.net线程池内幕
本文通過對.NET4.5的ThreadPool源碼的分析講解揭示.NET線程池的內(nèi)幕，并總結(jié)ThreadPool設(shè)計的好與不足。
線程池的作用
線程池，顧名思義，線程對象池。Task和TPL都有用到線程池，所以了解線程池的內(nèi)幕有助于你寫出更好的程序。由于篇幅有限，在這里我只講解以下核心概念：
線程池的大小
如何調(diào)用線程池添加任務(wù)
線程池如何執(zhí)行任務(wù)
Threadpool也支持操控IOCP的線程，但在這里我們不研究它，涉及到task和TPL的會在其各自的博客中做詳解。
線程池的大小
不管什么池，總有尺寸，ThreadPool也不例外。ThreadPool提供了4個方法來調(diào)整線程池的大小：
SetMaxThreads
GetMaxThreads
SetMinThreads
GetMinThreads
SetMaxThreads指定線程池最多可以有多少個線程，而GetMaxThreads自然就是獲取這個值。SetMinThreads指定線程池中最少存活的線程的數(shù)量，而GetMinThreads就是獲取這個值。
為何要設(shè)置一個最大數(shù)量和有一個最小數(shù)量呢？原來線程池的大小取決于若干因素，如虛擬地址空間的大小等。比如你的計算機是4g內(nèi)存，而一個線程的初始堆棧大小為1m，那么你最多能創(chuàng)建4g/1m的線程（忽略操作系統(tǒng)本身以及其他進程內(nèi)存分配）；正因為線程有內(nèi)存開銷，所以如果線程池的線程過多而又沒有被完全使用，那么這就是對內(nèi)存的一種浪費，所以限制線程池的最大數(shù)是很make sense的。
那么最小數(shù)又是為啥？線程池就是線程的對象池，對象池的最大的用處是重用對象。為啥要重用線程，因為線程的創(chuàng)建與銷毀都要占用大量的cpu時間。所以在高并發(fā)狀態(tài)下，線程池由于無需創(chuàng)建銷毀線程節(jié)約了大量時間，提高了系統(tǒng)的響應(yīng)能力和吞吐量。最小數(shù)可以讓你調(diào)整最小的存活線程數(shù)量來應(yīng)對不同的高并發(fā)場景。
如何調(diào)用線程池添加任務(wù)
線程池主要提供了2個方法來調(diào)用：QueueUserWorkItem和UnsafeQueueUserWorkItem。
兩個方法的代碼基本一致，除了attribute不同，QueueUserWorkItem可以被partial trust的代碼調(diào)用，而UnsafeQueueUserWorkItem只能被full trust的代碼調(diào)用。
{
StackCrawlMark stackMark = StackCrawlMark.LookForMyCaller;
? ? return ThreadPool.QueueUserWorkItemHelper(callBack, (object) null, ref stackMark, true);
}
QueueUserWorkItemHelper首先調(diào)用ThreadPool.EnsureVMInitialized()來確保CLR虛擬機初始化（VM是一個統(tǒng)稱，不是單指java虛擬機，也可以指CLR的execution engine），緊接著實例化ThreadPoolWorkQueue，最后調(diào)用ThreadPoolWorkQueue的Enqueue方法并傳入callback和true。
[SecurityCritical]
public void Enqueue(IThreadPoolWorkItem callback, bool forceGlobal)
{
ThreadPoolWorkQueueThreadLocals queueThreadLocals = (ThreadPoolWorkQueueThreadLocals) null;
if (!forceGlobal)
queueThreadLocals = ThreadPoolWorkQueueThreadLocals.threadLocals;
if (this.loggingEnabled)
FrameworkEventSource.Log.ThreadPoolEnqueueWorkObject((object) callback);
if (queueThreadLocals != null)
{
queueThreadLocals.workStealingQueue.LocalPush(callback);
}
else
{
ThreadPoolWorkQueue.QueueSegment comparand = this.queueHead;
while (!comparand.TryEnqueue(callback))
{
Interlocked.CompareExchange<ThreadPoolWorkQueue.QueueSegment>(ref comparand.Next, new ThreadPoolWorkQueue.QueueSegment(), (ThreadPoolWorkQueue.QueueSegment) null);
for (; comparand.Next != null; comparand = this.queueHead)
Interlocked.CompareExchange<ThreadPoolWorkQueue.QueueSegment>(ref this.queueHead, comparand.Next, comparand);
}
}
this.EnsureThreadRequested();
}
ThreadPoolWorkQueue主要包含2個“queue”(實際是數(shù)組)，一個為QueueSegment（global work queue），另一個是WorkStealingQueue(local work queue)。兩者具體的區(qū)別會在Task/TPL里講解，這里暫不解釋。
由于forceGlobal是true，所以執(zhí)行到了comparand.TryEnqueue(callback)，也就是QueueSegment.TryEnqueue。comparand先從隊列的頭(queueHead)開始enqueue，如果不行就繼續(xù)往下enqueue，成功后再賦值給queueHead。
讓我們來看看QueueSegment的源代碼：
public QueueSegment()
{
this.nodes = new IThreadPoolWorkItem[256];
}
public bool TryEnqueue(IThreadPoolWorkItem node)
{
int upper;
int lower;
this.GetIndexes(out upper, out lower);
while (upper != this.nodes.Length)
{
if (this.CompareExchangeIndexes(ref upper, upper + 1, ref lower, lower))
{
Volatile.Write<IThreadPoolWorkItem>(ref this.nodes[upper], node);
return true;
}
}
return false;
}
這個所謂的global work queue實際上是一個IThreadPoolWorkItem的數(shù)組，而且限死256，這是為啥？難道是因為和IIS線程池(也只有256個線程）對齊？使用interlock和內(nèi)存寫屏障volatile.write來保證nodes的正確性，比起同步鎖性能有很大的提高。最后調(diào)用EnsureThreadRequested，EnsureThreadRequested會調(diào)用QCall把請求發(fā)送至CLR，由CLR調(diào)度ThreadPool。
線程池如何執(zhí)行任務(wù)
線程被調(diào)度后通過ThreadPoolWorkQueue的Dispatch方法來執(zhí)行callback。
internal static bool Dispatch()
{
ThreadPoolWorkQueue threadPoolWorkQueue = ThreadPoolGlobals.workQueue;
int tickCount = Environment.TickCount;
threadPoolWorkQueue.MarkThreadRequestSatisfied();
threadPoolWorkQueue.loggingEnabled = FrameworkEventSource.Log.IsEnabled(EventLevel.Verbose, (EventKeywords) 18);
bool flag1 = true;
IThreadPoolWorkItem callback = (IThreadPoolWorkItem) null;
try
{
ThreadPoolWorkQueueThreadLocals tl = threadPoolWorkQueue.EnsureCurrentThreadHasQueue();
while ((long) (Environment.TickCount - tickCount) < (long) ThreadPoolGlobals.tpQuantum)
{
try
{
}
finally
{
bool missedSteal = false;
threadPoolWorkQueue.Dequeue(tl, out callback, out missedSteal);
if (callback == null)
flag1 = missedSteal;
else
threadPoolWorkQueue.EnsureThreadRequested();
}
if (callback == null)
return true;
if (threadPoolWorkQueue.loggingEnabled)
FrameworkEventSource.Log.ThreadPoolDequeueWorkObject((object) callback);
if (ThreadPoolGlobals.enableWorkerTracking)
{
bool flag2 = false;
try
{
try
{
}
finally
{
ThreadPool.ReportThreadStatus(true);
flag2 = true;
}
callback.ExecuteWorkItem();
callback = (IThreadPoolWorkItem) null;
}
finally
{
if (flag2)
ThreadPool.ReportThreadStatus(false);
}
}
else
{
callback.ExecuteWorkItem();
callback = (IThreadPoolWorkItem) null;
}
if (!ThreadPool.NotifyWorkItemComplete())
return false;
}
return true;
}
catch (ThreadAbortException ex)
{
if (callback != null)
callback.MarkAborted(ex);
flag1 = false;
}
finally
{
if (flag1)
threadPoolWorkQueue.EnsureThreadRequested();
}
return true;
}
while語句判斷如果執(zhí)行時間少于30ms會不斷繼續(xù)執(zhí)行下一個callback。這是因為大多數(shù)機器線程切換大概在30ms，如果該線程只執(zhí)行了不到30ms就在等待中斷線程切換那就太浪費CPU了，浪費可恥啊！
Dequeue負責找到需要執(zhí)行的callback：
public void Dequeue(ThreadPoolWorkQueueThreadLocals tl, out IThreadPoolWorkItem callback, out bool missedSteal)
{
callback = (IThreadPoolWorkItem) null;
missedSteal = false;
ThreadPoolWorkQueue.WorkStealingQueue workStealingQueue1 = tl.workStealingQueue;
workStealingQueue1.LocalPop(out callback);
if (callback == null)
{
for (ThreadPoolWorkQueue.QueueSegment comparand = this.queueTail; !comparand.TryDequeue(out callback) && comparand.Next != null && comparand.IsUsedUp(); comparand = this.queueTail)
Interlocked.CompareExchange<ThreadPoolWorkQueue.QueueSegment>(ref this.queueTail, comparand.Next, comparand);
}
if (callback != null)
return;
ThreadPoolWorkQueue.WorkStealingQueue[] current = ThreadPoolWorkQueue.allThreadQueues.Current;
int num = tl.random.Next(current.Length);
for (int length = current.Length; length > 0; --length)
{
ThreadPoolWorkQueue.WorkStealingQueue workStealingQueue2 = Volatile.Read<ThreadPoolWorkQueue.WorkStealingQueue>(ref current[num % current.Length]);
if (workStealingQueue2 != null && workStealingQueue2 != workStealingQueue1 && workStealingQueue2.TrySteal(out callback, ref missedSteal))
break;
++num;
}
}
因為我們把callback添加到了global work queue，所以local work queue(workStealingQueue.LocalPop(out?callback))找不到callback，local work queue查找callback會在task里講解。接著又去global work queue查找，先從global work queue的起始位置查找直至尾部，因此global work quque里的callback是FIFO的執(zhí)行順序。
public bool TryDequeue(out IThreadPoolWorkItem node)
{
int upper;
int lower;
this.GetIndexes(out upper, out lower);
while (lower != upper)
{
// ISSUE: explicit reference operation
// ISSUE: variable of a reference type
int& prevUpper = @upper;
// ISSUE: explicit reference operation
int newUpper = ^prevUpper;
// ISSUE: explicit reference operation
// ISSUE: variable of a reference type
int& prevLower = @lower;
// ISSUE: explicit reference operation
int newLower = ^prevLower + 1;
if (this.CompareExchangeIndexes(prevUpper, newUpper, prevLower, newLower))
{
SpinWait spinWait = new SpinWait();
while ((node = Volatile.Read<IThreadPoolWorkItem>(ref this.nodes[lower])) == null)
spinWait.SpinOnce();
this.nodes[lower] = (IThreadPoolWorkItem) null;
return true;
}
}
node = (IThreadPoolWorkItem) null;
return false;
}
使用自旋鎖和內(nèi)存讀屏障來避免內(nèi)核態(tài)和用戶態(tài)的切換，提高了獲取callback的性能。如果還是沒有callback，那么就從所有的local work queue里隨機選取一個，然后在該local work queue里“偷取”一個任務(wù)(callback)。
拿到callback后執(zhí)行callback.ExecuteWorkItem()，通知完成。
總結(jié)
ThreadPool提供了方法調(diào)整線程池最少活躍的線程來應(yīng)對不同的并發(fā)場景。ThreadPool帶有2個work queue，一個golbal一個local。執(zhí)行時先從local找任務(wù)，接著去global，最后才會去隨機選取一個local偷一個任務(wù)，其中g(shù)lobal是FIFO的執(zhí)行順序。Work queue實際上是數(shù)組，使用了大量的自旋鎖和內(nèi)存屏障來提高性能。但是在偷取任務(wù)上，是否可以考慮得更多，隨機選擇一個local太隨意。首先要考慮偷取的隊列上必須有可執(zhí)行任務(wù)；其次可以選取一個不在調(diào)度中的線程的local work queue，這樣降低了自旋鎖的可能性，加快了偷取的速度；最后，偷取的時候可以考慮像golang一樣偷取別人queue里一半的任務(wù)，因為執(zhí)行完偷到的這一個任務(wù)之后，下次該線程再次被調(diào)度到還是可能沒任務(wù)可執(zhí)行，還得去偷取別人的任務(wù)，這樣既浪費CPU時間，又讓任務(wù)在線程上分布不均勻，降低了系統(tǒng)吞吐量！
另外，如果禁用log和ETW trace，可以使ThreadPool的性能更進一步。
原文地址： http://www.cnblogs.com/newbier/p/6192882.html
.NET社區(qū)新聞，深度好文，微信中搜索dotNET跨平臺或掃描二維碼關(guān)注
總結(jié)
- 上一篇： 中间件和微服务，Docker以及原生云架
- 下一篇： Visual Studio中使用Git