From 62af2aab4a388788bcfc80836eb6d91546df42cd Mon Sep 17 00:00:00 2001 From: Scott Wittenburg Date: Wed, 18 Mar 2026 14:34:01 -0600 Subject: [PATCH 1/3] adiosComm: Expose Test method on Comm::Req wrapper The MPI version of ISend already returned a Comm::Req object that wrapped however many actual requests resulted from the send (there could be multiple if it was batched). This commit exposes MPI_Test as Test, which returns true if MPI_Test returns true for all the encapsulated request objects. --- source/adios2/helper/adiosComm.cpp | 14 +++++++ source/adios2/helper/adiosComm.h | 6 +++ source/adios2/helper/adiosCommDummy.cpp | 3 ++ source/adios2/helper/adiosCommMPI.cpp | 51 +++++++++++++++++++++++++ 4 files changed, 74 insertions(+) diff --git a/source/adios2/helper/adiosComm.cpp b/source/adios2/helper/adiosComm.cpp index c107af2262..c374953068 100644 --- a/source/adios2/helper/adiosComm.cpp +++ b/source/adios2/helper/adiosComm.cpp @@ -180,6 +180,20 @@ Comm::Status Comm::Req::Wait(const std::string &hint) return status; } +bool Comm::Req::TestComplete(const std::string &hint) +{ + bool result = false; + if (m_Impl) + { + result = m_Impl->TestComplete(hint); + if (result) + { + m_Impl.reset(); + } + } + return result; +} + Comm::Win::Win() = default; Comm::Win::Win(std::unique_ptr impl) : m_Impl(std::move(impl)) {} diff --git a/source/adios2/helper/adiosComm.h b/source/adios2/helper/adiosComm.h index d86cfe948f..226be63faf 100644 --- a/source/adios2/helper/adiosComm.h +++ b/source/adios2/helper/adiosComm.h @@ -331,6 +331,11 @@ class Comm::Req */ Comm::Status Wait(const std::string &hint = std::string()); + /** + * @brief Test for completion of a request. + */ + bool TestComplete(const std::string &hint = std::string()); + private: friend class CommImpl; @@ -523,6 +528,7 @@ class CommReqImpl public: virtual ~CommReqImpl() = 0; virtual Comm::Status Wait(const std::string &hint) = 0; + virtual bool TestComplete(const std::string &hint) = 0; }; class CommWinImpl diff --git a/source/adios2/helper/adiosCommDummy.cpp b/source/adios2/helper/adiosCommDummy.cpp index 527cd43fb3..8cbfa90dbe 100644 --- a/source/adios2/helper/adiosCommDummy.cpp +++ b/source/adios2/helper/adiosCommDummy.cpp @@ -35,6 +35,7 @@ class CommReqImplDummy : public CommReqImpl ~CommReqImplDummy() override; Comm::Status Wait(const std::string &hint) override; + bool TestComplete(const std::string &hint) override; }; CommReqImplDummy::~CommReqImplDummy() = default; @@ -334,6 +335,8 @@ Comm::Status CommReqImplDummy::Wait(const std::string &hint) return status; } +bool CommReqImplDummy::TestComplete(const std::string &hint) { return true; } + int CommWinImplDummy::Free(const std::string &hint) { return 0; } Comm CommDummy() diff --git a/source/adios2/helper/adiosCommMPI.cpp b/source/adios2/helper/adiosCommMPI.cpp index 3f558ddffe..c630740ede 100644 --- a/source/adios2/helper/adiosCommMPI.cpp +++ b/source/adios2/helper/adiosCommMPI.cpp @@ -112,6 +112,7 @@ class CommReqImplMPI : public CommReqImpl ~CommReqImplMPI() override; Comm::Status Wait(const std::string &hint) override; + bool TestComplete(const std::string &hint) override; /** Encapsulated MPI datatype of the requested operation. */ MPI_Datatype m_MPIDatatype = MPI_DATATYPE_NULL; @@ -658,6 +659,56 @@ Comm::Status CommReqImplMPI::Wait(const std::string &hint) return status; } +bool CommReqImplMPI::TestComplete(const std::string &hint) +{ + if (m_MPIReqs.empty()) + { + return true; + } + + int flag = 0; + int mpiReturn = MPI_SUCCESS; + std::vector mpiStatuses(m_MPIReqs.size()); + + if (m_MPIReqs.size() > 1) + { + mpiReturn = MPI_Testall(static_cast(m_MPIReqs.size()), m_MPIReqs.data(), &flag, + mpiStatuses.data()); + + // Only check for errors in statuses if the operation actually completed + if (flag != 0 && mpiReturn == MPI_ERR_IN_STATUS) + { + for (auto &mpiStatus : mpiStatuses) + { + if (mpiStatus.MPI_ERROR != MPI_SUCCESS) + { + mpiReturn = mpiStatus.MPI_ERROR; + break; + } + } + } + } + else + { + mpiReturn = MPI_Test(m_MPIReqs.data(), &flag, mpiStatuses.data()); + } + + // non-zero means the operations are complete + if (flag != 0) + { + // Throw or handle any errors we collected + CheckMPIReturn(mpiReturn, hint); + + // Since the requests are done, clear the internal container so the object + // registers as complete/empty for future calls. + m_MPIReqs.clear(); + return true; + } + + // Request(s) not complete, the caller can try again later + return false; +} + int CommWinImplMPI::Free(const std::string &hint) { return MPI_Win_free(&m_Win); } Comm CommWithMPI(MPI_Comm mpiComm) From c618fbbec569fcd107553152595c73728c0e5b90 Mon Sep 17 00:00:00 2001 From: Scott Wittenburg Date: Wed, 18 Mar 2026 14:40:10 -0600 Subject: [PATCH 2/3] adiosRerouting: Non-blocking send returns the request --- source/adios2/helper/adiosRerouting.cpp | 5 +++-- source/adios2/helper/adiosRerouting.h | 3 ++- testing/adios2/helper/TestRerouteMessage.cpp | 5 ++++- 3 files changed, 9 insertions(+), 4 deletions(-) diff --git a/source/adios2/helper/adiosRerouting.cpp b/source/adios2/helper/adiosRerouting.cpp index 678400fdbf..aee42b56db 100644 --- a/source/adios2/helper/adiosRerouting.cpp +++ b/source/adios2/helper/adiosRerouting.cpp @@ -23,7 +23,8 @@ namespace adios2 namespace helper { -void RerouteMessage::NonBlockingSendTo(helper::Comm &comm, int destRank, std::vector &buffer) +helper::Comm::Req RerouteMessage::NonBlockingSendTo(helper::Comm &comm, int destRank, + std::vector &buffer) { size_t pos = 0; buffer.resize(REROUTE_MESSAGE_SIZE); @@ -45,7 +46,7 @@ void RerouteMessage::NonBlockingSendTo(helper::Comm &comm, int destRank, std::ve // std::cout << ss.str(); - comm.Isend(buffer.data(), buffer.size(), destRank, 0); + return comm.Isend(buffer.data(), buffer.size(), destRank, 0); } void RerouteMessage::BlockingSendTo(helper::Comm &comm, int destRank, std::vector &buffer) diff --git a/source/adios2/helper/adiosRerouting.h b/source/adios2/helper/adiosRerouting.h index e458770a5a..50939c5da1 100644 --- a/source/adios2/helper/adiosRerouting.h +++ b/source/adios2/helper/adiosRerouting.h @@ -63,7 +63,8 @@ class RerouteMessage } // Send the contents of this message to another rank - void NonBlockingSendTo(helper::Comm &comm, int destRank, std::vector &buffer); + helper::Comm::Req NonBlockingSendTo(helper::Comm &comm, int destRank, + std::vector &buffer); void BlockingSendTo(helper::Comm &comm, int destRank, std::vector &buffer); // Receive a message from another rank to populate this message diff --git a/testing/adios2/helper/TestRerouteMessage.cpp b/testing/adios2/helper/TestRerouteMessage.cpp index 2650bd8b4b..312cd02165 100644 --- a/testing/adios2/helper/TestRerouteMessage.cpp +++ b/testing/adios2/helper/TestRerouteMessage.cpp @@ -35,7 +35,7 @@ void SendAndReceiveMessage(helper::Comm &comm, int destRank, int srcRank) origMsg.m_DestRank = destRank; origMsg.m_Offset = 2138; origMsg.m_Size = 1213; - origMsg.NonBlockingSendTo(comm, destRank, sendBuffer); + adios2::helper::Comm::Req sendReq = origMsg.NonBlockingSendTo(comm, destRank, sendBuffer); int ready = 0; helper::Comm::Status status; @@ -65,6 +65,9 @@ void SendAndReceiveMessage(helper::Comm &comm, int destRank, int srcRank) ASSERT_EQ(receivedMsg.m_DestRank, worldRank); ASSERT_EQ(receivedMsg.m_Offset, origMsg.m_Offset); ASSERT_EQ(receivedMsg.m_Size, origMsg.m_Size); + + // Make sure the request is complete + ASSERT_EQ(sendReq.TestComplete(), true); } } From 140091bb27d3842df1719bb005c561fe7e04d987 Mon Sep 17 00:00:00 2001 From: Scott Wittenburg Date: Wed, 18 Mar 2026 15:18:28 -0600 Subject: [PATCH 3/3] BP5Writer: Improvements to rerouting aggregation scheme Previously the buffer pool for non-blocking sends was a fixed size (set at 100), and had it no guards against re-using (and thus, obliterating) a buffer while a previous ISend using the buffer may still have been incomplete. This commit introduces a role-appropriate pool size, and also adds logic that forces the next ISend to wait if all buffers in the pool are actively being used by a prior ISend. To reduce the likelihood of the background messaging thread spinning and consuming CPU while nothing is happening, this commit introduces a conditional yield within the background thread loop. If no work is done during an iteration of the loop (message receipt and response, rerouting pair identified and messaged, etc...) then yield() is called at the end of the loop. MPI send-to-self can sometimes be slow, and it was used by each subcoordinator to signal itself to write at the beginning of the workflow. This commit replaces the self-send with direct signalling of the main thread to begin writing. Previously, once the loop finished, the thread exited immediately, possibly causing buffers associated with incomplete ISends to be destroyed. Now that we're collecting the request object from each ISend (stored in the buffer pool), this commit iterates the pool one final time to wait for any incomplete ISend to complete. --- .../engine/bp5/BP5Writer_WithRerouting.cpp | 218 ++++++++++++------ 1 file changed, 149 insertions(+), 69 deletions(-) diff --git a/source/adios2/engine/bp5/BP5Writer_WithRerouting.cpp b/source/adios2/engine/bp5/BP5Writer_WithRerouting.cpp index e56606baa3..91825d70c1 100644 --- a/source/adios2/engine/bp5/BP5Writer_WithRerouting.cpp +++ b/source/adios2/engine/bp5/BP5Writer_WithRerouting.cpp @@ -22,32 +22,68 @@ namespace { +struct ManagedBuffer +{ + std::vector m_Data; + adios2::helper::Comm::Req m_Request; + bool m_Active = false; +}; class BufferPool { public: - BufferPool(int size) { m_Pool.resize(size); } + BufferPool(size_t size) { m_Pool.resize(size); } ~BufferPool() = default; - std::vector &GetNextBuffer() + ManagedBuffer &GetNextBuffer() { - size_t bufferIdx = m_CurrentBufferIdx; + size_t startIdx = m_CurrentBufferIdx; - if (m_CurrentBufferIdx < m_Pool.size() - 1) + while (true) { - m_CurrentBufferIdx += 1; + ManagedBuffer &mb = m_Pool[m_CurrentBufferIdx]; + + // Advance index for next call + m_CurrentBufferIdx = (m_CurrentBufferIdx + 1) % m_Pool.size(); + + // If buffer was never used, it's safe + if (!mb.m_Active) + { + return mb; + } + + // Check if the previous MPI_Isend using this buffer is complete + if (mb.m_Request.TestComplete()) + { + mb.m_Active = false; + return mb; + } + + // If we've circled the whole pool and none have completed, we MUST wait + if (m_CurrentBufferIdx == startIdx) + { + mb.m_Request.Wait(); + mb.m_Active = false; + return mb; + } } - else + } + + void Flush() + { + for (auto &mb : m_Pool) { - m_CurrentBufferIdx = 0; + if (mb.m_Active) + { + mb.m_Request.Wait(); + mb.m_Active = false; + } } - - return m_Pool[bufferIdx]; } size_t m_CurrentBufferIdx = 0; - std::vector> m_Pool; + std::vector m_Pool; }; struct WriterGroupState @@ -186,30 +222,7 @@ void BP5Writer::ReroutingCommunicationLoop() m_Profiler.AddTimerWatch("Rerouting_Send_NB"); m_Profiler.AddTimerWatch("Rerouting_Send_B"); m_Profiler.AddTimerWatch("Rerouting_Recv_NB"); - - auto lf_SendNonBlocking = [](RerouteMessage &msg, adios2::helper::Comm &comm, int toRank, - std::vector &buffer, - adios2::profiling::JSONProfiler &profiler) { - profiler.Start("Rerouting_Send_NB"); - msg.NonBlockingSendTo(comm, toRank, buffer); - profiler.Stop("Rerouting_Send_NB"); - }; - - auto lf_SendBlocking = [](RerouteMessage &msg, adios2::helper::Comm &comm, int toRank, - std::vector &buffer, - adios2::profiling::JSONProfiler &profiler) { - profiler.Start("Rerouting_Send_B"); - msg.BlockingSendTo(comm, toRank, buffer); - profiler.Stop("Rerouting_Send_B"); - }; - - auto lf_RecvBlocking = [](RerouteMessage &msg, adios2::helper::Comm &comm, int fromRank, - std::vector &buffer, - adios2::profiling::JSONProfiler &profiler) { - profiler.Start("Rerouting_Recv_NB"); - msg.BlockingRecvFrom(comm, fromRank, buffer); - profiler.Stop("Rerouting_Recv_NB"); - }; + m_Profiler.AddTimerWatch("Rerouting_GetNextBuffer"); int subCoord = m_Aggregator->m_AggregatorRank; bool iAmSubCoord = m_RankMPI == subCoord; @@ -234,8 +247,21 @@ void BP5Writer::ReroutingCommunicationLoop() // Most sends are currently non-blocking. We use the pool to avoid a // situation where the buffer is destructed before the send is complete. - BufferPool sendBuffers(100); - std::vector recvBuffer; + size_t poolSize = 1; + if (iAmGlobalCoord) + { + // twice the number of subcoordinators + poolSize = 2 * m_Partitioning.m_Partitions.size(); + } + else if (iAmSubCoord) + { + // twice the number of ranks in my writer chain/group + poolSize = 2 * m_Partitioning.m_Partitions[m_Aggregator->m_SubStreamIndex].size(); + } + BufferPool sendBuffers(poolSize); + + std::vector recvBuffer; // For blocking recvs + std::vector sendBuffer; // For blocking sends int writingRank = -1; uint64_t currentFilePos = 0; uint64_t writeMoreCount = 0; @@ -247,6 +273,32 @@ void BP5Writer::ReroutingCommunicationLoop() bool expectingWriteCompletion = false; bool sentIdle = false; + auto lf_SendNonBlocking = [&sendBuffers](RerouteMessage &msg, adios2::helper::Comm &comm, + int toRank, + adios2::profiling::JSONProfiler &profiler) { + profiler.Start("Rerouting_Send_NB"); + profiler.Start("Rerouting_GetNextBuffer"); + ManagedBuffer &buffer = sendBuffers.GetNextBuffer(); + profiler.Stop("Rerouting_GetNextBuffer"); + buffer.m_Request = msg.NonBlockingSendTo(comm, toRank, buffer.m_Data); + buffer.m_Active = true; + profiler.Stop("Rerouting_Send_NB"); + }; + + auto lf_SendBlocking = [&sendBuffer](RerouteMessage &msg, adios2::helper::Comm &comm, + int toRank, adios2::profiling::JSONProfiler &profiler) { + profiler.Start("Rerouting_Send_B"); + msg.BlockingSendTo(comm, toRank, sendBuffer); + profiler.Stop("Rerouting_Send_B"); + }; + + auto lf_RecvBlocking = [&recvBuffer](RerouteMessage &msg, adios2::helper::Comm &comm, + int fromRank, adios2::profiling::JSONProfiler &profiler) { + profiler.Start("Rerouting_Recv_NB"); + msg.BlockingRecvFrom(comm, fromRank, recvBuffer); + profiler.Stop("Rerouting_Recv_NB"); + }; + if (iAmGlobalCoord) { // Global coordinator initializes the state it tracks for each subcoord @@ -299,6 +351,7 @@ void BP5Writer::ReroutingCommunicationLoop() while (lf_keepGoing()) { + bool workDone = false; // Flag to decide whether to yield on this iteration int msgReady = 0; helper::Comm::Status status = m_Comm.Iprobe(static_cast(helper::Comm::Constants::CommRecvAny), 0, &msgReady); @@ -307,7 +360,8 @@ void BP5Writer::ReroutingCommunicationLoop() if (msgReady) { RerouteMessage message; - lf_RecvBlocking(message, m_Comm, status.Source, recvBuffer, m_Profiler); + lf_RecvBlocking(message, m_Comm, status.Source, m_Profiler); + workDone = true; switch ((RerouteMessage::MessageType)message.m_MsgType) { @@ -383,8 +437,7 @@ void BP5Writer::ReroutingCommunicationLoop() closeAckMsg.m_SrcRank = m_RankMPI; closeAckMsg.m_DestRank = globalCoord; closeAckMsg.m_WildCard = static_cast(m_Aggregator->m_SubStreamIndex); - lf_SendBlocking(closeAckMsg, m_Comm, globalCoord, sendBuffers.GetNextBuffer(), - m_Profiler); + lf_SendBlocking(closeAckMsg, m_Comm, globalCoord, m_Profiler); break; case RerouteMessage::MessageType::GROUP_CLOSE_ACK: // msg for global coordinator @@ -433,8 +486,7 @@ void BP5Writer::ReroutingCommunicationLoop() inquiryMsg.m_MsgType = RerouteMessage::MessageType::STATUS_INQUIRY; inquiryMsg.m_SrcRank = m_RankMPI; inquiryMsg.m_DestRank = scRank; - lf_SendNonBlocking(inquiryMsg, m_Comm, scRank, - sendBuffers.GetNextBuffer(), m_Profiler); + lf_SendNonBlocking(inquiryMsg, m_Comm, scRank, m_Profiler); } } @@ -471,8 +523,7 @@ void BP5Writer::ReroutingCommunicationLoop() // The response to the status query is my subfile and queue size replyMsg.m_WildCard = static_cast(m_Aggregator->m_SubStreamIndex); replyMsg.m_Size = static_cast(writerQueue.size()); - lf_SendNonBlocking(replyMsg, m_Comm, globalCoord, sendBuffers.GetNextBuffer(), - m_Profiler); + lf_SendNonBlocking(replyMsg, m_Comm, globalCoord, m_Profiler); break; case RerouteMessage::MessageType::STATUS_REPLY: if (m_Parameters.verbose > 3) @@ -507,8 +558,7 @@ void BP5Writer::ReroutingCommunicationLoop() rejectMsg.m_MsgType = RerouteMessage::MessageType::REROUTE_REJECT; rejectMsg.m_SrcRank = message.m_SrcRank; rejectMsg.m_DestRank = message.m_DestRank; - lf_SendNonBlocking(rejectMsg, m_Comm, globalCoord, sendBuffers.GetNextBuffer(), - m_Profiler); + lf_SendNonBlocking(rejectMsg, m_Comm, globalCoord, m_Profiler); } else { @@ -527,8 +577,7 @@ void BP5Writer::ReroutingCommunicationLoop() ackMsg.m_SrcRank = message.m_SrcRank; ackMsg.m_DestRank = message.m_DestRank; ackMsg.m_WildCard = reroutedRank; - lf_SendNonBlocking(ackMsg, m_Comm, globalCoord, sendBuffers.GetNextBuffer(), - m_Profiler); + lf_SendNonBlocking(ackMsg, m_Comm, globalCoord, m_Profiler); } break; case RerouteMessage::MessageType::REROUTE_REJECT: @@ -579,8 +628,7 @@ void BP5Writer::ReroutingCommunicationLoop() adios2::helper::RerouteMessage writeMoreMsg; writeMoreMsg.m_MsgType = RerouteMessage::MessageType::WRITE_MORE; writeMoreMsg.m_WildCard = message.m_WildCard; // i.e. the rerouted writer rank - lf_SendNonBlocking(writeMoreMsg, m_Comm, message.m_DestRank, - sendBuffers.GetNextBuffer(), m_Profiler); + lf_SendNonBlocking(writeMoreMsg, m_Comm, message.m_DestRank, m_Profiler); groupIdlesNeeded.insert(message.m_DestRank); @@ -643,8 +691,7 @@ void BP5Writer::ReroutingCommunicationLoop() // done at this point. However, I need to do a blocking send because I // am about to return from this function, at which point my buffer pool // goes away. - lf_SendBlocking(writeCompleteMsg, m_Comm, m_TargetCoordinator, - sendBuffers.GetNextBuffer(), m_Profiler); + lf_SendBlocking(writeCompleteMsg, m_Comm, m_TargetCoordinator, m_Profiler); receivedGroupClose = true; continue; @@ -657,10 +704,10 @@ void BP5Writer::ReroutingCommunicationLoop() << m_TargetCoordinator << ") of write completion -- NONBLOCKING" << std::endl; } - lf_SendNonBlocking(writeCompleteMsg, m_Comm, m_TargetCoordinator, - sendBuffers.GetNextBuffer(), m_Profiler); + lf_SendNonBlocking(writeCompleteMsg, m_Comm, m_TargetCoordinator, m_Profiler); } + workDone = true; sentFinished = true; } } @@ -679,16 +726,34 @@ void BP5Writer::ReroutingCommunicationLoop() << std::endl; } writerQueue.pop(); - adios2::helper::RerouteMessage writeMsg; - writeMsg.m_MsgType = RerouteMessage::MessageType::DO_WRITE; - writeMsg.m_SrcRank = m_RankMPI; - writeMsg.m_DestRank = nextWriter; - writeMsg.m_WildCard = static_cast(m_Aggregator->m_SubStreamIndex); - writeMsg.m_Offset = currentFilePos; - writingRank = nextWriter; - expectingWriteCompletion = true; - lf_SendNonBlocking(writeMsg, m_Comm, nextWriter, sendBuffers.GetNextBuffer(), - m_Profiler); + + if (nextWriter == m_RankMPI) + { + // Don't use MPI to tell ourselves to write + std::unique_lock lck(m_WriteMutex); + m_TargetIndex = static_cast(m_Aggregator->m_SubStreamIndex); + m_DataPos = currentFilePos; + m_TargetCoordinator = m_RankMPI; + m_ReadyToWrite = true; + m_WriteCV.notify_one(); + + writingRank = m_RankMPI; + expectingWriteCompletion = true; + } + else + { + adios2::helper::RerouteMessage writeMsg; + writeMsg.m_MsgType = RerouteMessage::MessageType::DO_WRITE; + writeMsg.m_SrcRank = m_RankMPI; + writeMsg.m_DestRank = nextWriter; + writeMsg.m_WildCard = static_cast(m_Aggregator->m_SubStreamIndex); + writeMsg.m_Offset = currentFilePos; + writingRank = nextWriter; + expectingWriteCompletion = true; + lf_SendNonBlocking(writeMsg, m_Comm, nextWriter, m_Profiler); + } + + workDone = true; } else if (!sentIdle) { @@ -713,9 +778,9 @@ void BP5Writer::ReroutingCommunicationLoop() idleMsg.m_SrcRank = m_RankMPI; idleMsg.m_DestRank = globalCoord; idleMsg.m_WildCard = static_cast(m_Aggregator->m_SubStreamIndex); - lf_SendNonBlocking(idleMsg, m_Comm, globalCoord, sendBuffers.GetNextBuffer(), - m_Profiler); + lf_SendNonBlocking(idleMsg, m_Comm, globalCoord, m_Profiler); sentIdle = true; + workDone = true; } } @@ -751,11 +816,12 @@ void BP5Writer::ReroutingCommunicationLoop() rerouteReqMsg.m_MsgType = RerouteMessage::MessageType::REROUTE_REQUEST; rerouteReqMsg.m_SrcRank = writerSubcoordRank; rerouteReqMsg.m_DestRank = idleSubcoordRank; - lf_SendNonBlocking(rerouteReqMsg, m_Comm, writerSubcoordRank, - sendBuffers.GetNextBuffer(), m_Profiler); + lf_SendNonBlocking(rerouteReqMsg, m_Comm, writerSubcoordRank, m_Profiler); groupState[idleIdx].m_currentStatus = WriterGroupState::Status::PENDING; groupState[writerIdx].m_currentStatus = WriterGroupState::Status::PENDING; + + workDone = true; } else if (result == StateTraversal::SearchResult::FINISHED && groupIdlesNeeded.empty()) { @@ -785,8 +851,7 @@ void BP5Writer::ReroutingCommunicationLoop() } adios2::helper::RerouteMessage closeMsg; closeMsg.m_MsgType = RerouteMessage::MessageType::GROUP_CLOSE; - lf_SendNonBlocking(closeMsg, m_Comm, subCoordRanks[scIdx], - sendBuffers.GetNextBuffer(), m_Profiler); + lf_SendNonBlocking(closeMsg, m_Comm, subCoordRanks[scIdx], m_Profiler); } } @@ -858,6 +923,12 @@ void BP5Writer::ReroutingCommunicationLoop() } } } + + if (!workDone) + { + // Yield to avoid busy-cycling the CPU when nothing productive was done + std::this_thread::yield(); + } } // Before leaving this method, subcoordinators need to update the variable tracking @@ -867,6 +938,15 @@ void BP5Writer::ReroutingCommunicationLoop() m_DataPos = currentFilePos; } + if (m_Parameters.verbose > 3) + { + std::cout << "Rank " << m_RankMPI << " flushing pending sends..." << std::endl; + } + + // Ensure all ISends associated with buffers in the pool are complete + // before we let the pool go out of scope + sendBuffers.Flush(); + if (m_Parameters.verbose > 3) { std::cout << "Rank " << m_RankMPI << " Exit ReroutingCommunicationLoop" << std::endl;