X-Git-Url: http://git.megacz.com/?p=ghc-hetmet.git;a=blobdiff_plain;f=rts%2FSchedule.c;h=b3d523e111488b5830b650b656970889f045aced;hp=626c097a82ae47e9a1be8cc8b06eeeac1f2413fd;hb=a2a67cd520b9841114d69a87a423dabcb3b4368e;hpb=cd22cec99074fe9a4005c8d7b2de227fca285798 diff --git a/rts/Schedule.c b/rts/Schedule.c index 626c097..b3d523e 100644 --- a/rts/Schedule.c +++ b/rts/Schedule.c @@ -9,31 +9,25 @@ #include "PosixSource.h" #define KEEP_LOCKCLOSURE #include "Rts.h" -#include "SchedAPI.h" + +#include "sm/Storage.h" #include "RtsUtils.h" -#include "RtsFlags.h" -#include "OSThreads.h" -#include "Storage.h" #include "StgRun.h" -#include "Hooks.h" #include "Schedule.h" -#include "StgMiscClosures.h" #include "Interpreter.h" #include "Printer.h" #include "RtsSignals.h" #include "Sanity.h" #include "Stats.h" #include "STM.h" -#include "Timer.h" #include "Prelude.h" #include "ThreadLabels.h" -#include "LdvProfile.h" #include "Updates.h" #include "Proftimer.h" #include "ProfHeap.h" - -/* PARALLEL_HASKELL includes go here */ - +#include "Weak.h" +#include "eventlog/EventLog.h" +#include "sm/GC.h" // waitForGcThreads, releaseGCThreads, N #include "Sparks.h" #include "Capability.h" #include "Task.h" @@ -44,7 +38,8 @@ #include "Trace.h" #include "RaiseAsync.h" #include "Threads.h" -#include "ThrIOManager.h" +#include "Timer.h" +#include "ThreadPaused.h" #ifdef HAVE_SYS_TYPES_H #include @@ -61,12 +56,6 @@ #include #endif -// Turn off inlining when debugging - it obfuscates things -#ifdef DEBUG -# undef STATIC_INLINE -# define STATIC_INLINE static -#endif - /* ----------------------------------------------------------------------------- * Global variables * -------------------------------------------------------------------------- */ @@ -89,16 +78,24 @@ StgTSO *blackhole_queue = NULL; */ rtsBool blackholes_need_checking = rtsFalse; +/* Set to true when the latest garbage collection failed to reclaim + * enough space, and the runtime should proceed to shut itself down in + * an orderly fashion (emitting profiling info etc.) + */ +rtsBool heap_overflow = rtsFalse; + /* 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). + * + * NB. must be StgWord, we do xchg() on it. */ -nat recent_activity = ACTIVITY_YES; +volatile StgWord recent_activity = ACTIVITY_YES; /* if this flag is set as well, give up execution - * LOCK: none (changes once, from false->true) + * LOCK: none (changes monotonically) */ -rtsBool sched_state = SCHED_RUNNING; +volatile StgWord sched_state = SCHED_RUNNING; /* This is used in `TSO.h' and gcc 2.96 insists that this variable actually * exists - earlier gccs apparently didn't. @@ -137,17 +134,17 @@ static Capability *schedule (Capability *initialCapability, Task *task); // scheduler clearer. // static void schedulePreLoop (void); +static void scheduleFindWork (Capability *cap); #if defined(THREADED_RTS) -static void schedulePushWork(Capability *cap, Task *task); +static void scheduleYield (Capability **pcap, Task *task); #endif static void scheduleStartSignalHandlers (Capability *cap); static void scheduleCheckBlockedThreads (Capability *cap); static void scheduleCheckWakeupThreads(Capability *cap USED_IF_NOT_THREADS); static void scheduleCheckBlackHoles (Capability *cap); static void scheduleDetectDeadlock (Capability *cap, Task *task); -#if defined(PARALLEL_HASKELL) -static rtsBool scheduleGetRemoteWork(Capability *cap); -static void scheduleSendPendingMessages(void); +static void schedulePushWork(Capability *cap, Task *task); +#if defined(THREADED_RTS) static void scheduleActivateSpark(Capability *cap); #endif static void schedulePostRunThread(Capability *cap, StgTSO *t); @@ -193,18 +190,8 @@ static char *whatNext_strs[] = { STATIC_INLINE void addToRunQueue( Capability *cap, StgTSO *t ) { -#if defined(PARALLEL_HASKELL) - if (RtsFlags.ParFlags.doFairScheduling) { - // this does round-robin scheduling; good for concurrency - appendToRunQueue(cap,t); - } else { - // this does unfair scheduling; good for parallelism - pushOnRunQueue(cap,t); - } -#else // this does round-robin scheduling; good for concurrency appendToRunQueue(cap,t); -#endif } /* --------------------------------------------------------------------------- @@ -249,9 +236,6 @@ schedule (Capability *initialCapability, Task *task) StgTSO *t; Capability *cap; StgThreadReturnCode ret; -#if defined(PARALLEL_HASKELL) - rtsBool receivedFinish = rtsFalse; -#endif nat prev_what_next; rtsBool ready_to_gc; #if defined(THREADED_RTS) @@ -268,35 +252,20 @@ schedule (Capability *initialCapability, Task *task) "### NEW SCHEDULER LOOP (task: %p, cap: %p)", task, initialCapability); + if (running_finalizers) { + errorBelch("error: a C finalizer called back into Haskell.\n" + " This was previously allowed, but is disallowed in GHC 6.10.2 and later.\n" + " To create finalizers that may call back into Haskll, use\n" + " Foreign.Concurrent.newForeignPtr instead of Foreign.newForeignPtr."); + stg_exit(EXIT_FAILURE); + } + schedulePreLoop(); // ----------------------------------------------------------- // Scheduler loop starts here: -#if defined(PARALLEL_HASKELL) -#define TERMINATION_CONDITION (!receivedFinish) -#else -#define TERMINATION_CONDITION rtsTrue -#endif - - while (TERMINATION_CONDITION) { - -#if defined(THREADED_RTS) - if (first) { - // don't yield the first time, we want a chance to run this - // thread for a bit, even if there are others banging at the - // door. - first = rtsFalse; - ASSERT_FULL_CAPABILITY_INVARIANTS(cap,task); - } else { - // Yield the capability to higher-priority tasks if necessary. - yieldCapability(&cap, task); - } -#endif - -#if defined(THREADED_RTS) - schedulePushWork(cap,task); -#endif + while (1) { // Check whether we have re-entered the RTS from Haskell without // going via suspendThread()/resumeThread (i.e. a 'safe' foreign @@ -351,7 +320,14 @@ schedule (Capability *initialCapability, Task *task) #endif /* scheduleDoGC() deletes all the threads */ cap = scheduleDoGC(cap,task,rtsFalse); - break; + + // after scheduleDoGC(), we must be shutting down. Either some + // other Capability did the final GC, or we did it above, + // either way we can fall through to the SCHED_SHUTTING_DOWN + // case now. + ASSERT(sched_state == SCHED_SHUTTING_DOWN); + // fall through + case SCHED_SHUTTING_DOWN: debugTrace(DEBUG_sched, "SCHED_SHUTTING_DOWN"); // If we are a worker, just exit. If we're a bound thread @@ -365,74 +341,14 @@ schedule (Capability *initialCapability, Task *task) barf("sched_state: %d", sched_state); } -#if defined(THREADED_RTS) - // If the run queue is empty, take a spark and turn it into a thread. - { - if (emptyRunQueue(cap)) { - StgClosure *spark; - spark = findSpark(cap); - if (spark != NULL) { - debugTrace(DEBUG_sched, - "turning spark of closure %p into a thread", - (StgClosure *)spark); - createSparkThread(cap,spark); - } - } - } -#endif // THREADED_RTS - - scheduleStartSignalHandlers(cap); - - // Only check the black holes here if we've nothing else to do. - // During normal execution, the black hole list only gets checked - // at GC time, to avoid repeatedly traversing this possibly long - // list each time around the scheduler. - if (emptyRunQueue(cap)) { scheduleCheckBlackHoles(cap); } - - scheduleCheckWakeupThreads(cap); - - scheduleCheckBlockedThreads(cap); - -#if defined(PARALLEL_HASKELL) - /* message processing and work distribution goes here */ - - /* if messages have been buffered... a NOOP in THREADED_RTS */ - scheduleSendPendingMessages(); + scheduleFindWork(cap); - /* If the run queue is empty,...*/ - if (emptyRunQueue(cap)) { - /* ...take one of our own sparks and turn it into a thread */ - scheduleActivateSpark(cap); - - /* if this did not work, try to steal a spark from someone else */ - if (emptyRunQueue(cap)) { - receivedFinish = scheduleGetRemoteWork(cap); - continue; // a new round, (hopefully) with new work - /* - in GUM, this a) sends out a FISH and returns IF no fish is - out already - b) (blocking) awaits and receives messages - - in Eden, this is only the blocking receive, as b) in GUM. - */ - } - } - - /* since we perform a blocking receive and continue otherwise, - either we never reach here or we definitely have work! */ - // from here: non-empty run queue - ASSERT(!emptyRunQueue(cap)); - - if (PacketsWaiting()) { /* now process incoming messages, if any - pending... - - CAUTION: scheduleGetRemoteWork called - above, waits for messages as well! */ - processMessages(cap, &receivedFinish); - } -#endif // PARALLEL_HASKELL + /* work pushing, currently relevant only for THREADED_RTS: + (pushes threads, wakes up idle capabilities for stealing) */ + schedulePushWork(cap,task); scheduleDetectDeadlock(cap,task); + #if defined(THREADED_RTS) cap = task->cap; // reload cap, it might have changed #endif @@ -445,12 +361,28 @@ schedule (Capability *initialCapability, Task *task) // // win32: might be here due to awaitEvent() being abandoned // as a result of a console event having been delivered. - if ( emptyRunQueue(cap) ) { + +#if defined(THREADED_RTS) + if (first) + { + // XXX: ToDo + // // 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); + } + + yield: + scheduleYield(&cap,task); + if (emptyRunQueue(cap)) continue; // look for work again +#endif + #if !defined(THREADED_RTS) && !defined(mingw32_HOST_OS) + if ( emptyRunQueue(cap) ) { ASSERT(sched_state >= SCHED_INTERRUPTING); -#endif - continue; // nothing to do } +#endif // // Get a thread to run @@ -493,6 +425,15 @@ schedule (Capability *initialCapability, Task *task) } #endif + // If we're shutting down, and this thread has not yet been + // killed, kill it now. This sometimes happens when a finalizer + // thread is created by the final GC, or a thread previously + // in a foreign call returns. + if (sched_state >= SCHED_INTERRUPTING && + !(t->what_next == ThreadComplete || t->what_next == ThreadKilled)) { + deleteThread(cap,t); + } + /* context switches are initiated by the timer signal, unless * the user specified "context switch as often as possible", with * +RTS -C0 @@ -522,6 +463,7 @@ run_thread: ASSERT_FULL_CAPABILITY_INVARIANTS(cap,task); ASSERT(t->cap == cap); + ASSERT(t->bound ? t->bound->cap == cap : 1); prev_what_next = t->what_next; @@ -548,6 +490,8 @@ run_thread: } #endif + postEvent(cap, EVENT_RUN_THREAD, t->id, 0); + switch (prev_what_next) { case ThreadKilled: @@ -596,6 +540,8 @@ run_thread: t->saved_winerror = GetLastError(); #endif + postEvent (cap, EVENT_STOP_THREAD, t->id, ret); + #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 @@ -607,7 +553,7 @@ run_thread: debugTrace(DEBUG_sched, "--<< thread %lu (%s) stopped: blocked", (unsigned long)t->id, whatNext_strs[t->what_next]); - continue; + goto yield; } #endif @@ -624,7 +570,9 @@ run_thread: schedulePostRunThread(cap,t); - t = threadStackUnderflow(task,t); + if (ret != StackOverflow) { + t = threadStackUnderflow(task,t); + } ready_to_gc = rtsFalse; @@ -674,16 +622,99 @@ schedulePreLoop(void) } /* ----------------------------------------------------------------------------- + * scheduleFindWork() + * + * Search for work to do, and handle messages from elsewhere. + * -------------------------------------------------------------------------- */ + +static void +scheduleFindWork (Capability *cap) +{ + scheduleStartSignalHandlers(cap); + + // Only check the black holes here if we've nothing else to do. + // During normal execution, the black hole list only gets checked + // at GC time, to avoid repeatedly traversing this possibly long + // list each time around the scheduler. + if (emptyRunQueue(cap)) { scheduleCheckBlackHoles(cap); } + + scheduleCheckWakeupThreads(cap); + + scheduleCheckBlockedThreads(cap); + +#if defined(THREADED_RTS) + if (emptyRunQueue(cap)) { scheduleActivateSpark(cap); } +#endif +} + +#if defined(THREADED_RTS) +STATIC_INLINE rtsBool +shouldYieldCapability (Capability *cap, Task *task) +{ + // we need to yield this capability to someone else if.. + // - another thread is initiating a GC + // - another Task is returning from a foreign call + // - the thread at the head of the run queue cannot be run + // by this Task (it is bound to another Task, or it is unbound + // and this task it bound). + return (waiting_for_gc || + cap->returning_tasks_hd != NULL || + (!emptyRunQueue(cap) && (task->tso == NULL + ? cap->run_queue_hd->bound != NULL + : cap->run_queue_hd->bound != task))); +} + +// This is the single place where a Task goes to sleep. There are +// two reasons it might need to sleep: +// - there are no threads to run +// - we need to yield this Capability to someone else +// (see shouldYieldCapability()) +// +// Careful: the scheduler loop is quite delicate. Make sure you run +// the tests in testsuite/concurrent (all ways) after modifying this, +// and also check the benchmarks in nofib/parallel for regressions. + +static void +scheduleYield (Capability **pcap, Task *task) +{ + Capability *cap = *pcap; + + // if we have work, and we don't need to give up the Capability, continue. + if (!shouldYieldCapability(cap,task) && + (!emptyRunQueue(cap) || + !emptyWakeupQueue(cap) || + blackholes_need_checking || + sched_state >= SCHED_INTERRUPTING)) + return; + + // otherwise yield (sleep), and keep yielding if necessary. + do { + yieldCapability(&cap,task); + } + while (shouldYieldCapability(cap,task)); + + // note there may still be no threads on the run queue at this + // point, the caller has to check. + + *pcap = cap; + return; +} +#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) { + /* following code not for PARALLEL_HASKELL. I kept the call general, + future GUM versions might use pushing in a distributed setup */ +#if defined(THREADED_RTS) + Capability *free_caps[n_capabilities], *cap0; nat i, n_free_caps; @@ -692,9 +723,11 @@ schedulePushWork(Capability *cap USED_IF_THREADS, // 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; + if (cap->run_queue_hd == END_TSO_QUEUE) { + if (sparkPoolSizeCap(cap) < 2) return; + } else { + if (cap->run_queue_hd->_link == END_TSO_QUEUE && + sparkPoolSizeCap(cap) < 1) return; } // First grab as many free Capabilities as we can. @@ -726,7 +759,12 @@ schedulePushWork(Capability *cap USED_IF_THREADS, StgTSO *prev, *t, *next; rtsBool pushed_to_all; - debugTrace(DEBUG_sched, "excess threads on run queue and %d free capabilities, sharing...", n_free_caps); + debugTrace(DEBUG_sched, + "cap %d: %s and %d free capabilities, sharing...", + cap->no, + (!emptyRunQueue(cap) && cap->run_queue_hd->_link != END_TSO_QUEUE)? + "excess threads on run queue":"sparks to share (>=2)", + n_free_caps); i = 0; pushed_to_all = rtsFalse; @@ -752,6 +790,9 @@ schedulePushWork(Capability *cap USED_IF_THREADS, } else { debugTrace(DEBUG_sched, "pushing thread %lu to capability %d", (unsigned long)t->id, free_caps[i]->no); appendToRunQueue(free_caps[i],t); + + postEvent (cap, EVENT_MIGRATE_THREAD, t->id, free_caps[i]->no); + if (t->bound) { t->bound->cap = free_caps[i]; } t->cap = free_caps[i]; i++; @@ -760,6 +801,9 @@ schedulePushWork(Capability *cap USED_IF_THREADS, cap->run_queue_tl = prev; } +#ifdef SPARK_PUSHING + /* JB I left this code in place, it would work but is not necessary */ + // 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) { @@ -767,24 +811,30 @@ schedulePushWork(Capability *cap USED_IF_THREADS, // i is the next free capability to push to for (; i < n_free_caps; i++) { if (emptySparkPoolCap(free_caps[i])) { - spark = findSpark(cap); + spark = tryStealSpark(cap->sparks); if (spark != NULL) { debugTrace(DEBUG_sched, "pushing spark %p to capability %d", spark, free_caps[i]->no); + + postEvent(free_caps[i], EVENT_STEAL_SPARK, t->id, cap->no); + newSpark(&(free_caps[i]->r), spark); } } } } +#endif /* SPARK_PUSHING */ // release the capabilities for (i = 0; i < n_free_caps; i++) { task->cap = free_caps[i]; - releaseCapability(free_caps[i]); + releaseAndWakeupCapability(free_caps[i]); } } task->cap = cap; // reset to point to our Capability. + +#endif /* THREADED_RTS */ + } -#endif /* ---------------------------------------------------------------------------- * Start any pending signal handlers @@ -862,8 +912,13 @@ scheduleCheckBlackHoles (Capability *cap) { ACQUIRE_LOCK(&sched_mutex); if ( blackholes_need_checking ) { - checkBlackHoles(cap); blackholes_need_checking = rtsFalse; + // important that we reset the flag *before* checking the + // blackhole queue, otherwise we could get deadlock. This + // happens as follows: we wake up a thread that + // immediately runs on another Capability, blocks on a + // blackhole, and then we reset the blackholes_need_checking flag. + checkBlackHoles(cap); } RELEASE_LOCK(&sched_mutex); } @@ -876,12 +931,6 @@ scheduleCheckBlackHoles (Capability *cap) static void scheduleDetectDeadlock (Capability *cap, Task *task) { - -#if defined(PARALLEL_HASKELL) - // ToDo: add deadlock detection in GUM (similar to THREADED_RTS) -- HWL - return; -#endif - /* * Detect deadlock: when we have no threads to run, there are no * threads blocked, waiting for I/O, or sleeping, and all the @@ -907,12 +956,11 @@ scheduleDetectDeadlock (Capability *cap, Task *task) // 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*/); + cap = scheduleDoGC (cap, task, rtsTrue/*force major GC*/); + // when force_major == rtsTrue. scheduleDoGC sets + // recent_activity to ACTIVITY_DONE_GC and turns off the timer + // signal. - recent_activity = ACTIVITY_DONE_GC; - // disable timer signals (see #1623) - stopTimer(); - if ( !emptyRunQueue(cap) ) return; #if defined(RTS_USER_SIGNALS) && !defined(THREADED_RTS) @@ -965,7 +1013,7 @@ scheduleDetectDeadlock (Capability *cap, Task *task) * ------------------------------------------------------------------------- */ #if defined(PARALLEL_HASKELL) -static StgTSO * +static void scheduleSendPendingMessages(void) { @@ -984,80 +1032,20 @@ scheduleSendPendingMessages(void) #endif /* ---------------------------------------------------------------------------- - * Activate spark threads (PARALLEL_HASKELL only) + * Activate spark threads (PARALLEL_HASKELL and THREADED_RTS) * ------------------------------------------------------------------------- */ -#if defined(PARALLEL_HASKELL) +#if defined(THREADED_RTS) static void scheduleActivateSpark(Capability *cap) { - StgClosure *spark; - -/* We only want to stay here if the run queue is empty and we want some - work. We try to turn a spark into a thread, and add it to the run - queue, from where it will be picked up in the next iteration of the - scheduler loop. -*/ - if (!emptyRunQueue(cap)) - /* In the threaded RTS, another task might have pushed a thread - on our run queue in the meantime ? But would need a lock.. */ - return; - - spark = findSpark(cap); // defined in Sparks.c - - if (spark != NULL) { - debugTrace(DEBUG_sched, - "turning spark of closure %p into a thread", - (StgClosure *)spark); - createSparkThread(cap,spark); // defined in Sparks.c + if (anySparks()) + { + createSparkThread(cap); + debugTrace(DEBUG_sched, "creating a spark thread"); } } -#endif // PARALLEL_HASKELL - -/* ---------------------------------------------------------------------------- - * Get work from a remote node (PARALLEL_HASKELL only) - * ------------------------------------------------------------------------- */ - -#if defined(PARALLEL_HASKELL) -static rtsBool -scheduleGetRemoteWork(Capability *cap) -{ -#if defined(PARALLEL_HASKELL) - rtsBool receivedFinish = rtsFalse; - - // idle() , i.e. send all buffers, wait for work - if (RtsFlags.ParFlags.BufferTime) { - IF_PAR_DEBUG(verbose, - debugBelch("...send all pending data,")); - { - nat i; - for (i=1; i<=nPEs; i++) - sendImmediately(i); // send all messages away immediately - } - } - - /* this would be the place for fishing in GUM... - - if (no-earlier-fish-around) - sendFish(choosePe()); - */ - - // Eden:just look for incoming messages (blocking receive) - IF_PAR_DEBUG(verbose, - debugBelch("...wait for incoming messages...\n")); - processMessages(cap, &receivedFinish); // blocking receive... - - - return receivedFinish; - // reenter scheduling look after having received something - -#else /* !PARALLEL_HASKELL, i.e. THREADED_RTS */ - - return rtsFalse; /* return value unused in THREADED_RTS */ - -#endif /* PARALLEL_HASKELL */ -} -#endif // PARALLEL_HASKELL +#endif // PARALLEL_HASKELL || THREADED_RTS /* ---------------------------------------------------------------------------- * After running a thread... @@ -1085,7 +1073,7 @@ schedulePostRunThread (Capability *cap, StgTSO *t) // ATOMICALLY_FRAME, aborting the (nested) // transaction, and saving the stack of any // partially-evaluated thunks on the heap. - throwToSingleThreaded_(cap, t, NULL, rtsTrue, NULL); + throwToSingleThreaded_(cap, t, NULL, rtsTrue); ASSERT(get_itbl((StgClosure *)t->sp)->type == ATOMICALLY_FRAME); } @@ -1173,7 +1161,7 @@ scheduleHandleHeapOverflow( Capability *cap, StgTSO *t ) "--<< thread %ld (%s) stopped: HeapOverflow", (long)t->id, whatNext_strs[t->what_next]); - if (cap->context_switch) { + if (cap->r.rHpLim == NULL || cap->context_switch) { // Sometimes we miss a context switch, e.g. when calling // primitives in a tight loop, MAYBE_GC() doesn't check the // context switch flag, and we end up waiting for a GC. @@ -1270,7 +1258,7 @@ scheduleHandleYield( Capability *cap, StgTSO *t, nat prev_what_next ) static void scheduleHandleThreadBlocked( StgTSO *t -#if !defined(GRAN) && !defined(DEBUG) +#if !defined(DEBUG) STG_UNUSED #endif ) @@ -1315,6 +1303,12 @@ scheduleHandleThreadFinished (Capability *cap STG_UNUSED, Task *task, StgTSO *t) debugTrace(DEBUG_sched, "--++ thread %lu (%s) finished", (unsigned long)t->id, whatNext_strs[t->what_next]); + // blocked exceptions can now complete, even if the thread was in + // blocked mode (see #2910). This unconditionally calls + // lockTSO(), which ensures that we don't miss any threads that + // are engaged in throwTo() with this thread as a target. + awakenBlockedExceptionQueue (cap, t); + // // Check whether the thread that just completed was a bound // thread, and if so return with the result. @@ -1357,7 +1351,11 @@ scheduleHandleThreadFinished (Capability *cap STG_UNUSED, Task *task, StgTSO *t) *(task->ret) = NULL; } if (sched_state >= SCHED_INTERRUPTING) { - task->stat = Interrupted; + if (heap_overflow) { + task->stat = HeapExhausted; + } else { + task->stat = Interrupted; + } } else { task->stat = Killed; } @@ -1400,11 +1398,28 @@ scheduleDoGC (Capability *cap, Task *task USED_IF_THREADS, rtsBool force_major) #ifdef THREADED_RTS /* extern static volatile StgWord waiting_for_gc; lives inside capability.c */ - rtsBool was_waiting; + rtsBool gc_type, prev_pending_gc; nat i; #endif + if (sched_state == SCHED_SHUTTING_DOWN) { + // The final GC has already been done, and the system is + // shutting down. We'll probably deadlock if we try to GC + // now. + return cap; + } + #ifdef THREADED_RTS + if (sched_state < SCHED_INTERRUPTING + && RtsFlags.ParFlags.parGcEnabled + && N >= RtsFlags.ParFlags.parGcGen + && ! oldest_gen->steps[0].mark) + { + gc_type = PENDING_GC_PAR; + } else { + gc_type = PENDING_GC_SEQ; + } + // 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. @@ -1415,39 +1430,57 @@ scheduleDoGC (Capability *cap, Task *task USED_IF_THREADS, rtsBool force_major) // actually did the GC. But it's quite hard to arrange for all // the other tasks to sleep and stay asleep. // - + /* Other capabilities are prevented from running yet more Haskell threads if waiting_for_gc is set. Tested inside yieldCapability() and releaseCapability() in Capability.c */ - was_waiting = cas(&waiting_for_gc, 0, 1); - if (was_waiting) { + prev_pending_gc = cas(&waiting_for_gc, 0, gc_type); + if (prev_pending_gc) { do { - debugTrace(DEBUG_sched, "someone else is trying to GC..."); - if (cap) yieldCapability(&cap,task); + debugTrace(DEBUG_sched, "someone else is trying to GC (%d)...", + prev_pending_gc); + ASSERT(cap); + yieldCapability(&cap,task); } while (waiting_for_gc); return cap; // NOTE: task->cap might have changed here } setContextSwitches(); - for (i=0; i < n_capabilities; i++) { - debugTrace(DEBUG_sched, "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; - waitForReturnCapability(&pcap, task); - if (pcap != &capabilities[i]) { - barf("scheduleDoGC: got the wrong capability"); - } - } + + // The final shutdown GC is always single-threaded, because it's + // possible that some of the Capabilities have no worker threads. + + if (gc_type == PENDING_GC_SEQ) + { + postEvent(cap, EVENT_REQUEST_SEQ_GC, 0, 0); + // single-threaded GC: grab all the capabilities + for (i=0; i < n_capabilities; i++) { + debugTrace(DEBUG_sched, "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; + waitForReturnCapability(&pcap, task); + if (pcap != &capabilities[i]) { + barf("scheduleDoGC: got the wrong capability"); + } + } + } } + else + { + // multi-threaded GC: make sure all the Capabilities donate one + // GC thread each. + postEvent(cap, EVENT_REQUEST_PAR_GC, 0, 0); + debugTrace(DEBUG_sched, "ready_to_gc, grabbing GC threads"); - waiting_for_gc = rtsFalse; + waitForGcThreads(cap); + } #endif // so this happens periodically: @@ -1455,41 +1488,95 @@ scheduleDoGC (Capability *cap, Task *task USED_IF_THREADS, rtsBool force_major) IF_DEBUG(scheduler, printAllThreads()); +delete_threads_and_gc: /* * 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]); + if (sched_state == SCHED_INTERRUPTING) { + deleteAllThreads(cap); sched_state = SCHED_SHUTTING_DOWN; } heap_census = scheduleNeedHeapProfile(rtsTrue); - /* 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) + postEvent(cap, EVENT_GC_START, 0, 0); debugTrace(DEBUG_sched, "doing GC"); + // reset waiting_for_gc *before* GC, so that when the GC threads + // emerge they don't immediately re-enter the GC. + waiting_for_gc = 0; + GarbageCollect(force_major || heap_census, gc_type, cap); +#else + GarbageCollect(force_major || heap_census, 0, cap); #endif - GarbageCollect(force_major || heap_census); - + postEvent(cap, EVENT_GC_END, 0, 0); + + if (recent_activity == ACTIVITY_INACTIVE && force_major) + { + // We are doing a GC because the system has been idle for a + // timeslice and we need to check for deadlock. Record the + // fact that we've done a GC and turn off the timer signal; + // it will get re-enabled if we run any threads after the GC. + recent_activity = ACTIVITY_DONE_GC; + stopTimer(); + } + else + { + // the GC might have taken long enough for the timer to set + // recent_activity = ACTIVITY_INACTIVE, but we aren't + // necessarily deadlocked: + recent_activity = ACTIVITY_YES; + } + +#if defined(THREADED_RTS) + if (gc_type == PENDING_GC_PAR) + { + releaseGCThreads(cap); + } +#endif + if (heap_census) { debugTrace(DEBUG_sched, "performing heap census"); heapCensus(); performHeapProfile = rtsFalse; } + if (heap_overflow && sched_state < SCHED_INTERRUPTING) { + // GC set the heap_overflow flag, so we should proceed with + // an orderly shutdown now. Ultimately we want the main + // thread to return to its caller with HeapExhausted, at which + // point the caller should call hs_exit(). The first step is + // to delete all the threads. + // + // Another way to do this would be to raise an exception in + // the main thread, which we really should do because it gives + // the program a chance to clean up. But how do we find the + // main thread? It should presumably be the same one that + // gets ^C exceptions, but that's all done on the Haskell side + // (GHC.TopHandler). + sched_state = SCHED_INTERRUPTING; + goto delete_threads_and_gc; + } + +#ifdef SPARKBALANCE + /* JB + Once we are all together... this would be the place to balance all + spark pools. No concurrent stealing or adding of new sparks can + occur. Should be defined in Sparks.c. */ + balanceSparkPoolsCaps(n_capabilities, capabilities); +#endif + #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 (gc_type == PENDING_GC_SEQ) { + // 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; @@ -1740,6 +1827,7 @@ suspendThread (StgRegTable *reg) task = cap->running_task; tso = cap->r.rCurrentTSO; + postEvent(cap, EVENT_STOP_THREAD, tso->id, THREAD_SUSPENDED_FOREIGN_CALL); debugTrace(DEBUG_sched, "thread %lu did a safe foreign call", (unsigned long)cap->r.rCurrentTSO->id); @@ -1764,7 +1852,7 @@ suspendThread (StgRegTable *reg) suspendTask(cap,task); cap->in_haskell = rtsFalse; - releaseCapability_(cap); + releaseCapability_(cap,rtsFalse); RELEASE_LOCK(&cap->lock); @@ -1811,10 +1899,15 @@ resumeThread (void *task_) tso = task->suspended_tso; task->suspended_tso = NULL; tso->_link = END_TSO_QUEUE; // no write barrier reqd + + postEvent(cap, EVENT_RUN_THREAD, tso->id, 0); debugTrace(DEBUG_sched, "thread %lu: re-entering RTS", (unsigned long)tso->id); if (tso->why_blocked == BlockedOnCCall) { - awakenBlockedExceptionQueue(cap,tso); + // avoid locking the TSO if we don't have to + if (tso->blocked_exceptions != END_TSO_QUEUE) { + awakenBlockedExceptionQueue(cap,tso); + } tso->flags &= ~(TSO_BLOCKEX | TSO_INTERRUPTIBLE); } @@ -1864,6 +1957,7 @@ scheduleThreadOn(Capability *cap, StgWord cpu USED_IF_THREADS, StgTSO *tso) if (cpu == cap->no) { appendToRunQueue(cap,tso); } else { + postEvent (cap, EVENT_MIGRATE_THREAD, tso->id, capabilities[cpu].no); wakeupThreadOnCapability(cap, &capabilities[cpu], tso); } #else @@ -1916,15 +2010,32 @@ workerStart(Task *task) cap = task->cap; RELEASE_LOCK(&task->lock); + if (RtsFlags.ParFlags.setAffinity) { + setThreadAffinity(cap->no, n_capabilities); + } + // set the thread-local pointer to the Task: taskEnter(task); // schedule() runs without a lock. cap = schedule(cap,task); - // On exit from schedule(), we have a Capability. - releaseCapability(cap); + // On exit from schedule(), we have a Capability, but possibly not + // the same one we started with. + + // During shutdown, the requirement is that after all the + // Capabilities are shut down, all workers that are shutting down + // have finished workerTaskStop(). This is why we hold on to + // cap->lock until we've finished workerTaskStop() below. + // + // There may be workers still involved in foreign calls; those + // will just block in waitForReturnCapability() because the + // Capability has been shut down. + // + ACQUIRE_LOCK(&cap->lock); + releaseCapability_(cap,rtsFalse); workerTaskStop(task); + RELEASE_LOCK(&cap->lock); } #endif @@ -1967,7 +2078,7 @@ initScheduler(void) initTaskManager(); -#if defined(THREADED_RTS) || defined(PARALLEL_HASKELL) +#if defined(THREADED_RTS) initSparkPools(); #endif @@ -1990,8 +2101,6 @@ initScheduler(void) } #endif - trace(TRACE_sched, "start: %d capabilities", n_capabilities); - RELEASE_LOCK(&sched_mutex); } @@ -2006,16 +2115,14 @@ exitScheduler( { Task *task = NULL; -#if defined(THREADED_RTS) - ACQUIRE_LOCK(&sched_mutex); task = newBoundTask(); - RELEASE_LOCK(&sched_mutex); -#endif // If we haven't killed all the threads yet, do it now. if (sched_state < SCHED_SHUTTING_DOWN) { sched_state = SCHED_INTERRUPTING; - scheduleDoGC(NULL,task,rtsFalse); + waitForReturnCapability(&task->cap,task); + scheduleDoGC(task->cap,task,rtsFalse); + releaseCapability(task->cap); } sched_state = SCHED_SHUTTING_DOWN; @@ -2027,20 +2134,30 @@ exitScheduler( shutdownCapability(&capabilities[i], task, wait_foreign); } boundTaskExiting(task); - stopTaskManager(); } -#else - freeCapability(&MainCapability); #endif } void freeScheduler( void ) { - freeTaskManager(); - if (n_capabilities != 1) { - stgFree(capabilities); + nat still_running; + + ACQUIRE_LOCK(&sched_mutex); + still_running = freeTaskManager(); + // We can only free the Capabilities if there are no Tasks still + // running. We might have a Task about to return from a foreign + // call into waitForReturnCapability(), for example (actually, + // this should be the *only* thing that a still-running Task can + // do at this point, and it will block waiting for the + // Capability). + if (still_running == 0) { + freeCapabilities(); + if (n_capabilities != 1) { + stgFree(capabilities); + } } + RELEASE_LOCK(&sched_mutex); #if defined(THREADED_RTS) closeMutex(&sched_mutex); #endif @@ -2058,13 +2175,15 @@ static void performGC_(rtsBool force_major) { Task *task; + // We must grab a new Task here, because the existing Task may be // associated with a particular Capability, and chained onto the // suspended_ccalling_tasks queue. - ACQUIRE_LOCK(&sched_mutex); task = newBoundTask(); - RELEASE_LOCK(&sched_mutex); - scheduleDoGC(NULL,task,force_major); + + waitForReturnCapability(&task->cap,task); + scheduleDoGC(task->cap,task,force_major); + releaseCapability(task->cap); boundTaskExiting(task); } @@ -2170,13 +2289,6 @@ threadStackOverflow(Capability *cap, StgTSO *tso) tso->sp = (P_)&(tso->stack[tso->stack_size]); tso->why_blocked = NotBlocked; - IF_PAR_DEBUG(verbose, - debugBelch("@@ threadStackOverflow of TSO %d (now at %p): stack size increased to %ld\n", - tso->id, tso, tso->stack_size); - /* If we're debugging, just print out the top of the stack */ - printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size, - tso->sp+64))); - unlockTSO(dest); unlockTSO(tso); @@ -2197,9 +2309,18 @@ threadStackUnderflow (Task *task STG_UNUSED, StgTSO *tso) tso_size_w = tso_sizeW(tso); - if (tso_size_w < MBLOCK_SIZE_W || + if (tso_size_w < MBLOCK_SIZE_W || + // TSO is less than 2 mblocks (since the first mblock is + // shorter than MBLOCK_SIZE_W) + (tso_size_w - BLOCKS_PER_MBLOCK*BLOCK_SIZE_W) % MBLOCK_SIZE_W != 0 || + // or TSO is not a whole number of megablocks (ensuring + // precondition of splitLargeBlock() below) + (tso_size_w <= round_up_to_mblocks(RtsFlags.GcFlags.initialStkSize)) || + // or TSO is smaller than the minimum stack size (rounded up) (nat)(tso->stack + tso->stack_size - tso->sp) > tso->stack_size / 4) + // or stack is using more than 1/4 of the available space { + // then do nothing return tso; } @@ -2248,7 +2369,9 @@ interruptStgRts(void) { sched_state = SCHED_INTERRUPTING; setContextSwitches(); +#if defined(THREADED_RTS) wakeUpRts(); +#endif } /* ----------------------------------------------------------------------------- @@ -2264,16 +2387,15 @@ interruptStgRts(void) will have interrupted any blocking system call in progress anyway. -------------------------------------------------------------------------- */ -void -wakeUpRts(void) -{ #if defined(THREADED_RTS) +void wakeUpRts(void) +{ // This forces the IO Manager thread to wakeup, which will // in turn ensure that some OS thread wakes up and runs the // scheduler loop, which will cause a GC and deadlock check. ioManagerWakeup(); -#endif } +#endif /* ----------------------------------------------------------------------------- * checkBlackHoles() @@ -2306,6 +2428,10 @@ checkBlackHoles (Capability *cap) prev = &blackhole_queue; t = blackhole_queue; while (t != END_TSO_QUEUE) { + if (t->what_next == ThreadRelocated) { + t = t->_link; + continue; + } ASSERT(t->why_blocked == BlockedOnBlackHole); type = get_itbl(UNTAG_CLOSURE(t->block_info.closure))->type; if (type != BLACKHOLE && type != CAF_BLACKHOLE) {