.NET-Programmier-Thread-Pool-Insider

Dieser Artikel enthüllt die Insider-Geschichte des .NET-Thread-Pools, indem er den ThreadPool-Quellcode von .NET4.5 analysiert und erklärt, und fasst die Vor- und Nachteile des ThreadPool-Designs zusammen.

Die Rolle des Thread-Pools

Der Thread-Pool ist, wie der Name schon sagt, ein Thread-Objektpool. Sowohl Task als auch TPL verwenden Thread-Pools. Wenn Sie also die Insidergeschichte von Thread-Pools verstehen, können Sie bessere Programme schreiben. Aus Platzgründen werde ich hier nur die folgenden Kernkonzepte

erläutern:

• Die Größe des Thread-Pools

• So rufen Sie Threads auf und fügen Aufgaben zum Pool hinzu

• Wie der Thread-Pool Aufgaben ausführt

Threadpool unterstützt auch die Bearbeitung von IOCP-Threads, aber wir werden es tun Die hier nicht untersuchten Aufgaben und TPL werden in den jeweiligen Blogs ausführlich erläutert.

Thread-Pool-Größe

Egal welche Art von Pool es gibt, es gibt immer eine Größe, und ThreadPool ist keine Ausnahme. ThreadPool bietet 4 Methoden zum Anpassen der Größe des Thread-Pools:

• SetMaxThreads

• GetMaxThreads

• SetMinThreads

• GetMinThreads

SetMaxThreads gibt die maximale Anzahl von Threads an, die der Thread-Pool haben kann, und GetMaxThreads erhält diesen Wert natürlich. SetMinThreads gibt die Mindestanzahl überlebender Threads im Thread-Pool an und GetMinThreads ruft diesen Wert ab.

Warum müssen wir eine Höchstmenge und eine Mindestmenge festlegen? Es stellt sich heraus, dass die Größe des Thread-Pools von mehreren Faktoren abhängt, beispielsweise von der Größe des virtuellen Adressraums. Wenn Ihr Computer beispielsweise über 4 g Arbeitsspeicher verfügt und die anfängliche Stapelgröße eines Threads 1 m beträgt, können Sie bis zu 4 g/1 m Threads erstellen (ohne Berücksichtigung des Betriebssystems selbst und anderer Prozessspeicherzuweisungen), gerade weil Threads Speicher-Overhead haben Wenn der Thread zu viele Threads im Pool enthält und diese nicht vollständig genutzt werden, stellt dies eine Speicherverschwendung dar. Daher ist es sinnvoll, die maximale Anzahl von Thread-Pools zu begrenzen.

Was ist dann die Mindestanzahl? Der Thread-Pool ist der Objektpool von Threads. Der größte Nutzen des Objektpools besteht in der Wiederverwendung von Objekten. Warum sollten wir Threads wiederverwenden, weil das Erstellen und Zerstören von Threads viel CPU-Zeit in Anspruch nimmt? Daher spart der Thread-Pool in einem Zustand mit hoher Parallelität viel Zeit, da er keine Threads erstellen und zerstören muss, was die Reaktionsfähigkeit und den Durchsatz des Systems verbessert. Mit der Mindestanzahl können Sie die Mindestanzahl überlebender Threads anpassen, um verschiedene Szenarien mit hoher Parallelität zu bewältigen.

So rufen Sie den Thread-Pool auf, um eine Aufgabe hinzuzufügen

Der Thread-Pool bietet hauptsächlich zwei Methoden zum Aufrufen: QueueUserWorkItem und UnsafeQueueUserWorkItem .

Die Codes der beiden Methoden sind grundsätzlich gleich, mit Ausnahme der unterschiedlichen Attribute, die durch teilweise vertrauenswürdigen Code aufgerufen werden können, während UnsafeQueueUserWorkItem nur durch vollständige Vertrauenswürdigkeit aufgerufen werden kann Code.

public static bool QueueUserWorkItem(WaitCallback callBack)
{
StackCrawlMark stackMark = StackCrawlMark.LookForMyCaller;
return ThreadPool.QueueUserWorkItemHelper(callBack, (object) null, ref stackMark, true);
}

QueueUserWorkItemHelper ruft zuerst ThreadPool.EnsureVMInitialized() auf, um sicherzustellen, dass die virtuelle CLR-Maschine initialisiert wird (VM ist ein allgemeiner Begriff, nicht nur die virtuelle Java-Maschine, sondern auch die CLR-Ausführungs-Engine) und dann instanziiert ThreadPoolWorkQueue und ruft schließlich die Enqueue-Methode von ThreadPoolWorkQueue auf und übergibt Rückruf und 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 enthält hauptsächlich zwei „Warteschlangen“ (eigentlich Arrays), eine ist QueueSegment (globale Arbeitswarteschlange) und die andere ist WorkStealingQueue (lokale Arbeitswarteschlange). Die spezifischen Unterschiede zwischen den beiden werden in Task/TPL erläutert und hier nicht erläutert.

Da forceGlobal wahr ist, wird comparand.TryEnqueue(callback) ausgeführt, was QueueSegment.TryEnqueue ist. comparand beginnt mit dem Einreihen in die Warteschlange (queueHead). Wenn dies fehlschlägt, wird das Einreihen in die Warteschlange fortgesetzt. Nach Erfolg wird der Wert queueHead zugewiesen.

Werfen wir einen Blick auf den Quellcode von 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;
}

Diese sogenannte globale Arbeitswarteschlange ist eigentlich ein Array von IThreadPoolWorkItem und auf 256 begrenzt. Warum? Das? Liegt es daran, dass es am IIS-Thread-Pool ausgerichtet ist (der nur 256 Threads hat)? Verwenden Sie Interlock und die Speicherschreibbarriere volatile.write, um die Korrektheit der Knoten sicherzustellen, was die Leistung im Vergleich zu Synchronisationssperren erheblich verbessert.

Rufen Sie schließlich „EsureThreadRequested“ auf, um QCall aufzurufen, um die Anforderung an die CLR zu senden, und die CLR plant den ThreadPool.

Wie der Thread-Pool Aufgaben ausführt

Nachdem der Thread geplant wurde, wird der Rückruf über die Dispatch-Methode von ThreadPoolWorkQueue ausgeführt.

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;
}

Die while-Anweisung bestimmt, dass sie den nächsten Rückruf weiterhin ausführt, wenn die Ausführungszeit weniger als 30 ms beträgt. Dies liegt daran, dass die meisten Maschinen-Threadwechsel etwa 30 ms dauern. Wenn der Thread nur weniger als 30 ms lang ausgeführt wird und dann auf den Interrupt-Thread wartet, wäre das eine Verschwendung von CPU.

Dequeue ist dafür verantwortlich, den Rückruf zu finden, der ausgeführt werden muss:

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;
}
}

Da wir den Rückruf zur globalen Arbeitswarteschlange hinzugefügt haben, ist die lokale Arbeitswarteschlange (workStealingQueue.LocalPop (out callback)) finds Wenn der Callback nicht gefunden wird, wird das Finden des Callbacks in der lokalen Arbeitswarteschlange in der Aufgabe erläutert. Gehen Sie dann zur Suche in die globale Arbeitswarteschlange. Suchen Sie zunächst vom Anfang bis zum Ende der globalen Arbeitswarteschlange, sodass der Rückruf in der globalen Arbeitswarteschlange die Ausführungsreihenfolge des FIFO ist.

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）！