ghc/rts/Schedule.c

   1 /* ---------------------------------------------------------------------------
   2  *
   3  * (c) The GHC Team, 1998-2004
   4  *
   5  * Scheduler
   6  *
   7  * Different GHC ways use this scheduler quite differently (see comments below)
   8  * Here is the global picture:
   9  *
  10  * WAY  Name     CPP flag  What's it for
  11  * --------------------------------------
  12  * mp   GUM      PARALLEL_HASKELL          Parallel execution on a distrib. memory machine
  13  * s    SMP      SMP          Parallel execution on a shared memory machine
  14  * mg   GranSim  GRAN         Simulation of parallel execution
  15  * md   GUM/GdH  DIST         Distributed execution (based on GUM)
  16  *
  17  * --------------------------------------------------------------------------*/
  18
  19 /*
  20  * Version with support for distributed memory parallelism aka GUM (WAY=mp):
  21
  22    The main scheduling loop in GUM iterates until a finish message is received.
  23    In that case a global flag @receivedFinish@ is set and this instance of
  24    the RTS shuts down. See ghc/rts/parallel/HLComms.c:processMessages()
  25    for the handling of incoming messages, such as PP_FINISH.
  26    Note that in the parallel case we have a system manager that coordinates
  27    different PEs, each of which are running one instance of the RTS.
  28    See ghc/rts/parallel/SysMan.c for the main routine of the parallel program.
  29    From this routine processes executing ghc/rts/Main.c are spawned. -- HWL
  30
  31  * Version with support for simulating parallel execution aka GranSim (WAY=mg):
  32
  33    The main scheduling code in GranSim is quite different from that in std
  34    (concurrent) Haskell: while concurrent Haskell just iterates over the
  35    threads in the runnable queue, GranSim is event driven, i.e. it iterates
  36    over the events in the global event queue.  -- HWL
  37 */
  38
  39 #include "PosixSource.h"
  40 #include "Rts.h"
  41 #include "SchedAPI.h"
  42 #include "RtsUtils.h"
  43 #include "RtsFlags.h"
  44 #include "BlockAlloc.h"
  45 #include "OSThreads.h"
  46 #include "Storage.h"
  47 #include "StgRun.h"
  48 #include "Hooks.h"
  49 #define COMPILING_SCHEDULER
  50 #include "Schedule.h"
  51 #include "StgMiscClosures.h"
  52 #include "Interpreter.h"
  53 #include "Exception.h"
  54 #include "Printer.h"
  55 #include "Signals.h"
  56 #include "Sanity.h"
  57 #include "Stats.h"
  58 #include "STM.h"
  59 #include "Timer.h"
  60 #include "Prelude.h"
  61 #include "ThreadLabels.h"
  62 #include "LdvProfile.h"
  63 #include "Updates.h"
  64 #ifdef PROFILING
  65 #include "Proftimer.h"
  66 #include "ProfHeap.h"
  67 #endif
  68 #if defined(GRAN) || defined(PARALLEL_HASKELL)
  69 # include "GranSimRts.h"
  70 # include "GranSim.h"
  71 # include "ParallelRts.h"
  72 # include "Parallel.h"
  73 # include "ParallelDebug.h"
  74 # include "FetchMe.h"
  75 # include "HLC.h"
  76 #endif
  77 #include "Sparks.h"
  78 #include "Capability.h"
  79 #include  "Task.h"
  80
  81 #ifdef HAVE_SYS_TYPES_H
  82 #include <sys/types.h>
  83 #endif
  84 #ifdef HAVE_UNISTD_H
  85 #include <unistd.h>
  86 #endif
  87
  88 #include <string.h>
  89 #include <stdlib.h>
  90 #include <stdarg.h>
  91
  92 #ifdef HAVE_ERRNO_H
  93 #include <errno.h>
  94 #endif
  95
  96 // Turn off inlining when debugging - it obfuscates things
  97 #ifdef DEBUG
  98 # undef  STATIC_INLINE
  99 # define STATIC_INLINE static
 100 #endif
 101
 102 #ifdef THREADED_RTS
 103 #define USED_IN_THREADED_RTS
 104 #else
 105 #define USED_IN_THREADED_RTS STG_UNUSED
 106 #endif
 107
 108 #ifdef RTS_SUPPORTS_THREADS
 109 #define USED_WHEN_RTS_SUPPORTS_THREADS
 110 #else
 111 #define USED_WHEN_RTS_SUPPORTS_THREADS STG_UNUSED
 112 #endif
 113
 114 /* Main thread queue.
 115  * Locks required: sched_mutex.
 116  */
 117 StgMainThread *main_threads = NULL;
 118
 119 #if defined(GRAN)
 120
 121 StgTSO* ActiveTSO = NULL; /* for assigning system costs; GranSim-Light only */
 122 /* rtsTime TimeOfNextEvent, EndOfTimeSlice;            now in GranSim.c */
 123
 124 /*
 125    In GranSim we have a runnable and a blocked queue for each processor.
 126    In order to minimise code changes new arrays run_queue_hds/tls
 127    are created. run_queue_hd is then a short cut (macro) for
 128    run_queue_hds[CurrentProc] (see GranSim.h).
 129    -- HWL
 130 */
 131 StgTSO *run_queue_hds[MAX_PROC], *run_queue_tls[MAX_PROC];
 132 StgTSO *blocked_queue_hds[MAX_PROC], *blocked_queue_tls[MAX_PROC];
 133 StgTSO *ccalling_threadss[MAX_PROC];
 134 /* We use the same global list of threads (all_threads) in GranSim as in
 135    the std RTS (i.e. we are cheating). However, we don't use this list in
 136    the GranSim specific code at the moment (so we are only potentially
 137    cheating).  */
 138
 139 #else /* !GRAN */
 140
 141 /* Thread queues.
 142  * Locks required: sched_mutex.
 143  */
 144 StgTSO *run_queue_hd = NULL;
 145 StgTSO *run_queue_tl = NULL;
 146 StgTSO *blocked_queue_hd = NULL;
 147 StgTSO *blocked_queue_tl = NULL;
 148 StgTSO *blackhole_queue = NULL;
 149 StgTSO *sleeping_queue = NULL;    /* perhaps replace with a hash table? */
 150
 151 #endif
 152
 153 /* The blackhole_queue should be checked for threads to wake up.  See
 154  * Schedule.h for more thorough comment.
 155  */
 156 rtsBool blackholes_need_checking = rtsFalse;
 157
 158 /* Linked list of all threads.
 159  * Used for detecting garbage collected threads.
 160  */
 161 StgTSO *all_threads = NULL;
 162
 163 /* When a thread performs a safe C call (_ccall_GC, using old
 164  * terminology), it gets put on the suspended_ccalling_threads
 165  * list. Used by the garbage collector.
 166  */
 167 static StgTSO *suspended_ccalling_threads;
 168
 169 /* KH: The following two flags are shared memory locations.  There is no need
 170        to lock them, since they are only unset at the end of a scheduler
 171        operation.
 172 */
 173
 174 /* flag set by signal handler to precipitate a context switch */
 175 int context_switch = 0;
 176
 177 /* flag that tracks whether we have done any execution in this time slice. */
 178 nat recent_activity = ACTIVITY_YES;
 179
 180 /* if this flag is set as well, give up execution */
 181 rtsBool interrupted = rtsFalse;
 182
 183 /* Next thread ID to allocate.
 184  * Locks required: thread_id_mutex
 185  */
 186 static StgThreadID next_thread_id = 1;
 187
 188 /*
 189  * Pointers to the state of the current thread.
 190  * Rule of thumb: if CurrentTSO != NULL, then we're running a Haskell
 191  * thread.  If CurrentTSO == NULL, then we're at the scheduler level.
 192  */
 193
 194 /* The smallest stack size that makes any sense is:
 195  *    RESERVED_STACK_WORDS    (so we can get back from the stack overflow)
 196  *  + sizeofW(StgStopFrame)   (the stg_stop_thread_info frame)
 197  *  + 1                       (the closure to enter)
 198  *  + 1                       (stg_ap_v_ret)
 199  *  + 1                       (spare slot req'd by stg_ap_v_ret)
 200  *
 201  * A thread with this stack will bomb immediately with a stack
 202  * overflow, which will increase its stack size.
 203  */
 204
 205 #define MIN_STACK_WORDS (RESERVED_STACK_WORDS + sizeofW(StgStopFrame) + 3)
 206
 207
 208 #if defined(GRAN)
 209 StgTSO *CurrentTSO;
 210 #endif
 211
 212 /*  This is used in `TSO.h' and gcc 2.96 insists that this variable actually
 213  *  exists - earlier gccs apparently didn't.
 214  *  -= chak
 215  */
 216 StgTSO dummy_tso;
 217
 218 /*
 219  * Set to TRUE when entering a shutdown state (via shutdownHaskellAndExit()) --
 220  * in an MT setting, needed to signal that a worker thread shouldn't hang around
 221  * in the scheduler when it is out of work.
 222  */
 223 static rtsBool shutting_down_scheduler = rtsFalse;
 224
 225 #if defined(RTS_SUPPORTS_THREADS)
 226 /* ToDo: carefully document the invariants that go together
 227  *       with these synchronisation objects.
 228  */
 229 Mutex     sched_mutex       = INIT_MUTEX_VAR;
 230 Mutex     term_mutex        = INIT_MUTEX_VAR;
 231
 232 #endif /* RTS_SUPPORTS_THREADS */
 233
 234 #if defined(PARALLEL_HASKELL)
 235 StgTSO *LastTSO;
 236 rtsTime TimeOfLastYield;
 237 rtsBool emitSchedule = rtsTrue;
 238 #endif
 239
 240 #if DEBUG
 241 static char *whatNext_strs[] = {
 242   "(unknown)",
 243   "ThreadRunGHC",
 244   "ThreadInterpret",
 245   "ThreadKilled",
 246   "ThreadRelocated",
 247   "ThreadComplete"
 248 };
 249 #endif
 250
 251 /* -----------------------------------------------------------------------------
 252  * static function prototypes
 253  * -------------------------------------------------------------------------- */
 254
 255 #if defined(RTS_SUPPORTS_THREADS)
 256 static void taskStart(void);
 257 #endif
 258
 259 static void schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS,
 260                       Capability *initialCapability );
 261
 262 //
 263 // These function all encapsulate parts of the scheduler loop, and are
 264 // abstracted only to make the structure and control flow of the
 265 // scheduler clearer.
 266 //
 267 static void schedulePreLoop(void);
 268 static void scheduleStartSignalHandlers(void);
 269 static void scheduleCheckBlockedThreads(void);
 270 static void scheduleCheckBlackHoles(void);
 271 static void scheduleDetectDeadlock(void);
 272 #if defined(GRAN)
 273 static StgTSO *scheduleProcessEvent(rtsEvent *event);
 274 #endif
 275 #if defined(PARALLEL_HASKELL)
 276 static StgTSO *scheduleSendPendingMessages(void);
 277 static void scheduleActivateSpark(void);
 278 static rtsBool scheduleGetRemoteWork(rtsBool *receivedFinish);
 279 #endif
 280 #if defined(PAR) || defined(GRAN)
 281 static void scheduleGranParReport(void);
 282 #endif
 283 static void schedulePostRunThread(void);
 284 static rtsBool scheduleHandleHeapOverflow( Capability *cap, StgTSO *t );
 285 static void scheduleHandleStackOverflow( StgTSO *t);
 286 static rtsBool scheduleHandleYield( StgTSO *t, nat prev_what_next );
 287 static void scheduleHandleThreadBlocked( StgTSO *t );
 288 static rtsBool scheduleHandleThreadFinished( StgMainThread *mainThread,
 289                                              Capability *cap, StgTSO *t );
 290 static rtsBool scheduleDoHeapProfile(rtsBool ready_to_gc);
 291 static void scheduleDoGC(rtsBool force_major);
 292
 293 static void unblockThread(StgTSO *tso);
 294 static rtsBool checkBlackHoles(void);
 295 static SchedulerStatus waitThread_(/*out*/StgMainThread* m,
 296                                    Capability *initialCapability
 297                                    );
 298 static void scheduleThread_ (StgTSO* tso);
 299 static void AllRoots(evac_fn evac);
 300
 301 static StgTSO *threadStackOverflow(StgTSO *tso);
 302
 303 static void raiseAsync_(StgTSO *tso, StgClosure *exception,
 304                         rtsBool stop_at_atomically);
 305
 306 static void printThreadBlockage(StgTSO *tso);
 307 static void printThreadStatus(StgTSO *tso);
 308 void printThreadQueue(StgTSO *tso);
 309
 310 #if defined(PARALLEL_HASKELL)
 311 StgTSO * createSparkThread(rtsSpark spark);
 312 StgTSO * activateSpark (rtsSpark spark);
 313 #endif
 314
 315 /* ----------------------------------------------------------------------------
 316  * Starting Tasks
 317  * ------------------------------------------------------------------------- */
 318
 319 #if defined(RTS_SUPPORTS_THREADS)
 320 static nat startingWorkerThread = 0;
 321
 322 static void
 323 taskStart(void)
 324 {
 325   ACQUIRE_LOCK(&sched_mutex);
 326   startingWorkerThread--;
 327   schedule(NULL,NULL);
 328   taskStop();
 329   RELEASE_LOCK(&sched_mutex);
 330 }
 331
 332 void
 333 startSchedulerTaskIfNecessary(void)
 334 {
 335     if ( !EMPTY_RUN_QUEUE()
 336          && !shutting_down_scheduler // not if we're shutting down
 337          && startingWorkerThread==0)
 338     {
 339         // we don't want to start another worker thread
 340         // just because the last one hasn't yet reached the
 341         // "waiting for capability" state
 342         startingWorkerThread++;
 343         if (!maybeStartNewWorker(taskStart)) {
 344             startingWorkerThread--;
 345         }
 346     }
 347 }
 348 #endif
 349
 350 /* -----------------------------------------------------------------------------
 351  * Putting a thread on the run queue: different scheduling policies
 352  * -------------------------------------------------------------------------- */
 353
 354 STATIC_INLINE void
 355 addToRunQueue( StgTSO *t )
 356 {
 357 #if defined(PARALLEL_HASKELL)
 358     if (RtsFlags.ParFlags.doFairScheduling) {
 359         // this does round-robin scheduling; good for concurrency
 360         APPEND_TO_RUN_QUEUE(t);
 361     } else {
 362         // this does unfair scheduling; good for parallelism
 363         PUSH_ON_RUN_QUEUE(t);
 364     }
 365 #else
 366     // this does round-robin scheduling; good for concurrency
 367     APPEND_TO_RUN_QUEUE(t);
 368 #endif
 369 }
 370
 371 /* ---------------------------------------------------------------------------
 372    Main scheduling loop.
 373
 374    We use round-robin scheduling, each thread returning to the
 375    scheduler loop when one of these conditions is detected:
 376
 377       * out of heap space
 378       * timer expires (thread yields)
 379       * thread blocks
 380       * thread ends
 381       * stack overflow
 382
 383    Locking notes:  we acquire the scheduler lock once at the beginning
 384    of the scheduler loop, and release it when
 385
 386       * running a thread, or
 387       * waiting for work, or
 388       * waiting for a GC to complete.
 389
 390    GRAN version:
 391      In a GranSim setup this loop iterates over the global event queue.
 392      This revolves around the global event queue, which determines what
 393      to do next. Therefore, it's more complicated than either the
 394      concurrent or the parallel (GUM) setup.
 395
 396    GUM version:
 397      GUM iterates over incoming messages.
 398      It starts with nothing to do (thus CurrentTSO == END_TSO_QUEUE),
 399      and sends out a fish whenever it has nothing to do; in-between
 400      doing the actual reductions (shared code below) it processes the
 401      incoming messages and deals with delayed operations
 402      (see PendingFetches).
 403      This is not the ugliest code you could imagine, but it's bloody close.
 404
 405    ------------------------------------------------------------------------ */
 406
 407 static void
 408 schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS,
 409           Capability *initialCapability )
 410 {
 411   StgTSO *t;
 412   Capability *cap;
 413   StgThreadReturnCode ret;
 414 #if defined(GRAN)
 415   rtsEvent *event;
 416 #elif defined(PARALLEL_HASKELL)
 417   StgTSO *tso;
 418   GlobalTaskId pe;
 419   rtsBool receivedFinish = rtsFalse;
 420 # if defined(DEBUG)
 421   nat tp_size, sp_size; // stats only
 422 # endif
 423 #endif
 424   nat prev_what_next;
 425   rtsBool ready_to_gc;
 426
 427   // Pre-condition: sched_mutex is held.
 428   // We might have a capability, passed in as initialCapability.
 429   cap = initialCapability;
 430
 431 #if !defined(RTS_SUPPORTS_THREADS)
 432   // simply initialise it in the non-threaded case
 433   grabCapability(&cap);
 434 #endif
 435
 436   IF_DEBUG(scheduler,
 437            sched_belch("### NEW SCHEDULER LOOP (main thr: %p, cap: %p)",
 438                        mainThread, initialCapability);
 439       );
 440
 441   schedulePreLoop();
 442
 443   // -----------------------------------------------------------
 444   // Scheduler loop starts here:
 445
 446 #if defined(PARALLEL_HASKELL)
 447 #define TERMINATION_CONDITION        (!receivedFinish)
 448 #elif defined(GRAN)
 449 #define TERMINATION_CONDITION        ((event = get_next_event()) != (rtsEvent*)NULL)
 450 #else
 451 #define TERMINATION_CONDITION        rtsTrue
 452 #endif
 453
 454   while (TERMINATION_CONDITION) {
 455
 456 #if defined(GRAN)
 457       /* Choose the processor with the next event */
 458       CurrentProc = event->proc;
 459       CurrentTSO = event->tso;
 460 #endif
 461
 462 #if defined(RTS_SUPPORTS_THREADS)
 463       // Yield the capability to higher-priority tasks if necessary.
 464       //
 465       if (cap != NULL) {
 466           yieldCapability(&cap,
 467                           mainThread ? &mainThread->bound_thread_cond : NULL );
 468       }
 469
 470       // If we do not currently hold a capability, we wait for one
 471       //
 472       if (cap == NULL) {
 473           waitForCapability(&sched_mutex, &cap,
 474                             mainThread ? &mainThread->bound_thread_cond : NULL);
 475       }
 476
 477       // We now have a capability...
 478 #endif
 479
 480 #if 0 /* extra sanity checking */
 481       {
 482           StgMainThread *m;
 483           for (m = main_threads; m != NULL; m = m->link) {
 484               ASSERT(get_itbl(m->tso)->type == TSO);
 485           }
 486       }
 487 #endif
 488
 489     // Check whether we have re-entered the RTS from Haskell without
 490     // going via suspendThread()/resumeThread (i.e. a 'safe' foreign
 491     // call).
 492     if (cap->r.rInHaskell) {
 493           errorBelch("schedule: re-entered unsafely.\n"
 494                      "   Perhaps a 'foreign import unsafe' should be 'safe'?");
 495           stg_exit(1);
 496     }
 497
 498     //
 499     // Test for interruption.  If interrupted==rtsTrue, then either
 500     // we received a keyboard interrupt (^C), or the scheduler is
 501     // trying to shut down all the tasks (shutting_down_scheduler) in
 502     // the threaded RTS.
 503     //
 504     if (interrupted) {
 505         if (shutting_down_scheduler) {
 506             IF_DEBUG(scheduler, sched_belch("shutting down"));
 507             releaseCapability(cap);
 508             if (mainThread) {
 509                 mainThread->stat = Interrupted;
 510                 mainThread->ret  = NULL;
 511             }
 512             return;
 513         } else {
 514             IF_DEBUG(scheduler, sched_belch("interrupted"));
 515             deleteAllThreads();
 516         }
 517     }
 518
 519 #if defined(not_yet) && defined(SMP)
 520     //
 521     // Top up the run queue from our spark pool.  We try to make the
 522     // number of threads in the run queue equal to the number of
 523     // free capabilities.
 524     //
 525     {
 526         StgClosure *spark;
 527         if (EMPTY_RUN_QUEUE()) {
 528             spark = findSpark(rtsFalse);
 529             if (spark == NULL) {
 530                 break; /* no more sparks in the pool */
 531             } else {
 532                 createSparkThread(spark);
 533                 IF_DEBUG(scheduler,
 534                          sched_belch("==^^ turning spark of closure %p into a thread",
 535                                      (StgClosure *)spark));
 536             }
 537         }
 538     }
 539 #endif // SMP
 540
 541     scheduleStartSignalHandlers();
 542
 543     // Only check the black holes here if we've nothing else to do.
 544     // During normal execution, the black hole list only gets checked
 545     // at GC time, to avoid repeatedly traversing this possibly long
 546     // list each time around the scheduler.
 547     if (EMPTY_RUN_QUEUE()) { scheduleCheckBlackHoles(); }
 548
 549     scheduleCheckBlockedThreads();
 550
 551     scheduleDetectDeadlock();
 552
 553     // Normally, the only way we can get here with no threads to
 554     // run is if a keyboard interrupt received during
 555     // scheduleCheckBlockedThreads() or scheduleDetectDeadlock().
 556     // Additionally, it is not fatal for the
 557     // threaded RTS to reach here with no threads to run.
 558     //
 559     // win32: might be here due to awaitEvent() being abandoned
 560     // as a result of a console event having been delivered.
 561     if ( EMPTY_RUN_QUEUE() ) {
 562 #if !defined(RTS_SUPPORTS_THREADS) && !defined(mingw32_HOST_OS)
 563         ASSERT(interrupted);
 564 #endif
 565         continue; // nothing to do
 566     }
 567
 568 #if defined(PARALLEL_HASKELL)
 569     scheduleSendPendingMessages();
 570     if (EMPTY_RUN_QUEUE() && scheduleActivateSpark())
 571         continue;
 572
 573 #if defined(SPARKS)
 574     ASSERT(next_fish_to_send_at==0);  // i.e. no delayed fishes left!
 575 #endif
 576
 577     /* If we still have no work we need to send a FISH to get a spark
 578        from another PE */
 579     if (EMPTY_RUN_QUEUE()) {
 580         if (!scheduleGetRemoteWork(&receivedFinish)) continue;
 581         ASSERT(rtsFalse); // should not happen at the moment
 582     }
 583     // from here: non-empty run queue.
 584     //  TODO: merge above case with this, only one call processMessages() !
 585     if (PacketsWaiting()) {  /* process incoming messages, if
 586                                 any pending...  only in else
 587                                 because getRemoteWork waits for
 588                                 messages as well */
 589         receivedFinish = processMessages();
 590     }
 591 #endif
 592
 593 #if defined(GRAN)
 594     scheduleProcessEvent(event);
 595 #endif
 596
 597     //
 598     // Get a thread to run
 599     //
 600     ASSERT(run_queue_hd != END_TSO_QUEUE);
 601     POP_RUN_QUEUE(t);
 602
 603 #if defined(GRAN) || defined(PAR)
 604     scheduleGranParReport(); // some kind of debuging output
 605 #else
 606     // Sanity check the thread we're about to run.  This can be
 607     // expensive if there is lots of thread switching going on...
 608     IF_DEBUG(sanity,checkTSO(t));
 609 #endif
 610
 611 #if defined(RTS_SUPPORTS_THREADS)
 612     // Check whether we can run this thread in the current task.
 613     // If not, we have to pass our capability to the right task.
 614     {
 615       StgMainThread *m = t->main;
 616
 617       if(m)
 618       {
 619         if(m == mainThread)
 620         {
 621           IF_DEBUG(scheduler,
 622             sched_belch("### Running thread %d in bound thread", t->id));
 623           // yes, the Haskell thread is bound to the current native thread
 624         }
 625         else
 626         {
 627           IF_DEBUG(scheduler,
 628             sched_belch("### thread %d bound to another OS thread", t->id));
 629           // no, bound to a different Haskell thread: pass to that thread
 630           PUSH_ON_RUN_QUEUE(t);
 631           continue;
 632         }
 633       }
 634       else
 635       {
 636         if(mainThread != NULL)
 637         // The thread we want to run is unbound.
 638         {
 639           IF_DEBUG(scheduler,
 640             sched_belch("### this OS thread cannot run thread %d", t->id));
 641           // no, the current native thread is bound to a different
 642           // Haskell thread, so pass it to any worker thread
 643           PUSH_ON_RUN_QUEUE(t);
 644           continue;
 645         }
 646       }
 647     }
 648 #endif
 649
 650     cap->r.rCurrentTSO = t;
 651
 652     /* context switches are now initiated by the timer signal, unless
 653      * the user specified "context switch as often as possible", with
 654      * +RTS -C0
 655      */
 656     if ((RtsFlags.ConcFlags.ctxtSwitchTicks == 0
 657          && (run_queue_hd != END_TSO_QUEUE
 658              || blocked_queue_hd != END_TSO_QUEUE
 659              || sleeping_queue != END_TSO_QUEUE)))
 660         context_switch = 1;
 661
 662 run_thread:
 663
 664     RELEASE_LOCK(&sched_mutex);
 665
 666     IF_DEBUG(scheduler, sched_belch("-->> running thread %ld %s ...",
 667                               (long)t->id, whatNext_strs[t->what_next]));
 668
 669 #if defined(PROFILING)
 670     startHeapProfTimer();
 671 #endif
 672
 673     // ----------------------------------------------------------------------
 674     // Run the current thread
 675
 676     prev_what_next = t->what_next;
 677
 678     errno = t->saved_errno;
 679     cap->r.rInHaskell = rtsTrue;
 680
 681     recent_activity = ACTIVITY_YES;
 682
 683     switch (prev_what_next) {
 684
 685     case ThreadKilled:
 686     case ThreadComplete:
 687         /* Thread already finished, return to scheduler. */
 688         ret = ThreadFinished;
 689         break;
 690
 691     case ThreadRunGHC:
 692         ret = StgRun((StgFunPtr) stg_returnToStackTop, &cap->r);
 693         break;
 694
 695     case ThreadInterpret:
 696         ret = interpretBCO(cap);
 697         break;
 698
 699     default:
 700       barf("schedule: invalid what_next field");
 701     }
 702
 703 #if defined(SMP)
 704     // in SMP mode, we might return with a different capability than
 705     // we started with, if the Haskell thread made a foreign call.  So
 706     // let's find out what our current Capability is:
 707     cap = myCapability();
 708 #endif
 709
 710     cap->r.rInHaskell = rtsFalse;
 711
 712     // The TSO might have moved, eg. if it re-entered the RTS and a GC
 713     // happened.  So find the new location:
 714     t = cap->r.rCurrentTSO;
 715
 716     // And save the current errno in this thread.
 717     t->saved_errno = errno;
 718
 719     // ----------------------------------------------------------------------
 720
 721     /* Costs for the scheduler are assigned to CCS_SYSTEM */
 722 #if defined(PROFILING)
 723     stopHeapProfTimer();
 724     CCCS = CCS_SYSTEM;
 725 #endif
 726
 727     ACQUIRE_LOCK(&sched_mutex);
 728
 729     // We have run some Haskell code: there might be blackhole-blocked
 730     // threads to wake up now.
 731     if ( blackhole_queue != END_TSO_QUEUE ) {
 732         blackholes_need_checking = rtsTrue;
 733     }
 734
 735 #if defined(RTS_SUPPORTS_THREADS)
 736     IF_DEBUG(scheduler,debugBelch("sched (task %p): ", osThreadId()););
 737 #elif !defined(GRAN) && !defined(PARALLEL_HASKELL)
 738     IF_DEBUG(scheduler,debugBelch("sched: "););
 739 #endif
 740
 741     schedulePostRunThread();
 742
 743     ready_to_gc = rtsFalse;
 744
 745     switch (ret) {
 746     case HeapOverflow:
 747         ready_to_gc = scheduleHandleHeapOverflow(cap,t);
 748         break;
 749
 750     case StackOverflow:
 751         scheduleHandleStackOverflow(t);
 752         break;
 753
 754     case ThreadYielding:
 755         if (scheduleHandleYield(t, prev_what_next)) {
 756             // shortcut for switching between compiler/interpreter:
 757             goto run_thread;
 758         }
 759         break;
 760
 761     case ThreadBlocked:
 762         scheduleHandleThreadBlocked(t);
 763         break;
 764
 765     case ThreadFinished:
 766         if (scheduleHandleThreadFinished(mainThread, cap, t)) return;;
 767         break;
 768
 769     default:
 770       barf("schedule: invalid thread return code %d", (int)ret);
 771     }
 772
 773     if (scheduleDoHeapProfile(ready_to_gc)) { ready_to_gc = rtsFalse; }
 774     if (ready_to_gc) { scheduleDoGC(rtsFalse); }
 775   } /* end of while() */
 776
 777   IF_PAR_DEBUG(verbose,
 778                debugBelch("== Leaving schedule() after having received Finish\n"));
 779 }
 780
 781 /* ----------------------------------------------------------------------------
 782  * Setting up the scheduler loop
 783  * ASSUMES: sched_mutex
 784  * ------------------------------------------------------------------------- */
 785
 786 static void
 787 schedulePreLoop(void)
 788 {
 789 #if defined(GRAN)
 790     /* set up first event to get things going */
 791     /* ToDo: assign costs for system setup and init MainTSO ! */
 792     new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
 793               ContinueThread,
 794               CurrentTSO, (StgClosure*)NULL, (rtsSpark*)NULL);
 795
 796     IF_DEBUG(gran,
 797              debugBelch("GRAN: Init CurrentTSO (in schedule) = %p\n",
 798                         CurrentTSO);
 799              G_TSO(CurrentTSO, 5));
 800
 801     if (RtsFlags.GranFlags.Light) {
 802         /* Save current time; GranSim Light only */
 803         CurrentTSO->gran.clock = CurrentTime[CurrentProc];
 804     }
 805 #endif
 806 }
 807
 808 /* ----------------------------------------------------------------------------
 809  * Start any pending signal handlers
 810  * ASSUMES: sched_mutex
 811  * ------------------------------------------------------------------------- */
 812
 813 static void
 814 scheduleStartSignalHandlers(void)
 815 {
 816 #if defined(RTS_USER_SIGNALS) && !defined(RTS_SUPPORTS_THREADS)
 817     if (signals_pending()) {
 818       RELEASE_LOCK(&sched_mutex); /* ToDo: kill */
 819       startSignalHandlers();
 820       ACQUIRE_LOCK(&sched_mutex);
 821     }
 822 #endif
 823 }
 824
 825 /* ----------------------------------------------------------------------------
 826  * Check for blocked threads that can be woken up.
 827  * ASSUMES: sched_mutex
 828  * ------------------------------------------------------------------------- */
 829
 830 static void
 831 scheduleCheckBlockedThreads(void)
 832 {
 833     //
 834     // Check whether any waiting threads need to be woken up.  If the
 835     // run queue is empty, and there are no other tasks running, we
 836     // can wait indefinitely for something to happen.
 837     //
 838     if ( !EMPTY_QUEUE(blocked_queue_hd) || !EMPTY_QUEUE(sleeping_queue) )
 839     {
 840 #if defined(RTS_SUPPORTS_THREADS)
 841         // We shouldn't be here...
 842         barf("schedule: awaitEvent() in threaded RTS");
 843 #else
 844         awaitEvent( EMPTY_RUN_QUEUE() && !blackholes_need_checking );
 845 #endif
 846     }
 847 }
 848
 849
 850 /* ----------------------------------------------------------------------------
 851  * Check for threads blocked on BLACKHOLEs that can be woken up
 852  * ASSUMES: sched_mutex
 853  * ------------------------------------------------------------------------- */
 854 static void
 855 scheduleCheckBlackHoles( void )
 856 {
 857     if ( blackholes_need_checking )
 858     {
 859         checkBlackHoles();
 860         blackholes_need_checking = rtsFalse;
 861     }
 862 }
 863
 864 /* ----------------------------------------------------------------------------
 865  * Detect deadlock conditions and attempt to resolve them.
 866  * ASSUMES: sched_mutex
 867  * ------------------------------------------------------------------------- */
 868
 869 static void
 870 scheduleDetectDeadlock()
 871 {
 872
 873 #if defined(PARALLEL_HASKELL)
 874     // ToDo: add deadlock detection in GUM (similar to SMP) -- HWL
 875     return;
 876 #endif
 877
 878     /*
 879      * Detect deadlock: when we have no threads to run, there are no
 880      * threads blocked, waiting for I/O, or sleeping, and all the
 881      * other tasks are waiting for work, we must have a deadlock of
 882      * some description.
 883      */
 884     if ( EMPTY_THREAD_QUEUES() )
 885     {
 886 #if defined(RTS_SUPPORTS_THREADS)
 887         /*
 888          * In the threaded RTS, we only check for deadlock if there
 889          * has been no activity in a complete timeslice.  This means
 890          * we won't eagerly start a full GC just because we don't have
 891          * any threads to run currently.
 892          */
 893         if (recent_activity != ACTIVITY_INACTIVE) return;
 894 #endif
 895
 896         IF_DEBUG(scheduler, sched_belch("deadlocked, forcing major GC..."));
 897
 898         // Garbage collection can release some new threads due to
 899         // either (a) finalizers or (b) threads resurrected because
 900         // they are unreachable and will therefore be sent an
 901         // exception.  Any threads thus released will be immediately
 902         // runnable.
 903
 904         scheduleDoGC( rtsTrue/*force  major GC*/ );
 905         recent_activity = ACTIVITY_DONE_GC;
 906         if ( !EMPTY_RUN_QUEUE() ) return;
 907
 908 #if defined(RTS_USER_SIGNALS) && !defined(RTS_SUPPORTS_THREADS)
 909         /* If we have user-installed signal handlers, then wait
 910          * for signals to arrive rather then bombing out with a
 911          * deadlock.
 912          */
 913         if ( anyUserHandlers() ) {
 914             IF_DEBUG(scheduler,
 915                      sched_belch("still deadlocked, waiting for signals..."));
 916
 917             awaitUserSignals();
 918
 919             if (signals_pending()) {
 920                 RELEASE_LOCK(&sched_mutex);
 921                 startSignalHandlers();
 922                 ACQUIRE_LOCK(&sched_mutex);
 923             }
 924
 925             // either we have threads to run, or we were interrupted:
 926             ASSERT(!EMPTY_RUN_QUEUE() || interrupted);
 927         }
 928 #endif
 929
 930 #if !defined(RTS_SUPPORTS_THREADS)
 931         /* Probably a real deadlock.  Send the current main thread the
 932          * Deadlock exception (or in the SMP build, send *all* main
 933          * threads the deadlock exception, since none of them can make
 934          * progress).
 935          */
 936         {
 937             StgMainThread *m;
 938             m = main_threads;
 939             switch (m->tso->why_blocked) {
 940             case BlockedOnSTM:
 941             case BlockedOnBlackHole:
 942             case BlockedOnException:
 943             case BlockedOnMVar:
 944                 raiseAsync(m->tso, (StgClosure *)NonTermination_closure);
 945                 return;
 946             default:
 947                 barf("deadlock: main thread blocked in a strange way");
 948             }
 949         }
 950 #endif
 951     }
 952 }
 953
 954 /* ----------------------------------------------------------------------------
 955  * Process an event (GRAN only)
 956  * ------------------------------------------------------------------------- */
 957
 958 #if defined(GRAN)
 959 static StgTSO *
 960 scheduleProcessEvent(rtsEvent *event)
 961 {
 962     StgTSO *t;
 963
 964     if (RtsFlags.GranFlags.Light)
 965       GranSimLight_enter_system(event, &ActiveTSO); // adjust ActiveTSO etc
 966
 967     /* adjust time based on time-stamp */
 968     if (event->time > CurrentTime[CurrentProc] &&
 969         event->evttype != ContinueThread)
 970       CurrentTime[CurrentProc] = event->time;
 971
 972     /* Deal with the idle PEs (may issue FindWork or MoveSpark events) */
 973     if (!RtsFlags.GranFlags.Light)
 974       handleIdlePEs();
 975
 976     IF_DEBUG(gran, debugBelch("GRAN: switch by event-type\n"));
 977
 978     /* main event dispatcher in GranSim */
 979     switch (event->evttype) {
 980       /* Should just be continuing execution */
 981     case ContinueThread:
 982       IF_DEBUG(gran, debugBelch("GRAN: doing ContinueThread\n"));
 983       /* ToDo: check assertion
 984       ASSERT(run_queue_hd != (StgTSO*)NULL &&
 985              run_queue_hd != END_TSO_QUEUE);
 986       */
 987       /* Ignore ContinueThreads for fetching threads (if synchr comm) */
 988       if (!RtsFlags.GranFlags.DoAsyncFetch &&
 989           procStatus[CurrentProc]==Fetching) {
 990         debugBelch("ghuH: Spurious ContinueThread while Fetching ignored; TSO %d (%p) [PE %d]\n",
 991               CurrentTSO->id, CurrentTSO, CurrentProc);
 992         goto next_thread;
 993       }
 994       /* Ignore ContinueThreads for completed threads */
 995       if (CurrentTSO->what_next == ThreadComplete) {
 996         debugBelch("ghuH: found a ContinueThread event for completed thread %d (%p) [PE %d] (ignoring ContinueThread)\n",
 997               CurrentTSO->id, CurrentTSO, CurrentProc);
 998         goto next_thread;
 999       }
1000       /* Ignore ContinueThreads for threads that are being migrated */
1001       if (PROCS(CurrentTSO)==Nowhere) {
1002         debugBelch("ghuH: trying to run the migrating TSO %d (%p) [PE %d] (ignoring ContinueThread)\n",
1003               CurrentTSO->id, CurrentTSO, CurrentProc);
1004         goto next_thread;
1005       }
1006       /* The thread should be at the beginning of the run queue */
1007       if (CurrentTSO!=run_queue_hds[CurrentProc]) {
1008         debugBelch("ghuH: TSO %d (%p) [PE %d] is not at the start of the run_queue when doing a ContinueThread\n",
1009               CurrentTSO->id, CurrentTSO, CurrentProc);
1010         break; // run the thread anyway
1011       }
1012       /*
1013       new_event(proc, proc, CurrentTime[proc],
1014                 FindWork,
1015                 (StgTSO*)NULL, (StgClosure*)NULL, (rtsSpark*)NULL);
1016       goto next_thread;
1017       */ /* Catches superfluous CONTINUEs -- should be unnecessary */
1018       break; // now actually run the thread; DaH Qu'vam yImuHbej
1019
1020     case FetchNode:
1021       do_the_fetchnode(event);
1022       goto next_thread;             /* handle next event in event queue  */
1023
1024     case GlobalBlock:
1025       do_the_globalblock(event);
1026       goto next_thread;             /* handle next event in event queue  */
1027
1028     case FetchReply:
1029       do_the_fetchreply(event);
1030       goto next_thread;             /* handle next event in event queue  */
1031
1032     case UnblockThread:   /* Move from the blocked queue to the tail of */
1033       do_the_unblock(event);
1034       goto next_thread;             /* handle next event in event queue  */
1035
1036     case ResumeThread:  /* Move from the blocked queue to the tail of */
1037       /* the runnable queue ( i.e. Qu' SImqa'lu') */
1038       event->tso->gran.blocktime +=
1039         CurrentTime[CurrentProc] - event->tso->gran.blockedat;
1040       do_the_startthread(event);
1041       goto next_thread;             /* handle next event in event queue  */
1042
1043     case StartThread:
1044       do_the_startthread(event);
1045       goto next_thread;             /* handle next event in event queue  */
1046
1047     case MoveThread:
1048       do_the_movethread(event);
1049       goto next_thread;             /* handle next event in event queue  */
1050
1051     case MoveSpark:
1052       do_the_movespark(event);
1053       goto next_thread;             /* handle next event in event queue  */
1054
1055     case FindWork:
1056       do_the_findwork(event);
1057       goto next_thread;             /* handle next event in event queue  */
1058
1059     default:
1060       barf("Illegal event type %u\n", event->evttype);
1061     }  /* switch */
1062
1063     /* This point was scheduler_loop in the old RTS */
1064
1065     IF_DEBUG(gran, debugBelch("GRAN: after main switch\n"));
1066
1067     TimeOfLastEvent = CurrentTime[CurrentProc];
1068     TimeOfNextEvent = get_time_of_next_event();
1069     IgnoreEvents=(TimeOfNextEvent==0); // HWL HACK
1070     // CurrentTSO = ThreadQueueHd;
1071
1072     IF_DEBUG(gran, debugBelch("GRAN: time of next event is: %ld\n",
1073                          TimeOfNextEvent));
1074
1075     if (RtsFlags.GranFlags.Light)
1076       GranSimLight_leave_system(event, &ActiveTSO);
1077
1078     EndOfTimeSlice = CurrentTime[CurrentProc]+RtsFlags.GranFlags.time_slice;
1079
1080     IF_DEBUG(gran,
1081              debugBelch("GRAN: end of time-slice is %#lx\n", EndOfTimeSlice));
1082
1083     /* in a GranSim setup the TSO stays on the run queue */
1084     t = CurrentTSO;
1085     /* Take a thread from the run queue. */
1086     POP_RUN_QUEUE(t); // take_off_run_queue(t);
1087
1088     IF_DEBUG(gran,
1089              debugBelch("GRAN: About to run current thread, which is\n");
1090              G_TSO(t,5));
1091
1092     context_switch = 0; // turned on via GranYield, checking events and time slice
1093
1094     IF_DEBUG(gran,
1095              DumpGranEvent(GR_SCHEDULE, t));
1096
1097     procStatus[CurrentProc] = Busy;
1098 }
1099 #endif // GRAN
1100
1101 /* ----------------------------------------------------------------------------
1102  * Send pending messages (PARALLEL_HASKELL only)
1103  * ------------------------------------------------------------------------- */
1104
1105 #if defined(PARALLEL_HASKELL)
1106 static StgTSO *
1107 scheduleSendPendingMessages(void)
1108 {
1109     StgSparkPool *pool;
1110     rtsSpark spark;
1111     StgTSO *t;
1112
1113 # if defined(PAR) // global Mem.Mgmt., omit for now
1114     if (PendingFetches != END_BF_QUEUE) {
1115         processFetches();
1116     }
1117 # endif
1118
1119     if (RtsFlags.ParFlags.BufferTime) {
1120         // if we use message buffering, we must send away all message
1121         // packets which have become too old...
1122         sendOldBuffers();
1123     }
1124 }
1125 #endif
1126
1127 /* ----------------------------------------------------------------------------
1128  * Activate spark threads (PARALLEL_HASKELL only)
1129  * ------------------------------------------------------------------------- */
1130
1131 #if defined(PARALLEL_HASKELL)
1132 static void
1133 scheduleActivateSpark(void)
1134 {
1135 #if defined(SPARKS)
1136   ASSERT(EMPTY_RUN_QUEUE());
1137 /* We get here if the run queue is empty and want some work.
1138    We try to turn a spark into a thread, and add it to the run queue,
1139    from where it will be picked up in the next iteration of the scheduler
1140    loop.
1141 */
1142
1143       /* :-[  no local threads => look out for local sparks */
1144       /* the spark pool for the current PE */
1145       pool = &(cap.r.rSparks); // JB: cap = (old) MainCap
1146       if (advisory_thread_count < RtsFlags.ParFlags.maxThreads &&
1147           pool->hd < pool->tl) {
1148         /*
1149          * ToDo: add GC code check that we really have enough heap afterwards!!
1150          * Old comment:
1151          * If we're here (no runnable threads) and we have pending
1152          * sparks, we must have a space problem.  Get enough space
1153          * to turn one of those pending sparks into a
1154          * thread...
1155          */
1156
1157         spark = findSpark(rtsFalse);            /* get a spark */
1158         if (spark != (rtsSpark) NULL) {
1159           tso = createThreadFromSpark(spark);       /* turn the spark into a thread */
1160           IF_PAR_DEBUG(fish, // schedule,
1161                        debugBelch("==== schedule: Created TSO %d (%p); %d threads active\n",
1162                              tso->id, tso, advisory_thread_count));
1163
1164           if (tso==END_TSO_QUEUE) { /* failed to activate spark->back to loop */
1165             IF_PAR_DEBUG(fish, // schedule,
1166                          debugBelch("==^^ failed to create thread from spark @ %lx\n",
1167                             spark));
1168             return rtsFalse; /* failed to generate a thread */
1169           }                  /* otherwise fall through & pick-up new tso */
1170         } else {
1171           IF_PAR_DEBUG(fish, // schedule,
1172                        debugBelch("==^^ no local sparks (spark pool contains only NFs: %d)\n",
1173                              spark_queue_len(pool)));
1174           return rtsFalse;  /* failed to generate a thread */
1175         }
1176         return rtsTrue;  /* success in generating a thread */
1177   } else { /* no more threads permitted or pool empty */
1178     return rtsFalse;  /* failed to generateThread */
1179   }
1180 #else
1181   tso = NULL; // avoid compiler warning only
1182   return rtsFalse;  /* dummy in non-PAR setup */
1183 #endif // SPARKS
1184 }
1185 #endif // PARALLEL_HASKELL
1186
1187 /* ----------------------------------------------------------------------------
1188  * Get work from a remote node (PARALLEL_HASKELL only)
1189  * ------------------------------------------------------------------------- */
1190
1191 #if defined(PARALLEL_HASKELL)
1192 static rtsBool
1193 scheduleGetRemoteWork(rtsBool *receivedFinish)
1194 {
1195   ASSERT(EMPTY_RUN_QUEUE());
1196
1197   if (RtsFlags.ParFlags.BufferTime) {
1198         IF_PAR_DEBUG(verbose,
1199                 debugBelch("...send all pending data,"));
1200         {
1201           nat i;
1202           for (i=1; i<=nPEs; i++)
1203             sendImmediately(i); // send all messages away immediately
1204         }
1205   }
1206 # ifndef SPARKS
1207         //++EDEN++ idle() , i.e. send all buffers, wait for work
1208         // suppress fishing in EDEN... just look for incoming messages
1209         // (blocking receive)
1210   IF_PAR_DEBUG(verbose,
1211                debugBelch("...wait for incoming messages...\n"));
1212   *receivedFinish = processMessages(); // blocking receive...
1213
1214         // and reenter scheduling loop after having received something
1215         // (return rtsFalse below)
1216
1217 # else /* activate SPARKS machinery */
1218 /* We get here, if we have no work, tried to activate a local spark, but still
1219    have no work. We try to get a remote spark, by sending a FISH message.
1220    Thread migration should be added here, and triggered when a sequence of
1221    fishes returns without work. */
1222         delay = (RtsFlags.ParFlags.fishDelay!=0ll ? RtsFlags.ParFlags.fishDelay : 0ll);
1223
1224       /* =8-[  no local sparks => look for work on other PEs */
1225         /*
1226          * We really have absolutely no work.  Send out a fish
1227          * (there may be some out there already), and wait for
1228          * something to arrive.  We clearly can't run any threads
1229          * until a SCHEDULE or RESUME arrives, and so that's what
1230          * we're hoping to see.  (Of course, we still have to
1231          * respond to other types of messages.)
1232          */
1233         rtsTime now = msTime() /*CURRENT_TIME*/;
1234         IF_PAR_DEBUG(verbose,
1235                      debugBelch("--  now=%ld\n", now));
1236         IF_PAR_DEBUG(fish, // verbose,
1237              if (outstandingFishes < RtsFlags.ParFlags.maxFishes &&
1238                  (last_fish_arrived_at!=0 &&
1239                   last_fish_arrived_at+delay > now)) {
1240                debugBelch("--$$ <%llu> delaying FISH until %llu (last fish %llu, delay %llu)\n",
1241                      now, last_fish_arrived_at+delay,
1242                      last_fish_arrived_at,
1243                      delay);
1244              });
1245
1246         if (outstandingFishes < RtsFlags.ParFlags.maxFishes &&
1247             advisory_thread_count < RtsFlags.ParFlags.maxThreads) { // send a FISH, but when?
1248           if (last_fish_arrived_at==0 ||
1249               (last_fish_arrived_at+delay <= now)) {           // send FISH now!
1250             /* outstandingFishes is set in sendFish, processFish;
1251                avoid flooding system with fishes via delay */
1252     next_fish_to_send_at = 0;
1253   } else {
1254     /* ToDo: this should be done in the main scheduling loop to avoid the
1255              busy wait here; not so bad if fish delay is very small  */
1256     int iq = 0; // DEBUGGING -- HWL
1257     next_fish_to_send_at = last_fish_arrived_at+delay; // remember when to send
1258     /* send a fish when ready, but process messages that arrive in the meantime */
1259     do {
1260       if (PacketsWaiting()) {
1261         iq++; // DEBUGGING
1262         *receivedFinish = processMessages();
1263       }
1264       now = msTime();
1265     } while (!*receivedFinish || now<next_fish_to_send_at);
1266     // JB: This means the fish could become obsolete, if we receive
1267     // work. Better check for work again?
1268     // last line: while (!receivedFinish || !haveWork || now<...)
1269     // next line: if (receivedFinish || haveWork )
1270
1271     if (*receivedFinish) // no need to send a FISH if we are finishing anyway
1272       return rtsFalse;  // NB: this will leave scheduler loop
1273                         // immediately after return!
1274
1275     IF_PAR_DEBUG(fish, // verbose,
1276                debugBelch("--$$ <%llu> sent delayed fish (%d processMessages); active/total threads=%d/%d\n",now,iq,run_queue_len(),advisory_thread_count));
1277
1278   }
1279
1280     // JB: IMHO, this should all be hidden inside sendFish(...)
1281     /* pe = choosePE();
1282        sendFish(pe, thisPE, NEW_FISH_AGE, NEW_FISH_HISTORY,
1283                 NEW_FISH_HUNGER);
1284
1285     // Global statistics: count no. of fishes
1286     if (RtsFlags.ParFlags.ParStats.Global &&
1287          RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
1288            globalParStats.tot_fish_mess++;
1289            }
1290     */
1291
1292   /* delayed fishes must have been sent by now! */
1293   next_fish_to_send_at = 0;
1294   }
1295
1296   *receivedFinish = processMessages();
1297 # endif /* SPARKS */
1298
1299  return rtsFalse;
1300  /* NB: this function always returns rtsFalse, meaning the scheduler
1301     loop continues with the next iteration;
1302     rationale:
1303       return code means success in finding work; we enter this function
1304       if there is no local work, thus have to send a fish which takes
1305       time until it arrives with work; in the meantime we should process
1306       messages in the main loop;
1307  */
1308 }
1309 #endif // PARALLEL_HASKELL
1310
1311 /* ----------------------------------------------------------------------------
1312  * PAR/GRAN: Report stats & debugging info(?)
1313  * ------------------------------------------------------------------------- */
1314
1315 #if defined(PAR) || defined(GRAN)
1316 static void
1317 scheduleGranParReport(void)
1318 {
1319   ASSERT(run_queue_hd != END_TSO_QUEUE);
1320
1321   /* Take a thread from the run queue, if we have work */
1322   POP_RUN_QUEUE(t);  // take_off_run_queue(END_TSO_QUEUE);
1323
1324     /* If this TSO has got its outport closed in the meantime,
1325      *   it mustn't be run. Instead, we have to clean it up as if it was finished.
1326      * It has to be marked as TH_DEAD for this purpose.
1327      * If it is TH_TERM instead, it is supposed to have finished in the normal way.
1328
1329 JB: TODO: investigate wether state change field could be nuked
1330      entirely and replaced by the normal tso state (whatnext
1331      field). All we want to do is to kill tsos from outside.
1332      */
1333
1334     /* ToDo: write something to the log-file
1335     if (RTSflags.ParFlags.granSimStats && !sameThread)
1336         DumpGranEvent(GR_SCHEDULE, RunnableThreadsHd);
1337
1338     CurrentTSO = t;
1339     */
1340     /* the spark pool for the current PE */
1341     pool = &(cap.r.rSparks); //  cap = (old) MainCap
1342
1343     IF_DEBUG(scheduler,
1344              debugBelch("--=^ %d threads, %d sparks on [%#x]\n",
1345                    run_queue_len(), spark_queue_len(pool), CURRENT_PROC));
1346
1347     IF_PAR_DEBUG(fish,
1348              debugBelch("--=^ %d threads, %d sparks on [%#x]\n",
1349                    run_queue_len(), spark_queue_len(pool), CURRENT_PROC));
1350
1351     if (RtsFlags.ParFlags.ParStats.Full &&
1352         (t->par.sparkname != (StgInt)0) && // only log spark generated threads
1353         (emitSchedule || // forced emit
1354          (t && LastTSO && t->id != LastTSO->id))) {
1355       /*
1356          we are running a different TSO, so write a schedule event to log file
1357          NB: If we use fair scheduling we also have to write  a deschedule
1358              event for LastTSO; with unfair scheduling we know that the
1359              previous tso has blocked whenever we switch to another tso, so
1360              we don't need it in GUM for now
1361       */
1362       IF_PAR_DEBUG(fish, // schedule,
1363                    debugBelch("____ scheduling spark generated thread %d (%lx) (%lx) via a forced emit\n",t->id,t,t->par.sparkname));
1364
1365       DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
1366                        GR_SCHEDULE, t, (StgClosure *)NULL, 0, 0);
1367       emitSchedule = rtsFalse;
1368     }
1369 }
1370 #endif
1371
1372 /* ----------------------------------------------------------------------------
1373  * After running a thread...
1374  * ASSUMES: sched_mutex
1375  * ------------------------------------------------------------------------- */
1376
1377 static void
1378 schedulePostRunThread(void)
1379 {
1380 #if defined(PAR)
1381     /* HACK 675: if the last thread didn't yield, make sure to print a
1382        SCHEDULE event to the log file when StgRunning the next thread, even
1383        if it is the same one as before */
1384     LastTSO = t;
1385     TimeOfLastYield = CURRENT_TIME;
1386 #endif
1387
1388   /* some statistics gathering in the parallel case */
1389
1390 #if defined(GRAN) || defined(PAR) || defined(EDEN)
1391   switch (ret) {
1392     case HeapOverflow:
1393 # if defined(GRAN)
1394       IF_DEBUG(gran, DumpGranEvent(GR_DESCHEDULE, t));
1395       globalGranStats.tot_heapover++;
1396 # elif defined(PAR)
1397       globalParStats.tot_heapover++;
1398 # endif
1399       break;
1400
1401      case StackOverflow:
1402 # if defined(GRAN)
1403       IF_DEBUG(gran,
1404                DumpGranEvent(GR_DESCHEDULE, t));
1405       globalGranStats.tot_stackover++;
1406 # elif defined(PAR)
1407       // IF_DEBUG(par,
1408       // DumpGranEvent(GR_DESCHEDULE, t);
1409       globalParStats.tot_stackover++;
1410 # endif
1411       break;
1412
1413     case ThreadYielding:
1414 # if defined(GRAN)
1415       IF_DEBUG(gran,
1416                DumpGranEvent(GR_DESCHEDULE, t));
1417       globalGranStats.tot_yields++;
1418 # elif defined(PAR)
1419       // IF_DEBUG(par,
1420       // DumpGranEvent(GR_DESCHEDULE, t);
1421       globalParStats.tot_yields++;
1422 # endif
1423       break;
1424
1425     case ThreadBlocked:
1426 # if defined(GRAN)
1427       IF_DEBUG(scheduler,
1428                debugBelch("--<< thread %ld (%p; %s) stopped, blocking on node %p [PE %d] with BQ: ",
1429                           t->id, t, whatNext_strs[t->what_next], t->block_info.closure,
1430                           (t->block_info.closure==(StgClosure*)NULL ? 99 : where_is(t->block_info.closure)));
1431                if (t->block_info.closure!=(StgClosure*)NULL)
1432                  print_bq(t->block_info.closure);
1433                debugBelch("\n"));
1434
1435       // ??? needed; should emit block before
1436       IF_DEBUG(gran,
1437                DumpGranEvent(GR_DESCHEDULE, t));
1438       prune_eventq(t, (StgClosure *)NULL); // prune ContinueThreads for t
1439       /*
1440         ngoq Dogh!
1441       ASSERT(procStatus[CurrentProc]==Busy ||
1442               ((procStatus[CurrentProc]==Fetching) &&
1443               (t->block_info.closure!=(StgClosure*)NULL)));
1444       if (run_queue_hds[CurrentProc] == END_TSO_QUEUE &&
1445           !(!RtsFlags.GranFlags.DoAsyncFetch &&
1446             procStatus[CurrentProc]==Fetching))
1447         procStatus[CurrentProc] = Idle;
1448       */
1449 # elif defined(PAR)
1450 //++PAR++  blockThread() writes the event (change?)
1451 # endif
1452     break;
1453
1454   case ThreadFinished:
1455     break;
1456
1457   default:
1458     barf("parGlobalStats: unknown return code");
1459     break;
1460     }
1461 #endif
1462 }
1463
1464 /* -----------------------------------------------------------------------------
1465  * Handle a thread that returned to the scheduler with ThreadHeepOverflow
1466  * ASSUMES: sched_mutex
1467  * -------------------------------------------------------------------------- */
1468
1469 static rtsBool
1470 scheduleHandleHeapOverflow( Capability *cap, StgTSO *t )
1471 {
1472     // did the task ask for a large block?
1473     if (cap->r.rHpAlloc > BLOCK_SIZE) {
1474         // if so, get one and push it on the front of the nursery.
1475         bdescr *bd;
1476         lnat blocks;
1477
1478         blocks = (lnat)BLOCK_ROUND_UP(cap->r.rHpAlloc) / BLOCK_SIZE;
1479
1480         IF_DEBUG(scheduler,
1481                  debugBelch("--<< thread %ld (%s) stopped: requesting a large block (size %ld)\n",
1482                             (long)t->id, whatNext_strs[t->what_next], blocks));
1483
1484         // don't do this if the nursery is (nearly) full, we'll GC first.
1485         if (cap->r.rCurrentNursery->link != NULL ||
1486             cap->r.rNursery->n_blocks == 1) {  // paranoia to prevent infinite loop
1487                                                // if the nursery has only one block.
1488
1489             ACQUIRE_SM_LOCK
1490             bd = allocGroup( blocks );
1491             RELEASE_SM_LOCK
1492             cap->r.rNursery->n_blocks += blocks;
1493
1494             // link the new group into the list
1495             bd->link = cap->r.rCurrentNursery;
1496             bd->u.back = cap->r.rCurrentNursery->u.back;
1497             if (cap->r.rCurrentNursery->u.back != NULL) {
1498                 cap->r.rCurrentNursery->u.back->link = bd;
1499             } else {
1500 #if !defined(SMP)
1501                 ASSERT(g0s0->blocks == cap->r.rCurrentNursery &&
1502                        g0s0 == cap->r.rNursery);
1503 #endif
1504                 cap->r.rNursery->blocks = bd;
1505             }
1506             cap->r.rCurrentNursery->u.back = bd;
1507
1508             // initialise it as a nursery block.  We initialise the
1509             // step, gen_no, and flags field of *every* sub-block in
1510             // this large block, because this is easier than making
1511             // sure that we always find the block head of a large
1512             // block whenever we call Bdescr() (eg. evacuate() and
1513             // isAlive() in the GC would both have to do this, at
1514             // least).
1515             {
1516                 bdescr *x;
1517                 for (x = bd; x < bd + blocks; x++) {
1518                     x->step = cap->r.rNursery;
1519                     x->gen_no = 0;
1520                     x->flags = 0;
1521                 }
1522             }
1523
1524             // This assert can be a killer if the app is doing lots
1525             // of large block allocations.
1526             IF_DEBUG(sanity, checkNurserySanity(cap->r.rNursery));
1527
1528             // now update the nursery to point to the new block
1529             cap->r.rCurrentNursery = bd;
1530
1531             // we might be unlucky and have another thread get on the
1532             // run queue before us and steal the large block, but in that
1533             // case the thread will just end up requesting another large
1534             // block.
1535             PUSH_ON_RUN_QUEUE(t);
1536             return rtsFalse;  /* not actually GC'ing */
1537         }
1538     }
1539
1540     IF_DEBUG(scheduler,
1541              debugBelch("--<< thread %ld (%s) stopped: HeapOverflow\n",
1542                         (long)t->id, whatNext_strs[t->what_next]));
1543 #if defined(GRAN)
1544     ASSERT(!is_on_queue(t,CurrentProc));
1545 #elif defined(PARALLEL_HASKELL)
1546     /* Currently we emit a DESCHEDULE event before GC in GUM.
1547        ToDo: either add separate event to distinguish SYSTEM time from rest
1548        or just nuke this DESCHEDULE (and the following SCHEDULE) */
1549     if (0 && RtsFlags.ParFlags.ParStats.Full) {
1550         DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
1551                          GR_DESCHEDULE, t, (StgClosure *)NULL, 0, 0);
1552         emitSchedule = rtsTrue;
1553     }
1554 #endif
1555
1556     PUSH_ON_RUN_QUEUE(t);
1557     return rtsTrue;
1558     /* actual GC is done at the end of the while loop in schedule() */
1559 }
1560
1561 /* -----------------------------------------------------------------------------
1562  * Handle a thread that returned to the scheduler with ThreadStackOverflow
1563  * ASSUMES: sched_mutex
1564  * -------------------------------------------------------------------------- */
1565
1566 static void
1567 scheduleHandleStackOverflow( StgTSO *t)
1568 {
1569     IF_DEBUG(scheduler,debugBelch("--<< thread %ld (%s) stopped, StackOverflow\n",
1570                                   (long)t->id, whatNext_strs[t->what_next]));
1571     /* just adjust the stack for this thread, then pop it back
1572      * on the run queue.
1573      */
1574     {
1575         /* enlarge the stack */
1576         StgTSO *new_t = threadStackOverflow(t);
1577
1578         /* This TSO has moved, so update any pointers to it from the
1579          * main thread stack.  It better not be on any other queues...
1580          * (it shouldn't be).
1581          */
1582         if (t->main != NULL) {
1583             t->main->tso = new_t;
1584         }
1585         PUSH_ON_RUN_QUEUE(new_t);
1586     }
1587 }
1588
1589 /* -----------------------------------------------------------------------------
1590  * Handle a thread that returned to the scheduler with ThreadYielding
1591  * ASSUMES: sched_mutex
1592  * -------------------------------------------------------------------------- */
1593
1594 static rtsBool
1595 scheduleHandleYield( StgTSO *t, nat prev_what_next )
1596 {
1597     // Reset the context switch flag.  We don't do this just before
1598     // running the thread, because that would mean we would lose ticks
1599     // during GC, which can lead to unfair scheduling (a thread hogs
1600     // the CPU because the tick always arrives during GC).  This way
1601     // penalises threads that do a lot of allocation, but that seems
1602     // better than the alternative.
1603     context_switch = 0;
1604
1605     /* put the thread back on the run queue.  Then, if we're ready to
1606      * GC, check whether this is the last task to stop.  If so, wake
1607      * up the GC thread.  getThread will block during a GC until the
1608      * GC is finished.
1609      */
1610     IF_DEBUG(scheduler,
1611              if (t->what_next != prev_what_next) {
1612                  debugBelch("--<< thread %ld (%s) stopped to switch evaluators\n",
1613                             (long)t->id, whatNext_strs[t->what_next]);
1614              } else {
1615                  debugBelch("--<< thread %ld (%s) stopped, yielding\n",
1616                             (long)t->id, whatNext_strs[t->what_next]);
1617              }
1618         );
1619
1620     IF_DEBUG(sanity,
1621              //debugBelch("&& Doing sanity check on yielding TSO %ld.", t->id);
1622              checkTSO(t));
1623     ASSERT(t->link == END_TSO_QUEUE);
1624
1625     // Shortcut if we're just switching evaluators: don't bother
1626     // doing stack squeezing (which can be expensive), just run the
1627     // thread.
1628     if (t->what_next != prev_what_next) {
1629         return rtsTrue;
1630     }
1631
1632 #if defined(GRAN)
1633     ASSERT(!is_on_queue(t,CurrentProc));
1634
1635     IF_DEBUG(sanity,
1636              //debugBelch("&& Doing sanity check on all ThreadQueues (and their TSOs).");
1637              checkThreadQsSanity(rtsTrue));
1638
1639 #endif
1640
1641     addToRunQueue(t);
1642
1643 #if defined(GRAN)
1644     /* add a ContinueThread event to actually process the thread */
1645     new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
1646               ContinueThread,
1647               t, (StgClosure*)NULL, (rtsSpark*)NULL);
1648     IF_GRAN_DEBUG(bq,
1649                   debugBelch("GRAN: eventq and runnableq after adding yielded thread to queue again:\n");
1650                   G_EVENTQ(0);
1651                   G_CURR_THREADQ(0));
1652 #endif
1653     return rtsFalse;
1654 }
1655
1656 /* -----------------------------------------------------------------------------
1657  * Handle a thread that returned to the scheduler with ThreadBlocked
1658  * ASSUMES: sched_mutex
1659  * -------------------------------------------------------------------------- */
1660
1661 static void
1662 scheduleHandleThreadBlocked( StgTSO *t
1663 #if !defined(GRAN) && !defined(DEBUG)
1664     STG_UNUSED
1665 #endif
1666     )
1667 {
1668 #if defined(GRAN)
1669     IF_DEBUG(scheduler,
1670              debugBelch("--<< thread %ld (%p; %s) stopped, blocking on node %p [PE %d] with BQ: \n",
1671                         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)));
1672              if (t->block_info.closure!=(StgClosure*)NULL) print_bq(t->block_info.closure));
1673
1674     // ??? needed; should emit block before
1675     IF_DEBUG(gran,
1676              DumpGranEvent(GR_DESCHEDULE, t));
1677     prune_eventq(t, (StgClosure *)NULL); // prune ContinueThreads for t
1678     /*
1679       ngoq Dogh!
1680       ASSERT(procStatus[CurrentProc]==Busy ||
1681       ((procStatus[CurrentProc]==Fetching) &&
1682       (t->block_info.closure!=(StgClosure*)NULL)));
1683       if (run_queue_hds[CurrentProc] == END_TSO_QUEUE &&
1684       !(!RtsFlags.GranFlags.DoAsyncFetch &&
1685       procStatus[CurrentProc]==Fetching))
1686       procStatus[CurrentProc] = Idle;
1687     */
1688 #elif defined(PAR)
1689     IF_DEBUG(scheduler,
1690              debugBelch("--<< thread %ld (%p; %s) stopped, blocking on node %p with BQ: \n",
1691                         t->id, t, whatNext_strs[t->what_next], t->block_info.closure));
1692     IF_PAR_DEBUG(bq,
1693
1694                  if (t->block_info.closure!=(StgClosure*)NULL)
1695                  print_bq(t->block_info.closure));
1696
1697     /* Send a fetch (if BlockedOnGA) and dump event to log file */
1698     blockThread(t);
1699
1700     /* whatever we schedule next, we must log that schedule */
1701     emitSchedule = rtsTrue;
1702
1703 #else /* !GRAN */
1704
1705       // We don't need to do anything.  The thread is blocked, and it
1706       // has tidied up its stack and placed itself on whatever queue
1707       // it needs to be on.
1708
1709 #if !defined(SMP)
1710     ASSERT(t->why_blocked != NotBlocked);
1711              // This might not be true under SMP: we don't have
1712              // exclusive access to this TSO, so someone might have
1713              // woken it up by now.  This actually happens: try
1714              // conc023 +RTS -N2.
1715 #endif
1716
1717     IF_DEBUG(scheduler,
1718              debugBelch("--<< thread %d (%s) stopped: ",
1719                         t->id, whatNext_strs[t->what_next]);
1720              printThreadBlockage(t);
1721              debugBelch("\n"));
1722
1723     /* Only for dumping event to log file
1724        ToDo: do I need this in GranSim, too?
1725        blockThread(t);
1726     */
1727 #endif
1728 }
1729
1730 /* -----------------------------------------------------------------------------
1731  * Handle a thread that returned to the scheduler with ThreadFinished
1732  * ASSUMES: sched_mutex
1733  * -------------------------------------------------------------------------- */
1734
1735 static rtsBool
1736 scheduleHandleThreadFinished( StgMainThread *mainThread
1737                               USED_WHEN_RTS_SUPPORTS_THREADS,
1738                               Capability *cap,
1739                               StgTSO *t )
1740 {
1741     /* Need to check whether this was a main thread, and if so,
1742      * return with the return value.
1743      *
1744      * We also end up here if the thread kills itself with an
1745      * uncaught exception, see Exception.cmm.
1746      */
1747     IF_DEBUG(scheduler,debugBelch("--++ thread %d (%s) finished\n",
1748                                   t->id, whatNext_strs[t->what_next]));
1749
1750 #if defined(GRAN)
1751       endThread(t, CurrentProc); // clean-up the thread
1752 #elif defined(PARALLEL_HASKELL)
1753       /* For now all are advisory -- HWL */
1754       //if(t->priority==AdvisoryPriority) ??
1755       advisory_thread_count--; // JB: Caution with this counter, buggy!
1756
1757 # if defined(DIST)
1758       if(t->dist.priority==RevalPriority)
1759         FinishReval(t);
1760 # endif
1761
1762 # if defined(EDENOLD)
1763       // the thread could still have an outport... (BUG)
1764       if (t->eden.outport != -1) {
1765       // delete the outport for the tso which has finished...
1766         IF_PAR_DEBUG(eden_ports,
1767                    debugBelch("WARNING: Scheduler removes outport %d for TSO %d.\n",
1768                               t->eden.outport, t->id));
1769         deleteOPT(t);
1770       }
1771       // thread still in the process (HEAVY BUG! since outport has just been closed...)
1772       if (t->eden.epid != -1) {
1773         IF_PAR_DEBUG(eden_ports,
1774                    debugBelch("WARNING: Scheduler removes TSO %d from process %d .\n",
1775                            t->id, t->eden.epid));
1776         removeTSOfromProcess(t);
1777       }
1778 # endif
1779
1780 # if defined(PAR)
1781       if (RtsFlags.ParFlags.ParStats.Full &&
1782           !RtsFlags.ParFlags.ParStats.Suppressed)
1783         DumpEndEvent(CURRENT_PROC, t, rtsFalse /* not mandatory */);
1784
1785       //  t->par only contains statistics: left out for now...
1786       IF_PAR_DEBUG(fish,
1787                    debugBelch("**** end thread: ended sparked thread %d (%lx); sparkname: %lx\n",
1788                               t->id,t,t->par.sparkname));
1789 # endif
1790 #endif // PARALLEL_HASKELL
1791
1792       //
1793       // Check whether the thread that just completed was a main
1794       // thread, and if so return with the result.
1795       //
1796       // There is an assumption here that all thread completion goes
1797       // through this point; we need to make sure that if a thread
1798       // ends up in the ThreadKilled state, that it stays on the run
1799       // queue so it can be dealt with here.
1800       //
1801       if (
1802 #if defined(RTS_SUPPORTS_THREADS)
1803           mainThread != NULL
1804 #else
1805           mainThread->tso == t
1806 #endif
1807           )
1808       {
1809           // We are a bound thread: this must be our thread that just
1810           // completed.
1811           ASSERT(mainThread->tso == t);
1812
1813           if (t->what_next == ThreadComplete) {
1814               if (mainThread->ret) {
1815                   // NOTE: return val is tso->sp[1] (see StgStartup.hc)
1816                   *(mainThread->ret) = (StgClosure *)mainThread->tso->sp[1];
1817               }
1818               mainThread->stat = Success;
1819           } else {
1820               if (mainThread->ret) {
1821                   *(mainThread->ret) = NULL;
1822               }
1823               if (interrupted) {
1824                   mainThread->stat = Interrupted;
1825               } else {
1826                   mainThread->stat = Killed;
1827               }
1828           }
1829 #ifdef DEBUG
1830           removeThreadLabel((StgWord)mainThread->tso->id);
1831 #endif
1832           if (mainThread->prev == NULL) {
1833               ASSERT(mainThread == main_threads);
1834               main_threads = mainThread->link;
1835           } else {
1836               mainThread->prev->link = mainThread->link;
1837           }
1838           if (mainThread->link != NULL) {
1839               mainThread->link->prev = mainThread->prev;
1840           }
1841           releaseCapability(cap);
1842           return rtsTrue; // tells schedule() to return
1843       }
1844
1845 #ifdef RTS_SUPPORTS_THREADS
1846       ASSERT(t->main == NULL);
1847 #else
1848       if (t->main != NULL) {
1849           // Must be a main thread that is not the topmost one.  Leave
1850           // it on the run queue until the stack has unwound to the
1851           // point where we can deal with this.  Leaving it on the run
1852           // queue also ensures that the garbage collector knows about
1853           // this thread and its return value (it gets dropped from the
1854           // all_threads list so there's no other way to find it).
1855           APPEND_TO_RUN_QUEUE(t);
1856       }
1857 #endif
1858       return rtsFalse;
1859 }
1860
1861 /* -----------------------------------------------------------------------------
1862  * Perform a heap census, if PROFILING
1863  * -------------------------------------------------------------------------- */
1864
1865 static rtsBool
1866 scheduleDoHeapProfile( rtsBool ready_to_gc STG_UNUSED )
1867 {
1868 #if defined(PROFILING)
1869     // When we have +RTS -i0 and we're heap profiling, do a census at
1870     // every GC.  This lets us get repeatable runs for debugging.
1871     if (performHeapProfile ||
1872         (RtsFlags.ProfFlags.profileInterval==0 &&
1873          RtsFlags.ProfFlags.doHeapProfile && ready_to_gc)) {
1874         GarbageCollect(GetRoots, rtsTrue);
1875         heapCensus();
1876         performHeapProfile = rtsFalse;
1877         return rtsTrue;  // true <=> we already GC'd
1878     }
1879 #endif
1880     return rtsFalse;
1881 }
1882
1883 /* -----------------------------------------------------------------------------
1884  * Perform a garbage collection if necessary
1885  * ASSUMES: sched_mutex
1886  * -------------------------------------------------------------------------- */
1887
1888 static void
1889 scheduleDoGC( rtsBool force_major )
1890 {
1891     StgTSO *t;
1892 #ifdef SMP
1893     Capability *cap;
1894     static rtsBool waiting_for_gc;
1895     int n_capabilities = RtsFlags.ParFlags.nNodes - 1;
1896            // subtract one because we're already holding one.
1897     Capability *caps[n_capabilities];
1898 #endif
1899
1900 #ifdef SMP
1901     // In order to GC, there must be no threads running Haskell code.
1902     // Therefore, the GC thread needs to hold *all* the capabilities,
1903     // and release them after the GC has completed.
1904     //
1905     // This seems to be the simplest way: previous attempts involved
1906     // making all the threads with capabilities give up their
1907     // capabilities and sleep except for the *last* one, which
1908     // actually did the GC.  But it's quite hard to arrange for all
1909     // the other tasks to sleep and stay asleep.
1910     //
1911     // This does mean that there will be multiple entries in the
1912     // thread->capability hash table for the current thread, but
1913     // they will be removed as normal when the capabilities are
1914     // released again.
1915     //
1916
1917     // Someone else is already trying to GC
1918     if (waiting_for_gc) return;
1919     waiting_for_gc = rtsTrue;
1920
1921     while (n_capabilities > 0) {
1922         IF_DEBUG(scheduler, sched_belch("ready_to_gc, grabbing all the capabilies (%d left)", n_capabilities));
1923         waitForReturnCapability(&sched_mutex, &cap);
1924         n_capabilities--;
1925         caps[n_capabilities] = cap;
1926     }
1927
1928     waiting_for_gc = rtsFalse;
1929 #endif
1930
1931     /* Kick any transactions which are invalid back to their
1932      * atomically frames.  When next scheduled they will try to
1933      * commit, this commit will fail and they will retry.
1934      */
1935     {
1936         StgTSO *next;
1937
1938         for (t = all_threads; t != END_TSO_QUEUE; t = next) {
1939             if (t->what_next == ThreadRelocated) {
1940                 next = t->link;
1941             } else {
1942                 next = t->global_link;
1943                 if (t -> trec != NO_TREC && t -> why_blocked == NotBlocked) {
1944                     if (!stmValidateNestOfTransactions (t -> trec)) {
1945                         IF_DEBUG(stm, sched_belch("trec %p found wasting its time", t));
1946
1947                         // strip the stack back to the ATOMICALLY_FRAME, aborting
1948                         // the (nested) transaction, and saving the stack of any
1949                         // partially-evaluated thunks on the heap.
1950                         raiseAsync_(t, NULL, rtsTrue);
1951
1952 #ifdef REG_R1
1953                         ASSERT(get_itbl((StgClosure *)t->sp)->type == ATOMICALLY_FRAME);
1954 #endif
1955                     }
1956                 }
1957             }
1958         }
1959     }
1960
1961     // so this happens periodically:
1962     scheduleCheckBlackHoles();
1963
1964     IF_DEBUG(scheduler, printAllThreads());
1965
1966     /* everybody back, start the GC.
1967      * Could do it in this thread, or signal a condition var
1968      * to do it in another thread.  Either way, we need to
1969      * broadcast on gc_pending_cond afterward.
1970      */
1971 #if defined(RTS_SUPPORTS_THREADS)
1972     IF_DEBUG(scheduler,sched_belch("doing GC"));
1973 #endif
1974     GarbageCollect(GetRoots, force_major);
1975
1976 #if defined(SMP)
1977     {
1978         // release our stash of capabilities.
1979         nat i;
1980         for (i = 0; i < RtsFlags.ParFlags.nNodes-1; i++) {
1981             releaseCapability(caps[i]);
1982         }
1983     }
1984 #endif
1985
1986 #if defined(GRAN)
1987     /* add a ContinueThread event to continue execution of current thread */
1988     new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
1989               ContinueThread,
1990               t, (StgClosure*)NULL, (rtsSpark*)NULL);
1991     IF_GRAN_DEBUG(bq,
1992                   debugBelch("GRAN: eventq and runnableq after Garbage collection:\n\n");
1993                   G_EVENTQ(0);
1994                   G_CURR_THREADQ(0));
1995 #endif /* GRAN */
1996 }
1997
1998 /* ---------------------------------------------------------------------------
1999  * rtsSupportsBoundThreads(): is the RTS built to support bound threads?
2000  * used by Control.Concurrent for error checking.
2001  * ------------------------------------------------------------------------- */
2002
2003 StgBool
2004 rtsSupportsBoundThreads(void)
2005 {
2006 #if defined(RTS_SUPPORTS_THREADS)
2007   return rtsTrue;
2008 #else
2009   return rtsFalse;
2010 #endif
2011 }
2012
2013 /* ---------------------------------------------------------------------------
2014  * isThreadBound(tso): check whether tso is bound to an OS thread.
2015  * ------------------------------------------------------------------------- */
2016
2017 StgBool
2018 isThreadBound(StgTSO* tso USED_IN_THREADED_RTS)
2019 {
2020 #if defined(RTS_SUPPORTS_THREADS)
2021   return (tso->main != NULL);
2022 #endif
2023   return rtsFalse;
2024 }
2025
2026 /* ---------------------------------------------------------------------------
2027  * Singleton fork(). Do not copy any running threads.
2028  * ------------------------------------------------------------------------- */
2029
2030 #ifndef mingw32_HOST_OS
2031 #define FORKPROCESS_PRIMOP_SUPPORTED
2032 #endif
2033
2034 #ifdef FORKPROCESS_PRIMOP_SUPPORTED
2035 static void
2036 deleteThreadImmediately(StgTSO *tso);
2037 #endif
2038 StgInt
2039 forkProcess(HsStablePtr *entry
2040 #ifndef FORKPROCESS_PRIMOP_SUPPORTED
2041             STG_UNUSED
2042 #endif
2043            )
2044 {
2045 #ifdef FORKPROCESS_PRIMOP_SUPPORTED
2046   pid_t pid;
2047   StgTSO* t,*next;
2048   StgMainThread *m;
2049   SchedulerStatus rc;
2050
2051   IF_DEBUG(scheduler,sched_belch("forking!"));
2052   rts_lock(); // This not only acquires sched_mutex, it also
2053               // makes sure that no other threads are running
2054
2055   pid = fork();
2056
2057   if (pid) { /* parent */
2058
2059   /* just return the pid */
2060     rts_unlock();
2061     return pid;
2062
2063   } else { /* child */
2064
2065
2066       // delete all threads
2067     run_queue_hd = run_queue_tl = END_TSO_QUEUE;
2068
2069     for (t = all_threads; t != END_TSO_QUEUE; t = next) {
2070       next = t->link;
2071
2072         // don't allow threads to catch the ThreadKilled exception
2073       deleteThreadImmediately(t);
2074     }
2075
2076       // wipe the main thread list
2077     while((m = main_threads) != NULL) {
2078       main_threads = m->link;
2079 # ifdef THREADED_RTS
2080       closeCondition(&m->bound_thread_cond);
2081 # endif
2082       stgFree(m);
2083     }
2084
2085     rc = rts_evalStableIO(entry, NULL);  // run the action
2086     rts_checkSchedStatus("forkProcess",rc);
2087
2088     rts_unlock();
2089
2090     hs_exit();                      // clean up and exit
2091     stg_exit(0);
2092   }
2093 #else /* !FORKPROCESS_PRIMOP_SUPPORTED */
2094   barf("forkProcess#: primop not supported, sorry!\n");
2095   return -1;
2096 #endif
2097 }
2098
2099 /* ---------------------------------------------------------------------------
2100  * deleteAllThreads():  kill all the live threads.
2101  *
2102  * This is used when we catch a user interrupt (^C), before performing
2103  * any necessary cleanups and running finalizers.
2104  *
2105  * Locks: sched_mutex held.
2106  * ------------------------------------------------------------------------- */
2107
2108 void
2109 deleteAllThreads ( void )
2110 {
2111   StgTSO* t, *next;
2112   IF_DEBUG(scheduler,sched_belch("deleting all threads"));
2113   for (t = all_threads; t != END_TSO_QUEUE; t = next) {
2114       if (t->what_next == ThreadRelocated) {
2115           next = t->link;
2116       } else {
2117           next = t->global_link;
2118           deleteThread(t);
2119       }
2120   }
2121
2122   // The run queue now contains a bunch of ThreadKilled threads.  We
2123   // must not throw these away: the main thread(s) will be in there
2124   // somewhere, and the main scheduler loop has to deal with it.
2125   // Also, the run queue is the only thing keeping these threads from
2126   // being GC'd, and we don't want the "main thread has been GC'd" panic.
2127
2128   ASSERT(blocked_queue_hd == END_TSO_QUEUE);
2129   ASSERT(blackhole_queue == END_TSO_QUEUE);
2130   ASSERT(sleeping_queue == END_TSO_QUEUE);
2131 }
2132
2133 /* startThread and  insertThread are now in GranSim.c -- HWL */
2134
2135
2136 /* ---------------------------------------------------------------------------
2137  * Suspending & resuming Haskell threads.
2138  *
2139  * When making a "safe" call to C (aka _ccall_GC), the task gives back
2140  * its capability before calling the C function.  This allows another
2141  * task to pick up the capability and carry on running Haskell
2142  * threads.  It also means that if the C call blocks, it won't lock
2143  * the whole system.
2144  *
2145  * The Haskell thread making the C call is put to sleep for the
2146  * duration of the call, on the susepended_ccalling_threads queue.  We
2147  * give out a token to the task, which it can use to resume the thread
2148  * on return from the C function.
2149  * ------------------------------------------------------------------------- */
2150
2151 StgInt
2152 suspendThread( StgRegTable *reg )
2153 {
2154   nat tok;
2155   Capability *cap;
2156   int saved_errno = errno;
2157
2158   /* assume that *reg is a pointer to the StgRegTable part
2159    * of a Capability.
2160    */
2161   cap = (Capability *)((void *)((unsigned char*)reg - sizeof(StgFunTable)));
2162
2163   ACQUIRE_LOCK(&sched_mutex);
2164
2165   IF_DEBUG(scheduler,
2166            sched_belch("thread %d did a _ccall_gc", cap->r.rCurrentTSO->id));
2167
2168   // XXX this might not be necessary --SDM
2169   cap->r.rCurrentTSO->what_next = ThreadRunGHC;
2170
2171   threadPaused(cap->r.rCurrentTSO);
2172   cap->r.rCurrentTSO->link = suspended_ccalling_threads;
2173   suspended_ccalling_threads = cap->r.rCurrentTSO;
2174
2175   if(cap->r.rCurrentTSO->blocked_exceptions == NULL)  {
2176       cap->r.rCurrentTSO->why_blocked = BlockedOnCCall;
2177       cap->r.rCurrentTSO->blocked_exceptions = END_TSO_QUEUE;
2178   } else {
2179       cap->r.rCurrentTSO->why_blocked = BlockedOnCCall_NoUnblockExc;
2180   }
2181
2182   /* Use the thread ID as the token; it should be unique */
2183   tok = cap->r.rCurrentTSO->id;
2184
2185   /* Hand back capability */
2186   cap->r.rInHaskell = rtsFalse;
2187   releaseCapability(cap);
2188
2189 #if defined(RTS_SUPPORTS_THREADS)
2190   /* Preparing to leave the RTS, so ensure there's a native thread/task
2191      waiting to take over.
2192   */
2193   IF_DEBUG(scheduler, sched_belch("worker (token %d): leaving RTS", tok));
2194 #endif
2195
2196   RELEASE_LOCK(&sched_mutex);
2197
2198   errno = saved_errno;
2199   return tok;
2200 }
2201
2202 StgRegTable *
2203 resumeThread( StgInt tok )
2204 {
2205   StgTSO *tso, **prev;
2206   Capability *cap;
2207   int saved_errno = errno;
2208
2209 #if defined(RTS_SUPPORTS_THREADS)
2210   /* Wait for permission to re-enter the RTS with the result. */
2211   ACQUIRE_LOCK(&sched_mutex);
2212   waitForReturnCapability(&sched_mutex, &cap);
2213
2214   IF_DEBUG(scheduler, sched_belch("worker (token %d): re-entering RTS", tok));
2215 #else
2216   grabCapability(&cap);
2217 #endif
2218
2219   /* Remove the thread off of the suspended list */
2220   prev = &suspended_ccalling_threads;
2221   for (tso = suspended_ccalling_threads;
2222        tso != END_TSO_QUEUE;
2223        prev = &tso->link, tso = tso->link) {
2224     if (tso->id == (StgThreadID)tok) {
2225       *prev = tso->link;
2226       break;
2227     }
2228   }
2229   if (tso == END_TSO_QUEUE) {
2230     barf("resumeThread: thread not found");
2231   }
2232   tso->link = END_TSO_QUEUE;
2233
2234   if(tso->why_blocked == BlockedOnCCall) {
2235       awakenBlockedQueueNoLock(tso->blocked_exceptions);
2236       tso->blocked_exceptions = NULL;
2237   }
2238
2239   /* Reset blocking status */
2240   tso->why_blocked  = NotBlocked;
2241
2242   cap->r.rCurrentTSO = tso;
2243   cap->r.rInHaskell = rtsTrue;
2244   RELEASE_LOCK(&sched_mutex);
2245   errno = saved_errno;
2246   return &cap->r;
2247 }
2248
2249 /* ---------------------------------------------------------------------------
2250  * Comparing Thread ids.
2251  *
2252  * This is used from STG land in the implementation of the
2253  * instances of Eq/Ord for ThreadIds.
2254  * ------------------------------------------------------------------------ */
2255
2256 int
2257 cmp_thread(StgPtr tso1, StgPtr tso2)
2258 {
2259   StgThreadID id1 = ((StgTSO *)tso1)->id;
2260   StgThreadID id2 = ((StgTSO *)tso2)->id;
2261
2262   if (id1 < id2) return (-1);
2263   if (id1 > id2) return 1;
2264   return 0;
2265 }
2266
2267 /* ---------------------------------------------------------------------------
2268  * Fetching the ThreadID from an StgTSO.
2269  *
2270  * This is used in the implementation of Show for ThreadIds.
2271  * ------------------------------------------------------------------------ */
2272 int
2273 rts_getThreadId(StgPtr tso)
2274 {
2275   return ((StgTSO *)tso)->id;
2276 }
2277
2278 #ifdef DEBUG
2279 void
2280 labelThread(StgPtr tso, char *label)
2281 {
2282   int len;
2283   void *buf;
2284
2285   /* Caveat: Once set, you can only set the thread name to "" */
2286   len = strlen(label)+1;
2287   buf = stgMallocBytes(len * sizeof(char), "Schedule.c:labelThread()");
2288   strncpy(buf,label,len);
2289   /* Update will free the old memory for us */
2290   updateThreadLabel(((StgTSO *)tso)->id,buf);
2291 }
2292 #endif /* DEBUG */
2293
2294 /* ---------------------------------------------------------------------------
2295    Create a new thread.
2296
2297    The new thread starts with the given stack size.  Before the
2298    scheduler can run, however, this thread needs to have a closure
2299    (and possibly some arguments) pushed on its stack.  See
2300    pushClosure() in Schedule.h.
2301
2302    createGenThread() and createIOThread() (in SchedAPI.h) are
2303    convenient packaged versions of this function.
2304
2305    currently pri (priority) is only used in a GRAN setup -- HWL
2306    ------------------------------------------------------------------------ */
2307 #if defined(GRAN)
2308 /*   currently pri (priority) is only used in a GRAN setup -- HWL */
2309 StgTSO *
2310 createThread(nat size, StgInt pri)
2311 #else
2312 StgTSO *
2313 createThread(nat size)
2314 #endif
2315 {
2316     StgTSO *tso;
2317     nat stack_size;
2318
2319     /* First check whether we should create a thread at all */
2320 #if defined(PARALLEL_HASKELL)
2321   /* check that no more than RtsFlags.ParFlags.maxThreads threads are created */
2322   if (advisory_thread_count >= RtsFlags.ParFlags.maxThreads) {
2323     threadsIgnored++;
2324     debugBelch("{createThread}Daq ghuH: refusing to create another thread; no more than %d threads allowed (currently %d)\n",
2325           RtsFlags.ParFlags.maxThreads, advisory_thread_count);
2326     return END_TSO_QUEUE;
2327   }
2328   threadsCreated++;
2329 #endif
2330
2331 #if defined(GRAN)
2332   ASSERT(!RtsFlags.GranFlags.Light || CurrentProc==0);
2333 #endif
2334
2335   // ToDo: check whether size = stack_size - TSO_STRUCT_SIZEW
2336
2337   /* catch ridiculously small stack sizes */
2338   if (size < MIN_STACK_WORDS + TSO_STRUCT_SIZEW) {
2339     size = MIN_STACK_WORDS + TSO_STRUCT_SIZEW;
2340   }
2341
2342   stack_size = size - TSO_STRUCT_SIZEW;
2343
2344   tso = (StgTSO *)allocate(size);
2345   TICK_ALLOC_TSO(stack_size, 0);
2346
2347   SET_HDR(tso, &stg_TSO_info, CCS_SYSTEM);
2348 #if defined(GRAN)
2349   SET_GRAN_HDR(tso, ThisPE);
2350 #endif
2351
2352   // Always start with the compiled code evaluator
2353   tso->what_next = ThreadRunGHC;
2354
2355   tso->id = next_thread_id++;
2356   tso->why_blocked  = NotBlocked;
2357   tso->blocked_exceptions = NULL;
2358
2359   tso->saved_errno = 0;
2360   tso->main = NULL;
2361
2362   tso->stack_size   = stack_size;
2363   tso->max_stack_size = round_to_mblocks(RtsFlags.GcFlags.maxStkSize)
2364                               - TSO_STRUCT_SIZEW;
2365   tso->sp           = (P_)&(tso->stack) + stack_size;
2366
2367   tso->trec = NO_TREC;
2368
2369 #ifdef PROFILING
2370   tso->prof.CCCS = CCS_MAIN;
2371 #endif
2372
2373   /* put a stop frame on the stack */
2374   tso->sp -= sizeofW(StgStopFrame);
2375   SET_HDR((StgClosure*)tso->sp,(StgInfoTable *)&stg_stop_thread_info,CCS_SYSTEM);
2376   tso->link = END_TSO_QUEUE;
2377
2378   // ToDo: check this
2379 #if defined(GRAN)
2380   /* uses more flexible routine in GranSim */
2381   insertThread(tso, CurrentProc);
2382 #else
2383   /* In a non-GranSim setup the pushing of a TSO onto the runq is separated
2384    * from its creation
2385    */
2386 #endif
2387
2388 #if defined(GRAN)
2389   if (RtsFlags.GranFlags.GranSimStats.Full)
2390     DumpGranEvent(GR_START,tso);
2391 #elif defined(PARALLEL_HASKELL)
2392   if (RtsFlags.ParFlags.ParStats.Full)
2393     DumpGranEvent(GR_STARTQ,tso);
2394   /* HACk to avoid SCHEDULE
2395      LastTSO = tso; */
2396 #endif
2397
2398   /* Link the new thread on the global thread list.
2399    */
2400   tso->global_link = all_threads;
2401   all_threads = tso;
2402
2403 #if defined(DIST)
2404   tso->dist.priority = MandatoryPriority; //by default that is...
2405 #endif
2406
2407 #if defined(GRAN)
2408   tso->gran.pri = pri;
2409 # if defined(DEBUG)
2410   tso->gran.magic = TSO_MAGIC; // debugging only
2411 # endif
2412   tso->gran.sparkname   = 0;
2413   tso->gran.startedat   = CURRENT_TIME;
2414   tso->gran.exported    = 0;
2415   tso->gran.basicblocks = 0;
2416   tso->gran.allocs      = 0;
2417   tso->gran.exectime    = 0;
2418   tso->gran.fetchtime   = 0;
2419   tso->gran.fetchcount  = 0;
2420   tso->gran.blocktime   = 0;
2421   tso->gran.blockcount  = 0;
2422   tso->gran.blockedat   = 0;
2423   tso->gran.globalsparks = 0;
2424   tso->gran.localsparks  = 0;
2425   if (RtsFlags.GranFlags.Light)
2426     tso->gran.clock  = Now; /* local clock */
2427   else
2428     tso->gran.clock  = 0;
2429
2430   IF_DEBUG(gran,printTSO(tso));
2431 #elif defined(PARALLEL_HASKELL)
2432 # if defined(DEBUG)
2433   tso->par.magic = TSO_MAGIC; // debugging only
2434 # endif
2435   tso->par.sparkname   = 0;
2436   tso->par.startedat   = CURRENT_TIME;
2437   tso->par.exported    = 0;
2438   tso->par.basicblocks = 0;
2439   tso->par.allocs      = 0;
2440   tso->par.exectime    = 0;
2441   tso->par.fetchtime   = 0;
2442   tso->par.fetchcount  = 0;
2443   tso->par.blocktime   = 0;
2444   tso->par.blockcount  = 0;
2445   tso->par.blockedat   = 0;
2446   tso->par.globalsparks = 0;
2447   tso->par.localsparks  = 0;
2448 #endif
2449
2450 #if defined(GRAN)
2451   globalGranStats.tot_threads_created++;
2452   globalGranStats.threads_created_on_PE[CurrentProc]++;
2453   globalGranStats.tot_sq_len += spark_queue_len(CurrentProc);
2454   globalGranStats.tot_sq_probes++;
2455 #elif defined(PARALLEL_HASKELL)
2456   // collect parallel global statistics (currently done together with GC stats)
2457   if (RtsFlags.ParFlags.ParStats.Global &&
2458       RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
2459     //debugBelch("Creating thread %d @ %11.2f\n", tso->id, usertime());
2460     globalParStats.tot_threads_created++;
2461   }
2462 #endif
2463
2464 #if defined(GRAN)
2465   IF_GRAN_DEBUG(pri,
2466                 sched_belch("==__ schedule: Created TSO %d (%p);",
2467                       CurrentProc, tso, tso->id));
2468 #elif defined(PARALLEL_HASKELL)
2469   IF_PAR_DEBUG(verbose,
2470                sched_belch("==__ schedule: Created TSO %d (%p); %d threads active",
2471                            (long)tso->id, tso, advisory_thread_count));
2472 #else
2473   IF_DEBUG(scheduler,sched_belch("created thread %ld, stack size = %lx words",
2474                                  (long)tso->id, (long)tso->stack_size));
2475 #endif
2476   return tso;
2477 }
2478
2479 #if defined(PAR)
2480 /* RFP:
2481    all parallel thread creation calls should fall through the following routine.
2482 */
2483 StgTSO *
2484 createThreadFromSpark(rtsSpark spark)
2485 { StgTSO *tso;
2486   ASSERT(spark != (rtsSpark)NULL);
2487 // JB: TAKE CARE OF THIS COUNTER! BUGGY
2488   if (advisory_thread_count >= RtsFlags.ParFlags.maxThreads)
2489   { threadsIgnored++;
2490     barf("{createSparkThread}Daq ghuH: refusing to create another thread; no more than %d threads allowed (currently %d)",
2491           RtsFlags.ParFlags.maxThreads, advisory_thread_count);
2492     return END_TSO_QUEUE;
2493   }
2494   else
2495   { threadsCreated++;
2496     tso = createThread(RtsFlags.GcFlags.initialStkSize);
2497     if (tso==END_TSO_QUEUE)
2498       barf("createSparkThread: Cannot create TSO");
2499 #if defined(DIST)
2500     tso->priority = AdvisoryPriority;
2501 #endif
2502     pushClosure(tso,spark);
2503     addToRunQueue(tso);
2504     advisory_thread_count++;  // JB: TAKE CARE OF THIS COUNTER! BUGGY
2505   }
2506   return tso;
2507 }
2508 #endif
2509
2510 /*
2511   Turn a spark into a thread.
2512   ToDo: fix for SMP (needs to acquire SCHED_MUTEX!)
2513 */
2514 #if 0
2515 StgTSO *
2516 activateSpark (rtsSpark spark)
2517 {
2518   StgTSO *tso;
2519
2520   tso = createSparkThread(spark);
2521   if (RtsFlags.ParFlags.ParStats.Full) {
2522     //ASSERT(run_queue_hd == END_TSO_QUEUE); // I think ...
2523       IF_PAR_DEBUG(verbose,
2524                    debugBelch("==^^ activateSpark: turning spark of closure %p (%s) into a thread\n",
2525                               (StgClosure *)spark, info_type((StgClosure *)spark)));
2526   }
2527   // ToDo: fwd info on local/global spark to thread -- HWL
2528   // tso->gran.exported =  spark->exported;
2529   // tso->gran.locked =   !spark->global;
2530   // tso->gran.sparkname = spark->name;
2531
2532   return tso;
2533 }
2534 #endif
2535
2536 /* ---------------------------------------------------------------------------
2537  * scheduleThread()
2538  *
2539  * scheduleThread puts a thread on the head of the runnable queue.
2540  * This will usually be done immediately after a thread is created.
2541  * The caller of scheduleThread must create the thread using e.g.
2542  * createThread and push an appropriate closure
2543  * on this thread's stack before the scheduler is invoked.
2544  * ------------------------------------------------------------------------ */
2545
2546 void
2547 scheduleThreadLocked(StgTSO *tso)
2548 {
2549   // The thread goes at the *end* of the run-queue, to avoid possible
2550   // starvation of any threads already on the queue.
2551   APPEND_TO_RUN_QUEUE(tso);
2552   threadRunnable();
2553 }
2554
2555 void
2556 scheduleThread(StgTSO* tso)
2557 {
2558   ACQUIRE_LOCK(&sched_mutex);
2559   scheduleThreadLocked(tso);
2560   RELEASE_LOCK(&sched_mutex);
2561 }
2562
2563 #if defined(RTS_SUPPORTS_THREADS)
2564 static Condition bound_cond_cache;
2565 static int bound_cond_cache_full = 0;
2566 #endif
2567
2568
2569 SchedulerStatus
2570 scheduleWaitThread(StgTSO* tso, /*[out]*/HaskellObj* ret,
2571                    Capability *initialCapability)
2572 {
2573     // Precondition: sched_mutex must be held
2574     StgMainThread *m;
2575
2576     m = stgMallocBytes(sizeof(StgMainThread), "waitThread");
2577     m->tso = tso;
2578     tso->main = m;
2579     m->ret = ret;
2580     m->stat = NoStatus;
2581     m->link = main_threads;
2582     m->prev = NULL;
2583     if (main_threads != NULL) {
2584         main_threads->prev = m;
2585     }
2586     main_threads = m;
2587
2588 #if defined(RTS_SUPPORTS_THREADS)
2589     // Allocating a new condition for each thread is expensive, so we
2590     // cache one.  This is a pretty feeble hack, but it helps speed up
2591     // consecutive call-ins quite a bit.
2592     if (bound_cond_cache_full) {
2593         m->bound_thread_cond = bound_cond_cache;
2594         bound_cond_cache_full = 0;
2595     } else {
2596         initCondition(&m->bound_thread_cond);
2597     }
2598 #endif
2599
2600     /* Put the thread on the main-threads list prior to scheduling the TSO.
2601        Failure to do so introduces a race condition in the MT case (as
2602        identified by Wolfgang Thaller), whereby the new task/OS thread
2603        created by scheduleThread_() would complete prior to the thread
2604        that spawned it managed to put 'itself' on the main-threads list.
2605        The upshot of it all being that the worker thread wouldn't get to
2606        signal the completion of the its work item for the main thread to
2607        see (==> it got stuck waiting.)    -- sof 6/02.
2608     */
2609     IF_DEBUG(scheduler, sched_belch("waiting for thread (%d)", tso->id));
2610
2611     APPEND_TO_RUN_QUEUE(tso);
2612     // NB. Don't call threadRunnable() here, because the thread is
2613     // bound and only runnable by *this* OS thread, so waking up other
2614     // workers will just slow things down.
2615
2616     return waitThread_(m, initialCapability);
2617 }
2618
2619 /* ---------------------------------------------------------------------------
2620  * initScheduler()
2621  *
2622  * Initialise the scheduler.  This resets all the queues - if the
2623  * queues contained any threads, they'll be garbage collected at the
2624  * next pass.
2625  *
2626  * ------------------------------------------------------------------------ */
2627
2628 void
2629 initScheduler(void)
2630 {
2631 #if defined(GRAN)
2632   nat i;
2633
2634   for (i=0; i<=MAX_PROC; i++) {
2635     run_queue_hds[i]      = END_TSO_QUEUE;
2636     run_queue_tls[i]      = END_TSO_QUEUE;
2637     blocked_queue_hds[i]  = END_TSO_QUEUE;
2638     blocked_queue_tls[i]  = END_TSO_QUEUE;
2639     ccalling_threadss[i]  = END_TSO_QUEUE;
2640     blackhole_queue[i]    = END_TSO_QUEUE;
2641     sleeping_queue        = END_TSO_QUEUE;
2642   }
2643 #else
2644   run_queue_hd      = END_TSO_QUEUE;
2645   run_queue_tl      = END_TSO_QUEUE;
2646   blocked_queue_hd  = END_TSO_QUEUE;
2647   blocked_queue_tl  = END_TSO_QUEUE;
2648   blackhole_queue   = END_TSO_QUEUE;
2649   sleeping_queue    = END_TSO_QUEUE;
2650 #endif
2651
2652   suspended_ccalling_threads  = END_TSO_QUEUE;
2653
2654   main_threads = NULL;
2655   all_threads  = END_TSO_QUEUE;
2656
2657   context_switch = 0;
2658   interrupted    = 0;
2659
2660   RtsFlags.ConcFlags.ctxtSwitchTicks =
2661       RtsFlags.ConcFlags.ctxtSwitchTime / TICK_MILLISECS;
2662
2663 #if defined(RTS_SUPPORTS_THREADS)
2664   /* Initialise the mutex and condition variables used by
2665    * the scheduler. */
2666   initMutex(&sched_mutex);
2667   initMutex(&term_mutex);
2668 #endif
2669
2670   ACQUIRE_LOCK(&sched_mutex);
2671
2672   /* A capability holds the state a native thread needs in
2673    * order to execute STG code. At least one capability is
2674    * floating around (only SMP builds have more than one).
2675    */
2676   initCapabilities();
2677
2678 #if defined(RTS_SUPPORTS_THREADS)
2679   initTaskManager();
2680 #endif
2681
2682 #if defined(SMP)
2683   /* eagerly start some extra workers */
2684   startingWorkerThread = RtsFlags.ParFlags.nNodes;
2685   startTasks(RtsFlags.ParFlags.nNodes, taskStart);
2686 #endif
2687
2688 #if /* defined(SMP) ||*/ defined(PARALLEL_HASKELL)
2689   initSparkPools();
2690 #endif
2691
2692   RELEASE_LOCK(&sched_mutex);
2693 }
2694
2695 void
2696 exitScheduler( void )
2697 {
2698     interrupted = rtsTrue;
2699     shutting_down_scheduler = rtsTrue;
2700
2701 #if defined(RTS_SUPPORTS_THREADS)
2702     if (threadIsTask(osThreadId())) { taskStop(); }
2703     stopTaskManager();
2704     //
2705     // What can we do here?  There are a bunch of worker threads, it
2706     // might be nice to let them exit cleanly.  There may be some main
2707     // threads in the run queue; we should let them return to their
2708     // callers with an Interrupted state.  We can't in general wait
2709     // for all the running Tasks to stop, because some might be off in
2710     // a C call that is blocked.
2711     //
2712     // Letting the run queue drain is the safest thing.  That lets any
2713     // main threads return that can return, and cleans up all the
2714     // runnable threads.  Then we grab all the Capabilities to stop
2715     // anything unexpected happening while we shut down.
2716     //
2717     // ToDo: this doesn't let us get the time stats from the worker
2718     // tasks, because they haven't called taskStop().
2719     //
2720     ACQUIRE_LOCK(&sched_mutex);
2721     {
2722         nat i;
2723         for (i = 1000; i > 0; i--) {
2724             if (EMPTY_RUN_QUEUE()) {
2725                 IF_DEBUG(scheduler, sched_belch("run queue is empty"));
2726                 break;
2727             }
2728             IF_DEBUG(scheduler, sched_belch("yielding"));
2729             RELEASE_LOCK(&sched_mutex);
2730             prodWorker();
2731             yieldThread();
2732             ACQUIRE_LOCK(&sched_mutex);
2733         }
2734     }
2735
2736 #ifdef SMP
2737     {
2738         Capability *cap;
2739         int n_capabilities = RtsFlags.ParFlags.nNodes;
2740         Capability *caps[n_capabilities];
2741         nat i;
2742
2743         while (n_capabilities > 0) {
2744             IF_DEBUG(scheduler, sched_belch("exitScheduler: grabbing all the capabilies (%d left)", n_capabilities));
2745             waitForReturnCapability(&sched_mutex, &cap);
2746             n_capabilities--;
2747             caps[n_capabilities] = cap;
2748         }
2749     }
2750 #else
2751     {
2752         Capability *cap;
2753         waitForReturnCapability(&sched_mutex, &cap);
2754     }
2755 #endif
2756 #endif
2757 }
2758
2759 /* ----------------------------------------------------------------------------
2760    Managing the per-task allocation areas.
2761
2762    Each capability comes with an allocation area.  These are
2763    fixed-length block lists into which allocation can be done.
2764
2765    ToDo: no support for two-space collection at the moment???
2766    ------------------------------------------------------------------------- */
2767
2768 static SchedulerStatus
2769 waitThread_(StgMainThread* m, Capability *initialCapability)
2770 {
2771   SchedulerStatus stat;
2772
2773   // Precondition: sched_mutex must be held.
2774   IF_DEBUG(scheduler, sched_belch("new main thread (%d)", m->tso->id));
2775
2776 #if defined(GRAN)
2777   /* GranSim specific init */
2778   CurrentTSO = m->tso;                // the TSO to run
2779   procStatus[MainProc] = Busy;        // status of main PE
2780   CurrentProc = MainProc;             // PE to run it on
2781   schedule(m,initialCapability);
2782 #else
2783   schedule(m,initialCapability);
2784   ASSERT(m->stat != NoStatus);
2785 #endif
2786
2787   stat = m->stat;
2788
2789 #if defined(RTS_SUPPORTS_THREADS)
2790   // Free the condition variable, returning it to the cache if possible.
2791   if (!bound_cond_cache_full) {
2792       bound_cond_cache = m->bound_thread_cond;
2793       bound_cond_cache_full = 1;
2794   } else {
2795       closeCondition(&m->bound_thread_cond);
2796   }
2797 #endif
2798
2799   IF_DEBUG(scheduler, sched_belch("main thread (%d) finished", m->tso->id));
2800   stgFree(m);
2801
2802   // Postcondition: sched_mutex still held
2803   return stat;
2804 }
2805
2806 /* ---------------------------------------------------------------------------
2807    Where are the roots that we know about?
2808
2809         - all the threads on the runnable queue
2810         - all the threads on the blocked queue
2811         - all the threads on the sleeping queue
2812         - all the thread currently executing a _ccall_GC
2813         - all the "main threads"
2814
2815    ------------------------------------------------------------------------ */
2816
2817 /* This has to be protected either by the scheduler monitor, or by the
2818         garbage collection monitor (probably the latter).
2819         KH @ 25/10/99
2820 */
2821
2822 void
2823 GetRoots( evac_fn evac )
2824 {
2825 #if defined(GRAN)
2826   {
2827     nat i;
2828     for (i=0; i<=RtsFlags.GranFlags.proc; i++) {
2829       if ((run_queue_hds[i] != END_TSO_QUEUE) && ((run_queue_hds[i] != NULL)))
2830           evac((StgClosure **)&run_queue_hds[i]);
2831       if ((run_queue_tls[i] != END_TSO_QUEUE) && ((run_queue_tls[i] != NULL)))
2832           evac((StgClosure **)&run_queue_tls[i]);
2833
2834       if ((blocked_queue_hds[i] != END_TSO_QUEUE) && ((blocked_queue_hds[i] != NULL)))
2835           evac((StgClosure **)&blocked_queue_hds[i]);
2836       if ((blocked_queue_tls[i] != END_TSO_QUEUE) && ((blocked_queue_tls[i] != NULL)))
2837           evac((StgClosure **)&blocked_queue_tls[i]);
2838       if ((ccalling_threadss[i] != END_TSO_QUEUE) && ((ccalling_threadss[i] != NULL)))
2839           evac((StgClosure **)&ccalling_threads[i]);
2840     }
2841   }
2842
2843   markEventQueue();
2844
2845 #else /* !GRAN */
2846   if (run_queue_hd != END_TSO_QUEUE) {
2847       ASSERT(run_queue_tl != END_TSO_QUEUE);
2848       evac((StgClosure **)&run_queue_hd);
2849       evac((StgClosure **)&run_queue_tl);
2850   }
2851
2852   if (blocked_queue_hd != END_TSO_QUEUE) {
2853       ASSERT(blocked_queue_tl != END_TSO_QUEUE);
2854       evac((StgClosure **)&blocked_queue_hd);
2855       evac((StgClosure **)&blocked_queue_tl);
2856   }
2857
2858   if (sleeping_queue != END_TSO_QUEUE) {
2859       evac((StgClosure **)&sleeping_queue);
2860   }
2861 #endif
2862
2863   if (blackhole_queue != END_TSO_QUEUE) {
2864       evac((StgClosure **)&blackhole_queue);
2865   }
2866
2867   if (suspended_ccalling_threads != END_TSO_QUEUE) {
2868       evac((StgClosure **)&suspended_ccalling_threads);
2869   }
2870
2871 #if defined(PARALLEL_HASKELL) || defined(GRAN)
2872   markSparkQueue(evac);
2873 #endif
2874
2875 #if defined(RTS_USER_SIGNALS)
2876   // mark the signal handlers (signals should be already blocked)
2877   markSignalHandlers(evac);
2878 #endif
2879 }
2880
2881 /* -----------------------------------------------------------------------------
2882    performGC
2883
2884    This is the interface to the garbage collector from Haskell land.
2885    We provide this so that external C code can allocate and garbage
2886    collect when called from Haskell via _ccall_GC.
2887
2888    It might be useful to provide an interface whereby the programmer
2889    can specify more roots (ToDo).
2890
2891    This needs to be protected by the GC condition variable above.  KH.
2892    -------------------------------------------------------------------------- */
2893
2894 static void (*extra_roots)(evac_fn);
2895
2896 void
2897 performGC(void)
2898 {
2899   /* Obligated to hold this lock upon entry */
2900   ACQUIRE_LOCK(&sched_mutex);
2901   GarbageCollect(GetRoots,rtsFalse);
2902   RELEASE_LOCK(&sched_mutex);
2903 }
2904
2905 void
2906 performMajorGC(void)
2907 {
2908   ACQUIRE_LOCK(&sched_mutex);
2909   GarbageCollect(GetRoots,rtsTrue);
2910   RELEASE_LOCK(&sched_mutex);
2911 }
2912
2913 static void
2914 AllRoots(evac_fn evac)
2915 {
2916     GetRoots(evac);             // the scheduler's roots
2917     extra_roots(evac);          // the user's roots
2918 }
2919
2920 void
2921 performGCWithRoots(void (*get_roots)(evac_fn))
2922 {
2923   ACQUIRE_LOCK(&sched_mutex);
2924   extra_roots = get_roots;
2925   GarbageCollect(AllRoots,rtsFalse);
2926   RELEASE_LOCK(&sched_mutex);
2927 }
2928
2929 /* -----------------------------------------------------------------------------
2930    Stack overflow
2931
2932    If the thread has reached its maximum stack size, then raise the
2933    StackOverflow exception in the offending thread.  Otherwise
2934    relocate the TSO into a larger chunk of memory and adjust its stack
2935    size appropriately.
2936    -------------------------------------------------------------------------- */
2937
2938 static StgTSO *
2939 threadStackOverflow(StgTSO *tso)
2940 {
2941   nat new_stack_size, stack_words;
2942   lnat new_tso_size;
2943   StgPtr new_sp;
2944   StgTSO *dest;
2945
2946   IF_DEBUG(sanity,checkTSO(tso));
2947   if (tso->stack_size >= tso->max_stack_size) {
2948
2949     IF_DEBUG(gc,
2950              debugBelch("@@ threadStackOverflow of TSO %ld (%p): stack too large (now %ld; max is %ld)\n",
2951                    (long)tso->id, tso, (long)tso->stack_size, (long)tso->max_stack_size);
2952              /* If we're debugging, just print out the top of the stack */
2953              printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size,
2954                                               tso->sp+64)));
2955
2956     /* Send this thread the StackOverflow exception */
2957     raiseAsync(tso, (StgClosure *)stackOverflow_closure);
2958     return tso;
2959   }
2960
2961   /* Try to double the current stack size.  If that takes us over the
2962    * maximum stack size for this thread, then use the maximum instead.
2963    * Finally round up so the TSO ends up as a whole number of blocks.
2964    */
2965   new_stack_size = stg_min(tso->stack_size * 2, tso->max_stack_size);
2966   new_tso_size   = (lnat)BLOCK_ROUND_UP(new_stack_size * sizeof(W_) +
2967                                        TSO_STRUCT_SIZE)/sizeof(W_);
2968   new_tso_size = round_to_mblocks(new_tso_size);  /* Be MBLOCK-friendly */
2969   new_stack_size = new_tso_size - TSO_STRUCT_SIZEW;
2970
2971   IF_DEBUG(scheduler, debugBelch("== sched: increasing stack size from %d words to %d.\n", tso->stack_size, new_stack_size));
2972
2973   dest = (StgTSO *)allocate(new_tso_size);
2974   TICK_ALLOC_TSO(new_stack_size,0);
2975
2976   /* copy the TSO block and the old stack into the new area */
2977   memcpy(dest,tso,TSO_STRUCT_SIZE);
2978   stack_words = tso->stack + tso->stack_size - tso->sp;
2979   new_sp = (P_)dest + new_tso_size - stack_words;
2980   memcpy(new_sp, tso->sp, stack_words * sizeof(W_));
2981
2982   /* relocate the stack pointers... */
2983   dest->sp         = new_sp;
2984   dest->stack_size = new_stack_size;
2985
2986   /* Mark the old TSO as relocated.  We have to check for relocated
2987    * TSOs in the garbage collector and any primops that deal with TSOs.
2988    *
2989    * It's important to set the sp value to just beyond the end
2990    * of the stack, so we don't attempt to scavenge any part of the
2991    * dead TSO's stack.
2992    */
2993   tso->what_next = ThreadRelocated;
2994   tso->link = dest;
2995   tso->sp = (P_)&(tso->stack[tso->stack_size]);
2996   tso->why_blocked = NotBlocked;
2997
2998   IF_PAR_DEBUG(verbose,
2999                debugBelch("@@ threadStackOverflow of TSO %d (now at %p): stack size increased to %ld\n",
3000                      tso->id, tso, tso->stack_size);
3001                /* If we're debugging, just print out the top of the stack */
3002                printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size,
3003                                                 tso->sp+64)));
3004
3005   IF_DEBUG(sanity,checkTSO(tso));
3006 #if 0
3007   IF_DEBUG(scheduler,printTSO(dest));
3008 #endif
3009
3010   return dest;
3011 }
3012
3013 /* ---------------------------------------------------------------------------
3014    Wake up a queue that was blocked on some resource.
3015    ------------------------------------------------------------------------ */
3016
3017 #if defined(GRAN)
3018 STATIC_INLINE void
3019 unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node )
3020 {
3021 }
3022 #elif defined(PARALLEL_HASKELL)
3023 STATIC_INLINE void
3024 unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node )
3025 {
3026   /* write RESUME events to log file and
3027      update blocked and fetch time (depending on type of the orig closure) */
3028   if (RtsFlags.ParFlags.ParStats.Full) {
3029     DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
3030                      GR_RESUMEQ, ((StgTSO *)bqe), ((StgTSO *)bqe)->block_info.closure,
3031                      0, 0 /* spark_queue_len(ADVISORY_POOL) */);
3032     if (EMPTY_RUN_QUEUE())
3033       emitSchedule = rtsTrue;
3034
3035     switch (get_itbl(node)->type) {
3036         case FETCH_ME_BQ:
3037           ((StgTSO *)bqe)->par.fetchtime += CURRENT_TIME-((StgTSO *)bqe)->par.blockedat;
3038           break;
3039         case RBH:
3040         case FETCH_ME:
3041         case BLACKHOLE_BQ:
3042           ((StgTSO *)bqe)->par.blocktime += CURRENT_TIME-((StgTSO *)bqe)->par.blockedat;
3043           break;
3044 #ifdef DIST
3045         case MVAR:
3046           break;
3047 #endif
3048         default:
3049           barf("{unblockOneLocked}Daq Qagh: unexpected closure in blocking queue");
3050         }
3051       }
3052 }
3053 #endif
3054
3055 #if defined(GRAN)
3056 StgBlockingQueueElement *
3057 unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node)
3058 {
3059     StgTSO *tso;
3060     PEs node_loc, tso_loc;
3061
3062     node_loc = where_is(node); // should be lifted out of loop
3063     tso = (StgTSO *)bqe;  // wastes an assignment to get the type right
3064     tso_loc = where_is((StgClosure *)tso);
3065     if (IS_LOCAL_TO(PROCS(node),tso_loc)) { // TSO is local
3066       /* !fake_fetch => TSO is on CurrentProc is same as IS_LOCAL_TO */
3067       ASSERT(CurrentProc!=node_loc || tso_loc==CurrentProc);
3068       CurrentTime[CurrentProc] += RtsFlags.GranFlags.Costs.lunblocktime;
3069       // insertThread(tso, node_loc);
3070       new_event(tso_loc, tso_loc, CurrentTime[CurrentProc],
3071                 ResumeThread,
3072                 tso, node, (rtsSpark*)NULL);
3073       tso->link = END_TSO_QUEUE; // overwrite link just to be sure
3074       // len_local++;
3075       // len++;
3076     } else { // TSO is remote (actually should be FMBQ)
3077       CurrentTime[CurrentProc] += RtsFlags.GranFlags.Costs.mpacktime +
3078                                   RtsFlags.GranFlags.Costs.gunblocktime +
3079                                   RtsFlags.GranFlags.Costs.latency;
3080       new_event(tso_loc, CurrentProc, CurrentTime[CurrentProc],
3081                 UnblockThread,
3082                 tso, node, (rtsSpark*)NULL);
3083       tso->link = END_TSO_QUEUE; // overwrite link just to be sure
3084       // len++;
3085     }
3086     /* the thread-queue-overhead is accounted for in either Resume or UnblockThread */
3087     IF_GRAN_DEBUG(bq,
3088                   debugBelch(" %s TSO %d (%p) [PE %d] (block_info.closure=%p) (next=%p) ,",
3089                           (node_loc==tso_loc ? "Local" : "Global"),
3090                           tso->id, tso, CurrentProc, tso->block_info.closure, tso->link));
3091     tso->block_info.closure = NULL;
3092     IF_DEBUG(scheduler,debugBelch("-- Waking up thread %ld (%p)\n",
3093                              tso->id, tso));
3094 }
3095 #elif defined(PARALLEL_HASKELL)
3096 StgBlockingQueueElement *
3097 unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node)
3098 {
3099     StgBlockingQueueElement *next;
3100
3101     switch (get_itbl(bqe)->type) {
3102     case TSO:
3103       ASSERT(((StgTSO *)bqe)->why_blocked != NotBlocked);
3104       /* if it's a TSO just push it onto the run_queue */
3105       next = bqe->link;
3106       ((StgTSO *)bqe)->link = END_TSO_QUEUE; // debugging?
3107       APPEND_TO_RUN_QUEUE((StgTSO *)bqe);
3108       threadRunnable();
3109       unblockCount(bqe, node);
3110       /* reset blocking status after dumping event */
3111       ((StgTSO *)bqe)->why_blocked = NotBlocked;
3112       break;
3113
3114     case BLOCKED_FETCH:
3115       /* if it's a BLOCKED_FETCH put it on the PendingFetches list */
3116       next = bqe->link;
3117       bqe->link = (StgBlockingQueueElement *)PendingFetches;
3118       PendingFetches = (StgBlockedFetch *)bqe;
3119       break;
3120
3121 # if defined(DEBUG)
3122       /* can ignore this case in a non-debugging setup;
3123          see comments on RBHSave closures above */
3124     case CONSTR:
3125       /* check that the closure is an RBHSave closure */
3126       ASSERT(get_itbl((StgClosure *)bqe) == &stg_RBH_Save_0_info ||
3127              get_itbl((StgClosure *)bqe) == &stg_RBH_Save_1_info ||
3128              get_itbl((StgClosure *)bqe) == &stg_RBH_Save_2_info);
3129       break;
3130
3131     default:
3132       barf("{unblockOneLocked}Daq Qagh: Unexpected IP (%#lx; %s) in blocking queue at %#lx\n",
3133            get_itbl((StgClosure *)bqe), info_type((StgClosure *)bqe),
3134            (StgClosure *)bqe);
3135 # endif
3136     }
3137   IF_PAR_DEBUG(bq, debugBelch(", %p (%s)\n", bqe, info_type((StgClosure*)bqe)));
3138   return next;
3139 }
3140
3141 #else /* !GRAN && !PARALLEL_HASKELL */
3142 StgTSO *
3143 unblockOneLocked(StgTSO *tso)
3144 {
3145   StgTSO *next;
3146
3147   ASSERT(get_itbl(tso)->type == TSO);
3148   ASSERT(tso->why_blocked != NotBlocked);
3149   tso->why_blocked = NotBlocked;
3150   next = tso->link;
3151   tso->link = END_TSO_QUEUE;
3152   APPEND_TO_RUN_QUEUE(tso);
3153   threadRunnable();
3154   IF_DEBUG(scheduler,sched_belch("waking up thread %ld", (long)tso->id));
3155   return next;
3156 }
3157 #endif
3158
3159 #if defined(GRAN) || defined(PARALLEL_HASKELL)
3160 INLINE_ME StgBlockingQueueElement *
3161 unblockOne(StgBlockingQueueElement *bqe, StgClosure *node)
3162 {
3163   ACQUIRE_LOCK(&sched_mutex);
3164   bqe = unblockOneLocked(bqe, node);
3165   RELEASE_LOCK(&sched_mutex);
3166   return bqe;
3167 }
3168 #else
3169 INLINE_ME StgTSO *
3170 unblockOne(StgTSO *tso)
3171 {
3172   ACQUIRE_LOCK(&sched_mutex);
3173   tso = unblockOneLocked(tso);
3174   RELEASE_LOCK(&sched_mutex);
3175   return tso;
3176 }
3177 #endif
3178
3179 #if defined(GRAN)
3180 void
3181 awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node)
3182 {
3183   StgBlockingQueueElement *bqe;
3184   PEs node_loc;
3185   nat len = 0;
3186
3187   IF_GRAN_DEBUG(bq,
3188                 debugBelch("##-_ AwBQ for node %p on PE %d @ %ld by TSO %d (%p): \n", \
3189                       node, CurrentProc, CurrentTime[CurrentProc],
3190                       CurrentTSO->id, CurrentTSO));
3191
3192   node_loc = where_is(node);
3193
3194   ASSERT(q == END_BQ_QUEUE ||
3195          get_itbl(q)->type == TSO ||   // q is either a TSO or an RBHSave
3196          get_itbl(q)->type == CONSTR); // closure (type constructor)
3197   ASSERT(is_unique(node));
3198
3199   /* FAKE FETCH: magically copy the node to the tso's proc;
3200      no Fetch necessary because in reality the node should not have been
3201      moved to the other PE in the first place
3202   */
3203   if (CurrentProc!=node_loc) {
3204     IF_GRAN_DEBUG(bq,
3205                   debugBelch("## node %p is on PE %d but CurrentProc is %d (TSO %d); assuming fake fetch and adjusting bitmask (old: %#x)\n",
3206                         node, node_loc, CurrentProc, CurrentTSO->id,
3207                         // CurrentTSO, where_is(CurrentTSO),
3208                         node->header.gran.procs));
3209     node->header.gran.procs = (node->header.gran.procs) | PE_NUMBER(CurrentProc);
3210     IF_GRAN_DEBUG(bq,
3211                   debugBelch("## new bitmask of node %p is %#x\n",
3212                         node, node->header.gran.procs));
3213     if (RtsFlags.GranFlags.GranSimStats.Global) {
3214       globalGranStats.tot_fake_fetches++;
3215     }
3216   }
3217
3218   bqe = q;
3219   // ToDo: check: ASSERT(CurrentProc==node_loc);
3220   while (get_itbl(bqe)->type==TSO) { // q != END_TSO_QUEUE) {
3221     //next = bqe->link;
3222     /*
3223        bqe points to the current element in the queue
3224        next points to the next element in the queue
3225     */
3226     //tso = (StgTSO *)bqe;  // wastes an assignment to get the type right
3227     //tso_loc = where_is(tso);
3228     len++;
3229     bqe = unblockOneLocked(bqe, node);
3230   }
3231
3232   /* if this is the BQ of an RBH, we have to put back the info ripped out of
3233      the closure to make room for the anchor of the BQ */
3234   if (bqe!=END_BQ_QUEUE) {
3235     ASSERT(get_itbl(node)->type == RBH && get_itbl(bqe)->type == CONSTR);
3236     /*
3237     ASSERT((info_ptr==&RBH_Save_0_info) ||
3238            (info_ptr==&RBH_Save_1_info) ||
3239            (info_ptr==&RBH_Save_2_info));
3240     */
3241     /* cf. convertToRBH in RBH.c for writing the RBHSave closure */
3242     ((StgRBH *)node)->blocking_queue = (StgBlockingQueueElement *)((StgRBHSave *)bqe)->payload[0];
3243     ((StgRBH *)node)->mut_link       = (StgMutClosure *)((StgRBHSave *)bqe)->payload[1];
3244
3245     IF_GRAN_DEBUG(bq,
3246                   debugBelch("## Filled in RBH_Save for %p (%s) at end of AwBQ\n",
3247                         node, info_type(node)));
3248   }
3249
3250   /* statistics gathering */
3251   if (RtsFlags.GranFlags.GranSimStats.Global) {
3252     // globalGranStats.tot_bq_processing_time += bq_processing_time;
3253     globalGranStats.tot_bq_len += len;      // total length of all bqs awakened
3254     // globalGranStats.tot_bq_len_local += len_local;  // same for local TSOs only
3255     globalGranStats.tot_awbq++;             // total no. of bqs awakened
3256   }
3257   IF_GRAN_DEBUG(bq,
3258                 debugBelch("## BQ Stats of %p: [%d entries] %s\n",
3259                         node, len, (bqe!=END_BQ_QUEUE) ? "RBH" : ""));
3260 }
3261 #elif defined(PARALLEL_HASKELL)
3262 void
3263 awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node)
3264 {
3265   StgBlockingQueueElement *bqe;
3266
3267   ACQUIRE_LOCK(&sched_mutex);
3268
3269   IF_PAR_DEBUG(verbose,
3270                debugBelch("##-_ AwBQ for node %p on [%x]: \n",
3271                      node, mytid));
3272 #ifdef DIST
3273   //RFP
3274   if(get_itbl(q)->type == CONSTR || q==END_BQ_QUEUE) {
3275     IF_PAR_DEBUG(verbose, debugBelch("## ... nothing to unblock so lets just return. RFP (BUG?)\n"));
3276     return;
3277   }
3278 #endif
3279
3280   ASSERT(q == END_BQ_QUEUE ||
3281          get_itbl(q)->type == TSO ||
3282          get_itbl(q)->type == BLOCKED_FETCH ||
3283          get_itbl(q)->type == CONSTR);
3284
3285   bqe = q;
3286   while (get_itbl(bqe)->type==TSO ||
3287          get_itbl(bqe)->type==BLOCKED_FETCH) {
3288     bqe = unblockOneLocked(bqe, node);
3289   }
3290   RELEASE_LOCK(&sched_mutex);
3291 }
3292
3293 #else   /* !GRAN && !PARALLEL_HASKELL */
3294
3295 void
3296 awakenBlockedQueueNoLock(StgTSO *tso)
3297 {
3298   if (tso == NULL) return; // hack; see bug #1235728, and comments in
3299                            // Exception.cmm
3300   while (tso != END_TSO_QUEUE) {
3301     tso = unblockOneLocked(tso);
3302   }
3303 }
3304
3305 void
3306 awakenBlockedQueue(StgTSO *tso)
3307 {
3308   if (tso == NULL) return; // hack; see bug #1235728, and comments in
3309                            // Exception.cmm
3310   ACQUIRE_LOCK(&sched_mutex);
3311   while (tso != END_TSO_QUEUE) {
3312     tso = unblockOneLocked(tso);
3313   }
3314   RELEASE_LOCK(&sched_mutex);
3315 }
3316 #endif
3317
3318 /* ---------------------------------------------------------------------------
3319    Interrupt execution
3320    - usually called inside a signal handler so it mustn't do anything fancy.
3321    ------------------------------------------------------------------------ */
3322
3323 void
3324 interruptStgRts(void)
3325 {
3326     interrupted    = 1;
3327     context_switch = 1;
3328     threadRunnable();
3329     /* ToDo: if invoked from a signal handler, this threadRunnable
3330      * only works if there's another thread (not this one) waiting to
3331      * be woken up.
3332      */
3333 }
3334
3335 /* -----------------------------------------------------------------------------
3336    Unblock a thread
3337
3338    This is for use when we raise an exception in another thread, which
3339    may be blocked.
3340    This has nothing to do with the UnblockThread event in GranSim. -- HWL
3341    -------------------------------------------------------------------------- */
3342
3343 #if defined(GRAN) || defined(PARALLEL_HASKELL)
3344 /*
3345   NB: only the type of the blocking queue is different in GranSim and GUM
3346       the operations on the queue-elements are the same
3347       long live polymorphism!
3348
3349   Locks: sched_mutex is held upon entry and exit.
3350
3351 */
3352 static void
3353 unblockThread(StgTSO *tso)
3354 {
3355   StgBlockingQueueElement *t, **last;
3356
3357   switch (tso->why_blocked) {
3358
3359   case NotBlocked:
3360     return;  /* not blocked */
3361
3362   case BlockedOnSTM:
3363     // Be careful: nothing to do here!  We tell the scheduler that the thread
3364     // is runnable and we leave it to the stack-walking code to abort the
3365     // transaction while unwinding the stack.  We should perhaps have a debugging
3366     // test to make sure that this really happens and that the 'zombie' transaction
3367     // does not get committed.
3368     goto done;
3369
3370   case BlockedOnMVar:
3371     ASSERT(get_itbl(tso->block_info.closure)->type == MVAR);
3372     {
3373       StgBlockingQueueElement *last_tso = END_BQ_QUEUE;
3374       StgMVar *mvar = (StgMVar *)(tso->block_info.closure);
3375
3376       last = (StgBlockingQueueElement **)&mvar->head;
3377       for (t = (StgBlockingQueueElement *)mvar->head;
3378            t != END_BQ_QUEUE;
3379            last = &t->link, last_tso = t, t = t->link) {
3380         if (t == (StgBlockingQueueElement *)tso) {
3381           *last = (StgBlockingQueueElement *)tso->link;
3382           if (mvar->tail == tso) {
3383             mvar->tail = (StgTSO *)last_tso;
3384           }
3385           goto done;
3386         }
3387       }
3388       barf("unblockThread (MVAR): TSO not found");
3389     }
3390
3391   case BlockedOnBlackHole:
3392     ASSERT(get_itbl(tso->block_info.closure)->type == BLACKHOLE_BQ);
3393     {
3394       StgBlockingQueue *bq = (StgBlockingQueue *)(tso->block_info.closure);
3395
3396       last = &bq->blocking_queue;
3397       for (t = bq->blocking_queue;
3398            t != END_BQ_QUEUE;
3399            last = &t->link, t = t->link) {
3400         if (t == (StgBlockingQueueElement *)tso) {
3401           *last = (StgBlockingQueueElement *)tso->link;
3402           goto done;
3403         }
3404       }
3405       barf("unblockThread (BLACKHOLE): TSO not found");
3406     }
3407
3408   case BlockedOnException:
3409     {
3410       StgTSO *target  = tso->block_info.tso;
3411
3412       ASSERT(get_itbl(target)->type == TSO);
3413
3414       if (target->what_next == ThreadRelocated) {
3415           target = target->link;
3416           ASSERT(get_itbl(target)->type == TSO);
3417       }
3418
3419       ASSERT(target->blocked_exceptions != NULL);
3420
3421       last = (StgBlockingQueueElement **)&target->blocked_exceptions;
3422       for (t = (StgBlockingQueueElement *)target->blocked_exceptions;
3423            t != END_BQ_QUEUE;
3424            last = &t->link, t = t->link) {
3425         ASSERT(get_itbl(t)->type == TSO);
3426         if (t == (StgBlockingQueueElement *)tso) {
3427           *last = (StgBlockingQueueElement *)tso->link;
3428           goto done;
3429         }
3430       }
3431       barf("unblockThread (Exception): TSO not found");
3432     }
3433
3434   case BlockedOnRead:
3435   case BlockedOnWrite:
3436 #if defined(mingw32_HOST_OS)
3437   case BlockedOnDoProc:
3438 #endif
3439     {
3440       /* take TSO off blocked_queue */
3441       StgBlockingQueueElement *prev = NULL;
3442       for (t = (StgBlockingQueueElement *)blocked_queue_hd; t != END_BQ_QUEUE;
3443            prev = t, t = t->link) {
3444         if (t == (StgBlockingQueueElement *)tso) {
3445           if (prev == NULL) {
3446             blocked_queue_hd = (StgTSO *)t->link;
3447             if ((StgBlockingQueueElement *)blocked_queue_tl == t) {
3448               blocked_queue_tl = END_TSO_QUEUE;
3449             }
3450           } else {
3451             prev->link = t->link;
3452             if ((StgBlockingQueueElement *)blocked_queue_tl == t) {
3453               blocked_queue_tl = (StgTSO *)prev;
3454             }
3455           }
3456 #if defined(mingw32_HOST_OS)
3457           /* (Cooperatively) signal that the worker thread should abort
3458            * the request.
3459            */
3460           abandonWorkRequest(tso->block_info.async_result->reqID);
3461 #endif
3462           goto done;
3463         }
3464       }
3465       barf("unblockThread (I/O): TSO not found");
3466     }
3467
3468   case BlockedOnDelay:
3469     {
3470       /* take TSO off sleeping_queue */
3471       StgBlockingQueueElement *prev = NULL;
3472       for (t = (StgBlockingQueueElement *)sleeping_queue; t != END_BQ_QUEUE;
3473            prev = t, t = t->link) {
3474         if (t == (StgBlockingQueueElement *)tso) {
3475           if (prev == NULL) {
3476             sleeping_queue = (StgTSO *)t->link;
3477           } else {
3478             prev->link = t->link;
3479           }
3480           goto done;
3481         }
3482       }
3483       barf("unblockThread (delay): TSO not found");
3484     }
3485
3486   default:
3487     barf("unblockThread");
3488   }
3489
3490  done:
3491   tso->link = END_TSO_QUEUE;
3492   tso->why_blocked = NotBlocked;
3493   tso->block_info.closure = NULL;
3494   PUSH_ON_RUN_QUEUE(tso);
3495 }
3496 #else
3497 static void
3498 unblockThread(StgTSO *tso)
3499 {
3500   StgTSO *t, **last;
3501
3502   /* To avoid locking unnecessarily. */
3503   if (tso->why_blocked == NotBlocked) {
3504     return;
3505   }
3506
3507   switch (tso->why_blocked) {
3508
3509   case BlockedOnSTM:
3510     // Be careful: nothing to do here!  We tell the scheduler that the thread
3511     // is runnable and we leave it to the stack-walking code to abort the
3512     // transaction while unwinding the stack.  We should perhaps have a debugging
3513     // test to make sure that this really happens and that the 'zombie' transaction
3514     // does not get committed.
3515     goto done;
3516
3517   case BlockedOnMVar:
3518     ASSERT(get_itbl(tso->block_info.closure)->type == MVAR);
3519     {
3520       StgTSO *last_tso = END_TSO_QUEUE;
3521       StgMVar *mvar = (StgMVar *)(tso->block_info.closure);
3522
3523       last = &mvar->head;
3524       for (t = mvar->head; t != END_TSO_QUEUE;
3525            last = &t->link, last_tso = t, t = t->link) {
3526         if (t == tso) {
3527           *last = tso->link;
3528           if (mvar->tail == tso) {
3529             mvar->tail = last_tso;
3530           }
3531           goto done;
3532         }
3533       }
3534       barf("unblockThread (MVAR): TSO not found");
3535     }
3536
3537   case BlockedOnBlackHole:
3538     {
3539       last = &blackhole_queue;
3540       for (t = blackhole_queue; t != END_TSO_QUEUE;
3541            last = &t->link, t = t->link) {
3542         if (t == tso) {
3543           *last = tso->link;
3544           goto done;
3545         }
3546       }
3547       barf("unblockThread (BLACKHOLE): TSO not found");
3548     }
3549
3550   case BlockedOnException:
3551     {
3552       StgTSO *target  = tso->block_info.tso;
3553
3554       ASSERT(get_itbl(target)->type == TSO);
3555
3556       while (target->what_next == ThreadRelocated) {
3557           target = target->link;
3558           ASSERT(get_itbl(target)->type == TSO);
3559       }
3560
3561       ASSERT(target->blocked_exceptions != NULL);
3562
3563       last = &target->blocked_exceptions;
3564       for (t = target->blocked_exceptions; t != END_TSO_QUEUE;
3565            last = &t->link, t = t->link) {
3566         ASSERT(get_itbl(t)->type == TSO);
3567         if (t == tso) {
3568           *last = tso->link;
3569           goto done;
3570         }
3571       }
3572       barf("unblockThread (Exception): TSO not found");
3573     }
3574
3575   case BlockedOnRead:
3576   case BlockedOnWrite:
3577 #if defined(mingw32_HOST_OS)
3578   case BlockedOnDoProc:
3579 #endif
3580     {
3581       StgTSO *prev = NULL;
3582       for (t = blocked_queue_hd; t != END_TSO_QUEUE;
3583            prev = t, t = t->link) {
3584         if (t == tso) {
3585           if (prev == NULL) {
3586             blocked_queue_hd = t->link;
3587             if (blocked_queue_tl == t) {
3588               blocked_queue_tl = END_TSO_QUEUE;
3589             }
3590           } else {
3591             prev->link = t->link;
3592             if (blocked_queue_tl == t) {
3593               blocked_queue_tl = prev;
3594             }
3595           }
3596 #if defined(mingw32_HOST_OS)
3597           /* (Cooperatively) signal that the worker thread should abort
3598            * the request.
3599            */
3600           abandonWorkRequest(tso->block_info.async_result->reqID);
3601 #endif
3602           goto done;
3603         }
3604       }
3605       barf("unblockThread (I/O): TSO not found");
3606     }
3607
3608   case BlockedOnDelay:
3609     {
3610       StgTSO *prev = NULL;
3611       for (t = sleeping_queue; t != END_TSO_QUEUE;
3612            prev = t, t = t->link) {
3613         if (t == tso) {
3614           if (prev == NULL) {
3615             sleeping_queue = t->link;
3616           } else {
3617             prev->link = t->link;
3618           }
3619           goto done;
3620         }
3621       }
3622       barf("unblockThread (delay): TSO not found");
3623     }
3624
3625   default:
3626     barf("unblockThread");
3627   }
3628
3629  done:
3630   tso->link = END_TSO_QUEUE;
3631   tso->why_blocked = NotBlocked;
3632   tso->block_info.closure = NULL;
3633   APPEND_TO_RUN_QUEUE(tso);
3634 }
3635 #endif
3636
3637 /* -----------------------------------------------------------------------------
3638  * checkBlackHoles()
3639  *
3640  * Check the blackhole_queue for threads that can be woken up.  We do
3641  * this periodically: before every GC, and whenever the run queue is
3642  * empty.
3643  *
3644  * An elegant solution might be to just wake up all the blocked
3645  * threads with awakenBlockedQueue occasionally: they'll go back to
3646  * sleep again if the object is still a BLACKHOLE.  Unfortunately this
3647  * doesn't give us a way to tell whether we've actually managed to
3648  * wake up any threads, so we would be busy-waiting.
3649  *
3650  * -------------------------------------------------------------------------- */
3651
3652 static rtsBool
3653 checkBlackHoles( void )
3654 {
3655     StgTSO **prev, *t;
3656     rtsBool any_woke_up = rtsFalse;
3657     StgHalfWord type;
3658
3659     IF_DEBUG(scheduler, sched_belch("checking threads blocked on black holes"));
3660
3661     // ASSUMES: sched_mutex
3662     prev = &blackhole_queue;
3663     t = blackhole_queue;
3664     while (t != END_TSO_QUEUE) {
3665         ASSERT(t->why_blocked == BlockedOnBlackHole);
3666         type = get_itbl(t->block_info.closure)->type;
3667         if (type != BLACKHOLE && type != CAF_BLACKHOLE) {
3668             IF_DEBUG(sanity,checkTSO(t));
3669             t = unblockOneLocked(t);
3670             *prev = t;
3671             any_woke_up = rtsTrue;
3672         } else {
3673             prev = &t->link;
3674             t = t->link;
3675         }
3676     }
3677
3678     return any_woke_up;
3679 }
3680
3681 /* -----------------------------------------------------------------------------
3682  * raiseAsync()
3683  *
3684  * The following function implements the magic for raising an
3685  * asynchronous exception in an existing thread.
3686  *
3687  * We first remove the thread from any queue on which it might be
3688  * blocked.  The possible blockages are MVARs and BLACKHOLE_BQs.
3689  *
3690  * We strip the stack down to the innermost CATCH_FRAME, building
3691  * thunks in the heap for all the active computations, so they can
3692  * be restarted if necessary.  When we reach a CATCH_FRAME, we build
3693  * an application of the handler to the exception, and push it on
3694  * the top of the stack.
3695  *
3696  * How exactly do we save all the active computations?  We create an
3697  * AP_STACK for every UpdateFrame on the stack.  Entering one of these
3698  * AP_STACKs pushes everything from the corresponding update frame
3699  * upwards onto the stack.  (Actually, it pushes everything up to the
3700  * next update frame plus a pointer to the next AP_STACK object.
3701  * Entering the next AP_STACK object pushes more onto the stack until we
3702  * reach the last AP_STACK object - at which point the stack should look
3703  * exactly as it did when we killed the TSO and we can continue
3704  * execution by entering the closure on top of the stack.
3705  *
3706  * We can also kill a thread entirely - this happens if either (a) the
3707  * exception passed to raiseAsync is NULL, or (b) there's no
3708  * CATCH_FRAME on the stack.  In either case, we strip the entire
3709  * stack and replace the thread with a zombie.
3710  *
3711  * Locks: sched_mutex held upon entry nor exit.
3712  *
3713  * -------------------------------------------------------------------------- */
3714
3715 void
3716 deleteThread(StgTSO *tso)
3717 {
3718   if (tso->why_blocked != BlockedOnCCall &&
3719       tso->why_blocked != BlockedOnCCall_NoUnblockExc) {
3720       raiseAsync(tso,NULL);
3721   }
3722 }
3723
3724 #ifdef FORKPROCESS_PRIMOP_SUPPORTED
3725 static void
3726 deleteThreadImmediately(StgTSO *tso)
3727 { // for forkProcess only:
3728   // delete thread without giving it a chance to catch the KillThread exception
3729
3730   if (tso->what_next == ThreadComplete || tso->what_next == ThreadKilled) {
3731       return;
3732   }
3733
3734   if (tso->why_blocked != BlockedOnCCall &&
3735       tso->why_blocked != BlockedOnCCall_NoUnblockExc) {
3736     unblockThread(tso);
3737   }
3738
3739   tso->what_next = ThreadKilled;
3740 }
3741 #endif
3742
3743 void
3744 raiseAsyncWithLock(StgTSO *tso, StgClosure *exception)
3745 {
3746   /* When raising async exs from contexts where sched_mutex isn't held;
3747      use raiseAsyncWithLock(). */
3748   ACQUIRE_LOCK(&sched_mutex);
3749   raiseAsync(tso,exception);
3750   RELEASE_LOCK(&sched_mutex);
3751 }
3752
3753 void
3754 raiseAsync(StgTSO *tso, StgClosure *exception)
3755 {
3756     raiseAsync_(tso, exception, rtsFalse);
3757 }
3758
3759 static void
3760 raiseAsync_(StgTSO *tso, StgClosure *exception, rtsBool stop_at_atomically)
3761 {
3762     StgRetInfoTable *info;
3763     StgPtr sp;
3764
3765     // Thread already dead?
3766     if (tso->what_next == ThreadComplete || tso->what_next == ThreadKilled) {
3767         return;
3768     }
3769
3770     IF_DEBUG(scheduler,
3771              sched_belch("raising exception in thread %ld.", (long)tso->id));
3772
3773     // Remove it from any blocking queues
3774     unblockThread(tso);
3775
3776     sp = tso->sp;
3777
3778     // The stack freezing code assumes there's a closure pointer on
3779     // the top of the stack, so we have to arrange that this is the case...
3780     //
3781     if (sp[0] == (W_)&stg_enter_info) {
3782         sp++;
3783     } else {
3784         sp--;
3785         sp[0] = (W_)&stg_dummy_ret_closure;
3786     }
3787
3788     while (1) {
3789         nat i;
3790
3791         // 1. Let the top of the stack be the "current closure"
3792         //
3793         // 2. Walk up the stack until we find either an UPDATE_FRAME or a
3794         // CATCH_FRAME.
3795         //
3796         // 3. If it's an UPDATE_FRAME, then make an AP_STACK containing the
3797         // current closure applied to the chunk of stack up to (but not
3798         // including) the update frame.  This closure becomes the "current
3799         // closure".  Go back to step 2.
3800         //
3801         // 4. If it's a CATCH_FRAME, then leave the exception handler on
3802         // top of the stack applied to the exception.
3803         //
3804         // 5. If it's a STOP_FRAME, then kill the thread.
3805         //
3806         // NB: if we pass an ATOMICALLY_FRAME then abort the associated
3807         // transaction
3808
3809
3810         StgPtr frame;
3811
3812         frame = sp + 1;
3813         info = get_ret_itbl((StgClosure *)frame);
3814
3815         while (info->i.type != UPDATE_FRAME
3816                && (info->i.type != CATCH_FRAME || exception == NULL)
3817                && info->i.type != STOP_FRAME
3818                && (info->i.type != ATOMICALLY_FRAME || stop_at_atomically == rtsFalse))
3819         {
3820             if (info->i.type == CATCH_RETRY_FRAME || info->i.type == ATOMICALLY_FRAME) {
3821               // IF we find an ATOMICALLY_FRAME then we abort the
3822               // current transaction and propagate the exception.  In
3823               // this case (unlike ordinary exceptions) we do not care
3824               // whether the transaction is valid or not because its
3825               // possible validity cannot have caused the exception
3826               // and will not be visible after the abort.
3827               IF_DEBUG(stm,
3828                        debugBelch("Found atomically block delivering async exception\n"));
3829               stmAbortTransaction(tso -> trec);
3830               tso -> trec = stmGetEnclosingTRec(tso -> trec);
3831             }
3832             frame += stack_frame_sizeW((StgClosure *)frame);
3833             info = get_ret_itbl((StgClosure *)frame);
3834         }
3835
3836         switch (info->i.type) {
3837
3838         case ATOMICALLY_FRAME:
3839             ASSERT(stop_at_atomically);
3840             ASSERT(stmGetEnclosingTRec(tso->trec) == NO_TREC);
3841             stmCondemnTransaction(tso -> trec);
3842 #ifdef REG_R1
3843             tso->sp = frame;
3844 #else
3845             // R1 is not a register: the return convention for IO in
3846             // this case puts the return value on the stack, so we
3847             // need to set up the stack to return to the atomically
3848             // frame properly...
3849             tso->sp = frame - 2;
3850             tso->sp[1] = (StgWord) &stg_NO_FINALIZER_closure; // why not?
3851             tso->sp[0] = (StgWord) &stg_ut_1_0_unreg_info;
3852 #endif
3853             tso->what_next = ThreadRunGHC;
3854             return;
3855
3856         case CATCH_FRAME:
3857             // If we find a CATCH_FRAME, and we've got an exception to raise,
3858             // then build the THUNK raise(exception), and leave it on
3859             // top of the CATCH_FRAME ready to enter.
3860             //
3861         {
3862 #ifdef PROFILING
3863             StgCatchFrame *cf = (StgCatchFrame *)frame;
3864 #endif
3865             StgThunk *raise;
3866
3867             // we've got an exception to raise, so let's pass it to the
3868             // handler in this frame.
3869             //
3870             raise = (StgThunk *)allocate(sizeofW(StgThunk)+MIN_UPD_SIZE);
3871             TICK_ALLOC_SE_THK(1,0);
3872             SET_HDR(raise,&stg_raise_info,cf->header.prof.ccs);
3873             raise->payload[0] = exception;
3874
3875             // throw away the stack from Sp up to the CATCH_FRAME.
3876             //
3877             sp = frame - 1;
3878
3879             /* Ensure that async excpetions are blocked now, so we don't get
3880              * a surprise exception before we get around to executing the
3881              * handler.
3882              */
3883             if (tso->blocked_exceptions == NULL) {
3884                 tso->blocked_exceptions = END_TSO_QUEUE;
3885             }
3886
3887             /* Put the newly-built THUNK on top of the stack, ready to execute
3888              * when the thread restarts.
3889              */
3890             sp[0] = (W_)raise;
3891             sp[-1] = (W_)&stg_enter_info;
3892             tso->sp = sp-1;
3893             tso->what_next = ThreadRunGHC;
3894             IF_DEBUG(sanity, checkTSO(tso));
3895             return;
3896         }
3897
3898         case UPDATE_FRAME:
3899         {
3900             StgAP_STACK * ap;
3901             nat words;
3902
3903             // First build an AP_STACK consisting of the stack chunk above the
3904             // current update frame, with the top word on the stack as the
3905             // fun field.
3906             //
3907             words = frame - sp - 1;
3908             ap = (StgAP_STACK *)allocate(AP_STACK_sizeW(words));
3909
3910             ap->size = words;
3911             ap->fun  = (StgClosure *)sp[0];
3912             sp++;
3913             for(i=0; i < (nat)words; ++i) {
3914                 ap->payload[i] = (StgClosure *)*sp++;
3915             }
3916
3917             SET_HDR(ap,&stg_AP_STACK_info,
3918                     ((StgClosure *)frame)->header.prof.ccs /* ToDo */);
3919             TICK_ALLOC_UP_THK(words+1,0);
3920
3921             IF_DEBUG(scheduler,
3922                      debugBelch("sched: Updating ");
3923                      printPtr((P_)((StgUpdateFrame *)frame)->updatee);
3924                      debugBelch(" with ");
3925                      printObj((StgClosure *)ap);
3926                 );
3927
3928             // Replace the updatee with an indirection - happily
3929             // this will also wake up any threads currently
3930             // waiting on the result.
3931             //
3932             // Warning: if we're in a loop, more than one update frame on
3933             // the stack may point to the same object.  Be careful not to
3934             // overwrite an IND_OLDGEN in this case, because we'll screw
3935             // up the mutable lists.  To be on the safe side, don't
3936             // overwrite any kind of indirection at all.  See also
3937             // threadSqueezeStack in GC.c, where we have to make a similar
3938             // check.
3939             //
3940             if (!closure_IND(((StgUpdateFrame *)frame)->updatee)) {
3941                 // revert the black hole
3942                 UPD_IND_NOLOCK(((StgUpdateFrame *)frame)->updatee,
3943                                (StgClosure *)ap);
3944             }
3945             sp += sizeofW(StgUpdateFrame) - 1;
3946             sp[0] = (W_)ap; // push onto stack
3947             break;
3948         }
3949
3950         case STOP_FRAME:
3951             // We've stripped the entire stack, the thread is now dead.
3952             sp += sizeofW(StgStopFrame);
3953             tso->what_next = ThreadKilled;
3954             tso->sp = sp;
3955             return;
3956
3957         default:
3958             barf("raiseAsync");
3959         }
3960     }
3961     barf("raiseAsync");
3962 }
3963
3964 /* -----------------------------------------------------------------------------
3965    raiseExceptionHelper
3966
3967    This function is called by the raise# primitve, just so that we can
3968    move some of the tricky bits of raising an exception from C-- into
3969    C.  Who knows, it might be a useful re-useable thing here too.
3970    -------------------------------------------------------------------------- */
3971
3972 StgWord
3973 raiseExceptionHelper (StgTSO *tso, StgClosure *exception)
3974 {
3975     StgThunk *raise_closure = NULL;
3976     StgPtr p, next;
3977     StgRetInfoTable *info;
3978     //
3979     // This closure represents the expression 'raise# E' where E
3980     // is the exception raise.  It is used to overwrite all the
3981     // thunks which are currently under evaluataion.
3982     //
3983
3984     //
3985     // LDV profiling: stg_raise_info has THUNK as its closure
3986     // type. Since a THUNK takes at least MIN_UPD_SIZE words in its
3987     // payload, MIN_UPD_SIZE is more approprate than 1.  It seems that
3988     // 1 does not cause any problem unless profiling is performed.
3989     // However, when LDV profiling goes on, we need to linearly scan
3990     // small object pool, where raise_closure is stored, so we should
3991     // use MIN_UPD_SIZE.
3992     //
3993     // raise_closure = (StgClosure *)RET_STGCALL1(P_,allocate,
3994     //                                 sizeofW(StgClosure)+1);
3995     //
3996
3997     //
3998     // Walk up the stack, looking for the catch frame.  On the way,
3999     // we update any closures pointed to from update frames with the
4000     // raise closure that we just built.
4001     //
4002     p = tso->sp;
4003     while(1) {
4004         info = get_ret_itbl((StgClosure *)p);
4005         next = p + stack_frame_sizeW((StgClosure *)p);
4006         switch (info->i.type) {
4007
4008         case UPDATE_FRAME:
4009             // Only create raise_closure if we need to.
4010             if (raise_closure == NULL) {
4011                 raise_closure =
4012                     (StgThunk *)allocate(sizeofW(StgThunk)+MIN_UPD_SIZE);
4013                 SET_HDR(raise_closure, &stg_raise_info, CCCS);
4014                 raise_closure->payload[0] = exception;
4015             }
4016             UPD_IND(((StgUpdateFrame *)p)->updatee,(StgClosure *)raise_closure);
4017             p = next;
4018             continue;
4019
4020         case ATOMICALLY_FRAME:
4021             IF_DEBUG(stm, debugBelch("Found ATOMICALLY_FRAME at %p\n", p));
4022             tso->sp = p;
4023             return ATOMICALLY_FRAME;
4024
4025         case CATCH_FRAME:
4026             tso->sp = p;
4027             return CATCH_FRAME;
4028
4029         case CATCH_STM_FRAME:
4030             IF_DEBUG(stm, debugBelch("Found CATCH_STM_FRAME at %p\n", p));
4031             tso->sp = p;
4032             return CATCH_STM_FRAME;
4033
4034         case STOP_FRAME:
4035             tso->sp = p;
4036             return STOP_FRAME;
4037
4038         case CATCH_RETRY_FRAME:
4039         default:
4040             p = next;
4041             continue;
4042         }
4043     }
4044 }
4045
4046
4047 /* -----------------------------------------------------------------------------
4048    findRetryFrameHelper
4049
4050    This function is called by the retry# primitive.  It traverses the stack
4051    leaving tso->sp referring to the frame which should handle the retry.
4052
4053    This should either be a CATCH_RETRY_FRAME (if the retry# is within an orElse#)
4054    or should be a ATOMICALLY_FRAME (if the retry# reaches the top level).
4055
4056    We skip CATCH_STM_FRAMEs because retries are not considered to be exceptions,
4057    despite the similar implementation.
4058
4059    We should not expect to see CATCH_FRAME or STOP_FRAME because those should
4060    not be created within memory transactions.
4061    -------------------------------------------------------------------------- */
4062
4063 StgWord
4064 findRetryFrameHelper (StgTSO *tso)
4065 {
4066   StgPtr           p, next;
4067   StgRetInfoTable *info;
4068
4069   p = tso -> sp;
4070   while (1) {
4071     info = get_ret_itbl((StgClosure *)p);
4072     next = p + stack_frame_sizeW((StgClosure *)p);
4073     switch (info->i.type) {
4074
4075     case ATOMICALLY_FRAME:
4076       IF_DEBUG(stm, debugBelch("Found ATOMICALLY_FRAME at %p during retrry\n", p));
4077       tso->sp = p;
4078       return ATOMICALLY_FRAME;
4079
4080     case CATCH_RETRY_FRAME:
4081       IF_DEBUG(stm, debugBelch("Found CATCH_RETRY_FRAME at %p during retrry\n", p));
4082       tso->sp = p;
4083       return CATCH_RETRY_FRAME;
4084
4085     case CATCH_STM_FRAME:
4086     default:
4087       ASSERT(info->i.type != CATCH_FRAME);
4088       ASSERT(info->i.type != STOP_FRAME);
4089       p = next;
4090       continue;
4091     }
4092   }
4093 }
4094
4095 /* -----------------------------------------------------------------------------
4096    resurrectThreads is called after garbage collection on the list of
4097    threads found to be garbage.  Each of these threads will be woken
4098    up and sent a signal: BlockedOnDeadMVar if the thread was blocked
4099    on an MVar, or NonTermination if the thread was blocked on a Black
4100    Hole.
4101
4102    Locks: sched_mutex isn't held upon entry nor exit.
4103    -------------------------------------------------------------------------- */
4104
4105 void
4106 resurrectThreads( StgTSO *threads )
4107 {
4108   StgTSO *tso, *next;
4109
4110   for (tso = threads; tso != END_TSO_QUEUE; tso = next) {
4111     next = tso->global_link;
4112     tso->global_link = all_threads;
4113     all_threads = tso;
4114     IF_DEBUG(scheduler, sched_belch("resurrecting thread %d", tso->id));
4115
4116     switch (tso->why_blocked) {
4117     case BlockedOnMVar:
4118     case BlockedOnException:
4119       /* Called by GC - sched_mutex lock is currently held. */
4120       raiseAsync(tso,(StgClosure *)BlockedOnDeadMVar_closure);
4121       break;
4122     case BlockedOnBlackHole:
4123       raiseAsync(tso,(StgClosure *)NonTermination_closure);
4124       break;
4125     case BlockedOnSTM:
4126       raiseAsync(tso,(StgClosure *)BlockedIndefinitely_closure);
4127       break;
4128     case NotBlocked:
4129       /* This might happen if the thread was blocked on a black hole
4130        * belonging to a thread that we've just woken up (raiseAsync
4131        * can wake up threads, remember...).
4132        */
4133       continue;
4134     default:
4135       barf("resurrectThreads: thread blocked in a strange way");
4136     }
4137   }
4138 }
4139
4140 /* ----------------------------------------------------------------------------
4141  * Debugging: why is a thread blocked
4142  * [Also provides useful information when debugging threaded programs
4143  *  at the Haskell source code level, so enable outside of DEBUG. --sof 7/02]
4144    ------------------------------------------------------------------------- */
4145
4146 static void
4147 printThreadBlockage(StgTSO *tso)
4148 {
4149   switch (tso->why_blocked) {
4150   case BlockedOnRead:
4151     debugBelch("is blocked on read from fd %d", (int)(tso->block_info.fd));
4152     break;
4153   case BlockedOnWrite:
4154     debugBelch("is blocked on write to fd %d", (int)(tso->block_info.fd));
4155     break;
4156 #if defined(mingw32_HOST_OS)
4157     case BlockedOnDoProc:
4158     debugBelch("is blocked on proc (request: %ld)", tso->block_info.async_result->reqID);
4159     break;
4160 #endif
4161   case BlockedOnDelay:
4162     debugBelch("is blocked until %ld", (long)(tso->block_info.target));
4163     break;
4164   case BlockedOnMVar:
4165     debugBelch("is blocked on an MVar @ %p", tso->block_info.closure);
4166     break;
4167   case BlockedOnException:
4168     debugBelch("is blocked on delivering an exception to thread %d",
4169             tso->block_info.tso->id);
4170     break;
4171   case BlockedOnBlackHole:
4172     debugBelch("is blocked on a black hole");
4173     break;
4174   case NotBlocked:
4175     debugBelch("is not blocked");
4176     break;
4177 #if defined(PARALLEL_HASKELL)
4178   case BlockedOnGA:
4179     debugBelch("is blocked on global address; local FM_BQ is %p (%s)",
4180             tso->block_info.closure, info_type(tso->block_info.closure));
4181     break;
4182   case BlockedOnGA_NoSend:
4183     debugBelch("is blocked on global address (no send); local FM_BQ is %p (%s)",
4184             tso->block_info.closure, info_type(tso->block_info.closure));
4185     break;
4186 #endif
4187   case BlockedOnCCall:
4188     debugBelch("is blocked on an external call");
4189     break;
4190   case BlockedOnCCall_NoUnblockExc:
4191     debugBelch("is blocked on an external call (exceptions were already blocked)");
4192     break;
4193   case BlockedOnSTM:
4194     debugBelch("is blocked on an STM operation");
4195     break;
4196   default:
4197     barf("printThreadBlockage: strange tso->why_blocked: %d for TSO %d (%d)",
4198          tso->why_blocked, tso->id, tso);
4199   }
4200 }
4201
4202 static void
4203 printThreadStatus(StgTSO *tso)
4204 {
4205   switch (tso->what_next) {
4206   case ThreadKilled:
4207     debugBelch("has been killed");
4208     break;
4209   case ThreadComplete:
4210     debugBelch("has completed");
4211     break;
4212   default:
4213     printThreadBlockage(tso);
4214   }
4215 }
4216
4217 void
4218 printAllThreads(void)
4219 {
4220   StgTSO *t;
4221
4222 # if defined(GRAN)
4223   char time_string[TIME_STR_LEN], node_str[NODE_STR_LEN];
4224   ullong_format_string(TIME_ON_PROC(CurrentProc),
4225                        time_string, rtsFalse/*no commas!*/);
4226
4227   debugBelch("all threads at [%s]:\n", time_string);
4228 # elif defined(PARALLEL_HASKELL)
4229   char time_string[TIME_STR_LEN], node_str[NODE_STR_LEN];
4230   ullong_format_string(CURRENT_TIME,
4231                        time_string, rtsFalse/*no commas!*/);
4232
4233   debugBelch("all threads at [%s]:\n", time_string);
4234 # else
4235   debugBelch("all threads:\n");
4236 # endif
4237
4238   for (t = all_threads; t != END_TSO_QUEUE; ) {
4239     debugBelch("\tthread %4d @ %p ", t->id, (void *)t);
4240 #if defined(DEBUG)
4241     {
4242       void *label = lookupThreadLabel(t->id);
4243       if (label) debugBelch("[\"%s\"] ",(char *)label);
4244     }
4245 #endif
4246     if (t->what_next == ThreadRelocated) {
4247         debugBelch("has been relocated...\n");
4248         t = t->link;
4249     } else {
4250         printThreadStatus(t);
4251         debugBelch("\n");
4252         t = t->global_link;
4253     }
4254   }
4255 }
4256
4257 #ifdef DEBUG
4258
4259 // useful from gdb
4260 void
4261 printThreadQueue(StgTSO *t)
4262 {
4263     nat i = 0;
4264     for (; t != END_TSO_QUEUE; t = t->link) {
4265         debugBelch("\tthread %d @ %p ", t->id, (void *)t);
4266         if (t->what_next == ThreadRelocated) {
4267             debugBelch("has been relocated...\n");
4268         } else {
4269             printThreadStatus(t);
4270             debugBelch("\n");
4271         }
4272         i++;
4273     }
4274     debugBelch("%d threads on queue\n", i);
4275 }
4276
4277 /*
4278    Print a whole blocking queue attached to node (debugging only).
4279 */
4280 # if defined(PARALLEL_HASKELL)
4281 void
4282 print_bq (StgClosure *node)
4283 {
4284   StgBlockingQueueElement *bqe;
4285   StgTSO *tso;
4286   rtsBool end;
4287
4288   debugBelch("## BQ of closure %p (%s): ",
4289           node, info_type(node));
4290
4291   /* should cover all closures that may have a blocking queue */
4292   ASSERT(get_itbl(node)->type == BLACKHOLE_BQ ||
4293          get_itbl(node)->type == FETCH_ME_BQ ||
4294          get_itbl(node)->type == RBH ||
4295          get_itbl(node)->type == MVAR);
4296
4297   ASSERT(node!=(StgClosure*)NULL);         // sanity check
4298
4299   print_bqe(((StgBlockingQueue*)node)->blocking_queue);
4300 }
4301
4302 /*
4303    Print a whole blocking queue starting with the element bqe.
4304 */
4305 void
4306 print_bqe (StgBlockingQueueElement *bqe)
4307 {
4308   rtsBool end;
4309
4310   /*
4311      NB: In a parallel setup a BQ of an RBH must end with an RBH_Save closure;
4312   */
4313   for (end = (bqe==END_BQ_QUEUE);
4314        !end; // iterate until bqe points to a CONSTR
4315        end = (get_itbl(bqe)->type == CONSTR) || (bqe->link==END_BQ_QUEUE),
4316        bqe = end ? END_BQ_QUEUE : bqe->link) {
4317     ASSERT(bqe != END_BQ_QUEUE);                               // sanity check
4318     ASSERT(bqe != (StgBlockingQueueElement *)NULL);            // sanity check
4319     /* types of closures that may appear in a blocking queue */
4320     ASSERT(get_itbl(bqe)->type == TSO ||
4321            get_itbl(bqe)->type == BLOCKED_FETCH ||
4322            get_itbl(bqe)->type == CONSTR);
4323     /* only BQs of an RBH end with an RBH_Save closure */
4324     //ASSERT(get_itbl(bqe)->type != CONSTR || get_itbl(node)->type == RBH);
4325
4326     switch (get_itbl(bqe)->type) {
4327     case TSO:
4328       debugBelch(" TSO %u (%x),",
4329               ((StgTSO *)bqe)->id, ((StgTSO *)bqe));
4330       break;
4331     case BLOCKED_FETCH:
4332       debugBelch(" BF (node=%p, ga=((%x, %d, %x)),",
4333               ((StgBlockedFetch *)bqe)->node,
4334               ((StgBlockedFetch *)bqe)->ga.payload.gc.gtid,
4335               ((StgBlockedFetch *)bqe)->ga.payload.gc.slot,
4336               ((StgBlockedFetch *)bqe)->ga.weight);
4337       break;
4338     case CONSTR:
4339       debugBelch(" %s (IP %p),",
4340               (get_itbl(bqe) == &stg_RBH_Save_0_info ? "RBH_Save_0" :
4341                get_itbl(bqe) == &stg_RBH_Save_1_info ? "RBH_Save_1" :
4342                get_itbl(bqe) == &stg_RBH_Save_2_info ? "RBH_Save_2" :
4343                "RBH_Save_?"), get_itbl(bqe));
4344       break;
4345     default:
4346       barf("Unexpected closure type %s in blocking queue", // of %p (%s)",
4347            info_type((StgClosure *)bqe)); // , node, info_type(node));
4348       break;
4349     }
4350   } /* for */
4351   debugBelch("\n");
4352 }
4353 # elif defined(GRAN)
4354 void
4355 print_bq (StgClosure *node)
4356 {
4357   StgBlockingQueueElement *bqe;
4358   PEs node_loc, tso_loc;
4359   rtsBool end;
4360
4361   /* should cover all closures that may have a blocking queue */
4362   ASSERT(get_itbl(node)->type == BLACKHOLE_BQ ||
4363          get_itbl(node)->type == FETCH_ME_BQ ||
4364          get_itbl(node)->type == RBH);
4365
4366   ASSERT(node!=(StgClosure*)NULL);         // sanity check
4367   node_loc = where_is(node);
4368
4369   debugBelch("## BQ of closure %p (%s) on [PE %d]: ",
4370           node, info_type(node), node_loc);
4371
4372   /*
4373      NB: In a parallel setup a BQ of an RBH must end with an RBH_Save closure;
4374   */
4375   for (bqe = ((StgBlockingQueue*)node)->blocking_queue, end = (bqe==END_BQ_QUEUE);
4376        !end; // iterate until bqe points to a CONSTR
4377        end = (get_itbl(bqe)->type == CONSTR) || (bqe->link==END_BQ_QUEUE), bqe = end ? END_BQ_QUEUE : bqe->link) {
4378     ASSERT(bqe != END_BQ_QUEUE);             // sanity check
4379     ASSERT(bqe != (StgBlockingQueueElement *)NULL);  // sanity check
4380     /* types of closures that may appear in a blocking queue */
4381     ASSERT(get_itbl(bqe)->type == TSO ||
4382            get_itbl(bqe)->type == CONSTR);
4383     /* only BQs of an RBH end with an RBH_Save closure */
4384     ASSERT(get_itbl(bqe)->type != CONSTR || get_itbl(node)->type == RBH);
4385
4386     tso_loc = where_is((StgClosure *)bqe);
4387     switch (get_itbl(bqe)->type) {
4388     case TSO:
4389       debugBelch(" TSO %d (%p) on [PE %d],",
4390               ((StgTSO *)bqe)->id, (StgTSO *)bqe, tso_loc);
4391       break;
4392     case CONSTR:
4393       debugBelch(" %s (IP %p),",
4394               (get_itbl(bqe) == &stg_RBH_Save_0_info ? "RBH_Save_0" :
4395                get_itbl(bqe) == &stg_RBH_Save_1_info ? "RBH_Save_1" :
4396                get_itbl(bqe) == &stg_RBH_Save_2_info ? "RBH_Save_2" :
4397                "RBH_Save_?"), get_itbl(bqe));
4398       break;
4399     default:
4400       barf("Unexpected closure type %s in blocking queue of %p (%s)",
4401            info_type((StgClosure *)bqe), node, info_type(node));
4402       break;
4403     }
4404   } /* for */
4405   debugBelch("\n");
4406 }
4407 # endif
4408
4409 #if defined(PARALLEL_HASKELL)
4410 static nat
4411 run_queue_len(void)
4412 {
4413   nat i;
4414   StgTSO *tso;
4415
4416   for (i=0, tso=run_queue_hd;
4417        tso != END_TSO_QUEUE;
4418        i++, tso=tso->link)
4419     /* nothing */
4420
4421   return i;
4422 }
4423 #endif
4424
4425 void
4426 sched_belch(char *s, ...)
4427 {
4428   va_list ap;
4429   va_start(ap,s);
4430 #ifdef RTS_SUPPORTS_THREADS
4431   debugBelch("sched (task %p): ", osThreadId());
4432 #elif defined(PARALLEL_HASKELL)
4433   debugBelch("== ");
4434 #else
4435   debugBelch("sched: ");
4436 #endif
4437   vdebugBelch(s, ap);
4438   debugBelch("\n");
4439   va_end(ap);
4440 }
4441
4442 #endif /* DEBUG */