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

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