C++ posix线程池子

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

typedef void*(*tasks)(void*args);

void *master_threads(void *args);

class service
{
private:
    std::vector pid;
    std::queue> taskss;
    pthread_t master;
    pthread_mutex_t lock;
    pthread_mutex_t busy_cnt;
    pthread_cond_t queue_idle; //队列空闲
    pthread_cond_t queue_full; //队列满载
    uint8_t max = 255;         // FIXME: 最少的为1 个线程 最大为 255 线程, 网管线程最大为200个线程,判断队尾是否满载
    uint8_t min = 1;
    uint8_t busy_status;
    uint8_t idle_status;
    uint8_t live_status;
    uint8_t exit_status;
    bool shutdown;
    friend void* master_threads(void* args);
    friend void* common_threads(void* args);
public:
    explicit service(){
        for(int32_t i = 0 ; i < max ; i++){
            pid.push_back(0);
        }
    };
    explicit service(uint8_t Max, uint8_t Min) : max(Max), min(Min){
        for(int32_t i = 0 ; i < max ; i++){
            pid.push_back(0);
        }
    };
    int32_t init();
    void add(std::pair task);
    bool isalive(pthread_t pid);
    ~service();
};

int32_t service::init()
{
    if (pthread_mutex_init(&(this->lock), NULL) != 0 
            || pthread_mutex_init(&(this->busy_cnt), NULL) != 0 
            || pthread_cond_init(&(this->queue_not_empty), NULL) != 0 
            || pthread_cond_init(&(this->queue_not_full), NULL) != 0)
    {
        std::cout << "service init error" << std::endl;
        return -1;
    }

    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

    int32_t ret = pthread_create(&(this->master),&attr,master_threads,(void*)this); //调用和使用本线程管理的方法

    if(ret < 0){
        std::cout << "create master threads errorn";
        return ret;
    }

    return 0;
}

void* common_threads(void*args){
    service* serv = (service*)args;
    tasks task;
    void* arg;

    while (true)
    {
        pthread_mutex_lock(&(serv->lock));
        while ((serv->taskss.size() == 0) && (!serv->shutdown))
        {                                                          // FIXME: 检测该线程是否应该退出并使用
            pthread_cond_wait(&(serv->queue_idle), &(serv->lock)); // TODO: 等待有新的任务加入
            if (serv->exit_status > 0)
            {
                serv->exit_status--;
                if (serv->live_status > serv->min)
                {
                    serv->live_status--;
                    pthread_mutex_unlock(&(serv->lock));
                    pthread_exit(nullptr);
                }
            }
        }
        if (serv->shutdown)
        {
            pthread_mutex_unlock(&(serv->lock));
            pthread_exit(nullptr)
        }
        
        task = serv->taskss.front().first;
        arg = serv->taskss.front().second;
        serv->taskss.pop(); // TODO:任务队列的出队操作
        pthread_cond_broadcast(&(serv->queue_full));
        pthread_mutex_unlock(&(serv->lock));

        pthread_mutex_lock(&(serv->busy_cnt));
        serv->busy_status++;
        pthread_mutex_unlock(&(serv->busy_cnt));

        (*(task))(arg); //TODO:  做不同的任务处理
        
        pthread_mutex_lock(&(serv->busy_cnt));
        serv->busy_status--;
        pthread_mutex_unlock(&(serv->busy_cnt));
    }
    pthread_exit(nullptr);
}

void *master_threads(void *args)
{
    service *serv = (service *)args;
    while (!serv->shutdown)
    {
        sleep(1);
        pthread_mutex_lock(&(serv->lock));
        uint32_t queue_size = serv->pid.size(); // TODO:  保证原子读取 线程中的数据
        uint8_t alive_size = serv->live_status;
        pthread_mutex_unlock(&(serv->lock));

        pthread_mutex_lock(&(serv->busy_cnt));
        uint8_t busy_size = serv->busy_status; // TODO:处于繁忙状态的线程的数量
        pthread_mutex_unlock(&(serv->busy_cnt));

        if (queue_size >= serv->min && alive_size < serv->max) // FIXME: 一次增加5个线程,一般情况下最大只有是个线程,做一个记录以后可能有问题
        {
            pthread_mutex_lock(&(serv->lock));
            uint32_t add = 0;

            for (int32_t i = 0; i < serv->max && add < 5 && serv->live_status < serv->max; i++)
            {
                if (serv->pid[i] == 0 || !serv->isalive(serv->pid[i]))
                {
                    if (pthread_create(&(serv->pid[i]), NULL, common_threads, (void *)args))
                    {
                        std::cout << "create thread error, try again !!!n"
                                  << std::endl;
                        continue;
                    }
                    add++;
                    serv->live_status++; //可以使用的线程的数量
                }
            }
            pthread_mutex_unlock(&(serv->lock));
        }

        if ((busy_size * 2) < alive_size && alive_size > serv->min)
        {
            pthread_mutex_lock(&(serv->lock));
            serv->exit_status = 5;
            pthread_mutex_unlock(&(serv->lock));

            for (int32_t i = 0; i < 5; i++)
            {
                pthread_cond_signal(&(serv->queue_idle)); //自动销毁处于空闲状态的线程
            }
        }
    }
    return nullptr;
}

bool service::isalive(pthread_t tid){
    return pthread_kill(tid, 0) ==  ESRCH ? false : true;     //发0号信号，测试线程是否存活
}

void service::add(std::pair task){
    pthread_mutex_lock(&(lock));
    while(taskss.size() == max && !shutdown){
        pthread_cond_wait(&(queue_full),&lock);
    }

    if(shutdown){
        pthread_cond_broadcast(&queue_idle);
        pthread_mutex_unlock(&lock);
        return 0;
    }

    //FIXME: 清空回调函数的参数? 是否生效
    if(taskss.back().second != nullptr){
        taskss.back().second = nullptr;
    }

    taskss.push(task);

    pthread_cond_signal(&queue_idle); //等待任务处理的任务
    pthread_mutex_unlock(&lock);
}

service::~service(){
    shutdown = true;
    for(int32_t i = 0 ; i < live_status ; i++){
        pthread_cond_broadcast(&queue_idle);
    }

    pthread_mutex_lock(&(lock));
    pthread_mutex_destroy(&(lock));
    pthread_mutex_lock(&(busy_cnt));
    pthread_mutex_destroy(&(busy_cnt));
    pthread_cond_destroy(&(queue_idle));
    pthread_cond_destroy(&(queue_full));
}

该线程池考虑ARM 板卡的性能 cortexa53 255 并发没有问题 cortexa9 64 的并发没有问题,根据设备不同需要性能调优，但是不适合在增加高并发量

当前的并发设计，小并发优选线程池子

FIXME:

1. 有一个隐患， 设备如果长时间 运行 一直在栈上分配，16个字节的空间,运行一段时间 程序必然 栈溢出, 如果用的请修改
2. 任务参数清空,是必须的,我写的这个有一个bug,正在排查

解决: 首次队列为空,跳过清空
需要对线程池中的数据初始化为0