X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/96cedde3cdbc0b8ffc3f096a1b65d021b0226f99..1530fcaf86c1f6ef04bcd8e6163c9f85d1c41abd:/src/smpi/mpi/smpi_request.cpp diff --git a/src/smpi/mpi/smpi_request.cpp b/src/smpi/mpi/smpi_request.cpp index 227857cd19..c9d1068575 100644 --- a/src/smpi/mpi/smpi_request.cpp +++ b/src/smpi/mpi/smpi_request.cpp @@ -1,4 +1,4 @@ -/* Copyright (c) 2007-2019. The SimGrid Team. All rights reserved. */ +/* Copyright (c) 2007-2022. The SimGrid Team. All rights reserved. */ /* This program is free software; you can redistribute it and/or modify it * under the terms of the license (GNU LGPL) which comes with this package. */ @@ -8,18 +8,22 @@ #include "mc/mc.h" #include "private.hpp" #include "simgrid/Exception.hpp" +#include "simgrid/s4u/ConditionVariable.hpp" #include "simgrid/s4u/Exec.hpp" +#include "simgrid/s4u/Mutex.hpp" #include "smpi_comm.hpp" #include "smpi_datatype.hpp" #include "smpi_host.hpp" #include "smpi_op.hpp" +#include "src/kernel/EngineImpl.hpp" #include "src/kernel/activity/CommImpl.hpp" +#include "src/kernel/actor/ActorImpl.hpp" +#include "src/kernel/actor/SimcallObserver.hpp" #include "src/mc/mc_replay.hpp" -#include "src/simix/ActorImpl.hpp" #include "src/smpi/include/smpi_actor.hpp" -#include "xbt/config.hpp" #include +#include XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_request, smpi, "Logging specific to SMPI (request)"); @@ -28,217 +32,256 @@ static simgrid::config::Flag smpi_iprobe_sleep( static simgrid::config::Flag smpi_test_sleep( "smpi/test", "Minimum time to inject inside a call to MPI_Test", 1e-4); -std::vector smpi_ois_values; - -extern void (*smpi_comm_copy_data_callback) (smx_activity_t, void*, size_t); - -namespace simgrid{ -namespace smpi{ - -Request::Request(void* buf, int count, MPI_Datatype datatype, int src, int dst, int tag, MPI_Comm comm, unsigned flags) - : buf_(buf), old_type_(datatype), src_(src), dst_(dst), tag_(tag), comm_(comm), flags_(flags) +extern std::function smpi_comm_copy_data_callback; + +namespace simgrid::smpi { + +Request::Request(const void* buf, int count, MPI_Datatype datatype, aid_t src, aid_t dst, int tag, MPI_Comm comm, + unsigned flags, MPI_Op op) + : buf_(const_cast(buf)) + , old_buf_(buf_) + , type_(datatype) + , size_(datatype->size() * count) + , 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; - if (count==0){ - buf_ = nullptr; - }else { - buf_ = xbt_malloc(count*datatype->size()); - if ((datatype->flags() & DT_FLAG_DERIVED) && ((flags & MPI_REQ_SEND) != 0)) { - datatype->serialize(old_buf, buf_, count); - } - } - } - // This part handles the problem of non-contiguous memory (for the unserialisation at the reception) - old_buf_ = old_buf; - size_ = datatype->size() * count; datatype->ref(); comm_->ref(); + if(op != MPI_REPLACE && op != MPI_OP_NULL) + op_->ref(); action_ = nullptr; - detached_ = 0; + detached_ = false; detached_sender_ = nullptr; real_src_ = 0; - truncated_ = 0; + // get src_host if it's available (src is valid) + if (auto src_process = simgrid::s4u::Actor::by_pid(src)) + src_host_ = src_process->get_host(); + truncated_ = false; + unmatched_types_ = false; real_size_ = 0; real_tag_ = 0; if (flags & MPI_REQ_PERSISTENT) refcount_ = 1; else refcount_ = 0; - op_ = MPI_REPLACE; - cancelled_ = 0; + message_id_ = 0; + init_buffer(count); + this->add_f(); } -MPI_Comm Request::comm(){ - return comm_; -} - -int Request::src(){ - return src_; -} - -int Request::dst(){ - return dst_; +void Request::ref(){ + refcount_++; } -int Request::tag(){ - return tag_; -} +void Request::unref(MPI_Request* request) +{ + xbt_assert(*request != MPI_REQUEST_NULL, "freeing an already free request"); -int Request::flags(){ - return flags_; -} + (*request)->refcount_--; + if ((*request)->refcount_ < 0) { + (*request)->print_request("wrong refcount"); + xbt_die("Whoops, wrong refcount"); + } + if ((*request)->refcount_ == 0) { + if ((*request)->flags_ & MPI_REQ_GENERALIZED) { + ((*request)->generalized_funcs)->free_fn(((*request)->generalized_funcs)->extra_state); + } else { + Comm::unref((*request)->comm_); + Datatype::unref((*request)->type_); + } + if ((*request)->op_ != MPI_REPLACE && (*request)->op_ != MPI_OP_NULL) + Op::unref(&(*request)->op_); -int Request::detached(){ - return detached_; + (*request)->print_request("Destroying"); + F2C::free_f((*request)->f2c_id()); + delete *request; + *request = MPI_REQUEST_NULL; + } else { + (*request)->print_request("Decrementing"); + } } -MPI_Datatype Request::type(){ - return old_type_; +bool Request::match_types(MPI_Datatype stype, MPI_Datatype rtype){ + bool match = false; + if ((stype == rtype) || + //byte and packed always match with anything + (stype == MPI_PACKED || rtype == MPI_PACKED || stype == MPI_BYTE || rtype == MPI_BYTE) || + //complex datatypes - we don't properly match these yet, as it would mean checking each subtype recursively. + (stype->flags() & DT_FLAG_DERIVED || rtype->flags() & DT_FLAG_DERIVED) || + //duplicated datatypes, check if underlying is ok + (stype->duplicated_datatype()!=MPI_DATATYPE_NULL && match_types(stype->duplicated_datatype(), rtype)) || + (rtype->duplicated_datatype()!=MPI_DATATYPE_NULL && match_types(stype, rtype->duplicated_datatype()))) + match = true; + if (not match) + XBT_WARN("Mismatched datatypes : sending %s and receiving %s", stype->name().c_str(), rtype->name().c_str()); + return match; } -size_t Request::size(){ - return size_; -} -size_t Request::real_size(){ - return real_size_; +bool Request::match_common(MPI_Request req, MPI_Request sender, MPI_Request receiver) +{ + xbt_assert(sender, "Cannot match against null sender"); + xbt_assert(receiver, "Cannot match against null receiver"); + XBT_DEBUG("Trying to match %s of sender src %ld against %ld, tag %d against %d, id %d against %d", + (req == receiver ? "send" : "recv"), sender->src_, receiver->src_, sender->tag_, receiver->tag_, + sender->comm_->id(), receiver->comm_->id()); + + if ((receiver->comm_->id() == MPI_UNDEFINED || sender->comm_->id() == MPI_UNDEFINED || + receiver->comm_->id() == sender->comm_->id()) && + ((receiver->src_ == MPI_ANY_SOURCE && (receiver->comm_->group()->rank(sender->src_) != MPI_UNDEFINED)) || + receiver->src_ == sender->src_) && + ((receiver->tag_ == MPI_ANY_TAG && sender->tag_ >= 0) || receiver->tag_ == sender->tag_)) { + // we match, we can transfer some values + if (receiver->src_ == MPI_ANY_SOURCE) { + receiver->real_src_ = sender->src_; + receiver->src_host_ = sender->src_host_; + } + if (receiver->tag_ == MPI_ANY_TAG) + receiver->real_tag_ = sender->tag_; + if ((receiver->flags_ & MPI_REQ_PROBE) == 0 && receiver->real_size_ < sender->real_size_) { + XBT_DEBUG("Truncating message - should not happen: receiver size : %zu < sender size : %zu", receiver->real_size_, + sender->real_size_); + receiver->truncated_ = true; + } + //0-sized datatypes/counts should not interfere and match + if (sender->real_size_ != 0 && receiver->real_size_ != 0 && not match_types(sender->type_, receiver->type_)) + receiver->unmatched_types_ = true; + if (sender->detached_) + receiver->detached_sender_ = sender; // tie the sender to the receiver, as it is detached and has to be freed in + // the receiver + req->flags_ |= MPI_REQ_MATCHED; // mark as impossible to cancel anymore + XBT_DEBUG("match succeeded"); + return true; + } + return false; } -void Request::ref(){ - refcount_++; +void Request::init_buffer(int count){ +// FIXME Handle the case of a partial shared malloc. + // This part handles the problem of non-contiguous memory (for the unserialization at the reception) + if (not smpi_process()->replaying() && + ((((flags_ & MPI_REQ_RECV) != 0) && ((flags_ & MPI_REQ_ACCUMULATE) != 0)) || (type_->flags() & DT_FLAG_DERIVED))) { + // This part handles the problem of non-contiguous memory + old_buf_ = buf_; + if (count==0){ + buf_ = nullptr; + }else { + buf_ = xbt_malloc(count*type_->size()); + if ((type_->flags() & DT_FLAG_DERIVED) && ((flags_ & MPI_REQ_SEND) != 0)) { + type_->serialize(old_buf_, buf_, count); + } + } + } } -void Request::unref(MPI_Request* request) +bool Request::match_recv(void* a, void* b, simgrid::kernel::activity::CommImpl*) { - if((*request) != MPI_REQUEST_NULL){ - (*request)->refcount_--; - if((*request)->refcount_ < 0) { - (*request)->print_request("wrong refcount"); - xbt_die("Whoops, wrong refcount"); - } - if((*request)->refcount_==0){ - Datatype::unref((*request)->old_type_); - Comm::unref((*request)->comm_); - (*request)->print_request("Destroying"); - delete *request; - *request = MPI_REQUEST_NULL; - }else{ - (*request)->print_request("Decrementing"); + auto ref = static_cast(a); + auto req = static_cast(b); + bool match = match_common(req, req, ref); + if (not match || ref->comm_ == MPI_COMM_UNINITIALIZED || ref->comm_->is_smp_comm()) + return match; + + if (ref->comm_->get_received_messages_count(ref->comm_->group()->rank(req->src_), + ref->comm_->group()->rank(req->dst_), req->tag_) == req->message_id_) { + if (((ref->flags_ & MPI_REQ_PROBE) == 0) && ((req->flags_ & MPI_REQ_PROBE) == 0)) { + XBT_DEBUG("increasing count in comm %p, which was %u from pid %ld, to pid %ld with tag %d", ref->comm_, + ref->comm_->get_received_messages_count(ref->comm_->group()->rank(req->src_), + ref->comm_->group()->rank(req->dst_), req->tag_), + req->src_, req->dst_, req->tag_); + ref->comm_->increment_received_messages_count(ref->comm_->group()->rank(req->src_), + ref->comm_->group()->rank(req->dst_), req->tag_); + if (ref->real_size_ > req->real_size_) { + ref->real_size_ = req->real_size_; + } } - }else{ - xbt_die("freeing an already free request"); - } -} - -int Request::match_recv(void* a, void* b, simgrid::kernel::activity::CommImpl* ignored) -{ - 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_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_) - && ((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) - ref->real_src_ = req->src_; - if(ref->tag_ == MPI_ANY_TAG) - ref->real_tag_ = req->tag_; - if(ref->real_size_ < req->real_size_) - ref->truncated_ = 1; - if(req->detached_==1) - ref->detached_sender_=req; //tie the sender to the receiver, as it is detached and has to be freed in the receiver - if(req->cancelled_==0) - req->cancelled_=-1;//mark as uncancellable - XBT_DEBUG("match succeeded"); - return 1; - }else return 0; -} - -int Request::match_send(void* a, void* b, simgrid::kernel::activity::CommImpl* ignored) -{ - 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_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_) - && ((req->tag_ == MPI_ANY_TAG && ref->tag_ >=0)|| req->tag_ == ref->tag_)){ - if(req->src_ == MPI_ANY_SOURCE) - req->real_src_ = ref->src_; - if(req->tag_ == MPI_ANY_TAG) - req->real_tag_ = ref->tag_; - if(req->real_size_ < ref->real_size_) - req->truncated_ = 1; - if(ref->detached_==1) - req->detached_sender_=ref; //tie the sender to the receiver, as it is detached and has to be freed in the receiver - if(req->cancelled_==0) - req->cancelled_=-1;//mark as uncancellable - XBT_DEBUG("match succeeded"); - return 1; - } else - return 0; + } else { + match = false; + req->flags_ &= ~MPI_REQ_MATCHED; + ref->detached_sender_ = nullptr; + XBT_DEBUG("Refusing to match message, as its ID is not the one I expect. in comm %p, %u != %u, " + "from pid %ld to pid %ld, with tag %d", + ref->comm_, + ref->comm_->get_received_messages_count(ref->comm_->group()->rank(req->src_), + ref->comm_->group()->rank(req->dst_), req->tag_), + req->message_id_, req->src_, req->dst_, req->tag_); + } + return match; } -void Request::print_request(const char *message) +bool Request::match_send(void* a, void* b, simgrid::kernel::activity::CommImpl*) { - XBT_VERB("%s request %p [buf = %p, size = %zu, src = %d, dst = %d, tag = %d, flags = %x]", - message, this, buf_, size_, src_, dst_, tag_, flags_); + auto ref = static_cast(a); + auto req = static_cast(b); + return match_common(req, ref, req); } +void Request::print_request(const char* message) const +{ + XBT_VERB("%s request %p [buf = %p, size = %zu, src = %ld, dst = %ld, tag = %d, flags = %x]", message, this, buf_, + size_, src_, dst_, tag_, flags_); +} /* 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::bsend_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(), + dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm, + MPI_REQ_PERSISTENT | MPI_REQ_SEND | MPI_REQ_PREPARED | MPI_REQ_BSEND); +} +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(), - comm->group()->actor(dst)->get_pid(), tag, comm, + dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm, 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, + dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, 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, + dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm, MPI_REQ_PERSISTENT | MPI_REQ_ISEND | MPI_REQ_SEND | MPI_REQ_PREPARED); } - -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 */ + MPI_Request request; if(op==MPI_OP_NULL){ - request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, comm->group()->actor(src)->get_pid(), - comm->group()->actor(dst)->get_pid(), tag, comm, + request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, comm->group()->actor(src), + dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm, MPI_REQ_RMA | MPI_REQ_NON_PERSISTENT | MPI_REQ_ISEND | MPI_REQ_SEND | MPI_REQ_PREPARED); }else{ - request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, comm->group()->actor(src)->get_pid(), - comm->group()->actor(dst)->get_pid(), tag, comm, + request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, comm->group()->actor(src), + dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm, MPI_REQ_RMA | MPI_REQ_NON_PERSISTENT | MPI_REQ_ISEND | MPI_REQ_SEND | MPI_REQ_PREPARED | - MPI_REQ_ACCUMULATE); - request->op_ = op; + MPI_REQ_ACCUMULATE, + op); } return request; } MPI_Request Request::recv_init(void *buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm) { + aid_t source = MPI_PROC_NULL; + if (src == MPI_ANY_SOURCE) + source = MPI_ANY_SOURCE; + else if (src != MPI_PROC_NULL) + source = comm->group()->actor(src); return new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, - src == MPI_ANY_SOURCE ? MPI_ANY_SOURCE : comm->group()->actor(src)->get_pid(), + source, simgrid::s4u::this_actor::get_pid(), tag, comm, MPI_REQ_PERSISTENT | MPI_REQ_RECV | MPI_REQ_PREPARED); } @@ -246,101 +289,141 @@ MPI_Request Request::recv_init(void *buf, int count, MPI_Datatype datatype, int MPI_Request Request::rma_recv_init(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 */ + aid_t source = MPI_PROC_NULL; + if (src == MPI_ANY_SOURCE) + source = MPI_ANY_SOURCE; + else if (src != MPI_PROC_NULL) + source = comm->group()->actor(src); + MPI_Request request; if(op==MPI_OP_NULL){ - request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, comm->group()->actor(src)->get_pid(), - comm->group()->actor(dst)->get_pid(), tag, comm, + request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, source, + dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm, MPI_REQ_RMA | MPI_REQ_NON_PERSISTENT | MPI_REQ_RECV | MPI_REQ_PREPARED); }else{ - request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, comm->group()->actor(src)->get_pid(), - comm->group()->actor(dst)->get_pid(), tag, comm, - MPI_REQ_RMA | MPI_REQ_NON_PERSISTENT | MPI_REQ_RECV | MPI_REQ_PREPARED | MPI_REQ_ACCUMULATE); - request->op_ = op; + request = + new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, source, + dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm, + MPI_REQ_RMA | MPI_REQ_NON_PERSISTENT | MPI_REQ_RECV | MPI_REQ_PREPARED | MPI_REQ_ACCUMULATE, op); } return request; } MPI_Request Request::irecv_init(void *buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm) { + aid_t source = MPI_PROC_NULL; + if (src == MPI_ANY_SOURCE) + source = MPI_ANY_SOURCE; + else if (src != MPI_PROC_NULL) + source = comm->group()->actor(src); return new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, - src == MPI_ANY_SOURCE ? MPI_ANY_SOURCE : comm->group()->actor(src)->get_pid(), - simgrid::s4u::this_actor::get_pid(), tag, comm, + source, simgrid::s4u::this_actor::get_pid(), tag, comm, 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::ibsend(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(), - comm->group()->actor(dst)->get_pid(), tag, comm, - MPI_REQ_NON_PERSISTENT | MPI_REQ_ISEND | MPI_REQ_SEND); - request->start(); + auto request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(), + dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm, + MPI_REQ_NON_PERSISTENT | MPI_REQ_ISEND | MPI_REQ_SEND | MPI_REQ_BSEND); + if(dst != MPI_PROC_NULL) + request->start(); return request; } -MPI_Request Request::issend(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(), - comm->group()->actor(dst)->get_pid(), tag, comm, - MPI_REQ_NON_PERSISTENT | MPI_REQ_ISEND | MPI_REQ_SSEND | MPI_REQ_SEND); - request->start(); + auto request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(), + dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm, + MPI_REQ_NON_PERSISTENT | MPI_REQ_ISEND | MPI_REQ_SEND); + if(dst != MPI_PROC_NULL) + request->start(); return request; } +MPI_Request Request::issend(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) +{ + auto request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(), + dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm, + MPI_REQ_NON_PERSISTENT | MPI_REQ_ISEND | MPI_REQ_SSEND | MPI_REQ_SEND); + if(dst != MPI_PROC_NULL) + request->start(); + return request; +} MPI_Request Request::irecv(void *buf, int count, MPI_Datatype datatype, int src, 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, - src == MPI_ANY_SOURCE ? MPI_ANY_SOURCE : comm->group()->actor(src)->get_pid(), - simgrid::s4u::this_actor::get_pid(), tag, comm, MPI_REQ_NON_PERSISTENT | MPI_REQ_RECV); - request->start(); + aid_t source = MPI_PROC_NULL; + if (src == MPI_ANY_SOURCE) + source = MPI_ANY_SOURCE; + else if (src != MPI_PROC_NULL) + source = comm->group()->actor(src); + auto request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, source, + simgrid::s4u::this_actor::get_pid(), tag, comm, MPI_REQ_NON_PERSISTENT | MPI_REQ_RECV); + if(src != MPI_PROC_NULL) + request->start(); return request; } -void Request::recv(void *buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm, MPI_Status * status) +int Request::recv(void *buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm, MPI_Status * status) { - MPI_Request request = nullptr; /* MC needs the comm to be set to nullptr during the call */ - request = irecv(buf, count, datatype, src, tag, comm); - wait(&request,status); + MPI_Request request = irecv(buf, count, datatype, src, tag, comm); + int retval = wait(&request,status); request = nullptr; + return retval; } -void Request::send(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) +void Request::bsend(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(), - comm->group()->actor(dst)->get_pid(), tag, comm, MPI_REQ_NON_PERSISTENT | MPI_REQ_SEND); + auto request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(), + dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm, + MPI_REQ_NON_PERSISTENT | MPI_REQ_SEND | MPI_REQ_BSEND); - request->start(); + if(dst != MPI_PROC_NULL) + request->start(); + wait(&request, MPI_STATUS_IGNORE); + request = nullptr; +} + +void Request::send(const void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) +{ + auto request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(), + dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm, + MPI_REQ_NON_PERSISTENT | MPI_REQ_SEND); + if(dst != MPI_PROC_NULL) + request->start(); wait(&request, MPI_STATUS_IGNORE); 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(), - comm->group()->actor(dst)->get_pid(), tag, comm, - MPI_REQ_NON_PERSISTENT | MPI_REQ_SSEND | MPI_REQ_SEND); + auto request = new Request(buf == MPI_BOTTOM ? nullptr : buf, count, datatype, simgrid::s4u::this_actor::get_pid(), + dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL, tag, comm, + MPI_REQ_NON_PERSISTENT | MPI_REQ_SSEND | MPI_REQ_SEND); - request->start(); + if(dst != MPI_PROC_NULL) + request->start(); wait(&request,MPI_STATUS_IGNORE); 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) { - MPI_Request requests[2]; - MPI_Status stats[2]; - int myid = simgrid::s4u::this_actor::get_pid(); - if ((comm->group()->actor(dst)->get_pid() == myid) && (comm->group()->actor(src)->get_pid() == myid)) { + aid_t source = MPI_PROC_NULL; + if (src == MPI_ANY_SOURCE) + source = MPI_ANY_SOURCE; + else if (src != MPI_PROC_NULL) + source = comm->group()->actor(src); + aid_t destination = dst != MPI_PROC_NULL ? comm->group()->actor(dst) : MPI_PROC_NULL; + + std::array requests; + std::array stats; + if (aid_t myid = simgrid::s4u::this_actor::get_pid(); (destination == myid) && (source == myid)) { Datatype::copy(sendbuf, sendcount, sendtype, recvbuf, recvcount, recvtype); if (status != MPI_STATUS_IGNORE) { - status->MPI_SOURCE = src; + status->MPI_SOURCE = source; status->MPI_TAG = recvtag; status->MPI_ERROR = MPI_SUCCESS; status->count = sendcount * sendtype->size(); @@ -349,8 +432,8 @@ void Request::sendrecv(void *sendbuf, int sendcount, MPI_Datatype sendtype,int d } requests[0] = isend_init(sendbuf, sendcount, sendtype, dst, sendtag, comm); requests[1] = irecv_init(recvbuf, recvcount, recvtype, src, recvtag, comm); - startall(2, requests); - waitall(2, requests, stats); + startall(2, requests.data()); + waitall(2, requests.data(), stats.data()); unref(&requests[0]); unref(&requests[1]); if(status != MPI_STATUS_IGNORE) { @@ -361,48 +444,57 @@ void Request::sendrecv(void *sendbuf, int sendcount, MPI_Datatype sendtype,int d void Request::start() { - smx_mailbox_t mailbox; + s4u::Mailbox* mailbox; xbt_assert(action_ == nullptr, "Cannot (re-)start unfinished communication"); + //reinitialize temporary buffer for persistent requests + if(real_size_ > 0 && flags_ & MPI_REQ_FINISHED){ + buf_ = old_buf_; + init_buffer(real_size_/type_->size()); + } flags_ &= ~MPI_REQ_PREPARED; flags_ &= ~MPI_REQ_FINISHED; this->ref(); + // we make a copy here, as the size is modified by simix, and we may reuse the request in another receive later + real_size_=size_; if ((flags_ & MPI_REQ_RECV) != 0) { this->print_request("New recv"); simgrid::smpi::ActorExt* process = smpi_process_remote(simgrid::s4u::Actor::by_pid(dst_)); - int async_small_thresh = simgrid::config::get_value("smpi/async-small-thresh"); + simgrid::s4u::MutexPtr mut = process->mailboxes_mutex(); + if (smpi_cfg_async_small_thresh() != 0 || (flags_ & MPI_REQ_RMA) != 0) + mut->lock(); - xbt_mutex_t mut = process->mailboxes_mutex(); - if (async_small_thresh != 0 || (flags_ & MPI_REQ_RMA) != 0) - xbt_mutex_acquire(mut); + bool is_probe = ((flags_ & MPI_REQ_PROBE) != 0); + flags_ |= MPI_REQ_PROBE; - if (async_small_thresh == 0 && (flags_ & MPI_REQ_RMA) == 0) { + if (smpi_cfg_async_small_thresh() == 0 && (flags_ & MPI_REQ_RMA) == 0) { mailbox = process->mailbox(); - } else if (((flags_ & MPI_REQ_RMA) != 0) || static_cast(size_) < async_small_thresh) { + } else if (((flags_ & MPI_REQ_RMA) != 0) || static_cast(size_) < smpi_cfg_async_small_thresh()) { //We have to check both mailboxes (because SSEND messages are sent to the large mbox). //begin with the more appropriate one : the small one. mailbox = process->mailbox_small(); - XBT_DEBUG("Is there a corresponding send already posted in the small mailbox %p (in case of SSEND)?", mailbox); - smx_activity_t action = simcall_comm_iprobe(mailbox, 0, &match_recv, static_cast(this)); + XBT_DEBUG("Is there a corresponding send already posted in the small mailbox %s (in case of SSEND)?", + mailbox->get_cname()); + simgrid::kernel::activity::ActivityImplPtr action = mailbox->iprobe(0, &match_recv, static_cast(this)); if (action == nullptr) { mailbox = process->mailbox(); - XBT_DEBUG("No, nothing in the small mailbox test the other one : %p", mailbox); - action = simcall_comm_iprobe(mailbox, 0, &match_recv, static_cast(this)); + XBT_DEBUG("No, nothing in the small mailbox test the other one : %s", mailbox->get_cname()); + action = mailbox->iprobe(0, &match_recv, static_cast(this)); if (action == nullptr) { - XBT_DEBUG("Still nothing, switch back to the small mailbox : %p", mailbox); + XBT_DEBUG("Still nothing, switch back to the small mailbox : %s", mailbox->get_cname()); mailbox = process->mailbox_small(); } } else { - XBT_DEBUG("yes there was something for us in the large mailbox"); + XBT_DEBUG("yes there was something for us in the small mailbox"); } } else { mailbox = process->mailbox_small(); XBT_DEBUG("Is there a corresponding send already posted the small mailbox?"); - smx_activity_t action = simcall_comm_iprobe(mailbox, 0, &match_recv, static_cast(this)); + simgrid::kernel::activity::ActivityImplPtr action = mailbox->iprobe(0, &match_recv, static_cast(this)); if (action == nullptr) { XBT_DEBUG("No, nothing in the permanent receive mailbox"); @@ -411,41 +503,52 @@ void Request::start() XBT_DEBUG("yes there was something for us in the small mailbox"); } } + if (not is_probe) + flags_ &= ~MPI_REQ_PROBE; + kernel::actor::CommIrecvSimcall observer{process->get_actor()->get_impl(), + mailbox->get_impl(), + static_cast(buf_), + &real_size_, + &match_recv, + process->replaying() ? &smpi_comm_null_copy_buffer_callback + : smpi_comm_copy_data_callback, + this, + -1.0}; + observer.set_tag(tag_); + + action_ = kernel::actor::simcall_answered([&observer] { return kernel::activity::CommImpl::irecv(&observer); }, + &observer); - // we make a copy here, as the size is modified by simix, and we may reuse the request in another receive later - real_size_=size_; - action_ = simcall_comm_irecv( - process->get_actor()->get_impl(), mailbox, buf_, &real_size_, &match_recv, - process->replaying() ? &smpi_comm_null_copy_buffer_callback : smpi_comm_copy_data_callback, this, -1.0); XBT_DEBUG("recv simcall posted"); - if (async_small_thresh != 0 || (flags_ & MPI_REQ_RMA) != 0) - xbt_mutex_release(mut); + if (smpi_cfg_async_small_thresh() != 0 || (flags_ & MPI_REQ_RMA) != 0) + mut->unlock(); } else { /* the RECV flag was not set, so this is a send */ - simgrid::smpi::ActorExt* process = smpi_process_remote(simgrid::s4u::Actor::by_pid(dst_)); - int rank = src_; - if (TRACE_smpi_view_internals()) { - TRACE_smpi_send(rank, rank, dst_, tag_, size_); - } + const simgrid::smpi::ActorExt* process = smpi_process_remote(simgrid::s4u::Actor::by_pid(dst_)); + xbt_assert(process, "Actor pid=%ld is gone??", dst_); + if (TRACE_smpi_view_internals()) + TRACE_smpi_send(src_, src_, dst_, tag_, size_); this->print_request("New send"); + message_id_=comm_->get_sent_messages_count(comm_->group()->rank(src_), comm_->group()->rank(dst_), tag_); + comm_->increment_sent_messages_count(comm_->group()->rank(src_), comm_->group()->rank(dst_), tag_); + void* buf = buf_; if ((flags_ & MPI_REQ_SSEND) == 0 && - ((flags_ & MPI_REQ_RMA) != 0 || - static_cast(size_) < simgrid::config::get_value("smpi/send-is-detached-thresh"))) { + ((flags_ & MPI_REQ_RMA) != 0 || (flags_ & MPI_REQ_BSEND) != 0 || + static_cast(size_) < smpi_cfg_detached_send_thresh())) { void *oldbuf = nullptr; - detached_ = 1; + detached_ = true; XBT_DEBUG("Send request %p is detached", this); this->ref(); - if (not(old_type_->flags() & DT_FLAG_DERIVED)) { + if (not(type_->flags() & DT_FLAG_DERIVED)) { oldbuf = buf_; if (not process->replaying() && oldbuf != nullptr && size_ != 0) { - if ((smpi_privatize_global_variables != SmpiPrivStrategies::NONE) && - (static_cast(buf_) >= smpi_data_exe_start) && - (static_cast(buf_) < smpi_data_exe_start + smpi_data_exe_size)) { + if (smpi_switch_data_segment(simgrid::s4u::Actor::by_pid(src_), buf_)) XBT_DEBUG("Privatization : We are sending from a zone inside global memory. Switch data segment "); - smpi_switch_data_segment(simgrid::s4u::Actor::by_pid(src_)); - } + + //we need this temporary buffer even for bsend, as it will be released in the copy callback and we don't have a way to differentiate it + //so actually ... don't use manually attached buffer space. buf = xbt_malloc(size_); memcpy(buf,oldbuf,size_); XBT_DEBUG("buf %p copied into %p",oldbuf,buf); @@ -455,40 +558,45 @@ void Request::start() //if we are giving back the control to the user without waiting for completion, we have to inject timings double sleeptime = 0.0; - if (detached_ != 0 || ((flags_ & (MPI_REQ_ISEND | MPI_REQ_SSEND)) != 0)) { // issend should be treated as isend + if (detached_ || ((flags_ & (MPI_REQ_ISEND | MPI_REQ_SSEND)) != 0)) { // issend should be treated as isend // isend and send timings may be different - sleeptime = ((flags_ & MPI_REQ_ISEND) != 0) - ? simgrid::s4u::Actor::self()->get_host()->extension()->oisend(size_) - : simgrid::s4u::Actor::self()->get_host()->extension()->osend(size_); + sleeptime = + ((flags_ & MPI_REQ_ISEND) != 0) + ? simgrid::s4u::Actor::self()->get_host()->extension()->oisend( + size_, simgrid::s4u::Actor::by_pid(src_)->get_host(), simgrid::s4u::Actor::by_pid(dst_)->get_host()) + : simgrid::s4u::Actor::self()->get_host()->extension()->osend( + size_, simgrid::s4u::Actor::by_pid(src_)->get_host(), + simgrid::s4u::Actor::by_pid(dst_)->get_host()); } if(sleeptime > 0.0){ - simcall_process_sleep(sleeptime); + simgrid::s4u::this_actor::sleep_for(sleeptime); XBT_DEBUG("sending size of %zu : sleep %f ", size_, sleeptime); } - int async_small_thresh = simgrid::config::get_value("smpi/async-small-thresh"); - - xbt_mutex_t mut=process->mailboxes_mutex(); + simgrid::s4u::MutexPtr mut = process->mailboxes_mutex(); - if (async_small_thresh != 0 || (flags_ & MPI_REQ_RMA) != 0) - xbt_mutex_acquire(mut); + if (smpi_cfg_async_small_thresh() != 0 || (flags_ & MPI_REQ_RMA) != 0) + mut->lock(); - if (not(async_small_thresh != 0 || (flags_ & MPI_REQ_RMA) != 0)) { + if (not(smpi_cfg_async_small_thresh() != 0 || (flags_ & MPI_REQ_RMA) != 0)) { mailbox = process->mailbox(); - } else if (((flags_ & MPI_REQ_RMA) != 0) || static_cast(size_) < async_small_thresh) { // eager mode + } else if (((flags_ & MPI_REQ_RMA) != 0) || static_cast(size_) < smpi_cfg_async_small_thresh()) { // eager mode + bool is_probe = ((flags_ & MPI_REQ_PROBE) != 0); + flags_ |= MPI_REQ_PROBE; + mailbox = process->mailbox(); - XBT_DEBUG("Is there a corresponding recv already posted in the large mailbox %p?", mailbox); - smx_activity_t action = simcall_comm_iprobe(mailbox, 1, &match_send, static_cast(this)); - if (action == nullptr) { + XBT_DEBUG("Is there a corresponding recv already posted in the large mailbox %s?", mailbox->get_cname()); + if (not mailbox->iprobe(1, &match_send, static_cast(this))) { if ((flags_ & MPI_REQ_SSEND) == 0) { mailbox = process->mailbox_small(); - XBT_DEBUG("No, nothing in the large mailbox, message is to be sent on the small one %p", mailbox); + XBT_DEBUG("No, nothing in the large mailbox, message is to be sent on the small one %s", + mailbox->get_cname()); } else { mailbox = process->mailbox_small(); - XBT_DEBUG("SSEND : Is there a corresponding recv already posted in the small mailbox %p?", mailbox); - action = simcall_comm_iprobe(mailbox, 1, &match_send, static_cast(this)); - if (action == nullptr) { + XBT_DEBUG("SSEND : Is there a corresponding recv already posted in the small mailbox %s?", + mailbox->get_cname()); + if (not mailbox->iprobe(1, &match_send, static_cast(this))) { XBT_DEBUG("No, we are first, send to large mailbox"); mailbox = process->mailbox(); } @@ -496,26 +604,34 @@ void Request::start() } else { XBT_DEBUG("Yes there was something for us in the large mailbox"); } + if (not is_probe) + flags_ &= ~MPI_REQ_PROBE; } else { mailbox = process->mailbox(); - XBT_DEBUG("Send request %p is in the large mailbox %p (buf: %p)",mailbox, this,buf_); + XBT_DEBUG("Send request %p is in the large mailbox %s (buf: %p)", this, mailbox->get_cname(), buf_); } - // we make a copy here, as the size is modified by simix, and we may reuse the request in another receive later - real_size_=size_; - action_ = simcall_comm_isend( - simgrid::s4u::Actor::by_pid(src_)->get_impl(), mailbox, size_, -1.0, buf, real_size_, &match_send, + size_t payload_size_ = size_ + 16;//MPI enveloppe size (tag+dest+communicator) + kernel::actor::CommIsendSimcall observer{ + simgrid::kernel::EngineImpl::get_instance()->get_actor_by_pid(src_), mailbox->get_impl(), + static_cast(payload_size_), -1, static_cast(buf), real_size_, &match_send, &xbt_free_f, // how to free the userdata if a detached send fails - not process->replaying() ? smpi_comm_copy_data_callback : &smpi_comm_null_copy_buffer_callback, this, + process->replaying() ? &smpi_comm_null_copy_buffer_callback : smpi_comm_copy_data_callback, this, // detach if msg size < eager/rdv switch limit - detached_); + detached_}; + observer.set_tag(tag_); + action_ = kernel::actor::simcall_answered([&observer] { return kernel::activity::CommImpl::isend(&observer); }, + &observer); XBT_DEBUG("send simcall posted"); /* FIXME: detached sends are not traceable (action_ == nullptr) */ - if (action_ != nullptr) - simcall_set_category(action_, TRACE_internal_smpi_get_category()); - if (async_small_thresh != 0 || ((flags_ & MPI_REQ_RMA) != 0)) - xbt_mutex_release(mut); + if (action_ != nullptr) { + boost::static_pointer_cast(action_)->set_tracing_category( + smpi_process()->get_tracing_category()); + } + + if (smpi_cfg_async_small_thresh() != 0 || ((flags_ & MPI_REQ_RMA) != 0)) + mut->unlock(); } } @@ -525,64 +641,94 @@ void Request::startall(int count, MPI_Request * requests) return; for(int i = 0; i < count; i++) { - requests[i]->start(); + if(requests[i]->src_ != MPI_PROC_NULL && requests[i]->dst_ != MPI_PROC_NULL) + requests[i]->start(); } } void Request::cancel() { - if(cancelled_!=-1) - cancelled_=1; + this->flags_ |= MPI_REQ_CANCELLED; if (this->action_ != nullptr) (boost::static_pointer_cast(this->action_))->cancel(); } -int Request::test(MPI_Request * request, MPI_Status * status) { - //assume that request is not MPI_REQUEST_NULL (filtered in PMPI_Test or testall before) +int Request::test(MPI_Request * request, MPI_Status * status, int* flag) { + // assume that *request is not MPI_REQUEST_NULL (filtered in PMPI_Test or testall before) // to avoid deadlocks if used as a break condition, such as // while (MPI_Test(request, flag, status) && flag) dostuff... // because the time will not normally advance when only calls to MPI_Test are made -> deadlock // multiplier to the sleeptime, to increase speed of execution, each failed test will increase it + xbt_assert(*request != MPI_REQUEST_NULL); + static int nsleeps = 1; + int ret = MPI_SUCCESS; + if(smpi_test_sleep > 0) - simcall_process_sleep(nsleeps*smpi_test_sleep); + simgrid::s4u::this_actor::sleep_for(nsleeps * smpi_test_sleep); Status::empty(status); - int flag = 1; - if (((*request)->flags_ & MPI_REQ_PREPARED) == 0) { - if ((*request)->action_ != nullptr){ + *flag = 1; + + if ((*request)->flags_ & MPI_REQ_NBC){ + *flag = finish_nbc_requests(request, 1); + } + + if (((*request)->flags_ & (MPI_REQ_PREPARED | MPI_REQ_FINISHED)) == 0) { + if ((*request)->action_ != nullptr && ((*request)->flags_ & MPI_REQ_CANCELLED) == 0){ try{ - flag = simcall_comm_test((*request)->action_); - }catch (xbt_ex& e) { - return 0; + kernel::actor::ActorImpl* issuer = kernel::actor::ActorImpl::self(); + kernel::actor::ActivityTestSimcall observer{issuer, (*request)->action_.get()}; + *flag = kernel::actor::simcall_answered( + [&observer] { return observer.get_activity()->test(observer.get_issuer()); }, &observer); + } catch (const Exception&) { + *flag = 0; + return ret; } } - if (flag) { - finish_wait(request,status); + if (((*request)->flags_ & MPI_REQ_GENERALIZED) && not((*request)->flags_ & MPI_REQ_COMPLETE)) + *flag=0; + if (*flag) { + finish_wait(request, status); // may invalidate *request + if (*request != MPI_REQUEST_NULL && ((*request)->flags_ & MPI_REQ_GENERALIZED)){ + MPI_Status tmp_status; + MPI_Status* mystatus; + if (status == MPI_STATUS_IGNORE) { + mystatus = &tmp_status; + Status::empty(mystatus); + } else { + mystatus = status; + } + ret = ((*request)->generalized_funcs)->query_fn(((*request)->generalized_funcs)->extra_state, mystatus); + } nsleeps=1;//reset the number of sleeps we will do next time if (*request != MPI_REQUEST_NULL && ((*request)->flags_ & MPI_REQ_PERSISTENT) == 0) *request = MPI_REQUEST_NULL; - } else if (simgrid::config::get_value("smpi/grow-injected-times")) { + } else if (smpi_cfg_grow_injected_times()) { nsleeps++; } } - return flag; + return ret; } -int Request::testsome(int incount, MPI_Request requests[], int *indices, MPI_Status status[]) +int Request::testsome(int incount, MPI_Request requests[], int *count, int *indices, MPI_Status status[]) { - int count = 0; + int error=0; int count_dead = 0; + int flag = 0; MPI_Status stat; MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat; + *count = 0; for (int i = 0; i < incount; i++) { - if (requests[i] != MPI_REQUEST_NULL) { - if (test(&requests[i], pstat)) { - indices[i] = 1; - count++; + if (requests[i] != MPI_REQUEST_NULL && not (requests[i]->flags_ & MPI_REQ_FINISHED)) { + if (test(&requests[i], pstat, &flag) != MPI_SUCCESS) + error = 1; + if(flag) { + indices[*count] = i; if (status != MPI_STATUSES_IGNORE) - status[i] = *pstat; + status[*count] = *pstat; + (*count)++; if ((requests[i] != MPI_REQUEST_NULL) && (requests[i]->flags_ & MPI_REQ_NON_PERSISTENT)) requests[i] = MPI_REQUEST_NULL; } @@ -590,25 +736,26 @@ int Request::testsome(int incount, MPI_Request requests[], int *indices, MPI_Sta count_dead++; } } - if(count_dead==incount) - return MPI_UNDEFINED; - else return count; + if(count_dead==incount)*count=MPI_UNDEFINED; + if(error!=0) + return MPI_ERR_IN_STATUS; + else + return MPI_SUCCESS; } -int Request::testany(int count, MPI_Request requests[], int *index, MPI_Status * status) +int Request::testany(int count, MPI_Request requests[], int *index, int* flag, MPI_Status * status) { - std::vector comms; + std::vector comms; comms.reserve(count); - int i; - int flag = 0; - + *flag = 0; + int ret = MPI_SUCCESS; *index = MPI_UNDEFINED; std::vector map; /** Maps all matching comms back to their location in requests **/ - for(i = 0; i < count; i++) { + for (int i = 0; i < count; i++) { if ((requests[i] != MPI_REQUEST_NULL) && requests[i]->action_ && not(requests[i]->flags_ & MPI_REQ_PREPARED)) { - comms.push_back(requests[i]->action_); + comms.push_back(requests[i]->action_.get()); map.push_back(i); } } @@ -616,45 +763,81 @@ int Request::testany(int count, MPI_Request requests[], int *index, MPI_Status * //multiplier to the sleeptime, to increase speed of execution, each failed testany will increase it static int nsleeps = 1; if(smpi_test_sleep > 0) - simcall_process_sleep(nsleeps*smpi_test_sleep); + simgrid::s4u::this_actor::sleep_for(nsleeps * smpi_test_sleep); + ssize_t i; try{ - i = simcall_comm_testany(comms.data(), comms.size()); // The i-th element in comms matches! - }catch (xbt_ex& e) { + kernel::actor::ActorImpl* issuer = kernel::actor::ActorImpl::self(); + kernel::actor::ActivityTestanySimcall observer{issuer, comms}; + i = kernel::actor::simcall_answered( + [&observer] { + return kernel::activity::ActivityImpl::test_any(observer.get_issuer(), observer.get_activities()); + }, + &observer); + } catch (const Exception&) { + XBT_DEBUG("Exception in testany"); return 0; } - + if (i != -1) { // -1 is not MPI_UNDEFINED but a SIMIX return code. (nothing matches) *index = map[i]; - finish_wait(&requests[*index],status); - flag = 1; - nsleeps = 1; - if (requests[*index] != MPI_REQUEST_NULL && (requests[*index]->flags_ & MPI_REQ_NON_PERSISTENT)) { - requests[*index] = MPI_REQUEST_NULL; + if (requests[*index] != MPI_REQUEST_NULL && (requests[*index]->flags_ & MPI_REQ_GENERALIZED) && + not(requests[*index]->flags_ & MPI_REQ_COMPLETE)) { + *flag=0; + } else { + finish_wait(&requests[*index],status); + if (requests[*index] != MPI_REQUEST_NULL && (requests[*index]->flags_ & MPI_REQ_GENERALIZED)){ + MPI_Status tmp_status; + MPI_Status* mystatus; + if (status == MPI_STATUS_IGNORE) { + mystatus = &tmp_status; + Status::empty(mystatus); + } else { + mystatus = status; + } + ret=(requests[*index]->generalized_funcs)->query_fn((requests[*index]->generalized_funcs)->extra_state, mystatus); + } + + if (requests[*index] != MPI_REQUEST_NULL && requests[*index]->flags_ & MPI_REQ_NBC){ + *flag = finish_nbc_requests(&requests[*index] , 1); + } + + 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; } else { nsleeps++; } } else { + XBT_DEBUG("Testany on inactive handles, returning flag=1 but empty status"); //all requests are null or inactive, return true - flag = 1; + *flag = 1; + *index = MPI_UNDEFINED; Status::empty(status); } - return flag; + return ret; } -int Request::testall(int count, MPI_Request requests[], MPI_Status status[]) +int Request::testall(int count, MPI_Request requests[], int* outflag, MPI_Status status[]) { MPI_Status stat; MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat; - int flag=1; + int flag; + int error = 0; + *outflag = 1; for(int i=0; iflags_ & MPI_REQ_PREPARED)) { - if (test(&requests[i], pstat)!=1){ + int ret = test(&requests[i], pstat, &flag); + if (flag){ flag=0; }else{ - requests[i]=MPI_REQUEST_NULL; + *outflag=0; } + if (ret != MPI_SUCCESS) + error = 1; }else{ Status::empty(pstat); } @@ -662,7 +845,10 @@ int Request::testall(int count, MPI_Request requests[], MPI_Status status[]) status[i] = *pstat; } } - return flag; + if (error == 1) + return MPI_ERR_IN_STATUS; + else + return MPI_SUCCESS; } void Request::probe(int source, int tag, MPI_Comm comm, MPI_Status* status){ @@ -682,39 +868,44 @@ void Request::iprobe(int source, int tag, MPI_Comm comm, int* flag, MPI_Status* // This can speed up the execution of certain applications by an order of magnitude, such as HPL static int nsleeps = 1; double speed = s4u::this_actor::get_host()->get_speed(); - double maxrate = simgrid::config::get_value("smpi/iprobe-cpu-usage"); - MPI_Request request = new Request(nullptr, 0, MPI_CHAR, - source == MPI_ANY_SOURCE ? MPI_ANY_SOURCE : comm->group()->actor(source)->get_pid(), - simgrid::s4u::this_actor::get_pid(), tag, comm, MPI_REQ_PERSISTENT | MPI_REQ_RECV); + double maxrate = smpi_cfg_iprobe_cpu_usage(); + auto request = + new Request(nullptr, 0, MPI_CHAR, source == MPI_ANY_SOURCE ? MPI_ANY_SOURCE : comm->group()->actor(source), + simgrid::s4u::this_actor::get_pid(), tag, comm, MPI_REQ_PERSISTENT | MPI_REQ_RECV | MPI_REQ_PROBE); if (smpi_iprobe_sleep > 0) { /** Compute the number of flops we will sleep **/ s4u::this_actor::exec_init(/*nsleeps: See comment above */ nsleeps * - /*(in seconds)*/ smpi_iprobe_sleep * speed * maxrate) + /*(seconds * flop/s -> total flops)*/ smpi_iprobe_sleep * speed * maxrate) ->set_name("iprobe") + /* Not the entire CPU can be used when iprobing: This is important for + * the energy consumption caused by polling with iprobes. + * Note also that the number of flops that was + * computed above contains a maxrate factor and is hence reduced (maxrate < 1) + */ + ->set_bound(maxrate * speed) ->start() ->wait(); } // behave like a receive, but don't do it - smx_mailbox_t mailbox; + s4u::Mailbox* mailbox; request->print_request("New iprobe"); - // We have to test both mailboxes as we don't know if we will receive one one or another - if (simgrid::config::get_value("smpi/async-small-thresh") > 0) { + // We have to test both mailboxes as we don't know if we will receive one or another + if (smpi_cfg_async_small_thresh() > 0) { mailbox = smpi_process()->mailbox_small(); XBT_DEBUG("Trying to probe the perm recv mailbox"); - request->action_ = simcall_comm_iprobe(mailbox, 0, &match_recv, static_cast(request)); + request->action_ = mailbox->iprobe(0, &match_recv, static_cast(request)); } if (request->action_ == nullptr){ mailbox = smpi_process()->mailbox(); XBT_DEBUG("trying to probe the other mailbox"); - request->action_ = simcall_comm_iprobe(mailbox, 0, &match_recv, static_cast(request)); + request->action_ = mailbox->iprobe(0, &match_recv, static_cast(request)); } if (request->action_ != nullptr){ - simgrid::kernel::activity::CommImplPtr sync_comm = - boost::static_pointer_cast(request->action_); - MPI_Request req = static_cast(sync_comm->src_data); + kernel::activity::CommImplPtr sync_comm = boost::static_pointer_cast(request->action_); + const Request* req = static_cast(sync_comm->src_data_); *flag = 1; if (status != MPI_STATUS_IGNORE && (req->flags_ & MPI_REQ_PREPARED) == 0) { status->MPI_SOURCE = comm->group()->rank(req->src_); @@ -726,18 +917,51 @@ void Request::iprobe(int source, int tag, MPI_Comm comm, int* flag, MPI_Status* } else { *flag = 0; - if (simgrid::config::get_value("smpi/grow-injected-times")) + if (smpi_cfg_grow_injected_times()) nsleeps++; } unref(&request); + xbt_assert(request == MPI_REQUEST_NULL); +} + +int Request::finish_nbc_requests(MPI_Request* request, int test){ + int flag = 1; + int ret = 0; + if(test == 0) + ret = waitall((*request)->nbc_requests_.size(), (*request)->nbc_requests_.data(), MPI_STATUSES_IGNORE); + else{ + ret = testall((*request)->nbc_requests_.size(), (*request)->nbc_requests_.data(), &flag, MPI_STATUSES_IGNORE); + } + if(ret!=MPI_SUCCESS) + xbt_die("Failure when waiting on non blocking collective sub-requests"); + if(flag == 1){ + XBT_DEBUG("Finishing non blocking collective request with %zu sub-requests", (*request)->nbc_requests_.size()); + for(auto& req: (*request)->nbc_requests_){ + if((*request)->buf_!=nullptr && req!=MPI_REQUEST_NULL){//reduce case + void * buf=req->buf_; + if((*request)->type_->flags() & DT_FLAG_DERIVED) + buf=req->old_buf_; + if(req->flags_ & MPI_REQ_RECV ){ + if((*request)->op_!=MPI_OP_NULL){ + int count=(*request)->size_/ (*request)->type_->size(); + (*request)->op_->apply(buf, (*request)->buf_, &count, (*request)->type_); + } + smpi_free_tmp_buffer(static_cast(buf)); + } + } + if(req!=MPI_REQUEST_NULL) + Request::unref(&req); + } + (*request)->nbc_requests_.clear(); + } + return flag; } void Request::finish_wait(MPI_Request* request, MPI_Status * status) { MPI_Request req = *request; Status::empty(status); - - if (req->cancelled_==1){ + if((req->flags_ & MPI_REQ_CANCELLED) != 0 && (req->flags_ & MPI_REQ_MATCHED) == 0) { if (status!=MPI_STATUS_IGNORE) status->cancelled=1; if(req->detached_sender_ != nullptr) @@ -746,56 +970,61 @@ void Request::finish_wait(MPI_Request* request, MPI_Status * status) return; } - if (not((req->detached_ != 0) && ((req->flags_ & MPI_REQ_SEND) != 0)) && ((req->flags_ & MPI_REQ_PREPARED) == 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); - status->MPI_TAG = req->tag_ == MPI_ANY_TAG ? req->real_tag_ : req->tag_; - status->MPI_ERROR = req->truncated_ != 0 ? MPI_ERR_TRUNCATE : MPI_SUCCESS; + if ((req->flags_ & (MPI_REQ_PREPARED | MPI_REQ_GENERALIZED | MPI_REQ_FINISHED)) == 0) { + if (status != MPI_STATUS_IGNORE) { + if (req->src_== MPI_PROC_NULL || req->dst_== MPI_PROC_NULL){ + Status::empty(status); + status->MPI_SOURCE = MPI_PROC_NULL; + } else { + aid_t src = req->src_ == MPI_ANY_SOURCE ? req->real_src_ : req->src_; + status->MPI_SOURCE = req->comm_->group()->rank(src); + status->MPI_TAG = req->tag_ == MPI_ANY_TAG ? req->real_tag_ : req->tag_; + status->MPI_ERROR = req->truncated_ ? MPI_ERR_TRUNCATE : MPI_SUCCESS; + } // this handles the case were size in receive differs from size in send status->count = req->real_size_; } - - 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); + //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->type_; + + // FIXME Handle the case of a partial shared malloc. + if (not smpi_process()->replaying() && + (((req->flags_ & MPI_REQ_ACCUMULATE) != 0) || (datatype->flags() & DT_FLAG_DERIVED))) { + if (smpi_switch_data_segment(simgrid::s4u::Actor::self(), req->old_buf_)) + XBT_VERB("Privatization : We are unserializing to a zone in global memory Switch data segment "); + + if(datatype->flags() & DT_FLAG_DERIVED){ + // This part handles the problem of non-contiguous 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_); + req->buf_=nullptr; + } 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_); + req->buf_=nullptr; } - xbt_free(req->buf_); } } } if (TRACE_smpi_view_internals() && ((req->flags_ & MPI_REQ_RECV) != 0)) { - int rank = simgrid::s4u::this_actor::get_pid(); - int src_traced = (req->src_ == MPI_ANY_SOURCE ? req->real_src_ : req->src_); + aid_t rank = simgrid::s4u::this_actor::get_pid(); + aid_t src_traced = (req->src_ == MPI_ANY_SOURCE ? req->real_src_ : req->src_); TRACE_smpi_recv(src_traced, rank,req->tag_); } if(req->detached_sender_ != nullptr){ //integrate pseudo-timing for buffering of small messages, do not bother to execute the simcall if 0 - double sleeptime = - simgrid::s4u::Actor::self()->get_host()->extension()->orecv(req->real_size()); - if(sleeptime > 0.0){ - simcall_process_sleep(sleeptime); + simgrid::s4u::Host* dst_host = simgrid::s4u::Actor::by_pid(req->dst_)->get_host(); + if (double sleeptime = simgrid::s4u::Actor::self()->get_host()->extension()->orecv( + req->real_size(), req->src_host_, dst_host); + sleeptime > 0.0) { + simgrid::s4u::this_actor::sleep_for(sleeptime); XBT_DEBUG("receiving size of %zu : sleep %f ", req->real_size_, sleeptime); } unref(&(req->detached_sender_)); @@ -803,58 +1032,116 @@ void Request::finish_wait(MPI_Request* request, MPI_Status * status) if (req->flags_ & MPI_REQ_PERSISTENT) req->action_ = nullptr; req->flags_ |= MPI_REQ_FINISHED; - unref(request); + + if (req->truncated_ || req->unmatched_types_) { + char error_string[MPI_MAX_ERROR_STRING]; + int error_size; + int errkind; + if(req->truncated_ ) + errkind = MPI_ERR_TRUNCATE; + else + errkind = MPI_ERR_TYPE; + PMPI_Error_string(errkind, error_string, &error_size); + MPI_Errhandler err = (req->comm_) ? (req->comm_)->errhandler() : MPI_ERRHANDLER_NULL; + if (err == MPI_ERRHANDLER_NULL || err == MPI_ERRORS_RETURN) + XBT_WARN("recv - returned %.*s instead of MPI_SUCCESS", error_size, error_string); + else if (err == MPI_ERRORS_ARE_FATAL) + xbt_die("recv - returned %.*s instead of MPI_SUCCESS", error_size, error_string); + else + err->call((req->comm_), errkind); + if (err != MPI_ERRHANDLER_NULL) + simgrid::smpi::Errhandler::unref(err); + MC_assert(not MC_is_active()); /* Only fail in MC mode */ + } + if(req->src_ != MPI_PROC_NULL && req->dst_ != MPI_PROC_NULL) + unref(request); } -void Request::wait(MPI_Request * request, MPI_Status * status) +int Request::wait(MPI_Request * request, MPI_Status * status) { + // assume that *request is not MPI_REQUEST_NULL (filtered in PMPI_Wait before) + xbt_assert(*request != MPI_REQUEST_NULL); + + int ret=MPI_SUCCESS; + + if((*request)->src_ == MPI_PROC_NULL || (*request)->dst_ == MPI_PROC_NULL){ + if (status != MPI_STATUS_IGNORE) { + Status::empty(status); + status->MPI_SOURCE = MPI_PROC_NULL; + } + (*request)=MPI_REQUEST_NULL; + return ret; + } + (*request)->print_request("Waiting"); - if ((*request)->flags_ & MPI_REQ_PREPARED) { + if ((*request)->flags_ & (MPI_REQ_PREPARED | MPI_REQ_FINISHED)) { Status::empty(status); - return; + return ret; } if ((*request)->action_ != nullptr){ try{ // this is not a detached send - simcall_comm_wait((*request)->action_, -1.0); - }catch (xbt_ex& e) { + kernel::actor::ActorImpl* issuer = kernel::actor::ActorImpl::self(); + kernel::actor::ActivityWaitSimcall observer{issuer, (*request)->action_.get(), -1}; + kernel::actor::simcall_blocking([issuer, &observer] { observer.get_activity()->wait_for(issuer, -1); }, + &observer); + } catch (const CancelException&) { XBT_VERB("Request cancelled"); } } + if ((*request)->flags_ & MPI_REQ_GENERALIZED) { + if (not((*request)->flags_ & MPI_REQ_COMPLETE)) { + ((*request)->generalized_funcs)->mutex->lock(); + ((*request)->generalized_funcs)->cond->wait(((*request)->generalized_funcs)->mutex); + ((*request)->generalized_funcs)->mutex->unlock(); + } + MPI_Status tmp_status; + MPI_Status* mystatus; + if (status == MPI_STATUS_IGNORE) { + mystatus = &tmp_status; + Status::empty(mystatus); + } else { + mystatus = status; + } + ret = ((*request)->generalized_funcs)->query_fn(((*request)->generalized_funcs)->extra_state, mystatus); + } + + if ((*request)->truncated_) + ret = MPI_ERR_TRUNCATE; + + if ((*request)->flags_ & MPI_REQ_NBC) + finish_nbc_requests(request, 0); - finish_wait(request,status); + finish_wait(request, status); // may invalidate *request if (*request != MPI_REQUEST_NULL && (((*request)->flags_ & MPI_REQ_NON_PERSISTENT) != 0)) *request = MPI_REQUEST_NULL; + return ret; } int Request::waitany(int count, MPI_Request requests[], MPI_Status * status) { - s_xbt_dynar_t comms; // Keep it on stack to save some extra mallocs int index = MPI_UNDEFINED; if(count > 0) { - int size = 0; // Wait for a request to complete - xbt_dynar_init(&comms, sizeof(smx_activity_t), [](void*ptr){ - intrusive_ptr_release(*(simgrid::kernel::activity::ActivityImpl**)ptr); - }); - int *map = xbt_new(int, count); + std::vector comms; + std::vector map; XBT_DEBUG("Wait for one of %d", count); for(int i = 0; i < count; i++) { if (requests[i] != MPI_REQUEST_NULL && not(requests[i]->flags_ & MPI_REQ_PREPARED) && not(requests[i]->flags_ & MPI_REQ_FINISHED)) { if (requests[i]->action_ != nullptr) { XBT_DEBUG("Waiting any %p ", requests[i]); - intrusive_ptr_add_ref(requests[i]->action_.get()); - xbt_dynar_push_as(&comms, simgrid::kernel::activity::ActivityImpl*, requests[i]->action_.get()); - map[size] = i; - size++; + comms.push_back(requests[i]->action_.get()); + map.push_back(i); } else { // This is a finished detached request, let's return this one - size = 0; // so we free the dynar but don't do the waitany call + comms.clear(); // don't do the waitany call afterwards index = i; + if (requests[index]->flags_ & MPI_REQ_NBC) + finish_nbc_requests(&requests[index], 0); finish_wait(&requests[i], status); // cleanup if refcount = 0 if (requests[i] != MPI_REQUEST_NULL && (requests[i]->flags_ & MPI_REQ_NON_PERSISTENT)) requests[i] = MPI_REQUEST_NULL; // set to null @@ -862,15 +1149,21 @@ int Request::waitany(int count, MPI_Request requests[], MPI_Status * status) } } } - if (size > 0) { - XBT_DEBUG("Enter waitany for %lu comms", xbt_dynar_length(&comms)); - int i=MPI_UNDEFINED; + if (not comms.empty()) { + XBT_DEBUG("Enter waitany for %zu comms", comms.size()); + ssize_t i; try{ - // this is not a detached send - i = simcall_comm_waitany(&comms, -1); - }catch (xbt_ex& e) { - XBT_INFO("request %d cancelled ",i); - return i; + kernel::actor::ActorImpl* issuer = kernel::actor::ActorImpl::self(); + kernel::actor::ActivityWaitanySimcall observer{issuer, comms, -1}; + i = kernel::actor::simcall_blocking( + [&observer] { + kernel::activity::ActivityImpl::wait_any_for(observer.get_issuer(), observer.get_activities(), + observer.get_timeout()); + }, + &observer); + } catch (const CancelException&) { + XBT_INFO("request cancelled"); + i = -1; } // not MPI_UNDEFINED, as this is a simix return code @@ -885,18 +1178,16 @@ int Request::waitany(int count, MPI_Request requests[], MPI_Status * status) } } } - - xbt_dynar_free_data(&comms); - xbt_free(map); } + if (index==MPI_UNDEFINED) Status::empty(status); return index; } -static int sort_accumulates(MPI_Request a, MPI_Request b) +static int sort_accumulates(const Request* a, const Request* b) { return (a->tag() > b->tag()); } @@ -925,8 +1216,8 @@ int Request::waitall(int count, MPI_Request requests[], MPI_Status status[]) wait(&requests[c],pstat); index = c; } else { - index = waitany(count, (MPI_Request*)requests, pstat); - + index = waitany(count, requests, pstat); + if (index == MPI_UNDEFINED) break; @@ -943,12 +1234,9 @@ int Request::waitall(int count, MPI_Request requests[], MPI_Status status[]) } } - if (not accumulates.empty()) { - std::sort(accumulates.begin(), accumulates.end(), sort_accumulates); - for (auto& req : accumulates) { - finish_wait(&req, status); - } - } + std::sort(accumulates.begin(), accumulates.end(), sort_accumulates); + for (auto& req : accumulates) + finish_wait(&req, status); return retvalue; } @@ -956,51 +1244,109 @@ int Request::waitall(int count, MPI_Request requests[], MPI_Status status[]) int Request::waitsome(int incount, MPI_Request requests[], int *indices, MPI_Status status[]) { int count = 0; + int flag = 0; + int index = 0; MPI_Status stat; MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat; - + index = waitany(incount, requests, pstat); + if(index==MPI_UNDEFINED) return MPI_UNDEFINED; + if(status != MPI_STATUSES_IGNORE) { + status[count] = *pstat; + } + indices[count] = index; + count++; for (int i = 0; i < incount; i++) { - int index = waitany(incount, requests, pstat); - if(index!=MPI_UNDEFINED){ - indices[count] = index; - count++; - if(status != MPI_STATUSES_IGNORE) { - status[index] = *pstat; + if (i != index && requests[i] != MPI_REQUEST_NULL && not(requests[i]->flags_ & MPI_REQ_FINISHED)) { + test(&requests[i], pstat,&flag); + if (flag==1){ + indices[count] = i; + if(status != MPI_STATUSES_IGNORE) { + status[count] = *pstat; + } + if (requests[i] != MPI_REQUEST_NULL && (requests[i]->flags_ & MPI_REQ_NON_PERSISTENT)) + requests[i]=MPI_REQUEST_NULL; + count++; } - if (requests[index] != MPI_REQUEST_NULL && (requests[index]->flags_ & MPI_REQ_NON_PERSISTENT)) - requests[index] = MPI_REQUEST_NULL; - }else{ - return MPI_UNDEFINED; } } return count; } -MPI_Request Request::f2c(int id) { - char key[KEY_SIZE]; +MPI_Request Request::f2c(int id) +{ if(id==MPI_FORTRAN_REQUEST_NULL) - return static_cast(MPI_REQUEST_NULL); - return static_cast(F2C::f2c_lookup()->at(get_key_id(key, id))); + return MPI_REQUEST_NULL; + return static_cast(F2C::lookup()->at(id)); } -int Request::add_f() +void Request::free_f(int id) { - if (F2C::f2c_lookup() == nullptr) { - F2C::set_f2c_lookup(new std::unordered_map); + if (id != MPI_FORTRAN_REQUEST_NULL) { + F2C::lookup()->erase(id); } - char key[KEY_SIZE]; - (*(F2C::f2c_lookup()))[get_key_id(key, F2C::f2c_id())] = this; - F2C::f2c_id_increment(); - return F2C::f2c_id()-1; } -void Request::free_f(int id) +int Request::get_status(const Request* req, int* flag, MPI_Status* status) { - if (id != MPI_FORTRAN_REQUEST_NULL) { - char key[KEY_SIZE]; - F2C::f2c_lookup()->erase(get_key_id(key, id)); + if(req != MPI_REQUEST_NULL && req->action_ != nullptr) { + req->iprobe(req->comm_->group()->rank(req->src_), req->tag_, req->comm_, flag, status); + if(*flag) + return MPI_SUCCESS; + } + if (req != MPI_REQUEST_NULL && (req->flags_ & MPI_REQ_GENERALIZED) && not(req->flags_ & MPI_REQ_COMPLETE)) { + *flag = 0; + return MPI_SUCCESS; } + + *flag=1; + if(req != MPI_REQUEST_NULL && + status != MPI_STATUS_IGNORE) { + aid_t src = req->src_ == MPI_ANY_SOURCE ? req->real_src_ : req->src_; + status->MPI_SOURCE = req->comm_->group()->rank(src); + status->MPI_TAG = req->tag_ == MPI_ANY_TAG ? req->real_tag_ : req->tag_; + status->MPI_ERROR = req->truncated_ ? MPI_ERR_TRUNCATE : MPI_SUCCESS; + status->count = req->real_size_; + } + return MPI_SUCCESS; +} + +int Request::grequest_start(MPI_Grequest_query_function* query_fn, MPI_Grequest_free_function* free_fn, + MPI_Grequest_cancel_function* cancel_fn, void* extra_state, MPI_Request* request) +{ + *request = new Request(); + (*request)->flags_ |= MPI_REQ_GENERALIZED; + (*request)->flags_ |= MPI_REQ_PERSISTENT; + (*request)->refcount_ = 1; + ((*request)->generalized_funcs) = std::make_unique(); + ((*request)->generalized_funcs)->query_fn=query_fn; + ((*request)->generalized_funcs)->free_fn=free_fn; + ((*request)->generalized_funcs)->cancel_fn=cancel_fn; + ((*request)->generalized_funcs)->extra_state=extra_state; + ((*request)->generalized_funcs)->cond = simgrid::s4u::ConditionVariable::create(); + ((*request)->generalized_funcs)->mutex = simgrid::s4u::Mutex::create(); + return MPI_SUCCESS; } +int Request::grequest_complete(MPI_Request request) +{ + if ((not(request->flags_ & MPI_REQ_GENERALIZED)) || request->generalized_funcs->mutex == nullptr) + return MPI_ERR_REQUEST; + request->generalized_funcs->mutex->lock(); + request->flags_ |= MPI_REQ_COMPLETE; // in case wait would be called after complete + request->generalized_funcs->cond->notify_one(); + request->generalized_funcs->mutex->unlock(); + return MPI_SUCCESS; } + +void Request::start_nbc_requests(std::vector reqs){ + if (not reqs.empty()) { + nbc_requests_ = reqs; + Request::startall(reqs.size(), reqs.data()); + } +} + +std::vector Request::get_nbc_requests() const +{ + return nbc_requests_; } +} // namespace simgrid::smpi