Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
e755f008c809daecaded25747dd115fd68e909db
[simgrid.git] / src / smpi / colls / smpi_mvapich2_selector.cpp
1 /* selector for collective algorithms based on mvapich decision logic */
2
3 /* Copyright (c) 2009-2019. The SimGrid Team.
4  * All rights reserved.                                                     */
5
6 /* This program is free software; you can redistribute it and/or modify it
7  * under the terms of the license (GNU LGPL) which comes with this package. */
8
9 #include "colls_private.hpp"
10
11 #include "smpi_mvapich2_selector_stampede.hpp"
12
13 namespace simgrid{
14 namespace smpi{
15
16
17 int Coll_alltoall_mvapich2::alltoall( void *sendbuf, int sendcount,
18     MPI_Datatype sendtype,
19     void* recvbuf, int recvcount,
20     MPI_Datatype recvtype,
21     MPI_Comm comm)
22 {
23
24   if(mv2_alltoall_table_ppn_conf==NULL)
25     init_mv2_alltoall_tables_stampede();
26
27   int sendtype_size, recvtype_size, comm_size;
28   char * tmp_buf = NULL;
29   int mpi_errno=MPI_SUCCESS;
30   int range = 0;
31   int range_threshold = 0;
32   int conf_index = 0;
33   comm_size =  comm->size();
34
35   sendtype_size=sendtype->size();
36   recvtype_size=recvtype->size();
37   long nbytes = sendtype_size * sendcount;
38
39   /* check if safe to use partial subscription mode */
40
41   /* Search for the corresponding system size inside the tuning table */
42   while ((range < (mv2_size_alltoall_tuning_table[conf_index] - 1)) &&
43       (comm_size > mv2_alltoall_thresholds_table[conf_index][range].numproc)) {
44       range++;
45   }
46   /* Search for corresponding inter-leader function */
47   while ((range_threshold < (mv2_alltoall_thresholds_table[conf_index][range].size_table - 1))
48       && (nbytes >
49   mv2_alltoall_thresholds_table[conf_index][range].algo_table[range_threshold].max)
50   && (mv2_alltoall_thresholds_table[conf_index][range].algo_table[range_threshold].max != -1)) {
51       range_threshold++;
52   }
53   MV2_Alltoall_function = mv2_alltoall_thresholds_table[conf_index][range].algo_table[range_threshold]
54                                                                                       .MV2_pt_Alltoall_function;
55
56   if(sendbuf != MPI_IN_PLACE) {
57       mpi_errno = MV2_Alltoall_function(sendbuf, sendcount, sendtype,
58           recvbuf, recvcount, recvtype,
59           comm);
60   } else {
61       range_threshold = 0;
62       if(nbytes <
63           mv2_alltoall_thresholds_table[conf_index][range].in_place_algo_table[range_threshold].min
64           ||nbytes > mv2_alltoall_thresholds_table[conf_index][range].in_place_algo_table[range_threshold].max
65       ) {
66           tmp_buf = (char *)smpi_get_tmp_sendbuffer( comm_size * recvcount * recvtype_size );
67           Datatype::copy((char *)recvbuf,
68               comm_size*recvcount, recvtype,
69               (char *)tmp_buf,
70               comm_size*recvcount, recvtype);
71
72           mpi_errno = MV2_Alltoall_function(tmp_buf, recvcount, recvtype,
73               recvbuf, recvcount, recvtype,
74               comm );
75           smpi_free_tmp_buffer(tmp_buf);
76       } else {
77           mpi_errno = MPIR_Alltoall_inplace_MV2(sendbuf, sendcount, sendtype,
78               recvbuf, recvcount, recvtype,
79               comm );
80       }
81   }
82
83
84   return (mpi_errno);
85 }
86
87 int Coll_allgather_mvapich2::allgather(void *sendbuf, int sendcount, MPI_Datatype sendtype,
88     void *recvbuf, int recvcount, MPI_Datatype recvtype,
89     MPI_Comm comm)
90 {
91
92   int mpi_errno = MPI_SUCCESS;
93   long nbytes = 0, comm_size, recvtype_size;
94   int range = 0;
95   int partial_sub_ok = 0;
96   int conf_index = 0;
97   int range_threshold = 0;
98   int is_two_level = 0;
99   MPI_Comm shmem_comm;
100   //MPI_Comm *shmem_commptr=NULL;
101   /* Get the size of the communicator */
102   comm_size = comm->size();
103   recvtype_size=recvtype->size();
104   nbytes = recvtype_size * recvcount;
105
106   if(mv2_allgather_table_ppn_conf==NULL)
107     init_mv2_allgather_tables_stampede();
108
109   if(comm->get_leaders_comm()==MPI_COMM_NULL){
110     comm->init_smp();
111   }
112
113   if (comm->is_uniform()){
114     shmem_comm = comm->get_intra_comm();
115     int local_size = shmem_comm->size();
116     int i          = 0;
117     if (mv2_allgather_table_ppn_conf[0] == -1) {
118       // Indicating user defined tuning
119       conf_index = 0;
120       goto conf_check_end;
121     }
122     do {
123       if (local_size == mv2_allgather_table_ppn_conf[i]) {
124         conf_index = i;
125         partial_sub_ok = 1;
126         break;
127       }
128       i++;
129     } while(i < mv2_allgather_num_ppn_conf);
130   }
131   conf_check_end:
132   if (partial_sub_ok != 1) {
133     conf_index = 0;
134   }
135
136   /* Search for the corresponding system size inside the tuning table */
137   while ((range < (mv2_size_allgather_tuning_table[conf_index] - 1)) &&
138       (comm_size >
139   mv2_allgather_thresholds_table[conf_index][range].numproc)) {
140       range++;
141   }
142   /* Search for corresponding inter-leader function */
143   while ((range_threshold <
144       (mv2_allgather_thresholds_table[conf_index][range].size_inter_table - 1))
145       && (nbytes > mv2_allgather_thresholds_table[conf_index][range].inter_leader[range_threshold].max)
146       && (mv2_allgather_thresholds_table[conf_index][range].inter_leader[range_threshold].max !=
147           -1)) {
148       range_threshold++;
149   }
150
151   /* Set inter-leader pt */
152   MV2_Allgatherction =
153       mv2_allgather_thresholds_table[conf_index][range].inter_leader[range_threshold].
154       MV2_pt_Allgatherction;
155
156   is_two_level =  mv2_allgather_thresholds_table[conf_index][range].two_level[range_threshold];
157
158   /* intracommunicator */
159   if(is_two_level ==1){
160     if(partial_sub_ok ==1){
161       if (comm->is_blocked()){
162       mpi_errno = MPIR_2lvl_Allgather_MV2(sendbuf, sendcount, sendtype,
163                             recvbuf, recvcount, recvtype,
164                             comm);
165       }else{
166       mpi_errno = Coll_allgather_mpich::allgather(sendbuf, sendcount, sendtype,
167                             recvbuf, recvcount, recvtype,
168                             comm);
169       }
170     } else {
171       mpi_errno = MPIR_Allgather_RD_MV2(sendbuf, sendcount, sendtype,
172           recvbuf, recvcount, recvtype,
173           comm);
174     }
175   } else if(MV2_Allgatherction == &MPIR_Allgather_Bruck_MV2
176       || MV2_Allgatherction == &MPIR_Allgather_RD_MV2
177       || MV2_Allgatherction == &MPIR_Allgather_Ring_MV2) {
178       mpi_errno = MV2_Allgatherction(sendbuf, sendcount, sendtype,
179           recvbuf, recvcount, recvtype,
180           comm);
181   }else{
182       return MPI_ERR_OTHER;
183   }
184
185   return mpi_errno;
186 }
187
188 int Coll_gather_mvapich2::gather(void *sendbuf,
189     int sendcnt,
190     MPI_Datatype sendtype,
191     void *recvbuf,
192     int recvcnt,
193     MPI_Datatype recvtype,
194     int root, MPI_Comm  comm)
195 {
196   if(mv2_gather_thresholds_table==NULL)
197     init_mv2_gather_tables_stampede();
198
199   int mpi_errno = MPI_SUCCESS;
200   int range = 0;
201   int range_threshold = 0;
202   int range_intra_threshold = 0;
203   long nbytes = 0;
204   int comm_size = comm->size();
205   int rank      = comm->rank();
206
207   if (rank == root) {
208     int recvtype_size = recvtype->size();
209     nbytes            = recvcnt * recvtype_size;
210   } else {
211     int sendtype_size = sendtype->size();
212     nbytes            = sendcnt * sendtype_size;
213   }
214
215   /* Search for the corresponding system size inside the tuning table */
216   while ((range < (mv2_size_gather_tuning_table - 1)) &&
217       (comm_size > mv2_gather_thresholds_table[range].numproc)) {
218       range++;
219   }
220   /* Search for corresponding inter-leader function */
221   while ((range_threshold < (mv2_gather_thresholds_table[range].size_inter_table - 1))
222       && (nbytes >
223   mv2_gather_thresholds_table[range].inter_leader[range_threshold].max)
224   && (mv2_gather_thresholds_table[range].inter_leader[range_threshold].max !=
225       -1)) {
226       range_threshold++;
227   }
228
229   /* Search for corresponding intra node function */
230   while ((range_intra_threshold < (mv2_gather_thresholds_table[range].size_intra_table - 1))
231       && (nbytes >
232   mv2_gather_thresholds_table[range].intra_node[range_intra_threshold].max)
233   && (mv2_gather_thresholds_table[range].intra_node[range_intra_threshold].max !=
234       -1)) {
235       range_intra_threshold++;
236   }
237
238     if (comm->is_blocked() ) {
239         // Set intra-node function pt for gather_two_level
240         MV2_Gather_intra_node_function =
241                               mv2_gather_thresholds_table[range].intra_node[range_intra_threshold].
242                               MV2_pt_Gather_function;
243         //Set inter-leader pt
244         MV2_Gather_inter_leader_function =
245                               mv2_gather_thresholds_table[range].inter_leader[range_threshold].
246                               MV2_pt_Gather_function;
247         // We call Gather function
248         mpi_errno =
249             MV2_Gather_inter_leader_function(sendbuf, sendcnt, sendtype, recvbuf, recvcnt,
250                                              recvtype, root, comm);
251
252     } else {
253   // Indeed, direct (non SMP-aware)gather is MPICH one
254   mpi_errno = Coll_gather_mpich::gather(sendbuf, sendcnt, sendtype,
255       recvbuf, recvcnt, recvtype,
256       root, comm);
257   }
258
259   return mpi_errno;
260 }
261
262 int Coll_allgatherv_mvapich2::allgatherv(void *sendbuf, int sendcount, MPI_Datatype sendtype,
263     void *recvbuf, int *recvcounts, int *displs,
264     MPI_Datatype recvtype, MPI_Comm  comm )
265 {
266   int mpi_errno = MPI_SUCCESS;
267   int range = 0, comm_size, total_count, recvtype_size, i;
268   int range_threshold = 0;
269   long nbytes = 0;
270
271   if(mv2_allgatherv_thresholds_table==NULL)
272     init_mv2_allgatherv_tables_stampede();
273
274   comm_size = comm->size();
275   total_count = 0;
276   for (i = 0; i < comm_size; i++)
277     total_count += recvcounts[i];
278
279   recvtype_size=recvtype->size();
280   nbytes = total_count * recvtype_size;
281
282   /* Search for the corresponding system size inside the tuning table */
283   while ((range < (mv2_size_allgatherv_tuning_table - 1)) &&
284       (comm_size > mv2_allgatherv_thresholds_table[range].numproc)) {
285       range++;
286   }
287   /* Search for corresponding inter-leader function */
288   while ((range_threshold < (mv2_allgatherv_thresholds_table[range].size_inter_table - 1))
289       && (nbytes >
290   comm_size * mv2_allgatherv_thresholds_table[range].inter_leader[range_threshold].max)
291   && (mv2_allgatherv_thresholds_table[range].inter_leader[range_threshold].max !=
292       -1)) {
293       range_threshold++;
294   }
295   /* Set inter-leader pt */
296   MV2_Allgatherv_function =
297       mv2_allgatherv_thresholds_table[range].inter_leader[range_threshold].
298       MV2_pt_Allgatherv_function;
299
300   if (MV2_Allgatherv_function == &MPIR_Allgatherv_Rec_Doubling_MV2)
301     {
302     if (not(comm_size & (comm_size - 1))) {
303       mpi_errno =
304           MPIR_Allgatherv_Rec_Doubling_MV2(sendbuf, sendcount, sendtype, recvbuf, recvcounts, displs, recvtype, comm);
305         } else {
306             mpi_errno =
307                 MPIR_Allgatherv_Bruck_MV2(sendbuf, sendcount,
308                     sendtype, recvbuf,
309                     recvcounts, displs,
310                     recvtype, comm);
311         }
312     } else {
313         mpi_errno =
314             MV2_Allgatherv_function(sendbuf, sendcount, sendtype,
315                 recvbuf, recvcounts, displs,
316                 recvtype, comm);
317     }
318
319   return mpi_errno;
320 }
321
322
323
324 int Coll_allreduce_mvapich2::allreduce(void *sendbuf,
325     void *recvbuf,
326     int count,
327     MPI_Datatype datatype,
328     MPI_Op op, MPI_Comm comm)
329 {
330
331   int mpi_errno = MPI_SUCCESS;
332   //int rank = 0,
333   int comm_size = 0;
334
335   comm_size = comm->size();
336   //rank = comm->rank();
337
338   if (count == 0) {
339       return MPI_SUCCESS;
340   }
341
342   if (mv2_allreduce_thresholds_table == NULL)
343     init_mv2_allreduce_tables_stampede();
344
345   /* check if multiple threads are calling this collective function */
346
347   MPI_Aint sendtype_size = 0;
348   long nbytes = 0;
349   int is_commutative = 0;
350   MPI_Aint true_lb, true_extent;
351
352   sendtype_size=datatype->size();
353   nbytes = count * sendtype_size;
354
355   datatype->extent(&true_lb, &true_extent);
356   //MPI_Op *op_ptr;
357   //is_commutative = op->is_commutative();
358
359   {
360     int range = 0, range_threshold = 0, range_threshold_intra = 0;
361     int is_two_level = 0;
362
363     /* Search for the corresponding system size inside the tuning table */
364     while ((range < (mv2_size_allreduce_tuning_table - 1)) &&
365         (comm_size > mv2_allreduce_thresholds_table[range].numproc)) {
366         range++;
367     }
368     /* Search for corresponding inter-leader function */
369     /* skip mcast poiters if mcast is not available */
370     if(mv2_allreduce_thresholds_table[range].mcast_enabled != 1){
371         while ((range_threshold < (mv2_allreduce_thresholds_table[range].size_inter_table - 1))
372             && ((mv2_allreduce_thresholds_table[range].
373                 inter_leader[range_threshold].MV2_pt_Allreducection
374                 == &MPIR_Allreduce_mcst_reduce_redscat_gather_MV2) ||
375                 (mv2_allreduce_thresholds_table[range].
376                     inter_leader[range_threshold].MV2_pt_Allreducection
377                     == &MPIR_Allreduce_mcst_reduce_two_level_helper_MV2)
378             )) {
379             range_threshold++;
380         }
381     }
382     while ((range_threshold < (mv2_allreduce_thresholds_table[range].size_inter_table - 1))
383         && (nbytes >
384     mv2_allreduce_thresholds_table[range].inter_leader[range_threshold].max)
385     && (mv2_allreduce_thresholds_table[range].inter_leader[range_threshold].max != -1)) {
386         range_threshold++;
387     }
388     if(mv2_allreduce_thresholds_table[range].is_two_level_allreduce[range_threshold] == 1){
389         is_two_level = 1;
390     }
391     /* Search for corresponding intra-node function */
392     while ((range_threshold_intra <
393         (mv2_allreduce_thresholds_table[range].size_intra_table - 1))
394         && (nbytes >
395     mv2_allreduce_thresholds_table[range].intra_node[range_threshold_intra].max)
396     && (mv2_allreduce_thresholds_table[range].intra_node[range_threshold_intra].max !=
397         -1)) {
398         range_threshold_intra++;
399     }
400
401     MV2_Allreducection = mv2_allreduce_thresholds_table[range].inter_leader[range_threshold]
402                                                                                 .MV2_pt_Allreducection;
403
404     MV2_Allreduce_intra_function = mv2_allreduce_thresholds_table[range].intra_node[range_threshold_intra]
405                                                                                     .MV2_pt_Allreducection;
406
407     /* check if mcast is ready, otherwise replace mcast with other algorithm */
408     if((MV2_Allreducection == &MPIR_Allreduce_mcst_reduce_redscat_gather_MV2)||
409         (MV2_Allreducection == &MPIR_Allreduce_mcst_reduce_two_level_helper_MV2)){
410         {
411           MV2_Allreducection = &MPIR_Allreduce_pt2pt_rd_MV2;
412         }
413         if(is_two_level != 1) {
414             MV2_Allreducection = &MPIR_Allreduce_pt2pt_rd_MV2;
415         }
416     }
417
418     if(is_two_level == 1){
419         // check if shm is ready, if not use other algorithm first
420         if (is_commutative) {
421           if(comm->get_leaders_comm()==MPI_COMM_NULL){
422             comm->init_smp();
423           }
424           mpi_errno = MPIR_Allreduce_two_level_MV2(sendbuf, recvbuf, count,
425                                                      datatype, op, comm);
426                 } else {
427         mpi_errno = MPIR_Allreduce_pt2pt_rd_MV2(sendbuf, recvbuf, count,
428             datatype, op, comm);
429         }
430     } else {
431         mpi_errno = MV2_Allreducection(sendbuf, recvbuf, count,
432             datatype, op, comm);
433     }
434   }
435
436   //comm->ch.intra_node_done=0;
437
438   return (mpi_errno);
439
440
441 }
442
443
444 int Coll_alltoallv_mvapich2::alltoallv(void *sbuf, int *scounts, int *sdisps,
445     MPI_Datatype sdtype,
446     void *rbuf, int *rcounts, int *rdisps,
447     MPI_Datatype rdtype,
448     MPI_Comm  comm
449 )
450 {
451
452   if (sbuf == MPI_IN_PLACE) {
453       return Coll_alltoallv_ompi_basic_linear::alltoallv(sbuf, scounts, sdisps, sdtype,
454           rbuf, rcounts, rdisps,rdtype,
455           comm);
456   } else     /* For starters, just keep the original algorithm. */
457   return Coll_alltoallv_ring::alltoallv(sbuf, scounts, sdisps, sdtype,
458       rbuf, rcounts, rdisps,rdtype,
459       comm);
460 }
461
462
463 int Coll_barrier_mvapich2::barrier(MPI_Comm  comm)
464 {
465   return Coll_barrier_mvapich2_pair::barrier(comm);
466 }
467
468
469
470
471 int Coll_bcast_mvapich2::bcast(void *buffer,
472     int count,
473     MPI_Datatype datatype,
474     int root, MPI_Comm comm)
475 {
476     int mpi_errno = MPI_SUCCESS;
477     int comm_size/*, rank*/;
478     int two_level_bcast = 1;
479     long nbytes = 0;
480     int range = 0;
481     int range_threshold = 0;
482     int range_threshold_intra = 0;
483     int is_homogeneous, is_contig;
484     MPI_Aint type_size;
485     //, position;
486     void *tmp_buf = NULL;
487     MPI_Comm shmem_comm;
488     //MPID_Datatype *dtp;
489
490     if (count == 0)
491         return MPI_SUCCESS;
492     if(comm->get_leaders_comm()==MPI_COMM_NULL){
493       comm->init_smp();
494     }
495     if (not mv2_bcast_thresholds_table)
496       init_mv2_bcast_tables_stampede();
497     comm_size = comm->size();
498     //rank = comm->rank();
499
500     is_contig=1;
501 /*    if (HANDLE_GET_KIND(datatype) == HANDLE_KIND_BUILTIN)*/
502 /*        is_contig = 1;*/
503 /*    else {*/
504 /*        MPID_Datatype_get_ptr(datatype, dtp);*/
505 /*        is_contig = dtp->is_contig;*/
506 /*    }*/
507
508     is_homogeneous = 1;
509
510     /* MPI_Type_size() might not give the accurate size of the packed
511      * datatype for heterogeneous systems (because of padding, encoding,
512      * etc). On the other hand, MPI_Pack_size() can become very
513      * expensive, depending on the implementation, especially for
514      * heterogeneous systems. We want to use MPI_Type_size() wherever
515      * possible, and MPI_Pack_size() in other places.
516      */
517     //if (is_homogeneous) {
518         type_size=datatype->size();
519
520    /* } else {
521         MPIR_Pack_size_impl(1, datatype, &type_size);
522     }*/
523     nbytes =  (count) * (type_size);
524
525     /* Search for the corresponding system size inside the tuning table */
526     while ((range < (mv2_size_bcast_tuning_table - 1)) &&
527            (comm_size > mv2_bcast_thresholds_table[range].numproc)) {
528         range++;
529     }
530     /* Search for corresponding inter-leader function */
531     while ((range_threshold < (mv2_bcast_thresholds_table[range].size_inter_table - 1))
532            && (nbytes >
533                mv2_bcast_thresholds_table[range].inter_leader[range_threshold].max)
534            && (mv2_bcast_thresholds_table[range].inter_leader[range_threshold].max != -1)) {
535         range_threshold++;
536     }
537
538     /* Search for corresponding intra-node function */
539     while ((range_threshold_intra <
540             (mv2_bcast_thresholds_table[range].size_intra_table - 1))
541            && (nbytes >
542                mv2_bcast_thresholds_table[range].intra_node[range_threshold_intra].max)
543            && (mv2_bcast_thresholds_table[range].intra_node[range_threshold_intra].max !=
544                -1)) {
545         range_threshold_intra++;
546     }
547
548     MV2_Bcast_function =
549         mv2_bcast_thresholds_table[range].inter_leader[range_threshold].
550         MV2_pt_Bcast_function;
551
552     MV2_Bcast_intra_node_function =
553         mv2_bcast_thresholds_table[range].
554         intra_node[range_threshold_intra].MV2_pt_Bcast_function;
555
556 /*    if (mv2_user_bcast_intra == NULL && */
557 /*            MV2_Bcast_intra_node_function == &MPIR_Knomial_Bcast_intra_node_MV2) {*/
558 /*            MV2_Bcast_intra_node_function = &MPIR_Shmem_Bcast_MV2;*/
559 /*    }*/
560
561     if (mv2_bcast_thresholds_table[range].inter_leader[range_threshold].
562         zcpy_pipelined_knomial_factor != -1) {
563         zcpy_knomial_factor =
564             mv2_bcast_thresholds_table[range].inter_leader[range_threshold].
565             zcpy_pipelined_knomial_factor;
566     }
567
568     if (mv2_pipelined_zcpy_knomial_factor != -1) {
569         zcpy_knomial_factor = mv2_pipelined_zcpy_knomial_factor;
570     }
571
572     if(MV2_Bcast_intra_node_function == NULL) {
573         /* if tuning table do not have any intra selection, set func pointer to
574         ** default one for mcast intra node */
575         MV2_Bcast_intra_node_function = &MPIR_Shmem_Bcast_MV2;
576     }
577
578     /* Set value of pipeline segment size */
579     bcast_segment_size = mv2_bcast_thresholds_table[range].bcast_segment_size;
580
581     /* Set value of inter node knomial factor */
582     mv2_inter_node_knomial_factor = mv2_bcast_thresholds_table[range].inter_node_knomial_factor;
583
584     /* Set value of intra node knomial factor */
585     mv2_intra_node_knomial_factor = mv2_bcast_thresholds_table[range].intra_node_knomial_factor;
586
587     /* Check if we will use a two level algorithm or not */
588     two_level_bcast =
589 #if defined(_MCST_SUPPORT_)
590         mv2_bcast_thresholds_table[range].is_two_level_bcast[range_threshold]
591         || comm->ch.is_mcast_ok;
592 #else
593         mv2_bcast_thresholds_table[range].is_two_level_bcast[range_threshold];
594 #endif
595      if (two_level_bcast == 1) {
596        if (not is_contig || not is_homogeneous) {
597          tmp_buf = (void*)smpi_get_tmp_sendbuffer(nbytes);
598
599          /*            position = 0;*/
600          /*            if (rank == root) {*/
601          /*                mpi_errno =*/
602          /*                    MPIR_Pack_impl(buffer, count, datatype, tmp_buf, nbytes, &position);*/
603          /*                if (mpi_errno)*/
604          /*                    MPIU_ERR_POP(mpi_errno);*/
605          /*            }*/
606         }
607 #ifdef CHANNEL_MRAIL_GEN2
608         if ((mv2_enable_zcpy_bcast == 1) &&
609               (&MPIR_Pipelined_Bcast_Zcpy_MV2 == MV2_Bcast_function)) {
610           if (not is_contig || not is_homogeneous) {
611             mpi_errno = MPIR_Pipelined_Bcast_Zcpy_MV2(tmp_buf, nbytes, MPI_BYTE, root, comm);
612             } else {
613                 mpi_errno = MPIR_Pipelined_Bcast_Zcpy_MV2(buffer, count, datatype,
614                                                  root, comm);
615             }
616         } else
617 #endif /* defined(CHANNEL_MRAIL_GEN2) */
618         {
619             shmem_comm = comm->get_intra_comm();
620             if (not is_contig || not is_homogeneous) {
621               MPIR_Bcast_tune_inter_node_helper_MV2(tmp_buf, nbytes, MPI_BYTE, root, comm);
622             } else {
623               MPIR_Bcast_tune_inter_node_helper_MV2(buffer, count, datatype, root, comm);
624             }
625
626             /* We are now done with the inter-node phase */
627
628
629                     root = INTRA_NODE_ROOT;
630
631                     if (not is_contig || not is_homogeneous) {
632                       mpi_errno = MV2_Bcast_intra_node_function(tmp_buf, nbytes, MPI_BYTE, root, shmem_comm);
633                 } else {
634                     mpi_errno = MV2_Bcast_intra_node_function(buffer, count,
635                                                               datatype, root, shmem_comm);
636
637                 }
638         }
639         /*        if (not is_contig || not is_homogeneous) {*/
640         /*            if (rank != root) {*/
641         /*                position = 0;*/
642         /*                mpi_errno = MPIR_Unpack_impl(tmp_buf, nbytes, &position, buffer,*/
643         /*                                             count, datatype);*/
644         /*            }*/
645         /*        }*/
646     } else {
647         /* We use Knomial for intra node */
648         MV2_Bcast_intra_node_function = &MPIR_Knomial_Bcast_intra_node_MV2;
649 /*        if (mv2_enable_shmem_bcast == 0) {*/
650             /* Fall back to non-tuned version */
651 /*            MPIR_Bcast_intra_MV2(buffer, count, datatype, root, comm);*/
652 /*        } else {*/
653             mpi_errno = MV2_Bcast_function(buffer, count, datatype, root,
654                                            comm);
655
656 /*        }*/
657     }
658
659
660     return mpi_errno;
661
662 }
663
664
665
666 int Coll_reduce_mvapich2::reduce( void *sendbuf,
667     void *recvbuf,
668     int count,
669     MPI_Datatype datatype,
670     MPI_Op op, int root, MPI_Comm comm)
671 {
672   if(mv2_reduce_thresholds_table == NULL)
673     init_mv2_reduce_tables_stampede();
674
675   int mpi_errno = MPI_SUCCESS;
676   int range = 0;
677   int range_threshold = 0;
678   int range_intra_threshold = 0;
679   int is_commutative, pof2;
680   int comm_size = 0;
681   long nbytes = 0;
682   int sendtype_size;
683   int is_two_level = 0;
684
685   comm_size = comm->size();
686   sendtype_size=datatype->size();
687   nbytes = count * sendtype_size;
688
689   if (count == 0)
690     return MPI_SUCCESS;
691
692   is_commutative = (op==MPI_OP_NULL || op->is_commutative());
693
694   /* find nearest power-of-two less than or equal to comm_size */
695   for( pof2 = 1; pof2 <= comm_size; pof2 <<= 1 );
696   pof2 >>=1;
697
698
699   /* Search for the corresponding system size inside the tuning table */
700   while ((range < (mv2_size_reduce_tuning_table - 1)) &&
701       (comm_size > mv2_reduce_thresholds_table[range].numproc)) {
702       range++;
703   }
704   /* Search for corresponding inter-leader function */
705   while ((range_threshold < (mv2_reduce_thresholds_table[range].size_inter_table - 1))
706       && (nbytes >
707   mv2_reduce_thresholds_table[range].inter_leader[range_threshold].max)
708   && (mv2_reduce_thresholds_table[range].inter_leader[range_threshold].max !=
709       -1)) {
710       range_threshold++;
711   }
712
713   /* Search for corresponding intra node function */
714   while ((range_intra_threshold < (mv2_reduce_thresholds_table[range].size_intra_table - 1))
715       && (nbytes >
716   mv2_reduce_thresholds_table[range].intra_node[range_intra_threshold].max)
717   && (mv2_reduce_thresholds_table[range].intra_node[range_intra_threshold].max !=
718       -1)) {
719       range_intra_threshold++;
720   }
721
722   /* Set intra-node function pt for reduce_two_level */
723   MV2_Reduce_intra_function =
724       mv2_reduce_thresholds_table[range].intra_node[range_intra_threshold].
725       MV2_pt_Reduce_function;
726   /* Set inter-leader pt */
727   MV2_Reduce_function =
728       mv2_reduce_thresholds_table[range].inter_leader[range_threshold].
729       MV2_pt_Reduce_function;
730
731   if(mv2_reduce_intra_knomial_factor<0)
732     {
733       mv2_reduce_intra_knomial_factor = mv2_reduce_thresholds_table[range].intra_k_degree;
734     }
735   if(mv2_reduce_inter_knomial_factor<0)
736     {
737       mv2_reduce_inter_knomial_factor = mv2_reduce_thresholds_table[range].inter_k_degree;
738     }
739   if(mv2_reduce_thresholds_table[range].is_two_level_reduce[range_threshold] == 1){
740       is_two_level = 1;
741   }
742   /* We call Reduce function */
743   if(is_two_level == 1)
744     {
745        if (is_commutative == 1) {
746          if(comm->get_leaders_comm()==MPI_COMM_NULL){
747            comm->init_smp();
748          }
749          mpi_errno = MPIR_Reduce_two_level_helper_MV2(sendbuf, recvbuf, count,
750                                            datatype, op, root, comm);
751         } else {
752       mpi_errno = MPIR_Reduce_binomial_MV2(sendbuf, recvbuf, count,
753           datatype, op, root, comm);
754       }
755     } else if(MV2_Reduce_function == &MPIR_Reduce_inter_knomial_wrapper_MV2 ){
756         if(is_commutative ==1)
757           {
758             mpi_errno = MV2_Reduce_function(sendbuf, recvbuf, count,
759                 datatype, op, root, comm);
760           } else {
761               mpi_errno = MPIR_Reduce_binomial_MV2(sendbuf, recvbuf, count,
762                   datatype, op, root, comm);
763           }
764     } else if(MV2_Reduce_function == &MPIR_Reduce_redscat_gather_MV2){
765         if (/*(HANDLE_GET_KIND(op) == HANDLE_KIND_BUILTIN) &&*/ (count >= pof2))
766           {
767             mpi_errno = MV2_Reduce_function(sendbuf, recvbuf, count,
768                 datatype, op, root, comm);
769           } else {
770               mpi_errno = MPIR_Reduce_binomial_MV2(sendbuf, recvbuf, count,
771                   datatype, op, root, comm);
772           }
773     } else {
774         mpi_errno = MV2_Reduce_function(sendbuf, recvbuf, count,
775             datatype, op, root, comm);
776     }
777
778
779   return mpi_errno;
780
781 }
782
783
784 int Coll_reduce_scatter_mvapich2::reduce_scatter(void *sendbuf, void *recvbuf, int *recvcnts,
785     MPI_Datatype datatype, MPI_Op op,
786     MPI_Comm comm)
787 {
788   int mpi_errno = MPI_SUCCESS;
789   int i = 0, comm_size = comm->size(), total_count = 0, type_size =
790       0, nbytes = 0;
791   int is_commutative = 0;
792   int* disps          = new int[comm_size];
793
794   if(mv2_red_scat_thresholds_table==NULL)
795     init_mv2_reduce_scatter_tables_stampede();
796
797   is_commutative=(op==MPI_OP_NULL || op->is_commutative());
798   for (i = 0; i < comm_size; i++) {
799       disps[i] = total_count;
800       total_count += recvcnts[i];
801   }
802
803   type_size=datatype->size();
804   nbytes = total_count * type_size;
805
806   if (is_commutative) {
807     int range           = 0;
808     int range_threshold = 0;
809
810       /* Search for the corresponding system size inside the tuning table */
811       while ((range < (mv2_size_red_scat_tuning_table - 1)) &&
812           (comm_size > mv2_red_scat_thresholds_table[range].numproc)) {
813           range++;
814       }
815       /* Search for corresponding inter-leader function */
816       while ((range_threshold < (mv2_red_scat_thresholds_table[range].size_inter_table - 1))
817           && (nbytes >
818       mv2_red_scat_thresholds_table[range].inter_leader[range_threshold].max)
819       && (mv2_red_scat_thresholds_table[range].inter_leader[range_threshold].max !=
820           -1)) {
821           range_threshold++;
822       }
823
824       /* Set inter-leader pt */
825       MV2_Red_scat_function =
826           mv2_red_scat_thresholds_table[range].inter_leader[range_threshold].
827           MV2_pt_Red_scat_function;
828
829       mpi_errno = MV2_Red_scat_function(sendbuf, recvbuf,
830           recvcnts, datatype,
831           op, comm);
832   } else {
833       int is_block_regular = 1;
834       for (i = 0; i < (comm_size - 1); ++i) {
835           if (recvcnts[i] != recvcnts[i+1]) {
836               is_block_regular = 0;
837               break;
838           }
839       }
840       int pof2 = 1;
841       while (pof2 < comm_size) pof2 <<= 1;
842       if (pof2 == comm_size && is_block_regular) {
843           /* noncommutative, pof2 size, and block regular */
844           MPIR_Reduce_scatter_non_comm_MV2(sendbuf, recvbuf,
845               recvcnts, datatype,
846               op, comm);
847       }
848       mpi_errno =  Coll_reduce_scatter_mpich_rdb::reduce_scatter(sendbuf, recvbuf,
849           recvcnts, datatype,
850           op, comm);
851   }
852   delete[] disps;
853   return mpi_errno;
854
855 }
856
857
858
859 int Coll_scatter_mvapich2::scatter(void *sendbuf,
860     int sendcnt,
861     MPI_Datatype sendtype,
862     void *recvbuf,
863     int recvcnt,
864     MPI_Datatype recvtype,
865     int root, MPI_Comm comm)
866 {
867   int range = 0, range_threshold = 0, range_threshold_intra = 0;
868   int mpi_errno = MPI_SUCCESS;
869   //   int mpi_errno_ret = MPI_SUCCESS;
870   int rank, nbytes, comm_size;
871   int partial_sub_ok = 0;
872   int conf_index = 0;
873      MPI_Comm shmem_comm;
874   //    MPID_Comm *shmem_commptr=NULL;
875   if(mv2_scatter_thresholds_table==NULL)
876     init_mv2_scatter_tables_stampede();
877
878   if(comm->get_leaders_comm()==MPI_COMM_NULL){
879     comm->init_smp();
880   }
881
882   comm_size = comm->size();
883
884   rank = comm->rank();
885
886   if (rank == root) {
887     int sendtype_size = sendtype->size();
888     nbytes            = sendcnt * sendtype_size;
889   } else {
890     int recvtype_size = recvtype->size();
891     nbytes            = recvcnt * recvtype_size;
892   }
893
894     // check if safe to use partial subscription mode
895     if (comm->is_uniform()) {
896
897         shmem_comm = comm->get_intra_comm();
898         if (mv2_scatter_table_ppn_conf[0] == -1) {
899             // Indicating user defined tuning
900             conf_index = 0;
901         }else{
902           int local_size = shmem_comm->size();
903           int i          = 0;
904             do {
905                 if (local_size == mv2_scatter_table_ppn_conf[i]) {
906                     conf_index = i;
907                     partial_sub_ok = 1;
908                     break;
909                 }
910                 i++;
911             } while(i < mv2_scatter_num_ppn_conf);
912         }
913     }
914
915   if (partial_sub_ok != 1) {
916       conf_index = 0;
917   }
918
919   /* Search for the corresponding system size inside the tuning table */
920   while ((range < (mv2_size_scatter_tuning_table[conf_index] - 1)) &&
921       (comm_size > mv2_scatter_thresholds_table[conf_index][range].numproc)) {
922       range++;
923   }
924   /* Search for corresponding inter-leader function */
925   while ((range_threshold < (mv2_scatter_thresholds_table[conf_index][range].size_inter_table - 1))
926       && (nbytes >
927   mv2_scatter_thresholds_table[conf_index][range].inter_leader[range_threshold].max)
928   && (mv2_scatter_thresholds_table[conf_index][range].inter_leader[range_threshold].max != -1)) {
929       range_threshold++;
930   }
931
932   /* Search for corresponding intra-node function */
933   while ((range_threshold_intra <
934       (mv2_scatter_thresholds_table[conf_index][range].size_intra_table - 1))
935       && (nbytes >
936   mv2_scatter_thresholds_table[conf_index][range].intra_node[range_threshold_intra].max)
937   && (mv2_scatter_thresholds_table[conf_index][range].intra_node[range_threshold_intra].max !=
938       -1)) {
939       range_threshold_intra++;
940   }
941
942   MV2_Scatter_function = mv2_scatter_thresholds_table[conf_index][range].inter_leader[range_threshold]
943                                                                                       .MV2_pt_Scatter_function;
944
945   if(MV2_Scatter_function == &MPIR_Scatter_mcst_wrap_MV2) {
946 #if defined(_MCST_SUPPORT_)
947       if(comm->ch.is_mcast_ok == 1
948           && mv2_use_mcast_scatter == 1
949           && comm->ch.shmem_coll_ok == 1) {
950           MV2_Scatter_function = &MPIR_Scatter_mcst_MV2;
951       } else
952 #endif /*#if defined(_MCST_SUPPORT_) */
953         {
954           if(mv2_scatter_thresholds_table[conf_index][range].inter_leader[range_threshold + 1].
955               MV2_pt_Scatter_function != NULL) {
956               MV2_Scatter_function = mv2_scatter_thresholds_table[conf_index][range].inter_leader[range_threshold + 1]
957                                                                                                   .MV2_pt_Scatter_function;
958           } else {
959               /* Fallback! */
960               MV2_Scatter_function = &MPIR_Scatter_MV2_Binomial;
961           }
962         }
963   }
964
965   if( (MV2_Scatter_function == &MPIR_Scatter_MV2_two_level_Direct) ||
966       (MV2_Scatter_function == &MPIR_Scatter_MV2_two_level_Binomial)) {
967        if( comm->is_blocked()) {
968              MV2_Scatter_intra_function = mv2_scatter_thresholds_table[conf_index][range].intra_node[range_threshold_intra]
969                                 .MV2_pt_Scatter_function;
970
971              mpi_errno =
972                    MV2_Scatter_function(sendbuf, sendcnt, sendtype,
973                                         recvbuf, recvcnt, recvtype, root,
974                                         comm);
975          } else {
976       mpi_errno = MPIR_Scatter_MV2_Binomial(sendbuf, sendcnt, sendtype,
977           recvbuf, recvcnt, recvtype, root,
978           comm);
979
980       }
981   } else {
982       mpi_errno = MV2_Scatter_function(sendbuf, sendcnt, sendtype,
983           recvbuf, recvcnt, recvtype, root,
984           comm);
985   }
986   return (mpi_errno);
987 }
988
989 }
990 }
991
992 void smpi_coll_cleanup_mvapich2()
993 {
994   if (mv2_alltoall_thresholds_table)
995     delete[] mv2_alltoall_thresholds_table[0];
996   delete[] mv2_alltoall_thresholds_table;
997   delete[] mv2_size_alltoall_tuning_table;
998   delete[] mv2_alltoall_table_ppn_conf;
999
1000   delete[] mv2_gather_thresholds_table;
1001   if (mv2_allgather_thresholds_table)
1002     delete[] mv2_allgather_thresholds_table[0];
1003   delete[] mv2_size_allgather_tuning_table;
1004   delete[] mv2_allgather_table_ppn_conf;
1005   delete[] mv2_allgather_thresholds_table;
1006
1007   delete[] mv2_allgatherv_thresholds_table;
1008   delete[] mv2_reduce_thresholds_table;
1009   delete[] mv2_red_scat_thresholds_table;
1010   delete[] mv2_allreduce_thresholds_table;
1011   delete[] mv2_bcast_thresholds_table;
1012   if (mv2_scatter_thresholds_table)
1013     delete[] mv2_scatter_thresholds_table[0];
1014   delete[] mv2_scatter_thresholds_table;
1015   delete[] mv2_size_scatter_tuning_table;
1016   delete[] mv2_scatter_table_ppn_conf;
1017 }