c语言 简单线程池模型及实例

#ifndef		THREAD_POLL_H__
#define		THREAD_POLL_H__

#include 

typedef	void* (*funcCallback)(void *);

struct worker_st{

	pthread_t tid;
	struct manage_st *poll;
	int state;

	struct worker_st *prev;
	struct worker_st *next;
};

struct job_st{

	funcCallback callback;
	void *arg;

	struct job_st *prev;
	struct job_st *next;
};

typedef	struct manage_st{
	struct worker_st *workers;
	struct job_st *jobs;

	pthread_cond_t poll_cond;
	pthread_mutex_t poll_mutex;
}threadPoll;

int threadPoll_create(threadPoll *poll, int size);

int threadPoll_push(threadPoll *poll, funcCallback callback, void *data);

void threadPoll_destroy(threadPoll *poll);

#endif

#include 
#include 
#include 
#include 
#include 
#include 
#include 

#include "threadPoll.h"

#define	LL_ADD(me, list) do{	
	me->prev = NULL;			
	me->next = list;			
	if(list != NULL)	list->prev = me;	
	list = me;					
} while(0)
#define LL_DELETE(me, list) do{	
	if(me->prev != NULL)	me->prev->next = me->next;	
	if(me->next != NULL)	me->next->prev = me->prev;	
	if(me == list)	list = list->next;	
	me->next = me->prev = NULL;
} while(0)

static void *FuncCallBack(void *arg){

	struct worker_st *worker = arg;

	while(1){
		pthread_mutex_lock(&worker->poll->poll_mutex);
		while(worker->poll->jobs == NULL){
			if(worker->state)
			  break;
			pthread_cond_wait(&worker->poll->poll_cond, &worker->poll->poll_mutex);
		}
		if(worker->state){
			pthread_mutex_unlock(&worker->poll->poll_mutex);
			break;
		}
		struct job_st *job = worker->poll->jobs;
		LL_DELETE(job, worker->poll->jobs);
		pthread_cond_broadcast(&worker->poll->poll_cond);
		pthread_mutex_unlock(&worker->poll->poll_mutex);

		job->callback(job->arg);
		free(job);
	}
	free(worker);

	pthread_exit(NULL);
}

int threadPoll_create(threadPoll *poll, int size){

	if(poll == NULL)
	  return -1;

	int i, ret;

	memset(poll, 0, sizeof(*poll));
	pthread_cond_t blank_cond = PTHREAD_COND_INITIALIZER;
	memcpy(&poll->poll_cond, &blank_cond, sizeof(blank_cond));
	pthread_mutex_t blank_mutex = PTHREAD_MUTEX_INITIALIZER;
	memcpy(&poll->poll_mutex, &blank_mutex, sizeof(blank_mutex));

	for(i = 0; i < size; i++){
		struct worker_st *work = malloc(sizeof(*work));
		if(work == NULL)
		  return -2;
		memset(work, 0, sizeof(*work));

		work->poll = poll;
		work->state = 0;
		ret = pthread_create(&work->tid, NULL, FuncCallBack, work);
		if(ret){
			perror("pthread_create()");
			free(work);
			return -3;
		}

		LL_ADD(work, poll->workers);
	}
	


	return 0;
}

int threadPoll_push(threadPoll *poll, funcCallback callback, void *data){

	struct job_st *job = NULL;

	job = malloc(sizeof(*job));
	if(job == NULL)
	  return -1;

	job->callback = callback;
	job->arg = data;

	pthread_mutex_lock(&poll->poll_mutex);
	LL_ADD(job, poll->jobs);
	pthread_cond_signal(&poll->poll_cond);
	pthread_mutex_unlock(&poll->poll_mutex);

	return 0;
}

void threadPoll_destroy(threadPoll *poll){

	struct worker_st *me = NULL;

	pthread_mutex_lock(&poll->poll_mutex);
	while(poll->jobs != NULL){
		pthread_cond_wait(&poll->poll_cond, &poll->poll_mutex);
	}
	pthread_mutex_unlock(&poll->poll_mutex);

	for(me = poll->workers; me != NULL; me = me->next){
		me->state = 1;
	}
	
	pthread_mutex_lock(&poll->poll_mutex);
	pthread_cond_broadcast(&poll->poll_cond);
	pthread_mutex_unlock(&poll->poll_mutex);


#if 0
	for(me = poll->workers; me != NULL; me = me->next){
		pthread_join(me->tid, NULL);
	}
#endif

	return ;
}


#if 1

void* myPrint(void *num){

	int n = *(int*)num;
	int i, flag;
	if(n < 2)
	  return NULL;

	//flag = sqrt(n);
    flag = n / 2;
	for(i = 2; i <= flag; i++){
		if(n%i == 0){
			return NULL;
		}
	}

	fprintf(stdout, "%d is a primen", n);


	return NULL;
}

int main(){
	
	threadPoll poll;
	int i, ret;
	int arr[102400];

	ret = threadPoll_create(&poll, 10);
	if(ret){
		fprintf(stderr, "threadPoll_create() false! ret = %dn", ret);
		exit(1);
	}

	for(i = 0; i < 102400; i++){
		arr[i] = i;
		threadPoll_push(&poll, myPrint, (void*)(arr+i));
		//threadPoll_push(&poll, myPrint, (void*)(&i));
	}

	threadPoll_destroy(&poll);

	exit(0);
}


#endif

        测试环境：Ubuntu 21.10  gcc version 11.2.0