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