X-Git-Url: https://code.citadel.org/?a=blobdiff_plain;f=citadel%2Fthreads.c;h=1f066a0cd6e79e88f390db5bad0cec7f5d97e500;hb=cb8cd32734a354007d10d75959ba990ac8baaa19;hp=c13a2aed52c700e216e70b1d3095e8efe2879772;hpb=2a557d1e1aeee5b4bb8d829a1cff40c37f8d4d2c;p=citadel.git diff --git a/citadel/threads.c b/citadel/threads.c index c13a2aed5..1f066a0cd 100644 --- a/citadel/threads.c +++ b/citadel/threads.c @@ -1,11 +1,21 @@ /* - * $Id$ + * Thread handling stuff for Citadel server * - * Citadel "system dependent" stuff. - * See COPYING for copyright information. + * Copyright (c) 1987-2011 by the citadel.org team * - * Here's where we have the Citadel thread implimentation + * This program is open source software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 3 of the License, or + * (at your option) any later version. * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */ #include @@ -17,7 +27,9 @@ #include #include #include +#include +#include "sysdep.h" #if TIME_WITH_SYS_TIME # include # include @@ -29,6 +41,14 @@ # endif #endif +#ifdef HAVE_SYSCALL_H +# include +#else +# if HAVE_SYS_SYSCALL_H +# include +# endif +#endif + #include #include "threads.h" @@ -39,14 +59,10 @@ #include "citserver.h" #include "sysdep_decls.h" #include "context.h" +#include "event_client.h" -/* - * define this to use the new worker_thread method of handling connections - */ -//#define NEW_WORKER /* - * New thread interface. * To create a thread you must call one of the create thread functions. * You must pass it the address of (a pointer to a CtdlThreadNode initialised to NULL) like this * struct CtdlThreadNode *node = NULL; @@ -60,8 +76,6 @@ static int num_threads = 0; /* Current number of threads */ static int num_workers = 0; /* Current number of worker threads */ -long statcount = 0; /* are we doing a stats check? */ -static long stats_done = 0; CtdlThreadNode *CtdlThreadList = NULL; CtdlThreadNode *CtdlThreadSchedList = NULL; @@ -100,9 +114,6 @@ int try_critical_section(int which_one) * transaction; this could lead to deadlock. */ if ( (which_one != S_FLOORCACHE) -#ifdef DEBUG_MEMORY_LEAKS - && (which_one != S_DEBUGMEMLEAKS) -#endif && (which_one != S_RPLIST) ) { cdb_check_handles(); @@ -116,16 +127,13 @@ int try_critical_section(int which_one) */ void begin_critical_section(int which_one) { - /* CtdlLogPrintf(CTDL_DEBUG, "begin_critical_section(%d)\n", which_one); */ + /* syslog(LOG_DEBUG, "begin_critical_section(%d)\n", which_one); */ /* For all types of critical sections except those listed here, * ensure nobody ever tries to do a critical section within a * transaction; this could lead to deadlock. */ if ( (which_one != S_FLOORCACHE) -#ifdef DEBUG_MEMORY_LEAKS - && (which_one != S_DEBUGMEMLEAKS) -#endif && (which_one != S_RPLIST) ) { cdb_check_handles(); @@ -162,7 +170,7 @@ void ctdl_thread_internal_init_tsd(void) int ret; if ((ret = citthread_key_create(&ThreadKey, ctdl_thread_internal_dest_tsd))) { - CtdlLogPrintf(CTDL_EMERG, "citthread_key_create: %s\n", strerror(ret)); + syslog(LOG_EMERG, "citthread_key_create: %s\n", strerror(ret)); exit(CTDLEXIT_DB); } } @@ -243,7 +251,7 @@ void ctdl_thread_internal_init(void) /* Get ourself a thread entry */ this_thread = malloc(sizeof(CtdlThreadNode)); if (this_thread == NULL) { - CtdlLogPrintf(CTDL_EMERG, "Thread system, can't allocate CtdlThreadNode, exiting\n"); + syslog(LOG_EMERG, "Thread system, can't allocate CtdlThreadNode, exiting\n"); return; } // Ensuring this is zero'd means we make sure the thread doesn't start doing its thing until we are ready. @@ -258,7 +266,7 @@ void ctdl_thread_internal_init(void) this_thread->state = CTDL_THREAD_RUNNING; if ((ret = citthread_attr_init(&this_thread->attr))) { - CtdlLogPrintf(CTDL_EMERG, "Thread system, citthread_attr_init: %s\n", strerror(ret)); + syslog(LOG_EMERG, "Thread system, citthread_attr_init: %s\n", strerror(ret)); free(this_thread); return; } @@ -281,30 +289,6 @@ void ctdl_thread_internal_init(void) } -/* - * A function to update a threads load averages - */ - void ctdl_thread_internal_update_avgs(CtdlThreadNode *this_thread) - { - struct timeval now, result; - double last_duration; - - gettimeofday(&now, NULL); - timersub(&now, &(this_thread->last_state_change), &result); - /* I don't think these mutex's are needed here */ - citthread_mutex_lock(&this_thread->ThreadMutex); - // result now has a timeval for the time we spent in the last state since we last updated - last_duration = (double)result.tv_sec + ((double)result.tv_usec / (double) 1000000); - if (this_thread->state == CTDL_THREAD_SLEEPING) - this_thread->avg_sleeping += last_duration; - if (this_thread->state == CTDL_THREAD_RUNNING) - this_thread->avg_running += last_duration; - if (this_thread->state == CTDL_THREAD_BLOCKED) - this_thread->avg_blocked += last_duration; - memcpy (&this_thread->last_state_change, &now, sizeof (struct timeval)); - citthread_mutex_unlock(&this_thread->ThreadMutex); -} - /* * A function to chenge the state of a thread */ @@ -313,7 +297,6 @@ void ctdl_thread_internal_change_state (CtdlThreadNode *this_thread, enum CtdlTh /* * Wether we change state or not we need update the load values */ - ctdl_thread_internal_update_avgs(this_thread); /* This mutex not needed here? */ citthread_mutex_lock(&this_thread->ThreadMutex); /* To prevent race condition of a sleeping thread */ if ((new_state == CTDL_THREAD_STOP_REQ) && (this_thread->state > CTDL_THREAD_STOP_REQ)) @@ -333,10 +316,14 @@ void CtdlThreadStopAll(void) { /* First run any registered shutdown hooks. This probably doesn't belong here. */ PerformSessionHooks(EVT_SHUTDOWN); - + + /* then close all tcp ports so nobody else can talk to us anymore. */ + CtdlShutdownServiceHooks(); //FIXME: The signalling of the condition should not be in the critical_section // We need to build a list of threads we are going to signal and then signal them afterwards + ShutDownEventQueue(); + CtdlThreadNode *this_thread; begin_critical_section(S_THREAD_LIST); @@ -345,15 +332,14 @@ void CtdlThreadStopAll(void) GC_thread->state = CTDL_THREAD_STOP_REQ; while(this_thread) { -#ifdef THREADS_USESIGNALS if (!citthread_equal(this_thread->tid, GC_thread->tid)) citthread_kill(this_thread->tid, SIGHUP); -#endif + ctdl_thread_internal_change_state (this_thread, CTDL_THREAD_STOP_REQ); citthread_cond_signal(&this_thread->ThreadCond); citthread_cond_signal(&this_thread->SleepCond); this_thread->stop_ticker = time(NULL); - CtdlLogPrintf(CTDL_DEBUG, "Thread system stopping thread \"%s\" (0x%08lx).\n", + syslog(LOG_DEBUG, "Thread system stopping thread \"%s\" (0x%08lx).\n", this_thread->name, this_thread->tid); this_thread = this_thread->next; } @@ -368,7 +354,7 @@ void CtdlThreadWakeAll(void) { CtdlThreadNode *this_thread; - CtdlLogPrintf(CTDL_DEBUG, "Thread system waking all threads.\n"); + syslog(LOG_DEBUG, "Thread system waking all threads.\n"); begin_critical_section(S_THREAD_LIST); this_thread = CtdlThreadList; @@ -410,16 +396,18 @@ double CtdlThreadGetWorkerAvg(void) double CtdlThreadGetLoadAvg(void) { - double load_avg[3] ; + double load_avg[3] = {0.0, 0.0, 0.0}; - int ret; + int ret = 0; int smp_num_cpus; /* Borrowed this straight from procps */ smp_num_cpus = sysconf(_SC_NPROCESSORS_ONLN); if(smp_num_cpus<1) smp_num_cpus=1; /* SPARC glibc is buggy */ +#ifdef HAVE_GETLOADAVG ret = getloadavg(load_avg, 3); +#endif if (ret < 0) return 0; return load_avg[0] / smp_num_cpus; @@ -452,7 +440,7 @@ const char *CtdlThreadName(const char *name) if (!CT) { - CtdlLogPrintf(CTDL_WARNING, "Thread system WARNING. Attempt to CtdlThreadRename() a non thread. %s\n", name); + syslog(LOG_WARNING, "Thread system WARNING. Attempt to CtdlThreadRename() a non thread. %s\n", name); return NULL; } old_name = CT->name; @@ -475,14 +463,14 @@ void CtdlThreadCancel(CtdlThreadNode *thread) this_thread = thread; if (!this_thread) { - CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC. Attempt to CtdlThreadCancel() a non thread.\n"); + syslog(LOG_EMERG, "Thread system PANIC. Attempt to CtdlThreadCancel() a non thread.\n"); CtdlThreadStopAll(); return; } if (!this_thread->thread_func) { - CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC. Attempt to CtdlThreadCancel() the garbage collector.\n"); + syslog(LOG_EMERG, "Thread system PANIC. Attempt to CtdlThreadCancel() the garbage collector.\n"); CtdlThreadStopAll(); return; } @@ -501,20 +489,20 @@ int CtdlThreadCheckStop(void) if (!CT) { - CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC, CtdlThreadCheckStop() called by a non thread.\n"); + syslog(LOG_EMERG, "Thread system PANIC, CtdlThreadCheckStop() called by a non thread.\n"); CtdlThreadStopAll(); return -1; } state = CT->state; -#ifdef THREADS_USESIGNALS if (CT->signal) { - CtdlLogPrintf(CTDL_DEBUG, "Thread \"%s\" caught signal %d.\n", CT->name, CT->signal); + syslog(LOG_DEBUG, "Thread \"%s\" caught signal %d.\n", CT->name, CT->signal); + if (CT->signal == SIGHUP) + CT->state = CTDL_THREAD_STOP_REQ; CT->signal = 0; } -#endif if(state == CTDL_THREAD_STOP_REQ) { CT->state = CTDL_THREAD_STOPPING; @@ -544,10 +532,10 @@ void CtdlThreadStop(CtdlThreadNode *thread) return; if (!(this_thread->thread_func)) return; // Don't stop garbage collector -#ifdef THREADS_USESIGNALS + if (!citthread_equal(this_thread->tid, GC_thread->tid)) citthread_kill(this_thread->tid, SIGHUP); -#endif + ctdl_thread_internal_change_state (this_thread, CTDL_THREAD_STOP_REQ); citthread_cond_signal(&this_thread->ThreadCond); citthread_cond_signal(&this_thread->SleepCond); @@ -565,7 +553,7 @@ void CtdlThreadSleep(int secs) if (!CT) { - CtdlLogPrintf(CTDL_WARNING, "CtdlThreadSleep() called by something that is not a thread. Should we die?\n"); + syslog(LOG_WARNING, "CtdlThreadSleep() called by something that is not a thread. Should we die?\n"); return; } @@ -593,10 +581,20 @@ static void ctdl_internal_thread_cleanup(void *arg) * In here we were called by the current thread because it is exiting * NB. WE ARE THE CURRENT THREAD */ - CtdlLogPrintf(CTDL_NOTICE, "Thread \"%s\" (0x%08lx) exited.\n", CT->name, CT->tid); + if (CT) + { + const char *name = CT->name; + const pid_t tid = CT->tid; + + syslog(LOG_NOTICE, "Thread \"%s\" (0x%08lx) exited.\n", name, (unsigned long) tid); + } + else + { + syslog(LOG_NOTICE, "some ((unknown ? ? ?) Thread exited.\n"); + } #ifdef HAVE_BACKTRACE - eCrash_UnregisterThread(); +/// eCrash_UnregisterThread(); #endif citthread_mutex_lock(&CT->ThreadMutex); @@ -604,64 +602,22 @@ static void ctdl_internal_thread_cleanup(void *arg) citthread_mutex_unlock(&CT->ThreadMutex); } -/* - * A quick function to show the load averages - */ -void ctdl_thread_internal_calc_loadavg(void) -{ - CtdlThreadNode *that_thread; - double load_avg, worker_avg; - int workers = 0; - - that_thread = CtdlThreadList; - load_avg = 0; - worker_avg = 0; - while(that_thread) - { - /* Update load averages */ - ctdl_thread_internal_update_avgs(that_thread); - citthread_mutex_lock(&that_thread->ThreadMutex); - that_thread->load_avg = (that_thread->avg_sleeping + that_thread->avg_running) / (that_thread->avg_sleeping + that_thread->avg_running + that_thread->avg_blocked) * 100; - that_thread->avg_sleeping /= 2; - that_thread->avg_running /= 2; - that_thread->avg_blocked /= 2; - load_avg += that_thread->load_avg; - if (that_thread->flags & CTDLTHREAD_WORKER) - { - worker_avg += that_thread->load_avg; - workers++; - } -#ifdef WITH_THREADLOG - CtdlLogPrintf(CTDL_DEBUG, "CtdlThread, \"%s\" (%lu) \"%s\" %.2f %.2f %.2f %.2f\n", - that_thread->name, - that_thread->tid, - CtdlThreadStates[that_thread->state], - that_thread->avg_sleeping, - that_thread->avg_running, - that_thread->avg_blocked, - that_thread->load_avg); -#endif - citthread_mutex_unlock(&that_thread->ThreadMutex); - that_thread = that_thread->next; - } - CtdlThreadLoadAvg = load_avg/num_threads; - CtdlThreadWorkerAvg = worker_avg/workers; -#ifdef WITH_THREADLOG - CtdlLogPrintf(CTDL_INFO, "System load average %.2f, workers averag %.2f, threads %d, workers %d, sessions %d\n", CtdlThreadGetLoadAvg(), CtdlThreadWorkerAvg, num_threads, num_workers, num_sessions); -#endif -} - - /* * Garbage collection routine. * Gets called by main() in a loop to clean up the thread list periodically. */ void CtdlThreadGC (void) { + + + return; + /* FIXME this is a big deal, but I think it's causing corruption */ + + CtdlThreadNode *this_thread, *that_thread; int workers = 0, sys_workers; int ret=0; - + begin_critical_section(S_THREAD_LIST); /* Handle exiting of garbage collector thread */ @@ -669,7 +625,7 @@ void CtdlThreadGC (void) CtdlThreadList->state = CTDL_THREAD_EXITED; #ifdef WITH_THREADLOG - CtdlLogPrintf(CTDL_DEBUG, "Thread system running garbage collection.\n"); + syslog(LOG_DEBUG, "Thread system running garbage collection.\n"); #endif /* * Woke up to do garbage collection @@ -682,7 +638,10 @@ void CtdlThreadGC (void) if ((that_thread->state == CTDL_THREAD_STOP_REQ || that_thread->state == CTDL_THREAD_STOPPING) && (!citthread_equal(that_thread->tid, citthread_self()))) - CtdlLogPrintf(CTDL_DEBUG, "Waiting for thread %s (0x%08lx) to exit.\n", that_thread->name, that_thread->tid); + { + syslog(LOG_DEBUG, "Waiting for thread %s (0x%08lx) to exit.\n", that_thread->name, that_thread->tid); + terminate_stuck_sessions(); + } else { /** @@ -694,9 +653,9 @@ void CtdlThreadGC (void) if (that_thread->stop_ticker + 5 == time(NULL)) { - CtdlLogPrintf(CTDL_DEBUG, "Thread System: The thread \"%s\" (0x%08lx) failed to self terminate within 5 ticks. It would be cancelled now.\n", that_thread->name, that_thread->tid); + syslog(LOG_DEBUG, "Thread System: The thread \"%s\" (0x%08lx) failed to self terminate within 5 ticks. It would be cancelled now.\n", that_thread->name, that_thread->tid); if ((that_thread->flags & CTDLTHREAD_WORKER) == 0) - CtdlLogPrintf(CTDL_INFO, "Thread System: A non worker thread would have been canceled this may cause message loss.\n"); + syslog(LOG_INFO, "Thread System: A non worker thread would have been canceled this may cause message loss.\n"); // that_thread->state = CTDL_THREAD_CANCELLED; that_thread->stop_ticker++; // citthread_cancel(that_thread->tid); @@ -714,7 +673,7 @@ void CtdlThreadGC (void) if (citthread_equal(that_thread->tid, citthread_self()) && that_thread->thread_func) { /* Sanity check */ end_critical_section(S_THREAD_LIST); - CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC, a thread is trying to clean up after itself.\n"); + syslog(LOG_EMERG, "Thread system PANIC, a thread is trying to clean up after itself.\n"); abort(); return; } @@ -722,7 +681,7 @@ void CtdlThreadGC (void) if (num_threads <= 0) { /* Sanity check */ end_critical_section(S_THREAD_LIST); - CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC, num_threads <= 0 and trying to do Garbage Collection.\n"); + syslog(LOG_EMERG, "Thread system PANIC, num_threads <= 0 and trying to do Garbage Collection.\n"); abort(); return; } @@ -749,17 +708,17 @@ void CtdlThreadGC (void) */ ret = citthread_join (that_thread->tid, NULL); if (ret == EDEADLK) - CtdlLogPrintf(CTDL_DEBUG, "Garbage collection on own thread.\n"); + syslog(LOG_DEBUG, "Garbage collection on own thread.\n"); else if (ret == EINVAL) - CtdlLogPrintf(CTDL_DEBUG, "Garbage collection, that thread already joined on.\n"); + syslog(LOG_DEBUG, "Garbage collection, that thread already joined on.\n"); else if (ret == ESRCH) - CtdlLogPrintf(CTDL_DEBUG, "Garbage collection, no thread to join on.\n"); + syslog(LOG_DEBUG, "Garbage collection, no thread to join on.\n"); else if (ret != 0) - CtdlLogPrintf(CTDL_DEBUG, "Garbage collection, citthread_join returned an unknown error(%d).\n", ret); + syslog(LOG_DEBUG, "Garbage collection, citthread_join returned an unknown error(%d).\n", ret); /* * Now we own that thread entry */ - CtdlLogPrintf(CTDL_INFO, "Garbage Collection for thread \"%s\" (0x%08lx).\n", + syslog(LOG_INFO, "Garbage Collection for thread \"%s\" (0x%08lx).\n", that_thread->name, that_thread->tid); citthread_mutex_destroy(&that_thread->ThreadMutex); citthread_cond_destroy(&that_thread->ThreadCond); @@ -774,7 +733,7 @@ void CtdlThreadGC (void) /* Sanity check number of worker threads */ if (workers != sys_workers) { - CtdlLogPrintf(CTDL_EMERG, + syslog(LOG_EMERG, "Thread system PANIC, discrepancy in number of worker threads. Counted %d, should be %d.\n", workers, sys_workers ); @@ -802,7 +761,6 @@ static void *ctdl_internal_thread_func (void *arg) begin_critical_section(S_THREAD_LIST); this_thread = (CtdlThreadNode *) arg; gettimeofday(&this_thread->start_time, NULL); /* Time this thread started */ -// citthread_mutex_lock(&this_thread->ThreadMutex); // Register the cleanup function to take care of when we exit. citthread_cleanup_push(ctdl_internal_thread_cleanup, NULL); @@ -815,8 +773,6 @@ static void *ctdl_internal_thread_func (void *arg) * Other wise there is a window to allow this threads creation to continue to full grown and * therby prevent a shutdown of the server. */ -// citthread_mutex_unlock(&this_thread->ThreadMutex); - if (!CtdlThreadCheckStop()) { citthread_mutex_lock(&this_thread->ThreadMutex); @@ -827,11 +783,14 @@ static void *ctdl_internal_thread_func (void *arg) // Register for tracing #ifdef HAVE_BACKTRACE - eCrash_RegisterThread(this_thread->name, 0); +/// eCrash_RegisterThread(this_thread->name, 0); #endif // Tell the world we are here - CtdlLogPrintf(CTDL_NOTICE, "Created a new thread \"%s\" (0x%08lx).\n", +#if defined(HAVE_SYSCALL_H) && defined (SYS_gettid) + this_thread->reltid = syscall(SYS_gettid); +#endif + syslog(LOG_NOTICE, "Created a new thread \"%s\" (0x%08lx).\n", this_thread->name, this_thread->tid); /* @@ -876,7 +835,7 @@ CtdlThreadNode *ctdl_internal_init_thread_struct(CtdlThreadNode *this_thread, lo citthread_cond_destroy(&(this_thread->ThreadCond)); citthread_mutex_destroy(&(this_thread->SleepMutex)); citthread_cond_destroy(&(this_thread->SleepCond)); - CtdlLogPrintf(CTDL_EMERG, "Thread system, citthread_attr_init: %s\n", strerror(ret)); + syslog(LOG_EMERG, "Thread system, citthread_attr_init: %s\n", strerror(ret)); free(this_thread); return NULL; } @@ -888,7 +847,7 @@ CtdlThreadNode *ctdl_internal_init_thread_struct(CtdlThreadNode *this_thread, lo if (flags & CTDLTHREAD_BIGSTACK) { #ifdef WITH_THREADLOG - CtdlLogPrintf(CTDL_INFO, "Thread system. Creating BIG STACK thread.\n"); + syslog(LOG_INFO, "Thread system. Creating BIG STACK thread.\n"); #endif if ((ret = citthread_attr_setstacksize(&this_thread->attr, THREADSTACKSIZE))) { citthread_mutex_unlock(&this_thread->ThreadMutex); @@ -897,7 +856,7 @@ CtdlThreadNode *ctdl_internal_init_thread_struct(CtdlThreadNode *this_thread, lo citthread_mutex_destroy(&(this_thread->SleepMutex)); citthread_cond_destroy(&(this_thread->SleepCond)); citthread_attr_destroy(&this_thread->attr); - CtdlLogPrintf(CTDL_EMERG, "Thread system, citthread_attr_setstacksize: %s\n", + syslog(LOG_EMERG, "Thread system, citthread_attr_setstacksize: %s\n", strerror(ret)); free(this_thread); return NULL; @@ -926,13 +885,13 @@ CtdlThreadNode *ctdl_internal_create_thread(char *name, long flags, void *(*thre if (num_threads >= 32767) { - CtdlLogPrintf(CTDL_EMERG, "Thread system. Thread list full.\n"); + syslog(LOG_EMERG, "Thread system. Thread list full.\n"); return NULL; } this_thread = malloc(sizeof(CtdlThreadNode)); if (this_thread == NULL) { - CtdlLogPrintf(CTDL_EMERG, "Thread system, can't allocate CtdlThreadNode, exiting\n"); + syslog(LOG_EMERG, "Thread system, can't allocate CtdlThreadNode, exiting\n"); return NULL; } @@ -940,7 +899,7 @@ CtdlThreadNode *ctdl_internal_create_thread(char *name, long flags, void *(*thre if (ctdl_internal_init_thread_struct(this_thread, flags) == NULL) { free(this_thread); - CtdlLogPrintf(CTDL_EMERG, "Thread system, can't initialise CtdlThreadNode, exiting\n"); + syslog(LOG_EMERG, "Thread system, can't initialise CtdlThreadNode, exiting\n"); return NULL; } /* @@ -961,8 +920,6 @@ CtdlThreadNode *ctdl_internal_create_thread(char *name, long flags, void *(*thre this_thread->thread_func = thread_func; this_thread->user_args = args; -// citthread_mutex_lock(&this_thread->ThreadMutex); - begin_critical_section(S_THREAD_LIST); /* * We pass this_thread into the thread as its args so that it can find out information @@ -972,7 +929,7 @@ CtdlThreadNode *ctdl_internal_create_thread(char *name, long flags, void *(*thre if ((ret = citthread_create(&this_thread->tid, &this_thread->attr, ctdl_internal_thread_func, this_thread) != 0)) { end_critical_section(S_THREAD_LIST); - CtdlLogPrintf(CTDL_ALERT, "Thread system, Can't create thread: %s\n", + syslog(LOG_ALERT, "Thread system, Can't create thread: %s\n", strerror(ret)); citthread_mutex_unlock(&this_thread->ThreadMutex); citthread_mutex_destroy(&(this_thread->ThreadMutex)); @@ -983,7 +940,6 @@ CtdlThreadNode *ctdl_internal_create_thread(char *name, long flags, void *(*thre free(this_thread); return NULL; } - num_threads++; // Increase the count of threads in the system. if(this_thread->flags & CTDLTHREAD_WORKER) num_workers++; @@ -992,9 +948,7 @@ CtdlThreadNode *ctdl_internal_create_thread(char *name, long flags, void *(*thre CtdlThreadList = this_thread; if (this_thread->next) this_thread->next->prev = this_thread; - ctdl_thread_internal_calc_loadavg(); -// citthread_mutex_unlock(&this_thread->ThreadMutex); end_critical_section(S_THREAD_LIST); return this_thread; @@ -1016,73 +970,10 @@ CtdlThreadNode *CtdlThreadCreate(char *name, long flags, void *(*thread_func) (v -/* - * Internal function to schedule a thread. - * Must be called from within a S_THREAD_LIST critical section - */ -CtdlThreadNode *CtdlThreadSchedule(char *name, long flags, void *(*thread_func) (void *arg), void *args, time_t when) -{ - CtdlThreadNode *this_thread; - - if (num_threads >= 32767) - { - CtdlLogPrintf(CTDL_EMERG, "Thread system. Thread list full.\n"); - return NULL; - } - - this_thread = malloc(sizeof(CtdlThreadNode)); - if (this_thread == NULL) { - CtdlLogPrintf(CTDL_EMERG, "Thread system, can't allocate CtdlThreadNode, exiting\n"); - return NULL; - } - /* Initialise the thread structure */ - if (ctdl_internal_init_thread_struct(this_thread, flags) == NULL) - { - free(this_thread); - CtdlLogPrintf(CTDL_EMERG, "Thread system, can't initialise CtdlThreadNode, exiting\n"); - return NULL; - } - - /* - * If we got here we are going to create the thread so we must initilise the structure - * first because most implimentations of threading can't create it in a stopped state - * and it might want to do things with its structure that aren't initialised otherwise. - */ - if(name) - { - this_thread->name = name; - } - else - { - this_thread->name = "Un-named Thread"; - } - - this_thread->flags = flags; - this_thread->thread_func = thread_func; - this_thread->user_args = args; - - /* - * When to start this thread - */ - this_thread->when = when; - - begin_critical_section(S_SCHEDULE_LIST); - this_thread->next = CtdlThreadSchedList; - CtdlThreadSchedList = this_thread; - if (this_thread->next) - this_thread->next->prev = this_thread; - end_critical_section(S_SCHEDULE_LIST); - - return this_thread; -} - - - CtdlThreadNode *ctdl_thread_internal_start_scheduled (CtdlThreadNode *this_thread) { int ret = 0; -// citthread_mutex_lock(&that_thread->ThreadMutex); begin_critical_section(S_THREAD_LIST); /* * We pass this_thread into the thread as its args so that it can find out information @@ -1092,8 +983,7 @@ CtdlThreadNode *ctdl_thread_internal_start_scheduled (CtdlThreadNode *this_threa if ((ret = citthread_create(&this_thread->tid, &this_thread->attr, ctdl_internal_thread_func, this_thread) != 0)) { end_critical_section(S_THREAD_LIST); - CtdlLogPrintf(CTDL_DEBUG, "Failed to start scheduled thread \"%s\": %s\n", this_thread->name, strerror(ret)); -// citthread_mutex_unlock(&this_thread->ThreadMutex); + syslog(LOG_DEBUG, "Failed to start scheduled thread \"%s\": %s\n", this_thread->name, strerror(ret)); citthread_mutex_destroy(&(this_thread->ThreadMutex)); citthread_cond_destroy(&(this_thread->ThreadCond)); citthread_mutex_destroy(&(this_thread->SleepMutex)); @@ -1112,9 +1002,7 @@ CtdlThreadNode *ctdl_thread_internal_start_scheduled (CtdlThreadNode *this_threa CtdlThreadList = this_thread; if (this_thread->next) this_thread->next->prev = this_thread; -// citthread_mutex_unlock(&that_thread->ThreadMutex); - ctdl_thread_internal_calc_loadavg(); end_critical_section(S_THREAD_LIST); @@ -1138,7 +1026,7 @@ void ctdl_thread_internal_check_scheduled(void) now = time(NULL); #ifdef WITH_THREADLOG - CtdlLogPrintf(CTDL_DEBUG, "Checking for scheduled threads to start.\n"); + syslog(LOG_DEBUG, "Checking for scheduled threads to start.\n"); #endif this_thread = CtdlThreadSchedList; @@ -1159,14 +1047,14 @@ void ctdl_thread_internal_check_scheduled(void) that_thread->next = that_thread->prev = NULL; #ifdef WITH_THREADLOG - CtdlLogPrintf(CTDL_DEBUG, "About to start scheduled thread \"%s\".\n", that_thread->name); + syslog(LOG_DEBUG, "About to start scheduled thread \"%s\".\n", that_thread->name); #endif if (CT->state > CTDL_THREAD_STOP_REQ) { /* Only start it if the system is not stopping */ if (ctdl_thread_internal_start_scheduled (that_thread)) { #ifdef WITH_THREADLOG - CtdlLogPrintf(CTDL_INFO, "Thread system, Started a scheduled thread \"%s\" (0x%08lx).\n", + syslog(LOG_INFO, "Thread system, Started a scheduled thread \"%s\" (0x%08lx).\n", that_thread->name, that_thread->tid); #endif } @@ -1175,7 +1063,7 @@ void ctdl_thread_internal_check_scheduled(void) #ifdef WITH_THREADLOG else { - CtdlLogPrintf(CTDL_DEBUG, "Thread \"%s\" will start in %ld seconds.\n", + syslog(LOG_DEBUG, "Thread \"%s\" will start in %ld seconds.\n", that_thread->name, that_thread->when - time(NULL)); } #endif @@ -1217,7 +1105,7 @@ int CtdlThreadSelect(int n, fd_set *readfds, fd_set *writefds, fd_set *exceptfds citthread_mutex_lock(&CT->ThreadMutex); /* To prevent race condition of a sleeping thread */ if (GC_thread->state > CTDL_THREAD_STOP_REQ && CT->state <= CTDL_THREAD_STOP_REQ) { - CtdlLogPrintf(CTDL_DEBUG, "Thread %s (0x%08lx) refused stop request.\n", CT->name, CT->tid); + syslog(LOG_DEBUG, "Thread %s (0x%08lx) refused stop request.\n", CT->name, CT->tid); CT->state = CTDL_THREAD_RUNNING; } citthread_mutex_unlock(&CT->ThreadMutex); @@ -1234,39 +1122,11 @@ void *new_worker_thread(void *arg); extern void close_masters (void); -void *simulation_worker (void*arg) { - struct CitContext *this; - - this = CreateNewContext(); - CtdlThreadSleep(1); - this->kill_me = 1; - this->state = CON_IDLE; - dead_session_purge(1); - begin_critical_section(S_SESSION_TABLE); - stats_done++; - end_critical_section(S_SESSION_TABLE); - return NULL; -} - - -void *simulation_thread (void *arg) -{ - long stats = statcount; - - while(stats && !CtdlThreadCheckStop()) { - CtdlThreadCreate("Connection simulation worker", CTDLTHREAD_BIGSTACK, simulation_worker, NULL); - stats--; - } - CtdlThreadStopAll(); - return NULL; -} void go_threading(void) { int i; CtdlThreadNode *last_worker; - struct timeval start, now, result; - double last_duration; /* * Initialise the thread system @@ -1274,19 +1134,13 @@ void go_threading(void) ctdl_thread_internal_init(); /* Second call to module init functions now that threading is up */ - if (!statcount) - initialise_modules(1); - else { - CtdlLogPrintf(CTDL_EMERG, "Running connection simulation stats\n"); - gettimeofday(&start, NULL); - CtdlThreadCreate("Connection simulation master", CTDLTHREAD_BIGSTACK, simulation_thread, NULL); - } - + initialise_modules(1); + CtdlThreadCreate("select_on_master", CTDLTHREAD_BIGSTACK, select_on_master, NULL); /* * This thread is now used for garbage collection of other threads in the thread list */ - CtdlLogPrintf(CTDL_INFO, "Startup thread %d becoming garbage collector,\n", citthread_self()); + syslog(LOG_INFO, "Startup thread %ld becoming garbage collector,\n", (long) citthread_self()); /* * We do a lot of locking and unlocking of the thread list in here. @@ -1302,13 +1156,11 @@ void go_threading(void) if (exit_signal) { CtdlThreadStopAll(); -// close_masters(); } check_sched_shutdown(); if (CT->state > CTDL_THREAD_STOP_REQ) { begin_critical_section(S_THREAD_LIST); - ctdl_thread_internal_calc_loadavg(); end_critical_section(S_THREAD_LIST); ctdl_thread_internal_check_scheduled(); /* start scheduled threads */ @@ -1335,7 +1187,7 @@ void go_threading(void) if (last_worker) { #ifdef WITH_THREADLOG - CtdlLogPrintf(CTDL_DEBUG, "Thread system, stopping excess worker thread \"%s\" (0x%08lx).\n", + syslog(LOG_DEBUG, "Thread system, stopping excess worker thread \"%s\" (0x%08lx).\n", last_worker->name, last_worker->tid ); @@ -1351,36 +1203,26 @@ void go_threading(void) /* FIXME: come up with a better way to dynamically alter the number of threads * based on the system load */ - if (!statcount) { -#ifdef NEW_WORKER - if ((((CtdlThreadGetWorkers() < config.c_max_workers) && (CtdlThreadGetWorkers() <= num_sessions) ) || CtdlThreadGetWorkers() < config.c_min_workers) && (CT->state > CTDL_THREAD_STOP_REQ)) -#else - if ((((CtdlThreadGetWorkers() < config.c_max_workers) && (CtdlThreadGetWorkerAvg() > 60)) || CtdlThreadGetWorkers() < config.c_min_workers) && (CT->state > CTDL_THREAD_STOP_REQ)) -#endif /* NEW_WORKER */ + if ( (((CtdlThreadGetWorkers() < config.c_max_workers) + && (CtdlThreadGetWorkerAvg() > 60)) + || CtdlThreadGetWorkers() < config.c_min_workers) + && (CT->state > CTDL_THREAD_STOP_REQ) + ) { /* Only start new threads if we are not going to overload the machine */ /* Temporarily set to 10 should be enough to make sure we don't stranglew the server * at least until we make this a config option */ if (CtdlThreadGetLoadAvg() < ((double)10.00)) { for (i=0; i<5 ; i++) { -#ifdef NEW_WORKER - CtdlThreadCreate("Worker Thread (new)", - CTDLTHREAD_BIGSTACK + CTDLTHREAD_WORKER, - new_worker_thread, - NULL - ); -#else CtdlThreadCreate("Worker Thread", CTDLTHREAD_BIGSTACK + CTDLTHREAD_WORKER, worker_thread, NULL ); -#endif /* NEW_WORKER */ } } else - CtdlLogPrintf (CTDL_WARNING, "Server strangled due to machine load average too high.\n"); - } + syslog(LOG_WARNING, "Server strangled due to machine load average too high.\n"); } CtdlThreadGC(); @@ -1401,260 +1243,8 @@ void go_threading(void) * If the above loop exits we must be shutting down since we obviously have no threads */ ctdl_thread_internal_cleanup(); - - if (statcount) { - gettimeofday(&now, NULL); - timersub(&now, &start, &result); - last_duration = (double)result.tv_sec + ((double)result.tv_usec / (double) 1000000); - CtdlLogPrintf(CTDL_EMERG, "Simulated %ld connections in %f seconds\n", stats_done, last_duration); - } } -/* - * Starting a new implimentation of a worker thread. - * This new implimentation will be faster and do more work per thread. - */ - -/* - * Select on master socket. - * First worker thread in here acquires the lock and builds an FDSET of master sockets. - * then it goes into a loop selecting on the master sockets timing out every few milliseconds. - * If it times out it rebiulds its list and loops. - * If the select succeeds it creates a new context and returns. - * During this time the other workers are selecting on existing contexts or sleeping. - */ -void select_on_master(void) -{ - fd_set readfds; - struct ServiceFunctionHook *serviceptr; - int ssock; /* Descriptor for client socket */ - int highest; - int m, i; - int retval = 0; - struct timeval tv; - CitContext *con; - const char *old_name; - - - - old_name = CtdlThreadName("select_on_master"); - - /* Initialize the fdset. */ - FD_ZERO(&readfds); - highest = 0; - - /* First, add the various master sockets to the fdset. */ - for (serviceptr = ServiceHookTable; serviceptr != NULL; serviceptr = serviceptr->next ) { - m = serviceptr->msock; - FD_SET(m, &readfds); - if (m > highest) { - highest = m; - } - } - - tv.tv_sec = 1; /* wake up every 1 sec if no input */ - tv.tv_usec = 0; - retval = CtdlThreadSelect(highest + 1, &readfds, NULL, NULL, &tv); - - /* Select got an error or we are shutting down so get out */ - if (retval == 0 || CtdlThreadCheckStop()) { - CtdlThreadName(old_name); - return; - } - - /* Select says something happened on one of our master sockets so now we handle it */ - for (serviceptr = ServiceHookTable; serviceptr != NULL; serviceptr = serviceptr->next ) { - if (FD_ISSET(serviceptr->msock, &readfds)) { - ssock = accept(serviceptr->msock, NULL, 0); - if (ssock >= 0) { - CtdlLogPrintf(CTDL_DEBUG, "New client socket %d\n", ssock); - /* The master socket is non-blocking but the client - * sockets need to be blocking, otherwise certain - * operations barf on FreeBSD. Not a fatal error. - */ - if (fcntl(ssock, F_SETFL, 0) < 0) { - CtdlLogPrintf(CTDL_EMERG, - "citserver: Can't set socket to blocking: %s\n", - strerror(errno)); - } - - /* New context will be created already - * set up in the CON_EXECUTING state. - */ - con = CreateNewContext(); - CT->Context = con; - - /* Assign our new socket number to it. */ - con->client_socket = ssock; - con->h_command_function = serviceptr->h_command_function; - con->h_async_function = serviceptr->h_async_function; - con->ServiceName = serviceptr->ServiceName; - /* Determine whether it's a local socket */ - if (serviceptr->sockpath != NULL) - con->is_local_socket = 1; - - /* Set the SO_REUSEADDR socket option */ - i = 1; - setsockopt(ssock, SOL_SOCKET, SO_REUSEADDR, &i, sizeof(i)); - - become_session(con); - begin_session(con); - serviceptr->h_greeting_function(); - become_session(NULL); - con->state = CON_IDLE; - break; - } - } - } - - CtdlThreadName(old_name); -} - -/* - * Select on client socket. - * First worker thread in here acquires the lock and builds an FDSET of client sockets. - * then it selects on the client sockets timing out after 1 second. - * If it times out the thread goes off to check on housekeeping etc. - * If the select succeeds the thread goes off to handle the client request. - * If the list of client connections is empty the threads all sleep for one second - */ -CitContext *select_on_client(void) -{ - fd_set readfds; - struct timeval tv; - int retval = 0; - int highest=0; - const char *old_name; - - - old_name = CtdlThreadName("select_on_client"); - - /* Initialise the fdset */ - FD_ZERO(&readfds); - FD_SET(CT->Context->client_socket, &readfds); - highest = CT->Context->client_socket; - /* Now we can select on any connections that are waiting */ - - if (!CtdlThreadCheckStop()) - { - tv.tv_sec = config.c_sleeping; /* wake up every second if no input */ - tv.tv_usec = 0; - retval = select(highest + 1, &readfds, NULL, NULL, &tv); - } - else /* Shutting down? */ - { - CtdlThreadName(old_name); - return(NULL); - } - - - /* Now figure out who made this select() unblock. - * First, check for an error or exit condition. - */ - if (retval < 0) { - if (errno == EBADF) { - CtdlLogPrintf(CTDL_NOTICE, "select() failed: (%s)\n", - strerror(errno)); - } - if (errno != EINTR) { - CtdlLogPrintf(CTDL_EMERG, "Exiting (%s)\n", strerror(errno)); - CtdlThreadStopAll(); - } else if (!CtdlThreadCheckStop()) { - CtdlLogPrintf(CTDL_DEBUG, "Un handled select failure.\n"); - } - CtdlThreadName(old_name); - return NULL; - } - else if(retval == 0) - { - CtdlThreadName(old_name); - CT->Context->kill_me = 1; - CT->Context = NULL; - return CT->Context; - } - - CT->Context->state = CON_EXECUTING; - CT->Context->input_waiting = 1; - - CtdlThreadName(old_name); - return (CT->Context); -} - - - -/* - * Do the worker threads work when needed - */ -int execute_session(CitContext *bind_me) -{ - int force_purge; - - become_session(bind_me); - - /* If the client has sent a command, execute it. */ - if (CC->input_waiting) { - CC->h_command_function(); - CC->input_waiting = 0; - } - - /* If there are asynchronous messages waiting and the - * client supports it, do those now */ - if ((CC->is_async) && (CC->async_waiting) - && (CC->h_async_function != NULL)) { - CC->h_async_function(); - CC->async_waiting = 0; - } - - force_purge = CC->kill_me; - if (force_purge) - CT->Context = NULL; - become_session(NULL); - bind_me->state = CON_IDLE; - return force_purge; -} - - - - -/* - * A new worker_thread loop. - */ - -void *new_worker_thread(void *arg) -{ - CitContext *bind_me; - int force_purge; - - while (!CtdlThreadCheckStop()) { - - /* make doubly sure we're not holding any stale db handles - * which might cause a deadlock. - */ - cdb_check_handles(); - force_purge = 0; - bind_me = NULL; /* Which session shall we handle? */ - - if (CT->Context == NULL) - select_on_master(); - if (CtdlThreadCheckStop()) - break; - - if (CT->Context) - bind_me = select_on_client(); - if (CtdlThreadCheckStop()) - break; - - if (bind_me) - force_purge = execute_session(bind_me); - - dead_session_purge(force_purge); - if (CtdlThreadCheckStop()) - break; - - do_housekeeping(); - } - return NULL; -}