src/langs/pvmc/pvmc_comm.c

   1 #include <stddef.h>
   2 #include "converse.h"
   3 #include "pvmc.h"
   4
   5 CpvStaticDeclare(CmmTable,seq_table);
   6
   7 CpvStaticDeclare(int,pvmc_control_handler);
   8 CpvStaticDeclare(int,pvmc_msg_handler);
   9 CpvStaticDeclare(int*,send_seq_num);
  10 CpvStaticDeclare(int*,recv_seq_num);
  11
  12 CpvExtern(int,pvmc_barrier_num);
  13 CpvExtern(int,pvmc_at_barrier_num);
  14
  15 typedef struct msg_hdr_struct {
  16   char handler[CmiMsgHeaderSizeBytes];
  17   int sender;
  18   unsigned int seq_num;
  19 } msg_hdr;
  20
  21 typedef struct control_msg_struct {
  22   char handler[CmiMsgHeaderSizeBytes];
  23   int type;
  24 } control_msg;
  25
  26 static void pvmc_control_handler_func(void *);
  27 static void pvmc_msg_handler_func(void *);
  28
  29 void pvmc_init_comm(void)
  30 {
  31   int i;
  32
  33 #ifdef PVM_DEBUG
  34   PRINTF("Pe(%d) tid=%d:pvm_init_comm()\n",MYPE(),pvm_mytid());
  35 #endif
  36
  37   CpvInitialize(CmmTable,seq_table);
  38   CpvAccess(seq_table) = CmmNew();
  39
  40   CpvInitialize(int,pvmc_control_handler);
  41   CpvAccess(pvmc_control_handler)=
  42     CmiRegisterHandler(pvmc_control_handler_func);
  43
  44   CpvInitialize(int,pvmc_msg_handler);
  45   CpvAccess(pvmc_msg_handler)=CmiRegisterHandler(pvmc_msg_handler_func);
  46
  47   CpvInitialize(int*,recv_seq_num);
  48   CpvAccess(recv_seq_num)=(int *)MALLOC(CmiNumPes()*sizeof(int));
  49
  50   if (CpvAccess(recv_seq_num)==NULL) {
  51     PRINTF("Pe(%d) tid=%d:%s:%d pvmc_init_comm() can't allocate seq buffer\n",
  52            MYPE(),pvm_mytid(),__FILE__,__LINE__);
  53     exit(1);
  54   }
  55   for(i=0; i<CmiNumPes(); i++)
  56     CpvAccess(recv_seq_num)=0;
  57
  58   CpvInitialize(int*,send_seq_num);
  59   CpvAccess(send_seq_num)=(int *)MALLOC(CmiNumPes()*sizeof(int));
  60
  61   if (CpvAccess(send_seq_num)==NULL) {
  62     PRINTF("Pe(%d) tid=%d:%s:%d pvmc_init_comm() can't allocate seq buffer\n",
  63            MYPE(),pvm_mytid(),__FILE__,__LINE__);
  64     exit(1);
  65   }
  66   for(i=0; i<CmiNumPes(); i++)
  67     CpvAccess(send_seq_num)[i]=0;
  68
  69 }
  70
  71 void pvmc_send_control_msg(int type, int pe)
  72 {
  73   control_msg *msg;
  74
  75   msg=(control_msg *)CmiAlloc(sizeof(control_msg));
  76   msg->type=type;
  77   CmiSetHandler(msg,CpvAccess(pvmc_control_handler));
  78   CmiSyncSendAndFree(pe,sizeof(control_msg),msg);
  79 }
  80
  81 static void pvmc_control_handler_func(void *voidmsg)
  82 {
  83   control_msg *msg = (control_msg *)voidmsg;
  84   switch (msg->type)  {
  85   case PVMC_CTRL_AT_BARRIER:
  86     CpvAccess(pvmc_at_barrier_num)++;
  87     break;
  88   case PVMC_CTRL_THROUGH_BARRIER:
  89     CpvAccess(pvmc_barrier_num)++;
  90     break;
  91   case PVMC_CTRL_KILL:
  92     ConverseExit();
  93     exit(0);
  94     break;
  95   default:
  96     PRINTF("WARNING: %s:%d, Illegal control message\n",__FILE__,__LINE__);
  97   }
  98 }
  99
 100 static void pvmc_msg_handler_func(void *msg)
 101 {
 102   int seq_num;
 103   int sender;
 104   int pvm_tag;
 105   int tags[2];
 106   int rtags[2];
 107
 108   sender=((msg_hdr *)msg)->sender;
 109   seq_num=((msg_hdr *)msg)->seq_num;
 110
 111   tags[0]=sender;
 112   tags[1]=pvmc_gettag((char *)msg+sizeof(msg_hdr));
 113   CmmPut(CpvAccess(seq_table),2,tags,msg);
 114 }
 115
 116 int pvm_kill(int tid)
 117 {
 118   control_msg *exit_msg;
 119
 120 #ifdef PVM_DEBUG
 121   PRINTF("Pe(%d) tid=%d:pvm_kill(%d)\n",
 122         MYPE(),pvm_mytid(),tid);
 123 #endif
 124   pvmc_send_control_msg(PVMC_CTRL_KILL,TID2PE(tid));
 125
 126   return 0;
 127 }
 128
 129 int pvm_send(int pvm_tid, int tag)
 130 {
 131   void *msg;
 132   int msg_sz, conv_tid, conv_tag;
 133
 134 #ifdef PVM_DEBUG
 135   PRINTF("Pe(%d) tid=%d:pvm_send(%d,%d)\n",
 136         MYPE(),pvm_mytid(),pvm_tid,tag);
 137 #endif
 138   pvmc_settidtag(pvm_mytid(),tag);
 139
 140   if ((pvm_tid<1) || ( pvm_tid > CmiNumPes())) {
 141     PRINTF("Pe(%d) tid=%d:%s:%d pvm_send() illegal tid %d\n",
 142            MYPE(),pvm_mytid(),__FILE__,__LINE__,pvm_tid);
 143     return -1;
 144   } else conv_tid = pvm_tid-1;
 145
 146   if (tag<0) {
 147     PRINTF("Pe(%d) tid=%d:%s:%d pvm_send() illegal tag\n",
 148            MYPE(),pvm_mytid(),__FILE__,__LINE__);
 149     return -1;
 150   } else conv_tag = tag;
 151
 152   msg_sz = sizeof(msg_hdr)+pvmc_sendmsgsz();
 153   msg = CmiAlloc(msg_sz);
 154
 155   if (msg==NULL) {
 156     PRINTF("Pe(%d) tid=%d:%s:%d pvm_send() can't alloc msg buffer\n",
 157            MYPE(),pvm_mytid(),__FILE__,__LINE__);
 158     return -1;
 159   }
 160
 161   CmiSetHandler(msg,CpvAccess(pvmc_msg_handler));
 162   ((msg_hdr *)msg)->sender=MYPE();
 163   ((msg_hdr *)msg)->seq_num=CpvAccess(send_seq_num)[conv_tid];
 164   CpvAccess(send_seq_num)[conv_tid]++;
 165
 166   pvmc_packmsg((char *)msg + (int)sizeof(msg_hdr));
 167   CmiSyncSendAndFree(conv_tid,msg_sz,msg);
 168   return 0;
 169 }
 170
 171 int pvm_mcast(int *tids, int ntask, int msgtag)
 172 {
 173   int i;
 174
 175 #ifdef PVM_DEBUG
 176   PRINTF("Pe(%d) tid=%d:pvm_mcast(%x,%d,%d)\n",
 177         MYPE(),pvm_mytid(),tids,ntask,msgtag);
 178 #endif
 179   for(i=0;i<ntask;i++)
 180     pvm_send(tids[i],msgtag);
 181
 182   return 0;
 183 }
 184
 185 int pvm_nrecv(int tid, int tag)
 186 {
 187   int conv_tid, conv_tag;
 188   void *msg;
 189   int tags[2];
 190   int rtags[2];
 191   int sender, seq_num;
 192   int rbuf;
 193
 194 #ifdef PVM_DEBUG
 195   PRINTF("Pe(%d) tid=%d:pvm_nrecv(%d,%d)\n",
 196         MYPE(),pvm_mytid(),tid,tag);
 197 #endif
 198
 199   if (tid==-1)
 200     conv_tid=CmmWildCard;
 201   else conv_tid=tid-1;
 202
 203   if (tag==-1)
 204     conv_tag=CmmWildCard;
 205   else conv_tag=tag;
 206
 207   /*
 208    * Empty messages from machine layer.
 209    */
 210
 211   while(CmiDeliverMsgs(1)==0)
 212     ;
 213
 214   /*
 215    *  See if the message is already in the tag table and extract it.
 216    */
 217
 218   tags[0]=conv_tid;
 219   tags[1]=conv_tag;
 220   msg=CmmGet(CpvAccess(seq_table),2,tags,rtags);
 221   if (msg!=NULL) {
 222     sender = rtags[0];
 223     /*
 224     seq_num = CpvAccess(recv_seq_num)[sender];
 225
 226     if ((((msg_hdr *)msg)->seq_num) != seq_num)
 227       PRINTF("tid=%d:%s:%d pvm_recv() seq number mismatch, I'm confused\n",
 228              tid,__FILE__,__LINE__);
 229     else CpvAccess(recv_seq_num)[sender]++;
 230     */
 231
 232
 233     rbuf=pvm_setrbuf(pvm_mkbuf(PvmDataRaw));
 234     if (rbuf > 0)
 235       {
 236 #ifdef PVM_DEBUG
 237       PRINTF("Pe(%d) tid=%d:%s:%d pvm_nrecv() says pvm_setrbuf=%d\n",
 238         MYPE(),tid,__FILE__,__LINE__,rbuf);
 239 #endif
 240       pvm_freebuf(rbuf);
 241       }
 242     pvmc_unpackmsg(msg,(char *)msg+sizeof(msg_hdr));
 243
 244 #ifdef PVM_DEBUG
 245     PRINTF("Pe(%d) tid=%d:%s:%d pvm_nrecv() returning pvm_getrbuf()=%d\n",
 246         MYPE(),tid,__FILE__,__LINE__,pvm_getrbuf());
 247 #endif
 248     return pvm_getrbuf();
 249   }
 250   else return 0;  /* Non blocking receive returns immediately. */
 251 }
 252
 253 int pvm_recv(int tid, int tag)
 254 {
 255   int bufid=0;
 256
 257 #ifdef PVM_DEBUG
 258   PRINTF("Pe(%d) tid=%d:pvm_recv(%d,%d)\n",
 259         MYPE(),pvm_mytid(),tid,tag);
 260 #endif
 261   while (bufid==0)
 262     bufid=pvm_nrecv(tid,tag);
 263
 264 #ifdef PVM_DEBUG
 265   PRINTF("Pe(%d) tid=%d:pvm_recv(%d,%d) returning %d\n",
 266         MYPE(),pvm_mytid(),tid,tag,bufid);
 267 #endif
 268
 269   return bufid;
 270 }
 271
 272 int pvm_probe(int tid, int tag)
 273 {
 274   int conv_tid, conv_tag;
 275   void *msg;
 276   int tags[2];
 277   int rtags[2];
 278   int sender, seq_num;
 279
 280 #ifdef PVM_DEBUG
 281   PRINTF("Pe(%d) tid=%d:pvm_probe(%d,%d)\n",
 282         MYPE(),pvm_mytid(),tid,tag);
 283 #endif
 284   if (tid==-1)
 285     conv_tid=CmmWildCard;
 286   else conv_tid=tid;
 287
 288   if (tag==-1)
 289     conv_tag=CmmWildCard;
 290   else conv_tag=tag;
 291
 292   /*
 293    * Empty messages from machine layer.
 294    */
 295
 296   while(CmiDeliverMsgs(1)==0)
 297     ;
 298
 299   /*
 300    *  See if the message is already in the tag table
 301    */
 302
 303   tags[0]=conv_tid;
 304   tags[1]=conv_tag;
 305   msg=CmmProbe(CpvAccess(seq_table),2,tags,rtags);
 306   if (msg!=NULL) {
 307     /*
 308     sender = rtag[0];
 309     seq_num = CpvAccess(recv_seq_num)[sender];
 310
 311     if ((((msg_hdr *)msg)->seq_num) != seq_num)
 312       PRINTF("Pe(%d) tid=%d:%s:%d pvm_recv() seq num mismatch, I'm confused\n",
 313              MYPE(),pvm_mytid(),__FILE__,__LINE__);
 314     else CpvAccess(recv_seq_num)[sender]++;
 315     */
 316
 317   /*
 318    * We will just unpack the message, so bufinfo works, but this
 319    * should really just set up what bufinfo needs and unpack the
 320    * rest later
 321    */
 322     pvmc_unpackmsg(msg,(char *)msg+sizeof(msg_hdr));
 323
 324
 325 #ifdef PVM_DEBUG
 326     PRINTF("Pe(%d) tid=%d:%s:%d pvm_probe() returning pvm_getrbuf()=%d\n",
 327         MYPE(),tid,__FILE__,__LINE__,pvm_getrbuf());
 328 #endif
 329     return pvm_getrbuf();
 330   }
 331   else return 0;  /* Probe returns immediately. */
 332 }