root/orte/test/system/ofi_big_msg.c

DEFINITIONS

1. send_callback
2. main

   1 #include "orte_config.h"
   2 
   3 #include <stdio.h>
   4 #include <signal.h>
   5 #include <math.h>
   6 
   7 #include "opal/runtime/opal_progress.h"
   8 
   9 #include "orte/util/proc_info.h"
  10 #include "orte/util/name_fns.h"
  11 #include "orte/runtime/orte_globals.h"
  12 #include "orte/mca/rml/rml.h"
  13 #include "orte/mca/errmgr/errmgr.h"
  14 
  15 #include "orte/runtime/runtime.h"
  16 #include "orte/runtime/orte_wait.h"
  17 
  18 #define MY_TAG 12345
  19 #define MAX_COUNT 3
  20 
  21 static bool msg_recvd;
  22 static volatile bool msg_active;
  23 
  24 static void send_callback(int status, orte_process_name_t *peer,
  25                           opal_buffer_t* buffer, orte_rml_tag_t tag,
  26                           void* cbdata)
  27 
  28 {
  29     OBJ_RELEASE(buffer);
  30     if (ORTE_SUCCESS != status) {
  31         exit(1);
  32     }
  33     msg_active = false;
  34 }
  35 
  36 
  37 int
  38 main(int argc, char *argv[]){
  39     int count;
  40     int msgsize;
  41     uint8_t *msg;
  42     int i, j, rc;
  43     orte_process_name_t peer;
  44     double maxpower;
  45     opal_buffer_t *buf;
  46     orte_rml_recv_cb_t blob;
  47     int sock_conduit_id = 1;  //use the first one
  48 
  49     /*
  50      * Init
  51      */
  52     orte_init(&argc, &argv, ORTE_PROC_NON_MPI);
  53 
  54     if (argc > 1) {
  55         count = atoi(argv[1]);
  56         if (count < 0) {
  57             count = INT_MAX-1;
  58         }
  59     } else {
  60         count = MAX_COUNT;
  61     }
  62 
  63     peer.jobid = ORTE_PROC_MY_NAME->jobid;
  64     peer.vpid = ORTE_PROC_MY_NAME->vpid + 1;
  65     if (peer.vpid == orte_process_info.num_procs) {
  66         peer.vpid = 0;
  67     }
  68 
  69     for (j=1; j < count+1; j++) {
  70         /* rank0 starts ring */
  71         if (ORTE_PROC_MY_NAME->vpid == 0) {
  72             /* setup the initiating buffer - put random sized message in it */
  73             buf = OBJ_NEW(opal_buffer_t);
  74 
  75             //maxpower = (double)(j%7);
  76             maxpower = (double)(j%8);
  77             msgsize = (int)pow(10.0, maxpower);
  78             //msgsize += 1401000;
  79             opal_output(0, "Ring %d message size %d bytes", j, msgsize);
  80             msg = (uint8_t*)malloc(msgsize);
  81             opal_dss.pack(buf, msg, msgsize, OPAL_BYTE);
  82             free(msg);
  83             orte_rml.send_buffer_transport_nb(sock_conduit_id,&peer, buf, MY_TAG, orte_rml_send_callback, NULL);
  84 
  85             /* wait for it to come around */
  86             OBJ_CONSTRUCT(&blob, orte_rml_recv_cb_t);
  87             blob.active = true;
  88             orte_rml.recv_buffer_nb(ORTE_NAME_WILDCARD, MY_TAG,
  89                                     ORTE_RML_NON_PERSISTENT,
  90                                     orte_rml_recv_callback, &blob);
  91             ORTE_WAIT_FOR_COMPLETION(blob.active);
  92             OBJ_DESTRUCT(&blob);
  93 
  94             opal_output(0, "%s Ring %d completed", ORTE_NAME_PRINT(ORTE_PROC_MY_NAME), j);
  95         } else {
  96             /* wait for msg */
  97             OBJ_CONSTRUCT(&blob, orte_rml_recv_cb_t);
  98             blob.active = true;
  99             orte_rml.recv_buffer_nb(ORTE_NAME_WILDCARD, MY_TAG,
 100                                     ORTE_RML_NON_PERSISTENT,
 101                                     orte_rml_recv_callback, &blob);
 102             ORTE_WAIT_FOR_COMPLETION(blob.active);
 103 
 104             opal_output(0, "%s received message %d from %s", ORTE_NAME_PRINT(ORTE_PROC_MY_NAME), j, ORTE_NAME_PRINT(&blob.name));
 105 
 106             /* send it along */
 107             buf = OBJ_NEW(opal_buffer_t);
 108             opal_dss.copy_payload(buf, &blob.data);
 109             OBJ_DESTRUCT(&blob);
 110             msg_active = true;
 111             orte_rml.send_buffer_transport_nb(sock_conduit_id,&peer, buf, MY_TAG, send_callback, NULL);
 112             ORTE_WAIT_FOR_COMPLETION(msg_active);
 113         }
 114     }
 115 
 116     orte_finalize();
 117 
 118     return 0;
 119 }