ghc/rts/Schedule.c

   1 /* ---------------------------------------------------------------------------
   2  * $Id: Schedule.c,v 1.76 2000/08/15 14:18:43 simonmar Exp $
   3  *
   4  * (c) The GHC Team, 1998-2000
   5  *
   6  * Scheduler
   7  *
   8  * The main scheduling code in GranSim is quite different from that in std
   9  * (concurrent) Haskell: while concurrent Haskell just iterates over the
  10  * threads in the runnable queue, GranSim is event driven, i.e. it iterates
  11  * over the events in the global event queue.  -- HWL
  12  * --------------------------------------------------------------------------*/
  13
  14 //@node Main scheduling code, , ,
  15 //@section Main scheduling code
  16
  17 /* Version with scheduler monitor support for SMPs.
  18
  19    This design provides a high-level API to create and schedule threads etc.
  20    as documented in the SMP design document.
  21
  22    It uses a monitor design controlled by a single mutex to exercise control
  23    over accesses to shared data structures, and builds on the Posix threads
  24    library.
  25
  26    The majority of state is shared.  In order to keep essential per-task state,
  27    there is a Capability structure, which contains all the information
  28    needed to run a thread: its STG registers, a pointer to its TSO, a
  29    nursery etc.  During STG execution, a pointer to the capability is
  30    kept in a register (BaseReg).
  31
  32    In a non-SMP build, there is one global capability, namely MainRegTable.
  33
  34    SDM & KH, 10/99
  35 */
  36
  37 //@menu
  38 //* Includes::
  39 //* Variables and Data structures::
  40 //* Main scheduling loop::
  41 //* Suspend and Resume::
  42 //* Run queue code::
  43 //* Garbage Collextion Routines::
  44 //* Blocking Queue Routines::
  45 //* Exception Handling Routines::
  46 //* Debugging Routines::
  47 //* Index::
  48 //@end menu
  49
  50 //@node Includes, Variables and Data structures, Main scheduling code, Main scheduling code
  51 //@subsection Includes
  52
  53 #include "Rts.h"
  54 #include "SchedAPI.h"
  55 #include "RtsUtils.h"
  56 #include "RtsFlags.h"
  57 #include "Storage.h"
  58 #include "StgRun.h"
  59 #include "StgStartup.h"
  60 #include "GC.h"
  61 #include "Hooks.h"
  62 #include "Schedule.h"
  63 #include "StgMiscClosures.h"
  64 #include "Storage.h"
  65 #include "Evaluator.h"
  66 #include "Exception.h"
  67 #include "Printer.h"
  68 #include "Main.h"
  69 #include "Signals.h"
  70 #include "Sanity.h"
  71 #include "Stats.h"
  72 #include "Itimer.h"
  73 #include "Prelude.h"
  74 #if defined(GRAN) || defined(PAR)
  75 # include "GranSimRts.h"
  76 # include "GranSim.h"
  77 # include "ParallelRts.h"
  78 # include "Parallel.h"
  79 # include "ParallelDebug.h"
  80 # include "FetchMe.h"
  81 # include "HLC.h"
  82 #endif
  83 #include "Sparks.h"
  84
  85 #include <stdarg.h>
  86
  87 //@node Variables and Data structures, Prototypes, Includes, Main scheduling code
  88 //@subsection Variables and Data structures
  89
  90 /* Main threads:
  91  *
  92  * These are the threads which clients have requested that we run.
  93  *
  94  * In an SMP build, we might have several concurrent clients all
  95  * waiting for results, and each one will wait on a condition variable
  96  * until the result is available.
  97  *
  98  * In non-SMP, clients are strictly nested: the first client calls
  99  * into the RTS, which might call out again to C with a _ccall_GC, and
 100  * eventually re-enter the RTS.
 101  *
 102  * Main threads information is kept in a linked list:
 103  */
 104 //@cindex StgMainThread
 105 typedef struct StgMainThread_ {
 106   StgTSO *         tso;
 107   SchedulerStatus  stat;
 108   StgClosure **    ret;
 109 #ifdef SMP
 110   pthread_cond_t wakeup;
 111 #endif
 112   struct StgMainThread_ *link;
 113 } StgMainThread;
 114
 115 /* Main thread queue.
 116  * Locks required: sched_mutex.
 117  */
 118 static StgMainThread *main_threads;
 119
 120 /* Thread queues.
 121  * Locks required: sched_mutex.
 122  */
 123 #if defined(GRAN)
 124
 125 StgTSO* ActiveTSO = NULL; /* for assigning system costs; GranSim-Light only */
 126 /* rtsTime TimeOfNextEvent, EndOfTimeSlice;            now in GranSim.c */
 127
 128 /*
 129    In GranSim we have a runable and a blocked queue for each processor.
 130    In order to minimise code changes new arrays run_queue_hds/tls
 131    are created. run_queue_hd is then a short cut (macro) for
 132    run_queue_hds[CurrentProc] (see GranSim.h).
 133    -- HWL
 134 */
 135 StgTSO *run_queue_hds[MAX_PROC], *run_queue_tls[MAX_PROC];
 136 StgTSO *blocked_queue_hds[MAX_PROC], *blocked_queue_tls[MAX_PROC];
 137 StgTSO *ccalling_threadss[MAX_PROC];
 138 /* We use the same global list of threads (all_threads) in GranSim as in
 139    the std RTS (i.e. we are cheating). However, we don't use this list in
 140    the GranSim specific code at the moment (so we are only potentially
 141    cheating).  */
 142
 143 #else /* !GRAN */
 144
 145 StgTSO *run_queue_hd, *run_queue_tl;
 146 StgTSO *blocked_queue_hd, *blocked_queue_tl;
 147
 148 #endif
 149
 150 /* Linked list of all threads.
 151  * Used for detecting garbage collected threads.
 152  */
 153 StgTSO *all_threads;
 154
 155 /* Threads suspended in _ccall_GC.
 156  */
 157 static StgTSO *suspended_ccalling_threads;
 158
 159 static void GetRoots(void);
 160 static StgTSO *threadStackOverflow(StgTSO *tso);
 161
 162 /* KH: The following two flags are shared memory locations.  There is no need
 163        to lock them, since they are only unset at the end of a scheduler
 164        operation.
 165 */
 166
 167 /* flag set by signal handler to precipitate a context switch */
 168 //@cindex context_switch
 169 nat context_switch;
 170
 171 /* if this flag is set as well, give up execution */
 172 //@cindex interrupted
 173 rtsBool interrupted;
 174
 175 /* Next thread ID to allocate.
 176  * Locks required: sched_mutex
 177  */
 178 //@cindex next_thread_id
 179 StgThreadID next_thread_id = 1;
 180
 181 /*
 182  * Pointers to the state of the current thread.
 183  * Rule of thumb: if CurrentTSO != NULL, then we're running a Haskell
 184  * thread.  If CurrentTSO == NULL, then we're at the scheduler level.
 185  */
 186
 187 /* The smallest stack size that makes any sense is:
 188  *    RESERVED_STACK_WORDS    (so we can get back from the stack overflow)
 189  *  + sizeofW(StgStopFrame)   (the stg_stop_thread_info frame)
 190  *  + 1                       (the realworld token for an IO thread)
 191  *  + 1                       (the closure to enter)
 192  *
 193  * A thread with this stack will bomb immediately with a stack
 194  * overflow, which will increase its stack size.
 195  */
 196
 197 #define MIN_STACK_WORDS (RESERVED_STACK_WORDS + sizeofW(StgStopFrame) + 2)
 198
 199 /* Free capability list.
 200  * Locks required: sched_mutex.
 201  */
 202 #ifdef SMP
 203 //@cindex free_capabilities
 204 //@cindex n_free_capabilities
 205 Capability *free_capabilities; /* Available capabilities for running threads */
 206 nat n_free_capabilities;       /* total number of available capabilities */
 207 #else
 208 //@cindex MainRegTable
 209 Capability MainRegTable;       /* for non-SMP, we have one global capability */
 210 #endif
 211
 212 #if defined(GRAN)
 213 StgTSO *CurrentTSO;
 214 #endif
 215
 216 rtsBool ready_to_gc;
 217
 218 /* All our current task ids, saved in case we need to kill them later.
 219  */
 220 #ifdef SMP
 221 //@cindex task_ids
 222 task_info *task_ids;
 223 #endif
 224
 225 void            addToBlockedQueue ( StgTSO *tso );
 226
 227 static void     schedule          ( void );
 228        void     interruptStgRts   ( void );
 229 #if defined(GRAN)
 230 static StgTSO * createThread_     ( nat size, rtsBool have_lock, StgInt pri );
 231 #else
 232 static StgTSO * createThread_     ( nat size, rtsBool have_lock );
 233 #endif
 234
 235 static void     detectBlackHoles  ( void );
 236
 237 #ifdef DEBUG
 238 static void sched_belch(char *s, ...);
 239 #endif
 240
 241 #ifdef SMP
 242 //@cindex sched_mutex
 243 //@cindex term_mutex
 244 //@cindex thread_ready_cond
 245 //@cindex gc_pending_cond
 246 pthread_mutex_t sched_mutex       = PTHREAD_MUTEX_INITIALIZER;
 247 pthread_mutex_t term_mutex        = PTHREAD_MUTEX_INITIALIZER;
 248 pthread_cond_t  thread_ready_cond = PTHREAD_COND_INITIALIZER;
 249 pthread_cond_t  gc_pending_cond   = PTHREAD_COND_INITIALIZER;
 250
 251 nat await_death;
 252 #endif
 253
 254 #if defined(PAR)
 255 StgTSO *LastTSO;
 256 rtsTime TimeOfLastYield;
 257 #endif
 258
 259 #if DEBUG
 260 char *whatNext_strs[] = {
 261   "ThreadEnterGHC",
 262   "ThreadRunGHC",
 263   "ThreadEnterHugs",
 264   "ThreadKilled",
 265   "ThreadComplete"
 266 };
 267
 268 char *threadReturnCode_strs[] = {
 269   "HeapOverflow",                       /* might also be StackOverflow */
 270   "StackOverflow",
 271   "ThreadYielding",
 272   "ThreadBlocked",
 273   "ThreadFinished"
 274 };
 275 #endif
 276
 277 /*
 278  * The thread state for the main thread.
 279 // ToDo: check whether not needed any more
 280 StgTSO   *MainTSO;
 281  */
 282
 283 //@node Main scheduling loop, Suspend and Resume, Prototypes, Main scheduling code
 284 //@subsection Main scheduling loop
 285
 286 /* ---------------------------------------------------------------------------
 287    Main scheduling loop.
 288
 289    We use round-robin scheduling, each thread returning to the
 290    scheduler loop when one of these conditions is detected:
 291
 292       * out of heap space
 293       * timer expires (thread yields)
 294       * thread blocks
 295       * thread ends
 296       * stack overflow
 297
 298    Locking notes:  we acquire the scheduler lock once at the beginning
 299    of the scheduler loop, and release it when
 300
 301       * running a thread, or
 302       * waiting for work, or
 303       * waiting for a GC to complete.
 304
 305    GRAN version:
 306      In a GranSim setup this loop iterates over the global event queue.
 307      This revolves around the global event queue, which determines what
 308      to do next. Therefore, it's more complicated than either the
 309      concurrent or the parallel (GUM) setup.
 310
 311    GUM version:
 312      GUM iterates over incoming messages.
 313      It starts with nothing to do (thus CurrentTSO == END_TSO_QUEUE),
 314      and sends out a fish whenever it has nothing to do; in-between
 315      doing the actual reductions (shared code below) it processes the
 316      incoming messages and deals with delayed operations
 317      (see PendingFetches).
 318      This is not the ugliest code you could imagine, but it's bloody close.
 319
 320    ------------------------------------------------------------------------ */
 321 //@cindex schedule
 322 static void
 323 schedule( void )
 324 {
 325   StgTSO *t;
 326   Capability *cap;
 327   StgThreadReturnCode ret;
 328 #if defined(GRAN)
 329   rtsEvent *event;
 330 #elif defined(PAR)
 331   StgSparkPool *pool;
 332   rtsSpark spark;
 333   StgTSO *tso;
 334   GlobalTaskId pe;
 335 #endif
 336   rtsBool was_interrupted = rtsFalse;
 337
 338   ACQUIRE_LOCK(&sched_mutex);
 339
 340 #if defined(GRAN)
 341
 342   /* set up first event to get things going */
 343   /* ToDo: assign costs for system setup and init MainTSO ! */
 344   new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
 345             ContinueThread,
 346             CurrentTSO, (StgClosure*)NULL, (rtsSpark*)NULL);
 347
 348   IF_DEBUG(gran,
 349            fprintf(stderr, "GRAN: Init CurrentTSO (in schedule) = %p\n", CurrentTSO);
 350            G_TSO(CurrentTSO, 5));
 351
 352   if (RtsFlags.GranFlags.Light) {
 353     /* Save current time; GranSim Light only */
 354     CurrentTSO->gran.clock = CurrentTime[CurrentProc];
 355   }
 356
 357   event = get_next_event();
 358
 359   while (event!=(rtsEvent*)NULL) {
 360     /* Choose the processor with the next event */
 361     CurrentProc = event->proc;
 362     CurrentTSO = event->tso;
 363
 364 #elif defined(PAR)
 365
 366   while (!GlobalStopPending) {          /* GlobalStopPending set in par_exit */
 367
 368 #else
 369
 370   while (1) {
 371
 372 #endif
 373
 374     IF_DEBUG(scheduler, printAllThreads());
 375
 376     /* If we're interrupted (the user pressed ^C, or some other
 377      * termination condition occurred), kill all the currently running
 378      * threads.
 379      */
 380     if (interrupted) {
 381       IF_DEBUG(scheduler, sched_belch("interrupted"));
 382       for (t = run_queue_hd; t != END_TSO_QUEUE; t = t->link) {
 383         deleteThread(t);
 384       }
 385       for (t = blocked_queue_hd; t != END_TSO_QUEUE; t = t->link) {
 386         deleteThread(t);
 387       }
 388       run_queue_hd = run_queue_tl = END_TSO_QUEUE;
 389       blocked_queue_hd = blocked_queue_tl = END_TSO_QUEUE;
 390       interrupted = rtsFalse;
 391       was_interrupted = rtsTrue;
 392     }
 393
 394     /* Go through the list of main threads and wake up any
 395      * clients whose computations have finished.  ToDo: this
 396      * should be done more efficiently without a linear scan
 397      * of the main threads list, somehow...
 398      */
 399 #ifdef SMP
 400     {
 401       StgMainThread *m, **prev;
 402       prev = &main_threads;
 403       for (m = main_threads; m != NULL; m = m->link) {
 404         switch (m->tso->what_next) {
 405         case ThreadComplete:
 406           if (m->ret) {
 407             *(m->ret) = (StgClosure *)m->tso->sp[0];
 408           }
 409           *prev = m->link;
 410           m->stat = Success;
 411           pthread_cond_broadcast(&m->wakeup);
 412           break;
 413         case ThreadKilled:
 414           *prev = m->link;
 415           if (was_interrupted) {
 416             m->stat = Interrupted;
 417           } else {
 418             m->stat = Killed;
 419           }
 420           pthread_cond_broadcast(&m->wakeup);
 421           break;
 422         default:
 423           break;
 424         }
 425       }
 426     }
 427
 428 #else
 429 # if defined(PAR)
 430     /* in GUM do this only on the Main PE */
 431     if (IAmMainThread)
 432 # endif
 433     /* If our main thread has finished or been killed, return.
 434      */
 435     {
 436       StgMainThread *m = main_threads;
 437       if (m->tso->what_next == ThreadComplete
 438           || m->tso->what_next == ThreadKilled) {
 439         main_threads = main_threads->link;
 440         if (m->tso->what_next == ThreadComplete) {
 441           /* we finished successfully, fill in the return value */
 442           if (m->ret) { *(m->ret) = (StgClosure *)m->tso->sp[0]; };
 443           m->stat = Success;
 444           return;
 445         } else {
 446           if (was_interrupted) {
 447             m->stat = Interrupted;
 448           } else {
 449             m->stat = Killed;
 450           }
 451           return;
 452         }
 453       }
 454     }
 455 #endif
 456
 457     /* Top up the run queue from our spark pool.  We try to make the
 458      * number of threads in the run queue equal to the number of
 459      * free capabilities.
 460      */
 461 #if defined(SMP)
 462     {
 463       nat n = n_free_capabilities;
 464       StgTSO *tso = run_queue_hd;
 465
 466       /* Count the run queue */
 467       while (n > 0 && tso != END_TSO_QUEUE) {
 468         tso = tso->link;
 469         n--;
 470       }
 471
 472       for (; n > 0; n--) {
 473         StgClosure *spark;
 474         spark = findSpark();
 475         if (spark == NULL) {
 476           break; /* no more sparks in the pool */
 477         } else {
 478           /* I'd prefer this to be done in activateSpark -- HWL */
 479           /* tricky - it needs to hold the scheduler lock and
 480            * not try to re-acquire it -- SDM */
 481           StgTSO *tso;
 482           tso = createThread_(RtsFlags.GcFlags.initialStkSize, rtsTrue);
 483           pushClosure(tso,spark);
 484           PUSH_ON_RUN_QUEUE(tso);
 485 #ifdef PAR
 486           advisory_thread_count++;
 487 #endif
 488
 489           IF_DEBUG(scheduler,
 490                    sched_belch("turning spark of closure %p into a thread",
 491                                (StgClosure *)spark));
 492         }
 493       }
 494       /* We need to wake up the other tasks if we just created some
 495        * work for them.
 496        */
 497       if (n_free_capabilities - n > 1) {
 498           pthread_cond_signal(&thread_ready_cond);
 499       }
 500     }
 501 #endif /* SMP */
 502
 503     /* Check whether any waiting threads need to be woken up.  If the
 504      * run queue is empty, and there are no other tasks running, we
 505      * can wait indefinitely for something to happen.
 506      * ToDo: what if another client comes along & requests another
 507      * main thread?
 508      */
 509     if (blocked_queue_hd != END_TSO_QUEUE) {
 510       awaitEvent(
 511            (run_queue_hd == END_TSO_QUEUE)
 512 #ifdef SMP
 513         && (n_free_capabilities == RtsFlags.ParFlags.nNodes)
 514 #endif
 515         );
 516     }
 517
 518     /* check for signals each time around the scheduler */
 519 #ifndef mingw32_TARGET_OS
 520     if (signals_pending()) {
 521       start_signal_handlers();
 522     }
 523 #endif
 524
 525     /*
 526      * Detect deadlock: when we have no threads to run, there are no
 527      * threads waiting on I/O or sleeping, and all the other tasks are
 528      * waiting for work, we must have a deadlock of some description.
 529      *
 530      * We first try to find threads blocked on themselves (ie. black
 531      * holes), and generate NonTermination exceptions where necessary.
 532      *
 533      * If no threads are black holed, we have a deadlock situation, so
 534      * inform all the main threads.
 535      */
 536 #ifdef SMP
 537     if (blocked_queue_hd == END_TSO_QUEUE
 538         && run_queue_hd == END_TSO_QUEUE
 539         && (n_free_capabilities == RtsFlags.ParFlags.nNodes))
 540     {
 541         IF_DEBUG(scheduler, sched_belch("deadlocked, checking for black holes..."));
 542         detectBlackHoles();
 543         if (run_queue_hd == END_TSO_QUEUE) {
 544             StgMainThread *m;
 545             for (m = main_threads; m != NULL; m = m->link) {
 546                 m->ret = NULL;
 547                 m->stat = Deadlock;
 548                 pthread_cond_broadcast(&m->wakeup);
 549             }
 550             main_threads = NULL;
 551         }
 552     }
 553 #else /* ! SMP */
 554     if (blocked_queue_hd == END_TSO_QUEUE
 555         && run_queue_hd == END_TSO_QUEUE)
 556     {
 557         IF_DEBUG(scheduler, sched_belch("deadlocked, checking for black holes..."));
 558         detectBlackHoles();
 559         if (run_queue_hd == END_TSO_QUEUE) {
 560             StgMainThread *m = main_threads;
 561             m->ret = NULL;
 562             m->stat = Deadlock;
 563             main_threads = m->link;
 564             return;
 565         }
 566     }
 567 #endif
 568
 569 #ifdef SMP
 570     /* If there's a GC pending, don't do anything until it has
 571      * completed.
 572      */
 573     if (ready_to_gc) {
 574       IF_DEBUG(scheduler,sched_belch("waiting for GC"));
 575       pthread_cond_wait(&gc_pending_cond, &sched_mutex);
 576     }
 577
 578     /* block until we've got a thread on the run queue and a free
 579      * capability.
 580      */
 581     while (run_queue_hd == END_TSO_QUEUE || free_capabilities == NULL) {
 582       IF_DEBUG(scheduler, sched_belch("waiting for work"));
 583       pthread_cond_wait(&thread_ready_cond, &sched_mutex);
 584       IF_DEBUG(scheduler, sched_belch("work now available"));
 585     }
 586 #endif
 587
 588 #if defined(GRAN)
 589
 590     if (RtsFlags.GranFlags.Light)
 591       GranSimLight_enter_system(event, &ActiveTSO); // adjust ActiveTSO etc
 592
 593     /* adjust time based on time-stamp */
 594     if (event->time > CurrentTime[CurrentProc] &&
 595         event->evttype != ContinueThread)
 596       CurrentTime[CurrentProc] = event->time;
 597
 598     /* Deal with the idle PEs (may issue FindWork or MoveSpark events) */
 599     if (!RtsFlags.GranFlags.Light)
 600       handleIdlePEs();
 601
 602     IF_DEBUG(gran, fprintf(stderr, "GRAN: switch by event-type\n"))
 603
 604     /* main event dispatcher in GranSim */
 605     switch (event->evttype) {
 606       /* Should just be continuing execution */
 607     case ContinueThread:
 608       IF_DEBUG(gran, fprintf(stderr, "GRAN: doing ContinueThread\n"));
 609       /* ToDo: check assertion
 610       ASSERT(run_queue_hd != (StgTSO*)NULL &&
 611              run_queue_hd != END_TSO_QUEUE);
 612       */
 613       /* Ignore ContinueThreads for fetching threads (if synchr comm) */
 614       if (!RtsFlags.GranFlags.DoAsyncFetch &&
 615           procStatus[CurrentProc]==Fetching) {
 616         belch("ghuH: Spurious ContinueThread while Fetching ignored; TSO %d (%p) [PE %d]",
 617               CurrentTSO->id, CurrentTSO, CurrentProc);
 618         goto next_thread;
 619       }
 620       /* Ignore ContinueThreads for completed threads */
 621       if (CurrentTSO->what_next == ThreadComplete) {
 622         belch("ghuH: found a ContinueThread event for completed thread %d (%p) [PE %d] (ignoring ContinueThread)",
 623               CurrentTSO->id, CurrentTSO, CurrentProc);
 624         goto next_thread;
 625       }
 626       /* Ignore ContinueThreads for threads that are being migrated */
 627       if (PROCS(CurrentTSO)==Nowhere) {
 628         belch("ghuH: trying to run the migrating TSO %d (%p) [PE %d] (ignoring ContinueThread)",
 629               CurrentTSO->id, CurrentTSO, CurrentProc);
 630         goto next_thread;
 631       }
 632       /* The thread should be at the beginning of the run queue */
 633       if (CurrentTSO!=run_queue_hds[CurrentProc]) {
 634         belch("ghuH: TSO %d (%p) [PE %d] is not at the start of the run_queue when doing a ContinueThread",
 635               CurrentTSO->id, CurrentTSO, CurrentProc);
 636         break; // run the thread anyway
 637       }
 638       /*
 639       new_event(proc, proc, CurrentTime[proc],
 640                 FindWork,
 641                 (StgTSO*)NULL, (StgClosure*)NULL, (rtsSpark*)NULL);
 642       goto next_thread;
 643       */ /* Catches superfluous CONTINUEs -- should be unnecessary */
 644       break; // now actually run the thread; DaH Qu'vam yImuHbej
 645
 646     case FetchNode:
 647       do_the_fetchnode(event);
 648       goto next_thread;             /* handle next event in event queue  */
 649
 650     case GlobalBlock:
 651       do_the_globalblock(event);
 652       goto next_thread;             /* handle next event in event queue  */
 653
 654     case FetchReply:
 655       do_the_fetchreply(event);
 656       goto next_thread;             /* handle next event in event queue  */
 657
 658     case UnblockThread:   /* Move from the blocked queue to the tail of */
 659       do_the_unblock(event);
 660       goto next_thread;             /* handle next event in event queue  */
 661
 662     case ResumeThread:  /* Move from the blocked queue to the tail of */
 663       /* the runnable queue ( i.e. Qu' SImqa'lu') */
 664       event->tso->gran.blocktime +=
 665         CurrentTime[CurrentProc] - event->tso->gran.blockedat;
 666       do_the_startthread(event);
 667       goto next_thread;             /* handle next event in event queue  */
 668
 669     case StartThread:
 670       do_the_startthread(event);
 671       goto next_thread;             /* handle next event in event queue  */
 672
 673     case MoveThread:
 674       do_the_movethread(event);
 675       goto next_thread;             /* handle next event in event queue  */
 676
 677     case MoveSpark:
 678       do_the_movespark(event);
 679       goto next_thread;             /* handle next event in event queue  */
 680
 681     case FindWork:
 682       do_the_findwork(event);
 683       goto next_thread;             /* handle next event in event queue  */
 684
 685     default:
 686       barf("Illegal event type %u\n", event->evttype);
 687     }  /* switch */
 688
 689     /* This point was scheduler_loop in the old RTS */
 690
 691     IF_DEBUG(gran, belch("GRAN: after main switch"));
 692
 693     TimeOfLastEvent = CurrentTime[CurrentProc];
 694     TimeOfNextEvent = get_time_of_next_event();
 695     IgnoreEvents=(TimeOfNextEvent==0); // HWL HACK
 696     // CurrentTSO = ThreadQueueHd;
 697
 698     IF_DEBUG(gran, belch("GRAN: time of next event is: %ld",
 699                          TimeOfNextEvent));
 700
 701     if (RtsFlags.GranFlags.Light)
 702       GranSimLight_leave_system(event, &ActiveTSO);
 703
 704     EndOfTimeSlice = CurrentTime[CurrentProc]+RtsFlags.GranFlags.time_slice;
 705
 706     IF_DEBUG(gran,
 707              belch("GRAN: end of time-slice is %#lx", EndOfTimeSlice));
 708
 709     /* in a GranSim setup the TSO stays on the run queue */
 710     t = CurrentTSO;
 711     /* Take a thread from the run queue. */
 712     t = POP_RUN_QUEUE(); // take_off_run_queue(t);
 713
 714     IF_DEBUG(gran,
 715              fprintf(stderr, "GRAN: About to run current thread, which is\n");
 716              G_TSO(t,5))
 717
 718     context_switch = 0; // turned on via GranYield, checking events and time slice
 719
 720     IF_DEBUG(gran,
 721              DumpGranEvent(GR_SCHEDULE, t));
 722
 723     procStatus[CurrentProc] = Busy;
 724
 725 #elif defined(PAR)
 726
 727     if (PendingFetches != END_BF_QUEUE) {
 728         processFetches();
 729     }
 730
 731     /* ToDo: phps merge with spark activation above */
 732     /* check whether we have local work and send requests if we have none */
 733     if (run_queue_hd == END_TSO_QUEUE) {  /* no runnable threads */
 734       /* :-[  no local threads => look out for local sparks */
 735       /* the spark pool for the current PE */
 736       pool = &(MainRegTable.rSparks); // generalise to cap = &MainRegTable
 737       if (advisory_thread_count < RtsFlags.ParFlags.maxThreads &&
 738           pool->hd < pool->tl) {
 739         /*
 740          * ToDo: add GC code check that we really have enough heap afterwards!!
 741          * Old comment:
 742          * If we're here (no runnable threads) and we have pending
 743          * sparks, we must have a space problem.  Get enough space
 744          * to turn one of those pending sparks into a
 745          * thread...
 746          */
 747
 748         spark = findSpark();                /* get a spark */
 749         if (spark != (rtsSpark) NULL) {
 750           tso = activateSpark(spark);       /* turn the spark into a thread */
 751           IF_PAR_DEBUG(schedule,
 752                        belch("==== schedule: Created TSO %d (%p); %d threads active",
 753                              tso->id, tso, advisory_thread_count));
 754
 755           if (tso==END_TSO_QUEUE) { /* failed to activate spark->back to loop */
 756             belch("==^^ failed to activate spark");
 757             goto next_thread;
 758           }               /* otherwise fall through & pick-up new tso */
 759         } else {
 760           IF_PAR_DEBUG(verbose,
 761                        belch("==^^ no local sparks (spark pool contains only NFs: %d)",
 762                              spark_queue_len(pool)));
 763           goto next_thread;
 764         }
 765       } else
 766       /* =8-[  no local sparks => look for work on other PEs */
 767       {
 768         /*
 769          * We really have absolutely no work.  Send out a fish
 770          * (there may be some out there already), and wait for
 771          * something to arrive.  We clearly can't run any threads
 772          * until a SCHEDULE or RESUME arrives, and so that's what
 773          * we're hoping to see.  (Of course, we still have to
 774          * respond to other types of messages.)
 775          */
 776         if (//!fishing &&
 777             outstandingFishes < RtsFlags.ParFlags.maxFishes ) { // &&
 778           // (last_fish_arrived_at+FISH_DELAY < CURRENT_TIME)) {
 779           /* fishing set in sendFish, processFish;
 780              avoid flooding system with fishes via delay */
 781           pe = choosePE();
 782           sendFish(pe, mytid, NEW_FISH_AGE, NEW_FISH_HISTORY,
 783                    NEW_FISH_HUNGER);
 784         }
 785
 786         processMessages();
 787         goto next_thread;
 788         // ReSchedule(0);
 789       }
 790     } else if (PacketsWaiting()) {  /* Look for incoming messages */
 791       processMessages();
 792     }
 793
 794     /* Now we are sure that we have some work available */
 795     ASSERT(run_queue_hd != END_TSO_QUEUE);
 796     /* Take a thread from the run queue, if we have work */
 797     t = POP_RUN_QUEUE();  // take_off_run_queue(END_TSO_QUEUE);
 798
 799     /* ToDo: write something to the log-file
 800     if (RTSflags.ParFlags.granSimStats && !sameThread)
 801         DumpGranEvent(GR_SCHEDULE, RunnableThreadsHd);
 802
 803     CurrentTSO = t;
 804     */
 805     /* the spark pool for the current PE */
 806     pool = &(MainRegTable.rSparks); // generalise to cap = &MainRegTable
 807
 808     IF_DEBUG(scheduler, belch("--^^ %d sparks on [%#x] (hd=%x; tl=%x; base=%x, lim=%x)",
 809                               spark_queue_len(pool),
 810                               CURRENT_PROC,
 811                               pool->hd, pool->tl, pool->base, pool->lim));
 812
 813     IF_DEBUG(scheduler, belch("--== %d threads on [%#x] (hd=%x; tl=%x)",
 814                               run_queue_len(), CURRENT_PROC,
 815                               run_queue_hd, run_queue_tl));
 816
 817 #if 0
 818     if (t != LastTSO) {
 819       /*
 820          we are running a different TSO, so write a schedule event to log file
 821          NB: If we use fair scheduling we also have to write  a deschedule
 822              event for LastTSO; with unfair scheduling we know that the
 823              previous tso has blocked whenever we switch to another tso, so
 824              we don't need it in GUM for now
 825       */
 826       DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
 827                        GR_SCHEDULE, t, (StgClosure *)NULL, 0, 0);
 828
 829     }
 830 #endif
 831 #else /* !GRAN && !PAR */
 832
 833     /* grab a thread from the run queue
 834      */
 835     ASSERT(run_queue_hd != END_TSO_QUEUE);
 836     t = POP_RUN_QUEUE();
 837     IF_DEBUG(sanity,checkTSO(t));
 838
 839 #endif
 840
 841     /* grab a capability
 842      */
 843 #ifdef SMP
 844     cap = free_capabilities;
 845     free_capabilities = cap->link;
 846     n_free_capabilities--;
 847 #else
 848     cap = &MainRegTable;
 849 #endif
 850
 851     cap->rCurrentTSO = t;
 852
 853     /* context switches are now initiated by the timer signal, unless
 854      * the user specified "context switch as often as possible", with
 855      * +RTS -C0
 856      */
 857     if (RtsFlags.ConcFlags.ctxtSwitchTicks == 0
 858         && (run_queue_hd != END_TSO_QUEUE
 859             || blocked_queue_hd != END_TSO_QUEUE))
 860         context_switch = 1;
 861     else
 862         context_switch = 0;
 863
 864     RELEASE_LOCK(&sched_mutex);
 865
 866     IF_DEBUG(scheduler, sched_belch("-->> Running TSO %ld (%p) %s ...",
 867                               t->id, t, whatNext_strs[t->what_next]));
 868
 869     /* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
 870     /* Run the current thread
 871      */
 872     switch (cap->rCurrentTSO->what_next) {
 873     case ThreadKilled:
 874     case ThreadComplete:
 875       /* Thread already finished, return to scheduler. */
 876       ret = ThreadFinished;
 877       break;
 878     case ThreadEnterGHC:
 879       ret = StgRun((StgFunPtr) stg_enterStackTop, cap);
 880       break;
 881     case ThreadRunGHC:
 882       ret = StgRun((StgFunPtr) stg_returnToStackTop, cap);
 883       break;
 884     case ThreadEnterHugs:
 885 #ifdef INTERPRETER
 886       {
 887          StgClosure* c;
 888          IF_DEBUG(scheduler,sched_belch("entering Hugs"));
 889          c = (StgClosure *)(cap->rCurrentTSO->sp[0]);
 890          cap->rCurrentTSO->sp += 1;
 891          ret = enter(cap,c);
 892          break;
 893       }
 894 #else
 895       barf("Panic: entered a BCO but no bytecode interpreter in this build");
 896 #endif
 897     default:
 898       barf("schedule: invalid what_next field");
 899     }
 900     /* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
 901
 902     /* Costs for the scheduler are assigned to CCS_SYSTEM */
 903 #ifdef PROFILING
 904     CCCS = CCS_SYSTEM;
 905 #endif
 906
 907     ACQUIRE_LOCK(&sched_mutex);
 908
 909 #ifdef SMP
 910     IF_DEBUG(scheduler,fprintf(stderr,"scheduler (task %ld): ", pthread_self()););
 911 #elif !defined(GRAN) && !defined(PAR)
 912     IF_DEBUG(scheduler,fprintf(stderr,"scheduler: "););
 913 #endif
 914     t = cap->rCurrentTSO;
 915
 916 #if defined(PAR)
 917     /* HACK 675: if the last thread didn't yield, make sure to print a
 918        SCHEDULE event to the log file when StgRunning the next thread, even
 919        if it is the same one as before */
 920     LastTSO = t; //(ret == ThreadBlocked) ? END_TSO_QUEUE : t;
 921     TimeOfLastYield = CURRENT_TIME;
 922 #endif
 923
 924     switch (ret) {
 925     case HeapOverflow:
 926       /* make all the running tasks block on a condition variable,
 927        * maybe set context_switch and wait till they all pile in,
 928        * then have them wait on a GC condition variable.
 929        */
 930       IF_DEBUG(scheduler,belch("--<< thread %ld (%p; %s) stopped: HeapOverflow",
 931                                t->id, t, whatNext_strs[t->what_next]));
 932       threadPaused(t);
 933 #if defined(GRAN)
 934       ASSERT(!is_on_queue(t,CurrentProc));
 935 #endif
 936
 937       ready_to_gc = rtsTrue;
 938       context_switch = 1;               /* stop other threads ASAP */
 939       PUSH_ON_RUN_QUEUE(t);
 940       /* actual GC is done at the end of the while loop */
 941       break;
 942
 943     case StackOverflow:
 944       IF_DEBUG(scheduler,belch("--<< thread %ld (%p; %s) stopped, StackOverflow",
 945                                t->id, t, whatNext_strs[t->what_next]));
 946       /* just adjust the stack for this thread, then pop it back
 947        * on the run queue.
 948        */
 949       threadPaused(t);
 950       {
 951         StgMainThread *m;
 952         /* enlarge the stack */
 953         StgTSO *new_t = threadStackOverflow(t);
 954
 955         /* This TSO has moved, so update any pointers to it from the
 956          * main thread stack.  It better not be on any other queues...
 957          * (it shouldn't be).
 958          */
 959         for (m = main_threads; m != NULL; m = m->link) {
 960           if (m->tso == t) {
 961             m->tso = new_t;
 962           }
 963         }
 964         threadPaused(new_t);
 965         PUSH_ON_RUN_QUEUE(new_t);
 966       }
 967       break;
 968
 969     case ThreadYielding:
 970 #if defined(GRAN)
 971       IF_DEBUG(gran,
 972                DumpGranEvent(GR_DESCHEDULE, t));
 973       globalGranStats.tot_yields++;
 974 #elif defined(PAR)
 975       IF_DEBUG(par,
 976                DumpGranEvent(GR_DESCHEDULE, t));
 977 #endif
 978       /* put the thread back on the run queue.  Then, if we're ready to
 979        * GC, check whether this is the last task to stop.  If so, wake
 980        * up the GC thread.  getThread will block during a GC until the
 981        * GC is finished.
 982        */
 983       IF_DEBUG(scheduler,
 984                if (t->what_next == ThreadEnterHugs) {
 985                    /* ToDo: or maybe a timer expired when we were in Hugs?
 986                     * or maybe someone hit ctrl-C
 987                     */
 988                    belch("--<< thread %ld (%p; %s) stopped to switch to Hugs",
 989                          t->id, t, whatNext_strs[t->what_next]);
 990                } else {
 991                    belch("--<< thread %ld (%p; %s) stopped, yielding",
 992                          t->id, t, whatNext_strs[t->what_next]);
 993                }
 994                );
 995
 996       threadPaused(t);
 997
 998       IF_DEBUG(sanity,
 999                //belch("&& Doing sanity check on yielding TSO %ld.", t->id);
1000                checkTSO(t));
1001       ASSERT(t->link == END_TSO_QUEUE);
1002 #if defined(GRAN)
1003       ASSERT(!is_on_queue(t,CurrentProc));
1004
1005       IF_DEBUG(sanity,
1006                //belch("&& Doing sanity check on all ThreadQueues (and their TSOs).");
1007                checkThreadQsSanity(rtsTrue));
1008 #endif
1009       APPEND_TO_RUN_QUEUE(t);
1010 #if defined(GRAN)
1011       /* add a ContinueThread event to actually process the thread */
1012       new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
1013                 ContinueThread,
1014                 t, (StgClosure*)NULL, (rtsSpark*)NULL);
1015       IF_GRAN_DEBUG(bq,
1016                belch("GRAN: eventq and runnableq after adding yielded thread to queue again:");
1017                G_EVENTQ(0);
1018                G_CURR_THREADQ(0))
1019 #endif /* GRAN */
1020       break;
1021
1022     case ThreadBlocked:
1023 #if defined(GRAN)
1024       IF_DEBUG(scheduler,
1025                belch("--<< thread %ld (%p; %s) stopped, blocking on node %p [PE %d] with BQ: ",
1026                                t->id, t, whatNext_strs[t->what_next], t->block_info.closure, (t->block_info.closure==(StgClosure*)NULL ? 99 : where_is(t->block_info.closure)));
1027                if (t->block_info.closure!=(StgClosure*)NULL) print_bq(t->block_info.closure));
1028
1029       // ??? needed; should emit block before
1030       IF_DEBUG(gran,
1031                DumpGranEvent(GR_DESCHEDULE, t));
1032       prune_eventq(t, (StgClosure *)NULL); // prune ContinueThreads for t
1033       /*
1034         ngoq Dogh!
1035       ASSERT(procStatus[CurrentProc]==Busy ||
1036               ((procStatus[CurrentProc]==Fetching) &&
1037               (t->block_info.closure!=(StgClosure*)NULL)));
1038       if (run_queue_hds[CurrentProc] == END_TSO_QUEUE &&
1039           !(!RtsFlags.GranFlags.DoAsyncFetch &&
1040             procStatus[CurrentProc]==Fetching))
1041         procStatus[CurrentProc] = Idle;
1042       */
1043 #elif defined(PAR)
1044       IF_DEBUG(par,
1045                DumpGranEvent(GR_DESCHEDULE, t));
1046
1047       /* Send a fetch (if BlockedOnGA) and dump event to log file */
1048       blockThread(t);
1049
1050       IF_DEBUG(scheduler,
1051                belch("--<< thread %ld (%p; %s) stopped, blocking on node %p with BQ: ",
1052                                t->id, t, whatNext_strs[t->what_next], t->block_info.closure);
1053                if (t->block_info.closure!=(StgClosure*)NULL) print_bq(t->block_info.closure));
1054
1055 #else /* !GRAN */
1056       /* don't need to do anything.  Either the thread is blocked on
1057        * I/O, in which case we'll have called addToBlockedQueue
1058        * previously, or it's blocked on an MVar or Blackhole, in which
1059        * case it'll be on the relevant queue already.
1060        */
1061       IF_DEBUG(scheduler,
1062                fprintf(stderr, "--<< thread %d (%p) stopped: ", t->id, t);
1063                printThreadBlockage(t);
1064                fprintf(stderr, "\n"));
1065
1066       /* Only for dumping event to log file
1067          ToDo: do I need this in GranSim, too?
1068       blockThread(t);
1069       */
1070 #endif
1071       threadPaused(t);
1072       break;
1073
1074     case ThreadFinished:
1075       /* Need to check whether this was a main thread, and if so, signal
1076        * the task that started it with the return value.  If we have no
1077        * more main threads, we probably need to stop all the tasks until
1078        * we get a new one.
1079        */
1080       /* We also end up here if the thread kills itself with an
1081        * uncaught exception, see Exception.hc.
1082        */
1083       IF_DEBUG(scheduler,belch("--++ thread %d (%p) finished", t->id, t));
1084 #if defined(GRAN)
1085       endThread(t, CurrentProc); // clean-up the thread
1086 #elif defined(PAR)
1087       advisory_thread_count--;
1088       if (RtsFlags.ParFlags.ParStats.Full)
1089         DumpEndEvent(CURRENT_PROC, t, rtsFalse /* not mandatory */);
1090 #endif
1091       break;
1092
1093     default:
1094       barf("schedule: invalid thread return code %d", (int)ret);
1095     }
1096
1097 #ifdef SMP
1098     cap->link = free_capabilities;
1099     free_capabilities = cap;
1100     n_free_capabilities++;
1101 #endif
1102
1103 #ifdef SMP
1104     if (ready_to_gc && n_free_capabilities == RtsFlags.ParFlags.nNodes)
1105 #else
1106     if (ready_to_gc)
1107 #endif
1108       {
1109       /* everybody back, start the GC.
1110        * Could do it in this thread, or signal a condition var
1111        * to do it in another thread.  Either way, we need to
1112        * broadcast on gc_pending_cond afterward.
1113        */
1114 #ifdef SMP
1115       IF_DEBUG(scheduler,sched_belch("doing GC"));
1116 #endif
1117       GarbageCollect(GetRoots,rtsFalse);
1118       ready_to_gc = rtsFalse;
1119 #ifdef SMP
1120       pthread_cond_broadcast(&gc_pending_cond);
1121 #endif
1122 #if defined(GRAN)
1123       /* add a ContinueThread event to continue execution of current thread */
1124       new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
1125                 ContinueThread,
1126                 t, (StgClosure*)NULL, (rtsSpark*)NULL);
1127       IF_GRAN_DEBUG(bq,
1128                fprintf(stderr, "GRAN: eventq and runnableq after Garbage collection:\n");
1129                G_EVENTQ(0);
1130                G_CURR_THREADQ(0))
1131 #endif /* GRAN */
1132     }
1133 #if defined(GRAN)
1134   next_thread:
1135     IF_GRAN_DEBUG(unused,
1136                   print_eventq(EventHd));
1137
1138     event = get_next_event();
1139
1140 #elif defined(PAR)
1141   next_thread:
1142     /* ToDo: wait for next message to arrive rather than busy wait */
1143
1144 #else /* GRAN */
1145   /* not any more
1146   next_thread:
1147     t = take_off_run_queue(END_TSO_QUEUE);
1148   */
1149 #endif /* GRAN */
1150   } /* end of while(1) */
1151 }
1152
1153 /* A hack for Hugs concurrency support.  Needs sanitisation (?) */
1154 void deleteAllThreads ( void )
1155 {
1156   StgTSO* t;
1157   IF_DEBUG(scheduler,sched_belch("deleteAllThreads()"));
1158   for (t = run_queue_hd; t != END_TSO_QUEUE; t = t->link) {
1159     deleteThread(t);
1160   }
1161   for (t = blocked_queue_hd; t != END_TSO_QUEUE; t = t->link) {
1162     deleteThread(t);
1163   }
1164   run_queue_hd = run_queue_tl = END_TSO_QUEUE;
1165   blocked_queue_hd = blocked_queue_tl = END_TSO_QUEUE;
1166 }
1167
1168 /* startThread and  insertThread are now in GranSim.c -- HWL */
1169
1170 //@node Suspend and Resume, Run queue code, Main scheduling loop, Main scheduling code
1171 //@subsection Suspend and Resume
1172
1173 /* ---------------------------------------------------------------------------
1174  * Suspending & resuming Haskell threads.
1175  *
1176  * When making a "safe" call to C (aka _ccall_GC), the task gives back
1177  * its capability before calling the C function.  This allows another
1178  * task to pick up the capability and carry on running Haskell
1179  * threads.  It also means that if the C call blocks, it won't lock
1180  * the whole system.
1181  *
1182  * The Haskell thread making the C call is put to sleep for the
1183  * duration of the call, on the susepended_ccalling_threads queue.  We
1184  * give out a token to the task, which it can use to resume the thread
1185  * on return from the C function.
1186  * ------------------------------------------------------------------------- */
1187
1188 StgInt
1189 suspendThread( Capability *cap )
1190 {
1191   nat tok;
1192
1193   ACQUIRE_LOCK(&sched_mutex);
1194
1195   IF_DEBUG(scheduler,
1196            sched_belch("thread %d did a _ccall_gc", cap->rCurrentTSO->id));
1197
1198   threadPaused(cap->rCurrentTSO);
1199   cap->rCurrentTSO->link = suspended_ccalling_threads;
1200   suspended_ccalling_threads = cap->rCurrentTSO;
1201
1202   /* Use the thread ID as the token; it should be unique */
1203   tok = cap->rCurrentTSO->id;
1204
1205 #ifdef SMP
1206   cap->link = free_capabilities;
1207   free_capabilities = cap;
1208   n_free_capabilities++;
1209 #endif
1210
1211   RELEASE_LOCK(&sched_mutex);
1212   return tok;
1213 }
1214
1215 Capability *
1216 resumeThread( StgInt tok )
1217 {
1218   StgTSO *tso, **prev;
1219   Capability *cap;
1220
1221   ACQUIRE_LOCK(&sched_mutex);
1222
1223   prev = &suspended_ccalling_threads;
1224   for (tso = suspended_ccalling_threads;
1225        tso != END_TSO_QUEUE;
1226        prev = &tso->link, tso = tso->link) {
1227     if (tso->id == (StgThreadID)tok) {
1228       *prev = tso->link;
1229       break;
1230     }
1231   }
1232   if (tso == END_TSO_QUEUE) {
1233     barf("resumeThread: thread not found");
1234   }
1235
1236 #ifdef SMP
1237   while (free_capabilities == NULL) {
1238     IF_DEBUG(scheduler, sched_belch("waiting to resume"));
1239     pthread_cond_wait(&thread_ready_cond, &sched_mutex);
1240     IF_DEBUG(scheduler, sched_belch("resuming thread %d", tso->id));
1241   }
1242   cap = free_capabilities;
1243   free_capabilities = cap->link;
1244   n_free_capabilities--;
1245 #else
1246   cap = &MainRegTable;
1247 #endif
1248
1249   cap->rCurrentTSO = tso;
1250
1251   RELEASE_LOCK(&sched_mutex);
1252   return cap;
1253 }
1254
1255
1256 /* ---------------------------------------------------------------------------
1257  * Static functions
1258  * ------------------------------------------------------------------------ */
1259 static void unblockThread(StgTSO *tso);
1260
1261 /* ---------------------------------------------------------------------------
1262  * Comparing Thread ids.
1263  *
1264  * This is used from STG land in the implementation of the
1265  * instances of Eq/Ord for ThreadIds.
1266  * ------------------------------------------------------------------------ */
1267
1268 int cmp_thread(const StgTSO *tso1, const StgTSO *tso2)
1269 {
1270   StgThreadID id1 = tso1->id;
1271   StgThreadID id2 = tso2->id;
1272
1273   if (id1 < id2) return (-1);
1274   if (id1 > id2) return 1;
1275   return 0;
1276 }
1277
1278 /* ---------------------------------------------------------------------------
1279    Create a new thread.
1280
1281    The new thread starts with the given stack size.  Before the
1282    scheduler can run, however, this thread needs to have a closure
1283    (and possibly some arguments) pushed on its stack.  See
1284    pushClosure() in Schedule.h.
1285
1286    createGenThread() and createIOThread() (in SchedAPI.h) are
1287    convenient packaged versions of this function.
1288
1289    currently pri (priority) is only used in a GRAN setup -- HWL
1290    ------------------------------------------------------------------------ */
1291 //@cindex createThread
1292 #if defined(GRAN)
1293 /*   currently pri (priority) is only used in a GRAN setup -- HWL */
1294 StgTSO *
1295 createThread(nat stack_size, StgInt pri)
1296 {
1297   return createThread_(stack_size, rtsFalse, pri);
1298 }
1299
1300 static StgTSO *
1301 createThread_(nat size, rtsBool have_lock, StgInt pri)
1302 {
1303 #else
1304 StgTSO *
1305 createThread(nat stack_size)
1306 {
1307   return createThread_(stack_size, rtsFalse);
1308 }
1309
1310 static StgTSO *
1311 createThread_(nat size, rtsBool have_lock)
1312 {
1313 #endif
1314
1315     StgTSO *tso;
1316     nat stack_size;
1317
1318     /* First check whether we should create a thread at all */
1319 #if defined(PAR)
1320   /* check that no more than RtsFlags.ParFlags.maxThreads threads are created */
1321   if (advisory_thread_count >= RtsFlags.ParFlags.maxThreads) {
1322     threadsIgnored++;
1323     belch("{createThread}Daq ghuH: refusing to create another thread; no more than %d threads allowed (currently %d)",
1324           RtsFlags.ParFlags.maxThreads, advisory_thread_count);
1325     return END_TSO_QUEUE;
1326   }
1327   threadsCreated++;
1328 #endif
1329
1330 #if defined(GRAN)
1331   ASSERT(!RtsFlags.GranFlags.Light || CurrentProc==0);
1332 #endif
1333
1334   // ToDo: check whether size = stack_size - TSO_STRUCT_SIZEW
1335
1336   /* catch ridiculously small stack sizes */
1337   if (size < MIN_STACK_WORDS + TSO_STRUCT_SIZEW) {
1338     size = MIN_STACK_WORDS + TSO_STRUCT_SIZEW;
1339   }
1340
1341   stack_size = size - TSO_STRUCT_SIZEW;
1342
1343   tso = (StgTSO *)allocate(size);
1344   TICK_ALLOC_TSO(size-TSO_STRUCT_SIZEW, 0);
1345
1346   SET_HDR(tso, &TSO_info, CCS_SYSTEM);
1347 #if defined(GRAN)
1348   SET_GRAN_HDR(tso, ThisPE);
1349 #endif
1350   tso->what_next     = ThreadEnterGHC;
1351
1352   /* tso->id needs to be unique.  For now we use a heavyweight mutex to
1353    * protect the increment operation on next_thread_id.
1354    * In future, we could use an atomic increment instead.
1355    */
1356   if (!have_lock) { ACQUIRE_LOCK(&sched_mutex); }
1357   tso->id = next_thread_id++;
1358   if (!have_lock) { RELEASE_LOCK(&sched_mutex); }
1359
1360   tso->why_blocked  = NotBlocked;
1361   tso->blocked_exceptions = NULL;
1362
1363   tso->stack_size   = stack_size;
1364   tso->max_stack_size = round_to_mblocks(RtsFlags.GcFlags.maxStkSize)
1365                               - TSO_STRUCT_SIZEW;
1366   tso->sp           = (P_)&(tso->stack) + stack_size;
1367
1368 #ifdef PROFILING
1369   tso->prof.CCCS = CCS_MAIN;
1370 #endif
1371
1372   /* put a stop frame on the stack */
1373   tso->sp -= sizeofW(StgStopFrame);
1374   SET_HDR((StgClosure*)tso->sp,(StgInfoTable *)&stg_stop_thread_info,CCS_SYSTEM);
1375   tso->su = (StgUpdateFrame*)tso->sp;
1376
1377   // ToDo: check this
1378 #if defined(GRAN)
1379   tso->link = END_TSO_QUEUE;
1380   /* uses more flexible routine in GranSim */
1381   insertThread(tso, CurrentProc);
1382 #else
1383   /* In a non-GranSim setup the pushing of a TSO onto the runq is separated
1384    * from its creation
1385    */
1386 #endif
1387
1388 #if defined(GRAN) || defined(PAR)
1389   DumpGranEvent(GR_START,tso);
1390 #endif
1391
1392   /* Link the new thread on the global thread list.
1393    */
1394   tso->global_link = all_threads;
1395   all_threads = tso;
1396
1397 #if defined(GRAN)
1398   tso->gran.pri = pri;
1399 # if defined(DEBUG)
1400   tso->gran.magic = TSO_MAGIC; // debugging only
1401 # endif
1402   tso->gran.sparkname   = 0;
1403   tso->gran.startedat   = CURRENT_TIME;
1404   tso->gran.exported    = 0;
1405   tso->gran.basicblocks = 0;
1406   tso->gran.allocs      = 0;
1407   tso->gran.exectime    = 0;
1408   tso->gran.fetchtime   = 0;
1409   tso->gran.fetchcount  = 0;
1410   tso->gran.blocktime   = 0;
1411   tso->gran.blockcount  = 0;
1412   tso->gran.blockedat   = 0;
1413   tso->gran.globalsparks = 0;
1414   tso->gran.localsparks  = 0;
1415   if (RtsFlags.GranFlags.Light)
1416     tso->gran.clock  = Now; /* local clock */
1417   else
1418     tso->gran.clock  = 0;
1419
1420   IF_DEBUG(gran,printTSO(tso));
1421 #elif defined(PAR)
1422 # if defined(DEBUG)
1423   tso->par.magic = TSO_MAGIC; // debugging only
1424 # endif
1425   tso->par.sparkname   = 0;
1426   tso->par.startedat   = CURRENT_TIME;
1427   tso->par.exported    = 0;
1428   tso->par.basicblocks = 0;
1429   tso->par.allocs      = 0;
1430   tso->par.exectime    = 0;
1431   tso->par.fetchtime   = 0;
1432   tso->par.fetchcount  = 0;
1433   tso->par.blocktime   = 0;
1434   tso->par.blockcount  = 0;
1435   tso->par.blockedat   = 0;
1436   tso->par.globalsparks = 0;
1437   tso->par.localsparks  = 0;
1438 #endif
1439
1440 #if defined(GRAN)
1441   globalGranStats.tot_threads_created++;
1442   globalGranStats.threads_created_on_PE[CurrentProc]++;
1443   globalGranStats.tot_sq_len += spark_queue_len(CurrentProc);
1444   globalGranStats.tot_sq_probes++;
1445 #endif
1446
1447 #if defined(GRAN)
1448   IF_GRAN_DEBUG(pri,
1449                 belch("==__ schedule: Created TSO %d (%p);",
1450                       CurrentProc, tso, tso->id));
1451 #elif defined(PAR)
1452     IF_PAR_DEBUG(verbose,
1453                  belch("==__ schedule: Created TSO %d (%p); %d threads active",
1454                        tso->id, tso, advisory_thread_count));
1455 #else
1456   IF_DEBUG(scheduler,sched_belch("created thread %ld, stack size = %lx words",
1457                                  tso->id, tso->stack_size));
1458 #endif
1459   return tso;
1460 }
1461
1462 /*
1463   Turn a spark into a thread.
1464   ToDo: fix for SMP (needs to acquire SCHED_MUTEX!)
1465 */
1466 #if defined(PAR)
1467 //@cindex activateSpark
1468 StgTSO *
1469 activateSpark (rtsSpark spark)
1470 {
1471   StgTSO *tso;
1472
1473   ASSERT(spark != (rtsSpark)NULL);
1474   tso = createThread_(RtsFlags.GcFlags.initialStkSize, rtsTrue);
1475   if (tso!=END_TSO_QUEUE) {
1476     pushClosure(tso,spark);
1477     PUSH_ON_RUN_QUEUE(tso);
1478     advisory_thread_count++;
1479
1480     if (RtsFlags.ParFlags.ParStats.Full) {
1481       //ASSERT(run_queue_hd == END_TSO_QUEUE); // I think ...
1482       IF_PAR_DEBUG(verbose,
1483                    belch("==^^ activateSpark: turning spark of closure %p (%s) into a thread",
1484                          (StgClosure *)spark, info_type((StgClosure *)spark)));
1485     }
1486   } else {
1487     barf("activateSpark: Cannot create TSO");
1488   }
1489   // ToDo: fwd info on local/global spark to thread -- HWL
1490   // tso->gran.exported =  spark->exported;
1491   // tso->gran.locked =   !spark->global;
1492   // tso->gran.sparkname = spark->name;
1493
1494   return tso;
1495 }
1496 #endif
1497
1498 /* ---------------------------------------------------------------------------
1499  * scheduleThread()
1500  *
1501  * scheduleThread puts a thread on the head of the runnable queue.
1502  * This will usually be done immediately after a thread is created.
1503  * The caller of scheduleThread must create the thread using e.g.
1504  * createThread and push an appropriate closure
1505  * on this thread's stack before the scheduler is invoked.
1506  * ------------------------------------------------------------------------ */
1507
1508 void
1509 scheduleThread(StgTSO *tso)
1510 {
1511   if (tso==END_TSO_QUEUE){
1512     schedule();
1513     return;
1514   }
1515
1516   ACQUIRE_LOCK(&sched_mutex);
1517
1518   /* Put the new thread on the head of the runnable queue.  The caller
1519    * better push an appropriate closure on this thread's stack
1520    * beforehand.  In the SMP case, the thread may start running as
1521    * soon as we release the scheduler lock below.
1522    */
1523   PUSH_ON_RUN_QUEUE(tso);
1524   THREAD_RUNNABLE();
1525
1526 #if 0
1527   IF_DEBUG(scheduler,printTSO(tso));
1528 #endif
1529   RELEASE_LOCK(&sched_mutex);
1530 }
1531
1532 /* ---------------------------------------------------------------------------
1533  * startTasks()
1534  *
1535  * Start up Posix threads to run each of the scheduler tasks.
1536  * I believe the task ids are not needed in the system as defined.
1537  *  KH @ 25/10/99
1538  * ------------------------------------------------------------------------ */
1539
1540 #if defined(PAR) || defined(SMP)
1541 void *
1542 taskStart( void *arg STG_UNUSED )
1543 {
1544   rts_evalNothing(NULL);
1545 }
1546 #endif
1547
1548 /* ---------------------------------------------------------------------------
1549  * initScheduler()
1550  *
1551  * Initialise the scheduler.  This resets all the queues - if the
1552  * queues contained any threads, they'll be garbage collected at the
1553  * next pass.
1554  *
1555  * This now calls startTasks(), so should only be called once!  KH @ 25/10/99
1556  * ------------------------------------------------------------------------ */
1557
1558 #ifdef SMP
1559 static void
1560 term_handler(int sig STG_UNUSED)
1561 {
1562   stat_workerStop();
1563   ACQUIRE_LOCK(&term_mutex);
1564   await_death--;
1565   RELEASE_LOCK(&term_mutex);
1566   pthread_exit(NULL);
1567 }
1568 #endif
1569
1570 //@cindex initScheduler
1571 void
1572 initScheduler(void)
1573 {
1574 #if defined(GRAN)
1575   nat i;
1576
1577   for (i=0; i<=MAX_PROC; i++) {
1578     run_queue_hds[i]      = END_TSO_QUEUE;
1579     run_queue_tls[i]      = END_TSO_QUEUE;
1580     blocked_queue_hds[i]  = END_TSO_QUEUE;
1581     blocked_queue_tls[i]  = END_TSO_QUEUE;
1582     ccalling_threadss[i]  = END_TSO_QUEUE;
1583   }
1584 #else
1585   run_queue_hd      = END_TSO_QUEUE;
1586   run_queue_tl      = END_TSO_QUEUE;
1587   blocked_queue_hd  = END_TSO_QUEUE;
1588   blocked_queue_tl  = END_TSO_QUEUE;
1589 #endif
1590
1591   suspended_ccalling_threads  = END_TSO_QUEUE;
1592
1593   main_threads = NULL;
1594   all_threads  = END_TSO_QUEUE;
1595
1596   context_switch = 0;
1597   interrupted    = 0;
1598
1599   RtsFlags.ConcFlags.ctxtSwitchTicks =
1600       RtsFlags.ConcFlags.ctxtSwitchTime / TICK_MILLISECS;
1601
1602 #ifdef INTERPRETER
1603   ecafList = END_ECAF_LIST;
1604   clearECafTable();
1605 #endif
1606
1607   /* Install the SIGHUP handler */
1608 #ifdef SMP
1609   {
1610     struct sigaction action,oact;
1611
1612     action.sa_handler = term_handler;
1613     sigemptyset(&action.sa_mask);
1614     action.sa_flags = 0;
1615     if (sigaction(SIGTERM, &action, &oact) != 0) {
1616       barf("can't install TERM handler");
1617     }
1618   }
1619 #endif
1620
1621 #ifdef SMP
1622   /* Allocate N Capabilities */
1623   {
1624     nat i;
1625     Capability *cap, *prev;
1626     cap  = NULL;
1627     prev = NULL;
1628     for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
1629       cap = stgMallocBytes(sizeof(Capability), "initScheduler:capabilities");
1630       cap->link = prev;
1631       prev = cap;
1632     }
1633     free_capabilities = cap;
1634     n_free_capabilities = RtsFlags.ParFlags.nNodes;
1635   }
1636   IF_DEBUG(scheduler,fprintf(stderr,"scheduler: Allocated %d capabilities\n",
1637                              n_free_capabilities););
1638 #endif
1639
1640 #if defined(SMP) || defined(PAR)
1641   initSparkPools();
1642 #endif
1643 }
1644
1645 #ifdef SMP
1646 void
1647 startTasks( void )
1648 {
1649   nat i;
1650   int r;
1651   pthread_t tid;
1652
1653   /* make some space for saving all the thread ids */
1654   task_ids = stgMallocBytes(RtsFlags.ParFlags.nNodes * sizeof(task_info),
1655                             "initScheduler:task_ids");
1656
1657   /* and create all the threads */
1658   for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
1659     r = pthread_create(&tid,NULL,taskStart,NULL);
1660     if (r != 0) {
1661       barf("startTasks: Can't create new Posix thread");
1662     }
1663     task_ids[i].id = tid;
1664     task_ids[i].mut_time = 0.0;
1665     task_ids[i].mut_etime = 0.0;
1666     task_ids[i].gc_time = 0.0;
1667     task_ids[i].gc_etime = 0.0;
1668     task_ids[i].elapsedtimestart = elapsedtime();
1669     IF_DEBUG(scheduler,fprintf(stderr,"scheduler: Started task: %ld\n",tid););
1670   }
1671 }
1672 #endif
1673
1674 void
1675 exitScheduler( void )
1676 {
1677 #ifdef SMP
1678   nat i;
1679
1680   /* Don't want to use pthread_cancel, since we'd have to install
1681    * these silly exception handlers (pthread_cleanup_{push,pop}) around
1682    * all our locks.
1683    */
1684 #if 0
1685   /* Cancel all our tasks */
1686   for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
1687     pthread_cancel(task_ids[i].id);
1688   }
1689
1690   /* Wait for all the tasks to terminate */
1691   for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
1692     IF_DEBUG(scheduler,fprintf(stderr,"scheduler: waiting for task %ld\n",
1693                                task_ids[i].id));
1694     pthread_join(task_ids[i].id, NULL);
1695   }
1696 #endif
1697
1698   /* Send 'em all a SIGHUP.  That should shut 'em up.
1699    */
1700   await_death = RtsFlags.ParFlags.nNodes;
1701   for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
1702     pthread_kill(task_ids[i].id,SIGTERM);
1703   }
1704   while (await_death > 0) {
1705     sched_yield();
1706   }
1707 #endif
1708 }
1709
1710 /* -----------------------------------------------------------------------------
1711    Managing the per-task allocation areas.
1712
1713    Each capability comes with an allocation area.  These are
1714    fixed-length block lists into which allocation can be done.
1715
1716    ToDo: no support for two-space collection at the moment???
1717    -------------------------------------------------------------------------- */
1718
1719 /* -----------------------------------------------------------------------------
1720  * waitThread is the external interface for running a new computation
1721  * and waiting for the result.
1722  *
1723  * In the non-SMP case, we create a new main thread, push it on the
1724  * main-thread stack, and invoke the scheduler to run it.  The
1725  * scheduler will return when the top main thread on the stack has
1726  * completed or died, and fill in the necessary fields of the
1727  * main_thread structure.
1728  *
1729  * In the SMP case, we create a main thread as before, but we then
1730  * create a new condition variable and sleep on it.  When our new
1731  * main thread has completed, we'll be woken up and the status/result
1732  * will be in the main_thread struct.
1733  * -------------------------------------------------------------------------- */
1734
1735 int
1736 howManyThreadsAvail ( void )
1737 {
1738    int i = 0;
1739    StgTSO* q;
1740    for (q = run_queue_hd; q != END_TSO_QUEUE; q = q->link)
1741       i++;
1742    for (q = blocked_queue_hd; q != END_TSO_QUEUE; q = q->link)
1743       i++;
1744    return i;
1745 }
1746
1747 void
1748 finishAllThreads ( void )
1749 {
1750    do {
1751       while (run_queue_hd != END_TSO_QUEUE) {
1752          waitThread ( run_queue_hd, NULL );
1753       }
1754       while (blocked_queue_hd != END_TSO_QUEUE) {
1755          waitThread ( blocked_queue_hd, NULL );
1756       }
1757    } while
1758       (blocked_queue_hd != END_TSO_QUEUE ||
1759         run_queue_hd != END_TSO_QUEUE);
1760 }
1761
1762 SchedulerStatus
1763 waitThread(StgTSO *tso, /*out*/StgClosure **ret)
1764 {
1765   StgMainThread *m;
1766   SchedulerStatus stat;
1767
1768   ACQUIRE_LOCK(&sched_mutex);
1769
1770   m = stgMallocBytes(sizeof(StgMainThread), "waitThread");
1771
1772   m->tso = tso;
1773   m->ret = ret;
1774   m->stat = NoStatus;
1775 #ifdef SMP
1776   pthread_cond_init(&m->wakeup, NULL);
1777 #endif
1778
1779   m->link = main_threads;
1780   main_threads = m;
1781
1782   IF_DEBUG(scheduler, fprintf(stderr, "scheduler: new main thread (%d)\n",
1783                               m->tso->id));
1784
1785 #ifdef SMP
1786   do {
1787     pthread_cond_wait(&m->wakeup, &sched_mutex);
1788   } while (m->stat == NoStatus);
1789 #elif defined(GRAN)
1790   /* GranSim specific init */
1791   CurrentTSO = m->tso;                // the TSO to run
1792   procStatus[MainProc] = Busy;        // status of main PE
1793   CurrentProc = MainProc;             // PE to run it on
1794
1795   schedule();
1796 #else
1797   schedule();
1798   ASSERT(m->stat != NoStatus);
1799 #endif
1800
1801   stat = m->stat;
1802
1803 #ifdef SMP
1804   pthread_cond_destroy(&m->wakeup);
1805 #endif
1806
1807   IF_DEBUG(scheduler, fprintf(stderr, "scheduler: main thread (%d) finished\n",
1808                               m->tso->id));
1809   free(m);
1810
1811   RELEASE_LOCK(&sched_mutex);
1812
1813   return stat;
1814 }
1815
1816 //@node Run queue code, Garbage Collextion Routines, Suspend and Resume, Main scheduling code
1817 //@subsection Run queue code
1818
1819 #if 0
1820 /*
1821    NB: In GranSim we have many run queues; run_queue_hd is actually a macro
1822        unfolding to run_queue_hds[CurrentProc], thus CurrentProc is an
1823        implicit global variable that has to be correct when calling these
1824        fcts -- HWL
1825 */
1826
1827 /* Put the new thread on the head of the runnable queue.
1828  * The caller of createThread better push an appropriate closure
1829  * on this thread's stack before the scheduler is invoked.
1830  */
1831 static /* inline */ void
1832 add_to_run_queue(tso)
1833 StgTSO* tso;
1834 {
1835   ASSERT(tso!=run_queue_hd && tso!=run_queue_tl);
1836   tso->link = run_queue_hd;
1837   run_queue_hd = tso;
1838   if (run_queue_tl == END_TSO_QUEUE) {
1839     run_queue_tl = tso;
1840   }
1841 }
1842
1843 /* Put the new thread at the end of the runnable queue. */
1844 static /* inline */ void
1845 push_on_run_queue(tso)
1846 StgTSO* tso;
1847 {
1848   ASSERT(get_itbl((StgClosure *)tso)->type == TSO);
1849   ASSERT(run_queue_hd!=NULL && run_queue_tl!=NULL);
1850   ASSERT(tso!=run_queue_hd && tso!=run_queue_tl);
1851   if (run_queue_hd == END_TSO_QUEUE) {
1852     run_queue_hd = tso;
1853   } else {
1854     run_queue_tl->link = tso;
1855   }
1856   run_queue_tl = tso;
1857 }
1858
1859 /*
1860    Should be inlined because it's used very often in schedule.  The tso
1861    argument is actually only needed in GranSim, where we want to have the
1862    possibility to schedule *any* TSO on the run queue, irrespective of the
1863    actual ordering. Therefore, if tso is not the nil TSO then we traverse
1864    the run queue and dequeue the tso, adjusting the links in the queue.
1865 */
1866 //@cindex take_off_run_queue
1867 static /* inline */ StgTSO*
1868 take_off_run_queue(StgTSO *tso) {
1869   StgTSO *t, *prev;
1870
1871   /*
1872      qetlaHbogh Qu' ngaSbogh ghomDaQ {tso} yIteq!
1873
1874      if tso is specified, unlink that tso from the run_queue (doesn't have
1875      to be at the beginning of the queue); GranSim only
1876   */
1877   if (tso!=END_TSO_QUEUE) {
1878     /* find tso in queue */
1879     for (t=run_queue_hd, prev=END_TSO_QUEUE;
1880          t!=END_TSO_QUEUE && t!=tso;
1881          prev=t, t=t->link)
1882       /* nothing */ ;
1883     ASSERT(t==tso);
1884     /* now actually dequeue the tso */
1885     if (prev!=END_TSO_QUEUE) {
1886       ASSERT(run_queue_hd!=t);
1887       prev->link = t->link;
1888     } else {
1889       /* t is at beginning of thread queue */
1890       ASSERT(run_queue_hd==t);
1891       run_queue_hd = t->link;
1892     }
1893     /* t is at end of thread queue */
1894     if (t->link==END_TSO_QUEUE) {
1895       ASSERT(t==run_queue_tl);
1896       run_queue_tl = prev;
1897     } else {
1898       ASSERT(run_queue_tl!=t);
1899     }
1900     t->link = END_TSO_QUEUE;
1901   } else {
1902     /* take tso from the beginning of the queue; std concurrent code */
1903     t = run_queue_hd;
1904     if (t != END_TSO_QUEUE) {
1905       run_queue_hd = t->link;
1906       t->link = END_TSO_QUEUE;
1907       if (run_queue_hd == END_TSO_QUEUE) {
1908         run_queue_tl = END_TSO_QUEUE;
1909       }
1910     }
1911   }
1912   return t;
1913 }
1914
1915 #endif /* 0 */
1916
1917 //@node Garbage Collextion Routines, Blocking Queue Routines, Run queue code, Main scheduling code
1918 //@subsection Garbage Collextion Routines
1919
1920 /* ---------------------------------------------------------------------------
1921    Where are the roots that we know about?
1922
1923         - all the threads on the runnable queue
1924         - all the threads on the blocked queue
1925         - all the thread currently executing a _ccall_GC
1926         - all the "main threads"
1927
1928    ------------------------------------------------------------------------ */
1929
1930 /* This has to be protected either by the scheduler monitor, or by the
1931         garbage collection monitor (probably the latter).
1932         KH @ 25/10/99
1933 */
1934
1935 static void GetRoots(void)
1936 {
1937   StgMainThread *m;
1938
1939 #if defined(GRAN)
1940   {
1941     nat i;
1942     for (i=0; i<=RtsFlags.GranFlags.proc; i++) {
1943       if ((run_queue_hds[i] != END_TSO_QUEUE) && ((run_queue_hds[i] != NULL)))
1944         run_queue_hds[i]    = (StgTSO *)MarkRoot((StgClosure *)run_queue_hds[i]);
1945       if ((run_queue_tls[i] != END_TSO_QUEUE) && ((run_queue_tls[i] != NULL)))
1946         run_queue_tls[i]    = (StgTSO *)MarkRoot((StgClosure *)run_queue_tls[i]);
1947
1948       if ((blocked_queue_hds[i] != END_TSO_QUEUE) && ((blocked_queue_hds[i] != NULL)))
1949         blocked_queue_hds[i] = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_hds[i]);
1950       if ((blocked_queue_tls[i] != END_TSO_QUEUE) && ((blocked_queue_tls[i] != NULL)))
1951         blocked_queue_tls[i] = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_tls[i]);
1952       if ((ccalling_threadss[i] != END_TSO_QUEUE) && ((ccalling_threadss[i] != NULL)))
1953         ccalling_threadss[i] = (StgTSO *)MarkRoot((StgClosure *)ccalling_threadss[i]);
1954     }
1955   }
1956
1957   markEventQueue();
1958
1959 #else /* !GRAN */
1960   if (run_queue_hd != END_TSO_QUEUE) {
1961     ASSERT(run_queue_tl != END_TSO_QUEUE);
1962     run_queue_hd      = (StgTSO *)MarkRoot((StgClosure *)run_queue_hd);
1963     run_queue_tl      = (StgTSO *)MarkRoot((StgClosure *)run_queue_tl);
1964   }
1965
1966   if (blocked_queue_hd != END_TSO_QUEUE) {
1967     ASSERT(blocked_queue_tl != END_TSO_QUEUE);
1968     blocked_queue_hd  = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_hd);
1969     blocked_queue_tl  = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_tl);
1970   }
1971 #endif
1972
1973   for (m = main_threads; m != NULL; m = m->link) {
1974     m->tso = (StgTSO *)MarkRoot((StgClosure *)m->tso);
1975   }
1976   if (suspended_ccalling_threads != END_TSO_QUEUE)
1977     suspended_ccalling_threads =
1978       (StgTSO *)MarkRoot((StgClosure *)suspended_ccalling_threads);
1979
1980 #if defined(SMP) || defined(PAR) || defined(GRAN)
1981   markSparkQueue();
1982 #endif
1983 }
1984
1985 /* -----------------------------------------------------------------------------
1986    performGC
1987
1988    This is the interface to the garbage collector from Haskell land.
1989    We provide this so that external C code can allocate and garbage
1990    collect when called from Haskell via _ccall_GC.
1991
1992    It might be useful to provide an interface whereby the programmer
1993    can specify more roots (ToDo).
1994
1995    This needs to be protected by the GC condition variable above.  KH.
1996    -------------------------------------------------------------------------- */
1997
1998 void (*extra_roots)(void);
1999
2000 void
2001 performGC(void)
2002 {
2003   GarbageCollect(GetRoots,rtsFalse);
2004 }
2005
2006 void
2007 performMajorGC(void)
2008 {
2009   GarbageCollect(GetRoots,rtsTrue);
2010 }
2011
2012 static void
2013 AllRoots(void)
2014 {
2015   GetRoots();                   /* the scheduler's roots */
2016   extra_roots();                /* the user's roots */
2017 }
2018
2019 void
2020 performGCWithRoots(void (*get_roots)(void))
2021 {
2022   extra_roots = get_roots;
2023
2024   GarbageCollect(AllRoots,rtsFalse);
2025 }
2026
2027 /* -----------------------------------------------------------------------------
2028    Stack overflow
2029
2030    If the thread has reached its maximum stack size, then raise the
2031    StackOverflow exception in the offending thread.  Otherwise
2032    relocate the TSO into a larger chunk of memory and adjust its stack
2033    size appropriately.
2034    -------------------------------------------------------------------------- */
2035
2036 static StgTSO *
2037 threadStackOverflow(StgTSO *tso)
2038 {
2039   nat new_stack_size, new_tso_size, diff, stack_words;
2040   StgPtr new_sp;
2041   StgTSO *dest;
2042
2043   IF_DEBUG(sanity,checkTSO(tso));
2044   if (tso->stack_size >= tso->max_stack_size) {
2045
2046     IF_DEBUG(gc,
2047              belch("@@ threadStackOverflow of TSO %d (%p): stack too large (now %ld; max is %ld",
2048                    tso->id, tso, tso->stack_size, tso->max_stack_size);
2049              /* If we're debugging, just print out the top of the stack */
2050              printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size,
2051                                               tso->sp+64)));
2052
2053 #ifdef INTERPRETER
2054     fprintf(stderr, "fatal: stack overflow in Hugs; aborting\n" );
2055     exit(1);
2056 #else
2057     /* Send this thread the StackOverflow exception */
2058     raiseAsync(tso, (StgClosure *)stackOverflow_closure);
2059 #endif
2060     return tso;
2061   }
2062
2063   /* Try to double the current stack size.  If that takes us over the
2064    * maximum stack size for this thread, then use the maximum instead.
2065    * Finally round up so the TSO ends up as a whole number of blocks.
2066    */
2067   new_stack_size = stg_min(tso->stack_size * 2, tso->max_stack_size);
2068   new_tso_size   = (nat)BLOCK_ROUND_UP(new_stack_size * sizeof(W_) +
2069                                        TSO_STRUCT_SIZE)/sizeof(W_);
2070   new_tso_size = round_to_mblocks(new_tso_size);  /* Be MBLOCK-friendly */
2071   new_stack_size = new_tso_size - TSO_STRUCT_SIZEW;
2072
2073   IF_DEBUG(scheduler, fprintf(stderr,"scheduler: increasing stack size from %d words to %d.\n", tso->stack_size, new_stack_size));
2074
2075   dest = (StgTSO *)allocate(new_tso_size);
2076   TICK_ALLOC_TSO(new_tso_size-sizeofW(StgTSO),0);
2077
2078   /* copy the TSO block and the old stack into the new area */
2079   memcpy(dest,tso,TSO_STRUCT_SIZE);
2080   stack_words = tso->stack + tso->stack_size - tso->sp;
2081   new_sp = (P_)dest + new_tso_size - stack_words;
2082   memcpy(new_sp, tso->sp, stack_words * sizeof(W_));
2083
2084   /* relocate the stack pointers... */
2085   diff = (P_)new_sp - (P_)tso->sp; /* In *words* */
2086   dest->su    = (StgUpdateFrame *) ((P_)dest->su + diff);
2087   dest->sp    = new_sp;
2088   dest->stack_size = new_stack_size;
2089
2090   /* and relocate the update frame list */
2091   relocate_TSO(tso, dest);
2092
2093   /* Mark the old TSO as relocated.  We have to check for relocated
2094    * TSOs in the garbage collector and any primops that deal with TSOs.
2095    *
2096    * It's important to set the sp and su values to just beyond the end
2097    * of the stack, so we don't attempt to scavenge any part of the
2098    * dead TSO's stack.
2099    */
2100   tso->what_next = ThreadRelocated;
2101   tso->link = dest;
2102   tso->sp = (P_)&(tso->stack[tso->stack_size]);
2103   tso->su = (StgUpdateFrame *)tso->sp;
2104   tso->why_blocked = NotBlocked;
2105   dest->mut_link = NULL;
2106
2107   IF_PAR_DEBUG(verbose,
2108                belch("@@ threadStackOverflow of TSO %d (now at %p): stack size increased to %ld",
2109                      tso->id, tso, tso->stack_size);
2110                /* If we're debugging, just print out the top of the stack */
2111                printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size,
2112                                                 tso->sp+64)));
2113
2114   IF_DEBUG(sanity,checkTSO(tso));
2115 #if 0
2116   IF_DEBUG(scheduler,printTSO(dest));
2117 #endif
2118
2119   return dest;
2120 }
2121
2122 //@node Blocking Queue Routines, Exception Handling Routines, Garbage Collextion Routines, Main scheduling code
2123 //@subsection Blocking Queue Routines
2124
2125 /* ---------------------------------------------------------------------------
2126    Wake up a queue that was blocked on some resource.
2127    ------------------------------------------------------------------------ */
2128
2129 /* ToDo: check push_on_run_queue vs. PUSH_ON_RUN_QUEUE */
2130
2131 #if defined(GRAN)
2132 static inline void
2133 unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node )
2134 {
2135 }
2136 #elif defined(PAR)
2137 static inline void
2138 unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node )
2139 {
2140   /* write RESUME events to log file and
2141      update blocked and fetch time (depending on type of the orig closure) */
2142   if (RtsFlags.ParFlags.ParStats.Full) {
2143     DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
2144                      GR_RESUME, ((StgTSO *)bqe), ((StgTSO *)bqe)->block_info.closure,
2145                      0, 0 /* spark_queue_len(ADVISORY_POOL) */);
2146
2147     switch (get_itbl(node)->type) {
2148         case FETCH_ME_BQ:
2149           ((StgTSO *)bqe)->par.fetchtime += CURRENT_TIME-((StgTSO *)bqe)->par.blockedat;
2150           break;
2151         case RBH:
2152         case FETCH_ME:
2153         case BLACKHOLE_BQ:
2154           ((StgTSO *)bqe)->par.blocktime += CURRENT_TIME-((StgTSO *)bqe)->par.blockedat;
2155           break;
2156         default:
2157           barf("{unblockOneLocked}Daq Qagh: unexpected closure in blocking queue");
2158         }
2159       }
2160 }
2161 #endif
2162
2163 #if defined(GRAN)
2164 static StgBlockingQueueElement *
2165 unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node)
2166 {
2167     StgTSO *tso;
2168     PEs node_loc, tso_loc;
2169
2170     node_loc = where_is(node); // should be lifted out of loop
2171     tso = (StgTSO *)bqe;  // wastes an assignment to get the type right
2172     tso_loc = where_is((StgClosure *)tso);
2173     if (IS_LOCAL_TO(PROCS(node),tso_loc)) { // TSO is local
2174       /* !fake_fetch => TSO is on CurrentProc is same as IS_LOCAL_TO */
2175       ASSERT(CurrentProc!=node_loc || tso_loc==CurrentProc);
2176       CurrentTime[CurrentProc] += RtsFlags.GranFlags.Costs.lunblocktime;
2177       // insertThread(tso, node_loc);
2178       new_event(tso_loc, tso_loc, CurrentTime[CurrentProc],
2179                 ResumeThread,
2180                 tso, node, (rtsSpark*)NULL);
2181       tso->link = END_TSO_QUEUE; // overwrite link just to be sure
2182       // len_local++;
2183       // len++;
2184     } else { // TSO is remote (actually should be FMBQ)
2185       CurrentTime[CurrentProc] += RtsFlags.GranFlags.Costs.mpacktime +
2186                                   RtsFlags.GranFlags.Costs.gunblocktime +
2187                                   RtsFlags.GranFlags.Costs.latency;
2188       new_event(tso_loc, CurrentProc, CurrentTime[CurrentProc],
2189                 UnblockThread,
2190                 tso, node, (rtsSpark*)NULL);
2191       tso->link = END_TSO_QUEUE; // overwrite link just to be sure
2192       // len++;
2193     }
2194     /* the thread-queue-overhead is accounted for in either Resume or UnblockThread */
2195     IF_GRAN_DEBUG(bq,
2196                   fprintf(stderr," %s TSO %d (%p) [PE %d] (block_info.closure=%p) (next=%p) ,",
2197                           (node_loc==tso_loc ? "Local" : "Global"),
2198                           tso->id, tso, CurrentProc, tso->block_info.closure, tso->link));
2199     tso->block_info.closure = NULL;
2200     IF_DEBUG(scheduler,belch("-- Waking up thread %ld (%p)",
2201                              tso->id, tso));
2202 }
2203 #elif defined(PAR)
2204 static StgBlockingQueueElement *
2205 unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node)
2206 {
2207     StgBlockingQueueElement *next;
2208
2209     switch (get_itbl(bqe)->type) {
2210     case TSO:
2211       ASSERT(((StgTSO *)bqe)->why_blocked != NotBlocked);
2212       /* if it's a TSO just push it onto the run_queue */
2213       next = bqe->link;
2214       // ((StgTSO *)bqe)->link = END_TSO_QUEUE; // debugging?
2215       PUSH_ON_RUN_QUEUE((StgTSO *)bqe);
2216       THREAD_RUNNABLE();
2217       unblockCount(bqe, node);
2218       /* reset blocking status after dumping event */
2219       ((StgTSO *)bqe)->why_blocked = NotBlocked;
2220       break;
2221
2222     case BLOCKED_FETCH:
2223       /* if it's a BLOCKED_FETCH put it on the PendingFetches list */
2224       next = bqe->link;
2225       bqe->link = PendingFetches;
2226       PendingFetches = bqe;
2227       break;
2228
2229 # if defined(DEBUG)
2230       /* can ignore this case in a non-debugging setup;
2231          see comments on RBHSave closures above */
2232     case CONSTR:
2233       /* check that the closure is an RBHSave closure */
2234       ASSERT(get_itbl((StgClosure *)bqe) == &RBH_Save_0_info ||
2235              get_itbl((StgClosure *)bqe) == &RBH_Save_1_info ||
2236              get_itbl((StgClosure *)bqe) == &RBH_Save_2_info);
2237       break;
2238
2239     default:
2240       barf("{unblockOneLocked}Daq Qagh: Unexpected IP (%#lx; %s) in blocking queue at %#lx\n",
2241            get_itbl((StgClosure *)bqe), info_type((StgClosure *)bqe),
2242            (StgClosure *)bqe);
2243 # endif
2244     }
2245   // IF_DEBUG(scheduler,sched_belch("waking up thread %ld", tso->id));
2246   return next;
2247 }
2248
2249 #else /* !GRAN && !PAR */
2250 static StgTSO *
2251 unblockOneLocked(StgTSO *tso)
2252 {
2253   StgTSO *next;
2254
2255   ASSERT(get_itbl(tso)->type == TSO);
2256   ASSERT(tso->why_blocked != NotBlocked);
2257   tso->why_blocked = NotBlocked;
2258   next = tso->link;
2259   PUSH_ON_RUN_QUEUE(tso);
2260   THREAD_RUNNABLE();
2261   IF_DEBUG(scheduler,sched_belch("waking up thread %ld", tso->id));
2262   return next;
2263 }
2264 #endif
2265
2266 #if defined(GRAN) || defined(PAR)
2267 inline StgBlockingQueueElement *
2268 unblockOne(StgBlockingQueueElement *bqe, StgClosure *node)
2269 {
2270   ACQUIRE_LOCK(&sched_mutex);
2271   bqe = unblockOneLocked(bqe, node);
2272   RELEASE_LOCK(&sched_mutex);
2273   return bqe;
2274 }
2275 #else
2276 inline StgTSO *
2277 unblockOne(StgTSO *tso)
2278 {
2279   ACQUIRE_LOCK(&sched_mutex);
2280   tso = unblockOneLocked(tso);
2281   RELEASE_LOCK(&sched_mutex);
2282   return tso;
2283 }
2284 #endif
2285
2286 #if defined(GRAN)
2287 void
2288 awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node)
2289 {
2290   StgBlockingQueueElement *bqe;
2291   PEs node_loc;
2292   nat len = 0;
2293
2294   IF_GRAN_DEBUG(bq,
2295                 belch("## AwBQ for node %p on PE %d @ %ld by TSO %d (%p): ", \
2296                       node, CurrentProc, CurrentTime[CurrentProc],
2297                       CurrentTSO->id, CurrentTSO));
2298
2299   node_loc = where_is(node);
2300
2301   ASSERT(get_itbl(q)->type == TSO ||   // q is either a TSO or an RBHSave
2302          get_itbl(q)->type == CONSTR); // closure (type constructor)
2303   ASSERT(is_unique(node));
2304
2305   /* FAKE FETCH: magically copy the node to the tso's proc;
2306      no Fetch necessary because in reality the node should not have been
2307      moved to the other PE in the first place
2308   */
2309   if (CurrentProc!=node_loc) {
2310     IF_GRAN_DEBUG(bq,
2311                   belch("## node %p is on PE %d but CurrentProc is %d (TSO %d); assuming fake fetch and adjusting bitmask (old: %#x)",
2312                         node, node_loc, CurrentProc, CurrentTSO->id,
2313                         // CurrentTSO, where_is(CurrentTSO),
2314                         node->header.gran.procs));
2315     node->header.gran.procs = (node->header.gran.procs) | PE_NUMBER(CurrentProc);
2316     IF_GRAN_DEBUG(bq,
2317                   belch("## new bitmask of node %p is %#x",
2318                         node, node->header.gran.procs));
2319     if (RtsFlags.GranFlags.GranSimStats.Global) {
2320       globalGranStats.tot_fake_fetches++;
2321     }
2322   }
2323
2324   bqe = q;
2325   // ToDo: check: ASSERT(CurrentProc==node_loc);
2326   while (get_itbl(bqe)->type==TSO) { // q != END_TSO_QUEUE) {
2327     //next = bqe->link;
2328     /*
2329        bqe points to the current element in the queue
2330        next points to the next element in the queue
2331     */
2332     //tso = (StgTSO *)bqe;  // wastes an assignment to get the type right
2333     //tso_loc = where_is(tso);
2334     len++;
2335     bqe = unblockOneLocked(bqe, node);
2336   }
2337
2338   /* if this is the BQ of an RBH, we have to put back the info ripped out of
2339      the closure to make room for the anchor of the BQ */
2340   if (bqe!=END_BQ_QUEUE) {
2341     ASSERT(get_itbl(node)->type == RBH && get_itbl(bqe)->type == CONSTR);
2342     /*
2343     ASSERT((info_ptr==&RBH_Save_0_info) ||
2344            (info_ptr==&RBH_Save_1_info) ||
2345            (info_ptr==&RBH_Save_2_info));
2346     */
2347     /* cf. convertToRBH in RBH.c for writing the RBHSave closure */
2348     ((StgRBH *)node)->blocking_queue = (StgBlockingQueueElement *)((StgRBHSave *)bqe)->payload[0];
2349     ((StgRBH *)node)->mut_link       = (StgMutClosure *)((StgRBHSave *)bqe)->payload[1];
2350
2351     IF_GRAN_DEBUG(bq,
2352                   belch("## Filled in RBH_Save for %p (%s) at end of AwBQ",
2353                         node, info_type(node)));
2354   }
2355
2356   /* statistics gathering */
2357   if (RtsFlags.GranFlags.GranSimStats.Global) {
2358     // globalGranStats.tot_bq_processing_time += bq_processing_time;
2359     globalGranStats.tot_bq_len += len;      // total length of all bqs awakened
2360     // globalGranStats.tot_bq_len_local += len_local;  // same for local TSOs only
2361     globalGranStats.tot_awbq++;             // total no. of bqs awakened
2362   }
2363   IF_GRAN_DEBUG(bq,
2364                 fprintf(stderr,"## BQ Stats of %p: [%d entries] %s\n",
2365                         node, len, (bqe!=END_BQ_QUEUE) ? "RBH" : ""));
2366 }
2367 #elif defined(PAR)
2368 void
2369 awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node)
2370 {
2371   StgBlockingQueueElement *bqe, *next;
2372
2373   ACQUIRE_LOCK(&sched_mutex);
2374
2375   IF_PAR_DEBUG(verbose,
2376                belch("## AwBQ for node %p on [%x]: ",
2377                      node, mytid));
2378
2379   ASSERT(get_itbl(q)->type == TSO ||
2380          get_itbl(q)->type == BLOCKED_FETCH ||
2381          get_itbl(q)->type == CONSTR);
2382
2383   bqe = q;
2384   while (get_itbl(bqe)->type==TSO ||
2385          get_itbl(bqe)->type==BLOCKED_FETCH) {
2386     bqe = unblockOneLocked(bqe, node);
2387   }
2388   RELEASE_LOCK(&sched_mutex);
2389 }
2390
2391 #else   /* !GRAN && !PAR */
2392 void
2393 awakenBlockedQueue(StgTSO *tso)
2394 {
2395   ACQUIRE_LOCK(&sched_mutex);
2396   while (tso != END_TSO_QUEUE) {
2397     tso = unblockOneLocked(tso);
2398   }
2399   RELEASE_LOCK(&sched_mutex);
2400 }
2401 #endif
2402
2403 //@node Exception Handling Routines, Debugging Routines, Blocking Queue Routines, Main scheduling code
2404 //@subsection Exception Handling Routines
2405
2406 /* ---------------------------------------------------------------------------
2407    Interrupt execution
2408    - usually called inside a signal handler so it mustn't do anything fancy.
2409    ------------------------------------------------------------------------ */
2410
2411 void
2412 interruptStgRts(void)
2413 {
2414     interrupted    = 1;
2415     context_switch = 1;
2416 }
2417
2418 /* -----------------------------------------------------------------------------
2419    Unblock a thread
2420
2421    This is for use when we raise an exception in another thread, which
2422    may be blocked.
2423    This has nothing to do with the UnblockThread event in GranSim. -- HWL
2424    -------------------------------------------------------------------------- */
2425
2426 #if defined(GRAN) || defined(PAR)
2427 /*
2428   NB: only the type of the blocking queue is different in GranSim and GUM
2429       the operations on the queue-elements are the same
2430       long live polymorphism!
2431 */
2432 static void
2433 unblockThread(StgTSO *tso)
2434 {
2435   StgBlockingQueueElement *t, **last;
2436
2437   ACQUIRE_LOCK(&sched_mutex);
2438   switch (tso->why_blocked) {
2439
2440   case NotBlocked:
2441     return;  /* not blocked */
2442
2443   case BlockedOnMVar:
2444     ASSERT(get_itbl(tso->block_info.closure)->type == MVAR);
2445     {
2446       StgBlockingQueueElement *last_tso = END_BQ_QUEUE;
2447       StgMVar *mvar = (StgMVar *)(tso->block_info.closure);
2448
2449       last = (StgBlockingQueueElement **)&mvar->head;
2450       for (t = (StgBlockingQueueElement *)mvar->head;
2451            t != END_BQ_QUEUE;
2452            last = &t->link, last_tso = t, t = t->link) {
2453         if (t == (StgBlockingQueueElement *)tso) {
2454           *last = (StgBlockingQueueElement *)tso->link;
2455           if (mvar->tail == tso) {
2456             mvar->tail = (StgTSO *)last_tso;
2457           }
2458           goto done;
2459         }
2460       }
2461       barf("unblockThread (MVAR): TSO not found");
2462     }
2463
2464   case BlockedOnBlackHole:
2465     ASSERT(get_itbl(tso->block_info.closure)->type == BLACKHOLE_BQ);
2466     {
2467       StgBlockingQueue *bq = (StgBlockingQueue *)(tso->block_info.closure);
2468
2469       last = &bq->blocking_queue;
2470       for (t = bq->blocking_queue;
2471            t != END_BQ_QUEUE;
2472            last = &t->link, t = t->link) {
2473         if (t == (StgBlockingQueueElement *)tso) {
2474           *last = (StgBlockingQueueElement *)tso->link;
2475           goto done;
2476         }
2477       }
2478       barf("unblockThread (BLACKHOLE): TSO not found");
2479     }
2480
2481   case BlockedOnException:
2482     {
2483       StgTSO *target  = tso->block_info.tso;
2484
2485       ASSERT(get_itbl(target)->type == TSO);
2486       ASSERT(target->blocked_exceptions != NULL);
2487
2488       last = (StgBlockingQueueElement **)&target->blocked_exceptions;
2489       for (t = (StgBlockingQueueElement *)target->blocked_exceptions;
2490            t != END_BQ_QUEUE;
2491            last = &t->link, t = t->link) {
2492         ASSERT(get_itbl(t)->type == TSO);
2493         if (t == (StgBlockingQueueElement *)tso) {
2494           *last = (StgBlockingQueueElement *)tso->link;
2495           goto done;
2496         }
2497       }
2498       barf("unblockThread (Exception): TSO not found");
2499     }
2500
2501   case BlockedOnDelay:
2502   case BlockedOnRead:
2503   case BlockedOnWrite:
2504     {
2505       StgBlockingQueueElement *prev = NULL;
2506       for (t = (StgBlockingQueueElement *)blocked_queue_hd; t != END_BQ_QUEUE;
2507            prev = t, t = t->link) {
2508         if (t == (StgBlockingQueueElement *)tso) {
2509           if (prev == NULL) {
2510             blocked_queue_hd = (StgTSO *)t->link;
2511             if ((StgBlockingQueueElement *)blocked_queue_tl == t) {
2512               blocked_queue_tl = END_TSO_QUEUE;
2513             }
2514           } else {
2515             prev->link = t->link;
2516             if ((StgBlockingQueueElement *)blocked_queue_tl == t) {
2517               blocked_queue_tl = (StgTSO *)prev;
2518             }
2519           }
2520           goto done;
2521         }
2522       }
2523       barf("unblockThread (I/O): TSO not found");
2524     }
2525
2526   default:
2527     barf("unblockThread");
2528   }
2529
2530  done:
2531   tso->link = END_TSO_QUEUE;
2532   tso->why_blocked = NotBlocked;
2533   tso->block_info.closure = NULL;
2534   PUSH_ON_RUN_QUEUE(tso);
2535   RELEASE_LOCK(&sched_mutex);
2536 }
2537 #else
2538 static void
2539 unblockThread(StgTSO *tso)
2540 {
2541   StgTSO *t, **last;
2542
2543   ACQUIRE_LOCK(&sched_mutex);
2544   switch (tso->why_blocked) {
2545
2546   case NotBlocked:
2547     return;  /* not blocked */
2548
2549   case BlockedOnMVar:
2550     ASSERT(get_itbl(tso->block_info.closure)->type == MVAR);
2551     {
2552       StgTSO *last_tso = END_TSO_QUEUE;
2553       StgMVar *mvar = (StgMVar *)(tso->block_info.closure);
2554
2555       last = &mvar->head;
2556       for (t = mvar->head; t != END_TSO_QUEUE;
2557            last = &t->link, last_tso = t, t = t->link) {
2558         if (t == tso) {
2559           *last = tso->link;
2560           if (mvar->tail == tso) {
2561             mvar->tail = last_tso;
2562           }
2563           goto done;
2564         }
2565       }
2566       barf("unblockThread (MVAR): TSO not found");
2567     }
2568
2569   case BlockedOnBlackHole:
2570     ASSERT(get_itbl(tso->block_info.closure)->type == BLACKHOLE_BQ);
2571     {
2572       StgBlockingQueue *bq = (StgBlockingQueue *)(tso->block_info.closure);
2573
2574       last = &bq->blocking_queue;
2575       for (t = bq->blocking_queue; t != END_TSO_QUEUE;
2576            last = &t->link, t = t->link) {
2577         if (t == tso) {
2578           *last = tso->link;
2579           goto done;
2580         }
2581       }
2582       barf("unblockThread (BLACKHOLE): TSO not found");
2583     }
2584
2585   case BlockedOnException:
2586     {
2587       StgTSO *target  = tso->block_info.tso;
2588
2589       ASSERT(get_itbl(target)->type == TSO);
2590       ASSERT(target->blocked_exceptions != NULL);
2591
2592       last = &target->blocked_exceptions;
2593       for (t = target->blocked_exceptions; t != END_TSO_QUEUE;
2594            last = &t->link, t = t->link) {
2595         ASSERT(get_itbl(t)->type == TSO);
2596         if (t == tso) {
2597           *last = tso->link;
2598           goto done;
2599         }
2600       }
2601       barf("unblockThread (Exception): TSO not found");
2602     }
2603
2604   case BlockedOnDelay:
2605   case BlockedOnRead:
2606   case BlockedOnWrite:
2607     {
2608       StgTSO *prev = NULL;
2609       for (t = blocked_queue_hd; t != END_TSO_QUEUE;
2610            prev = t, t = t->link) {
2611         if (t == tso) {
2612           if (prev == NULL) {
2613             blocked_queue_hd = t->link;
2614             if (blocked_queue_tl == t) {
2615               blocked_queue_tl = END_TSO_QUEUE;
2616             }
2617           } else {
2618             prev->link = t->link;
2619             if (blocked_queue_tl == t) {
2620               blocked_queue_tl = prev;
2621             }
2622           }
2623           goto done;
2624         }
2625       }
2626       barf("unblockThread (I/O): TSO not found");
2627     }
2628
2629   default:
2630     barf("unblockThread");
2631   }
2632
2633  done:
2634   tso->link = END_TSO_QUEUE;
2635   tso->why_blocked = NotBlocked;
2636   tso->block_info.closure = NULL;
2637   PUSH_ON_RUN_QUEUE(tso);
2638   RELEASE_LOCK(&sched_mutex);
2639 }
2640 #endif
2641
2642 /* -----------------------------------------------------------------------------
2643  * raiseAsync()
2644  *
2645  * The following function implements the magic for raising an
2646  * asynchronous exception in an existing thread.
2647  *
2648  * We first remove the thread from any queue on which it might be
2649  * blocked.  The possible blockages are MVARs and BLACKHOLE_BQs.
2650  *
2651  * We strip the stack down to the innermost CATCH_FRAME, building
2652  * thunks in the heap for all the active computations, so they can
2653  * be restarted if necessary.  When we reach a CATCH_FRAME, we build
2654  * an application of the handler to the exception, and push it on
2655  * the top of the stack.
2656  *
2657  * How exactly do we save all the active computations?  We create an
2658  * AP_UPD for every UpdateFrame on the stack.  Entering one of these
2659  * AP_UPDs pushes everything from the corresponding update frame
2660  * upwards onto the stack.  (Actually, it pushes everything up to the
2661  * next update frame plus a pointer to the next AP_UPD object.
2662  * Entering the next AP_UPD object pushes more onto the stack until we
2663  * reach the last AP_UPD object - at which point the stack should look
2664  * exactly as it did when we killed the TSO and we can continue
2665  * execution by entering the closure on top of the stack.
2666  *
2667  * We can also kill a thread entirely - this happens if either (a) the
2668  * exception passed to raiseAsync is NULL, or (b) there's no
2669  * CATCH_FRAME on the stack.  In either case, we strip the entire
2670  * stack and replace the thread with a zombie.
2671  *
2672  * -------------------------------------------------------------------------- */
2673
2674 void
2675 deleteThread(StgTSO *tso)
2676 {
2677   raiseAsync(tso,NULL);
2678 }
2679
2680 void
2681 raiseAsync(StgTSO *tso, StgClosure *exception)
2682 {
2683   StgUpdateFrame* su = tso->su;
2684   StgPtr          sp = tso->sp;
2685
2686   /* Thread already dead? */
2687   if (tso->what_next == ThreadComplete || tso->what_next == ThreadKilled) {
2688     return;
2689   }
2690
2691   IF_DEBUG(scheduler, sched_belch("raising exception in thread %ld.", tso->id));
2692
2693   /* Remove it from any blocking queues */
2694   unblockThread(tso);
2695
2696   /* The stack freezing code assumes there's a closure pointer on
2697    * the top of the stack.  This isn't always the case with compiled
2698    * code, so we have to push a dummy closure on the top which just
2699    * returns to the next return address on the stack.
2700    */
2701   if ( LOOKS_LIKE_GHC_INFO((void*)*sp) ) {
2702     *(--sp) = (W_)&dummy_ret_closure;
2703   }
2704
2705   while (1) {
2706     int words = ((P_)su - (P_)sp) - 1;
2707     nat i;
2708     StgAP_UPD * ap;
2709
2710     /* If we find a CATCH_FRAME, and we've got an exception to raise,
2711      * then build PAP(handler,exception,realworld#), and leave it on
2712      * top of the stack ready to enter.
2713      */
2714     if (get_itbl(su)->type == CATCH_FRAME && exception != NULL) {
2715       StgCatchFrame *cf = (StgCatchFrame *)su;
2716       /* we've got an exception to raise, so let's pass it to the
2717        * handler in this frame.
2718        */
2719       ap = (StgAP_UPD *)allocate(sizeofW(StgPAP) + 2);
2720       TICK_ALLOC_UPD_PAP(3,0);
2721       SET_HDR(ap,&PAP_info,cf->header.prof.ccs);
2722
2723       ap->n_args = 2;
2724       ap->fun = cf->handler;    /* :: Exception -> IO a */
2725       ap->payload[0] = exception;
2726       ap->payload[1] = ARG_TAG(0); /* realworld token */
2727
2728       /* throw away the stack from Sp up to and including the
2729        * CATCH_FRAME.
2730        */
2731       sp = (P_)su + sizeofW(StgCatchFrame) - 1;
2732       tso->su = cf->link;
2733
2734       /* Restore the blocked/unblocked state for asynchronous exceptions
2735        * at the CATCH_FRAME.
2736        *
2737        * If exceptions were unblocked at the catch, arrange that they
2738        * are unblocked again after executing the handler by pushing an
2739        * unblockAsyncExceptions_ret stack frame.
2740        */
2741       if (!cf->exceptions_blocked) {
2742         *(sp--) = (W_)&unblockAsyncExceptionszh_ret_info;
2743       }
2744
2745       /* Ensure that async exceptions are blocked when running the handler.
2746        */
2747       if (tso->blocked_exceptions == NULL) {
2748         tso->blocked_exceptions = END_TSO_QUEUE;
2749       }
2750
2751       /* Put the newly-built PAP on top of the stack, ready to execute
2752        * when the thread restarts.
2753        */
2754       sp[0] = (W_)ap;
2755       tso->sp = sp;
2756       tso->what_next = ThreadEnterGHC;
2757       IF_DEBUG(sanity, checkTSO(tso));
2758       return;
2759     }
2760
2761     /* First build an AP_UPD consisting of the stack chunk above the
2762      * current update frame, with the top word on the stack as the
2763      * fun field.
2764      */
2765     ap = (StgAP_UPD *)allocate(AP_sizeW(words));
2766
2767     ASSERT(words >= 0);
2768
2769     ap->n_args = words;
2770     ap->fun    = (StgClosure *)sp[0];
2771     sp++;
2772     for(i=0; i < (nat)words; ++i) {
2773       ap->payload[i] = (StgClosure *)*sp++;
2774     }
2775
2776     switch (get_itbl(su)->type) {
2777
2778     case UPDATE_FRAME:
2779       {
2780         SET_HDR(ap,&AP_UPD_info,su->header.prof.ccs /* ToDo */);
2781         TICK_ALLOC_UP_THK(words+1,0);
2782
2783         IF_DEBUG(scheduler,
2784                  fprintf(stderr,  "scheduler: Updating ");
2785                  printPtr((P_)su->updatee);
2786                  fprintf(stderr,  " with ");
2787                  printObj((StgClosure *)ap);
2788                  );
2789
2790         /* Replace the updatee with an indirection - happily
2791          * this will also wake up any threads currently
2792          * waiting on the result.
2793          */
2794         UPD_IND_NOLOCK(su->updatee,ap);  /* revert the black hole */
2795         su = su->link;
2796         sp += sizeofW(StgUpdateFrame) -1;
2797         sp[0] = (W_)ap; /* push onto stack */
2798         break;
2799       }
2800
2801     case CATCH_FRAME:
2802       {
2803         StgCatchFrame *cf = (StgCatchFrame *)su;
2804         StgClosure* o;
2805
2806         /* We want a PAP, not an AP_UPD.  Fortunately, the
2807          * layout's the same.
2808          */
2809         SET_HDR(ap,&PAP_info,su->header.prof.ccs /* ToDo */);
2810         TICK_ALLOC_UPD_PAP(words+1,0);
2811
2812         /* now build o = FUN(catch,ap,handler) */
2813         o = (StgClosure *)allocate(sizeofW(StgClosure)+2);
2814         TICK_ALLOC_FUN(2,0);
2815         SET_HDR(o,&catch_info,su->header.prof.ccs /* ToDo */);
2816         o->payload[0] = (StgClosure *)ap;
2817         o->payload[1] = cf->handler;
2818
2819         IF_DEBUG(scheduler,
2820                  fprintf(stderr,  "scheduler: Built ");
2821                  printObj((StgClosure *)o);
2822                  );
2823
2824         /* pop the old handler and put o on the stack */
2825         su = cf->link;
2826         sp += sizeofW(StgCatchFrame) - 1;
2827         sp[0] = (W_)o;
2828         break;
2829       }
2830
2831     case SEQ_FRAME:
2832       {
2833         StgSeqFrame *sf = (StgSeqFrame *)su;
2834         StgClosure* o;
2835
2836         SET_HDR(ap,&PAP_info,su->header.prof.ccs /* ToDo */);
2837         TICK_ALLOC_UPD_PAP(words+1,0);
2838
2839         /* now build o = FUN(seq,ap) */
2840         o = (StgClosure *)allocate(sizeofW(StgClosure)+1);
2841         TICK_ALLOC_SE_THK(1,0);
2842         SET_HDR(o,&seq_info,su->header.prof.ccs /* ToDo */);
2843         o->payload[0] = (StgClosure *)ap;
2844
2845         IF_DEBUG(scheduler,
2846                  fprintf(stderr,  "scheduler: Built ");
2847                  printObj((StgClosure *)o);
2848                  );
2849
2850         /* pop the old handler and put o on the stack */
2851         su = sf->link;
2852         sp += sizeofW(StgSeqFrame) - 1;
2853         sp[0] = (W_)o;
2854         break;
2855       }
2856
2857     case STOP_FRAME:
2858       /* We've stripped the entire stack, the thread is now dead. */
2859       sp += sizeofW(StgStopFrame) - 1;
2860       sp[0] = (W_)exception;    /* save the exception */
2861       tso->what_next = ThreadKilled;
2862       tso->su = (StgUpdateFrame *)(sp+1);
2863       tso->sp = sp;
2864       return;
2865
2866     default:
2867       barf("raiseAsync");
2868     }
2869   }
2870   barf("raiseAsync");
2871 }
2872
2873 /* -----------------------------------------------------------------------------
2874    resurrectThreads is called after garbage collection on the list of
2875    threads found to be garbage.  Each of these threads will be woken
2876    up and sent a signal: BlockedOnDeadMVar if the thread was blocked
2877    on an MVar, or NonTermination if the thread was blocked on a Black
2878    Hole.
2879    -------------------------------------------------------------------------- */
2880
2881 void
2882 resurrectThreads( StgTSO *threads )
2883 {
2884   StgTSO *tso, *next;
2885
2886   for (tso = threads; tso != END_TSO_QUEUE; tso = next) {
2887     next = tso->global_link;
2888     tso->global_link = all_threads;
2889     all_threads = tso;
2890     IF_DEBUG(scheduler, sched_belch("resurrecting thread %d", tso->id));
2891
2892     switch (tso->why_blocked) {
2893     case BlockedOnMVar:
2894     case BlockedOnException:
2895       raiseAsync(tso,(StgClosure *)BlockedOnDeadMVar_closure);
2896       break;
2897     case BlockedOnBlackHole:
2898       raiseAsync(tso,(StgClosure *)NonTermination_closure);
2899       break;
2900     case NotBlocked:
2901       /* This might happen if the thread was blocked on a black hole
2902        * belonging to a thread that we've just woken up (raiseAsync
2903        * can wake up threads, remember...).
2904        */
2905       continue;
2906     default:
2907       barf("resurrectThreads: thread blocked in a strange way");
2908     }
2909   }
2910 }
2911
2912 /* -----------------------------------------------------------------------------
2913  * Blackhole detection: if we reach a deadlock, test whether any
2914  * threads are blocked on themselves.  Any threads which are found to
2915  * be self-blocked get sent a NonTermination exception.
2916  *
2917  * This is only done in a deadlock situation in order to avoid
2918  * performance overhead in the normal case.
2919  * -------------------------------------------------------------------------- */
2920
2921 static void
2922 detectBlackHoles( void )
2923 {
2924     StgTSO *t = all_threads;
2925     StgUpdateFrame *frame;
2926     StgClosure *blocked_on;
2927
2928     for (t = all_threads; t != END_TSO_QUEUE; t = t->global_link) {
2929
2930         if (t->why_blocked != BlockedOnBlackHole) {
2931             continue;
2932         }
2933
2934         blocked_on = t->block_info.closure;
2935
2936         for (frame = t->su; ; frame = frame->link) {
2937             switch (get_itbl(frame)->type) {
2938
2939             case UPDATE_FRAME:
2940                 if (frame->updatee == blocked_on) {
2941                     /* We are blocking on one of our own computations, so
2942                      * send this thread the NonTermination exception.
2943                      */
2944                     IF_DEBUG(scheduler,
2945                              sched_belch("thread %d is blocked on itself", t->id));
2946                     raiseAsync(t, (StgClosure *)NonTermination_closure);
2947                     goto done;
2948                 }
2949                 else {
2950                     continue;
2951                 }
2952
2953             case CATCH_FRAME:
2954             case SEQ_FRAME:
2955                 continue;
2956
2957             case STOP_FRAME:
2958                 break;
2959             }
2960             break;
2961         }
2962
2963     done:
2964     }
2965 }
2966
2967 //@node Debugging Routines, Index, Exception Handling Routines, Main scheduling code
2968 //@subsection Debugging Routines
2969
2970 /* -----------------------------------------------------------------------------
2971    Debugging: why is a thread blocked
2972    -------------------------------------------------------------------------- */
2973
2974 #ifdef DEBUG
2975
2976 void
2977 printThreadBlockage(StgTSO *tso)
2978 {
2979   switch (tso->why_blocked) {
2980   case BlockedOnRead:
2981     fprintf(stderr,"blocked on read from fd %d", tso->block_info.fd);
2982     break;
2983   case BlockedOnWrite:
2984     fprintf(stderr,"blocked on write to fd %d", tso->block_info.fd);
2985     break;
2986   case BlockedOnDelay:
2987 #if defined(HAVE_SETITIMER) || defined(mingw32_TARGET_OS)
2988     fprintf(stderr,"blocked on delay of %d ms", tso->block_info.delay);
2989 #else
2990     fprintf(stderr,"blocked on delay of %d ms",
2991             tso->block_info.target - getourtimeofday());
2992 #endif
2993     break;
2994   case BlockedOnMVar:
2995     fprintf(stderr,"blocked on an MVar");
2996     break;
2997   case BlockedOnException:
2998     fprintf(stderr,"blocked on delivering an exception to thread %d",
2999             tso->block_info.tso->id);
3000     break;
3001   case BlockedOnBlackHole:
3002     fprintf(stderr,"blocked on a black hole");
3003     break;
3004   case NotBlocked:
3005     fprintf(stderr,"not blocked");
3006     break;
3007 #if defined(PAR)
3008   case BlockedOnGA:
3009     fprintf(stderr,"blocked on global address; local FM_BQ is %p (%s)",
3010             tso->block_info.closure, info_type(tso->block_info.closure));
3011     break;
3012   case BlockedOnGA_NoSend:
3013     fprintf(stderr,"blocked on global address (no send); local FM_BQ is %p (%s)",
3014             tso->block_info.closure, info_type(tso->block_info.closure));
3015     break;
3016 #endif
3017   default:
3018     barf("printThreadBlockage: strange tso->why_blocked: %d for TSO %d (%d)",
3019          tso->why_blocked, tso->id, tso);
3020   }
3021 }
3022
3023 void
3024 printThreadStatus(StgTSO *tso)
3025 {
3026   switch (tso->what_next) {
3027   case ThreadKilled:
3028     fprintf(stderr,"has been killed");
3029     break;
3030   case ThreadComplete:
3031     fprintf(stderr,"has completed");
3032     break;
3033   default:
3034     printThreadBlockage(tso);
3035   }
3036 }
3037
3038 void
3039 printAllThreads(void)
3040 {
3041   StgTSO *t;
3042
3043   sched_belch("all threads:");
3044   for (t = all_threads; t != END_TSO_QUEUE; t = t->global_link) {
3045     fprintf(stderr, "\tthread %d is ", t->id);
3046     printThreadStatus(t);
3047     fprintf(stderr,"\n");
3048   }
3049 }
3050
3051 /*
3052    Print a whole blocking queue attached to node (debugging only).
3053 */
3054 //@cindex print_bq
3055 # if defined(PAR)
3056 void
3057 print_bq (StgClosure *node)
3058 {
3059   StgBlockingQueueElement *bqe;
3060   StgTSO *tso;
3061   rtsBool end;
3062
3063   fprintf(stderr,"## BQ of closure %p (%s): ",
3064           node, info_type(node));
3065
3066   /* should cover all closures that may have a blocking queue */
3067   ASSERT(get_itbl(node)->type == BLACKHOLE_BQ ||
3068          get_itbl(node)->type == FETCH_ME_BQ ||
3069          get_itbl(node)->type == RBH);
3070
3071   ASSERT(node!=(StgClosure*)NULL);         // sanity check
3072   /*
3073      NB: In a parallel setup a BQ of an RBH must end with an RBH_Save closure;
3074   */
3075   for (bqe = ((StgBlockingQueue*)node)->blocking_queue, end = (bqe==END_BQ_QUEUE);
3076        !end; // iterate until bqe points to a CONSTR
3077        end = (get_itbl(bqe)->type == CONSTR) || (bqe->link==END_BQ_QUEUE), bqe = end ? END_BQ_QUEUE : bqe->link) {
3078     ASSERT(bqe != END_BQ_QUEUE);             // sanity check
3079     ASSERT(bqe != (StgTSO*)NULL);            // sanity check
3080     /* types of closures that may appear in a blocking queue */
3081     ASSERT(get_itbl(bqe)->type == TSO ||
3082            get_itbl(bqe)->type == BLOCKED_FETCH ||
3083            get_itbl(bqe)->type == CONSTR);
3084     /* only BQs of an RBH end with an RBH_Save closure */
3085     ASSERT(get_itbl(bqe)->type != CONSTR || get_itbl(node)->type == RBH);
3086
3087     switch (get_itbl(bqe)->type) {
3088     case TSO:
3089       fprintf(stderr," TSO %d (%x),",
3090               ((StgTSO *)bqe)->id, ((StgTSO *)bqe));
3091       break;
3092     case BLOCKED_FETCH:
3093       fprintf(stderr," BF (node=%p, ga=((%x, %d, %x)),",
3094               ((StgBlockedFetch *)bqe)->node,
3095               ((StgBlockedFetch *)bqe)->ga.payload.gc.gtid,
3096               ((StgBlockedFetch *)bqe)->ga.payload.gc.slot,
3097               ((StgBlockedFetch *)bqe)->ga.weight);
3098       break;
3099     case CONSTR:
3100       fprintf(stderr," %s (IP %p),",
3101               (get_itbl(bqe) == &RBH_Save_0_info ? "RBH_Save_0" :
3102                get_itbl(bqe) == &RBH_Save_1_info ? "RBH_Save_1" :
3103                get_itbl(bqe) == &RBH_Save_2_info ? "RBH_Save_2" :
3104                "RBH_Save_?"), get_itbl(bqe));
3105       break;
3106     default:
3107       barf("Unexpected closure type %s in blocking queue of %p (%s)",
3108            info_type(bqe), node, info_type(node));
3109       break;
3110     }
3111   } /* for */
3112   fputc('\n', stderr);
3113 }
3114 # elif defined(GRAN)
3115 void
3116 print_bq (StgClosure *node)
3117 {
3118   StgBlockingQueueElement *bqe;
3119   PEs node_loc, tso_loc;
3120   rtsBool end;
3121
3122   /* should cover all closures that may have a blocking queue */
3123   ASSERT(get_itbl(node)->type == BLACKHOLE_BQ ||
3124          get_itbl(node)->type == FETCH_ME_BQ ||
3125          get_itbl(node)->type == RBH);
3126
3127   ASSERT(node!=(StgClosure*)NULL);         // sanity check
3128   node_loc = where_is(node);
3129
3130   fprintf(stderr,"## BQ of closure %p (%s) on [PE %d]: ",
3131           node, info_type(node), node_loc);
3132
3133   /*
3134      NB: In a parallel setup a BQ of an RBH must end with an RBH_Save closure;
3135   */
3136   for (bqe = ((StgBlockingQueue*)node)->blocking_queue, end = (bqe==END_BQ_QUEUE);
3137        !end; // iterate until bqe points to a CONSTR
3138        end = (get_itbl(bqe)->type == CONSTR) || (bqe->link==END_BQ_QUEUE), bqe = end ? END_BQ_QUEUE : bqe->link) {
3139     ASSERT(bqe != END_BQ_QUEUE);             // sanity check
3140     ASSERT(bqe != (StgBlockingQueueElement *)NULL);  // sanity check
3141     /* types of closures that may appear in a blocking queue */
3142     ASSERT(get_itbl(bqe)->type == TSO ||
3143            get_itbl(bqe)->type == CONSTR);
3144     /* only BQs of an RBH end with an RBH_Save closure */
3145     ASSERT(get_itbl(bqe)->type != CONSTR || get_itbl(node)->type == RBH);
3146
3147     tso_loc = where_is((StgClosure *)bqe);
3148     switch (get_itbl(bqe)->type) {
3149     case TSO:
3150       fprintf(stderr," TSO %d (%p) on [PE %d],",
3151               ((StgTSO *)bqe)->id, (StgTSO *)bqe, tso_loc);
3152       break;
3153     case CONSTR:
3154       fprintf(stderr," %s (IP %p),",
3155               (get_itbl(bqe) == &RBH_Save_0_info ? "RBH_Save_0" :
3156                get_itbl(bqe) == &RBH_Save_1_info ? "RBH_Save_1" :
3157                get_itbl(bqe) == &RBH_Save_2_info ? "RBH_Save_2" :
3158                "RBH_Save_?"), get_itbl(bqe));
3159       break;
3160     default:
3161       barf("Unexpected closure type %s in blocking queue of %p (%s)",
3162            info_type((StgClosure *)bqe), node, info_type(node));
3163       break;
3164     }
3165   } /* for */
3166   fputc('\n', stderr);
3167 }
3168 #else
3169 /*
3170    Nice and easy: only TSOs on the blocking queue
3171 */
3172 void
3173 print_bq (StgClosure *node)
3174 {
3175   StgTSO *tso;
3176
3177   ASSERT(node!=(StgClosure*)NULL);         // sanity check
3178   for (tso = ((StgBlockingQueue*)node)->blocking_queue;
3179        tso != END_TSO_QUEUE;
3180        tso=tso->link) {
3181     ASSERT(tso!=NULL && tso!=END_TSO_QUEUE);   // sanity check
3182     ASSERT(get_itbl(tso)->type == TSO);  // guess what, sanity check
3183     fprintf(stderr," TSO %d (%p),", tso->id, tso);
3184   }
3185   fputc('\n', stderr);
3186 }
3187 # endif
3188
3189 #if defined(PAR)
3190 static nat
3191 run_queue_len(void)
3192 {
3193   nat i;
3194   StgTSO *tso;
3195
3196   for (i=0, tso=run_queue_hd;
3197        tso != END_TSO_QUEUE;
3198        i++, tso=tso->link)
3199     /* nothing */
3200
3201   return i;
3202 }
3203 #endif
3204
3205 static void
3206 sched_belch(char *s, ...)
3207 {
3208   va_list ap;
3209   va_start(ap,s);
3210 #ifdef SMP
3211   fprintf(stderr, "scheduler (task %ld): ", pthread_self());
3212 #else
3213   fprintf(stderr, "scheduler: ");
3214 #endif
3215   vfprintf(stderr, s, ap);
3216   fprintf(stderr, "\n");
3217 }
3218
3219 #endif /* DEBUG */
3220
3221
3222 //@node Index,  , Debugging Routines, Main scheduling code
3223 //@subsection Index
3224
3225 //@index
3226 //* MainRegTable::  @cindex\s-+MainRegTable
3227 //* StgMainThread::  @cindex\s-+StgMainThread
3228 //* awaken_blocked_queue::  @cindex\s-+awaken_blocked_queue
3229 //* blocked_queue_hd::  @cindex\s-+blocked_queue_hd
3230 //* blocked_queue_tl::  @cindex\s-+blocked_queue_tl
3231 //* context_switch::  @cindex\s-+context_switch
3232 //* createThread::  @cindex\s-+createThread
3233 //* free_capabilities::  @cindex\s-+free_capabilities
3234 //* gc_pending_cond::  @cindex\s-+gc_pending_cond
3235 //* initScheduler::  @cindex\s-+initScheduler
3236 //* interrupted::  @cindex\s-+interrupted
3237 //* n_free_capabilities::  @cindex\s-+n_free_capabilities
3238 //* next_thread_id::  @cindex\s-+next_thread_id
3239 //* print_bq::  @cindex\s-+print_bq
3240 //* run_queue_hd::  @cindex\s-+run_queue_hd
3241 //* run_queue_tl::  @cindex\s-+run_queue_tl
3242 //* sched_mutex::  @cindex\s-+sched_mutex
3243 //* schedule::  @cindex\s-+schedule
3244 //* take_off_run_queue::  @cindex\s-+take_off_run_queue
3245 //* task_ids::  @cindex\s-+task_ids
3246 //* term_mutex::  @cindex\s-+term_mutex
3247 //* thread_ready_cond::  @cindex\s-+thread_ready_cond
3248 //@end index