X-Git-Url: http://git.megacz.com/?a=blobdiff_plain;f=ghc%2Frts%2FSchedule.c;h=5451cc5b33ff28b8527e40af9445d28c9ba61e85;hb=50027272414438955dbc41696541cbd25da55883;hp=d87f7ab99a5901334d66ee97a4a44ce207db54d6;hpb=1b28d4e1f43185ad8c8e7407c66413e1b358402b;p=ghc-hetmet.git diff --git a/ghc/rts/Schedule.c b/ghc/rts/Schedule.c index d87f7ab..5451cc5 100644 --- a/ghc/rts/Schedule.c +++ b/ghc/rts/Schedule.c @@ -1,20 +1,26 @@ /* --------------------------------------------------------------------------- - * $Id: Schedule.c,v 1.41 2000/01/13 14:34:05 hwloidl Exp $ + * $Id: Schedule.c,v 1.95 2001/03/23 16:36:21 simonmar Exp $ * - * (c) The GHC Team, 1998-1999 + * (c) The GHC Team, 1998-2000 * * Scheduler * - * The main scheduling code in GranSim is quite different from that in std - * (concurrent) Haskell: while concurrent Haskell just iterates over the - * threads in the runnable queue, GranSim is event driven, i.e. it iterates - * over the events in the global event queue. -- HWL + * Different GHC ways use this scheduler quite differently (see comments below) + * Here is the global picture: + * + * WAY Name CPP flag What's it for + * -------------------------------------- + * mp GUM PAR Parallel execution on a distributed memory machine + * s SMP SMP Parallel execution on a shared memory machine + * mg GranSim GRAN Simulation of parallel execution + * md GUM/GdH DIST Distributed execution (based on GUM) * --------------------------------------------------------------------------*/ //@node Main scheduling code, , , //@section Main scheduling code -/* Version with scheduler monitor support for SMPs. +/* + * Version with scheduler monitor support for SMPs (WAY=s): This design provides a high-level API to create and schedule threads etc. as documented in the SMP design document. @@ -32,12 +38,29 @@ In a non-SMP build, there is one global capability, namely MainRegTable. SDM & KH, 10/99 + + * Version with support for distributed memory parallelism aka GUM (WAY=mp): + + The main scheduling loop in GUM iterates until a finish message is received. + In that case a global flag @receivedFinish@ is set and this instance of + the RTS shuts down. See ghc/rts/parallel/HLComms.c:processMessages() + for the handling of incoming messages, such as PP_FINISH. + Note that in the parallel case we have a system manager that coordinates + different PEs, each of which are running one instance of the RTS. + See ghc/rts/parallel/SysMan.c for the main routine of the parallel program. + From this routine processes executing ghc/rts/Main.c are spawned. -- HWL + + * Version with support for simulating parallel execution aka GranSim (WAY=mg): + + The main scheduling code in GranSim is quite different from that in std + (concurrent) Haskell: while concurrent Haskell just iterates over the + threads in the runnable queue, GranSim is event driven, i.e. it iterates + over the events in the global event queue. -- HWL */ //@menu //* Includes:: //* Variables and Data structures:: -//* Prototypes:: //* Main scheduling loop:: //* Suspend and Resume:: //* Run queue code:: @@ -63,15 +86,15 @@ #include "Schedule.h" #include "StgMiscClosures.h" #include "Storage.h" -#include "Evaluator.h" +#include "Interpreter.h" #include "Exception.h" #include "Printer.h" #include "Main.h" #include "Signals.h" -#include "Profiling.h" #include "Sanity.h" #include "Stats.h" -#include "Sparks.h" +#include "Itimer.h" +#include "Prelude.h" #if defined(GRAN) || defined(PAR) # include "GranSimRts.h" # include "GranSim.h" @@ -81,6 +104,7 @@ # include "FetchMe.h" # include "HLC.h" #endif +#include "Sparks.h" #include @@ -120,29 +144,10 @@ static StgMainThread *main_threads; /* Thread queues. * Locks required: sched_mutex. */ - -#if DEBUG -char *whatNext_strs[] = { - "ThreadEnterGHC", - "ThreadRunGHC", - "ThreadEnterHugs", - "ThreadKilled", - "ThreadComplete" -}; - -char *threadReturnCode_strs[] = { - "HeapOverflow", /* might also be StackOverflow */ - "StackOverflow", - "ThreadYielding", - "ThreadBlocked", - "ThreadFinished" -}; -#endif - #if defined(GRAN) StgTSO* ActiveTSO = NULL; /* for assigning system costs; GranSim-Light only */ -// rtsTime TimeOfNextEvent, EndOfTimeSlice; now in GranSim.c +/* rtsTime TimeOfNextEvent, EndOfTimeSlice; now in GranSim.c */ /* In GranSim we have a runable and a blocked queue for each processor. @@ -154,24 +159,30 @@ StgTSO* ActiveTSO = NULL; /* for assigning system costs; GranSim-Light only */ StgTSO *run_queue_hds[MAX_PROC], *run_queue_tls[MAX_PROC]; StgTSO *blocked_queue_hds[MAX_PROC], *blocked_queue_tls[MAX_PROC]; StgTSO *ccalling_threadss[MAX_PROC]; +/* We use the same global list of threads (all_threads) in GranSim as in + the std RTS (i.e. we are cheating). However, we don't use this list in + the GranSim specific code at the moment (so we are only potentially + cheating). */ #else /* !GRAN */ -//@cindex run_queue_hd -//@cindex run_queue_tl -//@cindex blocked_queue_hd -//@cindex blocked_queue_tl StgTSO *run_queue_hd, *run_queue_tl; StgTSO *blocked_queue_hd, *blocked_queue_tl; +StgTSO *sleeping_queue; /* perhaps replace with a hash table? */ + +#endif + +/* Linked list of all threads. + * Used for detecting garbage collected threads. + */ +StgTSO *all_threads; /* Threads suspended in _ccall_GC. - * Locks required: sched_mutex. */ static StgTSO *suspended_ccalling_threads; static void GetRoots(void); static StgTSO *threadStackOverflow(StgTSO *tso); -#endif /* 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 @@ -224,11 +235,15 @@ Capability MainRegTable; /* for non-SMP, we have one global capability */ #endif #if defined(GRAN) -StgTSO *CurrentTSOs[MAX_PROC]; -#else -StgTSO *CurrentTSO; +StgTSO *CurrentTSO; #endif +/* This is used in `TSO.h' and gcc 2.96 insists that this variable actually + * exists - earlier gccs apparently didn't. + * -= chak + */ +StgTSO dummy_tso; + rtsBool ready_to_gc; /* All our current task ids, saved in case we need to kill them later. @@ -242,7 +257,13 @@ void addToBlockedQueue ( StgTSO *tso ); static void schedule ( void ); void interruptStgRts ( void ); +#if defined(GRAN) +static StgTSO * createThread_ ( nat size, rtsBool have_lock, StgInt pri ); +#else static StgTSO * createThread_ ( nat size, rtsBool have_lock ); +#endif + +static void detectBlackHoles ( void ); #ifdef DEBUG static void sched_belch(char *s, ...); @@ -264,6 +285,30 @@ nat await_death; #if defined(PAR) StgTSO *LastTSO; rtsTime TimeOfLastYield; +rtsBool emitSchedule = rtsTrue; +#endif + +#if DEBUG +char *whatNext_strs[] = { + "ThreadEnterGHC", + "ThreadRunGHC", + "ThreadEnterInterp", + "ThreadKilled", + "ThreadComplete" +}; + +char *threadReturnCode_strs[] = { + "HeapOverflow", /* might also be StackOverflow */ + "StackOverflow", + "ThreadYielding", + "ThreadBlocked", + "ThreadFinished" +}; +#endif + +#ifdef PAR +StgTSO * createSparkThread(rtsSpark spark); +StgTSO * activateSpark (rtsSpark spark); #endif /* @@ -272,20 +317,6 @@ rtsTime TimeOfLastYield; StgTSO *MainTSO; */ - -//@node Prototypes, Main scheduling loop, Variables and Data structures, Main scheduling code -//@subsection Prototypes - -#if 0 && defined(GRAN) -// ToDo: replace these with macros -static /* inline */ void add_to_run_queue(StgTSO* tso); -static /* inline */ void push_on_run_queue(StgTSO* tso); -static /* inline */ StgTSO *take_off_run_queue(StgTSO *tso); - -/* Thread management */ -void initScheduler(void); -#endif - //@node Main scheduling loop, Suspend and Resume, Prototypes, Main scheduling code //@subsection Main scheduling loop @@ -308,6 +339,21 @@ void initScheduler(void); * waiting for work, or * waiting for a GC to complete. + GRAN version: + In a GranSim setup this loop iterates over the global event queue. + This revolves around the global event queue, which determines what + to do next. Therefore, it's more complicated than either the + concurrent or the parallel (GUM) setup. + + GUM version: + GUM iterates over incoming messages. + It starts with nothing to do (thus CurrentTSO == END_TSO_QUEUE), + and sends out a fish whenever it has nothing to do; in-between + doing the actual reductions (shared code below) it processes the + incoming messages and deals with delayed operations + (see PendingFetches). + This is not the ugliest code you could imagine, but it's bloody close. + ------------------------------------------------------------------------ */ //@cindex schedule static void @@ -319,39 +365,64 @@ schedule( void ) #if defined(GRAN) rtsEvent *event; #elif defined(PAR) + StgSparkPool *pool; rtsSpark spark; StgTSO *tso; GlobalTaskId pe; + rtsBool receivedFinish = rtsFalse; +# if defined(DEBUG) + nat tp_size, sp_size; // stats only +# endif #endif + rtsBool was_interrupted = rtsFalse; ACQUIRE_LOCK(&sched_mutex); #if defined(GRAN) -# error ToDo: implement GranSim scheduler + + /* 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; + #elif defined(PAR) - while (!GlobalStopPending) { /* GlobalStopPending set in par_exit */ - if (PendingFetches != END_BF_QUEUE) { - processFetches(); - } + while (!receivedFinish) { /* set by processMessages */ + /* when receiving PP_FINISH message */ #else + while (1) { + #endif + IF_DEBUG(scheduler, printAllThreads()); + /* 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")); - for (t = run_queue_hd; t != END_TSO_QUEUE; t = t->link) { - deleteThread(t); - } - for (t = blocked_queue_hd; t != END_TSO_QUEUE; t = t->link) { - deleteThread(t); - } - run_queue_hd = run_queue_tl = END_TSO_QUEUE; - blocked_queue_hd = blocked_queue_tl = END_TSO_QUEUE; + deleteAllThreads(); + interrupted = rtsFalse; + was_interrupted = rtsTrue; } /* Go through the list of main threads and wake up any @@ -364,7 +435,7 @@ schedule( void ) StgMainThread *m, **prev; prev = &main_threads; for (m = main_threads; m != NULL; m = m->link) { - switch (m->tso->whatNext) { + switch (m->tso->what_next) { case ThreadComplete: if (m->ret) { *(m->ret) = (StgClosure *)m->tso->sp[0]; @@ -375,7 +446,11 @@ schedule( void ) break; case ThreadKilled: *prev = m->link; - m->stat = Killed; + if (was_interrupted) { + m->stat = Interrupted; + } else { + m->stat = Killed; + } pthread_cond_broadcast(&m->wakeup); break; default: @@ -383,21 +458,30 @@ schedule( void ) } } } + #else +# if defined(PAR) + /* in GUM do this only on the Main PE */ + if (IAmMainThread) +# endif /* If our main thread has finished or been killed, return. */ { StgMainThread *m = main_threads; - if (m->tso->whatNext == ThreadComplete - || m->tso->whatNext == ThreadKilled) { + if (m->tso->what_next == ThreadComplete + || m->tso->what_next == ThreadKilled) { main_threads = main_threads->link; - if (m->tso->whatNext == ThreadComplete) { + if (m->tso->what_next == ThreadComplete) { /* we finished successfully, fill in the return value */ if (m->ret) { *(m->ret) = (StgClosure *)m->tso->sp[0]; }; m->stat = Success; return; } else { - m->stat = Killed; + if (was_interrupted) { + m->stat = Interrupted; + } else { + m->stat = Killed; + } return; } } @@ -421,21 +505,16 @@ schedule( void ) for (; n > 0; n--) { StgClosure *spark; - spark = findSpark(); + spark = findSpark(rtsFalse); if (spark == NULL) { break; /* no more sparks in the pool */ } else { - // I'd prefer this to be done in activateSpark -- HWL - StgTSO *tso; - tso = createThread_(RtsFlags.GcFlags.initialStkSize, rtsTrue); - pushClosure(tso,spark); - PUSH_ON_RUN_QUEUE(tso); -#ifdef PAR - advisory_thread_count++; -#endif - + /* I'd prefer this to be done in activateSpark -- HWL */ + /* tricky - it needs to hold the scheduler lock and + * not try to re-acquire it -- SDM */ + createSparkThread(spark); IF_DEBUG(scheduler, - sched_belch("turning spark of closure %p into a thread", + sched_belch("==^^ turning spark of closure %p into a thread", (StgClosure *)spark)); } } @@ -454,7 +533,7 @@ schedule( void ) * ToDo: what if another client comes along & requests another * main thread? */ - if (blocked_queue_hd != END_TSO_QUEUE) { + if (blocked_queue_hd != END_TSO_QUEUE || sleeping_queue != END_TSO_QUEUE) { awaitEvent( (run_queue_hd == END_TSO_QUEUE) #ifdef SMP @@ -462,40 +541,61 @@ schedule( void ) #endif ); } - + /* we can be interrupted while waiting for I/O... */ + if (interrupted) continue; + /* check for signals each time around the scheduler */ -#ifndef __MINGW32__ +#ifndef mingw32_TARGET_OS if (signals_pending()) { start_signal_handlers(); } #endif - /* 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. Inform - * all the main threads. + /* + * 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. */ #ifdef SMP if (blocked_queue_hd == END_TSO_QUEUE && run_queue_hd == END_TSO_QUEUE - && (n_free_capabilities == RtsFlags.ParFlags.nNodes) - ) { - StgMainThread *m; - for (m = main_threads; m != NULL; m = m->link) { - m->ret = NULL; - m->stat = Deadlock; - pthread_cond_broadcast(&m->wakeup); - } - main_threads = NULL; + && sleeping_queue == END_TSO_QUEUE + && (n_free_capabilities == RtsFlags.ParFlags.nNodes)) + { + IF_DEBUG(scheduler, sched_belch("deadlocked, checking for black holes...")); + detectBlackHoles(); + if (run_queue_hd == END_TSO_QUEUE) { + StgMainThread *m; + for (m = main_threads; m != NULL; m = m->link) { + m->ret = NULL; + m->stat = Deadlock; + pthread_cond_broadcast(&m->wakeup); + } + main_threads = NULL; + } } +#elif defined(PAR) + /* ToDo: add deadlock detection in GUM (similar to SMP) -- HWL */ #else /* ! SMP */ if (blocked_queue_hd == END_TSO_QUEUE - && run_queue_hd == END_TSO_QUEUE) { - StgMainThread *m = main_threads; - m->ret = NULL; - m->stat = Deadlock; - main_threads = m->link; - return; + && run_queue_hd == END_TSO_QUEUE + && sleeping_queue == END_TSO_QUEUE) + { + IF_DEBUG(scheduler, sched_belch("deadlocked, checking for black holes...")); + detectBlackHoles(); + if (run_queue_hd == END_TSO_QUEUE) { + StgMainThread *m = main_threads; + m->ret = NULL; + m->stat = Deadlock; + main_threads = m->link; + return; + } } #endif @@ -519,15 +619,155 @@ schedule( void ) #endif #if defined(GRAN) -# error ToDo: implement GranSim scheduler + + 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, fprintf(stderr, "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")); + /* ToDo: check assertion + ASSERT(run_queue_hd != (StgTSO*)NULL && + run_queue_hd != END_TSO_QUEUE); + */ + /* 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]", + 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)", + 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)", + 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", + CurrentTSO->id, CurrentTSO, CurrentProc); + break; // run the thread anyway + } + /* + new_event(proc, proc, CurrentTime[proc], + FindWork, + (StgTSO*)NULL, (StgClosure*)NULL, (rtsSpark*)NULL); + goto next_thread; + */ /* Catches superfluous CONTINUEs -- should be unnecessary */ + break; // now actually run the thread; DaH Qu'vam yImuHbej + + case FetchNode: + do_the_fetchnode(event); + goto next_thread; /* handle next event in event queue */ + + case GlobalBlock: + do_the_globalblock(event); + goto next_thread; /* handle next event in event queue */ + + case FetchReply: + do_the_fetchreply(event); + goto next_thread; /* handle next event in event queue */ + + case UnblockThread: /* Move from the blocked queue to the tail of */ + do_the_unblock(event); + goto next_thread; /* handle next event in event queue */ + + case ResumeThread: /* Move from the blocked queue to the tail of */ + /* the runnable queue ( i.e. Qu' SImqa'lu') */ + event->tso->gran.blocktime += + CurrentTime[CurrentProc] - event->tso->gran.blockedat; + do_the_startthread(event); + goto next_thread; /* handle next event in event queue */ + + case StartThread: + do_the_startthread(event); + goto next_thread; /* handle next event in event queue */ + + case MoveThread: + do_the_movethread(event); + goto next_thread; /* handle next event in event queue */ + + case MoveSpark: + do_the_movespark(event); + goto next_thread; /* handle next event in event queue */ + + case FindWork: + do_the_findwork(event); + goto next_thread; /* handle next event in event queue */ + + default: + barf("Illegal event type %u\n", event->evttype); + } /* switch */ + + /* This point was scheduler_loop in the old RTS */ + + IF_DEBUG(gran, belch("GRAN: after main switch")); + + 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", + TimeOfNextEvent)); + + if (RtsFlags.GranFlags.Light) + GranSimLight_leave_system(event, &ActiveTSO); + + EndOfTimeSlice = CurrentTime[CurrentProc]+RtsFlags.GranFlags.time_slice; + + IF_DEBUG(gran, + belch("GRAN: end of time-slice is %#lx", EndOfTimeSlice)); + + /* in a GranSim setup the TSO stays on the run queue */ + t = CurrentTSO; + /* Take a thread from the run queue. */ + t = POP_RUN_QUEUE(); // take_off_run_queue(t); + + IF_DEBUG(gran, + fprintf(stderr, "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 + + IF_DEBUG(gran, + DumpGranEvent(GR_SCHEDULE, t)); + + procStatus[CurrentProc] = Busy; + #elif defined(PAR) - // ToDo: phps merge with spark activation above + if (PendingFetches != END_BF_QUEUE) { + processFetches(); + } + + /* ToDo: phps merge with spark activation above */ /* check whether we have local work and send requests if we have none */ - if (run_queue_hd == END_TSO_QUEUE) { /* no runnable threads */ + 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 if (advisory_thread_count < RtsFlags.ParFlags.maxThreads && - (pending_sparks_hd[REQUIRED_POOL] < pending_sparks_tl[REQUIRED_POOL] || - pending_sparks_hd[ADVISORY_POOL] < pending_sparks_tl[ADVISORY_POOL])) { + pool->hd < pool->tl) { /* * ToDo: add GC code check that we really have enough heap afterwards!! * Old comment: @@ -536,27 +776,31 @@ schedule( void ) * to turn one of those pending sparks into a * thread... */ - - spark = findSpark(); /* get a spark */ + + spark = findSpark(rtsFalse); /* get a spark */ if (spark != (rtsSpark) NULL) { tso = activateSpark(spark); /* turn the spark into a thread */ - IF_PAR_DEBUG(verbose, - belch("== [%x] schedule: Created TSO %p (%d); %d threads active", - mytid, tso, tso->id, advisory_thread_count)); + 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"); + 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 + } /* 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(ADVISORY_POOL))); + belch("==^^ no local sparks (spark pool contains only NFs: %d)", + spark_queue_len(pool))); goto next_thread; } - } else + } + + /* 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 * (there may be some out there already), and wait for @@ -565,47 +809,75 @@ schedule( void ) * we're hoping to see. (Of course, we still have to * respond to other types of messages.) */ - if (//!fishing && - outstandingFishes < RtsFlags.ParFlags.maxFishes ) { // && - // (last_fish_arrived_at+FISH_DELAY < CURRENT_TIME)) { - /* fishing set in sendFish, processFish; + TIME 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); + }); + + 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++; + } } - - processMessages(); + + receivedFinish = processMessages(); goto next_thread; - // ReSchedule(0); } } else if (PacketsWaiting()) { /* Look for incoming messages */ - processMessages(); + receivedFinish = processMessages(); } /* 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 */ - t = take_off_run_queue(END_TSO_QUEUE); + t = POP_RUN_QUEUE(); // take_off_run_queue(END_TSO_QUEUE); + IF_DEBUG(sanity,checkTSO(t)); /* 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 + + IF_DEBUG(scheduler, + belch("--=^ %d threads, %d sparks on [%#x]", + 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_DEBUG(scheduler, belch("--^^ %d sparks on [%#x] (hd=%x; tl=%x; lim=%x)", - spark_queue_len(ADVISORY_POOL), CURRENT_PROC, - pending_sparks_hd[ADVISORY_POOL], - pending_sparks_tl[ADVISORY_POOL], - pending_sparks_lim[ADVISORY_POOL])); - - IF_DEBUG(scheduler, belch("--== %d threads on [%#x] (hd=%x; tl=%x)", - run_queue_len(), CURRENT_PROC, - run_queue_hd, run_queue_tl)); - - if (t != LastTSO) { + if (RtsFlags.ParFlags.ParStats.Full && + (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 @@ -615,14 +887,17 @@ schedule( void ) */ 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); t = POP_RUN_QUEUE(); + IF_DEBUG(sanity,checkTSO(t)); #endif @@ -638,47 +913,43 @@ schedule( void ) cap->rCurrentTSO = t; - /* set the context_switch flag + /* context switches are now initiated by the timer signal, unless + * the user specified "context switch as often as possible", with + * +RTS -C0 */ - if (run_queue_hd == END_TSO_QUEUE) - context_switch = 0; + 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 = 1; + context_switch = 0; RELEASE_LOCK(&sched_mutex); - - IF_DEBUG(scheduler,sched_belch("running thread %d", t->id)); + + IF_DEBUG(scheduler, sched_belch("-->> Running TSO %ld (%p) %s ...", + t->id, t, whatNext_strs[t->what_next])); /* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */ /* Run the current thread */ - switch (cap->rCurrentTSO->whatNext) { + switch (cap->rCurrentTSO->what_next) { case ThreadKilled: case ThreadComplete: - /* Thread already finished, return to scheduler. */ - ret = ThreadFinished; - break; + /* Thread already finished, return to scheduler. */ + ret = ThreadFinished; + break; case ThreadEnterGHC: - ret = StgRun((StgFunPtr) stg_enterStackTop, cap); - break; + ret = StgRun((StgFunPtr) stg_enterStackTop, cap); + break; case ThreadRunGHC: - ret = StgRun((StgFunPtr) stg_returnToStackTop, cap); - break; - case ThreadEnterHugs: -#ifdef INTERPRETER - { - StgClosure* c; - IF_DEBUG(scheduler,sched_belch("entering Hugs")); - c = (StgClosure *)(cap->rCurrentTSO->sp[0]); - cap->rCurrentTSO->sp += 1; - ret = enter(cap,c); - break; - } -#else - barf("Panic: entered a BCO but no bytecode interpreter in this build"); -#endif + ret = StgRun((StgFunPtr) stg_returnToStackTop, cap); + break; + case ThreadEnterInterp: + ret = interpretBCO(cap); + break; default: - barf("schedule: invalid whatNext field"); + barf("schedule: invalid what_next field"); } /* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */ @@ -691,30 +962,71 @@ schedule( void ) #ifdef SMP IF_DEBUG(scheduler,fprintf(stderr,"scheduler (task %ld): ", pthread_self());); -#else +#elif !defined(GRAN) && !defined(PAR) IF_DEBUG(scheduler,fprintf(stderr,"scheduler: ");); #endif t = cap->rCurrentTSO; +#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 + if it is the same one as before */ + LastTSO = t; + TimeOfLastYield = CURRENT_TIME; +#endif + switch (ret) { case HeapOverflow: +#if defined(GRAN) + IF_DEBUG(gran, + DumpGranEvent(GR_DESCHEDULE, t)); + globalGranStats.tot_heapover++; +#elif defined(PAR) + // IF_DEBUG(par, + //DumpGranEvent(GR_DESCHEDULE, t); + globalParStats.tot_heapover++; +#endif /* 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 stopped: HeapOverflow", t->id)); + IF_DEBUG(scheduler,belch("--<< thread %ld (%p; %s) stopped: HeapOverflow", + t->id, t, 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 */ break; case StackOverflow: +#if defined(GRAN) + IF_DEBUG(gran, + DumpGranEvent(GR_DESCHEDULE, t)); + globalGranStats.tot_stackover++; +#elif defined(PAR) + // IF_DEBUG(par, + // DumpGranEvent(GR_DESCHEDULE, t); + globalParStats.tot_stackover++; +#endif + IF_DEBUG(scheduler,belch("--<< thread %ld (%p; %s) stopped, StackOverflow", + t->id, t, whatNext_strs[t->what_next])); /* just adjust the stack for this thread, then pop it back * on the run queue. */ - IF_DEBUG(scheduler,belch("thread %ld stopped, StackOverflow", t->id)); threadPaused(t); { StgMainThread *m; @@ -723,13 +1035,14 @@ schedule( void ) /* 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) + * (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); } break; @@ -740,8 +1053,9 @@ schedule( void ) DumpGranEvent(GR_DESCHEDULE, t)); globalGranStats.tot_yields++; #elif defined(PAR) - IF_DEBUG(par, - DumpGranEvent(GR_DESCHEDULE, t)); + // 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 @@ -749,36 +1063,107 @@ schedule( void ) * GC is finished. */ IF_DEBUG(scheduler, - if (t->whatNext == ThreadEnterHugs) { - /* ToDo: or maybe a timer expired when we were in Hugs? - * or maybe someone hit ctrl-C - */ - belch("thread %ld stopped to switch to Hugs", t->id); - } else { - belch("thread %ld stopped, yielding", t->id); - } - ); + if (t->what_next == ThreadEnterInterp) { + /* ToDo: or maybe a timer expired when we were in Hugs? + * or maybe someone hit ctrl-C + */ + belch("--<< thread %ld (%p; %s) stopped to switch to Hugs", + t->id, t, whatNext_strs[t->what_next]); + } else { + belch("--<< thread %ld (%p; %s) stopped, yielding", + t->id, t, whatNext_strs[t->what_next]); + } + ); + threadPaused(t); + + IF_DEBUG(sanity, + //belch("&& Doing sanity check on yielding TSO %ld.", t->id); + checkTSO(t)); + ASSERT(t->link == END_TSO_QUEUE); +#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; case ThreadBlocked: #if defined(GRAN) -# error ToDo: implement GranSim scheduler -#elif defined(PAR) - IF_DEBUG(par, + 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)); + + // ??? needed; should emit block before + IF_DEBUG(gran, DumpGranEvent(GR_DESCHEDULE, t)); -#else -#endif + 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, + 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, + + 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 stopped, ", t->id); + fprintf(stderr, "--<< thread %d (%p) stopped: ", t->id, t); printThreadBlockage(t); fprintf(stderr, "\n")); + + /* Only for dumping event to log file + ToDo: do I need this in GranSim, too? + blockThread(t); + */ +#endif threadPaused(t); break; @@ -788,19 +1173,30 @@ schedule( void ) * more main threads, we probably need to stop all the tasks until * we get a new one. */ - IF_DEBUG(scheduler,belch("thread %ld finished", t->id)); - t->whatNext = ThreadComplete; + /* We also end up here if the thread kills itself with an + * uncaught exception, see Exception.hc. + */ + IF_DEBUG(scheduler,belch("--++ thread %d (%p) finished", t->id, t)); #if defined(GRAN) - // ToDo: endThread(t, CurrentProc); // clean-up the thread + endThread(t, CurrentProc); // clean-up the thread #elif defined(PAR) + /* For now all are advisory -- HWL */ + //if(t->priority==AdvisoryPriority) ?? advisory_thread_count--; - if (RtsFlags.ParFlags.ParStats.Full) + +# ifdef DIST + if(t->dist.priority==RevalPriority) + FinishReval(t); +# endif + + if (RtsFlags.ParFlags.ParStats.Full && + !RtsFlags.ParFlags.ParStats.Suppressed) DumpEndEvent(CURRENT_PROC, t, rtsFalse /* not mandatory */); #endif break; default: - barf("doneThread: invalid thread return code"); + barf("schedule: invalid thread return code %d", (int)ret); } #ifdef SMP @@ -823,11 +1219,21 @@ schedule( void ) #ifdef SMP IF_DEBUG(scheduler,sched_belch("doing GC")); #endif - GarbageCollect(GetRoots); + GarbageCollect(GetRoots,rtsFalse); ready_to_gc = rtsFalse; #ifdef SMP pthread_cond_broadcast(&gc_pending_cond); #endif +#if defined(GRAN) + /* add a ContinueThread event to continue execution of current thread */ + new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc], + ContinueThread, + t, (StgClosure*)NULL, (rtsSpark*)NULL); + IF_GRAN_DEBUG(bq, + fprintf(stderr, "GRAN: eventq and runnableq after Garbage collection:\n"); + G_EVENTQ(0); + G_CURR_THREADQ(0)); +#endif /* GRAN */ } #if defined(GRAN) next_thread: @@ -847,21 +1253,33 @@ schedule( void ) */ #endif /* GRAN */ } /* end of while(1) */ + IF_PAR_DEBUG(verbose, + belch("== Leaving schedule() after having received Finish")); } -/* A hack for Hugs concurrency support. Needs sanitisation (?) */ +/* --------------------------------------------------------------------------- + * deleteAllThreads(): kill all the live threads. + * + * This is used when we catch a user interrupt (^C), before performing + * any necessary cleanups and running finalizers. + * ------------------------------------------------------------------------- */ + void deleteAllThreads ( void ) { StgTSO* t; - IF_DEBUG(scheduler,sched_belch("deleteAllThreads()")); + IF_DEBUG(scheduler,sched_belch("deleting all threads")); for (t = run_queue_hd; t != END_TSO_QUEUE; t = t->link) { - deleteThread(t); + deleteThread(t); } for (t = blocked_queue_hd; t != END_TSO_QUEUE; t = t->link) { - deleteThread(t); + deleteThread(t); + } + for (t = sleeping_queue; t != END_TSO_QUEUE; t = t->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; } /* startThread and insertThread are now in GranSim.c -- HWL */ @@ -892,7 +1310,7 @@ suspendThread( Capability *cap ) ACQUIRE_LOCK(&sched_mutex); IF_DEBUG(scheduler, - sched_belch("thread %d did a _ccall_gc\n", cap->rCurrentTSO->id)); + sched_belch("thread %d did a _ccall_gc", cap->rCurrentTSO->id)); threadPaused(cap->rCurrentTSO); cap->rCurrentTSO->link = suspended_ccalling_threads; @@ -931,6 +1349,7 @@ resumeThread( StgInt tok ) if (tso == END_TSO_QUEUE) { barf("resumeThread: thread not found"); } + tso->link = END_TSO_QUEUE; #ifdef SMP while (free_capabilities == NULL) { @@ -984,10 +1403,12 @@ int cmp_thread(const StgTSO *tso1, const StgTSO *tso2) createGenThread() and createIOThread() (in SchedAPI.h) are convenient packaged versions of this function. + + currently pri (priority) is only used in a GRAN setup -- HWL ------------------------------------------------------------------------ */ //@cindex createThread #if defined(GRAN) -/* currently pri (priority) is only used in a GRAN setup -- HWL */ +/* currently pri (priority) is only used in a GRAN setup -- HWL */ StgTSO * createThread(nat stack_size, StgInt pri) { @@ -1008,6 +1429,7 @@ static StgTSO * createThread_(nat size, rtsBool have_lock) { #endif + StgTSO *tso; nat stack_size; @@ -1034,23 +1456,21 @@ createThread_(nat size, rtsBool have_lock) size = MIN_STACK_WORDS + TSO_STRUCT_SIZEW; } - tso = (StgTSO *)allocate(size); - TICK_ALLOC_TSO(size-sizeofW(StgTSO),0); - stack_size = size - TSO_STRUCT_SIZEW; - // Hmm, this CCS_MAIN is not protected by a PROFILING cpp var; - SET_HDR(tso, &TSO_info, CCS_MAIN); + tso = (StgTSO *)allocate(size); + TICK_ALLOC_TSO(size-TSO_STRUCT_SIZEW, 0); + + SET_HDR(tso, &stg_TSO_info, CCS_SYSTEM); #if defined(GRAN) SET_GRAN_HDR(tso, ThisPE); #endif - tso->whatNext = ThreadEnterGHC; + tso->what_next = ThreadEnterGHC; /* tso->id needs to be unique. For now we use a heavyweight mutex to - protect the increment operation on next_thread_id. - In future, we could use an atomic increment instead. - */ - + * protect the increment operation on next_thread_id. + * In future, we could use an atomic increment instead. + */ if (!have_lock) { ACQUIRE_LOCK(&sched_mutex); } tso->id = next_thread_id++; if (!have_lock) { RELEASE_LOCK(&sched_mutex); } @@ -1058,7 +1478,6 @@ createThread_(nat size, rtsBool have_lock) tso->why_blocked = NotBlocked; tso->blocked_exceptions = NULL; - tso->splim = (P_)&(tso->stack) + RESERVED_STACK_WORDS; tso->stack_size = stack_size; tso->max_stack_size = round_to_mblocks(RtsFlags.GcFlags.maxStkSize) - TSO_STRUCT_SIZEW; @@ -1070,24 +1489,44 @@ createThread_(nat size, rtsBool have_lock) /* put a stop frame on the stack */ tso->sp -= sizeofW(StgStopFrame); - SET_HDR((StgClosure*)tso->sp,(StgInfoTable *)&stg_stop_thread_info,CCS_MAIN); + SET_HDR((StgClosure*)tso->sp,(StgInfoTable *)&stg_stop_thread_info,CCS_SYSTEM); tso->su = (StgUpdateFrame*)tso->sp; - IF_DEBUG(scheduler,belch("---- Initialised TSO %ld (%p), stack size = %lx words", - tso->id, tso, tso->stack_size)); - // ToDo: check this #if defined(GRAN) tso->link = END_TSO_QUEUE; /* uses more flexible routine in GranSim */ insertThread(tso, CurrentProc); #else - add_to_run_queue(tso); + /* In a non-GranSim setup the pushing of a TSO onto the runq is separated + * from its creation + */ +#endif + +#if defined(GRAN) + if (RtsFlags.GranFlags.GranSimStats.Full) + DumpGranEvent(GR_START,tso); +#elif defined(PAR) + if (RtsFlags.ParFlags.ParStats.Full) + DumpGranEvent(GR_STARTQ,tso); + /* HACk to avoid SCHEDULE + LastTSO = tso; */ +#endif + + /* Link the new thread on the global thread list. + */ + tso->global_link = all_threads; + all_threads = tso; + +#if defined(DIST) + tso->dist.priority = MandatoryPriority; //by default that is... #endif #if defined(GRAN) tso->gran.pri = pri; +# if defined(DEBUG) tso->gran.magic = TSO_MAGIC; // debugging only +# endif tso->gran.sparkname = 0; tso->gran.startedat = CURRENT_TIME; tso->gran.exported = 0; @@ -1108,6 +1547,9 @@ createThread_(nat size, rtsBool have_lock) IF_DEBUG(gran,printTSO(tso)); #elif defined(PAR) +# if defined(DEBUG) + tso->par.magic = TSO_MAGIC; // debugging only +# endif tso->par.sparkname = 0; tso->par.startedat = CURRENT_TIME; tso->par.exported = 0; @@ -1128,12 +1570,86 @@ createThread_(nat size, rtsBool have_lock) globalGranStats.threads_created_on_PE[CurrentProc]++; globalGranStats.tot_sq_len += spark_queue_len(CurrentProc); globalGranStats.tot_sq_probes++; +#elif defined(PAR) + // 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()); + globalParStats.tot_threads_created++; + } #endif +#if defined(GRAN) + IF_GRAN_DEBUG(pri, + 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)); +#else IF_DEBUG(scheduler,sched_belch("created thread %ld, stack size = %lx words", tso->id, tso->stack_size)); +#endif + return tso; +} + +#if defined(PAR) +/* RFP: + all parallel thread creation calls should fall through the following routine. +*/ +StgTSO * +createSparkThread(rtsSpark spark) +{ StgTSO *tso; + ASSERT(spark != (rtsSpark)NULL); + 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)", + RtsFlags.ParFlags.maxThreads, advisory_thread_count); + return END_TSO_QUEUE; + } + else + { threadsCreated++; + tso = createThread_(RtsFlags.GcFlags.initialStkSize, rtsTrue); + if (tso==END_TSO_QUEUE) + barf("createSparkThread: Cannot create TSO"); +#if defined(DIST) + tso->priority = AdvisoryPriority; +#endif + pushClosure(tso,spark); + PUSH_ON_RUN_QUEUE(tso); + advisory_thread_count++; + } + return tso; +} +#endif + +/* + Turn a spark into a thread. + ToDo: fix for SMP (needs to acquire SCHED_MUTEX!) +*/ +#if defined(PAR) +//@cindex activateSpark +StgTSO * +activateSpark (rtsSpark spark) +{ + StgTSO *tso; + + 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))); + } + // ToDo: fwd info on local/global spark to thread -- HWL + // tso->gran.exported = spark->exported; + // tso->gran.locked = !spark->global; + // tso->gran.sparkname = spark->name; + return tso; } +#endif /* --------------------------------------------------------------------------- * scheduleThread() @@ -1148,6 +1664,11 @@ createThread_(nat size, rtsBool have_lock) void scheduleThread(StgTSO *tso) { + if (tso==END_TSO_QUEUE){ + schedule(); + return; + } + ACQUIRE_LOCK(&sched_mutex); /* Put the new thread on the head of the runnable queue. The caller @@ -1158,7 +1679,9 @@ scheduleThread(StgTSO *tso) PUSH_ON_RUN_QUEUE(tso); THREAD_RUNNABLE(); +#if 0 IF_DEBUG(scheduler,printTSO(tso)); +#endif RELEASE_LOCK(&sched_mutex); } @@ -1170,12 +1693,11 @@ scheduleThread(StgTSO *tso) * KH @ 25/10/99 * ------------------------------------------------------------------------ */ -#ifdef SMP -static void * -taskStart( void *arg STG_UNUSED ) +#if defined(PAR) || defined(SMP) +void +taskStart(void) /* ( void *arg STG_UNUSED) */ { - schedule(); - return NULL; + scheduleThread(END_TSO_QUEUE); } #endif @@ -1214,22 +1736,26 @@ initScheduler(void) blocked_queue_hds[i] = END_TSO_QUEUE; blocked_queue_tls[i] = END_TSO_QUEUE; ccalling_threadss[i] = END_TSO_QUEUE; + sleeping_queue = END_TSO_QUEUE; } #else run_queue_hd = END_TSO_QUEUE; run_queue_tl = END_TSO_QUEUE; blocked_queue_hd = END_TSO_QUEUE; blocked_queue_tl = END_TSO_QUEUE; + sleeping_queue = END_TSO_QUEUE; #endif suspended_ccalling_threads = END_TSO_QUEUE; main_threads = NULL; + all_threads = END_TSO_QUEUE; context_switch = 0; interrupted = 0; - enteredCAFs = END_CAF_LIST; + RtsFlags.ConcFlags.ctxtSwitchTicks = + RtsFlags.ConcFlags.ctxtSwitchTime / TICK_MILLISECS; /* Install the SIGHUP handler */ #ifdef SMP @@ -1344,7 +1870,7 @@ exitScheduler( void ) -------------------------------------------------------------------------- */ /* ----------------------------------------------------------------------------- - * waitThread is the external interface for running a new computataion + * waitThread is the external interface for running a new computation * and waiting for the result. * * In the non-SMP case, we create a new main thread, push it on the @@ -1359,6 +1885,39 @@ exitScheduler( void ) * will be in the main_thread struct. * -------------------------------------------------------------------------- */ +int +howManyThreadsAvail ( void ) +{ + int i = 0; + StgTSO* q; + for (q = run_queue_hd; q != END_TSO_QUEUE; q = q->link) + i++; + for (q = blocked_queue_hd; q != END_TSO_QUEUE; q = q->link) + i++; + for (q = sleeping_queue; q != END_TSO_QUEUE; q = q->link) + i++; + return i; +} + +void +finishAllThreads ( void ) +{ + do { + while (run_queue_hd != END_TSO_QUEUE) { + waitThread ( run_queue_hd, NULL ); + } + while (blocked_queue_hd != END_TSO_QUEUE) { + waitThread ( blocked_queue_hd, NULL ); + } + while (sleeping_queue != END_TSO_QUEUE) { + waitThread ( blocked_queue_hd, NULL ); + } + } while + (blocked_queue_hd != END_TSO_QUEUE || + run_queue_hd != END_TSO_QUEUE || + sleeping_queue != END_TSO_QUEUE); +} + SchedulerStatus waitThread(StgTSO *tso, /*out*/StgClosure **ret) { @@ -1379,13 +1938,20 @@ waitThread(StgTSO *tso, /*out*/StgClosure **ret) m->link = main_threads; main_threads = m; - IF_DEBUG(scheduler, fprintf(stderr, "scheduler: new main thread (%d)\n", + IF_DEBUG(scheduler, fprintf(stderr, "== scheduler: new main thread (%d)\n", m->tso->id)); #ifdef SMP do { pthread_cond_wait(&m->wakeup, &sched_mutex); } while (m->stat == NoStatus); +#elif defined(GRAN) + /* GranSim specific init */ + CurrentTSO = m->tso; // the TSO to run + procStatus[MainProc] = Busy; // status of main PE + CurrentProc = MainProc; // PE to run it on + + schedule(); #else schedule(); ASSERT(m->stat != NoStatus); @@ -1397,7 +1963,7 @@ waitThread(StgTSO *tso, /*out*/StgClosure **ret) pthread_cond_destroy(&m->wakeup); #endif - IF_DEBUG(scheduler, fprintf(stderr, "scheduler: main thread (%d) finished\n", + IF_DEBUG(scheduler, fprintf(stderr, "== scheduler: main thread (%d) finished\n", m->tso->id)); free(m); @@ -1507,21 +2073,6 @@ take_off_run_queue(StgTSO *tso) { #endif /* 0 */ -nat -run_queue_len(void) -{ - nat i; - StgTSO *tso; - - for (i=0, tso=run_queue_hd; - tso != END_TSO_QUEUE; - i++, tso=tso->link) - /* nothing */ - - return i; -} - - //@node Garbage Collextion Routines, Blocking Queue Routines, Run queue code, Main scheduling code //@subsection Garbage Collextion Routines @@ -1530,6 +2081,7 @@ run_queue_len(void) - all the threads on the runnable queue - all the threads on the blocked queue + - all the threads on the sleeping queue - all the thread currently executing a _ccall_GC - all the "main threads" @@ -1543,42 +2095,51 @@ run_queue_len(void) static void GetRoots(void) { StgMainThread *m; - nat i; #if defined(GRAN) - for (i=0; i<=RtsFlags.GranFlags.proc; i++) { - if ((run_queue_hds[i] != END_TSO_QUEUE) && ((run_queue_hds[i] != NULL))) - run_queue_hds[i] = (StgTSO *)MarkRoot((StgClosure *)run_queue_hds[i]); - if ((run_queue_tls[i] != END_TSO_QUEUE) && ((run_queue_tls[i] != NULL))) - run_queue_tls[i] = (StgTSO *)MarkRoot((StgClosure *)run_queue_tls[i]); - - if ((blocked_queue_hds[i] != END_TSO_QUEUE) && ((blocked_queue_hds[i] != NULL))) - blocked_queue_hds[i] = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_hds[i]); - if ((blocked_queue_tls[i] != END_TSO_QUEUE) && ((blocked_queue_tls[i] != NULL))) - blocked_queue_tls[i] = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_tls[i]); - if ((ccalling_threadss[i] != END_TSO_QUEUE) && ((ccalling_threadss[i] != NULL))) - ccalling_threadss[i] = (StgTSO *)MarkRoot((StgClosure *)ccalling_threadss[i]); + { + nat i; + for (i=0; i<=RtsFlags.GranFlags.proc; i++) { + if ((run_queue_hds[i] != END_TSO_QUEUE) && ((run_queue_hds[i] != NULL))) + run_queue_hds[i] = (StgTSO *)MarkRoot((StgClosure *)run_queue_hds[i]); + if ((run_queue_tls[i] != END_TSO_QUEUE) && ((run_queue_tls[i] != NULL))) + run_queue_tls[i] = (StgTSO *)MarkRoot((StgClosure *)run_queue_tls[i]); + + if ((blocked_queue_hds[i] != END_TSO_QUEUE) && ((blocked_queue_hds[i] != NULL))) + blocked_queue_hds[i] = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_hds[i]); + if ((blocked_queue_tls[i] != END_TSO_QUEUE) && ((blocked_queue_tls[i] != NULL))) + blocked_queue_tls[i] = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_tls[i]); + if ((ccalling_threadss[i] != END_TSO_QUEUE) && ((ccalling_threadss[i] != NULL))) + ccalling_threadss[i] = (StgTSO *)MarkRoot((StgClosure *)ccalling_threadss[i]); + } } markEventQueue(); -#elif defined(PAR) - run_queue_hd = (StgTSO *)MarkRoot((StgClosure *)run_queue_hd); - run_queue_tl = (StgTSO *)MarkRoot((StgClosure *)run_queue_tl); - blocked_queue_hd = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_hd); - blocked_queue_tl = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_tl); -#else - run_queue_hd = (StgTSO *)MarkRoot((StgClosure *)run_queue_hd); - run_queue_tl = (StgTSO *)MarkRoot((StgClosure *)run_queue_tl); - blocked_queue_hd = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_hd); - blocked_queue_tl = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_tl); +#else /* !GRAN */ + if (run_queue_hd != END_TSO_QUEUE) { + ASSERT(run_queue_tl != END_TSO_QUEUE); + run_queue_hd = (StgTSO *)MarkRoot((StgClosure *)run_queue_hd); + run_queue_tl = (StgTSO *)MarkRoot((StgClosure *)run_queue_tl); + } + + if (blocked_queue_hd != END_TSO_QUEUE) { + ASSERT(blocked_queue_tl != END_TSO_QUEUE); + blocked_queue_hd = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_hd); + blocked_queue_tl = (StgTSO *)MarkRoot((StgClosure *)blocked_queue_tl); + } + + if (sleeping_queue != END_TSO_QUEUE) { + sleeping_queue = (StgTSO *)MarkRoot((StgClosure *)sleeping_queue); + } #endif for (m = main_threads; m != NULL; m = m->link) { m->tso = (StgTSO *)MarkRoot((StgClosure *)m->tso); } - suspended_ccalling_threads = - (StgTSO *)MarkRoot((StgClosure *)suspended_ccalling_threads); + if (suspended_ccalling_threads != END_TSO_QUEUE) + suspended_ccalling_threads = + (StgTSO *)MarkRoot((StgClosure *)suspended_ccalling_threads); #if defined(SMP) || defined(PAR) || defined(GRAN) markSparkQueue(); @@ -1603,7 +2164,13 @@ void (*extra_roots)(void); void performGC(void) { - GarbageCollect(GetRoots); + GarbageCollect(GetRoots,rtsFalse); +} + +void +performMajorGC(void) +{ + GarbageCollect(GetRoots,rtsTrue); } static void @@ -1618,15 +2185,16 @@ performGCWithRoots(void (*get_roots)(void)) { extra_roots = get_roots; - GarbageCollect(AllRoots); + GarbageCollect(AllRoots,rtsFalse); } /* ----------------------------------------------------------------------------- Stack overflow - If the thread has reached its maximum stack size, - then bomb out. Otherwise relocate the TSO into a larger chunk of - memory and adjust its stack size appropriately. + If the thread has reached its maximum stack size, then raise the + StackOverflow exception in the offending thread. Otherwise + relocate the TSO into a larger chunk of memory and adjust its stack + size appropriately. -------------------------------------------------------------------------- */ static StgTSO * @@ -1636,19 +2204,18 @@ threadStackOverflow(StgTSO *tso) StgPtr new_sp; StgTSO *dest; + IF_DEBUG(sanity,checkTSO(tso)); if (tso->stack_size >= tso->max_stack_size) { -#if 0 - /* 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)); -#endif -#ifdef INTERPRETER - fprintf(stderr, "fatal: stack overflow in Hugs; aborting\n" ); - exit(1); -#else + + 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); + /* 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))); + /* Send this thread the StackOverflow exception */ - raiseAsync(tso, (StgClosure *)&stackOverflow_closure); -#endif + raiseAsync(tso, (StgClosure *)stackOverflow_closure); return tso; } @@ -1662,7 +2229,7 @@ threadStackOverflow(StgTSO *tso) new_tso_size = round_to_mblocks(new_tso_size); /* Be MBLOCK-friendly */ new_stack_size = new_tso_size - TSO_STRUCT_SIZEW; - IF_DEBUG(scheduler, fprintf(stderr,"scheduler: increasing stack size from %d words to %d.\n", tso->stack_size, new_stack_size)); + IF_DEBUG(scheduler, fprintf(stderr,"== scheduler: 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_tso_size-sizeofW(StgTSO),0); @@ -1677,36 +2244,37 @@ threadStackOverflow(StgTSO *tso) diff = (P_)new_sp - (P_)tso->sp; /* In *words* */ dest->su = (StgUpdateFrame *) ((P_)dest->su + diff); dest->sp = new_sp; - dest->splim = (P_)dest->splim + (nat)((P_)dest - (P_)tso); dest->stack_size = new_stack_size; /* and relocate the update frame list */ relocate_TSO(tso, dest); - /* Mark the old one as dead so we don't try to scavenge it during - * garbage collection (the TSO will likely be on a mutables list in - * some generation, but it'll get collected soon enough). It's - * important to set the sp and su values to just beyond the end of - * the stack, so we don't attempt to scavenge any part of the dead - * TSO's stack. + /* Mark the old TSO as relocated. We have to check for relocated + * TSOs in the garbage collector and any primops that deal with TSOs. + * + * It's important to set the sp and su values to just beyond the end + * of the stack, so we don't attempt to scavenge any part of the + * dead TSO's stack. */ - tso->whatNext = ThreadKilled; + tso->what_next = ThreadRelocated; + tso->link = dest; tso->sp = (P_)&(tso->stack[tso->stack_size]); tso->su = (StgUpdateFrame *)tso->sp; 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", + tso->id, tso, tso->stack_size); + /* If we're debugging, just print out the top of the stack */ + printStackChunk(tso->sp, stg_min(tso->stack+tso->stack_size, + tso->sp+64))); + IF_DEBUG(sanity,checkTSO(tso)); #if 0 IF_DEBUG(scheduler,printTSO(dest)); #endif -#if 0 - /* This will no longer work: KH */ - if (tso == MainTSO) { /* hack */ - MainTSO = dest; - } -#endif return dest; } @@ -1717,10 +2285,11 @@ threadStackOverflow(StgTSO *tso) Wake up a queue that was blocked on some resource. ------------------------------------------------------------------------ */ -// ToDo: check push_on_run_queue vs. PUSH_ON_RUN_QUEUE - #if defined(GRAN) -# error FixME +static inline void +unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node ) +{ +} #elif defined(PAR) static inline void unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node ) @@ -1731,6 +2300,8 @@ unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node ) DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC, GR_RESUMEQ, ((StgTSO *)bqe), ((StgTSO *)bqe)->block_info.closure, 0, 0 /* spark_queue_len(ADVISORY_POOL) */); + if (EMPTY_RUN_QUEUE()) + emitSchedule = rtsTrue; switch (get_itbl(node)->type) { case FETCH_ME_BQ: @@ -1741,6 +2312,10 @@ unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node ) case BLACKHOLE_BQ: ((StgTSO *)bqe)->par.blocktime += CURRENT_TIME-((StgTSO *)bqe)->par.blockedat; break; +#ifdef DIST + case MVAR: + break; +#endif default: barf("{unblockOneLocked}Daq Qagh: unexpected closure in blocking queue"); } @@ -1749,7 +2324,45 @@ unblockCount ( StgBlockingQueueElement *bqe, StgClosure *node ) #endif #if defined(GRAN) -# error FixME +static StgBlockingQueueElement * +unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node) +{ + StgTSO *tso; + PEs node_loc, tso_loc; + + node_loc = where_is(node); // should be lifted out of loop + tso = (StgTSO *)bqe; // wastes an assignment to get the type right + tso_loc = where_is((StgClosure *)tso); + if (IS_LOCAL_TO(PROCS(node),tso_loc)) { // TSO is local + /* !fake_fetch => TSO is on CurrentProc is same as IS_LOCAL_TO */ + ASSERT(CurrentProc!=node_loc || tso_loc==CurrentProc); + CurrentTime[CurrentProc] += RtsFlags.GranFlags.Costs.lunblocktime; + // insertThread(tso, node_loc); + new_event(tso_loc, tso_loc, CurrentTime[CurrentProc], + ResumeThread, + tso, node, (rtsSpark*)NULL); + tso->link = END_TSO_QUEUE; // overwrite link just to be sure + // len_local++; + // len++; + } else { // TSO is remote (actually should be FMBQ) + CurrentTime[CurrentProc] += RtsFlags.GranFlags.Costs.mpacktime + + RtsFlags.GranFlags.Costs.gunblocktime + + RtsFlags.GranFlags.Costs.latency; + new_event(tso_loc, CurrentProc, CurrentTime[CurrentProc], + UnblockThread, + tso, node, (rtsSpark*)NULL); + tso->link = END_TSO_QUEUE; // overwrite link just to be sure + // len++; + } + /* 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) ,", + (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)", + tso->id, tso)); +} #elif defined(PAR) static StgBlockingQueueElement * unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node) @@ -1772,8 +2385,8 @@ unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node) case BLOCKED_FETCH: /* if it's a BLOCKED_FETCH put it on the PendingFetches list */ next = bqe->link; - bqe->link = PendingFetches; - PendingFetches = bqe; + bqe->link = (StgBlockingQueueElement *)PendingFetches; + PendingFetches = (StgBlockedFetch *)bqe; break; # if defined(DEBUG) @@ -1781,9 +2394,9 @@ unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node) see comments on RBHSave closures above */ case CONSTR: /* check that the closure is an RBHSave closure */ - ASSERT(get_itbl((StgClosure *)bqe) == &RBH_Save_0_info || - get_itbl((StgClosure *)bqe) == &RBH_Save_1_info || - get_itbl((StgClosure *)bqe) == &RBH_Save_2_info); + ASSERT(get_itbl((StgClosure *)bqe) == &stg_RBH_Save_0_info || + get_itbl((StgClosure *)bqe) == &stg_RBH_Save_1_info || + get_itbl((StgClosure *)bqe) == &stg_RBH_Save_2_info); break; default: @@ -1792,7 +2405,7 @@ unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node) (StgClosure *)bqe); # endif } - // IF_DEBUG(scheduler,sched_belch("waking up thread %ld", tso->id)); + IF_PAR_DEBUG(bq, fprintf(stderr, ", %p (%s)", bqe, info_type((StgClosure*)bqe))); return next; } @@ -1813,16 +2426,14 @@ unblockOneLocked(StgTSO *tso) } #endif -#if defined(GRAN) -# error FixME -#elif defined(PAR) -inline StgTSO * -unblockOne(StgTSO *tso, StgClosure *node) +#if defined(GRAN) || defined(PAR) +inline StgBlockingQueueElement * +unblockOne(StgBlockingQueueElement *bqe, StgClosure *node) { ACQUIRE_LOCK(&sched_mutex); - tso = unblockOneLocked(tso, node); + bqe = unblockOneLocked(bqe, node); RELEASE_LOCK(&sched_mutex); - return tso; + return bqe; } #else inline StgTSO * @@ -1836,70 +2447,22 @@ unblockOne(StgTSO *tso) #endif #if defined(GRAN) -# error FixME -#elif defined(PAR) void awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node) { - StgBlockingQueueElement *bqe, *next; - - ACQUIRE_LOCK(&sched_mutex); - - IF_PAR_DEBUG(verbose, - belch("## AwBQ for node %p on [%x]: ", - node, mytid)); - - ASSERT(get_itbl(q)->type == TSO || - get_itbl(q)->type == BLOCKED_FETCH || - get_itbl(q)->type == CONSTR); - - bqe = q; - while (get_itbl(bqe)->type==TSO || - get_itbl(bqe)->type==BLOCKED_FETCH) { - bqe = unblockOneLocked(bqe, node); - } - RELEASE_LOCK(&sched_mutex); -} - -#else /* !GRAN && !PAR */ -void -awakenBlockedQueue(StgTSO *tso) -{ - ACQUIRE_LOCK(&sched_mutex); - while (tso != END_TSO_QUEUE) { - tso = unblockOneLocked(tso); - } - RELEASE_LOCK(&sched_mutex); -} -#endif - -#if 0 -// ngoq ngo' - -#if defined(GRAN) -/* - Awakening a blocking queue in GranSim means checking for each of the - TSOs in the queue whether they are local or not, issuing a ResumeThread - or an UnblockThread event, respectively. The basic iteration over the - blocking queue is the same as in the standard setup. -*/ -void -awaken_blocked_queue(StgBlockingQueueElement *q, StgClosure *node) -{ - StgBlockingQueueElement *bqe, *next; - StgTSO *tso; - PEs node_loc, tso_loc; - rtsTime bq_processing_time = 0; - nat len = 0, len_local = 0; + StgBlockingQueueElement *bqe; + PEs node_loc; + nat len = 0; IF_GRAN_DEBUG(bq, - belch("## AwBQ for node %p on PE %d @ %ld by TSO %d (%p): ", \ + belch("##-_ AwBQ for node %p on PE %d @ %ld by TSO %d (%p): ", \ node, CurrentProc, CurrentTime[CurrentProc], CurrentTSO->id, CurrentTSO)); node_loc = where_is(node); - ASSERT(get_itbl(q)->type == TSO || // q is either a TSO or an RBHSave + ASSERT(q == END_BQ_QUEUE || + get_itbl(q)->type == TSO || // q is either a TSO or an RBHSave get_itbl(q)->type == CONSTR); // closure (type constructor) ASSERT(is_unique(node)); @@ -1922,63 +2485,32 @@ awaken_blocked_queue(StgBlockingQueueElement *q, StgClosure *node) } } - next = q; + bqe = q; // ToDo: check: ASSERT(CurrentProc==node_loc); - while (get_itbl(next)->type==TSO) { // q != END_TSO_QUEUE) { - bqe = next; - next = bqe->link; + while (get_itbl(bqe)->type==TSO) { // q != END_TSO_QUEUE) { + //next = bqe->link; /* bqe points to the current element in the queue next points to the next element in the queue */ - tso = (StgTSO *)bqe; // wastes an assignment to get the type right - tso_loc = where_is(tso); - if (IS_LOCAL_TO(PROCS(node),tso_loc)) { // TSO is local - /* !fake_fetch => TSO is on CurrentProc is same as IS_LOCAL_TO */ - ASSERT(CurrentProc!=node_loc || tso_loc==CurrentProc); - bq_processing_time += RtsFlags.GranFlags.Costs.lunblocktime; - // insertThread(tso, node_loc); - new_event(tso_loc, tso_loc, - CurrentTime[CurrentProc]+bq_processing_time, - ResumeThread, - tso, node, (rtsSpark*)NULL); - tso->link = END_TSO_QUEUE; // overwrite link just to be sure - len_local++; - len++; - } else { // TSO is remote (actually should be FMBQ) - bq_processing_time += RtsFlags.GranFlags.Costs.mpacktime; - bq_processing_time += RtsFlags.GranFlags.Costs.gunblocktime; - new_event(tso_loc, CurrentProc, - CurrentTime[CurrentProc]+bq_processing_time+ - RtsFlags.GranFlags.Costs.latency, - UnblockThread, - tso, node, (rtsSpark*)NULL); - tso->link = END_TSO_QUEUE; // overwrite link just to be sure - bq_processing_time += RtsFlags.GranFlags.Costs.mtidytime; - len++; - } - /* the thread-queue-overhead is accounted for in either Resume or UnblockThread */ - IF_GRAN_DEBUG(bq, - fprintf(stderr," %s TSO %d (%p) [PE %d] (blocked_on=%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)", - tso->id, tso)); + //tso = (StgTSO *)bqe; // wastes an assignment to get the type right + //tso_loc = where_is(tso); + len++; + bqe = unblockOneLocked(bqe, node); } /* if this is the BQ of an RBH, we have to put back the info ripped out of the closure to make room for the anchor of the BQ */ - if (next!=END_BQ_QUEUE) { - ASSERT(get_itbl(node)->type == RBH && get_itbl(next)->type == CONSTR); + if (bqe!=END_BQ_QUEUE) { + ASSERT(get_itbl(node)->type == RBH && get_itbl(bqe)->type == CONSTR); /* ASSERT((info_ptr==&RBH_Save_0_info) || (info_ptr==&RBH_Save_1_info) || (info_ptr==&RBH_Save_2_info)); */ /* cf. convertToRBH in RBH.c for writing the RBHSave closure */ - ((StgRBH *)node)->blocking_queue = ((StgRBHSave *)next)->payload[0]; - ((StgRBH *)node)->mut_link = ((StgRBHSave *)next)->payload[1]; + ((StgRBH *)node)->blocking_queue = (StgBlockingQueueElement *)((StgRBHSave *)bqe)->payload[0]; + ((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", @@ -1987,151 +2519,58 @@ awaken_blocked_queue(StgBlockingQueueElement *q, StgClosure *node) /* statistics gathering */ if (RtsFlags.GranFlags.GranSimStats.Global) { - globalGranStats.tot_bq_processing_time += bq_processing_time; + // globalGranStats.tot_bq_processing_time += bq_processing_time; globalGranStats.tot_bq_len += len; // total length of all bqs awakened - globalGranStats.tot_bq_len_local += len_local; // same for local TSOs only + // globalGranStats.tot_bq_len_local += len_local; // same for local TSOs only globalGranStats.tot_awbq++; // total no. of bqs awakened } IF_GRAN_DEBUG(bq, - fprintf(stderr,"## BQ Stats of %p: [%d entries, %d local] %s\n", - node, len, len_local, (next!=END_TSO_QUEUE) ? "RBH" : "")); + fprintf(stderr,"## BQ Stats of %p: [%d entries] %s\n", + node, len, (bqe!=END_BQ_QUEUE) ? "RBH" : "")); } - #elif defined(PAR) - -/* - Awakening a blocking queue in GUM has to check whether an entry in the - queue is a normal TSO or a BLOCKED_FETCH. The later indicates that a TSO is - waiting for the result of this computation on another PE. Thus, when - finding a BLOCKED_FETCH we have to send off a message to that PE. - Actually, we defer sending off a message, by just putting the BLOCKED_FETCH - onto the PendingFetches queue, which will be later traversed by - processFetches, sending off a RESUME message for each BLOCKED_FETCH. - - NB: There is no check for an RBHSave closure (type CONSTR) in the code - below. The reason is, if we awaken the BQ of an RBH closure (RBHSaves - only exist at the end of such BQs) we know that the closure has been - unpacked successfully on the other PE, and we can discard the info - contained in the RBHSave closure. The current closure will be turned - into a FetchMe closure anyway. -*/ void -awaken_blocked_queue(StgBlockingQueueElement *q, StgClosure *node) +awakenBlockedQueue(StgBlockingQueueElement *q, StgClosure *node) { - StgBlockingQueueElement *bqe, *next; + StgBlockingQueueElement *bqe; + + ACQUIRE_LOCK(&sched_mutex); IF_PAR_DEBUG(verbose, - belch("## AwBQ for node %p on [%x]: ", + belch("##-_ AwBQ for node %p on [%x]: ", node, mytid)); - - ASSERT(get_itbl(q)->type == TSO || +#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?)")); + return; + } +#endif + + ASSERT(q == END_BQ_QUEUE || + get_itbl(q)->type == TSO || get_itbl(q)->type == BLOCKED_FETCH || get_itbl(q)->type == CONSTR); - next = q; - while (get_itbl(next)->type==TSO || - get_itbl(next)->type==BLOCKED_FETCH) { - bqe = next; - switch (get_itbl(bqe)->type) { - case TSO: - /* if it's a TSO just push it onto the run_queue */ - next = bqe->link; -#if defined(DEBUG) - ((StgTSO *)bqe)->link = END_TSO_QUEUE; // debugging only -#endif - push_on_run_queue((StgTSO *)bqe); // HWL: was: PUSH_ON_RUN_QUEUE(tso); - - /* write RESUME events to log file and - update blocked and fetch time (depending on type of the orig closure) */ - if (RtsFlags.ParFlags.ParStats.Full) { - DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC, - GR_RESUMEQ, ((StgTSO *)bqe), ((StgTSO *)bqe)->block_info.closure, - 0, spark_queue_len(ADVISORY_POOL)); - - switch (get_itbl(node)->type) { - case FETCH_ME_BQ: - ((StgTSO *)bqe)->par.fetchtime += CURRENT_TIME-((StgTSO *)bqe)->par.blockedat; - break; - case RBH: - case FETCH_ME: - case BLACKHOLE_BQ: - ((StgTSO *)bqe)->par.blocktime += CURRENT_TIME-((StgTSO *)bqe)->par.blockedat; - break; - default: - barf("{awaken_blocked_queue}Daq Qagh: unexpected closure %p (%s) with blocking queue", - node, info_type(node)); - } - } - /* reset block_info.closure field after dumping event */ - ((StgTSO *)bqe)->block_info.closure = NULL; - - /* rest of this branch is debugging only */ - IF_PAR_DEBUG(verbose, - fprintf(stderr," TSO %d (%p) [PE %lx] (block_info.closure=%p) (next=%p) ,", - ((StgTSO *)bqe)->id, (StgTSO *)bqe, - mytid, ((StgTSO *)bqe)->block_info.closure, ((StgTSO *)bqe)->link)); - - IF_DEBUG(scheduler, - if (!RtsFlags.ParFlags.Debug.verbose) - belch("-- Waking up thread %ld (%p)", - ((StgTSO *)bqe)->id, (StgTSO *)bqe)); - break; - - case BLOCKED_FETCH: - /* if it's a BLOCKED_FETCH put it on the PendingFetches list */ - next = bqe->link; - bqe->link = PendingFetches; - PendingFetches = bqe; - // bqe.tso->block_info.closure = NULL; - - /* rest of this branch is debugging only */ - IF_PAR_DEBUG(verbose, - fprintf(stderr," BLOCKED_FETCH (%p) on node %p [PE %lx] (next=%p) ,", - ((StgBlockedFetch *)bqe), - ((StgBlockedFetch *)bqe)->node, - mytid, ((StgBlockedFetch *)bqe)->link)); - break; - -# if defined(DEBUG) - /* can ignore this case in a non-debugging setup; - see comments on RBHSave closures above */ - case CONSTR: - /* check that the closure is an RBHSave closure */ - ASSERT(get_itbl((StgClosure *)bqe) == &RBH_Save_0_info || - get_itbl((StgClosure *)bqe) == &RBH_Save_1_info || - get_itbl((StgClosure *)bqe) == &RBH_Save_2_info); - break; - - default: - barf("{awaken_blocked_queue}Daq Qagh: Unexpected IP (%#lx; %s) in blocking queue at %#lx\n", - get_itbl((StgClosure *)bqe), info_type((StgClosure *)bqe), - (StgClosure *)bqe); -# endif - } + bqe = q; + while (get_itbl(bqe)->type==TSO || + get_itbl(bqe)->type==BLOCKED_FETCH) { + bqe = unblockOneLocked(bqe, node); } + RELEASE_LOCK(&sched_mutex); } -#else /* !GRAN && !PAR */ - -void -awaken_blocked_queue(StgTSO *q) { awakenBlockedQueue(q); } - -/* +#else /* !GRAN && !PAR */ +void +awakenBlockedQueue(StgTSO *tso) { - StgTSO *tso; - - while (q != END_TSO_QUEUE) { - ASSERT(get_itbl(q)->type == TSO); - tso = q; - q = tso->link; - push_on_run_queue(tso); // HWL: was: PUSH_ON_RUN_QUEUE(tso); - //tso->block_info.closure = NULL; - IF_DEBUG(scheduler, belch("-- Waking up thread %ld (%p)", tso->id, tso)); + ACQUIRE_LOCK(&sched_mutex); + while (tso != END_TSO_QUEUE) { + tso = unblockOneLocked(tso); } + RELEASE_LOCK(&sched_mutex); } -*/ -#endif /* GRAN */ -#endif /* 0 */ +#endif //@node Exception Handling Routines, Debugging Routines, Blocking Queue Routines, Main scheduling code //@subsection Exception Handling Routines @@ -2156,6 +2595,142 @@ interruptStgRts(void) This has nothing to do with the UnblockThread event in GranSim. -- HWL -------------------------------------------------------------------------- */ +#if defined(GRAN) || defined(PAR) +/* + NB: only the type of the blocking queue is different in GranSim and GUM + the operations on the queue-elements are the same + long live polymorphism! +*/ +static void +unblockThread(StgTSO *tso) +{ + StgBlockingQueueElement *t, **last; + + ACQUIRE_LOCK(&sched_mutex); + switch (tso->why_blocked) { + + case NotBlocked: + return; /* not blocked */ + + case BlockedOnMVar: + ASSERT(get_itbl(tso->block_info.closure)->type == MVAR); + { + StgBlockingQueueElement *last_tso = END_BQ_QUEUE; + StgMVar *mvar = (StgMVar *)(tso->block_info.closure); + + last = (StgBlockingQueueElement **)&mvar->head; + for (t = (StgBlockingQueueElement *)mvar->head; + t != END_BQ_QUEUE; + last = &t->link, last_tso = t, t = t->link) { + if (t == (StgBlockingQueueElement *)tso) { + *last = (StgBlockingQueueElement *)tso->link; + if (mvar->tail == tso) { + mvar->tail = (StgTSO *)last_tso; + } + goto done; + } + } + barf("unblockThread (MVAR): TSO not found"); + } + + 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_BQ_QUEUE; + last = &t->link, t = t->link) { + if (t == (StgBlockingQueueElement *)tso) { + *last = (StgBlockingQueueElement *)tso->link; + goto done; + } + } + barf("unblockThread (BLACKHOLE): TSO not found"); + } + + case BlockedOnException: + { + StgTSO *target = tso->block_info.tso; + + ASSERT(get_itbl(target)->type == TSO); + + if (target->what_next == ThreadRelocated) { + target = target->link; + ASSERT(get_itbl(target)->type == TSO); + } + + ASSERT(target->blocked_exceptions != NULL); + + last = (StgBlockingQueueElement **)&target->blocked_exceptions; + for (t = (StgBlockingQueueElement *)target->blocked_exceptions; + t != END_BQ_QUEUE; + last = &t->link, t = t->link) { + ASSERT(get_itbl(t)->type == TSO); + if (t == (StgBlockingQueueElement *)tso) { + *last = (StgBlockingQueueElement *)tso->link; + goto done; + } + } + barf("unblockThread (Exception): TSO not found"); + } + + case BlockedOnRead: + case BlockedOnWrite: + { + /* take TSO off blocked_queue */ + StgBlockingQueueElement *prev = NULL; + for (t = (StgBlockingQueueElement *)blocked_queue_hd; t != END_BQ_QUEUE; + prev = t, t = t->link) { + if (t == (StgBlockingQueueElement *)tso) { + if (prev == NULL) { + blocked_queue_hd = (StgTSO *)t->link; + if ((StgBlockingQueueElement *)blocked_queue_tl == t) { + blocked_queue_tl = END_TSO_QUEUE; + } + } else { + prev->link = t->link; + if ((StgBlockingQueueElement *)blocked_queue_tl == t) { + blocked_queue_tl = (StgTSO *)prev; + } + } + goto done; + } + } + barf("unblockThread (I/O): TSO not found"); + } + + case BlockedOnDelay: + { + /* take TSO off sleeping_queue */ + StgBlockingQueueElement *prev = NULL; + for (t = (StgBlockingQueueElement *)sleeping_queue; t != END_BQ_QUEUE; + prev = t, t = t->link) { + if (t == (StgBlockingQueueElement *)tso) { + if (prev == NULL) { + sleeping_queue = (StgTSO *)t->link; + } else { + prev->link = t->link; + } + goto done; + } + } + barf("unblockThread (I/O): TSO not found"); + } + + default: + barf("unblockThread"); + } + + done: + tso->link = END_TSO_QUEUE; + tso->why_blocked = NotBlocked; + tso->block_info.closure = NULL; + PUSH_ON_RUN_QUEUE(tso); + RELEASE_LOCK(&sched_mutex); +} +#else static void unblockThread(StgTSO *tso) { @@ -2208,6 +2783,12 @@ unblockThread(StgTSO *tso) StgTSO *target = tso->block_info.tso; ASSERT(get_itbl(target)->type == TSO); + + while (target->what_next == ThreadRelocated) { + target = target->link; + ASSERT(get_itbl(target)->type == TSO); + } + ASSERT(target->blocked_exceptions != NULL); last = &target->blocked_exceptions; @@ -2222,7 +2803,6 @@ unblockThread(StgTSO *tso) barf("unblockThread (Exception): TSO not found"); } - case BlockedOnDelay: case BlockedOnRead: case BlockedOnWrite: { @@ -2247,6 +2827,23 @@ unblockThread(StgTSO *tso) barf("unblockThread (I/O): TSO not found"); } + case BlockedOnDelay: + { + StgTSO *prev = NULL; + for (t = sleeping_queue; t != END_TSO_QUEUE; + prev = t, t = t->link) { + if (t == tso) { + if (prev == NULL) { + sleeping_queue = t->link; + } else { + prev->link = t->link; + } + goto done; + } + } + barf("unblockThread (I/O): TSO not found"); + } + default: barf("unblockThread"); } @@ -2258,6 +2855,7 @@ unblockThread(StgTSO *tso) PUSH_ON_RUN_QUEUE(tso); RELEASE_LOCK(&sched_mutex); } +#endif /* ----------------------------------------------------------------------------- * raiseAsync() @@ -2304,7 +2902,7 @@ raiseAsync(StgTSO *tso, StgClosure *exception) StgPtr sp = tso->sp; /* Thread already dead? */ - if (tso->whatNext == ThreadComplete || tso->whatNext == ThreadKilled) { + if (tso->what_next == ThreadComplete || tso->what_next == ThreadKilled) { return; } @@ -2319,7 +2917,7 @@ raiseAsync(StgTSO *tso, StgClosure *exception) * returns to the next return address on the stack. */ if ( LOOKS_LIKE_GHC_INFO((void*)*sp) ) { - *(--sp) = (W_)&dummy_ret_closure; + *(--sp) = (W_)&stg_dummy_ret_closure; } while (1) { @@ -2328,25 +2926,27 @@ raiseAsync(StgTSO *tso, StgClosure *exception) StgAP_UPD * ap; /* If we find a CATCH_FRAME, and we've got an exception to raise, - * then build PAP(handler,exception), and leave it on top of - * the stack ready to enter. + * then build PAP(handler,exception,realworld#), and leave it on + * top of the stack ready to enter. */ if (get_itbl(su)->type == CATCH_FRAME && exception != NULL) { StgCatchFrame *cf = (StgCatchFrame *)su; /* we've got an exception to raise, so let's pass it to the * handler in this frame. */ - ap = (StgAP_UPD *)allocate(sizeofW(StgPAP) + 1); - TICK_ALLOC_UPD_PAP(2,0); - SET_HDR(ap,&PAP_info,cf->header.prof.ccs); + ap = (StgAP_UPD *)allocate(sizeofW(StgPAP) + 2); + TICK_ALLOC_UPD_PAP(3,0); + SET_HDR(ap,&stg_PAP_info,cf->header.prof.ccs); - ap->n_args = 1; - ap->fun = cf->handler; - ap->payload[0] = (P_)exception; + ap->n_args = 2; + ap->fun = cf->handler; /* :: Exception -> IO a */ + ap->payload[0] = exception; + ap->payload[1] = ARG_TAG(0); /* realworld token */ - /* sp currently points to the word above the CATCH_FRAME on the stack. + /* throw away the stack from Sp up to and including the + * CATCH_FRAME. */ - sp += sizeofW(StgCatchFrame); + sp = (P_)su + sizeofW(StgCatchFrame) - 1; tso->su = cf->link; /* Restore the blocked/unblocked state for asynchronous exceptions @@ -2357,7 +2957,7 @@ raiseAsync(StgTSO *tso, StgClosure *exception) * unblockAsyncExceptions_ret stack frame. */ if (!cf->exceptions_blocked) { - *(sp--) = (W_)&unblockAsyncExceptionszh_ret_info; + *(sp--) = (W_)&stg_unblockAsyncExceptionszh_ret_info; } /* Ensure that async exceptions are blocked when running the handler. @@ -2371,7 +2971,8 @@ raiseAsync(StgTSO *tso, StgClosure *exception) */ sp[0] = (W_)ap; tso->sp = sp; - tso->whatNext = ThreadEnterGHC; + tso->what_next = ThreadEnterGHC; + IF_DEBUG(sanity, checkTSO(tso)); return; } @@ -2387,14 +2988,14 @@ raiseAsync(StgTSO *tso, StgClosure *exception) ap->fun = (StgClosure *)sp[0]; sp++; for(i=0; i < (nat)words; ++i) { - ap->payload[i] = (P_)*sp++; + ap->payload[i] = (StgClosure *)*sp++; } switch (get_itbl(su)->type) { case UPDATE_FRAME: { - SET_HDR(ap,&AP_UPD_info,su->header.prof.ccs /* ToDo */); + SET_HDR(ap,&stg_AP_UPD_info,su->header.prof.ccs /* ToDo */); TICK_ALLOC_UP_THK(words+1,0); IF_DEBUG(scheduler, @@ -2407,14 +3008,24 @@ raiseAsync(StgTSO *tso, StgClosure *exception) /* Replace the updatee with an indirection - happily * this will also wake up any threads currently * waiting on the result. + * + * Warning: if we're in a loop, more than one update frame on + * the stack may point to the same object. Be careful not to + * overwrite an IND_OLDGEN in this case, because we'll screw + * up the mutable lists. To be on the safe side, don't + * overwrite any kind of indirection at all. See also + * threadSqueezeStack in GC.c, where we have to make a similar + * check. */ - UPD_IND_NOLOCK(su->updatee,ap); /* revert the black hole */ + if (!closure_IND(su->updatee)) { + UPD_IND_NOLOCK(su->updatee,ap); /* revert the black hole */ + } su = su->link; sp += sizeofW(StgUpdateFrame) -1; sp[0] = (W_)ap; /* push onto stack */ break; } - + case CATCH_FRAME: { StgCatchFrame *cf = (StgCatchFrame *)su; @@ -2423,13 +3034,13 @@ raiseAsync(StgTSO *tso, StgClosure *exception) /* We want a PAP, not an AP_UPD. Fortunately, the * layout's the same. */ - SET_HDR(ap,&PAP_info,su->header.prof.ccs /* ToDo */); + SET_HDR(ap,&stg_PAP_info,su->header.prof.ccs /* ToDo */); TICK_ALLOC_UPD_PAP(words+1,0); /* now build o = FUN(catch,ap,handler) */ o = (StgClosure *)allocate(sizeofW(StgClosure)+2); TICK_ALLOC_FUN(2,0); - SET_HDR(o,&catch_info,su->header.prof.ccs /* ToDo */); + SET_HDR(o,&stg_catch_info,su->header.prof.ccs /* ToDo */); o->payload[0] = (StgClosure *)ap; o->payload[1] = cf->handler; @@ -2450,14 +3061,14 @@ raiseAsync(StgTSO *tso, StgClosure *exception) StgSeqFrame *sf = (StgSeqFrame *)su; StgClosure* o; - SET_HDR(ap,&PAP_info,su->header.prof.ccs /* ToDo */); + SET_HDR(ap,&stg_PAP_info,su->header.prof.ccs /* ToDo */); TICK_ALLOC_UPD_PAP(words+1,0); /* now build o = FUN(seq,ap) */ o = (StgClosure *)allocate(sizeofW(StgClosure)+1); TICK_ALLOC_SE_THK(1,0); - SET_HDR(o,&seq_info,su->header.prof.ccs /* ToDo */); - payloadCPtr(o,0) = (StgClosure *)ap; + SET_HDR(o,&stg_seq_info,su->header.prof.ccs /* ToDo */); + o->payload[0] = (StgClosure *)ap; IF_DEBUG(scheduler, fprintf(stderr, "scheduler: Built "); @@ -2475,11 +3086,11 @@ raiseAsync(StgTSO *tso, StgClosure *exception) /* We've stripped the entire stack, the thread is now dead. */ sp += sizeofW(StgStopFrame) - 1; sp[0] = (W_)exception; /* save the exception */ - tso->whatNext = ThreadKilled; + tso->what_next = ThreadKilled; tso->su = (StgUpdateFrame *)(sp+1); tso->sp = sp; return; - + default: barf("raiseAsync"); } @@ -2487,6 +3098,105 @@ raiseAsync(StgTSO *tso, StgClosure *exception) barf("raiseAsync"); } +/* ----------------------------------------------------------------------------- + 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 + on an MVar, or NonTermination if the thread was blocked on a Black + Hole. + -------------------------------------------------------------------------- */ + +void +resurrectThreads( StgTSO *threads ) +{ + StgTSO *tso, *next; + + for (tso = threads; tso != END_TSO_QUEUE; tso = next) { + next = tso->global_link; + tso->global_link = all_threads; + all_threads = tso; + IF_DEBUG(scheduler, sched_belch("resurrecting thread %d", tso->id)); + + switch (tso->why_blocked) { + case BlockedOnMVar: + case BlockedOnException: + raiseAsync(tso,(StgClosure *)BlockedOnDeadMVar_closure); + break; + case BlockedOnBlackHole: + raiseAsync(tso,(StgClosure *)NonTermination_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 + * can wake up threads, remember...). + */ + continue; + default: + barf("resurrectThreads: thread blocked in a strange way"); + } + } +} + +/* ----------------------------------------------------------------------------- + * 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. + * -------------------------------------------------------------------------- */ + +static void +detectBlackHoles( void ) +{ + StgTSO *t = all_threads; + StgUpdateFrame *frame; + StgClosure *blocked_on; + + for (t = all_threads; t != END_TSO_QUEUE; t = t->global_link) { + + while (t->what_next == ThreadRelocated) { + t = t->link; + ASSERT(get_itbl(t)->type == TSO); + } + + if (t->why_blocked != BlockedOnBlackHole) { + continue; + } + + blocked_on = t->block_info.closure; + + for (frame = t->su; ; frame = frame->link) { + switch (get_itbl(frame)->type) { + + case UPDATE_FRAME: + if (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", t->id)); + raiseAsync(t, (StgClosure *)NonTermination_closure); + goto done; + } + else { + continue; + } + + case CATCH_FRAME: + case SEQ_FRAME: + continue; + + case STOP_FRAME: + break; + } + break; + } + + done: ; + } +} + //@node Debugging Routines, Index, Exception Handling Routines, Main scheduling code //@subsection Debugging Routines @@ -2496,39 +3206,91 @@ raiseAsync(StgTSO *tso, StgClosure *exception) #ifdef DEBUG -void printThreadBlockage(StgTSO *tso) +void +printThreadBlockage(StgTSO *tso) { switch (tso->why_blocked) { case BlockedOnRead: - fprintf(stderr,"blocked on read from fd %d", tso->block_info.fd); + fprintf(stderr,"is blocked on read from fd %d", tso->block_info.fd); break; case BlockedOnWrite: - fprintf(stderr,"blocked on write to fd %d", tso->block_info.fd); + fprintf(stderr,"is blocked on write to fd %d", tso->block_info.fd); break; case BlockedOnDelay: - fprintf(stderr,"blocked on delay of %d ms", tso->block_info.delay); + fprintf(stderr,"is blocked until %d", tso->block_info.target); break; case BlockedOnMVar: - fprintf(stderr,"blocked on an MVar"); + fprintf(stderr,"is blocked on an MVar"); break; case BlockedOnException: - fprintf(stderr,"blocked on delivering an exception to thread %d", + fprintf(stderr,"is blocked on delivering an exception to thread %d", tso->block_info.tso->id); break; case BlockedOnBlackHole: - fprintf(stderr,"blocked on a black hole"); + fprintf(stderr,"is blocked on a black hole"); break; case NotBlocked: - fprintf(stderr,"not blocked"); + fprintf(stderr,"is not blocked"); break; #if defined(PAR) case BlockedOnGA: - fprintf(stderr,"blocked on global address"); + fprintf(stderr,"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)", + tso->block_info.closure, info_type(tso->block_info.closure)); break; #endif + default: + barf("printThreadBlockage: strange tso->why_blocked: %d for TSO %d (%d)", + tso->why_blocked, tso->id, tso); + } +} + +void +printThreadStatus(StgTSO *tso) +{ + switch (tso->what_next) { + case ThreadKilled: + fprintf(stderr,"has been killed"); + break; + case ThreadComplete: + fprintf(stderr,"has completed"); + break; + default: + printThreadBlockage(tso); } } +void +printAllThreads(void) +{ + StgTSO *t; + +# 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!*/); + + sched_belch("all threads at [%s]:", time_string); +# elif defined(PAR) + char time_string[TIME_STR_LEN], node_str[NODE_STR_LEN]; + ullong_format_string(CURRENT_TIME, + time_string, rtsFalse/*no commas!*/); + + sched_belch("all threads at [%s]:", time_string); +# else + sched_belch("all threads:"); +# endif + + for (t = all_threads; t != END_TSO_QUEUE; t = t->global_link) { + fprintf(stderr, "\tthread %d ", t->id); + printThreadStatus(t); + fprintf(stderr,"\n"); + } +} + /* Print a whole blocking queue attached to node (debugging only). */ @@ -2547,27 +3309,41 @@ print_bq (StgClosure *node) /* should cover all closures that may have a blocking queue */ ASSERT(get_itbl(node)->type == BLACKHOLE_BQ || get_itbl(node)->type == FETCH_ME_BQ || - get_itbl(node)->type == RBH); + get_itbl(node)->type == RBH || + get_itbl(node)->type == MVAR); ASSERT(node!=(StgClosure*)NULL); // sanity check + + print_bqe(((StgBlockingQueue*)node)->blocking_queue); +} + +/* + Print a whole blocking queue starting with the element bqe. +*/ +void +print_bqe (StgBlockingQueueElement *bqe) +{ + rtsBool end; + /* NB: In a parallel setup a BQ of an RBH must end with an RBH_Save closure; */ - for (bqe = ((StgBlockingQueue*)node)->blocking_queue, end = (bqe==END_BQ_QUEUE); + for (end = (bqe==END_BQ_QUEUE); !end; // iterate until bqe points to a CONSTR - end = (get_itbl(bqe)->type == CONSTR) || (bqe->link==END_BQ_QUEUE), bqe = end ? END_BQ_QUEUE : bqe->link) { - ASSERT(bqe != END_BQ_QUEUE); // sanity check - ASSERT(bqe != (StgTSO*)NULL); // sanity check + end = (get_itbl(bqe)->type == CONSTR) || (bqe->link==END_BQ_QUEUE), + bqe = end ? END_BQ_QUEUE : bqe->link) { + ASSERT(bqe != END_BQ_QUEUE); // sanity check + ASSERT(bqe != (StgBlockingQueueElement *)NULL); // sanity check /* types of closures that may appear in a blocking queue */ ASSERT(get_itbl(bqe)->type == TSO || get_itbl(bqe)->type == BLOCKED_FETCH || get_itbl(bqe)->type == CONSTR); /* only BQs of an RBH end with an RBH_Save closure */ - ASSERT(get_itbl(bqe)->type != CONSTR || get_itbl(node)->type == RBH); + //ASSERT(get_itbl(bqe)->type != CONSTR || get_itbl(node)->type == RBH); switch (get_itbl(bqe)->type) { case TSO: - fprintf(stderr," TSO %d (%x),", + fprintf(stderr," TSO %u (%x),", ((StgTSO *)bqe)->id, ((StgTSO *)bqe)); break; case BLOCKED_FETCH: @@ -2579,14 +3355,14 @@ print_bq (StgClosure *node) break; case CONSTR: fprintf(stderr," %s (IP %p),", - (get_itbl(bqe) == &RBH_Save_0_info ? "RBH_Save_0" : - get_itbl(bqe) == &RBH_Save_1_info ? "RBH_Save_1" : - get_itbl(bqe) == &RBH_Save_2_info ? "RBH_Save_2" : + (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" : "RBH_Save_?"), get_itbl(bqe)); break; default: - barf("Unexpected closure type %s in blocking queue of %p (%s)", - info_type(bqe), node, info_type(node)); + barf("Unexpected closure type %s in blocking queue", // of %p (%s)", + info_type((StgClosure *)bqe)); // , node, info_type(node)); break; } } /* for */ @@ -2597,7 +3373,6 @@ void print_bq (StgClosure *node) { StgBlockingQueueElement *bqe; - StgTSO *tso; PEs node_loc, tso_loc; rtsBool end; @@ -2619,7 +3394,7 @@ print_bq (StgClosure *node) !end; // iterate until bqe points to a CONSTR end = (get_itbl(bqe)->type == CONSTR) || (bqe->link==END_BQ_QUEUE), bqe = end ? END_BQ_QUEUE : bqe->link) { ASSERT(bqe != END_BQ_QUEUE); // sanity check - ASSERT(bqe != (StgTSO*)NULL); // sanity check + ASSERT(bqe != (StgBlockingQueueElement *)NULL); // sanity check /* types of closures that may appear in a blocking queue */ ASSERT(get_itbl(bqe)->type == TSO || get_itbl(bqe)->type == CONSTR); @@ -2629,19 +3404,19 @@ print_bq (StgClosure *node) tso_loc = where_is((StgClosure *)bqe); switch (get_itbl(bqe)->type) { case TSO: - fprintf(stderr," TSO %d (%x) on [PE %d],", - ((StgTSO *)bqe)->id, ((StgTSO *)bqe), tso_loc); + fprintf(stderr," TSO %d (%p) on [PE %d],", + ((StgTSO *)bqe)->id, (StgTSO *)bqe, tso_loc); break; case CONSTR: fprintf(stderr," %s (IP %p),", - (get_itbl(bqe) == &RBH_Save_0_info ? "RBH_Save_0" : - get_itbl(bqe) == &RBH_Save_1_info ? "RBH_Save_1" : - get_itbl(bqe) == &RBH_Save_2_info ? "RBH_Save_2" : + (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" : "RBH_Save_?"), get_itbl(bqe)); break; default: barf("Unexpected closure type %s in blocking queue of %p (%s)", - info_type(bqe), node, info_type(node)); + info_type((StgClosure *)bqe), node, info_type(node)); break; } } /* for */ @@ -2660,26 +3435,27 @@ print_bq (StgClosure *node) for (tso = ((StgBlockingQueue*)node)->blocking_queue; tso != END_TSO_QUEUE; tso=tso->link) { - ASSERT(tso!=(StgTSO*)NULL && tso!=END_TSO_QUEUE); // sanity check + ASSERT(tso!=NULL && tso!=END_TSO_QUEUE); // sanity check ASSERT(get_itbl(tso)->type == TSO); // guess what, sanity check - fprintf(stderr," TSO %d (%x),", tso->id, tso); + fprintf(stderr," TSO %d (%p),", tso->id, tso); } fputc('\n', stderr); } # endif -/* A debugging function used all over the place in GranSim and GUM. - Dummy function in other setups. -*/ -# if !defined(GRAN) && !defined(PAR) -char * -info_type(StgClosure *closure){ - return "petaQ"; -} +#if defined(PAR) +static nat +run_queue_len(void) +{ + nat i; + StgTSO *tso; + + for (i=0, tso=run_queue_hd; + tso != END_TSO_QUEUE; + i++, tso=tso->link) + /* nothing */ -char * -info_type_by_ip(StgInfoTable *ip){ - return "petaQ"; + return i; } #endif @@ -2690,6 +3466,8 @@ sched_belch(char *s, ...) va_start(ap,s); #ifdef SMP fprintf(stderr, "scheduler (task %ld): ", pthread_self()); +#elif defined(PAR) + fprintf(stderr, "== "); #else fprintf(stderr, "scheduler: "); #endif @@ -2699,6 +3477,7 @@ sched_belch(char *s, ...) #endif /* DEBUG */ + //@node Index, , Debugging Routines, Main scheduling code //@subsection Index