X-Git-Url: http://git.megacz.com/?a=blobdiff_plain;f=ghc%2Frts%2FSchedule.c;h=8a21d7b58c7a782b4cc59441603dd5ca790cccf3;hb=ab3d1f285cef784138d99e70f7359ea6e67f6c25;hp=4017c3bbf01e599118cccf0d8aa96040e02ba0ab;hpb=6e817437ecf7820f86564f461d8a12bf432a42fb;p=ghc-hetmet.git diff --git a/ghc/rts/Schedule.c b/ghc/rts/Schedule.c index 4017c3b..8a21d7b 100644 --- a/ghc/rts/Schedule.c +++ b/ghc/rts/Schedule.c @@ -1,7 +1,6 @@ /* --------------------------------------------------------------------------- - * $Id: Schedule.c,v 1.189 2004/02/27 12:39:16 simonmar Exp $ * - * (c) The GHC Team, 1998-2003 + * (c) The GHC Team, 1998-2004 * * Scheduler * @@ -10,7 +9,7 @@ * * WAY Name CPP flag What's it for * -------------------------------------- - * mp GUM PAR Parallel execution on a distrib. memory machine + * mp GUM PARALLEL_HASKELL Parallel execution on a distrib. memory machine * s SMP SMP Parallel execution on a shared memory machine * mg GranSim GRAN Simulation of parallel execution * md GUM/GdH DIST Distributed execution (based on GUM) @@ -42,28 +41,31 @@ #include "SchedAPI.h" #include "RtsUtils.h" #include "RtsFlags.h" +#include "BlockAlloc.h" +#include "OSThreads.h" #include "Storage.h" #include "StgRun.h" -#include "StgStartup.h" #include "Hooks.h" #define COMPILING_SCHEDULER #include "Schedule.h" #include "StgMiscClosures.h" -#include "Storage.h" #include "Interpreter.h" #include "Exception.h" #include "Printer.h" #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" #endif -#if defined(GRAN) || defined(PAR) +#if defined(GRAN) || defined(PARALLEL_HASKELL) # include "GranSimRts.h" # include "GranSim.h" # include "ParallelRts.h" @@ -74,7 +76,6 @@ #endif #include "Sparks.h" #include "Capability.h" -#include "OSThreads.h" #include "Task.h" #ifdef HAVE_SYS_TYPES_H @@ -92,6 +93,12 @@ #include #endif +// Turn off inlining when debugging - it obfuscates things +#ifdef DEBUG +# undef STATIC_INLINE +# define STATIC_INLINE static +#endif + #ifdef THREADED_RTS #define USED_IN_THREADED_RTS #else @@ -109,9 +116,6 @@ */ StgMainThread *main_threads = NULL; -/* Thread queues. - * Locks required: sched_mutex. - */ #if defined(GRAN) StgTSO* ActiveTSO = NULL; /* for assigning system costs; GranSim-Light only */ @@ -134,14 +138,23 @@ StgTSO *ccalling_threadss[MAX_PROC]; #else /* !GRAN */ +/* Thread queues. + * Locks required: sched_mutex. + */ StgTSO *run_queue_hd = NULL; StgTSO *run_queue_tl = NULL; StgTSO *blocked_queue_hd = NULL; StgTSO *blocked_queue_tl = NULL; +StgTSO *blackhole_queue = NULL; StgTSO *sleeping_queue = NULL; /* perhaps replace with a hash table? */ #endif +/* The blackhole_queue should be checked for threads to wake up. See + * Schedule.h for more thorough comment. + */ +rtsBool blackholes_need_checking = rtsFalse; + /* Linked list of all threads. * Used for detecting garbage collected threads. */ @@ -153,15 +166,16 @@ StgTSO *all_threads = NULL; */ static StgTSO *suspended_ccalling_threads; -static StgTSO *threadStackOverflow(StgTSO *tso); - /* KH: The following two flags are shared memory locations. There is no need to lock them, since they are only unset at the end of a scheduler operation. */ /* flag set by signal handler to precipitate a context switch */ -nat context_switch = 0; +int context_switch = 0; + +/* flag that tracks whether we have done any execution in this time slice. */ +nat recent_activity = ACTIVITY_YES; /* if this flag is set as well, give up execution */ rtsBool interrupted = rtsFalse; @@ -201,8 +215,6 @@ StgTSO *CurrentTSO; */ StgTSO dummy_tso; -static rtsBool ready_to_gc; - /* * Set to TRUE when entering a shutdown state (via shutdownHaskellAndExit()) -- * in an MT setting, needed to signal that a worker thread shouldn't hang around @@ -210,13 +222,6 @@ static rtsBool ready_to_gc; */ static rtsBool shutting_down_scheduler = rtsFalse; -void addToBlockedQueue ( StgTSO *tso ); - -static void schedule ( StgMainThread *mainThread, Capability *initialCapability ); - void interruptStgRts ( void ); - -static void detectBlackHoles ( void ); - #if defined(RTS_SUPPORTS_THREADS) /* ToDo: carefully document the invariants that go together * with these synchronisation objects. @@ -226,7 +231,7 @@ Mutex term_mutex = INIT_MUTEX_VAR; #endif /* RTS_SUPPORTS_THREADS */ -#if defined(PAR) +#if defined(PARALLEL_HASKELL) StgTSO *LastTSO; rtsTime TimeOfLastYield; rtsBool emitSchedule = rtsTrue; @@ -234,6 +239,7 @@ rtsBool emitSchedule = rtsTrue; #if DEBUG static char *whatNext_strs[] = { + "(unknown)", "ThreadRunGHC", "ThreadInterpret", "ThreadKilled", @@ -242,7 +248,66 @@ static char *whatNext_strs[] = { }; #endif -#if defined(PAR) +/* ----------------------------------------------------------------------------- + * static function prototypes + * -------------------------------------------------------------------------- */ + +#if defined(RTS_SUPPORTS_THREADS) +static void taskStart(void); +#endif + +static void schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, + Capability *initialCapability ); + +// +// These function all encapsulate parts of the scheduler loop, and are +// abstracted only to make the structure and control flow of the +// scheduler clearer. +// +static void schedulePreLoop(void); +static void scheduleStartSignalHandlers(void); +static void scheduleCheckBlockedThreads(void); +static void scheduleCheckBlackHoles(void); +static void scheduleDetectDeadlock(void); +#if defined(GRAN) +static StgTSO *scheduleProcessEvent(rtsEvent *event); +#endif +#if defined(PARALLEL_HASKELL) +static StgTSO *scheduleSendPendingMessages(void); +static void scheduleActivateSpark(void); +static rtsBool scheduleGetRemoteWork(rtsBool *receivedFinish); +#endif +#if defined(PAR) || defined(GRAN) +static void scheduleGranParReport(void); +#endif +static void schedulePostRunThread(void); +static rtsBool scheduleHandleHeapOverflow( Capability *cap, StgTSO *t ); +static void scheduleHandleStackOverflow( StgTSO *t); +static rtsBool scheduleHandleYield( StgTSO *t, nat prev_what_next ); +static void scheduleHandleThreadBlocked( StgTSO *t ); +static rtsBool scheduleHandleThreadFinished( StgMainThread *mainThread, + Capability *cap, StgTSO *t ); +static rtsBool scheduleDoHeapProfile(rtsBool ready_to_gc); +static void scheduleDoGC(rtsBool force_major); + +static void unblockThread(StgTSO *tso); +static rtsBool checkBlackHoles(void); +static SchedulerStatus waitThread_(/*out*/StgMainThread* m, + Capability *initialCapability + ); +static void scheduleThread_ (StgTSO* tso); +static void AllRoots(evac_fn evac); + +static StgTSO *threadStackOverflow(StgTSO *tso); + +static void raiseAsync_(StgTSO *tso, StgClosure *exception, + rtsBool stop_at_atomically); + +static void printThreadBlockage(StgTSO *tso); +static void printThreadStatus(StgTSO *tso); +void printThreadQueue(StgTSO *tso); + +#if defined(PARALLEL_HASKELL) StgTSO * createSparkThread(rtsSpark spark); StgTSO * activateSpark (rtsSpark spark); #endif @@ -252,35 +317,57 @@ StgTSO * activateSpark (rtsSpark spark); * ------------------------------------------------------------------------- */ #if defined(RTS_SUPPORTS_THREADS) -static rtsBool startingWorkerThread = rtsFalse; +static nat startingWorkerThread = 0; -static void taskStart(void); static void taskStart(void) { ACQUIRE_LOCK(&sched_mutex); + startingWorkerThread--; schedule(NULL,NULL); + taskStop(); RELEASE_LOCK(&sched_mutex); } void startSchedulerTaskIfNecessary(void) { - if(run_queue_hd != END_TSO_QUEUE - || blocked_queue_hd != END_TSO_QUEUE - || sleeping_queue != END_TSO_QUEUE) - { - if(!startingWorkerThread) - { // we don't want to start another worker thread - // just because the last one hasn't yet reached the - // "waiting for capability" state - startingWorkerThread = rtsTrue; - startTask(taskStart); + if ( !EMPTY_RUN_QUEUE() + && !shutting_down_scheduler // not if we're shutting down + && startingWorkerThread==0) + { + // we don't want to start another worker thread + // just because the last one hasn't yet reached the + // "waiting for capability" state + startingWorkerThread++; + if (!maybeStartNewWorker(taskStart)) { + startingWorkerThread--; + } } - } } #endif +/* ----------------------------------------------------------------------------- + * Putting a thread on the run queue: different scheduling policies + * -------------------------------------------------------------------------- */ + +STATIC_INLINE void +addToRunQueue( StgTSO *t ) +{ +#if defined(PARALLEL_HASKELL) + if (RtsFlags.ParFlags.doFairScheduling) { + // this does round-robin scheduling; good for concurrency + APPEND_TO_RUN_QUEUE(t); + } else { + // this does unfair scheduling; good for parallelism + PUSH_ON_RUN_QUEUE(t); + } +#else + // this does round-robin scheduling; good for concurrency + APPEND_TO_RUN_QUEUE(t); +#endif +} + /* --------------------------------------------------------------------------- Main scheduling loop. @@ -316,18 +403,17 @@ startSchedulerTaskIfNecessary(void) This is not the ugliest code you could imagine, but it's bloody close. ------------------------------------------------------------------------ */ + 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; -#elif defined(PAR) - StgSparkPool *pool; - rtsSpark spark; +#elif defined(PARALLEL_HASKELL) StgTSO *tso; GlobalTaskId pe; rtsBool receivedFinish = rtsFalse; @@ -335,67 +421,50 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, nat tp_size, sp_size; // stats only # endif #endif - rtsBool was_interrupted = rtsFalse; - StgTSOWhatNext prev_what_next; + nat prev_what_next; + rtsBool ready_to_gc; // Pre-condition: sched_mutex is held. - -#if defined(RTS_SUPPORTS_THREADS) - // - // in the threaded case, the capability is either passed in via the - // initialCapability parameter, or initialized inside the scheduler - // loop - // - IF_DEBUG(scheduler, - sched_belch("### NEW SCHEDULER LOOP (main thr: %p, cap: %p)", - mainThread, initialCapability); - ); -#else + // We might have a capability, passed in as initialCapability. + cap = initialCapability; + +#if !defined(RTS_SUPPORTS_THREADS) // simply initialise it in the non-threaded case grabCapability(&cap); #endif -#if defined(GRAN) - /* set up first event to get things going */ - /* ToDo: assign costs for system setup and init MainTSO ! */ - new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc], - ContinueThread, - CurrentTSO, (StgClosure*)NULL, (rtsSpark*)NULL); - - IF_DEBUG(gran, - fprintf(stderr, "GRAN: Init CurrentTSO (in schedule) = %p\n", CurrentTSO); - G_TSO(CurrentTSO, 5)); - - if (RtsFlags.GranFlags.Light) { - /* Save current time; GranSim Light only */ - CurrentTSO->gran.clock = CurrentTime[CurrentProc]; - } - - event = get_next_event(); - - while (event!=(rtsEvent*)NULL) { - /* Choose the processor with the next event */ - CurrentProc = event->proc; - CurrentTSO = event->tso; + IF_DEBUG(scheduler, + sched_belch("### NEW SCHEDULER LOOP (main thr: %p, cap: %p)", + mainThread, initialCapability); + ); -#elif defined(PAR) + schedulePreLoop(); - while (!receivedFinish) { /* set by processMessages */ - /* when receiving PP_FINISH message */ + // ----------------------------------------------------------- + // Scheduler loop starts here: -#else // everything except GRAN and PAR +#if defined(PARALLEL_HASKELL) +#define TERMINATION_CONDITION (!receivedFinish) +#elif defined(GRAN) +#define TERMINATION_CONDITION ((event = get_next_event()) != (rtsEvent*)NULL) +#else +#define TERMINATION_CONDITION rtsTrue +#endif - while (1) { + while (TERMINATION_CONDITION) { +#if defined(GRAN) + /* Choose the processor with the next event */ + CurrentProc = event->proc; + CurrentTSO = event->tso; #endif - IF_DEBUG(scheduler, printAllThreads()); - #if defined(RTS_SUPPORTS_THREADS) // Yield the capability to higher-priority tasks if necessary. // if (cap != NULL) { - yieldCapability(&cap); + yieldCapability(&cap, + mainThread ? &mainThread->bound_thread_cond : NULL ); } // If we do not currently hold a capability, we wait for one @@ -408,257 +477,509 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, // We now have a capability... #endif - // - // If we're interrupted (the user pressed ^C, or some other - // termination condition occurred), kill all the currently running - // threads. - // - if (interrupted) { - IF_DEBUG(scheduler, sched_belch("interrupted")); - interrupted = rtsFalse; - was_interrupted = rtsTrue; -#if defined(RTS_SUPPORTS_THREADS) - // In the threaded RTS, deadlock detection doesn't work, - // so just exit right away. - prog_belch("interrupted"); - releaseCapability(cap); - RELEASE_LOCK(&sched_mutex); - shutdownHaskellAndExit(EXIT_SUCCESS); -#else - deleteAllThreads(); +#if 0 /* extra sanity checking */ + { + StgMainThread *m; + for (m = main_threads; m != NULL; m = m->link) { + ASSERT(get_itbl(m->tso)->type == TSO); + } + } #endif + + // Check whether we have re-entered the RTS from Haskell without + // going via suspendThread()/resumeThread (i.e. a 'safe' foreign + // call). + if (cap->r.rInHaskell) { + errorBelch("schedule: re-entered unsafely.\n" + " Perhaps a 'foreign import unsafe' should be 'safe'?"); + stg_exit(1); } // - // 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... + // Test for interruption. If interrupted==rtsTrue, then either + // we received a keyboard interrupt (^C), or the scheduler is + // trying to shut down all the tasks (shutting_down_scheduler) in + // the threaded RTS. // -#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); - 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(&m->bound_thread_cond); - continue; - } - } + if (interrupted) { + if (shutting_down_scheduler) { + IF_DEBUG(scheduler, sched_belch("shutting down")); + releaseCapability(cap); + if (mainThread) { + mainThread->stat = Interrupted; + mainThread->ret = NULL; + } + return; + } else { + IF_DEBUG(scheduler, sched_belch("interrupted")); + deleteAllThreads(); } } - -#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. - */ +#if defined(not_yet) && defined(SMP) + // + // Top up the run queue from our spark pool. We try to make the + // number of threads in the run queue equal to the number of + // free capabilities. + // { - 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; + StgClosure *spark; + if (EMPTY_RUN_QUEUE()) { + spark = findSpark(rtsFalse); + if (spark == NULL) { + break; /* no more sparks in the pool */ + } else { + createSparkThread(spark); + IF_DEBUG(scheduler, + sched_belch("==^^ turning spark of closure %p into a thread", + (StgClosure *)spark)); + } } - } } -#endif +#endif // SMP + scheduleStartSignalHandlers(); -#if defined(RTS_USER_SIGNALS) - // check for signals each time around the scheduler - if (signals_pending()) { - RELEASE_LOCK(&sched_mutex); /* ToDo: kill */ - startSignalHandlers(); - ACQUIRE_LOCK(&sched_mutex); - } -#endif + // Only check the black holes here if we've nothing else to do. + // During normal execution, the black hole list only gets checked + // at GC time, to avoid repeatedly traversing this possibly long + // list each time around the scheduler. + if (EMPTY_RUN_QUEUE()) { scheduleCheckBlackHoles(); } - /* 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() ); - } - /* we can be interrupted while waiting for I/O... */ - if (interrupted) continue; + scheduleCheckBlockedThreads(); - /* - * Detect deadlock: when we have no threads to run, there are no - * threads waiting on I/O or sleeping, and all the other tasks are - * waiting for work, we must have a deadlock of some description. - * - * We first try to find threads blocked on themselves (ie. black - * holes), and generate NonTermination exceptions where necessary. - * - * If no threads are black holed, we have a deadlock situation, so - * inform all the main threads. - */ -#if !defined(PAR) && !defined(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. - GarbageCollect(GetRoots,rtsTrue); + scheduleDetectDeadlock(); - if ( !EMPTY_RUN_QUEUE() ) { goto not_deadlocked; } + // Normally, the only way we can get here with no threads to + // run is if a keyboard interrupt received during + // scheduleCheckBlockedThreads() or scheduleDetectDeadlock(). + // Additionally, it is not fatal for the + // threaded RTS to reach here with no threads to run. + // + // win32: might be here due to awaitEvent() being abandoned + // as a result of a console event having been delivered. + if ( EMPTY_RUN_QUEUE() ) { +#if !defined(RTS_SUPPORTS_THREADS) && !defined(mingw32_HOST_OS) + ASSERT(interrupted); +#endif + continue; // nothing to do + } - IF_DEBUG(scheduler, - sched_belch("still deadlocked, checking for black holes...")); - detectBlackHoles(); +#if defined(PARALLEL_HASKELL) + scheduleSendPendingMessages(); + if (EMPTY_RUN_QUEUE() && scheduleActivateSpark()) + continue; - if ( !EMPTY_RUN_QUEUE() ) { goto not_deadlocked; } +#if defined(SPARKS) + ASSERT(next_fish_to_send_at==0); // i.e. no delayed fishes left! +#endif -#if defined(RTS_USER_SIGNALS) - /* If we have user-installed signal handlers, then wait - * for signals to arrive rather then bombing out with a - * deadlock. - */ - if ( anyUserHandlers() ) { - IF_DEBUG(scheduler, - sched_belch("still deadlocked, waiting for signals...")); + /* If we still have no work we need to send a FISH to get a spark + from another PE */ + if (EMPTY_RUN_QUEUE()) { + if (!scheduleGetRemoteWork(&receivedFinish)) continue; + ASSERT(rtsFalse); // should not happen at the moment + } + // from here: non-empty run queue. + // TODO: merge above case with this, only one call processMessages() ! + if (PacketsWaiting()) { /* process incoming messages, if + any pending... only in else + because getRemoteWork waits for + messages as well */ + receivedFinish = processMessages(); + } +#endif - awaitUserSignals(); +#if defined(GRAN) + scheduleProcessEvent(event); +#endif - // we might be interrupted... - if (interrupted) { continue; } + // + // Get a thread to run + // + ASSERT(run_queue_hd != END_TSO_QUEUE); + POP_RUN_QUEUE(t); - if (signals_pending()) { - RELEASE_LOCK(&sched_mutex); - startSignalHandlers(); - ACQUIRE_LOCK(&sched_mutex); - } - ASSERT(!EMPTY_RUN_QUEUE()); - goto not_deadlocked; - } +#if defined(GRAN) || defined(PAR) + scheduleGranParReport(); // some kind of debuging output +#else + // Sanity check the thread we're about to run. This can be + // expensive if there is lots of thread switching going on... + IF_DEBUG(sanity,checkTSO(t)); #endif - /* Probably a real deadlock. Send the current main thread the - * Deadlock exception (or in the SMP build, send *all* main - * threads the deadlock exception, since none of them can make - * progress). - */ +#if defined(RTS_SUPPORTS_THREADS) + // Check whether we can run this thread in the current task. + // If not, we have to pass our capability to the right task. + { + StgMainThread *m = t->main; + + if(m) + { + if(m == mainThread) { - StgMainThread *m; - 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); - break; - default: - barf("deadlock: main thread blocked in a strange way"); - } + IF_DEBUG(scheduler, + sched_belch("### Running thread %d in bound thread", t->id)); + // yes, the Haskell thread is bound to the current native thread + } + else + { + IF_DEBUG(scheduler, + sched_belch("### thread %d bound to another OS thread", t->id)); + // no, bound to a different Haskell thread: pass to that thread + PUSH_ON_RUN_QUEUE(t); + continue; + } + } + else + { + if(mainThread != NULL) + // The thread we want to run is unbound. + { + IF_DEBUG(scheduler, + sched_belch("### this OS thread cannot run thread %d", t->id)); + // no, the current native thread is bound to a different + // Haskell thread, so pass it to any worker thread + PUSH_ON_RUN_QUEUE(t); + continue; } + } } - not_deadlocked: - -#elif defined(RTS_SUPPORTS_THREADS) - // ToDo: add deadlock detection in threaded RTS -#elif defined(PAR) - // ToDo: add deadlock detection in GUM (similar to SMP) -- HWL #endif -#if defined(RTS_SUPPORTS_THREADS) - if ( EMPTY_RUN_QUEUE() ) { - continue; // nothing to do - } -#endif + cap->r.rCurrentTSO = t; + + /* context switches are now initiated by the timer signal, unless + * the user specified "context switch as often as possible", with + * +RTS -C0 + */ + if ((RtsFlags.ConcFlags.ctxtSwitchTicks == 0 + && (run_queue_hd != END_TSO_QUEUE + || blocked_queue_hd != END_TSO_QUEUE + || sleeping_queue != END_TSO_QUEUE))) + context_switch = 1; -#if defined(GRAN) - if (RtsFlags.GranFlags.Light) - GranSimLight_enter_system(event, &ActiveTSO); // adjust ActiveTSO etc +run_thread: - /* adjust time based on time-stamp */ - if (event->time > CurrentTime[CurrentProc] && - event->evttype != ContinueThread) - CurrentTime[CurrentProc] = event->time; - - /* Deal with the idle PEs (may issue FindWork or MoveSpark events) */ - if (!RtsFlags.GranFlags.Light) - handleIdlePEs(); + RELEASE_LOCK(&sched_mutex); - IF_DEBUG(gran, fprintf(stderr, "GRAN: switch by event-type\n")); + IF_DEBUG(scheduler, sched_belch("-->> running thread %ld %s ...", + (long)t->id, whatNext_strs[t->what_next])); + +#if defined(PROFILING) + startHeapProfTimer(); +#endif + + // ---------------------------------------------------------------------- + // Run the current thread + + prev_what_next = t->what_next; + + errno = t->saved_errno; + cap->r.rInHaskell = rtsTrue; + + recent_activity = ACTIVITY_YES; + + switch (prev_what_next) { + + case ThreadKilled: + case ThreadComplete: + /* Thread already finished, return to scheduler. */ + ret = ThreadFinished; + break; + + case ThreadRunGHC: + ret = StgRun((StgFunPtr) stg_returnToStackTop, &cap->r); + break; + + case ThreadInterpret: + ret = interpretBCO(cap); + break; + + default: + barf("schedule: invalid what_next field"); + } + +#if defined(SMP) + // in SMP mode, we might return with a different capability than + // we started with, if the Haskell thread made a foreign call. So + // let's find out what our current Capability is: + cap = myCapability(); +#endif + + // We have run some Haskell code: there might be blackhole-blocked + // threads to wake up now. + if ( blackhole_queue != END_TSO_QUEUE ) { + blackholes_need_checking = rtsTrue; + } + + cap->r.rInHaskell = rtsFalse; + + // The TSO might have moved, eg. if it re-entered the RTS and a GC + // happened. So find the new location: + t = cap->r.rCurrentTSO; + + // And save the current errno in this thread. + t->saved_errno = errno; + + // ---------------------------------------------------------------------- + + /* Costs for the scheduler are assigned to CCS_SYSTEM */ +#if defined(PROFILING) + stopHeapProfTimer(); + CCCS = CCS_SYSTEM; +#endif + + ACQUIRE_LOCK(&sched_mutex); + +#if defined(RTS_SUPPORTS_THREADS) + IF_DEBUG(scheduler,debugBelch("sched (task %p): ", osThreadId());); +#elif !defined(GRAN) && !defined(PARALLEL_HASKELL) + IF_DEBUG(scheduler,debugBelch("sched: ");); +#endif + + schedulePostRunThread(); + + ready_to_gc = rtsFalse; + + switch (ret) { + case HeapOverflow: + ready_to_gc = scheduleHandleHeapOverflow(cap,t); + break; + + case StackOverflow: + scheduleHandleStackOverflow(t); + break; + + case ThreadYielding: + if (scheduleHandleYield(t, prev_what_next)) { + // shortcut for switching between compiler/interpreter: + goto run_thread; + } + break; + + case ThreadBlocked: + scheduleHandleThreadBlocked(t); + break; + + case ThreadFinished: + if (scheduleHandleThreadFinished(mainThread, cap, t)) return;; + break; + + default: + barf("schedule: invalid thread return code %d", (int)ret); + } + + if (scheduleDoHeapProfile(ready_to_gc)) { ready_to_gc = rtsFalse; } + if (ready_to_gc) { scheduleDoGC(rtsFalse); } + } /* end of while() */ + + IF_PAR_DEBUG(verbose, + debugBelch("== Leaving schedule() after having received Finish\n")); +} + +/* ---------------------------------------------------------------------------- + * Setting up the scheduler loop + * ASSUMES: sched_mutex + * ------------------------------------------------------------------------- */ + +static void +schedulePreLoop(void) +{ +#if defined(GRAN) + /* set up first event to get things going */ + /* ToDo: assign costs for system setup and init MainTSO ! */ + new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc], + ContinueThread, + CurrentTSO, (StgClosure*)NULL, (rtsSpark*)NULL); + + IF_DEBUG(gran, + debugBelch("GRAN: Init CurrentTSO (in schedule) = %p\n", + CurrentTSO); + G_TSO(CurrentTSO, 5)); + + if (RtsFlags.GranFlags.Light) { + /* Save current time; GranSim Light only */ + CurrentTSO->gran.clock = CurrentTime[CurrentProc]; + } +#endif +} + +/* ---------------------------------------------------------------------------- + * Start any pending signal handlers + * ASSUMES: sched_mutex + * ------------------------------------------------------------------------- */ + +static void +scheduleStartSignalHandlers(void) +{ +#if defined(RTS_USER_SIGNALS) && !defined(RTS_SUPPORTS_THREADS) + if (signals_pending()) { + RELEASE_LOCK(&sched_mutex); /* ToDo: kill */ + startSignalHandlers(); + ACQUIRE_LOCK(&sched_mutex); + } +#endif +} + +/* ---------------------------------------------------------------------------- + * Check for blocked threads that can be woken up. + * ASSUMES: sched_mutex + * ------------------------------------------------------------------------- */ + +static void +scheduleCheckBlockedThreads(void) +{ + // + // 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) + // We shouldn't be here... + barf("schedule: awaitEvent() in threaded RTS"); +#else + awaitEvent( EMPTY_RUN_QUEUE() && !blackholes_need_checking ); +#endif + } +} + + +/* ---------------------------------------------------------------------------- + * Check for threads blocked on BLACKHOLEs that can be woken up + * ASSUMES: sched_mutex + * ------------------------------------------------------------------------- */ +static void +scheduleCheckBlackHoles( void ) +{ + if ( blackholes_need_checking ) + { + checkBlackHoles(); + blackholes_need_checking = rtsFalse; + } +} + +/* ---------------------------------------------------------------------------- + * Detect deadlock conditions and attempt to resolve them. + * ASSUMES: sched_mutex + * ------------------------------------------------------------------------- */ + +static void +scheduleDetectDeadlock() +{ + +#if defined(PARALLEL_HASKELL) + // ToDo: add deadlock detection in GUM (similar to SMP) -- HWL + return; +#endif + + /* + * Detect deadlock: when we have no threads to run, there are no + * threads blocked, waiting for I/O, or sleeping, and all the + * other tasks are waiting for work, we must have a deadlock of + * some description. + */ + if ( EMPTY_THREAD_QUEUES() ) + { +#if defined(RTS_SUPPORTS_THREADS) + /* + * In the threaded RTS, we only check for deadlock if there + * has been no activity in a complete timeslice. This means + * we won't eagerly start a full GC just because we don't have + * any threads to run currently. + */ + if (recent_activity != ACTIVITY_INACTIVE) return; +#endif + + 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 unreachable and will therefore be sent an + // exception. Any threads thus released will be immediately + // runnable. + + scheduleDoGC( rtsTrue/*force major GC*/ ); + recent_activity = ACTIVITY_DONE_GC; + if ( !EMPTY_RUN_QUEUE() ) return; + +#if defined(RTS_USER_SIGNALS) && !defined(RTS_SUPPORTS_THREADS) + /* If we have user-installed signal handlers, then wait + * for signals to arrive rather then bombing out with a + * deadlock. + */ + if ( anyUserHandlers() ) { + IF_DEBUG(scheduler, + sched_belch("still deadlocked, waiting for signals...")); + + awaitUserSignals(); + + if (signals_pending()) { + RELEASE_LOCK(&sched_mutex); + startSignalHandlers(); + ACQUIRE_LOCK(&sched_mutex); + } + + // either we have threads to run, or we were interrupted: + ASSERT(!EMPTY_RUN_QUEUE() || interrupted); + } +#endif + +#if !defined(RTS_SUPPORTS_THREADS) + /* Probably a real deadlock. Send the current main thread the + * Deadlock exception (or in the SMP build, send *all* main + * threads the deadlock exception, since none of them can make + * progress). + */ + { + StgMainThread *m; + m = main_threads; + switch (m->tso->why_blocked) { + case BlockedOnSTM: + case BlockedOnBlackHole: + case BlockedOnException: + case BlockedOnMVar: + raiseAsync(m->tso, (StgClosure *)NonTermination_closure); + return; + default: + barf("deadlock: main thread blocked in a strange way"); + } + } +#endif + } +} + +/* ---------------------------------------------------------------------------- + * Process an event (GRAN only) + * ------------------------------------------------------------------------- */ + +#if defined(GRAN) +static StgTSO * +scheduleProcessEvent(rtsEvent *event) +{ + StgTSO *t; + + if (RtsFlags.GranFlags.Light) + GranSimLight_enter_system(event, &ActiveTSO); // adjust ActiveTSO etc + + /* adjust time based on time-stamp */ + if (event->time > CurrentTime[CurrentProc] && + event->evttype != ContinueThread) + CurrentTime[CurrentProc] = event->time; + + /* Deal with the idle PEs (may issue FindWork or MoveSpark events) */ + if (!RtsFlags.GranFlags.Light) + handleIdlePEs(); + + 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); @@ -666,25 +987,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 } @@ -741,14 +1062,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) @@ -757,7 +1078,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; @@ -765,7 +1086,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 @@ -774,18 +1095,54 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, DumpGranEvent(GR_SCHEDULE, t)); procStatus[CurrentProc] = Busy; +} +#endif // GRAN -#elif defined(PAR) +/* ---------------------------------------------------------------------------- + * Send pending messages (PARALLEL_HASKELL only) + * ------------------------------------------------------------------------- */ + +#if defined(PARALLEL_HASKELL) +static StgTSO * +scheduleSendPendingMessages(void) +{ + StgSparkPool *pool; + rtsSpark spark; + StgTSO *t; + +# if defined(PAR) // global Mem.Mgmt., omit for now if (PendingFetches != END_BF_QUEUE) { processFetches(); } +# endif + + if (RtsFlags.ParFlags.BufferTime) { + // if we use message buffering, we must send away all message + // packets which have become too old... + sendOldBuffers(); + } +} +#endif + +/* ---------------------------------------------------------------------------- + * Activate spark threads (PARALLEL_HASKELL only) + * ------------------------------------------------------------------------- */ + +#if defined(PARALLEL_HASKELL) +static void +scheduleActivateSpark(void) +{ +#if defined(SPARKS) + ASSERT(EMPTY_RUN_QUEUE()); +/* We get here if the run queue is empty and want some work. + We try to turn a spark into a thread, and add it to the run queue, + from where it will be picked up in the next iteration of the scheduler + loop. +*/ - /* ToDo: phps merge with spark activation above */ - /* check whether we have local work and send requests if we have none */ - if (EMPTY_RUN_QUEUE()) { /* no runnable threads */ /* :-[ no local threads => look out for local sparks */ /* the spark pool for the current PE */ - pool = &(MainRegTable.rSparks); // generalise to cap = &MainRegTable + pool = &(cap.r.rSparks); // JB: cap = (old) MainCap if (advisory_thread_count < RtsFlags.ParFlags.maxThreads && pool->hd < pool->tl) { /* @@ -797,29 +1154,73 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, * thread... */ - spark = findSpark(rtsFalse); /* get a spark */ - 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", - tso->id, tso, advisory_thread_count)); - - if (tso==END_TSO_QUEUE) { /* failed to activate spark->back to loop */ - belch("==^^ failed to activate spark"); - 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)", - spark_queue_len(pool))); - goto next_thread; + spark = findSpark(rtsFalse); /* get a spark */ + if (spark != (rtsSpark) NULL) { + tso = createThreadFromSpark(spark); /* turn the spark into a thread */ + IF_PAR_DEBUG(fish, // schedule, + 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 */ + IF_PAR_DEBUG(fish, // schedule, + debugBelch("==^^ failed to create thread from spark @ %lx\n", + spark)); + return rtsFalse; /* failed to generate a thread */ + } /* otherwise fall through & pick-up new tso */ + } else { + IF_PAR_DEBUG(fish, // schedule, + debugBelch("==^^ no local sparks (spark pool contains only NFs: %d)\n", + spark_queue_len(pool))); + return rtsFalse; /* failed to generate a thread */ + } + return rtsTrue; /* success in generating a thread */ + } else { /* no more threads permitted or pool empty */ + return rtsFalse; /* failed to generateThread */ + } +#else + tso = NULL; // avoid compiler warning only + return rtsFalse; /* dummy in non-PAR setup */ +#endif // SPARKS +} +#endif // PARALLEL_HASKELL + +/* ---------------------------------------------------------------------------- + * Get work from a remote node (PARALLEL_HASKELL only) + * ------------------------------------------------------------------------- */ + +#if defined(PARALLEL_HASKELL) +static rtsBool +scheduleGetRemoteWork(rtsBool *receivedFinish) +{ + ASSERT(EMPTY_RUN_QUEUE()); + + if (RtsFlags.ParFlags.BufferTime) { + IF_PAR_DEBUG(verbose, + debugBelch("...send all pending data,")); + { + nat i; + for (i=1; i<=nPEs; i++) + sendImmediately(i); // send all messages away immediately } - } + } +# ifndef SPARKS + //++EDEN++ idle() , i.e. send all buffers, wait for work + // suppress fishing in EDEN... just look for incoming messages + // (blocking receive) + IF_PAR_DEBUG(verbose, + debugBelch("...wait for incoming messages...\n")); + *receivedFinish = processMessages(); // blocking receive... + + // and reenter scheduling loop after having received something + // (return rtsFalse below) + +# else /* activate SPARKS machinery */ +/* We get here, if we have no work, tried to activate a local spark, but still + have no work. We try to get a remote spark, by sending a FISH message. + Thread migration should be added here, and triggered when a sequence of + fishes returns without work. */ + delay = (RtsFlags.ParFlags.fishDelay!=0ll ? RtsFlags.ParFlags.fishDelay : 0ll); - /* If we still have no work we need to send a FISH to get a spark - from another PE - */ - if (EMPTY_RUN_QUEUE()) { /* =8-[ no local sparks => look for work on other PEs */ /* * We really have absolutely no work. Send out a fish @@ -829,75 +1230,128 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, * we're hoping to see. (Of course, we still have to * respond to other types of messages.) */ - TIME now = msTime() /*CURRENT_TIME*/; + rtsTime now = msTime() /*CURRENT_TIME*/; IF_PAR_DEBUG(verbose, - belch("-- now=%ld", 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)", - last_fish_arrived_at+RtsFlags.ParFlags.fishDelay, - last_fish_arrived_at, - RtsFlags.ParFlags.fishDelay, now); - }); - + debugBelch("-- now=%ld\n", now)); + IF_PAR_DEBUG(fish, // verbose, + if (outstandingFishes < RtsFlags.ParFlags.maxFishes && + (last_fish_arrived_at!=0 && + last_fish_arrived_at+delay > now)) { + debugBelch("--$$ <%llu> delaying FISH until %llu (last fish %llu, delay %llu)\n", + now, last_fish_arrived_at+delay, + last_fish_arrived_at, + delay); + }); + if (outstandingFishes < RtsFlags.ParFlags.maxFishes && - (last_fish_arrived_at==0 || - (last_fish_arrived_at+RtsFlags.ParFlags.fishDelay <= now))) { - /* outstandingFishes is set in sendFish, processFish; - avoid flooding system with fishes via delay */ - pe = choosePE(); - sendFish(pe, mytid, NEW_FISH_AGE, NEW_FISH_HISTORY, - NEW_FISH_HUNGER); - - // Global statistics: count no. of fishes - if (RtsFlags.ParFlags.ParStats.Global && - RtsFlags.GcFlags.giveStats > NO_GC_STATS) { - globalParStats.tot_fish_mess++; - } - } - - receivedFinish = processMessages(); - goto next_thread; + advisory_thread_count < RtsFlags.ParFlags.maxThreads) { // send a FISH, but when? + if (last_fish_arrived_at==0 || + (last_fish_arrived_at+delay <= now)) { // send FISH now! + /* outstandingFishes is set in sendFish, processFish; + avoid flooding system with fishes via delay */ + next_fish_to_send_at = 0; + } else { + /* ToDo: this should be done in the main scheduling loop to avoid the + busy wait here; not so bad if fish delay is very small */ + int iq = 0; // DEBUGGING -- HWL + next_fish_to_send_at = last_fish_arrived_at+delay; // remember when to send + /* send a fish when ready, but process messages that arrive in the meantime */ + do { + if (PacketsWaiting()) { + iq++; // DEBUGGING + *receivedFinish = processMessages(); } - } else if (PacketsWaiting()) { /* Look for incoming messages */ - receivedFinish = processMessages(); - } + now = msTime(); + } while (!*receivedFinish || now sent delayed fish (%d processMessages); active/total threads=%d/%d\n",now,iq,run_queue_len(),advisory_thread_count)); - /* Now we are sure that we have some work available */ - ASSERT(run_queue_hd != END_TSO_QUEUE); + } - /* Take a thread from the run queue, if we have work */ - POP_RUN_QUEUE(t); // take_off_run_queue(END_TSO_QUEUE); - IF_DEBUG(sanity,checkTSO(t)); + // JB: IMHO, this should all be hidden inside sendFish(...) + /* pe = choosePE(); + sendFish(pe, thisPE, NEW_FISH_AGE, NEW_FISH_HISTORY, + NEW_FISH_HUNGER); - /* ToDo: write something to the log-file + // Global statistics: count no. of fishes + if (RtsFlags.ParFlags.ParStats.Global && + RtsFlags.GcFlags.giveStats > NO_GC_STATS) { + globalParStats.tot_fish_mess++; + } + */ + + /* delayed fishes must have been sent by now! */ + next_fish_to_send_at = 0; + } + + *receivedFinish = processMessages(); +# endif /* SPARKS */ + + return rtsFalse; + /* NB: this function always returns rtsFalse, meaning the scheduler + loop continues with the next iteration; + rationale: + return code means success in finding work; we enter this function + if there is no local work, thus have to send a fish which takes + time until it arrives with work; in the meantime we should process + messages in the main loop; + */ +} +#endif // PARALLEL_HASKELL + +/* ---------------------------------------------------------------------------- + * PAR/GRAN: Report stats & debugging info(?) + * ------------------------------------------------------------------------- */ + +#if defined(PAR) || defined(GRAN) +static void +scheduleGranParReport(void) +{ + ASSERT(run_queue_hd != END_TSO_QUEUE); + + /* Take a thread from the run queue, if we have work */ + POP_RUN_QUEUE(t); // take_off_run_queue(END_TSO_QUEUE); + + /* If this TSO has got its outport closed in the meantime, + * it mustn't be run. Instead, we have to clean it up as if it was finished. + * It has to be marked as TH_DEAD for this purpose. + * If it is TH_TERM instead, it is supposed to have finished in the normal way. + +JB: TODO: investigate wether state change field could be nuked + entirely and replaced by the normal tso state (whatnext + field). All we want to do is to kill tsos from outside. + */ + + /* ToDo: write something to the log-file if (RTSflags.ParFlags.granSimStats && !sameThread) DumpGranEvent(GR_SCHEDULE, RunnableThreadsHd); CurrentTSO = t; */ /* the spark pool for the current PE */ - pool = &(MainRegTable.rSparks); // generalise to cap = &MainRegTable + pool = &(cap.r.rSparks); // cap = (old) MainCap 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 - if (0 && RtsFlags.ParFlags.ParStats.Full && - t && LastTSO && t->id != LastTSO->id && - LastTSO->why_blocked == NotBlocked && - LastTSO->what_next != ThreadComplete) { - // if previously scheduled TSO not blocked we have to record the context switch - DumpVeryRawGranEvent(TimeOfLastYield, CURRENT_PROC, CURRENT_PROC, - GR_DESCHEDULE, LastTSO, (StgClosure *)NULL, 0, 0); - } + IF_PAR_DEBUG(fish, + debugBelch("--=^ %d threads, %d sparks on [%#x]\n", + run_queue_len(), spark_queue_len(pool), CURRENT_PROC)); if (RtsFlags.ParFlags.ParStats.Full && - (emitSchedule /* forced emit */ || - (t && LastTSO && t->id != LastTSO->id))) { + (t->par.sparkname != (StgInt)0) && // only log spark generated threads + (emitSchedule || // forced emit + (t && LastTSO && t->id != LastTSO->id))) { /* we are running a different TSO, so write a schedule event to log file NB: If we use fair scheduling we also have to write a deschedule @@ -905,130 +1359,24 @@ schedule( StgMainThread *mainThread USED_WHEN_RTS_SUPPORTS_THREADS, previous tso has blocked whenever we switch to another tso, so we don't need it in GUM for now */ + IF_PAR_DEBUG(fish, // schedule, + debugBelch("____ scheduling spark generated thread %d (%lx) (%lx) via a forced emit\n",t->id,t,t->par.sparkname)); + DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC, GR_SCHEDULE, t, (StgClosure *)NULL, 0, 0); emitSchedule = rtsFalse; } - -# endif -#else /* !GRAN && !PAR */ - - // grab a thread from the run queue - ASSERT(run_queue_hd != END_TSO_QUEUE); - POP_RUN_QUEUE(t); - - // Sanity check the thread we're about to run. This can be - // expensive if there is lots of thread switching going on... - IF_DEBUG(sanity,checkTSO(t)); -#endif - -#ifdef THREADED_RTS - { - StgMainThread *m; - for(m = main_threads; m; m = m->link) - { - if(m->tso == t) - break; - } - - if(m) - { - if(m == mainThread) - { - IF_DEBUG(scheduler, - sched_belch("### Running thread %d in bound thread", t->id)); - // yes, the Haskell thread is bound to the current native thread - } - else - { - IF_DEBUG(scheduler, - sched_belch("### thread %d bound to another OS thread", t->id)); - // no, bound to a different Haskell thread: pass to that thread - PUSH_ON_RUN_QUEUE(t); - passCapability(&m->bound_thread_cond); - continue; - } - } - else - { - if(mainThread != NULL) - // The thread we want to run is bound. - { - IF_DEBUG(scheduler, - sched_belch("### this OS thread cannot run thread %d", t->id)); - // no, the current native thread is bound to a different - // Haskell thread, so pass it to any worker thread - PUSH_ON_RUN_QUEUE(t); - passCapabilityToWorker(); - continue; - } - } - } +} #endif - cap->r.rCurrentTSO = t; - - /* context switches are now initiated by the timer signal, unless - * the user specified "context switch as often as possible", with - * +RTS -C0 - */ - if ((RtsFlags.ConcFlags.ctxtSwitchTicks == 0 - && (run_queue_hd != END_TSO_QUEUE - || blocked_queue_hd != END_TSO_QUEUE - || sleeping_queue != END_TSO_QUEUE))) - 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])); - -#ifdef PROFILING - startHeapProfTimer(); -#endif +/* ---------------------------------------------------------------------------- + * After running a thread... + * ASSUMES: sched_mutex + * ------------------------------------------------------------------------- */ - /* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */ - /* Run the current thread - */ - prev_what_next = t->what_next; - 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"); - } - /* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */ - - /* Costs for the scheduler are assigned to CCS_SYSTEM */ -#ifdef PROFILING - stopHeapProfTimer(); - CCCS = CCS_SYSTEM; -#endif - - ACQUIRE_LOCK(&sched_mutex); - -#ifdef RTS_SUPPORTS_THREADS - IF_DEBUG(scheduler,fprintf(stderr,"sched (task %p): ", osThreadId());); -#elif !defined(GRAN) && !defined(PAR) - IF_DEBUG(scheduler,fprintf(stderr,"sched: ");); -#endif - t = cap->r.rCurrentTSO; - +static void +schedulePostRunThread(void) +{ #if defined(PAR) /* HACK 675: if the last thread didn't yield, make sure to print a SCHEDULE event to the log file when StgRunning the next thread, even @@ -1037,218 +1385,52 @@ run_thread: TimeOfLastYield = CURRENT_TIME; #endif - switch (ret) { + /* some statistics gathering in the parallel case */ + +#if defined(GRAN) || defined(PAR) || defined(EDEN) + switch (ret) { case HeapOverflow: -#if defined(GRAN) +# if defined(GRAN) IF_DEBUG(gran, DumpGranEvent(GR_DESCHEDULE, t)); globalGranStats.tot_heapover++; -#elif defined(PAR) +# elif defined(PAR) globalParStats.tot_heapover++; -#endif - - // did the task ask for a large block? - if (cap->r.rHpAlloc > BLOCK_SIZE_W) { - // 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; - - IF_DEBUG(scheduler,belch("--<< thread %ld (%s) stopped: requesting a large block (size %d)", - 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. - if (alloc_blocks + blocks < alloc_blocks_lim) { - - alloc_blocks += blocks; - bd = allocGroup( blocks ); - - // link the new group into the list - bd->link = cap->r.rCurrentNursery; - bd->u.back = cap->r.rCurrentNursery->u.back; - if (cap->r.rCurrentNursery->u.back != NULL) { - cap->r.rCurrentNursery->u.back->link = bd; - } else { - ASSERT(g0s0->blocks == cap->r.rCurrentNursery && - g0s0->blocks == cap->r.rNursery); - cap->r.rNursery = g0s0->blocks = bd; - } - cap->r.rCurrentNursery->u.back = bd; - - // initialise it as a nursery block. We initialise the - // step, gen_no, and flags field of *every* sub-block in - // this large block, because this is easier than making - // sure that we always find the block head of a large - // block whenever we call Bdescr() (eg. evacuate() and - // isAlive() in the GC would both have to do this, at - // least). - { - bdescr *x; - for (x = bd; x < bd + blocks; x++) { - x->step = g0s0; - x->gen_no = 0; - x->flags = 0; - } - } - - // don't forget to update the block count in g0s0. - g0s0->n_blocks += blocks; - // This assert can be a killer if the app is doing lots - // of large block allocations. - ASSERT(countBlocks(g0s0->blocks) == g0s0->n_blocks); - - // now update the nursery to point to the new block - cap->r.rCurrentNursery = bd; - - // we might be unlucky and have another thread get on the - // run queue before us and steal the large block, but in that - // case the thread will just end up requesting another large - // block. - PUSH_ON_RUN_QUEUE(t); - break; - } - } - - /* make all the running tasks block on a condition variable, - * 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])); - threadPaused(t); -#if defined(GRAN) - ASSERT(!is_on_queue(t,CurrentProc)); -#elif defined(PAR) - /* Currently we emit a DESCHEDULE event before GC in GUM. - ToDo: either add separate event to distinguish SYSTEM time from rest - or just nuke this DESCHEDULE (and the following SCHEDULE) */ - if (0 && RtsFlags.ParFlags.ParStats.Full) { - DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC, - GR_DESCHEDULE, t, (StgClosure *)NULL, 0, 0); - emitSchedule = rtsTrue; - } -#endif - - ready_to_gc = rtsTrue; - context_switch = 1; /* stop other threads ASAP */ - PUSH_ON_RUN_QUEUE(t); - /* actual GC is done at the end of the while loop */ +# endif break; - - case StackOverflow: -#if defined(GRAN) + + case StackOverflow: +# if defined(GRAN) IF_DEBUG(gran, DumpGranEvent(GR_DESCHEDULE, t)); globalGranStats.tot_stackover++; -#elif defined(PAR) +# elif defined(PAR) // IF_DEBUG(par, // DumpGranEvent(GR_DESCHEDULE, t); globalParStats.tot_stackover++; -#endif - IF_DEBUG(scheduler,belch("--<< thread %ld (%s) stopped, StackOverflow", - 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); - - /* This TSO has moved, so update any pointers to it from the - * main thread stack. It better not be on any other queues... - * (it shouldn't be). - */ - for (m = main_threads; m != NULL; m = m->link) { - if (m->tso == t) { - m->tso = new_t; - } - } - threadPaused(new_t); - PUSH_ON_RUN_QUEUE(new_t); - } +# endif break; case ThreadYielding: -#if defined(GRAN) +# if defined(GRAN) IF_DEBUG(gran, DumpGranEvent(GR_DESCHEDULE, t)); globalGranStats.tot_yields++; -#elif defined(PAR) +# elif defined(PAR) // IF_DEBUG(par, // DumpGranEvent(GR_DESCHEDULE, t); globalParStats.tot_yields++; -#endif - /* put the thread back on the run queue. Then, if we're ready to - * GC, check whether this is the last task to stop. If so, wake - * up the GC thread. getThread will block during a GC until the - * GC is finished. - */ - 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]); - } else { - belch("--<< thread %ld (%s) stopped, yielding", - t->id, whatNext_strs[t->what_next]); - } - ); - - IF_DEBUG(sanity, - //belch("&& Doing sanity check on yielding TSO %ld.", t->id); - checkTSO(t)); - ASSERT(t->link == END_TSO_QUEUE); - - // Shortcut if we're just switching evaluators: don't bother - // doing stack squeezing (which can be expensive), just run the - // thread. - if (t->what_next != prev_what_next) { - goto run_thread; - } - - threadPaused(t); - -#if defined(GRAN) - ASSERT(!is_on_queue(t,CurrentProc)); - - IF_DEBUG(sanity, - //belch("&& Doing sanity check on all ThreadQueues (and their TSOs)."); - checkThreadQsSanity(rtsTrue)); -#endif - -#if defined(PAR) - if (RtsFlags.ParFlags.doFairScheduling) { - /* this does round-robin scheduling; good for concurrency */ - APPEND_TO_RUN_QUEUE(t); - } else { - /* this does unfair scheduling; good for parallelism */ - PUSH_ON_RUN_QUEUE(t); - } -#else - // this does round-robin scheduling; good for concurrency - APPEND_TO_RUN_QUEUE(t); -#endif - -#if defined(GRAN) - /* add a ContinueThread event to actually process the thread */ - new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc], - ContinueThread, - t, (StgClosure*)NULL, (rtsSpark*)NULL); - IF_GRAN_DEBUG(bq, - belch("GRAN: eventq and runnableq after adding yielded thread to queue again:"); - G_EVENTQ(0); - G_CURR_THREADQ(0)); -#endif /* GRAN */ - break; +# endif + break; case ThreadBlocked: -#if defined(GRAN) +# if defined(GRAN) IF_DEBUG(scheduler, - belch("--<< thread %ld (%p; %s) stopped, blocking on node %p [PE %d] with BQ: ", - 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)); + debugBelch("--<< thread %ld (%p; %s) stopped, blocking on node %p [PE %d] with BQ: ", + 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); + debugBelch("\n")); // ??? needed; should emit block before IF_DEBUG(gran, @@ -1264,76 +1446,424 @@ run_thread: procStatus[CurrentProc]==Fetching)) procStatus[CurrentProc] = Idle; */ -#elif defined(PAR) - IF_DEBUG(scheduler, - belch("--<< thread %ld (%p; %s) stopped, blocking on node %p with BQ: ", - t->id, t, whatNext_strs[t->what_next], t->block_info.closure)); - IF_PAR_DEBUG(bq, +# elif defined(PAR) +//++PAR++ blockThread() writes the event (change?) +# endif + break; - if (t->block_info.closure!=(StgClosure*)NULL) - print_bq(t->block_info.closure)); + case ThreadFinished: + break; - /* Send a fetch (if BlockedOnGA) and dump event to log file */ - blockThread(t); + default: + barf("parGlobalStats: unknown return code"); + break; + } +#endif +} - /* whatever we schedule next, we must log that schedule */ - emitSchedule = rtsTrue; +/* ----------------------------------------------------------------------------- + * Handle a thread that returned to the scheduler with ThreadHeepOverflow + * ASSUMES: sched_mutex + * -------------------------------------------------------------------------- */ + +static rtsBool +scheduleHandleHeapOverflow( Capability *cap, StgTSO *t ) +{ + // did the task ask for a large block? + if (cap->r.rHpAlloc > BLOCK_SIZE) { + // if so, get one and push it on the front of the nursery. + bdescr *bd; + lnat blocks; + + blocks = (lnat)BLOCK_ROUND_UP(cap->r.rHpAlloc) / BLOCK_SIZE; + + IF_DEBUG(scheduler, + debugBelch("--<< thread %ld (%s) stopped: requesting a large block (size %ld)\n", + (long)t->id, whatNext_strs[t->what_next], blocks)); + + // don't do this if the nursery is (nearly) full, we'll GC first. + if (cap->r.rCurrentNursery->link != NULL || + cap->r.rNursery->n_blocks == 1) { // paranoia to prevent infinite loop + // if the nursery has only one block. + + bd = allocGroup( blocks ); + cap->r.rNursery->n_blocks += blocks; + + // link the new group into the list + bd->link = cap->r.rCurrentNursery; + bd->u.back = cap->r.rCurrentNursery->u.back; + if (cap->r.rCurrentNursery->u.back != NULL) { + cap->r.rCurrentNursery->u.back->link = bd; + } else { +#if !defined(SMP) + ASSERT(g0s0->blocks == cap->r.rCurrentNursery && + g0s0 == cap->r.rNursery); +#endif + cap->r.rNursery->blocks = bd; + } + cap->r.rCurrentNursery->u.back = bd; + + // initialise it as a nursery block. We initialise the + // step, gen_no, and flags field of *every* sub-block in + // this large block, because this is easier than making + // sure that we always find the block head of a large + // block whenever we call Bdescr() (eg. evacuate() and + // isAlive() in the GC would both have to do this, at + // least). + { + bdescr *x; + for (x = bd; x < bd + blocks; x++) { + x->step = cap->r.rNursery; + x->gen_no = 0; + x->flags = 0; + } + } + + // This assert can be a killer if the app is doing lots + // of large block allocations. + IF_DEBUG(sanity, checkNurserySanity(cap->r.rNursery)); + + // now update the nursery to point to the new block + cap->r.rCurrentNursery = bd; + + // we might be unlucky and have another thread get on the + // run queue before us and steal the large block, but in that + // case the thread will just end up requesting another large + // block. + PUSH_ON_RUN_QUEUE(t); + return rtsFalse; /* not actually GC'ing */ + } + } + + IF_DEBUG(scheduler, + debugBelch("--<< thread %ld (%s) stopped: HeapOverflow\n", + (long)t->id, whatNext_strs[t->what_next])); +#if defined(GRAN) + ASSERT(!is_on_queue(t,CurrentProc)); +#elif defined(PARALLEL_HASKELL) + /* Currently we emit a DESCHEDULE event before GC in GUM. + ToDo: either add separate event to distinguish SYSTEM time from rest + or just nuke this DESCHEDULE (and the following SCHEDULE) */ + if (0 && RtsFlags.ParFlags.ParStats.Full) { + DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC, + GR_DESCHEDULE, t, (StgClosure *)NULL, 0, 0); + emitSchedule = rtsTrue; + } +#endif + + PUSH_ON_RUN_QUEUE(t); + return rtsTrue; + /* actual GC is done at the end of the while loop in schedule() */ +} + +/* ----------------------------------------------------------------------------- + * Handle a thread that returned to the scheduler with ThreadStackOverflow + * ASSUMES: sched_mutex + * -------------------------------------------------------------------------- */ +static void +scheduleHandleStackOverflow( StgTSO *t) +{ + 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. + */ + { + /* enlarge the stack */ + StgTSO *new_t = threadStackOverflow(t); + + /* This TSO has moved, so update any pointers to it from the + * main thread stack. It better not be on any other queues... + * (it shouldn't be). + */ + if (t->main != NULL) { + t->main->tso = new_t; + } + PUSH_ON_RUN_QUEUE(new_t); + } +} + +/* ----------------------------------------------------------------------------- + * Handle a thread that returned to the scheduler with ThreadYielding + * ASSUMES: sched_mutex + * -------------------------------------------------------------------------- */ + +static rtsBool +scheduleHandleYield( StgTSO *t, nat prev_what_next ) +{ + // Reset the context switch flag. We don't do this just before + // running the thread, because that would mean we would lose ticks + // 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; + + /* put the thread back on the run queue. Then, if we're ready to + * GC, check whether this is the last task to stop. If so, wake + * up the GC thread. getThread will block during a GC until the + * GC is finished. + */ + IF_DEBUG(scheduler, + if (t->what_next != prev_what_next) { + debugBelch("--<< thread %ld (%s) stopped to switch evaluators\n", + (long)t->id, whatNext_strs[t->what_next]); + } else { + debugBelch("--<< thread %ld (%s) stopped, yielding\n", + (long)t->id, whatNext_strs[t->what_next]); + } + ); + + IF_DEBUG(sanity, + //debugBelch("&& Doing sanity check on yielding TSO %ld.", t->id); + checkTSO(t)); + ASSERT(t->link == END_TSO_QUEUE); + + // Shortcut if we're just switching evaluators: don't bother + // doing stack squeezing (which can be expensive), just run the + // thread. + if (t->what_next != prev_what_next) { + return rtsTrue; + } + +#if defined(GRAN) + ASSERT(!is_on_queue(t,CurrentProc)); + + IF_DEBUG(sanity, + //debugBelch("&& Doing sanity check on all ThreadQueues (and their TSOs)."); + checkThreadQsSanity(rtsTrue)); + +#endif + + addToRunQueue(t); + +#if defined(GRAN) + /* add a ContinueThread event to actually process the thread */ + new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc], + ContinueThread, + t, (StgClosure*)NULL, (rtsSpark*)NULL); + IF_GRAN_DEBUG(bq, + debugBelch("GRAN: eventq and runnableq after adding yielded thread to queue again:\n"); + G_EVENTQ(0); + G_CURR_THREADQ(0)); +#endif + return rtsFalse; +} + +/* ----------------------------------------------------------------------------- + * Handle a thread that returned to the scheduler with ThreadBlocked + * ASSUMES: sched_mutex + * -------------------------------------------------------------------------- */ + +static void +scheduleHandleThreadBlocked( StgTSO *t +#if !defined(GRAN) && !defined(DEBUG) + STG_UNUSED +#endif + ) +{ +#if defined(GRAN) + IF_DEBUG(scheduler, + 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)); + + // ??? needed; should emit block before + IF_DEBUG(gran, + DumpGranEvent(GR_DESCHEDULE, t)); + prune_eventq(t, (StgClosure *)NULL); // prune ContinueThreads for t + /* + ngoq Dogh! + ASSERT(procStatus[CurrentProc]==Busy || + ((procStatus[CurrentProc]==Fetching) && + (t->block_info.closure!=(StgClosure*)NULL))); + if (run_queue_hds[CurrentProc] == END_TSO_QUEUE && + !(!RtsFlags.GranFlags.DoAsyncFetch && + procStatus[CurrentProc]==Fetching)) + procStatus[CurrentProc] = Idle; + */ +#elif defined(PAR) + IF_DEBUG(scheduler, + debugBelch("--<< thread %ld (%p; %s) stopped, blocking on node %p with BQ: \n", + t->id, t, whatNext_strs[t->what_next], t->block_info.closure)); + IF_PAR_DEBUG(bq, + + if (t->block_info.closure!=(StgClosure*)NULL) + print_bq(t->block_info.closure)); + + /* Send a fetch (if BlockedOnGA) and dump event to log file */ + blockThread(t); + + /* whatever we schedule next, we must log that schedule */ + emitSchedule = rtsTrue; + #else /* !GRAN */ - /* don't need to do anything. Either the thread is blocked on - * I/O, in which case we'll have called addToBlockedQueue - * previously, or it's blocked on an MVar or Blackhole, in which - * case it'll be on the relevant queue already. - */ - IF_DEBUG(scheduler, - fprintf(stderr, "--<< thread %d (%s) stopped: ", - t->id, whatNext_strs[t->what_next]); - printThreadBlockage(t); - fprintf(stderr, "\n")); - fflush(stderr); - - /* Only for dumping event to log file - ToDo: do I need this in GranSim, too? - blockThread(t); - */ + + // We don't need to do anything. The thread is blocked, and it + // has tidied up its stack and placed itself on whatever queue + // it needs to be on. + +#if !defined(SMP) + ASSERT(t->why_blocked != NotBlocked); + // This might not be true under SMP: we don't have + // exclusive access to this TSO, so someone might have + // woken it up by now. This actually happens: try + // conc023 +RTS -N2. #endif - threadPaused(t); - break; - case ThreadFinished: - /* Need to check whether this was a main thread, and if so, signal - * the task that started it with the return value. If we have no - * more main threads, we probably need to stop all the tasks until - * we get a new one. - */ - /* 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", - t->id, whatNext_strs[t->what_next])); + IF_DEBUG(scheduler, + debugBelch("--<< thread %d (%s) stopped: ", + t->id, whatNext_strs[t->what_next]); + printThreadBlockage(t); + debugBelch("\n")); + + /* Only for dumping event to log file + ToDo: do I need this in GranSim, too? + blockThread(t); + */ +#endif +} + +/* ----------------------------------------------------------------------------- + * Handle a thread that returned to the scheduler with ThreadFinished + * ASSUMES: sched_mutex + * -------------------------------------------------------------------------- */ + +static rtsBool +scheduleHandleThreadFinished( StgMainThread *mainThread + USED_WHEN_RTS_SUPPORTS_THREADS, + Capability *cap, + StgTSO *t ) +{ + /* Need to check whether this was a main thread, and if so, + * return with the return value. + * + * We also end up here if the thread kills itself with an + * uncaught exception, see Exception.cmm. + */ + IF_DEBUG(scheduler,debugBelch("--++ thread %d (%s) finished\n", + t->id, whatNext_strs[t->what_next])); + #if defined(GRAN) endThread(t, CurrentProc); // clean-up the thread -#elif defined(PAR) +#elif defined(PARALLEL_HASKELL) /* For now all are advisory -- HWL */ //if(t->priority==AdvisoryPriority) ?? - advisory_thread_count--; + advisory_thread_count--; // JB: Caution with this counter, buggy! -# ifdef DIST +# if defined(DIST) if(t->dist.priority==RevalPriority) FinishReval(t); # endif - + +# if defined(EDENOLD) + // the thread could still have an outport... (BUG) + if (t->eden.outport != -1) { + // delete the outport for the tso which has finished... + IF_PAR_DEBUG(eden_ports, + debugBelch("WARNING: Scheduler removes outport %d for TSO %d.\n", + t->eden.outport, t->id)); + deleteOPT(t); + } + // thread still in the process (HEAVY BUG! since outport has just been closed...) + if (t->eden.epid != -1) { + IF_PAR_DEBUG(eden_ports, + debugBelch("WARNING: Scheduler removes TSO %d from process %d .\n", + t->id, t->eden.epid)); + removeTSOfromProcess(t); + } +# endif + +# if defined(PAR) if (RtsFlags.ParFlags.ParStats.Full && !RtsFlags.ParFlags.ParStats.Suppressed) DumpEndEvent(CURRENT_PROC, t, rtsFalse /* not mandatory */); + + // t->par only contains statistics: left out for now... + IF_PAR_DEBUG(fish, + debugBelch("**** end thread: ended sparked thread %d (%lx); sparkname: %lx\n", + t->id,t,t->par.sparkname)); +# endif +#endif // PARALLEL_HASKELL + + // + // 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 - break; - - default: - barf("schedule: invalid thread return code %d", (int)ret); - } + ) + { + // 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 (interrupted) { + mainThread->stat = Interrupted; + } else { + mainThread->stat = Killed; + } + } +#ifdef DEBUG + removeThreadLabel((StgWord)mainThread->tso->id); +#endif + if (mainThread->prev == NULL) { + ASSERT(mainThread == main_threads); + main_threads = mainThread->link; + } else { + mainThread->prev->link = mainThread->link; + } + if (mainThread->link != NULL) { + mainThread->link->prev = mainThread->prev; + } + releaseCapability(cap); + return rtsTrue; // tells schedule() to return + } -#ifdef PROFILING +#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 + return rtsFalse; +} + +/* ----------------------------------------------------------------------------- + * Perform a heap census, if PROFILING + * -------------------------------------------------------------------------- */ + +static rtsBool +scheduleDoHeapProfile( rtsBool ready_to_gc STG_UNUSED ) +{ +#if defined(PROFILING) // When we have +RTS -i0 and we're heap profiling, do a census at // every GC. This lets us get repeatable runs for debugging. if (performHeapProfile || @@ -1342,48 +1872,111 @@ run_thread: GarbageCollect(GetRoots, rtsTrue); heapCensus(); performHeapProfile = rtsFalse; - ready_to_gc = rtsFalse; // we already GC'd + return rtsTrue; // true <=> we already GC'd } #endif + return rtsFalse; +} - 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 - * broadcast on gc_pending_cond afterward. - */ -#if defined(RTS_SUPPORTS_THREADS) - IF_DEBUG(scheduler,sched_belch("doing GC")); +/* ----------------------------------------------------------------------------- + * Perform a garbage collection if necessary + * ASSUMES: sched_mutex + * -------------------------------------------------------------------------- */ + +static void +scheduleDoGC( rtsBool force_major ) +{ + StgTSO *t; +#ifdef SMP + Capability *cap; + static rtsBool waiting_for_gc; + int n_capabilities = RtsFlags.ParFlags.nNodes - 1; + // subtract one because we're already holding one. + Capability *caps[n_capabilities]; #endif - GarbageCollect(GetRoots,rtsFalse); - ready_to_gc = rtsFalse; -#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"); - G_EVENTQ(0); - G_CURR_THREADQ(0)); -#endif /* GRAN */ + +#ifdef SMP + // In order to GC, there must be no threads running Haskell code. + // Therefore, the GC thread needs to hold *all* the capabilities, + // and release them after the GC has completed. + // + // This seems to be the simplest way: previous attempts involved + // making all the threads with capabilities give up their + // capabilities and sleep except for the *last* one, which + // actually did the GC. But it's quite hard to arrange for all + // the other tasks to sleep and stay asleep. + // + + // Someone else is already trying to GC + if (waiting_for_gc) return; + waiting_for_gc = rtsTrue; + + while (n_capabilities > 0) { + IF_DEBUG(scheduler, sched_belch("ready_to_gc, grabbing all the capabilies (%d left)", n_capabilities)); + waitForReturnCapability(&sched_mutex, &cap); + n_capabilities--; + caps[n_capabilities] = cap; } -#if defined(GRAN) - next_thread: - IF_GRAN_DEBUG(unused, - print_eventq(EventHd)); + waiting_for_gc = rtsFalse; +#endif - event = get_next_event(); -#elif defined(PAR) - next_thread: - /* ToDo: wait for next message to arrive rather than busy wait */ -#endif /* GRAN */ + /* 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 (!stmValidateNestOfTransactions (t -> trec)) { + IF_DEBUG(stm, sched_belch("trec %p found wasting its time", t)); + + // strip the stack back to the ATOMICALLY_FRAME, aborting + // the (nested) transaction, and saving the stack of any + // partially-evaluated thunks on the heap. + raiseAsync_(t, NULL, rtsTrue); + +#ifdef REG_R1 + ASSERT(get_itbl((StgClosure *)t->sp)->type == ATOMICALLY_FRAME); +#endif + } + } + } + + // so this happens periodically: + scheduleCheckBlackHoles(); + + IF_DEBUG(scheduler, printAllThreads()); - } /* end of while(1) */ + /* everybody back, start the GC. + * Could do it in this thread, or signal a condition var + * to do it in another thread. Either way, we need to + * broadcast on gc_pending_cond afterward. + */ +#if defined(RTS_SUPPORTS_THREADS) + IF_DEBUG(scheduler,sched_belch("doing GC")); +#endif + GarbageCollect(GetRoots, force_major); + +#if defined(SMP) + { + // release our stash of capabilities. + nat i; + for (i = 0; i < RtsFlags.ParFlags.nNodes-1; i++) { + releaseCapability(caps[i]); + } + } +#endif - IF_PAR_DEBUG(verbose, - belch("== Leaving schedule() after having received Finish")); +#if defined(GRAN) + /* add a ContinueThread event to continue execution of current thread */ + new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc], + ContinueThread, + t, (StgClosure*)NULL, (rtsSpark*)NULL); + IF_GRAN_DEBUG(bq, + debugBelch("GRAN: eventq and runnableq after Garbage collection:\n\n"); + G_EVENTQ(0); + G_CURR_THREADQ(0)); +#endif /* GRAN */ } /* --------------------------------------------------------------------------- @@ -1394,7 +1987,7 @@ run_thread: StgBool rtsSupportsBoundThreads(void) { -#ifdef THREADED_RTS +#if defined(RTS_SUPPORTS_THREADS) return rtsTrue; #else return rtsFalse; @@ -1408,13 +2001,8 @@ rtsSupportsBoundThreads(void) 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; - } +#if defined(RTS_SUPPORTS_THREADS) + return (tso->main != NULL); #endif return rtsFalse; } @@ -1423,13 +2011,22 @@ isThreadBound(StgTSO* tso USED_IN_THREADED_RTS) * Singleton fork(). Do not copy any running threads. * ------------------------------------------------------------------------- */ +#ifndef mingw32_HOST_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; @@ -1463,18 +2060,12 @@ 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 - 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); @@ -1483,10 +2074,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 } /* --------------------------------------------------------------------------- @@ -1504,12 +2095,23 @@ deleteAllThreads ( void ) StgTSO* t, *next; IF_DEBUG(scheduler,sched_belch("deleting all threads")); for (t = all_threads; t != END_TSO_QUEUE; t = next) { - next = t->global_link; - deleteThread(t); + if (t->what_next == ThreadRelocated) { + next = t->link; + } else { + 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(blackhole_queue == END_TSO_QUEUE); + ASSERT(sleeping_queue == END_TSO_QUEUE); } /* startThread and insertThread are now in GranSim.c -- HWL */ @@ -1531,12 +2133,7 @@ deleteAllThreads ( void ) * ------------------------------------------------------------------------- */ StgInt -suspendThread( StgRegTable *reg, - rtsBool concCall -#if !defined(DEBUG) - STG_UNUSED -#endif - ) +suspendThread( StgRegTable *reg ) { nat tok; Capability *cap; @@ -1550,7 +2147,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; @@ -1559,22 +2156,18 @@ 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; /* Hand back capability */ + cap->r.rInHaskell = rtsFalse; releaseCapability(cap); #if defined(RTS_SUPPORTS_THREADS) @@ -1584,8 +2177,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; @@ -1593,8 +2184,7 @@ suspendThread( StgRegTable *reg, } StgRegTable * -resumeThread( StgInt tok, - rtsBool concCall STG_UNUSED ) +resumeThread( StgInt tok ) { StgTSO *tso, **prev; Capability *cap; @@ -1625,29 +2215,21 @@ 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; + cap->r.rInHaskell = rtsTrue; RELEASE_LOCK(&sched_mutex); errno = saved_errno; return &cap->r; } - -/* --------------------------------------------------------------------------- - * Static functions - * ------------------------------------------------------------------------ */ -static void unblockThread(StgTSO *tso); - /* --------------------------------------------------------------------------- * Comparing Thread ids. * @@ -1720,11 +2302,11 @@ createThread(nat size) nat stack_size; /* First check whether we should create a thread at all */ -#if defined(PAR) +#if defined(PARALLEL_HASKELL) /* 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; } @@ -1760,12 +2342,15 @@ createThread(nat size) 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) - TSO_STRUCT_SIZEW; tso->sp = (P_)&(tso->stack) + stack_size; + tso->trec = NO_TREC; + #ifdef PROFILING tso->prof.CCCS = CCS_MAIN; #endif @@ -1773,9 +2358,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 @@ -1787,7 +2373,7 @@ createThread(nat size) #if defined(GRAN) if (RtsFlags.GranFlags.GranSimStats.Full) DumpGranEvent(GR_START,tso); -#elif defined(PAR) +#elif defined(PARALLEL_HASKELL) if (RtsFlags.ParFlags.ParStats.Full) DumpGranEvent(GR_STARTQ,tso); /* HACk to avoid SCHEDULE @@ -1827,7 +2413,7 @@ createThread(nat size) tso->gran.clock = 0; IF_DEBUG(gran,printTSO(tso)); -#elif defined(PAR) +#elif defined(PARALLEL_HASKELL) # if defined(DEBUG) tso->par.magic = TSO_MAGIC; // debugging only # endif @@ -1851,26 +2437,26 @@ createThread(nat size) globalGranStats.threads_created_on_PE[CurrentProc]++; globalGranStats.tot_sq_len += spark_queue_len(CurrentProc); globalGranStats.tot_sq_probes++; -#elif defined(PAR) +#elif defined(PARALLEL_HASKELL) // collect parallel global statistics (currently done together with GC stats) if (RtsFlags.ParFlags.ParStats.Global && RtsFlags.GcFlags.giveStats > NO_GC_STATS) { - //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)); +#elif defined(PARALLEL_HASKELL) + IF_PAR_DEBUG(verbose, + sched_belch("==__ schedule: Created TSO %d (%p); %d threads active", + (long)tso->id, tso, advisory_thread_count)); #else IF_DEBUG(scheduler,sched_belch("created thread %ld, stack size = %lx words", - tso->id, tso->stack_size)); + (long)tso->id, (long)tso->stack_size)); #endif return tso; } @@ -1880,9 +2466,10 @@ createThread(nat size) all parallel thread creation calls should fall through the following routine. */ StgTSO * -createSparkThread(rtsSpark spark) +createThreadFromSpark(rtsSpark spark) { StgTSO *tso; ASSERT(spark != (rtsSpark)NULL); +// JB: TAKE CARE OF THIS COUNTER! BUGGY if (advisory_thread_count >= RtsFlags.ParFlags.maxThreads) { threadsIgnored++; barf("{createSparkThread}Daq ghuH: refusing to create another thread; no more than %d threads allowed (currently %d)", @@ -1898,8 +2485,8 @@ createSparkThread(rtsSpark spark) tso->priority = AdvisoryPriority; #endif pushClosure(tso,spark); - PUSH_ON_RUN_QUEUE(tso); - advisory_thread_count++; + addToRunQueue(tso); + advisory_thread_count++; // JB: TAKE CARE OF THIS COUNTER! BUGGY } return tso; } @@ -1909,7 +2496,7 @@ createSparkThread(rtsSpark spark) Turn a spark into a thread. ToDo: fix for SMP (needs to acquire SCHED_MUTEX!) */ -#if defined(PAR) +#if 0 StgTSO * activateSpark (rtsSpark spark) { @@ -1918,9 +2505,9 @@ activateSpark (rtsSpark spark) tso = createSparkThread(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", - (StgClosure *)spark, info_type((StgClosure *)spark))); + IF_PAR_DEBUG(verbose, + 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 // tso->gran.exported = spark->exported; @@ -1931,11 +2518,6 @@ activateSpark (rtsSpark spark) } #endif -static SchedulerStatus waitThread_(/*out*/StgMainThread* m, - Capability *initialCapability - ); - - /* --------------------------------------------------------------------------- * scheduleThread() * @@ -1946,14 +2528,13 @@ static SchedulerStatus waitThread_(/*out*/StgMainThread* m, * on this thread's stack before the scheduler is invoked. * ------------------------------------------------------------------------ */ -static void scheduleThread_ (StgTSO* tso); - -void +static 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 @@ -1965,9 +2546,11 @@ scheduleThread(StgTSO* tso) } #if defined(RTS_SUPPORTS_THREADS) -static Condition *bound_cond_cache = NULL; +static Condition bound_cond_cache; +static int bound_cond_cache_full = 0; #endif + SchedulerStatus scheduleWaitThread(StgTSO* tso, /*[out]*/HaskellObj* ret, Capability *initialCapability) @@ -1977,15 +2560,23 @@ scheduleWaitThread(StgTSO* tso, /*[out]*/HaskellObj* ret, 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; + #if defined(RTS_SUPPORTS_THREADS) // Allocating a new condition for each thread is expensive, so we // cache one. This is a pretty feeble hack, but it helps speed up // consecutive call-ins quite a bit. - if (bound_cond_cache != NULL) { - m->bound_thread_cond = *bound_cond_cache; - bound_cond_cache = NULL; + if (bound_cond_cache_full) { + m->bound_thread_cond = bound_cond_cache; + bound_cond_cache_full = 0; } else { initCondition(&m->bound_thread_cond); } @@ -2002,11 +2593,8 @@ scheduleWaitThread(StgTSO* tso, /*[out]*/HaskellObj* ret, */ IF_DEBUG(scheduler, sched_belch("waiting for thread (%d)", tso->id)); - m->link = main_threads; - main_threads = m; - - 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. @@ -2034,6 +2622,7 @@ initScheduler(void) blocked_queue_hds[i] = END_TSO_QUEUE; blocked_queue_tls[i] = END_TSO_QUEUE; ccalling_threadss[i] = END_TSO_QUEUE; + blackhole_queue[i] = END_TSO_QUEUE; sleeping_queue = END_TSO_QUEUE; } #else @@ -2041,6 +2630,7 @@ initScheduler(void) run_queue_tl = END_TSO_QUEUE; blocked_queue_hd = END_TSO_QUEUE; blocked_queue_tl = END_TSO_QUEUE; + blackhole_queue = END_TSO_QUEUE; sleeping_queue = END_TSO_QUEUE; #endif @@ -2062,9 +2652,7 @@ initScheduler(void) initMutex(&term_mutex); #endif -#if defined(RTS_SUPPORTS_THREADS) ACQUIRE_LOCK(&sched_mutex); -#endif /* A capability holds the state a native thread needs in * order to execute STG code. At least one capability is @@ -2073,25 +2661,31 @@ initScheduler(void) initCapabilities(); #if defined(RTS_SUPPORTS_THREADS) - /* start our haskell execution tasks */ - startTaskManager(0,taskStart); + initTaskManager(); +#endif + +#if defined(SMP) + /* eagerly start some extra workers */ + startingWorkerThread = RtsFlags.ParFlags.nNodes; + startTasks(RtsFlags.ParFlags.nNodes, taskStart); #endif -#if /* defined(SMP) ||*/ defined(PAR) +#if /* defined(SMP) ||*/ defined(PARALLEL_HASKELL) initSparkPools(); #endif RELEASE_LOCK(&sched_mutex); - } void exitScheduler( void ) { + interrupted = rtsTrue; + shutting_down_scheduler = rtsTrue; #if defined(RTS_SUPPORTS_THREADS) - stopTaskManager(); + if (threadIsTask(osThreadId())) { taskStop(); } + stopTaskManager(); #endif - shutting_down_scheduler = rtsTrue; } /* ---------------------------------------------------------------------------- @@ -2103,8 +2697,7 @@ exitScheduler( void ) ToDo: no support for two-space collection at the moment??? ------------------------------------------------------------------------- */ -static -SchedulerStatus +static SchedulerStatus waitThread_(StgMainThread* m, Capability *initialCapability) { SchedulerStatus stat; @@ -2127,8 +2720,9 @@ waitThread_(StgMainThread* m, Capability *initialCapability) #if defined(RTS_SUPPORTS_THREADS) // Free the condition variable, returning it to the cache if possible. - if (bound_cond_cache == NULL) { - *bound_cond_cache = m->bound_thread_cond; + if (!bound_cond_cache_full) { + bound_cond_cache = m->bound_thread_cond; + bound_cond_cache_full = 1; } else { closeCondition(&m->bound_thread_cond); } @@ -2198,11 +2792,15 @@ GetRoots( evac_fn evac ) } #endif + if (blackhole_queue != END_TSO_QUEUE) { + evac((StgClosure **)&blackhole_queue); + } + if (suspended_ccalling_threads != END_TSO_QUEUE) { evac((StgClosure **)&suspended_ccalling_threads); } -#if defined(PAR) || defined(GRAN) +#if defined(PARALLEL_HASKELL) || defined(GRAN) markSparkQueue(evac); #endif @@ -2210,25 +2808,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; - } - } - } } /* ----------------------------------------------------------------------------- @@ -2291,7 +2870,8 @@ performGCWithRoots(void (*get_roots)(evac_fn)) static StgTSO * threadStackOverflow(StgTSO *tso) { - nat new_stack_size, new_tso_size, stack_words; + nat new_stack_size, stack_words; + lnat new_tso_size; StgPtr new_sp; StgTSO *dest; @@ -2299,8 +2879,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))); @@ -2315,12 +2895,12 @@ threadStackOverflow(StgTSO *tso) * Finally round up so the TSO ends up as a whole number of blocks. */ new_stack_size = stg_min(tso->stack_size * 2, tso->max_stack_size); - new_tso_size = (nat)BLOCK_ROUND_UP(new_stack_size * sizeof(W_) + + new_tso_size = (lnat)BLOCK_ROUND_UP(new_stack_size * sizeof(W_) + TSO_STRUCT_SIZE)/sizeof(W_); 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); @@ -2346,10 +2926,9 @@ threadStackOverflow(StgTSO *tso) tso->link = dest; tso->sp = (P_)&(tso->stack[tso->stack_size]); tso->why_blocked = NotBlocked; - 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, @@ -2372,7 +2951,7 @@ STATIC_INLINE void unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node ) { } -#elif defined(PAR) +#elif defined(PARALLEL_HASKELL) STATIC_INLINE void unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node ) { @@ -2406,7 +2985,7 @@ unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node ) #endif #if defined(GRAN) -static StgBlockingQueueElement * +StgBlockingQueueElement * unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node) { StgTSO *tso; @@ -2438,15 +3017,15 @@ 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) -static StgBlockingQueueElement * +#elif defined(PARALLEL_HASKELL) +StgBlockingQueueElement * unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node) { StgBlockingQueueElement *next; @@ -2456,9 +3035,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; @@ -2487,12 +3066,12 @@ 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; } -#else /* !GRAN && !PAR */ -static StgTSO * +#else /* !GRAN && !PARALLEL_HASKELL */ +StgTSO * unblockOneLocked(StgTSO *tso) { StgTSO *next; @@ -2501,14 +3080,15 @@ 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 -#if defined(GRAN) || defined(PAR) +#if defined(GRAN) || defined(PARALLEL_HASKELL) INLINE_ME StgBlockingQueueElement * unblockOne(StgBlockingQueueElement *bqe, StgClosure *node) { @@ -2537,7 +3117,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)); @@ -2554,13 +3134,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++; @@ -2595,7 +3175,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))); } @@ -2607,10 +3187,10 @@ 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) +#elif defined(PARALLEL_HASKELL) void awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node) { @@ -2619,12 +3199,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 @@ -2642,9 +3222,8 @@ awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node) RELEASE_LOCK(&sched_mutex); } -#else /* !GRAN && !PAR */ +#else /* !GRAN && !PARALLEL_HASKELL */ -#ifdef RTS_SUPPORTS_THREADS void awakenBlockedQueueNoLock(StgTSO *tso) { @@ -2652,7 +3231,6 @@ awakenBlockedQueueNoLock(StgTSO *tso) tso = unblockOneLocked(tso); } } -#endif void awakenBlockedQueue(StgTSO *tso) @@ -2675,9 +3253,11 @@ interruptStgRts(void) { interrupted = 1; context_switch = 1; -#ifdef RTS_SUPPORTS_THREADS - wakeBlockedWorkerThread(); -#endif + threadRunnable(); + /* ToDo: if invoked from a signal handler, this threadRunnable + * only works if there's another thread (not this one) waiting to + * be woken up. + */ } /* ----------------------------------------------------------------------------- @@ -2688,7 +3268,7 @@ interruptStgRts(void) This has nothing to do with the UnblockThread event in GranSim. -- HWL -------------------------------------------------------------------------- */ -#if defined(GRAN) || defined(PAR) +#if defined(GRAN) || defined(PARALLEL_HASKELL) /* NB: only the type of the blocking queue is different in GranSim and GUM the operations on the queue-elements are the same @@ -2707,6 +3287,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); { @@ -2773,7 +3361,7 @@ unblockThread(StgTSO *tso) case BlockedOnRead: case BlockedOnWrite: -#if defined(mingw32_TARGET_OS) +#if defined(mingw32_HOST_OS) case BlockedOnDoProc: #endif { @@ -2793,6 +3381,12 @@ unblockThread(StgTSO *tso) blocked_queue_tl = (StgTSO *)prev; } } +#if defined(mingw32_HOST_OS) + /* (Cooperatively) signal that the worker thread should abort + * the request. + */ + abandonWorkRequest(tso->block_info.async_result->reqID); +#endif goto done; } } @@ -2840,6 +3434,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); { @@ -2861,12 +3463,9 @@ unblockThread(StgTSO *tso) } case BlockedOnBlackHole: - ASSERT(get_itbl(tso->block_info.closure)->type == BLACKHOLE_BQ); { - StgBlockingQueue *bq = (StgBlockingQueue *)(tso->block_info.closure); - - last = &bq->blocking_queue; - for (t = bq->blocking_queue; t != END_TSO_QUEUE; + last = &blackhole_queue; + for (t = blackhole_queue; t != END_TSO_QUEUE; last = &t->link, t = t->link) { if (t == tso) { *last = tso->link; @@ -2903,7 +3502,7 @@ unblockThread(StgTSO *tso) case BlockedOnRead: case BlockedOnWrite: -#if defined(mingw32_TARGET_OS) +#if defined(mingw32_HOST_OS) case BlockedOnDoProc: #endif { @@ -2922,6 +3521,12 @@ unblockThread(StgTSO *tso) blocked_queue_tl = prev; } } +#if defined(mingw32_HOST_OS) + /* (Cooperatively) signal that the worker thread should abort + * the request. + */ + abandonWorkRequest(tso->block_info.async_result->reqID); +#endif goto done; } } @@ -2953,11 +3558,54 @@ 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 /* ----------------------------------------------------------------------------- + * checkBlackHoles() + * + * Check the blackhole_queue for threads that can be woken up. We do + * this periodically: before every GC, and whenever the run queue is + * empty. + * + * An elegant solution might be to just wake up all the blocked + * threads with awakenBlockedQueue occasionally: they'll go back to + * sleep again if the object is still a BLACKHOLE. Unfortunately this + * doesn't give us a way to tell whether we've actually managed to + * wake up any threads, so we would be busy-waiting. + * + * -------------------------------------------------------------------------- */ + +static rtsBool +checkBlackHoles( void ) +{ + StgTSO **prev, *t; + rtsBool any_woke_up = rtsFalse; + StgHalfWord type; + + IF_DEBUG(scheduler, sched_belch("checking threads blocked on black holes")); + + // ASSUMES: sched_mutex + prev = &blackhole_queue; + t = blackhole_queue; + while (t != END_TSO_QUEUE) { + ASSERT(t->why_blocked == BlockedOnBlackHole); + type = get_itbl(t->block_info.closure)->type; + if (type != BLACKHOLE && type != CAF_BLACKHOLE) { + t = unblockOneLocked(t); + *prev = t; + any_woke_up = rtsTrue; + } else { + prev = &t->link; + t = t->link; + } + } + + return any_woke_up; +} + +/* ----------------------------------------------------------------------------- * raiseAsync() * * The following function implements the magic for raising an @@ -2994,9 +3642,13 @@ 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 static void deleteThreadImmediately(StgTSO *tso) { // for forkProcess only: @@ -3005,13 +3657,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) @@ -3026,6 +3680,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; @@ -3035,7 +3695,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); @@ -3069,6 +3729,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; @@ -3077,13 +3741,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 @@ -3093,12 +3789,12 @@ raiseAsync(StgTSO *tso, StgClosure *exception) #ifdef PROFILING StgCatchFrame *cf = (StgCatchFrame *)frame; #endif - StgClosure *raise; + StgThunk *raise; // we've got an exception to raise, so let's pass it to the // handler in this frame. // - raise = (StgClosure *)allocate(sizeofW(StgClosure)+1); + raise = (StgThunk *)allocate(sizeofW(StgThunk)+1); TICK_ALLOC_SE_THK(1,0); SET_HDR(raise,&stg_raise_info,cf->header.prof.ccs); raise->payload[0] = exception; @@ -3136,7 +3832,7 @@ raiseAsync(StgTSO *tso, StgClosure *exception) // fun field. // words = frame - sp - 1; - ap = (StgAP_STACK *)allocate(PAP_sizeW(words)); + ap = (StgAP_STACK *)allocate(AP_STACK_sizeW(words)); ap->size = words; ap->fun = (StgClosure *)sp[0]; @@ -3150,9 +3846,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); ); @@ -3170,7 +3866,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 @@ -3192,6 +3889,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) +{ + StgThunk *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 = + (StgThunk *)allocate(sizeofW(StgThunk)+MIN_UPD_SIZE); + SET_HDR(raise_closure, &stg_raise_info, CCCS); + raise_closure->payload[0] = exception; + } + UPD_IND(((StgUpdateFrame *)p)->updatee,(StgClosure *)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 @@ -3221,6 +4049,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 @@ -3233,140 +4064,77 @@ 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 * at the Haskell source code level, so enable outside of DEBUG. --sof 7/02] ------------------------------------------------------------------------- */ -static -void +static void 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", (int)(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", (int)(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: %ld)", tso->block_info.async_result->reqID); break; #endif case BlockedOnDelay: - fprintf(stderr,"is blocked until %d", tso->block_info.target); + debugBelch("is blocked until %ld", (long)(tso->block_info.target)); break; case BlockedOnMVar: - fprintf(stderr,"is blocked on an MVar"); + debugBelch("is blocked on an MVar @ %p", tso->block_info.closure); 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) +#if defined(PARALLEL_HASKELL) 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; + case BlockedOnSTM: + debugBelch("is blocked on an STM operation"); break; -#endif default: barf("printThreadBlockage: strange tso->why_blocked: %d for TSO %d (%d)", tso->why_blocked, tso->id, tso); } } -static -void +static void 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); @@ -3377,39 +4145,66 @@ 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); -# elif defined(PAR) + debugBelch("all threads at [%s]:\n", time_string); +# elif defined(PARALLEL_HASKELL) char time_string[TIME_STR_LEN], node_str[NODE_STR_LEN]; ullong_format_string(CURRENT_TIME, time_string, rtsFalse/*no commas!*/); - 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); - printThreadStatus(t); - fprintf(stderr,"\n"); + for (t = all_threads; t != END_TSO_QUEUE; ) { + debugBelch("\tthread %4d @ %p ", t->id, (void *)t); +#if defined(DEBUG) + { + void *label = lookupThreadLabel(t->id); + if (label) debugBelch("[\"%s\"] ",(char *)label); + } +#endif + if (t->what_next == ThreadRelocated) { + debugBelch("has been relocated...\n"); + t = t->link; + } else { + printThreadStatus(t); + debugBelch("\n"); + t = t->global_link; + } } } - + #ifdef DEBUG +// useful from gdb +void +printThreadQueue(StgTSO *t) +{ + nat i = 0; + for (; t != END_TSO_QUEUE; t = t->link) { + debugBelch("\tthread %d @ %p ", t->id, (void *)t); + if (t->what_next == ThreadRelocated) { + debugBelch("has been relocated...\n"); + } else { + printThreadStatus(t); + debugBelch("\n"); + } + i++; + } + debugBelch("%d threads on queue\n", i); +} + /* Print a whole blocking queue attached to node (debugging only). */ -# if defined(PAR) +# if defined(PARALLEL_HASKELL) void print_bq (StgClosure *node) { @@ -3417,7 +4212,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 */ @@ -3457,18 +4252,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" : @@ -3480,7 +4275,7 @@ print_bqe (StgBlockingQueueElement *bqe) break; } } /* for */ - fputc('\n', stderr); + debugBelch("\n"); } # elif defined(GRAN) void @@ -3498,7 +4293,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); /* @@ -3518,11 +4313,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" : @@ -3534,30 +4329,11 @@ print_bq (StgClosure *node) break; } } /* for */ - fputc('\n', stderr); -} -#else -/* - Nice and easy: only TSOs on the blocking queue -*/ -void -print_bq (StgClosure *node) -{ - StgTSO *tso; - - ASSERT(node!=(StgClosure*)NULL); // sanity check - for (tso = ((StgBlockingQueue*)node)->blocking_queue; - tso != END_TSO_QUEUE; - 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); - } - fputc('\n', stderr); + debugBelch("\n"); } # endif -#if defined(PAR) +#if defined(PARALLEL_HASKELL) static nat run_queue_len(void) { @@ -3579,15 +4355,14 @@ sched_belch(char *s, ...) va_list ap; va_start(ap,s); #ifdef RTS_SUPPORTS_THREADS - fprintf(stderr, "sched (task %p): ", osThreadId()); -#elif defined(PAR) - fprintf(stderr, "== "); + debugBelch("sched (task %p): ", osThreadId()); +#elif defined(PARALLEL_HASKELL) + 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); }