thread_safe_deque.hpp

Go to the documentation of this file.

 
#pragma once
#include <atomic>
#include <condition_variable>
#include <deque>
#include <mutex>
 
namespace csv {
    namespace internals {
        template<typename T>
        class ThreadSafeDeque {
        public:
            ThreadSafeDeque(size_t notify_size = 100) : _notify_size(notify_size) {}
            ThreadSafeDeque(const ThreadSafeDeque& other) {
                this->data = other.data;
                this->_notify_size = other._notify_size;
                this->_is_empty.store(other._is_empty.load(std::memory_order_acquire), std::memory_order_release);
            }
 
            ThreadSafeDeque(const std::deque<T>& source) : ThreadSafeDeque() {
                this->data = source;
                this->_is_empty.store(source.empty(), std::memory_order_release);
            }
 
            bool empty() const noexcept {
                return this->_is_empty.load(std::memory_order_acquire);
            }
 
            T& front() noexcept {
                std::lock_guard<std::mutex> lock{ this->_lock };
                return this->data.front();
            }
 
            T& operator[](size_t n) {
                return this->data[n];
            }
 
            void push_back(T&& item) {
                std::lock_guard<std::mutex> lock{ this->_lock };
                this->data.push_back(std::move(item));
                this->_is_empty.store(false, std::memory_order_release);
 
                if (this->data.size() >= _notify_size) {
                    this->_cond.notify_all();
                }
            }
 
            T pop_front() noexcept {
                std::lock_guard<std::mutex> lock{ this->_lock };
                T item = std::move(data.front());
                data.pop_front();
                
                // Update empty flag if we just emptied the deque
                if (this->data.empty()) {
                    this->_is_empty.store(true, std::memory_order_release);
                }
                
                return item;
            }
 
            constexpr bool is_waitable() const noexcept { return this->_is_waitable; }
 
            void wait() {
                if (!is_waitable()) {
                    return;
                }
 
                std::unique_lock<std::mutex> lock{ this->_lock };
                this->_cond.wait(lock, [this] { return this->data.size() >= _notify_size || !this->is_waitable(); });
                lock.unlock();
            }
 
            size_t size() const noexcept {
                std::lock_guard<std::mutex> lock{ this->_lock };
                return this->data.size();
            }
 
            typename std::deque<T>::iterator begin() noexcept {
                return this->data.begin();
            }
 
            typename std::deque<T>::iterator end() noexcept {
                return this->data.end();
            }
 
            void notify_all() {
                this->_is_waitable.store(true, std::memory_order_release);
                this->_cond.notify_all();
            }
 
            void kill_all() {
                this->_is_waitable.store(false, std::memory_order_release);
                this->_cond.notify_all();
            }
 
        private:
            std::atomic<bool> _is_empty{ true };     // Lock-free empty() check  
            std::atomic<bool> _is_waitable{ false }; // Lock-free is_waitable() check
            size_t _notify_size;
            mutable std::mutex _lock;
            std::condition_variable _cond;
            std::deque<T> data;
        };
    }
}