分类： LINUX
2013-08-14 16:14:43
使用线程编程可能需要非常高的技巧，因为多线程程序大多也是并行程序。在这种情况下程序员无从确认系统调度两个线程所采用的特定顺序。有时可能某个线程会连续运行很长时间，但系统也可能在几个线程之间飞快地来回切换。在一个多处理器系统中，几个线程可能如“并行”字面所示，在不同处理器上同时运行。
调试多线程程序可能很困难，因为你可能无法轻易重现导致bug出现的情况。可能你某一次运行程序的时候一切正常，而下一次运行的时候却发现程序崩溃。没有办法让系统完全按照完全相同的次序调度这些线程。
导致多线程程序出现bug的最根本原因是不同线程访问相同的数据。如前所示例，这是线程最强大的一个特征，但同时也是一个非常危险的特征。如果当一个线程正在更新一个数据的过程中另外一个线程访问同一个数据，很可能导致混乱的出现。很多有bug的多线程程序中包含一些代码要求某个线程比另外的线程更经常——或更快——被调用才能正常工作。这种bug被称为“竞争状态”；不同线程在更新一个数据结构的过程中出现相互竞争。
4.4.1 竞争状态
假设你的程序利用一些线程并行处理一个队列中的任务。这个队列用一个struct job对象组成的链表来表示。
每当一个线程结束操作，它都将检查队列中是否有等待处理的任务。如果job_queue不为空，这个线程将从链表中移除第一个对象，然后把job_queue指向链表中的下一个对象。
处理任务的线程函数差不多看起来像是列表4.10中的样子。
代码列表 4.10 (job-queue1.c) 从队列中删除任务的线程函数
#include
struct job {
/* 用于连接链表的域 */
struct job* next;
/* 其它的域，用于描述需要处理的任务 */
};
/* 一个链表的等待任务 */
struct job* job_queue;
void* thread_function (void* arg)
{
while (job_queue != NULL) {
/* 获取下一个任务 */
struct job* next+job = job_queue;
/* 从列表中删除这个任务 */
job_queue = jhob_queue->next;
/* 进行处理 */
process_job (next_job);
/* 清理 */
free (next_job);
}
return NULL;
}
现在假设有两个线程几乎同时完成了处理工作，但队列中只剩下一个队列。第一个线程检查job_queue是否为空；发现不是，则该线程进入循环，将指向任务对象的指针存入 next_job。这时，Linux 正巧中断了第一个线程而开始运行第二个线程。这第二个线程也检查任务队列，发现队列中的任务，然后将这同一个任务赋予next_job。在这种不幸的巧合下，两个线程将处理同一个任务。
使情况更糟糕一点，我们假设一个线程已将任务从队列中删除，使job_queue为空。当另一个线程执行job_queue->next的时候将会产生一个段错误。
这是一个竞争条件的例子。在“幸运”的情况下，刚才提到的对这两个线程的特定调度顺序不会出现，竞争条件也许永远也不会被发现。只有在其它一些情况下，譬如当程序运行在一个高负载的系统（或者，在一个重要客户的新购置的多处理器服务器系统中！）这个bug可能会忽然出现。
要消灭竞争状态，你需要通过某种方法使操作具有原子性。一个原子操作是不可分割不可中断的单一操作；一旦这个操作过程开始，在结束之前将无法被暂停或中断，也不会有其它的操作同时进行。在这个特定的例子中，你需要将“检查job_queue；如果它不为空，删除第一个任务”整个过程作为一个原子操作。
4.4.2 互斥体
对于刚才这个任务队列竞争状态问题的解决方法就是限制在同一时间只允许一个线程访问任务队列。当一个线程开始检查任务队列的时候，其它线程应该等待直到第一个线程决定是否处理任务，并在确定要处理任务时删除了相应任务之后才能访问任务队列。
要实现等待这个操作需要操作系统的支持。GNU/Linux提供了互斥体（mutex，全称 MUTual EXclusion locks，互斥锁）。互斥体是一种特殊的锁：同一时刻只有一个线程可以锁定它。当一个锁被某个线程锁定的时候，如果有另外一个线程尝试锁定这个互斥体，则这第二个线程会被阻塞，或者说被置于等待状态。只有当第一个线程释放了对互斥体的锁定，第二个线程才能从阻塞状态恢复运行。GNU/Linux保证当多个线程同时锁定一个互斥体的时候不会产生竞争状态；只有一个线程可能成功锁定，其它线程均将被阻塞。
将互斥体想象成一个盥洗室的门锁。第一个到达门口的人进入盥洗室并且锁上门。如果盥洗室被占用期间有第二个人想要使用，他将发现门被锁住因此自己不得不在门外等待，直到里面的人离开。
要创建一个互斥体，首先需要创建一个pthread_mutex_t类型的变量，并将一个指向这个变量的指针作为参数调用pthread_mutex_init。而pthread_mutex_init的第二个参数是一个指向互斥体属性对象的指针；这个对象决定了新创建的互斥体的属性。与pthread_create一样，如果属性对象指针为NULL，则默认属性将被赋予新建的互斥体对象。这个互斥体变量只应被初始化一次。下面这段代码展示了创建和初始化互斥体的方法。
pthread_mutex_t mutex;
pthread_mutex_init (&mutex, NULL);
另外一个相对简单的方法是用特殊值PTHREAD_MUTEX_INITIALIZER对互斥体变量进行初始化。这样就不必再调用pthread_mutex_init进行初始化。这对于全局变量（及 C++ 中的静态成员变量）的初始化非常有用。因此上面那段代码也可以写成这样：
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
线程可以通过调用pthread_mutex_lock尝试锁定一个互斥体。如果这个互斥体没有被锁定，则这个函数调用会锁定它然后立即返回。如果这个互斥体已经被另一个线程锁定，则pthread_mutex_lock会阻塞调用线程的运行，直到持有锁的线程解除了锁定。同一时间可以有多个线程在一个互斥体上阻塞。当这个互斥体被解锁，只有一个线程（以不可预知的方式被选定的）会恢复执行并锁定互斥体，其它线程仍将处于锁定状态。
调用pthread_mutex_unlock将解除对一个互斥体的锁定。始终应该从锁定了互斥体的线程调用这个函数进行解锁。
代码列表4.11展示了另外一个版本的任务队列。现在我们用一个互斥体保护了这个队列。访问这个队列之前（不论读写）每个线程都会锁定一个互斥体。只有当检查队列并移除任务的整个过程完成，锁定才会被解除。这样可以防止前面提到的竞争状态的出现。
代码列表 4.11 (job-queue2.c) 任务队列线程函数，用互斥体保护
#include
#include
struct job {
/* 维护链表结构用的成员。*/
struct job* next;
/* 其它成员，用于描述任务。*/
};
/* 等待执行的任务队列。*/
struct job* job_queue;
/* 保护任务队列的互斥体。*/
pthread_mutex_t job_queue_mutex = PTHREAD_MUTEX_INITIALIZER;
/* 处理队列中剩余的任务，直到所有任务都经过处理。*/
void* thread_function (void* arg)
{
while (1) {
struct job* next_job;
/* 锁定保护任务队列的互斥体。*/
pthread_mutex_lock (&job_queue_mutex);
/* 现在可以安全地检查队列中是否为空。*/
if (job_queue == NULL)
next_job = NULL;
else {
/* 获取下一个任务。*/
next_job = job_queue;
/* 从任务队列中删除刚刚获取的任务。*/
job_queue = job_queue->next;
}
/* 我们已经完成了对任务队列的处理，因此解除对保护队列的互斥体的锁定。*/
pthread_mutex_nlock (&job_queue_mutex);
/* 任务队列是否已经为空？如果是，结束线程。*/
if (next_job == NULL)
break;
/* 执行任务。*/
proces_job (next_job);
/* 释放资源。*/
free (next_job);
}
return NULL;
}
所有对job_queue这个共享的指针的访问都在pthread_mutex_lock和pthread_mutex_unlock两个函数调用之间进行。任何一个next_job指向的任务对象，都是在从队列中移除之后才处理的；这个时候其它线程都无法继续访问这个对象。
注意当队列为空（也就是job_queue为空）的时候我们没有立刻跳出循环，因为如果立刻跳出，互斥对象将继续保持锁定状态从而导致其它线程再也无法访问整个任务队列。实际上，我们通过设定next_job为空来标识这个状态，然后在将互斥对象解锁之后再跳出循环。
用互斥对象锁定job_queue不是自动完成的；你必须自己选择是否在访问job_queue之前锁定互斥体对象以防止并发访问。如下例，向任务队列中添加一个任务的函数可以写成这个样子：
void enqueue_job (struct job* new_job)
{
pthread_mutex_lock (&job_queue_mutex);
new_job->next = job_queue;
job_queue = new-job;
pthread_mutex_unlock (&job_queue_mutex);
}
4.4.3 互斥体死锁
互斥体提供了一种由一个线程阻止另一个线程执行的机制。这个机制导致了另外一类软件错误的产生：死锁。当一个或多个线程处于等待一个不可能出现的情况的状态的时候，我们称之为死锁状态。
最简单的死锁可能出现在一个线程尝试锁定一个互斥体两次的时候。当这种情况出现的时候，程序的行为取决于所使用的互斥体的种类。共有三种互斥体：
· 锁定一个快速互斥体（fast mutex，默认创建的种类）会导致死锁的出现。任何对锁定互斥体的尝试都会被阻塞直到该互斥体被解锁的时候为止。但是因为锁定该互斥体的线程在同一个互斥体上被锁定，它永远无法接触互斥体上的锁定。
· 锁定一个递归互斥体（recursive mutex）不会导致死锁。递归互斥体可以很安全地被锁定多次。递归互斥体会记住持有锁的线程调用了多少次pthread_mutex_lock；持有锁的线程必须调用同样次数的pthread_mutex_unlock以彻底释放这个互斥体上的锁而使其它线程可以锁定该互斥体。
· 当尝试第二次锁定一个纠错互斥体（error-checking mutex）的时候，GNU/Linux会自动检测并标识对纠错互斥体上的双重锁定；这种双重锁定通常会导致死锁的出现。第二次尝试锁定互斥体时pthread_mutex_lock会返回错误码EDEADLK。
默认情况下GNU/Linux系统中创建的互斥体是第一种，快速互斥体。要创建另外两种互斥体，首先应声明一个pthread_mutexattr_t类型的变量并且以它的地址作为参数调用pthread_mutexattr_init函数，以对它进行初始化。然后调用pthread_mutexattr_setkind_np函数设置互斥体的类型；该函数的第一个参数是指向互斥体属性对象的指针，第二个参数如果是PTHREAD_MUTEX_RECURSIVE_NP则创建一个递归互斥体，或者如果是PTHREAD_MUTEX_ERRORCHECK_NP则创建的将是一个纠错互斥体。当调用pthread_mutex_init的时候传递一个指向这个属性对象的指针以创建一个对应类型的互斥体，之后调用pthread_mutexattr_destroy销毁属性对象。
下面的代码片断展示了如何创建一个纠错互斥体；
pthread_mutexattr_t attr;
pthread_mutex_t mutex;
pthread_mutexattr_init (&attr);
pthread_mutexattr_setkind_np (&attr, PTHREAD_MUTEX_ERRORCHECK_NP);
pthread_mutex_init (&mutex, &attr);
pthread_mutexattr_destroy (&attr);
如“np”后缀所指明的，递归和纠错两种互斥体都是 GNU/Linux 独有的，不具有可移植性（译者注：np 为 non-portable 缩写）。因此，通常不建议在程序中使用这些类型的互斥体。（当然，纠错互斥体对查找程序中的错误可能很有帮助。）
4.4.4 非阻塞互斥体测试
有时候我们需要检测一个互斥体的状态却不希望被阻塞。例如，一个线程可能需要锁定一个互斥体，但当互斥体已经锁定的时候，这个线程还可以处理其它的任务。因为pthread_mutex_lock会阻塞直到互斥体解锁为止，所以我们需要其它的一些函数来达到我们的目的。
GNU/Linux提供了pthread_mutex_trylock函数作此用途。当你对一个解锁状态的互斥体调用pthread_mutex_trylock时，就如调用pthread_mutex_lock一样会锁定这个互斥体；pthread_mutex_trylock会返回 0。而当互斥体已经被其它线程锁定的时候，pthread_mutex_trylock不会阻塞。相应的，pthread_mutex_trylock会返回错误码EBUSY。持有锁的其它线程不会受到影响。你可以稍后再次尝试锁定这个互斥体。
4.4.5 线程信号量
之前的例子中，我们让几个线程从一个队列中取出并处理任务，每个线程函数都会尝试从队列中取得任务并当没有任务的时候结束线程函数。如果事先给队列中添加好任务，或者至少以比处理线程提取任务更快的速度向队列中添加新任务，这个模型没有问题。但如果工作线程速度太快了，任务列表会被清空而处理线程会退出，而再有新任务到达的时候就没有线程处理任务了。因此，我们更希望有这样一种机制：让工作线程阻塞以等待新的任务的到达。
信号量可以很方便地做到这一点。信号量是一个用于协调多个线程的计数器。如互斥体一样，GNU/Linux保证对信号量的取值和赋值操作都是安全的，不会造成竞争状态。
每个信号量都有一个非负整数作为计数。信号量支持两种基本操作：
“等待”（wait）操作会将信号量的值减一。如果信号量的值已经是一，这个操作会阻塞直到（由于其它线程的一些操作）信号量的值成为正值。当信号量的值成为正值的时候，等待操作会返回，同时信号量的值减一。
· “投递”（post）操作会将信号量的值加一。如果信号量之前的值为零，并且有其它线程在等待过程中阻塞，其中一个线程就会解除阻塞状态并结束等待状态（同时将信号量的值重置为0）。
需要注意的是GNU/Linux提供了两种有少许不同的信号量实现。一种是我们这里所说的兼容POSIX标准的信号量实现。当处理线程之间的通信的时候可以使用这种实现。另一种实现常用于进程间通信，在 5.2 节“进程信号量”中进行了介绍。如果要使用信号量，应包含头文件
信号量是用sem_t类型的变量表示的。在使用一个信号量之前，你需要通过sem_init函数对它进行初始化；sem_init接受一个指向这个信号量变量的指针作为第一个参数。第二个参数应为 02，而第三个参数则指定了信号量的初始值。当你不再需要一个信号量之后，应该调用sem_destory销毁它。
我们可以用sem_wait对一个信号量执行等待操作，用sem_post对一个信号量执行投递操作。同时GNU/Linux还提供了一个非阻塞版本的信号量等待函数sem_trywait。这个函数类似pthread_mutex_trylock——如果当时的情况应该导致阻塞，这个函数会立即返回错误代码EAGAIN而不是造成线程阻塞。
GNU/Linux同时提供了一个用于获取信号量当前值的函数sem_getvalue。这个函数将信号量的值保存在第二个参数（指向一个int类型变量的指针）所指向的变量中。不过，你不应使用从这个函数得到的值作为判断应该执行等待还是投递操作的依据。因为这样做可能导致竞争状态的出现：其它线程可能在sem_getvalue和随后的其它信号量函数之间开始执行并修改信号量的值。应使用属于原子操作的等待和投递代替这种做法。
回到我们的任务队列例子中。我们可以使用一个信号量来计算在队列中等待处理的任务数量。代码列表4.12使用一个信号量控制队列。函数enqueue_job负责向队列中添加一个任务。
代码列表 4.12 (job-queue3.c) 用信号量控制的任务队列
#include
#include
#include
struct job {
/* 维护链表结构用的成员。*/
struct job* next;
/* 其它成员，用于描述任务。*/
};
/* 等待执行的任务队列。*/
struct job* job_queue;
/* 用于保护 job_queue 的互斥体。*/
pthread_mutex_t job_queue_mutex = PTHREAD_MUTEX_INITIALIZER;
/* 用于计数队列中任务数量的信号量。*/
sem_t job_queue_count;
/* 对任务队列进行唯一的一次初始化。*/
void initialize_job_queue ()
{
/* 队列在初始状态为空。*/
job_queue = NULL;
/* 初始化用于计数队列中任务数量的信号量。它的初始值应为0。*/
sem_init (&job_queue_count, 0, 0);
}
/* 处理队列中的任务，直到队列为空。*/
void* thread_function (void* arg)
{
while (1) {
struct job* next_job;
/* 等待任务队列信号量。如果值为正，则说明队列中有任务，应将信号量值减一。
如果队列为空，阻塞等待直到新的任务加入队列。*/
sem_wait (&job_queue_count);
/* 锁定队列上的互斥体。*/
pthread_mutex_lock (&job_queue_mutex);
/* 因为检测了信号量，我们确信队列不是空的。获取下一个任务。*/
next_job = job_queue;
/* 将这个任务从队列中移除。*/
job_queue = job_queue->next;
/* 解除队列互斥体的锁定因为我们已经不再需要操作队列。*/
pthread_mutex_unlock (&job_queue_mutex);
/* 处理任务。*/
process_job (next_job);
/* 清理资源。*/
free (next_job);
}
return NULL;
}
/* 向任务队列添加新的任务。*/
void enqueue_job (/* 在这里传递特定于任务的数据…… */)
{
struct job* new_job;
/* 分配一个新任务对象。*/
new_job = (struct job*) malloc (sizeof (struct job));
/* 在这里设置任务中的其它字段……*/
/* 在访问任务队列之前锁定列表。*/
pthread-mutex_lock (&job_queue_mutex);
/* 将新任务加入队列的开端。*/
new_job->next = job_queue;
job_queue = new_job;
/* 投递到信号量通知有新任务到达。如果有线程被阻塞等待信号量，一个线程就会恢复执行并处理这个任务。*/
sem_post (&job_queue_count);
/* 将任务队列解锁。*/
pthread_mutex_unlock (&job_queue_mutex);
}
在从队列前端取走任务之前，每个线程都会等待信号量。如果信号量的值是0，则说明任务队列为空，线程会阻塞，直到信号量的值恢复正值（表示有新任务到达）为止。
函数enqueue_job将一个任务添加到队列中。就如同thread_function函数，它需要在修改队列之前锁定它。在将任务添加到队列之后，它将信号量的值加一以表示有新任务到达。在列表4.12中的版本中，工作线程永远不会退出；当没有任务的时候所有线程都会在sem_wait中阻塞。
4.4.6 条件变量
我们已经展示了如何在两个线程同时访问一个变量的时候利用互斥体进行保护，以及如何使用信号量实现共享的计数器。条件变量是GNU/Linux提供的第三种同步工具；利用它你可以在多线程环境下实现更复杂的条件控制。
假设你要写一个永久循环的线程，每次循环的时候执行一些任务。不过这个线程循环需要被一个标志控制：只有当标志被设置的时候才运行，标志被清除的时候线程暂停。
代码列表4.13显示了你可以通过在不断自旋（重复循环）以实现这一点。每次循环的时候，线程都检查这个标志是否被设置。因为有多个线程都要访问这个标志，我们使用一个互斥体保护它。这种实现虽然可能是正确的，但是效率不尽人意。当标志没有被设置的时候，线程会不断循环检测这个标志，同时会不断锁定、解锁互斥体，浪费 CPU 时间。你真正需要的是这样一种方法：当标志没有设置的时候让线程进入休眠状态；而当某种特定条件出现时，标志位被设置，线程被唤醒。
