rts/parallel/Dist.c

   1 #include "Dist.h"
   2
   3 #ifdef DIST /* whole file */
   4
   5 #include "RtsFlags.h"
   6 #include "RtsUtils.h"
   7 #include "ParallelRts.h"
   8 #include "Parallel.h" // nPEs,allPEs,mytid
   9 #include "HLC.h" //for sendReval
  10 #include "LLC.h" //for pvm stuff
  11 #include "FetchMe.h"     // for BLOCKED_FETCH_info
  12 #include "Storage.h"       // for recordMutable
  13
  14 /* hopefully the result>0  */
  15 StgWord32 cGetPECount(void)
  16 { return nPEs;
  17 }
  18
  19 /* return taskID, n is 1..count, n=1 is always the mainPE */
  20 StgPEId cGetPEId(StgWord32 n)
  21 { return allPEs[n-1];
  22 }
  23
  24 /* return the taskID */
  25 StgPEId cGetMyPEId(void)
  26 { return mytid;
  27 }
  28
  29 /* return the taskID of the owning PE of an MVar/TSO:
  30 - MVAR/TSOs get converted to REMOTE_REFs when shipped, and
  31   there is no mechanism for using these REMOTE_REFs
  32   apart from this code.
  33 */
  34
  35 StgPEId cGetCertainOwner(StgClosure *mv)
  36 { globalAddr *ga;
  37   switch(get_itbl(mv)->type)
  38   { case TSO:
  39     case MVAR:
  40       return  mytid; // must be local
  41     case REMOTE_REF:
  42       ga = LAGAlookup(mv);
  43       ASSERT(ga);
  44       return ga->payload.gc.gtid; // I know its global address
  45   }
  46   barf("Dist.c:cGetCertainOwner() wrong closure type %s",info_type(mv));
  47 }
  48
  49 /* for some additional fun, lets look up a certain host... */
  50 StgPEId cGetHostOwner(StgByteArray h) //okay h is a C string
  51 { int nArch,nHost,nTask,i;
  52   StgPEId dtid;
  53   struct pvmhostinfo *host;
  54   struct pvmtaskinfo *task;
  55
  56   dtid=0;
  57   pvm_config(&nHost,&nArch,&host);
  58   for(i=0;i<nHost;i++)
  59     if(strcmp(host[i].hi_name,h)==0)
  60     { dtid=host[i].hi_tid;
  61       break;
  62     }
  63   if(dtid==0) return 0; // no host of that name
  64
  65   for(i=0;i<nPEs;i++)
  66   { pvm_tasks(allPEs[i],&nTask,&task);
  67     ASSERT(nTask==1); //cause we lookup a single task
  68     if(task[0].ti_host==dtid)
  69       return allPEs[i];
  70   }
  71   return 0;  //know host, put no PE on it
  72 }
  73
  74 void cRevalIO(StgClosure *job,StgPEId p)
  75 { nat size;
  76   rtsPackBuffer *buffer=NULL;
  77
  78   ASSERT(get_itbl(job)->type==MVAR);
  79   job=((StgMVar*)job)->value; // extract the job from the MVar
  80
  81   ASSERT(closure_THUNK(job)); // must be a closure!!!!!
  82   ASSERT(p!=mytid);
  83
  84   buffer = PackNearbyGraph(job, END_TSO_QUEUE, &size,p);
  85   ASSERT(buffer != (rtsPackBuffer *)NULL);
  86   ASSERT(get_itbl(job)->type==RBH);
  87
  88   IF_PAR_DEBUG(verbose,
  89                belch("@;~) %x doing revalIO to %x\n",
  90                      mytid,p));
  91
  92   sendReval(p,size,buffer);
  93
  94   if (RtsFlags.ParFlags.ParStats.Global &&
  95       RtsFlags.GcFlags.giveStats > NO_GC_STATS) {
  96     globalParStats.tot_reval_mess++;
  97   }
  98
  99   /*
 100      We turn job into a FETCHME_BQ so that the thread will block
 101      when it enters it.
 102
 103      Note: it will not receive an ACK, thus no GA.
 104   */
 105
 106   ASSERT(get_itbl(job)->type==RBH);
 107
 108    /* put closure on mutables list, while it is still a RBH */
 109   recordMutable((StgMutClosure *)job);
 110
 111   /* actually turn it into a FETCH_ME_BQ */
 112   SET_INFO(job, &FETCH_ME_BQ_info);
 113   ((StgFetchMe *)job)->ga = 0;     //hope this won't make anyone barf!!!
 114   ((StgBlockingQueue*)job)->blocking_queue=END_BQ_QUEUE;
 115 }
 116
 117 #endif