Unix网络编程-同步

Posted 2020-07-16 爱橙子的OK绷

1、互斥锁（量）和条件变量

默认情况下互斥锁和条件变量用于线程间同步，若将它们放在共享内存区，也能用于进程间同步。

1.1 互斥锁

1、概述：
互斥锁(Mutex，也称互斥量)，防止多个线程对一个公共资源做读写操作的机制，以保证共享数据的完整性。

用以保护临界区，以保证任何时候只有一个线程（或进程）在访问共享资源(如代码段)。保护临界区的代码形式：

lock_the_mutex(...);
临界区
unlock_the_mutex(...);

任何时刻只有一个线程能够锁住一个给定的互斥锁。

下面的三个函数给一个互斥锁进行上锁和解锁：

#include<pthread.h>
int pthread_mutex_lock(pthread_mutex_t *mptr);
int pthread_mutex_trylock(pthread_mutex_t *mptr);
int pthread_mutex_unlock(pthread_mutex_t *mptr);

以上三个函数，如果调用成功均返回0，失败返回相应的error值。

如果尝试给一个已由某个线程锁住的互斥锁上锁，那么pthread_mutex_lock将阻塞到该互斥锁解锁为止。pthread_mutex_trylock是对应的非阻塞函数，如果该互斥锁已锁住，它就返回一个EBUSY错误。

如果有多个线程阻塞在等待同一个互斥锁上，那么当该互斥锁解锁时，哪一个线程会开始运行：不同线程被赋予不同优先级，同步函数将唤醒优先级最高的被阻塞线程。

2、互斥锁实现的生产者-消费者模型：

生产者-消费者问题也称有界缓冲区问题，若干个生产者和若干个消费者共享使用固定数目的缓冲区，因而带来的同步和通信问题。

各种IPC手段本身就是一个生产者-消费者问题的实例。
管道、FIFO和消息队列的同步是隐式同步，使用者只能通过指定的接口来使用这些IPC方式，其中的同步都由内核完成。

共享内存作为IPC，需要使用者进行显式同步，线程间共享全局数据也需要显式同步。

以多生产者，单消费者的模型为例：

在单个进程中有多个生产者线程和单个消费者线程，我们只关心多个生产者线程之间的同步，直到所有生产者线程都完成工作后，才启动消费者线程。

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#include <math.h>

#define MAXNITEMS 1000000
#define MAXNTHREADS 100
int nitems; //(1)生产者存放的条目数，只读(对于生产者或消费者)！！！
/*
*shared结构中的变量是共享数据
*/
struct {//(2)！！！
    pthread_mutex_t mutex;/*同步变量：互斥锁*/
    int buff[MAXNITEMS];//生产者会依次给buff数组存放数据
    int nput;/*nput是buff数组中下一次存放的元素下标*/
    int nval;/*nval是下一次存放的值（0,1,2等）*/
} shared = {
    PTHREAD_MUTEX_INITIALIZER //(3)对用于生产者线程间同步的互斥锁做初始化！！！
};
void *produce(void *), *consume(void *);

int main(int argc,char *argv[])
{
    /*
    *变量说明：tid_produce[]数组中保存每个线程的线程ID
    *count[]是每个线程计数器
    *tid_consume中保存单个的消费者的ID
    */
    int i, nthreads, count[MAXNTHREADS];
    pthread_t tid_produce[MAXNTHREADS], tid_consume;
    /*命令行参数个数判断*/
    if (argc != 3) {
        printf("usage: producer_consumer1 <#iterms> <#threads>\\n");
        exit(1);
    }
    /*
    * argv[1]中指定生产者存放的条目数
    * argv[2]中指定待创建的生产者线程的数目
    */
    nitems = min(atoi(argv[1]), MAXNITEMS);//(4)指定生产者存放的条目数！！！
    nthreads = min(atoi(argv[2]), MAXNTHREADS);//(5)创建多少个生产者线程！！！
    /*
    *set_concurrency函数用来告诉线程系统我们希望并发运行多少线程
    *即设置并发级别
    */
    set_concurrency(nthreads);//(6)！！！

    //(7)创建生产者线程：每个线程执行produce！！！
    for (i = 0; i < nthreads; i++) {//依次将buff[i]设置为i
        count[i] = 0;//计数器初始化为0,每个线程每次往缓冲区存放一个条目时给这个计数器加1.
        pthread_create(&tid_produce[i], NULL, produce, &count[i]);
    }

    //(8)等待所有生产者线程终止，并输出每个线程的计数器值！！！
    for (i = 0; i < nthreads; i++) {
        pthread_join(tid_produce[i], NULL);
        printf("count[%d] = %d\\n", i, count[i]);
    }
    //(9)然后启动单个消费者线程！！！
    pthread_create(&tid_consume, NULL, consume, NULL);
    //(10)接着等待消费者完成，然后终止进程！！！
    pthread_join(tid_consume, NULL);
    return 0;
}

//创建生产者线程
void *produce(void *arg)
{
    for ( ; ; ) {
        pthread_mutex_lock(&shared.mutex);//(1)上锁！！！
        //(2)临界区
        if (shared.nput >= nitems) {
            //说明此时已经生产完毕，解锁
            pthread_mutex_unlock(&shared.mutex);
            return (NULL);
        }
        shared.buff[shared.nput] = shared.nval;
        shared.nput++;
        shared.nval++;
        pthread_mutex_unlock(&shared.mutex);//(3)解锁！！！
        //count元素的增加(通过指针arg)不属于临界区,因为每个线程有各自的计数器
        *((int *) arg) += 1;
    }
}

//等待生产者线程，然后启动消费者线程
void *consume(void *arg)
{
    int i;
    /*
    *消费者只是验证buff中的条目是否正确，如果发现错误则输出一条信息
    *这个函数是只有一个实例在运行，而且是在所有的生产者线程都完成之后
    *因此不需要任何同步
    */
    for (i = 0; i < nitems; i++)
        if (shared.buff[i] != i)
            printf("buff[%d] = %d\\n", i, shared.buff[i]);
    return (NULL);
}

3、互斥锁的非正常终止：

若进程在持有互斥锁时终止，内核不会负责自动释放持有的锁。内核自动清理的唯一同步锁类型是fcntl记录锁。
若被锁住的互斥锁的持有进程或线程终止，会造成这个互斥锁无法解锁，因而死锁。线程可以安装线程清理程序，用来在被取消时能释放持有的锁。但这种释放可能会导致共享对象的状态被部分更新，造成不一致。

1.2 条件变量

互斥锁只能用于上锁，实现对某个共享对象的互斥访问，无法用于对某事件的等待。条件变量则用于等待。

#include<pthread.h>
int pthread_cond_wait(pthread_cond_t *cond,pthread_mutex_t *mutex);
int int pthread_cond_signal(pthread_cond_t *cond);

每个条件变量都需要关联一个互斥锁，用来提供对等待条件的互斥访问。

2、读写锁

读写锁可以在读数据与修改数据之间作区分。其规则如下：

1）没有线程持有写锁时，任意多的线程可以持有读锁。
2）仅当没有线程持有读锁或写锁时，才能分配写锁。

简言之，只要没有线程在写，那么所有线程都可以读；但是有线程要想写，必须是既没有线程在读，也没有线程在写。！！！

当已有线程持有读锁时，另一线程申请写锁则会阻塞，若后续还有读锁的申请，此时有两种策略：
1）对后续的读锁请求都通过，可能会造成因读锁不断被分配，写锁申请始终阻塞，“饿死”了写进程。
2）后续读锁请求都阻塞，等当前持有的读锁都结束后优先分配写锁。

与普通互斥锁相比，当被保护数据的读访问比写访问更为频繁时，读写锁能提供更高的并发度。

3、记录上锁

记录上锁是读写锁的一种扩展类型，它可用于有亲缘关系或无亲缘关系的进城之间共享某个文件的读与写。

执行上锁的函数是fcntl，锁由内核维护，其属主由进程ID标识。

特点：只用于不同进程间的上锁，而不是同一进程内不同线程间的上锁。

Unix内核没有记录这一概念，对记录的解释是由读写文件的应用进行的。每个记录就是文件中的一个字节范围。

使用fcntl记录上锁时，等待着的读出者优先还是等待着的写入者优先没有保证。

4、信号量

信号量是一种用于不同进程间，或一个给定进程内不同线程间同步手段的原语。
3中信号量类型：
1）Posix有名信号量：使用Posix IPC名字标识，可用于进程或线程间的同步。（可用于彼此无亲缘关系的进程间）
2）Posix基于内存的信号量（无名信号量）：存放在共享内存区，可用于进程或线程间的同步。（不可用于彼此无亲缘关系的进程间）
3）System V信号量：在内核中维护，可用于进程或线程间的同步。

Posix信号量不必在内核中维护（System V信号量由内核维护），由可能为路径名的名字来标识。

4.1 Posix信号量

1、概述

三种基本操作：
1）创建（create）：指定初始值。
2）等待（wait）：如果值小于等于0则阻塞，否则将其减一，又称P操作。
3）挂出（post）：将信号量的值加1，加后如果值大于0，则唤醒一个阻塞在等待上的线程，又称V操作。

信号量的wait和post与条件变量的wait和signal类似，区别是：因为永久的改变了信号量的值，信号量的操作总被记住（会影响到后续的操作）；条件变量的signal如果没有线程在等待，该信号将丢失（对后续操作没有影响）。

互斥锁是为上锁而优化的，条件变量是为等待优化的，信号量既可以上锁也可以等待，因此开销更大。

2、二值信号量
二值信号量，其值为0或1，资源锁住则信号量值为0，若资源可用则信号量值为1。

二值信号量可用于互斥，就像互斥锁一样。但互斥锁必须由锁住它的线程解锁，信号量的挂出却不必由执行过它的等待操作的同一线程执行。

二值信号量用于生产者消费者问题：考虑往某个缓冲区放一个条目的一个生产者，以及取走该条目的一个消费者，这种简化类型。

3、计数信号量
其值在0和某个限制值（32767内）之间，可统计资源数，信号量的值就是可用资源数。等待操作都等待信号量的值变为大于0（表示可用），然后将它减1；挂出操作则只是将信号量值加1（可用资源数增加），唤醒正在等待该信号量值变为大于0的任意线程。

4.2 System V信号量

System V信号量增加了另一级复杂度。

**计数信号量集：一个或多个信号量（构成一个集合），其中每个都是计数信号量。**System V信号量一般指计数信号量集。而Posix信号量一般指单个计数信号量。

5、信号量、互斥锁和条件变量的差异

信号量的意图在于进程间同步，这些进程可能共享也可能不共享内存区；互斥锁和条件变量的意图在于线程间同步，这些线程总是共享内存区；但是信号量也可用于线程间，互斥锁和条件变量也可用于进程间。

1）互斥锁总是由给他上锁的线程解锁，信号量的挂出也可由其他线程执行。
2）互斥锁要么被锁住，要么被解开（二值状态，类似于二值信号量）。
3）信号量有一个与之关联的状态（计数值），信号量的挂出操作总是被记住。然而当向一个条件变量发信号时，如果没有线程在等待，信号将丢失。

参考《Unix网络编程：卷2》