linux线程池.doc

linux线程池(zt) /phus/archive/2005/06/09/390745.aspxthrmgr.h文件/* Copyright (C) 2004 Trog * This program is free software; you can redistribute it and/or modify* it under the terms of the GNU General Public License as published by* the Free Software Foundation; either version 2 of the License, or* (at your option) any later version.* This program is distributed in the hope that it will be useful,* but WITHOUT ANY WARRANTY; without even the implied warranty of* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the* GNU General Public License for more details.* You should have received a copy of the GNU General Public License* along with this program; if not, write to the Free Software* Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.*/#ifndef _THRMGR_H_#define _THRMGR_H_#include #include typedef struct work_item_tag struct work_item_tag *next;void *data;struct timeval time_queued; work_item_t;typedef struct work_queue_tag work_item_t *head;work_item_t *tail;int item_count; work_queue_t;typedef enum POOL_INVALID,POOL_VALID,POOL_EXIT, pool_state_t;typedef struct threadpool_tag pthread_mutex_t pool_mutex;pthread_cond_t pool_cond;pthread_attr_t pool_attr;pool_state_t state;int thr_max;int thr_alive;int thr_idle;int idle_timeout;void (*handler)(void *);work_queue_t *queue; threadpool_t;threadpool_t *thrmgr_new(int max_threads, int idle_timeout, void (*handler)(void *);void thrmgr_destroy(threadpool_t *threadpool);int thrmgr_dispatch(threadpool_t *threadpool, void *user_data);#endifthrmgr.c文件/* Copyright (C) 2004 Trog * This program is free software; you can redistribute it and/or modify* it under the terms of the GNU General Public License as published by* the Free Software Foundation; either version 2 of the License, or* (at your option) any later version.* This program is distributed in the hope that it will be useful,* but WITHOUT ANY WARRANTY; without even the implied warranty of* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the* GNU General Public License for more details.* You should have received a copy of the GNU General Public License* along with this program; if not, write to the Free Software* Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.*/#include #include #include #include thrmgr.h#include others.h#include memory.h#include output.h#define FALSE (0)#define TRUE (1)work_queue_t *work_queue_new()work_queue_t *work_q;work_q = (work_queue_t *) mmalloc(sizeof(work_queue_t);work_q-head = work_q-tail = NULL;work_q-item_count = 0;return work_q;void work_queue_add(work_queue_t *work_q, void *data)work_item_t *work_item;if (!work_q) return;work_item = (work_item_t *) mmalloc(sizeof(work_item_t);work_item-next = NULL;work_item-data = data;gettimeofday(&(work_item-time_queued), NULL);if (work_q-head = NULL) work_q-head = work_q-tail = work_item;work_q-item_count = 1; else work_q-tail-next = work_item;work_q-tail = work_item;work_q-item_count+;return;void *work_queue_pop(work_queue_t *work_q)work_item_t *work_item;void *data;if (!work_q | !work_q-head) return NULL;work_item = work_q-head;data = work_item-data;work_q-head = work_item-next;if (work_q-head = NULL) work_q-tail = NULL;free(work_item);return data;void thrmgr_destroy(threadpool_t *threadpool)if (!threadpool | (threadpool-state != POOL_VALID) return;if (pthread_mutex_lock(&threadpool-pool_mutex) != 0) logg(!Mutex lock failedn);exit(-1);threadpool-state = POOL_EXIT;/* wait for threads to exit */if (threadpool-thr_alive 0) /*通知兄弟们收工*/if (pthread_cond_broadcast(&(threadpool-pool_cond) != 0) pthread_mutex_unlock(&threadpool-pool_mutex);return;while (threadpool-thr_alive 0) /*原来是这位老兄负责等最后一名兄弟的信号啊*/if (pthread_cond_wait (&threadpool-pool_cond, &threadpool-pool_mutex) != 0) pthread_mutex_unlock(&threadpool-pool_mutex);return;if (pthread_mutex_unlock(&threadpool-pool_mutex) != 0) logg(!Mutex unlock failedn);exit(-1);pthread_mutex_destroy(&(threadpool-pool_mutex);pthread_cond_destroy(&(threadpool-pool_cond);pthread_attr_destroy(&(threadpool-pool_attr);free(threadpool);return;threadpool_t *thrmgr_new(int max_threads, int idle_timeout, void (*handler)(void *)threadpool_t *threadpool;if (max_threads queue = work_queue_new();if (!threadpool-queue) free(threadpool);return NULL; threadpool-thr_max = max_threads;threadpool-thr_alive = 0;threadpool-thr_idle = 0;threadpool-idle_timeout = idle_timeout;threadpool-handler = handler;pthread_mutex_init(&(threadpool-pool_mutex), NULL);if (pthread_cond_init(&(threadpool-pool_cond), NULL) != 0) free(threadpool);return NULL;if (pthread_attr_init(&(threadpool-pool_attr) != 0) free(threadpool);return NULL;if (pthread_attr_setdetachstate(&(threadpool-pool_attr), PTHREAD_CREATE_DETACHED) != 0) free(threadpool);return NULL;threadpool-state = POOL_VALID;return threadpool;/*工作线程.该工作线程遍历工作链表,如果有活干就干,没活干就等活干,难怪叫民工*/void *thrmgr_worker(void *arg)threadpool_t *threadpool = (threadpool_t *) arg;void *job_data;int retval, must_exit = FALSE;struct timespec timeout;/* loop looking for work */for (;) if (pthread_mutex_lock(&(threadpool-pool_mutex) != 0) /* Fatal error */logg(!Fatal: mutex lock failedn);exit(-2);timeout.tv_sec = time(NULL) + threadpool-idle_timeout;timeout.tv_nsec = 0;threadpool-thr_idle+;while (job_data=work_queue_pop(threadpool-queue) = NULL)& (threadpool-state != POOL_EXIT) /* Sleep, awaiting wakeup ,注意,民工等一段时间,如果没有活干就结束该线程*/retval = pthread_cond_timedwait(&(threadpool-pool_cond),&(threadpool-pool_mutex), &timeout);if (retval = ETIMEDOUT) must_exit = TRUE;break;threadpool-thr_idle-;/要干活了,闲着的民工少了一位if (threadpool-state = POOL_EXIT) must_exit = TRUE;if (pthread_mutex_unlock(&(threadpool-pool_mutex) != 0) /* Fatal error */logg(!Fatal: mutex unlock failedn);exit(-2);if (job_data) threadpool-handler(job_data); else if (must_exit) /*如果没有等到活或者要结束整个线程池时,该线程收工*/break;if (pthread_mutex_lock(&(threadpool-pool_mutex) != 0) /* Fatal error */logg(!Fatal: mutex lock failedn);exit(-2);threadpool-thr_alive-;/活干完了,该走人了(人又少了一个)if (threadpool-thr_alive = 0) /* signal that all threads are finished */pthread_cond_broadcast(&threadpool-pool_cond);/人都跑光了,谁还听得到这个信号?多次一举吗?if (pthread_mutex_unlock(&(threadpool-pool_mutex) != 0) /* Fatal error */logg(!Fatal: mutex unlock failedn);exit(-2);return NULL;/*创建一个工作线程,如果目前有等待条件信号的工作线程,则唤醒该工作线程处理数据*/int thrmgr_dispatch(threadpool_t *threadpool, void *user_data)pthread_t thr_id;if (!threadpool) return FALSE;/* Lock the threadpool */if (pthread_mutex_lock(&(threadpool-pool_mutex) != 0) logg(!Mutex lock failedn);return FALSE;if (threadpool-state != POOL_VALID)