src/smpi/colls/bcast/bcast-arrival-pattern-aware-wait.cpp

   1 /* Copyright (c) 2013-2022. The SimGrid Team.
   2  * All rights reserved.                                                     */
   3
   4 /* This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the license (GNU LGPL) which comes with this package. */
   6
   7 #include "../colls_private.hpp"
   8
   9 int bcast_arrival_pattern_aware_wait_segment_size_in_byte = 8192;
  10
  11 #ifndef BCAST_ARRIVAL_PATTERN_AWARE_HEADER_SIZE
  12 #define BCAST_ARRIVAL_PATTERN_AWARE_HEADER_SIZE 1024
  13 #endif
  14
  15 #ifndef BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE
  16 #define BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE 128
  17 #endif
  18 namespace simgrid{
  19 namespace smpi{
  20 /* Non-topology-specific pipelined linear-bcast function */
  21 int bcast__arrival_pattern_aware_wait(void *buf, int count,
  22                                       MPI_Datatype datatype,
  23                                       int root, MPI_Comm comm)
  24 {
  25   MPI_Status status;
  26   MPI_Request request;
  27
  28   MPI_Status temp_status_array[BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE];
  29
  30   int rank, size;
  31   int i, j, k;
  32   int tag = -COLL_TAG_BCAST;
  33   int will_send[BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE];
  34
  35   int sent_count;
  36   int header_index;
  37   int flag_array[BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE];
  38   int already_sent[BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE];
  39
  40   int header_buf[BCAST_ARRIVAL_PATTERN_AWARE_HEADER_SIZE];
  41   char temp_buf[BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE];
  42
  43   int max_node = BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE;
  44   int header_size = BCAST_ARRIVAL_PATTERN_AWARE_HEADER_SIZE;
  45
  46   MPI_Aint extent;
  47   extent = datatype->get_extent();
  48
  49   /* source and destination */
  50   int to, from;
  51
  52
  53
  54   rank = comm->rank();
  55   size = comm->size();
  56
  57
  58   /* segment is segment size in number of elements (not bytes) */
  59   int segment = bcast_arrival_pattern_aware_wait_segment_size_in_byte / extent;
  60   segment =  segment == 0 ? 1 :segment;
  61   /* pipeline length */
  62   int pipe_length = count / segment;
  63
  64   /* use for buffer offset for sending and receiving data = segment size in byte */
  65   int increment = segment * extent;
  66
  67   /* if the input size is not divisible by segment size =>
  68      the small remainder will be done with native implementation */
  69   int remainder = count % segment;
  70
  71   /* if root is not zero send to rank zero first
  72      this can be modified to make it faster by using logical src, dst.
  73    */
  74   if (root != 0) {
  75     if (rank == root) {
  76       Request::send(buf, count, datatype, 0, tag, comm);
  77     } else if (rank == 0) {
  78       Request::recv(buf, count, datatype, root, tag, comm, &status);
  79     }
  80   }
  81
  82
  83   /* value == 0 means root has not send data (or header) to the node yet */
  84   for (i = 0; i < max_node; i++) {
  85     already_sent[i] = 0;
  86   }
  87
  88   /* when a message is smaller than a block size => no pipeline */
  89   if (count <= segment) {
  90     segment = count;
  91     pipe_length = 1;
  92   }
  93
  94   /* start pipeline bcast */
  95
  96   auto* send_request_array = new MPI_Request[size + pipe_length];
  97   auto* recv_request_array = new MPI_Request[size + pipe_length];
  98   auto* send_status_array  = new MPI_Status[size + pipe_length];
  99   auto* recv_status_array  = new MPI_Status[size + pipe_length];
 100
 101   /* root */
 102   if (rank == 0) {
 103     sent_count = 0;
 104
 105     for (i = 0; i < BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE; i++)
 106       will_send[i] = 0;
 107     while (sent_count < (size - 1)) {
 108       /* loop k times to let more processes arrive before start sending data */
 109       for (k = 0; k < 3; k++) {
 110         for (i = 1; i < size; i++) {
 111           if ((already_sent[i] == 0) && (will_send[i] == 0)) {
 112             Request::iprobe(i, MPI_ANY_TAG, comm, &flag_array[i],
 113                        &temp_status_array[i]);
 114             if (flag_array[i] == 1) {
 115               will_send[i] = 1;
 116               Request::recv(&temp_buf[i], 1, MPI_CHAR, i, tag, comm,
 117                        &status);
 118               i = 0;
 119             }
 120           }
 121         }
 122       }
 123
 124       header_index = 0;
 125
 126       /* recv 1-byte message */
 127       for (i = 1; i < size; i++) {
 128         /* message arrive */
 129         if ((will_send[i] == 1) && (already_sent[i] == 0)) {
 130           header_buf[header_index] = i;
 131           header_index++;
 132           sent_count++;
 133
 134           /* will send in the next step */
 135           already_sent[i] = 1;
 136         }
 137       }
 138
 139       /* send header followed by data */
 140       if (header_index != 0) {
 141         header_buf[header_index] = -1;
 142         to = header_buf[0];
 143
 144         /* send header */
 145         Request::send(header_buf, header_size, MPI_INT, to, tag, comm);
 146
 147         /* send data - pipeline */
 148         for (i = 0; i < pipe_length; i++) {
 149           send_request_array[i] = Request::isend((char *)buf + (i * increment), segment, datatype, to, tag, comm);
 150         }
 151         Request::waitall((pipe_length), send_request_array, send_status_array);
 152       }
 153
 154
 155       /* end - send header followed by data */
 156       /* randomly MPI_Send to one node */
 157       /* this part has been commented out - performance-wise */
 158       else if (2 == 3) {
 159         /* search for the first node that never received data before */
 160         for (i = 0; i < size; i++) {
 161           if (i == root)
 162             continue;
 163           if (already_sent[i] == 0) {
 164             header_buf[0] = i;
 165             header_buf[1] = -1;
 166             to = i;
 167
 168             Request::send(header_buf, header_size, MPI_INT, to, tag, comm);
 169
 170             /* still need to chop data so that we can use the same non-root code */
 171             for (j = 0; j < pipe_length; j++) {
 172               Request::send((char *)buf + (j * increment), segment, datatype, to, tag, comm);
 173             }
 174           }
 175         }
 176       }
 177     }                           /* end - while (send_count < size-1) loop */
 178   }
 179
 180   /* end - root */
 181   /* none root */
 182   else {
 183
 184     /* send 1-byte message to root */
 185     Request::send(temp_buf, 1, MPI_CHAR, 0, tag, comm);
 186
 187     /* wait for header forward when required */
 188     request = Request::irecv(header_buf, header_size, MPI_INT, MPI_ANY_SOURCE, tag, comm);
 189     Request::wait(&request, MPI_STATUS_IGNORE);
 190
 191     /* search for where it is */
 192     int myordering = 0;
 193     while (rank != header_buf[myordering]) {
 194       myordering++;
 195     }
 196
 197     to = header_buf[myordering + 1];
 198     if (myordering == 0) {
 199       from = 0;
 200     } else {
 201       from = header_buf[myordering - 1];
 202     }
 203
 204     /* send header when required */
 205     if (to != -1) {
 206       Request::send(header_buf, header_size, MPI_INT, to, tag, comm);
 207     }
 208
 209     /* receive data */
 210
 211     for (i = 0; i < pipe_length; i++) {
 212       recv_request_array[i] = Request::irecv((char *)buf + (i * increment), segment, datatype, from, tag, comm);
 213     }
 214
 215     /* forward data */
 216     if (to != -1) {
 217       for (i = 0; i < pipe_length; i++) {
 218         Request::wait(&recv_request_array[i], MPI_STATUS_IGNORE);
 219         send_request_array[i] = Request::isend((char *)buf + (i * increment), segment, datatype, to, tag, comm);
 220       }
 221       Request::waitall((pipe_length), send_request_array, send_status_array);
 222     }
 223
 224     /* recv only */
 225     else {
 226       Request::waitall((pipe_length), recv_request_array, recv_status_array);
 227     }
 228   }
 229
 230   delete[] send_request_array;
 231   delete[] recv_request_array;
 232   delete[] send_status_array;
 233   delete[] recv_status_array;
 234   /* end pipeline */
 235
 236   if ((remainder != 0) && (count > segment)) {
 237     XBT_INFO("MPI_bcast_arrival_pattern_aware_wait: count is not divisible by block size, use default MPI_bcast for remainder.");
 238     colls::bcast((char*)buf + (pipe_length * increment), remainder, datatype, root, comm);
 239   }
 240
 241   return MPI_SUCCESS;
 242 }
 243
 244 }
 245 }