X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/394a9cd893e93ee04bf4a8f6ff4ad5a8697a0e82..d74bb413a4b25a580d7cfc96065b638292a2e679:/src/smpi/smpi_base.c diff --git a/src/smpi/smpi_base.c b/src/smpi/smpi_base.c index 05a391a4b2..7296b10763 100644 --- a/src/smpi/smpi_base.c +++ b/src/smpi/smpi_base.c @@ -105,6 +105,8 @@ static void finish_wait(MPI_Request* request, MPI_Status* status) { status->MPI_SOURCE = (*request)->src; status->MPI_TAG = (*request)->tag; status->MPI_ERROR = MPI_SUCCESS; + status->_count = (*request)->size; // size in bytes + status->_cancelled = 0; // FIXME: cancellation of requests not handled yet } DEBUG3("finishing wait for %p [data = %p, complete = %d]", *request, data, data->complete); // data == *request if sender is first to finish its wait @@ -160,6 +162,15 @@ int smpi_mpi_testany(int count, MPI_Request requests[], int* index, MPI_Status* return flag; } + +void smpi_mpi_get_count(MPI_Status *status, MPI_Datatype datatype, int *count) { + int size = smpi_datatype_size(datatype); + *count = (int)(status->_count / size); + if ( (int)((*count) * size) != status->_count ) + *count = MPI_UNDEFINED; +} + + void smpi_mpi_wait(MPI_Request* request, MPI_Status* status) { MPI_Request data = (*request)->data; @@ -448,7 +459,7 @@ void smpi_mpi_reduce(void* sendbuf, void* recvbuf, int count, MPI_Datatype datat int rank, size, src, index, datasize; MPI_Request* requests; void** tmpbufs; - + rank = smpi_comm_rank(comm); size = smpi_comm_size(comm); if(rank != root) { @@ -457,7 +468,7 @@ void smpi_mpi_reduce(void* sendbuf, void* recvbuf, int count, MPI_Datatype datat } else { datasize = smpi_datatype_size(datatype); // Local copy from root - memcpy(recvbuf, sendbuf, count * datasize * sizeof(char)); + memcpy(recvbuf, sendbuf, count * datasize * sizeof(char)); // Receive buffers from senders //TODO: make a MPI_barrier here ? requests = xbt_new(MPI_Request, size - 1); @@ -497,12 +508,12 @@ FIXME: buggy implementation int rank, size, other, index, datasize; MPI_Request* requests; void** tmpbufs; - + rank = smpi_comm_rank(comm); size = smpi_comm_size(comm); datasize = smpi_datatype_size(datatype); // Local copy from self - memcpy(recvbuf, sendbuf, count * datasize * sizeof(char)); + memcpy(recvbuf, sendbuf, count * datasize * sizeof(char)); // Send/Recv buffers to/from others; //TODO: make a MPI_barrier here ? requests = xbt_new(MPI_Request, 2 * (size - 1)); @@ -534,3 +545,47 @@ FIXME: buggy implementation xbt_free(requests); */ } + +void smpi_mpi_scan(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype, MPI_Op op, MPI_Comm comm) { + int system_tag = 666; + int rank, size, other, index, datasize; + int total; + MPI_Request* requests; + void** tmpbufs; + + rank = smpi_comm_rank(comm); + size = smpi_comm_size(comm); + datasize = smpi_datatype_size(datatype); + // Local copy from self + memcpy(recvbuf, sendbuf, count * datasize * sizeof(char)); + // Send/Recv buffers to/from others; + total = rank + (size - (rank + 1)); + requests = xbt_new(MPI_Request, total); + tmpbufs = xbt_new(void*, rank); + 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); + index++; + } + for(other = rank + 1; other < size; other++) { + requests[index] = smpi_mpi_isend(sendbuf, count, datatype, other, system_tag, comm); + index++; + } + // Wait for completion of all comms. + for(other = 0; other < total; other++) { + index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE); + if(index == MPI_UNDEFINED) { + break; + } + if(index < rank) { + // #Request is below rank: it's a irecv + smpi_op_apply(op, tmpbufs[index], recvbuf, &count, &datatype); + } + } + for(index = 0; index < size - 1; index++) { + xbt_free(tmpbufs[index]); + } + xbt_free(tmpbufs); + xbt_free(requests); +}