Newer
Older
/*---------------------------------------------------------------------------*\
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\/ M anipulation |
-------------------------------------------------------------------------------
Copyright (C) 2015-2021 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/>.
\*---------------------------------------------------------------------------*/
#include "Pstream.H"
#include "PstreamBuffers.H"
#include "PstreamCombineReduceOps.H"
#include "flipOp.H"
// * * * * * * * * * * * * * Private Member Functions * * * * * * * * * * * //
template<class T, class CombineOp, class negateOp>
void Foam::mapDistributeBase::flipAndCombine
(
const labelUList& map,
const bool hasFlip,
const UList<T>& rhs,
const CombineOp& cop,
const negateOp& negOp,
List<T>& lhs
)
{
if (hasFlip)
{
forAll(map, i)
{
if (map[i] > 0)
{
label index = map[i]-1;
cop(lhs[index], rhs[i]);
}
else if (map[i] < 0)
{
label index = -map[i]-1;
cop(lhs[index], negOp(rhs[i]));
}
else
{
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
<< "At index " << i << " out of " << map.size()
<< " have illegal index " << map[i]
<< " for field " << rhs.size() << " with flipMap"
<< exit(FatalError);
}
}
}
else
{
forAll(map, i)
{
cop(lhs[map[i]], rhs[i]);
}
}
}
template<class T, class negateOp>
T Foam::mapDistributeBase::accessAndFlip
(
const UList<T>& fld,
const label index,
const bool hasFlip,
const negateOp& negOp
)
{
T t;
if (hasFlip)
{
if (index > 0)
{
t = fld[index-1];
}
else if (index < 0)
{
t = negOp(fld[-index-1]);
}
else
{
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
<< "Illegal index " << index
<< " into field of size " << fld.size()
<< " with face-flipping"
<< exit(FatalError);
t = fld[index];
}
}
else
{
t = fld[index];
}
return t;
}
// Distribute list.
template<class T, class negateOp>
void Foam::mapDistributeBase::distribute
(
const Pstream::commsTypes commsType,
const List<labelPair>& schedule,
const label constructSize,
const labelListList& subMap,
const bool subHasFlip,
const labelListList& constructMap,
const bool constructHasFlip,
List<T>& field,
const negateOp& negOp,
const int tag,
const label comm
const label myRank = Pstream::myProcNo(comm);
const label nProcs = Pstream::nProcs(comm);
if (!Pstream::parRun())
{
// Do only me to me.
const labelList& mySubMap = subMap[myRank];
List<T> subField(mySubMap.size());
forAll(mySubMap, i)
{
subField[i] = accessAndFlip(field, mySubMap[i], subHasFlip, negOp);
}
// Receive sub field from myself (subField)
const labelList& map = constructMap[myRank];
field.setSize(constructSize);
flipAndCombine
(
map,
constructHasFlip,
subField,
eqOp<T>(),
negOp,
field
);
return;
}
if (commsType == Pstream::commsTypes::blocking)
{
// Since buffered sending can reuse the field to collect the
// received data.
// Send sub field to neighbour
for (const int domain : Pstream::allProcs(comm))
{
const labelList& map = subMap[domain];
if (domain != myRank && map.size())
OPstream toNbr
(
Pstream::commsTypes::blocking,
domain,
0,
tag,
comm
);
List<T> subField(map.size());
forAll(subField, i)
{
subField[i] = accessAndFlip
(
field,
map[i],
subHasFlip,
negOp
);
}
toNbr << subField;
}
}
// Subset myself
const labelList& mySubMap = subMap[myRank];
List<T> subField(mySubMap.size());
forAll(mySubMap, i)
{
subField[i] = accessAndFlip(field, mySubMap[i], subHasFlip, negOp);
}
// Receive sub field from myself (subField)
const labelList& map = constructMap[myRank];
field.setSize(constructSize);
flipAndCombine
(
map,
constructHasFlip,
subField,
eqOp<T>(),
negOp,
field
);
// Receive sub field from neighbour
for (const int domain : Pstream::allProcs(comm))
{
const labelList& map = constructMap[domain];
if (domain != myRank && map.size())
IPstream fromNbr
(
Pstream::commsTypes::blocking,
domain,
0,
tag,
comm
);
List<T> subField(fromNbr);
checkReceivedSize(domain, map.size(), subField.size());
flipAndCombine
(
map,
constructHasFlip,
subField,
eqOp<T>(),
negOp,
field
);
}
}
}
else if (commsType == Pstream::commsTypes::scheduled)
{
// Need to make sure I don't overwrite field with received data
// since the data might need to be sent to another processor. So
// allocate a new field for the results.
List<T> newField(constructSize);
// Receive sub field from myself
{
const labelList& mySubMap = subMap[myRank];
List<T> subField(mySubMap.size());
forAll(subField, i)
{
subField[i] = accessAndFlip
(
field,
mySubMap[i],
subHasFlip,
negOp
);
}
// Receive sub field from myself (subField)
flipAndCombine
(
constructHasFlip,
subField,
eqOp<T>(),
negOp,
newField
);
}
// Schedule will already have pruned 0-sized comms
forAll(schedule, i)
{
const labelPair& twoProcs = schedule[i];
// twoProcs is a swap pair of processors. The first one is the
// one that needs to send first and then receive.
label sendProc = twoProcs[0];
label recvProc = twoProcs[1];
{
// I am send first, receive next
{
OPstream toNbr
(
Pstream::commsTypes::scheduled,
recvProc,
0,
const labelList& map = subMap[recvProc];
List<T> subField(map.size());
forAll(subField, i)
{
subField[i] = accessAndFlip
(
field,
map[i],
subHasFlip,
negOp
);
}
toNbr << subField;
}
{
IPstream fromNbr
(
Pstream::commsTypes::scheduled,
recvProc,
0,
List<T> subField(fromNbr);
const labelList& map = constructMap[recvProc];
checkReceivedSize(recvProc, map.size(), subField.size());
flipAndCombine
(
map,
constructHasFlip,
subField,
eqOp<T>(),
negOp,
newField
);
}
}
else
{
// I am receive first, send next
{
IPstream fromNbr
(
Pstream::commsTypes::scheduled,
sendProc,
0,
List<T> subField(fromNbr);
const labelList& map = constructMap[sendProc];
checkReceivedSize(sendProc, map.size(), subField.size());
flipAndCombine
(
map,
constructHasFlip,
subField,
eqOp<T>(),
negOp,
newField
);
}
{
OPstream toNbr
(
Pstream::commsTypes::scheduled,
sendProc,
0,
const labelList& map = subMap[sendProc];
List<T> subField(map.size());
forAll(subField, i)
{
subField[i] = accessAndFlip
(
field,
map[i],
subHasFlip,
negOp
);
}
toNbr << subField;
}
}
}
field.transfer(newField);
}
else if (commsType == Pstream::commsTypes::nonBlocking)
{
label nOutstanding = Pstream::nRequests();
PstreamBuffers pBufs(Pstream::commsTypes::nonBlocking, tag, comm);
for (const int domain : Pstream::allProcs(comm))
{
const labelList& map = subMap[domain];
if (domain != myRank && map.size())
{
// Put data into send buffer
UOPstream toDomain(domain, pBufs);
List<T> subField(map.size());
forAll(subField, i)
{
subField[i] = accessAndFlip
(
field,
map[i],
subHasFlip,
negOp
);
}
toDomain << subField;
}
}
// Start receiving. Do not block.
pBufs.finishedSends(false);
{
// Set up 'send' to myself
const labelList& mySub = subMap[myRank];
List<T> mySubField(mySub.size());
forAll(mySub, i)
{
mySubField[i] = accessAndFlip
(
field,
mySub[i],
subHasFlip,
negOp
);
}
// Combine bits. Note that can reuse field storage
field.setSize(constructSize);
// Receive sub field from myself
{
const labelList& map = constructMap[myRank];
flipAndCombine
(
map,
constructHasFlip,
mySubField,
eqOp<T>(),
negOp,
field
);
}
}
// Block ourselves, waiting only for the current comms
Pstream::waitRequests(nOutstanding);
// Consume
for (const int domain : Pstream::allProcs(comm))
{
const labelList& map = constructMap[domain];
if (domain != myRank && map.size())
{
UIPstream str(domain, pBufs);
List<T> recvField(str);
checkReceivedSize(domain, map.size(), recvField.size());
flipAndCombine
(
map,
constructHasFlip,
recvField,
eqOp<T>(),
negOp,
field
);
}
}
}
else
{
// Set up sends to neighbours
for (const int domain : Pstream::allProcs(comm))
{
const labelList& map = subMap[domain];
if (domain != myRank && map.size())
{
List<T>& subField = sendFields[domain];
subField.setSize(map.size());
forAll(map, i)
{
subField[i] = accessAndFlip
(
field,
map[i],
subHasFlip,
negOp
);
}
OPstream::write
(
Pstream::commsTypes::nonBlocking,
subField.cdata_bytes(),
);
}
}
// Set up receives from neighbours
for (const int domain : Pstream::allProcs(comm))
{
const labelList& map = constructMap[domain];
if (domain != myRank && map.size())
{
recvFields[domain].setSize(map.size());
IPstream::read
(
Pstream::commsTypes::nonBlocking,
recvFields[domain].data_bytes(),
recvFields[domain].size_bytes(),
);
}
}
// Set up 'send' to myself
{
const labelList& map = subMap[myRank];
List<T>& subField = sendFields[myRank];
subField.setSize(map.size());
forAll(map, i)
{
subField[i] = accessAndFlip
(
field,
map[i],
subHasFlip,
negOp
);
}
}
// Combine bits. Note that can reuse field storage
field.setSize(constructSize);
// Receive sub field from myself (sendFields[myRank])
const labelList& map = constructMap[myRank];
const List<T>& subField = sendFields[myRank];
flipAndCombine
(
map,
constructHasFlip,
subField,
eqOp<T>(),
negOp,
field
);
}
// Wait for all to finish
Pstream::waitRequests(nOutstanding);
// Collect neighbour fields
for (const int domain : Pstream::allProcs(comm))
{
const labelList& map = constructMap[domain];
if (domain != myRank && map.size())
{
const List<T>& subField = recvFields[domain];
checkReceivedSize(domain, map.size(), subField.size());
flipAndCombine
(
map,
constructHasFlip,
subField,
eqOp<T>(),
negOp,
field
);
}
}
}
}
else
{
<< "Unknown communication schedule " << int(commsType)
<< abort(FatalError);
}
}
// Distribute list.
template<class T, class CombineOp, class negateOp>
void Foam::mapDistributeBase::distribute
(
const Pstream::commsTypes commsType,
const List<labelPair>& schedule,
const label constructSize,
const labelListList& subMap,
const bool subHasFlip,
const labelListList& constructMap,
const bool constructHasFlip,
List<T>& field,
const CombineOp& cop,
const negateOp& negOp,
const int tag,
const label comm
const label myRank = Pstream::myProcNo(comm);
const label nProcs = Pstream::nProcs(comm);
if (!Pstream::parRun())
{
// Do only me to me.
const labelList& mySubMap = subMap[myRank];
List<T> subField(mySubMap.size());
forAll(mySubMap, i)
{
subField[i] = accessAndFlip(field, mySubMap[i], subHasFlip, negOp);
}
// Receive sub field from myself (subField)
const labelList& map = constructMap[myRank];
field.setSize(constructSize);
field = nullValue;
flipAndCombine(map, constructHasFlip, subField, cop, negOp, field);
return;
}
if (commsType == Pstream::commsTypes::blocking)
{
// Since buffered sending can reuse the field to collect the
// received data.
// Send sub field to neighbour
for (const int domain : Pstream::allProcs(comm))
{
const labelList& map = subMap[domain];
if (domain != myRank && map.size())
OPstream toNbr
(
Pstream::commsTypes::blocking,
domain,
0,
tag,
comm
);
List<T> subField(map.size());
forAll(subField, i)
{
subField[i] = accessAndFlip
(
field,
map[i],
subHasFlip,
negOp
);
}
toNbr << subField;
}
}
// Subset myself
const labelList& mySubMap = subMap[myRank];
List<T> subField(mySubMap.size());
forAll(mySubMap, i)
{
subField[i] = accessAndFlip(field, mySubMap[i], subHasFlip, negOp);
}
// Receive sub field from myself (subField)
const labelList& map = constructMap[myRank];
field.setSize(constructSize);
field = nullValue;
flipAndCombine(map, constructHasFlip, subField, cop, negOp, field);
// Receive sub field from neighbour
for (const int domain : Pstream::allProcs(comm))
{
const labelList& map = constructMap[domain];
if (domain != myRank && map.size())
IPstream fromNbr
(
Pstream::commsTypes::blocking,
domain,
0,
tag,
comm
);
List<T> subField(fromNbr);
checkReceivedSize(domain, map.size(), subField.size());
flipAndCombine
(
map,
constructHasFlip,
subField,
cop,
negOp,
field
);
}
}
}
else if (commsType == Pstream::commsTypes::scheduled)
{
// Need to make sure I don't overwrite field with received data
// since the data might need to be sent to another processor. So
// allocate a new field for the results.
List<T> newField(constructSize, nullValue);
{
const labelList& mySubMap = subMap[myRank];
// Subset myself
List<T> subField(mySubMap.size());
forAll(subField, i)
{
subField[i] = accessAndFlip
(
field,
mySubMap[i],
subHasFlip,
negOp
);
}
// Receive sub field from myself (subField)
const labelList& map = constructMap[myRank];
flipAndCombine
(
map,
constructHasFlip,
subField,
cop,
negOp,
newField
);
}
// Schedule will already have pruned 0-sized comms
forAll(schedule, i)
{
const labelPair& twoProcs = schedule[i];
// twoProcs is a swap pair of processors. The first one is the
// one that needs to send first and then receive.
label sendProc = twoProcs[0];
label recvProc = twoProcs[1];
{
// I am send first, receive next
{
OPstream toNbr
(
Pstream::commsTypes::scheduled,
recvProc,
0,
const labelList& map = subMap[recvProc];
List<T> subField(map.size());
forAll(subField, i)
{
subField[i] = accessAndFlip
(
field,
map[i],
subHasFlip,
negOp
);
}
toNbr << subField;
}
{
IPstream fromNbr
(
Pstream::commsTypes::scheduled,
recvProc,
0,
List<T> subField(fromNbr);
const labelList& map = constructMap[recvProc];
checkReceivedSize(recvProc, map.size(), subField.size());
flipAndCombine
(
map,
constructHasFlip,
subField,
cop,
negOp,
newField
);
}
}
else
{
// I am receive first, send next
{
IPstream fromNbr
(
Pstream::commsTypes::scheduled,
sendProc,
0,
List<T> subField(fromNbr);
const labelList& map = constructMap[sendProc];
checkReceivedSize(sendProc, map.size(), subField.size());
flipAndCombine
(
map,
constructHasFlip,
subField,
cop,
negOp,
newField
);
}
{
OPstream toNbr
(
Pstream::commsTypes::scheduled,
sendProc,
0,
const labelList& map = subMap[sendProc];
List<T> subField(map.size());
forAll(subField, i)
{
subField[i] = accessAndFlip
(
field,
map[i],
subHasFlip,
negOp
);
}
toNbr << subField;
}
}
}
field.transfer(newField);
}
else if (commsType == Pstream::commsTypes::nonBlocking)
{
label nOutstanding = Pstream::nRequests();
PstreamBuffers pBufs(Pstream::commsTypes::nonBlocking, tag, comm);
for (const int domain : Pstream::allProcs(comm))
{
const labelList& map = subMap[domain];
if (domain != myRank && map.size())
{
// Put data into send buffer
UOPstream toDomain(domain, pBufs);
List<T> subField(map.size());
forAll(subField, i)
{
subField[i] = accessAndFlip
(
field,
map[i],
subHasFlip,
negOp
);
}
toDomain << subField;
}
}
// Start receiving. Do not block.
pBufs.finishedSends(false);
{
// Set up 'send' to myself
const labelList& myMap = subMap[myRank];
List<T> mySubField(myMap.size());
forAll(myMap, i)
{
mySubField[i] = accessAndFlip
(
field,
myMap[i],
subHasFlip,
negOp
);
}
// Combine bits. Note that can reuse field storage
field.setSize(constructSize);
field = nullValue;
// Receive sub field from myself
{
const labelList& map = constructMap[myRank];