Subversion Repositories Code-Repo

#include "threadpool.h"
#include "list.h"
#include <stdlib.h>
#include <semaphore.h>
#include <pthread.h>

struct future {
        sem_t finished;
        struct list_elem elem;
        thread_pool_callable_func_t callable;
        void * callable_data;
        void * result;
};

struct thread_pool {
        pthread_mutex_t mutex_lock;
        pthread_cond_t condition_var;
        struct list future_list;
        bool shutting_down;
        int thread_list_length;
        pthread_t * thread_list;
};

static void * thread_func(void * arg) {
        struct thread_pool *t_pool = arg;
        pthread_mutex_lock(&t_pool->mutex_lock);
        while(1) {
                // Get lock and wait for signal while there are no jobs queued
                while (list_size(&t_pool->future_list) == 0) {
                        pthread_cond_wait(&t_pool->condition_var,&t_pool->mutex_lock);
                        if (t_pool->shutting_down) {
                                pthread_mutex_unlock(&t_pool->mutex_lock);
                                return NULL;
                        }
                }
                // Retreive and remove the first future in the list
                struct list_elem *e = list_pop_front(&t_pool->future_list);
                pthread_mutex_unlock(&t_pool->mutex_lock);
                struct future *f_entry = list_entry(e, struct future, elem);

                // Store results of the function
                f_entry->result = f_entry->callable(f_entry->callable_data);
                // Signal completion of the future
                sem_post(&f_entry->finished);

                pthread_mutex_lock(&t_pool->mutex_lock);
                if (t_pool->shutting_down)
                        return NULL;
        }
}

/* Create a new thread pool with n threads. */
struct thread_pool * thread_pool_new(int nthreads) {
        // Allocate a new thread pool structure
        struct thread_pool *t_pool = malloc(sizeof(struct thread_pool));

        // Initialize the thread pool variables
        pthread_mutex_init(&t_pool->mutex_lock,NULL);
        pthread_cond_init(&t_pool->condition_var,NULL);
        list_init(&t_pool->future_list);
        t_pool->shutting_down = false;
        t_pool->thread_list_length = nthreads;

        // Allocate and start each thread in the thread pool
        t_pool->thread_list = malloc(nthreads * sizeof(pthread_t));
        int i;
        for (i = 0; i < nthreads; i++)
                pthread_create(t_pool->thread_list + i, NULL, thread_func, t_pool);

        return t_pool;
}

/* Shutdown this thread pool.  May or may not execute already queued tasks. */
void thread_pool_shutdown(struct thread_pool * t_pool) {
        // Set the shutdown flag and notify all worker threads
        pthread_mutex_lock(&t_pool->mutex_lock);
        t_pool->shutting_down = true;
        pthread_cond_broadcast(&t_pool->condition_var);
        pthread_mutex_unlock(&t_pool->mutex_lock);

        // Wait for all worker threads to join before returning
        int i;
        for (i = 0; i < t_pool->thread_list_length; i++)
                pthread_join(t_pool->thread_list[i],NULL);

        free(t_pool->thread_list);
        free(t_pool);
}

/* Submit a callable to thread pool and return future.
 * The returned future can be used in future_get() and future_free() */
struct future * thread_pool_submit(struct thread_pool * t_pool, 
        thread_pool_callable_func_t callable, void * callable_data) {
        // Allocate and initialize the new future
        struct future *f_entry = malloc(sizeof(struct future));
        sem_init(&f_entry->finished,0,0);
        f_entry->callable = callable;
        f_entry->callable_data = callable_data;

        // Get the lock on the thread pool and enqueue the future to the end of the work queue
        pthread_mutex_lock(&t_pool->mutex_lock);
        list_push_back(&t_pool->future_list,&f_entry->elem);
        // Notify one worker thread to process the future
        pthread_cond_signal(&t_pool->condition_var);
        pthread_mutex_unlock(&t_pool->mutex_lock);
        return f_entry;
}

/* Make sure that thread pool has completed executing this callable, then return result. */
void * future_get(struct future * f_entry) {
        sem_wait(&f_entry->finished);
        return f_entry->result;
}

/* Deallocate this future.  Must be called after future_get() */
void future_free(struct future * f_entry) {
        free(f_entry);
}