APUE学习之多线程编程（二）：线程同步，apue多线程编程
为了保证临界资源的安全性和可靠性，线程不得不使用锁，同一时间只允许一个或几个线程访问变量。常用的锁有互斥量，读写锁，条件变量
一、互斥量
互斥量是用pthread_mutex_t数据类型表示的，在使用之前，必须对其进行初始化，可以把它设置为PTHREAD_MUTEX_INITIALIZER（只适于静态分配的互斥量），也可以通过调用pthread_mutex_init函数进行初始化，最后还要调用pthread_mutex_destroy进行释放。
#include <pthread.h>
int pthread_mutex_init(pthread_mutex_t *restrict mutex, const pthread_mutexattr_t *restrict attr);
int pthread_mutex_destroy(pthread_mutex_t *mutex);
要用默认的属性初始化互斥量，只需把attr设为NULL，后面在讨论互斥量属性。
对互斥量进行加锁，使用pthread_mutex_lock，如果互斥量已经上锁，调用线程将阻塞至互斥量解锁，对互斥量解锁，使用pthread_mutex_unlock，如果线程不希望被阻塞，它可以调用pthread_mutex_trylock尝试对互斥量进行加锁，如果互斥量未锁住，则成功加锁，如果互斥量已锁住，pthread_mutex_trylock就会失败，返回EBUSY。
#include <pthread.h>
int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);
int pthread_mutex_trylock(pthread_mutex_t *mutex);
例子：
![]()
#include <stdio.h>
#include <pthread.h>
struct foo
{
int f_count;
pthread_mutex_t f_lock;
int f_id;
};
struct foo * foo_alloc(
int id)
{
struct foo *fp =
NULL;
if ((fp =
malloc(
sizeof(
struct foo))) !=
NULL)
{
fp->f_count =
1;
fp->f_id =
id;
if (pthread_mutex_init(&fp->f_lock, NULL) !=
0)
{
free(fp);
return NULL;
}
}
return fp;
}
void foo_hold(
struct foo *
fp)
{
pthread_mutex_lock(&fp->
f_lock);
fp->f_count++
;
pthread_mutex_unlock(&fp->
f_lock);
}
void foo_rele(
struct foo *
fp)
{
pthread_mutex_lock(&fp->
f_lock);
if (--fp->f_count ==
0)
{
pthread_mutex_unlock(&fp->
f_lock);
pthread_mutex_destroy(&fp->
f_lock);
free(fp);
}
else
{
pthread_mutex_unlock(&fp->
f_lock);
}
}
View Code
上面的例子描述了用于保护某个数据结构的互斥量，我们在对象中嵌入引用计数，确保在所有使用该对象的线程完成数据访问之前，该对象的内存空间不会被释放。
如果线程对同一个互斥量加锁两次，那么它自身将陷入死锁状态。如果有一个以上的互斥量，且允许一个线程一直占有第一个互斥量，并且试图锁住第二个互斥量时处于阻塞状态，但是拥有第二个互斥量的线程也在试图锁住第一个互斥量，也阻塞，就死锁了。
可以通过仔细控制互斥量加锁的顺序来避免死锁的发生，譬如要求所有线程必须先锁住互斥量A才能锁住互斥量B。另一种办法是当线程无法获得下一个互斥量的时候，就释放自己已占有的互斥量，过一段时间再试。
例子：
![]()
#include
"apue.h"
#include <pthread.h>
#define NMASH 29
#define HASH(id) (((unsigned long)id) % NMASH)
struct foo *
fh[NMASH];
pthread_mutex_t hashlock =
PTHREAD_MUTEX_INITIALIZER;
struct foo
{
int f_count;
pthread_mutex_t f_lock;
int f_id;
struct foo *
f_next;
};
struct foo *foo_alloc(
int id)
{
struct foo *fp =
NULL;
int idx =
0;
if ((fp =
malloc(
sizeof(
struct foo))) !=
NULL)
{
fp->f_count =
1;
fp->f_id =
if;
if (pthread_mutex_init(&fp->f_lock, NULL) !=
0)
{
free(fp);
return NULL;
}
idx =
HASH(id);
pthread_mutex_lock(&
hashlock);
fp->f_next =
fh[idx];
fh[idx] =
fp;
pthread_mutex_lock(&fp->
f_lock);
pthread_mutex_unlock(&
hashlock);
pthread_mutex_unlock(&fp->
f_lock);
}
return fp;
}
void foo_hold(
struct foo *
fp)
{
pthread_mutex_lock(&fp->
f_lock);
fp->f_count++
;
pthread_mutex_unlock(&fp->
f_lock);
}
struct foo *foo_find(
int id)
{
struct foo *fp =
NULL;
pthread_mutex_lock(&
hashlock);
for (fp = fh[HASH(id)]; fp != NULL; fp = fp->
next)
{
if (fp->f_id =
id)
{
foo_hold(fp);
break;
}
}
pthread_mutex_unlock(&
hashlock);
return fp;
}
void foo_rele(
struct foo *
fp)
{
struct foo *tfp =
NULL;
int idx =
0;
pthread_mutex_lock(&fp->
f_lock);
if (fp->f_count ==
1)
{
pthread_mutex_unlock(&fp->
f_lock);
pthread_mutex_lock(&
hashlock);
pthread_mutex_lock(&fp->
f_lock);
if (fp->f_count !=
1)
{
fp->f_count--
;
pthread_mutex_unlock(&
hashlock);
pthread_mutex_unlock(&fp->
f_lock);
return;
}
idx = HASH(fp->
f_id);
tfp =
fh[idx];
if (tfp =
fp)
{
fh[idx] = fp->
f_next
}
else
{
while(tfp->next !=
fp)
{
tfp = tfp->
next;
}
tfp->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);
}
}
View Code
这个例子比上一个例子多了一个散列表和一个保护散列表的互斥量，加锁的顺序是先hashlock，再f_lock，注意这个顺序，就不会发生死锁，不过这样也导致代码太繁琐，最后一个函数解锁f_lock后重新加锁f_lock，需要重新考察f_count的值，因为可能在这期间被其他线程修改。
这样的方式太复杂，让hashlock也保护f_cout，事情会简单很多。
例子：
![]()
#include
"apue.h"
#include <pthread.h>
#define NMASH 29
#define HASH(id) (((unsigned long)id) % NMASH)
struct foo *
fh[NMASH];
pthread_mutex_t hashlock =
PTHREAD_MUTEX_INITIALIZER;
struct foo
{
int f_count;
pthread_mutex_t f_lock;
int f_id;
struct foo *
f_next;
};
struct foo *foo_alloc(
int id)
{
struct foo *fp =
NULL;
int idx =
0;
if ((fp =
malloc(
sizeof(
struct foo))) !=
NULL)
{
fp->f_count =
1;
fp->f_id =
if;
if (pthread_mutex_init(&fp->f_lock, NULL) !=
0)
{
free(fp);
return NULL;
}
idx =
HASH(id);
pthread_mutex_lock(&
hashlock);
fp->f_next =
fh[idx];
fh[idx] =
fp;
pthread_mutex_lock(&fp->
f_lock);
pthread_mutex_unlock(&
hashlock);
pthread_mutex_unlock(&fp->
f_lock);
}
return fp;
}
void foo_hold(
struct foo *
fp)
{
pthread_mutex_lock(&
hashlock);
fp->f_count++
;
pthread_mutex_unlock(&
hashlock);
}
struct foo *foo_find(
int id)
{
struct foo *fp =
NULL;
pthread_mutex_lock(&
hashlock);
for (fp = fh[HASH(id)]; fp != NULL; fp = fp->
next)
{
if (fp->f_id =
id)
{
foo_hold(fp);
break;
}
}
pthread_mutex_unlock(&
hashlock);
return fp;
}
void foo_rele(
struct foo *
fp)
{
struct foo *tfp =
NULL;
int idx =
0;
pthread_mutex_lock(&
hashlock);
if (fp->f_count ==
1)
{
idx = HASH(fp->
f_id);
tfp =
fh[idx];
if (tfp =
fp)
{
fh[idx] = fp->
f_next
}
else
{
while(tfp->next !=
fp)
{
tfp = tfp->
next;
}
tfp->next = fp->
f_next;
}
pthread_mutex_unlock(&
hashlock);
pthread_mutex_destroy(&fp->
f_lock);
free(fp);
}
else
{
fp->f_count--
;
pthread_mutex_unlock(&
hashlock);
}
}
View Code
当线程试图获取一个已加锁的互斥量时，pthread_mutex_timedlock互斥量原语允许绑定线程阻塞时间。pthread_mutex_timedlock和pthread_mutex_lock是基本等价的，但是达到超时时间后，pthread_mutex_timedlock会返回。超时时间指原意等待的绝对时间。这个超时时间是用timespec来表示的
#include <pthread.h>
#include <time.h>
int pthread_mutex_timedlock(pthread_mutex_t *restrict mutex, const struct timespec *restrict tsptr);
二、读写锁
读写锁与互斥量相似，不过读写锁允许更高的并行性，一次只有一个线程可以占有写模式的读写锁，但是多个线程可以同时占有读模式的读写锁，简单地来说，就说支持一个写者，多个读者。
当读写锁是写加锁状态时，所以试图对这个锁加锁的线程都会被阻塞，当读写锁在读加锁状态时，所以试图以读模式对它进行加锁的线程都可以得到访问权，但是希望以写模式加锁的线程会被阻塞。不过当有一个线程企图以写模式获取锁时，读写锁会阻塞后面的读模式锁请求，防止读模式锁长期占用。
可知，读写锁适用于对数据结构读的次数远大于写的情况，又称共享互斥锁，读共享，写互斥。
#include <pthread.h>
int pthread_rwlock_init(pthread_rwlock_init(pthread_rwlock_t *restrict rwlock, const pthread_rwlockattr_t *restrict attr);
int pthread_rwlock_destroy(pthread_rwlock_t *rwlock);
读写锁调用phtread_rwlock_init进行初始化，如果希望读写锁有默认的属性，传null给attr即可。
读的模式下锁定读写锁，需要调用phtread_rwlock_rdlock，写的模式下锁定读写锁，需要调用pthread_rwlock_wrlock，不过以何种方式锁定读写锁，都可以调用pthread_rwlock_unlock解锁。
#include <pthread.h>
int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_unlock(pthread_rwlock_t *rwlock);
例子：
![]()
#include <stdio.h>
#include <pthread.h>
struct job
{
struct job *
j_next;
struct job *
j_prev;
pthread_t j_id;
};
struct queue
{
struct job *
q_head;
struct job *
q_tail;
pthread_rwlock_t q_lock;
};
int queue_init(
struct queue *
qp)
{
int err;
qp->q_head =
NULL;
qp->q_tail =
NULL;
err = pthread_rwlock_init(&qb->
q_lock, NULL);
if (err !=
0)
{
return err;
}
return 0
}
void job_insert(
struct queue *qp,
struct job *
jp)
{
pthread_rwlock_wrlock(&qb->
q_lock);
jp->next = qp->
head;
jp->j_prev =
NULL;
if (qp->q_head !=
NULL)
{
qp->q_head->j_prev =
jp;
}
else
{
qp->tail =
jp;
}
qp->head =
jp;
pthread_rwlock_unlock(&qp->
q_lock);
}
void job_append(
struct queue *qp,
struct job *
jp)
{
pthread_rwlock_wrlock(&qp->
q_lock);
jp->j_next =
NULL;
jp->j_prev = qp->
tail;
if (qp->q_tail !=
NULL)
{
qp->q_tail->j_next =
jp;
}
qp->q_tail =
jp;
pthread_rwlock_unlock(&qp->
q_lock);
}
void job_remove(
struct queue *qp,
struct job *
jp)
{
pthread_rwlock_wrlock(&qp->
q_lock);
if (jp == qp->
q_head)
{
qp->q_head = jp->
j_next;
if (qp->q_tail ==
jp)
{
qp->tail =
NULL;
}
else
{
jp->next->j_prev = jp->
j_prev;
}
}
else if (jp == qp->
q_tail)
{
qp->q_tail = jp->
j_prev;
jp->j_prev->j_next =
NULL;
}
else
{
jp->j_prev->j_next = jp->
j_next;
jp->j_next->j_prev = jp->
j_prev;
}
pthread_rwlock_unlock(&qp->
q_lock);
}
struct job *job_find(
struct queue *
qp, pthread_t id)
{
struct job *
jp;
if (pthread_rwlock_rdlock(&qp->q_lock) !=
0)
{
return NULL;
}
for (jp = qb->q_head; jp != NULL; jp = jp->
j_next)
{
if (pthread_equal(jp->
j_id, id))
{
break;
}
}
pthread_rwlock_unlock(&qp->
q_lock);
return jp;
}
View Code
与互斥量一样，读写锁也有带超时的读写锁函数，避免陷入永久的阻塞。
#include <pthread.h>
#include <time.h>
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);
三、条件变量
条件变量与互斥量一起使用时，允许线程以无竞争的方式等待特定的条件发生。
条件本身由互斥量保护，线程在改变条件状态之前必须锁定互斥量。在使用条件变量之前，必须把它初始化，可以把常量PTHREAD_CON_INITIALIZE赋给静态分配的条件变量，也可用pthread_cond_init函数进行初始化。使用pthread_cond_destroy释放。
#include <pthread.h>
int pthread_cond_init(pthread_cond_t *restrict cond, const pthread_condattr_t *restrict attr);
int pthread_cond_destroy(pthread_con_t *cond);
如果需要一个默认属性的条件变量，把null给attr即可。
我们使用pthread_cond_wait等待条件变量为真，如果在给定时间内不能满足，则返回错误码。
#include<pthread.h>
int pthread_cond_wait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex)
int pthread_cond_timedwait(pthread_cond_t *restrict cond, phtread_mutex_t *restrict mutex, const struct timespec *restrict tsptr)
调用者把锁定的互斥量传给函数，函数自动把调用线程放到等待条件的线程列表上，对互斥量解锁，当pthread_cond_wait返回时，互斥量再次被锁住。pthread_cond_timedwait多了原意等待的时间。
有两个函数可用于通知线程条件已满足，pthread_cond_signal函数至少唤醒一个，pthread_cond_broadcast唤醒等待该条件的所有线程。
#include<phtread.h>
int pthread_cond_signal(pthread_cond_t *cond)
int pthread_cond_broadcast(pthread_cond_t *cond)
例子：
![]()
#include <pthread.h>
struct msg
{
struct msg *
m_next;
};
struct msg *
workq;
pthread_cond_t qready =
PTHREAD_COND_INITIALIZER;
pthread_mutex_t qlock =
PTHREAD_MUTEX_INITIALIZER;
void process_msg(
void)
{
struct msg *
mp;
for(;;)
{
pthread_mutex_lock(&
qlock);
while (workq ==
NULL)
{
pthread_cond_wait(&qready, &
qlock);
}
mp =
workq;
workq = mp->
m_next;
pthread_mutex_unlock(&
qlock);
}
}
void enqueue_msg(
struct msg *
mp)
{
pthread_mutex_lock(&
qlock);
mp->m_next =
workq;
workq =
mp;
pthread_mutex_unlock(&
qlock);
pthread_cond_signal(&
qready);
}
View Code
评论暂时关闭