ghc/rts/Schedule.c

   1 /* ---------------------------------------------------------------------------
   2  * $Id: Schedule.c,v 1.55 2000/03/17 13:30:24 simonmar Exp $
   3  *
   4  * (c) The GHC Team, 1998-2000
   5  *
   6  * Scheduler
   7  *
   8  * The main scheduling code in GranSim is quite different from that in std
   9  * (concurrent) Haskell: while concurrent Haskell just iterates over the
  10  * threads in the runnable queue, GranSim is event driven, i.e. it iterates
  11  * over the events in the global event queue.  -- HWL
  12  * --------------------------------------------------------------------------*/
  13
  14 //@node Main scheduling code, , ,
  15 //@section Main scheduling code
  16
  17 /* Version with scheduler monitor support for SMPs.
  18
  19    This design provides a high-level API to create and schedule threads etc.
  20    as documented in the SMP design document.
  21
  22    It uses a monitor design controlled by a single mutex to exercise control
  23    over accesses to shared data structures, and builds on the Posix threads
  24    library.
  25
  26    The majority of state is shared.  In order to keep essential per-task state,
  27    there is a Capability structure, which contains all the information
  28    needed to run a thread: its STG registers, a pointer to its TSO, a
  29    nursery etc.  During STG execution, a pointer to the capability is
  30    kept in a register (BaseReg).
  31
  32    In a non-SMP build, there is one global capability, namely MainRegTable.
  33
  34    SDM & KH, 10/99
  35 */
  36
  37 //@menu
  38 //* Includes::
  39 //* Variables and Data structures::
  40 //* Prototypes::
  41 //* Main scheduling loop::
  42 //* Suspend and Resume::
  43 //* Run queue code::
  44 //* Garbage Collextion Routines::
  45 //* Blocking Queue Routines::
  46 //* Exception Handling Routines::
  47 //* Debugging Routines::
  48 //* Index::
  49 //@end menu
  50
  51 //@node Includes, Variables and Data structures, Main scheduling code, Main scheduling code
  52 //@subsection Includes
  53
  54 #include "Rts.h"
  55 #include "SchedAPI.h"
  56 #include "RtsUtils.h"
  57 #include "RtsFlags.h"
  58 #include "Storage.h"
  59 #include "StgRun.h"
  60 #include "StgStartup.h"
  61 #include "GC.h"
  62 #include "Hooks.h"
  63 #include "Schedule.h"
  64 #include "StgMiscClosures.h"
  65 #include "Storage.h"
  66 #include "Evaluator.h"
  67 #include "Exception.h"
  68 #include "Printer.h"
  69 #include "Main.h"
  70 #include "Signals.h"
  71 #include "Profiling.h"
  72 #include "Sanity.h"
  73 #include "Stats.h"
  74 #include "Sparks.h"
  75 #include "Prelude.h"
  76 #if defined(GRAN) || defined(PAR)
  77 # include "GranSimRts.h"
  78 # include "GranSim.h"
  79 # include "ParallelRts.h"
  80 # include "Parallel.h"
  81 # include "ParallelDebug.h"
  82 # include "FetchMe.h"
  83 # include "HLC.h"
  84 #endif
  85
  86 #include <stdarg.h>
  87
  88 //@node Variables and Data structures, Prototypes, Includes, Main scheduling code
  89 //@subsection Variables and Data structures
  90
  91 /* Main threads:
  92  *
  93  * These are the threads which clients have requested that we run.
  94  *
  95  * In an SMP build, we might have several concurrent clients all
  96  * waiting for results, and each one will wait on a condition variable
  97  * until the result is available.
  98  *
  99  * In non-SMP, clients are strictly nested: the first client calls
 100  * into the RTS, which might call out again to C with a _ccall_GC, and
 101  * eventually re-enter the RTS.
 102  *
 103  * Main threads information is kept in a linked list:
 104  */
 105 //@cindex StgMainThread
 106 typedef struct StgMainThread_ {
 107   StgTSO *         tso;
 108   SchedulerStatus  stat;
 109   StgClosure **    ret;
 110 #ifdef SMP
 111   pthread_cond_t wakeup;
 112 #endif
 113   struct StgMainThread_ *link;
 114 } StgMainThread;
 115
 116 /* Main thread queue.
 117  * Locks required: sched_mutex.
 118  */
 119 static StgMainThread *main_threads;
 120
 121 /* Thread queues.
 122  * Locks required: sched_mutex.
 123  */
 124 #if defined(GRAN)
 125
 126 StgTSO* ActiveTSO = NULL; /* for assigning system costs; GranSim-Light only */
 127 /* rtsTime TimeOfNextEvent, EndOfTimeSlice;            now in GranSim.c */
 128
 129 /*
 130    In GranSim we have a runable and a blocked queue for each processor.
 131    In order to minimise code changes new arrays run_queue_hds/tls
 132    are created. run_queue_hd is then a short cut (macro) for
 133    run_queue_hds[CurrentProc] (see GranSim.h).
 134    -- HWL
 135 */
 136 StgTSO *run_queue_hds[MAX_PROC], *run_queue_tls[MAX_PROC];
 137 StgTSO *blocked_queue_hds[MAX_PROC], *blocked_queue_tls[MAX_PROC];
 138 StgTSO *ccalling_threadss[MAX_PROC];
 139 StgTSO *all_threadss[MAX_PROC];
 140
 141 #else /* !GRAN */
 142
 143 StgTSO *run_queue_hd, *run_queue_tl;
 144 StgTSO *blocked_queue_hd, *blocked_queue_tl;
 145
 146 /* Linked list of all threads.
 147  * Used for detecting garbage collected threads.
 148  */
 149 StgTSO *all_threads;
 150
 151 /* Threads suspended in _ccall_GC.
 152  */
 153 static StgTSO *suspended_ccalling_threads;
 154
 155 static void GetRoots(void);
 156 static StgTSO *threadStackOverflow(StgTSO *tso);
 157 #endif
 158
 159 /* KH: The following two flags are shared memory locations.  There is no need
 160        to lock them, since they are only unset at the end of a scheduler
 161        operation.
 162 */
 163
 164 /* flag set by signal handler to precipitate a context switch */
 165 //@cindex context_switch
 166 nat context_switch;
 167
 168 /* if this flag is set as well, give up execution */
 169 //@cindex interrupted
 170 rtsBool interrupted;
 171
 172 /* Next thread ID to allocate.
 173  * Locks required: sched_mutex
 174  */
 175 //@cindex next_thread_id
 176 StgThreadID next_thread_id = 1;
 177
 178 /*
 179  * Pointers to the state of the current thread.
 180  * Rule of thumb: if CurrentTSO != NULL, then we're running a Haskell
 181  * thread.  If CurrentTSO == NULL, then we're at the scheduler level.
 182  */
 183
 184 /* The smallest stack size that makes any sense is:
 185  *    RESERVED_STACK_WORDS    (so we can get back from the stack overflow)
 186  *  + sizeofW(StgStopFrame)   (the stg_stop_thread_info frame)
 187  *  + 1                       (the realworld token for an IO thread)
 188  *  + 1                       (the closure to enter)
 189  *
 190  * A thread with this stack will bomb immediately with a stack
 191  * overflow, which will increase its stack size.
 192  */
 193
 194 #define MIN_STACK_WORDS (RESERVED_STACK_WORDS + sizeofW(StgStopFrame) + 2)
 195
 196 /* Free capability list.
 197  * Locks required: sched_mutex.
 198  */
 199 #ifdef SMP
 200 //@cindex free_capabilities
 201 //@cindex n_free_capabilities
 202 Capability *free_capabilities; /* Available capabilities for running threads */
 203 nat n_free_capabilities;       /* total number of available capabilities */
 204 #else
 205 //@cindex MainRegTable
 206 Capability MainRegTable;       /* for non-SMP, we have one global capability */
 207 #endif
 208
 209 #if defined(GRAN)
 210 StgTSO      *CurrentTSOs[MAX_PROC];
 211 #else
 212 StgTSO      *CurrentTSO;
 213 #endif
 214
 215 rtsBool ready_to_gc;
 216
 217 /* All our current task ids, saved in case we need to kill them later.
 218  */
 219 #ifdef SMP
 220 //@cindex task_ids
 221 task_info *task_ids;
 222 #endif
 223
 224 void            addToBlockedQueue ( StgTSO *tso );
 225
 226 static void     schedule          ( void );
 227        void     interruptStgRts   ( void );
 228 static StgTSO * createThread_     ( nat size, rtsBool have_lock );
 229
 230 #ifdef DEBUG
 231 static void sched_belch(char *s, ...);
 232 #endif
 233
 234 #ifdef SMP
 235 //@cindex sched_mutex
 236 //@cindex term_mutex
 237 //@cindex thread_ready_cond
 238 //@cindex gc_pending_cond
 239 pthread_mutex_t sched_mutex       = PTHREAD_MUTEX_INITIALIZER;
 240 pthread_mutex_t term_mutex        = PTHREAD_MUTEX_INITIALIZER;
 241 pthread_cond_t  thread_ready_cond = PTHREAD_COND_INITIALIZER;
 242 pthread_cond_t  gc_pending_cond   = PTHREAD_COND_INITIALIZER;
 243
 244 nat await_death;
 245 #endif
 246
 247 #if defined(PAR)
 248 StgTSO *LastTSO;
 249 rtsTime TimeOfLastYield;
 250 #endif
 251
 252 /*
 253  * The thread state for the main thread.
 254 // ToDo: check whether not needed any more
 255 StgTSO   *MainTSO;
 256  */
 257
 258
 259 //@node Prototypes, Main scheduling loop, Variables and Data structures, Main scheduling code
 260 //@subsection Prototypes
 261
 262 //@node Main scheduling loop, Suspend and Resume, Prototypes, Main scheduling code
 263 //@subsection Main scheduling loop
 264
 265 /* ---------------------------------------------------------------------------
 266    Main scheduling loop.
 267
 268    We use round-robin scheduling, each thread returning to the
 269    scheduler loop when one of these conditions is detected:
 270
 271       * out of heap space
 272       * timer expires (thread yields)
 273       * thread blocks
 274       * thread ends
 275       * stack overflow
 276
 277    Locking notes:  we acquire the scheduler lock once at the beginning
 278    of the scheduler loop, and release it when
 279
 280       * running a thread, or
 281       * waiting for work, or
 282       * waiting for a GC to complete.
 283
 284    ------------------------------------------------------------------------ */
 285 //@cindex schedule
 286 static void
 287 schedule( void )
 288 {
 289   StgTSO *t;
 290   Capability *cap;
 291   StgThreadReturnCode ret;
 292 #if defined(GRAN)
 293   rtsEvent *event;
 294 #elif defined(PAR)
 295   rtsSpark spark;
 296   StgTSO *tso;
 297   GlobalTaskId pe;
 298 #endif
 299   rtsBool was_interrupted = rtsFalse;
 300
 301   ACQUIRE_LOCK(&sched_mutex);
 302
 303 #if defined(GRAN)
 304 # error ToDo: implement GranSim scheduler
 305 #elif defined(PAR)
 306   while (!GlobalStopPending) {          /* GlobalStopPending set in par_exit */
 307
 308     if (PendingFetches != END_BF_QUEUE) {
 309         processFetches();
 310     }
 311 #else
 312   while (1) {
 313 #endif
 314
 315     IF_DEBUG(scheduler, printAllThreads());
 316
 317     /* If we're interrupted (the user pressed ^C, or some other
 318      * termination condition occurred), kill all the currently running
 319      * threads.
 320      */
 321     if (interrupted) {
 322       IF_DEBUG(scheduler, sched_belch("interrupted"));
 323       for (t = run_queue_hd; t != END_TSO_QUEUE; t = t->link) {
 324         deleteThread(t);
 325       }
 326       for (t = blocked_queue_hd; t != END_TSO_QUEUE; t = t->link) {
 327         deleteThread(t);
 328       }
 329       run_queue_hd = run_queue_tl = END_TSO_QUEUE;
 330       blocked_queue_hd = blocked_queue_tl = END_TSO_QUEUE;
 331       interrupted = rtsFalse;
 332       was_interrupted = rtsTrue;
 333     }
 334
 335     /* Go through the list of main threads and wake up any
 336      * clients whose computations have finished.  ToDo: this
 337      * should be done more efficiently without a linear scan
 338      * of the main threads list, somehow...
 339      */
 340 #ifdef SMP
 341     {
 342       StgMainThread *m, **prev;
 343       prev = &main_threads;
 344       for (m = main_threads; m != NULL; m = m->link) {
 345         switch (m->tso->what_next) {
 346         case ThreadComplete:
 347           if (m->ret) {
 348             *(m->ret) = (StgClosure *)m->tso->sp[0];
 349           }
 350           *prev = m->link;
 351           m->stat = Success;
 352           pthread_cond_broadcast(&m->wakeup);
 353           break;
 354         case ThreadKilled:
 355           *prev = m->link;
 356           if (was_interrupted) {
 357             m->stat = Interrupted;
 358           } else {
 359             m->stat = Killed;
 360           }
 361           pthread_cond_broadcast(&m->wakeup);
 362           break;
 363         default:
 364           break;
 365         }
 366       }
 367     }
 368 #else
 369     /* If our main thread has finished or been killed, return.
 370      */
 371     {
 372       StgMainThread *m = main_threads;
 373       if (m->tso->what_next == ThreadComplete
 374           || m->tso->what_next == ThreadKilled) {
 375         main_threads = main_threads->link;
 376         if (m->tso->what_next == ThreadComplete) {
 377           /* we finished successfully, fill in the return value */
 378           if (m->ret) { *(m->ret) = (StgClosure *)m->tso->sp[0]; };
 379           m->stat = Success;
 380           return;
 381         } else {
 382           if (was_interrupted) {
 383             m->stat = Interrupted;
 384           } else {
 385             m->stat = Killed;
 386           }
 387           return;
 388         }
 389       }
 390     }
 391 #endif
 392
 393     /* Top up the run queue from our spark pool.  We try to make the
 394      * number of threads in the run queue equal to the number of
 395      * free capabilities.
 396      */
 397 #if defined(SMP)
 398     {
 399       nat n = n_free_capabilities;
 400       StgTSO *tso = run_queue_hd;
 401
 402       /* Count the run queue */
 403       while (n > 0 && tso != END_TSO_QUEUE) {
 404         tso = tso->link;
 405         n--;
 406       }
 407
 408       for (; n > 0; n--) {
 409         StgClosure *spark;
 410         spark = findSpark();
 411         if (spark == NULL) {
 412           break; /* no more sparks in the pool */
 413         } else {
 414           /* I'd prefer this to be done in activateSpark -- HWL */
 415           /* tricky - it needs to hold the scheduler lock and
 416            * not try to re-acquire it -- SDM */
 417           StgTSO *tso;
 418           tso = createThread_(RtsFlags.GcFlags.initialStkSize, rtsTrue);
 419           pushClosure(tso,spark);
 420           PUSH_ON_RUN_QUEUE(tso);
 421 #ifdef PAR
 422           advisory_thread_count++;
 423 #endif
 424
 425           IF_DEBUG(scheduler,
 426                    sched_belch("turning spark of closure %p into a thread",
 427                                (StgClosure *)spark));
 428         }
 429       }
 430       /* We need to wake up the other tasks if we just created some
 431        * work for them.
 432        */
 433       if (n_free_capabilities - n > 1) {
 434           pthread_cond_signal(&thread_ready_cond);
 435       }
 436     }
 437 #endif /* SMP */
 438
 439     /* Check whether any waiting threads need to be woken up.  If the
 440      * run queue is empty, and there are no other tasks running, we
 441      * can wait indefinitely for something to happen.
 442      * ToDo: what if another client comes along & requests another
 443      * main thread?
 444      */
 445     if (blocked_queue_hd != END_TSO_QUEUE) {
 446       awaitEvent(
 447            (run_queue_hd == END_TSO_QUEUE)
 448 #ifdef SMP
 449         && (n_free_capabilities == RtsFlags.ParFlags.nNodes)
 450 #endif
 451         );
 452     }
 453
 454     /* check for signals each time around the scheduler */
 455 #ifndef __MINGW32__
 456     if (signals_pending()) {
 457       start_signal_handlers();
 458     }
 459 #endif
 460
 461     /* Detect deadlock: when we have no threads to run, there are
 462      * no threads waiting on I/O or sleeping, and all the other
 463      * tasks are waiting for work, we must have a deadlock.  Inform
 464      * all the main threads.
 465      */
 466 #ifdef SMP
 467     if (blocked_queue_hd == END_TSO_QUEUE
 468         && run_queue_hd == END_TSO_QUEUE
 469         && (n_free_capabilities == RtsFlags.ParFlags.nNodes)
 470         ) {
 471       StgMainThread *m;
 472       for (m = main_threads; m != NULL; m = m->link) {
 473           m->ret = NULL;
 474           m->stat = Deadlock;
 475           pthread_cond_broadcast(&m->wakeup);
 476       }
 477       main_threads = NULL;
 478     }
 479 #else /* ! SMP */
 480     if (blocked_queue_hd == END_TSO_QUEUE
 481         && run_queue_hd == END_TSO_QUEUE) {
 482       StgMainThread *m = main_threads;
 483       m->ret = NULL;
 484       m->stat = Deadlock;
 485       main_threads = m->link;
 486       return;
 487     }
 488 #endif
 489
 490 #ifdef SMP
 491     /* If there's a GC pending, don't do anything until it has
 492      * completed.
 493      */
 494     if (ready_to_gc) {
 495       IF_DEBUG(scheduler,sched_belch("waiting for GC"));
 496       pthread_cond_wait(&gc_pending_cond, &sched_mutex);
 497     }
 498
 499     /* block until we've got a thread on the run queue and a free
 500      * capability.
 501      */
 502     while (run_queue_hd == END_TSO_QUEUE || free_capabilities == NULL) {
 503       IF_DEBUG(scheduler, sched_belch("waiting for work"));
 504       pthread_cond_wait(&thread_ready_cond, &sched_mutex);
 505       IF_DEBUG(scheduler, sched_belch("work now available"));
 506     }
 507 #endif
 508
 509 #if defined(GRAN)
 510 # error ToDo: implement GranSim scheduler
 511 #elif defined(PAR)
 512     /* ToDo: phps merge with spark activation above */
 513     /* check whether we have local work and send requests if we have none */
 514     if (run_queue_hd == END_TSO_QUEUE) {  /* no runnable threads */
 515       /* :-[  no local threads => look out for local sparks */
 516       if (advisory_thread_count < RtsFlags.ParFlags.maxThreads &&
 517           (pending_sparks_hd[REQUIRED_POOL] < pending_sparks_tl[REQUIRED_POOL] ||
 518            pending_sparks_hd[ADVISORY_POOL] < pending_sparks_tl[ADVISORY_POOL])) {
 519         /*
 520          * ToDo: add GC code check that we really have enough heap afterwards!!
 521          * Old comment:
 522          * If we're here (no runnable threads) and we have pending
 523          * sparks, we must have a space problem.  Get enough space
 524          * to turn one of those pending sparks into a
 525          * thread...
 526          */
 527
 528         spark = findSpark();                /* get a spark */
 529         if (spark != (rtsSpark) NULL) {
 530           tso = activateSpark(spark);       /* turn the spark into a thread */
 531           IF_PAR_DEBUG(verbose,
 532                        belch("== [%x] schedule: Created TSO %p (%d); %d threads active",
 533                              mytid, tso, tso->id, advisory_thread_count));
 534
 535           if (tso==END_TSO_QUEUE) { /* failed to activate spark->back to loop */
 536             belch("^^ failed to activate spark");
 537             goto next_thread;
 538           }               /* otherwise fall through & pick-up new tso */
 539         } else {
 540           IF_PAR_DEBUG(verbose,
 541                        belch("^^ no local sparks (spark pool contains only NFs: %d)",
 542                              spark_queue_len(ADVISORY_POOL)));
 543           goto next_thread;
 544         }
 545       } else
 546       /* =8-[  no local sparks => look for work on other PEs */
 547       {
 548         /*
 549          * We really have absolutely no work.  Send out a fish
 550          * (there may be some out there already), and wait for
 551          * something to arrive.  We clearly can't run any threads
 552          * until a SCHEDULE or RESUME arrives, and so that's what
 553          * we're hoping to see.  (Of course, we still have to
 554          * respond to other types of messages.)
 555          */
 556         if (//!fishing &&
 557             outstandingFishes < RtsFlags.ParFlags.maxFishes ) { // &&
 558           // (last_fish_arrived_at+FISH_DELAY < CURRENT_TIME)) {
 559           /* fishing set in sendFish, processFish;
 560              avoid flooding system with fishes via delay */
 561           pe = choosePE();
 562           sendFish(pe, mytid, NEW_FISH_AGE, NEW_FISH_HISTORY,
 563                    NEW_FISH_HUNGER);
 564         }
 565
 566         processMessages();
 567         goto next_thread;
 568         // ReSchedule(0);
 569       }
 570     } else if (PacketsWaiting()) {  /* Look for incoming messages */
 571       processMessages();
 572     }
 573
 574     /* Now we are sure that we have some work available */
 575     ASSERT(run_queue_hd != END_TSO_QUEUE);
 576     /* Take a thread from the run queue, if we have work */
 577     t = take_off_run_queue(END_TSO_QUEUE);
 578
 579     /* ToDo: write something to the log-file
 580     if (RTSflags.ParFlags.granSimStats && !sameThread)
 581         DumpGranEvent(GR_SCHEDULE, RunnableThreadsHd);
 582     */
 583
 584     CurrentTSO = t;
 585
 586     IF_DEBUG(scheduler, belch("--^^ %d sparks on [%#x] (hd=%x; tl=%x; lim=%x)",
 587                               spark_queue_len(ADVISORY_POOL), CURRENT_PROC,
 588                               pending_sparks_hd[ADVISORY_POOL],
 589                               pending_sparks_tl[ADVISORY_POOL],
 590                               pending_sparks_lim[ADVISORY_POOL]));
 591
 592     IF_DEBUG(scheduler, belch("--== %d threads on [%#x] (hd=%x; tl=%x)",
 593                               run_queue_len(), CURRENT_PROC,
 594                               run_queue_hd, run_queue_tl));
 595
 596     if (t != LastTSO) {
 597       /*
 598          we are running a different TSO, so write a schedule event to log file
 599          NB: If we use fair scheduling we also have to write  a deschedule
 600              event for LastTSO; with unfair scheduling we know that the
 601              previous tso has blocked whenever we switch to another tso, so
 602              we don't need it in GUM for now
 603       */
 604       DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
 605                        GR_SCHEDULE, t, (StgClosure *)NULL, 0, 0);
 606
 607     }
 608
 609 #else /* !GRAN && !PAR */
 610
 611     /* grab a thread from the run queue
 612      */
 613     t = POP_RUN_QUEUE();
 614     IF_DEBUG(sanity,checkTSO(t));
 615
 616 #endif
 617
 618     /* grab a capability
 619      */
 620 #ifdef SMP
 621     cap = free_capabilities;
 622     free_capabilities = cap->link;
 623     n_free_capabilities--;
 624 #else
 625     cap = &MainRegTable;
 626 #endif
 627
 628     cap->rCurrentTSO = t;
 629
 630     /* set the context_switch flag
 631      */
 632     if (run_queue_hd == END_TSO_QUEUE)
 633       context_switch = 0;
 634     else
 635       context_switch = 1;
 636
 637     RELEASE_LOCK(&sched_mutex);
 638
 639     IF_DEBUG(scheduler,sched_belch("running thread %d", t->id));
 640
 641     /* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
 642     /* Run the current thread
 643      */
 644     switch (cap->rCurrentTSO->what_next) {
 645     case ThreadKilled:
 646     case ThreadComplete:
 647       /* Thread already finished, return to scheduler. */
 648       ret = ThreadFinished;
 649       break;
 650     case ThreadEnterGHC:
 651       ret = StgRun((StgFunPtr) stg_enterStackTop, cap);
 652       break;
 653     case ThreadRunGHC:
 654       ret = StgRun((StgFunPtr) stg_returnToStackTop, cap);
 655       break;
 656     case ThreadEnterHugs:
 657 #ifdef INTERPRETER
 658       {
 659          StgClosure* c;
 660          IF_DEBUG(scheduler,sched_belch("entering Hugs"));
 661          c = (StgClosure *)(cap->rCurrentTSO->sp[0]);
 662          cap->rCurrentTSO->sp += 1;
 663          ret = enter(cap,c);
 664          break;
 665       }
 666 #else
 667       barf("Panic: entered a BCO but no bytecode interpreter in this build");
 668 #endif
 669     default:
 670       barf("schedule: invalid what_next field");
 671     }
 672     /* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
 673
 674     /* Costs for the scheduler are assigned to CCS_SYSTEM */
 675 #ifdef PROFILING
 676     CCCS = CCS_SYSTEM;
 677 #endif
 678
 679     ACQUIRE_LOCK(&sched_mutex);
 680
 681 #ifdef SMP
 682     IF_DEBUG(scheduler,fprintf(stderr,"scheduler (task %ld): ", pthread_self()););
 683 #else
 684     IF_DEBUG(scheduler,fprintf(stderr,"scheduler: "););
 685 #endif
 686     t = cap->rCurrentTSO;
 687
 688     switch (ret) {
 689     case HeapOverflow:
 690       /* make all the running tasks block on a condition variable,
 691        * maybe set context_switch and wait till they all pile in,
 692        * then have them wait on a GC condition variable.
 693        */
 694       IF_DEBUG(scheduler,belch("thread %ld stopped: HeapOverflow", t->id));
 695       threadPaused(t);
 696
 697       ready_to_gc = rtsTrue;
 698       context_switch = 1;               /* stop other threads ASAP */
 699       PUSH_ON_RUN_QUEUE(t);
 700       break;
 701
 702     case StackOverflow:
 703       /* just adjust the stack for this thread, then pop it back
 704        * on the run queue.
 705        */
 706       IF_DEBUG(scheduler,belch("thread %ld stopped, StackOverflow", t->id));
 707       threadPaused(t);
 708       {
 709         StgMainThread *m;
 710         /* enlarge the stack */
 711         StgTSO *new_t = threadStackOverflow(t);
 712
 713         /* This TSO has moved, so update any pointers to it from the
 714          * main thread stack.  It better not be on any other queues...
 715          * (it shouldn't be).
 716          */
 717         for (m = main_threads; m != NULL; m = m->link) {
 718           if (m->tso == t) {
 719             m->tso = new_t;
 720           }
 721         }
 722         threadPaused(new_t);
 723         PUSH_ON_RUN_QUEUE(new_t);
 724       }
 725       break;
 726
 727     case ThreadYielding:
 728 #if defined(GRAN)
 729       IF_DEBUG(gran,
 730                DumpGranEvent(GR_DESCHEDULE, t));
 731       globalGranStats.tot_yields++;
 732 #elif defined(PAR)
 733       IF_DEBUG(par,
 734                DumpGranEvent(GR_DESCHEDULE, t));
 735 #endif
 736       /* put the thread back on the run queue.  Then, if we're ready to
 737        * GC, check whether this is the last task to stop.  If so, wake
 738        * up the GC thread.  getThread will block during a GC until the
 739        * GC is finished.
 740        */
 741       IF_DEBUG(scheduler,
 742                if (t->what_next == ThreadEnterHugs) {
 743                  /* ToDo: or maybe a timer expired when we were in Hugs?
 744                   * or maybe someone hit ctrl-C
 745                   */
 746                  belch("thread %ld stopped to switch to Hugs", t->id);
 747                } else {
 748                  belch("thread %ld stopped, yielding", t->id);
 749                }
 750                );
 751       threadPaused(t);
 752       APPEND_TO_RUN_QUEUE(t);
 753       break;
 754
 755     case ThreadBlocked:
 756 #if defined(GRAN)
 757 # error ToDo: implement GranSim scheduler
 758 #elif defined(PAR)
 759       IF_DEBUG(par,
 760                DumpGranEvent(GR_DESCHEDULE, t));
 761 #else
 762 #endif
 763       /* don't need to do anything.  Either the thread is blocked on
 764        * I/O, in which case we'll have called addToBlockedQueue
 765        * previously, or it's blocked on an MVar or Blackhole, in which
 766        * case it'll be on the relevant queue already.
 767        */
 768       IF_DEBUG(scheduler,
 769                fprintf(stderr, "thread %d stopped, ", t->id);
 770                printThreadBlockage(t);
 771                fprintf(stderr, "\n"));
 772       threadPaused(t);
 773       break;
 774
 775     case ThreadFinished:
 776       /* Need to check whether this was a main thread, and if so, signal
 777        * the task that started it with the return value.  If we have no
 778        * more main threads, we probably need to stop all the tasks until
 779        * we get a new one.
 780        */
 781       IF_DEBUG(scheduler,belch("thread %ld finished", t->id));
 782       t->what_next = ThreadComplete;
 783 #if defined(GRAN)
 784       // ToDo: endThread(t, CurrentProc); // clean-up the thread
 785 #elif defined(PAR)
 786       advisory_thread_count--;
 787       if (RtsFlags.ParFlags.ParStats.Full)
 788         DumpEndEvent(CURRENT_PROC, t, rtsFalse /* not mandatory */);
 789 #endif
 790       break;
 791
 792     default:
 793       barf("doneThread: invalid thread return code");
 794     }
 795
 796 #ifdef SMP
 797     cap->link = free_capabilities;
 798     free_capabilities = cap;
 799     n_free_capabilities++;
 800 #endif
 801
 802 #ifdef SMP
 803     if (ready_to_gc && n_free_capabilities == RtsFlags.ParFlags.nNodes)
 804 #else
 805     if (ready_to_gc)
 806 #endif
 807       {
 808       /* everybody back, start the GC.
 809        * Could do it in this thread, or signal a condition var
 810        * to do it in another thread.  Either way, we need to
 811        * broadcast on gc_pending_cond afterward.
 812        */
 813 #ifdef SMP
 814       IF_DEBUG(scheduler,sched_belch("doing GC"));
 815 #endif
 816       GarbageCollect(GetRoots);
 817       ready_to_gc = rtsFalse;
 818 #ifdef SMP
 819       pthread_cond_broadcast(&gc_pending_cond);
 820 #endif
 821     }
 822 #if defined(GRAN)
 823   next_thread:
 824     IF_GRAN_DEBUG(unused,
 825                   print_eventq(EventHd));
 826
 827     event = get_next_event();
 828
 829 #elif defined(PAR)
 830   next_thread:
 831     /* ToDo: wait for next message to arrive rather than busy wait */
 832
 833 #else /* GRAN */
 834   /* not any more
 835   next_thread:
 836     t = take_off_run_queue(END_TSO_QUEUE);
 837   */
 838 #endif /* GRAN */
 839   } /* end of while(1) */
 840 }
 841
 842 /* A hack for Hugs concurrency support.  Needs sanitisation (?) */
 843 void deleteAllThreads ( void )
 844 {
 845   StgTSO* t;
 846   IF_DEBUG(scheduler,sched_belch("deleteAllThreads()"));
 847   for (t = run_queue_hd; t != END_TSO_QUEUE; t = t->link) {
 848     deleteThread(t);
 849   }
 850   for (t = blocked_queue_hd; t != END_TSO_QUEUE; t = t->link) {
 851     deleteThread(t);
 852   }
 853   run_queue_hd = run_queue_tl = END_TSO_QUEUE;
 854   blocked_queue_hd = blocked_queue_tl = END_TSO_QUEUE;
 855 }
 856
 857 /* startThread and  insertThread are now in GranSim.c -- HWL */
 858
 859 //@node Suspend and Resume, Run queue code, Main scheduling loop, Main scheduling code
 860 //@subsection Suspend and Resume
 861
 862 /* ---------------------------------------------------------------------------
 863  * Suspending & resuming Haskell threads.
 864  *
 865  * When making a "safe" call to C (aka _ccall_GC), the task gives back
 866  * its capability before calling the C function.  This allows another
 867  * task to pick up the capability and carry on running Haskell
 868  * threads.  It also means that if the C call blocks, it won't lock
 869  * the whole system.
 870  *
 871  * The Haskell thread making the C call is put to sleep for the
 872  * duration of the call, on the susepended_ccalling_threads queue.  We
 873  * give out a token to the task, which it can use to resume the thread
 874  * on return from the C function.
 875  * ------------------------------------------------------------------------- */
 876
 877 StgInt
 878 suspendThread( Capability *cap )
 879 {
 880   nat tok;
 881
 882   ACQUIRE_LOCK(&sched_mutex);
 883
 884   IF_DEBUG(scheduler,
 885            sched_belch("thread %d did a _ccall_gc\n", cap->rCurrentTSO->id));
 886
 887   threadPaused(cap->rCurrentTSO);
 888   cap->rCurrentTSO->link = suspended_ccalling_threads;
 889   suspended_ccalling_threads = cap->rCurrentTSO;
 890
 891   /* Use the thread ID as the token; it should be unique */
 892   tok = cap->rCurrentTSO->id;
 893
 894 #ifdef SMP
 895   cap->link = free_capabilities;
 896   free_capabilities = cap;
 897   n_free_capabilities++;
 898 #endif
 899
 900   RELEASE_LOCK(&sched_mutex);
 901   return tok;
 902 }
 903
 904 Capability *
 905 resumeThread( StgInt tok )
 906 {
 907   StgTSO *tso, **prev;
 908   Capability *cap;
 909
 910   ACQUIRE_LOCK(&sched_mutex);
 911
 912   prev = &suspended_ccalling_threads;
 913   for (tso = suspended_ccalling_threads;
 914        tso != END_TSO_QUEUE;
 915        prev = &tso->link, tso = tso->link) {
 916     if (tso->id == (StgThreadID)tok) {
 917       *prev = tso->link;
 918       break;
 919     }
 920   }
 921   if (tso == END_TSO_QUEUE) {
 922     barf("resumeThread: thread not found");
 923   }
 924
 925 #ifdef SMP
 926   while (free_capabilities == NULL) {
 927     IF_DEBUG(scheduler, sched_belch("waiting to resume"));
 928     pthread_cond_wait(&thread_ready_cond, &sched_mutex);
 929     IF_DEBUG(scheduler, sched_belch("resuming thread %d", tso->id));
 930   }
 931   cap = free_capabilities;
 932   free_capabilities = cap->link;
 933   n_free_capabilities--;
 934 #else
 935   cap = &MainRegTable;
 936 #endif
 937
 938   cap->rCurrentTSO = tso;
 939
 940   RELEASE_LOCK(&sched_mutex);
 941   return cap;
 942 }
 943
 944
 945 /* ---------------------------------------------------------------------------
 946  * Static functions
 947  * ------------------------------------------------------------------------ */
 948 static void unblockThread(StgTSO *tso);
 949
 950 /* ---------------------------------------------------------------------------
 951  * Comparing Thread ids.
 952  *
 953  * This is used from STG land in the implementation of the
 954  * instances of Eq/Ord for ThreadIds.
 955  * ------------------------------------------------------------------------ */
 956
 957 int cmp_thread(const StgTSO *tso1, const StgTSO *tso2)
 958 {
 959   StgThreadID id1 = tso1->id;
 960   StgThreadID id2 = tso2->id;
 961
 962   if (id1 < id2) return (-1);
 963   if (id1 > id2) return 1;
 964   return 0;
 965 }
 966
 967 /* ---------------------------------------------------------------------------
 968    Create a new thread.
 969
 970    The new thread starts with the given stack size.  Before the
 971    scheduler can run, however, this thread needs to have a closure
 972    (and possibly some arguments) pushed on its stack.  See
 973    pushClosure() in Schedule.h.
 974
 975    createGenThread() and createIOThread() (in SchedAPI.h) are
 976    convenient packaged versions of this function.
 977    ------------------------------------------------------------------------ */
 978 //@cindex createThread
 979 #if defined(GRAN)
 980 /* currently pri (priority) is only used in a GRAN setup -- HWL */
 981 StgTSO *
 982 createThread(nat stack_size, StgInt pri)
 983 {
 984   return createThread_(stack_size, rtsFalse, pri);
 985 }
 986
 987 static StgTSO *
 988 createThread_(nat size, rtsBool have_lock, StgInt pri)
 989 {
 990 #else
 991 StgTSO *
 992 createThread(nat stack_size)
 993 {
 994   return createThread_(stack_size, rtsFalse);
 995 }
 996
 997 static StgTSO *
 998 createThread_(nat size, rtsBool have_lock)
 999 {
1000 #endif
1001     StgTSO *tso;
1002     nat stack_size;
1003
1004     /* First check whether we should create a thread at all */
1005 #if defined(PAR)
1006   /* check that no more than RtsFlags.ParFlags.maxThreads threads are created */
1007   if (advisory_thread_count >= RtsFlags.ParFlags.maxThreads) {
1008     threadsIgnored++;
1009     belch("{createThread}Daq ghuH: refusing to create another thread; no more than %d threads allowed (currently %d)",
1010           RtsFlags.ParFlags.maxThreads, advisory_thread_count);
1011     return END_TSO_QUEUE;
1012   }
1013   threadsCreated++;
1014 #endif
1015
1016 #if defined(GRAN)
1017   ASSERT(!RtsFlags.GranFlags.Light || CurrentProc==0);
1018 #endif
1019
1020   // ToDo: check whether size = stack_size - TSO_STRUCT_SIZEW
1021
1022   /* catch ridiculously small stack sizes */
1023   if (size < MIN_STACK_WORDS + TSO_STRUCT_SIZEW) {
1024     size = MIN_STACK_WORDS + TSO_STRUCT_SIZEW;
1025   }
1026
1027   stack_size = size - TSO_STRUCT_SIZEW;
1028
1029   tso = (StgTSO *)allocate(size);
1030   TICK_ALLOC_TSO(size-TSO_STRUCT_SIZEW, 0);
1031
1032   SET_HDR(tso, &TSO_info, CCS_MAIN);
1033 #if defined(GRAN)
1034   SET_GRAN_HDR(tso, ThisPE);
1035 #endif
1036   tso->what_next     = ThreadEnterGHC;
1037
1038   /* tso->id needs to be unique.  For now we use a heavyweight mutex to
1039    * protect the increment operation on next_thread_id.
1040    * In future, we could use an atomic increment instead.
1041    */
1042   if (!have_lock) { ACQUIRE_LOCK(&sched_mutex); }
1043   tso->id = next_thread_id++;
1044   if (!have_lock) { RELEASE_LOCK(&sched_mutex); }
1045
1046   tso->why_blocked  = NotBlocked;
1047   tso->blocked_exceptions = NULL;
1048
1049   tso->splim        = (P_)&(tso->stack) + RESERVED_STACK_WORDS;
1050   tso->stack_size   = stack_size;
1051   tso->max_stack_size = round_to_mblocks(RtsFlags.GcFlags.maxStkSize)
1052                               - TSO_STRUCT_SIZEW;
1053   tso->sp           = (P_)&(tso->stack) + stack_size;
1054
1055 #ifdef PROFILING
1056   tso->prof.CCCS = CCS_MAIN;
1057 #endif
1058
1059   /* put a stop frame on the stack */
1060   tso->sp -= sizeofW(StgStopFrame);
1061   SET_HDR((StgClosure*)tso->sp,(StgInfoTable *)&stg_stop_thread_info,CCS_MAIN);
1062   tso->su = (StgUpdateFrame*)tso->sp;
1063
1064   IF_DEBUG(scheduler,belch("---- Initialised TSO %ld (%p), stack size = %lx words",
1065                            tso->id, tso, tso->stack_size));
1066
1067   // ToDo: check this
1068 #if defined(GRAN)
1069   tso->link = END_TSO_QUEUE;
1070   /* uses more flexible routine in GranSim */
1071   insertThread(tso, CurrentProc);
1072 #else
1073   /* In a non-GranSim setup the pushing of a TSO onto the runq is separated
1074    * from its creation
1075    */
1076 #endif
1077
1078   /* Link the new thread on the global thread list.
1079    */
1080 #if defined(GRAN)
1081 #error ToDo
1082 #else
1083   tso->global_link = all_threads;
1084   all_threads = tso;
1085 #endif
1086
1087 #if defined(GRAN)
1088   tso->gran.pri = pri;
1089   tso->gran.magic = TSO_MAGIC; // debugging only
1090   tso->gran.sparkname   = 0;
1091   tso->gran.startedat   = CURRENT_TIME;
1092   tso->gran.exported    = 0;
1093   tso->gran.basicblocks = 0;
1094   tso->gran.allocs      = 0;
1095   tso->gran.exectime    = 0;
1096   tso->gran.fetchtime   = 0;
1097   tso->gran.fetchcount  = 0;
1098   tso->gran.blocktime   = 0;
1099   tso->gran.blockcount  = 0;
1100   tso->gran.blockedat   = 0;
1101   tso->gran.globalsparks = 0;
1102   tso->gran.localsparks  = 0;
1103   if (RtsFlags.GranFlags.Light)
1104     tso->gran.clock  = Now; /* local clock */
1105   else
1106     tso->gran.clock  = 0;
1107
1108   IF_DEBUG(gran,printTSO(tso));
1109 #elif defined(PAR)
1110   tso->par.sparkname   = 0;
1111   tso->par.startedat   = CURRENT_TIME;
1112   tso->par.exported    = 0;
1113   tso->par.basicblocks = 0;
1114   tso->par.allocs      = 0;
1115   tso->par.exectime    = 0;
1116   tso->par.fetchtime   = 0;
1117   tso->par.fetchcount  = 0;
1118   tso->par.blocktime   = 0;
1119   tso->par.blockcount  = 0;
1120   tso->par.blockedat   = 0;
1121   tso->par.globalsparks = 0;
1122   tso->par.localsparks  = 0;
1123 #endif
1124
1125 #if defined(GRAN)
1126   globalGranStats.tot_threads_created++;
1127   globalGranStats.threads_created_on_PE[CurrentProc]++;
1128   globalGranStats.tot_sq_len += spark_queue_len(CurrentProc);
1129   globalGranStats.tot_sq_probes++;
1130 #endif
1131
1132   IF_DEBUG(scheduler,sched_belch("created thread %ld, stack size = %lx words",
1133                                  tso->id, tso->stack_size));
1134   return tso;
1135 }
1136
1137 /* ---------------------------------------------------------------------------
1138  * scheduleThread()
1139  *
1140  * scheduleThread puts a thread on the head of the runnable queue.
1141  * This will usually be done immediately after a thread is created.
1142  * The caller of scheduleThread must create the thread using e.g.
1143  * createThread and push an appropriate closure
1144  * on this thread's stack before the scheduler is invoked.
1145  * ------------------------------------------------------------------------ */
1146
1147 void
1148 scheduleThread(StgTSO *tso)
1149 {
1150   ACQUIRE_LOCK(&sched_mutex);
1151
1152   /* Put the new thread on the head of the runnable queue.  The caller
1153    * better push an appropriate closure on this thread's stack
1154    * beforehand.  In the SMP case, the thread may start running as
1155    * soon as we release the scheduler lock below.
1156    */
1157   PUSH_ON_RUN_QUEUE(tso);
1158   THREAD_RUNNABLE();
1159
1160   IF_DEBUG(scheduler,printTSO(tso));
1161   RELEASE_LOCK(&sched_mutex);
1162 }
1163
1164 /* ---------------------------------------------------------------------------
1165  * startTasks()
1166  *
1167  * Start up Posix threads to run each of the scheduler tasks.
1168  * I believe the task ids are not needed in the system as defined.
1169  *  KH @ 25/10/99
1170  * ------------------------------------------------------------------------ */
1171
1172 #ifdef SMP
1173 static void *
1174 taskStart( void *arg STG_UNUSED )
1175 {
1176   schedule();
1177   return NULL;
1178 }
1179 #endif
1180
1181 /* ---------------------------------------------------------------------------
1182  * initScheduler()
1183  *
1184  * Initialise the scheduler.  This resets all the queues - if the
1185  * queues contained any threads, they'll be garbage collected at the
1186  * next pass.
1187  *
1188  * This now calls startTasks(), so should only be called once!  KH @ 25/10/99
1189  * ------------------------------------------------------------------------ */
1190
1191 #ifdef SMP
1192 static void
1193 term_handler(int sig STG_UNUSED)
1194 {
1195   stat_workerStop();
1196   ACQUIRE_LOCK(&term_mutex);
1197   await_death--;
1198   RELEASE_LOCK(&term_mutex);
1199   pthread_exit(NULL);
1200 }
1201 #endif
1202
1203 //@cindex initScheduler
1204 void
1205 initScheduler(void)
1206 {
1207 #if defined(GRAN)
1208   nat i;
1209
1210   for (i=0; i<=MAX_PROC; i++) {
1211     run_queue_hds[i]      = END_TSO_QUEUE;
1212     run_queue_tls[i]      = END_TSO_QUEUE;
1213     blocked_queue_hds[i]  = END_TSO_QUEUE;
1214     blocked_queue_tls[i]  = END_TSO_QUEUE;
1215     ccalling_threadss[i]  = END_TSO_QUEUE;
1216   }
1217 #else
1218   run_queue_hd      = END_TSO_QUEUE;
1219   run_queue_tl      = END_TSO_QUEUE;
1220   blocked_queue_hd  = END_TSO_QUEUE;
1221   blocked_queue_tl  = END_TSO_QUEUE;
1222 #endif
1223
1224   suspended_ccalling_threads  = END_TSO_QUEUE;
1225
1226   main_threads = NULL;
1227   all_threads  = END_TSO_QUEUE;
1228
1229   context_switch = 0;
1230   interrupted    = 0;
1231
1232   enteredCAFs = END_CAF_LIST;
1233
1234   /* Install the SIGHUP handler */
1235 #ifdef SMP
1236   {
1237     struct sigaction action,oact;
1238
1239     action.sa_handler = term_handler;
1240     sigemptyset(&action.sa_mask);
1241     action.sa_flags = 0;
1242     if (sigaction(SIGTERM, &action, &oact) != 0) {
1243       barf("can't install TERM handler");
1244     }
1245   }
1246 #endif
1247
1248 #ifdef SMP
1249   /* Allocate N Capabilities */
1250   {
1251     nat i;
1252     Capability *cap, *prev;
1253     cap  = NULL;
1254     prev = NULL;
1255     for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
1256       cap = stgMallocBytes(sizeof(Capability), "initScheduler:capabilities");
1257       cap->link = prev;
1258       prev = cap;
1259     }
1260     free_capabilities = cap;
1261     n_free_capabilities = RtsFlags.ParFlags.nNodes;
1262   }
1263   IF_DEBUG(scheduler,fprintf(stderr,"scheduler: Allocated %d capabilities\n",
1264                              n_free_capabilities););
1265 #endif
1266
1267 #if defined(SMP) || defined(PAR)
1268   initSparkPools();
1269 #endif
1270 }
1271
1272 #ifdef SMP
1273 void
1274 startTasks( void )
1275 {
1276   nat i;
1277   int r;
1278   pthread_t tid;
1279
1280   /* make some space for saving all the thread ids */
1281   task_ids = stgMallocBytes(RtsFlags.ParFlags.nNodes * sizeof(task_info),
1282                             "initScheduler:task_ids");
1283
1284   /* and create all the threads */
1285   for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
1286     r = pthread_create(&tid,NULL,taskStart,NULL);
1287     if (r != 0) {
1288       barf("startTasks: Can't create new Posix thread");
1289     }
1290     task_ids[i].id = tid;
1291     task_ids[i].mut_time = 0.0;
1292     task_ids[i].mut_etime = 0.0;
1293     task_ids[i].gc_time = 0.0;
1294     task_ids[i].gc_etime = 0.0;
1295     task_ids[i].elapsedtimestart = elapsedtime();
1296     IF_DEBUG(scheduler,fprintf(stderr,"scheduler: Started task: %ld\n",tid););
1297   }
1298 }
1299 #endif
1300
1301 void
1302 exitScheduler( void )
1303 {
1304 #ifdef SMP
1305   nat i;
1306
1307   /* Don't want to use pthread_cancel, since we'd have to install
1308    * these silly exception handlers (pthread_cleanup_{push,pop}) around
1309    * all our locks.
1310    */
1311 #if 0
1312   /* Cancel all our tasks */
1313   for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
1314     pthread_cancel(task_ids[i].id);
1315   }
1316
1317   /* Wait for all the tasks to terminate */
1318   for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
1319     IF_DEBUG(scheduler,fprintf(stderr,"scheduler: waiting for task %ld\n",
1320                                task_ids[i].id));
1321     pthread_join(task_ids[i].id, NULL);
1322   }
1323 #endif
1324
1325   /* Send 'em all a SIGHUP.  That should shut 'em up.
1326    */
1327   await_death = RtsFlags.ParFlags.nNodes;
1328   for (i = 0; i < RtsFlags.ParFlags.nNodes; i++) {
1329     pthread_kill(task_ids[i].id,SIGTERM);
1330   }
1331   while (await_death > 0) {
1332     sched_yield();
1333   }
1334 #endif
1335 }
1336
1337 /* -----------------------------------------------------------------------------
1338    Managing the per-task allocation areas.
1339
1340    Each capability comes with an allocation area.  These are
1341    fixed-length block lists into which allocation can be done.
1342
1343    ToDo: no support for two-space collection at the moment???
1344    -------------------------------------------------------------------------- */
1345
1346 /* -----------------------------------------------------------------------------
1347  * waitThread is the external interface for running a new computataion
1348  * and waiting for the result.
1349  *
1350  * In the non-SMP case, we create a new main thread, push it on the
1351  * main-thread stack, and invoke the scheduler to run it.  The
1352  * scheduler will return when the top main thread on the stack has
1353  * completed or died, and fill in the necessary fields of the
1354  * main_thread structure.
1355  *
1356  * In the SMP case, we create a main thread as before, but we then
1357  * create a new condition variable and sleep on it.  When our new
1358  * main thread has completed, we'll be woken up and the status/result
1359  * will be in the main_thread struct.
1360  * -------------------------------------------------------------------------- */
1361
1362 SchedulerStatus
1363 waitThread(StgTSO *tso, /*out*/StgClosure **ret)
1364 {
1365   StgMainThread *m;
1366   SchedulerStatus stat;
1367
1368   ACQUIRE_LOCK(&sched_mutex);
1369
1370   m = stgMallocBytes(sizeof(StgMainThread), "waitThread");
1371
1372   m->tso = tso;
1373   m->ret = ret;
1374   m->stat = NoStatus;
1375 #ifdef SMP
1376   pthread_cond_init(&m->wakeup, NULL);
1377 #endif
1378
1379   m->link = main_threads;
1380   main_threads = m;
1381
1382   IF_DEBUG(scheduler, fprintf(stderr, "scheduler: new main thread (%d)\n",
1383                               m->tso->id));
1384
1385 #ifdef SMP
1386   do {
1387     pthread_cond_wait(&m->wakeup, &sched_mutex);
1388   } while (m->stat == NoStatus);
1389 #else
1390   schedule();
1391   ASSERT(m->stat != NoStatus);
1392 #endif
1393
1394   stat = m->stat;
1395
1396 #ifdef SMP
1397   pthread_cond_destroy(&m->wakeup);
1398 #endif
1399
1400   IF_DEBUG(scheduler, fprintf(stderr, "scheduler: main thread (%d) finished\n",
1401                               m->tso->id));
1402   free(m);
1403
1404   RELEASE_LOCK(&sched_mutex);
1405
1406   return stat;
1407 }
1408
1409 //@node Run queue code, Garbage Collextion Routines, Suspend and Resume, Main scheduling code
1410 //@subsection Run queue code
1411
1412 #if 0
1413 /*
1414    NB: In GranSim we have many run queues; run_queue_hd is actually a macro
1415        unfolding to run_queue_hds[CurrentProc], thus CurrentProc is an
1416        implicit global variable that has to be correct when calling these
1417        fcts -- HWL
1418 */
1419
1420 /* Put the new thread on the head of the runnable queue.
1421  * The caller of createThread better push an appropriate closure
1422  * on this thread's stack before the scheduler is invoked.
1423  */
1424 static /* inline */ void
1425 add_to_run_queue(tso)
1426 StgTSO* tso;
1427 {
1428   ASSERT(tso!=run_queue_hd && tso!=run_queue_tl);
1429   tso->link = run_queue_hd;
1430   run_queue_hd = tso;
1431   if (run_queue_tl == END_TSO_QUEUE) {
1432     run_queue_tl = tso;
1433   }
1434 }
1435
1436 /* Put the new thread at the end of the runnable queue. */
1437 static /* inline */ void
1438 push_on_run_queue(tso)
1439 StgTSO* tso;
1440 {
1441   ASSERT(get_itbl((StgClosure *)tso)->type == TSO);
1442   ASSERT(run_queue_hd!=NULL && run_queue_tl!=NULL);
1443   ASSERT(tso!=run_queue_hd && tso!=run_queue_tl);
1444   if (run_queue_hd == END_TSO_QUEUE) {
1445     run_queue_hd = tso;
1446   } else {
1447     run_queue_tl->link = tso;
1448   }
1449   run_queue_tl = tso;
1450 }
1451
1452 /*
1453    Should be inlined because it's used very often in schedule.  The tso
1454    argument is actually only needed in GranSim, where we want to have the
1455    possibility to schedule *any* TSO on the run queue, irrespective of the
1456    actual ordering. Therefore, if tso is not the nil TSO then we traverse
1457    the run queue and dequeue the tso, adjusting the links in the queue.
1458 */
1459 //@cindex take_off_run_queue
1460 static /* inline */ StgTSO*
1461 take_off_run_queue(StgTSO *tso) {
1462   StgTSO *t, *prev;
1463
1464   /*
1465      qetlaHbogh Qu' ngaSbogh ghomDaQ {tso} yIteq!
1466
1467      if tso is specified, unlink that tso from the run_queue (doesn't have
1468      to be at the beginning of the queue); GranSim only
1469   */
1470   if (tso!=END_TSO_QUEUE) {
1471     /* find tso in queue */
1472     for (t=run_queue_hd, prev=END_TSO_QUEUE;
1473          t!=END_TSO_QUEUE && t!=tso;
1474          prev=t, t=t->link)
1475       /* nothing */ ;
1476     ASSERT(t==tso);
1477     /* now actually dequeue the tso */
1478     if (prev!=END_TSO_QUEUE) {
1479       ASSERT(run_queue_hd!=t);
1480       prev->link = t->link;
1481     } else {
1482       /* t is at beginning of thread queue */
1483       ASSERT(run_queue_hd==t);
1484       run_queue_hd = t->link;
1485     }
1486     /* t is at end of thread queue */
1487     if (t->link==END_TSO_QUEUE) {
1488       ASSERT(t==run_queue_tl);
1489       run_queue_tl = prev;
1490     } else {
1491       ASSERT(run_queue_tl!=t);
1492     }
1493     t->link = END_TSO_QUEUE;
1494   } else {
1495     /* take tso from the beginning of the queue; std concurrent code */
1496     t = run_queue_hd;
1497     if (t != END_TSO_QUEUE) {
1498       run_queue_hd = t->link;
1499       t->link = END_TSO_QUEUE;
1500       if (run_queue_hd == END_TSO_QUEUE) {
1501         run_queue_tl = END_TSO_QUEUE;
1502       }
1503     }
1504   }
1505   return t;
1506 }
1507
1508 #endif /* 0 */
1509
1510 //@node Garbage Collextion Routines, Blocking Queue Routines, Run queue code, Main scheduling code
1511 //@subsection Garbage Collextion Routines
1512
1513 /* ---------------------------------------------------------------------------
1514    Where are the roots that we know about?
1515
1516         - all the threads on the runnable queue
1517         - all the threads on the blocked queue
1518         - all the thread currently executing a _ccall_GC
1519         - all the "main threads"
1520
1521    ------------------------------------------------------------------------ */
1522
1523 /* This has to be protected either by the scheduler monitor, or by the
1524         garbage collection monitor (probably the latter).
1525         KH @ 25/10/99
1526 */
1527
1528 static void GetRoots(void)
1529 {
1530   StgMainThread *m;
1531
1532 #if defined(GRAN)
1533   {
1534     nat i;
1535     for (i=0; i<=RtsFlags.GranFlags.proc; i++) {
1536       if ((run_queue_hds[i] != END_TSO_QUEUE) && ((run_queue_hds[i] != NULL)))
1537         run_queue_hds[i]    = (StgTSO *)MarkRoot((StgClosure *)run_queue_hds[i]);
1538       if ((run_queue_tls[i] != END_TSO_QUEUE) && ((run_queue_tls[i] != NULL)))
1539         run_queue_tls[i]    = (StgTSO *)MarkRoot((StgClosure *)run_queue_tls[i]);
1540
1541       if ((blocked_queue_hds[i] != END_TSO_QUEUE) && ((blocked_queue_hds[i] != NULL)))
1542         blocked_queue_hds[i] = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_hds[i]);
1543       if ((blocked_queue_tls[i] != END_TSO_QUEUE) && ((blocked_queue_tls[i] != NULL)))
1544         blocked_queue_tls[i] = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_tls[i]);
1545       if ((ccalling_threadss[i] != END_TSO_QUEUE) && ((ccalling_threadss[i] != NULL)))
1546         ccalling_threadss[i] = (StgTSO *)MarkRoot((StgClosure *)ccalling_threadss[i]);
1547     }
1548   }
1549
1550   markEventQueue();
1551
1552 #else /* !GRAN */
1553   run_queue_hd      = (StgTSO *)MarkRoot((StgClosure *)run_queue_hd);
1554   run_queue_tl      = (StgTSO *)MarkRoot((StgClosure *)run_queue_tl);
1555
1556   blocked_queue_hd  = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_hd);
1557   blocked_queue_tl  = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_tl);
1558 #endif
1559
1560   for (m = main_threads; m != NULL; m = m->link) {
1561     m->tso = (StgTSO *)MarkRoot((StgClosure *)m->tso);
1562   }
1563   suspended_ccalling_threads =
1564     (StgTSO *)MarkRoot((StgClosure *)suspended_ccalling_threads);
1565
1566 #if defined(SMP) || defined(PAR) || defined(GRAN)
1567   markSparkQueue();
1568 #endif
1569 }
1570
1571 /* -----------------------------------------------------------------------------
1572    performGC
1573
1574    This is the interface to the garbage collector from Haskell land.
1575    We provide this so that external C code can allocate and garbage
1576    collect when called from Haskell via _ccall_GC.
1577
1578    It might be useful to provide an interface whereby the programmer
1579    can specify more roots (ToDo).
1580
1581    This needs to be protected by the GC condition variable above.  KH.
1582    -------------------------------------------------------------------------- */
1583
1584 void (*extra_roots)(void);
1585
1586 void
1587 performGC(void)
1588 {
1589   GarbageCollect(GetRoots);
1590 }
1591
1592 static void
1593 AllRoots(void)
1594 {
1595   GetRoots();                   /* the scheduler's roots */
1596   extra_roots();                /* the user's roots */
1597 }
1598
1599 void
1600 performGCWithRoots(void (*get_roots)(void))
1601 {
1602   extra_roots = get_roots;
1603
1604   GarbageCollect(AllRoots);
1605 }
1606
1607 /* -----------------------------------------------------------------------------
1608    Stack overflow
1609
1610    If the thread has reached its maximum stack size, then raise the
1611    StackOverflow exception in the offending thread.  Otherwise
1612    relocate the TSO into a larger chunk of memory and adjust its stack
1613    size appropriately.
1614    -------------------------------------------------------------------------- */
1615
1616 static StgTSO *
1617 threadStackOverflow(StgTSO *tso)
1618 {
1619   nat new_stack_size, new_tso_size, diff, stack_words;
1620   StgPtr new_sp;
1621   StgTSO *dest;
1622
1623   IF_DEBUG(sanity,checkTSO(tso));
1624   if (tso->stack_size >= tso->max_stack_size) {
1625 #if 0
1626     /* If we're debugging, just print out the top of the stack */
1627     printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size,
1628                                      tso->sp+64));
1629 #endif
1630 #ifdef INTERPRETER
1631     fprintf(stderr, "fatal: stack overflow in Hugs; aborting\n" );
1632     exit(1);
1633 #else
1634     /* Send this thread the StackOverflow exception */
1635     raiseAsync(tso, (StgClosure *)stackOverflow_closure);
1636 #endif
1637     return tso;
1638   }
1639
1640   /* Try to double the current stack size.  If that takes us over the
1641    * maximum stack size for this thread, then use the maximum instead.
1642    * Finally round up so the TSO ends up as a whole number of blocks.
1643    */
1644   new_stack_size = stg_min(tso->stack_size * 2, tso->max_stack_size);
1645   new_tso_size   = (nat)BLOCK_ROUND_UP(new_stack_size * sizeof(W_) +
1646                                        TSO_STRUCT_SIZE)/sizeof(W_);
1647   new_tso_size = round_to_mblocks(new_tso_size);  /* Be MBLOCK-friendly */
1648   new_stack_size = new_tso_size - TSO_STRUCT_SIZEW;
1649
1650   IF_DEBUG(scheduler, fprintf(stderr,"scheduler: increasing stack size from %d words to %d.\n", tso->stack_size, new_stack_size));
1651
1652   dest = (StgTSO *)allocate(new_tso_size);
1653   TICK_ALLOC_TSO(new_tso_size-sizeofW(StgTSO),0);
1654
1655   /* copy the TSO block and the old stack into the new area */
1656   memcpy(dest,tso,TSO_STRUCT_SIZE);
1657   stack_words = tso->stack + tso->stack_size - tso->sp;
1658   new_sp = (P_)dest + new_tso_size - stack_words;
1659   memcpy(new_sp, tso->sp, stack_words * sizeof(W_));
1660
1661   /* relocate the stack pointers... */
1662   diff = (P_)new_sp - (P_)tso->sp; /* In *words* */
1663   dest->su    = (StgUpdateFrame *) ((P_)dest->su + diff);
1664   dest->sp    = new_sp;
1665   dest->splim = (P_)dest->splim + (nat)((P_)dest - (P_)tso);
1666   dest->stack_size = new_stack_size;
1667
1668   /* and relocate the update frame list */
1669   relocate_TSO(tso, dest);
1670
1671   /* Mark the old TSO as relocated.  We have to check for relocated
1672    * TSOs in the garbage collector and any primops that deal with TSOs.
1673    *
1674    * It's important to set the sp and su values to just beyond the end
1675    * of the stack, so we don't attempt to scavenge any part of the
1676    * dead TSO's stack.
1677    */
1678   tso->what_next = ThreadRelocated;
1679   tso->link = dest;
1680   tso->sp = (P_)&(tso->stack[tso->stack_size]);
1681   tso->su = (StgUpdateFrame *)tso->sp;
1682   tso->why_blocked = NotBlocked;
1683   dest->mut_link = NULL;
1684
1685   IF_DEBUG(sanity,checkTSO(tso));
1686 #if 0
1687   IF_DEBUG(scheduler,printTSO(dest));
1688 #endif
1689
1690   return dest;
1691 }
1692
1693 //@node Blocking Queue Routines, Exception Handling Routines, Garbage Collextion Routines, Main scheduling code
1694 //@subsection Blocking Queue Routines
1695
1696 /* ---------------------------------------------------------------------------
1697    Wake up a queue that was blocked on some resource.
1698    ------------------------------------------------------------------------ */
1699
1700 /* ToDo: check push_on_run_queue vs. PUSH_ON_RUN_QUEUE */
1701
1702 #if defined(GRAN)
1703 static inline void
1704 unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node )
1705 {
1706 }
1707 #elif defined(PAR)
1708 static inline void
1709 unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node )
1710 {
1711   /* write RESUME events to log file and
1712      update blocked and fetch time (depending on type of the orig closure) */
1713   if (RtsFlags.ParFlags.ParStats.Full) {
1714     DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
1715                      GR_RESUMEQ, ((StgTSO *)bqe), ((StgTSO *)bqe)->block_info.closure,
1716                      0, 0 /* spark_queue_len(ADVISORY_POOL) */);
1717
1718     switch (get_itbl(node)->type) {
1719         case FETCH_ME_BQ:
1720           ((StgTSO *)bqe)->par.fetchtime += CURRENT_TIME-((StgTSO *)bqe)->par.blockedat;
1721           break;
1722         case RBH:
1723         case FETCH_ME:
1724         case BLACKHOLE_BQ:
1725           ((StgTSO *)bqe)->par.blocktime += CURRENT_TIME-((StgTSO *)bqe)->par.blockedat;
1726           break;
1727         default:
1728           barf("{unblockOneLocked}Daq Qagh: unexpected closure in blocking queue");
1729         }
1730       }
1731 }
1732 #endif
1733
1734 #if defined(GRAN)
1735 static StgBlockingQueueElement *
1736 unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node)
1737 {
1738     StgBlockingQueueElement *next;
1739     PEs node_loc, tso_loc;
1740
1741     node_loc = where_is(node); // should be lifted out of loop
1742     tso = (StgTSO *)bqe;  // wastes an assignment to get the type right
1743     tso_loc = where_is(tso);
1744     if (IS_LOCAL_TO(PROCS(node),tso_loc)) { // TSO is local
1745       /* !fake_fetch => TSO is on CurrentProc is same as IS_LOCAL_TO */
1746       ASSERT(CurrentProc!=node_loc || tso_loc==CurrentProc);
1747       bq_processing_time += RtsFlags.GranFlags.Costs.lunblocktime;
1748       // insertThread(tso, node_loc);
1749       new_event(tso_loc, tso_loc,
1750                 CurrentTime[CurrentProc]+bq_processing_time,
1751                 ResumeThread,
1752                 tso, node, (rtsSpark*)NULL);
1753       tso->link = END_TSO_QUEUE; // overwrite link just to be sure
1754       // len_local++;
1755       // len++;
1756     } else { // TSO is remote (actually should be FMBQ)
1757       bq_processing_time += RtsFlags.GranFlags.Costs.mpacktime;
1758       bq_processing_time += RtsFlags.GranFlags.Costs.gunblocktime;
1759       new_event(tso_loc, CurrentProc,
1760                 CurrentTime[CurrentProc]+bq_processing_time+
1761                 RtsFlags.GranFlags.Costs.latency,
1762                 UnblockThread,
1763                 tso, node, (rtsSpark*)NULL);
1764       tso->link = END_TSO_QUEUE; // overwrite link just to be sure
1765       bq_processing_time += RtsFlags.GranFlags.Costs.mtidytime;
1766       // len++;
1767     }
1768     /* the thread-queue-overhead is accounted for in either Resume or UnblockThread */
1769     IF_GRAN_DEBUG(bq,
1770                   fprintf(stderr," %s TSO %d (%p) [PE %d] (blocked_on=%p) (next=%p) ,",
1771                           (node_loc==tso_loc ? "Local" : "Global"),
1772                           tso->id, tso, CurrentProc, tso->blocked_on, tso->link))
1773     tso->blocked_on = NULL;
1774     IF_DEBUG(scheduler,belch("-- Waking up thread %ld (%p)",
1775                              tso->id, tso));
1776   }
1777
1778   /* if this is the BQ of an RBH, we have to put back the info ripped out of
1779      the closure to make room for the anchor of the BQ */
1780   if (next!=END_BQ_QUEUE) {
1781     ASSERT(get_itbl(node)->type == RBH && get_itbl(next)->type == CONSTR);
1782     /*
1783     ASSERT((info_ptr==&RBH_Save_0_info) ||
1784            (info_ptr==&RBH_Save_1_info) ||
1785            (info_ptr==&RBH_Save_2_info));
1786     */
1787     /* cf. convertToRBH in RBH.c for writing the RBHSave closure */
1788     ((StgRBH *)node)->blocking_queue = ((StgRBHSave *)next)->payload[0];
1789     ((StgRBH *)node)->mut_link       = ((StgRBHSave *)next)->payload[1];
1790
1791     IF_GRAN_DEBUG(bq,
1792                   belch("## Filled in RBH_Save for %p (%s) at end of AwBQ",
1793                         node, info_type(node)));
1794   }
1795 }
1796 #elif defined(PAR)
1797 static StgBlockingQueueElement *
1798 unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node)
1799 {
1800     StgBlockingQueueElement *next;
1801
1802     switch (get_itbl(bqe)->type) {
1803     case TSO:
1804       ASSERT(((StgTSO *)bqe)->why_blocked != NotBlocked);
1805       /* if it's a TSO just push it onto the run_queue */
1806       next = bqe->link;
1807       // ((StgTSO *)bqe)->link = END_TSO_QUEUE; // debugging?
1808       PUSH_ON_RUN_QUEUE((StgTSO *)bqe);
1809       THREAD_RUNNABLE();
1810       unblockCount(bqe, node);
1811       /* reset blocking status after dumping event */
1812       ((StgTSO *)bqe)->why_blocked = NotBlocked;
1813       break;
1814
1815     case BLOCKED_FETCH:
1816       /* if it's a BLOCKED_FETCH put it on the PendingFetches list */
1817       next = bqe->link;
1818       bqe->link = PendingFetches;
1819       PendingFetches = bqe;
1820       break;
1821
1822 # if defined(DEBUG)
1823       /* can ignore this case in a non-debugging setup;
1824          see comments on RBHSave closures above */
1825     case CONSTR:
1826       /* check that the closure is an RBHSave closure */
1827       ASSERT(get_itbl((StgClosure *)bqe) == &RBH_Save_0_info ||
1828              get_itbl((StgClosure *)bqe) == &RBH_Save_1_info ||
1829              get_itbl((StgClosure *)bqe) == &RBH_Save_2_info);
1830       break;
1831
1832     default:
1833       barf("{unblockOneLocked}Daq Qagh: Unexpected IP (%#lx; %s) in blocking queue at %#lx\n",
1834            get_itbl((StgClosure *)bqe), info_type((StgClosure *)bqe),
1835            (StgClosure *)bqe);
1836 # endif
1837     }
1838   // IF_DEBUG(scheduler,sched_belch("waking up thread %ld", tso->id));
1839   return next;
1840 }
1841
1842 #else /* !GRAN && !PAR */
1843 static StgTSO *
1844 unblockOneLocked(StgTSO *tso)
1845 {
1846   StgTSO *next;
1847
1848   ASSERT(get_itbl(tso)->type == TSO);
1849   ASSERT(tso->why_blocked != NotBlocked);
1850   tso->why_blocked = NotBlocked;
1851   next = tso->link;
1852   PUSH_ON_RUN_QUEUE(tso);
1853   THREAD_RUNNABLE();
1854   IF_DEBUG(scheduler,sched_belch("waking up thread %ld", tso->id));
1855   return next;
1856 }
1857 #endif
1858
1859 #if defined(PAR) || defined(GRAN)
1860 inline StgTSO *
1861 unblockOne(StgTSO *tso, StgClosure *node)
1862 {
1863   ACQUIRE_LOCK(&sched_mutex);
1864   tso = unblockOneLocked(tso, node);
1865   RELEASE_LOCK(&sched_mutex);
1866   return tso;
1867 }
1868 #else
1869 inline StgTSO *
1870 unblockOne(StgTSO *tso)
1871 {
1872   ACQUIRE_LOCK(&sched_mutex);
1873   tso = unblockOneLocked(tso);
1874   RELEASE_LOCK(&sched_mutex);
1875   return tso;
1876 }
1877 #endif
1878
1879 #if defined(GRAN)
1880 void
1881 awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node)
1882 {
1883   StgBlockingQueueElement *bqe, *next;
1884   StgTSO *tso;
1885   PEs node_loc, tso_loc;
1886   rtsTime bq_processing_time = 0;
1887   nat len = 0, len_local = 0;
1888
1889   IF_GRAN_DEBUG(bq,
1890                 belch("## AwBQ for node %p on PE %d @ %ld by TSO %d (%p): ", \
1891                       node, CurrentProc, CurrentTime[CurrentProc],
1892                       CurrentTSO->id, CurrentTSO));
1893
1894   node_loc = where_is(node);
1895
1896   ASSERT(get_itbl(q)->type == TSO ||   // q is either a TSO or an RBHSave
1897          get_itbl(q)->type == CONSTR); // closure (type constructor)
1898   ASSERT(is_unique(node));
1899
1900   /* FAKE FETCH: magically copy the node to the tso's proc;
1901      no Fetch necessary because in reality the node should not have been
1902      moved to the other PE in the first place
1903   */
1904   if (CurrentProc!=node_loc) {
1905     IF_GRAN_DEBUG(bq,
1906                   belch("## node %p is on PE %d but CurrentProc is %d (TSO %d); assuming fake fetch and adjusting bitmask (old: %#x)",
1907                         node, node_loc, CurrentProc, CurrentTSO->id,
1908                         // CurrentTSO, where_is(CurrentTSO),
1909                         node->header.gran.procs));
1910     node->header.gran.procs = (node->header.gran.procs) | PE_NUMBER(CurrentProc);
1911     IF_GRAN_DEBUG(bq,
1912                   belch("## new bitmask of node %p is %#x",
1913                         node, node->header.gran.procs));
1914     if (RtsFlags.GranFlags.GranSimStats.Global) {
1915       globalGranStats.tot_fake_fetches++;
1916     }
1917   }
1918
1919   bqe = q;
1920   // ToDo: check: ASSERT(CurrentProc==node_loc);
1921   while (get_itbl(bqe)->type==TSO) { // q != END_TSO_QUEUE) {
1922     //next = bqe->link;
1923     /*
1924        bqe points to the current element in the queue
1925        next points to the next element in the queue
1926     */
1927     //tso = (StgTSO *)bqe;  // wastes an assignment to get the type right
1928     //tso_loc = where_is(tso);
1929     bqe = unblockOneLocked(bqe, node);
1930   }
1931
1932   /* statistics gathering */
1933   /* ToDo: fix counters
1934   if (RtsFlags.GranFlags.GranSimStats.Global) {
1935     globalGranStats.tot_bq_processing_time += bq_processing_time;
1936     globalGranStats.tot_bq_len += len;      // total length of all bqs awakened
1937     globalGranStats.tot_bq_len_local += len_local;  // same for local TSOs only
1938     globalGranStats.tot_awbq++;             // total no. of bqs awakened
1939   }
1940   IF_GRAN_DEBUG(bq,
1941                 fprintf(stderr,"## BQ Stats of %p: [%d entries, %d local] %s\n",
1942                         node, len, len_local, (next!=END_TSO_QUEUE) ? "RBH" : ""));
1943   */
1944 }
1945 #elif defined(PAR)
1946 void
1947 awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node)
1948 {
1949   StgBlockingQueueElement *bqe, *next;
1950
1951   ACQUIRE_LOCK(&sched_mutex);
1952
1953   IF_PAR_DEBUG(verbose,
1954                belch("## AwBQ for node %p on [%x]: ",
1955                      node, mytid));
1956
1957   ASSERT(get_itbl(q)->type == TSO ||
1958          get_itbl(q)->type == BLOCKED_FETCH ||
1959          get_itbl(q)->type == CONSTR);
1960
1961   bqe = q;
1962   while (get_itbl(bqe)->type==TSO ||
1963          get_itbl(bqe)->type==BLOCKED_FETCH) {
1964     bqe = unblockOneLocked(bqe, node);
1965   }
1966   RELEASE_LOCK(&sched_mutex);
1967 }
1968
1969 #else   /* !GRAN && !PAR */
1970 void
1971 awakenBlockedQueue(StgTSO *tso)
1972 {
1973   ACQUIRE_LOCK(&sched_mutex);
1974   while (tso != END_TSO_QUEUE) {
1975     tso = unblockOneLocked(tso);
1976   }
1977   RELEASE_LOCK(&sched_mutex);
1978 }
1979 #endif
1980
1981 //@node Exception Handling Routines, Debugging Routines, Blocking Queue Routines, Main scheduling code
1982 //@subsection Exception Handling Routines
1983
1984 /* ---------------------------------------------------------------------------
1985    Interrupt execution
1986    - usually called inside a signal handler so it mustn't do anything fancy.
1987    ------------------------------------------------------------------------ */
1988
1989 void
1990 interruptStgRts(void)
1991 {
1992     interrupted    = 1;
1993     context_switch = 1;
1994 }
1995
1996 /* -----------------------------------------------------------------------------
1997    Unblock a thread
1998
1999    This is for use when we raise an exception in another thread, which
2000    may be blocked.
2001    This has nothing to do with the UnblockThread event in GranSim. -- HWL
2002    -------------------------------------------------------------------------- */
2003
2004 static void
2005 unblockThread(StgTSO *tso)
2006 {
2007   StgTSO *t, **last;
2008
2009   ACQUIRE_LOCK(&sched_mutex);
2010   switch (tso->why_blocked) {
2011
2012   case NotBlocked:
2013     return;  /* not blocked */
2014
2015   case BlockedOnMVar:
2016     ASSERT(get_itbl(tso->block_info.closure)->type == MVAR);
2017     {
2018       StgTSO *last_tso = END_TSO_QUEUE;
2019       StgMVar *mvar = (StgMVar *)(tso->block_info.closure);
2020
2021       last = &mvar->head;
2022       for (t = mvar->head; t != END_TSO_QUEUE;
2023            last = &t->link, last_tso = t, t = t->link) {
2024         if (t == tso) {
2025           *last = tso->link;
2026           if (mvar->tail == tso) {
2027             mvar->tail = last_tso;
2028           }
2029           goto done;
2030         }
2031       }
2032       barf("unblockThread (MVAR): TSO not found");
2033     }
2034
2035   case BlockedOnBlackHole:
2036     ASSERT(get_itbl(tso->block_info.closure)->type == BLACKHOLE_BQ);
2037     {
2038       StgBlockingQueue *bq = (StgBlockingQueue *)(tso->block_info.closure);
2039
2040       last = &bq->blocking_queue;
2041       for (t = bq->blocking_queue; t != END_TSO_QUEUE;
2042            last = &t->link, t = t->link) {
2043         if (t == tso) {
2044           *last = tso->link;
2045           goto done;
2046         }
2047       }
2048       barf("unblockThread (BLACKHOLE): TSO not found");
2049     }
2050
2051   case BlockedOnException:
2052     {
2053       StgTSO *target  = tso->block_info.tso;
2054
2055       ASSERT(get_itbl(target)->type == TSO);
2056       ASSERT(target->blocked_exceptions != NULL);
2057
2058       last = &target->blocked_exceptions;
2059       for (t = target->blocked_exceptions; t != END_TSO_QUEUE;
2060            last = &t->link, t = t->link) {
2061         ASSERT(get_itbl(t)->type == TSO);
2062         if (t == tso) {
2063           *last = tso->link;
2064           goto done;
2065         }
2066       }
2067       barf("unblockThread (Exception): TSO not found");
2068     }
2069
2070   case BlockedOnDelay:
2071   case BlockedOnRead:
2072   case BlockedOnWrite:
2073     {
2074       StgTSO *prev = NULL;
2075       for (t = blocked_queue_hd; t != END_TSO_QUEUE;
2076            prev = t, t = t->link) {
2077         if (t == tso) {
2078           if (prev == NULL) {
2079             blocked_queue_hd = t->link;
2080             if (blocked_queue_tl == t) {
2081               blocked_queue_tl = END_TSO_QUEUE;
2082             }
2083           } else {
2084             prev->link = t->link;
2085             if (blocked_queue_tl == t) {
2086               blocked_queue_tl = prev;
2087             }
2088           }
2089           goto done;
2090         }
2091       }
2092       barf("unblockThread (I/O): TSO not found");
2093     }
2094
2095   default:
2096     barf("unblockThread");
2097   }
2098
2099  done:
2100   tso->link = END_TSO_QUEUE;
2101   tso->why_blocked = NotBlocked;
2102   tso->block_info.closure = NULL;
2103   PUSH_ON_RUN_QUEUE(tso);
2104   RELEASE_LOCK(&sched_mutex);
2105 }
2106
2107 /* -----------------------------------------------------------------------------
2108  * raiseAsync()
2109  *
2110  * The following function implements the magic for raising an
2111  * asynchronous exception in an existing thread.
2112  *
2113  * We first remove the thread from any queue on which it might be
2114  * blocked.  The possible blockages are MVARs and BLACKHOLE_BQs.
2115  *
2116  * We strip the stack down to the innermost CATCH_FRAME, building
2117  * thunks in the heap for all the active computations, so they can
2118  * be restarted if necessary.  When we reach a CATCH_FRAME, we build
2119  * an application of the handler to the exception, and push it on
2120  * the top of the stack.
2121  *
2122  * How exactly do we save all the active computations?  We create an
2123  * AP_UPD for every UpdateFrame on the stack.  Entering one of these
2124  * AP_UPDs pushes everything from the corresponding update frame
2125  * upwards onto the stack.  (Actually, it pushes everything up to the
2126  * next update frame plus a pointer to the next AP_UPD object.
2127  * Entering the next AP_UPD object pushes more onto the stack until we
2128  * reach the last AP_UPD object - at which point the stack should look
2129  * exactly as it did when we killed the TSO and we can continue
2130  * execution by entering the closure on top of the stack.
2131  *
2132  * We can also kill a thread entirely - this happens if either (a) the
2133  * exception passed to raiseAsync is NULL, or (b) there's no
2134  * CATCH_FRAME on the stack.  In either case, we strip the entire
2135  * stack and replace the thread with a zombie.
2136  *
2137  * -------------------------------------------------------------------------- */
2138
2139 void
2140 deleteThread(StgTSO *tso)
2141 {
2142   raiseAsync(tso,NULL);
2143 }
2144
2145 void
2146 raiseAsync(StgTSO *tso, StgClosure *exception)
2147 {
2148   StgUpdateFrame* su = tso->su;
2149   StgPtr          sp = tso->sp;
2150
2151   /* Thread already dead? */
2152   if (tso->what_next == ThreadComplete || tso->what_next == ThreadKilled) {
2153     return;
2154   }
2155
2156   IF_DEBUG(scheduler, sched_belch("raising exception in thread %ld.", tso->id));
2157
2158   /* Remove it from any blocking queues */
2159   unblockThread(tso);
2160
2161   /* The stack freezing code assumes there's a closure pointer on
2162    * the top of the stack.  This isn't always the case with compiled
2163    * code, so we have to push a dummy closure on the top which just
2164    * returns to the next return address on the stack.
2165    */
2166   if ( LOOKS_LIKE_GHC_INFO((void*)*sp) ) {
2167     *(--sp) = (W_)&dummy_ret_closure;
2168   }
2169
2170   while (1) {
2171     int words = ((P_)su - (P_)sp) - 1;
2172     nat i;
2173     StgAP_UPD * ap;
2174
2175     /* If we find a CATCH_FRAME, and we've got an exception to raise,
2176      * then build PAP(handler,exception,realworld#), and leave it on
2177      * top of the stack ready to enter.
2178      */
2179     if (get_itbl(su)->type == CATCH_FRAME && exception != NULL) {
2180       StgCatchFrame *cf = (StgCatchFrame *)su;
2181       /* we've got an exception to raise, so let's pass it to the
2182        * handler in this frame.
2183        */
2184       ap = (StgAP_UPD *)allocate(sizeofW(StgPAP) + 2);
2185       TICK_ALLOC_UPD_PAP(3,0);
2186       SET_HDR(ap,&PAP_info,cf->header.prof.ccs);
2187
2188       ap->n_args = 2;
2189       ap->fun = cf->handler;    /* :: Exception -> IO a */
2190       ap->payload[0] = (P_)exception;
2191       ap->payload[1] = ARG_TAG(0); /* realworld token */
2192
2193       /* throw away the stack from Sp up to and including the
2194        * CATCH_FRAME.
2195        */
2196       sp = (P_)su + sizeofW(StgCatchFrame) - 1;
2197       tso->su = cf->link;
2198
2199       /* Restore the blocked/unblocked state for asynchronous exceptions
2200        * at the CATCH_FRAME.
2201        *
2202        * If exceptions were unblocked at the catch, arrange that they
2203        * are unblocked again after executing the handler by pushing an
2204        * unblockAsyncExceptions_ret stack frame.
2205        */
2206       if (!cf->exceptions_blocked) {
2207         *(sp--) = (W_)&unblockAsyncExceptionszh_ret_info;
2208       }
2209
2210       /* Ensure that async exceptions are blocked when running the handler.
2211        */
2212       if (tso->blocked_exceptions == NULL) {
2213         tso->blocked_exceptions = END_TSO_QUEUE;
2214       }
2215
2216       /* Put the newly-built PAP on top of the stack, ready to execute
2217        * when the thread restarts.
2218        */
2219       sp[0] = (W_)ap;
2220       tso->sp = sp;
2221       tso->what_next = ThreadEnterGHC;
2222       return;
2223     }
2224
2225     /* First build an AP_UPD consisting of the stack chunk above the
2226      * current update frame, with the top word on the stack as the
2227      * fun field.
2228      */
2229     ap = (StgAP_UPD *)allocate(AP_sizeW(words));
2230
2231     ASSERT(words >= 0);
2232
2233     ap->n_args = words;
2234     ap->fun    = (StgClosure *)sp[0];
2235     sp++;
2236     for(i=0; i < (nat)words; ++i) {
2237       ap->payload[i] = (P_)*sp++;
2238     }
2239
2240     switch (get_itbl(su)->type) {
2241
2242     case UPDATE_FRAME:
2243       {
2244         SET_HDR(ap,&AP_UPD_info,su->header.prof.ccs /* ToDo */);
2245         TICK_ALLOC_UP_THK(words+1,0);
2246
2247         IF_DEBUG(scheduler,
2248                  fprintf(stderr,  "scheduler: Updating ");
2249                  printPtr((P_)su->updatee);
2250                  fprintf(stderr,  " with ");
2251                  printObj((StgClosure *)ap);
2252                  );
2253
2254         /* Replace the updatee with an indirection - happily
2255          * this will also wake up any threads currently
2256          * waiting on the result.
2257          */
2258         UPD_IND_NOLOCK(su->updatee,ap);  /* revert the black hole */
2259         su = su->link;
2260         sp += sizeofW(StgUpdateFrame) -1;
2261         sp[0] = (W_)ap; /* push onto stack */
2262         break;
2263       }
2264
2265     case CATCH_FRAME:
2266       {
2267         StgCatchFrame *cf = (StgCatchFrame *)su;
2268         StgClosure* o;
2269
2270         /* We want a PAP, not an AP_UPD.  Fortunately, the
2271          * layout's the same.
2272          */
2273         SET_HDR(ap,&PAP_info,su->header.prof.ccs /* ToDo */);
2274         TICK_ALLOC_UPD_PAP(words+1,0);
2275
2276         /* now build o = FUN(catch,ap,handler) */
2277         o = (StgClosure *)allocate(sizeofW(StgClosure)+2);
2278         TICK_ALLOC_FUN(2,0);
2279         SET_HDR(o,&catch_info,su->header.prof.ccs /* ToDo */);
2280         o->payload[0] = (StgClosure *)ap;
2281         o->payload[1] = cf->handler;
2282
2283         IF_DEBUG(scheduler,
2284                  fprintf(stderr,  "scheduler: Built ");
2285                  printObj((StgClosure *)o);
2286                  );
2287
2288         /* pop the old handler and put o on the stack */
2289         su = cf->link;
2290         sp += sizeofW(StgCatchFrame) - 1;
2291         sp[0] = (W_)o;
2292         break;
2293       }
2294
2295     case SEQ_FRAME:
2296       {
2297         StgSeqFrame *sf = (StgSeqFrame *)su;
2298         StgClosure* o;
2299
2300         SET_HDR(ap,&PAP_info,su->header.prof.ccs /* ToDo */);
2301         TICK_ALLOC_UPD_PAP(words+1,0);
2302
2303         /* now build o = FUN(seq,ap) */
2304         o = (StgClosure *)allocate(sizeofW(StgClosure)+1);
2305         TICK_ALLOC_SE_THK(1,0);
2306         SET_HDR(o,&seq_info,su->header.prof.ccs /* ToDo */);
2307         payloadCPtr(o,0) = (StgClosure *)ap;
2308
2309         IF_DEBUG(scheduler,
2310                  fprintf(stderr,  "scheduler: Built ");
2311                  printObj((StgClosure *)o);
2312                  );
2313
2314         /* pop the old handler and put o on the stack */
2315         su = sf->link;
2316         sp += sizeofW(StgSeqFrame) - 1;
2317         sp[0] = (W_)o;
2318         break;
2319       }
2320
2321     case STOP_FRAME:
2322       /* We've stripped the entire stack, the thread is now dead. */
2323       sp += sizeofW(StgStopFrame) - 1;
2324       sp[0] = (W_)exception;    /* save the exception */
2325       tso->what_next = ThreadKilled;
2326       tso->su = (StgUpdateFrame *)(sp+1);
2327       tso->sp = sp;
2328       return;
2329
2330     default:
2331       barf("raiseAsync");
2332     }
2333   }
2334   barf("raiseAsync");
2335 }
2336
2337 /* -----------------------------------------------------------------------------
2338    resurrectThreads is called after garbage collection on the list of
2339    threads found to be garbage.  Each of these threads will be woken
2340    up and sent a signal: BlockedOnDeadMVar if the thread was blocked
2341    on an MVar, or NonTermination if the thread was blocked on a Black
2342    Hole.
2343    -------------------------------------------------------------------------- */
2344
2345 void
2346 resurrectThreads( StgTSO *threads )
2347 {
2348   StgTSO *tso, *next;
2349
2350   for (tso = threads; tso != END_TSO_QUEUE; tso = next) {
2351     next = tso->global_link;
2352     tso->global_link = all_threads;
2353     all_threads = tso;
2354     IF_DEBUG(scheduler, sched_belch("resurrecting thread %d", tso->id));
2355
2356     switch (tso->why_blocked) {
2357     case BlockedOnMVar:
2358     case BlockedOnException:
2359       raiseAsync(tso,(StgClosure *)BlockedOnDeadMVar_closure);
2360       break;
2361     case BlockedOnBlackHole:
2362       raiseAsync(tso,(StgClosure *)NonTermination_closure);
2363       break;
2364     case NotBlocked:
2365       /* This might happen if the thread was blocked on a black hole
2366        * belonging to a thread that we've just woken up (raiseAsync
2367        * can wake up threads, remember...).
2368        */
2369       continue;
2370     default:
2371       barf("resurrectThreads: thread blocked in a strange way");
2372     }
2373   }
2374 }
2375
2376 //@node Debugging Routines, Index, Exception Handling Routines, Main scheduling code
2377 //@subsection Debugging Routines
2378
2379 /* -----------------------------------------------------------------------------
2380    Debugging: why is a thread blocked
2381    -------------------------------------------------------------------------- */
2382
2383 #ifdef DEBUG
2384
2385 void
2386 printThreadBlockage(StgTSO *tso)
2387 {
2388   switch (tso->why_blocked) {
2389   case BlockedOnRead:
2390     fprintf(stderr,"blocked on read from fd %d", tso->block_info.fd);
2391     break;
2392   case BlockedOnWrite:
2393     fprintf(stderr,"blocked on write to fd %d", tso->block_info.fd);
2394     break;
2395   case BlockedOnDelay:
2396     fprintf(stderr,"blocked on delay of %d ms", tso->block_info.delay);
2397     break;
2398   case BlockedOnMVar:
2399     fprintf(stderr,"blocked on an MVar");
2400     break;
2401   case BlockedOnException:
2402     fprintf(stderr,"blocked on delivering an exception to thread %d",
2403             tso->block_info.tso->id);
2404     break;
2405   case BlockedOnBlackHole:
2406     fprintf(stderr,"blocked on a black hole");
2407     break;
2408   case NotBlocked:
2409     fprintf(stderr,"not blocked");
2410     break;
2411 #if defined(PAR)
2412   case BlockedOnGA:
2413     fprintf(stderr,"blocked on global address");
2414     break;
2415 #endif
2416   }
2417 }
2418
2419 void
2420 printThreadStatus(StgTSO *tso)
2421 {
2422   switch (tso->what_next) {
2423   case ThreadKilled:
2424     fprintf(stderr,"has been killed");
2425     break;
2426   case ThreadComplete:
2427     fprintf(stderr,"has completed");
2428     break;
2429   default:
2430     printThreadBlockage(tso);
2431   }
2432 }
2433
2434 void
2435 printAllThreads(void)
2436 {
2437   StgTSO *t;
2438
2439   sched_belch("all threads:");
2440   for (t = all_threads; t != END_TSO_QUEUE; t = t->global_link) {
2441     fprintf(stderr, "\tthread %d is ", t->id);
2442     printThreadStatus(t);
2443     fprintf(stderr,"\n");
2444   }
2445 }
2446
2447 /*
2448    Print a whole blocking queue attached to node (debugging only).
2449 */
2450 //@cindex print_bq
2451 # if defined(PAR)
2452 void
2453 print_bq (StgClosure *node)
2454 {
2455   StgBlockingQueueElement *bqe;
2456   StgTSO *tso;
2457   rtsBool end;
2458
2459   fprintf(stderr,"## BQ of closure %p (%s): ",
2460           node, info_type(node));
2461
2462   /* should cover all closures that may have a blocking queue */
2463   ASSERT(get_itbl(node)->type == BLACKHOLE_BQ ||
2464          get_itbl(node)->type == FETCH_ME_BQ ||
2465          get_itbl(node)->type == RBH);
2466
2467   ASSERT(node!=(StgClosure*)NULL);         // sanity check
2468   /*
2469      NB: In a parallel setup a BQ of an RBH must end with an RBH_Save closure;
2470   */
2471   for (bqe = ((StgBlockingQueue*)node)->blocking_queue, end = (bqe==END_BQ_QUEUE);
2472        !end; // iterate until bqe points to a CONSTR
2473        end = (get_itbl(bqe)->type == CONSTR) || (bqe->link==END_BQ_QUEUE), bqe = end ? END_BQ_QUEUE : bqe->link) {
2474     ASSERT(bqe != END_BQ_QUEUE);             // sanity check
2475     ASSERT(bqe != (StgTSO*)NULL);            // sanity check
2476     /* types of closures that may appear in a blocking queue */
2477     ASSERT(get_itbl(bqe)->type == TSO ||
2478            get_itbl(bqe)->type == BLOCKED_FETCH ||
2479            get_itbl(bqe)->type == CONSTR);
2480     /* only BQs of an RBH end with an RBH_Save closure */
2481     ASSERT(get_itbl(bqe)->type != CONSTR || get_itbl(node)->type == RBH);
2482
2483     switch (get_itbl(bqe)->type) {
2484     case TSO:
2485       fprintf(stderr," TSO %d (%x),",
2486               ((StgTSO *)bqe)->id, ((StgTSO *)bqe));
2487       break;
2488     case BLOCKED_FETCH:
2489       fprintf(stderr," BF (node=%p, ga=((%x, %d, %x)),",
2490               ((StgBlockedFetch *)bqe)->node,
2491               ((StgBlockedFetch *)bqe)->ga.payload.gc.gtid,
2492               ((StgBlockedFetch *)bqe)->ga.payload.gc.slot,
2493               ((StgBlockedFetch *)bqe)->ga.weight);
2494       break;
2495     case CONSTR:
2496       fprintf(stderr," %s (IP %p),",
2497               (get_itbl(bqe) == &RBH_Save_0_info ? "RBH_Save_0" :
2498                get_itbl(bqe) == &RBH_Save_1_info ? "RBH_Save_1" :
2499                get_itbl(bqe) == &RBH_Save_2_info ? "RBH_Save_2" :
2500                "RBH_Save_?"), get_itbl(bqe));
2501       break;
2502     default:
2503       barf("Unexpected closure type %s in blocking queue of %p (%s)",
2504            info_type(bqe), node, info_type(node));
2505       break;
2506     }
2507   } /* for */
2508   fputc('\n', stderr);
2509 }
2510 # elif defined(GRAN)
2511 void
2512 print_bq (StgClosure *node)
2513 {
2514   StgBlockingQueueElement *bqe;
2515   StgTSO *tso;
2516   PEs node_loc, tso_loc;
2517   rtsBool end;
2518
2519   /* should cover all closures that may have a blocking queue */
2520   ASSERT(get_itbl(node)->type == BLACKHOLE_BQ ||
2521          get_itbl(node)->type == FETCH_ME_BQ ||
2522          get_itbl(node)->type == RBH);
2523
2524   ASSERT(node!=(StgClosure*)NULL);         // sanity check
2525   node_loc = where_is(node);
2526
2527   fprintf(stderr,"## BQ of closure %p (%s) on [PE %d]: ",
2528           node, info_type(node), node_loc);
2529
2530   /*
2531      NB: In a parallel setup a BQ of an RBH must end with an RBH_Save closure;
2532   */
2533   for (bqe = ((StgBlockingQueue*)node)->blocking_queue, end = (bqe==END_BQ_QUEUE);
2534        !end; // iterate until bqe points to a CONSTR
2535        end = (get_itbl(bqe)->type == CONSTR) || (bqe->link==END_BQ_QUEUE), bqe = end ? END_BQ_QUEUE : bqe->link) {
2536     ASSERT(bqe != END_BQ_QUEUE);             // sanity check
2537     ASSERT(bqe != (StgTSO*)NULL);            // sanity check
2538     /* types of closures that may appear in a blocking queue */
2539     ASSERT(get_itbl(bqe)->type == TSO ||
2540            get_itbl(bqe)->type == CONSTR);
2541     /* only BQs of an RBH end with an RBH_Save closure */
2542     ASSERT(get_itbl(bqe)->type != CONSTR || get_itbl(node)->type == RBH);
2543
2544     tso_loc = where_is((StgClosure *)bqe);
2545     switch (get_itbl(bqe)->type) {
2546     case TSO:
2547       fprintf(stderr," TSO %d (%x) on [PE %d],",
2548               ((StgTSO *)bqe)->id, ((StgTSO *)bqe), tso_loc);
2549       break;
2550     case CONSTR:
2551       fprintf(stderr," %s (IP %p),",
2552               (get_itbl(bqe) == &RBH_Save_0_info ? "RBH_Save_0" :
2553                get_itbl(bqe) == &RBH_Save_1_info ? "RBH_Save_1" :
2554                get_itbl(bqe) == &RBH_Save_2_info ? "RBH_Save_2" :
2555                "RBH_Save_?"), get_itbl(bqe));
2556       break;
2557     default:
2558       barf("Unexpected closure type %s in blocking queue of %p (%s)",
2559            info_type(bqe), node, info_type(node));
2560       break;
2561     }
2562   } /* for */
2563   fputc('\n', stderr);
2564 }
2565 #else
2566 /*
2567    Nice and easy: only TSOs on the blocking queue
2568 */
2569 void
2570 print_bq (StgClosure *node)
2571 {
2572   StgTSO *tso;
2573
2574   ASSERT(node!=(StgClosure*)NULL);         // sanity check
2575   for (tso = ((StgBlockingQueue*)node)->blocking_queue;
2576        tso != END_TSO_QUEUE;
2577        tso=tso->link) {
2578     ASSERT(tso!=NULL && tso!=END_TSO_QUEUE);   // sanity check
2579     ASSERT(get_itbl(tso)->type == TSO);  // guess what, sanity check
2580     fprintf(stderr," TSO %d (%p),", tso->id, tso);
2581   }
2582   fputc('\n', stderr);
2583 }
2584 # endif
2585
2586 #if defined(PAR)
2587 static nat
2588 run_queue_len(void)
2589 {
2590   nat i;
2591   StgTSO *tso;
2592
2593   for (i=0, tso=run_queue_hd;
2594        tso != END_TSO_QUEUE;
2595        i++, tso=tso->link)
2596     /* nothing */
2597
2598   return i;
2599 }
2600 #endif
2601
2602 static void
2603 sched_belch(char *s, ...)
2604 {
2605   va_list ap;
2606   va_start(ap,s);
2607 #ifdef SMP
2608   fprintf(stderr, "scheduler (task %ld): ", pthread_self());
2609 #else
2610   fprintf(stderr, "scheduler: ");
2611 #endif
2612   vfprintf(stderr, s, ap);
2613   fprintf(stderr, "\n");
2614 }
2615
2616 #endif /* DEBUG */
2617
2618 //@node Index,  , Debugging Routines, Main scheduling code
2619 //@subsection Index
2620
2621 //@index
2622 //* MainRegTable::  @cindex\s-+MainRegTable
2623 //* StgMainThread::  @cindex\s-+StgMainThread
2624 //* awaken_blocked_queue::  @cindex\s-+awaken_blocked_queue
2625 //* blocked_queue_hd::  @cindex\s-+blocked_queue_hd
2626 //* blocked_queue_tl::  @cindex\s-+blocked_queue_tl
2627 //* context_switch::  @cindex\s-+context_switch
2628 //* createThread::  @cindex\s-+createThread
2629 //* free_capabilities::  @cindex\s-+free_capabilities
2630 //* gc_pending_cond::  @cindex\s-+gc_pending_cond
2631 //* initScheduler::  @cindex\s-+initScheduler
2632 //* interrupted::  @cindex\s-+interrupted
2633 //* n_free_capabilities::  @cindex\s-+n_free_capabilities
2634 //* next_thread_id::  @cindex\s-+next_thread_id
2635 //* print_bq::  @cindex\s-+print_bq
2636 //* run_queue_hd::  @cindex\s-+run_queue_hd
2637 //* run_queue_tl::  @cindex\s-+run_queue_tl
2638 //* sched_mutex::  @cindex\s-+sched_mutex
2639 //* schedule::  @cindex\s-+schedule
2640 //* take_off_run_queue::  @cindex\s-+take_off_run_queue
2641 //* task_ids::  @cindex\s-+task_ids
2642 //* term_mutex::  @cindex\s-+term_mutex
2643 //* thread_ready_cond::  @cindex\s-+thread_ready_cond
2644 //@end index