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     int iteration = 0;
 105
 106     for (i = 0; i < BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE; i++)
 107       will_send[i] = 0;
 108     while (sent_count < (size - 1)) {
 109       iteration++;
 110
 111       /* loop k times to let more processes arrive before start sending data */
 112       for (k = 0; k < 3; k++) {
 113         for (i = 1; i < size; i++) {
 114           if ((already_sent[i] == 0) && (will_send[i] == 0)) {
 115             Request::iprobe(i, MPI_ANY_TAG, comm, &flag_array[i],
 116                        &temp_status_array[i]);
 117             if (flag_array[i] == 1) {
 118               will_send[i] = 1;
 119               Request::recv(&temp_buf[i], 1, MPI_CHAR, i, tag, comm,
 120                        &status);
 121               i = 0;
 122             }
 123           }
 124         }
 125       }
 126
 127       header_index = 0;
 128
 129       /* recv 1-byte message */
 130       for (i = 1; i < size; i++) {
 131         /* message arrive */
 132         if ((will_send[i] == 1) && (already_sent[i] == 0)) {
 133           header_buf[header_index] = i;
 134           header_index++;
 135           sent_count++;
 136
 137           /* will send in the next step */
 138           already_sent[i] = 1;
 139         }
 140       }
 141
 142       /* send header followed by data */
 143       if (header_index != 0) {
 144         header_buf[header_index] = -1;
 145         to = header_buf[0];
 146
 147         /* send header */
 148         Request::send(header_buf, header_size, MPI_INT, to, tag, comm);
 149
 150         /* send data - pipeline */
 151         for (i = 0; i < pipe_length; i++) {
 152           send_request_array[i] = Request::isend((char *)buf + (i * increment), segment, datatype, to, tag, comm);
 153         }
 154         Request::waitall((pipe_length), send_request_array, send_status_array);
 155       }
 156
 157
 158       /* end - send header followed by data */
 159       /* randomly MPI_Send to one node */
 160       /* this part has been commented out - performance-wise */
 161       else if (2 == 3) {
 162         /* search for the first node that never received data before */
 163         for (i = 0; i < size; i++) {
 164           if (i == root)
 165             continue;
 166           if (already_sent[i] == 0) {
 167             header_buf[0] = i;
 168             header_buf[1] = -1;
 169             to = i;
 170
 171             Request::send(header_buf, header_size, MPI_INT, to, tag, comm);
 172
 173             /* still need to chop data so that we can use the same non-root code */
 174             for (j = 0; j < pipe_length; j++) {
 175               Request::send((char *)buf + (j * increment), segment, datatype, to, tag, comm);
 176             }
 177           }
 178         }
 179       }
 180     }                           /* end - while (send_count < size-1) loop */
 181   }
 182
 183   /* end - root */
 184   /* none root */
 185   else {
 186
 187     /* send 1-byte message to root */
 188     Request::send(temp_buf, 1, MPI_CHAR, 0, tag, comm);
 189
 190     /* wait for header forward when required */
 191     request = Request::irecv(header_buf, header_size, MPI_INT, MPI_ANY_SOURCE, tag, comm);
 192     Request::wait(&request, MPI_STATUS_IGNORE);
 193
 194     /* search for where it is */
 195     int myordering = 0;
 196     while (rank != header_buf[myordering]) {
 197       myordering++;
 198     }
 199
 200     to = header_buf[myordering + 1];
 201     if (myordering == 0) {
 202       from = 0;
 203     } else {
 204       from = header_buf[myordering - 1];
 205     }
 206
 207     /* send header when required */
 208     if (to != -1) {
 209       Request::send(header_buf, header_size, MPI_INT, to, tag, comm);
 210     }
 211
 212     /* receive data */
 213
 214     for (i = 0; i < pipe_length; i++) {
 215       recv_request_array[i] = Request::irecv((char *)buf + (i * increment), segment, datatype, from, tag, comm);
 216     }
 217
 218     /* forward data */
 219     if (to != -1) {
 220       for (i = 0; i < pipe_length; i++) {
 221         Request::wait(&recv_request_array[i], MPI_STATUS_IGNORE);
 222         send_request_array[i] = Request::isend((char *)buf + (i * increment), segment, datatype, to, tag, comm);
 223       }
 224       Request::waitall((pipe_length), send_request_array, send_status_array);
 225     }
 226
 227     /* recv only */
 228     else {
 229       Request::waitall((pipe_length), recv_request_array, recv_status_array);
 230     }
 231   }
 232
 233   delete[] send_request_array;
 234   delete[] recv_request_array;
 235   delete[] send_status_array;
 236   delete[] recv_status_array;
 237   /* end pipeline */
 238
 239   if ((remainder != 0) && (count > segment)) {
 240     XBT_INFO("MPI_bcast_arrival_pattern_aware_wait: count is not divisible by block size, use default MPI_bcast for remainder.");
 241     colls::bcast((char*)buf + (pipe_length * increment), remainder, datatype, root, comm);
 242   }
 243
 244   return MPI_SUCCESS;
 245 }
 246
 247 }
 248 }