借助互斥量和条件变量实现读写锁


当CPU的发展不在遵循摩尔定律，并逐渐迈向多核时代的时候，为了充分发挥多核CPU的性能，多线程编程技术也应运而生。在多线程编程的过程中，线程同步机制的设计，会直接影响到程序的并发性能。

Posix标准为我们提供了多种线程间的同步方式，有：互斥量、信号量、信号灯等等。每种同步机制都有其特定的属性进而决定了它们不同的适用场合，本文将介绍如何借助互斥量和信号量自己实现一个读者优先的读写锁。

为什么会有读写锁？相比于普通锁，读写锁可以满足我们更多的需求。比如：可以保证一块数据能够被并发读，独占写。这对于读请求远大于写请求的多线程程序来说，可以提高程序的并发性能。下面先看一下读写锁的结构：

typedef struct rwlock_tag {

pthread_mutex_t mutex; /*保护读写锁的互斥量*/

pthread_cond_t read; /* （模拟读锁）*/

pthread_cond_t write; /* （模拟写锁）*/

int valid; /* 用来检测读写锁是否被正确初始化的标志*/

int r_active; /* 当前获取读锁的线程数*/

int w_active; /* 当前获取写锁的线程数 */

int r_wait; /* 当前正在等待的读线程数量*/

int w_wait; /* 当前正在等待的写线程数量*/

} rwlock;

下面是程序的代码：

rwlock.h

#include <pthread.h>

typedef struct rwlock_tag {
pthread_mutex_t     mutex;
pthread_cond_t      read;           /* wait for read */
pthread_cond_t      write;          /* wait for write */
int                 valid;          /* set when valid */
int                 r_active;       /* readers active */
int                 w_active;       /* writer active */
int                 r_wait;         /* readers waiting */
int                 w_wait;         /* writers waiting */
} rwlock_t;

#define RWLOCK_VALID 0xfacade

#define RWL_INITIALIZER \
{ PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER, \
PTHREAD_COND_INITIALIZER, RWLOCK_VALID, 0, 0, 0, 0}

extern int rwl_init (rwlock_t *rwl); //初始化读写锁
extern int rwl_destroy (rwlock_t *rwl); //销毁读写锁
extern int rwl_readlock (rwlock_t *rwl); //加读锁
extern int rwl_readtrylock (rwlock_t *rwl); //trylock
extern int rwl_readunlock (rwlock_t *rwl); //释放读锁
extern int rwl_writelock (rwlock_t *rwl); //加写锁
extern int rwl_writetrylock (rwlock_t *rwl); //trylock
extern int rwl_writeunlock (rwlock_t *rwl); //释放写锁

rwlock.cpp

<pre class="cpp" name="code">#include <pthread.h>
#include "rwlock.h"
#include "errors.h"

int rwl_init (rwlock_t *rwl)
{
int status;

rwl->r_active = 0;
rwl->r_wait = rwl->w_wait = 0;
rwl->w_active = 0;

status = pthread_mutex_init(&rwl->mutex, NULL);
if(status != 0)
{
return status;
}

status = pthread_cond_init(&rwl->read, NULL);
if(status != 0)
{
pthread_mutex_destroy(&rwl->mutex);
return status;
}

status = pthread_cond_init(&rwl->write, NULL);
if(status != 0)
{
pthread_cond_destroy(&rwl->read);
pthread_mutex_destroy(&rwl->mutex);
return status;
}
rwl->valid = RWLOCK_VALID; /*设置魔方数字标志*/

return 0;
}

int rwl_destroy (rwlock_t *rwl)
{
int status, status1, status2;

/*先判断读写锁是否被正确初始化*/
if (rwl->valid != RWLOCK_VALID)
return EINVAL;

status = pthread_mutex_lock (&rwl->mutex);
if (status != 0)
return status;

/*确保没有正在占有的读/写线程，如果有，返回busy*/
if(rwl->r_active > 0 || rwl->w_active)
{
pthread_mutex_unlock (&rwl->mutex);
return EBUSY;
}

/*确保没有正在等待的读/写线程，如果有，返回busy*/
if(rwl->r_wait != 0 || rwl->w_wait != 0)
{
pthread_mutex_unlock (&rwl->mutex);
return EBUSY;
}

rwl->valid = 0;
status = pthread_mutex_unlock (&rwl->mutex);
if (status != 0)
return status;

status = pthread_mutex_destroy (&rwl->mutex);
status1 = pthread_cond_destroy (&rwl->read);
status2 = pthread_cond_destroy (&rwl->write);

return (status == 0 ? status : (status1 == 0 ? status1 : status2));
}

/*读线程销毁时的清理函数，避免销毁一个正在等待的读线程时，使程序进入无限等待*/
static void rwl_readcleanup (void *arg)
{
rwlock_t *rwl = (rwlock_t *)arg;

rwl->r_wait--;
pthread_mutex_unlock (&rwl->mutex);
}

int rwl_readlock (rwlock_t *rwl)
{
int status;

if (rwl->valid != RWLOCK_VALID)
return EINVAL;

status = pthread_mutex_lock (&rwl->mutex);
if (status != 0)
return status;

/*如果有写线程正在占有，则进入等待*/
if(rwl->w_active)
{
rwl->r_wait++;
pthread_cleanup_push(rwl_readcleanup, (void*)rwl); //设置线程取消时的清理函数

while(rwl->w_active)
{
status = pthread_cond_wait(&rwl->read, &rwl->mutex);
if (status != 0)
break;
}
pthread_cleanup_pop(0);
rwl->r_wait--;
}
if (status == 0)
rwl->r_active++; //增加读占有线程数目

pthread_mutex_unlock (&rwl->mutex);

return status;
}

int rwl_readtrylock (rwlock_t *rwl)
{
int status, status2;

if (rwl->valid != RWLOCK_VALID)
return EINVAL;

status = pthread_mutex_lock (&rwl->mutex);
if (status != 0)
return status;

/*主要判断是否有写线程正在占有写锁*/
if(rwl->w_active)
status = EBUSY;
else
rwl->r_active++;

status = pthread_mutex_unlock(&rwl->mutex);

return (status2 != 0 ? status2 : status);
}

int rwl_readunlock (rwlock_t *rwl)
{
int status, status2;

if (rwl->valid != RWLOCK_VALID)
return EINVAL;

status = pthread_mutex_lock (&rwl->mutex);
if (status != 0)
return status;

/*读占有线程数目减少，如果此时已经没有读线程占有和等待，同时存在等待的写线程，则唤醒一个写线程*/
rwl->r_active--;
if(rwl->r_active == 0 && rwl->w_wait > 0)
status = pthread_cond_signal(&rwl->write);

status2 = pthread_mutex_unlock (&rwl->mutex);

return (status2 == 0 ? status : status2);
}

/*写线程销毁时的清理函数，避免销毁一个正在等待的写线程时，使程序进入无限等待*/
static void rwl_writecleanup (void *arg)
{
rwlock_t *rwl = (rwlock_t *)arg;

rwl->w_wait--;
pthread_mutex_unlock (&rwl->mutex);
}

int rwl_writelock (rwlock_t *rwl)
{
int status;

if (rwl->valid != RWLOCK_VALID)
return EINVAL;

status = pthread_mutex_lock (&rwl->mutex);
if (status != 0)
return status;

/*如果存在读占有线程或写占有线程，则进入等待*/
if(rwl->w_active || rwl->r_active > 0)
{
rwl->w_wait++;
pthread_cleanup_push(rwl_writecleanup, (void*)rwl); //设置线程取消时的清理函数
while(rwl->w_active || rwl->r_active > 0)
{
status = pthread_cond_wait(&rwl->write, &rwl->mutex);
if(status != 0)
break;
}
pthread_cleanup_pop(0);
rwl->w_wait--;
}

if(status == 0)
{
rwl->w_active = 1;
}
pthread_mutex_unlock (&rwl->mutex);

return status;
}

int rwl_writetrylock (rwlock_t *rwl)
{
int status, status2;

if (rwl->valid != RWLOCK_VALID)
return EINVAL;

status = pthread_mutex_lock (&rwl->mutex);
if (status != 0)
return status;

/*主要检测是否存在读/写占有线程*/
if(rwl->w_active || rwl->r_active > 0)
status = EBUSY;
else
rwl->w_active = 1;

status2 = pthread_mutex_unlock (&rwl->mutex);

return (status != 0 ? status : status2);
}

int rwl_writeunlock (rwlock_t *rwl)
{
int status;

if (rwl->valid != RWLOCK_VALID)
return EINVAL;

status = pthread_mutex_lock (&rwl->mutex);
if (status != 0)
return status;

/*由于是读者优先，所以首先检测是否存在读等待线程，有则唤醒*/
rwl->w_active = 0;
if(rwl->r_wait > 0)
{
status = pthread_cond_broadcast(&rwl->read);
if (status != 0)
{
pthread_mutex_unlock (&rwl->mutex);
return status;
}
}
else if(rwl->w_wait > 0) /*如果没有读等待线程，同时存在写等待线程则唤醒一个写等待线程*/
{
status = pthread_cond_signal(&rwl->write);
if(status != 0)
{
pthread_mutex_unlock(&rwl->mutex);
return status;
}
}
status = pthread_mutex_unlock(&rwl->mutex);

return status;
}

由于篇幅原因，测试用例便不再帖了。