X-Git-Url: http://git.megacz.com/?a=blobdiff_plain;f=ghc%2Frts%2FSchedule.c;h=9e2a5d053a481490f2859708d317994387854dc6;hb=6458c2c59780ea01a79f3c31b9b2545a0241606d;hp=c1c929994cb8fcad070f14bdbfac2421543db614;hpb=1b170b6ca71637cf3574eb1cbb7030c22459826d;p=ghc-hetmet.git diff --git a/ghc/rts/Schedule.c b/ghc/rts/Schedule.c index c1c9299..9e2a5d0 100644 --- a/ghc/rts/Schedule.c +++ b/ghc/rts/Schedule.c @@ -1,7 +1,6 @@ /* --------------------------------------------------------------------------- - * $Id: Schedule.c,v 1.182 2003/12/12 16:35:20 simonmar Exp $ * - * (c) The GHC Team, 1998-2000 + * (c) The GHC Team, 1998-2004 * * Scheduler * @@ -10,36 +9,14 @@ * * WAY Name CPP flag What's it for * -------------------------------------- - * mp GUM PAR Parallel execution on a distributed memory machine + * mp GUM PAR Parallel execution on a distrib. memory machine * s SMP SMP Parallel execution on a shared memory machine * mg GranSim GRAN Simulation of parallel execution * md GUM/GdH DIST Distributed execution (based on GUM) * * --------------------------------------------------------------------------*/ -//@node Main scheduling code, , , -//@section Main scheduling code - /* - * Version with scheduler monitor support for SMPs (WAY=s): - - This design provides a high-level API to create and schedule threads etc. - as documented in the SMP design document. - - It uses a monitor design controlled by a single mutex to exercise control - over accesses to shared data structures, and builds on the Posix threads - library. - - The majority of state is shared. In order to keep essential per-task state, - there is a Capability structure, which contains all the information - needed to run a thread: its STG registers, a pointer to its TSO, a - nursery etc. During STG execution, a pointer to the capability is - kept in a register (BaseReg). - - In a non-SMP build, there is one global capability, namely MainRegTable. - - SDM & KH, 10/99 - * Version with support for distributed memory parallelism aka GUM (WAY=mp): The main scheduling loop in GUM iterates until a finish message is received. @@ -59,30 +36,14 @@ over the events in the global event queue. -- HWL */ -//@menu -//* Includes:: -//* Variables and Data structures:: -//* Main scheduling loop:: -//* Suspend and Resume:: -//* Run queue code:: -//* Garbage Collextion Routines:: -//* Blocking Queue Routines:: -//* Exception Handling Routines:: -//* Debugging Routines:: -//* Index:: -//@end menu - -//@node Includes, Variables and Data structures, Main scheduling code, Main scheduling code -//@subsection Includes - #include "PosixSource.h" #include "Rts.h" #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" @@ -97,6 +58,8 @@ #include "Timer.h" #include "Prelude.h" #include "ThreadLabels.h" +#include "LdvProfile.h" +#include "Updates.h" #ifdef PROFILING #include "Proftimer.h" #include "ProfHeap.h" @@ -142,9 +105,6 @@ #define USED_WHEN_RTS_SUPPORTS_THREADS STG_UNUSED #endif -//@node Variables and Data structures, Prototypes, Includes, Main scheduling code -//@subsection Variables and Data structures - /* Main thread queue. * Locks required: sched_mutex. */ @@ -202,17 +162,14 @@ static StgTSO *threadStackOverflow(StgTSO *tso); */ /* flag set by signal handler to precipitate a context switch */ -//@cindex context_switch nat context_switch = 0; /* if this flag is set as well, give up execution */ -//@cindex interrupted rtsBool interrupted = rtsFalse; /* Next thread ID to allocate. * Locks required: thread_id_mutex */ -//@cindex next_thread_id static StgThreadID next_thread_id = 1; /* @@ -268,18 +225,6 @@ static void detectBlackHoles ( void ); Mutex sched_mutex = INIT_MUTEX_VAR; Mutex term_mutex = INIT_MUTEX_VAR; -/* - * A heavyweight solution to the problem of protecting - * the thread_id from concurrent update. - */ -Mutex thread_id_mutex = INIT_MUTEX_VAR; - - -# if defined(SMP) -static Condition gc_pending_cond = INIT_COND_VAR; -nat await_death; -# endif - #endif /* RTS_SUPPORTS_THREADS */ #if defined(PAR) @@ -290,6 +235,7 @@ rtsBool emitSchedule = rtsTrue; #if DEBUG static char *whatNext_strs[] = { + "(unknown)", "ThreadRunGHC", "ThreadInterpret", "ThreadKilled", @@ -303,11 +249,9 @@ StgTSO * createSparkThread(rtsSpark spark); StgTSO * activateSpark (rtsSpark spark); #endif -/* - * The thread state for the main thread. -// ToDo: check whether not needed any more -StgTSO *MainTSO; - */ +/* ---------------------------------------------------------------------------- + * Starting Tasks + * ------------------------------------------------------------------------- */ #if defined(RTS_SUPPORTS_THREADS) static rtsBool startingWorkerThread = rtsFalse; @@ -316,14 +260,10 @@ static void taskStart(void); static void taskStart(void) { - Capability *cap; - ACQUIRE_LOCK(&sched_mutex); startingWorkerThread = rtsFalse; - waitForWorkCapability(&sched_mutex, &cap, NULL); + schedule(NULL,NULL); RELEASE_LOCK(&sched_mutex); - - schedule(NULL,cap); } void @@ -338,15 +278,15 @@ 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; + } } } } #endif -//@node Main scheduling loop, Suspend and Resume, Prototypes, Main scheduling code -//@subsection Main scheduling loop - /* --------------------------------------------------------------------------- Main scheduling loop. @@ -382,13 +322,12 @@ startSchedulerTaskIfNecessary(void) This is not the ugliest code you could imagine, but it's bloody close. ------------------------------------------------------------------------ */ -//@cindex schedule static void schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, Capability *initialCapability ) { StgTSO *t; - Capability *cap = initialCapability; + Capability *cap; StgThreadReturnCode ret; #if defined(GRAN) rtsEvent *event; @@ -403,10 +342,12 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, # endif #endif rtsBool was_interrupted = rtsFalse; - StgTSOWhatNext prev_what_next; + nat prev_what_next; - ACQUIRE_LOCK(&sched_mutex); - + // Pre-condition: sched_mutex is held. + // We might have a capability, passed in as initialCapability. + cap = initialCapability; + #if defined(RTS_SUPPORTS_THREADS) // // in the threaded case, the capability is either passed in via the @@ -418,7 +359,7 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, mainThread, initialCapability); ); #else - /* simply initialise it in the non-threaded case */ + // simply initialise it in the non-threaded case grabCapability(&cap); #endif @@ -430,7 +371,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) { @@ -459,23 +400,20 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, IF_DEBUG(scheduler, printAllThreads()); #if defined(RTS_SUPPORTS_THREADS) - // - // Check to see whether there are any worker threads - // waiting to deposit external call results. If so, - // yield our capability... if we have a capability, that is. - // - if (cap != NULL) { - yieldToReturningWorker(&sched_mutex, &cap, - mainThread ? &mainThread->bound_thread_cond - : NULL); - } + // Yield the capability to higher-priority tasks if necessary. + // + if (cap != NULL) { + yieldCapability(&cap); + } - // If we do not currently hold a capability, we wait for one - if (cap == NULL) { - waitForWorkCapability(&sched_mutex, &cap, - mainThread ? &mainThread->bound_thread_cond - : NULL); - } + // If we do not currently hold a capability, we wait for one + // + if (cap == NULL) { + waitForCapability(&sched_mutex, &cap, + mainThread ? &mainThread->bound_thread_cond : NULL); + } + + // We now have a capability... #endif // @@ -490,7 +428,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); @@ -499,152 +437,6 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, #endif } - // - // Go through the list of main threads and wake up any - // clients whose computations have finished. ToDo: this - // should be done more efficiently without a linear scan - // of the main threads list, somehow... - // -#if defined(RTS_SUPPORTS_THREADS) - { - StgMainThread *m, **prev; - prev = &main_threads; - for (m = main_threads; m != NULL; prev = &m->link, m = m->link) { - if (m->tso->what_next == ThreadComplete - || m->tso->what_next == ThreadKilled) - { - if (m == mainThread) - { - if (m->tso->what_next == ThreadComplete) - { - if (m->ret) - { - // NOTE: return val is tso->sp[1] (see StgStartup.hc) - *(m->ret) = (StgClosure *)m->tso->sp[1]; - } - m->stat = Success; - } - else - { - if (m->ret) - { - *(m->ret) = NULL; - } - if (was_interrupted) - { - m->stat = Interrupted; - } - else - { - m->stat = Killed; - } - } - *prev = m->link; - -#ifdef DEBUG - removeThreadLabel((StgWord)m->tso->id); -#endif - releaseCapability(cap); - RELEASE_LOCK(&sched_mutex); - return; - } - else - { - // The current OS thread can not handle the fact that - // the Haskell thread "m" has ended. "m" is bound; - // the scheduler loop in it's bound OS thread has to - // return, so let's pass our capability directly to - // that thread. - passCapability(&sched_mutex, cap, &m->bound_thread_cond); - cap = NULL; - } - } - } - } - - // If we gave our capability away, go to the top to get it back - if (cap == NULL) { - continue; - } - -#else /* not threaded */ - -# if defined(PAR) - /* in GUM do this only on the Main PE */ - if (IAmMainThread) -# endif - /* If our main thread has finished or been killed, return. - */ - { - StgMainThread *m = main_threads; - if (m->tso->what_next == ThreadComplete - || m->tso->what_next == ThreadKilled) { -#ifdef DEBUG - removeThreadLabel((StgWord)m->tso->id); -#endif - main_threads = main_threads->link; - if (m->tso->what_next == ThreadComplete) { - // We finished successfully, fill in the return value - // NOTE: return val is tso->sp[1] (see StgStartup.hc) - if (m->ret) { *(m->ret) = (StgClosure *)m->tso->sp[1]; }; - m->stat = Success; - return; - } else { - if (m->ret) { *(m->ret) = NULL; }; - if (was_interrupted) { - m->stat = Interrupted; - } else { - m->stat = Killed; - } - return; - } - } - } -#endif - - -#if 0 /* defined(SMP) */ - /* Top up the run queue from our spark pool. We try to make the - * number of threads in the run queue equal to the number of - * free capabilities. - * - * Disable spark support in SMP for now, non-essential & requires - * a little bit of work to make it compile cleanly. -- sof 1/02. - */ - { - nat n = getFreeCapabilities(); - StgTSO *tso = run_queue_hd; - - /* Count the run queue */ - while (n > 0 && tso != END_TSO_QUEUE) { - tso = tso->link; - n--; - } - - for (; n > 0; n--) { - StgClosure *spark; - spark = findSpark(rtsFalse); - if (spark == NULL) { - break; /* no more sparks in the pool */ - } else { - /* I'd prefer this to be done in activateSpark -- HWL */ - /* tricky - it needs to hold the scheduler lock and - * not try to re-acquire it -- SDM */ - createSparkThread(spark); - IF_DEBUG(scheduler, - sched_belch("==^^ turning spark of closure %p into a thread", - (StgClosure *)spark)); - } - } - /* We need to wake up the other tasks if we just created some - * work for them. - */ - if (getFreeCapabilities() - n > 1) { - signalCondition( &thread_ready_cond ); - } - } -#endif // SMP - #if defined(RTS_USER_SIGNALS) // check for signals each time around the scheduler if (signals_pending()) { @@ -654,23 +446,20 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, } #endif - /* Check whether any waiting threads need to be woken up. If the - * run queue is empty, and there are no other tasks running, we - * can wait indefinitely for something to happen. - */ - if ( !EMPTY_QUEUE(blocked_queue_hd) || !EMPTY_QUEUE(sleeping_queue) -#if defined(RTS_SUPPORTS_THREADS) && !defined(SMP) + // + // Check whether any waiting threads need to be woken up. If the + // run queue is empty, and there are no other tasks running, we + // can wait indefinitely for something to happen. + // + if ( !EMPTY_QUEUE(blocked_queue_hd) || !EMPTY_QUEUE(sleeping_queue) +#if defined(RTS_SUPPORTS_THREADS) || EMPTY_RUN_QUEUE() #endif - ) + ) { - awaitEvent( EMPTY_RUN_QUEUE() -#if defined(SMP) - && allFreeCapabilities() -#endif - ); + awaitEvent( EMPTY_RUN_QUEUE() ); } - /* we can be interrupted while waiting for I/O... */ + // we can be interrupted while waiting for I/O... if (interrupted) continue; /* @@ -685,20 +474,9 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, * inform all the main threads. */ #if !defined(PAR) && !defined(RTS_SUPPORTS_THREADS) - if ( EMPTY_THREAD_QUEUES() -#if defined(RTS_SUPPORTS_THREADS) - && EMPTY_QUEUE(suspended_ccalling_threads) -#endif -#ifdef SMP - && allFreeCapabilities() -#endif - ) + if ( EMPTY_THREAD_QUEUES() ) { IF_DEBUG(scheduler, sched_belch("deadlocked, forcing major GC...")); -#if defined(THREADED_RTS) - /* and SMP mode ..? */ - releaseCapability(cap); -#endif // 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 @@ -718,19 +496,7 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, * for signals to arrive rather then bombing out with a * deadlock. */ -#if defined(RTS_SUPPORTS_THREADS) - if ( 0 ) { /* hmm..what to do? Simply stop waiting for - a signal with no runnable threads (or I/O - suspended ones) leads nowhere quick. - For now, simply shut down when we reach this - condition. - - ToDo: define precisely under what conditions - the Scheduler should shut down in an MT setting. - */ -#else if ( anyUserHandlers() ) { -#endif IF_DEBUG(scheduler, sched_belch("still deadlocked, waiting for signals...")); @@ -756,90 +522,31 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, */ { StgMainThread *m; -#if defined(RTS_SUPPORTS_THREADS) - for (m = main_threads; m != NULL; m = m->link) { - switch (m->tso->why_blocked) { - case BlockedOnBlackHole: - raiseAsync(m->tso, (StgClosure *)NonTermination_closure); - break; - case BlockedOnException: - case BlockedOnMVar: - raiseAsync(m->tso, (StgClosure *)Deadlock_closure); - break; - default: - barf("deadlock: main thread blocked in a strange way"); - } - } -#else m = main_threads; switch (m->tso->why_blocked) { case BlockedOnBlackHole: - raiseAsync(m->tso, (StgClosure *)NonTermination_closure); - break; case BlockedOnException: case BlockedOnMVar: - raiseAsync(m->tso, (StgClosure *)Deadlock_closure); + raiseAsync(m->tso, (StgClosure *)NonTermination_closure); break; default: barf("deadlock: main thread blocked in a strange way"); } -#endif } - -#if defined(RTS_SUPPORTS_THREADS) - /* ToDo: revisit conditions (and mechanism) for shutting - down a multi-threaded world */ - IF_DEBUG(scheduler, sched_belch("all done, i think...shutting down.")); - RELEASE_LOCK(&sched_mutex); - shutdownHaskell(); - return; -#endif } not_deadlocked: #elif defined(RTS_SUPPORTS_THREADS) - /* ToDo: add deadlock detection in threaded RTS */ + // ToDo: add deadlock detection in threaded RTS #elif defined(PAR) - /* ToDo: add deadlock detection in GUM (similar to SMP) -- HWL */ + // ToDo: add deadlock detection in GUM (similar to SMP) -- HWL #endif -#if defined(SMP) - /* If there's a GC pending, don't do anything until it has - * completed. - */ - if (ready_to_gc) { - IF_DEBUG(scheduler,sched_belch("waiting for GC")); - waitCondition( &gc_pending_cond, &sched_mutex ); - } -#endif - #if defined(RTS_SUPPORTS_THREADS) -#if defined(SMP) - /* block until we've got a thread on the run queue and a free - * capability. - * - */ - if ( EMPTY_RUN_QUEUE() ) { - /* Give up our capability */ - releaseCapability(cap); - - /* If we're in the process of shutting down (& running the - * a batch of finalisers), don't wait around. - */ - if ( shutting_down_scheduler ) { - RELEASE_LOCK(&sched_mutex); - return; - } - IF_DEBUG(scheduler, sched_belch("waiting for work")); - waitForWorkCapability(&sched_mutex, &cap, rtsTrue); - IF_DEBUG(scheduler, sched_belch("work now available")); - } -#else if ( EMPTY_RUN_QUEUE() ) { - continue; // nothing to do + continue; // nothing to do } #endif -#endif #if defined(GRAN) if (RtsFlags.GranFlags.Light) @@ -854,13 +561,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); @@ -868,25 +575,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 } @@ -943,14 +650,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) @@ -959,7 +666,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; @@ -967,7 +674,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 @@ -1003,16 +710,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; } @@ -1033,12 +740,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); @@ -1084,7 +791,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 @@ -1126,12 +833,7 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, #ifdef THREADED_RTS { - StgMainThread *m; - for(m = main_threads; m; m = m->link) - { - if(m->tso == t) - break; - } + StgMainThread *m = t->main; if(m) { @@ -1147,8 +849,7 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, sched_belch("### thread %d bound to another OS thread", t->id)); // no, bound to a different Haskell thread: pass to that thread PUSH_ON_RUN_QUEUE(t); - passCapability(&sched_mutex,cap,&m->bound_thread_cond); - cap = NULL; + passCapability(&m->bound_thread_cond); continue; } } @@ -1162,8 +863,7 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, // no, the current native thread is bound to a different // Haskell thread, so pass it to any worker thread PUSH_ON_RUN_QUEUE(t); - passCapabilityToWorker(&sched_mutex, cap); - cap = NULL; + passCapabilityToWorker(); continue; } } @@ -1181,15 +881,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(); @@ -1199,23 +897,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 */ @@ -1227,11 +937,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 @@ -1251,15 +960,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. @@ -1318,8 +1027,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)); @@ -1350,14 +1059,13 @@ 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. */ threadPaused(t); { - StgMainThread *m; /* enlarge the stack */ StgTSO *new_t = threadStackOverflow(t); @@ -1365,17 +1073,22 @@ run_thread: * main thread stack. It better not be on any other queues... * (it shouldn't be). */ - for (m = main_threads; m != NULL; m = m->link) { - if (m->tso == t) { - m->tso = new_t; - } + 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)); @@ -1392,16 +1105,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); @@ -1418,7 +1131,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 @@ -1441,7 +1154,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 */ @@ -1450,7 +1163,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)); @@ -1470,7 +1183,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, @@ -1490,11 +1203,10 @@ run_thread: * case it'll be on the relevant queue already. */ 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? @@ -1513,7 +1225,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 @@ -1531,8 +1243,74 @@ run_thread: !RtsFlags.ParFlags.ParStats.Suppressed) DumpEndEvent(CURRENT_PROC, t, rtsFalse /* not mandatory */); #endif + + // + // Check whether the thread that just completed was a main + // thread, and if so return with the result. + // + // There is an assumption here that all thread completion goes + // through this point; we need to make sure that if a thread + // ends up in the ThreadKilled state, that it stays on the run + // queue so it can be dealt with here. + // + if ( +#if defined(RTS_SUPPORTS_THREADS) + mainThread != NULL +#else + mainThread->tso == t +#endif + ) + { + // We are a bound thread: this must be our thread that just + // completed. + ASSERT(mainThread->tso == t); + + if (t->what_next == ThreadComplete) { + if (mainThread->ret) { + // NOTE: return val is tso->sp[1] (see StgStartup.hc) + *(mainThread->ret) = (StgClosure *)mainThread->tso->sp[1]; + } + mainThread->stat = Success; + } else { + if (mainThread->ret) { + *(mainThread->ret) = NULL; + } + if (was_interrupted) { + mainThread->stat = Interrupted; + } else { + mainThread->stat = Killed; + } + } +#ifdef DEBUG + removeThreadLabel((StgWord)mainThread->tso->id); +#endif + if (mainThread->prev == NULL) { + main_threads = mainThread->link; + } else { + mainThread->prev->link = mainThread->link; + } + if (mainThread->link != NULL) { + mainThread->link->prev = NULL; + } + releaseCapability(cap); + return; + } + +#ifdef RTS_SUPPORTS_THREADS + ASSERT(t->main == NULL); +#else + if (t->main != NULL) { + // Must be a main thread that is not the topmost one. Leave + // it on the run queue until the stack has unwound to the + // point where we can deal with this. Leaving it on the run + // queue also ensures that the garbage collector knows about + // this thread and its return value (it gets dropped from the + // all_threads list so there's no other way to find it). + APPEND_TO_RUN_QUEUE(t); + } +#endif break; - + default: barf("schedule: invalid thread return code %d", (int)ret); } @@ -1550,11 +1328,7 @@ run_thread: } #endif - if (ready_to_gc -#ifdef SMP - && allFreeCapabilities() -#endif - ) { + if (ready_to_gc) { /* 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 @@ -1565,16 +1339,13 @@ run_thread: #endif GarbageCollect(GetRoots,rtsFalse); ready_to_gc = rtsFalse; -#ifdef SMP - broadcastCondition(&gc_pending_cond); -#endif #if defined(GRAN) /* add a ContinueThread event to continue execution of current thread */ new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc], ContinueThread, t, (StgClosure*)NULL, (rtsSpark*)NULL); IF_GRAN_DEBUG(bq, - 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 */ @@ -1594,7 +1365,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")); } /* --------------------------------------------------------------------------- @@ -1620,12 +1391,7 @@ StgBool isThreadBound(StgTSO* tso USED_IN_THREADED_RTS) { #ifdef THREADED_RTS - StgMainThread *m; - for(m = main_threads; m; m = m->link) - { - if(m->tso == tso) - return rtsTrue; - } + return (tso->main != NULL); #endif return rtsFalse; } @@ -1634,13 +1400,22 @@ isThreadBound(StgTSO* tso USED_IN_THREADED_RTS) * Singleton fork(). Do not copy any running threads. * ------------------------------------------------------------------------- */ +#ifndef mingw32_TARGET_OS +#define FORKPROCESS_PRIMOP_SUPPORTED +#endif + +#ifdef FORKPROCESS_PRIMOP_SUPPORTED static void deleteThreadImmediately(StgTSO *tso); - +#endif StgInt -forkProcess(HsStablePtr *entry) +forkProcess(HsStablePtr *entry +#ifndef FORKPROCESS_PRIMOP_SUPPORTED + STG_UNUSED +#endif + ) { -#ifndef mingw32_TARGET_OS +#ifdef FORKPROCESS_PRIMOP_SUPPORTED pid_t pid; StgTSO* t,*next; StgMainThread *m; @@ -1674,17 +1449,17 @@ forkProcess(HsStablePtr *entry) // wipe the main thread list while((m = main_threads) != NULL) { main_threads = m->link; -#ifdef THREADED_RTS +# ifdef THREADED_RTS closeCondition(&m->bound_thread_cond); -#endif +# endif stgFree(m); } -#ifdef RTS_SUPPORTS_THREADS +# ifdef RTS_SUPPORTS_THREADS resetTaskManagerAfterFork(); // tell startTask() and friends that startingWorkerThread = rtsFalse; // we have no worker threads any more resetWorkerWakeupPipeAfterFork(); -#endif +# endif rc = rts_evalStableIO(entry, NULL); // run the action rts_checkSchedStatus("forkProcess",rc); @@ -1694,10 +1469,10 @@ forkProcess(HsStablePtr *entry) hs_exit(); // clean up and exit stg_exit(0); } -#else /* mingw32 */ - barf("forkProcess#: primop not implemented for mingw32, sorry!\n"); +#else /* !FORKPROCESS_PRIMOP_SUPPORTED */ + barf("forkProcess#: primop not supported, sorry!\n"); return -1; -#endif /* mingw32 */ +#endif } /* --------------------------------------------------------------------------- @@ -1718,17 +1493,20 @@ 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 */ -//@node Suspend and Resume, Run queue code, Main scheduling loop, Main scheduling code -//@subsection Suspend and Resume - /* --------------------------------------------------------------------------- * Suspending & resuming Haskell threads. * @@ -1745,12 +1523,7 @@ deleteAllThreads ( void ) * ------------------------------------------------------------------------- */ StgInt -suspendThread( StgRegTable *reg, - rtsBool concCall -#if !defined(RTS_SUPPORTS_THREADS) && !defined(DEBUG) - STG_UNUSED -#endif - ) +suspendThread( StgRegTable *reg ) { nat tok; Capability *cap; @@ -1764,7 +1537,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; @@ -1773,17 +1546,12 @@ suspendThread( StgRegTable *reg, cap->r.rCurrentTSO->link = suspended_ccalling_threads; suspended_ccalling_threads = cap->r.rCurrentTSO; -#if defined(RTS_SUPPORTS_THREADS) - if(cap->r.rCurrentTSO->blocked_exceptions == NULL) - { + if(cap->r.rCurrentTSO->blocked_exceptions == NULL) { cap->r.rCurrentTSO->why_blocked = BlockedOnCCall; cap->r.rCurrentTSO->blocked_exceptions = END_TSO_QUEUE; - } - else - { + } else { cap->r.rCurrentTSO->why_blocked = BlockedOnCCall_NoUnblockExc; } -#endif /* Use the thread ID as the token; it should be unique */ tok = cap->r.rCurrentTSO->id; @@ -1807,8 +1575,7 @@ suspendThread( StgRegTable *reg, } StgRegTable * -resumeThread( StgInt tok, - rtsBool concCall STG_UNUSED ) +resumeThread( StgInt tok ) { StgTSO *tso, **prev; Capability *cap; @@ -1817,7 +1584,7 @@ resumeThread( StgInt tok, #if defined(RTS_SUPPORTS_THREADS) /* Wait for permission to re-enter the RTS with the result. */ ACQUIRE_LOCK(&sched_mutex); - grabReturnCapability(&sched_mutex, &cap); + waitForReturnCapability(&sched_mutex, &cap); IF_DEBUG(scheduler, sched_belch("worker (token %d): re-entering RTS", tok)); #else @@ -1839,21 +1606,16 @@ resumeThread( StgInt tok, } tso->link = END_TSO_QUEUE; -#if defined(RTS_SUPPORTS_THREADS) - if(tso->why_blocked == BlockedOnCCall) - { + if(tso->why_blocked == BlockedOnCCall) { awakenBlockedQueueNoLock(tso->blocked_exceptions); tso->blocked_exceptions = NULL; } -#endif /* Reset blocking status */ tso->why_blocked = NotBlocked; cap->r.rCurrentTSO = tso; -#if defined(RTS_SUPPORTS_THREADS) RELEASE_LOCK(&sched_mutex); -#endif errno = saved_errno; return &cap->r; } @@ -1922,7 +1684,6 @@ labelThread(StgPtr tso, char *label) currently pri (priority) is only used in a GRAN setup -- HWL ------------------------------------------------------------------------ */ -//@cindex createThread #if defined(GRAN) /* currently pri (priority) is only used in a GRAN setup -- HWL */ StgTSO * @@ -1941,7 +1702,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; } @@ -1972,18 +1733,12 @@ createThread(nat size) // Always start with the compiled code evaluator tso->what_next = ThreadRunGHC; - /* tso->id needs to be unique. For now we use a heavyweight mutex to - * protect the increment operation on next_thread_id. - * In future, we could use an atomic increment instead. - */ - ACQUIRE_LOCK(&thread_id_mutex); tso->id = next_thread_id++; - RELEASE_LOCK(&thread_id_mutex); - tso->why_blocked = NotBlocked; tso->blocked_exceptions = NULL; tso->saved_errno = 0; + tso->main = NULL; tso->stack_size = stack_size; tso->max_stack_size = round_to_mblocks(RtsFlags.GcFlags.maxStkSize) @@ -1997,9 +1752,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 @@ -2079,22 +1835,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; } @@ -2134,7 +1890,6 @@ createSparkThread(rtsSpark spark) ToDo: fix for SMP (needs to acquire SCHED_MUTEX!) */ #if defined(PAR) -//@cindex activateSpark StgTSO * activateSpark (rtsSpark spark) { @@ -2144,7 +1899,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 @@ -2177,61 +1932,74 @@ void scheduleThread_(StgTSO *tso) { // Precondition: sched_mutex must be held. - - /* Put the new thread on the head of the runnable queue. The caller - * better push an appropriate closure on this thread's stack - * beforehand. In the SMP case, the thread may start running as - * soon as we release the scheduler lock below. - */ - PUSH_ON_RUN_QUEUE(tso); + // The thread goes at the *end* of the run-queue, to avoid possible + // starvation of any threads already on the queue. + APPEND_TO_RUN_QUEUE(tso); THREAD_RUNNABLE(); - -#if 0 - IF_DEBUG(scheduler,printTSO(tso)); -#endif } -void scheduleThread(StgTSO* tso) +void +scheduleThread(StgTSO* tso) { ACQUIRE_LOCK(&sched_mutex); scheduleThread_(tso); RELEASE_LOCK(&sched_mutex); } +#if defined(RTS_SUPPORTS_THREADS) +static Condition bound_cond_cache; +static int bound_cond_cache_full = 0; +#endif + + SchedulerStatus -scheduleWaitThread(StgTSO* tso, /*[out]*/HaskellObj* ret, Capability *initialCapability) -{ // Precondition: sched_mutex must be held - StgMainThread *m; +scheduleWaitThread(StgTSO* tso, /*[out]*/HaskellObj* ret, + Capability *initialCapability) +{ + // Precondition: sched_mutex must be held + StgMainThread *m; + + m = stgMallocBytes(sizeof(StgMainThread), "waitThread"); + m->tso = tso; + tso->main = m; + m->ret = ret; + m->stat = NoStatus; + m->link = main_threads; + m->prev = NULL; + if (main_threads != NULL) { + main_threads->prev = m; + } + main_threads = m; - m = stgMallocBytes(sizeof(StgMainThread), "waitThread"); - m->tso = tso; - m->ret = ret; - m->stat = NoStatus; #if defined(RTS_SUPPORTS_THREADS) -#if defined(THREADED_RTS) - initCondition(&m->bound_thread_cond); -#else - initCondition(&m->wakeup); -#endif + // Allocating a new condition for each thread is expensive, so we + // cache one. This is a pretty feeble hack, but it helps speed up + // consecutive call-ins quite a bit. + if (bound_cond_cache_full) { + m->bound_thread_cond = bound_cond_cache; + bound_cond_cache_full = 0; + } else { + initCondition(&m->bound_thread_cond); + } #endif - /* Put the thread on the main-threads list prior to scheduling the TSO. - Failure to do so introduces a race condition in the MT case (as - identified by Wolfgang Thaller), whereby the new task/OS thread - created by scheduleThread_() would complete prior to the thread - that spawned it managed to put 'itself' on the main-threads list. - The upshot of it all being that the worker thread wouldn't get to - signal the completion of the its work item for the main thread to - see (==> it got stuck waiting.) -- sof 6/02. - */ - IF_DEBUG(scheduler, sched_belch("waiting for thread (%d)", tso->id)); - - m->link = main_threads; - main_threads = m; - - scheduleThread_(tso); + /* Put the thread on the main-threads list prior to scheduling the TSO. + Failure to do so introduces a race condition in the MT case (as + identified by Wolfgang Thaller), whereby the new task/OS thread + created by scheduleThread_() would complete prior to the thread + that spawned it managed to put 'itself' on the main-threads list. + The upshot of it all being that the worker thread wouldn't get to + signal the completion of the its work item for the main thread to + see (==> it got stuck waiting.) -- sof 6/02. + */ + IF_DEBUG(scheduler, sched_belch("waiting for thread (%d)", tso->id)); + + APPEND_TO_RUN_QUEUE(tso); + // NB. Don't call THREAD_RUNNABLE() here, because the thread is + // bound and only runnable by *this* OS thread, so waking up other + // workers will just slow things down. - return waitThread_(m, initialCapability); + return waitThread_(m, initialCapability); } /* --------------------------------------------------------------------------- @@ -2243,18 +2011,6 @@ scheduleWaitThread(StgTSO* tso, /*[out]*/HaskellObj* ret, Capability *initialCap * * ------------------------------------------------------------------------ */ -#ifdef SMP -static void -term_handler(int sig STG_UNUSED) -{ - stat_workerStop(); - ACQUIRE_LOCK(&term_mutex); - await_death--; - RELEASE_LOCK(&term_mutex); - shutdownThread(); -} -#endif - void initScheduler(void) { @@ -2293,32 +2049,9 @@ initScheduler(void) * the scheduler. */ initMutex(&sched_mutex); initMutex(&term_mutex); - initMutex(&thread_id_mutex); - - initCondition(&thread_ready_cond); #endif -#if defined(SMP) - initCondition(&gc_pending_cond); -#endif - -#if defined(RTS_SUPPORTS_THREADS) ACQUIRE_LOCK(&sched_mutex); -#endif - - /* Install the SIGHUP handler */ -#if defined(SMP) - { - struct sigaction action,oact; - - action.sa_handler = term_handler; - sigemptyset(&action.sa_mask); - action.sa_flags = 0; - if (sigaction(SIGTERM, &action, &oact) != 0) { - barf("can't install TERM handler"); - } - } -#endif /* A capability holds the state a native thread needs in * order to execute STG code. At least one capability is @@ -2328,21 +2061,14 @@ initScheduler(void) #if defined(RTS_SUPPORTS_THREADS) /* start our haskell execution tasks */ -# if defined(SMP) - startTaskManager(RtsFlags.ParFlags.nNodes, taskStart); -# else startTaskManager(0,taskStart); -# endif #endif #if /* defined(SMP) ||*/ defined(PAR) initSparkPools(); #endif -#if defined(RTS_SUPPORTS_THREADS) RELEASE_LOCK(&sched_mutex); -#endif - } void @@ -2354,94 +2080,14 @@ exitScheduler( void ) shutting_down_scheduler = rtsTrue; } -/* ----------------------------------------------------------------------------- +/* ---------------------------------------------------------------------------- Managing the per-task allocation areas. Each capability comes with an allocation area. These are fixed-length block lists into which allocation can be done. ToDo: no support for two-space collection at the moment??? - -------------------------------------------------------------------------- */ - -/* ----------------------------------------------------------------------------- - * waitThread is the external interface for running a new computation - * and waiting for the result. - * - * In the non-SMP case, we create a new main thread, push it on the - * main-thread stack, and invoke the scheduler to run it. The - * scheduler will return when the top main thread on the stack has - * completed or died, and fill in the necessary fields of the - * main_thread structure. - * - * In the SMP case, we create a main thread as before, but we then - * create a new condition variable and sleep on it. When our new - * main thread has completed, we'll be woken up and the status/result - * will be in the main_thread struct. - * -------------------------------------------------------------------------- */ - -int -howManyThreadsAvail ( void ) -{ - int i = 0; - StgTSO* q; - for (q = run_queue_hd; q != END_TSO_QUEUE; q = q->link) - i++; - for (q = blocked_queue_hd; q != END_TSO_QUEUE; q = q->link) - i++; - for (q = sleeping_queue; q != END_TSO_QUEUE; q = q->link) - i++; - return i; -} - -void -finishAllThreads ( void ) -{ - do { - while (run_queue_hd != END_TSO_QUEUE) { - waitThread ( run_queue_hd, NULL, NULL ); - } - while (blocked_queue_hd != END_TSO_QUEUE) { - waitThread ( blocked_queue_hd, NULL, NULL ); - } - while (sleeping_queue != END_TSO_QUEUE) { - waitThread ( blocked_queue_hd, NULL, NULL ); - } - } while - (blocked_queue_hd != END_TSO_QUEUE || - run_queue_hd != END_TSO_QUEUE || - sleeping_queue != END_TSO_QUEUE); -} - -SchedulerStatus -waitThread(StgTSO *tso, /*out*/StgClosure **ret, Capability *initialCapability) -{ - StgMainThread *m; - SchedulerStatus stat; - - m = stgMallocBytes(sizeof(StgMainThread), "waitThread"); - m->tso = tso; - m->ret = ret; - m->stat = NoStatus; -#if defined(RTS_SUPPORTS_THREADS) -#if defined(THREADED_RTS) - initCondition(&m->bound_thread_cond); -#else - initCondition(&m->wakeup); -#endif -#endif - - /* see scheduleWaitThread() comment */ - ACQUIRE_LOCK(&sched_mutex); - m->link = main_threads; - main_threads = m; - - IF_DEBUG(scheduler, sched_belch("waiting for thread %d", tso->id)); - - stat = waitThread_(m,initialCapability); - - RELEASE_LOCK(&sched_mutex); - return stat; -} + ------------------------------------------------------------------------- */ static SchedulerStatus @@ -2452,150 +2098,36 @@ waitThread_(StgMainThread* m, Capability *initialCapability) // Precondition: sched_mutex must be held. IF_DEBUG(scheduler, sched_belch("new main thread (%d)", m->tso->id)); -#if defined(RTS_SUPPORTS_THREADS) && !defined(THREADED_RTS) - { // FIXME: does this still make sense? - // It's not for the threaded rts => SMP only - do { - waitCondition(&m->wakeup, &sched_mutex); - } while (m->stat == NoStatus); - } -#elif defined(GRAN) +#if defined(GRAN) /* GranSim specific init */ CurrentTSO = m->tso; // the TSO to run procStatus[MainProc] = Busy; // status of main PE CurrentProc = MainProc; // PE to run it on - - RELEASE_LOCK(&sched_mutex); schedule(m,initialCapability); #else - RELEASE_LOCK(&sched_mutex); schedule(m,initialCapability); - ACQUIRE_LOCK(&sched_mutex); ASSERT(m->stat != NoStatus); #endif stat = m->stat; #if defined(RTS_SUPPORTS_THREADS) -#if defined(THREADED_RTS) - closeCondition(&m->bound_thread_cond); -#else - closeCondition(&m->wakeup); -#endif + // Free the condition variable, returning it to the cache if possible. + if (!bound_cond_cache_full) { + bound_cond_cache = m->bound_thread_cond; + bound_cond_cache_full = 1; + } else { + closeCondition(&m->bound_thread_cond); + } #endif - IF_DEBUG(scheduler, fprintf(stderr, "== sched: main thread (%d) finished\n", - m->tso->id)); + IF_DEBUG(scheduler, sched_belch("main thread (%d) finished", m->tso->id)); stgFree(m); // Postcondition: sched_mutex still held return stat; } -//@node Run queue code, Garbage Collextion Routines, Suspend and Resume, Main scheduling code -//@subsection Run queue code - -#if 0 -/* - NB: In GranSim we have many run queues; run_queue_hd is actually a macro - unfolding to run_queue_hds[CurrentProc], thus CurrentProc is an - implicit global variable that has to be correct when calling these - fcts -- HWL -*/ - -/* Put the new thread on the head of the runnable queue. - * The caller of createThread better push an appropriate closure - * on this thread's stack before the scheduler is invoked. - */ -static /* inline */ void -add_to_run_queue(tso) -StgTSO* tso; -{ - ASSERT(tso!=run_queue_hd && tso!=run_queue_tl); - tso->link = run_queue_hd; - run_queue_hd = tso; - if (run_queue_tl == END_TSO_QUEUE) { - run_queue_tl = tso; - } -} - -/* Put the new thread at the end of the runnable queue. */ -static /* inline */ void -push_on_run_queue(tso) -StgTSO* tso; -{ - ASSERT(get_itbl((StgClosure *)tso)->type == TSO); - ASSERT(run_queue_hd!=NULL && run_queue_tl!=NULL); - ASSERT(tso!=run_queue_hd && tso!=run_queue_tl); - if (run_queue_hd == END_TSO_QUEUE) { - run_queue_hd = tso; - } else { - run_queue_tl->link = tso; - } - run_queue_tl = tso; -} - -/* - Should be inlined because it's used very often in schedule. The tso - argument is actually only needed in GranSim, where we want to have the - possibility to schedule *any* TSO on the run queue, irrespective of the - actual ordering. Therefore, if tso is not the nil TSO then we traverse - the run queue and dequeue the tso, adjusting the links in the queue. -*/ -//@cindex take_off_run_queue -static /* inline */ StgTSO* -take_off_run_queue(StgTSO *tso) { - StgTSO *t, *prev; - - /* - qetlaHbogh Qu' ngaSbogh ghomDaQ {tso} yIteq! - - if tso is specified, unlink that tso from the run_queue (doesn't have - to be at the beginning of the queue); GranSim only - */ - if (tso!=END_TSO_QUEUE) { - /* find tso in queue */ - for (t=run_queue_hd, prev=END_TSO_QUEUE; - t!=END_TSO_QUEUE && t!=tso; - prev=t, t=t->link) - /* nothing */ ; - ASSERT(t==tso); - /* now actually dequeue the tso */ - if (prev!=END_TSO_QUEUE) { - ASSERT(run_queue_hd!=t); - prev->link = t->link; - } else { - /* t is at beginning of thread queue */ - ASSERT(run_queue_hd==t); - run_queue_hd = t->link; - } - /* t is at end of thread queue */ - if (t->link==END_TSO_QUEUE) { - ASSERT(t==run_queue_tl); - run_queue_tl = prev; - } else { - ASSERT(run_queue_tl!=t); - } - t->link = END_TSO_QUEUE; - } else { - /* take tso from the beginning of the queue; std concurrent code */ - t = run_queue_hd; - if (t != END_TSO_QUEUE) { - run_queue_hd = t->link; - t->link = END_TSO_QUEUE; - if (run_queue_hd == END_TSO_QUEUE) { - run_queue_tl = END_TSO_QUEUE; - } - } - } - return t; -} - -#endif /* 0 */ - -//@node Garbage Collextion Routines, Blocking Queue Routines, Run queue code, Main scheduling code -//@subsection Garbage Collextion Routines - /* --------------------------------------------------------------------------- Where are the roots that we know about? @@ -2613,7 +2145,7 @@ take_off_run_queue(StgTSO *tso) { */ void -GetRoots(evac_fn evac) +GetRoots( evac_fn evac ) { #if defined(GRAN) { @@ -2665,25 +2197,6 @@ GetRoots(evac_fn evac) // mark the signal handlers (signals should be already blocked) markSignalHandlers(evac); #endif - - // main threads which have completed need to be retained until they - // are dealt with in the main scheduler loop. They won't be - // retained any other way: the GC will drop them from the - // all_threads list, so we have to be careful to treat them as roots - // here. - { - StgMainThread *m; - for (m = main_threads; m != NULL; m = m->link) { - switch (m->tso->what_next) { - case ThreadComplete: - case ThreadKilled: - evac((StgClosure **)&m->tso); - break; - default: - break; - } - } - } } /* ----------------------------------------------------------------------------- @@ -2754,8 +2267,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))); @@ -2775,7 +2288,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); @@ -2804,7 +2317,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, @@ -2818,9 +2331,6 @@ threadStackOverflow(StgTSO *tso) return dest; } -//@node Blocking Queue Routines, Exception Handling Routines, Garbage Collextion Routines, Main scheduling code -//@subsection Blocking Queue Routines - /* --------------------------------------------------------------------------- Wake up a queue that was blocked on some resource. ------------------------------------------------------------------------ */ @@ -2896,11 +2406,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) @@ -2914,8 +2424,8 @@ 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); + ((StgTSO *)bqe)->link = END_TSO_QUEUE; // debugging? + APPEND_TO_RUN_QUEUE((StgTSO *)bqe); THREAD_RUNNABLE(); unblockCount(bqe, node); /* reset blocking status after dumping event */ @@ -2945,7 +2455,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; } @@ -2959,9 +2469,10 @@ unblockOneLocked(StgTSO *tso) ASSERT(tso->why_blocked != NotBlocked); tso->why_blocked = NotBlocked; next = tso->link; - PUSH_ON_RUN_QUEUE(tso); + tso->link = END_TSO_QUEUE; + APPEND_TO_RUN_QUEUE(tso); THREAD_RUNNABLE(); - IF_DEBUG(scheduler,sched_belch("waking up thread %ld", tso->id)); + IF_DEBUG(scheduler,sched_belch("waking up thread %ld", (long)tso->id)); return next; } #endif @@ -2995,7 +2506,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)); @@ -3012,13 +2523,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++; @@ -3053,7 +2564,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))); } @@ -3065,7 +2576,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) @@ -3077,12 +2588,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 @@ -3102,7 +2613,6 @@ awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node) #else /* !GRAN && !PAR */ -#ifdef RTS_SUPPORTS_THREADS void awakenBlockedQueueNoLock(StgTSO *tso) { @@ -3110,7 +2620,6 @@ awakenBlockedQueueNoLock(StgTSO *tso) tso = unblockOneLocked(tso); } } -#endif void awakenBlockedQueue(StgTSO *tso) @@ -3123,9 +2632,6 @@ awakenBlockedQueue(StgTSO *tso) } #endif -//@node Exception Handling Routines, Debugging Routines, Blocking Queue Routines, Main scheduling code -//@subsection Exception Handling Routines - /* --------------------------------------------------------------------------- Interrupt execution - usually called inside a signal handler so it mustn't do anything fancy. @@ -3414,7 +2920,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 @@ -3458,6 +2964,7 @@ deleteThread(StgTSO *tso) raiseAsync(tso,NULL); } +#ifdef FORKPROCESS_PRIMOP_SUPPORTED static void deleteThreadImmediately(StgTSO *tso) { // for forkProcess only: @@ -3466,13 +2973,15 @@ deleteThreadImmediately(StgTSO *tso) if (tso->what_next == ThreadComplete || tso->what_next == ThreadKilled) { return; } -#if defined(RTS_SUPPORTS_THREADS) - if (tso->why_blocked != BlockedOnCCall - && tso->why_blocked != BlockedOnCCall_NoUnblockExc) -#endif + + if (tso->why_blocked != BlockedOnCCall && + tso->why_blocked != BlockedOnCCall_NoUnblockExc) { unblockThread(tso); + } + tso->what_next = ThreadKilled; } +#endif void raiseAsyncWithLock(StgTSO *tso, StgClosure *exception) @@ -3496,7 +3005,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); @@ -3611,9 +3120,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); ); @@ -3631,7 +3140,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 @@ -3653,6 +3163,77 @@ 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 CATCH_FRAME: + tso->sp = p; + return CATCH_FRAME; + + case STOP_FRAME: + tso->sp = p; + return STOP_FRAME; + + default: + 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 @@ -3709,7 +3290,7 @@ static void detectBlackHoles( void ) { StgTSO *tso = all_threads; - StgClosure *frame; + StgPtr frame; StgClosure *blocked_on; StgRetInfoTable *info; @@ -3725,10 +3306,10 @@ detectBlackHoles( void ) } blocked_on = tso->block_info.closure; - frame = (StgClosure *)tso->sp; + frame = tso->sp; while(1) { - info = get_ret_itbl(frame); + info = get_ret_itbl((StgClosure *)frame); switch (info->i.type) { case UPDATE_FRAME: if (((StgUpdateFrame *)frame)->updatee == blocked_on) { @@ -3741,7 +3322,7 @@ detectBlackHoles( void ) goto done; } - frame = (StgClosure *) ((StgUpdateFrame *)frame + 1); + frame = (StgPtr)((StgUpdateFrame *)frame + 1); continue; case STOP_FRAME: @@ -3749,21 +3330,18 @@ detectBlackHoles( void ) // normal stack frames; do nothing except advance the pointer default: - (StgPtr)frame += stack_frame_sizeW(frame); + frame += stack_frame_sizeW((StgClosure *)frame); } } done: ; } } -//@node Debugging Routines, Index, Exception Handling Routines, Main scheduling code -//@subsection Debugging Routines - -/* ----------------------------------------------------------------------------- +/* ---------------------------------------------------------------------------- * Debugging: why is a thread blocked * [Also provides useful information when debugging threaded programs * at the Haskell source code level, so enable outside of DEBUG. --sof 7/02] - -------------------------------------------------------------------------- */ + ------------------------------------------------------------------------- */ static void @@ -3771,50 +3349,48 @@ 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) 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 -#if defined(RTS_SUPPORTS_THREADS) 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; -#endif default: barf("printThreadBlockage: strange tso->why_blocked: %d for TSO %d (%d)", tso->why_blocked, tso->id, tso); @@ -3827,10 +3403,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); @@ -3848,23 +3424,23 @@ printAllThreads(void) 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); + debugBelch("\tthread %d @ %p ", t->id, (void *)t); label = lookupThreadLabel(t->id); - if (label) fprintf(stderr,"[\"%s\"] ",(char *)label); + if (label) debugBelch("[\"%s\"] ",(char *)label); printThreadStatus(t); - fprintf(stderr,"\n"); + debugBelch("\n"); } } @@ -3873,7 +3449,6 @@ printAllThreads(void) /* Print a whole blocking queue attached to node (debugging only). */ -//@cindex print_bq # if defined(PAR) void print_bq (StgClosure *node) @@ -3882,7 +3457,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 */ @@ -3922,18 +3497,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" : @@ -3945,7 +3520,7 @@ print_bqe (StgBlockingQueueElement *bqe) break; } } /* for */ - fputc('\n', stderr); + debugBelch("\n"); } # elif defined(GRAN) void @@ -3963,7 +3538,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); /* @@ -3983,11 +3558,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" : @@ -3999,7 +3574,7 @@ print_bq (StgClosure *node) break; } } /* for */ - fputc('\n', stderr); + debugBelch("\n"); } #else /* @@ -4016,9 +3591,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 @@ -4044,40 +3619,15 @@ 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); } #endif /* DEBUG */ - - -//@node Index, , Debugging Routines, Main scheduling code -//@subsection Index - -//@index -//* StgMainThread:: @cindex\s-+StgMainThread -//* awaken_blocked_queue:: @cindex\s-+awaken_blocked_queue -//* blocked_queue_hd:: @cindex\s-+blocked_queue_hd -//* blocked_queue_tl:: @cindex\s-+blocked_queue_tl -//* context_switch:: @cindex\s-+context_switch -//* createThread:: @cindex\s-+createThread -//* gc_pending_cond:: @cindex\s-+gc_pending_cond -//* initScheduler:: @cindex\s-+initScheduler -//* interrupted:: @cindex\s-+interrupted -//* next_thread_id:: @cindex\s-+next_thread_id -//* print_bq:: @cindex\s-+print_bq -//* run_queue_hd:: @cindex\s-+run_queue_hd -//* run_queue_tl:: @cindex\s-+run_queue_tl -//* sched_mutex:: @cindex\s-+sched_mutex -//* schedule:: @cindex\s-+schedule -//* take_off_run_queue:: @cindex\s-+take_off_run_queue -//* term_mutex:: @cindex\s-+term_mutex -//@end index