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             bd = allocGroup( blocks );
1490             cap->r.rNursery->n_blocks += blocks;
1491
1492             // link the new group into the list
1493             bd->link = cap->r.rCurrentNursery;
1494             bd->u.back = cap->r.rCurrentNursery->u.back;
1495             if (cap->r.rCurrentNursery->u.back != NULL) {
1496                 cap->r.rCurrentNursery->u.back->link = bd;
1497             } else {
1498 #if !defined(SMP)
1499                 ASSERT(g0s0->blocks == cap->r.rCurrentNursery &&
1500                        g0s0 == cap->r.rNursery);
1501 #endif
1502                 cap->r.rNursery->blocks = bd;
1503             }
1504             cap->r.rCurrentNursery->u.back = bd;
1505
1506             // initialise it as a nursery block.  We initialise the
1507             // step, gen_no, and flags field of *every* sub-block in
1508             // this large block, because this is easier than making
1509             // sure that we always find the block head of a large
1510             // block whenever we call Bdescr() (eg. evacuate() and
1511             // isAlive() in the GC would both have to do this, at
1512             // least).
1513             {
1514                 bdescr *x;
1515                 for (x = bd; x < bd + blocks; x++) {
1516                     x->step = cap->r.rNursery;
1517                     x->gen_no = 0;
1518                     x->flags = 0;
1519                 }
1520             }
1521
1522             // This assert can be a killer if the app is doing lots
1523             // of large block allocations.
1524             IF_DEBUG(sanity, checkNurserySanity(cap->r.rNursery));
1525
1526             // now update the nursery to point to the new block
1527             cap->r.rCurrentNursery = bd;
1528
1529             // we might be unlucky and have another thread get on the
1530             // run queue before us and steal the large block, but in that
1531             // case the thread will just end up requesting another large
1532             // block.
1533             PUSH_ON_RUN_QUEUE(t);
1534             return rtsFalse;  /* not actually GC'ing */
1535         }
1536     }
1537
1538     IF_DEBUG(scheduler,
1539              debugBelch("--<< thread %ld (%s) stopped: HeapOverflow\n",
1540                         (long)t->id, whatNext_strs[t->what_next]));
1541 #if defined(GRAN)
1542     ASSERT(!is_on_queue(t,CurrentProc));
1543 #elif defined(PARALLEL_HASKELL)
1544     /* Currently we emit a DESCHEDULE event before GC in GUM.
1545        ToDo: either add separate event to distinguish SYSTEM time from rest
1546        or just nuke this DESCHEDULE (and the following SCHEDULE) */
1547     if (0 && RtsFlags.ParFlags.ParStats.Full) {
1548         DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
1549                          GR_DESCHEDULE, t, (StgClosure *)NULL, 0, 0);
1550         emitSchedule = rtsTrue;
1551     }
1552 #endif
1553
1554     PUSH_ON_RUN_QUEUE(t);
1555     return rtsTrue;
1556     /* actual GC is done at the end of the while loop in schedule() */
1557 }
1558
1559 /* -----------------------------------------------------------------------------
1560  * Handle a thread that returned to the scheduler with ThreadStackOverflow
1561  * ASSUMES: sched_mutex
1562  * -------------------------------------------------------------------------- */
1563
1564 static void
1565 scheduleHandleStackOverflow( StgTSO *t)
1566 {
1567     IF_DEBUG(scheduler,debugBelch("--<< thread %ld (%s) stopped, StackOverflow\n",
1568                                   (long)t->id, whatNext_strs[t->what_next]));
1569     /* just adjust the stack for this thread, then pop it back
1570      * on the run queue.
1571      */
1572     {
1573         /* enlarge the stack */
1574         StgTSO *new_t = threadStackOverflow(t);
1575
1576         /* This TSO has moved, so update any pointers to it from the
1577          * main thread stack.  It better not be on any other queues...
1578          * (it shouldn't be).
1579          */
1580         if (t->main != NULL) {
1581             t->main->tso = new_t;
1582         }
1583         PUSH_ON_RUN_QUEUE(new_t);
1584     }
1585 }
1586
1587 /* -----------------------------------------------------------------------------
1588  * Handle a thread that returned to the scheduler with ThreadYielding
1589  * ASSUMES: sched_mutex
1590  * -------------------------------------------------------------------------- */
1591
1592 static rtsBool
1593 scheduleHandleYield( StgTSO *t, nat prev_what_next )
1594 {
1595     // Reset the context switch flag.  We don't do this just before
1596     // running the thread, because that would mean we would lose ticks
1597     // during GC, which can lead to unfair scheduling (a thread hogs
1598     // the CPU because the tick always arrives during GC).  This way
1599     // penalises threads that do a lot of allocation, but that seems
1600     // better than the alternative.
1601     context_switch = 0;
1602
1603     /* put the thread back on the run queue.  Then, if we're ready to
1604      * GC, check whether this is the last task to stop.  If so, wake
1605      * up the GC thread.  getThread will block during a GC until the
1606      * GC is finished.
1607      */
1608     IF_DEBUG(scheduler,
1609              if (t->what_next != prev_what_next) {
1610                  debugBelch("--<< thread %ld (%s) stopped to switch evaluators\n",
1611                             (long)t->id, whatNext_strs[t->what_next]);
1612              } else {
1613                  debugBelch("--<< thread %ld (%s) stopped, yielding\n",
1614                             (long)t->id, whatNext_strs[t->what_next]);
1615              }
1616         );
1617
1618     IF_DEBUG(sanity,
1619              //debugBelch("&& Doing sanity check on yielding TSO %ld.", t->id);
1620              checkTSO(t));
1621     ASSERT(t->link == END_TSO_QUEUE);
1622
1623     // Shortcut if we're just switching evaluators: don't bother
1624     // doing stack squeezing (which can be expensive), just run the
1625     // thread.
1626     if (t->what_next != prev_what_next) {
1627         return rtsTrue;
1628     }
1629
1630 #if defined(GRAN)
1631     ASSERT(!is_on_queue(t,CurrentProc));
1632
1633     IF_DEBUG(sanity,
1634              //debugBelch("&& Doing sanity check on all ThreadQueues (and their TSOs).");
1635              checkThreadQsSanity(rtsTrue));
1636
1637 #endif
1638
1639     addToRunQueue(t);
1640
1641 #if defined(GRAN)
1642     /* add a ContinueThread event to actually process the thread */
1643     new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
1644               ContinueThread,
1645               t, (StgClosure*)NULL, (rtsSpark*)NULL);
1646     IF_GRAN_DEBUG(bq,
1647                   debugBelch("GRAN: eventq and runnableq after adding yielded thread to queue again:\n");
1648                   G_EVENTQ(0);
1649                   G_CURR_THREADQ(0));
1650 #endif
1651     return rtsFalse;
1652 }
1653
1654 /* -----------------------------------------------------------------------------
1655  * Handle a thread that returned to the scheduler with ThreadBlocked
1656  * ASSUMES: sched_mutex
1657  * -------------------------------------------------------------------------- */
1658
1659 static void
1660 scheduleHandleThreadBlocked( StgTSO *t
1661 #if !defined(GRAN) && !defined(DEBUG)
1662     STG_UNUSED
1663 #endif
1664     )
1665 {
1666 #if defined(GRAN)
1667     IF_DEBUG(scheduler,
1668              debugBelch("--<< thread %ld (%p; %s) stopped, blocking on node %p [PE %d] with BQ: \n",
1669                         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)));
1670              if (t->block_info.closure!=(StgClosure*)NULL) print_bq(t->block_info.closure));
1671
1672     // ??? needed; should emit block before
1673     IF_DEBUG(gran,
1674              DumpGranEvent(GR_DESCHEDULE, t));
1675     prune_eventq(t, (StgClosure *)NULL); // prune ContinueThreads for t
1676     /*
1677       ngoq Dogh!
1678       ASSERT(procStatus[CurrentProc]==Busy ||
1679       ((procStatus[CurrentProc]==Fetching) &&
1680       (t->block_info.closure!=(StgClosure*)NULL)));
1681       if (run_queue_hds[CurrentProc] == END_TSO_QUEUE &&
1682       !(!RtsFlags.GranFlags.DoAsyncFetch &&
1683       procStatus[CurrentProc]==Fetching))
1684       procStatus[CurrentProc] = Idle;
1685     */
1686 #elif defined(PAR)
1687     IF_DEBUG(scheduler,
1688              debugBelch("--<< thread %ld (%p; %s) stopped, blocking on node %p with BQ: \n",
1689                         t->id, t, whatNext_strs[t->what_next], t->block_info.closure));
1690     IF_PAR_DEBUG(bq,
1691
1692                  if (t->block_info.closure!=(StgClosure*)NULL)
1693                  print_bq(t->block_info.closure));
1694
1695     /* Send a fetch (if BlockedOnGA) and dump event to log file */
1696     blockThread(t);
1697
1698     /* whatever we schedule next, we must log that schedule */
1699     emitSchedule = rtsTrue;
1700
1701 #else /* !GRAN */
1702
1703       // We don't need to do anything.  The thread is blocked, and it
1704       // has tidied up its stack and placed itself on whatever queue
1705       // it needs to be on.
1706
1707 #if !defined(SMP)
1708     ASSERT(t->why_blocked != NotBlocked);
1709              // This might not be true under SMP: we don't have
1710              // exclusive access to this TSO, so someone might have
1711              // woken it up by now.  This actually happens: try
1712              // conc023 +RTS -N2.
1713 #endif
1714
1715     IF_DEBUG(scheduler,
1716              debugBelch("--<< thread %d (%s) stopped: ",
1717                         t->id, whatNext_strs[t->what_next]);
1718              printThreadBlockage(t);
1719              debugBelch("\n"));
1720
1721     /* Only for dumping event to log file
1722        ToDo: do I need this in GranSim, too?
1723        blockThread(t);
1724     */
1725 #endif
1726 }
1727
1728 /* -----------------------------------------------------------------------------
1729  * Handle a thread that returned to the scheduler with ThreadFinished
1730  * ASSUMES: sched_mutex
1731  * -------------------------------------------------------------------------- */
1732
1733 static rtsBool
1734 scheduleHandleThreadFinished( StgMainThread *mainThread
1735                               USED_WHEN_RTS_SUPPORTS_THREADS,
1736                               Capability *cap,
1737                               StgTSO *t )
1738 {
1739     /* Need to check whether this was a main thread, and if so,
1740      * return with the return value.
1741      *
1742      * We also end up here if the thread kills itself with an
1743      * uncaught exception, see Exception.cmm.
1744      */
1745     IF_DEBUG(scheduler,debugBelch("--++ thread %d (%s) finished\n",
1746                                   t->id, whatNext_strs[t->what_next]));
1747
1748 #if defined(GRAN)
1749       endThread(t, CurrentProc); // clean-up the thread
1750 #elif defined(PARALLEL_HASKELL)
1751       /* For now all are advisory -- HWL */
1752       //if(t->priority==AdvisoryPriority) ??
1753       advisory_thread_count--; // JB: Caution with this counter, buggy!
1754
1755 # if defined(DIST)
1756       if(t->dist.priority==RevalPriority)
1757         FinishReval(t);
1758 # endif
1759
1760 # if defined(EDENOLD)
1761       // the thread could still have an outport... (BUG)
1762       if (t->eden.outport != -1) {
1763       // delete the outport for the tso which has finished...
1764         IF_PAR_DEBUG(eden_ports,
1765                    debugBelch("WARNING: Scheduler removes outport %d for TSO %d.\n",
1766                               t->eden.outport, t->id));
1767         deleteOPT(t);
1768       }
1769       // thread still in the process (HEAVY BUG! since outport has just been closed...)
1770       if (t->eden.epid != -1) {
1771         IF_PAR_DEBUG(eden_ports,
1772                    debugBelch("WARNING: Scheduler removes TSO %d from process %d .\n",
1773                            t->id, t->eden.epid));
1774         removeTSOfromProcess(t);
1775       }
1776 # endif
1777
1778 # if defined(PAR)
1779       if (RtsFlags.ParFlags.ParStats.Full &&
1780           !RtsFlags.ParFlags.ParStats.Suppressed)
1781         DumpEndEvent(CURRENT_PROC, t, rtsFalse /* not mandatory */);
1782
1783       //  t->par only contains statistics: left out for now...
1784       IF_PAR_DEBUG(fish,
1785                    debugBelch("**** end thread: ended sparked thread %d (%lx); sparkname: %lx\n",
1786                               t->id,t,t->par.sparkname));
1787 # endif
1788 #endif // PARALLEL_HASKELL
1789
1790       //
1791       // Check whether the thread that just completed was a main
1792       // thread, and if so return with the result.
1793       //
1794       // There is an assumption here that all thread completion goes
1795       // through this point; we need to make sure that if a thread
1796       // ends up in the ThreadKilled state, that it stays on the run
1797       // queue so it can be dealt with here.
1798       //
1799       if (
1800 #if defined(RTS_SUPPORTS_THREADS)
1801           mainThread != NULL
1802 #else
1803           mainThread->tso == t
1804 #endif
1805           )
1806       {
1807           // We are a bound thread: this must be our thread that just
1808           // completed.
1809           ASSERT(mainThread->tso == t);
1810
1811           if (t->what_next == ThreadComplete) {
1812               if (mainThread->ret) {
1813                   // NOTE: return val is tso->sp[1] (see StgStartup.hc)
1814                   *(mainThread->ret) = (StgClosure *)mainThread->tso->sp[1];
1815               }
1816               mainThread->stat = Success;
1817           } else {
1818               if (mainThread->ret) {
1819                   *(mainThread->ret) = NULL;
1820               }
1821               if (interrupted) {
1822                   mainThread->stat = Interrupted;
1823               } else {
1824                   mainThread->stat = Killed;
1825               }
1826           }
1827 #ifdef DEBUG
1828           removeThreadLabel((StgWord)mainThread->tso->id);
1829 #endif
1830           if (mainThread->prev == NULL) {
1831               ASSERT(mainThread == main_threads);
1832               main_threads = mainThread->link;
1833           } else {
1834               mainThread->prev->link = mainThread->link;
1835           }
1836           if (mainThread->link != NULL) {
1837               mainThread->link->prev = mainThread->prev;
1838           }
1839           releaseCapability(cap);
1840           return rtsTrue; // tells schedule() to return
1841       }
1842
1843 #ifdef RTS_SUPPORTS_THREADS
1844       ASSERT(t->main == NULL);
1845 #else
1846       if (t->main != NULL) {
1847           // Must be a main thread that is not the topmost one.  Leave
1848           // it on the run queue until the stack has unwound to the
1849           // point where we can deal with this.  Leaving it on the run
1850           // queue also ensures that the garbage collector knows about
1851           // this thread and its return value (it gets dropped from the
1852           // all_threads list so there's no other way to find it).
1853           APPEND_TO_RUN_QUEUE(t);
1854       }
1855 #endif
1856       return rtsFalse;
1857 }
1858
1859 /* -----------------------------------------------------------------------------
1860  * Perform a heap census, if PROFILING
1861  * -------------------------------------------------------------------------- */
1862
1863 static rtsBool
1864 scheduleDoHeapProfile( rtsBool ready_to_gc STG_UNUSED )
1865 {
1866 #if defined(PROFILING)
1867     // When we have +RTS -i0 and we're heap profiling, do a census at
1868     // every GC.  This lets us get repeatable runs for debugging.
1869     if (performHeapProfile ||
1870         (RtsFlags.ProfFlags.profileInterval==0 &&
1871          RtsFlags.ProfFlags.doHeapProfile && ready_to_gc)) {
1872         GarbageCollect(GetRoots, rtsTrue);
1873         heapCensus();
1874         performHeapProfile = rtsFalse;
1875         return rtsTrue;  // true <=> we already GC'd
1876     }
1877 #endif
1878     return rtsFalse;
1879 }
1880
1881 /* -----------------------------------------------------------------------------
1882  * Perform a garbage collection if necessary
1883  * ASSUMES: sched_mutex
1884  * -------------------------------------------------------------------------- */
1885
1886 static void
1887 scheduleDoGC( rtsBool force_major )
1888 {
1889     StgTSO *t;
1890 #ifdef SMP
1891     Capability *cap;
1892     static rtsBool waiting_for_gc;
1893     int n_capabilities = RtsFlags.ParFlags.nNodes - 1;
1894            // subtract one because we're already holding one.
1895     Capability *caps[n_capabilities];
1896 #endif
1897
1898 #ifdef SMP
1899     // In order to GC, there must be no threads running Haskell code.
1900     // Therefore, the GC thread needs to hold *all* the capabilities,
1901     // and release them after the GC has completed.
1902     //
1903     // This seems to be the simplest way: previous attempts involved
1904     // making all the threads with capabilities give up their
1905     // capabilities and sleep except for the *last* one, which
1906     // actually did the GC.  But it's quite hard to arrange for all
1907     // the other tasks to sleep and stay asleep.
1908     //
1909     // This does mean that there will be multiple entries in the
1910     // thread->capability hash table for the current thread, but
1911     // they will be removed as normal when the capabilities are
1912     // released again.
1913     //
1914
1915     // Someone else is already trying to GC
1916     if (waiting_for_gc) return;
1917     waiting_for_gc = rtsTrue;
1918
1919     while (n_capabilities > 0) {
1920         IF_DEBUG(scheduler, sched_belch("ready_to_gc, grabbing all the capabilies (%d left)", n_capabilities));
1921         waitForReturnCapability(&sched_mutex, &cap);
1922         n_capabilities--;
1923         caps[n_capabilities] = cap;
1924     }
1925
1926     waiting_for_gc = rtsFalse;
1927 #endif
1928
1929     /* Kick any transactions which are invalid back to their
1930      * atomically frames.  When next scheduled they will try to
1931      * commit, this commit will fail and they will retry.
1932      */
1933     {
1934         StgTSO *next;
1935
1936         for (t = all_threads; t != END_TSO_QUEUE; t = next) {
1937             if (t->what_next == ThreadRelocated) {
1938                 next = t->link;
1939             } else {
1940                 next = t->global_link;
1941                 if (t -> trec != NO_TREC && t -> why_blocked == NotBlocked) {
1942                     if (!stmValidateNestOfTransactions (t -> trec)) {
1943                         IF_DEBUG(stm, sched_belch("trec %p found wasting its time", t));
1944
1945                         // strip the stack back to the ATOMICALLY_FRAME, aborting
1946                         // the (nested) transaction, and saving the stack of any
1947                         // partially-evaluated thunks on the heap.
1948                         raiseAsync_(t, NULL, rtsTrue);
1949
1950 #ifdef REG_R1
1951                         ASSERT(get_itbl((StgClosure *)t->sp)->type == ATOMICALLY_FRAME);
1952 #endif
1953                     }
1954                 }
1955             }
1956         }
1957     }
1958
1959     // so this happens periodically:
1960     scheduleCheckBlackHoles();
1961
1962     IF_DEBUG(scheduler, printAllThreads());
1963
1964     /* everybody back, start the GC.
1965      * Could do it in this thread, or signal a condition var
1966      * to do it in another thread.  Either way, we need to
1967      * broadcast on gc_pending_cond afterward.
1968      */
1969 #if defined(RTS_SUPPORTS_THREADS)
1970     IF_DEBUG(scheduler,sched_belch("doing GC"));
1971 #endif
1972     GarbageCollect(GetRoots, force_major);
1973
1974 #if defined(SMP)
1975     {
1976         // release our stash of capabilities.
1977         nat i;
1978         for (i = 0; i < RtsFlags.ParFlags.nNodes-1; i++) {
1979             releaseCapability(caps[i]);
1980         }
1981     }
1982 #endif
1983
1984 #if defined(GRAN)
1985     /* add a ContinueThread event to continue execution of current thread */
1986     new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
1987               ContinueThread,
1988               t, (StgClosure*)NULL, (rtsSpark*)NULL);
1989     IF_GRAN_DEBUG(bq,
1990                   debugBelch("GRAN: eventq and runnableq after Garbage collection:\n\n");
1991                   G_EVENTQ(0);
1992                   G_CURR_THREADQ(0));
1993 #endif /* GRAN */
1994 }
1995
1996 /* ---------------------------------------------------------------------------
1997  * rtsSupportsBoundThreads(): is the RTS built to support bound threads?
1998  * used by Control.Concurrent for error checking.
1999  * ------------------------------------------------------------------------- */
2000
2001 StgBool
2002 rtsSupportsBoundThreads(void)
2003 {
2004 #if defined(RTS_SUPPORTS_THREADS)
2005   return rtsTrue;
2006 #else
2007   return rtsFalse;
2008 #endif
2009 }
2010
2011 /* ---------------------------------------------------------------------------
2012  * isThreadBound(tso): check whether tso is bound to an OS thread.
2013  * ------------------------------------------------------------------------- */
2014
2015 StgBool
2016 isThreadBound(StgTSO* tso USED_IN_THREADED_RTS)
2017 {
2018 #if defined(RTS_SUPPORTS_THREADS)
2019   return (tso->main != NULL);
2020 #endif
2021   return rtsFalse;
2022 }
2023
2024 /* ---------------------------------------------------------------------------
2025  * Singleton fork(). Do not copy any running threads.
2026  * ------------------------------------------------------------------------- */
2027
2028 #ifndef mingw32_HOST_OS
2029 #define FORKPROCESS_PRIMOP_SUPPORTED
2030 #endif
2031
2032 #ifdef FORKPROCESS_PRIMOP_SUPPORTED
2033 static void
2034 deleteThreadImmediately(StgTSO *tso);
2035 #endif
2036 StgInt
2037 forkProcess(HsStablePtr *entry
2038 #ifndef FORKPROCESS_PRIMOP_SUPPORTED
2039             STG_UNUSED
2040 #endif
2041            )
2042 {
2043 #ifdef FORKPROCESS_PRIMOP_SUPPORTED
2044   pid_t pid;
2045   StgTSO* t,*next;
2046   StgMainThread *m;
2047   SchedulerStatus rc;
2048
2049   IF_DEBUG(scheduler,sched_belch("forking!"));
2050   rts_lock(); // This not only acquires sched_mutex, it also
2051               // makes sure that no other threads are running
2052
2053   pid = fork();
2054
2055   if (pid) { /* parent */
2056
2057   /* just return the pid */
2058     rts_unlock();
2059     return pid;
2060
2061   } else { /* child */
2062
2063
2064       // delete all threads
2065     run_queue_hd = run_queue_tl = END_TSO_QUEUE;
2066
2067     for (t = all_threads; t != END_TSO_QUEUE; t = next) {
2068       next = t->link;
2069
2070         // don't allow threads to catch the ThreadKilled exception
2071       deleteThreadImmediately(t);
2072     }
2073
2074       // wipe the main thread list
2075     while((m = main_threads) != NULL) {
2076       main_threads = m->link;
2077 # ifdef THREADED_RTS
2078       closeCondition(&m->bound_thread_cond);
2079 # endif
2080       stgFree(m);
2081     }
2082
2083     rc = rts_evalStableIO(entry, NULL);  // run the action
2084     rts_checkSchedStatus("forkProcess",rc);
2085
2086     rts_unlock();
2087
2088     hs_exit();                      // clean up and exit
2089     stg_exit(0);
2090   }
2091 #else /* !FORKPROCESS_PRIMOP_SUPPORTED */
2092   barf("forkProcess#: primop not supported, sorry!\n");
2093   return -1;
2094 #endif
2095 }
2096
2097 /* ---------------------------------------------------------------------------
2098  * deleteAllThreads():  kill all the live threads.
2099  *
2100  * This is used when we catch a user interrupt (^C), before performing
2101  * any necessary cleanups and running finalizers.
2102  *
2103  * Locks: sched_mutex held.
2104  * ------------------------------------------------------------------------- */
2105
2106 void
2107 deleteAllThreads ( void )
2108 {
2109   StgTSO* t, *next;
2110   IF_DEBUG(scheduler,sched_belch("deleting all threads"));
2111   for (t = all_threads; t != END_TSO_QUEUE; t = next) {
2112       if (t->what_next == ThreadRelocated) {
2113           next = t->link;
2114       } else {
2115           next = t->global_link;
2116           deleteThread(t);
2117       }
2118   }
2119
2120   // The run queue now contains a bunch of ThreadKilled threads.  We
2121   // must not throw these away: the main thread(s) will be in there
2122   // somewhere, and the main scheduler loop has to deal with it.
2123   // Also, the run queue is the only thing keeping these threads from
2124   // being GC'd, and we don't want the "main thread has been GC'd" panic.
2125
2126   ASSERT(blocked_queue_hd == END_TSO_QUEUE);
2127   ASSERT(blackhole_queue == END_TSO_QUEUE);
2128   ASSERT(sleeping_queue == END_TSO_QUEUE);
2129 }
2130
2131 /* startThread and  insertThread are now in GranSim.c -- HWL */
2132
2133
2134 /* ---------------------------------------------------------------------------
2135  * Suspending & resuming Haskell threads.
2136  *
2137  * When making a "safe" call to C (aka _ccall_GC), the task gives back
2138  * its capability before calling the C function.  This allows another
2139  * task to pick up the capability and carry on running Haskell
2140  * threads.  It also means that if the C call blocks, it won't lock
2141  * the whole system.
2142  *
2143  * The Haskell thread making the C call is put to sleep for the
2144  * duration of the call, on the susepended_ccalling_threads queue.  We
2145  * give out a token to the task, which it can use to resume the thread
2146  * on return from the C function.
2147  * ------------------------------------------------------------------------- */
2148
2149 StgInt
2150 suspendThread( StgRegTable *reg )
2151 {
2152   nat tok;
2153   Capability *cap;
2154   int saved_errno = errno;
2155
2156   /* assume that *reg is a pointer to the StgRegTable part
2157    * of a Capability.
2158    */
2159   cap = (Capability *)((void *)((unsigned char*)reg - sizeof(StgFunTable)));
2160
2161   ACQUIRE_LOCK(&sched_mutex);
2162
2163   IF_DEBUG(scheduler,
2164            sched_belch("thread %d did a _ccall_gc", cap->r.rCurrentTSO->id));
2165
2166   // XXX this might not be necessary --SDM
2167   cap->r.rCurrentTSO->what_next = ThreadRunGHC;
2168
2169   threadPaused(cap->r.rCurrentTSO);
2170   cap->r.rCurrentTSO->link = suspended_ccalling_threads;
2171   suspended_ccalling_threads = cap->r.rCurrentTSO;
2172
2173   if(cap->r.rCurrentTSO->blocked_exceptions == NULL)  {
2174       cap->r.rCurrentTSO->why_blocked = BlockedOnCCall;
2175       cap->r.rCurrentTSO->blocked_exceptions = END_TSO_QUEUE;
2176   } else {
2177       cap->r.rCurrentTSO->why_blocked = BlockedOnCCall_NoUnblockExc;
2178   }
2179
2180   /* Use the thread ID as the token; it should be unique */
2181   tok = cap->r.rCurrentTSO->id;
2182
2183   /* Hand back capability */
2184   cap->r.rInHaskell = rtsFalse;
2185   releaseCapability(cap);
2186
2187 #if defined(RTS_SUPPORTS_THREADS)
2188   /* Preparing to leave the RTS, so ensure there's a native thread/task
2189      waiting to take over.
2190   */
2191   IF_DEBUG(scheduler, sched_belch("worker (token %d): leaving RTS", tok));
2192 #endif
2193
2194   RELEASE_LOCK(&sched_mutex);
2195
2196   errno = saved_errno;
2197   return tok;
2198 }
2199
2200 StgRegTable *
2201 resumeThread( StgInt tok )
2202 {
2203   StgTSO *tso, **prev;
2204   Capability *cap;
2205   int saved_errno = errno;
2206
2207 #if defined(RTS_SUPPORTS_THREADS)
2208   /* Wait for permission to re-enter the RTS with the result. */
2209   ACQUIRE_LOCK(&sched_mutex);
2210   waitForReturnCapability(&sched_mutex, &cap);
2211
2212   IF_DEBUG(scheduler, sched_belch("worker (token %d): re-entering RTS", tok));
2213 #else
2214   grabCapability(&cap);
2215 #endif
2216
2217   /* Remove the thread off of the suspended list */
2218   prev = &suspended_ccalling_threads;
2219   for (tso = suspended_ccalling_threads;
2220        tso != END_TSO_QUEUE;
2221        prev = &tso->link, tso = tso->link) {
2222     if (tso->id == (StgThreadID)tok) {
2223       *prev = tso->link;
2224       break;
2225     }
2226   }
2227   if (tso == END_TSO_QUEUE) {
2228     barf("resumeThread: thread not found");
2229   }
2230   tso->link = END_TSO_QUEUE;
2231
2232   if(tso->why_blocked == BlockedOnCCall) {
2233       awakenBlockedQueueNoLock(tso->blocked_exceptions);
2234       tso->blocked_exceptions = NULL;
2235   }
2236
2237   /* Reset blocking status */
2238   tso->why_blocked  = NotBlocked;
2239
2240   cap->r.rCurrentTSO = tso;
2241   cap->r.rInHaskell = rtsTrue;
2242   RELEASE_LOCK(&sched_mutex);
2243   errno = saved_errno;
2244   return &cap->r;
2245 }
2246
2247 /* ---------------------------------------------------------------------------
2248  * Comparing Thread ids.
2249  *
2250  * This is used from STG land in the implementation of the
2251  * instances of Eq/Ord for ThreadIds.
2252  * ------------------------------------------------------------------------ */
2253
2254 int
2255 cmp_thread(StgPtr tso1, StgPtr tso2)
2256 {
2257   StgThreadID id1 = ((StgTSO *)tso1)->id;
2258   StgThreadID id2 = ((StgTSO *)tso2)->id;
2259
2260   if (id1 < id2) return (-1);
2261   if (id1 > id2) return 1;
2262   return 0;
2263 }
2264
2265 /* ---------------------------------------------------------------------------
2266  * Fetching the ThreadID from an StgTSO.
2267  *
2268  * This is used in the implementation of Show for ThreadIds.
2269  * ------------------------------------------------------------------------ */
2270 int
2271 rts_getThreadId(StgPtr tso)
2272 {
2273   return ((StgTSO *)tso)->id;
2274 }
2275
2276 #ifdef DEBUG
2277 void
2278 labelThread(StgPtr tso, char *label)
2279 {
2280   int len;
2281   void *buf;
2282
2283   /* Caveat: Once set, you can only set the thread name to "" */
2284   len = strlen(label)+1;
2285   buf = stgMallocBytes(len * sizeof(char), "Schedule.c:labelThread()");
2286   strncpy(buf,label,len);
2287   /* Update will free the old memory for us */
2288   updateThreadLabel(((StgTSO *)tso)->id,buf);
2289 }
2290 #endif /* DEBUG */
2291
2292 /* ---------------------------------------------------------------------------
2293    Create a new thread.
2294
2295    The new thread starts with the given stack size.  Before the
2296    scheduler can run, however, this thread needs to have a closure
2297    (and possibly some arguments) pushed on its stack.  See
2298    pushClosure() in Schedule.h.
2299
2300    createGenThread() and createIOThread() (in SchedAPI.h) are
2301    convenient packaged versions of this function.
2302
2303    currently pri (priority) is only used in a GRAN setup -- HWL
2304    ------------------------------------------------------------------------ */
2305 #if defined(GRAN)
2306 /*   currently pri (priority) is only used in a GRAN setup -- HWL */
2307 StgTSO *
2308 createThread(nat size, StgInt pri)
2309 #else
2310 StgTSO *
2311 createThread(nat size)
2312 #endif
2313 {
2314
2315     StgTSO *tso;
2316     nat stack_size;
2317
2318     ACQUIRE_LOCK(&sched_mutex);
2319
2320     /* First check whether we should create a thread at all */
2321 #if defined(PARALLEL_HASKELL)
2322   /* check that no more than RtsFlags.ParFlags.maxThreads threads are created */
2323   if (advisory_thread_count >= RtsFlags.ParFlags.maxThreads) {
2324     threadsIgnored++;
2325     debugBelch("{createThread}Daq ghuH: refusing to create another thread; no more than %d threads allowed (currently %d)\n",
2326           RtsFlags.ParFlags.maxThreads, advisory_thread_count);
2327     RELEASE_LOCK(&sched_mutex);
2328     return END_TSO_QUEUE;
2329   }
2330   threadsCreated++;
2331 #endif
2332
2333 #if defined(GRAN)
2334   ASSERT(!RtsFlags.GranFlags.Light || CurrentProc==0);
2335 #endif
2336
2337   // ToDo: check whether size = stack_size - TSO_STRUCT_SIZEW
2338
2339   /* catch ridiculously small stack sizes */
2340   if (size < MIN_STACK_WORDS + TSO_STRUCT_SIZEW) {
2341     size = MIN_STACK_WORDS + TSO_STRUCT_SIZEW;
2342   }
2343
2344   stack_size = size - TSO_STRUCT_SIZEW;
2345
2346   tso = (StgTSO *)allocate(size);
2347   TICK_ALLOC_TSO(stack_size, 0);
2348
2349   SET_HDR(tso, &stg_TSO_info, CCS_SYSTEM);
2350 #if defined(GRAN)
2351   SET_GRAN_HDR(tso, ThisPE);
2352 #endif
2353
2354   // Always start with the compiled code evaluator
2355   tso->what_next = ThreadRunGHC;
2356
2357   tso->id = next_thread_id++;
2358   tso->why_blocked  = NotBlocked;
2359   tso->blocked_exceptions = NULL;
2360
2361   tso->saved_errno = 0;
2362   tso->main = NULL;
2363
2364   tso->stack_size   = stack_size;
2365   tso->max_stack_size = round_to_mblocks(RtsFlags.GcFlags.maxStkSize)
2366                               - TSO_STRUCT_SIZEW;
2367   tso->sp           = (P_)&(tso->stack) + stack_size;
2368
2369   tso->trec = NO_TREC;
2370
2371 #ifdef PROFILING
2372   tso->prof.CCCS = CCS_MAIN;
2373 #endif
2374
2375   /* put a stop frame on the stack */
2376   tso->sp -= sizeofW(StgStopFrame);
2377   SET_HDR((StgClosure*)tso->sp,(StgInfoTable *)&stg_stop_thread_info,CCS_SYSTEM);
2378   tso->link = END_TSO_QUEUE;
2379
2380   // ToDo: check this
2381 #if defined(GRAN)
2382   /* uses more flexible routine in GranSim */
2383   insertThread(tso, CurrentProc);
2384 #else
2385   /* In a non-GranSim setup the pushing of a TSO onto the runq is separated
2386    * from its creation
2387    */
2388 #endif
2389
2390 #if defined(GRAN)
2391   if (RtsFlags.GranFlags.GranSimStats.Full)
2392     DumpGranEvent(GR_START,tso);
2393 #elif defined(PARALLEL_HASKELL)
2394   if (RtsFlags.ParFlags.ParStats.Full)
2395     DumpGranEvent(GR_STARTQ,tso);
2396   /* HACk to avoid SCHEDULE
2397      LastTSO = tso; */
2398 #endif
2399
2400   /* Link the new thread on the global thread list.
2401    */
2402   tso->global_link = all_threads;
2403   all_threads = tso;
2404
2405 #if defined(DIST)
2406   tso->dist.priority = MandatoryPriority; //by default that is...
2407 #endif
2408
2409 #if defined(GRAN)
2410   tso->gran.pri = pri;
2411 # if defined(DEBUG)
2412   tso->gran.magic = TSO_MAGIC; // debugging only
2413 # endif
2414   tso->gran.sparkname   = 0;
2415   tso->gran.startedat   = CURRENT_TIME;
2416   tso->gran.exported    = 0;
2417   tso->gran.basicblocks = 0;
2418   tso->gran.allocs      = 0;
2419   tso->gran.exectime    = 0;
2420   tso->gran.fetchtime   = 0;
2421   tso->gran.fetchcount  = 0;
2422   tso->gran.blocktime   = 0;
2423   tso->gran.blockcount  = 0;
2424   tso->gran.blockedat   = 0;
2425   tso->gran.globalsparks = 0;
2426   tso->gran.localsparks  = 0;
2427   if (RtsFlags.GranFlags.Light)
2428     tso->gran.clock  = Now; /* local clock */
2429   else
2430     tso->gran.clock  = 0;
2431
2432   IF_DEBUG(gran,printTSO(tso));
2433 #elif defined(PARALLEL_HASKELL)
2434 # if defined(DEBUG)
2435   tso->par.magic = TSO_MAGIC; // debugging only
2436 # endif
2437   tso->par.sparkname   = 0;
2438   tso->par.startedat   = CURRENT_TIME;
2439   tso->par.exported    = 0;
2440   tso->par.basicblocks = 0;
2441   tso->par.allocs      = 0;
2442   tso->par.exectime    = 0;
2443   tso->par.fetchtime   = 0;
2444   tso->par.fetchcount  = 0;
2445   tso->par.blocktime   = 0;
2446   tso->par.blockcount  = 0;
2447   tso->par.blockedat   = 0;
2448   tso->par.globalsparks = 0;
2449   tso->par.localsparks  = 0;
2450 #endif
2451
2452 #if defined(GRAN)
2453   globalGranStats.tot_threads_created++;
2454   globalGranStats.threads_created_on_PE[CurrentProc]++;
2455   globalGranStats.tot_sq_len += spark_queue_len(CurrentProc);
2456   globalGranStats.tot_sq_probes++;
2457 #elif defined(PARALLEL_HASKELL)
2458   // collect parallel global statistics (currently done together with GC stats)
2459   if (RtsFlags.ParFlags.ParStats.Global &&
2460       RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
2461     //debugBelch("Creating thread %d @ %11.2f\n", tso->id, usertime());
2462     globalParStats.tot_threads_created++;
2463   }
2464 #endif
2465
2466 #if defined(GRAN)
2467   IF_GRAN_DEBUG(pri,
2468                 sched_belch("==__ schedule: Created TSO %d (%p);",
2469                       CurrentProc, tso, tso->id));
2470 #elif defined(PARALLEL_HASKELL)
2471   IF_PAR_DEBUG(verbose,
2472                sched_belch("==__ schedule: Created TSO %d (%p); %d threads active",
2473                            (long)tso->id, tso, advisory_thread_count));
2474 #else
2475   IF_DEBUG(scheduler,sched_belch("created thread %ld, stack size = %lx words",
2476                                  (long)tso->id, (long)tso->stack_size));
2477 #endif
2478   RELEASE_LOCK(&sched_mutex);
2479   return tso;
2480 }
2481
2482 #if defined(PAR)
2483 /* RFP:
2484    all parallel thread creation calls should fall through the following routine.
2485 */
2486 StgTSO *
2487 createThreadFromSpark(rtsSpark spark)
2488 { StgTSO *tso;
2489   ASSERT(spark != (rtsSpark)NULL);
2490 // JB: TAKE CARE OF THIS COUNTER! BUGGY
2491   if (advisory_thread_count >= RtsFlags.ParFlags.maxThreads)
2492   { threadsIgnored++;
2493     barf("{createSparkThread}Daq ghuH: refusing to create another thread; no more than %d threads allowed (currently %d)",
2494           RtsFlags.ParFlags.maxThreads, advisory_thread_count);
2495     return END_TSO_QUEUE;
2496   }
2497   else
2498   { threadsCreated++;
2499     tso = createThread(RtsFlags.GcFlags.initialStkSize);
2500     if (tso==END_TSO_QUEUE)
2501       barf("createSparkThread: Cannot create TSO");
2502 #if defined(DIST)
2503     tso->priority = AdvisoryPriority;
2504 #endif
2505     pushClosure(tso,spark);
2506     addToRunQueue(tso);
2507     advisory_thread_count++;  // JB: TAKE CARE OF THIS COUNTER! BUGGY
2508   }
2509   return tso;
2510 }
2511 #endif
2512
2513 /*
2514   Turn a spark into a thread.
2515   ToDo: fix for SMP (needs to acquire SCHED_MUTEX!)
2516 */
2517 #if 0
2518 StgTSO *
2519 activateSpark (rtsSpark spark)
2520 {
2521   StgTSO *tso;
2522
2523   tso = createSparkThread(spark);
2524   if (RtsFlags.ParFlags.ParStats.Full) {
2525     //ASSERT(run_queue_hd == END_TSO_QUEUE); // I think ...
2526       IF_PAR_DEBUG(verbose,
2527                    debugBelch("==^^ activateSpark: turning spark of closure %p (%s) into a thread\n",
2528                               (StgClosure *)spark, info_type((StgClosure *)spark)));
2529   }
2530   // ToDo: fwd info on local/global spark to thread -- HWL
2531   // tso->gran.exported =  spark->exported;
2532   // tso->gran.locked =   !spark->global;
2533   // tso->gran.sparkname = spark->name;
2534
2535   return tso;
2536 }
2537 #endif
2538
2539 /* ---------------------------------------------------------------------------
2540  * scheduleThread()
2541  *
2542  * scheduleThread puts a thread on the head of the runnable queue.
2543  * This will usually be done immediately after a thread is created.
2544  * The caller of scheduleThread must create the thread using e.g.
2545  * createThread and push an appropriate closure
2546  * on this thread's stack before the scheduler is invoked.
2547  * ------------------------------------------------------------------------ */
2548
2549 static void
2550 scheduleThread_(StgTSO *tso)
2551 {
2552   // The thread goes at the *end* of the run-queue, to avoid possible
2553   // starvation of any threads already on the queue.
2554   APPEND_TO_RUN_QUEUE(tso);
2555   threadRunnable();
2556 }
2557
2558 void
2559 scheduleThread(StgTSO* tso)
2560 {
2561   ACQUIRE_LOCK(&sched_mutex);
2562   scheduleThread_(tso);
2563   RELEASE_LOCK(&sched_mutex);
2564 }
2565
2566 #if defined(RTS_SUPPORTS_THREADS)
2567 static Condition bound_cond_cache;
2568 static int bound_cond_cache_full = 0;
2569 #endif
2570
2571
2572 SchedulerStatus
2573 scheduleWaitThread(StgTSO* tso, /*[out]*/HaskellObj* ret,
2574                    Capability *initialCapability)
2575 {
2576     // Precondition: sched_mutex must be held
2577     StgMainThread *m;
2578
2579     m = stgMallocBytes(sizeof(StgMainThread), "waitThread");
2580     m->tso = tso;
2581     tso->main = m;
2582     m->ret = ret;
2583     m->stat = NoStatus;
2584     m->link = main_threads;
2585     m->prev = NULL;
2586     if (main_threads != NULL) {
2587         main_threads->prev = m;
2588     }
2589     main_threads = m;
2590
2591 #if defined(RTS_SUPPORTS_THREADS)
2592     // Allocating a new condition for each thread is expensive, so we
2593     // cache one.  This is a pretty feeble hack, but it helps speed up
2594     // consecutive call-ins quite a bit.
2595     if (bound_cond_cache_full) {
2596         m->bound_thread_cond = bound_cond_cache;
2597         bound_cond_cache_full = 0;
2598     } else {
2599         initCondition(&m->bound_thread_cond);
2600     }
2601 #endif
2602
2603     /* Put the thread on the main-threads list prior to scheduling the TSO.
2604        Failure to do so introduces a race condition in the MT case (as
2605        identified by Wolfgang Thaller), whereby the new task/OS thread
2606        created by scheduleThread_() would complete prior to the thread
2607        that spawned it managed to put 'itself' on the main-threads list.
2608        The upshot of it all being that the worker thread wouldn't get to
2609        signal the completion of the its work item for the main thread to
2610        see (==> it got stuck waiting.)    -- sof 6/02.
2611     */
2612     IF_DEBUG(scheduler, sched_belch("waiting for thread (%d)", tso->id));
2613
2614     APPEND_TO_RUN_QUEUE(tso);
2615     // NB. Don't call threadRunnable() here, because the thread is
2616     // bound and only runnable by *this* OS thread, so waking up other
2617     // workers will just slow things down.
2618
2619     return waitThread_(m, initialCapability);
2620 }
2621
2622 /* ---------------------------------------------------------------------------
2623  * initScheduler()
2624  *
2625  * Initialise the scheduler.  This resets all the queues - if the
2626  * queues contained any threads, they'll be garbage collected at the
2627  * next pass.
2628  *
2629  * ------------------------------------------------------------------------ */
2630
2631 void
2632 initScheduler(void)
2633 {
2634 #if defined(GRAN)
2635   nat i;
2636
2637   for (i=0; i<=MAX_PROC; i++) {
2638     run_queue_hds[i]      = END_TSO_QUEUE;
2639     run_queue_tls[i]      = END_TSO_QUEUE;
2640     blocked_queue_hds[i]  = END_TSO_QUEUE;
2641     blocked_queue_tls[i]  = END_TSO_QUEUE;
2642     ccalling_threadss[i]  = END_TSO_QUEUE;
2643     blackhole_queue[i]    = END_TSO_QUEUE;
2644     sleeping_queue        = END_TSO_QUEUE;
2645   }
2646 #else
2647   run_queue_hd      = END_TSO_QUEUE;
2648   run_queue_tl      = END_TSO_QUEUE;
2649   blocked_queue_hd  = END_TSO_QUEUE;
2650   blocked_queue_tl  = END_TSO_QUEUE;
2651   blackhole_queue   = END_TSO_QUEUE;
2652   sleeping_queue    = END_TSO_QUEUE;
2653 #endif
2654
2655   suspended_ccalling_threads  = END_TSO_QUEUE;
2656
2657   main_threads = NULL;
2658   all_threads  = END_TSO_QUEUE;
2659
2660   context_switch = 0;
2661   interrupted    = 0;
2662
2663   RtsFlags.ConcFlags.ctxtSwitchTicks =
2664       RtsFlags.ConcFlags.ctxtSwitchTime / TICK_MILLISECS;
2665
2666 #if defined(RTS_SUPPORTS_THREADS)
2667   /* Initialise the mutex and condition variables used by
2668    * the scheduler. */
2669   initMutex(&sched_mutex);
2670   initMutex(&term_mutex);
2671 #endif
2672
2673   ACQUIRE_LOCK(&sched_mutex);
2674
2675   /* A capability holds the state a native thread needs in
2676    * order to execute STG code. At least one capability is
2677    * floating around (only SMP builds have more than one).
2678    */
2679   initCapabilities();
2680
2681 #if defined(RTS_SUPPORTS_THREADS)
2682   initTaskManager();
2683 #endif
2684
2685 #if defined(SMP)
2686   /* eagerly start some extra workers */
2687   startingWorkerThread = RtsFlags.ParFlags.nNodes;
2688   startTasks(RtsFlags.ParFlags.nNodes, taskStart);
2689 #endif
2690
2691 #if /* defined(SMP) ||*/ defined(PARALLEL_HASKELL)
2692   initSparkPools();
2693 #endif
2694
2695   RELEASE_LOCK(&sched_mutex);
2696 }
2697
2698 void
2699 exitScheduler( void )
2700 {
2701     interrupted = rtsTrue;
2702     shutting_down_scheduler = rtsTrue;
2703 #if defined(RTS_SUPPORTS_THREADS)
2704     if (threadIsTask(osThreadId())) { taskStop(); }
2705     stopTaskManager();
2706 #endif
2707 }
2708
2709 /* ----------------------------------------------------------------------------
2710    Managing the per-task allocation areas.
2711
2712    Each capability comes with an allocation area.  These are
2713    fixed-length block lists into which allocation can be done.
2714
2715    ToDo: no support for two-space collection at the moment???
2716    ------------------------------------------------------------------------- */
2717
2718 static SchedulerStatus
2719 waitThread_(StgMainThread* m, Capability *initialCapability)
2720 {
2721   SchedulerStatus stat;
2722
2723   // Precondition: sched_mutex must be held.
2724   IF_DEBUG(scheduler, sched_belch("new main thread (%d)", m->tso->id));
2725
2726 #if defined(GRAN)
2727   /* GranSim specific init */
2728   CurrentTSO = m->tso;                // the TSO to run
2729   procStatus[MainProc] = Busy;        // status of main PE
2730   CurrentProc = MainProc;             // PE to run it on
2731   schedule(m,initialCapability);
2732 #else
2733   schedule(m,initialCapability);
2734   ASSERT(m->stat != NoStatus);
2735 #endif
2736
2737   stat = m->stat;
2738
2739 #if defined(RTS_SUPPORTS_THREADS)
2740   // Free the condition variable, returning it to the cache if possible.
2741   if (!bound_cond_cache_full) {
2742       bound_cond_cache = m->bound_thread_cond;
2743       bound_cond_cache_full = 1;
2744   } else {
2745       closeCondition(&m->bound_thread_cond);
2746   }
2747 #endif
2748
2749   IF_DEBUG(scheduler, sched_belch("main thread (%d) finished", m->tso->id));
2750   stgFree(m);
2751
2752   // Postcondition: sched_mutex still held
2753   return stat;
2754 }
2755
2756 /* ---------------------------------------------------------------------------
2757    Where are the roots that we know about?
2758
2759         - all the threads on the runnable queue
2760         - all the threads on the blocked queue
2761         - all the threads on the sleeping queue
2762         - all the thread currently executing a _ccall_GC
2763         - all the "main threads"
2764
2765    ------------------------------------------------------------------------ */
2766
2767 /* This has to be protected either by the scheduler monitor, or by the
2768         garbage collection monitor (probably the latter).
2769         KH @ 25/10/99
2770 */
2771
2772 void
2773 GetRoots( evac_fn evac )
2774 {
2775 #if defined(GRAN)
2776   {
2777     nat i;
2778     for (i=0; i<=RtsFlags.GranFlags.proc; i++) {
2779       if ((run_queue_hds[i] != END_TSO_QUEUE) && ((run_queue_hds[i] != NULL)))
2780           evac((StgClosure **)&run_queue_hds[i]);
2781       if ((run_queue_tls[i] != END_TSO_QUEUE) && ((run_queue_tls[i] != NULL)))
2782           evac((StgClosure **)&run_queue_tls[i]);
2783
2784       if ((blocked_queue_hds[i] != END_TSO_QUEUE) && ((blocked_queue_hds[i] != NULL)))
2785           evac((StgClosure **)&blocked_queue_hds[i]);
2786       if ((blocked_queue_tls[i] != END_TSO_QUEUE) && ((blocked_queue_tls[i] != NULL)))
2787           evac((StgClosure **)&blocked_queue_tls[i]);
2788       if ((ccalling_threadss[i] != END_TSO_QUEUE) && ((ccalling_threadss[i] != NULL)))
2789           evac((StgClosure **)&ccalling_threads[i]);
2790     }
2791   }
2792
2793   markEventQueue();
2794
2795 #else /* !GRAN */
2796   if (run_queue_hd != END_TSO_QUEUE) {
2797       ASSERT(run_queue_tl != END_TSO_QUEUE);
2798       evac((StgClosure **)&run_queue_hd);
2799       evac((StgClosure **)&run_queue_tl);
2800   }
2801
2802   if (blocked_queue_hd != END_TSO_QUEUE) {
2803       ASSERT(blocked_queue_tl != END_TSO_QUEUE);
2804       evac((StgClosure **)&blocked_queue_hd);
2805       evac((StgClosure **)&blocked_queue_tl);
2806   }
2807
2808   if (sleeping_queue != END_TSO_QUEUE) {
2809       evac((StgClosure **)&sleeping_queue);
2810   }
2811 #endif
2812
2813   if (blackhole_queue != END_TSO_QUEUE) {
2814       evac((StgClosure **)&blackhole_queue);
2815   }
2816
2817   if (suspended_ccalling_threads != END_TSO_QUEUE) {
2818       evac((StgClosure **)&suspended_ccalling_threads);
2819   }
2820
2821 #if defined(PARALLEL_HASKELL) || defined(GRAN)
2822   markSparkQueue(evac);
2823 #endif
2824
2825 #if defined(RTS_USER_SIGNALS)
2826   // mark the signal handlers (signals should be already blocked)
2827   markSignalHandlers(evac);
2828 #endif
2829 }
2830
2831 /* -----------------------------------------------------------------------------
2832    performGC
2833
2834    This is the interface to the garbage collector from Haskell land.
2835    We provide this so that external C code can allocate and garbage
2836    collect when called from Haskell via _ccall_GC.
2837
2838    It might be useful to provide an interface whereby the programmer
2839    can specify more roots (ToDo).
2840
2841    This needs to be protected by the GC condition variable above.  KH.
2842    -------------------------------------------------------------------------- */
2843
2844 static void (*extra_roots)(evac_fn);
2845
2846 void
2847 performGC(void)
2848 {
2849   /* Obligated to hold this lock upon entry */
2850   ACQUIRE_LOCK(&sched_mutex);
2851   GarbageCollect(GetRoots,rtsFalse);
2852   RELEASE_LOCK(&sched_mutex);
2853 }
2854
2855 void
2856 performMajorGC(void)
2857 {
2858   ACQUIRE_LOCK(&sched_mutex);
2859   GarbageCollect(GetRoots,rtsTrue);
2860   RELEASE_LOCK(&sched_mutex);
2861 }
2862
2863 static void
2864 AllRoots(evac_fn evac)
2865 {
2866     GetRoots(evac);             // the scheduler's roots
2867     extra_roots(evac);          // the user's roots
2868 }
2869
2870 void
2871 performGCWithRoots(void (*get_roots)(evac_fn))
2872 {
2873   ACQUIRE_LOCK(&sched_mutex);
2874   extra_roots = get_roots;
2875   GarbageCollect(AllRoots,rtsFalse);
2876   RELEASE_LOCK(&sched_mutex);
2877 }
2878
2879 /* -----------------------------------------------------------------------------
2880    Stack overflow
2881
2882    If the thread has reached its maximum stack size, then raise the
2883    StackOverflow exception in the offending thread.  Otherwise
2884    relocate the TSO into a larger chunk of memory and adjust its stack
2885    size appropriately.
2886    -------------------------------------------------------------------------- */
2887
2888 static StgTSO *
2889 threadStackOverflow(StgTSO *tso)
2890 {
2891   nat new_stack_size, stack_words;
2892   lnat new_tso_size;
2893   StgPtr new_sp;
2894   StgTSO *dest;
2895
2896   IF_DEBUG(sanity,checkTSO(tso));
2897   if (tso->stack_size >= tso->max_stack_size) {
2898
2899     IF_DEBUG(gc,
2900              debugBelch("@@ threadStackOverflow of TSO %ld (%p): stack too large (now %ld; max is %ld)\n",
2901                    (long)tso->id, tso, (long)tso->stack_size, (long)tso->max_stack_size);
2902              /* If we're debugging, just print out the top of the stack */
2903              printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size,
2904                                               tso->sp+64)));
2905
2906     /* Send this thread the StackOverflow exception */
2907     raiseAsync(tso, (StgClosure *)stackOverflow_closure);
2908     return tso;
2909   }
2910
2911   /* Try to double the current stack size.  If that takes us over the
2912    * maximum stack size for this thread, then use the maximum instead.
2913    * Finally round up so the TSO ends up as a whole number of blocks.
2914    */
2915   new_stack_size = stg_min(tso->stack_size * 2, tso->max_stack_size);
2916   new_tso_size   = (lnat)BLOCK_ROUND_UP(new_stack_size * sizeof(W_) +
2917                                        TSO_STRUCT_SIZE)/sizeof(W_);
2918   new_tso_size = round_to_mblocks(new_tso_size);  /* Be MBLOCK-friendly */
2919   new_stack_size = new_tso_size - TSO_STRUCT_SIZEW;
2920
2921   IF_DEBUG(scheduler, debugBelch("== sched: increasing stack size from %d words to %d.\n", tso->stack_size, new_stack_size));
2922
2923   dest = (StgTSO *)allocate(new_tso_size);
2924   TICK_ALLOC_TSO(new_stack_size,0);
2925
2926   /* copy the TSO block and the old stack into the new area */
2927   memcpy(dest,tso,TSO_STRUCT_SIZE);
2928   stack_words = tso->stack + tso->stack_size - tso->sp;
2929   new_sp = (P_)dest + new_tso_size - stack_words;
2930   memcpy(new_sp, tso->sp, stack_words * sizeof(W_));
2931
2932   /* relocate the stack pointers... */
2933   dest->sp         = new_sp;
2934   dest->stack_size = new_stack_size;
2935
2936   /* Mark the old TSO as relocated.  We have to check for relocated
2937    * TSOs in the garbage collector and any primops that deal with TSOs.
2938    *
2939    * It's important to set the sp value to just beyond the end
2940    * of the stack, so we don't attempt to scavenge any part of the
2941    * dead TSO's stack.
2942    */
2943   tso->what_next = ThreadRelocated;
2944   tso->link = dest;
2945   tso->sp = (P_)&(tso->stack[tso->stack_size]);
2946   tso->why_blocked = NotBlocked;
2947
2948   IF_PAR_DEBUG(verbose,
2949                debugBelch("@@ threadStackOverflow of TSO %d (now at %p): stack size increased to %ld\n",
2950                      tso->id, tso, tso->stack_size);
2951                /* If we're debugging, just print out the top of the stack */
2952                printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size,
2953                                                 tso->sp+64)));
2954
2955   IF_DEBUG(sanity,checkTSO(tso));
2956 #if 0
2957   IF_DEBUG(scheduler,printTSO(dest));
2958 #endif
2959
2960   return dest;
2961 }
2962
2963 /* ---------------------------------------------------------------------------
2964    Wake up a queue that was blocked on some resource.
2965    ------------------------------------------------------------------------ */
2966
2967 #if defined(GRAN)
2968 STATIC_INLINE void
2969 unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node )
2970 {
2971 }
2972 #elif defined(PARALLEL_HASKELL)
2973 STATIC_INLINE void
2974 unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node )
2975 {
2976   /* write RESUME events to log file and
2977      update blocked and fetch time (depending on type of the orig closure) */
2978   if (RtsFlags.ParFlags.ParStats.Full) {
2979     DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
2980                      GR_RESUMEQ, ((StgTSO *)bqe), ((StgTSO *)bqe)->block_info.closure,
2981                      0, 0 /* spark_queue_len(ADVISORY_POOL) */);
2982     if (EMPTY_RUN_QUEUE())
2983       emitSchedule = rtsTrue;
2984
2985     switch (get_itbl(node)->type) {
2986         case FETCH_ME_BQ:
2987           ((StgTSO *)bqe)->par.fetchtime += CURRENT_TIME-((StgTSO *)bqe)->par.blockedat;
2988           break;
2989         case RBH:
2990         case FETCH_ME:
2991         case BLACKHOLE_BQ:
2992           ((StgTSO *)bqe)->par.blocktime += CURRENT_TIME-((StgTSO *)bqe)->par.blockedat;
2993           break;
2994 #ifdef DIST
2995         case MVAR:
2996           break;
2997 #endif
2998         default:
2999           barf("{unblockOneLocked}Daq Qagh: unexpected closure in blocking queue");
3000         }
3001       }
3002 }
3003 #endif
3004
3005 #if defined(GRAN)
3006 StgBlockingQueueElement *
3007 unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node)
3008 {
3009     StgTSO *tso;
3010     PEs node_loc, tso_loc;
3011
3012     node_loc = where_is(node); // should be lifted out of loop
3013     tso = (StgTSO *)bqe;  // wastes an assignment to get the type right
3014     tso_loc = where_is((StgClosure *)tso);
3015     if (IS_LOCAL_TO(PROCS(node),tso_loc)) { // TSO is local
3016       /* !fake_fetch => TSO is on CurrentProc is same as IS_LOCAL_TO */
3017       ASSERT(CurrentProc!=node_loc || tso_loc==CurrentProc);
3018       CurrentTime[CurrentProc] += RtsFlags.GranFlags.Costs.lunblocktime;
3019       // insertThread(tso, node_loc);
3020       new_event(tso_loc, tso_loc, CurrentTime[CurrentProc],
3021                 ResumeThread,
3022                 tso, node, (rtsSpark*)NULL);
3023       tso->link = END_TSO_QUEUE; // overwrite link just to be sure
3024       // len_local++;
3025       // len++;
3026     } else { // TSO is remote (actually should be FMBQ)
3027       CurrentTime[CurrentProc] += RtsFlags.GranFlags.Costs.mpacktime +
3028                                   RtsFlags.GranFlags.Costs.gunblocktime +
3029                                   RtsFlags.GranFlags.Costs.latency;
3030       new_event(tso_loc, CurrentProc, CurrentTime[CurrentProc],
3031                 UnblockThread,
3032                 tso, node, (rtsSpark*)NULL);
3033       tso->link = END_TSO_QUEUE; // overwrite link just to be sure
3034       // len++;
3035     }
3036     /* the thread-queue-overhead is accounted for in either Resume or UnblockThread */
3037     IF_GRAN_DEBUG(bq,
3038                   debugBelch(" %s TSO %d (%p) [PE %d] (block_info.closure=%p) (next=%p) ,",
3039                           (node_loc==tso_loc ? "Local" : "Global"),
3040                           tso->id, tso, CurrentProc, tso->block_info.closure, tso->link));
3041     tso->block_info.closure = NULL;
3042     IF_DEBUG(scheduler,debugBelch("-- Waking up thread %ld (%p)\n",
3043                              tso->id, tso));
3044 }
3045 #elif defined(PARALLEL_HASKELL)
3046 StgBlockingQueueElement *
3047 unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node)
3048 {
3049     StgBlockingQueueElement *next;
3050
3051     switch (get_itbl(bqe)->type) {
3052     case TSO:
3053       ASSERT(((StgTSO *)bqe)->why_blocked != NotBlocked);
3054       /* if it's a TSO just push it onto the run_queue */
3055       next = bqe->link;
3056       ((StgTSO *)bqe)->link = END_TSO_QUEUE; // debugging?
3057       APPEND_TO_RUN_QUEUE((StgTSO *)bqe);
3058       threadRunnable();
3059       unblockCount(bqe, node);
3060       /* reset blocking status after dumping event */
3061       ((StgTSO *)bqe)->why_blocked = NotBlocked;
3062       break;
3063
3064     case BLOCKED_FETCH:
3065       /* if it's a BLOCKED_FETCH put it on the PendingFetches list */
3066       next = bqe->link;
3067       bqe->link = (StgBlockingQueueElement *)PendingFetches;
3068       PendingFetches = (StgBlockedFetch *)bqe;
3069       break;
3070
3071 # if defined(DEBUG)
3072       /* can ignore this case in a non-debugging setup;
3073          see comments on RBHSave closures above */
3074     case CONSTR:
3075       /* check that the closure is an RBHSave closure */
3076       ASSERT(get_itbl((StgClosure *)bqe) == &stg_RBH_Save_0_info ||
3077              get_itbl((StgClosure *)bqe) == &stg_RBH_Save_1_info ||
3078              get_itbl((StgClosure *)bqe) == &stg_RBH_Save_2_info);
3079       break;
3080
3081     default:
3082       barf("{unblockOneLocked}Daq Qagh: Unexpected IP (%#lx; %s) in blocking queue at %#lx\n",
3083            get_itbl((StgClosure *)bqe), info_type((StgClosure *)bqe),
3084            (StgClosure *)bqe);
3085 # endif
3086     }
3087   IF_PAR_DEBUG(bq, debugBelch(", %p (%s)\n", bqe, info_type((StgClosure*)bqe)));
3088   return next;
3089 }
3090
3091 #else /* !GRAN && !PARALLEL_HASKELL */
3092 StgTSO *
3093 unblockOneLocked(StgTSO *tso)
3094 {
3095   StgTSO *next;
3096
3097   ASSERT(get_itbl(tso)->type == TSO);
3098   ASSERT(tso->why_blocked != NotBlocked);
3099   tso->why_blocked = NotBlocked;
3100   next = tso->link;
3101   tso->link = END_TSO_QUEUE;
3102   APPEND_TO_RUN_QUEUE(tso);
3103   threadRunnable();
3104   IF_DEBUG(scheduler,sched_belch("waking up thread %ld", (long)tso->id));
3105   return next;
3106 }
3107 #endif
3108
3109 #if defined(GRAN) || defined(PARALLEL_HASKELL)
3110 INLINE_ME StgBlockingQueueElement *
3111 unblockOne(StgBlockingQueueElement *bqe, StgClosure *node)
3112 {
3113   ACQUIRE_LOCK(&sched_mutex);
3114   bqe = unblockOneLocked(bqe, node);
3115   RELEASE_LOCK(&sched_mutex);
3116   return bqe;
3117 }
3118 #else
3119 INLINE_ME StgTSO *
3120 unblockOne(StgTSO *tso)
3121 {
3122   ACQUIRE_LOCK(&sched_mutex);
3123   tso = unblockOneLocked(tso);
3124   RELEASE_LOCK(&sched_mutex);
3125   return tso;
3126 }
3127 #endif
3128
3129 #if defined(GRAN)
3130 void
3131 awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node)
3132 {
3133   StgBlockingQueueElement *bqe;
3134   PEs node_loc;
3135   nat len = 0;
3136
3137   IF_GRAN_DEBUG(bq,
3138                 debugBelch("##-_ AwBQ for node %p on PE %d @ %ld by TSO %d (%p): \n", \
3139                       node, CurrentProc, CurrentTime[CurrentProc],
3140                       CurrentTSO->id, CurrentTSO));
3141
3142   node_loc = where_is(node);
3143
3144   ASSERT(q == END_BQ_QUEUE ||
3145          get_itbl(q)->type == TSO ||   // q is either a TSO or an RBHSave
3146          get_itbl(q)->type == CONSTR); // closure (type constructor)
3147   ASSERT(is_unique(node));
3148
3149   /* FAKE FETCH: magically copy the node to the tso's proc;
3150      no Fetch necessary because in reality the node should not have been
3151      moved to the other PE in the first place
3152   */
3153   if (CurrentProc!=node_loc) {
3154     IF_GRAN_DEBUG(bq,
3155                   debugBelch("## node %p is on PE %d but CurrentProc is %d (TSO %d); assuming fake fetch and adjusting bitmask (old: %#x)\n",
3156                         node, node_loc, CurrentProc, CurrentTSO->id,
3157                         // CurrentTSO, where_is(CurrentTSO),
3158                         node->header.gran.procs));
3159     node->header.gran.procs = (node->header.gran.procs) | PE_NUMBER(CurrentProc);
3160     IF_GRAN_DEBUG(bq,
3161                   debugBelch("## new bitmask of node %p is %#x\n",
3162                         node, node->header.gran.procs));
3163     if (RtsFlags.GranFlags.GranSimStats.Global) {
3164       globalGranStats.tot_fake_fetches++;
3165     }
3166   }
3167
3168   bqe = q;
3169   // ToDo: check: ASSERT(CurrentProc==node_loc);
3170   while (get_itbl(bqe)->type==TSO) { // q != END_TSO_QUEUE) {
3171     //next = bqe->link;
3172     /*
3173        bqe points to the current element in the queue
3174        next points to the next element in the queue
3175     */
3176     //tso = (StgTSO *)bqe;  // wastes an assignment to get the type right
3177     //tso_loc = where_is(tso);
3178     len++;
3179     bqe = unblockOneLocked(bqe, node);
3180   }
3181
3182   /* if this is the BQ of an RBH, we have to put back the info ripped out of
3183      the closure to make room for the anchor of the BQ */
3184   if (bqe!=END_BQ_QUEUE) {
3185     ASSERT(get_itbl(node)->type == RBH && get_itbl(bqe)->type == CONSTR);
3186     /*
3187     ASSERT((info_ptr==&RBH_Save_0_info) ||
3188            (info_ptr==&RBH_Save_1_info) ||
3189            (info_ptr==&RBH_Save_2_info));
3190     */
3191     /* cf. convertToRBH in RBH.c for writing the RBHSave closure */
3192     ((StgRBH *)node)->blocking_queue = (StgBlockingQueueElement *)((StgRBHSave *)bqe)->payload[0];
3193     ((StgRBH *)node)->mut_link       = (StgMutClosure *)((StgRBHSave *)bqe)->payload[1];
3194
3195     IF_GRAN_DEBUG(bq,
3196                   debugBelch("## Filled in RBH_Save for %p (%s) at end of AwBQ\n",
3197                         node, info_type(node)));
3198   }
3199
3200   /* statistics gathering */
3201   if (RtsFlags.GranFlags.GranSimStats.Global) {
3202     // globalGranStats.tot_bq_processing_time += bq_processing_time;
3203     globalGranStats.tot_bq_len += len;      // total length of all bqs awakened
3204     // globalGranStats.tot_bq_len_local += len_local;  // same for local TSOs only
3205     globalGranStats.tot_awbq++;             // total no. of bqs awakened
3206   }
3207   IF_GRAN_DEBUG(bq,
3208                 debugBelch("## BQ Stats of %p: [%d entries] %s\n",
3209                         node, len, (bqe!=END_BQ_QUEUE) ? "RBH" : ""));
3210 }
3211 #elif defined(PARALLEL_HASKELL)
3212 void
3213 awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node)
3214 {
3215   StgBlockingQueueElement *bqe;
3216
3217   ACQUIRE_LOCK(&sched_mutex);
3218
3219   IF_PAR_DEBUG(verbose,
3220                debugBelch("##-_ AwBQ for node %p on [%x]: \n",
3221                      node, mytid));
3222 #ifdef DIST
3223   //RFP
3224   if(get_itbl(q)->type == CONSTR || q==END_BQ_QUEUE) {
3225     IF_PAR_DEBUG(verbose, debugBelch("## ... nothing to unblock so lets just return. RFP (BUG?)\n"));
3226     return;
3227   }
3228 #endif
3229
3230   ASSERT(q == END_BQ_QUEUE ||
3231          get_itbl(q)->type == TSO ||
3232          get_itbl(q)->type == BLOCKED_FETCH ||
3233          get_itbl(q)->type == CONSTR);
3234
3235   bqe = q;
3236   while (get_itbl(bqe)->type==TSO ||
3237          get_itbl(bqe)->type==BLOCKED_FETCH) {
3238     bqe = unblockOneLocked(bqe, node);
3239   }
3240   RELEASE_LOCK(&sched_mutex);
3241 }
3242
3243 #else   /* !GRAN && !PARALLEL_HASKELL */
3244
3245 void
3246 awakenBlockedQueueNoLock(StgTSO *tso)
3247 {
3248   if (tso == NULL) return; // hack; see bug #1235728, and comments in
3249                            // Exception.cmm
3250   while (tso != END_TSO_QUEUE) {
3251     tso = unblockOneLocked(tso);
3252   }
3253 }
3254
3255 void
3256 awakenBlockedQueue(StgTSO *tso)
3257 {
3258   if (tso == NULL) return; // hack; see bug #1235728, and comments in
3259                            // Exception.cmm
3260   ACQUIRE_LOCK(&sched_mutex);
3261   while (tso != END_TSO_QUEUE) {
3262     tso = unblockOneLocked(tso);
3263   }
3264   RELEASE_LOCK(&sched_mutex);
3265 }
3266 #endif
3267
3268 /* ---------------------------------------------------------------------------
3269    Interrupt execution
3270    - usually called inside a signal handler so it mustn't do anything fancy.
3271    ------------------------------------------------------------------------ */
3272
3273 void
3274 interruptStgRts(void)
3275 {
3276     interrupted    = 1;
3277     context_switch = 1;
3278     threadRunnable();
3279     /* ToDo: if invoked from a signal handler, this threadRunnable
3280      * only works if there's another thread (not this one) waiting to
3281      * be woken up.
3282      */
3283 }
3284
3285 /* -----------------------------------------------------------------------------
3286    Unblock a thread
3287
3288    This is for use when we raise an exception in another thread, which
3289    may be blocked.
3290    This has nothing to do with the UnblockThread event in GranSim. -- HWL
3291    -------------------------------------------------------------------------- */
3292
3293 #if defined(GRAN) || defined(PARALLEL_HASKELL)
3294 /*
3295   NB: only the type of the blocking queue is different in GranSim and GUM
3296       the operations on the queue-elements are the same
3297       long live polymorphism!
3298
3299   Locks: sched_mutex is held upon entry and exit.
3300
3301 */
3302 static void
3303 unblockThread(StgTSO *tso)
3304 {
3305   StgBlockingQueueElement *t, **last;
3306
3307   switch (tso->why_blocked) {
3308
3309   case NotBlocked:
3310     return;  /* not blocked */
3311
3312   case BlockedOnSTM:
3313     // Be careful: nothing to do here!  We tell the scheduler that the thread
3314     // is runnable and we leave it to the stack-walking code to abort the
3315     // transaction while unwinding the stack.  We should perhaps have a debugging
3316     // test to make sure that this really happens and that the 'zombie' transaction
3317     // does not get committed.
3318     goto done;
3319
3320   case BlockedOnMVar:
3321     ASSERT(get_itbl(tso->block_info.closure)->type == MVAR);
3322     {
3323       StgBlockingQueueElement *last_tso = END_BQ_QUEUE;
3324       StgMVar *mvar = (StgMVar *)(tso->block_info.closure);
3325
3326       last = (StgBlockingQueueElement **)&mvar->head;
3327       for (t = (StgBlockingQueueElement *)mvar->head;
3328            t != END_BQ_QUEUE;
3329            last = &t->link, last_tso = t, t = t->link) {
3330         if (t == (StgBlockingQueueElement *)tso) {
3331           *last = (StgBlockingQueueElement *)tso->link;
3332           if (mvar->tail == tso) {
3333             mvar->tail = (StgTSO *)last_tso;
3334           }
3335           goto done;
3336         }
3337       }
3338       barf("unblockThread (MVAR): TSO not found");
3339     }
3340
3341   case BlockedOnBlackHole:
3342     ASSERT(get_itbl(tso->block_info.closure)->type == BLACKHOLE_BQ);
3343     {
3344       StgBlockingQueue *bq = (StgBlockingQueue *)(tso->block_info.closure);
3345
3346       last = &bq->blocking_queue;
3347       for (t = bq->blocking_queue;
3348            t != END_BQ_QUEUE;
3349            last = &t->link, t = t->link) {
3350         if (t == (StgBlockingQueueElement *)tso) {
3351           *last = (StgBlockingQueueElement *)tso->link;
3352           goto done;
3353         }
3354       }
3355       barf("unblockThread (BLACKHOLE): TSO not found");
3356     }
3357
3358   case BlockedOnException:
3359     {
3360       StgTSO *target  = tso->block_info.tso;
3361
3362       ASSERT(get_itbl(target)->type == TSO);
3363
3364       if (target->what_next == ThreadRelocated) {
3365           target = target->link;
3366           ASSERT(get_itbl(target)->type == TSO);
3367       }
3368
3369       ASSERT(target->blocked_exceptions != NULL);
3370
3371       last = (StgBlockingQueueElement **)&target->blocked_exceptions;
3372       for (t = (StgBlockingQueueElement *)target->blocked_exceptions;
3373            t != END_BQ_QUEUE;
3374            last = &t->link, t = t->link) {
3375         ASSERT(get_itbl(t)->type == TSO);
3376         if (t == (StgBlockingQueueElement *)tso) {
3377           *last = (StgBlockingQueueElement *)tso->link;
3378           goto done;
3379         }
3380       }
3381       barf("unblockThread (Exception): TSO not found");
3382     }
3383
3384   case BlockedOnRead:
3385   case BlockedOnWrite:
3386 #if defined(mingw32_HOST_OS)
3387   case BlockedOnDoProc:
3388 #endif
3389     {
3390       /* take TSO off blocked_queue */
3391       StgBlockingQueueElement *prev = NULL;
3392       for (t = (StgBlockingQueueElement *)blocked_queue_hd; t != END_BQ_QUEUE;
3393            prev = t, t = t->link) {
3394         if (t == (StgBlockingQueueElement *)tso) {
3395           if (prev == NULL) {
3396             blocked_queue_hd = (StgTSO *)t->link;
3397             if ((StgBlockingQueueElement *)blocked_queue_tl == t) {
3398               blocked_queue_tl = END_TSO_QUEUE;
3399             }
3400           } else {
3401             prev->link = t->link;
3402             if ((StgBlockingQueueElement *)blocked_queue_tl == t) {
3403               blocked_queue_tl = (StgTSO *)prev;
3404             }
3405           }
3406 #if defined(mingw32_HOST_OS)
3407           /* (Cooperatively) signal that the worker thread should abort
3408            * the request.
3409            */
3410           abandonWorkRequest(tso->block_info.async_result->reqID);
3411 #endif
3412           goto done;
3413         }
3414       }
3415       barf("unblockThread (I/O): TSO not found");
3416     }
3417
3418   case BlockedOnDelay:
3419     {
3420       /* take TSO off sleeping_queue */
3421       StgBlockingQueueElement *prev = NULL;
3422       for (t = (StgBlockingQueueElement *)sleeping_queue; t != END_BQ_QUEUE;
3423            prev = t, t = t->link) {
3424         if (t == (StgBlockingQueueElement *)tso) {
3425           if (prev == NULL) {
3426             sleeping_queue = (StgTSO *)t->link;
3427           } else {
3428             prev->link = t->link;
3429           }
3430           goto done;
3431         }
3432       }
3433       barf("unblockThread (delay): TSO not found");
3434     }
3435
3436   default:
3437     barf("unblockThread");
3438   }
3439
3440  done:
3441   tso->link = END_TSO_QUEUE;
3442   tso->why_blocked = NotBlocked;
3443   tso->block_info.closure = NULL;
3444   PUSH_ON_RUN_QUEUE(tso);
3445 }
3446 #else
3447 static void
3448 unblockThread(StgTSO *tso)
3449 {
3450   StgTSO *t, **last;
3451
3452   /* To avoid locking unnecessarily. */
3453   if (tso->why_blocked == NotBlocked) {
3454     return;
3455   }
3456
3457   switch (tso->why_blocked) {
3458
3459   case BlockedOnSTM:
3460     // Be careful: nothing to do here!  We tell the scheduler that the thread
3461     // is runnable and we leave it to the stack-walking code to abort the
3462     // transaction while unwinding the stack.  We should perhaps have a debugging
3463     // test to make sure that this really happens and that the 'zombie' transaction
3464     // does not get committed.
3465     goto done;
3466
3467   case BlockedOnMVar:
3468     ASSERT(get_itbl(tso->block_info.closure)->type == MVAR);
3469     {
3470       StgTSO *last_tso = END_TSO_QUEUE;
3471       StgMVar *mvar = (StgMVar *)(tso->block_info.closure);
3472
3473       last = &mvar->head;
3474       for (t = mvar->head; t != END_TSO_QUEUE;
3475            last = &t->link, last_tso = t, t = t->link) {
3476         if (t == tso) {
3477           *last = tso->link;
3478           if (mvar->tail == tso) {
3479             mvar->tail = last_tso;
3480           }
3481           goto done;
3482         }
3483       }
3484       barf("unblockThread (MVAR): TSO not found");
3485     }
3486
3487   case BlockedOnBlackHole:
3488     {
3489       last = &blackhole_queue;
3490       for (t = blackhole_queue; t != END_TSO_QUEUE;
3491            last = &t->link, t = t->link) {
3492         if (t == tso) {
3493           *last = tso->link;
3494           goto done;
3495         }
3496       }
3497       barf("unblockThread (BLACKHOLE): TSO not found");
3498     }
3499
3500   case BlockedOnException:
3501     {
3502       StgTSO *target  = tso->block_info.tso;
3503
3504       ASSERT(get_itbl(target)->type == TSO);
3505
3506       while (target->what_next == ThreadRelocated) {
3507           target = target->link;
3508           ASSERT(get_itbl(target)->type == TSO);
3509       }
3510
3511       ASSERT(target->blocked_exceptions != NULL);
3512
3513       last = &target->blocked_exceptions;
3514       for (t = target->blocked_exceptions; t != END_TSO_QUEUE;
3515            last = &t->link, t = t->link) {
3516         ASSERT(get_itbl(t)->type == TSO);
3517         if (t == tso) {
3518           *last = tso->link;
3519           goto done;
3520         }
3521       }
3522       barf("unblockThread (Exception): TSO not found");
3523     }
3524
3525   case BlockedOnRead:
3526   case BlockedOnWrite:
3527 #if defined(mingw32_HOST_OS)
3528   case BlockedOnDoProc:
3529 #endif
3530     {
3531       StgTSO *prev = NULL;
3532       for (t = blocked_queue_hd; t != END_TSO_QUEUE;
3533            prev = t, t = t->link) {
3534         if (t == tso) {
3535           if (prev == NULL) {
3536             blocked_queue_hd = t->link;
3537             if (blocked_queue_tl == t) {
3538               blocked_queue_tl = END_TSO_QUEUE;
3539             }
3540           } else {
3541             prev->link = t->link;
3542             if (blocked_queue_tl == t) {
3543               blocked_queue_tl = prev;
3544             }
3545           }
3546 #if defined(mingw32_HOST_OS)
3547           /* (Cooperatively) signal that the worker thread should abort
3548            * the request.
3549            */
3550           abandonWorkRequest(tso->block_info.async_result->reqID);
3551 #endif
3552           goto done;
3553         }
3554       }
3555       barf("unblockThread (I/O): TSO not found");
3556     }
3557
3558   case BlockedOnDelay:
3559     {
3560       StgTSO *prev = NULL;
3561       for (t = sleeping_queue; t != END_TSO_QUEUE;
3562            prev = t, t = t->link) {
3563         if (t == tso) {
3564           if (prev == NULL) {
3565             sleeping_queue = t->link;
3566           } else {
3567             prev->link = t->link;
3568           }
3569           goto done;
3570         }
3571       }
3572       barf("unblockThread (delay): TSO not found");
3573     }
3574
3575   default:
3576     barf("unblockThread");
3577   }
3578
3579  done:
3580   tso->link = END_TSO_QUEUE;
3581   tso->why_blocked = NotBlocked;
3582   tso->block_info.closure = NULL;
3583   APPEND_TO_RUN_QUEUE(tso);
3584 }
3585 #endif
3586
3587 /* -----------------------------------------------------------------------------
3588  * checkBlackHoles()
3589  *
3590  * Check the blackhole_queue for threads that can be woken up.  We do
3591  * this periodically: before every GC, and whenever the run queue is
3592  * empty.
3593  *
3594  * An elegant solution might be to just wake up all the blocked
3595  * threads with awakenBlockedQueue occasionally: they'll go back to
3596  * sleep again if the object is still a BLACKHOLE.  Unfortunately this
3597  * doesn't give us a way to tell whether we've actually managed to
3598  * wake up any threads, so we would be busy-waiting.
3599  *
3600  * -------------------------------------------------------------------------- */
3601
3602 static rtsBool
3603 checkBlackHoles( void )
3604 {
3605     StgTSO **prev, *t;
3606     rtsBool any_woke_up = rtsFalse;
3607     StgHalfWord type;
3608
3609     IF_DEBUG(scheduler, sched_belch("checking threads blocked on black holes"));
3610
3611     // ASSUMES: sched_mutex
3612     prev = &blackhole_queue;
3613     t = blackhole_queue;
3614     while (t != END_TSO_QUEUE) {
3615         ASSERT(t->why_blocked == BlockedOnBlackHole);
3616         type = get_itbl(t->block_info.closure)->type;
3617         if (type != BLACKHOLE && type != CAF_BLACKHOLE) {
3618             t = unblockOneLocked(t);
3619             *prev = t;
3620             any_woke_up = rtsTrue;
3621         } else {
3622             prev = &t->link;
3623             t = t->link;
3624         }
3625     }
3626
3627     return any_woke_up;
3628 }
3629
3630 /* -----------------------------------------------------------------------------
3631  * raiseAsync()
3632  *
3633  * The following function implements the magic for raising an
3634  * asynchronous exception in an existing thread.
3635  *
3636  * We first remove the thread from any queue on which it might be
3637  * blocked.  The possible blockages are MVARs and BLACKHOLE_BQs.
3638  *
3639  * We strip the stack down to the innermost CATCH_FRAME, building
3640  * thunks in the heap for all the active computations, so they can
3641  * be restarted if necessary.  When we reach a CATCH_FRAME, we build
3642  * an application of the handler to the exception, and push it on
3643  * the top of the stack.
3644  *
3645  * How exactly do we save all the active computations?  We create an
3646  * AP_STACK for every UpdateFrame on the stack.  Entering one of these
3647  * AP_STACKs pushes everything from the corresponding update frame
3648  * upwards onto the stack.  (Actually, it pushes everything up to the
3649  * next update frame plus a pointer to the next AP_STACK object.
3650  * Entering the next AP_STACK object pushes more onto the stack until we
3651  * reach the last AP_STACK object - at which point the stack should look
3652  * exactly as it did when we killed the TSO and we can continue
3653  * execution by entering the closure on top of the stack.
3654  *
3655  * We can also kill a thread entirely - this happens if either (a) the
3656  * exception passed to raiseAsync is NULL, or (b) there's no
3657  * CATCH_FRAME on the stack.  In either case, we strip the entire
3658  * stack and replace the thread with a zombie.
3659  *
3660  * Locks: sched_mutex held upon entry nor exit.
3661  *
3662  * -------------------------------------------------------------------------- */
3663
3664 void
3665 deleteThread(StgTSO *tso)
3666 {
3667   if (tso->why_blocked != BlockedOnCCall &&
3668       tso->why_blocked != BlockedOnCCall_NoUnblockExc) {
3669       raiseAsync(tso,NULL);
3670   }
3671 }
3672
3673 #ifdef FORKPROCESS_PRIMOP_SUPPORTED
3674 static void
3675 deleteThreadImmediately(StgTSO *tso)
3676 { // for forkProcess only:
3677   // delete thread without giving it a chance to catch the KillThread exception
3678
3679   if (tso->what_next == ThreadComplete || tso->what_next == ThreadKilled) {
3680       return;
3681   }
3682
3683   if (tso->why_blocked != BlockedOnCCall &&
3684       tso->why_blocked != BlockedOnCCall_NoUnblockExc) {
3685     unblockThread(tso);
3686   }
3687
3688   tso->what_next = ThreadKilled;
3689 }
3690 #endif
3691
3692 void
3693 raiseAsyncWithLock(StgTSO *tso, StgClosure *exception)
3694 {
3695   /* When raising async exs from contexts where sched_mutex isn't held;
3696      use raiseAsyncWithLock(). */
3697   ACQUIRE_LOCK(&sched_mutex);
3698   raiseAsync(tso,exception);
3699   RELEASE_LOCK(&sched_mutex);
3700 }
3701
3702 void
3703 raiseAsync(StgTSO *tso, StgClosure *exception)
3704 {
3705     raiseAsync_(tso, exception, rtsFalse);
3706 }
3707
3708 static void
3709 raiseAsync_(StgTSO *tso, StgClosure *exception, rtsBool stop_at_atomically)
3710 {
3711     StgRetInfoTable *info;
3712     StgPtr sp;
3713
3714     // Thread already dead?
3715     if (tso->what_next == ThreadComplete || tso->what_next == ThreadKilled) {
3716         return;
3717     }
3718
3719     IF_DEBUG(scheduler,
3720              sched_belch("raising exception in thread %ld.", (long)tso->id));
3721
3722     // Remove it from any blocking queues
3723     unblockThread(tso);
3724
3725     sp = tso->sp;
3726
3727     // The stack freezing code assumes there's a closure pointer on
3728     // the top of the stack, so we have to arrange that this is the case...
3729     //
3730     if (sp[0] == (W_)&stg_enter_info) {
3731         sp++;
3732     } else {
3733         sp--;
3734         sp[0] = (W_)&stg_dummy_ret_closure;
3735     }
3736
3737     while (1) {
3738         nat i;
3739
3740         // 1. Let the top of the stack be the "current closure"
3741         //
3742         // 2. Walk up the stack until we find either an UPDATE_FRAME or a
3743         // CATCH_FRAME.
3744         //
3745         // 3. If it's an UPDATE_FRAME, then make an AP_STACK containing the
3746         // current closure applied to the chunk of stack up to (but not
3747         // including) the update frame.  This closure becomes the "current
3748         // closure".  Go back to step 2.
3749         //
3750         // 4. If it's a CATCH_FRAME, then leave the exception handler on
3751         // top of the stack applied to the exception.
3752         //
3753         // 5. If it's a STOP_FRAME, then kill the thread.
3754         //
3755         // NB: if we pass an ATOMICALLY_FRAME then abort the associated
3756         // transaction
3757
3758
3759         StgPtr frame;
3760
3761         frame = sp + 1;
3762         info = get_ret_itbl((StgClosure *)frame);
3763
3764         while (info->i.type != UPDATE_FRAME
3765                && (info->i.type != CATCH_FRAME || exception == NULL)
3766                && info->i.type != STOP_FRAME
3767                && (info->i.type != ATOMICALLY_FRAME || stop_at_atomically == rtsFalse))
3768         {
3769             if (info->i.type == CATCH_RETRY_FRAME || info->i.type == ATOMICALLY_FRAME) {
3770               // IF we find an ATOMICALLY_FRAME then we abort the
3771               // current transaction and propagate the exception.  In
3772               // this case (unlike ordinary exceptions) we do not care
3773               // whether the transaction is valid or not because its
3774               // possible validity cannot have caused the exception
3775               // and will not be visible after the abort.
3776               IF_DEBUG(stm,
3777                        debugBelch("Found atomically block delivering async exception\n"));
3778               stmAbortTransaction(tso -> trec);
3779               tso -> trec = stmGetEnclosingTRec(tso -> trec);
3780             }
3781             frame += stack_frame_sizeW((StgClosure *)frame);
3782             info = get_ret_itbl((StgClosure *)frame);
3783         }
3784
3785         switch (info->i.type) {
3786
3787         case ATOMICALLY_FRAME:
3788             ASSERT(stop_at_atomically);
3789             ASSERT(stmGetEnclosingTRec(tso->trec) == NO_TREC);
3790             stmCondemnTransaction(tso -> trec);
3791 #ifdef REG_R1
3792             tso->sp = frame;
3793 #else
3794             // R1 is not a register: the return convention for IO in
3795             // this case puts the return value on the stack, so we
3796             // need to set up the stack to return to the atomically
3797             // frame properly...
3798             tso->sp = frame - 2;
3799             tso->sp[1] = (StgWord) &stg_NO_FINALIZER_closure; // why not?
3800             tso->sp[0] = (StgWord) &stg_ut_1_0_unreg_info;
3801 #endif
3802             tso->what_next = ThreadRunGHC;
3803             return;
3804
3805         case CATCH_FRAME:
3806             // If we find a CATCH_FRAME, and we've got an exception to raise,
3807             // then build the THUNK raise(exception), and leave it on
3808             // top of the CATCH_FRAME ready to enter.
3809             //
3810         {
3811 #ifdef PROFILING
3812             StgCatchFrame *cf = (StgCatchFrame *)frame;
3813 #endif
3814             StgThunk *raise;
3815
3816             // we've got an exception to raise, so let's pass it to the
3817             // handler in this frame.
3818             //
3819             raise = (StgThunk *)allocate(sizeofW(StgThunk)+1);
3820             TICK_ALLOC_SE_THK(1,0);
3821             SET_HDR(raise,&stg_raise_info,cf->header.prof.ccs);
3822             raise->payload[0] = exception;
3823
3824             // throw away the stack from Sp up to the CATCH_FRAME.
3825             //
3826             sp = frame - 1;
3827
3828             /* Ensure that async excpetions are blocked now, so we don't get
3829              * a surprise exception before we get around to executing the
3830              * handler.
3831              */
3832             if (tso->blocked_exceptions == NULL) {
3833                 tso->blocked_exceptions = END_TSO_QUEUE;
3834             }
3835
3836             /* Put the newly-built THUNK on top of the stack, ready to execute
3837              * when the thread restarts.
3838              */
3839             sp[0] = (W_)raise;
3840             sp[-1] = (W_)&stg_enter_info;
3841             tso->sp = sp-1;
3842             tso->what_next = ThreadRunGHC;
3843             IF_DEBUG(sanity, checkTSO(tso));
3844             return;
3845         }
3846
3847         case UPDATE_FRAME:
3848         {
3849             StgAP_STACK * ap;
3850             nat words;
3851
3852             // First build an AP_STACK consisting of the stack chunk above the
3853             // current update frame, with the top word on the stack as the
3854             // fun field.
3855             //
3856             words = frame - sp - 1;
3857             ap = (StgAP_STACK *)allocate(AP_STACK_sizeW(words));
3858
3859             ap->size = words;
3860             ap->fun  = (StgClosure *)sp[0];
3861             sp++;
3862             for(i=0; i < (nat)words; ++i) {
3863                 ap->payload[i] = (StgClosure *)*sp++;
3864             }
3865
3866             SET_HDR(ap,&stg_AP_STACK_info,
3867                     ((StgClosure *)frame)->header.prof.ccs /* ToDo */);
3868             TICK_ALLOC_UP_THK(words+1,0);
3869
3870             IF_DEBUG(scheduler,
3871                      debugBelch("sched: Updating ");
3872                      printPtr((P_)((StgUpdateFrame *)frame)->updatee);
3873                      debugBelch(" with ");
3874                      printObj((StgClosure *)ap);
3875                 );
3876
3877             // Replace the updatee with an indirection - happily
3878             // this will also wake up any threads currently
3879             // waiting on the result.
3880             //
3881             // Warning: if we're in a loop, more than one update frame on
3882             // the stack may point to the same object.  Be careful not to
3883             // overwrite an IND_OLDGEN in this case, because we'll screw
3884             // up the mutable lists.  To be on the safe side, don't
3885             // overwrite any kind of indirection at all.  See also
3886             // threadSqueezeStack in GC.c, where we have to make a similar
3887             // check.
3888             //
3889             if (!closure_IND(((StgUpdateFrame *)frame)->updatee)) {
3890                 // revert the black hole
3891                 UPD_IND_NOLOCK(((StgUpdateFrame *)frame)->updatee,
3892                                (StgClosure *)ap);
3893             }
3894             sp += sizeofW(StgUpdateFrame) - 1;
3895             sp[0] = (W_)ap; // push onto stack
3896             break;
3897         }
3898
3899         case STOP_FRAME:
3900             // We've stripped the entire stack, the thread is now dead.
3901             sp += sizeofW(StgStopFrame);
3902             tso->what_next = ThreadKilled;
3903             tso->sp = sp;
3904             return;
3905
3906         default:
3907             barf("raiseAsync");
3908         }
3909     }
3910     barf("raiseAsync");
3911 }
3912
3913 /* -----------------------------------------------------------------------------
3914    raiseExceptionHelper
3915
3916    This function is called by the raise# primitve, just so that we can
3917    move some of the tricky bits of raising an exception from C-- into
3918    C.  Who knows, it might be a useful re-useable thing here too.
3919    -------------------------------------------------------------------------- */
3920
3921 StgWord
3922 raiseExceptionHelper (StgTSO *tso, StgClosure *exception)
3923 {
3924     StgThunk *raise_closure = NULL;
3925     StgPtr p, next;
3926     StgRetInfoTable *info;
3927     //
3928     // This closure represents the expression 'raise# E' where E
3929     // is the exception raise.  It is used to overwrite all the
3930     // thunks which are currently under evaluataion.
3931     //
3932
3933     //
3934     // LDV profiling: stg_raise_info has THUNK as its closure
3935     // type. Since a THUNK takes at least MIN_UPD_SIZE words in its
3936     // payload, MIN_UPD_SIZE is more approprate than 1.  It seems that
3937     // 1 does not cause any problem unless profiling is performed.
3938     // However, when LDV profiling goes on, we need to linearly scan
3939     // small object pool, where raise_closure is stored, so we should
3940     // use MIN_UPD_SIZE.
3941     //
3942     // raise_closure = (StgClosure *)RET_STGCALL1(P_,allocate,
3943     //                                 sizeofW(StgClosure)+1);
3944     //
3945
3946     //
3947     // Walk up the stack, looking for the catch frame.  On the way,
3948     // we update any closures pointed to from update frames with the
3949     // raise closure that we just built.
3950     //
3951     p = tso->sp;
3952     while(1) {
3953         info = get_ret_itbl((StgClosure *)p);
3954         next = p + stack_frame_sizeW((StgClosure *)p);
3955         switch (info->i.type) {
3956
3957         case UPDATE_FRAME:
3958             // Only create raise_closure if we need to.
3959             if (raise_closure == NULL) {
3960                 raise_closure =
3961                     (StgThunk *)allocate(sizeofW(StgThunk)+MIN_UPD_SIZE);
3962                 SET_HDR(raise_closure, &stg_raise_info, CCCS);
3963                 raise_closure->payload[0] = exception;
3964             }
3965             UPD_IND(((StgUpdateFrame *)p)->updatee,(StgClosure *)raise_closure);
3966             p = next;
3967             continue;
3968
3969         case ATOMICALLY_FRAME:
3970             IF_DEBUG(stm, debugBelch("Found ATOMICALLY_FRAME at %p\n", p));
3971             tso->sp = p;
3972             return ATOMICALLY_FRAME;
3973
3974         case CATCH_FRAME:
3975             tso->sp = p;
3976             return CATCH_FRAME;
3977
3978         case CATCH_STM_FRAME:
3979             IF_DEBUG(stm, debugBelch("Found CATCH_STM_FRAME at %p\n", p));
3980             tso->sp = p;
3981             return CATCH_STM_FRAME;
3982
3983         case STOP_FRAME:
3984             tso->sp = p;
3985             return STOP_FRAME;
3986
3987         case CATCH_RETRY_FRAME:
3988         default:
3989             p = next;
3990             continue;
3991         }
3992     }
3993 }
3994
3995
3996 /* -----------------------------------------------------------------------------
3997    findRetryFrameHelper
3998
3999    This function is called by the retry# primitive.  It traverses the stack
4000    leaving tso->sp referring to the frame which should handle the retry.
4001
4002    This should either be a CATCH_RETRY_FRAME (if the retry# is within an orElse#)
4003    or should be a ATOMICALLY_FRAME (if the retry# reaches the top level).
4004
4005    We skip CATCH_STM_FRAMEs because retries are not considered to be exceptions,
4006    despite the similar implementation.
4007
4008    We should not expect to see CATCH_FRAME or STOP_FRAME because those should
4009    not be created within memory transactions.
4010    -------------------------------------------------------------------------- */
4011
4012 StgWord
4013 findRetryFrameHelper (StgTSO *tso)
4014 {
4015   StgPtr           p, next;
4016   StgRetInfoTable *info;
4017
4018   p = tso -> sp;
4019   while (1) {
4020     info = get_ret_itbl((StgClosure *)p);
4021     next = p + stack_frame_sizeW((StgClosure *)p);
4022     switch (info->i.type) {
4023
4024     case ATOMICALLY_FRAME:
4025       IF_DEBUG(stm, debugBelch("Found ATOMICALLY_FRAME at %p during retrry\n", p));
4026       tso->sp = p;
4027       return ATOMICALLY_FRAME;
4028
4029     case CATCH_RETRY_FRAME:
4030       IF_DEBUG(stm, debugBelch("Found CATCH_RETRY_FRAME at %p during retrry\n", p));
4031       tso->sp = p;
4032       return CATCH_RETRY_FRAME;
4033
4034     case CATCH_STM_FRAME:
4035     default:
4036       ASSERT(info->i.type != CATCH_FRAME);
4037       ASSERT(info->i.type != STOP_FRAME);
4038       p = next;
4039       continue;
4040     }
4041   }
4042 }
4043
4044 /* -----------------------------------------------------------------------------
4045    resurrectThreads is called after garbage collection on the list of
4046    threads found to be garbage.  Each of these threads will be woken
4047    up and sent a signal: BlockedOnDeadMVar if the thread was blocked
4048    on an MVar, or NonTermination if the thread was blocked on a Black
4049    Hole.
4050
4051    Locks: sched_mutex isn't held upon entry nor exit.
4052    -------------------------------------------------------------------------- */
4053
4054 void
4055 resurrectThreads( StgTSO *threads )
4056 {
4057   StgTSO *tso, *next;
4058
4059   for (tso = threads; tso != END_TSO_QUEUE; tso = next) {
4060     next = tso->global_link;
4061     tso->global_link = all_threads;
4062     all_threads = tso;
4063     IF_DEBUG(scheduler, sched_belch("resurrecting thread %d", tso->id));
4064
4065     switch (tso->why_blocked) {
4066     case BlockedOnMVar:
4067     case BlockedOnException:
4068       /* Called by GC - sched_mutex lock is currently held. */
4069       raiseAsync(tso,(StgClosure *)BlockedOnDeadMVar_closure);
4070       break;
4071     case BlockedOnBlackHole:
4072       raiseAsync(tso,(StgClosure *)NonTermination_closure);
4073       break;
4074     case BlockedOnSTM:
4075       raiseAsync(tso,(StgClosure *)BlockedIndefinitely_closure);
4076       break;
4077     case NotBlocked:
4078       /* This might happen if the thread was blocked on a black hole
4079        * belonging to a thread that we've just woken up (raiseAsync
4080        * can wake up threads, remember...).
4081        */
4082       continue;
4083     default:
4084       barf("resurrectThreads: thread blocked in a strange way");
4085     }
4086   }
4087 }
4088
4089 /* ----------------------------------------------------------------------------
4090  * Debugging: why is a thread blocked
4091  * [Also provides useful information when debugging threaded programs
4092  *  at the Haskell source code level, so enable outside of DEBUG. --sof 7/02]
4093    ------------------------------------------------------------------------- */
4094
4095 static void
4096 printThreadBlockage(StgTSO *tso)
4097 {
4098   switch (tso->why_blocked) {
4099   case BlockedOnRead:
4100     debugBelch("is blocked on read from fd %d", (int)(tso->block_info.fd));
4101     break;
4102   case BlockedOnWrite:
4103     debugBelch("is blocked on write to fd %d", (int)(tso->block_info.fd));
4104     break;
4105 #if defined(mingw32_HOST_OS)
4106     case BlockedOnDoProc:
4107     debugBelch("is blocked on proc (request: %ld)", tso->block_info.async_result->reqID);
4108     break;
4109 #endif
4110   case BlockedOnDelay:
4111     debugBelch("is blocked until %ld", (long)(tso->block_info.target));
4112     break;
4113   case BlockedOnMVar:
4114     debugBelch("is blocked on an MVar @ %p", tso->block_info.closure);
4115     break;
4116   case BlockedOnException:
4117     debugBelch("is blocked on delivering an exception to thread %d",
4118             tso->block_info.tso->id);
4119     break;
4120   case BlockedOnBlackHole:
4121     debugBelch("is blocked on a black hole");
4122     break;
4123   case NotBlocked:
4124     debugBelch("is not blocked");
4125     break;
4126 #if defined(PARALLEL_HASKELL)
4127   case BlockedOnGA:
4128     debugBelch("is blocked on global address; local FM_BQ is %p (%s)",
4129             tso->block_info.closure, info_type(tso->block_info.closure));
4130     break;
4131   case BlockedOnGA_NoSend:
4132     debugBelch("is blocked on global address (no send); local FM_BQ is %p (%s)",
4133             tso->block_info.closure, info_type(tso->block_info.closure));
4134     break;
4135 #endif
4136   case BlockedOnCCall:
4137     debugBelch("is blocked on an external call");
4138     break;
4139   case BlockedOnCCall_NoUnblockExc:
4140     debugBelch("is blocked on an external call (exceptions were already blocked)");
4141     break;
4142   case BlockedOnSTM:
4143     debugBelch("is blocked on an STM operation");
4144     break;
4145   default:
4146     barf("printThreadBlockage: strange tso->why_blocked: %d for TSO %d (%d)",
4147          tso->why_blocked, tso->id, tso);
4148   }
4149 }
4150
4151 static void
4152 printThreadStatus(StgTSO *tso)
4153 {
4154   switch (tso->what_next) {
4155   case ThreadKilled:
4156     debugBelch("has been killed");
4157     break;
4158   case ThreadComplete:
4159     debugBelch("has completed");
4160     break;
4161   default:
4162     printThreadBlockage(tso);
4163   }
4164 }
4165
4166 void
4167 printAllThreads(void)
4168 {
4169   StgTSO *t;
4170
4171 # if defined(GRAN)
4172   char time_string[TIME_STR_LEN], node_str[NODE_STR_LEN];
4173   ullong_format_string(TIME_ON_PROC(CurrentProc),
4174                        time_string, rtsFalse/*no commas!*/);
4175
4176   debugBelch("all threads at [%s]:\n", time_string);
4177 # elif defined(PARALLEL_HASKELL)
4178   char time_string[TIME_STR_LEN], node_str[NODE_STR_LEN];
4179   ullong_format_string(CURRENT_TIME,
4180                        time_string, rtsFalse/*no commas!*/);
4181
4182   debugBelch("all threads at [%s]:\n", time_string);
4183 # else
4184   debugBelch("all threads:\n");
4185 # endif
4186
4187   for (t = all_threads; t != END_TSO_QUEUE; ) {
4188     debugBelch("\tthread %4d @ %p ", t->id, (void *)t);
4189 #if defined(DEBUG)
4190     {
4191       void *label = lookupThreadLabel(t->id);
4192       if (label) debugBelch("[\"%s\"] ",(char *)label);
4193     }
4194 #endif
4195     if (t->what_next == ThreadRelocated) {
4196         debugBelch("has been relocated...\n");
4197         t = t->link;
4198     } else {
4199         printThreadStatus(t);
4200         debugBelch("\n");
4201         t = t->global_link;
4202     }
4203   }
4204 }
4205
4206 #ifdef DEBUG
4207
4208 // useful from gdb
4209 void
4210 printThreadQueue(StgTSO *t)
4211 {
4212     nat i = 0;
4213     for (; t != END_TSO_QUEUE; t = t->link) {
4214         debugBelch("\tthread %d @ %p ", t->id, (void *)t);
4215         if (t->what_next == ThreadRelocated) {
4216             debugBelch("has been relocated...\n");
4217         } else {
4218             printThreadStatus(t);
4219             debugBelch("\n");
4220         }
4221         i++;
4222     }
4223     debugBelch("%d threads on queue\n", i);
4224 }
4225
4226 /*
4227    Print a whole blocking queue attached to node (debugging only).
4228 */
4229 # if defined(PARALLEL_HASKELL)
4230 void
4231 print_bq (StgClosure *node)
4232 {
4233   StgBlockingQueueElement *bqe;
4234   StgTSO *tso;
4235   rtsBool end;
4236
4237   debugBelch("## BQ of closure %p (%s): ",
4238           node, info_type(node));
4239
4240   /* should cover all closures that may have a blocking queue */
4241   ASSERT(get_itbl(node)->type == BLACKHOLE_BQ ||
4242          get_itbl(node)->type == FETCH_ME_BQ ||
4243          get_itbl(node)->type == RBH ||
4244          get_itbl(node)->type == MVAR);
4245
4246   ASSERT(node!=(StgClosure*)NULL);         // sanity check
4247
4248   print_bqe(((StgBlockingQueue*)node)->blocking_queue);
4249 }
4250
4251 /*
4252    Print a whole blocking queue starting with the element bqe.
4253 */
4254 void
4255 print_bqe (StgBlockingQueueElement *bqe)
4256 {
4257   rtsBool end;
4258
4259   /*
4260      NB: In a parallel setup a BQ of an RBH must end with an RBH_Save closure;
4261   */
4262   for (end = (bqe==END_BQ_QUEUE);
4263        !end; // iterate until bqe points to a CONSTR
4264        end = (get_itbl(bqe)->type == CONSTR) || (bqe->link==END_BQ_QUEUE),
4265        bqe = end ? END_BQ_QUEUE : bqe->link) {
4266     ASSERT(bqe != END_BQ_QUEUE);                               // sanity check
4267     ASSERT(bqe != (StgBlockingQueueElement *)NULL);            // sanity check
4268     /* types of closures that may appear in a blocking queue */
4269     ASSERT(get_itbl(bqe)->type == TSO ||
4270            get_itbl(bqe)->type == BLOCKED_FETCH ||
4271            get_itbl(bqe)->type == CONSTR);
4272     /* only BQs of an RBH end with an RBH_Save closure */
4273     //ASSERT(get_itbl(bqe)->type != CONSTR || get_itbl(node)->type == RBH);
4274
4275     switch (get_itbl(bqe)->type) {
4276     case TSO:
4277       debugBelch(" TSO %u (%x),",
4278               ((StgTSO *)bqe)->id, ((StgTSO *)bqe));
4279       break;
4280     case BLOCKED_FETCH:
4281       debugBelch(" BF (node=%p, ga=((%x, %d, %x)),",
4282               ((StgBlockedFetch *)bqe)->node,
4283               ((StgBlockedFetch *)bqe)->ga.payload.gc.gtid,
4284               ((StgBlockedFetch *)bqe)->ga.payload.gc.slot,
4285               ((StgBlockedFetch *)bqe)->ga.weight);
4286       break;
4287     case CONSTR:
4288       debugBelch(" %s (IP %p),",
4289               (get_itbl(bqe) == &stg_RBH_Save_0_info ? "RBH_Save_0" :
4290                get_itbl(bqe) == &stg_RBH_Save_1_info ? "RBH_Save_1" :
4291                get_itbl(bqe) == &stg_RBH_Save_2_info ? "RBH_Save_2" :
4292                "RBH_Save_?"), get_itbl(bqe));
4293       break;
4294     default:
4295       barf("Unexpected closure type %s in blocking queue", // of %p (%s)",
4296            info_type((StgClosure *)bqe)); // , node, info_type(node));
4297       break;
4298     }
4299   } /* for */
4300   debugBelch("\n");
4301 }
4302 # elif defined(GRAN)
4303 void
4304 print_bq (StgClosure *node)
4305 {
4306   StgBlockingQueueElement *bqe;
4307   PEs node_loc, tso_loc;
4308   rtsBool end;
4309
4310   /* should cover all closures that may have a blocking queue */
4311   ASSERT(get_itbl(node)->type == BLACKHOLE_BQ ||
4312          get_itbl(node)->type == FETCH_ME_BQ ||
4313          get_itbl(node)->type == RBH);
4314
4315   ASSERT(node!=(StgClosure*)NULL);         // sanity check
4316   node_loc = where_is(node);
4317
4318   debugBelch("## BQ of closure %p (%s) on [PE %d]: ",
4319           node, info_type(node), node_loc);
4320
4321   /*
4322      NB: In a parallel setup a BQ of an RBH must end with an RBH_Save closure;
4323   */
4324   for (bqe = ((StgBlockingQueue*)node)->blocking_queue, end = (bqe==END_BQ_QUEUE);
4325        !end; // iterate until bqe points to a CONSTR
4326        end = (get_itbl(bqe)->type == CONSTR) || (bqe->link==END_BQ_QUEUE), bqe = end ? END_BQ_QUEUE : bqe->link) {
4327     ASSERT(bqe != END_BQ_QUEUE);             // sanity check
4328     ASSERT(bqe != (StgBlockingQueueElement *)NULL);  // sanity check
4329     /* types of closures that may appear in a blocking queue */
4330     ASSERT(get_itbl(bqe)->type == TSO ||
4331            get_itbl(bqe)->type == CONSTR);
4332     /* only BQs of an RBH end with an RBH_Save closure */
4333     ASSERT(get_itbl(bqe)->type != CONSTR || get_itbl(node)->type == RBH);
4334
4335     tso_loc = where_is((StgClosure *)bqe);
4336     switch (get_itbl(bqe)->type) {
4337     case TSO:
4338       debugBelch(" TSO %d (%p) on [PE %d],",
4339               ((StgTSO *)bqe)->id, (StgTSO *)bqe, tso_loc);
4340       break;
4341     case CONSTR:
4342       debugBelch(" %s (IP %p),",
4343               (get_itbl(bqe) == &stg_RBH_Save_0_info ? "RBH_Save_0" :
4344                get_itbl(bqe) == &stg_RBH_Save_1_info ? "RBH_Save_1" :
4345                get_itbl(bqe) == &stg_RBH_Save_2_info ? "RBH_Save_2" :
4346                "RBH_Save_?"), get_itbl(bqe));
4347       break;
4348     default:
4349       barf("Unexpected closure type %s in blocking queue of %p (%s)",
4350            info_type((StgClosure *)bqe), node, info_type(node));
4351       break;
4352     }
4353   } /* for */
4354   debugBelch("\n");
4355 }
4356 # endif
4357
4358 #if defined(PARALLEL_HASKELL)
4359 static nat
4360 run_queue_len(void)
4361 {
4362   nat i;
4363   StgTSO *tso;
4364
4365   for (i=0, tso=run_queue_hd;
4366        tso != END_TSO_QUEUE;
4367        i++, tso=tso->link)
4368     /* nothing */
4369
4370   return i;
4371 }
4372 #endif
4373
4374 void
4375 sched_belch(char *s, ...)
4376 {
4377   va_list ap;
4378   va_start(ap,s);
4379 #ifdef RTS_SUPPORTS_THREADS
4380   debugBelch("sched (task %p): ", osThreadId());
4381 #elif defined(PARALLEL_HASKELL)
4382   debugBelch("== ");
4383 #else
4384   debugBelch("sched: ");
4385 #endif
4386   vdebugBelch(s, ap);
4387   debugBelch("\n");
4388   va_end(ap);
4389 }
4390
4391 #endif /* DEBUG */