Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
attempt to please jenkins
[simgrid.git] / src / smpi / smpi_coll.cpp
1 /* smpi_coll.c -- various optimized routing for collectives                   */
2
3 /* Copyright (c) 2009-2015. 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 <stdio.h>
10 #include <string.h>
11 #include <assert.h>
12
13 #include "private.h"
14 #include "colls/colls.h"
15 #include "simgrid/sg_config.h"
16
17 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_coll, smpi, "Logging specific to SMPI (coll)");
18
19 s_mpi_coll_description_t mpi_coll_gather_description[] = {
20   {"default", "gather default collective", (void*)smpi_mpi_gather},
21    COLL_GATHERS(COLL_DESCRIPTION, COLL_COMMA), {NULL, NULL, NULL}      /* this array must be NULL terminated */
22 };
23
24
25 s_mpi_coll_description_t mpi_coll_allgather_description[] = {
26   {"default",
27    "allgather default collective",
28    (void*)smpi_mpi_allgather},
29 COLL_ALLGATHERS(COLL_DESCRIPTION, COLL_COMMA),
30   {NULL, NULL, NULL}      /* this array must be NULL terminated */
31 };
32
33 s_mpi_coll_description_t mpi_coll_allgatherv_description[] = {
34   {"default",
35    "allgatherv default collective",
36    (void*)smpi_mpi_allgatherv},
37 COLL_ALLGATHERVS(COLL_DESCRIPTION, COLL_COMMA),
38   {NULL, NULL, NULL}      /* this array must be NULL terminated */
39 };
40
41 s_mpi_coll_description_t mpi_coll_allreduce_description[] = {
42   {"default",
43    "allreduce default collective",
44    (void*)smpi_mpi_allreduce},
45 COLL_ALLREDUCES(COLL_DESCRIPTION, COLL_COMMA),
46   {NULL, NULL, NULL}      /* this array must be NULL terminated */
47 };
48
49 s_mpi_coll_description_t mpi_coll_reduce_scatter_description[] = {
50   {"default",
51    "reduce_scatter default collective",
52    (void*)smpi_mpi_reduce_scatter},
53 COLL_REDUCE_SCATTERS(COLL_DESCRIPTION, COLL_COMMA),
54   {NULL, NULL, NULL}      /* this array must be NULL terminated */
55 };
56
57 s_mpi_coll_description_t mpi_coll_scatter_description[] = {
58   {"default",
59    "scatter default collective",
60    (void*)smpi_mpi_scatter},
61 COLL_SCATTERS(COLL_DESCRIPTION, COLL_COMMA),
62   {NULL, NULL, NULL}      /* this array must be NULL terminated */
63 };
64
65 s_mpi_coll_description_t mpi_coll_barrier_description[] = {
66   {"default",
67    "barrier default collective",
68    (void*)smpi_mpi_barrier},
69 COLL_BARRIERS(COLL_DESCRIPTION, COLL_COMMA),
70   {NULL, NULL, NULL}      /* this array must be NULL terminated */
71 };
72 s_mpi_coll_description_t mpi_coll_alltoall_description[] = {
73   {"default",
74    "Ompi alltoall default collective",
75    (void*)smpi_coll_tuned_alltoall_ompi2},
76 COLL_ALLTOALLS(COLL_DESCRIPTION, COLL_COMMA),
77   {"bruck",
78    "Alltoall Bruck (SG) collective",
79    (void*)smpi_coll_tuned_alltoall_bruck},
80   {"basic_linear",
81    "Alltoall basic linear (SG) collective",
82    (void*)smpi_coll_tuned_alltoall_basic_linear},
83   {NULL, NULL, NULL}      /* this array must be NULL terminated */
84 };
85
86 s_mpi_coll_description_t mpi_coll_alltoallv_description[] = {
87   {"default",
88    "Ompi alltoallv default collective",
89    (void*)smpi_coll_basic_alltoallv},
90 COLL_ALLTOALLVS(COLL_DESCRIPTION, COLL_COMMA),
91   {NULL, NULL, NULL}      /* this array must be NULL terminated */
92 };
93
94 s_mpi_coll_description_t mpi_coll_bcast_description[] = {
95   {"default",
96    "bcast default collective ",
97    (void*)smpi_mpi_bcast},
98 COLL_BCASTS(COLL_DESCRIPTION, COLL_COMMA),
99   {NULL, NULL, NULL}      /* this array must be NULL terminated */
100 };
101
102 s_mpi_coll_description_t mpi_coll_reduce_description[] = {
103   {"default",
104    "reduce default collective",
105    (void*)smpi_mpi_reduce},
106 COLL_REDUCES(COLL_DESCRIPTION, COLL_COMMA),
107   {NULL, NULL, NULL}      /* this array must be NULL terminated */
108 };
109
110
111
112 /** Displays the long description of all registered models, and quit */
113 void coll_help(const char *category, s_mpi_coll_description_t * table)
114 {
115   int i;
116   printf("Long description of the %s models accepted by this simulator:\n",
117          category);
118   for (i = 0; table[i].name; i++)
119     printf("  %s: %s\n", table[i].name, table[i].description);
120 }
121
122 int find_coll_description(s_mpi_coll_description_t * table,
123                            char *name, const char *desc)
124 {
125   int i;
126   char *name_list = NULL;
127   int selector_on=0;
128   if(name==NULL){//no argument provided, use active selector's algorithm
129     name=(char*)sg_cfg_get_string("smpi/coll_selector");
130     selector_on=1;
131   }
132   for (i = 0; table[i].name; i++)
133     if (!strcmp(name, table[i].name)) {
134       if (strcmp(table[i].name,"default"))
135         XBT_INFO("Switch to algorithm %s for collective %s",table[i].name,desc);
136       return i;
137     }
138
139   if(selector_on){
140     // collective seems not handled by the active selector, try with default one
141     name=(char*)"default";
142     for (i = 0; table[i].name; i++)
143       if (!strcmp(name, table[i].name)) {
144         return i;
145     }
146   }
147   if (!table[0].name)
148     xbt_die("No collective is valid for '%s'! This is a bug.",name);
149   name_list = xbt_strdup(table[0].name);
150   for (i = 1; table[i].name; i++) {
151     name_list = static_cast<char*>(xbt_realloc(name_list,
152                     strlen(name_list) + strlen(table[i].name) + 3));
153     strcat(name_list, ", ");
154     strcat(name_list, table[i].name);
155   }
156   xbt_die("Collective '%s' is invalid! Valid collectives are: %s.", name, name_list);
157   return -1;
158 }
159
160 int (*mpi_coll_gather_fun)(void *, int, MPI_Datatype, void*, int, MPI_Datatype, int root, MPI_Comm);
161 int (*mpi_coll_allgather_fun)(void *, int, MPI_Datatype, void*, int, MPI_Datatype, MPI_Comm);
162 int (*mpi_coll_allgatherv_fun)(void *, int, MPI_Datatype, void*, int*, int*, MPI_Datatype, MPI_Comm);
163 int (*mpi_coll_allreduce_fun)(void *sbuf, void *rbuf, int rcount, MPI_Datatype dtype, MPI_Op op, MPI_Comm comm);
164 int (*mpi_coll_alltoall_fun)(void *, int, MPI_Datatype, void*, int, MPI_Datatype, MPI_Comm);
165 int (*mpi_coll_alltoallv_fun)(void *, int*, int*, MPI_Datatype, void*, int*, int*, MPI_Datatype, MPI_Comm);
166 int (*mpi_coll_bcast_fun)(void *buf, int count, MPI_Datatype datatype, int root, MPI_Comm com);
167 int (*mpi_coll_reduce_fun)(void *buf, void *rbuf, int count, MPI_Datatype datatype, MPI_Op op, int root, MPI_Comm comm);
168 int (*mpi_coll_reduce_scatter_fun)(void *sbuf, void *rbuf, int *rcounts,MPI_Datatype dtype,MPI_Op  op,MPI_Comm  comm);
169 int (*mpi_coll_scatter_fun)(void *sendbuf, int sendcount, MPI_Datatype sendtype,void *recvbuf, int recvcount, MPI_Datatype recvtype,int root, MPI_Comm comm);
170 int (*mpi_coll_barrier_fun)(MPI_Comm comm);
171
172
173 int smpi_coll_tuned_alltoall_ompi2(void *sendbuf, int sendcount,
174                                    MPI_Datatype sendtype, void *recvbuf,
175                                    int recvcount, MPI_Datatype recvtype,
176                                    MPI_Comm comm)
177 {
178   int size, sendsize;  
179   size = smpi_comm_size(comm);  
180   sendsize = smpi_datatype_size(sendtype) * sendcount;  
181   if (sendsize < 200 && size > 12) {
182     return
183         smpi_coll_tuned_alltoall_bruck(sendbuf, sendcount, sendtype,
184                                        recvbuf, recvcount, recvtype,
185                                        comm);
186   } else if (sendsize < 3000) {
187     return
188         smpi_coll_tuned_alltoall_basic_linear(sendbuf, sendcount,
189                                               sendtype, recvbuf,
190                                               recvcount, recvtype, comm);
191   } else {
192     return
193         smpi_coll_tuned_alltoall_ring(sendbuf, sendcount, sendtype,
194                                       recvbuf, recvcount, recvtype,
195                                       comm);
196   }
197 }
198
199 /**
200  * Alltoall Bruck
201  *
202  * Openmpi calls this routine when the message size sent to each rank < 2000 bytes and size < 12
203  * FIXME: uh, check smpi_pmpi again, but this routine is called for > 12, not
204  * less...
205  **/
206 int smpi_coll_tuned_alltoall_bruck(void *sendbuf, int sendcount,
207                                    MPI_Datatype sendtype, void *recvbuf,
208                                    int recvcount, MPI_Datatype recvtype,
209                                    MPI_Comm comm)
210 {
211   int system_tag = 777;
212   int i, rank, size, err, count;
213   MPI_Aint lb;
214   MPI_Aint sendext = 0;
215   MPI_Aint recvext = 0;
216   MPI_Request *requests;
217
218   // FIXME: check implementation
219   rank = smpi_comm_rank(comm);
220   size = smpi_comm_size(comm);
221   XBT_DEBUG("<%d> algorithm alltoall_bruck() called.", rank);
222   smpi_datatype_extent(sendtype, &lb, &sendext);
223   smpi_datatype_extent(recvtype, &lb, &recvext);
224   /* Local copy from self */
225   err =
226       smpi_datatype_copy((char *)sendbuf + rank * sendcount * sendext, 
227                          sendcount, sendtype, 
228                          (char *)recvbuf + rank * recvcount * recvext,
229                          recvcount, recvtype);
230   if (err == MPI_SUCCESS && size > 1) {
231     /* Initiate all send/recv to/from others. */
232     requests = xbt_new(MPI_Request, 2 * (size - 1));
233     count = 0;
234     /* Create all receives that will be posted first */
235     for (i = 0; i < size; ++i) {
236       if (i == rank) {
237         XBT_DEBUG("<%d> skip request creation [src = %d, recvcount = %d]",
238                rank, i, recvcount);
239         continue;
240       }
241       requests[count] =
242           smpi_irecv_init((char *)recvbuf + i * recvcount * recvext, recvcount,
243                           recvtype, i, system_tag, comm);
244       count++;
245     }
246     /* Now create all sends  */
247     for (i = 0; i < size; ++i) {
248       if (i == rank) {
249         XBT_DEBUG("<%d> skip request creation [dst = %d, sendcount = %d]",
250                rank, i, sendcount);
251         continue;
252       }
253       requests[count] =
254           smpi_isend_init((char *)sendbuf + i * sendcount * sendext, sendcount,
255                           sendtype, i, system_tag, comm);
256       count++;
257     }
258     /* Wait for them all. */
259     smpi_mpi_startall(count, requests);
260     XBT_DEBUG("<%d> wait for %d requests", rank, count);
261     smpi_mpi_waitall(count, requests, MPI_STATUS_IGNORE);
262     for(i = 0; i < count; i++) {
263       if(requests[i]!=MPI_REQUEST_NULL) smpi_mpi_request_free(&requests[i]);
264     }
265     xbt_free(requests);
266   }
267   return MPI_SUCCESS;
268 }
269
270 /**
271  * Alltoall basic_linear (STARMPI:alltoall-simple)
272  **/
273 int smpi_coll_tuned_alltoall_basic_linear(void *sendbuf, int sendcount,
274                                           MPI_Datatype sendtype,
275                                           void *recvbuf, int recvcount,
276                                           MPI_Datatype recvtype,
277                                           MPI_Comm comm)
278 {
279   int system_tag = 888;
280   int i, rank, size, err, count;
281   MPI_Aint lb = 0, sendext = 0, recvext = 0;
282   MPI_Request *requests;
283
284   /* Initialize. */
285   rank = smpi_comm_rank(comm);
286   size = smpi_comm_size(comm);
287   XBT_DEBUG("<%d> algorithm alltoall_basic_linear() called.", rank);
288   smpi_datatype_extent(sendtype, &lb, &sendext);
289   smpi_datatype_extent(recvtype, &lb, &recvext);
290   /* simple optimization */
291   err = smpi_datatype_copy((char *)sendbuf + rank * sendcount * sendext, 
292                            sendcount, sendtype, 
293                            (char *)recvbuf + rank * recvcount * recvext, 
294                            recvcount, recvtype);
295   if (err == MPI_SUCCESS && size > 1) {
296     /* Initiate all send/recv to/from others. */
297     requests = xbt_new(MPI_Request, 2 * (size - 1));
298     /* Post all receives first -- a simple optimization */
299     count = 0;
300     for (i = (rank + 1) % size; i != rank; i = (i + 1) % size) {
301       requests[count] =
302           smpi_irecv_init((char *)recvbuf + i * recvcount * recvext, recvcount, 
303                           recvtype, i, system_tag, comm);
304       count++;
305     }
306     /* Now post all sends in reverse order
307      *   - We would like to minimize the search time through message queue
308      *     when messages actually arrive in the order in which they were posted.
309      * TODO: check the previous assertion
310      */
311     for (i = (rank + size - 1) % size; i != rank; i = (i + size - 1) % size) {
312       requests[count] =
313           smpi_isend_init((char *)sendbuf + i * sendcount * sendext, sendcount,
314                           sendtype, i, system_tag, comm);
315       count++;
316     }
317     /* Wait for them all. */
318     smpi_mpi_startall(count, requests);
319     XBT_DEBUG("<%d> wait for %d requests", rank, count);
320     smpi_mpi_waitall(count, requests, MPI_STATUS_IGNORE);
321     for(i = 0; i < count; i++) {
322       if(requests[i]!=MPI_REQUEST_NULL) smpi_mpi_request_free(&requests[i]);
323     }
324     xbt_free(requests);
325   }
326   return err;
327 }
328
329 int smpi_coll_basic_alltoallv(void *sendbuf, int *sendcounts,
330                               int *senddisps, MPI_Datatype sendtype,
331                               void *recvbuf, int *recvcounts,
332                               int *recvdisps, MPI_Datatype recvtype,
333                               MPI_Comm comm)
334 {
335   int system_tag = 889;
336   int i, rank, size, err, count;
337   MPI_Aint lb = 0, sendext = 0, recvext = 0;
338   MPI_Request *requests;
339
340   /* Initialize. */
341   rank = smpi_comm_rank(comm);
342   size = smpi_comm_size(comm);
343   XBT_DEBUG("<%d> algorithm basic_alltoallv() called.", rank);
344   smpi_datatype_extent(sendtype, &lb, &sendext);
345   smpi_datatype_extent(recvtype, &lb, &recvext);
346   /* Local copy from self */
347   err =
348       smpi_datatype_copy((char *)sendbuf + senddisps[rank] * sendext, 
349                          sendcounts[rank], sendtype,
350                          (char *)recvbuf + recvdisps[rank] * recvext, 
351                          recvcounts[rank], recvtype);
352   if (err == MPI_SUCCESS && size > 1) {
353     /* Initiate all send/recv to/from others. */
354     requests = xbt_new(MPI_Request, 2 * (size - 1));
355     count = 0;
356     /* Create all receives that will be posted first */
357     for (i = 0; i < size; ++i) {
358       if (i == rank || recvcounts[i] == 0) {
359         XBT_DEBUG
360             ("<%d> skip request creation [src = %d, recvcounts[src] = %d]",
361              rank, i, recvcounts[i]);
362         continue;
363       }
364       requests[count] =
365           smpi_irecv_init((char *)recvbuf + recvdisps[i] * recvext, 
366                           recvcounts[i], recvtype, i, system_tag, comm);
367       count++;
368     }
369     /* Now create all sends  */
370     for (i = 0; i < size; ++i) {
371       if (i == rank || sendcounts[i] == 0) {
372         XBT_DEBUG
373             ("<%d> skip request creation [dst = %d, sendcounts[dst] = %d]",
374              rank, i, sendcounts[i]);
375         continue;
376       }
377       requests[count] =
378           smpi_isend_init((char *)sendbuf + senddisps[i] * sendext, 
379                           sendcounts[i], sendtype, i, system_tag, comm);
380       count++;
381     }
382     /* Wait for them all. */
383     smpi_mpi_startall(count, requests);
384     XBT_DEBUG("<%d> wait for %d requests", rank, count);
385     smpi_mpi_waitall(count, requests, MPI_STATUS_IGNORE);
386     for(i = 0; i < count; i++) {
387       if(requests[i]!=MPI_REQUEST_NULL) smpi_mpi_request_free(&requests[i]);
388     }
389     xbt_free(requests);
390   }
391   return err;
392 }