From 1be7d00ef7b3866c2c88a85827fdff46f5304609 Mon Sep 17 00:00:00 2001 From: Aleksey Kliger Date: Fri, 7 Apr 2023 13:32:02 -0400 Subject: [PATCH 1/4] Split PortableThreadPool.WorkerThread start and loop body For browser-wasm we will need to start the worker thread in a special way, and use callbacks to run the loop body. Current PR is just refactoring existing code. No functional change. --- .../System.Private.CoreLib.Shared.projitems | 1 + ...tableThreadPool.WorkerThread.NonBrowser.cs | 81 +++++++ .../PortableThreadPool.WorkerThread.cs | 223 +++++++----------- 3 files changed, 172 insertions(+), 133 deletions(-) create mode 100644 src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.NonBrowser.cs diff --git a/src/libraries/System.Private.CoreLib/src/System.Private.CoreLib.Shared.projitems b/src/libraries/System.Private.CoreLib/src/System.Private.CoreLib.Shared.projitems index ca9f5435d6a1f..15fd70ea73ace 100644 --- a/src/libraries/System.Private.CoreLib/src/System.Private.CoreLib.Shared.projitems +++ b/src/libraries/System.Private.CoreLib/src/System.Private.CoreLib.Shared.projitems @@ -2539,6 +2539,7 @@ + diff --git a/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.NonBrowser.cs b/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.NonBrowser.cs new file mode 100644 index 0000000000000..c60b5177d8784 --- /dev/null +++ b/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.NonBrowser.cs @@ -0,0 +1,81 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System.Diagnostics.Tracing; + +namespace System.Threading +{ + internal sealed partial class PortableThreadPool + { + /// + /// The worker thread infastructure for the CLR thread pool. + /// + private static partial class WorkerThread + { + + /// + /// Semaphore for controlling how many threads are currently working. + /// + private static readonly LowLevelLifoSemaphore s_semaphore = + new LowLevelLifoSemaphore( + 0, + MaxPossibleThreadCount, + AppContextConfigHelper.GetInt32Config( + "System.Threading.ThreadPool.UnfairSemaphoreSpinLimit", + SemaphoreSpinCountDefault, + false), + onWait: () => + { + if (NativeRuntimeEventSource.Log.IsEnabled()) + { + NativeRuntimeEventSource.Log.ThreadPoolWorkerThreadWait( + (uint)ThreadPoolInstance._separated.counts.VolatileRead().NumExistingThreads); + } + }); + + private static readonly ThreadStart s_workerThreadStart = WorkerThreadStart; + + private static void WorkerThreadStart() + { + Thread.CurrentThread.SetThreadPoolWorkerThreadName(); + + PortableThreadPool threadPoolInstance = ThreadPoolInstance; + + if (NativeRuntimeEventSource.Log.IsEnabled()) + { + NativeRuntimeEventSource.Log.ThreadPoolWorkerThreadStart( + (uint)threadPoolInstance._separated.counts.VolatileRead().NumExistingThreads); + } + + LowLevelLock threadAdjustmentLock = threadPoolInstance._threadAdjustmentLock; + LowLevelLifoSemaphore semaphore = s_semaphore; + + while (true) + { + bool spinWait = true; + while (semaphore.Wait(ThreadPoolThreadTimeoutMs, spinWait)) + { + WorkerDoWork(threadPoolInstance, ref spinWait); + } + + if (WorkerTimedOutMaybeStop(threadPoolInstance, threadAdjustmentLock)) + { + break; + } + } + } + + + private static void CreateWorkerThread() + { + // Thread pool threads must start in the default execution context without transferring the context, so + // using UnsafeStart() instead of Start() + Thread workerThread = new Thread(s_workerThreadStart); + workerThread.IsThreadPoolThread = true; + workerThread.IsBackground = true; + // thread name will be set in thread proc + workerThread.UnsafeStart(); + } + } + } +} diff --git a/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.cs b/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.cs index 96578b9de6b8d..4836a03385d4e 100644 --- a/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.cs +++ b/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.cs @@ -2,6 +2,7 @@ // The .NET Foundation licenses this file to you under the MIT license. using System.Diagnostics.Tracing; +using System.Runtime.CompilerServices; namespace System.Threading { @@ -28,148 +29,115 @@ private static partial class WorkerThread // preexisting threads from running out of memory when using new stack space in low-memory situations. public const int EstimatedAdditionalStackUsagePerThreadBytes = 64 << 10; // 64 KB - /// - /// Semaphore for controlling how many threads are currently working. - /// - private static readonly LowLevelLifoSemaphore s_semaphore = - new LowLevelLifoSemaphore( - 0, - MaxPossibleThreadCount, - AppContextConfigHelper.GetInt32Config( - "System.Threading.ThreadPool.UnfairSemaphoreSpinLimit", - SemaphoreSpinCountDefault, - false), - onWait: () => + [MethodImpl(MethodImplOptions.AggressiveInlining)] + private static void WorkerDoWork(PortableThreadPool threadPoolInstance, ref bool spinWait) + { + bool alreadyRemovedWorkingWorker = false; + while (TakeActiveRequest(threadPoolInstance)) + { + threadPoolInstance._separated.lastDequeueTime = Environment.TickCount; + if (!ThreadPoolWorkQueue.Dispatch()) { - if (NativeRuntimeEventSource.Log.IsEnabled()) - { - NativeRuntimeEventSource.Log.ThreadPoolWorkerThreadWait( - (uint)ThreadPoolInstance._separated.counts.VolatileRead().NumExistingThreads); - } - }); + // ShouldStopProcessingWorkNow() caused the thread to stop processing work, and it would have + // already removed this working worker in the counts. This typically happens when hill climbing + // decreases the worker thread count goal. + alreadyRemovedWorkingWorker = true; + break; + } - private static readonly ThreadStart s_workerThreadStart = WorkerThreadStart; + if (threadPoolInstance._separated.numRequestedWorkers <= 0) + { + break; + } - private static void WorkerThreadStart() - { - Thread.CurrentThread.SetThreadPoolWorkerThreadName(); + // In highly bursty cases with short bursts of work, especially in the portable thread pool + // implementation, worker threads are being released and entering Dispatch very quickly, not finding + // much work in Dispatch, and soon afterwards going back to Dispatch, causing extra thrashing on + // data and some interlocked operations, and similarly when the thread pool runs out of work. Since + // there is a pending request for work, introduce a slight delay before serving the next request. + // The spin-wait is mainly for when the sleep is not effective due to there being no other threads + // to schedule. + Thread.UninterruptibleSleep0(); + if (!Environment.IsSingleProcessor) + { + Thread.SpinWait(1); + } + } - PortableThreadPool threadPoolInstance = ThreadPoolInstance; + // Don't spin-wait on the semaphore next time if the thread was actively stopped from processing work, + // as it's unlikely that the worker thread count goal would be increased again so soon afterwards that + // the semaphore would be released within the spin-wait window + spinWait = !alreadyRemovedWorkingWorker; - if (NativeRuntimeEventSource.Log.IsEnabled()) + if (!alreadyRemovedWorkingWorker) { - NativeRuntimeEventSource.Log.ThreadPoolWorkerThreadStart( - (uint)threadPoolInstance._separated.counts.VolatileRead().NumExistingThreads); + // If we woke up but couldn't find a request, or ran out of work items to process, we need to update + // the number of working workers to reflect that we are done working for now + RemoveWorkingWorker(threadPoolInstance); } + } - LowLevelLock threadAdjustmentLock = threadPoolInstance._threadAdjustmentLock; - LowLevelLifoSemaphore semaphore = s_semaphore; + // returns true if the worker is shutting down + // returns false if we should do another iteration + [MethodImpl(MethodImplOptions.AggressiveInlining)] + private static bool WorkerTimedOutMaybeStop (PortableThreadPool threadPoolInstance, LowLevelLock threadAdjustmentLock) + { + // The thread cannot exit if it has IO pending, otherwise the IO may be canceled + if (IsIOPending) + { + return false; + } - while (true) + threadAdjustmentLock.Acquire(); + try { - bool spinWait = true; - while (semaphore.Wait(ThreadPoolThreadTimeoutMs, spinWait)) + // At this point, the thread's wait timed out. We are shutting down this thread. + // We are going to decrement the number of existing threads to no longer include this one + // and then change the max number of threads in the thread pool to reflect that we don't need as many + // as we had. Finally, we are going to tell hill climbing that we changed the max number of threads. + ThreadCounts counts = threadPoolInstance._separated.counts; + while (true) { - bool alreadyRemovedWorkingWorker = false; - while (TakeActiveRequest(threadPoolInstance)) + // Since this thread is currently registered as an existing thread, if more work comes in meanwhile, + // this thread would be expected to satisfy the new work. Ensure that NumExistingThreads is not + // decreased below NumProcessingWork, as that would be indicative of such a case. + if (counts.NumExistingThreads <= counts.NumProcessingWork) { - threadPoolInstance._separated.lastDequeueTime = Environment.TickCount; - if (!ThreadPoolWorkQueue.Dispatch()) - { - // ShouldStopProcessingWorkNow() caused the thread to stop processing work, and it would have - // already removed this working worker in the counts. This typically happens when hill climbing - // decreases the worker thread count goal. - alreadyRemovedWorkingWorker = true; - break; - } - - if (threadPoolInstance._separated.numRequestedWorkers <= 0) - { - break; - } - - // In highly bursty cases with short bursts of work, especially in the portable thread pool - // implementation, worker threads are being released and entering Dispatch very quickly, not finding - // much work in Dispatch, and soon afterwards going back to Dispatch, causing extra thrashing on - // data and some interlocked operations, and similarly when the thread pool runs out of work. Since - // there is a pending request for work, introduce a slight delay before serving the next request. - // The spin-wait is mainly for when the sleep is not effective due to there being no other threads - // to schedule. - Thread.UninterruptibleSleep0(); - if (!Environment.IsSingleProcessor) - { - Thread.SpinWait(1); - } + // In this case, enough work came in that this thread should not time out and should go back to work. + break; } - // Don't spin-wait on the semaphore next time if the thread was actively stopped from processing work, - // as it's unlikely that the worker thread count goal would be increased again so soon afterwards that - // the semaphore would be released within the spin-wait window - spinWait = !alreadyRemovedWorkingWorker; - - if (!alreadyRemovedWorkingWorker) + ThreadCounts newCounts = counts; + short newNumExistingThreads = --newCounts.NumExistingThreads; + short newNumThreadsGoal = + Math.Max( + threadPoolInstance.MinThreadsGoal, + Math.Min(newNumExistingThreads, counts.NumThreadsGoal)); + newCounts.NumThreadsGoal = newNumThreadsGoal; + + ThreadCounts oldCounts = + threadPoolInstance._separated.counts.InterlockedCompareExchange(newCounts, counts); + if (oldCounts == counts) { - // If we woke up but couldn't find a request, or ran out of work items to process, we need to update - // the number of working workers to reflect that we are done working for now - RemoveWorkingWorker(threadPoolInstance); - } - } - - // The thread cannot exit if it has IO pending, otherwise the IO may be canceled - if (IsIOPending) - { - continue; - } - - threadAdjustmentLock.Acquire(); - try - { - // At this point, the thread's wait timed out. We are shutting down this thread. - // We are going to decrement the number of existing threads to no longer include this one - // and then change the max number of threads in the thread pool to reflect that we don't need as many - // as we had. Finally, we are going to tell hill climbing that we changed the max number of threads. - ThreadCounts counts = threadPoolInstance._separated.counts; - while (true) - { - // Since this thread is currently registered as an existing thread, if more work comes in meanwhile, - // this thread would be expected to satisfy the new work. Ensure that NumExistingThreads is not - // decreased below NumProcessingWork, as that would be indicative of such a case. - if (counts.NumExistingThreads <= counts.NumProcessingWork) + HillClimbing.ThreadPoolHillClimber.ForceChange( + newNumThreadsGoal, + HillClimbing.StateOrTransition.ThreadTimedOut); + if (NativeRuntimeEventSource.Log.IsEnabled()) { - // In this case, enough work came in that this thread should not time out and should go back to work. - break; + NativeRuntimeEventSource.Log.ThreadPoolWorkerThreadStop((uint)newNumExistingThreads); } - - ThreadCounts newCounts = counts; - short newNumExistingThreads = --newCounts.NumExistingThreads; - short newNumThreadsGoal = - Math.Max( - threadPoolInstance.MinThreadsGoal, - Math.Min(newNumExistingThreads, counts.NumThreadsGoal)); - newCounts.NumThreadsGoal = newNumThreadsGoal; - - ThreadCounts oldCounts = - threadPoolInstance._separated.counts.InterlockedCompareExchange(newCounts, counts); - if (oldCounts == counts) - { - HillClimbing.ThreadPoolHillClimber.ForceChange( - newNumThreadsGoal, - HillClimbing.StateOrTransition.ThreadTimedOut); - if (NativeRuntimeEventSource.Log.IsEnabled()) - { - NativeRuntimeEventSource.Log.ThreadPoolWorkerThreadStop((uint)newNumExistingThreads); - } - return; - } - - counts = oldCounts; + return true; } - } - finally - { - threadAdjustmentLock.Release(); + + counts = oldCounts; } } + finally + { + threadAdjustmentLock.Release(); + } + // if we get here new work came in and we're going to keep running + return false; } /// @@ -300,17 +268,6 @@ private static bool TakeActiveRequest(PortableThreadPool threadPoolInstance) } return false; } - - private static void CreateWorkerThread() - { - // Thread pool threads must start in the default execution context without transferring the context, so - // using UnsafeStart() instead of Start() - Thread workerThread = new Thread(s_workerThreadStart); - workerThread.IsThreadPoolThread = true; - workerThread.IsBackground = true; - // thread name will be set in thread proc - workerThread.UnsafeStart(); - } } } } From 42d0d92bf4843a760a1f22da1f8c4ea935bf0c54 Mon Sep 17 00:00:00 2001 From: Aleksey Kliger Date: Tue, 18 Apr 2023 10:00:47 -0400 Subject: [PATCH 2/4] fix indentation --- .../Threading/PortableThreadPool.WorkerThread.NonBrowser.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.NonBrowser.cs b/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.NonBrowser.cs index c60b5177d8784..a35b2760e6733 100644 --- a/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.NonBrowser.cs +++ b/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.NonBrowser.cs @@ -55,12 +55,12 @@ private static void WorkerThreadStart() bool spinWait = true; while (semaphore.Wait(ThreadPoolThreadTimeoutMs, spinWait)) { - WorkerDoWork(threadPoolInstance, ref spinWait); + WorkerDoWork(threadPoolInstance, ref spinWait); } if (WorkerTimedOutMaybeStop(threadPoolInstance, threadAdjustmentLock)) { - break; + break; } } } From b4cc691e4804d5541bb2158b1961615cb01f0fd1 Mon Sep 17 00:00:00 2001 From: Aleksey Kliger Date: Tue, 18 Apr 2023 10:01:31 -0400 Subject: [PATCH 3/4] Change loop to use return instead of break --- .../src/System/Threading/PortableThreadPool.WorkerThread.cs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.cs b/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.cs index 4836a03385d4e..e0ab16257b7d0 100644 --- a/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.cs +++ b/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.cs @@ -104,7 +104,7 @@ private static bool WorkerTimedOutMaybeStop (PortableThreadPool threadPoolInstan if (counts.NumExistingThreads <= counts.NumProcessingWork) { // In this case, enough work came in that this thread should not time out and should go back to work. - break; + return false; } ThreadCounts newCounts = counts; @@ -136,8 +136,6 @@ private static bool WorkerTimedOutMaybeStop (PortableThreadPool threadPoolInstan { threadAdjustmentLock.Release(); } - // if we get here new work came in and we're going to keep running - return false; } /// From b30ca89d4ce0e07c4defeb76e9671672b9e89a50 Mon Sep 17 00:00:00 2001 From: Aleksey Kliger Date: Tue, 18 Apr 2023 10:02:04 -0400 Subject: [PATCH 4/4] rename utility method to ShouldExitWorker --- .../Threading/PortableThreadPool.WorkerThread.NonBrowser.cs | 2 +- .../src/System/Threading/PortableThreadPool.WorkerThread.cs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.NonBrowser.cs b/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.NonBrowser.cs index a35b2760e6733..2e634fd469d9d 100644 --- a/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.NonBrowser.cs +++ b/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.NonBrowser.cs @@ -58,7 +58,7 @@ private static void WorkerThreadStart() WorkerDoWork(threadPoolInstance, ref spinWait); } - if (WorkerTimedOutMaybeStop(threadPoolInstance, threadAdjustmentLock)) + if (ShouldExitWorker(threadPoolInstance, threadAdjustmentLock)) { break; } diff --git a/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.cs b/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.cs index e0ab16257b7d0..b8aac3a896b11 100644 --- a/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.cs +++ b/src/libraries/System.Private.CoreLib/src/System/Threading/PortableThreadPool.WorkerThread.cs @@ -80,7 +80,7 @@ private static void WorkerDoWork(PortableThreadPool threadPoolInstance, ref bool // returns true if the worker is shutting down // returns false if we should do another iteration [MethodImpl(MethodImplOptions.AggressiveInlining)] - private static bool WorkerTimedOutMaybeStop (PortableThreadPool threadPoolInstance, LowLevelLock threadAdjustmentLock) + private static bool ShouldExitWorker (PortableThreadPool threadPoolInstance, LowLevelLock threadAdjustmentLock) { // The thread cannot exit if it has IO pending, otherwise the IO may be canceled if (IsIOPending)