ghc/rts/parallel/LLComms.c

   1 /* ----------------------------------------------------------------------------
   2  * Time-stamp: <Tue Mar 21 2000 20:23:41 Stardate: [-30]4539.24 hwloidl>
   3  * $Id: LLComms.c,v 1.3 2000/03/31 03:09:37 hwloidl Exp $
   4  *
   5  * GUM Low-Level Inter-Task Communication
   6  *
   7  * This module defines PVM Routines for PE-PE  communication.
   8  * P. Trinder, December 5th. 1994.
   9  * Adapted for the new RTS
  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  * waitForPEOp waits for a packet from global task {\em who} with the
 277  * OpCode {\em op}.  Other OpCodes are handled by processUnexpected.
 278  */
 279 //@cindex waitForPEOp
 280 rtsPacket
 281 waitForPEOp(OpCode op, GlobalTaskId who)
 282 {
 283   rtsPacket p;
 284   int nbytes;
 285   OpCode opCode;
 286   GlobalTaskId sender_id;
 287   rtsBool match;
 288
 289   do {
 290     IF_PAR_DEBUG(verbose,
 291                   fprintf(stderr,"waitForPEOp: op = %x (%s), who = %x\n",
 292                           op, getOpName(op), who));
 293
 294     while((p = pvm_recv(ANY_TASK,ANY_OPCODE)) < 0)
 295       pvm_perror("waitForPEOp: Waiting for PEOp");
 296
 297     pvm_bufinfo( p, &nbytes, &opCode, &sender_id );
 298     IF_PAR_DEBUG(verbose,
 299                  fprintf(stderr,"waitForPEOp: received: OpCode = %x, sender_id = %x",
 300                        opCode, getOpName(opCode), sender_id));
 301
 302     match = (op == ANY_OPCODE || op == opCode) &&
 303             (who == ANY_TASK || who == sender_id);
 304
 305     if (match)
 306       return(p);
 307
 308     /* Handle the unexpected OpCodes */
 309     processUnexpected(p);
 310
 311   } while(rtsTrue);
 312 }
 313
 314 /*
 315  * processUnexpected processes unexpected messages. If the message is a
 316  * FINISH it exits the prgram, and PVM gracefully
 317  */
 318 //@cindex processUnexpected
 319 void
 320 processUnexpected(rtsPacket packet)
 321 {
 322     OpCode opCode = getOpcode(packet);
 323
 324     IF_PAR_DEBUG(verbose,
 325                  GlobalTaskId sender = senderTask(packet);
 326                  fprintf(stderr,"== [%x] processUnexpected: Received %x (%s), sender %x\n",
 327                        mytid, opCode, getOpName(opCode), sender));
 328
 329     switch (opCode) {
 330     case PP_FINISH:
 331         stg_exit(EXIT_SUCCESS);
 332         break;
 333
 334       /* Anything we're not prepared to deal with.  Note that ALL OpCodes
 335          are discarded during termination -- this helps prevent bizarre
 336          race conditions.  */
 337       default:
 338         if (!GlobalStopPending) {
 339           GlobalTaskId errorTask;
 340           OpCode opCode;
 341
 342           getOpcodeAndSender(packet,&opCode,&errorTask);
 343           fprintf(stderr,"Task %x: Unexpected OpCode %x from %x in processUnexpected",
 344                 mytid, opCode, errorTask );
 345
 346           stg_exit(EXIT_FAILURE);
 347         }
 348     }
 349 }
 350
 351 //@cindex getOpcode
 352 OpCode
 353 getOpcode(rtsPacket p)
 354 {
 355   int nbytes;
 356   OpCode OpCode;
 357   GlobalTaskId sender_id;
 358   pvm_bufinfo(p, &nbytes, &OpCode, &sender_id);
 359   return(OpCode);
 360 }
 361
 362 //@cindex getOpcodeAndSender
 363 void
 364 getOpcodeAndSender(rtsPacket p, OpCode *opCodep, GlobalTaskId *senderIdp)
 365 {
 366   int nbytes;
 367   pvm_bufinfo(p, &nbytes, opCodep, senderIdp);
 368 }
 369
 370 //@cindex senderTask
 371 GlobalTaskId
 372 senderTask(rtsPacket p)
 373 {
 374   int nbytes;
 375   OpCode opCode;
 376   GlobalTaskId sender_id;
 377   pvm_bufinfo(p, &nbytes, &opCode, &sender_id);
 378   return(sender_id);
 379 }
 380
 381 /*
 382  * PEStartUp does the low-level comms specific startup stuff for a
 383  * PE. It initialises the comms system, joins the appropriate groups,
 384  * synchronises with the other PEs. Receives and records in a global
 385  * variable the task-id of SysMan. If this is the main thread (discovered
 386  * in main.lc), identifies itself to SysMan. Finally it receives
 387  * from SysMan an array of the Global Task Ids of each PE, which is
 388  * returned as the value of the function.
 389  */
 390
 391 //@cindex startUpPE
 392 GlobalTaskId *
 393 startUpPE(nat nPEs)
 394 {
 395   int i;
 396   rtsPacket addr;
 397   long *buffer = (long *) stgMallocBytes(sizeof(long) * nPEs,
 398                                          "PEStartUp (buffer)");
 399   GlobalTaskId *thePEs = (GlobalTaskId *)
 400     stgMallocBytes(sizeof(GlobalTaskId) * nPEs,
 401                    "PEStartUp (PEs)");
 402
 403   mytid = _my_gtid;     /* Initialise PVM and get task id into global var.*/
 404
 405   IF_PAR_DEBUG(verbose,
 406                fprintf(stderr,"== [%x] PEStartup: Task id = [%x], No. PEs = %d \n",
 407                        mytid, mytid, nPEs));
 408   checkComms(pvm_joingroup(PEGROUP), "PEStartup");
 409   IF_PAR_DEBUG(verbose,
 410                fprintf(stderr,"== [%x] PEStartup: Joined PEGROUP\n", mytid));
 411   checkComms(pvm_joingroup(PECTLGROUP), "PEStartup");
 412   IF_PAR_DEBUG(verbose,
 413                fprintf(stderr,"== [%x] PEStartup: Joined PECTLGROUP\n", mytid));
 414   checkComms(pvm_barrier(PECTLGROUP, nPEs+1), "PEStartup");
 415   IF_PAR_DEBUG(verbose,
 416                fprintf(stderr,"== [%x] PEStartup, Passed PECTLGROUP barrier\n", mytid));
 417
 418   addr = waitForPEOp(PP_SYSMAN_TID, ANY_GLOBAL_TASK);
 419   SysManTask = senderTask(addr);
 420   if (IAmMainThread) {         /* Main Thread Identifies itself to SysMan */
 421     pvm_initsend(PvmDataDefault);
 422     pvm_send(SysManTask, PP_MAIN_TASK);
 423   }
 424   IF_PAR_DEBUG(verbose,
 425                fprintf(stderr,"== [%x] Thread waits for %s\n",
 426                        mytid, getOpName(PP_PETIDS)));
 427   addr = waitForPEOp(PP_PETIDS, ANY_GLOBAL_TASK);
 428   GetArgs(buffer, nPEs);
 429   for (i = 0; i < nPEs; ++i) {
 430     thePEs[i] = (GlobalTaskId) buffer[i];
 431     IF_PAR_DEBUG(verbose,
 432                  fprintf(stderr,"== [%x] PEStartup: PEs[%d] = %x \n",
 433                          mytid, i, thePEs[i]));
 434   }
 435   free(buffer);
 436   return thePEs;
 437 }
 438
 439 /*
 440  * PEShutdown does the low-level comms-specific shutdown stuff for a
 441  * single PE. It leaves the groups and then exits from pvm.
 442  */
 443 //@cindex shutDownPE
 444 void
 445 shutDownPE(void)
 446 {
 447   IF_PAR_DEBUG(verbose,
 448                fprintf(stderr, "== [%x] PEshutdown\n", mytid));
 449
 450   checkComms(pvm_lvgroup(PEGROUP),"PEShutDown");
 451   checkComms(pvm_lvgroup(PECTLGROUP),"PEShutDown");
 452   checkComms(pvm_exit(),"PEShutDown");
 453 }
 454
 455 #endif /* PAR -- whole file */
 456
 457 //@node Index,  , Auxiliary functions, GUM Low-Level Inter-Task Communication
 458 //@subsection Index
 459
 460 //@index
 461 //* getOpName::  @cindex\s-+getOpName
 462 //* traceSendOp::  @cindex\s-+traceSendOp
 463 //* sendOp::  @cindex\s-+sendOp
 464 //* sendOp1::  @cindex\s-+sendOp1
 465 //* sendOp2::  @cindex\s-+sendOp2
 466 //* sendOpV::  @cindex\s-+sendOpV
 467 //* sendOpNV::  @cindex\s-+sendOpNV
 468 //* sendOpN::  @cindex\s-+sendOpN
 469 //* waitForPEOp::  @cindex\s-+waitForPEOp
 470 //* processUnexpected::  @cindex\s-+processUnexpected
 471 //* getOpcode::  @cindex\s-+getOpcode
 472 //* getOpcodeAndSender::  @cindex\s-+getOpcodeAndSender
 473 //* senderTask::  @cindex\s-+senderTask
 474 //* startUpPE::  @cindex\s-+startUpPE
 475 //* shutDownPE::  @cindex\s-+shutDownPE
 476 //@end index