线程池

一、线程池功能组件

总共包含三个组件：线程池、线程执行任务，任务详情。

线程池包含条件等待，锁，链中线程任务某一个，链中job队列中某一个。

线程执行任务包含线程池，当前线程，当前线程的的prev和next。

任务详情包含回调函数，回调函数的参数，当前job的prev和next。

二、代码编写流程

1. 初始化线程池：ThreadPool pool
2. 为线程池添加线程：参数是线程池指针，线程数量
   2.1 如果线程数量小于1，赋值1
   2.2 为线程池pool分配内存空间
   2.3 创建条件信号量，将条件信号量赋值给线程池的条件信号量
   2.4 创建锁，将锁赋值给线程池的锁
   2.5 遍历线程数量，生成执行任务的结构体，并分配内存空间，同时初始化执行任务结构体
   2.6 创建线程pthread_create(&worker->thread,NULL,ntyWorkThread,(void*)worker)，将其指针给执行任务的线程，并且执行的方法是线程执行任务的方法，worker是线程方法的参数。通过worker的线程池的等待任务队列中，获取对应的job回调函数和回调函数的参数，并执行任务的回调函数，释放worker，线程退出
3. 为线程池添加任务: 参数是线程池指针，和job结构体
   3.1 加锁
   3.2 将线程池添加job
4. 关闭线程池
   4.1 遍历线程池，将所有执行线程设置终止条件设置为1
   4.2 加锁
   4.3 将线程池的当前work设置为NULL
   4.4 线程池的等待队列job设置为NULL
   4.5 广播信号量
   4.6 解锁

main的执行顺序

1. 初始化线程池结构体
2. 初始化线程池
3. 添加job
4. 关闭线程池

三、具体代码

//
// Created by rosy on 2022/7/24.
//
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>

#define MAX_NUM_THREADS  100
#define NUM_JOBS  1000

#define LL_ADD(item,list) do{ \
    item->prev = NULL; \
    item->next = list; \
    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)


typedef struct NWORK{
    pthread_t thread;
    struct THREAD_POOL *thread_pool;
    struct NWORK *prev;
    struct NWORK *next;
    int terminate;
} nWork;

typedef struct THREAD_POOL{
    pthread_cond_t job_cond;
    pthread_mutex_t  job_mtx;
    struct NWORK *works;
    struct NJOB *jobs;

} thread_pool;

typedef struct NJOB{
    void (*job_function)(struct NJOB *job);
    void *user_data;
    struct NJOB *prev;
    struct NJOB *next;
} nJob;

static void *work_do_job(void *ptr){
    nWork  *work = (nWork*)ptr;

    while(1) {
        pthread_mutex_lock(&work->thread_pool->job_mtx);

        while(work->thread_pool->jobs == NULL){
            if(work->terminate){
                break;
            }

            pthread_cond_wait(&work->thread_pool->job_cond,&work->thread_pool->job_mtx);
        }

        if(work->terminate){
            pthread_mutex_unlock(&work->thread_pool->job_mtx);
            break;

        }

        nJob *job = work->thread_pool->jobs;
        if(job != NULL){ //移除当前job
            LL_REMOVE(job,work->thread_pool->jobs);
        }

        pthread_mutex_unlock(&work->thread_pool->job_mtx);

        if(job == NULL) continue;

        job->job_function(job);
    }

    free(work);
    pthread_exit(NULL);
}

int createThreadPool(thread_pool* pool,int num_works){

    if(num_works < 1) {
        num_works = 1;
    }
    memset(pool,0,sizeof(thread_pool));

    pthread_cond_t blank_cond = PTHREAD_COND_INITIALIZER;
    memcpy(&pool->job_cond,&blank_cond,sizeof(pool->job_cond));

    pthread_mutex_t blank_mutex = PTHREAD_MUTEX_INITIALIZER;
    memcpy(&pool->job_mtx,&blank_mutex,sizeof(pool->job_mtx));

    for(int i = 0; i < num_works;i++){
        nWork *work = (nWork*)malloc(sizeof(nWork));
        if(work == NULL){
            perror("malloc work");
            return 1;
        }
        memset(work,0,sizeof(nWork));

        work->thread_pool = pool;

        int ret = pthread_create(&work->thread,NULL,work_do_job,(void *)work);
        if(ret){
            perror("pthread_create");
            free(work);
            return 1;
        }
        LL_ADD(work,work->thread_pool->works);
    }
    return 0;
}

void add_job(thread_pool *pool,nJob* job){

    pthread_mutex_lock(&pool->job_mtx);

    LL_ADD(job,pool->jobs);

    pthread_cond_signal(&pool->job_cond);
    pthread_mutex_unlock(&pool->job_mtx);


}

void shut_down(thread_pool *pool){
    nWork *work = NULL;
    for(work = pool->works; work != NULL; work = work->next){
        work->terminate = 1;
    }

    pthread_mutex_lock(&pool->job_mtx);

    pool->works = NULL;
    pool->jobs = NULL;

    pthread_cond_broadcast(&pool->job_cond);
    pthread_mutex_unlock(&pool->job_mtx);
}

void counter(nJob *job){
    int index = *(int*)job->user_data;

    printf("index:%d,selfid:%lu\n",index,pthread_self());

    free(job->user_data);
    free(job);
}

int main(int argc,char *argv[]) {
    printf("hello yes\n");
    thread_pool pool;
    createThreadPool(&pool,MAX_NUM_THREADS);

    printf("create thread pool\n");
    for(int i = 0; i < NUM_JOBS; i++){
        nJob *job = (nJob*)malloc(sizeof(nJob));
        if(job == NULL){
            perror("malloc");
            exit(1);
        }

        job->job_function = counter;
        job->user_data = malloc(sizeof(int));
        *(int*)job->user_data = i;

        add_job(&pool,job);
    }
    printf("关闭...\n");
    shut_down(&pool);

    getchar();
    printf("finish\n");

}