ghc/rts/Schedule.c

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