Something about sync
目前接触到的同步机制有如下：
- 互斥锁
- 条件变量
- 读写锁
- 信号量
- 自旋锁
- 屏障
- 原子操作
- 各类IPC机制（包括信号、管道、FIFO、socket、消息队列、共享内存）
接下来简单介绍几个同步机制，以下的同步机制，均基于共享内存模型实现
1. 互斥量
互斥量是最简单的同步机制，即互斥锁。多个进程(线程)均可以访问到一个互斥量，通过对互斥量加锁，从而来保护一个临界区,防止其它进程(线程)同时进入临界区，保护临界资源互斥访问。
对应POSIX 的API
// 初始化互斥锁
int pthread_mutex_init (pthread_mutex_t * restrict mutex , \
const pthread_mutexattr_t * restrict attr ) ;
int pthread_mutex_destroy (pthread_mutex_t * mutex ) ;
// 加锁
int pthread_mutex_lock (pthread_mutex_t *mutex) ;
// 解锁
int pthread_mutex_unlock (pthread_mutex_t *mutex) ;
// 尝试加锁
int pthread_mutex_trylock (pthread_mutex_t *mutex) ;
// 带超时的尝试加锁，防止死锁的一种方式
int pthread_mutex_timedlock (pthread_mutex_t * restrict mutex , \
const struct timespec * restrict abstime ) ;
2. 自旋锁
自旋锁相当于忙等。如果说互斥量是有智慧的等待，那么自旋锁就相当于死心塌地地苦苦追问。举一个不太恰当的例子：
你是一个宅男程序猿，你买了一件心爱的东西，快递上显示今天上午到，你焦急地等待，不停的询问快递小哥“到哪了？”“到哪了？”。
你是一个文艺青年，你买了一本书，快递显示今天上午到，然后你继续睡觉，你知道快递小哥会通知你，如果快递到了的话。
上面的例子一个描述自旋锁，一个表示互斥锁。所以简单理解它的执行流程就是：
- 自旋锁：不断尝试，是否可以获得锁，只要操作系统调度到它，他就去尝试是否能获取锁。
- 互斥锁：尝试一次没有获取到锁之后，就去休息了，然后等待外界（OS）告诉它可以获得锁，之后去加锁。
从调度的角度理解，对于自旋锁，由于需要一直检测锁是否可用，所以进程被放置到ready队列里待OS进行调度。对于mutex而言，由于一次尝试获取锁未成功，自己进入休眠，就被OS排队到阻塞队列中。等待有人唤醒。之后通过解锁进程/线程解除对互斥量的加锁，从而通过OS告知排在队列上的进程锁可用，把进程加入到ready队列中，等待调度。
相关自旋锁的实现，shui(杜炳阳)同学做了详细的分析。
对应POSIX实现的API
// 初始化自旋锁
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_unlock ( pthread_spinlock_t *lock ) ;
// 尝试加锁
int pthread_spin_trylock ( pthread_spinlock_t * lock ) ;
3. 条件变量
条件变量适合多个进程(线程)等待同一事件发生，然后去干某事。举一个简单的例子：
生产者和消费者模型：
多个消费者去等待生产者生产物品，消费者去消耗物品。当生产者生产出来一件物品时，便可以通知所有的消费者(当然也可以只通知其中一个等待的消费者)---可以去消耗物品了。这时多个消费者便去争抢物品，谁快谁拿到物品消耗。当物品被消耗完时，消费者就等待生产者。就类似于这样的场景。
条件变量必须配合互斥量一起工作。为什么？因为生产者生产出来的物品是临界资源，即所有进程和线程都可以使用的公共资源，则在一个时刻仅允许一个消费者去取。这时便使用互斥量去保护临界资源。
对应POSIX实现的API
// 初始化条件变量
int pthread_cond_init (pthread_cond_t * restrict cond , \
pthread_condattr_t * restrict attr ) ;
// 销毁条件变量
int pthread_cond_destroy ( pthread_cond_t * 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 *cond ) ;
// 通知所有在等待的进程或者线程锁可用
int pthread_cond_broadcast ( pthread_cond_t *cond ) ;
操作系统中的哲学家就餐问题，就可以通过条件变量解决。下面简单给下代码，毕竟这个问题大家都会。
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#include <signal.h>
#include <abort.h>
#include <errno.h>
#define PHILOSOPHER_COUNT 5
#define LEFT(p) ((p+4)%PHILOSOPHER_COUNT)
#define RIGHT(p) ((p+1)%PHILOSOPHER_COUNT)
enum {
EATING ,
THINKING ,
} ;
// Every ph has a cond
typedef struct {
int tid ;
pthread_cond_t cond ;
int status ;
} Philopher , *Philopher_p ;
// table is shared
pthread_mutex_t table_lock ;
Philopher_p ps;
void sig_handler (int signo) ;
void init_all_phil_status () {
ps = (Philopher_p) malloc (sizeof(Philopher)* PHILOSOPHER_COUNT) ;
if (NULL == ps ) {
fprintf(stderr , "Malloc error .No free space !\n") ;
abort () ;
}
// Init global lock
pthread_mutexattr_t lock_attr ;
pthread_mutexattr_init (&lock_attr) ;
pthread_mutexattr_settype ( &lock_attr , PTHREAD_MUTEX_ERRORCHECK ) ;
pthread_mutex_init (&table_lock , &lock_attr) ;
// Init every philosopher's status
for (int i = 0 ; i < PHILOSOPHER_COUNT ; i++ ) {
ps[i].status = THINKING ;
pthread_cond_init (&ps[i].cond , NULL) ;
}
// Regist sighandler
struct sigaction act , old_act ;
act.sa_handler = sig_handler ;
sigemptyset (&act.sa_mask) ;
act.sa_flags = 0 ;
sigaction (SIGINT , &act , &old_act ) ;
sigaction (SIGQUIT , &act , &old_act ) ;
printf ("Finish init !\n") ;
}
void philosopher_thinking (int index) {
// lock table ;
pthread_mutex_lock(&table_lock) ;
ps[index].status = THINKING ;
// judge left and right status
if (ps[LEFT(index)].status==THINKING) {
pthread_cond_signal (&ps[index].cond) ;
}
if ( ps[RIGHT(index)].status == THINKING) {
pthread_cond_signal(&ps[index].cond) ;
}
pthread_mutex_unlock (&table_lock) ;
sleep(1) ;
}
void philosopher_eating (int index) {
pthread_mutex_lock (&table_lock ) ;
// judge left and right status
if ((ps[LEFT(index)].status == THINKING) && (ps[RIGHT(index)].status == THINKING) ) {
ps[index].status = EATING ;
printf ("num %d philosopher is eating now\n" , ps[index].tid) ;
}
pthread_mutex_unlock (&table_lock) ;
sleep(1) ;
}
void * th_func (void * argu) {
long i = (long) argu ;
int index = (int)i ;
printf ("tid = %d\n" , index ) ;
ps[index].tid = index ;
while (1) {
philosopher_thinking (index) ;
philosopher_eating (index) ;
}
}
void destory_all_phol_status () {
free(ps) ;
printf ("Finish clear!\n") ;
}
void sig_handler (int signo) {
if ( SIGQUIT == signo ) {
printf ("Recv a SIGTERM signal , clear global var now ~\n") ;
} else if ( SIGTERM == signo ) {
printf ("Recv a SIGTERM signal , clear global var now ~\n") ;
}
destory_all_phol_status () ;
exit (0) ;
}
int main() {
atexit(destory_all_phol_status) ;
pthread_t tid ;
pthread_attr_t thread_attr ;
pthread_attr_init (&thread_attr) ;
pthread_attr_setdetachstate (&thread_attr, PTHREAD_CREATE_DETACHED) ;
init_all_phil_status() ;
for (int i = 0 ; i < 5 ; i++) {
pthread_create (&tid , &thread_attr , th_func , (void*)i) ;
}
pthread_attr_destroy (&thread_attr) ;
sleep (100) ;
return 0 ;
}
4. 读写锁
读写锁适合于使用在读操作多，写操作少的情况，比如数据库。读写锁读锁可以同时加很多，但是写锁是互斥的。当有进程或者线程要写时，必须等待所有的读进程或者线程都释放自己的读锁方可以写。数据库很多时候可能只是做一些查询。
POSIX对应的API
// 初始化读写锁
int pthread_rwlock_init ( pthread_rwlock_t * restrict rwlock , \
const pthread_rwlockattr_t * restrict attr ) ;
// 销毁读写锁
int pthread_rwlock_destroy (pthread_rwlock_t * 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_trywrlock ( pthread_rwlock_t * rwlock ) ;
// 带有超时的读写锁，避免死锁的一种方式
int pthread_rwlock_timedrdlock ( pthread_rwlock_t * restrict rwlock ,\
const struct timespec * restrict tsptr ) ;
int pthread_rwlock_timedwrlock ( pthread_rwlock_t * restrict rwlock , \
const struct timespec * restrict tsptr ) ;
我自己用的很少，似乎还没有场景需要我使用读写锁，平时一般的同步机制已经可以够用了，至于选择哪一种，必须要Benchmark以下，测试实际效率再进行抉择。
5. 屏障
屏障适合于，分配给多个工作线程一个大任务的一部分，最终需要一个线程去做最后的处理。举一个简单的例子，要排序1亿个数字，就可以使用多线程去做。开4个线程(或者更多，具体需要实际Benchmark，实测自己的机器上多少个线程时效率最快)。每个线程排序数据的1/4。最后，主线程做一一合并。具体的执行逻辑就是，多个工作线程，每个线程独立做一部分任务，完成了自己的任务时候，修改屏障计数，并自己去睡眠，不需要退出。当所有线程都到达屏障时，所有线程可以接着工作。
下面给出一个简单的多线程排序的例子，很简单，大家都会。
/// 多线程排序 ，同步使用屏障
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#include <string.h>
#include <strings.h>
#include <time.h>
#include <errno.h>
#include <sys/types.h>
#include <limits.h>
#include <sys/time.h>
#define THREAD_COUNT 4
#define NUM_COUNT 8000000L
#define PRE_THREAD_NUM_COUNT (NUM_COUNT/THREAD_COUNT)
typedef struct data {
long start ;
int id ;
} threadData , *threadDataP;
long *num ;
long *savedNum ;
// 声明屏障
pthread_barrier_t barrier ;
// 快排
void start_qsort ( long num[] , long start , long end ) {
long i = start , j = end ;
long base = num[start] ;
if ( i < j ) {
while (i < j ) {
while ( i < j && base <= num[j] ) {
j -- ;
}
if ( i < j ) {
num[i] = num[j] ;
}
while ( i < j && base > num[i] ) {
i ++ ;
}
if ( i < j ) {
num[j] = num[i] ;
}
}
num[i] = base ;
// 递归左
start_qsort ( num , start , i-1 ) ;
// 递归右
start_qsort (num , i+1 , end ) ;
}
}
void * thread_sort_part ( void * start_size ) {
// 分离出去线程
threadDataP tdp = (threadDataP) start_size ;
printf ("thread %d ready to sort !\n",tdp->id) ;
// 开始快排部分
start_qsort (num , tdp->start , tdp->start + PRE_THREAD_NUM_COUNT -1) ;
printf ("thread %d finished sort !\n" , tdp->id ) ;
pthread_barrier_wait (&barrier) ;
// Quit
pthread_exit ((void *)0) ;
}
void merge () {
// 记录每个块的开始位置
long threadBeginIndex [THREAD_COUNT] ;
// minidx 保存被选中的块（最小的那个）
// numIndex 保存数组中的数的索引
// tempNum 用来暂存找到的最小值
long i , minidx , numIndex , tempNum ;
// *******************************************************************************************
// 确保遍历遍历数组中所有的数
for ( i = 0 ; i < THREAD_COUNT ; ++i ) {
threadBeginIndex[i] = i * PRE_THREAD_NUM_COUNT ;
}
for ( numIndex = 0 ; numIndex < NUM_COUNT ; ++numIndex ) {
tempNum = LONG_MAX ;
// 这层循环的意思是找到所有的块中最小的那个块，把最小块的第一个元素放到最终的数组中去。然后让那个块的开头++
for ( i = 0 ; i < THREAD_COUNT ; i ++ ) {
if ( (threadBeginIndex[i] < (i+1)*PRE_THREAD_NUM_COUNT) && (num[threadBeginIndex[i]] < tempNum) ) {
tempNum = num[threadBeginIndex[i]] ;
minidx = i ;
}
}
savedNum[numIndex] = num[threadBeginIndex[minidx]] ;
threadBeginIndex[minidx]++ ;
}
printf ("Finished!\n") ;
}
int main () {
unsigned long i ;
struct timeval start , end ;
long long startusec , endusec ;
double elapsed ;
// errno
int err ;
pthread_t tid ;
pthread_attr_t thread_attr ;
// 线程数据
threadData td[THREAD_COUNT] ;
if ( NULL == (num = (long*) malloc (8*NUM_COUNT)) ) {
fprintf (stderr , "No free memory . Serious error !\n") ;
exit (1) ;
}
if ( NULL == (savedNum = (long*) malloc (8*NUM_COUNT) )) {
fprintf (stderr , "No free memory . Serious error !\n") ;
exit (1) ;
}
srandom (1) ;
for ( i = 0 ; i < NUM_COUNT ; ++i ) {
num[i] = random ()%10000000L ;
}
gettimeofday (&start , NULL ) ;
// 初始化屏障
pthread_barrier_init (&barrier , NULL , THREAD_COUNT+1) ;
// 初始化线程属性
pthread_attr_init ( &thread_attr ) ;
pthread_attr_setdetachstate ( &thread_attr , PTHREAD_CREATE_DETACHED ) ;
for ( i = 0 ; i < THREAD_COUNT ; ++i ) {
td[i].id = i ;
td[i].start = i * PRE_THREAD_NUM_COUNT ;
err = pthread_create (&tid , NULL , thread_sort_part , (void*)(&td[i])) ;
if ( 0 != err ) {
fprintf (stderr , "pthread_create error , serious error \n") ;
// 严重错误可以使用 long_jmp 跳转到指定清除程序或者函数处。 或者调一个统一处理的函数去。
exit (1) ;
}
}
// 主线程创建完协同线程后阻塞等待其排序完成。
pthread_barrier_wait (&barrier) ;
printf ("Begin merge !\n") ;
merge () ;
// 调用merge 合并所有分块
gettimeofday (&end , NULL) ;
// 计算排序使用的时间
startusec = start.tv_sec * 1000000 + start.tv_usec ;
endusec = end.tv_sec * 1000000 + end.tv_usec ;
elapsed = (double) (endusec - startusec ) /1000000.0 ;
printf ("sort tookk %.4f seconds\n" , elapsed) ;
// 清理操作
free ( num ) ;
free ( savedNum ) ;
pthread_barrier_destroy (&barrier) ;
pthread_attr_destroy ( &thread_attr ) ;
sleep (1) ;
return 0 ;
}
5. 信号量
在生产者消费者模型中，对任务数量的记录就可以使用信号量来做。可以理解为带计数的条件变量。当信号量的值小于0时，工作进程或者线程就会阻塞，等待物品到来。当生产者生产一个物品，会将信号量值加1操作。 这是会唤醒在信号量上阻塞的进程或者线程，它们去争抢物品。
POSIX 对应的API(匿名信号量)
// 初始化信号量
int sem_init ( sem_t * sem , int pshared , unsigned int value ) ;
// 销毁信号量
int sem_destroy ( sem_t * sem ) ;
// 信号量加1操作
int sem_post ( sem_t * sem ) ;
// 等待信号量值大于0 ， 并将信号量减1操作
int sem_wait (sem_t * sem ) ;
// 尝试等待
int sem_trywait ( sem_t * sem ) ;
运用信号量很容易实现生产者消费者模型。(但同时也存在问题，大家思考)
以下是一个简单的生产者消费者实现。
task_struct.h文件
#ifndef _TASK_STRUCT_H
#define _TASK_STRUCT_H
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <assert.h>
#include <pthread.h>
#include <semaphore.h>
#define BUFFER_SIZE 256
struct task_and_data {
void (*task_type_one) (const char *) ; //基本的打印工作
double (*task_type_two) (const char *) ; //计算传入的值
char task_buffer[BUFFER_SIZE] ; //任务缓冲区
int task_type ; //任务类型标志
} ;
typedef struct task_queue {
int thread_id ; //线程号
struct task_and_data task ; //任务结构
struct task_queue *next ; //指向下一个任务
} task_queue , *task_queue_p ;
typedef struct task_queue_head_rear { //队列的头和尾指针
task_queue_p head ;
task_queue_p rear ;
int task_count ; //任务计数
} task_queue_pointer ;
void print_string (const char * str ) ;
double calculate (const char * str ) ;
#endif
#include "task_struct.h"
pthread_mutex_t task_queue_lock ; //任务队列互斥
sem_t sem ; //申请信号量
task_queue_pointer task_queue_hr ; //申请任务队列头尾指针
void print_string (const char *str ) {
printf ("%s" , str ) ;
}
double calculate (const char * str ) {
double result = 1 ;
printf ("This expression is *** , now I will calculate it !\n") ;
return result ;
}
void init_sem_task_queue () {
pthread_mutex_init (&task_queue_lock , NULL) ;
sem_init (&sem, 0 , 0 ) ; //初始化信号量
task_queue_hr.rear = task_queue_hr.head = NULL ; //初始化任务队列为空
task_queue_hr.task_count = 0 ; //初始化任务数量为0
}
void * exec_th (void * data ) {
int tid = data ;
task_queue_p task_temp ;
struct task_and_data task ;
printf ("pthread %d waitting \n" , tid) ;
while (1) {
sem_wait (&sem) ; //等待信号到来
///锁住任务队列
pthread_mutex_lock (&task_queue_lock) ;
printf ("Tread %d is running \n" , data);
///当任务队列有任务时, 取出来, 执行
///取出队头指针
if (task_queue_hr.head != task_queue_hr.rear) {
task = (task_queue_hr.head)->task ;
task_temp = task_queue_hr.head ;
///修改队头指针
// printf ("task pointer value is %p \n" , task_temp) ;
///当当前任务不是最后一个任务时
task_queue_hr.head = (task_queue_hr.head)->next ;
free (task_temp) ;
} else {
if (task_queue_hr.head != NULL ) {
task = (task_queue_hr.head)->task ;
free (task_queue_hr.head) ;
task_queue_hr.head = task_queue_hr.rear = NULL ;
} else {
printf ("任务队列已空!\n") ;
}
}
///释放队列空间
///释放锁
pthread_mutex_unlock (&task_queue_lock) ;
switch (task.task_type) {
case 1 : (*(task.task_type_one))(task.task_buffer) ; break;
case 2 : break ;
}
}
}
void * task_th (void * data ) {
int count = 100 ;
task_queue_p temp_task_p ; //临时指向申请的任务
task_queue_p temp_task_before_p ; //指向先前一个节点
while (count) {
pthread_mutex_lock (&task_queue_lock) ; //锁住队列
temp_task_p = (task_queue_p)malloc (sizeof (task_queue)) ;
(temp_task_p ->task).task_type_one = print_string ;
sprintf (((temp_task_p ->task).task_buffer) , "%s%d%c","Hello this is a task " , count,'\n' ) ;
(temp_task_p ->task).task_type = 1 ; //表示执行打印函数
temp_task_p->next = NULL ;
///第一次插入时
if ((task_queue_hr.task_count) == 0) {
task_queue_hr.head = task_queue_hr.rear = temp_task_p ;
temp_task_before_p = temp_task_p ;
(task_queue_hr.task_count) ++ ;
} else {
///插入队列
task_queue_hr.rear = temp_task_p ;
temp_task_before_p ->next = temp_task_p ;
temp_task_before_p = temp_task_p ;
(task_queue_hr.task_count) ++ ;
}
count -- ; //添加任务次数减1
pthread_mutex_unlock (&task_queue_lock) ;
sem_post (&sem) ; //添加信号量
}
printf ("任务线程放置所有任务完毕\n") ;
pthread_exit (NULL) ;
}
int main (int argc , char* argv[]) {
pthread_t exec_th1 , exec_th2 , exec_th3 , task_thread ;
init_sem_task_queue () ; //初始化信号量和互斥量
pthread_create (&exec_th1 , NULL ,exec_th , (void*)1 ) ;
pthread_detach (exec_th1) ;
pthread_create (&exec_th2 , NULL ,exec_th ,(void*) 2 ) ;
pthread_detach (exec_th2) ;
pthread_create (&exec_th3 , NULL ,exec_th , (void*)3 ) ;
pthread_detach (exec_th3) ;
pthread_create (&task_thread , NULL , task_th , (void*)4 ) ;
pthread_detach (task_thread) ;
sleep (10000) ;
return 0 ;
}
各类进程间通信
可以使用各类进程间通信来完成同步操作。
- 信号(异步)
- 文件
- 管道
- Socket
- 共享内存
- 消息队列(个人用的比较少)
原子操作
CPU指令级别支持的，C语言支持一整套原子操作如下：
详细理解点击
旧版本的原子操作
static __inline__ void atomic_add(int i, atomic_t *v) ;
static __inline__ void atomic_sub(int i, atomic_t *v) ;
static __inline__ int atomic_sub_and_test(int i, atomic_t *v) ;
static __inline__ void atomic_inc(atomic_t *v) ;
static __inline__ void atomic_dec(atomic_t *v) ;
static __inline__ int atomic_dec_and_test(atomic_t *v) ;
... ...
新版本的原子操作
void __sync_add_and_fetch((x),1)
void __sync_sub_and_fetch((x),1)
void __sync_add_and_fetch((x),(y))
void __sync_sub_and_fetch((x),(y))
简单看一个函数
#include <stdio.h>
#include <stdlib.h>
static int count = 0 ;
int main() {
__sync_fetch_and_add (&count , 1 ) ;
__sync_fetch_and_sub (&count , 1) ;
return 0;
}
反汇编后看到的：
00000000004004f6 <main>:
4004f6: 55 push %rbp
4004f7: 48 89 e5 mov %rsp,%rbp
4004fa: f0 83 05 2e 0b 20 00 lock addl $0x1,0x200b2e(%rip) # 601030 <__TMC_END__>
400501: 01
400502: f0 83 2d 26 0b 20 00 lock subl $0x1,0x200b26(%rip) # 601030 <__TMC_END__>
400509: 01
40050a: b8 00 00 00 00 mov $0x0,%eax
40050f: 5d pop %rbp
400510: c3 retq
400511: 66 2e 0f 1f 84 00 00 nopw %cs:0x0(%rax,%rax,1)
400518: 00 00 00
40051b: 0f 1f 44 00 00 nopl 0x0(%rax,%rax,1)
指令是怎么实现的？
参考上述链接。
