Document UniqSupply
[ghc-hetmet.git] / rts / parallel / HLComms.c
1 /* ----------------------------------------------------------------------------
2  * Time-stamp: <Wed Mar 21 2001 16:34:41 Stardate: [-30]6363.45 hwloidl>
3  *
4  * High Level Communications Routines (HLComms.lc)
5  *
6  * Contains the high-level routines (i.e. communication
7  * subsystem independent) used by GUM
8  * 
9  * GUM 0.2x: Phil Trinder, Glasgow University, 12 December 1994
10  * GUM 3.xx: Phil Trinder, Simon Marlow July 1998
11  * GUM 4.xx: H-W. Loidl, Heriot-Watt University, November 1999 -
12  * 
13  * ------------------------------------------------------------------------- */
14
15 #ifdef PAR /* whole file */
16
17 //@node High Level Communications Routines, , ,
18 //@section High Level Communications Routines
19
20 //@menu
21 //* Macros etc::                
22 //* Includes::                  
23 //* GUM Message Sending and Unpacking Functions::  
24 //* Message-Processing Functions::  
25 //* GUM Message Processor::     
26 //* Miscellaneous Functions::   
27 //* Index::                     
28 //@end menu
29
30 //@node Macros etc, Includes, High Level Communications Routines, High Level Communications Routines
31 //@subsection Macros etc
32
33 /* Evidently not Posix */
34 /* #include "PosixSource.h" */
35
36 //@node Includes, GUM Message Sending and Unpacking Functions, Macros etc, High Level Communications Routines
37 //@subsection Includes
38
39 #include "Rts.h"
40 #include "RtsUtils.h"
41 #include "RtsFlags.h"
42 #include "Storage.h"   // for recordMutable
43 #include "HLC.h"
44 #include "Parallel.h"
45 #include "GranSimRts.h"
46 #include "ParallelRts.h"
47 #include "Sparks.h"
48 #include "FetchMe.h"     // for BLOCKED_FETCH_info etc
49 #if defined(DEBUG)
50 # include "ParallelDebug.h"
51 #endif
52 #include "StgMacros.h" // inlined IS_... fcts
53
54 #ifdef DIST
55 #include "SchedAPI.h" //for createIOThread
56 extern unsigned int context_switch; 
57 #endif /* DIST */
58
59 //@node GUM Message Sending and Unpacking Functions, Message-Processing Functions, Includes, High Level Communications Routines
60 //@subsection GUM Message Sending and Unpacking Functions
61
62 /*
63  * GUM Message Sending and Unpacking Functions
64  */
65
66 /*
67  * Allocate space for message processing
68  */
69
70 //@cindex gumPackBuffer
71 static rtsPackBuffer *gumPackBuffer;
72
73 //@cindex initMoreBuffers
74 rtsBool
75 initMoreBuffers(void)
76 {
77   if ((gumPackBuffer = (rtsPackBuffer *)stgMallocWords(RtsFlags.ParFlags.packBufferSize, 
78                                              "initMoreBuffers")) == NULL)
79     return rtsFalse;
80   return rtsTrue;
81 }
82
83 /*
84  * SendFetch packs the two global addresses and a load into a message +
85  * sends it.  
86
87 //@cindex FETCH
88
89    Structure of a FETCH message:
90
91          |    GA 1     |        GA 2          |
92          +------------------------------------+------+
93          | gtid | slot | weight | gtid | slot | load |
94          +------------------------------------+------+
95  */
96
97 //@cindex sendFetch
98 void
99 sendFetch(globalAddr *rga, globalAddr *lga, int load)
100 {
101   ASSERT(rga->weight > 0 && lga->weight > 0);
102   IF_PAR_DEBUG(fetch,
103                belch("~^** Sending Fetch for ((%x, %d, 0)); locally ((%x, %d, %x)), load = %d", 
104                      rga->payload.gc.gtid, rga->payload.gc.slot, 
105                      lga->payload.gc.gtid, lga->payload.gc.slot, lga->weight,
106                      load));
107
108
109   /* ToDo: Dump event
110   DumpRawGranEvent(CURRENT_PROC, taskIDtoPE(rga->payload.gc.gtid), 
111                    GR_FETCH, CurrentTSO, (StgClosure *)(lga->payload.gc.slot),
112                    0, spark_queue_len(ADVISORY_POOL));
113   */
114
115   sendOpV(PP_FETCH, rga->payload.gc.gtid, 6,
116           (StgWord) rga->payload.gc.gtid, (StgWord) rga->payload.gc.slot, 
117           (StgWord) lga->weight, (StgWord) lga->payload.gc.gtid, 
118           (StgWord) lga->payload.gc.slot, (StgWord) load);
119 }
120
121 /*
122  * unpackFetch unpacks a FETCH message into two Global addresses and a load
123  * figure.  
124 */
125
126 //@cindex unpackFetch
127 static void
128 unpackFetch(globalAddr *lga, globalAddr *rga, int *load)
129 {
130   long buf[6];
131
132   GetArgs(buf, 6); 
133
134   IF_PAR_DEBUG(fetch,
135                belch("~^** Unpacking Fetch for ((%x, %d, 0)) to ((%x, %d, %x)), load = %d", 
136                      (GlobalTaskId) buf[0], (int) buf[1], 
137                      (GlobalTaskId) buf[3], (int) buf[4], buf[2], buf[5]));
138
139   lga->weight = 1;
140   lga->payload.gc.gtid = (GlobalTaskId) buf[0];
141   lga->payload.gc.slot = (int) buf[1];
142
143   rga->weight = (unsigned) buf[2];
144   rga->payload.gc.gtid = (GlobalTaskId) buf[3];
145   rga->payload.gc.slot = (int) buf[4];
146
147   *load = (int) buf[5];
148
149   ASSERT(rga->weight > 0);
150 }
151
152 /*
153  * SendResume packs the remote blocking queue's GA and data into a message 
154  * and sends it.
155
156 //@cindex RESUME
157
158    Structure of a RESUME message:
159
160       -------------------------------
161       | weight | slot | n | data ...
162       -------------------------------
163
164    data is a packed graph represented as an rtsPackBuffer
165    n is the size of the graph (as returned by PackNearbyGraph) + packet hdr size
166  */
167
168 //@cindex sendResume
169 void
170 sendResume(globalAddr *rga, int nelem, rtsPackBuffer *packBuffer)
171 {
172   IF_PAR_DEBUG(fetch,
173                belch("~^[] Sending Resume (packet <<%d>> with %d elems) for ((%x, %d, %x)) to [%x]", 
174                      packBuffer->id, nelem,
175                      rga->payload.gc.gtid, rga->payload.gc.slot, rga->weight,
176                      rga->payload.gc.gtid));
177   IF_PAR_DEBUG(packet,
178                PrintPacket(packBuffer));
179
180   ASSERT(nelem==packBuffer->size);
181   /* check for magic end-of-buffer word */
182   IF_DEBUG(sanity, ASSERT(*(packBuffer->buffer+nelem) == END_OF_BUFFER_MARKER));
183
184   sendOpNV(PP_RESUME, rga->payload.gc.gtid, 
185            nelem + PACK_BUFFER_HDR_SIZE + DEBUG_HEADROOM, (StgPtr)packBuffer, 
186            2, (rtsWeight) rga->weight, (StgWord) rga->payload.gc.slot);
187 }
188
189 /*
190  * unpackResume unpacks a Resume message into two Global addresses and
191  * a data array.
192  */
193
194 //@cindex unpackResume
195 static void
196 unpackResume(globalAddr *lga, int *nelem, rtsPackBuffer *packBuffer)
197 {
198     long buf[3];
199
200     GetArgs(buf, 3); 
201
202     /*
203       RESUME event is written in awaken_blocked_queue
204     DumpRawGranEvent(CURRENT_PROC, taskIDtoPE(lga->payload.gc.gtid), 
205                      GR_RESUME, END_TSO_QUEUE, (StgClosure *)NULL, 0, 0);
206     */
207
208     lga->weight = (unsigned) buf[0];
209     lga->payload.gc.gtid = mytid;
210     lga->payload.gc.slot = (int) buf[1];
211
212     *nelem = (int) buf[2] - PACK_BUFFER_HDR_SIZE - DEBUG_HEADROOM;
213     GetArgs(packBuffer, *nelem + PACK_BUFFER_HDR_SIZE + DEBUG_HEADROOM);
214
215     IF_PAR_DEBUG(fetch,
216                  belch("~^[] Unpacking Resume (packet <<%d>> with %d elems) for ((%x, %d, %x))", 
217                        packBuffer->id, *nelem, mytid, (int) buf[1], (unsigned) buf[0]));
218
219     /* check for magic end-of-buffer word */
220     IF_DEBUG(sanity, ASSERT(*(packBuffer->buffer+*nelem) == END_OF_BUFFER_MARKER));
221 }
222
223 /*
224  * SendAck packs the global address being acknowledged, together with
225  * an array of global addresses for any closures shipped and sends them.
226
227 //@cindex ACK
228
229    Structure of an ACK message:
230
231       |        GA 1          |        GA 2          | 
232       +---------------------------------------------+-------
233       | weight | gtid | slot | weight | gtid | slot |  .....  ngas times
234       + --------------------------------------------+------- 
235
236  */
237
238 //@cindex sendAck
239 void
240 sendAck(GlobalTaskId task, int ngas, globalAddr *gagamap)
241 {
242   static long *buffer;
243   long *p;
244   int i;
245
246   if(ngas==0)
247     return; //don't send unnecessary messages!!
248   
249   buffer = (long *) gumPackBuffer;
250
251   for(i = 0, p = buffer; i < ngas; i++, p += 6) {
252     ASSERT(gagamap[1].weight > 0);
253     p[0] = (long) gagamap->weight;
254     p[1] = (long) gagamap->payload.gc.gtid;
255     p[2] = (long) gagamap->payload.gc.slot;
256     gagamap++;
257     p[3] = (long) gagamap->weight;
258     p[4] = (long) gagamap->payload.gc.gtid;
259     p[5] = (long) gagamap->payload.gc.slot;
260     gagamap++;
261   }
262   IF_PAR_DEBUG(schedule,
263                belch("~^,, Sending Ack (%d pairs) to [%x]\n", 
264                      ngas, task));
265
266   sendOpN(PP_ACK, task, p - buffer, (StgPtr)buffer);
267 }
268
269 /*
270  * unpackAck unpacks an Acknowledgement message into a Global address,
271  * a count of the number of global addresses following and a map of 
272  * Global addresses
273  */
274
275 //@cindex unpackAck
276 static void
277 unpackAck(int *ngas, globalAddr *gagamap)
278 {
279   long GAarraysize;
280   long buf[6];
281   
282   GetArgs(&GAarraysize, 1);
283   
284   *ngas = GAarraysize / 6;
285   
286   IF_PAR_DEBUG(schedule,
287                belch("~^,, Unpacking Ack (%d pairs) on [%x]\n", 
288                      *ngas, mytid));
289
290   while (GAarraysize > 0) {
291     GetArgs(buf, 6);
292     gagamap->weight = (rtsWeight) buf[0];
293     gagamap->payload.gc.gtid = (GlobalTaskId) buf[1];
294     gagamap->payload.gc.slot = (int) buf[2];
295     gagamap++;
296     gagamap->weight = (rtsWeight) buf[3];
297     gagamap->payload.gc.gtid = (GlobalTaskId) buf[4];
298     gagamap->payload.gc.slot = (int) buf[5];
299     ASSERT(gagamap->weight > 0);
300     gagamap++;
301     GAarraysize -= 6;
302   }
303 }
304
305 /*
306  * SendFish packs the global address being acknowledged, together with
307  * an array of global addresses for any closures shipped and sends them.
308
309 //@cindex FISH
310
311  Structure of a FISH message:
312
313      +----------------------------------+
314      | orig PE | age | history | hunger |
315      +----------------------------------+
316  */
317
318 //@cindex sendFish
319 void
320 sendFish(GlobalTaskId destPE, GlobalTaskId origPE, 
321          int age, int history, int hunger)
322 {
323   IF_PAR_DEBUG(fish,
324                belch("~^$$ Sending Fish to [%x] (%d outstanding fishes)", 
325                      destPE, outstandingFishes));
326
327   sendOpV(PP_FISH, destPE, 4, 
328           (StgWord) origPE, (StgWord) age, (StgWord) history, (StgWord) hunger);
329
330   if (origPE == mytid) {
331     //fishing = rtsTrue;
332     outstandingFishes++;
333   }
334 }
335
336 /*
337  * unpackFish unpacks a FISH message into the global task id of the
338  * originating PE and 3 data fields: the age, history and hunger of the
339  * fish. The history + hunger are not currently used.
340
341  */
342
343 //@cindex unpackFish
344 static void
345 unpackFish(GlobalTaskId *origPE, int *age, int *history, int *hunger)
346 {
347   long buf[4];
348   
349   GetArgs(buf, 4);
350   
351   IF_PAR_DEBUG(fish,
352                belch("~^$$ Unpacking Fish from [%x] (age=%d)", 
353                      (GlobalTaskId) buf[0], (int) buf[1]));
354
355   *origPE = (GlobalTaskId) buf[0];
356   *age = (int) buf[1];
357   *history = (int) buf[2];
358   *hunger = (int) buf[3];
359 }
360
361 /*
362  * SendFree sends (weight, slot) pairs for GAs that we no longer need
363  * references to.  
364
365 //@cindex FREE
366
367    Structure of a FREE message:
368    
369        +-----------------------------
370        | n | weight_1 | slot_1 | ...
371        +-----------------------------
372  */
373 //@cindex sendFree
374 void
375 sendFree(GlobalTaskId pe, int nelem, StgPtr data)
376 {
377     IF_PAR_DEBUG(free,
378                  belch("~^!! Sending Free (%d GAs) to [%x]", 
379                        nelem/2, pe));
380
381     sendOpN(PP_FREE, pe, nelem, data);
382 }
383
384 /*
385  * unpackFree unpacks a FREE message into the amount of data shipped and
386  * a data block.
387  */
388 //@cindex unpackFree
389 static void
390 unpackFree(int *nelem, StgWord *data)
391 {
392   long buf[1];
393   
394   GetArgs(buf, 1);
395   *nelem = (int) buf[0];
396
397   IF_PAR_DEBUG(free,
398                belch("~^!! Unpacking Free (%d GAs)", 
399                      *nelem/2));
400
401   GetArgs(data, *nelem);
402 }
403
404 /*
405  * SendSchedule sends a closure to be evaluated in response to a Fish
406  * message. The message is directed to the PE that originated the Fish
407  * (origPE), and includes the packed closure (data) along with its size
408  * (nelem).
409
410 //@cindex SCHEDULE
411
412    Structure of a SCHEDULE message:
413
414        +------------------------------------
415        | PE | n | pack buffer of a graph ...
416        +------------------------------------
417  */
418 //@cindex sendSchedule
419 void
420 sendSchedule(GlobalTaskId origPE, int nelem, rtsPackBuffer *packBuffer) 
421 {
422   IF_PAR_DEBUG(schedule,
423                belch("~^-- Sending Schedule (packet <<%d>> with %d elems) to [%x]\n", 
424                      packBuffer->id, nelem, origPE));
425   IF_PAR_DEBUG(packet,
426                PrintPacket(packBuffer));
427
428   ASSERT(nelem==packBuffer->size);
429   /* check for magic end-of-buffer word */
430   IF_DEBUG(sanity, ASSERT(*(packBuffer->buffer+nelem) == END_OF_BUFFER_MARKER));
431
432   sendOpN(PP_SCHEDULE, origPE, 
433           nelem + PACK_BUFFER_HDR_SIZE + DEBUG_HEADROOM, (StgPtr)packBuffer);
434 }
435
436 /*
437  * unpackSchedule unpacks a SCHEDULE message into the Global address of
438  * the closure shipped, the amount of data shipped (nelem) and the data
439  * block (data).
440  */
441
442 //@cindex unpackSchedule
443 static void
444 unpackSchedule(int *nelem, rtsPackBuffer *packBuffer)
445 {
446   long buf[1];
447
448   /* first, just unpack 1 word containing the total size (including header) */
449   GetArgs(buf, 1);
450   /* no. of elems, not counting the header of the pack buffer */
451   *nelem = (int) buf[0] - PACK_BUFFER_HDR_SIZE - DEBUG_HEADROOM;
452
453   /* automatic cast of flat pvm-data to rtsPackBuffer */
454   GetArgs(packBuffer, *nelem + PACK_BUFFER_HDR_SIZE + DEBUG_HEADROOM);
455
456   IF_PAR_DEBUG(schedule,
457                belch("~^-- Unpacking Schedule (packet <<%d>> with %d elems) on [%x]\n", 
458                      packBuffer->id, *nelem, mytid));
459
460   ASSERT(*nelem==packBuffer->size);
461   /* check for magic end-of-buffer word */
462   IF_DEBUG(sanity, ASSERT(*(packBuffer->buffer+*nelem) == END_OF_BUFFER_MARKER));
463 }
464
465 #ifdef DIST
466 /* sendReval is almost identical to the Schedule version, so we can unpack with unpackSchedule */
467 void
468 sendReval(GlobalTaskId origPE, int nelem, rtsPackBuffer *packBuffer) 
469 {  
470   IF_PAR_DEBUG(schedule,
471                belch("~^-- Sending Reval (packet <<%d>> with %d elems) to [%x]\n", 
472                      packBuffer->id, nelem, origPE));
473   IF_PAR_DEBUG(packet,
474                PrintPacket(packBuffer));
475
476   ASSERT(nelem==packBuffer->size);
477   /* check for magic end-of-buffer word */
478   IF_DEBUG(sanity, ASSERT(*(packBuffer->buffer+nelem) == END_OF_BUFFER_MARKER));
479
480   sendOpN(PP_REVAL, origPE, 
481           nelem + PACK_BUFFER_HDR_SIZE + DEBUG_HEADROOM, (StgPtr)packBuffer);
482 }
483
484 void FinishReval(StgTSO *t)
485 { StgClosure *res;
486   globalAddr ga;
487   nat size;
488   rtsPackBuffer *buffer=NULL;
489   
490   ga.payload.gc.slot = t->revalSlot;
491   ga.payload.gc.gtid = t->revalTid;
492   ga.weight = 0; 
493   
494   //find where the reval result is
495   res = GALAlookup(&ga);
496   ASSERT(res);
497   
498   IF_PAR_DEBUG(schedule,
499     printGA(&ga);
500     belch(" needs the result %08x\n",res));       
501   
502   //send off the result
503   buffer = PackNearbyGraph(res, END_TSO_QUEUE, &size,ga.payload.gc.gtid);
504   ASSERT(buffer != (rtsPackBuffer *)NULL);
505   sendResume(&ga, size, buffer);
506
507   IF_PAR_DEBUG(schedule,
508     belch("@;~) Reval Finished"));
509 }
510
511 #endif /* DIST */
512
513 //@node Message-Processing Functions, GUM Message Processor, GUM Message Sending and Unpacking Functions, High Level Communications Routines
514 //@subsection Message-Processing Functions
515
516 /*
517  * Message-Processing Functions
518  *
519  * The following routines process incoming GUM messages. Often reissuing
520  * messages in response.
521  *
522  * processFish unpacks a fish message, reissuing it if it's our own,
523  * sending work if we have it or sending it onwards otherwise.
524  */
525
526 /*
527  * processFetches constructs and sends resume messages for every
528  * BlockedFetch which is ready to be awakened.
529  * awaken_blocked_queue (in Schedule.c) is responsible for moving 
530  * BlockedFetches from a blocking queue to the PendingFetches queue.
531  */
532 void GetRoots(void);
533 extern StgBlockedFetch *PendingFetches;
534
535 nat
536 pending_fetches_len(void)
537 {
538   StgBlockedFetch *bf;
539   nat n;
540
541   for (n=0, bf=PendingFetches; bf != END_BF_QUEUE; n++, bf = (StgBlockedFetch *)(bf->link)) {
542     ASSERT(get_itbl(bf)->type==BLOCKED_FETCH);
543   }
544   return n;
545 }
546
547 //@cindex processFetches
548 void
549 processFetches(void) {
550   StgBlockedFetch *bf, *next;
551   StgClosure *closure;
552   StgInfoTable *ip;
553   globalAddr rga;
554   static rtsPackBuffer *packBuffer;
555     
556   IF_PAR_DEBUG(verbose,
557                belch("____ processFetches: %d pending fetches (root @ %p)",
558                      pending_fetches_len(), PendingFetches));
559   
560   for (bf = PendingFetches; 
561        bf != END_BF_QUEUE;
562        bf=next) {
563     /* the PendingFetches list contains only BLOCKED_FETCH closures */
564     ASSERT(get_itbl(bf)->type==BLOCKED_FETCH);
565     /* store link (we might overwrite it via blockFetch later on */
566     next = (StgBlockedFetch *)(bf->link);
567
568     /*
569      * Find the target at the end of the indirection chain, and
570      * process it in much the same fashion as the original target
571      * of the fetch.  Though we hope to find graph here, we could
572      * find a black hole (of any flavor) or even a FetchMe.
573      */
574     closure = bf->node;
575     /*
576       We evacuate BQs and update the node fields where necessary in GC.c
577       So, if we find an EVACUATED closure, something has gone Very Wrong
578       (and therefore we let the RTS crash most ungracefully).
579     */
580     ASSERT(get_itbl(closure)->type != EVACUATED);
581       //  closure = ((StgEvacuated *)closure)->evacuee;
582
583     closure = UNWIND_IND(closure);
584     //while ((ind = IS_INDIRECTION(closure)) != NULL) { closure = ind; }
585
586     ip = get_itbl(closure);
587     if (ip->type == FETCH_ME) {
588       /* Forward the Fetch to someone else */
589       rga.payload.gc.gtid = bf->ga.payload.gc.gtid;
590       rga.payload.gc.slot = bf->ga.payload.gc.slot;
591       rga.weight = bf->ga.weight;
592       
593       sendFetch(((StgFetchMe *)closure)->ga, &rga, 0 /* load */);
594
595       // Global statistics: count no. of fetches
596       if (RtsFlags.ParFlags.ParStats.Global &&
597           RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
598         globalParStats.tot_fetch_mess++;
599       }
600
601       IF_PAR_DEBUG(fetch,
602                    belch("__-> processFetches: Forwarding fetch from %lx to %lx",
603                          mytid, rga.payload.gc.gtid));
604
605     } else if (IS_BLACK_HOLE(closure)) {
606       IF_PAR_DEBUG(verbose,
607                    belch("__++ processFetches: trying to send a BLACK_HOLE => doing a blockFetch on closure %p (%s)",
608                          closure, info_type(closure)));
609       bf->node = closure;
610       blockFetch(bf, closure);
611     } else {
612       /* We now have some local graph to send back */
613       nat size;
614
615       packBuffer = gumPackBuffer;
616       IF_PAR_DEBUG(verbose,
617                    belch("__*> processFetches: PackNearbyGraph of closure %p (%s)",
618                          closure, info_type(closure)));
619
620       if ((packBuffer = PackNearbyGraph(closure, END_TSO_QUEUE, &size, bf->ga.payload.gc.gtid)) == NULL) {
621         // Put current BF back on list
622         bf->link = (StgBlockingQueueElement *)PendingFetches;
623         PendingFetches = (StgBlockedFetch *)bf;
624         // ToDo: check that nothing more has to be done to prepare for GC!
625         barf("processFetches: out of heap while packing graph; ToDo: call GC here");
626         GarbageCollect(GetRoots, rtsFalse); 
627         bf = PendingFetches;
628         PendingFetches = (StgBlockedFetch *)(bf->link);
629         closure = bf->node;
630         packBuffer = PackNearbyGraph(closure, END_TSO_QUEUE, &size, bf->ga.payload.gc.gtid);
631         ASSERT(packBuffer != (rtsPackBuffer *)NULL);
632       }
633       rga.payload.gc.gtid = bf->ga.payload.gc.gtid;
634       rga.payload.gc.slot = bf->ga.payload.gc.slot;
635       rga.weight = bf->ga.weight;
636       
637       sendResume(&rga, size, packBuffer);
638
639       // Global statistics: count no. of fetches
640       if (RtsFlags.ParFlags.ParStats.Global &&
641           RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
642         globalParStats.tot_resume_mess++;
643       }
644     }
645   }
646   PendingFetches = END_BF_QUEUE;
647 }
648
649 #if 0
650 /*
651   Alternatively to sending fetch messages directly from the FETCH_ME_entry
652   code we could just store the data about the remote data in a global
653   variable and send the fetch request from the main scheduling loop (similar
654   to processFetches above). This would save an expensive STGCALL in the entry 
655   code because we have to go back to the scheduler anyway.
656 */
657 //@cindex processFetches
658 void
659 processTheRealFetches(void) {
660   StgBlockedFetch *bf;
661   StgClosure *closure, *next;
662     
663   IF_PAR_DEBUG(verbose,
664                belch("__ processTheRealFetches: ");
665                printGA(&theGlobalFromGA);
666                printGA(&theGlobalToGA));
667
668   ASSERT(theGlobalFromGA.payload.gc.gtid != 0 &&
669          theGlobalToGA.payload.gc.gtid != 0);
670
671   /* the old version did this in the FETCH_ME entry code */
672   sendFetch(&theGlobalFromGA, &theGlobalToGA, 0/*load*/);
673   
674 }
675 #endif
676
677
678 /* 
679    Way of dealing with unwanted fish.
680    Used during startup/shutdown, or from unknown PEs 
681 */
682 void
683 bounceFish(void) { 
684   GlobalTaskId origPE;
685   int age, history, hunger;
686   
687   /* IF_PAR_DEBUG(verbose, */
688                belch(".... [%x] Bouncing unwanted FISH",mytid);
689
690   unpackFish(&origPE, &age, &history, &hunger);
691           
692   if (origPE == mytid) {
693     //fishing = rtsFalse;                   // fish has come home
694     outstandingFishes--;
695     last_fish_arrived_at = CURRENT_TIME;  // remember time (see schedule fct)
696     return;                               // that's all
697   }
698
699   /* otherwise, send it home to die */
700   sendFish(origPE, origPE, (age + 1), NEW_FISH_HISTORY, NEW_FISH_HUNGER);
701   // Global statistics: count no. of fetches
702       if (RtsFlags.ParFlags.ParStats.Global &&
703           RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
704         globalParStats.tot_fish_mess++;
705       }
706 }
707    
708 /*
709  * processFish unpacks a fish message, reissuing it if it's our own,
710  * sending work if we have it or sending it onwards otherwise.
711  */
712 //@cindex processFish
713 static void
714 processFish(void)
715 {
716   GlobalTaskId origPE;
717   int age, history, hunger;
718   rtsSpark spark;
719   static rtsPackBuffer *packBuffer; 
720
721   unpackFish(&origPE, &age, &history, &hunger);
722
723   if (origPE == mytid) {
724     //fishing = rtsFalse;                   // fish has come home
725     outstandingFishes--;
726     last_fish_arrived_at = CURRENT_TIME;  // remember time (see schedule fct)
727     return;                               // that's all
728   }
729
730   ASSERT(origPE != mytid);
731   IF_PAR_DEBUG(fish,
732                belch("$$__ processing fish; %d sparks available",
733                      spark_queue_len(&(MainRegTable.rSparks))));
734   while ((spark = findSpark(rtsTrue/*for_export*/)) != NULL) {
735     nat size;
736     // StgClosure *graph;
737
738     packBuffer = gumPackBuffer; 
739     ASSERT(closure_SHOULD_SPARK((StgClosure *)spark));
740     if ((packBuffer = PackNearbyGraph(spark, END_TSO_QUEUE, &size,origPE)) == NULL) {
741       IF_PAR_DEBUG(fish,
742                    belch("$$ GC while trying to satisfy FISH via PackNearbyGraph of node %p",
743                          (StgClosure *)spark));
744       barf("processFish: out of heap while packing graph; ToDo: call GC here");
745       GarbageCollect(GetRoots, rtsFalse);
746       /* Now go back and try again */
747     } else {
748       IF_PAR_DEBUG(verbose,
749                    if (RtsFlags.ParFlags.ParStats.Sparks)
750                      belch("==== STEALING spark %x; sending to %x", spark, origPE));
751       
752       IF_PAR_DEBUG(fish,
753                    belch("$$-- Replying to FISH from %x by sending graph @ %p (%s)",
754                          origPE, 
755                          (StgClosure *)spark, info_type((StgClosure *)spark)));
756       sendSchedule(origPE, size, packBuffer);
757       disposeSpark(spark);
758       // Global statistics: count no. of fetches
759       if (RtsFlags.ParFlags.ParStats.Global &&
760           RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
761         globalParStats.tot_schedule_mess++;
762       }
763
764       break;
765     }
766   }
767   if (spark == (rtsSpark)NULL) {
768     IF_PAR_DEBUG(fish,
769                  belch("$$^^ No sparks available for FISH from %x",
770                        origPE));
771     /* We have no sparks to give */
772     if (age < FISH_LIFE_EXPECTANCY) {
773       /* and the fish is atill young, send it to another PE to look for work */
774       sendFish(choosePE(), origPE,
775                (age + 1), NEW_FISH_HISTORY, NEW_FISH_HUNGER);
776
777       // Global statistics: count no. of fetches
778       if (RtsFlags.ParFlags.ParStats.Global &&
779           RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
780         globalParStats.tot_fish_mess++;
781       }
782     } else { /* otherwise, send it home to die */
783       sendFish(origPE, origPE, (age + 1), NEW_FISH_HISTORY, NEW_FISH_HUNGER);
784       // Global statistics: count no. of fetches
785       if (RtsFlags.ParFlags.ParStats.Global &&
786           RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
787         globalParStats.tot_fish_mess++;
788       }
789     }
790   }
791 }  /* processFish */
792
793 /*
794  * processFetch either returns the requested data (if available) 
795  * or blocks the remote blocking queue on a black hole (if not).
796  */
797
798 //@cindex processFetch
799 static void
800 processFetch(void)
801 {
802   globalAddr ga, rga;
803   int load;
804   StgClosure *closure;
805   StgInfoTable *ip;
806
807   unpackFetch(&ga, &rga, &load);
808   IF_PAR_DEBUG(fetch,
809                belch("%%%%__ Rcvd Fetch for ((%x, %d, 0)), Resume ((%x, %d, %x)) (load %d) from %x",
810                      ga.payload.gc.gtid, ga.payload.gc.slot,
811                      rga.payload.gc.gtid, rga.payload.gc.slot, rga.weight, load,
812                      rga.payload.gc.gtid));
813
814   closure = GALAlookup(&ga);
815   ASSERT(closure != (StgClosure *)NULL);
816   ip = get_itbl(closure);
817   if (ip->type == FETCH_ME) {
818     /* Forward the Fetch to someone else */
819     sendFetch(((StgFetchMe *)closure)->ga, &rga, load);
820
821     // Global statistics: count no. of fetches
822     if (RtsFlags.ParFlags.ParStats.Global &&
823         RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
824       globalParStats.tot_fetch_mess++;
825     }
826   } else if (rga.payload.gc.gtid == mytid) {
827     /* Our own FETCH forwarded back around to us */
828     StgFetchMeBlockingQueue *fmbq = (StgFetchMeBlockingQueue *)GALAlookup(&rga);
829     
830     IF_PAR_DEBUG(fetch,
831                  belch("%%%%== Fetch returned to sending PE; closure=%p (%s); receiver=%p (%s)",
832                        closure, info_type(closure), fmbq, info_type((StgClosure*)fmbq)));
833     /* We may have already discovered that the fetch target is our own. */
834     if ((StgClosure *)fmbq != closure) 
835       CommonUp((StgClosure *)fmbq, closure);
836     (void) addWeight(&rga);
837   } else if (IS_BLACK_HOLE(closure)) {
838     /* This includes RBH's and FMBQ's */
839     StgBlockedFetch *bf;
840
841     /* Can we assert something on the remote GA? */
842     ASSERT(GALAlookup(&rga) == NULL);
843
844     /* If we're hitting a BH or RBH or FMBQ we have to put a BLOCKED_FETCH
845        closure into the BQ in order to denote that when updating this node
846        the result should be sent to the originator of this fetch message. */
847     bf = (StgBlockedFetch *)createBlockedFetch(ga, rga);
848     IF_PAR_DEBUG(fetch,
849                  belch("%%++ Blocking Fetch ((%x, %d, %x)) on %p (%s)",
850                        rga.payload.gc.gtid, rga.payload.gc.slot, rga.weight, 
851                        closure, info_type(closure)));
852     blockFetch(bf, closure);
853   } else {                      
854     /* The target of the FetchMe is some local graph */
855     nat size;
856     // StgClosure *graph;
857     rtsPackBuffer *buffer = (rtsPackBuffer *)NULL;
858
859     if ((buffer = PackNearbyGraph(closure, END_TSO_QUEUE, &size, rga.payload.gc.gtid)) == NULL) {
860       barf("processFetch: out of heap while packing graph; ToDo: call GC here");
861       GarbageCollect(GetRoots, rtsFalse); 
862       closure = GALAlookup(&ga);
863       buffer = PackNearbyGraph(closure, END_TSO_QUEUE, &size, rga.payload.gc.gtid);
864       ASSERT(buffer != (rtsPackBuffer *)NULL);
865     }
866     sendResume(&rga, size, buffer);
867
868     // Global statistics: count no. of fetches
869     if (RtsFlags.ParFlags.ParStats.Global &&
870         RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
871       globalParStats.tot_resume_mess++;
872     }
873   }
874 }
875
876 /* 
877    The list of pending fetches must be a root-list for GC.
878    This routine is called from GC.c (same as marking GAs etc).
879 */
880 void
881 markPendingFetches(rtsBool major_gc) {
882
883   /* No need to traverse the list; this is done via the scavenge code
884      for a BLOCKED_FETCH closure, which evacuates the link field */
885
886   if (PendingFetches != END_BF_QUEUE ) {
887     IF_PAR_DEBUG(tables,
888                  fprintf(stderr, "@@@@ PendingFetches is root; evaced from %p to",
889                          PendingFetches));
890
891     PendingFetches = MarkRoot((StgClosure*)PendingFetches);
892
893     IF_PAR_DEBUG(verbose,
894                  fprintf(stderr, " %p\n", PendingFetches));
895
896   } else {
897     IF_PAR_DEBUG(tables,
898                  fprintf(stderr, "@@@@ PendingFetches is empty; no need to mark it\n"));
899   }
900 }
901
902 /*
903  * processFree unpacks a FREE message and adds the weights to our GAs.
904  */
905 //@cindex processFree
906 static void
907 processFree(void)
908 {
909   int nelem;
910   static StgWord *buffer;
911   int i;
912   globalAddr ga;
913
914   buffer = (StgWord *)gumPackBuffer;
915   unpackFree(&nelem, buffer);
916   IF_PAR_DEBUG(free,
917                belch("!!__ Rcvd Free (%d GAs)", nelem / 2));
918
919   ga.payload.gc.gtid = mytid;
920   for (i = 0; i < nelem;) {
921     ga.weight = (rtsWeight) buffer[i++];
922     ga.payload.gc.slot = (int) buffer[i++];
923     IF_PAR_DEBUG(free,
924                  fprintf(stderr, "!!-- Processing free "); 
925                  printGA(&ga);
926                  fputc('\n', stderr);
927                  );
928     (void) addWeight(&ga);
929   }
930 }
931
932 /*
933  * processResume unpacks a RESUME message into the graph, filling in
934  * the LA -> GA, and GA -> LA tables. Threads blocked on the original
935  * FetchMe (now a blocking queue) are awakened, and the blocking queue
936  * is converted into an indirection.  Finally it sends an ACK in response
937  * which contains any newly allocated GAs.
938  */
939
940 //@cindex processResume
941 static void
942 processResume(GlobalTaskId sender)
943 {
944   int nelem;
945   nat nGAs;
946   static rtsPackBuffer *packBuffer;
947   StgClosure *newGraph, *old;
948   globalAddr lga;
949   globalAddr *gagamap;
950   
951   packBuffer = (rtsPackBuffer *)gumPackBuffer;
952   unpackResume(&lga, &nelem, packBuffer);
953
954   IF_PAR_DEBUG(fetch,
955                fprintf(stderr, "[]__ Rcvd Resume for "); 
956                printGA(&lga);
957                fputc('\n', stderr));
958   IF_PAR_DEBUG(packet,
959                PrintPacket((rtsPackBuffer *)packBuffer));
960   
961   /* 
962    * We always unpack the incoming graph, even if we've received the
963    * requested node in some other data packet (and already awakened
964    * the blocking queue).
965   if (SAVE_Hp + packBuffer[0] >= SAVE_HpLim) {
966     ReallyPerformThreadGC(packBuffer[0], rtsFalse);
967     SAVE_Hp -= packBuffer[0];
968   }
969    */
970
971   // ToDo: Check for GC here !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
972
973   /* Do this *after* GC; we don't want to release the object early! */
974
975   if (lga.weight > 0)
976     (void) addWeight(&lga);
977
978   old = GALAlookup(&lga);
979
980   /* ToDo:  The closure that requested this graph must be one of these two?*/
981   ASSERT(get_itbl(old)->type == FETCH_ME_BQ || 
982          get_itbl(old)->type == RBH);
983
984   if (RtsFlags.ParFlags.ParStats.Full) {
985     StgBlockingQueueElement *bqe, *last_bqe;
986
987     IF_PAR_DEBUG(fetch,
988                  belch("[]-- Resume is REPLY to closure %lx", old));
989
990     /* Write REPLY events to the log file, indicating that the remote
991        data has arrived 
992        NB: we emit a REPLY only for the *last* elem in the queue; this is
993            the one that triggered the fetch message; all other entries
994            have just added themselves to the queue, waiting for the data 
995            they know that has been requested (see entry code for FETCH_ME_BQ)
996     */
997     if ((get_itbl(old)->type == FETCH_ME_BQ ||
998          get_itbl(old)->type == RBH)) {
999       for (bqe = ((StgFetchMeBlockingQueue *)old)->blocking_queue,
1000            last_bqe = END_BQ_QUEUE;
1001              get_itbl(bqe)->type==TSO || 
1002              get_itbl(bqe)->type==BLOCKED_FETCH;
1003            last_bqe = bqe, bqe = bqe->link) { /* nothing */ }
1004
1005       ASSERT(last_bqe==END_BQ_QUEUE || 
1006              get_itbl((StgClosure *)last_bqe)->type == TSO);
1007
1008       /* last_bqe now points to the TSO that triggered the FETCH */ 
1009       if (get_itbl((StgClosure *)last_bqe)->type == TSO)
1010         DumpRawGranEvent(CURRENT_PROC, taskIDtoPE(sender), 
1011                          GR_REPLY, ((StgTSO *)last_bqe), ((StgTSO *)last_bqe)->block_info.closure,
1012                          0, spark_queue_len(&(MainRegTable.rSparks)));
1013     }
1014   }
1015
1016   newGraph = UnpackGraph(packBuffer, &gagamap, &nGAs);
1017   ASSERT(newGraph != NULL);
1018
1019   /* 
1020    * Sometimes, unpacking will common up the resumee with the
1021    * incoming graph, but if it hasn't, we'd better do so now.
1022    */
1023    
1024   if (get_itbl(old)->type == FETCH_ME_BQ)
1025     CommonUp(old, newGraph);
1026
1027   IF_PAR_DEBUG(fetch,
1028                belch("[]-- Ready to resume unpacked graph at %p (%s)",
1029                      newGraph, info_type(newGraph)));
1030
1031   IF_PAR_DEBUG(tables,
1032                DebugPrintGAGAMap(gagamap, nGAs));
1033   
1034   sendAck(sender, nGAs, gagamap);
1035 }
1036
1037 /*
1038  * processSchedule unpacks a SCHEDULE message into the graph, filling
1039  * in the LA -> GA, and GA -> LA tables. The root of the graph is added to
1040  * the local spark queue.  Finally it sends an ACK in response
1041  * which contains any newly allocated GAs.
1042  */
1043 //@cindex processSchedule
1044 static void
1045 processSchedule(GlobalTaskId sender)
1046 {
1047   nat nelem, nGAs;
1048   rtsBool success;
1049   static rtsPackBuffer *packBuffer;
1050   StgClosure *newGraph;
1051   globalAddr *gagamap;
1052   
1053   packBuffer = gumPackBuffer;           /* HWL */
1054   unpackSchedule(&nelem, packBuffer);
1055
1056   IF_PAR_DEBUG(schedule,
1057                belch("--__ Rcvd Schedule (%d elems)", nelem));
1058   IF_PAR_DEBUG(packet,
1059                PrintPacket(packBuffer));
1060
1061   /*
1062    * For now, the graph is a closure to be sparked as an advisory
1063    * spark, but in future it may be a complete spark with
1064    * required/advisory status, priority etc.
1065    */
1066
1067   /*
1068   space_required = packBuffer[0];
1069   if (SAVE_Hp + space_required >= SAVE_HpLim) {
1070     ReallyPerformThreadGC(space_required, rtsFalse);
1071     SAVE_Hp -= space_required;
1072   }
1073   */
1074   // ToDo: check whether GC is necessary !!!!!!!!!!!!!!!!!!!!!
1075   newGraph = UnpackGraph(packBuffer, &gagamap, &nGAs);
1076   ASSERT(newGraph != NULL);
1077   success = add_to_spark_queue(newGraph, &(MainRegTable.rSparks));
1078
1079   if (RtsFlags.ParFlags.ParStats.Full && 
1080       RtsFlags.ParFlags.ParStats.Sparks && 
1081       success) 
1082     DumpRawGranEvent(CURRENT_PROC, CURRENT_PROC, 
1083                      GR_STOLEN, ((StgTSO *)NULL), newGraph, 
1084                      0, 0 /* spark_queue_len(ADVISORY_POOL) */);
1085
1086   IF_PAR_DEBUG(schedule,
1087                if (success)
1088                  belch("--^^  added spark to unpacked graph %p (%s); %d sparks available on [%x] (%s)", 
1089                      newGraph, info_type(newGraph), spark_queue_len(&(MainRegTable.rSparks)), mytid);
1090                else
1091                  belch("--^^  received non-sparkable closure %p (%s); nothing added to spark pool; %d sparks available on [%x]", 
1092                      newGraph, info_type(newGraph), spark_queue_len(&(MainRegTable.rSparks)), mytid));
1093   IF_PAR_DEBUG(packet,
1094                belch("*<    Unpacked graph with root at %p (%s):", 
1095                      newGraph, info_type(newGraph));
1096                PrintGraph(newGraph, 0));
1097
1098   IF_PAR_DEBUG(tables,
1099                DebugPrintGAGAMap(gagamap, nGAs));
1100
1101   sendAck(sender, nGAs, gagamap);
1102
1103   //fishing = rtsFalse;
1104   ASSERT(outstandingFishes>0);
1105   outstandingFishes--;
1106 }
1107
1108 /*
1109  * processAck unpacks an ACK, and uses the GAGA map to convert RBH's
1110  * (which represent shared thunks that have been shipped) into fetch-mes
1111  * to remote GAs.
1112  */
1113 //@cindex processAck
1114 static void
1115 processAck(void)
1116 {
1117   nat nGAs;
1118   globalAddr *gaga;
1119   globalAddr gagamap[256]; // ToDo: elim magic constant!!   MAX_GAS * 2];??
1120
1121   unpackAck(&nGAs, gagamap);
1122
1123   IF_PAR_DEBUG(tables,
1124                belch(",,,, Rcvd Ack (%d pairs)", nGAs);
1125                DebugPrintGAGAMap(gagamap, nGAs));
1126
1127   IF_DEBUG(sanity,
1128            checkGAGAMap(gagamap, nGAs));
1129
1130   /*
1131    * For each (oldGA, newGA) pair, set the GA of the corresponding
1132    * thunk to the newGA, convert the thunk to a FetchMe, and return
1133    * the weight from the oldGA.
1134    */
1135   for (gaga = gagamap; gaga < gagamap + nGAs * 2; gaga += 2) {
1136     StgClosure *old_closure = GALAlookup(gaga);
1137     StgClosure *new_closure = GALAlookup(gaga + 1);
1138
1139     ASSERT(old_closure != NULL);
1140     if (new_closure == NULL) {
1141       /* We don't have this closure, so we make a fetchme for it */
1142       globalAddr *ga = setRemoteGA(old_closure, gaga + 1, rtsTrue);
1143       
1144       /* convertToFetchMe should be done unconditionally here.
1145          Currently, we assign GAs to CONSTRs, too, (a bit of a hack),
1146          so we have to check whether it is an RBH before converting
1147
1148          ASSERT(get_itbl(old_closure)==RBH);
1149       */
1150       if (get_itbl(old_closure)->type==RBH)
1151         convertToFetchMe((StgRBH *)old_closure, ga);
1152     } else {
1153       /* 
1154        * Oops...we've got this one already; update the RBH to
1155        * point to the object we already know about, whatever it
1156        * happens to be.
1157        */
1158       CommonUp(old_closure, new_closure);
1159       
1160       /* 
1161        * Increase the weight of the object by the amount just
1162        * received in the second part of the ACK pair.
1163        */
1164       (void) addWeight(gaga + 1);
1165     }
1166     (void) addWeight(gaga);
1167   }
1168
1169   /* check the sanity of the LAGA and GALA tables after mincing them */
1170   IF_DEBUG(sanity, checkLAGAtable(rtsFalse));
1171 }
1172
1173 #ifdef DIST
1174
1175 void
1176 bounceReval(void) {  
1177   barf("Task %x: TODO: should send NACK in response to REVAL",mytid);     
1178 }
1179
1180 static void
1181 processReval(GlobalTaskId sender) //similar to schedule...
1182 { nat nelem, space_required, nGAs;
1183   static rtsPackBuffer *packBuffer;
1184   StgClosure *newGraph;
1185   globalAddr *gagamap;
1186   StgTSO*     tso;
1187   globalAddr *ga;
1188   
1189   packBuffer = gumPackBuffer;           /* HWL */
1190   unpackSchedule(&nelem, packBuffer); /* okay, since the structure is the same */
1191
1192   IF_PAR_DEBUG(packet,
1193                belch("@;~) [%x] Rcvd Reval (%d elems)", mytid, nelem);
1194                PrintPacket(packBuffer));
1195
1196   /*
1197   space_required = packBuffer[0];
1198   if (SAVE_Hp + space_required >= SAVE_HpLim) {
1199     ReallyPerformThreadGC(space_required, rtsFalse);
1200     SAVE_Hp -= space_required;
1201   }
1202   */
1203   
1204   // ToDo: check whether GC is necessary !!!!!!!!!!!!!!!!!!!!!
1205   newGraph = UnpackGraph(packBuffer, &gagamap, &nGAs);
1206   ASSERT(newGraph != NULL);
1207   
1208   IF_PAR_DEBUG(packet,
1209                belch("@;~)  Unpacked graph with root at %p (%s):", 
1210                      newGraph, info_type(newGraph));
1211                PrintGraph(newGraph, 0));
1212
1213   IF_PAR_DEBUG(tables,
1214                DebugPrintGAGAMap(gagamap, nGAs));
1215
1216   IF_PAR_DEBUG(tables, 
1217     printLAGAtable();   
1218     DebugPrintGAGAMap(gagamap, nGAs));   
1219
1220   //We don't send an Ack to the head!!!!
1221   ASSERT(nGAs>0);  
1222   sendAck(sender, nGAs-1, gagamap+2);
1223   
1224   IF_PAR_DEBUG(verbose,
1225                belch("@;~)  About to create Reval thread on behalf of %x", 
1226                      sender));
1227   
1228   tso=createGenThread(RtsFlags.GcFlags.initialStkSize,newGraph);
1229   tso->priority=RevalPriority;
1230   tso->revalSlot=gagamap->payload.gc.slot;//record who sent the reval
1231   tso->revalTid =gagamap->payload.gc.gtid;
1232   scheduleThread(tso);
1233   context_switch = 1; // switch at the earliest opportunity
1234
1235 #endif
1236
1237
1238 //@node GUM Message Processor, Miscellaneous Functions, Message-Processing Functions, High Level Communications Routines
1239 //@subsection GUM Message Processor
1240
1241 /*
1242  * GUM Message Processor
1243
1244  * processMessages processes any messages that have arrived, calling
1245  * appropriate routines depending on the message tag
1246  * (opcode). N.B. Unless profiling it assumes that there {\em ARE} messages
1247  * present and performs a blocking receive! During profiling it
1248  * busy-waits in order to record idle time.
1249  */
1250
1251 //@cindex processMessages
1252 rtsBool
1253 processMessages(void)
1254 {
1255   rtsPacket packet;
1256   OpCode opcode;
1257   GlobalTaskId task;
1258   rtsBool receivedFinish = rtsFalse;
1259
1260   do {
1261     packet = GetPacket();  /* Get next message; block until one available */
1262     getOpcodeAndSender(packet, &opcode, &task);
1263
1264     if (task==SysManTask) { 
1265       switch (opcode) { 
1266       case PP_PETIDS:
1267         processPEtids();
1268         break;
1269           
1270       case PP_FINISH:
1271         IF_PAR_DEBUG(verbose,
1272                      belch("==== received FINISH [%p]", mytid));
1273         /* this boolean value is returned and propagated to the main 
1274            scheduling loop, thus shutting-down this PE */
1275         receivedFinish = rtsTrue;
1276         break;  
1277           
1278       default:  
1279         barf("Task %x: received unknown opcode %x from SysMan",mytid, opcode);
1280       }
1281     } else if (taskIDtoPE(task)==0) { 
1282       /* When a new PE joins then potentially FISH & REVAL message may
1283          reach PES before they are notified of the new PEs existance.  The
1284          only solution is to bounce/fail these messages back to the sender.
1285          But we will worry about it once we start seeing these race
1286          conditions!  */
1287       switch (opcode) { 
1288       case PP_FISH:
1289         bounceFish();
1290         break;
1291 #ifdef DIST       
1292       case PP_REVAL:
1293         bounceReval();
1294         break;    
1295 #endif          
1296       case PP_PETIDS:
1297         belch("Task %x: Ignoring PVM session opened by another SysMan %x",mytid,task);
1298         break;
1299         
1300       case PP_FINISH:   
1301         break;
1302         
1303       default:  
1304         belch("Task %x: Ignoring opcode %x from unknown PE %x",mytid, opcode, task);
1305       }
1306     } else
1307       switch (opcode) {
1308       case PP_FETCH:
1309         processFetch();
1310         // Global statistics: count no. of fetches
1311         if (RtsFlags.ParFlags.ParStats.Global &&
1312             RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
1313           globalParStats.rec_fetch_mess++;
1314         }
1315         break;
1316
1317       case PP_RESUME:
1318         processResume(task);
1319         // Global statistics: count no. of fetches
1320         if (RtsFlags.ParFlags.ParStats.Global &&
1321             RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
1322           globalParStats.rec_resume_mess++;
1323         }
1324         break;
1325
1326       case PP_ACK:
1327         processAck();
1328         break;
1329
1330       case PP_FISH:
1331         processFish();
1332         // Global statistics: count no. of fetches
1333         if (RtsFlags.ParFlags.ParStats.Global &&
1334             RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
1335           globalParStats.rec_fish_mess++;
1336         }
1337         break;
1338
1339       case PP_FREE:
1340         processFree();
1341         break;
1342       
1343       case PP_SCHEDULE:
1344         processSchedule(task);
1345         // Global statistics: count no. of fetches
1346         if (RtsFlags.ParFlags.ParStats.Global &&
1347             RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
1348           globalParStats.rec_schedule_mess++;
1349         }
1350         break;
1351       
1352 #ifdef DIST      
1353       case PP_REVAL:
1354         processReval(task);
1355         // Global statistics: count no. of fetches
1356         if (RtsFlags.ParFlags.ParStats.Global &&
1357             RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
1358           globalParStats.rec_reval_mess++;
1359         }
1360         break;
1361 #endif
1362       
1363       default:
1364         /* Anything we're not prepared to deal with. */
1365         barf("Task %x: Unexpected opcode %x from %x",
1366              mytid, opcode, task);
1367       } /* switch */
1368
1369   } while (PacketsWaiting());   /* While there are messages: process them */
1370   return receivedFinish;
1371 }                               /* processMessages */
1372
1373 //@node Miscellaneous Functions, Index, GUM Message Processor, High Level Communications Routines
1374 //@subsection Miscellaneous Functions
1375
1376 /*
1377  * blockFetch blocks a BlockedFetch node on some kind of black hole.
1378  */
1379 //@cindex blockFetch
1380 void
1381 blockFetch(StgBlockedFetch *bf, StgClosure *bh) {
1382   bf->node = bh;
1383   switch (get_itbl(bh)->type) {
1384   case BLACKHOLE:
1385     bf->link = END_BQ_QUEUE;
1386     //((StgBlockingQueue *)bh)->header.info = &stg_BLACKHOLE_BQ_info;
1387     SET_INFO(bh, &stg_BLACKHOLE_BQ_info); // turn closure into a blocking queue
1388     ((StgBlockingQueue *)bh)->blocking_queue = (StgBlockingQueueElement *)bf;
1389     
1390     // put bh on the mutables list
1391     recordMutable((StgMutClosure *)bh);
1392     break;
1393     
1394   case BLACKHOLE_BQ:
1395     /* enqueue bf on blocking queue of closure bh */
1396     bf->link = ((StgBlockingQueue *)bh)->blocking_queue;
1397     ((StgBlockingQueue *)bh)->blocking_queue = (StgBlockingQueueElement *)bf;
1398
1399     // put bh on the mutables list; ToDo: check
1400     recordMutable((StgMutClosure *)bh);
1401     break;
1402
1403   case FETCH_ME_BQ:
1404     /* enqueue bf on blocking queue of closure bh */
1405     bf->link = ((StgFetchMeBlockingQueue *)bh)->blocking_queue;
1406     ((StgFetchMeBlockingQueue *)bh)->blocking_queue = (StgBlockingQueueElement *)bf;
1407
1408     // put bh on the mutables list; ToDo: check
1409     recordMutable((StgMutClosure *)bh);
1410     break;
1411     
1412   case RBH:
1413     /* enqueue bf on blocking queue of closure bh */
1414     bf->link = ((StgRBH *)bh)->blocking_queue;
1415     ((StgRBH *)bh)->blocking_queue = (StgBlockingQueueElement *)bf;
1416
1417     // put bh on the mutables list; ToDo: check
1418     recordMutable((StgMutClosure *)bh);
1419     break;
1420     
1421   default:
1422     barf("blockFetch: thought %p was a black hole (IP %#lx, %s)",
1423          (StgClosure *)bh, get_itbl((StgClosure *)bh), 
1424          info_type((StgClosure *)bh));
1425   }
1426   IF_PAR_DEBUG(bq,
1427                belch("##++ blockFetch: after block the BQ of %p (%s) is:",
1428                      bh, info_type(bh));
1429                print_bq(bh));
1430 }
1431
1432
1433 /*
1434   @blockThread@ is called from the main scheduler whenever tso returns with
1435   a ThreadBlocked return code; tso has already been added to a blocking
1436   queue (that's done in the entry code of the closure, because it is a 
1437   cheap operation we have to do in any case); the main purpose of this
1438   routine is to send a Fetch message in case we are blocking on a FETCHME(_BQ)
1439   closure, which is indicated by the tso.why_blocked field;
1440   we also write an entry into the log file if we are generating one
1441
1442   Should update exectime etc in the entry code already; but we don't have
1443   something like ``system time'' in the log file anyway, so this should
1444   even out the inaccuracies.
1445 */
1446
1447 //@cindex blockThread
1448 void
1449 blockThread(StgTSO *tso)
1450 {
1451   globalAddr *remote_ga=NULL;
1452   globalAddr *local_ga;
1453   globalAddr fmbq_ga;
1454
1455   // ASSERT(we are on some blocking queue)
1456   ASSERT(tso->block_info.closure != (StgClosure *)NULL);
1457
1458   /*
1459     We have to check why this thread has been blocked.
1460   */
1461   switch (tso->why_blocked) {
1462     case BlockedOnGA:
1463       /* the closure must be a FETCH_ME_BQ; tso came in here via 
1464          FETCH_ME entry code */
1465       ASSERT(get_itbl(tso->block_info.closure)->type==FETCH_ME_BQ);
1466
1467       /* HACK: the link field is used to hold the GA between FETCH_ME_entry
1468          end this point; if something (eg. GC) happens inbetween the whole
1469          thing will blow up 
1470          The problem is that the ga field of the FETCH_ME has been overwritten
1471          with the head of the blocking queue (which is tso). 
1472       */
1473       ASSERT(looks_like_ga(&theGlobalFromGA));
1474       // ASSERT(tso->link!=END_TSO_QUEUE && tso->link!=NULL);
1475       remote_ga = &theGlobalFromGA; //tso->link;
1476       tso->link = (StgTSO*)END_BQ_QUEUE;
1477       /* it was tso which turned node from FETCH_ME into FETCH_ME_BQ =>
1478          we have to send a Fetch message here! */
1479       if (RtsFlags.ParFlags.ParStats.Full) {
1480         /* Note that CURRENT_TIME may perform an unsafe call */
1481         tso->par.exectime += CURRENT_TIME - tso->par.blockedat;
1482         tso->par.fetchcount++;
1483         tso->par.blockedat = CURRENT_TIME;
1484         /* we are about to send off a FETCH message, so dump a FETCH event */
1485         DumpRawGranEvent(CURRENT_PROC, 
1486                          taskIDtoPE(remote_ga->payload.gc.gtid),
1487                          GR_FETCH, tso, tso->block_info.closure, 0, 0);
1488       }
1489       /* Phil T. claims that this was a workaround for a hard-to-find
1490        * bug, hence I'm leaving it out for now --SDM 
1491        */
1492       /* Assign a brand-new global address to the newly created FMBQ  */
1493       local_ga = makeGlobal(tso->block_info.closure, rtsFalse);
1494       splitWeight(&fmbq_ga, local_ga);
1495       ASSERT(fmbq_ga.weight == 1U << (BITS_IN(unsigned) - 1));
1496       
1497       sendFetch(remote_ga, &fmbq_ga, 0/*load*/);
1498
1499       // Global statistics: count no. of fetches
1500       if (RtsFlags.ParFlags.ParStats.Global &&
1501           RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
1502         globalParStats.tot_fetch_mess++;
1503       }
1504
1505       IF_DEBUG(sanity,
1506                theGlobalFromGA.payload.gc.gtid = (GlobalTaskId)0);
1507       break;
1508
1509     case BlockedOnGA_NoSend:
1510       /* the closure must be a FETCH_ME_BQ; tso came in here via 
1511          FETCH_ME_BQ entry code */
1512       ASSERT(get_itbl(tso->block_info.closure)->type==FETCH_ME_BQ);
1513
1514       /* Fetch message has been sent already */
1515       if (RtsFlags.ParFlags.ParStats.Full) {
1516         /* Note that CURRENT_TIME may perform an unsafe call */
1517         tso->par.exectime += CURRENT_TIME - tso->par.blockedat;
1518         tso->par.blockcount++;
1519         tso->par.blockedat = CURRENT_TIME;
1520         /* dump a block event, because fetch has been sent already */
1521         DumpRawGranEvent(CURRENT_PROC, thisPE,
1522                          GR_BLOCK, tso, tso->block_info.closure, 0, 0);
1523       }
1524       break;
1525
1526     case BlockedOnMVar:
1527     case BlockedOnBlackHole:
1528       /* the closure must be a BLACKHOLE_BQ or an RBH; tso came in here via 
1529          BLACKHOLE(_BQ) or CAF_BLACKHOLE or RBH entry code */
1530       ASSERT(get_itbl(tso->block_info.closure)->type==MVAR ||
1531              get_itbl(tso->block_info.closure)->type==BLACKHOLE_BQ ||
1532              get_itbl(tso->block_info.closure)->type==RBH);
1533
1534       /* if collecting stats update the execution time etc */
1535       if (RtsFlags.ParFlags.ParStats.Full) {
1536         /* Note that CURRENT_TIME may perform an unsafe call */
1537         tso->par.exectime += CURRENT_TIME - tso->par.blockedat;
1538         tso->par.blockcount++;
1539         tso->par.blockedat = CURRENT_TIME;
1540         DumpRawGranEvent(CURRENT_PROC, thisPE,
1541                          GR_BLOCK, tso, tso->block_info.closure, 0, 0);
1542       }
1543       break;
1544
1545     case BlockedOnDelay:
1546       /* Whats sort of stats shall we collect for an explicit threadDelay? */
1547       IF_PAR_DEBUG(verbose,
1548                belch("##++ blockThread: TSO %d blocked on ThreadDelay",
1549                      tso->id));
1550       break;
1551
1552     /* Check that the following is impossible to happen, indeed
1553     case BlockedOnException:
1554     case BlockedOnRead:
1555     case BlockedOnWrite:
1556     */
1557     default:
1558       barf("blockThread: impossible why_blocked code %d for TSO %d",
1559            tso->why_blocked, tso->id);
1560   }
1561
1562   IF_PAR_DEBUG(verbose,
1563                belch("##++ blockThread: TSO %d blocked on closure %p (%s); %s",
1564                      tso->id, tso->block_info.closure, info_type(tso->block_info.closure),
1565                      (tso->why_blocked==BlockedOnGA) ? "Sent FETCH for GA" : ""));
1566   
1567   IF_PAR_DEBUG(bq,
1568                print_bq(tso->block_info.closure));
1569 }
1570
1571 /*
1572  * ChoosePE selects a GlobalTaskId from the array of PEs 'at random'.
1573  * Important properties:
1574  *   - it varies during execution, even if the PE is idle
1575  *   - it's different for each PE
1576  *   - we never send a fish to ourselves
1577  */
1578 extern long lrand48 (void);
1579
1580 //@cindex choosePE
1581 GlobalTaskId
1582 choosePE(void)
1583 {
1584   long temp;
1585
1586   temp = lrand48() % nPEs;
1587   if (allPEs[temp] == mytid) {  /* Never send a FISH to yourself */
1588     temp = (temp + 1) % nPEs;
1589   }
1590   return allPEs[temp];
1591 }
1592
1593 /* 
1594  * allocate a BLOCKED_FETCH closure and fill it with the relevant fields
1595  * of the ga argument; called from processFetch when the local closure is
1596  * under evaluation
1597  */
1598 //@cindex createBlockedFetch
1599 StgClosure *
1600 createBlockedFetch (globalAddr ga, globalAddr rga)
1601 {
1602   StgBlockedFetch *bf;
1603   StgClosure *closure;
1604
1605   closure = GALAlookup(&ga);
1606   if ((bf = (StgBlockedFetch *)allocate(_HS + sizeofW(StgBlockedFetch))) == NULL) {
1607     barf("createBlockedFetch: out of heap while allocating heap for a BlocekdFetch; ToDo: call GC here");
1608     GarbageCollect(GetRoots, rtsFalse); 
1609     closure = GALAlookup(&ga);
1610     bf = (StgBlockedFetch *)allocate(_HS + sizeofW(StgBlockedFetch));
1611     // ToDo: check whether really guaranteed to succeed 2nd time around
1612   }
1613
1614   ASSERT(bf != (StgBlockedFetch *)NULL);
1615   SET_INFO((StgClosure *)bf, &stg_BLOCKED_FETCH_info);
1616   // ToDo: check whether other header info is needed
1617   bf->node = closure;
1618   bf->ga.payload.gc.gtid = rga.payload.gc.gtid;
1619   bf->ga.payload.gc.slot = rga.payload.gc.slot;
1620   bf->ga.weight = rga.weight;
1621   // bf->link = NULL;  debugging
1622
1623   IF_PAR_DEBUG(schedule,
1624                fprintf(stderr, "%%%%// created BF: bf=%p (%s) of closure , GA: ",
1625                        bf, info_type((StgClosure*)bf));
1626                printGA(&(bf->ga));
1627                fputc('\n',stderr));
1628   return (StgClosure *)bf;
1629 }
1630
1631 /*
1632  * waitForTermination enters a loop ignoring spurious messages while
1633  * waiting for the termination sequence to be completed.  
1634  */
1635 //@cindex waitForTermination
1636 void
1637 waitForTermination(void)
1638 {
1639   do {
1640     rtsPacket p = GetPacket();
1641     processUnexpectedMessage(p);
1642   } while (rtsTrue);
1643 }
1644
1645 #ifdef DEBUG
1646 //@cindex DebugPrintGAGAMap
1647 void
1648 DebugPrintGAGAMap(globalAddr *gagamap, int nGAs)
1649 {
1650   nat i;
1651   
1652   for (i = 0; i < nGAs; ++i, gagamap += 2)
1653     fprintf(stderr, "__ gagamap[%d] = ((%x, %d, %x)) -> ((%x, %d, %x))\n", i,
1654             gagamap[0].payload.gc.gtid, gagamap[0].payload.gc.slot, gagamap[0].weight,
1655             gagamap[1].payload.gc.gtid, gagamap[1].payload.gc.slot, gagamap[1].weight);
1656 }
1657
1658 //@cindex checkGAGAMap
1659 void
1660 checkGAGAMap(globalAddr *gagamap, int nGAs)
1661 {
1662   nat i;
1663   
1664   for (i = 0; i < (nat)nGAs; ++i, gagamap += 2) {
1665     ASSERT(looks_like_ga(gagamap));
1666     ASSERT(looks_like_ga(gagamap+1));
1667   }
1668 }
1669 #endif
1670
1671 //@cindex freeMsgBuffer
1672 static StgWord **freeMsgBuffer = NULL;
1673 //@cindex freeMsgIndex
1674 static nat      *freeMsgIndex  = NULL;
1675
1676 //@cindex prepareFreeMsgBuffers
1677 void
1678 prepareFreeMsgBuffers(void)
1679 {
1680   nat i;
1681   
1682   /* Allocate the freeMsg buffers just once and then hang onto them. */
1683   if (freeMsgIndex == NULL) {
1684     freeMsgIndex = (nat *) stgMallocBytes(nPEs * sizeof(nat), 
1685                                           "prepareFreeMsgBuffers (Index)");
1686     freeMsgBuffer = (StgWord **) stgMallocBytes(nPEs * sizeof(long *), 
1687                                           "prepareFreeMsgBuffers (Buffer)");
1688     
1689     for(i = 0; i < nPEs; i++) 
1690       if (i != (thisPE-1)) 
1691         freeMsgBuffer[i] = (StgPtr) stgMallocWords(RtsFlags.ParFlags.packBufferSize,
1692                                                "prepareFreeMsgBuffers (Buffer #i)");
1693       else
1694         freeMsgBuffer[i] = 0;
1695   }
1696   
1697   /* Initialize the freeMsg buffer pointers to point to the start of their
1698      buffers */
1699   for (i = 0; i < nPEs; i++)
1700     freeMsgIndex[i] = 0;
1701 }
1702
1703 //@cindex freeRemoteGA
1704 void
1705 freeRemoteGA(int pe, globalAddr *ga)
1706 {
1707   nat i;
1708   
1709   ASSERT(GALAlookup(ga) == NULL);
1710   
1711   if ((i = freeMsgIndex[pe]) + 2 >= RtsFlags.ParFlags.packBufferSize) {
1712     IF_PAR_DEBUG(free,
1713                  belch("!! Filled a free message buffer (sending remaining messages indivisually)"));   
1714
1715     sendFree(ga->payload.gc.gtid, i, freeMsgBuffer[pe]);
1716     i = 0;
1717   }
1718   freeMsgBuffer[pe][i++] = (StgWord) ga->weight;
1719   freeMsgBuffer[pe][i++] = (StgWord) ga->payload.gc.slot;
1720   freeMsgIndex[pe] = i;
1721
1722   IF_DEBUG(sanity,
1723            ga->weight = 0xdead0add;
1724            ga->payload.gc.gtid = 0xbbbbbbbb;
1725            ga->payload.gc.slot = 0xbbbbbbbb;);
1726 }
1727
1728 //@cindex sendFreeMessages
1729 void
1730 sendFreeMessages(void)
1731 {
1732   nat i;
1733   
1734   for (i = 0; i < nPEs; i++) 
1735     if (freeMsgIndex[i] > 0)
1736       sendFree(allPEs[i], freeMsgIndex[i], freeMsgBuffer[i]);
1737 }
1738
1739 /* synchronises with the other PEs. Receives and records in a global
1740  * variable the task-id of SysMan. If this is the main thread (discovered
1741  * in main.lc), identifies itself to SysMan. Finally it receives
1742  * from SysMan an array of the Global Task Ids of each PE, which is
1743  * returned as the value of the function.
1744  */
1745
1746 #if defined(PAR_TICKY)
1747 /* Has to see freeMsgIndex, so must be defined here not in ParTicky.c */
1748 //@cindex stats_CntFreeGA
1749 void
1750 stats_CntFreeGA (void) {  // stats only
1751
1752   // Global statistics: residency of thread and spark pool
1753   if (RtsFlags.ParFlags.ParStats.Global &&
1754       RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
1755     nat i, s;
1756   
1757     globalParStats.cnt_free_GA++;
1758     for (i = 0, s = 0; i < nPEs; i++) 
1759       s += globalParStats.tot_free_GA += freeMsgIndex[i]/2;
1760
1761     if ( s > globalParStats.res_free_GA )
1762       globalParStats.res_free_GA = s;
1763   }
1764 }
1765 #endif /* PAR_TICKY */
1766
1767 #endif /* PAR -- whole file */
1768
1769 //@node Index,  , Miscellaneous Functions, High Level Communications Routines
1770 //@subsection Index
1771
1772 //@index
1773 //* ACK::  @cindex\s-+ACK
1774 //* DebugPrintGAGAMap::  @cindex\s-+DebugPrintGAGAMap
1775 //* FETCH::  @cindex\s-+FETCH
1776 //* FISH::  @cindex\s-+FISH
1777 //* FREE::  @cindex\s-+FREE
1778 //* RESUME::  @cindex\s-+RESUME
1779 //* SCHEDULE::  @cindex\s-+SCHEDULE
1780 //* blockFetch::  @cindex\s-+blockFetch
1781 //* choosePE::  @cindex\s-+choosePE
1782 //* freeMsgBuffer::  @cindex\s-+freeMsgBuffer
1783 //* freeMsgIndex::  @cindex\s-+freeMsgIndex
1784 //* freeRemoteGA::  @cindex\s-+freeRemoteGA
1785 //* gumPackBuffer::  @cindex\s-+gumPackBuffer
1786 //* initMoreBuffers::  @cindex\s-+initMoreBuffers
1787 //* prepareFreeMsgBuffers::  @cindex\s-+prepareFreeMsgBuffers
1788 //* processAck::  @cindex\s-+processAck
1789 //* processFetch::  @cindex\s-+processFetch
1790 //* processFetches::  @cindex\s-+processFetches
1791 //* processFish::  @cindex\s-+processFish
1792 //* processFree::  @cindex\s-+processFree
1793 //* processMessages::  @cindex\s-+processMessages
1794 //* processResume::  @cindex\s-+processResume
1795 //* processSchedule::  @cindex\s-+processSchedule
1796 //* sendAck::  @cindex\s-+sendAck
1797 //* sendFetch::  @cindex\s-+sendFetch
1798 //* sendFish::  @cindex\s-+sendFish
1799 //* sendFree::  @cindex\s-+sendFree
1800 //* sendFreeMessages::  @cindex\s-+sendFreeMessages
1801 //* sendResume::  @cindex\s-+sendResume
1802 //* sendSchedule::  @cindex\s-+sendSchedule
1803 //* unpackAck::  @cindex\s-+unpackAck
1804 //* unpackFetch::  @cindex\s-+unpackFetch
1805 //* unpackFish::  @cindex\s-+unpackFish
1806 //* unpackFree::  @cindex\s-+unpackFree
1807 //* unpackResume::  @cindex\s-+unpackResume
1808 //* unpackSchedule::  @cindex\s-+unpackSchedule
1809 //* waitForTermination::  @cindex\s-+waitForTermination
1810 //@end index