1 /* ---------------------------------------------------------------------------
2 * $Id: Schedule.c,v 1.184 2003/12/18 12:24:59 simonmar Exp $
4 * (c) The GHC Team, 1998-2003
8 * Different GHC ways use this scheduler quite differently (see comments below)
9 * Here is the global picture:
11 * WAY Name CPP flag What's it for
12 * --------------------------------------
13 * mp GUM PAR Parallel execution on a distrib. memory machine
14 * s SMP SMP Parallel execution on a shared memory machine
15 * mg GranSim GRAN Simulation of parallel execution
16 * md GUM/GdH DIST Distributed execution (based on GUM)
18 * --------------------------------------------------------------------------*/
21 * Version with support for distributed memory parallelism aka GUM (WAY=mp):
23 The main scheduling loop in GUM iterates until a finish message is received.
24 In that case a global flag @receivedFinish@ is set and this instance of
25 the RTS shuts down. See ghc/rts/parallel/HLComms.c:processMessages()
26 for the handling of incoming messages, such as PP_FINISH.
27 Note that in the parallel case we have a system manager that coordinates
28 different PEs, each of which are running one instance of the RTS.
29 See ghc/rts/parallel/SysMan.c for the main routine of the parallel program.
30 From this routine processes executing ghc/rts/Main.c are spawned. -- HWL
32 * Version with support for simulating parallel execution aka GranSim (WAY=mg):
34 The main scheduling code in GranSim is quite different from that in std
35 (concurrent) Haskell: while concurrent Haskell just iterates over the
36 threads in the runnable queue, GranSim is event driven, i.e. it iterates
37 over the events in the global event queue. -- HWL
40 #include "PosixSource.h"
47 #include "StgStartup.h"
49 #define COMPILING_SCHEDULER
51 #include "StgMiscClosures.h"
53 #include "Interpreter.h"
54 #include "Exception.h"
61 #include "ThreadLabels.h"
63 #include "Proftimer.h"
66 #if defined(GRAN) || defined(PAR)
67 # include "GranSimRts.h"
69 # include "ParallelRts.h"
70 # include "Parallel.h"
71 # include "ParallelDebug.h"
76 #include "Capability.h"
77 #include "OSThreads.h"
80 #ifdef HAVE_SYS_TYPES_H
81 #include <sys/types.h>
96 #define USED_IN_THREADED_RTS
98 #define USED_IN_THREADED_RTS STG_UNUSED
101 #ifdef RTS_SUPPORTS_THREADS
102 #define USED_WHEN_RTS_SUPPORTS_THREADS
104 #define USED_WHEN_RTS_SUPPORTS_THREADS STG_UNUSED
107 /* Main thread queue.
108 * Locks required: sched_mutex.
110 StgMainThread *main_threads = NULL;
113 * Locks required: sched_mutex.
117 StgTSO* ActiveTSO = NULL; /* for assigning system costs; GranSim-Light only */
118 /* rtsTime TimeOfNextEvent, EndOfTimeSlice; now in GranSim.c */
121 In GranSim we have a runnable and a blocked queue for each processor.
122 In order to minimise code changes new arrays run_queue_hds/tls
123 are created. run_queue_hd is then a short cut (macro) for
124 run_queue_hds[CurrentProc] (see GranSim.h).
127 StgTSO *run_queue_hds[MAX_PROC], *run_queue_tls[MAX_PROC];
128 StgTSO *blocked_queue_hds[MAX_PROC], *blocked_queue_tls[MAX_PROC];
129 StgTSO *ccalling_threadss[MAX_PROC];
130 /* We use the same global list of threads (all_threads) in GranSim as in
131 the std RTS (i.e. we are cheating). However, we don't use this list in
132 the GranSim specific code at the moment (so we are only potentially
137 StgTSO *run_queue_hd = NULL;
138 StgTSO *run_queue_tl = NULL;
139 StgTSO *blocked_queue_hd = NULL;
140 StgTSO *blocked_queue_tl = NULL;
141 StgTSO *sleeping_queue = NULL; /* perhaps replace with a hash table? */
145 /* Linked list of all threads.
146 * Used for detecting garbage collected threads.
148 StgTSO *all_threads = NULL;
150 /* When a thread performs a safe C call (_ccall_GC, using old
151 * terminology), it gets put on the suspended_ccalling_threads
152 * list. Used by the garbage collector.
154 static StgTSO *suspended_ccalling_threads;
156 static StgTSO *threadStackOverflow(StgTSO *tso);
158 /* KH: The following two flags are shared memory locations. There is no need
159 to lock them, since they are only unset at the end of a scheduler
163 /* flag set by signal handler to precipitate a context switch */
164 nat context_switch = 0;
166 /* if this flag is set as well, give up execution */
167 rtsBool interrupted = rtsFalse;
169 /* Next thread ID to allocate.
170 * Locks required: thread_id_mutex
172 static StgThreadID next_thread_id = 1;
175 * Pointers to the state of the current thread.
176 * Rule of thumb: if CurrentTSO != NULL, then we're running a Haskell
177 * thread. If CurrentTSO == NULL, then we're at the scheduler level.
180 /* The smallest stack size that makes any sense is:
181 * RESERVED_STACK_WORDS (so we can get back from the stack overflow)
182 * + sizeofW(StgStopFrame) (the stg_stop_thread_info frame)
183 * + 1 (the closure to enter)
185 * + 1 (spare slot req'd by stg_ap_v_ret)
187 * A thread with this stack will bomb immediately with a stack
188 * overflow, which will increase its stack size.
191 #define MIN_STACK_WORDS (RESERVED_STACK_WORDS + sizeofW(StgStopFrame) + 3)
198 /* This is used in `TSO.h' and gcc 2.96 insists that this variable actually
199 * exists - earlier gccs apparently didn't.
204 static rtsBool ready_to_gc;
207 * Set to TRUE when entering a shutdown state (via shutdownHaskellAndExit()) --
208 * in an MT setting, needed to signal that a worker thread shouldn't hang around
209 * in the scheduler when it is out of work.
211 static rtsBool shutting_down_scheduler = rtsFalse;
213 void addToBlockedQueue ( StgTSO *tso );
215 static void schedule ( StgMainThread *mainThread, Capability *initialCapability );
216 void interruptStgRts ( void );
218 static void detectBlackHoles ( void );
220 #if defined(RTS_SUPPORTS_THREADS)
221 /* ToDo: carefully document the invariants that go together
222 * with these synchronisation objects.
224 Mutex sched_mutex = INIT_MUTEX_VAR;
225 Mutex term_mutex = INIT_MUTEX_VAR;
228 * A heavyweight solution to the problem of protecting
229 * the thread_id from concurrent update.
231 Mutex thread_id_mutex = INIT_MUTEX_VAR;
233 #endif /* RTS_SUPPORTS_THREADS */
237 rtsTime TimeOfLastYield;
238 rtsBool emitSchedule = rtsTrue;
242 static char *whatNext_strs[] = {
252 StgTSO * createSparkThread(rtsSpark spark);
253 StgTSO * activateSpark (rtsSpark spark);
256 /* ----------------------------------------------------------------------------
258 * ------------------------------------------------------------------------- */
260 #if defined(RTS_SUPPORTS_THREADS)
261 static rtsBool startingWorkerThread = rtsFalse;
263 static void taskStart(void);
267 ACQUIRE_LOCK(&sched_mutex);
269 RELEASE_LOCK(&sched_mutex);
273 startSchedulerTaskIfNecessary(void)
275 if(run_queue_hd != END_TSO_QUEUE
276 || blocked_queue_hd != END_TSO_QUEUE
277 || sleeping_queue != END_TSO_QUEUE)
279 if(!startingWorkerThread)
280 { // we don't want to start another worker thread
281 // just because the last one hasn't yet reached the
282 // "waiting for capability" state
283 startingWorkerThread = rtsTrue;
284 startTask(taskStart);
290 /* ---------------------------------------------------------------------------
291 Main scheduling loop.
293 We use round-robin scheduling, each thread returning to the
294 scheduler loop when one of these conditions is detected:
297 * timer expires (thread yields)
302 Locking notes: we acquire the scheduler lock once at the beginning
303 of the scheduler loop, and release it when
305 * running a thread, or
306 * waiting for work, or
307 * waiting for a GC to complete.
310 In a GranSim setup this loop iterates over the global event queue.
311 This revolves around the global event queue, which determines what
312 to do next. Therefore, it's more complicated than either the
313 concurrent or the parallel (GUM) setup.
316 GUM iterates over incoming messages.
317 It starts with nothing to do (thus CurrentTSO == END_TSO_QUEUE),
318 and sends out a fish whenever it has nothing to do; in-between
319 doing the actual reductions (shared code below) it processes the
320 incoming messages and deals with delayed operations
321 (see PendingFetches).
322 This is not the ugliest code you could imagine, but it's bloody close.
324 ------------------------------------------------------------------------ */
326 schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS,
327 Capability *initialCapability )
330 Capability *cap = initialCapability;
331 StgThreadReturnCode ret;
339 rtsBool receivedFinish = rtsFalse;
341 nat tp_size, sp_size; // stats only
344 rtsBool was_interrupted = rtsFalse;
345 StgTSOWhatNext prev_what_next;
347 // Pre-condition: sched_mutex is held.
349 #if defined(RTS_SUPPORTS_THREADS)
351 // in the threaded case, the capability is either passed in via the
352 // initialCapability parameter, or initialized inside the scheduler
356 sched_belch("### NEW SCHEDULER LOOP (main thr: %p, cap: %p)",
357 mainThread, initialCapability);
360 // simply initialise it in the non-threaded case
361 grabCapability(&cap);
365 /* set up first event to get things going */
366 /* ToDo: assign costs for system setup and init MainTSO ! */
367 new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
369 CurrentTSO, (StgClosure*)NULL, (rtsSpark*)NULL);
372 fprintf(stderr, "GRAN: Init CurrentTSO (in schedule) = %p\n", CurrentTSO);
373 G_TSO(CurrentTSO, 5));
375 if (RtsFlags.GranFlags.Light) {
376 /* Save current time; GranSim Light only */
377 CurrentTSO->gran.clock = CurrentTime[CurrentProc];
380 event = get_next_event();
382 while (event!=(rtsEvent*)NULL) {
383 /* Choose the processor with the next event */
384 CurrentProc = event->proc;
385 CurrentTSO = event->tso;
389 while (!receivedFinish) { /* set by processMessages */
390 /* when receiving PP_FINISH message */
392 #else // everything except GRAN and PAR
398 IF_DEBUG(scheduler, printAllThreads());
400 #if defined(RTS_SUPPORTS_THREADS)
401 // Yield the capability to higher-priority tasks if necessary.
404 yieldCapability(&cap);
407 // If we do not currently hold a capability, we wait for one
410 waitForCapability(&sched_mutex, &cap,
411 mainThread ? &mainThread->bound_thread_cond : NULL);
414 // We now have a capability...
418 // If we're interrupted (the user pressed ^C, or some other
419 // termination condition occurred), kill all the currently running
423 IF_DEBUG(scheduler, sched_belch("interrupted"));
424 interrupted = rtsFalse;
425 was_interrupted = rtsTrue;
426 #if defined(RTS_SUPPORTS_THREADS)
427 // In the threaded RTS, deadlock detection doesn't work,
428 // so just exit right away.
429 prog_belch("interrupted");
430 releaseCapability(cap);
431 RELEASE_LOCK(&sched_mutex);
432 shutdownHaskellAndExit(EXIT_SUCCESS);
439 // Go through the list of main threads and wake up any
440 // clients whose computations have finished. ToDo: this
441 // should be done more efficiently without a linear scan
442 // of the main threads list, somehow...
444 #if defined(RTS_SUPPORTS_THREADS)
446 StgMainThread *m, **prev;
447 prev = &main_threads;
448 for (m = main_threads; m != NULL; prev = &m->link, m = m->link) {
449 if (m->tso->what_next == ThreadComplete
450 || m->tso->what_next == ThreadKilled)
454 if (m->tso->what_next == ThreadComplete)
458 // NOTE: return val is tso->sp[1] (see StgStartup.hc)
459 *(m->ret) = (StgClosure *)m->tso->sp[1];
471 m->stat = Interrupted;
481 removeThreadLabel((StgWord)m->tso->id);
483 releaseCapability(cap);
488 // The current OS thread can not handle the fact that
489 // the Haskell thread "m" has ended. "m" is bound;
490 // the scheduler loop in it's bound OS thread has to
491 // return, so let's pass our capability directly to
493 passCapability(&m->bound_thread_cond);
500 #else /* not threaded */
503 /* in GUM do this only on the Main PE */
506 /* If our main thread has finished or been killed, return.
509 StgMainThread *m = main_threads;
510 if (m->tso->what_next == ThreadComplete
511 || m->tso->what_next == ThreadKilled) {
513 removeThreadLabel((StgWord)m->tso->id);
515 main_threads = main_threads->link;
516 if (m->tso->what_next == ThreadComplete) {
517 // We finished successfully, fill in the return value
518 // NOTE: return val is tso->sp[1] (see StgStartup.hc)
519 if (m->ret) { *(m->ret) = (StgClosure *)m->tso->sp[1]; };
523 if (m->ret) { *(m->ret) = NULL; };
524 if (was_interrupted) {
525 m->stat = Interrupted;
536 #if defined(RTS_USER_SIGNALS)
537 // check for signals each time around the scheduler
538 if (signals_pending()) {
539 RELEASE_LOCK(&sched_mutex); /* ToDo: kill */
540 startSignalHandlers();
541 ACQUIRE_LOCK(&sched_mutex);
545 /* Check whether any waiting threads need to be woken up. If the
546 * run queue is empty, and there are no other tasks running, we
547 * can wait indefinitely for something to happen.
549 if ( !EMPTY_QUEUE(blocked_queue_hd) || !EMPTY_QUEUE(sleeping_queue)
550 #if defined(RTS_SUPPORTS_THREADS)
555 awaitEvent( EMPTY_RUN_QUEUE() );
557 /* we can be interrupted while waiting for I/O... */
558 if (interrupted) continue;
561 * Detect deadlock: when we have no threads to run, there are no
562 * threads waiting on I/O or sleeping, and all the other tasks are
563 * waiting for work, we must have a deadlock of some description.
565 * We first try to find threads blocked on themselves (ie. black
566 * holes), and generate NonTermination exceptions where necessary.
568 * If no threads are black holed, we have a deadlock situation, so
569 * inform all the main threads.
571 #if !defined(PAR) && !defined(RTS_SUPPORTS_THREADS)
572 if ( EMPTY_THREAD_QUEUES() )
574 IF_DEBUG(scheduler, sched_belch("deadlocked, forcing major GC..."));
575 // Garbage collection can release some new threads due to
576 // either (a) finalizers or (b) threads resurrected because
577 // they are about to be send BlockedOnDeadMVar. Any threads
578 // thus released will be immediately runnable.
579 GarbageCollect(GetRoots,rtsTrue);
581 if ( !EMPTY_RUN_QUEUE() ) { goto not_deadlocked; }
584 sched_belch("still deadlocked, checking for black holes..."));
587 if ( !EMPTY_RUN_QUEUE() ) { goto not_deadlocked; }
589 #if defined(RTS_USER_SIGNALS)
590 /* If we have user-installed signal handlers, then wait
591 * for signals to arrive rather then bombing out with a
594 if ( anyUserHandlers() ) {
596 sched_belch("still deadlocked, waiting for signals..."));
600 // we might be interrupted...
601 if (interrupted) { continue; }
603 if (signals_pending()) {
604 RELEASE_LOCK(&sched_mutex);
605 startSignalHandlers();
606 ACQUIRE_LOCK(&sched_mutex);
608 ASSERT(!EMPTY_RUN_QUEUE());
613 /* Probably a real deadlock. Send the current main thread the
614 * Deadlock exception (or in the SMP build, send *all* main
615 * threads the deadlock exception, since none of them can make
621 switch (m->tso->why_blocked) {
622 case BlockedOnBlackHole:
623 raiseAsync(m->tso, (StgClosure *)NonTermination_closure);
625 case BlockedOnException:
627 raiseAsync(m->tso, (StgClosure *)Deadlock_closure);
630 barf("deadlock: main thread blocked in a strange way");
636 #elif defined(RTS_SUPPORTS_THREADS)
637 // ToDo: add deadlock detection in threaded RTS
639 // ToDo: add deadlock detection in GUM (similar to SMP) -- HWL
642 #if defined(RTS_SUPPORTS_THREADS)
643 if ( EMPTY_RUN_QUEUE() ) {
644 continue; // nothing to do
649 if (RtsFlags.GranFlags.Light)
650 GranSimLight_enter_system(event, &ActiveTSO); // adjust ActiveTSO etc
652 /* adjust time based on time-stamp */
653 if (event->time > CurrentTime[CurrentProc] &&
654 event->evttype != ContinueThread)
655 CurrentTime[CurrentProc] = event->time;
657 /* Deal with the idle PEs (may issue FindWork or MoveSpark events) */
658 if (!RtsFlags.GranFlags.Light)
661 IF_DEBUG(gran, fprintf(stderr, "GRAN: switch by event-type\n"));
663 /* main event dispatcher in GranSim */
664 switch (event->evttype) {
665 /* Should just be continuing execution */
667 IF_DEBUG(gran, fprintf(stderr, "GRAN: doing ContinueThread\n"));
668 /* ToDo: check assertion
669 ASSERT(run_queue_hd != (StgTSO*)NULL &&
670 run_queue_hd != END_TSO_QUEUE);
672 /* Ignore ContinueThreads for fetching threads (if synchr comm) */
673 if (!RtsFlags.GranFlags.DoAsyncFetch &&
674 procStatus[CurrentProc]==Fetching) {
675 belch("ghuH: Spurious ContinueThread while Fetching ignored; TSO %d (%p) [PE %d]",
676 CurrentTSO->id, CurrentTSO, CurrentProc);
679 /* Ignore ContinueThreads for completed threads */
680 if (CurrentTSO->what_next == ThreadComplete) {
681 belch("ghuH: found a ContinueThread event for completed thread %d (%p) [PE %d] (ignoring ContinueThread)",
682 CurrentTSO->id, CurrentTSO, CurrentProc);
685 /* Ignore ContinueThreads for threads that are being migrated */
686 if (PROCS(CurrentTSO)==Nowhere) {
687 belch("ghuH: trying to run the migrating TSO %d (%p) [PE %d] (ignoring ContinueThread)",
688 CurrentTSO->id, CurrentTSO, CurrentProc);
691 /* The thread should be at the beginning of the run queue */
692 if (CurrentTSO!=run_queue_hds[CurrentProc]) {
693 belch("ghuH: TSO %d (%p) [PE %d] is not at the start of the run_queue when doing a ContinueThread",
694 CurrentTSO->id, CurrentTSO, CurrentProc);
695 break; // run the thread anyway
698 new_event(proc, proc, CurrentTime[proc],
700 (StgTSO*)NULL, (StgClosure*)NULL, (rtsSpark*)NULL);
702 */ /* Catches superfluous CONTINUEs -- should be unnecessary */
703 break; // now actually run the thread; DaH Qu'vam yImuHbej
706 do_the_fetchnode(event);
707 goto next_thread; /* handle next event in event queue */
710 do_the_globalblock(event);
711 goto next_thread; /* handle next event in event queue */
714 do_the_fetchreply(event);
715 goto next_thread; /* handle next event in event queue */
717 case UnblockThread: /* Move from the blocked queue to the tail of */
718 do_the_unblock(event);
719 goto next_thread; /* handle next event in event queue */
721 case ResumeThread: /* Move from the blocked queue to the tail of */
722 /* the runnable queue ( i.e. Qu' SImqa'lu') */
723 event->tso->gran.blocktime +=
724 CurrentTime[CurrentProc] - event->tso->gran.blockedat;
725 do_the_startthread(event);
726 goto next_thread; /* handle next event in event queue */
729 do_the_startthread(event);
730 goto next_thread; /* handle next event in event queue */
733 do_the_movethread(event);
734 goto next_thread; /* handle next event in event queue */
737 do_the_movespark(event);
738 goto next_thread; /* handle next event in event queue */
741 do_the_findwork(event);
742 goto next_thread; /* handle next event in event queue */
745 barf("Illegal event type %u\n", event->evttype);
748 /* This point was scheduler_loop in the old RTS */
750 IF_DEBUG(gran, belch("GRAN: after main switch"));
752 TimeOfLastEvent = CurrentTime[CurrentProc];
753 TimeOfNextEvent = get_time_of_next_event();
754 IgnoreEvents=(TimeOfNextEvent==0); // HWL HACK
755 // CurrentTSO = ThreadQueueHd;
757 IF_DEBUG(gran, belch("GRAN: time of next event is: %ld",
760 if (RtsFlags.GranFlags.Light)
761 GranSimLight_leave_system(event, &ActiveTSO);
763 EndOfTimeSlice = CurrentTime[CurrentProc]+RtsFlags.GranFlags.time_slice;
766 belch("GRAN: end of time-slice is %#lx", EndOfTimeSlice));
768 /* in a GranSim setup the TSO stays on the run queue */
770 /* Take a thread from the run queue. */
771 POP_RUN_QUEUE(t); // take_off_run_queue(t);
774 fprintf(stderr, "GRAN: About to run current thread, which is\n");
777 context_switch = 0; // turned on via GranYield, checking events and time slice
780 DumpGranEvent(GR_SCHEDULE, t));
782 procStatus[CurrentProc] = Busy;
785 if (PendingFetches != END_BF_QUEUE) {
789 /* ToDo: phps merge with spark activation above */
790 /* check whether we have local work and send requests if we have none */
791 if (EMPTY_RUN_QUEUE()) { /* no runnable threads */
792 /* :-[ no local threads => look out for local sparks */
793 /* the spark pool for the current PE */
794 pool = &(MainRegTable.rSparks); // generalise to cap = &MainRegTable
795 if (advisory_thread_count < RtsFlags.ParFlags.maxThreads &&
796 pool->hd < pool->tl) {
798 * ToDo: add GC code check that we really have enough heap afterwards!!
800 * If we're here (no runnable threads) and we have pending
801 * sparks, we must have a space problem. Get enough space
802 * to turn one of those pending sparks into a
806 spark = findSpark(rtsFalse); /* get a spark */
807 if (spark != (rtsSpark) NULL) {
808 tso = activateSpark(spark); /* turn the spark into a thread */
809 IF_PAR_DEBUG(schedule,
810 belch("==== schedule: Created TSO %d (%p); %d threads active",
811 tso->id, tso, advisory_thread_count));
813 if (tso==END_TSO_QUEUE) { /* failed to activate spark->back to loop */
814 belch("==^^ failed to activate spark");
816 } /* otherwise fall through & pick-up new tso */
818 IF_PAR_DEBUG(verbose,
819 belch("==^^ no local sparks (spark pool contains only NFs: %d)",
820 spark_queue_len(pool)));
825 /* If we still have no work we need to send a FISH to get a spark
828 if (EMPTY_RUN_QUEUE()) {
829 /* =8-[ no local sparks => look for work on other PEs */
831 * We really have absolutely no work. Send out a fish
832 * (there may be some out there already), and wait for
833 * something to arrive. We clearly can't run any threads
834 * until a SCHEDULE or RESUME arrives, and so that's what
835 * we're hoping to see. (Of course, we still have to
836 * respond to other types of messages.)
838 TIME now = msTime() /*CURRENT_TIME*/;
839 IF_PAR_DEBUG(verbose,
840 belch("-- now=%ld", now));
841 IF_PAR_DEBUG(verbose,
842 if (outstandingFishes < RtsFlags.ParFlags.maxFishes &&
843 (last_fish_arrived_at!=0 &&
844 last_fish_arrived_at+RtsFlags.ParFlags.fishDelay > now)) {
845 belch("--$$ delaying FISH until %ld (last fish %ld, delay %ld, now %ld)",
846 last_fish_arrived_at+RtsFlags.ParFlags.fishDelay,
847 last_fish_arrived_at,
848 RtsFlags.ParFlags.fishDelay, now);
851 if (outstandingFishes < RtsFlags.ParFlags.maxFishes &&
852 (last_fish_arrived_at==0 ||
853 (last_fish_arrived_at+RtsFlags.ParFlags.fishDelay <= now))) {
854 /* outstandingFishes is set in sendFish, processFish;
855 avoid flooding system with fishes via delay */
857 sendFish(pe, mytid, NEW_FISH_AGE, NEW_FISH_HISTORY,
860 // Global statistics: count no. of fishes
861 if (RtsFlags.ParFlags.ParStats.Global &&
862 RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
863 globalParStats.tot_fish_mess++;
867 receivedFinish = processMessages();
870 } else if (PacketsWaiting()) { /* Look for incoming messages */
871 receivedFinish = processMessages();
874 /* Now we are sure that we have some work available */
875 ASSERT(run_queue_hd != END_TSO_QUEUE);
877 /* Take a thread from the run queue, if we have work */
878 POP_RUN_QUEUE(t); // take_off_run_queue(END_TSO_QUEUE);
879 IF_DEBUG(sanity,checkTSO(t));
881 /* ToDo: write something to the log-file
882 if (RTSflags.ParFlags.granSimStats && !sameThread)
883 DumpGranEvent(GR_SCHEDULE, RunnableThreadsHd);
887 /* the spark pool for the current PE */
888 pool = &(MainRegTable.rSparks); // generalise to cap = &MainRegTable
891 belch("--=^ %d threads, %d sparks on [%#x]",
892 run_queue_len(), spark_queue_len(pool), CURRENT_PROC));
895 if (0 && RtsFlags.ParFlags.ParStats.Full &&
896 t && LastTSO && t->id != LastTSO->id &&
897 LastTSO->why_blocked == NotBlocked &&
898 LastTSO->what_next != ThreadComplete) {
899 // if previously scheduled TSO not blocked we have to record the context switch
900 DumpVeryRawGranEvent(TimeOfLastYield, CURRENT_PROC, CURRENT_PROC,
901 GR_DESCHEDULE, LastTSO, (StgClosure *)NULL, 0, 0);
904 if (RtsFlags.ParFlags.ParStats.Full &&
905 (emitSchedule /* forced emit */ ||
906 (t && LastTSO && t->id != LastTSO->id))) {
908 we are running a different TSO, so write a schedule event to log file
909 NB: If we use fair scheduling we also have to write a deschedule
910 event for LastTSO; with unfair scheduling we know that the
911 previous tso has blocked whenever we switch to another tso, so
912 we don't need it in GUM for now
914 DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
915 GR_SCHEDULE, t, (StgClosure *)NULL, 0, 0);
916 emitSchedule = rtsFalse;
920 #else /* !GRAN && !PAR */
922 // grab a thread from the run queue
923 ASSERT(run_queue_hd != END_TSO_QUEUE);
926 // Sanity check the thread we're about to run. This can be
927 // expensive if there is lots of thread switching going on...
928 IF_DEBUG(sanity,checkTSO(t));
934 for(m = main_threads; m; m = m->link)
945 sched_belch("### Running thread %d in bound thread", t->id));
946 // yes, the Haskell thread is bound to the current native thread
951 sched_belch("### thread %d bound to another OS thread", t->id));
952 // no, bound to a different Haskell thread: pass to that thread
953 PUSH_ON_RUN_QUEUE(t);
954 passCapability(&m->bound_thread_cond);
960 if(mainThread != NULL)
961 // The thread we want to run is bound.
964 sched_belch("### this OS thread cannot run thread %d", t->id));
965 // no, the current native thread is bound to a different
966 // Haskell thread, so pass it to any worker thread
967 PUSH_ON_RUN_QUEUE(t);
968 passCapabilityToWorker();
975 cap->r.rCurrentTSO = t;
977 /* context switches are now initiated by the timer signal, unless
978 * the user specified "context switch as often as possible", with
981 if ((RtsFlags.ConcFlags.ctxtSwitchTicks == 0
982 && (run_queue_hd != END_TSO_QUEUE
983 || blocked_queue_hd != END_TSO_QUEUE
984 || sleeping_queue != END_TSO_QUEUE)))
991 RELEASE_LOCK(&sched_mutex);
993 IF_DEBUG(scheduler, sched_belch("-->> running thread %ld %s ...",
994 t->id, whatNext_strs[t->what_next]));
997 startHeapProfTimer();
1000 /* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
1001 /* Run the current thread
1003 prev_what_next = t->what_next;
1004 switch (prev_what_next) {
1006 case ThreadComplete:
1007 /* Thread already finished, return to scheduler. */
1008 ret = ThreadFinished;
1011 errno = t->saved_errno;
1012 ret = StgRun((StgFunPtr) stg_returnToStackTop, &cap->r);
1013 t->saved_errno = errno;
1015 case ThreadInterpret:
1016 ret = interpretBCO(cap);
1019 barf("schedule: invalid what_next field");
1021 /* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
1023 /* Costs for the scheduler are assigned to CCS_SYSTEM */
1025 stopHeapProfTimer();
1029 ACQUIRE_LOCK(&sched_mutex);
1031 #ifdef RTS_SUPPORTS_THREADS
1032 IF_DEBUG(scheduler,fprintf(stderr,"sched (task %p): ", osThreadId()););
1033 #elif !defined(GRAN) && !defined(PAR)
1034 IF_DEBUG(scheduler,fprintf(stderr,"sched: "););
1036 t = cap->r.rCurrentTSO;
1039 /* HACK 675: if the last thread didn't yield, make sure to print a
1040 SCHEDULE event to the log file when StgRunning the next thread, even
1041 if it is the same one as before */
1043 TimeOfLastYield = CURRENT_TIME;
1049 IF_DEBUG(gran, DumpGranEvent(GR_DESCHEDULE, t));
1050 globalGranStats.tot_heapover++;
1052 globalParStats.tot_heapover++;
1055 // did the task ask for a large block?
1056 if (cap->r.rHpAlloc > BLOCK_SIZE_W) {
1057 // if so, get one and push it on the front of the nursery.
1061 blocks = (nat)BLOCK_ROUND_UP(cap->r.rHpAlloc * sizeof(W_)) / BLOCK_SIZE;
1063 IF_DEBUG(scheduler,belch("--<< thread %ld (%s) stopped: requesting a large block (size %d)",
1064 t->id, whatNext_strs[t->what_next], blocks));
1066 // don't do this if it would push us over the
1067 // alloc_blocks_lim limit; we'll GC first.
1068 if (alloc_blocks + blocks < alloc_blocks_lim) {
1070 alloc_blocks += blocks;
1071 bd = allocGroup( blocks );
1073 // link the new group into the list
1074 bd->link = cap->r.rCurrentNursery;
1075 bd->u.back = cap->r.rCurrentNursery->u.back;
1076 if (cap->r.rCurrentNursery->u.back != NULL) {
1077 cap->r.rCurrentNursery->u.back->link = bd;
1079 ASSERT(g0s0->blocks == cap->r.rCurrentNursery &&
1080 g0s0->blocks == cap->r.rNursery);
1081 cap->r.rNursery = g0s0->blocks = bd;
1083 cap->r.rCurrentNursery->u.back = bd;
1085 // initialise it as a nursery block. We initialise the
1086 // step, gen_no, and flags field of *every* sub-block in
1087 // this large block, because this is easier than making
1088 // sure that we always find the block head of a large
1089 // block whenever we call Bdescr() (eg. evacuate() and
1090 // isAlive() in the GC would both have to do this, at
1094 for (x = bd; x < bd + blocks; x++) {
1101 // don't forget to update the block count in g0s0.
1102 g0s0->n_blocks += blocks;
1103 // This assert can be a killer if the app is doing lots
1104 // of large block allocations.
1105 ASSERT(countBlocks(g0s0->blocks) == g0s0->n_blocks);
1107 // now update the nursery to point to the new block
1108 cap->r.rCurrentNursery = bd;
1110 // we might be unlucky and have another thread get on the
1111 // run queue before us and steal the large block, but in that
1112 // case the thread will just end up requesting another large
1114 PUSH_ON_RUN_QUEUE(t);
1119 /* make all the running tasks block on a condition variable,
1120 * maybe set context_switch and wait till they all pile in,
1121 * then have them wait on a GC condition variable.
1123 IF_DEBUG(scheduler,belch("--<< thread %ld (%s) stopped: HeapOverflow",
1124 t->id, whatNext_strs[t->what_next]));
1127 ASSERT(!is_on_queue(t,CurrentProc));
1129 /* Currently we emit a DESCHEDULE event before GC in GUM.
1130 ToDo: either add separate event to distinguish SYSTEM time from rest
1131 or just nuke this DESCHEDULE (and the following SCHEDULE) */
1132 if (0 && RtsFlags.ParFlags.ParStats.Full) {
1133 DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
1134 GR_DESCHEDULE, t, (StgClosure *)NULL, 0, 0);
1135 emitSchedule = rtsTrue;
1139 ready_to_gc = rtsTrue;
1140 context_switch = 1; /* stop other threads ASAP */
1141 PUSH_ON_RUN_QUEUE(t);
1142 /* actual GC is done at the end of the while loop */
1148 DumpGranEvent(GR_DESCHEDULE, t));
1149 globalGranStats.tot_stackover++;
1152 // DumpGranEvent(GR_DESCHEDULE, t);
1153 globalParStats.tot_stackover++;
1155 IF_DEBUG(scheduler,belch("--<< thread %ld (%s) stopped, StackOverflow",
1156 t->id, whatNext_strs[t->what_next]));
1157 /* just adjust the stack for this thread, then pop it back
1163 /* enlarge the stack */
1164 StgTSO *new_t = threadStackOverflow(t);
1166 /* This TSO has moved, so update any pointers to it from the
1167 * main thread stack. It better not be on any other queues...
1168 * (it shouldn't be).
1170 for (m = main_threads; m != NULL; m = m->link) {
1175 threadPaused(new_t);
1176 PUSH_ON_RUN_QUEUE(new_t);
1180 case ThreadYielding:
1183 DumpGranEvent(GR_DESCHEDULE, t));
1184 globalGranStats.tot_yields++;
1187 // DumpGranEvent(GR_DESCHEDULE, t);
1188 globalParStats.tot_yields++;
1190 /* put the thread back on the run queue. Then, if we're ready to
1191 * GC, check whether this is the last task to stop. If so, wake
1192 * up the GC thread. getThread will block during a GC until the
1196 if (t->what_next != prev_what_next) {
1197 belch("--<< thread %ld (%s) stopped to switch evaluators",
1198 t->id, whatNext_strs[t->what_next]);
1200 belch("--<< thread %ld (%s) stopped, yielding",
1201 t->id, whatNext_strs[t->what_next]);
1206 //belch("&& Doing sanity check on yielding TSO %ld.", t->id);
1208 ASSERT(t->link == END_TSO_QUEUE);
1210 // Shortcut if we're just switching evaluators: don't bother
1211 // doing stack squeezing (which can be expensive), just run the
1213 if (t->what_next != prev_what_next) {
1220 ASSERT(!is_on_queue(t,CurrentProc));
1223 //belch("&& Doing sanity check on all ThreadQueues (and their TSOs).");
1224 checkThreadQsSanity(rtsTrue));
1228 if (RtsFlags.ParFlags.doFairScheduling) {
1229 /* this does round-robin scheduling; good for concurrency */
1230 APPEND_TO_RUN_QUEUE(t);
1232 /* this does unfair scheduling; good for parallelism */
1233 PUSH_ON_RUN_QUEUE(t);
1236 // this does round-robin scheduling; good for concurrency
1237 APPEND_TO_RUN_QUEUE(t);
1241 /* add a ContinueThread event to actually process the thread */
1242 new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
1244 t, (StgClosure*)NULL, (rtsSpark*)NULL);
1246 belch("GRAN: eventq and runnableq after adding yielded thread to queue again:");
1255 belch("--<< thread %ld (%p; %s) stopped, blocking on node %p [PE %d] with BQ: ",
1256 t->id, t, whatNext_strs[t->what_next], t->block_info.closure, (t->block_info.closure==(StgClosure*)NULL ? 99 : where_is(t->block_info.closure)));
1257 if (t->block_info.closure!=(StgClosure*)NULL) print_bq(t->block_info.closure));
1259 // ??? needed; should emit block before
1261 DumpGranEvent(GR_DESCHEDULE, t));
1262 prune_eventq(t, (StgClosure *)NULL); // prune ContinueThreads for t
1265 ASSERT(procStatus[CurrentProc]==Busy ||
1266 ((procStatus[CurrentProc]==Fetching) &&
1267 (t->block_info.closure!=(StgClosure*)NULL)));
1268 if (run_queue_hds[CurrentProc] == END_TSO_QUEUE &&
1269 !(!RtsFlags.GranFlags.DoAsyncFetch &&
1270 procStatus[CurrentProc]==Fetching))
1271 procStatus[CurrentProc] = Idle;
1275 belch("--<< thread %ld (%p; %s) stopped, blocking on node %p with BQ: ",
1276 t->id, t, whatNext_strs[t->what_next], t->block_info.closure));
1279 if (t->block_info.closure!=(StgClosure*)NULL)
1280 print_bq(t->block_info.closure));
1282 /* Send a fetch (if BlockedOnGA) and dump event to log file */
1285 /* whatever we schedule next, we must log that schedule */
1286 emitSchedule = rtsTrue;
1289 /* don't need to do anything. Either the thread is blocked on
1290 * I/O, in which case we'll have called addToBlockedQueue
1291 * previously, or it's blocked on an MVar or Blackhole, in which
1292 * case it'll be on the relevant queue already.
1295 fprintf(stderr, "--<< thread %d (%s) stopped: ",
1296 t->id, whatNext_strs[t->what_next]);
1297 printThreadBlockage(t);
1298 fprintf(stderr, "\n"));
1301 /* Only for dumping event to log file
1302 ToDo: do I need this in GranSim, too?
1309 case ThreadFinished:
1310 /* Need to check whether this was a main thread, and if so, signal
1311 * the task that started it with the return value. If we have no
1312 * more main threads, we probably need to stop all the tasks until
1315 /* We also end up here if the thread kills itself with an
1316 * uncaught exception, see Exception.hc.
1318 IF_DEBUG(scheduler,belch("--++ thread %d (%s) finished",
1319 t->id, whatNext_strs[t->what_next]));
1321 endThread(t, CurrentProc); // clean-up the thread
1323 /* For now all are advisory -- HWL */
1324 //if(t->priority==AdvisoryPriority) ??
1325 advisory_thread_count--;
1328 if(t->dist.priority==RevalPriority)
1332 if (RtsFlags.ParFlags.ParStats.Full &&
1333 !RtsFlags.ParFlags.ParStats.Suppressed)
1334 DumpEndEvent(CURRENT_PROC, t, rtsFalse /* not mandatory */);
1339 barf("schedule: invalid thread return code %d", (int)ret);
1343 // When we have +RTS -i0 and we're heap profiling, do a census at
1344 // every GC. This lets us get repeatable runs for debugging.
1345 if (performHeapProfile ||
1346 (RtsFlags.ProfFlags.profileInterval==0 &&
1347 RtsFlags.ProfFlags.doHeapProfile && ready_to_gc)) {
1348 GarbageCollect(GetRoots, rtsTrue);
1350 performHeapProfile = rtsFalse;
1351 ready_to_gc = rtsFalse; // we already GC'd
1356 /* everybody back, start the GC.
1357 * Could do it in this thread, or signal a condition var
1358 * to do it in another thread. Either way, we need to
1359 * broadcast on gc_pending_cond afterward.
1361 #if defined(RTS_SUPPORTS_THREADS)
1362 IF_DEBUG(scheduler,sched_belch("doing GC"));
1364 GarbageCollect(GetRoots,rtsFalse);
1365 ready_to_gc = rtsFalse;
1367 /* add a ContinueThread event to continue execution of current thread */
1368 new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
1370 t, (StgClosure*)NULL, (rtsSpark*)NULL);
1372 fprintf(stderr, "GRAN: eventq and runnableq after Garbage collection:\n");
1380 IF_GRAN_DEBUG(unused,
1381 print_eventq(EventHd));
1383 event = get_next_event();
1386 /* ToDo: wait for next message to arrive rather than busy wait */
1389 } /* end of while(1) */
1391 IF_PAR_DEBUG(verbose,
1392 belch("== Leaving schedule() after having received Finish"));
1395 /* ---------------------------------------------------------------------------
1396 * rtsSupportsBoundThreads(): is the RTS built to support bound threads?
1397 * used by Control.Concurrent for error checking.
1398 * ------------------------------------------------------------------------- */
1401 rtsSupportsBoundThreads(void)
1410 /* ---------------------------------------------------------------------------
1411 * isThreadBound(tso): check whether tso is bound to an OS thread.
1412 * ------------------------------------------------------------------------- */
1415 isThreadBound(StgTSO* tso USED_IN_THREADED_RTS)
1419 for(m = main_threads; m; m = m->link)
1428 /* ---------------------------------------------------------------------------
1429 * Singleton fork(). Do not copy any running threads.
1430 * ------------------------------------------------------------------------- */
1433 deleteThreadImmediately(StgTSO *tso);
1436 forkProcess(HsStablePtr *entry)
1438 #ifndef mingw32_TARGET_OS
1444 IF_DEBUG(scheduler,sched_belch("forking!"));
1445 rts_lock(); // This not only acquires sched_mutex, it also
1446 // makes sure that no other threads are running
1450 if (pid) { /* parent */
1452 /* just return the pid */
1456 } else { /* child */
1459 // delete all threads
1460 run_queue_hd = run_queue_tl = END_TSO_QUEUE;
1462 for (t = all_threads; t != END_TSO_QUEUE; t = next) {
1465 // don't allow threads to catch the ThreadKilled exception
1466 deleteThreadImmediately(t);
1469 // wipe the main thread list
1470 while((m = main_threads) != NULL) {
1471 main_threads = m->link;
1473 closeCondition(&m->bound_thread_cond);
1478 #ifdef RTS_SUPPORTS_THREADS
1479 resetTaskManagerAfterFork(); // tell startTask() and friends that
1480 startingWorkerThread = rtsFalse; // we have no worker threads any more
1481 resetWorkerWakeupPipeAfterFork();
1484 rc = rts_evalStableIO(entry, NULL); // run the action
1485 rts_checkSchedStatus("forkProcess",rc);
1489 hs_exit(); // clean up and exit
1493 barf("forkProcess#: primop not implemented for mingw32, sorry!\n");
1495 #endif /* mingw32 */
1498 /* ---------------------------------------------------------------------------
1499 * deleteAllThreads(): kill all the live threads.
1501 * This is used when we catch a user interrupt (^C), before performing
1502 * any necessary cleanups and running finalizers.
1504 * Locks: sched_mutex held.
1505 * ------------------------------------------------------------------------- */
1508 deleteAllThreads ( void )
1511 IF_DEBUG(scheduler,sched_belch("deleting all threads"));
1512 for (t = all_threads; t != END_TSO_QUEUE; t = next) {
1513 next = t->global_link;
1516 run_queue_hd = run_queue_tl = END_TSO_QUEUE;
1517 blocked_queue_hd = blocked_queue_tl = END_TSO_QUEUE;
1518 sleeping_queue = END_TSO_QUEUE;
1521 /* startThread and insertThread are now in GranSim.c -- HWL */
1524 /* ---------------------------------------------------------------------------
1525 * Suspending & resuming Haskell threads.
1527 * When making a "safe" call to C (aka _ccall_GC), the task gives back
1528 * its capability before calling the C function. This allows another
1529 * task to pick up the capability and carry on running Haskell
1530 * threads. It also means that if the C call blocks, it won't lock
1533 * The Haskell thread making the C call is put to sleep for the
1534 * duration of the call, on the susepended_ccalling_threads queue. We
1535 * give out a token to the task, which it can use to resume the thread
1536 * on return from the C function.
1537 * ------------------------------------------------------------------------- */
1540 suspendThread( StgRegTable *reg,
1549 int saved_errno = errno;
1551 /* assume that *reg is a pointer to the StgRegTable part
1554 cap = (Capability *)((void *)((unsigned char*)reg - sizeof(StgFunTable)));
1556 ACQUIRE_LOCK(&sched_mutex);
1559 sched_belch("thread %d did a _ccall_gc (is_concurrent: %d)", cap->r.rCurrentTSO->id,concCall));
1561 // XXX this might not be necessary --SDM
1562 cap->r.rCurrentTSO->what_next = ThreadRunGHC;
1564 threadPaused(cap->r.rCurrentTSO);
1565 cap->r.rCurrentTSO->link = suspended_ccalling_threads;
1566 suspended_ccalling_threads = cap->r.rCurrentTSO;
1568 #if defined(RTS_SUPPORTS_THREADS)
1569 if(cap->r.rCurrentTSO->blocked_exceptions == NULL)
1571 cap->r.rCurrentTSO->why_blocked = BlockedOnCCall;
1572 cap->r.rCurrentTSO->blocked_exceptions = END_TSO_QUEUE;
1576 cap->r.rCurrentTSO->why_blocked = BlockedOnCCall_NoUnblockExc;
1580 /* Use the thread ID as the token; it should be unique */
1581 tok = cap->r.rCurrentTSO->id;
1583 /* Hand back capability */
1584 releaseCapability(cap);
1586 #if defined(RTS_SUPPORTS_THREADS)
1587 /* Preparing to leave the RTS, so ensure there's a native thread/task
1588 waiting to take over.
1590 IF_DEBUG(scheduler, sched_belch("worker (token %d): leaving RTS", tok));
1593 /* Other threads _might_ be available for execution; signal this */
1595 RELEASE_LOCK(&sched_mutex);
1597 errno = saved_errno;
1602 resumeThread( StgInt tok,
1603 rtsBool concCall STG_UNUSED )
1605 StgTSO *tso, **prev;
1607 int saved_errno = errno;
1609 #if defined(RTS_SUPPORTS_THREADS)
1610 /* Wait for permission to re-enter the RTS with the result. */
1611 ACQUIRE_LOCK(&sched_mutex);
1612 waitForReturnCapability(&sched_mutex, &cap);
1614 IF_DEBUG(scheduler, sched_belch("worker (token %d): re-entering RTS", tok));
1616 grabCapability(&cap);
1619 /* Remove the thread off of the suspended list */
1620 prev = &suspended_ccalling_threads;
1621 for (tso = suspended_ccalling_threads;
1622 tso != END_TSO_QUEUE;
1623 prev = &tso->link, tso = tso->link) {
1624 if (tso->id == (StgThreadID)tok) {
1629 if (tso == END_TSO_QUEUE) {
1630 barf("resumeThread: thread not found");
1632 tso->link = END_TSO_QUEUE;
1634 #if defined(RTS_SUPPORTS_THREADS)
1635 if(tso->why_blocked == BlockedOnCCall)
1637 awakenBlockedQueueNoLock(tso->blocked_exceptions);
1638 tso->blocked_exceptions = NULL;
1642 /* Reset blocking status */
1643 tso->why_blocked = NotBlocked;
1645 cap->r.rCurrentTSO = tso;
1646 RELEASE_LOCK(&sched_mutex);
1647 errno = saved_errno;
1652 /* ---------------------------------------------------------------------------
1654 * ------------------------------------------------------------------------ */
1655 static void unblockThread(StgTSO *tso);
1657 /* ---------------------------------------------------------------------------
1658 * Comparing Thread ids.
1660 * This is used from STG land in the implementation of the
1661 * instances of Eq/Ord for ThreadIds.
1662 * ------------------------------------------------------------------------ */
1665 cmp_thread(StgPtr tso1, StgPtr tso2)
1667 StgThreadID id1 = ((StgTSO *)tso1)->id;
1668 StgThreadID id2 = ((StgTSO *)tso2)->id;
1670 if (id1 < id2) return (-1);
1671 if (id1 > id2) return 1;
1675 /* ---------------------------------------------------------------------------
1676 * Fetching the ThreadID from an StgTSO.
1678 * This is used in the implementation of Show for ThreadIds.
1679 * ------------------------------------------------------------------------ */
1681 rts_getThreadId(StgPtr tso)
1683 return ((StgTSO *)tso)->id;
1688 labelThread(StgPtr tso, char *label)
1693 /* Caveat: Once set, you can only set the thread name to "" */
1694 len = strlen(label)+1;
1695 buf = stgMallocBytes(len * sizeof(char), "Schedule.c:labelThread()");
1696 strncpy(buf,label,len);
1697 /* Update will free the old memory for us */
1698 updateThreadLabel(((StgTSO *)tso)->id,buf);
1702 /* ---------------------------------------------------------------------------
1703 Create a new thread.
1705 The new thread starts with the given stack size. Before the
1706 scheduler can run, however, this thread needs to have a closure
1707 (and possibly some arguments) pushed on its stack. See
1708 pushClosure() in Schedule.h.
1710 createGenThread() and createIOThread() (in SchedAPI.h) are
1711 convenient packaged versions of this function.
1713 currently pri (priority) is only used in a GRAN setup -- HWL
1714 ------------------------------------------------------------------------ */
1716 /* currently pri (priority) is only used in a GRAN setup -- HWL */
1718 createThread(nat size, StgInt pri)
1721 createThread(nat size)
1728 /* First check whether we should create a thread at all */
1730 /* check that no more than RtsFlags.ParFlags.maxThreads threads are created */
1731 if (advisory_thread_count >= RtsFlags.ParFlags.maxThreads) {
1733 belch("{createThread}Daq ghuH: refusing to create another thread; no more than %d threads allowed (currently %d)",
1734 RtsFlags.ParFlags.maxThreads, advisory_thread_count);
1735 return END_TSO_QUEUE;
1741 ASSERT(!RtsFlags.GranFlags.Light || CurrentProc==0);
1744 // ToDo: check whether size = stack_size - TSO_STRUCT_SIZEW
1746 /* catch ridiculously small stack sizes */
1747 if (size < MIN_STACK_WORDS + TSO_STRUCT_SIZEW) {
1748 size = MIN_STACK_WORDS + TSO_STRUCT_SIZEW;
1751 stack_size = size - TSO_STRUCT_SIZEW;
1753 tso = (StgTSO *)allocate(size);
1754 TICK_ALLOC_TSO(stack_size, 0);
1756 SET_HDR(tso, &stg_TSO_info, CCS_SYSTEM);
1758 SET_GRAN_HDR(tso, ThisPE);
1761 // Always start with the compiled code evaluator
1762 tso->what_next = ThreadRunGHC;
1764 /* tso->id needs to be unique. For now we use a heavyweight mutex to
1765 * protect the increment operation on next_thread_id.
1766 * In future, we could use an atomic increment instead.
1768 ACQUIRE_LOCK(&thread_id_mutex);
1769 tso->id = next_thread_id++;
1770 RELEASE_LOCK(&thread_id_mutex);
1772 tso->why_blocked = NotBlocked;
1773 tso->blocked_exceptions = NULL;
1775 tso->saved_errno = 0;
1777 tso->stack_size = stack_size;
1778 tso->max_stack_size = round_to_mblocks(RtsFlags.GcFlags.maxStkSize)
1780 tso->sp = (P_)&(tso->stack) + stack_size;
1783 tso->prof.CCCS = CCS_MAIN;
1786 /* put a stop frame on the stack */
1787 tso->sp -= sizeofW(StgStopFrame);
1788 SET_HDR((StgClosure*)tso->sp,(StgInfoTable *)&stg_stop_thread_info,CCS_SYSTEM);
1791 tso->link = END_TSO_QUEUE;
1792 /* uses more flexible routine in GranSim */
1793 insertThread(tso, CurrentProc);
1795 /* In a non-GranSim setup the pushing of a TSO onto the runq is separated
1801 if (RtsFlags.GranFlags.GranSimStats.Full)
1802 DumpGranEvent(GR_START,tso);
1804 if (RtsFlags.ParFlags.ParStats.Full)
1805 DumpGranEvent(GR_STARTQ,tso);
1806 /* HACk to avoid SCHEDULE
1810 /* Link the new thread on the global thread list.
1812 tso->global_link = all_threads;
1816 tso->dist.priority = MandatoryPriority; //by default that is...
1820 tso->gran.pri = pri;
1822 tso->gran.magic = TSO_MAGIC; // debugging only
1824 tso->gran.sparkname = 0;
1825 tso->gran.startedat = CURRENT_TIME;
1826 tso->gran.exported = 0;
1827 tso->gran.basicblocks = 0;
1828 tso->gran.allocs = 0;
1829 tso->gran.exectime = 0;
1830 tso->gran.fetchtime = 0;
1831 tso->gran.fetchcount = 0;
1832 tso->gran.blocktime = 0;
1833 tso->gran.blockcount = 0;
1834 tso->gran.blockedat = 0;
1835 tso->gran.globalsparks = 0;
1836 tso->gran.localsparks = 0;
1837 if (RtsFlags.GranFlags.Light)
1838 tso->gran.clock = Now; /* local clock */
1840 tso->gran.clock = 0;
1842 IF_DEBUG(gran,printTSO(tso));
1845 tso->par.magic = TSO_MAGIC; // debugging only
1847 tso->par.sparkname = 0;
1848 tso->par.startedat = CURRENT_TIME;
1849 tso->par.exported = 0;
1850 tso->par.basicblocks = 0;
1851 tso->par.allocs = 0;
1852 tso->par.exectime = 0;
1853 tso->par.fetchtime = 0;
1854 tso->par.fetchcount = 0;
1855 tso->par.blocktime = 0;
1856 tso->par.blockcount = 0;
1857 tso->par.blockedat = 0;
1858 tso->par.globalsparks = 0;
1859 tso->par.localsparks = 0;
1863 globalGranStats.tot_threads_created++;
1864 globalGranStats.threads_created_on_PE[CurrentProc]++;
1865 globalGranStats.tot_sq_len += spark_queue_len(CurrentProc);
1866 globalGranStats.tot_sq_probes++;
1868 // collect parallel global statistics (currently done together with GC stats)
1869 if (RtsFlags.ParFlags.ParStats.Global &&
1870 RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
1871 //fprintf(stderr, "Creating thread %d @ %11.2f\n", tso->id, usertime());
1872 globalParStats.tot_threads_created++;
1878 belch("==__ schedule: Created TSO %d (%p);",
1879 CurrentProc, tso, tso->id));
1881 IF_PAR_DEBUG(verbose,
1882 belch("==__ schedule: Created TSO %d (%p); %d threads active",
1883 tso->id, tso, advisory_thread_count));
1885 IF_DEBUG(scheduler,sched_belch("created thread %ld, stack size = %lx words",
1886 tso->id, tso->stack_size));
1893 all parallel thread creation calls should fall through the following routine.
1896 createSparkThread(rtsSpark spark)
1898 ASSERT(spark != (rtsSpark)NULL);
1899 if (advisory_thread_count >= RtsFlags.ParFlags.maxThreads)
1901 barf("{createSparkThread}Daq ghuH: refusing to create another thread; no more than %d threads allowed (currently %d)",
1902 RtsFlags.ParFlags.maxThreads, advisory_thread_count);
1903 return END_TSO_QUEUE;
1907 tso = createThread(RtsFlags.GcFlags.initialStkSize);
1908 if (tso==END_TSO_QUEUE)
1909 barf("createSparkThread: Cannot create TSO");
1911 tso->priority = AdvisoryPriority;
1913 pushClosure(tso,spark);
1914 PUSH_ON_RUN_QUEUE(tso);
1915 advisory_thread_count++;
1922 Turn a spark into a thread.
1923 ToDo: fix for SMP (needs to acquire SCHED_MUTEX!)
1927 activateSpark (rtsSpark spark)
1931 tso = createSparkThread(spark);
1932 if (RtsFlags.ParFlags.ParStats.Full) {
1933 //ASSERT(run_queue_hd == END_TSO_QUEUE); // I think ...
1934 IF_PAR_DEBUG(verbose,
1935 belch("==^^ activateSpark: turning spark of closure %p (%s) into a thread",
1936 (StgClosure *)spark, info_type((StgClosure *)spark)));
1938 // ToDo: fwd info on local/global spark to thread -- HWL
1939 // tso->gran.exported = spark->exported;
1940 // tso->gran.locked = !spark->global;
1941 // tso->gran.sparkname = spark->name;
1947 static SchedulerStatus waitThread_(/*out*/StgMainThread* m,
1948 Capability *initialCapability
1952 /* ---------------------------------------------------------------------------
1955 * scheduleThread puts a thread on the head of the runnable queue.
1956 * This will usually be done immediately after a thread is created.
1957 * The caller of scheduleThread must create the thread using e.g.
1958 * createThread and push an appropriate closure
1959 * on this thread's stack before the scheduler is invoked.
1960 * ------------------------------------------------------------------------ */
1962 static void scheduleThread_ (StgTSO* tso);
1965 scheduleThread_(StgTSO *tso)
1967 // Precondition: sched_mutex must be held.
1969 /* Put the new thread on the head of the runnable queue. The caller
1970 * better push an appropriate closure on this thread's stack
1971 * beforehand. In the SMP case, the thread may start running as
1972 * soon as we release the scheduler lock below.
1974 PUSH_ON_RUN_QUEUE(tso);
1978 IF_DEBUG(scheduler,printTSO(tso));
1982 void scheduleThread(StgTSO* tso)
1984 ACQUIRE_LOCK(&sched_mutex);
1985 scheduleThread_(tso);
1986 RELEASE_LOCK(&sched_mutex);
1990 scheduleWaitThread(StgTSO* tso, /*[out]*/HaskellObj* ret,
1991 Capability *initialCapability)
1993 // Precondition: sched_mutex must be held
1996 m = stgMallocBytes(sizeof(StgMainThread), "waitThread");
2000 #if defined(RTS_SUPPORTS_THREADS)
2001 initCondition(&m->bound_thread_cond);
2004 /* Put the thread on the main-threads list prior to scheduling the TSO.
2005 Failure to do so introduces a race condition in the MT case (as
2006 identified by Wolfgang Thaller), whereby the new task/OS thread
2007 created by scheduleThread_() would complete prior to the thread
2008 that spawned it managed to put 'itself' on the main-threads list.
2009 The upshot of it all being that the worker thread wouldn't get to
2010 signal the completion of the its work item for the main thread to
2011 see (==> it got stuck waiting.) -- sof 6/02.
2013 IF_DEBUG(scheduler, sched_belch("waiting for thread (%d)", tso->id));
2015 m->link = main_threads;
2018 scheduleThread_(tso);
2020 return waitThread_(m, initialCapability);
2023 /* ---------------------------------------------------------------------------
2026 * Initialise the scheduler. This resets all the queues - if the
2027 * queues contained any threads, they'll be garbage collected at the
2030 * ------------------------------------------------------------------------ */
2038 for (i=0; i<=MAX_PROC; i++) {
2039 run_queue_hds[i] = END_TSO_QUEUE;
2040 run_queue_tls[i] = END_TSO_QUEUE;
2041 blocked_queue_hds[i] = END_TSO_QUEUE;
2042 blocked_queue_tls[i] = END_TSO_QUEUE;
2043 ccalling_threadss[i] = END_TSO_QUEUE;
2044 sleeping_queue = END_TSO_QUEUE;
2047 run_queue_hd = END_TSO_QUEUE;
2048 run_queue_tl = END_TSO_QUEUE;
2049 blocked_queue_hd = END_TSO_QUEUE;
2050 blocked_queue_tl = END_TSO_QUEUE;
2051 sleeping_queue = END_TSO_QUEUE;
2054 suspended_ccalling_threads = END_TSO_QUEUE;
2056 main_threads = NULL;
2057 all_threads = END_TSO_QUEUE;
2062 RtsFlags.ConcFlags.ctxtSwitchTicks =
2063 RtsFlags.ConcFlags.ctxtSwitchTime / TICK_MILLISECS;
2065 #if defined(RTS_SUPPORTS_THREADS)
2066 /* Initialise the mutex and condition variables used by
2068 initMutex(&sched_mutex);
2069 initMutex(&term_mutex);
2070 initMutex(&thread_id_mutex);
2072 initCondition(&thread_ready_cond);
2075 #if defined(RTS_SUPPORTS_THREADS)
2076 ACQUIRE_LOCK(&sched_mutex);
2079 /* A capability holds the state a native thread needs in
2080 * order to execute STG code. At least one capability is
2081 * floating around (only SMP builds have more than one).
2085 #if defined(RTS_SUPPORTS_THREADS)
2086 /* start our haskell execution tasks */
2087 startTaskManager(0,taskStart);
2090 #if /* defined(SMP) ||*/ defined(PAR)
2094 RELEASE_LOCK(&sched_mutex);
2099 exitScheduler( void )
2101 #if defined(RTS_SUPPORTS_THREADS)
2104 shutting_down_scheduler = rtsTrue;
2107 /* ----------------------------------------------------------------------------
2108 Managing the per-task allocation areas.
2110 Each capability comes with an allocation area. These are
2111 fixed-length block lists into which allocation can be done.
2113 ToDo: no support for two-space collection at the moment???
2114 ------------------------------------------------------------------------- */
2118 waitThread_(StgMainThread* m, Capability *initialCapability)
2120 SchedulerStatus stat;
2122 // Precondition: sched_mutex must be held.
2123 IF_DEBUG(scheduler, sched_belch("new main thread (%d)", m->tso->id));
2126 /* GranSim specific init */
2127 CurrentTSO = m->tso; // the TSO to run
2128 procStatus[MainProc] = Busy; // status of main PE
2129 CurrentProc = MainProc; // PE to run it on
2130 schedule(m,initialCapability);
2132 schedule(m,initialCapability);
2133 ASSERT(m->stat != NoStatus);
2138 #if defined(RTS_SUPPORTS_THREADS)
2139 closeCondition(&m->bound_thread_cond);
2142 IF_DEBUG(scheduler, sched_belch("main thread (%d) finished", m->tso->id));
2145 // Postcondition: sched_mutex still held
2149 /* ---------------------------------------------------------------------------
2150 Where are the roots that we know about?
2152 - all the threads on the runnable queue
2153 - all the threads on the blocked queue
2154 - all the threads on the sleeping queue
2155 - all the thread currently executing a _ccall_GC
2156 - all the "main threads"
2158 ------------------------------------------------------------------------ */
2160 /* This has to be protected either by the scheduler monitor, or by the
2161 garbage collection monitor (probably the latter).
2166 GetRoots( evac_fn evac )
2171 for (i=0; i<=RtsFlags.GranFlags.proc; i++) {
2172 if ((run_queue_hds[i] != END_TSO_QUEUE) && ((run_queue_hds[i] != NULL)))
2173 evac((StgClosure **)&run_queue_hds[i]);
2174 if ((run_queue_tls[i] != END_TSO_QUEUE) && ((run_queue_tls[i] != NULL)))
2175 evac((StgClosure **)&run_queue_tls[i]);
2177 if ((blocked_queue_hds[i] != END_TSO_QUEUE) && ((blocked_queue_hds[i] != NULL)))
2178 evac((StgClosure **)&blocked_queue_hds[i]);
2179 if ((blocked_queue_tls[i] != END_TSO_QUEUE) && ((blocked_queue_tls[i] != NULL)))
2180 evac((StgClosure **)&blocked_queue_tls[i]);
2181 if ((ccalling_threadss[i] != END_TSO_QUEUE) && ((ccalling_threadss[i] != NULL)))
2182 evac((StgClosure **)&ccalling_threads[i]);
2189 if (run_queue_hd != END_TSO_QUEUE) {
2190 ASSERT(run_queue_tl != END_TSO_QUEUE);
2191 evac((StgClosure **)&run_queue_hd);
2192 evac((StgClosure **)&run_queue_tl);
2195 if (blocked_queue_hd != END_TSO_QUEUE) {
2196 ASSERT(blocked_queue_tl != END_TSO_QUEUE);
2197 evac((StgClosure **)&blocked_queue_hd);
2198 evac((StgClosure **)&blocked_queue_tl);
2201 if (sleeping_queue != END_TSO_QUEUE) {
2202 evac((StgClosure **)&sleeping_queue);
2206 if (suspended_ccalling_threads != END_TSO_QUEUE) {
2207 evac((StgClosure **)&suspended_ccalling_threads);
2210 #if defined(PAR) || defined(GRAN)
2211 markSparkQueue(evac);
2214 #if defined(RTS_USER_SIGNALS)
2215 // mark the signal handlers (signals should be already blocked)
2216 markSignalHandlers(evac);
2219 // main threads which have completed need to be retained until they
2220 // are dealt with in the main scheduler loop. They won't be
2221 // retained any other way: the GC will drop them from the
2222 // all_threads list, so we have to be careful to treat them as roots
2226 for (m = main_threads; m != NULL; m = m->link) {
2227 switch (m->tso->what_next) {
2228 case ThreadComplete:
2230 evac((StgClosure **)&m->tso);
2239 /* -----------------------------------------------------------------------------
2242 This is the interface to the garbage collector from Haskell land.
2243 We provide this so that external C code can allocate and garbage
2244 collect when called from Haskell via _ccall_GC.
2246 It might be useful to provide an interface whereby the programmer
2247 can specify more roots (ToDo).
2249 This needs to be protected by the GC condition variable above. KH.
2250 -------------------------------------------------------------------------- */
2252 static void (*extra_roots)(evac_fn);
2257 /* Obligated to hold this lock upon entry */
2258 ACQUIRE_LOCK(&sched_mutex);
2259 GarbageCollect(GetRoots,rtsFalse);
2260 RELEASE_LOCK(&sched_mutex);
2264 performMajorGC(void)
2266 ACQUIRE_LOCK(&sched_mutex);
2267 GarbageCollect(GetRoots,rtsTrue);
2268 RELEASE_LOCK(&sched_mutex);
2272 AllRoots(evac_fn evac)
2274 GetRoots(evac); // the scheduler's roots
2275 extra_roots(evac); // the user's roots
2279 performGCWithRoots(void (*get_roots)(evac_fn))
2281 ACQUIRE_LOCK(&sched_mutex);
2282 extra_roots = get_roots;
2283 GarbageCollect(AllRoots,rtsFalse);
2284 RELEASE_LOCK(&sched_mutex);
2287 /* -----------------------------------------------------------------------------
2290 If the thread has reached its maximum stack size, then raise the
2291 StackOverflow exception in the offending thread. Otherwise
2292 relocate the TSO into a larger chunk of memory and adjust its stack
2294 -------------------------------------------------------------------------- */
2297 threadStackOverflow(StgTSO *tso)
2299 nat new_stack_size, new_tso_size, stack_words;
2303 IF_DEBUG(sanity,checkTSO(tso));
2304 if (tso->stack_size >= tso->max_stack_size) {
2307 belch("@@ threadStackOverflow of TSO %d (%p): stack too large (now %ld; max is %ld)",
2308 tso->id, tso, tso->stack_size, tso->max_stack_size);
2309 /* If we're debugging, just print out the top of the stack */
2310 printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size,
2313 /* Send this thread the StackOverflow exception */
2314 raiseAsync(tso, (StgClosure *)stackOverflow_closure);
2318 /* Try to double the current stack size. If that takes us over the
2319 * maximum stack size for this thread, then use the maximum instead.
2320 * Finally round up so the TSO ends up as a whole number of blocks.
2322 new_stack_size = stg_min(tso->stack_size * 2, tso->max_stack_size);
2323 new_tso_size = (nat)BLOCK_ROUND_UP(new_stack_size * sizeof(W_) +
2324 TSO_STRUCT_SIZE)/sizeof(W_);
2325 new_tso_size = round_to_mblocks(new_tso_size); /* Be MBLOCK-friendly */
2326 new_stack_size = new_tso_size - TSO_STRUCT_SIZEW;
2328 IF_DEBUG(scheduler, fprintf(stderr,"== sched: increasing stack size from %d words to %d.\n", tso->stack_size, new_stack_size));
2330 dest = (StgTSO *)allocate(new_tso_size);
2331 TICK_ALLOC_TSO(new_stack_size,0);
2333 /* copy the TSO block and the old stack into the new area */
2334 memcpy(dest,tso,TSO_STRUCT_SIZE);
2335 stack_words = tso->stack + tso->stack_size - tso->sp;
2336 new_sp = (P_)dest + new_tso_size - stack_words;
2337 memcpy(new_sp, tso->sp, stack_words * sizeof(W_));
2339 /* relocate the stack pointers... */
2341 dest->stack_size = new_stack_size;
2343 /* Mark the old TSO as relocated. We have to check for relocated
2344 * TSOs in the garbage collector and any primops that deal with TSOs.
2346 * It's important to set the sp value to just beyond the end
2347 * of the stack, so we don't attempt to scavenge any part of the
2350 tso->what_next = ThreadRelocated;
2352 tso->sp = (P_)&(tso->stack[tso->stack_size]);
2353 tso->why_blocked = NotBlocked;
2354 dest->mut_link = NULL;
2356 IF_PAR_DEBUG(verbose,
2357 belch("@@ threadStackOverflow of TSO %d (now at %p): stack size increased to %ld",
2358 tso->id, tso, tso->stack_size);
2359 /* If we're debugging, just print out the top of the stack */
2360 printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size,
2363 IF_DEBUG(sanity,checkTSO(tso));
2365 IF_DEBUG(scheduler,printTSO(dest));
2371 /* ---------------------------------------------------------------------------
2372 Wake up a queue that was blocked on some resource.
2373 ------------------------------------------------------------------------ */
2377 unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node )
2382 unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node )
2384 /* write RESUME events to log file and
2385 update blocked and fetch time (depending on type of the orig closure) */
2386 if (RtsFlags.ParFlags.ParStats.Full) {
2387 DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
2388 GR_RESUMEQ, ((StgTSO *)bqe), ((StgTSO *)bqe)->block_info.closure,
2389 0, 0 /* spark_queue_len(ADVISORY_POOL) */);
2390 if (EMPTY_RUN_QUEUE())
2391 emitSchedule = rtsTrue;
2393 switch (get_itbl(node)->type) {
2395 ((StgTSO *)bqe)->par.fetchtime += CURRENT_TIME-((StgTSO *)bqe)->par.blockedat;
2400 ((StgTSO *)bqe)->par.blocktime += CURRENT_TIME-((StgTSO *)bqe)->par.blockedat;
2407 barf("{unblockOneLocked}Daq Qagh: unexpected closure in blocking queue");
2414 static StgBlockingQueueElement *
2415 unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node)
2418 PEs node_loc, tso_loc;
2420 node_loc = where_is(node); // should be lifted out of loop
2421 tso = (StgTSO *)bqe; // wastes an assignment to get the type right
2422 tso_loc = where_is((StgClosure *)tso);
2423 if (IS_LOCAL_TO(PROCS(node),tso_loc)) { // TSO is local
2424 /* !fake_fetch => TSO is on CurrentProc is same as IS_LOCAL_TO */
2425 ASSERT(CurrentProc!=node_loc || tso_loc==CurrentProc);
2426 CurrentTime[CurrentProc] += RtsFlags.GranFlags.Costs.lunblocktime;
2427 // insertThread(tso, node_loc);
2428 new_event(tso_loc, tso_loc, CurrentTime[CurrentProc],
2430 tso, node, (rtsSpark*)NULL);
2431 tso->link = END_TSO_QUEUE; // overwrite link just to be sure
2434 } else { // TSO is remote (actually should be FMBQ)
2435 CurrentTime[CurrentProc] += RtsFlags.GranFlags.Costs.mpacktime +
2436 RtsFlags.GranFlags.Costs.gunblocktime +
2437 RtsFlags.GranFlags.Costs.latency;
2438 new_event(tso_loc, CurrentProc, CurrentTime[CurrentProc],
2440 tso, node, (rtsSpark*)NULL);
2441 tso->link = END_TSO_QUEUE; // overwrite link just to be sure
2444 /* the thread-queue-overhead is accounted for in either Resume or UnblockThread */
2446 fprintf(stderr," %s TSO %d (%p) [PE %d] (block_info.closure=%p) (next=%p) ,",
2447 (node_loc==tso_loc ? "Local" : "Global"),
2448 tso->id, tso, CurrentProc, tso->block_info.closure, tso->link));
2449 tso->block_info.closure = NULL;
2450 IF_DEBUG(scheduler,belch("-- Waking up thread %ld (%p)",
2454 static StgBlockingQueueElement *
2455 unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node)
2457 StgBlockingQueueElement *next;
2459 switch (get_itbl(bqe)->type) {
2461 ASSERT(((StgTSO *)bqe)->why_blocked != NotBlocked);
2462 /* if it's a TSO just push it onto the run_queue */
2464 // ((StgTSO *)bqe)->link = END_TSO_QUEUE; // debugging?
2465 PUSH_ON_RUN_QUEUE((StgTSO *)bqe);
2467 unblockCount(bqe, node);
2468 /* reset blocking status after dumping event */
2469 ((StgTSO *)bqe)->why_blocked = NotBlocked;
2473 /* if it's a BLOCKED_FETCH put it on the PendingFetches list */
2475 bqe->link = (StgBlockingQueueElement *)PendingFetches;
2476 PendingFetches = (StgBlockedFetch *)bqe;
2480 /* can ignore this case in a non-debugging setup;
2481 see comments on RBHSave closures above */
2483 /* check that the closure is an RBHSave closure */
2484 ASSERT(get_itbl((StgClosure *)bqe) == &stg_RBH_Save_0_info ||
2485 get_itbl((StgClosure *)bqe) == &stg_RBH_Save_1_info ||
2486 get_itbl((StgClosure *)bqe) == &stg_RBH_Save_2_info);
2490 barf("{unblockOneLocked}Daq Qagh: Unexpected IP (%#lx; %s) in blocking queue at %#lx\n",
2491 get_itbl((StgClosure *)bqe), info_type((StgClosure *)bqe),
2495 IF_PAR_DEBUG(bq, fprintf(stderr, ", %p (%s)", bqe, info_type((StgClosure*)bqe)));
2499 #else /* !GRAN && !PAR */
2501 unblockOneLocked(StgTSO *tso)
2505 ASSERT(get_itbl(tso)->type == TSO);
2506 ASSERT(tso->why_blocked != NotBlocked);
2507 tso->why_blocked = NotBlocked;
2509 PUSH_ON_RUN_QUEUE(tso);
2511 IF_DEBUG(scheduler,sched_belch("waking up thread %ld", tso->id));
2516 #if defined(GRAN) || defined(PAR)
2517 INLINE_ME StgBlockingQueueElement *
2518 unblockOne(StgBlockingQueueElement *bqe, StgClosure *node)
2520 ACQUIRE_LOCK(&sched_mutex);
2521 bqe = unblockOneLocked(bqe, node);
2522 RELEASE_LOCK(&sched_mutex);
2527 unblockOne(StgTSO *tso)
2529 ACQUIRE_LOCK(&sched_mutex);
2530 tso = unblockOneLocked(tso);
2531 RELEASE_LOCK(&sched_mutex);
2538 awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node)
2540 StgBlockingQueueElement *bqe;
2545 belch("##-_ AwBQ for node %p on PE %d @ %ld by TSO %d (%p): ", \
2546 node, CurrentProc, CurrentTime[CurrentProc],
2547 CurrentTSO->id, CurrentTSO));
2549 node_loc = where_is(node);
2551 ASSERT(q == END_BQ_QUEUE ||
2552 get_itbl(q)->type == TSO || // q is either a TSO or an RBHSave
2553 get_itbl(q)->type == CONSTR); // closure (type constructor)
2554 ASSERT(is_unique(node));
2556 /* FAKE FETCH: magically copy the node to the tso's proc;
2557 no Fetch necessary because in reality the node should not have been
2558 moved to the other PE in the first place
2560 if (CurrentProc!=node_loc) {
2562 belch("## node %p is on PE %d but CurrentProc is %d (TSO %d); assuming fake fetch and adjusting bitmask (old: %#x)",
2563 node, node_loc, CurrentProc, CurrentTSO->id,
2564 // CurrentTSO, where_is(CurrentTSO),
2565 node->header.gran.procs));
2566 node->header.gran.procs = (node->header.gran.procs) | PE_NUMBER(CurrentProc);
2568 belch("## new bitmask of node %p is %#x",
2569 node, node->header.gran.procs));
2570 if (RtsFlags.GranFlags.GranSimStats.Global) {
2571 globalGranStats.tot_fake_fetches++;
2576 // ToDo: check: ASSERT(CurrentProc==node_loc);
2577 while (get_itbl(bqe)->type==TSO) { // q != END_TSO_QUEUE) {
2580 bqe points to the current element in the queue
2581 next points to the next element in the queue
2583 //tso = (StgTSO *)bqe; // wastes an assignment to get the type right
2584 //tso_loc = where_is(tso);
2586 bqe = unblockOneLocked(bqe, node);
2589 /* if this is the BQ of an RBH, we have to put back the info ripped out of
2590 the closure to make room for the anchor of the BQ */
2591 if (bqe!=END_BQ_QUEUE) {
2592 ASSERT(get_itbl(node)->type == RBH && get_itbl(bqe)->type == CONSTR);
2594 ASSERT((info_ptr==&RBH_Save_0_info) ||
2595 (info_ptr==&RBH_Save_1_info) ||
2596 (info_ptr==&RBH_Save_2_info));
2598 /* cf. convertToRBH in RBH.c for writing the RBHSave closure */
2599 ((StgRBH *)node)->blocking_queue = (StgBlockingQueueElement *)((StgRBHSave *)bqe)->payload[0];
2600 ((StgRBH *)node)->mut_link = (StgMutClosure *)((StgRBHSave *)bqe)->payload[1];
2603 belch("## Filled in RBH_Save for %p (%s) at end of AwBQ",
2604 node, info_type(node)));
2607 /* statistics gathering */
2608 if (RtsFlags.GranFlags.GranSimStats.Global) {
2609 // globalGranStats.tot_bq_processing_time += bq_processing_time;
2610 globalGranStats.tot_bq_len += len; // total length of all bqs awakened
2611 // globalGranStats.tot_bq_len_local += len_local; // same for local TSOs only
2612 globalGranStats.tot_awbq++; // total no. of bqs awakened
2615 fprintf(stderr,"## BQ Stats of %p: [%d entries] %s\n",
2616 node, len, (bqe!=END_BQ_QUEUE) ? "RBH" : ""));
2620 awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node)
2622 StgBlockingQueueElement *bqe;
2624 ACQUIRE_LOCK(&sched_mutex);
2626 IF_PAR_DEBUG(verbose,
2627 belch("##-_ AwBQ for node %p on [%x]: ",
2631 if(get_itbl(q)->type == CONSTR || q==END_BQ_QUEUE) {
2632 IF_PAR_DEBUG(verbose, belch("## ... nothing to unblock so lets just return. RFP (BUG?)"));
2637 ASSERT(q == END_BQ_QUEUE ||
2638 get_itbl(q)->type == TSO ||
2639 get_itbl(q)->type == BLOCKED_FETCH ||
2640 get_itbl(q)->type == CONSTR);
2643 while (get_itbl(bqe)->type==TSO ||
2644 get_itbl(bqe)->type==BLOCKED_FETCH) {
2645 bqe = unblockOneLocked(bqe, node);
2647 RELEASE_LOCK(&sched_mutex);
2650 #else /* !GRAN && !PAR */
2652 #ifdef RTS_SUPPORTS_THREADS
2654 awakenBlockedQueueNoLock(StgTSO *tso)
2656 while (tso != END_TSO_QUEUE) {
2657 tso = unblockOneLocked(tso);
2663 awakenBlockedQueue(StgTSO *tso)
2665 ACQUIRE_LOCK(&sched_mutex);
2666 while (tso != END_TSO_QUEUE) {
2667 tso = unblockOneLocked(tso);
2669 RELEASE_LOCK(&sched_mutex);
2673 /* ---------------------------------------------------------------------------
2675 - usually called inside a signal handler so it mustn't do anything fancy.
2676 ------------------------------------------------------------------------ */
2679 interruptStgRts(void)
2683 #ifdef RTS_SUPPORTS_THREADS
2684 wakeBlockedWorkerThread();
2688 /* -----------------------------------------------------------------------------
2691 This is for use when we raise an exception in another thread, which
2693 This has nothing to do with the UnblockThread event in GranSim. -- HWL
2694 -------------------------------------------------------------------------- */
2696 #if defined(GRAN) || defined(PAR)
2698 NB: only the type of the blocking queue is different in GranSim and GUM
2699 the operations on the queue-elements are the same
2700 long live polymorphism!
2702 Locks: sched_mutex is held upon entry and exit.
2706 unblockThread(StgTSO *tso)
2708 StgBlockingQueueElement *t, **last;
2710 switch (tso->why_blocked) {
2713 return; /* not blocked */
2716 ASSERT(get_itbl(tso->block_info.closure)->type == MVAR);
2718 StgBlockingQueueElement *last_tso = END_BQ_QUEUE;
2719 StgMVar *mvar = (StgMVar *)(tso->block_info.closure);
2721 last = (StgBlockingQueueElement **)&mvar->head;
2722 for (t = (StgBlockingQueueElement *)mvar->head;
2724 last = &t->link, last_tso = t, t = t->link) {
2725 if (t == (StgBlockingQueueElement *)tso) {
2726 *last = (StgBlockingQueueElement *)tso->link;
2727 if (mvar->tail == tso) {
2728 mvar->tail = (StgTSO *)last_tso;
2733 barf("unblockThread (MVAR): TSO not found");
2736 case BlockedOnBlackHole:
2737 ASSERT(get_itbl(tso->block_info.closure)->type == BLACKHOLE_BQ);
2739 StgBlockingQueue *bq = (StgBlockingQueue *)(tso->block_info.closure);
2741 last = &bq->blocking_queue;
2742 for (t = bq->blocking_queue;
2744 last = &t->link, t = t->link) {
2745 if (t == (StgBlockingQueueElement *)tso) {
2746 *last = (StgBlockingQueueElement *)tso->link;
2750 barf("unblockThread (BLACKHOLE): TSO not found");
2753 case BlockedOnException:
2755 StgTSO *target = tso->block_info.tso;
2757 ASSERT(get_itbl(target)->type == TSO);
2759 if (target->what_next == ThreadRelocated) {
2760 target = target->link;
2761 ASSERT(get_itbl(target)->type == TSO);
2764 ASSERT(target->blocked_exceptions != NULL);
2766 last = (StgBlockingQueueElement **)&target->blocked_exceptions;
2767 for (t = (StgBlockingQueueElement *)target->blocked_exceptions;
2769 last = &t->link, t = t->link) {
2770 ASSERT(get_itbl(t)->type == TSO);
2771 if (t == (StgBlockingQueueElement *)tso) {
2772 *last = (StgBlockingQueueElement *)tso->link;
2776 barf("unblockThread (Exception): TSO not found");
2780 case BlockedOnWrite:
2781 #if defined(mingw32_TARGET_OS)
2782 case BlockedOnDoProc:
2785 /* take TSO off blocked_queue */
2786 StgBlockingQueueElement *prev = NULL;
2787 for (t = (StgBlockingQueueElement *)blocked_queue_hd; t != END_BQ_QUEUE;
2788 prev = t, t = t->link) {
2789 if (t == (StgBlockingQueueElement *)tso) {
2791 blocked_queue_hd = (StgTSO *)t->link;
2792 if ((StgBlockingQueueElement *)blocked_queue_tl == t) {
2793 blocked_queue_tl = END_TSO_QUEUE;
2796 prev->link = t->link;
2797 if ((StgBlockingQueueElement *)blocked_queue_tl == t) {
2798 blocked_queue_tl = (StgTSO *)prev;
2804 barf("unblockThread (I/O): TSO not found");
2807 case BlockedOnDelay:
2809 /* take TSO off sleeping_queue */
2810 StgBlockingQueueElement *prev = NULL;
2811 for (t = (StgBlockingQueueElement *)sleeping_queue; t != END_BQ_QUEUE;
2812 prev = t, t = t->link) {
2813 if (t == (StgBlockingQueueElement *)tso) {
2815 sleeping_queue = (StgTSO *)t->link;
2817 prev->link = t->link;
2822 barf("unblockThread (delay): TSO not found");
2826 barf("unblockThread");
2830 tso->link = END_TSO_QUEUE;
2831 tso->why_blocked = NotBlocked;
2832 tso->block_info.closure = NULL;
2833 PUSH_ON_RUN_QUEUE(tso);
2837 unblockThread(StgTSO *tso)
2841 /* To avoid locking unnecessarily. */
2842 if (tso->why_blocked == NotBlocked) {
2846 switch (tso->why_blocked) {
2849 ASSERT(get_itbl(tso->block_info.closure)->type == MVAR);
2851 StgTSO *last_tso = END_TSO_QUEUE;
2852 StgMVar *mvar = (StgMVar *)(tso->block_info.closure);
2855 for (t = mvar->head; t != END_TSO_QUEUE;
2856 last = &t->link, last_tso = t, t = t->link) {
2859 if (mvar->tail == tso) {
2860 mvar->tail = last_tso;
2865 barf("unblockThread (MVAR): TSO not found");
2868 case BlockedOnBlackHole:
2869 ASSERT(get_itbl(tso->block_info.closure)->type == BLACKHOLE_BQ);
2871 StgBlockingQueue *bq = (StgBlockingQueue *)(tso->block_info.closure);
2873 last = &bq->blocking_queue;
2874 for (t = bq->blocking_queue; t != END_TSO_QUEUE;
2875 last = &t->link, t = t->link) {
2881 barf("unblockThread (BLACKHOLE): TSO not found");
2884 case BlockedOnException:
2886 StgTSO *target = tso->block_info.tso;
2888 ASSERT(get_itbl(target)->type == TSO);
2890 while (target->what_next == ThreadRelocated) {
2891 target = target->link;
2892 ASSERT(get_itbl(target)->type == TSO);
2895 ASSERT(target->blocked_exceptions != NULL);
2897 last = &target->blocked_exceptions;
2898 for (t = target->blocked_exceptions; t != END_TSO_QUEUE;
2899 last = &t->link, t = t->link) {
2900 ASSERT(get_itbl(t)->type == TSO);
2906 barf("unblockThread (Exception): TSO not found");
2910 case BlockedOnWrite:
2911 #if defined(mingw32_TARGET_OS)
2912 case BlockedOnDoProc:
2915 StgTSO *prev = NULL;
2916 for (t = blocked_queue_hd; t != END_TSO_QUEUE;
2917 prev = t, t = t->link) {
2920 blocked_queue_hd = t->link;
2921 if (blocked_queue_tl == t) {
2922 blocked_queue_tl = END_TSO_QUEUE;
2925 prev->link = t->link;
2926 if (blocked_queue_tl == t) {
2927 blocked_queue_tl = prev;
2933 barf("unblockThread (I/O): TSO not found");
2936 case BlockedOnDelay:
2938 StgTSO *prev = NULL;
2939 for (t = sleeping_queue; t != END_TSO_QUEUE;
2940 prev = t, t = t->link) {
2943 sleeping_queue = t->link;
2945 prev->link = t->link;
2950 barf("unblockThread (delay): TSO not found");
2954 barf("unblockThread");
2958 tso->link = END_TSO_QUEUE;
2959 tso->why_blocked = NotBlocked;
2960 tso->block_info.closure = NULL;
2961 PUSH_ON_RUN_QUEUE(tso);
2965 /* -----------------------------------------------------------------------------
2968 * The following function implements the magic for raising an
2969 * asynchronous exception in an existing thread.
2971 * We first remove the thread from any queue on which it might be
2972 * blocked. The possible blockages are MVARs and BLACKHOLE_BQs.
2974 * We strip the stack down to the innermost CATCH_FRAME, building
2975 * thunks in the heap for all the active computations, so they can
2976 * be restarted if necessary. When we reach a CATCH_FRAME, we build
2977 * an application of the handler to the exception, and push it on
2978 * the top of the stack.
2980 * How exactly do we save all the active computations? We create an
2981 * AP_STACK for every UpdateFrame on the stack. Entering one of these
2982 * AP_STACKs pushes everything from the corresponding update frame
2983 * upwards onto the stack. (Actually, it pushes everything up to the
2984 * next update frame plus a pointer to the next AP_STACK object.
2985 * Entering the next AP_STACK object pushes more onto the stack until we
2986 * reach the last AP_STACK object - at which point the stack should look
2987 * exactly as it did when we killed the TSO and we can continue
2988 * execution by entering the closure on top of the stack.
2990 * We can also kill a thread entirely - this happens if either (a) the
2991 * exception passed to raiseAsync is NULL, or (b) there's no
2992 * CATCH_FRAME on the stack. In either case, we strip the entire
2993 * stack and replace the thread with a zombie.
2995 * Locks: sched_mutex held upon entry nor exit.
2997 * -------------------------------------------------------------------------- */
3000 deleteThread(StgTSO *tso)
3002 raiseAsync(tso,NULL);
3006 deleteThreadImmediately(StgTSO *tso)
3007 { // for forkProcess only:
3008 // delete thread without giving it a chance to catch the KillThread exception
3010 if (tso->what_next == ThreadComplete || tso->what_next == ThreadKilled) {
3013 #if defined(RTS_SUPPORTS_THREADS)
3014 if (tso->why_blocked != BlockedOnCCall
3015 && tso->why_blocked != BlockedOnCCall_NoUnblockExc)
3018 tso->what_next = ThreadKilled;
3022 raiseAsyncWithLock(StgTSO *tso, StgClosure *exception)
3024 /* When raising async exs from contexts where sched_mutex isn't held;
3025 use raiseAsyncWithLock(). */
3026 ACQUIRE_LOCK(&sched_mutex);
3027 raiseAsync(tso,exception);
3028 RELEASE_LOCK(&sched_mutex);
3032 raiseAsync(StgTSO *tso, StgClosure *exception)
3034 StgRetInfoTable *info;
3037 // Thread already dead?
3038 if (tso->what_next == ThreadComplete || tso->what_next == ThreadKilled) {
3043 sched_belch("raising exception in thread %ld.", tso->id));
3045 // Remove it from any blocking queues
3050 // The stack freezing code assumes there's a closure pointer on
3051 // the top of the stack, so we have to arrange that this is the case...
3053 if (sp[0] == (W_)&stg_enter_info) {
3057 sp[0] = (W_)&stg_dummy_ret_closure;
3063 // 1. Let the top of the stack be the "current closure"
3065 // 2. Walk up the stack until we find either an UPDATE_FRAME or a
3068 // 3. If it's an UPDATE_FRAME, then make an AP_STACK containing the
3069 // current closure applied to the chunk of stack up to (but not
3070 // including) the update frame. This closure becomes the "current
3071 // closure". Go back to step 2.
3073 // 4. If it's a CATCH_FRAME, then leave the exception handler on
3074 // top of the stack applied to the exception.
3076 // 5. If it's a STOP_FRAME, then kill the thread.
3081 info = get_ret_itbl((StgClosure *)frame);
3083 while (info->i.type != UPDATE_FRAME
3084 && (info->i.type != CATCH_FRAME || exception == NULL)
3085 && info->i.type != STOP_FRAME) {
3086 frame += stack_frame_sizeW((StgClosure *)frame);
3087 info = get_ret_itbl((StgClosure *)frame);
3090 switch (info->i.type) {
3093 // If we find a CATCH_FRAME, and we've got an exception to raise,
3094 // then build the THUNK raise(exception), and leave it on
3095 // top of the CATCH_FRAME ready to enter.
3099 StgCatchFrame *cf = (StgCatchFrame *)frame;
3103 // we've got an exception to raise, so let's pass it to the
3104 // handler in this frame.
3106 raise = (StgClosure *)allocate(sizeofW(StgClosure)+1);
3107 TICK_ALLOC_SE_THK(1,0);
3108 SET_HDR(raise,&stg_raise_info,cf->header.prof.ccs);
3109 raise->payload[0] = exception;
3111 // throw away the stack from Sp up to the CATCH_FRAME.
3115 /* Ensure that async excpetions are blocked now, so we don't get
3116 * a surprise exception before we get around to executing the
3119 if (tso->blocked_exceptions == NULL) {
3120 tso->blocked_exceptions = END_TSO_QUEUE;
3123 /* Put the newly-built THUNK on top of the stack, ready to execute
3124 * when the thread restarts.
3127 sp[-1] = (W_)&stg_enter_info;
3129 tso->what_next = ThreadRunGHC;
3130 IF_DEBUG(sanity, checkTSO(tso));
3139 // First build an AP_STACK consisting of the stack chunk above the
3140 // current update frame, with the top word on the stack as the
3143 words = frame - sp - 1;
3144 ap = (StgAP_STACK *)allocate(PAP_sizeW(words));
3147 ap->fun = (StgClosure *)sp[0];
3149 for(i=0; i < (nat)words; ++i) {
3150 ap->payload[i] = (StgClosure *)*sp++;
3153 SET_HDR(ap,&stg_AP_STACK_info,
3154 ((StgClosure *)frame)->header.prof.ccs /* ToDo */);
3155 TICK_ALLOC_UP_THK(words+1,0);
3158 fprintf(stderr, "sched: Updating ");
3159 printPtr((P_)((StgUpdateFrame *)frame)->updatee);
3160 fprintf(stderr, " with ");
3161 printObj((StgClosure *)ap);
3164 // Replace the updatee with an indirection - happily
3165 // this will also wake up any threads currently
3166 // waiting on the result.
3168 // Warning: if we're in a loop, more than one update frame on
3169 // the stack may point to the same object. Be careful not to
3170 // overwrite an IND_OLDGEN in this case, because we'll screw
3171 // up the mutable lists. To be on the safe side, don't
3172 // overwrite any kind of indirection at all. See also
3173 // threadSqueezeStack in GC.c, where we have to make a similar
3176 if (!closure_IND(((StgUpdateFrame *)frame)->updatee)) {
3177 // revert the black hole
3178 UPD_IND_NOLOCK(((StgUpdateFrame *)frame)->updatee,ap);
3180 sp += sizeofW(StgUpdateFrame) - 1;
3181 sp[0] = (W_)ap; // push onto stack
3186 // We've stripped the entire stack, the thread is now dead.
3187 sp += sizeofW(StgStopFrame);
3188 tso->what_next = ThreadKilled;
3199 /* -----------------------------------------------------------------------------
3200 resurrectThreads is called after garbage collection on the list of
3201 threads found to be garbage. Each of these threads will be woken
3202 up and sent a signal: BlockedOnDeadMVar if the thread was blocked
3203 on an MVar, or NonTermination if the thread was blocked on a Black
3206 Locks: sched_mutex isn't held upon entry nor exit.
3207 -------------------------------------------------------------------------- */
3210 resurrectThreads( StgTSO *threads )
3214 for (tso = threads; tso != END_TSO_QUEUE; tso = next) {
3215 next = tso->global_link;
3216 tso->global_link = all_threads;
3218 IF_DEBUG(scheduler, sched_belch("resurrecting thread %d", tso->id));
3220 switch (tso->why_blocked) {
3222 case BlockedOnException:
3223 /* Called by GC - sched_mutex lock is currently held. */
3224 raiseAsync(tso,(StgClosure *)BlockedOnDeadMVar_closure);
3226 case BlockedOnBlackHole:
3227 raiseAsync(tso,(StgClosure *)NonTermination_closure);
3230 /* This might happen if the thread was blocked on a black hole
3231 * belonging to a thread that we've just woken up (raiseAsync
3232 * can wake up threads, remember...).
3236 barf("resurrectThreads: thread blocked in a strange way");
3241 /* -----------------------------------------------------------------------------
3242 * Blackhole detection: if we reach a deadlock, test whether any
3243 * threads are blocked on themselves. Any threads which are found to
3244 * be self-blocked get sent a NonTermination exception.
3246 * This is only done in a deadlock situation in order to avoid
3247 * performance overhead in the normal case.
3249 * Locks: sched_mutex is held upon entry and exit.
3250 * -------------------------------------------------------------------------- */
3253 detectBlackHoles( void )
3255 StgTSO *tso = all_threads;
3257 StgClosure *blocked_on;
3258 StgRetInfoTable *info;
3260 for (tso = all_threads; tso != END_TSO_QUEUE; tso = tso->global_link) {
3262 while (tso->what_next == ThreadRelocated) {
3264 ASSERT(get_itbl(tso)->type == TSO);
3267 if (tso->why_blocked != BlockedOnBlackHole) {
3270 blocked_on = tso->block_info.closure;
3272 frame = (StgClosure *)tso->sp;
3275 info = get_ret_itbl(frame);
3276 switch (info->i.type) {
3278 if (((StgUpdateFrame *)frame)->updatee == blocked_on) {
3279 /* We are blocking on one of our own computations, so
3280 * send this thread the NonTermination exception.
3283 sched_belch("thread %d is blocked on itself", tso->id));
3284 raiseAsync(tso, (StgClosure *)NonTermination_closure);
3288 frame = (StgClosure *) ((StgUpdateFrame *)frame + 1);
3294 // normal stack frames; do nothing except advance the pointer
3296 (StgPtr)frame += stack_frame_sizeW(frame);
3303 /* ----------------------------------------------------------------------------
3304 * Debugging: why is a thread blocked
3305 * [Also provides useful information when debugging threaded programs
3306 * at the Haskell source code level, so enable outside of DEBUG. --sof 7/02]
3307 ------------------------------------------------------------------------- */
3311 printThreadBlockage(StgTSO *tso)
3313 switch (tso->why_blocked) {
3315 fprintf(stderr,"is blocked on read from fd %d", tso->block_info.fd);
3317 case BlockedOnWrite:
3318 fprintf(stderr,"is blocked on write to fd %d", tso->block_info.fd);
3320 #if defined(mingw32_TARGET_OS)
3321 case BlockedOnDoProc:
3322 fprintf(stderr,"is blocked on proc (request: %d)", tso->block_info.async_result->reqID);
3325 case BlockedOnDelay:
3326 fprintf(stderr,"is blocked until %d", tso->block_info.target);
3329 fprintf(stderr,"is blocked on an MVar");
3331 case BlockedOnException:
3332 fprintf(stderr,"is blocked on delivering an exception to thread %d",
3333 tso->block_info.tso->id);
3335 case BlockedOnBlackHole:
3336 fprintf(stderr,"is blocked on a black hole");
3339 fprintf(stderr,"is not blocked");
3343 fprintf(stderr,"is blocked on global address; local FM_BQ is %p (%s)",
3344 tso->block_info.closure, info_type(tso->block_info.closure));
3346 case BlockedOnGA_NoSend:
3347 fprintf(stderr,"is blocked on global address (no send); local FM_BQ is %p (%s)",
3348 tso->block_info.closure, info_type(tso->block_info.closure));
3351 #if defined(RTS_SUPPORTS_THREADS)
3352 case BlockedOnCCall:
3353 fprintf(stderr,"is blocked on an external call");
3355 case BlockedOnCCall_NoUnblockExc:
3356 fprintf(stderr,"is blocked on an external call (exceptions were already blocked)");
3360 barf("printThreadBlockage: strange tso->why_blocked: %d for TSO %d (%d)",
3361 tso->why_blocked, tso->id, tso);
3367 printThreadStatus(StgTSO *tso)
3369 switch (tso->what_next) {
3371 fprintf(stderr,"has been killed");
3373 case ThreadComplete:
3374 fprintf(stderr,"has completed");
3377 printThreadBlockage(tso);
3382 printAllThreads(void)
3388 char time_string[TIME_STR_LEN], node_str[NODE_STR_LEN];
3389 ullong_format_string(TIME_ON_PROC(CurrentProc),
3390 time_string, rtsFalse/*no commas!*/);
3392 fprintf(stderr, "all threads at [%s]:\n", time_string);
3394 char time_string[TIME_STR_LEN], node_str[NODE_STR_LEN];
3395 ullong_format_string(CURRENT_TIME,
3396 time_string, rtsFalse/*no commas!*/);
3398 fprintf(stderr,"all threads at [%s]:\n", time_string);
3400 fprintf(stderr,"all threads:\n");
3403 for (t = all_threads; t != END_TSO_QUEUE; t = t->global_link) {
3404 fprintf(stderr, "\tthread %d @ %p ", t->id, (void *)t);
3405 label = lookupThreadLabel(t->id);
3406 if (label) fprintf(stderr,"[\"%s\"] ",(char *)label);
3407 printThreadStatus(t);
3408 fprintf(stderr,"\n");
3415 Print a whole blocking queue attached to node (debugging only).
3419 print_bq (StgClosure *node)
3421 StgBlockingQueueElement *bqe;
3425 fprintf(stderr,"## BQ of closure %p (%s): ",
3426 node, info_type(node));
3428 /* should cover all closures that may have a blocking queue */
3429 ASSERT(get_itbl(node)->type == BLACKHOLE_BQ ||
3430 get_itbl(node)->type == FETCH_ME_BQ ||
3431 get_itbl(node)->type == RBH ||
3432 get_itbl(node)->type == MVAR);
3434 ASSERT(node!=(StgClosure*)NULL); // sanity check
3436 print_bqe(((StgBlockingQueue*)node)->blocking_queue);
3440 Print a whole blocking queue starting with the element bqe.
3443 print_bqe (StgBlockingQueueElement *bqe)
3448 NB: In a parallel setup a BQ of an RBH must end with an RBH_Save closure;
3450 for (end = (bqe==END_BQ_QUEUE);
3451 !end; // iterate until bqe points to a CONSTR
3452 end = (get_itbl(bqe)->type == CONSTR) || (bqe->link==END_BQ_QUEUE),
3453 bqe = end ? END_BQ_QUEUE : bqe->link) {
3454 ASSERT(bqe != END_BQ_QUEUE); // sanity check
3455 ASSERT(bqe != (StgBlockingQueueElement *)NULL); // sanity check
3456 /* types of closures that may appear in a blocking queue */
3457 ASSERT(get_itbl(bqe)->type == TSO ||
3458 get_itbl(bqe)->type == BLOCKED_FETCH ||
3459 get_itbl(bqe)->type == CONSTR);
3460 /* only BQs of an RBH end with an RBH_Save closure */
3461 //ASSERT(get_itbl(bqe)->type != CONSTR || get_itbl(node)->type == RBH);
3463 switch (get_itbl(bqe)->type) {
3465 fprintf(stderr," TSO %u (%x),",
3466 ((StgTSO *)bqe)->id, ((StgTSO *)bqe));
3469 fprintf(stderr," BF (node=%p, ga=((%x, %d, %x)),",
3470 ((StgBlockedFetch *)bqe)->node,
3471 ((StgBlockedFetch *)bqe)->ga.payload.gc.gtid,
3472 ((StgBlockedFetch *)bqe)->ga.payload.gc.slot,
3473 ((StgBlockedFetch *)bqe)->ga.weight);
3476 fprintf(stderr," %s (IP %p),",
3477 (get_itbl(bqe) == &stg_RBH_Save_0_info ? "RBH_Save_0" :
3478 get_itbl(bqe) == &stg_RBH_Save_1_info ? "RBH_Save_1" :
3479 get_itbl(bqe) == &stg_RBH_Save_2_info ? "RBH_Save_2" :
3480 "RBH_Save_?"), get_itbl(bqe));
3483 barf("Unexpected closure type %s in blocking queue", // of %p (%s)",
3484 info_type((StgClosure *)bqe)); // , node, info_type(node));
3488 fputc('\n', stderr);
3490 # elif defined(GRAN)
3492 print_bq (StgClosure *node)
3494 StgBlockingQueueElement *bqe;
3495 PEs node_loc, tso_loc;
3498 /* should cover all closures that may have a blocking queue */
3499 ASSERT(get_itbl(node)->type == BLACKHOLE_BQ ||
3500 get_itbl(node)->type == FETCH_ME_BQ ||
3501 get_itbl(node)->type == RBH);
3503 ASSERT(node!=(StgClosure*)NULL); // sanity check
3504 node_loc = where_is(node);
3506 fprintf(stderr,"## BQ of closure %p (%s) on [PE %d]: ",
3507 node, info_type(node), node_loc);
3510 NB: In a parallel setup a BQ of an RBH must end with an RBH_Save closure;
3512 for (bqe = ((StgBlockingQueue*)node)->blocking_queue, end = (bqe==END_BQ_QUEUE);
3513 !end; // iterate until bqe points to a CONSTR
3514 end = (get_itbl(bqe)->type == CONSTR) || (bqe->link==END_BQ_QUEUE), bqe = end ? END_BQ_QUEUE : bqe->link) {
3515 ASSERT(bqe != END_BQ_QUEUE); // sanity check
3516 ASSERT(bqe != (StgBlockingQueueElement *)NULL); // sanity check
3517 /* types of closures that may appear in a blocking queue */
3518 ASSERT(get_itbl(bqe)->type == TSO ||
3519 get_itbl(bqe)->type == CONSTR);
3520 /* only BQs of an RBH end with an RBH_Save closure */
3521 ASSERT(get_itbl(bqe)->type != CONSTR || get_itbl(node)->type == RBH);
3523 tso_loc = where_is((StgClosure *)bqe);
3524 switch (get_itbl(bqe)->type) {
3526 fprintf(stderr," TSO %d (%p) on [PE %d],",
3527 ((StgTSO *)bqe)->id, (StgTSO *)bqe, tso_loc);
3530 fprintf(stderr," %s (IP %p),",
3531 (get_itbl(bqe) == &stg_RBH_Save_0_info ? "RBH_Save_0" :
3532 get_itbl(bqe) == &stg_RBH_Save_1_info ? "RBH_Save_1" :
3533 get_itbl(bqe) == &stg_RBH_Save_2_info ? "RBH_Save_2" :
3534 "RBH_Save_?"), get_itbl(bqe));
3537 barf("Unexpected closure type %s in blocking queue of %p (%s)",
3538 info_type((StgClosure *)bqe), node, info_type(node));
3542 fputc('\n', stderr);
3546 Nice and easy: only TSOs on the blocking queue
3549 print_bq (StgClosure *node)
3553 ASSERT(node!=(StgClosure*)NULL); // sanity check
3554 for (tso = ((StgBlockingQueue*)node)->blocking_queue;
3555 tso != END_TSO_QUEUE;
3557 ASSERT(tso!=NULL && tso!=END_TSO_QUEUE); // sanity check
3558 ASSERT(get_itbl(tso)->type == TSO); // guess what, sanity check
3559 fprintf(stderr," TSO %d (%p),", tso->id, tso);
3561 fputc('\n', stderr);
3572 for (i=0, tso=run_queue_hd;
3573 tso != END_TSO_QUEUE;
3582 sched_belch(char *s, ...)
3586 #ifdef RTS_SUPPORTS_THREADS
3587 fprintf(stderr, "sched (task %p): ", osThreadId());
3589 fprintf(stderr, "== ");
3591 fprintf(stderr, "sched: ");
3593 vfprintf(stderr, s, ap);
3594 fprintf(stderr, "\n");