IpcBlockingPriorityQueueImpl.hpp

/*
Copyright (c) 2017-2018,
Battelle Memorial Institute; Lawrence Livermore National Security, LLC; Alliance for Sustainable
Energy, LLC All rights reserved. See LICENSE file and DISCLAIMER for more details.
*/
#pragma once
 
#include "helics/external/optional.hpp"
 
#include <boost/interprocess/mapped_region.hpp>
#include <boost/interprocess/shared_memory_object.hpp>
#include <boost/interprocess/sync/interprocess_condition.hpp>
#include <boost/interprocess/sync/interprocess_mutex.hpp>
#include <chrono>
#include <utility>
 
namespace helics {
namespace ipc {
    namespace detail {
        struct dataIndex {
            int32_t offset;
            int32_t dataSize;
        };
        class dataBlock {
          private:
            unsigned char* origin = nullptr;
            unsigned char* next = nullptr;
            size_t totalSize = 0;
            dataIndex* next = nullptr;
            int dataCount = 0;
 
          public:
            dataBlock(unsigned char* newBlock, size_t blockSize);
 
            void swap(dataBlock& other) noexcept;
 
            int getCurrentCount() const { return dataCount; }
            bool isSpaceAvaialble(int sz) const;
            bool empty() const { return (dataCount == 0); }
 
            bool push(const unsigned char* block, int blockSize);
 
            int next_data_size() const;
 
            int pop(unsigned char* block);
 
            void reverse();
            void clear();
        };
        class IpcBlockingPriorityQueueImpl {
          private:
            mutable boost::interprocess::interprocess_mutex
                m_pushLock;  
            dataBlock pushData;
            mutable boost::interprocess::interprocess_mutex
                m_pullLock;  
            dataBlock pullData;
            mutable boost::interprocess::interprocess_mutex
                m_conditionLock;  
            bool queueEmptyFlag{true};  
            bool queueFullFlag{false};
            // the condition variable should be keyed of the conditionLock
            boost::interprocess::interprocess_condition
                condition_empty;  
            boost::interprocess::interprocess_condition
                condition_full;  
            unsigned char* data1 = nullptr;
            unsigned char* data2 = nullptr;
            size_t data1Size = 0;
            size_t data2Size = 0;
            unsigned char* dataPriority = nullptr;
            size_t dataPrioritySize = 0;
 
          public:
            IpcBlockingPriorityQueueImpl(void* dataBlock, size_t blockSize);
 
            void clear();
 
            IpcBlockingPriorityQueueImpl(const IpcBlockingPriorityQueueImpl&) = delete;
            IpcBlockingPriorityQueueImpl& operator=(const IpcBlockingPriorityQueueImpl&) = delete;
 
            void push(const unsigned char* data, size_t size);
            void pushPriority(const unsigned char* data, size_t size);
            bool try_push(const unsigned char* data, size_t size);
 
            bool try_pushPriority(const unsigned char* data, size_t size);
 
            stx::optional<std::pair<unsigned char*, int>> try_peek() const;
 
            bool try_pop(unsigned char* data, int maxSize);
 
            int pop(unsigned char* data, int maxSize);
 
            void pop(std::chrono::milliseconds timeout, unsigned char* data, int maxSize);
 
            bool empty() const;
        };
 
    }  // namespace detail
}  // namespace ipc
}  // namespace helics