Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Merge branch 'master' of git+ssh://scm.gforge.inria.fr//gitroot/simgrid/simgrid
[simgrid.git] / src / smpi / smpi_coll.c
1 /* smpi_coll.c -- various optimized routing for collectives                   */
2
3 /* Copyright (c) 2009, 2010. 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 s_mpi_coll_description_t mpi_coll_gather_description[] = {
18   {"default",
19    "gather default collective",
20    smpi_mpi_gather},
21 COLL_GATHERS(COLL_DESCRIPTION, COLL_COMMA),
22   {NULL, NULL, NULL}      /* this array must be NULL terminated */
23 };
24
25
26 s_mpi_coll_description_t mpi_coll_allgather_description[] = {
27   {"default",
28    "allgather default collective",
29    smpi_mpi_allgather},
30 COLL_ALLGATHERS(COLL_DESCRIPTION, COLL_COMMA),
31   {NULL, NULL, NULL}      /* this array must be NULL terminated */
32 };
33
34 s_mpi_coll_description_t mpi_coll_allgatherv_description[] = {
35   {"default",
36    "allgatherv default collective",
37    smpi_mpi_allgatherv},
38 COLL_ALLGATHERVS(COLL_DESCRIPTION, COLL_COMMA),
39   {NULL, NULL, NULL}      /* this array must be NULL terminated */
40 };
41
42 s_mpi_coll_description_t mpi_coll_allreduce_description[] = {
43   {"default",
44    "allreduce default collective",
45    smpi_mpi_allreduce},
46 COLL_ALLREDUCES(COLL_DESCRIPTION, COLL_COMMA),
47   {NULL, NULL, NULL}      /* this array must be NULL terminated */
48 };
49
50 s_mpi_coll_description_t mpi_coll_reduce_scatter_description[] = {
51   {"default",
52    "reduce_scatter default collective",
53    smpi_mpi_reduce_scatter},
54 COLL_REDUCE_SCATTERS(COLL_DESCRIPTION, COLL_COMMA),
55   {NULL, NULL, NULL}      /* this array must be NULL terminated */
56 };
57
58
59 s_mpi_coll_description_t mpi_coll_alltoall_description[] = {
60   {"default",
61    "Ompi alltoall default collective",
62    smpi_coll_tuned_alltoall_ompi2},
63 COLL_ALLTOALLS(COLL_DESCRIPTION, COLL_COMMA),
64   {"bruck",
65    "Alltoall Bruck (SG) collective",
66    smpi_coll_tuned_alltoall_bruck},
67   {"basic_linear",
68    "Alltoall basic linear (SG) collective",
69    smpi_coll_tuned_alltoall_basic_linear},
70   {NULL, NULL, NULL}      /* this array must be NULL terminated */
71 };
72
73 s_mpi_coll_description_t mpi_coll_alltoallv_description[] = {
74   {"default",
75    "Ompi alltoallv default collective",
76    smpi_coll_basic_alltoallv},
77 COLL_ALLTOALLVS(COLL_DESCRIPTION, COLL_COMMA),
78   {NULL, NULL, NULL}      /* this array must be NULL terminated */
79 };
80
81 s_mpi_coll_description_t mpi_coll_bcast_description[] = {
82   {"default",
83    "bcast default collective",
84    smpi_mpi_bcast},
85 COLL_BCASTS(COLL_DESCRIPTION, COLL_COMMA),
86   {NULL, NULL, NULL}      /* this array must be NULL terminated */
87 };
88
89 s_mpi_coll_description_t mpi_coll_reduce_description[] = {
90   {"default",
91    "reduce default collective",
92    smpi_mpi_reduce},
93 COLL_REDUCES(COLL_DESCRIPTION, COLL_COMMA),
94   {NULL, NULL, NULL}      /* this array must be NULL terminated */
95 };
96
97
98
99 /** Displays the long description of all registered models, and quit */
100 void coll_help(const char *category, s_mpi_coll_description_t * table)
101 {
102   int i;
103   printf("Long description of the %s models accepted by this simulator:\n",
104          category);
105   for (i = 0; table[i].name; i++)
106     printf("  %s: %s\n", table[i].name, table[i].description);
107 }
108
109 int find_coll_description(s_mpi_coll_description_t * table,
110                            char *name)
111 {
112   int i;
113   char *name_list = NULL;
114   int selector_on=0;
115   if(name==NULL){//no argument provided, use active selector's algorithm
116     name=(char*)sg_cfg_get_string("smpi/coll_selector");
117     selector_on=1;
118   }
119   for (i = 0; table[i].name; i++)
120     if (!strcmp(name, table[i].name)) {
121       return i;
122     }
123
124   if(selector_on){
125     // collective seems not handled by the active selector, try with default one
126     name=(char*)"default";
127     for (i = 0; table[i].name; i++)
128       if (!strcmp(name, table[i].name)) {
129         return i;
130     }
131   }
132   name_list = strdup(table[0].name);
133   for (i = 1; table[i].name; i++) {
134     name_list =
135         xbt_realloc(name_list,
136                     strlen(name_list) + strlen(table[i].name) + 3);
137     strcat(name_list, ", ");
138     strcat(name_list, table[i].name);
139   }
140   xbt_die("Model '%s' is invalid! Valid models are: %s.", name, name_list);
141   return -1;
142 }
143
144 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_coll, smpi,
145                                 "Logging specific to SMPI (coll)");
146
147 int (*mpi_coll_gather_fun)(void *, int, MPI_Datatype, void*, int, MPI_Datatype, int root, MPI_Comm);
148 int (*mpi_coll_allgather_fun)(void *, int, MPI_Datatype, void*, int, MPI_Datatype, MPI_Comm);
149 int (*mpi_coll_allgatherv_fun)(void *, int, MPI_Datatype, void*, int*, int*, MPI_Datatype, MPI_Comm);
150 int (*mpi_coll_allreduce_fun)(void *sbuf, void *rbuf, int rcount, MPI_Datatype dtype, MPI_Op op, MPI_Comm comm);
151 int (*mpi_coll_alltoall_fun)(void *, int, MPI_Datatype, void*, int, MPI_Datatype, MPI_Comm);
152 int (*mpi_coll_alltoallv_fun)(void *, int*, int*, MPI_Datatype, void*, int*, int*, MPI_Datatype, MPI_Comm);
153 int (*mpi_coll_bcast_fun)(void *buf, int count, MPI_Datatype datatype, int root, MPI_Comm com);
154 int (*mpi_coll_reduce_fun)(void *buf, void *rbuf, int count, MPI_Datatype datatype, MPI_Op op, int root, MPI_Comm comm);
155 int (*mpi_coll_reduce_scatter_fun)(void *sbuf, void *rbuf, int *rcounts,MPI_Datatype dtype,MPI_Op  op,MPI_Comm  comm);
156 struct s_proc_tree {
157   int PROCTREE_A;
158   int numChildren;
159   int *child;
160   int parent;
161   int me;
162   int root;
163   int isRoot;
164 };
165 typedef struct s_proc_tree *proc_tree_t;
166
167 /**
168  * alloc and init
169  **/
170 static proc_tree_t alloc_tree(int arity)
171 {
172   proc_tree_t tree;
173   int i;
174
175   tree = xbt_new(struct s_proc_tree, 1);
176   tree->PROCTREE_A = arity;
177   tree->isRoot = 0;
178   tree->numChildren = 0;
179   tree->child = xbt_new(int, arity);
180   for (i = 0; i < arity; i++) {
181     tree->child[i] = -1;
182   }
183   tree->root = -1;
184   tree->parent = -1;
185   return tree;
186 }
187
188 /**
189  * free
190  **/
191 static void free_tree(proc_tree_t tree)
192 {
193   xbt_free(tree->child);
194   xbt_free(tree);
195 }
196
197 /**
198  * Build the tree depending on a process rank (index) and the group size (extent)
199  * @param root the rank of the tree root
200  * @param rank the rank of the calling process
201  * @param size the total number of processes
202  **/
203 static void build_tree(int root, int rank, int size, proc_tree_t * tree)
204 {
205   int index = (rank - root + size) % size;
206   int firstChildIdx = index * (*tree)->PROCTREE_A + 1;
207   int i;
208
209   (*tree)->me = rank;
210   (*tree)->root = root;
211
212   for (i = 0; i < (*tree)->PROCTREE_A && firstChildIdx + i < size; i++) {
213     (*tree)->child[i] = (firstChildIdx + i + root) % size;
214     (*tree)->numChildren++;
215   }
216   if (rank == root) {
217     (*tree)->isRoot = 1;
218   } else {
219     (*tree)->isRoot = 0;
220     (*tree)->parent = (((index - 1) / (*tree)->PROCTREE_A) + root) % size;
221   }
222 }
223
224 /**
225  * bcast
226  **/
227 static void tree_bcast(void *buf, int count, MPI_Datatype datatype,
228                        MPI_Comm comm, proc_tree_t tree)
229 {
230   int system_tag = 999;         // used negative int but smpi_create_request() declares this illegal (to be checked)
231   int rank, i;
232   MPI_Request *requests;
233
234   rank = smpi_comm_rank(comm);
235   /* wait for data from my parent in the tree */
236   if (!tree->isRoot) {
237     XBT_DEBUG("<%d> tree_bcast(): i am not root: recv from %d, tag=%d)",
238            rank, tree->parent, system_tag + rank);
239     smpi_mpi_recv(buf, count, datatype, tree->parent, system_tag + rank,
240                   comm, MPI_STATUS_IGNORE);
241   }
242   requests = xbt_new(MPI_Request, tree->numChildren);
243   XBT_DEBUG("<%d> creates %d requests (1 per child)", rank,
244          tree->numChildren);
245   /* iniates sends to ranks lower in the tree */
246   for (i = 0; i < tree->numChildren; i++) {
247     if (tree->child[i] == -1) {
248       requests[i] = MPI_REQUEST_NULL;
249     } else {
250       XBT_DEBUG("<%d> send to <%d>, tag=%d", rank, tree->child[i],
251              system_tag + tree->child[i]);
252       requests[i] =
253           smpi_isend_init(buf, count, datatype, tree->child[i],
254                           system_tag + tree->child[i], comm);
255     }
256   }
257   smpi_mpi_startall(tree->numChildren, requests);
258   smpi_mpi_waitall(tree->numChildren, requests, MPI_STATUS_IGNORE);
259   xbt_free(requests);
260 }
261
262 /**
263  * anti-bcast
264  **/
265 static void tree_antibcast(void *buf, int count, MPI_Datatype datatype,
266                            MPI_Comm comm, proc_tree_t tree)
267 {
268   int system_tag = 999;         // used negative int but smpi_create_request() declares this illegal (to be checked)
269   int rank, i;
270   MPI_Request *requests;
271
272   rank = smpi_comm_rank(comm);
273   // everyone sends to its parent, except root.
274   if (!tree->isRoot) {
275     XBT_DEBUG("<%d> tree_antibcast(): i am not root: send to %d, tag=%d)",
276            rank, tree->parent, system_tag + rank);
277     smpi_mpi_send(buf, count, datatype, tree->parent, system_tag + rank,
278                   comm);
279   }
280   //every one receives as many messages as it has children
281   requests = xbt_new(MPI_Request, tree->numChildren);
282   XBT_DEBUG("<%d> creates %d requests (1 per child)", rank,
283          tree->numChildren);
284   for (i = 0; i < tree->numChildren; i++) {
285     if (tree->child[i] == -1) {
286       requests[i] = MPI_REQUEST_NULL;
287     } else {
288       XBT_DEBUG("<%d> recv from <%d>, tag=%d", rank, tree->child[i],
289              system_tag + tree->child[i]);
290       requests[i] =
291           smpi_irecv_init(buf, count, datatype, tree->child[i],
292                           system_tag + tree->child[i], comm);
293     }
294   }
295   smpi_mpi_startall(tree->numChildren, requests);
296   smpi_mpi_waitall(tree->numChildren, requests, MPI_STATUS_IGNORE);
297   xbt_free(requests);
298 }
299
300 /**
301  * bcast with a binary, ternary, or whatever tree ..
302  **/
303 void nary_tree_bcast(void *buf, int count, MPI_Datatype datatype, int root,
304                      MPI_Comm comm, int arity)
305 {
306   proc_tree_t tree = alloc_tree(arity);
307   int rank, size;
308
309   rank = smpi_comm_rank(comm);
310   size = smpi_comm_size(comm);
311   build_tree(root, rank, size, &tree);
312   tree_bcast(buf, count, datatype, comm, tree);
313   free_tree(tree);
314 }
315
316 /**
317  * barrier with a binary, ternary, or whatever tree ..
318  **/
319 void nary_tree_barrier(MPI_Comm comm, int arity)
320 {
321   proc_tree_t tree = alloc_tree(arity);
322   int rank, size;
323   char dummy = '$';
324
325   rank = smpi_comm_rank(comm);
326   size = smpi_comm_size(comm);
327   build_tree(0, rank, size, &tree);
328   tree_antibcast(&dummy, 1, MPI_CHAR, comm, tree);
329   tree_bcast(&dummy, 1, MPI_CHAR, comm, tree);
330   free_tree(tree);
331 }
332
333 int smpi_coll_tuned_alltoall_ompi2(void *sendbuf, int sendcount,
334                                    MPI_Datatype sendtype, void *recvbuf,
335                                    int recvcount, MPI_Datatype recvtype,
336                                    MPI_Comm comm)
337 {
338   int size, sendsize;   
339   size = smpi_comm_size(comm);  
340   sendsize = smpi_datatype_size(sendtype) * sendcount;  
341   if (sendsize < 200 && size > 12) {
342     return
343         smpi_coll_tuned_alltoall_bruck(sendbuf, sendcount, sendtype,
344                                        recvbuf, recvcount, recvtype,
345                                        comm);
346   } else if (sendsize < 3000) {
347     return
348         smpi_coll_tuned_alltoall_basic_linear(sendbuf, sendcount,
349                                               sendtype, recvbuf,
350                                               recvcount, recvtype, comm);
351   } else {
352     return
353         smpi_coll_tuned_alltoall_ring(sendbuf, sendcount, sendtype,
354                                       recvbuf, recvcount, recvtype,
355                                       comm);
356   }
357 }
358
359 /**
360  * Alltoall Bruck
361  *
362  * Openmpi calls this routine when the message size sent to each rank < 2000 bytes and size < 12
363  * FIXME: uh, check smpi_pmpi again, but this routine is called for > 12, not
364  * less...
365  **/
366 int smpi_coll_tuned_alltoall_bruck(void *sendbuf, int sendcount,
367                                    MPI_Datatype sendtype, void *recvbuf,
368                                    int recvcount, MPI_Datatype recvtype,
369                                    MPI_Comm comm)
370 {
371   int system_tag = 777;
372   int i, rank, size, err, count;
373   MPI_Aint lb;
374   MPI_Aint sendext = 0;
375   MPI_Aint recvext = 0;
376   MPI_Request *requests;
377
378   // FIXME: check implementation
379   rank = smpi_comm_rank(comm);
380   size = smpi_comm_size(comm);
381   XBT_DEBUG("<%d> algorithm alltoall_bruck() called.", rank);
382   err = smpi_datatype_extent(sendtype, &lb, &sendext);
383   err = smpi_datatype_extent(recvtype, &lb, &recvext);
384   /* Local copy from self */
385   err =
386       smpi_datatype_copy((char *)sendbuf + rank * sendcount * sendext, 
387                          sendcount, sendtype, 
388                          (char *)recvbuf + rank * recvcount * recvext,
389                          recvcount, recvtype);
390   if (err == MPI_SUCCESS && size > 1) {
391     /* Initiate all send/recv to/from others. */
392     requests = xbt_new(MPI_Request, 2 * (size - 1));
393     count = 0;
394     /* Create all receives that will be posted first */
395     for (i = 0; i < size; ++i) {
396       if (i == rank) {
397         XBT_DEBUG("<%d> skip request creation [src = %d, recvcount = %d]",
398                rank, i, recvcount);
399         continue;
400       }
401       requests[count] =
402           smpi_irecv_init((char *)recvbuf + i * recvcount * recvext, recvcount,
403                           recvtype, i, system_tag, comm);
404       count++;
405     }
406     /* Now create all sends  */
407     for (i = 0; i < size; ++i) {
408       if (i == rank) {
409         XBT_DEBUG("<%d> skip request creation [dst = %d, sendcount = %d]",
410                rank, i, sendcount);
411         continue;
412       }
413       requests[count] =
414           smpi_isend_init((char *)sendbuf + i * sendcount * sendext, sendcount,
415                           sendtype, i, system_tag, comm);
416       count++;
417     }
418     /* Wait for them all. */
419     smpi_mpi_startall(count, requests);
420     XBT_DEBUG("<%d> wait for %d requests", rank, count);
421     smpi_mpi_waitall(count, requests, MPI_STATUS_IGNORE);
422     xbt_free(requests);
423   }
424   return MPI_SUCCESS;
425 }
426
427 /**
428  * Alltoall basic_linear (STARMPI:alltoall-simple)
429  **/
430 int smpi_coll_tuned_alltoall_basic_linear(void *sendbuf, int sendcount,
431                                           MPI_Datatype sendtype,
432                                           void *recvbuf, int recvcount,
433                                           MPI_Datatype recvtype,
434                                           MPI_Comm comm)
435 {
436   int system_tag = 888;
437   int i, rank, size, err, count;
438   MPI_Aint lb = 0, sendext = 0, recvext = 0;
439   MPI_Request *requests;
440
441   /* Initialize. */
442   rank = smpi_comm_rank(comm);
443   size = smpi_comm_size(comm);
444   XBT_DEBUG("<%d> algorithm alltoall_basic_linear() called.", rank);
445   err = smpi_datatype_extent(sendtype, &lb, &sendext);
446   err = smpi_datatype_extent(recvtype, &lb, &recvext);
447   /* simple optimization */
448   err = smpi_datatype_copy((char *)sendbuf + rank * sendcount * sendext, 
449                            sendcount, sendtype, 
450                            (char *)recvbuf + rank * recvcount * recvext, 
451                            recvcount, recvtype);
452   if (err == MPI_SUCCESS && size > 1) {
453     /* Initiate all send/recv to/from others. */
454     requests = xbt_new(MPI_Request, 2 * (size - 1));
455     /* Post all receives first -- a simple optimization */
456     count = 0;
457     for (i = (rank + 1) % size; i != rank; i = (i + 1) % size) {
458       requests[count] =
459           smpi_irecv_init((char *)recvbuf + i * recvcount * recvext, recvcount, 
460                           recvtype, i, system_tag, comm);
461       count++;
462     }
463     /* Now post all sends in reverse order
464      *   - We would like to minimize the search time through message queue
465      *     when messages actually arrive in the order in which they were posted.
466      * TODO: check the previous assertion
467      */
468     for (i = (rank + size - 1) % size; i != rank; i = (i + size - 1) % size) {
469       requests[count] =
470           smpi_isend_init((char *)sendbuf + i * sendcount * sendext, sendcount,
471                           sendtype, i, system_tag, comm);
472       count++;
473     }
474     /* Wait for them all. */
475     smpi_mpi_startall(count, requests);
476     XBT_DEBUG("<%d> wait for %d requests", rank, count);
477     smpi_mpi_waitall(count, requests, MPI_STATUS_IGNORE);
478     xbt_free(requests);
479   }
480   return err;
481 }
482
483 int smpi_coll_basic_alltoallv(void *sendbuf, int *sendcounts,
484                               int *senddisps, MPI_Datatype sendtype,
485                               void *recvbuf, int *recvcounts,
486                               int *recvdisps, MPI_Datatype recvtype,
487                               MPI_Comm comm)
488 {
489   int system_tag = 889;
490   int i, rank, size, err, count;
491   MPI_Aint lb = 0, sendext = 0, recvext = 0;
492   MPI_Request *requests;
493
494   /* Initialize. */
495   rank = smpi_comm_rank(comm);
496   size = smpi_comm_size(comm);
497   XBT_DEBUG("<%d> algorithm basic_alltoallv() called.", rank);
498   err = smpi_datatype_extent(sendtype, &lb, &sendext);
499   err = smpi_datatype_extent(recvtype, &lb, &recvext);
500   /* Local copy from self */
501   err =
502       smpi_datatype_copy((char *)sendbuf + senddisps[rank] * sendext, 
503                          sendcounts[rank], sendtype,
504                          (char *)recvbuf + recvdisps[rank] * recvext, 
505                          recvcounts[rank], recvtype);
506   if (err == MPI_SUCCESS && size > 1) {
507     /* Initiate all send/recv to/from others. */
508     requests = xbt_new(MPI_Request, 2 * (size - 1));
509     count = 0;
510     /* Create all receives that will be posted first */
511     for (i = 0; i < size; ++i) {
512       if (i == rank || recvcounts[i] == 0) {
513         XBT_DEBUG
514             ("<%d> skip request creation [src = %d, recvcounts[src] = %d]",
515              rank, i, recvcounts[i]);
516         continue;
517       }
518       requests[count] =
519           smpi_irecv_init((char *)recvbuf + recvdisps[i] * recvext, 
520                           recvcounts[i], recvtype, i, system_tag, comm);
521       count++;
522     }
523     /* Now create all sends  */
524     for (i = 0; i < size; ++i) {
525       if (i == rank || sendcounts[i] == 0) {
526         XBT_DEBUG
527             ("<%d> skip request creation [dst = %d, sendcounts[dst] = %d]",
528              rank, i, sendcounts[i]);
529         continue;
530       }
531       requests[count] =
532           smpi_isend_init((char *)sendbuf + senddisps[i] * sendext, 
533                           sendcounts[i], sendtype, i, system_tag, comm);
534       count++;
535     }
536     /* Wait for them all. */
537     smpi_mpi_startall(count, requests);
538     XBT_DEBUG("<%d> wait for %d requests", rank, count);
539     smpi_mpi_waitall(count, requests, MPI_STATUS_IGNORE);
540     xbt_free(requests);
541   }
542   return err;
543 }