/* ----------------------------------------------------------------------------
- * Time-stamp: <Wed Jan 12 2000 13:32:25 Stardate: [-30]4193.86 hwloidl>
- * $Id: HLComms.c,v 1.2 2000/01/13 14:34:07 hwloidl Exp $
+ * Time-stamp: <Wed Mar 21 2001 16:34:41 Stardate: [-30]6363.45 hwloidl>
*
* High Level Communications Routines (HLComms.lc)
*
* Contains the high-level routines (i.e. communication
* subsystem independent) used by GUM
*
- * Phil Trinder, Glasgow University, 12 December 1994
- * Adapted for new RTS
- * Phil Trinder, Simon Marlow July 1998
- * H-W. Loidl, Heriot-Watt University, November 1999
+ * GUM 0.2x: Phil Trinder, Glasgow University, 12 December 1994
+ * GUM 3.xx: Phil Trinder, Simon Marlow July 1998
+ * GUM 4.xx: H-W. Loidl, Heriot-Watt University, November 1999 -
*
* ------------------------------------------------------------------------- */
//@node Macros etc, Includes, High Level Communications Routines, High Level Communications Routines
//@subsection Macros etc
-# ifndef _AIX
-# define NON_POSIX_SOURCE /* so says Solaris */
-# endif
+/* Evidently not Posix */
+/* #include "PosixSource.h" */
//@node Includes, GUM Message Sending and Unpacking Functions, Macros etc, High Level Communications Routines
//@subsection Includes
#include "Parallel.h"
#include "GranSimRts.h"
#include "ParallelRts.h"
+#include "Sparks.h"
#include "FetchMe.h" // for BLOCKED_FETCH_info etc
#if defined(DEBUG)
# include "ParallelDebug.h"
#endif
#include "StgMacros.h" // inlined IS_... fcts
+#ifdef DIST
+#include "SchedAPI.h" //for createIOThread
+extern unsigned int context_switch;
+#endif /* DIST */
+
//@node GUM Message Sending and Unpacking Functions, Message-Processing Functions, Includes, High Level Communications Routines
//@subsection GUM Message Sending and Unpacking Functions
{
ASSERT(rga->weight > 0 && lga->weight > 0);
IF_PAR_DEBUG(fetch,
- belch("** [%x] Sending Fetch for ((%x, %d, 0)); locally ((%x, %d, %x)), load = %d",
- mytid,
+ belch("~^** Sending Fetch for ((%x, %d, 0)); locally ((%x, %d, %x)), load = %d",
rga->payload.gc.gtid, rga->payload.gc.slot,
lga->payload.gc.gtid, lga->payload.gc.slot, lga->weight,
load));
GetArgs(buf, 6);
IF_PAR_DEBUG(fetch,
- belch("** [%x] Unpacking Fetch for ((%x, %d, 0)) to ((%x, %d, %x)), load = %d",
- mytid,
+ belch("~^** Unpacking Fetch for ((%x, %d, 0)) to ((%x, %d, %x)), load = %d",
(GlobalTaskId) buf[0], (int) buf[1],
(GlobalTaskId) buf[3], (int) buf[4], buf[2], buf[5]));
//@cindex sendResume
void
-sendResume(globalAddr *rga, int nelem, rtsPackBuffer *data) // StgPtr data)
+sendResume(globalAddr *rga, int nelem, rtsPackBuffer *packBuffer)
{
- IF_PAR_DEBUG(resume,
- PrintPacket(data);
- belch("[] [%x] Sending Resume for ((%x, %d, %x))",
- mytid,
- rga->payload.gc.gtid, rga->payload.gc.slot, rga->weight));
+ IF_PAR_DEBUG(fetch,
+ belch("~^[] Sending Resume (packet <<%d>> with %d elems) for ((%x, %d, %x)) to [%x]",
+ packBuffer->id, nelem,
+ rga->payload.gc.gtid, rga->payload.gc.slot, rga->weight,
+ rga->payload.gc.gtid));
+ IF_PAR_DEBUG(packet,
+ PrintPacket(packBuffer));
+
+ ASSERT(nelem==packBuffer->size);
+ /* check for magic end-of-buffer word */
+ IF_DEBUG(sanity, ASSERT(*(packBuffer->buffer+nelem) == END_OF_BUFFER_MARKER));
sendOpNV(PP_RESUME, rga->payload.gc.gtid,
- nelem + PACK_BUFFER_HDR_SIZE, (StgPtr)data,
+ nelem + PACK_BUFFER_HDR_SIZE + DEBUG_HEADROOM, (StgPtr)packBuffer,
2, (rtsWeight) rga->weight, (StgWord) rga->payload.gc.slot);
}
//@cindex unpackResume
static void
-unpackResume(globalAddr *lga, int *nelem, rtsPackBuffer *data)
+unpackResume(globalAddr *lga, int *nelem, rtsPackBuffer *packBuffer)
{
long buf[3];
GetArgs(buf, 3);
- IF_PAR_DEBUG(resume,
- belch("[] [%x] Unpacking Resume for ((%x, %d, %x))",
- mytid, mytid,
- (int) buf[1], (unsigned) buf[0]));
-
/*
RESUME event is written in awaken_blocked_queue
DumpRawGranEvent(CURRENT_PROC, taskIDtoPE(lga->payload.gc.gtid),
lga->payload.gc.gtid = mytid;
lga->payload.gc.slot = (int) buf[1];
- *nelem = (int) buf[2]; // includes PACK_BUFFER_HDR_SIZE;
- GetArgs(data, *nelem);
- *nelem -= PACK_BUFFER_HDR_SIZE;
+ *nelem = (int) buf[2] - PACK_BUFFER_HDR_SIZE - DEBUG_HEADROOM;
+ GetArgs(packBuffer, *nelem + PACK_BUFFER_HDR_SIZE + DEBUG_HEADROOM);
+
+ IF_PAR_DEBUG(fetch,
+ belch("~^[] Unpacking Resume (packet <<%d>> with %d elems) for ((%x, %d, %x))",
+ packBuffer->id, *nelem, mytid, (int) buf[1], (unsigned) buf[0]));
+
+ /* check for magic end-of-buffer word */
+ IF_DEBUG(sanity, ASSERT(*(packBuffer->buffer+*nelem) == END_OF_BUFFER_MARKER));
}
/*
long *p;
int i;
+ if(ngas==0)
+ return; //don't send unnecessary messages!!
+
buffer = (long *) gumPackBuffer;
for(i = 0, p = buffer; i < ngas; i++, p += 6) {
p[5] = (long) gagamap->payload.gc.slot;
gagamap++;
}
- IF_PAR_DEBUG(ack,
- belch(",, [%x] Sending Ack (%d pairs) to PE %x\n",
- mytid, ngas, task));
+ IF_PAR_DEBUG(schedule,
+ belch("~^,, Sending Ack (%d pairs) to [%x]\n",
+ ngas, task));
- sendOpN(PP_ACK, task, p - buffer, buffer);
+ sendOpN(PP_ACK, task, p - buffer, (StgPtr)buffer);
}
/*
*ngas = GAarraysize / 6;
- IF_PAR_DEBUG(ack,
- belch(",, [%x] Unpacking Ack (%d pairs) on %x\n",
- mytid, *ngas, mytid));
+ IF_PAR_DEBUG(schedule,
+ belch("~^,, Unpacking Ack (%d pairs) on [%x]\n",
+ *ngas, mytid));
while (GAarraysize > 0) {
GetArgs(buf, 6);
int age, int history, int hunger)
{
IF_PAR_DEBUG(fish,
- belch("$$ [%x] Sending Fish to %x (%d outstanding fishes)",
- mytid, destPE, outstandingFishes));
+ belch("~^$$ Sending Fish to [%x] (%d outstanding fishes)",
+ destPE, outstandingFishes));
sendOpV(PP_FISH, destPE, 4,
(StgWord) origPE, (StgWord) age, (StgWord) history, (StgWord) hunger);
GetArgs(buf, 4);
IF_PAR_DEBUG(fish,
- belch("$$ [%x] Unpacking Fish from PE %x (age=%d)",
- mytid, (GlobalTaskId) buf[0], (int) buf[1]));
+ belch("~^$$ Unpacking Fish from [%x] (age=%d)",
+ (GlobalTaskId) buf[0], (int) buf[1]));
*origPE = (GlobalTaskId) buf[0];
*age = (int) buf[1];
sendFree(GlobalTaskId pe, int nelem, StgPtr data)
{
IF_PAR_DEBUG(free,
- belch("!! [%x] Sending Free (%d GAs) to %x",
- mytid, nelem/2, pe));
+ belch("~^!! Sending Free (%d GAs) to [%x]",
+ nelem/2, pe));
sendOpN(PP_FREE, pe, nelem, data);
}
*/
//@cindex unpackFree
static void
-unpackFree(int *nelem, rtsPackBuffer *data)
+unpackFree(int *nelem, StgWord *data)
{
long buf[1];
*nelem = (int) buf[0];
IF_PAR_DEBUG(free,
- belch("!! [%x] Unpacking Free (%d GAs)",
- mytid, *nelem/2));
+ belch("~^!! Unpacking Free (%d GAs)",
+ *nelem/2));
GetArgs(data, *nelem);
}
*/
//@cindex sendSchedule
void
-sendSchedule(GlobalTaskId origPE, int nelem, rtsPackBuffer *data) // StgPtr data)
+sendSchedule(GlobalTaskId origPE, int nelem, rtsPackBuffer *packBuffer)
{
IF_PAR_DEBUG(schedule,
- PrintPacket(data);
- belch("-- [%x] Sending Schedule (%d elems) to %x\n",
- mytid, nelem, origPE));
+ belch("~^-- Sending Schedule (packet <<%d>> with %d elems) to [%x]\n",
+ packBuffer->id, nelem, origPE));
+ IF_PAR_DEBUG(packet,
+ PrintPacket(packBuffer));
+
+ ASSERT(nelem==packBuffer->size);
+ /* check for magic end-of-buffer word */
+ IF_DEBUG(sanity, ASSERT(*(packBuffer->buffer+nelem) == END_OF_BUFFER_MARKER));
- sendOpN(PP_SCHEDULE, origPE, nelem + PACK_BUFFER_HDR_SIZE, (StgPtr)data);
+ sendOpN(PP_SCHEDULE, origPE,
+ nelem + PACK_BUFFER_HDR_SIZE + DEBUG_HEADROOM, (StgPtr)packBuffer);
}
/*
//@cindex unpackSchedule
static void
-unpackSchedule(int *nelem, rtsPackBuffer *data)
+unpackSchedule(int *nelem, rtsPackBuffer *packBuffer)
{
- long buf[1];
+ long buf[1];
+
+ /* first, just unpack 1 word containing the total size (including header) */
+ GetArgs(buf, 1);
+ /* no. of elems, not counting the header of the pack buffer */
+ *nelem = (int) buf[0] - PACK_BUFFER_HDR_SIZE - DEBUG_HEADROOM;
- GetArgs(buf, 1);
- /* no. of elems, not counting the header of the pack buffer */
- *nelem = (int) buf[0] - PACK_BUFFER_HDR_SIZE;
+ /* automatic cast of flat pvm-data to rtsPackBuffer */
+ GetArgs(packBuffer, *nelem + PACK_BUFFER_HDR_SIZE + DEBUG_HEADROOM);
- IF_PAR_DEBUG(schedule,
- belch("-- [%x] Unpacking Schedule (%d elems) on %x\n",
- mytid, *nelem));
+ IF_PAR_DEBUG(schedule,
+ belch("~^-- Unpacking Schedule (packet <<%d>> with %d elems) on [%x]\n",
+ packBuffer->id, *nelem, mytid));
- /* automatic cast of flat pvm-data to rtsPackBuffer */
- GetArgs(data, *nelem + PACK_BUFFER_HDR_SIZE);
+ ASSERT(*nelem==packBuffer->size);
+ /* check for magic end-of-buffer word */
+ IF_DEBUG(sanity, ASSERT(*(packBuffer->buffer+*nelem) == END_OF_BUFFER_MARKER));
}
+#ifdef DIST
+/* sendReval is almost identical to the Schedule version, so we can unpack with unpackSchedule */
+void
+sendReval(GlobalTaskId origPE, int nelem, rtsPackBuffer *packBuffer)
+{
+ IF_PAR_DEBUG(schedule,
+ belch("~^-- Sending Reval (packet <<%d>> with %d elems) to [%x]\n",
+ packBuffer->id, nelem, origPE));
+ IF_PAR_DEBUG(packet,
+ PrintPacket(packBuffer));
+
+ ASSERT(nelem==packBuffer->size);
+ /* check for magic end-of-buffer word */
+ IF_DEBUG(sanity, ASSERT(*(packBuffer->buffer+nelem) == END_OF_BUFFER_MARKER));
+
+ sendOpN(PP_REVAL, origPE,
+ nelem + PACK_BUFFER_HDR_SIZE + DEBUG_HEADROOM, (StgPtr)packBuffer);
+}
+
+void FinishReval(StgTSO *t)
+{ StgClosure *res;
+ globalAddr ga;
+ nat size;
+ rtsPackBuffer *buffer=NULL;
+
+ ga.payload.gc.slot = t->revalSlot;
+ ga.payload.gc.gtid = t->revalTid;
+ ga.weight = 0;
+
+ //find where the reval result is
+ res = GALAlookup(&ga);
+ ASSERT(res);
+
+ IF_PAR_DEBUG(schedule,
+ printGA(&ga);
+ belch(" needs the result %08x\n",res));
+
+ //send off the result
+ buffer = PackNearbyGraph(res, END_TSO_QUEUE, &size,ga.payload.gc.gtid);
+ ASSERT(buffer != (rtsPackBuffer *)NULL);
+ sendResume(&ga, size, buffer);
+
+ IF_PAR_DEBUG(schedule,
+ belch("@;~) Reval Finished"));
+}
+
+#endif /* DIST */
+
//@node Message-Processing Functions, GUM Message Processor, GUM Message Sending and Unpacking Functions, High Level Communications Routines
//@subsection Message-Processing Functions
*/
/*
- * blockFetch blocks a BlockedFetch node on some kind of black hole.
- */
-//@cindex blockFetch
-static void
-blockFetch(StgBlockedFetch *bf, StgClosure *bh) {
- bf->node = bh;
- switch (get_itbl(bh)->type) {
- case BLACKHOLE:
- bf->link = END_BQ_QUEUE;
- //((StgBlockingQueue *)bh)->header.info = &BLACKHOLE_BQ_info;
- SET_INFO(bh, &BLACKHOLE_BQ_info); // turn closure into a blocking queue
- ((StgBlockingQueue *)bh)->blocking_queue = (StgBlockingQueueElement *)bf;
-
- // put bh on the mutables list
- recordMutable((StgMutClosure *)bh);
-
-# if 0
- /*
- * If we modify a black hole in the old generation, we have to
- * make sure it goes on the mutables list
- */
-
- if (bh <= StorageMgrInfo.OldLim) {
- MUT_LINK(bh) = (StgWord) StorageMgrInfo.OldMutables;
- StorageMgrInfo.OldMutables = bh;
- } else
- MUT_LINK(bh) = MUT_NOT_LINKED;
-# endif
- break;
-
- case BLACKHOLE_BQ:
- /* enqueue bf on blocking queue of closure bh */
- bf->link = ((StgBlockingQueue *)bh)->blocking_queue;
- ((StgBlockingQueue *)bh)->blocking_queue = (StgBlockingQueueElement *)bf;
-
- // put bh on the mutables list; ToDo: check
- recordMutable((StgMutClosure *)bh);
- break;
-
- case FETCH_ME_BQ:
- /* enqueue bf on blocking queue of closure bh */
- bf->link = ((StgFetchMeBlockingQueue *)bh)->blocking_queue;
- ((StgFetchMeBlockingQueue *)bh)->blocking_queue = (StgBlockingQueueElement *)bf;
-
- // put bh on the mutables list; ToDo: check
- recordMutable((StgMutClosure *)bh);
- break;
-
- case RBH:
- /* enqueue bf on blocking queue of closure bh */
- bf->link = ((StgRBH *)bh)->blocking_queue;
- ((StgRBH *)bh)->blocking_queue = (StgBlockingQueueElement *)bf;
-
- // put bh on the mutables list; ToDo: check
- recordMutable((StgMutClosure *)bh);
- break;
-
- default:
- barf("Panic (blockFetch): thought %p was a black hole (IP %#lx, %s)",
- (StgClosure *)bh, get_itbl((StgClosure *)bh),
- info_type((StgClosure *)bh));
- }
- IF_PAR_DEBUG(verbose,
- belch("## blockFetch: after block the BQ of %p (%s) is:",
- bh, info_type(bh));
- print_bq(bh));
-}
-
-
-/*
* processFetches constructs and sends resume messages for every
* BlockedFetch which is ready to be awakened.
* awaken_blocked_queue (in Schedule.c) is responsible for moving
//@cindex processFetches
void
processFetches(void) {
- StgBlockedFetch *bf;
- StgClosure *closure, *next;
+ StgBlockedFetch *bf, *next;
+ StgClosure *closure;
StgInfoTable *ip;
globalAddr rga;
static rtsPackBuffer *packBuffer;
IF_PAR_DEBUG(verbose,
- belch("__ processFetches: %d pending fetches",
- pending_fetches_len()));
+ belch("____ processFetches: %d pending fetches (root @ %p)",
+ pending_fetches_len(), PendingFetches));
for (bf = PendingFetches;
bf != END_BF_QUEUE;
- bf=(StgBlockedFetch *)(bf->link)) {
+ bf=next) {
/* the PendingFetches list contains only BLOCKED_FETCH closures */
ASSERT(get_itbl(bf)->type==BLOCKED_FETCH);
+ /* store link (we might overwrite it via blockFetch later on */
+ next = (StgBlockedFetch *)(bf->link);
/*
* Find the target at the end of the indirection chain, and
*/
closure = bf->node;
/*
- HACK 312: bf->node may have been evacuated since filling it; follow
- the evacuee in this case; the proper way to handle this is to
- traverse the blocking queue and update the node fields of
- BLOCKED_FETCH entries when evacuating an BLACKHOLE_BQ, FETCH_ME_BQ
- or RBH (but it's late and I'm tired)
+ We evacuate BQs and update the node fields where necessary in GC.c
+ So, if we find an EVACUATED closure, something has gone Very Wrong
+ (and therefore we let the RTS crash most ungracefully).
*/
- if (get_itbl(closure)->type == EVACUATED)
- closure = ((StgEvacuated *)closure)->evacuee;
+ ASSERT(get_itbl(closure)->type != EVACUATED);
+ // closure = ((StgEvacuated *)closure)->evacuee;
- while ((next = IS_INDIRECTION(closure)) != NULL) { closure = next; }
+ closure = UNWIND_IND(closure);
+ //while ((ind = IS_INDIRECTION(closure)) != NULL) { closure = ind; }
ip = get_itbl(closure);
if (ip->type == FETCH_ME) {
sendFetch(((StgFetchMe *)closure)->ga, &rga, 0 /* load */);
- IF_PAR_DEBUG(forward,
- belch("__ processFetches: Forwarding fetch from %lx to %lx",
+ // Global statistics: count no. of fetches
+ if (RtsFlags.ParFlags.ParStats.Global &&
+ RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
+ globalParStats.tot_fetch_mess++;
+ }
+
+ IF_PAR_DEBUG(fetch,
+ belch("__-> processFetches: Forwarding fetch from %lx to %lx",
mytid, rga.payload.gc.gtid));
} else if (IS_BLACK_HOLE(closure)) {
IF_PAR_DEBUG(verbose,
- belch("__ processFetches: trying to send a BLACK_HOLE => doign a blockFetch on closure %p (%s)",
+ belch("__++ processFetches: trying to send a BLACK_HOLE => doing a blockFetch on closure %p (%s)",
closure, info_type(closure)));
bf->node = closure;
blockFetch(bf, closure);
packBuffer = gumPackBuffer;
IF_PAR_DEBUG(verbose,
- belch("__ processFetches: PackNearbyGraph of closure %p (%s)",
+ belch("__*> processFetches: PackNearbyGraph of closure %p (%s)",
closure, info_type(closure)));
- if ((packBuffer = PackNearbyGraph(closure, END_TSO_QUEUE, &size)) == NULL) {
+ if ((packBuffer = PackNearbyGraph(closure, END_TSO_QUEUE, &size, bf->ga.payload.gc.gtid)) == NULL) {
// Put current BF back on list
bf->link = (StgBlockingQueueElement *)PendingFetches;
PendingFetches = (StgBlockedFetch *)bf;
// ToDo: check that nothing more has to be done to prepare for GC!
- GarbageCollect(GetRoots);
+ barf("processFetches: out of heap while packing graph; ToDo: call GC here");
+ GarbageCollect(GetRoots, rtsFalse);
bf = PendingFetches;
PendingFetches = (StgBlockedFetch *)(bf->link);
closure = bf->node;
- packBuffer = PackNearbyGraph(closure, END_TSO_QUEUE, &size);
+ packBuffer = PackNearbyGraph(closure, END_TSO_QUEUE, &size, bf->ga.payload.gc.gtid);
ASSERT(packBuffer != (rtsPackBuffer *)NULL);
}
rga.payload.gc.gtid = bf->ga.payload.gc.gtid;
rga.weight = bf->ga.weight;
sendResume(&rga, size, packBuffer);
+
+ // Global statistics: count no. of fetches
+ if (RtsFlags.ParFlags.ParStats.Global &&
+ RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
+ globalParStats.tot_resume_mess++;
+ }
}
}
PendingFetches = END_BF_QUEUE;
/* the old version did this in the FETCH_ME entry code */
sendFetch(&theGlobalFromGA, &theGlobalToGA, 0/*load*/);
-#if DEBUG
- theGlobalFromGA.payload.gc.gtid = 0;
- theGlobalToGA.payload.gc.gtid = 0;
-#endif DEBUG
}
#endif
+/*
+ Way of dealing with unwanted fish.
+ Used during startup/shutdown, or from unknown PEs
+*/
+void
+bounceFish(void) {
+ GlobalTaskId origPE;
+ int age, history, hunger;
+
+ /* IF_PAR_DEBUG(verbose, */
+ belch(".... [%x] Bouncing unwanted FISH",mytid);
+
+ unpackFish(&origPE, &age, &history, &hunger);
+
+ if (origPE == mytid) {
+ //fishing = rtsFalse; // fish has come home
+ outstandingFishes--;
+ last_fish_arrived_at = CURRENT_TIME; // remember time (see schedule fct)
+ return; // that's all
+ }
+
+ /* otherwise, send it home to die */
+ sendFish(origPE, origPE, (age + 1), NEW_FISH_HISTORY, NEW_FISH_HUNGER);
+ // Global statistics: count no. of fetches
+ if (RtsFlags.ParFlags.ParStats.Global &&
+ RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
+ globalParStats.tot_fish_mess++;
+ }
+}
+
/*
* processFish unpacks a fish message, reissuing it if it's our own,
* sending work if we have it or sending it onwards otherwise.
ASSERT(origPE != mytid);
IF_PAR_DEBUG(fish,
- belch("$$ [%x] processing fish; %d sparks available",
- mytid, spark_queue_len(ADVISORY_POOL)));
- while ((spark = findLocalSpark(rtsTrue)) != NULL) {
+ belch("$$__ processing fish; %d sparks available",
+ spark_queue_len(&(MainRegTable.rSparks))));
+ while ((spark = findSpark(rtsTrue/*for_export*/)) != NULL) {
nat size;
// StgClosure *graph;
packBuffer = gumPackBuffer;
ASSERT(closure_SHOULD_SPARK((StgClosure *)spark));
- if ((packBuffer = PackNearbyGraph(spark, END_TSO_QUEUE, &size)) == NULL) {
+ if ((packBuffer = PackNearbyGraph(spark, END_TSO_QUEUE, &size,origPE)) == NULL) {
IF_PAR_DEBUG(fish,
belch("$$ GC while trying to satisfy FISH via PackNearbyGraph of node %p",
(StgClosure *)spark));
- GarbageCollect(GetRoots);
+ barf("processFish: out of heap while packing graph; ToDo: call GC here");
+ GarbageCollect(GetRoots, rtsFalse);
/* Now go back and try again */
} else {
+ IF_PAR_DEBUG(verbose,
+ if (RtsFlags.ParFlags.ParStats.Sparks)
+ belch("==== STEALING spark %x; sending to %x", spark, origPE));
+
IF_PAR_DEBUG(fish,
- belch("$$ [%x] Replying to FISH from %x by sending graph @ %p (%s)",
- mytid, origPE,
+ belch("$$-- Replying to FISH from %x by sending graph @ %p (%s)",
+ origPE,
(StgClosure *)spark, info_type((StgClosure *)spark)));
sendSchedule(origPE, size, packBuffer);
disposeSpark(spark);
+ // Global statistics: count no. of fetches
+ if (RtsFlags.ParFlags.ParStats.Global &&
+ RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
+ globalParStats.tot_schedule_mess++;
+ }
+
break;
}
}
if (spark == (rtsSpark)NULL) {
IF_PAR_DEBUG(fish,
- belch("$$ [%x] No sparks available for FISH from %x",
- mytid, origPE));
+ belch("$$^^ No sparks available for FISH from %x",
+ origPE));
/* We have no sparks to give */
- if (age < FISH_LIFE_EXPECTANCY)
+ if (age < FISH_LIFE_EXPECTANCY) {
/* and the fish is atill young, send it to another PE to look for work */
sendFish(choosePE(), origPE,
(age + 1), NEW_FISH_HISTORY, NEW_FISH_HUNGER);
- /* otherwise, send it home to die */
- else
+ // Global statistics: count no. of fetches
+ if (RtsFlags.ParFlags.ParStats.Global &&
+ RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
+ globalParStats.tot_fish_mess++;
+ }
+ } else { /* otherwise, send it home to die */
sendFish(origPE, origPE, (age + 1), NEW_FISH_HISTORY, NEW_FISH_HUNGER);
+ // Global statistics: count no. of fetches
+ if (RtsFlags.ParFlags.ParStats.Global &&
+ RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
+ globalParStats.tot_fish_mess++;
+ }
}
+ }
} /* processFish */
/*
unpackFetch(&ga, &rga, &load);
IF_PAR_DEBUG(fetch,
- belch("%% [%x] Rcvd Fetch for ((%x, %d, 0)), Resume ((%x, %d, %x)) (load %d) from %x",
- mytid,
+ belch("%%%%__ Rcvd Fetch for ((%x, %d, 0)), Resume ((%x, %d, %x)) (load %d) from %x",
ga.payload.gc.gtid, ga.payload.gc.slot,
rga.payload.gc.gtid, rga.payload.gc.slot, rga.weight, load,
rga.payload.gc.gtid));
if (ip->type == FETCH_ME) {
/* Forward the Fetch to someone else */
sendFetch(((StgFetchMe *)closure)->ga, &rga, load);
+
+ // Global statistics: count no. of fetches
+ if (RtsFlags.ParFlags.ParStats.Global &&
+ RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
+ globalParStats.tot_fetch_mess++;
+ }
} else if (rga.payload.gc.gtid == mytid) {
/* Our own FETCH forwarded back around to us */
StgFetchMeBlockingQueue *fmbq = (StgFetchMeBlockingQueue *)GALAlookup(&rga);
IF_PAR_DEBUG(fetch,
- belch("%% [%x] Fetch returned to sending PE; closure=%p (%s); receiver=%p (%s)",
- mytid, closure, info_type(closure), fmbq, info_type(fmbq)));
+ belch("%%%%== Fetch returned to sending PE; closure=%p (%s); receiver=%p (%s)",
+ closure, info_type(closure), fmbq, info_type((StgClosure*)fmbq)));
/* We may have already discovered that the fetch target is our own. */
if ((StgClosure *)fmbq != closure)
CommonUp((StgClosure *)fmbq, closure);
/* This includes RBH's and FMBQ's */
StgBlockedFetch *bf;
+ /* Can we assert something on the remote GA? */
ASSERT(GALAlookup(&rga) == NULL);
/* If we're hitting a BH or RBH or FMBQ we have to put a BLOCKED_FETCH
closure into the BQ in order to denote that when updating this node
the result should be sent to the originator of this fetch message. */
bf = (StgBlockedFetch *)createBlockedFetch(ga, rga);
- blockFetch(bf, closure);
-
IF_PAR_DEBUG(fetch,
- belch("%% [%x] Blocking Fetch ((%x, %d, %x)) on %p (%s)",
- mytid,
+ belch("%%++ Blocking Fetch ((%x, %d, %x)) on %p (%s)",
rga.payload.gc.gtid, rga.payload.gc.slot, rga.weight,
closure, info_type(closure)));
- } else {
- /* The target of the FetchMe is some local graph */
- nat size;
- // StgClosure *graph;
- rtsPackBuffer *buffer = (rtsPackBuffer *)NULL;
-
- if ((buffer = PackNearbyGraph(closure, END_TSO_QUEUE, &size)) == NULL) {
- GarbageCollect(GetRoots);
- closure = GALAlookup(&ga);
- buffer = PackNearbyGraph(closure, END_TSO_QUEUE, &size);
- ASSERT(buffer != (rtsPackBuffer *)NULL);
- }
- sendResume(&rga, size, buffer);
+ blockFetch(bf, closure);
+ } else {
+ /* The target of the FetchMe is some local graph */
+ nat size;
+ // StgClosure *graph;
+ rtsPackBuffer *buffer = (rtsPackBuffer *)NULL;
+
+ if ((buffer = PackNearbyGraph(closure, END_TSO_QUEUE, &size, rga.payload.gc.gtid)) == NULL) {
+ barf("processFetch: out of heap while packing graph; ToDo: call GC here");
+ GarbageCollect(GetRoots, rtsFalse);
+ closure = GALAlookup(&ga);
+ buffer = PackNearbyGraph(closure, END_TSO_QUEUE, &size, rga.payload.gc.gtid);
+ ASSERT(buffer != (rtsPackBuffer *)NULL);
+ }
+ sendResume(&rga, size, buffer);
+
+ // Global statistics: count no. of fetches
+ if (RtsFlags.ParFlags.ParStats.Global &&
+ RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
+ globalParStats.tot_resume_mess++;
}
+ }
+}
+
+/*
+ The list of pending fetches must be a root-list for GC.
+ This routine is called from GC.c (same as marking GAs etc).
+*/
+void
+markPendingFetches(rtsBool major_gc) {
+
+ /* No need to traverse the list; this is done via the scavenge code
+ for a BLOCKED_FETCH closure, which evacuates the link field */
+
+ if (PendingFetches != END_BF_QUEUE ) {
+ IF_PAR_DEBUG(tables,
+ fprintf(stderr, "@@@@ PendingFetches is root; evaced from %p to",
+ PendingFetches));
+
+ PendingFetches = MarkRoot((StgClosure*)PendingFetches);
+
+ IF_PAR_DEBUG(verbose,
+ fprintf(stderr, " %p\n", PendingFetches));
+
+ } else {
+ IF_PAR_DEBUG(tables,
+ fprintf(stderr, "@@@@ PendingFetches is empty; no need to mark it\n"));
+ }
}
/*
buffer = (StgWord *)gumPackBuffer;
unpackFree(&nelem, buffer);
IF_PAR_DEBUG(free,
- belch("!! [%x] Rcvd Free (%d GAs)", mytid, nelem / 2));
+ belch("!!__ Rcvd Free (%d GAs)", nelem / 2));
ga.payload.gc.gtid = mytid;
for (i = 0; i < nelem;) {
ga.weight = (rtsWeight) buffer[i++];
ga.payload.gc.slot = (int) buffer[i++];
IF_PAR_DEBUG(free,
- fprintf(stderr, "!! [%x] Processing free ", mytid);
+ fprintf(stderr, "!!-- Processing free ");
printGA(&ga);
fputc('\n', stderr);
);
globalAddr lga;
globalAddr *gagamap;
- packBuffer = gumPackBuffer;
- unpackResume(&lga, &nelem, (StgPtr)packBuffer);
+ packBuffer = (rtsPackBuffer *)gumPackBuffer;
+ unpackResume(&lga, &nelem, packBuffer);
- IF_PAR_DEBUG(resume,
- fprintf(stderr, "[] [%x] Rcvd Resume for ", mytid);
+ IF_PAR_DEBUG(fetch,
+ fprintf(stderr, "[]__ Rcvd Resume for ");
printGA(&lga);
- fputc('\n', stderr);
+ fputc('\n', stderr));
+ IF_PAR_DEBUG(packet,
PrintPacket((rtsPackBuffer *)packBuffer));
/*
old = GALAlookup(&lga);
+ /* ToDo: The closure that requested this graph must be one of these two?*/
+ ASSERT(get_itbl(old)->type == FETCH_ME_BQ ||
+ get_itbl(old)->type == RBH);
+
if (RtsFlags.ParFlags.ParStats.Full) {
- // StgTSO *tso = END_TSO_QUEUE;
- StgBlockingQueueElement *bqe;
+ StgBlockingQueueElement *bqe, *last_bqe;
+
+ IF_PAR_DEBUG(fetch,
+ belch("[]-- Resume is REPLY to closure %lx", old));
/* Write REPLY events to the log file, indicating that the remote
- data has arrived */
- if (get_itbl(old)->type == FETCH_ME_BQ ||
- get_itbl(old)->type == RBH)
- for (bqe = ((StgFetchMeBlockingQueue *)old)->blocking_queue;
- bqe->link != END_BQ_QUEUE;
- bqe = bqe->link)
- if (get_itbl((StgClosure *)bqe)->type == TSO)
- DumpRawGranEvent(CURRENT_PROC, taskIDtoPE(sender),
- GR_REPLY, ((StgTSO *)bqe), ((StgTSO *)bqe)->block_info.closure,
- 0, spark_queue_len(ADVISORY_POOL));
+ data has arrived
+ NB: we emit a REPLY only for the *last* elem in the queue; this is
+ the one that triggered the fetch message; all other entries
+ have just added themselves to the queue, waiting for the data
+ they know that has been requested (see entry code for FETCH_ME_BQ)
+ */
+ if ((get_itbl(old)->type == FETCH_ME_BQ ||
+ get_itbl(old)->type == RBH)) {
+ for (bqe = ((StgFetchMeBlockingQueue *)old)->blocking_queue,
+ last_bqe = END_BQ_QUEUE;
+ get_itbl(bqe)->type==TSO ||
+ get_itbl(bqe)->type==BLOCKED_FETCH;
+ last_bqe = bqe, bqe = bqe->link) { /* nothing */ }
+
+ ASSERT(last_bqe==END_BQ_QUEUE ||
+ get_itbl((StgClosure *)last_bqe)->type == TSO);
+
+ /* last_bqe now points to the TSO that triggered the FETCH */
+ if (get_itbl((StgClosure *)last_bqe)->type == TSO)
+ DumpRawGranEvent(CURRENT_PROC, taskIDtoPE(sender),
+ GR_REPLY, ((StgTSO *)last_bqe), ((StgTSO *)last_bqe)->block_info.closure,
+ 0, spark_queue_len(&(MainRegTable.rSparks)));
+ }
}
newGraph = UnpackGraph(packBuffer, &gagamap, &nGAs);
if (get_itbl(old)->type == FETCH_ME_BQ)
CommonUp(old, newGraph);
- IF_PAR_DEBUG(resume,
+ IF_PAR_DEBUG(fetch,
+ belch("[]-- Ready to resume unpacked graph at %p (%s)",
+ newGraph, info_type(newGraph)));
+
+ IF_PAR_DEBUG(tables,
DebugPrintGAGAMap(gagamap, nGAs));
sendAck(sender, nGAs, gagamap);
static void
processSchedule(GlobalTaskId sender)
{
- nat nelem, space_required, nGAs;
+ nat nelem, nGAs;
rtsBool success;
static rtsPackBuffer *packBuffer;
StgClosure *newGraph;
unpackSchedule(&nelem, packBuffer);
IF_PAR_DEBUG(schedule,
- belch("-- [%x] Rcvd Schedule (%d elems)", mytid, nelem);
+ belch("--__ Rcvd Schedule (%d elems)", nelem));
+ IF_PAR_DEBUG(packet,
PrintPacket(packBuffer));
/*
SAVE_Hp -= space_required;
}
*/
- // ToDo: check whether GC is necessary !!!!!!!!!!!!!!!!!!!!!1
+ // ToDo: check whether GC is necessary !!!!!!!!!!!!!!!!!!!!!
newGraph = UnpackGraph(packBuffer, &gagamap, &nGAs);
ASSERT(newGraph != NULL);
- success = add_to_spark_queue(newGraph, rtsFalse);
+ success = add_to_spark_queue(newGraph, &(MainRegTable.rSparks));
- IF_PAR_DEBUG(pack,
+ if (RtsFlags.ParFlags.ParStats.Full &&
+ RtsFlags.ParFlags.ParStats.Sparks &&
+ success)
+ DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC,
+ GR_STOLEN, ((StgTSO *)NULL), newGraph,
+ 0, 0 /* spark_queue_len(ADVISORY_POOL) */);
+
+ IF_PAR_DEBUG(schedule,
if (success)
- belch("+* added spark to unpacked graph %p; %d sparks available on [%x]",
- newGraph, spark_queue_len(ADVISORY_POOL), mytid);
+ belch("--^^ added spark to unpacked graph %p (%s); %d sparks available on [%x] (%s)",
+ newGraph, info_type(newGraph), spark_queue_len(&(MainRegTable.rSparks)), mytid);
else
- belch("+* received non-sparkable closure %p; nothing added to spark pool; %d sparks available on [%x]",
- newGraph, spark_queue_len(ADVISORY_POOL), mytid);
- belch("-* Unpacked graph with root at %p (%s):",
+ belch("--^^ received non-sparkable closure %p (%s); nothing added to spark pool; %d sparks available on [%x]",
+ newGraph, info_type(newGraph), spark_queue_len(&(MainRegTable.rSparks)), mytid));
+ IF_PAR_DEBUG(packet,
+ belch("*< Unpacked graph with root at %p (%s):",
newGraph, info_type(newGraph));
PrintGraph(newGraph, 0));
- IF_PAR_DEBUG(pack,
+ IF_PAR_DEBUG(tables,
DebugPrintGAGAMap(gagamap, nGAs));
- if (nGAs > 0)
- sendAck(sender, nGAs, gagamap);
+ sendAck(sender, nGAs, gagamap);
//fishing = rtsFalse;
ASSERT(outstandingFishes>0);
unpackAck(&nGAs, gagamap);
- IF_PAR_DEBUG(ack,
- belch(",, [%x] Rcvd Ack (%d pairs)", mytid, nGAs);
+ IF_PAR_DEBUG(tables,
+ belch(",,,, Rcvd Ack (%d pairs)", nGAs);
DebugPrintGAGAMap(gagamap, nGAs));
+ IF_DEBUG(sanity,
+ checkGAGAMap(gagamap, nGAs));
+
/*
* For each (oldGA, newGA) pair, set the GA of the corresponding
* thunk to the newGA, convert the thunk to a FetchMe, and return
ASSERT(get_itbl(old_closure)==RBH);
*/
if (get_itbl(old_closure)->type==RBH)
- convertToFetchMe(old_closure, ga);
+ convertToFetchMe((StgRBH *)old_closure, ga);
} else {
/*
* Oops...we've got this one already; update the RBH to
}
(void) addWeight(gaga);
}
+
+ /* check the sanity of the LAGA and GALA tables after mincing them */
+ IF_DEBUG(sanity, checkLAGAtable(rtsFalse));
}
+#ifdef DIST
+
+void
+bounceReval(void) {
+ barf("Task %x: TODO: should send NACK in response to REVAL",mytid);
+}
+
+static void
+processReval(GlobalTaskId sender) //similar to schedule...
+{ nat nelem, space_required, nGAs;
+ static rtsPackBuffer *packBuffer;
+ StgClosure *newGraph;
+ globalAddr *gagamap;
+ StgTSO* tso;
+ globalAddr *ga;
+
+ packBuffer = gumPackBuffer; /* HWL */
+ unpackSchedule(&nelem, packBuffer); /* okay, since the structure is the same */
+
+ IF_PAR_DEBUG(packet,
+ belch("@;~) [%x] Rcvd Reval (%d elems)", mytid, nelem);
+ PrintPacket(packBuffer));
+
+ /*
+ space_required = packBuffer[0];
+ if (SAVE_Hp + space_required >= SAVE_HpLim) {
+ ReallyPerformThreadGC(space_required, rtsFalse);
+ SAVE_Hp -= space_required;
+ }
+ */
+
+ // ToDo: check whether GC is necessary !!!!!!!!!!!!!!!!!!!!!
+ newGraph = UnpackGraph(packBuffer, &gagamap, &nGAs);
+ ASSERT(newGraph != NULL);
+
+ IF_PAR_DEBUG(packet,
+ belch("@;~) Unpacked graph with root at %p (%s):",
+ newGraph, info_type(newGraph));
+ PrintGraph(newGraph, 0));
+
+ IF_PAR_DEBUG(tables,
+ DebugPrintGAGAMap(gagamap, nGAs));
+
+ IF_PAR_DEBUG(tables,
+ printLAGAtable();
+ DebugPrintGAGAMap(gagamap, nGAs));
+
+ //We don't send an Ack to the head!!!!
+ ASSERT(nGAs>0);
+ sendAck(sender, nGAs-1, gagamap+2);
+
+ IF_PAR_DEBUG(verbose,
+ belch("@;~) About to create Reval thread on behalf of %x",
+ sender));
+
+ tso=createGenThread(RtsFlags.GcFlags.initialStkSize,newGraph);
+ tso->priority=RevalPriority;
+ tso->revalSlot=gagamap->payload.gc.slot;//record who sent the reval
+ tso->revalTid =gagamap->payload.gc.gtid;
+ scheduleThread(tso);
+ context_switch = 1; // switch at the earliest opportunity
+}
+#endif
+
+
//@node GUM Message Processor, Miscellaneous Functions, Message-Processing Functions, High Level Communications Routines
//@subsection GUM Message Processor
*/
//@cindex processMessages
-void
+rtsBool
processMessages(void)
{
rtsPacket packet;
OpCode opcode;
GlobalTaskId task;
-
+ rtsBool receivedFinish = rtsFalse;
+
do {
packet = GetPacket(); /* Get next message; block until one available */
getOpcodeAndSender(packet, &opcode, &task);
- switch (opcode) {
- case PP_FINISH:
- IF_PAR_DEBUG(verbose,
- belch("== [%x] received FINISH", mytid));
- /* setting this global variables eventually terminates the main
- scheduling loop for this PE and causes a shut-down, sending
- PP_FINISH to SysMan */
- GlobalStopPending = rtsTrue;
- break;
+ if (task==SysManTask) {
+ switch (opcode) {
+ case PP_PETIDS:
+ processPEtids();
+ break;
+
+ case PP_FINISH:
+ IF_PAR_DEBUG(verbose,
+ belch("==== received FINISH [%p]", mytid));
+ /* this boolean value is returned and propagated to the main
+ scheduling loop, thus shutting-down this PE */
+ receivedFinish = rtsTrue;
+ break;
+
+ default:
+ barf("Task %x: received unknown opcode %x from SysMan",mytid, opcode);
+ }
+ } else if (taskIDtoPE(task)==0) {
+ /* When a new PE joins then potentially FISH & REVAL message may
+ reach PES before they are notified of the new PEs existance. The
+ only solution is to bounce/fail these messages back to the sender.
+ But we will worry about it once we start seeing these race
+ conditions! */
+ switch (opcode) {
+ case PP_FISH:
+ bounceFish();
+ break;
+#ifdef DIST
+ case PP_REVAL:
+ bounceReval();
+ break;
+#endif
+ case PP_PETIDS:
+ belch("Task %x: Ignoring PVM session opened by another SysMan %x",mytid,task);
+ break;
+
+ case PP_FINISH:
+ break;
+
+ default:
+ belch("Task %x: Ignoring opcode %x from unknown PE %x",mytid, opcode, task);
+ }
+ } else
+ switch (opcode) {
+ case PP_FETCH:
+ processFetch();
+ // Global statistics: count no. of fetches
+ if (RtsFlags.ParFlags.ParStats.Global &&
+ RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
+ globalParStats.rec_fetch_mess++;
+ }
+ break;
+
+ case PP_RESUME:
+ processResume(task);
+ // Global statistics: count no. of fetches
+ if (RtsFlags.ParFlags.ParStats.Global &&
+ RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
+ globalParStats.rec_resume_mess++;
+ }
+ break;
+
+ case PP_ACK:
+ processAck();
+ break;
+
+ case PP_FISH:
+ processFish();
+ // Global statistics: count no. of fetches
+ if (RtsFlags.ParFlags.ParStats.Global &&
+ RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
+ globalParStats.rec_fish_mess++;
+ }
+ break;
+
+ case PP_FREE:
+ processFree();
+ break;
+
+ case PP_SCHEDULE:
+ processSchedule(task);
+ // Global statistics: count no. of fetches
+ if (RtsFlags.ParFlags.ParStats.Global &&
+ RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
+ globalParStats.rec_schedule_mess++;
+ }
+ break;
+
+#ifdef DIST
+ case PP_REVAL:
+ processReval(task);
+ // Global statistics: count no. of fetches
+ if (RtsFlags.ParFlags.ParStats.Global &&
+ RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
+ globalParStats.rec_reval_mess++;
+ }
+ break;
+#endif
+
+ default:
+ /* Anything we're not prepared to deal with. */
+ barf("Task %x: Unexpected opcode %x from %x",
+ mytid, opcode, task);
+ } /* switch */
- case PP_FETCH:
- processFetch();
- break;
+ } while (PacketsWaiting()); /* While there are messages: process them */
+ return receivedFinish;
+} /* processMessages */
- case PP_RESUME:
- processResume(task);
- break;
+//@node Miscellaneous Functions, Index, GUM Message Processor, High Level Communications Routines
+//@subsection Miscellaneous Functions
+
+/*
+ * blockFetch blocks a BlockedFetch node on some kind of black hole.
+ */
+//@cindex blockFetch
+void
+blockFetch(StgBlockedFetch *bf, StgClosure *bh) {
+ bf->node = bh;
+ switch (get_itbl(bh)->type) {
+ case BLACKHOLE:
+ bf->link = END_BQ_QUEUE;
+ //((StgBlockingQueue *)bh)->header.info = &stg_BLACKHOLE_BQ_info;
+ SET_INFO(bh, &stg_BLACKHOLE_BQ_info); // turn closure into a blocking queue
+ ((StgBlockingQueue *)bh)->blocking_queue = (StgBlockingQueueElement *)bf;
+
+ // put bh on the mutables list
+ recordMutable((StgMutClosure *)bh);
+ break;
+
+ case BLACKHOLE_BQ:
+ /* enqueue bf on blocking queue of closure bh */
+ bf->link = ((StgBlockingQueue *)bh)->blocking_queue;
+ ((StgBlockingQueue *)bh)->blocking_queue = (StgBlockingQueueElement *)bf;
- case PP_ACK:
- processAck();
+ // put bh on the mutables list; ToDo: check
+ recordMutable((StgMutClosure *)bh);
+ break;
+
+ case FETCH_ME_BQ:
+ /* enqueue bf on blocking queue of closure bh */
+ bf->link = ((StgFetchMeBlockingQueue *)bh)->blocking_queue;
+ ((StgFetchMeBlockingQueue *)bh)->blocking_queue = (StgBlockingQueueElement *)bf;
+
+ // put bh on the mutables list; ToDo: check
+ recordMutable((StgMutClosure *)bh);
+ break;
+
+ case RBH:
+ /* enqueue bf on blocking queue of closure bh */
+ bf->link = ((StgRBH *)bh)->blocking_queue;
+ ((StgRBH *)bh)->blocking_queue = (StgBlockingQueueElement *)bf;
+
+ // put bh on the mutables list; ToDo: check
+ recordMutable((StgMutClosure *)bh);
+ break;
+
+ default:
+ barf("blockFetch: thought %p was a black hole (IP %#lx, %s)",
+ (StgClosure *)bh, get_itbl((StgClosure *)bh),
+ info_type((StgClosure *)bh));
+ }
+ IF_PAR_DEBUG(bq,
+ belch("##++ blockFetch: after block the BQ of %p (%s) is:",
+ bh, info_type(bh));
+ print_bq(bh));
+}
+
+
+/*
+ @blockThread@ is called from the main scheduler whenever tso returns with
+ a ThreadBlocked return code; tso has already been added to a blocking
+ queue (that's done in the entry code of the closure, because it is a
+ cheap operation we have to do in any case); the main purpose of this
+ routine is to send a Fetch message in case we are blocking on a FETCHME(_BQ)
+ closure, which is indicated by the tso.why_blocked field;
+ we also write an entry into the log file if we are generating one
+
+ Should update exectime etc in the entry code already; but we don't have
+ something like ``system time'' in the log file anyway, so this should
+ even out the inaccuracies.
+*/
+
+//@cindex blockThread
+void
+blockThread(StgTSO *tso)
+{
+ globalAddr *remote_ga=NULL;
+ globalAddr *local_ga;
+ globalAddr fmbq_ga;
+
+ // ASSERT(we are on some blocking queue)
+ ASSERT(tso->block_info.closure != (StgClosure *)NULL);
+
+ /*
+ We have to check why this thread has been blocked.
+ */
+ switch (tso->why_blocked) {
+ case BlockedOnGA:
+ /* the closure must be a FETCH_ME_BQ; tso came in here via
+ FETCH_ME entry code */
+ ASSERT(get_itbl(tso->block_info.closure)->type==FETCH_ME_BQ);
+
+ /* HACK: the link field is used to hold the GA between FETCH_ME_entry
+ end this point; if something (eg. GC) happens inbetween the whole
+ thing will blow up
+ The problem is that the ga field of the FETCH_ME has been overwritten
+ with the head of the blocking queue (which is tso).
+ */
+ ASSERT(looks_like_ga(&theGlobalFromGA));
+ // ASSERT(tso->link!=END_TSO_QUEUE && tso->link!=NULL);
+ remote_ga = &theGlobalFromGA; //tso->link;
+ tso->link = (StgTSO*)END_BQ_QUEUE;
+ /* it was tso which turned node from FETCH_ME into FETCH_ME_BQ =>
+ we have to send a Fetch message here! */
+ if (RtsFlags.ParFlags.ParStats.Full) {
+ /* Note that CURRENT_TIME may perform an unsafe call */
+ tso->par.exectime += CURRENT_TIME - tso->par.blockedat;
+ tso->par.fetchcount++;
+ tso->par.blockedat = CURRENT_TIME;
+ /* we are about to send off a FETCH message, so dump a FETCH event */
+ DumpRawGranEvent(CURRENT_PROC,
+ taskIDtoPE(remote_ga->payload.gc.gtid),
+ GR_FETCH, tso, tso->block_info.closure, 0, 0);
+ }
+ /* Phil T. claims that this was a workaround for a hard-to-find
+ * bug, hence I'm leaving it out for now --SDM
+ */
+ /* Assign a brand-new global address to the newly created FMBQ */
+ local_ga = makeGlobal(tso->block_info.closure, rtsFalse);
+ splitWeight(&fmbq_ga, local_ga);
+ ASSERT(fmbq_ga.weight == 1U << (BITS_IN(unsigned) - 1));
+
+ sendFetch(remote_ga, &fmbq_ga, 0/*load*/);
+
+ // Global statistics: count no. of fetches
+ if (RtsFlags.ParFlags.ParStats.Global &&
+ RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
+ globalParStats.tot_fetch_mess++;
+ }
+
+ IF_DEBUG(sanity,
+ theGlobalFromGA.payload.gc.gtid = (GlobalTaskId)0);
break;
- case PP_FISH:
- processFish();
+ case BlockedOnGA_NoSend:
+ /* the closure must be a FETCH_ME_BQ; tso came in here via
+ FETCH_ME_BQ entry code */
+ ASSERT(get_itbl(tso->block_info.closure)->type==FETCH_ME_BQ);
+
+ /* Fetch message has been sent already */
+ if (RtsFlags.ParFlags.ParStats.Full) {
+ /* Note that CURRENT_TIME may perform an unsafe call */
+ tso->par.exectime += CURRENT_TIME - tso->par.blockedat;
+ tso->par.blockcount++;
+ tso->par.blockedat = CURRENT_TIME;
+ /* dump a block event, because fetch has been sent already */
+ DumpRawGranEvent(CURRENT_PROC, thisPE,
+ GR_BLOCK, tso, tso->block_info.closure, 0, 0);
+ }
break;
- case PP_FREE:
- processFree();
+ case BlockedOnMVar:
+ case BlockedOnBlackHole:
+ /* the closure must be a BLACKHOLE_BQ or an RBH; tso came in here via
+ BLACKHOLE(_BQ) or CAF_BLACKHOLE or RBH entry code */
+ ASSERT(get_itbl(tso->block_info.closure)->type==MVAR ||
+ get_itbl(tso->block_info.closure)->type==BLACKHOLE_BQ ||
+ get_itbl(tso->block_info.closure)->type==RBH);
+
+ /* if collecting stats update the execution time etc */
+ if (RtsFlags.ParFlags.ParStats.Full) {
+ /* Note that CURRENT_TIME may perform an unsafe call */
+ tso->par.exectime += CURRENT_TIME - tso->par.blockedat;
+ tso->par.blockcount++;
+ tso->par.blockedat = CURRENT_TIME;
+ DumpRawGranEvent(CURRENT_PROC, thisPE,
+ GR_BLOCK, tso, tso->block_info.closure, 0, 0);
+ }
break;
-
- case PP_SCHEDULE:
- processSchedule(task);
+
+ case BlockedOnDelay:
+ /* Whats sort of stats shall we collect for an explicit threadDelay? */
+ IF_PAR_DEBUG(verbose,
+ belch("##++ blockThread: TSO %d blocked on ThreadDelay",
+ tso->id));
break;
+ /* Check that the following is impossible to happen, indeed
+ case BlockedOnException:
+ case BlockedOnRead:
+ case BlockedOnWrite:
+ */
default:
- /* Anything we're not prepared to deal with. */
- barf("Task %x: Unexpected opcode %x from %x",
- mytid, opcode, task);
- } /* switch */
-
- } while (PacketsWaiting()); /* While there are messages: process them */
-} /* processMessages */
+ barf("blockThread: impossible why_blocked code %d for TSO %d",
+ tso->why_blocked, tso->id);
+ }
-//@node Miscellaneous Functions, Index, GUM Message Processor, High Level Communications Routines
-//@subsection Miscellaneous Functions
+ IF_PAR_DEBUG(verbose,
+ belch("##++ blockThread: TSO %d blocked on closure %p (%s); %s",
+ tso->id, tso->block_info.closure, info_type(tso->block_info.closure),
+ (tso->why_blocked==BlockedOnGA) ? "Sent FETCH for GA" : ""));
+
+ IF_PAR_DEBUG(bq,
+ print_bq(tso->block_info.closure));
+}
/*
* ChoosePE selects a GlobalTaskId from the array of PEs 'at random'.
StgClosure *closure;
closure = GALAlookup(&ga);
- if ((bf = (StgBlockedFetch *)allocate(FIXED_HS + sizeofW(StgBlockedFetch))) == NULL) {
- GarbageCollect(GetRoots);
+ if ((bf = (StgBlockedFetch *)allocate(_HS + sizeofW(StgBlockedFetch))) == NULL) {
+ barf("createBlockedFetch: out of heap while allocating heap for a BlocekdFetch; ToDo: call GC here");
+ GarbageCollect(GetRoots, rtsFalse);
closure = GALAlookup(&ga);
- bf = (StgBlockedFetch *)allocate(FIXED_HS + sizeofW(StgBlockedFetch));
+ bf = (StgBlockedFetch *)allocate(_HS + sizeofW(StgBlockedFetch));
// ToDo: check whether really guaranteed to succeed 2nd time around
}
- ASSERT(bf != (StgClosure *)NULL);
- SET_INFO((StgClosure *)bf, &BLOCKED_FETCH_info);
+ ASSERT(bf != (StgBlockedFetch *)NULL);
+ SET_INFO((StgClosure *)bf, &stg_BLOCKED_FETCH_info);
// ToDo: check whether other header info is needed
bf->node = closure;
bf->ga.payload.gc.gtid = rga.payload.gc.gtid;
bf->ga.weight = rga.weight;
// bf->link = NULL; debugging
- IF_PAR_DEBUG(fetch,
- fprintf(stderr, "%% [%x] created BF: closure=%p (%s), GA: ",
- mytid, closure, info_type(closure));
+ IF_PAR_DEBUG(schedule,
+ fprintf(stderr, "%%%%// created BF: bf=%p (%s) of closure , GA: ",
+ bf, info_type((StgClosure*)bf));
printGA(&(bf->ga));
fputc('\n',stderr));
- return bf;
+ return (StgClosure *)bf;
}
/*
{
do {
rtsPacket p = GetPacket();
- processUnexpected(p);
+ processUnexpectedMessage(p);
} while (rtsTrue);
}
void
DebugPrintGAGAMap(globalAddr *gagamap, int nGAs)
{
- int i;
+ nat i;
for (i = 0; i < nGAs; ++i, gagamap += 2)
- fprintf(stderr, "gagamap[%d] = ((%x, %d, %x)) -> ((%x, %d, %x))\n", i,
+ fprintf(stderr, "__ gagamap[%d] = ((%x, %d, %x)) -> ((%x, %d, %x))\n", i,
gagamap[0].payload.gc.gtid, gagamap[0].payload.gc.slot, gagamap[0].weight,
gagamap[1].payload.gc.gtid, gagamap[1].payload.gc.slot, gagamap[1].weight);
}
+
+//@cindex checkGAGAMap
+void
+checkGAGAMap(globalAddr *gagamap, int nGAs)
+{
+ nat i;
+
+ for (i = 0; i < (nat)nGAs; ++i, gagamap += 2) {
+ ASSERT(looks_like_ga(gagamap));
+ ASSERT(looks_like_ga(gagamap+1));
+ }
+}
#endif
//@cindex freeMsgBuffer
static StgWord **freeMsgBuffer = NULL;
//@cindex freeMsgIndex
-static int *freeMsgIndex = NULL;
+static nat *freeMsgIndex = NULL;
//@cindex prepareFreeMsgBuffers
void
prepareFreeMsgBuffers(void)
{
- int i;
+ nat i;
/* Allocate the freeMsg buffers just once and then hang onto them. */
if (freeMsgIndex == NULL) {
- freeMsgIndex = (int *) stgMallocBytes(nPEs * sizeof(int),
+ freeMsgIndex = (nat *) stgMallocBytes(nPEs * sizeof(nat),
"prepareFreeMsgBuffers (Index)");
freeMsgBuffer = (StgWord **) stgMallocBytes(nPEs * sizeof(long *),
"prepareFreeMsgBuffers (Buffer)");
for(i = 0; i < nPEs; i++)
- if (i != thisPE)
+ if (i != (thisPE-1))
freeMsgBuffer[i] = (StgPtr) stgMallocWords(RtsFlags.ParFlags.packBufferSize,
"prepareFreeMsgBuffers (Buffer #i)");
+ else
+ freeMsgBuffer[i] = 0;
}
/* Initialize the freeMsg buffer pointers to point to the start of their
void
freeRemoteGA(int pe, globalAddr *ga)
{
- int i;
+ nat i;
ASSERT(GALAlookup(ga) == NULL);
if ((i = freeMsgIndex[pe]) + 2 >= RtsFlags.ParFlags.packBufferSize) {
IF_PAR_DEBUG(free,
- belch("Filled a free message buffer (sending remaining messages indivisually)"));
+ belch("!! Filled a free message buffer (sending remaining messages indivisually)"));
sendFree(ga->payload.gc.gtid, i, freeMsgBuffer[pe]);
i = 0;
freeMsgBuffer[pe][i++] = (StgWord) ga->payload.gc.slot;
freeMsgIndex[pe] = i;
-#ifdef DEBUG
- ga->weight = 0x0f0f0f0f;
- ga->payload.gc.gtid = 0x666;
- ga->payload.gc.slot = 0xdeaddead;
-#endif
+ IF_DEBUG(sanity,
+ ga->weight = 0xdead0add;
+ ga->payload.gc.gtid = 0xbbbbbbbb;
+ ga->payload.gc.slot = 0xbbbbbbbb;);
}
//@cindex sendFreeMessages
void
sendFreeMessages(void)
{
- int i;
+ nat i;
for (i = 0; i < nPEs; i++)
if (freeMsgIndex[i] > 0)
sendFree(allPEs[i], freeMsgIndex[i], freeMsgBuffer[i]);
}
+/* synchronises with the other PEs. Receives and records in a global
+ * variable the task-id of SysMan. If this is the main thread (discovered
+ * in main.lc), identifies itself to SysMan. Finally it receives
+ * from SysMan an array of the Global Task Ids of each PE, which is
+ * returned as the value of the function.
+ */
+
+#if defined(PAR_TICKY)
+/* Has to see freeMsgIndex, so must be defined here not in ParTicky.c */
+//@cindex stats_CntFreeGA
+void
+stats_CntFreeGA (void) { // stats only
+
+ // Global statistics: residency of thread and spark pool
+ if (RtsFlags.ParFlags.ParStats.Global &&
+ RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
+ nat i, s;
+
+ globalParStats.cnt_free_GA++;
+ for (i = 0, s = 0; i < nPEs; i++)
+ s += globalParStats.tot_free_GA += freeMsgIndex[i]/2;
+
+ if ( s > globalParStats.res_free_GA )
+ globalParStats.res_free_GA = s;
+ }
+}
+#endif /* PAR_TICKY */
+
#endif /* PAR -- whole file */
//@node Index, , Miscellaneous Functions, High Level Communications Routines
projects / ghc-hetmet.git / blobdiff