Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Add new entry in Release_Notes.
[simgrid.git] / src / smpi / mpi / smpi_win.cpp
index 7e8217c..af63fe6 100644 (file)
@@ -1,8 +1,9 @@
-/* Copyright (c) 2007-2021. The SimGrid Team. All rights reserved.          */
+/* Copyright (c) 2007-2023. 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. */
 
+#include <simgrid/modelchecker.h>
 #include "smpi_win.hpp"
 
 #include "private.hpp"
 #include "smpi_keyvals.hpp"
 #include "smpi_request.hpp"
 #include "src/smpi/include/smpi_actor.hpp"
+#include "src/mc/mc_replay.hpp"
 
 #include <algorithm>
+#include <mutex> // std::scoped_lock
 
 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_rma, smpi, "Logging specific to SMPI (RMA operations)");
 
@@ -26,22 +29,18 @@ XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_rma, smpi, "Logging specific to SMPI (RMA o
     return MPI_ERR_RMA_RANGE;\
   }
 
-#define CHECK_WIN_LOCKED(win)\
-  if(opened_==0){ /*check that post/start has been done*/\
-    int locked=0;\
-    for (auto const& it : win->lockers_)\
-      if (it == comm_->rank())\
-        locked = 1;\
-    if(locked != 1)\
-      return MPI_ERR_WIN;\
+#define CHECK_WIN_LOCKED(win)                                                                                          \
+  if (opened_ == 0) { /*check that post/start has been done*/                                                          \
+    bool locked = std::any_of(begin(win->lockers_), end(win->lockers_), [this](int it) { return it == this->rank_; }); \
+    if (not locked)                                                                                                    \
+      return MPI_ERR_WIN;                                                                                              \
   }
 
-namespace simgrid{
-namespace smpi{
+namespace simgrid::smpi {
 std::unordered_map<int, smpi_key_elem> Win::keyvals_;
 int Win::keyval_id_=0;
 
-Win::Win(void* base, MPI_Aint size, int disp_unit, MPI_Info info, MPI_Comm comm, int allocated, int dynamic)
+Win::Win(void* base, MPI_Aint size, int disp_unit, MPI_Info info, MPI_Comm comm, bool allocated, bool dynamic)
     : base_(base)
     , size_(size)
     , disp_unit_(disp_unit)
@@ -56,47 +55,57 @@ Win::Win(void* base, MPI_Aint size, int disp_unit, MPI_Info info, MPI_Comm comm,
   if(info!=MPI_INFO_NULL)
     info->ref();
   connected_wins_[rank_] = this;
-  if(rank_==0){
-    bar_ = new s4u::Barrier(comm->size());
-  }
   errhandler_->ref();
   comm->add_rma_win(this);
   comm->ref();
 
   colls::allgather(&connected_wins_[rank_], sizeof(MPI_Win), MPI_BYTE, connected_wins_.data(), sizeof(MPI_Win),
                    MPI_BYTE, comm);
-
-  colls::bcast(&bar_, sizeof(s4u::Barrier*), MPI_BYTE, 0, comm);
-
-  colls::barrier(comm);
+  if  (MC_is_active() || MC_record_replay_is_active()){
+    s4u::Barrier* bar_ptr;
+    if (rank_ == 0) {
+      bar_ = s4u::Barrier::create(comm->size());
+      bar_ptr = bar_.get();
+    }
+    colls::bcast(&bar_ptr, sizeof(s4u::Barrier*), MPI_BYTE, 0, comm);
+    if (rank_ != 0)
+      bar_ = s4u::BarrierPtr(bar_ptr);
+  }
   this->add_f();
 }
 
-Win::~Win(){
+int Win::del(Win* win){
   //As per the standard, perform a barrier to ensure every async comm is finished
-  bar_->wait();
-
-  int finished = finish_comms();
-  XBT_DEBUG("Win destructor - Finished %d RMA calls", finished);
-
-  if (info_ != MPI_INFO_NULL)
-    simgrid::smpi::Info::unref(info_);
-  if (errhandler_ != MPI_ERRHANDLER_NULL)
-    simgrid::smpi::Errhandler::unref(errhandler_);
-
-  comm_->remove_rma_win(this);
-
-  colls::barrier(comm_);
-  Comm::unref(comm_);
-  
-  if (rank_ == 0)
-    delete bar_;
+  if  (MC_is_active() || MC_record_replay_is_active())
+    win->bar_->wait();
+  else
+    colls::barrier(win->comm_);
+  win->flush_local_all();
+
+  if (win->info_ != MPI_INFO_NULL)
+    simgrid::smpi::Info::unref(win->info_);
+  if (win->errhandler_ != MPI_ERRHANDLER_NULL)
+    simgrid::smpi::Errhandler::unref(win->errhandler_);
+
+  win->comm_->remove_rma_win(win);
+
+  colls::barrier(win->comm_);
+  Comm::unref(win->comm_);
+  if (not win->lockers_.empty() || win->opened_ < 0) {
+    XBT_WARN("Freeing a locked or opened window");
+    return MPI_ERR_WIN;
+  }
+  if (win->allocated_)
+    xbt_free(win->base_);
+  for (auto m : {win->mut_, win->lock_mut_, win->atomic_mut_})
+    if (m->get_owner() != nullptr)
+      m->unlock();
 
-  if(allocated_ !=0)
-    xbt_free(base_);
+  F2C::free_f(win->f2c_id());
+  win->cleanup_attr<Win>();
 
-  F2C::free_f(this->f2c_id());
-  cleanup_attr<Win>();
+  delete win;
+  return MPI_SUCCESS;
 }
 
 int Win::attach(void* /*base*/, MPI_Aint size)
@@ -134,9 +143,6 @@ void Win::get_group(MPI_Group* group){
 
 MPI_Info Win::info()
 {
-  if (info_ == MPI_INFO_NULL)
-    info_ = new Info();
-  info_->ref();
   return info_;
 }
 
@@ -165,7 +171,7 @@ int Win::disp_unit() const
   return disp_unit_;
 }
 
-int Win::dynamic() const
+bool Win::dynamic() const
 {
   return dynamic_;
 }
@@ -186,32 +192,24 @@ void Win::set_name(const char* name){
 int Win::fence(int assert)
 {
   XBT_DEBUG("Entering fence");
-  if (opened_ == 0)
-    opened_=1;
+  opened_++;
   if (not (assert & MPI_MODE_NOPRECEDE)) {
     // This is not the first fence => finalize what came before
-    bar_->wait();
-    mut_->lock();
-    // This (simulated) mutex ensures that no process pushes to the vector of requests during the waitall.
-    // Without this, the vector could get redimensioned when another process pushes.
-    // This would result in the array used by Request::waitall() to be invalidated.
-    // Another solution would be to copy the data and cleanup the vector *before* Request::waitall
-
-    // start all requests that have been prepared by another process
-    if (not requests_.empty()) {
-      int size           = static_cast<int>(requests_.size());
-      MPI_Request* treqs = requests_.data();
-      Request::waitall(size, treqs, MPI_STATUSES_IGNORE);
-    }
+    if (MC_is_active() || MC_record_replay_is_active())
+      bar_->wait();
+    else
+      colls::barrier(comm_);
+    flush_local_all();
     count_=0;
-    mut_->unlock();
   }
 
   if (assert & MPI_MODE_NOSUCCEED) // there should be no ops after this one, tell we are closed.
     opened_=0;
   assert_ = assert;
-
-  bar_->wait();
+  if (MC_is_active() || MC_record_replay_is_active())
+    bar_->wait();
+  else
+    colls::barrier(comm_);
   XBT_DEBUG("Leaving fence");
 
   return MPI_SUCCESS;
@@ -223,32 +221,21 @@ int Win::put(const void *origin_addr, int origin_count, MPI_Datatype origin_data
   //get receiver pointer
   Win* recv_win = connected_wins_[target_rank];
 
-  if(opened_==0){//check that post/start has been done
-    // no fence or start .. lock ok ?
-    int locked=0;
-    for (auto const& it : recv_win->lockers_)
-      if (it == comm_->rank())
-        locked = 1;
-    if(locked != 1)
-      return MPI_ERR_WIN;
-  }
-
+  CHECK_WIN_LOCKED(recv_win)
   CHECK_RMA_REMOTE_WIN("MPI_Put", recv_win)
 
   void* recv_addr = static_cast<char*>(recv_win->base_) + target_disp * recv_win->disp_unit_;
 
-  if (target_rank != comm_->rank()) { // This is not for myself, so we need to send messages
+  if (target_rank != 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);
+        Request::rma_send_init(origin_addr, origin_count, origin_datatype, rank_, target_rank, SMPI_RMA_TAG + 1, comm_,
+                               MPI_OP_NULL);
 
     //prepare receiver request
-    // 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);
+    MPI_Request rreq = Request::rma_recv_init(recv_addr, target_count, target_datatype, rank_, target_rank,
+                                              SMPI_RMA_TAG + 1, recv_win->comm_, MPI_OP_NULL);
 
     //start send
     sreq->start();
@@ -256,16 +243,14 @@ int Win::put(const void *origin_addr, int origin_count, MPI_Datatype origin_data
     if(request!=nullptr){
       *request=sreq;
     }else{
-      mut_->lock();
+      const std::scoped_lock lock(*mut_);
       requests_.push_back(sreq);
-      mut_->unlock();
     }
 
     //push request to receiver's win
-    recv_win->mut_->lock();
+    const std::scoped_lock recv_lock(*recv_win->mut_);
     recv_win->requests_.push_back(rreq);
     rreq->start();
-    recv_win->mut_->unlock();
   } 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);
@@ -288,23 +273,21 @@ int Win::get( void *origin_addr, int origin_count, MPI_Datatype origin_datatype,
   const void* send_addr = static_cast<void*>(static_cast<char*>(send_win->base_) + target_disp * send_win->disp_unit_);
   XBT_DEBUG("Entering MPI_Get from %d", target_rank);
 
-  if(target_rank != comm_->rank()){
+  if (target_rank != rank_) {
     //prepare send_request
-    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);
+    MPI_Request sreq = Request::rma_send_init(send_addr, target_count, target_datatype, target_rank, 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, 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);
+    MPI_Request rreq = Request::rma_recv_init(origin_addr, origin_count, origin_datatype, target_rank, rank_,
+                                              SMPI_RMA_TAG + 2, comm_, MPI_OP_NULL);
 
     //start the send, with another process than us as sender.
     sreq->start();
-    //push request to receiver's win
-    send_win->mut_->lock();
-    send_win->requests_.push_back(sreq);
-    send_win->mut_->unlock();
+    // push request to sender's win
+    if (const std::scoped_lock send_lock(*send_win->mut_); true) {
+      send_win->requests_.push_back(sreq);
+    }
 
     //start recv
     rreq->start();
@@ -312,9 +295,8 @@ int Win::get( void *origin_addr, int origin_count, MPI_Datatype origin_datatype,
     if(request!=nullptr){
       *request=rreq;
     }else{
-      mut_->lock();
+      const std::scoped_lock lock(*mut_);
       requests_.push_back(rreq);
-      mut_->unlock();
     }
   } else {
     Datatype::copy(send_addr, target_count, target_datatype, origin_addr, origin_count, origin_datatype);
@@ -341,12 +323,11 @@ int Win::accumulate(const void *origin_addr, int origin_count, MPI_Datatype orig
   // 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, comm_->rank(), target_rank,
+  MPI_Request sreq = Request::rma_send_init(origin_addr, origin_count, origin_datatype, 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, recv_win->comm_->rank(),
-                                            recv_win->comm_->group()->rank(comm_->group()->actor(target_rank)),
+  MPI_Request rreq = Request::rma_recv_init(recv_addr, target_count, target_datatype, rank_, target_rank,
                                             SMPI_RMA_TAG - 3 - count_, recv_win->comm_, op);
 
   count_++;
@@ -354,19 +335,21 @@ int Win::accumulate(const void *origin_addr, int origin_count, MPI_Datatype orig
   // start send
   sreq->start();
   // push request to receiver's win
-  recv_win->mut_->lock();
-  recv_win->requests_.push_back(rreq);
-  rreq->start();
-  recv_win->mut_->unlock();
+  if (const std::scoped_lock recv_lock(*recv_win->mut_); true) {
+    recv_win->requests_.push_back(rreq);
+    rreq->start();
+  }
 
   if (request != nullptr) {
     *request = sreq;
   } else {
-    mut_->lock();
+    const std::scoped_lock lock(*mut_);
     requests_.push_back(sreq);
-    mut_->unlock();
   }
 
+  // FIXME: The current implementation fails to ensure the correct ordering of the accumulate requests.  The following
+  // 'flush' is a workaround to fix that.
+  flush(target_rank);
   XBT_DEBUG("Leaving MPI_Win_Accumulate");
   return MPI_SUCCESS;
 }
@@ -383,8 +366,8 @@ int Win::get_accumulate(const void* origin_addr, int origin_count, MPI_Datatype
 
   XBT_DEBUG("Entering MPI_Get_accumulate from %d", target_rank);
   //need to be sure ops are correctly ordered, so finish request here ? slow.
-  MPI_Request req;
-  send_win->atomic_mut_->lock();
+  MPI_Request req = MPI_REQUEST_NULL;
+  const std::scoped_lock lock(*send_win->atomic_mut_);
   get(result_addr, result_count, result_datatype, target_rank,
               target_disp, target_count, target_datatype, &req);
   if (req != MPI_REQUEST_NULL)
@@ -394,7 +377,6 @@ int Win::get_accumulate(const void* origin_addr, int origin_count, MPI_Datatype
               target_disp, target_count, target_datatype, op, &req);
   if (req != MPI_REQUEST_NULL)
     Request::wait(&req, MPI_STATUS_IGNORE);
-  send_win->atomic_mut_->unlock();
   return MPI_SUCCESS;
 }
 
@@ -408,7 +390,7 @@ int Win::compare_and_swap(const void* origin_addr, const void* compare_addr, voi
 
   XBT_DEBUG("Entering MPI_Compare_and_swap with %d", target_rank);
   MPI_Request req = MPI_REQUEST_NULL;
-  send_win->atomic_mut_->lock();
+  const std::scoped_lock lock(*send_win->atomic_mut_);
   get(result_addr, 1, datatype, target_rank,
               target_disp, 1, datatype, &req);
   if (req != MPI_REQUEST_NULL)
@@ -417,7 +399,6 @@ int Win::compare_and_swap(const void* origin_addr, const void* compare_addr, voi
     put(origin_addr, 1, datatype, target_rank,
               target_disp, 1, datatype);
   }
-  send_win->atomic_mut_->unlock();
   return MPI_SUCCESS;
 }
 
@@ -436,29 +417,24 @@ int Win::start(MPI_Group group, int /*assert*/)
   must complete, without further dependencies.  */
 
   //naive, blocking implementation.
-  int i             = 0;
-  int j             = 0;
-  int size          = group->size();
-  std::vector<MPI_Request> reqs(size);
-
   XBT_DEBUG("Entering MPI_Win_Start");
-  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++;
+  std::vector<MPI_Request> reqs;
+  for (int i = 0; i < group->size(); i++) {
+    int src = comm_->group()->rank(group->actor(i));
+    xbt_assert(src != MPI_UNDEFINED);
+    if (src != rank_)
+      reqs.emplace_back(Request::irecv_init(nullptr, 0, MPI_CHAR, src, SMPI_RMA_TAG + 4, comm_));
   }
-  size = i;
+  int size = static_cast<int>(reqs.size());
+
   Request::startall(size, reqs.data());
   Request::waitall(size, reqs.data(), MPI_STATUSES_IGNORE);
-  for (i = 0; i < size; i++) {
-    Request::unref(&reqs[i]);
-  }
+  for (auto& req : reqs)
+    Request::unref(&req);
+
   group->ref();
   dst_group_ = group;
-  opened_++; // we're open for business !
+  opened_--; // we're open for business !
   XBT_DEBUG("Leaving MPI_Win_Start");
   return MPI_SUCCESS;
 }
@@ -466,30 +442,24 @@ int Win::start(MPI_Group group, int /*assert*/)
 int Win::post(MPI_Group group, int /*assert*/)
 {
   //let's make a synchronous send here
-  int i             = 0;
-  int j             = 0;
-  int size = group->size();
-  std::vector<MPI_Request> reqs(size);
-
   XBT_DEBUG("Entering MPI_Win_Post");
-  while(j!=size){
-    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++;
+  std::vector<MPI_Request> reqs;
+  for (int i = 0; i < group->size(); i++) {
+    int dst = comm_->group()->rank(group->actor(i));
+    xbt_assert(dst != MPI_UNDEFINED);
+    if (dst != rank_)
+      reqs.emplace_back(Request::send_init(nullptr, 0, MPI_CHAR, dst, SMPI_RMA_TAG + 4, comm_));
   }
-  size=i;
+  int size = static_cast<int>(reqs.size());
 
   Request::startall(size, reqs.data());
   Request::waitall(size, reqs.data(), MPI_STATUSES_IGNORE);
-  for(i=0;i<size;i++){
-    Request::unref(&reqs[i]);
-  }
+  for (auto& req : reqs)
+    Request::unref(&req);
+
   group->ref();
   src_group_ = group;
-  opened_++; // we're open for business !
+  opened_--; // we're open for business !
   XBT_DEBUG("Leaving MPI_Win_Post");
   return MPI_SUCCESS;
 }
@@ -498,32 +468,24 @@ int Win::complete(){
   xbt_assert(opened_ != 0, "Complete called on already opened MPI_Win");
 
   XBT_DEBUG("Entering MPI_Win_Complete");
-  int i             = 0;
-  int j             = 0;
-  int size          = dst_group_->size();
-  std::vector<MPI_Request> reqs(size);
-
-  while(j!=size){
-    int dst = comm_->group()->rank(dst_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++;
+  std::vector<MPI_Request> reqs;
+  for (int i = 0; i < dst_group_->size(); i++) {
+    int dst = comm_->group()->rank(dst_group_->actor(i));
+    xbt_assert(dst != MPI_UNDEFINED);
+    if (dst != rank_)
+      reqs.emplace_back(Request::send_init(nullptr, 0, MPI_CHAR, dst, SMPI_RMA_TAG + 5, comm_));
   }
-  size=i;
+  int size = static_cast<int>(reqs.size());
+
   XBT_DEBUG("Win_complete - Sending sync messages to %d processes", size);
   Request::startall(size, reqs.data());
   Request::waitall(size, reqs.data(), MPI_STATUSES_IGNORE);
+  for (auto& req : reqs)
+    Request::unref(&req);
 
-  for(i=0;i<size;i++){
-    Request::unref(&reqs[i]);
-  }
+  flush_local_all();
 
-  int finished = finish_comms();
-  XBT_DEBUG("Win_complete - Finished %d RMA calls", finished);
-
-  opened_--; //we're closed for business !
+  opened_++; //we're closed for business !
   Group::unref(dst_group_);
   dst_group_ = MPI_GROUP_NULL;
   return MPI_SUCCESS;
@@ -532,30 +494,24 @@ int Win::complete(){
 int Win::wait(){
   //naive, blocking implementation.
   XBT_DEBUG("Entering MPI_Win_Wait");
-  int i             = 0;
-  int j             = 0;
-  int size          = src_group_->size();
-  std::vector<MPI_Request> reqs(size);
-
-  while(j!=size){
-    int src = comm_->group()->rank(src_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++;
+  std::vector<MPI_Request> reqs;
+  for (int i = 0; i < src_group_->size(); i++) {
+    int src = comm_->group()->rank(src_group_->actor(i));
+    xbt_assert(src != MPI_UNDEFINED);
+    if (src != rank_)
+      reqs.emplace_back(Request::irecv_init(nullptr, 0, MPI_CHAR, src, SMPI_RMA_TAG + 5, comm_));
   }
-  size=i;
+  int size = static_cast<int>(reqs.size());
+
   XBT_DEBUG("Win_wait - Receiving sync messages from %d processes", size);
   Request::startall(size, reqs.data());
   Request::waitall(size, reqs.data(), MPI_STATUSES_IGNORE);
-  for(i=0;i<size;i++){
-    Request::unref(&reqs[i]);
-  }
-  int finished = finish_comms();
-  XBT_DEBUG("Win_wait - Finished %d RMA calls", finished);
+  for (auto& req : reqs)
+    Request::unref(&req);
+
+  flush_local_all();
 
-  opened_--; //we're closed for business !
+  opened_++; //we're closed for business !
   Group::unref(src_group_);
   src_group_ = MPI_GROUP_NULL;
   return MPI_SUCCESS;
@@ -574,12 +530,9 @@ int Win::lock(int lock_type, int rank, int /*assert*/)
   } else if (not(target_win->mode_ == MPI_LOCK_SHARED && lock_type == MPI_LOCK_EXCLUSIVE))
     target_win->mode_ += lock_type; // don't set to exclusive if it's already shared
 
-  target_win->lockers_.push_back(comm_->rank());
+  target_win->lockers_.push_back(rank_);
 
-  int finished = finish_comms(rank);
-  XBT_DEBUG("Win_lock %d - Finished %d RMA calls", rank, finished);
-  finished = target_win->finish_comms(rank_);
-  XBT_DEBUG("Win_lock target %d - Finished %d RMA calls", rank, finished);
+  flush(rank);
   return MPI_SUCCESS;
 }
 
@@ -597,15 +550,12 @@ int Win::unlock(int rank){
   MPI_Win target_win = connected_wins_[rank];
   int target_mode = target_win->mode_;
   target_win->mode_= 0;
-  target_win->lockers_.remove(comm_->rank());
+  target_win->lockers_.remove(rank_);
   if (target_mode==MPI_LOCK_EXCLUSIVE){
     target_win->lock_mut_->unlock();
   }
 
-  int finished = finish_comms(rank);
-  XBT_DEBUG("Win_unlock %d - Finished %d RMA calls", rank, finished);
-  finished = target_win->finish_comms(rank_);
-  XBT_DEBUG("Win_unlock target %d - Finished %d RMA calls", rank, finished);
+  flush(rank);
   return MPI_SUCCESS;
 }
 
@@ -620,33 +570,36 @@ int Win::unlock_all(){
 }
 
 int Win::flush(int rank){
-  MPI_Win target_win = connected_wins_[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);
-  XBT_DEBUG("Win_flush on remote %d - Finished %d RMA calls", rank, finished);
+  int finished = finish_comms(rank);
+  XBT_DEBUG("Win_flush on local %d for remote %d - Finished %d RMA calls", rank_, rank, finished);
+  if (rank != rank_) {
+    finished = connected_wins_[rank]->finish_comms(rank_);
+    XBT_DEBUG("Win_flush on remote %d for local %d - Finished %d RMA calls", rank, rank_, finished);
+  }
   return MPI_SUCCESS;
 }
 
 int Win::flush_local(int rank){
   int finished = finish_comms(rank);
-  XBT_DEBUG("Win_flush_local for rank %d - Finished %d RMA calls", rank, finished);
+  XBT_DEBUG("Win_flush_local on local %d for remote %d - Finished %d RMA calls", rank_, rank, finished);
   return MPI_SUCCESS;
 }
 
 int Win::flush_all(){
   int finished = finish_comms();
-  XBT_DEBUG("Win_flush_all on local - Finished %d RMA calls", finished);
+  XBT_DEBUG("Win_flush_all on local %d - Finished %d RMA calls", rank_, finished);
   for (int i = 0; i < comm_->size(); i++) {
-    finished = connected_wins_[i]->finish_comms(rank_);
-    XBT_DEBUG("Win_flush_all on %d - Finished %d RMA calls", i, finished);
+    if (i != rank_) {
+      finished = connected_wins_[i]->finish_comms(rank_);
+      XBT_DEBUG("Win_flush_all on remote %d for local %d - Finished %d RMA calls", i, rank_, finished);
+    }
   }
   return MPI_SUCCESS;
 }
 
 int Win::flush_local_all(){
   int finished = finish_comms();
-  XBT_DEBUG("Win_flush_local_all - Finished %d RMA calls", finished);
+  XBT_DEBUG("Win_flush_local_all on local %d - Finished %d RMA calls", rank_, finished);
   return MPI_SUCCESS;
 }
 
@@ -655,7 +608,11 @@ Win* Win::f2c(int id){
 }
 
 int Win::finish_comms(){
-  mut_->lock();
+  // This (simulated) mutex ensures that no process pushes to the vector of requests during the waitall.
+  // Without this, the vector could get redimensioned when another process pushes.
+  // This would result in the array used by Request::waitall() to be invalidated.
+  // Another solution would be to copy the data and cleanup the vector *before* Request::waitall
+  const std::scoped_lock lock(*mut_);
   //Finish own requests
   int size = static_cast<int>(requests_.size());
   if (size > 0) {
@@ -663,12 +620,12 @@ int Win::finish_comms(){
     Request::waitall(size, treqs, MPI_STATUSES_IGNORE);
     requests_.clear();
   }
-  mut_->unlock();
   return size;
 }
 
 int Win::finish_comms(int rank){
-  mut_->lock();
+  // See comment about the mutex in finish_comms() above
+  const std::scoped_lock lock(*mut_);
   // Finish own requests
   // 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.
@@ -686,7 +643,6 @@ int Win::finish_comms(int rank){
     Request::waitall(size, treqs, MPI_STATUSES_IGNORE);
     myreqqs.clear();
   }
-  mut_->unlock();
   return size;
 }
 
@@ -723,5 +679,4 @@ void Win::set_errhandler(MPI_Errhandler errhandler)
   if (errhandler_ != MPI_ERRHANDLER_NULL)
     errhandler_->ref();
 }
-} // namespace smpi
-} // namespace simgrid
+} // namespace simgrid::smpi