/* ---------------------------------------------------------------------------
- * $Id: Schedule.c,v 1.42 2000/01/14 11:45:21 hwloidl Exp $
+ * $Id: Schedule.c,v 1.63 2000/04/04 15:02:02 simonmar Exp $
*
- * (c) The GHC Team, 1998-1999
+ * (c) The GHC Team, 1998-2000
*
* Scheduler
*
//@menu
//* Includes::
//* Variables and Data structures::
-//* Prototypes::
//* Main scheduling loop::
//* Suspend and Resume::
//* Run queue code::
#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"
# include "FetchMe.h"
# include "HLC.h"
#endif
+#include "Sparks.h"
#include <stdarg.h>
/* 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.
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;
+#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
#endif
#if defined(GRAN)
-StgTSO *CurrentTSOs[MAX_PROC];
-#else
-StgTSO *CurrentTSO;
+StgTSO *CurrentTSO;
#endif
rtsBool ready_to_gc;
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
#ifdef DEBUG
static void sched_belch(char *s, ...);
rtsTime TimeOfLastYield;
#endif
+#if DEBUG
+char *whatNext_strs[] = {
+ "ThreadEnterGHC",
+ "ThreadRunGHC",
+ "ThreadEnterHugs",
+ "ThreadKilled",
+ "ThreadComplete"
+};
+
+char *threadReturnCode_strs[] = {
+ "HeapOverflow", /* might also be StackOverflow */
+ "StackOverflow",
+ "ThreadYielding",
+ "ThreadBlocked",
+ "ThreadFinished"
+};
+#endif
+
/*
* The thread state for the main thread.
// ToDo: check whether not needed any more
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
* 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
#if defined(GRAN)
rtsEvent *event;
#elif defined(PAR)
+ StgSparkPool *pool;
rtsSpark spark;
StgTSO *tso;
GlobalTaskId pe;
#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();
- }
#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.
}
run_queue_hd = run_queue_tl = END_TSO_QUEUE;
blocked_queue_hd = blocked_queue_tl = END_TSO_QUEUE;
+ interrupted = rtsFalse;
+ was_interrupted = rtsTrue;
}
/* Go through the list of main threads and wake up any
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];
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:
}
}
}
+
#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;
}
}
if (spark == NULL) {
break; /* no more sparks in the pool */
} else {
- // I'd prefer this to be done in activateSpark -- HWL
+ /* 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 */
StgTSO *tso;
tso = createThread_(RtsFlags.GcFlags.initialStkSize, rtsTrue);
pushClosure(tso,spark);
main_threads = NULL;
}
#else /* ! SMP */
+ /*
+ In GUM all non-main PEs come in here with no work;
+ we ignore multiple main threads for now
+
if (blocked_queue_hd == END_TSO_QUEUE
&& run_queue_hd == END_TSO_QUEUE) {
StgMainThread *m = main_threads;
main_threads = m->link;
return;
}
+ */
#endif
#ifdef SMP
#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 */
/* :-[ 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:
spark = findSpark(); /* 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
/* 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);
/* 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 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 sparks on [%#x] (hd=%x; tl=%x; base=%x, lim=%x)",
+ spark_queue_len(pool),
+ CURRENT_PROC,
+ pool->hd, pool->tl, pool->base, pool->lim));
IF_DEBUG(scheduler, belch("--== %d threads on [%#x] (hd=%x; tl=%x)",
run_queue_len(), CURRENT_PROC,
run_queue_hd, run_queue_tl));
+#if 0
if (t != LastTSO) {
/*
we are running a different TSO, so write a schedule event to log file
GR_SCHEDULE, t, (StgClosure *)NULL, 0, 0);
}
-
+#endif
#else /* !GRAN && !PAR */
/* grab a thread from the run queue
*/
t = POP_RUN_QUEUE();
+ IF_DEBUG(sanity,checkTSO(t));
#endif
context_switch = 1;
RELEASE_LOCK(&sched_mutex);
-
+
+#if defined(GRAN) || defined(PAR)
+ IF_DEBUG(scheduler, belch("-->> Running TSO %ld (%p) %s ...",
+ t->id, t, whatNext_strs[t->what_next]));
+#else
IF_DEBUG(scheduler,sched_belch("running thread %d", t->id));
+#endif
/* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
/* Run the current thread
*/
- switch (cap->rCurrentTSO->whatNext) {
+ switch (cap->rCurrentTSO->what_next) {
case ThreadKilled:
case ThreadComplete:
/* Thread already finished, return to scheduler. */
barf("Panic: entered a BCO but no bytecode interpreter in this build");
#endif
default:
- barf("schedule: invalid whatNext field");
+ barf("schedule: invalid what_next field");
}
/* +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ */
#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; //(ret == ThreadBlocked) ? END_TSO_QUEUE : t;
+ TimeOfLastYield = CURRENT_TIME;
+#endif
+
switch (ret) {
case HeapOverflow:
/* 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));
+#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_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;
/* 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;
* 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 == ThreadEnterHugs) {
+ /* 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
APPEND_TO_RUN_QUEUE(t);
+#if defined(GRAN)
+ /* add a ContinueThread event to actually process the thread */
+ new_event(CurrentProc, CurrentProc, CurrentTime[CurrentProc],
+ ContinueThread,
+ t, (StgClosure*)NULL, (rtsSpark*)NULL);
+ IF_GRAN_DEBUG(bq,
+ 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
+ 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));
+ 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(par,
DumpGranEvent(GR_DESCHEDULE, t));
-#else
-#endif
+
+ /* Send a fetch (if BlockedOnGA) and dump event to log file */
+ blockThread(t);
+
+ 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 (t->block_info.closure!=(StgClosure*)NULL) print_bq(t->block_info.closure));
+
+#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;
* 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;
+ IF_DEBUG(scheduler,belch("--++ thread %d (%p) finished", t->id, t));
+ t->what_next = ThreadComplete;
#if defined(GRAN)
- // ToDo: endThread(t, CurrentProc); // clean-up the thread
+ endThread(t, CurrentProc); // clean-up the thread
#elif defined(PAR)
advisory_thread_count--;
if (RtsFlags.ParFlags.ParStats.Full)
#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:
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)
{
createThread_(nat size, rtsBool have_lock)
{
#endif
+
StgTSO *tso;
nat stack_size;
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;
+ tso = (StgTSO *)allocate(size);
+ TICK_ALLOC_TSO(size-TSO_STRUCT_SIZEW, 0);
+
SET_HDR(tso, &TSO_info, CCS_MAIN);
#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); }
insertThread(tso, CurrentProc);
#else
/* In a non-GranSim setup the pushing of a TSO onto the runq is separated
- from its creation
- */
+ * from its creation
+ */
+#endif
+
+#if defined(GRAN) || defined(PAR)
+ DumpGranEvent(GR_START,tso);
#endif
+ /* Link the new thread on the global thread list.
+ */
+ tso->global_link = all_threads;
+ all_threads = tso;
+
#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;
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;
globalGranStats.tot_sq_probes++;
#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;
+}
+
+/*
+ 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;
+
+ ASSERT(spark != (rtsSpark)NULL);
+ tso = createThread_(RtsFlags.GcFlags.initialStkSize, rtsTrue);
+ if (tso!=END_TSO_QUEUE) {
+ pushClosure(tso,spark);
+ PUSH_ON_RUN_QUEUE(tso);
+ advisory_thread_count++;
+
+ 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)));
+ }
+ } else {
+ barf("activateSpark: Cannot create TSO");
+ }
+ // 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()
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
* KH @ 25/10/99
* ------------------------------------------------------------------------ */
-#ifdef SMP
-static void *
+#if defined(PAR) || defined(SMP)
+void *
taskStart( void *arg STG_UNUSED )
{
- schedule();
- return NULL;
+ rts_evalNothing(NULL);
}
#endif
suspended_ccalling_threads = END_TSO_QUEUE;
main_threads = NULL;
+ all_threads = END_TSO_QUEUE;
context_switch = 0;
interrupted = 0;
-------------------------------------------------------------------------- */
/* -----------------------------------------------------------------------------
- * 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
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);
#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
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);
+ }
#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();
/* -----------------------------------------------------------------------------
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 *
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
+
+ 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)));
+
#ifdef INTERPRETER
fprintf(stderr, "fatal: stack overflow in Hugs; aborting\n" );
exit(1);
#else
/* Send this thread the StackOverflow exception */
- raiseAsync(tso, (StgClosure *)&stackOverflow_closure);
+ raiseAsync(tso, (StgClosure *)stackOverflow_closure);
#endif
return tso;
}
/* 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;
}
Wake up a queue that was blocked on some resource.
------------------------------------------------------------------------ */
-// ToDo: check push_on_run_queue vs. PUSH_ON_RUN_QUEUE
+/* ToDo: check push_on_run_queue vs. PUSH_ON_RUN_QUEUE */
#if defined(GRAN)
static inline void
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,
+ GR_RESUME, ((StgTSO *)bqe), ((StgTSO *)bqe)->block_info.closure,
0, 0 /* spark_queue_len(ADVISORY_POOL) */);
switch (get_itbl(node)->type) {
static StgBlockingQueueElement *
unblockOneLocked(StgBlockingQueueElement *bqe, StgClosure *node)
{
- StgBlockingQueueElement *next;
+ 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(tso);
+ 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);
- bq_processing_time += RtsFlags.GranFlags.Costs.lunblocktime;
+ CurrentTime[CurrentProc] += RtsFlags.GranFlags.Costs.lunblocktime;
// insertThread(tso, node_loc);
- new_event(tso_loc, tso_loc,
- CurrentTime[CurrentProc]+bq_processing_time,
+ 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)
- 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,
+ 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
- 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) ,",
+ 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->blocked_on, tso->link))
- tso->blocked_on = NULL;
+ 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));
- }
-
- /* 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);
- /*
- 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];
-
- IF_GRAN_DEBUG(bq,
- belch("## Filled in RBH_Save for %p (%s) at end of AwBQ",
- node, info_type(node)));
- }
}
#elif defined(PAR)
static StgBlockingQueueElement *
}
#endif
-#if defined(GRAN)
-inline StgTSO *
-unblockOne(StgTSO *tso, StgClosure *node)
-{
- ACQUIRE_LOCK(&sched_mutex);
- tso = unblockOneLocked(tso, node);
- RELEASE_LOCK(&sched_mutex);
- return tso;
-}
-#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 *
void
awakenBlockedQueue(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): ", \
*/
//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 (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 = (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",
+ node, info_type(node)));
+ }
+
/* statistics gathering */
- /* ToDo: fix counters
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)
void
}
#endif
-#if 0
-// ngoq ngo'
+//@node Exception Handling Routines, Debugging Routines, Blocking Queue Routines, Main scheduling code
+//@subsection Exception Handling Routines
+
+/* ---------------------------------------------------------------------------
+ Interrupt execution
+ - usually called inside a signal handler so it mustn't do anything fancy.
+ ------------------------------------------------------------------------ */
-#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)
+interruptStgRts(void)
{
- StgBlockingQueueElement *bqe, *next;
- StgTSO *tso;
- PEs node_loc, tso_loc;
- rtsTime bq_processing_time = 0;
- nat len = 0, len_local = 0;
-
- IF_GRAN_DEBUG(bq,
- 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
- get_itbl(q)->type == CONSTR); // closure (type constructor)
- ASSERT(is_unique(node));
-
- /* FAKE FETCH: magically copy the node to the tso's proc;
- no Fetch necessary because in reality the node should not have been
- moved to the other PE in the first place
- */
- if (CurrentProc!=node_loc) {
- IF_GRAN_DEBUG(bq,
- belch("## node %p is on PE %d but CurrentProc is %d (TSO %d); assuming fake fetch and adjusting bitmask (old: %#x)",
- node, node_loc, CurrentProc, CurrentTSO->id,
- // CurrentTSO, where_is(CurrentTSO),
- node->header.gran.procs));
- node->header.gran.procs = (node->header.gran.procs) | PE_NUMBER(CurrentProc);
- IF_GRAN_DEBUG(bq,
- belch("## new bitmask of node %p is %#x",
- node, node->header.gran.procs));
- if (RtsFlags.GranFlags.GranSimStats.Global) {
- globalGranStats.tot_fake_fetches++;
- }
- }
-
- next = q;
- // ToDo: check: ASSERT(CurrentProc==node_loc);
- while (get_itbl(next)->type==TSO) { // q != END_TSO_QUEUE) {
- bqe = next;
- 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));
- }
-
- /* 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);
- /*
- 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];
-
- IF_GRAN_DEBUG(bq,
- belch("## Filled in RBH_Save for %p (%s) at end of AwBQ",
- node, info_type(node)));
- }
-
- /* statistics gathering */
- if (RtsFlags.GranFlags.GranSimStats.Global) {
- 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_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" : ""));
+ interrupted = 1;
+ context_switch = 1;
}
-#elif defined(PAR)
+/* -----------------------------------------------------------------------------
+ Unblock a thread
+ This is for use when we raise an exception in another thread, which
+ may be blocked.
+ This has nothing to do with the UnblockThread event in GranSim. -- HWL
+ -------------------------------------------------------------------------- */
+
+#if defined(GRAN) || 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.
+ 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!
*/
-void
-awaken_blocked_queue(StgBlockingQueueElement *q, StgClosure *node)
+static void
+unblockThread(StgTSO *tso)
{
- StgBlockingQueueElement *bqe, *next;
-
- IF_PAR_DEBUG(verbose,
- belch("## AwBQ for node %p on [%x]: ",
- node, mytid));
+ StgBlockingQueueElement *t, **last;
- ASSERT(get_itbl(q)->type == TSO ||
- get_itbl(q)->type == BLOCKED_FETCH ||
- get_itbl(q)->type == CONSTR);
+ ACQUIRE_LOCK(&sched_mutex);
+ switch (tso->why_blocked) {
- 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);
+ case NotBlocked:
+ return; /* not blocked */
- /* 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));
+ case BlockedOnMVar:
+ ASSERT(get_itbl(tso->block_info.closure)->type == MVAR);
+ {
+ StgBlockingQueueElement *last_tso = END_BQ_QUEUE;
+ StgMVar *mvar = (StgMVar *)(tso->block_info.closure);
- 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));
+ 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;
}
}
- /* 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));
+ barf("unblockThread (MVAR): TSO not found");
+ }
- IF_DEBUG(scheduler,
- if (!RtsFlags.ParFlags.Debug.verbose)
- belch("-- Waking up thread %ld (%p)",
- ((StgTSO *)bqe)->id, (StgTSO *)bqe));
- break;
+ case BlockedOnBlackHole:
+ ASSERT(get_itbl(tso->block_info.closure)->type == BLACKHOLE_BQ);
+ {
+ StgBlockingQueue *bq = (StgBlockingQueue *)(tso->block_info.closure);
- 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;
+ 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");
+ }
- /* 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;
+ case BlockedOnException:
+ {
+ StgTSO *target = tso->block_info.tso;
-# 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;
+ ASSERT(get_itbl(target)->type == TSO);
+ ASSERT(target->blocked_exceptions != NULL);
- 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
+ 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");
}
- }
-}
-
-#else /* !GRAN && !PAR */
-
-void
-awaken_blocked_queue(StgTSO *q) { awakenBlockedQueue(q); }
-/*
-{
- StgTSO *tso;
+ case BlockedOnDelay:
+ case BlockedOnRead:
+ case BlockedOnWrite:
+ {
+ 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");
+ }
- 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));
+ default:
+ barf("unblockThread");
}
-}
-*/
-#endif /* GRAN */
-#endif /* 0 */
-
-//@node Exception Handling Routines, Debugging Routines, Blocking Queue Routines, Main scheduling code
-//@subsection Exception Handling Routines
-/* ---------------------------------------------------------------------------
- Interrupt execution
- - usually called inside a signal handler so it mustn't do anything fancy.
- ------------------------------------------------------------------------ */
-
-void
-interruptStgRts(void)
-{
- interrupted = 1;
- context_switch = 1;
+ done:
+ tso->link = END_TSO_QUEUE;
+ tso->why_blocked = NotBlocked;
+ tso->block_info.closure = NULL;
+ PUSH_ON_RUN_QUEUE(tso);
+ RELEASE_LOCK(&sched_mutex);
}
-
-/* -----------------------------------------------------------------------------
- Unblock a thread
-
- This is for use when we raise an exception in another thread, which
- may be blocked.
- This has nothing to do with the UnblockThread event in GranSim. -- HWL
- -------------------------------------------------------------------------- */
-
+#else
static void
unblockThread(StgTSO *tso)
{
PUSH_ON_RUN_QUEUE(tso);
RELEASE_LOCK(&sched_mutex);
}
+#endif
/* -----------------------------------------------------------------------------
* raiseAsync()
StgPtr sp = tso->sp;
/* Thread already dead? */
- if (tso->whatNext == ThreadComplete || tso->whatNext == ThreadKilled) {
+ if (tso->what_next == ThreadComplete || tso->what_next == ThreadKilled) {
return;
}
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);
+ ap = (StgAP_UPD *)allocate(sizeofW(StgPAP) + 2);
+ TICK_ALLOC_UPD_PAP(3,0);
SET_HDR(ap,&PAP_info,cf->header.prof.ccs);
- ap->n_args = 1;
- ap->fun = cf->handler;
+ ap->n_args = 2;
+ ap->fun = cf->handler; /* :: Exception -> IO a */
ap->payload[0] = (P_)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
*/
sp[0] = (W_)ap;
tso->sp = sp;
- tso->whatNext = ThreadEnterGHC;
+ tso->what_next = ThreadEnterGHC;
return;
}
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;
+ o->payload[0] = (StgClosure *)ap;
IF_DEBUG(scheduler,
fprintf(stderr, "scheduler: Built ");
/* 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;
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");
+ }
+ }
+}
+
//@node Debugging Routines, Index, Exception Handling Routines, Main scheduling code
//@subsection Debugging Routines
#ifdef DEBUG
-void printThreadBlockage(StgTSO *tso)
+void
+printThreadBlockage(StgTSO *tso)
{
switch (tso->why_blocked) {
case BlockedOnRead:
fprintf(stderr,"blocked on write to fd %d", tso->block_info.fd);
break;
case BlockedOnDelay:
+#if defined(HAVE_SETITIMER) || defined(mingw32_TARGET_OS)
fprintf(stderr,"blocked on delay of %d ms", tso->block_info.delay);
+#else
+ fprintf(stderr,"blocked on delay of %d ms",
+ tso->block_info.target - getourtimeofday());
+#endif
break;
case BlockedOnMVar:
fprintf(stderr,"blocked on an MVar");
break;
#if defined(PAR)
case BlockedOnGA:
- fprintf(stderr,"blocked on global address");
+ fprintf(stderr,"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,"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;
+
+ sched_belch("all threads:");
+ for (t = all_threads; t != END_TSO_QUEUE; t = t->global_link) {
+ fprintf(stderr, "\tthread %d is ", t->id);
+ printThreadStatus(t);
+ fprintf(stderr,"\n");
+ }
+}
+
/*
Print a whole blocking queue attached to node (debugging only).
*/
print_bq (StgClosure *node)
{
StgBlockingQueueElement *bqe;
- StgTSO *tso;
PEs node_loc, tso_loc;
rtsBool end;
!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);
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),",
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 */
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
#endif /* DEBUG */
+
//@node Index, , Debugging Routines, Main scheduling code
//@subsection Index
projects / ghc-hetmet.git / blobdiff