From c448ea2a1174246deeb9e17631fc3baec4c16228 Mon Sep 17 00:00:00 2001
From: Mark Olesen <Mark.Olesen@esi-group.com>
Date: Tue, 25 Feb 2025 08:39:16 +0100
Subject: [PATCH] ENH: add simple handling for point-to-point barriers

- provides an additional means of process synchronization
---
 .../test/parallel-barrier1/Make/files         |   3 +
 .../test/parallel-barrier1/Make/options       |   2 +
 .../Test-parallel-barrier1.cxx                | 125 ++++++++++++++++++
 src/OpenFOAM/db/IOstreams/Pstreams/UPstream.H |  34 ++++-
 src/Pstream/dummy/UPstream.C                  |  26 +++-
 src/Pstream/mpi/PstreamGlobals.H              |   2 +-
 src/Pstream/mpi/UPstream.C                    |  67 +++++++++-
 7 files changed, 248 insertions(+), 11 deletions(-)
 create mode 100644 applications/test/parallel-barrier1/Make/files
 create mode 100644 applications/test/parallel-barrier1/Make/options
 create mode 100644 applications/test/parallel-barrier1/Test-parallel-barrier1.cxx

diff --git a/applications/test/parallel-barrier1/Make/files b/applications/test/parallel-barrier1/Make/files
new file mode 100644
index 0000000000..9256068d0b
--- /dev/null
+++ b/applications/test/parallel-barrier1/Make/files
@@ -0,0 +1,3 @@
+Test-parallel-barrier1.cxx
+
+EXE = $(FOAM_USER_APPBIN)/Test-parallel-barrier1
diff --git a/applications/test/parallel-barrier1/Make/options b/applications/test/parallel-barrier1/Make/options
new file mode 100644
index 0000000000..18e6fe47af
--- /dev/null
+++ b/applications/test/parallel-barrier1/Make/options
@@ -0,0 +1,2 @@
+/* EXE_INC = */
+/* EXE_LIBS = */
diff --git a/applications/test/parallel-barrier1/Test-parallel-barrier1.cxx b/applications/test/parallel-barrier1/Test-parallel-barrier1.cxx
new file mode 100644
index 0000000000..586539752d
--- /dev/null
+++ b/applications/test/parallel-barrier1/Test-parallel-barrier1.cxx
@@ -0,0 +1,125 @@
+/*---------------------------------------------------------------------------*\
+  =========                 |
+  \\      /  F ield         | OpenFOAM: The Open Source CFD Toolbox
+   \\    /   O peration     |
+    \\  /    A nd           | www.openfoam.com
+     \\/     M anipulation  |
+-------------------------------------------------------------------------------
+    Copyright (C) 2025 OpenCFD Ltd.
+-------------------------------------------------------------------------------
+License
+    This file is part of OpenFOAM.
+
+    OpenFOAM is free software: you can redistribute it and/or modify it
+    under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    OpenFOAM is distributed in the hope that it will be useful, but WITHOUT
+    ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+    FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+    for more details.
+
+    You should have received a copy of the GNU General Public License
+    along with OpenFOAM.  If not, see <http://www.gnu.org/licenses/>.
+
+Application
+    Test-parallel-barrier1
+
+Description
+    Simple test of local barriers communication
+
+\*---------------------------------------------------------------------------*/
+
+#include "argList.H"
+#include "clockTime.H"
+#include "IPstream.H"
+#include "OPstream.H"
+
+using namespace Foam;
+
+// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
+
+int main(int argc, char *argv[])
+{
+    argList::noCheckProcessorDirectories();
+    argList::addVerboseOption();
+    argList::addOption("delay", "sec", "Seconds to sleep (default 2)");
+
+    #include "setRootCase.H"
+
+    if (!UPstream::parRun())
+    {
+        Info<< "###############" << nl
+            << "Not running in parallel. Stopping now" << nl
+            << "###############" << endl;
+        return 1;
+    }
+
+    const auto delay = args.getOrDefault<label>("delay", 2);
+
+    Info<< nl
+        << "Testing local barrier, sleep=" << delay << endl;
+
+
+    const auto myProci = UPstream::myProcNo(UPstream::worldComm);
+    const auto numProc = UPstream::nProcs(UPstream::worldComm);
+
+    constexpr int uniqTag = 1516;
+
+    clockTime timing;
+
+    if (UPstream::master(UPstream::worldComm))
+    {
+        // Wait for the last rank
+        UPstream::wait_done(numProc-1, UPstream::worldComm);
+
+        // Wait for any other rank
+        if (numProc > 2)
+        {
+            int from = UPstream::wait_done(-1, UPstream::worldComm, uniqTag);
+            Pout<< "done signal from: " << from << endl;
+        }
+    }
+    else if (myProci == numProc-1)
+    {
+        Foam::sleep(delay);
+        UPstream::send_done(UPstream::masterNo(), UPstream::worldComm);
+    }
+
+    // Cascade sequencing (and delays)
+    if (numProc > 7)
+    {
+        if (myProci == 2)
+        {
+            Foam::sleep(2*delay);
+            UPstream::send_done(4, UPstream::worldComm);
+        }
+        else if (myProci == 4)
+        {
+            UPstream::wait_done(2, UPstream::worldComm);
+            Foam::sleep(2*delay);
+            UPstream::send_done(5, UPstream::worldComm);
+        }
+        else if (myProci == 5)
+        {
+            UPstream::wait_done(4, UPstream::worldComm);
+        }
+    }
+
+    // Some arbitrary signaling rank
+    if ((numProc > 2) && (myProci == numProc/2))
+    {
+        Pout<< "send done signal " << myProci << " -> 0" << endl;
+        UPstream::send_done(UPstream::masterNo(), UPstream::worldComm, uniqTag);
+    }
+
+    Pout<< "done: " << timing.elapsedTime() << " s" << endl;
+
+    Info<< "\nEnd\n" << endl;
+
+    return 0;
+}
+
+
+// ************************************************************************* //
diff --git a/src/OpenFOAM/db/IOstreams/Pstreams/UPstream.H b/src/OpenFOAM/db/IOstreams/Pstreams/UPstream.H
index e9ff4fc371..a6c940fb9d 100644
--- a/src/OpenFOAM/db/IOstreams/Pstreams/UPstream.H
+++ b/src/OpenFOAM/db/IOstreams/Pstreams/UPstream.H
@@ -777,10 +777,38 @@ public:
         //- Impose a synchronisation barrier (optionally non-blocking)
         static void barrier
         (
-            const label communicator,
+            const int communicator,
             UPstream::Request* req = nullptr
         );
 
+        //- Impose a point-to-point synchronisation barrier
+        //- by sending a zero-byte \em "done" message to given rank.
+        //  A no-op for non-parallel
+        static void send_done
+        (
+            //! The destination rank
+            const int toProcNo,
+            //! The communicator index (eg, UPstream::worldComm)
+            const int communicator,
+            //! Message tag (must match on receiving side)
+            const int tag = UPstream::msgType()+1970
+        );
+
+        //- Impose a point-to-point synchronisation barrier
+        //- by receiving a zero-byte \em "done" message from given rank
+        //  A no-op for non-parallel
+        //  \returns the source rank (useful for ANY_SOURCE messages)
+        //  or -1 for non-parallel
+        static int wait_done
+        (
+            //! The source rank (negative == ANY_SOURCE)
+            const int fromProcNo,
+            //! The communicator index (eg, UPstream::worldComm)
+            const int communicator,
+            //! Message tag (must match on sending side)
+            const int tag = UPstream::msgType()+1970
+        );
+
         //- Probe for an incoming message.
         //
         //  \param commsType    Non-blocking or not
@@ -795,7 +823,7 @@ public:
             const UPstream::commsTypes commsType,
             const int fromProcNo,
             const int tag = UPstream::msgType(),
-            const label communicator = worldComm
+            const int communicator = worldComm
         );
 
 
@@ -824,7 +852,7 @@ public:
 
         //- Transfer the (wrapped) MPI request to the internal global list
         //- and invalidate the parameter (ignores null requests)
-        //  A no-op for non-parallel,
+        //  A no-op for non-parallel
         static void addRequest(UPstream::Request& req);
 
         //- Non-blocking comms: cancel and free outstanding request.
diff --git a/src/Pstream/dummy/UPstream.C b/src/Pstream/dummy/UPstream.C
index 50cf97f607..4f70336800 100644
--- a/src/Pstream/dummy/UPstream.C
+++ b/src/Pstream/dummy/UPstream.C
@@ -109,20 +109,40 @@ void Foam::UPstream::freeCommunicatorComponents(const label index)
 {}
 
 
-void Foam::UPstream::barrier(const label communicator, UPstream::Request* req)
+void Foam::UPstream::barrier(const int communicator, UPstream::Request* req)
 {}
 
 
+void Foam::UPstream::send_done
+(
+    const int toProc,
+    const int communicator,
+    const int tag
+)
+{}
+
+
+int Foam::UPstream::wait_done
+(
+    const int fromProc,
+    const int communicator,
+    const int tag
+)
+{
+    return -1;
+}
+
+
 std::pair<int,int64_t>
 Foam::UPstream::probeMessage
 (
     const UPstream::commsTypes commsType,
     const int fromProcNo,
     const int tag,
-    const label communicator
+    const int communicator
 )
 {
-    return std::pair<int,int64_t>(-1, 0);
+    return {-1, 0};
 }
 
 
diff --git a/src/Pstream/mpi/PstreamGlobals.H b/src/Pstream/mpi/PstreamGlobals.H
index 2ccb245e33..db6396e2e1 100644
--- a/src/Pstream/mpi/PstreamGlobals.H
+++ b/src/Pstream/mpi/PstreamGlobals.H
@@ -113,7 +113,7 @@ inline void push_request
         // Transcribe as UPstream::Request
         *req = UPstream::Request(request);
     }
-    else
+    else if (MPI_REQUEST_NULL != request)
     {
         // Push onto list of requests
         PstreamGlobals::outstandingRequests_.push_back(request);
diff --git a/src/Pstream/mpi/UPstream.C b/src/Pstream/mpi/UPstream.C
index 82659a6b07..1edb6af58b 100644
--- a/src/Pstream/mpi/UPstream.C
+++ b/src/Pstream/mpi/UPstream.C
@@ -1050,10 +1050,10 @@ bool Foam::UPstream::setSharedMemoryCommunicators()
 }
 
 
-void Foam::UPstream::barrier(const label communicator, UPstream::Request* req)
+void Foam::UPstream::barrier(const int communicator, UPstream::Request* req)
 {
     // No-op for non-parallel or not on communicator
-    if (!UPstream::parRun() || !UPstream::is_rank(communicator))
+    if (!UPstream::is_parallel(communicator))
     {
         PstreamGlobals::reset_request(req);
         return;
@@ -1099,19 +1099,78 @@ void Foam::UPstream::barrier(const label communicator, UPstream::Request* req)
 }
 
 
+void Foam::UPstream::send_done
+(
+    const int toProcNo,
+    const int communicator,
+    const int tag  // Message tag (must match on receiving side)
+)
+{
+    if (!UPstream::is_parallel(communicator))
+    {
+        // Nothing to do
+        return;
+    }
+
+    {
+        MPI_Send
+        (
+            nullptr, 0, MPI_BYTE, toProcNo, tag,
+            PstreamGlobals::MPICommunicators_[communicator]
+        );
+    }
+}
+
+
+int Foam::UPstream::wait_done
+(
+    const int fromProcNo,
+    const int communicator,
+    const int tag  // Message tag (must match on sending side)
+)
+{
+    if (!UPstream::is_parallel(communicator))
+    {
+        // Nothing to do
+        return -1;
+    }
+    else if (fromProcNo < 0)
+    {
+        MPI_Status status;
+        MPI_Recv
+        (
+            nullptr, 0, MPI_BYTE, MPI_ANY_SOURCE, tag,
+            PstreamGlobals::MPICommunicators_[communicator],
+           &status
+        );
+        return status.MPI_SOURCE;
+    }
+    else
+    {
+        MPI_Recv
+        (
+            nullptr, 0, MPI_BYTE, fromProcNo, tag,
+            PstreamGlobals::MPICommunicators_[communicator],
+            MPI_STATUS_IGNORE
+        );
+        return fromProcNo;
+    }
+}
+
+
 std::pair<int,int64_t>
 Foam::UPstream::probeMessage
 (
     const UPstream::commsTypes commsType,
     const int fromProcNo,
     const int tag,
-    const label communicator
+    const int communicator
 )
 {
     std::pair<int,int64_t> result(-1, 0);
 
     // No-op for non-parallel or not on communicator
-    if (!UPstream::parRun() || !UPstream::is_rank(communicator))
+    if (!UPstream::is_parallel(communicator))
     {
         return result;
     }