非常精简的Linux线程池实现（二）——使用信号量和自旋锁进一步简化程序

我在博客Linux C语言多线程库Pthread中条件变量的的正确用法逐步详解中介绍了条件变量的使用，并在博客非常精简的Linux线程池实现中利用条件变量和互斥锁实现了一个简单的线程池。我在介绍条件变量所对应的同步场景时曾提到过这种同步场景也可以用信号量更加方便地实现，只是性能稍微低一些。但是对于一个线程池来说，通常并不需要高频率地加解互斥锁和发送条件信号，所以影响不大。而且如果用信号量实现线程池的条件通知，那么可以采用自旋锁来实现互斥。信号量比条件变量慢，但是自旋锁也远比互斥锁高效。

相对于博客非常精简的Linux线程池实现，这里的改变并不多，主要改变了thread_pool.c中的两个函数：worker和thread_pool_add_task。另外还有一处对线程池结构体的小修改，使得线程池创建和销毁函数在分配和释放内存时稍微简化了一些。所有这些修改都是在thread_pool.c中进行的，所有其它文件不受影响。

//thread_pool.c

#include "thread_pool.h"
#include "queue.h"
#include <stdlib.h>
#include <pthread.h>
#include <semaphore.h>

struct thread_pool {
	pthread_spinlock_t lock;
	sem_t task_ready;
	queue_t tasks;
	unsigned int thread_count;
	pthread_t threads[0];
};

struct task {
	void* (*routine)(void *arg);
	void *arg;
};

static void * worker(thread_pool_t pool) {
	struct task *t;
	while(1) {
		sem_wait(&pool->task_ready);
		pthread_spin_lock(&pool->lock);
		t=(struct task*)queue_dequeue(pool->tasks);
		pthread_spin_unlock(&pool->lock);
		t->routine(t->arg);/*todo: report returned value*/
		free(t);
	}
	return NULL;
}

thread_pool_t thread_pool_create(unsigned int thread_count) {
	unsigned int i;
	thread_pool_t pool=NULL;
	pool=(thread_pool_t)malloc(sizeof(struct thread_pool)+sizeof(pthread_t)*thread_count);
	pool->thread_count=thread_count;
	pool->tasks=queue_create();
	
	pthread_spin_init(&pool->lock, PTHREAD_PROCESS_PRIVATE);
	sem_init(&pool->task_ready, 0, 0);
	
	for(i=0; i<thread_count; i++) {
		pthread_create(pool->threads+i, NULL, (void*(*)(void*))worker, pool);
	}
	return pool;
}

void thread_pool_add_task(thread_pool_t pool, void* (*routine)(void *arg), void *arg) {
	struct task *t;
	pthread_spin_lock(&pool->lock);
	t=(struct task*)queue_enqueue(pool->tasks, sizeof(struct task));
	pthread_spin_unlock(&pool->lock);
	t->routine=routine;
	t->arg=arg;
	sem_post(&pool->task_ready);//The maximum allowable value for a semaphore could be exceeded.
}

void thread_pool_destroy(thread_pool_t pool) {
	unsigned int i;
	for(i=0; i<pool->thread_count; i++) {
		pthread_cancel(pool->threads[i]);
	}
	for(i=0; i<pool->thread_count; i++) {
		pthread_join(pool->threads[i], NULL);
	}
	pthread_spin_destroy(&pool->lock);
	sem_destroy(&pool->task_ready);
	queue_destroy(pool->tasks);
	free(pool);
}

注意上面thread_pool_add_task函数中的最后一行注释，如果快速大量添加任务导致任务累积，信号量的值会一直增大最后可能超过数值类型限制，比如int是2^31-1，long long是2^63-1，具体是哪个要看信号量的具体实现方式。总之这个风险是有的，这时候sem_post会失效，后续的任务继续添加进来都无法通知给工作线程，直到工作线程完成一个任务把信号量值恢复到正常范围，不过正常情况下任务数量不会累积到如此恐怖的地步。