From 5af699682d0833fcca2e33acf623b84d9c00651f Mon Sep 17 00:00:00 2001 From: Koundinya Veluri Date: Fri, 22 Apr 2022 14:55:17 -0700 Subject: [PATCH 1/3] Fix some scaling issues with the global queue in the thread pool - The global concurrent queue is not scaling well in some situations on machines with a large number of processors. A lot of contention is seen in dequeuing work on some benchmarks. - I initially tried switching to queuing work from thread pool threads more independently with more efficient work stealing, but it looks like that may need more investigation/experimentation. This is a simpler change that seems to work reasonably well for now. - Beyond 32 procs, added additional concurrent queues, one per 16 procs. When a worker thread begins dispatching work, it assigns one of those additional queues to itself, trying to limit assignments to 16 worker threads per queue. - Work items queued by a worker thread queues to the assigned queue. The worker thread dequeues from the assigned queue first, and later tries to dequeue from other queues. Work items queued by non-thread-pool threads continue to go into the global queue. - When a worker thread stops dispatching work items, it unassigns itself from the queue, and may transfer work items from it if it was the last thread assigned to the queue. - In the observed cases, work items are distributed to the different queues and contention is reduced due to a limited number of threads operating on each queue --- .../System/Threading/ThreadPoolWorkQueue.cs | 236 +++++++++++++++++- 1 file changed, 225 insertions(+), 11 deletions(-) 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 840e0b68e7bf59..a967dfc5ee6a76 100644 --- a/src/libraries/System.Private.CoreLib/src/System/Threading/ThreadPoolWorkQueue.cs +++ b/src/libraries/System.Private.CoreLib/src/System/Threading/ThreadPoolWorkQueue.cs @@ -387,6 +387,11 @@ public int Count } } + private const int ProcessorsPerAssignableWorkItemQueue = 16; + private static readonly int AssignableWorkItemQueueCount = + Environment.ProcessorCount <= 32 ? 0 : + (Environment.ProcessorCount + (ProcessorsPerAssignableWorkItemQueue - 1)) / ProcessorsPerAssignableWorkItemQueue; + private bool _loggingEnabled; private bool _dispatchNormalPriorityWorkFirst; private int _mayHaveHighPriorityWorkItems; @@ -395,6 +400,16 @@ public int Count internal readonly ConcurrentQueue workItems = new ConcurrentQueue(); internal readonly ConcurrentQueue highPriorityWorkItems = new ConcurrentQueue(); + // SOS's ThreadPool command depends on the following name. The global queue doesn't scale well beyond a point of + // concurrency. Some additional queues may be added and assigned to a limited number of worker threads if necessary to + // help with limiting the concurrency level. + internal readonly ConcurrentQueue[] assignableWorkItemQueues = + new ConcurrentQueue[AssignableWorkItemQueueCount]; + + private readonly LowLevelLock _queueAssignmentLock = new(); + private readonly int[] _assignedWorkItemQueueThreadCounts = + AssignableWorkItemQueueCount > 0 ? new int[AssignableWorkItemQueueCount] : Array.Empty(); + [StructLayout(LayoutKind.Sequential)] private struct CacheLineSeparated { @@ -409,9 +424,126 @@ private struct CacheLineSeparated public ThreadPoolWorkQueue() { + for (int i = 0; i < AssignableWorkItemQueueCount; i++) + { + assignableWorkItemQueues[i] = new ConcurrentQueue(); + } + RefreshLoggingEnabled(); } + private void AssignWorkItemQueue(ThreadPoolWorkQueueThreadLocals tl) + { + Debug.Assert(AssignableWorkItemQueueCount > 0); + + _queueAssignmentLock.Acquire(); + + // Determine the first queue that has not yet been assigned to the limit of worker threads + int queueIndex = -1; + int minCount = int.MaxValue; + int minCountQueueIndex = 0; + for (int i = 0; i < AssignableWorkItemQueueCount; i++) + { + int count = _assignedWorkItemQueueThreadCounts[i]; + Debug.Assert(count >= 0); + if (count < ProcessorsPerAssignableWorkItemQueue) + { + queueIndex = i; + _assignedWorkItemQueueThreadCounts[queueIndex] = count + 1; + break; + } + + if (count < minCount) + { + minCount = count; + minCountQueueIndex = i; + } + } + + if (queueIndex < 0) + { + // All queues have been fully assigned. Choose the queue that has been assigned to the least number of worker + // threads. + queueIndex = minCountQueueIndex; + _assignedWorkItemQueueThreadCounts[queueIndex]++; + } + + _queueAssignmentLock.Release(); + + tl.queueIndex = queueIndex; + tl.assignedGlobalWorkItemQueue = assignableWorkItemQueues[queueIndex]; + } + + private void TryReassignWorkItemQueue(ThreadPoolWorkQueueThreadLocals tl) + { + Debug.Assert(AssignableWorkItemQueueCount > 0); + + int queueIndex = tl.queueIndex; + if (queueIndex == 0) + { + return; + } + + if (!_queueAssignmentLock.TryAcquire()) + { + return; + } + + // If the currently assigned queue is assigned to other worker threads, try to reassign an earlier queue to this + // worker thread if the earlier queue is not assigned to the limit of worker threads + Debug.Assert(_assignedWorkItemQueueThreadCounts[queueIndex] >= 0); + if (_assignedWorkItemQueueThreadCounts[queueIndex] > 1) + { + for (int i = 0; i < queueIndex; i++) + { + if (_assignedWorkItemQueueThreadCounts[i] < ProcessorsPerAssignableWorkItemQueue) + { + _assignedWorkItemQueueThreadCounts[queueIndex]--; + queueIndex = i; + _assignedWorkItemQueueThreadCounts[queueIndex]++; + break; + } + } + } + + _queueAssignmentLock.Release(); + + tl.queueIndex = queueIndex; + tl.assignedGlobalWorkItemQueue = assignableWorkItemQueues[queueIndex]; + } + + private void UnassignWorkItemQueue(ThreadPoolWorkQueueThreadLocals tl) + { + Debug.Assert(AssignableWorkItemQueueCount > 0); + + int queueIndex = tl.queueIndex; + + _queueAssignmentLock.Acquire(); + int newCount = --_assignedWorkItemQueueThreadCounts[queueIndex]; + _queueAssignmentLock.Release(); + + Debug.Assert(newCount >= 0); + if (newCount > 0) + { + return; + } + + // This queue is not assigned to any other worker threads. Move its work items to the global queue to prevent them + // from being starved for a long duration. + bool movedWorkItem = false; + ConcurrentQueue queue = tl.assignedGlobalWorkItemQueue; + while (_assignedWorkItemQueueThreadCounts[queueIndex] <= 0 && queue.TryDequeue(out object? workItem)) + { + workItems.Enqueue(workItem); + movedWorkItem = true; + } + + if (movedWorkItem) + { + EnsureThreadRequested(); + } + } + public ThreadPoolWorkQueueThreadLocals GetOrCreateThreadLocals() => ThreadPoolWorkQueueThreadLocals.threadLocals ?? CreateThreadLocals(); @@ -479,7 +611,11 @@ public void Enqueue(object callback, bool forceGlobal) } else { - workItems.Enqueue(callback); + ConcurrentQueue queue = + AssignableWorkItemQueueCount > 0 && (tl = ThreadPoolWorkQueueThreadLocals.threadLocals) != null + ? tl.assignedGlobalWorkItemQueue + : workItems; + queue.Enqueue(callback); } EnsureThreadRequested(); @@ -533,22 +669,42 @@ internal static bool LocalFindAndPop(object callback) return workItem; } - // Check for global work items + // Check for work items from the assigned global queue + if (AssignableWorkItemQueueCount > 0 && tl.assignedGlobalWorkItemQueue.TryDequeue(out workItem)) + { + return workItem; + } + + // Check for work items from the global queue if (workItems.TryDequeue(out workItem)) { return workItem; } + // Check for work items in other assignable global queues + uint randomValue = tl.random.NextUInt32(); + if (AssignableWorkItemQueueCount > 0) + { + int queueIndex = tl.queueIndex; + for (int c = AssignableWorkItemQueueCount, maxIndex = c - 1, i = (int)(randomValue % (uint)c); + c > 0; + i = i < maxIndex ? i + 1 : 0, c--) + { + if (i != queueIndex && assignableWorkItemQueues[i].TryDequeue(out workItem)) + { + return workItem; + } + } + } + // Try to steal from other threads' local work items WorkStealingQueue localWsq = tl.workStealingQueue; WorkStealingQueue[] queues = WorkStealingQueueList.Queues; - int c = queues.Length; - Debug.Assert(c > 0, "There must at least be a queue for this thread."); - int maxIndex = c - 1; - uint i = tl.random.NextUInt32() % (uint)c; - while (c > 0) + Debug.Assert(queues.Length > 0, "There must at least be a queue for this thread."); + for (int c = queues.Length, maxIndex = c - 1, i = (int)(randomValue % (uint)c); + c > 0; + i = i < maxIndex ? i + 1 : 0, c--) { - i = (i < maxIndex) ? i + 1 : 0; WorkStealingQueue otherQueue = queues[i]; if (otherQueue != localWsq && otherQueue.CanSteal) { @@ -558,7 +714,6 @@ internal static bool LocalFindAndPop(object callback) return workItem; } } - c--; } return null; @@ -594,7 +749,19 @@ public static long LocalCount } } - public long GlobalCount => (long)highPriorityWorkItems.Count + workItems.Count; + public long GlobalCount + { + get + { + long count = (long)highPriorityWorkItems.Count + workItems.Count; + for (int i = 0; i < AssignableWorkItemQueueCount; i++) + { + count += assignableWorkItemQueues[i].Count; + } + + return count; + } + } // Time in ms for which ThreadPoolWorkQueue.Dispatch keeps executing normal work items before either returning from // Dispatch (if YieldFromDispatchLoop is true), or performing periodic activities @@ -612,6 +779,11 @@ internal static bool Dispatch() ThreadPoolWorkQueue workQueue = ThreadPool.s_workQueue; ThreadPoolWorkQueueThreadLocals tl = workQueue.GetOrCreateThreadLocals(); + if (AssignableWorkItemQueueCount > 0) + { + workQueue.AssignWorkItemQueue(tl); + } + // Before dequeuing the first work item, acknowledge that the thread request has been satisfied workQueue.MarkThreadRequestSatisfied(); @@ -625,7 +797,12 @@ internal static bool Dispatch() if (dispatchNormalPriorityWorkFirst && !tl.workStealingQueue.CanSteal) { workQueue._dispatchNormalPriorityWorkFirst = !dispatchNormalPriorityWorkFirst; - workQueue.workItems.TryDequeue(out workItem); + ConcurrentQueue queue = + AssignableWorkItemQueueCount > 0 ? tl.assignedGlobalWorkItemQueue : workQueue.workItems; + if (!queue.TryDequeue(out workItem) && AssignableWorkItemQueueCount > 0) + { + workQueue.workItems.TryDequeue(out workItem); + } } if (workItem == null) @@ -635,6 +812,11 @@ internal static bool Dispatch() if (workItem == null) { + if (AssignableWorkItemQueueCount > 0) + { + workQueue.UnassignWorkItemQueue(tl); + } + // // No work. // If we missed a steal, though, there may be more work in the queue. @@ -689,6 +871,11 @@ internal static bool Dispatch() if (workItem == null) { + if (AssignableWorkItemQueueCount > 0) + { + workQueue.UnassignWorkItemQueue(tl); + } + // // No work. // If we missed a steal, though, there may be more work in the queue. @@ -753,6 +940,10 @@ internal static bool Dispatch() // processing work items. tl.TransferLocalWork(); tl.isProcessingHighPriorityWorkItems = false; + if (AssignableWorkItemQueueCount > 0) + { + workQueue.UnassignWorkItemQueue(tl); + } return false; } @@ -769,9 +960,20 @@ internal static bool Dispatch() // The runtime-specific thread pool implementation requires the Dispatch loop to return to the VM // periodically to let it perform its own work tl.isProcessingHighPriorityWorkItems = false; + if (AssignableWorkItemQueueCount > 0) + { + workQueue.UnassignWorkItemQueue(tl); + } return true; } + if (AssignableWorkItemQueueCount > 0) + { + // Due to hill climbing, over time arbitrary worker threads may stop working and eventually unbalance the + // queue assignments. Periodically try to reassign a queue to keep the assigned queues busy. + workQueue.TryReassignWorkItemQueue(tl); + } + // This method will continue to dispatch work items. Refresh the start tick count for the next dispatch // quantum and do some periodic activities. startTickCount = currentTickCount; @@ -823,6 +1025,8 @@ internal sealed class ThreadPoolWorkQueueThreadLocals public static ThreadPoolWorkQueueThreadLocals? threadLocals; public bool isProcessingHighPriorityWorkItems; + public int queueIndex; + public ConcurrentQueue assignedGlobalWorkItemQueue; public readonly ThreadPoolWorkQueue workQueue; public readonly ThreadPoolWorkQueue.WorkStealingQueue workStealingQueue; public readonly Thread currentThread; @@ -831,6 +1035,7 @@ internal sealed class ThreadPoolWorkQueueThreadLocals public ThreadPoolWorkQueueThreadLocals(ThreadPoolWorkQueue tpq) { + assignedGlobalWorkItemQueue = tpq.workItems; workQueue = tpq; workStealingQueue = new ThreadPoolWorkQueue.WorkStealingQueue(); ThreadPoolWorkQueue.WorkStealingQueueList.Add(workStealingQueue); @@ -1406,6 +1611,15 @@ internal static IEnumerable GetQueuedWorkItems() yield return workItem; } + // Enumerate assignable global queues + foreach (ConcurrentQueue queue in s_workQueue.assignableWorkItemQueues) + { + foreach (object workItem in queue) + { + yield return workItem; + } + } + // Enumerate global queue foreach (object workItem in s_workQueue.workItems) { From 467a8d9accd18efd301c23d9f1c2812690a58180 Mon Sep 17 00:00:00 2001 From: Koundinya Veluri Date: Thu, 19 May 2022 08:25:53 -0700 Subject: [PATCH 2/3] Rename a couple of fields --- .../System/Threading/ThreadPoolWorkQueue.cs | 56 +++++++++---------- 1 file changed, 28 insertions(+), 28 deletions(-) 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 a967dfc5ee6a76..7e7a249e66ee1c 100644 --- a/src/libraries/System.Private.CoreLib/src/System/Threading/ThreadPoolWorkQueue.cs +++ b/src/libraries/System.Private.CoreLib/src/System/Threading/ThreadPoolWorkQueue.cs @@ -388,7 +388,7 @@ public int Count } private const int ProcessorsPerAssignableWorkItemQueue = 16; - private static readonly int AssignableWorkItemQueueCount = + private static readonly int s_assignableWorkItemQueueCount = Environment.ProcessorCount <= 32 ? 0 : (Environment.ProcessorCount + (ProcessorsPerAssignableWorkItemQueue - 1)) / ProcessorsPerAssignableWorkItemQueue; @@ -403,12 +403,12 @@ public int Count // SOS's ThreadPool command depends on the following name. The global queue doesn't scale well beyond a point of // concurrency. Some additional queues may be added and assigned to a limited number of worker threads if necessary to // help with limiting the concurrency level. - internal readonly ConcurrentQueue[] assignableWorkItemQueues = - new ConcurrentQueue[AssignableWorkItemQueueCount]; + internal readonly ConcurrentQueue[] _assignableWorkItemQueues = + new ConcurrentQueue[s_assignableWorkItemQueueCount]; private readonly LowLevelLock _queueAssignmentLock = new(); private readonly int[] _assignedWorkItemQueueThreadCounts = - AssignableWorkItemQueueCount > 0 ? new int[AssignableWorkItemQueueCount] : Array.Empty(); + s_assignableWorkItemQueueCount > 0 ? new int[s_assignableWorkItemQueueCount] : Array.Empty(); [StructLayout(LayoutKind.Sequential)] private struct CacheLineSeparated @@ -424,9 +424,9 @@ private struct CacheLineSeparated public ThreadPoolWorkQueue() { - for (int i = 0; i < AssignableWorkItemQueueCount; i++) + for (int i = 0; i < s_assignableWorkItemQueueCount; i++) { - assignableWorkItemQueues[i] = new ConcurrentQueue(); + _assignableWorkItemQueues[i] = new ConcurrentQueue(); } RefreshLoggingEnabled(); @@ -434,7 +434,7 @@ public ThreadPoolWorkQueue() private void AssignWorkItemQueue(ThreadPoolWorkQueueThreadLocals tl) { - Debug.Assert(AssignableWorkItemQueueCount > 0); + Debug.Assert(s_assignableWorkItemQueueCount > 0); _queueAssignmentLock.Acquire(); @@ -442,7 +442,7 @@ private void AssignWorkItemQueue(ThreadPoolWorkQueueThreadLocals tl) int queueIndex = -1; int minCount = int.MaxValue; int minCountQueueIndex = 0; - for (int i = 0; i < AssignableWorkItemQueueCount; i++) + for (int i = 0; i < s_assignableWorkItemQueueCount; i++) { int count = _assignedWorkItemQueueThreadCounts[i]; Debug.Assert(count >= 0); @@ -471,12 +471,12 @@ private void AssignWorkItemQueue(ThreadPoolWorkQueueThreadLocals tl) _queueAssignmentLock.Release(); tl.queueIndex = queueIndex; - tl.assignedGlobalWorkItemQueue = assignableWorkItemQueues[queueIndex]; + tl.assignedGlobalWorkItemQueue = _assignableWorkItemQueues[queueIndex]; } private void TryReassignWorkItemQueue(ThreadPoolWorkQueueThreadLocals tl) { - Debug.Assert(AssignableWorkItemQueueCount > 0); + Debug.Assert(s_assignableWorkItemQueueCount > 0); int queueIndex = tl.queueIndex; if (queueIndex == 0) @@ -509,12 +509,12 @@ private void TryReassignWorkItemQueue(ThreadPoolWorkQueueThreadLocals tl) _queueAssignmentLock.Release(); tl.queueIndex = queueIndex; - tl.assignedGlobalWorkItemQueue = assignableWorkItemQueues[queueIndex]; + tl.assignedGlobalWorkItemQueue = _assignableWorkItemQueues[queueIndex]; } private void UnassignWorkItemQueue(ThreadPoolWorkQueueThreadLocals tl) { - Debug.Assert(AssignableWorkItemQueueCount > 0); + Debug.Assert(s_assignableWorkItemQueueCount > 0); int queueIndex = tl.queueIndex; @@ -612,7 +612,7 @@ public void Enqueue(object callback, bool forceGlobal) else { ConcurrentQueue queue = - AssignableWorkItemQueueCount > 0 && (tl = ThreadPoolWorkQueueThreadLocals.threadLocals) != null + s_assignableWorkItemQueueCount > 0 && (tl = ThreadPoolWorkQueueThreadLocals.threadLocals) != null ? tl.assignedGlobalWorkItemQueue : workItems; queue.Enqueue(callback); @@ -670,7 +670,7 @@ internal static bool LocalFindAndPop(object callback) } // Check for work items from the assigned global queue - if (AssignableWorkItemQueueCount > 0 && tl.assignedGlobalWorkItemQueue.TryDequeue(out workItem)) + if (s_assignableWorkItemQueueCount > 0 && tl.assignedGlobalWorkItemQueue.TryDequeue(out workItem)) { return workItem; } @@ -683,14 +683,14 @@ internal static bool LocalFindAndPop(object callback) // Check for work items in other assignable global queues uint randomValue = tl.random.NextUInt32(); - if (AssignableWorkItemQueueCount > 0) + if (s_assignableWorkItemQueueCount > 0) { int queueIndex = tl.queueIndex; - for (int c = AssignableWorkItemQueueCount, maxIndex = c - 1, i = (int)(randomValue % (uint)c); + for (int c = s_assignableWorkItemQueueCount, maxIndex = c - 1, i = (int)(randomValue % (uint)c); c > 0; i = i < maxIndex ? i + 1 : 0, c--) { - if (i != queueIndex && assignableWorkItemQueues[i].TryDequeue(out workItem)) + if (i != queueIndex && _assignableWorkItemQueues[i].TryDequeue(out workItem)) { return workItem; } @@ -754,9 +754,9 @@ public long GlobalCount get { long count = (long)highPriorityWorkItems.Count + workItems.Count; - for (int i = 0; i < AssignableWorkItemQueueCount; i++) + for (int i = 0; i < s_assignableWorkItemQueueCount; i++) { - count += assignableWorkItemQueues[i].Count; + count += _assignableWorkItemQueues[i].Count; } return count; @@ -779,7 +779,7 @@ internal static bool Dispatch() ThreadPoolWorkQueue workQueue = ThreadPool.s_workQueue; ThreadPoolWorkQueueThreadLocals tl = workQueue.GetOrCreateThreadLocals(); - if (AssignableWorkItemQueueCount > 0) + if (s_assignableWorkItemQueueCount > 0) { workQueue.AssignWorkItemQueue(tl); } @@ -798,8 +798,8 @@ internal static bool Dispatch() { workQueue._dispatchNormalPriorityWorkFirst = !dispatchNormalPriorityWorkFirst; ConcurrentQueue queue = - AssignableWorkItemQueueCount > 0 ? tl.assignedGlobalWorkItemQueue : workQueue.workItems; - if (!queue.TryDequeue(out workItem) && AssignableWorkItemQueueCount > 0) + s_assignableWorkItemQueueCount > 0 ? tl.assignedGlobalWorkItemQueue : workQueue.workItems; + if (!queue.TryDequeue(out workItem) && s_assignableWorkItemQueueCount > 0) { workQueue.workItems.TryDequeue(out workItem); } @@ -812,7 +812,7 @@ internal static bool Dispatch() if (workItem == null) { - if (AssignableWorkItemQueueCount > 0) + if (s_assignableWorkItemQueueCount > 0) { workQueue.UnassignWorkItemQueue(tl); } @@ -871,7 +871,7 @@ internal static bool Dispatch() if (workItem == null) { - if (AssignableWorkItemQueueCount > 0) + if (s_assignableWorkItemQueueCount > 0) { workQueue.UnassignWorkItemQueue(tl); } @@ -940,7 +940,7 @@ internal static bool Dispatch() // processing work items. tl.TransferLocalWork(); tl.isProcessingHighPriorityWorkItems = false; - if (AssignableWorkItemQueueCount > 0) + if (s_assignableWorkItemQueueCount > 0) { workQueue.UnassignWorkItemQueue(tl); } @@ -960,14 +960,14 @@ internal static bool Dispatch() // The runtime-specific thread pool implementation requires the Dispatch loop to return to the VM // periodically to let it perform its own work tl.isProcessingHighPriorityWorkItems = false; - if (AssignableWorkItemQueueCount > 0) + if (s_assignableWorkItemQueueCount > 0) { workQueue.UnassignWorkItemQueue(tl); } return true; } - if (AssignableWorkItemQueueCount > 0) + if (s_assignableWorkItemQueueCount > 0) { // Due to hill climbing, over time arbitrary worker threads may stop working and eventually unbalance the // queue assignments. Periodically try to reassign a queue to keep the assigned queues busy. @@ -1612,7 +1612,7 @@ internal static IEnumerable GetQueuedWorkItems() } // Enumerate assignable global queues - foreach (ConcurrentQueue queue in s_workQueue.assignableWorkItemQueues) + foreach (ConcurrentQueue queue in s_workQueue._assignableWorkItemQueues) { foreach (object workItem in queue) { From f28479a67525af4e22a2fdb8606df701b263a090 Mon Sep 17 00:00:00 2001 From: Koundinya Veluri Date: Thu, 2 Jun 2022 22:00:18 -0700 Subject: [PATCH 3/3] Address feedback --- .../System/Threading/ThreadPoolWorkQueue.cs | 30 ++++++++++--------- 1 file changed, 16 insertions(+), 14 deletions(-) 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 7e7a249e66ee1c..34116731072dea 100644 --- a/src/libraries/System.Private.CoreLib/src/System/Threading/ThreadPoolWorkQueue.cs +++ b/src/libraries/System.Private.CoreLib/src/System/Threading/ThreadPoolWorkQueue.cs @@ -686,9 +686,9 @@ internal static bool LocalFindAndPop(object callback) if (s_assignableWorkItemQueueCount > 0) { int queueIndex = tl.queueIndex; - for (int c = s_assignableWorkItemQueueCount, maxIndex = c - 1, i = (int)(randomValue % (uint)c); - c > 0; - i = i < maxIndex ? i + 1 : 0, c--) + int c = s_assignableWorkItemQueueCount; + int maxIndex = c - 1; + for (int i = (int)(randomValue % (uint)c); c > 0; i = i < maxIndex ? i + 1 : 0, c--) { if (i != queueIndex && _assignableWorkItemQueues[i].TryDequeue(out workItem)) { @@ -698,20 +698,22 @@ internal static bool LocalFindAndPop(object callback) } // Try to steal from other threads' local work items - WorkStealingQueue localWsq = tl.workStealingQueue; - WorkStealingQueue[] queues = WorkStealingQueueList.Queues; - Debug.Assert(queues.Length > 0, "There must at least be a queue for this thread."); - for (int c = queues.Length, maxIndex = c - 1, i = (int)(randomValue % (uint)c); - c > 0; - i = i < maxIndex ? i + 1 : 0, c--) { - WorkStealingQueue otherQueue = queues[i]; - if (otherQueue != localWsq && otherQueue.CanSteal) + WorkStealingQueue localWsq = tl.workStealingQueue; + WorkStealingQueue[] queues = WorkStealingQueueList.Queues; + int c = queues.Length; + Debug.Assert(c > 0, "There must at least be a queue for this thread."); + int maxIndex = c - 1; + for (int i = (int)(randomValue % (uint)c); c > 0; i = i < maxIndex ? i + 1 : 0, c--) { - workItem = otherQueue.TrySteal(ref missedSteal); - if (workItem != null) + WorkStealingQueue otherQueue = queues[i]; + if (otherQueue != localWsq && otherQueue.CanSteal) { - return workItem; + workItem = otherQueue.TrySteal(ref missedSteal); + if (workItem != null) + { + return workItem; + } } } }