-
Max Barrett authoredMax Barrett authored
threadpool.c 2.63 KiB
#include "threadpool.h"
#include "list.h"
#include <stdlib.h>
#include <stdio.h>
#include <assert.h>
#include <pthread.h>
struct thread_pool {
pthread_mutex_t lock;
pthread_cond_t cond;
struct list global_queue;
bool shutdown;
int threads;
struct worker *workers;
};
struct future {
fork_join_task_t task;
void* args;
void* result;
struct list_elem elem;
};
struct worker {
struct thread_pool* worker_threadpool;
pthread_t internal;
};
static void *start_routine(void * arg) {
struct worker *worker = (struct worker *) arg;
while (!worker->worker_threadpool->shutdown) {
pthread_mutex_lock(&worker->worker_threadpool->lock);
struct future *worker_future = list_entry(list_pop_front(&worker->worker_threadpool->global_queue), struct future, elem);
if (list_empty(&worker->worker_threadpool->global_queue)) {
pthread_cond_wait(&worker->worker_threadpool->cond, &worker->worker_threadpool->lock);
}
worker_future->result = worker_future->task(worker->worker_threadpool, worker_future->args);
}
pthread_mutex_unlock(&worker->worker_threadpool->lock);
return NULL;
}
struct thread_pool * thread_pool_new(int nthreads) {
struct thread_pool *threadpool = malloc(sizeof(struct thread_pool));
threadpool->threads = nthreads;
pthread_cond_init(&threadpool->cond, NULL);
pthread_mutex_init(&threadpool->lock, NULL);
list_init(&threadpool->global_queue);
threadpool->workers = malloc(sizeof(struct worker) * nthreads);
for (int i = 0; i < nthreads; i++) {
threadpool->workers[i].worker_threadpool = threadpool;
pthread_create(&threadpool->workers[i].internal, NULL, start_routine, &threadpool->workers[i]);
}
return threadpool;
}
void thread_pool_shutdown_and_destroy(struct thread_pool *threadpool) {
pthread_mutex_lock(&threadpool->lock);
threadpool->shutdown = true;
pthread_cond_broadcast(&threadpool->cond);
pthread_mutex_unlock(&threadpool->lock);
for (int i = 0; i < threadpool->threads; i++) {
printf("%d\n", i);
pthread_join(threadpool->workers[i].internal, NULL);
}
}
void * future_get(struct future *future) {
return future;
}
void future_free(struct future *future) {
free(future);
}
struct future * thread_pool_submit(struct thread_pool *pool, fork_join_task_t task, void * data) {
struct future *future = malloc(sizeof(struct future));
future->args = data;
future->task = task;
list_push_back(&pool->global_queue, &future->elem);
pthread_cond_signal(&pool->cond);
pthread_mutex_unlock(&pool->lock);
free(pool);
return future;
}