rts/Schedule.c

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