From 34e6adf8adedca18725407ab41cf7bc51c0dc5ce Mon Sep 17 00:00:00 2001 From: jhugard Date: Wed, 6 May 2026 14:06:20 -0700 Subject: [PATCH 1/6] Compile existing task queue test hooks only in unit-test builds --- Source/Task/TaskQueue.cpp | 6 +++++- Source/Task/TaskQueueImpl.h | 4 ++++ Source/Task/TaskQueueP.h | 2 ++ Source/Task/XTaskQueuePriv.h | 4 +++- 4 files changed, 14 insertions(+), 2 deletions(-) diff --git a/Source/Task/TaskQueue.cpp b/Source/Task/TaskQueue.cpp index 8fed26dc7..0dc4fbffd 100644 --- a/Source/Task/TaskQueue.cpp +++ b/Source/Task/TaskQueue.cpp @@ -1118,6 +1118,7 @@ bool TaskQueuePortImpl::ScheduleNextPendingCallback( // See VerifyDelayedCallbackTimerRaceOnManualQueue for full // analysis. The test hook here allows unit tests to verify // there is no race. +#ifdef HC_UNITTEST_API m_attachedContexts.Visit([&](ITaskQueuePortContext* portContext) { auto hooks = portContext->GetQueue()->GetTestHooks(); @@ -1128,6 +1129,7 @@ bool TaskQueuePortImpl::ScheduleNextPendingCallback( noDueTime); } }); +#endif } } @@ -2341,6 +2343,7 @@ STDAPI_(bool) XTaskQueueUninitialize( return ApiRefs::WaitZeroRefs(timeoutMilliseconds); } +#ifdef HC_UNITTEST_API /// /// Sets or clears test hooks on a task queue. /// @@ -2353,4 +2356,5 @@ STDAPI XTaskQueueSetTestHooks( RETURN_HR_IF(E_GAMERUNTIME_INVALID_HANDLE, aq == nullptr); aq->SetTestHooks(hooks); return S_OK; -} \ No newline at end of file +} +#endif diff --git a/Source/Task/TaskQueueImpl.h b/Source/Task/TaskQueueImpl.h index 7e4bfa5bf..57966f737 100644 --- a/Source/Task/TaskQueueImpl.h +++ b/Source/Task/TaskQueueImpl.h @@ -402,8 +402,10 @@ class TaskQueueImpl : public Api _In_ XTaskQueuePortHandle completionPort); XTaskQueueHandle __stdcall GetHandle() override { return &m_header; } +#ifdef HC_UNITTEST_API XTaskQueueTestHooks* __stdcall GetTestHooks() override { return m_testHooks; } void __stdcall SetTestHooks(_In_ XTaskQueueTestHooks* testHooks) override { m_testHooks = testHooks; } +#endif HRESULT __stdcall GetPortContext( _In_ XTaskQueuePort port, @@ -475,7 +477,9 @@ class TaskQueueImpl : public Api TerminationData m_termination; TaskQueuePortContextImpl m_work; TaskQueuePortContextImpl m_completion; +#ifdef HC_UNITTEST_API XTaskQueueTestHooks* m_testHooks = nullptr; +#endif #ifdef SUSPEND_API SuspendResumeHandler m_suspendHandler; diff --git a/Source/Task/TaskQueueP.h b/Source/Task/TaskQueueP.h index 3220f50aa..0f70d3c7c 100644 --- a/Source/Task/TaskQueueP.h +++ b/Source/Task/TaskQueueP.h @@ -125,8 +125,10 @@ struct ITaskQueuePortContext : IApi struct ITaskQueue : IApi { virtual XTaskQueueHandle __stdcall GetHandle() = 0; +#ifdef HC_UNITTEST_API virtual XTaskQueueTestHooks* __stdcall GetTestHooks() = 0; virtual void __stdcall SetTestHooks(_In_ XTaskQueueTestHooks* testHooks) = 0; +#endif virtual HRESULT __stdcall GetPortContext( _In_ XTaskQueuePort port, diff --git a/Source/Task/XTaskQueuePriv.h b/Source/Task/XTaskQueuePriv.h index 82cbdac0b..d20569bad 100644 --- a/Source/Task/XTaskQueuePriv.h +++ b/Source/Task/XTaskQueuePriv.h @@ -40,12 +40,13 @@ STDAPI_(void) XTaskQueueResumeTermination( _In_ XTaskQueueHandle queue ) noexcept; +#ifdef HC_UNITTEST_API /// /// This structure can be passed as a pointer to the task queue so unit tests /// can hook into its behavior. Some race conditions are very difficult to get /// to happen naturally so sometimes a hook is needed. A pointer to this /// structure will be stored on the task queue. It is up to the test to ensure -/// the structure lifetime exceeds that of the task queue under test. +/// the structure lifetime exceeds that of the task queue under test. /// struct XTaskQueueTestHooks { @@ -67,6 +68,7 @@ STDAPI XTaskQueueSetTestHooks( _In_ XTaskQueueHandle queue, _In_ XTaskQueueTestHooks* hooks ) noexcept; +#endif //----------------------------------------------------------------// // From a9ffe0f8175118479968a3bf90fdcddc89923bb0 Mon Sep 17 00:00:00 2001 From: jhugard Date: Wed, 6 May 2026 14:08:00 -0700 Subject: [PATCH 2/6] Add regression for early promotion during queue termination --- Source/Task/TaskQueue.cpp | 10 ++ Source/Task/XTaskQueuePriv.h | 6 + Tests/UnitTests/Tests/TaskQueueTests.cpp | 145 +++++++++++++++++++++++ 3 files changed, 161 insertions(+) diff --git a/Source/Task/TaskQueue.cpp b/Source/Task/TaskQueue.cpp index 0dc4fbffd..4855d01b4 100644 --- a/Source/Task/TaskQueue.cpp +++ b/Source/Task/TaskQueue.cpp @@ -962,6 +962,16 @@ void TaskQueuePortImpl::CancelPendingEntries( m_timer.Cancel(); m_timerDue = UINT64_MAX; + #ifdef HC_UNITTEST_API + // Test hook: let unit tests enqueue a sibling delayed callback after the + // due time has been reset but before the old SubmitPendingCallback() + // rescan treats the sibling entry as immediately due. + if (auto hooks = portContext->GetQueue()->GetTestHooks(); hooks != nullptr) + { + hooks->PendingEntriesRemovedDuringTermination(portContext->GetType()); + } + #endif + m_pendingList->remove_if([&](auto& entry, auto address) { if (entry.portContext == portContext) diff --git a/Source/Task/XTaskQueuePriv.h b/Source/Task/XTaskQueuePriv.h index d20569bad..c87cdc3c6 100644 --- a/Source/Task/XTaskQueuePriv.h +++ b/Source/Task/XTaskQueuePriv.h @@ -50,6 +50,12 @@ STDAPI_(void) XTaskQueueResumeTermination( /// struct XTaskQueueTestHooks { + virtual void PendingEntriesRemovedDuringTermination( + XTaskQueuePort port) + { + UNREFERENCED_PARAMETER(port); + } + virtual void NextPendingCallbackScheduled( XTaskQueuePort port, uint64_t lastDueTime, diff --git a/Tests/UnitTests/Tests/TaskQueueTests.cpp b/Tests/UnitTests/Tests/TaskQueueTests.cpp index 925c1b757..09e5ca247 100644 --- a/Tests/UnitTests/Tests/TaskQueueTests.cpp +++ b/Tests/UnitTests/Tests/TaskQueueTests.cpp @@ -2055,4 +2055,149 @@ DEFINE_TEST_CLASS(TaskQueueTests) VERIFY_IS_TRUE(canaryFired.load()); } + + DEFINE_TEST_CASE(VerifyTerminationDoesNotEarlyPromoteSiblingDelayedCallback) + { + using TestClock = std::chrono::steady_clock; + + struct TestBarrier + { + std::mutex mtx; + std::condition_variable cv; + bool phase1_ready = false; // terminate thread -> test thread + bool phase2_ready = false; // test thread -> terminate thread + }; + + struct TestHooks : public XTaskQueueTestHooks + { + explicit TestHooks(_In_ TestBarrier* barrier) : m_testBarrier(barrier) {} + + void PendingEntriesRemovedDuringTermination(XTaskQueuePort port) override + { + UNREFERENCED_PARAMETER(port); + + { + std::lock_guard lk(m_testBarrier->mtx); + m_testBarrier->phase1_ready = true; + } + m_testBarrier->cv.notify_all(); + + std::unique_lock lk(m_testBarrier->mtx); + m_testBarrier->cv.wait_for(lk, std::chrono::seconds(5), + [&] { return m_testBarrier->phase2_ready; }); + } + + private: + TestBarrier* m_testBarrier = nullptr; + }; + + struct CallbackState + { + std::atomic invoked{ false }; + std::atomic canceled{ false }; + std::atomic firedAtNs{ 0 }; + + static int64_t NowNs() + { + return std::chrono::duration_cast( + TestClock::now().time_since_epoch()).count(); + } + + static void CALLBACK Invoke(void* ctx, bool canceled) + { + auto* self = static_cast(ctx); + self->canceled.store(canceled, std::memory_order_release); + self->firedAtNs.store(NowNs(), std::memory_order_release); + self->invoked.store(true, std::memory_order_release); + } + }; + + AutoQueueHandle root; + VERIFY_SUCCEEDED(XTaskQueueCreate( + XTaskQueueDispatchMode::Manual, + XTaskQueueDispatchMode::Manual, + &root)); + + XTaskQueuePortHandle rootPort; + VERIFY_SUCCEEDED(XTaskQueueGetPort(root, XTaskQueuePort::Work, &rootPort)); + + AutoQueueHandle terminatingQueue; + AutoQueueHandle siblingQueue; + VERIFY_SUCCEEDED(XTaskQueueCreateComposite(rootPort, rootPort, &terminatingQueue)); + VERIFY_SUCCEEDED(XTaskQueueCreateComposite(rootPort, rootPort, &siblingQueue)); + + TestBarrier barrier; + TestHooks hooks(&barrier); + VERIFY_SUCCEEDED(XTaskQueueSetTestHooks(terminatingQueue, &hooks)); + + CallbackState terminatingState; + CallbackState siblingState; + + // Arm the shared timer with a deadline owned by the queue that will be + // terminated. The sibling callback is submitted while termination is + // blocked in the hook above. + VERIFY_SUCCEEDED(XTaskQueueSubmitDelayedCallback( + terminatingQueue, + XTaskQueuePort::Work, + 150, + &terminatingState, + &CallbackState::Invoke)); + + HRESULT terminateHr = S_OK; + std::thread terminator([&] + { + terminateHr = XTaskQueueTerminate(terminatingQueue, false, nullptr, nullptr); + }); + + { + std::unique_lock lk(barrier.mtx); + bool ok = barrier.cv.wait_for(lk, std::chrono::seconds(5), + [&] { return barrier.phase1_ready; }); + VERIFY_IS_TRUE(ok); + } + + constexpr int64_t siblingDelayNs = 300LL * 1000LL * 1000LL; + const int64_t siblingSubmittedAtNs = CallbackState::NowNs(); + VERIFY_SUCCEEDED(XTaskQueueSubmitDelayedCallback( + siblingQueue, + XTaskQueuePort::Work, + 300, + &siblingState, + &CallbackState::Invoke)); + + { + std::lock_guard lk(barrier.mtx); + barrier.phase2_ready = true; + } + barrier.cv.notify_all(); + + terminator.join(); + VERIFY_SUCCEEDED(terminateHr); + VERIFY_SUCCEEDED(XTaskQueueSetTestHooks(terminatingQueue, nullptr)); + + // The terminating queue's delayed callback is expected to surface as + // canceled. The sibling delayed callback must stay pending until its + // own deadline instead of being promoted immediately. + while (XTaskQueueDispatch(root, XTaskQueuePort::Work, 0)) + { + } + + VERIFY_IS_TRUE(terminatingState.invoked.load(std::memory_order_acquire)); + VERIFY_IS_TRUE(terminatingState.canceled.load(std::memory_order_acquire)); + VERIFY_IS_FALSE(siblingState.invoked.load(std::memory_order_acquire)); + + VERIFY_IS_FALSE(XTaskQueueDispatch(root, XTaskQueuePort::Work, 150)); + VERIFY_IS_TRUE(XTaskQueueDispatch(root, XTaskQueuePort::Work, 2000)); + + VERIFY_IS_TRUE(siblingState.invoked.load(std::memory_order_acquire)); + VERIFY_IS_FALSE(siblingState.canceled.load(std::memory_order_acquire)); + VERIFY_IS_GREATER_THAN_OR_EQUAL( + siblingState.firedAtNs.load(std::memory_order_acquire) - siblingSubmittedAtNs, + siblingDelayNs); + + XTaskQueueTerminate(siblingQueue, false, nullptr, nullptr); + while (XTaskQueueDispatch(root, XTaskQueuePort::Work, 0)) + { + } + } }; From 16d5a91d14370995c08001fcc03a5d62914b496a Mon Sep 17 00:00:00 2001 From: jhugard Date: Wed, 6 May 2026 14:08:42 -0700 Subject: [PATCH 3/6] Avoid early promotion when terminating shared task queues --- Source/Task/TaskQueue.cpp | 30 +++++++++++++----------------- 1 file changed, 13 insertions(+), 17 deletions(-) diff --git a/Source/Task/TaskQueue.cpp b/Source/Task/TaskQueue.cpp index 4855d01b4..c87f3b16d 100644 --- a/Source/Task/TaskQueue.cpp +++ b/Source/Task/TaskQueue.cpp @@ -955,22 +955,10 @@ void TaskQueuePortImpl::CancelPendingEntries( _In_ ITaskQueuePortContext* portContext, _In_ bool appendToQueue) { - // Stop wait timer and promote pending callbacks that are used - // by the queue that invoked this termination. Other callbacks - // are placed back on the pending list. - - m_timer.Cancel(); - m_timerDue = UINT64_MAX; - - #ifdef HC_UNITTEST_API - // Test hook: let unit tests enqueue a sibling delayed callback after the - // due time has been reset but before the old SubmitPendingCallback() - // rescan treats the sibling entry as immediately due. - if (auto hooks = portContext->GetQueue()->GetTestHooks(); hooks != nullptr) - { - hooks->PendingEntriesRemovedDuringTermination(portContext->GetType()); - } - #endif + // Only move entries owned by the terminating queue. Sibling delegates + // share this port's delayed-callback timer state, so leave m_timer and + // m_timerDue alone; if we removed the armed earliest entry, the existing + // timer simply takes one blank fire and re-arms for the next real item. m_pendingList->remove_if([&](auto& entry, auto address) { @@ -988,7 +976,15 @@ void TaskQueuePortImpl::CancelPendingEntries( return false; }); - SubmitPendingCallback(); +#ifdef HC_UNITTEST_API + // Test hook: let unit tests enqueue a sibling delayed callback while this + // termination path still owns the interleaving window that used to race + // with SubmitPendingCallback(). + if (auto hooks = portContext->GetQueue()->GetTestHooks(); hooks != nullptr) + { + hooks->PendingEntriesRemovedDuringTermination(portContext->GetType()); + } +#endif #ifdef _WIN32 From d690f7a7782a89443604e20b79820326ec750223 Mon Sep 17 00:00:00 2001 From: jhugard Date: Fri, 8 May 2026 20:27:44 -0700 Subject: [PATCH 4/6] Add stale delayed callback repro --- Source/Task/TaskQueue.cpp | 17 +++++ Source/Task/TaskQueueImpl.h | 7 ++ Source/Task/XTaskQueuePriv.h | 10 +++ Tests/UnitTests/Tests/TaskQueueTests.cpp | 81 ++++++++++++++++++++++++ 4 files changed, 115 insertions(+) diff --git a/Source/Task/TaskQueue.cpp b/Source/Task/TaskQueue.cpp index c87f3b16d..ae59f132a 100644 --- a/Source/Task/TaskQueue.cpp +++ b/Source/Task/TaskQueue.cpp @@ -2363,4 +2363,21 @@ STDAPI XTaskQueueSetTestHooks( aq->SetTestHooks(hooks); return S_OK; } + +STDAPI XTaskQueueSubmitPendingCallbackForTests( + _In_ XTaskQueueHandle queue, + _In_ XTaskQueuePort port + ) noexcept +{ + referenced_ptr aq(GetQueue(queue)); + RETURN_HR_IF(E_GAMERUNTIME_INVALID_HANDLE, aq == nullptr); + + referenced_ptr portContext; + RETURN_IF_FAILED(aq->GetPortContext(port, portContext.address_of())); + + auto* portImpl = static_cast(portContext->GetPort()); + portImpl->SubmitPendingCallbackForTests(); + return S_OK; +} #endif + diff --git a/Source/Task/TaskQueueImpl.h b/Source/Task/TaskQueueImpl.h index 57966f737..98c6755d2 100644 --- a/Source/Task/TaskQueueImpl.h +++ b/Source/Task/TaskQueueImpl.h @@ -215,6 +215,13 @@ class TaskQueuePortImpl: public Api void __stdcall SuspendPort(); void __stdcall ResumePort(); +#ifdef HC_UNITTEST_API + void __stdcall SubmitPendingCallbackForTests() + { + SubmitPendingCallback(); + } +#endif + private: struct WaitRegistration; diff --git a/Source/Task/XTaskQueuePriv.h b/Source/Task/XTaskQueuePriv.h index c87cdc3c6..3c1392e66 100644 --- a/Source/Task/XTaskQueuePriv.h +++ b/Source/Task/XTaskQueuePriv.h @@ -74,6 +74,16 @@ STDAPI XTaskQueueSetTestHooks( _In_ XTaskQueueHandle queue, _In_ XTaskQueueTestHooks* hooks ) noexcept; + +/// +/// Directly invokes the delayed-callback notification path for unit tests. +/// This is used to model stale threadpool timer callbacks that were already +/// queued before the timer was retargeted. +/// +STDAPI XTaskQueueSubmitPendingCallbackForTests( + _In_ XTaskQueueHandle queue, + _In_ XTaskQueuePort port + ) noexcept; #endif //----------------------------------------------------------------// diff --git a/Tests/UnitTests/Tests/TaskQueueTests.cpp b/Tests/UnitTests/Tests/TaskQueueTests.cpp index 09e5ca247..44480d4b5 100644 --- a/Tests/UnitTests/Tests/TaskQueueTests.cpp +++ b/Tests/UnitTests/Tests/TaskQueueTests.cpp @@ -2200,4 +2200,85 @@ DEFINE_TEST_CLASS(TaskQueueTests) { } } + + DEFINE_TEST_CASE(VerifyStaleDelayedCallbackDoesNotEarlyPromoteNextPendingEntry) + { + using TestClock = std::chrono::steady_clock; + + struct CallbackState + { + std::atomic invoked{ false }; + std::atomic canceled{ false }; + std::atomic firedAtNs{ 0 }; + + static int64_t NowNs() + { + return std::chrono::duration_cast( + TestClock::now().time_since_epoch()).count(); + } + + static void CALLBACK Invoke(void* ctx, bool canceled) + { + auto* self = static_cast(ctx); + self->canceled.store(canceled, std::memory_order_release); + self->firedAtNs.store(NowNs(), std::memory_order_release); + self->invoked.store(true, std::memory_order_release); + } + }; + + AutoQueueHandle queue; + VERIFY_SUCCEEDED(XTaskQueueCreate( + XTaskQueueDispatchMode::Manual, + XTaskQueueDispatchMode::Manual, + &queue)); + + CallbackState firstState; + CallbackState secondState; + + constexpr uint32_t firstDelayMs = 10; + constexpr uint32_t secondDelayMs = 1000; + constexpr int64_t firstDispatchUpperBoundNs = 500LL * 1000LL * 1000LL; + constexpr int64_t secondDelayNs = 1000LL * 1000LL * 1000LL; + const int64_t submittedAtNs = CallbackState::NowNs(); + + VERIFY_SUCCEEDED(XTaskQueueSubmitDelayedCallback( + queue, + XTaskQueuePort::Work, + firstDelayMs, + &firstState, + &CallbackState::Invoke)); + + const int64_t secondSubmittedAtNs = CallbackState::NowNs(); + VERIFY_SUCCEEDED(XTaskQueueSubmitDelayedCallback( + queue, + XTaskQueuePort::Work, + secondDelayMs, + &secondState, + &CallbackState::Invoke)); + + // Dispatch the first callback. The timer callback that promoted it has + // already re-armed the shared delayed-callback timer for secondState. + VERIFY_IS_TRUE(XTaskQueueDispatch(queue, XTaskQueuePort::Work, 5000)); + + VERIFY_IS_TRUE(firstState.invoked.load(std::memory_order_acquire)); + VERIFY_IS_FALSE(firstState.canceled.load(std::memory_order_acquire)); + VERIFY_IS_FALSE(secondState.invoked.load(std::memory_order_acquire)); + VERIFY_IS_LESS_THAN(CallbackState::NowNs() - submittedAtNs, firstDispatchUpperBoundNs); + + // Simulate a stale delayed-callback notification that was already + // queued before the timer was re-armed for secondState. This must not + // promote the later pending entry before its own deadline. + VERIFY_SUCCEEDED(XTaskQueueSubmitPendingCallbackForTests(queue, XTaskQueuePort::Work)); + + VERIFY_IS_FALSE(XTaskQueueDispatch(queue, XTaskQueuePort::Work, 0)); + VERIFY_IS_FALSE(XTaskQueueDispatch(queue, XTaskQueuePort::Work, 200)); + VERIFY_IS_FALSE(secondState.invoked.load(std::memory_order_acquire)); + + VERIFY_IS_TRUE(XTaskQueueDispatch(queue, XTaskQueuePort::Work, 2000)); + VERIFY_IS_TRUE(secondState.invoked.load(std::memory_order_acquire)); + VERIFY_IS_FALSE(secondState.canceled.load(std::memory_order_acquire)); + VERIFY_IS_GREATER_THAN_OR_EQUAL( + secondState.firedAtNs.load(std::memory_order_acquire) - secondSubmittedAtNs, + secondDelayNs); + } }; From 3030a23c384955306c16b8e3535e1da3a3ca26b7 Mon Sep 17 00:00:00 2001 From: jhugard Date: Mon, 11 May 2026 11:29:24 -0700 Subject: [PATCH 5/6] Fix stale delayed callback, empty-sweep wake, and STL timer teardown races Retarget delayed-callback wake handling so stale timer notifications cannot promote future work too early, keep the same-port empty-sweep rescue path iterative, and harden the STL wait timer so teardown cannot race an in-flight callback after a due heap entry is popped. The regression and hook changes remain grouped here because they cover the same delayed-callback scheduling line and the Linux timer backend that services it. --- Source/Task/TaskQueue.cpp | 249 ++++++++++++----- Source/Task/TaskQueueImpl.h | 5 +- Source/Task/ThreadPool_stl.cpp | 6 +- Source/Task/WaitTimer.h | 14 +- Source/Task/WaitTimer_stl.cpp | 335 ++++++++++++++++++----- Source/Task/WaitTimer_win32.cpp | 74 +++-- Source/Task/XTaskQueuePriv.h | 8 + Source/Task/iOS/ios_WaitTimer.mm | 44 ++- Source/Task/iOS/ios_WaitTimerImpl.h | 2 +- Tests/UnitTests/Tests/TaskQueueTests.cpp | 122 +++++++++ 10 files changed, 687 insertions(+), 172 deletions(-) diff --git a/Source/Task/TaskQueue.cpp b/Source/Task/TaskQueue.cpp index ae59f132a..cc6b1e3e3 100644 --- a/Source/Task/TaskQueue.cpp +++ b/Source/Task/TaskQueue.cpp @@ -362,7 +362,10 @@ HRESULT __stdcall TaskQueuePortImpl::QueueItem( } else { - entry.enqueueTime = m_timer.GetAbsoluteTime(waitMs); + // Delayed callbacks are ordered by a monotonic due time so stale timer + // callbacks and wall-clock adjustments cannot make one pending entry + // masquerade as another. + entry.enqueueTime = m_timer.GetDueTime(waitMs); RETURN_HR_IF(E_OUTOFMEMORY, !m_pendingList->push_back(entry)); // If the entry's enqueue time is < our current time, @@ -959,12 +962,17 @@ void TaskQueuePortImpl::CancelPendingEntries( // share this port's delayed-callback timer state, so leave m_timer and // m_timerDue alone; if we removed the armed earliest entry, the existing // timer simply takes one blank fire and re-arms for the next real item. + LocklessQueue entriesToAppend(*m_queueList.get()); m_pendingList->remove_if([&](auto& entry, auto address) { if (entry.portContext == portContext) { - if (!appendToQueue || !AppendEntry(entry, address)) + if (appendToQueue) + { + entriesToAppend.push_back(std::move(entry), address); + } + else { entry.portContext->Release(); m_pendingList->free_node(address); @@ -976,6 +984,22 @@ void TaskQueuePortImpl::CancelPendingEntries( return false; }); + while (appendToQueue) + { + QueueEntry entry = {}; + uint64_t address = 0; + if (!entriesToAppend.pop_front(entry, address)) + { + break; + } + + if (!AppendEntry(entry, address)) + { + entry.portContext->Release(); + m_queueList->free_node(address); + } + } + #ifdef HC_UNITTEST_API // Test hook: let unit tests enqueue a sibling delayed callback while this // termination path still owns the interleaving window that used to race @@ -1034,83 +1058,130 @@ void TaskQueuePortImpl::EraseQueue( } } -// Examines the pending callback list, optionally popping the entry off the -// list that matches m_timerDue, and schedules the timer for the next entry. -bool TaskQueuePortImpl::ScheduleNextPendingCallback( +// Promotes every delayed entry whose deadline has already arrived and then +// arms the timer for the next future deadline, if one remains. +// +// This replaces the older "pop exactly one entry whose enqueueTime matches the +// currently armed due time" flow. That older model made correctness depend on +// timestamps behaving like unique identities. By sweeping everything with +// enqueueTime <= now, equal-deadline siblings and stale timer callbacks both +// collapse into the same simple rule: if a callback is due, move it now; if it +// is still in the future, leave it pending and re-arm for the earliest future +// item. +void TaskQueuePortImpl::PromoteReadyPendingCallbacks( _In_ uint64_t dueTime, - _Out_ QueueEntry& dueEntry, - _Out_ uint64_t& dueEntryNode) + _In_ uint64_t now) { - QueueEntry nextItem = {}; - bool hasDueEntry = false; - bool hasNextItem = false; + for (;;) + { + // Collect due entries locally first and only touch the active queue + // after remove_if completes. Keeping the sweep phase and the publish + // phase separate preserves the "promote all ready entries" behavior + // without asking remove_if to coexist with queue wakeups and + // cross-queue node reuse at the same time. + LocklessQueue readyEntries(*m_queueList.get()); - dueEntryNode = 0; + QueueEntry nextItem = {}; + bool hasNextItem = false; - m_pendingList->remove_if([&](auto& entry, auto address) - { - if (!hasDueEntry && entry.enqueueTime == dueTime) - { - dueEntry = entry; - dueEntryNode = address; - hasDueEntry = true; - return true; - } - else if (!hasNextItem || nextItem.enqueueTime > entry.enqueueTime) + m_pendingList->remove_if([&](auto& entry, auto address) { - // remove_if works by removing items from the list and - // re-adding them if this callback returns false. If we - // are going to keep an item beyond this callback we need - // to make sure fields we're using stay valid. Only the - // port context is a risk. + // Any entry whose deadline has passed is ready right now, + // regardless of whether its timestamp aliases another entry or + // whether this timer fire is the original notification or a + // stale callback that arrived late. + if (entry.enqueueTime <= now) + { + readyEntries.push_back(std::move(entry), address); + + return true; + } - if (hasNextItem) + if (!hasNextItem || nextItem.enqueueTime > entry.enqueueTime) { - nextItem.portContext->Release(); + // remove_if works by removing items from the list and + // re-adding them if this callback returns false. If we + // are going to keep an item beyond this callback we need + // to make sure fields we're using stay valid. Only the + // port context is a risk. + + if (hasNextItem) + { + nextItem.portContext->Release(); + } + + nextItem = entry; + nextItem.portContext->AddRef(); + hasNextItem = true; } - nextItem = entry; - nextItem.portContext->AddRef(); - hasNextItem = true; - } + return false; + }); - return false; - }); + // Publish the ready entries after the pending-list walk finishes. + QueueEntry readyEntry = {}; + uint64_t readyEntryNode = 0; + while (readyEntries.pop_front(readyEntry, readyEntryNode)) + { + if (!AppendEntry(readyEntry, readyEntryNode)) + { + readyEntry.portContext->Release(); + m_queueList->free_node(readyEntryNode); + } + } - if (hasNextItem) - { - if (nextItem.portContext->GetStatus() == TaskQueuePortStatus::Active) + if (hasNextItem) { - while (true) + if (nextItem.portContext->GetStatus() == TaskQueuePortStatus::Active) { - if (m_timerDue.compare_exchange_weak(dueTime, nextItem.enqueueTime)) + while (true) { - m_timer.Start(nextItem.enqueueTime); - break; - } + // Publish the earliest future deadline that survived the + // ready sweep. If another thread already armed an even + // earlier timer, leave that earlier deadline in place. + if (m_timerDue.compare_exchange_weak(dueTime, nextItem.enqueueTime)) + { + m_timer.Start(nextItem.enqueueTime); + break; + } - dueTime = m_timerDue.load(); + dueTime = m_timerDue.load(); - if (dueTime <= nextItem.enqueueTime) - { - break; + if (dueTime <= nextItem.enqueueTime) + { + break; + } } } - } - else - { - // The port is no longer active. Pending entries are canceled - // when the port is terminated, but if we were iterating above - // it's possible that we removed an item while the termination was - // being processed and it got missed. - CancelPendingEntries(nextItem.portContext, true); + else + { + // The port is no longer active. Pending entries are canceled + // when the port is terminated, but if we were iterating above + // it's possible that we removed an item while the termination + // was being processed and it got missed. + CancelPendingEntries(nextItem.portContext, true); + } + + nextItem.portContext->Release(); + return; } - nextItem.portContext->Release(); - } - else - { + // No future entries remain in the pending list. uint64_t noDueTime = UINT64_MAX; + +#ifdef HC_UNITTEST_API + m_attachedContexts.Visit([&](ITaskQueuePortContext* portContext) + { + auto hooks = portContext->GetQueue()->GetTestHooks(); + if (hooks != nullptr) + { + hooks->NoNextPendingCallbackFound( + portContext->GetType(), + dueTime); + } + }); +#endif + if (m_timerDue.compare_exchange_strong(dueTime, noDueTime)) { // Bug fix: ScheduleNextPendingCallback timer race results @@ -1136,24 +1207,68 @@ bool TaskQueuePortImpl::ScheduleNextPendingCallback( } }); #endif + + // A concurrent QueueItem can append a future entry after our + // sweep has already concluded there is no next item, but before + // we publish UINT64_MAX here. Instead of recursing (which has + // no tail-call guarantee and risks stack growth under sustained + // contention), loop back for a rescue sweep. If nothing landed, + // the second pass is a cheap no-op. + if (dueTime != noDueTime) + { + now = m_timer.GetCurrentTime(); + dueTime = noDueTime; + continue; + } } - } - return hasDueEntry; + return; + } } void TaskQueuePortImpl::SubmitPendingCallback() { - QueueEntry dueEntry; - uint64_t dueEntryNode; - - if (ScheduleNextPendingCallback(m_timerDue.load(), dueEntry, dueEntryNode)) + while (true) { - if (!AppendEntry(dueEntry, dueEntryNode)) + uint64_t dueTime = m_timerDue.load(); + + if (dueTime == UINT64_MAX) + { + return; + } + + // Threadpool timer callbacks that were already queued can still arrive + // after the timer has been retargeted. Treat the callback as advisory and + // only sweep ready entries once the currently armed monotonic deadline has + // actually arrived. + // + // Important: do not just return on an "early" callback. On Win32 the + // threadpool timer's relative wait source is not the same clock object as + // std::chrono::steady_clock, so a legitimate one-shot fire can arrive a + // little before the stored steady-clock deadline. If we drop that callback + // without re-arming the timer, the pending entry can remain stranded until + // some unrelated later timer fire or termination path happens to flush it. + // + // Also do not blindly re-arm the due time we just read. Another thread can + // publish an earlier pending entry between the load above and Start() below. + // If this stale callback then overwrites the timer with the older deadline, + // the newer earlier entry can stay stranded until the older deadline fires. + // Only re-arm when m_timerDue still matches the due time we observed. + const uint64_t now = m_timer.GetCurrentTime(); + if (now < dueTime) { - dueEntry.portContext->Release(); - m_queueList->free_node(dueEntryNode); + uint64_t expectedDueTime = dueTime; + if (m_timerDue.compare_exchange_weak(expectedDueTime, dueTime)) + { + m_timer.Start(dueTime); + return; + } + + continue; } + + PromoteReadyPendingCallbacks(dueTime, now); + return; } } diff --git a/Source/Task/TaskQueueImpl.h b/Source/Task/TaskQueueImpl.h index 98c6755d2..9dd18f0b8 100644 --- a/Source/Task/TaskQueueImpl.h +++ b/Source/Task/TaskQueueImpl.h @@ -311,10 +311,9 @@ class TaskQueuePortImpl: public Api static void EraseQueue( _In_opt_ LocklessQueue* queue); - bool ScheduleNextPendingCallback( + void PromoteReadyPendingCallbacks( _In_ uint64_t dueTime, - _Out_ QueueEntry& dueEntry, - _Out_ uint64_t& dueEntryNode); + _In_ uint64_t now); void SubmitPendingCallback(); diff --git a/Source/Task/ThreadPool_stl.cpp b/Source/Task/ThreadPool_stl.cpp index a429e3e9a..80ce5cf17 100644 --- a/Source/Task/ThreadPool_stl.cpp +++ b/Source/Task/ThreadPool_stl.cpp @@ -182,8 +182,10 @@ namespace OS m_calls++; } - // Release lock before notify_all to optimize immediate awakes - m_wake.notify_all(); + // Release lock before notifying one waiting worker. Additional + // queued calls will issue additional wakes if more parallelism is + // needed. + m_wake.notify_one(); } private: diff --git a/Source/Task/WaitTimer.h b/Source/Task/WaitTimer.h index 33eca4bd8..0a80af5ab 100644 --- a/Source/Task/WaitTimer.h +++ b/Source/Task/WaitTimer.h @@ -6,8 +6,8 @@ namespace OS class WaitTimerImpl; - // A wait timer holds a single timeout in absolute - // time. Calling Start will reset any pending timeout. + // A wait timer holds a single timeout expressed as a monotonic due time. + // Calling Start will reset any pending timeout. class WaitTimer { public: @@ -17,10 +17,16 @@ namespace OS HRESULT Initialize(_In_opt_ void* context, _In_ WaitTimerCallback* callback) noexcept; void Terminate() noexcept; - void Start(_In_ uint64_t absoluteTime) noexcept; + // Arms the one-shot timer for the provided monotonic due time. + void Start(_In_ uint64_t dueTime) noexcept; void Cancel() noexcept; - uint64_t GetAbsoluteTime(_In_ uint32_t msFromNow) noexcept; + // Returns the current monotonic time used for delayed-callback + // ordering and stale-timer validation. + uint64_t GetCurrentTime() noexcept; + + // Returns a monotonic due time msFromNow milliseconds in the future. + uint64_t GetDueTime(_In_ uint32_t msFromNow) noexcept; private: std::atomic m_impl; diff --git a/Source/Task/WaitTimer_stl.cpp b/Source/Task/WaitTimer_stl.cpp index f4e098c7c..74565ffa7 100644 --- a/Source/Task/WaitTimer_stl.cpp +++ b/Source/Task/WaitTimer_stl.cpp @@ -1,33 +1,135 @@ #include "pch.h" #include "WaitTimer.h" -using Deadline = std::chrono::high_resolution_clock::time_point; +using Clock = std::chrono::steady_clock; +using Deadline = Clock::time_point; +using TimerDuration = std::chrono::nanoseconds; + +namespace +{ + // Keep the public WaitTimer surface on a plain integer so TaskQueue can use + // atomics without dragging chrono types through its state. The integer still + // represents steady-clock time, not wall-clock time. + Deadline DeadlineFromDueTime(uint64_t dueTime) noexcept + { + return Deadline(std::chrono::duration_cast(TimerDuration(dueTime))); + } + + uint64_t DueTimeFromDeadline(Deadline deadline) noexcept + { + return static_cast( + std::chrono::duration_cast(deadline.time_since_epoch()).count()); + } +} namespace OS { class TimerQueue; + // Keep callback payload and teardown coordination in shared state because the + // Worker can still hold a due heap entry after the owning WaitTimerImpl has + // been canceled or destroyed. + class WaitTimerState + { + public: + WaitTimerState(_In_opt_ void* context, _In_ WaitTimerCallback* callback) noexcept + : m_context(context), m_callback(callback) + {} + + uint64_t NextGeneration() noexcept + { + return ++m_generation; + } + + uint64_t Generation() const noexcept + { + return m_generation.load(std::memory_order_acquire); + } + + void BeginTerminate() noexcept + { + m_terminating.store(true, std::memory_order_release); + } + + bool TryBeginDispatch() noexcept + { + if (m_terminating.load(std::memory_order_acquire)) + { + return false; + } + + std::lock_guard lock{ m_lock }; + // Re-check under the lock so teardown cannot race with the in-flight + // count increment and then wait forever for a dispatch we started. + if (m_terminating.load(std::memory_order_relaxed)) + { + return false; + } + + ++m_inFlightDispatch; + return true; + } + + void EndDispatch() noexcept + { + std::lock_guard lock{ m_lock }; + ASSERT(m_inFlightDispatch != 0); + + --m_inFlightDispatch; + if (m_inFlightDispatch == 0) + { + m_quiesced.notify_all(); + } + } + + void WaitForQuiesce() noexcept + { + std::unique_lock lock{ m_lock }; + m_quiesced.wait(lock, [this]() noexcept + { + return m_inFlightDispatch == 0; + }); + } + + void InvokeCallback() noexcept + { + m_callback(m_context); + } + + private: + void* m_context; + WaitTimerCallback* m_callback; + std::atomic m_generation{ 0 }; + std::atomic m_terminating{ false }; + DefaultUnnamedMutex m_lock; + std::condition_variable m_quiesced; + uint32_t m_inFlightDispatch = 0; + }; + class WaitTimerImpl { public: ~WaitTimerImpl(); HRESULT Initialize(_In_opt_ void* context, _In_ WaitTimerCallback* callback); - void Start(_In_ uint64_t absoluteTime); + void Start(_In_ uint64_t dueTime); void Cancel(); - void InvokeCallback(); + void Terminate() noexcept; private: - - void* m_context; - WaitTimerCallback* m_callback; + std::shared_ptr m_state; std::shared_ptr m_timerQueue; }; struct TimerEntry { Deadline When; - WaitTimerImpl* Timer; - TimerEntry(Deadline d, WaitTimerImpl* t) : When{ d }, Timer{ t } {} + std::shared_ptr State; + // Each Start() pushes a new heap entry instead of searching/removing the + // old one. Generation lets the Worker discard superseded entries cheaply. + uint64_t Generation; + TimerEntry(Deadline d, std::shared_ptr state, uint64_t g) + : When{ d }, State{ std::move(state) }, Generation{ g } + {} }; struct TimerEntryComparator @@ -38,14 +140,19 @@ namespace OS } }; - class TimerQueue + // The queue is shared across timers, but it should still retire once the + // last timer goes away instead of leaking for process lifetime. + class TimerQueue : public std::enable_shared_from_this { public: bool Init() noexcept; ~TimerQueue(); - void Set(WaitTimerImpl* timer, Deadline deadline) noexcept; - void Remove(WaitTimerImpl const* timer) noexcept; + void AddTimer() noexcept; + void RemoveTimer() noexcept; + void Set(std::shared_ptr const& state, Deadline deadline) noexcept; + void Remove(WaitTimerState const* state) noexcept; + std::thread::id WorkerThreadId() const noexcept; private: void Worker() noexcept; @@ -57,6 +164,7 @@ namespace OS std::condition_variable m_cv; std::vector m_queue; // used as a heap std::thread m_t; + std::atomic m_timerCount{ 0 }; bool m_exitThread = false; bool m_initialized = false; }; @@ -77,7 +185,16 @@ namespace OS m_cv.notify_all(); if (m_t.joinable()) { - m_t.join(); + if (m_t.get_id() == std::this_thread::get_id()) + { + // Immediate-port callbacks can tear down their own timer from the + // Worker thread. Joining from that path would deadlock. + m_t.detach(); + } + else + { + m_t.join(); + } } } @@ -87,9 +204,11 @@ namespace OS try { - m_t = std::thread([this]() + // Capture a self-reference so the queue stays alive until the Worker + // exits even if the last timer concurrently clears the global slot. + m_t = std::thread([keepAlive = shared_from_this()]() { - Worker(); + keepAlive->Worker(); }); m_initialized = true; } @@ -101,41 +220,81 @@ namespace OS return m_initialized; } - void TimerQueue::Set(WaitTimerImpl* timer, Deadline deadline) noexcept + void TimerQueue::AddTimer() noexcept { + m_timerCount.fetch_add(1, std::memory_order_relaxed); + } + + void TimerQueue::RemoveTimer() noexcept + { + if (m_timerCount.fetch_sub(1, std::memory_order_acq_rel) != 1) { - std::lock_guard lock{ m_mutex }; + return; + } - for (auto& entry : m_queue) + // Last timer out clears the global queue and wakes the Worker so Linux + // teardown actually quiesces instead of depending on process exit. + { + std::lock_guard globalLock{ g_timerQueueMutex }; + if (g_timerQueue.get() == this) { - if (entry.Timer == timer) - { - entry.Timer = nullptr; - } + g_timerQueue.reset(); } + } + + { + std::lock_guard lock{ m_mutex }; + m_exitThread = true; + } + + m_cv.notify_one(); + } + + void TimerQueue::Set(std::shared_ptr const& state, Deadline deadline) noexcept + { + bool shouldNotify; + { + std::lock_guard lock{ m_mutex }; - m_queue.emplace_back(deadline, timer); + // Bump the generation so stale heap entries for this timer are + // skipped by the Worker on pop. This replaces the old O(N) + // nullification scan with an O(log N) push. + uint64_t gen = state->NextGeneration(); + + // Only wake the Worker when the new deadline might be earlier + // than the current heap top; otherwise the Worker is already + // sleeping for a deadline that is at least as early. + shouldNotify = m_queue.empty() || deadline < m_queue.front().When; + + m_queue.emplace_back(deadline, state, gen); std::push_heap(m_queue.begin(), m_queue.end(), TimerEntryComparator{}); } - m_cv.notify_all(); + if (shouldNotify) + { + m_cv.notify_one(); + } } - void TimerQueue::Remove(WaitTimerImpl const* timer) noexcept + void TimerQueue::Remove(WaitTimerState const* state) noexcept { std::lock_guard lock{ m_mutex }; - // since m_queue is a heap, removing elements is non trivial, instead we - // just clean the timer pointer and the entry will be popped eventually - + // Remove is only called during cancellation/teardown. + // Reset shared state entries so the Worker never dispatches them. for (auto& entry : m_queue) { - if (entry.Timer == timer) + if (entry.State.get() == state) { - entry.Timer = nullptr; + entry.State.reset(); } } } + std::thread::id TimerQueue::WorkerThreadId() const noexcept + { + return m_t.get_id(); + } + void TimerQueue::Worker() noexcept { std::unique_lock lock{ m_mutex }; @@ -143,28 +302,52 @@ namespace OS { while (!m_queue.empty()) { - Deadline next = Peek().When; - if (std::chrono::high_resolution_clock::now() < next) + auto& top = Peek(); + + // Discard stale/nullified entries without releasing the lock. + if (!top.State || + top.Generation != top.State->Generation()) + { + Pop(); + continue; + } + + if (Clock::now() < top.When) { break; } TimerEntry entry = Pop(); + if (!entry.State->TryBeginDispatch()) + { + continue; + } - // release the lock while invoking the callback, just in case timer - // gets destroyed on this thread or re-adds itself in the callback + // Release the lock while invoking the callback, just in case + // the timer gets destroyed on this thread or re-adds itself + // in the callback. lock.unlock(); - if (entry.Timer) // Timer is set to nullptr if the entry is removed + entry.State->InvokeCallback(); + entry.State->EndDispatch(); + lock.lock(); + } + + // Drain dead entries at the heap top so wait_until targets a + // live deadline rather than sleeping until a stale entry's time. + while (!m_queue.empty()) + { + auto& top = Peek(); + if (top.State && + top.Generation == top.State->Generation()) { - entry.Timer->InvokeCallback(); + break; } - lock.lock(); + Pop(); } if (!m_queue.empty()) { - Deadline next = Peek().When; - m_cv.wait_until(lock, next); + m_cv.wait_until(lock, Peek().When); } else { @@ -190,25 +373,19 @@ namespace OS WaitTimerImpl::~WaitTimerImpl() { - std::lock_guard lock{ g_timerQueueMutex }; - - // If we are the last one referencing the global timer the - // shared use count will be two (us + the global). If it is, - // clear out the global. We let our own reference reset - // as the class destructs. This puts it outside the mutex - // lock, which we want since there is some shutdown cost - // associated with shutting the timer down. - - if (g_timerQueue.use_count() == 2) - { - g_timerQueue.reset(); - } + Terminate(); } HRESULT WaitTimerImpl::Initialize(_In_opt_ void* context, _In_ WaitTimerCallback* callback) { - m_context = context; - m_callback = callback; + try + { + m_state = http_allocate_shared(context, callback); + } + catch (const std::bad_alloc&) + { + return E_OUTOFMEMORY; + } std::lock_guard lock{ g_timerQueueMutex }; @@ -231,23 +408,46 @@ namespace OS } m_timerQueue = g_timerQueue; + m_timerQueue->AddTimer(); return S_OK; } - void WaitTimerImpl::Start(_In_ uint64_t absoluteTime) + void WaitTimerImpl::Start(_In_ uint64_t dueTime) { - m_timerQueue->Set(this, Deadline(Deadline::duration(absoluteTime))); + m_timerQueue->Set(m_state, DeadlineFromDueTime(dueTime)); } void WaitTimerImpl::Cancel() { - m_timerQueue->Remove(this); + if (m_state != nullptr && m_timerQueue != nullptr) + { + m_timerQueue->Remove(m_state.get()); + } } - void WaitTimerImpl::InvokeCallback() + void WaitTimerImpl::Terminate() noexcept { - m_callback(m_context); + std::shared_ptr state = std::move(m_state); + std::shared_ptr timerQueue = std::move(m_timerQueue); + if (state == nullptr || timerQueue == nullptr) + { + return; + } + + // Block any new dispatch before removing queued entries so teardown has + // a single publish point that both the Worker and waiter observe. + state->BeginTerminate(); + timerQueue->Remove(state.get()); + + if (std::this_thread::get_id() != timerQueue->WorkerThreadId()) + { + // Delayed callbacks can run on Immediate queues and self-terminate on + // the Worker thread. Waiting there would deadlock on our own dispatch. + state->WaitForQuiesce(); + } + + timerQueue->RemoveTimer(); } WaitTimer::WaitTimer() noexcept @@ -281,13 +481,13 @@ namespace OS std::unique_ptr timer(m_impl.exchange(nullptr)); if (timer != nullptr) { - timer->Cancel(); + timer->Terminate(); } } - void WaitTimer::Start(_In_ uint64_t absoluteTime) noexcept + void WaitTimer::Start(_In_ uint64_t dueTime) noexcept { - m_impl.load()->Start(absoluteTime); + m_impl.load()->Start(dueTime); } void WaitTimer::Cancel() noexcept @@ -295,9 +495,14 @@ namespace OS m_impl.load()->Cancel(); } - uint64_t WaitTimer::GetAbsoluteTime(_In_ uint32_t msFromNow) noexcept + uint64_t WaitTimer::GetCurrentTime() noexcept + { + return DueTimeFromDeadline(Clock::now()); + } + + uint64_t WaitTimer::GetDueTime(_In_ uint32_t msFromNow) noexcept { - Deadline d = std::chrono::high_resolution_clock::now() + std::chrono::milliseconds(msFromNow); - return d.time_since_epoch().count(); + Deadline deadline = Clock::now() + std::chrono::milliseconds(msFromNow); + return DueTimeFromDeadline(deadline); } -} // Namespace \ No newline at end of file +} // Namespace diff --git a/Source/Task/WaitTimer_win32.cpp b/Source/Task/WaitTimer_win32.cpp index 81a1c544c..716dcf16c 100644 --- a/Source/Task/WaitTimer_win32.cpp +++ b/Source/Task/WaitTimer_win32.cpp @@ -1,6 +1,24 @@ #include "pch.h" #include "WaitTimer.h" +using Clock = std::chrono::steady_clock; +using Deadline = Clock::time_point; +using TimerDuration = std::chrono::nanoseconds; + +namespace +{ + Deadline DeadlineFromDueTime(uint64_t dueTime) noexcept + { + return Deadline(std::chrono::duration_cast(TimerDuration(dueTime))); + } + + uint64_t DueTimeFromDeadline(Deadline deadline) noexcept + { + return static_cast( + std::chrono::duration_cast(deadline.time_since_epoch()).count()); + } +} + namespace OS { class WaitTimerImpl @@ -37,12 +55,36 @@ namespace OS } } - void Start(_In_ uint64_t absoluteTime) + void Start(_In_ uint64_t dueTime) { LARGE_INTEGER li; FILETIME ft; - ASSERT((absoluteTime & 0x8000000000000000) == 0); - li.QuadPart = static_cast(absoluteTime); + + // The threadpool timer can run on its existing one-shot mechanism, + // but the due time we store in TaskQueue is now a steady-clock + // deadline. Convert that deadline back into a relative wait here so + // queue correctness no longer depends on wall-clock precision. + Deadline now = Clock::now(); + Deadline dueDeadline = DeadlineFromDueTime(dueTime); + TimerDuration remaining = dueDeadline > now + ? std::chrono::duration_cast(dueDeadline - now) + : TimerDuration::zero(); + + constexpr int64_t nanosPerTick = 100; + int64_t relativeTicks = + (static_cast(remaining.count()) + nanosPerTick - 1) / + nanosPerTick; + + // SetThreadpoolTimer expects negative values for relative waits. + // Clamp zero-or-past deadlines to the smallest relative delay so an + // already-due timer is queued immediately without switching back to + // absolute wall-clock FILETIME semantics. + if (relativeTicks <= 0) + { + relativeTicks = 1; + } + + li.QuadPart = -relativeTicks; ft.dwHighDateTime = li.HighPart; ft.dwLowDateTime = li.LowPart; @@ -108,9 +150,9 @@ namespace OS } } - void WaitTimer::Start(_In_ uint64_t absoluteTime) noexcept + void WaitTimer::Start(_In_ uint64_t dueTime) noexcept { - m_impl.load()->Start(absoluteTime); + m_impl.load()->Start(dueTime); } void WaitTimer::Cancel() noexcept @@ -118,20 +160,14 @@ namespace OS m_impl.load()->Cancel(); } - uint64_t WaitTimer::GetAbsoluteTime(_In_ uint32_t msFromNow) noexcept + uint64_t WaitTimer::GetCurrentTime() noexcept { - FILETIME ft; - ULARGE_INTEGER li; - GetSystemTimeAsFileTime(&ft); - ASSERT((ft.dwHighDateTime & 0x80000000) == 0); - - uint64_t hundredNanosFromNow = msFromNow; - hundredNanosFromNow *= 10000ULL; - - li.HighPart = ft.dwHighDateTime; - li.LowPart = ft.dwLowDateTime; - li.QuadPart += hundredNanosFromNow; + return DueTimeFromDeadline(Clock::now()); + } - return li.QuadPart; + uint64_t WaitTimer::GetDueTime(_In_ uint32_t msFromNow) noexcept + { + Deadline deadline = Clock::now() + std::chrono::milliseconds(msFromNow); + return DueTimeFromDeadline(deadline); } -} // Namespace \ No newline at end of file +} // Namespace diff --git a/Source/Task/XTaskQueuePriv.h b/Source/Task/XTaskQueuePriv.h index 3c1392e66..ae9833f90 100644 --- a/Source/Task/XTaskQueuePriv.h +++ b/Source/Task/XTaskQueuePriv.h @@ -56,6 +56,14 @@ struct XTaskQueueTestHooks UNREFERENCED_PARAMETER(port); } + virtual void NoNextPendingCallbackFound( + XTaskQueuePort port, + uint64_t dueTime) + { + UNREFERENCED_PARAMETER(port); + UNREFERENCED_PARAMETER(dueTime); + } + virtual void NextPendingCallbackScheduled( XTaskQueuePort port, uint64_t lastDueTime, diff --git a/Source/Task/iOS/ios_WaitTimer.mm b/Source/Task/iOS/ios_WaitTimer.mm index 470cc736c..8e75b6d3f 100644 --- a/Source/Task/iOS/ios_WaitTimer.mm +++ b/Source/Task/iOS/ios_WaitTimer.mm @@ -9,7 +9,23 @@ #include "ios_WaitTimerImpl.h" -using Deadline = std::chrono::high_resolution_clock::time_point; +using Clock = std::chrono::steady_clock; +using Deadline = Clock::time_point; +using TimerDuration = std::chrono::nanoseconds; + +namespace +{ + Deadline DeadlineFromDueTime(uint64_t dueTime) noexcept + { + return Deadline(std::chrono::duration_cast(TimerDuration(dueTime))); + } + + uint64_t DueTimeFromDeadline(Deadline deadline) noexcept + { + return static_cast( + std::chrono::duration_cast(deadline.time_since_epoch()).count()); + } +} WaitTimerImpl::WaitTimerImpl() : m_context(nullptr), @@ -34,12 +50,13 @@ return S_OK; } -void WaitTimerImpl::Start(_In_ uint64_t absoluteTime) +void WaitTimerImpl::Start(_In_ uint64_t dueTime) { Cancel(); - - auto duration = Deadline::duration(absoluteTime); - auto timePoint = Deadline(duration) - std::chrono::high_resolution_clock::now(); + + // NSTimer consumes a relative interval, so convert the stored steady-clock + // deadline back into a relative delay right before arming it. + auto timePoint = DeadlineFromDueTime(dueTime) - Clock::now(); auto ms = std::chrono::duration_cast(timePoint); m_timer = [NSTimer scheduledTimerWithTimeInterval:ms.count() / 1000.0 @@ -77,7 +94,7 @@ } } -HRESULT WaitTimer::Initialize(void *context, WaitTimerCallback *callback) noexcept +HRESULT WaitTimer::Initialize(_In_opt_ void* context, _In_ WaitTimerCallback* callback) noexcept { if (m_impl != nullptr || callback == nullptr) { @@ -94,9 +111,9 @@ return S_OK; } -void WaitTimer::Start(uint64_t absoluteTime) noexcept +void WaitTimer::Start(_In_ uint64_t dueTime) noexcept { - m_impl->Start(absoluteTime); + m_impl->Start(dueTime); } void WaitTimer::Cancel() noexcept @@ -104,8 +121,13 @@ m_impl->Cancel(); } -uint64_t WaitTimer::GetAbsoluteTime(uint32_t msFromNow) noexcept +uint64_t WaitTimer::GetCurrentTime() noexcept +{ + return DueTimeFromDeadline(Clock::now()); +} + +uint64_t WaitTimer::GetDueTime(_In_ uint32_t msFromNow) noexcept { - auto deadline = std::chrono::high_resolution_clock::now() + std::chrono::milliseconds(msFromNow); - return deadline.time_since_epoch().count(); + auto deadline = Clock::now() + std::chrono::milliseconds(msFromNow); + return DueTimeFromDeadline(deadline); } diff --git a/Source/Task/iOS/ios_WaitTimerImpl.h b/Source/Task/iOS/ios_WaitTimerImpl.h index 162f80700..17c255ec8 100644 --- a/Source/Task/iOS/ios_WaitTimerImpl.h +++ b/Source/Task/iOS/ios_WaitTimerImpl.h @@ -17,7 +17,7 @@ class WaitTimerImpl WaitTimerImpl(); ~WaitTimerImpl(); HRESULT Initialize(_In_opt_ void* context, _In_ WaitTimerCallback* callback); - void Start(_In_ uint64_t absoluteTime); + void Start(_In_ uint64_t dueTime); void Cancel(); void TimerFired(); diff --git a/Tests/UnitTests/Tests/TaskQueueTests.cpp b/Tests/UnitTests/Tests/TaskQueueTests.cpp index 44480d4b5..970fec363 100644 --- a/Tests/UnitTests/Tests/TaskQueueTests.cpp +++ b/Tests/UnitTests/Tests/TaskQueueTests.cpp @@ -2056,6 +2056,128 @@ DEFINE_TEST_CLASS(TaskQueueTests) VERIFY_IS_TRUE(canaryFired.load()); } + DEFINE_TEST_CASE(VerifyFutureDelayedCallbackQueuedDuringEmptySweepDoesNotStall) + { + // Regression: a future delayed callback can be queued on the same port + // after the ready sweep has concluded there is no next item, but before + // m_timerDue is reset to UINT64_MAX. In that interleaving the publisher + // still sees the stale prior due time, does not retarget the timer, and + // the empty-sweep path can strand the new callback with no armed wake. + + struct TestBarrier + { + std::mutex mtx; + std::condition_variable cv; + bool phase1_ready = false; // timer callback -> test thread + bool phase2_ready = false; // test thread -> timer callback + }; + + struct TestHooks : public XTaskQueueTestHooks + { + explicit TestHooks(_In_ TestBarrier* barrier) : m_testBarrier(barrier) {} + + void NoNextPendingCallbackFound(XTaskQueuePort port, uint64_t dueTime) override + { + UNREFERENCED_PARAMETER(port); + UNREFERENCED_PARAMETER(dueTime); + + std::unique_lock lk(m_testBarrier->mtx); + if (!m_hookArmed) + { + return; + } + + m_hookArmed = false; + m_testBarrier->phase1_ready = true; + lk.unlock(); + m_testBarrier->cv.notify_all(); + + lk.lock(); + m_testBarrier->cv.wait_for( + lk, + std::chrono::seconds(5), + [&] { return m_testBarrier->phase2_ready; }); + } + + private: + TestBarrier* m_testBarrier = nullptr; + bool m_hookArmed = true; + }; + + TestBarrier barrier; + TestHooks hooks(&barrier); + + AutoQueueHandle queue; + VERIFY_SUCCEEDED(XTaskQueueCreate( + XTaskQueueDispatchMode::Manual, + XTaskQueueDispatchMode::Immediate, + &queue)); + + VERIFY_SUCCEEDED(XTaskQueueSetTestHooks(queue, &hooks)); + + std::atomic firstFired{ false }; + std::atomic secondFired{ false }; + + auto markFired = [](void* ctx, bool cancel) + { + if (!cancel) + { + static_cast*>(ctx)->store(true); + } + }; + + VERIFY_SUCCEEDED(XTaskQueueSubmitDelayedCallback( + queue, + XTaskQueuePort::Work, + 1, + &firstFired, + markFired)); + + { + std::unique_lock lk(barrier.mtx); + bool ok = barrier.cv.wait_for( + lk, + std::chrono::seconds(5), + [&] { return barrier.phase1_ready; }); + VERIFY_IS_TRUE(ok); + } + + VERIFY_SUCCEEDED(XTaskQueueSubmitDelayedCallback( + queue, + XTaskQueuePort::Work, + 10, + &secondFired, + markFired)); + + { + std::lock_guard lk(barrier.mtx); + barrier.phase2_ready = true; + } + barrier.cv.notify_all(); + + VERIFY_SUCCEEDED(XTaskQueueSetTestHooks(queue, nullptr)); + + const uint64_t start = GetTickCount64(); + while ((!firstFired.load() || !secondFired.load()) && + GetTickCount64() - start < 2000) + { + XTaskQueueDispatch(queue, XTaskQueuePort::Work, 100); + } + + LOG_COMMENT(L"First callback fired: %s", + firstFired.load() ? L"yes" : L"no"); + LOG_COMMENT(L"Second callback fired: %s", + secondFired.load() ? L"yes" : L"NO -- callback stranded during empty sweep"); + + XTaskQueueTerminate(queue, false, nullptr, nullptr); + while (XTaskQueueDispatch(queue, XTaskQueuePort::Work, 0)) + { + } + + VERIFY_IS_TRUE(firstFired.load()); + VERIFY_IS_TRUE(secondFired.load()); + } + DEFINE_TEST_CASE(VerifyTerminationDoesNotEarlyPromoteSiblingDelayedCallback) { using TestClock = std::chrono::steady_clock; From 66e7fa453cc7b50dc73e5de29ac9c1f9acefa4ab Mon Sep 17 00:00:00 2001 From: jhugard Date: Mon, 11 May 2026 11:29:24 -0700 Subject: [PATCH 6/6] Fix iOS WaitTimer ceiling rounding and teardown Keep the iOS backend on monotonic due times without truncating sub-millisecond delays, and implement the missing WaitTimer termination path so the Apple backend matches the shared TaskQueue lifecycle contract. --- Source/Task/iOS/ios_WaitTimer.mm | 25 ++++++++++++++++++------- 1 file changed, 18 insertions(+), 7 deletions(-) diff --git a/Source/Task/iOS/ios_WaitTimer.mm b/Source/Task/iOS/ios_WaitTimer.mm index 8e75b6d3f..8ddc5d197 100644 --- a/Source/Task/iOS/ios_WaitTimer.mm +++ b/Source/Task/iOS/ios_WaitTimer.mm @@ -55,9 +55,13 @@ uint64_t DueTimeFromDeadline(Deadline deadline) noexcept Cancel(); // NSTimer consumes a relative interval, so convert the stored steady-clock - // deadline back into a relative delay right before arming it. - auto timePoint = DeadlineFromDueTime(dueTime) - Clock::now(); - auto ms = std::chrono::duration_cast(timePoint); + // deadline back into a relative delay right before arming it. Use ceiling + // rounding so that a sub-millisecond remainder is rounded up rather than + // truncated to zero, and clamp to zero so a past deadline never produces a + // negative interval that would cause a tight re-arm loop. + auto remaining = DeadlineFromDueTime(dueTime) - Clock::now(); + auto ms = std::max(std::chrono::milliseconds(0), + std::chrono::ceil(remaining)); m_timer = [NSTimer scheduledTimerWithTimeInterval:ms.count() / 1000.0 target:m_target @@ -88,10 +92,7 @@ uint64_t DueTimeFromDeadline(Deadline deadline) noexcept WaitTimer::~WaitTimer() noexcept { - if (m_impl != nullptr) - { - delete m_impl; - } + Terminate(); } HRESULT WaitTimer::Initialize(_In_opt_ void* context, _In_ WaitTimerCallback* callback) noexcept @@ -111,6 +112,16 @@ uint64_t DueTimeFromDeadline(Deadline deadline) noexcept return S_OK; } +void WaitTimer::Terminate() noexcept +{ + WaitTimerImpl* timer = m_impl.exchange(nullptr); + if (timer != nullptr) + { + timer->Cancel(); + delete timer; + } +} + void WaitTimer::Start(_In_ uint64_t dueTime) noexcept { m_impl->Start(dueTime);