Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Make internal collective flags negative (incorrect in MPI), to avoid confusion with...
[simgrid.git] / src / smpi / colls / bcast-arrival-pattern-aware-wait.c
1 #include "colls_private.h"
2
3 int bcast_arrival_pattern_aware_wait_segment_size_in_byte = 8192;
4
5 #ifndef BCAST_ARRIVAL_PATTERN_AWARE_HEADER_SIZE
6 #define BCAST_ARRIVAL_PATTERN_AWARE_HEADER_SIZE 1024
7 #endif
8
9 #ifndef BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE
10 #define BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE 128
11 #endif
12
13 /* Non-topology-specific pipelined linear-bcast function */
14 int smpi_coll_tuned_bcast_arrival_pattern_aware_wait(void *buf, int count,
15                                                      MPI_Datatype datatype,
16                                                      int root, MPI_Comm comm)
17 {
18   MPI_Status status;
19   MPI_Request request;
20   MPI_Request *send_request_array;
21   MPI_Request *recv_request_array;
22   MPI_Status *send_status_array;
23   MPI_Status *recv_status_array;
24
25
26   MPI_Status temp_status_array[BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE];
27
28   int rank, size;
29   int i, j, k;
30   int tag = -COLL_TAG_BCAST;
31   int will_send[BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE];
32
33   int sent_count;
34   int header_index;
35   int flag_array[BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE];
36   int already_sent[BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE];
37
38   int header_buf[BCAST_ARRIVAL_PATTERN_AWARE_HEADER_SIZE];
39   char temp_buf[BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE];
40
41   int max_node = BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE;
42   int header_size = BCAST_ARRIVAL_PATTERN_AWARE_HEADER_SIZE;
43
44   MPI_Aint extent;
45   extent = smpi_datatype_get_extent(datatype);
46
47   /* source and destination */
48   int to, from;
49
50
51
52   rank = smpi_comm_rank(MPI_COMM_WORLD);
53   size = smpi_comm_size(MPI_COMM_WORLD);
54
55
56   /* segment is segment size in number of elements (not bytes) */
57   int segment = bcast_arrival_pattern_aware_wait_segment_size_in_byte / extent;
58
59   /* pipeline length */
60   int pipe_length = count / segment;
61
62   /* use for buffer offset for sending and receiving data = segment size in byte */
63   int increment = segment * extent;
64
65   /* if the input size is not divisible by segment size => 
66      the small remainder will be done with native implementation */
67   int remainder = count % segment;
68
69   /* if root is not zero send to rank zero first
70      this can be modified to make it faster by using logical src, dst.
71    */
72   if (root != 0) {
73     if (rank == root) {
74       smpi_mpi_send(buf, count, datatype, 0, tag, comm);
75     } else if (rank == 0) {
76       smpi_mpi_recv(buf, count, datatype, root, tag, comm, &status);
77     }
78   }
79
80
81   /* value == 0 means root has not send data (or header) to the node yet */
82   for (i = 0; i < max_node; i++) {
83     already_sent[i] = 0;
84   }
85
86   /* when a message is smaller than a block size => no pipeline */
87   if (count <= segment) {
88     segment = count;
89     pipe_length = 1;
90   }
91
92   /* start pipeline bcast */
93
94   send_request_array =
95       (MPI_Request *) xbt_malloc((size + pipe_length) * sizeof(MPI_Request));
96   recv_request_array =
97       (MPI_Request *) xbt_malloc((size + pipe_length) * sizeof(MPI_Request));
98   send_status_array =
99       (MPI_Status *) xbt_malloc((size + pipe_length) * sizeof(MPI_Status));
100   recv_status_array =
101       (MPI_Status *) xbt_malloc((size + pipe_length) * sizeof(MPI_Status));
102
103   /* root */
104   if (rank == 0) {
105     sent_count = 0;
106     int iteration = 0;
107
108     for (i = 0; i < BCAST_ARRIVAL_PATTERN_AWARE_MAX_NODE; i++)
109       will_send[i] = 0;
110     while (sent_count < (size - 1)) {
111       iteration++;
112
113       /* loop k times to let more processes arrive before start sending data */
114       for (k = 0; k < 3; k++) {
115         for (i = 1; i < size; i++) {
116           if ((already_sent[i] == 0) && (will_send[i] == 0)) {
117             smpi_mpi_iprobe(i, MPI_ANY_TAG, MPI_COMM_WORLD, &flag_array[i],
118                        &temp_status_array[i]);
119             if (flag_array[i] == 1) {
120               will_send[i] = 1;
121               smpi_mpi_recv(&temp_buf[i], 1, MPI_CHAR, i, tag, MPI_COMM_WORLD,
122                        &status);
123               i = 0;
124             }
125           }
126         }
127       }
128
129       header_index = 0;
130
131       /* recv 1-byte message */
132       for (i = 1; i < size; i++) {
133         /* message arrive */
134         if ((will_send[i] == 1) && (already_sent[i] == 0)) {
135           header_buf[header_index] = i;
136           header_index++;
137           sent_count++;
138
139           /* will send in the next step */
140           already_sent[i] = 1;
141         }
142       }
143
144       /* send header followed by data */
145       if (header_index != 0) {
146         header_buf[header_index] = -1;
147         to = header_buf[0];
148
149         /* send header */
150         smpi_mpi_send(header_buf, header_size, MPI_INT, to, tag, comm);
151
152         /* send data - pipeline */
153         for (i = 0; i < pipe_length; i++) {
154           send_request_array[i] = smpi_mpi_isend((char *)buf + (i * increment), segment, datatype, to, tag, comm);
155         }
156         smpi_mpi_waitall((pipe_length), send_request_array, send_status_array);
157       }
158
159
160       /* end - send header followed by data */
161       /* randomly MPI_Send to one node */
162       /* this part has been commented out - performance-wise */
163       else if (2 == 3) {
164         /* search for the first node that never received data before */
165         for (i = 0; i < size; i++) {
166           if (i == root)
167             continue;
168           if (already_sent[i] == 0) {
169             header_buf[0] = i;
170             header_buf[1] = -1;
171             to = i;
172
173             smpi_mpi_send(header_buf, header_size, MPI_INT, to, tag, comm);
174
175             /* still need to chop data so that we can use the same non-root code */
176             for (j = 0; j < pipe_length; j++) {
177               smpi_mpi_send((char *)buf + (j * increment), segment, datatype, to, tag, comm);
178             }
179           }
180         }
181       }
182     }                           /* end - while (send_count < size-1) loop */
183   }
184
185   /* end - root */
186   /* none root */
187   else {
188
189     /* send 1-byte message to root */
190     smpi_mpi_send(temp_buf, 1, MPI_CHAR, 0, tag, comm);
191
192     /* wait for header forward when required */
193     request = smpi_mpi_irecv(header_buf, header_size, MPI_INT, MPI_ANY_SOURCE, tag, comm);
194     smpi_mpi_wait(&request, MPI_STATUS_IGNORE);
195
196     /* search for where it is */
197     int myordering = 0;
198     while (rank != header_buf[myordering]) {
199       myordering++;
200     }
201
202     to = header_buf[myordering + 1];
203     if (myordering == 0) {
204       from = 0;
205     } else {
206       from = header_buf[myordering - 1];
207     }
208
209     /* send header when required */
210     if (to != -1) {
211       smpi_mpi_send(header_buf, header_size, MPI_INT, to, tag, comm);
212     }
213
214     /* receive data */
215
216     for (i = 0; i < pipe_length; i++) {
217       recv_request_array[i] = smpi_mpi_irecv((char *)buf + (i * increment), segment, datatype, from, tag, comm);
218     }
219
220     /* forward data */
221     if (to != -1) {
222       for (i = 0; i < pipe_length; i++) {
223         smpi_mpi_wait(&recv_request_array[i], MPI_STATUS_IGNORE);
224         send_request_array[i] = smpi_mpi_isend((char *)buf + (i * increment), segment, datatype, to, tag, comm);
225       }
226       smpi_mpi_waitall((pipe_length), send_request_array, send_status_array);
227     }
228
229     /* recv only */
230     else {
231       smpi_mpi_waitall((pipe_length), recv_request_array, recv_status_array);
232     }
233   }
234
235   free(send_request_array);
236   free(recv_request_array);
237   free(send_status_array);
238   free(recv_status_array);
239   /* end pipeline */
240
241   /* when count is not divisible by block size, use default BCAST for the remainder */
242   if ((remainder != 0) && (count > segment)) {
243     XBT_WARN("MPI_bcast_arrival_pattern_aware_wait use default MPI_bcast.");              
244     smpi_mpi_bcast((char *)buf + (pipe_length * increment), remainder, datatype, root, comm);
245   }
246
247   return MPI_SUCCESS;
248 }