Skip to content
GitLab
Commit 537a916d authored by mattijs's avatar mattijs
Browse files

ENH: Pstream: expose tag

parent ff37232f
Hide whitespace changes
Inline Side-by-side
src/OpenFOAM/db/IOobjects/IOdictionary/IOdictionaryIO.C
View file @ 537a916d
......@@ -66,8 +66,13 @@ void Foam::IOdictionary::readFile(const bool masterOnly)
// Master reads headerclassname from file. Make sure this gets
// transfered as well as contents.
Pstream::scatter(comms, const_cast<word&>(headerClassName()));
Pstream::scatter(comms, note());
Pstream::scatter
(
comms,
const_cast<word&>(headerClassName()),
Pstream::msgType()
);
Pstream::scatter(comms, note(), Pstream::msgType());
// Get my communication order
const Pstream::commsStruct& myComm = comms[Pstream::myProcNo()];
......@@ -88,6 +93,7 @@ void Foam::IOdictionary::readFile(const bool masterOnly)
Pstream::scheduled,
myComm.above(),
0,
Pstream::msgType(),
IOstream::ASCII
);
IOdictionary::readData(fromAbove);
......
src/OpenFOAM/db/IOstreams/Pstreams/IPstream.C
View file @ 537a916d
......@@ -2,7 +2,7 @@
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2004-2010 OpenCFD Ltd.
\\ / A nd | Copyright (C) 2004-2011 OpenCFD Ltd.
\\/ M anipulation |
-------------------------------------------------------------------------------
License
......@@ -32,6 +32,7 @@ Foam::IPstream::IPstream
const commsTypes commsType,
const int fromProcNo,
const label bufSize,
const int tag,
streamFormat format,
versionNumber version
)
......@@ -43,7 +44,7 @@ Foam::IPstream::IPstream
fromProcNo,
buf_,
externalBufPosition_,
UPstream::msgType(), // tag
tag, // tag
false, // do not clear buf_ if at end
format,
version
......
src/OpenFOAM/db/IOstreams/Pstreams/IPstream.H
View file @ 537a916d
......@@ -2,7 +2,7 @@
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2004-2010 OpenCFD Ltd.
\\ / A nd | Copyright (C) 2004-2011 OpenCFD Ltd.
\\/ M anipulation |
-------------------------------------------------------------------------------
License
......@@ -68,6 +68,7 @@ public:
const commsTypes commsType,
const int fromProcNo,
const label bufSize = 0,
const int tag = UPstream::msgType(),
streamFormat format=BINARY,
versionNumber version=currentVersion
);
......
src/OpenFOAM/db/IOstreams/Pstreams/Pstream.H
View file @ 537a916d
......@@ -2,7 +2,7 @@
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2004-2010 OpenCFD Ltd.
\\ / A nd | Copyright (C) 2004-2011 OpenCFD Ltd.
\\/ M anipulation |
-------------------------------------------------------------------------------
License
......@@ -97,20 +97,31 @@ public:
(
const List<commsStruct>& comms,
T& Value,
const BinaryOp& bop
const BinaryOp& bop,
const int tag
);
//- Like above but switches between linear/tree communication
template <class T, class BinaryOp>
static void gather(T& Value, const BinaryOp& bop);
static void gather
(
T& Value,
const BinaryOp& bop,
const int tag = Pstream::msgType()
);
//- Scatter data. Distribute without modification. Reverse of gather
template <class T>
static void scatter(const List<commsStruct>& comms, T& Value);
static void scatter
(
const List<commsStruct>& comms,
T& Value,
const int tag
);
//- Like above but switches between linear/tree communication
template <class T>
static void scatter(T& Value);
static void scatter(T& Value, const int tag = Pstream::msgType());
// Combine variants. Inplace combine values from processors.
......@@ -121,24 +132,35 @@ public:
(
const List<commsStruct>& comms,
T& Value,
const CombineOp& cop
const CombineOp& cop,
const int tag
);
//- Like above but switches between linear/tree communication
template <class T, class CombineOp>
static void combineGather(T& Value, const CombineOp& cop);
static void combineGather
(
T& Value,
const CombineOp& cop,
const int tag = Pstream::msgType()
);
//- Scatter data. Reverse of combineGather
template <class T>
static void combineScatter
(
const List<commsStruct>& comms,
T& Value
T& Value,
const int tag
);
//- Like above but switches between linear/tree communication
template <class T>
static void combineScatter(T& Value);
static void combineScatter
(
T& Value,
const int tag = Pstream::msgType()
);
// Combine variants working on whole List at a time.
......@@ -147,24 +169,35 @@ public:
(
const List<commsStruct>& comms,
List<T>& Value,
const CombineOp& cop
const CombineOp& cop,
const int tag
);
//- Like above but switches between linear/tree communication
template <class T, class CombineOp>
static void listCombineGather(List<T>& Value, const CombineOp& cop);
static void listCombineGather
(
List<T>& Value,
const CombineOp& cop,
const int tag = Pstream::msgType()
);
//- Scatter data. Reverse of combineGather
template <class T>
static void listCombineScatter
(
const List<commsStruct>& comms,
List<T>& Value
List<T>& Value,
const int tag
);
//- Like above but switches between linear/tree communication
template <class T>
static void listCombineScatter(List<T>& Value);
static void listCombineScatter
(
List<T>& Value,
const int tag = Pstream::msgType()
);
// Combine variants working on whole map at a time. Container needs to
// have iterators and find() defined.
......@@ -174,7 +207,8 @@ public:
(
const List<commsStruct>& comms,
Container& Values,
const CombineOp& cop
const CombineOp& cop,
const int tag
);
//- Like above but switches between linear/tree communication
......@@ -182,7 +216,8 @@ public:
static void mapCombineGather
(
Container& Values,
const CombineOp& cop
const CombineOp& cop,
const int tag = Pstream::msgType()
);
//- Scatter data. Reverse of combineGather
......@@ -190,12 +225,17 @@ public:
static void mapCombineScatter
(
const List<commsStruct>& comms,
Container& Values
Container& Values,
const int tag
);
//- Like above but switches between linear/tree communication
template <class Container>
static void mapCombineScatter(Container& Values);
static void mapCombineScatter
(
Container& Values,
const int tag = Pstream::msgType()
);
......@@ -208,24 +248,34 @@ public:
static void gatherList
(
const List<commsStruct>& comms,
List<T>& Values
List<T>& Values,
const int tag
);
//- Like above but switches between linear/tree communication
template <class T>
static void gatherList(List<T>& Values);
static void gatherList
(
List<T>& Values,
const int tag = Pstream::msgType()
);
//- Scatter data. Reverse of gatherList
template <class T>
static void scatterList
(
const List<commsStruct>& comms,
List<T>& Values
List<T>& Values,
const int tag
);
//- Like above but switches between linear/tree communication
template <class T>
static void scatterList(List<T>& Values);
static void scatterList
(
List<T>& Values,
const int tag = Pstream::msgType()
);
// Exchange
......
src/OpenFOAM/db/IOstreams/Pstreams/PstreamBuffers.H
View file @ 537a916d
......@@ -2,7 +2,7 @@
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2004-2010 OpenCFD Ltd.
\\ / A nd | Copyright (C) 2004-2011 OpenCFD Ltd.
\\/ M anipulation |
-------------------------------------------------------------------------------
License
......@@ -137,6 +137,11 @@ public:
// Member functions
int tag() const
{
return tag_;
}
//- Mark all sends as having been done. This will start receives
// in non-blocking mode. If block will wait for all transfers to
// finish (only relevant for nonBlocking mode)
......
src/OpenFOAM/db/IOstreams/Pstreams/PstreamCombineReduceOps.H
View file @ 537a916d
......@@ -2,7 +2,7 @@
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2004-2010 OpenCFD Ltd.
\\ / A nd | Copyright (C) 2004-2011 OpenCFD Ltd.
\\/ M anipulation |
-------------------------------------------------------------------------------
License
......@@ -52,26 +52,38 @@ void combineReduce
(
const List<UPstream::commsStruct>& comms,
T& Value,
const CombineOp& cop
const CombineOp& cop,
const int tag
)
{
Pstream::combineGather(comms, Value, cop);
Pstream::combineScatter(comms, Value);
Pstream::combineGather(comms, Value, cop, tag);
Pstream::combineScatter(comms, Value, tag);
}
template <class T, class CombineOp>
void combineReduce(T& Value, const CombineOp& cop)
void combineReduce
(
T& Value,
const CombineOp& cop,
const int tag = Pstream::msgType()
)
{
if (UPstream::nProcs() < UPstream::nProcsSimpleSum)
{
Pstream::combineGather(UPstream::linearCommunication(), Value, cop);
Pstream::combineScatter(UPstream::linearCommunication(), Value);
Pstream::combineGather
(
UPstream::linearCommunication(),
Value,
cop,
tag
);
Pstream::combineScatter(UPstream::linearCommunication(), Value, tag);
}
else
{
Pstream::combineGather(UPstream::treeCommunication(), Value, cop);
Pstream::combineScatter(UPstream::treeCommunication(), Value);
Pstream::combineGather(UPstream::treeCommunication(), Value, cop, tag);
Pstream::combineScatter(UPstream::treeCommunication(), Value, tag);
}
}
......
src/OpenFOAM/db/IOstreams/Pstreams/PstreamReduceOps.H
View file @ 537a916d
......@@ -2,7 +2,7 @@
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2004-2010 OpenCFD Ltd.
\\ / A nd | Copyright (C) 2004-2011 OpenCFD Ltd.
\\/ M anipulation |
-------------------------------------------------------------------------------
License
......@@ -42,11 +42,12 @@ void reduce
(
const List<UPstream::commsStruct>& comms,
T& Value,
const BinaryOp& bop
const BinaryOp& bop,
const int tag
)
{
Pstream::gather(comms, Value, bop);
Pstream::scatter(comms, Value);
Pstream::gather(comms, Value, bop, tag);
Pstream::scatter(comms, Value, tag);
}
......@@ -55,16 +56,17 @@ template <class T, class BinaryOp>
void reduce
(
T& Value,
const BinaryOp& bop
const BinaryOp& bop,
const int tag = Pstream::msgType()
)
{
if (UPstream::nProcs() < UPstream::nProcsSimpleSum)
{
reduce(UPstream::linearCommunication(), Value, bop);
reduce(UPstream::linearCommunication(), Value, bop, tag);
}
else
{
reduce(UPstream::treeCommunication(), Value, bop);
reduce(UPstream::treeCommunication(), Value, bop, tag);
}
}
......@@ -74,18 +76,19 @@ template <class T, class BinaryOp>
T returnReduce
(
const T& Value,
const BinaryOp& bop
const BinaryOp& bop,
const int tag = Pstream::msgType()
)
{
T WorkValue(Value);
if (UPstream::nProcs() < UPstream::nProcsSimpleSum)
{
reduce(UPstream::linearCommunication(), WorkValue, bop);
reduce(UPstream::linearCommunication(), WorkValue, bop, tag);
}
else
{
reduce(UPstream::treeCommunication(), WorkValue, bop);
reduce(UPstream::treeCommunication(), WorkValue, bop, tag);
}
return WorkValue;
......@@ -93,7 +96,12 @@ T returnReduce
// Insist there is a specialisation for the reduction of a scalar
void reduce(scalar& Value, const sumOp<scalar>& bop);
void reduce
(
scalar& Value,
const sumOp<scalar>& bop,
const int tag = Pstream::msgType()
);
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
......
src/OpenFOAM/db/IOstreams/Pstreams/UIPstream.C
View file @ 537a916d
......@@ -84,13 +84,17 @@ Foam::UIPstream::~UIPstream()
{
if (debug)
{
Pout<< "UIPstream::~UIPstream() : clearing externalBuf_ of size "
<< externalBuf_.size() << endl;
Pout<< "UIPstream::~UIPstream() : tag:" << tag_
<< " fromProcNo:" << fromProcNo_
<< " clearing externalBuf_ of size "
<< externalBuf_.size()
<< " messageSize_:" << messageSize_ << endl;
}
externalBuf_.clearStorage();
}
}
// * * * * * * * * * * * * * * Member Functions * * * * * * * * * * * * * * //
Foam::Istream& Foam::UIPstream::read(token& t)
......
src/OpenFOAM/db/IOstreams/Pstreams/UPstream.H
View file @ 537a916d
......@@ -2,7 +2,7 @@
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2004-2010 OpenCFD Ltd.
\\ / A nd | Copyright (C) 2004-2011 OpenCFD Ltd.
\\/ M anipulation |
-------------------------------------------------------------------------------
License
......@@ -262,14 +262,23 @@ public:
// Spawns slave processes and initialises inter-communication
static bool init(int& argc, char**& argv);
//- Non-blocking comms: wait until all have finished.
static void waitRequests();
// Non-blocking comms
//- Get number of outstanding requests
static label nRequests();
//- Truncate number of outstanding requests
static void resetRequests(const label sz);
//- Wait until all requests (from start onwards) have finished.
static void waitRequests(const label start = 0);
//- Non-blocking comms: has request i finished?
static bool finishedRequest(const label i);
//- Non-blocking comms: has request i finished?
static bool finishedRequest(const label i);
//- Is this a parallel run?
static bool parRun()
static bool& parRun()
{
return parRun_;
}
......
src/OpenFOAM/db/IOstreams/Pstreams/combineGatherScatter.C
View file @ 537a916d
......@@ -2,7 +2,7 @@
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2004-2010 OpenCFD Ltd.
\\ / A nd | Copyright (C) 2004-2011 OpenCFD Ltd.
\\/ M anipulation |
-------------------------------------------------------------------------------
License
......@@ -50,7 +50,8 @@ void Pstream::combineGather
(
const List<UPstream::commsStruct>& comms,
T& Value,
const CombineOp& cop
const CombineOp& cop,
const int tag
)
{
if (UPstream::parRun())
......@@ -71,7 +72,8 @@ void Pstream::combineGather
UPstream::scheduled,
belowID,
reinterpret_cast<char*>(&value),
sizeof(T)
sizeof(T),
tag
);
if (debug & 2)
......@@ -84,7 +86,7 @@ void Pstream::combineGather
}
else
{
IPstream fromBelow(UPstream::scheduled, belowID);
IPstream fromBelow(UPstream::scheduled, belowID, 0, tag);
T value(fromBelow);
if (debug & 2)
......@@ -113,12 +115,13 @@ void Pstream::combineGather
UPstream::scheduled,
myComm.above(),
reinterpret_cast<const char*>(&Value),
sizeof(T)
sizeof(T),
tag
);
}
else
{
OPstream toAbove(UPstream::scheduled, myComm.above());
OPstream toAbove(UPstream::scheduled, myComm.above(), 0, tag);
toAbove << Value;
}
}
......@@ -127,15 +130,15 @@ void Pstream::combineGather
template <class T, class CombineOp>
void Pstream::combineGather(T& Value, const CombineOp& cop)
void Pstream::combineGather(T& Value, const CombineOp& cop, const int tag)
{
if (UPstream::nProcs() < UPstream::nProcsSimpleSum)
{
combineGather(UPstream::linearCommunication(), Value, cop);
combineGather(UPstream::linearCommunication(), Value, cop, tag);
}
else
{
combineGather(UPstream::treeCommunication(), Value, cop);
combineGather(UPstream::treeCommunication(), Value, cop, tag);
}
}
......@@ -144,7 +147,8 @@ template <class T>
void Pstream::combineScatter
(
const List<UPstream::commsStruct>& comms,
T& Value
T& Value,
const int tag
)
{
if (UPstream::parRun())
......@@ -162,12 +166,13 @@ void Pstream::combineScatter
UPstream::scheduled,
myComm.above(),
reinterpret_cast<char*>(&Value),
sizeof(T)
sizeof(T),
tag
);
}
else
{
IPstream fromAbove(UPstream::scheduled, myComm.above());
IPstream fromAbove(UPstream::scheduled, myComm.above(), 0, tag);
Value = T(fromAbove);
}
......@@ -195,12 +200,13 @@ void Pstream::combineScatter
UPstream::scheduled,
belowID,
reinterpret_cast<const char*>(&Value),
sizeof(T)
sizeof(T),
tag
);
}
else
{
OPstream toBelow(UPstream::scheduled, belowID);
OPstream toBelow(UPstream::scheduled, belowID, 0, tag);
toBelow << Value;
}
}
......@@ -209,15 +215,15 @@ void Pstream::combineScatter
template <class T>
void Pstream::combineScatter(T& Value)
void Pstream::combineScatter(T& Value, const int tag)
{
if (UPstream::nProcs() < UPstream::nProcsSimpleSum)
{
combineScatter(UPstream::linearCommunication(), Value);
combineScatter(UPstream::linearCommunication(), Value, tag);
}
else
{