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

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