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)
+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)
{
void *old_buf = nullptr;
// FIXME Handle the case of a partial shared malloc.
size_ = datatype->size() * count;
datatype->ref();
comm_->ref();
+ if(op != MPI_REPLACE && op != MPI_OP_NULL)
+ op_->ref();
action_ = nullptr;
detached_ = 0;
detached_sender_ = nullptr;
refcount_ = 1;
else
refcount_ = 0;
- op_ = MPI_REPLACE;
cancelled_ = 0;
generalized_funcs=nullptr;
nbc_requests_=nullptr;
Comm::unref((*request)->comm_);
Datatype::unref((*request)->old_type_);
}
+ if ((*request)->op_!=MPI_REPLACE && (*request)->op_!=MPI_OP_NULL)
+ Op::unref(&(*request)->op_);
+
(*request)->print_request("Destroying");
delete *request;
*request = MPI_REQUEST_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,
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;
}
}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;
+ MPI_REQ_RMA | MPI_REQ_NON_PERSISTENT | MPI_REQ_RECV | MPI_REQ_PREPARED | MPI_REQ_ACCUMULATE, op);
}
return request;
}
/* FIXME: detached sends are not traceable (action_ == nullptr) */
if (action_ != nullptr) {
- std::string category = smpi_process()->get_tracing_category();
- simgrid::simix::simcall([this, category] { this->action_->set_category(category); });
+ boost::static_pointer_cast<kernel::activity::CommImpl>(action_)->set_tracing_category(
+ smpi_process()->get_tracing_category());
}
if (async_small_thresh != 0 || ((flags_ & MPI_REQ_RMA) != 0))
if ((*request)->nbc_requests_size_>0){
ret = waitall((*request)->nbc_requests_size_, (*request)->nbc_requests_, MPI_STATUSES_IGNORE);
for (int i = 0; i < (*request)->nbc_requests_size_; i++) {
+ if((*request)->buf_!=nullptr && (*request)->nbc_requests_[i]!=MPI_REQUEST_NULL){//reduce case
+ void * buf=(*request)->nbc_requests_[i]->buf_;
+ if((*request)->old_type_->flags() & DT_FLAG_DERIVED)
+ buf=(*request)->nbc_requests_[i]->old_buf_;
+ if((*request)->nbc_requests_[i]->flags_ & MPI_REQ_RECV ){
+ if((*request)->op_!=MPI_OP_NULL){
+ int count=(*request)->size_/ (*request)->old_type_->size();
+ (*request)->op_->apply(buf, (*request)->buf_, &count, (*request)->old_type_);
+ }
+ smpi_free_tmp_buffer(buf);
+ }
+ }
if((*request)->nbc_requests_[i]!=MPI_REQUEST_NULL)
Request::unref(&((*request)->nbc_requests_[i]));
}