src/smpi/colls/allreduce-smp-binomial.c

   1 #include "colls_private.h"
   2 /* IMPLEMENTED BY PITCH PATARASUK
   3    Non-topoloty-specific (however, number of cores/node need to be changed)
   4    all-reduce operation designed for smp clusters
   5    It uses 2-layer communication: binomial for both intra-communication
   6    inter-communication*/
   7
   8 /* change number of core per smp-node
   9    we assume that number of core per process will be the same for all implementations */
  10 #ifndef NUM_CORE
  11 #define NUM_CORE 8
  12 #endif
  13
  14 /* ** NOTE **
  15    Use -DMPICH2 if this code does not compile.
  16    MPICH1 code also work on MPICH2 on our cluster and the performance are similar.
  17    This code assume commutative and associative reduce operator (MPI_SUM, MPI_MAX, etc).
  18 */
  19
  20 //#include <star-reduction.c>
  21
  22 /*
  23 This fucntion performs all-reduce operation as follow.
  24 1) binomial_tree reduce inside each SMP node
  25 2) binomial_tree reduce intra-communication between root of each SMP node
  26 3) binomial_tree bcast intra-communication between root of each SMP node
  27 4) binomial_tree bcast inside each SMP node
  28 */
  29 int smpi_coll_tuned_allreduce_smp_binomial(void *send_buf, void *recv_buf,
  30                                            int count, MPI_Datatype dtype,
  31                                            MPI_Op op, MPI_Comm comm)
  32 {
  33   int comm_size, rank;
  34   void *tmp_buf;
  35   int tag = COLL_TAG_ALLREDUCE;
  36   int mask, src, dst;
  37
  38
  39   int num_core = simcall_host_get_core(SIMIX_host_self());
  40   // do we use the default one or the number of cores in the platform ?
  41   // if the number of cores is one, the platform may be simulated with 1 node = 1 core
  42   if (num_core == 1) num_core = NUM_CORE;
  43   MPI_Status status;
  44
  45   comm_size=smpi_comm_size(comm);
  46   rank=smpi_comm_rank(comm);
  47   MPI_Aint extent, lb;
  48   smpi_datatype_extent(dtype, &lb, &extent);
  49   tmp_buf = (void *) xbt_malloc(count * extent);
  50
  51   /* compute intra and inter ranking */
  52   int intra_rank, inter_rank;
  53   intra_rank = rank % num_core;
  54   inter_rank = rank / num_core;
  55
  56   /* size of processes participate in intra communications =>
  57      should be equal to number of machines */
  58   int inter_comm_size = (comm_size + num_core - 1) / num_core;
  59
  60   /* copy input buffer to output buffer */
  61   smpi_mpi_sendrecv(send_buf, count, dtype, rank, tag,
  62                recv_buf, count, dtype, rank, tag, comm, &status);
  63
  64   /* start binomial reduce intra communication inside each SMP node */
  65   mask = 1;
  66   while (mask < num_core) {
  67     if ((mask & intra_rank) == 0) {
  68       src = (inter_rank * num_core) + (intra_rank | mask);
  69       if (src < comm_size) {
  70         smpi_mpi_recv(tmp_buf, count, dtype, src, tag, comm, &status);
  71         smpi_op_apply(op, tmp_buf, recv_buf, &count, &dtype);
  72       }
  73     } else {
  74       dst = (inter_rank * num_core) + (intra_rank & (~mask));
  75       smpi_mpi_send(recv_buf, count, dtype, dst, tag, comm);
  76       break;
  77     }
  78     mask <<= 1;
  79   }
  80
  81   /* start binomial reduce inter-communication between each SMP nodes:
  82      each node only have one process that can communicate to other nodes */
  83   if (intra_rank == 0) {
  84     mask = 1;
  85     while (mask < inter_comm_size) {
  86       if ((mask & inter_rank) == 0) {
  87         src = (inter_rank | mask) * num_core;
  88         if (src < comm_size) {
  89           smpi_mpi_recv(tmp_buf, count, dtype, src, tag, comm, &status);
  90           smpi_op_apply(op, tmp_buf, recv_buf, &count, &dtype);
  91         }
  92       } else {
  93         dst = (inter_rank & (~mask)) * num_core;
  94         smpi_mpi_send(recv_buf, count, dtype, dst, tag, comm);
  95         break;
  96       }
  97       mask <<= 1;
  98     }
  99   }
 100
 101   /* start binomial broadcast inter-communication between each SMP nodes:
 102      each node only have one process that can communicate to other nodes */
 103   if (intra_rank == 0) {
 104     mask = 1;
 105     while (mask < inter_comm_size) {
 106       if (inter_rank & mask) {
 107         src = (inter_rank - mask) * num_core;
 108         smpi_mpi_recv(recv_buf, count, dtype, src, tag, comm, &status);
 109         break;
 110       }
 111       mask <<= 1;
 112     }
 113     mask >>= 1;
 114
 115     while (mask > 0) {
 116       if (inter_rank < inter_comm_size) {
 117         dst = (inter_rank + mask) * num_core;
 118         if (dst < comm_size) {
 119           smpi_mpi_send(recv_buf, count, dtype, dst, tag, comm);
 120         }
 121       }
 122       mask >>= 1;
 123     }
 124   }
 125
 126   /* start binomial broadcast intra-communication inside each SMP nodes */
 127   int num_core_in_current_smp = num_core;
 128   if (inter_rank == (inter_comm_size - 1)) {
 129     num_core_in_current_smp = comm_size - (inter_rank * num_core);
 130   }
 131   mask = 1;
 132   while (mask < num_core_in_current_smp) {
 133     if (intra_rank & mask) {
 134       src = (inter_rank * num_core) + (intra_rank - mask);
 135       smpi_mpi_recv(recv_buf, count, dtype, src, tag, comm, &status);
 136       break;
 137     }
 138     mask <<= 1;
 139   }
 140   mask >>= 1;
 141
 142   while (mask > 0) {
 143     dst = (inter_rank * num_core) + (intra_rank + mask);
 144     if (dst < comm_size) {
 145       smpi_mpi_send(recv_buf, count, dtype, dst, tag, comm);
 146     }
 147     mask >>= 1;
 148   }
 149
 150   free(tmp_buf);
 151   return MPI_SUCCESS;
 152 }