X-Git-Url: https://code.citadel.org/?a=blobdiff_plain;f=citadel%2Fsysdep.c;h=c09fd2a06292edca76dae0ebbad3997440b76818;hb=cbc2a603ed4c3995836e45fd400e44e12868f5f8;hp=dcd3b249f78a8606353c32b3c8e3f586495b9ee9;hpb=2ff61cb4add87dd2592b59440dafae856da51b6b;p=citadel.git diff --git a/citadel/sysdep.c b/citadel/sysdep.c index dcd3b249f..c09fd2a06 100644 --- a/citadel/sysdep.c +++ b/citadel/sysdep.c @@ -49,9 +49,6 @@ #include #include #include -#ifdef HAVE_PTHREAD_H -#include -#endif #include #include "citadel.h" #include "server.h" @@ -73,6 +70,10 @@ #endif #include "ctdl_module.h" +#include "threads.h" +#include "user_ops.h" +#include "control.h" + #ifdef DEBUG_MEMORY_LEAKS struct igheap { @@ -86,15 +87,12 @@ struct igheap *igheap = NULL; #endif -pthread_mutex_t Critters[MAX_SEMAPHORES]; /* Things needing locking */ -pthread_key_t MyConKey; /* TSD key for MyContext() */ +citthread_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; @@ -102,30 +100,16 @@ int enable_syslog = 0; /* - * Create an interface to lprintf that follows the coding convention. - * This is here until such time as we have replaced all calls to lprintf with CtdlLogPrintf - */ - -void CtdlLogPrintf(enum LogLevel loglevel, const char *format, ...) -{ - va_list arg_ptr; - va_start(arg_ptr, format); - vlprintf(loglevel, format, arg_ptr); - va_end(arg_ptr); -} - - -/* - * lprintf() ... Write logging information + * CtdlLogPrintf() ... Write logging information */ -void lprintf(enum LogLevel loglevel, const char *format, ...) { +void CtdlLogPrintf(enum LogLevel loglevel, const char *format, ...) { va_list arg_ptr; va_start(arg_ptr, format); - vlprintf(loglevel, format, arg_ptr); + vCtdlLogPrintf(loglevel, format, arg_ptr); va_end(arg_ptr); } -void vlprintf(enum LogLevel loglevel, const char *format, va_list arg_ptr) +void vCtdlLogPrintf(enum LogLevel loglevel, const char *format, va_list arg_ptr) { char buf[SIZ], buf2[SIZ]; @@ -160,7 +144,7 @@ void vlprintf(enum LogLevel loglevel, const char *format, va_list arg_ptr) tim.tm_mday, tim.tm_hour, tim.tm_min, tim.tm_sec, (long)tv.tv_usec); } - vsprintf(buf2, format, arg_ptr); + vsnprintf(buf2, SIZ, format, arg_ptr); fprintf(stderr, "%s%s", buf, buf2); fflush(stderr); @@ -179,20 +163,14 @@ volatile int restart_server = 0; volatile int running_as_daemon = 0; static RETSIGTYPE signal_cleanup(int signum) { - CtdlLogPrintf(CTDL_DEBUG, "Caught signal %d; shutting down.\n", signum); - exit_signal = signum; -} - - - - -void InitialiseSemaphores(void) -{ - int i; - - /* Set up a bunch of semaphores to be used for critical sections */ - for (i=0; isignal = signum; + else +#endif + { + CtdlLogPrintf(CTDL_DEBUG, "Caught signal %d; shutting down.\n", signum); + exit_signal = signum; } } @@ -226,7 +204,7 @@ void init_sysdep(void) { * CitContext structure (in the ContextList linked list) of the * session to which the calling thread is currently bound. */ - if (pthread_key_create(&MyConKey, NULL) != 0) { + if (citthread_key_create(&MyConKey, NULL) != 0) { CtdlLogPrintf(CTDL_CRIT, "Can't create TSD key: %s\n", strerror(errno)); } @@ -263,59 +241,6 @@ void init_sysdep(void) { -/* - * 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 @@ -477,7 +402,7 @@ struct CitContext *MyContext(void) { register struct CitContext *c; - return ((c = (struct CitContext *) pthread_getspecific(MyConKey), + return ((c = (struct CitContext *) citthread_getspecific(MyConKey), c == NULL) ? &masterCC : c ); } @@ -498,14 +423,17 @@ struct CitContext *CreateNewContext(void) { CtdlLogPrintf(CTDL_ALERT, "citserver: can't allocate memory!!\n"); return NULL; } - memset(me, 0, sizeof(struct CitContext)); + memset(me, 0, sizeof(struct CitContext)); + + /* Give the contaxt a name. Hopefully makes it easier to track */ + strcpy (me->user.fullname, "SYS_notauth"); + /* The new context will be created already in the CON_EXECUTING state * in order to prevent another thread from grabbing it while it's * being set up. */ me->state = CON_EXECUTING; - /* * Generate a unique session number and insert this context into * the list. @@ -520,10 +448,59 @@ struct CitContext *CreateNewContext(void) { } ++num_sessions; end_critical_section(S_SESSION_TABLE); - return(me); + return (me); +} + + +struct CitContext *CtdlGetContextArray(int *count) +{ + int nContexts, i; + struct CitContext *nptr, *cptr; + + nContexts = num_sessions; + nptr = malloc(sizeof(struct CitContext) * nContexts); + if (!nptr) + return NULL; + begin_critical_section(S_SESSION_TABLE); + for (cptr = ContextList, i=0; cptr != NULL && i < nContexts; cptr = cptr->next, i++) + memcpy(&nptr[i], cptr, sizeof (struct CitContext)); + end_critical_section (S_SESSION_TABLE); + + *count = i; + return nptr; } + +/** + * This function fills in a context and its user field correctly + * Then creates/loads that user + */ +void CtdlFillSystemContext(struct CitContext *context, char *name) +{ + char sysname[USERNAME_SIZE]; + + memset(context, 0, sizeof(struct CitContext)); + context->internal_pgm = 1; + context->cs_pid = 0; + strcpy (sysname, "SYS_"); + strcat (sysname, name); + /* internal_create_user has the side effect of loading the user regardless of wether they + * already existed or needed to be created + */ + internal_create_user (sysname, &(context->user), -1) ; + + /* Check to see if the system user needs upgrading */ + if (context->user.usernum == 0) + { /* old system user with number 0, upgrade it */ + context->user.usernum = get_new_user_number(); + CtdlLogPrintf(CTDL_DEBUG, "Upgrading system user \"%s\" from user number 0 to user number %d\n", context->user.fullname, context->user.usernum); + /* add user to the database */ + putuser(&(context->user)); + cdb_store(CDB_USERSBYNUMBER, &(context->user.usernum), sizeof(long), context->user.fullname, strlen(context->user.fullname)+1); + } +} + /* * The following functions implement output buffering. If the kernel supplies * native TCP buffering (Linux & *BSD), use that; otherwise, emulate it with @@ -604,14 +581,16 @@ void unbuffer_output(void) { /* * client_write() ... Send binary data to the client. */ -void client_write(char *buf, int nbytes) +int client_write(char *buf, int nbytes) { int bytes_written = 0; int retval; #ifndef HAVE_TCP_BUFFERING int old_buffer_len = 0; #endif + fd_set wset; t_context *Ctx; + int fdflags; Ctx = CC; if (Ctx->redirect_buffer != NULL) { @@ -623,7 +602,7 @@ void client_write(char *buf, int nbytes) memcpy(&Ctx->redirect_buffer[Ctx->redirect_len], buf, nbytes); Ctx->redirect_len += nbytes; Ctx->redirect_buffer[Ctx->redirect_len] = 0; - return; + return 0; } #ifndef HAVE_TCP_BUFFERING @@ -633,7 +612,7 @@ void client_write(char *buf, int nbytes) Ctx->buffer_len += nbytes; Ctx->output_buffer = realloc(Ctx->output_buffer, Ctx->buffer_len); memcpy(&Ctx->output_buffer[old_buffer_len], buf, nbytes); - return; + return 0; } #endif @@ -642,11 +621,27 @@ void client_write(char *buf, int nbytes) #ifdef HAVE_OPENSSL if (Ctx->redirect_ssl) { client_write_ssl(buf, nbytes); - return; + return 0; } #endif + fdflags = fcntl(Ctx->client_socket, F_GETFL); + while (bytes_written < nbytes) { + if ((fdflags & O_NONBLOCK) == O_NONBLOCK) { + FD_ZERO(&wset); + FD_SET(Ctx->client_socket, &wset); + if (select(1, NULL, &wset, NULL, NULL) == -1) { + CtdlLogPrintf(CTDL_ERR, + "client_write(%d bytes) select failed: %s (%d)\n", + nbytes - bytes_written, + strerror(errno), errno); + cit_backtrace(); + Ctx->kill_me = 1; + return -1; + } + } + retval = write(Ctx->client_socket, &buf[bytes_written], nbytes - bytes_written); if (retval < 1) { @@ -657,10 +652,11 @@ void client_write(char *buf, int nbytes) cit_backtrace(); // CtdlLogPrintf(CTDL_DEBUG, "Tried to send: %s", &buf[bytes_written]); Ctx->kill_me = 1; - return; + return -1; } bytes_written = bytes_written + retval; } + return 0; } @@ -712,6 +708,15 @@ int client_read_to(char *buf, int bytes, int timeout) retval = select( (fd)+1, &rfds, NULL, NULL, &tv); + if (retval < 0) + { + if (errno == EINTR) + { + CtdlLogPrintf(CTDL_DEBUG, "Interrupted select().\n"); + CC->kill_me = 1; + return (-1); + } + } if (FD_ISSET(fd, &rfds) == 0) { return(0); @@ -789,24 +794,30 @@ void context_cleanup(void) * There are no threads so no critical_section stuff is needed. */ ptr = ContextList; + + /* We need to update the ContextList because some modules may want to itterate it + * Question is should we NULL it before iterating here or should we just keep updating it + * as we remove items? + * + * Answer is to NULL it first to prevent modules from doing any actions on the list at all + */ + ContextList=NULL; while (ptr != NULL){ /* Remove the session from the active list */ rem = ptr->next; --num_sessions; - lprintf(CTDL_DEBUG, "Purging session %d\n", ptr->cs_pid); + CtdlLogPrintf(CTDL_DEBUG, "Purging session %d\n", ptr->cs_pid); RemoveContext(ptr); free (ptr); ptr = rem; } - } -/* - * The system-dependent part of master_cleanup() - close the master socket. - */ -void sysdep_master_cleanup(void) { + +void close_masters (void) +{ struct ServiceFunctionHook *serviceptr; /* @@ -816,20 +827,32 @@ void sysdep_master_cleanup(void) { serviceptr = serviceptr->next ) { if (serviceptr->tcp_port > 0) + { CtdlLogPrintf(CTDL_INFO, "Closing listener on port %d\n", serviceptr->tcp_port); - + serviceptr->tcp_port = 0; + } + if (serviceptr->sockpath != NULL) CtdlLogPrintf(CTDL_INFO, "Closing listener on '%s'\n", serviceptr->sockpath); close(serviceptr->msock); - /* If it's a Unix domain socket, remove the file. */ if (serviceptr->sockpath != NULL) { unlink(serviceptr->sockpath); + serviceptr->sockpath = NULL; } } +} + + +/* + * The system-dependent part of master_cleanup() - close the master socket. + */ +void sysdep_master_cleanup(void) { + + close_masters(); context_cleanup(); @@ -846,6 +869,8 @@ void sysdep_master_cleanup(void) { CtdlDestroyFixedOutputHooks(); CtdlDestroySessionHooks(); CtdlDestroyServiceHook(); + CtdlDestroyRoomHooks(); + CtdlDestroyDirectoryServiceFuncs(); #ifdef HAVE_BACKTRACE eCrash_Uninit(); #endif @@ -927,11 +952,7 @@ void start_daemon(int unused) { else { fp = fopen(file_pid_file, "w"); if (fp != NULL) { - /* - * NB.. The pid file contains the pid of the actual server. - * This is not the pid of the watcher process - */ - fprintf(fp, ""F_PID_T"\n", current_child); + fprintf(fp, ""F_PID_T"\n", getpid()); fclose(fp); } waitpid(current_child, &status, 0); @@ -992,835 +1013,27 @@ int convert_login(char NameToConvert[]) { } } - - -/* - * 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. - * If your thread function returns it will be started again without creating a new thread. - * If your thread function wants to exit it should call CtdlThreadExit(ret_code); - * 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; - -/* - * 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; -/* - * Pinched the following bits regarding signals from Kannel.org - */ - -/* - * Change this thread's signal mask to block user-visible signals - * (HUP, TERM, QUIT, INT), and store the old signal mask in - * *old_set_storage. - * Return 0 for success, or -1 if an error occurred. - */ - - /* - * This does not work in Darwin alias MacOS X alias Mach kernel, - * however. So we define a dummy function doing nothing. - */ -#if defined(DARWIN_OLD) - static int pthread_sigmask(); -#endif - -static int ctdl_thread_internal_block_signals(sigset_t *old_set_storage) -{ - int ret; - sigset_t block_signals; - - ret = sigemptyset(&block_signals); - if (ret != 0) { - CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC. Couldn't initialize signal set\n"); - return -1; - } - ret = sigaddset(&block_signals, SIGHUP); - ret |= sigaddset(&block_signals, SIGTERM); - ret |= sigaddset(&block_signals, SIGQUIT); - ret |= sigaddset(&block_signals, SIGINT); - if (ret != 0) { - CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC. Couldn't add signal to signal set.\n"); - return -1; - } - ret = pthread_sigmask(SIG_BLOCK, &block_signals, old_set_storage); - if (ret != 0) { - CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC. Couldn't disable signals for thread creation\n"); - return -1; - } - return 0; -} - -static void ctdl_thread_internal_restore_signals(sigset_t *old_set) -{ - int ret; - - ret = pthread_sigmask(SIG_SETMASK, old_set, NULL); - if (ret != 0) { - CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC. Couldn't restore signal set.\n"); - } -} - - -void ctdl_thread_internal_cleanup(void) -{ - int i; - - for (i=0; istate = 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 = strdup("Garbage Collection Thread"); - - pthread_mutex_init (&(this_thread->ThreadMutex), NULL); - pthread_cond_init (&(this_thread->ThreadCond), NULL); - - this_thread->tid = GC_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); - // 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)); -} - -/* - * 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 - */ - pthread_mutex_lock(&this_thread->ThreadMutex); /* To prevent race condition of a sleeping thread */ - ctdl_thread_internal_update_avgs(this_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) - { - if (this_thread->thread_func) // Don't tell garbage collector to stop - { - ctdl_thread_internal_change_state (this_thread, CTDL_THREAD_STOP_REQ); -// pthread_mutex_lock(&this_thread->ThreadMutex); - pthread_cond_signal(&this_thread->ThreadCond); -// pthread_mutex_unlock(&this_thread->ThreadMutex); - 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 signal that we need to do garbage collection on the thread list - */ -void CtdlThreadGC(void) -{ - struct CtdlThreadNode *this_thread; - - CtdlLogPrintf(CTDL_DEBUG, "Thread system signalling garbage collection.\n"); - - begin_critical_section(S_THREAD_LIST); - this_thread = CtdlThreadList; - while(this_thread) - { - if (!this_thread->thread_func) - { -// pthread_mutex_lock(&this_thread->ThreadMutex); - pthread_cond_signal(&this_thread->ThreadCond); -// pthread_mutex_unlock(&this_thread->ThreadMutex); - } - 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; -} - -/* - * A function to find the thread structure for this thread - */ -struct CtdlThreadNode *CtdlThreadSelf(void) -{ - pthread_t self_tid; - struct CtdlThreadNode *this_thread; - - self_tid = pthread_self(); - - begin_critical_section(S_THREAD_LIST); - this_thread = CtdlThreadList; - while(this_thread) - { - if (pthread_equal(self_tid, this_thread->tid)) - { - end_critical_section(S_THREAD_LIST); - return this_thread; - } - this_thread = this_thread->next; - } - end_critical_section(S_THREAD_LIST); - return NULL; -} - - - - -/* - * A function to rename a thread - * Returns a char * and the caller owns the memory and should free it - */ -char *CtdlThreadName(struct CtdlThreadNode *thread, char *name) -{ - struct CtdlThreadNode *this_thread; - char *old_name; - - if (!thread) - this_thread = CtdlThreadSelf(); - else - this_thread = thread; - if (!this_thread) - { - CtdlLogPrintf(CTDL_WARNING, "Thread system WARNING. Attempt to CtdlThreadRename() a non thread. %s\n", name); - return NULL; - } - begin_critical_section(S_THREAD_LIST); - old_name = this_thread->name; - if (name) - this_thread->name = strdup (name); - else - old_name = strdup(old_name); - end_critical_section (S_THREAD_LIST); - return (old_name); -} - - -/* - * A function to force a thread to exit - */ -void CtdlThreadCancel(struct CtdlThreadNode *thread) -{ - struct CtdlThreadNode *this_thread; - - if (!thread) - this_thread = CtdlThreadSelf(); - 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; - } - - begin_critical_section(S_THREAD_LIST); - ctdl_thread_internal_change_state (this_thread, CTDL_THREAD_CANCELLED); - pthread_cancel(this_thread->tid); - end_critical_section (S_THREAD_LIST); -} - - - -/* - * A function for a thread to check if it has been asked to stop - */ -int CtdlThreadCheckStop(struct CtdlThreadNode *this_thread) -{ - if (!this_thread) - { - CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC, CtdlThreadCheckStop() called by a non thread.\n"); - CtdlThreadStopAll(); - return -1; - } - pthread_mutex_lock(&this_thread->ThreadMutex); - if(this_thread->state == CTDL_THREAD_STOP_REQ) - { - this_thread->state = CTDL_THREAD_STOPPING; - pthread_mutex_unlock(&this_thread->ThreadMutex); - return -1; - } - else if(this_thread->state < CTDL_THREAD_STOP_REQ) - { - pthread_mutex_unlock(&this_thread->ThreadMutex); - return -1; - } - pthread_mutex_unlock(&this_thread->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 = CtdlThreadSelf(); - else - this_thread = thread; - if (!this_thread) - return; - if (!(this_thread->thread_func)) - return; // Don't stop garbage collector - - begin_critical_section (S_THREAD_LIST); - ctdl_thread_internal_change_state (this_thread, CTDL_THREAD_STOP_REQ); -// pthread_mutex_lock(&this_thread->ThreadMutex); - pthread_cond_signal(&this_thread->ThreadCond); -// pthread_mutex_unlock(&this_thread->ThreadMutex); - end_critical_section(S_THREAD_LIST); -} - -/* - * 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; - struct CtdlThreadNode *self; - - - self = CtdlThreadSelf(); - if (!self) - { - CtdlLogPrintf(CTDL_WARNING, "CtdlThreadSleep() called by something that is not a thread. Should we die?\n"); - return; - } - - begin_critical_section(S_THREAD_LIST); - ctdl_thread_internal_change_state (self, CTDL_THREAD_SLEEPING); - pthread_mutex_lock(&self->ThreadMutex); /* Prevent something asking us to awaken before we've gone to sleep */ - end_critical_section(S_THREAD_LIST); - - 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; - pthread_cond_timedwait(&self->ThreadCond, &self->ThreadMutex, &wake_time); - begin_critical_section(S_THREAD_LIST); - pthread_mutex_unlock(&self->ThreadMutex); - ctdl_thread_internal_change_state (self, CTDL_THREAD_RUNNING); - end_critical_section(S_THREAD_LIST); -} - - -/* - * Routine to clean up our thread function on exit - */ -static void ctdl_internal_thread_cleanup(void *arg) -{ - struct CtdlThreadNode *this_thread; - this_thread = CtdlThreadSelf(); - /* - * 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", this_thread->name, this_thread->tid); - begin_critical_section(S_THREAD_LIST); - #ifdef HAVE_BACKTRACE - eCrash_UnregisterThread(); - #endif - pthread_mutex_lock(&this_thread->ThreadMutex); - this_thread->state = CTDL_THREAD_EXITED; // needs to be last thing else house keeping will unlink us too early - pthread_mutex_unlock(&this_thread->ThreadMutex); - end_critical_section(S_THREAD_LIST); -// CtdlThreadGC(); -} - -/* - * 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 */ - pthread_mutex_lock(&that_thread->ThreadMutex); - ctdl_thread_internal_update_avgs(that_thread); - 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 ctdl_internal_thread_gc (void) -{ - struct CtdlThreadNode *this_thread, *that_thread; - int workers = 0; - - /* 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? */ - if (that_thread->state != CTDL_THREAD_EXITED) - { - if(that_thread->flags & CTDLTHREAD_WORKER) - workers++; /* Sanity check on number of worker threads */ - continue; - } - - if (pthread_equal(that_thread->tid, pthread_self()) && that_thread->thread_func) - { /* Sanity check */ - CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC, a thread is trying to clean up after itself.\n"); - abort(); - return; - } - - if (num_threads <= 0) - { /* Sanity check */ - CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC, num_threads <= 0 and trying to do Garbage Collection.\n"); - abort(); - return; - } - - /* If we are unlinking the list head then the next becomes the list head */ - if (that_thread == CtdlThreadList) - CtdlThreadList = that_thread->next; - if(that_thread->prev) - that_thread->prev->next = that_thread->next; - if(that_thread->next) - that_thread->next->prev = that_thread->next; - num_threads--; - if(that_thread->flags & CTDLTHREAD_WORKER) - num_workers--; /* This is a wroker thread so reduce the count. */ - - /* - * 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 - * If that thread has no function it must be the garbage collector - */ - if (that_thread->thread_func) - pthread_join (that_thread->tid, NULL); - - /* - * Now we own that thread entry - */ - CtdlLogPrintf(CTDL_INFO, "Garbage Collection for thread \"%s\" (%ld).\n", that_thread->name, that_thread->tid); - if(that_thread->name) - free(that_thread->name); - pthread_mutex_destroy(&that_thread->ThreadMutex); - pthread_cond_destroy(&that_thread->ThreadCond); - pthread_attr_destroy(&that_thread->attr); - free(that_thread); - } - - /* Sanity check number of worker threads */ - if (workers != num_workers) - { - CtdlLogPrintf(CTDL_EMERG, - "Thread system PANIC, discrepancy in number of worker threads. Counted %d, should be %d.\n", - workers, num_workers - ); - abort(); - } -} - - - - -/* - * 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); - // Get our thread data structure - this_thread = (struct CtdlThreadNode *) arg; - this_thread->state = CTDL_THREAD_RUNNING; - this_thread->pid = getpid(); - 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. */ - end_critical_section(S_THREAD_LIST); - - // Tell the world we are here - CtdlLogPrintf(CTDL_NOTICE, "Created a new thread \"%s\" (%ld). \n", this_thread->name, this_thread->tid); - - // Register the cleanup function to take care of when we exit. - pthread_cleanup_push(ctdl_internal_thread_cleanup, NULL); - - - /* - * run the thread to do the work - */ - 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; - int sigtrick = 0; - sigset_t old_signal_set; - - 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)); - - this_thread->state = CTDL_THREAD_CREATE; - - if ((ret = pthread_attr_init(&this_thread->attr))) { - 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))) { - CtdlLogPrintf(CTDL_EMERG, "Thread system, pthread_attr_setstacksize: %s\n", - strerror(ret)); - pthread_attr_destroy(&this_thread->attr); - 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 = strdup(name); - } - else - { - this_thread->name = strdup("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; - pthread_mutex_init (&(this_thread->ThreadMutex), NULL); - pthread_cond_init (&(this_thread->ThreadCond), NULL); - - /* - * We want to make sure that only the main thread handles signals, - * so that each signal is handled exactly once. To do this, we - * make sure that each new thread has all the signals that we - * handle blocked. To avoid race conditions, we block them in - * the spawning thread first, then create the new thread (which - * inherits the settings), and then restore the old settings in - * the spawning thread. This means that there is a brief period - * when no signals will be processed, but during that time they - * should be queued by the operating system. - */ - if (pthread_equal(GC_thread, pthread_self())) - sigtrick = ctdl_thread_internal_block_signals(&old_signal_set) == 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)); - if (this_thread->name) - free (this_thread->name); - pthread_mutex_destroy(&(this_thread->ThreadMutex)); - pthread_cond_destroy(&(this_thread->ThreadCond)); - pthread_attr_destroy(&this_thread->attr); - free(this_thread); - if (sigtrick) - ctdl_thread_internal_restore_signals(&old_signal_set); - return NULL; - } - - if (sigtrick) - ctdl_thread_internal_restore_signals(&old_signal_set); - - 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; - // Register for tracing - #ifdef HAVE_BACKTRACE - eCrash_RegisterThread(this_thread->name, 0); - #endif - 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; -} - - - -/* - * 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, const struct timeval *timeout, struct CtdlThreadNode *self) -{ - int ret; - - ctdl_thread_internal_change_state(self, CTDL_THREAD_BLOCKED); - ret = select(n, readfds, writefds, exceptfds, timeout); - ctdl_thread_internal_change_state(self, 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 * few seconds, because running it after every single unbind would waste a lot * of CPU time and keep the context list locked too much. To force it to run * anyway, set "force" to nonzero. - * - * - * After that's done, we raise the size of the worker thread pool - * if such an action is appropriate. */ void dead_session_purge(int force) { struct CitContext *ptr, *ptr2; /* general-purpose utility pointer */ struct CitContext *rem = NULL; /* list of sessions to be destroyed */ - CtdlThreadPushName("dead_session_purge"); - if (force == 0) { if ( (time(NULL) - last_purge) < 5 ) { - CtdlThreadPopName(); return; /* Too soon, go away */ } } time(&last_purge); - begin_critical_section(S_SESSION_TABLE); + if (try_critical_section(S_SESSION_TABLE)) + return; + ptr = ContextList; while (ptr) { ptr2 = ptr; @@ -1839,11 +1052,9 @@ void dead_session_purge(int force) { } --num_sessions; - /* And put it on our to-be-destroyed list */ ptr2->next = rem; rem = ptr2; - } } end_critical_section(S_SESSION_TABLE); @@ -1859,18 +1070,6 @@ void dead_session_purge(int force) { rem = rem->next; free(ptr); } - - /* Raise the size of the worker thread pool if necessary. */ - begin_critical_section(S_THREAD_LIST); - if ( (num_sessions > num_workers) - && (num_workers < config.c_max_workers) ) { - ctdl_internal_create_thread("Worker Thread", CTDLTHREAD_BIGSTACK + CTDLTHREAD_WORKER, worker_thread, NULL); - } - end_critical_section(S_THREAD_LIST); - // FIXME: reduce the number of worker threads too - - CtdlThreadPopName(); - } @@ -1890,20 +1089,43 @@ void InitializeMasterCC(void) { - - /* * Bind a thread to a context. (It's inline merely to speed things up.) */ INLINE void become_session(struct CitContext *which_con) { - pthread_setspecific(MyConKey, (void *)which_con ); + citthread_setspecific(MyConKey, (void *)which_con ); } /* * This loop just keeps going and going and going... - */ + */ +/* + * FIXME: + * This current implimentation of worker_thread creates a bottle neck in several situations + * The first thing to remember is that a single thread can handle more than one connection at a time. + * More threads mean less memory for the system to run in. + * So for efficiency we want every thread to be doing something useful or waiting in the main loop for + * something to happen anywhere. + * This current implimentation requires worker threads to wait in other locations, after it has + * been committed to a single connection which is very wasteful. + * As an extreme case consider this: + * A slow client connects and this slow client sends only one character each second. + * With this current implimentation a single worker thread is dispatched to handle that connection + * until such times as the client timeout expires, an error occurs on the socket or the client + * completes its transmission. + * THIS IS VERY BAD since that thread could have handled a read from many more clients in each one + * second interval between chars. + * + * It is my intention to re-write this code and the associated client_getln, client_read functions + * to allow any thread to read data on behalf of any connection (context). + * To do this I intend to have this main loop read chars into a buffer stored in the context. + * Once the correct criteria for a full buffer is met then we will dispatch a thread to + * process it. + * This worker thread loop also needs to be able to handle binary data. + */ + void *worker_thread(void *arg) { int i; int highest; @@ -1918,11 +1140,8 @@ void *worker_thread(void *arg) { int force_purge = 0; int m; - CT_PUSH(); - - cdb_allocate_tsd(); - while (!CtdlThreadCheckStop(CT)) { + while (!CtdlThreadCheckStop()) { /* make doubly sure we're not holding any stale db handles * which might cause a deadlock. @@ -1968,14 +1187,13 @@ do_select: force_purge = 0; } } - if (!CtdlThreadCheckStop(CT)) { + if (!CtdlThreadCheckStop()) { tv.tv_sec = 1; /* wake up every second if no input */ tv.tv_usec = 0; - retval = CtdlThreadSelect(highest + 1, &readfds, NULL, NULL, &tv, CT); -// retval = select(highest + 1, &readfds, NULL, NULL, &tv); + retval = CtdlThreadSelect(highest + 1, &readfds, NULL, NULL, &tv); } - - if (CtdlThreadCheckStop(CT)) return(NULL); + else + return NULL; /* Now figure out who made this select() unblock. * First, check for an error or exit condition. @@ -1989,12 +1207,14 @@ do_select: force_purge = 0; if (errno != EINTR) { CtdlLogPrintf(CTDL_EMERG, "Exiting (%s)\n", strerror(errno)); CtdlThreadStopAll(); - } else if (!CtdlThreadCheckStop(CT)) { - CtdlLogPrintf(CTDL_DEBUG, "Un handled select failure.\n"); + } else { + CtdlLogPrintf(CTDL_DEBUG, "Interrupted CtdlThreadSelect.\n"); + if (CtdlThreadCheckStop()) return(NULL); goto do_select; } } else if(retval == 0) { + if (CtdlThreadCheckStop()) return(NULL); goto SKIP_SELECT; } /* Next, check to see if it's a new client connecting @@ -2102,7 +1322,6 @@ SKIP_SELECT: dead_session_purge(force_purge); do_housekeeping(); - check_sched_shutdown(); } /* If control reaches this point, the server is shutting down */ return(NULL);