#pragma once #include #include #include #include #include #include // TODO: cleanup includes. struct BaseThreadQueue { virtual bool IsEmpty() = 0; }; struct MultiQueueWaiter { std::mutex m; std::condition_variable cv; bool HasState(std::initializer_list queues) { for (BaseThreadQueue* queue : queues) { if (!queue->IsEmpty()) return true; } return false; } void Wait(std::initializer_list queues) { // We cannot have a single condition variable wait on all of the different // mutexes, so we have a global condition variable that every queue will // notify. After it is notified we check to see if any of the queues have // data; if they do, we return. // // We repoll every 5 seconds because it's not possible to atomically check // the state of every queue and then setup the condition variable. So, if // Wait() is called, HasState() returns false, and then in the time after // HasState() is called data gets posted but before we begin waiting for // the condition variable, we will miss the notification. The timeout of 5 // means that if this happens we will delay operation for 5 seconds. while (!HasState(queues)) { std::unique_lock l(m); cv.wait_for(l, std::chrono::seconds(5)); } } }; // A threadsafe-queue. http://stackoverflow.com/a/16075550 template struct ThreadedQueue : public BaseThreadQueue { public: ThreadedQueue(MultiQueueWaiter* waiter) : waiter_(waiter) {} // Add an element to the front of the queue. void PriorityEnqueue(T&& t) { std::lock_guard lock(mutex_); priority_.push(std::move(t)); waiter_->cv.notify_all(); } // Add an element to the queue. void Enqueue(T&& t) { std::lock_guard lock(mutex_); queue_.push(std::move(t)); waiter_->cv.notify_all(); } // Return all elements in the queue. std::vector DequeueAll() { std::lock_guard lock(mutex_); std::vector result; result.reserve(priority_.size() + queue_.size()); while (!priority_.empty()) { result.emplace_back(std::move(priority_.front())); priority_.pop(); } while (!queue_.empty()) { result.emplace_back(std::move(queue_.front())); queue_.pop(); } return result; } bool IsEmpty() { std::lock_guard lock(mutex_); return queue_.empty(); } /* // Get the "front"-element. // If the queue is empty, wait untill an element is avaiable. T Dequeue() { std::unique_lock lock(mutex_); while (priority_.empty() && queue_.empty()) { // release lock as long as the wait and reaquire it afterwards. cv_.wait(lock); } if (!priority_.empty()) { auto val = std::move(priority_.front()); priority_.pop(); return val; } auto val = std::move(queue_.front()); queue_.pop(); return val; } */ // Get the first element from the queue without blocking. Returns a null // value if the queue is empty. optional TryDequeue() { std::lock_guard lock(mutex_); if (priority_.empty() && queue_.empty()) return nullopt; if (!priority_.empty()) { auto val = std::move(priority_.front()); priority_.pop(); return std::move(val); } auto val = std::move(queue_.front()); queue_.pop(); return std::move(val); } private: std::queue priority_; mutable std::mutex mutex_; std::queue queue_; MultiQueueWaiter* waiter_; };