1. 线程
2. 线程包的接口
3. 多线程echo server
4. 多线程程序中的共享变量
5. 信号量 - 线程同步的理论
6. 信号量 - 接口
7. 信号量 - 二元信号量
8. 信号量 - 生产者/消费者模型
9. 信号量 - 读者/写者模型
10. 线程实际使用 - 预线程服务器

线程

void *thread(void *vargp);

int main(){
    pthread_t tid;
    pthread_create(&tid, NULL, thread, NULL);
    pthread_join(tid,NULL);
    exit(0);
}

void *thread(void *vargp){
    printf("No.%d thread prints Hello world!\n", pthread_self());
    return NULL;
}

线程包的接口

#include <pthread.h>
//定义一个参数是void *, 返回值也是void * 的函数叫 func
typedef void *(func)(void *);
//成功返回0, 不成功返回非零数
int pthread_create(pthread_t *tid, pthread_attr_t *attr, func *f, void *arg);

#include <pthread.h>

pthread_t pthread_self(void);

1. 线程例程返回的时候, 线程会隐式的终止
2. 调用pthread_exit函数, 线程会显式的终止. 如果主线程调用这个函数, 会等待其他所有对等线程终止, 再终止主线程和整个进程, 这个函数返回值为 thread_return .

           #include <pthread.h>
           //成功返回0, 不成功返回非零
           void pthread_exit(void *thread_return);
   

3. 对等线程调用Linux系统的exit函数, 进程会立刻终止, 所有线程也一起挂了
4. 一个对等线程通过pthread_cancel(tid)来终止ID为tid的线程.

#include <pthread.h>
//成功返回0, 不成功返回非零
int pthread_join(pthread_t tid, void **thread_return);

#include <pthread.h>
//成功返回0, 不成功返回非零
int pthread_detach(pthread_t tid);

#include <pthread.h>

pthread_once_t once_control = PTHREAD_ONCE_INIT;

//固定返回0
int pthread_once(pthread_once_t *once_control, void (*init_routine)(void));

多线程echo server

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include "csapp.h"

#define MAXLINE 8192

void echo(int connfd);

//线程例程声明
void *thread(void *vargp);

int main(int argc, char **argv) {
    //一个整数的指针connfdp, 用于指向已连接描述符的地址
    int listenfd, *connfdp;
    socklen_t clientlen;
    struct sockaddr_storage clientaddr;

    //声明线程id
    pthread_t tid;

    if (argc != 2) {
        fprintf(stderr, "usage: %s <port>\n", argv[0]);
        exit(0);
    }

    listenfd = Open_listenfd(argv[1]);

    while (1) {
        clientlen = sizeof(struct sockaddr_storage);
        //每次申请新内存, 然后在新的内存里装入新的已连接描述符
        //申请一块新的内存
        connfdp = Malloc(sizeof(int));

        //accept函数返回的connfdp存放在新的区域里, 这样可以确保每个线程分配到的connfdp指针指向不同的内存区域.
        *connfdp = Accept(listenfd, (SA *) &clientaddr, &clientlen);

        //启动新线程
        Pthread_create(&tid, NULL, thread, connfdp);
    }
}

//线程例程的实际执行代码
void *thread(void *vargp){
    //强制转换vargp的类型, 然后取出传入的connfdp指针的值, 就是新的已连接描述符
    int connfd = *((int *) vargp);
    //将自己变成分离线程, 为了结束的时候可以回收内存
    Pthread_detach(pthread_self());
    //释放vargp指向的内存, 也就是本来存放当前线程对应的已连接描述符的内存
    Free(vargp);
    echo(connfd);
    Close(connfd);
    return NULL;
}


void echo(int connfd){
    size_t n;
    char buf[MAXLINE];
    rio_t rio;

    Rio_readinitb(&rio, connfd);

    while ((n = Rio_readlineb(&rio, buf, MAXLINE)) != 0) {
        printf("server received %d bytes \n", (int) n);
        Rio_writen(connfd, buf, n);
    }

}

练习12.5 为何只在线程例程中关闭了已连接描述符?

多线程程序中的共享变量

1. 每个线程独立的使用线程ID, 栈, 栈指针, 程序计数器, 条件码 和通用目的寄存器
2. 所有线程共享进程上下文的剩余部分, 也就是整个虚拟地址空间, 包括只读代码区, 数据区, 堆, 共享库代码
3. 所有的线程对外都是一个进程, 因此也共享操作系统为进程提供的上下文, 其中最核心的就是文件描述符

#include "csapp.h"
#define N 2

void *thread(void *vargp);

char **ptr;

int main(){
    int i;
    pthread_t tid;
    char *msgs[N] = {
            "Hello from foo",
            "Hello from bar",
    };

    ptr = msgs;

    for (i = 0; i < N; i++) {
        Pthread_create(&tid, NULL, thread, (void *) i);
    }
    sleep(1);
    printf("tid is %u\n", tid);
}

void *thread(void *vargp){
    int myid = (int) vargp;
    static int cnt = 0;
    printf("[%d]: %s (cnt = %d)\n", myid, ptr[myid], ++cnt);
    return NULL;
}

1. char **ptr;是全局变量, 显然被共享, 两个线程都从其中获得了自己要输入的字符串
2. main函数中声明的变量虽然这些是局部变量, 把线程看做一个过程的话, 这些过程都被main函数调用, 这些局部变量, 也只有一个副本, 被所有线程共享.
3. 线程的局部变量, 这个要首先从程序结构而不是并发上去理解, myid是局部非静态变量, 所以只被当前过程使用, 过程变成线程例程的时候, 各个例程使用各自的myid局部变量, 这在输出结果中也能看到, myid是不同的.
4. static int cnt, 依然要先从程序结构去判断, 在之前章节已经知道, 局部的静态变量会存储在一个单独的位置, 过程结束依然存在. 这里也是这样, 两个线程共享线程代码中定义的静态变量, 从输出结果中就可以看到.

练习 12.6 填写上边程序中各个变量的共享情况

信号量 - 线程同步理论

#include "csapp.h"

void *thread(void *vargp);

volatile long cnt = 0;

int main(int argc, char **argv){
    long niters;
    pthread_t tid1, tid2;

    if (argc != 2) {
        printf("usage: %s <niters>\n", argv[0]);
        exit(0);
    }

    niters = atoi(argv[1]);

    Pthread_create(&tid1, NULL, thread, &niters);
    Pthread_create(&tid2, NULL, thread, &niters);

    Pthread_join(tid1, NULL);
    Pthread_join(tid2, NULL);

    if (cnt != 2 * niters) {
        printf("BOOM! cnt = %ld\n", cnt);
    } else {
        printf("OK cnt = %ld\n", cnt);
    }
}

void *thread(void *vargp){
    long i, niters = *(long *) vargp;

    for (i = 0; i < niters; i++) {
        cnt++;
    }
}

1. 取数步骤Li, movq cnt(%rip), %rdx, 从全局变量cnt的地址中将值装入寄存器%rdx
2. 自增步骤Ui, addq $1, %rdx, 将%rdx自增1
3. 写回步骤Si, movq %rdx, cnt(%rip), 将%rdx中自增1后的值写回cnt

练习题 12.7 根据指令顺序完成下表

练习 12.8 使用进度图下列轨迹线是否安全?

1. H1,L1,U1,S1,H2,L2,U2,S2,T2,T1
2. H2,L2,H1,L1,U1,S1,T1,U2,S2,T2
3. H1,H2,L2,U2,S2,L1,U1,S1,T1,T2

信号量 - 实际操作

1. P操作: 检查信号量, 如果非0, 将s减1, 然后立刻返回(即不阻塞, 执行后续代码); 如果是0, 就挂起当前线程, 直到s不再是0.
2. V操作: 将s加1, 然后检测是不是有任何线程阻塞在P操作, 如果有, 重启这些线程中的一个.

#include <semaphore.h>

//声明信号量
sem_t sem;

//初始化信号量
int sem_init(sem_t *sem, 0, unsinged int value);

//P操作
int sem_wait(sem_t *s);

//V操作
int sem_post(sem_t *s);

//上边这些函数成功都返回0, 出错返回-1

void P(sem_t *sem)
{
    if (sem_wait(sem) < 0)
        unix_error("P error");
}

void V(sem_t *sem)
{
    if (sem_post(sem) < 0)
        unix_error("V error");
}

信号量 - 二元信号量

#include "csapp.h"

void *thread(void *vargp);

volatile long cnt = 0;

//声明信号量
sem_t mutex;

int main(int argc, char **argv){

    long niters;
    pthread_t tid1, tid2;

    if (argc != 2) {
        printf("usage: %s <niters>\n", argv[0]);
        exit(0);
    }

    niters = atoi(argv[1]);

    //初始化信号量
    Sem_init(&mutex, 0, 1);

    Pthread_create(&tid1, NULL, thread, &niters);
    Pthread_create(&tid2, NULL, thread, &niters);

    Pthread_join(tid1, NULL);
    Pthread_join(tid2, NULL);

    if (cnt != 2 * niters) {
        printf("BOOM! cnt = %ld\n", cnt);
    } else {
        printf("OK cnt = %ld\n", cnt);
    }
}

void *thread(void *vargp){
    long i, niters = *(long *) vargp;

    for (i = 0; i < niters; i++) {
        P(&mutex);
        cnt++;
        V(&mutex);
    }
}

信号量 - 生产者消费者模型

#include "csapp.h"

#ifndef C_SBUF_H
#define C_SBUF_H

typedef struct {
    int *buf;       //缓冲区数组, 采用动态分配内存
    int n;          //最大槽位, 也就是数组的最大长度
    int front;      //记录数组第一项的索引
    int rear;       //记录数组最后一项的索引
    sem_t mutex;    //用于控制访问缓冲区的互斥信号量
    sem_t slots;    //统计空的插槽
    sem_t items;    //统计可用的数据(装了数据的插槽)
} sbuf_t;

#endif //C_SBUF_H

1. sbuf_init函数初始化缓冲区, 指定n个槽位, 然后会给buf动态分配空间, 同时初始化信号量, mutex作为互斥信号量为1, slots一开始全部可用, 设置为n. 而可用数据一开始是0.
2. 生产者生产数据, 调用sbuf_insert 向其中放入数据, 由于要根据slot的数量来放入数据, 因此需要加上slots信号量, 然后实际放数据的时候, 因为要访问缓冲区, 还必须加上mutex信号量. 然后插入一个数据, 此时会因为V操作将slots信号量减去1. 如果slots已经是0, 就会阻塞.
3. 消费者取用数据, 调用sbuf_remove 取出数据, 由于要根据items的数量来取出数据, 因此需要加上items信号量, 然后实际取数据的时候, 因为要访问缓冲区, 还必须加上mutex信号量. 然后取走一个数据, 将front指针移动到下一个位置(这里使用了取余的技巧, 让front超过长度再移动到索引0的位置), 此时会因为V操作将items信号量减去1. 如果items已经是0, 就会阻塞.
4. 需要结束生产者和消费者程序的时候, 释放结构中的动态分配的部分

void sbuf_init(sbuf_t *sp, int n){

    //动态分配n个int的空间, 这里就是用int当做生产和消费的数据类型
    sp->buf = Calloc(n, sizeof(int));
    //数组长度设置为n
    sp->n = n;
    //一开始没有数据, front和rear都是0
    sp->front = sp->rear = 0;
    //访问缓冲区的互斥信号量要初始化为1
    Sem_init(&sp->mutex, 0, 1);
    //初始没有数据, 可用槽位就是n
    Sem_init(&sp->slots, 0, n);
    //初始可用数据是0
    Sem_init(&sp->items, 0, 0);
}

void sbuf_insert(sbuf_t *sp, int item){

    //检查是不是有空槽
    P(&sp->slots);
    //通过检查空槽的话, 要读写缓冲区了, 必须要互斥访问
    P(&sp->mutex);
    //都通过了, 放入数据
    //这里用了一个小技巧, rear增加后不断取余, 索引会反复在1-9-0之间变动.
    sp->buf[(++sp->rear) % (sp->n)] = item;
    //解除互斥访问
    V(&sp->mutex);
    //这里很关键, 有了空槽写完之后, 要给谁加1. 由于空槽的数量已经被P减1了, 又放入了一个数据, 显然不能再P(&sp->slots), 否则这个slots失去意义了
    //所以是要给items 加1, 表示现在可用数据多了1个
    V(&sp->items);
}

int sbuf_remove(sbuf_t *sp){

    int data;
    //检查是不是有可读取的数据
    P(&sp->items);
    //如果有可读取数据, 要读写缓冲区了, 必须要互斥访问
    P(&sp->mutex);
    //取出数据, 这里的技巧和之前相同, 也让front在1-9-0之间变动.
    data = sp->buf[(++sp->front) % (sp->n)];
    //解除互斥访问
    V(&sp->mutex);
    //基于同样的分析, 这里应该给slots加1, 表示空出来一个槽
    V(&sp->slots);
    return data;
}

void subf_deinit(sbuf_t *sp) {
    Free(sp->buf);
}

练习 12.9 指出下列场景互斥锁是否必要

1. p = 1, c = 1, n > 1
2. p = 1, c = 1, n = 1
3. p > 1, c > 1, n = 1

信号量 - 读者/写者模型

int readcnt=0;
sem_t mutex=1, W=1;

void writer(void){
    while (1) {
        P(&W);
        do something;
        V(&W);
    }
}

void reader(void){
    while(1) {
        //读者进来的代码
        P(&mutex);
        readcnt++;
        if (readcnt == 1) {
            P(&W);
        }
        V(&mutex);

        do something;

        //读者干完活要离开的代码
        P(&mutex);
        readcnt--;
        if (readcnt == 0) {
            V(&W);
        }
        V(&mutex);
    }
}

练习 12.10 数量不平衡时候出现的问题

线程实际使用 - 预线程服务器

#include "csapp.h"
#include "sbuf.h"
#include "echo.h"

#define NTHREADS 4
#define SBUFSIZE 16

void echo_cnt(int connfd);
void *thread(void *vargp);

sbuf_t sbuf;

int main(int argc, char **argv){
    int i, listenfd, connfd;
    socklen_t clientlen;
    struct sockaddr_storage clientaddr;
    pthread_t tid;

    if (argc != 2) {
        fprintf(stderr, "usage: %s <port>\n", argv[0]);
        exit(0);
    }

    listenfd = Open_listenfd(argv[1]);

    //初始化工作缓冲区, 创建指定数量的线程
    sbuf_init(&sbuf, SBUFSIZE);
    for (i = 0; i < NTHREADS; i++) {
        Pthread_create(&tid, NULL, thread, NULL);
    }

    //主线程之后做的事情, 就是等待连接然后将已连接描述符放入缓冲区
    while(1){
        clientlen = sizeof(struct sockaddr_storage);
        connfd = Accept(listenfd, (SA *) &clientaddr, &clientlen);

        //使用编写好的库, 添加connfd到缓冲区中
        sbuf_insert(&sbuf, connfd);
    }
}

void *thread(void *vargp){
    //将自己变成分离线程
    Pthread_detach(pthread_self());
    //不断循环的等待从缓冲区拿描述符然后干活
    while (1) {
        int connfd = sbuf_remove(&sbuf);
        echo_cnt(connfd);
        Close(connfd);
    }
}

#include "csapp.h"
#include "echo.h"

static int byte_cnt;
static sem_t mutex;

static void init_echo_cnt(void){
    Sem_init(&mutex, 0, 1);
    byte_cnt = 0;
}

void echo_cnt(int connfd){
    int n;
    char buf[MAXLINE];
    rio_t rio;
    static pthread_once_t once = PTHREAD_ONCE_INIT;

    Pthread_once(&once, init_echo_cnt);
    Rio_readinitb(&rio, connfd);
    while ((n = Rio_readlineb(&rio, buf, MAXLINE)) != 0) {
        P(&mutex);
        byte_cnt += n;
        printf("Server received %d (%d total) bytes on fd %d", n, byte_cnt, connfd);
        V(&mutex);
        Rio_writen(connfd, buf, n);
    }
}