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       //
1406       // Check whether the thread that just completed was a bound
1407       // thread, and if so return with the result.
1408       //
1409       // There is an assumption here that all thread completion goes
1410       // through this point; we need to make sure that if a thread
1411       // ends up in the ThreadKilled state, that it stays on the run
1412       // queue so it can be dealt with here.
1413       //
1414
1415       if (t->bound) {
1416
1417           if (t->bound != task) {
1418 #if !defined(THREADED_RTS)
1419               // Must be a bound thread that is not the topmost one.  Leave
1420               // it on the run queue until the stack has unwound to the
1421               // point where we can deal with this.  Leaving it on the run
1422               // queue also ensures that the garbage collector knows about
1423               // this thread and its return value (it gets dropped from the
1424               // step->threads list so there's no other way to find it).
1425               appendToRunQueue(cap,t);
1426               return rtsFalse;
1427 #else
1428               // this cannot happen in the threaded RTS, because a
1429               // bound thread can only be run by the appropriate Task.
1430               barf("finished bound thread that isn't mine");
1431 #endif
1432           }
1433
1434           ASSERT(task->tso == t);
1435
1436           if (t->what_next == ThreadComplete) {
1437               if (task->ret) {
1438                   // NOTE: return val is tso->sp[1] (see StgStartup.hc)
1439                   *(task->ret) = (StgClosure *)task->tso->sp[1];
1440               }
1441               task->stat = Success;
1442           } else {
1443               if (task->ret) {
1444                   *(task->ret) = NULL;
1445               }
1446               if (sched_state >= SCHED_INTERRUPTING) {
1447                   if (heap_overflow) {
1448                       task->stat = HeapExhausted;
1449                   } else {
1450                       task->stat = Interrupted;
1451                   }
1452               } else {
1453                   task->stat = Killed;
1454               }
1455           }
1456 #ifdef DEBUG
1457           removeThreadLabel((StgWord)task->tso->id);
1458 #endif
1459           return rtsTrue; // tells schedule() to return
1460       }
1461
1462       return rtsFalse;
1463 }
1464
1465 /* -----------------------------------------------------------------------------
1466  * Perform a heap census
1467  * -------------------------------------------------------------------------- */
1468
1469 static rtsBool
1470 scheduleNeedHeapProfile( rtsBool ready_to_gc STG_UNUSED )
1471 {
1472     // When we have +RTS -i0 and we're heap profiling, do a census at
1473     // every GC.  This lets us get repeatable runs for debugging.
1474     if (performHeapProfile ||
1475         (RtsFlags.ProfFlags.profileInterval==0 &&
1476          RtsFlags.ProfFlags.doHeapProfile && ready_to_gc)) {
1477         return rtsTrue;
1478     } else {
1479         return rtsFalse;
1480     }
1481 }
1482
1483 /* -----------------------------------------------------------------------------
1484  * Perform a garbage collection if necessary
1485  * -------------------------------------------------------------------------- */
1486
1487 static Capability *
1488 scheduleDoGC (Capability *cap, Task *task USED_IF_THREADS, rtsBool force_major)
1489 {
1490     rtsBool heap_census;
1491 #ifdef THREADED_RTS
1492     /* extern static volatile StgWord waiting_for_gc;
1493        lives inside capability.c */
1494     rtsBool gc_type, prev_pending_gc;
1495     nat i;
1496 #endif
1497
1498     if (sched_state == SCHED_SHUTTING_DOWN) {
1499         // The final GC has already been done, and the system is
1500         // shutting down.  We'll probably deadlock if we try to GC
1501         // now.
1502         return cap;
1503     }
1504
1505 #ifdef THREADED_RTS
1506     if (sched_state < SCHED_INTERRUPTING
1507         && RtsFlags.ParFlags.parGcEnabled
1508         && N >= RtsFlags.ParFlags.parGcGen
1509         && ! oldest_gen->steps[0].mark)
1510     {
1511         gc_type = PENDING_GC_PAR;
1512     } else {
1513         gc_type = PENDING_GC_SEQ;
1514     }
1515
1516     // In order to GC, there must be no threads running Haskell code.
1517     // Therefore, the GC thread needs to hold *all* the capabilities,
1518     // and release them after the GC has completed.
1519     //
1520     // This seems to be the simplest way: previous attempts involved
1521     // making all the threads with capabilities give up their
1522     // capabilities and sleep except for the *last* one, which
1523     // actually did the GC.  But it's quite hard to arrange for all
1524     // the other tasks to sleep and stay asleep.
1525     //
1526
1527     /*  Other capabilities are prevented from running yet more Haskell
1528         threads if waiting_for_gc is set. Tested inside
1529         yieldCapability() and releaseCapability() in Capability.c */
1530
1531     prev_pending_gc = cas(&waiting_for_gc, 0, gc_type);
1532     if (prev_pending_gc) {
1533         do {
1534             debugTrace(DEBUG_sched, "someone else is trying to GC (%d)...",
1535                        prev_pending_gc);
1536             ASSERT(cap);
1537             yieldCapability(&cap,task);
1538         } while (waiting_for_gc);
1539         return cap;  // NOTE: task->cap might have changed here
1540     }
1541
1542     setContextSwitches();
1543
1544     // The final shutdown GC is always single-threaded, because it's
1545     // possible that some of the Capabilities have no worker threads.
1546
1547     if (gc_type == PENDING_GC_SEQ)
1548     {
1549         // single-threaded GC: grab all the capabilities
1550         for (i=0; i < n_capabilities; i++) {
1551             debugTrace(DEBUG_sched, "ready_to_gc, grabbing all the capabilies (%d/%d)", i, n_capabilities);
1552             if (cap != &capabilities[i]) {
1553                 Capability *pcap = &capabilities[i];
1554                 // we better hope this task doesn't get migrated to
1555                 // another Capability while we're waiting for this one.
1556                 // It won't, because load balancing happens while we have
1557                 // all the Capabilities, but even so it's a slightly
1558                 // unsavoury invariant.
1559                 task->cap = pcap;
1560                 waitForReturnCapability(&pcap, task);
1561                 if (pcap != &capabilities[i]) {
1562                     barf("scheduleDoGC: got the wrong capability");
1563                 }
1564             }
1565         }
1566     }
1567     else
1568     {
1569         // multi-threaded GC: make sure all the Capabilities donate one
1570         // GC thread each.
1571         debugTrace(DEBUG_sched, "ready_to_gc, grabbing GC threads");
1572
1573         waitForGcThreads(cap);
1574     }
1575 #endif
1576
1577     // so this happens periodically:
1578     if (cap) scheduleCheckBlackHoles(cap);
1579
1580     IF_DEBUG(scheduler, printAllThreads());
1581
1582 delete_threads_and_gc:
1583     /*
1584      * We now have all the capabilities; if we're in an interrupting
1585      * state, then we should take the opportunity to delete all the
1586      * threads in the system.
1587      */
1588     if (sched_state == SCHED_INTERRUPTING) {
1589         deleteAllThreads(cap);
1590         sched_state = SCHED_SHUTTING_DOWN;
1591     }
1592
1593     heap_census = scheduleNeedHeapProfile(rtsTrue);
1594
1595 #if defined(THREADED_RTS)
1596     debugTrace(DEBUG_sched, "doing GC");
1597     // reset waiting_for_gc *before* GC, so that when the GC threads
1598     // emerge they don't immediately re-enter the GC.
1599     waiting_for_gc = 0;
1600     GarbageCollect(force_major || heap_census, gc_type, cap);
1601 #else
1602     GarbageCollect(force_major || heap_census, 0, cap);
1603 #endif
1604
1605     if (heap_census) {
1606         debugTrace(DEBUG_sched, "performing heap census");
1607         heapCensus();
1608         performHeapProfile = rtsFalse;
1609     }
1610
1611     if (heap_overflow && sched_state < SCHED_INTERRUPTING) {
1612         // GC set the heap_overflow flag, so we should proceed with
1613         // an orderly shutdown now.  Ultimately we want the main
1614         // thread to return to its caller with HeapExhausted, at which
1615         // point the caller should call hs_exit().  The first step is
1616         // to delete all the threads.
1617         //
1618         // Another way to do this would be to raise an exception in
1619         // the main thread, which we really should do because it gives
1620         // the program a chance to clean up.  But how do we find the
1621         // main thread?  It should presumably be the same one that
1622         // gets ^C exceptions, but that's all done on the Haskell side
1623         // (GHC.TopHandler).
1624         sched_state = SCHED_INTERRUPTING;
1625         goto delete_threads_and_gc;
1626     }
1627
1628 #ifdef SPARKBALANCE
1629     /* JB
1630        Once we are all together... this would be the place to balance all
1631        spark pools. No concurrent stealing or adding of new sparks can
1632        occur. Should be defined in Sparks.c. */
1633     balanceSparkPoolsCaps(n_capabilities, capabilities);
1634 #endif
1635
1636     if (force_major)
1637     {
1638         // We've just done a major GC and we don't need the timer
1639         // signal turned on any more (#1623).
1640         // NB. do this *before* releasing the Capabilities, to avoid
1641         // deadlocks!
1642         recent_activity = ACTIVITY_DONE_GC;
1643         stopTimer();
1644     }
1645
1646 #if defined(THREADED_RTS)
1647     if (gc_type == PENDING_GC_SEQ) {
1648         // release our stash of capabilities.
1649         for (i = 0; i < n_capabilities; i++) {
1650             if (cap != &capabilities[i]) {
1651                 task->cap = &capabilities[i];
1652                 releaseCapability(&capabilities[i]);
1653             }
1654         }
1655     }
1656     if (cap) {
1657         task->cap = cap;
1658     } else {
1659         task->cap = NULL;
1660     }
1661 #endif
1662
1663     return cap;
1664 }
1665
1666 /* ---------------------------------------------------------------------------
1667  * Singleton fork(). Do not copy any running threads.
1668  * ------------------------------------------------------------------------- */
1669
1670 pid_t
1671 forkProcess(HsStablePtr *entry
1672 #ifndef FORKPROCESS_PRIMOP_SUPPORTED
1673             STG_UNUSED
1674 #endif
1675            )
1676 {
1677 #ifdef FORKPROCESS_PRIMOP_SUPPORTED
1678     Task *task;
1679     pid_t pid;
1680     StgTSO* t,*next;
1681     Capability *cap;
1682     nat s;
1683
1684 #if defined(THREADED_RTS)
1685     if (RtsFlags.ParFlags.nNodes > 1) {
1686         errorBelch("forking not supported with +RTS -N<n> greater than 1");
1687         stg_exit(EXIT_FAILURE);
1688     }
1689 #endif
1690
1691     debugTrace(DEBUG_sched, "forking!");
1692
1693     // ToDo: for SMP, we should probably acquire *all* the capabilities
1694     cap = rts_lock();
1695
1696     // no funny business: hold locks while we fork, otherwise if some
1697     // other thread is holding a lock when the fork happens, the data
1698     // structure protected by the lock will forever be in an
1699     // inconsistent state in the child.  See also #1391.
1700     ACQUIRE_LOCK(&sched_mutex);
1701     ACQUIRE_LOCK(&cap->lock);
1702     ACQUIRE_LOCK(&cap->running_task->lock);
1703
1704     pid = fork();
1705
1706     if (pid) { // parent
1707
1708         RELEASE_LOCK(&sched_mutex);
1709         RELEASE_LOCK(&cap->lock);
1710         RELEASE_LOCK(&cap->running_task->lock);
1711
1712         // just return the pid
1713         rts_unlock(cap);
1714         return pid;
1715
1716     } else { // child
1717
1718 #if defined(THREADED_RTS)
1719         initMutex(&sched_mutex);
1720         initMutex(&cap->lock);
1721         initMutex(&cap->running_task->lock);
1722 #endif
1723
1724         // Now, all OS threads except the thread that forked are
1725         // stopped.  We need to stop all Haskell threads, including
1726         // those involved in foreign calls.  Also we need to delete
1727         // all Tasks, because they correspond to OS threads that are
1728         // now gone.
1729
1730         for (s = 0; s < total_steps; s++) {
1731           for (t = all_steps[s].threads; t != END_TSO_QUEUE; t = next) {
1732             if (t->what_next == ThreadRelocated) {
1733                 next = t->_link;
1734             } else {
1735                 next = t->global_link;
1736                 // don't allow threads to catch the ThreadKilled
1737                 // exception, but we do want to raiseAsync() because these
1738                 // threads may be evaluating thunks that we need later.
1739                 deleteThread_(cap,t);
1740             }
1741           }
1742         }
1743
1744         // Empty the run queue.  It seems tempting to let all the
1745         // killed threads stay on the run queue as zombies to be
1746         // cleaned up later, but some of them correspond to bound
1747         // threads for which the corresponding Task does not exist.
1748         cap->run_queue_hd = END_TSO_QUEUE;
1749         cap->run_queue_tl = END_TSO_QUEUE;
1750
1751         // Any suspended C-calling Tasks are no more, their OS threads
1752         // don't exist now:
1753         cap->suspended_ccalling_tasks = NULL;
1754
1755         // Empty the threads lists.  Otherwise, the garbage
1756         // collector may attempt to resurrect some of these threads.
1757         for (s = 0; s < total_steps; s++) {
1758             all_steps[s].threads = END_TSO_QUEUE;
1759         }
1760
1761         // Wipe the task list, except the current Task.
1762         ACQUIRE_LOCK(&sched_mutex);
1763         for (task = all_tasks; task != NULL; task=task->all_link) {
1764             if (task != cap->running_task) {
1765 #if defined(THREADED_RTS)
1766                 initMutex(&task->lock); // see #1391
1767 #endif
1768                 discardTask(task);
1769             }
1770         }
1771         RELEASE_LOCK(&sched_mutex);
1772
1773 #if defined(THREADED_RTS)
1774         // Wipe our spare workers list, they no longer exist.  New
1775         // workers will be created if necessary.
1776         cap->spare_workers = NULL;
1777         cap->returning_tasks_hd = NULL;
1778         cap->returning_tasks_tl = NULL;
1779 #endif
1780
1781         // On Unix, all timers are reset in the child, so we need to start
1782         // the timer again.
1783         initTimer();
1784         startTimer();
1785
1786         cap = rts_evalStableIO(cap, entry, NULL);  // run the action
1787         rts_checkSchedStatus("forkProcess",cap);
1788
1789         rts_unlock(cap);
1790         hs_exit();                      // clean up and exit
1791         stg_exit(EXIT_SUCCESS);
1792     }
1793 #else /* !FORKPROCESS_PRIMOP_SUPPORTED */
1794     barf("forkProcess#: primop not supported on this platform, sorry!\n");
1795     return -1;
1796 #endif
1797 }
1798
1799 /* ---------------------------------------------------------------------------
1800  * Delete all the threads in the system
1801  * ------------------------------------------------------------------------- */
1802
1803 static void
1804 deleteAllThreads ( Capability *cap )
1805 {
1806     // NOTE: only safe to call if we own all capabilities.
1807
1808     StgTSO* t, *next;
1809     nat s;
1810
1811     debugTrace(DEBUG_sched,"deleting all threads");
1812     for (s = 0; s < total_steps; s++) {
1813       for (t = all_steps[s].threads; t != END_TSO_QUEUE; t = next) {
1814         if (t->what_next == ThreadRelocated) {
1815             next = t->_link;
1816         } else {
1817             next = t->global_link;
1818             deleteThread(cap,t);
1819         }
1820       }
1821     }
1822
1823     // The run queue now contains a bunch of ThreadKilled threads.  We
1824     // must not throw these away: the main thread(s) will be in there
1825     // somewhere, and the main scheduler loop has to deal with it.
1826     // Also, the run queue is the only thing keeping these threads from
1827     // being GC'd, and we don't want the "main thread has been GC'd" panic.
1828
1829 #if !defined(THREADED_RTS)
1830     ASSERT(blocked_queue_hd == END_TSO_QUEUE);
1831     ASSERT(sleeping_queue == END_TSO_QUEUE);
1832 #endif
1833 }
1834
1835 /* -----------------------------------------------------------------------------
1836    Managing the suspended_ccalling_tasks list.
1837    Locks required: sched_mutex
1838    -------------------------------------------------------------------------- */
1839
1840 STATIC_INLINE void
1841 suspendTask (Capability *cap, Task *task)
1842 {
1843     ASSERT(task->next == NULL && task->prev == NULL);
1844     task->next = cap->suspended_ccalling_tasks;
1845     task->prev = NULL;
1846     if (cap->suspended_ccalling_tasks) {
1847         cap->suspended_ccalling_tasks->prev = task;
1848     }
1849     cap->suspended_ccalling_tasks = task;
1850 }
1851
1852 STATIC_INLINE void
1853 recoverSuspendedTask (Capability *cap, Task *task)
1854 {
1855     if (task->prev) {
1856         task->prev->next = task->next;
1857     } else {
1858         ASSERT(cap->suspended_ccalling_tasks == task);
1859         cap->suspended_ccalling_tasks = task->next;
1860     }
1861     if (task->next) {
1862         task->next->prev = task->prev;
1863     }
1864     task->next = task->prev = NULL;
1865 }
1866
1867 /* ---------------------------------------------------------------------------
1868  * Suspending & resuming Haskell threads.
1869  *
1870  * When making a "safe" call to C (aka _ccall_GC), the task gives back
1871  * its capability before calling the C function.  This allows another
1872  * task to pick up the capability and carry on running Haskell
1873  * threads.  It also means that if the C call blocks, it won't lock
1874  * the whole system.
1875  *
1876  * The Haskell thread making the C call is put to sleep for the
1877  * duration of the call, on the susepended_ccalling_threads queue.  We
1878  * give out a token to the task, which it can use to resume the thread
1879  * on return from the C function.
1880  * ------------------------------------------------------------------------- */
1881
1882 void *
1883 suspendThread (StgRegTable *reg)
1884 {
1885   Capability *cap;
1886   int saved_errno;
1887   StgTSO *tso;
1888   Task *task;
1889 #if mingw32_HOST_OS
1890   StgWord32 saved_winerror;
1891 #endif
1892
1893   saved_errno = errno;
1894 #if mingw32_HOST_OS
1895   saved_winerror = GetLastError();
1896 #endif
1897
1898   /* assume that *reg is a pointer to the StgRegTable part of a Capability.
1899    */
1900   cap = regTableToCapability(reg);
1901
1902   task = cap->running_task;
1903   tso = cap->r.rCurrentTSO;
1904
1905   debugTrace(DEBUG_sched,
1906              "thread %lu did a safe foreign call",
1907              (unsigned long)cap->r.rCurrentTSO->id);
1908
1909   // XXX this might not be necessary --SDM
1910   tso->what_next = ThreadRunGHC;
1911
1912   threadPaused(cap,tso);
1913
1914   if ((tso->flags & TSO_BLOCKEX) == 0)  {
1915       tso->why_blocked = BlockedOnCCall;
1916       tso->flags |= TSO_BLOCKEX;
1917       tso->flags &= ~TSO_INTERRUPTIBLE;
1918   } else {
1919       tso->why_blocked = BlockedOnCCall_NoUnblockExc;
1920   }
1921
1922   // Hand back capability
1923   task->suspended_tso = tso;
1924
1925   ACQUIRE_LOCK(&cap->lock);
1926
1927   suspendTask(cap,task);
1928   cap->in_haskell = rtsFalse;
1929   releaseCapability_(cap,rtsFalse);
1930
1931   RELEASE_LOCK(&cap->lock);
1932
1933 #if defined(THREADED_RTS)
1934   /* Preparing to leave the RTS, so ensure there's a native thread/task
1935      waiting to take over.
1936   */
1937   debugTrace(DEBUG_sched, "thread %lu: leaving RTS", (unsigned long)tso->id);
1938 #endif
1939
1940   errno = saved_errno;
1941 #if mingw32_HOST_OS
1942   SetLastError(saved_winerror);
1943 #endif
1944   return task;
1945 }
1946
1947 StgRegTable *
1948 resumeThread (void *task_)
1949 {
1950     StgTSO *tso;
1951     Capability *cap;
1952     Task *task = task_;
1953     int saved_errno;
1954 #if mingw32_HOST_OS
1955     StgWord32 saved_winerror;
1956 #endif
1957
1958     saved_errno = errno;
1959 #if mingw32_HOST_OS
1960     saved_winerror = GetLastError();
1961 #endif
1962
1963     cap = task->cap;
1964     // Wait for permission to re-enter the RTS with the result.
1965     waitForReturnCapability(&cap,task);
1966     // we might be on a different capability now... but if so, our
1967     // entry on the suspended_ccalling_tasks list will also have been
1968     // migrated.
1969
1970     // Remove the thread from the suspended list
1971     recoverSuspendedTask(cap,task);
1972
1973     tso = task->suspended_tso;
1974     task->suspended_tso = NULL;
1975     tso->_link = END_TSO_QUEUE; // no write barrier reqd
1976     debugTrace(DEBUG_sched, "thread %lu: re-entering RTS", (unsigned long)tso->id);
1977
1978     if (tso->why_blocked == BlockedOnCCall) {
1979         awakenBlockedExceptionQueue(cap,tso);
1980         tso->flags &= ~(TSO_BLOCKEX | TSO_INTERRUPTIBLE);
1981     }
1982
1983     /* Reset blocking status */
1984     tso->why_blocked  = NotBlocked;
1985
1986     cap->r.rCurrentTSO = tso;
1987     cap->in_haskell = rtsTrue;
1988     errno = saved_errno;
1989 #if mingw32_HOST_OS
1990     SetLastError(saved_winerror);
1991 #endif
1992
1993     /* We might have GC'd, mark the TSO dirty again */
1994     dirty_TSO(cap,tso);
1995
1996     IF_DEBUG(sanity, checkTSO(tso));
1997
1998     return &cap->r;
1999 }
2000
2001 /* ---------------------------------------------------------------------------
2002  * scheduleThread()
2003  *
2004  * scheduleThread puts a thread on the end  of the runnable queue.
2005  * This will usually be done immediately after a thread is created.
2006  * The caller of scheduleThread must create the thread using e.g.
2007  * createThread and push an appropriate closure
2008  * on this thread's stack before the scheduler is invoked.
2009  * ------------------------------------------------------------------------ */
2010
2011 void
2012 scheduleThread(Capability *cap, StgTSO *tso)
2013 {
2014     // The thread goes at the *end* of the run-queue, to avoid possible
2015     // starvation of any threads already on the queue.
2016     appendToRunQueue(cap,tso);
2017 }
2018
2019 void
2020 scheduleThreadOn(Capability *cap, StgWord cpu USED_IF_THREADS, StgTSO *tso)
2021 {
2022 #if defined(THREADED_RTS)
2023     tso->flags |= TSO_LOCKED; // we requested explicit affinity; don't
2024                               // move this thread from now on.
2025     cpu %= RtsFlags.ParFlags.nNodes;
2026     if (cpu == cap->no) {
2027         appendToRunQueue(cap,tso);
2028     } else {
2029         wakeupThreadOnCapability(cap, &capabilities[cpu], tso);
2030     }
2031 #else
2032     appendToRunQueue(cap,tso);
2033 #endif
2034 }
2035
2036 Capability *
2037 scheduleWaitThread (StgTSO* tso, /*[out]*/HaskellObj* ret, Capability *cap)
2038 {
2039     Task *task;
2040
2041     // We already created/initialised the Task
2042     task = cap->running_task;
2043
2044     // This TSO is now a bound thread; make the Task and TSO
2045     // point to each other.
2046     tso->bound = task;
2047     tso->cap = cap;
2048
2049     task->tso = tso;
2050     task->ret = ret;
2051     task->stat = NoStatus;
2052
2053     appendToRunQueue(cap,tso);
2054
2055     debugTrace(DEBUG_sched, "new bound thread (%lu)", (unsigned long)tso->id);
2056
2057     cap = schedule(cap,task);
2058
2059     ASSERT(task->stat != NoStatus);
2060     ASSERT_FULL_CAPABILITY_INVARIANTS(cap,task);
2061
2062     debugTrace(DEBUG_sched, "bound thread (%lu) finished", (unsigned long)task->tso->id);
2063     return cap;
2064 }
2065
2066 /* ----------------------------------------------------------------------------
2067  * Starting Tasks
2068  * ------------------------------------------------------------------------- */
2069
2070 #if defined(THREADED_RTS)
2071 void OSThreadProcAttr
2072 workerStart(Task *task)
2073 {
2074     Capability *cap;
2075
2076     // See startWorkerTask().
2077     ACQUIRE_LOCK(&task->lock);
2078     cap = task->cap;
2079     RELEASE_LOCK(&task->lock);
2080
2081     // set the thread-local pointer to the Task:
2082     taskEnter(task);
2083
2084     // schedule() runs without a lock.
2085     cap = schedule(cap,task);
2086
2087     // On exit from schedule(), we have a Capability, but possibly not
2088     // the same one we started with.
2089
2090     // During shutdown, the requirement is that after all the
2091     // Capabilities are shut down, all workers that are shutting down
2092     // have finished workerTaskStop().  This is why we hold on to
2093     // cap->lock until we've finished workerTaskStop() below.
2094     //
2095     // There may be workers still involved in foreign calls; those
2096     // will just block in waitForReturnCapability() because the
2097     // Capability has been shut down.
2098     //
2099     ACQUIRE_LOCK(&cap->lock);
2100     releaseCapability_(cap,rtsFalse);
2101     workerTaskStop(task);
2102     RELEASE_LOCK(&cap->lock);
2103 }
2104 #endif
2105
2106 /* ---------------------------------------------------------------------------
2107  * initScheduler()
2108  *
2109  * Initialise the scheduler.  This resets all the queues - if the
2110  * queues contained any threads, they'll be garbage collected at the
2111  * next pass.
2112  *
2113  * ------------------------------------------------------------------------ */
2114
2115 void
2116 initScheduler(void)
2117 {
2118 #if !defined(THREADED_RTS)
2119   blocked_queue_hd  = END_TSO_QUEUE;
2120   blocked_queue_tl  = END_TSO_QUEUE;
2121   sleeping_queue    = END_TSO_QUEUE;
2122 #endif
2123
2124   blackhole_queue   = END_TSO_QUEUE;
2125
2126   sched_state    = SCHED_RUNNING;
2127   recent_activity = ACTIVITY_YES;
2128
2129 #if defined(THREADED_RTS)
2130   /* Initialise the mutex and condition variables used by
2131    * the scheduler. */
2132   initMutex(&sched_mutex);
2133 #endif
2134
2135   ACQUIRE_LOCK(&sched_mutex);
2136
2137   /* A capability holds the state a native thread needs in
2138    * order to execute STG code. At least one capability is
2139    * floating around (only THREADED_RTS builds have more than one).
2140    */
2141   initCapabilities();
2142
2143   initTaskManager();
2144
2145 #if defined(THREADED_RTS) || defined(PARALLEL_HASKELL)
2146   initSparkPools();
2147 #endif
2148
2149 #if defined(THREADED_RTS)
2150   /*
2151    * Eagerly start one worker to run each Capability, except for
2152    * Capability 0.  The idea is that we're probably going to start a
2153    * bound thread on Capability 0 pretty soon, so we don't want a
2154    * worker task hogging it.
2155    */
2156   {
2157       nat i;
2158       Capability *cap;
2159       for (i = 1; i < n_capabilities; i++) {
2160           cap = &capabilities[i];
2161           ACQUIRE_LOCK(&cap->lock);
2162           startWorkerTask(cap, workerStart);
2163           RELEASE_LOCK(&cap->lock);
2164       }
2165   }
2166 #endif
2167
2168   trace(TRACE_sched, "start: %d capabilities", n_capabilities);
2169
2170   RELEASE_LOCK(&sched_mutex);
2171 }
2172
2173 void
2174 exitScheduler(
2175     rtsBool wait_foreign
2176 #if !defined(THREADED_RTS)
2177                          __attribute__((unused))
2178 #endif
2179 )
2180                /* see Capability.c, shutdownCapability() */
2181 {
2182     Task *task = NULL;
2183
2184 #if defined(THREADED_RTS)
2185     ACQUIRE_LOCK(&sched_mutex);
2186     task = newBoundTask();
2187     RELEASE_LOCK(&sched_mutex);
2188 #endif
2189
2190     // If we haven't killed all the threads yet, do it now.
2191     if (sched_state < SCHED_SHUTTING_DOWN) {
2192         sched_state = SCHED_INTERRUPTING;
2193 #if defined(THREADED_RTS)
2194         waitForReturnCapability(&task->cap,task);
2195         scheduleDoGC(task->cap,task,rtsFalse);
2196         releaseCapability(task->cap);
2197 #else
2198         scheduleDoGC(&MainCapability,task,rtsFalse);
2199 #endif
2200     }
2201     sched_state = SCHED_SHUTTING_DOWN;
2202
2203 #if defined(THREADED_RTS)
2204     {
2205         nat i;
2206
2207         for (i = 0; i < n_capabilities; i++) {
2208             shutdownCapability(&capabilities[i], task, wait_foreign);
2209         }
2210         boundTaskExiting(task);
2211     }
2212 #endif
2213 }
2214
2215 void
2216 freeScheduler( void )
2217 {
2218     nat still_running;
2219
2220     ACQUIRE_LOCK(&sched_mutex);
2221     still_running = freeTaskManager();
2222     // We can only free the Capabilities if there are no Tasks still
2223     // running.  We might have a Task about to return from a foreign
2224     // call into waitForReturnCapability(), for example (actually,
2225     // this should be the *only* thing that a still-running Task can
2226     // do at this point, and it will block waiting for the
2227     // Capability).
2228     if (still_running == 0) {
2229         freeCapabilities();
2230         if (n_capabilities != 1) {
2231             stgFree(capabilities);
2232         }
2233     }
2234     RELEASE_LOCK(&sched_mutex);
2235 #if defined(THREADED_RTS)
2236     closeMutex(&sched_mutex);
2237 #endif
2238 }
2239
2240 /* -----------------------------------------------------------------------------
2241    performGC
2242
2243    This is the interface to the garbage collector from Haskell land.
2244    We provide this so that external C code can allocate and garbage
2245    collect when called from Haskell via _ccall_GC.
2246    -------------------------------------------------------------------------- */
2247
2248 static void
2249 performGC_(rtsBool force_major)
2250 {
2251     Task *task;
2252
2253     // We must grab a new Task here, because the existing Task may be
2254     // associated with a particular Capability, and chained onto the
2255     // suspended_ccalling_tasks queue.
2256     ACQUIRE_LOCK(&sched_mutex);
2257     task = newBoundTask();
2258     RELEASE_LOCK(&sched_mutex);
2259
2260     waitForReturnCapability(&task->cap,task);
2261     scheduleDoGC(task->cap,task,force_major);
2262     releaseCapability(task->cap);
2263     boundTaskExiting(task);
2264 }
2265
2266 void
2267 performGC(void)
2268 {
2269     performGC_(rtsFalse);
2270 }
2271
2272 void
2273 performMajorGC(void)
2274 {
2275     performGC_(rtsTrue);
2276 }
2277
2278 /* -----------------------------------------------------------------------------
2279    Stack overflow
2280
2281    If the thread has reached its maximum stack size, then raise the
2282    StackOverflow exception in the offending thread.  Otherwise
2283    relocate the TSO into a larger chunk of memory and adjust its stack
2284    size appropriately.
2285    -------------------------------------------------------------------------- */
2286
2287 static StgTSO *
2288 threadStackOverflow(Capability *cap, StgTSO *tso)
2289 {
2290   nat new_stack_size, stack_words;
2291   lnat new_tso_size;
2292   StgPtr new_sp;
2293   StgTSO *dest;
2294
2295   IF_DEBUG(sanity,checkTSO(tso));
2296
2297   // don't allow throwTo() to modify the blocked_exceptions queue
2298   // while we are moving the TSO:
2299   lockClosure((StgClosure *)tso);
2300
2301   if (tso->stack_size >= tso->max_stack_size && !(tso->flags & TSO_BLOCKEX)) {
2302       // NB. never raise a StackOverflow exception if the thread is
2303       // inside Control.Exceptino.block.  It is impractical to protect
2304       // against stack overflow exceptions, since virtually anything
2305       // can raise one (even 'catch'), so this is the only sensible
2306       // thing to do here.  See bug #767.
2307
2308       debugTrace(DEBUG_gc,
2309                  "threadStackOverflow of TSO %ld (%p): stack too large (now %ld; max is %ld)",
2310                  (long)tso->id, tso, (long)tso->stack_size, (long)tso->max_stack_size);
2311       IF_DEBUG(gc,
2312                /* If we're debugging, just print out the top of the stack */
2313                printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size,
2314                                                 tso->sp+64)));
2315
2316       // Send this thread the StackOverflow exception
2317       unlockTSO(tso);
2318       throwToSingleThreaded(cap, tso, (StgClosure *)stackOverflow_closure);
2319       return tso;
2320   }
2321
2322   /* Try to double the current stack size.  If that takes us over the
2323    * maximum stack size for this thread, then use the maximum instead
2324    * (that is, unless we're already at or over the max size and we
2325    * can't raise the StackOverflow exception (see above), in which
2326    * case just double the size). Finally round up so the TSO ends up as
2327    * a whole number of blocks.
2328    */
2329   if (tso->stack_size >= tso->max_stack_size) {
2330       new_stack_size = tso->stack_size * 2;
2331   } else {
2332       new_stack_size = stg_min(tso->stack_size * 2, tso->max_stack_size);
2333   }
2334   new_tso_size   = (lnat)BLOCK_ROUND_UP(new_stack_size * sizeof(W_) +
2335                                        TSO_STRUCT_SIZE)/sizeof(W_);
2336   new_tso_size = round_to_mblocks(new_tso_size);  /* Be MBLOCK-friendly */
2337   new_stack_size = new_tso_size - TSO_STRUCT_SIZEW;
2338
2339   debugTrace(DEBUG_sched,
2340              "increasing stack size from %ld words to %d.",
2341              (long)tso->stack_size, new_stack_size);
2342
2343   dest = (StgTSO *)allocateLocal(cap,new_tso_size);
2344   TICK_ALLOC_TSO(new_stack_size,0);
2345
2346   /* copy the TSO block and the old stack into the new area */
2347   memcpy(dest,tso,TSO_STRUCT_SIZE);
2348   stack_words = tso->stack + tso->stack_size - tso->sp;
2349   new_sp = (P_)dest + new_tso_size - stack_words;
2350   memcpy(new_sp, tso->sp, stack_words * sizeof(W_));
2351
2352   /* relocate the stack pointers... */
2353   dest->sp         = new_sp;
2354   dest->stack_size = new_stack_size;
2355
2356   /* Mark the old TSO as relocated.  We have to check for relocated
2357    * TSOs in the garbage collector and any primops that deal with TSOs.
2358    *
2359    * It's important to set the sp value to just beyond the end
2360    * of the stack, so we don't attempt to scavenge any part of the
2361    * dead TSO's stack.
2362    */
2363   tso->what_next = ThreadRelocated;
2364   setTSOLink(cap,tso,dest);
2365   tso->sp = (P_)&(tso->stack[tso->stack_size]);
2366   tso->why_blocked = NotBlocked;
2367
2368   IF_PAR_DEBUG(verbose,
2369                debugBelch("@@ threadStackOverflow of TSO %d (now at %p): stack size increased to %ld\n",
2370                      tso->id, tso, tso->stack_size);
2371                /* If we're debugging, just print out the top of the stack */
2372                printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size,
2373                                                 tso->sp+64)));
2374
2375   unlockTSO(dest);
2376   unlockTSO(tso);
2377
2378   IF_DEBUG(sanity,checkTSO(dest));
2379 #if 0
2380   IF_DEBUG(scheduler,printTSO(dest));
2381 #endif
2382
2383   return dest;
2384 }
2385
2386 static StgTSO *
2387 threadStackUnderflow (Task *task STG_UNUSED, StgTSO *tso)
2388 {
2389     bdescr *bd, *new_bd;
2390     lnat free_w, tso_size_w;
2391     StgTSO *new_tso;
2392
2393     tso_size_w = tso_sizeW(tso);
2394
2395     if (tso_size_w < MBLOCK_SIZE_W ||
2396         (nat)(tso->stack + tso->stack_size - tso->sp) > tso->stack_size / 4)
2397     {
2398         return tso;
2399     }
2400
2401     // don't allow throwTo() to modify the blocked_exceptions queue
2402     // while we are moving the TSO:
2403     lockClosure((StgClosure *)tso);
2404
2405     // this is the number of words we'll free
2406     free_w = round_to_mblocks(tso_size_w/2);
2407
2408     bd = Bdescr((StgPtr)tso);
2409     new_bd = splitLargeBlock(bd, free_w / BLOCK_SIZE_W);
2410     bd->free = bd->start + TSO_STRUCT_SIZEW;
2411
2412     new_tso = (StgTSO *)new_bd->start;
2413     memcpy(new_tso,tso,TSO_STRUCT_SIZE);
2414     new_tso->stack_size = new_bd->free - new_tso->stack;
2415
2416     debugTrace(DEBUG_sched, "thread %ld: reducing TSO size from %lu words to %lu",
2417                (long)tso->id, tso_size_w, tso_sizeW(new_tso));
2418
2419     tso->what_next = ThreadRelocated;
2420     tso->_link = new_tso; // no write barrier reqd: same generation
2421
2422     // The TSO attached to this Task may have moved, so update the
2423     // pointer to it.
2424     if (task->tso == tso) {
2425         task->tso = new_tso;
2426     }
2427
2428     unlockTSO(new_tso);
2429     unlockTSO(tso);
2430
2431     IF_DEBUG(sanity,checkTSO(new_tso));
2432
2433     return new_tso;
2434 }
2435
2436 /* ---------------------------------------------------------------------------
2437    Interrupt execution
2438    - usually called inside a signal handler so it mustn't do anything fancy.
2439    ------------------------------------------------------------------------ */
2440
2441 void
2442 interruptStgRts(void)
2443 {
2444     sched_state = SCHED_INTERRUPTING;
2445     setContextSwitches();
2446     wakeUpRts();
2447 }
2448
2449 /* -----------------------------------------------------------------------------
2450    Wake up the RTS
2451
2452    This function causes at least one OS thread to wake up and run the
2453    scheduler loop.  It is invoked when the RTS might be deadlocked, or
2454    an external event has arrived that may need servicing (eg. a
2455    keyboard interrupt).
2456
2457    In the single-threaded RTS we don't do anything here; we only have
2458    one thread anyway, and the event that caused us to want to wake up
2459    will have interrupted any blocking system call in progress anyway.
2460    -------------------------------------------------------------------------- */
2461
2462 void
2463 wakeUpRts(void)
2464 {
2465 #if defined(THREADED_RTS)
2466     // This forces the IO Manager thread to wakeup, which will
2467     // in turn ensure that some OS thread wakes up and runs the
2468     // scheduler loop, which will cause a GC and deadlock check.
2469     ioManagerWakeup();
2470 #endif
2471 }
2472
2473 /* -----------------------------------------------------------------------------
2474  * checkBlackHoles()
2475  *
2476  * Check the blackhole_queue for threads that can be woken up.  We do
2477  * this periodically: before every GC, and whenever the run queue is
2478  * empty.
2479  *
2480  * An elegant solution might be to just wake up all the blocked
2481  * threads with awakenBlockedQueue occasionally: they'll go back to
2482  * sleep again if the object is still a BLACKHOLE.  Unfortunately this
2483  * doesn't give us a way to tell whether we've actually managed to
2484  * wake up any threads, so we would be busy-waiting.
2485  *
2486  * -------------------------------------------------------------------------- */
2487
2488 static rtsBool
2489 checkBlackHoles (Capability *cap)
2490 {
2491     StgTSO **prev, *t;
2492     rtsBool any_woke_up = rtsFalse;
2493     StgHalfWord type;
2494
2495     // blackhole_queue is global:
2496     ASSERT_LOCK_HELD(&sched_mutex);
2497
2498     debugTrace(DEBUG_sched, "checking threads blocked on black holes");
2499
2500     // ASSUMES: sched_mutex
2501     prev = &blackhole_queue;
2502     t = blackhole_queue;
2503     while (t != END_TSO_QUEUE) {
2504         ASSERT(t->why_blocked == BlockedOnBlackHole);
2505         type = get_itbl(UNTAG_CLOSURE(t->block_info.closure))->type;
2506         if (type != BLACKHOLE && type != CAF_BLACKHOLE) {
2507             IF_DEBUG(sanity,checkTSO(t));
2508             t = unblockOne(cap, t);
2509             *prev = t;
2510             any_woke_up = rtsTrue;
2511         } else {
2512             prev = &t->_link;
2513             t = t->_link;
2514         }
2515     }
2516
2517     return any_woke_up;
2518 }
2519
2520 /* -----------------------------------------------------------------------------
2521    Deleting threads
2522
2523    This is used for interruption (^C) and forking, and corresponds to
2524    raising an exception but without letting the thread catch the
2525    exception.
2526    -------------------------------------------------------------------------- */
2527
2528 static void
2529 deleteThread (Capability *cap, StgTSO *tso)
2530 {
2531     // NOTE: must only be called on a TSO that we have exclusive
2532     // access to, because we will call throwToSingleThreaded() below.
2533     // The TSO must be on the run queue of the Capability we own, or
2534     // we must own all Capabilities.
2535
2536     if (tso->why_blocked != BlockedOnCCall &&
2537         tso->why_blocked != BlockedOnCCall_NoUnblockExc) {
2538         throwToSingleThreaded(cap,tso,NULL);
2539     }
2540 }
2541
2542 #ifdef FORKPROCESS_PRIMOP_SUPPORTED
2543 static void
2544 deleteThread_(Capability *cap, StgTSO *tso)
2545 { // for forkProcess only:
2546   // like deleteThread(), but we delete threads in foreign calls, too.
2547
2548     if (tso->why_blocked == BlockedOnCCall ||
2549         tso->why_blocked == BlockedOnCCall_NoUnblockExc) {
2550         unblockOne(cap,tso);
2551         tso->what_next = ThreadKilled;
2552     } else {
2553         deleteThread(cap,tso);
2554     }
2555 }
2556 #endif
2557
2558 /* -----------------------------------------------------------------------------
2559    raiseExceptionHelper
2560
2561    This function is called by the raise# primitve, just so that we can
2562    move some of the tricky bits of raising an exception from C-- into
2563    C.  Who knows, it might be a useful re-useable thing here too.
2564    -------------------------------------------------------------------------- */
2565
2566 StgWord
2567 raiseExceptionHelper (StgRegTable *reg, StgTSO *tso, StgClosure *exception)
2568 {
2569     Capability *cap = regTableToCapability(reg);
2570     StgThunk *raise_closure = NULL;
2571     StgPtr p, next;
2572     StgRetInfoTable *info;
2573     //
2574     // This closure represents the expression 'raise# E' where E
2575     // is the exception raise.  It is used to overwrite all the
2576     // thunks which are currently under evaluataion.
2577     //
2578
2579     // OLD COMMENT (we don't have MIN_UPD_SIZE now):
2580     // LDV profiling: stg_raise_info has THUNK as its closure
2581     // type. Since a THUNK takes at least MIN_UPD_SIZE words in its
2582     // payload, MIN_UPD_SIZE is more approprate than 1.  It seems that
2583     // 1 does not cause any problem unless profiling is performed.
2584     // However, when LDV profiling goes on, we need to linearly scan
2585     // small object pool, where raise_closure is stored, so we should
2586     // use MIN_UPD_SIZE.
2587     //
2588     // raise_closure = (StgClosure *)RET_STGCALL1(P_,allocate,
2589     //                                 sizeofW(StgClosure)+1);
2590     //
2591
2592     //
2593     // Walk up the stack, looking for the catch frame.  On the way,
2594     // we update any closures pointed to from update frames with the
2595     // raise closure that we just built.
2596     //
2597     p = tso->sp;
2598     while(1) {
2599         info = get_ret_itbl((StgClosure *)p);
2600         next = p + stack_frame_sizeW((StgClosure *)p);
2601         switch (info->i.type) {
2602
2603         case UPDATE_FRAME:
2604             // Only create raise_closure if we need to.
2605             if (raise_closure == NULL) {
2606                 raise_closure =
2607                     (StgThunk *)allocateLocal(cap,sizeofW(StgThunk)+1);
2608                 SET_HDR(raise_closure, &stg_raise_info, CCCS);
2609                 raise_closure->payload[0] = exception;
2610             }
2611             UPD_IND(((StgUpdateFrame *)p)->updatee,(StgClosure *)raise_closure);
2612             p = next;
2613             continue;
2614
2615         case ATOMICALLY_FRAME:
2616             debugTrace(DEBUG_stm, "found ATOMICALLY_FRAME at %p", p);
2617             tso->sp = p;
2618             return ATOMICALLY_FRAME;
2619
2620         case CATCH_FRAME:
2621             tso->sp = p;
2622             return CATCH_FRAME;
2623
2624         case CATCH_STM_FRAME:
2625             debugTrace(DEBUG_stm, "found CATCH_STM_FRAME at %p", p);
2626             tso->sp = p;
2627             return CATCH_STM_FRAME;
2628
2629         case STOP_FRAME:
2630             tso->sp = p;
2631             return STOP_FRAME;
2632
2633         case CATCH_RETRY_FRAME:
2634         default:
2635             p = next;
2636             continue;
2637         }
2638     }
2639 }
2640
2641
2642 /* -----------------------------------------------------------------------------
2643    findRetryFrameHelper
2644
2645    This function is called by the retry# primitive.  It traverses the stack
2646    leaving tso->sp referring to the frame which should handle the retry.
2647
2648    This should either be a CATCH_RETRY_FRAME (if the retry# is within an orElse#)
2649    or should be a ATOMICALLY_FRAME (if the retry# reaches the top level).
2650
2651    We skip CATCH_STM_FRAMEs (aborting and rolling back the nested tx that they
2652    create) because retries are not considered to be exceptions, despite the
2653    similar implementation.
2654
2655    We should not expect to see CATCH_FRAME or STOP_FRAME because those should
2656    not be created within memory transactions.
2657    -------------------------------------------------------------------------- */
2658
2659 StgWord
2660 findRetryFrameHelper (StgTSO *tso)
2661 {
2662   StgPtr           p, next;
2663   StgRetInfoTable *info;
2664
2665   p = tso -> sp;
2666   while (1) {
2667     info = get_ret_itbl((StgClosure *)p);
2668     next = p + stack_frame_sizeW((StgClosure *)p);
2669     switch (info->i.type) {
2670
2671     case ATOMICALLY_FRAME:
2672         debugTrace(DEBUG_stm,
2673                    "found ATOMICALLY_FRAME at %p during retry", p);
2674         tso->sp = p;
2675         return ATOMICALLY_FRAME;
2676
2677     case CATCH_RETRY_FRAME:
2678         debugTrace(DEBUG_stm,
2679                    "found CATCH_RETRY_FRAME at %p during retrry", p);
2680         tso->sp = p;
2681         return CATCH_RETRY_FRAME;
2682
2683     case CATCH_STM_FRAME: {
2684         StgTRecHeader *trec = tso -> trec;
2685         StgTRecHeader *outer = stmGetEnclosingTRec(trec);
2686         debugTrace(DEBUG_stm,
2687                    "found CATCH_STM_FRAME at %p during retry", p);
2688         debugTrace(DEBUG_stm, "trec=%p outer=%p", trec, outer);
2689         stmAbortTransaction(tso -> cap, trec);
2690         stmFreeAbortedTRec(tso -> cap, trec);
2691         tso -> trec = outer;
2692         p = next;
2693         continue;
2694     }
2695
2696
2697     default:
2698       ASSERT(info->i.type != CATCH_FRAME);
2699       ASSERT(info->i.type != STOP_FRAME);
2700       p = next;
2701       continue;
2702     }
2703   }
2704 }
2705
2706 /* -----------------------------------------------------------------------------
2707    resurrectThreads is called after garbage collection on the list of
2708    threads found to be garbage.  Each of these threads will be woken
2709    up and sent a signal: BlockedOnDeadMVar if the thread was blocked
2710    on an MVar, or NonTermination if the thread was blocked on a Black
2711    Hole.
2712
2713    Locks: assumes we hold *all* the capabilities.
2714    -------------------------------------------------------------------------- */
2715
2716 void
2717 resurrectThreads (StgTSO *threads)
2718 {
2719     StgTSO *tso, *next;
2720     Capability *cap;
2721     step *step;
2722
2723     for (tso = threads; tso != END_TSO_QUEUE; tso = next) {
2724         next = tso->global_link;
2725
2726         step = Bdescr((P_)tso)->step;
2727         tso->global_link = step->threads;
2728         step->threads = tso;
2729
2730         debugTrace(DEBUG_sched, "resurrecting thread %lu", (unsigned long)tso->id);
2731
2732         // Wake up the thread on the Capability it was last on
2733         cap = tso->cap;
2734
2735         switch (tso->why_blocked) {
2736         case BlockedOnMVar:
2737         case BlockedOnException:
2738             /* Called by GC - sched_mutex lock is currently held. */
2739             throwToSingleThreaded(cap, tso,
2740                                   (StgClosure *)blockedOnDeadMVar_closure);
2741             break;
2742         case BlockedOnBlackHole:
2743             throwToSingleThreaded(cap, tso,
2744                                   (StgClosure *)nonTermination_closure);
2745             break;
2746         case BlockedOnSTM:
2747             throwToSingleThreaded(cap, tso,
2748                                   (StgClosure *)blockedIndefinitely_closure);
2749             break;
2750         case NotBlocked:
2751             /* This might happen if the thread was blocked on a black hole
2752              * belonging to a thread that we've just woken up (raiseAsync
2753              * can wake up threads, remember...).
2754              */
2755             continue;
2756         default:
2757             barf("resurrectThreads: thread blocked in a strange way");
2758         }
2759     }
2760 }
2761
2762 /* -----------------------------------------------------------------------------
2763    performPendingThrowTos is called after garbage collection, and
2764    passed a list of threads that were found to have pending throwTos
2765    (tso->blocked_exceptions was not empty), and were blocked.
2766    Normally this doesn't happen, because we would deliver the
2767    exception directly if the target thread is blocked, but there are
2768    small windows where it might occur on a multiprocessor (see
2769    throwTo()).
2770
2771    NB. we must be holding all the capabilities at this point, just
2772    like resurrectThreads().
2773    -------------------------------------------------------------------------- */
2774
2775 void
2776 performPendingThrowTos (StgTSO *threads)
2777 {
2778     StgTSO *tso, *next;
2779     Capability *cap;
2780     step *step;
2781
2782     for (tso = threads; tso != END_TSO_QUEUE; tso = next) {
2783         next = tso->global_link;
2784
2785         step = Bdescr((P_)tso)->step;
2786         tso->global_link = step->threads;
2787         step->threads = tso;
2788
2789         debugTrace(DEBUG_sched, "performing blocked throwTo to thread %lu", (unsigned long)tso->id);
2790
2791         cap = tso->cap;
2792         maybePerformBlockedException(cap, tso);
2793     }
2794 }