自已建网站卖东西要多少钱,手机用什么软件做网站,宜昌广告制作公司,wordpress博客修改前言在即将发布的 .NET 6 runtime 中#xff0c;默认的线程池实现从 C 代码改为了 C##xff0c;更方便我们学习线程池的设计了。https://github.com/dotnet/runtime/tree/release/6.0/src/libraries/System.Threading.ThreadPool新的线程池实现位于 PortableThreadPool 中默认的线程池实现从 C 代码改为了 C#更方便我们学习线程池的设计了。https://github.com/dotnet/runtime/tree/release/6.0/src/libraries/System.Threading.ThreadPool新的线程池实现位于 PortableThreadPool 中原 ThreadPool 中的对外公开的接口会直接调用 PortableThreadPool 中的实现。通过设置环境变量 ThreadPool_UsePortableThreadPool 为 0 可以设置成使用老的线程池实现。https://github.com/dotnet/runtime/pull/43841/commits/b0d47b84a6845a70f011d1b0d3ce5adde9a4d7b7本文以 .NET 6 runtime 源码作为学习材料对线程池的设计进行介绍。从目前的理解上来看其整体的设计与原来 C 的实现并没有特别大的出入。注意本文不涉及细节的代码实现主要为大家介绍其整体设计。所展示的代码并非原封不动的源码而是为了方便理解的简化版。ThreadPool.SetMaxThreads(int workerThreads, int completionPortThreads) 中的 completionPortThreads 所相关的 IOCP线程池 是 .NET Framework 时代的遗留产物用于管理 Windows 平台专有的 IOCP 的回调线程池。目前没看到有什么地方在用它了completionPortThreads 这个参数也已经没有意义底层IO库是自己维护的IO等待线程池。本文只涉及 worker thread 池的介绍。本文理解并不完整也不一定完全正确有异议的地方欢迎留言讨论。为了解释问题一部分代码会运行在 .NET 6 之前的环境中。任务的调度线程池的待执行任务被存放在一个队列系统中。这个系统包括一个 全局队列以及绑定在每一个 Worker Thread 上 的 本地队列 。而线程池中的每一个线程都在执行 while(true) 的循环从这个队列系统中领取并执行任务。在 ThreadPool.QueueUserWorkItem 的重载方法 ThreadPool.QueueUserWorkItemTState(ActionTState callBack, TState state, bool preferLocal) 里有一个 preferLocal 参数。调用不带 preferLocal 参数的 ThreadPool.QueueUserWorkItem 方法重载任务会被放到全局队列。当 preferLocal 为 true 的时候如果调用 ThreadPool.QueueUserWorkItem 代码的线程正好是个线程池里的某个线程则该任务就会进入该线程的本地队列中。除此之外的情况则会被放到全局队列中等待未来被某个 Worker Thread 捡走。在线程池外的线程中调用不管 preferLocal 传的是什么任务都会被放到全局队列。基本调度单元本地队列和全局队列的元素类型被定义为 object实际的任务类型分为两类在从队列系统取到任务之后会判断类型并执行对应的方法。IThreadPoolWorkItem 实现类的实例。/// summaryRepresents a work item that can be executed by the ThreadPool./summary
public interface IThreadPoolWorkItem
{void Execute();
}执行 Execute 方法也就代表着任务的执行。IThreadPoolWorkItem 的具体实现有很多例如通过 ThreadPool.QueueUserWorkItem(WaitCallback callBack) 传入的 callBack 委托实例会被包装到一个 QueueUserWorkItemCallback 实例里。QueueUserWorkItemCallback 是 IThreadPoolWorkItem 的实现类。Taskclass Task
{internal void InnerInvoke();
}执行 InnerInvoke 会执行 Task 所包含的委托。全局队列全局队列 是由 ThreadPoolWorkQueue 维护的同时它也是整个队列系统的入口直接被 ThreadPool 所引用。public static class ThreadPool
{internal static readonly ThreadPoolWorkQueue s_workQueue new ThreadPoolWorkQueue();public static bool QueueUserWorkItem(WaitCallback callBack, object state){object tpcallBack new QueueUserWorkItemCallback(callBack!, state);s_workQueue.Enqueue(tpcallBack, forceGlobal: true);return true;}
}internal sealed class ThreadPoolWorkQueue
{// 全局队列internal readonly ConcurrentQueueobject workItems new ConcurrentQueueobject();// forceGlobal 为 true 时push 到全局队列否则就放到本地队列public void Enqueue(object callback, bool forceGlobal);
}本地队列线程池中的每一个线程都会绑定一个 ThreadPoolWorkQueueThreadLocals 实例在 workStealingQueue 这个字段上保存着本地队列。internal sealed class ThreadPoolWorkQueueThreadLocals
{// 绑定在线程池线程上[ThreadStatic]public static ThreadPoolWorkQueueThreadLocals threadLocals;// 持有全局队列的引用以便能在需要的时候将任务转移到全局队列上public readonly ThreadPoolWorkQueue workQueue;// 本地队列的直接维护者public readonly ThreadPoolWorkQueue.WorkStealingQueue workStealingQueue;public readonly Thread currentThread;public ThreadPoolWorkQueueThreadLocals(ThreadPoolWorkQueue tpq){workQueue tpq;workStealingQueue new ThreadPoolWorkQueue.WorkStealingQueue();// WorkStealingQueueList 会集中管理 workStealingQueueThreadPoolWorkQueue.WorkStealingQueueList.Add(workStealingQueue);currentThread Thread.CurrentThread;}// 提供将本地队列中的任务转移到全局队列中去的功能// 当 ThreadPool 通过后文将会介绍的 HillClimbing 算法判断得出当前线程是多余的线程后// 会调用此方法对任务进行转移public void TransferLocalWork(){while (workStealingQueue.LocalPop() is object cb){workQueue.Enqueue(cb, forceGlobal: true);}}~ThreadPoolWorkQueueThreadLocals(){if (null ! workStealingQueue){// TransferLocalWork 真正的目的并非是为了在这里被调用这边只是确保任务不会丢的 fallback 逻辑TransferLocalWork();ThreadPoolWorkQueue.WorkStealingQueueList.Remove(workStealingQueue);}}
}偷窃机制这里思考一个问题为什么本地队列的名字会被叫做 WorkStealingQueue 呢所有 Worker Thread 的 WorkStealingQueue 都被集中在 WorkStealingQueueList 中。对线程池中其他所有线程可见。Worker Thread 的 while(true) 中优先会从自身的 WorkStealingQueue 中取任务。如果本地队列已经被清空就会从全局队列中取任务。例如下图的 Thread1 取全局队列中领取了一个任务。同时 Thread3 也没活干了但是全局队列中的任务被 Thread1 抢走了。这时候就会去 从 Thread2 的本地队列中抢 Thread2 的活。Worker Thread 的生命周期管理接下来我们把格局放大关注点从 Worker Thread 的打工日常转移到对它们的生命周期管理上来。为了更方便的解释线程管理的机制这边使用下面使用一些代码做演示。代码参考自 https://devblogs.microsoft.com/dotnet/performance-improvements-in-net-6/#threading。线程生命注入实验Task.Run 会将 Task 调度到线程池中执行下面的示例代码中等效于 ThreadPool.QueueUserWorkItem(WaitCallback callBack)会把 Task 放到队列系统的全局队列中顺便一提如果在一个线程池线程中执行 Task.Run 会将 Task 调度到此线程池线程的本地队列中。.NET 5 实验一 默认线程池配置static void Main(string[] args)
{var sw Stopwatch.StartNew();var tcs new TaskCompletionSource();var tasks new ListTask();for (int i 1; i Environment.ProcessorCount * 2; i){int id i;Console.WriteLine($Loop Id: {id:00} | {sw.Elapsed.TotalSeconds:0.000} | Busy Threads: {GetBusyThreads()});tasks.Add(Task.Run(() {Console.WriteLine($Task Id: {id:00} | {sw.Elapsed.TotalSeconds:0.000} | Busy Threads: {GetBusyThreads()});tcs.Task.Wait();}));}tasks.Add(Task.Run(() {Console.WriteLine($Task SetResult | {sw.Elapsed.TotalSeconds:0.000} | Busy Threads: {GetBusyThreads()});tcs.SetResult();}));Task.WaitAll(tasks.ToArray());Console.WriteLine($Done: | {sw.Elapsed.TotalSeconds:0.000});
}static int GetBusyThreads()
{ThreadPool.GetAvailableThreads(out var available, out _);ThreadPool.GetMaxThreads(out var max, out _);return max - available;
}首先在代码在 .NET 5 环境中运行以下代码CPU 逻辑核心数 12。Loop Id: 01 | 0.000 | Busy Threads: 0
Loop Id: 02 | 0.112 | Busy Threads: 1
Loop Id: 03 | 0.112 | Busy Threads: 2
Loop Id: 04 | 0.113 | Busy Threads: 4
Loop Id: 05 | 0.113 | Busy Threads: 7
Loop Id: 06 | 0.113 | Busy Threads: 10
Loop Id: 07 | 0.113 | Busy Threads: 10
Task Id: 01 | 0.113 | Busy Threads: 11
Task Id: 02 | 0.113 | Busy Threads: 12
Task Id: 03 | 0.113 | Busy Threads: 12
Task Id: 07 | 0.113 | Busy Threads: 12
Task Id: 04 | 0.113 | Busy Threads: 12
Task Id: 05 | 0.113 | Busy Threads: 12
Loop Id: 08 | 0.113 | Busy Threads: 10
Task Id: 08 | 0.113 | Busy Threads: 12
Loop Id: 09 | 0.113 | Busy Threads: 11
Loop Id: 10 | 0.113 | Busy Threads: 12
Loop Id: 11 | 0.114 | Busy Threads: 12
Loop Id: 12 | 0.114 | Busy Threads: 12
Loop Id: 13 | 0.114 | Busy Threads: 12
Loop Id: 14 | 0.114 | Busy Threads: 12
Loop Id: 15 | 0.114 | Busy Threads: 12
Loop Id: 16 | 0.114 | Busy Threads: 12
Loop Id: 17 | 0.114 | Busy Threads: 12
Loop Id: 18 | 0.114 | Busy Threads: 12
Loop Id: 19 | 0.114 | Busy Threads: 12
Loop Id: 20 | 0.114 | Busy Threads: 12
Loop Id: 21 | 0.114 | Busy Threads: 12
Loop Id: 22 | 0.114 | Busy Threads: 12
Loop Id: 23 | 0.114 | Busy Threads: 12
Loop Id: 24 | 0.114 | Busy Threads: 12
Task Id: 09 | 0.114 | Busy Threads: 12
Task Id: 06 | 0.114 | Busy Threads: 12
Task Id: 10 | 0.114 | Busy Threads: 12
Task Id: 11 | 0.114 | Busy Threads: 12
Task Id: 12 | 0.114 | Busy Threads: 12
Task Id: 13 | 1.091 | Busy Threads: 13
Task Id: 14 | 1.594 | Busy Threads: 14
Task Id: 15 | 2.099 | Busy Threads: 15
Task Id: 16 | 3.102 | Busy Threads: 16
Task Id: 17 | 3.603 | Busy Threads: 17
Task Id: 18 | 4.107 | Busy Threads: 18
Task Id: 19 | 4.611 | Busy Threads: 19
Task Id: 20 | 5.113 | Busy Threads: 20
Task Id: 21 | 5.617 | Busy Threads: 21
Task Id: 22 | 6.122 | Busy Threads: 22
Task Id: 23 | 7.128 | Busy Threads: 23
Task Id: 24 | 7.632 | Busy Threads: 24
Task SetResult | 8.135 | Busy Threads: 25
Done: | 8.136Task.Run 会把 Task 调度到线程池上执行前 24 个 task 都会被阻塞住直到第 25 个被执行。每次都会打印出当前线程池中正在执行任务的线程数也就是创建完成的线程数。可以观察到以下结果前几次循环线程随着 Task 数量递增后面几次循环直到循环结束为止线程数一直维持在 12 没有发生变化。线程数在达到 12 之前零间隔时间增加。第 12 到 第 13 线程间隔 1s 不到往后约 500ms 增加一个线程。.NET 5 实验二 调整 ThreadPool 设置在上面的代码最前面加入以下两行代码继续在 .NET 5 环境运行一次。ThreadPool.GetMinThreads(out int defaultMinThreads, out int completionPortThreads);
Console.WriteLine($DefaultMinThreads: {defaultMinThreads});
ThreadPool.SetMinThreads(14, completionPortThreads);运行结果如下DefaultMinThreads: 12
Loop Id: 01 | 0.000 | Busy Threads: 0
Loop Id: 02 | 0.003 | Busy Threads: 1
Loop Id: 03 | 0.003 | Busy Threads: 2
Loop Id: 04 | 0.003 | Busy Threads: 5
Loop Id: 05 | 0.004 | Busy Threads: 8
Task Id: 01 | 0.004 | Busy Threads: 10
Task Id: 03 | 0.004 | Busy Threads: 10
Loop Id: 06 | 0.004 | Busy Threads: 10
Task Id: 02 | 0.004 | Busy Threads: 10
Task Id: 04 | 0.004 | Busy Threads: 10
Task Id: 05 | 0.004 | Busy Threads: 12
Loop Id: 07 | 0.004 | Busy Threads: 9
Loop Id: 08 | 0.004 | Busy Threads: 10
Loop Id: 09 | 0.004 | Busy Threads: 11
Loop Id: 10 | 0.004 | Busy Threads: 12
Task Id: 08 | 0.004 | Busy Threads: 14
Task Id: 06 | 0.004 | Busy Threads: 14
Task Id: 09 | 0.004 | Busy Threads: 14
Task Id: 10 | 0.004 | Busy Threads: 14
Loop Id: 11 | 0.004 | Busy Threads: 14
Loop Id: 12 | 0.004 | Busy Threads: 14
Loop Id: 13 | 0.004 | Busy Threads: 14
Loop Id: 14 | 0.004 | Busy Threads: 14
Loop Id: 15 | 0.004 | Busy Threads: 14
Loop Id: 16 | 0.004 | Busy Threads: 14
Loop Id: 17 | 0.004 | Busy Threads: 14
Loop Id: 18 | 0.004 | Busy Threads: 14
Loop Id: 19 | 0.004 | Busy Threads: 14
Loop Id: 20 | 0.004 | Busy Threads: 14
Loop Id: 21 | 0.004 | Busy Threads: 14
Loop Id: 22 | 0.004 | Busy Threads: 14
Task Id: 11 | 0.004 | Busy Threads: 14
Loop Id: 23 | 0.004 | Busy Threads: 14
Loop Id: 24 | 0.005 | Busy Threads: 14
Task Id: 07 | 0.005 | Busy Threads: 14
Task Id: 12 | 0.005 | Busy Threads: 14
Task Id: 13 | 0.005 | Busy Threads: 14
Task Id: 14 | 0.005 | Busy Threads: 14
Task Id: 15 | 0.982 | Busy Threads: 15
Task Id: 16 | 1.486 | Busy Threads: 16
Task Id: 17 | 1.991 | Busy Threads: 17
Task Id: 18 | 2.997 | Busy Threads: 18
Task Id: 19 | 3.501 | Busy Threads: 19
Task Id: 20 | 4.004 | Busy Threads: 20
Task Id: 21 | 4.509 | Busy Threads: 21
Task Id: 22 | 5.014 | Busy Threads: 22
Task Id: 23 | 5.517 | Busy Threads: 23
Task Id: 24 | 6.021 | Busy Threads: 24
Task SetResult | 6.522 | Busy Threads: 25
Done: | 6.523在调整完线程池的最小线程数量之后线程注入速度发生转折的时间点从第 12默认min threads 个线程换到了第 14修改后的min threads个线程。整体时间也从 8s 缩到 6s。.NET 5 实验三 tcs.Task.Wait() 改为 Thread.Sleepstatic void Main(string[] args)
{var sw Stopwatch.StartNew();var tasks new ListTask();for (int i 1; i Environment.ProcessorCount * 2; i){int id i;Console.WriteLine($Loop Id: {id:00} | {sw.Elapsed.TotalSeconds:0.000} | Busy Threads: {GetBusyThreads()});tasks.Add(Task.Run(() {Console.WriteLine($Task Id: {id:00} | {sw.Elapsed.TotalSeconds:0.000} | Busy Threads: {GetBusyThreads()});Thread.Sleep(Environment.ProcessorCount * 1000);}));}Task.WhenAll(tasks.ToArray()).ContinueWith(_ {Console.WriteLine($Done: | {sw.Elapsed.TotalSeconds:0.000});});Console.ReadLine();
}Loop Id: 01 | 0.000 | Busy Threads: 0
Loop Id: 02 | 0.027 | Busy Threads: 1
Loop Id: 03 | 0.027 | Busy Threads: 2
Loop Id: 04 | 0.027 | Busy Threads: 3
Loop Id: 05 | 0.028 | Busy Threads: 4
Loop Id: 06 | 0.028 | Busy Threads: 10
Loop Id: 07 | 0.028 | Busy Threads: 9
Loop Id: 08 | 0.028 | Busy Threads: 9
Loop Id: 09 | 0.028 | Busy Threads: 10
Loop Id: 10 | 0.028 | Busy Threads: 12
Loop Id: 11 | 0.028 | Busy Threads: 12
Loop Id: 12 | 0.028 | Busy Threads: 12
Loop Id: 13 | 0.028 | Busy Threads: 12
Loop Id: 14 | 0.028 | Busy Threads: 12
Loop Id: 15 | 0.028 | Busy Threads: 12
Loop Id: 16 | 0.028 | Busy Threads: 12
Loop Id: 17 | 0.028 | Busy Threads: 12
Loop Id: 18 | 0.028 | Busy Threads: 12
Loop Id: 19 | 0.028 | Busy Threads: 12
Loop Id: 20 | 0.028 | Busy Threads: 12
Loop Id: 21 | 0.028 | Busy Threads: 12
Loop Id: 22 | 0.028 | Busy Threads: 12
Loop Id: 23 | 0.028 | Busy Threads: 12
Loop Id: 24 | 0.028 | Busy Threads: 12
Task Id: 01 | 0.029 | Busy Threads: 12
Task Id: 05 | 0.029 | Busy Threads: 12
Task Id: 03 | 0.029 | Busy Threads: 12
Task Id: 08 | 0.029 | Busy Threads: 12
Task Id: 09 | 0.029 | Busy Threads: 12
Task Id: 10 | 0.029 | Busy Threads: 12
Task Id: 06 | 0.029 | Busy Threads: 12
Task Id: 11 | 0.029 | Busy Threads: 12
Task Id: 12 | 0.029 | Busy Threads: 12
Task Id: 04 | 0.029 | Busy Threads: 12
Task Id: 02 | 0.029 | Busy Threads: 12
Task Id: 07 | 0.029 | Busy Threads: 12
Task Id: 13 | 1.018 | Busy Threads: 13
Task Id: 14 | 1.522 | Busy Threads: 14
Task Id: 15 | 2.025 | Busy Threads: 15
Task Id: 16 | 2.530 | Busy Threads: 16
Task Id: 17 | 3.530 | Busy Threads: 17
Task Id: 18 | 4.035 | Busy Threads: 18
Task Id: 19 | 4.537 | Busy Threads: 19
Task Id: 20 | 5.040 | Busy Threads: 20
Task Id: 21 | 5.545 | Busy Threads: 21
Task Id: 22 | 6.048 | Busy Threads: 22
Task Id: 23 | 7.049 | Busy Threads: 23
Task Id: 24 | 8.056 | Busy Threads: 24
Done: | 20.060达到 min threads 默认12之后线程注入速度明显变慢最快间隔 500ms。.NET 6 实验一 默认 ThreadPool 设置将 .NET 5 实验一的代码在 .NET 6 执行一次Loop Id: 01 | 0.001 | Busy Threads: 0
Loop Id: 02 | 0.018 | Busy Threads: 1
Loop Id: 03 | 0.018 | Busy Threads: 3
Loop Id: 04 | 0.018 | Busy Threads: 6
Loop Id: 05 | 0.018 | Busy Threads: 4
Loop Id: 06 | 0.018 | Busy Threads: 5
Loop Id: 07 | 0.018 | Busy Threads: 6
Loop Id: 08 | 0.018 | Busy Threads: 8
Task Id: 01 | 0.018 | Busy Threads: 11
Task Id: 04 | 0.018 | Busy Threads: 11
Task Id: 03 | 0.018 | Busy Threads: 11
Task Id: 02 | 0.018 | Busy Threads: 11
Task Id: 05 | 0.018 | Busy Threads: 11
Loop Id: 09 | 0.018 | Busy Threads: 12
Loop Id: 10 | 0.018 | Busy Threads: 12
Loop Id: 11 | 0.018 | Busy Threads: 12
Loop Id: 12 | 0.018 | Busy Threads: 12
Loop Id: 13 | 0.018 | Busy Threads: 12
Task Id: 09 | 0.018 | Busy Threads: 12
Loop Id: 14 | 0.018 | Busy Threads: 12
Loop Id: 15 | 0.018 | Busy Threads: 12
Loop Id: 16 | 0.018 | Busy Threads: 12
Loop Id: 17 | 0.018 | Busy Threads: 12
Task Id: 06 | 0.018 | Busy Threads: 12
Loop Id: 18 | 0.018 | Busy Threads: 12
Loop Id: 19 | 0.018 | Busy Threads: 12
Loop Id: 20 | 0.018 | Busy Threads: 12
Loop Id: 21 | 0.018 | Busy Threads: 12
Loop Id: 22 | 0.018 | Busy Threads: 12
Loop Id: 23 | 0.018 | Busy Threads: 12
Loop Id: 24 | 0.018 | Busy Threads: 12
Task Id: 10 | 0.018 | Busy Threads: 12
Task Id: 07 | 0.019 | Busy Threads: 12
Task Id: 11 | 0.019 | Busy Threads: 12
Task Id: 08 | 0.019 | Busy Threads: 12
Task Id: 12 | 0.019 | Busy Threads: 12
Task Id: 13 | 0.020 | Busy Threads: 16
Task Id: 14 | 0.020 | Busy Threads: 17
Task Id: 15 | 0.020 | Busy Threads: 18
Task Id: 16 | 0.020 | Busy Threads: 19
Task Id: 17 | 0.020 | Busy Threads: 20
Task Id: 18 | 0.020 | Busy Threads: 21
Task Id: 19 | 0.020 | Busy Threads: 22
Task Id: 20 | 0.020 | Busy Threads: 23
Task Id: 21 | 0.020 | Busy Threads: 24
Task Id: 23 | 0.020 | Busy Threads: 24
Task Id: 22 | 0.020 | Busy Threads: 24
Task Id: 24 | 0.020 | Busy Threads: 24
Task SetResult | 0.045 | Busy Threads: 25
Done: | 0.046与实验一相比虽然线程数仍然停留在 12 了一段时间但随后线程就立即增长了后文会介绍 .NET 6 在这方面做出的改进。.NET 6 实验二 调整 ThreadPool 设置将 .NET 5 实验二的代码在 .NET 6 中执行一次DefaultMinThreads: 12
Loop Id: 01 | 0.001 | Busy Threads: 0
Loop Id: 02 | 0.014 | Busy Threads: 1
Loop Id: 03 | 0.014 | Busy Threads: 2
Loop Id: 04 | 0.015 | Busy Threads: 5
Loop Id: 05 | 0.015 | Busy Threads: 4
Loop Id: 06 | 0.015 | Busy Threads: 5
Loop Id: 07 | 0.015 | Busy Threads: 7
Loop Id: 08 | 0.015 | Busy Threads: 8
Loop Id: 09 | 0.015 | Busy Threads: 11
Task Id: 06 | 0.015 | Busy Threads: 9
Task Id: 01 | 0.015 | Busy Threads: 9
Task Id: 02 | 0.015 | Busy Threads: 9
Task Id: 05 | 0.015 | Busy Threads: 9
Task Id: 03 | 0.015 | Busy Threads: 9
Task Id: 04 | 0.015 | Busy Threads: 9
Task Id: 07 | 0.015 | Busy Threads: 9
Task Id: 08 | 0.016 | Busy Threads: 9
Task Id: 09 | 0.016 | Busy Threads: 9
Loop Id: 10 | 0.016 | Busy Threads: 9
Loop Id: 11 | 0.016 | Busy Threads: 10
Loop Id: 12 | 0.016 | Busy Threads: 11
Loop Id: 13 | 0.016 | Busy Threads: 13
Task Id: 10 | 0.016 | Busy Threads: 14
Loop Id: 14 | 0.016 | Busy Threads: 14
Loop Id: 15 | 0.016 | Busy Threads: 14
Loop Id: 16 | 0.016 | Busy Threads: 14
Task Id: 11 | 0.016 | Busy Threads: 14
Loop Id: 17 | 0.016 | Busy Threads: 14
Loop Id: 18 | 0.016 | Busy Threads: 14
Loop Id: 19 | 0.016 | Busy Threads: 14
Loop Id: 20 | 0.016 | Busy Threads: 14
Loop Id: 21 | 0.016 | Busy Threads: 14
Loop Id: 22 | 0.016 | Busy Threads: 14
Loop Id: 23 | 0.016 | Busy Threads: 14
Loop Id: 24 | 0.016 | Busy Threads: 14
Task Id: 12 | 0.016 | Busy Threads: 14
Task Id: 13 | 0.016 | Busy Threads: 14
Task Id: 14 | 0.016 | Busy Threads: 14
Task Id: 15 | 0.017 | Busy Threads: 18
Task Id: 16 | 0.017 | Busy Threads: 19
Task Id: 17 | 0.017 | Busy Threads: 20
Task Id: 18 | 0.017 | Busy Threads: 21
Task Id: 19 | 0.017 | Busy Threads: 22
Task Id: 20 | 0.018 | Busy Threads: 23
Task Id: 21 | 0.018 | Busy Threads: 24
Task Id: 22 | 0.018 | Busy Threads: 25
Task Id: 23 | 0.018 | Busy Threads: 26
Task Id: 24 | 0.018 | Busy Threads: 26
Task SetResult | 0.018 | Busy Threads: 25
Done: | 0.019前半部分有部分日志乱序可以看到与实验三一样维持在最大线程数一小段时间之后立即就开始了线程增长。.NET 6 实验三 tcs.Task.Wait() 改为 Thread.Sleep将 .NET 5 实验三的代码在 .NET 6 中执行一次Loop Id: 01 | 0.003 | Busy Threads: 0
Loop Id: 02 | 0.024 | Busy Threads: 1
Loop Id: 03 | 0.025 | Busy Threads: 2
Loop Id: 04 | 0.025 | Busy Threads: 3
Loop Id: 05 | 0.025 | Busy Threads: 7
Loop Id: 06 | 0.025 | Busy Threads: 5
Loop Id: 07 | 0.025 | Busy Threads: 6
Loop Id: 08 | 0.025 | Busy Threads: 7
Loop Id: 09 | 0.025 | Busy Threads: 9
Loop Id: 10 | 0.025 | Busy Threads: 10
Loop Id: 11 | 0.026 | Busy Threads: 10
Loop Id: 12 | 0.026 | Busy Threads: 11
Loop Id: 13 | 0.026 | Busy Threads: 12
Loop Id: 14 | 0.026 | Busy Threads: 12
Loop Id: 15 | 0.026 | Busy Threads: 12
Loop Id: 16 | 0.026 | Busy Threads: 12
Loop Id: 17 | 0.026 | Busy Threads: 12
Loop Id: 18 | 0.026 | Busy Threads: 12
Loop Id: 19 | 0.026 | Busy Threads: 12
Loop Id: 20 | 0.026 | Busy Threads: 12
Loop Id: 21 | 0.026 | Busy Threads: 12
Loop Id: 22 | 0.026 | Busy Threads: 12
Loop Id: 23 | 0.026 | Busy Threads: 12
Loop Id: 24 | 0.026 | Busy Threads: 12
Task Id: 01 | 0.026 | Busy Threads: 12
Task Id: 02 | 0.026 | Busy Threads: 12
Task Id: 05 | 0.026 | Busy Threads: 12
Task Id: 04 | 0.026 | Busy Threads: 12
Task Id: 06 | 0.026 | Busy Threads: 12
Task Id: 08 | 0.026 | Busy Threads: 12
Task Id: 09 | 0.026 | Busy Threads: 12
Task Id: 03 | 0.026 | Busy Threads: 12
Task Id: 11 | 0.026 | Busy Threads: 12
Task Id: 10 | 0.026 | Busy Threads: 12
Task Id: 07 | 0.026 | Busy Threads: 12
Task Id: 12 | 0.026 | Busy Threads: 12
Task Id: 13 | 1.026 | Busy Threads: 13
Task Id: 14 | 2.027 | Busy Threads: 14
Task Id: 15 | 3.028 | Busy Threads: 15
Task Id: 16 | 4.030 | Busy Threads: 16
Task Id: 17 | 5.031 | Busy Threads: 17
Task Id: 18 | 6.032 | Busy Threads: 18
Task Id: 19 | 6.533 | Busy Threads: 19
Task Id: 20 | 7.035 | Busy Threads: 20
Task Id: 21 | 8.036 | Busy Threads: 21
Task Id: 22 | 8.537 | Busy Threads: 22
Task Id: 23 | 9.538 | Busy Threads: 23
Task Id: 24 | 10.039 | Busy Threads: 24
Done: | 22.041结果与 .NET 5 的实验三相差不大。线程注入对照上述的几组实验结果接下来以 .NET 6 中 C# 实现的 ThreadPool 作为资料来理解一下线程注入的几个阶段按个人理解进行的划分仅供参考。1. 第一个线程的出现随着任务被调度到队列上第一个线程被创建出来。下面是线程池在执行第一个任务的时候的代码摘要涉及到计数的并执行相关处理的地方代码都使用了 while(xxx) Interlocked 的方式来进行并发控制可以理解成乐观锁。这一阶段实际上我们只需要关注到 ThreadPoolWorkQueue.EnsureThreadRequested 方法就行了。可利用 Rider 的反编译 Debug 功能帮助我们学习。下面是第一个 Task.Run 的代码执行路径注意执行环节是 Main Threadpublic static class ThreadPool
{internal static readonly ThreadPoolWorkQueue s_workQueue new ThreadPoolWorkQueue();public static bool QueueUserWorkItem(WaitCallback callBack, object state){object tpcallBack new QueueUserWorkItemCallback(callBack!, state);s_workQueue.Enqueue(tpcallBack, forceGlobal: true);return true;}
}internal sealed class ThreadPoolWorkQueue
{[StructLayout(LayoutKind.Sequential)]private struct CacheLineSeparated{private readonly Internal.PaddingFor32 pad1;public volatile int numOutstandingThreadRequests;private readonly Internal.PaddingFor32 pad2;}private CacheLineSeparated _separated;public void Enqueue(object callback, bool forceGlobal){// 线程池中执行的任务有两种IThreadPoolWorkItem、TaskDebug.Assert((callback is IThreadPoolWorkItem) ^ (callback is Task));if (loggingEnabled FrameworkEventSource.Log.IsEnabled())FrameworkEventSource.Log.ThreadPoolEnqueueWorkObject(callback);ThreadPoolWorkQueueThreadLocals? tl null;if (!forceGlobal)// 获取本地队列如果执行改代码的线程不是线程池线程// 那这边是获取不到的就算 forceGlobal 是 false// 也会把任务放到全局队列tl ThreadPoolWorkQueueThreadLocals.threadLocals;if (null ! tl){// 放到本地队列tl.workStealingQueue.LocalPush(callback);}else{// 当道全局队列workItems.Enqueue(callback);}EnsureThreadRequested();}internal void EnsureThreadRequested(){//// If we have not yet requested #procs threads, then request a new thread.//// CoreCLR: Note that there is a separate count in the VM which has already been incremented// by the VM by the time we reach this point.//int count _separated.numOutstandingThreadRequests;while (count Environment.ProcessorCount){int prev Interlocked.CompareExchange(ref _separated.numOutstandingThreadRequests, count 1, count);if (prev count){ThreadPool.RequestWorkerThread();break;}count prev;}}public static class ThreadPool{/// summary/// This method is called to request a new thread pool worker to handle pending work./// /summaryinternal static void RequestWorkerThread() PortableThreadPool.ThreadPoolInstance.RequestWorker();}internal sealed class PortableThreadPool{public static readonly PortableThreadPool ThreadPoolInstance new PortableThreadPool();internal void RequestWorker(){// The order of operations here is important. MaybeAddWorkingWorker() and EnsureRunning() use speculative checks to// do their work and the memory barrier from the interlocked operation is necessary in this case for correctness.Interlocked.Increment(ref _separated.numRequestedWorkers);WorkerThread.MaybeAddWorkingWorker(this);// 初始化 GateThreadGateThread.EnsureRunning(this);}/// summary/// The worker thread infastructure for the CLR thread pool./// /summaryprivate static class WorkerThread{internal static void MaybeAddWorkingWorker(PortableThreadPool threadPoolInstance){ThreadCounts counts threadPoolInstance._separated.counts;short numExistingThreads, numProcessingWork, newNumExistingThreads, newNumProcessingWork;// 这个 while (true) 是确保计算出正确的待创建线程数while (true){numProcessingWork counts.NumProcessingWork;if (numProcessingWork counts.NumThreadsGoal){return;}newNumProcessingWork (short)(numProcessingWork 1);numExistingThreads counts.NumExistingThreads;newNumExistingThreads Math.Max(numExistingThreads, newNumProcessingWork);ThreadCounts newCounts counts;newCounts.NumProcessingWork newNumProcessingWork;newCounts.NumExistingThreads newNumExistingThreads;ThreadCounts oldCounts threadPoolInstance._separated.counts.InterlockedCompareExchange(newCounts, counts);if (oldCounts counts){break;}counts oldCounts;}int toCreate newNumExistingThreads - numExistingThreads;int toRelease newNumProcessingWork - numProcessingWork;if (toRelease 0){s_semaphore.Release(toRelease);}while (toCreate 0){if (TryCreateWorkerThread()){toCreate--;continue;}counts threadPoolInstance._separated.counts;while (true){ThreadCounts newCounts counts;newCounts.SubtractNumProcessingWork((short)toCreate);newCounts.SubtractNumExistingThreads((short)toCreate);ThreadCounts oldCounts threadPoolInstance._separated.counts.InterlockedCompareExchange(newCounts, counts);if (oldCounts counts){break;}counts oldCounts;}break;}}private static bool TryCreateWorkerThread(){try{// Thread pool threads must start in the default execution context without transferring the context, so// using UnsafeStart() instead of Start()Thread workerThread new Thread(s_workerThreadStart);workerThread.IsThreadPoolThread true;workerThread.IsBackground true;// thread name will be set in thread procworkerThread.UnsafeStart();}catch (ThreadStartException){return false;}catch (OutOfMemoryException){return false;}return true;}}}}2. 达到 min threads 之前的线程数增长细心的朋友会发现上面代码里 EnsureThreadRequested 方法有一个终止条件_separated.numOutstandingThreadRequests Environment.ProcessorCount每次新增一个 ThreadRequested这个数就会 1似乎允许创建的最大 Worker Thread 是 Environment.ProcessorCount其实 ThreadPoolWorkQueue 维护的 NumOutstandingThreadRequests 这个值会在线程池线程真正跑起来之后会在 ThreadPoolWorkQueue.Dispatch方法中 -1。也就是说只要有一个线程真正运行起来了就能创建第 Environment.ProcessorCount 1 个Thread。当然在向 ThreadPoolWorkQueue 加入第13个任务的时候第13个 Worker Thread 就算不允许创建也没关系因为任务已经入队了会被运行起来的 Worker Thread 取走。min threads 初始值为 运行环境 CPU 核心数可通过 ThreadPool.SetMinThreads 进行设置参数有效范围是 [1, max threads]。PortableThreadPool里维护了一个计数器 PortableThreadPool.ThreadPoolInstance._separated.counts记录了 Worker Thread 相关的三个数值NumProcessingWork当前正在执行任务的 Worker Thread。NumExistingThreads当前线程池中实际有的 Worker Thread。NumThreadsGoal当前允许创建的最大 Worker Thread初始值为 min threads。internal class PortableThreadPool{public static readonly PortableThreadPool ThreadPoolInstance new PortableThreadPool();private CacheLineSeparated _separated;private struct CacheLineSeparated{public ThreadCounts counts;}/// summary/// Tracks information on the number of threads we want/have in different states in our thread pool./// /summaryprivate struct ThreadCounts{/// summary/// Number of threads processing work items./// /summarypublic short NumProcessingWork { get; set; }/// summary/// Number of thread pool threads that currently exist./// /summarypublic short NumExistingThreads { get; set; }// summary/// Max possible thread pool threads we want to have./// /summarypublic short NumThreadsGoal { get; set; }}}3. 避免饥饿机制Starvation Avoidance上面讲到随着任务进入队列系统Worker Thread 将随之增长直到达到 NumThreadsGoal。NumThreadsGoal 是12前 12 个线程都被堵住了加入到队列系统的第 13 个任务没办法被这前 12 个线程领走执行。在这种情况下线程池的 Starvation Avoidance 机制就起到作用了。在上述所说的第一个阶段除了线程池中的第一个线程会被创建之外GateThread 也会随之被初始化。在第一阶段的代码摘录中可以看到 GateThread 的初始化。internal sealed class PortableThreadPool
{public static readonly PortableThreadPool ThreadPoolInstance new PortableThreadPool();internal void RequestWorker(){Interlocked.Increment(ref _separated.numRequestedWorkers);WorkerThread.MaybeAddWorkingWorker(this);// 初始化 GateThreadGateThread.EnsureRunning(this);}
}在 GateThread 是一个独立的线程每隔 500ms 进行检查一下如果 NumProcessingWork NumThreadsGoalWorkerThread.MaybeAddWorkingWorker 不添加 Worker Thread 的判断条件就设置新的 NumThreadsGoal NumProcessingWork 1并调用 WorkerThread.MaybeAddWorkingWorker这样新的 Worker Thread 就可以被 WorkerThread.MaybeAddWorkingWorker 创建。这就解释了为什么 .NET 5 实验一、二在线程数达到min threadsNumThreadsGoal 的默认值之后后面 Worker Thread 的增长是每 500ms 一个。由于在第三阶段中线程的增长会比较缓慢有经验的开发会在应用启动的时候设置一个较大的 min threads使其较晚或不进入第三阶段。线程注入在 .NET 6 中的改进.NET 6 与 .NET 5 的实验二相比达到 min threads 之后线程的增长速度有明显的差异而两者的实验三却相差不大。** .NET 6 对于 Task.Wait 导致线程池线程阻塞的场景进行了优化但如果并非此原因导致的线程数不够用依旧是 Starvation Avoidance 的策略。**新的 ThreadPool 提供了一个 ThreadPool.NotifyThreadBlocked 的内部接口里面会调用 GateThread.Wake 去唤醒 GateThread 本来 500ms 执行一次的逻辑这 500ms 的间隔时间是通过 AutoResetEvent 实现的所以 GateThread.Wake 也很简单。关键代码示意非真实代码internal class PortableThreadPool
{public bool NotifyThreadBlocked(){// ...GateThread.Wake(this);return true;}private static class GateThread{private static readonly AutoResetEvent DelayEvent new AutoResetEvent(initialState: false);// GateThread 入口方法private static void GateThreadStart(){while(true){DelayEvent.WaitOne(500);// ...}}public static void Wake(PortableThreadPool threadPoolInstance){DelayEvent.Set();EnsureRunning(threadPoolInstance);}}爬山算法Hill Climbing除了上述介绍的线程注入机制外从CLR 4.0开始线程池内实现了一个根据采集到线程池吞吐率数据每次任务完成时记录数据推导出该算法认为最优的线程池线程数量。算法实现位于 HillClimbing.ThreadPoolHillClimber.Update有兴趣的朋友可以去看一下。public (int newThreadCount, int newSampleMs) Update(int currentThreadCount, double sampleDurationSeconds, int numCompletions)currentThreadCount当前线程数sampleDurationSeconds采样间隔numCompletions这段采样时间间隔内完成的任务数newThreadCount新的线程数newSample新的采样间隔时间不必要线程的销毁如果线程需要被移除的时候本地队列还存在待执行任务则会将这些任务转移到全局队列中。在以下几个场景中线程池将会销毁掉不需要的线程并不一定全面只限于笔者当前认知。在无法从队列系统领取到任务时。通过爬山算法认定当前线程属于多余线程时。参考资料https://www.codeproject.com/Articles/3813/NET-s-ThreadPool-Class-Behind-The-Sceneshttps://devblogs.microsoft.com/dotnet/performance-improvements-in-net-6/#threadinghttps://mattwarren.org/2017/04/13/The-CLR-Thread-Pool-Thread-Injection-Algorithm/https://docs.microsoft.com/zh-CN/previous-versions/msp-n-p/ff963549(vpandp.10)?redirectedfromMSDN#thread-injection
