From 3c1eda61bc5b0d9c846abd9bb53b0bef26cb2880 Mon Sep 17 00:00:00 2001 From: pini Date: Wed, 21 Apr 2010 13:46:50 +0000 Subject: [PATCH] Added persistent communications (working but not yet fully compliant with the rdv protocol). git-svn-id: svn+ssh://scm.gforge.inria.fr/svn/simgrid/simgrid/trunk@7636 48e7efb5-ca39-0410-a469-dd3cf9ba447f --- include/smpi/smpi.h | 8 +- src/smpi/private.h | 20 ++++- src/smpi/smpi_base.c | 169 ++++++++++++++++++++++++++++------------- src/smpi/smpi_coll.c | 45 ++++------- src/smpi/smpi_global.c | 27 ++++--- src/smpi/smpi_mpi.c | 82 ++++++++++++++++++++ 6 files changed, 251 insertions(+), 100 deletions(-) diff --git a/include/smpi/smpi.h b/include/smpi/smpi.h index 4fe9a6d836..efdcace4e0 100644 --- a/include/smpi/smpi.h +++ b/include/smpi/smpi.h @@ -58,8 +58,6 @@ typedef struct { int MPI_SOURCE; int MPI_TAG; int MPI_ERROR; - int _count; - int _cancelled; } MPI_Status; #define MPI_STATUS_IGNORE NULL @@ -187,6 +185,11 @@ XBT_PUBLIC(int) MPI_Comm_dup(MPI_Comm comm, MPI_Comm* newcomm); XBT_PUBLIC(int) MPI_Comm_create(MPI_Comm comm, MPI_Group group, MPI_Comm* newcomm); XBT_PUBLIC(int) MPI_Comm_free(MPI_Comm* comm); +XBT_PUBLIC(int) MPI_Send_init(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm, MPI_Request* request); +XBT_PUBLIC(int) MPI_Recv_init(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm, MPI_Request* request); +XBT_PUBLIC(int) MPI_Start(MPI_Request* request); +XBT_PUBLIC(int) MPI_Startall(int count, MPI_Request* requests); +XBT_PUBLIC(int) MPI_Request_free(MPI_Request* request); XBT_PUBLIC(int) MPI_Irecv(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm, MPI_Request* request); XBT_PUBLIC(int) MPI_Isend(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm, MPI_Request* request); XBT_PUBLIC(int) MPI_Recv(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm, MPI_Status* status); @@ -195,7 +198,6 @@ XBT_PUBLIC(int) MPI_Sendrecv(void* sendbuf, int sendcount, MPI_Datatype sendtype XBT_PUBLIC(int) MPI_Sendrecv_replace(void* buf, int count, MPI_Datatype datatype, int dst, int sendtag, int src, int recvtag, MPI_Comm comm, MPI_Status* status); -XBT_PUBLIC(int) MPI_Get_count(MPI_Status *status, MPI_Datatype datatype, int *count); XBT_PUBLIC(int) MPI_Test(MPI_Request* request, int* flag, MPI_Status* status); XBT_PUBLIC(int) MPI_Testany(int count, MPI_Request requests[], int* index, int* flag, MPI_Status* status); XBT_PUBLIC(int) MPI_Wait(MPI_Request* request, MPI_Status* status); diff --git a/src/smpi/private.h b/src/smpi/private.h index 0ddb0faefe..4a5eccde73 100644 --- a/src/smpi/private.h +++ b/src/smpi/private.h @@ -9,16 +9,24 @@ struct s_smpi_process_data; typedef struct s_smpi_process_data* smpi_process_data_t; +#define PERSISTENT 0x0 +#define NON_PERSISTENT 0x1 +#define SEND 0x0 +#define RECV 0x2 + typedef struct s_smpi_mpi_request { - MPI_Comm comm; + void* buf; + size_t size; int src; int dst; int tag; - size_t size; + MPI_Comm comm; smx_rdv_t rdv; smx_comm_t pair; int complete; MPI_Request match; + unsigned flags; + MPI_Request ack; } s_smpi_mpi_request_t; void smpi_process_init(int* argc, char*** argv); @@ -29,6 +37,7 @@ smpi_process_data_t smpi_process_remote_data(int index); int smpi_process_count(void); int smpi_process_index(void); xbt_os_timer_t smpi_process_timer(void); +void print_request(const char* message, MPI_Request request); void smpi_process_post_send(MPI_Comm comm, MPI_Request request); void smpi_process_post_recv(MPI_Request request); @@ -61,7 +70,14 @@ MPI_Group smpi_comm_group(MPI_Comm comm); int smpi_comm_size(MPI_Comm comm); int smpi_comm_rank(MPI_Comm comm); +MPI_Request smpi_mpi_send_init(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm); +MPI_Request smpi_mpi_recv_init(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm); +void smpi_mpi_start(MPI_Request request); +void smpi_mpi_startall(int count, MPI_Request* requests); +void smpi_mpi_request_free(MPI_Request* request); +MPI_Request smpi_isend_init(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm); MPI_Request smpi_mpi_isend(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm); +MPI_Request smpi_irecv_init(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm); MPI_Request smpi_mpi_irecv(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm); void smpi_mpi_recv(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm, MPI_Status* status); void smpi_mpi_send(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm); diff --git a/src/smpi/smpi_base.c b/src/smpi/smpi_base.c index e1a0c567dc..a60f2098d2 100644 --- a/src/smpi/smpi_base.c +++ b/src/smpi/smpi_base.c @@ -40,48 +40,107 @@ void smpi_process_destroy(void) { DEBUG1("<%d> Process left the game", index); } -/* MPI Low level calls */ -MPI_Request smpi_mpi_isend(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) { +static MPI_Request build_request(void* buf, int count, MPI_Datatype datatype, int src, int dst, int tag, MPI_Comm comm, unsigned flags) { MPI_Request request; - size_t size = smpi_datatype_size(datatype) * count; - if (size > EAGER_LIMIT) { - /* Warning: this (zero-length synchronous) call will come back here with size == 0 */ - DEBUG1("RDV send to %d", dst); - } request = xbt_new(s_smpi_mpi_request_t, 1); - request->comm = comm; - request->src = smpi_comm_rank(comm); + request->buf = buf; + request->size = smpi_datatype_size(datatype) * count; + request->src = src; request->dst = dst; request->tag = tag; - request->size = size; + request->comm = comm; + request->rdv = NULL; + request->pair = NULL; request->complete = 0; request->match = MPI_REQUEST_NULL; - smpi_process_post_send(comm, request); - DEBUG3("NEW send request %p with rdv %p and match %p", request, request->rdv, request->match); - request->pair = SIMIX_network_isend(request->rdv, request->size, -1.0, buf, request->size, NULL); + request->flags = flags; + if(request->size <= EAGER_LIMIT) { + request->ack = MPI_REQUEST_NULL; + } else { + request->ack = xbt_new(s_smpi_mpi_request_t, 1); + request->ack->buf = NULL; + request->ack->size = 0; + request->ack->src = dst; + request->ack->dst = src; + request->ack->tag = RDV_TAG; + request->ack->comm = comm; + request->ack->rdv = NULL; + request->ack->pair = NULL; + request->ack->complete = 0; + request->ack->match = MPI_REQUEST_NULL; + request->ack->flags = NON_PERSISTENT | ((request->flags & RECV) == RECV ? SEND : RECV); + smpi_mpi_start(request->ack); + } return request; } -MPI_Request smpi_mpi_irecv(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm) { - MPI_Request request; - size_t size = smpi_datatype_size(datatype) * count; +/* MPI Low level calls */ +MPI_Request smpi_mpi_send_init(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) { + MPI_Request request = build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag, comm, PERSISTENT | SEND); + + return request; +} - if (size > EAGER_LIMIT) { - /* Warning: this (zero-length synchronous) call will come back here with size == 0 */ - DEBUG1("RDV recv from %d", src); +MPI_Request smpi_mpi_recv_init(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm) { + MPI_Request request = build_request(buf, count, datatype, src, smpi_comm_rank(comm), tag, comm, PERSISTENT | RECV); + + return request; +} + +void smpi_mpi_start(MPI_Request request) { + xbt_assert0(request->complete == 0, "Cannot start a non-finished communication"); + if(request->size > EAGER_LIMIT) { + print_request("RDV ack", request->ack); + smpi_mpi_wait(&request->ack, MPI_STATUS_IGNORE); } - request = xbt_new(s_smpi_mpi_request_t, 1); - request->comm = comm; - request->src = src; - request->dst = smpi_comm_rank(comm); - request->tag = tag; - request->size = size; - request->complete = 0; - request->match = MPI_REQUEST_NULL; - smpi_process_post_recv(request); - DEBUG3("NEW recv request %p with rdv %p and match %p", request, request->rdv, request->match); - request->pair = SIMIX_network_irecv(request->rdv, buf, &request->size); + if((request->flags & RECV) == RECV) { + smpi_process_post_recv(request); + print_request("New recv", request); + request->pair = SIMIX_network_irecv(request->rdv, request->buf, &request->size); + } else { + smpi_process_post_send(request->comm, request); // FIXME + print_request("New send", request); + request->pair = SIMIX_network_isend(request->rdv, request->size, -1.0, request->buf, request->size, NULL); + } +} + +void smpi_mpi_startall(int count, MPI_Request* requests) { + int i; + + for(i = 0; i < count; i++) { + smpi_mpi_start(requests[i]); + } +} + +void smpi_mpi_request_free(MPI_Request* request) { + xbt_free(*request); + *request = MPI_REQUEST_NULL; +} + +MPI_Request smpi_isend_init(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) { + MPI_Request request = build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag, comm, NON_PERSISTENT | SEND); + + return request; +} + +MPI_Request smpi_mpi_isend(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) { + MPI_Request request = smpi_isend_init(buf, count, datatype, dst, tag, comm); + + smpi_mpi_start(request); + return request; +} + +MPI_Request smpi_irecv_init(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm) { + MPI_Request request = build_request(buf, count, datatype, src, smpi_comm_rank(comm), tag, comm, NON_PERSISTENT | RECV); + + return request; +} + +MPI_Request smpi_mpi_irecv(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm) { + MPI_Request request = smpi_irecv_init(buf, count, datatype, src, tag, comm); + + smpi_mpi_start(request); return request; } @@ -103,8 +162,9 @@ void smpi_mpi_sendrecv(void* sendbuf, int sendcount, MPI_Datatype sendtype, int MPI_Request requests[2]; MPI_Status stats[2]; - requests[0] = smpi_mpi_isend(sendbuf, sendcount, sendtype, dst, sendtag, comm); - requests[1] = smpi_mpi_irecv(recvbuf, recvcount, recvtype, src, recvtag, comm); + requests[0] = smpi_isend_init(sendbuf, sendcount, sendtype, dst, sendtag, comm); + requests[1] = smpi_irecv_init(recvbuf, recvcount, recvtype, src, recvtag, comm); + smpi_mpi_startall(2, requests); smpi_mpi_waitall(2, requests, stats); if(status != MPI_STATUS_IGNORE) { // Copy receive status @@ -118,17 +178,16 @@ static void finish_wait(MPI_Request* request, MPI_Status* status) { status->MPI_TAG = (*request)->tag; status->MPI_ERROR = MPI_SUCCESS; } - DEBUG5("finishing wait for %p [src = %d, dst = %d, tag = %d, complete = %d]", - *request, (*request)->src, (*request)->dst, (*request)->tag, (*request)->complete); - DEBUG2("match %p, complete %d", (*request)->match, (*request)->match ? (*request)->match->complete : -1); + print_request("finishing wait", *request); if((*request)->complete == 1) { SIMIX_rdv_destroy((*request)->rdv); } else { (*request)->match->complete = 1; (*request)->match->match = MPI_REQUEST_NULL; } - xbt_free(*request); - *request = MPI_REQUEST_NULL; + if(((*request)->flags & NON_PERSISTENT) == NON_PERSISTENT) { + smpi_mpi_request_free(request); + } } int smpi_mpi_test(MPI_Request* request, MPI_Status* status) { @@ -159,8 +218,7 @@ int smpi_mpi_testany(int count, MPI_Request requests[], int* index, MPI_Status* } void smpi_mpi_wait(MPI_Request* request, MPI_Status* status) { - DEBUG6("wait for request %p (%p) [src = %d, dst = %d, tag = %d, complete = %d]", - *request, (*request)->pair, (*request)->src, (*request)->dst, (*request)->tag, (*request)->complete); + print_request("wait", *request); // data is null if receiver waits before sender enters the rdv if((*request)->complete) { SIMIX_communication_destroy((*request)->pair); @@ -193,8 +251,7 @@ int smpi_mpi_waitany(int count, MPI_Request requests[], MPI_Status* status) { DEBUG0("Wait for one of"); for(i = 0; i < count; i++) { if(requests[i] != MPI_REQUEST_NULL && requests[i]->complete == 0) { - DEBUG4(" request %p [src = %d, dst = %d, tag = %d]", - requests[i], requests[i]->src, requests[i]->dst, requests[i]->tag); + print_request(" ", requests[i]); xbt_dynar_push(comms, &requests[i]->pair); map[size] = i; size++; @@ -279,11 +336,12 @@ void smpi_mpi_gather(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* index = 0; for(src = 0; src < size; src++) { if(src != root) { - requests[index] = smpi_mpi_irecv(&((char*)recvbuf)[src * recvcount * recvsize], recvcount, recvtype, src, system_tag, comm); + requests[index] = smpi_irecv_init(&((char*)recvbuf)[src * recvcount * recvsize], recvcount, recvtype, src, system_tag, comm); index++; } } // Wait for completion of irecv's. + smpi_mpi_startall(size - 1, requests); smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE); xbt_free(requests); } @@ -308,11 +366,12 @@ void smpi_mpi_gatherv(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* index = 0; for(src = 0; src < size; src++) { if(src != root) { - requests[index] = smpi_mpi_irecv(&((char*)recvbuf)[displs[src]], recvcounts[src], recvtype, src, system_tag, comm); + requests[index] = smpi_irecv_init(&((char*)recvbuf)[displs[src]], recvcounts[src], recvtype, src, system_tag, comm); index++; } } // Wait for completion of irecv's. + smpi_mpi_startall(size - 1, requests); smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE); xbt_free(requests); } @@ -334,13 +393,14 @@ void smpi_mpi_allgather(void* sendbuf, int sendcount, MPI_Datatype sendtype, voi index = 0; for(other = 0; other < size; other++) { if(other != rank) { - requests[index] = smpi_mpi_isend(sendbuf, sendcount, sendtype, other, system_tag, comm); + requests[index] = smpi_isend_init(sendbuf, sendcount, sendtype, other, system_tag, comm); index++; - requests[index] = smpi_mpi_irecv(&((char*)recvbuf)[other * recvcount * recvsize], recvcount, recvtype, other, system_tag, comm); + requests[index] = smpi_irecv_init(&((char*)recvbuf)[other * recvcount * recvsize], recvcount, recvtype, other, system_tag, comm); index++; } } // Wait for completion of all comms. + smpi_mpi_startall(2 * (size - 1), requests); smpi_mpi_waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE); xbt_free(requests); } @@ -361,13 +421,14 @@ void smpi_mpi_allgatherv(void* sendbuf, int sendcount, MPI_Datatype sendtype, vo index = 0; for(other = 0; other < size; other++) { if(other != rank) { - requests[index] = smpi_mpi_isend(sendbuf, sendcount, sendtype, other, system_tag, comm); + requests[index] = smpi_isend_init(sendbuf, sendcount, sendtype, other, system_tag, comm); index++; - requests[index] = smpi_mpi_irecv(&((char*)recvbuf)[displs[other]], recvcounts[other], recvtype, other, system_tag, comm); + requests[index] = smpi_irecv_init(&((char*)recvbuf)[displs[other]], recvcounts[other], recvtype, other, system_tag, comm); index++; } } // Wait for completion of all comms. + smpi_mpi_startall(2 * (size - 1), requests); smpi_mpi_waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE); xbt_free(requests); } @@ -392,11 +453,12 @@ void smpi_mpi_scatter(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* index = 0; for(dst = 0; dst < size; dst++) { if(dst != root) { - requests[index] = smpi_mpi_isend(&((char*)sendbuf)[dst * sendcount * sendsize], sendcount, sendtype, dst, system_tag, comm); + requests[index] = smpi_isend_init(&((char*)sendbuf)[dst * sendcount * sendsize], sendcount, sendtype, dst, system_tag, comm); index++; } } // Wait for completion of isend's. + smpi_mpi_startall(size - 1, requests); smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE); xbt_free(requests); } @@ -422,11 +484,12 @@ void smpi_mpi_scatterv(void* sendbuf, int* sendcounts, int* displs, MPI_Datatype index = 0; for(dst = 0; dst < size; dst++) { if(dst != root) { - requests[index] = smpi_mpi_isend(&((char*)sendbuf)[displs[dst]], sendcounts[dst], sendtype, dst, system_tag, comm); + requests[index] = smpi_isend_init(&((char*)sendbuf)[displs[dst]], sendcounts[dst], sendtype, dst, system_tag, comm); index++; } } // Wait for completion of isend's. + smpi_mpi_startall(size - 1, requests); smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE); xbt_free(requests); } @@ -455,11 +518,12 @@ void smpi_mpi_reduce(void* sendbuf, void* recvbuf, int count, MPI_Datatype datat for(src = 0; src < size; src++) { if(src != root) { tmpbufs[index] = xbt_malloc(count * datasize); - requests[index] = smpi_mpi_irecv(tmpbufs[index], count, datatype, src, system_tag, comm); + requests[index] = smpi_irecv_init(tmpbufs[index], count, datatype, src, system_tag, comm); index++; } } // Wait for completion of irecv's. + smpi_mpi_startall(size - 1, requests); for(src = 0; src < size - 1; src++) { index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE); if(index == MPI_UNDEFINED) { @@ -543,14 +607,15 @@ void smpi_mpi_scan(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatyp index = 0; for(other = 0; other < rank; other++) { tmpbufs[index] = xbt_malloc(count * datasize); - requests[index] = smpi_mpi_irecv(tmpbufs[index], count, datatype, other, system_tag, comm); + requests[index] = smpi_irecv_init(tmpbufs[index], count, datatype, other, system_tag, comm); index++; } for(other = rank + 1; other < size; other++) { - requests[index] = smpi_mpi_isend(sendbuf, count, datatype, other, system_tag, comm); + requests[index] = smpi_isend_init(sendbuf, count, datatype, other, system_tag, comm); index++; } // Wait for completion of all comms. + smpi_mpi_startall(size - 1, requests); for(other = 0; other < total; other++) { index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE); if(index == MPI_UNDEFINED) { diff --git a/src/smpi/smpi_coll.c b/src/smpi/smpi_coll.c index 6dcc1ddc00..874d32a48d 100644 --- a/src/smpi/smpi_coll.c +++ b/src/smpi/smpi_coll.c @@ -109,9 +109,10 @@ static void tree_bcast(void* buf, int count, MPI_Datatype datatype, int root, MP requests[i] = MPI_REQUEST_NULL; } else { DEBUG3("<%d> send to <%d>, tag=%d", rank, tree->child[i], system_tag + tree->child[i]); - requests[i] = smpi_mpi_isend(buf, count, datatype, tree->child[i], system_tag + tree->child[i], comm); + requests[i] = smpi_isend_init(buf, count, datatype, tree->child[i], system_tag + tree->child[i], comm); } } + smpi_mpi_startall(tree->numChildren, requests); smpi_mpi_waitall(tree->numChildren, requests, MPI_STATUS_IGNORE); xbt_free(requests); } @@ -139,9 +140,10 @@ static void tree_antibcast(void* buf, int count, MPI_Datatype datatype, int root requests[i] = MPI_REQUEST_NULL; } else { DEBUG3("<%d> recv from <%d>, tag=%d", rank, tree->child[i], system_tag + tree->child[i]); - requests[i] = smpi_mpi_irecv(buf, count, datatype, tree->child[i], system_tag + tree->child[i], comm); + requests[i] = smpi_irecv_init(buf, count, datatype, tree->child[i], system_tag + tree->child[i], comm); } } + smpi_mpi_startall(tree->numChildren, requests); smpi_mpi_waitall(tree->numChildren, requests, MPI_STATUS_IGNORE); xbt_free(requests); } @@ -205,7 +207,7 @@ int smpi_coll_tuned_alltoall_bruck(void* sendbuf, int sendcount, MPI_Datatype se DEBUG3("<%d> skip request creation [src = %d, recvcount = %d]", rank, i, recvcount); continue; } - requests[count] = smpi_mpi_irecv(&((char*)recvbuf)[i * recvextent], recvcount, recvtype, i, system_tag, comm); + requests[count] = smpi_irecv_init(&((char*)recvbuf)[i * recvextent], recvcount, recvtype, i, system_tag, comm); count++; } /* Now create all sends */ @@ -214,16 +216,11 @@ int smpi_coll_tuned_alltoall_bruck(void* sendbuf, int sendcount, MPI_Datatype se DEBUG3("<%d> skip request creation [dst = %d, sendcount = %d]", rank, i, sendcount); continue; } - requests[count] = smpi_mpi_isend(&((char*)sendbuf)[i * sendextent], sendcount, sendtype, i, system_tag, comm); + requests[count] = smpi_isend_init(&((char*)sendbuf)[i * sendextent], sendcount, sendtype, i, system_tag, comm); count++; } - /* Wait for them all. If there's an error, note that we don't - * care what the error was -- just that there *was* an error. The - * PML will finish all requests, even if one or more of them fail. - * i.e., by the end of this call, all the requests are free-able. - * So free them anyway -- even if there was an error, and return - * the error after we free everything. - */ + /* Wait for them all.*/ + smpi_mpi_startall(count, requests); DEBUG2("<%d> wait for %d requests", rank, count); smpi_mpi_waitall(count, requests, MPI_STATUS_IGNORE); xbt_free(requests); @@ -256,7 +253,7 @@ int smpi_coll_tuned_alltoall_basic_linear(void *sendbuf, int sendcount, MPI_Data /* Post all receives first -- a simple optimization */ count = 0; for(i = (rank + 1) % size; i != rank; i = (i + 1) % size) { - requests[count] = smpi_mpi_irecv(&((char*)recvbuf)[i * recvinc], recvcount, recvtype, i, system_tag, comm); + requests[count] = smpi_irecv_init(&((char*)recvbuf)[i * recvinc], recvcount, recvtype, i, system_tag, comm); count++; } /* Now post all sends in reverse order @@ -265,16 +262,11 @@ int smpi_coll_tuned_alltoall_basic_linear(void *sendbuf, int sendcount, MPI_Data * TODO: check the previous assertion */ for(i = (rank + size - 1) % size; i != rank; i = (i + size - 1) % size ) { - requests[count] = smpi_mpi_isend(&((char*)sendbuf)[i * sendinc], sendcount, sendtype, i, system_tag, comm); + requests[count] = smpi_isend_init(&((char*)sendbuf)[i * sendinc], sendcount, sendtype, i, system_tag, comm); count++; } - /* Wait for them all. If there's an error, note that we don't - * care what the error was -- just that there *was* an error. The - * PML will finish all requests, even if one or more of them fail. - * i.e., by the end of this call, all the requests are free-able. - * So free them anyway -- even if there was an error, and return - * the error after we free everything. - */ + /* Wait for them all.*/ + smpi_mpi_startall(count, requests); DEBUG2("<%d> wait for %d requests", rank, count); smpi_mpi_waitall(count, requests, MPI_STATUS_IGNORE); xbt_free(requests); @@ -336,7 +328,7 @@ int smpi_coll_basic_alltoallv(void* sendbuf, int* sendcounts, int* senddisps, MP DEBUG3("<%d> skip request creation [src = %d, recvcounts[src] = %d]", rank, i, recvcounts[i]); continue; } - requests[count] = smpi_mpi_irecv(&((char*)recvbuf)[recvdisps[i] * recvextent], recvcounts[i], recvtype, i, system_tag, comm); + requests[count] = smpi_irecv_init(&((char*)recvbuf)[recvdisps[i] * recvextent], recvcounts[i], recvtype, i, system_tag, comm); count++; } /* Now create all sends */ @@ -345,16 +337,11 @@ int smpi_coll_basic_alltoallv(void* sendbuf, int* sendcounts, int* senddisps, MP DEBUG3("<%d> skip request creation [dst = %d, sendcounts[dst] = %d]", rank, i, sendcounts[i]); continue; } - requests[count] = smpi_mpi_isend(&((char*)sendbuf)[senddisps[i] * sendextent], sendcounts[i], sendtype, i, system_tag, comm); + requests[count] = smpi_isend_init(&((char*)sendbuf)[senddisps[i] * sendextent], sendcounts[i], sendtype, i, system_tag, comm); count++; } - /* Wait for them all. If there's an error, note that we don't - * care what the error was -- just that there *was* an error. The - * PML will finish all requests, even if one or more of them fail. - * i.e., by the end of this call, all the requests are free-able. - * So free them anyway -- even if there was an error, and return - * the error after we free everything. - */ + /* Wait for them all.*/ + smpi_mpi_startall(count, requests); DEBUG2("<%d> wait for %d requests", rank, count); smpi_mpi_waitall(count, requests, MPI_STATUS_IGNORE); xbt_free(requests); diff --git a/src/smpi/smpi_global.c b/src/smpi/smpi_global.c index 2efa9cbb5f..9a277df819 100644 --- a/src/smpi/smpi_global.c +++ b/src/smpi/smpi_global.c @@ -51,20 +51,27 @@ MPI_Comm smpi_process_comm_self(void) { return data->comm_self; } +void print_request(const char* message, MPI_Request request) { + char* req = bprintf("[buf = %p, size = %zu, src = %d, dst = %d, tag= %d, complete = %d, flags = %u]", + request->buf, request->size, request->src, request->dst, request->tag, request->complete, request->flags); + + DEBUG5("%s (request %p with rdv %p and match %p) %s", + message, request, request->rdv, request->match, req); + free(req); +} + void smpi_process_post_send(MPI_Comm comm, MPI_Request request) { int index = smpi_group_index(smpi_comm_group(comm), request->dst); smpi_process_data_t data = smpi_process_remote_data(index); xbt_fifo_item_t item; MPI_Request req; - DEBUG5("isend for request %p [src = %d, dst = %d, tag = %d, size = %zu]", - request, request->src, request->dst, request->tag, request->size); + print_request("Isend", request); xbt_fifo_foreach(data->pending_recv, item, req, MPI_Request) { if(req->comm == request->comm && (req->src == MPI_ANY_SOURCE || req->src == request->src) && (req->tag == MPI_ANY_TAG || req->tag == request->tag)){ - DEBUG4("find matching request %p [src = %d, dst = %d, tag = %d]", - req, req->src, req->dst, req->tag); + print_request("Match found", req); xbt_fifo_remove_item(data->pending_recv, item); /* Materialize the *_ANY_* fields from corresponding irecv request */ req->src = request->src; @@ -73,9 +80,6 @@ void smpi_process_post_send(MPI_Comm comm, MPI_Request request) { request->rdv = req->rdv; request->match = req; return; - } else { - DEBUG4("not matching request %p [src = %d, dst = %d, tag = %d]", - req, req->src, req->dst, req->tag); } } request->rdv = SIMIX_rdv_create(NULL); @@ -87,14 +91,12 @@ void smpi_process_post_recv(MPI_Request request) { xbt_fifo_item_t item; MPI_Request req; - DEBUG5("irecv for request %p [src = %d, dst = %d, tag = %d, size = %zu]", - request, request->src, request->dst, request->tag, request->size); + print_request("Irecv", request); xbt_fifo_foreach(data->pending_sent, item, req, MPI_Request) { if(req->comm == request->comm && (request->src == MPI_ANY_SOURCE || req->src == request->src) && (request->tag == MPI_ANY_TAG || req->tag == request->tag)){ - DEBUG4("find matching request %p [src = %d, dst = %d, tag = %d]", - req, req->src, req->dst, req->tag); + print_request("Match found", req); xbt_fifo_remove_item(data->pending_sent, item); /* Materialize the *_ANY_* fields from the irecv request */ req->match = request; @@ -103,9 +105,6 @@ void smpi_process_post_recv(MPI_Request request) { request->rdv = req->rdv; request->match = req; return; - } else { - DEBUG4("not matching request %p [src = %d, dst = %d, tag = %d]", - req, req->src, req->dst, req->tag); } } request->rdv = SIMIX_rdv_create(NULL); diff --git a/src/smpi/smpi_mpi.c b/src/smpi/smpi_mpi.c index 75cbab620c..60bbfd675e 100644 --- a/src/smpi/smpi_mpi.c +++ b/src/smpi/smpi_mpi.c @@ -681,6 +681,88 @@ int MPI_Comm_free(MPI_Comm* comm) { return retval; } +int MPI_Send_init(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm, MPI_Request* request) { + int retval; + int rank = comm != MPI_COMM_NULL ? smpi_comm_rank(comm) : -1; + + smpi_bench_end(rank, "Send_init"); + if(request == NULL) { + retval = MPI_ERR_ARG; + } else if (comm == MPI_COMM_NULL) { + retval = MPI_ERR_COMM; + } else { + *request = smpi_mpi_send_init(buf, count, datatype, dst, tag, comm); + retval = MPI_SUCCESS; + } + smpi_bench_begin(rank, "Send_init"); + return retval; +} + +int MPI_Recv_init(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm, MPI_Request* request) { + int retval; + int rank = comm != MPI_COMM_NULL ? smpi_comm_rank(comm) : -1; + + smpi_bench_end(rank, "Recv_init"); + if(request == NULL) { + retval = MPI_ERR_ARG; + } else if (comm == MPI_COMM_NULL) { + retval = MPI_ERR_COMM; + } else { + *request = smpi_mpi_recv_init(buf, count, datatype, src, tag, comm); + retval = MPI_SUCCESS; + } + smpi_bench_begin(rank, "Recv_init"); + return retval; +} + +int MPI_Start(MPI_Request* request) { + int retval; + MPI_Comm comm = (*request)->comm; + int rank = comm != MPI_COMM_NULL ? smpi_comm_rank(comm) : -1; + + smpi_bench_end(rank, "Start"); + if(request == NULL) { + retval = MPI_ERR_ARG; + } else { + smpi_mpi_start(*request); + retval = MPI_SUCCESS; + } + smpi_bench_begin(rank, "Start"); + return retval; +} + +int MPI_Startall(int count, MPI_Request* requests) { + int retval; + MPI_Comm comm = count > 0 && requests ? requests[0]->comm : MPI_COMM_NULL; + int rank = comm != MPI_COMM_NULL ? smpi_comm_rank(comm) : -1; + + smpi_bench_end(rank, "Startall"); + if(requests == NULL) { + retval = MPI_ERR_ARG; + } else { + smpi_mpi_startall(count, requests); + retval = MPI_SUCCESS; + } + smpi_bench_begin(rank, "Startall"); + return retval; +} + +int MPI_Request_free(MPI_Request* request) { + int retval; + MPI_Comm comm = (*request)->comm; + int rank = comm != MPI_COMM_NULL ? smpi_comm_rank(comm) : -1; + + smpi_bench_end(rank, "Request_free"); + if(request == NULL) { + retval = MPI_ERR_ARG; + } else { + smpi_mpi_request_free(request); + retval = MPI_SUCCESS; + } + smpi_bench_begin(rank, "Request_free"); + return retval; +} + int MPI_Irecv(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm, MPI_Request* request) { int retval; int rank = comm != MPI_COMM_NULL ? smpi_comm_rank(comm) : -1; -- 2.20.1