diff --git a/src/libraries/System.Private.CoreLib/src/System/Threading/ThreadInt64PersistentCounter.cs b/src/libraries/System.Private.CoreLib/src/System/Threading/ThreadInt64PersistentCounter.cs index 0f7fbc06a9a8eb..29cf2dce305657 100644 --- a/src/libraries/System.Private.CoreLib/src/System/Threading/ThreadInt64PersistentCounter.cs +++ b/src/libraries/System.Private.CoreLib/src/System/Threading/ThreadInt64PersistentCounter.cs @@ -15,6 +15,7 @@ internal sealed class ThreadInt64PersistentCounter private static List? t_nodeFinalizationHelpers; private long _overflowCount; + private long _lastReturnedCount; // dummy node serving as a start and end of the ring list private readonly ThreadLocalNode _nodes; @@ -31,6 +32,13 @@ public static void Increment(object threadLocalCountObject) Unsafe.As(threadLocalCountObject).Increment(); } + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public static void Decrement(object threadLocalCountObject) + { + Debug.Assert(threadLocalCountObject is ThreadLocalNode); + Unsafe.As(threadLocalCountObject).Decrement(); + } + [MethodImpl(MethodImplOptions.AggressiveInlining)] public static void Add(object threadLocalCountObject, uint count) { @@ -76,6 +84,17 @@ public long Count count += node.Count; node = node._next; } + + // Ensure that the returned value is monotonically increasing + long lastReturnedCount = _lastReturnedCount; + if (count > lastReturnedCount) + { + _lastReturnedCount = count; + } + else + { + count = lastReturnedCount; + } } finally { @@ -134,6 +153,18 @@ public void Increment() OnAddOverflow(1); } + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public void Decrement() + { + if (_count != 0) + { + _count--; + return; + } + + OnAddOverflow(-1); + } + public void Add(uint count) { Debug.Assert(count != 0); @@ -149,7 +180,7 @@ public void Add(uint count) } [MethodImpl(MethodImplOptions.NoInlining)] - private void OnAddOverflow(uint count) + private void OnAddOverflow(long count) { Debug.Assert(count != 0); @@ -161,7 +192,7 @@ private void OnAddOverflow(uint count) counter._lock.Acquire(); try { - counter._overflowCount += (long)_count + count; + counter._overflowCount += _count + count; _count = 0; } finally diff --git a/src/libraries/System.Private.CoreLib/src/System/Threading/ThreadPoolWorkQueue.cs b/src/libraries/System.Private.CoreLib/src/System/Threading/ThreadPoolWorkQueue.cs index 696c2960b05b5b..ebd1e58bae4fe1 100644 --- a/src/libraries/System.Private.CoreLib/src/System/Threading/ThreadPoolWorkQueue.cs +++ b/src/libraries/System.Private.CoreLib/src/System/Threading/ThreadPoolWorkQueue.cs @@ -1204,6 +1204,9 @@ void IThreadPoolWorkItem.Execute() Interlocked.MemoryBarrier(); if (!_workItems.TryDequeue(out T workItem)) { + // Discount a work item here to avoid counting this queue processing work item + ThreadInt64PersistentCounter.Decrement( + ThreadPoolWorkQueueThreadLocals.threadLocals!.threadLocalCompletionCountObject!); return; } @@ -1247,7 +1250,11 @@ void IThreadPoolWorkItem.Execute() currentThread.ResetThreadPoolThread(); } - ThreadInt64PersistentCounter.Add(tl.threadLocalCompletionCountObject!, completedCount); + // Discount a work item here to avoid counting this queue processing work item + if (completedCount > 1) + { + ThreadInt64PersistentCounter.Add(tl.threadLocalCompletionCountObject!, completedCount - 1); + } } } diff --git a/src/libraries/System.Threading.ThreadPool/tests/System.Threading.ThreadPool.Tests.csproj b/src/libraries/System.Threading.ThreadPool/tests/System.Threading.ThreadPool.Tests.csproj index 0cb21c9d38492b..a50fbed0ffb702 100644 --- a/src/libraries/System.Threading.ThreadPool/tests/System.Threading.ThreadPool.Tests.csproj +++ b/src/libraries/System.Threading.ThreadPool/tests/System.Threading.ThreadPool.Tests.csproj @@ -2,6 +2,7 @@ true $(NetCoreAppCurrent) + true true diff --git a/src/libraries/System.Threading.ThreadPool/tests/ThreadPoolTests.cs b/src/libraries/System.Threading.ThreadPool/tests/ThreadPoolTests.cs index 59764268029fac..2d305482d9e114 100644 --- a/src/libraries/System.Threading.ThreadPool/tests/ThreadPoolTests.cs +++ b/src/libraries/System.Threading.ThreadPool/tests/ThreadPoolTests.cs @@ -1265,6 +1265,42 @@ public static void PrioritizationExperimentConfigVarTest() }).Dispose(); } + + [ConditionalFact(nameof(IsThreadingAndRemoteExecutorSupported))] + [PlatformSpecific(TestPlatforms.Windows)] + public static unsafe void ThreadPoolCompletedWorkItemCountTest() + { + // Run in a separate process to test in a clean thread pool environment such that we don't count external work items + RemoteExecutor.Invoke(() => + { + const int WorkItemCount = 4; + + int completedWorkItemCount = 0; + using var allWorkItemsCompleted = new AutoResetEvent(false); + + IOCompletionCallback callback = + (errorCode, numBytes, innerNativeOverlapped) => + { + Overlapped.Free(innerNativeOverlapped); + if (Interlocked.Increment(ref completedWorkItemCount) == WorkItemCount) + { + allWorkItemsCompleted.Set(); + } + }; + for (int i = 0; i < WorkItemCount; i++) + { + ThreadPool.UnsafeQueueNativeOverlapped(new Overlapped().Pack(callback, null)); + } + + allWorkItemsCompleted.CheckedWait(); + + // Allow work items to be marked as completed during this time + ThreadTestHelpers.WaitForCondition(() => ThreadPool.CompletedWorkItemCount >= WorkItemCount); + Thread.Sleep(50); + Assert.Equal(WorkItemCount, ThreadPool.CompletedWorkItemCount); + }).Dispose(); + } + public static bool IsThreadingAndRemoteExecutorSupported => PlatformDetection.IsThreadingSupported && RemoteExecutor.IsSupported;