c语言线程池的简单实现

一、背景

在某种CPU密集型的应用场景中，处理计算任务耗时较多（如图像处理），考虑多核CPU的优势，若能把计算分担到多个线程中处理则能有效利用CPU；

但是，若过开启过多的线程，线程创建销毁、线程间切换所带来的开销也不容小觑；

二、相关知识

2.1 思路整理

对于这种场景，设计线程池对任务进行处理，即所有待处理的任务集中在队列里头，N个线程轮流去取队列进行计算；

2.2 队列的实现

队列使用之前的一篇文章实现的《链式队列》，队列数据为任务的回调函数、任务的工作参数；

使用的接口如下：

队列申请：queue_alloc
入列操作：queue_push
出列操作：queue_pop
队列销毁：queue_free

2.2 线程相关接口
互斥锁初始化：pthread_mutex_init(pthread_mutex_t *mutex,const pthread_mutexattr_t *attr);

上锁：pthread_mutex_lock(pthread_mutex_t *mutex);

解锁：pthread_mutex_unlock(pthread_mutex_t *mutex);

条件变量初始化：pthread_cond_init(pthread_cond_t *cond,const pthread_cond_t *attr);

线程挂起等待：pthread_cond_wait(pthread_cond_t *cond,pthread_mutex_t *mutex);

唤醒单个：pthread_cond_signal(pthread_cond_t *cond);

全部唤醒：pthread_cond_broadcast(pthread_cond_t *cond);

三、实现

结构体的定义，struct tpool 为线程池的管理结构，struct routine 为待执行的任务单元：

#define MAX_THREAD_NUM 16
#define MAX_TASKITEM 1024

typedef struct tpool
{
u8 enable;

queue_t *queue;

pthread_attr_t attr;
pthread_mutex_t mutex;
pthread_cond_t cond;
pthread_t tids[MAX_THREAD_NUM];

u16 num;
} tpool_t;

struct routine {
void *args;
void (*callback)(void *);
};

线程池的初始化，预先启动MAX_THREAD_NUM个子线程，每个子线程就绪等待：

tpool_t *tpool_alloc(u16 num)
{
int ret = FAILURE;
int ix = 0;

tpool_t *phead = NULL;

if ( num == 0 || num > MAX_THREAD_NUM ) {
goto _E1;
}

phead = calloc(1, sizeof(tpool_t));
if ( !phead ) {
goto _E1;
}

phead->enable = 1;
phead->num = num;
phead->queue = queue_alloc(MAX_TASKITEM);
if ( !phead->queue ) {
goto _E2;
}

ret  = pthread_attr_init(&phead->attr);
ret |= pthread_mutex_init(&phead->mutex, NULL);
ret |= pthread_cond_init(&phead->cond, NULL);
if ( SUCCESS != ret ) {
goto _E3;
}

ret = pthread_attr_setdetachstate(&phead->attr, PTHREAD_CREATE_DETACHED);
if ( SUCCESS != ret ) {
goto _E4;
}

for ( ix = 0; ix < num; ix++ ) {
ret = pthread_create(&phead->tids[ix], NULL, __worker, phead);
if ( SUCCESS != ret ) {
goto _E4;
}
}

ret = SUCCESS;
goto _E1;

_E4:
pthread_mutex_destroy(&phead->mutex);
pthread_cond_destroy(&phead->cond);
pthread_attr_destroy(&phead->attr);
_E3:
queue_free(&phead->queue, free);
_E2:
FREE_POINTER(phead);
_E1:
return phead;
}

子线程的实现如下，一个是在队列为空的时候挂起休眠，被唤醒后取队列中的工作单元进行调用，直到队列空再次进入休眠；

由于队列为共享资源，所以多线程操作下需要使用锁进行保护；

static int __worker_routine(tpool_t *phead)
{
struct routine *prt = NULL;

pthread_mutex_lock(&phead->mutex);
if ( queue_isempty(phead->queue) ) {
printf("Thread #%u go sleep!\n", (u32)pthread_self());
pthread_cond_wait(&phead->cond, &phead->mutex);
printf("Thread #%u wakeup!\n", (u32)pthread_self());
}

prt = (struct routine *)queue_pop(phead->queue);
pthread_mutex_unlock(&phead->mutex);

if ( prt ) {
prt->callback(prt->args);
return SUCCESS;
}
return FAILURE;
}

static void *__worker(void *args)
{
tpool_t *phead = (tpool_t *)args;

if ( !args ) {
return NULL;
}

while ( phead->enable ) {
if ( SUCCESS != __worker_routine(phead) ) {
printf("__worker_routine return, thread quit!\n");
break;
}
}

return NULL;
}


上述的工作子线程为消费者，还需要主线程去充当生产者，给工作线程投放工作任务；
int tpool_routine_add(tpool_t *phead, void (*callback)(void *), void *args)
{
struct routine *prt = NULL;

if ( !phead || !callback || !args ) {
return FAILURE;
}

prt = (struct routine *)calloc(1, sizeof(struct routine));
if ( !prt ) {
return FAILURE;
}
prt->callback = callback;
prt->args = args;

pthread_mutex_lock(&phead->mutex);
if ( SUCCESS != queue_push(phead->queue, NULL, prt) ) {
FREE_POINTER(prt);
pthread_mutex_unlock(&phead->mutex);
return FAILURE;
}
pthread_cond_signal(&phead->cond);
pthread_mutex_unlock(&phead->mutex);
return SUCCESS;
}

线程池的销毁，这个就是将线程使能位清空，然后等待所有子线程退出，然后销毁相关成员；
注意该函数有可能阻塞；

int tpool_destory(tpool_t *phead)
{
int ix = 0;

if ( !phead ) {
return FAILURE;
}

phead->enable = 0;

pthread_mutex_lock(&phead->mutex);
pthread_cond_broadcast(&phead->cond);
pthread_mutex_unlock(&phead->mutex);

for ( ix = 0; ix < phead->num; ix++ ) {
pthread_join(phead->tids[ix], NULL);
}

pthread_mutex_destroy(&phead->mutex);
pthread_cond_destroy(&phead->cond);
pthread_attr_destroy(&phead->attr);

return SUCCESS;
}


测试函数：
工作单元如下，使用休眠10秒模拟耗时操作：

struct item
{
int value;
};

void test_worker(void *args)
{
struct item *pitem = (struct item *)args;

if ( !args ) {
printf("NULL\n");
return;
}

printf("begin, %d\n", pitem->value);
sleep(10);
printf("end, %d\n", pitem->value);

free(pitem);
}

int main()
{
int ret = FAILURE;

struct item *pitem = NULL;

tpool_t *phead = NULL;

ASSERT_FAIL(NULL, phead = tpool_alloc(10));

sleep(2);
printf("1\n");
ASSERT_FAIL(NULL, pitem = (struct item *)calloc(1, sizeof(struct item)));
pitem->value = 1;
ASSERT(SUCCESS, ret = tpool_routine_add(phead, test_worker, pitem));

printf("2\n");
ASSERT_FAIL(NULL, pitem = (struct item *)calloc(1, sizeof(struct item)));
pitem->value = 2;
ASSERT(SUCCESS, ret = tpool_routine_add(phead, test_worker, pitem));

printf("3\n");
ASSERT_FAIL(NULL, pitem = (struct item *)calloc(1, sizeof(struct item)));
pitem->value = 3;
ASSERT(SUCCESS, ret = tpool_routine_add(phead, test_worker, pitem));
sleep(2);

printf("Close\n");
ASSERT(SUCCESS, ret = tpool_destory(phead));
_E1:
printf("Result: %s\n", ret ? "FAILURE" : "SUCCESS");
return ret ? EXIT_FAILURE : EXIT_SUCCESS;
}

四、总结

测试结果如下：



结果表示，当工作子线程进行任务处理时（sleep10模拟），推送新的工作任务并不会打断当前任务，而是由其他空闲线程被唤醒进行处理；

同时在发出退出信号时，由于程序中实现的为等待当前任务结束后在退出，所以也有一定的延迟性；

参考文章：

[1] C语言实现简单线程池，http://www.cnblogs.com/newth/archive/2012/05/09/2492459.html

[2] 互斥锁和条件变量，http://www.cnblogs.com/zendu/p/4981480.html