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;
}
static int nsleeps = 1;
int ret = MPI_SUCCESS;
- // are we testing a request meant for non blocking comms ?
+ // Are we testing a request meant for non blocking collectives ?
// If so, test all the subrequests.
if ((*request)->nbc_requests_size_>0){
ret = testall((*request)->nbc_requests_size_, (*request)->nbc_requests_, flag, MPI_STATUSES_IGNORE);
status->MPI_ERROR = req->truncated_ != 0 ? MPI_ERR_TRUNCATE : MPI_SUCCESS;
// this handles the case were size in receive differs from size in send
status->count = req->real_size_;
-// int flag;
-// Request::get_status(req,&flag,status);
}
req->print_request("Finishing");
int Request::wait(MPI_Request * request, MPI_Status * status)
{
int ret=MPI_SUCCESS;
+ // Are we waiting on a request meant for non blocking collectives ?
+ // If so, wait for all the subrequests.
+ 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)->nbc_requests_[i]!=MPI_REQUEST_NULL)
+ Request::unref(&((*request)->nbc_requests_[i]));
+ }
+ delete[] (*request)->nbc_requests_;
+ (*request)->nbc_requests_size_=0;
+ unref(request);
+ return ret;
+ }
+
(*request)->print_request("Waiting");
if ((*request)->flags_ & MPI_REQ_PREPARED) {
Status::empty(status);
if(req != MPI_REQUEST_NULL && req->action_ != nullptr) {
req->iprobe(req->src_, req->tag_, req->comm_, flag, status);
- if(flag)
+ if(*flag)
return MPI_SUCCESS;
}
if (req != MPI_REQUEST_NULL &&
}
*flag=1;
- if(status != MPI_STATUS_IGNORE) {
+ if(req != MPI_REQUEST_NULL &&
+ 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_;
nbc_requests_size_=size;
}
+int Request::get_nbc_requests_size(){
+ return nbc_requests_size_;
+}
+
+MPI_Request* Request::get_nbc_requests(){
+ return nbc_requests_;
+}
+
}
}