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