diff options
Diffstat (limited to 'rts/Schedule.c')
| -rw-r--r-- | rts/Schedule.c | 4589 |
1 files changed, 4589 insertions, 0 deletions
diff --git a/rts/Schedule.c b/rts/Schedule.c new file mode 100644 index 0000000000..52fd4d5df6 --- /dev/null +++ b/rts/Schedule.c @@ -0,0 +1,4589 @@ +/* --------------------------------------------------------------------------- + * + * (c) The GHC Team, 1998-2005 + * + * The scheduler and thread-related functionality + * + * --------------------------------------------------------------------------*/ + +#include "PosixSource.h" +#include "Rts.h" +#include "SchedAPI.h" +#include "RtsUtils.h" +#include "RtsFlags.h" +#include "BlockAlloc.h" +#include "OSThreads.h" +#include "Storage.h" +#include "StgRun.h" +#include "Hooks.h" +#include "Schedule.h" +#include "StgMiscClosures.h" +#include "Interpreter.h" +#include "Exception.h" +#include "Printer.h" +#include "RtsSignals.h" +#include "Sanity.h" +#include "Stats.h" +#include "STM.h" +#include "Timer.h" +#include "Prelude.h" +#include "ThreadLabels.h" +#include "LdvProfile.h" +#include "Updates.h" +#ifdef PROFILING +#include "Proftimer.h" +#include "ProfHeap.h" +#endif +#if defined(GRAN) || defined(PARALLEL_HASKELL) +# include "GranSimRts.h" +# include "GranSim.h" +# include "ParallelRts.h" +# include "Parallel.h" +# include "ParallelDebug.h" +# include "FetchMe.h" +# include "HLC.h" +#endif +#include "Sparks.h" +#include "Capability.h" +#include "Task.h" +#include "AwaitEvent.h" +#if defined(mingw32_HOST_OS) +#include "win32/IOManager.h" +#endif + +#ifdef HAVE_SYS_TYPES_H +#include <sys/types.h> +#endif +#ifdef HAVE_UNISTD_H +#include <unistd.h> +#endif + +#include <string.h> +#include <stdlib.h> +#include <stdarg.h> + +#ifdef HAVE_ERRNO_H +#include <errno.h> +#endif + +// Turn off inlining when debugging - it obfuscates things +#ifdef DEBUG +# undef STATIC_INLINE +# define STATIC_INLINE static +#endif + +/* ----------------------------------------------------------------------------- + * Global variables + * -------------------------------------------------------------------------- */ + +#if defined(GRAN) + +StgTSO* ActiveTSO = NULL; /* for assigning system costs; GranSim-Light only */ +/* rtsTime TimeOfNextEvent, EndOfTimeSlice; now in GranSim.c */ + +/* + In GranSim we have a runnable and a blocked queue for each processor. + In order to minimise code changes new arrays run_queue_hds/tls + are created. run_queue_hd is then a short cut (macro) for + run_queue_hds[CurrentProc] (see GranSim.h). + -- HWL +*/ +StgTSO *run_queue_hds[MAX_PROC], *run_queue_tls[MAX_PROC]; +StgTSO *blocked_queue_hds[MAX_PROC], *blocked_queue_tls[MAX_PROC]; +StgTSO *ccalling_threadss[MAX_PROC]; +/* We use the same global list of threads (all_threads) in GranSim as in + the std RTS (i.e. we are cheating). However, we don't use this list in + the GranSim specific code at the moment (so we are only potentially + cheating). */ + +#else /* !GRAN */ + +#if !defined(THREADED_RTS) +// Blocked/sleeping thrads +StgTSO *blocked_queue_hd = NULL; +StgTSO *blocked_queue_tl = NULL; +StgTSO *sleeping_queue = NULL; // perhaps replace with a hash table? +#endif + +/* Threads blocked on blackholes. + * LOCK: sched_mutex+capability, or all capabilities + */ +StgTSO *blackhole_queue = NULL; +#endif + +/* The blackhole_queue should be checked for threads to wake up. See + * Schedule.h for more thorough comment. + * LOCK: none (doesn't matter if we miss an update) + */ +rtsBool blackholes_need_checking = rtsFalse; + +/* Linked list of all threads. + * Used for detecting garbage collected threads. + * LOCK: sched_mutex+capability, or all capabilities + */ +StgTSO *all_threads = NULL; + +/* flag set by signal handler to precipitate a context switch + * LOCK: none (just an advisory flag) + */ +int context_switch = 0; + +/* flag that tracks whether we have done any execution in this time slice. + * LOCK: currently none, perhaps we should lock (but needs to be + * updated in the fast path of the scheduler). + */ +nat recent_activity = ACTIVITY_YES; + +/* if this flag is set as well, give up execution + * LOCK: none (changes once, from false->true) + */ +rtsBool sched_state = SCHED_RUNNING; + +/* Next thread ID to allocate. + * LOCK: sched_mutex + */ +static StgThreadID next_thread_id = 1; + +/* The smallest stack size that makes any sense is: + * RESERVED_STACK_WORDS (so we can get back from the stack overflow) + * + sizeofW(StgStopFrame) (the stg_stop_thread_info frame) + * + 1 (the closure to enter) + * + 1 (stg_ap_v_ret) + * + 1 (spare slot req'd by stg_ap_v_ret) + * + * A thread with this stack will bomb immediately with a stack + * overflow, which will increase its stack size. + */ +#define MIN_STACK_WORDS (RESERVED_STACK_WORDS + sizeofW(StgStopFrame) + 3) + +#if defined(GRAN) +StgTSO *CurrentTSO; +#endif + +/* This is used in `TSO.h' and gcc 2.96 insists that this variable actually + * exists - earlier gccs apparently didn't. + * -= chak + */ +StgTSO dummy_tso; + +/* + * Set to TRUE when entering a shutdown state (via shutdownHaskellAndExit()) -- + * in an MT setting, needed to signal that a worker thread shouldn't hang around + * in the scheduler when it is out of work. + */ +rtsBool shutting_down_scheduler = rtsFalse; + +/* + * This mutex protects most of the global scheduler data in + * the THREADED_RTS runtime. + */ +#if defined(THREADED_RTS) +Mutex sched_mutex; +#endif + +#if defined(PARALLEL_HASKELL) +StgTSO *LastTSO; +rtsTime TimeOfLastYield; +rtsBool emitSchedule = rtsTrue; +#endif + +/* ----------------------------------------------------------------------------- + * static function prototypes + * -------------------------------------------------------------------------- */ + +static Capability *schedule (Capability *initialCapability, Task *task); + +// +// These function all encapsulate parts of the scheduler loop, and are +// abstracted only to make the structure and control flow of the +// scheduler clearer. +// +static void schedulePreLoop (void); +#if defined(THREADED_RTS) +static void schedulePushWork(Capability *cap, Task *task); +#endif +static void scheduleStartSignalHandlers (Capability *cap); +static void scheduleCheckBlockedThreads (Capability *cap); +static void scheduleCheckWakeupThreads(Capability *cap USED_IF_NOT_THREADS); +static void scheduleCheckBlackHoles (Capability *cap); +static void scheduleDetectDeadlock (Capability *cap, Task *task); +#if defined(GRAN) +static StgTSO *scheduleProcessEvent(rtsEvent *event); +#endif +#if defined(PARALLEL_HASKELL) +static StgTSO *scheduleSendPendingMessages(void); +static void scheduleActivateSpark(void); +static rtsBool scheduleGetRemoteWork(rtsBool *receivedFinish); +#endif +#if defined(PAR) || defined(GRAN) +static void scheduleGranParReport(void); +#endif +static void schedulePostRunThread(void); +static rtsBool scheduleHandleHeapOverflow( Capability *cap, StgTSO *t ); +static void scheduleHandleStackOverflow( Capability *cap, Task *task, + StgTSO *t); +static rtsBool scheduleHandleYield( Capability *cap, StgTSO *t, + nat prev_what_next ); +static void scheduleHandleThreadBlocked( StgTSO *t ); +static rtsBool scheduleHandleThreadFinished( Capability *cap, Task *task, + StgTSO *t ); +static rtsBool scheduleDoHeapProfile(rtsBool ready_to_gc); +static Capability *scheduleDoGC(Capability *cap, Task *task, + rtsBool force_major, + void (*get_roots)(evac_fn)); + +static void unblockThread(Capability *cap, StgTSO *tso); +static rtsBool checkBlackHoles(Capability *cap); +static void AllRoots(evac_fn evac); + +static StgTSO *threadStackOverflow(Capability *cap, StgTSO *tso); + +static void raiseAsync_(Capability *cap, StgTSO *tso, StgClosure *exception, + rtsBool stop_at_atomically, StgPtr stop_here); + +static void deleteThread (Capability *cap, StgTSO *tso); +static void deleteAllThreads (Capability *cap); + +#ifdef DEBUG +static void printThreadBlockage(StgTSO *tso); +static void printThreadStatus(StgTSO *tso); +void printThreadQueue(StgTSO *tso); +#endif + +#if defined(PARALLEL_HASKELL) +StgTSO * createSparkThread(rtsSpark spark); +StgTSO * activateSpark (rtsSpark spark); +#endif + +#ifdef DEBUG +static char *whatNext_strs[] = { + "(unknown)", + "ThreadRunGHC", + "ThreadInterpret", + "ThreadKilled", + "ThreadRelocated", + "ThreadComplete" +}; +#endif + +/* ----------------------------------------------------------------------------- + * Putting a thread on the run queue: different scheduling policies + * -------------------------------------------------------------------------- */ + +STATIC_INLINE void +addToRunQueue( Capability *cap, StgTSO *t ) +{ +#if defined(PARALLEL_HASKELL) + if (RtsFlags.ParFlags.doFairScheduling) { + // this does round-robin scheduling; good for concurrency + appendToRunQueue(cap,t); + } else { + // this does unfair scheduling; good for parallelism + pushOnRunQueue(cap,t); + } +#else + // this does round-robin scheduling; good for concurrency + appendToRunQueue(cap,t); +#endif +} + +/* --------------------------------------------------------------------------- + Main scheduling loop. + + We use round-robin scheduling, each thread returning to the + scheduler loop when one of these conditions is detected: + + * out of heap space + * timer expires (thread yields) + * thread blocks + * thread ends + * stack overflow + + GRAN version: + In a GranSim setup this loop iterates over the global event queue. + This revolves around the global event queue, which determines what + to do next. Therefore, it's more complicated than either the + concurrent or the parallel (GUM) setup. + + GUM version: + GUM iterates over incoming messages. + It starts with nothing to do (thus CurrentTSO == END_TSO_QUEUE), + and sends out a fish whenever it has nothing to do; in-between + doing the actual reductions (shared code below) it processes the + incoming messages and deals with delayed operations + (see PendingFetches). + This is not the ugliest code you could imagine, but it's bloody close. + + ------------------------------------------------------------------------ */ + +static Capability * +schedule (Capability *initialCapability, Task *task) +{ + StgTSO *t; + Capability *cap; + StgThreadReturnCode ret; +#if defined(GRAN) + rtsEvent *event; +#elif defined(PARALLEL_HASKELL) + StgTSO *tso; + GlobalTaskId pe; + rtsBool receivedFinish = rtsFalse; +# if defined(DEBUG) + nat tp_size, sp_size; // stats only +# endif +#endif + nat prev_what_next; + rtsBool ready_to_gc; +#if defined(THREADED_RTS) + rtsBool first = rtsTrue; +#endif + + cap = initialCapability; + + // Pre-condition: this task owns initialCapability. + // The sched_mutex is *NOT* held + // NB. on return, we still hold a capability. + + IF_DEBUG(scheduler, + sched_belch("### NEW SCHEDULER LOOP (task: %p, cap: %p)", + task, initialCapability); + ); + + schedulePreLoop(); + + // ----------------------------------------------------------- + // Scheduler loop starts here: + +#if defined(PARALLEL_HASKELL) +#define TERMINATION_CONDITION (!receivedFinish) +#elif defined(GRAN) +#define TERMINATION_CONDITION ((event = get_next_event()) != (rtsEvent*)NULL) +#else +#define TERMINATION_CONDITION rtsTrue +#endif + + while (TERMINATION_CONDITION) { + +#if defined(GRAN) + /* Choose the processor with the next event */ + CurrentProc = event->proc; + CurrentTSO = event->tso; +#endif + +#if defined(THREADED_RTS) + if (first) { + // don't yield the first time, we want a chance to run this + // thread for a bit, even if there are others banging at the + // door. + first = rtsFalse; + ASSERT_FULL_CAPABILITY_INVARIANTS(cap,task); + } else { + // Yield the capability to higher-priority tasks if necessary. + yieldCapability(&cap, task); + } +#endif + +#if defined(THREADED_RTS) + schedulePushWork(cap,task); +#endif + + // Check whether we have re-entered the RTS from Haskell without + // going via suspendThread()/resumeThread (i.e. a 'safe' foreign + // call). + if (cap->in_haskell) { + errorBelch("schedule: re-entered unsafely.\n" + " Perhaps a 'foreign import unsafe' should be 'safe'?"); + stg_exit(EXIT_FAILURE); + } + + // The interruption / shutdown sequence. + // + // In order to cleanly shut down the runtime, we want to: + // * make sure that all main threads return to their callers + // with the state 'Interrupted'. + // * clean up all OS threads assocated with the runtime + // * free all memory etc. + // + // So the sequence for ^C goes like this: + // + // * ^C handler sets sched_state := SCHED_INTERRUPTING and + // arranges for some Capability to wake up + // + // * all threads in the system are halted, and the zombies are + // placed on the run queue for cleaning up. We acquire all + // the capabilities in order to delete the threads, this is + // done by scheduleDoGC() for convenience (because GC already + // needs to acquire all the capabilities). We can't kill + // threads involved in foreign calls. + // + // * sched_state := SCHED_INTERRUPTED + // + // * somebody calls shutdownHaskell(), which calls exitScheduler() + // + // * sched_state := SCHED_SHUTTING_DOWN + // + // * all workers exit when the run queue on their capability + // drains. All main threads will also exit when their TSO + // reaches the head of the run queue and they can return. + // + // * eventually all Capabilities will shut down, and the RTS can + // exit. + // + // * We might be left with threads blocked in foreign calls, + // we should really attempt to kill these somehow (TODO); + + switch (sched_state) { + case SCHED_RUNNING: + break; + case SCHED_INTERRUPTING: + IF_DEBUG(scheduler, sched_belch("SCHED_INTERRUPTING")); +#if defined(THREADED_RTS) + discardSparksCap(cap); +#endif + /* scheduleDoGC() deletes all the threads */ + cap = scheduleDoGC(cap,task,rtsFalse,GetRoots); + break; + case SCHED_INTERRUPTED: + IF_DEBUG(scheduler, sched_belch("SCHED_INTERRUPTED")); + break; + case SCHED_SHUTTING_DOWN: + IF_DEBUG(scheduler, sched_belch("SCHED_SHUTTING_DOWN")); + // If we are a worker, just exit. If we're a bound thread + // then we will exit below when we've removed our TSO from + // the run queue. + if (task->tso == NULL && emptyRunQueue(cap)) { + return cap; + } + break; + default: + barf("sched_state: %d", sched_state); + } + +#if defined(THREADED_RTS) + // If the run queue is empty, take a spark and turn it into a thread. + { + if (emptyRunQueue(cap)) { + StgClosure *spark; + spark = findSpark(cap); + if (spark != NULL) { + IF_DEBUG(scheduler, + sched_belch("turning spark of closure %p into a thread", + (StgClosure *)spark)); + createSparkThread(cap,spark); + } + } + } +#endif // THREADED_RTS + + scheduleStartSignalHandlers(cap); + + // Only check the black holes here if we've nothing else to do. + // During normal execution, the black hole list only gets checked + // at GC time, to avoid repeatedly traversing this possibly long + // list each time around the scheduler. + if (emptyRunQueue(cap)) { scheduleCheckBlackHoles(cap); } + + scheduleCheckWakeupThreads(cap); + + scheduleCheckBlockedThreads(cap); + + scheduleDetectDeadlock(cap,task); +#if defined(THREADED_RTS) + cap = task->cap; // reload cap, it might have changed +#endif + + // Normally, the only way we can get here with no threads to + // run is if a keyboard interrupt received during + // scheduleCheckBlockedThreads() or scheduleDetectDeadlock(). + // Additionally, it is not fatal for the + // threaded RTS to reach here with no threads to run. + // + // win32: might be here due to awaitEvent() being abandoned + // as a result of a console event having been delivered. + if ( emptyRunQueue(cap) ) { +#if !defined(THREADED_RTS) && !defined(mingw32_HOST_OS) + ASSERT(sched_state >= SCHED_INTERRUPTING); +#endif + continue; // nothing to do + } + +#if defined(PARALLEL_HASKELL) + scheduleSendPendingMessages(); + if (emptyRunQueue(cap) && scheduleActivateSpark()) + continue; + +#if defined(SPARKS) + ASSERT(next_fish_to_send_at==0); // i.e. no delayed fishes left! +#endif + + /* If we still have no work we need to send a FISH to get a spark + from another PE */ + if (emptyRunQueue(cap)) { + if (!scheduleGetRemoteWork(&receivedFinish)) continue; + ASSERT(rtsFalse); // should not happen at the moment + } + // from here: non-empty run queue. + // TODO: merge above case with this, only one call processMessages() ! + if (PacketsWaiting()) { /* process incoming messages, if + any pending... only in else + because getRemoteWork waits for + messages as well */ + receivedFinish = processMessages(); + } +#endif + +#if defined(GRAN) + scheduleProcessEvent(event); +#endif + + // + // Get a thread to run + // + t = popRunQueue(cap); + +#if defined(GRAN) || defined(PAR) + scheduleGranParReport(); // some kind of debuging output +#else + // Sanity check the thread we're about to run. This can be + // expensive if there is lots of thread switching going on... + IF_DEBUG(sanity,checkTSO(t)); +#endif + +#if defined(THREADED_RTS) + // Check whether we can run this thread in the current task. + // If not, we have to pass our capability to the right task. + { + Task *bound = t->bound; + + if (bound) { + if (bound == task) { + IF_DEBUG(scheduler, + sched_belch("### Running thread %d in bound thread", + t->id)); + // yes, the Haskell thread is bound to the current native thread + } else { + IF_DEBUG(scheduler, + sched_belch("### thread %d bound to another OS thread", + t->id)); + // no, bound to a different Haskell thread: pass to that thread + pushOnRunQueue(cap,t); + continue; + } + } else { + // The thread we want to run is unbound. + if (task->tso) { + IF_DEBUG(scheduler, + sched_belch("### this OS thread cannot run thread %d", t->id)); + // no, the current native thread is bound to a different + // Haskell thread, so pass it to any worker thread + pushOnRunQueue(cap,t); + continue; + } + } + } +#endif + + cap->r.rCurrentTSO = t; + + /* context switches are initiated by the timer signal, unless + * the user specified "context switch as often as possible", with + * +RTS -C0 + */ + if (RtsFlags.ConcFlags.ctxtSwitchTicks == 0 + && !emptyThreadQueues(cap)) { + context_switch = 1; + } + +run_thread: + + IF_DEBUG(scheduler, sched_belch("-->> running thread %ld %s ...", + (long)t->id, whatNext_strs[t->what_next])); + +#if defined(PROFILING) + startHeapProfTimer(); +#endif + + // ---------------------------------------------------------------------- + // Run the current thread + + ASSERT_FULL_CAPABILITY_INVARIANTS(cap,task); + ASSERT(t->cap == cap); + + prev_what_next = t->what_next; + + errno = t->saved_errno; + cap->in_haskell = rtsTrue; + + dirtyTSO(t); + + recent_activity = ACTIVITY_YES; + + switch (prev_what_next) { + + case ThreadKilled: + case ThreadComplete: + /* Thread already finished, return to scheduler. */ + ret = ThreadFinished; + break; + + case ThreadRunGHC: + { + StgRegTable *r; + r = StgRun((StgFunPtr) stg_returnToStackTop, &cap->r); + cap = regTableToCapability(r); + ret = r->rRet; + break; + } + + case ThreadInterpret: + cap = interpretBCO(cap); + ret = cap->r.rRet; + break; + + default: + barf("schedule: invalid what_next field"); + } + + cap->in_haskell = rtsFalse; + + // The TSO might have moved, eg. if it re-entered the RTS and a GC + // happened. So find the new location: + t = cap->r.rCurrentTSO; + + // We have run some Haskell code: there might be blackhole-blocked + // threads to wake up now. + // Lock-free test here should be ok, we're just setting a flag. + if ( blackhole_queue != END_TSO_QUEUE ) { + blackholes_need_checking = rtsTrue; + } + + // And save the current errno in this thread. + // XXX: possibly bogus for SMP because this thread might already + // be running again, see code below. + t->saved_errno = errno; + +#if defined(THREADED_RTS) + // If ret is ThreadBlocked, and this Task is bound to the TSO that + // blocked, we are in limbo - the TSO is now owned by whatever it + // is blocked on, and may in fact already have been woken up, + // perhaps even on a different Capability. It may be the case + // that task->cap != cap. We better yield this Capability + // immediately and return to normaility. + if (ret == ThreadBlocked) { + IF_DEBUG(scheduler, + sched_belch("--<< thread %d (%s) stopped: blocked\n", + t->id, whatNext_strs[t->what_next])); + continue; + } +#endif + + ASSERT_FULL_CAPABILITY_INVARIANTS(cap,task); + ASSERT(t->cap == cap); + + // ---------------------------------------------------------------------- + + // Costs for the scheduler are assigned to CCS_SYSTEM +#if defined(PROFILING) + stopHeapProfTimer(); + CCCS = CCS_SYSTEM; +#endif + +#if defined(THREADED_RTS) + IF_DEBUG(scheduler,debugBelch("sched (task %p): ", (void *)(unsigned long)(unsigned int)osThreadId());); +#elif !defined(GRAN) && !defined(PARALLEL_HASKELL) + IF_DEBUG(scheduler,debugBelch("sched: ");); +#endif + + schedulePostRunThread(); + + ready_to_gc = rtsFalse; + + switch (ret) { + case HeapOverflow: + ready_to_gc = scheduleHandleHeapOverflow(cap,t); + break; + + case StackOverflow: + scheduleHandleStackOverflow(cap,task,t); + break; + + case ThreadYielding: + if (scheduleHandleYield(cap, t, prev_what_next)) { + // shortcut for switching between compiler/interpreter: + goto run_thread; + } + break; + + case ThreadBlocked: + scheduleHandleThreadBlocked(t); + break; + + case ThreadFinished: + if (scheduleHandleThreadFinished(cap, task, t)) return cap; + ASSERT_FULL_CAPABILITY_INVARIANTS(cap,task); + break; + + default: + barf("schedule: invalid thread return code %d", (int)ret); + } + + if (scheduleDoHeapProfile(ready_to_gc)) { ready_to_gc = rtsFalse; } + if (ready_to_gc) { + cap = scheduleDoGC(cap,task,rtsFalse,GetRoots); + } + } /* end of while() */ + + IF_PAR_DEBUG(verbose, + debugBelch("== Leaving schedule() after having received Finish\n")); +} + +/* ---------------------------------------------------------------------------- + * Setting up the scheduler loop + * ------------------------------------------------------------------------- */ + +static void +schedulePreLoop(void) +{ +#if defined(GRAN) + /* set up first event to get things going */ + /* ToDo: assign costs for system setup and init MainTSO ! */ + new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc], + ContinueThread, + CurrentTSO, (StgClosure*)NULL, (rtsSpark*)NULL); + + IF_DEBUG(gran, + debugBelch("GRAN: Init CurrentTSO (in schedule) = %p\n", + CurrentTSO); + G_TSO(CurrentTSO, 5)); + + if (RtsFlags.GranFlags.Light) { + /* Save current time; GranSim Light only */ + CurrentTSO->gran.clock = CurrentTime[CurrentProc]; + } +#endif +} + +/* ----------------------------------------------------------------------------- + * schedulePushWork() + * + * Push work to other Capabilities if we have some. + * -------------------------------------------------------------------------- */ + +#if defined(THREADED_RTS) +static void +schedulePushWork(Capability *cap USED_IF_THREADS, + Task *task USED_IF_THREADS) +{ + Capability *free_caps[n_capabilities], *cap0; + nat i, n_free_caps; + + // migration can be turned off with +RTS -qg + if (!RtsFlags.ParFlags.migrate) return; + + // Check whether we have more threads on our run queue, or sparks + // in our pool, that we could hand to another Capability. + if ((emptyRunQueue(cap) || cap->run_queue_hd->link == END_TSO_QUEUE) + && sparkPoolSizeCap(cap) < 2) { + return; + } + + // First grab as many free Capabilities as we can. + for (i=0, n_free_caps=0; i < n_capabilities; i++) { + cap0 = &capabilities[i]; + if (cap != cap0 && tryGrabCapability(cap0,task)) { + if (!emptyRunQueue(cap0) || cap->returning_tasks_hd != NULL) { + // it already has some work, we just grabbed it at + // the wrong moment. Or maybe it's deadlocked! + releaseCapability(cap0); + } else { + free_caps[n_free_caps++] = cap0; + } + } + } + + // we now have n_free_caps free capabilities stashed in + // free_caps[]. Share our run queue equally with them. This is + // probably the simplest thing we could do; improvements we might + // want to do include: + // + // - giving high priority to moving relatively new threads, on + // the gournds that they haven't had time to build up a + // working set in the cache on this CPU/Capability. + // + // - giving low priority to moving long-lived threads + + if (n_free_caps > 0) { + StgTSO *prev, *t, *next; + rtsBool pushed_to_all; + + IF_DEBUG(scheduler, sched_belch("excess threads on run queue and %d free capabilities, sharing...", n_free_caps)); + + i = 0; + pushed_to_all = rtsFalse; + + if (cap->run_queue_hd != END_TSO_QUEUE) { + prev = cap->run_queue_hd; + t = prev->link; + prev->link = END_TSO_QUEUE; + for (; t != END_TSO_QUEUE; t = next) { + next = t->link; + t->link = END_TSO_QUEUE; + if (t->what_next == ThreadRelocated + || t->bound == task // don't move my bound thread + || tsoLocked(t)) { // don't move a locked thread + prev->link = t; + prev = t; + } else if (i == n_free_caps) { + pushed_to_all = rtsTrue; + i = 0; + // keep one for us + prev->link = t; + prev = t; + } else { + IF_DEBUG(scheduler, sched_belch("pushing thread %d to capability %d", t->id, free_caps[i]->no)); + appendToRunQueue(free_caps[i],t); + if (t->bound) { t->bound->cap = free_caps[i]; } + t->cap = free_caps[i]; + i++; + } + } + cap->run_queue_tl = prev; + } + + // If there are some free capabilities that we didn't push any + // threads to, then try to push a spark to each one. + if (!pushed_to_all) { + StgClosure *spark; + // i is the next free capability to push to + for (; i < n_free_caps; i++) { + if (emptySparkPoolCap(free_caps[i])) { + spark = findSpark(cap); + if (spark != NULL) { + IF_DEBUG(scheduler, sched_belch("pushing spark %p to capability %d", spark, free_caps[i]->no)); + newSpark(&(free_caps[i]->r), spark); + } + } + } + } + + // release the capabilities + for (i = 0; i < n_free_caps; i++) { + task->cap = free_caps[i]; + releaseCapability(free_caps[i]); + } + } + task->cap = cap; // reset to point to our Capability. +} +#endif + +/* ---------------------------------------------------------------------------- + * Start any pending signal handlers + * ------------------------------------------------------------------------- */ + +#if defined(RTS_USER_SIGNALS) && (!defined(THREADED_RTS) || defined(mingw32_HOST_OS)) +static void +scheduleStartSignalHandlers(Capability *cap) +{ + if (signals_pending()) { // safe outside the lock + startSignalHandlers(cap); + } +} +#else +static void +scheduleStartSignalHandlers(Capability *cap STG_UNUSED) +{ +} +#endif + +/* ---------------------------------------------------------------------------- + * Check for blocked threads that can be woken up. + * ------------------------------------------------------------------------- */ + +static void +scheduleCheckBlockedThreads(Capability *cap USED_IF_NOT_THREADS) +{ +#if !defined(THREADED_RTS) + // + // Check whether any waiting threads need to be woken up. If the + // run queue is empty, and there are no other tasks running, we + // can wait indefinitely for something to happen. + // + if ( !emptyQueue(blocked_queue_hd) || !emptyQueue(sleeping_queue) ) + { + awaitEvent( emptyRunQueue(cap) && !blackholes_need_checking ); + } +#endif +} + + +/* ---------------------------------------------------------------------------- + * Check for threads woken up by other Capabilities + * ------------------------------------------------------------------------- */ + +static void +scheduleCheckWakeupThreads(Capability *cap USED_IF_THREADS) +{ +#if defined(THREADED_RTS) + // Any threads that were woken up by other Capabilities get + // appended to our run queue. + if (!emptyWakeupQueue(cap)) { + ACQUIRE_LOCK(&cap->lock); + if (emptyRunQueue(cap)) { + cap->run_queue_hd = cap->wakeup_queue_hd; + cap->run_queue_tl = cap->wakeup_queue_tl; + } else { + cap->run_queue_tl->link = cap->wakeup_queue_hd; + cap->run_queue_tl = cap->wakeup_queue_tl; + } + cap->wakeup_queue_hd = cap->wakeup_queue_tl = END_TSO_QUEUE; + RELEASE_LOCK(&cap->lock); + } +#endif +} + +/* ---------------------------------------------------------------------------- + * Check for threads blocked on BLACKHOLEs that can be woken up + * ------------------------------------------------------------------------- */ +static void +scheduleCheckBlackHoles (Capability *cap) +{ + if ( blackholes_need_checking ) // check without the lock first + { + ACQUIRE_LOCK(&sched_mutex); + if ( blackholes_need_checking ) { + checkBlackHoles(cap); + blackholes_need_checking = rtsFalse; + } + RELEASE_LOCK(&sched_mutex); + } +} + +/* ---------------------------------------------------------------------------- + * Detect deadlock conditions and attempt to resolve them. + * ------------------------------------------------------------------------- */ + +static void +scheduleDetectDeadlock (Capability *cap, Task *task) +{ + +#if defined(PARALLEL_HASKELL) + // ToDo: add deadlock detection in GUM (similar to THREADED_RTS) -- HWL + return; +#endif + + /* + * Detect deadlock: when we have no threads to run, there are no + * threads blocked, waiting for I/O, or sleeping, and all the + * other tasks are waiting for work, we must have a deadlock of + * some description. + */ + if ( emptyThreadQueues(cap) ) + { +#if defined(THREADED_RTS) + /* + * In the threaded RTS, we only check for deadlock if there + * has been no activity in a complete timeslice. This means + * we won't eagerly start a full GC just because we don't have + * any threads to run currently. + */ + if (recent_activity != ACTIVITY_INACTIVE) return; +#endif + + IF_DEBUG(scheduler, sched_belch("deadlocked, forcing major GC...")); + + // Garbage collection can release some new threads due to + // either (a) finalizers or (b) threads resurrected because + // they are unreachable and will therefore be sent an + // exception. Any threads thus released will be immediately + // runnable. + cap = scheduleDoGC (cap, task, rtsTrue/*force major GC*/, GetRoots); + + recent_activity = ACTIVITY_DONE_GC; + + if ( !emptyRunQueue(cap) ) return; + +#if defined(RTS_USER_SIGNALS) && (!defined(THREADED_RTS) || defined(mingw32_HOST_OS)) + /* If we have user-installed signal handlers, then wait + * for signals to arrive rather then bombing out with a + * deadlock. + */ + if ( anyUserHandlers() ) { + IF_DEBUG(scheduler, + sched_belch("still deadlocked, waiting for signals...")); + + awaitUserSignals(); + + if (signals_pending()) { + startSignalHandlers(cap); + } + + // either we have threads to run, or we were interrupted: + ASSERT(!emptyRunQueue(cap) || sched_state >= SCHED_INTERRUPTING); + } +#endif + +#if !defined(THREADED_RTS) + /* Probably a real deadlock. Send the current main thread the + * Deadlock exception. + */ + if (task->tso) { + switch (task->tso->why_blocked) { + case BlockedOnSTM: + case BlockedOnBlackHole: + case BlockedOnException: + case BlockedOnMVar: + raiseAsync(cap, task->tso, (StgClosure *)NonTermination_closure); + return; + default: + barf("deadlock: main thread blocked in a strange way"); + } + } + return; +#endif + } +} + +/* ---------------------------------------------------------------------------- + * Process an event (GRAN only) + * ------------------------------------------------------------------------- */ + +#if defined(GRAN) +static StgTSO * +scheduleProcessEvent(rtsEvent *event) +{ + StgTSO *t; + + if (RtsFlags.GranFlags.Light) + GranSimLight_enter_system(event, &ActiveTSO); // adjust ActiveTSO etc + + /* adjust time based on time-stamp */ + if (event->time > CurrentTime[CurrentProc] && + event->evttype != ContinueThread) + CurrentTime[CurrentProc] = event->time; + + /* Deal with the idle PEs (may issue FindWork or MoveSpark events) */ + if (!RtsFlags.GranFlags.Light) + handleIdlePEs(); + + IF_DEBUG(gran, debugBelch("GRAN: switch by event-type\n")); + + /* main event dispatcher in GranSim */ + switch (event->evttype) { + /* Should just be continuing execution */ + case ContinueThread: + IF_DEBUG(gran, debugBelch("GRAN: doing ContinueThread\n")); + /* ToDo: check assertion + ASSERT(run_queue_hd != (StgTSO*)NULL && + run_queue_hd != END_TSO_QUEUE); + */ + /* Ignore ContinueThreads for fetching threads (if synchr comm) */ + if (!RtsFlags.GranFlags.DoAsyncFetch && + procStatus[CurrentProc]==Fetching) { + debugBelch("ghuH: Spurious ContinueThread while Fetching ignored; TSO %d (%p) [PE %d]\n", + CurrentTSO->id, CurrentTSO, CurrentProc); + goto next_thread; + } + /* Ignore ContinueThreads for completed threads */ + if (CurrentTSO->what_next == ThreadComplete) { + debugBelch("ghuH: found a ContinueThread event for completed thread %d (%p) [PE %d] (ignoring ContinueThread)\n", + CurrentTSO->id, CurrentTSO, CurrentProc); + goto next_thread; + } + /* Ignore ContinueThreads for threads that are being migrated */ + if (PROCS(CurrentTSO)==Nowhere) { + debugBelch("ghuH: trying to run the migrating TSO %d (%p) [PE %d] (ignoring ContinueThread)\n", + CurrentTSO->id, CurrentTSO, CurrentProc); + goto next_thread; + } + /* The thread should be at the beginning of the run queue */ + if (CurrentTSO!=run_queue_hds[CurrentProc]) { + debugBelch("ghuH: TSO %d (%p) [PE %d] is not at the start of the run_queue when doing a ContinueThread\n", + CurrentTSO->id, CurrentTSO, CurrentProc); + break; // run the thread anyway + } + /* + new_event(proc, proc, CurrentTime[proc], + FindWork, + (StgTSO*)NULL, (StgClosure*)NULL, (rtsSpark*)NULL); + goto next_thread; + */ /* Catches superfluous CONTINUEs -- should be unnecessary */ + break; // now actually run the thread; DaH Qu'vam yImuHbej + + case FetchNode: + do_the_fetchnode(event); + goto next_thread; /* handle next event in event queue */ + + case GlobalBlock: + do_the_globalblock(event); + goto next_thread; /* handle next event in event queue */ + + case FetchReply: + do_the_fetchreply(event); + goto next_thread; /* handle next event in event queue */ + + case UnblockThread: /* Move from the blocked queue to the tail of */ + do_the_unblock(event); + goto next_thread; /* handle next event in event queue */ + + case ResumeThread: /* Move from the blocked queue to the tail of */ + /* the runnable queue ( i.e. Qu' SImqa'lu') */ + event->tso->gran.blocktime += + CurrentTime[CurrentProc] - event->tso->gran.blockedat; + do_the_startthread(event); + goto next_thread; /* handle next event in event queue */ + + case StartThread: + do_the_startthread(event); + goto next_thread; /* handle next event in event queue */ + + case MoveThread: + do_the_movethread(event); + goto next_thread; /* handle next event in event queue */ + + case MoveSpark: + do_the_movespark(event); + goto next_thread; /* handle next event in event queue */ + + case FindWork: + do_the_findwork(event); + goto next_thread; /* handle next event in event queue */ + + default: + barf("Illegal event type %u\n", event->evttype); + } /* switch */ + + /* This point was scheduler_loop in the old RTS */ + + IF_DEBUG(gran, debugBelch("GRAN: after main switch\n")); + + TimeOfLastEvent = CurrentTime[CurrentProc]; + TimeOfNextEvent = get_time_of_next_event(); + IgnoreEvents=(TimeOfNextEvent==0); // HWL HACK + // CurrentTSO = ThreadQueueHd; + + IF_DEBUG(gran, debugBelch("GRAN: time of next event is: %ld\n", + TimeOfNextEvent)); + + if (RtsFlags.GranFlags.Light) + GranSimLight_leave_system(event, &ActiveTSO); + + EndOfTimeSlice = CurrentTime[CurrentProc]+RtsFlags.GranFlags.time_slice; + + IF_DEBUG(gran, + debugBelch("GRAN: end of time-slice is %#lx\n", EndOfTimeSlice)); + + /* in a GranSim setup the TSO stays on the run queue */ + t = CurrentTSO; + /* Take a thread from the run queue. */ + POP_RUN_QUEUE(t); // take_off_run_queue(t); + + IF_DEBUG(gran, + debugBelch("GRAN: About to run current thread, which is\n"); + G_TSO(t,5)); + + context_switch = 0; // turned on via GranYield, checking events and time slice + + IF_DEBUG(gran, + DumpGranEvent(GR_SCHEDULE, t)); + + procStatus[CurrentProc] = Busy; +} +#endif // GRAN + +/* ---------------------------------------------------------------------------- + * Send pending messages (PARALLEL_HASKELL only) + * ------------------------------------------------------------------------- */ + +#if defined(PARALLEL_HASKELL) +static StgTSO * +scheduleSendPendingMessages(void) +{ + StgSparkPool *pool; + rtsSpark spark; + StgTSO *t; + +# if defined(PAR) // global Mem.Mgmt., omit for now + if (PendingFetches != END_BF_QUEUE) { + processFetches(); + } +# endif + + if (RtsFlags.ParFlags.BufferTime) { + // if we use message buffering, we must send away all message + // packets which have become too old... + sendOldBuffers(); + } +} +#endif + +/* ---------------------------------------------------------------------------- + * Activate spark threads (PARALLEL_HASKELL only) + * ------------------------------------------------------------------------- */ + +#if defined(PARALLEL_HASKELL) +static void +scheduleActivateSpark(void) +{ +#if defined(SPARKS) + ASSERT(emptyRunQueue()); +/* We get here if the run queue is empty and want some work. + We try to turn a spark into a thread, and add it to the run queue, + from where it will be picked up in the next iteration of the scheduler + loop. +*/ + + /* :-[ no local threads => look out for local sparks */ + /* the spark pool for the current PE */ + pool = &(cap.r.rSparks); // JB: cap = (old) MainCap + if (advisory_thread_count < RtsFlags.ParFlags.maxThreads && + pool->hd < pool->tl) { + /* + * ToDo: add GC code check that we really have enough heap afterwards!! + * Old comment: + * If we're here (no runnable threads) and we have pending + * sparks, we must have a space problem. Get enough space + * to turn one of those pending sparks into a + * thread... + */ + + spark = findSpark(rtsFalse); /* get a spark */ + if (spark != (rtsSpark) NULL) { + tso = createThreadFromSpark(spark); /* turn the spark into a thread */ + IF_PAR_DEBUG(fish, // schedule, + debugBelch("==== schedule: Created TSO %d (%p); %d threads active\n", + tso->id, tso, advisory_thread_count)); + + if (tso==END_TSO_QUEUE) { /* failed to activate spark->back to loop */ + IF_PAR_DEBUG(fish, // schedule, + debugBelch("==^^ failed to create thread from spark @ %lx\n", + spark)); + return rtsFalse; /* failed to generate a thread */ + } /* otherwise fall through & pick-up new tso */ + } else { + IF_PAR_DEBUG(fish, // schedule, + debugBelch("==^^ no local sparks (spark pool contains only NFs: %d)\n", + spark_queue_len(pool))); + return rtsFalse; /* failed to generate a thread */ + } + return rtsTrue; /* success in generating a thread */ + } else { /* no more threads permitted or pool empty */ + return rtsFalse; /* failed to generateThread */ + } +#else + tso = NULL; // avoid compiler warning only + return rtsFalse; /* dummy in non-PAR setup */ +#endif // SPARKS +} +#endif // PARALLEL_HASKELL + +/* ---------------------------------------------------------------------------- + * Get work from a remote node (PARALLEL_HASKELL only) + * ------------------------------------------------------------------------- */ + +#if defined(PARALLEL_HASKELL) +static rtsBool +scheduleGetRemoteWork(rtsBool *receivedFinish) +{ + ASSERT(emptyRunQueue()); + + if (RtsFlags.ParFlags.BufferTime) { + IF_PAR_DEBUG(verbose, + debugBelch("...send all pending data,")); + { + nat i; + for (i=1; i<=nPEs; i++) + sendImmediately(i); // send all messages away immediately + } + } +# ifndef SPARKS + //++EDEN++ idle() , i.e. send all buffers, wait for work + // suppress fishing in EDEN... just look for incoming messages + // (blocking receive) + IF_PAR_DEBUG(verbose, + debugBelch("...wait for incoming messages...\n")); + *receivedFinish = processMessages(); // blocking receive... + + // and reenter scheduling loop after having received something + // (return rtsFalse below) + +# else /* activate SPARKS machinery */ +/* We get here, if we have no work, tried to activate a local spark, but still + have no work. We try to get a remote spark, by sending a FISH message. + Thread migration should be added here, and triggered when a sequence of + fishes returns without work. */ + delay = (RtsFlags.ParFlags.fishDelay!=0ll ? RtsFlags.ParFlags.fishDelay : 0ll); + + /* =8-[ no local sparks => look for work on other PEs */ + /* + * We really have absolutely no work. Send out a fish + * (there may be some out there already), and wait for + * something to arrive. We clearly can't run any threads + * until a SCHEDULE or RESUME arrives, and so that's what + * we're hoping to see. (Of course, we still have to + * respond to other types of messages.) + */ + rtsTime now = msTime() /*CURRENT_TIME*/; + IF_PAR_DEBUG(verbose, + debugBelch("-- now=%ld\n", now)); + IF_PAR_DEBUG(fish, // verbose, + if (outstandingFishes < RtsFlags.ParFlags.maxFishes && + (last_fish_arrived_at!=0 && + last_fish_arrived_at+delay > now)) { + debugBelch("--$$ <%llu> delaying FISH until %llu (last fish %llu, delay %llu)\n", + now, last_fish_arrived_at+delay, + last_fish_arrived_at, + delay); + }); + + if (outstandingFishes < RtsFlags.ParFlags.maxFishes && + advisory_thread_count < RtsFlags.ParFlags.maxThreads) { // send a FISH, but when? + if (last_fish_arrived_at==0 || + (last_fish_arrived_at+delay <= now)) { // send FISH now! + /* outstandingFishes is set in sendFish, processFish; + avoid flooding system with fishes via delay */ + next_fish_to_send_at = 0; + } else { + /* ToDo: this should be done in the main scheduling loop to avoid the + busy wait here; not so bad if fish delay is very small */ + int iq = 0; // DEBUGGING -- HWL + next_fish_to_send_at = last_fish_arrived_at+delay; // remember when to send + /* send a fish when ready, but process messages that arrive in the meantime */ + do { + if (PacketsWaiting()) { + iq++; // DEBUGGING + *receivedFinish = processMessages(); + } + now = msTime(); + } while (!*receivedFinish || now<next_fish_to_send_at); + // JB: This means the fish could become obsolete, if we receive + // work. Better check for work again? + // last line: while (!receivedFinish || !haveWork || now<...) + // next line: if (receivedFinish || haveWork ) + + if (*receivedFinish) // no need to send a FISH if we are finishing anyway + return rtsFalse; // NB: this will leave scheduler loop + // immediately after return! + + IF_PAR_DEBUG(fish, // verbose, + debugBelch("--$$ <%llu> sent delayed fish (%d processMessages); active/total threads=%d/%d\n",now,iq,run_queue_len(),advisory_thread_count)); + + } + + // JB: IMHO, this should all be hidden inside sendFish(...) + /* pe = choosePE(); + sendFish(pe, thisPE, NEW_FISH_AGE, NEW_FISH_HISTORY, + NEW_FISH_HUNGER); + + // Global statistics: count no. of fishes + if (RtsFlags.ParFlags.ParStats.Global && + RtsFlags.GcFlags.giveStats > NO_GC_STATS) { + globalParStats.tot_fish_mess++; + } + */ + + /* delayed fishes must have been sent by now! */ + next_fish_to_send_at = 0; + } + + *receivedFinish = processMessages(); +# endif /* SPARKS */ + + return rtsFalse; + /* NB: this function always returns rtsFalse, meaning the scheduler + loop continues with the next iteration; + rationale: + return code means success in finding work; we enter this function + if there is no local work, thus have to send a fish which takes + time until it arrives with work; in the meantime we should process + messages in the main loop; + */ +} +#endif // PARALLEL_HASKELL + +/* ---------------------------------------------------------------------------- + * PAR/GRAN: Report stats & debugging info(?) + * ------------------------------------------------------------------------- */ + +#if defined(PAR) || defined(GRAN) +static void +scheduleGranParReport(void) +{ + ASSERT(run_queue_hd != END_TSO_QUEUE); + + /* Take a thread from the run queue, if we have work */ + POP_RUN_QUEUE(t); // take_off_run_queue(END_TSO_QUEUE); + + /* If this TSO has got its outport closed in the meantime, + * it mustn't be run. Instead, we have to clean it up as if it was finished. + * It has to be marked as TH_DEAD for this purpose. + * If it is TH_TERM instead, it is supposed to have finished in the normal way. + +JB: TODO: investigate wether state change field could be nuked + entirely and replaced by the normal tso state (whatnext + field). All we want to do is to kill tsos from outside. + */ + + /* ToDo: write something to the log-file + if (RTSflags.ParFlags.granSimStats && !sameThread) + DumpGranEvent(GR_SCHEDULE, RunnableThreadsHd); + + CurrentTSO = t; + */ + /* the spark pool for the current PE */ + pool = &(cap.r.rSparks); // cap = (old) MainCap + + IF_DEBUG(scheduler, + debugBelch("--=^ %d threads, %d sparks on [%#x]\n", + run_queue_len(), spark_queue_len(pool), CURRENT_PROC)); + + IF_PAR_DEBUG(fish, + debugBelch("--=^ %d threads, %d sparks on [%#x]\n", + run_queue_len(), spark_queue_len(pool), CURRENT_PROC)); + + if (RtsFlags.ParFlags.ParStats.Full && + (t->par.sparkname != (StgInt)0) && // only log spark generated threads + (emitSchedule || // forced emit + (t && LastTSO && t->id != LastTSO->id))) { + /* + we are running a different TSO, so write a schedule event to log file + NB: If we use fair scheduling we also have to write a deschedule + event for LastTSO; with unfair scheduling we know that the + previous tso has blocked whenever we switch to another tso, so + we don't need it in GUM for now + */ + IF_PAR_DEBUG(fish, // schedule, + debugBelch("____ scheduling spark generated thread %d (%lx) (%lx) via a forced emit\n",t->id,t,t->par.sparkname)); + + DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC, + GR_SCHEDULE, t, (StgClosure *)NULL, 0, 0); + emitSchedule = rtsFalse; + } +} +#endif + +/* ---------------------------------------------------------------------------- + * After running a thread... + * ------------------------------------------------------------------------- */ + +static void +schedulePostRunThread(void) +{ +#if defined(PAR) + /* HACK 675: if the last thread didn't yield, make sure to print a + SCHEDULE event to the log file when StgRunning the next thread, even + if it is the same one as before */ + LastTSO = t; + TimeOfLastYield = CURRENT_TIME; +#endif + + /* some statistics gathering in the parallel case */ + +#if defined(GRAN) || defined(PAR) || defined(EDEN) + switch (ret) { + case HeapOverflow: +# if defined(GRAN) + IF_DEBUG(gran, DumpGranEvent(GR_DESCHEDULE, t)); + globalGranStats.tot_heapover++; +# elif defined(PAR) + globalParStats.tot_heapover++; +# endif + break; + + case StackOverflow: +# if defined(GRAN) + IF_DEBUG(gran, + DumpGranEvent(GR_DESCHEDULE, t)); + globalGranStats.tot_stackover++; +# elif defined(PAR) + // IF_DEBUG(par, + // DumpGranEvent(GR_DESCHEDULE, t); + globalParStats.tot_stackover++; +# endif + break; + + case ThreadYielding: +# if defined(GRAN) + IF_DEBUG(gran, + DumpGranEvent(GR_DESCHEDULE, t)); + globalGranStats.tot_yields++; +# elif defined(PAR) + // IF_DEBUG(par, + // DumpGranEvent(GR_DESCHEDULE, t); + globalParStats.tot_yields++; +# endif + break; + + case ThreadBlocked: +# if defined(GRAN) + IF_DEBUG(scheduler, + debugBelch("--<< thread %ld (%p; %s) stopped, blocking on node %p [PE %d] with BQ: ", + 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))); + if (t->block_info.closure!=(StgClosure*)NULL) + print_bq(t->block_info.closure); + debugBelch("\n")); + + // ??? needed; should emit block before + IF_DEBUG(gran, + DumpGranEvent(GR_DESCHEDULE, t)); + prune_eventq(t, (StgClosure *)NULL); // prune ContinueThreads for t + /* + ngoq Dogh! + ASSERT(procStatus[CurrentProc]==Busy || + ((procStatus[CurrentProc]==Fetching) && + (t->block_info.closure!=(StgClosure*)NULL))); + if (run_queue_hds[CurrentProc] == END_TSO_QUEUE && + !(!RtsFlags.GranFlags.DoAsyncFetch && + procStatus[CurrentProc]==Fetching)) + procStatus[CurrentProc] = Idle; + */ +# elif defined(PAR) +//++PAR++ blockThread() writes the event (change?) +# endif + break; + + case ThreadFinished: + break; + + default: + barf("parGlobalStats: unknown return code"); + break; + } +#endif +} + +/* ----------------------------------------------------------------------------- + * Handle a thread that returned to the scheduler with ThreadHeepOverflow + * -------------------------------------------------------------------------- */ + +static rtsBool +scheduleHandleHeapOverflow( Capability *cap, StgTSO *t ) +{ + // did the task ask for a large block? + if (cap->r.rHpAlloc > BLOCK_SIZE) { + // if so, get one and push it on the front of the nursery. + bdescr *bd; + lnat blocks; + + blocks = (lnat)BLOCK_ROUND_UP(cap->r.rHpAlloc) / BLOCK_SIZE; + + IF_DEBUG(scheduler, + debugBelch("--<< thread %ld (%s) stopped: requesting a large block (size %ld)\n", + (long)t->id, whatNext_strs[t->what_next], blocks)); + + // don't do this if the nursery is (nearly) full, we'll GC first. + if (cap->r.rCurrentNursery->link != NULL || + cap->r.rNursery->n_blocks == 1) { // paranoia to prevent infinite loop + // if the nursery has only one block. + + ACQUIRE_SM_LOCK + bd = allocGroup( blocks ); + RELEASE_SM_LOCK + cap->r.rNursery->n_blocks += blocks; + + // link the new group into the list + bd->link = cap->r.rCurrentNursery; + bd->u.back = cap->r.rCurrentNursery->u.back; + if (cap->r.rCurrentNursery->u.back != NULL) { + cap->r.rCurrentNursery->u.back->link = bd; + } else { +#if !defined(THREADED_RTS) + ASSERT(g0s0->blocks == cap->r.rCurrentNursery && + g0s0 == cap->r.rNursery); +#endif + cap->r.rNursery->blocks = bd; + } + cap->r.rCurrentNursery->u.back = bd; + + // initialise it as a nursery block. We initialise the + // step, gen_no, and flags field of *every* sub-block in + // this large block, because this is easier than making + // sure that we always find the block head of a large + // block whenever we call Bdescr() (eg. evacuate() and + // isAlive() in the GC would both have to do this, at + // least). + { + bdescr *x; + for (x = bd; x < bd + blocks; x++) { + x->step = cap->r.rNursery; + x->gen_no = 0; + x->flags = 0; + } + } + + // This assert can be a killer if the app is doing lots + // of large block allocations. + IF_DEBUG(sanity, checkNurserySanity(cap->r.rNursery)); + + // now update the nursery to point to the new block + cap->r.rCurrentNursery = bd; + + // we might be unlucky and have another thread get on the + // run queue before us and steal the large block, but in that + // case the thread will just end up requesting another large + // block. + pushOnRunQueue(cap,t); + return rtsFalse; /* not actually GC'ing */ + } + } + + IF_DEBUG(scheduler, + debugBelch("--<< thread %ld (%s) stopped: HeapOverflow\n", + (long)t->id, whatNext_strs[t->what_next])); +#if defined(GRAN) + ASSERT(!is_on_queue(t,CurrentProc)); +#elif defined(PARALLEL_HASKELL) + /* Currently we emit a DESCHEDULE event before GC in GUM. + ToDo: either add separate event to distinguish SYSTEM time from rest + or just nuke this DESCHEDULE (and the following SCHEDULE) */ + if (0 && RtsFlags.ParFlags.ParStats.Full) { + DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC, + GR_DESCHEDULE, t, (StgClosure *)NULL, 0, 0); + emitSchedule = rtsTrue; + } +#endif + + pushOnRunQueue(cap,t); + return rtsTrue; + /* actual GC is done at the end of the while loop in schedule() */ +} + +/* ----------------------------------------------------------------------------- + * Handle a thread that returned to the scheduler with ThreadStackOverflow + * -------------------------------------------------------------------------- */ + +static void +scheduleHandleStackOverflow (Capability *cap, Task *task, StgTSO *t) +{ + IF_DEBUG(scheduler,debugBelch("--<< thread %ld (%s) stopped, StackOverflow\n", + (long)t->id, whatNext_strs[t->what_next])); + /* just adjust the stack for this thread, then pop it back + * on the run queue. + */ + { + /* enlarge the stack */ + StgTSO *new_t = threadStackOverflow(cap, t); + + /* The TSO attached to this Task may have moved, so update the + * pointer to it. + */ + if (task->tso == t) { + task->tso = new_t; + } + pushOnRunQueue(cap,new_t); + } +} + +/* ----------------------------------------------------------------------------- + * Handle a thread that returned to the scheduler with ThreadYielding + * -------------------------------------------------------------------------- */ + +static rtsBool +scheduleHandleYield( Capability *cap, StgTSO *t, nat prev_what_next ) +{ + // Reset the context switch flag. We don't do this just before + // running the thread, because that would mean we would lose ticks + // during GC, which can lead to unfair scheduling (a thread hogs + // the CPU because the tick always arrives during GC). This way + // penalises threads that do a lot of allocation, but that seems + // better than the alternative. + context_switch = 0; + + /* put the thread back on the run queue. Then, if we're ready to + * GC, check whether this is the last task to stop. If so, wake + * up the GC thread. getThread will block during a GC until the + * GC is finished. + */ + IF_DEBUG(scheduler, + if (t->what_next != prev_what_next) { + debugBelch("--<< thread %ld (%s) stopped to switch evaluators\n", + (long)t->id, whatNext_strs[t->what_next]); + } else { + debugBelch("--<< thread %ld (%s) stopped, yielding\n", + (long)t->id, whatNext_strs[t->what_next]); + } + ); + + IF_DEBUG(sanity, + //debugBelch("&& Doing sanity check on yielding TSO %ld.", t->id); + checkTSO(t)); + ASSERT(t->link == END_TSO_QUEUE); + + // Shortcut if we're just switching evaluators: don't bother + // doing stack squeezing (which can be expensive), just run the + // thread. + if (t->what_next != prev_what_next) { + return rtsTrue; + } + +#if defined(GRAN) + ASSERT(!is_on_queue(t,CurrentProc)); + + IF_DEBUG(sanity, + //debugBelch("&& Doing sanity check on all ThreadQueues (and their TSOs)."); + checkThreadQsSanity(rtsTrue)); + +#endif + + addToRunQueue(cap,t); + +#if defined(GRAN) + /* add a ContinueThread event to actually process the thread */ + new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc], + ContinueThread, + t, (StgClosure*)NULL, (rtsSpark*)NULL); + IF_GRAN_DEBUG(bq, + debugBelch("GRAN: eventq and runnableq after adding yielded thread to queue again:\n"); + G_EVENTQ(0); + G_CURR_THREADQ(0)); +#endif + return rtsFalse; +} + +/* ----------------------------------------------------------------------------- + * Handle a thread that returned to the scheduler with ThreadBlocked + * -------------------------------------------------------------------------- */ + +static void +scheduleHandleThreadBlocked( StgTSO *t +#if !defined(GRAN) && !defined(DEBUG) + STG_UNUSED +#endif + ) +{ +#if defined(GRAN) + IF_DEBUG(scheduler, + debugBelch("--<< thread %ld (%p; %s) stopped, blocking on node %p [PE %d] with BQ: \n", + 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))); + if (t->block_info.closure!=(StgClosure*)NULL) print_bq(t->block_info.closure)); + + // ??? needed; should emit block before + IF_DEBUG(gran, + DumpGranEvent(GR_DESCHEDULE, t)); + prune_eventq(t, (StgClosure *)NULL); // prune ContinueThreads for t + /* + ngoq Dogh! + ASSERT(procStatus[CurrentProc]==Busy || + ((procStatus[CurrentProc]==Fetching) && + (t->block_info.closure!=(StgClosure*)NULL))); + if (run_queue_hds[CurrentProc] == END_TSO_QUEUE && + !(!RtsFlags.GranFlags.DoAsyncFetch && + procStatus[CurrentProc]==Fetching)) + procStatus[CurrentProc] = Idle; + */ +#elif defined(PAR) + IF_DEBUG(scheduler, + debugBelch("--<< thread %ld (%p; %s) stopped, blocking on node %p with BQ: \n", + t->id, t, whatNext_strs[t->what_next], t->block_info.closure)); + IF_PAR_DEBUG(bq, + + if (t->block_info.closure!=(StgClosure*)NULL) + print_bq(t->block_info.closure)); + + /* Send a fetch (if BlockedOnGA) and dump event to log file */ + blockThread(t); + + /* whatever we schedule next, we must log that schedule */ + emitSchedule = rtsTrue; + +#else /* !GRAN */ + + // We don't need to do anything. The thread is blocked, and it + // has tidied up its stack and placed itself on whatever queue + // it needs to be on. + +#if !defined(THREADED_RTS) + ASSERT(t->why_blocked != NotBlocked); + // This might not be true under THREADED_RTS: we don't have + // exclusive access to this TSO, so someone might have + // woken it up by now. This actually happens: try + // conc023 +RTS -N2. +#endif + + IF_DEBUG(scheduler, + debugBelch("--<< thread %d (%s) stopped: ", + t->id, whatNext_strs[t->what_next]); + printThreadBlockage(t); + debugBelch("\n")); + + /* Only for dumping event to log file + ToDo: do I need this in GranSim, too? + blockThread(t); + */ +#endif +} + +/* ----------------------------------------------------------------------------- + * Handle a thread that returned to the scheduler with ThreadFinished + * -------------------------------------------------------------------------- */ + +static rtsBool +scheduleHandleThreadFinished (Capability *cap STG_UNUSED, Task *task, StgTSO *t) +{ + /* Need to check whether this was a main thread, and if so, + * return with the return value. + * + * We also end up here if the thread kills itself with an + * uncaught exception, see Exception.cmm. + */ + IF_DEBUG(scheduler,debugBelch("--++ thread %d (%s) finished\n", + t->id, whatNext_strs[t->what_next])); + +#if defined(GRAN) + endThread(t, CurrentProc); // clean-up the thread +#elif defined(PARALLEL_HASKELL) + /* For now all are advisory -- HWL */ + //if(t->priority==AdvisoryPriority) ?? + advisory_thread_count--; // JB: Caution with this counter, buggy! + +# if defined(DIST) + if(t->dist.priority==RevalPriority) + FinishReval(t); +# endif + +# if defined(EDENOLD) + // the thread could still have an outport... (BUG) + if (t->eden.outport != -1) { + // delete the outport for the tso which has finished... + IF_PAR_DEBUG(eden_ports, + debugBelch("WARNING: Scheduler removes outport %d for TSO %d.\n", + t->eden.outport, t->id)); + deleteOPT(t); + } + // thread still in the process (HEAVY BUG! since outport has just been closed...) + if (t->eden.epid != -1) { + IF_PAR_DEBUG(eden_ports, + debugBelch("WARNING: Scheduler removes TSO %d from process %d .\n", + t->id, t->eden.epid)); + removeTSOfromProcess(t); + } +# endif + +# if defined(PAR) + if (RtsFlags.ParFlags.ParStats.Full && + !RtsFlags.ParFlags.ParStats.Suppressed) + DumpEndEvent(CURRENT_PROC, t, rtsFalse /* not mandatory */); + + // t->par only contains statistics: left out for now... + IF_PAR_DEBUG(fish, + debugBelch("**** end thread: ended sparked thread %d (%lx); sparkname: %lx\n", + t->id,t,t->par.sparkname)); +# endif +#endif // PARALLEL_HASKELL + + // + // Check whether the thread that just completed was a bound + // thread, and if so return with the result. + // + // There is an assumption here that all thread completion goes + // through this point; we need to make sure that if a thread + // ends up in the ThreadKilled state, that it stays on the run + // queue so it can be dealt with here. + // + + if (t->bound) { + + if (t->bound != task) { +#if !defined(THREADED_RTS) + // Must be a bound thread that is not the topmost one. Leave + // it on the run queue until the stack has unwound to the + // point where we can deal with this. Leaving it on the run + // queue also ensures that the garbage collector knows about + // this thread and its return value (it gets dropped from the + // all_threads list so there's no other way to find it). + appendToRunQueue(cap,t); + return rtsFalse; +#else + // this cannot happen in the threaded RTS, because a + // bound thread can only be run by the appropriate Task. + barf("finished bound thread that isn't mine"); +#endif + } + + ASSERT(task->tso == t); + + if (t->what_next == ThreadComplete) { + if (task->ret) { + // NOTE: return val is tso->sp[1] (see StgStartup.hc) + *(task->ret) = (StgClosure *)task->tso->sp[1]; + } + task->stat = Success; + } else { + if (task->ret) { + *(task->ret) = NULL; + } + if (sched_state >= SCHED_INTERRUPTING) { + task->stat = Interrupted; + } else { + task->stat = Killed; + } + } +#ifdef DEBUG + removeThreadLabel((StgWord)task->tso->id); +#endif + return rtsTrue; // tells schedule() to return + } + + return rtsFalse; +} + +/* ----------------------------------------------------------------------------- + * Perform a heap census, if PROFILING + * -------------------------------------------------------------------------- */ + +static rtsBool +scheduleDoHeapProfile( rtsBool ready_to_gc STG_UNUSED ) +{ +#if defined(PROFILING) + // When we have +RTS -i0 and we're heap profiling, do a census at + // every GC. This lets us get repeatable runs for debugging. + if (performHeapProfile || + (RtsFlags.ProfFlags.profileInterval==0 && + RtsFlags.ProfFlags.doHeapProfile && ready_to_gc)) { + + // checking black holes is necessary before GC, otherwise + // there may be threads that are unreachable except by the + // blackhole queue, which the GC will consider to be + // deadlocked. + scheduleCheckBlackHoles(&MainCapability); + + IF_DEBUG(scheduler, sched_belch("garbage collecting before heap census")); + GarbageCollect(GetRoots, rtsTrue); + + IF_DEBUG(scheduler, sched_belch("performing heap census")); + heapCensus(); + + performHeapProfile = rtsFalse; + return rtsTrue; // true <=> we already GC'd + } +#endif + return rtsFalse; +} + +/* ----------------------------------------------------------------------------- + * Perform a garbage collection if necessary + * -------------------------------------------------------------------------- */ + +static Capability * +scheduleDoGC (Capability *cap, Task *task USED_IF_THREADS, + rtsBool force_major, void (*get_roots)(evac_fn)) +{ + StgTSO *t; +#ifdef THREADED_RTS + static volatile StgWord waiting_for_gc; + rtsBool was_waiting; + nat i; +#endif + +#ifdef THREADED_RTS + // In order to GC, there must be no threads running Haskell code. + // Therefore, the GC thread needs to hold *all* the capabilities, + // and release them after the GC has completed. + // + // This seems to be the simplest way: previous attempts involved + // making all the threads with capabilities give up their + // capabilities and sleep except for the *last* one, which + // actually did the GC. But it's quite hard to arrange for all + // the other tasks to sleep and stay asleep. + // + + was_waiting = cas(&waiting_for_gc, 0, 1); + if (was_waiting) { + do { + IF_DEBUG(scheduler, sched_belch("someone else is trying to GC...")); + if (cap) yieldCapability(&cap,task); + } while (waiting_for_gc); + return cap; // NOTE: task->cap might have changed here + } + + for (i=0; i < n_capabilities; i++) { + IF_DEBUG(scheduler, sched_belch("ready_to_gc, grabbing all the capabilies (%d/%d)", i, n_capabilities)); + if (cap != &capabilities[i]) { + Capability *pcap = &capabilities[i]; + // we better hope this task doesn't get migrated to + // another Capability while we're waiting for this one. + // It won't, because load balancing happens while we have + // all the Capabilities, but even so it's a slightly + // unsavoury invariant. + task->cap = pcap; + context_switch = 1; + waitForReturnCapability(&pcap, task); + if (pcap != &capabilities[i]) { + barf("scheduleDoGC: got the wrong capability"); + } + } + } + + waiting_for_gc = rtsFalse; +#endif + + /* Kick any transactions which are invalid back to their + * atomically frames. When next scheduled they will try to + * commit, this commit will fail and they will retry. + */ + { + StgTSO *next; + + for (t = all_threads; t != END_TSO_QUEUE; t = next) { + if (t->what_next == ThreadRelocated) { + next = t->link; + } else { + next = t->global_link; + if (t -> trec != NO_TREC && t -> why_blocked == NotBlocked) { + if (!stmValidateNestOfTransactions (t -> trec)) { + IF_DEBUG(stm, sched_belch("trec %p found wasting its time", t)); + + // strip the stack back to the + // ATOMICALLY_FRAME, aborting the (nested) + // transaction, and saving the stack of any + // partially-evaluated thunks on the heap. + raiseAsync_(&capabilities[0], t, NULL, rtsTrue, NULL); + +#ifdef REG_R1 + ASSERT(get_itbl((StgClosure *)t->sp)->type == ATOMICALLY_FRAME); +#endif + } + } + } + } + } + + // so this happens periodically: + if (cap) scheduleCheckBlackHoles(cap); + + IF_DEBUG(scheduler, printAllThreads()); + + /* + * We now have all the capabilities; if we're in an interrupting + * state, then we should take the opportunity to delete all the + * threads in the system. + */ + if (sched_state >= SCHED_INTERRUPTING) { + deleteAllThreads(&capabilities[0]); + sched_state = SCHED_INTERRUPTED; + } + + /* everybody back, start the GC. + * Could do it in this thread, or signal a condition var + * to do it in another thread. Either way, we need to + * broadcast on gc_pending_cond afterward. + */ +#if defined(THREADED_RTS) + IF_DEBUG(scheduler,sched_belch("doing GC")); +#endif + GarbageCollect(get_roots, force_major); + +#if defined(THREADED_RTS) + // release our stash of capabilities. + for (i = 0; i < n_capabilities; i++) { + if (cap != &capabilities[i]) { + task->cap = &capabilities[i]; + releaseCapability(&capabilities[i]); + } + } + if (cap) { + task->cap = cap; + } else { + task->cap = NULL; + } +#endif + +#if defined(GRAN) + /* add a ContinueThread event to continue execution of current thread */ + new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc], + ContinueThread, + t, (StgClosure*)NULL, (rtsSpark*)NULL); + IF_GRAN_DEBUG(bq, + debugBelch("GRAN: eventq and runnableq after Garbage collection:\n\n"); + G_EVENTQ(0); + G_CURR_THREADQ(0)); +#endif /* GRAN */ + + return cap; +} + +/* --------------------------------------------------------------------------- + * rtsSupportsBoundThreads(): is the RTS built to support bound threads? + * used by Control.Concurrent for error checking. + * ------------------------------------------------------------------------- */ + +StgBool +rtsSupportsBoundThreads(void) +{ +#if defined(THREADED_RTS) + return rtsTrue; +#else + return rtsFalse; +#endif +} + +/* --------------------------------------------------------------------------- + * isThreadBound(tso): check whether tso is bound to an OS thread. + * ------------------------------------------------------------------------- */ + +StgBool +isThreadBound(StgTSO* tso USED_IF_THREADS) +{ +#if defined(THREADED_RTS) + return (tso->bound != NULL); +#endif + return rtsFalse; +} + +/* --------------------------------------------------------------------------- + * Singleton fork(). Do not copy any running threads. + * ------------------------------------------------------------------------- */ + +#if !defined(mingw32_HOST_OS) +#define FORKPROCESS_PRIMOP_SUPPORTED +#endif + +#ifdef FORKPROCESS_PRIMOP_SUPPORTED +static void +deleteThread_(Capability *cap, StgTSO *tso); +#endif +StgInt +forkProcess(HsStablePtr *entry +#ifndef FORKPROCESS_PRIMOP_SUPPORTED + STG_UNUSED +#endif + ) +{ +#ifdef FORKPROCESS_PRIMOP_SUPPORTED + Task *task; + pid_t pid; + StgTSO* t,*next; + Capability *cap; + +#if defined(THREADED_RTS) + if (RtsFlags.ParFlags.nNodes > 1) { + errorBelch("forking not supported with +RTS -N<n> greater than 1"); + stg_exit(EXIT_FAILURE); + } +#endif + + IF_DEBUG(scheduler,sched_belch("forking!")); + + // ToDo: for SMP, we should probably acquire *all* the capabilities + cap = rts_lock(); + + pid = fork(); + + if (pid) { // parent + + // just return the pid + rts_unlock(cap); + return pid; + + } else { // child + + // Now, all OS threads except the thread that forked are + // stopped. We need to stop all Haskell threads, including + // those involved in foreign calls. Also we need to delete + // all Tasks, because they correspond to OS threads that are + // now gone. + + for (t = all_threads; t != END_TSO_QUEUE; t = next) { + if (t->what_next == ThreadRelocated) { + next = t->link; + } else { + next = t->global_link; + // don't allow threads to catch the ThreadKilled + // exception, but we do want to raiseAsync() because these + // threads may be evaluating thunks that we need later. + deleteThread_(cap,t); + } + } + + // Empty the run queue. It seems tempting to let all the + // killed threads stay on the run queue as zombies to be + // cleaned up later, but some of them correspond to bound + // threads for which the corresponding Task does not exist. + cap->run_queue_hd = END_TSO_QUEUE; + cap->run_queue_tl = END_TSO_QUEUE; + + // Any suspended C-calling Tasks are no more, their OS threads + // don't exist now: + cap->suspended_ccalling_tasks = NULL; + + // Empty the all_threads list. Otherwise, the garbage + // collector may attempt to resurrect some of these threads. + all_threads = END_TSO_QUEUE; + + // Wipe the task list, except the current Task. + ACQUIRE_LOCK(&sched_mutex); + for (task = all_tasks; task != NULL; task=task->all_link) { + if (task != cap->running_task) { + discardTask(task); + } + } + RELEASE_LOCK(&sched_mutex); + +#if defined(THREADED_RTS) + // Wipe our spare workers list, they no longer exist. New + // workers will be created if necessary. + cap->spare_workers = NULL; + cap->returning_tasks_hd = NULL; + cap->returning_tasks_tl = NULL; +#endif + + cap = rts_evalStableIO(cap, entry, NULL); // run the action + rts_checkSchedStatus("forkProcess",cap); + + rts_unlock(cap); + hs_exit(); // clean up and exit + stg_exit(EXIT_SUCCESS); + } +#else /* !FORKPROCESS_PRIMOP_SUPPORTED */ + barf("forkProcess#: primop not supported on this platform, sorry!\n"); + return -1; +#endif +} + +/* --------------------------------------------------------------------------- + * Delete all the threads in the system + * ------------------------------------------------------------------------- */ + +static void +deleteAllThreads ( Capability *cap ) +{ + StgTSO* t, *next; + IF_DEBUG(scheduler,sched_belch("deleting all threads")); + for (t = all_threads; t != END_TSO_QUEUE; t = next) { + if (t->what_next == ThreadRelocated) { + next = t->link; + } else { + next = t->global_link; + deleteThread(cap,t); + } + } + + // The run queue now contains a bunch of ThreadKilled threads. We + // must not throw these away: the main thread(s) will be in there + // somewhere, and the main scheduler loop has to deal with it. + // Also, the run queue is the only thing keeping these threads from + // being GC'd, and we don't want the "main thread has been GC'd" panic. + +#if !defined(THREADED_RTS) + ASSERT(blocked_queue_hd == END_TSO_QUEUE); + ASSERT(sleeping_queue == END_TSO_QUEUE); +#endif +} + +/* ----------------------------------------------------------------------------- + Managing the suspended_ccalling_tasks list. + Locks required: sched_mutex + -------------------------------------------------------------------------- */ + +STATIC_INLINE void +suspendTask (Capability *cap, Task *task) +{ + ASSERT(task->next == NULL && task->prev == NULL); + task->next = cap->suspended_ccalling_tasks; + task->prev = NULL; + if (cap->suspended_ccalling_tasks) { + cap->suspended_ccalling_tasks->prev = task; + } + cap->suspended_ccalling_tasks = task; +} + +STATIC_INLINE void +recoverSuspendedTask (Capability *cap, Task *task) +{ + if (task->prev) { + task->prev->next = task->next; + } else { + ASSERT(cap->suspended_ccalling_tasks == task); + cap->suspended_ccalling_tasks = task->next; + } + if (task->next) { + task->next->prev = task->prev; + } + task->next = task->prev = NULL; +} + +/* --------------------------------------------------------------------------- + * Suspending & resuming Haskell threads. + * + * When making a "safe" call to C (aka _ccall_GC), the task gives back + * its capability before calling the C function. This allows another + * task to pick up the capability and carry on running Haskell + * threads. It also means that if the C call blocks, it won't lock + * the whole system. + * + * The Haskell thread making the C call is put to sleep for the + * duration of the call, on the susepended_ccalling_threads queue. We + * give out a token to the task, which it can use to resume the thread + * on return from the C function. + * ------------------------------------------------------------------------- */ + +void * +suspendThread (StgRegTable *reg) +{ + Capability *cap; + int saved_errno = errno; + StgTSO *tso; + Task *task; + + /* assume that *reg is a pointer to the StgRegTable part of a Capability. + */ + cap = regTableToCapability(reg); + + task = cap->running_task; + tso = cap->r.rCurrentTSO; + + IF_DEBUG(scheduler, + sched_belch("thread %d did a safe foreign call", cap->r.rCurrentTSO->id)); + + // XXX this might not be necessary --SDM + tso->what_next = ThreadRunGHC; + + threadPaused(cap,tso); + + if(tso->blocked_exceptions == NULL) { + tso->why_blocked = BlockedOnCCall; + tso->blocked_exceptions = END_TSO_QUEUE; + } else { + tso->why_blocked = BlockedOnCCall_NoUnblockExc; + } + + // Hand back capability + task->suspended_tso = tso; + + ACQUIRE_LOCK(&cap->lock); + + suspendTask(cap,task); + cap->in_haskell = rtsFalse; + releaseCapability_(cap); + + RELEASE_LOCK(&cap->lock); + +#if defined(THREADED_RTS) + /* Preparing to leave the RTS, so ensure there's a native thread/task + waiting to take over. + */ + IF_DEBUG(scheduler, sched_belch("thread %d: leaving RTS", tso->id)); +#endif + + errno = saved_errno; + return task; +} + +StgRegTable * +resumeThread (void *task_) +{ + StgTSO *tso; + Capability *cap; + int saved_errno = errno; + Task *task = task_; + + cap = task->cap; + // Wait for permission to re-enter the RTS with the result. + waitForReturnCapability(&cap,task); + // we might be on a different capability now... but if so, our + // entry on the suspended_ccalling_tasks list will also have been + // migrated. + + // Remove the thread from the suspended list + recoverSuspendedTask(cap,task); + + tso = task->suspended_tso; + task->suspended_tso = NULL; + tso->link = END_TSO_QUEUE; + IF_DEBUG(scheduler, sched_belch("thread %d: re-entering RTS", tso->id)); + + if (tso->why_blocked == BlockedOnCCall) { + awakenBlockedQueue(cap,tso->blocked_exceptions); + tso->blocked_exceptions = NULL; + } + + /* Reset blocking status */ + tso->why_blocked = NotBlocked; + + cap->r.rCurrentTSO = tso; + cap->in_haskell = rtsTrue; + errno = saved_errno; + + /* We might have GC'd, mark the TSO dirty again */ + dirtyTSO(tso); + + IF_DEBUG(sanity, checkTSO(tso)); + + return &cap->r; +} + +/* --------------------------------------------------------------------------- + * Comparing Thread ids. + * + * This is used from STG land in the implementation of the + * instances of Eq/Ord for ThreadIds. + * ------------------------------------------------------------------------ */ + +int +cmp_thread(StgPtr tso1, StgPtr tso2) +{ + StgThreadID id1 = ((StgTSO *)tso1)->id; + StgThreadID id2 = ((StgTSO *)tso2)->id; + + if (id1 < id2) return (-1); + if (id1 > id2) return 1; + return 0; +} + +/* --------------------------------------------------------------------------- + * Fetching the ThreadID from an StgTSO. + * + * This is used in the implementation of Show for ThreadIds. + * ------------------------------------------------------------------------ */ +int +rts_getThreadId(StgPtr tso) +{ + return ((StgTSO *)tso)->id; +} + +#ifdef DEBUG +void +labelThread(StgPtr tso, char *label) +{ + int len; + void *buf; + + /* Caveat: Once set, you can only set the thread name to "" */ + len = strlen(label)+1; + buf = stgMallocBytes(len * sizeof(char), "Schedule.c:labelThread()"); + strncpy(buf,label,len); + /* Update will free the old memory for us */ + updateThreadLabel(((StgTSO *)tso)->id,buf); +} +#endif /* DEBUG */ + +/* --------------------------------------------------------------------------- + Create a new thread. + + The new thread starts with the given stack size. Before the + scheduler can run, however, this thread needs to have a closure + (and possibly some arguments) pushed on its stack. See + pushClosure() in Schedule.h. + + createGenThread() and createIOThread() (in SchedAPI.h) are + convenient packaged versions of this function. + + currently pri (priority) is only used in a GRAN setup -- HWL + ------------------------------------------------------------------------ */ +#if defined(GRAN) +/* currently pri (priority) is only used in a GRAN setup -- HWL */ +StgTSO * +createThread(nat size, StgInt pri) +#else +StgTSO * +createThread(Capability *cap, nat size) +#endif +{ + StgTSO *tso; + nat stack_size; + + /* sched_mutex is *not* required */ + + /* First check whether we should create a thread at all */ +#if defined(PARALLEL_HASKELL) + /* check that no more than RtsFlags.ParFlags.maxThreads threads are created */ + if (advisory_thread_count >= RtsFlags.ParFlags.maxThreads) { + threadsIgnored++; + debugBelch("{createThread}Daq ghuH: refusing to create another thread; no more than %d threads allowed (currently %d)\n", + RtsFlags.ParFlags.maxThreads, advisory_thread_count); + return END_TSO_QUEUE; + } + threadsCreated++; +#endif + +#if defined(GRAN) + ASSERT(!RtsFlags.GranFlags.Light || CurrentProc==0); +#endif + + // ToDo: check whether size = stack_size - TSO_STRUCT_SIZEW + + /* catch ridiculously small stack sizes */ + if (size < MIN_STACK_WORDS + TSO_STRUCT_SIZEW) { + size = MIN_STACK_WORDS + TSO_STRUCT_SIZEW; + } + + stack_size = size - TSO_STRUCT_SIZEW; + + tso = (StgTSO *)allocateLocal(cap, size); + TICK_ALLOC_TSO(stack_size, 0); + + SET_HDR(tso, &stg_TSO_info, CCS_SYSTEM); +#if defined(GRAN) + SET_GRAN_HDR(tso, ThisPE); +#endif + + // Always start with the compiled code evaluator + tso->what_next = ThreadRunGHC; + + tso->why_blocked = NotBlocked; + tso->blocked_exceptions = NULL; + tso->flags = TSO_DIRTY; + + tso->saved_errno = 0; + tso->bound = NULL; + tso->cap = cap; + + tso->stack_size = stack_size; + tso->max_stack_size = round_to_mblocks(RtsFlags.GcFlags.maxStkSize) + - TSO_STRUCT_SIZEW; + tso->sp = (P_)&(tso->stack) + stack_size; + + tso->trec = NO_TREC; + +#ifdef PROFILING + tso->prof.CCCS = CCS_MAIN; +#endif + + /* put a stop frame on the stack */ + tso->sp -= sizeofW(StgStopFrame); + SET_HDR((StgClosure*)tso->sp,(StgInfoTable *)&stg_stop_thread_info,CCS_SYSTEM); + tso->link = END_TSO_QUEUE; + + // ToDo: check this +#if defined(GRAN) + /* uses more flexible routine in GranSim */ + insertThread(tso, CurrentProc); +#else + /* In a non-GranSim setup the pushing of a TSO onto the runq is separated + * from its creation + */ +#endif + +#if defined(GRAN) + if (RtsFlags.GranFlags.GranSimStats.Full) + DumpGranEvent(GR_START,tso); +#elif defined(PARALLEL_HASKELL) + if (RtsFlags.ParFlags.ParStats.Full) + DumpGranEvent(GR_STARTQ,tso); + /* HACk to avoid SCHEDULE + LastTSO = tso; */ +#endif + + /* Link the new thread on the global thread list. + */ + ACQUIRE_LOCK(&sched_mutex); + tso->id = next_thread_id++; // while we have the mutex + tso->global_link = all_threads; + all_threads = tso; + RELEASE_LOCK(&sched_mutex); + +#if defined(DIST) + tso->dist.priority = MandatoryPriority; //by default that is... +#endif + +#if defined(GRAN) + tso->gran.pri = pri; +# if defined(DEBUG) + tso->gran.magic = TSO_MAGIC; // debugging only +# endif + tso->gran.sparkname = 0; + tso->gran.startedat = CURRENT_TIME; + tso->gran.exported = 0; + tso->gran.basicblocks = 0; + tso->gran.allocs = 0; + tso->gran.exectime = 0; + tso->gran.fetchtime = 0; + tso->gran.fetchcount = 0; + tso->gran.blocktime = 0; + tso->gran.blockcount = 0; + tso->gran.blockedat = 0; + tso->gran.globalsparks = 0; + tso->gran.localsparks = 0; + if (RtsFlags.GranFlags.Light) + tso->gran.clock = Now; /* local clock */ + else + tso->gran.clock = 0; + + IF_DEBUG(gran,printTSO(tso)); +#elif defined(PARALLEL_HASKELL) +# if defined(DEBUG) + tso->par.magic = TSO_MAGIC; // debugging only +# endif + tso->par.sparkname = 0; + tso->par.startedat = CURRENT_TIME; + tso->par.exported = 0; + tso->par.basicblocks = 0; + tso->par.allocs = 0; + tso->par.exectime = 0; + tso->par.fetchtime = 0; + tso->par.fetchcount = 0; + tso->par.blocktime = 0; + tso->par.blockcount = 0; + tso->par.blockedat = 0; + tso->par.globalsparks = 0; + tso->par.localsparks = 0; +#endif + +#if defined(GRAN) + globalGranStats.tot_threads_created++; + globalGranStats.threads_created_on_PE[CurrentProc]++; + globalGranStats.tot_sq_len += spark_queue_len(CurrentProc); + globalGranStats.tot_sq_probes++; +#elif defined(PARALLEL_HASKELL) + // collect parallel global statistics (currently done together with GC stats) + if (RtsFlags.ParFlags.ParStats.Global && + RtsFlags.GcFlags.giveStats > NO_GC_STATS) { + //debugBelch("Creating thread %d @ %11.2f\n", tso->id, usertime()); + globalParStats.tot_threads_created++; + } +#endif + +#if defined(GRAN) + IF_GRAN_DEBUG(pri, + sched_belch("==__ schedule: Created TSO %d (%p);", + CurrentProc, tso, tso->id)); +#elif defined(PARALLEL_HASKELL) + IF_PAR_DEBUG(verbose, + sched_belch("==__ schedule: Created TSO %d (%p); %d threads active", + (long)tso->id, tso, advisory_thread_count)); +#else + IF_DEBUG(scheduler,sched_belch("created thread %ld, stack size = %lx words", + (long)tso->id, (long)tso->stack_size)); +#endif + return tso; +} + +#if defined(PAR) +/* RFP: + all parallel thread creation calls should fall through the following routine. +*/ +StgTSO * +createThreadFromSpark(rtsSpark spark) +{ StgTSO *tso; + ASSERT(spark != (rtsSpark)NULL); +// JB: TAKE CARE OF THIS COUNTER! BUGGY + if (advisory_thread_count >= RtsFlags.ParFlags.maxThreads) + { threadsIgnored++; + barf("{createSparkThread}Daq ghuH: refusing to create another thread; no more than %d threads allowed (currently %d)", + RtsFlags.ParFlags.maxThreads, advisory_thread_count); + return END_TSO_QUEUE; + } + else + { threadsCreated++; + tso = createThread(RtsFlags.GcFlags.initialStkSize); + if (tso==END_TSO_QUEUE) + barf("createSparkThread: Cannot create TSO"); +#if defined(DIST) + tso->priority = AdvisoryPriority; +#endif + pushClosure(tso,spark); + addToRunQueue(tso); + advisory_thread_count++; // JB: TAKE CARE OF THIS COUNTER! BUGGY + } + return tso; +} +#endif + +/* + Turn a spark into a thread. + ToDo: fix for SMP (needs to acquire SCHED_MUTEX!) +*/ +#if 0 +StgTSO * +activateSpark (rtsSpark spark) +{ + StgTSO *tso; + + tso = createSparkThread(spark); + if (RtsFlags.ParFlags.ParStats.Full) { + //ASSERT(run_queue_hd == END_TSO_QUEUE); // I think ... + IF_PAR_DEBUG(verbose, + debugBelch("==^^ activateSpark: turning spark of closure %p (%s) into a thread\n", + (StgClosure *)spark, info_type((StgClosure *)spark))); + } + // ToDo: fwd info on local/global spark to thread -- HWL + // tso->gran.exported = spark->exported; + // tso->gran.locked = !spark->global; + // tso->gran.sparkname = spark->name; + + return tso; +} +#endif + +/* --------------------------------------------------------------------------- + * scheduleThread() + * + * scheduleThread puts a thread on the end of the runnable queue. + * This will usually be done immediately after a thread is created. + * The caller of scheduleThread must create the thread using e.g. + * createThread and push an appropriate closure + * on this thread's stack before the scheduler is invoked. + * ------------------------------------------------------------------------ */ + +void +scheduleThread(Capability *cap, StgTSO *tso) +{ + // The thread goes at the *end* of the run-queue, to avoid possible + // starvation of any threads already on the queue. + appendToRunQueue(cap,tso); +} + +void +scheduleThreadOn(Capability *cap, StgWord cpu USED_IF_THREADS, StgTSO *tso) +{ +#if defined(THREADED_RTS) + tso->flags |= TSO_LOCKED; // we requested explicit affinity; don't + // move this thread from now on. + cpu %= RtsFlags.ParFlags.nNodes; + if (cpu == cap->no) { + appendToRunQueue(cap,tso); + } else { + Capability *target_cap = &capabilities[cpu]; + if (tso->bound) { + tso->bound->cap = target_cap; + } + tso->cap = target_cap; + wakeupThreadOnCapability(target_cap,tso); + } +#else + appendToRunQueue(cap,tso); +#endif +} + +Capability * +scheduleWaitThread (StgTSO* tso, /*[out]*/HaskellObj* ret, Capability *cap) +{ + Task *task; + + // We already created/initialised the Task + task = cap->running_task; + + // This TSO is now a bound thread; make the Task and TSO + // point to each other. + tso->bound = task; + tso->cap = cap; + + task->tso = tso; + task->ret = ret; + task->stat = NoStatus; + + appendToRunQueue(cap,tso); + + IF_DEBUG(scheduler, sched_belch("new bound thread (%d)", tso->id)); + +#if defined(GRAN) + /* GranSim specific init */ + CurrentTSO = m->tso; // the TSO to run + procStatus[MainProc] = Busy; // status of main PE + CurrentProc = MainProc; // PE to run it on +#endif + + cap = schedule(cap,task); + + ASSERT(task->stat != NoStatus); + ASSERT_FULL_CAPABILITY_INVARIANTS(cap,task); + + IF_DEBUG(scheduler, sched_belch("bound thread (%d) finished", task->tso->id)); + return cap; +} + +/* ---------------------------------------------------------------------------- + * Starting Tasks + * ------------------------------------------------------------------------- */ + +#if defined(THREADED_RTS) +void +workerStart(Task *task) +{ + Capability *cap; + + // See startWorkerTask(). + ACQUIRE_LOCK(&task->lock); + cap = task->cap; + RELEASE_LOCK(&task->lock); + + // set the thread-local pointer to the Task: + taskEnter(task); + + // schedule() runs without a lock. + cap = schedule(cap,task); + + // On exit from schedule(), we have a Capability. + releaseCapability(cap); + taskStop(task); +} +#endif + +/* --------------------------------------------------------------------------- + * initScheduler() + * + * Initialise the scheduler. This resets all the queues - if the + * queues contained any threads, they'll be garbage collected at the + * next pass. + * + * ------------------------------------------------------------------------ */ + +void +initScheduler(void) +{ +#if defined(GRAN) + nat i; + for (i=0; i<=MAX_PROC; i++) { + run_queue_hds[i] = END_TSO_QUEUE; + run_queue_tls[i] = END_TSO_QUEUE; + blocked_queue_hds[i] = END_TSO_QUEUE; + blocked_queue_tls[i] = END_TSO_QUEUE; + ccalling_threadss[i] = END_TSO_QUEUE; + blackhole_queue[i] = END_TSO_QUEUE; + sleeping_queue = END_TSO_QUEUE; + } +#elif !defined(THREADED_RTS) + blocked_queue_hd = END_TSO_QUEUE; + blocked_queue_tl = END_TSO_QUEUE; + sleeping_queue = END_TSO_QUEUE; +#endif + + blackhole_queue = END_TSO_QUEUE; + all_threads = END_TSO_QUEUE; + + context_switch = 0; + sched_state = SCHED_RUNNING; + + RtsFlags.ConcFlags.ctxtSwitchTicks = + RtsFlags.ConcFlags.ctxtSwitchTime / TICK_MILLISECS; + +#if defined(THREADED_RTS) + /* Initialise the mutex and condition variables used by + * the scheduler. */ + initMutex(&sched_mutex); +#endif + + ACQUIRE_LOCK(&sched_mutex); + + /* A capability holds the state a native thread needs in + * order to execute STG code. At least one capability is + * floating around (only THREADED_RTS builds have more than one). + */ + initCapabilities(); + + initTaskManager(); + +#if defined(THREADED_RTS) || defined(PARALLEL_HASKELL) + initSparkPools(); +#endif + +#if defined(THREADED_RTS) + /* + * Eagerly start one worker to run each Capability, except for + * Capability 0. The idea is that we're probably going to start a + * bound thread on Capability 0 pretty soon, so we don't want a + * worker task hogging it. + */ + { + nat i; + Capability *cap; + for (i = 1; i < n_capabilities; i++) { + cap = &capabilities[i]; + ACQUIRE_LOCK(&cap->lock); + startWorkerTask(cap, workerStart); + RELEASE_LOCK(&cap->lock); + } + } +#endif + + RELEASE_LOCK(&sched_mutex); +} + +void +exitScheduler( void ) +{ + Task *task = NULL; + +#if defined(THREADED_RTS) + ACQUIRE_LOCK(&sched_mutex); + task = newBoundTask(); + RELEASE_LOCK(&sched_mutex); +#endif + + // If we haven't killed all the threads yet, do it now. + if (sched_state < SCHED_INTERRUPTED) { + sched_state = SCHED_INTERRUPTING; + scheduleDoGC(NULL,task,rtsFalse,GetRoots); + } + sched_state = SCHED_SHUTTING_DOWN; + +#if defined(THREADED_RTS) + { + nat i; + + for (i = 0; i < n_capabilities; i++) { + shutdownCapability(&capabilities[i], task); + } + boundTaskExiting(task); + stopTaskManager(); + } +#endif +} + +/* --------------------------------------------------------------------------- + Where are the roots that we know about? + + - all the threads on the runnable queue + - all the threads on the blocked queue + - all the threads on the sleeping queue + - all the thread currently executing a _ccall_GC + - all the "main threads" + + ------------------------------------------------------------------------ */ + +/* This has to be protected either by the scheduler monitor, or by the + garbage collection monitor (probably the latter). + KH @ 25/10/99 +*/ + +void +GetRoots( evac_fn evac ) +{ + nat i; + Capability *cap; + Task *task; + +#if defined(GRAN) + for (i=0; i<=RtsFlags.GranFlags.proc; i++) { + if ((run_queue_hds[i] != END_TSO_QUEUE) && ((run_queue_hds[i] != NULL))) + evac((StgClosure **)&run_queue_hds[i]); + if ((run_queue_tls[i] != END_TSO_QUEUE) && ((run_queue_tls[i] != NULL))) + evac((StgClosure **)&run_queue_tls[i]); + + if ((blocked_queue_hds[i] != END_TSO_QUEUE) && ((blocked_queue_hds[i] != NULL))) + evac((StgClosure **)&blocked_queue_hds[i]); + if ((blocked_queue_tls[i] != END_TSO_QUEUE) && ((blocked_queue_tls[i] != NULL))) + evac((StgClosure **)&blocked_queue_tls[i]); + if ((ccalling_threadss[i] != END_TSO_QUEUE) && ((ccalling_threadss[i] != NULL))) + evac((StgClosure **)&ccalling_threads[i]); + } + + markEventQueue(); + +#else /* !GRAN */ + + for (i = 0; i < n_capabilities; i++) { + cap = &capabilities[i]; + evac((StgClosure **)(void *)&cap->run_queue_hd); + evac((StgClosure **)(void *)&cap->run_queue_tl); +#if defined(THREADED_RTS) + evac((StgClosure **)(void *)&cap->wakeup_queue_hd); + evac((StgClosure **)(void *)&cap->wakeup_queue_tl); +#endif + for (task = cap->suspended_ccalling_tasks; task != NULL; + task=task->next) { + IF_DEBUG(scheduler,sched_belch("evac'ing suspended TSO %d", task->suspended_tso->id)); + evac((StgClosure **)(void *)&task->suspended_tso); + } + + } + + +#if !defined(THREADED_RTS) + evac((StgClosure **)(void *)&blocked_queue_hd); + evac((StgClosure **)(void *)&blocked_queue_tl); + evac((StgClosure **)(void *)&sleeping_queue); +#endif +#endif + + // evac((StgClosure **)&blackhole_queue); + +#if defined(THREADED_RTS) || defined(PARALLEL_HASKELL) || defined(GRAN) + markSparkQueue(evac); +#endif + +#if defined(RTS_USER_SIGNALS) + // mark the signal handlers (signals should be already blocked) + markSignalHandlers(evac); +#endif +} + +/* ----------------------------------------------------------------------------- + performGC + + This is the interface to the garbage collector from Haskell land. + We provide this so that external C code can allocate and garbage + collect when called from Haskell via _ccall_GC. + + It might be useful to provide an interface whereby the programmer + can specify more roots (ToDo). + + This needs to be protected by the GC condition variable above. KH. + -------------------------------------------------------------------------- */ + +static void (*extra_roots)(evac_fn); + +static void +performGC_(rtsBool force_major, void (*get_roots)(evac_fn)) +{ + Task *task = myTask(); + + if (task == NULL) { + ACQUIRE_LOCK(&sched_mutex); + task = newBoundTask(); + RELEASE_LOCK(&sched_mutex); + scheduleDoGC(NULL,task,force_major, get_roots); + boundTaskExiting(task); + } else { + scheduleDoGC(NULL,task,force_major, get_roots); + } +} + +void +performGC(void) +{ + performGC_(rtsFalse, GetRoots); +} + +void +performMajorGC(void) +{ + performGC_(rtsTrue, GetRoots); +} + +static void +AllRoots(evac_fn evac) +{ + GetRoots(evac); // the scheduler's roots + extra_roots(evac); // the user's roots +} + +void +performGCWithRoots(void (*get_roots)(evac_fn)) +{ + extra_roots = get_roots; + performGC_(rtsFalse, AllRoots); +} + +/* ----------------------------------------------------------------------------- + Stack overflow + + If the thread has reached its maximum stack size, then raise the + StackOverflow exception in the offending thread. Otherwise + relocate the TSO into a larger chunk of memory and adjust its stack + size appropriately. + -------------------------------------------------------------------------- */ + +static StgTSO * +threadStackOverflow(Capability *cap, StgTSO *tso) +{ + nat new_stack_size, stack_words; + lnat new_tso_size; + StgPtr new_sp; + StgTSO *dest; + + IF_DEBUG(sanity,checkTSO(tso)); + if (tso->stack_size >= tso->max_stack_size) { + + IF_DEBUG(gc, + debugBelch("@@ threadStackOverflow of TSO %ld (%p): stack too large (now %ld; max is %ld)\n", + (long)tso->id, tso, (long)tso->stack_size, (long)tso->max_stack_size); + /* If we're debugging, just print out the top of the stack */ + printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size, + tso->sp+64))); + + /* Send this thread the StackOverflow exception */ + raiseAsync(cap, tso, (StgClosure *)stackOverflow_closure); + return tso; + } + + /* Try to double the current stack size. If that takes us over the + * maximum stack size for this thread, then use the maximum instead. + * Finally round up so the TSO ends up as a whole number of blocks. + */ + new_stack_size = stg_min(tso->stack_size * 2, tso->max_stack_size); + new_tso_size = (lnat)BLOCK_ROUND_UP(new_stack_size * sizeof(W_) + + TSO_STRUCT_SIZE)/sizeof(W_); + new_tso_size = round_to_mblocks(new_tso_size); /* Be MBLOCK-friendly */ + new_stack_size = new_tso_size - TSO_STRUCT_SIZEW; + + IF_DEBUG(scheduler, sched_belch("increasing stack size from %ld words to %d.\n", (long)tso->stack_size, new_stack_size)); + + dest = (StgTSO *)allocate(new_tso_size); + TICK_ALLOC_TSO(new_stack_size,0); + + /* copy the TSO block and the old stack into the new area */ + memcpy(dest,tso,TSO_STRUCT_SIZE); + stack_words = tso->stack + tso->stack_size - tso->sp; + new_sp = (P_)dest + new_tso_size - stack_words; + memcpy(new_sp, tso->sp, stack_words * sizeof(W_)); + + /* relocate the stack pointers... */ + dest->sp = new_sp; + dest->stack_size = new_stack_size; + + /* Mark the old TSO as relocated. We have to check for relocated + * TSOs in the garbage collector and any primops that deal with TSOs. + * + * It's important to set the sp value to just beyond the end + * of the stack, so we don't attempt to scavenge any part of the + * dead TSO's stack. + */ + tso->what_next = ThreadRelocated; + tso->link = dest; + tso->sp = (P_)&(tso->stack[tso->stack_size]); + tso->why_blocked = NotBlocked; + + IF_PAR_DEBUG(verbose, + debugBelch("@@ threadStackOverflow of TSO %d (now at %p): stack size increased to %ld\n", + tso->id, tso, tso->stack_size); + /* If we're debugging, just print out the top of the stack */ + printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size, + tso->sp+64))); + + IF_DEBUG(sanity,checkTSO(tso)); +#if 0 + IF_DEBUG(scheduler,printTSO(dest)); +#endif + + return dest; +} + +/* --------------------------------------------------------------------------- + Wake up a queue that was blocked on some resource. + ------------------------------------------------------------------------ */ + +#if defined(GRAN) +STATIC_INLINE void +unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node ) +{ +} +#elif defined(PARALLEL_HASKELL) +STATIC_INLINE void +unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node ) +{ + /* write RESUME events to log file and + update blocked and fetch time (depending on type of the orig closure) */ + if (RtsFlags.ParFlags.ParStats.Full) { + DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC, + GR_RESUMEQ, ((StgTSO *)bqe), ((StgTSO *)bqe)->block_info.closure, + 0, 0 /* spark_queue_len(ADVISORY_POOL) */); + if (emptyRunQueue()) + emitSchedule = rtsTrue; + + switch (get_itbl(node)->type) { + case FETCH_ME_BQ: + ((StgTSO *)bqe)->par.fetchtime += CURRENT_TIME-((StgTSO *)bqe)->par.blockedat; + break; + case RBH: + case FETCH_ME: + case BLACKHOLE_BQ: + ((StgTSO *)bqe)->par.blocktime += CURRENT_TIME-((StgTSO *)bqe)->par.blockedat; + break; +#ifdef DIST + case MVAR: + break; +#endif + default: + barf("{unblockOne}Daq Qagh: unexpected closure in blocking queue"); + } + } +} +#endif + +#if defined(GRAN) +StgBlockingQueueElement * +unblockOne(StgBlockingQueueElement *bqe, StgClosure *node) +{ + StgTSO *tso; + PEs node_loc, tso_loc; + + node_loc = where_is(node); // should be lifted out of loop + tso = (StgTSO *)bqe; // wastes an assignment to get the type right + tso_loc = where_is((StgClosure *)tso); + if (IS_LOCAL_TO(PROCS(node),tso_loc)) { // TSO is local + /* !fake_fetch => TSO is on CurrentProc is same as IS_LOCAL_TO */ + ASSERT(CurrentProc!=node_loc || tso_loc==CurrentProc); + CurrentTime[CurrentProc] += RtsFlags.GranFlags.Costs.lunblocktime; + // insertThread(tso, node_loc); + new_event(tso_loc, tso_loc, CurrentTime[CurrentProc], + ResumeThread, + tso, node, (rtsSpark*)NULL); + tso->link = END_TSO_QUEUE; // overwrite link just to be sure + // len_local++; + // len++; + } else { // TSO is remote (actually should be FMBQ) + CurrentTime[CurrentProc] += RtsFlags.GranFlags.Costs.mpacktime + + RtsFlags.GranFlags.Costs.gunblocktime + + RtsFlags.GranFlags.Costs.latency; + new_event(tso_loc, CurrentProc, CurrentTime[CurrentProc], + UnblockThread, + tso, node, (rtsSpark*)NULL); + tso->link = END_TSO_QUEUE; // overwrite link just to be sure + // len++; + } + /* the thread-queue-overhead is accounted for in either Resume or UnblockThread */ + IF_GRAN_DEBUG(bq, + debugBelch(" %s TSO %d (%p) [PE %d] (block_info.closure=%p) (next=%p) ,", + (node_loc==tso_loc ? "Local" : "Global"), + tso->id, tso, CurrentProc, tso->block_info.closure, tso->link)); + tso->block_info.closure = NULL; + IF_DEBUG(scheduler,debugBelch("-- Waking up thread %ld (%p)\n", + tso->id, tso)); +} +#elif defined(PARALLEL_HASKELL) +StgBlockingQueueElement * +unblockOne(StgBlockingQueueElement *bqe, StgClosure *node) +{ + StgBlockingQueueElement *next; + + switch (get_itbl(bqe)->type) { + case TSO: + ASSERT(((StgTSO *)bqe)->why_blocked != NotBlocked); + /* if it's a TSO just push it onto the run_queue */ + next = bqe->link; + ((StgTSO *)bqe)->link = END_TSO_QUEUE; // debugging? + APPEND_TO_RUN_QUEUE((StgTSO *)bqe); + threadRunnable(); + unblockCount(bqe, node); + /* reset blocking status after dumping event */ + ((StgTSO *)bqe)->why_blocked = NotBlocked; + break; + + case BLOCKED_FETCH: + /* if it's a BLOCKED_FETCH put it on the PendingFetches list */ + next = bqe->link; + bqe->link = (StgBlockingQueueElement *)PendingFetches; + PendingFetches = (StgBlockedFetch *)bqe; + break; + +# if defined(DEBUG) + /* can ignore this case in a non-debugging setup; + see comments on RBHSave closures above */ + case CONSTR: + /* check that the closure is an RBHSave closure */ + ASSERT(get_itbl((StgClosure *)bqe) == &stg_RBH_Save_0_info || + get_itbl((StgClosure *)bqe) == &stg_RBH_Save_1_info || + get_itbl((StgClosure *)bqe) == &stg_RBH_Save_2_info); + break; + + default: + barf("{unblockOne}Daq Qagh: Unexpected IP (%#lx; %s) in blocking queue at %#lx\n", + get_itbl((StgClosure *)bqe), info_type((StgClosure *)bqe), + (StgClosure *)bqe); +# endif + } + IF_PAR_DEBUG(bq, debugBelch(", %p (%s)\n", bqe, info_type((StgClosure*)bqe))); + return next; +} +#endif + +StgTSO * +unblockOne(Capability *cap, StgTSO *tso) +{ + StgTSO *next; + + ASSERT(get_itbl(tso)->type == TSO); + ASSERT(tso->why_blocked != NotBlocked); + + tso->why_blocked = NotBlocked; + next = tso->link; + tso->link = END_TSO_QUEUE; + +#if defined(THREADED_RTS) + if (tso->cap == cap || (!tsoLocked(tso) && RtsFlags.ParFlags.wakeupMigrate)) { + // We are waking up this thread on the current Capability, which + // might involve migrating it from the Capability it was last on. + if (tso->bound) { + ASSERT(tso->bound->cap == tso->cap); + tso->bound->cap = cap; + } + tso->cap = cap; + appendToRunQueue(cap,tso); + // we're holding a newly woken thread, make sure we context switch + // quickly so we can migrate it if necessary. + context_switch = 1; + } else { + // we'll try to wake it up on the Capability it was last on. + wakeupThreadOnCapability(tso->cap, tso); + } +#else + appendToRunQueue(cap,tso); + context_switch = 1; +#endif + + IF_DEBUG(scheduler,sched_belch("waking up thread %ld on cap %d", (long)tso->id, tso->cap->no)); + return next; +} + + +#if defined(GRAN) +void +awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node) +{ + StgBlockingQueueElement *bqe; + PEs node_loc; + nat len = 0; + + IF_GRAN_DEBUG(bq, + debugBelch("##-_ AwBQ for node %p on PE %d @ %ld by TSO %d (%p): \n", \ + node, CurrentProc, CurrentTime[CurrentProc], + CurrentTSO->id, CurrentTSO)); + + node_loc = where_is(node); + + ASSERT(q == END_BQ_QUEUE || + get_itbl(q)->type == TSO || // q is either a TSO or an RBHSave + get_itbl(q)->type == CONSTR); // closure (type constructor) + ASSERT(is_unique(node)); + + /* FAKE FETCH: magically copy the node to the tso's proc; + no Fetch necessary because in reality the node should not have been + moved to the other PE in the first place + */ + if (CurrentProc!=node_loc) { + IF_GRAN_DEBUG(bq, + debugBelch("## node %p is on PE %d but CurrentProc is %d (TSO %d); assuming fake fetch and adjusting bitmask (old: %#x)\n", + node, node_loc, CurrentProc, CurrentTSO->id, + // CurrentTSO, where_is(CurrentTSO), + node->header.gran.procs)); + node->header.gran.procs = (node->header.gran.procs) | PE_NUMBER(CurrentProc); + IF_GRAN_DEBUG(bq, + debugBelch("## new bitmask of node %p is %#x\n", + node, node->header.gran.procs)); + if (RtsFlags.GranFlags.GranSimStats.Global) { + globalGranStats.tot_fake_fetches++; + } + } + + bqe = q; + // ToDo: check: ASSERT(CurrentProc==node_loc); + while (get_itbl(bqe)->type==TSO) { // q != END_TSO_QUEUE) { + //next = bqe->link; + /* + bqe points to the current element in the queue + next points to the next element in the queue + */ + //tso = (StgTSO *)bqe; // wastes an assignment to get the type right + //tso_loc = where_is(tso); + len++; + bqe = unblockOne(bqe, node); + } + + /* if this is the BQ of an RBH, we have to put back the info ripped out of + the closure to make room for the anchor of the BQ */ + if (bqe!=END_BQ_QUEUE) { + ASSERT(get_itbl(node)->type == RBH && get_itbl(bqe)->type == CONSTR); + /* + ASSERT((info_ptr==&RBH_Save_0_info) || + (info_ptr==&RBH_Save_1_info) || + (info_ptr==&RBH_Save_2_info)); + */ + /* cf. convertToRBH in RBH.c for writing the RBHSave closure */ + ((StgRBH *)node)->blocking_queue = (StgBlockingQueueElement *)((StgRBHSave *)bqe)->payload[0]; + ((StgRBH *)node)->mut_link = (StgMutClosure *)((StgRBHSave *)bqe)->payload[1]; + + IF_GRAN_DEBUG(bq, + debugBelch("## Filled in RBH_Save for %p (%s) at end of AwBQ\n", + node, info_type(node))); + } + + /* statistics gathering */ + if (RtsFlags.GranFlags.GranSimStats.Global) { + // globalGranStats.tot_bq_processing_time += bq_processing_time; + globalGranStats.tot_bq_len += len; // total length of all bqs awakened + // globalGranStats.tot_bq_len_local += len_local; // same for local TSOs only + globalGranStats.tot_awbq++; // total no. of bqs awakened + } + IF_GRAN_DEBUG(bq, + debugBelch("## BQ Stats of %p: [%d entries] %s\n", + node, len, (bqe!=END_BQ_QUEUE) ? "RBH" : "")); +} +#elif defined(PARALLEL_HASKELL) +void +awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node) +{ + StgBlockingQueueElement *bqe; + + IF_PAR_DEBUG(verbose, + debugBelch("##-_ AwBQ for node %p on [%x]: \n", + node, mytid)); +#ifdef DIST + //RFP + if(get_itbl(q)->type == CONSTR || q==END_BQ_QUEUE) { + IF_PAR_DEBUG(verbose, debugBelch("## ... nothing to unblock so lets just return. RFP (BUG?)\n")); + return; + } +#endif + + ASSERT(q == END_BQ_QUEUE || + get_itbl(q)->type == TSO || + get_itbl(q)->type == BLOCKED_FETCH || + get_itbl(q)->type == CONSTR); + + bqe = q; + while (get_itbl(bqe)->type==TSO || + get_itbl(bqe)->type==BLOCKED_FETCH) { + bqe = unblockOne(bqe, node); + } +} + +#else /* !GRAN && !PARALLEL_HASKELL */ + +void +awakenBlockedQueue(Capability *cap, StgTSO *tso) +{ + if (tso == NULL) return; // hack; see bug #1235728, and comments in + // Exception.cmm + while (tso != END_TSO_QUEUE) { + tso = unblockOne(cap,tso); + } +} +#endif + +/* --------------------------------------------------------------------------- + Interrupt execution + - usually called inside a signal handler so it mustn't do anything fancy. + ------------------------------------------------------------------------ */ + +void +interruptStgRts(void) +{ + sched_state = SCHED_INTERRUPTING; + context_switch = 1; +#if defined(THREADED_RTS) + prodAllCapabilities(); +#endif +} + +/* ----------------------------------------------------------------------------- + Unblock a thread + + This is for use when we raise an exception in another thread, which + may be blocked. + This has nothing to do with the UnblockThread event in GranSim. -- HWL + -------------------------------------------------------------------------- */ + +#if defined(GRAN) || defined(PARALLEL_HASKELL) +/* + NB: only the type of the blocking queue is different in GranSim and GUM + the operations on the queue-elements are the same + long live polymorphism! + + Locks: sched_mutex is held upon entry and exit. + +*/ +static void +unblockThread(Capability *cap, StgTSO *tso) +{ + StgBlockingQueueElement *t, **last; + + switch (tso->why_blocked) { + + case NotBlocked: + return; /* not blocked */ + + case BlockedOnSTM: + // Be careful: nothing to do here! We tell the scheduler that the thread + // is runnable and we leave it to the stack-walking code to abort the + // transaction while unwinding the stack. We should perhaps have a debugging + // test to make sure that this really happens and that the 'zombie' transaction + // does not get committed. + goto done; + + case BlockedOnMVar: + ASSERT(get_itbl(tso->block_info.closure)->type == MVAR); + { + StgBlockingQueueElement *last_tso = END_BQ_QUEUE; + StgMVar *mvar = (StgMVar *)(tso->block_info.closure); + + last = (StgBlockingQueueElement **)&mvar->head; + for (t = (StgBlockingQueueElement *)mvar->head; + t != END_BQ_QUEUE; + last = &t->link, last_tso = t, t = t->link) { + if (t == (StgBlockingQueueElement *)tso) { + *last = (StgBlockingQueueElement *)tso->link; + if (mvar->tail == tso) { + mvar->tail = (StgTSO *)last_tso; + } + goto done; + } + } + barf("unblockThread (MVAR): TSO not found"); + } + + case BlockedOnBlackHole: + ASSERT(get_itbl(tso->block_info.closure)->type == BLACKHOLE_BQ); + { + StgBlockingQueue *bq = (StgBlockingQueue *)(tso->block_info.closure); + + last = &bq->blocking_queue; + for (t = bq->blocking_queue; + t != END_BQ_QUEUE; + last = &t->link, t = t->link) { + if (t == (StgBlockingQueueElement *)tso) { + *last = (StgBlockingQueueElement *)tso->link; + goto done; + } + } + barf("unblockThread (BLACKHOLE): TSO not found"); + } + + case BlockedOnException: + { + StgTSO *target = tso->block_info.tso; + + ASSERT(get_itbl(target)->type == TSO); + + if (target->what_next == ThreadRelocated) { + target = target->link; + ASSERT(get_itbl(target)->type == TSO); + } + + ASSERT(target->blocked_exceptions != NULL); + + last = (StgBlockingQueueElement **)&target->blocked_exceptions; + for (t = (StgBlockingQueueElement *)target->blocked_exceptions; + t != END_BQ_QUEUE; + last = &t->link, t = t->link) { + ASSERT(get_itbl(t)->type == TSO); + if (t == (StgBlockingQueueElement *)tso) { + *last = (StgBlockingQueueElement *)tso->link; + goto done; + } + } + barf("unblockThread (Exception): TSO not found"); + } + + case BlockedOnRead: + case BlockedOnWrite: +#if defined(mingw32_HOST_OS) + case BlockedOnDoProc: +#endif + { + /* take TSO off blocked_queue */ + StgBlockingQueueElement *prev = NULL; + for (t = (StgBlockingQueueElement *)blocked_queue_hd; t != END_BQ_QUEUE; + prev = t, t = t->link) { + if (t == (StgBlockingQueueElement *)tso) { + if (prev == NULL) { + blocked_queue_hd = (StgTSO *)t->link; + if ((StgBlockingQueueElement *)blocked_queue_tl == t) { + blocked_queue_tl = END_TSO_QUEUE; + } + } else { + prev->link = t->link; + if ((StgBlockingQueueElement *)blocked_queue_tl == t) { + blocked_queue_tl = (StgTSO *)prev; + } + } +#if defined(mingw32_HOST_OS) + /* (Cooperatively) signal that the worker thread should abort + * the request. + */ + abandonWorkRequest(tso->block_info.async_result->reqID); +#endif + goto done; + } + } + barf("unblockThread (I/O): TSO not found"); + } + + case BlockedOnDelay: + { + /* take TSO off sleeping_queue */ + StgBlockingQueueElement *prev = NULL; + for (t = (StgBlockingQueueElement *)sleeping_queue; t != END_BQ_QUEUE; + prev = t, t = t->link) { + if (t == (StgBlockingQueueElement *)tso) { + if (prev == NULL) { + sleeping_queue = (StgTSO *)t->link; + } else { + prev->link = t->link; + } + goto done; + } + } + barf("unblockThread (delay): TSO not found"); + } + + default: + barf("unblockThread"); + } + + done: + tso->link = END_TSO_QUEUE; + tso->why_blocked = NotBlocked; + tso->block_info.closure = NULL; + pushOnRunQueue(cap,tso); +} +#else +static void +unblockThread(Capability *cap, StgTSO *tso) +{ + StgTSO *t, **last; + + /* To avoid locking unnecessarily. */ + if (tso->why_blocked == NotBlocked) { + return; + } + + switch (tso->why_blocked) { + + case BlockedOnSTM: + // Be careful: nothing to do here! We tell the scheduler that the thread + // is runnable and we leave it to the stack-walking code to abort the + // transaction while unwinding the stack. We should perhaps have a debugging + // test to make sure that this really happens and that the 'zombie' transaction + // does not get committed. + goto done; + + case BlockedOnMVar: + ASSERT(get_itbl(tso->block_info.closure)->type == MVAR); + { + StgTSO *last_tso = END_TSO_QUEUE; + StgMVar *mvar = (StgMVar *)(tso->block_info.closure); + + last = &mvar->head; + for (t = mvar->head; t != END_TSO_QUEUE; + last = &t->link, last_tso = t, t = t->link) { + if (t == tso) { + *last = tso->link; + if (mvar->tail == tso) { + mvar->tail = last_tso; + } + goto done; + } + } + barf("unblockThread (MVAR): TSO not found"); + } + + case BlockedOnBlackHole: + { + last = &blackhole_queue; + for (t = blackhole_queue; t != END_TSO_QUEUE; + last = &t->link, t = t->link) { + if (t == tso) { + *last = tso->link; + goto done; + } + } + barf("unblockThread (BLACKHOLE): TSO not found"); + } + + case BlockedOnException: + { + StgTSO *target = tso->block_info.tso; + + ASSERT(get_itbl(target)->type == TSO); + + while (target->what_next == ThreadRelocated) { + target = target->link; + ASSERT(get_itbl(target)->type == TSO); + } + + ASSERT(target->blocked_exceptions != NULL); + + last = &target->blocked_exceptions; + for (t = target->blocked_exceptions; t != END_TSO_QUEUE; + last = &t->link, t = t->link) { + ASSERT(get_itbl(t)->type == TSO); + if (t == tso) { + *last = tso->link; + goto done; + } + } + barf("unblockThread (Exception): TSO not found"); + } + +#if !defined(THREADED_RTS) + case BlockedOnRead: + case BlockedOnWrite: +#if defined(mingw32_HOST_OS) + case BlockedOnDoProc: +#endif + { + StgTSO *prev = NULL; + for (t = blocked_queue_hd; t != END_TSO_QUEUE; + prev = t, t = t->link) { + if (t == tso) { + if (prev == NULL) { + blocked_queue_hd = t->link; + if (blocked_queue_tl == t) { + blocked_queue_tl = END_TSO_QUEUE; + } + } else { + prev->link = t->link; + if (blocked_queue_tl == t) { + blocked_queue_tl = prev; + } + } +#if defined(mingw32_HOST_OS) + /* (Cooperatively) signal that the worker thread should abort + * the request. + */ + abandonWorkRequest(tso->block_info.async_result->reqID); +#endif + goto done; + } + } + barf("unblockThread (I/O): TSO not found"); + } + + case BlockedOnDelay: + { + StgTSO *prev = NULL; + for (t = sleeping_queue; t != END_TSO_QUEUE; + prev = t, t = t->link) { + if (t == tso) { + if (prev == NULL) { + sleeping_queue = t->link; + } else { + prev->link = t->link; + } + goto done; + } + } + barf("unblockThread (delay): TSO not found"); + } +#endif + + default: + barf("unblockThread"); + } + + done: + tso->link = END_TSO_QUEUE; + tso->why_blocked = NotBlocked; + tso->block_info.closure = NULL; + appendToRunQueue(cap,tso); + + // We might have just migrated this TSO to our Capability: + if (tso->bound) { + tso->bound->cap = cap; + } + tso->cap = cap; +} +#endif + +/* ----------------------------------------------------------------------------- + * checkBlackHoles() + * + * Check the blackhole_queue for threads that can be woken up. We do + * this periodically: before every GC, and whenever the run queue is + * empty. + * + * An elegant solution might be to just wake up all the blocked + * threads with awakenBlockedQueue occasionally: they'll go back to + * sleep again if the object is still a BLACKHOLE. Unfortunately this + * doesn't give us a way to tell whether we've actually managed to + * wake up any threads, so we would be busy-waiting. + * + * -------------------------------------------------------------------------- */ + +static rtsBool +checkBlackHoles (Capability *cap) +{ + StgTSO **prev, *t; + rtsBool any_woke_up = rtsFalse; + StgHalfWord type; + + // blackhole_queue is global: + ASSERT_LOCK_HELD(&sched_mutex); + + IF_DEBUG(scheduler, sched_belch("checking threads blocked on black holes")); + + // ASSUMES: sched_mutex + prev = &blackhole_queue; + t = blackhole_queue; + while (t != END_TSO_QUEUE) { + ASSERT(t->why_blocked == BlockedOnBlackHole); + type = get_itbl(t->block_info.closure)->type; + if (type != BLACKHOLE && type != CAF_BLACKHOLE) { + IF_DEBUG(sanity,checkTSO(t)); + t = unblockOne(cap, t); + // urk, the threads migrate to the current capability + // here, but we'd like to keep them on the original one. + *prev = t; + any_woke_up = rtsTrue; + } else { + prev = &t->link; + t = t->link; + } + } + + return any_woke_up; +} + +/* ----------------------------------------------------------------------------- + * raiseAsync() + * + * The following function implements the magic for raising an + * asynchronous exception in an existing thread. + * + * We first remove the thread from any queue on which it might be + * blocked. The possible blockages are MVARs and BLACKHOLE_BQs. + * + * We strip the stack down to the innermost CATCH_FRAME, building + * thunks in the heap for all the active computations, so they can + * be restarted if necessary. When we reach a CATCH_FRAME, we build + * an application of the handler to the exception, and push it on + * the top of the stack. + * + * How exactly do we save all the active computations? We create an + * AP_STACK for every UpdateFrame on the stack. Entering one of these + * AP_STACKs pushes everything from the corresponding update frame + * upwards onto the stack. (Actually, it pushes everything up to the + * next update frame plus a pointer to the next AP_STACK object. + * Entering the next AP_STACK object pushes more onto the stack until we + * reach the last AP_STACK object - at which point the stack should look + * exactly as it did when we killed the TSO and we can continue + * execution by entering the closure on top of the stack. + * + * We can also kill a thread entirely - this happens if either (a) the + * exception passed to raiseAsync is NULL, or (b) there's no + * CATCH_FRAME on the stack. In either case, we strip the entire + * stack and replace the thread with a zombie. + * + * ToDo: in THREADED_RTS mode, this function is only safe if either + * (a) we hold all the Capabilities (eg. in GC, or if there is only + * one Capability), or (b) we own the Capability that the TSO is + * currently blocked on or on the run queue of. + * + * -------------------------------------------------------------------------- */ + +void +raiseAsync(Capability *cap, StgTSO *tso, StgClosure *exception) +{ + raiseAsync_(cap, tso, exception, rtsFalse, NULL); +} + +void +suspendComputation(Capability *cap, StgTSO *tso, StgPtr stop_here) +{ + raiseAsync_(cap, tso, NULL, rtsFalse, stop_here); +} + +static void +raiseAsync_(Capability *cap, StgTSO *tso, StgClosure *exception, + rtsBool stop_at_atomically, StgPtr stop_here) +{ + StgRetInfoTable *info; + StgPtr sp, frame; + nat i; + + // Thread already dead? + if (tso->what_next == ThreadComplete || tso->what_next == ThreadKilled) { + return; + } + + IF_DEBUG(scheduler, + sched_belch("raising exception in thread %ld.", (long)tso->id)); + + // Remove it from any blocking queues + unblockThread(cap,tso); + + // mark it dirty; we're about to change its stack. + dirtyTSO(tso); + + sp = tso->sp; + + // The stack freezing code assumes there's a closure pointer on + // the top of the stack, so we have to arrange that this is the case... + // + if (sp[0] == (W_)&stg_enter_info) { + sp++; + } else { + sp--; + sp[0] = (W_)&stg_dummy_ret_closure; + } + + frame = sp + 1; + while (stop_here == NULL || frame < stop_here) { + + // 1. Let the top of the stack be the "current closure" + // + // 2. Walk up the stack until we find either an UPDATE_FRAME or a + // CATCH_FRAME. + // + // 3. If it's an UPDATE_FRAME, then make an AP_STACK containing the + // current closure applied to the chunk of stack up to (but not + // including) the update frame. This closure becomes the "current + // closure". Go back to step 2. + // + // 4. If it's a CATCH_FRAME, then leave the exception handler on + // top of the stack applied to the exception. + // + // 5. If it's a STOP_FRAME, then kill the thread. + // + // NB: if we pass an ATOMICALLY_FRAME then abort the associated + // transaction + + info = get_ret_itbl((StgClosure *)frame); + + switch (info->i.type) { + + case UPDATE_FRAME: + { + StgAP_STACK * ap; + nat words; + + // First build an AP_STACK consisting of the stack chunk above the + // current update frame, with the top word on the stack as the + // fun field. + // + words = frame - sp - 1; + ap = (StgAP_STACK *)allocateLocal(cap,AP_STACK_sizeW(words)); + + ap->size = words; + ap->fun = (StgClosure *)sp[0]; + sp++; + for(i=0; i < (nat)words; ++i) { + ap->payload[i] = (StgClosure *)*sp++; + } + + SET_HDR(ap,&stg_AP_STACK_info, + ((StgClosure *)frame)->header.prof.ccs /* ToDo */); + TICK_ALLOC_UP_THK(words+1,0); + + IF_DEBUG(scheduler, + debugBelch("sched: Updating "); + printPtr((P_)((StgUpdateFrame *)frame)->updatee); + debugBelch(" with "); + printObj((StgClosure *)ap); + ); + + // Replace the updatee with an indirection + // + // Warning: if we're in a loop, more than one update frame on + // the stack may point to the same object. Be careful not to + // overwrite an IND_OLDGEN in this case, because we'll screw + // up the mutable lists. To be on the safe side, don't + // overwrite any kind of indirection at all. See also + // threadSqueezeStack in GC.c, where we have to make a similar + // check. + // + if (!closure_IND(((StgUpdateFrame *)frame)->updatee)) { + // revert the black hole + UPD_IND_NOLOCK(((StgUpdateFrame *)frame)->updatee, + (StgClosure *)ap); + } + sp += sizeofW(StgUpdateFrame) - 1; + sp[0] = (W_)ap; // push onto stack + frame = sp + 1; + continue; //no need to bump frame + } + + case STOP_FRAME: + // We've stripped the entire stack, the thread is now dead. + tso->what_next = ThreadKilled; + tso->sp = frame + sizeofW(StgStopFrame); + return; + + case CATCH_FRAME: + // If we find a CATCH_FRAME, and we've got an exception to raise, + // then build the THUNK raise(exception), and leave it on + // top of the CATCH_FRAME ready to enter. + // + { +#ifdef PROFILING + StgCatchFrame *cf = (StgCatchFrame *)frame; +#endif + StgThunk *raise; + + if (exception == NULL) break; + + // we've got an exception to raise, so let's pass it to the + // handler in this frame. + // + raise = (StgThunk *)allocateLocal(cap,sizeofW(StgThunk)+1); + TICK_ALLOC_SE_THK(1,0); + SET_HDR(raise,&stg_raise_info,cf->header.prof.ccs); + raise->payload[0] = exception; + + // throw away the stack from Sp up to the CATCH_FRAME. + // + sp = frame - 1; + + /* Ensure that async excpetions are blocked now, so we don't get + * a surprise exception before we get around to executing the + * handler. + */ + if (tso->blocked_exceptions == NULL) { + tso->blocked_exceptions = END_TSO_QUEUE; + } + + /* Put the newly-built THUNK on top of the stack, ready to execute + * when the thread restarts. + */ + sp[0] = (W_)raise; + sp[-1] = (W_)&stg_enter_info; + tso->sp = sp-1; + tso->what_next = ThreadRunGHC; + IF_DEBUG(sanity, checkTSO(tso)); + return; + } + + case ATOMICALLY_FRAME: + if (stop_at_atomically) { + ASSERT(stmGetEnclosingTRec(tso->trec) == NO_TREC); + stmCondemnTransaction(cap, tso -> trec); +#ifdef REG_R1 + tso->sp = frame; +#else + // R1 is not a register: the return convention for IO in + // this case puts the return value on the stack, so we + // need to set up the stack to return to the atomically + // frame properly... + tso->sp = frame - 2; + tso->sp[1] = (StgWord) &stg_NO_FINALIZER_closure; // why not? + tso->sp[0] = (StgWord) &stg_ut_1_0_unreg_info; +#endif + tso->what_next = ThreadRunGHC; + return; + } + // Not stop_at_atomically... fall through and abort the + // transaction. + + case CATCH_RETRY_FRAME: + // IF we find an ATOMICALLY_FRAME then we abort the + // current transaction and propagate the exception. In + // this case (unlike ordinary exceptions) we do not care + // whether the transaction is valid or not because its + // possible validity cannot have caused the exception + // and will not be visible after the abort. + IF_DEBUG(stm, + debugBelch("Found atomically block delivering async exception\n")); + StgTRecHeader *trec = tso -> trec; + StgTRecHeader *outer = stmGetEnclosingTRec(trec); + stmAbortTransaction(cap, trec); + tso -> trec = outer; + break; + + default: + break; + } + + // move on to the next stack frame + frame += stack_frame_sizeW((StgClosure *)frame); + } + + // if we got here, then we stopped at stop_here + ASSERT(stop_here != NULL); +} + +/* ----------------------------------------------------------------------------- + Deleting threads + + This is used for interruption (^C) and forking, and corresponds to + raising an exception but without letting the thread catch the + exception. + -------------------------------------------------------------------------- */ + +static void +deleteThread (Capability *cap, StgTSO *tso) +{ + if (tso->why_blocked != BlockedOnCCall && + tso->why_blocked != BlockedOnCCall_NoUnblockExc) { + raiseAsync(cap,tso,NULL); + } +} + +#ifdef FORKPROCESS_PRIMOP_SUPPORTED +static void +deleteThread_(Capability *cap, StgTSO *tso) +{ // for forkProcess only: + // like deleteThread(), but we delete threads in foreign calls, too. + + if (tso->why_blocked == BlockedOnCCall || + tso->why_blocked == BlockedOnCCall_NoUnblockExc) { + unblockOne(cap,tso); + tso->what_next = ThreadKilled; + } else { + deleteThread(cap,tso); + } +} +#endif + +/* ----------------------------------------------------------------------------- + raiseExceptionHelper + + This function is called by the raise# primitve, just so that we can + move some of the tricky bits of raising an exception from C-- into + C. Who knows, it might be a useful re-useable thing here too. + -------------------------------------------------------------------------- */ + +StgWord +raiseExceptionHelper (StgRegTable *reg, StgTSO *tso, StgClosure *exception) +{ + Capability *cap = regTableToCapability(reg); + StgThunk *raise_closure = NULL; + StgPtr p, next; + StgRetInfoTable *info; + // + // This closure represents the expression 'raise# E' where E + // is the exception raise. It is used to overwrite all the + // thunks which are currently under evaluataion. + // + + // OLD COMMENT (we don't have MIN_UPD_SIZE now): + // LDV profiling: stg_raise_info has THUNK as its closure + // type. Since a THUNK takes at least MIN_UPD_SIZE words in its + // payload, MIN_UPD_SIZE is more approprate than 1. It seems that + // 1 does not cause any problem unless profiling is performed. + // However, when LDV profiling goes on, we need to linearly scan + // small object pool, where raise_closure is stored, so we should + // use MIN_UPD_SIZE. + // + // raise_closure = (StgClosure *)RET_STGCALL1(P_,allocate, + // sizeofW(StgClosure)+1); + // + + // + // Walk up the stack, looking for the catch frame. On the way, + // we update any closures pointed to from update frames with the + // raise closure that we just built. + // + p = tso->sp; + while(1) { + info = get_ret_itbl((StgClosure *)p); + next = p + stack_frame_sizeW((StgClosure *)p); + switch (info->i.type) { + + case UPDATE_FRAME: + // Only create raise_closure if we need to. + if (raise_closure == NULL) { + raise_closure = + (StgThunk *)allocateLocal(cap,sizeofW(StgThunk)+1); + SET_HDR(raise_closure, &stg_raise_info, CCCS); + raise_closure->payload[0] = exception; + } + UPD_IND(((StgUpdateFrame *)p)->updatee,(StgClosure *)raise_closure); + p = next; + continue; + + case ATOMICALLY_FRAME: + IF_DEBUG(stm, debugBelch("Found ATOMICALLY_FRAME at %p\n", p)); + tso->sp = p; + return ATOMICALLY_FRAME; + + case CATCH_FRAME: + tso->sp = p; + return CATCH_FRAME; + + case CATCH_STM_FRAME: + IF_DEBUG(stm, debugBelch("Found CATCH_STM_FRAME at %p\n", p)); + tso->sp = p; + return CATCH_STM_FRAME; + + case STOP_FRAME: + tso->sp = p; + return STOP_FRAME; + + case CATCH_RETRY_FRAME: + default: + p = next; + continue; + } + } +} + + +/* ----------------------------------------------------------------------------- + findRetryFrameHelper + + This function is called by the retry# primitive. It traverses the stack + leaving tso->sp referring to the frame which should handle the retry. + + This should either be a CATCH_RETRY_FRAME (if the retry# is within an orElse#) + or should be a ATOMICALLY_FRAME (if the retry# reaches the top level). + + We skip CATCH_STM_FRAMEs because retries are not considered to be exceptions, + despite the similar implementation. + + We should not expect to see CATCH_FRAME or STOP_FRAME because those should + not be created within memory transactions. + -------------------------------------------------------------------------- */ + +StgWord +findRetryFrameHelper (StgTSO *tso) +{ + StgPtr p, next; + StgRetInfoTable *info; + + p = tso -> sp; + while (1) { + info = get_ret_itbl((StgClosure *)p); + next = p + stack_frame_sizeW((StgClosure *)p); + switch (info->i.type) { + + case ATOMICALLY_FRAME: + IF_DEBUG(stm, debugBelch("Found ATOMICALLY_FRAME at %p during retrry\n", p)); + tso->sp = p; + return ATOMICALLY_FRAME; + + case CATCH_RETRY_FRAME: + IF_DEBUG(stm, debugBelch("Found CATCH_RETRY_FRAME at %p during retrry\n", p)); + tso->sp = p; + return CATCH_RETRY_FRAME; + + case CATCH_STM_FRAME: + default: + ASSERT(info->i.type != CATCH_FRAME); + ASSERT(info->i.type != STOP_FRAME); + p = next; + continue; + } + } +} + +/* ----------------------------------------------------------------------------- + resurrectThreads is called after garbage collection on the list of + threads found to be garbage. Each of these threads will be woken + up and sent a signal: BlockedOnDeadMVar if the thread was blocked + on an MVar, or NonTermination if the thread was blocked on a Black + Hole. + + Locks: assumes we hold *all* the capabilities. + -------------------------------------------------------------------------- */ + +void +resurrectThreads (StgTSO *threads) +{ + StgTSO *tso, *next; + Capability *cap; + + for (tso = threads; tso != END_TSO_QUEUE; tso = next) { + next = tso->global_link; + tso->global_link = all_threads; + all_threads = tso; + IF_DEBUG(scheduler, sched_belch("resurrecting thread %d", tso->id)); + + // Wake up the thread on the Capability it was last on + cap = tso->cap; + + switch (tso->why_blocked) { + case BlockedOnMVar: + case BlockedOnException: + /* Called by GC - sched_mutex lock is currently held. */ + raiseAsync(cap, tso,(StgClosure *)BlockedOnDeadMVar_closure); + break; + case BlockedOnBlackHole: + raiseAsync(cap, tso,(StgClosure *)NonTermination_closure); + break; + case BlockedOnSTM: + raiseAsync(cap, tso,(StgClosure *)BlockedIndefinitely_closure); + break; + case NotBlocked: + /* This might happen if the thread was blocked on a black hole + * belonging to a thread that we've just woken up (raiseAsync + * can wake up threads, remember...). + */ + continue; + default: + barf("resurrectThreads: thread blocked in a strange way"); + } + } +} + +/* ---------------------------------------------------------------------------- + * Debugging: why is a thread blocked + * [Also provides useful information when debugging threaded programs + * at the Haskell source code level, so enable outside of DEBUG. --sof 7/02] + ------------------------------------------------------------------------- */ + +#if DEBUG +static void +printThreadBlockage(StgTSO *tso) +{ + switch (tso->why_blocked) { + case BlockedOnRead: + debugBelch("is blocked on read from fd %d", (int)(tso->block_info.fd)); + break; + case BlockedOnWrite: + debugBelch("is blocked on write to fd %d", (int)(tso->block_info.fd)); + break; +#if defined(mingw32_HOST_OS) + case BlockedOnDoProc: + debugBelch("is blocked on proc (request: %ld)", tso->block_info.async_result->reqID); + break; +#endif + case BlockedOnDelay: + debugBelch("is blocked until %ld", (long)(tso->block_info.target)); + break; + case BlockedOnMVar: + debugBelch("is blocked on an MVar @ %p", tso->block_info.closure); + break; + case BlockedOnException: + debugBelch("is blocked on delivering an exception to thread %d", + tso->block_info.tso->id); + break; + case BlockedOnBlackHole: + debugBelch("is blocked on a black hole"); + break; + case NotBlocked: + debugBelch("is not blocked"); + break; +#if defined(PARALLEL_HASKELL) + case BlockedOnGA: + debugBelch("is blocked on global address; local FM_BQ is %p (%s)", + tso->block_info.closure, info_type(tso->block_info.closure)); + break; + case BlockedOnGA_NoSend: + debugBelch("is blocked on global address (no send); local FM_BQ is %p (%s)", + tso->block_info.closure, info_type(tso->block_info.closure)); + break; +#endif + case BlockedOnCCall: + debugBelch("is blocked on an external call"); + break; + case BlockedOnCCall_NoUnblockExc: + debugBelch("is blocked on an external call (exceptions were already blocked)"); + break; + case BlockedOnSTM: + debugBelch("is blocked on an STM operation"); + break; + default: + barf("printThreadBlockage: strange tso->why_blocked: %d for TSO %d (%d)", + tso->why_blocked, tso->id, tso); + } +} + +void +printThreadStatus(StgTSO *t) +{ + debugBelch("\tthread %4d @ %p ", t->id, (void *)t); + { + void *label = lookupThreadLabel(t->id); + if (label) debugBelch("[\"%s\"] ",(char *)label); + } + if (t->what_next == ThreadRelocated) { + debugBelch("has been relocated...\n"); + } else { + switch (t->what_next) { + case ThreadKilled: + debugBelch("has been killed"); + break; + case ThreadComplete: + debugBelch("has completed"); + break; + default: + printThreadBlockage(t); + } + debugBelch("\n"); + } +} + +void +printAllThreads(void) +{ + StgTSO *t, *next; + nat i; + Capability *cap; + +# if defined(GRAN) + char time_string[TIME_STR_LEN], node_str[NODE_STR_LEN]; + ullong_format_string(TIME_ON_PROC(CurrentProc), + time_string, rtsFalse/*no commas!*/); + + debugBelch("all threads at [%s]:\n", time_string); +# elif defined(PARALLEL_HASKELL) + char time_string[TIME_STR_LEN], node_str[NODE_STR_LEN]; + ullong_format_string(CURRENT_TIME, + time_string, rtsFalse/*no commas!*/); + + debugBelch("all threads at [%s]:\n", time_string); +# else + debugBelch("all threads:\n"); +# endif + + for (i = 0; i < n_capabilities; i++) { + cap = &capabilities[i]; + debugBelch("threads on capability %d:\n", cap->no); + for (t = cap->run_queue_hd; t != END_TSO_QUEUE; t = t->link) { + printThreadStatus(t); + } + } + + debugBelch("other threads:\n"); + for (t = all_threads; t != END_TSO_QUEUE; t = next) { + if (t->why_blocked != NotBlocked) { + printThreadStatus(t); + } + if (t->what_next == ThreadRelocated) { + next = t->link; + } else { + next = t->global_link; + } + } +} + +// useful from gdb +void +printThreadQueue(StgTSO *t) +{ + nat i = 0; + for (; t != END_TSO_QUEUE; t = t->link) { + printThreadStatus(t); + i++; + } + debugBelch("%d threads on queue\n", i); +} + +/* + Print a whole blocking queue attached to node (debugging only). +*/ +# if defined(PARALLEL_HASKELL) +void +print_bq (StgClosure *node) +{ + StgBlockingQueueElement *bqe; + StgTSO *tso; + rtsBool end; + + debugBelch("## BQ of closure %p (%s): ", + node, info_type(node)); + + /* should cover all closures that may have a blocking queue */ + ASSERT(get_itbl(node)->type == BLACKHOLE_BQ || + get_itbl(node)->type == FETCH_ME_BQ || + get_itbl(node)->type == RBH || + get_itbl(node)->type == MVAR); + + ASSERT(node!=(StgClosure*)NULL); // sanity check + + print_bqe(((StgBlockingQueue*)node)->blocking_queue); +} + +/* + Print a whole blocking queue starting with the element bqe. +*/ +void +print_bqe (StgBlockingQueueElement *bqe) +{ + rtsBool end; + + /* + NB: In a parallel setup a BQ of an RBH must end with an RBH_Save closure; + */ + for (end = (bqe==END_BQ_QUEUE); + !end; // iterate until bqe points to a CONSTR + end = (get_itbl(bqe)->type == CONSTR) || (bqe->link==END_BQ_QUEUE), + bqe = end ? END_BQ_QUEUE : bqe->link) { + ASSERT(bqe != END_BQ_QUEUE); // sanity check + ASSERT(bqe != (StgBlockingQueueElement *)NULL); // sanity check + /* types of closures that may appear in a blocking queue */ + ASSERT(get_itbl(bqe)->type == TSO || + get_itbl(bqe)->type == BLOCKED_FETCH || + get_itbl(bqe)->type == CONSTR); + /* only BQs of an RBH end with an RBH_Save closure */ + //ASSERT(get_itbl(bqe)->type != CONSTR || get_itbl(node)->type == RBH); + + switch (get_itbl(bqe)->type) { + case TSO: + debugBelch(" TSO %u (%x),", + ((StgTSO *)bqe)->id, ((StgTSO *)bqe)); + break; + case BLOCKED_FETCH: + debugBelch(" BF (node=%p, ga=((%x, %d, %x)),", + ((StgBlockedFetch *)bqe)->node, + ((StgBlockedFetch *)bqe)->ga.payload.gc.gtid, + ((StgBlockedFetch *)bqe)->ga.payload.gc.slot, + ((StgBlockedFetch *)bqe)->ga.weight); + break; + case CONSTR: + debugBelch(" %s (IP %p),", + (get_itbl(bqe) == &stg_RBH_Save_0_info ? "RBH_Save_0" : + get_itbl(bqe) == &stg_RBH_Save_1_info ? "RBH_Save_1" : + get_itbl(bqe) == &stg_RBH_Save_2_info ? "RBH_Save_2" : + "RBH_Save_?"), get_itbl(bqe)); + break; + default: + barf("Unexpected closure type %s in blocking queue", // of %p (%s)", + info_type((StgClosure *)bqe)); // , node, info_type(node)); + break; + } + } /* for */ + debugBelch("\n"); +} +# elif defined(GRAN) +void +print_bq (StgClosure *node) +{ + StgBlockingQueueElement *bqe; + PEs node_loc, tso_loc; + rtsBool end; + + /* should cover all closures that may have a blocking queue */ + ASSERT(get_itbl(node)->type == BLACKHOLE_BQ || + get_itbl(node)->type == FETCH_ME_BQ || + get_itbl(node)->type == RBH); + + ASSERT(node!=(StgClosure*)NULL); // sanity check + node_loc = where_is(node); + + debugBelch("## BQ of closure %p (%s) on [PE %d]: ", + node, info_type(node), node_loc); + + /* + NB: In a parallel setup a BQ of an RBH must end with an RBH_Save closure; + */ + for (bqe = ((StgBlockingQueue*)node)->blocking_queue, end = (bqe==END_BQ_QUEUE); + !end; // iterate until bqe points to a CONSTR + end = (get_itbl(bqe)->type == CONSTR) || (bqe->link==END_BQ_QUEUE), bqe = end ? END_BQ_QUEUE : bqe->link) { + ASSERT(bqe != END_BQ_QUEUE); // sanity check + ASSERT(bqe != (StgBlockingQueueElement *)NULL); // sanity check + /* types of closures that may appear in a blocking queue */ + ASSERT(get_itbl(bqe)->type == TSO || + get_itbl(bqe)->type == CONSTR); + /* only BQs of an RBH end with an RBH_Save closure */ + ASSERT(get_itbl(bqe)->type != CONSTR || get_itbl(node)->type == RBH); + + tso_loc = where_is((StgClosure *)bqe); + switch (get_itbl(bqe)->type) { + case TSO: + debugBelch(" TSO %d (%p) on [PE %d],", + ((StgTSO *)bqe)->id, (StgTSO *)bqe, tso_loc); + break; + case CONSTR: + debugBelch(" %s (IP %p),", + (get_itbl(bqe) == &stg_RBH_Save_0_info ? "RBH_Save_0" : + get_itbl(bqe) == &stg_RBH_Save_1_info ? "RBH_Save_1" : + get_itbl(bqe) == &stg_RBH_Save_2_info ? "RBH_Save_2" : + "RBH_Save_?"), get_itbl(bqe)); + break; + default: + barf("Unexpected closure type %s in blocking queue of %p (%s)", + info_type((StgClosure *)bqe), node, info_type(node)); + break; + } + } /* for */ + debugBelch("\n"); +} +# endif + +#if defined(PARALLEL_HASKELL) +static nat +run_queue_len(void) +{ + nat i; + StgTSO *tso; + + for (i=0, tso=run_queue_hd; + tso != END_TSO_QUEUE; + i++, tso=tso->link) { + /* nothing */ + } + + return i; +} +#endif + +void +sched_belch(char *s, ...) +{ + va_list ap; + va_start(ap,s); +#ifdef THREADED_RTS + debugBelch("sched (task %p): ", (void *)(unsigned long)(unsigned int)osThreadId()); +#elif defined(PARALLEL_HASKELL) + debugBelch("== "); +#else + debugBelch("sched: "); +#endif + vdebugBelch(s, ap); + debugBelch("\n"); + va_end(ap); +} + +#endif /* DEBUG */ |