src/server/thrpool.c

/*
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
 *
 * Copyright 2011 Olaf Wintermann. All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 *   1. Redistributions of source code must retain the above copyright
 *      notice, this list of conditions and the following disclaimer.
 *
 *   2. Redistributions in binary form must reproduce the above copyright
 *      notice, this list of conditions and the following disclaimer in the
 *      documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include "thrpool.h"


threadpool_t* threadpool_new(int n) {
    threadpool_t *pool = malloc(sizeof(threadpool_t));
    pool->queue = NULL;
    pool->queue_len = 0;

    pthread_mutex_init(&pool->queue_lock, NULL);
    pthread_mutex_init(&pool->avlbl_lock, NULL);
    pthread_cond_init(&pool->available, NULL);

    /* create pool threads */
    for(int i=0;i<n;i++) {
        pthread_t t;
        if (pthread_create(&t, NULL, threadpool_func, pool) != 0) {
            perror("Error: threadpool_new: pthread_create");
            return NULL;
        }
    }

    return pool;
}

void* threadpool_func(void *data) {
    threadpool_t *pool = (threadpool_t*)data;

    for(;;) {
        threadpool_job *job = threadpool_get_job(pool);
        if(job == NULL) {
            break;
        }

        job->callback(job->data);

        free(job);
    }
    return NULL;
}

threadpool_job* threadpool_get_job(threadpool_t *pool) {
    pthread_mutex_lock(&pool->queue_lock);

    threadpool_job *job = NULL;
    while(job == NULL) {
        if(pool->queue_len == 0) {
            pthread_cond_wait(&pool->available, &pool->queue_lock);
            continue;
        } else {
            pool_queue_t *q = pool->queue;
            job = q->job;
            pool->queue = q->next;
            pool->queue_len--;
            free(q);
        }
    }

    pthread_mutex_unlock(&pool->queue_lock);

    return job;
}

void threadpool_run(threadpool_t *pool, job_callback_f func, void *data) {
    threadpool_job *job = malloc(sizeof(threadpool_job));
    job->callback = func;
    job->data = data;

    pool_queue_t *q = malloc(sizeof(pool_queue_t));
    q->job = job;
    q->next = NULL;

    pthread_mutex_lock(&pool->queue_lock);
    if(pool->queue == NULL) {
        pool->queue = q;
    } else {
        pool_queue_t *last_elem = pool->queue;
        while(last_elem->next != NULL) {
            last_elem = last_elem->next;
        }
        last_elem->next = q;
    }
    pool->queue_len++;

    if(pool->queue_len == 1) {
        pthread_cond_signal(&pool->available);
    }

    pthread_mutex_unlock(&pool->queue_lock);

}