-/* Copyright (c) 2007-2017. The SimGrid Team. All rights reserved. */
+/* Copyright (c) 2007-2018. 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. */
XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_rma, smpi, "Logging specific to SMPI (RMA operations)");
+using simgrid::s4u::Actor;
+
namespace simgrid{
namespace smpi{
std::unordered_map<int, smpi_key_elem> Win::keyvals_;
Win::Win(void *base, MPI_Aint size, int disp_unit, MPI_Info info, MPI_Comm comm, int allocated, int dynamic): base_(base), size_(size), disp_unit_(disp_unit), assert_(0), info_(info), comm_(comm), allocated_(allocated), dynamic_(dynamic){
int comm_size = comm->size();
- rank_ = comm->rank();
+ rank_ = comm->rank();
XBT_DEBUG("Creating window");
if(info!=MPI_INFO_NULL)
info->ref();
- name_ = nullptr;
- opened_ = 0;
- group_ = MPI_GROUP_NULL;
- requests_ = new std::vector<MPI_Request>();
- mut_=xbt_mutex_init();
- lock_mut_=xbt_mutex_init();
- atomic_mut_=xbt_mutex_init();
- connected_wins_ = new MPI_Win[comm_size];
+ name_ = nullptr;
+ opened_ = 0;
+ group_ = MPI_GROUP_NULL;
+ requests_ = new std::vector<MPI_Request>();
+ mut_ = xbt_mutex_init();
+ lock_mut_ = xbt_mutex_init();
+ atomic_mut_ = xbt_mutex_init();
+ connected_wins_ = new MPI_Win[comm_size];
connected_wins_[rank_] = this;
- count_ = 0;
+ count_ = 0;
if(rank_==0){
bar_ = MSG_barrier_init(comm_size);
}
mode_=0;
comm->add_rma_win(this);
+ comm->ref();
Colls::allgather(&(connected_wins_[rank_]), sizeof(MPI_Win), MPI_BYTE, connected_wins_, sizeof(MPI_Win),
MPI_BYTE, comm);
comm_->remove_rma_win(this);
Colls::barrier(comm_);
- int rank=comm_->rank();
- if(rank == 0)
+ Comm::unref(comm_);
+
+ if (rank_ == 0)
MSG_barrier_destroy(bar_);
xbt_mutex_destroy(mut_);
xbt_mutex_destroy(lock_mut_);
return MPI_ERR_ARG;
void* recv_addr = static_cast<void*> ( static_cast<char*>(recv_win->base_) + target_disp * recv_win->disp_unit_);
- XBT_DEBUG("Entering MPI_Put to %d", target_rank);
- if(target_rank != comm_->rank()){
- //prepare send_request
- MPI_Request sreq = Request::rma_send_init(origin_addr, origin_count, origin_datatype, smpi_process()->index(),
- comm_->group()->actor(target_rank)->getPid()-1, SMPI_RMA_TAG+1, comm_, MPI_OP_NULL);
+ if (target_rank != comm_->rank()) { // This is not for myself, so we need to send messages
+ XBT_DEBUG("Entering MPI_Put to remote rank %d", target_rank);
+ // prepare send_request
+ MPI_Request sreq =
+ // TODO cheinrich Check for rank / pid conversion
+ Request::rma_send_init(origin_addr, origin_count, origin_datatype, comm_->rank(), target_rank, SMPI_RMA_TAG + 1,
+ comm_, MPI_OP_NULL);
//prepare receiver request
- MPI_Request rreq = Request::rma_recv_init(recv_addr, target_count, target_datatype, smpi_process()->index(),
- comm_->group()->actor(target_rank)->getPid()-1, SMPI_RMA_TAG+1, recv_win->comm_, MPI_OP_NULL);
+ // TODO cheinrich Check for rank / pid conversion
+ MPI_Request rreq = Request::rma_recv_init(recv_addr, target_count, target_datatype, recv_win->comm_->rank(),
+ target_rank, SMPI_RMA_TAG + 1, recv_win->comm_, MPI_OP_NULL);
//start send
sreq->start();
xbt_mutex_release(recv_win->mut_);
}else{
+ XBT_DEBUG("Entering MPI_Put from myself to myself, rank %d", target_rank);
Datatype::copy(origin_addr, origin_count, origin_datatype, recv_addr, target_count, target_datatype);
if(request!=nullptr)
*request = MPI_REQUEST_NULL;
if(target_rank != comm_->rank()){
//prepare send_request
- MPI_Request sreq = Request::rma_send_init(send_addr, target_count, target_datatype,
- comm_->group()->actor(target_rank)->getPid()-1, smpi_process()->index(), SMPI_RMA_TAG+2, send_win->comm_,
- MPI_OP_NULL);
+ MPI_Request sreq = Request::rma_send_init(send_addr, target_count, target_datatype, target_rank,
+ send_win->comm_->rank(), SMPI_RMA_TAG + 2, send_win->comm_, MPI_OP_NULL);
//prepare receiver request
- MPI_Request rreq = Request::rma_recv_init(origin_addr, origin_count, origin_datatype,
- comm_->group()->actor(target_rank)->getPid()-1, smpi_process()->index(), SMPI_RMA_TAG+2, comm_,
- MPI_OP_NULL);
+ MPI_Request rreq = Request::rma_recv_init(
+ origin_addr, origin_count, origin_datatype, target_rank,
+ comm_->rank(), // TODO cheinrich Check here if comm_->rank() and above send_win->comm_->rank() are correct
+ SMPI_RMA_TAG + 2, comm_, MPI_OP_NULL);
//start the send, with another process than us as sender.
sreq->start();
//As the tag will be used for ordering of the operations, substract count from it (to avoid collisions with other SMPI tags, SMPI_RMA_TAG is set below all the other ones we use )
//prepare send_request
- MPI_Request sreq = Request::rma_send_init(origin_addr, origin_count, origin_datatype,
- smpi_process()->index(), comm_->group()->actor(target_rank)->getPid()-1, SMPI_RMA_TAG-3-count_, comm_, op);
+ MPI_Request sreq = Request::rma_send_init(origin_addr, origin_count, origin_datatype, comm_->rank(), target_rank,
+ SMPI_RMA_TAG - 3 - count_, comm_, op);
- //prepare receiver request
- MPI_Request rreq = Request::rma_recv_init(recv_addr, target_count, target_datatype,
- smpi_process()->index(), comm_->group()->actor(target_rank)->getPid()-1, SMPI_RMA_TAG-3-count_, recv_win->comm_, op);
+ // prepare receiver request
+ MPI_Request rreq = Request::rma_recv_init(recv_addr, target_count, target_datatype, recv_win->comm_->rank(),
+ recv_win->comm_->group()->rank(comm_->group()->actor(target_rank)), SMPI_RMA_TAG - 3 - count_, recv_win->comm_, op);
- count_++;
+ count_++;
- //start send
- sreq->start();
- //push request to receiver's win
- xbt_mutex_acquire(recv_win->mut_);
- recv_win->requests_->push_back(rreq);
- rreq->start();
- xbt_mutex_release(recv_win->mut_);
+ // start send
+ sreq->start();
+ // push request to receiver's win
+ xbt_mutex_acquire(recv_win->mut_);
+ recv_win->requests_->push_back(rreq);
+ rreq->start();
+ xbt_mutex_release(recv_win->mut_);
- if(request!=nullptr){
- *request=sreq;
- }else{
- xbt_mutex_acquire(mut_);
- requests_->push_back(sreq);
- xbt_mutex_release(mut_);
- }
+ if (request != nullptr) {
+ *request = sreq;
+ } else {
+ xbt_mutex_acquire(mut_);
+ requests_->push_back(sreq);
+ xbt_mutex_release(mut_);
+ }
XBT_DEBUG("Leaving MPI_Win_Accumulate");
return MPI_SUCCESS;
}
int Win::start(MPI_Group group, int assert){
- /* From MPI forum advices
- The call to MPI_WIN_COMPLETE does not return until the put call has completed at the origin; and the target window
- will be accessed by the put operation only after the call to MPI_WIN_START has matched a call to MPI_WIN_POST by
- the target process. This still leaves much choice to implementors. The call to MPI_WIN_START can block until the
- matching call to MPI_WIN_POST occurs at all target processes. One can also have implementations where the call to
- MPI_WIN_START is nonblocking, but the call to MPI_PUT blocks until the matching call to MPI_WIN_POST occurred; or
- implementations where the first two calls are nonblocking, but the call to MPI_WIN_COMPLETE blocks until the call
- to MPI_WIN_POST occurred; or even implementations where all three calls can complete before any target process
- called MPI_WIN_POST --- the data put must be buffered, in this last case, so as to allow the put to complete at the
- origin ahead of its completion at the target. However, once the call to MPI_WIN_POST is issued, the sequence above
- must complete, without further dependencies. */
+ /* From MPI forum advices
+ The call to MPI_WIN_COMPLETE does not return until the put call has completed at the origin; and the target window
+ will be accessed by the put operation only after the call to MPI_WIN_START has matched a call to MPI_WIN_POST by
+ the target process. This still leaves much choice to implementors. The call to MPI_WIN_START can block until the
+ matching call to MPI_WIN_POST occurs at all target processes. One can also have implementations where the call to
+ MPI_WIN_START is nonblocking, but the call to MPI_PUT blocks until the matching call to MPI_WIN_POST occurred; or
+ implementations where the first two calls are nonblocking, but the call to MPI_WIN_COMPLETE blocks until the call
+ to MPI_WIN_POST occurred; or even implementations where all three calls can complete before any target process
+ called MPI_WIN_POST --- the data put must be buffered, in this last case, so as to allow the put to complete at the
+ origin ahead of its completion at the target. However, once the call to MPI_WIN_POST is issued, the sequence above
+ must complete, without further dependencies. */
//naive, blocking implementation.
- int i = 0;
- int j = 0;
- int size = group->size();
- MPI_Request* reqs = xbt_new0(MPI_Request, size);
+ int i = 0;
+ int j = 0;
+ int size = group->size();
+ MPI_Request* reqs = xbt_new0(MPI_Request, size);
XBT_DEBUG("Entering MPI_Win_Start");
- while (j != size) {
- int src = group->actor(j)->getPid()-1;
- if (src != smpi_process()->index() && src != MPI_UNDEFINED) {
- reqs[i] = Request::irecv_init(nullptr, 0, MPI_CHAR, src, SMPI_RMA_TAG + 4, MPI_COMM_WORLD);
- i++;
- }
- j++;
+ while (j != size) {
+ int src = comm_->group()->rank(group->actor(j));
+ if (src != rank_ && src != MPI_UNDEFINED) { // TODO cheinrich: The check of MPI_UNDEFINED should be useless here
+ reqs[i] = Request::irecv_init(nullptr, 0, MPI_CHAR, src, SMPI_RMA_TAG + 4, comm_);
+ i++;
+ }
+ j++;
}
- size=i;
+ size = i;
Request::startall(size, reqs);
Request::waitall(size, reqs, MPI_STATUSES_IGNORE);
- for(i=0;i<size;i++){
+ for (i = 0; i < size; i++) {
Request::unref(&reqs[i]);
}
xbt_free(reqs);
XBT_DEBUG("Entering MPI_Win_Post");
while(j!=size){
- int dst=group->actor(j)->getPid()-1;
- if(dst!=smpi_process()->index() && dst!=MPI_UNDEFINED){
- reqs[i]=Request::send_init(nullptr, 0, MPI_CHAR, dst, SMPI_RMA_TAG+4, MPI_COMM_WORLD);
+ int dst = comm_->group()->rank(group->actor(j));
+ if (dst != rank_ && dst != MPI_UNDEFINED) {
+ reqs[i] = Request::send_init(nullptr, 0, MPI_CHAR, dst, SMPI_RMA_TAG + 4, comm_);
i++;
}
j++;
XBT_DEBUG("Entering MPI_Win_Complete");
int i = 0;
int j = 0;
- int size = group_->size();
+ int size = group_->size();
MPI_Request* reqs = xbt_new0(MPI_Request, size);
while(j!=size){
- int dst=group_->actor(j)->getPid()-1;
- if(dst!=smpi_process()->index() && dst!=MPI_UNDEFINED){
- reqs[i]=Request::send_init(nullptr, 0, MPI_CHAR, dst, SMPI_RMA_TAG+5, MPI_COMM_WORLD);
+ int dst = comm_->group()->rank(group_->actor(j));
+ if (dst != rank_ && dst != MPI_UNDEFINED) {
+ reqs[i] = Request::send_init(nullptr, 0, MPI_CHAR, dst, SMPI_RMA_TAG + 5, comm_);
i++;
}
j++;
MPI_Request* reqs = xbt_new0(MPI_Request, size);
while(j!=size){
- int src=group_->actor(j)->getPid()-1;
- if(src!=smpi_process()->index() && src!=MPI_UNDEFINED){
- reqs[i]=Request::irecv_init(nullptr, 0, MPI_CHAR, src,SMPI_RMA_TAG+5, MPI_COMM_WORLD);
+ int src = comm_->group()->rank(group_->actor(j));
+ if (src != rank_ && src != MPI_UNDEFINED) {
+ reqs[i] = Request::irecv_init(nullptr, 0, MPI_CHAR, src, SMPI_RMA_TAG + 5, comm_);
i++;
}
j++;
int Win::flush(int rank){
MPI_Win target_win = connected_wins_[rank];
- int finished = finish_comms(rank);
+ int finished = finish_comms(rank_);
XBT_DEBUG("Win_flush on local %d - Finished %d RMA calls", rank_, finished);
- finished = target_win->finish_comms(rank_);
+ finished = target_win->finish_comms(rank);
XBT_DEBUG("Win_flush on remote %d - Finished %d RMA calls", rank, finished);
return MPI_SUCCESS;
}
size = 0;
std::vector<MPI_Request> myreqqs;
std::vector<MPI_Request>::iterator iter = reqqs->begin();
+ int proc_id = comm_->group()->actor(rank)->get_pid();
while (iter != reqqs->end()){
- if(((*iter)!=MPI_REQUEST_NULL) && (((*iter)->src() == rank) || ((*iter)->dst() == rank))){
+ // Let's see if we're either the destination or the sender of this request
+ // because we only wait for requests that we are responsible for.
+ // Also use the process id here since the request itself returns from src()
+ // and dst() the process id, NOT the rank (which only exists in the context of a communicator).
+ if (((*iter) != MPI_REQUEST_NULL) && (((*iter)->src() == proc_id) || ((*iter)->dst() == proc_id))) {
myreqqs.push_back(*iter);
iter = reqqs->erase(iter);
size++;
return size;
}
-
+int Win::shared_query(int rank, MPI_Aint* size, int* disp_unit, void* baseptr)
+{
+ MPI_Win target_win = rank != MPI_PROC_NULL ? connected_wins_[rank] : nullptr;
+ for (int i = 0; not target_win && i < comm_->size(); i++) {
+ if (connected_wins_[i]->size_ > 0)
+ target_win = connected_wins_[i];
+ }
+ if (target_win) {
+ *size = target_win->size_;
+ *disp_unit = target_win->disp_unit_;
+ *static_cast<void**>(baseptr) = target_win->base_;
+ } else {
+ *size = 0;
+ *static_cast<void**>(baseptr) = xbt_malloc(0);
+ }
+ return MPI_SUCCESS;
+}
}
}
