diff options
Diffstat (limited to 'graphics/asymptote/LspCpp/include/LibLsp/JsonRpc/threaded_queue.h')
-rw-r--r-- | graphics/asymptote/LspCpp/include/LibLsp/JsonRpc/threaded_queue.h | 249 |
1 files changed, 249 insertions, 0 deletions
diff --git a/graphics/asymptote/LspCpp/include/LibLsp/JsonRpc/threaded_queue.h b/graphics/asymptote/LspCpp/include/LibLsp/JsonRpc/threaded_queue.h new file mode 100644 index 0000000000..cca9f9a486 --- /dev/null +++ b/graphics/asymptote/LspCpp/include/LibLsp/JsonRpc/threaded_queue.h @@ -0,0 +1,249 @@ +#pragma once + +#include <boost/optional/optional.hpp> +#include <algorithm> +#include <atomic> +#include <condition_variable> +#include <deque> +#include <memory> +#include <mutex> +#include <tuple> +#include <utility> +#include <vector> + + +struct MultiQueueWaiter; + +struct BaseThreadQueue { + virtual ~BaseThreadQueue() = default; + + virtual bool IsEmpty() = 0; + + std::shared_ptr<MultiQueueWaiter> waiter; +}; + +// std::lock accepts two or more arguments. We define an overload for one +// argument. +namespace std { +template <typename Lockable> +void lock(Lockable& l) { + l.lock(); +} +} // namespace std + +template <typename... Queue> +struct MultiQueueLock { + MultiQueueLock(Queue... lockable) : tuple_{lockable...} { lock(); } + ~MultiQueueLock() { unlock(); } + void lock() { lock_impl(typename std::index_sequence_for<Queue...>{}); } + void unlock() { unlock_impl(typename std::index_sequence_for<Queue...>{}); } + + private: + template <size_t... Is> + void lock_impl(std::index_sequence<Is...>) { + std::lock(std::get<Is>(tuple_)->mutex...); + } + + template <size_t... Is> + void unlock_impl(std::index_sequence<Is...>) { + (void)std::initializer_list<int>{ + (std::get<Is>(tuple_)->mutex.unlock(), 0)...}; + } + + std::tuple<Queue...> tuple_; +}; + +struct MultiQueueWaiter { + static bool HasState(std::initializer_list<BaseThreadQueue*> queues); + + bool ValidateWaiter(std::initializer_list<BaseThreadQueue*> queues); + + template <typename... BaseThreadQueue> + bool Wait(std::atomic<bool>& quit, BaseThreadQueue... queues) { + MultiQueueLock<BaseThreadQueue...> l(queues...); + while (!quit.load(std::memory_order_relaxed)) { + if (HasState({ queues... })) + return false; + cv.wait(l); + } + return true; + } + template <typename... BaseThreadQueue> + void WaitUntil(std::chrono::steady_clock::time_point t, + BaseThreadQueue... queues) { + MultiQueueLock<BaseThreadQueue...> l(queues...); + if (!HasState({ queues... })) + cv.wait_until(l, t); + } + template <typename... BaseThreadQueue> + void Wait(BaseThreadQueue... queues) { + assert(ValidateWaiter({queues...})); + + MultiQueueLock<BaseThreadQueue...> l(queues...); + while (!HasState({queues...})) + cv.wait(l); + } + + std::condition_variable_any cv; +}; + +// A threadsafe-queue. http://stackoverflow.com/a/16075550 +template <class T> +struct ThreadedQueue : public BaseThreadQueue { + public: + ThreadedQueue() : ThreadedQueue(std::make_shared<MultiQueueWaiter>()) {} + + explicit ThreadedQueue(std::shared_ptr<MultiQueueWaiter> waiter) + : total_count_(0) { + this->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. + void Enqueue(T&& t, bool priority) { + { + std::lock_guard<std::mutex> lock(mutex); + if (priority) + priority_.push_back(std::move(t)); + else + queue_.push_back(std::move(t)); + ++total_count_; + } + waiter->cv.notify_one(); + } + + // Add a set of elements to the queue. + void EnqueueAll(std::vector<T>&& elements, bool priority) { + if (elements.empty()) + return; + + { + std::lock_guard<std::mutex> 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(); + } + + // 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<std::mutex> lock(mutex); + waiter->cv.wait(lock, + [&]() { return !priority_.empty() || !queue_.empty(); }); + + auto execute = [&](std::deque<T>* 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. + boost::optional<T> TryDequeue(bool priority) { + std::lock_guard<std::mutex> lock(mutex); + + auto pop = [&](std::deque<T>* q) { + auto val = std::move(q->front()); + q->pop_front(); + --total_count_; + return std::move(val); + }; + + auto get_result = [&](std::deque<T>* first, + std::deque<T>* second) -> boost::optional<T> { + if (!first->empty()) + return pop(first); + if (!second->empty()) + return pop(second); + return {}; + }; + + if (priority) + return get_result(&priority_, &queue_); + return get_result(&queue_, &priority_); + } + // Return all elements in the queue. + std::vector<T> DequeueAll() { + std::lock_guard<std::mutex> lock(mutex); + + total_count_ = 0; + + std::vector<T> 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; + } + std::vector<T> TryDequeueSome(size_t num) { + std::lock_guard<std::mutex> lock(mutex); + + std::vector<T> result; + num = std::min(num, priority_.size() + queue_.size()); + total_count_ -= num; + result.reserve(num); + while (num) + { + if(!priority_.empty()) { + result.emplace_back(std::move(priority_.front())); + priority_.pop_front(); + } + else + { + break; + } + num -= 1; + } + while (num) + { + if (!queue_.empty()) { + result.emplace_back(std::move(queue_.front())); + queue_.pop_front(); + } + else + { + break; + } + num -= 1; + } + return result; + } + template <typename Fn> + void Iterate(Fn fn) { + std::lock_guard<std::mutex> lock(mutex); + for (auto& entry : priority_) + fn(entry); + for (auto& entry : queue_) + fn(entry); + } + + mutable std::mutex mutex; + + private: + std::atomic<int> total_count_; + std::deque<T> priority_; + std::deque<T> queue_; +}; |