1 // Licensed to the .NET Foundation under one or more agreements.
2 // The .NET Foundation licenses this file to you under the MIT license.
3 // See the LICENSE file in the project root for more information.
5 /*=============================================================================
9 ** Purpose: Class for creating and managing a threadpool
12 =============================================================================*/
14 using System.Collections.Concurrent;
15 using System.Collections.Generic;
16 using System.Diagnostics;
17 using System.Diagnostics.CodeAnalysis;
18 using System.Diagnostics.Tracing;
19 using System.Runtime.CompilerServices;
20 using System.Runtime.InteropServices;
21 using System.Threading.Tasks;
22 using Internal.Runtime.CompilerServices;
24 namespace System.Threading
26 internal static class ThreadPoolGlobals
28 public static readonly int processorCount = Environment.ProcessorCount;
30 public static volatile bool threadPoolInitialized;
31 public static bool enableWorkerTracking;
33 public static readonly ThreadPoolWorkQueue workQueue = new ThreadPoolWorkQueue();
35 /// <summary>Shim used to invoke <see cref="IAsyncStateMachineBox.MoveNext"/> of the supplied <see cref="IAsyncStateMachineBox"/>.</summary>
36 internal static readonly Action<object?> s_invokeAsyncStateMachineBox = state =>
38 if (!(state is IAsyncStateMachineBox box))
40 ThrowHelper.ThrowArgumentOutOfRangeException(ExceptionArgument.state);
48 [StructLayout(LayoutKind.Sequential)] // enforce layout so that padding reduces false sharing
49 internal sealed class ThreadPoolWorkQueue
51 internal static class WorkStealingQueueList
53 private static volatile WorkStealingQueue[] _queues = new WorkStealingQueue[0];
55 public static WorkStealingQueue[] Queues => _queues;
57 public static void Add(WorkStealingQueue queue)
59 Debug.Assert(queue != null);
62 WorkStealingQueue[] oldQueues = _queues;
63 Debug.Assert(Array.IndexOf(oldQueues, queue) == -1);
65 var newQueues = new WorkStealingQueue[oldQueues.Length + 1];
66 Array.Copy(oldQueues, 0, newQueues, 0, oldQueues.Length);
67 newQueues[newQueues.Length - 1] = queue;
68 if (Interlocked.CompareExchange(ref _queues, newQueues, oldQueues) == oldQueues)
75 public static void Remove(WorkStealingQueue queue)
77 Debug.Assert(queue != null);
80 WorkStealingQueue[] oldQueues = _queues;
81 if (oldQueues.Length == 0)
86 int pos = Array.IndexOf(oldQueues, queue);
89 Debug.Fail("Should have found the queue");
93 var newQueues = new WorkStealingQueue[oldQueues.Length - 1];
96 Array.Copy(oldQueues, 1, newQueues, 0, newQueues.Length);
98 else if (pos == oldQueues.Length - 1)
100 Array.Copy(oldQueues, 0, newQueues, 0, newQueues.Length);
104 Array.Copy(oldQueues, 0, newQueues, 0, pos);
105 Array.Copy(oldQueues, pos + 1, newQueues, pos, newQueues.Length - pos);
108 if (Interlocked.CompareExchange(ref _queues, newQueues, oldQueues) == oldQueues)
116 internal sealed class WorkStealingQueue
118 private const int INITIAL_SIZE = 32;
119 internal volatile object?[] m_array = new object[INITIAL_SIZE]; // SOS's ThreadPool command depends on this name
120 private volatile int m_mask = INITIAL_SIZE - 1;
123 // in debug builds, start at the end so we exercise the index reset logic.
124 private const int START_INDEX = int.MaxValue;
126 private const int START_INDEX = 0;
129 private volatile int m_headIndex = START_INDEX;
130 private volatile int m_tailIndex = START_INDEX;
132 private SpinLock m_foreignLock = new SpinLock(enableThreadOwnerTracking: false);
134 public void LocalPush(object obj)
136 int tail = m_tailIndex;
138 // We're going to increment the tail; if we'll overflow, then we need to reset our counts
139 if (tail == int.MaxValue)
141 bool lockTaken = false;
144 m_foreignLock.Enter(ref lockTaken);
146 if (m_tailIndex == int.MaxValue)
149 // Rather than resetting to zero, we'll just mask off the bits we don't care about.
150 // This way we don't need to rearrange the items already in the queue; they'll be found
151 // correctly exactly where they are. One subtlety here is that we need to make sure that
152 // if head is currently < tail, it remains that way. This happens to just fall out from
153 // the bit-masking, because we only do this if tail == int.MaxValue, meaning that all
154 // bits are set, so all of the bits we're keeping will also be set. Thus it's impossible
155 // for the head to end up > than the tail, since you can't set any more bits than all of
158 m_headIndex = m_headIndex & m_mask;
159 m_tailIndex = tail = m_tailIndex & m_mask;
160 Debug.Assert(m_headIndex <= m_tailIndex);
166 m_foreignLock.Exit(useMemoryBarrier: true);
170 // When there are at least 2 elements' worth of space, we can take the fast path.
171 if (tail < m_headIndex + m_mask)
173 Volatile.Write(ref m_array[tail & m_mask], obj);
174 m_tailIndex = tail + 1;
178 // We need to contend with foreign pops, so we lock.
179 bool lockTaken = false;
182 m_foreignLock.Enter(ref lockTaken);
184 int head = m_headIndex;
185 int count = m_tailIndex - m_headIndex;
187 // If there is still space (one left), just add the element.
190 // We're full; expand the queue by doubling its size.
191 var newArray = new object?[m_array.Length << 1];
192 for (int i = 0; i < m_array.Length; i++)
193 newArray[i] = m_array[(i + head) & m_mask];
195 // Reset the field values, incl. the mask.
198 m_tailIndex = tail = count;
199 m_mask = (m_mask << 1) | 1;
202 Volatile.Write(ref m_array[tail & m_mask], obj);
203 m_tailIndex = tail + 1;
208 m_foreignLock.Exit(useMemoryBarrier: false);
213 [SuppressMessage("Microsoft.Concurrency", "CA8001", Justification = "Reviewed for thread safety")]
214 public bool LocalFindAndPop(object obj)
216 // Fast path: check the tail. If equal, we can skip the lock.
217 if (m_array[(m_tailIndex - 1) & m_mask] == obj)
219 object? unused = LocalPop();
220 Debug.Assert(unused == null || unused == obj);
221 return unused != null;
224 // Else, do an O(N) search for the work item. The theory of work stealing and our
225 // inlining logic is that most waits will happen on recently queued work. And
226 // since recently queued work will be close to the tail end (which is where we
227 // begin our search), we will likely find it quickly. In the worst case, we
228 // will traverse the whole local queue; this is typically not going to be a
229 // problem (although degenerate cases are clearly an issue) because local work
230 // queues tend to be somewhat shallow in length, and because if we fail to find
231 // the work item, we are about to block anyway (which is very expensive).
232 for (int i = m_tailIndex - 2; i >= m_headIndex; i--)
234 if (m_array[i & m_mask] == obj)
236 // If we found the element, block out steals to avoid interference.
237 bool lockTaken = false;
240 m_foreignLock.Enter(ref lockTaken);
242 // If we encountered a race condition, bail.
243 if (m_array[i & m_mask] == null)
246 // Otherwise, null out the element.
247 Volatile.Write(ref m_array[i & m_mask], null);
249 // And then check to see if we can fix up the indexes (if we're at
250 // the edge). If we can't, we just leave nulls in the array and they'll
251 // get filtered out eventually (but may lead to superfluous resizing).
252 if (i == m_tailIndex)
254 else if (i == m_headIndex)
262 m_foreignLock.Exit(useMemoryBarrier: false);
270 public object? LocalPop() => m_headIndex < m_tailIndex ? LocalPopCore() : null;
272 [SuppressMessage("Microsoft.Concurrency", "CA8001", Justification = "Reviewed for thread safety")]
273 private object? LocalPopCore()
277 int tail = m_tailIndex;
278 if (m_headIndex >= tail)
283 // Decrement the tail using a fence to ensure subsequent read doesn't come before.
285 Interlocked.Exchange(ref m_tailIndex, tail);
287 // If there is no interaction with a take, we can head down the fast path.
288 if (m_headIndex <= tail)
290 int idx = tail & m_mask;
291 object? obj = Volatile.Read(ref m_array[idx]);
293 // Check for nulls in the array.
294 if (obj == null) continue;
301 // Interaction with takes: 0 or 1 elements left.
302 bool lockTaken = false;
305 m_foreignLock.Enter(ref lockTaken);
307 if (m_headIndex <= tail)
309 // Element still available. Take it.
310 int idx = tail & m_mask;
311 object? obj = Volatile.Read(ref m_array[idx]);
313 // Check for nulls in the array.
314 if (obj == null) continue;
321 // If we encountered a race condition and element was stolen, restore the tail.
322 m_tailIndex = tail + 1;
329 m_foreignLock.Exit(useMemoryBarrier: false);
335 public bool CanSteal => m_headIndex < m_tailIndex;
337 public object? TrySteal(ref bool missedSteal)
346 m_foreignLock.TryEnter(ref taken);
349 // Increment head, and ensure read of tail doesn't move before it (fence).
350 int head = m_headIndex;
351 Interlocked.Exchange(ref m_headIndex, head + 1);
353 if (head < m_tailIndex)
355 int idx = head & m_mask;
356 object? obj = Volatile.Read(ref m_array[idx]);
358 // Check for nulls in the array.
359 if (obj == null) continue;
366 // Failed, restore head.
374 m_foreignLock.Exit(useMemoryBarrier: false);
388 bool lockTaken = false;
391 m_foreignLock.Enter(ref lockTaken);
392 return Math.Max(0, m_tailIndex - m_headIndex);
398 m_foreignLock.Exit(useMemoryBarrier: false);
405 internal bool loggingEnabled;
406 internal readonly ConcurrentQueue<object> workItems = new ConcurrentQueue<object>(); // SOS's ThreadPool command depends on this name
408 private Internal.PaddingFor32 pad1;
410 private volatile int numOutstandingThreadRequests = 0;
412 private Internal.PaddingFor32 pad2;
414 public ThreadPoolWorkQueue()
416 loggingEnabled = FrameworkEventSource.Log.IsEnabled(EventLevel.Verbose, FrameworkEventSource.Keywords.ThreadPool | FrameworkEventSource.Keywords.ThreadTransfer);
419 public ThreadPoolWorkQueueThreadLocals GetOrCreateThreadLocals() =>
420 ThreadPoolWorkQueueThreadLocals.threadLocals ?? CreateThreadLocals();
422 [MethodImpl(MethodImplOptions.NoInlining)]
423 private ThreadPoolWorkQueueThreadLocals CreateThreadLocals()
425 Debug.Assert(ThreadPoolWorkQueueThreadLocals.threadLocals == null);
427 return (ThreadPoolWorkQueueThreadLocals.threadLocals = new ThreadPoolWorkQueueThreadLocals(this));
430 internal void EnsureThreadRequested()
433 // If we have not yet requested #procs threads, then request a new thread.
435 // CoreCLR: Note that there is a separate count in the VM which has already been incremented
436 // by the VM by the time we reach this point.
438 int count = numOutstandingThreadRequests;
439 while (count < ThreadPoolGlobals.processorCount)
441 int prev = Interlocked.CompareExchange(ref numOutstandingThreadRequests, count + 1, count);
444 ThreadPool.RequestWorkerThread();
451 internal void MarkThreadRequestSatisfied()
454 // One of our outstanding thread requests has been satisfied.
455 // Decrement the count so that future calls to EnsureThreadRequested will succeed.
457 // CoreCLR: Note that there is a separate count in the VM which has already been decremented
458 // by the VM by the time we reach this point.
460 int count = numOutstandingThreadRequests;
463 int prev = Interlocked.CompareExchange(ref numOutstandingThreadRequests, count - 1, count);
472 public void Enqueue(object callback, bool forceGlobal)
474 Debug.Assert((callback is IThreadPoolWorkItem) ^ (callback is Task));
477 System.Diagnostics.Tracing.FrameworkEventSource.Log.ThreadPoolEnqueueWorkObject(callback);
479 ThreadPoolWorkQueueThreadLocals? tl = null;
481 tl = ThreadPoolWorkQueueThreadLocals.threadLocals;
485 tl.workStealingQueue.LocalPush(callback);
489 workItems.Enqueue(callback);
492 EnsureThreadRequested();
495 internal bool LocalFindAndPop(object callback)
497 ThreadPoolWorkQueueThreadLocals tl = ThreadPoolWorkQueueThreadLocals.threadLocals;
498 return tl != null && tl.workStealingQueue.LocalFindAndPop(callback);
501 public object? Dequeue(ThreadPoolWorkQueueThreadLocals tl, ref bool missedSteal)
503 WorkStealingQueue localWsq = tl.workStealingQueue;
506 if ((callback = localWsq.LocalPop()) == null && // first try the local queue
507 !workItems.TryDequeue(out callback)) // then try the global queue
509 // finally try to steal from another thread's local queue
510 WorkStealingQueue[] queues = WorkStealingQueueList.Queues;
511 int c = queues.Length;
512 Debug.Assert(c > 0, "There must at least be a queue for this thread.");
513 int maxIndex = c - 1;
514 int i = tl.random.Next(c);
517 i = (i < maxIndex) ? i + 1 : 0;
518 WorkStealingQueue otherQueue = queues[i];
519 if (otherQueue != localWsq && otherQueue.CanSteal)
521 callback = otherQueue.TrySteal(ref missedSteal);
522 if (callback != null)
534 public long LocalCount
539 foreach (WorkStealingQueue workStealingQueue in WorkStealingQueueList.Queues)
541 count += workStealingQueue.Count;
547 public long GlobalCount => workItems.Count;
550 /// Dispatches work items to this thread.
553 /// <c>true</c> if this thread did as much work as was available or its quantum expired.
554 /// <c>false</c> if this thread stopped working early.
556 internal static bool Dispatch()
558 ThreadPoolWorkQueue outerWorkQueue = ThreadPoolGlobals.workQueue;
561 // Save the start time
563 int startTickCount = Environment.TickCount;
566 // Update our records to indicate that an outstanding request for a thread has now been fulfilled.
567 // From this point on, we are responsible for requesting another thread if we stop working for any
568 // reason, and we believe there might still be work in the queue.
570 // CoreCLR: Note that if this thread is aborted before we get a chance to request another one, the VM will
571 // record a thread request on our behalf. So we don't need to worry about getting aborted right here.
573 outerWorkQueue.MarkThreadRequestSatisfied();
575 // Has the desire for logging changed since the last time we entered?
576 outerWorkQueue.loggingEnabled = FrameworkEventSource.Log.IsEnabled(EventLevel.Verbose, FrameworkEventSource.Keywords.ThreadPool | FrameworkEventSource.Keywords.ThreadTransfer);
579 // Assume that we're going to need another thread if this one returns to the VM. We'll set this to
580 // false later, but only if we're absolutely certain that the queue is empty.
582 bool needAnotherThread = true;
583 object? outerWorkItem = null;
587 // Set up our thread-local data
589 // Use operate on workQueue local to try block so it can be enregistered
590 ThreadPoolWorkQueue workQueue = outerWorkQueue;
591 ThreadPoolWorkQueueThreadLocals tl = workQueue.GetOrCreateThreadLocals();
592 Thread currentThread = tl.currentThread;
594 // Start on clean ExecutionContext and SynchronizationContext
595 currentThread._executionContext = null;
596 currentThread._synchronizationContext = null;
599 // Loop until our quantum expires or there is no work.
601 while (ThreadPool.KeepDispatching(startTickCount))
603 bool missedSteal = false;
604 // Use operate on workItem local to try block so it can be enregistered
605 object? workItem = outerWorkItem = workQueue.Dequeue(tl, ref missedSteal);
607 if (workItem == null)
611 // If we missed a steal, though, there may be more work in the queue.
612 // Instead of looping around and trying again, we'll just request another thread. Hopefully the thread
613 // that owns the contended work-stealing queue will pick up its own workitems in the meantime,
614 // which will be more efficient than this thread doing it anyway.
616 needAnotherThread = missedSteal;
618 // Tell the VM we're returning normally, not because Hill Climbing asked us to return.
622 if (workQueue.loggingEnabled)
623 System.Diagnostics.Tracing.FrameworkEventSource.Log.ThreadPoolDequeueWorkObject(workItem);
626 // If we found work, there may be more work. Ask for another thread so that the other work can be processed
627 // in parallel. Note that this will only ask for a max of #procs threads, so it's safe to call it for every dequeue.
629 workQueue.EnsureThreadRequested();
632 // Execute the workitem outside of any finally blocks, so that it can be aborted if needed.
634 if (ThreadPoolGlobals.enableWorkerTracking)
636 bool reportedStatus = false;
639 ThreadPool.ReportThreadStatus(isWorking: true);
640 reportedStatus = true;
641 if (workItem is Task task)
643 task.ExecuteFromThreadPool(currentThread);
647 Debug.Assert(workItem is IThreadPoolWorkItem);
648 Unsafe.As<IThreadPoolWorkItem>(workItem).Execute();
654 ThreadPool.ReportThreadStatus(isWorking: false);
657 else if (workItem is Task task)
659 // Check for Task first as it's currently faster to type check
660 // for Task and then Unsafe.As for the interface, rather than
661 // vice versa, in particular when the object implements a bunch
663 task.ExecuteFromThreadPool(currentThread);
667 Debug.Assert(workItem is IThreadPoolWorkItem);
668 Unsafe.As<IThreadPoolWorkItem>(workItem).Execute();
671 currentThread.ResetThreadPoolThread();
674 outerWorkItem = workItem = null;
676 // Return to clean ExecutionContext and SynchronizationContext
677 ExecutionContext.ResetThreadPoolThread(currentThread);
680 // Notify the VM that we executed this workitem. This is also our opportunity to ask whether Hill Climbing wants
681 // us to return the thread to the pool or not.
683 if (!ThreadPool.NotifyWorkItemComplete())
687 // If we get here, it's because our quantum expired. Tell the VM we're returning normally.
693 // If we are exiting for any reason other than that the queue is definitely empty, ask for another
694 // thread to pick up where we left off.
696 if (needAnotherThread)
697 outerWorkQueue.EnsureThreadRequested();
702 // Simple random number generator. We don't need great randomness, we just need a little and for it to be fast.
703 internal struct FastRandom // xorshift prng
705 private uint _w, _x, _y, _z;
707 public FastRandom(int seed)
715 public int Next(int maxValue)
717 Debug.Assert(maxValue > 0);
719 uint t = _x ^ (_x << 11);
720 _x = _y; _y = _z; _z = _w;
721 _w = _w ^ (_w >> 19) ^ (t ^ (t >> 8));
723 return (int)(_w % (uint)maxValue);
727 // Holds a WorkStealingQueue, and removes it from the list when this object is no longer referenced.
728 internal sealed class ThreadPoolWorkQueueThreadLocals
731 public static ThreadPoolWorkQueueThreadLocals threadLocals;
733 public readonly ThreadPoolWorkQueue workQueue;
734 public readonly ThreadPoolWorkQueue.WorkStealingQueue workStealingQueue;
735 public readonly Thread currentThread;
736 public FastRandom random = new FastRandom(Thread.CurrentThread.ManagedThreadId); // mutable struct, do not copy or make readonly
738 public ThreadPoolWorkQueueThreadLocals(ThreadPoolWorkQueue tpq)
741 workStealingQueue = new ThreadPoolWorkQueue.WorkStealingQueue();
742 ThreadPoolWorkQueue.WorkStealingQueueList.Add(workStealingQueue);
743 currentThread = Thread.CurrentThread;
746 ~ThreadPoolWorkQueueThreadLocals()
748 // Transfer any pending workitems into the global queue so that they will be executed by another thread
749 if (null != workStealingQueue)
751 if (null != workQueue)
754 while ((cb = workStealingQueue.LocalPop()) != null)
756 Debug.Assert(null != cb);
757 workQueue.Enqueue(cb, forceGlobal: true);
761 ThreadPoolWorkQueue.WorkStealingQueueList.Remove(workStealingQueue);
766 public delegate void WaitCallback(object? state);
768 public delegate void WaitOrTimerCallback(object? state, bool timedOut); // signaled or timed out
770 internal abstract class QueueUserWorkItemCallbackBase : IThreadPoolWorkItem
773 private volatile int executed;
775 [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Performance", "CA1821:RemoveEmptyFinalizers")]
776 ~QueueUserWorkItemCallbackBase()
779 executed != 0, "A QueueUserWorkItemCallback was never called!");
783 public virtual void Execute()
786 GC.SuppressFinalize(this);
788 0 == Interlocked.Exchange(ref executed, 1),
789 "A QueueUserWorkItemCallback was called twice!");
794 internal sealed class QueueUserWorkItemCallback : QueueUserWorkItemCallbackBase
796 private WaitCallback? _callback; // SOS's ThreadPool command depends on this name
797 private readonly object? _state;
798 private readonly ExecutionContext _context;
800 private static readonly Action<QueueUserWorkItemCallback> s_executionContextShim = quwi =>
802 Debug.Assert(quwi._callback != null);
803 WaitCallback callback = quwi._callback;
804 quwi._callback = null;
806 callback(quwi._state);
809 internal QueueUserWorkItemCallback(WaitCallback callback, object? state, ExecutionContext context)
811 Debug.Assert(context != null);
813 _callback = callback;
818 public override void Execute()
822 ExecutionContext.RunForThreadPoolUnsafe(_context, s_executionContextShim, this);
826 internal sealed class QueueUserWorkItemCallback<TState> : QueueUserWorkItemCallbackBase
828 private Action<TState>? _callback; // SOS's ThreadPool command depends on this name
829 private readonly TState _state;
830 private readonly ExecutionContext _context;
832 internal QueueUserWorkItemCallback(Action<TState> callback, TState state, ExecutionContext context)
834 Debug.Assert(callback != null);
836 _callback = callback;
841 public override void Execute()
845 Debug.Assert(_callback != null);
846 Action<TState> callback = _callback;
849 ExecutionContext.RunForThreadPoolUnsafe(_context, callback, in _state);
853 internal sealed class QueueUserWorkItemCallbackDefaultContext : QueueUserWorkItemCallbackBase
855 private WaitCallback? _callback; // SOS's ThreadPool command depends on this name
856 private readonly object? _state;
858 internal QueueUserWorkItemCallbackDefaultContext(WaitCallback callback, object? state)
860 Debug.Assert(callback != null);
862 _callback = callback;
866 public override void Execute()
868 ExecutionContext.CheckThreadPoolAndContextsAreDefault();
871 Debug.Assert(_callback != null);
872 WaitCallback callback = _callback;
877 // ThreadPoolWorkQueue.Dispatch will handle notifications and reset EC and SyncCtx back to default
881 internal sealed class QueueUserWorkItemCallbackDefaultContext<TState> : QueueUserWorkItemCallbackBase
883 private Action<TState>? _callback; // SOS's ThreadPool command depends on this name
884 private readonly TState _state;
886 internal QueueUserWorkItemCallbackDefaultContext(Action<TState> callback, TState state)
888 Debug.Assert(callback != null);
890 _callback = callback;
894 public override void Execute()
896 ExecutionContext.CheckThreadPoolAndContextsAreDefault();
899 Debug.Assert(_callback != null);
900 Action<TState> callback = _callback;
905 // ThreadPoolWorkQueue.Dispatch will handle notifications and reset EC and SyncCtx back to default
909 internal sealed class _ThreadPoolWaitOrTimerCallback
911 private WaitOrTimerCallback _waitOrTimerCallback;
912 private ExecutionContext? _executionContext;
913 private object? _state;
914 private static readonly ContextCallback _ccbt = new ContextCallback(WaitOrTimerCallback_Context_t);
915 private static readonly ContextCallback _ccbf = new ContextCallback(WaitOrTimerCallback_Context_f);
917 internal _ThreadPoolWaitOrTimerCallback(WaitOrTimerCallback waitOrTimerCallback, object? state, bool flowExecutionContext)
919 _waitOrTimerCallback = waitOrTimerCallback;
922 if (flowExecutionContext)
924 // capture the exection context
925 _executionContext = ExecutionContext.Capture();
929 private static void WaitOrTimerCallback_Context_t(object? state) =>
930 WaitOrTimerCallback_Context(state, timedOut: true);
932 private static void WaitOrTimerCallback_Context_f(object? state) =>
933 WaitOrTimerCallback_Context(state, timedOut: false);
935 private static void WaitOrTimerCallback_Context(object? state, bool timedOut)
937 _ThreadPoolWaitOrTimerCallback helper = (_ThreadPoolWaitOrTimerCallback)state!;
938 helper._waitOrTimerCallback(helper._state, timedOut);
942 internal static void PerformWaitOrTimerCallback(_ThreadPoolWaitOrTimerCallback helper, bool timedOut)
944 Debug.Assert(helper != null, "Null state passed to PerformWaitOrTimerCallback!");
945 // call directly if it is an unsafe call OR EC flow is suppressed
946 ExecutionContext? context = helper._executionContext;
949 WaitOrTimerCallback callback = helper._waitOrTimerCallback;
950 callback(helper._state, timedOut);
954 ExecutionContext.Run(context, timedOut ? _ccbt : _ccbf, helper);
959 public static partial class ThreadPool
961 [CLSCompliant(false)]
962 public static RegisteredWaitHandle RegisterWaitForSingleObject(
963 WaitHandle waitObject,
964 WaitOrTimerCallback callBack,
966 uint millisecondsTimeOutInterval,
967 bool executeOnlyOnce // NOTE: we do not allow other options that allow the callback to be queued as an APC
970 if (millisecondsTimeOutInterval > (uint)int.MaxValue && millisecondsTimeOutInterval != uint.MaxValue)
971 throw new ArgumentOutOfRangeException(nameof(millisecondsTimeOutInterval), SR.ArgumentOutOfRange_LessEqualToIntegerMaxVal);
972 return RegisterWaitForSingleObject(waitObject, callBack, state, millisecondsTimeOutInterval, executeOnlyOnce, true);
975 [CLSCompliant(false)]
976 public static RegisteredWaitHandle UnsafeRegisterWaitForSingleObject(
977 WaitHandle waitObject,
978 WaitOrTimerCallback callBack,
980 uint millisecondsTimeOutInterval,
981 bool executeOnlyOnce // NOTE: we do not allow other options that allow the callback to be queued as an APC
984 if (millisecondsTimeOutInterval > (uint)int.MaxValue && millisecondsTimeOutInterval != uint.MaxValue)
985 throw new ArgumentOutOfRangeException(nameof(millisecondsTimeOutInterval), SR.ArgumentOutOfRange_NeedNonNegOrNegative1);
986 return RegisterWaitForSingleObject(waitObject, callBack, state, millisecondsTimeOutInterval, executeOnlyOnce, false);
989 public static RegisteredWaitHandle RegisterWaitForSingleObject(
990 WaitHandle waitObject,
991 WaitOrTimerCallback callBack,
993 int millisecondsTimeOutInterval,
994 bool executeOnlyOnce // NOTE: we do not allow other options that allow the callback to be queued as an APC
997 if (millisecondsTimeOutInterval < -1)
998 throw new ArgumentOutOfRangeException(nameof(millisecondsTimeOutInterval), SR.ArgumentOutOfRange_NeedNonNegOrNegative1);
999 return RegisterWaitForSingleObject(waitObject, callBack, state, (uint)millisecondsTimeOutInterval, executeOnlyOnce, true);
1002 public static RegisteredWaitHandle UnsafeRegisterWaitForSingleObject(
1003 WaitHandle waitObject,
1004 WaitOrTimerCallback callBack,
1006 int millisecondsTimeOutInterval,
1007 bool executeOnlyOnce // NOTE: we do not allow other options that allow the callback to be queued as an APC
1010 if (millisecondsTimeOutInterval < -1)
1011 throw new ArgumentOutOfRangeException(nameof(millisecondsTimeOutInterval), SR.ArgumentOutOfRange_NeedNonNegOrNegative1);
1012 return RegisterWaitForSingleObject(waitObject, callBack, state, (uint)millisecondsTimeOutInterval, executeOnlyOnce, false);
1015 public static RegisteredWaitHandle RegisterWaitForSingleObject(
1016 WaitHandle waitObject,
1017 WaitOrTimerCallback callBack,
1019 long millisecondsTimeOutInterval,
1020 bool executeOnlyOnce // NOTE: we do not allow other options that allow the callback to be queued as an APC
1023 if (millisecondsTimeOutInterval < -1)
1024 throw new ArgumentOutOfRangeException(nameof(millisecondsTimeOutInterval), SR.ArgumentOutOfRange_NeedNonNegOrNegative1);
1025 if (millisecondsTimeOutInterval > (uint)int.MaxValue)
1026 throw new ArgumentOutOfRangeException(nameof(millisecondsTimeOutInterval), SR.ArgumentOutOfRange_LessEqualToIntegerMaxVal);
1027 return RegisterWaitForSingleObject(waitObject, callBack, state, (uint)millisecondsTimeOutInterval, executeOnlyOnce, true);
1030 public static RegisteredWaitHandle UnsafeRegisterWaitForSingleObject(
1031 WaitHandle waitObject,
1032 WaitOrTimerCallback callBack,
1034 long millisecondsTimeOutInterval,
1035 bool executeOnlyOnce // NOTE: we do not allow other options that allow the callback to be queued as an APC
1038 if (millisecondsTimeOutInterval < -1)
1039 throw new ArgumentOutOfRangeException(nameof(millisecondsTimeOutInterval), SR.ArgumentOutOfRange_NeedNonNegOrNegative1);
1040 if (millisecondsTimeOutInterval > (uint)int.MaxValue)
1041 throw new ArgumentOutOfRangeException(nameof(millisecondsTimeOutInterval), SR.ArgumentOutOfRange_LessEqualToIntegerMaxVal);
1042 return RegisterWaitForSingleObject(waitObject, callBack, state, (uint)millisecondsTimeOutInterval, executeOnlyOnce, false);
1045 public static RegisteredWaitHandle RegisterWaitForSingleObject(
1046 WaitHandle waitObject,
1047 WaitOrTimerCallback callBack,
1050 bool executeOnlyOnce
1053 long tm = (long)timeout.TotalMilliseconds;
1055 throw new ArgumentOutOfRangeException(nameof(timeout), SR.ArgumentOutOfRange_NeedNonNegOrNegative1);
1056 if (tm > (long)int.MaxValue)
1057 throw new ArgumentOutOfRangeException(nameof(timeout), SR.ArgumentOutOfRange_LessEqualToIntegerMaxVal);
1058 return RegisterWaitForSingleObject(waitObject, callBack, state, (uint)tm, executeOnlyOnce, true);
1061 public static RegisteredWaitHandle UnsafeRegisterWaitForSingleObject(
1062 WaitHandle waitObject,
1063 WaitOrTimerCallback callBack,
1066 bool executeOnlyOnce
1069 long tm = (long)timeout.TotalMilliseconds;
1071 throw new ArgumentOutOfRangeException(nameof(timeout), SR.ArgumentOutOfRange_NeedNonNegOrNegative1);
1072 if (tm > (long)int.MaxValue)
1073 throw new ArgumentOutOfRangeException(nameof(timeout), SR.ArgumentOutOfRange_LessEqualToIntegerMaxVal);
1074 return RegisterWaitForSingleObject(waitObject, callBack, state, (uint)tm, executeOnlyOnce, false);
1077 public static bool QueueUserWorkItem(WaitCallback callBack) =>
1078 QueueUserWorkItem(callBack, null);
1080 public static bool QueueUserWorkItem(WaitCallback callBack, object? state)
1082 if (callBack == null)
1084 ThrowHelper.ThrowArgumentNullException(ExceptionArgument.callBack);
1087 EnsureInitialized();
1089 ExecutionContext? context = ExecutionContext.Capture();
1091 object tpcallBack = (context == null || context.IsDefault) ?
1092 new QueueUserWorkItemCallbackDefaultContext(callBack!, state) :
1093 (object)new QueueUserWorkItemCallback(callBack!, state, context);
1095 ThreadPoolGlobals.workQueue.Enqueue(tpcallBack, forceGlobal: true);
1100 public static bool QueueUserWorkItem<TState>(Action<TState> callBack, TState state, bool preferLocal)
1102 if (callBack == null)
1104 ThrowHelper.ThrowArgumentNullException(ExceptionArgument.callBack);
1107 EnsureInitialized();
1109 ExecutionContext? context = ExecutionContext.Capture();
1111 object tpcallBack = (context == null || context.IsDefault) ?
1112 new QueueUserWorkItemCallbackDefaultContext<TState>(callBack!, state) :
1113 (object)new QueueUserWorkItemCallback<TState>(callBack!, state, context);
1115 ThreadPoolGlobals.workQueue.Enqueue(tpcallBack, forceGlobal: !preferLocal);
1120 public static bool UnsafeQueueUserWorkItem<TState>(Action<TState> callBack, TState state, bool preferLocal)
1122 if (callBack == null)
1124 ThrowHelper.ThrowArgumentNullException(ExceptionArgument.callBack);
1127 // If the callback is the runtime-provided invocation of an IAsyncStateMachineBox,
1128 // then we can queue the Task state directly to the ThreadPool instead of
1129 // wrapping it in a QueueUserWorkItemCallback.
1131 // This occurs when user code queues its provided continuation to the ThreadPool;
1132 // internally we call UnsafeQueueUserWorkItemInternal directly for Tasks.
1133 if (ReferenceEquals(callBack, ThreadPoolGlobals.s_invokeAsyncStateMachineBox))
1135 if (!(state is IAsyncStateMachineBox))
1137 // The provided state must be the internal IAsyncStateMachineBox (Task) type
1138 ThrowHelper.ThrowArgumentOutOfRangeException(ExceptionArgument.state);
1141 UnsafeQueueUserWorkItemInternal((object)state!, preferLocal);
1145 EnsureInitialized();
1147 ThreadPoolGlobals.workQueue.Enqueue(
1148 new QueueUserWorkItemCallbackDefaultContext<TState>(callBack!, state), forceGlobal: !preferLocal);
1153 public static bool UnsafeQueueUserWorkItem(WaitCallback callBack, object? state)
1155 if (callBack == null)
1157 ThrowHelper.ThrowArgumentNullException(ExceptionArgument.callBack);
1160 EnsureInitialized();
1162 object tpcallBack = new QueueUserWorkItemCallbackDefaultContext(callBack!, state);
1164 ThreadPoolGlobals.workQueue.Enqueue(tpcallBack, forceGlobal: true);
1169 public static bool UnsafeQueueUserWorkItem(IThreadPoolWorkItem callBack, bool preferLocal)
1171 if (callBack == null)
1173 ThrowHelper.ThrowArgumentNullException(ExceptionArgument.callBack);
1175 if (callBack is Task)
1177 // Prevent code from queueing a derived Task that also implements the interface,
1178 // as that would bypass Task.Start and its safety checks.
1179 ThrowHelper.ThrowArgumentOutOfRangeException(ExceptionArgument.callBack);
1182 UnsafeQueueUserWorkItemInternal(callBack!, preferLocal);
1186 internal static void UnsafeQueueUserWorkItemInternal(object callBack, bool preferLocal)
1188 Debug.Assert((callBack is IThreadPoolWorkItem) ^ (callBack is Task));
1190 EnsureInitialized();
1192 ThreadPoolGlobals.workQueue.Enqueue(callBack, forceGlobal: !preferLocal);
1195 // This method tries to take the target callback out of the current thread's queue.
1196 internal static bool TryPopCustomWorkItem(object workItem)
1198 Debug.Assert(null != workItem);
1200 ThreadPoolGlobals.threadPoolInitialized && // if not initialized, so there's no way this workitem was ever queued.
1201 ThreadPoolGlobals.workQueue.LocalFindAndPop(workItem);
1204 // Get all workitems. Called by TaskScheduler in its debugger hooks.
1205 internal static IEnumerable<object> GetQueuedWorkItems()
1207 // Enumerate global queue
1208 foreach (object workItem in ThreadPoolGlobals.workQueue.workItems)
1210 yield return workItem;
1213 // Enumerate each local queue
1214 foreach (ThreadPoolWorkQueue.WorkStealingQueue wsq in ThreadPoolWorkQueue.WorkStealingQueueList.Queues)
1216 if (wsq != null && wsq.m_array != null)
1218 object?[] items = wsq.m_array;
1219 for (int i = 0; i < items.Length; i++)
1221 object? item = items[i];
1231 internal static IEnumerable<object> GetLocallyQueuedWorkItems()
1233 ThreadPoolWorkQueue.WorkStealingQueue wsq = ThreadPoolWorkQueueThreadLocals.threadLocals.workStealingQueue;
1234 if (wsq != null && wsq.m_array != null)
1236 object?[] items = wsq.m_array;
1237 for (int i = 0; i < items.Length; i++)
1239 object? item = items[i];
1246 internal static IEnumerable<object> GetGloballyQueuedWorkItems() => ThreadPoolGlobals.workQueue.workItems;
1248 private static object[] ToObjectArray(IEnumerable<object> workitems)
1251 foreach (object item in workitems)
1256 object[] result = new object[i];
1258 foreach (object item in workitems)
1260 if (i < result.Length) //just in case someone calls us while the queues are in motion
1268 // This is the method the debugger will actually call, if it ends up calling
1269 // into ThreadPool directly. Tests can use this to simulate a debugger, as well.
1270 internal static object[] GetQueuedWorkItemsForDebugger() =>
1271 ToObjectArray(GetQueuedWorkItems());
1273 internal static object[] GetGloballyQueuedWorkItemsForDebugger() =>
1274 ToObjectArray(GetGloballyQueuedWorkItems());
1276 internal static object[] GetLocallyQueuedWorkItemsForDebugger() =>
1277 ToObjectArray(GetLocallyQueuedWorkItems());
1280 /// Gets the number of work items that are currently queued to be processed.
1283 /// For a thread pool implementation that may have different types of work items, the count includes all types that can
1284 /// be tracked, which may only be the user work items including tasks. Some implementations may also include queued
1285 /// timer and wait callbacks in the count. On Windows, the count is unlikely to include the number of pending IO
1286 /// completions, as they get posted directly to an IO completion port.
1288 public static long PendingWorkItemCount
1292 ThreadPoolWorkQueue workQueue = ThreadPoolGlobals.workQueue;
1293 return workQueue.LocalCount + workQueue.GlobalCount + PendingUnmanagedWorkItemCount;