Linux系统编程9：多线程同步

多线程同步主要有信号量、互斥量、条件变量和读写锁四种方式。

0. 背景

竞争

#include <stdio.h>
#include <pthread.h>
 
void* func(void* arg){
    printf("enter func\n");
    sleep(1);
    printf("do something\n");
    sleep(1);
    printf("level func\n");
}
 
int main(int argc,int argv[]){  
    pthread_t tids[3];
    int i;
    for(i=0;i<3;i++){
        pthread_create(&tids[i],NULL,func,&mutex);
    }
    for(i=0;i<3;i++){
        pthread_join(tids[i],NULL);
    }
}

1. 信号量

1.1 操作

No.	操作	函数
1	创建	int sem_init(sem_t *sem, int pshared, unsigned int value)
2	销毁	int sem_destroy(sem_t *sem)
3	阻塞等待	int sem_wait(sem_t *sem)
4	非阻塞等待	int sem_trywait(sem_t * sem)
5	触发	int sem_post(sem_t *sem)

1.1.1 创建

int sem_init(sem_t *sem, int pshared, unsigned int value)

No.	参数	含义
1	sem	信号量对象
2	pshared	信号量类型。0：线程共享；<0：进程共享
3	value	初始值

• 返回值

No.	返回值	含义
1	0	成功
1	-1	失败

1.1.2 销毁

int sem_destroy(sem_t *sem)

No.	参数	含义
1	sem	信号量对象

• 返回值

No.	返回值	含义
1	0	成功
2	-1	失败

1.2 等待

1.2.1 阻塞等待

int sem_wait(sem_t *sem)

No.	参数	含义
1	sem	信号量对象

• 返回值

No.	返回值	含义
1	0	成功
2	-1	失败

1.2.2 非阻塞等待

int sem_trywait(sem_t * sem)

No.	参数	含义
1	sem	信号量对象

• 返回值

No.	返回值	含义
1	0	成功
2	-1	失败

1.3 触发

int sem_post(sem_t *sem) 

No.	参数	含义
1	sem	信号量对象

• 返回值

No.	返回值	含义
1	0	成功
2	-1	失败

• 示例
  解决竞争

#include <stdio.h>
#include <semaphore.h>
#include <pthread.h>
 
void* func(void* arg){
    sem_wait(arg);
    printf("enter func\n");
    sleep(1);
    printf("do something\n");
    sleep(1);
    printf("level func\n");
    sem_post(arg);
}
 
int main(int argc,int argv[]){
    sem_t sem;
    sem_init(&sem,0,1);
     
    pthread_t tids[3];
    int i;
    for(i=0;i<3;i++){
        pthread_create(&tids[i],NULL,func,&sem);
    }
    for(i=0;i<3;i++){
        pthread_join(tids[i],NULL);
    }
    sem_destroy(&sem);
 
}

2. 互斥量

• 比喻
  ATM取款
  toilet

2.1 分类

No.	分类	实现	特点
1	静态分配互斥量	pthread_mutex_t mutex= PTHREAD_MUTEX_INITIALIZER;	简单
2	动态分配互斥量	pthread_mutex_init(&mutex, NULL);pthread_mutex_destroy(&mutex);	可以设置更多的选项

2.2 操作

No.	操作	函数
1	加锁	int pthread_mutex_lock(pthread_t *mutex)
2	尝试加锁	int pthread_mutex_trylock(pthread_t *mutex)
3	解锁	int pthread_mutex_unlock(pthread_t *mutex)

• 参数

No.	参数	含义
1	mutex	互斥锁

• 示例

#include <stdio.h>
#include <pthread.h>
 
void* func(void* arg){
    pthread_mutex_lock(arg);
    printf("enter func\n");
    sleep(1);
    printf("do something\n");
    sleep(1);
    printf("level func\n");
    pthread_mutex_unlock(arg);
}
 
int main(int argc,int argv[]){
    pthread_mutex_t mutex;
    pthread_mutex_init(&mutex,NULL);
     
    pthread_t tids[3];
    int i;
    for(i=0;i<3;i++){
        pthread_create(&tids[i],NULL,func,&mutex);
    }
    for(i=0;i<3;i++){
        pthread_join(tids[i],NULL);
    }
    pthread_mutex_destroy(&mutex);
}

更加安全的做法

#include <stdio.h>
#include <pthread.h>
 
void* func(void* arg){
    pthread_cleanup_push(pthread_mutex_unlock,arg);
 
    pthread_mutex_lock(arg);
    printf("enter func\n");
    sleep(1);
    printf("do something\n");
    sleep(1);
    printf("level func\n");
    pthread_exit(0);
 
    pthread_cleanup_pop(0);
}
 
int main(int argc,int argv[]){
    pthread_mutex_t mutex;
    pthread_mutex_init(&mutex,NULL);
     
    pthread_t tids[3];
    int i;
    for(i=0;i<3;i++){
        pthread_create(&tids[i],NULL,func,&mutex);
    }
    for(i=0;i<3;i++){
        pthread_join(tids[i],NULL);
    }
    pthread_mutex_destroy(&mutex);
}

3 条件变量

• 概念
  线程挂起直到共享数据的某些条件得到满足

3.1 分类

No.	分类	实现	特点
1	静态分配条件变量	pthread_cond_t cond = PTHREAD_COND_INITIALIZER;	简单
2	动态分配静态变量	pthread_cond_init(&cond, NULL);pthread_cond_destroy(&cond);	可以设置更多的选项

3.2 操作

No.	操作	函数
1	条件等待	int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex)
2	计时等待	int pthread_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex, const struct timespec *abstime)
1	单个激活	int pthread_cond_signal(pthread_cond_t *cond)
2	全部激活	int pthread_cond_broadcast(pthread_cond_t *cond)

3.2.1 等待

3.2.1.1 条件等待

int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex)

No.	参数	含义
1	cond	条件变量
2	mutex	互斥锁

3.2.1.2 计时等待

int pthread_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex, const struct timespec *abstime)

No.	参数	含义
1	cond	条件变量
2	mutex	互斥锁
3	abstime	等待时间

• 返回值

No.	返回值	含义
1	ETIMEDOUT	超时结束等待

3.2.2 激活

3.2.2.1 单个激活

int pthread_cond_signal(pthread_cond_t *cond)

No.	参数	含义
1	cond	条件变量

• 返回值

No.	返回值	含义
1	0	成功
2	正数	错误码

3.2.2.2 全部激活

int pthread_cond_broadcast(pthread_cond_t *cond)

No.	参数	含义
1	cond	条件变量

• 返回值

No.	返回值	含义
1	0	成功
2	正数	错误码

套路

条件变量一般与互斥锁一起使用。

• 条件变量发送信号

pthread_mutex_lock(&mutex);
 
// do something
if(判断条件){
    pthread_cond_signal(&cond);// 唤醒单个
    // 或者
    pthread_cond_broadcast(&cond);// 唤醒多个
}
 
pthread_mutex_unlock(&mutex);

• 条件变量等待信号

pthread_mutex_lock(&mutex);
 
while(判断条件){
    pthread_cond_wait(&cond,&mutex);
}
// do something
// 把判断条件改为false
pthread_mutex_unlock(&mutex);

流程分析

• 主线程：等待子线程发信号。
• 子线程：每隔3秒计数一次，当数字是3的倍数时通知主进程。

#include <pthread.h>
#include <unistd.h>
#include <iostream>

using namespace std;

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
bool condition = false;

#define LOG(msg) cout << __func__<< ":" << msg << endl;

void* ChildFunc(void*){
    int i = 0;
    while(true){
        LOG("Enter");
        pthread_mutex_lock(&mutex);
        LOG("Get Lock");
        cout << ++i << endl;
        if(0 == i%3){
            condition = true;
            LOG("Begin Send Single");
            pthread_cond_signal(&cond);// 唤醒单个
            LOG("End Send Single");
        }
        sleep(3);
        pthread_mutex_unlock(&mutex);
        LOG("Lose Lock");
        LOG("Leave");
    }
}
void MainFunc(){
    while(true){
        LOG("Enter");
        pthread_mutex_lock(&mutex);
        LOG("Get Lock");
        while(!condition){
            LOG("Begin Wait Single");
            pthread_cond_wait(&cond,&mutex);
            LOG("End Wait Single");
        }
        condition = false;
        pthread_mutex_unlock(&mutex);
        LOG("Lose Lock");
        LOG("Leave");
    }
}

int main(){
    pthread_t tid;
    pthread_create(&tid,NULL,ChildFunc,NULL);
    MainFunc();
    pthread_join(tid,NULL);
    return 0;
}

问题：

• 主线程在pthread_cond_wait()时，是否释放互斥锁？
• 主线程在什么时候重新获得互斥锁？

案例

老板与会计的约定：每笔超过1000元的支出必须老板批准同意，低于1000元的会计可以自行决定。

#include <stdio.h>
#include <pthread.h>
#include <signal.h>
 
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
 
int currency = 0 ;
int signal_cnt = 0;
 
void* checkout(void* arg){
    sleep(1);
    for(;;){
        pthread_mutex_lock(&mutex);
        printf("checkout enter\n");
        currency = rand()%2000;     
        printf("spend %d\n",currency);
        if(currency >= 1000){
            printf("\033[42;31msignal boss:%d\033[0m\n");
            pthread_cond_signal(&cond);
            signal_cnt++;
        }
        printf("checkout leave\n");
        pthread_mutex_unlock(&mutex);
        //sched_yield();
        sleep(1);
    }
}
 
void* boss(void* arg){
    for(;;){
        pthread_mutex_lock(&mutex);
        printf("boss enter\n");
        while(currency < 1000){
            printf("boss wait\n");
            pthread_cond_wait(&cond,&mutex);
        }
        signal_cnt--;
        printf("\033[46;31mboss agress:%d signal_cnt:%d\033[0m\n",currency,signal_cnt);
        currency = 0;
        printf("boss leave\n");
        pthread_mutex_unlock(&mutex);
    }
}
 
int main(){
    typedef void*(*func_t)(void*);
    func_t funcs[2] = {boss,checkout};
    pthread_t tids[2];
    int i;
    pthread_setconcurrency(2);
    for(i=0;i<2;i++){
        pthread_create(&tids[i],NULL,funcs[i],tids);
    }
    for(i=0;i<2;i++){
        pthread_join(tids[i],NULL);
    }   
}

问题

• 互斥锁和条件变量能否不作为全局变量？

4. 读写锁

资源访问分为两种情况：读操作和写操作。

读写锁比mutex有更高的适用性，可以多个线程同时占用读模式的读写锁，但是只能一个线程占用写模式的读写锁。

1. 当读写锁是写加锁状态时，在这个锁被解锁之前，所有试图对这个锁加锁的线程都会被阻塞；
2. 当读写锁在读加锁状态时，所有试图以读模式对它进行加锁的线程都可以得到访问权，但是以写模式对它进行枷锁的线程将阻塞；
3. 当读写锁在读模式锁状态时，如果有另外线程试图以写模式加锁，读写锁通常会阻塞随后的读模式锁请求，这样可以避免读模式锁长期占用，而等待的写模式锁请求长期阻塞；
   这种锁适用对数据结构进行读的次数比写的次数多的情况下，因为可以进行读锁共享。

• 比喻

新闻发布会
领导发言与聊天

• 概念
  共享独占
  读取锁(共享)
  写入锁(独占)

4.1 分类

No.	分类	实现	特点
1	静态分配读写锁	pthread_rwlock_t rwlock = PTHREAD_RWLOCK_INITIALIZER	简单
2	动态分配读写锁	pthread_rwlock_init(&rwlock, NULL);pthread_rwlock_destroy(&rwlock);	可以设置更多的选项

4.2 操作

4.2.1 加锁

4.2.1.1 读取锁

int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_tryrdlock(pthread_rwlock_t *rwlock);

4.2.1.2 写入锁

int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_trywrlock(pthread_rwlock_t *rwlock);

4.2.2 解锁

int pthread_rwlock_unlock(pthread_rwlock_t *rwlock)

• 示例

火车票的查询与购买

#include <stdio.h>
#include <pthread.h>
 
pthread_rwlock_t rwlock = PTHREAD_RWLOCK_INITIALIZER;
int count = 10000;
int put_cur = 0;
void* search(void* arg){
    for(;;){
        pthread_rwlock_rdlock(&rwlock);
        printf("leave %d\n",count); 
        usleep(500000);
        pthread_rwlock_unlock(&rwlock);
    }
}
void rollback(void* arg){
    count -= put_cur;
    printf("rollback %d  to %d\n",put_cur,count);
    pthread_rwlock_unlock(&rwlock);
}
void* put(void* arg){
    pthread_cleanup_push(rollback,NULL);
    for(;;){
        pthread_rwlock_wrlock(&rwlock);
        put_cur = rand()%1000;
        count += put_cur;
        printf("put %d ok,leave %d\n",put_cur,count);
        usleep(500000);
        pthread_rwlock_unlock(&rwlock);
    }
    pthread_cleanup_pop(0);
}
void* hacker(void* arg){
    sleep(2);
    pthread_t* ptids = arg;
    pthread_cancel(ptids[0]);
    printf("\033[41;34mcancel %lu\033[0m\n",ptids[0]);
}
void* get(void* arg){
    for(;;){
        pthread_rwlock_wrlock(&rwlock);
        int cur = rand()%1000;
        if(count>cur){
            count -= cur;
            printf("crash %d leave %d\n",cur,count);
        }else{
            printf("leave not enought %d\n",count);
        }
        usleep(500000);
        pthread_rwlock_unlock(&rwlock);
    }
}
 
int main(){
    pthread_t tids[4];
    typedef void*(*func_t)(void*);
    func_t funcs[4] = {put,get,search,hacker};
    int i=0;
    pthread_setconcurrency(4);
    for(i=0;i<4;i++){
        pthread_create(&tids[i],NULL,funcs[i],tids);
    }
    for(i=0;i<4;i++){
        pthread_join(tids[i],NULL);
    }
}