Logo AND Algorithmique Numérique Distribuée

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