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