#include

#include

#include

#include

#include

#include

//定义链表操作add delete 宏函数

//定义添加宏函数

#define LL_ADD(item, list) do {\

item->prev = NULL;\

item->next = list;\

if (list != NULL) list->prev = item;\

list = item;\

} while (0)

//定义删除宏函数

#define LL_REMOVE(item, list) do {\

if (item->prev != NULL) item->prev->next = item->next;\

if (item->next != NULL) item->next->prev = item->prev;\

if (list == item) list = item->next;\

item->prev = item->next = NULL;\

} while (0)

//定义Worker链表，双向链表

typedef struct NWORKER {

pthread_t threadid; //线程id

int terminate; //退出机制，标志位 释放线程，将terminate置1

struct NMANAGER *pool; //定义一个双向链表

struct NWORKER *next;

struct NWORKER *prev;

} nWorker;

typedef struct NJOB {

void (*func)(void *arg); //回调函数，分析这个的功能？ void (*func)(void ):是个指向函数地址的指针，此函数地址是线程开始执行的起始函数地址。起始函数的地址由func确定。

void *user_data; //

struct NJOB *next;

struct NJOB *prev;

} nJob;

typedef struct NMANAGER {

nWorker *workers;

nJob *jobs;

int sum_thread; // 总数

int free_thread; //free_thread = sum_thread;

pthread_mutex_t jobs_mtx; //引入互斥量

pthread_cond_t jobs_cond; //引入条件变量，线程挂起就是因为条件变量

} nManager;

typedef nManager nThreadPool; //取别名

void *nWorkerCallback(void *arg) {

nWorker *worker = (nWorker*)arg;

while (1) {

// jobs != null

pthread_mutex_lock(&worker->pool->jobs_mtx);

while (worker->pool->jobs == NULL) { //任务队列为空

if (worker->terminate == 1) break; //terminate用法

pthread_cond_wait(&worker->pool->jobs_cond, &worker->pool->jobs_mtx); //条件等待

}

if (worker->terminate == 1) {

pthread_mutex_unlock(&worker->pool->jobs_mtx);

break;

}

// ll_remove(item, jobs)

nJob *job = worker->pool->jobs;

LL_REMOVE(job, worker->pool->jobs);

pthread_mutex_unlock(&worker->pool->jobs_mtx);

// enter

worker->pool->free_thread --;

job->func(job->user_data); //不断执行任务

// end

worker->pool->free_thread ++;

free(job);

// jobs->func(jobs);

}

free(worker);

}

int nThreadPoolCreate(nThreadPool *pool, int numWorkers) {

if (pool == NULL) return -1;

if (numWorkers < 1) numWorkers = 1;

memset(pool, 0, sizeof(nThreadPool)); //

pthread_mutex_t blank_mutex = PTHREAD_MUTEX_INITIALIZER;

memcpy(&pool->jobs_mtx, &blank_mutex, sizeof(pthread_mutex_t));

pthread_cond_t blank_cond = PTHREAD_COND_INITIALIZER;

memcpy(&pool->jobs_cond, &blank_cond, sizeof(pthread_cond_t));

int i = 0;

for (i = 0;i < numWorkers;i ++) {

nWorker *worker = (nWorker*)malloc(sizeof(nWorker));

if (worker == NULL) {

perror("malloc");

return 1;

}

memset(worker, 0, sizeof(nWorker));

int ret = pthread_create(&worker->threadid, NULL, nWorkerCallback, worker);

if (ret) {

perror("pthread_create");

//free(worker);

nWorker *w = pool->workers;

for (w = pool->workers; w != NULL; w = w->next) {

w->terminate = 1;

}

return 1;

}

LL_ADD(worker, pool->workers);

}

return 0;

}

int nThreadPoolDestory(nThreadPool *pool) {

nWorker *w = pool->workers;

for (w = pool->workers; w != NULL; w = w->next) {

w->terminate = 1;

}

pthread_mutex_lock(&pool->jobs_mtx);

pthread_cond_broadcast(&pool->jobs_cond);

pthread_mutex_unlock(&pool->jobs_mtx);

}

int nThreadPoolPushJob(nThreadPool *pool, nJob *job) {

pthread_mutex_lock(&pool->jobs_mtx);

LL_ADD(job, pool->jobs);

pthread_cond_signal(&pool->jobs_cond);

pthread_mutex_unlock(&pool->jobs_mtx);

}

//增加线程API

int nThreadpoolADDitem( ){

}

// API

// int nThreadPoolCreate(nThreadPool *pool, int numWorkers)

// int nThreadPoolPushJob(nThreadPool *pool, nJob *job)

// int nThreadPoolDestory(nThreadPool *pool)