ghc/rts/Schedule.c

   1 /* ---------------------------------------------------------------------------
   2  * $Id: Schedule.c,v 1.87 2001/01/24 15:46:19 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 "Interpreter.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   "ThreadEnterInterp",
 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 ThreadEnterInterp:
 890 #ifdef GHCI
 891       {
 892          IF_DEBUG(scheduler,sched_belch("entering interpreter"));
 893          ret = interpretBCO(cap);
 894          break;
 895       }
 896 #else
 897       barf("Panic: entered a BCO but no bytecode interpreter in this build");
 898 #endif
 899     default:
 900       barf("schedule: invalid what_next field");
 901     }
 902     /* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
 903
 904     /* Costs for the scheduler are assigned to CCS_SYSTEM */
 905 #ifdef PROFILING
 906     CCCS = CCS_SYSTEM;
 907 #endif
 908
 909     ACQUIRE_LOCK(&sched_mutex);
 910
 911 #ifdef SMP
 912     IF_DEBUG(scheduler,fprintf(stderr,"scheduler (task %ld): ", pthread_self()););
 913 #elif !defined(GRAN) && !defined(PAR)
 914     IF_DEBUG(scheduler,fprintf(stderr,"scheduler: "););
 915 #endif
 916     t = cap->rCurrentTSO;
 917
 918 #if defined(PAR)
 919     /* HACK 675: if the last thread didn't yield, make sure to print a
 920        SCHEDULE event to the log file when StgRunning the next thread, even
 921        if it is the same one as before */
 922     LastTSO = t; //(ret == ThreadBlocked) ? END_TSO_QUEUE : t;
 923     TimeOfLastYield = CURRENT_TIME;
 924 #endif
 925
 926     switch (ret) {
 927     case HeapOverflow:
 928       /* make all the running tasks block on a condition variable,
 929        * maybe set context_switch and wait till they all pile in,
 930        * then have them wait on a GC condition variable.
 931        */
 932       IF_DEBUG(scheduler,belch("--<< thread %ld (%p; %s) stopped: HeapOverflow",
 933                                t->id, t, whatNext_strs[t->what_next]));
 934       threadPaused(t);
 935 #if defined(GRAN)
 936       ASSERT(!is_on_queue(t,CurrentProc));
 937 #endif
 938
 939       ready_to_gc = rtsTrue;
 940       context_switch = 1;               /* stop other threads ASAP */
 941       PUSH_ON_RUN_QUEUE(t);
 942       /* actual GC is done at the end of the while loop */
 943       break;
 944
 945     case StackOverflow:
 946       IF_DEBUG(scheduler,belch("--<< thread %ld (%p; %s) stopped, StackOverflow",
 947                                t->id, t, whatNext_strs[t->what_next]));
 948       /* just adjust the stack for this thread, then pop it back
 949        * on the run queue.
 950        */
 951       threadPaused(t);
 952       {
 953         StgMainThread *m;
 954         /* enlarge the stack */
 955         StgTSO *new_t = threadStackOverflow(t);
 956
 957         /* This TSO has moved, so update any pointers to it from the
 958          * main thread stack.  It better not be on any other queues...
 959          * (it shouldn't be).
 960          */
 961         for (m = main_threads; m != NULL; m = m->link) {
 962           if (m->tso == t) {
 963             m->tso = new_t;
 964           }
 965         }
 966         threadPaused(new_t);
 967         PUSH_ON_RUN_QUEUE(new_t);
 968       }
 969       break;
 970
 971     case ThreadYielding:
 972 #if defined(GRAN)
 973       IF_DEBUG(gran,
 974                DumpGranEvent(GR_DESCHEDULE, t));
 975       globalGranStats.tot_yields++;
 976 #elif defined(PAR)
 977       IF_DEBUG(par,
 978                DumpGranEvent(GR_DESCHEDULE, t));
 979 #endif
 980       /* put the thread back on the run queue.  Then, if we're ready to
 981        * GC, check whether this is the last task to stop.  If so, wake
 982        * up the GC thread.  getThread will block during a GC until the
 983        * GC is finished.
 984        */
 985       IF_DEBUG(scheduler,
 986                if (t->what_next == ThreadEnterInterp) {
 987                    /* ToDo: or maybe a timer expired when we were in Hugs?
 988                     * or maybe someone hit ctrl-C
 989                     */
 990                    belch("--<< thread %ld (%p; %s) stopped to switch to Hugs",
 991                          t->id, t, whatNext_strs[t->what_next]);
 992                } else {
 993                    belch("--<< thread %ld (%p; %s) stopped, yielding",
 994                          t->id, t, whatNext_strs[t->what_next]);
 995                }
 996                );
 997
 998       threadPaused(t);
 999
1000       IF_DEBUG(sanity,
1001                //belch("&& Doing sanity check on yielding TSO %ld.", t->id);
1002                checkTSO(t));
1003       ASSERT(t->link == END_TSO_QUEUE);
1004 #if defined(GRAN)
1005       ASSERT(!is_on_queue(t,CurrentProc));
1006
1007       IF_DEBUG(sanity,
1008                //belch("&& Doing sanity check on all ThreadQueues (and their TSOs).");
1009                checkThreadQsSanity(rtsTrue));
1010 #endif
1011       APPEND_TO_RUN_QUEUE(t);
1012 #if defined(GRAN)
1013       /* add a ContinueThread event to actually process the thread */
1014       new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
1015                 ContinueThread,
1016                 t, (StgClosure*)NULL, (rtsSpark*)NULL);
1017       IF_GRAN_DEBUG(bq,
1018                belch("GRAN: eventq and runnableq after adding yielded thread to queue again:");
1019                G_EVENTQ(0);
1020                G_CURR_THREADQ(0))
1021 #endif /* GRAN */
1022       break;
1023
1024     case ThreadBlocked:
1025 #if defined(GRAN)
1026       IF_DEBUG(scheduler,
1027                belch("--<< thread %ld (%p; %s) stopped, blocking on node %p [PE %d] with BQ: ",
1028                                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)));
1029                if (t->block_info.closure!=(StgClosure*)NULL) print_bq(t->block_info.closure));
1030
1031       // ??? needed; should emit block before
1032       IF_DEBUG(gran,
1033                DumpGranEvent(GR_DESCHEDULE, t));
1034       prune_eventq(t, (StgClosure *)NULL); // prune ContinueThreads for t
1035       /*
1036         ngoq Dogh!
1037       ASSERT(procStatus[CurrentProc]==Busy ||
1038               ((procStatus[CurrentProc]==Fetching) &&
1039               (t->block_info.closure!=(StgClosure*)NULL)));
1040       if (run_queue_hds[CurrentProc] == END_TSO_QUEUE &&
1041           !(!RtsFlags.GranFlags.DoAsyncFetch &&
1042             procStatus[CurrentProc]==Fetching))
1043         procStatus[CurrentProc] = Idle;
1044       */
1045 #elif defined(PAR)
1046       IF_DEBUG(par,
1047                DumpGranEvent(GR_DESCHEDULE, t));
1048
1049       /* Send a fetch (if BlockedOnGA) and dump event to log file */
1050       blockThread(t);
1051
1052       IF_DEBUG(scheduler,
1053                belch("--<< thread %ld (%p; %s) stopped, blocking on node %p with BQ: ",
1054                                t->id, t, whatNext_strs[t->what_next], t->block_info.closure);
1055                if (t->block_info.closure!=(StgClosure*)NULL) print_bq(t->block_info.closure));
1056
1057 #else /* !GRAN */
1058       /* don't need to do anything.  Either the thread is blocked on
1059        * I/O, in which case we'll have called addToBlockedQueue
1060        * previously, or it's blocked on an MVar or Blackhole, in which
1061        * case it'll be on the relevant queue already.
1062        */
1063       IF_DEBUG(scheduler,
1064                fprintf(stderr, "--<< thread %d (%p) stopped: ", t->id, t);
1065                printThreadBlockage(t);
1066                fprintf(stderr, "\n"));
1067
1068       /* Only for dumping event to log file
1069          ToDo: do I need this in GranSim, too?
1070       blockThread(t);
1071       */
1072 #endif
1073       threadPaused(t);
1074       break;
1075
1076     case ThreadFinished:
1077       /* Need to check whether this was a main thread, and if so, signal
1078        * the task that started it with the return value.  If we have no
1079        * more main threads, we probably need to stop all the tasks until
1080        * we get a new one.
1081        */
1082       /* We also end up here if the thread kills itself with an
1083        * uncaught exception, see Exception.hc.
1084        */
1085       IF_DEBUG(scheduler,belch("--++ thread %d (%p) finished", t->id, t));
1086 #if defined(GRAN)
1087       endThread(t, CurrentProc); // clean-up the thread
1088 #elif defined(PAR)
1089       advisory_thread_count--;
1090       if (RtsFlags.ParFlags.ParStats.Full)
1091         DumpEndEvent(CURRENT_PROC, t, rtsFalse /* not mandatory */);
1092 #endif
1093       break;
1094
1095     default:
1096       barf("schedule: invalid thread return code %d", (int)ret);
1097     }
1098
1099 #ifdef SMP
1100     cap->link = free_capabilities;
1101     free_capabilities = cap;
1102     n_free_capabilities++;
1103 #endif
1104
1105 #ifdef SMP
1106     if (ready_to_gc && n_free_capabilities == RtsFlags.ParFlags.nNodes)
1107 #else
1108     if (ready_to_gc)
1109 #endif
1110       {
1111       /* everybody back, start the GC.
1112        * Could do it in this thread, or signal a condition var
1113        * to do it in another thread.  Either way, we need to
1114        * broadcast on gc_pending_cond afterward.
1115        */
1116 #ifdef SMP
1117       IF_DEBUG(scheduler,sched_belch("doing GC"));
1118 #endif
1119       GarbageCollect(GetRoots,rtsFalse);
1120       ready_to_gc = rtsFalse;
1121 #ifdef SMP
1122       pthread_cond_broadcast(&gc_pending_cond);
1123 #endif
1124 #if defined(GRAN)
1125       /* add a ContinueThread event to continue execution of current thread */
1126       new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
1127                 ContinueThread,
1128                 t, (StgClosure*)NULL, (rtsSpark*)NULL);
1129       IF_GRAN_DEBUG(bq,
1130                fprintf(stderr, "GRAN: eventq and runnableq after Garbage collection:\n");
1131                G_EVENTQ(0);
1132                G_CURR_THREADQ(0))
1133 #endif /* GRAN */
1134     }
1135 #if defined(GRAN)
1136   next_thread:
1137     IF_GRAN_DEBUG(unused,
1138                   print_eventq(EventHd));
1139
1140     event = get_next_event();
1141
1142 #elif defined(PAR)
1143   next_thread:
1144     /* ToDo: wait for next message to arrive rather than busy wait */
1145
1146 #else /* GRAN */
1147   /* not any more
1148   next_thread:
1149     t = take_off_run_queue(END_TSO_QUEUE);
1150   */
1151 #endif /* GRAN */
1152   } /* end of while(1) */
1153 }
1154
1155 /* ---------------------------------------------------------------------------
1156  * deleteAllThreads():  kill all the live threads.
1157  *
1158  * This is used when we catch a user interrupt (^C), before performing
1159  * any necessary cleanups and running finalizers.
1160  * ------------------------------------------------------------------------- */
1161
1162 void deleteAllThreads ( void )
1163 {
1164   StgTSO* t;
1165   IF_DEBUG(scheduler,sched_belch("deleting all threads"));
1166   for (t = run_queue_hd; t != END_TSO_QUEUE; t = t->link) {
1167       deleteThread(t);
1168   }
1169   for (t = blocked_queue_hd; t != END_TSO_QUEUE; t = t->link) {
1170       deleteThread(t);
1171   }
1172   for (t = sleeping_queue; t != END_TSO_QUEUE; t = t->link) {
1173       deleteThread(t);
1174   }
1175   run_queue_hd = run_queue_tl = END_TSO_QUEUE;
1176   blocked_queue_hd = blocked_queue_tl = END_TSO_QUEUE;
1177   sleeping_queue = END_TSO_QUEUE;
1178 }
1179
1180 /* startThread and  insertThread are now in GranSim.c -- HWL */
1181
1182 //@node Suspend and Resume, Run queue code, Main scheduling loop, Main scheduling code
1183 //@subsection Suspend and Resume
1184
1185 /* ---------------------------------------------------------------------------
1186  * Suspending & resuming Haskell threads.
1187  *
1188  * When making a "safe" call to C (aka _ccall_GC), the task gives back
1189  * its capability before calling the C function.  This allows another
1190  * task to pick up the capability and carry on running Haskell
1191  * threads.  It also means that if the C call blocks, it won't lock
1192  * the whole system.
1193  *
1194  * The Haskell thread making the C call is put to sleep for the
1195  * duration of the call, on the susepended_ccalling_threads queue.  We
1196  * give out a token to the task, which it can use to resume the thread
1197  * on return from the C function.
1198  * ------------------------------------------------------------------------- */
1199
1200 StgInt
1201 suspendThread( Capability *cap )
1202 {
1203   nat tok;
1204
1205   ACQUIRE_LOCK(&sched_mutex);
1206
1207   IF_DEBUG(scheduler,
1208            sched_belch("thread %d did a _ccall_gc", cap->rCurrentTSO->id));
1209
1210   threadPaused(cap->rCurrentTSO);
1211   cap->rCurrentTSO->link = suspended_ccalling_threads;
1212   suspended_ccalling_threads = cap->rCurrentTSO;
1213
1214   /* Use the thread ID as the token; it should be unique */
1215   tok = cap->rCurrentTSO->id;
1216
1217 #ifdef SMP
1218   cap->link = free_capabilities;
1219   free_capabilities = cap;
1220   n_free_capabilities++;
1221 #endif
1222
1223   RELEASE_LOCK(&sched_mutex);
1224   return tok;
1225 }
1226
1227 Capability *
1228 resumeThread( StgInt tok )
1229 {
1230   StgTSO *tso, **prev;
1231   Capability *cap;
1232
1233   ACQUIRE_LOCK(&sched_mutex);
1234
1235   prev = &suspended_ccalling_threads;
1236   for (tso = suspended_ccalling_threads;
1237        tso != END_TSO_QUEUE;
1238        prev = &tso->link, tso = tso->link) {
1239     if (tso->id == (StgThreadID)tok) {
1240       *prev = tso->link;
1241       break;
1242     }
1243   }
1244   if (tso == END_TSO_QUEUE) {
1245     barf("resumeThread: thread not found");
1246   }
1247   tso->link = END_TSO_QUEUE;
1248
1249 #ifdef SMP
1250   while (free_capabilities == NULL) {
1251     IF_DEBUG(scheduler, sched_belch("waiting to resume"));
1252     pthread_cond_wait(&thread_ready_cond, &sched_mutex);
1253     IF_DEBUG(scheduler, sched_belch("resuming thread %d", tso->id));
1254   }
1255   cap = free_capabilities;
1256   free_capabilities = cap->link;
1257   n_free_capabilities--;
1258 #else
1259   cap = &MainRegTable;
1260 #endif
1261
1262   cap->rCurrentTSO = tso;
1263
1264   RELEASE_LOCK(&sched_mutex);
1265   return cap;
1266 }
1267
1268
1269 /* ---------------------------------------------------------------------------
1270  * Static functions
1271  * ------------------------------------------------------------------------ */
1272 static void unblockThread(StgTSO *tso);
1273
1274 /* ---------------------------------------------------------------------------
1275  * Comparing Thread ids.
1276  *
1277  * This is used from STG land in the implementation of the
1278  * instances of Eq/Ord for ThreadIds.
1279  * ------------------------------------------------------------------------ */
1280
1281 int cmp_thread(const StgTSO *tso1, const StgTSO *tso2)
1282 {
1283   StgThreadID id1 = tso1->id;
1284   StgThreadID id2 = tso2->id;
1285
1286   if (id1 < id2) return (-1);
1287   if (id1 > id2) return 1;
1288   return 0;
1289 }
1290
1291 /* ---------------------------------------------------------------------------
1292    Create a new thread.
1293
1294    The new thread starts with the given stack size.  Before the
1295    scheduler can run, however, this thread needs to have a closure
1296    (and possibly some arguments) pushed on its stack.  See
1297    pushClosure() in Schedule.h.
1298
1299    createGenThread() and createIOThread() (in SchedAPI.h) are
1300    convenient packaged versions of this function.
1301
1302    currently pri (priority) is only used in a GRAN setup -- HWL
1303    ------------------------------------------------------------------------ */
1304 //@cindex createThread
1305 #if defined(GRAN)
1306 /*   currently pri (priority) is only used in a GRAN setup -- HWL */
1307 StgTSO *
1308 createThread(nat stack_size, StgInt pri)
1309 {
1310   return createThread_(stack_size, rtsFalse, pri);
1311 }
1312
1313 static StgTSO *
1314 createThread_(nat size, rtsBool have_lock, StgInt pri)
1315 {
1316 #else
1317 StgTSO *
1318 createThread(nat stack_size)
1319 {
1320   return createThread_(stack_size, rtsFalse);
1321 }
1322
1323 static StgTSO *
1324 createThread_(nat size, rtsBool have_lock)
1325 {
1326 #endif
1327
1328     StgTSO *tso;
1329     nat stack_size;
1330
1331     /* First check whether we should create a thread at all */
1332 #if defined(PAR)
1333   /* check that no more than RtsFlags.ParFlags.maxThreads threads are created */
1334   if (advisory_thread_count >= RtsFlags.ParFlags.maxThreads) {
1335     threadsIgnored++;
1336     belch("{createThread}Daq ghuH: refusing to create another thread; no more than %d threads allowed (currently %d)",
1337           RtsFlags.ParFlags.maxThreads, advisory_thread_count);
1338     return END_TSO_QUEUE;
1339   }
1340   threadsCreated++;
1341 #endif
1342
1343 #if defined(GRAN)
1344   ASSERT(!RtsFlags.GranFlags.Light || CurrentProc==0);
1345 #endif
1346
1347   // ToDo: check whether size = stack_size - TSO_STRUCT_SIZEW
1348
1349   /* catch ridiculously small stack sizes */
1350   if (size < MIN_STACK_WORDS + TSO_STRUCT_SIZEW) {
1351     size = MIN_STACK_WORDS + TSO_STRUCT_SIZEW;
1352   }
1353
1354   stack_size = size - TSO_STRUCT_SIZEW;
1355
1356   tso = (StgTSO *)allocate(size);
1357   TICK_ALLOC_TSO(size-TSO_STRUCT_SIZEW, 0);
1358
1359   SET_HDR(tso, &stg_TSO_info, CCS_SYSTEM);
1360 #if defined(GRAN)
1361   SET_GRAN_HDR(tso, ThisPE);
1362 #endif
1363   tso->what_next     = ThreadEnterGHC;
1364
1365   /* tso->id needs to be unique.  For now we use a heavyweight mutex to
1366    * protect the increment operation on next_thread_id.
1367    * In future, we could use an atomic increment instead.
1368    */
1369   if (!have_lock) { ACQUIRE_LOCK(&sched_mutex); }
1370   tso->id = next_thread_id++;
1371   if (!have_lock) { RELEASE_LOCK(&sched_mutex); }
1372
1373   tso->why_blocked  = NotBlocked;
1374   tso->blocked_exceptions = NULL;
1375
1376   tso->stack_size   = stack_size;
1377   tso->max_stack_size = round_to_mblocks(RtsFlags.GcFlags.maxStkSize)
1378                               - TSO_STRUCT_SIZEW;
1379   tso->sp           = (P_)&(tso->stack) + stack_size;
1380
1381 #ifdef PROFILING
1382   tso->prof.CCCS = CCS_MAIN;
1383 #endif
1384
1385   /* put a stop frame on the stack */
1386   tso->sp -= sizeofW(StgStopFrame);
1387   SET_HDR((StgClosure*)tso->sp,(StgInfoTable *)&stg_stop_thread_info,CCS_SYSTEM);
1388   tso->su = (StgUpdateFrame*)tso->sp;
1389
1390   // ToDo: check this
1391 #if defined(GRAN)
1392   tso->link = END_TSO_QUEUE;
1393   /* uses more flexible routine in GranSim */
1394   insertThread(tso, CurrentProc);
1395 #else
1396   /* In a non-GranSim setup the pushing of a TSO onto the runq is separated
1397    * from its creation
1398    */
1399 #endif
1400
1401 #if defined(GRAN) || defined(PAR)
1402   DumpGranEvent(GR_START,tso);
1403 #endif
1404
1405   /* Link the new thread on the global thread list.
1406    */
1407   tso->global_link = all_threads;
1408   all_threads = tso;
1409
1410 #if defined(GRAN)
1411   tso->gran.pri = pri;
1412 # if defined(DEBUG)
1413   tso->gran.magic = TSO_MAGIC; // debugging only
1414 # endif
1415   tso->gran.sparkname   = 0;
1416   tso->gran.startedat   = CURRENT_TIME;
1417   tso->gran.exported    = 0;
1418   tso->gran.basicblocks = 0;
1419   tso->gran.allocs      = 0;
1420   tso->gran.exectime    = 0;
1421   tso->gran.fetchtime   = 0;
1422   tso->gran.fetchcount  = 0;
1423   tso->gran.blocktime   = 0;
1424   tso->gran.blockcount  = 0;
1425   tso->gran.blockedat   = 0;
1426   tso->gran.globalsparks = 0;
1427   tso->gran.localsparks  = 0;
1428   if (RtsFlags.GranFlags.Light)
1429     tso->gran.clock  = Now; /* local clock */
1430   else
1431     tso->gran.clock  = 0;
1432
1433   IF_DEBUG(gran,printTSO(tso));
1434 #elif defined(PAR)
1435 # if defined(DEBUG)
1436   tso->par.magic = TSO_MAGIC; // debugging only
1437 # endif
1438   tso->par.sparkname   = 0;
1439   tso->par.startedat   = CURRENT_TIME;
1440   tso->par.exported    = 0;
1441   tso->par.basicblocks = 0;
1442   tso->par.allocs      = 0;
1443   tso->par.exectime    = 0;
1444   tso->par.fetchtime   = 0;
1445   tso->par.fetchcount  = 0;
1446   tso->par.blocktime   = 0;
1447   tso->par.blockcount  = 0;
1448   tso->par.blockedat   = 0;
1449   tso->par.globalsparks = 0;
1450   tso->par.localsparks  = 0;
1451 #endif
1452
1453 #if defined(GRAN)
1454   globalGranStats.tot_threads_created++;
1455   globalGranStats.threads_created_on_PE[CurrentProc]++;
1456   globalGranStats.tot_sq_len += spark_queue_len(CurrentProc);
1457   globalGranStats.tot_sq_probes++;
1458 #endif
1459
1460 #if defined(GRAN)
1461   IF_GRAN_DEBUG(pri,
1462                 belch("==__ schedule: Created TSO %d (%p);",
1463                       CurrentProc, tso, tso->id));
1464 #elif defined(PAR)
1465     IF_PAR_DEBUG(verbose,
1466                  belch("==__ schedule: Created TSO %d (%p); %d threads active",
1467                        tso->id, tso, advisory_thread_count));
1468 #else
1469   IF_DEBUG(scheduler,sched_belch("created thread %ld, stack size = %lx words",
1470                                  tso->id, tso->stack_size));
1471 #endif
1472   return tso;
1473 }
1474
1475 /*
1476   Turn a spark into a thread.
1477   ToDo: fix for SMP (needs to acquire SCHED_MUTEX!)
1478 */
1479 #if defined(PAR)
1480 //@cindex activateSpark
1481 StgTSO *
1482 activateSpark (rtsSpark spark)
1483 {
1484   StgTSO *tso;
1485
1486   ASSERT(spark != (rtsSpark)NULL);
1487   tso = createThread_(RtsFlags.GcFlags.initialStkSize, rtsTrue);
1488   if (tso!=END_TSO_QUEUE) {
1489     pushClosure(tso,spark);
1490     PUSH_ON_RUN_QUEUE(tso);
1491     advisory_thread_count++;
1492
1493     if (RtsFlags.ParFlags.ParStats.Full) {
1494       //ASSERT(run_queue_hd == END_TSO_QUEUE); // I think ...
1495       IF_PAR_DEBUG(verbose,
1496                    belch("==^^ activateSpark: turning spark of closure %p (%s) into a thread",
1497                          (StgClosure *)spark, info_type((StgClosure *)spark)));
1498     }
1499   } else {
1500     barf("activateSpark: Cannot create TSO");
1501   }
1502   // ToDo: fwd info on local/global spark to thread -- HWL
1503   // tso->gran.exported =  spark->exported;
1504   // tso->gran.locked =   !spark->global;
1505   // tso->gran.sparkname = spark->name;
1506
1507   return tso;
1508 }
1509 #endif
1510
1511 /* ---------------------------------------------------------------------------
1512  * scheduleThread()
1513  *
1514  * scheduleThread puts a thread on the head of the runnable queue.
1515  * This will usually be done immediately after a thread is created.
1516  * The caller of scheduleThread must create the thread using e.g.
1517  * createThread and push an appropriate closure
1518  * on this thread's stack before the scheduler is invoked.
1519  * ------------------------------------------------------------------------ */
1520
1521 void
1522 scheduleThread(StgTSO *tso)
1523 {
1524   if (tso==END_TSO_QUEUE){
1525     schedule();
1526     return;
1527   }
1528
1529   ACQUIRE_LOCK(&sched_mutex);
1530
1531   /* Put the new thread on the head of the runnable queue.  The caller
1532    * better push an appropriate closure on this thread's stack
1533    * beforehand.  In the SMP case, the thread may start running as
1534    * soon as we release the scheduler lock below.
1535    */
1536   PUSH_ON_RUN_QUEUE(tso);
1537   THREAD_RUNNABLE();
1538
1539 #if 0
1540   IF_DEBUG(scheduler,printTSO(tso));
1541 #endif
1542   RELEASE_LOCK(&sched_mutex);
1543 }
1544
1545 /* ---------------------------------------------------------------------------
1546  * startTasks()
1547  *
1548  * Start up Posix threads to run each of the scheduler tasks.
1549  * I believe the task ids are not needed in the system as defined.
1550  *  KH @ 25/10/99
1551  * ------------------------------------------------------------------------ */
1552
1553 #if defined(PAR) || defined(SMP)
1554 void *
1555 taskStart( void *arg STG_UNUSED )
1556 {
1557   rts_evalNothing(NULL);
1558 }
1559 #endif
1560
1561 /* ---------------------------------------------------------------------------
1562  * initScheduler()
1563  *
1564  * Initialise the scheduler.  This resets all the queues - if the
1565  * queues contained any threads, they'll be garbage collected at the
1566  * next pass.
1567  *
1568  * This now calls startTasks(), so should only be called once!  KH @ 25/10/99
1569  * ------------------------------------------------------------------------ */
1570
1571 #ifdef SMP
1572 static void
1573 term_handler(int sig STG_UNUSED)
1574 {
1575   stat_workerStop();
1576   ACQUIRE_LOCK(&term_mutex);
1577   await_death--;
1578   RELEASE_LOCK(&term_mutex);
1579   pthread_exit(NULL);
1580 }
1581 #endif
1582
1583 //@cindex initScheduler
1584 void
1585 initScheduler(void)
1586 {
1587 #if defined(GRAN)
1588   nat i;
1589
1590   for (i=0; i<=MAX_PROC; i++) {
1591     run_queue_hds[i]      = END_TSO_QUEUE;
1592     run_queue_tls[i]      = END_TSO_QUEUE;
1593     blocked_queue_hds[i]  = END_TSO_QUEUE;
1594     blocked_queue_tls[i]  = END_TSO_QUEUE;
1595     ccalling_threadss[i]  = END_TSO_QUEUE;
1596     sleeping_queue        = END_TSO_QUEUE;
1597   }
1598 #else
1599   run_queue_hd      = END_TSO_QUEUE;
1600   run_queue_tl      = END_TSO_QUEUE;
1601   blocked_queue_hd  = END_TSO_QUEUE;
1602   blocked_queue_tl  = END_TSO_QUEUE;
1603   sleeping_queue    = END_TSO_QUEUE;
1604 #endif
1605
1606   suspended_ccalling_threads  = END_TSO_QUEUE;
1607
1608   main_threads = NULL;
1609   all_threads  = END_TSO_QUEUE;
1610
1611   context_switch = 0;
1612   interrupted    = 0;
1613
1614   RtsFlags.ConcFlags.ctxtSwitchTicks =
1615       RtsFlags.ConcFlags.ctxtSwitchTime / TICK_MILLISECS;
1616
1617 #ifdef INTERPRETER
1618   ecafList = END_ECAF_LIST;
1619   clearECafTable();
1620 #endif
1621
1622   /* Install the SIGHUP handler */
1623 #ifdef SMP
1624   {
1625     struct sigaction action,oact;
1626
1627     action.sa_handler = term_handler;
1628     sigemptyset(&action.sa_mask);
1629     action.sa_flags = 0;
1630     if (sigaction(SIGTERM, &action, &oact) != 0) {
1631       barf("can't install TERM handler");
1632     }
1633   }
1634 #endif
1635
1636 #ifdef SMP
1637   /* Allocate N Capabilities */
1638   {
1639     nat i;
1640     Capability *cap, *prev;
1641     cap  = NULL;
1642     prev = NULL;
1643     for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
1644       cap = stgMallocBytes(sizeof(Capability), "initScheduler:capabilities");
1645       cap->link = prev;
1646       prev = cap;
1647     }
1648     free_capabilities = cap;
1649     n_free_capabilities = RtsFlags.ParFlags.nNodes;
1650   }
1651   IF_DEBUG(scheduler,fprintf(stderr,"scheduler: Allocated %d capabilities\n",
1652                              n_free_capabilities););
1653 #endif
1654
1655 #if defined(SMP) || defined(PAR)
1656   initSparkPools();
1657 #endif
1658 }
1659
1660 #ifdef SMP
1661 void
1662 startTasks( void )
1663 {
1664   nat i;
1665   int r;
1666   pthread_t tid;
1667
1668   /* make some space for saving all the thread ids */
1669   task_ids = stgMallocBytes(RtsFlags.ParFlags.nNodes * sizeof(task_info),
1670                             "initScheduler:task_ids");
1671
1672   /* and create all the threads */
1673   for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
1674     r = pthread_create(&tid,NULL,taskStart,NULL);
1675     if (r != 0) {
1676       barf("startTasks: Can't create new Posix thread");
1677     }
1678     task_ids[i].id = tid;
1679     task_ids[i].mut_time = 0.0;
1680     task_ids[i].mut_etime = 0.0;
1681     task_ids[i].gc_time = 0.0;
1682     task_ids[i].gc_etime = 0.0;
1683     task_ids[i].elapsedtimestart = elapsedtime();
1684     IF_DEBUG(scheduler,fprintf(stderr,"scheduler: Started task: %ld\n",tid););
1685   }
1686 }
1687 #endif
1688
1689 void
1690 exitScheduler( void )
1691 {
1692 #ifdef SMP
1693   nat i;
1694
1695   /* Don't want to use pthread_cancel, since we'd have to install
1696    * these silly exception handlers (pthread_cleanup_{push,pop}) around
1697    * all our locks.
1698    */
1699 #if 0
1700   /* Cancel all our tasks */
1701   for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
1702     pthread_cancel(task_ids[i].id);
1703   }
1704
1705   /* Wait for all the tasks to terminate */
1706   for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
1707     IF_DEBUG(scheduler,fprintf(stderr,"scheduler: waiting for task %ld\n",
1708                                task_ids[i].id));
1709     pthread_join(task_ids[i].id, NULL);
1710   }
1711 #endif
1712
1713   /* Send 'em all a SIGHUP.  That should shut 'em up.
1714    */
1715   await_death = RtsFlags.ParFlags.nNodes;
1716   for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
1717     pthread_kill(task_ids[i].id,SIGTERM);
1718   }
1719   while (await_death > 0) {
1720     sched_yield();
1721   }
1722 #endif
1723 }
1724
1725 /* -----------------------------------------------------------------------------
1726    Managing the per-task allocation areas.
1727
1728    Each capability comes with an allocation area.  These are
1729    fixed-length block lists into which allocation can be done.
1730
1731    ToDo: no support for two-space collection at the moment???
1732    -------------------------------------------------------------------------- */
1733
1734 /* -----------------------------------------------------------------------------
1735  * waitThread is the external interface for running a new computation
1736  * and waiting for the result.
1737  *
1738  * In the non-SMP case, we create a new main thread, push it on the
1739  * main-thread stack, and invoke the scheduler to run it.  The
1740  * scheduler will return when the top main thread on the stack has
1741  * completed or died, and fill in the necessary fields of the
1742  * main_thread structure.
1743  *
1744  * In the SMP case, we create a main thread as before, but we then
1745  * create a new condition variable and sleep on it.  When our new
1746  * main thread has completed, we'll be woken up and the status/result
1747  * will be in the main_thread struct.
1748  * -------------------------------------------------------------------------- */
1749
1750 int
1751 howManyThreadsAvail ( void )
1752 {
1753    int i = 0;
1754    StgTSO* q;
1755    for (q = run_queue_hd; q != END_TSO_QUEUE; q = q->link)
1756       i++;
1757    for (q = blocked_queue_hd; q != END_TSO_QUEUE; q = q->link)
1758       i++;
1759    for (q = sleeping_queue; q != END_TSO_QUEUE; q = q->link)
1760       i++;
1761    return i;
1762 }
1763
1764 void
1765 finishAllThreads ( void )
1766 {
1767    do {
1768       while (run_queue_hd != END_TSO_QUEUE) {
1769          waitThread ( run_queue_hd, NULL );
1770       }
1771       while (blocked_queue_hd != END_TSO_QUEUE) {
1772          waitThread ( blocked_queue_hd, NULL );
1773       }
1774       while (sleeping_queue != END_TSO_QUEUE) {
1775          waitThread ( blocked_queue_hd, NULL );
1776       }
1777    } while
1778       (blocked_queue_hd != END_TSO_QUEUE ||
1779        run_queue_hd     != END_TSO_QUEUE ||
1780        sleeping_queue   != END_TSO_QUEUE);
1781 }
1782
1783 SchedulerStatus
1784 waitThread(StgTSO *tso, /*out*/StgClosure **ret)
1785 {
1786   StgMainThread *m;
1787   SchedulerStatus stat;
1788
1789   ACQUIRE_LOCK(&sched_mutex);
1790
1791   m = stgMallocBytes(sizeof(StgMainThread), "waitThread");
1792
1793   m->tso = tso;
1794   m->ret = ret;
1795   m->stat = NoStatus;
1796 #ifdef SMP
1797   pthread_cond_init(&m->wakeup, NULL);
1798 #endif
1799
1800   m->link = main_threads;
1801   main_threads = m;
1802
1803   IF_DEBUG(scheduler, fprintf(stderr, "scheduler: new main thread (%d)\n",
1804                               m->tso->id));
1805
1806 #ifdef SMP
1807   do {
1808     pthread_cond_wait(&m->wakeup, &sched_mutex);
1809   } while (m->stat == NoStatus);
1810 #elif defined(GRAN)
1811   /* GranSim specific init */
1812   CurrentTSO = m->tso;                // the TSO to run
1813   procStatus[MainProc] = Busy;        // status of main PE
1814   CurrentProc = MainProc;             // PE to run it on
1815
1816   schedule();
1817 #else
1818   schedule();
1819   ASSERT(m->stat != NoStatus);
1820 #endif
1821
1822   stat = m->stat;
1823
1824 #ifdef SMP
1825   pthread_cond_destroy(&m->wakeup);
1826 #endif
1827
1828   IF_DEBUG(scheduler, fprintf(stderr, "scheduler: main thread (%d) finished\n",
1829                               m->tso->id));
1830   free(m);
1831
1832   RELEASE_LOCK(&sched_mutex);
1833
1834   return stat;
1835 }
1836
1837 //@node Run queue code, Garbage Collextion Routines, Suspend and Resume, Main scheduling code
1838 //@subsection Run queue code
1839
1840 #if 0
1841 /*
1842    NB: In GranSim we have many run queues; run_queue_hd is actually a macro
1843        unfolding to run_queue_hds[CurrentProc], thus CurrentProc is an
1844        implicit global variable that has to be correct when calling these
1845        fcts -- HWL
1846 */
1847
1848 /* Put the new thread on the head of the runnable queue.
1849  * The caller of createThread better push an appropriate closure
1850  * on this thread's stack before the scheduler is invoked.
1851  */
1852 static /* inline */ void
1853 add_to_run_queue(tso)
1854 StgTSO* tso;
1855 {
1856   ASSERT(tso!=run_queue_hd && tso!=run_queue_tl);
1857   tso->link = run_queue_hd;
1858   run_queue_hd = tso;
1859   if (run_queue_tl == END_TSO_QUEUE) {
1860     run_queue_tl = tso;
1861   }
1862 }
1863
1864 /* Put the new thread at the end of the runnable queue. */
1865 static /* inline */ void
1866 push_on_run_queue(tso)
1867 StgTSO* tso;
1868 {
1869   ASSERT(get_itbl((StgClosure *)tso)->type == TSO);
1870   ASSERT(run_queue_hd!=NULL && run_queue_tl!=NULL);
1871   ASSERT(tso!=run_queue_hd && tso!=run_queue_tl);
1872   if (run_queue_hd == END_TSO_QUEUE) {
1873     run_queue_hd = tso;
1874   } else {
1875     run_queue_tl->link = tso;
1876   }
1877   run_queue_tl = tso;
1878 }
1879
1880 /*
1881    Should be inlined because it's used very often in schedule.  The tso
1882    argument is actually only needed in GranSim, where we want to have the
1883    possibility to schedule *any* TSO on the run queue, irrespective of the
1884    actual ordering. Therefore, if tso is not the nil TSO then we traverse
1885    the run queue and dequeue the tso, adjusting the links in the queue.
1886 */
1887 //@cindex take_off_run_queue
1888 static /* inline */ StgTSO*
1889 take_off_run_queue(StgTSO *tso) {
1890   StgTSO *t, *prev;
1891
1892   /*
1893      qetlaHbogh Qu' ngaSbogh ghomDaQ {tso} yIteq!
1894
1895      if tso is specified, unlink that tso from the run_queue (doesn't have
1896      to be at the beginning of the queue); GranSim only
1897   */
1898   if (tso!=END_TSO_QUEUE) {
1899     /* find tso in queue */
1900     for (t=run_queue_hd, prev=END_TSO_QUEUE;
1901          t!=END_TSO_QUEUE && t!=tso;
1902          prev=t, t=t->link)
1903       /* nothing */ ;
1904     ASSERT(t==tso);
1905     /* now actually dequeue the tso */
1906     if (prev!=END_TSO_QUEUE) {
1907       ASSERT(run_queue_hd!=t);
1908       prev->link = t->link;
1909     } else {
1910       /* t is at beginning of thread queue */
1911       ASSERT(run_queue_hd==t);
1912       run_queue_hd = t->link;
1913     }
1914     /* t is at end of thread queue */
1915     if (t->link==END_TSO_QUEUE) {
1916       ASSERT(t==run_queue_tl);
1917       run_queue_tl = prev;
1918     } else {
1919       ASSERT(run_queue_tl!=t);
1920     }
1921     t->link = END_TSO_QUEUE;
1922   } else {
1923     /* take tso from the beginning of the queue; std concurrent code */
1924     t = run_queue_hd;
1925     if (t != END_TSO_QUEUE) {
1926       run_queue_hd = t->link;
1927       t->link = END_TSO_QUEUE;
1928       if (run_queue_hd == END_TSO_QUEUE) {
1929         run_queue_tl = END_TSO_QUEUE;
1930       }
1931     }
1932   }
1933   return t;
1934 }
1935
1936 #endif /* 0 */
1937
1938 //@node Garbage Collextion Routines, Blocking Queue Routines, Run queue code, Main scheduling code
1939 //@subsection Garbage Collextion Routines
1940
1941 /* ---------------------------------------------------------------------------
1942    Where are the roots that we know about?
1943
1944         - all the threads on the runnable queue
1945         - all the threads on the blocked queue
1946         - all the threads on the sleeping queue
1947         - all the thread currently executing a _ccall_GC
1948         - all the "main threads"
1949
1950    ------------------------------------------------------------------------ */
1951
1952 /* This has to be protected either by the scheduler monitor, or by the
1953         garbage collection monitor (probably the latter).
1954         KH @ 25/10/99
1955 */
1956
1957 static void GetRoots(void)
1958 {
1959   StgMainThread *m;
1960
1961 #if defined(GRAN)
1962   {
1963     nat i;
1964     for (i=0; i<=RtsFlags.GranFlags.proc; i++) {
1965       if ((run_queue_hds[i] != END_TSO_QUEUE) && ((run_queue_hds[i] != NULL)))
1966         run_queue_hds[i]    = (StgTSO *)MarkRoot((StgClosure *)run_queue_hds[i]);
1967       if ((run_queue_tls[i] != END_TSO_QUEUE) && ((run_queue_tls[i] != NULL)))
1968         run_queue_tls[i]    = (StgTSO *)MarkRoot((StgClosure *)run_queue_tls[i]);
1969
1970       if ((blocked_queue_hds[i] != END_TSO_QUEUE) && ((blocked_queue_hds[i] != NULL)))
1971         blocked_queue_hds[i] = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_hds[i]);
1972       if ((blocked_queue_tls[i] != END_TSO_QUEUE) && ((blocked_queue_tls[i] != NULL)))
1973         blocked_queue_tls[i] = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_tls[i]);
1974       if ((ccalling_threadss[i] != END_TSO_QUEUE) && ((ccalling_threadss[i] != NULL)))
1975         ccalling_threadss[i] = (StgTSO *)MarkRoot((StgClosure *)ccalling_threadss[i]);
1976     }
1977   }
1978
1979   markEventQueue();
1980
1981 #else /* !GRAN */
1982   if (run_queue_hd != END_TSO_QUEUE) {
1983     ASSERT(run_queue_tl != END_TSO_QUEUE);
1984     run_queue_hd      = (StgTSO *)MarkRoot((StgClosure *)run_queue_hd);
1985     run_queue_tl      = (StgTSO *)MarkRoot((StgClosure *)run_queue_tl);
1986   }
1987
1988   if (blocked_queue_hd != END_TSO_QUEUE) {
1989     ASSERT(blocked_queue_tl != END_TSO_QUEUE);
1990     blocked_queue_hd  = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_hd);
1991     blocked_queue_tl  = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_tl);
1992   }
1993
1994   if (sleeping_queue != END_TSO_QUEUE) {
1995     sleeping_queue  = (StgTSO *)MarkRoot((StgClosure *)sleeping_queue);
1996   }
1997 #endif
1998
1999   for (m = main_threads; m != NULL; m = m->link) {
2000     m->tso = (StgTSO *)MarkRoot((StgClosure *)m->tso);
2001   }
2002   if (suspended_ccalling_threads != END_TSO_QUEUE)
2003     suspended_ccalling_threads =
2004       (StgTSO *)MarkRoot((StgClosure *)suspended_ccalling_threads);
2005
2006 #if defined(SMP) || defined(PAR) || defined(GRAN)
2007   markSparkQueue();
2008 #endif
2009
2010 #if defined(GHCI)
2011   markCafs();
2012 #endif
2013 }
2014
2015 /* -----------------------------------------------------------------------------
2016    performGC
2017
2018    This is the interface to the garbage collector from Haskell land.
2019    We provide this so that external C code can allocate and garbage
2020    collect when called from Haskell via _ccall_GC.
2021
2022    It might be useful to provide an interface whereby the programmer
2023    can specify more roots (ToDo).
2024
2025    This needs to be protected by the GC condition variable above.  KH.
2026    -------------------------------------------------------------------------- */
2027
2028 void (*extra_roots)(void);
2029
2030 void
2031 performGC(void)
2032 {
2033   GarbageCollect(GetRoots,rtsFalse);
2034 }
2035
2036 void
2037 performMajorGC(void)
2038 {
2039   GarbageCollect(GetRoots,rtsTrue);
2040 }
2041
2042 static void
2043 AllRoots(void)
2044 {
2045   GetRoots();                   /* the scheduler's roots */
2046   extra_roots();                /* the user's roots */
2047 }
2048
2049 void
2050 performGCWithRoots(void (*get_roots)(void))
2051 {
2052   extra_roots = get_roots;
2053
2054   GarbageCollect(AllRoots,rtsFalse);
2055 }
2056
2057 /* -----------------------------------------------------------------------------
2058    Stack overflow
2059
2060    If the thread has reached its maximum stack size, then raise the
2061    StackOverflow exception in the offending thread.  Otherwise
2062    relocate the TSO into a larger chunk of memory and adjust its stack
2063    size appropriately.
2064    -------------------------------------------------------------------------- */
2065
2066 static StgTSO *
2067 threadStackOverflow(StgTSO *tso)
2068 {
2069   nat new_stack_size, new_tso_size, diff, stack_words;
2070   StgPtr new_sp;
2071   StgTSO *dest;
2072
2073   IF_DEBUG(sanity,checkTSO(tso));
2074   if (tso->stack_size >= tso->max_stack_size) {
2075
2076     IF_DEBUG(gc,
2077              belch("@@ threadStackOverflow of TSO %d (%p): stack too large (now %ld; max is %ld",
2078                    tso->id, tso, tso->stack_size, tso->max_stack_size);
2079              /* If we're debugging, just print out the top of the stack */
2080              printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size,
2081                                               tso->sp+64)));
2082
2083 #ifdef INTERPRETER
2084     fprintf(stderr, "fatal: stack overflow in Hugs; aborting\n" );
2085     exit(1);
2086 #else
2087     /* Send this thread the StackOverflow exception */
2088     raiseAsync(tso, (StgClosure *)stackOverflow_closure);
2089 #endif
2090     return tso;
2091   }
2092
2093   /* Try to double the current stack size.  If that takes us over the
2094    * maximum stack size for this thread, then use the maximum instead.
2095    * Finally round up so the TSO ends up as a whole number of blocks.
2096    */
2097   new_stack_size = stg_min(tso->stack_size * 2, tso->max_stack_size);
2098   new_tso_size   = (nat)BLOCK_ROUND_UP(new_stack_size * sizeof(W_) +
2099                                        TSO_STRUCT_SIZE)/sizeof(W_);
2100   new_tso_size = round_to_mblocks(new_tso_size);  /* Be MBLOCK-friendly */
2101   new_stack_size = new_tso_size - TSO_STRUCT_SIZEW;
2102
2103   IF_DEBUG(scheduler, fprintf(stderr,"scheduler: increasing stack size from %d words to %d.\n", tso->stack_size, new_stack_size));
2104
2105   dest = (StgTSO *)allocate(new_tso_size);
2106   TICK_ALLOC_TSO(new_tso_size-sizeofW(StgTSO),0);
2107
2108   /* copy the TSO block and the old stack into the new area */
2109   memcpy(dest,tso,TSO_STRUCT_SIZE);
2110   stack_words = tso->stack + tso->stack_size - tso->sp;
2111   new_sp = (P_)dest + new_tso_size - stack_words;
2112   memcpy(new_sp, tso->sp, stack_words * sizeof(W_));
2113
2114   /* relocate the stack pointers... */
2115   diff = (P_)new_sp - (P_)tso->sp; /* In *words* */
2116   dest->su    = (StgUpdateFrame *) ((P_)dest->su + diff);
2117   dest->sp    = new_sp;
2118   dest->stack_size = new_stack_size;
2119
2120   /* and relocate the update frame list */
2121   relocate_TSO(tso, dest);
2122
2123   /* Mark the old TSO as relocated.  We have to check for relocated
2124    * TSOs in the garbage collector and any primops that deal with TSOs.
2125    *
2126    * It's important to set the sp and su values to just beyond the end
2127    * of the stack, so we don't attempt to scavenge any part of the
2128    * dead TSO's stack.
2129    */
2130   tso->what_next = ThreadRelocated;
2131   tso->link = dest;
2132   tso->sp = (P_)&(tso->stack[tso->stack_size]);
2133   tso->su = (StgUpdateFrame *)tso->sp;
2134   tso->why_blocked = NotBlocked;
2135   dest->mut_link = NULL;
2136
2137   IF_PAR_DEBUG(verbose,
2138                belch("@@ threadStackOverflow of TSO %d (now at %p): stack size increased to %ld",
2139                      tso->id, tso, tso->stack_size);
2140                /* If we're debugging, just print out the top of the stack */
2141                printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size,
2142                                                 tso->sp+64)));
2143
2144   IF_DEBUG(sanity,checkTSO(tso));
2145 #if 0
2146   IF_DEBUG(scheduler,printTSO(dest));
2147 #endif
2148
2149   return dest;
2150 }
2151
2152 //@node Blocking Queue Routines, Exception Handling Routines, Garbage Collextion Routines, Main scheduling code
2153 //@subsection Blocking Queue Routines
2154
2155 /* ---------------------------------------------------------------------------
2156    Wake up a queue that was blocked on some resource.
2157    ------------------------------------------------------------------------ */
2158
2159 #if defined(GRAN)
2160 static inline void
2161 unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node )
2162 {
2163 }
2164 #elif defined(PAR)
2165 static inline void
2166 unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node )
2167 {
2168   /* write RESUME events to log file and
2169      update blocked and fetch time (depending on type of the orig closure) */
2170   if (RtsFlags.ParFlags.ParStats.Full) {
2171     DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
2172                      GR_RESUME, ((StgTSO *)bqe), ((StgTSO *)bqe)->block_info.closure,
2173                      0, 0 /* spark_queue_len(ADVISORY_POOL) */);
2174
2175     switch (get_itbl(node)->type) {
2176         case FETCH_ME_BQ:
2177           ((StgTSO *)bqe)->par.fetchtime += CURRENT_TIME-((StgTSO *)bqe)->par.blockedat;
2178           break;
2179         case RBH:
2180         case FETCH_ME:
2181         case BLACKHOLE_BQ:
2182           ((StgTSO *)bqe)->par.blocktime += CURRENT_TIME-((StgTSO *)bqe)->par.blockedat;
2183           break;
2184         default:
2185           barf("{unblockOneLocked}Daq Qagh: unexpected closure in blocking queue");
2186         }
2187       }
2188 }
2189 #endif
2190
2191 #if defined(GRAN)
2192 static StgBlockingQueueElement *
2193 unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node)
2194 {
2195     StgTSO *tso;
2196     PEs node_loc, tso_loc;
2197
2198     node_loc = where_is(node); // should be lifted out of loop
2199     tso = (StgTSO *)bqe;  // wastes an assignment to get the type right
2200     tso_loc = where_is((StgClosure *)tso);
2201     if (IS_LOCAL_TO(PROCS(node),tso_loc)) { // TSO is local
2202       /* !fake_fetch => TSO is on CurrentProc is same as IS_LOCAL_TO */
2203       ASSERT(CurrentProc!=node_loc || tso_loc==CurrentProc);
2204       CurrentTime[CurrentProc] += RtsFlags.GranFlags.Costs.lunblocktime;
2205       // insertThread(tso, node_loc);
2206       new_event(tso_loc, tso_loc, CurrentTime[CurrentProc],
2207                 ResumeThread,
2208                 tso, node, (rtsSpark*)NULL);
2209       tso->link = END_TSO_QUEUE; // overwrite link just to be sure
2210       // len_local++;
2211       // len++;
2212     } else { // TSO is remote (actually should be FMBQ)
2213       CurrentTime[CurrentProc] += RtsFlags.GranFlags.Costs.mpacktime +
2214                                   RtsFlags.GranFlags.Costs.gunblocktime +
2215                                   RtsFlags.GranFlags.Costs.latency;
2216       new_event(tso_loc, CurrentProc, CurrentTime[CurrentProc],
2217                 UnblockThread,
2218                 tso, node, (rtsSpark*)NULL);
2219       tso->link = END_TSO_QUEUE; // overwrite link just to be sure
2220       // len++;
2221     }
2222     /* the thread-queue-overhead is accounted for in either Resume or UnblockThread */
2223     IF_GRAN_DEBUG(bq,
2224                   fprintf(stderr," %s TSO %d (%p) [PE %d] (block_info.closure=%p) (next=%p) ,",
2225                           (node_loc==tso_loc ? "Local" : "Global"),
2226                           tso->id, tso, CurrentProc, tso->block_info.closure, tso->link));
2227     tso->block_info.closure = NULL;
2228     IF_DEBUG(scheduler,belch("-- Waking up thread %ld (%p)",
2229                              tso->id, tso));
2230 }
2231 #elif defined(PAR)
2232 static StgBlockingQueueElement *
2233 unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node)
2234 {
2235     StgBlockingQueueElement *next;
2236
2237     switch (get_itbl(bqe)->type) {
2238     case TSO:
2239       ASSERT(((StgTSO *)bqe)->why_blocked != NotBlocked);
2240       /* if it's a TSO just push it onto the run_queue */
2241       next = bqe->link;
2242       // ((StgTSO *)bqe)->link = END_TSO_QUEUE; // debugging?
2243       PUSH_ON_RUN_QUEUE((StgTSO *)bqe);
2244       THREAD_RUNNABLE();
2245       unblockCount(bqe, node);
2246       /* reset blocking status after dumping event */
2247       ((StgTSO *)bqe)->why_blocked = NotBlocked;
2248       break;
2249
2250     case BLOCKED_FETCH:
2251       /* if it's a BLOCKED_FETCH put it on the PendingFetches list */
2252       next = bqe->link;
2253       bqe->link = PendingFetches;
2254       PendingFetches = bqe;
2255       break;
2256
2257 # if defined(DEBUG)
2258       /* can ignore this case in a non-debugging setup;
2259          see comments on RBHSave closures above */
2260     case CONSTR:
2261       /* check that the closure is an RBHSave closure */
2262       ASSERT(get_itbl((StgClosure *)bqe) == &stg_RBH_Save_0_info ||
2263              get_itbl((StgClosure *)bqe) == &stg_RBH_Save_1_info ||
2264              get_itbl((StgClosure *)bqe) == &stg_RBH_Save_2_info);
2265       break;
2266
2267     default:
2268       barf("{unblockOneLocked}Daq Qagh: Unexpected IP (%#lx; %s) in blocking queue at %#lx\n",
2269            get_itbl((StgClosure *)bqe), info_type((StgClosure *)bqe),
2270            (StgClosure *)bqe);
2271 # endif
2272     }
2273   // IF_DEBUG(scheduler,sched_belch("waking up thread %ld", tso->id));
2274   return next;
2275 }
2276
2277 #else /* !GRAN && !PAR */
2278 static StgTSO *
2279 unblockOneLocked(StgTSO *tso)
2280 {
2281   StgTSO *next;
2282
2283   ASSERT(get_itbl(tso)->type == TSO);
2284   ASSERT(tso->why_blocked != NotBlocked);
2285   tso->why_blocked = NotBlocked;
2286   next = tso->link;
2287   PUSH_ON_RUN_QUEUE(tso);
2288   THREAD_RUNNABLE();
2289   IF_DEBUG(scheduler,sched_belch("waking up thread %ld", tso->id));
2290   return next;
2291 }
2292 #endif
2293
2294 #if defined(GRAN) || defined(PAR)
2295 inline StgBlockingQueueElement *
2296 unblockOne(StgBlockingQueueElement *bqe, StgClosure *node)
2297 {
2298   ACQUIRE_LOCK(&sched_mutex);
2299   bqe = unblockOneLocked(bqe, node);
2300   RELEASE_LOCK(&sched_mutex);
2301   return bqe;
2302 }
2303 #else
2304 inline StgTSO *
2305 unblockOne(StgTSO *tso)
2306 {
2307   ACQUIRE_LOCK(&sched_mutex);
2308   tso = unblockOneLocked(tso);
2309   RELEASE_LOCK(&sched_mutex);
2310   return tso;
2311 }
2312 #endif
2313
2314 #if defined(GRAN)
2315 void
2316 awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node)
2317 {
2318   StgBlockingQueueElement *bqe;
2319   PEs node_loc;
2320   nat len = 0;
2321
2322   IF_GRAN_DEBUG(bq,
2323                 belch("## AwBQ for node %p on PE %d @ %ld by TSO %d (%p): ", \
2324                       node, CurrentProc, CurrentTime[CurrentProc],
2325                       CurrentTSO->id, CurrentTSO));
2326
2327   node_loc = where_is(node);
2328
2329   ASSERT(get_itbl(q)->type == TSO ||   // q is either a TSO or an RBHSave
2330          get_itbl(q)->type == CONSTR); // closure (type constructor)
2331   ASSERT(is_unique(node));
2332
2333   /* FAKE FETCH: magically copy the node to the tso's proc;
2334      no Fetch necessary because in reality the node should not have been
2335      moved to the other PE in the first place
2336   */
2337   if (CurrentProc!=node_loc) {
2338     IF_GRAN_DEBUG(bq,
2339                   belch("## node %p is on PE %d but CurrentProc is %d (TSO %d); assuming fake fetch and adjusting bitmask (old: %#x)",
2340                         node, node_loc, CurrentProc, CurrentTSO->id,
2341                         // CurrentTSO, where_is(CurrentTSO),
2342                         node->header.gran.procs));
2343     node->header.gran.procs = (node->header.gran.procs) | PE_NUMBER(CurrentProc);
2344     IF_GRAN_DEBUG(bq,
2345                   belch("## new bitmask of node %p is %#x",
2346                         node, node->header.gran.procs));
2347     if (RtsFlags.GranFlags.GranSimStats.Global) {
2348       globalGranStats.tot_fake_fetches++;
2349     }
2350   }
2351
2352   bqe = q;
2353   // ToDo: check: ASSERT(CurrentProc==node_loc);
2354   while (get_itbl(bqe)->type==TSO) { // q != END_TSO_QUEUE) {
2355     //next = bqe->link;
2356     /*
2357        bqe points to the current element in the queue
2358        next points to the next element in the queue
2359     */
2360     //tso = (StgTSO *)bqe;  // wastes an assignment to get the type right
2361     //tso_loc = where_is(tso);
2362     len++;
2363     bqe = unblockOneLocked(bqe, node);
2364   }
2365
2366   /* if this is the BQ of an RBH, we have to put back the info ripped out of
2367      the closure to make room for the anchor of the BQ */
2368   if (bqe!=END_BQ_QUEUE) {
2369     ASSERT(get_itbl(node)->type == RBH && get_itbl(bqe)->type == CONSTR);
2370     /*
2371     ASSERT((info_ptr==&RBH_Save_0_info) ||
2372            (info_ptr==&RBH_Save_1_info) ||
2373            (info_ptr==&RBH_Save_2_info));
2374     */
2375     /* cf. convertToRBH in RBH.c for writing the RBHSave closure */
2376     ((StgRBH *)node)->blocking_queue = (StgBlockingQueueElement *)((StgRBHSave *)bqe)->payload[0];
2377     ((StgRBH *)node)->mut_link       = (StgMutClosure *)((StgRBHSave *)bqe)->payload[1];
2378
2379     IF_GRAN_DEBUG(bq,
2380                   belch("## Filled in RBH_Save for %p (%s) at end of AwBQ",
2381                         node, info_type(node)));
2382   }
2383
2384   /* statistics gathering */
2385   if (RtsFlags.GranFlags.GranSimStats.Global) {
2386     // globalGranStats.tot_bq_processing_time += bq_processing_time;
2387     globalGranStats.tot_bq_len += len;      // total length of all bqs awakened
2388     // globalGranStats.tot_bq_len_local += len_local;  // same for local TSOs only
2389     globalGranStats.tot_awbq++;             // total no. of bqs awakened
2390   }
2391   IF_GRAN_DEBUG(bq,
2392                 fprintf(stderr,"## BQ Stats of %p: [%d entries] %s\n",
2393                         node, len, (bqe!=END_BQ_QUEUE) ? "RBH" : ""));
2394 }
2395 #elif defined(PAR)
2396 void
2397 awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node)
2398 {
2399   StgBlockingQueueElement *bqe, *next;
2400
2401   ACQUIRE_LOCK(&sched_mutex);
2402
2403   IF_PAR_DEBUG(verbose,
2404                belch("## AwBQ for node %p on [%x]: ",
2405                      node, mytid));
2406
2407   ASSERT(get_itbl(q)->type == TSO ||
2408          get_itbl(q)->type == BLOCKED_FETCH ||
2409          get_itbl(q)->type == CONSTR);
2410
2411   bqe = q;
2412   while (get_itbl(bqe)->type==TSO ||
2413          get_itbl(bqe)->type==BLOCKED_FETCH) {
2414     bqe = unblockOneLocked(bqe, node);
2415   }
2416   RELEASE_LOCK(&sched_mutex);
2417 }
2418
2419 #else   /* !GRAN && !PAR */
2420 void
2421 awakenBlockedQueue(StgTSO *tso)
2422 {
2423   ACQUIRE_LOCK(&sched_mutex);
2424   while (tso != END_TSO_QUEUE) {
2425     tso = unblockOneLocked(tso);
2426   }
2427   RELEASE_LOCK(&sched_mutex);
2428 }
2429 #endif
2430
2431 //@node Exception Handling Routines, Debugging Routines, Blocking Queue Routines, Main scheduling code
2432 //@subsection Exception Handling Routines
2433
2434 /* ---------------------------------------------------------------------------
2435    Interrupt execution
2436    - usually called inside a signal handler so it mustn't do anything fancy.
2437    ------------------------------------------------------------------------ */
2438
2439 void
2440 interruptStgRts(void)
2441 {
2442     interrupted    = 1;
2443     context_switch = 1;
2444 }
2445
2446 /* -----------------------------------------------------------------------------
2447    Unblock a thread
2448
2449    This is for use when we raise an exception in another thread, which
2450    may be blocked.
2451    This has nothing to do with the UnblockThread event in GranSim. -- HWL
2452    -------------------------------------------------------------------------- */
2453
2454 #if defined(GRAN) || defined(PAR)
2455 /*
2456   NB: only the type of the blocking queue is different in GranSim and GUM
2457       the operations on the queue-elements are the same
2458       long live polymorphism!
2459 */
2460 static void
2461 unblockThread(StgTSO *tso)
2462 {
2463   StgBlockingQueueElement *t, **last;
2464
2465   ACQUIRE_LOCK(&sched_mutex);
2466   switch (tso->why_blocked) {
2467
2468   case NotBlocked:
2469     return;  /* not blocked */
2470
2471   case BlockedOnMVar:
2472     ASSERT(get_itbl(tso->block_info.closure)->type == MVAR);
2473     {
2474       StgBlockingQueueElement *last_tso = END_BQ_QUEUE;
2475       StgMVar *mvar = (StgMVar *)(tso->block_info.closure);
2476
2477       last = (StgBlockingQueueElement **)&mvar->head;
2478       for (t = (StgBlockingQueueElement *)mvar->head;
2479            t != END_BQ_QUEUE;
2480            last = &t->link, last_tso = t, t = t->link) {
2481         if (t == (StgBlockingQueueElement *)tso) {
2482           *last = (StgBlockingQueueElement *)tso->link;
2483           if (mvar->tail == tso) {
2484             mvar->tail = (StgTSO *)last_tso;
2485           }
2486           goto done;
2487         }
2488       }
2489       barf("unblockThread (MVAR): TSO not found");
2490     }
2491
2492   case BlockedOnBlackHole:
2493     ASSERT(get_itbl(tso->block_info.closure)->type == BLACKHOLE_BQ);
2494     {
2495       StgBlockingQueue *bq = (StgBlockingQueue *)(tso->block_info.closure);
2496
2497       last = &bq->blocking_queue;
2498       for (t = bq->blocking_queue;
2499            t != END_BQ_QUEUE;
2500            last = &t->link, t = t->link) {
2501         if (t == (StgBlockingQueueElement *)tso) {
2502           *last = (StgBlockingQueueElement *)tso->link;
2503           goto done;
2504         }
2505       }
2506       barf("unblockThread (BLACKHOLE): TSO not found");
2507     }
2508
2509   case BlockedOnException:
2510     {
2511       StgTSO *target  = tso->block_info.tso;
2512
2513       ASSERT(get_itbl(target)->type == TSO);
2514       ASSERT(target->blocked_exceptions != NULL);
2515
2516       last = (StgBlockingQueueElement **)&target->blocked_exceptions;
2517       for (t = (StgBlockingQueueElement *)target->blocked_exceptions;
2518            t != END_BQ_QUEUE;
2519            last = &t->link, t = t->link) {
2520         ASSERT(get_itbl(t)->type == TSO);
2521         if (t == (StgBlockingQueueElement *)tso) {
2522           *last = (StgBlockingQueueElement *)tso->link;
2523           goto done;
2524         }
2525       }
2526       barf("unblockThread (Exception): TSO not found");
2527     }
2528
2529   case BlockedOnRead:
2530   case BlockedOnWrite:
2531     {
2532       StgBlockingQueueElement *prev = NULL;
2533       for (t = (StgBlockingQueueElement *)blocked_queue_hd; t != END_BQ_QUEUE;
2534            prev = t, t = t->link) {
2535         if (t == (StgBlockingQueueElement *)tso) {
2536           if (prev == NULL) {
2537             blocked_queue_hd = (StgTSO *)t->link;
2538             if ((StgBlockingQueueElement *)blocked_queue_tl == t) {
2539               blocked_queue_tl = END_TSO_QUEUE;
2540             }
2541           } else {
2542             prev->link = t->link;
2543             if ((StgBlockingQueueElement *)blocked_queue_tl == t) {
2544               blocked_queue_tl = (StgTSO *)prev;
2545             }
2546           }
2547           goto done;
2548         }
2549       }
2550       barf("unblockThread (I/O): TSO not found");
2551     }
2552
2553   case BlockedOnDelay:
2554     {
2555       StgBlockingQueueElement *prev = NULL;
2556       for (t = (StgBlockingQueueElement *)sleeping_queue; t != END_BQ_QUEUE;
2557            prev = t, t = t->link) {
2558         if (t == (StgBlockingQueueElement *)tso) {
2559           if (prev == NULL) {
2560             sleeping_queue = (StgTSO *)t->link;
2561           } else {
2562             prev->link = t->link;
2563           }
2564           goto done;
2565         }
2566       }
2567       barf("unblockThread (I/O): TSO not found");
2568     }
2569
2570   default:
2571     barf("unblockThread");
2572   }
2573
2574  done:
2575   tso->link = END_TSO_QUEUE;
2576   tso->why_blocked = NotBlocked;
2577   tso->block_info.closure = NULL;
2578   PUSH_ON_RUN_QUEUE(tso);
2579   RELEASE_LOCK(&sched_mutex);
2580 }
2581 #else
2582 static void
2583 unblockThread(StgTSO *tso)
2584 {
2585   StgTSO *t, **last;
2586
2587   ACQUIRE_LOCK(&sched_mutex);
2588   switch (tso->why_blocked) {
2589
2590   case NotBlocked:
2591     return;  /* not blocked */
2592
2593   case BlockedOnMVar:
2594     ASSERT(get_itbl(tso->block_info.closure)->type == MVAR);
2595     {
2596       StgTSO *last_tso = END_TSO_QUEUE;
2597       StgMVar *mvar = (StgMVar *)(tso->block_info.closure);
2598
2599       last = &mvar->head;
2600       for (t = mvar->head; t != END_TSO_QUEUE;
2601            last = &t->link, last_tso = t, t = t->link) {
2602         if (t == tso) {
2603           *last = tso->link;
2604           if (mvar->tail == tso) {
2605             mvar->tail = last_tso;
2606           }
2607           goto done;
2608         }
2609       }
2610       barf("unblockThread (MVAR): TSO not found");
2611     }
2612
2613   case BlockedOnBlackHole:
2614     ASSERT(get_itbl(tso->block_info.closure)->type == BLACKHOLE_BQ);
2615     {
2616       StgBlockingQueue *bq = (StgBlockingQueue *)(tso->block_info.closure);
2617
2618       last = &bq->blocking_queue;
2619       for (t = bq->blocking_queue; t != END_TSO_QUEUE;
2620            last = &t->link, t = t->link) {
2621         if (t == tso) {
2622           *last = tso->link;
2623           goto done;
2624         }
2625       }
2626       barf("unblockThread (BLACKHOLE): TSO not found");
2627     }
2628
2629   case BlockedOnException:
2630     {
2631       StgTSO *target  = tso->block_info.tso;
2632
2633       ASSERT(get_itbl(target)->type == TSO);
2634       ASSERT(target->blocked_exceptions != NULL);
2635
2636       last = &target->blocked_exceptions;
2637       for (t = target->blocked_exceptions; t != END_TSO_QUEUE;
2638            last = &t->link, t = t->link) {
2639         ASSERT(get_itbl(t)->type == TSO);
2640         if (t == tso) {
2641           *last = tso->link;
2642           goto done;
2643         }
2644       }
2645       barf("unblockThread (Exception): TSO not found");
2646     }
2647
2648   case BlockedOnRead:
2649   case BlockedOnWrite:
2650     {
2651       StgTSO *prev = NULL;
2652       for (t = blocked_queue_hd; t != END_TSO_QUEUE;
2653            prev = t, t = t->link) {
2654         if (t == tso) {
2655           if (prev == NULL) {
2656             blocked_queue_hd = t->link;
2657             if (blocked_queue_tl == t) {
2658               blocked_queue_tl = END_TSO_QUEUE;
2659             }
2660           } else {
2661             prev->link = t->link;
2662             if (blocked_queue_tl == t) {
2663               blocked_queue_tl = prev;
2664             }
2665           }
2666           goto done;
2667         }
2668       }
2669       barf("unblockThread (I/O): TSO not found");
2670     }
2671
2672   case BlockedOnDelay:
2673     {
2674       StgTSO *prev = NULL;
2675       for (t = sleeping_queue; t != END_TSO_QUEUE;
2676            prev = t, t = t->link) {
2677         if (t == tso) {
2678           if (prev == NULL) {
2679             sleeping_queue = t->link;
2680           } else {
2681             prev->link = t->link;
2682           }
2683           goto done;
2684         }
2685       }
2686       barf("unblockThread (I/O): TSO not found");
2687     }
2688
2689   default:
2690     barf("unblockThread");
2691   }
2692
2693  done:
2694   tso->link = END_TSO_QUEUE;
2695   tso->why_blocked = NotBlocked;
2696   tso->block_info.closure = NULL;
2697   PUSH_ON_RUN_QUEUE(tso);
2698   RELEASE_LOCK(&sched_mutex);
2699 }
2700 #endif
2701
2702 /* -----------------------------------------------------------------------------
2703  * raiseAsync()
2704  *
2705  * The following function implements the magic for raising an
2706  * asynchronous exception in an existing thread.
2707  *
2708  * We first remove the thread from any queue on which it might be
2709  * blocked.  The possible blockages are MVARs and BLACKHOLE_BQs.
2710  *
2711  * We strip the stack down to the innermost CATCH_FRAME, building
2712  * thunks in the heap for all the active computations, so they can
2713  * be restarted if necessary.  When we reach a CATCH_FRAME, we build
2714  * an application of the handler to the exception, and push it on
2715  * the top of the stack.
2716  *
2717  * How exactly do we save all the active computations?  We create an
2718  * AP_UPD for every UpdateFrame on the stack.  Entering one of these
2719  * AP_UPDs pushes everything from the corresponding update frame
2720  * upwards onto the stack.  (Actually, it pushes everything up to the
2721  * next update frame plus a pointer to the next AP_UPD object.
2722  * Entering the next AP_UPD object pushes more onto the stack until we
2723  * reach the last AP_UPD object - at which point the stack should look
2724  * exactly as it did when we killed the TSO and we can continue
2725  * execution by entering the closure on top of the stack.
2726  *
2727  * We can also kill a thread entirely - this happens if either (a) the
2728  * exception passed to raiseAsync is NULL, or (b) there's no
2729  * CATCH_FRAME on the stack.  In either case, we strip the entire
2730  * stack and replace the thread with a zombie.
2731  *
2732  * -------------------------------------------------------------------------- */
2733
2734 void
2735 deleteThread(StgTSO *tso)
2736 {
2737   raiseAsync(tso,NULL);
2738 }
2739
2740 void
2741 raiseAsync(StgTSO *tso, StgClosure *exception)
2742 {
2743   StgUpdateFrame* su = tso->su;
2744   StgPtr          sp = tso->sp;
2745
2746   /* Thread already dead? */
2747   if (tso->what_next == ThreadComplete || tso->what_next == ThreadKilled) {
2748     return;
2749   }
2750
2751   IF_DEBUG(scheduler, sched_belch("raising exception in thread %ld.", tso->id));
2752
2753   /* Remove it from any blocking queues */
2754   unblockThread(tso);
2755
2756   /* The stack freezing code assumes there's a closure pointer on
2757    * the top of the stack.  This isn't always the case with compiled
2758    * code, so we have to push a dummy closure on the top which just
2759    * returns to the next return address on the stack.
2760    */
2761   if ( LOOKS_LIKE_GHC_INFO((void*)*sp) ) {
2762     *(--sp) = (W_)&stg_dummy_ret_closure;
2763   }
2764
2765   while (1) {
2766     int words = ((P_)su - (P_)sp) - 1;
2767     nat i;
2768     StgAP_UPD * ap;
2769
2770     /* If we find a CATCH_FRAME, and we've got an exception to raise,
2771      * then build PAP(handler,exception,realworld#), and leave it on
2772      * top of the stack ready to enter.
2773      */
2774     if (get_itbl(su)->type == CATCH_FRAME && exception != NULL) {
2775       StgCatchFrame *cf = (StgCatchFrame *)su;
2776       /* we've got an exception to raise, so let's pass it to the
2777        * handler in this frame.
2778        */
2779       ap = (StgAP_UPD *)allocate(sizeofW(StgPAP) + 2);
2780       TICK_ALLOC_UPD_PAP(3,0);
2781       SET_HDR(ap,&stg_PAP_info,cf->header.prof.ccs);
2782
2783       ap->n_args = 2;
2784       ap->fun = cf->handler;    /* :: Exception -> IO a */
2785       ap->payload[0] = exception;
2786       ap->payload[1] = ARG_TAG(0); /* realworld token */
2787
2788       /* throw away the stack from Sp up to and including the
2789        * CATCH_FRAME.
2790        */
2791       sp = (P_)su + sizeofW(StgCatchFrame) - 1;
2792       tso->su = cf->link;
2793
2794       /* Restore the blocked/unblocked state for asynchronous exceptions
2795        * at the CATCH_FRAME.
2796        *
2797        * If exceptions were unblocked at the catch, arrange that they
2798        * are unblocked again after executing the handler by pushing an
2799        * unblockAsyncExceptions_ret stack frame.
2800        */
2801       if (!cf->exceptions_blocked) {
2802         *(sp--) = (W_)&stg_unblockAsyncExceptionszh_ret_info;
2803       }
2804
2805       /* Ensure that async exceptions are blocked when running the handler.
2806        */
2807       if (tso->blocked_exceptions == NULL) {
2808         tso->blocked_exceptions = END_TSO_QUEUE;
2809       }
2810
2811       /* Put the newly-built PAP on top of the stack, ready to execute
2812        * when the thread restarts.
2813        */
2814       sp[0] = (W_)ap;
2815       tso->sp = sp;
2816       tso->what_next = ThreadEnterGHC;
2817       IF_DEBUG(sanity, checkTSO(tso));
2818       return;
2819     }
2820
2821     /* First build an AP_UPD consisting of the stack chunk above the
2822      * current update frame, with the top word on the stack as the
2823      * fun field.
2824      */
2825     ap = (StgAP_UPD *)allocate(AP_sizeW(words));
2826
2827     ASSERT(words >= 0);
2828
2829     ap->n_args = words;
2830     ap->fun    = (StgClosure *)sp[0];
2831     sp++;
2832     for(i=0; i < (nat)words; ++i) {
2833       ap->payload[i] = (StgClosure *)*sp++;
2834     }
2835
2836     switch (get_itbl(su)->type) {
2837
2838     case UPDATE_FRAME:
2839       {
2840         SET_HDR(ap,&stg_AP_UPD_info,su->header.prof.ccs /* ToDo */);
2841         TICK_ALLOC_UP_THK(words+1,0);
2842
2843         IF_DEBUG(scheduler,
2844                  fprintf(stderr,  "scheduler: Updating ");
2845                  printPtr((P_)su->updatee);
2846                  fprintf(stderr,  " with ");
2847                  printObj((StgClosure *)ap);
2848                  );
2849
2850         /* Replace the updatee with an indirection - happily
2851          * this will also wake up any threads currently
2852          * waiting on the result.
2853          */
2854         UPD_IND_NOLOCK(su->updatee,ap);  /* revert the black hole */
2855         su = su->link;
2856         sp += sizeofW(StgUpdateFrame) -1;
2857         sp[0] = (W_)ap; /* push onto stack */
2858         break;
2859       }
2860
2861     case CATCH_FRAME:
2862       {
2863         StgCatchFrame *cf = (StgCatchFrame *)su;
2864         StgClosure* o;
2865
2866         /* We want a PAP, not an AP_UPD.  Fortunately, the
2867          * layout's the same.
2868          */
2869         SET_HDR(ap,&stg_PAP_info,su->header.prof.ccs /* ToDo */);
2870         TICK_ALLOC_UPD_PAP(words+1,0);
2871
2872         /* now build o = FUN(catch,ap,handler) */
2873         o = (StgClosure *)allocate(sizeofW(StgClosure)+2);
2874         TICK_ALLOC_FUN(2,0);
2875         SET_HDR(o,&stg_catch_info,su->header.prof.ccs /* ToDo */);
2876         o->payload[0] = (StgClosure *)ap;
2877         o->payload[1] = cf->handler;
2878
2879         IF_DEBUG(scheduler,
2880                  fprintf(stderr,  "scheduler: Built ");
2881                  printObj((StgClosure *)o);
2882                  );
2883
2884         /* pop the old handler and put o on the stack */
2885         su = cf->link;
2886         sp += sizeofW(StgCatchFrame) - 1;
2887         sp[0] = (W_)o;
2888         break;
2889       }
2890
2891     case SEQ_FRAME:
2892       {
2893         StgSeqFrame *sf = (StgSeqFrame *)su;
2894         StgClosure* o;
2895
2896         SET_HDR(ap,&stg_PAP_info,su->header.prof.ccs /* ToDo */);
2897         TICK_ALLOC_UPD_PAP(words+1,0);
2898
2899         /* now build o = FUN(seq,ap) */
2900         o = (StgClosure *)allocate(sizeofW(StgClosure)+1);
2901         TICK_ALLOC_SE_THK(1,0);
2902         SET_HDR(o,&stg_seq_info,su->header.prof.ccs /* ToDo */);
2903         o->payload[0] = (StgClosure *)ap;
2904
2905         IF_DEBUG(scheduler,
2906                  fprintf(stderr,  "scheduler: Built ");
2907                  printObj((StgClosure *)o);
2908                  );
2909
2910         /* pop the old handler and put o on the stack */
2911         su = sf->link;
2912         sp += sizeofW(StgSeqFrame) - 1;
2913         sp[0] = (W_)o;
2914         break;
2915       }
2916
2917     case STOP_FRAME:
2918       /* We've stripped the entire stack, the thread is now dead. */
2919       sp += sizeofW(StgStopFrame) - 1;
2920       sp[0] = (W_)exception;    /* save the exception */
2921       tso->what_next = ThreadKilled;
2922       tso->su = (StgUpdateFrame *)(sp+1);
2923       tso->sp = sp;
2924       return;
2925
2926     default:
2927       barf("raiseAsync");
2928     }
2929   }
2930   barf("raiseAsync");
2931 }
2932
2933 /* -----------------------------------------------------------------------------
2934    resurrectThreads is called after garbage collection on the list of
2935    threads found to be garbage.  Each of these threads will be woken
2936    up and sent a signal: BlockedOnDeadMVar if the thread was blocked
2937    on an MVar, or NonTermination if the thread was blocked on a Black
2938    Hole.
2939    -------------------------------------------------------------------------- */
2940
2941 void
2942 resurrectThreads( StgTSO *threads )
2943 {
2944   StgTSO *tso, *next;
2945
2946   for (tso = threads; tso != END_TSO_QUEUE; tso = next) {
2947     next = tso->global_link;
2948     tso->global_link = all_threads;
2949     all_threads = tso;
2950     IF_DEBUG(scheduler, sched_belch("resurrecting thread %d", tso->id));
2951
2952     switch (tso->why_blocked) {
2953     case BlockedOnMVar:
2954     case BlockedOnException:
2955       raiseAsync(tso,(StgClosure *)BlockedOnDeadMVar_closure);
2956       break;
2957     case BlockedOnBlackHole:
2958       raiseAsync(tso,(StgClosure *)NonTermination_closure);
2959       break;
2960     case NotBlocked:
2961       /* This might happen if the thread was blocked on a black hole
2962        * belonging to a thread that we've just woken up (raiseAsync
2963        * can wake up threads, remember...).
2964        */
2965       continue;
2966     default:
2967       barf("resurrectThreads: thread blocked in a strange way");
2968     }
2969   }
2970 }
2971
2972 /* -----------------------------------------------------------------------------
2973  * Blackhole detection: if we reach a deadlock, test whether any
2974  * threads are blocked on themselves.  Any threads which are found to
2975  * be self-blocked get sent a NonTermination exception.
2976  *
2977  * This is only done in a deadlock situation in order to avoid
2978  * performance overhead in the normal case.
2979  * -------------------------------------------------------------------------- */
2980
2981 static void
2982 detectBlackHoles( void )
2983 {
2984     StgTSO *t = all_threads;
2985     StgUpdateFrame *frame;
2986     StgClosure *blocked_on;
2987
2988     for (t = all_threads; t != END_TSO_QUEUE; t = t->global_link) {
2989
2990         if (t->why_blocked != BlockedOnBlackHole) {
2991             continue;
2992         }
2993
2994         blocked_on = t->block_info.closure;
2995
2996         for (frame = t->su; ; frame = frame->link) {
2997             switch (get_itbl(frame)->type) {
2998
2999             case UPDATE_FRAME:
3000                 if (frame->updatee == blocked_on) {
3001                     /* We are blocking on one of our own computations, so
3002                      * send this thread the NonTermination exception.
3003                      */
3004                     IF_DEBUG(scheduler,
3005                              sched_belch("thread %d is blocked on itself", t->id));
3006                     raiseAsync(t, (StgClosure *)NonTermination_closure);
3007                     goto done;
3008                 }
3009                 else {
3010                     continue;
3011                 }
3012
3013             case CATCH_FRAME:
3014             case SEQ_FRAME:
3015                 continue;
3016
3017             case STOP_FRAME:
3018                 break;
3019             }
3020             break;
3021         }
3022
3023     done: ;
3024     }
3025 }
3026
3027 //@node Debugging Routines, Index, Exception Handling Routines, Main scheduling code
3028 //@subsection Debugging Routines
3029
3030 /* -----------------------------------------------------------------------------
3031    Debugging: why is a thread blocked
3032    -------------------------------------------------------------------------- */
3033
3034 #ifdef DEBUG
3035
3036 void
3037 printThreadBlockage(StgTSO *tso)
3038 {
3039   switch (tso->why_blocked) {
3040   case BlockedOnRead:
3041     fprintf(stderr,"is blocked on read from fd %d", tso->block_info.fd);
3042     break;
3043   case BlockedOnWrite:
3044     fprintf(stderr,"is blocked on write to fd %d", tso->block_info.fd);
3045     break;
3046   case BlockedOnDelay:
3047     fprintf(stderr,"is blocked until %d", tso->block_info.target);
3048     break;
3049   case BlockedOnMVar:
3050     fprintf(stderr,"is blocked on an MVar");
3051     break;
3052   case BlockedOnException:
3053     fprintf(stderr,"is blocked on delivering an exception to thread %d",
3054             tso->block_info.tso->id);
3055     break;
3056   case BlockedOnBlackHole:
3057     fprintf(stderr,"is blocked on a black hole");
3058     break;
3059   case NotBlocked:
3060     fprintf(stderr,"is not blocked");
3061     break;
3062 #if defined(PAR)
3063   case BlockedOnGA:
3064     fprintf(stderr,"is blocked on global address; local FM_BQ is %p (%s)",
3065             tso->block_info.closure, info_type(tso->block_info.closure));
3066     break;
3067   case BlockedOnGA_NoSend:
3068     fprintf(stderr,"is blocked on global address (no send); local FM_BQ is %p (%s)",
3069             tso->block_info.closure, info_type(tso->block_info.closure));
3070     break;
3071 #endif
3072   default:
3073     barf("printThreadBlockage: strange tso->why_blocked: %d for TSO %d (%d)",
3074          tso->why_blocked, tso->id, tso);
3075   }
3076 }
3077
3078 void
3079 printThreadStatus(StgTSO *tso)
3080 {
3081   switch (tso->what_next) {
3082   case ThreadKilled:
3083     fprintf(stderr,"has been killed");
3084     break;
3085   case ThreadComplete:
3086     fprintf(stderr,"has completed");
3087     break;
3088   default:
3089     printThreadBlockage(tso);
3090   }
3091 }
3092
3093 void
3094 printAllThreads(void)
3095 {
3096   StgTSO *t;
3097
3098   sched_belch("all threads:");
3099   for (t = all_threads; t != END_TSO_QUEUE; t = t->global_link) {
3100     fprintf(stderr, "\tthread %d ", t->id);
3101     printThreadStatus(t);
3102     fprintf(stderr,"\n");
3103   }
3104 }
3105
3106 /*
3107    Print a whole blocking queue attached to node (debugging only).
3108 */
3109 //@cindex print_bq
3110 # if defined(PAR)
3111 void
3112 print_bq (StgClosure *node)
3113 {
3114   StgBlockingQueueElement *bqe;
3115   StgTSO *tso;
3116   rtsBool end;
3117
3118   fprintf(stderr,"## BQ of closure %p (%s): ",
3119           node, info_type(node));
3120
3121   /* should cover all closures that may have a blocking queue */
3122   ASSERT(get_itbl(node)->type == BLACKHOLE_BQ ||
3123          get_itbl(node)->type == FETCH_ME_BQ ||
3124          get_itbl(node)->type == RBH);
3125
3126   ASSERT(node!=(StgClosure*)NULL);         // sanity check
3127   /*
3128      NB: In a parallel setup a BQ of an RBH must end with an RBH_Save closure;
3129   */
3130   for (bqe = ((StgBlockingQueue*)node)->blocking_queue, end = (bqe==END_BQ_QUEUE);
3131        !end; // iterate until bqe points to a CONSTR
3132        end = (get_itbl(bqe)->type == CONSTR) || (bqe->link==END_BQ_QUEUE), bqe = end ? END_BQ_QUEUE : bqe->link) {
3133     ASSERT(bqe != END_BQ_QUEUE);             // sanity check
3134     ASSERT(bqe != (StgTSO*)NULL);            // sanity check
3135     /* types of closures that may appear in a blocking queue */
3136     ASSERT(get_itbl(bqe)->type == TSO ||
3137            get_itbl(bqe)->type == BLOCKED_FETCH ||
3138            get_itbl(bqe)->type == CONSTR);
3139     /* only BQs of an RBH end with an RBH_Save closure */
3140     ASSERT(get_itbl(bqe)->type != CONSTR || get_itbl(node)->type == RBH);
3141
3142     switch (get_itbl(bqe)->type) {
3143     case TSO:
3144       fprintf(stderr," TSO %d (%x),",
3145               ((StgTSO *)bqe)->id, ((StgTSO *)bqe));
3146       break;
3147     case BLOCKED_FETCH:
3148       fprintf(stderr," BF (node=%p, ga=((%x, %d, %x)),",
3149               ((StgBlockedFetch *)bqe)->node,
3150               ((StgBlockedFetch *)bqe)->ga.payload.gc.gtid,
3151               ((StgBlockedFetch *)bqe)->ga.payload.gc.slot,
3152               ((StgBlockedFetch *)bqe)->ga.weight);
3153       break;
3154     case CONSTR:
3155       fprintf(stderr," %s (IP %p),",
3156               (get_itbl(bqe) == &stg_RBH_Save_0_info ? "RBH_Save_0" :
3157                get_itbl(bqe) == &stg_RBH_Save_1_info ? "RBH_Save_1" :
3158                get_itbl(bqe) == &stg_RBH_Save_2_info ? "RBH_Save_2" :
3159                "RBH_Save_?"), get_itbl(bqe));
3160       break;
3161     default:
3162       barf("Unexpected closure type %s in blocking queue of %p (%s)",
3163            info_type(bqe), node, info_type(node));
3164       break;
3165     }
3166   } /* for */
3167   fputc('\n', stderr);
3168 }
3169 # elif defined(GRAN)
3170 void
3171 print_bq (StgClosure *node)
3172 {
3173   StgBlockingQueueElement *bqe;
3174   PEs node_loc, tso_loc;
3175   rtsBool end;
3176
3177   /* should cover all closures that may have a blocking queue */
3178   ASSERT(get_itbl(node)->type == BLACKHOLE_BQ ||
3179          get_itbl(node)->type == FETCH_ME_BQ ||
3180          get_itbl(node)->type == RBH);
3181
3182   ASSERT(node!=(StgClosure*)NULL);         // sanity check
3183   node_loc = where_is(node);
3184
3185   fprintf(stderr,"## BQ of closure %p (%s) on [PE %d]: ",
3186           node, info_type(node), node_loc);
3187
3188   /*
3189      NB: In a parallel setup a BQ of an RBH must end with an RBH_Save closure;
3190   */
3191   for (bqe = ((StgBlockingQueue*)node)->blocking_queue, end = (bqe==END_BQ_QUEUE);
3192        !end; // iterate until bqe points to a CONSTR
3193        end = (get_itbl(bqe)->type == CONSTR) || (bqe->link==END_BQ_QUEUE), bqe = end ? END_BQ_QUEUE : bqe->link) {
3194     ASSERT(bqe != END_BQ_QUEUE);             // sanity check
3195     ASSERT(bqe != (StgBlockingQueueElement *)NULL);  // sanity check
3196     /* types of closures that may appear in a blocking queue */
3197     ASSERT(get_itbl(bqe)->type == TSO ||
3198            get_itbl(bqe)->type == CONSTR);
3199     /* only BQs of an RBH end with an RBH_Save closure */
3200     ASSERT(get_itbl(bqe)->type != CONSTR || get_itbl(node)->type == RBH);
3201
3202     tso_loc = where_is((StgClosure *)bqe);
3203     switch (get_itbl(bqe)->type) {
3204     case TSO:
3205       fprintf(stderr," TSO %d (%p) on [PE %d],",
3206               ((StgTSO *)bqe)->id, (StgTSO *)bqe, tso_loc);
3207       break;
3208     case CONSTR:
3209       fprintf(stderr," %s (IP %p),",
3210               (get_itbl(bqe) == &stg_RBH_Save_0_info ? "RBH_Save_0" :
3211                get_itbl(bqe) == &stg_RBH_Save_1_info ? "RBH_Save_1" :
3212                get_itbl(bqe) == &stg_RBH_Save_2_info ? "RBH_Save_2" :
3213                "RBH_Save_?"), get_itbl(bqe));
3214       break;
3215     default:
3216       barf("Unexpected closure type %s in blocking queue of %p (%s)",
3217            info_type((StgClosure *)bqe), node, info_type(node));
3218       break;
3219     }
3220   } /* for */
3221   fputc('\n', stderr);
3222 }
3223 #else
3224 /*
3225    Nice and easy: only TSOs on the blocking queue
3226 */
3227 void
3228 print_bq (StgClosure *node)
3229 {
3230   StgTSO *tso;
3231
3232   ASSERT(node!=(StgClosure*)NULL);         // sanity check
3233   for (tso = ((StgBlockingQueue*)node)->blocking_queue;
3234        tso != END_TSO_QUEUE;
3235        tso=tso->link) {
3236     ASSERT(tso!=NULL && tso!=END_TSO_QUEUE);   // sanity check
3237     ASSERT(get_itbl(tso)->type == TSO);  // guess what, sanity check
3238     fprintf(stderr," TSO %d (%p),", tso->id, tso);
3239   }
3240   fputc('\n', stderr);
3241 }
3242 # endif
3243
3244 #if defined(PAR)
3245 static nat
3246 run_queue_len(void)
3247 {
3248   nat i;
3249   StgTSO *tso;
3250
3251   for (i=0, tso=run_queue_hd;
3252        tso != END_TSO_QUEUE;
3253        i++, tso=tso->link)
3254     /* nothing */
3255
3256   return i;
3257 }
3258 #endif
3259
3260 static void
3261 sched_belch(char *s, ...)
3262 {
3263   va_list ap;
3264   va_start(ap,s);
3265 #ifdef SMP
3266   fprintf(stderr, "scheduler (task %ld): ", pthread_self());
3267 #else
3268   fprintf(stderr, "scheduler: ");
3269 #endif
3270   vfprintf(stderr, s, ap);
3271   fprintf(stderr, "\n");
3272 }
3273
3274 #endif /* DEBUG */
3275
3276
3277 //@node Index,  , Debugging Routines, Main scheduling code
3278 //@subsection Index
3279
3280 //@index
3281 //* MainRegTable::  @cindex\s-+MainRegTable
3282 //* StgMainThread::  @cindex\s-+StgMainThread
3283 //* awaken_blocked_queue::  @cindex\s-+awaken_blocked_queue
3284 //* blocked_queue_hd::  @cindex\s-+blocked_queue_hd
3285 //* blocked_queue_tl::  @cindex\s-+blocked_queue_tl
3286 //* context_switch::  @cindex\s-+context_switch
3287 //* createThread::  @cindex\s-+createThread
3288 //* free_capabilities::  @cindex\s-+free_capabilities
3289 //* gc_pending_cond::  @cindex\s-+gc_pending_cond
3290 //* initScheduler::  @cindex\s-+initScheduler
3291 //* interrupted::  @cindex\s-+interrupted
3292 //* n_free_capabilities::  @cindex\s-+n_free_capabilities
3293 //* next_thread_id::  @cindex\s-+next_thread_id
3294 //* print_bq::  @cindex\s-+print_bq
3295 //* run_queue_hd::  @cindex\s-+run_queue_hd
3296 //* run_queue_tl::  @cindex\s-+run_queue_tl
3297 //* sched_mutex::  @cindex\s-+sched_mutex
3298 //* schedule::  @cindex\s-+schedule
3299 //* take_off_run_queue::  @cindex\s-+take_off_run_queue
3300 //* task_ids::  @cindex\s-+task_ids
3301 //* term_mutex::  @cindex\s-+term_mutex
3302 //* thread_ready_cond::  @cindex\s-+thread_ready_cond
3303 //@end index