ghc/rts/Schedule.c

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