Guard delayed timer re-arm against stale callbacks

Author: jhugard

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<QueueEntry> 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,29 +1058,46 @@ 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)
+{
+    // Collect due entries locally first and only touch the active queue after
+    // remove_if completes. The callback passed to LocklessQueue::remove_if owns
+    // any removed node addresses, but mutating the ready queue and signaling
+    // dispatchers from inside that walk makes the pending-list sweep interleave
+    // with new work publication. 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<QueueEntry> readyEntries(*m_queueList.get());
+
     QueueEntry nextItem = {};
-    bool hasDueEntry = false;
     bool hasNextItem = false;
 
-    dueEntryNode = 0;
-
     m_pendingList->remove_if([&](auto& entry, auto address)
     {
-        if (!hasDueEntry && entry.enqueueTime == dueTime)
+        // 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)
         {
-            dueEntry = entry;
-            dueEntryNode = address;
-            hasDueEntry = true;
+            readyEntries.push_back(std::move(entry), address);
+
             return true;
         }
-        else if (!hasNextItem || nextItem.enqueueTime > entry.enqueueTime)
+
+        if (!hasNextItem || nextItem.enqueueTime > entry.enqueueTime)
         {
             // remove_if works by removing items from the list and
             // re-adding them if this callback returns false. If we
@@ -1077,12 +1118,30 @@ bool TaskQueuePortImpl::ScheduleNextPendingCallback(
         return false;
     });
 
+    // Publish the ready entries after the pending-list walk finishes. This
+    // keeps queue wakeups and threadpool submissions out of remove_if's
+    // critical section while still publishing each promoted ready entry
+    // individually once the sweep is complete.
+    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)
         {
             while (true)
             {
+                // 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 and do not overwrite it.
                 if (m_timerDue.compare_exchange_weak(dueTime, nextItem.enqueueTime))
                 {
                     m_timer.Start(nextItem.enqueueTime);
@@ -1139,21 +1198,51 @@ bool TaskQueuePortImpl::ScheduleNextPendingCallback(
         }
     }
 
-    return hasDueEntry;
 }
 
 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)
         {
-            dueEntry.portContext->Release();
-            m_queueList->free_node(dueEntryNode);
+            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)
+        {
+            uint64_t expectedDueTime = dueTime;
+            if (m_timerDue.compare_exchange_weak(expectedDueTime, dueTime))
+            {
+                m_timer.Start(dueTime);
+                return;
+            }
+
+            continue;
+        }
+
+        PromoteReadyPendingCallbacks(dueTime, now);
+        return;
     }
 }
 

Source/Task/TaskQueueImpl.h

@@ -311,10 +311,9 @@ class TaskQueuePortImpl: public Api<ApiId::TaskQueuePort, ITaskQueuePort>
     static void EraseQueue(
         _In_opt_ LocklessQueue<QueueEntry>* queue);
 
-    bool ScheduleNextPendingCallback(
+    void PromoteReadyPendingCallbacks(
         _In_ uint64_t dueTime,
-        _Out_ QueueEntry& dueEntry,
-        _Out_ uint64_t& dueEntryNode);
+        _In_ uint64_t now);
 
     void SubmitPendingCallback();
 

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<WaitTimerImpl*> m_impl;

Source/Task/WaitTimer_stl.cpp

@@ -1,7 +1,26 @@
 #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<Clock::duration>(TimerDuration(dueTime)));
+    }
+
+    uint64_t DueTimeFromDeadline(Deadline deadline) noexcept
+    {
+        return static_cast<uint64_t>(
+            std::chrono::duration_cast<TimerDuration>(deadline.time_since_epoch()).count());
+    }
+}
 
 namespace OS
 {
@@ -12,7 +31,7 @@ namespace OS
     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();
 
@@ -144,7 +163,7 @@ namespace OS
             while (!m_queue.empty())
             {
                 Deadline next = Peek().When;
-                if (std::chrono::high_resolution_clock::now() < next)
+                if (Clock::now() < next)
                 {
                     break;
                 }
@@ -235,9 +254,9 @@ namespace OS
         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(this, DeadlineFromDueTime(dueTime));
     }
 
     void WaitTimerImpl::Cancel()
@@ -285,19 +304,24 @@ 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
     {
         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