Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
add another ompi bcast algo
[simgrid.git] / src / smpi / colls / bcast-ompi-pipeline.c
1  #include "colls_private.h"
2
3
4 #define MAXTREEFANOUT 32
5
6 #define COLL_TUNED_COMPUTED_SEGCOUNT(SEGSIZE, TYPELNG, SEGCOUNT)        \
7     if( ((SEGSIZE) >= (TYPELNG)) &&                                     \
8         ((SEGSIZE) < ((TYPELNG) * (SEGCOUNT))) ) {                      \
9         size_t residual;                                                \
10         (SEGCOUNT) = (int)((SEGSIZE) / (TYPELNG));                      \
11         residual = (SEGSIZE) - (SEGCOUNT) * (TYPELNG);                  \
12         if( residual > ((TYPELNG) >> 1) )                               \
13             (SEGCOUNT)++;                                               \
14     }                                                                   \
15
16  typedef struct ompi_coll_tree_t {
17         int32_t tree_root;
18         int32_t tree_fanout;
19         int32_t tree_bmtree;
20         int32_t tree_prev;
21         int32_t tree_next[MAXTREEFANOUT];
22         int32_t tree_nextsize;
23     } ompi_coll_tree_t;
24
25     ompi_coll_tree_t*
26     ompi_coll_tuned_topo_build_chain( int fanout,
27                                      MPI_Comm com,
28                                      int root );
29
30 ompi_coll_tree_t*
31 ompi_coll_tuned_topo_build_chain( int fanout,
32                                   MPI_Comm comm,
33                                   int root )
34 {
35     int rank, size;
36     int srank; /* shifted rank */
37     int i,maxchainlen;
38     int mark,head,len;
39     ompi_coll_tree_t *chain;
40
41     XBT_DEBUG("coll:tuned:topo:build_chain fo %d rt %d", fanout, root);
42
43     /* 
44      * Get size and rank of the process in this communicator 
45      */
46     size = smpi_comm_size(comm);
47     rank = smpi_comm_rank(comm);
48
49     if( fanout < 1 ) {
50         XBT_DEBUG("coll:tuned:topo:build_chain WARNING invalid fanout of ZERO, forcing to 1 (pipeline)!");
51         fanout = 1;
52     }
53     if (fanout>MAXTREEFANOUT) {
54         XBT_DEBUG("coll:tuned:topo:build_chain WARNING invalid fanout %d bigger than max %d, forcing to max!", fanout, MAXTREEFANOUT);
55         fanout = MAXTREEFANOUT;
56     }
57
58     /*
59      * Allocate space for topology arrays if needed 
60      */
61     chain = (ompi_coll_tree_t*)malloc( sizeof(ompi_coll_tree_t) );
62     if (!chain) {
63         XBT_DEBUG("coll:tuned:topo:build_chain PANIC out of memory");
64         fflush(stdout);
65         return NULL;
66     }
67     chain->tree_root     = MPI_UNDEFINED;
68     chain->tree_nextsize = -1;
69     for(i=0;i<fanout;i++) chain->tree_next[i] = -1;
70
71     /* 
72      * Set root & numchain
73      */
74     chain->tree_root = root;
75     if( (size - 1) < fanout ) { 
76         chain->tree_nextsize = size-1;
77         fanout = size-1;
78     } else {
79         chain->tree_nextsize = fanout;
80     }
81     
82     /*
83      * Shift ranks
84      */
85     srank = rank - root;
86     if (srank < 0) srank += size;
87
88     /*
89      * Special case - fanout == 1
90      */
91     if( fanout == 1 ) {
92         if( srank == 0 ) chain->tree_prev = -1;
93         else chain->tree_prev = (srank-1+root)%size;
94
95         if( (srank + 1) >= size) {
96             chain->tree_next[0] = -1;
97             chain->tree_nextsize = 0;
98         } else {
99             chain->tree_next[0] = (srank+1+root)%size;
100             chain->tree_nextsize = 1;
101         }
102         return chain;
103     }
104
105     /* Let's handle the case where there is just one node in the communicator */
106     if( size == 1 ) {
107         chain->tree_next[0] = -1;
108         chain->tree_nextsize = 0;
109         chain->tree_prev = -1;
110         return chain;
111     }
112     /*
113      * Calculate maximum chain length
114      */
115     maxchainlen = (size-1) / fanout;
116     if( (size-1) % fanout != 0 ) {
117         maxchainlen++;
118         mark = (size-1)%fanout;
119     } else {
120         mark = fanout+1;
121     }
122
123     /*
124      * Find your own place in the list of shifted ranks
125      */
126     if( srank != 0 ) {
127         int column;
128         if( srank-1 < (mark * maxchainlen) ) {
129             column = (srank-1)/maxchainlen;
130             head = 1+column*maxchainlen;
131             len = maxchainlen;
132         } else {
133             column = mark + (srank-1-mark*maxchainlen)/(maxchainlen-1);
134             head = mark*maxchainlen+1+(column-mark)*(maxchainlen-1);
135             len = maxchainlen-1;
136         }
137
138         if( srank == head ) {
139             chain->tree_prev = 0; /*root*/
140         } else {
141             chain->tree_prev = srank-1; /* rank -1 */
142         }
143         if( srank == (head + len - 1) ) {
144             chain->tree_next[0] = -1;
145             chain->tree_nextsize = 0;
146         } else {
147             if( (srank + 1) < size ) {
148                 chain->tree_next[0] = srank+1;
149                 chain->tree_nextsize = 1;
150             } else {
151                 chain->tree_next[0] = -1;
152                 chain->tree_nextsize = 0;    
153             }
154         }
155     }
156     
157     /*
158      * Unshift values 
159      */
160     if( rank == root ) {
161         chain->tree_prev = -1;
162         chain->tree_next[0] = (root+1)%size;
163         for( i = 1; i < fanout; i++ ) {
164             chain->tree_next[i] = chain->tree_next[i-1] + maxchainlen;
165             if( i > mark ) {
166                 chain->tree_next[i]--;
167             }
168             chain->tree_next[i] %= size;
169         }
170         chain->tree_nextsize = fanout;
171     } else {
172         chain->tree_prev = (chain->tree_prev+root)%size;
173         if( chain->tree_next[0] != -1 ) {
174             chain->tree_next[0] = (chain->tree_next[0]+root)%size;
175         }
176     }
177
178     return chain;
179 }
180
181 smpi_coll_tuned_bcast_ompi_pipeline( void* buffer,
182                                       int original_count, 
183                                       MPI_Datatype datatype, 
184                                       int root,
185                                       MPI_Comm comm)
186 {
187     int count_by_segment = original_count;
188     size_t type_size;
189     int segsize;
190     //mca_coll_tuned_module_t *tuned_module = (mca_coll_tuned_module_t*) module;
191     //mca_coll_tuned_comm_t *data = tuned_module->tuned_data;
192     
193 //    return ompi_coll_tuned_bcast_intra_generic( buffer, count, datatype, root, comm, module,
194 //                                                count_by_segment, data->cached_pipeline );
195     ompi_coll_tree_t * tree = ompi_coll_tuned_topo_build_chain( 1, comm, root );
196     int err = 0, line, i;
197     int rank, size;
198     int segindex;
199     int num_segments; /* Number of segments */
200     int sendcount;    /* number of elements sent in this segment */ 
201     size_t realsegsize;
202     char *tmpbuf;
203     ptrdiff_t extent;
204     MPI_Request recv_reqs[2] = {MPI_REQUEST_NULL, MPI_REQUEST_NULL};
205     MPI_Request *send_reqs = NULL;
206     int req_index;
207     
208     /**
209      * Determine number of elements sent per operation.
210      */
211     type_size = smpi_datatype_size(datatype);
212
213     size = smpi_comm_size(comm);
214     rank = smpi_comm_rank(comm);
215     xbt_assert( size > 1 );
216
217
218     const double a_p16  = 3.2118e-6; /* [1 / byte] */
219     const double b_p16  = 8.7936;   
220     const double a_p64  = 2.3679e-6; /* [1 / byte] */
221     const double b_p64  = 1.1787;     
222     const double a_p128 = 1.6134e-6; /* [1 / byte] */
223     const double b_p128 = 2.1102;
224     size_t message_size;
225
226     /* else we need data size for decision function */
227     message_size = type_size * (unsigned long)original_count;   /* needed for decision */
228
229     if (size < (a_p128 * message_size + b_p128)) {
230             //Pipeline with 128KB segments 
231             segsize = 1024  << 7;
232     }else if (size < (a_p64 * message_size + b_p64)) {
233             // Pipeline with 64KB segments 
234             segsize = 1024 << 6;
235     }else if (size < (a_p16 * message_size + b_p16)) {
236             //Pipeline with 16KB segments 
237             segsize = 1024 << 4;
238     }
239
240     COLL_TUNED_COMPUTED_SEGCOUNT( segsize, type_size, count_by_segment );
241
242     XBT_DEBUG("coll:tuned:bcast_intra_pipeline rank %d ss %5d type_size %lu count_by_segment %d",
243                  smpi_comm_rank(comm), segsize, (unsigned long)type_size, count_by_segment);
244
245
246
247     extent = smpi_datatype_get_extent (datatype);
248     num_segments = (original_count + count_by_segment - 1) / count_by_segment;
249     realsegsize = count_by_segment * extent;
250     
251     /* Set the buffer pointers */
252     tmpbuf = (char *) buffer;
253
254     if( tree->tree_nextsize != 0 ) {
255         send_reqs = xbt_new(MPI_Request, tree->tree_nextsize  );
256     }
257
258     /* Root code */
259     if( rank == root ) {
260         /* 
261            For each segment:
262            - send segment to all children.
263              The last segment may have less elements than other segments.
264         */
265         sendcount = count_by_segment;
266         for( segindex = 0; segindex < num_segments; segindex++ ) {
267             if( segindex == (num_segments - 1) ) {
268                 sendcount = original_count - segindex * count_by_segment;
269             }
270             for( i = 0; i < tree->tree_nextsize; i++ ) { 
271                 send_reqs[i] = smpi_mpi_isend(tmpbuf, sendcount, datatype,
272                                          tree->tree_next[i], 
273                                          777, comm);
274            } 
275
276             /* complete the sends before starting the next sends */
277             smpi_mpi_waitall( tree->tree_nextsize, send_reqs, 
278                                          MPI_STATUSES_IGNORE );
279
280             /* update tmp buffer */
281             tmpbuf += realsegsize;
282
283         }
284     } 
285     
286     /* Intermediate nodes code */
287     else if( tree->tree_nextsize > 0 ) { 
288         /* 
289            Create the pipeline. 
290            1) Post the first receive
291            2) For segments 1 .. num_segments
292               - post new receive
293               - wait on the previous receive to complete
294               - send this data to children
295            3) Wait on the last segment
296            4) Compute number of elements in last segment.
297            5) Send the last segment to children
298          */
299         req_index = 0;
300         recv_reqs[req_index]=smpi_mpi_irecv(tmpbuf, count_by_segment, datatype,
301                            tree->tree_prev, 777,
302                            comm);
303         
304         for( segindex = 1; segindex < num_segments; segindex++ ) {
305             
306             req_index = req_index ^ 0x1;
307             
308             /* post new irecv */
309             recv_reqs[req_index]= smpi_mpi_irecv( tmpbuf + realsegsize, count_by_segment,
310                                 datatype, tree->tree_prev, 
311                                 777, 
312                                 comm);
313             
314             /* wait for and forward the previous segment to children */
315             smpi_mpi_wait( &recv_reqs[req_index ^ 0x1], 
316                                      MPI_STATUSES_IGNORE );
317             
318             for( i = 0; i < tree->tree_nextsize; i++ ) { 
319                 send_reqs[i]=smpi_mpi_isend(tmpbuf, count_by_segment, datatype,
320                                          tree->tree_next[i], 
321                                          777, comm );
322             } 
323             
324             /* complete the sends before starting the next iteration */
325             smpi_mpi_waitall( tree->tree_nextsize, send_reqs, 
326                                          MPI_STATUSES_IGNORE );
327             
328             /* Update the receive buffer */
329             tmpbuf += realsegsize;
330         }
331
332         /* Process the last segment */
333         smpi_mpi_wait( &recv_reqs[req_index], MPI_STATUSES_IGNORE );
334         sendcount = original_count - (num_segments - 1) * count_by_segment;
335         for( i = 0; i < tree->tree_nextsize; i++ ) {
336             send_reqs[i] = smpi_mpi_isend(tmpbuf, sendcount, datatype,
337                                      tree->tree_next[i], 
338                                      777, comm);
339         }
340         
341         smpi_mpi_waitall( tree->tree_nextsize, send_reqs, 
342                                      MPI_STATUSES_IGNORE );
343     }
344   
345     /* Leaf nodes */
346     else {
347         /* 
348            Receive all segments from parent in a loop:
349            1) post irecv for the first segment
350            2) for segments 1 .. num_segments
351               - post irecv for the next segment
352               - wait on the previous segment to arrive
353            3) wait for the last segment
354         */
355         req_index = 0;
356         recv_reqs[req_index] = smpi_mpi_irecv(tmpbuf, count_by_segment, datatype,
357                                  tree->tree_prev, 777,
358                                  comm);
359
360         for( segindex = 1; segindex < num_segments; segindex++ ) {
361             req_index = req_index ^ 0x1;
362             tmpbuf += realsegsize;
363             /* post receive for the next segment */
364             recv_reqs[req_index] = smpi_mpi_irecv(tmpbuf, count_by_segment, datatype, 
365                                      tree->tree_prev, 777, 
366                                      comm);
367             /* wait on the previous segment */
368             smpi_mpi_wait( &recv_reqs[req_index ^ 0x1], 
369                                      MPI_STATUS_IGNORE );
370         }
371
372         smpi_mpi_wait( &recv_reqs[req_index], MPI_STATUS_IGNORE );
373     }
374
375     if( NULL != send_reqs ) free(send_reqs);
376
377     return (MPI_SUCCESS);
378 }