X-Git-Url: http://git.megacz.com/?a=blobdiff_plain;f=ghc%2Frts%2FSchedule.c;h=bbc6a8b5a8c632960cbdf6e3e143c467cfd78180;hb=c137ecd7e6e83d0f9c39b15ccdb9f2355f243c91;hp=dcd32dad766457390730bf1cd820459d1ccc75be;hpb=6b7f3c93728b29302748b5cde56101588bdcb9bf;p=ghc-hetmet.git diff --git a/ghc/rts/Schedule.c b/ghc/rts/Schedule.c index dcd32da..bbc6a8b 100644 --- a/ghc/rts/Schedule.c +++ b/ghc/rts/Schedule.c @@ -1,41 +1,11 @@ /* --------------------------------------------------------------------------- * - * (c) The GHC Team, 1998-2004 + * (c) The GHC Team, 1998-2005 * - * Scheduler - * - * Different GHC ways use this scheduler quite differently (see comments below) - * Here is the global picture: - * - * WAY Name CPP flag What's it for - * -------------------------------------- - * mp GUM PARALLEL_HASKELL Parallel execution on a distrib. memory machine - * s SMP SMP Parallel execution on a shared memory machine - * mg GranSim GRAN Simulation of parallel execution - * md GUM/GdH DIST Distributed execution (based on GUM) + * The scheduler and thread-related functionality * * --------------------------------------------------------------------------*/ -/* - * Version with support for distributed memory parallelism aka GUM (WAY=mp): - - The main scheduling loop in GUM iterates until a finish message is received. - In that case a global flag @receivedFinish@ is set and this instance of - the RTS shuts down. See ghc/rts/parallel/HLComms.c:processMessages() - for the handling of incoming messages, such as PP_FINISH. - Note that in the parallel case we have a system manager that coordinates - different PEs, each of which are running one instance of the RTS. - See ghc/rts/parallel/SysMan.c for the main routine of the parallel program. - From this routine processes executing ghc/rts/Main.c are spawned. -- HWL - - * Version with support for simulating parallel execution aka GranSim (WAY=mg): - - The main scheduling code in GranSim is quite different from that in std - (concurrent) Haskell: while concurrent Haskell just iterates over the - threads in the runnable queue, GranSim is event driven, i.e. it iterates - over the events in the global event queue. -- HWL -*/ - #include "PosixSource.h" #include "Rts.h" #include "SchedAPI.h" @@ -46,13 +16,12 @@ #include "Storage.h" #include "StgRun.h" #include "Hooks.h" -#define COMPILING_SCHEDULER #include "Schedule.h" #include "StgMiscClosures.h" #include "Interpreter.h" #include "Exception.h" #include "Printer.h" -#include "Signals.h" +#include "RtsSignals.h" #include "Sanity.h" #include "Stats.h" #include "STM.h" @@ -76,7 +45,11 @@ #endif #include "Sparks.h" #include "Capability.h" -#include "Task.h" +#include "Task.h" +#include "AwaitEvent.h" +#if defined(mingw32_HOST_OS) +#include "win32/IOManager.h" +#endif #ifdef HAVE_SYS_TYPES_H #include @@ -99,22 +72,9 @@ # define STATIC_INLINE static #endif -#ifdef THREADED_RTS -#define USED_IN_THREADED_RTS -#else -#define USED_IN_THREADED_RTS STG_UNUSED -#endif - -#ifdef RTS_SUPPORTS_THREADS -#define USED_WHEN_RTS_SUPPORTS_THREADS -#else -#define USED_WHEN_RTS_SUPPORTS_THREADS STG_UNUSED -#endif - -/* Main thread queue. - * Locks required: sched_mutex. - */ -StgMainThread *main_threads = NULL; +/* ----------------------------------------------------------------------------- + * Global variables + * -------------------------------------------------------------------------- */ #if defined(GRAN) @@ -138,59 +98,52 @@ StgTSO *ccalling_threadss[MAX_PROC]; #else /* !GRAN */ -/* Thread queues. - * Locks required: sched_mutex. - */ -StgTSO *run_queue_hd = NULL; -StgTSO *run_queue_tl = NULL; +#if !defined(THREADED_RTS) +// Blocked/sleeping thrads StgTSO *blocked_queue_hd = NULL; StgTSO *blocked_queue_tl = NULL; -StgTSO *blackhole_queue = NULL; -StgTSO *sleeping_queue = NULL; /* perhaps replace with a hash table? */ +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; -/* When a thread performs a safe C call (_ccall_GC, using old - * terminology), it gets put on the suspended_ccalling_threads - * list. Used by the garbage collector. +/* flag set by signal handler to precipitate a context switch + * LOCK: none (just an advisory flag) */ -static StgTSO *suspended_ccalling_threads; - -/* KH: The following two flags are shared memory locations. There is no need - to lock them, since they are only unset at the end of a scheduler - operation. -*/ - -/* flag set by signal handler to precipitate a context switch */ int context_switch = 0; -/* flag that tracks whether we have done any execution in this time slice. */ +/* 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 */ -rtsBool interrupted = rtsFalse; +/* 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. - * Locks required: thread_id_mutex + * LOCK: sched_mutex */ static StgThreadID next_thread_id = 1; -/* - * Pointers to the state of the current thread. - * Rule of thumb: if CurrentTSO != NULL, then we're running a Haskell - * thread. If CurrentTSO == NULL, then we're at the scheduler level. - */ - /* 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) @@ -201,10 +154,8 @@ static StgThreadID next_thread_id = 1; * 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 @@ -220,16 +171,15 @@ StgTSO dummy_tso; * in an MT setting, needed to signal that a worker thread shouldn't hang around * in the scheduler when it is out of work. */ -static rtsBool shutting_down_scheduler = rtsFalse; +rtsBool shutting_down_scheduler = rtsFalse; -#if defined(RTS_SUPPORTS_THREADS) -/* ToDo: carefully document the invariants that go together - * with these synchronisation objects. +/* + * This mutex protects most of the global scheduler data in + * the THREADED_RTS runtime. */ -Mutex sched_mutex = INIT_MUTEX_VAR; -Mutex term_mutex = INIT_MUTEX_VAR; - -#endif /* RTS_SUPPORTS_THREADS */ +#if defined(THREADED_RTS) +Mutex sched_mutex; +#endif #if defined(PARALLEL_HASKELL) StgTSO *LastTSO; @@ -237,38 +187,25 @@ rtsTime TimeOfLastYield; rtsBool emitSchedule = rtsTrue; #endif -#if DEBUG -static char *whatNext_strs[] = { - "(unknown)", - "ThreadRunGHC", - "ThreadInterpret", - "ThreadKilled", - "ThreadRelocated", - "ThreadComplete" -}; -#endif - /* ----------------------------------------------------------------------------- * static function prototypes * -------------------------------------------------------------------------- */ -#if defined(RTS_SUPPORTS_THREADS) -static void taskStart(void); -#endif - -static void schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, - Capability *initialCapability ); +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); -static void scheduleStartSignalHandlers(void); -static void scheduleCheckBlockedThreads(void); -static void scheduleCheckBlackHoles(void); -static void scheduleDetectDeadlock(void); +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 scheduleCheckBlackHoles (Capability *cap); +static void scheduleDetectDeadlock (Capability *cap, Task *task); #if defined(GRAN) static StgTSO *scheduleProcessEvent(rtsEvent *event); #endif @@ -282,69 +219,50 @@ static void scheduleGranParReport(void); #endif static void schedulePostRunThread(void); static rtsBool scheduleHandleHeapOverflow( Capability *cap, StgTSO *t ); -static void scheduleHandleStackOverflow( StgTSO *t); -static rtsBool scheduleHandleYield( StgTSO *t, nat prev_what_next ); +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( StgMainThread *mainThread, - Capability *cap, StgTSO *t ); +static rtsBool scheduleHandleThreadFinished( Capability *cap, Task *task, + StgTSO *t ); static rtsBool scheduleDoHeapProfile(rtsBool ready_to_gc); -static void scheduleDoGC(rtsBool force_major); - -static void unblockThread(StgTSO *tso); -static rtsBool checkBlackHoles(void); -static SchedulerStatus waitThread_(/*out*/StgMainThread* m, - Capability *initialCapability - ); -static void scheduleThread_ (StgTSO* tso); +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(StgTSO *tso); +static StgTSO *threadStackOverflow(Capability *cap, StgTSO *tso); -static void raiseAsync_(StgTSO *tso, StgClosure *exception, - rtsBool stop_at_atomically); +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 -/* ---------------------------------------------------------------------------- - * Starting Tasks - * ------------------------------------------------------------------------- */ - -#if defined(RTS_SUPPORTS_THREADS) -static nat startingWorkerThread = 0; - -static void -taskStart(void) -{ - ACQUIRE_LOCK(&sched_mutex); - startingWorkerThread--; - schedule(NULL,NULL); - taskStop(); - RELEASE_LOCK(&sched_mutex); -} - -void -startSchedulerTaskIfNecessary(void) -{ - if ( !EMPTY_RUN_QUEUE() - && !shutting_down_scheduler // not if we're shutting down - && startingWorkerThread==0) - { - // we don't want to start another worker thread - // just because the last one hasn't yet reached the - // "waiting for capability" state - startingWorkerThread++; - if (!maybeStartNewWorker(taskStart)) { - startingWorkerThread--; - } - } -} +#ifdef DEBUG +static char *whatNext_strs[] = { + "(unknown)", + "ThreadRunGHC", + "ThreadInterpret", + "ThreadKilled", + "ThreadRelocated", + "ThreadComplete" +}; #endif /* ----------------------------------------------------------------------------- @@ -352,22 +270,22 @@ startSchedulerTaskIfNecessary(void) * -------------------------------------------------------------------------- */ STATIC_INLINE void -addToRunQueue( StgTSO *t ) +addToRunQueue( Capability *cap, StgTSO *t ) { #if defined(PARALLEL_HASKELL) if (RtsFlags.ParFlags.doFairScheduling) { // this does round-robin scheduling; good for concurrency - APPEND_TO_RUN_QUEUE(t); + appendToRunQueue(cap,t); } else { // this does unfair scheduling; good for parallelism - PUSH_ON_RUN_QUEUE(t); + pushOnRunQueue(cap,t); } #else // this does round-robin scheduling; good for concurrency - APPEND_TO_RUN_QUEUE(t); + appendToRunQueue(cap,t); #endif } - + /* --------------------------------------------------------------------------- Main scheduling loop. @@ -380,13 +298,6 @@ addToRunQueue( StgTSO *t ) * thread ends * stack overflow - Locking notes: we acquire the scheduler lock once at the beginning - of the scheduler loop, and release it when - - * running a thread, or - * waiting for work, or - * waiting for a GC to complete. - GRAN version: In a GranSim setup this loop iterates over the global event queue. This revolves around the global event queue, which determines what @@ -404,9 +315,8 @@ addToRunQueue( StgTSO *t ) ------------------------------------------------------------------------ */ -static void -schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, - Capability *initialCapability ) +static Capability * +schedule (Capability *initialCapability, Task *task) { StgTSO *t; Capability *cap; @@ -423,19 +333,19 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, #endif nat prev_what_next; rtsBool ready_to_gc; +#if defined(THREADED_RTS) + rtsBool first = rtsTrue; +#endif - // Pre-condition: sched_mutex is held. - // We might have a capability, passed in as initialCapability. cap = initialCapability; -#if !defined(RTS_SUPPORTS_THREADS) - // simply initialise it in the non-threaded case - grabCapability(&cap); -#endif + // 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 (main thr: %p, cap: %p)", - mainThread, initialCapability); + sched_belch("### NEW SCHEDULER LOOP (task: %p, cap: %p)", + task, initialCapability); ); schedulePreLoop(); @@ -459,96 +369,125 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, CurrentTSO = event->tso; #endif -#if defined(RTS_SUPPORTS_THREADS) - // Yield the capability to higher-priority tasks if necessary. - // - if (cap != NULL) { - yieldCapability(&cap, - mainThread ? &mainThread->bound_thread_cond : NULL ); - } - - // If we do not currently hold a capability, we wait for one - // - if (cap == NULL) { - waitForCapability(&sched_mutex, &cap, - mainThread ? &mainThread->bound_thread_cond : NULL); +#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); } - - // We now have a capability... #endif - -#if 0 /* extra sanity checking */ - { - StgMainThread *m; - for (m = main_threads; m != NULL; m = m->link) { - ASSERT(get_itbl(m->tso)->type == TSO); - } - } + +#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->r.rInHaskell) { + if (cap->in_haskell) { errorBelch("schedule: re-entered unsafely.\n" " Perhaps a 'foreign import unsafe' should be 'safe'?"); - stg_exit(1); + 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. // - // Test for interruption. If interrupted==rtsTrue, then either - // we received a keyboard interrupt (^C), or the scheduler is - // trying to shut down all the tasks (shutting_down_scheduler) in - // the threaded RTS. + // So the sequence for ^C goes like this: // - if (interrupted) { - if (shutting_down_scheduler) { - IF_DEBUG(scheduler, sched_belch("shutting down")); - releaseCapability(cap); - if (mainThread) { - mainThread->stat = Interrupted; - mainThread->ret = NULL; - } - return; - } else { - IF_DEBUG(scheduler, sched_belch("interrupted")); - deleteAllThreads(); + // * ^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(not_yet) && defined(SMP) - // - // Top up the run queue from our spark pool. We try to make the - // number of threads in the run queue equal to the number of - // free capabilities. - // +#if defined(THREADED_RTS) + // If the run queue is empty, take a spark and turn it into a thread. { - StgClosure *spark; - if (EMPTY_RUN_QUEUE()) { - spark = findSpark(rtsFalse); - if (spark == NULL) { - break; /* no more sparks in the pool */ - } else { - createSparkThread(spark); + if (emptyRunQueue(cap)) { + StgClosure *spark; + spark = findSpark(cap); + if (spark != NULL) { IF_DEBUG(scheduler, - sched_belch("==^^ turning spark of closure %p into a thread", + sched_belch("turning spark of closure %p into a thread", (StgClosure *)spark)); + createSparkThread(cap,spark); } } } -#endif // SMP +#endif // THREADED_RTS - scheduleStartSignalHandlers(); + 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 (EMPTY_RUN_QUEUE()) { scheduleCheckBlackHoles(); } + if (emptyRunQueue(cap)) { scheduleCheckBlackHoles(cap); } - scheduleCheckBlockedThreads(); + scheduleCheckBlockedThreads(cap); - scheduleDetectDeadlock(); + 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 @@ -558,16 +497,16 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, // // win32: might be here due to awaitEvent() being abandoned // as a result of a console event having been delivered. - if ( EMPTY_RUN_QUEUE() ) { -#if !defined(RTS_SUPPORTS_THREADS) && !defined(mingw32_HOST_OS) - ASSERT(interrupted); + 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 (EMPTY_RUN_QUEUE() && scheduleActivateSpark()) + if (emptyRunQueue(cap) && scheduleActivateSpark()) continue; #if defined(SPARKS) @@ -576,7 +515,7 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, /* If we still have no work we need to send a FISH to get a spark from another PE */ - if (EMPTY_RUN_QUEUE()) { + if (emptyRunQueue(cap)) { if (!scheduleGetRemoteWork(&receivedFinish)) continue; ASSERT(rtsFalse); // should not happen at the moment } @@ -597,8 +536,7 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, // // Get a thread to run // - ASSERT(run_queue_hd != END_TSO_QUEUE); - POP_RUN_QUEUE(t); + t = popRunQueue(cap); #if defined(GRAN) || defined(PAR) scheduleGranParReport(); // some kind of debuging output @@ -608,61 +546,53 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, IF_DEBUG(sanity,checkTSO(t)); #endif -#if defined(RTS_SUPPORTS_THREADS) +#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. { - StgMainThread *m = t->main; + Task *bound = t->bound; - if(m) - { - if(m == mainThread) - { - 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 - PUSH_ON_RUN_QUEUE(t); - continue; - } - } - else - { - if(mainThread != NULL) - // The thread we want to run is unbound. - { - 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 - PUSH_ON_RUN_QUEUE(t); - continue; + 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 now initiated by the timer signal, unless + /* 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 - && (run_queue_hd != END_TSO_QUEUE - || blocked_queue_hd != END_TSO_QUEUE - || sleeping_queue != END_TSO_QUEUE))) + if (RtsFlags.ConcFlags.ctxtSwitchTicks == 0 + && !emptyThreadQueues(cap)) { context_switch = 1; - + } + run_thread: - RELEASE_LOCK(&sched_mutex); - IF_DEBUG(scheduler, sched_belch("-->> running thread %ld %s ...", (long)t->id, whatNext_strs[t->what_next])); @@ -673,67 +603,88 @@ run_thread: // ---------------------------------------------------------------------- // Run the current thread + ASSERT_FULL_CAPABILITY_INVARIANTS(cap,task); + prev_what_next = t->what_next; errno = t->saved_errno; - cap->r.rInHaskell = rtsTrue; + 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: - ret = StgRun((StgFunPtr) stg_returnToStackTop, &cap->r); + { + StgRegTable *r; + r = StgRun((StgFunPtr) stg_returnToStackTop, &cap->r); + cap = regTableToCapability(r); + ret = r->rRet; break; - + } + case ThreadInterpret: - ret = interpretBCO(cap); + cap = interpretBCO(cap); + ret = cap->r.rRet; break; - + default: - barf("schedule: invalid what_next field"); + barf("schedule: invalid what_next field"); } -#if defined(SMP) - // in SMP mode, we might return with a different capability than - // we started with, if the Haskell thread made a foreign call. So - // let's find out what our current Capability is: - cap = myCapability(); -#endif + 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; } - - cap->r.rInHaskell = 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; - + // 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); + // ---------------------------------------------------------------------- - /* Costs for the scheduler are assigned to CCS_SYSTEM */ + // Costs for the scheduler are assigned to CCS_SYSTEM #if defined(PROFILING) stopHeapProfTimer(); CCCS = CCS_SYSTEM; #endif - ACQUIRE_LOCK(&sched_mutex); - -#if defined(RTS_SUPPORTS_THREADS) - IF_DEBUG(scheduler,debugBelch("sched (task %p): ", osThreadId());); +#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 @@ -748,11 +699,11 @@ run_thread: break; case StackOverflow: - scheduleHandleStackOverflow(t); + scheduleHandleStackOverflow(cap,task,t); break; case ThreadYielding: - if (scheduleHandleYield(t, prev_what_next)) { + if (scheduleHandleYield(cap, t, prev_what_next)) { // shortcut for switching between compiler/interpreter: goto run_thread; } @@ -763,7 +714,8 @@ run_thread: break; case ThreadFinished: - if (scheduleHandleThreadFinished(mainThread, cap, t)) return;; + if (scheduleHandleThreadFinished(cap, task, t)) return cap; + ASSERT_FULL_CAPABILITY_INVARIANTS(cap,task); break; default: @@ -771,7 +723,9 @@ run_thread: } if (scheduleDoHeapProfile(ready_to_gc)) { ready_to_gc = rtsFalse; } - if (ready_to_gc) { scheduleDoGC(rtsFalse); } + if (ready_to_gc) { + cap = scheduleDoGC(cap,task,rtsFalse,GetRoots); + } } /* end of while() */ IF_PAR_DEBUG(verbose, @@ -780,7 +734,6 @@ run_thread: /* ---------------------------------------------------------------------------- * Setting up the scheduler loop - * ASSUMES: sched_mutex * ------------------------------------------------------------------------- */ static void @@ -805,72 +758,181 @@ schedulePreLoop(void) #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; + + // 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 + 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]; } + 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 - * ASSUMES: sched_mutex * ------------------------------------------------------------------------- */ +#if defined(RTS_USER_SIGNALS) && (!defined(THREADED_RTS) || defined(mingw32_HOST_OS)) static void -scheduleStartSignalHandlers(void) +scheduleStartSignalHandlers(Capability *cap) { -#if defined(RTS_USER_SIGNALS) && !defined(RTS_SUPPORTS_THREADS) - if (signals_pending()) { - RELEASE_LOCK(&sched_mutex); /* ToDo: kill */ - startSignalHandlers(); - ACQUIRE_LOCK(&sched_mutex); + if (signals_pending()) { // safe outside the lock + startSignalHandlers(cap); } -#endif } +#else +static void +scheduleStartSignalHandlers(Capability *cap STG_UNUSED) +{ +} +#endif /* ---------------------------------------------------------------------------- * Check for blocked threads that can be woken up. - * ASSUMES: sched_mutex * ------------------------------------------------------------------------- */ static void -scheduleCheckBlockedThreads(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 ( !EMPTY_QUEUE(blocked_queue_hd) || !EMPTY_QUEUE(sleeping_queue) ) + if ( !emptyQueue(blocked_queue_hd) || !emptyQueue(sleeping_queue) ) { -#if defined(RTS_SUPPORTS_THREADS) - // We shouldn't be here... - barf("schedule: awaitEvent() in threaded RTS"); -#else - awaitEvent( EMPTY_RUN_QUEUE() ); + awaitEvent( emptyRunQueue(cap) && !blackholes_need_checking ); + } #endif } /* ---------------------------------------------------------------------------- * Check for threads blocked on BLACKHOLEs that can be woken up - * ASSUMES: sched_mutex * ------------------------------------------------------------------------- */ static void -scheduleCheckBlackHoles( void ) +scheduleCheckBlackHoles (Capability *cap) { - if ( blackholes_need_checking ) + if ( blackholes_need_checking ) // check without the lock first { - checkBlackHoles(); - blackholes_need_checking = rtsFalse; + 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. - * ASSUMES: sched_mutex * ------------------------------------------------------------------------- */ static void -scheduleDetectDeadlock() +scheduleDetectDeadlock (Capability *cap, Task *task) { #if defined(PARALLEL_HASKELL) - // ToDo: add deadlock detection in GUM (similar to SMP) -- HWL + // ToDo: add deadlock detection in GUM (similar to THREADED_RTS) -- HWL return; #endif @@ -880,9 +942,9 @@ scheduleDetectDeadlock() * other tasks are waiting for work, we must have a deadlock of * some description. */ - if ( EMPTY_THREAD_QUEUES() ) + if ( emptyThreadQueues(cap) ) { -#if defined(RTS_SUPPORTS_THREADS) +#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 @@ -899,12 +961,13 @@ scheduleDetectDeadlock() // 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); - scheduleDoGC( rtsTrue/*force major GC*/ ); recent_activity = ACTIVITY_DONE_GC; - if ( !EMPTY_RUN_QUEUE() ) return; + + if ( !emptyRunQueue(cap) ) return; -#if defined(RTS_USER_SIGNALS) && !defined(RTS_SUPPORTS_THREADS) +#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. @@ -916,36 +979,31 @@ scheduleDetectDeadlock() awaitUserSignals(); if (signals_pending()) { - RELEASE_LOCK(&sched_mutex); - startSignalHandlers(); - ACQUIRE_LOCK(&sched_mutex); + startSignalHandlers(cap); } // either we have threads to run, or we were interrupted: - ASSERT(!EMPTY_RUN_QUEUE() || interrupted); + ASSERT(!emptyRunQueue(cap) || sched_state >= SCHED_INTERRUPTING); } #endif -#if !defined(RTS_SUPPORTS_THREADS) +#if !defined(THREADED_RTS) /* Probably a real deadlock. Send the current main thread the - * Deadlock exception (or in the SMP build, send *all* main - * threads the deadlock exception, since none of them can make - * progress). + * Deadlock exception. */ - { - StgMainThread *m; - m = main_threads; - switch (m->tso->why_blocked) { + if (task->tso) { + switch (task->tso->why_blocked) { case BlockedOnSTM: case BlockedOnBlackHole: case BlockedOnException: case BlockedOnMVar: - raiseAsync(m->tso, (StgClosure *)NonTermination_closure); + raiseAsync(cap, task->tso, (StgClosure *)NonTermination_closure); return; default: barf("deadlock: main thread blocked in a strange way"); } } + return; #endif } } @@ -1132,7 +1190,7 @@ static void scheduleActivateSpark(void) { #if defined(SPARKS) - ASSERT(EMPTY_RUN_QUEUE()); + 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 @@ -1191,7 +1249,7 @@ scheduleActivateSpark(void) static rtsBool scheduleGetRemoteWork(rtsBool *receivedFinish) { - ASSERT(EMPTY_RUN_QUEUE()); + ASSERT(emptyRunQueue()); if (RtsFlags.ParFlags.BufferTime) { IF_PAR_DEBUG(verbose, @@ -1370,7 +1428,6 @@ JB: TODO: investigate wether state change field could be nuked /* ---------------------------------------------------------------------------- * After running a thread... - * ASSUMES: sched_mutex * ------------------------------------------------------------------------- */ static void @@ -1462,7 +1519,6 @@ schedulePostRunThread(void) /* ----------------------------------------------------------------------------- * Handle a thread that returned to the scheduler with ThreadHeepOverflow - * ASSUMES: sched_mutex * -------------------------------------------------------------------------- */ static rtsBool @@ -1485,7 +1541,9 @@ scheduleHandleHeapOverflow( Capability *cap, StgTSO *t ) 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 @@ -1494,7 +1552,7 @@ scheduleHandleHeapOverflow( Capability *cap, StgTSO *t ) if (cap->r.rCurrentNursery->u.back != NULL) { cap->r.rCurrentNursery->u.back->link = bd; } else { -#if !defined(SMP) +#if !defined(THREADED_RTS) ASSERT(g0s0->blocks == cap->r.rCurrentNursery && g0s0 == cap->r.rNursery); #endif @@ -1529,7 +1587,7 @@ scheduleHandleHeapOverflow( Capability *cap, StgTSO *t ) // run queue before us and steal the large block, but in that // case the thread will just end up requesting another large // block. - PUSH_ON_RUN_QUEUE(t); + pushOnRunQueue(cap,t); return rtsFalse; /* not actually GC'ing */ } } @@ -1550,18 +1608,17 @@ scheduleHandleHeapOverflow( Capability *cap, StgTSO *t ) } #endif - PUSH_ON_RUN_QUEUE(t); + 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 - * ASSUMES: sched_mutex * -------------------------------------------------------------------------- */ static void -scheduleHandleStackOverflow( StgTSO *t) +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])); @@ -1570,26 +1627,24 @@ scheduleHandleStackOverflow( StgTSO *t) */ { /* enlarge the stack */ - StgTSO *new_t = threadStackOverflow(t); + StgTSO *new_t = threadStackOverflow(cap, t); - /* This TSO has moved, so update any pointers to it from the - * main thread stack. It better not be on any other queues... - * (it shouldn't be). + /* The TSO attached to this Task may have moved, so update the + * pointer to it. */ - if (t->main != NULL) { - t->main->tso = new_t; + if (task->tso == t) { + task->tso = new_t; } - PUSH_ON_RUN_QUEUE(new_t); + pushOnRunQueue(cap,new_t); } } /* ----------------------------------------------------------------------------- * Handle a thread that returned to the scheduler with ThreadYielding - * ASSUMES: sched_mutex * -------------------------------------------------------------------------- */ static rtsBool -scheduleHandleYield( StgTSO *t, nat prev_what_next ) +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 @@ -1635,7 +1690,7 @@ scheduleHandleYield( StgTSO *t, nat prev_what_next ) #endif - addToRunQueue(t); + addToRunQueue(cap,t); #if defined(GRAN) /* add a ContinueThread event to actually process the thread */ @@ -1652,7 +1707,6 @@ scheduleHandleYield( StgTSO *t, nat prev_what_next ) /* ----------------------------------------------------------------------------- * Handle a thread that returned to the scheduler with ThreadBlocked - * ASSUMES: sched_mutex * -------------------------------------------------------------------------- */ static void @@ -1675,100 +1729,883 @@ scheduleHandleThreadBlocked( StgTSO *t /* ngoq Dogh! ASSERT(procStatus[CurrentProc]==Busy || - ((procStatus[Curren - * from its creation - */ + ((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 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; */ + 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 +} - /* Link the new thread on the global thread list. - */ - tso->global_link = all_threads; - all_threads = tso; +/* ----------------------------------------------------------------------------- + * Handle a thread that returned to the scheduler with ThreadFinished + * -------------------------------------------------------------------------- */ -#if defined(DIST) - tso->dist.priority = MandatoryPriority; //by default that is... -#endif +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) - tso->gran.pri = pri; -# if defined(DEBUG) - tso->gran.magic = TSO_MAGIC; // debugging only + 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 - 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 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_DEBUG(gran,printTSO(tso)); -#elif defined(PARALLEL_HASKELL) -# if defined(DEBUG) - tso->par.magic = TSO_MAGIC; // debugging only +# 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 - 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 +#endif // PARALLEL_HASKELL -#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 + // + // 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 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)); + 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 - IF_DEBUG(scheduler,sched_belch("created thread %ld, stack size = %lx words", - (long)tso->id, (long)tso->stack_size)); + // 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 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) { + 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->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; + return tso; } #if defined(PAR) @@ -1831,85 +2668,83 @@ activateSpark (rtsSpark spark) /* --------------------------------------------------------------------------- * scheduleThread() * - * scheduleThread puts a thread on the head of the runnable queue. + * 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. * ------------------------------------------------------------------------ */ -static void -scheduleThread_(StgTSO *tso) +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. - APPEND_TO_RUN_QUEUE(tso); - threadRunnable(); + // 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 -scheduleThread(StgTSO* tso) +Capability * +scheduleWaitThread (StgTSO* tso, /*[out]*/HaskellObj* ret, Capability *cap) { - ACQUIRE_LOCK(&sched_mutex); - scheduleThread_(tso); - RELEASE_LOCK(&sched_mutex); -} + Task *task; + + // We already created/initialised the Task + task = cap->running_task; -#if defined(RTS_SUPPORTS_THREADS) -static Condition bound_cond_cache; -static int bound_cond_cache_full = 0; + // This TSO is now a bound thread; make the Task and TSO + // point to each other. + tso->bound = task; + + 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; +} -SchedulerStatus -scheduleWaitThread(StgTSO* tso, /*[out]*/HaskellObj* ret, - Capability *initialCapability) +/* ---------------------------------------------------------------------------- + * Starting Tasks + * ------------------------------------------------------------------------- */ + +#if defined(THREADED_RTS) +void +workerStart(Task *task) { - // Precondition: sched_mutex must be held - StgMainThread *m; + Capability *cap; - m = stgMallocBytes(sizeof(StgMainThread), "waitThread"); - m->tso = tso; - tso->main = m; - m->ret = ret; - m->stat = NoStatus; - m->link = main_threads; - m->prev = NULL; - if (main_threads != NULL) { - main_threads->prev = m; - } - main_threads = m; + // See startWorkerTask(). + ACQUIRE_LOCK(&task->lock); + cap = task->cap; + RELEASE_LOCK(&task->lock); -#if defined(RTS_SUPPORTS_THREADS) - // Allocating a new condition for each thread is expensive, so we - // cache one. This is a pretty feeble hack, but it helps speed up - // consecutive call-ins quite a bit. - if (bound_cond_cache_full) { - m->bound_thread_cond = bound_cond_cache; - bound_cond_cache_full = 0; - } else { - initCondition(&m->bound_thread_cond); - } -#endif + // set the thread-local pointer to the Task: + taskEnter(task); - /* Put the thread on the main-threads list prior to scheduling the TSO. - Failure to do so introduces a race condition in the MT case (as - identified by Wolfgang Thaller), whereby the new task/OS thread - created by scheduleThread_() would complete prior to the thread - that spawned it managed to put 'itself' on the main-threads list. - The upshot of it all being that the worker thread wouldn't get to - signal the completion of the its work item for the main thread to - see (==> it got stuck waiting.) -- sof 6/02. - */ - IF_DEBUG(scheduler, sched_belch("waiting for thread (%d)", tso->id)); - - APPEND_TO_RUN_QUEUE(tso); - // NB. Don't call threadRunnable() here, because the thread is - // bound and only runnable by *this* OS thread, so waking up other - // workers will just slow things down. + // schedule() runs without a lock. + cap = schedule(cap,task); - return waitThread_(m, initialCapability); + // On exit from schedule(), we have a Capability. + releaseCapability(cap); + taskStop(task); } +#endif /* --------------------------------------------------------------------------- * initScheduler() @@ -1925,7 +2760,6 @@ 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; @@ -1935,53 +2769,58 @@ initScheduler(void) blackhole_queue[i] = END_TSO_QUEUE; sleeping_queue = END_TSO_QUEUE; } -#else - run_queue_hd = END_TSO_QUEUE; - run_queue_tl = END_TSO_QUEUE; +#elif !defined(THREADED_RTS) blocked_queue_hd = END_TSO_QUEUE; blocked_queue_tl = END_TSO_QUEUE; - blackhole_queue = END_TSO_QUEUE; sleeping_queue = END_TSO_QUEUE; -#endif - - suspended_ccalling_threads = END_TSO_QUEUE; +#endif - main_threads = NULL; - all_threads = END_TSO_QUEUE; + blackhole_queue = END_TSO_QUEUE; + all_threads = END_TSO_QUEUE; context_switch = 0; - interrupted = 0; + sched_state = SCHED_RUNNING; RtsFlags.ConcFlags.ctxtSwitchTicks = RtsFlags.ConcFlags.ctxtSwitchTime / TICK_MILLISECS; -#if defined(RTS_SUPPORTS_THREADS) +#if defined(THREADED_RTS) /* Initialise the mutex and condition variables used by * the scheduler. */ initMutex(&sched_mutex); - initMutex(&term_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 SMP builds have more than one). + * floating around (only THREADED_RTS builds have more than one). */ initCapabilities(); - -#if defined(RTS_SUPPORTS_THREADS) + initTaskManager(); -#endif -#if defined(SMP) - /* eagerly start some extra workers */ - startingWorkerThread = RtsFlags.ParFlags.nNodes; - startTasks(RtsFlags.ParFlags.nNodes, taskStart); +#if defined(THREADED_RTS) || defined(PARALLEL_HASKELL) + initSparkPools(); #endif -#if /* defined(SMP) ||*/ defined(PARALLEL_HASKELL) - initSparkPools(); +#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); @@ -1990,59 +2829,32 @@ initScheduler(void) void exitScheduler( void ) { - interrupted = rtsTrue; - shutting_down_scheduler = rtsTrue; -#if defined(RTS_SUPPORTS_THREADS) - if (threadIsTask(osThreadId())) { taskStop(); } - stopTaskManager(); -#endif -} - -/* ---------------------------------------------------------------------------- - Managing the per-task allocation areas. - - Each capability comes with an allocation area. These are - fixed-length block lists into which allocation can be done. - - ToDo: no support for two-space collection at the moment??? - ------------------------------------------------------------------------- */ - -static SchedulerStatus -waitThread_(StgMainThread* m, Capability *initialCapability) -{ - SchedulerStatus stat; + Task *task = NULL; - // Precondition: sched_mutex must be held. - IF_DEBUG(scheduler, sched_belch("new main thread (%d)", m->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 - schedule(m,initialCapability); -#else - schedule(m,initialCapability); - ASSERT(m->stat != NoStatus); +#if defined(THREADED_RTS) + ACQUIRE_LOCK(&sched_mutex); + task = newBoundTask(); + RELEASE_LOCK(&sched_mutex); #endif - stat = m->stat; + // 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(RTS_SUPPORTS_THREADS) - // Free the condition variable, returning it to the cache if possible. - if (!bound_cond_cache_full) { - bound_cond_cache = m->bound_thread_cond; - bound_cond_cache_full = 1; - } else { - closeCondition(&m->bound_thread_cond); - } +#if defined(THREADED_RTS) + { + nat i; + + for (i = 0; i < n_capabilities; i++) { + shutdownCapability(&capabilities[i], task); + } + boundTaskExiting(task); + stopTaskManager(); + } #endif - - IF_DEBUG(scheduler, sched_belch("main thread (%d) finished", m->tso->id)); - stgFree(m); - - // Postcondition: sched_mutex still held - return stat; } /* --------------------------------------------------------------------------- @@ -2064,59 +2876,57 @@ waitThread_(StgMainThread* m, Capability *initialCapability) void GetRoots( evac_fn evac ) { -#if defined(GRAN) - { 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]); + 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(); + markEventQueue(); #else /* !GRAN */ - if (run_queue_hd != END_TSO_QUEUE) { - ASSERT(run_queue_tl != END_TSO_QUEUE); - evac((StgClosure **)&run_queue_hd); - evac((StgClosure **)&run_queue_tl); - } - - if (blocked_queue_hd != END_TSO_QUEUE) { - ASSERT(blocked_queue_tl != END_TSO_QUEUE); - evac((StgClosure **)&blocked_queue_hd); - evac((StgClosure **)&blocked_queue_tl); - } - - if (sleeping_queue != END_TSO_QUEUE) { - evac((StgClosure **)&sleeping_queue); - } -#endif - if (blackhole_queue != END_TSO_QUEUE) { - evac((StgClosure **)&blackhole_queue); - } + for (i = 0; i < n_capabilities; i++) { + cap = &capabilities[i]; + evac((StgClosure **)&cap->run_queue_hd); + evac((StgClosure **)&cap->run_queue_tl); + + 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 **)&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 - if (suspended_ccalling_threads != END_TSO_QUEUE) { - evac((StgClosure **)&suspended_ccalling_threads); - } + // evac((StgClosure **)&blackhole_queue); -#if defined(PARALLEL_HASKELL) || defined(GRAN) - markSparkQueue(evac); +#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); + // mark the signal handlers (signals should be already blocked) + markSignalHandlers(evac); #endif } @@ -2135,21 +2945,32 @@ GetRoots( evac_fn evac ) 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) { - /* Obligated to hold this lock upon entry */ - ACQUIRE_LOCK(&sched_mutex); - GarbageCollect(GetRoots,rtsFalse); - RELEASE_LOCK(&sched_mutex); + performGC_(rtsFalse, GetRoots); } void performMajorGC(void) { - ACQUIRE_LOCK(&sched_mutex); - GarbageCollect(GetRoots,rtsTrue); - RELEASE_LOCK(&sched_mutex); + performGC_(rtsTrue, GetRoots); } static void @@ -2162,10 +2983,8 @@ AllRoots(evac_fn evac) void performGCWithRoots(void (*get_roots)(evac_fn)) { - ACQUIRE_LOCK(&sched_mutex); - extra_roots = get_roots; - GarbageCollect(AllRoots,rtsFalse); - RELEASE_LOCK(&sched_mutex); + extra_roots = get_roots; + performGC_(rtsFalse, AllRoots); } /* ----------------------------------------------------------------------------- @@ -2178,7 +2997,7 @@ performGCWithRoots(void (*get_roots)(evac_fn)) -------------------------------------------------------------------------- */ static StgTSO * -threadStackOverflow(StgTSO *tso) +threadStackOverflow(Capability *cap, StgTSO *tso) { nat new_stack_size, stack_words; lnat new_tso_size; @@ -2196,7 +3015,7 @@ threadStackOverflow(StgTSO *tso) tso->sp+64))); /* Send this thread the StackOverflow exception */ - raiseAsync(tso, (StgClosure *)stackOverflow_closure); + raiseAsync(cap, tso, (StgClosure *)stackOverflow_closure); return tso; } @@ -2210,7 +3029,7 @@ threadStackOverflow(StgTSO *tso) new_tso_size = round_to_mblocks(new_tso_size); /* Be MBLOCK-friendly */ new_stack_size = new_tso_size - TSO_STRUCT_SIZEW; - IF_DEBUG(scheduler, debugBelch("== sched: increasing stack size from %d words to %d.\n", tso->stack_size, new_stack_size)); + 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); @@ -2271,7 +3090,7 @@ unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node ) DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC, GR_RESUMEQ, ((StgTSO *)bqe), ((StgTSO *)bqe)->block_info.closure, 0, 0 /* spark_queue_len(ADVISORY_POOL) */); - if (EMPTY_RUN_QUEUE()) + if (emptyRunQueue()) emitSchedule = rtsTrue; switch (get_itbl(node)->type) { @@ -2288,7 +3107,7 @@ unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node ) break; #endif default: - barf("{unblockOneLocked}Daq Qagh: unexpected closure in blocking queue"); + barf("{unblockOne}Daq Qagh: unexpected closure in blocking queue"); } } } @@ -2296,7 +3115,7 @@ unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node ) #if defined(GRAN) StgBlockingQueueElement * -unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node) +unblockOne(StgBlockingQueueElement *bqe, StgClosure *node) { StgTSO *tso; PEs node_loc, tso_loc; @@ -2336,7 +3155,7 @@ unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node) } #elif defined(PARALLEL_HASKELL) StgBlockingQueueElement * -unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node) +unblockOne(StgBlockingQueueElement *bqe, StgClosure *node) { StgBlockingQueueElement *next; @@ -2371,7 +3190,7 @@ unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node) break; default: - barf("{unblockOneLocked}Daq Qagh: Unexpected IP (%#lx; %s) in blocking queue at %#lx\n", + 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 @@ -2379,10 +3198,10 @@ unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node) IF_PAR_DEBUG(bq, debugBelch(", %p (%s)\n", bqe, info_type((StgClosure*)bqe))); return next; } +#endif -#else /* !GRAN && !PARALLEL_HASKELL */ StgTSO * -unblockOneLocked(StgTSO *tso) +unblockOne(Capability *cap, StgTSO *tso) { StgTSO *next; @@ -2391,32 +3210,21 @@ unblockOneLocked(StgTSO *tso) tso->why_blocked = NotBlocked; next = tso->link; tso->link = END_TSO_QUEUE; - APPEND_TO_RUN_QUEUE(tso); - threadRunnable(); + + // We might have just migrated this TSO to our Capability: + if (tso->bound) { + tso->bound->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; IF_DEBUG(scheduler,sched_belch("waking up thread %ld", (long)tso->id)); return next; } -#endif -#if defined(GRAN) || defined(PARALLEL_HASKELL) -INLINE_ME StgBlockingQueueElement * -unblockOne(StgBlockingQueueElement *bqe, StgClosure *node) -{ - ACQUIRE_LOCK(&sched_mutex); - bqe = unblockOneLocked(bqe, node); - RELEASE_LOCK(&sched_mutex); - return bqe; -} -#else -INLINE_ME StgTSO * -unblockOne(StgTSO *tso) -{ - ACQUIRE_LOCK(&sched_mutex); - tso = unblockOneLocked(tso); - RELEASE_LOCK(&sched_mutex); - return tso; -} -#endif #if defined(GRAN) void @@ -2468,7 +3276,7 @@ awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node) //tso = (StgTSO *)bqe; // wastes an assignment to get the type right //tso_loc = where_is(tso); len++; - bqe = unblockOneLocked(bqe, node); + bqe = unblockOne(bqe, node); } /* if this is the BQ of an RBH, we have to put back the info ripped out of @@ -2506,8 +3314,6 @@ awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node) { StgBlockingQueueElement *bqe; - ACQUIRE_LOCK(&sched_mutex); - IF_PAR_DEBUG(verbose, debugBelch("##-_ AwBQ for node %p on [%x]: \n", node, mytid)); @@ -2527,29 +3333,20 @@ awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node) bqe = q; while (get_itbl(bqe)->type==TSO || get_itbl(bqe)->type==BLOCKED_FETCH) { - bqe = unblockOneLocked(bqe, node); + bqe = unblockOne(bqe, node); } - RELEASE_LOCK(&sched_mutex); } #else /* !GRAN && !PARALLEL_HASKELL */ void -awakenBlockedQueueNoLock(StgTSO *tso) -{ - while (tso != END_TSO_QUEUE) { - tso = unblockOneLocked(tso); - } -} - -void -awakenBlockedQueue(StgTSO *tso) +awakenBlockedQueue(Capability *cap, StgTSO *tso) { - ACQUIRE_LOCK(&sched_mutex); - while (tso != END_TSO_QUEUE) { - tso = unblockOneLocked(tso); - } - RELEASE_LOCK(&sched_mutex); + if (tso == NULL) return; // hack; see bug #1235728, and comments in + // Exception.cmm + while (tso != END_TSO_QUEUE) { + tso = unblockOne(cap,tso); + } } #endif @@ -2561,13 +3358,11 @@ awakenBlockedQueue(StgTSO *tso) void interruptStgRts(void) { - interrupted = 1; + sched_state = SCHED_INTERRUPTING; context_switch = 1; - threadRunnable(); - /* ToDo: if invoked from a signal handler, this threadRunnable - * only works if there's another thread (not this one) waiting to - * be woken up. - */ +#if defined(THREADED_RTS) + prodAllCapabilities(); +#endif } /* ----------------------------------------------------------------------------- @@ -2588,7 +3383,7 @@ interruptStgRts(void) */ static void -unblockThread(StgTSO *tso) +unblockThread(Capability *cap, StgTSO *tso) { StgBlockingQueueElement *t, **last; @@ -2729,11 +3524,11 @@ unblockThread(StgTSO *tso) tso->link = END_TSO_QUEUE; tso->why_blocked = NotBlocked; tso->block_info.closure = NULL; - PUSH_ON_RUN_QUEUE(tso); + pushOnRunQueue(cap,tso); } #else static void -unblockThread(StgTSO *tso) +unblockThread(Capability *cap, StgTSO *tso) { StgTSO *t, **last; @@ -2810,6 +3605,7 @@ unblockThread(StgTSO *tso) barf("unblockThread (Exception): TSO not found"); } +#if !defined(THREADED_RTS) case BlockedOnRead: case BlockedOnWrite: #if defined(mingw32_HOST_OS) @@ -2859,6 +3655,7 @@ unblockThread(StgTSO *tso) } barf("unblockThread (delay): TSO not found"); } +#endif default: barf("unblockThread"); @@ -2868,7 +3665,12 @@ unblockThread(StgTSO *tso) tso->link = END_TSO_QUEUE; tso->why_blocked = NotBlocked; tso->block_info.closure = NULL; - APPEND_TO_RUN_QUEUE(tso); + appendToRunQueue(cap,tso); + + // We might have just migrated this TSO to our Capability: + if (tso->bound) { + tso->bound->cap = cap; + } } #endif @@ -2888,12 +3690,15 @@ unblockThread(StgTSO *tso) * -------------------------------------------------------------------------- */ static rtsBool -checkBlackHoles( void ) +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 @@ -2903,7 +3708,10 @@ checkBlackHoles( void ) ASSERT(t->why_blocked == BlockedOnBlackHole); type = get_itbl(t->block_info.closure)->type; if (type != BLACKHOLE && type != CAF_BLACKHOLE) { - t = unblockOneLocked(t); + 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 { @@ -2945,59 +3753,32 @@ checkBlackHoles( void ) * CATCH_FRAME on the stack. In either case, we strip the entire * stack and replace the thread with a zombie. * - * Locks: sched_mutex held upon entry nor exit. + * 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 -deleteThread(StgTSO *tso) -{ - if (tso->why_blocked != BlockedOnCCall && - tso->why_blocked != BlockedOnCCall_NoUnblockExc) { - raiseAsync(tso,NULL); - } -} - -#ifdef FORKPROCESS_PRIMOP_SUPPORTED -static void -deleteThreadImmediately(StgTSO *tso) -{ // for forkProcess only: - // delete thread without giving it a chance to catch the KillThread exception - - if (tso->what_next == ThreadComplete || tso->what_next == ThreadKilled) { - return; - } - - if (tso->why_blocked != BlockedOnCCall && - tso->why_blocked != BlockedOnCCall_NoUnblockExc) { - unblockThread(tso); - } - - tso->what_next = ThreadKilled; -} -#endif - + * -------------------------------------------------------------------------- */ + void -raiseAsyncWithLock(StgTSO *tso, StgClosure *exception) +raiseAsync(Capability *cap, StgTSO *tso, StgClosure *exception) { - /* When raising async exs from contexts where sched_mutex isn't held; - use raiseAsyncWithLock(). */ - ACQUIRE_LOCK(&sched_mutex); - raiseAsync(tso,exception); - RELEASE_LOCK(&sched_mutex); + raiseAsync_(cap, tso, exception, rtsFalse, NULL); } void -raiseAsync(StgTSO *tso, StgClosure *exception) +suspendComputation(Capability *cap, StgTSO *tso, StgPtr stop_here) { - raiseAsync_(tso, exception, rtsFalse); + raiseAsync_(cap, tso, NULL, rtsFalse, stop_here); } static void -raiseAsync_(StgTSO *tso, StgClosure *exception, rtsBool stop_at_atomically) +raiseAsync_(Capability *cap, StgTSO *tso, StgClosure *exception, + rtsBool stop_at_atomically, StgPtr stop_here) { StgRetInfoTable *info; - StgPtr sp; + StgPtr sp, frame; + nat i; // Thread already dead? if (tso->what_next == ThreadComplete || tso->what_next == ThreadKilled) { @@ -3008,7 +3789,10 @@ raiseAsync_(StgTSO *tso, StgClosure *exception, rtsBool stop_at_atomically) sched_belch("raising exception in thread %ld.", (long)tso->id)); // Remove it from any blocking queues - unblockThread(tso); + unblockThread(cap,tso); + + // mark it dirty; we're about to change its stack. + dirtyTSO(tso); sp = tso->sp; @@ -3022,8 +3806,8 @@ raiseAsync_(StgTSO *tso, StgClosure *exception, rtsBool stop_at_atomically) sp[0] = (W_)&stg_dummy_ret_closure; } - while (1) { - nat i; + frame = sp + 1; + while (stop_here == NULL || frame < stop_here) { // 1. Let the top of the stack be the "current closure" // @@ -3043,95 +3827,10 @@ raiseAsync_(StgTSO *tso, StgClosure *exception, rtsBool stop_at_atomically) // NB: if we pass an ATOMICALLY_FRAME then abort the associated // transaction - - StgPtr frame; - - frame = sp + 1; info = get_ret_itbl((StgClosure *)frame); - - while (info->i.type != UPDATE_FRAME - && (info->i.type != CATCH_FRAME || exception == NULL) - && info->i.type != STOP_FRAME - && (info->i.type != ATOMICALLY_FRAME || stop_at_atomically == rtsFalse)) - { - if (info->i.type == CATCH_RETRY_FRAME || info->i.type == ATOMICALLY_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")); - stmAbortTransaction(tso -> trec); - tso -> trec = stmGetEnclosingTRec(tso -> trec); - } - frame += stack_frame_sizeW((StgClosure *)frame); - info = get_ret_itbl((StgClosure *)frame); - } - + switch (info->i.type) { - - case ATOMICALLY_FRAME: - ASSERT(stop_at_atomically); - ASSERT(stmGetEnclosingTRec(tso->trec) == NO_TREC); - stmCondemnTransaction(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; - 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; - - // we've got an exception to raise, so let's pass it to the - // handler in this frame. - // - raise = (StgThunk *)allocate(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 UPDATE_FRAME: { StgAP_STACK * ap; @@ -3142,7 +3841,7 @@ raiseAsync_(StgTSO *tso, StgClosure *exception, rtsBool stop_at_atomically) // fun field. // words = frame - sp - 1; - ap = (StgAP_STACK *)allocate(AP_STACK_sizeW(words)); + ap = (StgAP_STACK *)allocateLocal(cap,AP_STACK_sizeW(words)); ap->size = words; ap->fun = (StgClosure *)sp[0]; @@ -3162,9 +3861,7 @@ raiseAsync_(StgTSO *tso, StgClosure *exception, rtsBool stop_at_atomically) printObj((StgClosure *)ap); ); - // Replace the updatee with an indirection - happily - // this will also wake up any threads currently - // waiting on the result. + // 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 @@ -3181,22 +3878,140 @@ raiseAsync_(StgTSO *tso, StgClosure *exception, rtsBool stop_at_atomically) } sp += sizeofW(StgUpdateFrame) - 1; sp[0] = (W_)ap; // push onto stack - break; + frame = sp + 1; + continue; //no need to bump frame } - + case STOP_FRAME: // We've stripped the entire stack, the thread is now dead. - sp += sizeofW(StgStopFrame); tso->what_next = ThreadKilled; - tso->sp = sp; + 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: - barf("raiseAsync"); + 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); } - barf("raiseAsync"); } +#endif /* ----------------------------------------------------------------------------- raiseExceptionHelper @@ -3207,8 +4022,9 @@ raiseAsync_(StgTSO *tso, StgClosure *exception, rtsBool stop_at_atomically) -------------------------------------------------------------------------- */ StgWord -raiseExceptionHelper (StgTSO *tso, StgClosure *exception) +raiseExceptionHelper (StgRegTable *reg, StgTSO *tso, StgClosure *exception) { + Capability *cap = regTableToCapability(reg); StgThunk *raise_closure = NULL; StgPtr p, next; StgRetInfoTable *info; @@ -3218,7 +4034,7 @@ raiseExceptionHelper (StgTSO *tso, StgClosure *exception) // 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 @@ -3246,7 +4062,7 @@ raiseExceptionHelper (StgTSO *tso, StgClosure *exception) // Only create raise_closure if we need to. if (raise_closure == NULL) { raise_closure = - (StgThunk *)allocate(sizeofW(StgThunk)+MIN_UPD_SIZE); + (StgThunk *)allocateLocal(cap,sizeofW(StgThunk)+1); SET_HDR(raise_closure, &stg_raise_info, CCCS); raise_closure->payload[0] = exception; } @@ -3336,42 +4152,51 @@ findRetryFrameHelper (StgTSO *tso) on an MVar, or NonTermination if the thread was blocked on a Black Hole. - Locks: sched_mutex isn't held upon entry nor exit. + Locks: assumes we hold *all* the capabilities. -------------------------------------------------------------------------- */ void -resurrectThreads( StgTSO *threads ) +resurrectThreads (StgTSO *threads) { - StgTSO *tso, *next; - - 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)); - - switch (tso->why_blocked) { - case BlockedOnMVar: - case BlockedOnException: - /* Called by GC - sched_mutex lock is currently held. */ - raiseAsync(tso,(StgClosure *)BlockedOnDeadMVar_closure); - break; - case BlockedOnBlackHole: - raiseAsync(tso,(StgClosure *)NonTermination_closure); - break; - case BlockedOnSTM: - raiseAsync(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"); + 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 for a + // bound thread, or last_free_capability otherwise. + if (tso->bound) { + cap = tso->bound->cap; + } else { + cap = last_free_capability; + } + + 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"); + } } - } } /* ---------------------------------------------------------------------------- @@ -3380,6 +4205,7 @@ resurrectThreads( StgTSO *threads ) * at the Haskell source code level, so enable outside of DEBUG. --sof 7/02] ------------------------------------------------------------------------- */ +#if DEBUG static void printThreadBlockage(StgTSO *tso) { @@ -3436,12 +4262,170 @@ printThreadBlockage(StgTSO *tso) } } -static void -printThreadStatus(StgTSO *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) { - switch (tso->what_next) { - case ThreadKilled: - debugBelch("has + 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 || @@ -3495,33 +4479,34 @@ printThreadStatus(StgTSO *tso) 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; + 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 RTS_SUPPORTS_THREADS - debugBelch("sched (task %p): ", osThreadId()); + 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("== "); + debugBelch("== "); #else - debugBelch("sched: "); + debugBelch("sched: "); #endif - vdebugBelch(s, ap); - debugBelch("\n"); - va_end(ap); + vdebugBelch(s, ap); + debugBelch("\n"); + va_end(ap); } #endif /* DEBUG */