-/* Copyright (c) 2007-2019. The SimGrid Team. All rights reserved. */
+/* Copyright (c) 2007-2020. 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. */
namespace simgrid{
namespace smpi{
-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<void*>(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<void*>(buf))
+ , old_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 = const_cast<void*>(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 unserialization at the reception)
- old_buf_ = old_buf;
- size_ = datatype->size() * count;
datatype->ref();
comm_->ref();
if(op != MPI_REPLACE && op != MPI_OP_NULL)
generalized_funcs=nullptr;
nbc_requests_=nullptr;
nbc_requests_size_=0;
+ init_buffer(count);
}
void Request::ref(){
return false;
}
+void Request::init_buffer(int count){
+ void *old_buf = nullptr;
+// 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 ((((flags_ & MPI_REQ_RECV) != 0) && ((flags_ & MPI_REQ_ACCUMULATE) != 0)) || (old_type_->flags() & DT_FLAG_DERIVED)) {
+ // This part handles the problem of non-contiguous memory
+ old_buf = const_cast<void*>(buf_);
+ if (count==0){
+ buf_ = nullptr;
+ }else {
+ buf_ = xbt_malloc(count*old_type_->size());
+ if ((old_type_->flags() & DT_FLAG_DERIVED) && ((flags_ & MPI_REQ_SEND) != 0)) {
+ old_type_->serialize(old_buf, buf_, count);
+ }
+ }
+ }
+ old_buf_ = old_buf;
+}
+
bool Request::match_recv(void* a, void* b, simgrid::kernel::activity::CommImpl*)
{
MPI_Request ref = static_cast<MPI_Request>(a);
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_/old_type_->size());
+ }
flags_ &= ~MPI_REQ_PREPARED;
flags_ &= ~MPI_REQ_FINISHED;
this->ref();
mailbox = process->mailbox_small();
XBT_DEBUG("Is there a corresponding send already posted in the small mailbox %s (in case of SSEND)?",
mailbox->get_cname());
- smx_activity_t action = mailbox->iprobe(0, &match_recv, static_cast<void*>(this));
+ simgrid::kernel::activity::ActivityImplPtr action = mailbox->iprobe(0, &match_recv, static_cast<void*>(this));
if (action == nullptr) {
mailbox = process->mailbox();
} else {
mailbox = process->mailbox_small();
XBT_DEBUG("Is there a corresponding send already posted the small mailbox?");
- smx_activity_t action = mailbox->iprobe(0, &match_recv, static_cast<void*>(this));
+ simgrid::kernel::activity::ActivityImplPtr action = mailbox->iprobe(0, &match_recv, static_cast<void*>(this));
if (action == nullptr) {
XBT_DEBUG("No, nothing in the permanent receive mailbox");
} else if (((flags_ & MPI_REQ_RMA) != 0) || static_cast<int>(size_) < smpi_cfg_async_small_thresh()) { // eager mode
mailbox = process->mailbox();
XBT_DEBUG("Is there a corresponding recv already posted in the large mailbox %s?", mailbox->get_cname());
- smx_activity_t action = mailbox->iprobe(1, &match_send, static_cast<void*>(this));
+ simgrid::kernel::activity::ActivityImplPtr action = mailbox->iprobe(1, &match_send, static_cast<void*>(this));
if (action == nullptr) {
if ((flags_ & MPI_REQ_SSEND) == 0) {
mailbox = process->mailbox_small();
// 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_;
+ size_t payload_size_ = size_ + 16;//MPI enveloppe size (tag+dest+communicator)
action_ = simcall_comm_isend(
- simgrid::s4u::Actor::by_pid(src_)->get_impl(), mailbox->get_impl(), size_, -1.0, buf, real_size_, &match_send,
+ simgrid::s4u::Actor::by_pid(src_)->get_impl(), mailbox->get_impl(), payload_size_, -1.0, buf, real_size_, &match_send,
&xbt_free_f, // how to free the userdata if a detached send fails
process->replaying() ? &smpi_comm_null_copy_buffer_callback : smpi_comm_copy_data_callback, this,
// detach if msg size < eager/rdv switch limit
Status::empty(status);
*flag = 1;
- if (((*request)->flags_ & MPI_REQ_PREPARED) == 0) {
+ if (((*request)->flags_ & (MPI_REQ_PREPARED | MPI_REQ_FINISHED)) == 0) {
if ((*request)->action_ != nullptr && (*request)->cancelled_ != 1){
try{
- *flag = simcall_comm_test((*request)->action_);
+ *flag = simcall_comm_test((*request)->action_.get());
} catch (const Exception&) {
*flag = 0;
return ret;
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;
}
}
}
}
(*request)->print_request("Waiting");
- if ((*request)->flags_ & MPI_REQ_PREPARED) {
+ if ((*request)->flags_ & (MPI_REQ_PREPARED | MPI_REQ_FINISHED)) {
Status::empty(status);
return ret;
}
if ((*request)->action_ != nullptr){
try{
// this is not a detached send
- simcall_comm_wait((*request)->action_, -1.0);
+ simcall_comm_wait((*request)->action_.get(), -1.0);
} catch (const Exception&) {
XBT_VERB("Request cancelled");
}
}
}
-int Request::get_status(MPI_Request req, int* flag, MPI_Status* status)
+int Request::get_status(const Request* req, int* flag, MPI_Status* status)
{
*flag=0;
