/* This Source Code Form is subject to the terms of the Mozilla Public * License, v. 2.0. If a copy of the MPL was not distributed with this * file, You can obtain one at https://mozilla.org/MPL/2.0/. * * (c) ZeroTier, Inc. * https://www.zerotier.com/ */ #ifndef ZT_BLOCKINGQUEUE_HPP #define ZT_BLOCKINGQUEUE_HPP #include #include #include #include #include #include namespace ZeroTier { /** * Simple C++11 thread-safe queue * * Do not use in node/ since we have not gone C++11 there yet. */ template class BlockingQueue { public: BlockingQueue(void) : r(true) { } inline void post(T t) { std::lock_guard lock(m); q.push(t); c.notify_one(); } inline void postLimit(T t, const unsigned long limit) { std::unique_lock lock(m); for (;;) { if (q.size() < limit) { q.push(t); c.notify_one(); break; } if (! r) break; gc.wait(lock); } } inline void stop(void) { std::lock_guard lock(m); r = false; c.notify_all(); gc.notify_all(); } inline bool get(T& value) { std::unique_lock lock(m); if (! r) return false; while (q.empty()) { c.wait(lock); if (! r) { gc.notify_all(); return false; } } value = q.front(); q.pop(); gc.notify_all(); return true; } inline std::vector drain() { std::vector v; while (! q.empty()) { v.push_back(q.front()); q.pop(); } return v; } enum TimedWaitResult { OK, TIMED_OUT, STOP }; inline TimedWaitResult get(T& value, const unsigned long ms) { const std::chrono::milliseconds ms2 { ms }; std::unique_lock lock(m); if (! r) return STOP; while (q.empty()) { if (c.wait_for(lock, ms2) == std::cv_status::timeout) return ((r) ? TIMED_OUT : STOP); else if (! r) return STOP; } value = q.front(); q.pop(); return OK; } inline size_t size() const { return q.size(); } private: std::queue q; mutable std::mutex m; mutable std::condition_variable c, gc; std::atomic_bool r; }; } // namespace ZeroTier #endif