分类： C/C++
2011-12-06 14:13:37
我编写的并不是线程安全的队列例程，事实上我创建了一个“数据包装”或“控制”结构，它可以是任何线程支持的数据结构。看一下 control.h：
| #include typedef struct data_control { pthread_mutex_t mutex; pthread_cond_t cond; int active; } data_control; |
现在您看到了 data_control 结构定义，以下是它的视觉表示：
图像中的锁代表互斥对象，它允许对数据结构进行互斥访问。黄色的星代表条件变量，它可以睡眠，直到所讨论的数据结构改变为止。on/off 开关表示整数 "active"，它告诉线程此数据是否是活动的。在代码中，我使用整数 active 作为标志，告诉工作队列何时应该关闭。以下是 control.c：
| /* control.c ** Copyright 2000 Daniel Robbins, Gentoo Technologies, Inc. ** Author: Daniel Robbins ** Date: 16 Jun 2000 ** ** These routines provide an easy way to make any type of ** data-structure thread-aware. Simply associate a data_control ** structure with the data structure (by creating a new struct, for ** example). Then, simply lock and unlock the mutex, or ** wait/signal/broadcast on the condition variable in the data_control ** structure as needed. ** ** data_control structs contain an int called "active". This int is ** intended to be used for a specific kind of multithreaded design, ** where each thread checks the state of "active" every time it locks ** the mutex. If active is 0, the thread knows that instead of doing ** its normal routine, it should stop itself. If active is 1, it ** should continue as normal. So, by setting active to 0, a ** controlling thread can easily inform a thread work crew to shut ** down instead of processing new jobs. Use the control_activate() ** and control_deactivate() functions, which will also broadcast on ** the data_control struct's condition variable, so that all threads ** stuck in pthread_cond_wait() will wake up, have an opportunity to ** notice the change, and then terminate. */ #include "control.h" int control_init(data_control *mycontrol) { int mystatus; if (pthread_mutex_init(&(mycontrol->mutex),NULL)) return 1; if (pthread_cond_init(&(mycontrol->cond),NULL)) return 1; mycontrol->active=0; return 0; } int control_destroy(data_control *mycontrol) { int mystatus; if (pthread_cond_destroy(&(mycontrol->cond))) return 1; if (pthread_cond_destroy(&(mycontrol->cond))) return 1; mycontrol->active=0; return 0; } int control_activate(data_control *mycontrol) { int mystatus; if (pthread_mutex_lock(&(mycontrol->mutex))) return 0; mycontrol->active=1; pthread_mutex_unlock(&(mycontrol->mutex)); pthread_cond_broadcast(&(mycontrol->cond)); return 1; } int control_deactivate(data_control *mycontrol) { int mystatus; if (pthread_mutex_lock(&(mycontrol->mutex))) return 0; mycontrol->active=0; pthread_mutex_unlock(&(mycontrol->mutex)); pthread_cond_broadcast(&(mycontrol->cond)); return 1; } |
在开始调试之前，还需要一个文件。以下是 dbug.h：
| #define dabort() \ { printf("Aborting at line %d in source file %s\n",__LINE__,__FILE__); abort(); } |
此代码用于处理工作组代码中的不可纠正错误。
说到工作组代码，以下就是：
现在来快速初排代码。定义的第一个结构称作 "wq"，它包含了 data_control 和队列头。data_control 结构用于仲裁对整个队列的访问，包括队列中的节点。下一步工作是定义实际的工作节点。要使代码符合本文中的示例，此处所包含的都是作业号。
接着，创建清除队列。注释说明了它的工作方式。好，现在让我们跳过 threadfunc()、join_threads()、create_threads() 和 initialize_structs() 调用，直接跳到 main()。所做的第一件事就是初始化结构 -- 这包括初始化 data_controls 和队列，以及激活工作队列。
现在初始化线程。如果看一下 create_threads() 调用，似乎一切正常 -- 除了一件事。请注意，我们正在分配清除节点，以及初始化它的线程号和 TID 组件。我们还将清除节点作为初始自变量传递给每一个新的工作程序线程。为什么这样做？
因为当某个工作程序线程退出时，它会将其清除节点连接到清除队列，然后终止。那时，主线程会在清除队列中检测到这个节点（利用条件变量），并将这个 节点移出队列。因为 TID（线程标识）存储在清除节点中，所以主线程可以确切知道哪个线程已终止了。然后，主线程将调用 pthread_join(tid)，并联接适当的工作程序线程。如果没有做记录，那么主线程就需要按任意顺序联接工作程序线程，可能是按它们的创建顺 序。由于线程不一定按此顺序终止，那么主线程可能会在已经联接了十个线程时，等待联接另一个线程。您能理解这种设计决策是如何使关闭代码加速的吗（尤其在 使用几百个工作程序线程的情况下）？
我们已启动了工作程序线程（它们已经完成了执行 threadfunc()，稍后将讨论此函数），现在主线程开始将工作节点插入工作队列。首先，它锁定 wq 的控制互斥对象，然后分配 16000 个工作包，将它们逐个插入队列。完成之后，将调用 pthread_cond_broadcast()，于是所有正在睡眠的线程会被唤醒，并开始执行工作。此时，主线程将睡眠两秒钟，然后释放工作队列，并 通知工作程序线程终止活动。接着，主线程会调用 join_threads() 函数来清除所有工作程序线程。
现在来讨论 threadfunc()，这是所有工作程序线程都要执行的代码。当工作程序线程启动时，它会立即锁定工作队列互斥对象，获取一个工作节点（如果有的 话），然后对它进行处理。如果没有工作，则调用 pthread_cond_wait()。您会注意到这个调用在一个非常紧凑的 while() 循环中，这是非常重要的。当从 pthread_cond_wait() 调用中苏醒时，决不能认为条件肯定发生了 -- 它 可能发生了，也可能没有发生。如果发生了这种情况，即错误地唤醒了线程，而列表是空的，那么 while 循环将再次调用 pthread_cond_wait()。
如果有一个工作节点，那么我们只打印它的作业号，释放它并退出。然而，实际代码会执行一些更实质性的操作。在 while() 循环结尾，我们锁定了互斥对象，以便检查 active 变量，以及在循环顶部检查新的工作节点。如果执行完此代码，就会发现如果 wq.control.active 是 0，while 循环就会终止，并会执行 threadfunc() 结尾处的清除代码。
工作程序线程的清除代码部件非常有趣。首先，由于 pthread_cond_wait() 返回了锁定的互斥对象，它会对 work_queue 解锁。然后，它锁定清除队列，添加清除代码（包含了 TID，主线程将使用此 TID 来调用 pthread_join()），然后再对清除队列解锁。此后，它发信号给所有 cq 等待者 (pthread_cond_signal(&cq.control.cond))，于是主线程就知道有一个待处理的新节点。我们不使用 pthread_cond_broadcast()，因为没有这个必要 -- 只有一个线程（主线程）在等待清除队列中的新节点。当它调用 join_threads() 时，工作程序线程将打印关闭消息，然后终止，等待主线程发出的 pthread_join() 调用。
如果要查看关于如何使用条件变量的简单示例，请参考 join_threads() 函数。如果还有工作程序线程，join_threads() 会一直执行，等待清除队列中新的清除节点。如果有新节点，我们会将此节点移出队列、对清除队列解锁（从而使工作程序可以添加清除节点）、联接新的工作程序 线程（使用存储在清除节点中的 TID）、释放清除节点、减少“现有”线程的数量，然后继续。
现在已经到了“POSIX 线程详解”系列的尾声，希望您已经准备好开始将多线程代码添加到您自己的应用程序中。有关详细信息，请参阅 部分，这部分内容还包含了本文中使用的所有源码的 tar 文件。下一个系列中再见！
