diff -r 1fdbf4170ef4 -r b8bf95b39952 src/server/util/thrpool.c --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/src/server/util/thrpool.c Sat Jan 14 13:53:44 2012 +0100 @@ -0,0 +1,122 @@ +/* + * 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 +#include +#include +#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;icallback(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); + +}