From ff10cceadf9f1cf44f2ff1639e59e536bafcae98 Mon Sep 17 00:00:00 2001 From: Arkadiusz Lachowicz Date: Mon, 13 Jul 2026 08:16:30 +0200 Subject: [PATCH 1/2] Add async queue teardown diagnostics --- include/nbl/system/IAsyncQueueDispatcher.h | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/include/nbl/system/IAsyncQueueDispatcher.h b/include/nbl/system/IAsyncQueueDispatcher.h index d5b0cb8a1a..50e49d755d 100644 --- a/include/nbl/system/IAsyncQueueDispatcher.h +++ b/include/nbl/system/IAsyncQueueDispatcher.h @@ -61,8 +61,13 @@ class IAsyncQueueDispatcherBase inline request_base_t() = default; inline ~request_base_t() { - // fully cleaned up - assert(!future); + const auto currentState = state.query(); + const bool hasNoFuture = future==nullptr; + const bool isInitial = currentState==STATE::INITIAL; + + // Dispatcher storage may only destroy fully recycled request slots. + assert(hasNoFuture); + assert(isInitial); } // ban certain operators @@ -488,7 +493,15 @@ class IAsyncQueueDispatcher : public IThreadHandler, pro } protected: - inline ~IAsyncQueueDispatcher() {} + inline ~IAsyncQueueDispatcher() + { + const auto begin = cb_begin.load(); + const auto end = cb_end.load(); + const bool isIdle = begin==end; + + // Request storage must not be destroyed while queued work is still pending. + assert(isIdle); + } inline void background_work() {} private: From 7fdb862ef4891987b1e26d0b38e82f482aeb15e7 Mon Sep 17 00:00:00 2001 From: Arkadiusz Lachowicz Date: Mon, 13 Jul 2026 09:35:52 +0200 Subject: [PATCH 2/2] Fix async queue teardown --- include/nbl/system/IAsyncQueueDispatcher.h | 67 ++++++++++++++++++++++ include/nbl/system/ISystem.h | 4 ++ include/nbl/system/IThreadHandler.h | 7 ++- src/nbl/ui/CWindowManagerWin32.h | 6 +- 4 files changed, 82 insertions(+), 2 deletions(-) diff --git a/include/nbl/system/IAsyncQueueDispatcher.h b/include/nbl/system/IAsyncQueueDispatcher.h index 50e49d755d..3ba3b6e5dc 100644 --- a/include/nbl/system/IAsyncQueueDispatcher.h +++ b/include/nbl/system/IAsyncQueueDispatcher.h @@ -444,6 +444,8 @@ class IAsyncQueueDispatcher : public IThreadHandler, pro request_t request_pool[MaxRequestCount]; atomic_counter_t cb_begin = 0u; atomic_counter_t cb_end = 0u; + std::atomic_bool m_acceptsSubmissions = true; + atomic_counter_t m_activeSubmissions = 0u; static inline counter_t wrapAround(counter_t x) { @@ -470,6 +472,10 @@ class IAsyncQueueDispatcher : public IThreadHandler, pro template void request(future_t* _future, Args&&... args) { + if (!beginSubmission()) + return; + auto submissionCleanup = core::makeRAIIExiter([this]() -> void { this->finishSubmission(); }); + // get next output index const auto virtualIx = cb_end++; // protect against overflow by waiting for the worker to catch up @@ -493,6 +499,17 @@ class IAsyncQueueDispatcher : public IThreadHandler, pro } protected: + inline void shutdown() + { + if (!beginShutdown()) + return; + + // accepted submissions must publish cb_end before drain observes the queue + waitForSubmissions(); + drain(); + base_t::stopThread(); + } + inline ~IAsyncQueueDispatcher() { const auto begin = cb_begin.load(); @@ -505,6 +522,56 @@ class IAsyncQueueDispatcher : public IThreadHandler, pro inline void background_work() {} private: + inline bool beginSubmission() + { + // register first, so shutdown can't miss a request already entering submission + m_activeSubmissions.fetch_add(1u); + + const bool acceptsRequests = m_acceptsSubmissions.load(); + assert(acceptsRequests); + if (!acceptsRequests) + { + finishSubmission(); + return false; + } + return true; + } + + inline void finishSubmission() + { + const auto previousSubmissions = m_activeSubmissions.fetch_sub(1u); + assert(previousSubmissions!=0u); + m_activeSubmissions.notify_all(); + } + + inline bool beginShutdown() + { + return m_acceptsSubmissions.exchange(false); + } + + inline void waitForSubmissions() + { + for (auto submissions=m_activeSubmissions.load(); submissions!=0u; submissions=m_activeSubmissions.load()) + m_activeSubmissions.wait(submissions); + } + + inline void drain() + { + while (true) + { + const auto begin = cb_begin.load(); + const auto end = cb_end.load(); + if (begin==end) + return; + + { + auto global_lk = base_t::createLock(); + base_t::m_cvar.notify_one(); + } + cb_begin.wait(begin); + } + } + template void work(lock_t& lock, Args&&... optional_internal_state) { diff --git a/include/nbl/system/ISystem.h b/include/nbl/system/ISystem.h index 36c19f1961..fe40188153 100644 --- a/include/nbl/system/ISystem.h +++ b/include/nbl/system/ISystem.h @@ -274,6 +274,10 @@ class NBL_API2 ISystem : public core::IReferenceCounted { //waitForInitComplete(); init is a NOOP } + inline ~CAsyncQueue() + { + this->shutdown(); + } void process_request(base_t::future_base_t* _future_base, SRequestType& req); diff --git a/include/nbl/system/IThreadHandler.h b/include/nbl/system/IThreadHandler.h index f069025edc..1bcfaa82d2 100644 --- a/include/nbl/system/IThreadHandler.h +++ b/include/nbl/system/IThreadHandler.h @@ -165,6 +165,11 @@ class IThreadHandler } protected: + void stopThread() + { + terminate(); + } + void thread() { CRTP* this_ = static_cast(this); @@ -219,4 +224,4 @@ class IThreadHandler } -#endif \ No newline at end of file +#endif diff --git a/src/nbl/ui/CWindowManagerWin32.h b/src/nbl/ui/CWindowManagerWin32.h index 93ffb67071..0c422f9bf0 100644 --- a/src/nbl/ui/CWindowManagerWin32.h +++ b/src/nbl/ui/CWindowManagerWin32.h @@ -202,6 +202,10 @@ class NBL_API2 CWindowManagerWin32 final : public IWindowManagerWin32, public IC { //waitForInitComplete(); init is a NOOP } + inline ~CAsyncQueue() + { + this->shutdown(); + } inline void init() {} @@ -216,4 +220,4 @@ class NBL_API2 CWindowManagerWin32 final : public IWindowManagerWin32, public IC } #endif -#endif \ No newline at end of file +#endif