Commit 2238b3bf authored by mattijs's avatar mattijs
Browse files

ENH: Pstream: added maxCommsSize setting to do (unstructured) parallel transfers in blocks.

This is controlled by the setting maxCommsSize in etc/controlDict which
specifies the max number of bytes per exchange. If set to <= 0 it
is ignored. This max size of messages is important when doing e.g.
load balancing which can send over whole meshes.
parent 166abf70
......@@ -3,7 +3,7 @@
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2011-2016 OpenFOAM Foundation
\\/ M anipulation |
\\/ M anipulation | Copyright (C) 2016 OpenCFD Ltd.
-------------------------------------------------------------------------------
License
This file is part of OpenFOAM.
......@@ -55,6 +55,37 @@ class Pstream
:
public UPstream
{
// Private Static Functions
//- Exchange contiguous data. Sends sendData, receives into
// recvData. If block=true will wait for all transfers to finish.
// Data provided and received as container.
template<class Container, class T>
static void exchangeContainer
(
const UList<Container>& sendBufs,
const labelUList& recvSizes,
List<Container>& recvBufs,
const int tag,
const label comm,
const bool block
);
//- Exchange contiguous data. Sends sendData, receives into
// recvData. If block=true will wait for all transfers to finish.
// Data provided and received as pointers.
template<class T>
static void exchangeBuf
(
const labelUList& sendSizes, // number of T, not number of char
const UList<const char*>& sendBufs,
const labelUList& recvSizes, // number of T, not number of char
List<char*>& recvBufs,
const int tag,
const label comm,
const bool block
);
protected:
......@@ -63,6 +94,7 @@ protected:
//- Transfer buffer
DynamicList<char> buf_;
public:
// Declare name of the class and its debug switch
......
......@@ -3,7 +3,7 @@
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2011-2016 OpenFOAM Foundation
\\/ M anipulation |
\\/ M anipulation | Copyright (C) 2016 OpenCFD Ltd.
-------------------------------------------------------------------------------
License
This file is part of OpenFOAM.
......@@ -32,7 +32,6 @@ Description
#ifndef PstreamReduceOps_H
#define PstreamReduceOps_H
#include "Pstream.H"
#include "ops.H"
#include "vector2D.H"
......
......@@ -464,4 +464,16 @@ registerOptSwitch
);
int Foam::UPstream::maxCommsSize
(
Foam::debug::optimisationSwitch("maxCommsSize", 0)
);
registerOptSwitch
(
"maxCommsSize",
int,
Foam::UPstream::maxCommsSize
);
// ************************************************************************* //
......@@ -268,6 +268,9 @@ public:
//- Number of polling cycles in processor updates
static int nPollProcInterfaces;
//- Optional maximum message size (bytes)
static int maxCommsSize;
//- Default communicator (all processors)
static label worldComm;
......
......@@ -3,7 +3,7 @@
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2011-2016 OpenFOAM Foundation
\\/ M anipulation |
\\/ M anipulation | Copyright (C) 2016 OpenCFD Ltd.
-------------------------------------------------------------------------------
License
This file is part of OpenFOAM.
......@@ -28,11 +28,157 @@ Description
#include "Pstream.H"
#include "contiguous.H"
#include "PstreamCombineReduceOps.H"
#include "UPstream.H"
#include "PstreamReduceOps.H"
// * * * * * * * * * * * * * * * Member Functions * * * * * * * * * * * * * //
template<class Container, class T>
void Foam::Pstream::exchangeContainer
(
const UList<Container>& sendBufs,
const labelUList& recvSizes,
List<Container>& recvBufs,
const int tag,
const label comm,
const bool block
)
{
label startOfRequests = Pstream::nRequests();
// Set up receives
// ~~~~~~~~~~~~~~~
forAll(recvSizes, proci)
{
if (proci != Pstream::myProcNo(comm) && recvSizes[proci] > 0)
{
UIPstream::read
(
UPstream::nonBlocking,
proci,
reinterpret_cast<char*>(recvBufs[proci].begin()),
recvSizes[proci]*sizeof(T),
tag,
comm
);
}
}
// Set up sends
// ~~~~~~~~~~~~
forAll(sendBufs, proci)
{
if (proci != Pstream::myProcNo(comm) && sendBufs[proci].size() > 0)
{
if
(
!UOPstream::write
(
UPstream::nonBlocking,
proci,
reinterpret_cast<const char*>(sendBufs[proci].begin()),
sendBufs[proci].size()*sizeof(T),
tag,
comm
)
)
{
FatalErrorInFunction
<< "Cannot send outgoing message. "
<< "to:" << proci << " nBytes:"
<< label(sendBufs[proci].size()*sizeof(T))
<< Foam::abort(FatalError);
}
}
}
// Wait for all to finish
// ~~~~~~~~~~~~~~~~~~~~~~
if (block)
{
Pstream::waitRequests(startOfRequests);
}
}
template<class T>
void Foam::Pstream::exchangeBuf
(
const labelUList& sendSizes,
const UList<const char*>& sendBufs,
const labelUList& recvSizes,
List<char*>& recvBufs,
const int tag,
const label comm,
const bool block
)
{
label startOfRequests = Pstream::nRequests();
// Set up receives
// ~~~~~~~~~~~~~~~
forAll(recvSizes, proci)
{
if (proci != Pstream::myProcNo(comm) && recvSizes[proci] > 0)
{
UIPstream::read
(
UPstream::nonBlocking,
proci,
recvBufs[proci],
recvSizes[proci]*sizeof(T),
tag,
comm
);
}
}
// Set up sends
// ~~~~~~~~~~~~
forAll(sendBufs, proci)
{
if (proci != Pstream::myProcNo(comm) && sendSizes[proci] > 0)
{
if
(
!UOPstream::write
(
UPstream::nonBlocking,
proci,
sendBufs[proci],
sendSizes[proci]*sizeof(T),
tag,
comm
)
)
{
FatalErrorInFunction
<< "Cannot send outgoing message. "
<< "to:" << proci << " nBytes:"
<< label(sendSizes[proci]*sizeof(T))
<< Foam::abort(FatalError);
}
}
}
// Wait for all to finish
// ~~~~~~~~~~~~~~~~~~~~~~
if (block)
{
Pstream::waitRequests(startOfRequests);
}
}
template<class Container, class T>
void Foam::Pstream::exchange
(
......@@ -63,11 +209,7 @@ void Foam::Pstream::exchange
if (UPstream::parRun() && UPstream::nProcs(comm) > 1)
{
label startOfRequests = Pstream::nRequests();
// Set up receives
// ~~~~~~~~~~~~~~~
// Presize all receive buffers
forAll(recvSizes, proci)
{
label nRecv = recvSizes[proci];
......@@ -75,55 +217,121 @@ void Foam::Pstream::exchange
if (proci != Pstream::myProcNo(comm) && nRecv > 0)
{
recvBufs[proci].setSize(nRecv);
UIPstream::read
(
UPstream::nonBlocking,
proci,
reinterpret_cast<char*>(recvBufs[proci].begin()),
nRecv*sizeof(T),
tag,
comm
);
}
}
// Set up sends
// ~~~~~~~~~~~~
forAll(sendBufs, proci)
if (Pstream::maxCommsSize <= 0)
{
if (proci != Pstream::myProcNo(comm) && sendBufs[proci].size() > 0)
// Do the exchanging in one go
exchangeContainer<Container, T>
(
sendBufs,
recvSizes,
recvBufs,
tag,
comm,
block
);
}
else
{
// Determine the number of chunks to send. Note that we
// only have to look at the sending data since we are
// guaranteed that some processor's sending size is some other
// processor's receive size. Also we can ignore any local comms.
label maxNSend = 0;
forAll(sendBufs, proci)
{
if
(
!UOPstream::write
(
UPstream::nonBlocking,
proci,
reinterpret_cast<const char*>(sendBufs[proci].begin()),
sendBufs[proci].size()*sizeof(T),
tag,
comm
)
)
if (proci != Pstream::myProcNo(comm))
{
FatalErrorInFunction
<< "Cannot send outgoing message. "
<< "to:" << proci << " nBytes:"
<< label(sendBufs[proci].size()*sizeof(T))
<< Foam::abort(FatalError);
maxNSend = max(maxNSend, sendBufs[proci].size());
}
}
}
const label maxNBytes = sizeof(T)*maxNSend;
// Wait for all to finish
// ~~~~~~~~~~~~~~~~~~~~~~
// We need to send maxNBytes bytes so the number of iterations:
// maxNBytes iterations
// --------- ----------
// 0 0
// 1..maxCommsSize 1
// maxCommsSize+1..2*maxCommsSize 2
// etc.
if (block)
{
Pstream::waitRequests(startOfRequests);
label nIter;
if (maxNBytes == 0)
{
nIter = 0;
}
else
{
nIter = (maxNBytes-1)/Pstream::maxCommsSize+1;
}
reduce(nIter, maxOp<label>(), tag, comm);
List<const char*> charSendBufs(sendBufs.size());
List<char*> charRecvBufs(sendBufs.size());
labelList nRecv(sendBufs.size());
labelList startRecv(sendBufs.size(), 0);
labelList nSend(sendBufs.size());
labelList startSend(sendBufs.size(), 0);
for (label iter = 0; iter < nIter; iter++)
{
forAll(sendBufs, proci)
{
nSend[proci] = min
(
Pstream::maxCommsSize,
sendBufs[proci].size()-startSend[proci]
);
charSendBufs[proci] =
(
nSend[proci] > 0
? reinterpret_cast<const char*>
(
&(sendBufs[proci][startSend[proci]])
)
: nullptr
);
nRecv[proci] = min
(
Pstream::maxCommsSize,
recvBufs[proci].size()-startRecv[proci]
);
charRecvBufs[proci] =
(
nRecv[proci] > 0
? reinterpret_cast<char*>
(
&(recvBufs[proci][startRecv[proci]])
)
: nullptr
);
}
exchangeBuf<T>
(
nSend,
charSendBufs,
nRecv,
charRecvBufs,
tag,
comm,
block
);
forAll(nSend, proci)
{
startSend[proci] += nSend[proci];
startRecv[proci] += nRecv[proci];
}
}
}
}
......
......@@ -28,6 +28,7 @@ License
#include "dynamicCode.H"
#include "dynamicCodeContext.H"
#include "dlLibraryTable.H"
#include "Pstream.H"
#include "PstreamReduceOps.H"
#include "OSspecific.H"
#include "Ostream.H"
......
......@@ -3,7 +3,7 @@
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2011-2016 OpenFOAM Foundation
\\/ M anipulation |
\\/ M anipulation | Copyright (C) 2016 OpenCFD Ltd.
-------------------------------------------------------------------------------
License
This file is part of OpenFOAM.
......@@ -23,7 +23,7 @@ License
\*---------------------------------------------------------------------------*/
#include "UPstream.H"
#include "Pstream.H"
#include "PstreamReduceOps.H"
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
......
......@@ -3,7 +3,7 @@
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2011-2016 OpenFOAM Foundation
\\/ M anipulation |
\\/ M anipulation | Copyright (C) 2016 OpenCFD Ltd.
-------------------------------------------------------------------------------
License
This file is part of OpenFOAM.
......@@ -23,7 +23,7 @@ License
\*---------------------------------------------------------------------------*/
#include "UPstream.H"
#include "Pstream.H"
#include "PstreamReduceOps.H"
#include "OSspecific.H"
#include "PstreamGlobals.H"
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment