RecordSupply.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_RECORDSUPPLY_H
#define EVIO_6_0_RECORDSUPPLY_H


#include <string>
#include <memory>
#include <vector>
#include <atomic>
#include <mutex>


#include "ByteOrder.h"
#include "Compressor.h"
#include "RecordRingItem.h"
#include "EvioException.h"
#include "Disruptor/Util.h"
#include "Disruptor/Sequence.h"
#include "Disruptor/ISequence.h"
#include "Disruptor/RingBuffer.h"
#include "Disruptor/ISequenceBarrier.h"
#include "Disruptor/TimeoutException.h"
#include "Disruptor/SpinCountBackoffWaitStrategy.h"


namespace evio {


    class RecordSupply {

    private:

        std::mutex supplyMutex;

        ByteOrder order {ByteOrder::ENDIAN_LOCAL};

        uint32_t maxEventCount = 0;
        uint32_t maxBufferSize = 0;
        Compressor::CompressionType compressionType {Compressor::UNCOMPRESSED};
        uint32_t compressionThreadCount = 1;
        uint32_t ringSize = 0;


        std::shared_ptr<Disruptor::RingBuffer<std::shared_ptr<RecordRingItem>>> ringBuffer = nullptr;


        // Stuff for reporting errors

        std::atomic<bool> haveErrorCondition{false};
        std::string error {""};

        // Stuff for reporting conditions (disk is full)

        std::atomic<bool> diskFull{false};

        // Stuff for compression threads

        std::shared_ptr<Disruptor::ISequenceBarrier> compressBarrier;
        std::vector<std::shared_ptr<Disruptor::ISequence>> compressSeqs;
        std::vector<int64_t> nextCompressSeqs;
        std::vector<int64_t> availableCompressSeqs;

        // Stuff for writing thread

        std::shared_ptr<Disruptor::ISequenceBarrier> writeBarrier;
        std::vector<std::shared_ptr<Disruptor::ISequence>> writeSeqs;
        int64_t nextWriteSeq = 0L;
        int64_t availableWriteSeq = 0L;

        // For thread safety in getToWrite() & releaseWriter()

        int64_t lastSequenceReleased = -1L;
        int64_t maxSequence = -1L;
        uint32_t between = 0;


    public:

        RecordSupply();
        // No need to copy these things
        RecordSupply(const RecordSupply & supply) = delete;
        RecordSupply(uint32_t ringSize, ByteOrder order,
                     uint32_t threadCount, uint32_t maxEventCount, uint32_t maxBufferSize,
                     Compressor::CompressionType & compressionType);

        ~RecordSupply() {
            compressSeqs.clear();
            nextCompressSeqs.clear();
            nextCompressSeqs.clear();
            availableCompressSeqs.clear();
            writeSeqs.clear();
            ringBuffer.reset();
        }

        void errorAlert();

        uint32_t getMaxRingBytes();
        uint32_t getRingSize();
        ByteOrder & getOrder();
        uint64_t getFillLevel();
        int64_t getLastSequence();

        std::shared_ptr<RecordRingItem> get();
        void publish(std::shared_ptr<RecordRingItem> & item);
        std::shared_ptr<RecordRingItem> getToCompress(uint32_t threadNumber);
        std::shared_ptr<RecordRingItem> getToWrite();

        void releaseCompressor(std::shared_ptr<RecordRingItem> & item);
        bool releaseWriterSequential(std::shared_ptr<RecordRingItem> & item);
        bool releaseWriter(std::shared_ptr<RecordRingItem> & item);
        void release(uint32_t threadNum, int64_t sequenceNum);

        bool haveError();
        void haveError(bool err);
        std::string getError();
        void setError(std::string & err);

        bool isDiskFull();
        void setDiskFull(bool full);

    };

}


#endif //EVIO_6_0_RECORDSUPPLY_H