X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/73ce0df461cf8867f8002555273f54f7a8fb2d2f..35f18859ce88a96b7e07de4d7d70867424feb8df:/src/smpi/mpi/smpi_request.cpp diff --git a/src/smpi/mpi/smpi_request.cpp b/src/smpi/mpi/smpi_request.cpp index 88c539ea51..05cdc18c1e 100644 --- a/src/smpi/mpi/smpi_request.cpp +++ b/src/smpi/mpi/smpi_request.cpp @@ -36,14 +36,14 @@ extern void (*smpi_comm_copy_data_callback)(simgrid::kernel::activity::CommImpl* namespace simgrid{ namespace smpi{ -Request::Request(void* buf, int count, MPI_Datatype datatype, int src, int dst, int tag, MPI_Comm comm, unsigned flags, MPI_Op op) - : buf_(buf), old_type_(datatype), src_(src), dst_(dst), tag_(tag), comm_(comm), flags_(flags), op_(op) +Request::Request(const void* buf, int count, MPI_Datatype datatype, int src, int dst, int tag, MPI_Comm comm, unsigned flags, MPI_Op op) + : buf_(const_cast(buf)), old_type_(datatype), src_(src), dst_(dst), tag_(tag), comm_(comm), flags_(flags), op_(op) { void *old_buf = nullptr; // FIXME Handle the case of a partial shared malloc. if ((((flags & MPI_REQ_RECV) != 0) && ((flags & MPI_REQ_ACCUMULATE) != 0)) || (datatype->flags() & DT_FLAG_DERIVED)) { // This part handles the problem of non-contiguous memory - old_buf = buf; + old_buf = const_cast(buf); if (count==0){ buf_ = nullptr; }else { @@ -92,6 +92,7 @@ void Request::unref(MPI_Request* request) if((*request)->refcount_==0){ if ((*request)->flags_ & MPI_REQ_GENERALIZED){ ((*request)->generalized_funcs)->free_fn(((*request)->generalized_funcs)->extra_state); + delete (*request)->generalized_funcs; }else{ Comm::unref((*request)->comm_); Datatype::unref((*request)->old_type_); @@ -114,11 +115,12 @@ int Request::match_recv(void* a, void* b, simgrid::kernel::activity::CommImpl*) { MPI_Request ref = static_cast(a); MPI_Request req = static_cast(b); - XBT_DEBUG("Trying to match a recv of src %d against %d, tag %d against %d",ref->src_,req->src_, ref->tag_, req->tag_); + XBT_DEBUG("Trying to match a recv of src %d against %d, tag %d against %d, id %d against %d",ref->src_,req->src_, ref->tag_, req->tag_,ref->comm_->id(),req->comm_->id()); xbt_assert(ref, "Cannot match recv against null reference"); xbt_assert(req, "Cannot match recv against null request"); - if((ref->src_ == MPI_ANY_SOURCE || req->src_ == ref->src_) + if((ref->comm_->id()==MPI_UNDEFINED || req->comm_->id() == MPI_UNDEFINED || (ref->comm_->id()==req->comm_->id())) + && ((ref->src_ == MPI_ANY_SOURCE && (ref->comm_->group()->rank(req->src_) != MPI_UNDEFINED)) || req->src_ == ref->src_) && ((ref->tag_ == MPI_ANY_TAG && req->tag_ >=0) || req->tag_ == ref->tag_)){ //we match, we can transfer some values if(ref->src_ == MPI_ANY_SOURCE) @@ -140,11 +142,12 @@ int Request::match_send(void* a, void* b, simgrid::kernel::activity::CommImpl*) { MPI_Request ref = static_cast(a); MPI_Request req = static_cast(b); - XBT_DEBUG("Trying to match a send of src %d against %d, tag %d against %d",ref->src_,req->src_, ref->tag_, req->tag_); + XBT_DEBUG("Trying to match a send of src %d against %d, tag %d against %d, id %d against %d",ref->src_,req->src_, ref->tag_, req->tag_,ref->comm_->id(),req->comm_->id()); xbt_assert(ref, "Cannot match send against null reference"); xbt_assert(req, "Cannot match send against null request"); - if((req->src_ == MPI_ANY_SOURCE || req->src_ == ref->src_) + if((ref->comm_->id()==MPI_UNDEFINED || req->comm_->id() == MPI_UNDEFINED || (ref->comm_->id()==req->comm_->id())) + && ((req->src_ == MPI_ANY_SOURCE && (req->comm_->group()->rank(ref->src_) != MPI_UNDEFINED)) || req->src_ == ref->src_) && ((req->tag_ == MPI_ANY_TAG && ref->tag_ >=0)|| req->tag_ == ref->tag_)){ if(req->src_ == MPI_ANY_SOURCE) req->real_src_ = ref->src_; @@ -170,7 +173,7 @@ void Request::print_request(const char *message) /* factories, to hide the internal flags from the caller */ -MPI_Request Request::send_init(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) +MPI_Request Request::send_init(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) { return new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(), @@ -178,14 +181,14 @@ MPI_Request Request::send_init(void *buf, int count, MPI_Datatype datatype, int MPI_REQ_PERSISTENT | MPI_REQ_SEND | MPI_REQ_PREPARED); } -MPI_Request Request::ssend_init(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) +MPI_Request Request::ssend_init(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) { return new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(), comm->group()->actor(dst)->get_pid(), tag, comm, MPI_REQ_PERSISTENT | MPI_REQ_SSEND | MPI_REQ_SEND | MPI_REQ_PREPARED); } -MPI_Request Request::isend_init(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) +MPI_Request Request::isend_init(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) { return new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(), comm->group()->actor(dst)->get_pid(), tag, comm, @@ -193,7 +196,7 @@ MPI_Request Request::isend_init(void *buf, int count, MPI_Datatype datatype, int } -MPI_Request Request::rma_send_init(void *buf, int count, MPI_Datatype datatype, int src, int dst, int tag, MPI_Comm comm, +MPI_Request Request::rma_send_init(const void *buf, int count, MPI_Datatype datatype, int src, int dst, int tag, MPI_Comm comm, MPI_Op op) { MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */ @@ -242,7 +245,7 @@ MPI_Request Request::irecv_init(void *buf, int count, MPI_Datatype datatype, int MPI_REQ_PERSISTENT | MPI_REQ_RECV | MPI_REQ_PREPARED); } -MPI_Request Request::isend(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) +MPI_Request Request::isend(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) { MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */ request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(), @@ -252,7 +255,7 @@ MPI_Request Request::isend(void *buf, int count, MPI_Datatype datatype, int dst, return request; } -MPI_Request Request::issend(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) +MPI_Request Request::issend(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) { MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */ request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(), @@ -281,7 +284,7 @@ void Request::recv(void *buf, int count, MPI_Datatype datatype, int src, int tag request = nullptr; } -void Request::send(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) +void Request::send(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) { MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */ request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(), @@ -292,7 +295,7 @@ void Request::send(void *buf, int count, MPI_Datatype datatype, int dst, int tag request = nullptr; } -void Request::ssend(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) +void Request::ssend(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) { MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */ request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(), @@ -304,7 +307,7 @@ void Request::ssend(void *buf, int count, MPI_Datatype datatype, int dst, int ta request = nullptr; } -void Request::sendrecv(void *sendbuf, int sendcount, MPI_Datatype sendtype,int dst, int sendtag, +void Request::sendrecv(const 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) { @@ -547,7 +550,7 @@ int Request::test(MPI_Request * request, MPI_Status * status, int* flag) { if ((*request)->action_ != nullptr){ try{ *flag = simcall_comm_test((*request)->action_); - }catch (xbt_ex& e) { + } catch (const Exception&) { *flag = 0; return ret; } @@ -638,7 +641,8 @@ int Request::testany(int count, MPI_Request requests[], int *index, int* flag, M simcall_process_sleep(nsleeps*smpi_test_sleep); try{ i = simcall_comm_testany(comms.data(), comms.size()); // The i-th element in comms matches! - }catch (xbt_ex& e) { + } catch (const Exception&) { + XBT_DEBUG("Exception in testany"); return 0; } @@ -665,6 +669,7 @@ int Request::testany(int count, MPI_Request requests[], int *index, int* flag, M if (requests[*index] != MPI_REQUEST_NULL && (requests[*index]->flags_ & MPI_REQ_NON_PERSISTENT)) requests[*index] = MPI_REQUEST_NULL; + XBT_DEBUG("Testany - returning with index %d", *index); *flag=1; } nsleeps = 1; @@ -672,8 +677,10 @@ int Request::testany(int count, MPI_Request requests[], int *index, int* flag, M nsleeps++; } } else { + XBT_DEBUG("Testany on inactive handles, returning flag=1 but empty status"); //all requests are null or inactive, return true *flag = 1; + *index = MPI_UNDEFINED; Status::empty(status); } @@ -684,7 +691,8 @@ int Request::testall(int count, MPI_Request requests[], int* outflag, MPI_Status { MPI_Status stat; MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat; - int flag, error=0; + int flag; + int error = 0; int ret=MPI_SUCCESS; *outflag = 1; for(int i=0; idetached_ && ((req->flags_ & MPI_REQ_SEND) != 0)) && ((req->flags_ & MPI_REQ_PREPARED) == 0) && - ((req->flags_ & MPI_REQ_GENERALIZED) == 0)) { + if ((req->flags_ & (MPI_REQ_PREPARED | MPI_REQ_GENERALIZED | MPI_REQ_FINISHED)) == 0) { if(status != MPI_STATUS_IGNORE) { int src = req->src_ == MPI_ANY_SOURCE ? req->real_src_ : req->src_; status->MPI_SOURCE = req->comm_->group()->rank(src); @@ -807,32 +815,34 @@ void Request::finish_wait(MPI_Request* request, MPI_Status * status) // this handles the case were size in receive differs from size in send status->count = req->real_size_; } + //detached send will be finished at the other end + if (not(req->detached_ && ((req->flags_ & MPI_REQ_SEND) != 0))) { + req->print_request("Finishing"); + MPI_Datatype datatype = req->old_type_; + + // FIXME Handle the case of a partial shared malloc. + if (((req->flags_ & MPI_REQ_ACCUMULATE) != 0) || + (datatype->flags() & DT_FLAG_DERIVED)) { // && (not smpi_is_shared(req->old_buf_))){ + + if (not smpi_process()->replaying() && smpi_privatize_global_variables != SmpiPrivStrategies::NONE && + static_cast(req->old_buf_) >= smpi_data_exe_start && + static_cast(req->old_buf_) < smpi_data_exe_start + smpi_data_exe_size) { + XBT_VERB("Privatization : We are unserializing to a zone in global memory Switch data segment "); + smpi_switch_data_segment(simgrid::s4u::Actor::self()); + } - req->print_request("Finishing"); - MPI_Datatype datatype = req->old_type_; - -// FIXME Handle the case of a partial shared malloc. - if (((req->flags_ & MPI_REQ_ACCUMULATE) != 0) || - (datatype->flags() & DT_FLAG_DERIVED)) { // && (not smpi_is_shared(req->old_buf_))){ - - if (not smpi_process()->replaying() && smpi_privatize_global_variables != SmpiPrivStrategies::NONE && - static_cast(req->old_buf_) >= smpi_data_exe_start && - static_cast(req->old_buf_) < smpi_data_exe_start + smpi_data_exe_size) { - XBT_VERB("Privatization : We are unserializing to a zone in global memory Switch data segment "); - smpi_switch_data_segment(simgrid::s4u::Actor::self()); - } - - if(datatype->flags() & DT_FLAG_DERIVED){ - // This part handles the problem of non-contignous memory the unserialization at the reception - if ((req->flags_ & MPI_REQ_RECV) && datatype->size() != 0) - datatype->unserialize(req->buf_, req->old_buf_, req->real_size_/datatype->size() , req->op_); - xbt_free(req->buf_); - } else if (req->flags_ & MPI_REQ_RECV) { // apply op on contiguous buffer for accumulate - if (datatype->size() != 0) { - int n = req->real_size_ / datatype->size(); - req->op_->apply(req->buf_, req->old_buf_, &n, datatype); + if(datatype->flags() & DT_FLAG_DERIVED){ + // This part handles the problem of non-contignous memory the unserialization at the reception + if ((req->flags_ & MPI_REQ_RECV) && datatype->size() != 0) + datatype->unserialize(req->buf_, req->old_buf_, req->real_size_/datatype->size() , req->op_); + xbt_free(req->buf_); + } else if (req->flags_ & MPI_REQ_RECV) { // apply op on contiguous buffer for accumulate + if (datatype->size() != 0) { + int n = req->real_size_ / datatype->size(); + req->op_->apply(req->buf_, req->old_buf_, &n, datatype); + } + xbt_free(req->buf_); } - xbt_free(req->buf_); } } } @@ -875,7 +885,7 @@ int Request::wait(MPI_Request * request, MPI_Status * status) int count=(*request)->size_/ (*request)->old_type_->size(); (*request)->op_->apply(buf, (*request)->buf_, &count, (*request)->old_type_); } - smpi_free_tmp_buffer(buf); + smpi_free_tmp_buffer(static_cast(buf)); } } if((*request)->nbc_requests_[i]!=MPI_REQUEST_NULL) @@ -898,7 +908,7 @@ int Request::wait(MPI_Request * request, MPI_Status * status) try{ // this is not a detached send simcall_comm_wait((*request)->action_, -1.0); - }catch (xbt_ex& e) { + } catch (const Exception&) { XBT_VERB("Request cancelled"); } } @@ -961,8 +971,8 @@ int Request::waitany(int count, MPI_Request requests[], MPI_Status * status) try{ // this is not a detached send i = simcall_comm_waitany(comms.data(), comms.size(), -1); - }catch (xbt_ex& e) { - XBT_INFO("request %d cancelled ",i); + } catch (const Exception&) { + XBT_INFO("request %d cancelled ", i); return i; } @@ -1059,6 +1069,8 @@ int Request::waitsome(int incount, MPI_Request requests[], int *indices, MPI_Sta indices[count] = index; count++; for (int i = 0; i < incount; i++) { + if (i==index) + continue; if (requests[i] != MPI_REQUEST_NULL) { test(&requests[i], pstat,&flag); if (flag==1){ @@ -1135,7 +1147,7 @@ int Request::grequest_start( MPI_Grequest_query_function *query_fn, MPI_Grequest (*request)->flags_ |= MPI_REQ_GENERALIZED; (*request)->flags_ |= MPI_REQ_PERSISTENT; (*request)->refcount_ = 1; - ((*request)->generalized_funcs)=xbt_new0(s_smpi_mpi_generalized_request_funcs_t ,1); + ((*request)->generalized_funcs) = new s_smpi_mpi_generalized_request_funcs_t; ((*request)->generalized_funcs)->query_fn=query_fn; ((*request)->generalized_funcs)->free_fn=free_fn; ((*request)->generalized_funcs)->cancel_fn=cancel_fn; @@ -1156,8 +1168,13 @@ int Request::grequest_complete( MPI_Request request){ } void Request::set_nbc_requests(MPI_Request* reqs, int size){ - nbc_requests_=reqs; - nbc_requests_size_=size; + nbc_requests_size_ = size; + if (size > 0) { + nbc_requests_ = reqs; + } else { + delete[] reqs; + nbc_requests_ = nullptr; + } } int Request::get_nbc_requests_size(){