Skip to content

Commit d955059

Browse files
Transfer ThreadPool local queue to high-pri queue on Task blocking (#109989)
Added for now under the same config flag as is being used for other work prioritization experimentation. Co-authored-by: Stephen Toub <[email protected]>
1 parent 2c860a7 commit d955059

File tree

3 files changed

+65
-1
lines changed

3 files changed

+65
-1
lines changed

src/libraries/System.Private.CoreLib/src/System/Threading/Tasks/Task.cs

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3056,6 +3056,27 @@ private bool SpinThenBlockingWait(int millisecondsTimeout, CancellationToken can
30563056
bool returnValue = SpinWait(millisecondsTimeout);
30573057
if (!returnValue)
30583058
{
3059+
#if CORECLR
3060+
if (ThreadPoolWorkQueue.s_prioritizationExperiment)
3061+
{
3062+
// We're about to block waiting for the task to complete, which is expensive, and if
3063+
// the task being waited on depends on some other work to run, this thread could end up
3064+
// waiting for some other thread to do work. If the two threads are part of the same scheduler,
3065+
// such as the thread pool, that could lead to a (temporary) deadlock. This is made worse by
3066+
// it also leading to a possible priority inversion on previously queued work. Each thread in
3067+
// the thread pool has a local queue. A key motivator for this local queue is it allows this
3068+
// thread to create work items that it will then prioritize above all other work in the
3069+
// pool. However, while this thread makes its own local queue the top priority, that queue is
3070+
// every other thread's lowest priority. If this thread blocks, all of its created work that's
3071+
// supposed to be high priority becomes low priority, and work that's typically part of a
3072+
// currently in-flight operation gets deprioritized relative to new requests coming into the
3073+
// pool, which can lead to the whole system slowing down or even deadlocking. To address that,
3074+
// just before we block, we move all local work into a global queue, so that it's at least
3075+
// prioritized by other threads more fairly with respect to other work.
3076+
ThreadPoolWorkQueue.TransferAllLocalWorkItemsToHighPriorityGlobalQueue();
3077+
}
3078+
#endif
3079+
30593080
var mres = new SetOnInvokeMres();
30603081
try
30613082
{

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

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -701,6 +701,27 @@ public void EnqueueAtHighPriority(object workItem)
701701
EnsureThreadRequested();
702702
}
703703

704+
internal static void TransferAllLocalWorkItemsToHighPriorityGlobalQueue()
705+
{
706+
// If there's no local queue, there's nothing to transfer.
707+
if (ThreadPoolWorkQueueThreadLocals.threadLocals is not ThreadPoolWorkQueueThreadLocals tl)
708+
{
709+
return;
710+
}
711+
712+
// Pop each work item off the local queue and push it onto the global. This is a
713+
// bounded loop as no other thread is allowed to push into this thread's queue.
714+
ThreadPoolWorkQueue queue = ThreadPool.s_workQueue;
715+
while (tl.workStealingQueue.LocalPop() is object workItem)
716+
{
717+
queue.highPriorityWorkItems.Enqueue(workItem);
718+
}
719+
720+
Volatile.Write(ref queue._mayHaveHighPriorityWorkItems, true);
721+
722+
queue.EnsureThreadRequested();
723+
}
724+
704725
internal static bool LocalFindAndPop(object callback)
705726
{
706727
ThreadPoolWorkQueueThreadLocals? tl = ThreadPoolWorkQueueThreadLocals.threadLocals;

src/libraries/System.Threading.ThreadPool/tests/ThreadPoolTests.cs

Lines changed: 23 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1263,12 +1263,13 @@ public static void PrioritizationExperimentConfigVarTest()
12631263
RemoteExecutor.Invoke(() =>
12641264
{
12651265
const int WorkItemCountPerKind = 100;
1266+
const int Kinds = 3;
12661267

12671268
int completedWorkItemCount = 0;
12681269
var allWorkItemsCompleted = new AutoResetEvent(false);
12691270
Action<int> workItem = _ =>
12701271
{
1271-
if (Interlocked.Increment(ref completedWorkItemCount) == WorkItemCountPerKind * 3)
1272+
if (Interlocked.Increment(ref completedWorkItemCount) == WorkItemCountPerKind * Kinds)
12721273
{
12731274
allWorkItemsCompleted.Set();
12741275
}
@@ -1305,6 +1306,27 @@ public static void PrioritizationExperimentConfigVarTest()
13051306
0,
13061307
preferLocal: false);
13071308

1309+
ThreadPool.UnsafeQueueUserWorkItem(
1310+
_ =>
1311+
{
1312+
// Enqueue tasks from a thread pool thread into the local queue,
1313+
// then block this thread until a queued task completes.
1314+
1315+
startTest.CheckedWait();
1316+
1317+
Task queued = null;
1318+
for (int i = 0; i < WorkItemCountPerKind; i++)
1319+
{
1320+
queued = Task.Run(() => workItem(0));
1321+
}
1322+
1323+
queued
1324+
.ContinueWith(_ => { }) // prevent wait inlining
1325+
.Wait();
1326+
},
1327+
0,
1328+
preferLocal: false);
1329+
13081330
t = new Thread(() =>
13091331
{
13101332
// Enqueue local work from thread pool worker threads

0 commit comments

Comments
 (0)