APUE 学习笔记(八) 线程同步

1. 进程的所有信息对该进程内的所有线程都是共享的

包括可执行的程序文本、程序全局内存、堆内存以及文件描述符

线程包含了表示进程内执行环境必需的信息，包括线程ID、寄存器值、栈、调度优先级和策略、信号屏蔽字、线程私有数据

判断线程相等时采用 pthread_equal 函数

线程创建时并不能保证哪个线程会先执行，不能在线程调度上做出任何假设

#include <unistd.h>

#include <stdio.h>

#include <pthread.h>

void printid(const char* str)

{

    pid_t     pid = getpid();

    pthread_t tid = pthread_self();

    fprintf(stdout, "%s pid:%u,tid:%u\n", str, (unsigned int)pid, (unsigned int)tid);

}

void* thread_func(void* arg)

{

    printid("new thread: ");

    return (void*);

}

int main(int argc, char* argv[])

{

    pthread_t tid;

    int ret = pthread_create(&tid, NULL, thread_func, NULL);

    if (ret != ) {

        fprintf(stderr, "pthread_create error\n");

        return -;

    }

    printid("main thread: ");

    sleep();

    return ;

}

这段代码中需要两个地方需要处理主线程和新线程之间的竞争：

(1)主线程需要休眠，如果主线程不休眠，就可能在新线程有机会运行之前就退出终止了

(2)新线程通过 pthread_self 函数获取自己的线程ID，而不是从共享内存读出或者从线程的启动例程以参数的形式接收到

pthread_create 会通过第一个参数返回新建线程的线程ID，而新线程并不能安全的使用这个ID，因为新线程可能在主线程调用 pthread_create 回填线程ID之前就运行了，这时新线程使用的是未经初始化的ID，并不是正确的线程ID

#include <pthread.h>

#include <stdio.h>

#include <unistd.h>

#include <assert.h>

#define NTHREADS 5

void* PrintHello(void* threadId)

{

    int tid = ((int)threadId);

    fprintf(stdout, "Hello world, thread %d\n", tid);

    pthread_exit(NULL);

}

int main(int argc, char* argv[])

{

    pthread_t threads[NTHREADS];

    int rc = ;

    for (int i = ; i < NTHREADS; ++i) {

        fprintf(stdout, "In main: creating thread %d\n", i);

        rc = pthread_create(&threads[i], NULL, PrintHello, (void*)i);

        if (rc != ) {

            fprintf(stderr, "error:return code from pthread_create is %d\n", rc);

            exit(-);

        }

    }

    pthread_exit(NULL);

}

上述代码，我们创建了5个线程，每个线程打印一条包含线程编号的语句

可以预想到：每次运行程序时，结果不尽相同。因为线程创建时并不能保证哪个线程会先执行，不能在线程调度上做出任何假设

假如我们将上述代码中

 rc = pthread_create(&threads[i], NULL, PrintHello, (void*)i);

 void* PrintHello(void* threadId)

 {

     int tid = ((int)threadId);

     fprintf(stdout, "Hello world, thread %d\n", tid);

     pthread_exit(NULL);

 }

改为以下：

rc = pthread_create(&threads[i], NULL, PrintHello, (void*)&i);

void* PrintHello(void* threadId)

{

    int tid = *((int*)threadId);

    fprintf(stdout, "Hello world, thread %d\n", tid);

    pthread_exit(NULL);

}

仅有的差别就是线程执行函数的参数传递不同，执行改过之后的程序：

我们可以看到程序执行结果完全不同并且不正确

对于修改前：直接传递变量i 的值，这是值语义，之后线程操作的只是变量i 的副本，跟原来的变量i 没有任何关系，没有竞争出现

对于修改后：传递的是变量i 的地址(地址)，这是引用语义，之后线程操作的是原变量i，这时多个线程就出现了竞争，

因为这时变量i 的地址是共享内存，对所有线程可见，其余5个线程通过共享内存在读这个变量i，而主线程通过 i++在写这个变量值

这之间并没有任何同步，所以5个线程读取的值并不正确

#include <stdio.h>

#include <unistd.h>

#include <pthread.h>

#include <stdlib.h>

#include <assert.h>

#include <math.h>

#define NTHREADS 5

void* busywork(void* ptr)

{

    int tid = (int)ptr;

    fprintf(stdout, "Thread %d starting...\n", tid);

    double result = 0.0;

    for (int i = ; i < ; ++i) {

        result = result + sin(i) * tan(i);

    }

    fprintf(stdout, "Thread %d done. Result = %e\n", tid, result);

    pthread_exit((void*)ptr);

}

int main(int argc, char* argv[])

{

    pthread_t thread[NTHREADS];

    pthread_attr_t attr;

    /* Initialize and set thread detached attribute */

    pthread_attr_init(&attr);

    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

    for (int i = ; i < NTHREADS; ++i) {

        fprintf(stdout, "Main: creating thread %d\n", i);

        int rc = pthread_create(&thread[i], &attr, busywork, (void*)i);

        if (rc != ) {

            fprintf(stderr, "error:return code from pthread_create is %d\n", rc);

            exit(-);

        }

    }

    /* Free attribute and wait for the other threads */

    void* status;

    pthread_attr_destroy(&attr);

    for (int i = ; i < NTHREADS; ++i) {

        int rc = pthread_join(thread[i], &status);

        if (rc != ) {

            fprintf(stderr, "error:return code from pthread_join id %d\n", rc);

            exit(-);

        }

        fprintf(stdout, "Main:completed join with thread %d having a status of %d\n", i, (int)status);

    }

    fprintf(stdout, "Main: program completed. Exiting\n");

    pthread_exit(NULL);

}

2.线程可以通过phread_cancel函数来请求取消同一进程中的其它线程

3.进程原语和线程原语的比较

在默认情况下，线程的终止状态会保存到对该线程调用pthread_join，如果线程已经处于分离状态，那么不能用pthread_join函数等待它的终止状态。

3.线程同步

当多个控制线程共享相同的内存时，需要确保每个线程都看到一致的数据视图

(1)读写互斥

在变量修改时间多于1个存储器访问周期的处理器结构中，当存储器读与存储器写这两个周期交叉时，就有潜在的不一致性

为了解决这个问题，线程必须使用锁，在同一时间只允许一个线程访问该变量

(2)写写互斥

当多个线程在同一时间修改同一变量时，也需要同步

变量递增操作：

a.从内存单元读入寄存器

b.在寄存器中进行变量值的增加

c.把新值写回内存单元

如果修改操作是原子操作，那么就不存在竞争。然而在现代计算机系统中，存储器访问需要多个总线周期，通常在多个处理器上交叉进程

(3) 互斥量

对互斥量进行加锁以后，任何其它试图再次对互斥量加锁的线程将会被阻塞直到当前线程释放该互斥锁

#include <unistd.h>

#include <stdio.h>

#include <stdlib.h>

#include <pthread.h>

struct foo {

    int        f_count;

    pthread_t  f_lock;

    /* more stuff here... */

};

struct foo* foo_alloc(void)

{

    struct foo* fp = malloc(sizeof(struct foo));

    if (fp != NULL) {

        fp->f_count = ;

        int ret = pthread_mutex_init(&fp->f_lock, NULL);

        if (ret != ) {

            free(fp);

            return NULL;

        }

    }

    return fp;

}

/* increase a reference to the object */

void foo_increase(struct foo* fp)

{

    assert(fp != NULL);

    pthread_mutex_lock(&fp->f_lock);

    fp->f_count++;

    pthread_mutex_unlock(&fp->f_lock);

}

/* decrease a reference to the object */

void foo_decrease(struct foo* fp)

{

    assert(fp != NULL);

    pthread_mutex_lock(&fp->f_lock);

    if (--fp->f_count == ) {

        pthread_mutex_unlock(&fp->f_lock);

        pthread_mutex_destroy(&fp->f_lock);

        free(fp);

    } else {

        pthread_mutex_unlock(&fp->f_lock);

    }

}

4.避免死锁

两个线程都在相互请求另一个线程拥有的资源，这两个线程都无法向前运行，就产生了死锁

 #include <unistd.h>

 #include <stdio.h>

 #include <stdlib.h>

 #include <assert.h>

 #include <pthread.h>

 #define NHASH 29

 #define HASH(fp)  (((unsigned long)(fp)) % NHASH)

 pthread_mutex_t hashlock = PTHREAD_MUTEX_INITIALIZER;

 struct foo {

     struct foo*      f_next;

     int              f_count;

     pthread_mutex_t  f_lock;

     int              f_id;

     /* more stuff here... */

 };

 struct foo* fh[NHASH];

 struct foo* foo_alloc(void)

 {

     struct foo* fp = malloc(sizeof(struct foo));

     if (fp != NULL) {

         fp->f_count = ;

         int ret = pthread_mutex_init(&fp->f_lock, NULL);

         if (ret != ) {

             free(fp);

             return NULL;

         }

         int idx = HASH(fp);

         pthread_mutex_lock(&hashlock);

         fp->f_next = fh[idx];

         fh[idx] = fp->f_next;

         pthread_mutex_lock(&fp->f_lock);

         pthread_mutex_unlock(&hashlock);

         /*  continue initialization...... */

         pthread_mutex_unlock(&fp->f_lock);

     }

     return fp;

 }

 /* increase a reference to the object */

 void foo_increase(struct foo* fp)

 {

     assert(fp != NULL);

     pthread_mutex_lock(&fp->f_lock);

     fp->f_count++;

     pthread_mutex_unlock(&fp->f_lock);

 }

 /* find an existing object */

 struct foo* foo_find(int id)

 {

     struct foo* fp;

     int idx = HASH(fp);

     pthread_mutex_lock(&hashlock);

     for (fp = fh[idx]; fp != NULL; fp = fp->f_next) {

         if (fp->f_id == id) {

             foo_increase(fp);

             break;

         }

     }

     pthread_mutex_unlock(&hashlock);

     return fp;

 }

 /* decrease a reference to the object */

 void foo_decrease(struct foo* fp)

 {

     assert(fp != NULL);

     struct foo* tfp = NULL;

     int idx = ;

     pthread_mutex_lock(&fp->f_lock);

     if (fp->f_count == ) {

         pthread_mutex_unlock(&fp->f_lock);

         pthread_mutex_lock(&hashlock);

         pthread_mutex_lock(&fp->f_lock);

         /* need to recheck the condition */

         if (fp->f_count != ) {

             fp->f_count--;

             pthread_mutex_unlock(&fp->f_lock);

             pthread_mutex_unlock(&hashlock);

             return;

         }

         /* remove from list */

         idx = HASH(fp);

         tfp = fh[idx];

         if (tfp == fp) {

             fh[idx] = fp->f_next;

         } else {

             while (tfp->f_next != fp) {

                 tfp = tfp->f_next;

             }

             tfp->f_next = fp->f_next;

         }

         pthread_mutex_unlock(&hashlock);

         pthread_mutex_unlock(&fp->f_lock);

         pthread_mutex_destroy(&fp->f_lock);

         free(fp);

     } else {

         fp->f_count--;

         pthread_mutex_unlock(&fp->f_lock);

     }

 }

上述代码描述了两个互斥量的使用方法

当同时需要两个互斥量时，总是让它们以相同的顺序加锁，以避免死锁

hashlock维护着一个用于跟踪foo数据结构的散列列表，这样就保护foo结构中的fh散列表和f_next散列链字段

foo结构中的f_lock互斥量保护对foo结构中的其他字段的访问

在foo_decrease函数中，首先需要读取 f_count 这个时候我们必须对其加锁访问，防止其它线程在此过程中修改此变量

互斥读取到f_count变量后，如果它是最后一个引用，则需要从散列表中删除这个结构，这时需要读取全局的数据结构fh结构体数组，我们必须先对结构体互斥量解锁(第78行)，才可以重新获得散列表锁(第79行)，然后重新获取结构体互斥量(第80行)，因为两个互斥量的加锁顺序必须保持绝对一致。线程可能为了满足锁顺序而阻塞，其它线程可能就在此阻塞过程中对结构引用计数加1，所以必须重新检查条件。也就是说在第78行至80行之间，其它线程是可以对结构体引用计数加1的。

上述代码过于复杂，可以简化如下：

 #include <unistd.h>

 #include <stdio.h>

 #include <stdlib.h>

 #include <assert.h>

 #include <pthread.h>

 #define NHASH 29

 #define HASH(fp)  (((unsigned long)(fp)) % NHASH)

 struct foo {

     struct foo*       f_next;       /* protected by hashlock */

     int               f_count;      /* protected by hashlock */

     pthread_mutex_t   f_lock;

     int               f_id;

     /* more stuff here */

 };

 struct foo* fh[NHASH];

 pthread_mutex_t hashlock = PTHREAD_MUTEX_INITIALIZER;

 struct foo* foo_alloc(void)

 {

     int idx = ;

     struct foo* fp = malloc(sizeof(struct foo));

     if (fp != NULL) {

         fp->f_count = ;

         int ret = pthread_mutex_init(&fp->f_lock, NULL);

         if (ret != ) {

             free(fp);

             return NULL;

         }

         idx = HASH(fp);

         pthread_mutex_lock(&hashlock);

         fp->f_next = fh[idx];

         fh[idx] = fp->f_next;

         pthread_mutex_lock(&fp->f_count);

         pthread_mutex_unlock(&hashlock);

         /* continue initialization */

     }

     return fp;

 }

 void foo_increase(struct foo* fp)

 {

     assert(fp != NULL);

     pthread_mutex_lock(&hashlock);

     fp->f_count++;

     pthread_mutex_unlock(&hashlock);

 }

 struct foo* foo_find(int id)

 {

     struct foo* fp = NULL;

     int idx = HASH(fp);

     pthread_mutex_lock(&hashlock);

     for (fp = fh[idx]; fp != NULL; fp = fp->f_next) {

         if (fp->f_id == id) {

             fp->f_count++;

             break;

         }

     }

     pthread_mutex_unlock(&hashlock);

     return fp;

 }

 void foo_decrease(struct foo* fp)

 {

     assert(fp != NULL);

     struct foo* tfp = NULL;

     int idx = ;

     pthread_mutex_lock(&hashlock);

     if (--fp->f_count == ) {

         idx = HASH(fp);

         tfp = fh[idx];

         if (tfp == fp) {

             fh[idx] = fp->f_next;

         } else {

             while (tfp->f_next != fp) {

                 tfp = tfp->f_next;

             }

             tfp->f_next = fp->f_next;

         }

         pthread_mutex_unlock(&hashlock);

         pthread_mutex_destroy(&fp->f_lock);

         free(fp);

     } else {

         pthread_mutex_unlock(&hashlock);

     }

 }

如果锁的粒度太粗，就会出现很多线程阻塞等待相同的锁，源自并发性的改善微乎其微

如果锁的粒度太细，那么过多的锁会使系统性能开销受到影响，而且代码变得相当复杂

5.读写锁

互斥量要么是锁住状态，要么是不加锁状态，而且一次只有一个线程可以对其加锁

读写锁有三种状态：读状态加锁，写状态加锁，不加锁。一次只有一个线程可以占有写状态锁，多个线程可以同时占有多个读状态锁

读写锁非常适合于对数据结构读的次数远大于写的情况

读写锁也叫做共享-独占锁

6.条件变量

7. 线程属性

#include<pthread.h>

int pthread_attr_init(pthread_attr_t* attr);

int pthread_attr_destroy(pthread_attr_t* attr);

如果对现有的某个线程的终止状态不感兴趣的话，可以使用pthread_detach函数让操作系统在线程退出时回收它所占用的资源

pthread_attr_getstack pthread_attr_setstack 这两个函数可以用于管理stackaddr线程属性

pthread_attr_getstacksize pthread_attr_setstacksize 这两个函数可以用于管理stacksize线程属性

8.互斥量属性

#include <pthread.h>

int pthread_mutexattr_init(pthread_mutexattr_t* attr);

int pthread_mutexattr_destroy(pthread_mutexattr_t* attr);

9.重入、线程安全

如果一个函数在同一时刻可以被多个线程安全地调用，则该函数是线程安全的。

巴特西