この記事では、.NET4.5 の ThreadPool ソース コードを分析して説明することで .NET スレッド プールの内部事情を明らかにし、ThreadPool 設計の長所と短所をまとめます。
スレッドプールの役割
スレッドプールは、名前が示すように、スレッドオブジェクトのプールです。タスクと TPL は両方ともスレッド プールを使用するため、スレッド プールの内部事情を理解することは、より良いプログラムを作成するのに役立ちます。スペースが限られているため、ここでは次の主要な概念のみを説明します:
スレッド プールのサイズ
どのような種類のプールであっても、常にサイズがあり、ThreadPool も例外ではありません。 ThreadPool には、スレッド プールのサイズを調整するための 4 つのメソッドが用意されています: s は、スレッド プールが持つことができるスレッドの最大数を指定します。 、GetMaxThreads 当然この値が取得されます。 SetMinThreads は、スレッド プール内で存続するスレッドの最小数を指定し、GetMinThreads はこの値を取得します。
では、最小数は何でしょうか?スレッド プールはスレッドのオブジェクト プールです。オブジェクト プールの最大の用途は、オブジェクトを再利用することです。なぜスレッドを再利用する必要があるのでしょうか? スレッドの作成と破棄には多くの CPU 時間がかかるからです。したがって、同時実行性が高い状態では、スレッド プールはスレッドを作成および破棄する必要がないため、時間を大幅に節約し、システムの応答性とスループットが向上します。 [最小数] を使用すると、さまざまな同時実行性の高いシナリオに対処するために、存続するスレッドの最小数を調整できます。
2 つのメソッドのコードは基本的に同じですが、QueueUserWorkItem は部分信頼コードで呼び出すことができるのに対し、UnsafeQueueUserWorkItem は完全信頼コードでのみ呼び出すことができます。
public static bool QueueUserWorkItem(WaitCallback callBack)
{
StackCrawlMark stackMark = StackCrawlMark.LookForMyCaller;
return ThreadPool.QueueUserWorkItemHelper(callBack, (object) null, ref stackMark, true);
}QueueUserWorkItemHelper は、最初に ThreadPool.EnsureVMInitialized() を呼び出して CLR 仮想マシンが初期化されていることを確認し (VM は Java 仮想マシンだけでなく CLR 実行エンジンも含む一般的な用語です)、次に ThreadPoolWorkQueue をインスタンス化して、最後に呼び出しますThreadPoolWorkQueue メソッドの Enqueue をコールバックと 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 つの「キュー」(実際には配列) が含まれており、1 つは QueueSegment (グローバル ワーク キュー)、もう 1 つは WorkStealingQueue (ローカル ワーク キュー) です。 2 つの具体的な違いについては、Task/TPL で説明するため、ここでは説明しません。
forceGlobalがtrueなので、comparand.TryEnqueue(callback)が実行され、これがQueueSegment.TryEnqueueになります。 comparand はキューの先頭 (queueHead) からエンキューを開始し、成功した後はキューへの値を割り当てます。
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;
}
スレッド プールがタスクを実行する方法
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;
}Dequeue は、実行する必要があるコールバックを見つける責任があります:
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;
}
}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;
}使用自旋锁和内存读屏障来避免内核态和用户态的切换，提高了获取callback的性能。如果还是没有callback，那么就从所有的local work queue里随机选取一个，然后在该local work queue里“偷取”一个任务(callback)。
拿到callback后执行callback.ExecuteWorkItem()，通知完成。
总结
ThreadPool提供了方法调整线程池最少活跃的线程来应对不同的并发场景。ThreadPool带有2个work queue，一个golbal一个local。
执行时先从local找任务，接着去global，最后才会去随机选取一个local偷一个任务，其中global是FIFO的执行顺序。
Work queue实际上是数组，使用了大量的自旋锁和内存屏障来提高性能。但是在偷取任务上，是否可以考虑得更多，随机选择一个local太随意。
首先要考虑偷取的队列上必须有可执行任务；其次可以选取一个不在调度中的线程的local work queue，这样降低了自旋锁的可能性，加快了偷取的速度；最后，偷取的时候可以考虑像golang一样偷取别人queue里一半的任务，因为执行完偷到的这一个任务之后，下次该线程再次被调度到还是可能没任务可执行，还得去偷取别人的任务，这样既浪费CPU时间，又让任务在线程上分布不均匀，降低了系统吞吐量！
另外，如果禁用log和ETW trace，可以使ThreadPool的性能更进一步。
以上就是.NET编程之线程池内幕的内容，更多相关内容请关注PHP中文网（www.php.cn）！
![[Web フロントエンド] Node.js クイック スタート](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
