乐趣区

关于redis:redis后台线程条件变量源码解析

jiezi

定义

条件变量容许一个线程就某个共享变量（或其余共享资源）的状态变动告诉其余线程，并让其余线程期待（阻塞于）这一告诉。

标记变量

有时候线程之间不必条件变量和锁，用一个标记变量会看起来很简略，很吸
引人。例如，期待代码：

while (ready == 0)
    ; // spin

相干的发信号代码看起来像这样：

ready = 1;

千万不要这么做。因为：

  • 首先，少数状况下性能差（长时间的自旋节约 CPU）。
  • 其次，容易出错。这些不正规的同步办法半数以上都是有问题的。

条件变量 API

采纳了条件变量（condition variable），这一问题就迎刃而解：容许一个线程休眠（等
待）直至接获另一线程的告诉（收到信号）去执行某些操作（例如，呈现一些“状况”后，
期待者必须立刻做出响应）。线程阻塞时不占用 CPU。

#include <pthread.h>

int pthread_cond_signal(pthread_cond_t *cond);
int pthread_cond_broadcast(pthread_cond_t *cond);
int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);
pthread_cond_signal/pthread_cond_broadcast

函数 pthread_cond_signal()和 pthread_cond_broadcast()之间的差异在于，二者对阻塞于 pthread_cond_wait()的多个线程解决形式不同。pthread_cond_signal()函数只保障唤醒至多一条受到阻塞的线程，而 pthread_cond_broadcast()则会唤醒所有遭阻塞的线程。

只有当仅需唤醒一条（且无论是其中哪条）期待线程来解决共享变量的状态变动时，才应应用 pthread_cond_signal()。利用这种形式的典型状况是，所有期待线程都在执行完全相同的工作 。同理，函数 pthread_cond_broadcast() 所解决的状况是：处于期待状态的所有线程执行
的工作不同（即各线程关联于条件变量的断定条件不同）。

pthread_cond_wait

条件变量总是要与一个互斥量相干。将这些对象通过函数参数传递给 pthread_cond_wait()，后者执行如下操作步骤。

  • 解锁互斥量 mutex。
  • 梗塞调用线程，直至另一线程就条件变量 cond 发出信号。
  • 从新锁定 mutex。

通常状况下代码会以如下形式访问共享变量:

pthread_mutex_lock(&mtx);
while(/* Check that shared variable is not in state we want */)
    pthread_cond_wait(&cond, &mtx);
pthread_mutex_unlock(&mtx);

代码执行流程如下：

  1. 线程在筹备查看共享变量状态时锁定互斥量。
  2. 查看共享变量的状态。
  3. 如果共享变量未处于预期状态，线程应在期待条件变量并进入休眠前解锁互斥量（以便其余线程能拜访该共享变量）。
  4. 当线程因为条件变量的告诉而被再度唤醒时，必须对互斥量再次加锁，因为在典型状况下，线程会立刻访问共享变量。

函数 pthread_cond_wait()会 主动执行 最初两步中对互斥量的解锁和加锁动作。第 3 步中互斥量的开释与陷入对条件变量的期待同属于一个 原子操作 。换句话说，在函数 pthread_cond_wait() 的调用线程陷入对条件变量的期待之前，其余线程不可能获取到该互斥量，也不可能就该条件变量发出信号。

测试条件变量的判断条件（predicate）

必须由一个 while 循环，而不是 if 语句 ，来管制对 pthread_cond_wait() 的调用。这是因为，当代码从 pthread_cond_wait()返回时，并不能确定判断条件的状态，所以应该立刻从新查看判断条件，在条件不满足的状况下持续休眠期待。

从 pthread_cond_wait()返回时，之所以不能对判断条件的状态做任何假如，其理由如下。

  • 其余线程可能会率先醒来。兴许有多个线程在期待获取与条件变量相干的互斥量。即便就互斥量发出通知的线程将判断条件置为预期状态，其余线程仍然有可能率先获取互斥量并扭转相干共享变量的状态，进而扭转判断条件的状态。
  • 设计时设置“宽松的”判断条件或者更为简略。有时，用条件变量来表征可能性而非确定性，在设计应用程序时会更为简略。换言之，就条件变量发送信号意味着“可能有些事件”须要接管信号的线程去响应，而不是“肯定有一些事件”要做。应用这种办法，能够基于判断条件的近似状况来发送条件变量告诉，接管信号的线程能够通过再次查看判断条件来确定是否真的须要做些什么。
  • 可能会产生虚伪唤醒的状况。在一些实现中，即便没有任何其余线程真地就条件变量发出信号，期待此条件变量的线程仍有可能醒来。在一些多处理器零碎上，为确保高效实现而采纳的技术会导致此类（不常见的）虚伪唤醒。

redis5.0.8 源码

void *bioProcessBackgroundJobs(void *arg) {
    struct bio_job *job;
    unsigned long type = (unsigned long) arg;
    pthread_mutex_lock(&bio_mutex[type]);

    while(1) {
        /* The loop always starts with the lock hold. */
        if (listLength(bio_jobs[type]) == 0) {pthread_cond_wait(&bio_newjob_cond[type],&bio_mutex[type]);
            continue;
        }
        /* Pop the job from the queue. */
        ln = listFirst(bio_jobs[type]);
        job = ln->value;
        /* It is now possible to unlock the background system as we know have
         * a stand alone job structure to process.*/
        pthread_mutex_unlock(&bio_mutex[type]);

        /* Process the job accordingly to its type. */
        ......

        /* Lock again before reiterating the loop, if there are no longer
         * jobs to process we'll block again in pthread_cond_wait(). */
        pthread_mutex_lock(&bio_mutex[type]);
        listDelNode(bio_jobs[type],ln);
        bio_pending[type]--;

        /* Unblock threads blocked on bioWaitStepOfType() if any. */
        pthread_cond_broadcast(&bio_step_cond[type]);
    }
}
代码解释

  1. continue 跳转到外层 while，相当于将 if 转换为 while 条件。

    if (listLength(bio_jobs[type]) == 0) {pthread_cond_wait(&bio_newjob_cond[type],&bio_mutex[type]);
 continue;
}

  2. 弹出 bio_jobs 队首工作后，即可解锁。因为bio_jobs 对列是共享变量。

    /* Pop the job from the queue. */
ln = listFirst(bio_jobs[type]);
job = ln->value;
/* It is now possible to unlock the background system as we know have
 * a stand alone job structure to process.*/
pthread_mutex_unlock(&bio_mutex[type]);

小结

线程提供的弱小共享是有代价的。多线程应用程序必须应用互斥量和条件变量等 同步原语 来协调对共享变量的拜访。

  • 互斥量提供了对共享变量的独占式拜访。
  • 条件变量容许一个或多个线程等待告诉：其余线程扭转了共享变量的状态。

参考资料

  1. Linux/Unix 零碎编程手册
  2. redis5.0.8 源码
退出移动版