X-Git-Url: http://git.megacz.com/?p=ghc-hetmet.git;a=blobdiff_plain;f=rts%2FSchedule.c;h=1bd68c22974e142074df3310e3a52d4f163b0094;hp=49e25be3292088327c4ecd1eed0629cb29ffe01f;hb=842e9d6628a27cf1f420d53f6a5901935dc50c54;hpb=88b35c172f9434fd98b700f706074d142914a8bb diff --git a/rts/Schedule.c b/rts/Schedule.c index 49e25be..1bd68c2 100644 --- a/rts/Schedule.c +++ b/rts/Schedule.c @@ -7,11 +7,11 @@ * --------------------------------------------------------------------------*/ #include "PosixSource.h" +#define KEEP_LOCKCLOSURE #include "Rts.h" #include "SchedAPI.h" #include "RtsUtils.h" #include "RtsFlags.h" -#include "BlockAlloc.h" #include "OSThreads.h" #include "Storage.h" #include "StgRun.h" @@ -29,10 +29,8 @@ #include "ThreadLabels.h" #include "LdvProfile.h" #include "Updates.h" -#ifdef PROFILING #include "Proftimer.h" #include "ProfHeap.h" -#endif #if defined(GRAN) || defined(PARALLEL_HASKELL) # include "GranSimRts.h" # include "GranSim.h" @@ -52,6 +50,7 @@ #include "Trace.h" #include "RaiseAsync.h" #include "Threads.h" +#include "ThrIOManager.h" #ifdef HAVE_SYS_TYPES_H #include @@ -216,13 +215,11 @@ static rtsBool scheduleHandleYield( Capability *cap, StgTSO *t, static void scheduleHandleThreadBlocked( StgTSO *t ); static rtsBool scheduleHandleThreadFinished( Capability *cap, Task *task, StgTSO *t ); -static rtsBool scheduleDoHeapProfile(rtsBool ready_to_gc); +static rtsBool scheduleNeedHeapProfile(rtsBool ready_to_gc); static Capability *scheduleDoGC(Capability *cap, Task *task, - rtsBool force_major, - void (*get_roots)(evac_fn)); + rtsBool force_major); static rtsBool checkBlackHoles(Capability *cap); -static void AllRoots(evac_fn evac); static StgTSO *threadStackOverflow(Capability *cap, StgTSO *tso); @@ -421,7 +418,7 @@ schedule (Capability *initialCapability, Task *task) discardSparksCap(cap); #endif /* scheduleDoGC() deletes all the threads */ - cap = scheduleDoGC(cap,task,rtsFalse,GetRoots); + cap = scheduleDoGC(cap,task,rtsFalse); break; case SCHED_SHUTTING_DOWN: debugTrace(DEBUG_sched, "SCHED_SHUTTING_DOWN"); @@ -574,9 +571,7 @@ run_thread: debugTrace(DEBUG_sched, "-->> running thread %ld %s ...", (long)t->id, whatNext_strs[t->what_next]); -#if defined(PROFILING) startHeapProfTimer(); -#endif // Check for exceptions blocked on this thread maybePerformBlockedException (cap, t); @@ -590,11 +585,27 @@ run_thread: prev_what_next = t->what_next; errno = t->saved_errno; +#if mingw32_HOST_OS + SetLastError(t->saved_winerror); +#endif + cap->in_haskell = rtsTrue; dirtyTSO(t); - recent_activity = ACTIVITY_YES; +#if defined(THREADED_RTS) + if (recent_activity == ACTIVITY_DONE_GC) { + // ACTIVITY_DONE_GC means we turned off the timer signal to + // conserve power (see #1623). Re-enable it here. + nat prev; + prev = xchg(&recent_activity, ACTIVITY_YES); + if (prev == ACTIVITY_DONE_GC) { + startTimer(); + } + } else { + recent_activity = ACTIVITY_YES; + } +#endif switch (prev_what_next) { @@ -639,6 +650,10 @@ run_thread: // XXX: possibly bogus for SMP because this thread might already // be running again, see code below. t->saved_errno = errno; +#if mingw32_HOST_OS + // Similarly for Windows error code + t->saved_winerror = GetLastError(); +#endif #if defined(THREADED_RTS) // If ret is ThreadBlocked, and this Task is bound to the TSO that @@ -661,8 +676,8 @@ run_thread: // ---------------------------------------------------------------------- // Costs for the scheduler are assigned to CCS_SYSTEM -#if defined(PROFILING) stopHeapProfTimer(); +#if defined(PROFILING) CCCS = CCS_SYSTEM; #endif @@ -699,9 +714,8 @@ run_thread: barf("schedule: invalid thread return code %d", (int)ret); } - if (scheduleDoHeapProfile(ready_to_gc)) { ready_to_gc = rtsFalse; } - if (ready_to_gc) { - cap = scheduleDoGC(cap,task,rtsFalse,GetRoots); + if (ready_to_gc || scheduleNeedHeapProfile(ready_to_gc)) { + cap = scheduleDoGC(cap,task,rtsFalse); } } /* end of while() */ @@ -852,11 +866,12 @@ schedulePushWork(Capability *cap USED_IF_THREADS, * Start any pending signal handlers * ------------------------------------------------------------------------- */ -#if defined(RTS_USER_SIGNALS) && (!defined(THREADED_RTS) || defined(mingw32_HOST_OS)) +#if defined(RTS_USER_SIGNALS) && !defined(THREADED_RTS) static void scheduleStartSignalHandlers(Capability *cap) { - if (signals_pending()) { // safe outside the lock + if (RtsFlags.MiscFlags.install_signal_handlers && signals_pending()) { + // safe outside the lock startSignalHandlers(cap); } } @@ -968,18 +983,20 @@ 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*/, GetRoots); + cap = scheduleDoGC (cap, task, rtsTrue/*force major GC*/); recent_activity = ACTIVITY_DONE_GC; + // disable timer signals (see #1623) + stopTimer(); if ( !emptyRunQueue(cap) ) return; -#if defined(RTS_USER_SIGNALS) && (!defined(THREADED_RTS) || defined(mingw32_HOST_OS)) +#if defined(RTS_USER_SIGNALS) && !defined(THREADED_RTS) /* If we have user-installed signal handlers, then wait * for signals to arrive rather then bombing out with a * deadlock. */ - if ( anyUserHandlers() ) { + if ( RtsFlags.MiscFlags.install_signal_handlers && anyUserHandlers() ) { debugTrace(DEBUG_sched, "still deadlocked, waiting for signals..."); @@ -1617,7 +1634,16 @@ scheduleHandleHeapOverflow( Capability *cap, StgTSO *t ) } #endif - pushOnRunQueue(cap,t); + if (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. + // See #1984, and concurrent/should_run/1984 + context_switch = 0; + addToRunQueue(cap,t); + } else { + pushOnRunQueue(cap,t); + } return rtsTrue; /* actual GC is done at the end of the while loop in schedule() */ } @@ -1770,13 +1796,14 @@ scheduleHandleThreadBlocked( StgTSO *t // 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 + // ASSERT(t->why_blocked != NotBlocked); + // Not true: for example, + // - in THREADED_RTS, the thread may already have been woken + // up by another Capability. This actually happens: try + // conc023 +RTS -N2. + // - the thread may have woken itself up already, because + // threadPaused() might have raised a blocked throwTo + // exception, see maybePerformBlockedException(). #ifdef DEBUG if (traceClass(DEBUG_sched)) { @@ -1909,36 +1936,21 @@ scheduleHandleThreadFinished (Capability *cap STG_UNUSED, Task *task, StgTSO *t) } /* ----------------------------------------------------------------------------- - * Perform a heap census, if PROFILING + * Perform a heap census * -------------------------------------------------------------------------- */ static rtsBool -scheduleDoHeapProfile( rtsBool ready_to_gc STG_UNUSED ) +scheduleNeedHeapProfile( 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); - - debugTrace(DEBUG_sched, "garbage collecting before heap census"); - GarbageCollect(GetRoots, rtsTrue); - - debugTrace(DEBUG_sched, "performing heap census"); - heapCensus(); - - performHeapProfile = rtsFalse; - return rtsTrue; // true <=> we already GC'd + return rtsTrue; + } else { + return rtsFalse; } -#endif - return rtsFalse; } /* ----------------------------------------------------------------------------- @@ -1946,10 +1958,10 @@ scheduleDoHeapProfile( rtsBool ready_to_gc STG_UNUSED ) * -------------------------------------------------------------------------- */ static Capability * -scheduleDoGC (Capability *cap, Task *task USED_IF_THREADS, - rtsBool force_major, void (*get_roots)(evac_fn)) +scheduleDoGC (Capability *cap, Task *task USED_IF_THREADS, rtsBool force_major) { StgTSO *t; + rtsBool heap_census; #ifdef THREADED_RTS static volatile StgWord waiting_for_gc; rtsBool was_waiting; @@ -2057,6 +2069,8 @@ scheduleDoGC (Capability *cap, Task *task USED_IF_THREADS, deleteAllThreads(&capabilities[0]); 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 @@ -2066,8 +2080,14 @@ scheduleDoGC (Capability *cap, Task *task USED_IF_THREADS, #if defined(THREADED_RTS) debugTrace(DEBUG_sched, "doing GC"); #endif - GarbageCollect(get_roots, force_major); + GarbageCollect(force_major || heap_census); + if (heap_census) { + debugTrace(DEBUG_sched, "performing heap census"); + heapCensus(); + performHeapProfile = rtsFalse; + } + #if defined(THREADED_RTS) // release our stash of capabilities. for (i = 0; i < n_capabilities; i++) { @@ -2101,7 +2121,7 @@ scheduleDoGC (Capability *cap, Task *task USED_IF_THREADS, * Singleton fork(). Do not copy any running threads. * ------------------------------------------------------------------------- */ -StgInt +pid_t forkProcess(HsStablePtr *entry #ifndef FORKPROCESS_PRIMOP_SUPPORTED STG_UNUSED @@ -2126,16 +2146,34 @@ forkProcess(HsStablePtr *entry // ToDo: for SMP, we should probably acquire *all* the capabilities cap = rts_lock(); + // no funny business: hold locks while we fork, otherwise if some + // other thread is holding a lock when the fork happens, the data + // structure protected by the lock will forever be in an + // inconsistent state in the child. See also #1391. + ACQUIRE_LOCK(&sched_mutex); + ACQUIRE_LOCK(&cap->lock); + ACQUIRE_LOCK(&cap->running_task->lock); + pid = fork(); if (pid) { // parent + RELEASE_LOCK(&sched_mutex); + RELEASE_LOCK(&cap->lock); + RELEASE_LOCK(&cap->running_task->lock); + // just return the pid rts_unlock(cap); return pid; } else { // child +#if defined(THREADED_RTS) + initMutex(&sched_mutex); + initMutex(&cap->lock); + initMutex(&cap->running_task->lock); +#endif + // 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 @@ -2173,6 +2211,9 @@ forkProcess(HsStablePtr *entry ACQUIRE_LOCK(&sched_mutex); for (task = all_tasks; task != NULL; task=task->all_link) { if (task != cap->running_task) { +#if defined(THREADED_RTS) + initMutex(&task->lock); // see #1391 +#endif discardTask(task); } } @@ -2186,6 +2227,11 @@ forkProcess(HsStablePtr *entry cap->returning_tasks_tl = NULL; #endif + // On Unix, all timers are reset in the child, so we need to start + // the timer again. + initTimer(); + startTimer(); + cap = rts_evalStableIO(cap, entry, NULL); // run the action rts_checkSchedStatus("forkProcess",cap); @@ -2282,9 +2328,17 @@ void * suspendThread (StgRegTable *reg) { Capability *cap; - int saved_errno = errno; + int saved_errno; StgTSO *tso; Task *task; +#if mingw32_HOST_OS + StgWord32 saved_winerror; +#endif + + saved_errno = errno; +#if mingw32_HOST_OS + saved_winerror = GetLastError(); +#endif /* assume that *reg is a pointer to the StgRegTable part of a Capability. */ @@ -2329,6 +2383,9 @@ suspendThread (StgRegTable *reg) #endif errno = saved_errno; +#if mingw32_HOST_OS + SetLastError(saved_winerror); +#endif return task; } @@ -2337,8 +2394,16 @@ resumeThread (void *task_) { StgTSO *tso; Capability *cap; - int saved_errno = errno; Task *task = task_; + int saved_errno; +#if mingw32_HOST_OS + StgWord32 saved_winerror; +#endif + + saved_errno = errno; +#if mingw32_HOST_OS + saved_winerror = GetLastError(); +#endif cap = task->cap; // Wait for permission to re-enter the RTS with the result. @@ -2366,6 +2431,9 @@ resumeThread (void *task_) cap->r.rCurrentTSO = tso; cap->in_haskell = rtsTrue; errno = saved_errno; +#if mingw32_HOST_OS + SetLastError(saved_winerror); +#endif /* We might have GC'd, mark the TSO dirty again */ dirtyTSO(tso); @@ -2508,10 +2576,8 @@ initScheduler(void) context_switch = 0; sched_state = SCHED_RUNNING; + recent_activity = ACTIVITY_YES; - RtsFlags.ConcFlags.ctxtSwitchTicks = - RtsFlags.ConcFlags.ctxtSwitchTime / TICK_MILLISECS; - #if defined(THREADED_RTS) /* Initialise the mutex and condition variables used by * the scheduler. */ @@ -2557,7 +2623,13 @@ initScheduler(void) } void -exitScheduler( void ) +exitScheduler( + rtsBool wait_foreign +#if !defined(THREADED_RTS) + __attribute__((unused)) +#endif +) + /* see Capability.c, shutdownCapability() */ { Task *task = NULL; @@ -2570,7 +2642,7 @@ exitScheduler( void ) // 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,GetRoots); + scheduleDoGC(NULL,task,rtsFalse); } sched_state = SCHED_SHUTTING_DOWN; @@ -2579,11 +2651,24 @@ exitScheduler( void ) nat i; for (i = 0; i < n_capabilities; i++) { - shutdownCapability(&capabilities[i], task); + shutdownCapability(&capabilities[i], task, wait_foreign); } boundTaskExiting(task); stopTaskManager(); } +#else + freeCapability(&MainCapability); +#endif +} + +void +freeScheduler( void ) +{ + freeTaskManager(); + if (n_capabilities != 1) { + stgFree(capabilities); + } +#if defined(THREADED_RTS) closeMutex(&sched_mutex); #endif } @@ -2663,7 +2748,9 @@ GetRoots( evac_fn evac ) #if defined(RTS_USER_SIGNALS) // mark the signal handlers (signals should be already blocked) - markSignalHandlers(evac); + if (RtsFlags.MiscFlags.install_signal_handlers) { + markSignalHandlers(evac); + } #endif } @@ -2673,17 +2760,10 @@ GetRoots( evac_fn evac ) This is the interface to the garbage collector from Haskell land. We provide this so that external C code can allocate and garbage collect when called from Haskell via _ccall_GC. - - It might be useful to provide an interface whereby the programmer - can specify more roots (ToDo). - - This needs to be protected by the GC condition variable above. KH. -------------------------------------------------------------------------- */ -static void (*extra_roots)(evac_fn); - static void -performGC_(rtsBool force_major, void (*get_roots)(evac_fn)) +performGC_(rtsBool force_major) { Task *task; // We must grab a new Task here, because the existing Task may be @@ -2692,34 +2772,20 @@ performGC_(rtsBool force_major, void (*get_roots)(evac_fn)) ACQUIRE_LOCK(&sched_mutex); task = newBoundTask(); RELEASE_LOCK(&sched_mutex); - scheduleDoGC(NULL,task,force_major, get_roots); + scheduleDoGC(NULL,task,force_major); boundTaskExiting(task); } void performGC(void) { - performGC_(rtsFalse, GetRoots); + performGC_(rtsFalse); } void performMajorGC(void) { - performGC_(rtsTrue, GetRoots); -} - -static void -AllRoots(evac_fn evac) -{ - GetRoots(evac); // the scheduler's roots - extra_roots(evac); // the user's roots -} - -void -performGCWithRoots(void (*get_roots)(evac_fn)) -{ - extra_roots = get_roots; - performGC_(rtsFalse, AllRoots); + performGC_(rtsTrue); } /* ----------------------------------------------------------------------------- @@ -2745,7 +2811,12 @@ threadStackOverflow(Capability *cap, StgTSO *tso) // while we are moving the TSO: lockClosure((StgClosure *)tso); - if (tso->stack_size >= tso->max_stack_size) { + if (tso->stack_size >= tso->max_stack_size && !(tso->flags & TSO_BLOCKEX)) { + // NB. never raise a StackOverflow exception if the thread is + // inside Control.Exceptino.block. It is impractical to protect + // against stack overflow exceptions, since virtually anything + // can raise one (even 'catch'), so this is the only sensible + // thing to do here. See bug #767. debugTrace(DEBUG_gc, "threadStackOverflow of TSO %ld (%p): stack too large (now %ld; max is %ld)", @@ -2775,7 +2846,7 @@ threadStackOverflow(Capability *cap, StgTSO *tso) "increasing stack size from %ld words to %d.", (long)tso->stack_size, new_stack_size); - dest = (StgTSO *)allocate(new_tso_size); + dest = (StgTSO *)allocateLocal(cap,new_tso_size); TICK_ALLOC_TSO(new_stack_size,0); /* copy the TSO block and the old stack into the new area */ @@ -2848,17 +2919,10 @@ void wakeUpRts(void) { #if defined(THREADED_RTS) -#if !defined(mingw32_HOST_OS) // 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(); -#else - // On Windows this might be safe enough, because we aren't - // in a signal handler. Later we should use the IO Manager, - // though. - prodOneCapability(); -#endif #endif } @@ -2894,7 +2958,7 @@ checkBlackHoles (Capability *cap) t = blackhole_queue; while (t != END_TSO_QUEUE) { ASSERT(t->why_blocked == BlockedOnBlackHole); - type = get_itbl(t->block_info.closure)->type; + type = get_itbl(UNTAG_CLOSURE(t->block_info.closure))->type; if (type != BLACKHOLE && type != CAF_BLACKHOLE) { IF_DEBUG(sanity,checkTSO(t)); t = unblockOne(cap, t); @@ -3042,8 +3106,9 @@ raiseExceptionHelper (StgRegTable *reg, StgTSO *tso, StgClosure *exception) This should either be a CATCH_RETRY_FRAME (if the retry# is within an orElse#) or should be a ATOMICALLY_FRAME (if the retry# reaches the top level). - We skip CATCH_STM_FRAMEs because retries are not considered to be exceptions, - despite the similar implementation. + We skip CATCH_STM_FRAMEs (aborting and rolling back the nested tx that they + create) because retries are not considered to be exceptions, despite the + similar implementation. We should not expect to see CATCH_FRAME or STOP_FRAME because those should not be created within memory transactions. @@ -3063,7 +3128,7 @@ findRetryFrameHelper (StgTSO *tso) case ATOMICALLY_FRAME: debugTrace(DEBUG_stm, - "found ATOMICALLY_FRAME at %p during retrry", p); + "found ATOMICALLY_FRAME at %p during retry", p); tso->sp = p; return ATOMICALLY_FRAME; @@ -3073,7 +3138,20 @@ findRetryFrameHelper (StgTSO *tso) tso->sp = p; return CATCH_RETRY_FRAME; - case CATCH_STM_FRAME: + case CATCH_STM_FRAME: { + StgTRecHeader *trec = tso -> trec; + StgTRecHeader *outer = stmGetEnclosingTRec(trec); + debugTrace(DEBUG_stm, + "found CATCH_STM_FRAME at %p during retry", p); + debugTrace(DEBUG_stm, "trec=%p outer=%p", trec, outer); + stmAbortTransaction(tso -> cap, trec); + stmFreeAbortedTRec(tso -> cap, trec); + tso -> trec = outer; + p = next; + continue; + } + + default: ASSERT(info->i.type != CATCH_FRAME); ASSERT(info->i.type != STOP_FRAME);