Associating threadpool queues with CPU cores. #18403
Conversation
VSadov
added
the
* NO MERGE *
VSadov
force-pushed
the
queue
branch
2 times, most recently
from
0dc3949
to
7de443a
Compare
|
@VSadov, what is the expected action item at this point with this PR? |
|
@stephentoub - nothing yet. I had an idea and tried to do some experiments. The idea is that queue-per-core approach (vs. current queue-per-thread) could result in fewer cases of cross-core task stealing. Fewer queues would imply that we would not steal as much. However the change would have to permit occasionally concurrent pushes/pops of the tasks and it is a bit delicate to do that both correctly and efficiently. |
VSadov
changed the title
VSadov
force-pushed
the
queue
branch
8 times, most recently
from
e9f8baf
to
392a802
Compare
VSadov
force-pushed
the
queue
branch
9 times, most recently
from
94ef9cf
to
1dec7bf
Compare
|
On a small machine we are same or faster. Sometimes significantly faster. == On 6-core HT (12 logical) Coffee Lake New TP Old TP (built locally from master) |
|
Even better story on a manycore machine - == On 24-core HT (48 logical) Ivy Bridge New TP Old TP (built locally from master) |
|
These are all ideal cases - I am using @benaadams TP benchmark/stress app ( https://github.com/benaadams/ThreadPoolTaskTesting ) where tasks do not wait for anything, or do much work at all. Even GC is suspended while scenarios run. In reality tasks do more work than that and often perform some measure of waiting (intentional or not). TP compensates by adding a few more workers. |
|
There is also an issue with It looks like the test has stravation/deadlock issues due to semaphore waiting in tasks. I have seen the test failing on old TP as well. |
|
There appears to be an issue with threads trampling on one another due to queue sharing. Repro: using System;
using System.Diagnostics;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
static class TaskTreeOrderTest
{
private static readonly int ThreadCount = Environment.ProcessorCount * 2;
private const byte TreeDepth = 8;
private const byte ItemCountPerNode = 8;
private static void Main()
{
ThreadPool.SetMinThreads(ThreadCount, ThreadCount);
ThreadPool.SetMaxThreads(ThreadCount, ThreadCount);
var threadArrayCapacity = (ThreadCount + 1) * 16;
var taskLatencySquareSumsUs = new double[threadArrayCapacity];
const int LatencyBucketCount = 6;
var taskLatencyCountsBuckets = new int[LatencyBucketCount][];
var taskLatencySumsUsBuckets = new double[LatencyBucketCount][];
for (int b = 0; b < LatencyBucketCount; ++b)
{
taskLatencyCountsBuckets[b] = new int[threadArrayCapacity];
taskLatencySumsUsBuckets[b] = new double[threadArrayCapacity];
}
(byte, byte) TaskCommon(object data)
{
var startTicksShifted = Stopwatch.GetTimestamp() << 16;
var ulongData = (ulong)(long)data;
var scheduleTicksShifted = (long)(ulongData & ~(ulong)0xffff);
var latencyUs = scheduleTicksShifted == 0 ? 0 : TicksToUs((startTicksShifted - scheduleTicksShifted) >> 16);
var depth = (byte)((ushort)ulongData >> 8);
var itemCount = (byte)ulongData;
if (scheduleTicksShifted != 0)
{
var threadArrayIndex = t_threadArrayIndex;
if (threadArrayIndex == 0)
threadArrayIndex = InitializeThreadArrayIndex();
int bucketIndex;
if (latencyUs < 1)
bucketIndex = 0;
else if (latencyUs < 10)
bucketIndex = 1;
else if (latencyUs < 100)
bucketIndex = 2;
else if (latencyUs < 1000)
bucketIndex = 3;
else if (latencyUs < 10000)
bucketIndex = 4;
else
bucketIndex = 5;
++taskLatencyCountsBuckets[bucketIndex][threadArrayIndex];
taskLatencySumsUsBuckets[bucketIndex][threadArrayIndex] += latencyUs;
taskLatencySquareSumsUs[threadArrayIndex] += latencyUs <= 1 ? 1 : latencyUs * latencyUs;
}
return (depth, itemCount);
}
// Simulate parallel test execution under directory tree
Func<object, Task<bool>> searchStart = null;
searchStart = async data =>
{
(var depth, var itemCount) = TaskCommon(data);
Random rng = t_rng;
if (rng == null)
t_rng = rng = new Random();
if (depth == 0)
{
for (int i = 0; i < itemCount; ++i)
{
// Simulate rare test failure
if (rng.Next(1_000_000) == 0)
return false;
}
return true;
}
Task<bool> processRemainingItemsTask = null;
if (itemCount > 1)
processRemainingItemsTask = ScheduleTask(searchStart, depth, (byte)(itemCount - 1));
bool success = itemCount == 0 || await searchStart(PackParameters(0, (byte)(depth - 1), ItemCountPerNode));
if (processRemainingItemsTask != null && !await processRemainingItemsTask)
success = false;
return success;
};
var scheduleTaskWorkItem = new ScheduleTaskWorkItem(searchStart);
ScheduleGlobalWorkItem(scheduleTaskWorkItem);
var sw = new Stopwatch();
var sb = new StringBuilder();
var taskLatencyCountDiffs = new int[LatencyBucketCount];
var taskLatencySumDiffsUs = new double[LatencyBucketCount];
var taskLatencyBucketNames = new string[]
{
" < 1 us",
" < 10 us",
" < 100 us",
" < 1000 us",
" < 10000 us",
" >= 10000 us"
};
while (true)
{
double taskLatencySquareSumDiff = 0;
for (int b = 0; b < LatencyBucketCount; ++b)
{
taskLatencyCountDiffs[b] = 0;
taskLatencySumDiffsUs[b] = 0;
for (int i = 16; i < threadArrayCapacity; i += 16)
{
taskLatencyCountDiffs[b] -= taskLatencyCountsBuckets[b][i];
taskLatencySumDiffsUs[b] -= taskLatencySumsUsBuckets[b][i];
taskLatencySquareSumDiff -= taskLatencySquareSumsUs[i];
}
}
sw.Restart();
Thread.Sleep(500);
sw.Stop();
for (int b = 0; b < LatencyBucketCount; ++b)
{
for (int i = 16; i < threadArrayCapacity; i += 16)
{
taskLatencyCountDiffs[b] += taskLatencyCountsBuckets[b][i];
taskLatencySumDiffsUs[b] += taskLatencySumsUsBuckets[b][i];
taskLatencySquareSumDiff += taskLatencySquareSumsUs[i];
}
}
var pendingWorkItemCount = ThreadPool.PendingWorkItemCount;
double elapsedS = sw.Elapsed.TotalSeconds;
for (int b = 0; b < LatencyBucketCount; ++b)
{
sb.AppendLine(
$"{taskLatencyBucketNames[b]} | " +
$"Count/s: {taskLatencyCountDiffs[b] / elapsedS,15:0.000} | " +
$"Average latency (us): {taskLatencySumDiffsUs[b] / taskLatencyCountDiffs[b],15:0.000}");
}
sb.AppendLine($"Biased average latency (us) | {Math.Sqrt(taskLatencySquareSumDiff / taskLatencyCountDiffs.Sum()),15:0.000}");
sb.AppendLine($" Count/ms | {taskLatencyCountDiffs.Sum() / (elapsedS * 1000),15:0.000}");
sb.AppendLine($" Pending work item count | {pendingWorkItemCount,11}");
Console.WriteLine(sb.ToString());
sb.Clear();
}
}
private static int s_previousThreadArrayIndex;
[ThreadStatic]
private static int t_threadArrayIndex;
[ThreadStatic]
private static Random t_rng;
private static readonly double TicksPerUs = Stopwatch.Frequency / (double)1_000_000;
private static double TicksToUs(long ticks) => ticks / TicksPerUs;
[MethodImpl(MethodImplOptions.NoInlining)]
private static int InitializeThreadArrayIndex()
{
if (t_threadArrayIndex != 0)
throw new InvalidOperationException();
int i = Interlocked.Increment(ref s_previousThreadArrayIndex) * 16;
t_threadArrayIndex = i;
return i;
}
private static object PackParameters(long ticks, byte depth, byte itemCount) =>
(ticks << 16) | (ushort)((ushort)depth << 8) | itemCount;
private static Task<bool> ScheduleTask(Func<object, Task<bool>> func, byte depth, byte itemCount)
{
var t = new Task<Task<bool>>(func, PackParameters(Stopwatch.GetTimestamp(), depth, itemCount));
t.Start();
return t.Unwrap();
}
private static void ScheduleGlobalWorkItem(IThreadPoolWorkItem workItem) =>
ThreadPool.UnsafeQueueUserWorkItem(workItem, preferLocal: false);
private sealed class ScheduleTaskWorkItem : IThreadPoolWorkItem
{
private Func<object, Task<bool>> _taskStart;
public ScheduleTaskWorkItem(Func<object, Task<bool>> taskStart) => _taskStart = taskStart;
async void IThreadPoolWorkItem.Execute()
{
var startTicks = Stopwatch.GetTimestamp();
await ScheduleTask(_taskStart, TreeDepth, ItemCountPerNode);
var endTicks = Stopwatch.GetTimestamp();
Console.WriteLine(
$"-------------------------------- " +
$"{TicksToUs(endTicks - startTicks) / 1000,15:0.000 ms} " +
$"--------------------------------{Environment.NewLine}");
ScheduleGlobalWorkItem(this);
}
}
}I tried to sort of simulate parallel test execution with a depth-first search through simulated directories, or just a general depth-first search. Upon visiting a node in the tree (a directory), a task is scheduled to process the remaining children at that depth while the visitor processes one child in depth-first fashion. Each of those tasks would again schedule another task to process the remaining children while processing one child. Before: Highest latencies, pending work item count, and throughput are somewhat stable. After (without prototype fixes above): Probably due to With prototype fixes above: Pending work item count has stabilized somewhat, but still spikes frequently due to the issue (not always visible in the output). Highest latencies still worse than baseline. Throughput is lower still. I believe due to thread assignment to queues being balanced on proc ID changes, there is even more thread movement between queues and more cross-trampling of work streams. In the baseline one thread's perspective would be one task scheduled per depth while going down the tree on one path. When a leaf is reached, the thread would pop the most recently scheduled work item (at the lowest depth) and continue processing that top-level branch of the tree in depth-first order. Meanwhile another thread would steal the oldest item (a high-depth node) and mostly independently process another top-level branch, and so on for the remaining threads. With this change when there are more threads than the processor count, consider two threads T0 and T1 operating on one queue. Suppose T0 is going down the tree on one path and added a few work items one for each depth. T1 would initially steal an old item (a top-level high-depth node) and start processing that similarly to the baseline. However, while T0 is happily processing one top-level branch, T1 tramples on it by scheduling a bunch of work items for the other top-level branch. T0 gets detracted from its work stream and is forced to work on T1's payload. So the search is still mostly depth-first but at times it becomes a bit more breadth-first. This keeps happening from time to time, and every time it happens it causes work items that are already in their shared local queue to get backed up. I'm not sure exactly why it's slower, I had expected equivalent throughput, bad latency, and unexpected order of execution (which can show up in the UI if this was actual test execution). The issue also exists on Linux and numbers are somewhat similar. The issue mostly disappears with the prototype fixes when It's quite possible I did something wrong in the test as well, so please do double-check it, but I think the theory at least is valid. I'm not sure there is a good way to solve that problem with a shared queue model. It may help in this particular case to always steal items in LIFO order but that would not solve the problem and I don't think that's the right solution. As far as I can tell the best way to solve this problem is to not allow threads to trample on one another by scheduling and taking work items from the same queue such that their work stream occurs in natural order and is not detracted, which would imply a thread-local queue model. Perhaps a better model would be to use thread-local queues and to associate them with a proc ID (along with balancing thread assignments, checking proc ID frequently, etc., which I still feel is a bit unfortunate but may be necessary to solve some issues). Such that only the thread can ever add items to its own queue, but other threads associated with the same proc ID can take items from it in LIFO order before going to the global queue. Work items coming in from outside the thread pool could be spread out among processors in some uniform fashion, as inserting them at the opposite end of the queue should not trample on the thread's work stream. It feels to me like that could be implemented efficiently. |
|
You are describing a "replicating task" scenario. It generally benefits from LIFO order. How much is the benefit - depends on the actual pattern of replication. Some patterns benefit more, some less, but generally LIFO is helpful and that is why local queues are LIFO. With shared queues the LIFO property should hold statistically, but forcing a lot of context switches would weaken it. |
|
Re:
Ultimately, the starvation issue is because there is a blocking wait on a task in a task. It is generally a dangerous thing to do since you start depending on scheduling order. It is not hard to fix in a more general way though. (I had very good results here: #26037 and that is just one direction to consider). |
|
Actually, after thinking about this more, I think it is reasonable to search all local queues as long as task has not yet dequeued (which task can tell). With that condition it is not as "unbounded". We would have to wait until someone else dequeues it anyways, so we could as well keep scanning. I will take proposed In reply to: 536803228 [](ancestors = 536803228) |
Technically, work is being done in LIFO order, but in effect a single thread is ending up processing somewhat random nodes in the tree or subtrees, along with work item counts in local queues getting backed up, which is also not ideal. These issues are not so much caused by context switching but because of sharing queues between two threads. I get that the proposed model in the paper would get benefits from having a processor run a set of (related) work items as long as possible and in a reasonable order to get cache and other benefits, but it only works well when it can be guaranteed that there is only one thread at any given time that is processing work from and inserting work into a queue (except for cases where a work item stalls, but including when a stalled work item is unblocked, in that it cannot run concurrently with another thread that is processing the same queue with a work item in flight). This would be easier to do when there is full control over scheduling. The model is also hurt by processor IDs changing. I don't think the model works well when there are multiple threads operating on the same queue, or when thread-queue assignments can be frequently incorrect.
The perf cliff appears to be at 9-16 threads on my 4-core 8-thread machine. Not sure yet why throughput is worse after my fixes above, but I tried ensuring that a thread's current local queue does not change until the next work item and it didn't seem to help. But anyway, throughput isn't the only problem.
I know, but there are comments in the Side note, on my 4-core 8-thread machine it looks like proc ID assignments are unstable only between 5-8 concurrently running threads. Between 1-4 threads and at 9+ threads the IDs are reasonably stable at least for 40-50 ms on average. So maybe something to do with Windows and hyperthreading. |
|
There could also be processor cache issues due to the issue above. Although this test case doesn't explicitly cover that, when a small subtree of work items access similar memory locations, processing items in different or somewhat random orders can negatively affect the processor cache. |
|
I have pushed a change so that TryRemove scans all local queues as long as the task is still inlineable. I get roughly the same results as with master on the |
|
By now I think I've at least convinced myself that I'm not willing to go ahead with this PR (more specifically this particular design change proposal), even with fixes to the above issues, because I don't think this is the best design to tackle. The current design is not ideal and certainly can be improved with some of the ideas from this proposal and the referenced paper, and I think at least those options should be explored seriously before settling on a design choice. I think there is merit to some of the ideas here and I think they can be implemented in a way that fits better. Of course, this is only my opinion, others may disagree. |
|
First - I am very grateful for the validation that you have done. Thanks!!! Now - I disagree with your conclusion ... obviously :-) -- We make assumption about core jumping being infrequent in a lot of places - in GC, in Concurrent Collections, in Array Pool, in basically all the places where we associate data with cores or with threads. Thread-associated data could be more tricky in this respect. When a thread happens to change its core, in core-associated case we will switch to a different piece of data associated with that core. There would be a delay due to core ID caching, but we will notice the core change soon enough and switch. -- I welcome that! These are awesome scenarios and the pool must handle these reasonably well. However, I think we should not aggressively optimize for that. We should definitely address crazy anomalies - lock ups, crashes, excessive slowdowns..., but a gradual degrade in a compute scenario as more threads are introduced is probably ok. -- |
|
To reiterate the above:
This should be very rare. In particular TryPop and Enqueue should not experience mutual contentions at any meaningful rate. (That is why spinning there is trivial) |
|
I am not seeing the regression on x4 threads scenario. My fast machine: https://gist.github.com/VSadov/fec967fd44eaa21598debbac1a8f630c In either case
The tests are pretty noisy though. It looks like results depend a lot on what is going on on the machine, whether CPU is thermal throttled, etc. Took some time to get decent readings. |
|
There is a latency regression on x1 threads scenario though. Throughput is still higher with the new queue, but average latency is definitely worse. My fast machine: https://gist.github.com/VSadov/3961a0de1dcaf8dcf17b044b240409ae |
|
The results are puzzling in two ways:
In master we have a tapered shape with odds of higher latencies gradually decreasing: In the case of the new queue we have a "thick tail" and then sudden drop. |
|
In particular in x1 case associating queues with threads or cores should not differ much. Indeed, from the results, most tasks are dispatched in about the same time, but there is a fraction that seem to be "neglected" like if sometimes we have a queue that noone works with for about 1000us. I suspect what we are seeing is not the high rate of core changing, but the core sampling being too coarse. The duration of "neglect" is capped by the core ID caching delay, which appears to be way too conservative. |
|
Reducing sampling delay of core ID to 1/50 calls results in a dramatic latency improvement. https://gist.github.com/VSadov/7e5e165b3176f7e4981a54c2d91eb1ee So the culprit is the core ID caching being too conservative. |
| internal const int MaxSegmentLength = 1024 * 1024; | ||
|
|
||
| /// <summary> | ||
| /// Lock used to protect cross-segment operations"/> |
There was a problem hiding this comment.
The comment is also too vague. Thanks!
…ly associated with CPU cores.
The implementation avoids 1:1 strict mapping between work queues and working threads which is often a far greater number than the number of cores.
This implementation improves handling of cases where the number of workers needs to exceeds the number of cores and causes precipitous performance drops because:
- the cost of snooping/stealing does not grow with the number of workers.
- the local queues tend to be deeper (since there are fewer of them) and thus stealing is less frequent and less contentious with enqueuing/popping.
- reduced likelihood that the workitem will have to execute on a core different from where it was scheduled.
…ueue to ensure that continuously nonempty global queue does not delay retirement of old segments. This is not a very common scenario, generally happening only at start up, but mitigation is also simple.
Moving to a different core while dispatching is a rare occurence, but when it does happen we do not want to neglect our local queue for too long since that can result in a batch of high-latency outliers.
| return true; | ||
| } | ||
|
|
||
| for (int i = 0; i < _localQueues.Length; ++i) |
There was a problem hiding this comment.
Take _localQueues to local to avoid bounds check in loop? https://github.com/dotnet/coreclr/issues/27510
There was a problem hiding this comment.
We generally do not make this far. The task is either in the local queue or stolen and executed.
However, there is code in existence that relies on task inlining (or more specifically on task.Wait not blocking), so we check other queues if task did not run yet.
Storing queues in a temp is still better. Thanks!
|
Thank you for your contribution. As announced in dotnet/coreclr#27549 this repository will be moving to dotnet/runtime on November 13. If you would like to continue working on this PR after this date, the easiest way to move the change to dotnet/runtime is:
|
VSadov
added
the
post-consolidation
|
Thank you for your contribution. As announced in #27549 the dotnet/runtime repository will be used going forward for changes to this code base. Closing this PR as no more changes will be accepted into master for this repository. If you’d like to continue working on this change please move it to dotnet/runtime. |
New implementation of ThreadPool work queues as lock-free queues softly associated with CPU cores.
The implementation avoids 1:1 strict mapping between work queues and working threads which is often a far greater number than the number of cores.
This implementation improves handling of cases where the number of workers needs to exceeds the number of cores. It avoids precipitous performance drops because:
Fixes:#19088
Also may alleviate issues as described in http://labs.criteo.com/2018/10/net-threadpool-starvation-and-how-queuing-makes-it-worse/