summaryrefslogtreecommitdiff
path: root/graphics/asymptote/LspCpp/include/LibLsp/JsonRpc/threaded_queue.h
diff options
context:
space:
mode:
Diffstat (limited to 'graphics/asymptote/LspCpp/include/LibLsp/JsonRpc/threaded_queue.h')
-rw-r--r--graphics/asymptote/LspCpp/include/LibLsp/JsonRpc/threaded_queue.h493
1 files changed, 282 insertions, 211 deletions
diff --git a/graphics/asymptote/LspCpp/include/LibLsp/JsonRpc/threaded_queue.h b/graphics/asymptote/LspCpp/include/LibLsp/JsonRpc/threaded_queue.h
index e8fc6d13ea..9680058370 100644
--- a/graphics/asymptote/LspCpp/include/LibLsp/JsonRpc/threaded_queue.h
+++ b/graphics/asymptote/LspCpp/include/LibLsp/JsonRpc/threaded_queue.h
@@ -15,236 +15,307 @@
struct MultiQueueWaiter;
-struct BaseThreadQueue {
- virtual ~BaseThreadQueue() = default;
+struct BaseThreadQueue
+{
+ virtual ~BaseThreadQueue() = default;
- virtual bool IsEmpty() = 0;
+ virtual bool IsEmpty() = 0;
- std::shared_ptr<MultiQueueWaiter> waiter;
+ 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 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_;
+} // 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;
+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>()) {}
+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;
- }
+ 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_; }
+ // 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) {
+ // 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_;
+ {
+ 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();
}
- waiter->cv.notify_one();
- }
- // Add a set of elements to the queue.
- void EnqueueAll(std::vector<T>&& elements, bool priority) {
- if (elements.empty())
- return;
+ // 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::lock_guard<std::mutex> lock(mutex);
- total_count_ += elements.size();
- for (T& element : elements) {
+ 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.
+ 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) -> optional<T>
+ {
+ if (!first->empty())
+ {
+ return pop(first);
+ }
+ if (!second->empty())
+ {
+ return pop(second);
+ }
+ return {};
+ };
+
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.
- 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) -> 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_;
+ {
+ 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_;
};