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

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