#pragma once #include "utils.h" #include "work_thread.h" #include #include #include #include #include #include #include // TODO: cleanup includes. struct BaseThreadQueue { virtual bool IsEmpty() = 0; virtual ~BaseThreadQueue() = default; }; // TODO Remove after migration to C++14 namespace { template struct index_sequence {}; template struct make_index_sequence { using type = typename make_index_sequence::type; }; template struct make_index_sequence<0, Is...> { using type = index_sequence; }; } // namespace // std::lock accepts two or more arguments. We define an overload for one // argument. namespace std { template void lock(Lockable& l) { l.lock(); } } // namespace std template struct MultiQueueLock { MultiQueueLock(Queue... lockable) : tuple_{lockable...} { lock(); } ~MultiQueueLock() { unlock(); } void lock() { lock_impl(typename make_index_sequence::type{}); } void unlock() { unlock_impl(typename make_index_sequence::type{}); } private: template void lock_impl(index_sequence) { std::lock(std::get(tuple_)->mutex_...); } template void unlock_impl(index_sequence) { (void)std::initializer_list{ (std::get(tuple_)->mutex_.unlock(), 0)...}; } std::tuple tuple_; }; struct MultiQueueWaiter { std::condition_variable_any cv; static bool HasState(std::initializer_list queues) { for (BaseThreadQueue* queue : queues) { if (!queue->IsEmpty()) return true; } return false; } template void Wait(BaseThreadQueue... queues) { MultiQueueLock l(queues...); while (!HasState({queues...})) cv.wait(l); } }; // A threadsafe-queue. http://stackoverflow.com/a/16075550 template struct ThreadedQueue : public BaseThreadQueue { public: ThreadedQueue() : total_count_(0) { owned_waiter_ = MakeUnique(); waiter_ = owned_waiter_.get(); owned_waiter1_ = MakeUnique(); waiter1_ = owned_waiter1_.get(); } // TODO remove waiter1 after split of on_indexed explicit ThreadedQueue(MultiQueueWaiter* waiter, MultiQueueWaiter* waiter1 = nullptr) : total_count_(0), waiter_(waiter), waiter1_(waiter1) {} // Returns the number of elements in the queue. This is lock-free. size_t Size() const { return total_count_; } // Add an element to the queue. template ::*push)(T&&)> void Push(T&& t, bool priority) { std::lock_guard lock(mutex_); if (priority) (priority_.*push)(std::move(t)); else (queue_.*push)(std::move(t)); ++total_count_; waiter_->cv.notify_one(); if (waiter1_) waiter1_->cv.notify_one(); } void PushFront(T&& t, bool priority = false) { Push<&std::deque::push_front>(std::move(t), priority); } void PushBack(T&& t, bool priority = false) { Push<&std::deque::push_back>(std::move(t), priority); } // Add a set of elements to the queue. void EnqueueAll(std::vector&& elements, bool priority = false) { if (elements.empty()) return; std::lock_guard lock(mutex_); total_count_ += elements.size(); for (T& element : elements) { if (priority) priority_.push_back(std::move(element)); else queue_.push_back(std::move(element)); } elements.clear(); waiter_->cv.notify_all(); } // Return all elements in the queue. std::vector DequeueAll() { std::lock_guard lock(mutex_); total_count_ = 0; std::vector result; result.reserve(priority_.size() + queue_.size()); while (!priority_.empty()) { result.emplace_back(std::move(priority_.front())); priority_.pop_front(); } while (!queue_.empty()) { result.emplace_back(std::move(queue_.front())); queue_.pop_front(); } return result; } // Returns true if the queue is empty. This is lock-free. bool IsEmpty() { return total_count_ == 0; } // Get the first element from the queue. Blocks until one is available. T Dequeue() { std::unique_lock lock(mutex_); waiter_->cv.wait(lock, [&]() { return !priority_.empty() || !queue_.empty(); }); auto execute = [&](std::deque* q) { auto val = std::move(q->front()); q->pop_front(); --total_count_; return std::move(val); }; if (!priority_.empty()) return execute(&priority_); return execute(&queue_); } // Get the first element from the queue without blocking. Returns a null // value if the queue is empty. optional TryPopFrontHelper(int which) { std::lock_guard lock(mutex_); auto execute = [&](std::deque* q) { auto val = std::move(q->front()); q->pop_front(); --total_count_; return std::move(val); }; if (which & 2 && priority_.size()) return execute(&priority_); if (which & 1 && queue_.size()) return execute(&queue_); return nullopt; } optional TryPopFront() { return TryPopFrontHelper(3); } optional TryPopBack() { std::lock_guard lock(mutex_); auto execute = [&](std::deque* q) { auto val = std::move(q->back()); q->pop_back(); --total_count_; return std::move(val); }; // Reversed if (queue_.size()) return execute(&queue_); if (priority_.size()) return execute(&priority_); return nullopt; } optional TryPopFrontLow() { return TryPopFrontHelper(1); } optional TryPopFrontHigh() { return TryPopFrontHelper(2); } mutable std::mutex mutex_; private: std::atomic total_count_; std::deque priority_; std::deque queue_; MultiQueueWaiter* waiter_; std::unique_ptr owned_waiter_; // TODO remove waiter1 after split of on_indexed MultiQueueWaiter* waiter1_; std::unique_ptr owned_waiter1_; };