// Copyright 2017-2018 ccls Authors // SPDX-License-Identifier: Apache-2.0 #pragma once #include "utils.hh" #include #include #include #include #include #include #include // 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 namespace ccls { struct BaseThreadQueue { virtual bool IsEmpty() = 0; virtual ~BaseThreadQueue() = default; }; template struct MultiQueueLock { MultiQueueLock(Queue... lockable) : tuple_{lockable...} { lock(); } ~MultiQueueLock() { unlock(); } void lock() { lock_impl(typename std::index_sequence_for{}); } void unlock() { unlock_impl(typename std::index_sequence_for{}); } private: template void lock_impl(std::index_sequence) { std::lock(std::get(tuple_)->mutex_...); } template void unlock_impl(std::index_sequence) { (std::get(tuple_)->mutex_.unlock(), ...); } 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() { owned_waiter_ = std::make_unique(); waiter_ = owned_waiter_.get(); } explicit ThreadedQueue(MultiQueueWaiter *waiter) : waiter_(waiter) {} // 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(); } void PushBack(T &&t, bool priority = false) { Push<&std::deque::push_back>(std::move(t), priority); } // 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. std::optional TryPopFront() { 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 (priority_.size()) return execute(&priority_); if (queue_.size()) return execute(&queue_); return std::nullopt; } template void Iterate(Fn fn) { std::lock_guard lock(mutex_); for (auto &entry : priority_) fn(entry); for (auto &entry : queue_) fn(entry); } mutable std::mutex mutex_; private: std::atomic total_count_{0}; std::deque priority_; std::deque queue_; MultiQueueWaiter *waiter_; std::unique_ptr owned_waiter_; }; } // namespace ccls