ghc/rts/Schedule.c

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