ghc/rts/Schedule.c

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