citadel/threads.c

   1 /*
   2  * Thread handling stuff for Citadel server
   3  *
   4  * Copyright (c) 1987-2010 by the citadel.org team
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License as published by
   8  * the Free Software Foundation; either version 3 of the License, or
   9  * (at your option) any later version.
  10  *
  11  * This program is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14  * GNU General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * along with this program; if not, write to the Free Software
  18  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  19  */
  20
  21 #include <stdlib.h>
  22 #include <unistd.h>
  23 #include <stdio.h>
  24 #include <sys/types.h>
  25 #include <errno.h>
  26 #include <sys/socket.h>
  27 #include <unistd.h>
  28 #include <fcntl.h>
  29 #include <signal.h>
  30
  31 #include "sysdep.h"
  32 #if TIME_WITH_SYS_TIME
  33 # include <sys/time.h>
  34 # include <time.h>
  35 #else
  36 # if HAVE_SYS_TIME_H
  37 #  include <sys/time.h>
  38 # else
  39 #  include <time.h>
  40 # endif
  41 #endif
  42
  43 #ifdef HAVE_SYSCALL_H
  44 # include <syscall.h>
  45 #else
  46 # if HAVE_SYS_SYSCALL_H
  47 #  include <sys/syscall.h>
  48 # endif
  49 #endif
  50
  51 #include <libcitadel.h>
  52
  53 #include "threads.h"
  54 #include "ctdl_module.h"
  55 #include "modules_init.h"
  56 #include "housekeeping.h"
  57 #include "config.h"
  58 #include "citserver.h"
  59 #include "sysdep_decls.h"
  60 #include "context.h"
  61 #include "event_client.h"
  62
  63 /*
  64  * define this to use the new worker_thread method of handling connections
  65  */
  66 //#define NEW_WORKER
  67
  68 /*
  69  * New thread interface.
  70  * To create a thread you must call one of the create thread functions.
  71  * You must pass it the address of (a pointer to a CtdlThreadNode initialised to NULL) like this
  72  * struct CtdlThreadNode *node = NULL;
  73  * pass in &node
  74  * If the thread is created *node will point to the thread control structure for the created thread.
  75  * If the thread creation fails *node remains NULL
  76  * Do not free the memory pointed to by *node, it doesn't belong to you.
  77  * This new interface duplicates much of the eCrash stuff. We should go for closer integration since that would
  78  * remove the need for the calls to eCrashRegisterThread and friends
  79  */
  80
  81 static int num_threads = 0;                     /* Current number of threads */
  82 static int num_workers = 0;                     /* Current number of worker threads */
  83 long statcount = 0;             /* are we doing a stats check? */
  84 static long stats_done = 0;
  85
  86 CtdlThreadNode *CtdlThreadList = NULL;
  87 CtdlThreadNode *CtdlThreadSchedList = NULL;
  88
  89 static CtdlThreadNode *GC_thread = NULL;
  90 static char *CtdlThreadStates[CTDL_THREAD_LAST_STATE];
  91 double CtdlThreadLoadAvg = 0;
  92 double CtdlThreadWorkerAvg = 0;
  93 citthread_key_t ThreadKey;
  94
  95 citthread_mutex_t Critters[MAX_SEMAPHORES];     /* Things needing locking */
  96
  97
  98
  99 void InitialiseSemaphores(void)
 100 {
 101         int i;
 102
 103         /* Set up a bunch of semaphores to be used for critical sections */
 104         for (i=0; i<MAX_SEMAPHORES; ++i) {
 105                 citthread_mutex_init(&Critters[i], NULL);
 106         }
 107 }
 108
 109
 110
 111
 112 /*
 113  * Obtain a semaphore lock to begin a critical section.
 114  * but only if no one else has one
 115  */
 116 int try_critical_section(int which_one)
 117 {
 118         /* For all types of critical sections except those listed here,
 119          * ensure nobody ever tries to do a critical section within a
 120          * transaction; this could lead to deadlock.
 121          */
 122         if (    (which_one != S_FLOORCACHE)
 123 #ifdef DEBUG_MEMORY_LEAKS
 124                 && (which_one != S_DEBUGMEMLEAKS)
 125 #endif
 126                 && (which_one != S_RPLIST)
 127         ) {
 128                 cdb_check_handles();
 129         }
 130         return (citthread_mutex_trylock(&Critters[which_one]));
 131 }
 132
 133
 134 /*
 135  * Obtain a semaphore lock to begin a critical section.
 136  */
 137 void begin_critical_section(int which_one)
 138 {
 139         /* CtdlLogPrintf(CTDL_DEBUG, "begin_critical_section(%d)\n", which_one); */
 140
 141         /* For all types of critical sections except those listed here,
 142          * ensure nobody ever tries to do a critical section within a
 143          * transaction; this could lead to deadlock.
 144          */
 145         if (    (which_one != S_FLOORCACHE)
 146 #ifdef DEBUG_MEMORY_LEAKS
 147                 && (which_one != S_DEBUGMEMLEAKS)
 148 #endif
 149                 && (which_one != S_RPLIST)
 150         ) {
 151                 cdb_check_handles();
 152         }
 153         citthread_mutex_lock(&Critters[which_one]);
 154 }
 155
 156 /*
 157  * Release a semaphore lock to end a critical section.
 158  */
 159 void end_critical_section(int which_one)
 160 {
 161         citthread_mutex_unlock(&Critters[which_one]);
 162 }
 163
 164
 165 /*
 166  * A function to destroy the TSD
 167  */
 168 static void ctdl_thread_internal_dest_tsd(void *arg)
 169 {
 170         if (arg != NULL) {
 171                 check_handles(arg);
 172                 free(arg);
 173         }
 174 }
 175
 176
 177 /*
 178  * A function to initialise the thread TSD
 179  */
 180 void ctdl_thread_internal_init_tsd(void)
 181 {
 182         int ret;
 183
 184         if ((ret = citthread_key_create(&ThreadKey, ctdl_thread_internal_dest_tsd))) {
 185                 CtdlLogPrintf(CTDL_EMERG, "citthread_key_create: %s\n", strerror(ret));
 186                 exit(CTDLEXIT_DB);
 187         }
 188 }
 189
 190 /*
 191  * Ensure that we have a key for thread-specific data.
 192  *
 193  * This should be called immediately after startup by any thread
 194  *
 195  */
 196 void CtdlThreadAllocTSD(void)
 197 {
 198         ThreadTSD *tsd;
 199
 200         if (citthread_getspecific(ThreadKey) != NULL)
 201                 return;
 202
 203         tsd = malloc(sizeof(ThreadTSD));
 204
 205         tsd->tid = NULL;
 206
 207         memset(tsd->cursors, 0, sizeof tsd->cursors);
 208         tsd->self = NULL;
 209
 210         citthread_setspecific(ThreadKey, tsd);
 211 }
 212
 213
 214 void ctdl_thread_internal_free_tsd(void)
 215 {
 216         ctdl_thread_internal_dest_tsd(citthread_getspecific(ThreadKey));
 217         citthread_setspecific(ThreadKey, NULL);
 218 }
 219
 220
 221 void ctdl_thread_internal_cleanup(void)
 222 {
 223         int i;
 224         CtdlThreadNode *this_thread, *that_thread;
 225
 226         for (i=0; i<CTDL_THREAD_LAST_STATE; i++)
 227         {
 228                 free (CtdlThreadStates[i]);
 229         }
 230
 231         /* Clean up the scheduled thread list */
 232         this_thread = CtdlThreadSchedList;
 233         while (this_thread)
 234         {
 235                 that_thread = this_thread;
 236                 this_thread = this_thread->next;
 237                 citthread_mutex_destroy(&that_thread->ThreadMutex);
 238                 citthread_cond_destroy(&that_thread->ThreadCond);
 239                 citthread_mutex_destroy(&that_thread->SleepMutex);
 240                 citthread_cond_destroy(&that_thread->SleepCond);
 241                 citthread_attr_destroy(&that_thread->attr);
 242                 free(that_thread);
 243         }
 244         ctdl_thread_internal_free_tsd();
 245 }
 246
 247 void ctdl_thread_internal_init(void)
 248 {
 249         CtdlThreadNode *this_thread;
 250         int ret = 0;
 251
 252         CtdlThreadStates[CTDL_THREAD_INVALID] = strdup ("Invalid Thread");
 253         CtdlThreadStates[CTDL_THREAD_VALID] = strdup("Valid Thread");
 254         CtdlThreadStates[CTDL_THREAD_CREATE] = strdup("Thread being Created");
 255         CtdlThreadStates[CTDL_THREAD_CANCELLED] = strdup("Thread Cancelled");
 256         CtdlThreadStates[CTDL_THREAD_EXITED] = strdup("Thread Exited");
 257         CtdlThreadStates[CTDL_THREAD_STOPPING] = strdup("Thread Stopping");
 258         CtdlThreadStates[CTDL_THREAD_STOP_REQ] = strdup("Thread Stop Requested");
 259         CtdlThreadStates[CTDL_THREAD_SLEEPING] = strdup("Thread Sleeping");
 260         CtdlThreadStates[CTDL_THREAD_RUNNING] = strdup("Thread Running");
 261         CtdlThreadStates[CTDL_THREAD_BLOCKED] = strdup("Thread Blocked");
 262
 263         /* Get ourself a thread entry */
 264         this_thread = malloc(sizeof(CtdlThreadNode));
 265         if (this_thread == NULL) {
 266                 CtdlLogPrintf(CTDL_EMERG, "Thread system, can't allocate CtdlThreadNode, exiting\n");
 267                 return;
 268         }
 269         // Ensuring this is zero'd means we make sure the thread doesn't start doing its thing until we are ready.
 270         memset (this_thread, 0, sizeof(CtdlThreadNode));
 271
 272         citthread_mutex_init (&(this_thread->ThreadMutex), NULL);
 273         citthread_cond_init (&(this_thread->ThreadCond), NULL);
 274         citthread_mutex_init (&(this_thread->SleepMutex), NULL);
 275         citthread_cond_init (&(this_thread->SleepCond), NULL);
 276
 277         /* We are garbage collector so create us as running */
 278         this_thread->state = CTDL_THREAD_RUNNING;
 279
 280         if ((ret = citthread_attr_init(&this_thread->attr))) {
 281                 CtdlLogPrintf(CTDL_EMERG, "Thread system, citthread_attr_init: %s\n", strerror(ret));
 282                 free(this_thread);
 283                 return;
 284         }
 285
 286         this_thread->name = "Garbage Collection Thread";
 287
 288         this_thread->tid = citthread_self();
 289         GC_thread = this_thread;
 290         CT = this_thread;
 291
 292         num_threads++;  // Increase the count of threads in the system.
 293
 294         this_thread->next = CtdlThreadList;
 295         CtdlThreadList = this_thread;
 296         if (this_thread->next)
 297                 this_thread->next->prev = this_thread;
 298         /* Set up start times */
 299         gettimeofday(&this_thread->start_time, NULL);           /* Time this thread started */
 300         memcpy(&this_thread->last_state_change, &this_thread->start_time, sizeof (struct timeval));     /* Changed state so mark it. */
 301 }
 302
 303
 304 /*
 305  * A function to update a threads load averages
 306  */
 307  void ctdl_thread_internal_update_avgs(CtdlThreadNode *this_thread)
 308  {
 309         struct timeval now, result;
 310         double last_duration;
 311
 312         gettimeofday(&now, NULL);
 313         timersub(&now, &(this_thread->last_state_change), &result);
 314         /* I don't think these mutex's are needed here */
 315         citthread_mutex_lock(&this_thread->ThreadMutex);
 316         // result now has a timeval for the time we spent in the last state since we last updated
 317         last_duration = (double)result.tv_sec + ((double)result.tv_usec / (double) 1000000);
 318         if (this_thread->state == CTDL_THREAD_SLEEPING)
 319                 this_thread->avg_sleeping += last_duration;
 320         if (this_thread->state == CTDL_THREAD_RUNNING)
 321                 this_thread->avg_running += last_duration;
 322         if (this_thread->state == CTDL_THREAD_BLOCKED)
 323                 this_thread->avg_blocked += last_duration;
 324         memcpy (&this_thread->last_state_change, &now, sizeof (struct timeval));
 325         citthread_mutex_unlock(&this_thread->ThreadMutex);
 326 }
 327
 328 /*
 329  * A function to chenge the state of a thread
 330  */
 331 void ctdl_thread_internal_change_state (CtdlThreadNode *this_thread, enum CtdlThreadState new_state)
 332 {
 333         /*
 334          * Wether we change state or not we need update the load values
 335          */
 336         ctdl_thread_internal_update_avgs(this_thread);
 337         /* This mutex not needed here? */
 338         citthread_mutex_lock(&this_thread->ThreadMutex); /* To prevent race condition of a sleeping thread */
 339         if ((new_state == CTDL_THREAD_STOP_REQ) && (this_thread->state > CTDL_THREAD_STOP_REQ))
 340                 this_thread->state = new_state;
 341         if (((new_state == CTDL_THREAD_SLEEPING) || (new_state == CTDL_THREAD_BLOCKED)) && (this_thread->state == CTDL_THREAD_RUNNING))
 342                 this_thread->state = new_state;
 343         if ((new_state == CTDL_THREAD_RUNNING) && ((this_thread->state == CTDL_THREAD_SLEEPING) || (this_thread->state == CTDL_THREAD_BLOCKED)))
 344                 this_thread->state = new_state;
 345         citthread_mutex_unlock(&this_thread->ThreadMutex);
 346 }
 347
 348
 349 /*
 350  * A function to tell all threads to exit
 351  */
 352 void CtdlThreadStopAll(void)
 353 {
 354         /* First run any registered shutdown hooks.  This probably doesn't belong here. */
 355         PerformSessionHooks(EVT_SHUTDOWN);
 356
 357         /* then close all tcp ports so nobody else can talk to us anymore. */
 358         CtdlShutdownServiceHooks();
 359         //FIXME: The signalling of the condition should not be in the critical_section
 360         // We need to build a list of threads we are going to signal and then signal them afterwards
 361
 362         ShutDownEventQueue();
 363
 364         CtdlThreadNode *this_thread;
 365
 366         begin_critical_section(S_THREAD_LIST);
 367         this_thread = CtdlThreadList;
 368         // Ask the GC thread to stop first so everything knows we are shutting down.
 369         GC_thread->state = CTDL_THREAD_STOP_REQ;
 370         while(this_thread)
 371         {
 372                 if (!citthread_equal(this_thread->tid, GC_thread->tid))
 373                         citthread_kill(this_thread->tid, SIGHUP);
 374
 375                 ctdl_thread_internal_change_state (this_thread, CTDL_THREAD_STOP_REQ);
 376                 citthread_cond_signal(&this_thread->ThreadCond);
 377                 citthread_cond_signal(&this_thread->SleepCond);
 378                 this_thread->stop_ticker = time(NULL);
 379                 CtdlLogPrintf(CTDL_DEBUG, "Thread system stopping thread \"%s\" (0x%08lx).\n",
 380                         this_thread->name, this_thread->tid);
 381                 this_thread = this_thread->next;
 382         }
 383         end_critical_section(S_THREAD_LIST);
 384 }
 385
 386
 387 /*
 388  * A function to wake up all sleeping threads
 389  */
 390 void CtdlThreadWakeAll(void)
 391 {
 392         CtdlThreadNode *this_thread;
 393
 394         CtdlLogPrintf(CTDL_DEBUG, "Thread system waking all threads.\n");
 395
 396         begin_critical_section(S_THREAD_LIST);
 397         this_thread = CtdlThreadList;
 398         while(this_thread)
 399         {
 400                 if (!this_thread->thread_func)
 401                 {
 402                         citthread_cond_signal(&this_thread->ThreadCond);
 403                         citthread_cond_signal(&this_thread->SleepCond);
 404                 }
 405                 this_thread = this_thread->next;
 406         }
 407         end_critical_section(S_THREAD_LIST);
 408 }
 409
 410
 411 /*
 412  * A function to return the number of threads running in the system
 413  */
 414 int CtdlThreadGetCount(void)
 415 {
 416         return  num_threads;
 417 }
 418
 419 int CtdlThreadGetWorkers(void)
 420 {
 421         return  num_workers;
 422 }
 423
 424 double CtdlThreadGetWorkerAvg(void)
 425 {
 426         double ret;
 427
 428         begin_critical_section(S_THREAD_LIST);
 429         ret =  CtdlThreadWorkerAvg;
 430         end_critical_section(S_THREAD_LIST);
 431         return ret;
 432 }
 433
 434 double CtdlThreadGetLoadAvg(void)
 435 {
 436         double load_avg[3] = {0.0, 0.0, 0.0};
 437
 438         int ret = 0;
 439         int smp_num_cpus;
 440
 441         /* Borrowed this straight from procps */
 442         smp_num_cpus = sysconf(_SC_NPROCESSORS_ONLN);
 443         if(smp_num_cpus<1) smp_num_cpus=1; /* SPARC glibc is buggy */
 444
 445 #ifdef HAVE_GETLOADAVG
 446         ret = getloadavg(load_avg, 3);
 447 #endif
 448         if (ret < 0)
 449                 return 0;
 450         return load_avg[0] / smp_num_cpus;
 451 /*
 452  * This old chunk of code return a value that indicated the load on citserver
 453  * This value could easily reach 100 % even when citserver was doing very little and
 454  * hence the machine has much more spare capacity.
 455  * Because this value was used to determine if the machine was under heavy load conditions
 456  * from other processes in the system then citserver could be strangled un-necesarily
 457  * What we are actually trying to achieve is to strangle citserver if the machine is heavily loaded.
 458  * So we have changed this.
 459
 460         begin_critical_section(S_THREAD_LIST);
 461         ret =  CtdlThreadLoadAvg;
 462         end_critical_section(S_THREAD_LIST);
 463         return ret;
 464 */
 465 }
 466
 467
 468
 469
 470 /*
 471  * A function to rename a thread
 472  * Returns a const char *
 473  */
 474 const char *CtdlThreadName(const char *name)
 475 {
 476         const char *old_name;
 477
 478         if (!CT)
 479         {
 480                 CtdlLogPrintf(CTDL_WARNING, "Thread system WARNING. Attempt to CtdlThreadRename() a non thread. %s\n", name);
 481                 return NULL;
 482         }
 483         old_name = CT->name;
 484         if (name)
 485                 CT->name = name;
 486         return (old_name);
 487 }
 488
 489
 490 /*
 491  * A function to force a thread to exit
 492  */
 493 void CtdlThreadCancel(CtdlThreadNode *thread)
 494 {
 495         CtdlThreadNode *this_thread;
 496
 497         if (!thread)
 498                 this_thread = CT;
 499         else
 500                 this_thread = thread;
 501         if (!this_thread)
 502         {
 503                 CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC. Attempt to CtdlThreadCancel() a non thread.\n");
 504                 CtdlThreadStopAll();
 505                 return;
 506         }
 507
 508         if (!this_thread->thread_func)
 509         {
 510                 CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC. Attempt to CtdlThreadCancel() the garbage collector.\n");
 511                 CtdlThreadStopAll();
 512                 return;
 513         }
 514
 515         ctdl_thread_internal_change_state (this_thread, CTDL_THREAD_CANCELLED);
 516         citthread_cancel(this_thread->tid);
 517 }
 518
 519
 520 /*
 521  * A function for a thread to check if it has been asked to stop
 522  */
 523 int CtdlThreadCheckStop(void)
 524 {
 525         int state;
 526
 527         if (!CT)
 528         {
 529                 CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC, CtdlThreadCheckStop() called by a non thread.\n");
 530                 CtdlThreadStopAll();
 531                 return -1;
 532         }
 533
 534         state = CT->state;
 535
 536         if (CT->signal)
 537         {
 538                 CtdlLogPrintf(CTDL_DEBUG, "Thread \"%s\" caught signal %d.\n", CT->name, CT->signal);
 539                 if (CT->signal == SIGHUP)
 540                         CT->state = CTDL_THREAD_STOP_REQ;
 541                 CT->signal = 0;
 542         }
 543         if(state == CTDL_THREAD_STOP_REQ)
 544         {
 545                 CT->state = CTDL_THREAD_STOPPING;
 546                 return -1;
 547         }
 548         else if((state < CTDL_THREAD_STOP_REQ) && (state > CTDL_THREAD_CREATE))
 549         {
 550                 return -1;
 551         }
 552         return 0;
 553 }
 554
 555
 556 /*
 557  * A function to ask a thread to exit
 558  * The thread must call CtdlThreadCheckStop() periodically to determine if it should exit
 559  */
 560 void CtdlThreadStop(CtdlThreadNode *thread)
 561 {
 562         CtdlThreadNode *this_thread;
 563
 564         if (!thread)
 565                 this_thread = CT;
 566         else
 567                 this_thread = thread;
 568         if (!this_thread)
 569                 return;
 570         if (!(this_thread->thread_func))
 571                 return;         // Don't stop garbage collector
 572
 573         if (!citthread_equal(this_thread->tid, GC_thread->tid))
 574                 citthread_kill(this_thread->tid, SIGHUP);
 575
 576         ctdl_thread_internal_change_state (this_thread, CTDL_THREAD_STOP_REQ);
 577         citthread_cond_signal(&this_thread->ThreadCond);
 578         citthread_cond_signal(&this_thread->SleepCond);
 579         this_thread->stop_ticker = time(NULL);
 580 }
 581
 582 /*
 583  * So we now have a sleep command that works with threads but it is in seconds
 584  */
 585 void CtdlThreadSleep(int secs)
 586 {
 587         struct timespec wake_time;
 588         struct timeval time_now;
 589
 590
 591         if (!CT)
 592         {
 593                 CtdlLogPrintf(CTDL_WARNING, "CtdlThreadSleep() called by something that is not a thread. Should we die?\n");
 594                 return;
 595         }
 596
 597         memset (&wake_time, 0, sizeof(struct timespec));
 598         gettimeofday(&time_now, NULL);
 599         wake_time.tv_sec = time_now.tv_sec + secs;
 600         wake_time.tv_nsec = time_now.tv_usec * 10;
 601
 602         ctdl_thread_internal_change_state (CT, CTDL_THREAD_SLEEPING);
 603
 604         citthread_mutex_lock(&CT->ThreadMutex); /* Prevent something asking us to awaken before we've gone to sleep */
 605         citthread_cond_timedwait(&CT->SleepCond, &CT->ThreadMutex, &wake_time);
 606         citthread_mutex_unlock(&CT->ThreadMutex);
 607
 608         ctdl_thread_internal_change_state (CT, CTDL_THREAD_RUNNING);
 609 }
 610
 611
 612 /*
 613  * Routine to clean up our thread function on exit
 614  */
 615 static void ctdl_internal_thread_cleanup(void *arg)
 616 {
 617         /*
 618          * In here we were called by the current thread because it is exiting
 619          * NB. WE ARE THE CURRENT THREAD
 620          */
 621         if (CT)
 622         {
 623                 const char *name = CT->name;
 624                 const pid_t tid = CT->tid;
 625
 626                 CtdlLogPrintf(CTDL_NOTICE, "Thread \"%s\" (0x%08lx) exited.\n", name, (unsigned long) tid);
 627         }
 628         else
 629         {
 630                 CtdlLogPrintf(CTDL_NOTICE, "some ((unknown ? ? ?) Thread exited.\n");
 631         }
 632
 633         #ifdef HAVE_BACKTRACE
 634 ///     eCrash_UnregisterThread();
 635         #endif
 636
 637         citthread_mutex_lock(&CT->ThreadMutex);
 638         CT->state = CTDL_THREAD_EXITED; // needs to be last thing else house keeping will unlink us too early
 639         citthread_mutex_unlock(&CT->ThreadMutex);
 640 }
 641
 642 /*
 643  * A quick function to show the load averages
 644  */
 645 void ctdl_thread_internal_calc_loadavg(void)
 646 {
 647         CtdlThreadNode *that_thread;
 648         double load_avg, worker_avg;
 649         int workers = 0;
 650
 651         that_thread = CtdlThreadList;
 652         load_avg = 0;
 653         worker_avg = 0;
 654         while(that_thread)
 655         {
 656                 /* Update load averages */
 657                 ctdl_thread_internal_update_avgs(that_thread);
 658                 citthread_mutex_lock(&that_thread->ThreadMutex);
 659                 that_thread->load_avg = (that_thread->avg_sleeping + that_thread->avg_running) / (that_thread->avg_sleeping + that_thread->avg_running + that_thread->avg_blocked) * 100;
 660                 that_thread->avg_sleeping /= 2;
 661                 that_thread->avg_running /= 2;
 662                 that_thread->avg_blocked /= 2;
 663                 load_avg += that_thread->load_avg;
 664                 if (that_thread->flags & CTDLTHREAD_WORKER)
 665                 {
 666                         worker_avg += that_thread->load_avg;
 667                         workers++;
 668                 }
 669 #ifdef WITH_THREADLOG
 670                 CtdlLogPrintf(CTDL_DEBUG, "CtdlThread, \"%s\" (%lu) \"%s\" %.2f %.2f %.2f %.2f\n",
 671                         that_thread->name,
 672                         that_thread->tid,
 673                         CtdlThreadStates[that_thread->state],
 674                         that_thread->avg_sleeping,
 675                         that_thread->avg_running,
 676                         that_thread->avg_blocked,
 677                         that_thread->load_avg);
 678 #endif
 679                 citthread_mutex_unlock(&that_thread->ThreadMutex);
 680                 that_thread = that_thread->next;
 681         }
 682         CtdlThreadLoadAvg = load_avg/num_threads;
 683         CtdlThreadWorkerAvg = worker_avg/workers;
 684 #ifdef WITH_THREADLOG
 685         CtdlLogPrintf(CTDL_INFO, "System load average %.2f, workers averag %.2f, threads %d, workers %d, sessions %d\n", CtdlThreadGetLoadAvg(), CtdlThreadWorkerAvg, num_threads, num_workers, num_sessions);
 686 #endif
 687 }
 688
 689
 690 /*
 691  * Garbage collection routine.
 692  * Gets called by main() in a loop to clean up the thread list periodically.
 693  */
 694 void CtdlThreadGC (void)
 695 {
 696         CtdlThreadNode *this_thread, *that_thread;
 697         int workers = 0, sys_workers;
 698         int ret=0;
 699
 700         begin_critical_section(S_THREAD_LIST);
 701
 702         /* Handle exiting of garbage collector thread */
 703         if(num_threads == 1)
 704                 CtdlThreadList->state = CTDL_THREAD_EXITED;
 705
 706 #ifdef WITH_THREADLOG
 707         CtdlLogPrintf(CTDL_DEBUG, "Thread system running garbage collection.\n");
 708 #endif
 709         /*
 710          * Woke up to do garbage collection
 711          */
 712         this_thread = CtdlThreadList;
 713         while(this_thread)
 714         {
 715                 that_thread = this_thread;
 716                 this_thread = this_thread->next;
 717
 718                 if ((that_thread->state == CTDL_THREAD_STOP_REQ || that_thread->state == CTDL_THREAD_STOPPING)
 719                         && (!citthread_equal(that_thread->tid, citthread_self())))
 720                 {
 721                         CtdlLogPrintf(CTDL_DEBUG, "Waiting for thread %s (0x%08lx) to exit.\n", that_thread->name, that_thread->tid);
 722                         terminate_stuck_sessions();
 723                 }
 724                 else
 725                 {
 726                         /**
 727                          * Catch the situation where a worker was asked to stop but couldn't and we are not
 728                          * shutting down.
 729                          */
 730                         that_thread->stop_ticker = 0;
 731                 }
 732
 733                 if (that_thread->stop_ticker + 5 == time(NULL))
 734                 {
 735                         CtdlLogPrintf(CTDL_DEBUG, "Thread System: The thread \"%s\" (0x%08lx) failed to self terminate within 5 ticks. It would be cancelled now.\n", that_thread->name, that_thread->tid);
 736                         if ((that_thread->flags & CTDLTHREAD_WORKER) == 0)
 737                                 CtdlLogPrintf(CTDL_INFO, "Thread System: A non worker thread would have been canceled this may cause message loss.\n");
 738 //                      that_thread->state = CTDL_THREAD_CANCELLED;
 739                         that_thread->stop_ticker++;
 740 //                      citthread_cancel(that_thread->tid);
 741 //                      continue;
 742                 }
 743
 744                 /* Do we need to clean up this thread? */
 745                 if ((that_thread->state != CTDL_THREAD_EXITED) && (that_thread->state != CTDL_THREAD_CANCELLED))
 746                 {
 747                         if(that_thread->flags & CTDLTHREAD_WORKER)
 748                                 workers++;      /* Sanity check on number of worker threads */
 749                         continue;
 750                 }
 751
 752                 if (citthread_equal(that_thread->tid, citthread_self()) && that_thread->thread_func)
 753                 {       /* Sanity check */
 754                         end_critical_section(S_THREAD_LIST);
 755                         CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC, a thread is trying to clean up after itself.\n");
 756                         abort();
 757                         return;
 758                 }
 759
 760                 if (num_threads <= 0)
 761                 {       /* Sanity check */
 762                         end_critical_section(S_THREAD_LIST);
 763                         CtdlLogPrintf(CTDL_EMERG, "Thread system PANIC, num_threads <= 0 and trying to do Garbage Collection.\n");
 764                         abort();
 765                         return;
 766                 }
 767
 768                 if(that_thread->flags & CTDLTHREAD_WORKER)
 769                         num_workers--;  /* This is a wroker thread so reduce the count. */
 770                 num_threads--;
 771                 /* If we are unlinking the list head then the next becomes the list head */
 772                 if(that_thread->prev)
 773                         that_thread->prev->next = that_thread->next;
 774                 else
 775                         CtdlThreadList = that_thread->next;
 776                 if(that_thread->next)
 777                         that_thread->next->prev = that_thread->prev;
 778
 779                 citthread_cond_signal(&that_thread->ThreadCond);
 780                 citthread_cond_signal(&that_thread->SleepCond); // Make sure this thread is awake
 781                 citthread_mutex_lock(&that_thread->ThreadMutex);        // Make sure it has done what its doing
 782                 citthread_mutex_unlock(&that_thread->ThreadMutex);
 783                 /*
 784                  * Join on the thread to do clean up and prevent memory leaks
 785                  * Also makes sure the thread has cleaned up after itself before we remove it from the list
 786                  * We can join on the garbage collector thread the join should just return EDEADLCK
 787                  */
 788                 ret = citthread_join (that_thread->tid, NULL);
 789                 if (ret == EDEADLK)
 790                         CtdlLogPrintf(CTDL_DEBUG, "Garbage collection on own thread.\n");
 791                 else if (ret == EINVAL)
 792                         CtdlLogPrintf(CTDL_DEBUG, "Garbage collection, that thread already joined on.\n");
 793                 else if (ret == ESRCH)
 794                         CtdlLogPrintf(CTDL_DEBUG, "Garbage collection, no thread to join on.\n");
 795                 else if (ret != 0)
 796                         CtdlLogPrintf(CTDL_DEBUG, "Garbage collection, citthread_join returned an unknown error(%d).\n", ret);
 797                 /*
 798                  * Now we own that thread entry
 799                  */
 800                 CtdlLogPrintf(CTDL_INFO, "Garbage Collection for thread \"%s\" (0x%08lx).\n",
 801                         that_thread->name, that_thread->tid);
 802                 citthread_mutex_destroy(&that_thread->ThreadMutex);
 803                 citthread_cond_destroy(&that_thread->ThreadCond);
 804                 citthread_mutex_destroy(&that_thread->SleepMutex);
 805                 citthread_cond_destroy(&that_thread->SleepCond);
 806                 citthread_attr_destroy(&that_thread->attr);
 807                 free(that_thread);
 808         }
 809         sys_workers = num_workers;
 810         end_critical_section(S_THREAD_LIST);
 811
 812         /* Sanity check number of worker threads */
 813         if (workers != sys_workers)
 814         {
 815                 CtdlLogPrintf(CTDL_EMERG,
 816                         "Thread system PANIC, discrepancy in number of worker threads. Counted %d, should be %d.\n",
 817                         workers, sys_workers
 818                         );
 819                 abort();
 820         }
 821 }
 822
 823
 824
 825
 826 /*
 827  * Runtime function for a Citadel Thread.
 828  * This initialises the threads environment and then calls the user supplied thread function
 829  * Note that this is the REAL thread function and wraps the users thread function.
 830  */
 831 static void *ctdl_internal_thread_func (void *arg)
 832 {
 833         CtdlThreadNode *this_thread;
 834         void *ret = NULL;
 835
 836         /* lock and unlock the thread list.
 837          * This causes this thread to wait until all its creation stuff has finished before it
 838          * can continue its execution.
 839          */
 840         begin_critical_section(S_THREAD_LIST);
 841         this_thread = (CtdlThreadNode *) arg;
 842         gettimeofday(&this_thread->start_time, NULL);           /* Time this thread started */
 843
 844         // Register the cleanup function to take care of when we exit.
 845         citthread_cleanup_push(ctdl_internal_thread_cleanup, NULL);
 846         // Get our thread data structure
 847         CtdlThreadAllocTSD();
 848         CT = this_thread;
 849         this_thread->pid = getpid();
 850         memcpy(&this_thread->last_state_change, &this_thread->start_time, sizeof (struct timeval));     /* Changed state so mark it. */
 851         /* Only change to running state if we weren't asked to stop during the create cycle
 852          * Other wise there is a window to allow this threads creation to continue to full grown and
 853          * therby prevent a shutdown of the server.
 854          */
 855         if (!CtdlThreadCheckStop())
 856         {
 857                 citthread_mutex_lock(&this_thread->ThreadMutex);
 858                 this_thread->state = CTDL_THREAD_RUNNING;
 859                 citthread_mutex_unlock(&this_thread->ThreadMutex);
 860         }
 861         end_critical_section(S_THREAD_LIST);
 862
 863         // Register for tracing
 864         #ifdef HAVE_BACKTRACE
 865 ///     eCrash_RegisterThread(this_thread->name, 0);
 866         #endif
 867
 868         // Tell the world we are here
 869 #if defined(HAVE_SYSCALL_H) && defined (SYS_gettid)
 870         this_thread->reltid = syscall(SYS_gettid);
 871 #endif
 872         CtdlLogPrintf(CTDL_NOTICE, "Created a new thread \"%s\" (0x%08lx).\n",
 873                 this_thread->name, this_thread->tid);
 874
 875         /*
 876          * run the thread to do the work but only if we haven't been asked to stop
 877          */
 878         if (!CtdlThreadCheckStop())
 879                 ret = (this_thread->thread_func)(this_thread->user_args);
 880
 881         /*
 882          * Our thread is exiting either because it wanted to end or because the server is stopping
 883          * We need to clean up
 884          */
 885         citthread_cleanup_pop(1);       // Execute our cleanup routine and remove it
 886
 887         return(ret);
 888 }
 889
 890
 891
 892
 893 /*
 894  * Function to initialise an empty thread structure
 895  */
 896 CtdlThreadNode *ctdl_internal_init_thread_struct(CtdlThreadNode *this_thread, long flags)
 897 {
 898         int ret = 0;
 899
 900         // Ensuring this is zero'd means we make sure the thread doesn't start doing its thing until we are ready.
 901         memset (this_thread, 0, sizeof(CtdlThreadNode));
 902
 903         /* Create the mutex's early so we can use them */
 904         citthread_mutex_init (&(this_thread->ThreadMutex), NULL);
 905         citthread_cond_init (&(this_thread->ThreadCond), NULL);
 906         citthread_mutex_init (&(this_thread->SleepMutex), NULL);
 907         citthread_cond_init (&(this_thread->SleepCond), NULL);
 908
 909         this_thread->state = CTDL_THREAD_CREATE;
 910
 911         if ((ret = citthread_attr_init(&this_thread->attr))) {
 912                 citthread_mutex_unlock(&this_thread->ThreadMutex);
 913                 citthread_mutex_destroy(&(this_thread->ThreadMutex));
 914                 citthread_cond_destroy(&(this_thread->ThreadCond));
 915                 citthread_mutex_destroy(&(this_thread->SleepMutex));
 916                 citthread_cond_destroy(&(this_thread->SleepCond));
 917                 CtdlLogPrintf(CTDL_EMERG, "Thread system, citthread_attr_init: %s\n", strerror(ret));
 918                 free(this_thread);
 919                 return NULL;
 920         }
 921
 922         /* Our per-thread stacks need to be bigger than the default size,
 923          * otherwise the MIME parser crashes on FreeBSD, and the IMAP service
 924          * crashes on 64-bit Linux.
 925          */
 926         if (flags & CTDLTHREAD_BIGSTACK)
 927         {
 928 #ifdef WITH_THREADLOG
 929                 CtdlLogPrintf(CTDL_INFO, "Thread system. Creating BIG STACK thread.\n");
 930 #endif
 931                 if ((ret = citthread_attr_setstacksize(&this_thread->attr, THREADSTACKSIZE))) {
 932                         citthread_mutex_unlock(&this_thread->ThreadMutex);
 933                         citthread_mutex_destroy(&(this_thread->ThreadMutex));
 934                         citthread_cond_destroy(&(this_thread->ThreadCond));
 935                         citthread_mutex_destroy(&(this_thread->SleepMutex));
 936                         citthread_cond_destroy(&(this_thread->SleepCond));
 937                         citthread_attr_destroy(&this_thread->attr);
 938                         CtdlLogPrintf(CTDL_EMERG, "Thread system, citthread_attr_setstacksize: %s\n",
 939                                 strerror(ret));
 940                         free(this_thread);
 941                         return NULL;
 942                 }
 943         }
 944
 945         /* Set this new thread with an avg_blocked of 2. We do this so that its creation affects the
 946          * load average for the system. If we don't do this then we create a mass of threads at the same time
 947          * because the creation didn't affect the load average.
 948          */
 949         this_thread->avg_blocked = 2;
 950
 951         return (this_thread);
 952 }
 953
 954
 955
 956
 957 /*
 958  * Internal function to create a thread.
 959  */
 960 CtdlThreadNode *ctdl_internal_create_thread(char *name, long flags, void *(*thread_func) (void *arg), void *args)
 961 {
 962         int ret = 0;
 963         CtdlThreadNode *this_thread;
 964
 965         if (num_threads >= 32767)
 966         {
 967                 CtdlLogPrintf(CTDL_EMERG, "Thread system. Thread list full.\n");
 968                 return NULL;
 969         }
 970
 971         this_thread = malloc(sizeof(CtdlThreadNode));
 972         if (this_thread == NULL) {
 973                 CtdlLogPrintf(CTDL_EMERG, "Thread system, can't allocate CtdlThreadNode, exiting\n");
 974                 return NULL;
 975         }
 976
 977         /* Initialise the thread structure */
 978         if (ctdl_internal_init_thread_struct(this_thread, flags) == NULL)
 979         {
 980                 free(this_thread);
 981                 CtdlLogPrintf(CTDL_EMERG, "Thread system, can't initialise CtdlThreadNode, exiting\n");
 982                 return NULL;
 983         }
 984         /*
 985          * If we got here we are going to create the thread so we must initilise the structure
 986          * first because most implimentations of threading can't create it in a stopped state
 987          * and it might want to do things with its structure that aren't initialised otherwise.
 988          */
 989         if(name)
 990         {
 991                 this_thread->name = name;
 992         }
 993         else
 994         {
 995                 this_thread->name = "Un-named Thread";
 996         }
 997
 998         this_thread->flags = flags;
 999         this_thread->thread_func = thread_func;
1000         this_thread->user_args = args;
1001
1002         begin_critical_section(S_THREAD_LIST);
1003         /*
1004          * We pass this_thread into the thread as its args so that it can find out information
1005          * about itself and it has a bit of storage space for itself, not to mention that the REAL
1006          * thread function needs to finish off the setup of the structure
1007          */
1008         if ((ret = citthread_create(&this_thread->tid, &this_thread->attr, ctdl_internal_thread_func, this_thread) != 0))
1009         {
1010                 end_critical_section(S_THREAD_LIST);
1011                 CtdlLogPrintf(CTDL_ALERT, "Thread system, Can't create thread: %s\n",
1012                         strerror(ret));
1013                 citthread_mutex_unlock(&this_thread->ThreadMutex);
1014                 citthread_mutex_destroy(&(this_thread->ThreadMutex));
1015                 citthread_cond_destroy(&(this_thread->ThreadCond));
1016                 citthread_mutex_destroy(&(this_thread->SleepMutex));
1017                 citthread_cond_destroy(&(this_thread->SleepCond));
1018                 citthread_attr_destroy(&this_thread->attr);
1019                 free(this_thread);
1020                 return NULL;
1021         }
1022         num_threads++;  // Increase the count of threads in the system.
1023         if(this_thread->flags & CTDLTHREAD_WORKER)
1024                 num_workers++;
1025
1026         this_thread->next = CtdlThreadList;
1027         CtdlThreadList = this_thread;
1028         if (this_thread->next)
1029                 this_thread->next->prev = this_thread;
1030         ctdl_thread_internal_calc_loadavg();
1031
1032         end_critical_section(S_THREAD_LIST);
1033
1034         return this_thread;
1035 }
1036
1037 /*
1038  * Wrapper function to create a thread
1039  * ensures the critical section and other protections are in place.
1040  * char *name = name to give to thread, if NULL, use generic name
1041  * int flags = flags to determine type of thread and standard facilities
1042  */
1043 CtdlThreadNode *CtdlThreadCreate(char *name, long flags, void *(*thread_func) (void *arg), void *args)
1044 {
1045         CtdlThreadNode *ret = NULL;
1046
1047         ret = ctdl_internal_create_thread(name, flags, thread_func, args);
1048         return ret;
1049 }
1050
1051
1052
1053 CtdlThreadNode *ctdl_thread_internal_start_scheduled (CtdlThreadNode *this_thread)
1054 {
1055         int ret = 0;
1056
1057         begin_critical_section(S_THREAD_LIST);
1058         /*
1059          * We pass this_thread into the thread as its args so that it can find out information
1060          * about itself and it has a bit of storage space for itself, not to mention that the REAL
1061          * thread function needs to finish off the setup of the structure
1062          */
1063         if ((ret = citthread_create(&this_thread->tid, &this_thread->attr, ctdl_internal_thread_func, this_thread) != 0))
1064         {
1065                 end_critical_section(S_THREAD_LIST);
1066                 CtdlLogPrintf(CTDL_DEBUG, "Failed to start scheduled thread \"%s\": %s\n", this_thread->name, strerror(ret));
1067                 citthread_mutex_destroy(&(this_thread->ThreadMutex));
1068                 citthread_cond_destroy(&(this_thread->ThreadCond));
1069                 citthread_mutex_destroy(&(this_thread->SleepMutex));
1070                 citthread_cond_destroy(&(this_thread->SleepCond));
1071                 citthread_attr_destroy(&this_thread->attr);
1072                 free(this_thread);
1073                 return NULL;
1074         }
1075
1076
1077         num_threads++;  // Increase the count of threads in the system.
1078         if(this_thread->flags & CTDLTHREAD_WORKER)
1079                 num_workers++;
1080
1081         this_thread->next = CtdlThreadList;
1082         CtdlThreadList = this_thread;
1083         if (this_thread->next)
1084                 this_thread->next->prev = this_thread;
1085
1086         ctdl_thread_internal_calc_loadavg();
1087         end_critical_section(S_THREAD_LIST);
1088
1089
1090         return this_thread;
1091 }
1092
1093
1094
1095 void ctdl_thread_internal_check_scheduled(void)
1096 {
1097         CtdlThreadNode *this_thread, *that_thread;
1098         time_t now;
1099
1100         /* Don't start scheduled threads if the system wants single user mode */
1101         if (CtdlWantSingleUser())
1102                 return;
1103
1104         if (try_critical_section(S_SCHEDULE_LIST))
1105                 return; /* If this list is locked we wait till the next chance */
1106
1107         now = time(NULL);
1108
1109 #ifdef WITH_THREADLOG
1110         CtdlLogPrintf(CTDL_DEBUG, "Checking for scheduled threads to start.\n");
1111 #endif
1112
1113         this_thread = CtdlThreadSchedList;
1114         while(this_thread)
1115         {
1116                 that_thread = this_thread;
1117                 this_thread = this_thread->next;
1118
1119                 if (now > that_thread->when)
1120                 {
1121                         /* Unlink from schedule list */
1122                         if (that_thread->prev)
1123                                 that_thread->prev->next = that_thread->next;
1124                         else
1125                                 CtdlThreadSchedList = that_thread->next;
1126                         if (that_thread->next)
1127                                 that_thread->next->prev = that_thread->prev;
1128
1129                         that_thread->next = that_thread->prev = NULL;
1130 #ifdef WITH_THREADLOG
1131                         CtdlLogPrintf(CTDL_DEBUG, "About to start scheduled thread \"%s\".\n", that_thread->name);
1132 #endif
1133                         if (CT->state > CTDL_THREAD_STOP_REQ)
1134                         {       /* Only start it if the system is not stopping */
1135                                 if (ctdl_thread_internal_start_scheduled (that_thread))
1136                                 {
1137 #ifdef WITH_THREADLOG
1138                                         CtdlLogPrintf(CTDL_INFO, "Thread system, Started a scheduled thread \"%s\" (0x%08lx).\n",
1139                                                 that_thread->name, that_thread->tid);
1140 #endif
1141                                 }
1142                         }
1143                 }
1144 #ifdef WITH_THREADLOG
1145                 else
1146                 {
1147                         CtdlLogPrintf(CTDL_DEBUG, "Thread \"%s\" will start in %ld seconds.\n",
1148                                 that_thread->name, that_thread->when - time(NULL));
1149                 }
1150 #endif
1151         }
1152         end_critical_section(S_SCHEDULE_LIST);
1153 }
1154
1155
1156 /*
1157  * A warapper function for select so we can show a thread as blocked
1158  */
1159 int CtdlThreadSelect(int n, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout)
1160 {
1161         int ret = 0;
1162
1163         ctdl_thread_internal_change_state(CT, CTDL_THREAD_BLOCKED);
1164         if (!CtdlThreadCheckStop())
1165                 ret = select(n, readfds, writefds, exceptfds, timeout);
1166         /**
1167          * If the select returned <= 0 then it failed due to an error
1168          * or timeout so this thread could stop if asked to do so.
1169          * Anything else means it needs to continue unless the system is shutting down
1170          */
1171         if (ret > 0)
1172         {
1173                 /**
1174                  * The select says this thread needs to do something useful.
1175                  * This thread was in an idle state so it may have been asked to stop
1176                  * but if the system isn't shutting down this thread is no longer
1177                  * idle and select has given it a task to do so it must not stop
1178                  * In this condition we need to force it into the running state.
1179                  * CtdlThreadGC will clear its ticker for us.
1180                  *
1181                  * FIXME: there is still a small hole here. It is possible for the sequence of locking
1182                  * to allow the state to get changed to STOP_REQ just after this code if the other thread
1183                  * has decided to change the state before this lock, it there fore has to wait till the lock
1184                  * completes but it will continue to change the state. We need something a bit better here.
1185                  */
1186                 citthread_mutex_lock(&CT->ThreadMutex); /* To prevent race condition of a sleeping thread */
1187                 if (GC_thread->state > CTDL_THREAD_STOP_REQ && CT->state <= CTDL_THREAD_STOP_REQ)
1188                 {
1189                         CtdlLogPrintf(CTDL_DEBUG, "Thread %s (0x%08lx) refused stop request.\n", CT->name, CT->tid);
1190                         CT->state = CTDL_THREAD_RUNNING;
1191                 }
1192                 citthread_mutex_unlock(&CT->ThreadMutex);
1193         }
1194
1195         ctdl_thread_internal_change_state(CT, CTDL_THREAD_RUNNING);
1196
1197         return ret;
1198 }
1199
1200
1201
1202 void *new_worker_thread(void *arg);
1203 extern void close_masters (void);
1204
1205
1206 void *simulation_worker (void*arg) {
1207         struct CitContext *this;
1208
1209         this = CreateNewContext();
1210         CtdlThreadSleep(1);
1211         this->kill_me = 1;
1212         this->state = CON_IDLE;
1213         dead_session_purge(1);
1214         begin_critical_section(S_SESSION_TABLE);
1215         stats_done++;
1216         end_critical_section(S_SESSION_TABLE);
1217         return NULL;
1218 }
1219
1220
1221 void *simulation_thread (void *arg)
1222 {
1223         long stats = statcount;
1224
1225         while(stats && !CtdlThreadCheckStop()) {
1226                 CtdlThreadCreate("Connection simulation worker", CTDLTHREAD_BIGSTACK, simulation_worker, NULL);
1227                 stats--;
1228         }
1229         CtdlThreadStopAll();
1230         return NULL;
1231 }
1232
1233 void go_threading(void)
1234 {
1235         int i;
1236         CtdlThreadNode *last_worker;
1237         struct timeval start, now, result;
1238         double last_duration;
1239
1240         /*
1241          * Initialise the thread system
1242          */
1243         ctdl_thread_internal_init();
1244
1245         /* Second call to module init functions now that threading is up */
1246         if (!statcount) {
1247                 initialise_modules(1);
1248                 CtdlThreadCreate("select_on_master", CTDLTHREAD_BIGSTACK, select_on_master, NULL);
1249         }
1250         else {
1251                 CtdlLogPrintf(CTDL_EMERG, "Running connection simulation stats\n");
1252                 gettimeofday(&start, NULL);
1253                 CtdlThreadCreate("Connection simulation master", CTDLTHREAD_BIGSTACK, simulation_thread, NULL);
1254         }
1255
1256
1257         /*
1258          * This thread is now used for garbage collection of other threads in the thread list
1259          */
1260         CtdlLogPrintf(CTDL_INFO, "Startup thread %ld becoming garbage collector,\n", (long) citthread_self());
1261
1262         /*
1263          * We do a lot of locking and unlocking of the thread list in here.
1264          * We do this so that we can repeatedly release time for other threads
1265          * that may be waiting on the thread list.
1266          * We are a low priority thread so we can afford to do this
1267          */
1268
1269         while (CtdlThreadGetCount())
1270         {
1271                 if (CT->signal)
1272                         exit_signal = CT->signal;
1273                 if (exit_signal)
1274                 {
1275                         CtdlThreadStopAll();
1276                 }
1277                 check_sched_shutdown();
1278                 if (CT->state > CTDL_THREAD_STOP_REQ)
1279                 {
1280                         begin_critical_section(S_THREAD_LIST);
1281                         ctdl_thread_internal_calc_loadavg();
1282                         end_critical_section(S_THREAD_LIST);
1283
1284                         ctdl_thread_internal_check_scheduled(); /* start scheduled threads */
1285                 }
1286
1287                 /* Reduce the size of the worker thread pool if necessary. */
1288                 if ((CtdlThreadGetWorkers() > config.c_min_workers + 1) && (CtdlThreadWorkerAvg < 20) && (CT->state > CTDL_THREAD_STOP_REQ))
1289                 {
1290                         /* Ask a worker thread to stop as we no longer need it */
1291                         begin_critical_section(S_THREAD_LIST);
1292                         last_worker = CtdlThreadList;
1293                         while (last_worker)
1294                         {
1295                                 citthread_mutex_lock(&last_worker->ThreadMutex);
1296                                 if (last_worker->flags & CTDLTHREAD_WORKER && (last_worker->state > CTDL_THREAD_STOPPING) && (last_worker->Context == NULL))
1297                                 {
1298                                         citthread_mutex_unlock(&last_worker->ThreadMutex);
1299                                         break;
1300                                 }
1301                                 citthread_mutex_unlock(&last_worker->ThreadMutex);
1302                                 last_worker = last_worker->next;
1303                         }
1304                         end_critical_section(S_THREAD_LIST);
1305                         if (last_worker)
1306                         {
1307 #ifdef WITH_THREADLOG
1308                                 CtdlLogPrintf(CTDL_DEBUG, "Thread system, stopping excess worker thread \"%s\" (0x%08lx).\n",
1309                                         last_worker->name,
1310                                         last_worker->tid
1311                                         );
1312 #endif
1313                                 CtdlThreadStop(last_worker);
1314                         }
1315                 }
1316
1317                 /*
1318                  * If all our workers are working hard, start some more to help out
1319                  * with things
1320                  */
1321                 /* FIXME: come up with a better way to dynamically alter the number of threads
1322                  * based on the system load
1323                  */
1324                 if (!statcount) {
1325                 if ((((CtdlThreadGetWorkers() < config.c_max_workers) && (CtdlThreadGetWorkerAvg() > 60)) || CtdlThreadGetWorkers() < config.c_min_workers) && (CT->state > CTDL_THREAD_STOP_REQ))
1326                 {
1327                         /* Only start new threads if we are not going to overload the machine */
1328                         /* Temporarily set to 10 should be enough to make sure we don't stranglew the server
1329                          * at least until we make this a config option */
1330                         if (CtdlThreadGetLoadAvg() < ((double)10.00)) {
1331                                 for (i=0; i<5 ; i++) {
1332                                         CtdlThreadCreate("Worker Thread",
1333                                                 CTDLTHREAD_BIGSTACK + CTDLTHREAD_WORKER,
1334                                                 worker_thread,
1335                                                 NULL
1336                                                 );
1337                                 }
1338                         }
1339                         else
1340                                 CtdlLogPrintf (CTDL_WARNING, "Server strangled due to machine load average too high.\n");
1341                 }
1342                 }
1343
1344                 CtdlThreadGC();
1345
1346                 if (CtdlThreadGetCount() <= 1) // Shutting down clean up the garbage collector
1347                 {
1348                         CtdlThreadGC();
1349                 }
1350
1351 #ifdef THREADS_USESIGNALS
1352                 if (CtdlThreadGetCount() && CT->state > CTDL_THREAD_STOP_REQ)
1353 #else
1354                 if (CtdlThreadGetCount())
1355 #endif
1356                         CtdlThreadSleep(1);
1357         }
1358         /*
1359          * If the above loop exits we must be shutting down since we obviously have no threads
1360          */
1361         ctdl_thread_internal_cleanup();
1362
1363         if (statcount) {
1364                 gettimeofday(&now, NULL);
1365                 timersub(&now, &start, &result);
1366                 last_duration = (double)result.tv_sec + ((double)result.tv_usec / (double) 1000000);
1367                 CtdlLogPrintf(CTDL_EMERG, "Simulated %ld connections in %f seconds\n", stats_done, last_duration);
1368         }
1369 }
1370
1371
1372
1373