thread_pool.cc

// usage : g++ thread_pool.cc -o tp  -pthread --std=c++11
// remove chrono header and timer, you may modify this file to c
// usage : gcc xxx.c -o tp_in_c -pthread --std=c99
#include <stdio.h>
#include <stdlib.h>
#include <math.h>
#include <pthread.h>
#include <unistd.h>

#include <chrono> // for timer

typedef std::chrono::high_resolution_clock hrclock;
typedef std::chrono::duration<double, std::milli> mil;
typedef std::chrono::duration<int64_t, std::ratio<1,1000000000>> nal;
typedef std::chrono::time_point<hrclock, nal> time_point;

typedef struct {
    double d;
} thread_data;

typedef struct {
    int jlo;
    int jhi;
    int status; // 0: ready 
                // 1: adding data 
                // 2: data handling
                // 3: done
                // -1: finish
    int next_free;
    int _id;
} thread_node;

int current_free_node = 0 ;
int node_in_use = 0;
pthread_mutex_t mutex;
thread_node * nodes;

void back(int idx) {
    pthread_mutex_lock(&mutex);
    nodes[idx].next_free = current_free_node;
    current_free_node = idx;
    nodes[idx].status = 0;
    node_in_use -- ;
    pthread_mutex_unlock(&mutex);
}

int alloc() {
    pthread_mutex_lock(&mutex);
    int rt = current_free_node;
    if(current_free_node != -1) {
        node_in_use ++ ;
        current_free_node = nodes[current_free_node].next_free;
        nodes[rt].status = 1;
    }
    pthread_mutex_unlock(&mutex);
    return rt;
}

thread_data data[10000][10000];

void data_init() {
    for(int i = 0 ; i < 10000 ; i++ ) {
        for(int j = 0; j < 10000; j++ ) {
            data[i][j].d = 1;
        }
    }
}

void data_dump() {
    for(int i = 9990 ; i < 10000 ; i++ ) {
        for(int j = 9990 ; j < 10000 ; j ++ ) {
            printf("%10.3f ",data[i][j].d);
        }
        printf("\n");
    }
}

void * thread_handler(void * _data) {
    thread_node * _v = (thread_node *) _data;
    for(;;) {
        while(_v -> status != 2)  {
            if(_v-> status == -1 ) {
                return NULL;
            }
            usleep(10);
        }
        for(int i = 1 ; i < 10000 ; i++ )
            for(int j = _v->jlo; j < _v->jhi ; j++ )
                data[i][j].d = sqrt(i+j) + data[i-1][j].d + data[i][j].d;
        back(_v->_id);
    }
}

int main(void) {
    int thread_size = 2;
    int j_range = 300;

    time_point t0, t1;
    data_init();
    t0 = hrclock::now();
    // normal type
    for(int i = 1 ; i < 10000 ; i ++ ) {
        for(int j = 0 ; j < 10000 ; j ++ ) {
            data[i][j].d = sqrt(i+j) + data[i-1][j].d + data[i][j].d;
        }
    }
    t1 = hrclock::now();
    data_dump();
    printf("time cost 1 : %f ms\n",mil(t1 - t0 ).count());
    data_init();
    t0 = hrclock::now();
    // with thread 
    pthread_mutex_init(&mutex,NULL);
    nodes = (thread_node *)malloc(thread_size * sizeof(thread_node));
    for(int i = 0 ; i < thread_size - 1; i ++ ) {
        nodes[i].status = 0;
        nodes[i].next_free = i + 1;
        nodes[i]._id = i;
    }
    nodes[thread_size-1].status = 0;
    nodes[thread_size-1].next_free = -1;
    nodes[thread_size-1]._id = thread_size-1;
    pthread_t pt[thread_size];
    for(int i = 0 ; i < thread_size; i ++ ) {
        int ret = pthread_create(&(pt[i]),NULL,thread_handler,&(nodes[i]));
        if(ret != 0 ) {
            fprintf(stderr,"crate thread  %d failed .. \n", i);
            return -1;
        }
    }
    for(int j = 0 ; j < 10000 ; j += j_range ) {
        int off = 0;
        while((off = alloc()) == -1 ) usleep(10);
        nodes[off].jlo = j;
        int hi = j + j_range;
        if(hi > 10000 )
            nodes[off].jhi = 10000;
        else
            nodes[off].jhi = hi;
        //printf("%d => %d\n",nodes[off].jlo,nodes[off].jhi);
        nodes[off].status = 2;
        //printf("dd: %d\n",j);
    }
    while(node_in_use > 0) {
        //printf("--%d\n",node_in_use);
        usleep(10);
    }
    for(int i = 0 ; i < thread_size; i ++ ) {
        nodes[i].status = -1;
    }
    for(int i = 0 ; i < thread_size; i ++ ) {
        pthread_join(pt[i],NULL);
    }
    pthread_mutex_destroy(&mutex);
    t1 = hrclock::now();
    data_dump();
    printf("time cost 2 : %f ms\n",mil(t1 - t0 ).count());
    
    return 0;
}