ghc/rts/parallel/LLComms.c

   1 /* ----------------------------------------------------------------------------
   2  * Time-stamp: <Mon Mar 19 2001 22:10:38 Stardate: [-30]6354.62 hwloidl>
   3  * $Id: LLComms.c,v 1.5 2001/08/14 13:40:10 sewardj Exp $
   4  *
   5  * GUM Low-Level Inter-Task Communication
   6  *
   7  * This module defines PVM Routines for PE-PE  communication.
   8  *
   9  * P. Trinder, December 5th. 1994.
  10  * P. Trinder, July 1998
  11  * H-W. Loidl, November 1999 -
  12  --------------------------------------------------------------------------- */
  13
  14 #ifdef PAR /* whole file */
  15
  16 //@node GUM Low-Level Inter-Task Communication, , ,
  17 //@section GUM Low-Level Inter-Task Communication
  18
  19 /*
  20  *This module defines the routines which communicate between PEs.  The
  21  *code is based on Kevin Hammond's GRIP RTS. (OpCodes.h defines
  22  *PEOp1 etc. in terms of sendOp1 etc.).
  23  *
  24  *Routine       &       Arguments
  25  *              &
  26  *sendOp        &       0                       \\
  27  *sendOp1       &       1                       \\
  28  *sendOp2       &       2                       \\
  29  *sendOpN       &       vector                  \\
  30  *sendOpV       &       variable                \\
  31  *sendOpNV      &       variable+ vector        \\
  32  *
  33  *First the standard include files.
  34  */
  35
  36 //@menu
  37 //* Macros etc::
  38 //* Includes::
  39 //* Auxiliary functions::
  40 //* Index::
  41 //@end menu
  42
  43 //@node Macros etc, Includes, GUM Low-Level Inter-Task Communication, GUM Low-Level Inter-Task Communication
  44 //@subsection Macros etc
  45
  46 /* Evidently not Posix */
  47 /* #include "PosixSource.h" */
  48
  49 #define UNUSED           /* nothing */
  50
  51 //@node Includes, Auxiliary functions, Macros etc, GUM Low-Level Inter-Task Communication
  52 //@subsection Includes
  53
  54 #include "Rts.h"
  55 #include "RtsFlags.h"
  56 #include "RtsUtils.h"
  57 #include "Parallel.h"
  58 #include "ParallelRts.h"
  59 #if defined(DEBUG)
  60 # include "ParallelDebug.h"
  61 #endif
  62 #include "LLC.h"
  63
  64 #ifdef __STDC__
  65 #include <stdarg.h>
  66 #else
  67 #include <varargs.h>
  68 #endif
  69
  70 /* Cannot use std macro when compiling for SysMan */
  71 /* debugging enabled */
  72 // #define IF_PAR_DEBUG(c,s)  { s; }
  73 /* debugging disabled */
  74 #define IF_PAR_DEBUG(c,s)  /* nothing */
  75
  76 //@node Auxiliary functions, Index, Includes, GUM Low-Level Inter-Task Communication
  77 //@subsection Auxiliary functions
  78
  79 /*
  80  * heapChkCounter tracks the number of heap checks since the last probe.
  81  * Not currently used! We check for messages when a thread is resheduled.
  82  */
  83 int heapChkCounter = 0;
  84
  85 /*
  86  * Then some miscellaneous functions.
  87  * getOpName returns the character-string name of any OpCode.
  88  */
  89
  90 char *UserPEOpNames[] = { PEOP_NAMES };
  91
  92 //@cindex getOpName
  93 char *
  94 getOpName(nat op)
  95 {
  96     if (op >= MIN_PEOPS && op <= MAX_PEOPS)
  97         return (UserPEOpNames[op - MIN_PEOPS]);
  98     else
  99         return ("Unknown PE OpCode");
 100 }
 101
 102 /*
 103  * traceSendOp handles the tracing of messages.
 104  */
 105
 106 //@cindex traceSendOp
 107 static void
 108 traceSendOp(OpCode op, GlobalTaskId dest UNUSED,
 109              unsigned int data1 UNUSED, unsigned int data2 UNUSED)
 110 {
 111     char *OpName;
 112
 113     OpName = getOpName(op);
 114     IF_PAR_DEBUG(trace,
 115                  fprintf(stderr," %s [%x,%x] sent from %x to %x",
 116                        OpName, data1, data2, mytid, dest));
 117 }
 118
 119 /*
 120  * sendOp sends a 0-argument message with OpCode {\em op} to
 121  * the global task {\em task}.
 122  */
 123
 124 //@cindex sendOp
 125 void
 126 sendOp(OpCode op, GlobalTaskId task)
 127 {
 128     traceSendOp(op, task,0,0);
 129
 130     pvm_initsend(PvmDataRaw);
 131     pvm_send(task, op);
 132 }
 133
 134 /*
 135  * sendOp1 sends a 1-argument message with OpCode {\em op}
 136  * to the global task {\em task}.
 137  */
 138
 139 //@cindex sendOp1
 140 void
 141 sendOp1(OpCode op, GlobalTaskId task, StgWord arg1)
 142 {
 143     traceSendOp(op, task, arg1,0);
 144
 145     pvm_initsend(PvmDataRaw);
 146     PutArg1(arg1);
 147     pvm_send(task, op);
 148 }
 149
 150
 151 /*
 152  * sendOp2 is used by the FP code only.
 153  */
 154
 155 //@cindex sendOp2
 156 void
 157 sendOp2(OpCode op, GlobalTaskId task, StgWord arg1, StgWord arg2)
 158 {
 159     traceSendOp(op, task, arg1, arg2);
 160
 161     pvm_initsend(PvmDataRaw);
 162     PutArg1(arg1);
 163     PutArg2(arg2);
 164     pvm_send(task, op);
 165 }
 166
 167 /*
 168  *
 169  * sendOpV takes a variable number of arguments, as specified by {\em n}.
 170  * For example,
 171  *
 172  *    sendOpV( PP_STATS, StatsTask, 3, start_time, stop_time, sparkcount);
 173  */
 174
 175 //@cindex sendOpV
 176 void
 177 sendOpV(OpCode op, GlobalTaskId task, int n, ...)
 178 {
 179     va_list ap;
 180     int i;
 181     StgWord arg;
 182
 183     va_start(ap, n);
 184
 185     traceSendOp(op, task, 0, 0);
 186
 187     pvm_initsend(PvmDataRaw);
 188
 189     for (i = 0; i < n; ++i) {
 190         arg = va_arg(ap, StgWord);
 191         PutArgN(i, arg);
 192     }
 193     va_end(ap);
 194
 195     pvm_send(task, op);
 196 }
 197
 198 /*
 199  *
 200  * sendOpNV takes a variable-size datablock, as specified by {\em
 201  * nelem} and a variable number of arguments, as specified by {\em
 202  * narg}. N.B. The datablock and the additional arguments are contiguous
 203  * and are copied over together.  For example,
 204  *
 205  *        sendOpNV(PP_RESUME, tsoga.pe, 6, nelem, data,
 206  *          (W_) ga.weight, (W_) ga.loc.gc.gtid, (W_) ga.loc.gc.slot,
 207  *          (W_) tsoga.weight, (W_) tsoga.loc.gc.gtid, (W_) tsoga.loc.gc.slot);
 208  *
 209  * Important: The variable arguments must all be StgWords.
 210
 211  sendOpNV(_, tid, m, n, data, x1, ..., xm):
 212
 213                          |   n elems
 214      +------------------------------
 215      | x1 | ... | xm | n | data ....
 216      +------------------------------
 217  */
 218
 219 //@cindex sendOpNV
 220 void
 221 sendOpNV(OpCode op, GlobalTaskId task, int nelem,
 222          StgWord *datablock, int narg, ...)
 223 {
 224     va_list ap;
 225     int i;
 226     StgWord arg;
 227
 228     va_start(ap, narg);
 229
 230     traceSendOp(op, task, 0, 0);
 231     IF_PAR_DEBUG(trace,
 232                  fprintf(stderr,"~~ sendOpNV: op = %x (%s), task = %x, narg = %d, nelem = %d",
 233                        op, getOpName(op), task, narg, nelem));
 234
 235     pvm_initsend(PvmDataRaw);
 236
 237     for (i = 0; i < narg; ++i) {
 238         arg = va_arg(ap, StgWord);
 239         IF_PAR_DEBUG(trace,
 240                      fprintf(stderr,"~~ sendOpNV: arg = %d\n",arg));
 241         PutArgN(i, arg);
 242     }
 243     arg = (StgWord) nelem;
 244     PutArgN(narg, arg);
 245
 246 /*  for (i=0; i < nelem; ++i) fprintf(stderr, "%d ",datablock[i]); */
 247 /*  fprintf(stderr," in sendOpNV\n");*/
 248
 249     PutArgs(datablock, nelem);
 250     va_end(ap);
 251
 252     pvm_send(task, op);
 253 }
 254
 255 /*
 256  * sendOpN take a variable size array argument, whose size is given by
 257  * {\em n}.  For example,
 258  *
 259  *    sendOpN( PP_STATS, StatsTask, 3, stats_array);
 260  */
 261
 262 //@cindex sendOpN
 263 void
 264 sendOpN(OpCode op, GlobalTaskId task, int n, StgPtr args)
 265 {
 266     long arg;
 267
 268     traceSendOp(op, task, 0, 0);
 269
 270     pvm_initsend(PvmDataRaw);
 271     arg = (long) n;
 272     PutArgN(0, arg);
 273     PutArgs(args, n);
 274     pvm_send(task, op);
 275 }
 276
 277 /*
 278  * broadcastOpN is as sendOpN but broadcasts to all members of a group.
 279  */
 280
 281 void
 282 broadcastOpN(OpCode op, char *group, int n, StgPtr args)
 283 {
 284   long arg;
 285
 286   //traceSendOp(op, task, 0, 0);
 287
 288   pvm_initsend(PvmDataRaw);
 289   arg = (long) n;
 290   PutArgN(0, arg);
 291   PutArgs(args, n);
 292   pvm_bcast(group, op);
 293 }
 294
 295 /*
 296    waitForPEOp waits for a packet from global task who with the
 297    OpCode op.  If ignore is true all other messages are simply ignored;
 298    otherwise they are handled by processUnexpected.
 299  */
 300 //@cindex waitForPEOp
 301 rtsPacket
 302 waitForPEOp(OpCode op, GlobalTaskId who, void(*processUnexpected)(rtsPacket) )
 303 {
 304   rtsPacket p;
 305   int nbytes;
 306   OpCode opCode;
 307   GlobalTaskId sender_id;
 308   rtsBool match;
 309
 310   IF_PAR_DEBUG(verbose,
 311                fprintf(stderr,"~~ waitForPEOp: expecting op = %x (%s), who = [%x]\n",
 312                        op, getOpName(op), who));
 313
 314   do {
 315     while((p = pvm_recv(ANY_TASK,ANY_OPCODE)) < 0)
 316       pvm_perror("waitForPEOp: Waiting for PEOp");
 317
 318     pvm_bufinfo( p, &nbytes, &opCode, &sender_id );
 319     match = (op == ANY_OPCODE || op == opCode) &&
 320             (who == ANY_TASK || who == sender_id);
 321
 322     if (match) {
 323       IF_PAR_DEBUG(verbose,
 324                    fprintf(stderr,
 325                            "~~waitForPEOp: Qapla! received: OpCode = %#x (%s), sender_id = [%x]",
 326                            opCode, getOpName(opCode), sender_id));
 327
 328       return(p);
 329     }
 330
 331     /* Handle the unexpected OpCodes */
 332     if (processUnexpected!=NULL) {
 333       (*processUnexpected)(p);
 334     } else {
 335       IF_PAR_DEBUG(verbose,
 336                    fprintf(stderr,
 337                            "~~ waitForPEOp: ignoring OpCode = %#x (%s), sender_id = [%x]",
 338                            opCode, getOpName(opCode), sender_id));
 339     }
 340
 341   } while(rtsTrue);
 342 }
 343
 344 /*
 345   processUnexpected processes unexpected messages. If the message is a
 346   FINISH it exits the prgram, and PVM gracefully
 347  */
 348 //@cindex processUnexpectedMessage
 349 void
 350 processUnexpectedMessage(rtsPacket packet) {
 351     OpCode opCode = getOpcode(packet);
 352
 353     IF_PAR_DEBUG(verbose,
 354                  GlobalTaskId sender = senderTask(packet);
 355                  fprintf(stderr,"~~ [%x] processUnexpected: Received %x (%s), sender %x\n",
 356                        mytid, opCode, getOpName(opCode), sender));
 357
 358     switch (opCode) {
 359     case PP_FINISH:
 360         stg_exit(EXIT_SUCCESS);
 361         break;
 362
 363       /* Anything we're not prepared to deal with.  Note that ALL OpCodes
 364          are discarded during termination -- this helps prevent bizarre
 365          race conditions.  */
 366       default:
 367         // if (!GlobalStopPending)
 368         {
 369           GlobalTaskId errorTask;
 370           OpCode opCode;
 371
 372           getOpcodeAndSender(packet, &opCode, &errorTask);
 373           fprintf(stderr,"== Task %x: Unexpected OpCode %x from %x in processUnexpected",
 374                 mytid, opCode, errorTask );
 375
 376           stg_exit(EXIT_FAILURE);
 377         }
 378     }
 379 }
 380
 381 //@cindex getOpcode
 382 OpCode
 383 getOpcode(rtsPacket p)
 384 {
 385   int nbytes;
 386   OpCode OpCode;
 387   GlobalTaskId sender_id;
 388   /* read PVM buffer */
 389   pvm_bufinfo(p, &nbytes, &OpCode, &sender_id);
 390   /* return tag of the buffer as opcode */
 391   return(OpCode);
 392 }
 393
 394 //@cindex getOpcodeAndSender
 395 void
 396 getOpcodeAndSender(rtsPacket p, OpCode *opCodep, GlobalTaskId *senderIdp)
 397 {
 398   int nbytes;
 399   /* read PVM buffer */
 400   pvm_bufinfo(p, &nbytes, opCodep, senderIdp);
 401 }
 402
 403 //@cindex senderTask
 404 GlobalTaskId
 405 senderTask(rtsPacket p)
 406 {
 407   int nbytes;
 408   OpCode opCode;
 409   GlobalTaskId sender_id;
 410   /* read PVM buffer */
 411   pvm_bufinfo(p, &nbytes, &opCode, &sender_id);
 412   return(sender_id);
 413 }
 414
 415 /*
 416  * startUpPE does the low-level comms specific startup stuff for a
 417  * PE. It initialises the comms system, joins the appropriate groups
 418  * allocates the PE buffer
 419  */
 420
 421 //@cindex startUpPE
 422 void
 423 startUpPE(void)
 424 {
 425   mytid = _my_gtid;     /* Initialise PVM and get task id into global var.*/
 426
 427   IF_PAR_DEBUG(verbose,
 428                fprintf(stderr,"== [%x] PEStartup: Task id = [%x], No. PEs = %d \n",
 429                        mytid, mytid, nPEs));
 430   checkComms(pvm_joingroup(PEGROUP), "PEStartup");
 431   IF_PAR_DEBUG(verbose,
 432                fprintf(stderr,"== [%x] PEStartup: Joined PEGROUP\n", mytid));
 433 }
 434
 435 /*
 436  * PEShutdown does the low-level comms-specific shutdown stuff for a
 437  * single PE. It leaves the groups and then exits from pvm.
 438  */
 439 //@cindex shutDownPE
 440 void
 441 shutDownPE(void)
 442 {
 443   IF_PAR_DEBUG(verbose,
 444                fprintf(stderr, "== [%x] PEshutdown\n", mytid));
 445
 446   checkComms(pvm_lvgroup(PEGROUP),"PEShutDown");
 447   checkComms(pvm_exit(),"PEShutDown");
 448 }
 449
 450 /*
 451    Extract the exit code out of a PP_FINISH packet (used in SysMan)
 452 */
 453 int
 454 getExitCode(int nbytes, GlobalTaskId *sender_idp) {
 455   int exitCode=0;
 456
 457   if (nbytes==4) {               // Notification from a task doing pvm_exit
 458     GetArgs(sender_idp,1);       // Presumably this must be MainPE Id
 459     exitCode = -1;
 460   } else if (nbytes==8) {        // Doing a controlled shutdown
 461     GetArgs(&exitCode,1);        // HACK: controlled shutdown == 2 values
 462     GetArgs(&exitCode,1);
 463   } else {
 464     exitCode = -2;               // everything else
 465   }
 466   return exitCode;
 467 }
 468
 469 #endif /* PAR -- whole file */
 470
 471 //@node Index,  , Auxiliary functions, GUM Low-Level Inter-Task Communication
 472 //@subsection Index
 473
 474 //@index
 475 //* getOpName::  @cindex\s-+getOpName
 476 //* traceSendOp::  @cindex\s-+traceSendOp
 477 //* sendOp::  @cindex\s-+sendOp
 478 //* sendOp1::  @cindex\s-+sendOp1
 479 //* sendOp2::  @cindex\s-+sendOp2
 480 //* sendOpV::  @cindex\s-+sendOpV
 481 //* sendOpNV::  @cindex\s-+sendOpNV
 482 //* sendOpN::  @cindex\s-+sendOpN
 483 //* waitForPEOp::  @cindex\s-+waitForPEOp
 484 //* processUnexpectedMessage::  @cindex\s-+processUnexpectedMessage
 485 //* getOpcode::  @cindex\s-+getOpcode
 486 //* getOpcodeAndSender::  @cindex\s-+getOpcodeAndSender
 487 //* senderTask::  @cindex\s-+senderTask
 488 //* startUpPE::  @cindex\s-+startUpPE
 489 //* shutDownPE::  @cindex\s-+shutDownPE
 490 //@end index