Slight alteration to worker thread startup code.
Altered the order of the shutdown so that we shutdown the modules and
the ports before closing the database. I'm curious about the remaining
locks in the data base after shutdown.
Still to come.
Wrappers for the pthread calls so that they can return to sysdep.c (the
portability file).
Further speed improvements.
Module to get thread data for user display.
setup.c snprintf.c \
stress.c support.c sysdep.c user_ops.c userlist.c \
whobbs.c \
- crc16.c journaling.c citadel_dirs.c
+ crc16.c journaling.c citadel_dirs.c threads.c
include Make_sources
file_ops.o msgbase.o euidindex.o \
locate_host.o housekeeping.o html.o \
internet_addressing.o journaling.o \
- parsedate.o genstamp.o ecrash.o \
+ parsedate.o genstamp.o ecrash.o threads.o\
clientsocket.o modules_init.o modules_upgrade.o $(AUTH) $(SERV_MODULES)
citserver: $(SERV_OBJS)
#include "citadel.h"
#include "server.h"
#include "sysdep_decls.h"
+#include "threads.h"
#include "citserver.h"
#include "config.h"
#include "database.h"
/* Close the AdjRefCount queue file */
AdjRefCount(-1, 0);
+ /* Do system-dependent stuff */
+ sysdep_master_cleanup();
+
/* Close databases */
lprintf(CTDL_INFO, "Closing databases\n");
close_databases();
- /* Do system-dependent stuff */
- sysdep_master_cleanup();
-
#ifdef DEBUG_MEMORY_LEAKS
dump_heap();
#endif
#include "room_ops.h"
#include "user_ops.h"
#include "database.h"
+#include "threads.h"
#ifndef HAVE_SNPRINTF
#include "snprintf.h"
#endif
-#include <pthread.h>
#include "citadel.h"
#include "server.h"
#include "citserver.h"
#include "database.h"
#include "msgbase.h"
#include "sysdep_decls.h"
+#include "threads.h"
#include "config.h"
#include "ctdl_module.h"
#include "room_ops.h"
#include "msgbase.h"
#include "citserver.h"
+#include "threads.h"
#ifndef HAVE_SNPRINTF
#include "snprintf.h"
#include "journaling.h"
#include "ctdl_module.h"
-
+#include "threads.h"
/*
* Terminate idle sessions. This function pounds through the session table
#include "journaling.h"
#include "ctdl_module.h"
+#include "threads.h"
struct jnlq *jnlq = NULL; /* journal queue */
#include "citadel_dirs.h"
#include "clientsocket.h"
#include "serv_network.h"
-
+#include "threads.h"
long config_msgnum;
struct addresses_to_be_filed *atbf = NULL;
#include "citserver.h"
#include "control.h"
#include "citadel_dirs.h"
+#include "threads.h"
struct floor *floorcache[MAXFLOORS];
#include <stdarg.h>
#include <grp.h>
#include <pwd.h>
-#ifdef HAVE_PTHREAD_H
-#include <pthread.h>
-#endif
#ifdef HAVE_SYS_PRCTL_H
#include <sys/prctl.h>
#endif
#include "server.h"
#include "serv_extensions.h"
#include "sysdep_decls.h"
+#include "threads.h"
#include "citserver.h"
#include "support.h"
#include "config.h"
master_cleanup(exit_signal);
return(0);
}
-
-
-
-void go_threading(void)
-{
- int i;
- struct CtdlThreadNode *last_worker;
-
- /*
- * Initialise the thread system
- */
- ctdl_thread_internal_init();
- /*
- * Now create a bunch of worker threads.
- */
- CtdlLogPrintf(CTDL_DEBUG, "Starting %d worker threads\n", config.c_min_workers);
- begin_critical_section(S_THREAD_LIST);
- i=0; /* Always start at least 1 worker thread */
- do
- {
- ctdl_internal_create_thread("Worker Thread", CTDLTHREAD_BIGSTACK + CTDLTHREAD_WORKER, worker_thread, NULL);
- } while (++i < config.c_min_workers);
- end_critical_section(S_THREAD_LIST);
-
- /* Second call to module init functions now that threading is up */
- initialise_modules(1);
-
- /*
- * 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", pthread_self());
-
- /*
- * We do a lot of locking and unlocking of the thread list in here.
- * We do this so that we can repeatedly release time for other threads
- * that may be waiting on the thread list.
- * We are a low priority thread so we can afford to do this
- */
-
- while (CtdlThreadGetCount())
- {
- if (CT->signal)
- exit_signal = CT->signal;
- if (exit_signal)
- CtdlThreadStopAll();
- 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 */
- }
-
- /* Reduce the size of the worker thread pool if necessary. */
- if ((CtdlThreadGetWorkers() > config.c_min_workers) && (CtdlThreadWorkerAvg < 20) && (CT->state > CTDL_THREAD_STOP_REQ))
- {
- /* Ask a worker thread to stop as we no longer need it */
- begin_critical_section(S_THREAD_LIST);
- last_worker = CtdlThreadList;
- while (last_worker)
- {
- pthread_mutex_lock(&last_worker->ThreadMutex);
- if (last_worker->flags & CTDLTHREAD_WORKER && last_worker->state > CTDL_THREAD_STOPPING)
- {
- pthread_mutex_unlock(&last_worker->ThreadMutex);
- break;
- }
- pthread_mutex_unlock(&last_worker->ThreadMutex);
- last_worker = last_worker->next;
- }
- end_critical_section(S_THREAD_LIST);
- if (last_worker)
- {
-#ifdef WITH_THREADLOG
- CtdlLogPrintf(CTDL_DEBUG, "Thread system, stopping excess worker thread \"%s\" (%ld).\n",
- last_worker->name,
- last_worker->tid
- );
-#endif
- CtdlThreadStop(last_worker);
- }
- }
-
- /*
- * If all our workers are working hard, start some more to help out
- * with things
- */
- /* FIXME: come up with a better way to dynamically alter the number of threads
- * based on the system load
- */
-// if ((CtdlThreadGetWorkers() < config.c_max_workers) && (CtdlThreadGetWorkers() < num_sessions))
- // && (CtdlThreadLoadAvg < 90) )
- if ((CtdlThreadGetWorkers() < config.c_max_workers) && (CtdlThreadGetWorkerAvg() > 60) && (CtdlThreadGetLoadAvg() < 90) && (CT->state > CTDL_THREAD_STOP_REQ))
- {
- for (i=0; i<5 ; i++)
-// for (i=0; i< (num_sessions - CtdlThreadGetWorkers()) ; i++)
-// for (i=0; i< (10 - (55 - CtdlThreadWorkerAvg) / CtdlThreadWorkerAvg / CtdlThreadGetWorkers()) ; i++)
- {
-// begin_critical_section(S_THREAD_LIST);
- CtdlThreadCreate("Worker Thread",
- CTDLTHREAD_BIGSTACK + CTDLTHREAD_WORKER,
- worker_thread,
- NULL
- );
-// end_critical_section(S_THREAD_LIST);
- }
- }
-
- CtdlThreadGC();
-
- if (CtdlThreadGetCount() <= 1) // Shutting down clean up the garbage collector
- {
- CtdlThreadGC();
- }
-
- if (CtdlThreadGetCount())
- CtdlThreadSleep(1);
- }
- /*
- * If the above loop exits we must be shutting down since we obviously have no threads
- */
- ctdl_thread_internal_cleanup();
-}
#include <errno.h>
#include <stdarg.h>
#include <grp.h>
-#ifdef HAVE_PTHREAD_H
-#include <pthread.h>
-#endif
#include <libcitadel.h>
#include "citadel.h"
#include "server.h"
#endif
#include "ctdl_module.h"
+#include "threads.h"
#ifdef DEBUG_MEMORY_LEAKS
struct igheap {
#endif
-pthread_mutex_t Critters[MAX_SEMAPHORES]; /* Things needing locking */
pthread_key_t MyConKey; /* TSD key for MyContext() */
int verbosity = DEFAULT_VERBOSITY; /* Logging level */
struct CitContext masterCC;
time_t last_purge = 0; /* Last dead session purge */
-static int num_threads = 0; /* Current number of threads */
-static int num_workers = 0; /* Current number of worker threads */
int num_sessions = 0; /* Current number of sessions */
int syslog_facility = LOG_DAEMON;
-
-void InitialiseSemaphores(void)
-{
- int i;
-
- /* Set up a bunch of semaphores to be used for critical sections */
- for (i=0; i<MAX_SEMAPHORES; ++i) {
- pthread_mutex_init(&Critters[i], NULL);
- }
-}
-
-
-
/*
* Some initialization stuff...
*/
-/*
- * Obtain a semaphore lock to begin a critical section.
- * but only if no one else has one
- */
-int try_critical_section(int 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();
- }
- return (pthread_mutex_trylock(&Critters[which_one]));
-}
-
-
-/*
- * Obtain a semaphore lock to begin a critical section.
- */
-void begin_critical_section(int which_one)
-{
- /* CtdlLogPrintf(CTDL_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();
- }
- pthread_mutex_lock(&Critters[which_one]);
-}
-
-/*
- * Release a semaphore lock to end a critical section.
- */
-void end_critical_section(int which_one)
-{
- pthread_mutex_unlock(&Critters[which_one]);
-}
-
-
/*
* This is a generic function to set up a master socket for listening on
}
}
-
-
-/*
- * 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;
- * pass in &node
- * If the thread is created *node will point to the thread control structure for the created thread.
- * If the thread creation fails *node remains NULL
- * Do not free the memory pointed to by *node, it doesn't belong to you.
- * This new interface duplicates much of the eCrash stuff. We should go for closer integration since that would
- * remove the need for the calls to eCrashRegisterThread and friends
- */
-
-
-struct CtdlThreadNode *CtdlThreadList = NULL;
-struct CtdlThreadNode *CtdlThreadSchedList = NULL;
-
-/*
- * Condition variable and Mutex for thread garbage collection
- */
-/*static pthread_mutex_t thread_gc_mutex = PTHREAD_MUTEX_INITIALIZER;
-static pthread_cond_t thread_gc_cond = PTHREAD_COND_INITIALIZER;
-*/
-static pthread_t GC_thread;
-static char *CtdlThreadStates[CTDL_THREAD_LAST_STATE];
-double CtdlThreadLoadAvg = 0;
-double CtdlThreadWorkerAvg = 0;
-pthread_key_t ThreadKey;
-
-/*
- * A function to destroy the TSD
- */
-static void ctdl_thread_internal_dest_tsd(void *arg)
-{
- if (arg != NULL) {
- check_handles(arg);
- free(arg);
- }
-}
-
-
-/*
- * A function to initialise the thread TSD
- */
-void ctdl_thread_internal_init_tsd(void)
-{
- int ret;
-
- if ((ret = pthread_key_create(&ThreadKey, ctdl_thread_internal_dest_tsd))) {
- lprintf(CTDL_EMERG, "pthread_key_create: %s\n",
- strerror(ret));
- exit(CTDLEXIT_DB);
- }
-}
-
-/*
- * Ensure that we have a key for thread-specific data.
- *
- * This should be called immediately after startup by any thread
- *
- */
-void CtdlThreadAllocTSD(void)
-{
- ThreadTSD *tsd;
-
- if (pthread_getspecific(ThreadKey) != NULL)
- return;
-
- tsd = malloc(sizeof(ThreadTSD));
-
- tsd->tid = NULL;
-
- memset(tsd->cursors, 0, sizeof tsd->cursors);
- tsd->self = NULL;
-
- pthread_setspecific(ThreadKey, tsd);
-}
-
-
-void ctdl_thread_internal_free_tsd(void)
-{
- ctdl_thread_internal_dest_tsd(pthread_getspecific(ThreadKey));
- pthread_setspecific(ThreadKey, NULL);
-}
-
-
-void ctdl_thread_internal_cleanup(void)
-{
- int i;
- struct CtdlThreadNode *this_thread, *that_thread;
-
- for (i=0; i<CTDL_THREAD_LAST_STATE; i++)
- {
- free (CtdlThreadStates[i]);
- }
-
- /* Clean up the scheduled thread list */
- this_thread = CtdlThreadSchedList;
- while (this_thread)
- {
- that_thread = this_thread;
- this_thread = this_thread->next;
- pthread_mutex_destroy(&that_thread->ThreadMutex);
- pthread_cond_destroy(&that_thread->ThreadCond);
- pthread_mutex_destroy(&that_thread->SleepMutex);
- pthread_cond_destroy(&that_thread->SleepCond);
- pthread_attr_destroy(&that_thread->attr);
- free(that_thread);
- }
- ctdl_thread_internal_free_tsd();
-}
-
-void ctdl_thread_internal_init(void)
-{
- struct CtdlThreadNode *this_thread;
- int ret = 0;
-
- GC_thread = pthread_self();
- CtdlThreadStates[CTDL_THREAD_INVALID] = strdup ("Invalid Thread");
- CtdlThreadStates[CTDL_THREAD_VALID] = strdup("Valid Thread");
- CtdlThreadStates[CTDL_THREAD_CREATE] = strdup("Thread being Created");
- CtdlThreadStates[CTDL_THREAD_CANCELLED] = strdup("Thread Cancelled");
- CtdlThreadStates[CTDL_THREAD_EXITED] = strdup("Thread Exited");
- CtdlThreadStates[CTDL_THREAD_STOPPING] = strdup("Thread Stopping");
- CtdlThreadStates[CTDL_THREAD_STOP_REQ] = strdup("Thread Stop Requested");
- CtdlThreadStates[CTDL_THREAD_SLEEPING] = strdup("Thread Sleeping");
- CtdlThreadStates[CTDL_THREAD_RUNNING] = strdup("Thread Running");
- CtdlThreadStates[CTDL_THREAD_BLOCKED] = strdup("Thread Blocked");
-
- /* Get ourself a thread entry */
- this_thread = malloc(sizeof(struct CtdlThreadNode));
- if (this_thread == NULL) {
- CtdlLogPrintf(CTDL_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.
- memset (this_thread, 0, sizeof(struct CtdlThreadNode));
-
- pthread_mutex_init (&(this_thread->ThreadMutex), NULL);
- pthread_cond_init (&(this_thread->ThreadCond), NULL);
- pthread_mutex_init (&(this_thread->SleepMutex), NULL);
- pthread_cond_init (&(this_thread->SleepCond), NULL);
-
- /* We are garbage collector so create us as running */
- this_thread->state = CTDL_THREAD_RUNNING;
-
- if ((ret = pthread_attr_init(&this_thread->attr))) {
- CtdlLogPrintf(CTDL_EMERG, "Thread system, pthread_attr_init: %s\n", strerror(ret));
- free(this_thread);
- return;
- }
-
- this_thread->name = "Garbage Collection Thread";
-
- this_thread->tid = GC_thread;
- CT = this_thread;
-
- num_threads++; // Increase the count of threads in the system.
-
- this_thread->next = CtdlThreadList;
- CtdlThreadList = this_thread;
- if (this_thread->next)
- this_thread->next->prev = this_thread;
- /* Set up start times */
- gettimeofday(&this_thread->start_time, NULL); /* Time this thread started */
- memcpy(&this_thread->last_state_change, &this_thread->start_time, sizeof (struct timeval)); /* Changed state so mark it. */
-}
-
-
-/*
- * A function to update a threads load averages
- */
- void ctdl_thread_internal_update_avgs(struct CtdlThreadNode *this_thread)
- {
- struct timeval now, result;
- double last_duration;
-
- gettimeofday(&now, NULL);
- timersub(&now, &(this_thread->last_state_change), &result);
- pthread_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));
- pthread_mutex_unlock(&this_thread->ThreadMutex);
-}
-
-/*
- * A function to chenge the state of a thread
- */
-void ctdl_thread_internal_change_state (struct CtdlThreadNode *this_thread, enum CtdlThreadState new_state)
-{
- /*
- * Wether we change state or not we need update the load values
- */
- ctdl_thread_internal_update_avgs(this_thread);
- pthread_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))
- this_thread->state = new_state;
- if (((new_state == CTDL_THREAD_SLEEPING) || (new_state == CTDL_THREAD_BLOCKED)) && (this_thread->state == CTDL_THREAD_RUNNING))
- this_thread->state = new_state;
- if ((new_state == CTDL_THREAD_RUNNING) && ((this_thread->state == CTDL_THREAD_SLEEPING) || (this_thread->state == CTDL_THREAD_BLOCKED)))
- this_thread->state = new_state;
- pthread_mutex_unlock(&this_thread->ThreadMutex);
-}
-
-
-/*
- * A function to tell all threads to exit
- */
-void CtdlThreadStopAll(void)
-{
- //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
-
- struct CtdlThreadNode *this_thread;
-
- begin_critical_section(S_THREAD_LIST);
- this_thread = CtdlThreadList;
- while(this_thread)
- {
-#ifdef THREADS_USESIGNALS
- pthread_kill(this_thread->tid, SIGHUP);
-#endif
- ctdl_thread_internal_change_state (this_thread, CTDL_THREAD_STOP_REQ);
- pthread_cond_signal(&this_thread->ThreadCond);
- pthread_cond_signal(&this_thread->SleepCond);
- CtdlLogPrintf(CTDL_DEBUG, "Thread system stopping thread \"%s\" (%ld).\n", this_thread->name, this_thread->tid);
- this_thread = this_thread->next;
- }
- end_critical_section(S_THREAD_LIST);
-}
-
-
-/*
- * A function to wake up all sleeping threads
- */
-void CtdlThreadWakeAll(void)
-{
- struct CtdlThreadNode *this_thread;
-
- CtdlLogPrintf(CTDL_DEBUG, "Thread system waking all threads.\n");
-
- begin_critical_section(S_THREAD_LIST);
- this_thread = CtdlThreadList;
- while(this_thread)
- {
- if (!this_thread->thread_func)
- {
- pthread_cond_signal(&this_thread->ThreadCond);
- pthread_cond_signal(&this_thread->SleepCond);
- }
- this_thread = this_thread->next;
- }
- end_critical_section(S_THREAD_LIST);
-}
-
-
-/*
- * A function to return the number of threads running in the system
- */
-int CtdlThreadGetCount(void)
-{
- return num_threads;
-}
-
-int CtdlThreadGetWorkers(void)
-{
- return num_workers;
-}
-
-double CtdlThreadGetWorkerAvg(void)
-{
- double ret;
-
- begin_critical_section(S_THREAD_LIST);
- ret = CtdlThreadWorkerAvg;
- end_critical_section(S_THREAD_LIST);
- return ret;
-}
-
-double CtdlThreadGetLoadAvg(void)
-{
- double ret;
-
- begin_critical_section(S_THREAD_LIST);
- ret = CtdlThreadLoadAvg;
- end_critical_section(S_THREAD_LIST);
- return ret;
-}
-
-
-
-
-/*
- * A function to rename a thread
- * Returns a const char *
- */
-const char *CtdlThreadName(const char *name)
-{
- const char *old_name;
-
- if (!CT)
- {
- CtdlLogPrintf(CTDL_WARNING, "Thread system WARNING. Attempt to CtdlThreadRename() a non thread. %s\n", name);
- return NULL;
- }
-// FIXME: do we need this lock? I think not since the pointer asignmaent should be atomic
- pthread_mutex_lock(&CT->ThreadMutex);
- old_name = CT->name;
- if (name)
- CT->name = name;
- pthread_mutex_unlock(&CT->ThreadMutex);
- return (old_name);
-}
-
-
-/*
- * A function to force a thread to exit
- */
-void CtdlThreadCancel(struct CtdlThreadNode *thread)
-{
- struct CtdlThreadNode *this_thread;
-
- if (!thread)
- this_thread = CT;
- else
- this_thread = thread;
- if (!this_thread)
- {
- CtdlLogPrintf(CTDL_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");
- CtdlThreadStopAll();
- return;
- }
-
- ctdl_thread_internal_change_state (this_thread, CTDL_THREAD_CANCELLED);
- pthread_cancel(this_thread->tid);
-}
-
-
-
-/*
- * A function for a thread to check if it has been asked to stop
- */
-int CtdlThreadCheckStop(void)
-{
- int state;
-
- if (!CT)
- {
- CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC, CtdlThreadCheckStop() called by a non thread.\n");
- CtdlThreadStopAll();
- return -1;
- }
-
- state = CT->state;
-
-#ifdef THREADS_USERSIGNALS
- if (CT->signal)
- CtdlLogPrintf(CTDL_DEBUG, "Thread \"%s\" caught signal %d.\n", CT->name, CT->signal);
-#endif
- pthread_mutex_lock(&CT->ThreadMutex);
- if(state == CTDL_THREAD_STOP_REQ)
- {
- CT->state = CTDL_THREAD_STOPPING;
- pthread_mutex_unlock(&CT->ThreadMutex);
- return -1;
- }
- else if((state < CTDL_THREAD_STOP_REQ) && (state > CTDL_THREAD_CREATE))
- {
- pthread_mutex_unlock(&CT->ThreadMutex);
- return -1;
- }
- pthread_mutex_unlock(&CT->ThreadMutex);
- return 0;
-}
-
-
-/*
- * A function to ask a thread to exit
- * The thread must call CtdlThreadCheckStop() periodically to determine if it should exit
- */
-void CtdlThreadStop(struct CtdlThreadNode *thread)
-{
- struct CtdlThreadNode *this_thread;
-
- if (!thread)
- this_thread = CT;
- else
- this_thread = thread;
- if (!this_thread)
- return;
- if (!(this_thread->thread_func))
- return; // Don't stop garbage collector
-#ifdef THREADS_USESIGNALS
- pthread_kill(this_thread->tid, SIGHUP);
-#endif
- ctdl_thread_internal_change_state (this_thread, CTDL_THREAD_STOP_REQ);
- pthread_cond_signal(&this_thread->ThreadCond);
- pthread_cond_signal(&this_thread->SleepCond);
-}
-
-/*
- * So we now have a sleep command that works with threads but it is in seconds
- */
-void CtdlThreadSleep(int secs)
-{
- struct timespec wake_time;
- struct timeval time_now;
-
-
- if (!CT)
- {
- CtdlLogPrintf(CTDL_WARNING, "CtdlThreadSleep() called by something that is not a thread. Should we die?\n");
- return;
- }
-
- memset (&wake_time, 0, sizeof(struct timespec));
- gettimeofday(&time_now, NULL);
- wake_time.tv_sec = time_now.tv_sec + secs;
- wake_time.tv_nsec = time_now.tv_usec * 10;
-
- ctdl_thread_internal_change_state (CT, CTDL_THREAD_SLEEPING);
-
- pthread_mutex_lock(&CT->ThreadMutex); /* Prevent something asking us to awaken before we've gone to sleep */
- pthread_cond_timedwait(&CT->SleepCond, &CT->ThreadMutex, &wake_time);
- pthread_mutex_unlock(&CT->ThreadMutex);
-
- ctdl_thread_internal_change_state (CT, CTDL_THREAD_RUNNING);
-}
-
-
-/*
- * Routine to clean up our thread function on exit
- */
-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\" (%ld) exited.\n", CT->name, CT->tid);
-
- #ifdef HAVE_BACKTRACE
- eCrash_UnregisterThread();
- #endif
-
- pthread_mutex_lock(&CT->ThreadMutex);
- CT->state = CTDL_THREAD_EXITED; // needs to be last thing else house keeping will unlink us too early
- pthread_mutex_unlock(&CT->ThreadMutex);
-}
-
-/*
- * A quick function to show the load averages
- */
-void ctdl_thread_internal_calc_loadavg(void)
-{
- struct 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);
- pthread_mutex_lock(&that_thread->ThreadMutex);
- that_thread->load_avg = that_thread->avg_sleeping + that_thread->avg_running + that_thread->avg_blocked;
- that_thread->load_avg = that_thread->avg_running / that_thread->load_avg * 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\" (%ld) \"%s\" %f %f %f %f.\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
- pthread_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 %f, workers averag %f, threads %d, workers %d, sessions %d\n", CtdlThreadLoadAvg, 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)
-{
- struct CtdlThreadNode *this_thread, *that_thread;
- int workers = 0;
- int ret=0;
-
- begin_critical_section(S_THREAD_LIST);
-
- /* Handle exiting of garbage collector thread */
- if(num_threads == 1)
- CtdlThreadList->state = CTDL_THREAD_EXITED;
-
-#ifdef WITH_THREADLOG
- CtdlLogPrintf(CTDL_DEBUG, "Thread system running garbage collection.\n");
-#endif
- /*
- * Woke up to do garbage collection
- */
- this_thread = CtdlThreadList;
- while(this_thread)
- {
- that_thread = this_thread;
- this_thread = this_thread->next;
-
- /* Do we need to clean up this thread? */
- pthread_mutex_lock(&that_thread->ThreadMutex);
- if (that_thread->state != CTDL_THREAD_EXITED)
- {
- if(that_thread->flags & CTDLTHREAD_WORKER)
- workers++; /* Sanity check on number of worker threads */
- pthread_mutex_unlock(&that_thread->ThreadMutex);
- continue;
- }
-
- if (pthread_equal(that_thread->tid, pthread_self()) && that_thread->thread_func)
- { /* Sanity check */
- pthread_mutex_unlock(&that_thread->ThreadMutex);
- end_critical_section(S_THREAD_LIST);
- CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC, a thread is trying to clean up after itself.\n");
- abort();
- return;
- }
-
- if (num_threads <= 0)
- { /* Sanity check */
- pthread_mutex_unlock(&that_thread->ThreadMutex);
- end_critical_section(S_THREAD_LIST);
- CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC, num_threads <= 0 and trying to do Garbage Collection.\n");
- abort();
- return;
- }
-
- if(that_thread->flags & CTDLTHREAD_WORKER)
- num_workers--; /* This is a wroker thread so reduce the count. */
- num_threads--;
- /* If we are unlinking the list head then the next becomes the list head */
- if(that_thread->prev)
- that_thread->prev->next = that_thread->next;
- else
- CtdlThreadList = that_thread->next;
- if(that_thread->next)
- that_thread->next->prev = that_thread->prev;
-
- pthread_mutex_unlock(&that_thread->ThreadMutex);
- pthread_cond_signal(&that_thread->ThreadCond);
- pthread_cond_signal(&that_thread->SleepCond); // Make sure this thread is awake
- pthread_mutex_lock(&that_thread->ThreadMutex); // Make sure it has done what its doing
- pthread_mutex_unlock(&that_thread->ThreadMutex);
- /*
- * Join on the thread to do clean up and prevent memory leaks
- * Also makes sure the thread has cleaned up after itself before we remove it from the list
- * We can join on the garbage collector thread the join should just return EDEADLCK
- */
- ret = pthread_join (that_thread->tid, NULL);
- if (ret == EDEADLK)
- CtdlLogPrintf(CTDL_DEBUG, "Garbage collection on own thread.\n");
- else if (ret == EINVAL)
- CtdlLogPrintf(CTDL_DEBUG, "Garbage collection, that thread already joined on.\n");
- else if (ret == ESRCH)
- CtdlLogPrintf(CTDL_DEBUG, "Garbage collection, no thread to join on.\n");
- else if (ret != 0)
- CtdlLogPrintf(CTDL_DEBUG, "Garbage collection, pthread_join returned an unknown error.\n");
- /*
- * Now we own that thread entry
- */
- CtdlLogPrintf(CTDL_INFO, "Garbage Collection for thread \"%s\" (%ld).\n", that_thread->name, that_thread->tid);
- pthread_mutex_destroy(&that_thread->ThreadMutex);
- pthread_cond_destroy(&that_thread->ThreadCond);
- pthread_mutex_destroy(&that_thread->SleepMutex);
- pthread_cond_destroy(&that_thread->SleepCond);
- pthread_attr_destroy(&that_thread->attr);
- free(that_thread);
- }
-
- /* Sanity check number of worker threads */
- if (workers != num_workers)
- {
- end_critical_section(S_THREAD_LIST);
- CtdlLogPrintf(CTDL_EMERG,
- "Thread system PANIC, discrepancy in number of worker threads. Counted %d, should be %d.\n",
- workers, num_workers
- );
- abort();
- }
- end_critical_section(S_THREAD_LIST);
-}
-
-
-
-
-/*
- * Runtime function for a Citadel Thread.
- * This initialises the threads environment and then calls the user supplied thread function
- * Note that this is the REAL thread function and wraps the users thread function.
- */
-static void *ctdl_internal_thread_func (void *arg)
-{
- struct CtdlThreadNode *this_thread;
- void *ret = NULL;
-
- /* lock and unlock the thread list.
- * This causes this thread to wait until all its creation stuff has finished before it
- * can continue its execution.
- */
- begin_critical_section(S_THREAD_LIST);
- this_thread = (struct CtdlThreadNode *) arg;
- gettimeofday(&this_thread->start_time, NULL); /* Time this thread started */
- pthread_mutex_lock(&this_thread->ThreadMutex);
-
- // Register the cleanup function to take care of when we exit.
- pthread_cleanup_push(ctdl_internal_thread_cleanup, NULL);
- // Get our thread data structure
- CtdlThreadAllocTSD();
- CT = this_thread;
- this_thread->pid = getpid();
- memcpy(&this_thread->last_state_change, &this_thread->start_time, sizeof (struct timeval)); /* Changed state so mark it. */
- /* Only change to running state if we weren't asked to stop during the create cycle
- * Other wise there is a window to allow this threads creation to continue to full grown and
- * therby prevent a shutdown of the server.
- */
- pthread_mutex_unlock(&this_thread->ThreadMutex);
-
- if (!CtdlThreadCheckStop())
- {
- pthread_mutex_lock(&this_thread->ThreadMutex);
- this_thread->state = CTDL_THREAD_RUNNING;
- pthread_mutex_unlock(&this_thread->ThreadMutex);
- }
- end_critical_section(S_THREAD_LIST);
-
- // Register for tracing
- #ifdef HAVE_BACKTRACE
- eCrash_RegisterThread(this_thread->name, 0);
- #endif
-
- // Tell the world we are here
- CtdlLogPrintf(CTDL_NOTICE, "Created a new thread \"%s\" (%ld). \n", this_thread->name, this_thread->tid);
-
-
-
- /*
- * run the thread to do the work but only if we haven't been asked to stop
- */
- if (!CtdlThreadCheckStop())
- ret = (this_thread->thread_func)(this_thread->user_args);
-
- /*
- * Our thread is exiting either because it wanted to end or because the server is stopping
- * We need to clean up
- */
- pthread_cleanup_pop(1); // Execute our cleanup routine and remove it
-
- return(ret);
-}
-
-
-
-/*
- * Internal function to create a thread.
- * Must be called from within a S_THREAD_LIST critical section
- */
-struct CtdlThreadNode *ctdl_internal_create_thread(char *name, long flags, void *(*thread_func) (void *arg), void *args)
-{
- int ret = 0;
- struct CtdlThreadNode *this_thread;
-
- if (num_threads >= 32767)
- {
- CtdlLogPrintf(CTDL_EMERG, "Thread system. Thread list full.\n");
- return NULL;
- }
-
- this_thread = malloc(sizeof(struct CtdlThreadNode));
- if (this_thread == NULL) {
- CtdlLogPrintf(CTDL_EMERG, "Thread system, can't allocate CtdlThreadNode, exiting\n");
- return NULL;
- }
- // Ensuring this is zero'd means we make sure the thread doesn't start doing its thing until we are ready.
- memset (this_thread, 0, sizeof(struct CtdlThreadNode));
-
- /* Create the mutex's early so we can use them */
- pthread_mutex_init (&(this_thread->ThreadMutex), NULL);
- pthread_cond_init (&(this_thread->ThreadCond), NULL);
- pthread_mutex_init (&(this_thread->SleepMutex), NULL);
- pthread_cond_init (&(this_thread->SleepCond), NULL);
-
- pthread_mutex_lock(&this_thread->ThreadMutex);
-
- this_thread->state = CTDL_THREAD_CREATE;
-
- if ((ret = pthread_attr_init(&this_thread->attr))) {
- pthread_mutex_unlock(&this_thread->ThreadMutex);
- pthread_mutex_destroy(&(this_thread->ThreadMutex));
- pthread_cond_destroy(&(this_thread->ThreadCond));
- pthread_mutex_destroy(&(this_thread->SleepMutex));
- pthread_cond_destroy(&(this_thread->SleepCond));
- CtdlLogPrintf(CTDL_EMERG, "Thread system, pthread_attr_init: %s\n", strerror(ret));
- free(this_thread);
- return NULL;
- }
-
- /* Our per-thread stacks need to be bigger than the default size,
- * otherwise the MIME parser crashes on FreeBSD, and the IMAP service
- * crashes on 64-bit Linux.
- */
- if (flags & CTDLTHREAD_BIGSTACK)
- {
-#ifdef WITH_THREADLOG
- CtdlLogPrintf(CTDL_INFO, "Thread system. Creating BIG STACK thread.\n");
-#endif
- if ((ret = pthread_attr_setstacksize(&this_thread->attr, THREADSTACKSIZE))) {
- pthread_mutex_unlock(&this_thread->ThreadMutex);
- pthread_mutex_destroy(&(this_thread->ThreadMutex));
- pthread_cond_destroy(&(this_thread->ThreadCond));
- pthread_mutex_destroy(&(this_thread->SleepMutex));
- pthread_cond_destroy(&(this_thread->SleepCond));
- pthread_attr_destroy(&this_thread->attr);
- CtdlLogPrintf(CTDL_EMERG, "Thread system, pthread_attr_setstacksize: %s\n",
- strerror(ret));
- free(this_thread);
- 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;
- /* Set this new thread with an avg_blocked of 2. We do this so that its creation affects the
- * load average for the system. If we don't do this then we create a mass of threads at the same time
- * because the creation didn't affect the load average.
- */
- this_thread->avg_blocked = 2;
-
- /*
- * We pass this_thread into the thread as its args so that it can find out information
- * about itself and it has a bit of storage space for itself, not to mention that the REAL
- * thread function needs to finish off the setup of the structure
- */
- if ((ret = pthread_create(&this_thread->tid, &this_thread->attr, ctdl_internal_thread_func, this_thread) != 0))
- {
-
- CtdlLogPrintf(CTDL_ALERT, "Thread system, Can't create thread: %s\n",
- strerror(ret));
- pthread_mutex_unlock(&this_thread->ThreadMutex);
- pthread_mutex_destroy(&(this_thread->ThreadMutex));
- pthread_cond_destroy(&(this_thread->ThreadCond));
- pthread_mutex_destroy(&(this_thread->SleepMutex));
- pthread_cond_destroy(&(this_thread->SleepCond));
- pthread_attr_destroy(&this_thread->attr);
- free(this_thread);
- return NULL;
- }
-
- num_threads++; // Increase the count of threads in the system.
- if(this_thread->flags & CTDLTHREAD_WORKER)
- num_workers++;
-
- this_thread->next = CtdlThreadList;
- CtdlThreadList = this_thread;
- if (this_thread->next)
- this_thread->next->prev = this_thread;
-
- pthread_mutex_unlock(&this_thread->ThreadMutex);
-
- ctdl_thread_internal_calc_loadavg();
- return this_thread;
-}
-
-/*
- * Wrapper function to create a thread
- * ensures the critical section and other protections are in place.
- * char *name = name to give to thread, if NULL, use generic name
- * int flags = flags to determine type of thread and standard facilities
- */
-struct CtdlThreadNode *CtdlThreadCreate(char *name, long flags, void *(*thread_func) (void *arg), void *args)
-{
- struct CtdlThreadNode *ret = NULL;
-
- begin_critical_section(S_THREAD_LIST);
- ret = ctdl_internal_create_thread(name, flags, thread_func, args);
- end_critical_section(S_THREAD_LIST);
- return ret;
-}
-
-
-
-/*
- * Internal function to schedule a thread.
- * Must be called from within a S_THREAD_LIST critical section
- */
-struct CtdlThreadNode *CtdlThreadSchedule(char *name, long flags, void *(*thread_func) (void *arg), void *args, time_t when)
-{
- int ret = 0;
- struct CtdlThreadNode *this_thread;
-
- if (num_threads >= 32767)
- {
- CtdlLogPrintf(CTDL_EMERG, "Thread system. Thread list full.\n");
- return NULL;
- }
-
- this_thread = malloc(sizeof(struct CtdlThreadNode));
- if (this_thread == NULL) {
- CtdlLogPrintf(CTDL_EMERG, "Thread system, can't allocate CtdlThreadNode, exiting\n");
- return NULL;
- }
- // Ensuring this is zero'd means we make sure the thread doesn't start doing its thing until we are ready.
- memset (this_thread, 0, sizeof(struct CtdlThreadNode));
-
- /* Create the mutex's early so we can use them */
- pthread_mutex_init (&(this_thread->ThreadMutex), NULL);
- pthread_cond_init (&(this_thread->ThreadCond), NULL);
- pthread_mutex_init (&(this_thread->SleepMutex), NULL);
- pthread_cond_init (&(this_thread->SleepCond), NULL);
-
- this_thread->state = CTDL_THREAD_CREATE;
-
- if ((ret = pthread_attr_init(&this_thread->attr))) {
- pthread_mutex_destroy(&(this_thread->ThreadMutex));
- pthread_cond_destroy(&(this_thread->ThreadCond));
- pthread_mutex_destroy(&(this_thread->SleepMutex));
- pthread_cond_destroy(&(this_thread->SleepCond));
- CtdlLogPrintf(CTDL_EMERG, "Thread system, pthread_attr_init: %s\n", strerror(ret));
- free(this_thread);
- return NULL;
- }
-
- /* Our per-thread stacks need to be bigger than the default size,
- * otherwise the MIME parser crashes on FreeBSD, and the IMAP service
- * crashes on 64-bit Linux.
- */
- if (flags & CTDLTHREAD_BIGSTACK)
- {
- CtdlLogPrintf(CTDL_INFO, "Thread system. Creating BIG STACK thread.\n");
- if ((ret = pthread_attr_setstacksize(&this_thread->attr, THREADSTACKSIZE))) {
- pthread_mutex_destroy(&(this_thread->ThreadMutex));
- pthread_cond_destroy(&(this_thread->ThreadCond));
- pthread_mutex_destroy(&(this_thread->SleepMutex));
- pthread_cond_destroy(&(this_thread->SleepCond));
- pthread_attr_destroy(&this_thread->attr);
- CtdlLogPrintf(CTDL_EMERG, "Thread system, pthread_attr_setstacksize: %s\n",
- strerror(ret));
- free(this_thread);
- 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;
- /* Set this new thread with an avg_blocked of 2. We do this so that its creation affects the
- * load average for the system. If we don't do this then we create a mass of threads at the same time
- * because the creation didn't affect the load average.
- */
- this_thread->avg_blocked = 2;
-
- /*
- * 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;
-}
-
-
-
-struct CtdlThreadNode *ctdl_thread_internal_start_scheduled (struct CtdlThreadNode *this_thread)
-{
- int ret = 0;
-
- /*
- * We pass this_thread into the thread as its args so that it can find out information
- * about itself and it has a bit of storage space for itself, not to mention that the REAL
- * thread function needs to finish off the setup of the structure
- */
- if ((ret = pthread_create(&this_thread->tid, &this_thread->attr, ctdl_internal_thread_func, this_thread) != 0))
- {
-
- CtdlLogPrintf(CTDL_ALERT, "Thread system, Can't create thread: %s\n",
- strerror(ret));
- return NULL;
- }
-
-
- num_threads++; // Increase the count of threads in the system.
- if(this_thread->flags & CTDLTHREAD_WORKER)
- num_workers++;
-
- this_thread->next = CtdlThreadList;
- CtdlThreadList = this_thread;
- if (this_thread->next)
- this_thread->next->prev = this_thread;
-
- return this_thread;
-}
-
-
-
-void ctdl_thread_internal_check_scheduled(void)
-{
- struct CtdlThreadNode *this_thread, *that_thread;
- time_t now;
-
- if (try_critical_section(S_SCHEDULE_LIST))
- return; /* If this list is locked we wait till the next chance */
-
- now = time(NULL);
-
-#ifdef WITH_THREADLOG
- CtdlLogPrintf(CTDL_DEBUG, "Checking for scheduled threads to start.\n");
-#endif
-
- this_thread = CtdlThreadSchedList;
- while(this_thread)
- {
- that_thread = this_thread;
- this_thread = this_thread->next;
-
- if (now > that_thread->when)
- {
- /* Unlink from schedule list */
- if (that_thread->prev)
- that_thread->prev->next = that_thread->next;
- else
- CtdlThreadSchedList = that_thread->next;
- if (that_thread->next)
- that_thread->next->prev = that_thread->prev;
-
- that_thread->next = that_thread->prev = NULL;
-#ifdef WITH_THREADLOG
- CtdlLogPrintf(CTDL_DEBUG, "About to start scheduled thread \"%s\".\n", that_thread->name);
-#endif
- begin_critical_section(S_THREAD_LIST);
- if (CT->state > CTDL_THREAD_STOP_REQ)
- { /* Only start it if the system is not stopping */
- pthread_mutex_lock(&that_thread->ThreadMutex);
- if (ctdl_thread_internal_start_scheduled (that_thread) == NULL)
- {
-#ifdef WITH_THREADLOG
- CtdlLogPrintf(CTDL_DEBUG, "Failed to start scheduled thread \"%s\".\n", that_thread->name);
-#endif
- pthread_mutex_unlock(&that_thread->ThreadMutex);
- pthread_mutex_destroy(&(that_thread->ThreadMutex));
- pthread_cond_destroy(&(that_thread->ThreadCond));
- pthread_mutex_destroy(&(that_thread->SleepMutex));
- pthread_cond_destroy(&(that_thread->SleepCond));
- pthread_attr_destroy(&that_thread->attr);
- free(that_thread);
- }
- else
- {
- CtdlLogPrintf(CTDL_INFO, "Thread system, Started a scheduled thread \"%s\" (%ld).\n",
- that_thread->name, that_thread->tid);
- pthread_mutex_unlock(&that_thread->ThreadMutex);
- ctdl_thread_internal_calc_loadavg();
- }
- }
- end_critical_section(S_THREAD_LIST);
- }
- else
- {
-#ifdef WITH_THREADLOG
- CtdlLogPrintf(CTDL_DEBUG, "Thread \"%s\" will start in %ld seconds.\n", that_thread->name, that_thread->when - time(NULL));
-#endif
- }
- }
- end_critical_section(S_SCHEDULE_LIST);
-}
-
-
-/*
- * A warapper function for select so we can show a thread as blocked
- */
-int CtdlThreadSelect(int n, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout)
-{
- int ret;
-
- ctdl_thread_internal_change_state(CT, CTDL_THREAD_BLOCKED);
- ret = select(n, readfds, writefds, exceptfds, timeout);
- ctdl_thread_internal_change_state(CT, CTDL_THREAD_RUNNING);
- return ret;
-}
-
/*
* Purge all sessions which have the 'kill_me' flag set.
* This function has code to prevent it from running more than once every
extern int enable_syslog;
void init_sysdep (void);
-void begin_critical_section (int which_one);
-void end_critical_section (int which_one);
int ig_tcp_server (char *ip_addr, int port_number, int queue_len,char **errormessage);
int ig_uds_server(char *sockpath, int queue_len, char **errormessage);
struct CitContext *MyContext (void);
void InitializeMasterCC(void);
void init_master_fdset(void);
void create_worker(void);
-void InitialiseSemaphores(void);
extern int num_sessions;
-/*
- * Thread stuff
- */
-#define CTDLTHREAD_BIGSTACK 0x0001
-#define CTDLTHREAD_WORKER 0x0002
-
-enum CtdlThreadState {
- CTDL_THREAD_INVALID,
- CTDL_THREAD_VALID,
- CTDL_THREAD_CREATE,
- CTDL_THREAD_CANCELLED,
- CTDL_THREAD_EXITED,
- CTDL_THREAD_STOPPING,
- CTDL_THREAD_STOP_REQ, /* Do NOT put any running states before this state */
- CTDL_THREAD_SLEEPING,
- CTDL_THREAD_BLOCKED,
- CTDL_THREAD_RUNNING,
- CTDL_THREAD_LAST_STATE
-};
-
-extern struct CtdlThreadNode {
- pthread_t tid; /* id as returned by pthread_create() */
- pid_t pid; /* pid, as best the OS will let us determine */
- time_t when; /* When to start a scheduled thread */
- struct CitConext *Context; /* The session context that this thread mught be working on or NULL if none */
- long number; /* A unigue number for this thread (not implimented yet) */
- int wakefd_recv; /* An fd that this thread can sleep on (not implimented yet) */
- int wakefd_send; /* An fd that this thread can send out on (Not implimented yet) */
- int signal; /* A field to store a signal we caught. */
- const char *name; /* A name for this thread */
- void *(*thread_func) (void *arg); /* The actual function that does this threads work */
- void *user_args; /* Arguments passed to this threads work function */
- long flags; /* Flags that describe this thread */
- enum CtdlThreadState state; /* Flag to show state of this thread */
- pthread_mutex_t ThreadMutex; /* A mutex to sync this thread to others if this thread allows (also used for sleeping) */
- pthread_cond_t ThreadCond; /* A condition variable to sync this thread with others */
- pthread_mutex_t SleepMutex; /* A mutex for sleeping */
- pthread_cond_t SleepCond; /* A condition variable for sleeping */
- pthread_attr_t attr; /* Attributes of this thread */
- struct timeval start_time; /* Time this thread was started */
- struct timeval last_state_change; /* Time when this thread last changed state */
- double avg_sleeping; /* Average sleeping time */
- double avg_running; /* Average running time */
- double avg_blocked; /* Average blocked time */
- double load_avg; /* Load average for this thread */
- struct CtdlThreadNode *prev; /* Previous thread in the thread table */
- struct CtdlThreadNode *next; /* Next thread in the thread table */
-} *CtdlThreadList;
-
-typedef struct {
- DB_TXN *tid; /* Transaction handle */
- DBC *cursors[MAXCDB]; /* Cursors, for traversals... */
- struct CtdlThreadNode *self; /* Pointer to this threads control structure */
-}ThreadTSD ;
-
-extern double CtdlThreadLoadAvg;
-extern double CtdlThreadWorkerAvg;
-extern pthread_key_t ThreadKey;
-
-void ctdl_thread_internal_init_tsd(void);
-void ctdl_internal_thread_gc (void);
-void ctdl_thread_internal_init(void);
-void ctdl_thread_internal_cleanup(void);
-void ctdl_thread_internal_calc_loadavg(void);
-void ctdl_thread_internal_free_tsd(void);
-struct CtdlThreadNode *ctdl_internal_create_thread(char *name, long flags, void *(*thread_func) (void *arg), void *args);
-void ctdl_thread_internal_check_scheduled(void);
-
-
extern int SyslogFacility(char *name);
extern int syslog_facility;
--- /dev/null
+/*
+ * $Id: sysdep.c 5882 2007-12-13 19:46:05Z davew $
+ *
+ * Citadel "system dependent" stuff.
+ * See copyright.txt for copyright information.
+ *
+ * Here's where we have the Citadel thread implimentation
+ *
+ */
+
+#include <errno.h>
+
+#include "threads.h"
+#include "ctdl_module.h"
+#include "modules_init.h"
+#include "housekeeping.h"
+#include "config.h"
+
+/*
+ * 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;
+ * pass in &node
+ * If the thread is created *node will point to the thread control structure for the created thread.
+ * If the thread creation fails *node remains NULL
+ * Do not free the memory pointed to by *node, it doesn't belong to you.
+ * This new interface duplicates much of the eCrash stuff. We should go for closer integration since that would
+ * remove the need for the calls to eCrashRegisterThread and friends
+ */
+
+static int num_threads = 0; /* Current number of threads */
+static int num_workers = 0; /* Current number of worker threads */
+
+struct CtdlThreadNode *CtdlThreadList = NULL;
+struct CtdlThreadNode *CtdlThreadSchedList = NULL;
+
+/*
+ * Condition variable and Mutex for thread garbage collection
+ */
+/*static pthread_mutex_t thread_gc_mutex = PTHREAD_MUTEX_INITIALIZER;
+static pthread_cond_t thread_gc_cond = PTHREAD_COND_INITIALIZER;
+*/
+static pthread_t GC_thread;
+static char *CtdlThreadStates[CTDL_THREAD_LAST_STATE];
+double CtdlThreadLoadAvg = 0;
+double CtdlThreadWorkerAvg = 0;
+pthread_key_t ThreadKey;
+
+pthread_mutex_t Critters[MAX_SEMAPHORES]; /* Things needing locking */
+
+
+
+void InitialiseSemaphores(void)
+{
+ int i;
+
+ /* Set up a bunch of semaphores to be used for critical sections */
+ for (i=0; i<MAX_SEMAPHORES; ++i) {
+ pthread_mutex_init(&Critters[i], NULL);
+ }
+}
+
+
+
+
+/*
+ * Obtain a semaphore lock to begin a critical section.
+ * but only if no one else has one
+ */
+int try_critical_section(int 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();
+ }
+ return (pthread_mutex_trylock(&Critters[which_one]));
+}
+
+
+/*
+ * Obtain a semaphore lock to begin a critical section.
+ */
+void begin_critical_section(int which_one)
+{
+ /* CtdlLogPrintf(CTDL_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();
+ }
+ pthread_mutex_lock(&Critters[which_one]);
+}
+
+/*
+ * Release a semaphore lock to end a critical section.
+ */
+void end_critical_section(int which_one)
+{
+ pthread_mutex_unlock(&Critters[which_one]);
+}
+
+
+/*
+ * A function to destroy the TSD
+ */
+static void ctdl_thread_internal_dest_tsd(void *arg)
+{
+ if (arg != NULL) {
+ check_handles(arg);
+ free(arg);
+ }
+}
+
+
+/*
+ * A function to initialise the thread TSD
+ */
+void ctdl_thread_internal_init_tsd(void)
+{
+ int ret;
+
+ if ((ret = pthread_key_create(&ThreadKey, ctdl_thread_internal_dest_tsd))) {
+ lprintf(CTDL_EMERG, "pthread_key_create: %s\n",
+ strerror(ret));
+ exit(CTDLEXIT_DB);
+ }
+}
+
+/*
+ * Ensure that we have a key for thread-specific data.
+ *
+ * This should be called immediately after startup by any thread
+ *
+ */
+void CtdlThreadAllocTSD(void)
+{
+ ThreadTSD *tsd;
+
+ if (pthread_getspecific(ThreadKey) != NULL)
+ return;
+
+ tsd = malloc(sizeof(ThreadTSD));
+
+ tsd->tid = NULL;
+
+ memset(tsd->cursors, 0, sizeof tsd->cursors);
+ tsd->self = NULL;
+
+ pthread_setspecific(ThreadKey, tsd);
+}
+
+
+void ctdl_thread_internal_free_tsd(void)
+{
+ ctdl_thread_internal_dest_tsd(pthread_getspecific(ThreadKey));
+ pthread_setspecific(ThreadKey, NULL);
+}
+
+
+void ctdl_thread_internal_cleanup(void)
+{
+ int i;
+ struct CtdlThreadNode *this_thread, *that_thread;
+
+ for (i=0; i<CTDL_THREAD_LAST_STATE; i++)
+ {
+ free (CtdlThreadStates[i]);
+ }
+
+ /* Clean up the scheduled thread list */
+ this_thread = CtdlThreadSchedList;
+ while (this_thread)
+ {
+ that_thread = this_thread;
+ this_thread = this_thread->next;
+ pthread_mutex_destroy(&that_thread->ThreadMutex);
+ pthread_cond_destroy(&that_thread->ThreadCond);
+ pthread_mutex_destroy(&that_thread->SleepMutex);
+ pthread_cond_destroy(&that_thread->SleepCond);
+ pthread_attr_destroy(&that_thread->attr);
+ free(that_thread);
+ }
+ ctdl_thread_internal_free_tsd();
+}
+
+void ctdl_thread_internal_init(void)
+{
+ struct CtdlThreadNode *this_thread;
+ int ret = 0;
+
+ GC_thread = pthread_self();
+ CtdlThreadStates[CTDL_THREAD_INVALID] = strdup ("Invalid Thread");
+ CtdlThreadStates[CTDL_THREAD_VALID] = strdup("Valid Thread");
+ CtdlThreadStates[CTDL_THREAD_CREATE] = strdup("Thread being Created");
+ CtdlThreadStates[CTDL_THREAD_CANCELLED] = strdup("Thread Cancelled");
+ CtdlThreadStates[CTDL_THREAD_EXITED] = strdup("Thread Exited");
+ CtdlThreadStates[CTDL_THREAD_STOPPING] = strdup("Thread Stopping");
+ CtdlThreadStates[CTDL_THREAD_STOP_REQ] = strdup("Thread Stop Requested");
+ CtdlThreadStates[CTDL_THREAD_SLEEPING] = strdup("Thread Sleeping");
+ CtdlThreadStates[CTDL_THREAD_RUNNING] = strdup("Thread Running");
+ CtdlThreadStates[CTDL_THREAD_BLOCKED] = strdup("Thread Blocked");
+
+ /* Get ourself a thread entry */
+ this_thread = malloc(sizeof(struct CtdlThreadNode));
+ if (this_thread == NULL) {
+ CtdlLogPrintf(CTDL_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.
+ memset (this_thread, 0, sizeof(struct CtdlThreadNode));
+
+ pthread_mutex_init (&(this_thread->ThreadMutex), NULL);
+ pthread_cond_init (&(this_thread->ThreadCond), NULL);
+ pthread_mutex_init (&(this_thread->SleepMutex), NULL);
+ pthread_cond_init (&(this_thread->SleepCond), NULL);
+
+ /* We are garbage collector so create us as running */
+ this_thread->state = CTDL_THREAD_RUNNING;
+
+ if ((ret = pthread_attr_init(&this_thread->attr))) {
+ CtdlLogPrintf(CTDL_EMERG, "Thread system, pthread_attr_init: %s\n", strerror(ret));
+ free(this_thread);
+ return;
+ }
+
+ this_thread->name = "Garbage Collection Thread";
+
+ this_thread->tid = GC_thread;
+ CT = this_thread;
+
+ num_threads++; // Increase the count of threads in the system.
+
+ this_thread->next = CtdlThreadList;
+ CtdlThreadList = this_thread;
+ if (this_thread->next)
+ this_thread->next->prev = this_thread;
+ /* Set up start times */
+ gettimeofday(&this_thread->start_time, NULL); /* Time this thread started */
+ memcpy(&this_thread->last_state_change, &this_thread->start_time, sizeof (struct timeval)); /* Changed state so mark it. */
+}
+
+
+/*
+ * A function to update a threads load averages
+ */
+ void ctdl_thread_internal_update_avgs(struct CtdlThreadNode *this_thread)
+ {
+ struct timeval now, result;
+ double last_duration;
+
+ gettimeofday(&now, NULL);
+ timersub(&now, &(this_thread->last_state_change), &result);
+ pthread_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));
+ pthread_mutex_unlock(&this_thread->ThreadMutex);
+}
+
+/*
+ * A function to chenge the state of a thread
+ */
+void ctdl_thread_internal_change_state (struct CtdlThreadNode *this_thread, enum CtdlThreadState new_state)
+{
+ /*
+ * Wether we change state or not we need update the load values
+ */
+ ctdl_thread_internal_update_avgs(this_thread);
+ pthread_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))
+ this_thread->state = new_state;
+ if (((new_state == CTDL_THREAD_SLEEPING) || (new_state == CTDL_THREAD_BLOCKED)) && (this_thread->state == CTDL_THREAD_RUNNING))
+ this_thread->state = new_state;
+ if ((new_state == CTDL_THREAD_RUNNING) && ((this_thread->state == CTDL_THREAD_SLEEPING) || (this_thread->state == CTDL_THREAD_BLOCKED)))
+ this_thread->state = new_state;
+ pthread_mutex_unlock(&this_thread->ThreadMutex);
+}
+
+
+/*
+ * A function to tell all threads to exit
+ */
+void CtdlThreadStopAll(void)
+{
+ //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
+
+ struct CtdlThreadNode *this_thread;
+
+ begin_critical_section(S_THREAD_LIST);
+ this_thread = CtdlThreadList;
+ while(this_thread)
+ {
+#ifdef THREADS_USESIGNALS
+ pthread_kill(this_thread->tid, SIGHUP);
+#endif
+ ctdl_thread_internal_change_state (this_thread, CTDL_THREAD_STOP_REQ);
+ pthread_cond_signal(&this_thread->ThreadCond);
+ pthread_cond_signal(&this_thread->SleepCond);
+ CtdlLogPrintf(CTDL_DEBUG, "Thread system stopping thread \"%s\" (%ld).\n", this_thread->name, this_thread->tid);
+ this_thread = this_thread->next;
+ }
+ end_critical_section(S_THREAD_LIST);
+}
+
+
+/*
+ * A function to wake up all sleeping threads
+ */
+void CtdlThreadWakeAll(void)
+{
+ struct CtdlThreadNode *this_thread;
+
+ CtdlLogPrintf(CTDL_DEBUG, "Thread system waking all threads.\n");
+
+ begin_critical_section(S_THREAD_LIST);
+ this_thread = CtdlThreadList;
+ while(this_thread)
+ {
+ if (!this_thread->thread_func)
+ {
+ pthread_cond_signal(&this_thread->ThreadCond);
+ pthread_cond_signal(&this_thread->SleepCond);
+ }
+ this_thread = this_thread->next;
+ }
+ end_critical_section(S_THREAD_LIST);
+}
+
+
+/*
+ * A function to return the number of threads running in the system
+ */
+int CtdlThreadGetCount(void)
+{
+ return num_threads;
+}
+
+int CtdlThreadGetWorkers(void)
+{
+ return num_workers;
+}
+
+double CtdlThreadGetWorkerAvg(void)
+{
+ double ret;
+
+ begin_critical_section(S_THREAD_LIST);
+ ret = CtdlThreadWorkerAvg;
+ end_critical_section(S_THREAD_LIST);
+ return ret;
+}
+
+double CtdlThreadGetLoadAvg(void)
+{
+ double ret;
+
+ begin_critical_section(S_THREAD_LIST);
+ ret = CtdlThreadLoadAvg;
+ end_critical_section(S_THREAD_LIST);
+ return ret;
+}
+
+
+
+
+/*
+ * A function to rename a thread
+ * Returns a const char *
+ */
+const char *CtdlThreadName(const char *name)
+{
+ const char *old_name;
+
+ if (!CT)
+ {
+ CtdlLogPrintf(CTDL_WARNING, "Thread system WARNING. Attempt to CtdlThreadRename() a non thread. %s\n", name);
+ return NULL;
+ }
+// FIXME: do we need this lock? I think not since the pointer asignmaent should be atomic
+ pthread_mutex_lock(&CT->ThreadMutex);
+ old_name = CT->name;
+ if (name)
+ CT->name = name;
+ pthread_mutex_unlock(&CT->ThreadMutex);
+ return (old_name);
+}
+
+
+/*
+ * A function to force a thread to exit
+ */
+void CtdlThreadCancel(struct CtdlThreadNode *thread)
+{
+ struct CtdlThreadNode *this_thread;
+
+ if (!thread)
+ this_thread = CT;
+ else
+ this_thread = thread;
+ if (!this_thread)
+ {
+ CtdlLogPrintf(CTDL_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");
+ CtdlThreadStopAll();
+ return;
+ }
+
+ ctdl_thread_internal_change_state (this_thread, CTDL_THREAD_CANCELLED);
+ pthread_cancel(this_thread->tid);
+}
+
+
+
+/*
+ * A function for a thread to check if it has been asked to stop
+ */
+int CtdlThreadCheckStop(void)
+{
+ int state;
+
+ if (!CT)
+ {
+ CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC, CtdlThreadCheckStop() called by a non thread.\n");
+ CtdlThreadStopAll();
+ return -1;
+ }
+
+ state = CT->state;
+
+#ifdef THREADS_USERSIGNALS
+ if (CT->signal)
+ CtdlLogPrintf(CTDL_DEBUG, "Thread \"%s\" caught signal %d.\n", CT->name, CT->signal);
+#endif
+ pthread_mutex_lock(&CT->ThreadMutex);
+ if(state == CTDL_THREAD_STOP_REQ)
+ {
+ CT->state = CTDL_THREAD_STOPPING;
+ pthread_mutex_unlock(&CT->ThreadMutex);
+ return -1;
+ }
+ else if((state < CTDL_THREAD_STOP_REQ) && (state > CTDL_THREAD_CREATE))
+ {
+ pthread_mutex_unlock(&CT->ThreadMutex);
+ return -1;
+ }
+ pthread_mutex_unlock(&CT->ThreadMutex);
+ return 0;
+}
+
+
+/*
+ * A function to ask a thread to exit
+ * The thread must call CtdlThreadCheckStop() periodically to determine if it should exit
+ */
+void CtdlThreadStop(struct CtdlThreadNode *thread)
+{
+ struct CtdlThreadNode *this_thread;
+
+ if (!thread)
+ this_thread = CT;
+ else
+ this_thread = thread;
+ if (!this_thread)
+ return;
+ if (!(this_thread->thread_func))
+ return; // Don't stop garbage collector
+#ifdef THREADS_USESIGNALS
+ pthread_kill(this_thread->tid, SIGHUP);
+#endif
+ ctdl_thread_internal_change_state (this_thread, CTDL_THREAD_STOP_REQ);
+ pthread_cond_signal(&this_thread->ThreadCond);
+ pthread_cond_signal(&this_thread->SleepCond);
+}
+
+/*
+ * So we now have a sleep command that works with threads but it is in seconds
+ */
+void CtdlThreadSleep(int secs)
+{
+ struct timespec wake_time;
+ struct timeval time_now;
+
+
+ if (!CT)
+ {
+ CtdlLogPrintf(CTDL_WARNING, "CtdlThreadSleep() called by something that is not a thread. Should we die?\n");
+ return;
+ }
+
+ memset (&wake_time, 0, sizeof(struct timespec));
+ gettimeofday(&time_now, NULL);
+ wake_time.tv_sec = time_now.tv_sec + secs;
+ wake_time.tv_nsec = time_now.tv_usec * 10;
+
+ ctdl_thread_internal_change_state (CT, CTDL_THREAD_SLEEPING);
+
+ pthread_mutex_lock(&CT->ThreadMutex); /* Prevent something asking us to awaken before we've gone to sleep */
+ pthread_cond_timedwait(&CT->SleepCond, &CT->ThreadMutex, &wake_time);
+ pthread_mutex_unlock(&CT->ThreadMutex);
+
+ ctdl_thread_internal_change_state (CT, CTDL_THREAD_RUNNING);
+}
+
+
+/*
+ * Routine to clean up our thread function on exit
+ */
+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\" (%ld) exited.\n", CT->name, CT->tid);
+
+ #ifdef HAVE_BACKTRACE
+ eCrash_UnregisterThread();
+ #endif
+
+ pthread_mutex_lock(&CT->ThreadMutex);
+ CT->state = CTDL_THREAD_EXITED; // needs to be last thing else house keeping will unlink us too early
+ pthread_mutex_unlock(&CT->ThreadMutex);
+}
+
+/*
+ * A quick function to show the load averages
+ */
+void ctdl_thread_internal_calc_loadavg(void)
+{
+ struct 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);
+ pthread_mutex_lock(&that_thread->ThreadMutex);
+ that_thread->load_avg = that_thread->avg_sleeping + that_thread->avg_running + that_thread->avg_blocked;
+ that_thread->load_avg = that_thread->avg_running / that_thread->load_avg * 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\" (%ld) \"%s\" %f %f %f %f.\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
+ pthread_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 %f, workers averag %f, threads %d, workers %d, sessions %d\n", CtdlThreadLoadAvg, 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)
+{
+ struct CtdlThreadNode *this_thread, *that_thread;
+ int workers = 0;
+ int ret=0;
+
+ begin_critical_section(S_THREAD_LIST);
+
+ /* Handle exiting of garbage collector thread */
+ if(num_threads == 1)
+ CtdlThreadList->state = CTDL_THREAD_EXITED;
+
+#ifdef WITH_THREADLOG
+ CtdlLogPrintf(CTDL_DEBUG, "Thread system running garbage collection.\n");
+#endif
+ /*
+ * Woke up to do garbage collection
+ */
+ this_thread = CtdlThreadList;
+ while(this_thread)
+ {
+ that_thread = this_thread;
+ this_thread = this_thread->next;
+
+ /* Do we need to clean up this thread? */
+ pthread_mutex_lock(&that_thread->ThreadMutex);
+ if (that_thread->state != CTDL_THREAD_EXITED)
+ {
+ if(that_thread->flags & CTDLTHREAD_WORKER)
+ workers++; /* Sanity check on number of worker threads */
+ pthread_mutex_unlock(&that_thread->ThreadMutex);
+ continue;
+ }
+
+ if (pthread_equal(that_thread->tid, pthread_self()) && that_thread->thread_func)
+ { /* Sanity check */
+ pthread_mutex_unlock(&that_thread->ThreadMutex);
+ end_critical_section(S_THREAD_LIST);
+ CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC, a thread is trying to clean up after itself.\n");
+ abort();
+ return;
+ }
+
+ if (num_threads <= 0)
+ { /* Sanity check */
+ pthread_mutex_unlock(&that_thread->ThreadMutex);
+ end_critical_section(S_THREAD_LIST);
+ CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC, num_threads <= 0 and trying to do Garbage Collection.\n");
+ abort();
+ return;
+ }
+
+ if(that_thread->flags & CTDLTHREAD_WORKER)
+ num_workers--; /* This is a wroker thread so reduce the count. */
+ num_threads--;
+ /* If we are unlinking the list head then the next becomes the list head */
+ if(that_thread->prev)
+ that_thread->prev->next = that_thread->next;
+ else
+ CtdlThreadList = that_thread->next;
+ if(that_thread->next)
+ that_thread->next->prev = that_thread->prev;
+
+ pthread_mutex_unlock(&that_thread->ThreadMutex);
+ pthread_cond_signal(&that_thread->ThreadCond);
+ pthread_cond_signal(&that_thread->SleepCond); // Make sure this thread is awake
+ pthread_mutex_lock(&that_thread->ThreadMutex); // Make sure it has done what its doing
+ pthread_mutex_unlock(&that_thread->ThreadMutex);
+ /*
+ * Join on the thread to do clean up and prevent memory leaks
+ * Also makes sure the thread has cleaned up after itself before we remove it from the list
+ * We can join on the garbage collector thread the join should just return EDEADLCK
+ */
+ ret = pthread_join (that_thread->tid, NULL);
+ if (ret == EDEADLK)
+ CtdlLogPrintf(CTDL_DEBUG, "Garbage collection on own thread.\n");
+ else if (ret == EINVAL)
+ CtdlLogPrintf(CTDL_DEBUG, "Garbage collection, that thread already joined on.\n");
+ else if (ret == ESRCH)
+ CtdlLogPrintf(CTDL_DEBUG, "Garbage collection, no thread to join on.\n");
+ else if (ret != 0)
+ CtdlLogPrintf(CTDL_DEBUG, "Garbage collection, pthread_join returned an unknown error.\n");
+ /*
+ * Now we own that thread entry
+ */
+ CtdlLogPrintf(CTDL_INFO, "Garbage Collection for thread \"%s\" (%ld).\n", that_thread->name, that_thread->tid);
+ pthread_mutex_destroy(&that_thread->ThreadMutex);
+ pthread_cond_destroy(&that_thread->ThreadCond);
+ pthread_mutex_destroy(&that_thread->SleepMutex);
+ pthread_cond_destroy(&that_thread->SleepCond);
+ pthread_attr_destroy(&that_thread->attr);
+ free(that_thread);
+ }
+
+ /* Sanity check number of worker threads */
+ if (workers != num_workers)
+ {
+ end_critical_section(S_THREAD_LIST);
+ CtdlLogPrintf(CTDL_EMERG,
+ "Thread system PANIC, discrepancy in number of worker threads. Counted %d, should be %d.\n",
+ workers, num_workers
+ );
+ abort();
+ }
+ end_critical_section(S_THREAD_LIST);
+}
+
+
+
+
+/*
+ * Runtime function for a Citadel Thread.
+ * This initialises the threads environment and then calls the user supplied thread function
+ * Note that this is the REAL thread function and wraps the users thread function.
+ */
+static void *ctdl_internal_thread_func (void *arg)
+{
+ struct CtdlThreadNode *this_thread;
+ void *ret = NULL;
+
+ /* lock and unlock the thread list.
+ * This causes this thread to wait until all its creation stuff has finished before it
+ * can continue its execution.
+ */
+ begin_critical_section(S_THREAD_LIST);
+ this_thread = (struct CtdlThreadNode *) arg;
+ gettimeofday(&this_thread->start_time, NULL); /* Time this thread started */
+ pthread_mutex_lock(&this_thread->ThreadMutex);
+
+ // Register the cleanup function to take care of when we exit.
+ pthread_cleanup_push(ctdl_internal_thread_cleanup, NULL);
+ // Get our thread data structure
+ CtdlThreadAllocTSD();
+ CT = this_thread;
+ this_thread->pid = getpid();
+ memcpy(&this_thread->last_state_change, &this_thread->start_time, sizeof (struct timeval)); /* Changed state so mark it. */
+ /* Only change to running state if we weren't asked to stop during the create cycle
+ * Other wise there is a window to allow this threads creation to continue to full grown and
+ * therby prevent a shutdown of the server.
+ */
+ pthread_mutex_unlock(&this_thread->ThreadMutex);
+
+ if (!CtdlThreadCheckStop())
+ {
+ pthread_mutex_lock(&this_thread->ThreadMutex);
+ this_thread->state = CTDL_THREAD_RUNNING;
+ pthread_mutex_unlock(&this_thread->ThreadMutex);
+ }
+ end_critical_section(S_THREAD_LIST);
+
+ // Register for tracing
+ #ifdef HAVE_BACKTRACE
+ eCrash_RegisterThread(this_thread->name, 0);
+ #endif
+
+ // Tell the world we are here
+ CtdlLogPrintf(CTDL_NOTICE, "Created a new thread \"%s\" (%ld). \n", this_thread->name, this_thread->tid);
+
+
+
+ /*
+ * run the thread to do the work but only if we haven't been asked to stop
+ */
+ if (!CtdlThreadCheckStop())
+ ret = (this_thread->thread_func)(this_thread->user_args);
+
+ /*
+ * Our thread is exiting either because it wanted to end or because the server is stopping
+ * We need to clean up
+ */
+ pthread_cleanup_pop(1); // Execute our cleanup routine and remove it
+
+ return(ret);
+}
+
+
+
+/*
+ * Internal function to create a thread.
+ * Must be called from within a S_THREAD_LIST critical section
+ */
+struct CtdlThreadNode *ctdl_internal_create_thread(char *name, long flags, void *(*thread_func) (void *arg), void *args)
+{
+ int ret = 0;
+ struct CtdlThreadNode *this_thread;
+
+ if (num_threads >= 32767)
+ {
+ CtdlLogPrintf(CTDL_EMERG, "Thread system. Thread list full.\n");
+ return NULL;
+ }
+
+ this_thread = malloc(sizeof(struct CtdlThreadNode));
+ if (this_thread == NULL) {
+ CtdlLogPrintf(CTDL_EMERG, "Thread system, can't allocate CtdlThreadNode, exiting\n");
+ return NULL;
+ }
+ // Ensuring this is zero'd means we make sure the thread doesn't start doing its thing until we are ready.
+ memset (this_thread, 0, sizeof(struct CtdlThreadNode));
+
+ /* Create the mutex's early so we can use them */
+ pthread_mutex_init (&(this_thread->ThreadMutex), NULL);
+ pthread_cond_init (&(this_thread->ThreadCond), NULL);
+ pthread_mutex_init (&(this_thread->SleepMutex), NULL);
+ pthread_cond_init (&(this_thread->SleepCond), NULL);
+
+ pthread_mutex_lock(&this_thread->ThreadMutex);
+
+ this_thread->state = CTDL_THREAD_CREATE;
+
+ if ((ret = pthread_attr_init(&this_thread->attr))) {
+ pthread_mutex_unlock(&this_thread->ThreadMutex);
+ pthread_mutex_destroy(&(this_thread->ThreadMutex));
+ pthread_cond_destroy(&(this_thread->ThreadCond));
+ pthread_mutex_destroy(&(this_thread->SleepMutex));
+ pthread_cond_destroy(&(this_thread->SleepCond));
+ CtdlLogPrintf(CTDL_EMERG, "Thread system, pthread_attr_init: %s\n", strerror(ret));
+ free(this_thread);
+ return NULL;
+ }
+
+ /* Our per-thread stacks need to be bigger than the default size,
+ * otherwise the MIME parser crashes on FreeBSD, and the IMAP service
+ * crashes on 64-bit Linux.
+ */
+ if (flags & CTDLTHREAD_BIGSTACK)
+ {
+#ifdef WITH_THREADLOG
+ CtdlLogPrintf(CTDL_INFO, "Thread system. Creating BIG STACK thread.\n");
+#endif
+ if ((ret = pthread_attr_setstacksize(&this_thread->attr, THREADSTACKSIZE))) {
+ pthread_mutex_unlock(&this_thread->ThreadMutex);
+ pthread_mutex_destroy(&(this_thread->ThreadMutex));
+ pthread_cond_destroy(&(this_thread->ThreadCond));
+ pthread_mutex_destroy(&(this_thread->SleepMutex));
+ pthread_cond_destroy(&(this_thread->SleepCond));
+ pthread_attr_destroy(&this_thread->attr);
+ CtdlLogPrintf(CTDL_EMERG, "Thread system, pthread_attr_setstacksize: %s\n",
+ strerror(ret));
+ free(this_thread);
+ 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;
+ /* Set this new thread with an avg_blocked of 2. We do this so that its creation affects the
+ * load average for the system. If we don't do this then we create a mass of threads at the same time
+ * because the creation didn't affect the load average.
+ */
+ this_thread->avg_blocked = 2;
+
+ /*
+ * We pass this_thread into the thread as its args so that it can find out information
+ * about itself and it has a bit of storage space for itself, not to mention that the REAL
+ * thread function needs to finish off the setup of the structure
+ */
+ if ((ret = pthread_create(&this_thread->tid, &this_thread->attr, ctdl_internal_thread_func, this_thread) != 0))
+ {
+
+ CtdlLogPrintf(CTDL_ALERT, "Thread system, Can't create thread: %s\n",
+ strerror(ret));
+ pthread_mutex_unlock(&this_thread->ThreadMutex);
+ pthread_mutex_destroy(&(this_thread->ThreadMutex));
+ pthread_cond_destroy(&(this_thread->ThreadCond));
+ pthread_mutex_destroy(&(this_thread->SleepMutex));
+ pthread_cond_destroy(&(this_thread->SleepCond));
+ pthread_attr_destroy(&this_thread->attr);
+ free(this_thread);
+ return NULL;
+ }
+
+ num_threads++; // Increase the count of threads in the system.
+ if(this_thread->flags & CTDLTHREAD_WORKER)
+ num_workers++;
+
+ this_thread->next = CtdlThreadList;
+ CtdlThreadList = this_thread;
+ if (this_thread->next)
+ this_thread->next->prev = this_thread;
+
+ pthread_mutex_unlock(&this_thread->ThreadMutex);
+
+ ctdl_thread_internal_calc_loadavg();
+ return this_thread;
+}
+
+/*
+ * Wrapper function to create a thread
+ * ensures the critical section and other protections are in place.
+ * char *name = name to give to thread, if NULL, use generic name
+ * int flags = flags to determine type of thread and standard facilities
+ */
+struct CtdlThreadNode *CtdlThreadCreate(char *name, long flags, void *(*thread_func) (void *arg), void *args)
+{
+ struct CtdlThreadNode *ret = NULL;
+
+ begin_critical_section(S_THREAD_LIST);
+ ret = ctdl_internal_create_thread(name, flags, thread_func, args);
+ end_critical_section(S_THREAD_LIST);
+ return ret;
+}
+
+
+
+/*
+ * Internal function to schedule a thread.
+ * Must be called from within a S_THREAD_LIST critical section
+ */
+struct CtdlThreadNode *CtdlThreadSchedule(char *name, long flags, void *(*thread_func) (void *arg), void *args, time_t when)
+{
+ int ret = 0;
+ struct CtdlThreadNode *this_thread;
+
+ if (num_threads >= 32767)
+ {
+ CtdlLogPrintf(CTDL_EMERG, "Thread system. Thread list full.\n");
+ return NULL;
+ }
+
+ this_thread = malloc(sizeof(struct CtdlThreadNode));
+ if (this_thread == NULL) {
+ CtdlLogPrintf(CTDL_EMERG, "Thread system, can't allocate CtdlThreadNode, exiting\n");
+ return NULL;
+ }
+ // Ensuring this is zero'd means we make sure the thread doesn't start doing its thing until we are ready.
+ memset (this_thread, 0, sizeof(struct CtdlThreadNode));
+
+ /* Create the mutex's early so we can use them */
+ pthread_mutex_init (&(this_thread->ThreadMutex), NULL);
+ pthread_cond_init (&(this_thread->ThreadCond), NULL);
+ pthread_mutex_init (&(this_thread->SleepMutex), NULL);
+ pthread_cond_init (&(this_thread->SleepCond), NULL);
+
+ this_thread->state = CTDL_THREAD_CREATE;
+
+ if ((ret = pthread_attr_init(&this_thread->attr))) {
+ pthread_mutex_destroy(&(this_thread->ThreadMutex));
+ pthread_cond_destroy(&(this_thread->ThreadCond));
+ pthread_mutex_destroy(&(this_thread->SleepMutex));
+ pthread_cond_destroy(&(this_thread->SleepCond));
+ CtdlLogPrintf(CTDL_EMERG, "Thread system, pthread_attr_init: %s\n", strerror(ret));
+ free(this_thread);
+ return NULL;
+ }
+
+ /* Our per-thread stacks need to be bigger than the default size,
+ * otherwise the MIME parser crashes on FreeBSD, and the IMAP service
+ * crashes on 64-bit Linux.
+ */
+ if (flags & CTDLTHREAD_BIGSTACK)
+ {
+ CtdlLogPrintf(CTDL_INFO, "Thread system. Creating BIG STACK thread.\n");
+ if ((ret = pthread_attr_setstacksize(&this_thread->attr, THREADSTACKSIZE))) {
+ pthread_mutex_destroy(&(this_thread->ThreadMutex));
+ pthread_cond_destroy(&(this_thread->ThreadCond));
+ pthread_mutex_destroy(&(this_thread->SleepMutex));
+ pthread_cond_destroy(&(this_thread->SleepCond));
+ pthread_attr_destroy(&this_thread->attr);
+ CtdlLogPrintf(CTDL_EMERG, "Thread system, pthread_attr_setstacksize: %s\n",
+ strerror(ret));
+ free(this_thread);
+ 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;
+ /* Set this new thread with an avg_blocked of 2. We do this so that its creation affects the
+ * load average for the system. If we don't do this then we create a mass of threads at the same time
+ * because the creation didn't affect the load average.
+ */
+ this_thread->avg_blocked = 2;
+
+ /*
+ * 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;
+}
+
+
+
+struct CtdlThreadNode *ctdl_thread_internal_start_scheduled (struct CtdlThreadNode *this_thread)
+{
+ int ret = 0;
+
+ /*
+ * We pass this_thread into the thread as its args so that it can find out information
+ * about itself and it has a bit of storage space for itself, not to mention that the REAL
+ * thread function needs to finish off the setup of the structure
+ */
+ if ((ret = pthread_create(&this_thread->tid, &this_thread->attr, ctdl_internal_thread_func, this_thread) != 0))
+ {
+
+ CtdlLogPrintf(CTDL_ALERT, "Thread system, Can't create thread: %s\n",
+ strerror(ret));
+ return NULL;
+ }
+
+
+ num_threads++; // Increase the count of threads in the system.
+ if(this_thread->flags & CTDLTHREAD_WORKER)
+ num_workers++;
+
+ this_thread->next = CtdlThreadList;
+ CtdlThreadList = this_thread;
+ if (this_thread->next)
+ this_thread->next->prev = this_thread;
+
+ return this_thread;
+}
+
+
+
+void ctdl_thread_internal_check_scheduled(void)
+{
+ struct CtdlThreadNode *this_thread, *that_thread;
+ time_t now;
+
+ if (try_critical_section(S_SCHEDULE_LIST))
+ return; /* If this list is locked we wait till the next chance */
+
+ now = time(NULL);
+
+#ifdef WITH_THREADLOG
+ CtdlLogPrintf(CTDL_DEBUG, "Checking for scheduled threads to start.\n");
+#endif
+
+ this_thread = CtdlThreadSchedList;
+ while(this_thread)
+ {
+ that_thread = this_thread;
+ this_thread = this_thread->next;
+
+ if (now > that_thread->when)
+ {
+ /* Unlink from schedule list */
+ if (that_thread->prev)
+ that_thread->prev->next = that_thread->next;
+ else
+ CtdlThreadSchedList = that_thread->next;
+ if (that_thread->next)
+ that_thread->next->prev = that_thread->prev;
+
+ that_thread->next = that_thread->prev = NULL;
+#ifdef WITH_THREADLOG
+ CtdlLogPrintf(CTDL_DEBUG, "About to start scheduled thread \"%s\".\n", that_thread->name);
+#endif
+ begin_critical_section(S_THREAD_LIST);
+ if (CT->state > CTDL_THREAD_STOP_REQ)
+ { /* Only start it if the system is not stopping */
+ pthread_mutex_lock(&that_thread->ThreadMutex);
+ if (ctdl_thread_internal_start_scheduled (that_thread) == NULL)
+ {
+#ifdef WITH_THREADLOG
+ CtdlLogPrintf(CTDL_DEBUG, "Failed to start scheduled thread \"%s\".\n", that_thread->name);
+#endif
+ pthread_mutex_unlock(&that_thread->ThreadMutex);
+ pthread_mutex_destroy(&(that_thread->ThreadMutex));
+ pthread_cond_destroy(&(that_thread->ThreadCond));
+ pthread_mutex_destroy(&(that_thread->SleepMutex));
+ pthread_cond_destroy(&(that_thread->SleepCond));
+ pthread_attr_destroy(&that_thread->attr);
+ free(that_thread);
+ }
+ else
+ {
+ CtdlLogPrintf(CTDL_INFO, "Thread system, Started a scheduled thread \"%s\" (%ld).\n",
+ that_thread->name, that_thread->tid);
+ pthread_mutex_unlock(&that_thread->ThreadMutex);
+ ctdl_thread_internal_calc_loadavg();
+ }
+ }
+ end_critical_section(S_THREAD_LIST);
+ }
+ else
+ {
+#ifdef WITH_THREADLOG
+ CtdlLogPrintf(CTDL_DEBUG, "Thread \"%s\" will start in %ld seconds.\n", that_thread->name, that_thread->when - time(NULL));
+#endif
+ }
+ }
+ end_critical_section(S_SCHEDULE_LIST);
+}
+
+
+/*
+ * A warapper function for select so we can show a thread as blocked
+ */
+int CtdlThreadSelect(int n, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout)
+{
+ int ret;
+
+ ctdl_thread_internal_change_state(CT, CTDL_THREAD_BLOCKED);
+ ret = select(n, readfds, writefds, exceptfds, timeout);
+ ctdl_thread_internal_change_state(CT, CTDL_THREAD_RUNNING);
+ return ret;
+}
+
+
+
+
+
+
+void go_threading(void)
+{
+ int i;
+ struct CtdlThreadNode *last_worker;
+
+ /*
+ * Initialise the thread system
+ */
+ ctdl_thread_internal_init();
+ /*
+ * Now create a bunch of worker threads.
+ */
+// CtdlLogPrintf(CTDL_DEBUG, "Starting %d worker threads\n", config.c_min_workers);
+// begin_critical_section(S_THREAD_LIST);
+// i=0; /* Always start at least 1 worker thread */
+// do
+// {
+// ctdl_internal_create_thread("Worker Thread", CTDLTHREAD_BIGSTACK + CTDLTHREAD_WORKER, worker_thread, NULL);
+// } while (++i < config.c_min_workers);
+// end_critical_section(S_THREAD_LIST);
+
+ /* Second call to module init functions now that threading is up */
+ initialise_modules(1);
+
+ /*
+ * 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", pthread_self());
+
+ /*
+ * We do a lot of locking and unlocking of the thread list in here.
+ * We do this so that we can repeatedly release time for other threads
+ * that may be waiting on the thread list.
+ * We are a low priority thread so we can afford to do this
+ */
+
+ while (CtdlThreadGetCount())
+ {
+ if (CT->signal)
+ exit_signal = CT->signal;
+ if (exit_signal)
+ CtdlThreadStopAll();
+ 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 */
+ }
+
+ /* Reduce the size of the worker thread pool if necessary. */
+ if ((CtdlThreadGetWorkers() > config.c_min_workers) && (CtdlThreadWorkerAvg < 20) && (CT->state > CTDL_THREAD_STOP_REQ))
+ {
+ /* Ask a worker thread to stop as we no longer need it */
+ begin_critical_section(S_THREAD_LIST);
+ last_worker = CtdlThreadList;
+ while (last_worker)
+ {
+ pthread_mutex_lock(&last_worker->ThreadMutex);
+ if (last_worker->flags & CTDLTHREAD_WORKER && last_worker->state > CTDL_THREAD_STOPPING)
+ {
+ pthread_mutex_unlock(&last_worker->ThreadMutex);
+ break;
+ }
+ pthread_mutex_unlock(&last_worker->ThreadMutex);
+ last_worker = last_worker->next;
+ }
+ end_critical_section(S_THREAD_LIST);
+ if (last_worker)
+ {
+#ifdef WITH_THREADLOG
+ CtdlLogPrintf(CTDL_DEBUG, "Thread system, stopping excess worker thread \"%s\" (%ld).\n",
+ last_worker->name,
+ last_worker->tid
+ );
+#endif
+ CtdlThreadStop(last_worker);
+ }
+ }
+
+ /*
+ * If all our workers are working hard, start some more to help out
+ * with things
+ */
+ /* FIXME: come up with a better way to dynamically alter the number of threads
+ * based on the system load
+ */
+// if ((CtdlThreadGetWorkers() < config.c_max_workers) && (CtdlThreadGetWorkers() < num_sessions))
+ // && (CtdlThreadLoadAvg < 90) )
+ if ((((CtdlThreadGetWorkers() < config.c_max_workers) && (CtdlThreadGetWorkerAvg() > 60) && (CtdlThreadGetLoadAvg() < 90) ) || CtdlThreadGetWorkers() < config.c_min_workers) && (CT->state > CTDL_THREAD_STOP_REQ))
+ {
+ for (i=0; i<5 ; i++)
+// for (i=0; i< (num_sessions - CtdlThreadGetWorkers()) ; i++)
+// for (i=0; i< (10 - (55 - CtdlThreadWorkerAvg) / CtdlThreadWorkerAvg / CtdlThreadGetWorkers()) ; i++)
+ {
+// begin_critical_section(S_THREAD_LIST);
+ CtdlThreadCreate("Worker Thread",
+ CTDLTHREAD_BIGSTACK + CTDLTHREAD_WORKER,
+ worker_thread,
+ NULL
+ );
+// end_critical_section(S_THREAD_LIST);
+ }
+ }
+
+ CtdlThreadGC();
+
+ if (CtdlThreadGetCount() <= 1) // Shutting down clean up the garbage collector
+ {
+ CtdlThreadGC();
+ }
+
+ if (CtdlThreadGetCount())
+ CtdlThreadSleep(1);
+ }
+ /*
+ * If the above loop exits we must be shutting down since we obviously have no threads
+ */
+ ctdl_thread_internal_cleanup();
+}
--- /dev/null
+/* $Id:$ */
+
+#ifndef THREADS_H
+#define THREADS_H
+
+#include "sysdep.h"
+
+#ifdef HAVE_PTHREAD_H
+#include <pthread.h>
+#endif
+
+#include <sys/time.h>
+#include <string.h>
+
+#ifdef HAVE_DB_H
+#include <db.h>
+#elif defined(HAVE_DB4_DB_H)
+#include <db4/db.h>
+#else
+#error Neither <db.h> nor <db4/db.h> was found by configure. Install db4-devel.
+#endif
+
+#include "server.h"
+
+/*
+ * Thread stuff
+ */
+#define CTDLTHREAD_BIGSTACK 0x0001
+#define CTDLTHREAD_WORKER 0x0002
+
+enum CtdlThreadState {
+ CTDL_THREAD_INVALID,
+ CTDL_THREAD_VALID,
+ CTDL_THREAD_CREATE,
+ CTDL_THREAD_CANCELLED,
+ CTDL_THREAD_EXITED,
+ CTDL_THREAD_STOPPING,
+ CTDL_THREAD_STOP_REQ, /* Do NOT put any running states before this state */
+ CTDL_THREAD_SLEEPING,
+ CTDL_THREAD_BLOCKED,
+ CTDL_THREAD_RUNNING,
+ CTDL_THREAD_LAST_STATE
+};
+
+extern struct CtdlThreadNode {
+ pthread_t tid; /* id as returned by pthread_create() */
+ pid_t pid; /* pid, as best the OS will let us determine */
+ time_t when; /* When to start a scheduled thread */
+ struct CitConext *Context; /* The session context that this thread mught be working on or NULL if none */
+ long number; /* A unigue number for this thread (not implimented yet) */
+ int wakefd_recv; /* An fd that this thread can sleep on (not implimented yet) */
+ int wakefd_send; /* An fd that this thread can send out on (Not implimented yet) */
+ int signal; /* A field to store a signal we caught. */
+ const char *name; /* A name for this thread */
+ void *(*thread_func) (void *arg); /* The actual function that does this threads work */
+ void *user_args; /* Arguments passed to this threads work function */
+ long flags; /* Flags that describe this thread */
+ enum CtdlThreadState state; /* Flag to show state of this thread */
+ pthread_mutex_t ThreadMutex; /* A mutex to sync this thread to others if this thread allows (also used for sleeping) */
+ pthread_cond_t ThreadCond; /* A condition variable to sync this thread with others */
+ pthread_mutex_t SleepMutex; /* A mutex for sleeping */
+ pthread_cond_t SleepCond; /* A condition variable for sleeping */
+ pthread_attr_t attr; /* Attributes of this thread */
+ struct timeval start_time; /* Time this thread was started */
+ struct timeval last_state_change; /* Time when this thread last changed state */
+ double avg_sleeping; /* Average sleeping time */
+ double avg_running; /* Average running time */
+ double avg_blocked; /* Average blocked time */
+ double load_avg; /* Load average for this thread */
+ struct CtdlThreadNode *prev; /* Previous thread in the thread table */
+ struct CtdlThreadNode *next; /* Next thread in the thread table */
+} *CtdlThreadList;
+
+typedef struct {
+ DB_TXN *tid; /* Transaction handle */
+ DBC *cursors[MAXCDB]; /* Cursors, for traversals... */
+ struct CtdlThreadNode *self; /* Pointer to this threads control structure */
+}ThreadTSD ;
+
+extern double CtdlThreadLoadAvg;
+extern double CtdlThreadWorkerAvg;
+extern pthread_key_t ThreadKey;
+
+void ctdl_thread_internal_init_tsd(void);
+void ctdl_internal_thread_gc (void);
+void ctdl_thread_internal_init(void);
+void ctdl_thread_internal_cleanup(void);
+void ctdl_thread_internal_calc_loadavg(void);
+void ctdl_thread_internal_free_tsd(void);
+struct CtdlThreadNode *ctdl_internal_create_thread(char *name, long flags, void *(*thread_func) (void *arg), void *args);
+void ctdl_thread_internal_check_scheduled(void);
+
+void InitialiseSemaphores(void);
+int try_critical_section (int which_one);
+void begin_critical_section (int which_one);
+void end_critical_section (int which_one);
+void go_threading(void);
+
+#endif // THREADS_H
#include "citserver.h"
#include "citadel_dirs.h"
#include "genstamp.h"
+#include "threads.h"
/* These pipes are used to talk to the chkpwd daemon, which is forked during startup */
int chkpwd_write_pipe[2];
projects / citadel.git / commitdiff