rts/Schedule.c

   1 /* ---------------------------------------------------------------------------
   2  *
   3  * (c) The GHC Team, 1998-2006
   4  *
   5  * The scheduler and thread-related functionality
   6  *
   7  * --------------------------------------------------------------------------*/
   8
   9 #include "PosixSource.h"
  10 #define KEEP_LOCKCLOSURE
  11 #include "Rts.h"
  12 #include "SchedAPI.h"
  13 #include "RtsUtils.h"
  14 #include "RtsFlags.h"
  15 #include "OSThreads.h"
  16 #include "Storage.h"
  17 #include "StgRun.h"
  18 #include "Hooks.h"
  19 #include "Schedule.h"
  20 #include "StgMiscClosures.h"
  21 #include "Interpreter.h"
  22 #include "Printer.h"
  23 #include "RtsSignals.h"
  24 #include "Sanity.h"
  25 #include "Stats.h"
  26 #include "STM.h"
  27 #include "Timer.h"
  28 #include "Prelude.h"
  29 #include "ThreadLabels.h"
  30 #include "LdvProfile.h"
  31 #include "Updates.h"
  32 #include "Proftimer.h"
  33 #include "ProfHeap.h"
  34 #include "GC.h"
  35 #include "Weak.h"
  36 #include "EventLog.h"
  37
  38 /* PARALLEL_HASKELL includes go here */
  39
  40 #include "Sparks.h"
  41 #include "Capability.h"
  42 #include "Task.h"
  43 #include "AwaitEvent.h"
  44 #if defined(mingw32_HOST_OS)
  45 #include "win32/IOManager.h"
  46 #endif
  47 #include "Trace.h"
  48 #include "RaiseAsync.h"
  49 #include "Threads.h"
  50 #include "ThrIOManager.h"
  51
  52 #ifdef HAVE_SYS_TYPES_H
  53 #include <sys/types.h>
  54 #endif
  55 #ifdef HAVE_UNISTD_H
  56 #include <unistd.h>
  57 #endif
  58
  59 #include <string.h>
  60 #include <stdlib.h>
  61 #include <stdarg.h>
  62
  63 #ifdef HAVE_ERRNO_H
  64 #include <errno.h>
  65 #endif
  66
  67 // Turn off inlining when debugging - it obfuscates things
  68 #ifdef DEBUG
  69 # undef  STATIC_INLINE
  70 # define STATIC_INLINE static
  71 #endif
  72
  73 /* -----------------------------------------------------------------------------
  74  * Global variables
  75  * -------------------------------------------------------------------------- */
  76
  77 #if !defined(THREADED_RTS)
  78 // Blocked/sleeping thrads
  79 StgTSO *blocked_queue_hd = NULL;
  80 StgTSO *blocked_queue_tl = NULL;
  81 StgTSO *sleeping_queue = NULL;    // perhaps replace with a hash table?
  82 #endif
  83
  84 /* Threads blocked on blackholes.
  85  * LOCK: sched_mutex+capability, or all capabilities
  86  */
  87 StgTSO *blackhole_queue = NULL;
  88
  89 /* The blackhole_queue should be checked for threads to wake up.  See
  90  * Schedule.h for more thorough comment.
  91  * LOCK: none (doesn't matter if we miss an update)
  92  */
  93 rtsBool blackholes_need_checking = rtsFalse;
  94
  95 /* Set to true when the latest garbage collection failed to reclaim
  96  * enough space, and the runtime should proceed to shut itself down in
  97  * an orderly fashion (emitting profiling info etc.)
  98  */
  99 rtsBool heap_overflow = rtsFalse;
 100
 101 /* flag that tracks whether we have done any execution in this time slice.
 102  * LOCK: currently none, perhaps we should lock (but needs to be
 103  * updated in the fast path of the scheduler).
 104  *
 105  * NB. must be StgWord, we do xchg() on it.
 106  */
 107 volatile StgWord recent_activity = ACTIVITY_YES;
 108
 109 /* if this flag is set as well, give up execution
 110  * LOCK: none (changes monotonically)
 111  */
 112 volatile StgWord sched_state = SCHED_RUNNING;
 113
 114 /*  This is used in `TSO.h' and gcc 2.96 insists that this variable actually
 115  *  exists - earlier gccs apparently didn't.
 116  *  -= chak
 117  */
 118 StgTSO dummy_tso;
 119
 120 /*
 121  * Set to TRUE when entering a shutdown state (via shutdownHaskellAndExit()) --
 122  * in an MT setting, needed to signal that a worker thread shouldn't hang around
 123  * in the scheduler when it is out of work.
 124  */
 125 rtsBool shutting_down_scheduler = rtsFalse;
 126
 127 /*
 128  * This mutex protects most of the global scheduler data in
 129  * the THREADED_RTS runtime.
 130  */
 131 #if defined(THREADED_RTS)
 132 Mutex sched_mutex;
 133 #endif
 134
 135 #if !defined(mingw32_HOST_OS)
 136 #define FORKPROCESS_PRIMOP_SUPPORTED
 137 #endif
 138
 139 /* -----------------------------------------------------------------------------
 140  * static function prototypes
 141  * -------------------------------------------------------------------------- */
 142
 143 static Capability *schedule (Capability *initialCapability, Task *task);
 144
 145 //
 146 // These function all encapsulate parts of the scheduler loop, and are
 147 // abstracted only to make the structure and control flow of the
 148 // scheduler clearer.
 149 //
 150 static void schedulePreLoop (void);
 151 static void scheduleFindWork (Capability *cap);
 152 #if defined(THREADED_RTS)
 153 static void scheduleYield (Capability **pcap, Task *task);
 154 #endif
 155 static void scheduleStartSignalHandlers (Capability *cap);
 156 static void scheduleCheckBlockedThreads (Capability *cap);
 157 static void scheduleCheckWakeupThreads(Capability *cap USED_IF_NOT_THREADS);
 158 static void scheduleCheckBlackHoles (Capability *cap);
 159 static void scheduleDetectDeadlock (Capability *cap, Task *task);
 160 static void schedulePushWork(Capability *cap, Task *task);
 161 #if defined(THREADED_RTS)
 162 static void scheduleActivateSpark(Capability *cap);
 163 #endif
 164 static void schedulePostRunThread(Capability *cap, StgTSO *t);
 165 static rtsBool scheduleHandleHeapOverflow( Capability *cap, StgTSO *t );
 166 static void scheduleHandleStackOverflow( Capability *cap, Task *task,
 167                                          StgTSO *t);
 168 static rtsBool scheduleHandleYield( Capability *cap, StgTSO *t,
 169                                     nat prev_what_next );
 170 static void scheduleHandleThreadBlocked( StgTSO *t );
 171 static rtsBool scheduleHandleThreadFinished( Capability *cap, Task *task,
 172                                              StgTSO *t );
 173 static rtsBool scheduleNeedHeapProfile(rtsBool ready_to_gc);
 174 static Capability *scheduleDoGC(Capability *cap, Task *task,
 175                                 rtsBool force_major);
 176
 177 static rtsBool checkBlackHoles(Capability *cap);
 178
 179 static StgTSO *threadStackOverflow(Capability *cap, StgTSO *tso);
 180 static StgTSO *threadStackUnderflow(Task *task, StgTSO *tso);
 181
 182 static void deleteThread (Capability *cap, StgTSO *tso);
 183 static void deleteAllThreads (Capability *cap);
 184
 185 #ifdef FORKPROCESS_PRIMOP_SUPPORTED
 186 static void deleteThread_(Capability *cap, StgTSO *tso);
 187 #endif
 188
 189 #ifdef DEBUG
 190 static char *whatNext_strs[] = {
 191   "(unknown)",
 192   "ThreadRunGHC",
 193   "ThreadInterpret",
 194   "ThreadKilled",
 195   "ThreadRelocated",
 196   "ThreadComplete"
 197 };
 198 #endif
 199
 200 /* -----------------------------------------------------------------------------
 201  * Putting a thread on the run queue: different scheduling policies
 202  * -------------------------------------------------------------------------- */
 203
 204 STATIC_INLINE void
 205 addToRunQueue( Capability *cap, StgTSO *t )
 206 {
 207     // this does round-robin scheduling; good for concurrency
 208     appendToRunQueue(cap,t);
 209 }
 210
 211 /* ---------------------------------------------------------------------------
 212    Main scheduling loop.
 213
 214    We use round-robin scheduling, each thread returning to the
 215    scheduler loop when one of these conditions is detected:
 216
 217       * out of heap space
 218       * timer expires (thread yields)
 219       * thread blocks
 220       * thread ends
 221       * stack overflow
 222
 223    GRAN version:
 224      In a GranSim setup this loop iterates over the global event queue.
 225      This revolves around the global event queue, which determines what
 226      to do next. Therefore, it's more complicated than either the
 227      concurrent or the parallel (GUM) setup.
 228   This version has been entirely removed (JB 2008/08).
 229
 230    GUM version:
 231      GUM iterates over incoming messages.
 232      It starts with nothing to do (thus CurrentTSO == END_TSO_QUEUE),
 233      and sends out a fish whenever it has nothing to do; in-between
 234      doing the actual reductions (shared code below) it processes the
 235      incoming messages and deals with delayed operations
 236      (see PendingFetches).
 237      This is not the ugliest code you could imagine, but it's bloody close.
 238
 239   (JB 2008/08) This version was formerly indicated by a PP-Flag PAR,
 240   now by PP-flag PARALLEL_HASKELL. The Eden RTS (in GHC-6.x) uses it,
 241   as well as future GUM versions. This file has been refurbished to
 242   only contain valid code, which is however incomplete, refers to
 243   invalid includes etc.
 244
 245    ------------------------------------------------------------------------ */
 246
 247 static Capability *
 248 schedule (Capability *initialCapability, Task *task)
 249 {
 250   StgTSO *t;
 251   Capability *cap;
 252   StgThreadReturnCode ret;
 253   nat prev_what_next;
 254   rtsBool ready_to_gc;
 255 #if defined(THREADED_RTS)
 256   rtsBool first = rtsTrue;
 257 #endif
 258
 259   cap = initialCapability;
 260
 261   // Pre-condition: this task owns initialCapability.
 262   // The sched_mutex is *NOT* held
 263   // NB. on return, we still hold a capability.
 264
 265   debugTrace (DEBUG_sched,
 266               "### NEW SCHEDULER LOOP (task: %p, cap: %p)",
 267               task, initialCapability);
 268
 269   if (running_finalizers) {
 270       errorBelch("error: a C finalizer called back into Haskell.\n"
 271                  "   This was previously allowed, but is disallowed in GHC 6.10.2 and later.\n"
 272                  "   To create finalizers that may call back into Haskll, use\n"
 273                  "   Foreign.Concurrent.newForeignPtr instead of Foreign.newForeignPtr.");
 274       stg_exit(EXIT_FAILURE);
 275   }
 276
 277   schedulePreLoop();
 278
 279   // -----------------------------------------------------------
 280   // Scheduler loop starts here:
 281
 282   while (1) {
 283
 284     // Check whether we have re-entered the RTS from Haskell without
 285     // going via suspendThread()/resumeThread (i.e. a 'safe' foreign
 286     // call).
 287     if (cap->in_haskell) {
 288           errorBelch("schedule: re-entered unsafely.\n"
 289                      "   Perhaps a 'foreign import unsafe' should be 'safe'?");
 290           stg_exit(EXIT_FAILURE);
 291     }
 292
 293     // The interruption / shutdown sequence.
 294     //
 295     // In order to cleanly shut down the runtime, we want to:
 296     //   * make sure that all main threads return to their callers
 297     //     with the state 'Interrupted'.
 298     //   * clean up all OS threads assocated with the runtime
 299     //   * free all memory etc.
 300     //
 301     // So the sequence for ^C goes like this:
 302     //
 303     //   * ^C handler sets sched_state := SCHED_INTERRUPTING and
 304     //     arranges for some Capability to wake up
 305     //
 306     //   * all threads in the system are halted, and the zombies are
 307     //     placed on the run queue for cleaning up.  We acquire all
 308     //     the capabilities in order to delete the threads, this is
 309     //     done by scheduleDoGC() for convenience (because GC already
 310     //     needs to acquire all the capabilities).  We can't kill
 311     //     threads involved in foreign calls.
 312     //
 313     //   * somebody calls shutdownHaskell(), which calls exitScheduler()
 314     //
 315     //   * sched_state := SCHED_SHUTTING_DOWN
 316     //
 317     //   * all workers exit when the run queue on their capability
 318     //     drains.  All main threads will also exit when their TSO
 319     //     reaches the head of the run queue and they can return.
 320     //
 321     //   * eventually all Capabilities will shut down, and the RTS can
 322     //     exit.
 323     //
 324     //   * We might be left with threads blocked in foreign calls,
 325     //     we should really attempt to kill these somehow (TODO);
 326
 327     switch (sched_state) {
 328     case SCHED_RUNNING:
 329         break;
 330     case SCHED_INTERRUPTING:
 331         debugTrace(DEBUG_sched, "SCHED_INTERRUPTING");
 332 #if defined(THREADED_RTS)
 333         discardSparksCap(cap);
 334 #endif
 335         /* scheduleDoGC() deletes all the threads */
 336         cap = scheduleDoGC(cap,task,rtsFalse);
 337
 338         // after scheduleDoGC(), we must be shutting down.  Either some
 339         // other Capability did the final GC, or we did it above,
 340         // either way we can fall through to the SCHED_SHUTTING_DOWN
 341         // case now.
 342         ASSERT(sched_state == SCHED_SHUTTING_DOWN);
 343         // fall through
 344
 345     case SCHED_SHUTTING_DOWN:
 346         debugTrace(DEBUG_sched, "SCHED_SHUTTING_DOWN");
 347         // If we are a worker, just exit.  If we're a bound thread
 348         // then we will exit below when we've removed our TSO from
 349         // the run queue.
 350         if (task->tso == NULL && emptyRunQueue(cap)) {
 351             return cap;
 352         }
 353         break;
 354     default:
 355         barf("sched_state: %d", sched_state);
 356     }
 357
 358     scheduleFindWork(cap);
 359
 360     /* work pushing, currently relevant only for THREADED_RTS:
 361        (pushes threads, wakes up idle capabilities for stealing) */
 362     schedulePushWork(cap,task);
 363
 364     scheduleDetectDeadlock(cap,task);
 365
 366 #if defined(THREADED_RTS)
 367     cap = task->cap;    // reload cap, it might have changed
 368 #endif
 369
 370     // Normally, the only way we can get here with no threads to
 371     // run is if a keyboard interrupt received during
 372     // scheduleCheckBlockedThreads() or scheduleDetectDeadlock().
 373     // Additionally, it is not fatal for the
 374     // threaded RTS to reach here with no threads to run.
 375     //
 376     // win32: might be here due to awaitEvent() being abandoned
 377     // as a result of a console event having been delivered.
 378
 379 #if defined(THREADED_RTS)
 380     if (first)
 381     {
 382     // XXX: ToDo
 383     //     // don't yield the first time, we want a chance to run this
 384     //     // thread for a bit, even if there are others banging at the
 385     //     // door.
 386     //     first = rtsFalse;
 387     //     ASSERT_FULL_CAPABILITY_INVARIANTS(cap,task);
 388     }
 389
 390   yield:
 391     scheduleYield(&cap,task);
 392     if (emptyRunQueue(cap)) continue; // look for work again
 393 #endif
 394
 395 #if !defined(THREADED_RTS) && !defined(mingw32_HOST_OS)
 396     if ( emptyRunQueue(cap) ) {
 397         ASSERT(sched_state >= SCHED_INTERRUPTING);
 398     }
 399 #endif
 400
 401     //
 402     // Get a thread to run
 403     //
 404     t = popRunQueue(cap);
 405
 406     // Sanity check the thread we're about to run.  This can be
 407     // expensive if there is lots of thread switching going on...
 408     IF_DEBUG(sanity,checkTSO(t));
 409
 410 #if defined(THREADED_RTS)
 411     // Check whether we can run this thread in the current task.
 412     // If not, we have to pass our capability to the right task.
 413     {
 414         Task *bound = t->bound;
 415
 416         if (bound) {
 417             if (bound == task) {
 418                 debugTrace(DEBUG_sched,
 419                            "### Running thread %lu in bound thread", (unsigned long)t->id);
 420                 // yes, the Haskell thread is bound to the current native thread
 421             } else {
 422                 debugTrace(DEBUG_sched,
 423                            "### thread %lu bound to another OS thread", (unsigned long)t->id);
 424                 // no, bound to a different Haskell thread: pass to that thread
 425                 pushOnRunQueue(cap,t);
 426                 continue;
 427             }
 428         } else {
 429             // The thread we want to run is unbound.
 430             if (task->tso) {
 431                 debugTrace(DEBUG_sched,
 432                            "### this OS thread cannot run thread %lu", (unsigned long)t->id);
 433                 // no, the current native thread is bound to a different
 434                 // Haskell thread, so pass it to any worker thread
 435                 pushOnRunQueue(cap,t);
 436                 continue;
 437             }
 438         }
 439     }
 440 #endif
 441
 442     // If we're shutting down, and this thread has not yet been
 443     // killed, kill it now.  This sometimes happens when a finalizer
 444     // thread is created by the final GC, or a thread previously
 445     // in a foreign call returns.
 446     if (sched_state >= SCHED_INTERRUPTING &&
 447         !(t->what_next == ThreadComplete || t->what_next == ThreadKilled)) {
 448         deleteThread(cap,t);
 449     }
 450
 451     /* context switches are initiated by the timer signal, unless
 452      * the user specified "context switch as often as possible", with
 453      * +RTS -C0
 454      */
 455     if (RtsFlags.ConcFlags.ctxtSwitchTicks == 0
 456         && !emptyThreadQueues(cap)) {
 457         cap->context_switch = 1;
 458     }
 459
 460 run_thread:
 461
 462     // CurrentTSO is the thread to run.  t might be different if we
 463     // loop back to run_thread, so make sure to set CurrentTSO after
 464     // that.
 465     cap->r.rCurrentTSO = t;
 466
 467     debugTrace(DEBUG_sched, "-->> running thread %ld %s ...",
 468                               (long)t->id, whatNext_strs[t->what_next]);
 469
 470     startHeapProfTimer();
 471
 472     // Check for exceptions blocked on this thread
 473     maybePerformBlockedException (cap, t);
 474
 475     // ----------------------------------------------------------------------
 476     // Run the current thread
 477
 478     ASSERT_FULL_CAPABILITY_INVARIANTS(cap,task);
 479     ASSERT(t->cap == cap);
 480     ASSERT(t->bound ? t->bound->cap == cap : 1);
 481
 482     prev_what_next = t->what_next;
 483
 484     errno = t->saved_errno;
 485 #if mingw32_HOST_OS
 486     SetLastError(t->saved_winerror);
 487 #endif
 488
 489     cap->in_haskell = rtsTrue;
 490
 491     dirty_TSO(cap,t);
 492
 493 #if defined(THREADED_RTS)
 494     if (recent_activity == ACTIVITY_DONE_GC) {
 495         // ACTIVITY_DONE_GC means we turned off the timer signal to
 496         // conserve power (see #1623).  Re-enable it here.
 497         nat prev;
 498         prev = xchg((P_)&recent_activity, ACTIVITY_YES);
 499         if (prev == ACTIVITY_DONE_GC) {
 500             startTimer();
 501         }
 502     } else {
 503         recent_activity = ACTIVITY_YES;
 504     }
 505 #endif
 506
 507     postEvent(cap, EVENT_RUN_THREAD, t->id, 0);
 508
 509     switch (prev_what_next) {
 510
 511     case ThreadKilled:
 512     case ThreadComplete:
 513         /* Thread already finished, return to scheduler. */
 514         ret = ThreadFinished;
 515         break;
 516
 517     case ThreadRunGHC:
 518     {
 519         StgRegTable *r;
 520         r = StgRun((StgFunPtr) stg_returnToStackTop, &cap->r);
 521         cap = regTableToCapability(r);
 522         ret = r->rRet;
 523         break;
 524     }
 525
 526     case ThreadInterpret:
 527         cap = interpretBCO(cap);
 528         ret = cap->r.rRet;
 529         break;
 530
 531     default:
 532         barf("schedule: invalid what_next field");
 533     }
 534
 535     cap->in_haskell = rtsFalse;
 536
 537     // The TSO might have moved, eg. if it re-entered the RTS and a GC
 538     // happened.  So find the new location:
 539     t = cap->r.rCurrentTSO;
 540
 541     // We have run some Haskell code: there might be blackhole-blocked
 542     // threads to wake up now.
 543     // Lock-free test here should be ok, we're just setting a flag.
 544     if ( blackhole_queue != END_TSO_QUEUE ) {
 545         blackholes_need_checking = rtsTrue;
 546     }
 547
 548     // And save the current errno in this thread.
 549     // XXX: possibly bogus for SMP because this thread might already
 550     // be running again, see code below.
 551     t->saved_errno = errno;
 552 #if mingw32_HOST_OS
 553     // Similarly for Windows error code
 554     t->saved_winerror = GetLastError();
 555 #endif
 556
 557     postEvent (cap, EVENT_STOP_THREAD, t->id, ret);
 558
 559 #if defined(THREADED_RTS)
 560     // If ret is ThreadBlocked, and this Task is bound to the TSO that
 561     // blocked, we are in limbo - the TSO is now owned by whatever it
 562     // is blocked on, and may in fact already have been woken up,
 563     // perhaps even on a different Capability.  It may be the case
 564     // that task->cap != cap.  We better yield this Capability
 565     // immediately and return to normaility.
 566     if (ret == ThreadBlocked) {
 567         debugTrace(DEBUG_sched,
 568                    "--<< thread %lu (%s) stopped: blocked",
 569                    (unsigned long)t->id, whatNext_strs[t->what_next]);
 570         goto yield;
 571     }
 572 #endif
 573
 574     ASSERT_FULL_CAPABILITY_INVARIANTS(cap,task);
 575     ASSERT(t->cap == cap);
 576
 577     // ----------------------------------------------------------------------
 578
 579     // Costs for the scheduler are assigned to CCS_SYSTEM
 580     stopHeapProfTimer();
 581 #if defined(PROFILING)
 582     CCCS = CCS_SYSTEM;
 583 #endif
 584
 585     schedulePostRunThread(cap,t);
 586
 587     t = threadStackUnderflow(task,t);
 588
 589     ready_to_gc = rtsFalse;
 590
 591     switch (ret) {
 592     case HeapOverflow:
 593         ready_to_gc = scheduleHandleHeapOverflow(cap,t);
 594         break;
 595
 596     case StackOverflow:
 597         scheduleHandleStackOverflow(cap,task,t);
 598         break;
 599
 600     case ThreadYielding:
 601         if (scheduleHandleYield(cap, t, prev_what_next)) {
 602             // shortcut for switching between compiler/interpreter:
 603             goto run_thread;
 604         }
 605         break;
 606
 607     case ThreadBlocked:
 608         scheduleHandleThreadBlocked(t);
 609         break;
 610
 611     case ThreadFinished:
 612         if (scheduleHandleThreadFinished(cap, task, t)) return cap;
 613         ASSERT_FULL_CAPABILITY_INVARIANTS(cap,task);
 614         break;
 615
 616     default:
 617       barf("schedule: invalid thread return code %d", (int)ret);
 618     }
 619
 620     if (ready_to_gc || scheduleNeedHeapProfile(ready_to_gc)) {
 621       cap = scheduleDoGC(cap,task,rtsFalse);
 622     }
 623   } /* end of while() */
 624 }
 625
 626 /* ----------------------------------------------------------------------------
 627  * Setting up the scheduler loop
 628  * ------------------------------------------------------------------------- */
 629
 630 static void
 631 schedulePreLoop(void)
 632 {
 633   // initialisation for scheduler - what cannot go into initScheduler()
 634 }
 635
 636 /* -----------------------------------------------------------------------------
 637  * scheduleFindWork()
 638  *
 639  * Search for work to do, and handle messages from elsewhere.
 640  * -------------------------------------------------------------------------- */
 641
 642 static void
 643 scheduleFindWork (Capability *cap)
 644 {
 645     scheduleStartSignalHandlers(cap);
 646
 647     // Only check the black holes here if we've nothing else to do.
 648     // During normal execution, the black hole list only gets checked
 649     // at GC time, to avoid repeatedly traversing this possibly long
 650     // list each time around the scheduler.
 651     if (emptyRunQueue(cap)) { scheduleCheckBlackHoles(cap); }
 652
 653     scheduleCheckWakeupThreads(cap);
 654
 655     scheduleCheckBlockedThreads(cap);
 656
 657 #if defined(THREADED_RTS)
 658     if (emptyRunQueue(cap)) { scheduleActivateSpark(cap); }
 659 #endif
 660 }
 661
 662 #if defined(THREADED_RTS)
 663 STATIC_INLINE rtsBool
 664 shouldYieldCapability (Capability *cap, Task *task)
 665 {
 666     // we need to yield this capability to someone else if..
 667     //   - another thread is initiating a GC
 668     //   - another Task is returning from a foreign call
 669     //   - the thread at the head of the run queue cannot be run
 670     //     by this Task (it is bound to another Task, or it is unbound
 671     //     and this task it bound).
 672     return (waiting_for_gc ||
 673             cap->returning_tasks_hd != NULL ||
 674             (!emptyRunQueue(cap) && (task->tso == NULL
 675                                      ? cap->run_queue_hd->bound != NULL
 676                                      : cap->run_queue_hd->bound != task)));
 677 }
 678
 679 // This is the single place where a Task goes to sleep.  There are
 680 // two reasons it might need to sleep:
 681 //    - there are no threads to run
 682 //    - we need to yield this Capability to someone else
 683 //      (see shouldYieldCapability())
 684 //
 685 // Careful: the scheduler loop is quite delicate.  Make sure you run
 686 // the tests in testsuite/concurrent (all ways) after modifying this,
 687 // and also check the benchmarks in nofib/parallel for regressions.
 688
 689 static void
 690 scheduleYield (Capability **pcap, Task *task)
 691 {
 692     Capability *cap = *pcap;
 693
 694     // if we have work, and we don't need to give up the Capability, continue.
 695     if (!shouldYieldCapability(cap,task) &&
 696         (!emptyRunQueue(cap) ||
 697          !emptyWakeupQueue(cap) ||
 698          blackholes_need_checking ||
 699          sched_state >= SCHED_INTERRUPTING))
 700         return;
 701
 702     // otherwise yield (sleep), and keep yielding if necessary.
 703     do {
 704         yieldCapability(&cap,task);
 705     }
 706     while (shouldYieldCapability(cap,task));
 707
 708     // note there may still be no threads on the run queue at this
 709     // point, the caller has to check.
 710
 711     *pcap = cap;
 712     return;
 713 }
 714 #endif
 715
 716 /* -----------------------------------------------------------------------------
 717  * schedulePushWork()
 718  *
 719  * Push work to other Capabilities if we have some.
 720  * -------------------------------------------------------------------------- */
 721
 722 static void
 723 schedulePushWork(Capability *cap USED_IF_THREADS,
 724                  Task *task      USED_IF_THREADS)
 725 {
 726   /* following code not for PARALLEL_HASKELL. I kept the call general,
 727      future GUM versions might use pushing in a distributed setup */
 728 #if defined(THREADED_RTS)
 729
 730     Capability *free_caps[n_capabilities], *cap0;
 731     nat i, n_free_caps;
 732
 733     // migration can be turned off with +RTS -qg
 734     if (!RtsFlags.ParFlags.migrate) return;
 735
 736     // Check whether we have more threads on our run queue, or sparks
 737     // in our pool, that we could hand to another Capability.
 738     if (cap->run_queue_hd == END_TSO_QUEUE) {
 739         if (sparkPoolSizeCap(cap) < 2) return;
 740     } else {
 741         if (cap->run_queue_hd->_link == END_TSO_QUEUE &&
 742             sparkPoolSizeCap(cap) < 1) return;
 743     }
 744
 745     // First grab as many free Capabilities as we can.
 746     for (i=0, n_free_caps=0; i < n_capabilities; i++) {
 747         cap0 = &capabilities[i];
 748         if (cap != cap0 && tryGrabCapability(cap0,task)) {
 749             if (!emptyRunQueue(cap0) || cap->returning_tasks_hd != NULL) {
 750                 // it already has some work, we just grabbed it at
 751                 // the wrong moment.  Or maybe it's deadlocked!
 752                 releaseCapability(cap0);
 753             } else {
 754                 free_caps[n_free_caps++] = cap0;
 755             }
 756         }
 757     }
 758
 759     // we now have n_free_caps free capabilities stashed in
 760     // free_caps[].  Share our run queue equally with them.  This is
 761     // probably the simplest thing we could do; improvements we might
 762     // want to do include:
 763     //
 764     //   - giving high priority to moving relatively new threads, on
 765     //     the gournds that they haven't had time to build up a
 766     //     working set in the cache on this CPU/Capability.
 767     //
 768     //   - giving low priority to moving long-lived threads
 769
 770     if (n_free_caps > 0) {
 771         StgTSO *prev, *t, *next;
 772         rtsBool pushed_to_all;
 773
 774         debugTrace(DEBUG_sched,
 775                    "cap %d: %s and %d free capabilities, sharing...",
 776                    cap->no,
 777                    (!emptyRunQueue(cap) && cap->run_queue_hd->_link != END_TSO_QUEUE)?
 778                    "excess threads on run queue":"sparks to share (>=2)",
 779                    n_free_caps);
 780
 781         i = 0;
 782         pushed_to_all = rtsFalse;
 783
 784         if (cap->run_queue_hd != END_TSO_QUEUE) {
 785             prev = cap->run_queue_hd;
 786             t = prev->_link;
 787             prev->_link = END_TSO_QUEUE;
 788             for (; t != END_TSO_QUEUE; t = next) {
 789                 next = t->_link;
 790                 t->_link = END_TSO_QUEUE;
 791                 if (t->what_next == ThreadRelocated
 792                     || t->bound == task // don't move my bound thread
 793                     || tsoLocked(t)) {  // don't move a locked thread
 794                     setTSOLink(cap, prev, t);
 795                     prev = t;
 796                 } else if (i == n_free_caps) {
 797                     pushed_to_all = rtsTrue;
 798                     i = 0;
 799                     // keep one for us
 800                     setTSOLink(cap, prev, t);
 801                     prev = t;
 802                 } else {
 803                     debugTrace(DEBUG_sched, "pushing thread %lu to capability %d", (unsigned long)t->id, free_caps[i]->no);
 804                     appendToRunQueue(free_caps[i],t);
 805
 806         postEvent (cap, EVENT_MIGRATE_THREAD, t->id, free_caps[i]->no);
 807
 808                     if (t->bound) { t->bound->cap = free_caps[i]; }
 809                     t->cap = free_caps[i];
 810                     i++;
 811                 }
 812             }
 813             cap->run_queue_tl = prev;
 814         }
 815
 816 #ifdef SPARK_PUSHING
 817         /* JB I left this code in place, it would work but is not necessary */
 818
 819         // If there are some free capabilities that we didn't push any
 820         // threads to, then try to push a spark to each one.
 821         if (!pushed_to_all) {
 822             StgClosure *spark;
 823             // i is the next free capability to push to
 824             for (; i < n_free_caps; i++) {
 825                 if (emptySparkPoolCap(free_caps[i])) {
 826                     spark = tryStealSpark(cap->sparks);
 827                     if (spark != NULL) {
 828                         debugTrace(DEBUG_sched, "pushing spark %p to capability %d", spark, free_caps[i]->no);
 829
 830       postEvent(free_caps[i], EVENT_STEAL_SPARK, t->id, cap->no);
 831
 832                         newSpark(&(free_caps[i]->r), spark);
 833                     }
 834                 }
 835             }
 836         }
 837 #endif /* SPARK_PUSHING */
 838
 839         // release the capabilities
 840         for (i = 0; i < n_free_caps; i++) {
 841             task->cap = free_caps[i];
 842             releaseAndWakeupCapability(free_caps[i]);
 843         }
 844     }
 845     task->cap = cap; // reset to point to our Capability.
 846
 847 #endif /* THREADED_RTS */
 848
 849 }
 850
 851 /* ----------------------------------------------------------------------------
 852  * Start any pending signal handlers
 853  * ------------------------------------------------------------------------- */
 854
 855 #if defined(RTS_USER_SIGNALS) && !defined(THREADED_RTS)
 856 static void
 857 scheduleStartSignalHandlers(Capability *cap)
 858 {
 859     if (RtsFlags.MiscFlags.install_signal_handlers && signals_pending()) {
 860         // safe outside the lock
 861         startSignalHandlers(cap);
 862     }
 863 }
 864 #else
 865 static void
 866 scheduleStartSignalHandlers(Capability *cap STG_UNUSED)
 867 {
 868 }
 869 #endif
 870
 871 /* ----------------------------------------------------------------------------
 872  * Check for blocked threads that can be woken up.
 873  * ------------------------------------------------------------------------- */
 874
 875 static void
 876 scheduleCheckBlockedThreads(Capability *cap USED_IF_NOT_THREADS)
 877 {
 878 #if !defined(THREADED_RTS)
 879     //
 880     // Check whether any waiting threads need to be woken up.  If the
 881     // run queue is empty, and there are no other tasks running, we
 882     // can wait indefinitely for something to happen.
 883     //
 884     if ( !emptyQueue(blocked_queue_hd) || !emptyQueue(sleeping_queue) )
 885     {
 886         awaitEvent( emptyRunQueue(cap) && !blackholes_need_checking );
 887     }
 888 #endif
 889 }
 890
 891
 892 /* ----------------------------------------------------------------------------
 893  * Check for threads woken up by other Capabilities
 894  * ------------------------------------------------------------------------- */
 895
 896 static void
 897 scheduleCheckWakeupThreads(Capability *cap USED_IF_THREADS)
 898 {
 899 #if defined(THREADED_RTS)
 900     // Any threads that were woken up by other Capabilities get
 901     // appended to our run queue.
 902     if (!emptyWakeupQueue(cap)) {
 903         ACQUIRE_LOCK(&cap->lock);
 904         if (emptyRunQueue(cap)) {
 905             cap->run_queue_hd = cap->wakeup_queue_hd;
 906             cap->run_queue_tl = cap->wakeup_queue_tl;
 907         } else {
 908             setTSOLink(cap, cap->run_queue_tl, cap->wakeup_queue_hd);
 909             cap->run_queue_tl = cap->wakeup_queue_tl;
 910         }
 911         cap->wakeup_queue_hd = cap->wakeup_queue_tl = END_TSO_QUEUE;
 912         RELEASE_LOCK(&cap->lock);
 913     }
 914 #endif
 915 }
 916
 917 /* ----------------------------------------------------------------------------
 918  * Check for threads blocked on BLACKHOLEs that can be woken up
 919  * ------------------------------------------------------------------------- */
 920 static void
 921 scheduleCheckBlackHoles (Capability *cap)
 922 {
 923     if ( blackholes_need_checking ) // check without the lock first
 924     {
 925         ACQUIRE_LOCK(&sched_mutex);
 926         if ( blackholes_need_checking ) {
 927             blackholes_need_checking = rtsFalse;
 928             // important that we reset the flag *before* checking the
 929             // blackhole queue, otherwise we could get deadlock.  This
 930             // happens as follows: we wake up a thread that
 931             // immediately runs on another Capability, blocks on a
 932             // blackhole, and then we reset the blackholes_need_checking flag.
 933             checkBlackHoles(cap);
 934         }
 935         RELEASE_LOCK(&sched_mutex);
 936     }
 937 }
 938
 939 /* ----------------------------------------------------------------------------
 940  * Detect deadlock conditions and attempt to resolve them.
 941  * ------------------------------------------------------------------------- */
 942
 943 static void
 944 scheduleDetectDeadlock (Capability *cap, Task *task)
 945 {
 946     /*
 947      * Detect deadlock: when we have no threads to run, there are no
 948      * threads blocked, waiting for I/O, or sleeping, and all the
 949      * other tasks are waiting for work, we must have a deadlock of
 950      * some description.
 951      */
 952     if ( emptyThreadQueues(cap) )
 953     {
 954 #if defined(THREADED_RTS)
 955         /*
 956          * In the threaded RTS, we only check for deadlock if there
 957          * has been no activity in a complete timeslice.  This means
 958          * we won't eagerly start a full GC just because we don't have
 959          * any threads to run currently.
 960          */
 961         if (recent_activity != ACTIVITY_INACTIVE) return;
 962 #endif
 963
 964         debugTrace(DEBUG_sched, "deadlocked, forcing major GC...");
 965
 966         // Garbage collection can release some new threads due to
 967         // either (a) finalizers or (b) threads resurrected because
 968         // they are unreachable and will therefore be sent an
 969         // exception.  Any threads thus released will be immediately
 970         // runnable.
 971         cap = scheduleDoGC (cap, task, rtsTrue/*force major GC*/);
 972         // when force_major == rtsTrue. scheduleDoGC sets
 973         // recent_activity to ACTIVITY_DONE_GC and turns off the timer
 974         // signal.
 975
 976         if ( !emptyRunQueue(cap) ) return;
 977
 978 #if defined(RTS_USER_SIGNALS) && !defined(THREADED_RTS)
 979         /* If we have user-installed signal handlers, then wait
 980          * for signals to arrive rather then bombing out with a
 981          * deadlock.
 982          */
 983         if ( RtsFlags.MiscFlags.install_signal_handlers && anyUserHandlers() ) {
 984             debugTrace(DEBUG_sched,
 985                        "still deadlocked, waiting for signals...");
 986
 987             awaitUserSignals();
 988
 989             if (signals_pending()) {
 990                 startSignalHandlers(cap);
 991             }
 992
 993             // either we have threads to run, or we were interrupted:
 994             ASSERT(!emptyRunQueue(cap) || sched_state >= SCHED_INTERRUPTING);
 995
 996             return;
 997         }
 998 #endif
 999
1000 #if !defined(THREADED_RTS)
1001         /* Probably a real deadlock.  Send the current main thread the
1002          * Deadlock exception.
1003          */
1004         if (task->tso) {
1005             switch (task->tso->why_blocked) {
1006             case BlockedOnSTM:
1007             case BlockedOnBlackHole:
1008             case BlockedOnException:
1009             case BlockedOnMVar:
1010                 throwToSingleThreaded(cap, task->tso,
1011                                       (StgClosure *)nonTermination_closure);
1012                 return;
1013             default:
1014                 barf("deadlock: main thread blocked in a strange way");
1015             }
1016         }
1017         return;
1018 #endif
1019     }
1020 }
1021
1022
1023 /* ----------------------------------------------------------------------------
1024  * Send pending messages (PARALLEL_HASKELL only)
1025  * ------------------------------------------------------------------------- */
1026
1027 #if defined(PARALLEL_HASKELL)
1028 static void
1029 scheduleSendPendingMessages(void)
1030 {
1031
1032 # if defined(PAR) // global Mem.Mgmt., omit for now
1033     if (PendingFetches != END_BF_QUEUE) {
1034         processFetches();
1035     }
1036 # endif
1037
1038     if (RtsFlags.ParFlags.BufferTime) {
1039         // if we use message buffering, we must send away all message
1040         // packets which have become too old...
1041         sendOldBuffers();
1042     }
1043 }
1044 #endif
1045
1046 /* ----------------------------------------------------------------------------
1047  * Activate spark threads (PARALLEL_HASKELL and THREADED_RTS)
1048  * ------------------------------------------------------------------------- */
1049
1050 #if defined(THREADED_RTS)
1051 static void
1052 scheduleActivateSpark(Capability *cap)
1053 {
1054     if (anySparks())
1055     {
1056         createSparkThread(cap);
1057         debugTrace(DEBUG_sched, "creating a spark thread");
1058     }
1059 }
1060 #endif // PARALLEL_HASKELL || THREADED_RTS
1061
1062 /* ----------------------------------------------------------------------------
1063  * After running a thread...
1064  * ------------------------------------------------------------------------- */
1065
1066 static void
1067 schedulePostRunThread (Capability *cap, StgTSO *t)
1068 {
1069     // We have to be able to catch transactions that are in an
1070     // infinite loop as a result of seeing an inconsistent view of
1071     // memory, e.g.
1072     //
1073     //   atomically $ do
1074     //       [a,b] <- mapM readTVar [ta,tb]
1075     //       when (a == b) loop
1076     //
1077     // and a is never equal to b given a consistent view of memory.
1078     //
1079     if (t -> trec != NO_TREC && t -> why_blocked == NotBlocked) {
1080         if (!stmValidateNestOfTransactions (t -> trec)) {
1081             debugTrace(DEBUG_sched | DEBUG_stm,
1082                        "trec %p found wasting its time", t);
1083
1084             // strip the stack back to the
1085             // ATOMICALLY_FRAME, aborting the (nested)
1086             // transaction, and saving the stack of any
1087             // partially-evaluated thunks on the heap.
1088             throwToSingleThreaded_(cap, t, NULL, rtsTrue);
1089
1090             ASSERT(get_itbl((StgClosure *)t->sp)->type == ATOMICALLY_FRAME);
1091         }
1092     }
1093
1094   /* some statistics gathering in the parallel case */
1095 }
1096
1097 /* -----------------------------------------------------------------------------
1098  * Handle a thread that returned to the scheduler with ThreadHeepOverflow
1099  * -------------------------------------------------------------------------- */
1100
1101 static rtsBool
1102 scheduleHandleHeapOverflow( Capability *cap, StgTSO *t )
1103 {
1104     // did the task ask for a large block?
1105     if (cap->r.rHpAlloc > BLOCK_SIZE) {
1106         // if so, get one and push it on the front of the nursery.
1107         bdescr *bd;
1108         lnat blocks;
1109
1110         blocks = (lnat)BLOCK_ROUND_UP(cap->r.rHpAlloc) / BLOCK_SIZE;
1111
1112         debugTrace(DEBUG_sched,
1113                    "--<< thread %ld (%s) stopped: requesting a large block (size %ld)\n",
1114                    (long)t->id, whatNext_strs[t->what_next], blocks);
1115
1116         // don't do this if the nursery is (nearly) full, we'll GC first.
1117         if (cap->r.rCurrentNursery->link != NULL ||
1118             cap->r.rNursery->n_blocks == 1) {  // paranoia to prevent infinite loop
1119                                                // if the nursery has only one block.
1120
1121             ACQUIRE_SM_LOCK
1122             bd = allocGroup( blocks );
1123             RELEASE_SM_LOCK
1124             cap->r.rNursery->n_blocks += blocks;
1125
1126             // link the new group into the list
1127             bd->link = cap->r.rCurrentNursery;
1128             bd->u.back = cap->r.rCurrentNursery->u.back;
1129             if (cap->r.rCurrentNursery->u.back != NULL) {
1130                 cap->r.rCurrentNursery->u.back->link = bd;
1131             } else {
1132 #if !defined(THREADED_RTS)
1133                 ASSERT(g0s0->blocks == cap->r.rCurrentNursery &&
1134                        g0s0 == cap->r.rNursery);
1135 #endif
1136                 cap->r.rNursery->blocks = bd;
1137             }
1138             cap->r.rCurrentNursery->u.back = bd;
1139
1140             // initialise it as a nursery block.  We initialise the
1141             // step, gen_no, and flags field of *every* sub-block in
1142             // this large block, because this is easier than making
1143             // sure that we always find the block head of a large
1144             // block whenever we call Bdescr() (eg. evacuate() and
1145             // isAlive() in the GC would both have to do this, at
1146             // least).
1147             {
1148                 bdescr *x;
1149                 for (x = bd; x < bd + blocks; x++) {
1150                     x->step = cap->r.rNursery;
1151                     x->gen_no = 0;
1152                     x->flags = 0;
1153                 }
1154             }
1155
1156             // This assert can be a killer if the app is doing lots
1157             // of large block allocations.
1158             IF_DEBUG(sanity, checkNurserySanity(cap->r.rNursery));
1159
1160             // now update the nursery to point to the new block
1161             cap->r.rCurrentNursery = bd;
1162
1163             // we might be unlucky and have another thread get on the
1164             // run queue before us and steal the large block, but in that
1165             // case the thread will just end up requesting another large
1166             // block.
1167             pushOnRunQueue(cap,t);
1168             return rtsFalse;  /* not actually GC'ing */
1169         }
1170     }
1171
1172     debugTrace(DEBUG_sched,
1173                "--<< thread %ld (%s) stopped: HeapOverflow",
1174                (long)t->id, whatNext_strs[t->what_next]);
1175
1176     if (cap->r.rHpLim == NULL || cap->context_switch) {
1177         // Sometimes we miss a context switch, e.g. when calling
1178         // primitives in a tight loop, MAYBE_GC() doesn't check the
1179         // context switch flag, and we end up waiting for a GC.
1180         // See #1984, and concurrent/should_run/1984
1181         cap->context_switch = 0;
1182         addToRunQueue(cap,t);
1183     } else {
1184         pushOnRunQueue(cap,t);
1185     }
1186     return rtsTrue;
1187     /* actual GC is done at the end of the while loop in schedule() */
1188 }
1189
1190 /* -----------------------------------------------------------------------------
1191  * Handle a thread that returned to the scheduler with ThreadStackOverflow
1192  * -------------------------------------------------------------------------- */
1193
1194 static void
1195 scheduleHandleStackOverflow (Capability *cap, Task *task, StgTSO *t)
1196 {
1197     debugTrace (DEBUG_sched,
1198                 "--<< thread %ld (%s) stopped, StackOverflow",
1199                 (long)t->id, whatNext_strs[t->what_next]);
1200
1201     /* just adjust the stack for this thread, then pop it back
1202      * on the run queue.
1203      */
1204     {
1205         /* enlarge the stack */
1206         StgTSO *new_t = threadStackOverflow(cap, t);
1207
1208         /* The TSO attached to this Task may have moved, so update the
1209          * pointer to it.
1210          */
1211         if (task->tso == t) {
1212             task->tso = new_t;
1213         }
1214         pushOnRunQueue(cap,new_t);
1215     }
1216 }
1217
1218 /* -----------------------------------------------------------------------------
1219  * Handle a thread that returned to the scheduler with ThreadYielding
1220  * -------------------------------------------------------------------------- */
1221
1222 static rtsBool
1223 scheduleHandleYield( Capability *cap, StgTSO *t, nat prev_what_next )
1224 {
1225     // Reset the context switch flag.  We don't do this just before
1226     // running the thread, because that would mean we would lose ticks
1227     // during GC, which can lead to unfair scheduling (a thread hogs
1228     // the CPU because the tick always arrives during GC).  This way
1229     // penalises threads that do a lot of allocation, but that seems
1230     // better than the alternative.
1231     cap->context_switch = 0;
1232
1233     /* put the thread back on the run queue.  Then, if we're ready to
1234      * GC, check whether this is the last task to stop.  If so, wake
1235      * up the GC thread.  getThread will block during a GC until the
1236      * GC is finished.
1237      */
1238 #ifdef DEBUG
1239     if (t->what_next != prev_what_next) {
1240         debugTrace(DEBUG_sched,
1241                    "--<< thread %ld (%s) stopped to switch evaluators",
1242                    (long)t->id, whatNext_strs[t->what_next]);
1243     } else {
1244         debugTrace(DEBUG_sched,
1245                    "--<< thread %ld (%s) stopped, yielding",
1246                    (long)t->id, whatNext_strs[t->what_next]);
1247     }
1248 #endif
1249
1250     IF_DEBUG(sanity,
1251              //debugBelch("&& Doing sanity check on yielding TSO %ld.", t->id);
1252              checkTSO(t));
1253     ASSERT(t->_link == END_TSO_QUEUE);
1254
1255     // Shortcut if we're just switching evaluators: don't bother
1256     // doing stack squeezing (which can be expensive), just run the
1257     // thread.
1258     if (t->what_next != prev_what_next) {
1259         return rtsTrue;
1260     }
1261
1262     addToRunQueue(cap,t);
1263
1264     return rtsFalse;
1265 }
1266
1267 /* -----------------------------------------------------------------------------
1268  * Handle a thread that returned to the scheduler with ThreadBlocked
1269  * -------------------------------------------------------------------------- */
1270
1271 static void
1272 scheduleHandleThreadBlocked( StgTSO *t
1273 #if !defined(DEBUG)
1274     STG_UNUSED
1275 #endif
1276     )
1277 {
1278
1279       // We don't need to do anything.  The thread is blocked, and it
1280       // has tidied up its stack and placed itself on whatever queue
1281       // it needs to be on.
1282
1283     // ASSERT(t->why_blocked != NotBlocked);
1284     // Not true: for example,
1285     //    - in THREADED_RTS, the thread may already have been woken
1286     //      up by another Capability.  This actually happens: try
1287     //      conc023 +RTS -N2.
1288     //    - the thread may have woken itself up already, because
1289     //      threadPaused() might have raised a blocked throwTo
1290     //      exception, see maybePerformBlockedException().
1291
1292 #ifdef DEBUG
1293     if (traceClass(DEBUG_sched)) {
1294         debugTraceBegin("--<< thread %lu (%s) stopped: ",
1295                         (unsigned long)t->id, whatNext_strs[t->what_next]);
1296         printThreadBlockage(t);
1297         debugTraceEnd();
1298     }
1299 #endif
1300 }
1301
1302 /* -----------------------------------------------------------------------------
1303  * Handle a thread that returned to the scheduler with ThreadFinished
1304  * -------------------------------------------------------------------------- */
1305
1306 static rtsBool
1307 scheduleHandleThreadFinished (Capability *cap STG_UNUSED, Task *task, StgTSO *t)
1308 {
1309     /* Need to check whether this was a main thread, and if so,
1310      * return with the return value.
1311      *
1312      * We also end up here if the thread kills itself with an
1313      * uncaught exception, see Exception.cmm.
1314      */
1315     debugTrace(DEBUG_sched, "--++ thread %lu (%s) finished",
1316                (unsigned long)t->id, whatNext_strs[t->what_next]);
1317
1318     // blocked exceptions can now complete, even if the thread was in
1319     // blocked mode (see #2910).  This unconditionally calls
1320     // lockTSO(), which ensures that we don't miss any threads that
1321     // are engaged in throwTo() with this thread as a target.
1322     awakenBlockedExceptionQueue (cap, t);
1323
1324       //
1325       // Check whether the thread that just completed was a bound
1326       // thread, and if so return with the result.
1327       //
1328       // There is an assumption here that all thread completion goes
1329       // through this point; we need to make sure that if a thread
1330       // ends up in the ThreadKilled state, that it stays on the run
1331       // queue so it can be dealt with here.
1332       //
1333
1334       if (t->bound) {
1335
1336           if (t->bound != task) {
1337 #if !defined(THREADED_RTS)
1338               // Must be a bound thread that is not the topmost one.  Leave
1339               // it on the run queue until the stack has unwound to the
1340               // point where we can deal with this.  Leaving it on the run
1341               // queue also ensures that the garbage collector knows about
1342               // this thread and its return value (it gets dropped from the
1343               // step->threads list so there's no other way to find it).
1344               appendToRunQueue(cap,t);
1345               return rtsFalse;
1346 #else
1347               // this cannot happen in the threaded RTS, because a
1348               // bound thread can only be run by the appropriate Task.
1349               barf("finished bound thread that isn't mine");
1350 #endif
1351           }
1352
1353           ASSERT(task->tso == t);
1354
1355           if (t->what_next == ThreadComplete) {
1356               if (task->ret) {
1357                   // NOTE: return val is tso->sp[1] (see StgStartup.hc)
1358                   *(task->ret) = (StgClosure *)task->tso->sp[1];
1359               }
1360               task->stat = Success;
1361           } else {
1362               if (task->ret) {
1363                   *(task->ret) = NULL;
1364               }
1365               if (sched_state >= SCHED_INTERRUPTING) {
1366                   if (heap_overflow) {
1367                       task->stat = HeapExhausted;
1368                   } else {
1369                       task->stat = Interrupted;
1370                   }
1371               } else {
1372                   task->stat = Killed;
1373               }
1374           }
1375 #ifdef DEBUG
1376           removeThreadLabel((StgWord)task->tso->id);
1377 #endif
1378           return rtsTrue; // tells schedule() to return
1379       }
1380
1381       return rtsFalse;
1382 }
1383
1384 /* -----------------------------------------------------------------------------
1385  * Perform a heap census
1386  * -------------------------------------------------------------------------- */
1387
1388 static rtsBool
1389 scheduleNeedHeapProfile( rtsBool ready_to_gc STG_UNUSED )
1390 {
1391     // When we have +RTS -i0 and we're heap profiling, do a census at
1392     // every GC.  This lets us get repeatable runs for debugging.
1393     if (performHeapProfile ||
1394         (RtsFlags.ProfFlags.profileInterval==0 &&
1395          RtsFlags.ProfFlags.doHeapProfile && ready_to_gc)) {
1396         return rtsTrue;
1397     } else {
1398         return rtsFalse;
1399     }
1400 }
1401
1402 /* -----------------------------------------------------------------------------
1403  * Perform a garbage collection if necessary
1404  * -------------------------------------------------------------------------- */
1405
1406 static Capability *
1407 scheduleDoGC (Capability *cap, Task *task USED_IF_THREADS, rtsBool force_major)
1408 {
1409     rtsBool heap_census;
1410 #ifdef THREADED_RTS
1411     /* extern static volatile StgWord waiting_for_gc;
1412        lives inside capability.c */
1413     rtsBool gc_type, prev_pending_gc;
1414     nat i;
1415 #endif
1416
1417     if (sched_state == SCHED_SHUTTING_DOWN) {
1418         // The final GC has already been done, and the system is
1419         // shutting down.  We'll probably deadlock if we try to GC
1420         // now.
1421         return cap;
1422     }
1423
1424 #ifdef THREADED_RTS
1425     if (sched_state < SCHED_INTERRUPTING
1426         && RtsFlags.ParFlags.parGcEnabled
1427         && N >= RtsFlags.ParFlags.parGcGen
1428         && ! oldest_gen->steps[0].mark)
1429     {
1430         gc_type = PENDING_GC_PAR;
1431     } else {
1432         gc_type = PENDING_GC_SEQ;
1433     }
1434
1435     // In order to GC, there must be no threads running Haskell code.
1436     // Therefore, the GC thread needs to hold *all* the capabilities,
1437     // and release them after the GC has completed.
1438     //
1439     // This seems to be the simplest way: previous attempts involved
1440     // making all the threads with capabilities give up their
1441     // capabilities and sleep except for the *last* one, which
1442     // actually did the GC.  But it's quite hard to arrange for all
1443     // the other tasks to sleep and stay asleep.
1444     //
1445
1446     /*  Other capabilities are prevented from running yet more Haskell
1447         threads if waiting_for_gc is set. Tested inside
1448         yieldCapability() and releaseCapability() in Capability.c */
1449
1450     prev_pending_gc = cas(&waiting_for_gc, 0, gc_type);
1451     if (prev_pending_gc) {
1452         do {
1453             debugTrace(DEBUG_sched, "someone else is trying to GC (%d)...",
1454                        prev_pending_gc);
1455             ASSERT(cap);
1456             yieldCapability(&cap,task);
1457         } while (waiting_for_gc);
1458         return cap;  // NOTE: task->cap might have changed here
1459     }
1460
1461     setContextSwitches();
1462
1463     // The final shutdown GC is always single-threaded, because it's
1464     // possible that some of the Capabilities have no worker threads.
1465
1466     if (gc_type == PENDING_GC_SEQ)
1467     {
1468         postEvent(cap, EVENT_REQUEST_SEQ_GC, 0, 0);
1469         // single-threaded GC: grab all the capabilities
1470         for (i=0; i < n_capabilities; i++) {
1471             debugTrace(DEBUG_sched, "ready_to_gc, grabbing all the capabilies (%d/%d)", i, n_capabilities);
1472             if (cap != &capabilities[i]) {
1473                 Capability *pcap = &capabilities[i];
1474                 // we better hope this task doesn't get migrated to
1475                 // another Capability while we're waiting for this one.
1476                 // It won't, because load balancing happens while we have
1477                 // all the Capabilities, but even so it's a slightly
1478                 // unsavoury invariant.
1479                 task->cap = pcap;
1480                 waitForReturnCapability(&pcap, task);
1481                 if (pcap != &capabilities[i]) {
1482                     barf("scheduleDoGC: got the wrong capability");
1483                 }
1484             }
1485         }
1486     }
1487     else
1488     {
1489         // multi-threaded GC: make sure all the Capabilities donate one
1490         // GC thread each.
1491         postEvent(cap, EVENT_REQUEST_PAR_GC, 0, 0);
1492         debugTrace(DEBUG_sched, "ready_to_gc, grabbing GC threads");
1493
1494         waitForGcThreads(cap);
1495     }
1496 #endif
1497
1498     // so this happens periodically:
1499     if (cap) scheduleCheckBlackHoles(cap);
1500
1501     IF_DEBUG(scheduler, printAllThreads());
1502
1503 delete_threads_and_gc:
1504     /*
1505      * We now have all the capabilities; if we're in an interrupting
1506      * state, then we should take the opportunity to delete all the
1507      * threads in the system.
1508      */
1509     if (sched_state == SCHED_INTERRUPTING) {
1510         deleteAllThreads(cap);
1511         sched_state = SCHED_SHUTTING_DOWN;
1512     }
1513
1514     heap_census = scheduleNeedHeapProfile(rtsTrue);
1515
1516 #if defined(THREADED_RTS)
1517     postEvent(cap, EVENT_GC_START, 0, 0);
1518     debugTrace(DEBUG_sched, "doing GC");
1519     // reset waiting_for_gc *before* GC, so that when the GC threads
1520     // emerge they don't immediately re-enter the GC.
1521     waiting_for_gc = 0;
1522     GarbageCollect(force_major || heap_census, gc_type, cap);
1523 #else
1524     GarbageCollect(force_major || heap_census, 0, cap);
1525 #endif
1526     postEvent(cap, EVENT_GC_END, 0, 0);
1527
1528     if (recent_activity == ACTIVITY_INACTIVE && force_major)
1529     {
1530         // We are doing a GC because the system has been idle for a
1531         // timeslice and we need to check for deadlock.  Record the
1532         // fact that we've done a GC and turn off the timer signal;
1533         // it will get re-enabled if we run any threads after the GC.
1534         recent_activity = ACTIVITY_DONE_GC;
1535         stopTimer();
1536     }
1537     else
1538     {
1539         // the GC might have taken long enough for the timer to set
1540         // recent_activity = ACTIVITY_INACTIVE, but we aren't
1541         // necessarily deadlocked:
1542         recent_activity = ACTIVITY_YES;
1543     }
1544
1545 #if defined(THREADED_RTS)
1546     if (gc_type == PENDING_GC_PAR)
1547     {
1548         releaseGCThreads(cap);
1549     }
1550 #endif
1551
1552     if (heap_census) {
1553         debugTrace(DEBUG_sched, "performing heap census");
1554         heapCensus();
1555         performHeapProfile = rtsFalse;
1556     }
1557
1558     if (heap_overflow && sched_state < SCHED_INTERRUPTING) {
1559         // GC set the heap_overflow flag, so we should proceed with
1560         // an orderly shutdown now.  Ultimately we want the main
1561         // thread to return to its caller with HeapExhausted, at which
1562         // point the caller should call hs_exit().  The first step is
1563         // to delete all the threads.
1564         //
1565         // Another way to do this would be to raise an exception in
1566         // the main thread, which we really should do because it gives
1567         // the program a chance to clean up.  But how do we find the
1568         // main thread?  It should presumably be the same one that
1569         // gets ^C exceptions, but that's all done on the Haskell side
1570         // (GHC.TopHandler).
1571         sched_state = SCHED_INTERRUPTING;
1572         goto delete_threads_and_gc;
1573     }
1574
1575 #ifdef SPARKBALANCE
1576     /* JB
1577        Once we are all together... this would be the place to balance all
1578        spark pools. No concurrent stealing or adding of new sparks can
1579        occur. Should be defined in Sparks.c. */
1580     balanceSparkPoolsCaps(n_capabilities, capabilities);
1581 #endif
1582
1583 #if defined(THREADED_RTS)
1584     if (gc_type == PENDING_GC_SEQ) {
1585         // release our stash of capabilities.
1586         for (i = 0; i < n_capabilities; i++) {
1587             if (cap != &capabilities[i]) {
1588                 task->cap = &capabilities[i];
1589                 releaseCapability(&capabilities[i]);
1590             }
1591         }
1592     }
1593     if (cap) {
1594         task->cap = cap;
1595     } else {
1596         task->cap = NULL;
1597     }
1598 #endif
1599
1600     return cap;
1601 }
1602
1603 /* ---------------------------------------------------------------------------
1604  * Singleton fork(). Do not copy any running threads.
1605  * ------------------------------------------------------------------------- */
1606
1607 pid_t
1608 forkProcess(HsStablePtr *entry
1609 #ifndef FORKPROCESS_PRIMOP_SUPPORTED
1610             STG_UNUSED
1611 #endif
1612            )
1613 {
1614 #ifdef FORKPROCESS_PRIMOP_SUPPORTED
1615     Task *task;
1616     pid_t pid;
1617     StgTSO* t,*next;
1618     Capability *cap;
1619     nat s;
1620
1621 #if defined(THREADED_RTS)
1622     if (RtsFlags.ParFlags.nNodes > 1) {
1623         errorBelch("forking not supported with +RTS -N<n> greater than 1");
1624         stg_exit(EXIT_FAILURE);
1625     }
1626 #endif
1627
1628     debugTrace(DEBUG_sched, "forking!");
1629
1630     // ToDo: for SMP, we should probably acquire *all* the capabilities
1631     cap = rts_lock();
1632
1633     // no funny business: hold locks while we fork, otherwise if some
1634     // other thread is holding a lock when the fork happens, the data
1635     // structure protected by the lock will forever be in an
1636     // inconsistent state in the child.  See also #1391.
1637     ACQUIRE_LOCK(&sched_mutex);
1638     ACQUIRE_LOCK(&cap->lock);
1639     ACQUIRE_LOCK(&cap->running_task->lock);
1640
1641     pid = fork();
1642
1643     if (pid) { // parent
1644
1645         RELEASE_LOCK(&sched_mutex);
1646         RELEASE_LOCK(&cap->lock);
1647         RELEASE_LOCK(&cap->running_task->lock);
1648
1649         // just return the pid
1650         rts_unlock(cap);
1651         return pid;
1652
1653     } else { // child
1654
1655 #if defined(THREADED_RTS)
1656         initMutex(&sched_mutex);
1657         initMutex(&cap->lock);
1658         initMutex(&cap->running_task->lock);
1659 #endif
1660
1661         // Now, all OS threads except the thread that forked are
1662         // stopped.  We need to stop all Haskell threads, including
1663         // those involved in foreign calls.  Also we need to delete
1664         // all Tasks, because they correspond to OS threads that are
1665         // now gone.
1666
1667         for (s = 0; s < total_steps; s++) {
1668           for (t = all_steps[s].threads; t != END_TSO_QUEUE; t = next) {
1669             if (t->what_next == ThreadRelocated) {
1670                 next = t->_link;
1671             } else {
1672                 next = t->global_link;
1673                 // don't allow threads to catch the ThreadKilled
1674                 // exception, but we do want to raiseAsync() because these
1675                 // threads may be evaluating thunks that we need later.
1676                 deleteThread_(cap,t);
1677             }
1678           }
1679         }
1680
1681         // Empty the run queue.  It seems tempting to let all the
1682         // killed threads stay on the run queue as zombies to be
1683         // cleaned up later, but some of them correspond to bound
1684         // threads for which the corresponding Task does not exist.
1685         cap->run_queue_hd = END_TSO_QUEUE;
1686         cap->run_queue_tl = END_TSO_QUEUE;
1687
1688         // Any suspended C-calling Tasks are no more, their OS threads
1689         // don't exist now:
1690         cap->suspended_ccalling_tasks = NULL;
1691
1692         // Empty the threads lists.  Otherwise, the garbage
1693         // collector may attempt to resurrect some of these threads.
1694         for (s = 0; s < total_steps; s++) {
1695             all_steps[s].threads = END_TSO_QUEUE;
1696         }
1697
1698         // Wipe the task list, except the current Task.
1699         ACQUIRE_LOCK(&sched_mutex);
1700         for (task = all_tasks; task != NULL; task=task->all_link) {
1701             if (task != cap->running_task) {
1702 #if defined(THREADED_RTS)
1703                 initMutex(&task->lock); // see #1391
1704 #endif
1705                 discardTask(task);
1706             }
1707         }
1708         RELEASE_LOCK(&sched_mutex);
1709
1710 #if defined(THREADED_RTS)
1711         // Wipe our spare workers list, they no longer exist.  New
1712         // workers will be created if necessary.
1713         cap->spare_workers = NULL;
1714         cap->returning_tasks_hd = NULL;
1715         cap->returning_tasks_tl = NULL;
1716 #endif
1717
1718         // On Unix, all timers are reset in the child, so we need to start
1719         // the timer again.
1720         initTimer();
1721         startTimer();
1722
1723         cap = rts_evalStableIO(cap, entry, NULL);  // run the action
1724         rts_checkSchedStatus("forkProcess",cap);
1725
1726         rts_unlock(cap);
1727         hs_exit();                      // clean up and exit
1728         stg_exit(EXIT_SUCCESS);
1729     }
1730 #else /* !FORKPROCESS_PRIMOP_SUPPORTED */
1731     barf("forkProcess#: primop not supported on this platform, sorry!\n");
1732     return -1;
1733 #endif
1734 }
1735
1736 /* ---------------------------------------------------------------------------
1737  * Delete all the threads in the system
1738  * ------------------------------------------------------------------------- */
1739
1740 static void
1741 deleteAllThreads ( Capability *cap )
1742 {
1743     // NOTE: only safe to call if we own all capabilities.
1744
1745     StgTSO* t, *next;
1746     nat s;
1747
1748     debugTrace(DEBUG_sched,"deleting all threads");
1749     for (s = 0; s < total_steps; s++) {
1750       for (t = all_steps[s].threads; t != END_TSO_QUEUE; t = next) {
1751         if (t->what_next == ThreadRelocated) {
1752             next = t->_link;
1753         } else {
1754             next = t->global_link;
1755             deleteThread(cap,t);
1756         }
1757       }
1758     }
1759
1760     // The run queue now contains a bunch of ThreadKilled threads.  We
1761     // must not throw these away: the main thread(s) will be in there
1762     // somewhere, and the main scheduler loop has to deal with it.
1763     // Also, the run queue is the only thing keeping these threads from
1764     // being GC'd, and we don't want the "main thread has been GC'd" panic.
1765
1766 #if !defined(THREADED_RTS)
1767     ASSERT(blocked_queue_hd == END_TSO_QUEUE);
1768     ASSERT(sleeping_queue == END_TSO_QUEUE);
1769 #endif
1770 }
1771
1772 /* -----------------------------------------------------------------------------
1773    Managing the suspended_ccalling_tasks list.
1774    Locks required: sched_mutex
1775    -------------------------------------------------------------------------- */
1776
1777 STATIC_INLINE void
1778 suspendTask (Capability *cap, Task *task)
1779 {
1780     ASSERT(task->next == NULL && task->prev == NULL);
1781     task->next = cap->suspended_ccalling_tasks;
1782     task->prev = NULL;
1783     if (cap->suspended_ccalling_tasks) {
1784         cap->suspended_ccalling_tasks->prev = task;
1785     }
1786     cap->suspended_ccalling_tasks = task;
1787 }
1788
1789 STATIC_INLINE void
1790 recoverSuspendedTask (Capability *cap, Task *task)
1791 {
1792     if (task->prev) {
1793         task->prev->next = task->next;
1794     } else {
1795         ASSERT(cap->suspended_ccalling_tasks == task);
1796         cap->suspended_ccalling_tasks = task->next;
1797     }
1798     if (task->next) {
1799         task->next->prev = task->prev;
1800     }
1801     task->next = task->prev = NULL;
1802 }
1803
1804 /* ---------------------------------------------------------------------------
1805  * Suspending & resuming Haskell threads.
1806  *
1807  * When making a "safe" call to C (aka _ccall_GC), the task gives back
1808  * its capability before calling the C function.  This allows another
1809  * task to pick up the capability and carry on running Haskell
1810  * threads.  It also means that if the C call blocks, it won't lock
1811  * the whole system.
1812  *
1813  * The Haskell thread making the C call is put to sleep for the
1814  * duration of the call, on the susepended_ccalling_threads queue.  We
1815  * give out a token to the task, which it can use to resume the thread
1816  * on return from the C function.
1817  * ------------------------------------------------------------------------- */
1818
1819 void *
1820 suspendThread (StgRegTable *reg)
1821 {
1822   Capability *cap;
1823   int saved_errno;
1824   StgTSO *tso;
1825   Task *task;
1826 #if mingw32_HOST_OS
1827   StgWord32 saved_winerror;
1828 #endif
1829
1830   saved_errno = errno;
1831 #if mingw32_HOST_OS
1832   saved_winerror = GetLastError();
1833 #endif
1834
1835   /* assume that *reg is a pointer to the StgRegTable part of a Capability.
1836    */
1837   cap = regTableToCapability(reg);
1838
1839   task = cap->running_task;
1840   tso = cap->r.rCurrentTSO;
1841
1842   postEvent(cap, EVENT_STOP_THREAD, tso->id, THREAD_SUSPENDED_FOREIGN_CALL);
1843   debugTrace(DEBUG_sched,
1844              "thread %lu did a safe foreign call",
1845              (unsigned long)cap->r.rCurrentTSO->id);
1846
1847   // XXX this might not be necessary --SDM
1848   tso->what_next = ThreadRunGHC;
1849
1850   threadPaused(cap,tso);
1851
1852   if ((tso->flags & TSO_BLOCKEX) == 0)  {
1853       tso->why_blocked = BlockedOnCCall;
1854       tso->flags |= TSO_BLOCKEX;
1855       tso->flags &= ~TSO_INTERRUPTIBLE;
1856   } else {
1857       tso->why_blocked = BlockedOnCCall_NoUnblockExc;
1858   }
1859
1860   // Hand back capability
1861   task->suspended_tso = tso;
1862
1863   ACQUIRE_LOCK(&cap->lock);
1864
1865   suspendTask(cap,task);
1866   cap->in_haskell = rtsFalse;
1867   releaseCapability_(cap,rtsFalse);
1868
1869   RELEASE_LOCK(&cap->lock);
1870
1871 #if defined(THREADED_RTS)
1872   /* Preparing to leave the RTS, so ensure there's a native thread/task
1873      waiting to take over.
1874   */
1875   debugTrace(DEBUG_sched, "thread %lu: leaving RTS", (unsigned long)tso->id);
1876 #endif
1877
1878   errno = saved_errno;
1879 #if mingw32_HOST_OS
1880   SetLastError(saved_winerror);
1881 #endif
1882   return task;
1883 }
1884
1885 StgRegTable *
1886 resumeThread (void *task_)
1887 {
1888     StgTSO *tso;
1889     Capability *cap;
1890     Task *task = task_;
1891     int saved_errno;
1892 #if mingw32_HOST_OS
1893     StgWord32 saved_winerror;
1894 #endif
1895
1896     saved_errno = errno;
1897 #if mingw32_HOST_OS
1898     saved_winerror = GetLastError();
1899 #endif
1900
1901     cap = task->cap;
1902     // Wait for permission to re-enter the RTS with the result.
1903     waitForReturnCapability(&cap,task);
1904     // we might be on a different capability now... but if so, our
1905     // entry on the suspended_ccalling_tasks list will also have been
1906     // migrated.
1907
1908     // Remove the thread from the suspended list
1909     recoverSuspendedTask(cap,task);
1910
1911     tso = task->suspended_tso;
1912     task->suspended_tso = NULL;
1913     tso->_link = END_TSO_QUEUE; // no write barrier reqd
1914
1915     postEvent(cap, EVENT_RUN_THREAD, tso->id, 0);
1916     debugTrace(DEBUG_sched, "thread %lu: re-entering RTS", (unsigned long)tso->id);
1917
1918     if (tso->why_blocked == BlockedOnCCall) {
1919         // avoid locking the TSO if we don't have to
1920         if (tso->blocked_exceptions != END_TSO_QUEUE) {
1921             awakenBlockedExceptionQueue(cap,tso);
1922         }
1923         tso->flags &= ~(TSO_BLOCKEX | TSO_INTERRUPTIBLE);
1924     }
1925
1926     /* Reset blocking status */
1927     tso->why_blocked  = NotBlocked;
1928
1929     cap->r.rCurrentTSO = tso;
1930     cap->in_haskell = rtsTrue;
1931     errno = saved_errno;
1932 #if mingw32_HOST_OS
1933     SetLastError(saved_winerror);
1934 #endif
1935
1936     /* We might have GC'd, mark the TSO dirty again */
1937     dirty_TSO(cap,tso);
1938
1939     IF_DEBUG(sanity, checkTSO(tso));
1940
1941     return &cap->r;
1942 }
1943
1944 /* ---------------------------------------------------------------------------
1945  * scheduleThread()
1946  *
1947  * scheduleThread puts a thread on the end  of the runnable queue.
1948  * This will usually be done immediately after a thread is created.
1949  * The caller of scheduleThread must create the thread using e.g.
1950  * createThread and push an appropriate closure
1951  * on this thread's stack before the scheduler is invoked.
1952  * ------------------------------------------------------------------------ */
1953
1954 void
1955 scheduleThread(Capability *cap, StgTSO *tso)
1956 {
1957     // The thread goes at the *end* of the run-queue, to avoid possible
1958     // starvation of any threads already on the queue.
1959     appendToRunQueue(cap,tso);
1960 }
1961
1962 void
1963 scheduleThreadOn(Capability *cap, StgWord cpu USED_IF_THREADS, StgTSO *tso)
1964 {
1965 #if defined(THREADED_RTS)
1966     tso->flags |= TSO_LOCKED; // we requested explicit affinity; don't
1967                               // move this thread from now on.
1968     cpu %= RtsFlags.ParFlags.nNodes;
1969     if (cpu == cap->no) {
1970         appendToRunQueue(cap,tso);
1971     } else {
1972         postEvent (cap, EVENT_MIGRATE_THREAD, tso->id, capabilities[cpu].no);
1973         wakeupThreadOnCapability(cap, &capabilities[cpu], tso);
1974     }
1975 #else
1976     appendToRunQueue(cap,tso);
1977 #endif
1978 }
1979
1980 Capability *
1981 scheduleWaitThread (StgTSO* tso, /*[out]*/HaskellObj* ret, Capability *cap)
1982 {
1983     Task *task;
1984
1985     // We already created/initialised the Task
1986     task = cap->running_task;
1987
1988     // This TSO is now a bound thread; make the Task and TSO
1989     // point to each other.
1990     tso->bound = task;
1991     tso->cap = cap;
1992
1993     task->tso = tso;
1994     task->ret = ret;
1995     task->stat = NoStatus;
1996
1997     appendToRunQueue(cap,tso);
1998
1999     debugTrace(DEBUG_sched, "new bound thread (%lu)", (unsigned long)tso->id);
2000
2001     cap = schedule(cap,task);
2002
2003     ASSERT(task->stat != NoStatus);
2004     ASSERT_FULL_CAPABILITY_INVARIANTS(cap,task);
2005
2006     debugTrace(DEBUG_sched, "bound thread (%lu) finished", (unsigned long)task->tso->id);
2007     return cap;
2008 }
2009
2010 /* ----------------------------------------------------------------------------
2011  * Starting Tasks
2012  * ------------------------------------------------------------------------- */
2013
2014 #if defined(THREADED_RTS)
2015 void OSThreadProcAttr
2016 workerStart(Task *task)
2017 {
2018     Capability *cap;
2019
2020     // See startWorkerTask().
2021     ACQUIRE_LOCK(&task->lock);
2022     cap = task->cap;
2023     RELEASE_LOCK(&task->lock);
2024
2025     if (RtsFlags.ParFlags.setAffinity) {
2026         setThreadAffinity(cap->no, n_capabilities);
2027     }
2028
2029     // set the thread-local pointer to the Task:
2030     taskEnter(task);
2031
2032     // schedule() runs without a lock.
2033     cap = schedule(cap,task);
2034
2035     // On exit from schedule(), we have a Capability, but possibly not
2036     // the same one we started with.
2037
2038     // During shutdown, the requirement is that after all the
2039     // Capabilities are shut down, all workers that are shutting down
2040     // have finished workerTaskStop().  This is why we hold on to
2041     // cap->lock until we've finished workerTaskStop() below.
2042     //
2043     // There may be workers still involved in foreign calls; those
2044     // will just block in waitForReturnCapability() because the
2045     // Capability has been shut down.
2046     //
2047     ACQUIRE_LOCK(&cap->lock);
2048     releaseCapability_(cap,rtsFalse);
2049     workerTaskStop(task);
2050     RELEASE_LOCK(&cap->lock);
2051 }
2052 #endif
2053
2054 /* ---------------------------------------------------------------------------
2055  * initScheduler()
2056  *
2057  * Initialise the scheduler.  This resets all the queues - if the
2058  * queues contained any threads, they'll be garbage collected at the
2059  * next pass.
2060  *
2061  * ------------------------------------------------------------------------ */
2062
2063 void
2064 initScheduler(void)
2065 {
2066 #if !defined(THREADED_RTS)
2067   blocked_queue_hd  = END_TSO_QUEUE;
2068   blocked_queue_tl  = END_TSO_QUEUE;
2069   sleeping_queue    = END_TSO_QUEUE;
2070 #endif
2071
2072   blackhole_queue   = END_TSO_QUEUE;
2073
2074   sched_state    = SCHED_RUNNING;
2075   recent_activity = ACTIVITY_YES;
2076
2077 #if defined(THREADED_RTS)
2078   /* Initialise the mutex and condition variables used by
2079    * the scheduler. */
2080   initMutex(&sched_mutex);
2081 #endif
2082
2083   ACQUIRE_LOCK(&sched_mutex);
2084
2085   /* A capability holds the state a native thread needs in
2086    * order to execute STG code. At least one capability is
2087    * floating around (only THREADED_RTS builds have more than one).
2088    */
2089   initCapabilities();
2090
2091   initTaskManager();
2092
2093 #if defined(THREADED_RTS)
2094   initSparkPools();
2095 #endif
2096
2097 #if defined(THREADED_RTS)
2098   /*
2099    * Eagerly start one worker to run each Capability, except for
2100    * Capability 0.  The idea is that we're probably going to start a
2101    * bound thread on Capability 0 pretty soon, so we don't want a
2102    * worker task hogging it.
2103    */
2104   {
2105       nat i;
2106       Capability *cap;
2107       for (i = 1; i < n_capabilities; i++) {
2108           cap = &capabilities[i];
2109           ACQUIRE_LOCK(&cap->lock);
2110           startWorkerTask(cap, workerStart);
2111           RELEASE_LOCK(&cap->lock);
2112       }
2113   }
2114 #endif
2115
2116   RELEASE_LOCK(&sched_mutex);
2117 }
2118
2119 void
2120 exitScheduler(
2121     rtsBool wait_foreign
2122 #if !defined(THREADED_RTS)
2123                          __attribute__((unused))
2124 #endif
2125 )
2126                /* see Capability.c, shutdownCapability() */
2127 {
2128     Task *task = NULL;
2129
2130     task = newBoundTask();
2131
2132     // If we haven't killed all the threads yet, do it now.
2133     if (sched_state < SCHED_SHUTTING_DOWN) {
2134         sched_state = SCHED_INTERRUPTING;
2135         waitForReturnCapability(&task->cap,task);
2136         scheduleDoGC(task->cap,task,rtsFalse);
2137         releaseCapability(task->cap);
2138     }
2139     sched_state = SCHED_SHUTTING_DOWN;
2140
2141 #if defined(THREADED_RTS)
2142     {
2143         nat i;
2144
2145         for (i = 0; i < n_capabilities; i++) {
2146             shutdownCapability(&capabilities[i], task, wait_foreign);
2147         }
2148         boundTaskExiting(task);
2149     }
2150 #endif
2151 }
2152
2153 void
2154 freeScheduler( void )
2155 {
2156     nat still_running;
2157
2158     ACQUIRE_LOCK(&sched_mutex);
2159     still_running = freeTaskManager();
2160     // We can only free the Capabilities if there are no Tasks still
2161     // running.  We might have a Task about to return from a foreign
2162     // call into waitForReturnCapability(), for example (actually,
2163     // this should be the *only* thing that a still-running Task can
2164     // do at this point, and it will block waiting for the
2165     // Capability).
2166     if (still_running == 0) {
2167         freeCapabilities();
2168         if (n_capabilities != 1) {
2169             stgFree(capabilities);
2170         }
2171     }
2172     RELEASE_LOCK(&sched_mutex);
2173 #if defined(THREADED_RTS)
2174     closeMutex(&sched_mutex);
2175 #endif
2176 }
2177
2178 /* -----------------------------------------------------------------------------
2179    performGC
2180
2181    This is the interface to the garbage collector from Haskell land.
2182    We provide this so that external C code can allocate and garbage
2183    collect when called from Haskell via _ccall_GC.
2184    -------------------------------------------------------------------------- */
2185
2186 static void
2187 performGC_(rtsBool force_major)
2188 {
2189     Task *task;
2190
2191     // We must grab a new Task here, because the existing Task may be
2192     // associated with a particular Capability, and chained onto the
2193     // suspended_ccalling_tasks queue.
2194     task = newBoundTask();
2195
2196     waitForReturnCapability(&task->cap,task);
2197     scheduleDoGC(task->cap,task,force_major);
2198     releaseCapability(task->cap);
2199     boundTaskExiting(task);
2200 }
2201
2202 void
2203 performGC(void)
2204 {
2205     performGC_(rtsFalse);
2206 }
2207
2208 void
2209 performMajorGC(void)
2210 {
2211     performGC_(rtsTrue);
2212 }
2213
2214 /* -----------------------------------------------------------------------------
2215    Stack overflow
2216
2217    If the thread has reached its maximum stack size, then raise the
2218    StackOverflow exception in the offending thread.  Otherwise
2219    relocate the TSO into a larger chunk of memory and adjust its stack
2220    size appropriately.
2221    -------------------------------------------------------------------------- */
2222
2223 static StgTSO *
2224 threadStackOverflow(Capability *cap, StgTSO *tso)
2225 {
2226   nat new_stack_size, stack_words;
2227   lnat new_tso_size;
2228   StgPtr new_sp;
2229   StgTSO *dest;
2230
2231   IF_DEBUG(sanity,checkTSO(tso));
2232
2233   // don't allow throwTo() to modify the blocked_exceptions queue
2234   // while we are moving the TSO:
2235   lockClosure((StgClosure *)tso);
2236
2237   if (tso->stack_size >= tso->max_stack_size && !(tso->flags & TSO_BLOCKEX)) {
2238       // NB. never raise a StackOverflow exception if the thread is
2239       // inside Control.Exceptino.block.  It is impractical to protect
2240       // against stack overflow exceptions, since virtually anything
2241       // can raise one (even 'catch'), so this is the only sensible
2242       // thing to do here.  See bug #767.
2243
2244       debugTrace(DEBUG_gc,
2245                  "threadStackOverflow of TSO %ld (%p): stack too large (now %ld; max is %ld)",
2246                  (long)tso->id, tso, (long)tso->stack_size, (long)tso->max_stack_size);
2247       IF_DEBUG(gc,
2248                /* If we're debugging, just print out the top of the stack */
2249                printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size,
2250                                                 tso->sp+64)));
2251
2252       // Send this thread the StackOverflow exception
2253       unlockTSO(tso);
2254       throwToSingleThreaded(cap, tso, (StgClosure *)stackOverflow_closure);
2255       return tso;
2256   }
2257
2258   /* Try to double the current stack size.  If that takes us over the
2259    * maximum stack size for this thread, then use the maximum instead
2260    * (that is, unless we're already at or over the max size and we
2261    * can't raise the StackOverflow exception (see above), in which
2262    * case just double the size). Finally round up so the TSO ends up as
2263    * a whole number of blocks.
2264    */
2265   if (tso->stack_size >= tso->max_stack_size) {
2266       new_stack_size = tso->stack_size * 2;
2267   } else {
2268       new_stack_size = stg_min(tso->stack_size * 2, tso->max_stack_size);
2269   }
2270   new_tso_size   = (lnat)BLOCK_ROUND_UP(new_stack_size * sizeof(W_) +
2271                                        TSO_STRUCT_SIZE)/sizeof(W_);
2272   new_tso_size = round_to_mblocks(new_tso_size);  /* Be MBLOCK-friendly */
2273   new_stack_size = new_tso_size - TSO_STRUCT_SIZEW;
2274
2275   debugTrace(DEBUG_sched,
2276              "increasing stack size from %ld words to %d.",
2277              (long)tso->stack_size, new_stack_size);
2278
2279   dest = (StgTSO *)allocateLocal(cap,new_tso_size);
2280   TICK_ALLOC_TSO(new_stack_size,0);
2281
2282   /* copy the TSO block and the old stack into the new area */
2283   memcpy(dest,tso,TSO_STRUCT_SIZE);
2284   stack_words = tso->stack + tso->stack_size - tso->sp;
2285   new_sp = (P_)dest + new_tso_size - stack_words;
2286   memcpy(new_sp, tso->sp, stack_words * sizeof(W_));
2287
2288   /* relocate the stack pointers... */
2289   dest->sp         = new_sp;
2290   dest->stack_size = new_stack_size;
2291
2292   /* Mark the old TSO as relocated.  We have to check for relocated
2293    * TSOs in the garbage collector and any primops that deal with TSOs.
2294    *
2295    * It's important to set the sp value to just beyond the end
2296    * of the stack, so we don't attempt to scavenge any part of the
2297    * dead TSO's stack.
2298    */
2299   tso->what_next = ThreadRelocated;
2300   setTSOLink(cap,tso,dest);
2301   tso->sp = (P_)&(tso->stack[tso->stack_size]);
2302   tso->why_blocked = NotBlocked;
2303
2304   unlockTSO(dest);
2305   unlockTSO(tso);
2306
2307   IF_DEBUG(sanity,checkTSO(dest));
2308 #if 0
2309   IF_DEBUG(scheduler,printTSO(dest));
2310 #endif
2311
2312   return dest;
2313 }
2314
2315 static StgTSO *
2316 threadStackUnderflow (Task *task STG_UNUSED, StgTSO *tso)
2317 {
2318     bdescr *bd, *new_bd;
2319     lnat free_w, tso_size_w;
2320     StgTSO *new_tso;
2321
2322     tso_size_w = tso_sizeW(tso);
2323
2324     if (tso_size_w < MBLOCK_SIZE_W ||
2325           // TSO is less than 2 mblocks (since the first mblock is
2326           // shorter than MBLOCK_SIZE_W)
2327         (tso_size_w - BLOCKS_PER_MBLOCK*BLOCK_SIZE_W) % MBLOCK_SIZE_W != 0 ||
2328           // or TSO is not a whole number of megablocks (ensuring
2329           // precondition of splitLargeBlock() below)
2330         (tso_size_w <= round_up_to_mblocks(RtsFlags.GcFlags.initialStkSize)) ||
2331           // or TSO is smaller than the minimum stack size (rounded up)
2332         (nat)(tso->stack + tso->stack_size - tso->sp) > tso->stack_size / 4)
2333           // or stack is using more than 1/4 of the available space
2334     {
2335         // then do nothing
2336         return tso;
2337     }
2338
2339     // don't allow throwTo() to modify the blocked_exceptions queue
2340     // while we are moving the TSO:
2341     lockClosure((StgClosure *)tso);
2342
2343     // this is the number of words we'll free
2344     free_w = round_to_mblocks(tso_size_w/2);
2345
2346     bd = Bdescr((StgPtr)tso);
2347     new_bd = splitLargeBlock(bd, free_w / BLOCK_SIZE_W);
2348     bd->free = bd->start + TSO_STRUCT_SIZEW;
2349
2350     new_tso = (StgTSO *)new_bd->start;
2351     memcpy(new_tso,tso,TSO_STRUCT_SIZE);
2352     new_tso->stack_size = new_bd->free - new_tso->stack;
2353
2354     debugTrace(DEBUG_sched, "thread %ld: reducing TSO size from %lu words to %lu",
2355                (long)tso->id, tso_size_w, tso_sizeW(new_tso));
2356
2357     tso->what_next = ThreadRelocated;
2358     tso->_link = new_tso; // no write barrier reqd: same generation
2359
2360     // The TSO attached to this Task may have moved, so update the
2361     // pointer to it.
2362     if (task->tso == tso) {
2363         task->tso = new_tso;
2364     }
2365
2366     unlockTSO(new_tso);
2367     unlockTSO(tso);
2368
2369     IF_DEBUG(sanity,checkTSO(new_tso));
2370
2371     return new_tso;
2372 }
2373
2374 /* ---------------------------------------------------------------------------
2375    Interrupt execution
2376    - usually called inside a signal handler so it mustn't do anything fancy.
2377    ------------------------------------------------------------------------ */
2378
2379 void
2380 interruptStgRts(void)
2381 {
2382     sched_state = SCHED_INTERRUPTING;
2383     setContextSwitches();
2384     wakeUpRts();
2385 }
2386
2387 /* -----------------------------------------------------------------------------
2388    Wake up the RTS
2389
2390    This function causes at least one OS thread to wake up and run the
2391    scheduler loop.  It is invoked when the RTS might be deadlocked, or
2392    an external event has arrived that may need servicing (eg. a
2393    keyboard interrupt).
2394
2395    In the single-threaded RTS we don't do anything here; we only have
2396    one thread anyway, and the event that caused us to want to wake up
2397    will have interrupted any blocking system call in progress anyway.
2398    -------------------------------------------------------------------------- */
2399
2400 void
2401 wakeUpRts(void)
2402 {
2403 #if defined(THREADED_RTS)
2404     // This forces the IO Manager thread to wakeup, which will
2405     // in turn ensure that some OS thread wakes up and runs the
2406     // scheduler loop, which will cause a GC and deadlock check.
2407     ioManagerWakeup();
2408 #endif
2409 }
2410
2411 /* -----------------------------------------------------------------------------
2412  * checkBlackHoles()
2413  *
2414  * Check the blackhole_queue for threads that can be woken up.  We do
2415  * this periodically: before every GC, and whenever the run queue is
2416  * empty.
2417  *
2418  * An elegant solution might be to just wake up all the blocked
2419  * threads with awakenBlockedQueue occasionally: they'll go back to
2420  * sleep again if the object is still a BLACKHOLE.  Unfortunately this
2421  * doesn't give us a way to tell whether we've actually managed to
2422  * wake up any threads, so we would be busy-waiting.
2423  *
2424  * -------------------------------------------------------------------------- */
2425
2426 static rtsBool
2427 checkBlackHoles (Capability *cap)
2428 {
2429     StgTSO **prev, *t;
2430     rtsBool any_woke_up = rtsFalse;
2431     StgHalfWord type;
2432
2433     // blackhole_queue is global:
2434     ASSERT_LOCK_HELD(&sched_mutex);
2435
2436     debugTrace(DEBUG_sched, "checking threads blocked on black holes");
2437
2438     // ASSUMES: sched_mutex
2439     prev = &blackhole_queue;
2440     t = blackhole_queue;
2441     while (t != END_TSO_QUEUE) {
2442         if (t->what_next == ThreadRelocated) {
2443             t = t->_link;
2444             continue;
2445         }
2446         ASSERT(t->why_blocked == BlockedOnBlackHole);
2447         type = get_itbl(UNTAG_CLOSURE(t->block_info.closure))->type;
2448         if (type != BLACKHOLE && type != CAF_BLACKHOLE) {
2449             IF_DEBUG(sanity,checkTSO(t));
2450             t = unblockOne(cap, t);
2451             *prev = t;
2452             any_woke_up = rtsTrue;
2453         } else {
2454             prev = &t->_link;
2455             t = t->_link;
2456         }
2457     }
2458
2459     return any_woke_up;
2460 }
2461
2462 /* -----------------------------------------------------------------------------
2463    Deleting threads
2464
2465    This is used for interruption (^C) and forking, and corresponds to
2466    raising an exception but without letting the thread catch the
2467    exception.
2468    -------------------------------------------------------------------------- */
2469
2470 static void
2471 deleteThread (Capability *cap, StgTSO *tso)
2472 {
2473     // NOTE: must only be called on a TSO that we have exclusive
2474     // access to, because we will call throwToSingleThreaded() below.
2475     // The TSO must be on the run queue of the Capability we own, or
2476     // we must own all Capabilities.
2477
2478     if (tso->why_blocked != BlockedOnCCall &&
2479         tso->why_blocked != BlockedOnCCall_NoUnblockExc) {
2480         throwToSingleThreaded(cap,tso,NULL);
2481     }
2482 }
2483
2484 #ifdef FORKPROCESS_PRIMOP_SUPPORTED
2485 static void
2486 deleteThread_(Capability *cap, StgTSO *tso)
2487 { // for forkProcess only:
2488   // like deleteThread(), but we delete threads in foreign calls, too.
2489
2490     if (tso->why_blocked == BlockedOnCCall ||
2491         tso->why_blocked == BlockedOnCCall_NoUnblockExc) {
2492         unblockOne(cap,tso);
2493         tso->what_next = ThreadKilled;
2494     } else {
2495         deleteThread(cap,tso);
2496     }
2497 }
2498 #endif
2499
2500 /* -----------------------------------------------------------------------------
2501    raiseExceptionHelper
2502
2503    This function is called by the raise# primitve, just so that we can
2504    move some of the tricky bits of raising an exception from C-- into
2505    C.  Who knows, it might be a useful re-useable thing here too.
2506    -------------------------------------------------------------------------- */
2507
2508 StgWord
2509 raiseExceptionHelper (StgRegTable *reg, StgTSO *tso, StgClosure *exception)
2510 {
2511     Capability *cap = regTableToCapability(reg);
2512     StgThunk *raise_closure = NULL;
2513     StgPtr p, next;
2514     StgRetInfoTable *info;
2515     //
2516     // This closure represents the expression 'raise# E' where E
2517     // is the exception raise.  It is used to overwrite all the
2518     // thunks which are currently under evaluataion.
2519     //
2520
2521     // OLD COMMENT (we don't have MIN_UPD_SIZE now):
2522     // LDV profiling: stg_raise_info has THUNK as its closure
2523     // type. Since a THUNK takes at least MIN_UPD_SIZE words in its
2524     // payload, MIN_UPD_SIZE is more approprate than 1.  It seems that
2525     // 1 does not cause any problem unless profiling is performed.
2526     // However, when LDV profiling goes on, we need to linearly scan
2527     // small object pool, where raise_closure is stored, so we should
2528     // use MIN_UPD_SIZE.
2529     //
2530     // raise_closure = (StgClosure *)RET_STGCALL1(P_,allocate,
2531     //                                 sizeofW(StgClosure)+1);
2532     //
2533
2534     //
2535     // Walk up the stack, looking for the catch frame.  On the way,
2536     // we update any closures pointed to from update frames with the
2537     // raise closure that we just built.
2538     //
2539     p = tso->sp;
2540     while(1) {
2541         info = get_ret_itbl((StgClosure *)p);
2542         next = p + stack_frame_sizeW((StgClosure *)p);
2543         switch (info->i.type) {
2544
2545         case UPDATE_FRAME:
2546             // Only create raise_closure if we need to.
2547             if (raise_closure == NULL) {
2548                 raise_closure =
2549                     (StgThunk *)allocateLocal(cap,sizeofW(StgThunk)+1);
2550                 SET_HDR(raise_closure, &stg_raise_info, CCCS);
2551                 raise_closure->payload[0] = exception;
2552             }
2553             UPD_IND(((StgUpdateFrame *)p)->updatee,(StgClosure *)raise_closure);
2554             p = next;
2555             continue;
2556
2557         case ATOMICALLY_FRAME:
2558             debugTrace(DEBUG_stm, "found ATOMICALLY_FRAME at %p", p);
2559             tso->sp = p;
2560             return ATOMICALLY_FRAME;
2561
2562         case CATCH_FRAME:
2563             tso->sp = p;
2564             return CATCH_FRAME;
2565
2566         case CATCH_STM_FRAME:
2567             debugTrace(DEBUG_stm, "found CATCH_STM_FRAME at %p", p);
2568             tso->sp = p;
2569             return CATCH_STM_FRAME;
2570
2571         case STOP_FRAME:
2572             tso->sp = p;
2573             return STOP_FRAME;
2574
2575         case CATCH_RETRY_FRAME:
2576         default:
2577             p = next;
2578             continue;
2579         }
2580     }
2581 }
2582
2583
2584 /* -----------------------------------------------------------------------------
2585    findRetryFrameHelper
2586
2587    This function is called by the retry# primitive.  It traverses the stack
2588    leaving tso->sp referring to the frame which should handle the retry.
2589
2590    This should either be a CATCH_RETRY_FRAME (if the retry# is within an orElse#)
2591    or should be a ATOMICALLY_FRAME (if the retry# reaches the top level).
2592
2593    We skip CATCH_STM_FRAMEs (aborting and rolling back the nested tx that they
2594    create) because retries are not considered to be exceptions, despite the
2595    similar implementation.
2596
2597    We should not expect to see CATCH_FRAME or STOP_FRAME because those should
2598    not be created within memory transactions.
2599    -------------------------------------------------------------------------- */
2600
2601 StgWord
2602 findRetryFrameHelper (StgTSO *tso)
2603 {
2604   StgPtr           p, next;
2605   StgRetInfoTable *info;
2606
2607   p = tso -> sp;
2608   while (1) {
2609     info = get_ret_itbl((StgClosure *)p);
2610     next = p + stack_frame_sizeW((StgClosure *)p);
2611     switch (info->i.type) {
2612
2613     case ATOMICALLY_FRAME:
2614         debugTrace(DEBUG_stm,
2615                    "found ATOMICALLY_FRAME at %p during retry", p);
2616         tso->sp = p;
2617         return ATOMICALLY_FRAME;
2618
2619     case CATCH_RETRY_FRAME:
2620         debugTrace(DEBUG_stm,
2621                    "found CATCH_RETRY_FRAME at %p during retrry", p);
2622         tso->sp = p;
2623         return CATCH_RETRY_FRAME;
2624
2625     case CATCH_STM_FRAME: {
2626         StgTRecHeader *trec = tso -> trec;
2627         StgTRecHeader *outer = stmGetEnclosingTRec(trec);
2628         debugTrace(DEBUG_stm,
2629                    "found CATCH_STM_FRAME at %p during retry", p);
2630         debugTrace(DEBUG_stm, "trec=%p outer=%p", trec, outer);
2631         stmAbortTransaction(tso -> cap, trec);
2632         stmFreeAbortedTRec(tso -> cap, trec);
2633         tso -> trec = outer;
2634         p = next;
2635         continue;
2636     }
2637
2638
2639     default:
2640       ASSERT(info->i.type != CATCH_FRAME);
2641       ASSERT(info->i.type != STOP_FRAME);
2642       p = next;
2643       continue;
2644     }
2645   }
2646 }
2647
2648 /* -----------------------------------------------------------------------------
2649    resurrectThreads is called after garbage collection on the list of
2650    threads found to be garbage.  Each of these threads will be woken
2651    up and sent a signal: BlockedOnDeadMVar if the thread was blocked
2652    on an MVar, or NonTermination if the thread was blocked on a Black
2653    Hole.
2654
2655    Locks: assumes we hold *all* the capabilities.
2656    -------------------------------------------------------------------------- */
2657
2658 void
2659 resurrectThreads (StgTSO *threads)
2660 {
2661     StgTSO *tso, *next;
2662     Capability *cap;
2663     step *step;
2664
2665     for (tso = threads; tso != END_TSO_QUEUE; tso = next) {
2666         next = tso->global_link;
2667
2668         step = Bdescr((P_)tso)->step;
2669         tso->global_link = step->threads;
2670         step->threads = tso;
2671
2672         debugTrace(DEBUG_sched, "resurrecting thread %lu", (unsigned long)tso->id);
2673
2674         // Wake up the thread on the Capability it was last on
2675         cap = tso->cap;
2676
2677         switch (tso->why_blocked) {
2678         case BlockedOnMVar:
2679         case BlockedOnException:
2680             /* Called by GC - sched_mutex lock is currently held. */
2681             throwToSingleThreaded(cap, tso,
2682                                   (StgClosure *)blockedOnDeadMVar_closure);
2683             break;
2684         case BlockedOnBlackHole:
2685             throwToSingleThreaded(cap, tso,
2686                                   (StgClosure *)nonTermination_closure);
2687             break;
2688         case BlockedOnSTM:
2689             throwToSingleThreaded(cap, tso,
2690                                   (StgClosure *)blockedIndefinitely_closure);
2691             break;
2692         case NotBlocked:
2693             /* This might happen if the thread was blocked on a black hole
2694              * belonging to a thread that we've just woken up (raiseAsync
2695              * can wake up threads, remember...).
2696              */
2697             continue;
2698         default:
2699             barf("resurrectThreads: thread blocked in a strange way");
2700         }
2701     }
2702 }
2703
2704 /* -----------------------------------------------------------------------------
2705    performPendingThrowTos is called after garbage collection, and
2706    passed a list of threads that were found to have pending throwTos
2707    (tso->blocked_exceptions was not empty), and were blocked.
2708    Normally this doesn't happen, because we would deliver the
2709    exception directly if the target thread is blocked, but there are
2710    small windows where it might occur on a multiprocessor (see
2711    throwTo()).
2712
2713    NB. we must be holding all the capabilities at this point, just
2714    like resurrectThreads().
2715    -------------------------------------------------------------------------- */
2716
2717 void
2718 performPendingThrowTos (StgTSO *threads)
2719 {
2720     StgTSO *tso, *next;
2721     Capability *cap;
2722     step *step;
2723
2724     for (tso = threads; tso != END_TSO_QUEUE; tso = next) {
2725         next = tso->global_link;
2726
2727         step = Bdescr((P_)tso)->step;
2728         tso->global_link = step->threads;
2729         step->threads = tso;
2730
2731         debugTrace(DEBUG_sched, "performing blocked throwTo to thread %lu", (unsigned long)tso->id);
2732
2733         cap = tso->cap;
2734         maybePerformBlockedException(cap, tso);
2735     }
2736 }