.NET?6線程池ThreadPool實現(xiàn)概述
前言
在即將發(fā)布的 .NET 6 runtime 中，默認的線程池實現(xiàn)從 C++ 代碼改為了 C#，更方便我們學(xué)習(xí)線程池的設(shè)計了。
https://github.com/dotnet/runtime/tree/release/6.0/src/libraries/System.Threading.ThreadPool
新的線程池實現(xiàn)位于 PortableThreadPool 中，原 ThreadPool 中的對外公開的接口會直接調(diào)用 PortableThreadPool 中的實現(xiàn)。
通過設(shè)置環(huán)境變量 ThreadPool_UsePortableThreadPool 為 0 可以設(shè)置成使用老的線程池實現(xiàn)。
https://github.com/dotnet/runtime/pull/43841/commits/b0d47b84a6845a70f011d1b0d3ce5adde9a4d7b7
本文以 .NET 6 runtime 源碼作為學(xué)習(xí)材料，對線程池的設(shè)計進行介紹。從目前的理解上來看，其整體的設(shè)計與原來 C++ 的實現(xiàn)并沒有特別大的出入。
注意：
- 本文不涉及細節(jié)的代碼實現(xiàn)，主要為大家介紹其整體設(shè)計。所展示的代碼并非原封不動的源碼，而是為了方便理解的簡化版。
ThreadPool.SetMaxThreads(int workerThreads, int completionPortThreads)中的completionPortThreads所相關(guān)的IOCP線程池是 .NET Framework 時代的遺留產(chǎn)物，用于管理 Windows 平臺專有的 IOCP 的回調(diào)線程池。目前沒看到有什么地方在用它了，completionPortThreads 這個參數(shù)也已經(jīng)沒有意義，底層IO庫是自己維護的IO等待線程池。本文只涉及 worker thread 池的介紹。- 本文理解并不完整也不一定完全正確，有異議的地方歡迎留言討論。
- 為了解釋問題，一部分代碼會運行在 .NET 6 之前的環(huán)境中。
任務(wù)的調(diào)度
線程池的待執(zhí)行任務(wù)被存放在一個隊列系統(tǒng)中。這個系統(tǒng)包括一個 全局隊列，以及綁定在每一個 Worker Thread 上 的 本地隊列 。而線程池中的每一個線程都在執(zhí)行 while(true) 的循環(huán)，從這個隊列系統(tǒng)中領(lǐng)取并執(zhí)行任務(wù)。
在 ThreadPool.QueueUserWorkItem 的重載方法 ThreadPool.QueueUserWorkItem<TState>(Action<TState> callBack, TState state, bool preferLocal) 里有一個 preferLocal 參數(shù)。
調(diào)用不帶 preferLocal 參數(shù)的 ThreadPool.QueueUserWorkItem 方法重載，任務(wù)會被放到全局隊列。
當 preferLocal 為 true 的時候，如果調(diào)用 ThreadPool.QueueUserWorkItem 代碼的線程正好是個線程池里的某個線程，則該任務(wù)就會進入該線程的本地隊列中。除此之外的情況則會被放到全局隊列中等待未來被某個 Worker Thread 撿走。
在線程池外的線程中調(diào)用，不管 preferLocal 傳的是什么，任務(wù)都會被放到全局隊列。
基本調(diào)度單元
本地隊列和全局隊列的元素類型被定義為 object，實際的任務(wù)類型分為兩類，在從隊列系統(tǒng)取到任務(wù)之后會判斷類型并執(zhí)行對應(yīng)的方法。
IThreadPoolWorkItem 實現(xiàn)類的實例。
/// <summary>Represents a work item that can be executed by the ThreadPool.</summary>
public interface IThreadPoolWorkItem
{
void Execute();
}
執(zhí)行 Execute 方法也就代表著任務(wù)的執(zhí)行。
IThreadPoolWorkItem 的具體實現(xiàn)有很多，例如通過 ThreadPool.QueueUserWorkItem(WaitCallback callBack) 傳入的 callBack 委托實例會被包裝到一個 QueueUserWorkItemCallback 實例里。QueueUserWorkItemCallback 是 IThreadPoolWorkItem 的實現(xiàn)類。
Task
class Task
{
internal void InnerInvoke();
}
執(zhí)行 InnerInvoke 會執(zhí)行 Task 所包含的委托。
全局隊列
全局隊列 是由 ThreadPoolWorkQueue 維護的，同時它也是整個隊列系統(tǒng)的入口，直接被 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 ConcurrentQueue<object> workItems = new ConcurrentQueue<object>();
// forceGlobal 為 true 時，push 到全局隊列，否則就放到本地隊列
public void Enqueue(object callback, bool forceGlobal);
}
本地隊列
線程池中的每一個線程都會綁定一個 ThreadPoolWorkQueueThreadLocals 實例，在 workStealingQueue 這個字段上保存著本地隊列。
internal sealed class ThreadPoolWorkQueueThreadLocals
{
// 綁定在線程池線程上
[ThreadStatic]
public static ThreadPoolWorkQueueThreadLocals threadLocals;
// 持有全局隊列的引用，以便能在需要的時候?qū)⑷蝿?wù)轉(zhuǎn)移到全局隊列上
public readonly ThreadPoolWorkQueue workQueue;
// 本地隊列的直接維護者
public readonly ThreadPoolWorkQueue.WorkStealingQueue workStealingQueue;
public readonly Thread currentThread;
public ThreadPoolWorkQueueThreadLocals(ThreadPoolWorkQueue tpq)
{
workQueue = tpq;
workStealingQueue = new ThreadPoolWorkQueue.WorkStealingQueue();
// WorkStealingQueueList 會集中管理 workStealingQueue
ThreadPoolWorkQueue.WorkStealingQueueList.Add(workStealingQueue);
currentThread = Thread.CurrentThread;
}
// 提供將本地隊列中的任務(wù)轉(zhuǎn)移到全局隊列中去的功能，
// 當 ThreadPool 通過后文將會介紹的 HillClimbing 算法判斷得出當前線程是多余的線程后，
// 會調(diào)用此方法對任務(wù)進行轉(zhuǎn)移
public void TransferLocalWork()
{
while (workStealingQueue.LocalPop() is object cb)
{
workQueue.Enqueue(cb, forceGlobal: true);
}
}
~ThreadPoolWorkQueueThreadLocals()
{
if (null != workStealingQueue)
{
// TransferLocalWork 真正的目的并非是為了在這里被調(diào)用，這邊只是確保任務(wù)不會丟的 fallback 邏輯
TransferLocalWork();
ThreadPoolWorkQueue.WorkStealingQueueList.Remove(workStealingQueue);
}
}
}
偷竊機制
這里思考一個問題，為什么本地隊列的名字會被叫做 WorkStealingQueue 呢？
所有 Worker Thread 的 WorkStealingQueue 都被集中在 WorkStealingQueueList 中。對線程池中其他所有線程可見。
Worker Thread 的 while(true) 中優(yōu)先會從自身的 WorkStealingQueue 中取任務(wù)。如果本地隊列已經(jīng)被清空，就會從全局隊列中取任務(wù)。例如下圖的 Thread1 取全局隊列中領(lǐng)取了一個任務(wù)。
同時 Thread3 也沒活干了，但是全局隊列中的任務(wù)被 Thread1 搶走了。這時候就會去 從 Thread2 的本地隊列中搶 Thread2 的活。
Worker Thread 的生命周期管理
接下來我們把格局放大，關(guān)注點從 Worker Thread 的打工日常轉(zhuǎn)移到對它們的生命周期管理上來。
為了更方便的解釋線程管理的機制，這邊使用下面使用一些代碼做演示。
代碼參考自 https://devblogs.microsoft.com/dotnet/performance-improvements-in-net-6/。
線程注入實驗
Task.Run 會將 Task 調(diào)度到線程池中執(zhí)行，下面的示例代碼中等效于 ThreadPool.QueueUserWorkItem(WaitCallback callBack)，會把 Task 放到隊列系統(tǒng)的全局隊列中（順便一提，如果在一個線程池線程中執(zhí)行 Task.Run 會將 Task 調(diào)度到此線程池線程的本地隊列中）。
.NET 5 實驗一 默認線程池配置
static void Main(string[] args)
{
var sw = Stopwatch.StartNew();
var tcs = new TaskCompletionSource();
var tasks = new List<Task>();
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 環(huán)境中運行以下代碼，CPU 邏輯核心數(shù) 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.136
Task.Run 會把 Task 調(diào)度到線程池上執(zhí)行，前 24 個 task 都會被阻塞住，直到第 25 個被執(zhí)行。每次都會打印出當前線程池中正在執(zhí)行任務(wù)的線程數(shù)（也就是創(chuàng)建完成的線程數(shù)）。
可以觀察到以下結(jié)果：
- 前幾次循環(huán)，線程隨著 Task 數(shù)量遞增，后面幾次循環(huán)直到循環(huán)結(jié)束為止，線程數(shù)一直維持在 12 沒有發(fā)生變化。
- 線程數(shù)在達到 12 之前，零間隔時間增加。第 12 到 第 13 線程間隔 1s 不到，往后約 500ms 增加一個線程。
.NET 5 實驗二 調(diào)整 ThreadPool 設(shè)置
在上面的代碼最前面加入以下兩行代碼，繼續(xù)在 .NET 5 環(huán)境運行一次。
ThreadPool.GetMinThreads(out int defaultMinThreads, out int completionPortThreads);
Console.WriteLine($"DefaultMinThreads: {defaultMinThreads}");
ThreadPool.SetMinThreads(14, completionPortThreads);
運行結(jié)果如下
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
在調(diào)整完線程池的最小線程數(shù)量之后，線程注入速度發(fā)生轉(zhuǎn)折的時間點從第 12（默認min threads） 個線程換到了第 14（修改后的min threads）個線程。
整體時間也從 8s 縮到 6s。
.NET 5 實驗三 tcs.Task.Wait() 改為 Thread.Sleep
static void Main(string[] args)
{
var sw = Stopwatch.StartNew();
var tasks = new List<Task>();
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 設(shè)置
將 .NET 5 實驗一的代碼在 .NET 6 執(zhí)行一次
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
與實驗一相比，雖然線程數(shù)仍然停留在 12 了一段時間，但隨后線程就立即增長了，后文會介紹 .NET 6 在這方面做出的改進。
.NET 6 實驗二 調(diào)整 ThreadPool 設(shè)置
將 .NET 5 實驗二的代碼在 .NET 6 中執(zhí)行一次
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
前半部分有部分日志亂序，可以看到，與實驗三一樣，維持在最大線程數(shù)一小段時間之后，立即就開始了線程增長。
.NET 6 實驗三 tcs.Task.Wait() 改為 Thread.Sleep
將 .NET 5 實驗三的代碼在 .NET 6 中執(zhí)行一次
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
結(jié)果與 .NET 5 的實驗三相差不大。
線程注入
對照上述的幾組實驗結(jié)果，接下來以 .NET 6 中 C# 實現(xiàn)的 ThreadPool 作為資料來理解一下線程注入的幾個階段（按個人理解進行的劃分，僅供參考）。
1. 第一個線程的出現(xiàn)
隨著任務(wù)被調(diào)度到隊列上，第一個線程被創(chuàng)建出來。
下面是線程池在執(zhí)行第一個任務(wù)的時候的代碼摘要，涉及到計數(shù)的并執(zhí)行相關(guān)處理的地方，代碼都使用了 while(xxx) + Interlocked 的方式來進行并發(fā)控制，可以理解成樂觀鎖。這一階段，實際上我們只需要關(guān)注到 ThreadPoolWorkQueue.EnsureThreadRequested 方法就行了。
可利用 Rider 的反編譯 Debug 功能幫助我們學(xué)習(xí)。
下面是第一個 Task.Run 的代碼執(zhí)行路徑
注意：執(zhí)行環(huán)節(jié)是 Main Thread
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
{
[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)
{
// 線程池中執(zhí)行的任務(wù)有兩種：IThreadPoolWorkItem、Task
Debug.Assert((callback is IThreadPoolWorkItem) ^ (callback is Task));
if (loggingEnabled && FrameworkEventSource.Log.IsEnabled())
FrameworkEventSource.Log.ThreadPoolEnqueueWorkObject(callback);
ThreadPoolWorkQueueThreadLocals? tl = null;
if (!forceGlobal)
// 獲取本地隊列，如果執(zhí)行改代碼的線程不是線程池線程，
// 那這邊是獲取不到的，就算 forceGlobal 是 false，
// 也會把任務(wù)放到全局隊列
tl = ThreadPoolWorkQueueThreadLocals.threadLocals;
if (null != tl)
{
// 放到本地隊列
tl.workStealingQueue.LocalPush(callback);
}
else
{
// 當?shù)廊株犃?
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.
/// </summary>
internal 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);
// 初始化 GateThread
GateThread.EnsureRunning(this);
}
/// <summary>
/// The worker thread infastructure for the CLR thread pool.
/// </summary>
private static class WorkerThread
{
internal static void MaybeAddWorkingWorker(PortableThreadPool threadPoolInstance)
{
ThreadCounts counts = threadPoolInstance._separated.counts;
short numExistingThreads, numProcessingWork, newNumExistingThreads, newNumProcessingWork;
// 這個 while (true) 是確保計算出正確的待創(chuàng)建線程數(shù)
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 proc
workerThread.UnsafeStart();
}
catch (ThreadStartException)
{
return false;
}
catch (OutOfMemoryException)
{
return false;
}
return true;
}
}
}
}
2. 達到 min threads 之前的線程數(shù)增長
細心的朋友會發(fā)現(xiàn)上面代碼里 EnsureThreadRequested 方法有一個終止條件，_separated.numOutstandingThreadRequests == Environment.ProcessorCount，每次新增一個 ThreadRequested，這個數(shù)就會 +1，似乎允許創(chuàng)建的最大 Worker Thread 是 Environment.ProcessorCount？
其實 ThreadPoolWorkQueue 維護的 NumOutstandingThreadRequests 這個值會在線程池線程真正跑起來之后，會在 ThreadPoolWorkQueue.Dispatch方法中 -1。也就是說，只要有一個線程真正運行起來了，就能創(chuàng)建第 Environment.ProcessorCount + 1 個Thread。當然，在向 ThreadPoolWorkQueue 加入第13個任務(wù)的時候，第13個 Worker Thread 就算不允許創(chuàng)建也沒關(guān)系，因為任務(wù)已經(jīng)入隊了，會被運行起來的 Worker Thread 取走。
min threads 初始值為 運行環(huán)境 CPU 核心數(shù)，可通過 ThreadPool.SetMinThreads 進行設(shè)置，參數(shù)有效范圍是 [1, max threads]。
PortableThreadPool里維護了一個計數(shù)器 PortableThreadPool.ThreadPoolInstance._separated.counts，記錄了 Worker Thread 相關(guān)的三個數(shù)值：
- NumProcessingWork：當前正在執(zhí)行任務(wù)的 Worker Thread。
- NumExistingThreads：當前線程池中實際有的 Worker Thread。
- NumThreadsGoal：當前允許創(chuàng)建的最大 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.
/// </summary>
private struct ThreadCounts
{
/// <summary>
/// Number of threads processing work items.
/// </summary>
public short NumProcessingWork { get; set; }
/// <summary>
/// Number of thread pool threads that currently exist.
/// </summary>
public short NumExistingThreads { get; set; }
// <summary>
/// Max possible thread pool threads we want to have.
/// </summary>
public short NumThreadsGoal { get; set; }
}
}
3. 避免饑餓機制（Starvation Avoidance）
上面講到，隨著任務(wù)進入隊列系統(tǒng)，Worker Thread 將隨之增長，直到達到 NumThreadsGoal。
NumThreadsGoal 是12，前 12 個線程都被堵住了，加入到隊列系統(tǒng)的第 13 個任務(wù)沒辦法被這前 12 個線程領(lǐng)走執(zhí)行。
在這種情況下，線程池的 Starvation Avoidance 機制就起到作用了。
在上述所說的第一個階段，除了線程池中的第一個線程會被創(chuàng)建之外，GateThread 也會隨之被初始化。在第一階段的代碼摘錄中，可以看到 GateThread 的初始化。
internal sealed class PortableThreadPool
{
public static readonly PortableThreadPool ThreadPoolInstance = new PortableThreadPool();
internal void RequestWorker()
{
Interlocked.Increment(ref _separated.numRequestedWorkers);
WorkerThread.MaybeAddWorkingWorker(this);
// 初始化 GateThread
GateThread.EnsureRunning(this);
}
}
在 GateThread 是一個獨立的線程，每隔 500ms 進行檢查一下，如果 NumProcessingWork >= NumThreadsGoal（WorkerThread.MaybeAddWorkingWorker 不添加 Worker Thread 的判斷條件），就設(shè)置新的 NumThreadsGoal = NumProcessingWork + 1，并調(diào)用 WorkerThread.MaybeAddWorkingWorker，這樣新的 Worker Thread 就可以被 WorkerThread.MaybeAddWorkingWorker 創(chuàng)建。
這就解釋了，為什么 .NET 5 實驗一、二在線程數(shù)達到min threads（NumThreadsGoal 的默認值）之后，后面 Worker Thread 的增長是每 500ms 一個。
由于在第三階段中，線程的增長會比較緩慢，有經(jīng)驗的開發(fā)會在應(yīng)用啟動的時候設(shè)置一個較大的 min threads，使其較晚或不進入第三階段。
線程注入在 .NET 6 中的改進
.NET 6 與 .NET 5 的實驗二相比，達到 min threads 之后，線程的增長速度有明顯的差異，而兩者的實驗三卻相差不大。
.NET 6 對于 Task.Wait 導(dǎo)致線程池線程阻塞的場景進行了優(yōu)化，但如果并非此原因?qū)е碌木€程數(shù)不夠用，依舊是 Starvation Avoidance 的策略。
新的 ThreadPool 提供了一個 ThreadPool.NotifyThreadBlocked 的內(nèi)部接口，里面會調(diào)用 GateThread.Wake 去喚醒 GateThread 本來 500ms 執(zhí)行一次的邏輯，這 500ms 的間隔時間是通過 AutoResetEvent 實現(xiàn)的，所以 GateThread.Wake 也很簡單。
關(guān)鍵代碼示意，非真實代碼：
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開始，線程池內(nèi)實現(xiàn)了一個根據(jù)采集到線程池吞吐率數(shù)據(jù)（每次任務(wù)完成時記錄數(shù)據(jù)），推導(dǎo)出該算法認為最優(yōu)的線程池線程數(shù)量。
算法實現(xiàn)位于 HillClimbing.ThreadPoolHillClimber.Update，有興趣的朋友可以去看一下。
public (int newThreadCount, int newSampleMs) Update(int currentThreadCount, double sampleDurationSeconds, int numCompletions)
- currentThreadCount：當前線程數(shù)
- sampleDurationSeconds：采樣間隔
- numCompletions：這段采樣時間間隔內(nèi)完成的任務(wù)數(shù)
- newThreadCount：新的線程數(shù)
- newSample：新的采樣間隔時間
不必要線程的銷毀
如果線程需要被移除的時候，本地隊列還存在待執(zhí)行任務(wù)，則會將這些任務(wù)轉(zhuǎn)移到全局隊列中。
在以下幾個場景中，線程池將會銷毀掉不需要的線程，并不一定全面，只限于筆者當前認知。
- 在無法從隊列系統(tǒng)領(lǐng)取到任務(wù)時。
- 通過爬山算法認定當前線程屬于多余線程時。
參考資料
https://www.codeproject.com/Articles/3813/NET-s-ThreadPool-Class-Behind-The-Scenes
https://devblogs.microsoft.com/dotnet/performance-improvements-in-net-6/
https://mattwarren.org/2017/04/13/The-CLR-Thread-Pool-Thread-Injection-Algorithm/
https://docs.microsoft.com/zh-CN/previous-versions/msp-n-p/ff963549(v=pandp.10)?redirectedfrom=MSDN
到此這篇關(guān)于.NET 6線程池ThreadPool實現(xiàn)方法的文章就介紹到這了。希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。
相關(guān)文章
本文主要是MVC4的入門介紹，演示如何創(chuàng)建一個MVC4的應(yīng)用程序。2016-04-04
在ASP.NET 2.0中操作數(shù)據(jù)之六十九：處理Computed Columns列
SQL Server創(chuàng)建表的時候允許我們創(chuàng)建computed columns列，這種列的值源與表達式，并且是只讀的，所以我們不能通過Insert和Update為其賦值，本文具體講解創(chuàng)建和調(diào)用顯示的方法和步驟。2016-05-05
解讀ASP.NET 5 & MVC6系列教程（9）：日志框架
這篇文章主要介紹了ASP.NET 5 日志框架,通過基于接口的編程機制和DI依賴注入機制，我們可以很容易實現(xiàn)第三方日志provider的擴展，從而將日志記錄到我們想記錄的任意地方,需要的朋友可以參考下2016-06-06
ASP.NET MVC4入門教程（五）：從控制器訪問數(shù)據(jù)模型
本文介紹ASP.NET MVC4中在創(chuàng)建控制器的時候，同時使用Entity Framework為模型生成增刪改查等方法，方便控制器調(diào)用。2016-04-04
在ASP.NET 2.0中操作數(shù)據(jù)之三十三：基于DataList和Repeater使用DropDownList過濾的主/
前面已經(jīng)介紹過使用DropDownList過濾的主/從報表，不過當時是基于GridView，本文算是復(fù)習(xí)一下，基于DataList和Repeater再次實現(xiàn)一下相同的功能。2016-05-05
解讀ASP.NET 5 & MVC6系列教程（2）：初識項目
這篇文章主要介紹ASP.NET 5中新建項目的結(jié)構(gòu)和之前的差異,介紹的比較細致,需要的朋友可以參考下。2016-06-06
在ASP.NET 2.0中操作數(shù)據(jù)之四十三：DataList和Repeater數(shù)據(jù)排序（二）
上篇已經(jīng)介紹了DropDownList隱式使用ViewState，本文主要介紹詳細介紹直接使用ViewState存儲排序的條件，并配合上一頁、下一頁按鈕，實現(xiàn)DataList和Repeater自定義排序的功能。2016-05-05
Microsoft .Net Remoting系列教程之三:Remoting事件處理全接觸
本文主要講解.Net Remoting中的Remoting事件處理，需要的朋友可以參考下。2016-05-05
AlternatingItemTemplate類似于 ItemTemplate 元素
AlternatingItemTemplate類似于 ItemTemplate 元素...2006-10-10
web.config文件的中文解釋...2006-10-10