后台C++开发你一定要知道的条件变量

  • 2021-01-11
  • 浏览 (71)

后台C++开发你一定要知道的条件变量

今天因为工作需要,需要帮同事用C语言(不是C++)写一个生产者消费者的任务队列工具库,考虑到不能使用任何第三库和C++的任何特性,所以我将任务队列做成一个链表,生产者在队列尾部加入任务,消费者在队列头部取出任务。很快就写好了,代码如下: /** * 线程池工具, ctrip_thread_pool.h * zhangyl 2018.03.23 */ #ifndef __CTRIP_THREAD_POOL_H__ #define __CTRIP_THREAD_POOL_H__ #include <pthread.h> #ifndef NULL #define NULL 0 #endif #define PUBLIC PUBLIC struct ctrip_task { struct ctrip_task* pNext; int value; }; struct ctrip_thread_info { //线程退出标志 int thread_running; int thread_num; int tasknum; struct ctrip_task* tasks; pthread_t* threadid; pthread_mutex_t mutex; pthread_cond_t cond; }; /* 初始化线程池线程数目 * @param thread_num 线程数目, 默认为8个 */ PUBLIC void ctrip_init_thread_pool(int thread_num); /* 销毁线程池 */ PUBLIC void ctrip_destroy_thread_pool(); /**向任务池中增加一个任务 * @param t 需要增加的任务 */ PUBLIC void ctrip_thread_pool_add_task(struct ctrip_task* t); /**从任务池中取出一个任务 * @return 返回得到的任务 */ struct ctrip_task* ctrip_thread_pool_retrieve_task(); /**执行任务池中的任务 * @param t 需要执行的任务 */ PUBLIC void ctrip_thread_pool_do_task(struct ctrip_task* t); /**线程函数 * @param thread_param 线程参数 */ void* ctrip_thread_routine(void* thread_param); #endif //!__CTRIP_THREAD_POOL_H__ /** * 线程池工具, ctrip_thread_pool.c * zhangyl 2018.03.23 */ #include "ctrip_thread_pool.h" #include <stdio.h> #include <stdlib.h> struct ctrip_thread_info g_threadinfo; int thread_running = 0; void ctrip_init_thread_pool(int thread_num) { if (thread_num <= 0) thread_num = 5; pthread_mutex_init(&g_threadinfo.mutex, NULL); pthread_cond_init(&g_threadinfo.cond, NULL); g_threadinfo.thread_num = thread_num; g_threadinfo.thread_running = 1; g_threadinfo.tasknum = 0; g_threadinfo.tasks = NULL; thread_running = 1; g_threadinfo.threadid = (pthread_t*)malloc(sizeof(pthread_t) * thread_num); int i; for (i = 0; i < thread_num; ++i) { pthread_create(&g_threadinfo.threadid[i], NULL, ctrip_thread_routine, NULL); } } void ctrip_destroy_thread_pool() { g_threadinfo.thread_running = 0; thread_running = 0; pthread_cond_broadcast(&g_threadinfo.cond); int i; for (i = 0; i < g_threadinfo.thread_num; ++i) { pthread_join(g_threadinfo.threadid[i], NULL); } free(g_threadinfo.threadid); pthread_mutex_destroy(&g_threadinfo.mutex); pthread_cond_destroy(&g_threadinfo.cond); } void ctrip_thread_pool_add_task(struct ctrip_task* t) { if (t == NULL) return; pthread_mutex_lock(&g_threadinfo.mutex); struct ctrip_task* head = g_threadinfo.tasks; if (head == NULL) g_threadinfo.tasks = t; else { while (head->pNext != NULL) { head = head->pNext; } head->pNext = t; } ++g_threadinfo.tasknum; //当有变化后,使用signal通知wait函数 pthread_cond_signal(&g_threadinfo.cond); pthread_mutex_unlock(&g_threadinfo.mutex); } struct ctrip_task* ctrip_thread_pool_retrieve_task() { struct ctrip_task* head = g_threadinfo.tasks; if (head != NULL) { g_threadinfo.tasks = head->pNext; --g_threadinfo.tasknum; printf("retrieve a task, task value is [%d]\n", head->value); return head; } printf("no task\n"); return NULL; } void* ctrip_thread_routine(void* thread_param) { printf("thread NO.%d start.\n", (int)pthread_self()); while (thread_running/*g_threadinfo.thread_running*/) { struct ctrip_task* current = NULL; pthread_mutex_lock(&g_threadinfo.mutex); while (g_threadinfo.tasknum <= 0) { //如果获得了互斥锁,但是条件不合适的话,wait会释放锁,不往下执行。 //当变化后,条件合适,将直接获得锁。 pthread_cond_wait(&g_threadinfo.cond, &g_threadinfo.mutex); if (!g_threadinfo.thread_running) break; current = ctrip_thread_pool_retrieve_task(); if (current != NULL) break; }// end inner-while-loop pthread_mutex_unlock(&g_threadinfo.mutex); ctrip_thread_pool_do_task(current); }// end outer-while-loop printf("thread NO.%d exit.\n", (int)pthread_self()); } void ctrip_thread_pool_do_task(struct ctrip_task* t) { if (t == NULL) return; //TODO: do your work here printf("task value is [%d]\n", t->value); //TODO:如果t需要释放,记得在这里释放 } 测试代码如下: // ctrip_thread_pool.cpp : Defines the entry point for the console application. // //#include "stdafx.h" #include "ctrip_thread_pool.h" #include <stdlib.h> #include <unistd.h> int main(int argc, char* argv[]) { ctrip_init_thread_pool(5); struct ctrip_task* task = NULL; int i; for (i = 0; i < 100; ++i) { task = (struct ctrip_task*)malloc(sizeof(struct ctrip_task)); task->value = i + 1; task->pNext = NULL; printf("add task, task value [%d]\n", task->value); ctrip_thread_pool_add_task(task); } sleep(10); ctrip_destroy_thread_pool(); return 0; } 代码很快就写好了,但是每次程序只能执行前几个加到任务池子里面的任务,导致池子有不少任务积累在池子里面。甚是奇怪,我也看了半天才看出结果。聪明的你,能看出上述代码为啥只能执行加到池子里面的前几个任务?先不要看答案,自己想一会儿。 linux条件变量是做后台开发必须熟练掌握的基础知识,而条件变量使用存在以下几个非常让人迷惑的地方,讲解如下 第一、必须要结合一个互斥体一起使用。使用流程如下: pthread_mutex_lock(&g_threadinfo.mutex) pthread_cond_wait(&g_threadinfo.cond, &g_threadinfo.mutex); pthread_mutex_unlock(&g_threadinfo.mutex); 上面的代码,我们分为一二三步,当条件不满足是pthread_cond_wait会挂起线程,但是不知道你有没有注意到,如果在第二步挂起线程的话,第一步的mutex已经被上锁,谁来解锁?mutex的使用原则是谁上锁谁解锁,所以不可能在其他线程来给这个mutex解锁,但是这个线程已经挂起了,这就死锁了。所以pthread_cond_wait在挂起之前,额外做的一个事情就是给绑定的mutex解锁。反过来,如果条件满足,pthread_cond_wait不挂起线程,pthread_cond_wait将什么也不做,这样就接着走pthread_mutex_unlock解锁的流程。而在这个加锁和解锁之间的代码就是我们操作受保护资源的地方。 第二,不知道你有没有注意到pthread_cond_wait是放在一个while循环里面的: pthread_mutex_lock(&g_threadinfo.mutex); while (g_threadinfo.tasknum <= 0) { //如果获得了互斥锁,但是条件不合适的话,wait会释放锁,不往下执行。 //当变化后,条件合适,将直接获得锁。 pthread_cond_wait(&g_threadinfo.cond, &g_threadinfo.mutex); if (!g_threadinfo.thread_running) break; current = ctrip_thread_pool_retrieve_task(); if (current != NULL) break; }// end inner-while-loop pthread_mutex_unlock(&g_threadinfo.mutex); 注意,我说的是内层的while循环,不是外层的。pthread_cond_wait一定要放在一个while循环里面吗?一定要的。这里有一个非常重要的关于条件变量的基础知识,叫条件变量的虚假唤醒(spurious wakeup),那啥叫条件变量的虚假唤醒呢?假设pthread_cond_wait不放在这个while循环里面,正常情况下,pthread_cond_wait因为条件不满足,挂起线程。然后,外部条件满足以后,调用pthread_cond_signal或pthread_cond_broadcast来唤醒挂起的线程。这没啥问题。但是条件变量可能在某些情况下也被唤醒,这个时候pthread_cond_wait处继续往下执行,但是这个时候,条件并不满足(比如任务队列中仍然为空)。这种唤醒我们叫“虚假唤醒”。为了避免虚假唤醒时,做无意义的动作,我们将pthread_cond_wait放到while循环条件中,这样即使被虚假唤醒了,由于while条件(比如任务队列是否为空,资源数量是否大于0)仍然为true,导致线程进行继续挂起。有人说条件变量是最不可能用错的线程之间同步技术,我却觉得这是最容易使用错误的线程之间同步技术。 上述代码存在的问题是,只考虑了任务队列开始为空,生产者后来添加了任务,条件变量被唤醒,然后消费者取任务执行的逻辑。假如一开始池中就有任务呢?这个原因导致,只有开始的几个添加到任务队列中任务被执行。因为一旦任务队列不为空。内层while循环条件将不再满足,导致消费者线程不再从任务队列中取任务消费。正确的代码如下: /** * 线程池工具, ctrip_thread_pool.c(修正后的代码) * zhangyl 2018.03.23 */ #include "ctrip_thread_pool.h" #include <stdio.h> #include <stdlib.h> struct ctrip_thread_info g_threadinfo; void ctrip_init_thread_pool(int thread_num) { if (thread_num <= 0) thread_num = 5; pthread_mutex_init(&g_threadinfo.mutex, NULL); pthread_cond_init(&g_threadinfo.cond, NULL); g_threadinfo.thread_num = thread_num; g_threadinfo.thread_running = 1; g_threadinfo.tasknum = 0; g_threadinfo.tasks = NULL; g_threadinfo.threadid = (pthread_t*)malloc(sizeof(pthread_t) * thread_num); int i; for (i = 0; i < thread_num; ++i) { pthread_create(&g_threadinfo.threadid[i], NULL, ctrip_thread_routine, NULL); } } void ctrip_destroy_thread_pool() { g_threadinfo.thread_running = 0; pthread_cond_broadcast(&g_threadinfo.cond); int i; for (i = 0; i < g_threadinfo.thread_num; ++i) { pthread_join(g_threadinfo.threadid[i], NULL); } free(g_threadinfo.threadid); pthread_mutex_destroy(&g_threadinfo.mutex); pthread_cond_destroy(&g_threadinfo.cond); } void ctrip_thread_pool_add_task(struct ctrip_task* t) { if (t == NULL) return; pthread_mutex_lock(&g_threadinfo.mutex); struct ctrip_task* head = g_threadinfo.tasks; if (head == NULL) g_threadinfo.tasks = t; else { while (head->pNext != NULL) { head = head->pNext; } head->pNext = t; } ++g_threadinfo.tasknum; //当有变化后,使用signal通知wait函数 pthread_cond_signal(&g_threadinfo.cond); pthread_mutex_unlock(&g_threadinfo.mutex); } struct ctrip_task* ctrip_thread_pool_retrieve_task() { struct ctrip_task* head = g_threadinfo.tasks; if (head != NULL) { g_threadinfo.tasks = head->pNext; --g_threadinfo.tasknum; printf("retrieve a task, task value is [%d]\n", head->value); return head; } printf("no task\n"); return NULL; } void* ctrip_thread_routine(void* thread_param) { printf("thread NO.%d start.\n", (int)pthread_self()); while (g_threadinfo.thread_running) { struct ctrip_task* current = NULL; pthread_mutex_lock(&g_threadinfo.mutex); while (g_threadinfo.tasknum <= 0) { //如果获得了互斥锁,但是条件不合适的话,wait会释放锁,不往下执行。 //当变化后,条件合适,将直接获得锁。 pthread_cond_wait(&g_threadinfo.cond, &g_threadinfo.mutex); if (!g_threadinfo.thread_running) break; }// end inner-while-loop current = ctrip_thread_pool_retrieve_task(); pthread_mutex_unlock(&g_threadinfo.mutex); ctrip_thread_pool_do_task(current); }// end outer-while-loop printf("thread NO.%d exit.\n", (int)pthread_self()); } void ctrip_thread_pool_do_task(struct ctrip_task* t) { if (t == NULL) return; //TODO: do your work here printf("task value is [%d]\n", t->value); //TODO:如果t需要释放,记得在这里释放 } ok,不知道你有没有看明白呀?