Design and implementation of thread pool

Principle and implementation of thread pool

Explain the content

In these two days, I focused on the basic component of thread pool , And I tried to write a copy of the code of the thread pool , Here, I would like to share with you my learning achievements .

1. Role of thread pool

1.1 Advantages of thread pool

Thread pool is a tool to help us manage threads , It maintains multiple threads , It can reduce the consumption of resources , Improve system performance . And by using thread pools , We developers can better focus on the task code , Regardless of how the thread is executed , Decouple task submission and execution .

1.2 Where thread pools are used

Suppose that the time it takes a server to complete a task is ：T1 Create thread time ,T2 The time at which the task is executed in the thread ,T3 Destroy thread time . If ：T1 + T3 Far greater than T2, You can use a thread pool , To improve server performance .

2. How thread pools work

Thread pool is mainly composed of three parts ： In scheduler , Task queue , Work queue . A thread pool starts by creating several threads and putting them in a work queue in the form of a queue , And monitor the task queue with condition variables , If the task queue is empty, it will block , In this way, after all threads are created, they will be blocked in the place waiting for the task to arrive . Then the programmer writes the task and adds it to the task queue , Can wake up blocked threads , To complete the task .

3. Implementation of thread pool

#include<queue>
#include<pthread.h>
#include<stdlib.h>
using namespace std;
class njobs;
class threadpool;
typedef void NJOBCALLBACK(njobs *job); // Callback function 

class njobs
{
    
public:
    njobs();
    NJOBCALLBACK *m_job_function;
    void *m_user_data; // Task data 
};

class nworkers
{
    
public:
    nworkers();
    ~nworkers();
    pthread_t m_thread; 
    int m_terminate; // Whether to terminate 
};

class threadpool
{
    
public:
    threadpool();
    int ThreadPoolCreate(int max_thread); // Create thread 
    int ThreadPoolQueue(njobs *job);// Task in the team 
    static void *WorkerThread(void *ptr);

private:
    int m_max_thread;
    queue<njobs> m_waiting_jobs;// Task waiting queue 
    queue<nworkers> m_workers;// The initial thread included 
    pthread_mutex_t m_jobs_mtx;
    pthread_cond_t m_jobs_cond;
};


//pthread_create Incoming parameter 
struct workaddjob
{
    
    nworkers *worker;
    threadpool *pool;
};


```cpp
#include "threadpool.h"
#include<stdio.h>
njobs::njobs()
{
    
    m_job_function = NULL;
    m_user_data = NULL;
}

nworkers::nworkers()
{
    
    m_terminate=0;
}

nworkers::~nworkers()
{
    
    pthread_exit(NULL);
}



threadpool::threadpool()
{
    
    m_max_thread = 0;
    pthread_mutex_init(&m_jobs_mtx, NULL);
    pthread_cond_init(&m_jobs_cond, NULL);
}

void* threadpool::WorkerThread(void *ptr)
{
    
    workaddjob *wb = (workaddjob *)ptr;
    nworkers *curworker = wb->worker;
    threadpool *curpool = wb->pool;
    while(1)
    {
    
    //pthread_cond_wait It must be locked before 
        pthread_mutex_lock(&curpool->m_jobs_mtx);
        while(curpool->m_waiting_jobs.empty())
        {
    
            if(curworker->m_terminate) break;
            // The function will eventually block here ,pthread_cond_wait Will unlock first , Then lock it when you wake up .
            pthread_cond_wait(&curpool->m_jobs_cond,&curpool->m_jobs_mtx);
        }
        if(curworker->m_terminate)
        {
    
            pthread_mutex_unlock(&curpool->m_jobs_mtx);
            break;
        }
        njobs job = curpool->m_waiting_jobs.front();
        curpool->m_waiting_jobs.pop();
        pthread_mutex_unlock(&curpool->m_jobs_mtx);
        job.m_job_function(&job);
    }
}

int threadpool::ThreadPoolCreate(int max_thread)
{
    
    if(max_thread < 1) max_thread = 1;
    else m_max_thread=max_thread;
    // Loop creation thread 
    for(int i = 0;i < m_max_thread;i++)
    {
    
        nworkers *worker = new nworkers;
        if (worker == NULL)
        {
    
            perror("malloc");
            return 1;
        }
        workaddjob *wb=new workaddjob;
        wb->pool=this;
        wb->worker =worker;
        int ret = pthread_create(&worker->m_thread, NULL, this->WorkerThread, (void *)wb);
        if (ret)
        {
    
            perror("pthread_create");
            delete worker;
            return 1;
        }
        m_workers.push(*worker);
    }
    return 0;

}
// Task in the team 
int threadpool::ThreadPoolQueue(njobs *job)
{
    
    if(job == NULL)
    {
    
        perror("job add");
        return 1;
    }
    pthread_mutex_lock(&m_jobs_mtx);
    m_waiting_jobs.push(*job);
    pthread_cond_signal(&m_jobs_cond);
    pthread_mutex_unlock(&m_jobs_mtx);
    return  0;
}

// Here are the test cases 
```cpp
#include <iostream>
#include<threadpool.h>
using namespace std;

#define MAX_THREAD 80
#define COUNTER_SIZE 1000

void counter(njobs *job) {
    

    int *index = reinterpret_cast<int *>(job->m_user_data);

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


}


int main(int argc, char *argv[]) {
    

    threadpool pool;

    pool.ThreadPoolCreate(MAX_THREAD);

    int i = 0;
    for (i = 0;i < COUNTER_SIZE;i ++) {
    
        njobs *job = (njobs*)malloc(sizeof(njobs));
        if (job == NULL)
        {
    
            perror("malloc");
            exit(1);
        }
        job->m_job_function = counter;
        job->m_user_data =reinterpret_cast<void *>(i);
        pool.ThreadPoolQueue(job);

    }
    printf("finish\n");
    getchar();
    return 0;
}

The content is to keep creating new threads and passing numbers in. Let's see the results

It can be seen that the thread pool runs in no order .