53 { commsTypes::buffered,
"buffered" },
54 { commsTypes::scheduled,
"scheduled" },
55 { commsTypes::nonBlocking,
"nonBlocking" },
57 { commsTypes::buffered,
"blocking" },
114 #define PrintControl(Ctrl, Name) \ 115 if (UPstream::usingTopoControl(topoControls::Ctrl)) \ 117 os << (count++ ? ' ' : '(') << Name; \ 143 void Foam::UPstream::setParRun(
const int nProcs,
const bool haveThreads)
145 parRun_ = (nProcs > 0);
146 haveThreads_ = haveThreads;
149 labelRange singleProc(1);
165 comm = newCommunicator(-1, singleProc,
false);
170 <<
"problem : comm:" << comm
176 comm = newCommunicator(-2, singleProc,
false);
181 <<
"problem : comm:" << comm
199 comm = newCommunicator(-1, labelRange(nProcs),
true);
204 <<
"problem : comm:" << comm
213 singleProc.start() = globalRanki;
214 comm = newCommunicator(-2, singleProc,
true);
219 <<
"problem : comm:" << comm
224 Pout.
prefix() =
'[' + std::to_string(globalRanki) +
"] ";
230 Perr<<
"UPstream::setParRun :" 231 <<
" nProcs:" << nProcs
232 <<
" haveThreads:" << haveThreads
238 Foam::label Foam::UPstream::getAvailableCommIndex(
const label parentIndex)
241 if (!freeComms_.empty())
244 index = freeComms_.back();
245 freeComms_.pop_back();
248 myProcNo_[index] = -1;
249 parentComm_[index] = parentIndex;
251 procIDs_[index].clear();
253 linearCommunication_[index].clear();
254 treeCommunication_[index].clear();
259 index = parentComm_.size();
261 myProcNo_.push_back(-1);
262 parentComm_.push_back(parentIndex);
264 procIDs_.emplace_back();
266 linearCommunication_.emplace_back(index);
267 treeCommunication_.emplace_back(index);
271 linearCommunication_[index].linear(
true);
272 treeCommunication_[index].linear(
false);
280 const label parentIndex,
282 const bool withComponents
285 const label index = getAvailableCommIndex(parentIndex);
289 Perr<<
"Allocate communicator [" 290 << index <<
"] from [" << parentIndex
291 <<
"] ranks : " << subRanks <<
nl 298 auto& procIds = procIDs_[index];
306 ? (parentIndex < 0 || subRanks.
contains(myProcNo_[parentIndex]))
310 procIds.resize_nocopy(subRanks.
size());
311 std::iota(procIds.begin(), procIds.end(), subRanks.
start());
321 allocateCommunicatorComponents(parentIndex, index);
330 const label parentIndex,
332 const bool withComponents
335 const label index = getAvailableCommIndex(parentIndex);
339 Perr<<
"Allocate communicator [" 340 << index <<
"] from [" << parentIndex
348 auto& procIds = procIDs_[index];
356 ? (parentIndex < 0 || subRanks.contains(myProcNo_[parentIndex]))
360 procIds.resize_nocopy(subRanks.size());
363 bool monotonicOrder =
true;
364 for (
const auto ranki : subRanks)
373 if (monotonicOrder &&
count)
375 monotonicOrder = (procIds[
count-1] < ranki);
378 procIds[
count] = ranki;
384 auto last = procIds.begin() +
count;
386 last = std::unique(procIds.begin(), last);
387 count = label(last - procIds.begin());
390 procIds.resize(
count);
400 allocateCommunicatorComponents(parentIndex, index);
409 const label parentIndex
417 <<
"Attempted to duplicate an invalid communicator: " 423 const label index = getAvailableCommIndex(parentIndex);
427 Perr<<
"Duplicate communicator [" 428 << index <<
"] from [" << parentIndex <<
"]" <<
endl;
433 myProcNo_[index] = -1;
434 procIDs_[index].clear();
438 dupCommunicatorComponents(parentIndex, index);
447 const label parentIndex,
457 <<
"Attempted to split an invalid communicator: " 463 const label index = getAvailableCommIndex(parentIndex);
467 Perr<<
"Split communicator [" 468 << index <<
"] from [" << parentIndex
469 <<
"] using colour=" << colour
470 <<
" (two_step=" << two_step <<
")" <<
endl;
475 myProcNo_[index] = -1;
476 procIDs_[index].clear();
480 splitCommunicatorComponents(parentIndex, index, colour, two_step);
487 bool Foam::UPstream::setHostCommunicators(
const int numPerNode)
498 if (
FOAM_UNLIKELY(commInterNode_ >= 0 || commLocalNode_ >= 0))
502 <<
"Node communicator(s) already created!" <<
endl 507 commInterNode_ = getAvailableCommIndex(constWorldComm_);
508 commLocalNode_ = getAvailableCommIndex(constWorldComm_);
515 if (commLocalNode_ < commInterNode_)
517 std::swap(commLocalNode_, commInterNode_);
522 Perr<<
"Allocating node communicators " 523 << commInterNode_ <<
", " << commLocalNode_
524 <<
" on parent : " << constWorldComm_ <<
nl 534 const int myNodeId = (worldRank/numPerNode);
538 DynamicList<int> nodeGroup(numPerNode);
539 DynamicList<int> nodeLeader(1+worldSize/numPerNode);
541 for (
int proci = 0; proci < worldSize; ++proci)
543 if (myNodeId == (proci/numPerNode))
545 nodeGroup.push_back(proci);
548 if ((proci % numPerNode) == 0)
551 nodeLeader.push_back(proci);
555 procIDs_[commInterNode_] = std::move(nodeLeader);
556 procIDs_[commLocalNode_] = std::move(nodeGroup);
568 const SHA1Digest myDigest(SHA1(
hostName()).digest());
570 List<SHA1Digest> digests(worldSize);
571 digests[worldRank] = myDigest;
576 digests.data_bytes(),
578 UPstream::constWorldComm_
583 DynamicList<int> nodeGroup(64);
584 DynamicList<int> nodeLeader(64);
585 DynamicList<SHA1Digest> uniqDigests(64);
587 for (
int proci = 0; proci < worldSize; ++proci)
589 const auto& dig = digests[proci];
593 nodeGroup.push_back(proci);
596 if (!uniqDigests.contains(dig))
599 uniqDigests.push_back(dig);
600 nodeLeader.push_back(proci);
604 procIDs_[commInterNode_] = std::move(nodeLeader);
605 procIDs_[commLocalNode_] = std::move(nodeGroup);
611 numNodes_ = procIDs_[commInterNode_].size();
621 allocateCommunicatorComponents(constWorldComm_, commInterNode_);
622 allocateCommunicatorComponents(constWorldComm_, commLocalNode_);
630 const label communicator,
631 const bool withComponents
635 if (communicator < 0)
642 Perr<<
"Communicators : Freeing communicator " << communicator
643 <<
" parent: " << parentComm_[communicator]
644 <<
" myProcNo: " << myProcNo_[communicator]
648 if (withComponents && parRun())
650 freeCommunicatorComponents(communicator);
653 myProcNo_[communicator] = -1;
654 parentComm_[communicator] = -1;
656 linearCommunication_[communicator].clear();
669 procID = parentRanks[procID];
682 int procID = baseProcID;
689 return parentRanks.find(procID);
696 const label currentComm,
697 const int currentProcID
710 if (linearCommunication_[communicator].empty())
712 linearCommunication_[communicator].
init(communicator,
true);
729 if (treeCommunication_[communicator].empty())
731 treeCommunication_[communicator].
init(communicator,
false);
753 comms.printGraph(
Info());
768 parRun_ && (constWorldComm_ == communicator)
769 && (nodeCommsControl_ > 0)
772 && (numNodes_ > 1) && (numNodes_ >= nodeCommsMin_)
774 && (numNodes_ < procIDs_[constWorldComm_].size())
777 && (commInterNode_ > constWorldComm_)
778 && (commLocalNode_ > constWorldComm_)
785 static std::unique_ptr<List<int>> singleton;
792 (commInterNode_ <= constWorldComm_)
793 || (commInterNode_ >= procIDs_.size())
799 singleton = std::make_unique<List<int>>();
800 auto& offsets = *singleton;
802 const auto& procs = procIDs_[commInterNode_];
807 const auto count = procs.size();
809 offsets.resize(
count+1);
828 if (singleton.empty())
847 offsets[nodei+1] - offsets[nodei]
858 bool Foam::UPstream::parRun_(
false);
860 bool Foam::UPstream::haveThreads_(
false);
862 bool Foam::UPstream::noInitialCommDup_(
false);
864 int Foam::UPstream::msgType_(1);
878 Foam::UPstream::linearCommunication_(16);
881 Foam::UPstream::treeCommunication_(16);
884 int Foam::UPstream::constWorldComm_(0);
885 int Foam::UPstream::commInterNode_(-1);
886 int Foam::UPstream::commLocalNode_(-1);
887 int Foam::UPstream::numNodes_(1);
991 "nPollProcInterfaces",
1011 class addcommsTypeToOpt
1017 addcommsTypeToOpt(
const char*
name)
1022 virtual ~addcommsTypeToOpt() =
default;
1036 addcommsTypeToOpt addcommsTypeToOpt_(
"commsType");
registerOptSwitch("fa:geometryOrder", int, faMesh::geometryOrder_)
static bool floatTransfer
Should compact transfer be used in which floats replace doubles reducing the bandwidth requirement at...
Abstract base class for registered object with I/O. Used in debug symbol registration.
prefixOSstream Perr
OSstream wrapped stderr (std::cerr) with parallel prefix.
EnumType read(Istream &is) const
Read a word from Istream and return the corresponding enumeration.
label findLower(const ListType &input, const T &val, const label start, const ComparePredicate &comp)
Binary search to find the index of the last element in a sorted list that is less than value...
static bool usingNodeComms(const int communicator)
True if node topology-aware routines have been enabled, it is running in parallel, the starting point is the world-communicator and it is not an odd corner case (ie, all processes on one node, all processes on different nodes)
static const Enum< commsTypes > commsTypeNames
Enumerated names for the communication types.
errorManipArg< error, int > exit(error &err, const int errNo=1)
static void printCommTree(int communicator, bool linear=false)
Debugging: print the communication tree.
static void writeData(Ostream &os, const Type &val)
commsTypes
Communications types.
error FatalError
Error stream (stdout output on all processes), with additional 'FOAM FATAL ERROR' header text and sta...
#define FatalErrorInFunction
Report an error message using Foam::FatalError.
Central-differencing interpolation scheme class.
static void printNodeCommsControl(Ostream &os)
Report the node-communication settings.
static void freeCommunicator(const label communicator, const bool withComponents=true)
Free a previously allocated communicator.
An interval of (signed) integers defined by a start and a size.
const int nPredefinedComm
A range or interval of labels defined by a start and a size.
An Istream is an abstract base class for all input systems (streams, files, token lists etc)...
constexpr char nl
The newline '\n' character (0x0a)
static const commsStructList & linearCommunication(int communicator)
Linear communication schedule (special case) for given communicator.
IntRange< int > rangeType
Int ranges are used for MPI ranks (processes)
static int maxCommsSize
Optional maximum message size (bytes)
bool contains(const IntType value) const noexcept
True if the (global) value is within the range.
static int nProcsSimpleSum
Number of processors to change from linear to tree communication.
static label newCommunicator(const label parent, const labelRange &subRanks, const bool withComponents=true)
Create new communicator with sub-ranks on the parent communicator.
Ostream & endl(Ostream &os)
Add newline and flush stream.
static bool & parRun() noexcept
Test if this a parallel run.
static int nodeCommsMin_
Minimum number of nodes before topology-aware routines are enabled.
#define PrintControl(Ctrl, Name)
void init(int communicator)
Reset communicator index, fill tree with empty entries.
static label procNo(const label comm, const int baseProcID)
Return processor number in communicator (given physical processor number) (= reverse of baseProcNo) ...
static constexpr int commGlobal() noexcept
Communicator for all ranks, irrespective of any local worlds.
bool broadcast(Type *values, int count, MPI_Datatype datatype, const int communicator)
static int topologyControl_
Selection of topology-aware routines as a bitmask combination of the topoControls enumerations...
static int myProcNo(const label communicator=worldComm)
Rank of this process in the communicator (starting from masterNo()). Negative if the process is not a...
static label worldComm
Communicator for all ranks. May differ from commGlobal() if local worlds are in use.
static const commsStructList & treeCommunication(int communicator)
Tree communication schedule (standard case) for given communicator.
static int tuning_NBX_
Tuning parameters for non-blocking exchange (NBX)
UList< label > labelUList
A UList of labels.
static label parent(int communicator)
The parent communicator.
void reduce(T &value, [[maybe_unused]] BinaryOp bop, [[maybe_unused]] const int tag=UPstream::msgType(), const int communicator=UPstream::worldComm)
Reduce inplace (cf. MPI Allreduce)
Useful combination of include files which define Sin, Sout and Serr and the use of IO streams general...
static int nProcsNonblockingExchange
Number of processors to change to nonBlocking consensual exchange (NBX). Ignored for zero or negative...
Functions used by OpenFOAM that are specific to POSIX compliant operating systems and need to be repl...
unsigned int count(const UList< bool > &bools, const bool val=true)
Count number of 'true' entries.
AccessType combine(const UList< T > &lists, AccessOp aop=accessOp< T >())
Combines sub-lists into a single list.
static label nProcs(const label communicator=worldComm)
Number of ranks in parallel run (for given communicator). It is 1 for serial run. ...
static int nPollProcInterfaces
Number of polling cycles in processor updates.
word name(const expressions::valueTypeCode typeCode)
A word representation of a valueTypeCode. Empty for expressions::valueTypeCode::INVALID.
static const commsStructList & whichCommunication(const int communicator, bool linear=false)
Communication schedule for all-to-master (proc 0) as linear/tree/none with switching based on UPstrea...
static void printTopoControl(Ostream &os)
Report the topology routines settings.
void sort(UList< T > &list)
Sort the list.
static constexpr int masterNo() noexcept
Relative rank for the master process - is always 0.
const string & prefix() const noexcept
The stream prefix.
static constexpr int commSelf() noexcept
Communicator within the current rank only.
#define FOAM_UNLIKELY(cond)
IntType start() const noexcept
The (inclusive) lower value of the range.
void gather(const Type *sendData, Type *recvData, int count, MPI_Datatype datatype, const int communicator, UPstream::Request *req=nullptr)
static label warnComm
Debugging: warn for use of any communicator differing from warnComm.
errorManip< error > abort(error &err)
string hostName()
Return the system's host name, as per hostname(1)
An Ostream is an abstract base class for all output systems (streams, files, token lists...
int optimisationSwitch(const char *name, const int deflt=0)
Lookup optimisation switch or add default value.
static void mpiAllGather(Type *allData, int count, const int communicator=UPstream::worldComm)
Gather/scatter identically-sized data.
int debug
Static debugging option.
OBJstream os(runTime.globalPath()/outputName)
Collection of communication structures.
defineTypeNameAndDebug(combustionModel, 0)
static const List< int > & interNode_offsets()
Processor offsets corresponding to the inter-node communicator.
Wrapper for internally indexed communicator label. Always invokes UPstream::allocateCommunicatorCompo...
static commsTypes defaultCommsType
Default commsType.
static label dupCommunicator(const label parent)
Duplicate the parent communicator.
Enum is a wrapper around a list of names/values that represent particular enumeration (or int) values...
static const rangeType & localNode_parentProcs()
The range (start/size) of the commLocalNode ranks in terms of the (const) world communicator processo...
static label nComms() noexcept
Number of currently defined communicators.
bool empty() const noexcept
True if range is empty (zero-sized)
static bool master(const label communicator=worldComm)
True if process corresponds to the master rank in the communicator.
static int numNodes() noexcept
The number of shared/host nodes in the (const) world communicator.
messageStream Info
Information stream (stdout output on master, null elsewhere)
static List< int > & procID(int communicator)
The list of ranks within a given communicator.
dictionary & optimisationSwitches()
The OptimisationSwitches sub-dictionary in the central controlDict(s).
static constexpr unsigned size_bytes() noexcept
The number of bytes in digest (20)
static label splitCommunicator(const label parent, const int colour, const bool two_step=true)
Allocate a new communicator by splitting the parent communicator on the given colour.
IntType size() const noexcept
The size of the range.
static const int mpiBufferSize
MPI buffer-size (bytes)
void addOptimisationObject(const char *name, simpleRegIOobject *obj)
Register optimisation switch read/write object.
static const List< T > & null() noexcept
Return a null List (reference to a nullObject). Behaves like an empty List.
prefixOSstream Pout
OSstream wrapped stdout (std::cout) with parallel prefix.
static int baseProcNo(label comm, int procID)
Return physical processor number (i.e. processor number in worldComm) given communicator and processo...
static int nodeCommsControl_
Use of host/node topology-aware routines.
A class representing the concept of 1 (one) that can be used to avoid manipulating objects known to b...
FlatOutput::OutputAdaptor< Container, Delimiters > flatOutput(const Container &obj, Delimiters delim)
Global flatOutput() function with specified output delimiters.