OFstreamCollator.H

Go to the documentation of this file.

/*---------------------------------------------------------------------------*\
  =========                 |
  \\      /  F ield         | OpenFOAM: The Open Source CFD Toolbox
   \\    /   O peration     |
    \\  /    A nd           | www.openfoam.com
     \\/     M anipulation  |
-------------------------------------------------------------------------------
    Copyright (C) 2017-2018 OpenFOAM Foundation
    Copyright (C) 2019-2025 OpenCFD Ltd.
-------------------------------------------------------------------------------
License
    This file is part of OpenFOAM.
 
    OpenFOAM is free software: you can redistribute it and/or modify it
    under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.
 
    OpenFOAM is distributed in the hope that it will be useful, but WITHOUT
    ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
    FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
    for more details.
 
    You should have received a copy of the GNU General Public License
    along with OpenFOAM.  If not, see <http://www.gnu.org/licenses/>.
 
Class
    Foam::OFstreamCollator
 
Description
    Threaded file writer.
 
    Collects all data from all processors and writes as single
    'decomposedBlockData' file. The operation is determined by the
    buffer size (maxThreadFileBufferSize setting):
    - local size of data is larger than buffer: receive and write processor
    by processor (i.e. 'scheduled'). Does not use a thread, no file size
    limit.
    - total size of data is larger than buffer (but local is not):
    thread does all the collecting and writing of the processors. No file
    size limit.
    - total size of data is less than buffer:
    collecting is done locally; the thread only does the writing
    (since the data has already been collected)
 
SourceFiles
    OFstreamCollator.C
 
\*---------------------------------------------------------------------------*/
 
#ifndef Foam_OFstreamCollator_H
#define Foam_OFstreamCollator_H
 
#include "IOstream.H"
#include "List.H"
#include "CircularBuffer.H"  // As FIFO
#include "dictionary.H"
 
#include <mutex>
#include <thread>
 
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
 
namespace Foam
{
 
/*---------------------------------------------------------------------------*\
                        Class OFstreamCollator Declaration
\*---------------------------------------------------------------------------*/
 
class OFstreamCollator
{
    // Private Class
 
        //- Holds data to be written
        struct writeData
        {
            const label comm_;
            const word objectType_;
            const fileName pathName_;
            DynamicList<char> localData_;
            const labelList sizes_;
            List<List<char>> procData_;
            const IOstreamOption streamOpt_;
            IOstreamOption::atomicType atomic_;
            IOstreamOption::appendType append_;
            const dictionary headerEntries_;
 
            writeData() = delete;                      // No default construct
            writeData(const writeData&) = delete;      // No copy construct
            writeData(writeData&&) = delete;           // No move construct
            void operator=(const writeData&) = delete; // No copy assign
            void operator=(writeData&&) = delete;      // No move assign
 
            //- Construct without local or proc data
            writeData
            (
                const label comm,
                const word& objectType,
                const fileName& pathName,
                const labelUList& sizes,
                IOstreamOption streamOpt,
                IOstreamOption::atomicType atomic,
                IOstreamOption::appendType append,
                const dictionary& headerEntries
            )
            :
                comm_(comm),
                objectType_(objectType),
                pathName_(pathName),
                sizes_(sizes),
                streamOpt_(streamOpt),
                atomic_(atomic),
                append_(append),
                headerEntries_(headerEntries)
            {}
 
            //- Move reset local data
            void transfer(DynamicList<char>& localData)
            {
                localData_.transfer(localData);
            }
 
            //- The (approximate) size of local + any optional proc data
            off_t size() const
            {
                off_t total = localData_.size();
                for (const auto& data : procData_)
                {
                    total += data.size();
                }
                return total;
            }
        };
 
 
    // Private Data
 
        //- Total amount of storage to use for object stack below
        const off_t maxBufferSize_;
 
        mutable std::mutex mutex_;
 
        std::unique_ptr<std::thread> thread_;
 
        //- FIFO of files to write and their contents
        CircularBuffer<writeData*> objects_;
 
        //- Whether thread is running (and not exited)
        bool threadRunning_;
 
        //- 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
 
        //- Write actual file
        static bool writeFile
        (
            const label comm,
            const word& objectType,
            const fileName& fName,
            const UList<char>& localData,
            const labelUList& recvSizes,
            const UList<std::string_view>& procData,
            IOstreamOption streamOpt,
            IOstreamOption::atomicType atomic,
            IOstreamOption::appendType append,
            const dictionary& headerEntries
        );
 
        //- Write all files in stack
        static void* writeAll(void *threadarg);
 
        //- Wait for total size of objects_ (master + optional slave data)
        //  to be wantedSize less than overall maxBufferSize.
        void waitForBufferSpace(const off_t wantedSize) const;
 
 
public:
 
    // Declare name of the class and its debug switch
    TypeName("OFstreamCollator");
 
 
    // Constructors
 
        //- Construct from buffer size (0 = do not use thread)
        //- and with worldComm
        explicit OFstreamCollator(const off_t maxBufferSize);
 
        //- Construct from buffer size (0 = do not use thread)
        //- and specified communicator
        OFstreamCollator(const off_t maxBufferSize, const label comm);
 
 
    //- Destructor
    virtual ~OFstreamCollator();
 
 
    // Member Functions
 
        //- Write file with contents, possibly taking ownership of the
        //- content.
        //  Blocks until write-thread has space available
        //  (total file sizes < maxBufferSize)
        bool write
        (
            const word& objectType,
            const fileName& fName,
            DynamicList<char>&& localData,
            IOstreamOption streamOpt,
            IOstreamOption::atomicType atomic,
            IOstreamOption::appendType append,
            const bool useThread = true,
            const dictionary& headerEntries = dictionary::null
        );
 
        //- Write file with contents.
        FOAM_DEPRECATED_FOR(2023-09, "use write with movable content")
        bool write
        (
            const word& objectType,
            const fileName& fName,
            const std::string& s,
            IOstreamOption streamOpt,
            IOstreamOption::atomicType atomic,
            IOstreamOption::appendType append,
            const bool useThread = true,
            const dictionary& headerEntries = dictionary::null
        )
        {
            DynamicList<char> charData(label(s.size()));
            charData.resize(s.size());
            std::copy(s.begin(), s.end(), charData.begin());
 
            return write
            (
                objectType,
                fName,
                std::move(charData),
                streamOpt,
                atomic,
                append,
                useThread,
                headerEntries
            );
        }
 
        //- Wait for all thread actions to have finished
        void waitAll();
};
 
 
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
 
} // End namespace Foam
 
// * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * //
 
#endif
 
// ************************************************************************* //