Android C++系列：Linux线程（四）线程同步
多个线程同时访问共享数据时可能会冲突，这跟我们前面信号文章所说的可重入性是同样的问题。比如两个线程都要把某个全局变量增加1，这个操作在某平台需要三条指令完成:
从内存读变量值到寄存器；
寄存器的值加1；
将寄存器的值写回内存
假设两个线程在多处理器平台上同时执行这三条指令，则可能导致下图所示的结果，最后变量只加了一次而非两次。
实例：
#include
#include
#include
#define NLOOP 5000 int counter;
void *doit(void *);
/* incremented by threads */
int main(int argc, char **argv) {
pthread_t tidA, tidB;
pthread_create(&tidA, NULL, &doit, NULL);
pthread_create(&tidB, NULL, &doit, NULL);
/* wait for both threads to terminate */
pthread_join(tidA, NULL);
pthread_join(tidB, NULL);
return 0;
}
void *doit(void *vptr) {
int i, val;
for (i = 0; i < NLOOP; i++) {
val = counter;
printf("%x: %d\n", (unsigned int)pthread_self(), val + 1);
counter = val + 1;
}
return NULL;
}
我们创建两个线程，各自把counter增加5000次，正常情况下最后counter应该等于 10000，但事实上每次运行该程序的结果都不一样，有时候数到5000多，有时候数到6000 多。
1. 线程为什么要同步
2. 互斥量
mutex操作原语：
pthread_mutex_t
pthread_mutex_init
pthread_mutex_destroy
pthread_mutex_lock
pthread_mutex_trylock
pthread_mutex_unlock
2.1 临界区（Critical Section）
保证在某一时刻只有一个线程能访问数据的简便办法。在任意时刻只允许一个线程对共 享资源进行访问。如果有多个线程试图同时访问临界区，那么 在有一个线程进入后其他所有试图访问此临界区的线程将被挂起，并一直持续到进入临界区的线程离开。临界区在被释放后，其他线程可以继续抢占，并以此达到用原子方式操作共享资源的目的。
2.2 临界区的选定
临界区的选定因尽可能小，如果选定太大会影响程序的并行处理性能。
2.3 互斥量实例
#include
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
int pthread_mutex_destroy(pthread_mutex_t *mutex);
int pthread_mutex_init(pthread_mutex_t *restrict mutex, const pthread_mutexattr_t *restrict attr);
int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_trylock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);
实例：
#include
#include
#include
#define NLOOP 5000
int counter; /* incremented by threads */
pthread_mutex_t counter_mutex = PTHREAD_MUTEX_INITIALIZER;
void *doit(void *);
int main(int argc, char **argv) {
pthread_t tidA, tidB;
pthread_create(&tidA, NULL, doit, NULL);
pthread_create(&tidB, NULL, doit, NULL);
/* wait for both threads to terminate */
pthread_join(tidA, NULL);
pthread_join(tidB, NULL);
return 0;
}
void *doit(void *vptr) {
int i, val;
for (i = 0; i < NLOOP; i++) {
pthread_mutex_lock(&counter_mutex);
val = counter;
printf("%x: %d\n", (unsigned int)pthread_self(), val + 1);
counter = val + 1;
pthread_mutex_unlock(&counter_mutex);
}
return NULL;
}
这样运行结果就正常了，每次运行都能数到10000。
3. 死锁
4. 读写锁
读共享，写独占
pthread_rwlock_t
pthread_rwlock_init
pthread_rwlock_destroy
pthread_rwlock_rdlock
pthread_rwlock_wrlock
pthread_rwlock_tryrdlock
pthread_rwlock_trywrlock
pthread_rwlock_unlock
实例：
#include
#include
int counter;
pthread_rwlock_t rwlock; //3个线程不定时写同一全局资源，5个线程不定时读同一全局资源
void *th_write(void *arg)
{
int t;
while (1) {
pthread_rwlock_wrlock(&rwlock);
t = counter;
usleep(100);
printf("write %x : counter=%d ++counter=%d\n", (int)pthread_self(), t, ++counter);
pthread_rwlock_unlock(&rwlock);
usleep(100);
}
}
void *th_read(void *arg) {
while (1) {
pthread_rwlock_rdlock(&rwlock);
printf("read %x : %d\n", (int)pthread_self(), counter);
pthread_rwlock_unlock(&rwlock);
usleep(100);
}
}
int main(void) {
int i;
pthread_t tid[8];
pthread_rwlock_init(&rwlock, NULL);
for (i = 0; i < 3; i++)
pthread_create(&tid[i], NULL, th_write, NULL);
for (i = 0; i < 5; i++)
pthread_create(&tid[i+3], NULL, th_read, NULL);
pthread_rwlock_destroy(&rwlock);
for (i = 0; i < 8; i++)
pthread_join(tid[i], NULL);
return 0;
}
5. 条件变量
条件变量给多个线程提供了一个汇合的场所,条件变量控制原语:
pthread_cond_t
pthread_cond_init
pthread_cond_destroy
pthread_cond_wait
pthread_cond_timedwait
pthread_cond_signal
pthread_cond_broadcast
生产者消费者模型:
#include
#include
#include
struct msg {
struct msg *next;
int num;
};
struct msg *head;
pthread_cond_t has_product = PTHREAD_COND_INITIALIZER; pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
void *consumer(void *p) {
struct msg *mp;
for (;;) {
pthread_mutex_lock(&lock);
while (head == NULL)
pthread_cond_wait(&has_product, &lock);
mp = head;
head = mp->next;
pthread_mutex_unlock(&lock);
printf("Consume %d\n", mp->num); free(mp);
sleep(rand() % 5);
}
}
void *producer(void *p) {
struct msg *mp;
for (;;) {
mp = malloc(sizeof(struct msg));
mp->num = rand() % 1000 + 1;
printf("Produce %d\n", mp->num);
pthread_mutex_lock(&lock);
mp->next = head;
head = mp;
pthread_mutex_unlock(&lock);
pthread_cond_signal(&has_product);
sleep(rand() % 5);
}
}
int main(int argc, char *argv[]) {
pthread_t pid, cid;
srand(time(NULL));
pthread_create(&pid, NULL, producer, NULL);
pthread_create(&cid, NULL, consumer, NULL);
pthread_join(pid, NULL);
pthread_join(cid, NULL);
return 0;
}
6. 信号量
信号量控制原语
sem_t
sem_init
sem_wait
sem_trywait
sem_timedwait
sem_post
sem_destroy
生产者消费者实例：
#include
#include
#include
#include
#define NUM 5
int queue[NUM];
sem_t blank_number, product_number;
void *producer(void *arg) {
int p = 0;
while (1) {
sem_wait(&blank_number);
queue[p] = rand() % 1000 + 1;
printf("Produce %d\n", queue[p]);
sem_post(&product_number);
p = (p+1)%NUM;
sleep(rand()%5);
}
}
void *consumer(void *arg) {
int c = 0;
while (1) {
sem_wait(&product_number);
printf("Consume %d\n", queue[c]); queue[c] = 0; sem_post(&blank_number);
c = (c+1)%NUM;
sleep(rand()%5);
}
}
int main(int argc, char *argv[]) {
pthread_t pid, cid;
sem_init(&blank_number, 0, NUM);
sem_init(&product_number, 0, 0);
pthread_create(&pid, NULL, producer, NULL);
pthread_create(&cid, NULL, consumer, NULL);
pthread_join(pid, NULL);
pthread_join(cid, NULL);
sem_destroy(&blank_number);
sem_destroy(&product_number);
return 0;
}
7. 进程间锁
7.1 进程间pthread_mutex
#include
int pthread_mutexattr_init(pthread_mutexattr_t *attr);
int pthread_mutexattr_setpshared(pthread_mutexattr_t *attr, int pshared);
int pthread_mutexattr_destroy(pthread_mutexattr_t *attr);
pshared:
线程锁:PTHREAD_PROCESS_PRIVATE ;
进程锁:PTHREAD_PROCESS_SHARED；
默认情况是线程锁
实例：
#include
#include
#include
#include
#include
#include
#include
#include
struct mt {
int num;
pthread_mutex_t mutex;
pthread_mutexattr_t mutexattr;
};
int main(void) {
int fd, i;
struct mt *mm;
pid_t pid;
fd = open("mt_test", O_CREAT | O_RDWR, 0777);
/* 不需要write,文件里初始值为0 */
ftruncate(fd, sizeof(*mm));
mm = mmap(NULL, sizeof(*mm), PROT_READ|PROT_WRITE, MAP_SHARED, fd, 0);
close(fd);
memset(mm, 0, sizeof(*mm));
/* 初始化互斥对象属性 */
pthread_mutexattr_init(&mm->mutexattr);
/* 设置互斥对象为PTHREAD_PROCESS_SHARED共享，即可以在多个进程的线程访问,PTHREAD_PROCESS_PRIVATE 为同一进程的线程共享 */
pthread_mutexattr_setpshared(&mm->mutexattr,PTHREAD_PROCESS_SHARED);
pthread_mutex_init(&mm->mutex, &mm->mutexattr);
pid = fork();
if (pid == 0){
/* 加10次。相当于加10 */
for (i=0;i<10;i++){
pthread_mutex_lock(&mm->mutex);
(mm->num)++;
printf("num++:%d\n",mm->num);
pthread_mutex_unlock(&mm->mutex);
sleep(1);
}
}else if (pid > 0) {
/* 父进程完成x+2,加10次，相当于加20 */
for (i=0; i<10; i++){
pthread_mutex_lock(&mm->mutex);
mm->num += 2;
printf("num+=2:%d\n",mm->num);
pthread_mutex_unlock(&mm->mutex);
sleep(1);
}
wait(NULL);
}
pthread_mutex_destroy(&mm->mutex);
pthread_mutexattr_destroy(&mm->mutexattr);
/* 父子均需要释放 */
munmap(mm,sizeof(*mm));
unlink("mt_test");
return 0;
}
7.2 文件锁
使用fcntl提供文件锁
struct flock {
...
short l_type; /*Type of lock: F_RDLCK, F_WRLCK, F_UNLCK */
short l_whece; /* How to interpret l_start: SEEK_SET,SEEK_CUR,SEEK_END */
off_t l_start; /* Starting offset for lock*/
off_t l_len; /*Number of bytes to lock*/
pid_t l_pid; /* PID of process blocking our lock (F_GETLK only) */
实例：
#include
#include
#include
#include
#include
#include
void sys_err(char *str) {
perror(str);
exit(1);
}
int main(int argc, char *argv[]) {
int fd;
struct flock f_lock;
if (argc < 2) {
printf("./a.out filename\n");
exit(1);
}
if ((fd = open(argv[1], O_RDWR)) < 0)
sys_err("open");
//f_lock.l_type = F_WRLCK;
f_lock.l_type = F_RDLCK;
f_lock.l_whence = SEEK_SET;
f_lock.l_start = 0;
f_lock.l_len = 0; //0表示整个文件加锁
fcntl(fd, F_SETLKW, &f_lock);
printf("get flock\n");
sleep(10);
f_lock.l_type = F_UNLCK;
fcntl(fd, F_SETLKW, &f_lock);
printf("un flock\n");
close(fd);
return 0;
}
8. 总结
本文介绍了线程同步机制：为什么要同步、互斥量、死锁、读写锁、条件变量、信号量、进程间锁等概念与机制以及相关示例。