EventWriter.h

Go to the documentation of this file.

//
// Copyright 2020, Jefferson Science Associates, LLC.
// Subject to the terms in the LICENSE file found in the top-level directory.
//
// EPSCI Group
// Thomas Jefferson National Accelerator Facility
// 12000, Jefferson Ave, Newport News, VA 23606
// (757)-269-7100


#ifndef EVIO_6_0_EVENTWRITER_H
#define EVIO_6_0_EVENTWRITER_H


#include <cstdlib>
#include <iostream>
#include <iomanip>
#include <fstream>
#include <vector>
#include <string>
#include <queue>
#include <chrono>
#include <memory>
#include <bitset>
#include <exception>
#include <atomic>
#include <algorithm>
#include <future>
#include <sys/stat.h>
#include <sys/statvfs.h>

#ifdef USE_FILESYSTEMLIB
    #include <experimental/filesystem>
#endif


#include "HeaderType.h"
#include "FileHeader.h"
#include "ByteBuffer.h"
#include "ByteOrder.h"
#include "RecordOutput.h"
#include "RecordHeader.h"
#include "Compressor.h"
#include "RecordSupply.h"
#include "RecordCompressor.h"
#include "Util.h"
#include "EvioException.h"
#include "EvioBank.h"


#include "Disruptor/Util.h"
#include <boost/thread.hpp>
#include <boost/chrono.hpp>

#ifdef USE_FILESYSTEMLIB
namespace fs = std::experimental::filesystem;
#endif


namespace evio {

    class EventWriter {

    private:


        class RecordWriter {

        private:

            EventWriter * writer = nullptr;
            std::shared_ptr<RecordSupply> supply;
            boost::thread thd;
            std::atomic_long lastSeqProcessed{-1};

            std::shared_ptr<RecordRingItem> storedItem;
            std::atomic_bool forceToDisk{false};
            std::atomic<std::uint64_t> forcedRecordId{0};

        public:

            RecordWriter(EventWriter * pwriter, std::shared_ptr<RecordSupply> & recordSupply) :
                    writer(pwriter), supply(recordSupply)  {
            }

            RecordWriter(RecordWriter && obj) noexcept :
                    writer(obj.writer),
                    supply(std::move(obj.supply)),
                    thd(std::move(obj.thd)) {

                lastSeqProcessed.store(obj.lastSeqProcessed);
            }

            RecordWriter & operator=(RecordWriter && obj) noexcept {
                if (this != &obj) {
                    writer = obj.writer;
                    lastSeqProcessed.store(obj.lastSeqProcessed);
                    supply = std::move(obj.supply);
                    thd  = std::move(obj.thd);
                }
                return *this;
            }

            // Do not free writer!
            ~RecordWriter() {
                thd.interrupt();
                if (thd.try_join_for(boost::chrono::milliseconds(500))) {
                    std::cout << "     RecordWriter thread did not quit after 1/2 sec" << std::endl;
                }
            }

            void startThread() {
                thd = boost::thread([this]() {this->run();});
            }

            void stopThread() {
                // Send signal to interrupt it
                thd.interrupt();
                // Wait for it to stop
                thd.join();
            }

            void waitForLastItem() {
                //cout << "WRITE: supply last = " << supply->getLastSequence() << ", lasSeqProcessed = " << lastSeqProcessed <<
                //" supply->getLast > lastSeq = " <<  (supply->getLastSequence() > lastSeqProcessed)  <<  endl;
                while (supply->getLastSequence() > lastSeqProcessed.load()) {
                    std::this_thread::sleep_for(std::chrono::milliseconds(1));
                }

                // Stop this thread, not the calling thread
                stopThread();
            }

            void setForcedRecordId(uint64_t id) {
                forcedRecordId = id;
                forceToDisk = true;
            }

            std::shared_ptr<RecordRingItem> storeRecordCopy(std::shared_ptr<RecordRingItem> & rec) {
                // Call copy constructor of RecordRingItem, then make into shared pointer
                storedItem = std::make_shared<RecordRingItem>(*(rec.get()));
                return storedItem;
            }

            void run() {

                try {

                    while (true) {

                        // Get the next record for this thread to write
                        // shared_ptr<RecordRingItem>
                        auto item = supply->getToWrite();

                        {
                            // Only allow interruption when blocked on trying to get item
                            boost::this_thread::disable_interruption d1;

                            int64_t currentSeq = item->getSequence();
                            // Pull record out of wrapping object
                            auto record = item->getRecord();

                            // Only need to check the disk when writing the first record following
                            // a file split. That first write will create the file. If there isn't
                            // enough room, then flag will be set.
                            bool checkDisk = item->isCheckDisk();

                            // Check the disk before we try to write if about to create another file,
                            // we're told to check the disk, and we're not forcing to disk
                            if ((writer->bytesWritten < 1) && checkDisk && (!forceToDisk.load())) {

                                // If there isn't enough free space to write the complete, projected
                                // size file, and we're not trying to force the write ...
                                // store a COPY of the record for now and release the original so
                                // that writeEventToFile() does not block.
                                //
                                // So here is the problem. We are stuck in a loop here if disk is full.
                                // If events are flowing and since writing data to file is the bottleneck,
                                // it is likely that all records have been filled and published onto
                                // the ring. AND, writeEventToFile() blocks in a spin as it tries to get the
                                // next record from the ring which, unfortunately, never comes.
                                //
                                // When writeEventToFile() blocks, it can't respond by returning a "false" value.
                                // This plays havoc with code like the emu which is not expecting the write
                                // to block (at least not for very long).
                                //
                                // To break writeEventToFile() out of its spinning block, we make a copy of the
                                // item we're trying to write and release the original record. This allows
                                // writeEventToFile() to grab a new (newly released) record, write an event into
                                // it, and return to the caller. From then on, writeEventToFile() can prevent
                                // blocking by checking for a full disk (which it couldn't do before since
                                // the signal came too late).

                                while (writer->fullDisk() && (!forceToDisk.load())) {
                                    // Wait for a sec and try again
                                    std::this_thread::sleep_for(std::chrono::seconds(1));

                                    // If we released the item in a previous loop, don't do it again
                                    if (!item->isAlreadyReleased()) {
                                        // Copy item
                                        auto copiedItem = storeRecordCopy(item);
                                        // Release original so we don't block writeEvent()
                                        supply->releaseWriter(item);
                                        item = copiedItem;
                                    }

                                    // Wait until space opens up
                                }

                                // If we're here, there must be free space available even
                                // if there previously wasn't.
                            }

                            // Do write
                            // Write current item to file
                            //cout << "EventWriter: Calling writeToFileMT(item)\n";
                            writer->writeToFileMT(item, forceToDisk.load());

                            // Turn off forced write to disk, if the record which
                            // initially triggered it, has now been written.
                            if (forceToDisk.load() && (forcedRecordId.load() == item->getId())) {
                                //cout << "EventWriter: WROTE the record that triggered force, reset to false\n";
                                forceToDisk = false;
                            }

                            // Now we're done with this sequence
                            lastSeqProcessed = currentSeq;

                            // Split file if needed
                            if (item->splitFileAfterWrite()) {
                                writer->splitFile();
                            }

                            // Release back to supply
                            supply->releaseWriter(item);
                        }
                    }
                }
                catch (boost::thread_interrupted & e) {
                    std::cout << "EventWriter: INTERRUPTED, return" << std::endl;
                }
            }
        };



        class FileCloser {

            class CloseAsyncFChan {

                // Store quantities from exterior classes or store quantities that
                // may change between when this object is created and when this thread is run.
                std::shared_ptr<std::fstream> afChannel;
                std::shared_ptr<std::future<void>> future;
                std::shared_ptr<RecordSupply> supply;
                std::shared_ptr<RecordRingItem> item;
                FileHeader fHeader;
                std::shared_ptr<std::vector<uint32_t>> recLengths;
                uint64_t bytesWrittenToFile;
                uint32_t recordNum;
                bool addTrailer;
                bool writeIndx;
                bool noFileWriting;
                ByteOrder byteOrder;

                // A couple of things used to clean up after thread is done
                FileCloser *closer;
                std::shared_ptr<CloseAsyncFChan> sharedPtrOfMe;

                // Local storage
                uint32_t hdrBufferBytes = RecordHeader::HEADER_SIZE_BYTES + 2048;
                ByteBuffer hdrBuffer{hdrBufferBytes};
                uint8_t *hdrArray = hdrBuffer.array();

                // Thread which does the file writing
                boost::thread thd;

            public:

                CloseAsyncFChan(std::shared_ptr<std::fstream> &afc,
                                std::shared_ptr<std::future<void>> &future1,
                                std::shared_ptr<RecordSupply> &supply,
                                std::shared_ptr<RecordRingItem> &ringItem,
                                FileHeader &fileHeader, std::shared_ptr<std::vector<uint32_t>> recordLengths,
                                uint64_t bytesWritten, uint32_t recordNumber,
                                bool addingTrailer, bool writeIndex, bool noWriting,
                                ByteOrder &order, FileCloser *fc) :

                        afChannel(afc), future(future1), supply(supply),
                        item(ringItem), byteOrder(order) {

                    fHeader            = fileHeader;
                    recLengths         = recordLengths;
                    bytesWrittenToFile = bytesWritten;
                    recordNum          = recordNumber;
                    addTrailer         = addingTrailer;
                    writeIndx          = writeIndex;
                    noFileWriting      = noWriting;
                    closer             = fc;

                    hdrBuffer.order(order);

                    startThread();
                }


                void setSharedPointerOfThis(std::shared_ptr<CloseAsyncFChan> & mySharedPtr) {
                    sharedPtrOfMe = mySharedPtr;
                }


                void stopThread() {
                    // Send signal to interrupt it
                    thd.interrupt();

                    // Wait for it to stop
                    if (thd.try_join_for(boost::chrono::milliseconds(500))) {
                        std::cout << "CloseAsyncFChan thread did not quit after 1/2 sec" << std::endl;
                    }

                    try {
                        // When this thread is done, remove itself from vector
                        closer->removeThread(sharedPtrOfMe);
                    }
                    catch (std::exception & e) {}
                }


            private:


                void startThread() {
                    thd = boost::thread([this]() { this->run(); });
                }

                void run() {
                    // Finish writing to current file
                    if (future != nullptr) {
                        try {
                            // Wait for last write to end before we continue
                            future->get();
                        }
                        catch (std::exception &e) {}
                    }

                    // Release resources back to the ring
                    std::cout << "Closer: releaseWriterSequential, will release item seq = " << item->getSequence() << std::endl;
                    supply->releaseWriterSequential(item);

                    try {
                        if (addTrailer && !noFileWriting) {
                            writeTrailerToFile();
                        }
                    }
                    catch (std::exception &e) {}

                    try {
                        afChannel->close();
                    }
                    catch (std::exception &e) {
                        std::cout << e.what() << std::endl;
                    }

                    try {
                        // When this thread is done, remove itself from vector
                        closer->removeThread(sharedPtrOfMe);
                    }
                    catch (std::exception & e) {}
                }


                void writeTrailerToFile() {

                    // Keep track of where we are right now which is just before trailer
                    uint64_t trailerPosition = bytesWrittenToFile;

                    // If we're NOT adding a record index, just write trailer
                    if (!writeIndx) {
                        try {
                            // hdrBuffer is only used in this method
                            hdrBuffer.position(0).limit(RecordHeader::HEADER_SIZE_BYTES);
                            RecordHeader::writeTrailer(hdrBuffer, 0, recordNum, nullptr);
                        }
                        catch (EvioException &e) {/* never happen */}

                        afChannel->write(reinterpret_cast<char *>(hdrArray),
                                         RecordHeader::HEADER_SIZE_BYTES);
                        if (afChannel->fail()) {
                            throw EvioException("error writing to file");
                        }
                    }
                    else {
                        // Write trailer with index

                        // How many bytes are we writing here?
                        uint32_t bytesToWrite = RecordHeader::HEADER_SIZE_BYTES + 4*recLengths->size();

                        // Make sure our array can hold everything
                        if (hdrBufferBytes < bytesToWrite) {
                            hdrBuffer = ByteBuffer(bytesToWrite);
                            hdrArray = hdrBuffer.array();
                        }
                        hdrBuffer.limit(bytesToWrite).position(0);

                        // Place data into hdrBuffer - both header and index
                        try {
                            RecordHeader::writeTrailer(hdrBuffer, (size_t)0, recordNum, recLengths);
                        }
                        catch (EvioException &e) {/* never happen */}
                        afChannel->write(reinterpret_cast<char *>(hdrArray), bytesToWrite);
                        if (afChannel->fail()) {
                            throw EvioException("error writing to file");
                        }
                    }

                    // Update file header's trailer position word
                    if (!byteOrder.isLocalEndian()) {
                        trailerPosition = SWAP_64(trailerPosition);
                    }

                    afChannel->seekg(FileHeader::TRAILER_POSITION_OFFSET);
                    afChannel->write(reinterpret_cast<char *>(&trailerPosition), sizeof(trailerPosition));
                    if (afChannel->fail()) {
                        throw EvioException("error writing to file");
                    }

                    // Update file header's bit-info word
                    if (writeIndx) {
                        uint32_t bitInfo = fHeader.setBitInfo(fHeader.hasFirstEvent(),
                                                              fHeader.hasDictionary(),
                                                              true);
                        if (!byteOrder.isLocalEndian()) {
                            bitInfo = SWAP_32(bitInfo);
                        }
                        afChannel->seekg(FileHeader::BIT_INFO_OFFSET);
                        afChannel->write(reinterpret_cast<char *>(&bitInfo), sizeof(bitInfo));
                        if (afChannel->fail()) {
                            throw EvioException("error writing to  file");
                        }
                    }

                    // Update file header's record count word
                    uint32_t recordCount = recordNum - 1;
                    if (!byteOrder.isLocalEndian()) {
                        recordCount = SWAP_32(recordCount);
                    }
                    afChannel->seekg(FileHeader::RECORD_COUNT_OFFSET);
                    afChannel->write(reinterpret_cast<char *>(&recordCount), sizeof(recordCount));
                    if (afChannel->fail()) {
                        throw EvioException("error writing to file");
                    }
                }
            };


        private:


            std::vector<std::shared_ptr<CloseAsyncFChan>> threads;


        public:


            void close() {
                for (const std::shared_ptr<CloseAsyncFChan> & thread : threads) {
                    thread->stopThread();
                }
                threads.clear();
            }


            void removeThread(std::shared_ptr<CloseAsyncFChan> & thread) {
                // Look for this pointer among the shared pointers
                threads.erase(std::remove(threads.begin(), threads.end(), thread), threads.end());
            }


            void closeAsyncFile( std::shared_ptr<std::fstream> &afc,
                                 std::shared_ptr<std::future<void>> &future1,
                                 std::shared_ptr<RecordSupply> &supply,
                                 std::shared_ptr<RecordRingItem> &ringItem,
                                 FileHeader &fileHeader, std::shared_ptr<std::vector<uint32_t>> &recordLengths,
                                 uint64_t bytesWritten, uint32_t recordNumber,
                                 bool addingTrailer, bool writeIndex, bool noFileWriting,
                                 ByteOrder &order) {

                auto a = std::make_shared<CloseAsyncFChan>(afc, future1, supply, ringItem,
                                                           fileHeader, recordLengths,
                                                           bytesWritten, recordNumber,
                                                           addingTrailer, writeIndex,
                                                           noFileWriting, order, this);

                threads.push_back(a);
                a->setSharedPointerOfThis(a);
            }

            ~FileCloser() {
                close();
            }
        };


        //-------------------------------------------------------------------------------------


    private:


        std::shared_ptr<RecordOutput> commonRecord;

        std::shared_ptr<RecordOutput> currentRecord;

        std::shared_ptr<RecordRingItem> currentRingItem;

        std::shared_ptr<RecordSupply> supply;

        uint32_t maxSupplyBytes = 0;

        Compressor::CompressionType compressionType{Compressor::UNCOMPRESSED};

        uint32_t compressionFactor;

        std::shared_ptr<std::vector<uint32_t>> recordLengths;

        //TODO: DOES THIS NEED TO BE ATOMIC IF MT write????????????????????????????????
        size_t bytesWritten = 0ULL;

        bool addingTrailer = true;

        bool addTrailerIndex = false;

        std::vector<uint8_t> headerArray;

        std::vector<RecordCompressor> recordCompressorThreads;

        std::vector<EventWriter::RecordWriter> recordWriterThread;

        uint32_t recordsWritten = 0;

        uint32_t recordNumber = 1;

        std::string xmlDictionary;

        std::vector<uint8_t> dictionaryByteArray;

        std::vector<uint8_t> firstEventByteArray;

        bool haveFirstEvent = false;

        bool closed = false;

        bool toFile = false;

        bool append = false;

        bool hasAppendDictionary = false;

        uint32_t eventsWrittenTotal = 0;

        ByteOrder byteOrder {ByteOrder::ENDIAN_LOCAL};

        //-----------------------
        // Buffer related members
        //-----------------------

        uint32_t sourceId = 0;

        uint32_t bufferSize = 0;

        std::shared_ptr<ByteBuffer> buffer;

        std::shared_ptr<ByteBuffer> usedBuffer;

        std::vector<std::shared_ptr<ByteBuffer>> internalBuffers;

        uint32_t commonRecordBytesToBuffer = 0;

        uint32_t eventsWrittenToBuffer = 0;

        //-----------------------
        // File related members
        //-----------------------

        bool diskIsFull = false;

        std::atomic_bool diskIsFullVolatile{false};

        bool fileOpen = false;

        uint64_t idCounter = 0ULL;

        FileHeader fileHeader;

        FileHeader appendFileHeader;

        std::string currentFileName;

#ifdef USE_FILESYSTEMLIB

        fs::path currentFilePath;
#endif

        std::shared_ptr<std::future<void>> future1;

        std::shared_ptr<RecordRingItem> ringItem1;

        uint32_t futureIndex = 0;

        std::shared_ptr<std::fstream> asyncFileChannel = nullptr;

        uint64_t fileWritingPosition = 0ULL;

        uint32_t splitNumber = 0;

        uint32_t splitCount = 0;

        std::string baseFileName;

        uint32_t specifierCount = 0;

        uint32_t runNumber = 0;

        uint64_t split = 0ULL;

        uint32_t splitIncrement = 0;

        uint64_t splitEventBytes = 0ULL;

        uint32_t splitEventCount = 0;

        uint32_t streamId = 0;

        uint32_t streamCount = 1;

        bool singleThreadedCompression = false;

        bool overWriteOK = false;

        uint32_t eventsWrittenToFile = 0;

        bool hasTrailerWithIndex = false;

        uint32_t userHeaderLength = 0;

        uint32_t userHeaderPadding = 0;

        uint32_t indexLength = 0;

        std::shared_ptr<FileCloser> fileCloser;

        //-----------------------
        bool noFileWriting = false;
        //-----------------------


    public:


        explicit EventWriter(std::string & filename,
                             const ByteOrder & byteOrder = ByteOrder::nativeOrder(),
                             bool append = false);

        EventWriter(std::string & filename,
                    std::string & dictionary,
                    const ByteOrder & byteOrder = ByteOrder::nativeOrder(),
                    bool append = false);

        EventWriter(std::string baseName, const std::string & directory, const std::string & runType,
                    uint32_t runNumber, uint64_t split, uint32_t maxRecordSize, uint32_t maxEventCount,
                    const ByteOrder & byteOrder, const std::string & xmlDictionary, bool overWriteOK,
                    bool append, std::shared_ptr<EvioBank> firstEvent, uint32_t streamId, uint32_t splitNumber,
                    uint32_t splitIncrement, uint32_t streamCount, Compressor::CompressionType compressionType,
                    uint32_t compressionThreads, uint32_t ringSize, uint32_t bufferSize);

        //---------------------------------------------
        // BUFFER Constructors
        //---------------------------------------------

        explicit EventWriter(std::shared_ptr<ByteBuffer> & buf);
        EventWriter(std::shared_ptr<ByteBuffer> & buf, std::string & xmlDictionary);
        EventWriter(std::shared_ptr<ByteBuffer> & buf, uint32_t maxRecordSize, uint32_t maxEventCount,
                    const std::string & xmlDictionary, uint32_t recordNumber,
                    Compressor::CompressionType compressionType);

    private:

        void reInitializeBuffer(std::shared_ptr<ByteBuffer> & buf, const std::bitset<24> *bitInfo,
                                uint32_t recordNumber, bool useCurrentBitInfo);

        static void staticWriteFunction(EventWriter *pWriter, const char* data, size_t len);
        static void staticDoNothingFunction(EventWriter *pWriter);

    public:

        bool isDiskFull();
        void setBuffer(std::shared_ptr<ByteBuffer> & buf, std::bitset<24> *bitInfo, uint32_t recNumber);
        void setBuffer(std::shared_ptr<ByteBuffer> & buf);

    private:

        std::shared_ptr<ByteBuffer> getBuffer();


    public:

        std::shared_ptr<ByteBuffer> getByteBuffer();
        void setSourceId(int sId);
        void setEventType(int type);

        bool writingToFile() const;
        bool isClosed() const;

        std::string getCurrentFilename() const;
        size_t getBytesWrittenToBuffer() const;
        std::string getCurrentFilePath() const;
        uint32_t getSplitNumber() const;
        uint32_t getSplitCount() const;
        uint32_t getRecordNumber() const;
        uint32_t getEventsWritten() const;
        ByteOrder getByteOrder() const;

        void setStartingRecordNumber(uint32_t startingRecordNumber);

        void setFirstEvent(std::shared_ptr<EvioNode> & node);
        void setFirstEvent(std::shared_ptr<ByteBuffer> & buf);
        void setFirstEvent(std::shared_ptr<EvioBank> bank);

    private:

        void createCommonRecord(const std::string & xmlDict,
                                std::shared_ptr<EvioBank> const & firstBank,
                                std::shared_ptr<EvioNode> const & firstNode,
                                std::shared_ptr<ByteBuffer> const & firstBuf);

        void writeFileHeader() ;


    public :

        void flush();
        void close();

    protected:

        void examineFileHeader();

    private:

        void toAppendPosition();

    public:

        bool hasRoom(uint32_t bytes);

        bool writeEvent(std::shared_ptr<EvioNode> & node, bool force);
        bool writeEvent(std::shared_ptr<EvioNode> & node, bool force, bool duplicate);
        bool writeEventToFile(std::shared_ptr<EvioNode> & node, bool force, bool duplicate);

        bool writeEvent(std::shared_ptr<ByteBuffer> & bankBuffer);
        bool writeEvent(std::shared_ptr<ByteBuffer> & bankBuffer, bool force);

        bool writeEvent(std::shared_ptr<EvioBank> bank);
        bool writeEvent(std::shared_ptr<EvioBank> bank, bool force);

    private:

        bool writeEvent(std::shared_ptr<EvioBank> bank,
                        std::shared_ptr<ByteBuffer> bankBuffer, bool force);
        bool writeEventToFile(std::shared_ptr<EvioBank> bank,
                              std::shared_ptr<ByteBuffer> bankBuffer, bool force);

        bool fullDisk();

        void compressAndWriteToFile(bool force);
        bool tryCompressAndWriteToFile(bool force);

        bool writeToFile(bool force, bool checkDisk);
        void writeToFileMT(std::shared_ptr<RecordRingItem> & item, bool force);

        void splitFile();
        void writeTrailerToFile(bool writeIndex);
        void flushCurrentRecordToBuffer() ;
        bool writeToBuffer(std::shared_ptr<EvioBank> & bank, std::shared_ptr<ByteBuffer> & bankBuffer) ;

        uint32_t trailerBytes();
        void writeTrailerToBuffer(bool writeIndex);

    };

}


#endif //EVIO_6_0_EVENTWRITER_H