pthread 线程池实现快速排序并计时 c++

下面是使用 pthread 线程池实现快速排序并计时的 C 代码，其中线程池的实现使用了一个队列：

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#include <sys/time.h>

#define MAX_THREADS 4

typedef struct thread_pool {
    pthread_t *threads;
    int thread_count;
    int queue_size;
    int *queue;
    int head;
    int tail;
    pthread_mutex_t lock;
    pthread_cond_t empty;
    pthread_cond_t full;
    int shutdown;
} thread_pool_t;

typedef struct quicksort_args {
    int *arr;
    int left;
    int right;
} quicksort_args_t;

thread_pool_t thread_pool_create(int thread_count, int queue_size);
void *thread_pool_work(void *thread_pool_);
void thread_pool_add_work(thread_pool_t *thread_pool_, int work);
int thread_pool_get_work(thread_pool_t *thread_pool_);
void thread_pool_destroy(thread_pool_t *thread_pool_);

void quicksort(int *arr, int left, int right);
void *quicksort_thread(void *args_);

int main() {
    int n, i;
    struct timeval start, end;
    double time_used;
    
    printf("Enter number of elements: ");
    scanf("%d", &n);
    
    int *arr = (int*)malloc(n * sizeof(int));
    
    printf("Enter %d integers:\n", n);
    for(i = 0; i < n; ++i) {
        scanf("%d", &arr[i]);
    }
    
    thread_pool_t thread_pool = thread_pool_create(MAX_THREADS, n);
    
    gettimeofday(&start, NULL);
    
    quicksort_args_t args = {arr, 0, n-1};
    thread_pool_add_work(&thread_pool, (int)&args);
    
    while(thread_pool.head != thread_pool.tail);
    
    gettimeofday(&end, NULL);
    time_used = (end.tv_sec - start.tv_sec) + (end.tv_usec - start.tv_usec) / 1000000.0;
    
    printf("Sorted array:\n");
    for(i = 0; i < n; ++i) {
        printf("%d ", arr[i]);
    }
    printf("\nTime used: %lf seconds\n", time_used);
    
    free(arr);
    thread_pool_destroy(&thread_pool);
    
    return 0;
}

thread_pool_t thread_pool_create(int thread_count, int queue_size) {
    thread_pool_t thread_pool;
    int i;
    thread_pool.thread_count = thread_count;
    thread_pool.queue_size = queue_size;
    thread_pool.queue = (int*)malloc(queue_size * sizeof(int));
    thread_pool.head = 0;
    thread_pool.tail = 0;
    thread_pool.shutdown = 0;
    pthread_mutex_init(&thread_pool.lock, NULL);
    pthread_cond_init(&thread_pool.empty, NULL);
    pthread_cond_init(&thread_pool.full, NULL);
    thread_pool.threads = (pthread_t*)malloc(thread_count * sizeof(pthread_t));
    for(i = 0; i < thread_count; ++i) {
        pthread_create(&thread_pool.threads[i], NULL, thread_pool_work, (void*)&thread_pool);
    }
    return thread_pool;
}

void thread_pool_add_work(thread_pool_t *thread_pool_, int work) {
    pthread_mutex_lock(&thread_pool_->lock);
    while(thread_pool_->tail == thread_pool_->queue_size && !thread_pool_->shutdown) {
        pthread_cond_wait(&thread_pool_->full, &thread_pool_->lock);
    }
    if(thread_pool_->shutdown) {
        pthread_mutex_unlock(&thread_pool_->lock);
        return;
    }
    thread_pool_->queue[thread_pool_->tail++] = work;
    pthread_cond_signal(&thread_pool_->empty);
    pthread_mutex_unlock(&thread_pool_->lock);
}

int thread_pool_get_work(thread_pool_t *thread_pool_) {
    int work = -1;
    pthread_mutex_lock(&thread_pool_->lock);
    while(thread_pool_->head == thread_pool_->tail && !thread_pool_->shutdown) {
        pthread_cond_wait(&thread_pool_->empty, &thread_pool_->lock);
    }
    if(thread_pool_->shutdown) {
        pthread_mutex_unlock(&thread_pool_->lock);
        return work;
    }
    work = thread_pool_->queue[thread_pool_->head++];
    pthread_cond_signal(&thread_pool_->full);
    pthread_mutex_unlock(&thread_pool_->lock);
    return work;
}

void thread_pool_destroy(thread_pool_t *thread_pool_) {
    int i;
    thread_pool_->shutdown = 1;
    pthread_cond_broadcast(&thread_pool_->empty);
    pthread_cond_broadcast(&thread_pool_->full);
    for(i = 0; i < thread_pool_->thread_count; ++i) {
        pthread_join(thread_pool_->threads[i], NULL);
    }
    free(thread_pool_->queue);
    free(thread_pool_->threads);
    pthread_mutex_destroy(&thread_pool_->lock);
    pthread_cond_destroy(&thread_pool_->empty);
    pthread_cond_destroy(&thread_pool_->full);
}

void *thread_pool_work(void *thread_pool_) {
    int work;
    while((work = thread_pool_get_work((thread_pool_t*)thread_pool_)) != -1) {
        quicksort_args_t *args = (quicksort_args_t*)work;
        quicksort(args->arr, args->left, args->right);
        free(args);
    }
    pthread_exit(NULL);
}

void quicksort(int *arr, int left, int right) {
    if(left < right) {
        int pivot = arr[left];
        int i = left, j = right;
        while(i < j) {
            while(i < j && arr[j] >= pivot) {
                --j;
            }
            if(i < j) {
                arr[i++] = arr[j];
            }
            while(i < j && arr[i] < pivot) {
                ++i;
            }
            if(i < j) {
                arr[j--] = arr[i];
            }
        }
        arr[i] = pivot;
        quicksort_args_t *args1 = (quicksort_args_t*)malloc(sizeof(quicksort_args_t));
        args1->arr = arr;
        args1->left = left;
        args1->right = i-1;
        quicksort_args_t *args2 = (quicksort_args_t*)malloc(sizeof(quicksort_args_t));
        args2->arr = arr;
        args2->left = i+1;
        args2->right = right;
        thread_pool_add_work(&thread_pool, (int)args1);
        thread_pool_add_work(&thread_pool, (int)args2);
    }
}

void *quicksort_thread(void *args_) {
    quicksort_args_t *args = (quicksort_args_t*)args_;
    quicksort(args->arr, args->left, args->right);
    free(args);
    pthread_exit(NULL);
}

其中，`thread_pool_create()` 函数用于创建线程池，`thread_pool_add_work()` 函数用于向线程池中添加任务，`thread_pool_get_work()` 函数用于从线程池中取出任务，`thread_pool_destroy()` 函数用于销毁线程池。`quicksort()` 函数用于实现快速排序，其中在递归调用时将新的任务加入到线程池中。`quicksort_thread()` 函数是 `quicksort()` 函数在新线程中的入口点函数。

在 `main()` 函数中，首先读入要排序的数字序列，然后创建线程池，将要排序的序列整体作为一个任务加入到线程池中，最后等待线程池中所有任务完成后输出排序结果和用时，并销毁线程池。

需要注意的是，这个实现中并没有对线程池中的线程进行平衡负载，即没有考虑任务的大小或者复杂度，而是简单地将任务按顺序插入队列，这可能会导致线程池中的某些线程忙碌，而其它线程处于饥饿状态的情况。如果需要更好的负载均衡，可以考虑使用更高级的线程池实现。