Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
40b1b5891a103af5d2861f1d36a21be907f57148
[simgrid.git] / src / smpi / smpi_base.c
1 #include "private.h"
2 #include "xbt/time.h"
3
4 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_base, smpi,
5                                 "Logging specific to SMPI (base)");
6 XBT_LOG_EXTERNAL_CATEGORY(smpi_base);
7 XBT_LOG_EXTERNAL_CATEGORY(smpi_bench);
8 XBT_LOG_EXTERNAL_CATEGORY(smpi_kernel);
9 XBT_LOG_EXTERNAL_CATEGORY(smpi_mpi);
10 XBT_LOG_EXTERNAL_CATEGORY(smpi_mpi_dt);
11 XBT_LOG_EXTERNAL_CATEGORY(smpi_coll);
12 XBT_LOG_EXTERNAL_CATEGORY(smpi_receiver);
13 XBT_LOG_EXTERNAL_CATEGORY(smpi_sender);
14 XBT_LOG_EXTERNAL_CATEGORY(smpi_util);
15
16 void smpi_process_init(int* argc, char*** argv) {
17   int index;
18   smpi_process_data_t data;
19   smx_process_t proc;
20
21   proc = SIMIX_process_self();
22   index = atoi((*argv)[1]);
23   data = smpi_process_remote_data(index);
24   SIMIX_process_set_data(proc, data);
25   if (*argc > 2) {
26     memmove(&(*argv)[1], &(*argv)[2], sizeof(char *) * (*argc - 2));
27     (*argv)[(*argc) - 1] = NULL;
28   }
29   (*argc)--;
30   DEBUG2("<%d> New process in the game: %p", index, proc);
31 }
32
33 void smpi_process_destroy(void) {
34   int index = smpi_process_index();
35
36   DEBUG1("<%d> Process left the game", index);
37 }
38
39 /* MPI Low level calls */
40 MPI_Request smpi_mpi_isend(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) {
41   MPI_Request request;
42
43   request = xbt_new(s_smpi_mpi_request_t, 1);
44   request->comm = comm;
45   request->src = smpi_comm_rank(comm);
46   request->dst = dst;
47   request->tag = tag;
48   request->size = smpi_datatype_size(datatype) * count;
49   request->complete = 0;
50   smpi_process_post_send(comm, request);
51   request->pair = SIMIX_network_isend(request->rdv, request->size, -1.0, buf, request->size, request);
52   return request;
53 }
54
55 MPI_Request smpi_mpi_irecv(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm) {
56   MPI_Request request;
57
58   request = xbt_new(s_smpi_mpi_request_t, 1);
59   request->comm = comm;
60   request->src = src;
61   request->dst = smpi_comm_rank(comm);
62   request->tag = tag;
63   request->size = smpi_datatype_size(datatype) * count;
64   request->complete = 0;
65   smpi_process_post_recv(request);
66   request->pair = SIMIX_network_irecv(request->rdv, buf, &request->size);
67   return request;
68 }
69
70 void smpi_mpi_recv(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm, MPI_Status* status) {
71   MPI_Request request;
72
73   request = smpi_mpi_irecv(buf, count, datatype, src, tag, comm);
74   smpi_mpi_wait(&request, status);
75 }
76
77 void smpi_mpi_send(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm) {
78   MPI_Request request;
79
80   request = smpi_mpi_isend(buf, count, datatype, src, tag, comm);
81   smpi_mpi_wait(&request, MPI_STATUS_IGNORE);
82 }
83
84 void smpi_mpi_sendrecv(void* sendbuf, int sendcount, MPI_Datatype sendtype, int dst, int sendtag, void* recvbuf, int recvcount, MPI_Datatype recvtype, int src, int recvtag, MPI_Comm comm, MPI_Status* status) {
85   MPI_Request requests[2];
86   MPI_Status stats[2];
87
88   requests[0] = smpi_mpi_isend(sendbuf, sendcount, sendtype, dst, sendtag, comm);
89   requests[1] = smpi_mpi_irecv(recvbuf, recvcount, recvtype, src, recvtag, comm);
90   smpi_mpi_waitall(2, requests, stats);
91   if(status != MPI_STATUS_IGNORE) {
92     // Copy receive status
93     memcpy(status, &stats[1], sizeof(MPI_Status));
94   }
95 }
96
97 static void finish_wait(MPI_Request* request, MPI_Status* status) {
98   MPI_Request data = (MPI_Request)SIMIX_communication_get_data((*request)->pair);
99
100   xbt_assert0(data != MPI_REQUEST_NULL, "Erroneous situation");
101   if(status != MPI_STATUS_IGNORE) {
102     status->MPI_SOURCE = (*request)->src;
103     status->MPI_TAG = (*request)->tag;
104     status->MPI_ERROR = MPI_SUCCESS;
105   }
106   DEBUG3("finishing wait for %p [data = %p, complete = %d]", *request, data, data->complete);
107   // data == *request if sender is first to finish its wait
108   // data != *request if receiver is first to finish its wait
109   if(data->complete == 0) {
110     // first arrives here
111     data->complete = 1;
112     if(data != *request) {
113       // receveiver cleans its part
114       xbt_free(*request);
115     }
116   } else {
117     // second arrives here
118     if(data != *request) {
119       // receiver cleans everything
120       xbt_free(data);
121     }
122     xbt_free(*request);
123   }
124   *request = MPI_REQUEST_NULL;
125 }
126
127 int smpi_mpi_test(MPI_Request* request, MPI_Status* status) {
128   MPI_Request data = (MPI_Request)SIMIX_communication_get_data((*request)->pair);
129   int flag = data && data->complete == 1;
130
131   if(flag) {
132     finish_wait(request, status);
133   }
134   return flag;
135 }
136
137 int smpi_mpi_testany(int count, MPI_Request requests[], int* index, MPI_Status* status) {
138   MPI_Request data;
139   int i, flag;
140
141   *index = MPI_UNDEFINED;
142   flag = 0;
143   for(i = 0; i < count; i++) {
144     if(requests[i] != MPI_REQUEST_NULL) {
145       data = (MPI_Request)SIMIX_communication_get_data(requests[i]->pair);
146       if(data != MPI_REQUEST_NULL && data->complete == 1) {
147         finish_wait(&requests[i], status);
148         *index = i;
149         flag = 1;
150         break;
151       }
152     }
153   }
154   return flag;
155 }
156
157 void smpi_mpi_wait(MPI_Request* request, MPI_Status* status) {
158   MPI_Request data = (MPI_Request)SIMIX_communication_get_data((*request)->pair);
159
160   DEBUG6("wait for request %p (%p: %p) [src = %d, dst = %d, tag = %d]",
161          *request, (*request)->pair, data, (*request)->src, (*request)->dst, (*request)->tag);
162   // data is null if receiver waits before sender enters the rdv
163   if(data == MPI_REQUEST_NULL || data->complete == 0) {
164     SIMIX_network_wait((*request)->pair, -1.0);
165   }
166   finish_wait(request, status);
167 }
168
169 int smpi_mpi_waitany(int count, MPI_Request requests[], MPI_Status* status) {
170   xbt_dynar_t comms;
171   MPI_Request data;
172   int i, size, index;
173   int* map;
174
175   index = MPI_UNDEFINED;
176   if(count > 0) {
177     // First check for already completed requests
178     for(i = 0; i < count; i++) {
179       if(requests[i] != MPI_REQUEST_NULL) {
180         data = (MPI_Request)SIMIX_communication_get_data(requests[i]->pair);
181         if(data != MPI_REQUEST_NULL && data->complete == 1) {
182           index = i;
183           break;
184         }
185       }
186     }
187     if(index == MPI_UNDEFINED) {
188       // Otherwise, wait for a request to complete
189       comms = xbt_dynar_new(sizeof(smx_comm_t), NULL);
190       map = xbt_new(int, count);
191       size = 0;
192       DEBUG0("Wait for one of");
193       for(i = 0; i < count; i++) {
194         if(requests[i] != MPI_REQUEST_NULL && requests[i]->complete == 0) {
195          DEBUG4("   request %p [src = %d, dst = %d, tag = %d]",
196                 requests[i], requests[i]->src, requests[i]->dst, requests[i]->tag);
197           xbt_dynar_push(comms, &requests[i]->pair);
198           map[size] = i;
199           size++;
200         }
201       }
202       if(size > 0) {
203         index = SIMIX_network_waitany(comms);
204         index = map[index];
205       }
206       xbt_free(map);
207       xbt_dynar_free_container(&comms);
208     }
209     if(index != MPI_UNDEFINED) {
210       finish_wait(&requests[index], status);
211     }
212   }
213   return index;
214 }
215
216 void smpi_mpi_waitall(int count, MPI_Request requests[],  MPI_Status status[]) {
217   int index;
218   MPI_Status stat;
219
220   while(count > 0) {
221     index = smpi_mpi_waitany(count, requests, &stat);
222     if(index == MPI_UNDEFINED) {
223       break;
224     }
225     if(status != MPI_STATUS_IGNORE) {
226       memcpy(&status[index], &stat, sizeof(stat));
227     }
228     // Move the last request to the found position
229     requests[index] = requests[count - 1];
230     requests[count - 1] = MPI_REQUEST_NULL;
231     count--;
232   }
233 }
234
235 int smpi_mpi_waitsome(int incount, MPI_Request requests[], int* indices, MPI_Status status[]) {
236   MPI_Request data;
237   int i, count;
238
239   count = 0;
240   for(i = 0; i < count; i++) {
241     if(requests[i] != MPI_REQUEST_NULL) {
242       data = (MPI_Request)SIMIX_communication_get_data(requests[i]->pair);
243       if(data != MPI_REQUEST_NULL && data->complete == 1) {
244         finish_wait(&requests[i], status != MPI_STATUS_IGNORE ? &status[i] : MPI_STATUS_IGNORE);
245         indices[count] = i;
246         count++;
247       }
248     }
249   }
250   return count;
251 }
252
253 void smpi_mpi_bcast(void* buf, int count, MPI_Datatype datatype, int root, MPI_Comm comm) {
254   // arity=2: a binary tree, arity=4 seem to be a good setting (see P2P-MPI))
255   nary_tree_bcast(buf, count, datatype, root, comm, 4);
256 }
257
258 void smpi_mpi_barrier(MPI_Comm comm) {
259   // arity=2: a binary tree, arity=4 seem to be a good setting (see P2P-MPI))
260   nary_tree_barrier(comm, 4);
261 }
262
263 void smpi_mpi_gather(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm) {
264   int system_tag = 666;
265   int rank, size, src, index, sendsize, recvsize;
266   MPI_Request* requests;
267
268   rank = smpi_comm_rank(comm);
269   size = smpi_comm_size(comm);
270   if(rank != root) {
271     // Send buffer to root
272     smpi_mpi_send(sendbuf, sendcount, sendtype, root, system_tag, comm);
273   } else {
274     sendsize = smpi_datatype_size(sendtype);
275     recvsize = smpi_datatype_size(recvtype);
276     // Local copy from root
277     memcpy(&((char*)recvbuf)[root * recvcount * recvsize], sendbuf, sendcount * sendsize * sizeof(char));
278     // Receive buffers from senders
279     requests = xbt_new(MPI_Request, size - 1);
280     index = 0;
281     for(src = 0; src < size; src++) {
282       if(src != root) {
283         requests[index] = smpi_mpi_irecv(&((char*)recvbuf)[src * recvcount * recvsize], recvcount, recvtype, src, system_tag, comm);
284         index++;
285       }
286     }
287     // Wait for completion of irecv's.
288     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
289     xbt_free(requests);
290   }
291 }
292
293 void smpi_mpi_gatherv(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int* recvcounts, int* displs, MPI_Datatype recvtype, int root, MPI_Comm comm) {
294   int system_tag = 666;
295   int rank, size, src, index, sendsize;
296   MPI_Request* requests;
297
298   rank = smpi_comm_rank(comm);
299   size = smpi_comm_size(comm);
300   if(rank != root) {
301     // Send buffer to root
302     smpi_mpi_send(sendbuf, sendcount, sendtype, root, system_tag, comm);
303   } else {
304     sendsize = smpi_datatype_size(sendtype);
305     // Local copy from root
306     memcpy(&((char*)recvbuf)[displs[root]], sendbuf, sendcount * sendsize * sizeof(char));
307     // Receive buffers from senders
308     requests = xbt_new(MPI_Request, size - 1);
309     index = 0;
310     for(src = 0; src < size; src++) {
311       if(src != root) {
312         requests[index] = smpi_mpi_irecv(&((char*)recvbuf)[displs[src]], recvcounts[src], recvtype, src, system_tag, comm);
313         index++;
314       }
315     }
316     // Wait for completion of irecv's.
317     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
318     xbt_free(requests);
319   }
320 }
321
322 void smpi_mpi_allgather(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, MPI_Comm comm) {
323   int system_tag = 666;
324   int rank, size, other, index, sendsize, recvsize;
325   MPI_Request* requests;
326
327   rank = smpi_comm_rank(comm);
328   size = smpi_comm_size(comm);
329   sendsize = smpi_datatype_size(sendtype);
330   recvsize = smpi_datatype_size(recvtype);
331   // Local copy from self
332   memcpy(&((char*)recvbuf)[rank * recvcount * recvsize], sendbuf, sendcount * sendsize * sizeof(char));
333   // Send/Recv buffers to/from others;
334   requests = xbt_new(MPI_Request, 2 * (size - 1));
335   index = 0;
336   for(other = 0; other < size; other++) {
337     if(other != rank) {
338       requests[index] = smpi_mpi_isend(sendbuf, sendcount, sendtype, other, system_tag, comm);
339       index++;
340       requests[index] = smpi_mpi_irecv(&((char*)recvbuf)[other * recvcount * recvsize], recvcount, recvtype, other, system_tag, comm);
341       index++;
342     }
343   }
344   // Wait for completion of all comms.
345   smpi_mpi_waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE);
346   xbt_free(requests);
347 }
348
349 void smpi_mpi_allgatherv(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int* recvcounts, int* displs, MPI_Datatype recvtype, MPI_Comm comm) {
350   int system_tag = 666;
351   int rank, size, other, index, sendsize, recvsize;
352   MPI_Request* requests;
353
354   rank = smpi_comm_rank(comm);
355   size = smpi_comm_size(comm);
356   sendsize = smpi_datatype_size(sendtype);
357   recvsize = smpi_datatype_size(recvtype);
358   // Local copy from self
359   memcpy(&((char*)recvbuf)[displs[rank]], sendbuf, sendcount * sendsize * sizeof(char));
360   // Send buffers to others;
361   requests = xbt_new(MPI_Request, 2 * (size - 1));
362   index = 0;
363   for(other = 0; other < size; other++) {
364     if(other != rank) {
365       requests[index] = smpi_mpi_isend(sendbuf, sendcount, sendtype, other, system_tag, comm);
366       index++;
367       requests[index] = smpi_mpi_irecv(&((char*)recvbuf)[displs[other]], recvcounts[other], recvtype, other, system_tag, comm);
368       index++;
369     }
370   }
371   // Wait for completion of all comms.
372   smpi_mpi_waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE);
373   xbt_free(requests);
374 }
375
376 void smpi_mpi_scatter(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm) {
377   int system_tag = 666;
378   int rank, size, dst, index, sendsize, recvsize;
379   MPI_Request* requests;
380
381   rank = smpi_comm_rank(comm);
382   size = smpi_comm_size(comm);
383   if(rank != root) {
384     // Recv buffer from root
385     smpi_mpi_recv(recvbuf, recvcount, recvtype, root, system_tag, comm, MPI_STATUS_IGNORE);
386   } else {
387     sendsize = smpi_datatype_size(sendtype);
388     recvsize = smpi_datatype_size(recvtype);
389     // Local copy from root
390     memcpy(recvbuf, &((char*)sendbuf)[root * sendcount * sendsize], recvcount * recvsize * sizeof(char));
391     // Send buffers to receivers
392     requests = xbt_new(MPI_Request, size - 1);
393     index = 0;
394     for(dst = 0; dst < size; dst++) {
395       if(dst != root) {
396         requests[index] = smpi_mpi_isend(&((char*)sendbuf)[dst * sendcount * sendsize], sendcount, sendtype, dst, system_tag, comm);
397         index++;
398       }
399     }
400     // Wait for completion of isend's.
401     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
402     xbt_free(requests);
403   }
404 }
405
406 void smpi_mpi_scatterv(void* sendbuf, int* sendcounts, int* displs, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm) {
407   int system_tag = 666;
408   int rank, size, dst, index, sendsize, recvsize;
409   MPI_Request* requests;
410
411   rank = smpi_comm_rank(comm);
412   size = smpi_comm_size(comm);
413   if(rank != root) {
414     // Recv buffer from root
415     smpi_mpi_recv(recvbuf, recvcount, recvtype, root, system_tag, comm, MPI_STATUS_IGNORE);
416   } else {
417     sendsize = smpi_datatype_size(sendtype);
418     recvsize = smpi_datatype_size(recvtype);
419     // Local copy from root
420     memcpy(recvbuf, &((char*)sendbuf)[displs[root]], recvcount * recvsize * sizeof(char));
421     // Send buffers to receivers
422     requests = xbt_new(MPI_Request, size - 1);
423     index = 0;
424     for(dst = 0; dst < size; dst++) {
425       if(dst != root) {
426         requests[index] = smpi_mpi_isend(&((char*)sendbuf)[displs[dst]], sendcounts[dst], sendtype, dst, system_tag, comm);
427         index++;
428       }
429     }
430     // Wait for completion of isend's.
431     smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
432     xbt_free(requests);
433   }
434 }
435
436 void smpi_mpi_reduce(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype, MPI_Op op, int root, MPI_Comm comm) {
437   int system_tag = 666;
438   int rank, size, src, index, datasize;
439   MPI_Request* requests;
440   void** tmpbufs;
441  
442   rank = smpi_comm_rank(comm);
443   size = smpi_comm_size(comm);
444   if(rank != root) {
445     // Send buffer to root
446     smpi_mpi_send(sendbuf, count, datatype, root, system_tag, comm);
447   } else {
448     datasize = smpi_datatype_size(datatype);
449     // Local copy from root
450     memcpy(recvbuf, sendbuf, count * datasize * sizeof(char)); 
451     // Receive buffers from senders
452     //TODO: make a MPI_barrier here ?
453     requests = xbt_new(MPI_Request, size - 1);
454     tmpbufs = xbt_new(void*, size - 1);
455     index = 0;
456     for(src = 0; src < size; src++) {
457       if(src != root) {
458         tmpbufs[index] = xbt_malloc(count * datasize);
459         requests[index] = smpi_mpi_irecv(tmpbufs[index], count, datatype, src, system_tag, comm);
460         index++;
461       }
462     }
463     // Wait for completion of irecv's.
464     for(src = 0; src < size - 1; src++) {
465       index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE);
466       if(index == MPI_UNDEFINED) {
467         break;
468       }
469       smpi_op_apply(op, tmpbufs[index], recvbuf, &count, &datatype);
470     }
471     for(index = 0; index < size - 1; index++) {
472       xbt_free(tmpbufs[index]);
473     }
474     xbt_free(tmpbufs);
475     xbt_free(requests);
476   }
477 }
478
479 void smpi_mpi_allreduce(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype, MPI_Op op, MPI_Comm comm) {
480   smpi_mpi_reduce(sendbuf, recvbuf, count, datatype, op, 0, comm);
481   smpi_mpi_bcast(recvbuf, count, datatype, 0, comm);
482
483 /*
484 FIXME: buggy implementation
485
486   int system_tag = 666;
487   int rank, size, other, index, datasize;
488   MPI_Request* requests;
489   void** tmpbufs;
490  
491   rank = smpi_comm_rank(comm);
492   size = smpi_comm_size(comm);
493   datasize = smpi_datatype_size(datatype);
494   // Local copy from self
495   memcpy(recvbuf, sendbuf, count * datasize * sizeof(char)); 
496   // Send/Recv buffers to/from others;
497   //TODO: make a MPI_barrier here ?
498   requests = xbt_new(MPI_Request, 2 * (size - 1));
499   tmpbufs = xbt_new(void*, size - 1);
500   index = 0;
501   for(other = 0; other < size; other++) {
502     if(other != rank) {
503       tmpbufs[index / 2] = xbt_malloc(count * datasize);
504       requests[index] = smpi_mpi_isend(sendbuf, count, datatype, other, system_tag, comm);
505       requests[index + 1] = smpi_mpi_irecv(tmpbufs[index / 2], count, datatype, other, system_tag, comm);
506       index += 2;
507     }
508   }
509   // Wait for completion of all comms.
510   for(other = 0; other < 2 * (size - 1); other++) {
511     index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE);
512     if(index == MPI_UNDEFINED) {
513       break;
514     }
515     if((index & 1) == 1) {
516       // Request is odd: it's a irecv
517       smpi_op_apply(op, tmpbufs[index / 2], recvbuf, &count, &datatype);
518     }
519   }
520   for(index = 0; index < size - 1; index++) {
521     xbt_free(tmpbufs[index]);
522   }
523   xbt_free(tmpbufs);
524   xbt_free(requests);
525 */
526 }