/* ---------------------------------------------------------------------------- * Time-stamp: * * 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