From cdc27df7c8d4c2b8074a1bcb260cf1c4824ff5b1 Mon Sep 17 00:00:00 2001 From: Richard Fuchs Date: Fri, 5 Dec 2014 08:33:37 -0500 Subject: [PATCH] serialize fetcher and worker threads fetching callids depends on the workers finishing their processing and removing the processed acc records, so we must wait to fetch more callids until the workers are done --- mediator.c | 24 ++++++++++++++++++++---- 1 file changed, 20 insertions(+), 4 deletions(-) diff --git a/mediator.c b/mediator.c index ab67912..8556633 100644 --- a/mediator.c +++ b/mediator.c @@ -37,6 +37,8 @@ pthread_t signal_thread, static GQueue callid_process_queue = G_QUEUE_INIT; static pthread_cond_t callid_process_cond = PTHREAD_COND_INITIALIZER; static pthread_mutex_t callid_process_lock = PTHREAD_MUTEX_INITIALIZER; +static int process_threads_busy; +static pthread_cond_t process_idle_cond = PTHREAD_COND_INITIALIZER; /**********************************************************************/ @@ -252,8 +254,6 @@ static void *callid_fetcher(void *p) { if (medmysql_fetch_callids(&callids)) abort(); - /* XXX check length of queue, sleep if too large */ - if (!callids.length) { pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL); usleep(config_interval * 1000000); @@ -270,11 +270,20 @@ static void *callid_fetcher(void *p) { callids.head->prev = callid_process_queue.tail; callid_process_queue.length += callids.length; } + /* our queue is now invalid and will be cleared on next iteration */ + /* work to do! */ + + /* this wakes up all worker threads. we then wait for them to finish + * before fetching the next batch. this is necessary so we don't keep + * polling the same callids. */ pthread_cond_broadcast(&callid_process_cond); + pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL); + do + pthread_cond_wait(&process_idle_cond, &callid_process_lock); + while (process_threads_busy); + pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL); pthread_mutex_unlock(&callid_process_lock); - - /* our queue is now invalid and will be cleared on next iteration */ } medmysql_cleanup(); @@ -303,6 +312,10 @@ static void *callid_worker(void *p) { if (medmysql_batch_start(&batches)) abort(); + pthread_mutex_lock(&callid_process_lock); + process_threads_busy++; + pthread_mutex_unlock(&callid_process_lock); + while (!mediator_shutdown) { pthread_mutex_lock(&callid_process_lock); @@ -330,9 +343,12 @@ static void *callid_worker(void *p) { syslog(LOG_DEBUG, "Runtime for record group was %"PRIu64" ms.", runtime); #endif + process_threads_busy--; pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL); + pthread_cond_signal(&process_idle_cond); pthread_cond_wait(&callid_process_cond, &callid_process_lock); pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL); + process_threads_busy++; #ifdef WITH_TIME_CALC gettimeofday(&tv_start, NULL);