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 = 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);
return match_common(req, ref, req);
}
-void Request::print_request(const char *message)
+void Request::print_request(const char* message) const
{
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_);
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();
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_));
+ const simgrid::smpi::ActorExt* process = smpi_process_remote(simgrid::s4u::Actor::by_pid(dst_));
xbt_assert(process, "Actor pid=%d is gone??", dst_);
int rank = src_;
if (TRACE_smpi_view_internals()) {
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_.get());
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;
}
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());
}
wait(&requests[c],pstat);
index = c;
} else {
- index = waitany(count, (MPI_Request*)requests, pstat);
-
+ index = waitany(count, requests, pstat);
+
if (index == MPI_UNDEFINED)
break;
int index = 0;
MPI_Status stat;
MPI_Status *pstat = status == MPI_STATUSES_IGNORE ? MPI_STATUS_IGNORE : &stat;
- index = waitany(incount, (MPI_Request*)requests, pstat);
+ index = waitany(incount, requests, pstat);
if(index==MPI_UNDEFINED) return MPI_UNDEFINED;
if(status != MPI_STATUSES_IGNORE) {
status[count] = *pstat;
{
char key[KEY_SIZE];
if(id==MPI_FORTRAN_REQUEST_NULL)
- return static_cast<MPI_Request>(MPI_REQUEST_NULL);
+ return MPI_REQUEST_NULL;
return static_cast<MPI_Request>(F2C::f2c_lookup()->at(get_key(key,id)));
}
}
}
-int Request::get_nbc_requests_size(){
+int Request::get_nbc_requests_size() const
+{
return nbc_requests_size_;
}
-MPI_Request* Request::get_nbc_requests(){
+MPI_Request* Request::get_nbc_requests() const
+{
return nbc_requests_;
}
-
}
}