X-Git-Url: https://code.citadel.org/?a=blobdiff_plain;f=citadel%2Fsysdep.c;h=1c7df5e9c4857333406d8e68a7aa3a96cf57bc4c;hb=4ce1695b8f1f04cdbfb0af09b902859259d9918d;hp=6abc426e7391bad9879cd0b1872b1ba1869d2290;hpb=294b6ce260a0fba6e9bb28e2a9d9f42d84044372;p=citadel.git diff --git a/citadel/sysdep.c b/citadel/sysdep.c index 6abc426e7..1c7df5e9c 100644 --- a/citadel/sysdep.c +++ b/citadel/sysdep.c @@ -1,8 +1,6 @@ /* - * $Id$ - * * Citadel "system dependent" stuff. - * See copyright.txt for copyright information. + * See COPYING for copyright information. * * Here's where we (hopefully) have most parts of the Citadel server that * would need to be altered to run the server in a non-POSIX environment. @@ -49,9 +47,7 @@ #include #include #include -#ifdef HAVE_PTHREAD_H -#include -#endif +#define SHOW_ME_VAPPEND_PRINTF #include #include "citadel.h" #include "server.h" @@ -63,6 +59,7 @@ #include "housekeeping.h" #include "modules/crypto/serv_crypto.h" /* Needed for init_ssl, client_write_ssl, client_read_ssl, destruct_ssl */ #include "ecrash.h" +#include "context.h" #ifdef HAVE_SYS_SELECT_H #include @@ -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,83 +87,83 @@ struct igheap *igheap = NULL; #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; 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); -} - +int print_to_logfile = 1; /* - * 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]; if (enable_syslog) { vsyslog((syslog_facility | loglevel), format, arg_ptr); } /* stderr output code */ - if (enable_syslog || running_as_daemon) return; + if (enable_syslog || !print_to_logfile) return; /* if we run in forground and syslog is disabled, log to terminal */ if (loglevel <= verbosity) { struct timeval tv; struct tm tim; time_t unixtime; + CitContext *CCC = CC; + ThreadTSD *cTSD = CTP; + CtdlThreadNode *node = NULL; + long lwpid = 0; + char formatbuf[SIZ]; + char LWP[64]; + char SESS[64]; + + if (cTSD != NULL) { + node = cTSD->self; + } + + if ((node != NULL) && (node->reltid != 0)) { + lwpid = node->reltid; + } gettimeofday(&tv, NULL); + /* Promote to time_t; types differ on some OSes (like darwin) */ unixtime = tv.tv_sec; localtime_r(&unixtime, &tim); - if (CC->cs_pid != 0) { - sprintf(buf, - "%04d/%02d/%02d %2d:%02d:%02d.%06ld [%3d] ", - tim.tm_year + 1900, tim.tm_mon + 1, - tim.tm_mday, tim.tm_hour, tim.tm_min, - tim.tm_sec, (long)tv.tv_usec, - CC->cs_pid); - } else { - sprintf(buf, - "%04d/%02d/%02d %2d:%02d:%02d.%06ld ", - tim.tm_year + 1900, tim.tm_mon + 1, - tim.tm_mday, tim.tm_hour, tim.tm_min, - tim.tm_sec, (long)tv.tv_usec); + + *LWP = '\0'; + if (lwpid != 0) { + snprintf(LWP, 64, "[LWP:%ld] ", lwpid); + } + + *SESS = '\0'; + if (CCC != NULL) { + if (CCC->cs_pid != 0) { + snprintf(SESS, 64, " [%3d] ", CCC->cs_pid); + } + else if (CCC->user.usernum != 0) { + snprintf(SESS, 64, " [:%ld] ", CCC->user.usernum); + } } - vsprintf(buf2, format, arg_ptr); - fprintf(stderr, "%s%s", buf, buf2); + snprintf(formatbuf, SIZ, + "%04d/%02d/%02d %2d:%02d:%02d.%06ld %s%s%s", + tim.tm_year + 1900, tim.tm_mon + 1, + tim.tm_mday, tim.tm_hour, tim.tm_min, + tim.tm_sec, (long)tv.tv_usec, + LWP, SESS, format + ); + + vfprintf(stderr, formatbuf, arg_ptr); fflush(stderr); } } @@ -179,23 +180,20 @@ 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 + { + CtdlLogPrintf(CTDL_DEBUG, "Caught signal %d; shutting down.\n", signum); + exit_signal = signum; } } +static RETSIGTYPE signal_exit(int signum) { + exit(1); +} + /* @@ -226,7 +224,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)); } @@ -236,21 +234,22 @@ void init_sysdep(void) { * call signal_cleanup() to gracefully shut down the server. */ sigemptyset(&set); - sigaddset(&set, SIGINT); - sigaddset(&set, SIGQUIT); + sigaddset(&set, SIGINT); // intr = shutdown + // sigaddset(&set, SIGQUIT); // quit = force quit sigaddset(&set, SIGHUP); sigaddset(&set, SIGTERM); - // sigaddset(&set, SIGSEGV); commented out because - // sigaddset(&set, SIGILL); we want core dumps + // sigaddset(&set, SIGSEGV); // we want core dumps + // sigaddset(&set, SIGILL); // we want core dumps // sigaddset(&set, SIGBUS); sigprocmask(SIG_UNBLOCK, &set, NULL); - signal(SIGINT, signal_cleanup); - signal(SIGQUIT, signal_cleanup); + signal(SIGINT, signal_cleanup); // intr = shutdown + // signal(SIGQUIT, signal_cleanup); // quit = force quit signal(SIGHUP, signal_cleanup); signal(SIGTERM, signal_cleanup); - // signal(SIGSEGV, signal_cleanup); commented out because - // signal(SIGILL, signal_cleanup); we want core dumps + signal(SIGUSR2, signal_exit); + // signal(SIGSEGV, signal_cleanup); // we want coredumps + // signal(SIGILL, signal_cleanup); // we want core dumps // signal(SIGBUS, signal_cleanup); /* @@ -262,158 +261,139 @@ void init_sysdep(void) { } - -/* - * Obtain a semaphore lock to begin a critical section. - * but only if no one else has one +/* + * This is a generic function to set up a master socket for listening on + * a TCP port. The server shuts down if the bind fails. (IPv4/IPv6 version) + * + * ip_addr IP address to bind + * port_number port number to bind + * queue_len number of incoming connections to allow in the queue */ -int try_critical_section(int which_one) +int ctdl_tcp_server(char *ip_addr, int port_number, int queue_len, char *errormessage) { - /* 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) + struct protoent *p; + struct sockaddr_in6 sin6; + struct sockaddr_in sin4; + int s, i, b; + int ip_version = 6; + + memset(&sin6, 0, sizeof(sin6)); + memset(&sin4, 0, sizeof(sin4)); + sin6.sin6_family = AF_INET6; + sin4.sin_family = AF_INET; + + if ( (ip_addr == NULL) /* any IPv6 */ + || (IsEmptyStr(ip_addr)) + || (!strcmp(ip_addr, "*")) ) { - cdb_check_handles(); + ip_version = 6; + sin6.sin6_addr = in6addr_any; } - 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(); + else if (!strcmp(ip_addr, "0.0.0.0")) /* any IPv4 */ + { + ip_version = 4; + sin4.sin_addr.s_addr = INADDR_ANY; } - 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 - * a TCP port. The server shuts down if the bind fails. - * - */ -int ig_tcp_server(char *ip_addr, int port_number, int queue_len, char **errormessage) -{ - struct sockaddr_in sin; - int s, i; - int actual_queue_len; - - actual_queue_len = queue_len; - if (actual_queue_len < 5) actual_queue_len = 5; - - memset(&sin, 0, sizeof(sin)); - sin.sin_family = AF_INET; - sin.sin_port = htons((u_short)port_number); - if (ip_addr == NULL) { - sin.sin_addr.s_addr = INADDR_ANY; + else if ((strchr(ip_addr, '.')) && (!strchr(ip_addr, ':'))) /* specific IPv4 */ + { + ip_version = 4; + if (inet_pton(AF_INET, ip_addr, &sin4.sin_addr) <= 0) { + snprintf(errormessage, SIZ, + "Error binding to [%s] : %s", ip_addr, strerror(errno) + ); + CtdlLogPrintf(CTDL_ALERT, "%s\n", errormessage); + return (-1); + } } - else { - sin.sin_addr.s_addr = inet_addr(ip_addr); + else /* specific IPv6 */ + { + ip_version = 6; + if (inet_pton(AF_INET6, ip_addr, &sin6.sin6_addr) <= 0) { + snprintf(errormessage, SIZ, + "Error binding to [%s] : %s", ip_addr, strerror(errno) + ); + CtdlLogPrintf(CTDL_ALERT, "%s\n", errormessage); + return (-1); + } } - - if (sin.sin_addr.s_addr == !INADDR_ANY) { - sin.sin_addr.s_addr = INADDR_ANY; + + if (port_number == 0) { + snprintf(errormessage, SIZ, + "Can't start: no port number specified." + ); + CtdlLogPrintf(CTDL_ALERT, "%s\n", errormessage); + return (-1); } + sin6.sin6_port = htons((u_short) port_number); + sin4.sin_port = htons((u_short) port_number); - s = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP); + p = getprotobyname("tcp"); + s = socket( ((ip_version == 6) ? PF_INET6 : PF_INET), SOCK_STREAM, (p->p_proto)); if (s < 0) { - *errormessage = (char*) malloc(SIZ + 1); - snprintf(*errormessage, SIZ, - "citserver: Can't create a socket: %s", - strerror(errno)); - CtdlLogPrintf(CTDL_EMERG, "%s\n", *errormessage); - return(-1); + snprintf(errormessage, SIZ, + "Can't create a listening socket: %s", strerror(errno) + ); + CtdlLogPrintf(CTDL_ALERT, "%s\n", errormessage); + return (-1); } - + /* Set some socket options that make sense. */ i = 1; setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &i, sizeof(i)); - if (bind(s, (struct sockaddr *)&sin, sizeof(sin)) < 0) { - *errormessage = (char*) malloc(SIZ + 1); - snprintf(*errormessage, SIZ, - "citserver: Can't bind: %s", - strerror(errno)); - CtdlLogPrintf(CTDL_EMERG, "%s\n", *errormessage); - close(s); - return(-1); + if (ip_version == 6) { + b = bind(s, (struct sockaddr *) &sin6, sizeof(sin6)); } - - /* set to nonblock - we need this for some obscure situations */ - if (fcntl(s, F_SETFL, O_NONBLOCK) < 0) { - *errormessage = (char*) malloc(SIZ + 1); - snprintf(*errormessage, SIZ, - "citserver: Can't set socket to non-blocking: %s", - strerror(errno)); - CtdlLogPrintf(CTDL_EMERG, "%s\n", *errormessage); - close(s); - return(-1); + else { + b = bind(s, (struct sockaddr *) &sin4, sizeof(sin4)); } - if (listen(s, actual_queue_len) < 0) { - *errormessage = (char*) malloc(SIZ + 1); - snprintf(*errormessage, SIZ, - "citserver: Can't listen: %s", - strerror(errno)); - CtdlLogPrintf(CTDL_EMERG, "%s\n", *errormessage); - close(s); - return(-1); + if (b < 0) { + snprintf(errormessage, SIZ, + "Can't bind: %s", strerror(errno) + ); + CtdlLogPrintf(CTDL_ALERT, "%s\n", errormessage); + return (-1); } - return(s); + fcntl(s, F_SETFL, O_NONBLOCK); + + if (listen(s, ((queue_len >= 5) ? queue_len : 5) ) < 0) { + snprintf(errormessage, SIZ, + "Can't listen: %s", strerror(errno) + ); + CtdlLogPrintf(CTDL_ALERT, "%s\n", errormessage); + return (-1); + } + return (s); } + + /* * Create a Unix domain socket and listen on it */ -int ig_uds_server(char *sockpath, int queue_len, char **errormessage) +int ctdl_uds_server(char *sockpath, int queue_len, char *errormessage) { struct sockaddr_un addr; int s; int i; int actual_queue_len; +#ifdef HAVE_STRUCT_UCRED + int passcred = 1; +#endif actual_queue_len = queue_len; if (actual_queue_len < 5) actual_queue_len = 5; i = unlink(sockpath); - if (i != 0) if (errno != ENOENT) { - *errormessage = (char*) malloc(SIZ + 1); - snprintf(*errormessage, SIZ, "citserver: can't unlink %s: %s", - sockpath, strerror(errno)); - CtdlLogPrintf(CTDL_EMERG, "%s\n", *errormessage); + if ((i != 0) && (errno != ENOENT)) { + snprintf(errormessage, SIZ, "citserver: can't unlink %s: %s", + sockpath, strerror(errno) + ); + CtdlLogPrintf(CTDL_EMERG, "%s\n", errormessage); return(-1); } @@ -423,43 +403,43 @@ int ig_uds_server(char *sockpath, int queue_len, char **errormessage) s = socket(AF_UNIX, SOCK_STREAM, 0); if (s < 0) { - *errormessage = (char*) malloc(SIZ + 1); - snprintf(*errormessage, SIZ, + snprintf(errormessage, SIZ, "citserver: Can't create a socket: %s", strerror(errno)); - CtdlLogPrintf(CTDL_EMERG, "%s\n", *errormessage); + CtdlLogPrintf(CTDL_EMERG, "%s\n", errormessage); return(-1); } if (bind(s, (struct sockaddr *)&addr, sizeof(addr)) < 0) { - *errormessage = (char*) malloc(SIZ + 1); - snprintf(*errormessage, SIZ, + snprintf(errormessage, SIZ, "citserver: Can't bind: %s", strerror(errno)); - CtdlLogPrintf(CTDL_EMERG, "%s\n", *errormessage); + CtdlLogPrintf(CTDL_EMERG, "%s\n", errormessage); return(-1); } /* set to nonblock - we need this for some obscure situations */ if (fcntl(s, F_SETFL, O_NONBLOCK) < 0) { - *errormessage = (char*) malloc(SIZ + 1); - snprintf(*errormessage, SIZ, + snprintf(errormessage, SIZ, "citserver: Can't set socket to non-blocking: %s", strerror(errno)); - CtdlLogPrintf(CTDL_EMERG, "%s\n", *errormessage); + CtdlLogPrintf(CTDL_EMERG, "%s\n", errormessage); close(s); return(-1); } if (listen(s, actual_queue_len) < 0) { - *errormessage = (char*) malloc(SIZ + 1); - snprintf(*errormessage, SIZ, + snprintf(errormessage, SIZ, "citserver: Can't listen: %s", strerror(errno)); - CtdlLogPrintf(CTDL_EMERG, "%s\n", *errormessage); + CtdlLogPrintf(CTDL_EMERG, "%s\n", errormessage); return(-1); } +#ifdef HAVE_STRUCT_UCRED + setsockopt(s, SOL_SOCKET, SO_PASSCRED, &passcred, sizeof(passcred)); +#endif + chmod(sockpath, S_ISGID|S_IRUSR|S_IWUSR|S_IXUSR|S_IRGRP|S_IWGRP|S_IXGRP|S_IROTH|S_IWOTH|S_IXOTH); return(s); } @@ -467,67 +447,8 @@ int ig_uds_server(char *sockpath, int queue_len, char **errormessage) /* - * Return a pointer to the CitContext structure bound to the thread which - * called this function. If there's no such binding (for example, if it's - * called by the housekeeper thread) then a generic 'master' CC is returned. - * - * This function is used *VERY* frequently and must be kept small. - */ -struct CitContext *MyContext(void) { - - register struct CitContext *c; - - return ((c = (struct CitContext *) pthread_getspecific(MyConKey), - c == NULL) ? &masterCC : c - ); -} - - -/* - * Initialize a new context and place it in the list. The session number - * used to be the PID (which is why it's called cs_pid), but that was when we - * had one process per session. Now we just assign them sequentially, starting - * at 1 (don't change it to 0 because masterCC uses 0). - */ -struct CitContext *CreateNewContext(void) { - struct CitContext *me; - static int next_pid = 0; - - me = (struct CitContext *) malloc(sizeof(struct CitContext)); - if (me == NULL) { - CtdlLogPrintf(CTDL_ALERT, "citserver: can't allocate memory!!\n"); - return NULL; - } - memset(me, 0, sizeof(struct CitContext)); - - /* 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. - */ - begin_critical_section(S_SESSION_TABLE); - me->cs_pid = ++next_pid; - me->prev = NULL; - me->next = ContextList; - ContextList = me; - if (me->next != NULL) { - me->next->prev = me; - } - ++num_sessions; - end_critical_section(S_SESSION_TABLE); - return(me); -} - - -/* - * The following functions implement output buffering. If the kernel supplies - * native TCP buffering (Linux & *BSD), use that; otherwise, emulate it with - * user-space buffering. + * The following functions implement output buffering on operating systems which + * support it (such as Linux and various BSD flavors). */ #ifndef HAVE_DARWIN #ifdef TCP_CORK @@ -540,113 +461,126 @@ struct CitContext *CreateNewContext(void) { #endif /* TCP_CORK */ #endif /* HAVE_DARWIN */ -#ifdef HAVE_TCP_BUFFERING static unsigned on = 1, off = 0; -void buffer_output(void) { - struct CitContext *ctx = MyContext(); - setsockopt(ctx->client_socket, IPPROTO_TCP, TCP_CORK, &on, 4); - ctx->buffering = 1; -} - -void unbuffer_output(void) { - struct CitContext *ctx = MyContext(); - setsockopt(ctx->client_socket, IPPROTO_TCP, TCP_CORK, &off, 4); - ctx->buffering = 0; -} -void flush_output(void) { - struct CitContext *ctx = MyContext(); - setsockopt(ctx->client_socket, IPPROTO_TCP, TCP_CORK, &off, 4); - setsockopt(ctx->client_socket, IPPROTO_TCP, TCP_CORK, &on, 4); -} -#else -#ifdef HAVE_DARWIN -/* Stub functions for Darwin/OS X where TCP buffering isn't liked at all */ void buffer_output(void) { - CC->buffering = 0; +#ifdef HAVE_TCP_BUFFERING +#ifdef HAVE_OPENSSL + if (!CC->redirect_ssl) +#endif + setsockopt(CC->client_socket, IPPROTO_TCP, TCP_CORK, &on, 4); +#endif } + void unbuffer_output(void) { - CC->buffering = 0; -} -void flush_output(void) { -} -#else -void buffer_output(void) { - if (CC->buffering == 0) { - CC->buffering = 1; - CC->buffer_len = 0; - CC->output_buffer = malloc(SIZ); - } +#ifdef HAVE_TCP_BUFFERING +#ifdef HAVE_OPENSSL + if (!CC->redirect_ssl) +#endif + setsockopt(CC->client_socket, IPPROTO_TCP, TCP_CORK, &off, 4); +#endif } void flush_output(void) { - if (CC->buffering == 1) { - client_write(CC->output_buffer, CC->buffer_len); - CC->buffer_len = 0; - } +#ifdef HAVE_TCP_BUFFERING + struct CitContext *CCC = CC; + setsockopt(CCC->client_socket, IPPROTO_TCP, TCP_CORK, &off, 4); + setsockopt(CCC->client_socket, IPPROTO_TCP, TCP_CORK, &on, 4); +#endif } -void unbuffer_output(void) { - if (CC->buffering == 1) { - CC->buffering = 0; - /* We don't call flush_output because we can't. */ - client_write(CC->output_buffer, CC->buffer_len); - CC->buffer_len = 0; - free(CC->output_buffer); - CC->output_buffer = NULL; - } -} -#endif /* HAVE_DARWIN */ -#endif /* HAVE_TCP_BUFFERING */ +/* +static void flush_client_inbuf(void) +{ + CitContext *CCC=CC; + FlushStrBuf(CCC->ReadBuf); + CCC->RecvBuf->ReadWritePointer = NULL; +} +*/ /* * client_write() ... Send binary data to the client. */ -void client_write(char *buf, int nbytes) +int client_write(const char *buf, int nbytes) { int bytes_written = 0; int retval; #ifndef HAVE_TCP_BUFFERING int old_buffer_len = 0; #endif - t_context *Ctx; + fd_set wset; + CitContext *Ctx; + int fdflags; + + if (nbytes < 1) return(0); Ctx = CC; - if (Ctx->redirect_buffer != NULL) { - if ((Ctx->redirect_len + nbytes + 2) >= Ctx->redirect_alloc) { - Ctx->redirect_alloc = (Ctx->redirect_alloc * 2) + nbytes; - Ctx->redirect_buffer = realloc(Ctx->redirect_buffer, - Ctx->redirect_alloc); - } - memcpy(&Ctx->redirect_buffer[Ctx->redirect_len], buf, nbytes); - Ctx->redirect_len += nbytes; - Ctx->redirect_buffer[Ctx->redirect_len] = 0; - return; - } -#ifndef HAVE_TCP_BUFFERING - /* If we're buffering for later, do that now. */ - if (Ctx->buffering) { - old_buffer_len = Ctx->buffer_len; - Ctx->buffer_len += nbytes; - Ctx->output_buffer = realloc(Ctx->output_buffer, Ctx->buffer_len); - memcpy(&Ctx->output_buffer[old_buffer_len], buf, nbytes); - return; +#ifdef BIGBAD_IODBG + { + int rv = 0; + char fn [SIZ]; + FILE *fd; + + snprintf(fn, SIZ, "/tmp/foolog_%s.%d", Ctx->ServiceName, Ctx->cs_pid); + + fd = fopen(fn, "a+"); + if (fd) + { + fprintf(fd, "Sending: BufSize: %d BufContent: [", + nbytes); + rv = fwrite(buf, nbytes, 1, fd); + fprintf(fd, "]\n"); + fclose(fd); + } } #endif - - /* Ok, at this point we're not buffering. Go ahead and write. */ +// flush_client_inbuf(); + if (Ctx->redirect_buffer != NULL) { + StrBufAppendBufPlain(Ctx->redirect_buffer, + buf, nbytes, 0); + return 0; + } #ifdef HAVE_OPENSSL if (Ctx->redirect_ssl) { client_write_ssl(buf, nbytes); - return; + return 0; } #endif + if (Ctx->client_socket == -1) return -1; + + fdflags = fcntl(Ctx->client_socket, F_GETFL); + + while ((bytes_written < nbytes) && (Ctx->client_socket != -1)){ + if ((fdflags & O_NONBLOCK) == O_NONBLOCK) { + FD_ZERO(&wset); + FD_SET(Ctx->client_socket, &wset); + if (select(1, NULL, &wset, NULL, NULL) == -1) { + if (errno == EINTR) + { + CtdlLogPrintf(CTDL_DEBUG, "client_write(%d bytes) select() interrupted.\n", nbytes-bytes_written); + if (CtdlThreadCheckStop()) { + CC->kill_me = 1; + return (-1); + } else { + /* can't trust fd's and stuff so we need to re-create them */ + continue; + } + } else { + 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; + } + } + } - while (bytes_written < nbytes) { retval = write(Ctx->client_socket, &buf[bytes_written], nbytes - bytes_written); if (retval < 1) { @@ -657,21 +591,25 @@ 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; } +void cputbuf(const StrBuf *Buf) { + client_write(ChrPtr(Buf), StrLength(Buf)); +} + /* - * cprintf() ... Send formatted printable data to the client. It is - * implemented in terms of client_write() but remains in - * sysdep.c in case we port to somewhere without va_args... + * cprintf() Send formatted printable data to the client. + * Implemented in terms of client_write() so it's technically not sysdep... */ void cprintf(const char *format, ...) { va_list arg_ptr; - char buf[1024]; + char buf[1024]; va_start(arg_ptr, format); if (vsnprintf(buf, sizeof buf, format, arg_ptr) == -1) @@ -683,137 +621,386 @@ void cprintf(const char *format, ...) { /* * Read data from the client socket. - * Return values are: - * 1 Requested number of bytes has been read. - * 0 Request timed out. - * -1 The socket is broken. - * If the socket breaks, the session will be terminated. + * + * sock socket fd to read from + * buf buffer to read into + * bytes number of bytes to read + * timeout Number of seconds to wait before timing out + * + * Possible return values: + * 1 Requested number of bytes has been read. + * 0 Request timed out. + * -1 Connection is broken, or other error. */ -int client_read_to(char *buf, int bytes, int timeout) +int client_read_blob(StrBuf *Target, int bytes, int timeout) { - int len,rlen; - fd_set rfds; - int fd; - struct timeval tv; - int retval; + CitContext *CCC=CC; + const char *Error; + int retval = 0; #ifdef HAVE_OPENSSL - if (CC->redirect_ssl) { - return (client_read_ssl(buf, bytes, timeout)); - } + if (CCC->redirect_ssl) { +#ifdef BIGBAD_IODBG + int rv = 0; + char fn [SIZ]; + FILE *fd; + + snprintf(fn, SIZ, "/tmp/foolog_%s.%d", CCC->ServiceName, CCC->cs_pid); + + fd = fopen(fn, "a+"); + fprintf(fd, "Reading BLOB: BufSize: %d ", + bytes); + rv = fwrite(ChrPtr(Target), StrLength(Target), 1, fd); + fprintf(fd, "]\n"); + + + fclose(fd); #endif - len = 0; - fd = CC->client_socket; - while(lenkill_me = 1; - return(-1); +#ifdef BIGBAD_IODBG + snprintf(fn, SIZ, "/tmp/foolog_%s.%d", CCC->ServiceName, CCC->cs_pid); + + fd = fopen(fn, "a+"); + fprintf(fd, "Read: %d BufContent: [", + StrLength(Target)); + rv = fwrite(ChrPtr(Target), StrLength(Target), 1, fd); + fprintf(fd, "]\n"); + + + fclose(fd); +#endif + } + else +#endif + { +#ifdef BIGBAD_IODBG + int rv = 0; + char fn [SIZ]; + FILE *fd; + + snprintf(fn, SIZ, "/tmp/foolog_%s.%d", CCC->ServiceName, CCC->cs_pid); + + fd = fopen(fn, "a+"); + fprintf(fd, "Reading BLOB: BufSize: %d ", + bytes); + rv = fwrite(ChrPtr(Target), StrLength(Target), 1, fd); + fprintf(fd, "]\n"); + + + fclose(fd); +#endif + retval = StrBufReadBLOBBuffered(Target, + CCC->RecvBuf.Buf, + &CCC->RecvBuf.ReadWritePointer, + &CCC->client_socket, + 1, + bytes, + O_TERM, + &Error); + if (retval < 0) { + CtdlLogPrintf(CTDL_CRIT, + "%s failed: %s\n", + __FUNCTION__, + Error); + return retval; } - len = len + rlen; +#ifdef BIGBAD_IODBG + snprintf(fn, SIZ, "/tmp/foolog_%s.%d", CCC->ServiceName, CCC->cs_pid); + + fd = fopen(fn, "a+"); + fprintf(fd, "Read: %d BufContent: [", + StrLength(Target)); + rv = fwrite(ChrPtr(Target), StrLength(Target), 1, fd); + fprintf(fd, "]\n"); + + + fclose(fd); +#endif } - return(1); + return retval; } + /* - * Read data from the client socket with default timeout. - * (This is implemented in terms of client_read_to() and could be - * justifiably moved out of sysdep.c) + * to make client_read_random_blob() more efficient, increase buffer size. + * just use in greeting function, else your buffer may be flushed */ -INLINE int client_read(char *buf, int bytes) +void client_set_inbound_buf(long N) { - return(client_read_to(buf, bytes, config.c_sleeping)); + CitContext *CCC=CC; + FlushStrBuf(CCC->RecvBuf.Buf); + ReAdjustEmptyBuf(CCC->RecvBuf.Buf, N * SIZ, N * SIZ); } - -/* - * client_getln() ... Get a LF-terminated line of text from the client. - * (This is implemented in terms of client_read() and could be - * justifiably moved out of sysdep.c) - */ -int client_getln(char *buf, int bufsize) +int client_read_random_blob(StrBuf *Target, int timeout) { - int i, retval; + CitContext *CCC=CC; + int rc; - /* Read one character at a time. - */ - for (i = 0;;i++) { - retval = client_read(&buf[i], 1); - if (retval != 1 || buf[i] == '\n' || i == (bufsize-1)) - break; + rc = client_read_blob(Target, 1, timeout); + if (rc > 0) + { + long len; + const char *pch; + + len = StrLength(CCC->RecvBuf.Buf); + pch = ChrPtr(CCC->RecvBuf.Buf); + + if (len > 0) + { + if (CCC->RecvBuf.ReadWritePointer != NULL) { + len -= CCC->RecvBuf.ReadWritePointer - pch; + pch = CCC->RecvBuf.ReadWritePointer; + } + StrBufAppendBufPlain(Target, pch, len, 0); + FlushStrBuf(CCC->RecvBuf.Buf); + CCC->RecvBuf.ReadWritePointer = NULL; +#ifdef BIGBAD_IODBG + { + int rv = 0; + char fn [SIZ]; + FILE *fd; + + snprintf(fn, SIZ, "/tmp/foolog_%s.%d", CCC->ServiceName, CCC->cs_pid); + + fd = fopen(fn, "a+"); + fprintf(fd, "Read: BufSize: %d BufContent: [", + StrLength(Target)); + rv = fwrite(ChrPtr(Target), StrLength(Target), 1, fd); + fprintf(fd, "]\n"); + + + fclose(fd); + } +#endif + + return StrLength(Target); + } + return rc; } + else + return rc; +} - /* If we got a long line, discard characters until the newline. - */ - if (i == (bufsize-1)) - while (buf[i] != '\n' && retval == 1) - retval = client_read(&buf[i], 1); +int client_read_to(char *buf, int bytes, int timeout) +{ + CitContext *CCC=CC; + int rc; - /* Strip the trailing LF, and the trailing CR if present. - */ - buf[i] = 0; - while ( (i > 0) - && ( (buf[i - 1]==13) - || ( buf[i - 1]==10)) ) { - i--; - buf[i] = 0; + rc = client_read_blob(CCC->MigrateBuf, bytes, timeout); + if (rc < 0) + { + *buf = '\0'; + return rc; + } + else + { + memcpy(buf, + ChrPtr(CCC->MigrateBuf), + StrLength(CCC->MigrateBuf) + 1); + FlushStrBuf(CCC->MigrateBuf); + return rc; } - if (retval < 0) safestrncpy(&buf[i], "000", bufsize - i); - return(retval); +} + + +int HaveMoreLinesWaiting(CitContext *CCC) +{ + if ((CCC->kill_me == 1) || ( + (CCC->RecvBuf.ReadWritePointer == NULL) && + (StrLength(CCC->RecvBuf.Buf) == 0) && + (CCC->client_socket != -1)) ) + return 0; + else + return 1; } /* - * Cleanup any contexts that are left lying around + * Read data from the client socket with default timeout. + * (This is implemented in terms of client_read_to() and could be + * justifiably moved out of sysdep.c) */ -void context_cleanup(void) +INLINE int client_read(char *buf, int bytes) { - struct CitContext *ptr = NULL; - struct CitContext *rem = NULL; + return(client_read_to(buf, bytes, config.c_sleeping)); +} - /* - * Clean up the contexts. - * 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 +int CtdlClientGetLine(StrBuf *Target) +{ + CitContext *CCC=CC; + const char *Error; + int rc; + + FlushStrBuf(Target); +#ifdef HAVE_OPENSSL + if (CCC->redirect_ssl) { +#ifdef BIGBAD_IODBG + char fn [SIZ]; + FILE *fd; + int len = 0; + int rlen = 0; + int nlen = 0; + int nrlen = 0; + const char *pch; + + snprintf(fn, SIZ, "/tmp/foolog_%s.%d", CCC->ServiceName, CCC->cs_pid); + + fd = fopen(fn, "a+"); + pch = ChrPtr(CCC->RecvBuf.Buf); + len = StrLength(CCC->RecvBuf.Buf); + if (CCC->RecvBuf.ReadWritePointer != NULL) + rlen = CCC->RecvBuf.ReadWritePointer - pch; + else + rlen = 0; + +/* fprintf(fd, "\n\n\nBufSize: %d BufPos: %d \nBufContent: [%s]\n\n_____________________\n", + len, rlen, pch); +*/ + fprintf(fd, "\n\n\nSSL1: BufSize: %d BufPos: %d \n_____________________\n", + len, rlen); +#endif + rc = client_readline_sslbuffer(Target, + CCC->RecvBuf.Buf, + &CCC->RecvBuf.ReadWritePointer, + 1); +#ifdef BIGBAD_IODBG + pch = ChrPtr(CCC->RecvBuf.Buf); + nlen = StrLength(CCC->RecvBuf.Buf); + if (CCC->RecvBuf.ReadWritePointer != NULL) + nrlen = CCC->RecvBuf.ReadWritePointer - pch; + else + nrlen = 0; +/* + fprintf(fd, "\n\n\nBufSize: was: %d is: %d BufPos: was: %d is: %d \nBufContent: [%s]\n\n_____________________\n", + len, nlen, rlen, nrlen, pch); +*/ + fprintf(fd, "\n\n\nSSL2: BufSize: was: %d is: %d BufPos: was: %d is: %d \n", + len, nlen, rlen, nrlen); + + fprintf(fd, "SSL3: Read: BufSize: %d BufContent: [%s]\n\n*************\n", + StrLength(Target), ChrPtr(Target)); + fclose(fd); + + if (rc < 0) + CtdlLogPrintf(CTDL_CRIT, + "%s failed\n", + __FUNCTION__); +#endif + return rc; + } + else +#endif + { +#ifdef BIGBAD_IODBG + char fn [SIZ]; + FILE *fd; + int len, rlen, nlen, nrlen; + const char *pch; + + snprintf(fn, SIZ, "/tmp/foolog_%s.%d", CCC->ServiceName, CCC->cs_pid); + + fd = fopen(fn, "a+"); + pch = ChrPtr(CCC->RecvBuf.Buf); + len = StrLength(CCC->RecvBuf.Buf); + if (CCC->RecvBuf.ReadWritePointer != NULL) + rlen = CCC->RecvBuf.ReadWritePointer - pch; + else + rlen = 0; + +/* fprintf(fd, "\n\n\nBufSize: %d BufPos: %d \nBufContent: [%s]\n\n_____________________\n", + len, rlen, pch); +*/ + fprintf(fd, "\n\n\nBufSize: %d BufPos: %d \n_____________________\n", + len, rlen); +#endif + rc = StrBufTCP_read_buffered_line_fast(Target, + CCC->RecvBuf.Buf, + &CCC->RecvBuf.ReadWritePointer, + &CCC->client_socket, + 5, + 1, + &Error); + +#ifdef BIGBAD_IODBG + pch = ChrPtr(CCC->RecvBuf.Buf); + nlen = StrLength(CCC->RecvBuf.Buf); + if (CCC->RecvBuf.ReadWritePointer != NULL) + nrlen = CCC->RecvBuf.ReadWritePointer - pch; + else + nrlen = 0; +/* + fprintf(fd, "\n\n\nBufSize: was: %d is: %d BufPos: was: %d is: %d \nBufContent: [%s]\n\n_____________________\n", + len, nlen, rlen, nrlen, pch); +*/ + fprintf(fd, "\n\n\nBufSize: was: %d is: %d BufPos: was: %d is: %d \n", + len, nlen, rlen, nrlen); + + fprintf(fd, "Read: BufSize: %d BufContent: [%s]\n\n*************\n", + StrLength(Target), ChrPtr(Target)); + fclose(fd); + + if ((rc < 0) && (Error != NULL)) + CtdlLogPrintf(CTDL_CRIT, + "%s failed: %s\n", + __FUNCTION__, + Error); +#endif + return rc; + } +} + + +/* + * client_getln() ... Get a LF-terminated line of text from the client. + * (This is implemented in terms of client_read() and could be + * justifiably moved out of sysdep.c) + */ +int client_getln(char *buf, int bufsize) +{ + int i, retval; + CitContext *CCC=CC; + const char *pCh; + + retval = CtdlClientGetLine(CCC->MigrateBuf); + if (retval < 0) + return(retval >= 0); + + + i = StrLength(CCC->MigrateBuf); + pCh = ChrPtr(CCC->MigrateBuf); + /* Strip the trailing LF, and the trailing CR if present. */ - 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); - RemoveContext(ptr); - free (ptr); - ptr = rem; + if (bufsize <= i) + i = bufsize - 1; + while ( (i > 0) + && ( (pCh[i - 1]==13) + || ( pCh[i - 1]==10)) ) { + i--; } + memcpy(buf, pCh, i); + buf[i] = 0; + + FlushStrBuf(CCC->MigrateBuf); + if (retval < 0) { + safestrncpy(&buf[i], "000", bufsize - i); + } + return(retval >= 0); } /* - * The system-dependent part of master_cleanup() - close the master socket. + * Cleanup any contexts that are left lying around */ -void sysdep_master_cleanup(void) { + + +void close_masters (void) +{ struct ServiceFunctionHook *serviceptr; /* @@ -823,20 +1010,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(); @@ -854,31 +1053,14 @@ void sysdep_master_cleanup(void) { CtdlDestroySessionHooks(); CtdlDestroyServiceHook(); CtdlDestroyRoomHooks(); - CtdlDestroyDirectoryServiceFuncs(); + CtdlDestroySearchHooks(); #ifdef HAVE_BACKTRACE - eCrash_Uninit(); +/// eCrash_Uninit(); #endif } -/* - * Terminate another session. - * (This could justifiably be moved out of sysdep.c because it - * no longer does anything that is system-dependent.) - */ -void kill_session(int session_to_kill) { - struct CitContext *ptr; - - begin_critical_section(S_SESSION_TABLE); - for (ptr = ContextList; ptr != NULL; ptr = ptr->next) { - if (ptr->cs_pid == session_to_kill) { - ptr->kill_me = 1; - } - } - end_critical_section(S_SESSION_TABLE); -} - pid_t current_child; void graceful_shutdown(int signum) { kill(current_child, signum); @@ -886,6 +1068,9 @@ void graceful_shutdown(int signum) { exit(0); } +int nFireUps = 0; +int nFireUpsNonRestart = 0; +pid_t ForkedPid = 1; /* * Start running as a daemon. @@ -902,7 +1087,10 @@ void start_daemon(int unused) { * We don't just call close() because we don't want these fd's * to be reused for other files. */ - chdir(ctdl_run_dir); + if (chdir(ctdl_run_dir) != 0) + CtdlLogPrintf(CTDL_EMERG, + "unable to change into directory [%s]: %s", + ctdl_run_dir, strerror(errno)); child = fork(); if (child != 0) { @@ -915,9 +1103,13 @@ void start_daemon(int unused) { setsid(); umask(0); - freopen("/dev/null", "r", stdin); - freopen("/dev/null", "w", stdout); - freopen("/dev/null", "w", stderr); + if ((freopen("/dev/null", "r", stdin) != stdin) || + (freopen("/dev/null", "w", stdout) != stdout) || + (freopen("/dev/null", "w", stderr) != stderr)) + CtdlLogPrintf(CTDL_EMERG, + "unable to reopen stdin/out/err %s", + strerror(errno)); + do { current_child = fork(); @@ -936,40 +1128,29 @@ 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); } - do_restart = 0; + nFireUpsNonRestart = nFireUps; + + /* Exit code 0 means the watcher should exit */ + if (WIFEXITED(status) && (WEXITSTATUS(status) == CTDLEXIT_SHUTDOWN)) { + do_restart = 0; + } - /* Did the main process exit with an actual exit code? */ - if (WIFEXITED(status)) { - - /* Exit code 0 means the watcher should exit */ - if (WEXITSTATUS(status) == 0) { - do_restart = 0; - } - - /* Exit code 101-109 means the watcher should exit */ - else if ( (WEXITSTATUS(status) >= 101) && (WEXITSTATUS(status) <= 109) ) { - do_restart = 0; - } - - /* Any other exit code means we should restart. */ - else { - do_restart = 1; - } + /* Exit code 101-109 means the watcher should exit */ + else if (WIFEXITED(status) && (WEXITSTATUS(status) >= 101) && (WEXITSTATUS(status) <= 109)) { + do_restart = 0; } - /* Any other type of termination (signals, etc.) should also restart. */ + /* Any other exit code, or no exit code, means we should restart. */ else { do_restart = 1; + nFireUps++; + ForkedPid = current_child; } } while (do_restart); @@ -980,6 +1161,34 @@ void start_daemon(int unused) { +void checkcrash(void) +{ + if (nFireUpsNonRestart != nFireUps) + { + StrBuf *CrashMail; + + CrashMail = NewStrBuf(); + CtdlLogPrintf(CTDL_ALERT, "Posting crash message\n"); + StrBufPrintf(CrashMail, + " \n" + " The Citadel server process (citserver) terminated unexpectedly." + "\n \n" + " This could be the result of a bug in the server program, or some external " + "factor.\n \n" + " You can obtain more information about this by enabling core dumps.\n \n" + " For more information, please see:\n \n" + " http://citadel.org/doku.php/faq:mastering_your_os:gdb#how.do.i.make.my.system.produce.core-files" + "\n \n" + + " If you have already done this, the core dump is likely to be found at %score.%d\n" + , + ctdl_run_dir, ForkedPid); + CtdlAideMessage(ChrPtr(CrashMail), "Citadel server process terminated unexpectedly"); + FreeStrBuf(&CrashMail); + } +} + + /* * Generic routine to convert a login name to a full name (gecos) * Returns nonzero if a conversion took place @@ -1003,1212 +1212,243 @@ int convert_login(char NameToConvert[]) { +/* + * This loop just keeps going and going and going... + */ /* - * 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 + * 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 highest; + CitContext *ptr; + CitContext *bind_me = NULL; + fd_set readfds; + int retval = 0; + struct timeval tv; + int force_purge = 0; + + while (!CtdlThreadCheckStop()) { -struct CtdlThreadNode *CtdlThreadList = NULL; -struct CtdlThreadNode *CtdlThreadSchedList = NULL; + /* make doubly sure we're not holding any stale db handles + * which might cause a deadlock. + */ + cdb_check_handles(); +do_select: force_purge = 0; + bind_me = NULL; /* Which session shall we handle? */ -/* - * 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; + /* Initialize the fdset. */ + FD_ZERO(&readfds); + highest = 0; -/* - * A function to destroy the TSD - */ -static void ctdl_thread_internal_dest_tsd(void *arg) -{ - if (arg != NULL) { - check_handles(arg); - free(arg); - } -} + begin_critical_section(S_SESSION_TABLE); + for (ptr = ContextList; ptr != NULL; ptr = ptr->next) { + int client_socket; + if ((ptr->state == CON_SYS) && (ptr->client_socket == 0)) + continue; + client_socket = ptr->client_socket; + /* Dont select on dead sessions only truly idle ones */ + if ((ptr->state == CON_IDLE) && + (CC->kill_me == 0) && + (client_socket > 0)) + { + FD_SET(client_socket, &readfds); + if (client_socket > highest) + highest = client_socket; + } + if ((bind_me == NULL) && (ptr->state == CON_READY)) { + bind_me = ptr; + ptr->state = CON_EXECUTING; + break; + } + if ((bind_me == NULL) && (ptr->state == CON_GREETING)) { + bind_me = ptr; + ptr->state = CON_STARTING; + break; + } + } + end_critical_section(S_SESSION_TABLE); + if (bind_me) { + goto SKIP_SELECT; + } -/* - * 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); - } -} + /* If we got this far, it means that there are no sessions + * which a previous thread marked for attention, so we go + * ahead and get ready to select(). + */ -/* - * 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 (!CtdlThreadCheckStop()) { + tv.tv_sec = 1; /* wake up every second if no input */ + tv.tv_usec = 0; + retval = CtdlThreadSelect(highest + 1, &readfds, NULL, NULL, &tv); + } + else + return NULL; - if (pthread_getspecific(ThreadKey) != NULL) - return; + /* Now figure out who made this select() unblock. + * First, check for an error or exit condition. + */ + if (retval < 0) { + if (errno == EBADF) { + CtdlLogPrintf(CTDL_NOTICE, "select() failed: (%s)\n", + strerror(errno)); + goto do_select; + } + if (errno != EINTR) { + CtdlLogPrintf(CTDL_EMERG, "Exiting (%s)\n", strerror(errno)); + CtdlThreadStopAll(); + continue; + } else { + CtdlLogPrintf(CTDL_DEBUG, "Interrupted CtdlThreadSelect.\n"); + if (CtdlThreadCheckStop()) return(NULL); + goto do_select; + } + } + else if(retval == 0) { + if (CtdlThreadCheckStop()) return(NULL); + } - tsd = malloc(sizeof(ThreadTSD)); + /* It must be a client socket. Find a context that has data + * waiting on its socket *and* is in the CON_IDLE state. Any + * active sockets other than our chosen one are marked as + * CON_READY so the next thread that comes around can just bind + * to one without having to select() again. + */ + begin_critical_section(S_SESSION_TABLE); + for (ptr = ContextList; ptr != NULL; ptr = ptr->next) { + int checkfd = ptr->client_socket; + if ((checkfd != -1) && (ptr->state == CON_IDLE) ){ + if (FD_ISSET(checkfd, &readfds)) { + ptr->input_waiting = 1; + if (!bind_me) { + bind_me = ptr; /* I choose you! */ + bind_me->state = CON_EXECUTING; + } + else { + ptr->state = CON_READY; + } + } else if ((ptr->is_async) && (ptr->async_waiting) && (ptr->h_async_function)) { + if (!bind_me) { + bind_me = ptr; /* I choose you! */ + bind_me->state = CON_EXECUTING; + } + else { + ptr->state = CON_READY; + } + } + } + } + end_critical_section(S_SESSION_TABLE); - tsd->tid = NULL; +SKIP_SELECT: + /* We're bound to a session */ + if (bind_me != NULL) { + become_session(bind_me); - memset(tsd->cursors, 0, sizeof tsd->cursors); - tsd->self = NULL; - - pthread_setspecific(ThreadKey, tsd); -} + if (bind_me->state == CON_STARTING) { + bind_me->state = CON_EXECUTING; + begin_session(bind_me); + bind_me->h_greeting_function(); + } + /* If the client has sent a command, execute it. */ + if (CC->input_waiting) { + CC->h_command_function(); + while (HaveMoreLinesWaiting(CC)) + CC->h_command_function(); -void ctdl_thread_internal_free_tsd(void) -{ - ctdl_thread_internal_dest_tsd(pthread_getspecific(ThreadKey)); - pthread_setspecific(ThreadKey, NULL); -} + CC->input_waiting = 0; + } + /* If there are asynchronous messages waiting and the + * client supports it, do those now */ + if ((CC->is_async) && (CC->async_waiting) + && (CC->h_async_function != NULL)) { + CC->h_async_function(); + CC->async_waiting = 0; + } + + force_purge = CC->kill_me; + become_session(NULL); + bind_me->state = CON_IDLE; + } -void ctdl_thread_internal_cleanup(void) -{ - int i; - struct CtdlThreadNode *this_thread, *that_thread; - - for (i=0; inext; - 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); + dead_session_purge(force_purge); + do_housekeeping(); } - ctdl_thread_internal_free_tsd(); + /* If control reaches this point, the server is shutting down */ + return(NULL); } -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 + * A function to handle selecting on master sockets. + * In other words it handles new connections. + * It is a thread. */ -void ctdl_thread_internal_change_state (struct CtdlThreadNode *this_thread, enum CtdlThreadState new_state) +void *select_on_master (void *arg) { - /* - * 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) - { - 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) -{ - int ret; - - begin_critical_section(S_THREAD_LIST); - ret = num_threads; - end_critical_section(S_THREAD_LIST); - return ret; -} - -int CtdlThreadGetWorkers(void) -{ - int ret; - - begin_critical_section(S_THREAD_LIST); - ret = num_workers; - end_critical_section(S_THREAD_LIST); - return ret; -} - -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; - } - 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) -{ - if (!CT) - { - CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC, CtdlThreadCheckStop() called by a non thread.\n"); - CtdlThreadStopAll(); - return -1; - } - pthread_mutex_lock(&CT->ThreadMutex); - if(CT->state == CTDL_THREAD_STOP_REQ) - { - CT->state = CTDL_THREAD_STOPPING; - pthread_mutex_unlock(&CT->ThreadMutex); - return -1; - } - else if((CT->state < CTDL_THREAD_STOP_REQ) && (CT->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 - - 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) - { - CtdlLogPrintf(CTDL_INFO, "Thread system. Creating BIG STACK thread.\n"); - 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 - * 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. - */ -void dead_session_purge(int force) { - struct CitContext *ptr, *ptr2; /* general-purpose utility pointer */ - struct CitContext *rem = NULL; /* list of sessions to be destroyed */ - - if (force == 0) { - if ( (time(NULL) - last_purge) < 5 ) { - return; /* Too soon, go away */ - } - } - time(&last_purge); - - if (try_critical_section(S_SESSION_TABLE)) - return; - - ptr = ContextList; - while (ptr) { - ptr2 = ptr; - ptr = ptr->next; - - if ( (ptr2->state == CON_IDLE) && (ptr2->kill_me) ) { - /* Remove the session from the active list */ - if (ptr2->prev) { - ptr2->prev->next = ptr2->next; - } - else { - ContextList = ptr2->next; - } - if (ptr2->next) { - ptr2->next->prev = ptr2->prev; - } - - --num_sessions; - /* And put it on our to-be-destroyed list */ - ptr2->next = rem; - rem = ptr2; - } - } - end_critical_section(S_SESSION_TABLE); - - /* Now that we no longer have the session list locked, we can take - * our time and destroy any sessions on the to-be-killed list, which - * is allocated privately on this thread's stack. - */ - while (rem != NULL) { - CtdlLogPrintf(CTDL_DEBUG, "Purging session %d\n", rem->cs_pid); - RemoveContext(rem); - ptr = rem; - rem = rem->next; - free(ptr); - } -} - - - - - -/* - * masterCC is the context we use when not attached to a session. This - * function initializes it. - */ -void InitializeMasterCC(void) { - memset(&masterCC, 0, sizeof(struct CitContext)); - masterCC.internal_pgm = 1; - masterCC.cs_pid = 0; -} - - - - - - -/* - * 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 ); -} - - - -/* - * This loop just keeps going and going and going... - */ -void *worker_thread(void *arg) { - int i; - int highest; - struct CitContext *ptr; - struct CitContext *bind_me = NULL; - fd_set readfds; - int retval = 0; - struct CitContext *con= NULL; /* Temporary context pointer */ struct ServiceFunctionHook *serviceptr; - int ssock; /* Descriptor for client socket */ + fd_set master_fds; + int highest; struct timeval tv; - int force_purge = 0; + int ssock; /* Descriptor for client socket */ + CitContext *con= NULL; /* Temporary context pointer */ int m; - - - while (!CtdlThreadCheckStop()) { + int i; + int retval; + struct CitContext select_on_master_CC; - /* make doubly sure we're not holding any stale db handles - * which might cause a deadlock. - */ - cdb_check_handles(); -do_select: force_purge = 0; - bind_me = NULL; /* Which session shall we handle? */ + CtdlFillSystemContext(&select_on_master_CC, "select_on_master"); + citthread_setspecific(MyConKey, (void *)&select_on_master_CC); + while (!CtdlThreadCheckStop()) { /* Initialize the fdset. */ - FD_ZERO(&readfds); + FD_ZERO(&master_fds); highest = 0; - begin_critical_section(S_SESSION_TABLE); - for (ptr = ContextList; ptr != NULL; ptr = ptr->next) { - if (ptr->state == CON_IDLE) { - FD_SET(ptr->client_socket, &readfds); - if (ptr->client_socket > highest) - highest = ptr->client_socket; - } - if ((bind_me == NULL) && (ptr->state == CON_READY)) { - bind_me = ptr; - ptr->state = CON_EXECUTING; - } - } - end_critical_section(S_SESSION_TABLE); - - if (bind_me) { - goto SKIP_SELECT; - } - - /* If we got this far, it means that there are no sessions - * which a previous thread marked for attention, so we go - * ahead and get ready to select(). - */ - /* First, add the various master sockets to the fdset. */ for (serviceptr = ServiceHookTable; serviceptr != NULL; serviceptr = serviceptr->next ) { m = serviceptr->msock; - FD_SET(m, &readfds); + FD_SET(m, &master_fds); if (m > highest) { highest = m; } } if (!CtdlThreadCheckStop()) { - tv.tv_sec = 1; /* wake up every second if no input */ + tv.tv_sec = 60; /* wake up every second if no input */ tv.tv_usec = 0; - retval = CtdlThreadSelect(highest + 1, &readfds, NULL, NULL, &tv); + retval = CtdlThreadSelect(highest + 1, &master_fds, NULL, NULL, &tv); } - - if (CtdlThreadCheckStop()) return(NULL); + else + return NULL; /* Now figure out who made this select() unblock. * First, check for an error or exit condition. @@ -2217,18 +1457,20 @@ do_select: force_purge = 0; if (errno == EBADF) { CtdlLogPrintf(CTDL_NOTICE, "select() failed: (%s)\n", strerror(errno)); - goto do_select; + continue; } if (errno != EINTR) { CtdlLogPrintf(CTDL_EMERG, "Exiting (%s)\n", strerror(errno)); CtdlThreadStopAll(); - } else if (!CtdlThreadCheckStop()) { - CtdlLogPrintf(CTDL_DEBUG, "Un handled select failure.\n"); - goto do_select; + } else { + CtdlLogPrintf(CTDL_DEBUG, "Interrupted CtdlThreadSelect.\n"); + if (CtdlThreadCheckStop()) return(NULL); + continue; } } else if(retval == 0) { - goto SKIP_SELECT; + if (CtdlThreadCheckStop()) return(NULL); + continue; } /* Next, check to see if it's a new client connecting * on a master socket. @@ -2236,7 +1478,7 @@ do_select: force_purge = 0; else for (serviceptr = ServiceHookTable; serviceptr != NULL; serviceptr = serviceptr->next ) { - if (FD_ISSET(serviceptr->msock, &readfds)) { + if (FD_ISSET(serviceptr->msock, &master_fds)) { ssock = accept(serviceptr->msock, NULL, 0); if (ssock >= 0) { CtdlLogPrintf(CTDL_DEBUG, @@ -2264,6 +1506,7 @@ do_select: force_purge = 0; serviceptr->h_command_function; con->h_async_function = serviceptr->h_async_function; + con->h_greeting_function = serviceptr->h_greeting_function; con->ServiceName = serviceptr->ServiceName; @@ -2277,69 +1520,19 @@ do_select: force_purge = 0; SO_REUSEADDR, &i, sizeof(i)); - become_session(con); - begin_session(con); - serviceptr->h_greeting_function(); - become_session(NULL); - con->state = CON_IDLE; - goto do_select; - } - } - } + con->state = CON_GREETING; - /* It must be a client socket. Find a context that has data - * waiting on its socket *and* is in the CON_IDLE state. Any - * active sockets other than our chosen one are marked as - * CON_READY so the next thread that comes around can just bind - * to one without having to select() again. - */ - begin_critical_section(S_SESSION_TABLE); - for (ptr = ContextList; ptr != NULL; ptr = ptr->next) { - if ( (FD_ISSET(ptr->client_socket, &readfds)) - && (ptr->state != CON_EXECUTING) ) { - ptr->input_waiting = 1; - if (!bind_me) { - bind_me = ptr; /* I choose you! */ - bind_me->state = CON_EXECUTING; + retval--; + if (retval == 0) + break; } - else { - ptr->state = CON_READY; - } - } - } - end_critical_section(S_SESSION_TABLE); - -SKIP_SELECT: - /* We're bound to a session */ - if (bind_me != NULL) { - become_session(bind_me); - - /* If the client has sent a command, execute it. */ - if (CC->input_waiting) { - CC->h_command_function(); - CC->input_waiting = 0; - } - - /* If there are asynchronous messages waiting and the - * client supports it, do those now */ - if ((CC->is_async) && (CC->async_waiting) - && (CC->h_async_function != NULL)) { - CC->h_async_function(); - CC->async_waiting = 0; } - - force_purge = CC->kill_me; - become_session(NULL); - bind_me->state = CON_IDLE; } - - dead_session_purge(force_purge); - do_housekeeping(); } - /* If control reaches this point, the server is shutting down */ - return(NULL); -} + CtdlClearSystemContext(); + return NULL; +}