Skip to content

Fix a race condition in the thread pool #68171

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kouvel merged 3 commits into from
Apr 19, 2022
Merged

Fix a race condition in the thread pool #68171

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
318f045
Fix a race condition in the thread pool
Apr 14, 2022
4387dcf
Add a test
Apr 19, 2022
8f72207
Reduce test iterations
Apr 19, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 25 additions & 37 deletions src/libraries/System.Private.CoreLib/src/System/Threading/ThreadPoolWorkQueue.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -476,7 +476,6 @@ public void Enqueue(object callback, bool forceGlobal)
if (!forceGlobal && (tl = ThreadPoolWorkQueueThreadLocals.threadLocals) != null)
{
tl.workStealingQueue.LocalPush(callback);
tl.workState |= ThreadPoolWorkQueueThreadLocals.WorkState.MayHaveLocalWorkItems;
}
else
{
Expand Down Expand Up @@ -510,30 +509,21 @@ internal static bool LocalFindAndPop(object callback)
public object? Dequeue(ThreadPoolWorkQueueThreadLocals tl, ref bool missedSteal)
{
// Check for local work items
object? workItem;
ThreadPoolWorkQueueThreadLocals.WorkState tlWorkState = tl.workState;
if ((tlWorkState & ThreadPoolWorkQueueThreadLocals.WorkState.MayHaveLocalWorkItems) != 0)
object? workItem = tl.workStealingQueue.LocalPop();
if (workItem != null)
{
workItem = tl.workStealingQueue.LocalPop();
if (workItem != null)
{
return workItem;
}

Debug.Assert(tlWorkState == tl.workState);
tl.workState = tlWorkState &= ~ThreadPoolWorkQueueThreadLocals.WorkState.MayHaveLocalWorkItems;
return workItem;
}

// Check for high-priority work items
if ((tlWorkState & ThreadPoolWorkQueueThreadLocals.WorkState.IsProcessingHighPriorityWorkItems) != 0)
if (tl.isProcessingHighPriorityWorkItems)
{
if (highPriorityWorkItems.TryDequeue(out workItem))
{
return workItem;
}

Debug.Assert(tlWorkState == tl.workState);
tl.workState = tlWorkState &= ~ThreadPoolWorkQueueThreadLocals.WorkState.IsProcessingHighPriorityWorkItems;
tl.isProcessingHighPriorityWorkItems = false;
}
else if (
_mayHaveHighPriorityWorkItems != 0 &&
Expand Down Expand Up @@ -579,14 +569,14 @@ private bool TryStartProcessingHighPriorityWorkItemsAndDequeue(
ThreadPoolWorkQueueThreadLocals tl,
[MaybeNullWhen(false)] out object workItem)
{
Debug.Assert((tl.workState & ThreadPoolWorkQueueThreadLocals.WorkState.IsProcessingHighPriorityWorkItems) == 0);
Debug.Assert(!tl.isProcessingHighPriorityWorkItems);

if (!highPriorityWorkItems.TryDequeue(out workItem))
{
return false;
}

tl.workState |= ThreadPoolWorkQueueThreadLocals.WorkState.IsProcessingHighPriorityWorkItems;
tl.isProcessingHighPriorityWorkItems = true;
_mayHaveHighPriorityWorkItems = 1;
return true;
}
Expand Down Expand Up @@ -632,8 +622,7 @@ internal static bool Dispatch()
// take over the thread, sustaining starvation. For example, when worker threads are continually starved,
// high-priority work items may always be queued and normal-priority work items may not get a chance to run.
bool dispatchNormalPriorityWorkFirst = workQueue._dispatchNormalPriorityWorkFirst;
if (dispatchNormalPriorityWorkFirst &&
(tl.workState & ThreadPoolWorkQueueThreadLocals.WorkState.MayHaveLocalWorkItems) == 0)
if (dispatchNormalPriorityWorkFirst && !tl.workStealingQueue.CanSteal)
{
workQueue._dispatchNormalPriorityWorkFirst = !dispatchNormalPriorityWorkFirst;
workQueue.workItems.TryDequeue(out workItem);
Expand Down Expand Up @@ -670,7 +659,7 @@ internal static bool Dispatch()
// reason that may have a dependency on other queued work items.
workQueue.EnsureThreadRequested();

// After this point, this method is no longer responsible for ensuring thread requests
// After this point, this method is no longer responsible for ensuring thread requests except for missed steals
}

// Has the desire for logging changed since the last time we entered?
Expand Down Expand Up @@ -700,8 +689,18 @@ internal static bool Dispatch()

if (workItem == null)
{
// May have missed a steal, but this method is not responsible for ensuring thread requests anymore. See
// the dequeue before the loop.
//
// No work.
// If we missed a steal, though, there may be more work in the queue.
// Instead of looping around and trying again, we'll just request another thread. Hopefully the thread
// that owns the contended work-stealing queue will pick up its own workitems in the meantime,
// which will be more efficient than this thread doing it anyway.
//
if (missedSteal)
{
workQueue.EnsureThreadRequested();
}

return true;
}
}
Expand Down Expand Up @@ -753,7 +752,7 @@ internal static bool Dispatch()
// to ensure that they would not be heavily delayed. Tell the caller that this thread was requested to stop
// processing work items.
tl.TransferLocalWork();
tl.ResetWorkItemProcessingState();
tl.isProcessingHighPriorityWorkItems = false;
return false;
}

Expand All @@ -769,7 +768,7 @@ 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.ResetWorkItemProcessingState();
tl.isProcessingHighPriorityWorkItems = false;
return true;
}

Expand Down Expand Up @@ -823,7 +822,7 @@ internal sealed class ThreadPoolWorkQueueThreadLocals
[ThreadStatic]
public static ThreadPoolWorkQueueThreadLocals? threadLocals;

public WorkState workState;
public bool isProcessingHighPriorityWorkItems;
public readonly ThreadPoolWorkQueue workQueue;
public readonly ThreadPoolWorkQueue.WorkStealingQueue workStealingQueue;
public readonly Thread currentThread;
Expand All @@ -839,16 +838,12 @@ public ThreadPoolWorkQueueThreadLocals(ThreadPoolWorkQueue tpq)
threadLocalCompletionCountObject = ThreadPool.GetOrCreateThreadLocalCompletionCountObject();
}

public void ResetWorkItemProcessingState() => workState &= ~WorkState.IsProcessingHighPriorityWorkItems;

public void TransferLocalWork()
{
while (workStealingQueue.LocalPop() is object cb)
{
workQueue.Enqueue(cb, forceGlobal: true);
}

workState &= ~WorkState.MayHaveLocalWorkItems;
}

~ThreadPoolWorkQueueThreadLocals()
Expand All @@ -860,13 +855,6 @@ public void TransferLocalWork()
ThreadPoolWorkQueue.WorkStealingQueueList.Remove(workStealingQueue);
}
}

[Flags]
public enum WorkState
{
MayHaveLocalWorkItems = 1 << 0,
IsProcessingHighPriorityWorkItems = 1 << 1
}
}

// A strongly typed callback for ThreadPoolTypedWorkItemQueue<T, TCallback>.
Expand Down Expand Up @@ -948,7 +936,7 @@ void IThreadPoolWorkItem.Execute()
// yield to the thread pool after some time. The threshold used is half of the thread pool's dispatch quantum,
// which the thread pool uses for doing periodic work.
if (++completedCount == uint.MaxValue ||
(tl.workState & ThreadPoolWorkQueueThreadLocals.WorkState.MayHaveLocalWorkItems) != 0 ||
tl.workStealingQueue.CanSteal ||
(uint)(Environment.TickCount - startTimeMs) >= ThreadPoolWorkQueue.DispatchQuantumMs / 2 ||
!_workItems.TryDequeue(out workItem))
{
Expand Down
60 changes: 60 additions & 0 deletions src/libraries/System.Threading.ThreadPool/tests/ThreadPoolTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Reflection;
using System.Threading.Tasks;
Expand Down Expand Up @@ -1020,6 +1021,65 @@ public static void CooperativeBlockingWithProcessingThreadsAndGoalThreadsAndAddW
}).Dispose();
}

[ConditionalFact(nameof(IsThreadingAndRemoteExecutorSupported))]
public void FileStreamFlushAsyncThreadPoolDeadlockTest()
{
// This test was occasionally causing the deadlock described in https://github.com/dotnet/runtime/pull/68171. Run it
// in a remote process to test it with a dedicated thread pool.
RemoteExecutor.Invoke(async () =>
{
const int OneKibibyte = 1 << 10;
const int FourKibibytes = OneKibibyte << 2;
const int FileSize = 1024;

string destinationFilePath = null;
try
{
destinationFilePath = CreateFileWithRandomContent(FileSize);

static string CreateFileWithRandomContent(int fileSize)
{
string filePath = Path.GetTempFileName();

@danmoseley danmoseley Apr 19, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

aside, better to use the TempFile class, which allows the using pattern instead of writing out try/finally. It also avoids the problematic Path.GetTempFileName() which fails if someone else has leaked temp files.
It has a ctor that takes an array, too.

runtime/src/libraries/Common/tests/System/IO/TempFile.cs

Line 22 in 19eb98d

public TempFile(string path, byte[] data)

@kouvel kouvel Apr 19, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh ok, I'll fix it up in a follow-up PR

File.WriteAllBytes(filePath, CreateArray(fileSize));
return filePath;
}

static byte[] CreateArray(int count)
{
var result = new byte[count];
const int Seed = 12345;
var random = new Random(Seed);
random.NextBytes(result);
return result;
}

for (int j = 0; j < 100; j++)
{
using var fileStream =
new FileStream(
destinationFilePath,
FileMode.Create,
FileAccess.Write,
FileShare.Read,
FourKibibytes,
FileOptions.None);
for (int i = 0; i < FileSize; i++)
{
fileStream.WriteByte(default);
await fileStream.FlushAsync();
}
}
}
finally
{
if (!string.IsNullOrEmpty(destinationFilePath) && File.Exists(destinationFilePath))
{
File.Delete(destinationFilePath);
}
}
}).Dispose();
}

public static bool IsThreadingAndRemoteExecutorSupported =>
PlatformDetection.IsThreadingSupported && RemoteExecutor.IsSupported;
}
Expand Down