MaybeWakeWaiters

Author: VSadov

src/libraries/System.Private.CoreLib/src/System/Threading/LowLevelLifoSemaphore.cs

@@ -17,7 +17,7 @@ internal sealed partial class LowLevelLifoSemaphore
     {
         // Spinning in the threadpool semaphore is not always useful and benefits vary greatly by scenario.
         //
-        // Example1: An app periodically with rough time span T runs a task and waits for task`s completion.
+        // Example1: An app periodically with rough time span T runs a task and waits for task's completion.
         //           The app would benefit if a threadpool worker spins for longer than T as worker would not need to be woken up.
         //
         // Example2: The new workitems may be produced by non-pool threads and could only arrive if pool threads start blocking.
@@ -146,11 +146,6 @@ private bool WaitSlow(int timeoutMs, short tpThreadCount)
                 }
             }
 
-            return WaitNoSpin(timeoutMs, allowFastWake: true);
-        }
-
-        public bool WaitNoSpin(int timeoutMs, bool allowFastWake = false)
-        {
             // Now we will try registering as a waiter and wait.
             // If signaled before that, we have to acquire as this can be the last thread that could take that signal.
             // The difference with spinning above is that we are not waiting for a signal. We should typically
@@ -172,24 +167,67 @@ public bool WaitNoSpin(int timeoutMs, bool allowFastWake = false)
                 Counts countsBeforeUpdate = _separated._counts.InterlockedCompareExchange(newCounts, counts);
                 if (countsBeforeUpdate == counts)
                 {
-                    return counts.SignalCount != 0 || WaitAsWaiter(timeoutMs, allowFastWake);
+                    return counts.SignalCount != 0 || WaitAsWaiter(timeoutMs, allowFastWake: true);
                 }
 
                 Backoff.Exponential(collisionCount++);
             }
         }
 
+        public bool WaitNoSpin(int timeoutMs)
+        {
+            Counts counts = _separated._counts.InterlockedIncrementWaiterCount();
+
+            // If there were pending signals, we may end in a condition that requires
+            // waking a waiter.
+            // Perhaps the current thread will be such waiter, but we should still
+            // go through wait/wake routine (vs. just claiming the signal) as the caller
+            // wants to park the thread.
+            MaybeWakeWaiters(counts);
+
+            return WaitAsWaiter(timeoutMs, allowFastWake: false);
+        }
+
+        private void MaybeWakeWaiters(Counts counts)
+        {
+            // Check if waiters need to be woken
+            uint collisionCount = 0;
+            while (true)
+            {
+                // Determine how many waiters to wake.
+                // The number of wakes should not be more than the signal count, not more than waiter count and discount any pending wakes.
+                int countOfWaitersToWake = (int)Math.Min(counts.SignalCount, counts.WaiterCount) - counts.CountOfWaitersSignaledToWake;
+                if (countOfWaitersToWake <= 0)
+                {
+                    // No waiters to wake. This is the most common case.
+                    break;
+                }
+
+                Counts newCounts = counts;
+                newCounts.AddCountOfWaitersSignaledToWake((uint)countOfWaitersToWake);
+                Counts countsBeforeUpdate = _separated._counts.InterlockedCompareExchange(newCounts, counts);
+                if (countsBeforeUpdate == counts)
+                {
+                    Debug.Assert(_maximumSignalCount - counts.SignalCount >= 1);
+                    if (countOfWaitersToWake > 0)
+                    {
+                        ReleaseCore(countOfWaitersToWake);
+                    }
+
+                    break;
+                }
+
+                // collision, try again.
+                Backoff.Exponential(collisionCount++);
+
+                counts = _separated._counts;
+            }
+        }
+
         private bool WaitAsWaiter(int timeoutMs, bool allowFastWake)
         {
             Debug.Assert(timeoutMs >= -1);
 
-            // If the caller wants that the threads spend time waiting require a 4 usec cooldown
-            // before reintroducing the thread. The sleep/wake transition typically takes care of that,
-            // but the blocker has fast wake paths and the underlying OS API may have those as well,
-            // thus fast wake ups are hard to avoid completely.
-            // If fast wake happened when parking was desired, we hold up the thread a bit
-            // before reintroducing.
-            long cooldown = Stopwatch.Frequency * 4 / 1000000;
             while (true)
             {
                 long blockingStart = allowFastWake ? 0 : Stopwatch.GetTimestamp();
@@ -260,36 +298,7 @@ public void Signal()
         {
             // Increment signal count. This enables one-shot acquire.
             Counts counts = _separated._counts.InterlockedIncrementSignalCount();
-
-            // Now check if waiters need to be woken
-            uint collisionCount = 0;
-            while (true)
-            {
-                // Determine how many waiters to wake.
-                // The number of wakes should not be more than the signal count, not more than waiter count and discount any pending wakes.
-                int countOfWaitersToWake = (int)Math.Min(counts.SignalCount, counts.WaiterCount) - counts.CountOfWaitersSignaledToWake;
-                if (countOfWaitersToWake <= 0)
-                {
-                    // No waiters to wake. This is the most common case.
-                    return;
-                }
-
-                Counts newCounts = counts;
-                newCounts.AddCountOfWaitersSignaledToWake((uint)countOfWaitersToWake);
-                Counts countsBeforeUpdate = _separated._counts.InterlockedCompareExchange(newCounts, counts);
-                if (countsBeforeUpdate == counts)
-                {
-                    Debug.Assert(_maximumSignalCount - counts.SignalCount >= 1);
-                    if (countOfWaitersToWake > 0)
-                        ReleaseCore(countOfWaitersToWake);
-                    return;
-                }
-
-                // collision, try again.
-                Backoff.Exponential(collisionCount++);
-
-                counts = _separated._counts;
-            }
+            MaybeWakeWaiters(counts);
         }
 
         private bool Block(int timeoutMs)
@@ -458,6 +467,13 @@ public void InterlockedDecrementWaiterCount()
                 Debug.Assert(countsAfterUpdate.WaiterCount != ushort.MaxValue); // underflow check
             }
 
+            public Counts InterlockedIncrementWaiterCount()
+            {
+                var countsAfterUpdate = new Counts(Interlocked.Add(ref _data, unchecked((ulong)1) << WaiterCountShift));
+                Debug.Assert(countsAfterUpdate.WaiterCount != ushort.MaxValue); // overflow check
+                return countsAfterUpdate;
+            }
+
             public ushort CountOfWaitersSignaledToWake
             {
                 get => GetUInt16Value(CountOfWaitersSignaledToWakeShift);

src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.cs

@@ -108,7 +108,6 @@ private static void WorkerThreadStart()
                     bool isIdle = false;
                     while (isIdle ?
                         semaphore.WaitNoSpin(timeoutMs) :
-                        //semaphore.Wait(timeoutMs, threadPoolInstance._separated.counts.NumExistingThreads) :
                         semaphore.Wait(timeoutMs, threadPoolInstance._separated.counts.NumExistingThreads))
                     {
                         WorkerDoWork(threadPoolInstance, out isIdle);

src/libraries/System.Private.CoreLib/src/System/Threading/ThreadPoolWorkQueue.cs

@@ -366,7 +366,10 @@ public bool LocalFindAndPop(object obj)
                                 else
                                 {
                                     // Failed, restore head.
-                                    m_headIndex = head;
+                                    // This write must complete before we return with a missed steal and check if
+                                    // there is a pending thread request because the thread that responds
+                                    // to the request must see the write to not conclude that the queue is empty.
+                                    Interlocked.Exchange(ref m_headIndex, head);
                                 }
                             }
                         }