From 1b47034e24d2b4613c171f77cbcc2189d3ab5cea Mon Sep 17 00:00:00 2001
From: mattijs <mattijs>
Date: Mon, 3 Feb 2020 11:26:43 +0000
Subject: [PATCH] ENH: Pstream: use native reduce in SPDP mode. Fixes #1574.

---
 .../db/IOstreams/Pstreams/PstreamReduceOps.H  |  44 ++++++
 src/Pstream/dummy/UPstream.C                  |  45 ++++++
 src/Pstream/mpi/UPstream.C                    | 133 +++++++++++++++++-
 3 files changed, 221 insertions(+), 1 deletion(-)

diff --git a/src/OpenFOAM/db/IOstreams/Pstreams/PstreamReduceOps.H b/src/OpenFOAM/db/IOstreams/Pstreams/PstreamReduceOps.H
index 1df03ba933..414e9097b0 100644
--- a/src/OpenFOAM/db/IOstreams/Pstreams/PstreamReduceOps.H
+++ b/src/OpenFOAM/db/IOstreams/Pstreams/PstreamReduceOps.H
@@ -200,6 +200,50 @@ void reduce
 );
 
 
+#if defined(WM_SPDP)
+void reduce
+(
+    solveScalar& Value,
+    const sumOp<solveScalar>& bop,
+    const int tag = Pstream::msgType(),
+    const label comm = UPstream::worldComm
+);
+
+void reduce
+(
+    solveScalar& Value,
+    const minOp<solveScalar>& bop,
+    const int tag = Pstream::msgType(),
+    const label comm = UPstream::worldComm
+);
+
+void reduce
+(
+    Vector2D<solveScalar>& Value,
+    const sumOp<Vector2D<solveScalar>>& bop,
+    const int tag = Pstream::msgType(),
+    const label comm = UPstream::worldComm
+);
+
+void sumReduce
+(
+    solveScalar& Value,
+    label& Count,
+    const int tag = Pstream::msgType(),
+    const label comm = UPstream::worldComm
+);
+
+void reduce
+(
+    solveScalar& Value,
+    const sumOp<solveScalar>& bop,
+    const int tag,
+    const label comm,
+    label& request
+);
+#endif
+
+
 // * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
 
 } // End namespace Foam
diff --git a/src/Pstream/dummy/UPstream.C b/src/Pstream/dummy/UPstream.C
index 313acad689..bf7d9af6e8 100644
--- a/src/Pstream/dummy/UPstream.C
+++ b/src/Pstream/dummy/UPstream.C
@@ -96,6 +96,51 @@ void Foam::reduce(scalar&, const sumOp<scalar>&, const int, const label, label&)
 {}
 
 
+#if defined(WM_SPDP)
+void Foam::reduce
+(
+    solveScalar& Value,
+    const sumOp<solveScalar>& bop,
+    const int tag,
+    const label comm
+)
+{}
+void Foam::reduce
+(
+    solveScalar& Value,
+    const minOp<solveScalar>& bop,
+    const int tag,
+    const label comm
+)
+{}
+void Foam::reduce
+(
+    Vector2D<solveScalar>& Value,
+    const sumOp<Vector2D<solveScalar>>& bop,
+    const int tag,
+    const label comm
+)
+{}
+void Foam::sumReduce
+(
+    solveScalar& Value,
+    label& Count,
+    const int tag,
+    const label comm
+)
+{}
+void Foam::reduce
+(
+    solveScalar& Value,
+    const sumOp<solveScalar>& bop,
+    const int tag,
+    const label comm,
+    label& request
+)
+{}
+#endif
+
+
 void Foam::UPstream::allToAll
 (
     const labelUList& sendData,
diff --git a/src/Pstream/mpi/UPstream.C b/src/Pstream/mpi/UPstream.C
index 98fedcd537..65db848206 100644
--- a/src/Pstream/mpi/UPstream.C
+++ b/src/Pstream/mpi/UPstream.C
@@ -40,10 +40,15 @@ License
 #include <cstdlib>
 #include <csignal>
 
-#if defined(WM_SP) || defined(WM_SPDP)
+#if defined(WM_SP)
     #define MPI_SCALAR MPI_FLOAT
+    #define MPI_SOLVESCALAR MPI_FLOAT
+#elif defined(WM_SPDP)
+    #define MPI_SCALAR MPI_FLOAT
+    #define MPI_SOLVESCALAR MPI_DOUBLE
 #elif defined(WM_DP)
     #define MPI_SCALAR MPI_DOUBLE
+    #define MPI_SOLVESCALAR MPI_DOUBLE
 #endif
 
 // * * * * * * * * * * * * * * Static Data Members * * * * * * * * * * * * * //
@@ -478,6 +483,132 @@ void Foam::reduce
 }
 
 
+#if defined(WM_SPDP)
+void Foam::reduce
+(
+    solveScalar& Value,
+    const sumOp<solveScalar>& bop,
+    const int tag,
+    const label communicator
+)
+{
+    if (UPstream::warnComm != -1 && communicator != UPstream::warnComm)
+    {
+        Pout<< "** reducing:" << Value << " with comm:" << communicator
+            << " warnComm:" << UPstream::warnComm
+            << endl;
+        error::printStack(Pout);
+    }
+    allReduce(Value, 1, MPI_SOLVESCALAR, MPI_SUM, bop, tag, communicator);
+}
+
+
+void Foam::reduce
+(
+    solveScalar& Value,
+    const minOp<solveScalar>& bop,
+    const int tag,
+    const label communicator
+)
+{
+    if (UPstream::warnComm != -1 && communicator != UPstream::warnComm)
+    {
+        Pout<< "** reducing:" << Value << " with comm:" << communicator
+            << " warnComm:" << UPstream::warnComm
+            << endl;
+        error::printStack(Pout);
+    }
+    allReduce(Value, 1, MPI_SOLVESCALAR, MPI_MIN, bop, tag, communicator);
+}
+
+
+void Foam::reduce
+(
+    Vector2D<solveScalar>& Value,
+    const sumOp<Vector2D<solveScalar>>& bop,
+    const int tag,
+    const label communicator
+)
+{
+    if (UPstream::warnComm != -1 && communicator != UPstream::warnComm)
+    {
+        Pout<< "** reducing:" << Value << " with comm:" << communicator
+            << " warnComm:" << UPstream::warnComm
+            << endl;
+        error::printStack(Pout);
+    }
+    allReduce(Value, 2, MPI_SOLVESCALAR, MPI_SUM, bop, tag, communicator);
+}
+
+
+void Foam::sumReduce
+(
+    solveScalar& Value,
+    label& Count,
+    const int tag,
+    const label communicator
+)
+{
+    if (UPstream::warnComm != -1 && communicator != UPstream::warnComm)
+    {
+        Pout<< "** reducing:" << Value << " with comm:" << communicator
+            << " warnComm:" << UPstream::warnComm
+            << endl;
+        error::printStack(Pout);
+    }
+    Vector2D<solveScalar> twoScalars(Value, solveScalar(Count));
+    reduce(twoScalars, sumOp<Vector2D<solveScalar>>(), tag, communicator);
+
+    Value = twoScalars.x();
+    Count = twoScalars.y();
+}
+
+
+void Foam::reduce
+(
+    solveScalar& Value,
+    const sumOp<solveScalar>& bop,
+    const int tag,
+    const label communicator,
+    label& requestID
+)
+{
+#ifdef MPIX_COMM_TYPE_SHARED
+    // Assume mpich2 with non-blocking collectives extensions. Once mpi3
+    // is available this will change.
+    MPI_Request request;
+    solveScalar v = Value;
+    MPIX_Ireduce
+    (
+        &v,
+        &Value,
+        1,
+        MPI_SOLVESCALAR,
+        MPI_SUM,
+        0,              //root
+        PstreamGlobals::MPICommunicators_[communicator],
+        &request
+    );
+
+    requestID = PstreamGlobals::outstandingRequests_.size();
+    PstreamGlobals::outstandingRequests_.append(request);
+
+    if (UPstream::debug)
+    {
+        Pout<< "UPstream::allocateRequest for non-blocking reduce"
+            << " : request:" << requestID
+            << endl;
+    }
+#else
+    // Non-blocking not yet implemented in mpi
+    reduce(Value, bop, tag, communicator);
+    requestID = -1;
+#endif
+}
+#endif
+
+
+
 void Foam::UPstream::allToAll
 (
     const labelUList& sendData,