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);
16 #define EAGER_LIMIT 65536
19 void smpi_process_init(int* argc, char*** argv) {
21 smpi_process_data_t data;
24 proc = SIMIX_process_self();
25 index = atoi((*argv)[1]);
26 data = smpi_process_remote_data(index);
27 SIMIX_process_set_data(proc, data);
30 memmove(&(*argv)[1], &(*argv)[2], sizeof(char *) * (*argc - 2));
31 (*argv)[(*argc) - 1] = NULL;
34 DEBUG2("<%d> New process in the game: %p", index, proc);
37 void smpi_process_destroy(void) {
38 int index = smpi_process_index();
40 DEBUG1("<%d> Process left the game", index);
43 static MPI_Request build_request(void* buf, int count, MPI_Datatype datatype, int src, int dst, int tag, MPI_Comm comm, unsigned flags) {
46 request = xbt_new(s_smpi_mpi_request_t, 1);
48 request->size = smpi_datatype_size(datatype) * count;
55 request->complete = 0;
56 request->match = MPI_REQUEST_NULL;
57 request->flags = flags;
58 if(request->size <= EAGER_LIMIT) {
59 request->ack = MPI_REQUEST_NULL;
61 request->ack = xbt_new(s_smpi_mpi_request_t, 1);
62 request->ack->buf = NULL;
63 request->ack->size = 0;
64 request->ack->src = dst;
65 request->ack->dst = src;
66 request->ack->tag = RDV_TAG;
67 request->ack->comm = comm;
68 request->ack->rdv = NULL;
69 request->ack->pair = NULL;
70 request->ack->complete = 0;
71 request->ack->match = MPI_REQUEST_NULL;
72 request->ack->flags = NON_PERSISTENT | ((request->flags & RECV) == RECV ? SEND : RECV);
73 smpi_mpi_start(request->ack);
78 /* MPI Low level calls */
79 MPI_Request smpi_mpi_send_init(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) {
80 MPI_Request request = build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag, comm, PERSISTENT | SEND);
85 MPI_Request smpi_mpi_recv_init(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm) {
86 MPI_Request request = build_request(buf, count, datatype, src, smpi_comm_rank(comm), tag, comm, PERSISTENT | RECV);
91 void smpi_mpi_start(MPI_Request request) {
92 xbt_assert0(request->complete == 0, "Cannot start a non-finished communication");
93 if(request->size > EAGER_LIMIT) {
94 print_request("RDV ack", request->ack);
95 smpi_mpi_wait(&request->ack, MPI_STATUS_IGNORE);
97 if((request->flags & RECV) == RECV) {
98 smpi_process_post_recv(request);
99 print_request("New recv", request);
100 request->pair = SIMIX_network_irecv(request->rdv, request->buf, &request->size);
102 smpi_process_post_send(request->comm, request); // FIXME
103 print_request("New send", request);
104 request->pair = SIMIX_network_isend(request->rdv, request->size, -1.0, request->buf, request->size, NULL);
108 void smpi_mpi_startall(int count, MPI_Request* requests) {
111 for(i = 0; i < count; i++) {
112 smpi_mpi_start(requests[i]);
116 void smpi_mpi_request_free(MPI_Request* request) {
118 *request = MPI_REQUEST_NULL;
121 MPI_Request smpi_isend_init(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) {
122 MPI_Request request = build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag, comm, NON_PERSISTENT | SEND);
127 MPI_Request smpi_mpi_isend(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) {
128 MPI_Request request = smpi_isend_init(buf, count, datatype, dst, tag, comm);
130 smpi_mpi_start(request);
134 MPI_Request smpi_irecv_init(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm) {
135 MPI_Request request = build_request(buf, count, datatype, src, smpi_comm_rank(comm), tag, comm, NON_PERSISTENT | RECV);
140 MPI_Request smpi_mpi_irecv(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm) {
141 MPI_Request request = smpi_irecv_init(buf, count, datatype, src, tag, comm);
143 smpi_mpi_start(request);
147 void smpi_mpi_recv(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm, MPI_Status* status) {
150 request = smpi_mpi_irecv(buf, count, datatype, src, tag, comm);
151 smpi_mpi_wait(&request, status);
154 void smpi_mpi_send(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) {
157 request = smpi_mpi_isend(buf, count, datatype, dst, tag, comm);
158 smpi_mpi_wait(&request, MPI_STATUS_IGNORE);
161 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) {
162 MPI_Request requests[2];
165 requests[0] = smpi_isend_init(sendbuf, sendcount, sendtype, dst, sendtag, comm);
166 requests[1] = smpi_irecv_init(recvbuf, recvcount, recvtype, src, recvtag, comm);
167 smpi_mpi_startall(2, requests);
168 smpi_mpi_waitall(2, requests, stats);
169 if(status != MPI_STATUS_IGNORE) {
170 // Copy receive status
171 memcpy(status, &stats[1], sizeof(MPI_Status));
175 static void finish_wait(MPI_Request* request, MPI_Status* status) {
176 if(status != MPI_STATUS_IGNORE) {
177 status->MPI_SOURCE = (*request)->src;
178 status->MPI_TAG = (*request)->tag;
179 status->MPI_ERROR = MPI_SUCCESS;
181 print_request("finishing wait", *request);
182 if((*request)->complete == 1) {
183 SIMIX_rdv_destroy((*request)->rdv);
185 (*request)->match->complete = 1;
186 (*request)->match->match = MPI_REQUEST_NULL;
188 if(((*request)->flags & NON_PERSISTENT) == NON_PERSISTENT) {
189 smpi_mpi_request_free(request);
193 int smpi_mpi_test(MPI_Request* request, MPI_Status* status) {
194 int flag = (*request)->complete;
197 SIMIX_communication_destroy((*request)->pair);
198 finish_wait(request, status);
203 int smpi_mpi_testany(int count, MPI_Request requests[], int* index, MPI_Status* status) {
206 *index = MPI_UNDEFINED;
208 for(i = 0; i < count; i++) {
209 if(requests[i] != MPI_REQUEST_NULL && requests[i]->complete) {
210 SIMIX_communication_destroy(requests[i]->pair);
211 finish_wait(&requests[i], status);
220 void smpi_mpi_wait(MPI_Request* request, MPI_Status* status) {
221 print_request("wait", *request);
222 // data is null if receiver waits before sender enters the rdv
223 if((*request)->complete) {
224 SIMIX_communication_destroy((*request)->pair);
226 SIMIX_network_wait((*request)->pair, -1.0);
228 finish_wait(request, status);
231 int smpi_mpi_waitany(int count, MPI_Request requests[], MPI_Status* status) {
236 index = MPI_UNDEFINED;
238 // First check for already completed requests
239 for(i = 0; i < count; i++) {
240 if(requests[i] != MPI_REQUEST_NULL && requests[i]->complete) {
242 SIMIX_communication_destroy(requests[index]->pair); // always succeeds (but cleans the simix layer)
246 if(index == MPI_UNDEFINED) {
247 // Otherwise, wait for a request to complete
248 comms = xbt_dynar_new(sizeof(smx_comm_t), NULL);
249 map = xbt_new(int, count);
251 DEBUG0("Wait for one of");
252 for(i = 0; i < count; i++) {
253 if(requests[i] != MPI_REQUEST_NULL && requests[i]->complete == 0) {
254 print_request(" ", requests[i]);
255 xbt_dynar_push(comms, &requests[i]->pair);
261 index = SIMIX_network_waitany(comms);
265 xbt_dynar_free(&comms);
267 if(index != MPI_UNDEFINED) {
268 finish_wait(&requests[index], status);
274 void smpi_mpi_waitall(int count, MPI_Request requests[], MPI_Status status[]) {
279 index = smpi_mpi_waitany(count, requests, &stat);
280 if(index == MPI_UNDEFINED) {
283 if(status != MPI_STATUS_IGNORE) {
284 memcpy(&status[index], &stat, sizeof(stat));
286 // FIXME: check this -v
287 // Move the last request to the found position
288 requests[index] = requests[count - 1];
289 requests[count - 1] = MPI_REQUEST_NULL;
294 int smpi_mpi_waitsome(int incount, MPI_Request requests[], int* indices, MPI_Status status[]) {
298 for(i = 0; i < incount; i++) {
299 if(requests[i] != MPI_REQUEST_NULL && requests[i]->complete) {
300 SIMIX_communication_destroy(requests[i]->pair);
301 finish_wait(&requests[i], status != MPI_STATUS_IGNORE ? &status[i] : MPI_STATUS_IGNORE);
309 void smpi_mpi_bcast(void* buf, int count, MPI_Datatype datatype, int root, MPI_Comm comm) {
310 // arity=2: a binary tree, arity=4 seem to be a good setting (see P2P-MPI))
311 nary_tree_bcast(buf, count, datatype, root, comm, 4);
314 void smpi_mpi_barrier(MPI_Comm comm) {
315 // arity=2: a binary tree, arity=4 seem to be a good setting (see P2P-MPI))
316 nary_tree_barrier(comm, 4);
319 void smpi_mpi_gather(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm) {
320 int system_tag = 666;
321 int rank, size, src, index, sendsize, recvsize;
322 MPI_Request* requests;
324 rank = smpi_comm_rank(comm);
325 size = smpi_comm_size(comm);
327 // Send buffer to root
328 smpi_mpi_send(sendbuf, sendcount, sendtype, root, system_tag, comm);
330 sendsize = smpi_datatype_size(sendtype);
331 recvsize = smpi_datatype_size(recvtype);
332 // Local copy from root
333 memcpy(&((char*)recvbuf)[root * recvcount * recvsize], sendbuf, sendcount * sendsize * sizeof(char));
334 // Receive buffers from senders
335 requests = xbt_new(MPI_Request, size - 1);
337 for(src = 0; src < size; src++) {
339 requests[index] = smpi_irecv_init(&((char*)recvbuf)[src * recvcount * recvsize], recvcount, recvtype, src, system_tag, comm);
343 // Wait for completion of irecv's.
344 smpi_mpi_startall(size - 1, requests);
345 smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
350 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) {
351 int system_tag = 666;
352 int rank, size, src, index, sendsize;
353 MPI_Request* requests;
355 rank = smpi_comm_rank(comm);
356 size = smpi_comm_size(comm);
358 // Send buffer to root
359 smpi_mpi_send(sendbuf, sendcount, sendtype, root, system_tag, comm);
361 sendsize = smpi_datatype_size(sendtype);
362 // Local copy from root
363 memcpy(&((char*)recvbuf)[displs[root]], sendbuf, sendcount * sendsize * sizeof(char));
364 // Receive buffers from senders
365 requests = xbt_new(MPI_Request, size - 1);
367 for(src = 0; src < size; src++) {
369 requests[index] = smpi_irecv_init(&((char*)recvbuf)[displs[src]], recvcounts[src], recvtype, src, system_tag, comm);
373 // Wait for completion of irecv's.
374 smpi_mpi_startall(size - 1, requests);
375 smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
380 void smpi_mpi_allgather(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, MPI_Comm comm) {
381 int system_tag = 666;
382 int rank, size, other, index, sendsize, recvsize;
383 MPI_Request* requests;
385 rank = smpi_comm_rank(comm);
386 size = smpi_comm_size(comm);
387 sendsize = smpi_datatype_size(sendtype);
388 recvsize = smpi_datatype_size(recvtype);
389 // Local copy from self
390 memcpy(&((char*)recvbuf)[rank * recvcount * recvsize], sendbuf, sendcount * sendsize * sizeof(char));
391 // Send/Recv buffers to/from others;
392 requests = xbt_new(MPI_Request, 2 * (size - 1));
394 for(other = 0; other < size; other++) {
396 requests[index] = smpi_isend_init(sendbuf, sendcount, sendtype, other, system_tag, comm);
398 requests[index] = smpi_irecv_init(&((char*)recvbuf)[other * recvcount * recvsize], recvcount, recvtype, other, system_tag, comm);
402 // Wait for completion of all comms.
403 smpi_mpi_startall(2 * (size - 1), requests);
404 smpi_mpi_waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE);
408 void smpi_mpi_allgatherv(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int* recvcounts, int* displs, MPI_Datatype recvtype, MPI_Comm comm) {
409 int system_tag = 666;
410 int rank, size, other, index, sendsize, recvsize;
411 MPI_Request* requests;
413 rank = smpi_comm_rank(comm);
414 size = smpi_comm_size(comm);
415 sendsize = smpi_datatype_size(sendtype);
416 recvsize = smpi_datatype_size(recvtype);
417 // Local copy from self
418 memcpy(&((char*)recvbuf)[displs[rank]], sendbuf, sendcount * sendsize * sizeof(char));
419 // Send buffers to others;
420 requests = xbt_new(MPI_Request, 2 * (size - 1));
422 for(other = 0; other < size; other++) {
424 requests[index] = smpi_isend_init(sendbuf, sendcount, sendtype, other, system_tag, comm);
426 requests[index] = smpi_irecv_init(&((char*)recvbuf)[displs[other]], recvcounts[other], recvtype, other, system_tag, comm);
430 // Wait for completion of all comms.
431 smpi_mpi_startall(2 * (size - 1), requests);
432 smpi_mpi_waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE);
436 void smpi_mpi_scatter(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm) {
437 int system_tag = 666;
438 int rank, size, dst, index, sendsize, recvsize;
439 MPI_Request* requests;
441 rank = smpi_comm_rank(comm);
442 size = smpi_comm_size(comm);
444 // Recv buffer from root
445 smpi_mpi_recv(recvbuf, recvcount, recvtype, root, system_tag, comm, MPI_STATUS_IGNORE);
447 sendsize = smpi_datatype_size(sendtype);
448 recvsize = smpi_datatype_size(recvtype);
449 // Local copy from root
450 memcpy(recvbuf, &((char*)sendbuf)[root * sendcount * sendsize], recvcount * recvsize * sizeof(char));
451 // Send buffers to receivers
452 requests = xbt_new(MPI_Request, size - 1);
454 for(dst = 0; dst < size; dst++) {
456 requests[index] = smpi_isend_init(&((char*)sendbuf)[dst * sendcount * sendsize], sendcount, sendtype, dst, system_tag, comm);
460 // Wait for completion of isend's.
461 smpi_mpi_startall(size - 1, requests);
462 smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
467 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) {
468 int system_tag = 666;
469 int rank, size, dst, index, sendsize, recvsize;
470 MPI_Request* requests;
472 rank = smpi_comm_rank(comm);
473 size = smpi_comm_size(comm);
475 // Recv buffer from root
476 smpi_mpi_recv(recvbuf, recvcount, recvtype, root, system_tag, comm, MPI_STATUS_IGNORE);
478 sendsize = smpi_datatype_size(sendtype);
479 recvsize = smpi_datatype_size(recvtype);
480 // Local copy from root
481 memcpy(recvbuf, &((char*)sendbuf)[displs[root]], recvcount * recvsize * sizeof(char));
482 // Send buffers to receivers
483 requests = xbt_new(MPI_Request, size - 1);
485 for(dst = 0; dst < size; dst++) {
487 requests[index] = smpi_isend_init(&((char*)sendbuf)[displs[dst]], sendcounts[dst], sendtype, dst, system_tag, comm);
491 // Wait for completion of isend's.
492 smpi_mpi_startall(size - 1, requests);
493 smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE);
498 void smpi_mpi_reduce(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype, MPI_Op op, int root, MPI_Comm comm) {
499 int system_tag = 666;
500 int rank, size, src, index, datasize;
501 MPI_Request* requests;
504 rank = smpi_comm_rank(comm);
505 size = smpi_comm_size(comm);
507 // Send buffer to root
508 smpi_mpi_send(sendbuf, count, datatype, root, system_tag, comm);
510 datasize = smpi_datatype_size(datatype);
511 // Local copy from root
512 memcpy(recvbuf, sendbuf, count * datasize * sizeof(char));
513 // Receive buffers from senders
514 //TODO: make a MPI_barrier here ?
515 requests = xbt_new(MPI_Request, size - 1);
516 tmpbufs = xbt_new(void*, size - 1);
518 for(src = 0; src < size; src++) {
520 tmpbufs[index] = xbt_malloc(count * datasize);
521 requests[index] = smpi_irecv_init(tmpbufs[index], count, datatype, src, system_tag, comm);
525 // Wait for completion of irecv's.
526 smpi_mpi_startall(size - 1, requests);
527 for(src = 0; src < size - 1; src++) {
528 index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE);
529 if(index == MPI_UNDEFINED) {
532 smpi_op_apply(op, tmpbufs[index], recvbuf, &count, &datatype);
534 for(index = 0; index < size - 1; index++) {
535 xbt_free(tmpbufs[index]);
542 void smpi_mpi_allreduce(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype, MPI_Op op, MPI_Comm comm) {
543 smpi_mpi_reduce(sendbuf, recvbuf, count, datatype, op, 0, comm);
544 smpi_mpi_bcast(recvbuf, count, datatype, 0, comm);
547 FIXME: buggy implementation
549 int system_tag = 666;
550 int rank, size, other, index, datasize;
551 MPI_Request* requests;
554 rank = smpi_comm_rank(comm);
555 size = smpi_comm_size(comm);
556 datasize = smpi_datatype_size(datatype);
557 // Local copy from self
558 memcpy(recvbuf, sendbuf, count * datasize * sizeof(char));
559 // Send/Recv buffers to/from others;
560 //TODO: make a MPI_barrier here ?
561 requests = xbt_new(MPI_Request, 2 * (size - 1));
562 tmpbufs = xbt_new(void*, size - 1);
564 for(other = 0; other < size; other++) {
566 tmpbufs[index / 2] = xbt_malloc(count * datasize);
567 requests[index] = smpi_mpi_isend(sendbuf, count, datatype, other, system_tag, comm);
568 requests[index + 1] = smpi_mpi_irecv(tmpbufs[index / 2], count, datatype, other, system_tag, comm);
572 // Wait for completion of all comms.
573 for(other = 0; other < 2 * (size - 1); other++) {
574 index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE);
575 if(index == MPI_UNDEFINED) {
578 if((index & 1) == 1) {
579 // Request is odd: it's a irecv
580 smpi_op_apply(op, tmpbufs[index / 2], recvbuf, &count, &datatype);
583 for(index = 0; index < size - 1; index++) {
584 xbt_free(tmpbufs[index]);
591 void smpi_mpi_scan(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype, MPI_Op op, MPI_Comm comm) {
592 int system_tag = 666;
593 int rank, size, other, index, datasize;
595 MPI_Request* requests;
598 rank = smpi_comm_rank(comm);
599 size = smpi_comm_size(comm);
600 datasize = smpi_datatype_size(datatype);
601 // Local copy from self
602 memcpy(recvbuf, sendbuf, count * datasize * sizeof(char));
603 // Send/Recv buffers to/from others;
604 total = rank + (size - (rank + 1));
605 requests = xbt_new(MPI_Request, total);
606 tmpbufs = xbt_new(void*, rank);
608 for(other = 0; other < rank; other++) {
609 tmpbufs[index] = xbt_malloc(count * datasize);
610 requests[index] = smpi_irecv_init(tmpbufs[index], count, datatype, other, system_tag, comm);
613 for(other = rank + 1; other < size; other++) {
614 requests[index] = smpi_isend_init(sendbuf, count, datatype, other, system_tag, comm);
617 // Wait for completion of all comms.
618 smpi_mpi_startall(size - 1, requests);
619 for(other = 0; other < total; other++) {
620 index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE);
621 if(index == MPI_UNDEFINED) {
625 // #Request is below rank: it's a irecv
626 smpi_op_apply(op, tmpbufs[index], recvbuf, &count, &datatype);
629 for(index = 0; index < size - 1; index++) {
630 xbt_free(tmpbufs[index]);