Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
MPI_Comm -> C++
[simgrid.git] / src / smpi / colls / bcast-arrival-scatter.cpp
1 /* Copyright (c) 2013-2014. The SimGrid Team.
2  * All rights reserved.                                                     */
3
4 /* This program is free software; you can redistribute it and/or modify it
5  * under the terms of the license (GNU LGPL) which comes with this package. */
6
7 #include "colls_private.h"
8
9 #ifndef BCAST_ARRIVAL_PATTERN_AWARE_HEADER_SIZE
10 #define BCAST_ARRIVAL_PATTERN_AWARE_HEADER_SIZE 128
11 #endif
12
13 #ifndef BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE
14 #define BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE 128
15 #endif
16
17 /* Non-topology-specific pipelined linear-bcast function */
18 int smpi_coll_tuned_bcast_arrival_scatter(void *buf, int count,
19                                           MPI_Datatype datatype, int root,
20                                           MPI_Comm comm)
21 {
22   int tag = -COLL_TAG_BCAST;//in order to use ANY_TAG, make this one positive
23   int header_tag = 10;
24   MPI_Status status;
25
26   int curr_remainder;
27   int curr_size;
28   int curr_increment;
29   int send_offset;
30   int recv_offset;
31   int send_count;
32   int recv_count;
33
34   MPI_Status temp_status_array[BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE];
35
36   int rank, size;
37   int i, k;
38
39   int sent_count;
40   int header_index;
41   int flag_array[BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE];
42   int already_sent[BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE];
43   int header_buf[BCAST_ARRIVAL_PATTERN_AWARE_HEADER_SIZE];
44   char temp_buf[BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE];
45   int will_send[BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE];
46   int max_node = BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE;
47   int header_size = BCAST_ARRIVAL_PATTERN_AWARE_HEADER_SIZE;
48
49   MPI_Aint extent;
50   extent = smpi_datatype_get_extent(datatype);
51
52
53   /* source and destination */
54   int to, from;
55
56   rank = comm->rank();
57   size = comm->size();
58
59   /* message too small */
60   if (count < size) {
61     XBT_WARN("MPI_bcast_arrival_scatter use default MPI_bcast.");
62     smpi_mpi_bcast(buf, count, datatype, root, comm);
63     return MPI_SUCCESS;        
64   }
65
66
67
68   /* if root is not zero send to rank zero first
69      this can be modified to make it faster by using logical src, dst.
70    */
71   if (root != 0) {
72     if (rank == root) {
73       smpi_mpi_send(buf, count, datatype, 0, tag - 1, comm);
74     } else if (rank == 0) {
75       smpi_mpi_recv(buf, count, datatype, root, tag - 1, comm, &status);
76     }
77   }
78
79
80   /* value == 0 means root has not send data (or header) to the node yet */
81   for (i = 0; i < max_node; i++) {
82     already_sent[i] = 0;
83   }
84
85   /* start bcast */
86
87   /* root */
88   if (rank == 0) {
89
90     for (i = 0; i < max_node; i++)
91       will_send[i] = 0;
92
93     sent_count = 0;
94     while (sent_count < (size - 1)) {
95
96       for (k = 0; k < 3; k++) {
97         for (i = 1; i < size; i++) {
98           if ((already_sent[i] == 0) && (will_send[i] == 0)) {
99             smpi_mpi_iprobe(i, MPI_ANY_TAG, comm, &flag_array[i],
100                        &temp_status_array[i]);
101             if (flag_array[i] == 1) {
102               will_send[i] = 1;
103               smpi_mpi_recv(&temp_buf[i], 1, MPI_CHAR, i, tag, comm,
104                        &status);
105               i = 0;
106             }
107           }
108         }
109       }
110       header_index = 0;
111
112       /* recv 1-byte message in this round */
113       for (i = 1; i < size; i++) {
114         /* message arrive */
115         if ((will_send[i] == 1) && (already_sent[i] == 0)) {
116           header_buf[header_index] = i;
117           header_index++;
118           sent_count++;
119
120           /* will send in the next step */
121           already_sent[i] = 1;
122         }
123       }
124
125       /*
126          if (header_index != 0) {
127          printf("header index = %d node = ",header_index);
128          for (i=0;i<header_index;i++) {
129          printf("%d ",header_buf[i]);
130          }
131          printf("\n");
132          }
133        */
134
135       /* send header followed by data */
136       if (header_index != 0) {
137         header_buf[header_index] = -1;
138
139         /* send header */
140         for (i = 0; i < header_index; i++) {
141           to = header_buf[i];
142           smpi_mpi_send(header_buf, header_size, MPI_INT, to, header_tag, comm);
143         }
144
145         curr_remainder = count % header_index;
146         curr_size = (count / header_index);
147         curr_increment = curr_size * extent;
148
149         /* send data */
150
151         for (i = 0; i < header_index; i++) {
152           to = header_buf[i];
153           if ((i == (header_index - 1)) || (curr_size == 0))
154             curr_size += curr_remainder;
155           //printf("Root send to %d index %d\n",to,(i*curr_increment));
156           smpi_mpi_send((char *) buf + (i * curr_increment), curr_size, datatype, to,
157                    tag, comm);
158         }
159       }
160     }                           /* while (sent_count < size-1) */
161   }
162
163   /* rank 0 */
164   /* none root */
165   else {
166     /* send 1-byte message to root */
167     smpi_mpi_send(temp_buf, 1, MPI_CHAR, 0, tag, comm);
168
169     /* wait for header forward when required */
170     smpi_mpi_recv(header_buf, header_size, MPI_INT, 0, header_tag, comm, &status);
171
172     /* search for where it is */
173     int myordering = 0;
174     while (rank != header_buf[myordering]) {
175       myordering++;
176     }
177
178     int total_nodes = 0;
179     while (header_buf[total_nodes] != -1) {
180       total_nodes++;
181     }
182
183     curr_remainder = count % total_nodes;
184     curr_size = (count / total_nodes);
185     curr_increment = curr_size * extent;
186     int recv_size = curr_size;
187
188     /* receive data */
189     if (myordering == (total_nodes - 1))
190       recv_size += curr_remainder;
191     smpi_mpi_recv((char *) buf + (myordering * curr_increment), recv_size, datatype,
192              0, tag, comm, &status);
193
194     /* at this point all nodes in this set perform all-gather operation */
195     
196     to = (myordering == (total_nodes - 1)) ? header_buf[0] : header_buf[myordering + 1];
197     from = (myordering == 0) ? header_buf[total_nodes - 1] : header_buf[myordering - 1];
198
199
200     /* last segment may have a larger size since it also include the remainder */
201     int last_segment_ptr = (total_nodes - 1) * (count / total_nodes) * extent;
202
203
204     /* allgather */
205     for (i = 0; i < total_nodes - 1; i++) {
206       send_offset =
207           ((myordering - i + total_nodes) % total_nodes) * curr_increment;
208       recv_offset =
209           ((myordering - i - 1 + total_nodes) % total_nodes) * curr_increment;
210
211       /* adjust size */
212       if (send_offset != last_segment_ptr)
213         send_count = curr_size;
214       else
215         send_count = curr_size + curr_remainder;
216
217       if (recv_offset != last_segment_ptr)
218         recv_count = curr_size;
219       else
220         recv_count = curr_size + curr_remainder;
221
222       //printf("\t\tnode %d sent_to %d recv_from %d send_size %d recv_size %d\n",rank,to,from,send_count,recv_count);
223       //printf("\tnode %d sent_offset %d send_count %d\n",rank,send_offset,send_count);
224
225
226       smpi_mpi_sendrecv((char *) buf + send_offset, send_count, datatype, to,
227                    tag + i, (char *) buf + recv_offset, recv_count, datatype,
228                    from, tag + i, comm, &status);
229     }
230   }                             /* non-root */
231
232   return MPI_SUCCESS;
233 }