QueuePolicies.hpp

Go to the documentation of this file.

 
#pragma once
 
#include <cstddef>
#include <cstdint>
#include <queue>
#include <type_traits>
#include <utility>
 
#if defined(_WIN32) || defined(_WIN64) || defined(__unix__) || defined(__APPLE__) || (defined(_GLIBCXX_HAS_GTHREADS) && _GLIBCXX_HAS_GTHREADS) || defined(_LIBCPP_HAS_THREAD_API_PTHREAD)
#include <mutex>
#ifndef CORIUM_HAS_STD_MUTEX
#define CORIUM_HAS_STD_MUTEX 1
#endif
#else
#ifndef CORIUM_HAS_STD_MUTEX
#define CORIUM_HAS_STD_MUTEX 0
#endif
#endif
 
#include "corium/Events.hpp"
#include "corium/MpscRingBuffer.hpp"
 
namespace corium {
 
enum class EventPriority : uint8_t {
    High = 0,
    Normal = 1,
    Low = 2
};
 
template <typename EventVariant = DefaultEvents, size_t Capacity = 1024>
class BoundedMpscQueuePolicy {
public:
    using EventType = EventVariant;
    static constexpr std::size_t capacity = Capacity; 
 
    struct PushResult {
        bool pushed;
        bool wasEmpty;
    };
 
    PushResult tryPush(EventVariant&& event, EventPriority priority = EventPriority::Normal)
    {
        (void)priority;
        auto res = _ringBuffer.tryPush(std::move(event));
        return {res.pushed, res.wasEmpty};
    }
 
    PushResult tryPush(const EventVariant& event, EventPriority priority = EventPriority::Normal)
    {
        EventVariant copy = event;
        return tryPush(std::move(copy), priority);
    }
 
    bool tryPop(EventVariant& event)
    {
        return _ringBuffer.tryPop(event);
    }
 
    [[nodiscard]] bool empty() const noexcept
    {
        return _ringBuffer.empty();
    }
 
private:
    MpscRingBuffer<EventVariant, Capacity> _ringBuffer;
};
 
template <
    typename EventVariant = DefaultEvents,
    size_t HighCapacity = 256,
    size_t NormalCapacity = 1024,
    size_t LowCapacity = 0
>
class PriorityMpscQueuePolicy {
public:
    using EventType = EventVariant;
    static constexpr std::size_t capacity = HighCapacity + NormalCapacity + (LowCapacity > 0 ? LowCapacity : 0);
 
    struct PushResult {
        bool pushed;
        bool wasEmpty;
    };
 
    PushResult tryPush(EventVariant&& event, EventPriority priority = EventPriority::Normal)
    {
        bool wasOverallEmpty = empty();
        bool pushed = false;
 
        switch (priority) {
            case EventPriority::High: {
                auto res = _highRingBuffer.tryPush(std::move(event));
                pushed = res.pushed;
                break;
            }
            case EventPriority::Normal: {
                auto res = _normalRingBuffer.tryPush(std::move(event));
                pushed = res.pushed;
                break;
            }
            case EventPriority::Low: {
                if constexpr (LowCapacity > 0) {
                    auto res = _lowRingBuffer.tryPush(std::move(event));
                    pushed = res.pushed;
                } else {
                    auto res = _normalRingBuffer.tryPush(std::move(event));
                    pushed = res.pushed;
                }
                break;
            }
        }
 
        return {pushed, wasOverallEmpty};
    }
 
    PushResult tryPush(const EventVariant& event, EventPriority priority = EventPriority::Normal)
    {
        EventVariant copy = event;
        return tryPush(std::move(copy), priority);
    }
 
    bool tryPop(EventVariant& event)
    {
        if (_highRingBuffer.tryPop(event)) {
            return true;
        }
        if (_normalRingBuffer.tryPop(event)) {
            return true;
        }
        if constexpr (LowCapacity > 0) {
            if (_lowRingBuffer.tryPop(event)) {
                return true;
            }
        }
        return false;
    }
 
    [[nodiscard]] bool empty() const noexcept
    {
        bool isEmpty = _highRingBuffer.empty() && _normalRingBuffer.empty();
        if constexpr (LowCapacity > 0) {
            isEmpty = isEmpty && _lowRingBuffer.empty();
        }
        return isEmpty;
    }
 
private:
    MpscRingBuffer<EventVariant, HighCapacity> _highRingBuffer;
    MpscRingBuffer<EventVariant, NormalCapacity> _normalRingBuffer;
    [[no_unique_address]] std::conditional_t<(LowCapacity > 0),
        MpscRingBuffer<EventVariant, (LowCapacity > 0 ? LowCapacity : 1)>,
        std::monostate
    > _lowRingBuffer{};
};
 
template <typename EventVariant = DefaultEvents>
class BlockingQueuePolicy {
public:
    using EventType = EventVariant;
 
    struct PushResult {
        bool pushed;
        bool wasEmpty;
    };
 
    PushResult tryPush(EventVariant event, EventPriority priority = EventPriority::Normal)
    {
        (void)priority;
#if CORIUM_HAS_STD_MUTEX
        std::lock_guard<std::mutex> lock(_mutex);
#endif
        bool wasEmpty = _queue.empty();
        _queue.push(std::move(event));
        return {true, wasEmpty};
    }
 
    bool tryPop(EventVariant& event)
    {
#if CORIUM_HAS_STD_MUTEX
        std::lock_guard<std::mutex> lock(_mutex);
#endif
        if (_queue.empty()) {
            return false;
        }
        event = std::move(_queue.front());
        _queue.pop();
        return true;
    }
 
    [[nodiscard]] bool empty() const
    {
#if CORIUM_HAS_STD_MUTEX
        std::lock_guard<std::mutex> lock(_mutex);
#endif
        return _queue.empty();
    }
 
private:
    std::queue<EventVariant> _queue;
#if CORIUM_HAS_STD_MUTEX
    mutable std::mutex _mutex;
#endif
};
 
template <typename EventVariant = DefaultEvents>
class NoQueuePolicy {
public:
    using EventType = EventVariant;
 
    struct PushResult {
        bool pushed = false;
        bool wasEmpty = false;
    };
 
    PushResult tryPush(EventVariant, EventPriority = EventPriority::Normal) noexcept
    {
        return {false, false};
    }
 
    bool tryPop(EventVariant&) noexcept
    {
        return false;
    }
 
    [[nodiscard]] bool empty() const noexcept
    {
        return true;
    }
};
 
} // namespace corium