ghc/rts/Schedule.c

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