UNIX 环境高级编程笔记之线程

1 线程标识

进程 ID 在整个系统中唯一标识，但线程 ID 只有在它所属的进程上下文中才有意义

进程 ID 由 32 位非负整型表示，线程 ID 则是一个结构体

#icnlude <pthread.h>
int pthread_equal(pthread_t tid1, pthread_t tid2); // 两个线程比较
pthread_t pthread_self(void); // 获取自身的线程 ID

2 线程创建

传统进程模型中，每个进程只有一个控制线程；POSIX 线程的情况下，也是以单进程中的单个控制线程启动的，创建多个控制线程前，程序的行为与传统的进程并无区别

#include <pthread.h>
int pthread_create(pthread_t *restrict tidp,
                   const pthread_attr_t *restrict attr,
                   void *(*strat_rtn)(void *), void *restrict arg); // 创建线程

tidp 指向线程数据结构的指针作为线程 ID， attr 线程的属性，start_rtn 线程从该函数地址开始运行，arg 传入起始函数的参数

线程创建并不保证哪个会先运行；新建的线程可访问进程的地址空间，并继承调用线程的浮点环境和信号屏蔽字，但挂起信号集会被清除

实验

void print_id(char* str) {
    pthread_t tid = pthread_self();
    pid_t pid = getpid();
    while (1) {
        sleep(1);
        printf("%s: pid: %lu, tid: %lu\n", str, pid, tid);
    }
}

void* thr_fn(void* arg) {
    print_id("new thread");
}

int main() {
    pthread_t ntid;
    pthread_create(&ntid, NULL, thr_fn, NULL);
    printf("main print ntid: %lu\n", ntid); // ntid 保存创建新线程 ID
    print_id("main thread");
}

编译及运行结果如下，多运行几次，主线程和新建线程打印顺序可能会不一样

main print ntid: c6f6000
main thread: pid: 64593, tid: b25fa380
new thread: pid: 64593, tid: c6f6000
main thread: pid: 64593, tid: b25fa380
new thread: pid: 64593, tid: c6f6000
main thread: pid: 64593, tid: b25fa380
new thread: pid: 64593, tid: c6f6000
...

3 线程终止（上）

进程中的任意线程调用 exit、_Exit、_exit 整个进程都会终止

单个线程的退出：return、被其他线程取消 pthread_cancel、pthread_exit

#include <pthread.h>
void pthread_exit(void *rval_ptr); // 线程返回，传入无类型指针，join 可以访问到这个指针
int pthread_join(pthread_t thread, void **rval_ptr); // 等待某个线程返回或取消，rval_ptr 获得返回状态

实验一

void* thr_fn1(void* arg) {
    printf("thread 1\n");
    pthread_exit((void*) "t1 exit\n"); // 传入返回状态
}

void* thr_fn2(void* arg) {
    printf("thread 1\n");
    pthread_exit((void*) "t2 exit\n");
}

int main() {
    pthread_t ntid1;
    pthread_t ntid2;
    void* status;
    pthread_create(&ntid1, NULL, thr_fn1, NULL);
    pthread_create(&ntid2, NULL, thr_fn2, NULL);
    pthread_join(ntid1, &status); // 等待线程退出，并获取状态
    printf("%s", (char*) status);
    pthread_join(ntid2, &status);
    printf("%s", (char*) status);
}

编译及运行结果如下

thread 1
thread 1
t1 exit
t2 exit

实验二

struct foo {
    int a, b;
};

void print_foo(char* s, struct foo* fp) {
    printf("%s foo address: 0x%lx\n", s, (unsigned long) fp);
    printf("a: %d, b: %d\n", fp->a, fp->b);
};

void* thr_fn(void* arg) {
    struct foo f = {1, 2}; // 栈上分配，栈帧销毁会被回收
    print_foo("new thread", &f);
    pthread_exit((void*) &f);
}

int main() {
    pthread_t ntid1;
    void* status;
    pthread_create(&ntid1, NULL, thr_fn, NULL);
    pthread_join(ntid1, &status);
    print_foo("main thread", (struct foo *) status);
}

编译及运行结果如下

new thread foo address: 0x7f75bfee4ee8
a: 1, b: 2
main thread foo address: 0x7f75bfee4ee8
a: 0, b: 0  --> 有可能会得到乱七八糟的值，甚至不给读直接退出了，因为新建的线程退出这片空间也不属于你了

pthread 不属于默认库，gcc 编译时加上 -pthread

实验三

struct foo {
    int a, b;
};

void print_foo(char* s, struct foo* fp) {
    printf("%s foo address: 0x%lx\n", s, (unsigned long) fp);
    printf("a: %d, b: %d\n", fp->a, fp->b);
};

void* thr_fn(void* arg) {
    struct foo* fp = (struct foo*) malloc(sizeof(struct foo)); // 堆上分配
    fp->a = 1;
    fp->b = 2;
    print_foo("new thread", fp);
    pthread_exit((void*) fp);
}

int main() {
    pthread_t ntid;
    void* status;
    pthread_create(&ntid, NULL, thr_fn, NULL);
    pthread_join(ntid, &status);
    print_foo("main thread", (struct foo*) status);
    free(status); // 记得回收
}

编译及运行结果如下

new thread foo address: 0x7f6640000b60
a: 1, b: 2
main thread foo address: 0x7f6640000b60
a: 1, b: 2

4 线程终止（下）

#include <pthread.h>
int pthread_cancel(pthread_t tid); // 申请取消同进程下的其他线程，被取消的线程可以选择忽略或控制如何取消
int pthread_cleanup_push(void (*rtn)(void *), void *arg); // 同进程终止处理程序一样，这是为线程添加清理程序
int pthread_cleanup_pop(int execute); // push pop 想象成栈

清理程序被调用的时机

• pthread_exit
• 线程被其他线程取消 pthread_cancel
• 非 0 execute 参数调用 pthread_cleanup_pop

注意：正常 return 不会执行清理程序；push 与 pop 一定要成对出现，否则会编译不过

实验

void clean(void* arg) {
    printf("clean %s\n", (char*) arg);
}

void* thr_fn1(void* arg) {
    pthread_cleanup_push(clean, (void*) "t1");
        printf("thread 1\n");
        pthread_exit(NULL);
        printf("next"); // 这行没打印，确定不是由下方 pop 执行清理程序
    pthread_cleanup_pop(1);
}

void* thr_fn2(void* arg) {
    pthread_cleanup_push(clean, (void*) "t2");
        printf("thread 2\n");
    pthread_cleanup_pop(1);
}

void* thr_fn3(void* arg) {
    pthread_cleanup_push(clean, (void*) "t3");
        printf("thread 3\n");
        pause();
        printf("next"); // 这一行没打印，确定已被取消
    pthread_cleanup_pop(1);
}

void* thr_fn4(void* arg) {
    pthread_cleanup_push(clean, (void*) "t4");
        printf("thread 4\n");
        return NULL; // return 等价于 exit，所以整个进程会结束，不会执行线程清理程序
    pthread_cleanup_pop(1);
}

int main() {
    pthread_t tid1, tid2, tid3, tid4;
    pthread_create(&tid1, NULL, thr_fn1, NULL);
    pthread_create(&tid2, NULL, thr_fn2, NULL);
    pthread_create(&tid3, NULL, thr_fn3, NULL);
    pthread_cancel(tid3);
    sleep(1); // 确保以上线程任务执行完毕
    pthread_create(&tid4, NULL, thr_fn4, NULL);
    sleep(1);
    printf("next"); // 这行并不会打印，证明整个线程被干掉了
}

编译及运行结果如下，顺序不一致正常

thread 1
thread 2
thread 3
clean t1
clean t2
clean t3
thread 4

进程原语与线程原语的比较

进程原语	线程原语
fork	pthread_create
exit	pthread_exit
waitpid	pthread_join
atexit	pthread_cancel_push
getpid	pthread_self
abort	pthread_cancel

5 线程同步

变量操作分解为

1. 内存单元的值读入寄存器
2. CPU 对寄存器的值做操作
3. 寄存器的值写入内存单元

多个线程在同一时间对变量做操作，不同步的话就可能出现问题

5.1 互斥量

使用 pthread 互斥接口保护数据，确保统一时间只有一个线程访问数据，本质是一把锁

#include <pthread.h>
int pthread_mutex_init(pthread_mutex_t *restrict mutex,
                       const pthread_mutexattr_t *restrict attr); // 使用前先初始化
int pthread_mutex_destroy(pthread_mutex_t *mutex); // malloc 申请的需要手动释放
int pthread_mutex_lock(pthread_mutex_t *restrict mutex); // 加锁，互斥量已上锁会阻塞
int pthread_mutex_trylock(pthread_mutex_t *restrict mutex); // 不会阻塞的加锁
int pthread_mutex_unlock(pthread_mutex_t *restrict mutex); // 解锁

实验

struct {
    int f_count;
    pthread_mutex_t f_lock;
} typedef foo;

foo* foo_alloc(int count) { // 构造器
    foo* fp = (foo*) malloc(sizeof(foo)); // 堆上分配
    fp->f_count = count;
    pthread_mutex_init(&fp->f_lock, NULL); // 初始化
    return fp;
}

void foo_inc(foo * fp) {
    pthread_mutex_lock(&fp->f_lock); // 加锁
    fp->f_count++;
    pthread_mutex_unlock(&fp->f_lock); // 解锁
}

void* thr_fn(void* arg) {
    foo* fp = (foo*) arg;
    for (int i = 0; i < 100; ++i) {
        foo_inc(fp);
        printf("tid: 0x%x, current f_count: %d\n", pthread_self(), fp->f_count);
    }
    pthread_exit(NULL);
}

int main() {
    pthread_t tid1, tid2;
    foo* fp = foo_alloc(0);
    pthread_create(&tid1, NULL, thr_fn, fp);
    pthread_create(&tid2, NULL, thr_fn, fp);
    sleep(1);
    printf("f_count: %d\n", fp->f_count);
    pthread_mutex_destroy(&fp->f_lock);
    free(fp);
}

编译及运行结果如下

tid: 0x7032000, current f_count: 1
tid: 0x70b5000, current f_count: 2
tid: 0x7032000, current f_count: 3
tid: 0x70b5000, current f_count: 4
...
tid: 0x70b5000, current f_count: 197
tid: 0x70b5000, current f_count: 198
tid: 0x7032000, current f_count: 199
tid: 0x7032000, current f_count: 200
f_count: 200

5.2 避免死锁

多次加锁造成死锁

int main() {
    pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER; // 互斥量的另一种初始化方式
    pthread_mutex_init(&lock, NULL);
    pthread_mutex_lock(&lock);
    pthread_mutex_lock(&lock);
    pthread_mutex_unlock(&lock);
    pthread_mutex_unlock(&lock);
    printf("hello\n"); // 不会打印，证明死锁
}

线程 1 占有 A 资源，线程 2 占有 B 资源，需要同时拥有 A、B 资源才能继续进行下去（只要加锁顺序一致，就能避免）

void* thr_fn1(void* arg) {
    pthread_mutex_t* logs = (pthread_mutex_t*) arg;
    for (int i = 0; i < 200000; ++i) {
        pthread_mutex_lock(&logs[1]); // 1 加锁，1、0 交换一下，两个线程顺序加锁就不会死锁
        pthread_mutex_lock(&logs[0]); // 0 加锁
        pthread_mutex_unlock(&logs[0]); // 解锁
        pthread_mutex_unlock(&logs[1]); // 解锁
    }
}

void* thr_fn2(void* arg) {
    pthread_mutex_t* logs = (pthread_mutex_t*) arg;
    for (int i = 0; i < 200000; ++i) {
        pthread_mutex_lock(&logs[0]); // 0 加锁
        pthread_mutex_lock(&logs[1]); // 1 加锁
        pthread_mutex_unlock(&logs[0]); // 解锁
        pthread_mutex_unlock(&logs[1]); // 解锁
    }
    printf("hello\n"); // 不打印死锁
}

int main() {
    pthread_mutex_t lock1 = PTHREAD_MUTEX_INITIALIZER;
    pthread_mutex_t lock2 = PTHREAD_MUTEX_INITIALIZER;
    pthread_mutex_t* locks = (pthread_mutex_t*) malloc(sizeof(pthread_mutex_t) * 2);
    locks[0] = lock1;
    locks[1] = lock2;
    pthread_t tid1, tid2;
    pthread_create(&tid1, NULL, thr_fn1, locks);
    pthread_create(&tid2, NULL, thr_fn2, locks);
    pause();
}

5.3 pthread_mutex_timedlock 函数

#include <pthread.h>
#include <time.h>
int pthread_mutex_timedlock(pthread_mutex_t *restrict mutex, 
                            const struct timespec *restrict tsptr); // 到了时间还未获得锁，就不阻塞
// MacOS 不支持

实验

int main(void)
{
    struct timespec tout;
    pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
    pthread_mutex_lock(&lock);
    clock_gettime(CLOCK_REALTIME, &tout);
    localtime(&tout.tv_sec);
    tout.tv_sec += 1; 
    pthread_mutex_timedlock(&lock, &tout);
    printf("hello\n"); // 打印，没死锁
}

5.4 读写锁

读加读锁，写加写锁，读写互斥，写写互斥，读读不互斥。适合读多于写的情况

#include <pthread.h>
// 使用前要初始化，释放底层内存前必须销毁
int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock,
                        const pthread_rwlockattr_t *restrict attr);
int pthread_rwlock_destroy(pthread_rwlock_t *restrict rwlock);
int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock); // 写锁
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock); // 读锁
int pthread_rwlock_unlock(pthread_rwlock_t *rwlock); // 无锁
int pthread_rwlock_tryrdlock(pthread_rwlock_t *rwlock); // 加写锁，加不到不阻塞
int pthread_rwlock_rtywrlock(pthread_rwlock_t *rwlock); // 加读锁，加不到不阻塞
int pthread_rwlock_timedrdlock(pthread_rwlock_t *rwlock,
                               const struct timespec *restrict tsptr); // 带超时的写锁
int pthread_rwlock_timedwrlock(pthread_rwlock_t *rwlock,
                               const struct timespec *restrict tsptr); // 带超时的读锁

5.5 条件变量

条件变量是用来等待而不是用来上锁的。条件变量用来自动阻塞一个线程，直到某特殊情况发生为止。与互斥量搭配使用

#include <pthread.h>
// 使用前要初始化，释放底层内存前必须销毁
int pthread_cond_init(pthread_cond_t *restrict cond,
                      const pthread_cond_t *restrict attr);
int pthread_cond_destroy(pthread_cond_t *restrict cond);
int pthread_cond_wait(pthread_cond_t *restrict cond,
                      pthread_mutex_t *restrict mutex);
int pthread_cond_timedwait(pthread_cond_t *restrict cond,
                           pthread_mutex_t *restrict mutex,
                           const struct timespec *restrict tsptr ); // 带超时版本
int pthread_cond_signal(pthread_cond_t *restrict cond); // 唤醒一个等待该条件的线程
int pthread_cond_broadcast(pthread_cond_t *restrict cond);  // 唤醒所有等待该条件的线程

5.6 自旋锁

不通过休眠使进程阻塞，而是在获取锁之前一直处于忙等（自旋）阻塞状态。自旋锁可用于以下情况：锁被持有的时间短，而且线程并不希望在重新调度上花费太多的成本。

很多互斥量的实现非常高效，以至于应用程序采用互斥锁的性能与曾经采用过自旋锁的性能基本是相同的。事实上，有些互斥量的实现在试图获取互斥量的时候会自旋一小段时间，只有在自旋计数到达某一阈值的时候才会休眠。这些因素，加上现代处理器的进步，使得上下文切换越来越快，也使得自旋锁只在某些特定的情况下有用。

#include <pthread.h>
// 与互斥量接口一样
int pthread_spin_init(pthread_spinlock_t *lock, int pshared);
int pthread_spin_destroy(pthread_spinlock_t *lock);
int pthread_spin_lock(pthread_spinlock_t *lock);
int pthread_spin_trylock(pthread_spinlock_t *lock);
int pthread_spin_unlock(pthread_spinlock_t *lock);

5.7 屏障

#include <pthread.h>
int pthread_barrier_init(pthread_barrier_t *restrict barrier,
                const pthread_barrierattr_t *restrict attr, 
                unsigned int count);
int pthread_barrier_destroy(pthread_barrier_t *barrier);
int pthread_barrier_wait(pthread_barrier_t *barrier);

初始化屏障时，可以使用 count 参数指定，在允许所有线程继续运行之前，必须到达屏障的线程数目

调用 pthread_barrier_wait 的线程在屏障计数未满足条件时，会进入休眠状态。如果该线程是最后一个调用 pthread_barrier_wait 的线程，就满足了屏障计数，所有的线程都被唤醒。