songtianlun/openfoam
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ENH: Improvements to the fileHandler and collated IO

Browse Source 
Improvements to existing functionality
--------------------------------------
  - MPI is initialised without thread support if it is not needed e.g. uncollated
  - Use native c++11 threading; avoids problem with static destruction order.
  - etc/cellModels now only read if needed.
  - etc/controlDict can now be read from the environment variable FOAM_CONTROLDICT
  - Uniform files (e.g. '0/uniform/time') are now read only once on the master only
    (with the masterUncollated or collated file handlers)
  - collated format writes to 'processorsNNN' instead of 'processors'.  The file
    format is unchanged.
  - Thread buffer and file buffer size are no longer limited to 2Gb.

The global controlDict file contains parameters for file handling.  Under some
circumstances, e.g. running in parallel on a system without NFS, the user may
need to set some parameters, e.g. fileHandler, before the global controlDict
file is read from file.  To support this, OpenFOAM now allows the global
controlDict to be read as a string set to the FOAM_CONTROLDICT environment
variable.

The FOAM_CONTROLDICT environment variable can be set to the content the global
controlDict file, e.g. from a sh/bash shell:

    export FOAM_CONTROLDICT=$(foamDictionary $FOAM_ETC/controlDict)

FOAM_CONTROLDICT can then be passed to mpirun using the -x option, e.g.:

    mpirun -np 2 -x FOAM_CONTROLDICT simpleFoam -parallel

Note that while this avoids the need for NFS to read the OpenFOAM configuration
the executable still needs to load shared libraries which must either be copied
locally or available via NFS or equivalent.

New: Multiple IO ranks
----------------------
The masterUncollated and collated fileHandlers can now use multiple ranks for
writing e.g.:

    mpirun -np 6 simpleFoam -parallel -ioRanks '(0 3)'

In this example ranks 0 ('processor0') and 3 ('processor3') now handle all the
I/O.  Rank 0 handles 0,1,2 and rank 3 handles 3,4,5.  The set of IO ranks should always
include 0 as first element and be sorted in increasing order.

The collated fileHandler uses the directory naming processorsNNN_XXX-YYY where
NNN is the total number of processors and XXX and YYY are first and last
processor in the rank, e.g. in above example the directories would be

    processors6_0-2
    processors6_3-5

and each of the collated files in these contains data of the local ranks
only. The same naming also applies when e.g. running decomposePar:

decomposePar -fileHandler collated -ioRanks '(0 3)'

New: Distributed data
---------------------

The individual root directories can be placed on different hosts with different
paths if necessary.  In the current framework it is necessary to specify the
root per slave process but this has been simplified with the option of specifying
the root per host with the -hostRoots command line option:

    mpirun -np 6 simpleFoam -parallel -ioRanks '(0 3)' \
        -hostRoots '("machineA" "/tmp/" "machineB" "/tmp")'

The hostRoots option is followed by a list of machine name + root directory, the
machine name can contain regular expressions.

New: hostCollated
-----------------

The new hostCollated fileHandler automatically sets the 'ioRanks' according to
the host name with the lowest rank e.g. to run simpleFoam on 6 processors with
ranks 0-2 on machineA and ranks 3-5 on machineB with the machines specified in
the hostfile:

    mpirun -np 6 --hostfile hostfile simpleFoam -parallel -fileHandler hostCollated

This is equivalent to

    mpirun -np 6 --hostfile hostfile simpleFoam -parallel -fileHandler collated -ioRanks '(0 3)'

This example will write directories:

    processors6_0-2/
    processors6_3-5/

A typical example would use distributed data e.g. no two nodes, machineA and
machineB, each with three processes:

    decomposePar -fileHandler collated -case cavity

    # Copy case (constant/*, system/*, processors6/) to master:
    rsync -a cavity machineA:/tmp/

    # Create root on slave:
    ssh machineB mkdir -p /tmp/cavity

    # Run
    mpirun --hostfile hostfile icoFoam \
        -case /tmp/cavity -parallel -fileHandler hostCollated \
        -hostRoots '("machineA" "/tmp" "machineB" "/tmp")'

Contributed by Mattijs Janssens
This commit is contained in:
Henry Weller 2018-03-21 12:42:22 +00:00 committed by mattijs
parent d469bbae4b
commit 8959b8e00a
37 changed files with 4093 additions and 1479 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
applications/utilities/parallelProcessing/decomposePar/decomposePar.C
View File

@ -356,6 +356,9 @@ int main(int argc, char *argv[])
)
);
// Give file handler a chance to determine the output directory
const_cast<fileOperation&>(fileHandler()).setNProcs(nDomains);
if (decomposeFieldsOnly)
{
// Sanity check on previously decomposed case
@ -395,22 +398,42 @@ int main(int argc, char *argv[])
Info<< "Removing " << nProcs
<< " existing processor directories" << endl;
fileHandler().rmDir
// Remove existing processors directory
fileNameList dirs
(
runTime.path()/word("processors"),
true // silent (may not have been collated)
);
// remove existing processor dirs
// reverse order to avoid gaps if someone interrupts the process
for (label proci = nProcs-1; proci >= 0; --proci)
{
fileName procDir
fileHandler().readDir
(
runTime.path()/(word("processor") + name(proci))
);
runTime.path(),
fileName::Type::DIRECTORY
)
);
forAllReverse(dirs, diri)
{
const fileName& d = dirs[diri];
fileHandler().rmDir(procDir);
// Starts with 'processors'
if (d.find("processors") == 0)
{
if (fileHandler().exists(d))
{
fileHandler().rmDir(d);
}
}
// Starts with 'processor'
if (d.find("processor") == 0)
{
// Check that integer after processor
fileName num(d.substr(9));
label proci = -1;
if (Foam::read(num.c_str(), proci))
{
if (fileHandler().exists(d))
{
fileHandler().rmDir(d);
}
}
}
}
procDirsProblem = false;

3
applications/utilities/parallelProcessing/reconstructPar/reconstructPar.C
View File

@ -216,6 +216,9 @@ int main(int argc, char *argv[])
<< exit(FatalError);
}
// Warn fileHandler of number of processors
const_cast<fileOperation&>(fileHandler()).setNProcs(nProcs);
// Create the processor databases
PtrList<Time> databases(nProcs);

26
applications/utilities/postProcessing/graphics/PVReaders/vtkPVFoam/vtkPVFoam.C
View File

@ -33,6 +33,7 @@ License
#include "Time.H"
#include "patchZones.H"
#include "IOobjectList.H"
#include "collatedFileOperation.H"
// VTK includes
#include "vtkDataArraySelection.h"
@ -268,7 +269,7 @@ Foam::word Foam::vtkPVFoam::getReaderPartName(const int partId) const
Foam::vtkPVFoam::vtkPVFoam
(
const char* const FileName,
const char* const vtkFileName,
vtkPVFoamReader* reader
)
:
@ -294,12 +295,19 @@ Foam::vtkPVFoam::vtkPVFoam
{
if (debug)
{
Info<< "vtkPVFoam - " << FileName << nl;
Info<< "vtkPVFoam - " << vtkFileName << nl;
printMemory();
}
fileName FileName(vtkFileName);
// Make sure not to use the threaded version - it does not like
// being loaded as a shared library - static cleanup order is problematic.
// For now just disable the threaded writer.
fileOperations::collatedFileOperation::maxThreadFileBufferSize = 0;
// avoid argList and get rootPath/caseName directly from the file
fileName fullCasePath(fileName(FileName).path());
fileName fullCasePath(FileName.path());
if (!isDir(fullCasePath))
{
@ -314,8 +322,20 @@ Foam::vtkPVFoam::vtkPVFoam
setEnv("FOAM_EXECUTABLE", "paraview", false);
// Set the case as an environment variable - some BCs might use this
if (fullCasePath.name().find("processors", 0) == 0)
{
// FileName e.g. "cavity/processors256/processor1.OpenFOAM
// Remove the processors section so it goes into processorDDD
// checking below.
fullCasePath = fullCasePath.path()/fileName(FileName.name()).lessExt();
}
if (fullCasePath.name().find("processor", 0) == 0)
{
// Give filehandler opportunity to analyse number of processors
(void)fileHandler().filePath(fullCasePath);
const fileName globalCase = fullCasePath.path();
setEnv("FOAM_CASE", globalCase, true);

146
src/OSspecific/POSIX/POSIX.C
View File

@ -489,6 +489,10 @@ mode_t Foam::mode(const fileName& name, const bool followLink)
if (POSIX::debug)
{
Pout<< FUNCTION_NAME << " : name:" << name << endl;
if ((POSIX::debug & 2) && !Pstream::master())
{
error::printStack(Pout);
}
}
// Ignore an empty name => always 0
@ -516,10 +520,6 @@ Foam::fileName::Type Foam::type(const fileName& name, const bool followLink)
if (POSIX::debug)
{
Pout<< FUNCTION_NAME << " : name:" << name << endl;
if ((POSIX::debug & 2) && !Pstream::master())
{
error::printStack(Pout);
}
}
mode_t m = mode(name, followLink);
@ -1678,142 +1678,4 @@ Foam::fileNameList Foam::dlLoaded()
}
static Foam::DynamicList<Foam::autoPtr<pthread_t>> threads_;
static Foam::DynamicList<Foam::autoPtr<pthread_mutex_t>> mutexes_;
Foam::label Foam::allocateThread()
{
forAll(threads_, i)
{
if (!threads_[i].valid())
{
if (POSIX::debug)
{
Pout<< FUNCTION_NAME << " : reusing index:" << i << endl;
}
// Reuse entry
threads_[i].reset(new pthread_t());
return i;
}
}
const label index = threads_.size();
if (POSIX::debug)
{
Pout<< FUNCTION_NAME << " : new index:" << index << endl;
}
threads_.append(autoPtr<pthread_t>(new pthread_t()));
return index;
}
void Foam::createThread
(
const label index,
void *(*start_routine) (void *),
void *arg
)
{
if (POSIX::debug)
{
Pout<< FUNCTION_NAME << " : index:" << index << endl;
}
if (pthread_create(&threads_[index](), nullptr, start_routine, arg))
{
FatalErrorInFunction
<< "Failed starting thread " << index << exit(FatalError);
}
}
void Foam::joinThread(const label index)
{
if (POSIX::debug)
{
Pout<< FUNCTION_NAME << " : index:" << index << endl;
}
if (pthread_join(threads_[index](), nullptr))
{
FatalErrorInFunction << "Failed joining thread " << index
<< exit(FatalError);
}
}
void Foam::freeThread(const label index)
{
if (POSIX::debug)
{
Pout<< FUNCTION_NAME << " : index:" << index << endl;
}
threads_[index].clear();
}
Foam::label Foam::allocateMutex()
{
forAll(mutexes_, i)
{
if (!mutexes_[i].valid())
{
if (POSIX::debug)
{
Pout<< FUNCTION_NAME << " : reusing index:" << i << endl;
}
// Reuse entry
mutexes_[i].reset(new pthread_mutex_t());
return i;
}
}
const label index = mutexes_.size();
if (POSIX::debug)
{
Pout<< FUNCTION_NAME << " : new index:" << index << endl;
}
mutexes_.append(autoPtr<pthread_mutex_t>(new pthread_mutex_t()));
return index;
}
void Foam::lockMutex(const label index)
{
if (POSIX::debug)
{
Pout<< FUNCTION_NAME << " : index:" << index << endl;
}
if (pthread_mutex_lock(&mutexes_[index]()))
{
FatalErrorInFunction << "Failed locking mutex " << index
<< exit(FatalError);
}
}
void Foam::unlockMutex(const label index)
{
if (POSIX::debug)
{
Pout<< FUNCTION_NAME << " : index:" << index << endl;
}
if (pthread_mutex_unlock(&mutexes_[index]()))
{
FatalErrorInFunction << "Failed unlocking mutex " << index
<< exit(FatalError);
}
}
void Foam::freeMutex(const label index)
{
if (POSIX::debug)
{
Pout<< FUNCTION_NAME << " : index:" << index << endl;
}
mutexes_[index].clear();
}
// ************************************************************************* //

3
src/OpenFOAM/Make/files
View File

@ -11,9 +11,11 @@ global/etcFiles/etcFiles.C
fileOps = global/fileOperations
$(fileOps)/fileOperation/fileOperation.C
$(fileOps)/fileOperationInitialise/fileOperationInitialise.C
$(fileOps)/uncollatedFileOperation/uncollatedFileOperation.C
$(fileOps)/masterUncollatedFileOperation/masterUncollatedFileOperation.C
$(fileOps)/collatedFileOperation/collatedFileOperation.C
$(fileOps)/collatedFileOperation/hostCollatedFileOperation.C
$(fileOps)/collatedFileOperation/threadedCollatedOFstream.C
$(fileOps)/collatedFileOperation/OFstreamCollator.C
@ -288,7 +290,6 @@ $(Time)/TimeIO.C
$(Time)/findTimes.C
$(Time)/subCycleTime.C
$(Time)/subLoopTime.C
$(Time)/findInstance.C
$(Time)/timeSelector.C
$(Time)/instant/instant.C

143
src/OpenFOAM/db/IOobjects/decomposedBlockData/decomposedBlockData.C
View File

@ -2,7 +2,7 @@
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2017 OpenFOAM Foundation
\\ / A nd | Copyright (C) 2017-2018 OpenFOAM Foundation
\\/ M anipulation |
-------------------------------------------------------------------------------
License
@ -681,14 +681,15 @@ void Foam::decomposedBlockData::gather
List<int> recvOffsets;
List<int> recvSizes;
if (UPstream::master())
if (UPstream::master(comm))
{
recvOffsets.setSize(nProcs);
forAll(recvOffsets, proci)
{
// Note: truncating long int to int since UPstream::gather limited
// to ints
recvOffsets[proci] =
reinterpret_cast<char*>(&datas[proci])
- data0Ptr;
int(reinterpret_cast<char*>(&datas[proci]) - data0Ptr);
}
recvSizes.setSize(nProcs, sizeof(label));
}
@ -748,7 +749,8 @@ void Foam::decomposedBlockData::gatherSlaveData
&& (UPstream::myProcNo(comm) < startProc+nProcs)
)
{
nSend = data.byteSize();
// Note: UPstream::gather limited to int
nSend = int(data.byteSize());
}
UPstream::gather
@ -764,6 +766,46 @@ void Foam::decomposedBlockData::gatherSlaveData
}
Foam::label Foam::decomposedBlockData::calcNumProcs
(
const label comm,
const off_t maxBufferSize,
const labelUList& recvSizes,
const label startProci
)
{
const label nProcs = UPstream::nProcs(comm);
label nSendProcs = -1;
if (UPstream::master(comm))
{
off_t totalSize = recvSizes[startProci];
label proci = startProci+1;
while (proci < nProcs && (totalSize+recvSizes[proci] < maxBufferSize))
{
totalSize += recvSizes[proci];
proci++;
}
nSendProcs = proci-startProci;
}
// Scatter nSendProcs
label n;
UPstream::scatter
(
reinterpret_cast<const char*>(&nSendProcs),
List<int>(nProcs, sizeof(nSendProcs)),
List<int>(nProcs, 0),
reinterpret_cast<char*>(&n),
sizeof(n),
comm
);
return n;
}
bool Foam::decomposedBlockData::writeBlocks
(
const label comm,
@ -772,8 +814,7 @@ bool Foam::decomposedBlockData::writeBlocks
const UList<char>& data,
const labelUList& recvSizes,
const bool haveSlaveData,
const List<char>& slaveData,
const PtrList<SubList<char>>& slaveData,
const UPstream::commsTypes commsType,
const bool syncReturnState
@ -784,17 +825,15 @@ bool Foam::decomposedBlockData::writeBlocks
Pout<< "decomposedBlockData::writeBlocks:"
<< " stream:" << (osPtr.valid() ? osPtr().name() : "invalid")
<< " data:" << data.size()
<< " haveSlaveData:" << haveSlaveData
<< " (master only) slaveData:" << slaveData.size()
<< " commsType:" << Pstream::commsTypeNames[commsType] << endl;
}
const label nProcs = UPstream::nProcs(comm);
bool ok = true;
if (haveSlaveData)
if (slaveData.size())
{
// Already have gathered the slave data. communicator only used to
// check who is the master
@ -821,8 +860,7 @@ bool Foam::decomposedBlockData::writeBlocks
os << nl << nl << "// Processor" << proci << nl;
start[proci] = os.stdStream().tellp();
os << SubList<char>(slaveData, recvSizes[proci], slaveOffset);
os << slaveData[proci];
slaveOffset += recvSizes[proci];
}
@ -897,44 +935,24 @@ bool Foam::decomposedBlockData::writeBlocks
// maxMasterFileBufferSize
// Starting slave processor and number of processors
labelPair startAndSize(1, nProcs-1);
label startProc = 1;
label nSendProcs = nProcs-1;
while (startAndSize[1] > 0)
while (nSendProcs > 0)
{
labelPair masterData(startAndSize);
if (UPstream::master(comm))
{
label totalSize = recvSizes[masterData[0]];
label proci = masterData[0]+1;
while
(
proci < nProcs
&& (
totalSize+recvSizes[proci]
< fileOperations::masterUncollatedFileOperation::
maxMasterFileBufferSize
)
)
{
totalSize += recvSizes[proci];
++proci;
}
masterData[1] = proci-masterData[0];
}
// Scatter masterData
UPstream::scatter
nSendProcs = calcNumProcs
(
reinterpret_cast<const char*>(masterData.cdata()),
List<int>(nProcs, sizeof(masterData)),
List<int>(nProcs, 0),
reinterpret_cast<char*>(startAndSize.data()),
sizeof(startAndSize),
comm
comm,
off_t
(
fileOperations::masterUncollatedFileOperation::
maxMasterFileBufferSize
),
recvSizes,
startProc
);
if (startAndSize[0] == nProcs || startAndSize[1] == 0)
if (startProc == nProcs || nSendProcs == 0)
{
break;
}
@ -949,8 +967,8 @@ bool Foam::decomposedBlockData::writeBlocks
data,
recvSizes,
startAndSize[0], // startProc,
startAndSize[1], // nProcs,
startProc, // startProc,
nSendProcs, // nProcs,
sliceOffsets,
recvData
@ -963,9 +981,9 @@ bool Foam::decomposedBlockData::writeBlocks
// Write slaves
for
(
label proci = startAndSize[0];
proci < startAndSize[0]+startAndSize[1];
++proci
label proci = startProc;
proci < startProc+nSendProcs;
proci++
)
{
os << nl << nl << "// Processor" << proci << nl;
@ -981,7 +999,7 @@ bool Foam::decomposedBlockData::writeBlocks
}
}
startAndSize[0] += startAndSize[1];
startProc += nSendProcs;
}
if (UPstream::master(comm))
@ -1027,7 +1045,7 @@ bool Foam::decomposedBlockData::writeData(Ostream& os) const
);
IOobject io(*this);
if (Pstream::master())
if (Pstream::master(comm_))
{
IStringStream is
(
@ -1043,7 +1061,7 @@ bool Foam::decomposedBlockData::writeData(Ostream& os) const
// version
string versionString(os.version().str());
Pstream::scatter(versionString);
Pstream::scatter(versionString, Pstream::msgType(), comm_);
// stream
string formatString;
@ -1051,21 +1069,21 @@ bool Foam::decomposedBlockData::writeData(Ostream& os) const
OStringStream os;
os << os.format();
formatString = os.str();
Pstream::scatter(formatString);
Pstream::scatter(formatString, Pstream::msgType(), comm_);
}
//word masterName(name());
//Pstream::scatter(masterName);
//Pstream::scatter(masterName, Pstream::msgType(), comm_);
Pstream::scatter(io.headerClassName());
Pstream::scatter(io.note());
Pstream::scatter(io.headerClassName(), Pstream::msgType(), comm_);
Pstream::scatter(io.note(), Pstream::msgType(), comm_);
//Pstream::scatter(io.instance(), Pstream::msgType(), comm);
//Pstream::scatter(io.local(), Pstream::msgType(), comm);
fileName masterLocation(instance()/db().dbDir()/local());
Pstream::scatter(masterLocation);
Pstream::scatter(masterLocation, Pstream::msgType(), comm_);
if (!Pstream::master())
if (!Pstream::master(comm_))
{
writeHeader
(
@ -1081,7 +1099,7 @@ bool Foam::decomposedBlockData::writeData(Ostream& os) const
os.writeQuoted(str, false);
if (!Pstream::master())
if (!Pstream::master(comm_))
{
IOobject::writeEndDivider(os);
}
@ -1108,10 +1126,10 @@ bool Foam::decomposedBlockData::writeObject
}
labelList recvSizes;
gather(comm_, this->byteSize(), recvSizes);
gather(comm_, label(this->byteSize()), recvSizes);
List<std::streamoff> start;
List<char> slaveData; // dummy already received slave data
PtrList<SubList<char>> slaveData; // dummy slave data
return writeBlocks
(
comm_,
@ -1119,7 +1137,6 @@ bool Foam::decomposedBlockData::writeObject
start,
*this,
recvSizes,
false, // don't have slave data
slaveData,
commsType_
);

18
src/OpenFOAM/db/IOobjects/decomposedBlockData/decomposedBlockData.H
View File

@ -2,7 +2,7 @@
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2017 OpenFOAM Foundation
\\ / A nd | Copyright (C) 2017-2018 OpenFOAM Foundation
\\/ M anipulation |
-------------------------------------------------------------------------------
License
@ -66,6 +66,16 @@ protected:
// Protected member functions
//- Helper: determine number of processors whose recvSizes fits
// ito maxBufferSize
static label calcNumProcs
(
const label comm,
const off_t maxBufferSize,
const labelUList& recvSizes,
const label startProci
);
//- Read data into *this. ISstream is only valid on master.
static bool readBlocks
(
@ -202,12 +212,12 @@ public:
const label comm,
autoPtr<OSstream>& osPtr,
List<std::streamoff>& start,
const UList<char>&,
const UList<char>& masterData,
const labelUList& recvSizes,
const bool haveSlaveData, // does master have slaveData
const List<char>& slaveData, // optional slave data (on master)
// optional slave data (on master)
const PtrList<SubList<char>>& slaveData,
const UPstream::commsTypes,
const bool syncReturnState = true

6
src/OpenFOAM/db/IOstreams/Pstreams/UPstream.H
View File

@ -372,10 +372,8 @@ public:
static void addValidParOptions(HashTable<string>& validParOptions);
//- Initialisation function called from main
// Spawns slave processes and initialises inter-communication.
// \note warns if MPI has already been initialized.
// Fatal if MPI has already been finalized.
static bool init(int& argc, char**& argv);
// Spawns slave processes and initialises inter-communication
static bool init(int& argc, char**& argv, const bool needsThread);
//- Special purpose initialisation function.
// Performs a basic MPI_Init without any other setup.

30
src/OpenFOAM/db/Time/Time.C
View File

@ -714,6 +714,36 @@ Foam::instantList Foam::Time::times() const
}
Foam::word Foam::Time::findInstance
(
const fileName& dir,
const word& name,
const IOobject::readOption rOpt,
const word& stopInstance
) const
{
IOobject startIO
(
name, // name might be empty!
timeName(),
dir,
*this,
rOpt
);
IOobject io
(
fileHandler().findInstance
(
startIO,
timeOutputValue(),
stopInstance
)
);
return io.instance();
}
Foam::word Foam::Time::findInstancePath
(
const fileName& directory,

3
src/OpenFOAM/db/Time/Time.H
View File

@ -196,9 +196,6 @@ protected:
//- Read the control dictionary and set the write controls etc.
virtual void readDict();
//- Find IOobject in the objectPath
static bool exists(IOobject& io);
private:

236
src/OpenFOAM/db/Time/findInstance.C
View File

@ -1,236 +0,0 @@
/*---------------------------------------------------------------------------*\
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2011-2017 OpenFOAM Foundation
\\/ M anipulation |
-------------------------------------------------------------------------------
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/>.
Description
If "name" is empty: return the location of "directory"
If "name" is not empty: return the location of "directory" containing the
file "name".
Used in reading mesh data.
\*---------------------------------------------------------------------------*/
#include "Time.H"
#include "IOobject.H"
#include "IOList.H"
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
bool Foam::Time::exists(IOobject& io)
{
// Generate filename for object
fileName objPath(fileHandler().objectPath(io, word::null));
// Test for either directory or a (valid) file & IOobject
bool ok;
if (io.name().empty())
{
ok = fileHandler().isDir(objPath);
}
else
{
ok =
fileHandler().isFile(objPath)
&& io.typeHeaderOk<IOList<label>>(false);// object with local scope
}
if (!ok)
{
// Re-test with raw objectPath. This is for backwards
// compatibility
fileName originalPath(io.objectPath());
if (originalPath != objPath)
{
// Test for either directory or a (valid) file & IOobject
if (io.name().empty())
{
ok = fileHandler().isDir(originalPath);
}
else
{
ok =
fileHandler().isFile(originalPath)
&& io.typeHeaderOk<IOList<label>>(false);
}
}
}
return ok;
}
Foam::word Foam::Time::findInstance
(
const fileName& dir,
const word& name,
const IOobject::readOption rOpt,
const word& stopInstance
) const
{
// Note: - if name is empty, just check the directory itself
// - check both for isFile and headerOk since the latter does a
// filePath so searches for the file.
// - check for an object with local file scope (so no looking up in
// parent directory in case of parallel)
{
IOobject io
(
name, // name might be empty!
timeName(),
dir,
*this
);
if (exists(io))
{
if (debug)
{
InfoInFunction
<< "Found exact match for \"" << name
<< "\" in " << timeName()/dir
<< endl;
}
return timeName();
}
}
// Search back through the time directories to find the time
// closest to and lower than current time
instantList ts = times();
label instanceI;
for (instanceI = ts.size()-1; instanceI >= 0; --instanceI)
{
if (ts[instanceI].value() <= timeOutputValue())
{
break;
}
}
// continue searching from here
for (; instanceI >= 0; --instanceI)
{
IOobject io
(
name, // name might be empty!
ts[instanceI].name(),
dir,
*this
);
if (exists(io))
{
if (debug)
{
InfoInFunction
<< "Found instance match for \"" << name
<< "\" in " << ts[instanceI].name()/dir
<< endl;
}
return ts[instanceI].name();
}
// Check if hit minimum instance
if (ts[instanceI].name() == stopInstance)
{
if (debug)
{
//InfoInFunction
Pout<< "findInstance : "
<< "Hit stopInstance " << stopInstance
<< endl;
}
if
(
rOpt == IOobject::MUST_READ
|| rOpt == IOobject::MUST_READ_IF_MODIFIED
)
{
if (name.empty())
{
FatalErrorInFunction
<< "Cannot find directory "
<< dir << " in times " << timeName()
<< " down to " << stopInstance
<< exit(FatalError);
}
else
{
FatalErrorInFunction
<< "Cannot find file \"" << name << "\" in directory "
<< dir << " in times " << timeName()
<< " down to " << stopInstance
<< exit(FatalError);
}
}
return ts[instanceI].name();
}
}
// not in any of the time directories, try constant
// Note. This needs to be a hard-coded constant, rather than the
// constant function of the time, because the latter points to
// the case constant directory in parallel cases
IOobject io
(
name,
constant(),
dir,
*this
);
if (exists(io))
{
if (debug)
{
InfoInFunction
<< "Found constant match for \"" << name
<< "\" in " << constant()/dir
<< endl;
}
return constant();
}
if (rOpt == IOobject::MUST_READ || rOpt == IOobject::MUST_READ_IF_MODIFIED)
{
FatalErrorInFunction
<< "Cannot find file \"" << name << "\" in directory "
<< dir << " in times " << timeName()
<< " down to " << constant()
<< exit(FatalError);
}
return constant();
}
// ************************************************************************* //

5
src/OpenFOAM/db/regIOobject/regIOobjectRead.C
View File

@ -2,7 +2,7 @@
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2011-2016 OpenFOAM Foundation
\\ / A nd | Copyright (C) 2011-2018 OpenFOAM Foundation
\\/ M anipulation | Copyright (C) 2015-2016 OpenCFD Ltd.
-------------------------------------------------------------------------------
License
@ -183,7 +183,8 @@ void Foam::regIOobject::close()
if (IFstream::debug)
{
Pout<< "regIOobject::close() : "
<< "finished reading " << isPtr_().name()
<< "finished reading "
<< (isPtr_.valid() ? isPtr_().name() : "dummy")
<< endl;
}

190
src/OpenFOAM/global/argList/argList.C
View File

@ -40,8 +40,9 @@ License
#include "foamVersion.H"
#include "stringOps.H"
#include "CStringList.H"
#include "uncollatedFileOperation.H"
#include "masterUncollatedFileOperation.H"
#include "stringListOps.H"
#include "fileOperation.H"
#include "fileOperationInitialise.H"
#include "IOmanip.H"
#include <cctype>
@ -83,6 +84,12 @@ Foam::argList::initValidTables::initValidTables()
"decomposeParDict", "file",
"read decomposePar dictionary from specified location"
);
argList::addOption
(
"hostRoots", "(((host1 dir1) .. (hostN dirN))",
"slave root directories (per host) for distributed running"
);
validParOptions.set("hostRoots", "((host1 dir1) .. (hostN dirN))");
argList::addBoolOption
(
@ -324,6 +331,7 @@ void Foam::argList::noParallel()
removeOption("parallel");
removeOption("roots");
removeOption("decomposeParDict");
removeOption("hostRoots");
validParOptions.clear();
}
@ -687,6 +695,34 @@ Foam::argList::argList
options_(argc),
distributed_(false)
{
// Check for fileHandler
word handlerType(getEnv("FOAM_FILEHANDLER"));
for (int argI = 0; argI < argc; ++argI)
{
if (argv[argI][0] == '-')
{
const char *optionName = &argv[argI][1];
if (string(optionName) == "fileHandler")
{
handlerType = argv[argI+1];
break;
}
}
}
if (handlerType.empty())
{
handlerType = fileOperation::defaultFileHandler;
}
// Detect any parallel options
bool needsThread = fileOperations::fileOperationInitialise::New
(
handlerType,
argc,
argv
)().needsThreading();
// Check if this run is a parallel run by searching for any parallel option
// If found call runPar which might filter argv
for (int argi = 1; argi < argc; ++argi)
@ -697,7 +733,7 @@ Foam::argList::argList
if (validParOptions.found(optName))
{
parRunControl_.runPar(argc, argv);
parRunControl_.runPar(argc, argv, needsThread);
break;
}
}
@ -942,6 +978,58 @@ void Foam::argList::parse
Foam::fileHandler(handler);
}
stringList slaveProcs;
stringList slaveMachine;
const int writeHostsSwitch = debug::infoSwitch("writeHosts", 1);
// Collect slave machine/pid, and check that the build is identical
if (parRunControl_.parRun())
{
if (Pstream::master())
{
slaveProcs.setSize(Pstream::nProcs() - 1);
slaveMachine.setSize(Pstream::nProcs() - 1);
label proci = 0;
for
(
int slave = Pstream::firstSlave();
slave <= Pstream::lastSlave();
slave++
)
{
IPstream fromSlave(Pstream::commsTypes::scheduled, slave);
string slaveBuild;
label slavePid;
fromSlave >> slaveBuild >> slaveMachine[proci] >> slavePid;
slaveProcs[proci] = slaveMachine[proci] + "." + name(slavePid);
proci++;
// Check build string to make sure all processors are running
// the same build
if (slaveBuild != Foam::FOAMbuild)
{
FatalErrorIn(executable())
<< "Master is running version " << Foam::FOAMbuild
<< "; slave " << proci << " is running version "
<< slaveBuild
<< exit(FatalError);
}
}
}
else
{
OPstream toMaster
(
Pstream::commsTypes::scheduled,
Pstream::masterNo()
);
toMaster << string(Foam::FOAMbuild) << hostName() << pid();
}
}
// Case is a single processor run unless it is running parallel
int nProcs = 1;
@ -999,6 +1087,52 @@ void Foam::argList::parse
dictNProcs = roots.size()+1;
}
}
else if (options_.found("hostRoots"))
{
source = "-hostRoots";
IStringStream is(options_["hostRoots"]);
List<Tuple2<wordRe, fileName>> hostRoots(is);
roots.setSize(Pstream::nProcs()-1);
forAll(hostRoots, i)
{
const Tuple2<wordRe, fileName>& hostRoot = hostRoots[i];
const wordRe& re = hostRoot.first();
labelList matchedRoots(findStrings(re, slaveMachine));
forAll(matchedRoots, matchi)
{
label slavei = matchedRoots[matchi];
if (roots[slavei] != wordRe())
{
FatalErrorInFunction
<< "Slave " << slaveMachine[slavei]
<< " has multiple matching roots in "
<< hostRoots << exit(FatalError);
}
else
{
roots[slavei] = hostRoot.second();
}
}
}
// Check
forAll(roots, slavei)
{
if (roots[slavei] == wordRe())
{
FatalErrorInFunction
<< "Slave " << slaveMachine[slavei]
<< " has no matching roots in "
<< hostRoots << exit(FatalError);
}
}
if (roots.size() != 1)
{
dictNProcs = roots.size()+1;
}
}
else if (checkProcessorDirectories_)
{
// Use values from decomposeParDict, the location was already
@ -1172,55 +1306,6 @@ void Foam::argList::parse
case_ = globalCase_;
}
stringList slaveProcs;
const int writeHostsSwitch = debug::infoSwitch("writeHosts", 1);
// Collect slave machine/pid, and check that the build is identical
if (parRunControl_.parRun())
{
if (Pstream::master())
{
slaveProcs.setSize(Pstream::nProcs() - 1);
label proci = 0;
for
(
int slave = Pstream::firstSlave();
slave <= Pstream::lastSlave();
slave++
)
{
IPstream fromSlave(Pstream::commsTypes::scheduled, slave);
string slaveBuild;
string slaveMachine;
label slavePid;
fromSlave >> slaveBuild >> slaveMachine >> slavePid;
slaveProcs[proci++] = slaveMachine + "." + name(slavePid);
// Check build string to make sure all processors are running
// the same build
if (slaveBuild != Foam::FOAMbuild)
{
FatalErrorIn(executable())
<< "Master is running version " << Foam::FOAMbuild
<< "; slave " << proci << " is running version "
<< slaveBuild
<< exit(FatalError);
}
}
}
else
{
OPstream toMaster
(
Pstream::commsTypes::scheduled,
Pstream::masterNo()
);
toMaster << string(Foam::FOAMbuild) << hostName() << pid();
}
}
// Keep or discard slave and root information for reporting:
if (Pstream::master() && parRunControl_.parRun())
{
@ -1412,6 +1497,7 @@ bool Foam::argList::unsetOption(const word& optName)
optName == "case"
|| optName == "parallel"
|| optName == "roots"
|| optName == "hostRoots"
)
{
FatalErrorInFunction

6
src/OpenFOAM/global/argList/parRun.H
View File

@ -2,7 +2,7 @@
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2011 OpenFOAM Foundation
\\ / A nd | Copyright (C) 2011-2018 OpenFOAM Foundation
\\/ M anipulation |
-------------------------------------------------------------------------------
License
@ -70,11 +70,11 @@ public:
}
//- Initialize Pstream for a parallel run
void runPar(int& argc, char**& argv)
void runPar(int& argc, char**& argv, const bool needsThread)
{
RunPar = true;
if (!Pstream::init(argc, argv))
if (!Pstream::init(argc, argv, needsThread))
{
Info<< "Failed to start parallel run" << endl;
Pstream::exit(1);

36
src/OpenFOAM/global/debug/debug.C
View File

@ -2,7 +2,7 @@
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2011-2017 OpenFOAM Foundation
\\ / A nd | Copyright (C) 2011-2018 OpenFOAM Foundation
\\/ M anipulation |
-------------------------------------------------------------------------------
License
@ -95,21 +95,31 @@ Foam::dictionary& Foam::debug::controlDict()
{
if (!controlDictPtr_)
{
fileNameList controlDictFiles = findEtcFiles("controlDict", true);
controlDictPtr_ = new dictionary();
forAllReverse(controlDictFiles, cdfi)
string controlDictString(getEnv("FOAM_CONTROLDICT"));
if (!controlDictString.empty())
{
IFstream ifs(controlDictFiles[cdfi]);
if (!ifs.good())
// Read from environment
IStringStream is(controlDictString);
controlDictPtr_ = new dictionary(is);
}
else
{
fileNameList controlDictFiles = findEtcFiles("controlDict", true);
controlDictPtr_ = new dictionary();
forAllReverse(controlDictFiles, cdfi)
{
SafeFatalIOErrorInFunction
(
ifs,
"Cannot open controlDict"
);
IFstream ifs(controlDictFiles[cdfi]);
if (!ifs.good())
{
SafeFatalIOErrorInFunction
(
ifs,
"Cannot open controlDict"
);
}
controlDictPtr_->merge(dictionary(ifs));
}
controlDictPtr_->merge(dictionary(ifs));
}
}

372
src/OpenFOAM/global/fileOperations/collatedFileOperation/OFstreamCollator.C
View File

@ -2,7 +2,7 @@
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2017 OpenFOAM Foundation
\\ / A nd | Copyright (C) 2017-2018 OpenFOAM Foundation
\\/ M anipulation |
-------------------------------------------------------------------------------
License
@ -26,6 +26,7 @@ License
#include "OFstreamCollator.H"
#include "OFstream.H"
#include "decomposedBlockData.H"
#include "masterUncollatedFileOperation.H"
// * * * * * * * * * * * * * * Static Data Members * * * * * * * * * * * * * //
@ -44,8 +45,7 @@ bool Foam::OFstreamCollator::writeFile
const fileName& fName,
const string& masterData,
const labelUList& recvSizes,
const bool haveSlaveData, // does master have slaveData
const UList<char>& slaveData, // on master: slave data
const PtrList<SubList<char>>& slaveData, // optional slave data
IOstream::streamFormat fmt,
IOstream::versionNumber ver,
IOstream::compressionType cmp,
@ -54,9 +54,22 @@ bool Foam::OFstreamCollator::writeFile
{
if (debug)
{
Pout<< "OFstreamCollator : Writing " << masterData.size()
Pout<< "OFstreamCollator : Writing master " << masterData.size()
<< " bytes to " << fName
<< " using comm " << comm << endl;
if (slaveData.size())
{
Pout<< "OFstreamCollator : Slave data" << endl;
forAll(slaveData, proci)
{
if (slaveData.set(proci))
{
Pout<< " " << proci
<< " size:" << slaveData[proci].size()
<< endl;
}
}
}
}
autoPtr<OSstream> osPtr;
@ -76,17 +89,20 @@ bool Foam::OFstreamCollator::writeFile
);
// We don't have IOobject so cannot use IOobject::writeHeader
OSstream& os = osPtr();
decomposedBlockData::writeHeader
(
os,
ver,
fmt,
typeName,
"",
fName,
fName.name()
);
if (!append)
{
OSstream& os = osPtr();
decomposedBlockData::writeHeader
(
os,
ver,
fmt,
typeName,
"",
fName,
fName.name()
);
}
}
@ -109,9 +125,13 @@ bool Foam::OFstreamCollator::writeFile
start,
slice,
recvSizes,
haveSlaveData,
slaveData,
UPstream::commsTypes::nonBlocking, //scheduled,
(
fileOperations::masterUncollatedFileOperation::
maxMasterFileBufferSize == 0
? UPstream::commsTypes::scheduled
: UPstream::commsTypes::nonBlocking
),
false // do not reduce return state
);
@ -132,7 +152,11 @@ bool Foam::OFstreamCollator::writeFile
{
sum += recvSizes[i];
}
Pout<< " (overall " << sum << ")";
// Use ostringstream to display long int (until writing these is
// supported)
std::ostringstream os;
os << sum;
Pout<< " (overall " << os.str() << ")";
}
Pout<< " to " << fName
<< " using comm " << comm << endl;
@ -151,12 +175,13 @@ void* Foam::OFstreamCollator::writeAll(void *threadarg)
{
writeData* ptr = nullptr;
lockMutex(handler.mutex_);
if (handler.objects_.size())
{
ptr = handler.objects_.pop();
std::lock_guard<std::mutex> guard(handler.mutex_);
if (handler.objects_.size())
{
ptr = handler.objects_.pop();
}
}
unlockMutex(handler.mutex_);
if (!ptr)
{
@ -164,6 +189,28 @@ void* Foam::OFstreamCollator::writeAll(void *threadarg)
}
else
{
// Convert storage to pointers
PtrList<SubList<char>> slaveData;
if (ptr->slaveData_.size())
{
slaveData.setSize(ptr->slaveData_.size());
forAll(slaveData, proci)
{
if (ptr->slaveData_.set(proci))
{
slaveData.set
(
proci,
new SubList<char>
(
ptr->slaveData_[proci],
ptr->sizes_[proci]
)
);
}
}
}
bool ok = writeFile
(
ptr->comm_,
@ -171,9 +218,7 @@ void* Foam::OFstreamCollator::writeAll(void *threadarg)
ptr->pathName_,
ptr->data_,
ptr->sizes_,
ptr->haveSlaveData_,
ptr->slaveData_,
slaveData,
ptr->format_,
ptr->version_,
ptr->compression_,
@ -196,9 +241,10 @@ void* Foam::OFstreamCollator::writeAll(void *threadarg)
Pout<< "OFstreamCollator : Exiting write thread " << endl;
}
lockMutex(handler.mutex_);
handler.threadRunning_ = false;
unlockMutex(handler.mutex_);
{
std::lock_guard<std::mutex> guard(handler.mutex_);
handler.threadRunning_ = false;
}
return nullptr;
}
@ -211,12 +257,13 @@ void Foam::OFstreamCollator::waitForBufferSpace(const off_t wantedSize) const
// Count files to be written
off_t totalSize = 0;
lockMutex(mutex_);
forAllConstIter(FIFOStack<writeData*>, objects_, iter)
{
totalSize += iter()->size();
std::lock_guard<std::mutex> guard(mutex_);
forAllConstIter(FIFOStack<writeData*>, objects_, iter)
{
totalSize += iter()->size();
}
}
unlockMutex(mutex_);
if (totalSize == 0 || (totalSize+wantedSize) <= maxBufferSize_)
{
@ -225,13 +272,12 @@ void Foam::OFstreamCollator::waitForBufferSpace(const off_t wantedSize) const
if (debug)
{
lockMutex(mutex_);
std::lock_guard<std::mutex> guard(mutex_);
Pout<< "OFstreamCollator : Waiting for buffer space."
<< " Currently in use:" << totalSize
<< " limit:" << maxBufferSize_
<< " files:" << objects_.size()
<< endl;
unlockMutex(mutex_);
}
sleep(5);
@ -244,25 +290,34 @@ void Foam::OFstreamCollator::waitForBufferSpace(const off_t wantedSize) const
Foam::OFstreamCollator::OFstreamCollator(const off_t maxBufferSize)
:
maxBufferSize_(maxBufferSize),
mutex_
(
maxBufferSize_ > 0
? allocateMutex()
: -1
),
thread_
(
maxBufferSize_ > 0
? allocateThread()
: -1
),
threadRunning_(false),
comm_
localComm_(UPstream::worldComm),
threadComm_
(
UPstream::allocateCommunicator
(
UPstream::worldComm,
identity(UPstream::nProcs(UPstream::worldComm))
localComm_,
identity(UPstream::nProcs(localComm_))
)
)
{}
Foam::OFstreamCollator::OFstreamCollator
(
const off_t maxBufferSize,
const label comm
)
:
maxBufferSize_(maxBufferSize),
threadRunning_(false),
localComm_(comm),
threadComm_
(
UPstream::allocateCommunicator
(
localComm_,
identity(UPstream::nProcs(localComm_))
)
)
{}
@ -272,26 +327,19 @@ Foam::OFstreamCollator::OFstreamCollator(const off_t maxBufferSize)
Foam::OFstreamCollator::~OFstreamCollator()
{
if (threadRunning_)
if (thread_.valid())
{
if (debug)
{
Pout<< "~OFstreamCollator : Waiting for write thread" << endl;
}
thread_().join();
thread_.clear();
}
joinThread(thread_);
}
if (thread_ != -1)
if (threadComm_ != -1)
{
freeThread(thread_);
}
if (mutex_ != -1)
{
freeMutex(mutex_);
}
if (comm_ != -1)
{
UPstream::freeCommunicator(comm_);
UPstream::freeCommunicator(threadComm_);
}
}
@ -312,7 +360,8 @@ bool Foam::OFstreamCollator::write
// Determine (on master) sizes to receive. Note: do NOT use thread
// communicator
labelList recvSizes;
decomposedBlockData::gather(Pstream::worldComm, data.size(), recvSizes);
decomposedBlockData::gather(localComm_, label(data.size()), recvSizes);
off_t totalSize = 0;
label maxLocalSize = 0;
{
@ -321,8 +370,8 @@ bool Foam::OFstreamCollator::write
totalSize += recvSizes[proci];
maxLocalSize = max(maxLocalSize, recvSizes[proci]);
}
Pstream::scatter(totalSize, Pstream::msgType(), Pstream::worldComm);
Pstream::scatter(maxLocalSize, Pstream::msgType(), Pstream::worldComm);
Pstream::scatter(totalSize, Pstream::msgType(), localComm_);
Pstream::scatter(maxLocalSize, Pstream::msgType(), localComm_);
}
if (maxBufferSize_ == 0 || maxLocalSize > maxBufferSize_)
@ -330,18 +379,17 @@ bool Foam::OFstreamCollator::write
if (debug)
{
Pout<< "OFstreamCollator : non-thread gather and write of " << fName
<< " using worldComm" << endl;
<< " using local comm " << localComm_ << endl;
}
// Direct collating and writing (so master blocks until all written!)
const List<char> dummySlaveData;
const PtrList<SubList<char>> dummySlaveData;
return writeFile
(
UPstream::worldComm,
localComm_,
typeName,
fName,
data,
recvSizes,
false, // no slave data provided yet
dummySlaveData,
fmt,
ver,
@ -360,22 +408,28 @@ bool Foam::OFstreamCollator::write
<< fName << endl;
}
if (Pstream::master())
if (Pstream::master(localComm_))
{
waitForBufferSpace(totalSize);
}
// Allocate local buffer for all collated data
// Receive in chunks of labelMax (2^31-1) since this is the maximum
// size that a List can be
autoPtr<writeData> fileAndDataPtr
(
new writeData
(
comm_, // Note: comm not actually used anymore
threadComm_, // Note: comm not actually used anymore
typeName,
fName,
data,
(
Pstream::master(localComm_)
? data // Only used on master
: string::null
),
recvSizes,
true, // have slave data (collected below)
fmt,
ver,
cmp,
@ -384,40 +438,84 @@ bool Foam::OFstreamCollator::write
);
writeData& fileAndData = fileAndDataPtr();
// Gather the slave data and insert into fileAndData
PtrList<List<char>>& slaveData = fileAndData.slaveData_;
UList<char> slice(const_cast<char*>(data.data()), label(data.size()));
List<int> slaveOffsets;
decomposedBlockData::gatherSlaveData
(
Pstream::worldComm, // Note: using simulation thread
slice,
recvSizes,
1, // startProc,
Pstream::nProcs()-1, // n procs
slaveData.setSize(recvSizes.size());
slaveOffsets,
fileAndData.slaveData_
);
// Append to thread buffer
lockMutex(mutex_);
objects_.push(fileAndDataPtr.ptr());
unlockMutex(mutex_);
// Start thread if not running
lockMutex(mutex_);
if (!threadRunning_)
// Gather all data onto master. Is done in local communicator since
// not in write thread. Note that we do not store in contiguous
// buffer since that would limit to 2G chars.
label startOfRequests = Pstream::nRequests();
if (Pstream::master(localComm_))
{
createThread(thread_, writeAll, this);
if (debug)
for (label proci = 1; proci < slaveData.size(); proci++)
{
Pout<< "OFstreamCollator : Started write thread "
<< thread_ << endl;
slaveData.set(proci, new List<char>(recvSizes[proci]));
UIPstream::read
(
UPstream::commsTypes::nonBlocking,
proci,
reinterpret_cast<char*>(slaveData[proci].begin()),
slaveData[proci].byteSize(),
Pstream::msgType(),
localComm_
);
}
}
else
{
if
(
!UOPstream::write
(
UPstream::commsTypes::nonBlocking,
0,
reinterpret_cast<const char*>(slice.begin()),
slice.byteSize(),
Pstream::msgType(),
localComm_
)
)
{
FatalErrorInFunction
<< "Cannot send outgoing message. "
<< "to:" << 0 << " nBytes:"
<< label(slice.byteSize())
<< Foam::abort(FatalError);
}
}
Pstream::waitRequests(startOfRequests);
{
std::lock_guard<std::mutex> guard(mutex_);
// Append to thread buffer
objects_.push(fileAndDataPtr.ptr());
// Start thread if not running
if (!threadRunning_)
{
if (thread_.valid())
{
if (debug)
{
Pout<< "OFstreamCollator : Waiting for write thread"
<< endl;
}
thread_().join();
}
if (debug)
{
Pout<< "OFstreamCollator : Starting write thread"
<< endl;
}
thread_.reset(new std::thread(writeAll, this));
threadRunning_ = true;
}
threadRunning_ = true;
}
unlockMutex(mutex_);
return true;
}
@ -426,57 +524,65 @@ bool Foam::OFstreamCollator::write
if (debug)
{
Pout<< "OFstreamCollator : thread gather and write of " << fName
<< " in thread " << thread_
<< " using communicator " << comm_ << endl;
<< " using communicator " << threadComm_ << endl;
}
if (!UPstream::haveThreads())
{
FatalErrorInFunction
<< "mpi does not seem to have thread support."
<< "Please increase the buffer size 'maxThreadFileBufferSize'"
<< " Make sure to set buffer size 'maxThreadFileBufferSize'"
<< " to at least " << totalSize
<< " to be able to do the collating before threading."
<< exit(FatalError);
}
if (Pstream::master())
if (Pstream::master(localComm_))
{
waitForBufferSpace(data.size());
}
lockMutex(mutex_);
// Push all file info on buffer. Note that no slave data provided
// so it will trigger communication inside the thread
objects_.push
(
new writeData
(
comm_,
typeName,
fName,
data,
recvSizes,
false, // Have no slave data; collect in thread
fmt,
ver,
cmp,
append
)
);
unlockMutex(mutex_);
lockMutex(mutex_);
if (!threadRunning_)
{
createThread(thread_, writeAll, this);
if (debug)
std::lock_guard<std::mutex> guard(mutex_);
// Push all file info on buffer. Note that no slave data provided
// so it will trigger communication inside the thread
objects_.push
(
new writeData
(
threadComm_,
typeName,
fName,
data,
recvSizes,
fmt,
ver,
cmp,
append
)
);
if (!threadRunning_)
{
Pout<< "OFstreamCollator : Started write thread " << endl;
if (thread_.valid())
{
if (debug)
{
Pout<< "OFstreamCollator : Waiting for write thread"
<< endl;
}
thread_().join();
}
if (debug)
{
Pout<< "OFstreamCollator : Starting write thread" << endl;
}
thread_.reset(new std::thread(writeAll, this));
threadRunning_ = true;
}
threadRunning_ = true;
}
unlockMutex(mutex_);
return true;
}

45
src/OpenFOAM/global/fileOperations/collatedFileOperation/OFstreamCollator.H
View File

@ -2,7 +2,7 @@
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2017 OpenFOAM Foundation
\\ / A nd | Copyright (C) 2017-2018 OpenFOAM Foundation
\\/ M anipulation |
-------------------------------------------------------------------------------
License
@ -51,9 +51,12 @@ SourceFiles
#ifndef OFstreamCollator_H
#define OFstreamCollator_H
#include <thread>
#include <mutex>
#include "IOstream.H"
#include "labelList.H"
#include "FIFOStack.H"
#include "SubList.H"
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
@ -77,10 +80,7 @@ class OFstreamCollator
const fileName pathName_;
const string data_;
const labelList sizes_;
const bool haveSlaveData_;
List<char> slaveData_;
PtrList<List<char>> slaveData_;
const IOstream::streamFormat format_;
const IOstream::versionNumber version_;
const IOstream::compressionType compression_;
@ -93,7 +93,6 @@ class OFstreamCollator
const fileName& pathName,
const string& data,
const labelList& sizes,
const bool haveSlaveData,
IOstream::streamFormat format,
IOstream::versionNumber version,
IOstream::compressionType compression,
@ -105,7 +104,6 @@ class OFstreamCollator
pathName_(pathName),
data_(data),
sizes_(sizes),
haveSlaveData_(haveSlaveData),
slaveData_(0),
format_(format),
version_(version),
@ -116,27 +114,39 @@ class OFstreamCollator
//- (approximate) size of master + any optional slave data
off_t size() const
{
return data_.size() + slaveData_.size();
off_t sz = data_.size();
forAll(slaveData_, i)
{
if (slaveData_.set(i))
{
sz += slaveData_[i].size();
}
}
return sz;
}
};
// Private data
//- Total amount of storage to use for object stack below
const off_t maxBufferSize_;
//pthread_mutex_t mutex_;
label mutex_;
mutable std::mutex mutex_;
//pthread_t thread_;
label thread_;
autoPtr<std::thread> thread_;
//- Stack of files to write + contents
FIFOStack<writeData*> objects_;
//- Whether thread is running (and not exited)
bool threadRunning_;
//- Communicator to use for all parallel ops
label comm_;
//- Communicator to use for all parallel ops (in simulation thread)
label localComm_;
//- Communicator to use for all parallel ops (in write thread)
label threadComm_;
// Private Member Functions
@ -149,8 +159,7 @@ class OFstreamCollator
const fileName& fName,
const string& masterData,
const labelUList& recvSizes,
const bool haveSlaveData, // (does master) have slave data
const UList<char>& slaveData, // (on master) all slave data
const PtrList<SubList<char>>& slaveData,
IOstream::streamFormat fmt,
IOstream::versionNumber ver,
IOstream::compressionType cmp,
@ -176,6 +185,10 @@ public:
//- Construct from buffer size. 0 = do not use thread
OFstreamCollator(const off_t maxBufferSize);
//- Construct from buffer size (0 = do not use thread) and local
// thread
OFstreamCollator(const off_t maxBufferSize, const label comm);
//- Destructor
virtual ~OFstreamCollator();

339
src/OpenFOAM/global/fileOperations/collatedFileOperation/collatedFileOperation.C
View File

@ -2,7 +2,7 @@
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2017 OpenFOAM Foundation
\\ / A nd | Copyright (C) 2017-2018 OpenFOAM Foundation
\\/ M anipulation |
-------------------------------------------------------------------------------
License
@ -57,12 +57,63 @@ namespace fileOperations
float,
collatedFileOperation::maxThreadFileBufferSize
);
// Mark as needing threaded mpi
addNamedToRunTimeSelectionTable
(
fileOperationInitialise,
collatedFileOperationInitialise,
word,
collated
);
}
}
// * * * * * * * * * * * * * Private Member Functions * * * * * * * * * * * //
Foam::labelList Foam::fileOperations::collatedFileOperation::ioRanks()
{
labelList ioRanks;
string ioRanksString(getEnv("FOAM_IORANKS"));
if (!ioRanksString.empty())
{
IStringStream is(ioRanksString);
is >> ioRanks;
}
return ioRanks;
}
bool Foam::fileOperations::collatedFileOperation::isMasterRank
(
const label proci
)
const
{
if (Pstream::parRun())
{
return Pstream::master(comm_);
}
else
{
// Use any IO ranks
if (ioRanks_.size())
{
// Find myself in IO rank
return findIndex(ioRanks_, proci) != -1;
}
else
{
// Assume all in single communicator
return proci == 0;
}
}
}
bool Foam::fileOperations::collatedFileOperation::appendObject
(
const regIOobject& io,
@ -74,14 +125,12 @@ bool Foam::fileOperations::collatedFileOperation::appendObject
{
// Append to processors/ file
fileName prefix;
fileName postfix;
label proci = splitProcessorPath(io.objectPath(), prefix, postfix);
label proci = detectProcessorPath(io.objectPath());
if (debug)
{
Pout<< "writeObject:" << " : For local object : "
<< io.name()
Pout<< "collatedFileOperation::writeObject :"
<< " For local object : " << io.name()
<< " appending processor " << proci
<< " data to " << pathName << endl;
}
@ -93,12 +142,35 @@ bool Foam::fileOperations::collatedFileOperation::appendObject
<< exit(FatalError);
}
const bool isMaster = isMasterRank(proci);
// Determine the local rank if the pathName is a per-rank one
label localProci = proci;
{
fileName path, procDir, local;
label groupStart, groupSize, nProcs;
splitProcessorPath
(
pathName,
path,
procDir,
local,
groupStart,
groupSize,
nProcs
);
if (groupSize > 0 && groupStart != -1)
{
localProci = proci-groupStart;
}
}
// Create string from all data to write
string buf;
{
OStringStream os(fmt, ver);
if (proci == 0)
if (isMaster)
{
if (!io.writeHeader(os))
{
@ -112,7 +184,7 @@ bool Foam::fileOperations::collatedFileOperation::appendObject
return false;
}
if (proci == 0)
if (isMaster)
{
IOobject::writeEndDivider(os);
}
@ -121,8 +193,6 @@ bool Foam::fileOperations::collatedFileOperation::appendObject
}
bool append = (proci > 0);
// Note: cannot do append + compression. This is a limitation
// of ogzstream (or rather most compressed formats)
@ -132,7 +202,7 @@ bool Foam::fileOperations::collatedFileOperation::appendObject
IOstream::BINARY,
ver,
IOstream::UNCOMPRESSED, // no compression
append
!isMaster
);
if (!os.good())
@ -142,7 +212,7 @@ bool Foam::fileOperations::collatedFileOperation::appendObject
<< exit(FatalIOError);
}
if (proci == 0)
if (isMaster)
{
IOobject::writeBanner(os)
<< "FoamFile\n{\n"
@ -162,7 +232,7 @@ bool Foam::fileOperations::collatedFileOperation::appendObject
const_cast<char*>(buf.data()),
label(buf.size())
);
os << nl << "// Processor" << proci << nl << slice << nl;
os << nl << "// Processor" << localProci << nl << slice << nl;
return os.good();
}
@ -175,8 +245,23 @@ Foam::fileOperations::collatedFileOperation::collatedFileOperation
const bool verbose
)
:
masterUncollatedFileOperation(false),
writer_(maxThreadFileBufferSize)
masterUncollatedFileOperation
(
(
ioRanks().size()
? UPstream::allocateCommunicator
(
UPstream::worldComm,
subRanks(Pstream::nProcs())
)
: UPstream::worldComm
),
false
),
myComm_(comm_),
writer_(maxThreadFileBufferSize, comm_),
nProcs_(Pstream::nProcs()),
ioRanks_(ioRanks())
{
if (verbose)
{
@ -195,12 +280,97 @@ Foam::fileOperations::collatedFileOperation::collatedFileOperation
{
Info<< " Threading activated "
"since maxThreadFileBufferSize > 0." << nl
<< " Requires thread support enabled in MPI, "
"otherwise the simulation" << nl
<< " may \"hang\". If thread support cannot be "
"enabled, deactivate threading" << nl
<< " by setting maxThreadFileBufferSize to 0 in "
"the OpenFOAM etc/controlDict"
<< " Requires large enough buffer to collect all data"
" or thread support " << nl
<< " enabled in MPI. If thread support cannot be "
"enabled, deactivate" << nl
<< " threading by setting maxThreadFileBufferSize "
"to 0 in" << nl
<< " $FOAM_ETC/controlDict"
<< endl;
}
if (ioRanks_.size())
{
// Print a bit of information
stringList ioRanks(Pstream::nProcs());
if (Pstream::master(comm_))
{
ioRanks[Pstream::myProcNo()] = hostName()+"."+name(pid());
}
Pstream::gatherList(ioRanks);
Info<< " IO nodes:" << endl;
forAll(ioRanks, proci)
{
if (!ioRanks[proci].empty())
{
Info<< " " << ioRanks[proci] << endl;
}
}
}
if
(
regIOobject::fileModificationChecking
== regIOobject::inotifyMaster
)
{
WarningInFunction
<< "Resetting fileModificationChecking to inotify" << endl;
}
if
(
regIOobject::fileModificationChecking
== regIOobject::timeStampMaster
)
{
WarningInFunction
<< "Resetting fileModificationChecking to timeStamp" << endl;
}
}
}
Foam::fileOperations::collatedFileOperation::collatedFileOperation
(
const label comm,
const labelList& ioRanks,
const word& typeName,
const bool verbose
)
:
masterUncollatedFileOperation(comm, false),
myComm_(-1),
writer_(maxThreadFileBufferSize, comm),
nProcs_(Pstream::nProcs()),
ioRanks_(ioRanks)
{
if (verbose)
{
Info<< "I/O : " << typeName
<< " (maxThreadFileBufferSize " << maxThreadFileBufferSize
<< ')' << endl;
if (maxThreadFileBufferSize == 0)
{
Info<< " Threading not activated "
"since maxThreadFileBufferSize = 0." << nl
<< " Writing may run slowly for large file sizes."
<< endl;
}
else
{
Info<< " Threading activated "
"since maxThreadFileBufferSize > 0." << nl
<< " Requires large enough buffer to collect all data"
" or thread support " << nl
<< " enabled in MPI. If thread support cannot be "
"enabled, deactivate" << nl
<< " threading by setting maxThreadFileBufferSize "
"to 0 in the OpenFOAM etc/controlDict" << nl
<< endl;
}
@ -227,6 +397,17 @@ Foam::fileOperations::collatedFileOperation::collatedFileOperation
}
// * * * * * * * * * * * * * * * * Destructor * * * * * * * * * * * * * * * //
Foam::fileOperations::collatedFileOperation::~collatedFileOperation()
{
if (myComm_ != -1 && myComm_ != UPstream::worldComm)
{
UPstream::freeCommunicator(myComm_);
}
}
// * * * * * * * * * * * * * * * Member Functions * * * * * * * * * * * * * //
Foam::fileName Foam::fileOperations::collatedFileOperation::objectPath
@ -238,19 +419,21 @@ Foam::fileName Foam::fileOperations::collatedFileOperation::objectPath
// Replacement for objectPath
if (io.time().processorCase())
{
return masterUncollatedFileOperation::objectPath
return masterUncollatedFileOperation::localObjectPath
(
io,
fileOperation::PROCESSORSOBJECT,
fileOperation::PROCOBJECT,
"dummy", // not used for processorsobject
io.instance()
);
}
else
{
return masterUncollatedFileOperation::objectPath
return masterUncollatedFileOperation::localObjectPath
(
io,
fileOperation::OBJECT,
word::null,
io.instance()
);
}
@ -276,8 +459,8 @@ bool Foam::fileOperations::collatedFileOperation::writeObject
if (debug)
{
Pout<< "writeObject:"
<< " : For object : " << io.name()
Pout<< "collatedFileOperation::writeObject :"
<< " For object : " << io.name()
<< " falling back to master-only output to " << io.path()
<< endl;
}
@ -313,7 +496,7 @@ bool Foam::fileOperations::collatedFileOperation::writeObject
else
{
// Construct the equivalent processors/ directory
fileName path(processorsPath(io, inst));
fileName path(processorsPath(io, inst, processorsDir(io)));
mkDir(path);
fileName pathName(path/io.name());
@ -322,7 +505,8 @@ bool Foam::fileOperations::collatedFileOperation::writeObject
{
if (debug)
{
Pout<< "writeObject:" << " : For global object : " << io.name()
Pout<< "collatedFileOperation::writeObject :"
<< " For global object : " << io.name()
<< " falling back to master-only output to " << pathName
<< endl;
}
@ -359,11 +543,11 @@ bool Foam::fileOperations::collatedFileOperation::writeObject
else if (!Pstream::parRun())
{
// Special path for e.g. decomposePar. Append to
// processors/ file
// processorsDDD/ file
if (debug)
{
Pout<< "writeObject:"
<< " : For object : " << io.name()
Pout<< "collatedFileOperation::writeObject :"
<< " For object : " << io.name()
<< " appending to " << pathName << endl;
}
@ -373,8 +557,8 @@ bool Foam::fileOperations::collatedFileOperation::writeObject
{
if (debug)
{
Pout<< "writeObject:"
<< " : For object : " << io.name()
Pout<< "collatedFileOperation::writeObject :"
<< " For object : " << io.name()
<< " starting collating output to " << pathName << endl;
}
@ -386,7 +570,7 @@ bool Foam::fileOperations::collatedFileOperation::writeObject
{
return false;
}
if (Pstream::master() && !io.writeHeader(os))
if (Pstream::master(comm_) && !io.writeHeader(os))
{
return false;
}
@ -395,7 +579,7 @@ bool Foam::fileOperations::collatedFileOperation::writeObject
{
return false;
}
if (Pstream::master())
if (Pstream::master(comm_))
{
IOobject::writeEndDivider(os);
}
@ -406,4 +590,89 @@ bool Foam::fileOperations::collatedFileOperation::writeObject
}
Foam::word Foam::fileOperations::collatedFileOperation::processorsDir
(
const fileName& fName
) const
{
if (Pstream::parRun())
{
const List<int>& procs(UPstream::procID(comm_));
word procDir(processorsBaseDir+Foam::name(Pstream::nProcs()));
if (procs.size() != Pstream::nProcs())
{
procDir +=
+ "_"
+ Foam::name(procs[0])
+ "-"
+ Foam::name(procs.last());
}
return procDir;
}
else
{
word procDir(processorsBaseDir+Foam::name(nProcs_));
if (ioRanks_.size())
{
// Detect current processor number
label proci = detectProcessorPath(fName);
if (proci != -1)
{
// Find lowest io rank
label minProc = 0;
label maxProc = nProcs_-1;
forAll(ioRanks_, i)
{
if (ioRanks_[i] >= nProcs_)
{
break;
}
else if (ioRanks_[i] <= proci)
{
minProc = ioRanks_[i];
}
else
{
maxProc = ioRanks_[i]-1;
break;
}
}
procDir +=
+ "_"
+ Foam::name(minProc)
+ "-"
+ Foam::name(maxProc);
}
}
return procDir;
}
}
Foam::word Foam::fileOperations::collatedFileOperation::processorsDir
(
const IOobject& io
) const
{
return processorsDir(io.objectPath());
}
void Foam::fileOperations::collatedFileOperation::setNProcs(const label nProcs)
{
nProcs_ = nProcs;
if (debug)
{
Pout<< "collatedFileOperation::setNProcs :"
<< " Setting number of processors to " << nProcs_ << endl;
}
}
// ************************************************************************* //

81
src/OpenFOAM/global/fileOperations/collatedFileOperation/collatedFileOperation.H
View File

@ -43,6 +43,7 @@ SourceFiles
#include "masterUncollatedFileOperation.H"
#include "OFstreamCollator.H"
#include "fileOperationInitialise.H"
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
@ -59,14 +60,33 @@ class collatedFileOperation
:
public masterUncollatedFileOperation
{
// Private data
protected:
// Protected data
//- Any communicator allocated by me
const label myComm_;
//- Threaded writer
mutable OFstreamCollator writer_;
// For non-parallel operation
//- Number of processors (overall)
label nProcs_;
//- Ranks of IO handlers
const labelList ioRanks_;
// Private Member Functions
static labelList ioRanks();
//- Is proci master of communicator (in parallel) or master of
// the io ranks (non-parallel)
bool isMasterRank(const label proci) const;
//- Append to processors/ file
bool appendObject
(
@ -97,9 +117,18 @@ public:
//- Construct null
collatedFileOperation(const bool verbose);
//- Construct from user communicator
collatedFileOperation
(
const label comm,
const labelList& ioRanks,
const word& typeName,
const bool verbose
);
//- Destructor
virtual ~collatedFileOperation() = default;
virtual ~collatedFileOperation();
// Member Functions
@ -123,6 +152,54 @@ public:
IOstream::compressionType compression=IOstream::UNCOMPRESSED,
const bool valid = true
) const;
// Other
//- Actual name of processors dir
virtual word processorsDir(const IOobject&) const;
//- Actual name of processors dir
virtual word processorsDir(const fileName&) const;
//- Set number of processor directories/results. Only used in
// decomposePar
virtual void setNProcs(const label nProcs);
};
/*---------------------------------------------------------------------------*\
Class collatedFileOperationInitialise Declaration
\*---------------------------------------------------------------------------*/
class collatedFileOperationInitialise
:
public masterUncollatedFileOperationInitialise
{
public:
// Constructors
//- Construct from components
collatedFileOperationInitialise(int& argc, char**& argv)
:
masterUncollatedFileOperationInitialise(argc, argv)
{}
//- Destructor
virtual ~collatedFileOperationInitialise()
{}
// Member Functions
//- Needs threading
virtual bool needsThreading() const
{
return
collatedFileOperation::maxThreadFileBufferSize
> 0;
}
};

176
src/OpenFOAM/global/fileOperations/collatedFileOperation/hostCollatedFileOperation.C Normal file
View File

@ -0,0 +1,176 @@
/*---------------------------------------------------------------------------*\
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2017-2018 OpenFOAM Foundation
\\/ M anipulation |
-------------------------------------------------------------------------------
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 "hostCollatedFileOperation.H"
#include "addToRunTimeSelectionTable.H"
#include "bitSet.H"
/* * * * * * * * * * * * * * * Static Member Data * * * * * * * * * * * * * */
namespace Foam
{
namespace fileOperations
{
defineTypeNameAndDebug(hostCollatedFileOperation, 0);
addToRunTimeSelectionTable
(
fileOperation,
hostCollatedFileOperation,
word
);
// Register initialisation routine. Signals need for threaded mpi and
// handles command line arguments
addNamedToRunTimeSelectionTable
(
fileOperationInitialise,
hostCollatedFileOperationInitialise,
word,
hostCollated
);
}
}
// * * * * * * * * * * * * * Private Member Functions * * * * * * * * * * * //
Foam::labelList Foam::fileOperations::hostCollatedFileOperation::subRanks
(
const label n
)
{
DynamicList<label> subRanks(64);
string ioRanksString(getEnv("FOAM_IORANKS"));
if (!ioRanksString.empty())
{
IStringStream is(ioRanksString);
labelList ioRanks(is);
if (findIndex(ioRanks, 0) == -1)
{
FatalErrorInFunction
<< "Rank 0 (master) should be in the IO ranks. Currently "
<< ioRanks << exit(FatalError);
}
// The lowest numbered rank is the IO rank
const bitSet isIOrank(n, ioRanks);
for (label proci = Pstream::myProcNo(); proci >= 0; --proci)
{
if (isIOrank[proci])
{
// Found my master. Collect all processors with same master
subRanks.append(proci);
for
(
label rank = proci+1;
rank < n && !isIOrank[rank];
++rank
)
{
subRanks.append(rank);
}
break;
}
}
}
else
{
// Normal operation: one lowest rank per hostname is the writer
const string myHostName(hostName());
stringList hosts(Pstream::nProcs());
hosts[Pstream::myProcNo()] = myHostName;
Pstream::gatherList(hosts);
Pstream::scatterList(hosts);
// Collect procs with same hostname
forAll(hosts, proci)
{
if (hosts[proci] == myHostName)
{
subRanks.append(proci);
}
}
}
return subRanks;
}
// * * * * * * * * * * * * * * * * Constructors * * * * * * * * * * * * * * //
Foam::fileOperations::hostCollatedFileOperation::hostCollatedFileOperation
(
const bool verbose
)
:
collatedFileOperation
(
UPstream::allocateCommunicator
(
UPstream::worldComm,
subRanks(Pstream::nProcs())
),
(Pstream::parRun() ? labelList(0) : ioRanks()), // processor dirs
typeName,
verbose
)
{
if (verbose)
{
// Print a bit of information
stringList ioRanks(Pstream::nProcs());
if (Pstream::master(comm_))
{
ioRanks[Pstream::myProcNo()] = hostName()+"."+name(pid());
}
Pstream::gatherList(ioRanks);
Info<< " IO nodes:" << endl;
forAll(ioRanks, proci)
{
if (!ioRanks[proci].empty())
{
Info<< " " << ioRanks[proci] << endl;
}
}
}
}
// * * * * * * * * * * * * * * * * Destructor * * * * * * * * * * * * * * * //
Foam::fileOperations::hostCollatedFileOperation::~hostCollatedFileOperation()
{
if (comm_ != -1 && comm_ != UPstream::worldComm)
{
UPstream::freeCommunicator(comm_);
}
}
// ************************************************************************* //

134
src/OpenFOAM/global/fileOperations/collatedFileOperation/hostCollatedFileOperation.H Normal file
View File

@ -0,0 +1,134 @@
/*---------------------------------------------------------------------------*\
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2017-2018 OpenFOAM Foundation
\\/ M anipulation |
-------------------------------------------------------------------------------
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/>.
Class
Foam::fileOperations::hostCollatedFileOperation
Description
Version of collatedFileOperation with multiple read/write ranks.
In parallel it will assume ranks are sorted according to hostname
and the lowest rank per hostname will be the IO rank. The output directories
will get a unique name processors<N>_<low>-<high> where N is the overall
number of processors and low and high is the range of ranks contained
in the files. Each of these subsets uses its own communicator.
Instead of using the hostnames the IO ranks can be assigned using the
FOAM_IORANKS environment variable (also when running non-parallel), e.g.
when decomposing into 4:
FOAM_IORANKS='(0 2)' decomposePar -fileHandler hostCollated
will generate
processors4_0-1/
containing data for processors 0 to 1
processors4_2-3/
containing data for processors 2 to 3
See also
collatedFileOperation
SourceFiles
hostCollatedFileOperation.C
\*---------------------------------------------------------------------------*/
#ifndef fileOperations_hostCollatedFileOperation_H
#define fileOperations_hostCollatedFileOperation_H
#include "collatedFileOperation.H"
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
namespace Foam
{
namespace fileOperations
{
/*---------------------------------------------------------------------------*\
Class hostCollatedFileOperation Declaration
\*---------------------------------------------------------------------------*/
class hostCollatedFileOperation
:
public collatedFileOperation
{
// Private Member Functions
//- Get the list of processors part of this set
static labelList subRanks(const label n);
public:
//- Runtime type information
TypeName("hostCollated");
// Constructors
//- Construct null
hostCollatedFileOperation(const bool verbose);
//- Destructor
virtual ~hostCollatedFileOperation();
};
/*---------------------------------------------------------------------------*\
Class hostCollatedFileOperationInitialise Declaration
\*---------------------------------------------------------------------------*/
class hostCollatedFileOperationInitialise
:
public collatedFileOperationInitialise
{
public:
// Constructors
//- Construct from components
hostCollatedFileOperationInitialise(int& argc, char**& argv)
:
collatedFileOperationInitialise(argc, argv)
{}
//- Destructor
virtual ~hostCollatedFileOperationInitialise()
{}
};
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
} // End namespace fileOperations
} // End namespace Foam
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
#endif
// ************************************************************************* //

866
src/OpenFOAM/global/fileOperations/fileOperation/fileOperation.C
View File

@ -2,7 +2,7 @@
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2017 OpenFOAM Foundation
\\ / A nd | Copyright (C) 2017-2018 OpenFOAM Foundation
\\/ M anipulation |
-------------------------------------------------------------------------------
License
@ -28,11 +28,11 @@ License
#include "regIOobject.H"
#include "argList.H"
#include "HashSet.H"
#include "masterUncollatedFileOperation.H"
#include "objectRegistry.H"
#include "decomposedBlockData.H"
#include "polyMesh.H"
#include "registerSwitch.H"
#include "Time.H"
/* * * * * * * * * * * * * * * Static Member Data * * * * * * * * * * * * * */
@ -43,6 +43,28 @@ namespace Foam
defineTypeNameAndDebug(fileOperation, 0);
defineRunTimeSelectionTable(fileOperation, word);
template<>
const char* Foam::NamedEnum
<
fileOperation::pathType,
12
>::names[] =
{
"notFound",
"absolute",
"objectPath",
"writeObject",
"uncollatedProc",
"globalProc",
"localProc",
"parentObjectPath",
"findInstance",
"uncollatedProcInstance",
"globalProcInstance",
"localProcInstance"
};
const NamedEnum<fileOperation::pathType, 12> fileOperation::pathTypeNames_;
word fileOperation::defaultFileHandler
(
debug::optimisationSwitches().lookupOrAddDefault
@ -56,7 +78,7 @@ namespace Foam
);
}
Foam::word Foam::fileOperation::processorsDir = "processors";
Foam::word Foam::fileOperation::processorsBaseDir = "processors";
// * * * * * * * * * * * * * Private Member Functions * * * * * * * * * * * //
@ -133,6 +155,80 @@ Foam::instantList Foam::fileOperation::sortTimes
}
void Foam::fileOperation::mergeTimes
(
const instantList& extraTimes,
const word& constantName,
instantList& times
)
{
if (extraTimes.size())
{
bool haveConstant =
(
times.size() > 0
&& times[0].name() == constantName
);
bool haveExtraConstant =
(
extraTimes.size() > 0
&& extraTimes[0].name() == constantName
);
// Combine times
instantList combinedTimes(times.size()+extraTimes.size());
label sz = 0;
label extrai = 0;
if (haveExtraConstant)
{
extrai = 1;
if (!haveConstant)
{
combinedTimes[sz++] = extraTimes[0]; // constant
}
}
forAll(times, i)
{
combinedTimes[sz++] = times[i];
}
for (; extrai < extraTimes.size(); extrai++)
{
combinedTimes[sz++] = extraTimes[extrai];
}
combinedTimes.setSize(sz);
times.transfer(combinedTimes);
// Sort
if (times.size() > 1)
{
label starti = 0;
if (times[0].name() == constantName)
{
starti = 1;
}
std::sort(&times[starti], times.end(), instant::less());
// Filter out duplicates
label newi = starti+1;
for (label i = newi; i < times.size(); i++)
{
if (times[i].value() != times[i-1].value())
{
if (newi != i)
{
times[newi] = times[i];
}
newi++;
}
}
times.setSize(newi);
}
}
}
bool Foam::fileOperation::isFileOrDir(const bool isFile, const fileName& f)
{
return
@ -141,8 +237,182 @@ bool Foam::fileOperation::isFileOrDir(const bool isFile, const fileName& f)
}
Foam::tmpNrc<Foam::fileOperation::dirIndexList>
Foam::fileOperation::lookupProcessorsPath(const fileName& fName) const
{
// If path is local to a processor (e.g. contains 'processor2')
// find the corresponding actual processor directory (e.g. 'processors4')
// and index (2)
fileName path;
fileName pDir;
fileName local;
label gStart;
label gSz;
label numProcs;
label proci =
splitProcessorPath(fName, path, pDir, local, gStart, gSz, numProcs);
if (proci != -1)
{
const fileName procPath(path/pDir);
HashTable<dirIndexList>::const_iterator iter =
procsDirs_.find(procPath);
if (iter != procsDirs_.end())
{
return iter();
}
// Read all directories to see any beginning with processor
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
DynamicList<dirIndex> procDirs;
// Note: use parallel synchronised reading so cache will be same
// order on all processors
fileNameList dirNames(readDir(path, fileName::Type::DIRECTORY));
// Extract info from processorsDDD or processorDDD:
// - highest processor number
// - directory+offset containing data for proci
label maxProc = -1;
forAll(dirNames, i)
{
const fileName& dirN = dirNames[i];
// Analyse directory name
fileName rp, rd, rl;
label rStart, rSize, rNum;
label readProci =
splitProcessorPath(dirN, rp, rd, rl, rStart, rSize, rNum);
maxProc = max(maxProc, readProci);
if (proci == readProci)
{
// Found "processorDDD". No need for index.
procDirs.append
(
dirIndex
(
dirN,
Tuple2<pathType, label>(PROCUNCOLLATED, -1)
)
);
}
else if (proci >= rStart && proci < rStart+rSize)
{
// "processorsDDD_start-end"
// Found the file that contains the data for proci
procDirs.append
(
dirIndex
(
dirN,
Tuple2<pathType, label>(PROCOBJECT, proci-rStart)
)
);
}
if (rNum != -1)
{
// Direct detection of processorsDDD
maxProc = rNum-1;
if (rStart == -1)
{
// "processorsDDD"
procDirs.append
(
dirIndex
(
dirN,
Tuple2<pathType, label>(PROCBASEOBJECT, proci)
)
);
}
}
}
if (!Pstream::parRun())
{
// If (as a side effect) we found the number of decompositions
// use it
if (maxProc != -1)
{
const_cast<fileOperation&>(*this).setNProcs(maxProc+1);
}
}
if (returnReduce(procDirs.size(), sumOp<label>()))
{
procsDirs_.insert(procPath, procDirs);
if (debug)
{
Pout<< "fileOperation::lookupProcessorsPath : For:" << procPath
<< " detected:" << procDirs << endl;
}
// Make sure to return a reference
return procsDirs_[procPath];
}
}
return tmpNrc<dirIndexList>(new dirIndexList(0, dirIndex()));
}
bool Foam::fileOperation::exists(IOobject& io) const
{
// Generate output filename for object
fileName objPath(objectPath(io, word::null));
// Test for either directory or a (valid) file & IOobject
bool ok;
if (io.name().empty())
{
ok = isDir(objPath);
}
else
{
ok =
isFile(objPath)
&& io.typeHeaderOk<IOList<label>>(false);// object with local scope
}
if (!ok)
{
// Re-test with searched for objectPath. This is for backwards
// compatibility
fileName originalPath(filePath(io.objectPath()));
if (originalPath != objPath)
{
// Test for either directory or a (valid) file & IOobject
if (io.name().empty())
{
ok = isDir(originalPath);
}
else
{
ok =
isFile(originalPath)
&& io.typeHeaderOk<IOList<label>>(false);
}
}
}
return ok;
}
// * * * * * * * * * * * * * * * * Constructors * * * * * * * * * * * * * * //
Foam::fileOperation::fileOperation(label comm)
:
comm_(comm)
{}
Foam::autoPtr<Foam::fileOperation> Foam::fileOperation::New
(
const word& handlerType,
@ -238,37 +508,64 @@ bool Foam::fileOperation::writeObject
}
//Foam::fileName Foam::fileOperation::objectPath(const fileName& fName) const
//{
// return fName;
//}
Foam::fileName Foam::fileOperation::filePath(const fileName& fName) const
{
fileName path;
fileName local;
label proci = fileOperations::masterUncollatedFileOperation::
splitProcessorPath
(
fName,
path,
local
);
if (debug)
{
Pout<< "fileOperation::filePath :" << " fName:" << fName << endl;
}
fileName procsName(path/processorsDir/local);
fileName path;
fileName pDir;
fileName local;
label gStart;
label gSz;
label numProcs;
label proci =
splitProcessorPath(fName, path, pDir, local, gStart, gSz, numProcs);
if (numProcs != -1)
{
WarningInFunction << "Filename is already adapted:" << fName << endl;
}
// Give preference to processors variant
if (proci != -1 && exists(procsName))
if (proci != -1)
{
return procsName;
}
else if (exists(fName))
{
return fName;
// Get all processor directories
tmpNrc<dirIndexList> procDirs(lookupProcessorsPath(fName));
forAll(procDirs(), i)
{
const fileName& procDir = procDirs()[i].first();
fileName collatedName(path/procDir/local);
if (exists(collatedName))
{
if (debug)
{
Pout<< "fileOperation::filePath : " << collatedName << endl;
}
return collatedName;
}
}
}
return fileName::null;
if (exists(fName))
{
if (debug)
{
Pout<< "fileOperation::filePath : " << fName << endl;
}
return fName;
}
else
{
if (debug)
{
Pout<< "fileOperation::filePath : Not found" << endl;
}
return fileName::null;
}
}
@ -377,8 +674,8 @@ Foam::instantList Foam::fileOperation::findTimes
{
if (debug)
{
Pout<< FUNCTION_NAME
<< " : Finding times in directory " << directory << endl;
Pout<< "fileOperation::findTimes : Finding times in directory "
<< directory << endl;
}
// Read directory entries into a list
@ -393,103 +690,188 @@ Foam::instantList Foam::fileOperation::findTimes
instantList times = sortTimes(dirEntries, constantName);
// Check if directory is processorXXX
fileName procsDir
(
fileOperations::masterUncollatedFileOperation::processorsPath
(
directory
)
);
if (!procsDir.empty() && procsDir != directory)
// Get all processor directories
tmpNrc<dirIndexList> procDirs(lookupProcessorsPath(directory));
forAll(procDirs(), i)
{
fileNameList extraEntries
(
Foam::readDir
(
procsDir,
fileName::DIRECTORY
)
);
instantList extraTimes = sortTimes(extraEntries, constantName);
if (extraTimes.size())
const fileName& procDir = procDirs()[i].first();
fileName collDir(processorsPath(directory, procDir));
if (!collDir.empty() && collDir != directory)
{
bool haveConstant =
fileNameList extraEntries
(
times.size() > 0
&& times[0].name() == constantName
Foam::readDir
(
collDir,
fileName::DIRECTORY
)
);
bool haveExtraConstant =
mergeTimes
(
extraTimes.size() > 0
&& extraTimes[0].name() == constantName
sortTimes(extraEntries, constantName),
constantName,
times
);
// Combine times
instantList combinedTimes(times.size()+extraTimes.size());
label sz = 0;
label extrai = 0;
if (haveExtraConstant)
{
extrai = 1;
if (!haveConstant)
{
combinedTimes[sz++] = extraTimes[0]; // constant
}
}
forAll(times, i)
{
combinedTimes[sz++] = times[i];
}
for (; extrai < extraTimes.size(); extrai++)
{
combinedTimes[sz++] = extraTimes[extrai];
}
combinedTimes.setSize(sz);
times.transfer(combinedTimes);
// Sort
if (times.size() > 1)
{
label starti = 0;
if (times[0].name() == constantName)
{
starti = 1;
}
std::sort(&times[starti], times.end(), instant::less());
// Filter out duplicates
label newi = starti+1;
for (label i = newi; i < times.size(); i++)
{
if (times[i].value() != times[i-1].value())
{
if (newi != i)
{
times[newi] = times[i];
}
newi++;
}
}
times.setSize(newi);
}
}
}
if (debug)
{
Pout<< FUNCTION_NAME
<< " : Found times:" << times << endl;
Pout<< "fileOperation::findTimes : Found times:" << times << endl;
}
return times;
}
Foam::IOobject Foam::fileOperation::findInstance
(
const IOobject& startIO,
const scalar startValue,
const word& stopInstance
) const
{
const Time& time = startIO.time();
IOobject io(startIO);
// Note: - if name is empty, just check the directory itself
// - check both for isFile and headerOk since the latter does a
// filePath so searches for the file.
// - check for an object with local file scope (so no looking up in
// parent directory in case of parallel)
if (exists(io))
{
if (debug)
{
InfoInFunction
<< "Found exact match for \"" << io.name()
<< "\" in " << io.instance()/io.local()
<< endl;
}
return io;
}
// Search back through the time directories to find the time
// closest to and lower than current time
instantList ts = time.times();
label instanceI;
for (instanceI = ts.size()-1; instanceI >= 0; --instanceI)
{
if (ts[instanceI].value() <= startValue)
{
break;
}
}
// continue searching from here
for (; instanceI >= 0; --instanceI)
{
// Shortcut: if actual directory is the timeName we've already tested it
if (ts[instanceI].name() == startIO.instance())
{
continue;
}
io.instance() = ts[instanceI].name();
if (exists(io))
{
if (debug)
{
InfoInFunction
<< "Found exact match for \"" << io.name()
<< "\" in " << io.instance()/io.local()
<< endl;
}
return io;
}
// Check if hit minimum instance
if (ts[instanceI].name() == stopInstance)
{
if (debug)
{
InfoInFunction
<< "Hit stopInstance " << stopInstance << endl;
}
if
(
startIO.readOpt() == IOobject::MUST_READ
|| startIO.readOpt() == IOobject::MUST_READ_IF_MODIFIED
)
{
if (io.name().empty())
{
FatalErrorInFunction
<< "Cannot find directory "
<< io.local() << " in times " << startIO.instance()
<< " down to " << stopInstance
<< exit(FatalError);
}
else
{
FatalErrorInFunction
<< "Cannot find file \"" << io.name()
<< "\" in directory " << io.local()
<< " in times " << startIO.instance()
<< " down to " << stopInstance
<< exit(FatalError);
}
}
return io;
}
}
// times() usually already includes the constant() so would have been
// checked above. Re-test if
// - times() is empty. Sometimes this can happen (e.g. decomposePar with
// collated)
// - times()[0] is not constant
if (!ts.size() || ts[0].name() != time.constant())
{
// Note. This needs to be a hard-coded constant, rather than the
// constant function of the time, because the latter points to
// the case constant directory in parallel cases
io.instance() = time.constant();
if (exists(io))
{
if (debug)
{
InfoInFunction
<< "Found constant match for \"" << io.name()
<< "\" in " << io.instance()/io.local()
<< endl;
}
return io;
}
}
if
(
startIO.readOpt() == IOobject::MUST_READ
|| startIO.readOpt() == IOobject::MUST_READ_IF_MODIFIED
)
{
FatalErrorInFunction
<< "Cannot find file \"" << io.name() << "\" in directory "
<< io.local() << " in times " << startIO.instance()
<< " down to " << time.constant()
<< exit(FatalError);
}
return io;
}
Foam::fileNameList Foam::fileOperation::readObjects
(
const objectRegistry& db,
@ -518,19 +900,9 @@ Foam::fileNameList Foam::fileOperation::readObjects
else
{
// Get processors equivalent of path
fileName procsPath(filePath(path));
fileName prefix;
fileName postfix;
label proci = fileOperations::masterUncollatedFileOperation::
splitProcessorPath
(
path,
prefix,
postfix
);
fileName procsPath(prefix/processorsDir/postfix);
if (proci != -1 && Foam::isDir(procsPath))
if (!procsPath.empty())
{
newInstance = instance;
objectNames = Foam::readDir(procsPath, fileName::FILE);
@ -540,46 +912,252 @@ Foam::fileNameList Foam::fileOperation::readObjects
}
void Foam::fileOperation::setNProcs(const label nProcs)
{}
Foam::label Foam::fileOperation::nProcs
(
const fileName& dir,
const fileName& local
) const
{
if (Foam::isDir(dir/processorsDir))
label nProcs = 0;
if (Pstream::master(comm_))
{
fileName pointsFile
(
dir
/processorsDir
/"constant"
/local
/polyMesh::meshSubDir
/"points"
);
fileNameList dirNames(Foam::readDir(dir, fileName::Type::DIRECTORY));
if (Foam::isFile(pointsFile))
// Detect any processorsDDD or processorDDD
label maxProc = -1;
forAll(dirNames, i)
{
return decomposedBlockData::numBlocks(pointsFile);
const fileName& dirN = dirNames[i];
fileName path, pDir, local;
label start, size, n;
maxProc = max
(
maxProc,
splitProcessorPath(dirN, path, pDir, local, start, size, n)
);
if (n != -1)
{
// Direct detection of processorsDDD
maxProc = n-1;
break;
}
}
nProcs = maxProc+1;
if (nProcs == 0 && Foam::isDir(dir/processorsBaseDir))
{
fileName pointsFile
(
dir
/processorsBaseDir
/"constant"
/local
/polyMesh::meshSubDir
/"points"
);
if (Foam::isFile(pointsFile))
{
nProcs = decomposedBlockData::numBlocks(pointsFile);
}
else
{
WarningInFunction << "Cannot read file " << pointsFile
<< " to determine the number of decompositions."
<< " Returning 1" << endl;
}
}
}
Pstream::scatter(nProcs, Pstream::msgType(), comm_);
return nProcs;
}
Foam::fileName Foam::fileOperation::processorsCasePath
(
const IOobject& io,
const word& procsDir
) const
{
return io.rootPath()/io.time().globalCaseName()/procsDir;
}
Foam::fileName Foam::fileOperation::processorsPath
(
const IOobject& io,
const word& instance,
const word& procsDir
) const
{
return
processorsCasePath(io, procsDir)
/instance
/io.db().dbDir()
/io.local();
}
Foam::fileName Foam::fileOperation::processorsPath
(
const fileName& dir,
const word& procsDir
) const
{
// Check if directory is processorDDD
word caseName(dir.name());
std::string::size_type pos = caseName.find("processor");
if (pos == 0)
{
if (caseName.size() <= 9 || caseName[9] == 's')
{
WarningInFunction << "Directory " << dir
<< " does not end in old-style processorDDD" << endl;
}
return dir.path()/procsDir;
}
else
{
return fileName::null;
}
}
Foam::label Foam::fileOperation::splitProcessorPath
(
const fileName& objectPath,
fileName& path,
fileName& procDir,
fileName& local,
label& groupStart,
label& groupSize,
label& nProcs
)
{
path.clear();
procDir.clear();
local.clear();
// Potentially detected start of number of processors in local group
groupStart = -1;
groupSize = 0;
// Potentially detected number of processors
nProcs = -1;
// Search for processor at start of line or /processor
std::string::size_type pos = objectPath.find("processor");
if (pos == string::npos)
{
return -1;
}
// "processorDDD"
// "processorsNNN"
// "processorsNNN_AA-BB"
if (pos > 0 && objectPath[pos-1] != '/')
{
// Directory not starting with "processor" e.g. "somenamewithprocessor"
return -1;
}
procDir = objectPath;
// Strip leading directory
if (pos > 0)
{
path = objectPath.substr(0, pos-1);
procDir = objectPath.substr(pos);
}
// Strip trailing local directory
pos = procDir.find('/');
if (pos != string::npos)
{
local = procDir.substr(pos+1);
procDir = procDir.substr(0, pos);
}
// Now procDir is e.g.
// - processor0
// - processors0
// - processorBananas
// Look for number after "processor"
fileName f(procDir.substr(9));
if (f.size() && f[0] == 's')
{
// "processsorsNNN"
f = f.substr(1);
// Detect "processorsNNN_AA-BB"
{
std::string::size_type fromStart = f.find("_");
std::string::size_type toStart = f.find("-");
if (fromStart != string::npos && toStart != string::npos)
{
string nProcsName(f.substr(0, fromStart));
string fromName(f.substr(fromStart+1, toStart-(fromStart+1)));
string toName(f.substr(toStart+1));
label groupEnd = -1;
if
(
Foam::read(fromName.c_str(), groupStart)
&& Foam::read(toName.c_str(), groupEnd)
&& Foam::read(nProcsName.c_str(), nProcs)
)
{
groupSize = groupEnd-groupStart+1;
return -1;
}
}
}
// Detect "processorsN"
label n;
if (Foam::read(f.c_str(), n))
{
nProcs = n;
}
return -1;
}
else
{
// Detect "processorN"
label proci;
if (Foam::read(f.c_str(), proci))
{
return proci;
}
else
{
WarningInFunction << "Cannot read file " << pointsFile
<< " to determine the number of decompositions."
<< " Falling back to looking for processor.*" << endl;
return -1;
}
}
}
label nProcs = 0;
while
(
isDir(dir/(word("processor") + name(nProcs)))
)
{
++nProcs;
}
return nProcs;
Foam::label Foam::fileOperation::detectProcessorPath(const fileName& fName)
{
fileName path, pDir, local;
label start, size, nProcs;
return splitProcessorPath(fName, path, pDir, local, start, size, nProcs);
}

152
src/OpenFOAM/global/fileOperations/fileOperation/fileOperation.H
View File

@ -45,6 +45,10 @@ Description
#include "instantList.H"
#include "fileMonitor.H"
#include "labelList.H"
#include "Switch.H"
#include "tmpNrc.H"
#include "NamedEnum.H"
#include "Tuple2.H"
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
@ -56,16 +60,55 @@ class regIOobject;
class objectRegistry;
class Time;
// Description of processor directory naming:
// - processor directory naming
// - whether directory contains a range (so differs on different processors)
// - index in range
//typedef Tuple2<fileName, Tuple2<bool, label>> dirIndex;
//typedef List<dirIndex> dirIndexList;
/*---------------------------------------------------------------------------*\
Class fileOperation Declaration
\*---------------------------------------------------------------------------*/
class fileOperation
{
public:
//- Enumeration for the location of an IOobject
enum pathType
{
NOTFOUND, // not found
ABSOLUTE, // instance is absolute directory
OBJECT, // io.objectPath() exists
WRITEOBJECT, // write path exists
PROCUNCOLLATED, // objectPath exists in processor0
PROCBASEOBJECT, // objectPath exists in specified, constant
// processorsDir (usually 'processorsDDD')
PROCOBJECT, // objectPath exists in locally differing
// processorsDir (e.g. 'processorsDDD_0-1')
PARENTOBJECT, // parent of object path
FINDINSTANCE, // file found in time directory
PROCUNCOLLATEDINSTANCE, // as PROCUNCOLLATED but with instance
PROCBASEINSTANCE, // as PROCBASEOBJECT but with instance
PROCINSTANCE // as PROCOBJECT but with instance
};
static const NamedEnum<pathType, 12> pathTypeNames_;
typedef Tuple2<fileName, Tuple2<pathType, label>> dirIndex;
typedef List<dirIndex> dirIndexList;
protected:
// Protected data
//- Communicator to use
const label comm_;
//- Detected processors directories
mutable HashTable<dirIndexList> procsDirs_;
//- file-change monitor for all registered files
mutable autoPtr<fileMonitor> monitorPtr_;
@ -77,15 +120,35 @@ protected:
//- Sort directory entries according to time value
static instantList sortTimes(const fileNameList&, const word&);
//- Merge two times
static void mergeTimes
(
const instantList& extraTimes,
const word& constantName,
instantList& times
);
//- Helper: check for file (isFile) or directory (!isFile)
static bool isFileOrDir(const bool isFile, const fileName&);
//- Detect presence of processorsDDD
void cacheProcessorsPath(const fileName& fName) const;
//- Lookup name of processorsDDD using cache. Return empty fileName
// if not found
tmpNrc<dirIndexList> lookupProcessorsPath(const fileName&) const;
//- Does ioobject exist. Is either a directory (empty name()) or
// a file
bool exists(IOobject& io) const;
public:
// Static data
//- The processors directory name (usually "processors")
static word processorsDir;
//- Return the processors directory name (usually "processors")
static word processorsBaseDir;
//- Default fileHandler
static word defaultFileHandler;
@ -93,19 +156,6 @@ public:
// Public data types
//- Enumeration for the location of an IOobject
enum pathType
{
NOTFOUND, // not found
ABSOLUTE, // instance is absolute directory
OBJECT, // objectPath exists
PROCESSORSOBJECT, // objectPath exists in processors/
PARENTOBJECT, // parent of object path
FINDINSTANCE, // file found in time directory
PROCESSORSFINDINSTANCE // as above but in processors/
};
//- Runtime type information
TypeName("fileOperation");
@ -116,8 +166,8 @@ public:
// Constructors
//- Construct null
fileOperation() = default;
//- Construct from communicator
explicit fileOperation(const label comm);
// Declare run-time constructor selection table
@ -413,6 +463,24 @@ public:
// Other
//- Actual name of processors dir (for use in mode PROCOBJECT,
// PROCINSTANCE)
virtual word processorsDir(const IOobject& io) const
{
return processorsBaseDir;
}
//- Actual name of processors dir (for use in mode PROCOBJECT,
// PROCINSTANCE)
virtual word processorsDir(const fileName&) const
{
return processorsBaseDir;
}
//- Set number of processor directories/results. Only used in
// decomposePar
virtual void setNProcs(const label nProcs);
//- Get number of processor directories/results. Used for e.g.
// reconstructPar, argList checking
virtual label nProcs
@ -424,9 +492,59 @@ public:
//- Get sorted list of times
virtual instantList findTimes(const fileName&, const word&) const;
//- Find instance where IOobject is. Fails if cannot be found
// and readOpt() is MUST_READ/MUST_READ_IF_MODIFIED. Otherwise
// returns stopInstance.
virtual IOobject findInstance
(
const IOobject& io,
const scalar startValue,
const word& stopInstance
) const;
//- Callback for time change
virtual void setTime(const Time&) const
{}
//- Generate path (like io.path) from root+casename with any
// 'processorXXX' replaced by procDir (usually 'processsors')
fileName processorsCasePath
(
const IOobject&,
const word& procDir
) const;
//- Generate path (like io.path) with provided instance and any
// 'processorXXX' replaced by procDir (usually 'processsors')
fileName processorsPath
(
const IOobject&,
const word& instance,
const word& procDir
) const;
//- Operating on fileName: replace processorXXX with procDir
fileName processorsPath(const fileName&, const word& procDir) const;
//- Split fileName into part before 'processor' and part after.
// Returns -1 or processor number and optionally number
// of processors. Use with care.
// - path/"processor"+Foam::name(proci)/local reconstructs input
// - path/"processors"+Foam::name(nProcs)/local reconstructs
// collated processors equivalence
static label splitProcessorPath
(
const fileName&,
fileName& path,
fileName& procDir,
fileName& local,
label& groupStart,
label& groupSize,
label& nProcs
);
//- Detect processor number from '/aa/bb/processorDDD/cc'
static label detectProcessorPath(const fileName&);
};

88
src/OpenFOAM/global/fileOperations/fileOperationInitialise/fileOperationInitialise.C Normal file
View File

@ -0,0 +1,88 @@
/*---------------------------------------------------------------------------*\
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2017-2018 OpenFOAM Foundation
\\/ M anipulation |
-------------------------------------------------------------------------------
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 "fileOperationInitialise.H"
#include "addToRunTimeSelectionTable.H"
#include "OSspecific.H"
/* * * * * * * * * * * * * * * Static Member Data * * * * * * * * * * * * * */
namespace Foam
{
namespace fileOperations
{
defineTypeNameAndDebug(fileOperationInitialise, 0);
defineRunTimeSelectionTable(fileOperationInitialise, word);
}
}
// * * * * * * * * * * * * * * * * Constructors * * * * * * * * * * * * * * //
Foam::fileOperations::fileOperationInitialise::fileOperationInitialise
(
int& argc,
char**& argv
)
{}
Foam::autoPtr<Foam::fileOperations::fileOperationInitialise>
Foam::fileOperations::fileOperationInitialise::New
(
const word& type,
int& argc,
char**& argv
)
{
if (debug)
{
InfoInFunction << "Constructing fileOperationInitialise" << endl;
}
wordConstructorTable::iterator cstrIter =
wordConstructorTablePtr_->find(type);
if (cstrIter == wordConstructorTablePtr_->end())
{
FatalErrorInFunction
<< "Unknown fileOperationInitialise type "
<< type << nl << nl
<< "Valid fileOperationInitialise types are" << endl
<< wordConstructorTablePtr_->sortedToc()
<< abort(FatalError);
}
return autoPtr<fileOperationInitialise>(cstrIter()(argc, argv));
}
// * * * * * * * * * * * * * * * * Destructor * * * * * * * * * * * * * * * //
Foam::fileOperations::fileOperationInitialise::~fileOperationInitialise()
{}
// ************************************************************************* //

102
src/OpenFOAM/global/fileOperations/fileOperationInitialise/fileOperationInitialise.H Normal file
View File

@ -0,0 +1,102 @@
/*---------------------------------------------------------------------------*\
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2017-2018 OpenFOAM Foundation
\\/ M anipulation |
-------------------------------------------------------------------------------
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/>.
Class
Foam::fileOperationInitialise
\*---------------------------------------------------------------------------*/
#ifndef fileOperationInitialise_H
#define fileOperationInitialise_H
#include "runTimeSelectionTables.H"
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
namespace Foam
{
namespace fileOperations
{
/*---------------------------------------------------------------------------*\
Class fileOperationInitialise Declaration
\*---------------------------------------------------------------------------*/
class fileOperationInitialise
{
public:
//- Runtime type information
TypeName("fileOperationInitialise");
// Constructors
//- Construct components
fileOperationInitialise(int& argc, char**& argv);
// Declare run-time constructor selection table
declareRunTimeSelectionTable
(
autoPtr,
fileOperationInitialise,
word,
(
int& argc, char**& argv
),
(argc, argv)
);
// Selectors
//- Select type
static autoPtr<fileOperationInitialise> New
(
const word& type, int& argc, char**& argv
);
//- Destructor
virtual ~fileOperationInitialise();
// Member Functions
//- Needs threading
virtual bool needsThreading() const = 0;
};
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
} // End namespace fileOperations
} // End namespace Foam
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
#endif
// ************************************************************************* //

84
src/OpenFOAM/global/fileOperations/fileOperationInitialise/unthreadedInitialise.H Normal file
View File

@ -0,0 +1,84 @@
/*---------------------------------------------------------------------------*\
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2017-2018 OpenFOAM Foundation
\\/ M anipulation |
-------------------------------------------------------------------------------
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/>.
Class
Foam::unthreadedInitialise
\*---------------------------------------------------------------------------*/
#ifndef unthreadedInitialise_H
#define unthreadedInitialise_H
#include "fileOperationInitialise.H"
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
namespace Foam
{
namespace fileOperations
{
/*---------------------------------------------------------------------------*\
Class unthreadedInitialise Declaration
\*---------------------------------------------------------------------------*/
class unthreadedInitialise
:
public fileOperationInitialise
{
public:
// Constructors
//- Construct from components
unthreadedInitialise(int& argc, char**& argv)
:
fileOperationInitialise(argc, argv)
{}
//- Destructor
virtual ~unthreadedInitialise()
{}
// Member Functions
//- Needs threading
virtual bool needsThreading() const
{
return false;
}
};
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
} // End namespace fileOperations
} // End namespace Foam
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
#endif
// ************************************************************************* //

1816
src/OpenFOAM/global/fileOperations/masterUncollatedFileOperation/masterUncollatedFileOperation.C
View File

File diff suppressed because it is too large Load Diff

155
src/OpenFOAM/global/fileOperations/masterUncollatedFileOperation/masterUncollatedFileOperation.H
View File

@ -28,6 +28,31 @@ Description
fileOperations that performs all file operations on the master processor.
Requires the calls to be parallel synchronised!
Limitations: - no /processor in filename
- no /uniform/ in the filename
The main logic is in ::filePath which returns a
- same path on all processors. This can either be a global file
(system/controlDict, processorXXX/0/uniform/) or a collated file
(processors/0/p)
- same path on all processors of the local communicator
(processors4_0-1/0/p)
- different path on all processors (processor0/0/p)
system/controlDict:
filePath worldmaster: <globalRoot>/system/controlDict
localmaster: ,,
slave : ,,
processor0/uniform/time
filePath worldmaster: <globalRoot>/processorXXX/uniform/time
localmaster: ,,
slave : ,,
processors0/0/p
processors10/0/p
processors10_2-4/0/p
\*---------------------------------------------------------------------------*/
#ifndef fileOperations_masterUncollatedFileOperation_H
@ -36,6 +61,9 @@ Description
#include "fileOperation.H"
#include "OSspecific.H"
#include "HashPtrTable.H"
#include "Switch.H"
#include "unthreadedInitialise.H"
#include "boolList.H"
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
@ -59,6 +87,9 @@ protected:
// Protected data
//- Any communicator allocated by me
const label myComm_;
//- Cached times for a given directory
mutable HashPtrTable<instantList> times_;
@ -354,18 +385,29 @@ protected:
// Private Member Functions
//- Get the list of processors that are part of this communicator
static labelList subRanks(const label n);
template<class Type>
Type scatterList(const UList<Type>&) const;
Type scatterList(const UList<Type>&, const int, const label comm) const;
template<class Type, class fileOp>
Type masterOp(const fileName&, const fileOp& fop) const;
Type masterOp
(
const fileName&,
const fileOp& fop,
const int tag,
const label comm
) const;
template<class Type, class fileOp>
Type masterOp
(
const fileName&,
const fileName&,
const fileOp& fop
const fileOp& fop,
const int tag,
const label comm
) const;
//- Equivalent of Time::findInstance
@ -375,26 +417,32 @@ protected:
const instant& t
);
//- Search for object; return info on how it was found
//- Search (locally!) for object; return info on how it was found.
// Does not do any parallel communication.
// checkGlobal : also check undecomposed case
// isFile : true:check for file; false:check for directory
fileName filePathInfo
// isFile : true:check for file false:check for directory
// searchType : how was found
// processorsDir : name of processor directory
// instance : instance
virtual fileName filePathInfo
(
const bool checkGlobal,
const bool isFile,
const IOobject& io,
const IOobject&,
const bool search,
pathType&,
word&
pathType& searchType,
word& processorsDir,
word& instance
) const;
//- Construct filePath
static fileName objectPath
fileName localObjectPath
(
const IOobject&,
const pathType&,
const word&
);
const pathType& searchType,
const word& processorsDir,
const word& instancePath
) const;
//- Read file contents and send to processors
static void readAndSend
@ -405,6 +453,29 @@ protected:
PstreamBuffers& pBufs
);
//- Detect file (possibly compressed), read file contents and send
// to processors
static void readAndSend
(
const fileName& fName,
const labelUList& procs,
PstreamBuffers& pBufs
);
//- Read files on comms master
static autoPtr<ISstream> read
(
IOobject& io,
const label comm,
const bool uniform, // on comms master only
const fileNameList& filePaths, // on comms master only
const boolList& procValid // on comms master only
);
//- Helper: check IO for local existence. Like filePathInfo but
// without parent searchign and instance searching
bool exists(const dirIndexList&, IOobject& io) const;
public:
@ -425,9 +496,12 @@ public:
//- Construct null
masterUncollatedFileOperation(const bool verbose);
//- Construct from communicator
masterUncollatedFileOperation(const label comm, const bool verbose);
//- Destructor
virtual ~masterUncollatedFileOperation() = default;
virtual ~masterUncollatedFileOperation();
// Member Functions
@ -682,30 +756,19 @@ public:
//- Get sorted list of times
virtual instantList findTimes(const fileName&, const word&) const;
//- Find instance where IOobject is. Fails if cannot be found
// and readOpt() is MUST_READ/MUST_READ_IF_MODIFIED. Otherwise
// returns stopInstance.
virtual IOobject findInstance
(
const IOobject& io,
const scalar startValue,
const word& stopInstance
) const;
//- Callback for time change
virtual void setTime(const Time&) const;
//- root+casename with any 'processorXXX' replaced by 'processsors'
static fileName processorsCasePath(const IOobject&);
//- Like io.path with provided instance and any 'processorXXX'
// replaced by 'processsors'
static fileName processorsPath(const IOobject&, const word&);
//- Operating on fileName: replace processorXXX with processors
static fileName processorsPath(const fileName&);
//- Split fileName into part before processor and part after.
// Returns -1 or processor number. Use with care.
// - path/"processor" + Foam::name(proci)/local reconstructs input
// - path/"processors"/local reconstructs processors equivalence
static label splitProcessorPath
(
const fileName&,
fileName& path,
fileName& local
);
//- Return cached times
const HashPtrTable<instantList>& times() const
{
@ -714,6 +777,28 @@ public:
};
/*---------------------------------------------------------------------------*\
Class masterUncollatedFileOperationInitialise Declaration
\*---------------------------------------------------------------------------*/
class masterUncollatedFileOperationInitialise
:
public unthreadedInitialise
{
public:
// Constructors
//- Construct from components
masterUncollatedFileOperationInitialise(int& argc, char**& argv);
//- Destructor
virtual ~masterUncollatedFileOperationInitialise()
{}
};
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
} // End namespace fileOperations

58
src/OpenFOAM/global/fileOperations/masterUncollatedFileOperation/masterUncollatedFileOperationTemplates.C
View File

@ -2,7 +2,7 @@
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2017 OpenFOAM Foundation
\\ / A nd | Copyright (C) 2017-2018 OpenFOAM Foundation
\\/ M anipulation |
-------------------------------------------------------------------------------
License
@ -31,14 +31,16 @@ License
template<class Type>
Type Foam::fileOperations::masterUncollatedFileOperation::scatterList
(
const UList<Type>& masterLst
const UList<Type>& masterLst,
const int tag,
const label comm
) const
{
// TBD: more efficient scatter
PstreamBuffers pBufs(UPstream::commsTypes::nonBlocking);
if (Pstream::master())
PstreamBuffers pBufs(UPstream::commsTypes::nonBlocking, tag, comm);
if (Pstream::master(comm))
{
for (label proci = 1; proci < Pstream::nProcs(); proci++)
for (label proci = 1; proci < Pstream::nProcs(comm); proci++)
{
UOPstream os(proci, pBufs);
os << masterLst[proci];
@ -48,9 +50,9 @@ Type Foam::fileOperations::masterUncollatedFileOperation::scatterList
Type myResult;
if (Pstream::master())
if (Pstream::master(comm))
{
myResult = masterLst[Pstream::myProcNo()];
myResult = masterLst[Pstream::myProcNo(comm)];
}
else
{
@ -65,21 +67,25 @@ template<class Type, class fileOp>
Type Foam::fileOperations::masterUncollatedFileOperation::masterOp
(
const fileName& fName,
const fileOp& fop
const fileOp& fop,
const int tag,
const label comm
) const
{
if (IFstream::debug)
{
Pout<< "masterUncollatedFileOperation : Operation on " << fName << endl;
Pout<< "masterUncollatedFileOperation::masterOp : Operation "
<< typeid(fileOp).name()
<< " on " << fName << endl;
}
if (Pstream::parRun())
{
List<fileName> filePaths(Pstream::nProcs());
filePaths[Pstream::myProcNo()] = fName;
Pstream::gatherList(filePaths);
List<fileName> filePaths(Pstream::nProcs(comm));
filePaths[Pstream::myProcNo(comm)] = fName;
Pstream::gatherList(filePaths, tag, comm);
List<Type> result(Pstream::nProcs());
if (Pstream::master())
List<Type> result(filePaths.size());
if (Pstream::master(comm))
{
result = fop(filePaths[0]);
for (label i = 1; i < filePaths.size(); i++)
@ -91,7 +97,7 @@ Type Foam::fileOperations::masterUncollatedFileOperation::masterOp
}
}
return scatterList(result);
return scatterList(result, tag, comm);
}
else
{
@ -105,7 +111,9 @@ Type Foam::fileOperations::masterUncollatedFileOperation::masterOp
(
const fileName& src,
const fileName& dest,
const fileOp& fop
const fileOp& fop,
const int tag,
const label comm
) const
{
if (IFstream::debug)
@ -115,16 +123,16 @@ Type Foam::fileOperations::masterUncollatedFileOperation::masterOp
}
if (Pstream::parRun())
{
List<fileName> srcs(Pstream::nProcs());
srcs[Pstream::myProcNo()] = src;
Pstream::gatherList(srcs);
List<fileName> srcs(Pstream::nProcs(comm));
srcs[Pstream::myProcNo(comm)] = src;
Pstream::gatherList(srcs, tag, comm);
List<fileName> dests(Pstream::nProcs());
dests[Pstream::myProcNo()] = dest;
Pstream::gatherList(dests);
List<fileName> dests(srcs.size());
dests[Pstream::myProcNo(comm)] = dest;
Pstream::gatherList(dests, tag, comm);
List<Type> result(Pstream::nProcs());
if (Pstream::master())
List<Type> result(Pstream::nProcs(comm));
if (Pstream::master(comm))
{
result = fop(srcs[0], dests[0]);
for (label i = 1; i < srcs.size(); i++)
@ -136,7 +144,7 @@ Type Foam::fileOperations::masterUncollatedFileOperation::masterOp
}
}
return scatterList(result);
return scatterList(result, tag, comm);
}
else
{

66
src/OpenFOAM/global/fileOperations/uncollatedFileOperation/uncollatedFileOperation.C
View File

@ -27,9 +27,9 @@ License
#include "Time.H"
#include "Fstream.H"
#include "addToRunTimeSelectionTable.H"
#include "masterUncollatedFileOperation.H"
#include "decomposedBlockData.H"
#include "dummyISstream.H"
#include "unthreadedInitialise.H"
/* * * * * * * * * * * * * * * Static Member Data * * * * * * * * * * * * * */
@ -39,6 +39,15 @@ namespace fileOperations
{
defineTypeNameAndDebug(uncollatedFileOperation, 0);
addToRunTimeSelectionTable(fileOperation, uncollatedFileOperation, word);
// Mark as not needing threaded mpi
addNamedToRunTimeSelectionTable
(
fileOperationInitialise,
unthreadedInitialise,
word,
uncollated
);
}
}
@ -102,17 +111,20 @@ Foam::fileName Foam::fileOperations::uncollatedFileOperation::filePathInfo
// Check if parallel "procesors" directory
if (io.time().processorCase())
{
fileName path = fileOperations::masterUncollatedFileOperation::
processorsPath
tmpNrc<dirIndexList> pDirs
(
io,
io.instance()
lookupProcessorsPath(io.objectPath())
);
fileName objectPath = path/io.name();
if (isFileOrDir(isFile, objectPath))
forAll(pDirs(), i)
{
return objectPath;
const fileName& pDir = pDirs()[i].first();
fileName objPath =
processorsPath(io, io.instance(), pDir)
/io.name();
if (objPath != objectPath && isFileOrDir(isFile, objPath))
{
return objPath;
}
}
}
@ -153,6 +165,8 @@ Foam::fileOperations::uncollatedFileOperation::uncollatedFileOperation
(
const bool verbose
)
:
fileOperation(Pstream::worldComm)
{
if (verbose)
{
@ -533,15 +547,7 @@ Foam::fileOperations::uncollatedFileOperation::readStream
{
// Analyse the objectpath to find out the processor we're trying
// to access
fileName path;
fileName local;
label proci = fileOperations::masterUncollatedFileOperation::
splitProcessorPath
(
io.objectPath(),
path,
local
);
label proci = detectProcessorPath(io.objectPath());
if (proci == -1)
{
@ -551,6 +557,26 @@ Foam::fileOperations::uncollatedFileOperation::readStream
<< exit(FatalIOError);
}
// Analyse the fileName for any processor subset. Note: this
// should really be part of filePath() which should return
// both file and index in file.
fileName path, procDir, local;
label groupStart, groupSize, nProcs;
splitProcessorPath
(
fName,
path,
procDir,
local,
groupStart,
groupSize,
nProcs
);
if (groupStart != -1 && groupSize > 0)
{
proci = proci-groupStart;
}
// Read data and return as stream
return decomposedBlockData::readBlock(proci, isPtr(), io);
}
@ -570,8 +596,8 @@ bool Foam::fileOperations::uncollatedFileOperation::read
{
if (debug)
{
Pout<< "uncollatedFileOperation::read() : "
<< "reading object " << io.objectPath()
Pout<< "uncollatedFileOperation::read :"
<< " Reading object " << io.objectPath()
<< " from file " << endl;
}

27
src/OpenFOAM/include/OSspecific.H
View File

@ -256,33 +256,6 @@ bool dlSymFound(void* handle, const std::string& symbol);
fileNameList dlLoaded();
// Thread handling
//- Allocate a thread
label allocateThread();
//- Start a thread
void createThread(const label, void *(*start_routine) (void *), void *arg);
//- Wait for thread
void joinThread(const label);
//- Delete a thread
void freeThread(const label);
//- Allocate a mutex variable
label allocateMutex();
//- Lock a mutex variable
void lockMutex(const label);
//- Unlock a mutex variable
void unlockMutex(const label);
//- Free a mutex variable
void freeMutex(const label);
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
} // End namespace Foam

5
src/OpenFOAM/memory/tmp/tmpNrc.H
View File

@ -67,7 +67,7 @@ class tmpNrc
mutable T* ptr_;
//- The type (managed pointer | const-reference object)
refType type_;
mutable refType type_;
public:
@ -160,6 +160,9 @@ public:
//- delete object and set pointer to nullptr
inline void clear() const;
//- Swaps the managed object with other tmpNrc.
inline void swap(tmpNrc<T>& other) noexcept;
// Member operators

4
src/Pstream/dummy/UPstream.C
View File

@ -2,7 +2,7 @@
========= |
\\ / F ield | OpenFOAM: The Open Source CFD Toolbox
\\ / O peration |
\\ / A nd | Copyright (C) 2011-2017 OpenFOAM Foundation
\\ / A nd | Copyright (C) 2011-2018 OpenFOAM Foundation
\\/ M anipulation | Copyright (C) 2016 OpenCFD Ltd.
-------------------------------------------------------------------------------
License
@ -42,7 +42,7 @@ bool Foam::UPstream::initNull()
}
bool Foam::UPstream::init(int& argc, char**& argv)
bool Foam::UPstream::init(int& argc, char**& argv, const bool needsThread)
{
FatalErrorInFunction
<< "The dummy Pstream library cannot be used in parallel mode"

13
src/Pstream/mpi/UPstream.C
View File

@ -101,7 +101,7 @@ bool Foam::UPstream::initNull()
}
bool Foam::UPstream::init(int& argc, char**& argv)
bool Foam::UPstream::init(int& argc, char**& argv, const bool needsThread)
{
int flag = 0;
@ -129,17 +129,16 @@ bool Foam::UPstream::init(int& argc, char**& argv)
//MPI_Init(&argc, &argv);
int wanted_thread_support = MPI_THREAD_SINGLE;
if (fileOperations::collatedFileOperation::maxThreadFileBufferSize > 0)
{
wanted_thread_support = MPI_THREAD_MULTIPLE;
}
int provided_thread_support;
MPI_Init_thread
(
&argc,
&argv,
wanted_thread_support,
(
needsThread
? MPI_THREAD_MULTIPLE
: MPI_THREAD_SINGLE
),
&provided_thread_support
);

8
tutorials/IO/fileHandler/Allclean
View File

@ -1,7 +1,11 @@
#!/bin/sh
cd ${0%/*} || exit 1 # Run from this directory
. $WM_PROJECT_DIR/bin/tools/CleanFunctions # Tutorial clean functions
cd ${0%/*} || exit 1 # Run from this directory
# Source tutorial run functions
. $WM_PROJECT_DIR/bin/tools/CleanFunctions
cleanCase
rm -rf machineA
rm -rf machineB
#------------------------------------------------------------------------------

74
tutorials/IO/fileHandler/Allrun
View File

@ -22,13 +22,33 @@ runApplication decomposePar -fileHandler collated
runParallel $application -fileHandler collated
runApplication reconstructPar -latestTime -fileHandler collated
#- Delete collated files
rm -rf processors
#- Test writing uncollated format
runApplication -s uncollated decomposePar -fileHandler uncollated
runApplication -s uncollated decomposePar -fileHandler uncollated -force
runParallel -s uncollated $application -fileHandler uncollated
#- Test uncollated+distributed running: copy to different roots
rm -rf machineA/fileHandler
mkdir -p machineA/fileHandler
( cd machineA/fileHandler && \
cp -r ../../processor[0-1] . && \
cp -r ../../system . && \
mkdir -p constant && cp ../../constant/* constant
)
#- Note: slave node does not need constant&system since these are global
rm -rf machineB/fileHandler
mkdir -p machineB/fileHandler
( cd machineB/fileHandler && \
cp -r ../../processor[2-3] .
)
#- Run with different roots
( d=$PWD && \
cd machineA/fileHandler && \
runParallel -s multiRoot $application \
-fileHandler masterUncollated -ioRanks '(0 2)' \
-roots "(\"$d/machineA\" \"$d/machineB\" \"$d/machineB\")"
)
#- Restart from uncollated
runParallel -s collated $application -fileHandler collated
runApplication -s collated reconstructPar -latestTime -fileHandler collated
@ -36,4 +56,50 @@ runApplication -s collated reconstructPar -latestTime -fileHandler collated
#- Convert the parallel format to uncollated
runParallel foamFormatConvert -fileHandler uncollated
#- Restart with multiple IO ranks
runParallel -s multiCollated \
$application -fileHandler collated -ioRanks '(0 2)'
#- Reconstruct the multi-rank format. Delete the collated directory
# since conflicts with the multi-collated directory
rm -rf processors4
runApplication -s multiCollated reconstructPar -latestTime \
-fileHandler collated -ioRanks '(0 2)'
#- Convert the multi-rank format to uncollated
runParallel -s uncollated foamFormatConvert -fileHandler uncollated
#- Restart from multiCollated using collated
runParallel -s uncollated_from_multiCollated \
$application -fileHandler uncollated
#- Test collated+distributed running: copy to different roots
# Important: make sure to copy uniform since we're copying it
#- Delete all processor directories
runApplication -s collated decomposePar \
-fileHandler collated -force -copyUniform
rm -rf machineA/fileHandler
mkdir -p machineA/fileHandler
( cd machineA/fileHandler && \
cp -r ../../processor* . && \
cp -r ../../system . && \
mkdir -p constant && cp ../../constant/* constant
)
#- Note: slave node does not need constant&system since these are global
rm -rf machineB/fileHandler
mkdir -p machineB/fileHandler
#- Run with different roots
( d=$PWD && \
cd machineA/fileHandler && \
runParallel -s distributed_multiCollated $application \
-fileHandler collated -ioRanks '(0 2)' \
-roots "(\"$d/machineA\" \"$d/machineB\" \"$d/machineB\")"
)
#------------------------------------------------------------------------------

7
tutorials/IO/fileHandler/system/controlDict
View File

@ -37,7 +37,7 @@ writeFormat binary;
writePrecision 6;
writeCompression uncompressed;
writeCompression false;
timeFormat general;
@ -45,9 +45,4 @@ timePrecision 6;
runTimeModifiable yes;
// OptimisationSwitches
// {
// fileHandler collated;
// }
// ************************************************************************* //