ENH: replace PstreamBuffers with scheduled read/writes for VTK output

- use the new updates to globalIndex to manage the bookkeeping

src/fileFormats/vtk/output/foamVtkOutput.C

@@ -39,7 +39,7 @@ License
 #include "globalIndex.H"
 #include "instant.H"
 #include "Fstream.H"
-#include "PstreamBuffers.H"
+#include "Pstream.H"
 #include "OSspecific.H"
 
 // * * * * * * * * * * * * * * Global Functions  * * * * * * * * * * * * * * //
@@ -129,25 +129,17 @@ void Foam::vtk::writeListParallel
     const globalIndex& procOffset
 )
 {
-    // List sizes
+    // Gather sizes - master information, offsets are irrelevant
-    const globalIndex sizes(values.size());
+    const globalIndex procAddr
+    (
+        UPstream::listGatherValues<label>(values.size()),
+        globalIndex::SIZES
+    );
 
-    PstreamBuffers pBufs(Pstream::commsTypes::nonBlocking);
-
-    // Send to master
-    if (!Pstream::master())
-    {
-        UOPstream os(Pstream::masterNo(), pBufs);
-        os.write(values.cdata_bytes(), values.size_bytes());
-    }
-
-    pBufs.finishedSends();
 
     if (Pstream::master())
     {
-        // Master data
+        // Write master data - with value offset
-
-        // Write with offset
         const label offsetId = procOffset.offset(0);
         for (const label val : values)
         {
@@ -155,21 +147,38 @@ void Foam::vtk::writeListParallel
         }
 
         // Receive and write
-        for (const int proci : Pstream::subProcs())
+        DynamicList<label> recvData(procAddr.maxNonLocalSize());
+
+        for (const label proci : procAddr.subProcs())
         {
-            List<label> recv(sizes.localSize(proci));
+            recvData.resize_nocopy(procAddr.localSize(proci));
+            UIPstream::read
+            (
+                UPstream::commsTypes::scheduled,
+                proci,
+                recvData.data_bytes(),
+                recvData.size_bytes()
+            );
 
-            UIPstream is(proci, pBufs);
+            // With value offset
-            is.read(recv.data_bytes(), recv.size_bytes());
-
-            // Write with offset
             const label offsetId = procOffset.offset(proci);
-            for (const label val : recv)
+            for (const label val : recvData)
             {
                 vtk::write(fmt, val + offsetId);
             }
         }
     }
+    else
+    {
+        // Send
+        UOPstream::write
+        (
+            UPstream::commsTypes::scheduled,
+            Pstream::masterNo(),
+            values.cdata_bytes(),
+            values.size_bytes()
+        );
+    }
 }
 
 

src/fileFormats/vtk/output/foamVtkOutputTemplates.C

@@ -26,7 +26,7 @@ License
 \*---------------------------------------------------------------------------*/
 
 #include "globalIndex.H"
-#include "PstreamBuffers.H"
+#include "Pstream.H"
 #include "ListOps.H"
 
 // * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
@@ -131,33 +131,31 @@ void Foam::vtk::writeValueParallel
     const label count
 )
 {
+    if (!is_contiguous<Type>::value)
+    {
+        // Non-contiguous data does not make sense
+        FatalErrorInFunction
+            << "Contiguous data only" << endl
+            << Foam::exit(FatalError);
+    }
+
+    // Gather [count, value] tuples, including from master
+    const List<std::pair<label, Type>> countValues
+    (
+        UPstream::listGatherValues<std::pair<label, Type>>
+        (
+            std::pair<label, Type>(count, val)
+        )
+    );
+
     if (Pstream::master())
     {
-        vtk::write(fmt, val, count);
+        for (const auto& countVal : countValues)
-
-        label subCount;
-        Type subValue;
-
-        // Receive each [size, value] tuple
-        for (const int proci : Pstream::subProcs())
         {
-            IPstream is(Pstream::commsTypes::blocking, proci);
+            // Write [count, value] tuple
-            is >> subCount >> subValue;
+            vtk::write(fmt, countVal.first, countVal.second);
-
-            vtk::write(fmt, subValue, subCount);
         }
     }
-    else
-    {
-        OPstream os
-        (
-            Pstream::commsTypes::blocking,
-            Pstream::masterNo()
-        );
-
-        // Send [size, value] tuple
-        os << count << val;
-    }
 }
 
 
@@ -168,26 +166,22 @@ void Foam::vtk::writeListParallel
     const UList<Type>& values
 )
 {
-    // List sizes
+    if (!is_contiguous<Type>::value)
-    const globalIndex sizes(values.size());
-
-    PstreamBuffers pBufs(Pstream::commsTypes::nonBlocking);
-
-    // Send to master
-    if (!Pstream::master())
     {
-        UOPstream os(Pstream::masterNo(), pBufs);
+        // Non-contiguous data does not make sense
-        if (is_contiguous<Type>::value)
+        FatalErrorInFunction
-        {
+            << "Contiguous data only" << endl
-            os.write(values.cdata_bytes(), values.size_bytes());
+            << Foam::exit(FatalError);
-        }
-        else
-        {
-            os << values;
-        }
     }
 
-    pBufs.finishedSends();
+
+    // Gather sizes - master information, offsets are irrelevant
+    const globalIndex procAddr
+    (
+        UPstream::listGatherValues<label>(values.size()),
+        globalIndex::SIZES
+    );
+
 
     if (Pstream::master())
     {
@@ -195,25 +189,32 @@ void Foam::vtk::writeListParallel
         vtk::writeList(fmt, values);
 
         // Receive and write
-        for (const int proci : Pstream::subProcs())
+        DynamicList<Type> recvData(procAddr.maxNonLocalSize());
+
+        for (const label proci : procAddr.subProcs())
         {
-            UIPstream is(proci, pBufs);
+            recvData.resize_nocopy(procAddr.localSize(proci));
-
+            UIPstream::read
-            {
+            (
-                List<Type> recv(sizes.localSize(proci));
+                UPstream::commsTypes::scheduled,
-
+                proci,
-                if (is_contiguous<Type>::value)
+                recvData.data_bytes(),
-                {
+                recvData.size_bytes()
-                    is.read(recv.data_bytes(), recv.size_bytes());
+            );
-                }
+            vtk::writeList(fmt, recvData);
-                else
-                {
-                    is >> recv;
-                }
-                vtk::writeList(fmt, recv);
-            }
         }
     }
+    else
+    {
+        // Send
+        UOPstream::write
+        (
+            UPstream::commsTypes::scheduled,
+            Pstream::masterNo(),
+            values.cdata_bytes(),
+            values.size_bytes()
+        );
+    }
 }
 
 
@@ -225,21 +226,28 @@ void Foam::vtk::writeListParallel
     const labelUList& addressing
 )
 {
-    UIndirectList<Type> send(values, addressing);
+    if (!is_contiguous<Type>::value)
-
-    // List sizes
-    const globalIndex sizes(send.size());
-
-    PstreamBuffers pBufs(Pstream::commsTypes::nonBlocking);
-
-    // Send to master
-    if (!Pstream::master())
     {
-        UOPstream os(Pstream::masterNo(), pBufs);
+        // Non-contiguous data does not make sense
-        os << send;
+        FatalErrorInFunction
+            << "Contiguous data only" << endl
+            << Foam::exit(FatalError);
     }
 
-    pBufs.finishedSends();
+
+    List<Type> sendData;
+    if (!Pstream::master())
+    {
+        sendData = UIndirectList<Type>(values, addressing);
+    }
+
+    // Gather sizes - master information, offsets are irrelevant
+    const globalIndex procAddr
+    (
+        UPstream::listGatherValues<label>(sendData.size()),
+        globalIndex::SIZES
+    );
+
 
     if (Pstream::master())
     {
@@ -247,17 +255,31 @@ void Foam::vtk::writeListParallel
         vtk::writeList(fmt, values, addressing);
 
         // Receive and write
-        for (const int proci : Pstream::subProcs())
+        DynamicList<Type> recvData(procAddr.maxNonLocalSize());
-        {
-            UIPstream is(proci, pBufs);
 
-            {
+        for (const label proci : procAddr.subProcs())
-                List<Type> recv;
+        {
-                is >> recv;
+            recvData.resize_nocopy(procAddr.localSize(proci));
-                vtk::writeList(fmt, recv);
+            UIPstream::read
-            }
+            (
+                UPstream::commsTypes::scheduled,
+                proci,
+                recvData.data_bytes(),
+                recvData.size_bytes()
+            );
+            vtk::writeList(fmt, recvData);
         }
     }
+    else
+    {
+        UOPstream::write
+        (
+            UPstream::commsTypes::scheduled,
+            Pstream::masterNo(),
+            sendData.cdata_bytes(),
+            sendData.size_bytes()
+        );
+    }
 }
 
 
@@ -269,33 +291,28 @@ void Foam::vtk::writeListParallel
     const bitSet& selected
 )
 {
-    List<Type> send;
+    if (!is_contiguous<Type>::value)
-    if (!Pstream::master())
     {
-        send = subset(selected, values);
+        // Non-contiguous data does not make sense
+        FatalErrorInFunction
+            << "Contiguous data only" << endl
+            << Foam::exit(FatalError);
     }
 
-    // List sizes.
-    // NOTE okay to skip proc0 since we only need sizes (not offsets)
-    const globalIndex sizes(send.size());
 
-    PstreamBuffers pBufs(Pstream::commsTypes::nonBlocking);
+    List<Type> sendData;
-
-    // Send to master
     if (!Pstream::master())
     {
-        UOPstream os(Pstream::masterNo(), pBufs);
+        sendData = subset(selected, values);
-        if (is_contiguous<Type>::value)
-        {
-            os.write(send.cdata_bytes(), send.size_bytes());
-        }
-        else
-        {
-            os << send;
-        }
     }
 
-    pBufs.finishedSends();
+    // Gather sizes - master information, offsets are irrelevant
+    const globalIndex procAddr
+    (
+        UPstream::listGatherValues<label>(sendData.size()),
+        globalIndex::SIZES
+    );
+
 
     if (Pstream::master())
     {
@@ -303,25 +320,31 @@ void Foam::vtk::writeListParallel
         vtk::writeList(fmt, values, selected);
 
         // Receive and write
-        for (const int proci : Pstream::subProcs())
+        DynamicList<Type> recvData(procAddr.maxNonLocalSize());
+
+        for (const label proci : procAddr.subProcs())
         {
-            UIPstream is(proci, pBufs);
+            recvData.resize_nocopy(procAddr.localSize(proci));
-
+            UIPstream::read
-            {
+            (
-                List<Type> recv(sizes.localSize(proci));
+                UPstream::commsTypes::scheduled,
-
+                proci,
-                if (is_contiguous<Type>::value)
+                recvData.data_bytes(),
-                {
+                recvData.size_bytes()
-                    is.read(recv.data_bytes(), recv.size_bytes());
+            );
-                }
+            vtk::writeList(fmt, recvData);
-                else
-                {
-                    is >> recv;
-                }
-                vtk::writeList(fmt, recv);
-            }
         }
     }
+    else
+    {
+        UOPstream::write
+        (
+            UPstream::commsTypes::scheduled,
+            Pstream::masterNo(),
+            sendData.cdata_bytes(),
+            sendData.size_bytes()
+        );
+    }
 }
 
 
@@ -333,28 +356,27 @@ void Foam::vtk::writeListsParallel
     const UList<Type>& values2
 )
 {
-    // List sizes
+    if (!is_contiguous<Type>::value)
-    const globalIndex sizes1(values1.size());
-    const globalIndex sizes2(values2.size());
-
-    PstreamBuffers pBufs(Pstream::commsTypes::nonBlocking);
-
-    // Send to master
-    if (!Pstream::master())
     {
-        UOPstream os(Pstream::masterNo(), pBufs);
+        // Non-contiguous data does not make sense
-        if (is_contiguous<Type>::value)
+        FatalErrorInFunction
-        {
+            << "Contiguous data only" << endl
-            os.write(values1.cdata_bytes(), values1.size_bytes());
+            << Foam::exit(FatalError);
-            os.write(values2.cdata_bytes(), values2.size_bytes());
-        }
-        else
-        {
-            os << values1 << values2;
-        }
     }
 
-    pBufs.finishedSends();
+
+    // Gather sizes - master information and offsets are irrelevant
+    const globalIndex procAddr1
+    (
+        UPstream::listGatherValues<label>(values1.size()),
+        globalIndex::SIZES
+    );
+    const globalIndex procAddr2
+    (
+        UPstream::listGatherValues<label>(values2.size()),
+        globalIndex::SIZES
+    );
+
 
     if (Pstream::master())
     {
@@ -362,46 +384,55 @@ void Foam::vtk::writeListsParallel
         vtk::writeList(fmt, values1);
         vtk::writeList(fmt, values2);
 
-        // Reserve max receive size
+        // Receive and write
-        DynamicList<Type> recv
+        DynamicList<Type> recvData
         (
-            max(sizes1.maxNonLocalSize(), sizes2.maxNonLocalSize())
+            max(procAddr1.maxNonLocalSize(), procAddr2.maxNonLocalSize())
         );
 
-        // Receive and write
+        for (const label proci : procAddr1.subProcs())
-        for (const int proci : Pstream::subProcs())
         {
-            UIPstream is(proci, pBufs);
-
             // values1
-            {
+            recvData.resize_nocopy(procAddr1.localSize(proci));
-                List<Type> recv(sizes1.localSize(proci));
+            UIPstream::read
-                if (is_contiguous<Type>::value)
+            (
-                {
+                UPstream::commsTypes::scheduled,
-                    is.read(recv.data_bytes(), recv.size_bytes());
+                proci,
-                }
+                recvData.data_bytes(),
-                else
+                recvData.size_bytes()
-                {
+            );
-                    is >> recv;
+            vtk::writeList(fmt, recvData);
-                }
-                vtk::writeList(fmt, recv);
-            }
 
             // values2
-            {
+            recvData.resize_nocopy(procAddr2.localSize(proci));
-                List<Type> recv(sizes2.localSize(proci));
+            UIPstream::read
-                if (is_contiguous<Type>::value)
+            (
-                {
+                UPstream::commsTypes::scheduled,
-                    is.read(recv.data_bytes(), recv.size_bytes());
+                proci,
-                }
+                recvData.data_bytes(),
-                else
+                recvData.size_bytes()
-                {
+            );
-                    is >> recv;
+            vtk::writeList(fmt, recvData);
-                }
-                vtk::writeList(fmt, recv);
-            }
         }
     }
+    else
+    {
+        UOPstream::write
+        (
+            UPstream::commsTypes::scheduled,
+            Pstream::masterNo(),
+            values1.cdata_bytes(),
+            values1.size_bytes()
+        );
+
+        UOPstream::write
+        (
+            UPstream::commsTypes::scheduled,
+            Pstream::masterNo(),
+            values2.cdata_bytes(),
+            values2.size_bytes()
+        );
+    }
 }
 
 
@@ -414,45 +445,91 @@ void Foam::vtk::writeListsParallel
     const labelUList& addressing
 )
 {
-    UIndirectList<Type> send2(values2, addressing);
+    if (!is_contiguous<Type>::value)
-
-    PstreamBuffers pBufs(Pstream::commsTypes::nonBlocking);
-
-    // Send to master
-    if (!Pstream::master())
     {
-        UOPstream os(Pstream::masterNo(), pBufs);
+        // Non-contiguous data does not make sense
-        os << values1 << send2;
+        FatalErrorInFunction
+            << "Contiguous data only" << endl
+            << Foam::exit(FatalError);
     }
 
-    pBufs.finishedSends();
+
+    List<Type> sendData2;
+    if (!Pstream::master())
+    {
+        sendData2 = UIndirectList<Type>(values2, addressing);
+    }
+
+
+    // Gather sizes - master information, offsets are irrelevant
+    const globalIndex procAddr1
+    (
+        UPstream::listGatherValues<label>(values1.size()),
+        globalIndex::SIZES
+    );
+    const globalIndex procAddr2
+    (
+        UPstream::listGatherValues<label>(sendData2.size()),
+        globalIndex::SIZES
+    );
+
 
     if (Pstream::master())
     {
         // Write master data
+
         vtk::writeList(fmt, values1);
         vtk::writeList(fmt, values2, addressing);
 
         // Receive and write
-        for (const int proci : Pstream::subProcs())
+        DynamicList<Type> recvData
+        (
+            max(procAddr1.maxNonLocalSize(), procAddr2.maxNonLocalSize())
+        );
+
+        for (const label proci : procAddr1.subProcs())
         {
-            UIPstream is(proci, pBufs);
-
             // values1
-            {
+            recvData.resize_nocopy(procAddr1.localSize(proci));
-                List<Type> recv;
+            UIPstream::read
-                is >> recv;
+            (
-                vtk::writeList(fmt, recv);
+                UPstream::commsTypes::scheduled,
-            }
+                proci,
+                recvData.data_bytes(),
+                recvData.size_bytes()
+            );
+            vtk::writeList(fmt, recvData);
 
-            // values2 (send2)
+            // values2
-            {
+            recvData.resize_nocopy(procAddr2.localSize(proci));
-                List<Type> recv;
+            UIPstream::read
-                is >> recv;
+            (
-                vtk::writeList(fmt, recv);
+                UPstream::commsTypes::scheduled,
-            }
+                proci,
+                recvData.data_bytes(),
+                recvData.size_bytes()
+            );
+            vtk::writeList(fmt, recvData);
         }
     }
+    else
+    {
+        UOPstream::write
+        (
+            UPstream::commsTypes::scheduled,
+            Pstream::masterNo(),
+            values1.cdata_bytes(),
+            values1.size_bytes()
+        );
+
+        UOPstream::write
+        (
+            UPstream::commsTypes::scheduled,
+            Pstream::masterNo(),
+            sendData2.cdata_bytes(),
+            sendData2.size_bytes()
+        );
+    }
 }