citadel/threads.c

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