diff options
| -rw-r--r-- | zencore/include/zencore/blockingqueue.h | 73 | ||||
| -rw-r--r-- | zencore/include/zencore/refcount.h | 9 | ||||
| -rw-r--r-- | zenhttp/iothreadpool.cpp | 6 | ||||
| -rw-r--r-- | zenhttp/iothreadpool.h | 8 | ||||
| -rw-r--r-- | zenserver/upstream/upstreamcache.cpp | 68 |
5 files changed, 96 insertions, 68 deletions
diff --git a/zencore/include/zencore/blockingqueue.h b/zencore/include/zencore/blockingqueue.h new file mode 100644 index 000000000..fbb4480a1 --- /dev/null +++ b/zencore/include/zencore/blockingqueue.h @@ -0,0 +1,73 @@ +#pragma once + +#include <atomic> +#include <deque> +#include <mutex> + +namespace zen { + +template<typename T> +class BlockingQueue +{ +public: + BlockingQueue() = default; + + ~BlockingQueue() { CompleteAdding(); } + + void Enqueue(T&& Item) + { + { + std::lock_guard Lock(m_Lock); + m_Queue.emplace_back(std::move(Item)); + m_Size++; + } + + m_NewItemSignal.notify_one(); + } + + bool WaitAndDequeue(T& Item) + { + if (m_CompleteAdding.load()) + { + return false; + } + + std::unique_lock Lock(m_Lock); + m_NewItemSignal.wait(Lock, [this]() { return !m_Queue.empty() || m_CompleteAdding.load(); }); + + if (!m_Queue.empty()) + { + Item = std::move(m_Queue.front()); + m_Queue.pop_front(); + m_Size--; + + return true; + } + + return false; + } + + void CompleteAdding() + { + if (!m_CompleteAdding.load()) + { + m_CompleteAdding.store(true); + m_NewItemSignal.notify_all(); + } + } + + std::size_t Size() const + { + std::unique_lock Lock(m_Lock); + return m_Queue.size(); + } + +private: + mutable std::mutex m_Lock; + std::condition_variable m_NewItemSignal; + std::deque<T> m_Queue; + std::atomic_bool m_CompleteAdding{false}; + std::atomic_uint32_t m_Size; +}; + +} // namespace zen diff --git a/zencore/include/zencore/refcount.h b/zencore/include/zencore/refcount.h index 320718f5b..7167ab3b5 100644 --- a/zencore/include/zencore/refcount.h +++ b/zencore/include/zencore/refcount.h @@ -4,6 +4,8 @@ #include "atomic.h" #include "zencore.h" +#include <concepts> + namespace zen { /** @@ -114,6 +116,10 @@ public: inline Ref(T* Ptr) : m_Ref(Ptr) { m_Ref && m_Ref->AddRef(); } inline ~Ref() { m_Ref && m_Ref->Release(); } + template<typename DerivedType> + requires std::derived_from<DerivedType, T> + inline Ref(const Ref<DerivedType>& Rhs) : Ref(Rhs.m_Ref) {} + [[nodiscard]] inline bool IsNull() const { return m_Ref == nullptr; } inline explicit operator bool() const { return m_Ref != nullptr; } inline T* operator->() const { return m_Ref; } @@ -152,6 +158,9 @@ public: private: T* m_Ref = nullptr; + + template<class T> + friend class Ref; }; void refcount_forcelink(); diff --git a/zenhttp/iothreadpool.cpp b/zenhttp/iothreadpool.cpp index 4f1a6642b..6087e69ec 100644 --- a/zenhttp/iothreadpool.cpp +++ b/zenhttp/iothreadpool.cpp @@ -4,6 +4,8 @@ #include <zencore/except.h> +#if ZEN_PLATFORM_WINDOWS + namespace zen { WinIoThreadPool::WinIoThreadPool(int InThreadCount) @@ -32,6 +34,8 @@ WinIoThreadPool::~WinIoThreadPool() void WinIoThreadPool::CreateIocp(HANDLE IoHandle, PTP_WIN32_IO_CALLBACK Callback, void* Context, std::error_code& ErrorCode) { + ZEN_ASSERT(!m_ThreadPoolIo); + m_ThreadPoolIo = CreateThreadpoolIo(IoHandle, Callback, Context, &m_CallbackEnvironment); if (!m_ThreadPoolIo) @@ -41,3 +45,5 @@ WinIoThreadPool::CreateIocp(HANDLE IoHandle, PTP_WIN32_IO_CALLBACK Callback, voi } } // namespace zen + +#endif diff --git a/zenhttp/iothreadpool.h b/zenhttp/iothreadpool.h index 4418b940b..8333964c3 100644 --- a/zenhttp/iothreadpool.h +++ b/zenhttp/iothreadpool.h @@ -2,9 +2,12 @@ #pragma once -#include <zencore/windows.h> +#include <zencore/zencore.h> -#include <system_error> +#if ZEN_PLATFORM_WINDOWS +# include <zencore/windows.h> + +# include <system_error> namespace zen { @@ -31,3 +34,4 @@ private: }; } // namespace zen +#endif diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp index a183e7ebf..ba5991b3f 100644 --- a/zenserver/upstream/upstreamcache.cpp +++ b/zenserver/upstream/upstreamcache.cpp @@ -8,6 +8,7 @@ #include <zencore/compactbinarybuilder.h> #include <zencore/compactbinarypackage.h> #include <zencore/compactbinaryvalidation.h> +#include <zencore/blockingqueue.h> #include <zencore/fmtutils.h> #include <zencore/stats.h> #include <zencore/stream.h> @@ -23,7 +24,6 @@ #include <algorithm> #include <atomic> -#include <deque> #include <thread> #include <unordered_map> @@ -33,70 +33,6 @@ using namespace std::literals; namespace detail { - template<typename T> - class BlockingQueue - { - public: - BlockingQueue() = default; - - ~BlockingQueue() { CompleteAdding(); } - - void Enqueue(T&& Item) - { - { - std::lock_guard Lock(m_Lock); - m_Queue.emplace_back(std::move(Item)); - m_Size++; - } - - m_NewItemSignal.notify_one(); - } - - bool WaitAndDequeue(T& Item) - { - if (m_CompleteAdding.load()) - { - return false; - } - - std::unique_lock Lock(m_Lock); - m_NewItemSignal.wait(Lock, [this]() { return !m_Queue.empty() || m_CompleteAdding.load(); }); - - if (!m_Queue.empty()) - { - Item = std::move(m_Queue.front()); - m_Queue.pop_front(); - m_Size--; - - return true; - } - - return false; - } - - void CompleteAdding() - { - if (!m_CompleteAdding.load()) - { - m_CompleteAdding.store(true); - m_NewItemSignal.notify_all(); - } - } - - std::size_t Size() const - { - std::unique_lock Lock(m_Lock); - return m_Queue.size(); - } - - private: - mutable std::mutex m_Lock; - std::condition_variable m_NewItemSignal; - std::deque<T> m_Queue; - std::atomic_bool m_CompleteAdding{false}; - std::atomic_uint32_t m_Size; - }; - class JupiterUpstreamEndpoint final : public UpstreamEndpoint { public: @@ -1029,7 +965,7 @@ private: spdlog::logger& Log() { return m_Log; } - using UpstreamQueue = detail::BlockingQueue<UpstreamCacheRecord>; + using UpstreamQueue = BlockingQueue<UpstreamCacheRecord>; struct RunState { |