This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via e86d4d5217db1828d5cdfca1c019b27609278aac (commit) via c81023996a845207b689e3f7f36f99a720969d9a (commit) from e06a92b3380444c8c64551e984229cd80dd96699 (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit e86d4d5217db1828d5cdfca1c019b27609278aac Author: Phil Carns <[email protected]> Date: Mon Aug 1 21:48:19 2011 -0500 configurable num threads in aesocket commit c81023996a845207b689e3f7f36f99a720969d9a Author: Phil Carns <[email protected]> Date: Mon Aug 1 21:21:34 2011 -0500 add thread pool support to opcache ----------------------------------------------------------------------- Summary of changes: code/src/aesop/opcache.c | 105 ++++++++++++++++++++++++- code/src/aesop/opcache.h | 8 ++ code/src/common/resources/aesocket/aesocket.c | 36 ++++++++- 3 files changed, 144 insertions(+), 5 deletions(-) Diff of changes: diff --git a/code/src/aesop/opcache.c b/code/src/aesop/opcache.c index 0857720..122365e 100644 --- a/code/src/aesop/opcache.c +++ b/code/src/aesop/opcache.c @@ -16,13 +16,81 @@ struct ae_opcache int array_count; int size; int count; - triton_mutex_t mutex; ae_ops_t free_list; #endif + triton_mutex_t mutex; + triton_cond_t cond; + int num_ops_in_use; + void (*completion_fn)(struct ae_opcache* opcache, struct ae_op* op); + int num_threads; + pthread_t* tids; + ae_ops_t thread_queue; int typesize; int member_offset; }; +static void* thread_pool_fn(void* foo); + +static void* thread_pool_fn(void* foo) +{ + ae_opcache_t cache = (ae_opcache_t)foo; + struct ae_op* op; + + while(cache->num_threads > 0) + { + triton_mutex_lock(&cache->mutex); + while((op = ae_ops_dequeue(&cache->thread_queue)) == NULL) + { + pthread_cond_wait(&cache->cond, &cache->mutex); + } + triton_mutex_unlock(&cache->mutex); + cache->completion_fn(cache, op); + } + return(NULL); +} + +triton_ret_t ae_opcache_set_threads(ae_opcache_t cache, + void(*completion_fn)(ae_opcache_t opcache, struct ae_op* op), + int num_threads) +{ + int i; + int ret; + + cache->num_threads = num_threads; + cache->tids = (pthread_t*)malloc(num_threads*sizeof(pthread_t)); + if(!cache->tids) + return(TRITON_ERR_NOMEM); + cache->completion_fn = completion_fn; + + for(i=0; i<num_threads; i++) + { + ret = pthread_create(&cache->tids[i], NULL, thread_pool_fn, cache); + if(ret != 0) + { + return(TRITON_ERR_UNKNOWN); + } + } + + return(TRITON_SUCCESS); +} + +void ae_opcache_complete_op_threaded(ae_opcache_t cache, struct ae_op* op) +{ + triton_mutex_lock(&cache->mutex); + if(cache->num_ops_in_use > 1) + { + ae_ops_enqueue(op, &cache->thread_queue); + triton_mutex_unlock(&cache->mutex); + triton_cond_signal(&cache->cond); + } else + { + triton_mutex_unlock(&cache->mutex); + cache->completion_fn(cache, op); + } + return; +} + + triton_ret_t ae_opcache_init(int typesize, int member_offset, int init_size, ae_opcache_t *cache) { struct ae_opcache *c; @@ -43,9 +111,13 @@ triton_ret_t ae_opcache_init(int typesize, int member_offset, int init_size, ae_ } c->count = 0; ae_ops_init(&(c->free_list)); - triton_mutex_init(&c->mutex, NULL); #endif + triton_mutex_init(&c->mutex, NULL); + triton_cond_init(&c->cond, NULL); + ae_ops_init(&c->thread_queue); + c->num_ops_in_use = 0; + c->num_threads = 0; c->typesize = typesize; c->member_offset = member_offset; *cache = c; @@ -85,8 +157,9 @@ triton_ret_t ae_opcache_double_size(ae_opcache_t cache) void ae_opcache_destroy(ae_opcache_t cache) { -#ifndef TRITON_OPCACHE_MALLOC int i; + int nthreads; +#ifndef TRITON_OPCACHE_MALLOC triton_mutex_lock(&cache->mutex); for(i = 0; i < cache->array_count; ++i) { @@ -94,6 +167,16 @@ void ae_opcache_destroy(ae_opcache_t cache) } triton_mutex_unlock(&cache->mutex); #endif + nthreads = cache->num_threads; + cache->num_threads = 0; + for(i=0; i<nthreads; i++) + { + pthread_join(cache->tids[i], NULL); + } + free(cache->tids); + triton_mutex_destroy(&cache->mutex); + triton_cond_destroy(&cache->cond); + ae_ops_destroy(&cache->thread_queue); free(cache); return; } @@ -123,6 +206,12 @@ struct ae_op *ae_opcache_get(ae_opcache_t cache) op = (struct ae_op *) ( (char*) malloc (cache->typesize) + cache->member_offset); ae_op_clear(op); + if(cache->num_threads > 0) + { + triton_mutex_lock(&cache->mutex); + cache->num_ops_in_use++; + triton_mutex_unlock(&cache->mutex); + } #else int aind, count; @@ -149,6 +238,7 @@ struct ae_op *ae_opcache_get(ae_opcache_t cache) { op = ae_ops_dequeue(&cache->free_list); } + cache->num_op_in_use++; triton_mutex_unlock(&cache->mutex); assert(op); #endif @@ -158,10 +248,17 @@ struct ae_op *ae_opcache_get(ae_opcache_t cache) void ae_opcache_put(ae_opcache_t cache, struct ae_op *op) { #ifdef TRITON_OPCACHE_MALLOC - free ((char*) op - cache->member_offset); + free ((char*) op - cache->member_offset); + if(cache->num_threads > 0) + { + triton_mutex_lock(&cache->mutex); + cache->num_ops_in_use--; + triton_mutex_unlock(&cache->mutex); + } #else triton_mutex_lock(&cache->mutex); ae_ops_enqueue(op, &cache->free_list); + cache->num_ops_in_use--; triton_mutex_unlock(&cache->mutex); #endif diff --git a/code/src/aesop/opcache.h b/code/src/aesop/opcache.h index a4b1da0..8baa328 100644 --- a/code/src/aesop/opcache.h +++ b/code/src/aesop/opcache.h @@ -21,6 +21,7 @@ typedef struct ae_opcache *ae_opcache_t; ae_opcache_put(__opcache, __op); \ } while(0) +void ae_opcache_complete_op_threaded(ae_opcache_t cache, struct ae_op* op); /** * Create an opcache. @@ -30,6 +31,13 @@ ae_ret_t ae_opcache_init(int typesize, int member_offset, int init_size, ae_opcache_t *cache); /** + * Activates a thread pool to run callbacks for the opcache + */ +triton_ret_t ae_opcache_set_threads(ae_opcache_t cache, + void(*completion_fn)(ae_opcache_t opcache, struct ae_op* op), + int num_threads); + +/** * Destroy the given opcache. * Note that all entries obtained from this cache are released and * invalidated. diff --git a/code/src/common/resources/aesocket/aesocket.c b/code/src/common/resources/aesocket/aesocket.c index a2eb690..24afb6b 100644 --- a/code/src/common/resources/aesocket/aesocket.c +++ b/code/src/common/resources/aesocket/aesocket.c @@ -3,6 +3,7 @@ #include "src/aesop/opcache.h" #include "src/common/resources/aesocket/aesocket.h" #include "src/common/triton-init.h" +#include "src/zeroconf/zeroconf.h" #include <assert.h> #include <pthread.h> @@ -24,6 +25,7 @@ struct ae_context static ae_opcache_t aesocket_opcache = NULL; static int triton_aesocket_resource_id; +static int nthreads = 0; struct aesocket_op { @@ -56,7 +58,7 @@ static void aesocket_fd_ready( triton_mutex_unlock(&aesocket_mutex); ev_io_stop(eloop, io); - ae_opcache_complete_op(aesocket_opcache, op, triton_ret_t, TRITON_SUCCESS); + ae_opcache_complete_op_threaded(aesocket_opcache, op); return; } @@ -231,12 +233,37 @@ __attribute__ ((constructor)) triton_aesocket_finalize, NULL, "aesop.control"); } +static void aesocket_completion_fn(ae_opcache_t cache, struct ae_op* op) +{ + ae_opcache_complete_op(cache, op, triton_ret_t, TRITON_SUCCESS); +} + +static triton_ret_t nthreads_updater(const char* key, const char* value) +{ + int ret; + + ret = sscanf(value, "%d", &nthreads); + if(ret != 1) + { + return(TRITON_ERR_INVAL); + } + + return(TRITON_SUCCESS); +} + triton_ret_t triton_aesocket_init( void) { triton_ret_t tret; int ret; + tret = triton_zeroconf_register("triton.aesocket.nthreads", "2", + nthreads_updater, "Number of threads the aesocket resource will use for callbacks."); + if(tret != TRITON_SUCCESS) + { + return(tret); + } + event_loop_thread = pthread_self(); ae_ops_init(&posted_oplist); @@ -252,6 +279,13 @@ triton_ret_t triton_aesocket_init( return tret; } + tret = ae_opcache_set_threads(aesocket_opcache, aesocket_completion_fn, + nthreads); + if (tret != TRITON_SUCCESS) + { + return tret; + } + return ae_resource_register(&triton_aesocket_resource, &triton_aesocket_resource_id); } hooks/post-receive -- Triton Repository