简单实现LINUX C下的线程池.

时间 2019-12-25

标签简单实现 linux 线程栏目 Linux 繁體版

原文原文链接

本篇文章由导学宝转自：http://www.cnblogs.com/freezee/archive/2012/02/29/2373958.html
html

What I write, what I lose.socket

以前有点时间, 从新熟悉Linux的进程间通信的东西.ide

因而想起以前项目中本身写啦个很简单的线程池. 函数

此次想从新写下.测试

主要目的是用进程间或者线程间通讯的阻塞/取消阻塞方法实现对线程池线程的等待做业和开始做业.this

算是对这些代码的一种实践.spa

以上..net

===================================================================线程

我对一个简单线程池的一些理解.code

1.建立大量的线程.

2.工做线程的执行体功能为:

while(1)

{

//按照必定条件(A)阻塞.

//按照任务的参数设置开始执行任务.

}

3.控制线程的功能为.

{

//接受新任务的参数, 通常为回调函数+参数. (为保持兼容, 我设置的格式为 (void*)(*thread_task)(void*) + void* . 跟线程建立保持形式兼容.)

//按照必定规则查找空闲的线程.

//将接受的新任务参数赋给这条线程数据体.

//解除这条线程的阻塞条件.

}

===================================================================

common-thread-pool.c 线程池主要实现+一个简单的测试代码.

接口没有拿出来.

thread-control.h 提供线程池线程的等待做业和开始做业接口.

thread-control-xxxxx.c thread-control.h的接口实现. 能够使用多种方式.

common-thread-pool.c

View Code

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <pthread.h>
#include <assert.h>
#include <sys/types.h>#define DBGPRINTF_DEBUG printf#define DBGPRINTF_ERROR printf#define ASSERT  assert


#include "thread-control.h" typedef void*(*thread_task_func)(void* arg);/*线程执行任务的数据.*/ struct _thread_task_t
{
    int taskid;                     /*任务id.*/     thread_task_func task_func;     /*任务函数及参数*/     void* task_arg;
};
typedef struct _thread_task_t thread_task_t;/*线程状态.*/ typedef enum {
    ethread_status_unknown = ,
    ethread_status_idle ,
    ethread_status_running ,
    ethread_status_terminel ,
    ethread_status_cannotuse ,
}thread_status_e;/*线程数据.*/ struct _thread_data_t
{
    int thread_id;
    pthread_t pid;
    thread_status_e status;

    thread_task_t thread_task;
    THREAD_CONTROL thread_control;
};
typedef struct _thread_data_t thread_data_t;/*线程池数据.*/ struct _thread_pool_t
{
    thread_data_t* thread_data_set;
    int num_thread;
    int taskid_base;

    pthread_mutex_t thread_pool_lock;
};
typedef struct _thread_pool_t thread_pool_t;

thread_pool_t g_thread_pool;/*设置线程状态.*/ int thread_pool_setthreadstatus(thread_data_t* thread_data, thread_status_e status)
{
    thread_pool_t* thread_pool = &g_thread_pool;
    pthread_mutex_lock(&(thread_pool->thread_pool_lock));

    thread_data->status = status;

    pthread_mutex_unlock(&(thread_pool->thread_pool_lock));

    return ;
}/*线程池线程函数体.*/ void* thread_pool_func(void* arg)
{
    sleep(1);   //Wait pthread_t count.      thread_data_t* thread_data = (thread_data_t*)arg;
    DBGPRINTF_DEBUG("Thread start run. Thread_id = %d, pid = 0x%x . \n", 
            thread_data->thread_id, (unsigned int)thread_data->pid); 

    /* Continue to wait the task, then based on new task_func and task_arg to perform this task. */     while(1)
    {
        thread_control_wait(thread_data->thread_control);
        
        //Need to lock? Yes.         thread_pool_setthreadstatus(thread_data, ethread_status_running);
        DBGPRINTF_DEBUG("Task start. taskid = %d .\n", thread_data->thread_task.taskid);

        thread_data->thread_task.task_func(thread_data->thread_task.task_arg);

        DBGPRINTF_DEBUG("Task end. taskid = %d .\n", thread_data->thread_task.taskid);
        //Need to lock?Yes.         thread_pool_setthreadstatus(thread_data, ethread_status_idle);
    }

    DBGPRINTF_DEBUG("Thread end run. Thread_id = %d, pid = 0x%x . \n", 
            thread_data->thread_id, (unsigned int)thread_data->pid); 
}int thread_task_init(thread_task_t* thread_task)
{

    thread_task->taskid             = -1;
    thread_task->task_func          = NULL;
    thread_task->task_arg           = NULL;

    return ;
}int thread_data_init(thread_data_t* thread_data)
{
    thread_data->thread_id = -1;
    thread_data->pid = 0x0;
    thread_data->status = ethread_status_unknown ,

    thread_task_init(&(thread_data->thread_task));
    thread_control_init(&(thread_data->thread_control));

    return ;
}int thread_pool_create(int num_thread)
{
    ASSERT(num_thread >  && num_thread <= 10*1024);
    thread_pool_t* thread_pool = &g_thread_pool;
    
    int i = ;
    thread_pool->thread_data_set = (thread_data_t*)malloc(sizeof(thread_data_t) * num_thread);
    ASSERT(thread_pool->thread_data_set != NULL);
    thread_pool->num_thread = num_thread;
    thread_pool->taskid_base = -1;
    pthread_mutex_init(&(thread_pool->thread_pool_lock), NULL);

    for(i=; i<num_thread; i++)
    {
        thread_data_t* thread_data = thread_pool->thread_data_set+i;
        thread_data_init(thread_data);
        thread_data->thread_id = i;
        thread_data->status = ethread_status_idle;

        /* pthread_create set to detached. */         pthread_attr_t attr;
        pthread_attr_init(&attr);
        pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
        int ret = pthread_create(&(thread_data->pid), &attr, thread_pool_func, thread_data);
        if(ret != )
        {
            DBGPRINTF_DEBUG("pthread_create error[%d].\n", i);
            break;
        }
    }

    sleep(2);

    return ;
}void* test_func(void* arg)
{
    int t_sleep = (int)arg;
    DBGPRINTF_DEBUG("Test func. Sleep %d .\n", t_sleep);
    /* int a[2048*1024] = {0}; int i = 0; for(i=0; i<2028*1024; i++) { a[i] = i*i; } DBGPRINTF_DEBUG("a[0]=%d. \n", a[0]); */     sleep(t_sleep);

    DBGPRINTF_DEBUG("Test func finished. \n");

    return NULL;
}/*查询可接收任务的线程.*/ int thread_pool_queryfree(thread_data_t** thread_data_found)
{
    *thread_data_found = NULL;
    thread_pool_t* thread_pool = &g_thread_pool;
    pthread_mutex_lock(&(thread_pool->thread_pool_lock));

    int i = ;
    for(i=; i<thread_pool->num_thread; i++)
    {
        thread_data_t* thread_data = thread_pool->thread_data_set+i;
        if(thread_data->status == ethread_status_idle)
        {
            *thread_data_found = thread_data;
            break;
        }
    }

    pthread_mutex_unlock(&(thread_pool->thread_pool_lock));

    return ;
}/*分配taskid.*/ int thread_pool_gettaskid(int* taskid)
{
    thread_pool_t* thread_pool = &g_thread_pool;
    pthread_mutex_lock(&(thread_pool->thread_pool_lock));

    thread_pool->taskid_base ++;
    *taskid = thread_pool->taskid_base;

    pthread_mutex_unlock(&(thread_pool->thread_pool_lock));

    return ;
}/*向线程池增长任务.*/ int thread_pool_addtask(thread_task_func task_func, void* arg)
{
    /* Find a free thread. */     thread_data_t* thread_data_found = NULL;
    thread_pool_queryfree(&thread_data_found);

    if(thread_data_found != NULL)
    {
        DBGPRINTF_DEBUG("Thread [%d] perferm this task.\n", thread_data_found->thread_id);

        /* Set task data. */         thread_data_found->thread_task.task_func    = task_func;       
        thread_data_found->thread_task.task_arg     = arg;       
        thread_pool_gettaskid(&(thread_data_found->thread_task.taskid));

        /* Start the task. */         thread_pool_setthreadstatus(thread_data_found, ethread_status_running);
        thread_control_start(thread_data_found->thread_control);
        DBGPRINTF_DEBUG("Thread [%d] Add task[%d] finished.\n", 
                thread_data_found->thread_id, thread_data_found->thread_task.taskid);
    }
    else     {
        DBGPRINTF_ERROR("Thread pool full. Task not added.\n");
    }

    return ;
}int main()
{
    thread_pool_create(10);
    //thread_pool_create(10);     thread_pool_addtask(test_func, (void*)(1<<));
    thread_pool_addtask(test_func, (void*)(1<<1));
    thread_pool_addtask(test_func, (void*)(1<<2));
    thread_pool_addtask(test_func, (void*)(1<<3));
    thread_pool_addtask(test_func, (void*)(1<<4));
    thread_pool_addtask(test_func, (void*)(1<<5));
    thread_pool_addtask(test_func, (void*)(1<<6));
    thread_pool_addtask(test_func, (void*)(1<<7));

    sleep(6);
    thread_pool_addtask(test_func, (void*)(1<<));
    thread_pool_addtask(test_func, (void*)(1<<1));
    thread_pool_addtask(test_func, (void*)(1<<2));
    thread_pool_addtask(test_func, (void*)(1<<3));
    thread_pool_addtask(test_func, (void*)(1<<4));
    thread_pool_addtask(test_func, (void*)(1<<5));
    thread_pool_addtask(test_func, (void*)(1<<6));
    thread_pool_addtask(test_func, (void*)(1<<7));

    sleep(100000);

    return ;
}

复制代码

thread-control.h

View Code

#define THREAD_CONTROL  void*int thread_control_init(THREAD_CONTROL* thread_control);int thread_control_deinit(THREAD_CONTROL* thread_control);int thread_control_wait(THREAD_CONTROL thread_control);int thread_control_start(THREAD_CONTROL thread_control);

复制代码

thread-control.h的接口实现. 能够使用多种方式.

只要进程间通讯/线程间通讯中存在阻塞等待/解除阻塞等待的均可以拿来做实验.

好比:条件变量.
thread-control-condition.c

View Code

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <pthread.h>
#include <assert.h>
#include <sys/types.h>#define DBGPRINTF_DEBUG printf#define DBGPRINTF_ERROR printf#define ASSERT  assert


#include "thread-control.h" struct _thread_control_cond_t
{
    pthread_mutex_t lock;
    pthread_cond_t condition;
};
typedef struct _thread_control_cond_t thread_control_cond_t;int thread_control_init(THREAD_CONTROL* thread_control)
{
    *thread_control = NULL;

    thread_control_cond_t* cond = (thread_control_cond_t*)malloc(sizeof(thread_control_cond_t)); 
    assert(cond != NULL);

    pthread_mutex_init(&(cond->lock), NULL);
    pthread_cond_init(&(cond->condition), NULL);

    *thread_control = cond;

    return ;
}int thread_control_deinit(THREAD_CONTROL* thread_control)
{

    thread_control_cond_t* cond = (thread_control_cond_t*)(*thread_control); 

    pthread_mutex_destroy(&(cond->lock));
    pthread_cond_destroy(&(cond->condition));

    free(cond);
    *thread_control = NULL;
    
    return ;
}int thread_control_wait(THREAD_CONTROL thread_control)
{
    thread_control_cond_t* cond = (thread_control_cond_t*)(thread_control); 

    //Wait pthread condition.     pthread_mutex_lock(&(cond->lock));
    pthread_cond_wait(&(cond->condition), &(cond->lock));
    pthread_mutex_unlock(&(cond->lock));

    return ;
}int thread_control_start(THREAD_CONTROL thread_control)
{
    thread_control_cond_t* cond = (thread_control_cond_t*)(thread_control); 

    //start pthread condition.     pthread_mutex_lock(&(cond->lock));
    pthread_cond_signal(&(cond->condition));
    pthread_mutex_unlock(&(cond->lock));

    return ;
}

复制代码

好比:有名管道.

thread-control-fifopipe.c

View Code

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <pthread.h>
#include <assert.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/stat.h>#define DBGPRINTF_DEBUG printf#define DBGPRINTF_ERROR printf#define ASSERT  assert


#include "thread-control.h" static int path_index = ;#define LEN_CMD_PATH    10struct _fifopipe_control_t
{
    char fifopipe_cmd_path[LEN_CMD_PATH];
};
typedef struct _fifopipe_control_t fifopipe_control_t;int thread_control_init(THREAD_CONTROL* thread_control)
{
    *thread_control = NULL;

    fifopipe_control_t* fifopipe_control = (fifopipe_control_t*)malloc(sizeof(fifopipe_control_t));
    assert(fifopipe_control != NULL);

    path_index ++;
    snprintf(fifopipe_control->fifopipe_cmd_path, LEN_CMD_PATH, "./xxx%d", path_index);

    int ret = mkfifo(fifopipe_control->fifopipe_cmd_path, 0666/*(O_CREAT | O_RDWR)*/);
    assert(ret == );

    *thread_control = fifopipe_control;

    return ;
}int thread_control_deinit(THREAD_CONTROL* thread_control)
{
    fifopipe_control_t* fifopipe_control = (fifopipe_control_t*)(*thread_control);
    

    free(fifopipe_control);
    *thread_control = NULL;

    return ;
}int thread_control_wait(THREAD_CONTROL thread_control)
{
    fifopipe_control_t* fifopipe_control = (fifopipe_control_t*)(thread_control);
    
    int fd = open(fifopipe_control->fifopipe_cmd_path, O_RDONLY, );
    assert(fd>);

    char tmp = ;
    read(fd, &tmp, 1);

    return ;
}int thread_control_start(THREAD_CONTROL thread_control)
{
    fifopipe_control_t* fifopipe_control = (fifopipe_control_t*)(thread_control);
    
    int fd = open(fifopipe_control->fifopipe_cmd_path, O_WRONLY, );
    assert(fd>);

    char tmp = ;
    write(fd, &tmp, 1);

    return ;
}

复制代码

好比:管道, 消息队列, socket, while(condition?){sleep}等等.

以上代码中, 注释的比较少.

差很少.其实我都有点不知道本身在写什么.