ghc/rts/Schedule.c

   1 /* ---------------------------------------------------------------------------
   2  *
   3  * (c) The GHC Team, 1998-2004
   4  *
   5  * Scheduler
   6  *
   7  * Different GHC ways use this scheduler quite differently (see comments below)
   8  * Here is the global picture:
   9  *
  10  * WAY  Name     CPP flag  What's it for
  11  * --------------------------------------
  12  * mp   GUM      PARALLEL_HASKELL          Parallel execution on a distrib. memory machine
  13  * s    SMP      SMP          Parallel execution on a shared memory machine
  14  * mg   GranSim  GRAN         Simulation of parallel execution
  15  * md   GUM/GdH  DIST         Distributed execution (based on GUM)
  16  *
  17  * --------------------------------------------------------------------------*/
  18
  19 /*
  20  * Version with support for distributed memory parallelism aka GUM (WAY=mp):
  21
  22    The main scheduling loop in GUM iterates until a finish message is received.
  23    In that case a global flag @receivedFinish@ is set and this instance of
  24    the RTS shuts down. See ghc/rts/parallel/HLComms.c:processMessages()
  25    for the handling of incoming messages, such as PP_FINISH.
  26    Note that in the parallel case we have a system manager that coordinates
  27    different PEs, each of which are running one instance of the RTS.
  28    See ghc/rts/parallel/SysMan.c for the main routine of the parallel program.
  29    From this routine processes executing ghc/rts/Main.c are spawned. -- HWL
  30
  31  * Version with support for simulating parallel execution aka GranSim (WAY=mg):
  32
  33    The main scheduling code in GranSim is quite different from that in std
  34    (concurrent) Haskell: while concurrent Haskell just iterates over the
  35    threads in the runnable queue, GranSim is event driven, i.e. it iterates
  36    over the events in the global event queue.  -- HWL
  37 */
  38
  39 #include "PosixSource.h"
  40 #include "Rts.h"
  41 #include "SchedAPI.h"
  42 #include "RtsUtils.h"
  43 #include "RtsFlags.h"
  44 #include "BlockAlloc.h"
  45 #include "OSThreads.h"
  46 #include "Storage.h"
  47 #include "StgRun.h"
  48 #include "Hooks.h"
  49 #define COMPILING_SCHEDULER
  50 #include "Schedule.h"
  51 #include "StgMiscClosures.h"
  52 #include "Interpreter.h"
  53 #include "Exception.h"
  54 #include "Printer.h"
  55 #include "Signals.h"
  56 #include "Sanity.h"
  57 #include "Stats.h"
  58 #include "STM.h"
  59 #include "Timer.h"
  60 #include "Prelude.h"
  61 #include "ThreadLabels.h"
  62 #include "LdvProfile.h"
  63 #include "Updates.h"
  64 #ifdef PROFILING
  65 #include "Proftimer.h"
  66 #include "ProfHeap.h"
  67 #endif
  68 #if defined(GRAN) || defined(PARALLEL_HASKELL)
  69 # include "GranSimRts.h"
  70 # include "GranSim.h"
  71 # include "ParallelRts.h"
  72 # include "Parallel.h"
  73 # include "ParallelDebug.h"
  74 # include "FetchMe.h"
  75 # include "HLC.h"
  76 #endif
  77 #include "Sparks.h"
  78 #include "Capability.h"
  79 #include  "Task.h"
  80
  81 #ifdef HAVE_SYS_TYPES_H
  82 #include <sys/types.h>
  83 #endif
  84 #ifdef HAVE_UNISTD_H
  85 #include <unistd.h>
  86 #endif
  87
  88 #include <string.h>
  89 #include <stdlib.h>
  90 #include <stdarg.h>
  91
  92 #ifdef HAVE_ERRNO_H
  93 #include <errno.h>
  94 #endif
  95
  96 // Turn off inlining when debugging - it obfuscates things
  97 #ifdef DEBUG
  98 # undef  STATIC_INLINE
  99 # define STATIC_INLINE static
 100 #endif
 101
 102 #ifdef THREADED_RTS
 103 #define USED_IN_THREADED_RTS
 104 #else
 105 #define USED_IN_THREADED_RTS STG_UNUSED
 106 #endif
 107
 108 #ifdef RTS_SUPPORTS_THREADS
 109 #define USED_WHEN_RTS_SUPPORTS_THREADS
 110 #else
 111 #define USED_WHEN_RTS_SUPPORTS_THREADS STG_UNUSED
 112 #endif
 113
 114 /* Main thread queue.
 115  * Locks required: sched_mutex.
 116  */
 117 StgMainThread *main_threads = NULL;
 118
 119 #if defined(GRAN)
 120
 121 StgTSO* ActiveTSO = NULL; /* for assigning system costs; GranSim-Light only */
 122 /* rtsTime TimeOfNextEvent, EndOfTimeSlice;            now in GranSim.c */
 123
 124 /*
 125    In GranSim we have a runnable and a blocked queue for each processor.
 126    In order to minimise code changes new arrays run_queue_hds/tls
 127    are created. run_queue_hd is then a short cut (macro) for
 128    run_queue_hds[CurrentProc] (see GranSim.h).
 129    -- HWL
 130 */
 131 StgTSO *run_queue_hds[MAX_PROC], *run_queue_tls[MAX_PROC];
 132 StgTSO *blocked_queue_hds[MAX_PROC], *blocked_queue_tls[MAX_PROC];
 133 StgTSO *ccalling_threadss[MAX_PROC];
 134 /* We use the same global list of threads (all_threads) in GranSim as in
 135    the std RTS (i.e. we are cheating). However, we don't use this list in
 136    the GranSim specific code at the moment (so we are only potentially
 137    cheating).  */
 138
 139 #else /* !GRAN */
 140
 141 /* Thread queues.
 142  * Locks required: sched_mutex.
 143  */
 144 StgTSO *run_queue_hd = NULL;
 145 StgTSO *run_queue_tl = NULL;
 146 StgTSO *blocked_queue_hd = NULL;
 147 StgTSO *blocked_queue_tl = NULL;
 148 StgTSO *blackhole_queue = NULL;
 149 StgTSO *sleeping_queue = NULL;    /* perhaps replace with a hash table? */
 150
 151 #endif
 152
 153 /* The blackhole_queue should be checked for threads to wake up.  See
 154  * Schedule.h for more thorough comment.
 155  */
 156 rtsBool blackholes_need_checking = rtsFalse;
 157
 158 /* Linked list of all threads.
 159  * Used for detecting garbage collected threads.
 160  */
 161 StgTSO *all_threads = NULL;
 162
 163 /* When a thread performs a safe C call (_ccall_GC, using old
 164  * terminology), it gets put on the suspended_ccalling_threads
 165  * list. Used by the garbage collector.
 166  */
 167 static StgTSO *suspended_ccalling_threads;
 168
 169 /* KH: The following two flags are shared memory locations.  There is no need
 170        to lock them, since they are only unset at the end of a scheduler
 171        operation.
 172 */
 173
 174 /* flag set by signal handler to precipitate a context switch */
 175 int context_switch = 0;
 176
 177 /* flag that tracks whether we have done any execution in this time slice. */
 178 nat recent_activity = ACTIVITY_YES;
 179
 180 /* if this flag is set as well, give up execution */
 181 rtsBool interrupted = rtsFalse;
 182
 183 /* Next thread ID to allocate.
 184  * Locks required: thread_id_mutex
 185  */
 186 static StgThreadID next_thread_id = 1;
 187
 188 /*
 189  * Pointers to the state of the current thread.
 190  * Rule of thumb: if CurrentTSO != NULL, then we're running a Haskell
 191  * thread.  If CurrentTSO == NULL, then we're at the scheduler level.
 192  */
 193
 194 /* The smallest stack size that makes any sense is:
 195  *    RESERVED_STACK_WORDS    (so we can get back from the stack overflow)
 196  *  + sizeofW(StgStopFrame)   (the stg_stop_thread_info frame)
 197  *  + 1                       (the closure to enter)
 198  *  + 1                       (stg_ap_v_ret)
 199  *  + 1                       (spare slot req'd by stg_ap_v_ret)
 200  *
 201  * A thread with this stack will bomb immediately with a stack
 202  * overflow, which will increase its stack size.
 203  */
 204
 205 #define MIN_STACK_WORDS (RESERVED_STACK_WORDS + sizeofW(StgStopFrame) + 3)
 206
 207
 208 #if defined(GRAN)
 209 StgTSO *CurrentTSO;
 210 #endif
 211
 212 /*  This is used in `TSO.h' and gcc 2.96 insists that this variable actually
 213  *  exists - earlier gccs apparently didn't.
 214  *  -= chak
 215  */
 216 StgTSO dummy_tso;
 217
 218 /*
 219  * Set to TRUE when entering a shutdown state (via shutdownHaskellAndExit()) --
 220  * in an MT setting, needed to signal that a worker thread shouldn't hang around
 221  * in the scheduler when it is out of work.
 222  */
 223 static rtsBool shutting_down_scheduler = rtsFalse;
 224
 225 #if defined(RTS_SUPPORTS_THREADS)
 226 /* ToDo: carefully document the invariants that go together
 227  *       with these synchronisation objects.
 228  */
 229 Mutex     sched_mutex       = INIT_MUTEX_VAR;
 230 Mutex     term_mutex        = INIT_MUTEX_VAR;
 231
 232 #endif /* RTS_SUPPORTS_THREADS */
 233
 234 #if defined(PARALLEL_HASKELL)
 235 StgTSO *LastTSO;
 236 rtsTime TimeOfLastYield;
 237 rtsBool emitSchedule = rtsTrue;
 238 #endif
 239
 240 #if DEBUG
 241 static char *whatNext_strs[] = {
 242   "(unknown)",
 243   "ThreadRunGHC",
 244   "ThreadInterpret",
 245   "ThreadKilled",
 246   "ThreadRelocated",
 247   "ThreadComplete"
 248 };
 249 #endif
 250
 251 /* -----------------------------------------------------------------------------
 252  * static function prototypes
 253  * -------------------------------------------------------------------------- */
 254
 255 #if defined(RTS_SUPPORTS_THREADS)
 256 static void taskStart(void);
 257 #endif
 258
 259 static void schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS,
 260                       Capability *initialCapability );
 261
 262 //
 263 // These function all encapsulate parts of the scheduler loop, and are
 264 // abstracted only to make the structure and control flow of the
 265 // scheduler clearer.
 266 //
 267 static void schedulePreLoop(void);
 268 static void scheduleStartSignalHandlers(void);
 269 static void scheduleCheckBlockedThreads(void);
 270 static void scheduleCheckBlackHoles(void);
 271 static void scheduleDetectDeadlock(void);
 272 #if defined(GRAN)
 273 static StgTSO *scheduleProcessEvent(rtsEvent *event);
 274 #endif
 275 #if defined(PARALLEL_HASKELL)
 276 static StgTSO *scheduleSendPendingMessages(void);
 277 static void scheduleActivateSpark(void);
 278 static rtsBool scheduleGetRemoteWork(rtsBool *receivedFinish);
 279 #endif
 280 #if defined(PAR) || defined(GRAN)
 281 static void scheduleGranParReport(void);
 282 #endif
 283 static void schedulePostRunThread(void);
 284 static rtsBool scheduleHandleHeapOverflow( Capability *cap, StgTSO *t );
 285 static void scheduleHandleStackOverflow( StgTSO *t);
 286 static rtsBool scheduleHandleYield( StgTSO *t, nat prev_what_next );
 287 static void scheduleHandleThreadBlocked( StgTSO *t );
 288 static rtsBool scheduleHandleThreadFinished( StgMainThread *mainThread,
 289                                              Capability *cap, StgTSO *t );
 290 static rtsBool scheduleDoHeapProfile(rtsBool ready_to_gc);
 291 static void scheduleDoGC(Capability *cap);
 292
 293 static void unblockThread(StgTSO *tso);
 294 static rtsBool checkBlackHoles(void);
 295 static SchedulerStatus waitThread_(/*out*/StgMainThread* m,
 296                                    Capability *initialCapability
 297                                    );
 298 static void scheduleThread_ (StgTSO* tso);
 299 static void AllRoots(evac_fn evac);
 300
 301 static StgTSO *threadStackOverflow(StgTSO *tso);
 302
 303 static void raiseAsync_(StgTSO *tso, StgClosure *exception,
 304                         rtsBool stop_at_atomically);
 305
 306 static void printThreadBlockage(StgTSO *tso);
 307 static void printThreadStatus(StgTSO *tso);
 308
 309 #if defined(PARALLEL_HASKELL)
 310 StgTSO * createSparkThread(rtsSpark spark);
 311 StgTSO * activateSpark (rtsSpark spark);
 312 #endif
 313
 314 /* ----------------------------------------------------------------------------
 315  * Starting Tasks
 316  * ------------------------------------------------------------------------- */
 317
 318 #if defined(RTS_SUPPORTS_THREADS)
 319 static nat startingWorkerThread = 0;
 320
 321 static void
 322 taskStart(void)
 323 {
 324   ACQUIRE_LOCK(&sched_mutex);
 325   startingWorkerThread--;
 326   schedule(NULL,NULL);
 327   taskStop();
 328   RELEASE_LOCK(&sched_mutex);
 329 }
 330
 331 void
 332 startSchedulerTaskIfNecessary(void)
 333 {
 334     if ( !EMPTY_RUN_QUEUE()
 335          && !shutting_down_scheduler // not if we're shutting down
 336          && startingWorkerThread==0)
 337     {
 338         // we don't want to start another worker thread
 339         // just because the last one hasn't yet reached the
 340         // "waiting for capability" state
 341         startingWorkerThread++;
 342         if (!maybeStartNewWorker(taskStart)) {
 343             startingWorkerThread--;
 344         }
 345     }
 346 }
 347 #endif
 348
 349 /* -----------------------------------------------------------------------------
 350  * Putting a thread on the run queue: different scheduling policies
 351  * -------------------------------------------------------------------------- */
 352
 353 STATIC_INLINE void
 354 addToRunQueue( StgTSO *t )
 355 {
 356 #if defined(PARALLEL_HASKELL)
 357     if (RtsFlags.ParFlags.doFairScheduling) {
 358         // this does round-robin scheduling; good for concurrency
 359         APPEND_TO_RUN_QUEUE(t);
 360     } else {
 361         // this does unfair scheduling; good for parallelism
 362         PUSH_ON_RUN_QUEUE(t);
 363     }
 364 #else
 365     // this does round-robin scheduling; good for concurrency
 366     APPEND_TO_RUN_QUEUE(t);
 367 #endif
 368 }
 369
 370 /* ---------------------------------------------------------------------------
 371    Main scheduling loop.
 372
 373    We use round-robin scheduling, each thread returning to the
 374    scheduler loop when one of these conditions is detected:
 375
 376       * out of heap space
 377       * timer expires (thread yields)
 378       * thread blocks
 379       * thread ends
 380       * stack overflow
 381
 382    Locking notes:  we acquire the scheduler lock once at the beginning
 383    of the scheduler loop, and release it when
 384
 385       * running a thread, or
 386       * waiting for work, or
 387       * waiting for a GC to complete.
 388
 389    GRAN version:
 390      In a GranSim setup this loop iterates over the global event queue.
 391      This revolves around the global event queue, which determines what
 392      to do next. Therefore, it's more complicated than either the
 393      concurrent or the parallel (GUM) setup.
 394
 395    GUM version:
 396      GUM iterates over incoming messages.
 397      It starts with nothing to do (thus CurrentTSO == END_TSO_QUEUE),
 398      and sends out a fish whenever it has nothing to do; in-between
 399      doing the actual reductions (shared code below) it processes the
 400      incoming messages and deals with delayed operations
 401      (see PendingFetches).
 402      This is not the ugliest code you could imagine, but it's bloody close.
 403
 404    ------------------------------------------------------------------------ */
 405
 406 static void
 407 schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS,
 408           Capability *initialCapability )
 409 {
 410   StgTSO *t;
 411   Capability *cap;
 412   StgThreadReturnCode ret;
 413 #if defined(GRAN)
 414   rtsEvent *event;
 415 #elif defined(PARALLEL_HASKELL)
 416   StgTSO *tso;
 417   GlobalTaskId pe;
 418   rtsBool receivedFinish = rtsFalse;
 419 # if defined(DEBUG)
 420   nat tp_size, sp_size; // stats only
 421 # endif
 422 #endif
 423   nat prev_what_next;
 424   rtsBool ready_to_gc;
 425
 426   // Pre-condition: sched_mutex is held.
 427   // We might have a capability, passed in as initialCapability.
 428   cap = initialCapability;
 429
 430 #if !defined(RTS_SUPPORTS_THREADS)
 431   // simply initialise it in the non-threaded case
 432   grabCapability(&cap);
 433 #endif
 434
 435   IF_DEBUG(scheduler,
 436            sched_belch("### NEW SCHEDULER LOOP (main thr: %p, cap: %p)",
 437                        mainThread, initialCapability);
 438       );
 439
 440   schedulePreLoop();
 441
 442   // -----------------------------------------------------------
 443   // Scheduler loop starts here:
 444
 445 #if defined(PARALLEL_HASKELL)
 446 #define TERMINATION_CONDITION        (!receivedFinish)
 447 #elif defined(GRAN)
 448 #define TERMINATION_CONDITION        ((event = get_next_event()) != (rtsEvent*)NULL)
 449 #else
 450 #define TERMINATION_CONDITION        rtsTrue
 451 #endif
 452
 453   while (TERMINATION_CONDITION) {
 454
 455 #if defined(GRAN)
 456       /* Choose the processor with the next event */
 457       CurrentProc = event->proc;
 458       CurrentTSO = event->tso;
 459 #endif
 460
 461 #if defined(RTS_SUPPORTS_THREADS)
 462       // Yield the capability to higher-priority tasks if necessary.
 463       //
 464       if (cap != NULL) {
 465           yieldCapability(&cap);
 466       }
 467
 468       // If we do not currently hold a capability, we wait for one
 469       //
 470       if (cap == NULL) {
 471           waitForCapability(&sched_mutex, &cap,
 472                             mainThread ? &mainThread->bound_thread_cond : NULL);
 473       }
 474
 475       // We now have a capability...
 476 #endif
 477
 478 #if 0 /* extra sanity checking */
 479       {
 480           StgMainThread *m;
 481           for (m = main_threads; m != NULL; m = m->link) {
 482               ASSERT(get_itbl(m->tso)->type == TSO);
 483           }
 484       }
 485 #endif
 486
 487     // Check whether we have re-entered the RTS from Haskell without
 488     // going via suspendThread()/resumeThread (i.e. a 'safe' foreign
 489     // call).
 490     if (cap->r.rInHaskell) {
 491           errorBelch("schedule: re-entered unsafely.\n"
 492                      "   Perhaps a 'foreign import unsafe' should be 'safe'?");
 493           stg_exit(1);
 494     }
 495
 496     //
 497     // Test for interruption.  If interrupted==rtsTrue, then either
 498     // we received a keyboard interrupt (^C), or the scheduler is
 499     // trying to shut down all the tasks (shutting_down_scheduler) in
 500     // the threaded RTS.
 501     //
 502     if (interrupted) {
 503         if (shutting_down_scheduler) {
 504             IF_DEBUG(scheduler, sched_belch("shutting down"));
 505             releaseCapability(cap);
 506             if (mainThread) {
 507                 mainThread->stat = Interrupted;
 508                 mainThread->ret  = NULL;
 509             }
 510             return;
 511         } else {
 512             IF_DEBUG(scheduler, sched_belch("interrupted"));
 513             deleteAllThreads();
 514         }
 515     }
 516
 517 #if defined(not_yet) && defined(SMP)
 518     //
 519     // Top up the run queue from our spark pool.  We try to make the
 520     // number of threads in the run queue equal to the number of
 521     // free capabilities.
 522     //
 523     {
 524         StgClosure *spark;
 525         if (EMPTY_RUN_QUEUE()) {
 526             spark = findSpark(rtsFalse);
 527             if (spark == NULL) {
 528                 break; /* no more sparks in the pool */
 529             } else {
 530                 createSparkThread(spark);
 531                 IF_DEBUG(scheduler,
 532                          sched_belch("==^^ turning spark of closure %p into a thread",
 533                                      (StgClosure *)spark));
 534             }
 535         }
 536     }
 537 #endif // SMP
 538
 539     scheduleStartSignalHandlers();
 540
 541     // Only check the black holes here if we've nothing else to do.
 542     // During normal execution, the black hole list only gets checked
 543     // at GC time, to avoid repeatedly traversing this possibly long
 544     // list each time around the scheduler.
 545     if (EMPTY_RUN_QUEUE()) { scheduleCheckBlackHoles(); }
 546
 547     scheduleCheckBlockedThreads();
 548
 549     scheduleDetectDeadlock();
 550
 551     // Normally, the only way we can get here with no threads to
 552     // run is if a keyboard interrupt received during
 553     // scheduleCheckBlockedThreads() or scheduleDetectDeadlock().
 554     // Additionally, it is not fatal for the
 555     // threaded RTS to reach here with no threads to run.
 556     //
 557     // win32: might be here due to awaitEvent() being abandoned
 558     // as a result of a console event having been delivered.
 559     if ( EMPTY_RUN_QUEUE() ) {
 560 #if !defined(RTS_SUPPORTS_THREADS) && !defined(mingw32_HOST_OS)
 561         ASSERT(interrupted);
 562 #endif
 563         continue; // nothing to do
 564     }
 565
 566 #if defined(PARALLEL_HASKELL)
 567     scheduleSendPendingMessages();
 568     if (EMPTY_RUN_QUEUE() && scheduleActivateSpark())
 569         continue;
 570
 571 #if defined(SPARKS)
 572     ASSERT(next_fish_to_send_at==0);  // i.e. no delayed fishes left!
 573 #endif
 574
 575     /* If we still have no work we need to send a FISH to get a spark
 576        from another PE */
 577     if (EMPTY_RUN_QUEUE()) {
 578         if (!scheduleGetRemoteWork(&receivedFinish)) continue;
 579         ASSERT(rtsFalse); // should not happen at the moment
 580     }
 581     // from here: non-empty run queue.
 582     //  TODO: merge above case with this, only one call processMessages() !
 583     if (PacketsWaiting()) {  /* process incoming messages, if
 584                                 any pending...  only in else
 585                                 because getRemoteWork waits for
 586                                 messages as well */
 587         receivedFinish = processMessages();
 588     }
 589 #endif
 590
 591 #if defined(GRAN)
 592     scheduleProcessEvent(event);
 593 #endif
 594
 595     //
 596     // Get a thread to run
 597     //
 598     ASSERT(run_queue_hd != END_TSO_QUEUE);
 599     POP_RUN_QUEUE(t);
 600
 601 #if defined(GRAN) || defined(PAR)
 602     scheduleGranParReport(); // some kind of debuging output
 603 #else
 604     // Sanity check the thread we're about to run.  This can be
 605     // expensive if there is lots of thread switching going on...
 606     IF_DEBUG(sanity,checkTSO(t));
 607 #endif
 608
 609 #if defined(RTS_SUPPORTS_THREADS)
 610     // Check whether we can run this thread in the current task.
 611     // If not, we have to pass our capability to the right task.
 612     {
 613       StgMainThread *m = t->main;
 614
 615       if(m)
 616       {
 617         if(m == mainThread)
 618         {
 619           IF_DEBUG(scheduler,
 620             sched_belch("### Running thread %d in bound thread", t->id));
 621           // yes, the Haskell thread is bound to the current native thread
 622         }
 623         else
 624         {
 625           IF_DEBUG(scheduler,
 626             sched_belch("### thread %d bound to another OS thread", t->id));
 627           // no, bound to a different Haskell thread: pass to that thread
 628           PUSH_ON_RUN_QUEUE(t);
 629           passCapability(&m->bound_thread_cond);
 630           continue;
 631         }
 632       }
 633       else
 634       {
 635         if(mainThread != NULL)
 636         // The thread we want to run is bound.
 637         {
 638           IF_DEBUG(scheduler,
 639             sched_belch("### this OS thread cannot run thread %d", t->id));
 640           // no, the current native thread is bound to a different
 641           // Haskell thread, so pass it to any worker thread
 642           PUSH_ON_RUN_QUEUE(t);
 643           passCapabilityToWorker();
 644           continue;
 645         }
 646       }
 647     }
 648 #endif
 649
 650     cap->r.rCurrentTSO = t;
 651
 652     /* context switches are now initiated by the timer signal, unless
 653      * the user specified "context switch as often as possible", with
 654      * +RTS -C0
 655      */
 656     if ((RtsFlags.ConcFlags.ctxtSwitchTicks == 0
 657          && (run_queue_hd != END_TSO_QUEUE
 658              || blocked_queue_hd != END_TSO_QUEUE
 659              || sleeping_queue != END_TSO_QUEUE)))
 660         context_switch = 1;
 661
 662 run_thread:
 663
 664     RELEASE_LOCK(&sched_mutex);
 665
 666     IF_DEBUG(scheduler, sched_belch("-->> running thread %ld %s ...",
 667                               (long)t->id, whatNext_strs[t->what_next]));
 668
 669 #if defined(PROFILING)
 670     startHeapProfTimer();
 671 #endif
 672
 673     // ----------------------------------------------------------------------
 674     // Run the current thread
 675
 676     prev_what_next = t->what_next;
 677
 678     errno = t->saved_errno;
 679     cap->r.rInHaskell = rtsTrue;
 680
 681     recent_activity = ACTIVITY_YES;
 682
 683     switch (prev_what_next) {
 684
 685     case ThreadKilled:
 686     case ThreadComplete:
 687         /* Thread already finished, return to scheduler. */
 688         ret = ThreadFinished;
 689         break;
 690
 691     case ThreadRunGHC:
 692         ret = StgRun((StgFunPtr) stg_returnToStackTop, &cap->r);
 693         break;
 694
 695     case ThreadInterpret:
 696         ret = interpretBCO(cap);
 697         break;
 698
 699     default:
 700       barf("schedule: invalid what_next field");
 701     }
 702
 703 #if defined(SMP)
 704     // in SMP mode, we might return with a different capability than
 705     // we started with, if the Haskell thread made a foreign call.  So
 706     // let's find out what our current Capability is:
 707     cap = myCapability();
 708 #endif
 709
 710     // We have run some Haskell code: there might be blackhole-blocked
 711     // threads to wake up now.
 712     if ( blackhole_queue != END_TSO_QUEUE ) {
 713         blackholes_need_checking = rtsTrue;
 714     }
 715
 716     cap->r.rInHaskell = rtsFalse;
 717
 718     // The TSO might have moved, eg. if it re-entered the RTS and a GC
 719     // happened.  So find the new location:
 720     t = cap->r.rCurrentTSO;
 721
 722     // And save the current errno in this thread.
 723     t->saved_errno = errno;
 724
 725     // ----------------------------------------------------------------------
 726
 727     /* Costs for the scheduler are assigned to CCS_SYSTEM */
 728 #if defined(PROFILING)
 729     stopHeapProfTimer();
 730     CCCS = CCS_SYSTEM;
 731 #endif
 732
 733     ACQUIRE_LOCK(&sched_mutex);
 734
 735 #if defined(RTS_SUPPORTS_THREADS)
 736     IF_DEBUG(scheduler,debugBelch("sched (task %p): ", osThreadId()););
 737 #elif !defined(GRAN) && !defined(PARALLEL_HASKELL)
 738     IF_DEBUG(scheduler,debugBelch("sched: "););
 739 #endif
 740
 741     schedulePostRunThread();
 742
 743     ready_to_gc = rtsFalse;
 744
 745     switch (ret) {
 746     case HeapOverflow:
 747         ready_to_gc = scheduleHandleHeapOverflow(cap,t);
 748         break;
 749
 750     case StackOverflow:
 751         scheduleHandleStackOverflow(t);
 752         break;
 753
 754     case ThreadYielding:
 755         if (scheduleHandleYield(t, prev_what_next)) {
 756             // shortcut for switching between compiler/interpreter:
 757             goto run_thread;
 758         }
 759         break;
 760
 761     case ThreadBlocked:
 762         scheduleHandleThreadBlocked(t);
 763         break;
 764
 765     case ThreadFinished:
 766         if (scheduleHandleThreadFinished(mainThread, cap, t)) return;;
 767         break;
 768
 769     default:
 770       barf("schedule: invalid thread return code %d", (int)ret);
 771     }
 772
 773     if (scheduleDoHeapProfile(ready_to_gc)) { ready_to_gc = rtsFalse; }
 774     if (ready_to_gc) { scheduleDoGC(cap); }
 775   } /* end of while() */
 776
 777   IF_PAR_DEBUG(verbose,
 778                debugBelch("== Leaving schedule() after having received Finish\n"));
 779 }
 780
 781 /* ----------------------------------------------------------------------------
 782  * Setting up the scheduler loop
 783  * ASSUMES: sched_mutex
 784  * ------------------------------------------------------------------------- */
 785
 786 static void
 787 schedulePreLoop(void)
 788 {
 789 #if defined(GRAN)
 790     /* set up first event to get things going */
 791     /* ToDo: assign costs for system setup and init MainTSO ! */
 792     new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
 793               ContinueThread,
 794               CurrentTSO, (StgClosure*)NULL, (rtsSpark*)NULL);
 795
 796     IF_DEBUG(gran,
 797              debugBelch("GRAN: Init CurrentTSO (in schedule) = %p\n",
 798                         CurrentTSO);
 799              G_TSO(CurrentTSO, 5));
 800
 801     if (RtsFlags.GranFlags.Light) {
 802         /* Save current time; GranSim Light only */
 803         CurrentTSO->gran.clock = CurrentTime[CurrentProc];
 804     }
 805 #endif
 806 }
 807
 808 /* ----------------------------------------------------------------------------
 809  * Start any pending signal handlers
 810  * ASSUMES: sched_mutex
 811  * ------------------------------------------------------------------------- */
 812
 813 static void
 814 scheduleStartSignalHandlers(void)
 815 {
 816 #if defined(RTS_USER_SIGNALS) && !defined(RTS_SUPPORTS_THREADS)
 817     if (signals_pending()) {
 818       RELEASE_LOCK(&sched_mutex); /* ToDo: kill */
 819       startSignalHandlers();
 820       ACQUIRE_LOCK(&sched_mutex);
 821     }
 822 #endif
 823 }
 824
 825 /* ----------------------------------------------------------------------------
 826  * Check for blocked threads that can be woken up.
 827  * ASSUMES: sched_mutex
 828  * ------------------------------------------------------------------------- */
 829
 830 static void
 831 scheduleCheckBlockedThreads(void)
 832 {
 833     //
 834     // Check whether any waiting threads need to be woken up.  If the
 835     // run queue is empty, and there are no other tasks running, we
 836     // can wait indefinitely for something to happen.
 837     //
 838     if ( !EMPTY_QUEUE(blocked_queue_hd) || !EMPTY_QUEUE(sleeping_queue) )
 839     {
 840 #if defined(RTS_SUPPORTS_THREADS)
 841         // We shouldn't be here...
 842         barf("schedule: awaitEvent() in threaded RTS");
 843 #endif
 844         awaitEvent( EMPTY_RUN_QUEUE() && !blackholes_need_checking );
 845     }
 846 }
 847
 848
 849 /* ----------------------------------------------------------------------------
 850  * Check for threads blocked on BLACKHOLEs that can be woken up
 851  * ASSUMES: sched_mutex
 852  * ------------------------------------------------------------------------- */
 853 static void
 854 scheduleCheckBlackHoles( void )
 855 {
 856     if ( blackholes_need_checking )
 857     {
 858         checkBlackHoles();
 859         blackholes_need_checking = rtsFalse;
 860     }
 861 }
 862
 863 /* ----------------------------------------------------------------------------
 864  * Detect deadlock conditions and attempt to resolve them.
 865  * ASSUMES: sched_mutex
 866  * ------------------------------------------------------------------------- */
 867
 868 static void
 869 scheduleDetectDeadlock(void)
 870 {
 871
 872 #if defined(PARALLEL_HASKELL)
 873     // ToDo: add deadlock detection in GUM (similar to SMP) -- HWL
 874     return;
 875 #endif
 876
 877     /*
 878      * Detect deadlock: when we have no threads to run, there are no
 879      * threads blocked, waiting for I/O, or sleeping, and all the
 880      * other tasks are waiting for work, we must have a deadlock of
 881      * some description.
 882      */
 883     if ( EMPTY_THREAD_QUEUES() )
 884     {
 885 #if defined(RTS_SUPPORTS_THREADS)
 886         /*
 887          * In the threaded RTS, we only check for deadlock if there
 888          * has been no activity in a complete timeslice.  This means
 889          * we won't eagerly start a full GC just because we don't have
 890          * any threads to run currently.
 891          */
 892         if (recent_activity != ACTIVITY_INACTIVE) return;
 893 #endif
 894
 895         IF_DEBUG(scheduler, sched_belch("deadlocked, forcing major GC..."));
 896
 897         // Garbage collection can release some new threads due to
 898         // either (a) finalizers or (b) threads resurrected because
 899         // they are unreachable and will therefore be sent an
 900         // exception.  Any threads thus released will be immediately
 901         // runnable.
 902
 903         GarbageCollect(GetRoots,rtsTrue);
 904         recent_activity = ACTIVITY_DONE_GC;
 905         if ( !EMPTY_RUN_QUEUE() ) return;
 906
 907 #if defined(RTS_USER_SIGNALS) && !defined(RTS_SUPPORTS_THREADS)
 908         /* If we have user-installed signal handlers, then wait
 909          * for signals to arrive rather then bombing out with a
 910          * deadlock.
 911          */
 912         if ( anyUserHandlers() ) {
 913             IF_DEBUG(scheduler,
 914                      sched_belch("still deadlocked, waiting for signals..."));
 915
 916             awaitUserSignals();
 917
 918             if (signals_pending()) {
 919                 RELEASE_LOCK(&sched_mutex);
 920                 startSignalHandlers();
 921                 ACQUIRE_LOCK(&sched_mutex);
 922             }
 923
 924             // either we have threads to run, or we were interrupted:
 925             ASSERT(!EMPTY_RUN_QUEUE() || interrupted);
 926         }
 927 #endif
 928
 929 #if !defined(RTS_SUPPORTS_THREADS)
 930         /* Probably a real deadlock.  Send the current main thread the
 931          * Deadlock exception (or in the SMP build, send *all* main
 932          * threads the deadlock exception, since none of them can make
 933          * progress).
 934          */
 935         {
 936             StgMainThread *m;
 937             m = main_threads;
 938             switch (m->tso->why_blocked) {
 939             case BlockedOnSTM:
 940             case BlockedOnBlackHole:
 941             case BlockedOnException:
 942             case BlockedOnMVar:
 943                 raiseAsync(m->tso, (StgClosure *)NonTermination_closure);
 944                 return;
 945             default:
 946                 barf("deadlock: main thread blocked in a strange way");
 947             }
 948         }
 949 #endif
 950     }
 951 }
 952
 953 /* ----------------------------------------------------------------------------
 954  * Process an event (GRAN only)
 955  * ------------------------------------------------------------------------- */
 956
 957 #if defined(GRAN)
 958 static StgTSO *
 959 scheduleProcessEvent(rtsEvent *event)
 960 {
 961     StgTSO *t;
 962
 963     if (RtsFlags.GranFlags.Light)
 964       GranSimLight_enter_system(event, &ActiveTSO); // adjust ActiveTSO etc
 965
 966     /* adjust time based on time-stamp */
 967     if (event->time > CurrentTime[CurrentProc] &&
 968         event->evttype != ContinueThread)
 969       CurrentTime[CurrentProc] = event->time;
 970
 971     /* Deal with the idle PEs (may issue FindWork or MoveSpark events) */
 972     if (!RtsFlags.GranFlags.Light)
 973       handleIdlePEs();
 974
 975     IF_DEBUG(gran, debugBelch("GRAN: switch by event-type\n"));
 976
 977     /* main event dispatcher in GranSim */
 978     switch (event->evttype) {
 979       /* Should just be continuing execution */
 980     case ContinueThread:
 981       IF_DEBUG(gran, debugBelch("GRAN: doing ContinueThread\n"));
 982       /* ToDo: check assertion
 983       ASSERT(run_queue_hd != (StgTSO*)NULL &&
 984              run_queue_hd != END_TSO_QUEUE);
 985       */
 986       /* Ignore ContinueThreads for fetching threads (if synchr comm) */
 987       if (!RtsFlags.GranFlags.DoAsyncFetch &&
 988           procStatus[CurrentProc]==Fetching) {
 989         debugBelch("ghuH: Spurious ContinueThread while Fetching ignored; TSO %d (%p) [PE %d]\n",
 990               CurrentTSO->id, CurrentTSO, CurrentProc);
 991         goto next_thread;
 992       }
 993       /* Ignore ContinueThreads for completed threads */
 994       if (CurrentTSO->what_next == ThreadComplete) {
 995         debugBelch("ghuH: found a ContinueThread event for completed thread %d (%p) [PE %d] (ignoring ContinueThread)\n",
 996               CurrentTSO->id, CurrentTSO, CurrentProc);
 997         goto next_thread;
 998       }
 999       /* Ignore ContinueThreads for threads that are being migrated */
1000       if (PROCS(CurrentTSO)==Nowhere) {
1001         debugBelch("ghuH: trying to run the migrating TSO %d (%p) [PE %d] (ignoring ContinueThread)\n",
1002               CurrentTSO->id, CurrentTSO, CurrentProc);
1003         goto next_thread;
1004       }
1005       /* The thread should be at the beginning of the run queue */
1006       if (CurrentTSO!=run_queue_hds[CurrentProc]) {
1007         debugBelch("ghuH: TSO %d (%p) [PE %d] is not at the start of the run_queue when doing a ContinueThread\n",
1008               CurrentTSO->id, CurrentTSO, CurrentProc);
1009         break; // run the thread anyway
1010       }
1011       /*
1012       new_event(proc, proc, CurrentTime[proc],
1013                 FindWork,
1014                 (StgTSO*)NULL, (StgClosure*)NULL, (rtsSpark*)NULL);
1015       goto next_thread;
1016       */ /* Catches superfluous CONTINUEs -- should be unnecessary */
1017       break; // now actually run the thread; DaH Qu'vam yImuHbej
1018
1019     case FetchNode:
1020       do_the_fetchnode(event);
1021       goto next_thread;             /* handle next event in event queue  */
1022
1023     case GlobalBlock:
1024       do_the_globalblock(event);
1025       goto next_thread;             /* handle next event in event queue  */
1026
1027     case FetchReply:
1028       do_the_fetchreply(event);
1029       goto next_thread;             /* handle next event in event queue  */
1030
1031     case UnblockThread:   /* Move from the blocked queue to the tail of */
1032       do_the_unblock(event);
1033       goto next_thread;             /* handle next event in event queue  */
1034
1035     case ResumeThread:  /* Move from the blocked queue to the tail of */
1036       /* the runnable queue ( i.e. Qu' SImqa'lu') */
1037       event->tso->gran.blocktime +=
1038         CurrentTime[CurrentProc] - event->tso->gran.blockedat;
1039       do_the_startthread(event);
1040       goto next_thread;             /* handle next event in event queue  */
1041
1042     case StartThread:
1043       do_the_startthread(event);
1044       goto next_thread;             /* handle next event in event queue  */
1045
1046     case MoveThread:
1047       do_the_movethread(event);
1048       goto next_thread;             /* handle next event in event queue  */
1049
1050     case MoveSpark:
1051       do_the_movespark(event);
1052       goto next_thread;             /* handle next event in event queue  */
1053
1054     case FindWork:
1055       do_the_findwork(event);
1056       goto next_thread;             /* handle next event in event queue  */
1057
1058     default:
1059       barf("Illegal event type %u\n", event->evttype);
1060     }  /* switch */
1061
1062     /* This point was scheduler_loop in the old RTS */
1063
1064     IF_DEBUG(gran, debugBelch("GRAN: after main switch\n"));
1065
1066     TimeOfLastEvent = CurrentTime[CurrentProc];
1067     TimeOfNextEvent = get_time_of_next_event();
1068     IgnoreEvents=(TimeOfNextEvent==0); // HWL HACK
1069     // CurrentTSO = ThreadQueueHd;
1070
1071     IF_DEBUG(gran, debugBelch("GRAN: time of next event is: %ld\n",
1072                          TimeOfNextEvent));
1073
1074     if (RtsFlags.GranFlags.Light)
1075       GranSimLight_leave_system(event, &ActiveTSO);
1076
1077     EndOfTimeSlice = CurrentTime[CurrentProc]+RtsFlags.GranFlags.time_slice;
1078
1079     IF_DEBUG(gran,
1080              debugBelch("GRAN: end of time-slice is %#lx\n", EndOfTimeSlice));
1081
1082     /* in a GranSim setup the TSO stays on the run queue */
1083     t = CurrentTSO;
1084     /* Take a thread from the run queue. */
1085     POP_RUN_QUEUE(t); // take_off_run_queue(t);
1086
1087     IF_DEBUG(gran,
1088              debugBelch("GRAN: About to run current thread, which is\n");
1089              G_TSO(t,5));
1090
1091     context_switch = 0; // turned on via GranYield, checking events and time slice
1092
1093     IF_DEBUG(gran,
1094              DumpGranEvent(GR_SCHEDULE, t));
1095
1096     procStatus[CurrentProc] = Busy;
1097 }
1098 #endif // GRAN
1099
1100 /* ----------------------------------------------------------------------------
1101  * Send pending messages (PARALLEL_HASKELL only)
1102  * ------------------------------------------------------------------------- */
1103
1104 #if defined(PARALLEL_HASKELL)
1105 static StgTSO *
1106 scheduleSendPendingMessages(void)
1107 {
1108     StgSparkPool *pool;
1109     rtsSpark spark;
1110     StgTSO *t;
1111
1112 # if defined(PAR) // global Mem.Mgmt., omit for now
1113     if (PendingFetches != END_BF_QUEUE) {
1114         processFetches();
1115     }
1116 # endif
1117
1118     if (RtsFlags.ParFlags.BufferTime) {
1119         // if we use message buffering, we must send away all message
1120         // packets which have become too old...
1121         sendOldBuffers();
1122     }
1123 }
1124 #endif
1125
1126 /* ----------------------------------------------------------------------------
1127  * Activate spark threads (PARALLEL_HASKELL only)
1128  * ------------------------------------------------------------------------- */
1129
1130 #if defined(PARALLEL_HASKELL)
1131 static void
1132 scheduleActivateSpark(void)
1133 {
1134 #if defined(SPARKS)
1135   ASSERT(EMPTY_RUN_QUEUE());
1136 /* We get here if the run queue is empty and want some work.
1137    We try to turn a spark into a thread, and add it to the run queue,
1138    from where it will be picked up in the next iteration of the scheduler
1139    loop.
1140 */
1141
1142       /* :-[  no local threads => look out for local sparks */
1143       /* the spark pool for the current PE */
1144       pool = &(cap.r.rSparks); // JB: cap = (old) MainCap
1145       if (advisory_thread_count < RtsFlags.ParFlags.maxThreads &&
1146           pool->hd < pool->tl) {
1147         /*
1148          * ToDo: add GC code check that we really have enough heap afterwards!!
1149          * Old comment:
1150          * If we're here (no runnable threads) and we have pending
1151          * sparks, we must have a space problem.  Get enough space
1152          * to turn one of those pending sparks into a
1153          * thread...
1154          */
1155
1156         spark = findSpark(rtsFalse);            /* get a spark */
1157         if (spark != (rtsSpark) NULL) {
1158           tso = createThreadFromSpark(spark);       /* turn the spark into a thread */
1159           IF_PAR_DEBUG(fish, // schedule,
1160                        debugBelch("==== schedule: Created TSO %d (%p); %d threads active\n",
1161                              tso->id, tso, advisory_thread_count));
1162
1163           if (tso==END_TSO_QUEUE) { /* failed to activate spark->back to loop */
1164             IF_PAR_DEBUG(fish, // schedule,
1165                          debugBelch("==^^ failed to create thread from spark @ %lx\n",
1166                             spark));
1167             return rtsFalse; /* failed to generate a thread */
1168           }                  /* otherwise fall through & pick-up new tso */
1169         } else {
1170           IF_PAR_DEBUG(fish, // schedule,
1171                        debugBelch("==^^ no local sparks (spark pool contains only NFs: %d)\n",
1172                              spark_queue_len(pool)));
1173           return rtsFalse;  /* failed to generate a thread */
1174         }
1175         return rtsTrue;  /* success in generating a thread */
1176   } else { /* no more threads permitted or pool empty */
1177     return rtsFalse;  /* failed to generateThread */
1178   }
1179 #else
1180   tso = NULL; // avoid compiler warning only
1181   return rtsFalse;  /* dummy in non-PAR setup */
1182 #endif // SPARKS
1183 }
1184 #endif // PARALLEL_HASKELL
1185
1186 /* ----------------------------------------------------------------------------
1187  * Get work from a remote node (PARALLEL_HASKELL only)
1188  * ------------------------------------------------------------------------- */
1189
1190 #if defined(PARALLEL_HASKELL)
1191 static rtsBool
1192 scheduleGetRemoteWork(rtsBool *receivedFinish)
1193 {
1194   ASSERT(EMPTY_RUN_QUEUE());
1195
1196   if (RtsFlags.ParFlags.BufferTime) {
1197         IF_PAR_DEBUG(verbose,
1198                 debugBelch("...send all pending data,"));
1199         {
1200           nat i;
1201           for (i=1; i<=nPEs; i++)
1202             sendImmediately(i); // send all messages away immediately
1203         }
1204   }
1205 # ifndef SPARKS
1206         //++EDEN++ idle() , i.e. send all buffers, wait for work
1207         // suppress fishing in EDEN... just look for incoming messages
1208         // (blocking receive)
1209   IF_PAR_DEBUG(verbose,
1210                debugBelch("...wait for incoming messages...\n"));
1211   *receivedFinish = processMessages(); // blocking receive...
1212
1213         // and reenter scheduling loop after having received something
1214         // (return rtsFalse below)
1215
1216 # else /* activate SPARKS machinery */
1217 /* We get here, if we have no work, tried to activate a local spark, but still
1218    have no work. We try to get a remote spark, by sending a FISH message.
1219    Thread migration should be added here, and triggered when a sequence of
1220    fishes returns without work. */
1221         delay = (RtsFlags.ParFlags.fishDelay!=0ll ? RtsFlags.ParFlags.fishDelay : 0ll);
1222
1223       /* =8-[  no local sparks => look for work on other PEs */
1224         /*
1225          * We really have absolutely no work.  Send out a fish
1226          * (there may be some out there already), and wait for
1227          * something to arrive.  We clearly can't run any threads
1228          * until a SCHEDULE or RESUME arrives, and so that's what
1229          * we're hoping to see.  (Of course, we still have to
1230          * respond to other types of messages.)
1231          */
1232         rtsTime now = msTime() /*CURRENT_TIME*/;
1233         IF_PAR_DEBUG(verbose,
1234                      debugBelch("--  now=%ld\n", now));
1235         IF_PAR_DEBUG(fish, // verbose,
1236              if (outstandingFishes < RtsFlags.ParFlags.maxFishes &&
1237                  (last_fish_arrived_at!=0 &&
1238                   last_fish_arrived_at+delay > now)) {
1239                debugBelch("--$$ <%llu> delaying FISH until %llu (last fish %llu, delay %llu)\n",
1240                      now, last_fish_arrived_at+delay,
1241                      last_fish_arrived_at,
1242                      delay);
1243              });
1244
1245         if (outstandingFishes < RtsFlags.ParFlags.maxFishes &&
1246             advisory_thread_count < RtsFlags.ParFlags.maxThreads) { // send a FISH, but when?
1247           if (last_fish_arrived_at==0 ||
1248               (last_fish_arrived_at+delay <= now)) {           // send FISH now!
1249             /* outstandingFishes is set in sendFish, processFish;
1250                avoid flooding system with fishes via delay */
1251     next_fish_to_send_at = 0;
1252   } else {
1253     /* ToDo: this should be done in the main scheduling loop to avoid the
1254              busy wait here; not so bad if fish delay is very small  */
1255     int iq = 0; // DEBUGGING -- HWL
1256     next_fish_to_send_at = last_fish_arrived_at+delay; // remember when to send
1257     /* send a fish when ready, but process messages that arrive in the meantime */
1258     do {
1259       if (PacketsWaiting()) {
1260         iq++; // DEBUGGING
1261         *receivedFinish = processMessages();
1262       }
1263       now = msTime();
1264     } while (!*receivedFinish || now<next_fish_to_send_at);
1265     // JB: This means the fish could become obsolete, if we receive
1266     // work. Better check for work again?
1267     // last line: while (!receivedFinish || !haveWork || now<...)
1268     // next line: if (receivedFinish || haveWork )
1269
1270     if (*receivedFinish) // no need to send a FISH if we are finishing anyway
1271       return rtsFalse;  // NB: this will leave scheduler loop
1272                         // immediately after return!
1273
1274     IF_PAR_DEBUG(fish, // verbose,
1275                debugBelch("--$$ <%llu> sent delayed fish (%d processMessages); active/total threads=%d/%d\n",now,iq,run_queue_len(),advisory_thread_count));
1276
1277   }
1278
1279     // JB: IMHO, this should all be hidden inside sendFish(...)
1280     /* pe = choosePE();
1281        sendFish(pe, thisPE, NEW_FISH_AGE, NEW_FISH_HISTORY,
1282                 NEW_FISH_HUNGER);
1283
1284     // Global statistics: count no. of fishes
1285     if (RtsFlags.ParFlags.ParStats.Global &&
1286          RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
1287            globalParStats.tot_fish_mess++;
1288            }
1289     */
1290
1291   /* delayed fishes must have been sent by now! */
1292   next_fish_to_send_at = 0;
1293   }
1294
1295   *receivedFinish = processMessages();
1296 # endif /* SPARKS */
1297
1298  return rtsFalse;
1299  /* NB: this function always returns rtsFalse, meaning the scheduler
1300     loop continues with the next iteration;
1301     rationale:
1302       return code means success in finding work; we enter this function
1303       if there is no local work, thus have to send a fish which takes
1304       time until it arrives with work; in the meantime we should process
1305       messages in the main loop;
1306  */
1307 }
1308 #endif // PARALLEL_HASKELL
1309
1310 /* ----------------------------------------------------------------------------
1311  * PAR/GRAN: Report stats & debugging info(?)
1312  * ------------------------------------------------------------------------- */
1313
1314 #if defined(PAR) || defined(GRAN)
1315 static void
1316 scheduleGranParReport(void)
1317 {
1318   ASSERT(run_queue_hd != END_TSO_QUEUE);
1319
1320   /* Take a thread from the run queue, if we have work */
1321   POP_RUN_QUEUE(t);  // take_off_run_queue(END_TSO_QUEUE);
1322
1323     /* If this TSO has got its outport closed in the meantime,
1324      *   it mustn't be run. Instead, we have to clean it up as if it was finished.
1325      * It has to be marked as TH_DEAD for this purpose.
1326      * If it is TH_TERM instead, it is supposed to have finished in the normal way.
1327
1328 JB: TODO: investigate wether state change field could be nuked
1329      entirely and replaced by the normal tso state (whatnext
1330      field). All we want to do is to kill tsos from outside.
1331      */
1332
1333     /* ToDo: write something to the log-file
1334     if (RTSflags.ParFlags.granSimStats && !sameThread)
1335         DumpGranEvent(GR_SCHEDULE, RunnableThreadsHd);
1336
1337     CurrentTSO = t;
1338     */
1339     /* the spark pool for the current PE */
1340     pool = &(cap.r.rSparks); //  cap = (old) MainCap
1341
1342     IF_DEBUG(scheduler,
1343              debugBelch("--=^ %d threads, %d sparks on [%#x]\n",
1344                    run_queue_len(), spark_queue_len(pool), CURRENT_PROC));
1345
1346     IF_PAR_DEBUG(fish,
1347              debugBelch("--=^ %d threads, %d sparks on [%#x]\n",
1348                    run_queue_len(), spark_queue_len(pool), CURRENT_PROC));
1349
1350     if (RtsFlags.ParFlags.ParStats.Full &&
1351         (t->par.sparkname != (StgInt)0) && // only log spark generated threads
1352         (emitSchedule || // forced emit
1353          (t && LastTSO && t->id != LastTSO->id))) {
1354       /*
1355          we are running a different TSO, so write a schedule event to log file
1356          NB: If we use fair scheduling we also have to write  a deschedule
1357              event for LastTSO; with unfair scheduling we know that the
1358              previous tso has blocked whenever we switch to another tso, so
1359              we don't need it in GUM for now
1360       */
1361       IF_PAR_DEBUG(fish, // schedule,
1362                    debugBelch("____ scheduling spark generated thread %d (%lx) (%lx) via a forced emit\n",t->id,t,t->par.sparkname));
1363
1364       DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
1365                        GR_SCHEDULE, t, (StgClosure *)NULL, 0, 0);
1366       emitSchedule = rtsFalse;
1367     }
1368 }
1369 #endif
1370
1371 /* ----------------------------------------------------------------------------
1372  * After running a thread...
1373  * ASSUMES: sched_mutex
1374  * ------------------------------------------------------------------------- */
1375
1376 static void
1377 schedulePostRunThread(void)
1378 {
1379 #if defined(PAR)
1380     /* HACK 675: if the last thread didn't yield, make sure to print a
1381        SCHEDULE event to the log file when StgRunning the next thread, even
1382        if it is the same one as before */
1383     LastTSO = t;
1384     TimeOfLastYield = CURRENT_TIME;
1385 #endif
1386
1387   /* some statistics gathering in the parallel case */
1388
1389 #if defined(GRAN) || defined(PAR) || defined(EDEN)
1390   switch (ret) {
1391     case HeapOverflow:
1392 # if defined(GRAN)
1393       IF_DEBUG(gran, DumpGranEvent(GR_DESCHEDULE, t));
1394       globalGranStats.tot_heapover++;
1395 # elif defined(PAR)
1396       globalParStats.tot_heapover++;
1397 # endif
1398       break;
1399
1400      case StackOverflow:
1401 # if defined(GRAN)
1402       IF_DEBUG(gran,
1403                DumpGranEvent(GR_DESCHEDULE, t));
1404       globalGranStats.tot_stackover++;
1405 # elif defined(PAR)
1406       // IF_DEBUG(par,
1407       // DumpGranEvent(GR_DESCHEDULE, t);
1408       globalParStats.tot_stackover++;
1409 # endif
1410       break;
1411
1412     case ThreadYielding:
1413 # if defined(GRAN)
1414       IF_DEBUG(gran,
1415                DumpGranEvent(GR_DESCHEDULE, t));
1416       globalGranStats.tot_yields++;
1417 # elif defined(PAR)
1418       // IF_DEBUG(par,
1419       // DumpGranEvent(GR_DESCHEDULE, t);
1420       globalParStats.tot_yields++;
1421 # endif
1422       break;
1423
1424     case ThreadBlocked:
1425 # if defined(GRAN)
1426       IF_DEBUG(scheduler,
1427                debugBelch("--<< thread %ld (%p; %s) stopped, blocking on node %p [PE %d] with BQ: ",
1428                           t->id, t, whatNext_strs[t->what_next], t->block_info.closure,
1429                           (t->block_info.closure==(StgClosure*)NULL ? 99 : where_is(t->block_info.closure)));
1430                if (t->block_info.closure!=(StgClosure*)NULL)
1431                  print_bq(t->block_info.closure);
1432                debugBelch("\n"));
1433
1434       // ??? needed; should emit block before
1435       IF_DEBUG(gran,
1436                DumpGranEvent(GR_DESCHEDULE, t));
1437       prune_eventq(t, (StgClosure *)NULL); // prune ContinueThreads for t
1438       /*
1439         ngoq Dogh!
1440       ASSERT(procStatus[CurrentProc]==Busy ||
1441               ((procStatus[CurrentProc]==Fetching) &&
1442               (t->block_info.closure!=(StgClosure*)NULL)));
1443       if (run_queue_hds[CurrentProc] == END_TSO_QUEUE &&
1444           !(!RtsFlags.GranFlags.DoAsyncFetch &&
1445             procStatus[CurrentProc]==Fetching))
1446         procStatus[CurrentProc] = Idle;
1447       */
1448 # elif defined(PAR)
1449 //++PAR++  blockThread() writes the event (change?)
1450 # endif
1451     break;
1452
1453   case ThreadFinished:
1454     break;
1455
1456   default:
1457     barf("parGlobalStats: unknown return code");
1458     break;
1459     }
1460 #endif
1461 }
1462
1463 /* -----------------------------------------------------------------------------
1464  * Handle a thread that returned to the scheduler with ThreadHeepOverflow
1465  * ASSUMES: sched_mutex
1466  * -------------------------------------------------------------------------- */
1467
1468 static rtsBool
1469 scheduleHandleHeapOverflow( Capability *cap, StgTSO *t )
1470 {
1471     // did the task ask for a large block?
1472     if (cap->r.rHpAlloc > BLOCK_SIZE) {
1473         // if so, get one and push it on the front of the nursery.
1474         bdescr *bd;
1475         lnat blocks;
1476
1477         blocks = (lnat)BLOCK_ROUND_UP(cap->r.rHpAlloc) / BLOCK_SIZE;
1478
1479         IF_DEBUG(scheduler,
1480                  debugBelch("--<< thread %ld (%s) stopped: requesting a large block (size %ld)\n",
1481                             (long)t->id, whatNext_strs[t->what_next], blocks));
1482
1483         // don't do this if the nursery is (nearly) full, we'll GC first.
1484         if (cap->r.rCurrentNursery->link != NULL ||
1485             cap->r.rNursery->n_blocks == 1) {  // paranoia to prevent infinite loop
1486                                                // if the nursery has only one block.
1487
1488             bd = allocGroup( blocks );
1489             cap->r.rNursery->n_blocks += blocks;
1490
1491             // link the new group into the list
1492             bd->link = cap->r.rCurrentNursery;
1493             bd->u.back = cap->r.rCurrentNursery->u.back;
1494             if (cap->r.rCurrentNursery->u.back != NULL) {
1495                 cap->r.rCurrentNursery->u.back->link = bd;
1496             } else {
1497 #if !defined(SMP)
1498                 ASSERT(g0s0->blocks == cap->r.rCurrentNursery &&
1499                        g0s0 == cap->r.rNursery);
1500 #endif
1501                 cap->r.rNursery->blocks = bd;
1502             }
1503             cap->r.rCurrentNursery->u.back = bd;
1504
1505             // initialise it as a nursery block.  We initialise the
1506             // step, gen_no, and flags field of *every* sub-block in
1507             // this large block, because this is easier than making
1508             // sure that we always find the block head of a large
1509             // block whenever we call Bdescr() (eg. evacuate() and
1510             // isAlive() in the GC would both have to do this, at
1511             // least).
1512             {
1513                 bdescr *x;
1514                 for (x = bd; x < bd + blocks; x++) {
1515                     x->step = cap->r.rNursery;
1516                     x->gen_no = 0;
1517                     x->flags = 0;
1518                 }
1519             }
1520
1521             // This assert can be a killer if the app is doing lots
1522             // of large block allocations.
1523             IF_DEBUG(sanity, checkNurserySanity(cap->r.rNursery));
1524
1525             // now update the nursery to point to the new block
1526             cap->r.rCurrentNursery = bd;
1527
1528             // we might be unlucky and have another thread get on the
1529             // run queue before us and steal the large block, but in that
1530             // case the thread will just end up requesting another large
1531             // block.
1532             PUSH_ON_RUN_QUEUE(t);
1533             return rtsFalse;  /* not actually GC'ing */
1534         }
1535     }
1536
1537     IF_DEBUG(scheduler,
1538              debugBelch("--<< thread %ld (%s) stopped: HeapOverflow\n",
1539                         (long)t->id, whatNext_strs[t->what_next]));
1540 #if defined(GRAN)
1541     ASSERT(!is_on_queue(t,CurrentProc));
1542 #elif defined(PARALLEL_HASKELL)
1543     /* Currently we emit a DESCHEDULE event before GC in GUM.
1544        ToDo: either add separate event to distinguish SYSTEM time from rest
1545        or just nuke this DESCHEDULE (and the following SCHEDULE) */
1546     if (0 && RtsFlags.ParFlags.ParStats.Full) {
1547         DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
1548                          GR_DESCHEDULE, t, (StgClosure *)NULL, 0, 0);
1549         emitSchedule = rtsTrue;
1550     }
1551 #endif
1552
1553     PUSH_ON_RUN_QUEUE(t);
1554     return rtsTrue;
1555     /* actual GC is done at the end of the while loop in schedule() */
1556 }
1557
1558 /* -----------------------------------------------------------------------------
1559  * Handle a thread that returned to the scheduler with ThreadStackOverflow
1560  * ASSUMES: sched_mutex
1561  * -------------------------------------------------------------------------- */
1562
1563 static void
1564 scheduleHandleStackOverflow( StgTSO *t)
1565 {
1566     IF_DEBUG(scheduler,debugBelch("--<< thread %ld (%s) stopped, StackOverflow\n",
1567                                   (long)t->id, whatNext_strs[t->what_next]));
1568     /* just adjust the stack for this thread, then pop it back
1569      * on the run queue.
1570      */
1571     {
1572         /* enlarge the stack */
1573         StgTSO *new_t = threadStackOverflow(t);
1574
1575         /* This TSO has moved, so update any pointers to it from the
1576          * main thread stack.  It better not be on any other queues...
1577          * (it shouldn't be).
1578          */
1579         if (t->main != NULL) {
1580             t->main->tso = new_t;
1581         }
1582         PUSH_ON_RUN_QUEUE(new_t);
1583     }
1584 }
1585
1586 /* -----------------------------------------------------------------------------
1587  * Handle a thread that returned to the scheduler with ThreadYielding
1588  * ASSUMES: sched_mutex
1589  * -------------------------------------------------------------------------- */
1590
1591 static rtsBool
1592 scheduleHandleYield( StgTSO *t, nat prev_what_next )
1593 {
1594     // Reset the context switch flag.  We don't do this just before
1595     // running the thread, because that would mean we would lose ticks
1596     // during GC, which can lead to unfair scheduling (a thread hogs
1597     // the CPU because the tick always arrives during GC).  This way
1598     // penalises threads that do a lot of allocation, but that seems
1599     // better than the alternative.
1600     context_switch = 0;
1601
1602     /* put the thread back on the run queue.  Then, if we're ready to
1603      * GC, check whether this is the last task to stop.  If so, wake
1604      * up the GC thread.  getThread will block during a GC until the
1605      * GC is finished.
1606      */
1607     IF_DEBUG(scheduler,
1608              if (t->what_next != prev_what_next) {
1609                  debugBelch("--<< thread %ld (%s) stopped to switch evaluators\n",
1610                             (long)t->id, whatNext_strs[t->what_next]);
1611              } else {
1612                  debugBelch("--<< thread %ld (%s) stopped, yielding\n",
1613                             (long)t->id, whatNext_strs[t->what_next]);
1614              }
1615         );
1616
1617     IF_DEBUG(sanity,
1618              //debugBelch("&& Doing sanity check on yielding TSO %ld.", t->id);
1619              checkTSO(t));
1620     ASSERT(t->link == END_TSO_QUEUE);
1621
1622     // Shortcut if we're just switching evaluators: don't bother
1623     // doing stack squeezing (which can be expensive), just run the
1624     // thread.
1625     if (t->what_next != prev_what_next) {
1626         return rtsTrue;
1627     }
1628
1629 #if defined(GRAN)
1630     ASSERT(!is_on_queue(t,CurrentProc));
1631
1632     IF_DEBUG(sanity,
1633              //debugBelch("&& Doing sanity check on all ThreadQueues (and their TSOs).");
1634              checkThreadQsSanity(rtsTrue));
1635
1636 #endif
1637
1638     addToRunQueue(t);
1639
1640 #if defined(GRAN)
1641     /* add a ContinueThread event to actually process the thread */
1642     new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
1643               ContinueThread,
1644               t, (StgClosure*)NULL, (rtsSpark*)NULL);
1645     IF_GRAN_DEBUG(bq,
1646                   debugBelch("GRAN: eventq and runnableq after adding yielded thread to queue again:\n");
1647                   G_EVENTQ(0);
1648                   G_CURR_THREADQ(0));
1649 #endif
1650     return rtsFalse;
1651 }
1652
1653 /* -----------------------------------------------------------------------------
1654  * Handle a thread that returned to the scheduler with ThreadBlocked
1655  * ASSUMES: sched_mutex
1656  * -------------------------------------------------------------------------- */
1657
1658 static void
1659 scheduleHandleThreadBlocked( StgTSO *t
1660 #if !defined(GRAN) && !defined(DEBUG)
1661     STG_UNUSED
1662 #endif
1663     )
1664 {
1665 #if defined(GRAN)
1666     IF_DEBUG(scheduler,
1667              debugBelch("--<< thread %ld (%p; %s) stopped, blocking on node %p [PE %d] with BQ: \n",
1668                         t->id, t, whatNext_strs[t->what_next], t->block_info.closure, (t->block_info.closure==(StgClosure*)NULL ? 99 : where_is(t->block_info.closure)));
1669              if (t->block_info.closure!=(StgClosure*)NULL) print_bq(t->block_info.closure));
1670
1671     // ??? needed; should emit block before
1672     IF_DEBUG(gran,
1673              DumpGranEvent(GR_DESCHEDULE, t));
1674     prune_eventq(t, (StgClosure *)NULL); // prune ContinueThreads for t
1675     /*
1676       ngoq Dogh!
1677       ASSERT(procStatus[CurrentProc]==Busy ||
1678       ((procStatus[CurrentProc]==Fetching) &&
1679       (t->block_info.closure!=(StgClosure*)NULL)));
1680       if (run_queue_hds[CurrentProc] == END_TSO_QUEUE &&
1681       !(!RtsFlags.GranFlags.DoAsyncFetch &&
1682       procStatus[CurrentProc]==Fetching))
1683       procStatus[CurrentProc] = Idle;
1684     */
1685 #elif defined(PAR)
1686     IF_DEBUG(scheduler,
1687              debugBelch("--<< thread %ld (%p; %s) stopped, blocking on node %p with BQ: \n",
1688                         t->id, t, whatNext_strs[t->what_next], t->block_info.closure));
1689     IF_PAR_DEBUG(bq,
1690
1691                  if (t->block_info.closure!=(StgClosure*)NULL)
1692                  print_bq(t->block_info.closure));
1693
1694     /* Send a fetch (if BlockedOnGA) and dump event to log file */
1695     blockThread(t);
1696
1697     /* whatever we schedule next, we must log that schedule */
1698     emitSchedule = rtsTrue;
1699
1700 #else /* !GRAN */
1701
1702       // We don't need to do anything.  The thread is blocked, and it
1703       // has tidied up its stack and placed itself on whatever queue
1704       // it needs to be on.
1705
1706 #if !defined(SMP)
1707     ASSERT(t->why_blocked != NotBlocked);
1708              // This might not be true under SMP: we don't have
1709              // exclusive access to this TSO, so someone might have
1710              // woken it up by now.  This actually happens: try
1711              // conc023 +RTS -N2.
1712 #endif
1713
1714     IF_DEBUG(scheduler,
1715              debugBelch("--<< thread %d (%s) stopped: ",
1716                         t->id, whatNext_strs[t->what_next]);
1717              printThreadBlockage(t);
1718              debugBelch("\n"));
1719
1720     /* Only for dumping event to log file
1721        ToDo: do I need this in GranSim, too?
1722        blockThread(t);
1723     */
1724 #endif
1725 }
1726
1727 /* -----------------------------------------------------------------------------
1728  * Handle a thread that returned to the scheduler with ThreadFinished
1729  * ASSUMES: sched_mutex
1730  * -------------------------------------------------------------------------- */
1731
1732 static rtsBool
1733 scheduleHandleThreadFinished( StgMainThread *mainThread
1734                               USED_WHEN_RTS_SUPPORTS_THREADS,
1735                               Capability *cap,
1736                               StgTSO *t )
1737 {
1738     /* Need to check whether this was a main thread, and if so,
1739      * return with the return value.
1740      *
1741      * We also end up here if the thread kills itself with an
1742      * uncaught exception, see Exception.cmm.
1743      */
1744     IF_DEBUG(scheduler,debugBelch("--++ thread %d (%s) finished\n",
1745                                   t->id, whatNext_strs[t->what_next]));
1746
1747 #if defined(GRAN)
1748       endThread(t, CurrentProc); // clean-up the thread
1749 #elif defined(PARALLEL_HASKELL)
1750       /* For now all are advisory -- HWL */
1751       //if(t->priority==AdvisoryPriority) ??
1752       advisory_thread_count--; // JB: Caution with this counter, buggy!
1753
1754 # if defined(DIST)
1755       if(t->dist.priority==RevalPriority)
1756         FinishReval(t);
1757 # endif
1758
1759 # if defined(EDENOLD)
1760       // the thread could still have an outport... (BUG)
1761       if (t->eden.outport != -1) {
1762       // delete the outport for the tso which has finished...
1763         IF_PAR_DEBUG(eden_ports,
1764                    debugBelch("WARNING: Scheduler removes outport %d for TSO %d.\n",
1765                               t->eden.outport, t->id));
1766         deleteOPT(t);
1767       }
1768       // thread still in the process (HEAVY BUG! since outport has just been closed...)
1769       if (t->eden.epid != -1) {
1770         IF_PAR_DEBUG(eden_ports,
1771                    debugBelch("WARNING: Scheduler removes TSO %d from process %d .\n",
1772                            t->id, t->eden.epid));
1773         removeTSOfromProcess(t);
1774       }
1775 # endif
1776
1777 # if defined(PAR)
1778       if (RtsFlags.ParFlags.ParStats.Full &&
1779           !RtsFlags.ParFlags.ParStats.Suppressed)
1780         DumpEndEvent(CURRENT_PROC, t, rtsFalse /* not mandatory */);
1781
1782       //  t->par only contains statistics: left out for now...
1783       IF_PAR_DEBUG(fish,
1784                    debugBelch("**** end thread: ended sparked thread %d (%lx); sparkname: %lx\n",
1785                               t->id,t,t->par.sparkname));
1786 # endif
1787 #endif // PARALLEL_HASKELL
1788
1789       //
1790       // Check whether the thread that just completed was a main
1791       // thread, and if so return with the result.
1792       //
1793       // There is an assumption here that all thread completion goes
1794       // through this point; we need to make sure that if a thread
1795       // ends up in the ThreadKilled state, that it stays on the run
1796       // queue so it can be dealt with here.
1797       //
1798       if (
1799 #if defined(RTS_SUPPORTS_THREADS)
1800           mainThread != NULL
1801 #else
1802           mainThread->tso == t
1803 #endif
1804           )
1805       {
1806           // We are a bound thread: this must be our thread that just
1807           // completed.
1808           ASSERT(mainThread->tso == t);
1809
1810           if (t->what_next == ThreadComplete) {
1811               if (mainThread->ret) {
1812                   // NOTE: return val is tso->sp[1] (see StgStartup.hc)
1813                   *(mainThread->ret) = (StgClosure *)mainThread->tso->sp[1];
1814               }
1815               mainThread->stat = Success;
1816           } else {
1817               if (mainThread->ret) {
1818                   *(mainThread->ret) = NULL;
1819               }
1820               if (interrupted) {
1821                   mainThread->stat = Interrupted;
1822               } else {
1823                   mainThread->stat = Killed;
1824               }
1825           }
1826 #ifdef DEBUG
1827           removeThreadLabel((StgWord)mainThread->tso->id);
1828 #endif
1829           if (mainThread->prev == NULL) {
1830               ASSERT(mainThread == main_threads);
1831               main_threads = mainThread->link;
1832           } else {
1833               mainThread->prev->link = mainThread->link;
1834           }
1835           if (mainThread->link != NULL) {
1836               mainThread->link->prev = mainThread->prev;
1837           }
1838           releaseCapability(cap);
1839           return rtsTrue; // tells schedule() to return
1840       }
1841
1842 #ifdef RTS_SUPPORTS_THREADS
1843       ASSERT(t->main == NULL);
1844 #else
1845       if (t->main != NULL) {
1846           // Must be a main thread that is not the topmost one.  Leave
1847           // it on the run queue until the stack has unwound to the
1848           // point where we can deal with this.  Leaving it on the run
1849           // queue also ensures that the garbage collector knows about
1850           // this thread and its return value (it gets dropped from the
1851           // all_threads list so there's no other way to find it).
1852           APPEND_TO_RUN_QUEUE(t);
1853       }
1854 #endif
1855       return rtsFalse;
1856 }
1857
1858 /* -----------------------------------------------------------------------------
1859  * Perform a heap census, if PROFILING
1860  * -------------------------------------------------------------------------- */
1861
1862 static rtsBool
1863 scheduleDoHeapProfile( rtsBool ready_to_gc STG_UNUSED )
1864 {
1865 #if defined(PROFILING)
1866     // When we have +RTS -i0 and we're heap profiling, do a census at
1867     // every GC.  This lets us get repeatable runs for debugging.
1868     if (performHeapProfile ||
1869         (RtsFlags.ProfFlags.profileInterval==0 &&
1870          RtsFlags.ProfFlags.doHeapProfile && ready_to_gc)) {
1871         GarbageCollect(GetRoots, rtsTrue);
1872         heapCensus();
1873         performHeapProfile = rtsFalse;
1874         return rtsTrue;  // true <=> we already GC'd
1875     }
1876 #endif
1877     return rtsFalse;
1878 }
1879
1880 /* -----------------------------------------------------------------------------
1881  * Perform a garbage collection if necessary
1882  * ASSUMES: sched_mutex
1883  * -------------------------------------------------------------------------- */
1884
1885 static void
1886 scheduleDoGC( Capability *cap STG_UNUSED )
1887 {
1888     StgTSO *t;
1889 #ifdef SMP
1890     static rtsBool waiting_for_gc;
1891     int n_capabilities = RtsFlags.ParFlags.nNodes - 1;
1892            // subtract one because we're already holding one.
1893     Capability *caps[n_capabilities];
1894 #endif
1895
1896 #ifdef SMP
1897     // In order to GC, there must be no threads running Haskell code.
1898     // Therefore, the GC thread needs to hold *all* the capabilities,
1899     // and release them after the GC has completed.
1900     //
1901     // This seems to be the simplest way: previous attempts involved
1902     // making all the threads with capabilities give up their
1903     // capabilities and sleep except for the *last* one, which
1904     // actually did the GC.  But it's quite hard to arrange for all
1905     // the other tasks to sleep and stay asleep.
1906     //
1907
1908     // Someone else is already trying to GC
1909     if (waiting_for_gc) return;
1910     waiting_for_gc = rtsTrue;
1911
1912     caps[n_capabilities] = cap;
1913     while (n_capabilities > 0) {
1914         IF_DEBUG(scheduler, sched_belch("ready_to_gc, grabbing all the capabilies (%d left)", n_capabilities));
1915         waitForReturnCapability(&sched_mutex, &cap);
1916         n_capabilities--;
1917         caps[n_capabilities] = cap;
1918     }
1919
1920     waiting_for_gc = rtsFalse;
1921 #endif
1922
1923     /* Kick any transactions which are invalid back to their
1924      * atomically frames.  When next scheduled they will try to
1925      * commit, this commit will fail and they will retry.
1926      */
1927     for (t = all_threads; t != END_TSO_QUEUE; t = t -> link) {
1928         if (t -> what_next != ThreadRelocated && t -> trec != NO_TREC && t -> why_blocked == NotBlocked) {
1929             if (!stmValidateTransaction (t -> trec)) {
1930                 IF_DEBUG(stm, sched_belch("trec %p found wasting its time", t));
1931
1932                 // strip the stack back to the ATOMICALLY_FRAME, aborting
1933                 // the (nested) transaction, and saving the stack of any
1934                 // partially-evaluated thunks on the heap.
1935                 raiseAsync_(t, NULL, rtsTrue);
1936
1937 #ifdef REG_R1
1938                 ASSERT(get_itbl((StgClosure *)t->sp)->type == ATOMICALLY_FRAME);
1939 #endif
1940             }
1941         }
1942     }
1943
1944     // so this happens periodically:
1945     scheduleCheckBlackHoles();
1946
1947     IF_DEBUG(scheduler, printAllThreads());
1948
1949     /* everybody back, start the GC.
1950      * Could do it in this thread, or signal a condition var
1951      * to do it in another thread.  Either way, we need to
1952      * broadcast on gc_pending_cond afterward.
1953      */
1954 #if defined(RTS_SUPPORTS_THREADS)
1955     IF_DEBUG(scheduler,sched_belch("doing GC"));
1956 #endif
1957     GarbageCollect(GetRoots,rtsFalse);
1958
1959 #if defined(SMP)
1960     {
1961         // release our stash of capabilities.
1962         nat i;
1963         for (i = 0; i < RtsFlags.ParFlags.nNodes-1; i++) {
1964             releaseCapability(caps[i]);
1965         }
1966     }
1967 #endif
1968
1969 #if defined(GRAN)
1970     /* add a ContinueThread event to continue execution of current thread */
1971     new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
1972               ContinueThread,
1973               t, (StgClosure*)NULL, (rtsSpark*)NULL);
1974     IF_GRAN_DEBUG(bq,
1975                   debugBelch("GRAN: eventq and runnableq after Garbage collection:\n\n");
1976                   G_EVENTQ(0);
1977                   G_CURR_THREADQ(0));
1978 #endif /* GRAN */
1979 }
1980
1981 /* ---------------------------------------------------------------------------
1982  * rtsSupportsBoundThreads(): is the RTS built to support bound threads?
1983  * used by Control.Concurrent for error checking.
1984  * ------------------------------------------------------------------------- */
1985
1986 StgBool
1987 rtsSupportsBoundThreads(void)
1988 {
1989 #if defined(RTS_SUPPORTS_THREADS)
1990   return rtsTrue;
1991 #else
1992   return rtsFalse;
1993 #endif
1994 }
1995
1996 /* ---------------------------------------------------------------------------
1997  * isThreadBound(tso): check whether tso is bound to an OS thread.
1998  * ------------------------------------------------------------------------- */
1999
2000 StgBool
2001 isThreadBound(StgTSO* tso USED_IN_THREADED_RTS)
2002 {
2003 #if defined(RTS_SUPPORTS_THREADS)
2004   return (tso->main != NULL);
2005 #endif
2006   return rtsFalse;
2007 }
2008
2009 /* ---------------------------------------------------------------------------
2010  * Singleton fork(). Do not copy any running threads.
2011  * ------------------------------------------------------------------------- */
2012
2013 #ifndef mingw32_HOST_OS
2014 #define FORKPROCESS_PRIMOP_SUPPORTED
2015 #endif
2016
2017 #ifdef FORKPROCESS_PRIMOP_SUPPORTED
2018 static void
2019 deleteThreadImmediately(StgTSO *tso);
2020 #endif
2021 StgInt
2022 forkProcess(HsStablePtr *entry
2023 #ifndef FORKPROCESS_PRIMOP_SUPPORTED
2024             STG_UNUSED
2025 #endif
2026            )
2027 {
2028 #ifdef FORKPROCESS_PRIMOP_SUPPORTED
2029   pid_t pid;
2030   StgTSO* t,*next;
2031   StgMainThread *m;
2032   SchedulerStatus rc;
2033
2034   IF_DEBUG(scheduler,sched_belch("forking!"));
2035   rts_lock(); // This not only acquires sched_mutex, it also
2036               // makes sure that no other threads are running
2037
2038   pid = fork();
2039
2040   if (pid) { /* parent */
2041
2042   /* just return the pid */
2043     rts_unlock();
2044     return pid;
2045
2046   } else { /* child */
2047
2048
2049       // delete all threads
2050     run_queue_hd = run_queue_tl = END_TSO_QUEUE;
2051
2052     for (t = all_threads; t != END_TSO_QUEUE; t = next) {
2053       next = t->link;
2054
2055         // don't allow threads to catch the ThreadKilled exception
2056       deleteThreadImmediately(t);
2057     }
2058
2059       // wipe the main thread list
2060     while((m = main_threads) != NULL) {
2061       main_threads = m->link;
2062 # ifdef THREADED_RTS
2063       closeCondition(&m->bound_thread_cond);
2064 # endif
2065       stgFree(m);
2066     }
2067
2068     rc = rts_evalStableIO(entry, NULL);  // run the action
2069     rts_checkSchedStatus("forkProcess",rc);
2070
2071     rts_unlock();
2072
2073     hs_exit();                      // clean up and exit
2074     stg_exit(0);
2075   }
2076 #else /* !FORKPROCESS_PRIMOP_SUPPORTED */
2077   barf("forkProcess#: primop not supported, sorry!\n");
2078   return -1;
2079 #endif
2080 }
2081
2082 /* ---------------------------------------------------------------------------
2083  * deleteAllThreads():  kill all the live threads.
2084  *
2085  * This is used when we catch a user interrupt (^C), before performing
2086  * any necessary cleanups and running finalizers.
2087  *
2088  * Locks: sched_mutex held.
2089  * ------------------------------------------------------------------------- */
2090
2091 void
2092 deleteAllThreads ( void )
2093 {
2094   StgTSO* t, *next;
2095   IF_DEBUG(scheduler,sched_belch("deleting all threads"));
2096   for (t = all_threads; t != END_TSO_QUEUE; t = next) {
2097       if (t->what_next == ThreadRelocated) {
2098           next = t->link;
2099       } else {
2100           next = t->global_link;
2101           deleteThread(t);
2102       }
2103   }
2104
2105   // The run queue now contains a bunch of ThreadKilled threads.  We
2106   // must not throw these away: the main thread(s) will be in there
2107   // somewhere, and the main scheduler loop has to deal with it.
2108   // Also, the run queue is the only thing keeping these threads from
2109   // being GC'd, and we don't want the "main thread has been GC'd" panic.
2110
2111   ASSERT(blocked_queue_hd == END_TSO_QUEUE);
2112   ASSERT(blackhole_queue == END_TSO_QUEUE);
2113   ASSERT(sleeping_queue == END_TSO_QUEUE);
2114 }
2115
2116 /* startThread and  insertThread are now in GranSim.c -- HWL */
2117
2118
2119 /* ---------------------------------------------------------------------------
2120  * Suspending & resuming Haskell threads.
2121  *
2122  * When making a "safe" call to C (aka _ccall_GC), the task gives back
2123  * its capability before calling the C function.  This allows another
2124  * task to pick up the capability and carry on running Haskell
2125  * threads.  It also means that if the C call blocks, it won't lock
2126  * the whole system.
2127  *
2128  * The Haskell thread making the C call is put to sleep for the
2129  * duration of the call, on the susepended_ccalling_threads queue.  We
2130  * give out a token to the task, which it can use to resume the thread
2131  * on return from the C function.
2132  * ------------------------------------------------------------------------- */
2133
2134 StgInt
2135 suspendThread( StgRegTable *reg )
2136 {
2137   nat tok;
2138   Capability *cap;
2139   int saved_errno = errno;
2140
2141   /* assume that *reg is a pointer to the StgRegTable part
2142    * of a Capability.
2143    */
2144   cap = (Capability *)((void *)((unsigned char*)reg - sizeof(StgFunTable)));
2145
2146   ACQUIRE_LOCK(&sched_mutex);
2147
2148   IF_DEBUG(scheduler,
2149            sched_belch("thread %d did a _ccall_gc", cap->r.rCurrentTSO->id));
2150
2151   // XXX this might not be necessary --SDM
2152   cap->r.rCurrentTSO->what_next = ThreadRunGHC;
2153
2154   threadPaused(cap->r.rCurrentTSO);
2155   cap->r.rCurrentTSO->link = suspended_ccalling_threads;
2156   suspended_ccalling_threads = cap->r.rCurrentTSO;
2157
2158   if(cap->r.rCurrentTSO->blocked_exceptions == NULL)  {
2159       cap->r.rCurrentTSO->why_blocked = BlockedOnCCall;
2160       cap->r.rCurrentTSO->blocked_exceptions = END_TSO_QUEUE;
2161   } else {
2162       cap->r.rCurrentTSO->why_blocked = BlockedOnCCall_NoUnblockExc;
2163   }
2164
2165   /* Use the thread ID as the token; it should be unique */
2166   tok = cap->r.rCurrentTSO->id;
2167
2168   /* Hand back capability */
2169   cap->r.rInHaskell = rtsFalse;
2170   releaseCapability(cap);
2171
2172 #if defined(RTS_SUPPORTS_THREADS)
2173   /* Preparing to leave the RTS, so ensure there's a native thread/task
2174      waiting to take over.
2175   */
2176   IF_DEBUG(scheduler, sched_belch("worker (token %d): leaving RTS", tok));
2177 #endif
2178
2179   RELEASE_LOCK(&sched_mutex);
2180
2181   errno = saved_errno;
2182   return tok;
2183 }
2184
2185 StgRegTable *
2186 resumeThread( StgInt tok )
2187 {
2188   StgTSO *tso, **prev;
2189   Capability *cap;
2190   int saved_errno = errno;
2191
2192 #if defined(RTS_SUPPORTS_THREADS)
2193   /* Wait for permission to re-enter the RTS with the result. */
2194   ACQUIRE_LOCK(&sched_mutex);
2195   waitForReturnCapability(&sched_mutex, &cap);
2196
2197   IF_DEBUG(scheduler, sched_belch("worker (token %d): re-entering RTS", tok));
2198 #else
2199   grabCapability(&cap);
2200 #endif
2201
2202   /* Remove the thread off of the suspended list */
2203   prev = &suspended_ccalling_threads;
2204   for (tso = suspended_ccalling_threads;
2205        tso != END_TSO_QUEUE;
2206        prev = &tso->link, tso = tso->link) {
2207     if (tso->id == (StgThreadID)tok) {
2208       *prev = tso->link;
2209       break;
2210     }
2211   }
2212   if (tso == END_TSO_QUEUE) {
2213     barf("resumeThread: thread not found");
2214   }
2215   tso->link = END_TSO_QUEUE;
2216
2217   if(tso->why_blocked == BlockedOnCCall) {
2218       awakenBlockedQueueNoLock(tso->blocked_exceptions);
2219       tso->blocked_exceptions = NULL;
2220   }
2221
2222   /* Reset blocking status */
2223   tso->why_blocked  = NotBlocked;
2224
2225   cap->r.rCurrentTSO = tso;
2226   cap->r.rInHaskell = rtsTrue;
2227   RELEASE_LOCK(&sched_mutex);
2228   errno = saved_errno;
2229   return &cap->r;
2230 }
2231
2232 /* ---------------------------------------------------------------------------
2233  * Comparing Thread ids.
2234  *
2235  * This is used from STG land in the implementation of the
2236  * instances of Eq/Ord for ThreadIds.
2237  * ------------------------------------------------------------------------ */
2238
2239 int
2240 cmp_thread(StgPtr tso1, StgPtr tso2)
2241 {
2242   StgThreadID id1 = ((StgTSO *)tso1)->id;
2243   StgThreadID id2 = ((StgTSO *)tso2)->id;
2244
2245   if (id1 < id2) return (-1);
2246   if (id1 > id2) return 1;
2247   return 0;
2248 }
2249
2250 /* ---------------------------------------------------------------------------
2251  * Fetching the ThreadID from an StgTSO.
2252  *
2253  * This is used in the implementation of Show for ThreadIds.
2254  * ------------------------------------------------------------------------ */
2255 int
2256 rts_getThreadId(StgPtr tso)
2257 {
2258   return ((StgTSO *)tso)->id;
2259 }
2260
2261 #ifdef DEBUG
2262 void
2263 labelThread(StgPtr tso, char *label)
2264 {
2265   int len;
2266   void *buf;
2267
2268   /* Caveat: Once set, you can only set the thread name to "" */
2269   len = strlen(label)+1;
2270   buf = stgMallocBytes(len * sizeof(char), "Schedule.c:labelThread()");
2271   strncpy(buf,label,len);
2272   /* Update will free the old memory for us */
2273   updateThreadLabel(((StgTSO *)tso)->id,buf);
2274 }
2275 #endif /* DEBUG */
2276
2277 /* ---------------------------------------------------------------------------
2278    Create a new thread.
2279
2280    The new thread starts with the given stack size.  Before the
2281    scheduler can run, however, this thread needs to have a closure
2282    (and possibly some arguments) pushed on its stack.  See
2283    pushClosure() in Schedule.h.
2284
2285    createGenThread() and createIOThread() (in SchedAPI.h) are
2286    convenient packaged versions of this function.
2287
2288    currently pri (priority) is only used in a GRAN setup -- HWL
2289    ------------------------------------------------------------------------ */
2290 #if defined(GRAN)
2291 /*   currently pri (priority) is only used in a GRAN setup -- HWL */
2292 StgTSO *
2293 createThread(nat size, StgInt pri)
2294 #else
2295 StgTSO *
2296 createThread(nat size)
2297 #endif
2298 {
2299
2300     StgTSO *tso;
2301     nat stack_size;
2302
2303     /* First check whether we should create a thread at all */
2304 #if defined(PARALLEL_HASKELL)
2305   /* check that no more than RtsFlags.ParFlags.maxThreads threads are created */
2306   if (advisory_thread_count >= RtsFlags.ParFlags.maxThreads) {
2307     threadsIgnored++;
2308     debugBelch("{createThread}Daq ghuH: refusing to create another thread; no more than %d threads allowed (currently %d)\n",
2309           RtsFlags.ParFlags.maxThreads, advisory_thread_count);
2310     return END_TSO_QUEUE;
2311   }
2312   threadsCreated++;
2313 #endif
2314
2315 #if defined(GRAN)
2316   ASSERT(!RtsFlags.GranFlags.Light || CurrentProc==0);
2317 #endif
2318
2319   // ToDo: check whether size = stack_size - TSO_STRUCT_SIZEW
2320
2321   /* catch ridiculously small stack sizes */
2322   if (size < MIN_STACK_WORDS + TSO_STRUCT_SIZEW) {
2323     size = MIN_STACK_WORDS + TSO_STRUCT_SIZEW;
2324   }
2325
2326   stack_size = size - TSO_STRUCT_SIZEW;
2327
2328   tso = (StgTSO *)allocate(size);
2329   TICK_ALLOC_TSO(stack_size, 0);
2330
2331   SET_HDR(tso, &stg_TSO_info, CCS_SYSTEM);
2332 #if defined(GRAN)
2333   SET_GRAN_HDR(tso, ThisPE);
2334 #endif
2335
2336   // Always start with the compiled code evaluator
2337   tso->what_next = ThreadRunGHC;
2338
2339   tso->id = next_thread_id++;
2340   tso->why_blocked  = NotBlocked;
2341   tso->blocked_exceptions = NULL;
2342
2343   tso->saved_errno = 0;
2344   tso->main = NULL;
2345
2346   tso->stack_size   = stack_size;
2347   tso->max_stack_size = round_to_mblocks(RtsFlags.GcFlags.maxStkSize)
2348                               - TSO_STRUCT_SIZEW;
2349   tso->sp           = (P_)&(tso->stack) + stack_size;
2350
2351   tso->trec = NO_TREC;
2352
2353 #ifdef PROFILING
2354   tso->prof.CCCS = CCS_MAIN;
2355 #endif
2356
2357   /* put a stop frame on the stack */
2358   tso->sp -= sizeofW(StgStopFrame);
2359   SET_HDR((StgClosure*)tso->sp,(StgInfoTable *)&stg_stop_thread_info,CCS_SYSTEM);
2360   tso->link = END_TSO_QUEUE;
2361
2362   // ToDo: check this
2363 #if defined(GRAN)
2364   /* uses more flexible routine in GranSim */
2365   insertThread(tso, CurrentProc);
2366 #else
2367   /* In a non-GranSim setup the pushing of a TSO onto the runq is separated
2368    * from its creation
2369    */
2370 #endif
2371
2372 #if defined(GRAN)
2373   if (RtsFlags.GranFlags.GranSimStats.Full)
2374     DumpGranEvent(GR_START,tso);
2375 #elif defined(PARALLEL_HASKELL)
2376   if (RtsFlags.ParFlags.ParStats.Full)
2377     DumpGranEvent(GR_STARTQ,tso);
2378   /* HACk to avoid SCHEDULE
2379      LastTSO = tso; */
2380 #endif
2381
2382   /* Link the new thread on the global thread list.
2383    */
2384   tso->global_link = all_threads;
2385   all_threads = tso;
2386
2387 #if defined(DIST)
2388   tso->dist.priority = MandatoryPriority; //by default that is...
2389 #endif
2390
2391 #if defined(GRAN)
2392   tso->gran.pri = pri;
2393 # if defined(DEBUG)
2394   tso->gran.magic = TSO_MAGIC; // debugging only
2395 # endif
2396   tso->gran.sparkname   = 0;
2397   tso->gran.startedat   = CURRENT_TIME;
2398   tso->gran.exported    = 0;
2399   tso->gran.basicblocks = 0;
2400   tso->gran.allocs      = 0;
2401   tso->gran.exectime    = 0;
2402   tso->gran.fetchtime   = 0;
2403   tso->gran.fetchcount  = 0;
2404   tso->gran.blocktime   = 0;
2405   tso->gran.blockcount  = 0;
2406   tso->gran.blockedat   = 0;
2407   tso->gran.globalsparks = 0;
2408   tso->gran.localsparks  = 0;
2409   if (RtsFlags.GranFlags.Light)
2410     tso->gran.clock  = Now; /* local clock */
2411   else
2412     tso->gran.clock  = 0;
2413
2414   IF_DEBUG(gran,printTSO(tso));
2415 #elif defined(PARALLEL_HASKELL)
2416 # if defined(DEBUG)
2417   tso->par.magic = TSO_MAGIC; // debugging only
2418 # endif
2419   tso->par.sparkname   = 0;
2420   tso->par.startedat   = CURRENT_TIME;
2421   tso->par.exported    = 0;
2422   tso->par.basicblocks = 0;
2423   tso->par.allocs      = 0;
2424   tso->par.exectime    = 0;
2425   tso->par.fetchtime   = 0;
2426   tso->par.fetchcount  = 0;
2427   tso->par.blocktime   = 0;
2428   tso->par.blockcount  = 0;
2429   tso->par.blockedat   = 0;
2430   tso->par.globalsparks = 0;
2431   tso->par.localsparks  = 0;
2432 #endif
2433
2434 #if defined(GRAN)
2435   globalGranStats.tot_threads_created++;
2436   globalGranStats.threads_created_on_PE[CurrentProc]++;
2437   globalGranStats.tot_sq_len += spark_queue_len(CurrentProc);
2438   globalGranStats.tot_sq_probes++;
2439 #elif defined(PARALLEL_HASKELL)
2440   // collect parallel global statistics (currently done together with GC stats)
2441   if (RtsFlags.ParFlags.ParStats.Global &&
2442       RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
2443     //debugBelch("Creating thread %d @ %11.2f\n", tso->id, usertime());
2444     globalParStats.tot_threads_created++;
2445   }
2446 #endif
2447
2448 #if defined(GRAN)
2449   IF_GRAN_DEBUG(pri,
2450                 sched_belch("==__ schedule: Created TSO %d (%p);",
2451                       CurrentProc, tso, tso->id));
2452 #elif defined(PARALLEL_HASKELL)
2453   IF_PAR_DEBUG(verbose,
2454                sched_belch("==__ schedule: Created TSO %d (%p); %d threads active",
2455                            (long)tso->id, tso, advisory_thread_count));
2456 #else
2457   IF_DEBUG(scheduler,sched_belch("created thread %ld, stack size = %lx words",
2458                                  (long)tso->id, (long)tso->stack_size));
2459 #endif
2460   return tso;
2461 }
2462
2463 #if defined(PAR)
2464 /* RFP:
2465    all parallel thread creation calls should fall through the following routine.
2466 */
2467 StgTSO *
2468 createThreadFromSpark(rtsSpark spark)
2469 { StgTSO *tso;
2470   ASSERT(spark != (rtsSpark)NULL);
2471 // JB: TAKE CARE OF THIS COUNTER! BUGGY
2472   if (advisory_thread_count >= RtsFlags.ParFlags.maxThreads)
2473   { threadsIgnored++;
2474     barf("{createSparkThread}Daq ghuH: refusing to create another thread; no more than %d threads allowed (currently %d)",
2475           RtsFlags.ParFlags.maxThreads, advisory_thread_count);
2476     return END_TSO_QUEUE;
2477   }
2478   else
2479   { threadsCreated++;
2480     tso = createThread(RtsFlags.GcFlags.initialStkSize);
2481     if (tso==END_TSO_QUEUE)
2482       barf("createSparkThread: Cannot create TSO");
2483 #if defined(DIST)
2484     tso->priority = AdvisoryPriority;
2485 #endif
2486     pushClosure(tso,spark);
2487     addToRunQueue(tso);
2488     advisory_thread_count++;  // JB: TAKE CARE OF THIS COUNTER! BUGGY
2489   }
2490   return tso;
2491 }
2492 #endif
2493
2494 /*
2495   Turn a spark into a thread.
2496   ToDo: fix for SMP (needs to acquire SCHED_MUTEX!)
2497 */
2498 #if 0
2499 StgTSO *
2500 activateSpark (rtsSpark spark)
2501 {
2502   StgTSO *tso;
2503
2504   tso = createSparkThread(spark);
2505   if (RtsFlags.ParFlags.ParStats.Full) {
2506     //ASSERT(run_queue_hd == END_TSO_QUEUE); // I think ...
2507       IF_PAR_DEBUG(verbose,
2508                    debugBelch("==^^ activateSpark: turning spark of closure %p (%s) into a thread\n",
2509                               (StgClosure *)spark, info_type((StgClosure *)spark)));
2510   }
2511   // ToDo: fwd info on local/global spark to thread -- HWL
2512   // tso->gran.exported =  spark->exported;
2513   // tso->gran.locked =   !spark->global;
2514   // tso->gran.sparkname = spark->name;
2515
2516   return tso;
2517 }
2518 #endif
2519
2520 /* ---------------------------------------------------------------------------
2521  * scheduleThread()
2522  *
2523  * scheduleThread puts a thread on the head of the runnable queue.
2524  * This will usually be done immediately after a thread is created.
2525  * The caller of scheduleThread must create the thread using e.g.
2526  * createThread and push an appropriate closure
2527  * on this thread's stack before the scheduler is invoked.
2528  * ------------------------------------------------------------------------ */
2529
2530 static void
2531 scheduleThread_(StgTSO *tso)
2532 {
2533   // The thread goes at the *end* of the run-queue, to avoid possible
2534   // starvation of any threads already on the queue.
2535   APPEND_TO_RUN_QUEUE(tso);
2536   threadRunnable();
2537 }
2538
2539 void
2540 scheduleThread(StgTSO* tso)
2541 {
2542   ACQUIRE_LOCK(&sched_mutex);
2543   scheduleThread_(tso);
2544   RELEASE_LOCK(&sched_mutex);
2545 }
2546
2547 #if defined(RTS_SUPPORTS_THREADS)
2548 static Condition bound_cond_cache;
2549 static int bound_cond_cache_full = 0;
2550 #endif
2551
2552
2553 SchedulerStatus
2554 scheduleWaitThread(StgTSO* tso, /*[out]*/HaskellObj* ret,
2555                    Capability *initialCapability)
2556 {
2557     // Precondition: sched_mutex must be held
2558     StgMainThread *m;
2559
2560     m = stgMallocBytes(sizeof(StgMainThread), "waitThread");
2561     m->tso = tso;
2562     tso->main = m;
2563     m->ret = ret;
2564     m->stat = NoStatus;
2565     m->link = main_threads;
2566     m->prev = NULL;
2567     if (main_threads != NULL) {
2568         main_threads->prev = m;
2569     }
2570     main_threads = m;
2571
2572 #if defined(RTS_SUPPORTS_THREADS)
2573     // Allocating a new condition for each thread is expensive, so we
2574     // cache one.  This is a pretty feeble hack, but it helps speed up
2575     // consecutive call-ins quite a bit.
2576     if (bound_cond_cache_full) {
2577         m->bound_thread_cond = bound_cond_cache;
2578         bound_cond_cache_full = 0;
2579     } else {
2580         initCondition(&m->bound_thread_cond);
2581     }
2582 #endif
2583
2584     /* Put the thread on the main-threads list prior to scheduling the TSO.
2585        Failure to do so introduces a race condition in the MT case (as
2586        identified by Wolfgang Thaller), whereby the new task/OS thread
2587        created by scheduleThread_() would complete prior to the thread
2588        that spawned it managed to put 'itself' on the main-threads list.
2589        The upshot of it all being that the worker thread wouldn't get to
2590        signal the completion of the its work item for the main thread to
2591        see (==> it got stuck waiting.)    -- sof 6/02.
2592     */
2593     IF_DEBUG(scheduler, sched_belch("waiting for thread (%d)", tso->id));
2594
2595     APPEND_TO_RUN_QUEUE(tso);
2596     // NB. Don't call threadRunnable() here, because the thread is
2597     // bound and only runnable by *this* OS thread, so waking up other
2598     // workers will just slow things down.
2599
2600     return waitThread_(m, initialCapability);
2601 }
2602
2603 /* ---------------------------------------------------------------------------
2604  * initScheduler()
2605  *
2606  * Initialise the scheduler.  This resets all the queues - if the
2607  * queues contained any threads, they'll be garbage collected at the
2608  * next pass.
2609  *
2610  * ------------------------------------------------------------------------ */
2611
2612 void
2613 initScheduler(void)
2614 {
2615 #if defined(GRAN)
2616   nat i;
2617
2618   for (i=0; i<=MAX_PROC; i++) {
2619     run_queue_hds[i]      = END_TSO_QUEUE;
2620     run_queue_tls[i]      = END_TSO_QUEUE;
2621     blocked_queue_hds[i]  = END_TSO_QUEUE;
2622     blocked_queue_tls[i]  = END_TSO_QUEUE;
2623     ccalling_threadss[i]  = END_TSO_QUEUE;
2624     blackhole_queue[i]    = END_TSO_QUEUE;
2625     sleeping_queue        = END_TSO_QUEUE;
2626   }
2627 #else
2628   run_queue_hd      = END_TSO_QUEUE;
2629   run_queue_tl      = END_TSO_QUEUE;
2630   blocked_queue_hd  = END_TSO_QUEUE;
2631   blocked_queue_tl  = END_TSO_QUEUE;
2632   blackhole_queue   = END_TSO_QUEUE;
2633   sleeping_queue    = END_TSO_QUEUE;
2634 #endif
2635
2636   suspended_ccalling_threads  = END_TSO_QUEUE;
2637
2638   main_threads = NULL;
2639   all_threads  = END_TSO_QUEUE;
2640
2641   context_switch = 0;
2642   interrupted    = 0;
2643
2644   RtsFlags.ConcFlags.ctxtSwitchTicks =
2645       RtsFlags.ConcFlags.ctxtSwitchTime / TICK_MILLISECS;
2646
2647 #if defined(RTS_SUPPORTS_THREADS)
2648   /* Initialise the mutex and condition variables used by
2649    * the scheduler. */
2650   initMutex(&sched_mutex);
2651   initMutex(&term_mutex);
2652 #endif
2653
2654   ACQUIRE_LOCK(&sched_mutex);
2655
2656   /* A capability holds the state a native thread needs in
2657    * order to execute STG code. At least one capability is
2658    * floating around (only SMP builds have more than one).
2659    */
2660   initCapabilities();
2661
2662 #if defined(RTS_SUPPORTS_THREADS)
2663   initTaskManager();
2664 #endif
2665
2666 #if defined(SMP)
2667   /* eagerly start some extra workers */
2668   startingWorkerThread = RtsFlags.ParFlags.nNodes;
2669   startTasks(RtsFlags.ParFlags.nNodes, taskStart);
2670 #endif
2671
2672 #if /* defined(SMP) ||*/ defined(PARALLEL_HASKELL)
2673   initSparkPools();
2674 #endif
2675
2676   RELEASE_LOCK(&sched_mutex);
2677 }
2678
2679 void
2680 exitScheduler( void )
2681 {
2682     interrupted = rtsTrue;
2683     shutting_down_scheduler = rtsTrue;
2684 #if defined(RTS_SUPPORTS_THREADS)
2685     if (threadIsTask(osThreadId())) { taskStop(); }
2686     stopTaskManager();
2687 #endif
2688 }
2689
2690 /* ----------------------------------------------------------------------------
2691    Managing the per-task allocation areas.
2692
2693    Each capability comes with an allocation area.  These are
2694    fixed-length block lists into which allocation can be done.
2695
2696    ToDo: no support for two-space collection at the moment???
2697    ------------------------------------------------------------------------- */
2698
2699 static SchedulerStatus
2700 waitThread_(StgMainThread* m, Capability *initialCapability)
2701 {
2702   SchedulerStatus stat;
2703
2704   // Precondition: sched_mutex must be held.
2705   IF_DEBUG(scheduler, sched_belch("new main thread (%d)", m->tso->id));
2706
2707 #if defined(GRAN)
2708   /* GranSim specific init */
2709   CurrentTSO = m->tso;                // the TSO to run
2710   procStatus[MainProc] = Busy;        // status of main PE
2711   CurrentProc = MainProc;             // PE to run it on
2712   schedule(m,initialCapability);
2713 #else
2714   schedule(m,initialCapability);
2715   ASSERT(m->stat != NoStatus);
2716 #endif
2717
2718   stat = m->stat;
2719
2720 #if defined(RTS_SUPPORTS_THREADS)
2721   // Free the condition variable, returning it to the cache if possible.
2722   if (!bound_cond_cache_full) {
2723       bound_cond_cache = m->bound_thread_cond;
2724       bound_cond_cache_full = 1;
2725   } else {
2726       closeCondition(&m->bound_thread_cond);
2727   }
2728 #endif
2729
2730   IF_DEBUG(scheduler, sched_belch("main thread (%d) finished", m->tso->id));
2731   stgFree(m);
2732
2733   // Postcondition: sched_mutex still held
2734   return stat;
2735 }
2736
2737 /* ---------------------------------------------------------------------------
2738    Where are the roots that we know about?
2739
2740         - all the threads on the runnable queue
2741         - all the threads on the blocked queue
2742         - all the threads on the sleeping queue
2743         - all the thread currently executing a _ccall_GC
2744         - all the "main threads"
2745
2746    ------------------------------------------------------------------------ */
2747
2748 /* This has to be protected either by the scheduler monitor, or by the
2749         garbage collection monitor (probably the latter).
2750         KH @ 25/10/99
2751 */
2752
2753 void
2754 GetRoots( evac_fn evac )
2755 {
2756 #if defined(GRAN)
2757   {
2758     nat i;
2759     for (i=0; i<=RtsFlags.GranFlags.proc; i++) {
2760       if ((run_queue_hds[i] != END_TSO_QUEUE) && ((run_queue_hds[i] != NULL)))
2761           evac((StgClosure **)&run_queue_hds[i]);
2762       if ((run_queue_tls[i] != END_TSO_QUEUE) && ((run_queue_tls[i] != NULL)))
2763           evac((StgClosure **)&run_queue_tls[i]);
2764
2765       if ((blocked_queue_hds[i] != END_TSO_QUEUE) && ((blocked_queue_hds[i] != NULL)))
2766           evac((StgClosure **)&blocked_queue_hds[i]);
2767       if ((blocked_queue_tls[i] != END_TSO_QUEUE) && ((blocked_queue_tls[i] != NULL)))
2768           evac((StgClosure **)&blocked_queue_tls[i]);
2769       if ((ccalling_threadss[i] != END_TSO_QUEUE) && ((ccalling_threadss[i] != NULL)))
2770           evac((StgClosure **)&ccalling_threads[i]);
2771     }
2772   }
2773
2774   markEventQueue();
2775
2776 #else /* !GRAN */
2777   if (run_queue_hd != END_TSO_QUEUE) {
2778       ASSERT(run_queue_tl != END_TSO_QUEUE);
2779       evac((StgClosure **)&run_queue_hd);
2780       evac((StgClosure **)&run_queue_tl);
2781   }
2782
2783   if (blocked_queue_hd != END_TSO_QUEUE) {
2784       ASSERT(blocked_queue_tl != END_TSO_QUEUE);
2785       evac((StgClosure **)&blocked_queue_hd);
2786       evac((StgClosure **)&blocked_queue_tl);
2787   }
2788
2789   if (sleeping_queue != END_TSO_QUEUE) {
2790       evac((StgClosure **)&sleeping_queue);
2791   }
2792 #endif
2793
2794   if (blackhole_queue != END_TSO_QUEUE) {
2795       evac((StgClosure **)&blackhole_queue);
2796   }
2797
2798   if (suspended_ccalling_threads != END_TSO_QUEUE) {
2799       evac((StgClosure **)&suspended_ccalling_threads);
2800   }
2801
2802 #if defined(PARALLEL_HASKELL) || defined(GRAN)
2803   markSparkQueue(evac);
2804 #endif
2805
2806 #if defined(RTS_USER_SIGNALS)
2807   // mark the signal handlers (signals should be already blocked)
2808   markSignalHandlers(evac);
2809 #endif
2810 }
2811
2812 /* -----------------------------------------------------------------------------
2813    performGC
2814
2815    This is the interface to the garbage collector from Haskell land.
2816    We provide this so that external C code can allocate and garbage
2817    collect when called from Haskell via _ccall_GC.
2818
2819    It might be useful to provide an interface whereby the programmer
2820    can specify more roots (ToDo).
2821
2822    This needs to be protected by the GC condition variable above.  KH.
2823    -------------------------------------------------------------------------- */
2824
2825 static void (*extra_roots)(evac_fn);
2826
2827 void
2828 performGC(void)
2829 {
2830   /* Obligated to hold this lock upon entry */
2831   ACQUIRE_LOCK(&sched_mutex);
2832   GarbageCollect(GetRoots,rtsFalse);
2833   RELEASE_LOCK(&sched_mutex);
2834 }
2835
2836 void
2837 performMajorGC(void)
2838 {
2839   ACQUIRE_LOCK(&sched_mutex);
2840   GarbageCollect(GetRoots,rtsTrue);
2841   RELEASE_LOCK(&sched_mutex);
2842 }
2843
2844 static void
2845 AllRoots(evac_fn evac)
2846 {
2847     GetRoots(evac);             // the scheduler's roots
2848     extra_roots(evac);          // the user's roots
2849 }
2850
2851 void
2852 performGCWithRoots(void (*get_roots)(evac_fn))
2853 {
2854   ACQUIRE_LOCK(&sched_mutex);
2855   extra_roots = get_roots;
2856   GarbageCollect(AllRoots,rtsFalse);
2857   RELEASE_LOCK(&sched_mutex);
2858 }
2859
2860 /* -----------------------------------------------------------------------------
2861    Stack overflow
2862
2863    If the thread has reached its maximum stack size, then raise the
2864    StackOverflow exception in the offending thread.  Otherwise
2865    relocate the TSO into a larger chunk of memory and adjust its stack
2866    size appropriately.
2867    -------------------------------------------------------------------------- */
2868
2869 static StgTSO *
2870 threadStackOverflow(StgTSO *tso)
2871 {
2872   nat new_stack_size, stack_words;
2873   lnat new_tso_size;
2874   StgPtr new_sp;
2875   StgTSO *dest;
2876
2877   IF_DEBUG(sanity,checkTSO(tso));
2878   if (tso->stack_size >= tso->max_stack_size) {
2879
2880     IF_DEBUG(gc,
2881              debugBelch("@@ threadStackOverflow of TSO %ld (%p): stack too large (now %ld; max is %ld)\n",
2882                    (long)tso->id, tso, (long)tso->stack_size, (long)tso->max_stack_size);
2883              /* If we're debugging, just print out the top of the stack */
2884              printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size,
2885                                               tso->sp+64)));
2886
2887     /* Send this thread the StackOverflow exception */
2888     raiseAsync(tso, (StgClosure *)stackOverflow_closure);
2889     return tso;
2890   }
2891
2892   /* Try to double the current stack size.  If that takes us over the
2893    * maximum stack size for this thread, then use the maximum instead.
2894    * Finally round up so the TSO ends up as a whole number of blocks.
2895    */
2896   new_stack_size = stg_min(tso->stack_size * 2, tso->max_stack_size);
2897   new_tso_size   = (lnat)BLOCK_ROUND_UP(new_stack_size * sizeof(W_) +
2898                                        TSO_STRUCT_SIZE)/sizeof(W_);
2899   new_tso_size = round_to_mblocks(new_tso_size);  /* Be MBLOCK-friendly */
2900   new_stack_size = new_tso_size - TSO_STRUCT_SIZEW;
2901
2902   IF_DEBUG(scheduler, debugBelch("== sched: increasing stack size from %d words to %d.\n", tso->stack_size, new_stack_size));
2903
2904   dest = (StgTSO *)allocate(new_tso_size);
2905   TICK_ALLOC_TSO(new_stack_size,0);
2906
2907   /* copy the TSO block and the old stack into the new area */
2908   memcpy(dest,tso,TSO_STRUCT_SIZE);
2909   stack_words = tso->stack + tso->stack_size - tso->sp;
2910   new_sp = (P_)dest + new_tso_size - stack_words;
2911   memcpy(new_sp, tso->sp, stack_words * sizeof(W_));
2912
2913   /* relocate the stack pointers... */
2914   dest->sp         = new_sp;
2915   dest->stack_size = new_stack_size;
2916
2917   /* Mark the old TSO as relocated.  We have to check for relocated
2918    * TSOs in the garbage collector and any primops that deal with TSOs.
2919    *
2920    * It's important to set the sp value to just beyond the end
2921    * of the stack, so we don't attempt to scavenge any part of the
2922    * dead TSO's stack.
2923    */
2924   tso->what_next = ThreadRelocated;
2925   tso->link = dest;
2926   tso->sp = (P_)&(tso->stack[tso->stack_size]);
2927   tso->why_blocked = NotBlocked;
2928
2929   IF_PAR_DEBUG(verbose,
2930                debugBelch("@@ threadStackOverflow of TSO %d (now at %p): stack size increased to %ld\n",
2931                      tso->id, tso, tso->stack_size);
2932                /* If we're debugging, just print out the top of the stack */
2933                printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size,
2934                                                 tso->sp+64)));
2935
2936   IF_DEBUG(sanity,checkTSO(tso));
2937 #if 0
2938   IF_DEBUG(scheduler,printTSO(dest));
2939 #endif
2940
2941   return dest;
2942 }
2943
2944 /* ---------------------------------------------------------------------------
2945    Wake up a queue that was blocked on some resource.
2946    ------------------------------------------------------------------------ */
2947
2948 #if defined(GRAN)
2949 STATIC_INLINE void
2950 unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node )
2951 {
2952 }
2953 #elif defined(PARALLEL_HASKELL)
2954 STATIC_INLINE void
2955 unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node )
2956 {
2957   /* write RESUME events to log file and
2958      update blocked and fetch time (depending on type of the orig closure) */
2959   if (RtsFlags.ParFlags.ParStats.Full) {
2960     DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
2961                      GR_RESUMEQ, ((StgTSO *)bqe), ((StgTSO *)bqe)->block_info.closure,
2962                      0, 0 /* spark_queue_len(ADVISORY_POOL) */);
2963     if (EMPTY_RUN_QUEUE())
2964       emitSchedule = rtsTrue;
2965
2966     switch (get_itbl(node)->type) {
2967         case FETCH_ME_BQ:
2968           ((StgTSO *)bqe)->par.fetchtime += CURRENT_TIME-((StgTSO *)bqe)->par.blockedat;
2969           break;
2970         case RBH:
2971         case FETCH_ME:
2972         case BLACKHOLE_BQ:
2973           ((StgTSO *)bqe)->par.blocktime += CURRENT_TIME-((StgTSO *)bqe)->par.blockedat;
2974           break;
2975 #ifdef DIST
2976         case MVAR:
2977           break;
2978 #endif
2979         default:
2980           barf("{unblockOneLocked}Daq Qagh: unexpected closure in blocking queue");
2981         }
2982       }
2983 }
2984 #endif
2985
2986 #if defined(GRAN)
2987 StgBlockingQueueElement *
2988 unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node)
2989 {
2990     StgTSO *tso;
2991     PEs node_loc, tso_loc;
2992
2993     node_loc = where_is(node); // should be lifted out of loop
2994     tso = (StgTSO *)bqe;  // wastes an assignment to get the type right
2995     tso_loc = where_is((StgClosure *)tso);
2996     if (IS_LOCAL_TO(PROCS(node),tso_loc)) { // TSO is local
2997       /* !fake_fetch => TSO is on CurrentProc is same as IS_LOCAL_TO */
2998       ASSERT(CurrentProc!=node_loc || tso_loc==CurrentProc);
2999       CurrentTime[CurrentProc] += RtsFlags.GranFlags.Costs.lunblocktime;
3000       // insertThread(tso, node_loc);
3001       new_event(tso_loc, tso_loc, CurrentTime[CurrentProc],
3002                 ResumeThread,
3003                 tso, node, (rtsSpark*)NULL);
3004       tso->link = END_TSO_QUEUE; // overwrite link just to be sure
3005       // len_local++;
3006       // len++;
3007     } else { // TSO is remote (actually should be FMBQ)
3008       CurrentTime[CurrentProc] += RtsFlags.GranFlags.Costs.mpacktime +
3009                                   RtsFlags.GranFlags.Costs.gunblocktime +
3010                                   RtsFlags.GranFlags.Costs.latency;
3011       new_event(tso_loc, CurrentProc, CurrentTime[CurrentProc],
3012                 UnblockThread,
3013                 tso, node, (rtsSpark*)NULL);
3014       tso->link = END_TSO_QUEUE; // overwrite link just to be sure
3015       // len++;
3016     }
3017     /* the thread-queue-overhead is accounted for in either Resume or UnblockThread */
3018     IF_GRAN_DEBUG(bq,
3019                   debugBelch(" %s TSO %d (%p) [PE %d] (block_info.closure=%p) (next=%p) ,",
3020                           (node_loc==tso_loc ? "Local" : "Global"),
3021                           tso->id, tso, CurrentProc, tso->block_info.closure, tso->link));
3022     tso->block_info.closure = NULL;
3023     IF_DEBUG(scheduler,debugBelch("-- Waking up thread %ld (%p)\n",
3024                              tso->id, tso));
3025 }
3026 #elif defined(PARALLEL_HASKELL)
3027 StgBlockingQueueElement *
3028 unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node)
3029 {
3030     StgBlockingQueueElement *next;
3031
3032     switch (get_itbl(bqe)->type) {
3033     case TSO:
3034       ASSERT(((StgTSO *)bqe)->why_blocked != NotBlocked);
3035       /* if it's a TSO just push it onto the run_queue */
3036       next = bqe->link;
3037       ((StgTSO *)bqe)->link = END_TSO_QUEUE; // debugging?
3038       APPEND_TO_RUN_QUEUE((StgTSO *)bqe);
3039       threadRunnable();
3040       unblockCount(bqe, node);
3041       /* reset blocking status after dumping event */
3042       ((StgTSO *)bqe)->why_blocked = NotBlocked;
3043       break;
3044
3045     case BLOCKED_FETCH:
3046       /* if it's a BLOCKED_FETCH put it on the PendingFetches list */
3047       next = bqe->link;
3048       bqe->link = (StgBlockingQueueElement *)PendingFetches;
3049       PendingFetches = (StgBlockedFetch *)bqe;
3050       break;
3051
3052 # if defined(DEBUG)
3053       /* can ignore this case in a non-debugging setup;
3054          see comments on RBHSave closures above */
3055     case CONSTR:
3056       /* check that the closure is an RBHSave closure */
3057       ASSERT(get_itbl((StgClosure *)bqe) == &stg_RBH_Save_0_info ||
3058              get_itbl((StgClosure *)bqe) == &stg_RBH_Save_1_info ||
3059              get_itbl((StgClosure *)bqe) == &stg_RBH_Save_2_info);
3060       break;
3061
3062     default:
3063       barf("{unblockOneLocked}Daq Qagh: Unexpected IP (%#lx; %s) in blocking queue at %#lx\n",
3064            get_itbl((StgClosure *)bqe), info_type((StgClosure *)bqe),
3065            (StgClosure *)bqe);
3066 # endif
3067     }
3068   IF_PAR_DEBUG(bq, debugBelch(", %p (%s)\n", bqe, info_type((StgClosure*)bqe)));
3069   return next;
3070 }
3071
3072 #else /* !GRAN && !PARALLEL_HASKELL */
3073 StgTSO *
3074 unblockOneLocked(StgTSO *tso)
3075 {
3076   StgTSO *next;
3077
3078   ASSERT(get_itbl(tso)->type == TSO);
3079   ASSERT(tso->why_blocked != NotBlocked);
3080   tso->why_blocked = NotBlocked;
3081   next = tso->link;
3082   tso->link = END_TSO_QUEUE;
3083   APPEND_TO_RUN_QUEUE(tso);
3084   threadRunnable();
3085   IF_DEBUG(scheduler,sched_belch("waking up thread %ld", (long)tso->id));
3086   return next;
3087 }
3088 #endif
3089
3090 #if defined(GRAN) || defined(PARALLEL_HASKELL)
3091 INLINE_ME StgBlockingQueueElement *
3092 unblockOne(StgBlockingQueueElement *bqe, StgClosure *node)
3093 {
3094   ACQUIRE_LOCK(&sched_mutex);
3095   bqe = unblockOneLocked(bqe, node);
3096   RELEASE_LOCK(&sched_mutex);
3097   return bqe;
3098 }
3099 #else
3100 INLINE_ME StgTSO *
3101 unblockOne(StgTSO *tso)
3102 {
3103   ACQUIRE_LOCK(&sched_mutex);
3104   tso = unblockOneLocked(tso);
3105   RELEASE_LOCK(&sched_mutex);
3106   return tso;
3107 }
3108 #endif
3109
3110 #if defined(GRAN)
3111 void
3112 awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node)
3113 {
3114   StgBlockingQueueElement *bqe;
3115   PEs node_loc;
3116   nat len = 0;
3117
3118   IF_GRAN_DEBUG(bq,
3119                 debugBelch("##-_ AwBQ for node %p on PE %d @ %ld by TSO %d (%p): \n", \
3120                       node, CurrentProc, CurrentTime[CurrentProc],
3121                       CurrentTSO->id, CurrentTSO));
3122
3123   node_loc = where_is(node);
3124
3125   ASSERT(q == END_BQ_QUEUE ||
3126          get_itbl(q)->type == TSO ||   // q is either a TSO or an RBHSave
3127          get_itbl(q)->type == CONSTR); // closure (type constructor)
3128   ASSERT(is_unique(node));
3129
3130   /* FAKE FETCH: magically copy the node to the tso's proc;
3131      no Fetch necessary because in reality the node should not have been
3132      moved to the other PE in the first place
3133   */
3134   if (CurrentProc!=node_loc) {
3135     IF_GRAN_DEBUG(bq,
3136                   debugBelch("## node %p is on PE %d but CurrentProc is %d (TSO %d); assuming fake fetch and adjusting bitmask (old: %#x)\n",
3137                         node, node_loc, CurrentProc, CurrentTSO->id,
3138                         // CurrentTSO, where_is(CurrentTSO),
3139                         node->header.gran.procs));
3140     node->header.gran.procs = (node->header.gran.procs) | PE_NUMBER(CurrentProc);
3141     IF_GRAN_DEBUG(bq,
3142                   debugBelch("## new bitmask of node %p is %#x\n",
3143                         node, node->header.gran.procs));
3144     if (RtsFlags.GranFlags.GranSimStats.Global) {
3145       globalGranStats.tot_fake_fetches++;
3146     }
3147   }
3148
3149   bqe = q;
3150   // ToDo: check: ASSERT(CurrentProc==node_loc);
3151   while (get_itbl(bqe)->type==TSO) { // q != END_TSO_QUEUE) {
3152     //next = bqe->link;
3153     /*
3154        bqe points to the current element in the queue
3155        next points to the next element in the queue
3156     */
3157     //tso = (StgTSO *)bqe;  // wastes an assignment to get the type right
3158     //tso_loc = where_is(tso);
3159     len++;
3160     bqe = unblockOneLocked(bqe, node);
3161   }
3162
3163   /* if this is the BQ of an RBH, we have to put back the info ripped out of
3164      the closure to make room for the anchor of the BQ */
3165   if (bqe!=END_BQ_QUEUE) {
3166     ASSERT(get_itbl(node)->type == RBH && get_itbl(bqe)->type == CONSTR);
3167     /*
3168     ASSERT((info_ptr==&RBH_Save_0_info) ||
3169            (info_ptr==&RBH_Save_1_info) ||
3170            (info_ptr==&RBH_Save_2_info));
3171     */
3172     /* cf. convertToRBH in RBH.c for writing the RBHSave closure */
3173     ((StgRBH *)node)->blocking_queue = (StgBlockingQueueElement *)((StgRBHSave *)bqe)->payload[0];
3174     ((StgRBH *)node)->mut_link       = (StgMutClosure *)((StgRBHSave *)bqe)->payload[1];
3175
3176     IF_GRAN_DEBUG(bq,
3177                   debugBelch("## Filled in RBH_Save for %p (%s) at end of AwBQ\n",
3178                         node, info_type(node)));
3179   }
3180
3181   /* statistics gathering */
3182   if (RtsFlags.GranFlags.GranSimStats.Global) {
3183     // globalGranStats.tot_bq_processing_time += bq_processing_time;
3184     globalGranStats.tot_bq_len += len;      // total length of all bqs awakened
3185     // globalGranStats.tot_bq_len_local += len_local;  // same for local TSOs only
3186     globalGranStats.tot_awbq++;             // total no. of bqs awakened
3187   }
3188   IF_GRAN_DEBUG(bq,
3189                 debugBelch("## BQ Stats of %p: [%d entries] %s\n",
3190                         node, len, (bqe!=END_BQ_QUEUE) ? "RBH" : ""));
3191 }
3192 #elif defined(PARALLEL_HASKELL)
3193 void
3194 awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node)
3195 {
3196   StgBlockingQueueElement *bqe;
3197
3198   ACQUIRE_LOCK(&sched_mutex);
3199
3200   IF_PAR_DEBUG(verbose,
3201                debugBelch("##-_ AwBQ for node %p on [%x]: \n",
3202                      node, mytid));
3203 #ifdef DIST
3204   //RFP
3205   if(get_itbl(q)->type == CONSTR || q==END_BQ_QUEUE) {
3206     IF_PAR_DEBUG(verbose, debugBelch("## ... nothing to unblock so lets just return. RFP (BUG?)\n"));
3207     return;
3208   }
3209 #endif
3210
3211   ASSERT(q == END_BQ_QUEUE ||
3212          get_itbl(q)->type == TSO ||
3213          get_itbl(q)->type == BLOCKED_FETCH ||
3214          get_itbl(q)->type == CONSTR);
3215
3216   bqe = q;
3217   while (get_itbl(bqe)->type==TSO ||
3218          get_itbl(bqe)->type==BLOCKED_FETCH) {
3219     bqe = unblockOneLocked(bqe, node);
3220   }
3221   RELEASE_LOCK(&sched_mutex);
3222 }
3223
3224 #else   /* !GRAN && !PARALLEL_HASKELL */
3225
3226 void
3227 awakenBlockedQueueNoLock(StgTSO *tso)
3228 {
3229   while (tso != END_TSO_QUEUE) {
3230     tso = unblockOneLocked(tso);
3231   }
3232 }
3233
3234 void
3235 awakenBlockedQueue(StgTSO *tso)
3236 {
3237   ACQUIRE_LOCK(&sched_mutex);
3238   while (tso != END_TSO_QUEUE) {
3239     tso = unblockOneLocked(tso);
3240   }
3241   RELEASE_LOCK(&sched_mutex);
3242 }
3243 #endif
3244
3245 /* ---------------------------------------------------------------------------
3246    Interrupt execution
3247    - usually called inside a signal handler so it mustn't do anything fancy.
3248    ------------------------------------------------------------------------ */
3249
3250 void
3251 interruptStgRts(void)
3252 {
3253     interrupted    = 1;
3254     context_switch = 1;
3255     threadRunnable();
3256     /* ToDo: if invoked from a signal handler, this threadRunnable
3257      * only works if there's another thread (not this one) waiting to
3258      * be woken up.
3259      */
3260 }
3261
3262 /* -----------------------------------------------------------------------------
3263    Unblock a thread
3264
3265    This is for use when we raise an exception in another thread, which
3266    may be blocked.
3267    This has nothing to do with the UnblockThread event in GranSim. -- HWL
3268    -------------------------------------------------------------------------- */
3269
3270 #if defined(GRAN) || defined(PARALLEL_HASKELL)
3271 /*
3272   NB: only the type of the blocking queue is different in GranSim and GUM
3273       the operations on the queue-elements are the same
3274       long live polymorphism!
3275
3276   Locks: sched_mutex is held upon entry and exit.
3277
3278 */
3279 static void
3280 unblockThread(StgTSO *tso)
3281 {
3282   StgBlockingQueueElement *t, **last;
3283
3284   switch (tso->why_blocked) {
3285
3286   case NotBlocked:
3287     return;  /* not blocked */
3288
3289   case BlockedOnSTM:
3290     // Be careful: nothing to do here!  We tell the scheduler that the thread
3291     // is runnable and we leave it to the stack-walking code to abort the
3292     // transaction while unwinding the stack.  We should perhaps have a debugging
3293     // test to make sure that this really happens and that the 'zombie' transaction
3294     // does not get committed.
3295     goto done;
3296
3297   case BlockedOnMVar:
3298     ASSERT(get_itbl(tso->block_info.closure)->type == MVAR);
3299     {
3300       StgBlockingQueueElement *last_tso = END_BQ_QUEUE;
3301       StgMVar *mvar = (StgMVar *)(tso->block_info.closure);
3302
3303       last = (StgBlockingQueueElement **)&mvar->head;
3304       for (t = (StgBlockingQueueElement *)mvar->head;
3305            t != END_BQ_QUEUE;
3306            last = &t->link, last_tso = t, t = t->link) {
3307         if (t == (StgBlockingQueueElement *)tso) {
3308           *last = (StgBlockingQueueElement *)tso->link;
3309           if (mvar->tail == tso) {
3310             mvar->tail = (StgTSO *)last_tso;
3311           }
3312           goto done;
3313         }
3314       }
3315       barf("unblockThread (MVAR): TSO not found");
3316     }
3317
3318   case BlockedOnBlackHole:
3319     ASSERT(get_itbl(tso->block_info.closure)->type == BLACKHOLE_BQ);
3320     {
3321       StgBlockingQueue *bq = (StgBlockingQueue *)(tso->block_info.closure);
3322
3323       last = &bq->blocking_queue;
3324       for (t = bq->blocking_queue;
3325            t != END_BQ_QUEUE;
3326            last = &t->link, t = t->link) {
3327         if (t == (StgBlockingQueueElement *)tso) {
3328           *last = (StgBlockingQueueElement *)tso->link;
3329           goto done;
3330         }
3331       }
3332       barf("unblockThread (BLACKHOLE): TSO not found");
3333     }
3334
3335   case BlockedOnException:
3336     {
3337       StgTSO *target  = tso->block_info.tso;
3338
3339       ASSERT(get_itbl(target)->type == TSO);
3340
3341       if (target->what_next == ThreadRelocated) {
3342           target = target->link;
3343           ASSERT(get_itbl(target)->type == TSO);
3344       }
3345
3346       ASSERT(target->blocked_exceptions != NULL);
3347
3348       last = (StgBlockingQueueElement **)&target->blocked_exceptions;
3349       for (t = (StgBlockingQueueElement *)target->blocked_exceptions;
3350            t != END_BQ_QUEUE;
3351            last = &t->link, t = t->link) {
3352         ASSERT(get_itbl(t)->type == TSO);
3353         if (t == (StgBlockingQueueElement *)tso) {
3354           *last = (StgBlockingQueueElement *)tso->link;
3355           goto done;
3356         }
3357       }
3358       barf("unblockThread (Exception): TSO not found");
3359     }
3360
3361   case BlockedOnRead:
3362   case BlockedOnWrite:
3363 #if defined(mingw32_HOST_OS)
3364   case BlockedOnDoProc:
3365 #endif
3366     {
3367       /* take TSO off blocked_queue */
3368       StgBlockingQueueElement *prev = NULL;
3369       for (t = (StgBlockingQueueElement *)blocked_queue_hd; t != END_BQ_QUEUE;
3370            prev = t, t = t->link) {
3371         if (t == (StgBlockingQueueElement *)tso) {
3372           if (prev == NULL) {
3373             blocked_queue_hd = (StgTSO *)t->link;
3374             if ((StgBlockingQueueElement *)blocked_queue_tl == t) {
3375               blocked_queue_tl = END_TSO_QUEUE;
3376             }
3377           } else {
3378             prev->link = t->link;
3379             if ((StgBlockingQueueElement *)blocked_queue_tl == t) {
3380               blocked_queue_tl = (StgTSO *)prev;
3381             }
3382           }
3383 #if defined(mingw32_HOST_OS)
3384           /* (Cooperatively) signal that the worker thread should abort
3385            * the request.
3386            */
3387           abandonWorkRequest(tso->block_info.async_result->reqID);
3388 #endif
3389           goto done;
3390         }
3391       }
3392       barf("unblockThread (I/O): TSO not found");
3393     }
3394
3395   case BlockedOnDelay:
3396     {
3397       /* take TSO off sleeping_queue */
3398       StgBlockingQueueElement *prev = NULL;
3399       for (t = (StgBlockingQueueElement *)sleeping_queue; t != END_BQ_QUEUE;
3400            prev = t, t = t->link) {
3401         if (t == (StgBlockingQueueElement *)tso) {
3402           if (prev == NULL) {
3403             sleeping_queue = (StgTSO *)t->link;
3404           } else {
3405             prev->link = t->link;
3406           }
3407           goto done;
3408         }
3409       }
3410       barf("unblockThread (delay): TSO not found");
3411     }
3412
3413   default:
3414     barf("unblockThread");
3415   }
3416
3417  done:
3418   tso->link = END_TSO_QUEUE;
3419   tso->why_blocked = NotBlocked;
3420   tso->block_info.closure = NULL;
3421   PUSH_ON_RUN_QUEUE(tso);
3422 }
3423 #else
3424 static void
3425 unblockThread(StgTSO *tso)
3426 {
3427   StgTSO *t, **last;
3428
3429   /* To avoid locking unnecessarily. */
3430   if (tso->why_blocked == NotBlocked) {
3431     return;
3432   }
3433
3434   switch (tso->why_blocked) {
3435
3436   case BlockedOnSTM:
3437     // Be careful: nothing to do here!  We tell the scheduler that the thread
3438     // is runnable and we leave it to the stack-walking code to abort the
3439     // transaction while unwinding the stack.  We should perhaps have a debugging
3440     // test to make sure that this really happens and that the 'zombie' transaction
3441     // does not get committed.
3442     goto done;
3443
3444   case BlockedOnMVar:
3445     ASSERT(get_itbl(tso->block_info.closure)->type == MVAR);
3446     {
3447       StgTSO *last_tso = END_TSO_QUEUE;
3448       StgMVar *mvar = (StgMVar *)(tso->block_info.closure);
3449
3450       last = &mvar->head;
3451       for (t = mvar->head; t != END_TSO_QUEUE;
3452            last = &t->link, last_tso = t, t = t->link) {
3453         if (t == tso) {
3454           *last = tso->link;
3455           if (mvar->tail == tso) {
3456             mvar->tail = last_tso;
3457           }
3458           goto done;
3459         }
3460       }
3461       barf("unblockThread (MVAR): TSO not found");
3462     }
3463
3464   case BlockedOnBlackHole:
3465     {
3466       last = &blackhole_queue;
3467       for (t = blackhole_queue; t != END_TSO_QUEUE;
3468            last = &t->link, t = t->link) {
3469         if (t == tso) {
3470           *last = tso->link;
3471           goto done;
3472         }
3473       }
3474       barf("unblockThread (BLACKHOLE): TSO not found");
3475     }
3476
3477   case BlockedOnException:
3478     {
3479       StgTSO *target  = tso->block_info.tso;
3480
3481       ASSERT(get_itbl(target)->type == TSO);
3482
3483       while (target->what_next == ThreadRelocated) {
3484           target = target->link;
3485           ASSERT(get_itbl(target)->type == TSO);
3486       }
3487
3488       ASSERT(target->blocked_exceptions != NULL);
3489
3490       last = &target->blocked_exceptions;
3491       for (t = target->blocked_exceptions; t != END_TSO_QUEUE;
3492            last = &t->link, t = t->link) {
3493         ASSERT(get_itbl(t)->type == TSO);
3494         if (t == tso) {
3495           *last = tso->link;
3496           goto done;
3497         }
3498       }
3499       barf("unblockThread (Exception): TSO not found");
3500     }
3501
3502   case BlockedOnRead:
3503   case BlockedOnWrite:
3504 #if defined(mingw32_HOST_OS)
3505   case BlockedOnDoProc:
3506 #endif
3507     {
3508       StgTSO *prev = NULL;
3509       for (t = blocked_queue_hd; t != END_TSO_QUEUE;
3510            prev = t, t = t->link) {
3511         if (t == tso) {
3512           if (prev == NULL) {
3513             blocked_queue_hd = t->link;
3514             if (blocked_queue_tl == t) {
3515               blocked_queue_tl = END_TSO_QUEUE;
3516             }
3517           } else {
3518             prev->link = t->link;
3519             if (blocked_queue_tl == t) {
3520               blocked_queue_tl = prev;
3521             }
3522           }
3523 #if defined(mingw32_HOST_OS)
3524           /* (Cooperatively) signal that the worker thread should abort
3525            * the request.
3526            */
3527           abandonWorkRequest(tso->block_info.async_result->reqID);
3528 #endif
3529           goto done;
3530         }
3531       }
3532       barf("unblockThread (I/O): TSO not found");
3533     }
3534
3535   case BlockedOnDelay:
3536     {
3537       StgTSO *prev = NULL;
3538       for (t = sleeping_queue; t != END_TSO_QUEUE;
3539            prev = t, t = t->link) {
3540         if (t == tso) {
3541           if (prev == NULL) {
3542             sleeping_queue = t->link;
3543           } else {
3544             prev->link = t->link;
3545           }
3546           goto done;
3547         }
3548       }
3549       barf("unblockThread (delay): TSO not found");
3550     }
3551
3552   default:
3553     barf("unblockThread");
3554   }
3555
3556  done:
3557   tso->link = END_TSO_QUEUE;
3558   tso->why_blocked = NotBlocked;
3559   tso->block_info.closure = NULL;
3560   APPEND_TO_RUN_QUEUE(tso);
3561 }
3562 #endif
3563
3564 /* -----------------------------------------------------------------------------
3565  * checkBlackHoles()
3566  *
3567  * Check the blackhole_queue for threads that can be woken up.  We do
3568  * this periodically: before every GC, and whenever the run queue is
3569  * empty.
3570  *
3571  * An elegant solution might be to just wake up all the blocked
3572  * threads with awakenBlockedQueue occasionally: they'll go back to
3573  * sleep again if the object is still a BLACKHOLE.  Unfortunately this
3574  * doesn't give us a way to tell whether we've actually managed to
3575  * wake up any threads, so we would be busy-waiting.
3576  *
3577  * -------------------------------------------------------------------------- */
3578
3579 static rtsBool
3580 checkBlackHoles( void )
3581 {
3582     StgTSO **prev, *t;
3583     rtsBool any_woke_up = rtsFalse;
3584     StgHalfWord type;
3585
3586     IF_DEBUG(scheduler, sched_belch("checking threads blocked on black holes"));
3587
3588     // ASSUMES: sched_mutex
3589     prev = &blackhole_queue;
3590     t = blackhole_queue;
3591     while (t != END_TSO_QUEUE) {
3592         ASSERT(t->why_blocked == BlockedOnBlackHole);
3593         type = get_itbl(t->block_info.closure)->type;
3594         if (type != BLACKHOLE && type != CAF_BLACKHOLE) {
3595             t = unblockOneLocked(t);
3596             *prev = t;
3597             any_woke_up = rtsTrue;
3598         } else {
3599             prev = &t->link;
3600             t = t->link;
3601         }
3602     }
3603
3604     return any_woke_up;
3605 }
3606
3607 /* -----------------------------------------------------------------------------
3608  * raiseAsync()
3609  *
3610  * The following function implements the magic for raising an
3611  * asynchronous exception in an existing thread.
3612  *
3613  * We first remove the thread from any queue on which it might be
3614  * blocked.  The possible blockages are MVARs and BLACKHOLE_BQs.
3615  *
3616  * We strip the stack down to the innermost CATCH_FRAME, building
3617  * thunks in the heap for all the active computations, so they can
3618  * be restarted if necessary.  When we reach a CATCH_FRAME, we build
3619  * an application of the handler to the exception, and push it on
3620  * the top of the stack.
3621  *
3622  * How exactly do we save all the active computations?  We create an
3623  * AP_STACK for every UpdateFrame on the stack.  Entering one of these
3624  * AP_STACKs pushes everything from the corresponding update frame
3625  * upwards onto the stack.  (Actually, it pushes everything up to the
3626  * next update frame plus a pointer to the next AP_STACK object.
3627  * Entering the next AP_STACK object pushes more onto the stack until we
3628  * reach the last AP_STACK object - at which point the stack should look
3629  * exactly as it did when we killed the TSO and we can continue
3630  * execution by entering the closure on top of the stack.
3631  *
3632  * We can also kill a thread entirely - this happens if either (a) the
3633  * exception passed to raiseAsync is NULL, or (b) there's no
3634  * CATCH_FRAME on the stack.  In either case, we strip the entire
3635  * stack and replace the thread with a zombie.
3636  *
3637  * Locks: sched_mutex held upon entry nor exit.
3638  *
3639  * -------------------------------------------------------------------------- */
3640
3641 void
3642 deleteThread(StgTSO *tso)
3643 {
3644   if (tso->why_blocked != BlockedOnCCall &&
3645       tso->why_blocked != BlockedOnCCall_NoUnblockExc) {
3646       raiseAsync(tso,NULL);
3647   }
3648 }
3649
3650 #ifdef FORKPROCESS_PRIMOP_SUPPORTED
3651 static void
3652 deleteThreadImmediately(StgTSO *tso)
3653 { // for forkProcess only:
3654   // delete thread without giving it a chance to catch the KillThread exception
3655
3656   if (tso->what_next == ThreadComplete || tso->what_next == ThreadKilled) {
3657       return;
3658   }
3659
3660   if (tso->why_blocked != BlockedOnCCall &&
3661       tso->why_blocked != BlockedOnCCall_NoUnblockExc) {
3662     unblockThread(tso);
3663   }
3664
3665   tso->what_next = ThreadKilled;
3666 }
3667 #endif
3668
3669 void
3670 raiseAsyncWithLock(StgTSO *tso, StgClosure *exception)
3671 {
3672   /* When raising async exs from contexts where sched_mutex isn't held;
3673      use raiseAsyncWithLock(). */
3674   ACQUIRE_LOCK(&sched_mutex);
3675   raiseAsync(tso,exception);
3676   RELEASE_LOCK(&sched_mutex);
3677 }
3678
3679 void
3680 raiseAsync(StgTSO *tso, StgClosure *exception)
3681 {
3682     raiseAsync_(tso, exception, rtsFalse);
3683 }
3684
3685 static void
3686 raiseAsync_(StgTSO *tso, StgClosure *exception, rtsBool stop_at_atomically)
3687 {
3688     StgRetInfoTable *info;
3689     StgPtr sp;
3690
3691     // Thread already dead?
3692     if (tso->what_next == ThreadComplete || tso->what_next == ThreadKilled) {
3693         return;
3694     }
3695
3696     IF_DEBUG(scheduler,
3697              sched_belch("raising exception in thread %ld.", (long)tso->id));
3698
3699     // Remove it from any blocking queues
3700     unblockThread(tso);
3701
3702     sp = tso->sp;
3703
3704     // The stack freezing code assumes there's a closure pointer on
3705     // the top of the stack, so we have to arrange that this is the case...
3706     //
3707     if (sp[0] == (W_)&stg_enter_info) {
3708         sp++;
3709     } else {
3710         sp--;
3711         sp[0] = (W_)&stg_dummy_ret_closure;
3712     }
3713
3714     while (1) {
3715         nat i;
3716
3717         // 1. Let the top of the stack be the "current closure"
3718         //
3719         // 2. Walk up the stack until we find either an UPDATE_FRAME or a
3720         // CATCH_FRAME.
3721         //
3722         // 3. If it's an UPDATE_FRAME, then make an AP_STACK containing the
3723         // current closure applied to the chunk of stack up to (but not
3724         // including) the update frame.  This closure becomes the "current
3725         // closure".  Go back to step 2.
3726         //
3727         // 4. If it's a CATCH_FRAME, then leave the exception handler on
3728         // top of the stack applied to the exception.
3729         //
3730         // 5. If it's a STOP_FRAME, then kill the thread.
3731         //
3732         // NB: if we pass an ATOMICALLY_FRAME then abort the associated
3733         // transaction
3734
3735
3736         StgPtr frame;
3737
3738         frame = sp + 1;
3739         info = get_ret_itbl((StgClosure *)frame);
3740
3741         while (info->i.type != UPDATE_FRAME
3742                && (info->i.type != CATCH_FRAME || exception == NULL)
3743                && info->i.type != STOP_FRAME
3744                && (info->i.type != ATOMICALLY_FRAME || stop_at_atomically == rtsFalse))
3745         {
3746             if (info->i.type == CATCH_RETRY_FRAME || info->i.type == ATOMICALLY_FRAME) {
3747               // IF we find an ATOMICALLY_FRAME then we abort the
3748               // current transaction and propagate the exception.  In
3749               // this case (unlike ordinary exceptions) we do not care
3750               // whether the transaction is valid or not because its
3751               // possible validity cannot have caused the exception
3752               // and will not be visible after the abort.
3753               IF_DEBUG(stm,
3754                        debugBelch("Found atomically block delivering async exception\n"));
3755               stmAbortTransaction(tso -> trec);
3756               tso -> trec = stmGetEnclosingTRec(tso -> trec);
3757             }
3758             frame += stack_frame_sizeW((StgClosure *)frame);
3759             info = get_ret_itbl((StgClosure *)frame);
3760         }
3761
3762         switch (info->i.type) {
3763
3764         case ATOMICALLY_FRAME:
3765             ASSERT(stop_at_atomically);
3766             ASSERT(stmGetEnclosingTRec(tso->trec) == NO_TREC);
3767             stmCondemnTransaction(tso -> trec);
3768 #ifdef REG_R1
3769             tso->sp = frame;
3770 #else
3771             // R1 is not a register: the return convention for IO in
3772             // this case puts the return value on the stack, so we
3773             // need to set up the stack to return to the atomically
3774             // frame properly...
3775             tso->sp = frame - 2;
3776             tso->sp[1] = (StgWord) &stg_NO_FINALIZER_closure; // why not?
3777             tso->sp[0] = (StgWord) &stg_ut_1_0_unreg_info;
3778 #endif
3779             tso->what_next = ThreadRunGHC;
3780             return;
3781
3782         case CATCH_FRAME:
3783             // If we find a CATCH_FRAME, and we've got an exception to raise,
3784             // then build the THUNK raise(exception), and leave it on
3785             // top of the CATCH_FRAME ready to enter.
3786             //
3787         {
3788 #ifdef PROFILING
3789             StgCatchFrame *cf = (StgCatchFrame *)frame;
3790 #endif
3791             StgThunk *raise;
3792
3793             // we've got an exception to raise, so let's pass it to the
3794             // handler in this frame.
3795             //
3796             raise = (StgThunk *)allocate(sizeofW(StgThunk)+1);
3797             TICK_ALLOC_SE_THK(1,0);
3798             SET_HDR(raise,&stg_raise_info,cf->header.prof.ccs);
3799             raise->payload[0] = exception;
3800
3801             // throw away the stack from Sp up to the CATCH_FRAME.
3802             //
3803             sp = frame - 1;
3804
3805             /* Ensure that async excpetions are blocked now, so we don't get
3806              * a surprise exception before we get around to executing the
3807              * handler.
3808              */
3809             if (tso->blocked_exceptions == NULL) {
3810                 tso->blocked_exceptions = END_TSO_QUEUE;
3811             }
3812
3813             /* Put the newly-built THUNK on top of the stack, ready to execute
3814              * when the thread restarts.
3815              */
3816             sp[0] = (W_)raise;
3817             sp[-1] = (W_)&stg_enter_info;
3818             tso->sp = sp-1;
3819             tso->what_next = ThreadRunGHC;
3820             IF_DEBUG(sanity, checkTSO(tso));
3821             return;
3822         }
3823
3824         case UPDATE_FRAME:
3825         {
3826             StgAP_STACK * ap;
3827             nat words;
3828
3829             // First build an AP_STACK consisting of the stack chunk above the
3830             // current update frame, with the top word on the stack as the
3831             // fun field.
3832             //
3833             words = frame - sp - 1;
3834             ap = (StgAP_STACK *)allocate(AP_STACK_sizeW(words));
3835
3836             ap->size = words;
3837             ap->fun  = (StgClosure *)sp[0];
3838             sp++;
3839             for(i=0; i < (nat)words; ++i) {
3840                 ap->payload[i] = (StgClosure *)*sp++;
3841             }
3842
3843             SET_HDR(ap,&stg_AP_STACK_info,
3844                     ((StgClosure *)frame)->header.prof.ccs /* ToDo */);
3845             TICK_ALLOC_UP_THK(words+1,0);
3846
3847             IF_DEBUG(scheduler,
3848                      debugBelch("sched: Updating ");
3849                      printPtr((P_)((StgUpdateFrame *)frame)->updatee);
3850                      debugBelch(" with ");
3851                      printObj((StgClosure *)ap);
3852                 );
3853
3854             // Replace the updatee with an indirection - happily
3855             // this will also wake up any threads currently
3856             // waiting on the result.
3857             //
3858             // Warning: if we're in a loop, more than one update frame on
3859             // the stack may point to the same object.  Be careful not to
3860             // overwrite an IND_OLDGEN in this case, because we'll screw
3861             // up the mutable lists.  To be on the safe side, don't
3862             // overwrite any kind of indirection at all.  See also
3863             // threadSqueezeStack in GC.c, where we have to make a similar
3864             // check.
3865             //
3866             if (!closure_IND(((StgUpdateFrame *)frame)->updatee)) {
3867                 // revert the black hole
3868                 UPD_IND_NOLOCK(((StgUpdateFrame *)frame)->updatee,
3869                                (StgClosure *)ap);
3870             }
3871             sp += sizeofW(StgUpdateFrame) - 1;
3872             sp[0] = (W_)ap; // push onto stack
3873             break;
3874         }
3875
3876         case STOP_FRAME:
3877             // We've stripped the entire stack, the thread is now dead.
3878             sp += sizeofW(StgStopFrame);
3879             tso->what_next = ThreadKilled;
3880             tso->sp = sp;
3881             return;
3882
3883         default:
3884             barf("raiseAsync");
3885         }
3886     }
3887     barf("raiseAsync");
3888 }
3889
3890 /* -----------------------------------------------------------------------------
3891    raiseExceptionHelper
3892
3893    This function is called by the raise# primitve, just so that we can
3894    move some of the tricky bits of raising an exception from C-- into
3895    C.  Who knows, it might be a useful re-useable thing here too.
3896    -------------------------------------------------------------------------- */
3897
3898 StgWord
3899 raiseExceptionHelper (StgTSO *tso, StgClosure *exception)
3900 {
3901     StgThunk *raise_closure = NULL;
3902     StgPtr p, next;
3903     StgRetInfoTable *info;
3904     //
3905     // This closure represents the expression 'raise# E' where E
3906     // is the exception raise.  It is used to overwrite all the
3907     // thunks which are currently under evaluataion.
3908     //
3909
3910     //
3911     // LDV profiling: stg_raise_info has THUNK as its closure
3912     // type. Since a THUNK takes at least MIN_UPD_SIZE words in its
3913     // payload, MIN_UPD_SIZE is more approprate than 1.  It seems that
3914     // 1 does not cause any problem unless profiling is performed.
3915     // However, when LDV profiling goes on, we need to linearly scan
3916     // small object pool, where raise_closure is stored, so we should
3917     // use MIN_UPD_SIZE.
3918     //
3919     // raise_closure = (StgClosure *)RET_STGCALL1(P_,allocate,
3920     //                                 sizeofW(StgClosure)+1);
3921     //
3922
3923     //
3924     // Walk up the stack, looking for the catch frame.  On the way,
3925     // we update any closures pointed to from update frames with the
3926     // raise closure that we just built.
3927     //
3928     p = tso->sp;
3929     while(1) {
3930         info = get_ret_itbl((StgClosure *)p);
3931         next = p + stack_frame_sizeW((StgClosure *)p);
3932         switch (info->i.type) {
3933
3934         case UPDATE_FRAME:
3935             // Only create raise_closure if we need to.
3936             if (raise_closure == NULL) {
3937                 raise_closure =
3938                     (StgThunk *)allocate(sizeofW(StgThunk)+MIN_UPD_SIZE);
3939                 SET_HDR(raise_closure, &stg_raise_info, CCCS);
3940                 raise_closure->payload[0] = exception;
3941             }
3942             UPD_IND(((StgUpdateFrame *)p)->updatee,(StgClosure *)raise_closure);
3943             p = next;
3944             continue;
3945
3946         case ATOMICALLY_FRAME:
3947             IF_DEBUG(stm, debugBelch("Found ATOMICALLY_FRAME at %p\n", p));
3948             tso->sp = p;
3949             return ATOMICALLY_FRAME;
3950
3951         case CATCH_FRAME:
3952             tso->sp = p;
3953             return CATCH_FRAME;
3954
3955         case CATCH_STM_FRAME:
3956             IF_DEBUG(stm, debugBelch("Found CATCH_STM_FRAME at %p\n", p));
3957             tso->sp = p;
3958             return CATCH_STM_FRAME;
3959
3960         case STOP_FRAME:
3961             tso->sp = p;
3962             return STOP_FRAME;
3963
3964         case CATCH_RETRY_FRAME:
3965         default:
3966             p = next;
3967             continue;
3968         }
3969     }
3970 }
3971
3972
3973 /* -----------------------------------------------------------------------------
3974    findRetryFrameHelper
3975
3976    This function is called by the retry# primitive.  It traverses the stack
3977    leaving tso->sp referring to the frame which should handle the retry.
3978
3979    This should either be a CATCH_RETRY_FRAME (if the retry# is within an orElse#)
3980    or should be a ATOMICALLY_FRAME (if the retry# reaches the top level).
3981
3982    We skip CATCH_STM_FRAMEs because retries are not considered to be exceptions,
3983    despite the similar implementation.
3984
3985    We should not expect to see CATCH_FRAME or STOP_FRAME because those should
3986    not be created within memory transactions.
3987    -------------------------------------------------------------------------- */
3988
3989 StgWord
3990 findRetryFrameHelper (StgTSO *tso)
3991 {
3992   StgPtr           p, next;
3993   StgRetInfoTable *info;
3994
3995   p = tso -> sp;
3996   while (1) {
3997     info = get_ret_itbl((StgClosure *)p);
3998     next = p + stack_frame_sizeW((StgClosure *)p);
3999     switch (info->i.type) {
4000
4001     case ATOMICALLY_FRAME:
4002       IF_DEBUG(stm, debugBelch("Found ATOMICALLY_FRAME at %p during retrry\n", p));
4003       tso->sp = p;
4004       return ATOMICALLY_FRAME;
4005
4006     case CATCH_RETRY_FRAME:
4007       IF_DEBUG(stm, debugBelch("Found CATCH_RETRY_FRAME at %p during retrry\n", p));
4008       tso->sp = p;
4009       return CATCH_RETRY_FRAME;
4010
4011     case CATCH_STM_FRAME:
4012     default:
4013       ASSERT(info->i.type != CATCH_FRAME);
4014       ASSERT(info->i.type != STOP_FRAME);
4015       p = next;
4016       continue;
4017     }
4018   }
4019 }
4020
4021 /* -----------------------------------------------------------------------------
4022    resurrectThreads is called after garbage collection on the list of
4023    threads found to be garbage.  Each of these threads will be woken
4024    up and sent a signal: BlockedOnDeadMVar if the thread was blocked
4025    on an MVar, or NonTermination if the thread was blocked on a Black
4026    Hole.
4027
4028    Locks: sched_mutex isn't held upon entry nor exit.
4029    -------------------------------------------------------------------------- */
4030
4031 void
4032 resurrectThreads( StgTSO *threads )
4033 {
4034   StgTSO *tso, *next;
4035
4036   for (tso = threads; tso != END_TSO_QUEUE; tso = next) {
4037     next = tso->global_link;
4038     tso->global_link = all_threads;
4039     all_threads = tso;
4040     IF_DEBUG(scheduler, sched_belch("resurrecting thread %d", tso->id));
4041
4042     switch (tso->why_blocked) {
4043     case BlockedOnMVar:
4044     case BlockedOnException:
4045       /* Called by GC - sched_mutex lock is currently held. */
4046       raiseAsync(tso,(StgClosure *)BlockedOnDeadMVar_closure);
4047       break;
4048     case BlockedOnBlackHole:
4049       raiseAsync(tso,(StgClosure *)NonTermination_closure);
4050       break;
4051     case BlockedOnSTM:
4052       raiseAsync(tso,(StgClosure *)BlockedIndefinitely_closure);
4053       break;
4054     case NotBlocked:
4055       /* This might happen if the thread was blocked on a black hole
4056        * belonging to a thread that we've just woken up (raiseAsync
4057        * can wake up threads, remember...).
4058        */
4059       continue;
4060     default:
4061       barf("resurrectThreads: thread blocked in a strange way");
4062     }
4063   }
4064 }
4065
4066 /* ----------------------------------------------------------------------------
4067  * Debugging: why is a thread blocked
4068  * [Also provides useful information when debugging threaded programs
4069  *  at the Haskell source code level, so enable outside of DEBUG. --sof 7/02]
4070    ------------------------------------------------------------------------- */
4071
4072 static void
4073 printThreadBlockage(StgTSO *tso)
4074 {
4075   switch (tso->why_blocked) {
4076   case BlockedOnRead:
4077     debugBelch("is blocked on read from fd %d", (int)(tso->block_info.fd));
4078     break;
4079   case BlockedOnWrite:
4080     debugBelch("is blocked on write to fd %d", (int)(tso->block_info.fd));
4081     break;
4082 #if defined(mingw32_HOST_OS)
4083     case BlockedOnDoProc:
4084     debugBelch("is blocked on proc (request: %ld)", tso->block_info.async_result->reqID);
4085     break;
4086 #endif
4087   case BlockedOnDelay:
4088     debugBelch("is blocked until %ld", (long)(tso->block_info.target));
4089     break;
4090   case BlockedOnMVar:
4091     debugBelch("is blocked on an MVar");
4092     break;
4093   case BlockedOnException:
4094     debugBelch("is blocked on delivering an exception to thread %d",
4095             tso->block_info.tso->id);
4096     break;
4097   case BlockedOnBlackHole:
4098     debugBelch("is blocked on a black hole");
4099     break;
4100   case NotBlocked:
4101     debugBelch("is not blocked");
4102     break;
4103 #if defined(PARALLEL_HASKELL)
4104   case BlockedOnGA:
4105     debugBelch("is blocked on global address; local FM_BQ is %p (%s)",
4106             tso->block_info.closure, info_type(tso->block_info.closure));
4107     break;
4108   case BlockedOnGA_NoSend:
4109     debugBelch("is blocked on global address (no send); local FM_BQ is %p (%s)",
4110             tso->block_info.closure, info_type(tso->block_info.closure));
4111     break;
4112 #endif
4113   case BlockedOnCCall:
4114     debugBelch("is blocked on an external call");
4115     break;
4116   case BlockedOnCCall_NoUnblockExc:
4117     debugBelch("is blocked on an external call (exceptions were already blocked)");
4118     break;
4119   case BlockedOnSTM:
4120     debugBelch("is blocked on an STM operation");
4121     break;
4122   default:
4123     barf("printThreadBlockage: strange tso->why_blocked: %d for TSO %d (%d)",
4124          tso->why_blocked, tso->id, tso);
4125   }
4126 }
4127
4128 static void
4129 printThreadStatus(StgTSO *tso)
4130 {
4131   switch (tso->what_next) {
4132   case ThreadKilled:
4133     debugBelch("has been killed");
4134     break;
4135   case ThreadComplete:
4136     debugBelch("has completed");
4137     break;
4138   default:
4139     printThreadBlockage(tso);
4140   }
4141 }
4142
4143 void
4144 printAllThreads(void)
4145 {
4146   StgTSO *t;
4147
4148 # if defined(GRAN)
4149   char time_string[TIME_STR_LEN], node_str[NODE_STR_LEN];
4150   ullong_format_string(TIME_ON_PROC(CurrentProc),
4151                        time_string, rtsFalse/*no commas!*/);
4152
4153   debugBelch("all threads at [%s]:\n", time_string);
4154 # elif defined(PARALLEL_HASKELL)
4155   char time_string[TIME_STR_LEN], node_str[NODE_STR_LEN];
4156   ullong_format_string(CURRENT_TIME,
4157                        time_string, rtsFalse/*no commas!*/);
4158
4159   debugBelch("all threads at [%s]:\n", time_string);
4160 # else
4161   debugBelch("all threads:\n");
4162 # endif
4163
4164   for (t = all_threads; t != END_TSO_QUEUE; ) {
4165     debugBelch("\tthread %d @ %p ", t->id, (void *)t);
4166 #if defined(DEBUG)
4167     {
4168       void *label = lookupThreadLabel(t->id);
4169       if (label) debugBelch("[\"%s\"] ",(char *)label);
4170     }
4171 #endif
4172     if (t->what_next == ThreadRelocated) {
4173         debugBelch("has been relocated...\n");
4174         t = t->link;
4175     } else {
4176         printThreadStatus(t);
4177         debugBelch("\n");
4178         t = t->global_link;
4179     }
4180   }
4181 }
4182
4183 #ifdef DEBUG
4184
4185 /*
4186    Print a whole blocking queue attached to node (debugging only).
4187 */
4188 # if defined(PARALLEL_HASKELL)
4189 void
4190 print_bq (StgClosure *node)
4191 {
4192   StgBlockingQueueElement *bqe;
4193   StgTSO *tso;
4194   rtsBool end;
4195
4196   debugBelch("## BQ of closure %p (%s): ",
4197           node, info_type(node));
4198
4199   /* should cover all closures that may have a blocking queue */
4200   ASSERT(get_itbl(node)->type == BLACKHOLE_BQ ||
4201          get_itbl(node)->type == FETCH_ME_BQ ||
4202          get_itbl(node)->type == RBH ||
4203          get_itbl(node)->type == MVAR);
4204
4205   ASSERT(node!=(StgClosure*)NULL);         // sanity check
4206
4207   print_bqe(((StgBlockingQueue*)node)->blocking_queue);
4208 }
4209
4210 /*
4211    Print a whole blocking queue starting with the element bqe.
4212 */
4213 void
4214 print_bqe (StgBlockingQueueElement *bqe)
4215 {
4216   rtsBool end;
4217
4218   /*
4219      NB: In a parallel setup a BQ of an RBH must end with an RBH_Save closure;
4220   */
4221   for (end = (bqe==END_BQ_QUEUE);
4222        !end; // iterate until bqe points to a CONSTR
4223        end = (get_itbl(bqe)->type == CONSTR) || (bqe->link==END_BQ_QUEUE),
4224        bqe = end ? END_BQ_QUEUE : bqe->link) {
4225     ASSERT(bqe != END_BQ_QUEUE);                               // sanity check
4226     ASSERT(bqe != (StgBlockingQueueElement *)NULL);            // sanity check
4227     /* types of closures that may appear in a blocking queue */
4228     ASSERT(get_itbl(bqe)->type == TSO ||
4229            get_itbl(bqe)->type == BLOCKED_FETCH ||
4230            get_itbl(bqe)->type == CONSTR);
4231     /* only BQs of an RBH end with an RBH_Save closure */
4232     //ASSERT(get_itbl(bqe)->type != CONSTR || get_itbl(node)->type == RBH);
4233
4234     switch (get_itbl(bqe)->type) {
4235     case TSO:
4236       debugBelch(" TSO %u (%x),",
4237               ((StgTSO *)bqe)->id, ((StgTSO *)bqe));
4238       break;
4239     case BLOCKED_FETCH:
4240       debugBelch(" BF (node=%p, ga=((%x, %d, %x)),",
4241               ((StgBlockedFetch *)bqe)->node,
4242               ((StgBlockedFetch *)bqe)->ga.payload.gc.gtid,
4243               ((StgBlockedFetch *)bqe)->ga.payload.gc.slot,
4244               ((StgBlockedFetch *)bqe)->ga.weight);
4245       break;
4246     case CONSTR:
4247       debugBelch(" %s (IP %p),",
4248               (get_itbl(bqe) == &stg_RBH_Save_0_info ? "RBH_Save_0" :
4249                get_itbl(bqe) == &stg_RBH_Save_1_info ? "RBH_Save_1" :
4250                get_itbl(bqe) == &stg_RBH_Save_2_info ? "RBH_Save_2" :
4251                "RBH_Save_?"), get_itbl(bqe));
4252       break;
4253     default:
4254       barf("Unexpected closure type %s in blocking queue", // of %p (%s)",
4255            info_type((StgClosure *)bqe)); // , node, info_type(node));
4256       break;
4257     }
4258   } /* for */
4259   debugBelch("\n");
4260 }
4261 # elif defined(GRAN)
4262 void
4263 print_bq (StgClosure *node)
4264 {
4265   StgBlockingQueueElement *bqe;
4266   PEs node_loc, tso_loc;
4267   rtsBool end;
4268
4269   /* should cover all closures that may have a blocking queue */
4270   ASSERT(get_itbl(node)->type == BLACKHOLE_BQ ||
4271          get_itbl(node)->type == FETCH_ME_BQ ||
4272          get_itbl(node)->type == RBH);
4273
4274   ASSERT(node!=(StgClosure*)NULL);         // sanity check
4275   node_loc = where_is(node);
4276
4277   debugBelch("## BQ of closure %p (%s) on [PE %d]: ",
4278           node, info_type(node), node_loc);
4279
4280   /*
4281      NB: In a parallel setup a BQ of an RBH must end with an RBH_Save closure;
4282   */
4283   for (bqe = ((StgBlockingQueue*)node)->blocking_queue, end = (bqe==END_BQ_QUEUE);
4284        !end; // iterate until bqe points to a CONSTR
4285        end = (get_itbl(bqe)->type == CONSTR) || (bqe->link==END_BQ_QUEUE), bqe = end ? END_BQ_QUEUE : bqe->link) {
4286     ASSERT(bqe != END_BQ_QUEUE);             // sanity check
4287     ASSERT(bqe != (StgBlockingQueueElement *)NULL);  // sanity check
4288     /* types of closures that may appear in a blocking queue */
4289     ASSERT(get_itbl(bqe)->type == TSO ||
4290            get_itbl(bqe)->type == CONSTR);
4291     /* only BQs of an RBH end with an RBH_Save closure */
4292     ASSERT(get_itbl(bqe)->type != CONSTR || get_itbl(node)->type == RBH);
4293
4294     tso_loc = where_is((StgClosure *)bqe);
4295     switch (get_itbl(bqe)->type) {
4296     case TSO:
4297       debugBelch(" TSO %d (%p) on [PE %d],",
4298               ((StgTSO *)bqe)->id, (StgTSO *)bqe, tso_loc);
4299       break;
4300     case CONSTR:
4301       debugBelch(" %s (IP %p),",
4302               (get_itbl(bqe) == &stg_RBH_Save_0_info ? "RBH_Save_0" :
4303                get_itbl(bqe) == &stg_RBH_Save_1_info ? "RBH_Save_1" :
4304                get_itbl(bqe) == &stg_RBH_Save_2_info ? "RBH_Save_2" :
4305                "RBH_Save_?"), get_itbl(bqe));
4306       break;
4307     default:
4308       barf("Unexpected closure type %s in blocking queue of %p (%s)",
4309            info_type((StgClosure *)bqe), node, info_type(node));
4310       break;
4311     }
4312   } /* for */
4313   debugBelch("\n");
4314 }
4315 # endif
4316
4317 #if defined(PARALLEL_HASKELL)
4318 static nat
4319 run_queue_len(void)
4320 {
4321   nat i;
4322   StgTSO *tso;
4323
4324   for (i=0, tso=run_queue_hd;
4325        tso != END_TSO_QUEUE;
4326        i++, tso=tso->link)
4327     /* nothing */
4328
4329   return i;
4330 }
4331 #endif
4332
4333 void
4334 sched_belch(char *s, ...)
4335 {
4336   va_list ap;
4337   va_start(ap,s);
4338 #ifdef RTS_SUPPORTS_THREADS
4339   debugBelch("sched (task %p): ", osThreadId());
4340 #elif defined(PARALLEL_HASKELL)
4341   debugBelch("== ");
4342 #else
4343   debugBelch("sched: ");
4344 #endif
4345   vdebugBelch(s, ap);
4346   debugBelch("\n");
4347   va_end(ap);
4348 }
4349
4350 #endif /* DEBUG */