WriterMT.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_WRITERMT_H
#define EVIO_6_0_WRITERMT_H


#include <iostream>
#include <iomanip>
#include <fstream>
#include <vector>
#include <string>
#include <thread>
#include <queue>
#include <chrono>
#include <memory>
#include <atomic>


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


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


namespace evio {


    class WriterMT {


    private:

        class RecordWriter {

        private:

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

        public:

            RecordWriter(WriterMT * 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() {
//std::cout << "WRITE: supply last = " << supply->getLastSequence() << ", lasSeqProcessed = " << lastSeqProcessed <<
//" supply->getLast > lastSeq = " <<  (supply->getLastSequence() > lastSeqProcessed)  <<  std::endl;
                while (supply->getLastSequence() > lastSeqProcessed.load()) {
                    std::this_thread::sleep_for(std::chrono::milliseconds(1));
                }

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

            void run() {
                int64_t currentSeq;

                try {
                    while (true) {

                        std::cout << "   RecordWriter: try getting record to write" << std::endl;
                        // Get the next record for this thread to write
                        auto item = supply->getToWrite();

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

                            currentSeq = item->getSequence();
                            // Pull record out of wrapping object
                            std::shared_ptr<RecordOutput> & record = item->getRecord();

                            // Do write
                            auto & header = record->getHeader();
                            int bytesToWrite = header->getLength();
                            // Record length of this record
                            writer->recordLengths->push_back(bytesToWrite);
                            // Followed by events in record
                            writer->recordLengths->push_back(header->getEntries());
                            writer->writerBytesWritten += bytesToWrite;

                            auto buf = record->getBinaryBuffer();
                            std::cout << "   RecordWriter: use outFile to write file, buf pos = " << buf->position() <<
                                 ", lim = " << buf->limit() << ", bytesToWrite = " << bytesToWrite << std::endl;
                            writer->outFile.write(reinterpret_cast<const char *>(buf->array()), bytesToWrite);
                            if (writer->outFile.fail()) {
                                throw EvioException("failed write to file");
                            }

                            record->reset();

                            // Release back to supply
                            supply->releaseWriter(item);

                            // Now we're done with this sequence
                            lastSeqProcessed = currentSeq;
                        }
                    }
                }
                catch (boost::thread_interrupted & e) {
                    //cout << "   RecordWriter: INTERRUPTED, return" << endl;
                }
            }
        };


    private:


        size_t writerBytesWritten = 0ULL;
        uint8_t* firstEvent = nullptr;
        uint32_t firstEventLength = 0;
        uint32_t maxEventCount = 0;
        uint32_t maxBufferSize = 0;
        uint32_t recordNumber = 1;
        uint32_t compressionThreadCount = 1;

        std::string fileName = "";

        std::ofstream outFile;

        FileHeader fileHeader;

        std::string dictionary;

        std::shared_ptr<ByteBuffer> dictionaryFirstEventBuffer;

        ByteOrder byteOrder {ByteOrder::ENDIAN_LOCAL};

        std::shared_ptr<RecordOutput> outputRecord;

        std::vector<uint8_t> headerArray;

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

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

        std::shared_ptr<RecordSupply> supply;

        std::vector<RecordWriter> recordWriterThreads;

        std::vector<RecordCompressor> recordCompressorThreads;

        std::shared_ptr<RecordRingItem> ringItem;


        bool addingTrailer = true;
        bool addTrailerIndex = false;
        bool closed = false;
        bool opened = false;
        bool firstRecordWritten = false;
        bool haveDictionary = false;
        bool haveFirstEvent = false;
        bool haveUserHeader = false;


    public:

        WriterMT();

        WriterMT(const ByteOrder & order, uint32_t maxEventCount, uint32_t maxBufferSize,
                 Compressor::CompressionType compType, uint32_t compressionThreads);

        explicit WriterMT(
                const HeaderType & hType,
                const ByteOrder & order = ByteOrder::ENDIAN_LITTLE,
                uint32_t maxEventCount = 0,
                uint32_t maxBufferSize = 0,
                const std::string & dictionary = "",
                uint8_t* firstEvent = nullptr,
                uint32_t firstEventLen = 0,
                Compressor::CompressionType compressionType = Compressor::UNCOMPRESSED,
                uint32_t compressionThreads = 1,
                bool addTrailerIndex = false,
                uint32_t ringSize = 16);

        explicit WriterMT(const std::string & filename);

        WriterMT(const std::string & filename, const ByteOrder & order, uint32_t maxEventCount, uint32_t maxBufferSize,
                 Compressor::CompressionType compressionType, uint32_t compressionThreads);

        ~WriterMT() = default;


    private:

        std::shared_ptr<ByteBuffer> createDictionaryRecord();
        void writeTrailer(bool writeIndex, uint32_t recordNum);

    public:

        const ByteOrder & getByteOrder() const;
//    ByteBuffer   & getBuffer();
        FileHeader   & getFileHeader();
//    RecordHeader & getRecordHeader();
//    RecordOutput & getRecord();
        Compressor::CompressionType getCompressionType();

        bool addTrailer() const;
        void addTrailer(bool add);
        bool addTrailerWithIndex();
        void addTrailerWithIndex(bool addTrailingIndex);

        void open(const std::string & filename);
        void open(const std::string & filename, uint8_t* userHdr, uint32_t userLen);

        std::shared_ptr<ByteBuffer> createHeader(uint8_t* userHdr, uint32_t userLen);
        std::shared_ptr<ByteBuffer> createHeader(ByteBuffer & userHdr);

        void writeRecord(RecordOutput & record);

        // Use internal RecordOutput to write individual events

        void addEvent(uint8_t* buffer, uint32_t offset, uint32_t length);
        void addEvent(ByteBuffer & buffer);
//    void addEvent(EvioBank & bank);
        void addEvent(EvioNode & node);

        void reset();
        void close();

    };

}


#endif //EVIO_6_0_WRITERMT_H