+++ /dev/null
-/* ----------------------------------------------------------------------------
- * 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
- *
- * 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 -
- *
- * ------------------------------------------------------------------------- */
-
-#ifdef PAR /* whole file */
-
-//@node High Level Communications Routines, , ,
-//@section High Level Communications Routines
-
-//@menu
-//* Macros etc::
-//* Includes::
-//* GUM Message Sending and Unpacking Functions::
-//* Message-Processing Functions::
-//* GUM Message Processor::
-//* Miscellaneous Functions::
-//* Index::
-//@end menu
-
-//@node Macros etc, Includes, High Level Communications Routines, High Level Communications Routines
-//@subsection Macros etc
-
-/* Evidently not Posix */
-/* #include "PosixSource.h" */
-
-//@node Includes, GUM Message Sending and Unpacking Functions, Macros etc, High Level Communications Routines
-//@subsection Includes
-
-#include "Rts.h"
-#include "RtsUtils.h"
-#include "RtsFlags.h"
-#include "Storage.h" // for recordMutable
-#include "HLC.h"
-#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
-
-/*
- * GUM Message Sending and Unpacking Functions
- */
-
-/*
- * Allocate space for message processing
- */
-
-//@cindex gumPackBuffer
-static rtsPackBuffer *gumPackBuffer;
-
-//@cindex initMoreBuffers
-rtsBool
-initMoreBuffers(void)
-{
- if ((gumPackBuffer = (rtsPackBuffer *)stgMallocWords(RtsFlags.ParFlags.packBufferSize,
- "initMoreBuffers")) == NULL)
- return rtsFalse;
- return rtsTrue;
-}
-
-/*
- * SendFetch packs the two global addresses and a load into a message +
- * sends it.
-
-//@cindex FETCH
-
- Structure of a FETCH message:
-
- | GA 1 | GA 2 |
- +------------------------------------+------+
- | gtid | slot | weight | gtid | slot | load |
- +------------------------------------+------+
- */
-
-//@cindex sendFetch
-void
-sendFetch(globalAddr *rga, globalAddr *lga, int load)
-{
- ASSERT(rga->weight > 0 && lga->weight > 0);
- IF_PAR_DEBUG(fetch,
- 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));
-
-
- /* ToDo: Dump event
- DumpRawGranEvent(CURRENT_PROC, taskIDtoPE(rga->payload.gc.gtid),
- GR_FETCH, CurrentTSO, (StgClosure *)(lga->payload.gc.slot),
- 0, spark_queue_len(ADVISORY_POOL));
- */
-
- sendOpV(PP_FETCH, rga->payload.gc.gtid, 6,
- (StgWord) rga->payload.gc.gtid, (StgWord) rga->payload.gc.slot,
- (StgWord) lga->weight, (StgWord) lga->payload.gc.gtid,
- (StgWord) lga->payload.gc.slot, (StgWord) load);
-}
-
-/*
- * unpackFetch unpacks a FETCH message into two Global addresses and a load
- * figure.
-*/
-
-//@cindex unpackFetch
-static void
-unpackFetch(globalAddr *lga, globalAddr *rga, int *load)
-{
- long buf[6];
-
- GetArgs(buf, 6);
-
- IF_PAR_DEBUG(fetch,
- 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]));
-
- lga->weight = 1;
- lga->payload.gc.gtid = (GlobalTaskId) buf[0];
- lga->payload.gc.slot = (int) buf[1];
-
- rga->weight = (unsigned) buf[2];
- rga->payload.gc.gtid = (GlobalTaskId) buf[3];
- rga->payload.gc.slot = (int) buf[4];
-
- *load = (int) buf[5];
-
- ASSERT(rga->weight > 0);
-}
-
-/*
- * SendResume packs the remote blocking queue's GA and data into a message
- * and sends it.
-
-//@cindex RESUME
-
- Structure of a RESUME message:
-
- -------------------------------
- | weight | slot | n | data ...
- -------------------------------
-
- data is a packed graph represented as an rtsPackBuffer
- n is the size of the graph (as returned by PackNearbyGraph) + packet hdr size
- */
-
-//@cindex sendResume
-void
-sendResume(globalAddr *rga, int nelem, rtsPackBuffer *packBuffer)
-{
- 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 + DEBUG_HEADROOM, (StgPtr)packBuffer,
- 2, (rtsWeight) rga->weight, (StgWord) rga->payload.gc.slot);
-}
-
-/*
- * unpackResume unpacks a Resume message into two Global addresses and
- * a data array.
- */
-
-//@cindex unpackResume
-static void
-unpackResume(globalAddr *lga, int *nelem, rtsPackBuffer *packBuffer)
-{
- long buf[3];
-
- GetArgs(buf, 3);
-
- /*
- RESUME event is written in awaken_blocked_queue
- DumpRawGranEvent(CURRENT_PROC, taskIDtoPE(lga->payload.gc.gtid),
- GR_RESUME, END_TSO_QUEUE, (StgClosure *)NULL, 0, 0);
- */
-
- lga->weight = (unsigned) buf[0];
- lga->payload.gc.gtid = mytid;
- lga->payload.gc.slot = (int) buf[1];
-
- *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));
-}
-
-/*
- * SendAck packs the global address being acknowledged, together with
- * an array of global addresses for any closures shipped and sends them.
-
-//@cindex ACK
-
- Structure of an ACK message:
-
- | GA 1 | GA 2 |
- +---------------------------------------------+-------
- | weight | gtid | slot | weight | gtid | slot | ..... ngas times
- + --------------------------------------------+-------
-
- */
-
-//@cindex sendAck
-void
-sendAck(GlobalTaskId task, int ngas, globalAddr *gagamap)
-{
- static long *buffer;
- 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) {
- ASSERT(gagamap[1].weight > 0);
- p[0] = (long) gagamap->weight;
- p[1] = (long) gagamap->payload.gc.gtid;
- p[2] = (long) gagamap->payload.gc.slot;
- gagamap++;
- p[3] = (long) gagamap->weight;
- p[4] = (long) gagamap->payload.gc.gtid;
- p[5] = (long) gagamap->payload.gc.slot;
- gagamap++;
- }
- IF_PAR_DEBUG(schedule,
- belch("~^,, Sending Ack (%d pairs) to [%x]\n",
- ngas, task));
-
- sendOpN(PP_ACK, task, p - buffer, (StgPtr)buffer);
-}
-
-/*
- * unpackAck unpacks an Acknowledgement message into a Global address,
- * a count of the number of global addresses following and a map of
- * Global addresses
- */
-
-//@cindex unpackAck
-static void
-unpackAck(int *ngas, globalAddr *gagamap)
-{
- long GAarraysize;
- long buf[6];
-
- GetArgs(&GAarraysize, 1);
-
- *ngas = GAarraysize / 6;
-
- IF_PAR_DEBUG(schedule,
- belch("~^,, Unpacking Ack (%d pairs) on [%x]\n",
- *ngas, mytid));
-
- while (GAarraysize > 0) {
- GetArgs(buf, 6);
- gagamap->weight = (rtsWeight) buf[0];
- gagamap->payload.gc.gtid = (GlobalTaskId) buf[1];
- gagamap->payload.gc.slot = (int) buf[2];
- gagamap++;
- gagamap->weight = (rtsWeight) buf[3];
- gagamap->payload.gc.gtid = (GlobalTaskId) buf[4];
- gagamap->payload.gc.slot = (int) buf[5];
- ASSERT(gagamap->weight > 0);
- gagamap++;
- GAarraysize -= 6;
- }
-}
-
-/*
- * SendFish packs the global address being acknowledged, together with
- * an array of global addresses for any closures shipped and sends them.
-
-//@cindex FISH
-
- Structure of a FISH message:
-
- +----------------------------------+
- | orig PE | age | history | hunger |
- +----------------------------------+
- */
-
-//@cindex sendFish
-void
-sendFish(GlobalTaskId destPE, GlobalTaskId origPE,
- int age, int history, int hunger)
-{
- IF_PAR_DEBUG(fish,
- belch("~^$$ Sending Fish to [%x] (%d outstanding fishes)",
- destPE, outstandingFishes));
-
- sendOpV(PP_FISH, destPE, 4,
- (StgWord) origPE, (StgWord) age, (StgWord) history, (StgWord) hunger);
-
- if (origPE == mytid) {
- //fishing = rtsTrue;
- outstandingFishes++;
- }
-}
-
-/*
- * unpackFish unpacks a FISH message into the global task id of the
- * originating PE and 3 data fields: the age, history and hunger of the
- * fish. The history + hunger are not currently used.
-
- */
-
-//@cindex unpackFish
-static void
-unpackFish(GlobalTaskId *origPE, int *age, int *history, int *hunger)
-{
- long buf[4];
-
- GetArgs(buf, 4);
-
- IF_PAR_DEBUG(fish,
- belch("~^$$ Unpacking Fish from [%x] (age=%d)",
- (GlobalTaskId) buf[0], (int) buf[1]));
-
- *origPE = (GlobalTaskId) buf[0];
- *age = (int) buf[1];
- *history = (int) buf[2];
- *hunger = (int) buf[3];
-}
-
-/*
- * SendFree sends (weight, slot) pairs for GAs that we no longer need
- * references to.
-
-//@cindex FREE
-
- Structure of a FREE message:
-
- +-----------------------------
- | n | weight_1 | slot_1 | ...
- +-----------------------------
- */
-//@cindex sendFree
-void
-sendFree(GlobalTaskId pe, int nelem, StgPtr data)
-{
- IF_PAR_DEBUG(free,
- belch("~^!! Sending Free (%d GAs) to [%x]",
- nelem/2, pe));
-
- sendOpN(PP_FREE, pe, nelem, data);
-}
-
-/*
- * unpackFree unpacks a FREE message into the amount of data shipped and
- * a data block.
- */
-//@cindex unpackFree
-static void
-unpackFree(int *nelem, StgWord *data)
-{
- long buf[1];
-
- GetArgs(buf, 1);
- *nelem = (int) buf[0];
-
- IF_PAR_DEBUG(free,
- belch("~^!! Unpacking Free (%d GAs)",
- *nelem/2));
-
- GetArgs(data, *nelem);
-}
-
-/*
- * SendSchedule sends a closure to be evaluated in response to a Fish
- * message. The message is directed to the PE that originated the Fish
- * (origPE), and includes the packed closure (data) along with its size
- * (nelem).
-
-//@cindex SCHEDULE
-
- Structure of a SCHEDULE message:
-
- +------------------------------------
- | PE | n | pack buffer of a graph ...
- +------------------------------------
- */
-//@cindex sendSchedule
-void
-sendSchedule(GlobalTaskId origPE, int nelem, rtsPackBuffer *packBuffer)
-{
- IF_PAR_DEBUG(schedule,
- 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 + DEBUG_HEADROOM, (StgPtr)packBuffer);
-}
-
-/*
- * unpackSchedule unpacks a SCHEDULE message into the Global address of
- * the closure shipped, the amount of data shipped (nelem) and the data
- * block (data).
- */
-
-//@cindex unpackSchedule
-static void
-unpackSchedule(int *nelem, rtsPackBuffer *packBuffer)
-{
- 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;
-
- /* automatic cast of flat pvm-data to rtsPackBuffer */
- GetArgs(packBuffer, *nelem + PACK_BUFFER_HDR_SIZE + DEBUG_HEADROOM);
-
- IF_PAR_DEBUG(schedule,
- belch("~^-- Unpacking Schedule (packet <<%d>> with %d elems) on [%x]\n",
- packBuffer->id, *nelem, mytid));
-
- 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
-
-/*
- * Message-Processing Functions
- *
- * The following routines process incoming GUM messages. Often reissuing
- * messages in response.
- *
- * processFish unpacks a fish message, reissuing it if it's our own,
- * sending work if we have it or sending it onwards otherwise.
- */
-
-/*
- * 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
- * BlockedFetches from a blocking queue to the PendingFetches queue.
- */
-void GetRoots(void);
-extern StgBlockedFetch *PendingFetches;
-
-nat
-pending_fetches_len(void)
-{
- StgBlockedFetch *bf;
- nat n;
-
- for (n=0, bf=PendingFetches; bf != END_BF_QUEUE; n++, bf = (StgBlockedFetch *)(bf->link)) {
- ASSERT(get_itbl(bf)->type==BLOCKED_FETCH);
- }
- return n;
-}
-
-//@cindex processFetches
-void
-processFetches(void) {
- StgBlockedFetch *bf, *next;
- StgClosure *closure;
- StgInfoTable *ip;
- globalAddr rga;
- static rtsPackBuffer *packBuffer;
-
- IF_PAR_DEBUG(verbose,
- belch("____ processFetches: %d pending fetches (root @ %p)",
- pending_fetches_len(), PendingFetches));
-
- for (bf = PendingFetches;
- bf != END_BF_QUEUE;
- 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
- * process it in much the same fashion as the original target
- * of the fetch. Though we hope to find graph here, we could
- * find a black hole (of any flavor) or even a FetchMe.
- */
- closure = bf->node;
- /*
- 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).
- */
- ASSERT(get_itbl(closure)->type != EVACUATED);
- // closure = ((StgEvacuated *)closure)->evacuee;
-
- closure = UNWIND_IND(closure);
- //while ((ind = IS_INDIRECTION(closure)) != NULL) { closure = ind; }
-
- ip = get_itbl(closure);
- if (ip->type == FETCH_ME) {
- /* Forward the Fetch to someone else */
- rga.payload.gc.gtid = bf->ga.payload.gc.gtid;
- rga.payload.gc.slot = bf->ga.payload.gc.slot;
- rga.weight = bf->ga.weight;
-
- sendFetch(((StgFetchMe *)closure)->ga, &rga, 0 /* load */);
-
- // 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 => doing a blockFetch on closure %p (%s)",
- closure, info_type(closure)));
- bf->node = closure;
- blockFetch(bf, closure);
- } else {
- /* We now have some local graph to send back */
- nat size;
-
- packBuffer = gumPackBuffer;
- IF_PAR_DEBUG(verbose,
- belch("__*> processFetches: PackNearbyGraph of closure %p (%s)",
- closure, info_type(closure)));
-
- 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!
- 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, bf->ga.payload.gc.gtid);
- ASSERT(packBuffer != (rtsPackBuffer *)NULL);
- }
- rga.payload.gc.gtid = bf->ga.payload.gc.gtid;
- rga.payload.gc.slot = bf->ga.payload.gc.slot;
- 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;
-}
-
-#if 0
-/*
- Alternatively to sending fetch messages directly from the FETCH_ME_entry
- code we could just store the data about the remote data in a global
- variable and send the fetch request from the main scheduling loop (similar
- to processFetches above). This would save an expensive STGCALL in the entry
- code because we have to go back to the scheduler anyway.
-*/
-//@cindex processFetches
-void
-processTheRealFetches(void) {
- StgBlockedFetch *bf;
- StgClosure *closure, *next;
-
- IF_PAR_DEBUG(verbose,
- belch("__ processTheRealFetches: ");
- printGA(&theGlobalFromGA);
- printGA(&theGlobalToGA));
-
- ASSERT(theGlobalFromGA.payload.gc.gtid != 0 &&
- theGlobalToGA.payload.gc.gtid != 0);
-
- /* the old version did this in the FETCH_ME entry code */
- sendFetch(&theGlobalFromGA, &theGlobalToGA, 0/*load*/);
-
-}
-#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.
- */
-//@cindex processFish
-static void
-processFish(void)
-{
- GlobalTaskId origPE;
- int age, history, hunger;
- rtsSpark spark;
- static rtsPackBuffer *packBuffer;
-
- 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
- }
-
- ASSERT(origPE != mytid);
- IF_PAR_DEBUG(fish,
- 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,origPE)) == NULL) {
- IF_PAR_DEBUG(fish,
- belch("$$ GC while trying to satisfy FISH via PackNearbyGraph of node %p",
- (StgClosure *)spark));
- 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("$$-- 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("$$^^ No sparks available for FISH from %x",
- origPE));
- /* We have no sparks to give */
- 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);
-
- // 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 */
-
-/*
- * processFetch either returns the requested data (if available)
- * or blocks the remote blocking queue on a black hole (if not).
- */
-
-//@cindex processFetch
-static void
-processFetch(void)
-{
- globalAddr ga, rga;
- int load;
- StgClosure *closure;
- StgInfoTable *ip;
-
- unpackFetch(&ga, &rga, &load);
- IF_PAR_DEBUG(fetch,
- 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));
-
- closure = GALAlookup(&ga);
- ASSERT(closure != (StgClosure *)NULL);
- ip = get_itbl(closure);
- 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("%%%%== 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);
- (void) addWeight(&rga);
- } else if (IS_BLACK_HOLE(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);
- IF_PAR_DEBUG(fetch,
- belch("%%++ Blocking Fetch ((%x, %d, %x)) on %p (%s)",
- rga.payload.gc.gtid, rga.payload.gc.slot, rga.weight,
- closure, info_type(closure)));
- 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"));
- }
-}
-
-/*
- * processFree unpacks a FREE message and adds the weights to our GAs.
- */
-//@cindex processFree
-static void
-processFree(void)
-{
- int nelem;
- static StgWord *buffer;
- int i;
- globalAddr ga;
-
- buffer = (StgWord *)gumPackBuffer;
- unpackFree(&nelem, buffer);
- IF_PAR_DEBUG(free,
- 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, "!!-- Processing free ");
- printGA(&ga);
- fputc('\n', stderr);
- );
- (void) addWeight(&ga);
- }
-}
-
-/*
- * processResume unpacks a RESUME message into the graph, filling in
- * the LA -> GA, and GA -> LA tables. Threads blocked on the original
- * FetchMe (now a blocking queue) are awakened, and the blocking queue
- * is converted into an indirection. Finally it sends an ACK in response
- * which contains any newly allocated GAs.
- */
-
-//@cindex processResume
-static void
-processResume(GlobalTaskId sender)
-{
- int nelem;
- nat nGAs;
- static rtsPackBuffer *packBuffer;
- StgClosure *newGraph, *old;
- globalAddr lga;
- globalAddr *gagamap;
-
- packBuffer = (rtsPackBuffer *)gumPackBuffer;
- unpackResume(&lga, &nelem, packBuffer);
-
- IF_PAR_DEBUG(fetch,
- fprintf(stderr, "[]__ Rcvd Resume for ");
- printGA(&lga);
- fputc('\n', stderr));
- IF_PAR_DEBUG(packet,
- PrintPacket((rtsPackBuffer *)packBuffer));
-
- /*
- * We always unpack the incoming graph, even if we've received the
- * requested node in some other data packet (and already awakened
- * the blocking queue).
- if (SAVE_Hp + packBuffer[0] >= SAVE_HpLim) {
- ReallyPerformThreadGC(packBuffer[0], rtsFalse);
- SAVE_Hp -= packBuffer[0];
- }
- */
-
- // ToDo: Check for GC here !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
-
- /* Do this *after* GC; we don't want to release the object early! */
-
- if (lga.weight > 0)
- (void) addWeight(&lga);
-
- 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) {
- 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
- 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);
- ASSERT(newGraph != NULL);
-
- /*
- * Sometimes, unpacking will common up the resumee with the
- * incoming graph, but if it hasn't, we'd better do so now.
- */
-
- if (get_itbl(old)->type == FETCH_ME_BQ)
- CommonUp(old, newGraph);
-
- 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);
-}
-
-/*
- * processSchedule unpacks a SCHEDULE message into the graph, filling
- * in the LA -> GA, and GA -> LA tables. The root of the graph is added to
- * the local spark queue. Finally it sends an ACK in response
- * which contains any newly allocated GAs.
- */
-//@cindex processSchedule
-static void
-processSchedule(GlobalTaskId sender)
-{
- nat nelem, nGAs;
- rtsBool success;
- static rtsPackBuffer *packBuffer;
- StgClosure *newGraph;
- globalAddr *gagamap;
-
- packBuffer = gumPackBuffer; /* HWL */
- unpackSchedule(&nelem, packBuffer);
-
- IF_PAR_DEBUG(schedule,
- belch("--__ Rcvd Schedule (%d elems)", nelem));
- IF_PAR_DEBUG(packet,
- PrintPacket(packBuffer));
-
- /*
- * For now, the graph is a closure to be sparked as an advisory
- * spark, but in future it may be a complete spark with
- * required/advisory status, priority etc.
- */
-
- /*
- 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);
- success = add_to_spark_queue(newGraph, &(MainRegTable.rSparks));
-
- 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 (%s); %d sparks available on [%x] (%s)",
- newGraph, info_type(newGraph), spark_queue_len(&(MainRegTable.rSparks)), mytid);
- else
- 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(tables,
- DebugPrintGAGAMap(gagamap, nGAs));
-
- sendAck(sender, nGAs, gagamap);
-
- //fishing = rtsFalse;
- ASSERT(outstandingFishes>0);
- outstandingFishes--;
-}
-
-/*
- * processAck unpacks an ACK, and uses the GAGA map to convert RBH's
- * (which represent shared thunks that have been shipped) into fetch-mes
- * to remote GAs.
- */
-//@cindex processAck
-static void
-processAck(void)
-{
- nat nGAs;
- globalAddr *gaga;
- globalAddr gagamap[256]; // ToDo: elim magic constant!! MAX_GAS * 2];??
-
- unpackAck(&nGAs, gagamap);
-
- 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
- * the weight from the oldGA.
- */
- for (gaga = gagamap; gaga < gagamap + nGAs * 2; gaga += 2) {
- StgClosure *old_closure = GALAlookup(gaga);
- StgClosure *new_closure = GALAlookup(gaga + 1);
-
- ASSERT(old_closure != NULL);
- if (new_closure == NULL) {
- /* We don't have this closure, so we make a fetchme for it */
- globalAddr *ga = setRemoteGA(old_closure, gaga + 1, rtsTrue);
-
- /* convertToFetchMe should be done unconditionally here.
- Currently, we assign GAs to CONSTRs, too, (a bit of a hack),
- so we have to check whether it is an RBH before converting
-
- ASSERT(get_itbl(old_closure)==RBH);
- */
- if (get_itbl(old_closure)->type==RBH)
- convertToFetchMe((StgRBH *)old_closure, ga);
- } else {
- /*
- * Oops...we've got this one already; update the RBH to
- * point to the object we already know about, whatever it
- * happens to be.
- */
- CommonUp(old_closure, new_closure);
-
- /*
- * Increase the weight of the object by the amount just
- * received in the second part of the ACK pair.
- */
- (void) addWeight(gaga + 1);
- }
- (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
-
-/*
- * GUM Message Processor
-
- * processMessages processes any messages that have arrived, calling
- * appropriate routines depending on the message tag
- * (opcode). N.B. Unless profiling it assumes that there {\em ARE} messages
- * present and performs a blocking receive! During profiling it
- * busy-waits in order to record idle time.
- */
-
-//@cindex processMessages
-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);
-
- 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 */
-
- } while (PacketsWaiting()); /* While there are messages: process them */
- return receivedFinish;
-} /* processMessages */
-
-//@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;
-
- // 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 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 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 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:
- barf("blockThread: impossible why_blocked code %d for TSO %d",
- tso->why_blocked, tso->id);
- }
-
- 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'.
- * Important properties:
- * - it varies during execution, even if the PE is idle
- * - it's different for each PE
- * - we never send a fish to ourselves
- */
-extern long lrand48 (void);
-
-//@cindex choosePE
-GlobalTaskId
-choosePE(void)
-{
- long temp;
-
- temp = lrand48() % nPEs;
- if (allPEs[temp] == mytid) { /* Never send a FISH to yourself */
- temp = (temp + 1) % nPEs;
- }
- return allPEs[temp];
-}
-
-/*
- * allocate a BLOCKED_FETCH closure and fill it with the relevant fields
- * of the ga argument; called from processFetch when the local closure is
- * under evaluation
- */
-//@cindex createBlockedFetch
-StgClosure *
-createBlockedFetch (globalAddr ga, globalAddr rga)
-{
- StgBlockedFetch *bf;
- StgClosure *closure;
-
- closure = GALAlookup(&ga);
- 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(_HS + sizeofW(StgBlockedFetch));
- // ToDo: check whether really guaranteed to succeed 2nd time around
- }
-
- 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.payload.gc.slot = rga.payload.gc.slot;
- bf->ga.weight = rga.weight;
- // bf->link = NULL; debugging
-
- 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 (StgClosure *)bf;
-}
-
-/*
- * waitForTermination enters a loop ignoring spurious messages while
- * waiting for the termination sequence to be completed.
- */
-//@cindex waitForTermination
-void
-waitForTermination(void)
-{
- do {
- rtsPacket p = GetPacket();
- processUnexpectedMessage(p);
- } while (rtsTrue);
-}
-
-#ifdef DEBUG
-//@cindex DebugPrintGAGAMap
-void
-DebugPrintGAGAMap(globalAddr *gagamap, int nGAs)
-{
- nat i;
-
- for (i = 0; i < nGAs; ++i, gagamap += 2)
- 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 nat *freeMsgIndex = NULL;
-
-//@cindex prepareFreeMsgBuffers
-void
-prepareFreeMsgBuffers(void)
-{
- nat i;
-
- /* Allocate the freeMsg buffers just once and then hang onto them. */
- if (freeMsgIndex == NULL) {
- 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-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
- buffers */
- for (i = 0; i < nPEs; i++)
- freeMsgIndex[i] = 0;
-}
-
-//@cindex freeRemoteGA
-void
-freeRemoteGA(int pe, globalAddr *ga)
-{
- 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)"));
-
- sendFree(ga->payload.gc.gtid, i, freeMsgBuffer[pe]);
- i = 0;
- }
- freeMsgBuffer[pe][i++] = (StgWord) ga->weight;
- freeMsgBuffer[pe][i++] = (StgWord) ga->payload.gc.slot;
- freeMsgIndex[pe] = i;
-
- IF_DEBUG(sanity,
- ga->weight = 0xdead0add;
- ga->payload.gc.gtid = 0xbbbbbbbb;
- ga->payload.gc.slot = 0xbbbbbbbb;);
-}
-
-//@cindex sendFreeMessages
-void
-sendFreeMessages(void)
-{
- 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
-//@subsection Index
-
-//@index
-//* ACK:: @cindex\s-+ACK
-//* DebugPrintGAGAMap:: @cindex\s-+DebugPrintGAGAMap
-//* FETCH:: @cindex\s-+FETCH
-//* FISH:: @cindex\s-+FISH
-//* FREE:: @cindex\s-+FREE
-//* RESUME:: @cindex\s-+RESUME
-//* SCHEDULE:: @cindex\s-+SCHEDULE
-//* blockFetch:: @cindex\s-+blockFetch
-//* choosePE:: @cindex\s-+choosePE
-//* freeMsgBuffer:: @cindex\s-+freeMsgBuffer
-//* freeMsgIndex:: @cindex\s-+freeMsgIndex
-//* freeRemoteGA:: @cindex\s-+freeRemoteGA
-//* gumPackBuffer:: @cindex\s-+gumPackBuffer
-//* initMoreBuffers:: @cindex\s-+initMoreBuffers
-//* prepareFreeMsgBuffers:: @cindex\s-+prepareFreeMsgBuffers
-//* processAck:: @cindex\s-+processAck
-//* processFetch:: @cindex\s-+processFetch
-//* processFetches:: @cindex\s-+processFetches
-//* processFish:: @cindex\s-+processFish
-//* processFree:: @cindex\s-+processFree
-//* processMessages:: @cindex\s-+processMessages
-//* processResume:: @cindex\s-+processResume
-//* processSchedule:: @cindex\s-+processSchedule
-//* sendAck:: @cindex\s-+sendAck
-//* sendFetch:: @cindex\s-+sendFetch
-//* sendFish:: @cindex\s-+sendFish
-//* sendFree:: @cindex\s-+sendFree
-//* sendFreeMessages:: @cindex\s-+sendFreeMessages
-//* sendResume:: @cindex\s-+sendResume
-//* sendSchedule:: @cindex\s-+sendSchedule
-//* unpackAck:: @cindex\s-+unpackAck
-//* unpackFetch:: @cindex\s-+unpackFetch
-//* unpackFish:: @cindex\s-+unpackFish
-//* unpackFree:: @cindex\s-+unpackFree
-//* unpackResume:: @cindex\s-+unpackResume
-//* unpackSchedule:: @cindex\s-+unpackSchedule
-//* waitForTermination:: @cindex\s-+waitForTermination
-//@end index
projects / ghc-hetmet.git / blobdiff