threadsafecircularqueue.h

Go to the documentation of this file.

 
#pragma once
 
#include <config.h>
 
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <chrono>
 
#include <cassert>
#include <cstdint>
#include <ctime>
 
#include <type_traits>
 
template <typename ValueType, uint32_t capacity>
class ThreadSafeCircularQueue {
    using index_type = uint32_t;
    struct StorageType {
        int overflows {0}; 
        time_t time; 
        ValueType value;
    };
    StorageType data[capacity];
    std::atomic<index_type> read_index;
    std::mutex read_mutex;
    std::condition_variable read_bell;
 
    std::atomic<index_type> write_index;
    std::mutex write_mutex;
    std::condition_variable write_bell;
 
    inline index_type next (index_type index) {
        return (index + 1) % capacity;
    }
 
    inline void write_common() {
        data[write_index].time = time (nullptr);
        write_index = next (write_index);
        data[write_index].overflows = 0;
        read_bell.notify_one();
    }
 
public:
    using value_type = ValueType;
 
    template <typename VT>
    void push_back (VT &&value) {
        std::unique_lock<std::mutex> lock (write_mutex);
        while (next (write_index) == read_index) {
            write_bell.wait (lock);
        }
        data[write_index].value = std::move (value);
        write_common();
    }
 
    template <typename VT>
    bool try_push_back (VT &&value, 
                        const std::chrono::milliseconds &wait_duration = std::chrono::milliseconds::zero()) {
        std::unique_lock<std::mutex> lock (write_mutex);
        if (next (write_index) == read_index) {
            if (!read_bell.wait_for (lock, wait_duration, [this] ()->bool {
                return (next (write_index) != read_index);
            })) {
                data[write_index].overflows++;
                return false;
            }
        }
        data[write_index].value = std::move (value);
        write_common();
        return true;
    }
 
    ValueType pop_front (int *overflows = nullptr, time_t *sent_time = nullptr) {
        std::unique_lock<std::mutex> lock (read_mutex);
        while (read_index == write_index) {
            read_bell.wait (lock);
        }
        if (overflows) {
            *overflows = data[read_index].overflows;
        }
        if (sent_time) {
            *sent_time = data[read_index].time;
        }
        ValueType temp = std::move (data[read_index].value);
        read_index = next (read_index);
        write_bell.notify_one();
        return temp;
    }
 
    bool try_pop_front (ValueType *value,
                        int *overflows = nullptr,
                        time_t *sent_time = nullptr,
                        std::chrono::milliseconds wait_duration = std::chrono::milliseconds::zero()) {
        std::unique_lock<std::mutex> lock (read_mutex);
        if (read_index == write_index) {
            if (!read_bell.wait_for (lock, wait_duration, [this] ()->bool {
                return (read_index != write_index);
            })) {
                return false;
            };
        }
        if (overflows) {
            *overflows = data[read_index].overflows;
        }
        if (sent_time) {
            *sent_time = data[read_index].time;
        }
        *value = std::move (data[read_index].value);
        read_index = next (read_index);
        write_bell.notify_one();
        return true;
    }
};