ghc/rts/Schedule.c

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