APR源代码分析-线程同步篇

在线程同步方面，Posix标准定义了3种同步模型，分别为互斥量、条件变量和读写锁。APR也“浅”封装了这3种模型，只是在“读写锁”一块儿还没有全部完成。

线程同步的源代码的位置在$(APR_HOME)/locks目录下，本篇blog着重分析unix子目录下的thread_mutex.c、thread_rwlock.c和thread_cond.c文件的内容，其相应头文件为(APR_HOME)/include/apr_thread_mutex.h、apr_thread_rwlock.h和apr_thread_cond.h。

由于APR的封装过于“浅显”，实际上也并没有多少值得分析的“靓点”。所以本篇实际上是在讨论线程同步的3种运行模型。

一、互斥量
互斥量是线程同步中最基本的同步方式。互斥量用于保护代码中的临界区，以保证在任一时刻只有一个线程或进程访问临界区。

1、互斥量的初始化
在POSIX Thread中提供两种互斥量的初始化方式，如下：
(1) 静态初始化
互斥量首先是一个变量，Pthread提供预定义的值来支持互斥量的静态初始化。举例如下：
static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
既然是静态初始化，那么必然要求上面的mutex变量需要静态分配。在APR中并不支持apr_thread_mutex_t的使用预定值的静态初始化(但可以变通的利用下面的方式进行静态分配的mutex的初始化)。

(2) 动态初始化
除了上面的情况，如果mutex变量在堆上或在共享内存中分配的话，我们就需要调用一个初始化函数来动态初始化该变量了。在Pthread中的对应接口为pthread_mutex_init。APR封装了这一接口，我们可以使用下面方式在APR中初始化一个apr_thread_mutex_t变量。
        apr_thread_mutex_t *mutex = NULL;
        apr_pool_t  *pool = NULL;
        apr_status_t  stat;

        stat = apr_pool_create(&pool, NULL);
        if (stat != APR_SUCCESS) {
                printf("error in pool %d\n", stat);
        } else {
                printf("ok in pool\n");
        }

        stat = apr_thread_mutex_create(&mutex, APR_THREAD_MUTEX_DEFAULT, pool);
        if (stat != APR_SUCCESS) {
                printf("error %d in mutex\n", stat);
        } else {
                printf("ok in mutex\n");
        }

2、互斥锁的软弱性所在
互斥锁之软弱性在于其是一种协作性锁，其运作时对各线程有一定的要求，即“所有要访问临界区的线程必须首先获取这个互斥锁，离开临界区后释放该锁”，一旦某一线程不遵循该要求，那么这个互斥锁就形同虚设了。如下面的例子：
举例：我们有两个线程，一个线程A遵循要求，每次访问临界区均先获取锁，然后将临界区的变量x按偶数值递增，另一个线程B不遵循要求直接修改x值，这样即使在线程A获取锁的情况下仍能修改临界区的变量x。

static apr_thread_mutex_t       *mutex = NULL;
static int                                x       = 0;
static apr_thread_t             *t1     = NULL;
static apr_thread_t             *t2     = NULL;

static void * APR_THREAD_FUNC thread_func1(apr_thread_t *thd, void *data)
{
apr_time_t now;
apr_time_exp_t xt;

        while (1) {
                apr_thread_mutex_lock(mutex);
                now = apr_time_now();
                apr_time_exp_lt(&xt, now);
                printf("[threadA]: own the lock, time[%02d:%02d:%02d]\n", xt.tm_hour, xt.tm_min,
                         xt.tm_sec);
                printf("[threadA]: x = %d\n", x);
                if (x % 2 || x == 0) {
                        x += 2;
                } else {
                        printf("[threadA]: Warning: x变量值被破坏，现重新修正之\n");
                        x += 1;
                }
                apr_thread_mutex_unlock(mutex);
                now = apr_time_now();
                apr_time_exp_lt(&xt, now);
                printf("[threadA]: release the lock, time[%02d:%02d:%02d]\n", xt.tm_hour, xt.tm_min,
                         xt.tm_sec);
                sleep(2);
        }

return NULL;
}

static void * APR_THREAD_FUNC thread_func2(apr_thread_t *thd, void *data)
{
        apr_time_t      now;
        apr_time_exp_t xt;
        while (1) {
                x ++;
                now = apr_time_now();
                apr_time_exp_lt(&xt, now);
                printf("[threadB]: modify the var, time[%02d:%02d:%02d]\n", xt.tm_hour, xt.tm_min, xt.tm_sec);
                sleep(2);
        }

return NULL;
}

int main(int argc, const char * const * argv, const char * const *env)
{
apr_app_initialize(&argc, &argv, &env);
apr_status_t stat;

//…

        /*
         * 创建线程
         */
        stat = apr_thread_create(&t1, NULL, thread_func1, NULL, pool);
        stat = apr_thread_create(&t2, NULL, thread_func2, NULL, pool);

//…

apr_terminate();
return 0;
}
//output
… …
[threadA]: own the lock, time[10:10:15]
[threadB]: modify the var, time[10:10:15]
[threadA]: x = 10
[threadA]: Warning: x变量值被破坏，现重新修正之
[threadA]: release the lock, time[10:10:15]
当然这个例子不一定很精确的表明threadB在threadA拥有互斥量的时候修改了x值。

二、条件变量
互斥量一般用于被设计被短时间持有的锁，一旦我们不能确定等待输入的时间时，我们可以使用条件变量来完成同步。我们曾经说过I/O复用，在我们调用poll或者select的时候实际上就是在内核与用户进程之间达成了一个协议，即当某个I/O描述符事件发生的时候内核通知用户进程并且将处于挂起状态的用户进程唤醒。而这里我们所说的条件变量让对等的线程间达成协议，即“某一线程发现某一条件满足时必须发信号给阻塞在该条件上的线程，将后者唤醒”。这样我们就有了两种角色的线程，分别为
(1) 给条件变量发送信号的线程
其流程大致为：
{
        获取条件变量关联锁；
        修改条件为真；
        调用apr_thread_cond_signal通知阻塞线程条件满足了；—— (a)
        释放变量关联锁；
}
(2) 在条件变量上等待的线程
其流程大致为：
{
        获取条件变量关联锁；
        while (条件为假) { ——————— (c)
                调用apr_thread_cond_wait阻塞在条件变量上等待；—— (b)
        }
        修改条件；
        释放变量关联锁；
}
上面两个流程中，理解三点最关键：
a) apr_thread_cond_signal中调用的pthread_cond_signal保证至少有一个阻塞在条件变量上的线程恢复；在《Unix网络编程 Vol2》中也谈过这里存在着一个race。即在发送cond信号的同时，该发送线程仍然持有条件变量关联锁，那么那个恢复线程的apr_thread_cond_wait返回时仍然拿不到这把锁就会再次挂起。这里的这个race要看各个平台实现是如何处理的了。
b) apr_thread_cond_wait中调用的pthread_cond_wait原子的将调用线程挂起，并释放其持有的条件变量关联锁；
c) 这里之所以使用while反复测试条件，是防止“伪唤醒”的存在，即条件并未满足就被唤醒。所以无论怎样，唤醒后我都需要重新测试一下条件，保证该条件的的确确满足了。

条件变量在解决“生产者-消费者”问题中有很好的应用，在我以前的一篇blog中也说过这个问题。

三、读写锁
前面说过，互斥量把想进入临界区而又试图获取互斥量的所有线程都阻塞住了。读写锁则改进了互斥量的这种霸道行为，它区分读临界区数据和修改临界区数据两种情况。这样如果有线程持有读锁的话，这时再有线程想读临界区的数据也是可以再获取读锁的。读锁和写锁的分配规则在《Unix网络编程 Vol2》中有详细说明，这里不详述。

四、小结
三种同步方式如何选择？场合不同选择也不同。互斥量在于完全同步的临界区访问；条件变量在解决“生产者-消费者”模型问题上有独到之处；读写锁则在区分对临界区读写的时候使用。