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

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