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