X-Git-Url: http://git.megacz.com/?a=blobdiff_plain;f=ghc%2Frts%2FSchedule.c;h=ecd47aa41b9d9edf7729dfa1cf71c1ee3f6e609a;hb=153b9cb9b11e05c4edb1b6bc0a7b972660e41f70;hp=262e4c454157564f14211835572d4ab9902517c0;hpb=00912bda4c438c9f6a526e6c2d79cc581e7fbc6a;p=ghc-hetmet.git diff --git a/ghc/rts/Schedule.c b/ghc/rts/Schedule.c index 262e4c4..ecd47aa 100644 --- a/ghc/rts/Schedule.c +++ b/ghc/rts/Schedule.c @@ -1,7 +1,6 @@ /* --------------------------------------------------------------------------- - * $Id: Schedule.c,v 1.194 2004/03/13 00:56:45 sof Exp $ * - * (c) The GHC Team, 1998-2003 + * (c) The GHC Team, 1998-2004 * * Scheduler * @@ -42,9 +41,9 @@ #include "SchedAPI.h" #include "RtsUtils.h" #include "RtsFlags.h" +#include "BlockAlloc.h" #include "Storage.h" #include "StgRun.h" -#include "StgStartup.h" #include "Hooks.h" #define COMPILING_SCHEDULER #include "Schedule.h" @@ -56,9 +55,12 @@ #include "Signals.h" #include "Sanity.h" #include "Stats.h" +#include "STM.h" #include "Timer.h" #include "Prelude.h" #include "ThreadLabels.h" +#include "LdvProfile.h" +#include "Updates.h" #ifdef PROFILING #include "Proftimer.h" #include "ProfHeap.h" @@ -161,7 +163,7 @@ static StgTSO *threadStackOverflow(StgTSO *tso); */ /* flag set by signal handler to precipitate a context switch */ -nat context_switch = 0; +int context_switch = 0; /* if this flag is set as well, give up execution */ rtsBool interrupted = rtsFalse; @@ -215,7 +217,11 @@ void addToBlockedQueue ( StgTSO *tso ); static void schedule ( StgMainThread *mainThread, Capability *initialCapability ); void interruptStgRts ( void ); +#if !defined(PAR) && !defined(RTS_SUPPORTS_THREADS) static void detectBlackHoles ( void ); +#endif + +static void raiseAsync_(StgTSO *tso, StgClosure *exception, rtsBool stop_at_atomically); #if defined(RTS_SUPPORTS_THREADS) /* ToDo: carefully document the invariants that go together @@ -234,6 +240,7 @@ rtsBool emitSchedule = rtsTrue; #if DEBUG static char *whatNext_strs[] = { + "(unknown)", "ThreadRunGHC", "ThreadInterpret", "ThreadKilled", @@ -276,7 +283,10 @@ startSchedulerTaskIfNecessary(void) // just because the last one hasn't yet reached the // "waiting for capability" state startingWorkerThread = rtsTrue; - startTask(taskStart); + if(!startTask(taskStart)) + { + startingWorkerThread = rtsFalse; + } } } } @@ -337,7 +347,7 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, # endif #endif rtsBool was_interrupted = rtsFalse; - StgTSOWhatNext prev_what_next; + nat prev_what_next; // Pre-condition: sched_mutex is held. // We might have a capability, passed in as initialCapability. @@ -366,7 +376,7 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, CurrentTSO, (StgClosure*)NULL, (rtsSpark*)NULL); IF_DEBUG(gran, - fprintf(stderr, "GRAN: Init CurrentTSO (in schedule) = %p\n", CurrentTSO); + debugBelch("GRAN: Init CurrentTSO (in schedule) = %p\n", CurrentTSO); G_TSO(CurrentTSO, 5)); if (RtsFlags.GranFlags.Light) { @@ -423,7 +433,7 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, #if defined(RTS_SUPPORTS_THREADS) // In the threaded RTS, deadlock detection doesn't work, // so just exit right away. - prog_belch("interrupted"); + errorBelch("interrupted"); releaseCapability(cap); RELEASE_LOCK(&sched_mutex); shutdownHaskellAndExit(EXIT_SUCCESS); @@ -446,13 +456,13 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, // 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 ( !EMPTY_QUEUE(blocked_queue_hd) || !EMPTY_QUEUE(sleeping_queue) ) + { #if defined(RTS_SUPPORTS_THREADS) - || EMPTY_RUN_QUEUE() + // We shouldn't be here... + barf("schedule: awaitEvent() in threaded RTS"); #endif - ) - { - awaitEvent( EMPTY_RUN_QUEUE() ); + awaitEvent( EMPTY_RUN_QUEUE() ); } // we can be interrupted while waiting for I/O... if (interrupted) continue; @@ -472,18 +482,13 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, if ( EMPTY_THREAD_QUEUES() ) { IF_DEBUG(scheduler, sched_belch("deadlocked, forcing major GC...")); + // Garbage collection can release some new threads due to // either (a) finalizers or (b) threads resurrected because - // they are about to be send BlockedOnDeadMVar. Any threads - // thus released will be immediately runnable. + // they are unreachable and will therefore be sent an + // exception. Any threads thus released will be immediately + // runnable. GarbageCollect(GetRoots,rtsTrue); - - if ( !EMPTY_RUN_QUEUE() ) { goto not_deadlocked; } - - IF_DEBUG(scheduler, - sched_belch("still deadlocked, checking for black holes...")); - detectBlackHoles(); - if ( !EMPTY_RUN_QUEUE() ) { goto not_deadlocked; } #if defined(RTS_USER_SIGNALS) @@ -537,7 +542,10 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, // ToDo: add deadlock detection in GUM (similar to SMP) -- HWL #endif -#if defined(RTS_SUPPORTS_THREADS) +#if defined(RTS_SUPPORTS_THREADS) || defined(mingw32_HOST_OS) + /* win32: might be back here due to awaitEvent() being abandoned + * as a result of a console event having been delivered. + */ if ( EMPTY_RUN_QUEUE() ) { continue; // nothing to do } @@ -556,13 +564,13 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, if (!RtsFlags.GranFlags.Light) handleIdlePEs(); - IF_DEBUG(gran, fprintf(stderr, "GRAN: switch by event-type\n")); + IF_DEBUG(gran, debugBelch("GRAN: switch by event-type\n")); /* main event dispatcher in GranSim */ switch (event->evttype) { /* Should just be continuing execution */ case ContinueThread: - IF_DEBUG(gran, fprintf(stderr, "GRAN: doing ContinueThread\n")); + IF_DEBUG(gran, debugBelch("GRAN: doing ContinueThread\n")); /* ToDo: check assertion ASSERT(run_queue_hd != (StgTSO*)NULL && run_queue_hd != END_TSO_QUEUE); @@ -570,25 +578,25 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, /* Ignore ContinueThreads for fetching threads (if synchr comm) */ if (!RtsFlags.GranFlags.DoAsyncFetch && procStatus[CurrentProc]==Fetching) { - belch("ghuH: Spurious ContinueThread while Fetching ignored; TSO %d (%p) [PE %d]", + debugBelch("ghuH: Spurious ContinueThread while Fetching ignored; TSO %d (%p) [PE %d]\n", CurrentTSO->id, CurrentTSO, CurrentProc); goto next_thread; } /* Ignore ContinueThreads for completed threads */ if (CurrentTSO->what_next == ThreadComplete) { - belch("ghuH: found a ContinueThread event for completed thread %d (%p) [PE %d] (ignoring ContinueThread)", + debugBelch("ghuH: found a ContinueThread event for completed thread %d (%p) [PE %d] (ignoring ContinueThread)\n", CurrentTSO->id, CurrentTSO, CurrentProc); goto next_thread; } /* Ignore ContinueThreads for threads that are being migrated */ if (PROCS(CurrentTSO)==Nowhere) { - belch("ghuH: trying to run the migrating TSO %d (%p) [PE %d] (ignoring ContinueThread)", + debugBelch("ghuH: trying to run the migrating TSO %d (%p) [PE %d] (ignoring ContinueThread)\n", CurrentTSO->id, CurrentTSO, CurrentProc); goto next_thread; } /* The thread should be at the beginning of the run queue */ if (CurrentTSO!=run_queue_hds[CurrentProc]) { - belch("ghuH: TSO %d (%p) [PE %d] is not at the start of the run_queue when doing a ContinueThread", + debugBelch("ghuH: TSO %d (%p) [PE %d] is not at the start of the run_queue when doing a ContinueThread\n", CurrentTSO->id, CurrentTSO, CurrentProc); break; // run the thread anyway } @@ -645,14 +653,14 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, /* This point was scheduler_loop in the old RTS */ - IF_DEBUG(gran, belch("GRAN: after main switch")); + IF_DEBUG(gran, debugBelch("GRAN: after main switch\n")); TimeOfLastEvent = CurrentTime[CurrentProc]; TimeOfNextEvent = get_time_of_next_event(); IgnoreEvents=(TimeOfNextEvent==0); // HWL HACK // CurrentTSO = ThreadQueueHd; - IF_DEBUG(gran, belch("GRAN: time of next event is: %ld", + IF_DEBUG(gran, debugBelch("GRAN: time of next event is: %ld\n", TimeOfNextEvent)); if (RtsFlags.GranFlags.Light) @@ -661,7 +669,7 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, EndOfTimeSlice = CurrentTime[CurrentProc]+RtsFlags.GranFlags.time_slice; IF_DEBUG(gran, - belch("GRAN: end of time-slice is %#lx", EndOfTimeSlice)); + debugBelch("GRAN: end of time-slice is %#lx\n", EndOfTimeSlice)); /* in a GranSim setup the TSO stays on the run queue */ t = CurrentTSO; @@ -669,7 +677,7 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, POP_RUN_QUEUE(t); // take_off_run_queue(t); IF_DEBUG(gran, - fprintf(stderr, "GRAN: About to run current thread, which is\n"); + debugBelch("GRAN: About to run current thread, which is\n"); G_TSO(t,5)); context_switch = 0; // turned on via GranYield, checking events and time slice @@ -705,16 +713,16 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, if (spark != (rtsSpark) NULL) { tso = activateSpark(spark); /* turn the spark into a thread */ IF_PAR_DEBUG(schedule, - belch("==== schedule: Created TSO %d (%p); %d threads active", + debugBelch("==== schedule: Created TSO %d (%p); %d threads active\n", tso->id, tso, advisory_thread_count)); if (tso==END_TSO_QUEUE) { /* failed to activate spark->back to loop */ - belch("==^^ failed to activate spark"); + debugBelch("==^^ failed to activate spark\n"); goto next_thread; } /* otherwise fall through & pick-up new tso */ } else { IF_PAR_DEBUG(verbose, - belch("==^^ no local sparks (spark pool contains only NFs: %d)", + debugBelch("==^^ no local sparks (spark pool contains only NFs: %d)\n", spark_queue_len(pool))); goto next_thread; } @@ -735,12 +743,12 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, */ TIME now = msTime() /*CURRENT_TIME*/; IF_PAR_DEBUG(verbose, - belch("-- now=%ld", now)); + debugBelch("-- now=%ld\n", now)); IF_PAR_DEBUG(verbose, if (outstandingFishes < RtsFlags.ParFlags.maxFishes && (last_fish_arrived_at!=0 && last_fish_arrived_at+RtsFlags.ParFlags.fishDelay > now)) { - belch("--$$ delaying FISH until %ld (last fish %ld, delay %ld, now %ld)", + debugBelch("--$$ delaying FISH until %ld (last fish %ld, delay %ld, now %ld)\n", last_fish_arrived_at+RtsFlags.ParFlags.fishDelay, last_fish_arrived_at, RtsFlags.ParFlags.fishDelay, now); @@ -786,7 +794,7 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, pool = &(MainRegTable.rSparks); // generalise to cap = &MainRegTable IF_DEBUG(scheduler, - belch("--=^ %d threads, %d sparks on [%#x]", + debugBelch("--=^ %d threads, %d sparks on [%#x]\n", run_queue_len(), spark_queue_len(pool), CURRENT_PROC)); # if 1 @@ -876,15 +884,13 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, || blocked_queue_hd != END_TSO_QUEUE || sleeping_queue != END_TSO_QUEUE))) context_switch = 1; - else - context_switch = 0; run_thread: RELEASE_LOCK(&sched_mutex); IF_DEBUG(scheduler, sched_belch("-->> running thread %ld %s ...", - t->id, whatNext_strs[t->what_next])); + (long)t->id, whatNext_strs[t->what_next])); #ifdef PROFILING startHeapProfTimer(); @@ -894,23 +900,35 @@ run_thread: /* Run the current thread */ prev_what_next = t->what_next; + + errno = t->saved_errno; + switch (prev_what_next) { + case ThreadKilled: case ThreadComplete: /* Thread already finished, return to scheduler. */ ret = ThreadFinished; break; + case ThreadRunGHC: - errno = t->saved_errno; ret = StgRun((StgFunPtr) stg_returnToStackTop, &cap->r); - t->saved_errno = errno; break; + case ThreadInterpret: ret = interpretBCO(cap); break; + default: barf("schedule: invalid what_next field"); } + + // The TSO might have moved, so find the new location: + t = cap->r.rCurrentTSO; + + // And save the current errno in this thread. + t->saved_errno = errno; + /* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */ /* Costs for the scheduler are assigned to CCS_SYSTEM */ @@ -922,11 +940,10 @@ run_thread: ACQUIRE_LOCK(&sched_mutex); #ifdef RTS_SUPPORTS_THREADS - IF_DEBUG(scheduler,fprintf(stderr,"sched (task %p): ", osThreadId());); + IF_DEBUG(scheduler,debugBelch("sched (task %p): ", osThreadId());); #elif !defined(GRAN) && !defined(PAR) - IF_DEBUG(scheduler,fprintf(stderr,"sched: ");); + IF_DEBUG(scheduler,debugBelch("sched: ");); #endif - t = cap->r.rCurrentTSO; #if defined(PAR) /* HACK 675: if the last thread didn't yield, make sure to print a @@ -946,15 +963,15 @@ run_thread: #endif // did the task ask for a large block? - if (cap->r.rHpAlloc > BLOCK_SIZE_W) { + if (cap->r.rHpAlloc > BLOCK_SIZE) { // if so, get one and push it on the front of the nursery. bdescr *bd; nat blocks; - blocks = (nat)BLOCK_ROUND_UP(cap->r.rHpAlloc * sizeof(W_)) / BLOCK_SIZE; + blocks = (nat)BLOCK_ROUND_UP(cap->r.rHpAlloc) / BLOCK_SIZE; - IF_DEBUG(scheduler,belch("--<< thread %ld (%s) stopped: requesting a large block (size %d)", - t->id, whatNext_strs[t->what_next], blocks)); + IF_DEBUG(scheduler,debugBelch("--<< thread %ld (%s) stopped: requesting a large block (size %d)\n", + (long)t->id, whatNext_strs[t->what_next], blocks)); // don't do this if it would push us over the // alloc_blocks_lim limit; we'll GC first. @@ -1013,8 +1030,8 @@ run_thread: * maybe set context_switch and wait till they all pile in, * then have them wait on a GC condition variable. */ - IF_DEBUG(scheduler,belch("--<< thread %ld (%s) stopped: HeapOverflow", - t->id, whatNext_strs[t->what_next])); + IF_DEBUG(scheduler,debugBelch("--<< thread %ld (%s) stopped: HeapOverflow\n", + (long)t->id, whatNext_strs[t->what_next])); threadPaused(t); #if defined(GRAN) ASSERT(!is_on_queue(t,CurrentProc)); @@ -1045,8 +1062,8 @@ run_thread: // DumpGranEvent(GR_DESCHEDULE, t); globalParStats.tot_stackover++; #endif - IF_DEBUG(scheduler,belch("--<< thread %ld (%s) stopped, StackOverflow", - t->id, whatNext_strs[t->what_next])); + IF_DEBUG(scheduler,debugBelch("--<< thread %ld (%s) stopped, StackOverflow\n", + (long)t->id, whatNext_strs[t->what_next])); /* just adjust the stack for this thread, then pop it back * on the run queue. */ @@ -1062,12 +1079,19 @@ run_thread: if (t->main != NULL) { t->main->tso = new_t; } - threadPaused(new_t); PUSH_ON_RUN_QUEUE(new_t); } break; case ThreadYielding: + // Reset the context switch flag. We don't do this just before + // running the thread, because that would mean we would lose ticks + // during GC, which can lead to unfair scheduling (a thread hogs + // the CPU because the tick always arrives during GC). This way + // penalises threads that do a lot of allocation, but that seems + // better than the alternative. + context_switch = 0; + #if defined(GRAN) IF_DEBUG(gran, DumpGranEvent(GR_DESCHEDULE, t)); @@ -1084,16 +1108,16 @@ run_thread: */ IF_DEBUG(scheduler, if (t->what_next != prev_what_next) { - belch("--<< thread %ld (%s) stopped to switch evaluators", - t->id, whatNext_strs[t->what_next]); + debugBelch("--<< thread %ld (%s) stopped to switch evaluators\n", + (long)t->id, whatNext_strs[t->what_next]); } else { - belch("--<< thread %ld (%s) stopped, yielding", - t->id, whatNext_strs[t->what_next]); + debugBelch("--<< thread %ld (%s) stopped, yielding\n", + (long)t->id, whatNext_strs[t->what_next]); } ); IF_DEBUG(sanity, - //belch("&& Doing sanity check on yielding TSO %ld.", t->id); + //debugBelch("&& Doing sanity check on yielding TSO %ld.", t->id); checkTSO(t)); ASSERT(t->link == END_TSO_QUEUE); @@ -1110,7 +1134,7 @@ run_thread: ASSERT(!is_on_queue(t,CurrentProc)); IF_DEBUG(sanity, - //belch("&& Doing sanity check on all ThreadQueues (and their TSOs)."); + //debugBelch("&& Doing sanity check on all ThreadQueues (and their TSOs)."); checkThreadQsSanity(rtsTrue)); #endif @@ -1133,7 +1157,7 @@ run_thread: ContinueThread, t, (StgClosure*)NULL, (rtsSpark*)NULL); IF_GRAN_DEBUG(bq, - belch("GRAN: eventq and runnableq after adding yielded thread to queue again:"); + debugBelch("GRAN: eventq and runnableq after adding yielded thread to queue again:\n"); G_EVENTQ(0); G_CURR_THREADQ(0)); #endif /* GRAN */ @@ -1142,7 +1166,7 @@ run_thread: case ThreadBlocked: #if defined(GRAN) IF_DEBUG(scheduler, - belch("--<< thread %ld (%p; %s) stopped, blocking on node %p [PE %d] with BQ: ", + debugBelch("--<< thread %ld (%p; %s) stopped, blocking on node %p [PE %d] with BQ: \n", t->id, t, whatNext_strs[t->what_next], t->block_info.closure, (t->block_info.closure==(StgClosure*)NULL ? 99 : where_is(t->block_info.closure))); if (t->block_info.closure!=(StgClosure*)NULL) print_bq(t->block_info.closure)); @@ -1162,7 +1186,7 @@ run_thread: */ #elif defined(PAR) IF_DEBUG(scheduler, - belch("--<< thread %ld (%p; %s) stopped, blocking on node %p with BQ: ", + 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, @@ -1181,12 +1205,12 @@ run_thread: * previously, or it's blocked on an MVar or Blackhole, in which * case it'll be on the relevant queue already. */ + ASSERT(t->why_blocked != NotBlocked); IF_DEBUG(scheduler, - fprintf(stderr, "--<< thread %d (%s) stopped: ", + debugBelch("--<< thread %d (%s) stopped: ", t->id, whatNext_strs[t->what_next]); printThreadBlockage(t); - fprintf(stderr, "\n")); - fflush(stderr); + debugBelch("\n")); /* Only for dumping event to log file ToDo: do I need this in GranSim, too? @@ -1205,7 +1229,7 @@ run_thread: /* We also end up here if the thread kills itself with an * uncaught exception, see Exception.hc. */ - IF_DEBUG(scheduler,belch("--++ thread %d (%s) finished", + IF_DEBUG(scheduler,debugBelch("--++ thread %d (%s) finished\n", t->id, whatNext_strs[t->what_next])); #if defined(GRAN) endThread(t, CurrentProc); // clean-up the thread @@ -1309,6 +1333,26 @@ run_thread: #endif if (ready_to_gc) { + /* 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. */ + for (t = all_threads; t != END_TSO_QUEUE; t = t -> link) { + if (t -> what_next != ThreadRelocated && t -> trec != NO_TREC && t -> why_blocked == NotBlocked) { + if (!stmValidateTransaction (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_(t, NULL, rtsTrue); + +#ifdef REG_R1 + ASSERT(get_itbl((StgClosure *)t->sp)->type == ATOMICALLY_FRAME); +#endif + } + } + } + /* 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 @@ -1325,7 +1369,7 @@ run_thread: ContinueThread, t, (StgClosure*)NULL, (rtsSpark*)NULL); IF_GRAN_DEBUG(bq, - fprintf(stderr, "GRAN: eventq and runnableq after Garbage collection:\n"); + debugBelch("GRAN: eventq and runnableq after Garbage collection:\n\n"); G_EVENTQ(0); G_CURR_THREADQ(0)); #endif /* GRAN */ @@ -1345,7 +1389,7 @@ run_thread: } /* end of while(1) */ IF_PAR_DEBUG(verbose, - belch("== Leaving schedule() after having received Finish")); + debugBelch("== Leaving schedule() after having received Finish\n")); } /* --------------------------------------------------------------------------- @@ -1380,7 +1424,7 @@ isThreadBound(StgTSO* tso USED_IN_THREADED_RTS) * Singleton fork(). Do not copy any running threads. * ------------------------------------------------------------------------- */ -#ifndef mingw32_TARGET_OS +#ifndef mingw32_HOST_OS #define FORKPROCESS_PRIMOP_SUPPORTED #endif @@ -1435,12 +1479,6 @@ forkProcess(HsStablePtr *entry stgFree(m); } -# ifdef RTS_SUPPORTS_THREADS - resetTaskManagerAfterFork(); // tell startTask() and friends that - startingWorkerThread = rtsFalse; // we have no worker threads any more - resetWorkerWakeupPipeAfterFork(); -# endif - rc = rts_evalStableIO(entry, NULL); // run the action rts_checkSchedStatus("forkProcess",rc); @@ -1473,9 +1511,15 @@ deleteAllThreads ( void ) next = t->global_link; deleteThread(t); } - run_queue_hd = run_queue_tl = END_TSO_QUEUE; - blocked_queue_hd = blocked_queue_tl = END_TSO_QUEUE; - sleeping_queue = END_TSO_QUEUE; + + // 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. + + ASSERT(blocked_queue_hd == END_TSO_QUEUE); + ASSERT(sleeping_queue == END_TSO_QUEUE); } /* startThread and insertThread are now in GranSim.c -- HWL */ @@ -1497,12 +1541,7 @@ deleteAllThreads ( void ) * ------------------------------------------------------------------------- */ StgInt -suspendThread( StgRegTable *reg, - rtsBool concCall -#if !defined(DEBUG) - STG_UNUSED -#endif - ) +suspendThread( StgRegTable *reg ) { nat tok; Capability *cap; @@ -1516,7 +1555,7 @@ suspendThread( StgRegTable *reg, ACQUIRE_LOCK(&sched_mutex); IF_DEBUG(scheduler, - sched_belch("thread %d did a _ccall_gc (is_concurrent: %d)", cap->r.rCurrentTSO->id,concCall)); + sched_belch("thread %d did a _ccall_gc", cap->r.rCurrentTSO->id)); // XXX this might not be necessary --SDM cap->r.rCurrentTSO->what_next = ThreadRunGHC; @@ -1545,8 +1584,6 @@ suspendThread( StgRegTable *reg, IF_DEBUG(scheduler, sched_belch("worker (token %d): leaving RTS", tok)); #endif - /* Other threads _might_ be available for execution; signal this */ - THREAD_RUNNABLE(); RELEASE_LOCK(&sched_mutex); errno = saved_errno; @@ -1554,8 +1591,7 @@ suspendThread( StgRegTable *reg, } StgRegTable * -resumeThread( StgInt tok, - rtsBool concCall STG_UNUSED ) +resumeThread( StgInt tok ) { StgTSO *tso, **prev; Capability *cap; @@ -1682,7 +1718,7 @@ createThread(nat size) /* check that no more than RtsFlags.ParFlags.maxThreads threads are created */ if (advisory_thread_count >= RtsFlags.ParFlags.maxThreads) { threadsIgnored++; - belch("{createThread}Daq ghuH: refusing to create another thread; no more than %d threads allowed (currently %d)", + 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; } @@ -1725,6 +1761,8 @@ createThread(nat size) - TSO_STRUCT_SIZEW; tso->sp = (P_)&(tso->stack) + stack_size; + tso->trec = NO_TREC; + #ifdef PROFILING tso->prof.CCCS = CCS_MAIN; #endif @@ -1732,9 +1770,10 @@ createThread(nat size) /* 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) - tso->link = END_TSO_QUEUE; /* uses more flexible routine in GranSim */ insertThread(tso, CurrentProc); #else @@ -1814,22 +1853,22 @@ createThread(nat size) // collect parallel global statistics (currently done together with GC stats) if (RtsFlags.ParFlags.ParStats.Global && RtsFlags.GcFlags.giveStats > NO_GC_STATS) { - //fprintf(stderr, "Creating thread %d @ %11.2f\n", tso->id, usertime()); + //debugBelch("Creating thread %d @ %11.2f\n", tso->id, usertime()); globalParStats.tot_threads_created++; } #endif #if defined(GRAN) IF_GRAN_DEBUG(pri, - belch("==__ schedule: Created TSO %d (%p);", + sched_belch("==__ schedule: Created TSO %d (%p);", CurrentProc, tso, tso->id)); #elif defined(PAR) IF_PAR_DEBUG(verbose, - belch("==__ schedule: Created TSO %d (%p); %d threads active", - tso->id, tso, advisory_thread_count)); + 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", - tso->id, tso->stack_size)); + (long)tso->id, (long)tso->stack_size)); #endif return tso; } @@ -1878,7 +1917,7 @@ activateSpark (rtsSpark spark) if (RtsFlags.ParFlags.ParStats.Full) { //ASSERT(run_queue_hd == END_TSO_QUEUE); // I think ... IF_PAR_DEBUG(verbose, - belch("==^^ activateSpark: turning spark of closure %p (%s) into a thread", + debugBelch("==^^ activateSpark: turning spark of closure %p (%s) into a thread\n", (StgClosure *)spark, info_type((StgClosure *)spark))); } // ToDo: fwd info on local/global spark to thread -- HWL @@ -1910,9 +1949,10 @@ static void scheduleThread_ (StgTSO* tso); void scheduleThread_(StgTSO *tso) { - // Precondition: sched_mutex must be held. - PUSH_ON_RUN_QUEUE(tso); - THREAD_RUNNABLE(); + // 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(); } void @@ -1971,8 +2011,8 @@ scheduleWaitThread(StgTSO* tso, /*[out]*/HaskellObj* ret, */ IF_DEBUG(scheduler, sched_belch("waiting for thread (%d)", tso->id)); - PUSH_ON_RUN_QUEUE(tso); - // NB. Don't call THREAD_RUNNABLE() here, because the thread is + 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. @@ -2244,8 +2284,8 @@ threadStackOverflow(StgTSO *tso) if (tso->stack_size >= tso->max_stack_size) { IF_DEBUG(gc, - belch("@@ threadStackOverflow of TSO %d (%p): stack too large (now %ld; max is %ld)", - tso->id, tso, tso->stack_size, tso->max_stack_size); + debugBelch("@@ threadStackOverflow of TSO %ld (%p): stack too large (now %ld; max is %ld)\n", + (long)tso->id, tso, (long)tso->stack_size, (long)tso->max_stack_size); /* If we're debugging, just print out the top of the stack */ printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size, tso->sp+64))); @@ -2265,7 +2305,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, fprintf(stderr,"== sched: increasing stack size from %d words to %d.\n", tso->stack_size, new_stack_size)); + IF_DEBUG(scheduler, debugBelch("== sched: increasing stack size from %d words to %d.\n", tso->stack_size, new_stack_size)); dest = (StgTSO *)allocate(new_tso_size); TICK_ALLOC_TSO(new_stack_size,0); @@ -2294,7 +2334,7 @@ threadStackOverflow(StgTSO *tso) dest->mut_link = NULL; IF_PAR_DEBUG(verbose, - belch("@@ threadStackOverflow of TSO %d (now at %p): stack size increased to %ld", + 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, @@ -2383,11 +2423,11 @@ unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node) } /* the thread-queue-overhead is accounted for in either Resume or UnblockThread */ IF_GRAN_DEBUG(bq, - fprintf(stderr," %s TSO %d (%p) [PE %d] (block_info.closure=%p) (next=%p) ,", + debugBelch(" %s TSO %d (%p) [PE %d] (block_info.closure=%p) (next=%p) ,", (node_loc==tso_loc ? "Local" : "Global"), tso->id, tso, CurrentProc, tso->block_info.closure, tso->link)); tso->block_info.closure = NULL; - IF_DEBUG(scheduler,belch("-- Waking up thread %ld (%p)", + IF_DEBUG(scheduler,debugBelch("-- Waking up thread %ld (%p)\n", tso->id, tso)); } #elif defined(PAR) @@ -2401,9 +2441,9 @@ unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node) ASSERT(((StgTSO *)bqe)->why_blocked != NotBlocked); /* if it's a TSO just push it onto the run_queue */ next = bqe->link; - // ((StgTSO *)bqe)->link = END_TSO_QUEUE; // debugging? - PUSH_ON_RUN_QUEUE((StgTSO *)bqe); - THREAD_RUNNABLE(); + ((StgTSO *)bqe)->link = END_TSO_QUEUE; // debugging? + APPEND_TO_RUN_QUEUE((StgTSO *)bqe); + threadRunnable(); unblockCount(bqe, node); /* reset blocking status after dumping event */ ((StgTSO *)bqe)->why_blocked = NotBlocked; @@ -2432,7 +2472,7 @@ unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node) (StgClosure *)bqe); # endif } - IF_PAR_DEBUG(bq, fprintf(stderr, ", %p (%s)", bqe, info_type((StgClosure*)bqe))); + IF_PAR_DEBUG(bq, debugBelch(", %p (%s)\n", bqe, info_type((StgClosure*)bqe))); return next; } @@ -2446,9 +2486,10 @@ unblockOneLocked(StgTSO *tso) ASSERT(tso->why_blocked != NotBlocked); tso->why_blocked = NotBlocked; next = tso->link; - PUSH_ON_RUN_QUEUE(tso); - THREAD_RUNNABLE(); - IF_DEBUG(scheduler,sched_belch("waking up thread %ld", tso->id)); + tso->link = END_TSO_QUEUE; + APPEND_TO_RUN_QUEUE(tso); + threadRunnable(); + IF_DEBUG(scheduler,sched_belch("waking up thread %ld", (long)tso->id)); return next; } #endif @@ -2482,7 +2523,7 @@ awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node) nat len = 0; IF_GRAN_DEBUG(bq, - belch("##-_ AwBQ for node %p on PE %d @ %ld by TSO %d (%p): ", \ + debugBelch("##-_ AwBQ for node %p on PE %d @ %ld by TSO %d (%p): \n", \ node, CurrentProc, CurrentTime[CurrentProc], CurrentTSO->id, CurrentTSO)); @@ -2499,13 +2540,13 @@ awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node) */ if (CurrentProc!=node_loc) { IF_GRAN_DEBUG(bq, - belch("## node %p is on PE %d but CurrentProc is %d (TSO %d); assuming fake fetch and adjusting bitmask (old: %#x)", + debugBelch("## node %p is on PE %d but CurrentProc is %d (TSO %d); assuming fake fetch and adjusting bitmask (old: %#x)\n", node, node_loc, CurrentProc, CurrentTSO->id, // CurrentTSO, where_is(CurrentTSO), node->header.gran.procs)); node->header.gran.procs = (node->header.gran.procs) | PE_NUMBER(CurrentProc); IF_GRAN_DEBUG(bq, - belch("## new bitmask of node %p is %#x", + debugBelch("## new bitmask of node %p is %#x\n", node, node->header.gran.procs)); if (RtsFlags.GranFlags.GranSimStats.Global) { globalGranStats.tot_fake_fetches++; @@ -2540,7 +2581,7 @@ awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node) ((StgRBH *)node)->mut_link = (StgMutClosure *)((StgRBHSave *)bqe)->payload[1]; IF_GRAN_DEBUG(bq, - belch("## Filled in RBH_Save for %p (%s) at end of AwBQ", + debugBelch("## Filled in RBH_Save for %p (%s) at end of AwBQ\n", node, info_type(node))); } @@ -2552,7 +2593,7 @@ awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node) globalGranStats.tot_awbq++; // total no. of bqs awakened } IF_GRAN_DEBUG(bq, - fprintf(stderr,"## BQ Stats of %p: [%d entries] %s\n", + debugBelch("## BQ Stats of %p: [%d entries] %s\n", node, len, (bqe!=END_BQ_QUEUE) ? "RBH" : "")); } #elif defined(PAR) @@ -2564,12 +2605,12 @@ awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node) ACQUIRE_LOCK(&sched_mutex); IF_PAR_DEBUG(verbose, - belch("##-_ AwBQ for node %p on [%x]: ", + debugBelch("##-_ AwBQ for node %p on [%x]: \n", node, mytid)); #ifdef DIST //RFP if(get_itbl(q)->type == CONSTR || q==END_BQ_QUEUE) { - IF_PAR_DEBUG(verbose, belch("## ... nothing to unblock so lets just return. RFP (BUG?)")); + IF_PAR_DEBUG(verbose, debugBelch("## ... nothing to unblock so lets just return. RFP (BUG?)\n")); return; } #endif @@ -2618,9 +2659,6 @@ interruptStgRts(void) { interrupted = 1; context_switch = 1; -#ifdef RTS_SUPPORTS_THREADS - wakeBlockedWorkerThread(); -#endif } /* ----------------------------------------------------------------------------- @@ -2650,6 +2688,14 @@ unblockThread(StgTSO *tso) case NotBlocked: return; /* not blocked */ + case BlockedOnSTM: + // Be careful: nothing to do here! We tell the scheduler that the thread + // is runnable and we leave it to the stack-walking code to abort the + // transaction while unwinding the stack. We should perhaps have a debugging + // test to make sure that this really happens and that the 'zombie' transaction + // does not get committed. + goto done; + case BlockedOnMVar: ASSERT(get_itbl(tso->block_info.closure)->type == MVAR); { @@ -2716,7 +2762,7 @@ unblockThread(StgTSO *tso) case BlockedOnRead: case BlockedOnWrite: -#if defined(mingw32_TARGET_OS) +#if defined(mingw32_HOST_OS) case BlockedOnDoProc: #endif { @@ -2783,6 +2829,14 @@ unblockThread(StgTSO *tso) switch (tso->why_blocked) { + case BlockedOnSTM: + // Be careful: nothing to do here! We tell the scheduler that the thread + // is runnable and we leave it to the stack-walking code to abort the + // transaction while unwinding the stack. We should perhaps have a debugging + // test to make sure that this really happens and that the 'zombie' transaction + // does not get committed. + goto done; + case BlockedOnMVar: ASSERT(get_itbl(tso->block_info.closure)->type == MVAR); { @@ -2846,7 +2900,7 @@ unblockThread(StgTSO *tso) case BlockedOnRead: case BlockedOnWrite: -#if defined(mingw32_TARGET_OS) +#if defined(mingw32_HOST_OS) case BlockedOnDoProc: #endif { @@ -2896,7 +2950,7 @@ unblockThread(StgTSO *tso) tso->link = END_TSO_QUEUE; tso->why_blocked = NotBlocked; tso->block_info.closure = NULL; - PUSH_ON_RUN_QUEUE(tso); + APPEND_TO_RUN_QUEUE(tso); } #endif @@ -2937,7 +2991,10 @@ unblockThread(StgTSO *tso) void deleteThread(StgTSO *tso) { - raiseAsync(tso,NULL); + if (tso->why_blocked != BlockedOnCCall && + tso->why_blocked != BlockedOnCCall_NoUnblockExc) { + raiseAsync(tso,NULL); + } } #ifdef FORKPROCESS_PRIMOP_SUPPORTED @@ -2972,6 +3029,12 @@ raiseAsyncWithLock(StgTSO *tso, StgClosure *exception) void raiseAsync(StgTSO *tso, StgClosure *exception) { + raiseAsync_(tso, exception, rtsFalse); +} + +static void +raiseAsync_(StgTSO *tso, StgClosure *exception, rtsBool stop_at_atomically) +{ StgRetInfoTable *info; StgPtr sp; @@ -2981,7 +3044,7 @@ raiseAsync(StgTSO *tso, StgClosure *exception) } IF_DEBUG(scheduler, - sched_belch("raising exception in thread %ld.", tso->id)); + sched_belch("raising exception in thread %ld.", (long)tso->id)); // Remove it from any blocking queues unblockThread(tso); @@ -3015,6 +3078,10 @@ raiseAsync(StgTSO *tso, StgClosure *exception) // top of the stack applied to the exception. // // 5. If it's a STOP_FRAME, then kill the thread. + // + // NB: if we pass an ATOMICALLY_FRAME then abort the associated + // transaction + StgPtr frame; @@ -3023,13 +3090,45 @@ raiseAsync(StgTSO *tso, StgClosure *exception) while (info->i.type != UPDATE_FRAME && (info->i.type != CATCH_FRAME || exception == NULL) - && info->i.type != STOP_FRAME) { + && 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 @@ -3096,9 +3195,9 @@ raiseAsync(StgTSO *tso, StgClosure *exception) TICK_ALLOC_UP_THK(words+1,0); IF_DEBUG(scheduler, - fprintf(stderr, "sched: Updating "); + debugBelch("sched: Updating "); printPtr((P_)((StgUpdateFrame *)frame)->updatee); - fprintf(stderr, " with "); + debugBelch(" with "); printObj((StgClosure *)ap); ); @@ -3116,7 +3215,8 @@ raiseAsync(StgTSO *tso, StgClosure *exception) // if (!closure_IND(((StgUpdateFrame *)frame)->updatee)) { // revert the black hole - UPD_IND_NOLOCK(((StgUpdateFrame *)frame)->updatee,ap); + UPD_IND_NOLOCK(((StgUpdateFrame *)frame)->updatee, + (StgClosure *)ap); } sp += sizeofW(StgUpdateFrame) - 1; sp[0] = (W_)ap; // push onto stack @@ -3138,6 +3238,137 @@ raiseAsync(StgTSO *tso, StgClosure *exception) } /* ----------------------------------------------------------------------------- + raiseExceptionHelper + + This function is called by the raise# primitve, just so that we can + move some of the tricky bits of raising an exception from C-- into + C. Who knows, it might be a useful re-useable thing here too. + -------------------------------------------------------------------------- */ + +StgWord +raiseExceptionHelper (StgTSO *tso, StgClosure *exception) +{ + StgClosure *raise_closure = NULL; + StgPtr p, next; + StgRetInfoTable *info; + // + // This closure represents the expression 'raise# E' where E + // is the exception raise. It is used to overwrite all the + // thunks which are currently under evaluataion. + // + + // + // LDV profiling: stg_raise_info has THUNK as its closure + // type. Since a THUNK takes at least MIN_UPD_SIZE words in its + // payload, MIN_UPD_SIZE is more approprate than 1. It seems that + // 1 does not cause any problem unless profiling is performed. + // However, when LDV profiling goes on, we need to linearly scan + // small object pool, where raise_closure is stored, so we should + // use MIN_UPD_SIZE. + // + // raise_closure = (StgClosure *)RET_STGCALL1(P_,allocate, + // sizeofW(StgClosure)+1); + // + + // + // Walk up the stack, looking for the catch frame. On the way, + // we update any closures pointed to from update frames with the + // raise closure that we just built. + // + p = tso->sp; + while(1) { + info = get_ret_itbl((StgClosure *)p); + next = p + stack_frame_sizeW((StgClosure *)p); + switch (info->i.type) { + + case UPDATE_FRAME: + // Only create raise_closure if we need to. + if (raise_closure == NULL) { + raise_closure = + (StgClosure *)allocate(sizeofW(StgClosure)+MIN_UPD_SIZE); + SET_HDR(raise_closure, &stg_raise_info, CCCS); + raise_closure->payload[0] = exception; + } + UPD_IND(((StgUpdateFrame *)p)->updatee,raise_closure); + p = next; + continue; + + case ATOMICALLY_FRAME: + IF_DEBUG(stm, debugBelch("Found ATOMICALLY_FRAME at %p\n", p)); + tso->sp = p; + return ATOMICALLY_FRAME; + + case CATCH_FRAME: + tso->sp = p; + return CATCH_FRAME; + + case CATCH_STM_FRAME: + IF_DEBUG(stm, debugBelch("Found CATCH_STM_FRAME at %p\n", p)); + tso->sp = p; + return CATCH_STM_FRAME; + + case STOP_FRAME: + tso->sp = p; + return STOP_FRAME; + + case CATCH_RETRY_FRAME: + default: + p = next; + continue; + } + } +} + + +/* ----------------------------------------------------------------------------- + findRetryFrameHelper + + This function is called by the retry# primitive. It traverses the stack + leaving tso->sp referring to the frame which should handle the retry. + + This should either be a CATCH_RETRY_FRAME (if the retry# is within an orElse#) + or should be a ATOMICALLY_FRAME (if the retry# reaches the top level). + + We skip CATCH_STM_FRAMEs because retries are not considered to be exceptions, + despite the similar implementation. + + We should not expect to see CATCH_FRAME or STOP_FRAME because those should + not be created within memory transactions. + -------------------------------------------------------------------------- */ + +StgWord +findRetryFrameHelper (StgTSO *tso) +{ + StgPtr p, next; + StgRetInfoTable *info; + + p = tso -> sp; + while (1) { + info = get_ret_itbl((StgClosure *)p); + next = p + stack_frame_sizeW((StgClosure *)p); + switch (info->i.type) { + + case ATOMICALLY_FRAME: + IF_DEBUG(stm, debugBelch("Found ATOMICALLY_FRAME at %p during retrry\n", p)); + tso->sp = p; + return ATOMICALLY_FRAME; + + case CATCH_RETRY_FRAME: + IF_DEBUG(stm, debugBelch("Found CATCH_RETRY_FRAME at %p during retrry\n", p)); + tso->sp = p; + return CATCH_RETRY_FRAME; + + case CATCH_STM_FRAME: + default: + ASSERT(info->i.type != CATCH_FRAME); + ASSERT(info->i.type != STOP_FRAME); + p = next; + continue; + } + } +} + +/* ----------------------------------------------------------------------------- resurrectThreads is called after garbage collection on the list of threads found to be garbage. Each of these threads will be woken up and sent a signal: BlockedOnDeadMVar if the thread was blocked @@ -3167,6 +3398,9 @@ resurrectThreads( StgTSO *threads ) 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 @@ -3179,68 +3413,6 @@ resurrectThreads( StgTSO *threads ) } } -/* ----------------------------------------------------------------------------- - * Blackhole detection: if we reach a deadlock, test whether any - * threads are blocked on themselves. Any threads which are found to - * be self-blocked get sent a NonTermination exception. - * - * This is only done in a deadlock situation in order to avoid - * performance overhead in the normal case. - * - * Locks: sched_mutex is held upon entry and exit. - * -------------------------------------------------------------------------- */ - -static void -detectBlackHoles( void ) -{ - StgTSO *tso = all_threads; - StgClosure *frame; - StgClosure *blocked_on; - StgRetInfoTable *info; - - for (tso = all_threads; tso != END_TSO_QUEUE; tso = tso->global_link) { - - while (tso->what_next == ThreadRelocated) { - tso = tso->link; - ASSERT(get_itbl(tso)->type == TSO); - } - - if (tso->why_blocked != BlockedOnBlackHole) { - continue; - } - blocked_on = tso->block_info.closure; - - frame = (StgClosure *)tso->sp; - - while(1) { - info = get_ret_itbl(frame); - switch (info->i.type) { - case UPDATE_FRAME: - if (((StgUpdateFrame *)frame)->updatee == blocked_on) { - /* We are blocking on one of our own computations, so - * send this thread the NonTermination exception. - */ - IF_DEBUG(scheduler, - sched_belch("thread %d is blocked on itself", tso->id)); - raiseAsync(tso, (StgClosure *)NonTermination_closure); - goto done; - } - - frame = (StgClosure *) ((StgUpdateFrame *)frame + 1); - continue; - - case STOP_FRAME: - goto done; - - // normal stack frames; do nothing except advance the pointer - default: - (StgPtr)frame += stack_frame_sizeW(frame); - } - } - done: ; - } -} - /* ---------------------------------------------------------------------------- * Debugging: why is a thread blocked * [Also provides useful information when debugging threaded programs @@ -3253,47 +3425,50 @@ printThreadBlockage(StgTSO *tso) { switch (tso->why_blocked) { case BlockedOnRead: - fprintf(stderr,"is blocked on read from fd %d", tso->block_info.fd); + debugBelch("is blocked on read from fd %d", tso->block_info.fd); break; case BlockedOnWrite: - fprintf(stderr,"is blocked on write to fd %d", tso->block_info.fd); + debugBelch("is blocked on write to fd %d", tso->block_info.fd); break; -#if defined(mingw32_TARGET_OS) +#if defined(mingw32_HOST_OS) case BlockedOnDoProc: - fprintf(stderr,"is blocked on proc (request: %d)", tso->block_info.async_result->reqID); + debugBelch("is blocked on proc (request: %d)", tso->block_info.async_result->reqID); break; #endif case BlockedOnDelay: - fprintf(stderr,"is blocked until %d", tso->block_info.target); + debugBelch("is blocked until %d", tso->block_info.target); break; case BlockedOnMVar: - fprintf(stderr,"is blocked on an MVar"); + debugBelch("is blocked on an MVar"); break; case BlockedOnException: - fprintf(stderr,"is blocked on delivering an exception to thread %d", + debugBelch("is blocked on delivering an exception to thread %d", tso->block_info.tso->id); break; case BlockedOnBlackHole: - fprintf(stderr,"is blocked on a black hole"); + debugBelch("is blocked on a black hole"); break; case NotBlocked: - fprintf(stderr,"is not blocked"); + debugBelch("is not blocked"); break; #if defined(PAR) case BlockedOnGA: - fprintf(stderr,"is blocked on global address; local FM_BQ is %p (%s)", + debugBelch("is blocked on global address; local FM_BQ is %p (%s)", tso->block_info.closure, info_type(tso->block_info.closure)); break; case BlockedOnGA_NoSend: - fprintf(stderr,"is blocked on global address (no send); local FM_BQ is %p (%s)", + debugBelch("is blocked on global address (no send); local FM_BQ is %p (%s)", tso->block_info.closure, info_type(tso->block_info.closure)); break; #endif case BlockedOnCCall: - fprintf(stderr,"is blocked on an external call"); + debugBelch("is blocked on an external call"); break; case BlockedOnCCall_NoUnblockExc: - fprintf(stderr,"is blocked on an external call (exceptions were already blocked)"); + debugBelch("is blocked on an external call (exceptions were already blocked)"); + break; + case BlockedOnSTM: + debugBelch("is blocked on an STM operation"); break; default: barf("printThreadBlockage: strange tso->why_blocked: %d for TSO %d (%d)", @@ -3307,10 +3482,10 @@ printThreadStatus(StgTSO *tso) { switch (tso->what_next) { case ThreadKilled: - fprintf(stderr,"has been killed"); + debugBelch("has been killed"); break; case ThreadComplete: - fprintf(stderr,"has completed"); + debugBelch("has completed"); break; default: printThreadBlockage(tso); @@ -3321,30 +3496,33 @@ void printAllThreads(void) { StgTSO *t; - void *label; # 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!*/); - fprintf(stderr, "all threads at [%s]:\n", time_string); + debugBelch("all threads at [%s]:\n", time_string); # elif defined(PAR) char time_string[TIME_STR_LEN], node_str[NODE_STR_LEN]; ullong_format_string(CURRENT_TIME, time_string, rtsFalse/*no commas!*/); - fprintf(stderr,"all threads at [%s]:\n", time_string); + debugBelch("all threads at [%s]:\n", time_string); # else - fprintf(stderr,"all threads:\n"); + debugBelch("all threads:\n"); # endif for (t = all_threads; t != END_TSO_QUEUE; t = t->global_link) { - fprintf(stderr, "\tthread %d @ %p ", t->id, (void *)t); - label = lookupThreadLabel(t->id); - if (label) fprintf(stderr,"[\"%s\"] ",(char *)label); + debugBelch("\tthread %d @ %p ", t->id, (void *)t); +#if defined(DEBUG) + { + void *label = lookupThreadLabel(t->id); + if (label) debugBelch("[\"%s\"] ",(char *)label); + } +#endif printThreadStatus(t); - fprintf(stderr,"\n"); + debugBelch("\n"); } } @@ -3361,7 +3539,7 @@ print_bq (StgClosure *node) StgTSO *tso; rtsBool end; - fprintf(stderr,"## BQ of closure %p (%s): ", + debugBelch("## BQ of closure %p (%s): ", node, info_type(node)); /* should cover all closures that may have a blocking queue */ @@ -3401,18 +3579,18 @@ print_bqe (StgBlockingQueueElement *bqe) switch (get_itbl(bqe)->type) { case TSO: - fprintf(stderr," TSO %u (%x),", + debugBelch(" TSO %u (%x),", ((StgTSO *)bqe)->id, ((StgTSO *)bqe)); break; case BLOCKED_FETCH: - fprintf(stderr," BF (node=%p, ga=((%x, %d, %x)),", + 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: - fprintf(stderr," %s (IP %p),", + 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" : @@ -3424,7 +3602,7 @@ print_bqe (StgBlockingQueueElement *bqe) break; } } /* for */ - fputc('\n', stderr); + debugBelch("\n"); } # elif defined(GRAN) void @@ -3442,7 +3620,7 @@ print_bq (StgClosure *node) ASSERT(node!=(StgClosure*)NULL); // sanity check node_loc = where_is(node); - fprintf(stderr,"## BQ of closure %p (%s) on [PE %d]: ", + debugBelch("## BQ of closure %p (%s) on [PE %d]: ", node, info_type(node), node_loc); /* @@ -3462,11 +3640,11 @@ print_bq (StgClosure *node) tso_loc = where_is((StgClosure *)bqe); switch (get_itbl(bqe)->type) { case TSO: - fprintf(stderr," TSO %d (%p) on [PE %d],", + debugBelch(" TSO %d (%p) on [PE %d],", ((StgTSO *)bqe)->id, (StgTSO *)bqe, tso_loc); break; case CONSTR: - fprintf(stderr," %s (IP %p),", + 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" : @@ -3478,7 +3656,7 @@ print_bq (StgClosure *node) break; } } /* for */ - fputc('\n', stderr); + debugBelch("\n"); } #else /* @@ -3495,9 +3673,9 @@ print_bq (StgClosure *node) tso=tso->link) { ASSERT(tso!=NULL && tso!=END_TSO_QUEUE); // sanity check ASSERT(get_itbl(tso)->type == TSO); // guess what, sanity check - fprintf(stderr," TSO %d (%p),", tso->id, tso); + debugBelch(" TSO %d (%p),", tso->id, tso); } - fputc('\n', stderr); + debugBelch("\n"); } # endif @@ -3523,15 +3701,14 @@ sched_belch(char *s, ...) va_list ap; va_start(ap,s); #ifdef RTS_SUPPORTS_THREADS - fprintf(stderr, "sched (task %p): ", osThreadId()); + debugBelch("sched (task %p): ", osThreadId()); #elif defined(PAR) - fprintf(stderr, "== "); + debugBelch("== "); #else - fprintf(stderr, "sched: "); + debugBelch("sched: "); #endif - vfprintf(stderr, s, ap); - fprintf(stderr, "\n"); - fflush(stderr); + vdebugBelch(s, ap); + debugBelch("\n"); va_end(ap); }