Zum Inhalt
wannacode

Thread-Pool und Aufgaben

Kurz gesagt: Ein Thread-Pool sind ein paar Worker vor einer Aufgabenwarteschlange statt eines neuen Threads pro Aufgabe; in .NET ist das der gemeinsame Pool mit Task.Run, Parallel.For und Task.WhenAll.

Problem

Ein Thread für jedes Bild

Der Shop startet einen Sale. Verkäufer laden Bilder ihrer Produkte hoch, und jedes Bild braucht ein Vorschaubild für die Produktliste. Ein Vorschaubild zu erzeugen ist Arbeit, die rechnet: Sie geht durch jeden Pixel (siehe Arbeit, die wartet, und Arbeit, die rechnet).

Die erste Version startet für jedes hochgeladene Bild einen Thread:

var threads = new List<Thread>();
foreach (string image in uploads)
{
    var thread = new Thread(() => SaveThumbnail(image));
    thread.Start();
    threads.Add(thread);
}

Mit 20 Bildern funktioniert alles auf deinem Rechner: „Fertig: 20 Vorschaubilder“. In der ersten Stunde des Sales kommen 10.000 auf einmal.

  • Jeder Thread braucht einen eigenen Stack (meist bis zu 1 MB) und einen Platz im Betriebssystem. Tausende Threads fressen den Speicher des Servers.
  • Selbst wenn Speicher da ist, wechseln sich 10.000 Threads auf 8 Kernen nur ab. Der Prozessor verschwendet Zeit mit dem Umschalten, und jedes Bild wird langsamer.

Der Server wird nicht langsamer, er fällt um. Und ein Server, der down ist, bedient auch die Kunden nicht.

Die Idee

Ein paar Worker und eine Warteschlange

Ein Thread-Pool sind ein paar Worker, einmal erzeugt, und davor eine Warteschlange von Aufgaben. Eine Aufgabe ist „verkleinere dieses Bild“. Ein freier Worker nimmt die nächste Aufgabe aus der Warteschlange, erledigt sie und kommt für eine neue zurück.

  • Die Zahl der Worker ist konstant. 10.000 Bilder ergeben nicht 10.000 Threads, sondern 10.000 Aufgaben in der Warteschlange. Eine Aufgabe ist ein kleines Objekt; ein Thread ist teuer.
  • Die Worker werden wiederverwendet. Für jedes Bild wird kein Thread erzeugt und zerstört.
  • Die Warteschlange fängt die Spitze ab. Kommen Uploads schneller, als die Worker sie bearbeiten, warten die Aufgaben, statt den Server zu Fall zu bringen.

Das ist Producer und Consumer: Der Code, der Uploads annimmt, legt Aufgaben in die Warteschlange, und die Worker sind die Consumer.

Den Pool hast du schon in der Lektion über Threads kennengelernt: Task.Run, Task<int> und .Result. Hier sehen wir, was darin passiert, wie man die Ergebnisse ordnet, was mit Fehlern geschieht und wie viele Worker du brauchst.

Sieh es dir an

Drei Worker

Ein Vorschaubild zu erzeugen ist hier eine Pause, 2 ms für jedes KB, und die Größe kommt aus dem Namen des Bildes. So ist die Ausgabe bei jedem Lauf gleich:

// Vorschaubild erzeugen: Hier ist es eine Pause, und die Größe kommt aus dem Namen.
static int Thumbnail(string image)
{
    if (image.StartsWith("defekt"))
    {
        throw new ArgumentException("defekte Datei");
    }
    int kb = 20 + 3 * image.Length;
    Thread.Sleep(2 * kb);
    return kb;
}

Acht Bilder, ein Pool aus drei Workern:

int[] sizes = new int[images.Length];
var options = new ParallelOptions
{
    MaxDegreeOfParallelism = 3,
};
Parallel.For(0, images.Length, options, i =>
{
    sizes[i] = Thumbnail(images[i]);
});
int total = 0;
for (int i = 0; i < images.Length; i++)
{
    Console.WriteLine($"{images[i]}: {sizes[i]} KB");
    total += sizes[i];
}
Console.WriteLine($"Gesamt: {total} KB");
Schritt für Schritt, in Worten
  1. 8 Bilder warten. Parallel.For nutzt höchstens 3 Threads des .NET-Pools.
  2. Parallel.For verteilt die Indizes. Die drei Threads nehmen die ersten drei Bilder.
  3. Thread 3 wird als Erstes fertig: Hut ist das dritte Bild, und sein Ergebnis kommt an dritter Stelle. Der Worker hält nicht an: Er nimmt die nächste Aufgabe.
  4. Die Warteschlange ist leer. Thread 1 ist fertig und hat nichts mehr zu nehmen, also wartet er. Die anderen beenden ihre letzten Bilder.
  5. Jedes Ergebnis steht in sizes[i], also ist die Reihenfolge die der Bilder. Gesamt 379 KB.
string[] images = { "Kleid.jpg", "Mantel.jpg", "Hut.jpg", "Jacke.jpg", "Rucksack.jpg", "Gürtel.jpg", "Mütze.jpg", "Uhr.jpg" };
int[] sizes = new int[images.Length];
var options = new ParallelOptions
{
    MaxDegreeOfParallelism = 3,
};
Parallel.For(0, images.Length, options, i =>
{
    sizes[i] = Thumbnail(images[i]);
});
int total = 0;
for (int i = 0; i < images.Length; i++)
{
    Console.WriteLine($"{images[i]}: {sizes[i]} KB");
    total += sizes[i];
}
Console.WriteLine($"Gesamt: {total} KB");
Ausgabe
Kleid.jpg: 47 KB
Mantel.jpg: 50 KB
Hut.jpg: 41 KB
Jacke.jpg: 47 KB
Rucksack.jpg: 56 KB
Gürtel.jpg: 50 KB
Mütze.jpg: 47 KB
Uhr.jpg: 41 KB
Gesamt: 379 KB
  • .NET hat einen Thread-Pool (ThreadPool) für das ganze Programm und entscheidet selbst, wie viele Threads er hält. Du erzeugst ihn nicht und schließt ihn nicht.
  • Parallel.For führt den Rumpf für jeden Index im Pool aus und wartet auf alle. MaxDegreeOfParallelism = 3 heißt „höchstens drei gleichzeitig“: Das ist der Pool aus drei.
  • Jede Aufgabe schreibt in ihren eigenen Platz sizes[i], also braucht es keinen lock, und die Ergebnisse stehen in der Reihenfolge der Bilder.
Genauer, für Neugierige

.NET entscheidet selbst, wie viele Threads der Pool hält: Er startet mit so vielen, wie es Kerne gibt, und fügt weitere hinzu, wenn Aufgaben lange ohne Rechnen herumsitzen. ThreadPool.GetMaxThreads und SetMinThreads stellen das ein, aber das brauchst du selten.

Parallel.ForEach(images, options, image => …) ist dasselbe für eine Sammlung, und Parallel.ForEachAsync ist für Aufgaben mit await. Der Thread, der Parallel.For aufruft, arbeitet mit und wartet am Ende, rufe es also in einer Webanwendung nicht im Thread der Anfrage auf.

Wie es in den anderen Sprachen ist: In Java ist der Pool ein Objekt, das du erzeugst und schließt (ExecutorService), in Python ist es ThreadPoolExecutor, und Node hat keinen fertigen Pool von Workern.

Probier es aus

Finde den Fehler

Ein Kollege hat die Vorschaubilder mit einem Pool umgeschrieben, und der Sale wurde nicht schneller. Sechs Bilder brauchen so lange wie ohne Pool:

var watch = Stopwatch.StartNew();
var sizes = new List<int>();
foreach (string image in images)
{
    Task<int> size = Task.Run(() => Thumbnail(image));
    sizes.Add(await size);
}
Console.WriteLine($"Dauerte über 0,5 s: {watch.ElapsedMilliseconds > 500}");
Ausgabe
Dauerte über 0,5 s: True

Wähle die Zeile mit dem Fehler.

Die Idee

Ergebnisse in Eingabereihenfolge

Aufgaben werden in einer Reihenfolge fertig, die du nicht kennst: Sie hängt von den Bildern ab, von der Last, davon, welcher Worker frei war. Gib also nie „in der Reihenfolge des Fertigwerdens“ aus, wenn du jedes Mal dieselbe Ausgabe willst.

Zwei Wege liefern bei jedem Lauf dasselbe Ergebnis:

  • In Eingabereihenfolge. Du hältst die Promises in einer Liste in der Reihenfolge der Aufgaben und wartest in dieser Reihenfolge darauf. So funktioniert der Code oben. In C# liefert Task.WhenAll ein Array in der Reihenfolge der Tasks, nicht in der Reihenfolge des Fertigwerdens:
  • Eine Summe oder eine Anzahl. Brauchst du nur das Ergebnis insgesamt, ist die Reihenfolge egal: 47 + 50 ist dasselbe wie 50 + 47.
Task<int>[] tasks = images.Select(image => Task.Run(() => Thumbnail(image))).ToArray();
int[] sizes = await Task.WhenAll(tasks);
for (int i = 0; i < images.Length; i++)
{
    Console.WriteLine($"{images[i]}: {sizes[i]} KB");
}
Ausgabe
Kleid.jpg: 47 KB
Mantel.jpg: 50 KB
Hut.jpg: 41 KB
Jacke.jpg: 47 KB

Manchmal willst du die Ergebnisse in der Reihenfolge des Fertigwerdens: für einen Balken „3 von 8 fertig“ oder um die ersten fertigen Bilder zu zeigen. Task.WhenAny liefert sie so. Dann sammelst du nur Dinge, die nicht von der Reihenfolge abhängen:

var running = images.Select(image => Task.Run(() => Thumbnail(image))).ToList();
int count = 0;
int total = 0;
while (running.Count > 0)
{
    Task<int> done = await Task.WhenAny(running);
    running.Remove(done);
    total += await done;
    count++;
}
Console.WriteLine($"Fertig: {count} von {images.Length}, gesamt: {total} KB");
Ausgabe
Fertig: 8 von 8, gesamt: 379 KB
Task.WhenAny liefert die erste fertige Task aus der Liste. Du entfernst sie und wartest auf die nächste. Bei vielen Tasks ist das O(n²), deshalb hat .NET 9 Task.WhenEach bekommen, das sie einzeln liefert.

Die Idee

Ein Fehler in einer Aufgabe

Eines der Bilder ist defekt, und thumbnail wirft eine Exception. Sie kommt nicht in dem Thread heraus, der die Aufgabe gestartet hat: Der ist längst weitergezogen. Der Pool fängt sie ab und bewahrt sie im Promise des Ergebnisses auf. Sie kommt heraus, wenn du auf dieses Ergebnis wartest:

string[] images = { "Kleid.jpg", "Mantel.jpg", "defekt.jpg", "Hut.jpg" };
var tasks = images.Select(image => Task.Run(() => Thumbnail(image))).ToList();
for (int i = 0; i < images.Length; i++)
{
    try
    {
        int kb = await tasks[i];
        Console.WriteLine($"{images[i]}: {kb} KB");
    }
    catch (ArgumentException e)
    {
        Console.WriteLine($"{images[i]}: Fehler: {e.Message}");
    }
}
Ausgabe
Kleid.jpg: 47 KB
Mantel.jpg: 50 KB
defekt.jpg: Fehler: defekte Datei
Hut.jpg: 41 KB
await wirft die Exception selbst (ArgumentException). .Result würde sie in eine AggregateException verpackt werfen. Den anderen Bildern passiert nichts.

Probier es aus

Wohin ist der Fehler gegangen

Zwei der Bilder sind defekt. Was gibt das Programm aus?

string[] images = { "Kleid.jpg", "defekt-1.jpg", "defekt-2.jpg" };
Task<int[]> all = Task.WhenAll(images.Select(image => Task.Run(() => Thumbnail(image))));
try
{
    await all;
}
catch (ArgumentException e)
{
    Console.WriteLine($"Abgefangen: {e.Message}");
}
Console.WriteLine($"Fehler in der Aufgabe: {all.Exception!.InnerExceptions.Count}");
Antwort

Die Idee

Wie viele Worker

Das hängt davon ab, was sie tun:

  • Arbeit, die rechnet (Vorschaubilder, PDFs, Komprimierung): so viele wie Kerne. Ein neunter Worker auf 8 Kernen hat keinen Platz zum Laufen; er wechselt sich nur mit den anderen ab.
  • Arbeit, die wartet (E-Mail, Zahlung, Datenbank): mehr. Während ein Worker auf das Netzwerk wartet, ist der Kern für einen anderen frei. Wartet eine Aufgabe W ms und rechnet C ms, passen etwa 1 + W / C Worker auf einen Kern.
static int PoolSize(int cores, int waitMs, int computeMs)
{
    return cores * (1 + waitMs / computeMs);
}


Console.WriteLine($"Vorschaubilder (nur Rechnen): {PoolSize(8, 0, 50)}");
Console.WriteLine($"E-Mails (90 ms Warten, 10 ms Rechnen): {PoolSize(8, 90, 10)}");
Ausgabe
Vorschaubilder (nur Rechnen): 8
E-Mails (90 ms Warten, 10 ms Rechnen): 80

Die Zahl der Kerne kommt aus Environment.ProcessorCount. Die Formel ist ein Anfang, kein Gesetz: Miss nach. Gehen die E-Mails über einen Server mit einem Limit von 20 Verbindungen, warten 80 Worker nur in einer Schlange davor.

Probier es aus

Ergänze

Wie viele Worker für eine Aufgabe, die waitMs wartet und computeMs rechnet?

static int PoolSize(int cores, int waitMs, int computeMs)
{
    return cores * (1 + );
}
Genauer, für Neugierige

In C# hältst du für Arbeit, die wartet, gar keine Threads: await gibt den Thread frei, solange die Anfrage auf das Netzwerk wartet (die Lektion async und await). Deshalb bedient ein Webserver Tausende Anfragen mit einem kleinen Pool. Die Ressourcen dahinter (Datenbankverbindungen) werden begrenzt, zum Beispiel mit einem SemaphoreSlim.

Übung

Die Rechnungen des Sales

Löse die Aufgabe in deinem Editor.

Nach dem Sale braucht jede Bestellung eine Rechnung. Invoice(amount) erzeugt das PDF (hier eine Pause von 50 ms) und gibt den Betrag mit 20 % MwSt. zurück. Eine Bestellung mit einem Betrag von 0 oder weniger ist ungültig, und die Funktion wirft eine Exception mit dem Text „ungültiger Betrag“.

Die Bestellungen sind 1001 bis 1008 mit den Beträgen 50 €, 125 €, 80 €, 0 €, 240 €, 35 €, 90 € und 15 €. Erzeuge die Rechnungen in einem Pool aus 4 Workern. Gib sie in der Reihenfolge der Bestellungen aus, die ungültige mit dem Text des Fehlers. Am Ende: die Summe der guten Rechnungen und wie viele gescheitert sind.

Ausgabe
Bestellung 1001: 60 €
Bestellung 1002: 150 €
Bestellung 1003: 96 €
Bestellung 1004: Fehler: ungültiger Betrag
Bestellung 1005: 288 €
Bestellung 1006: 42 €
Bestellung 1007: 108 €
Bestellung 1008: 18 €
Gesamt: 762 €, mit Fehler: 1

Weitere Aufgaben

  • Eine begrenzte Warteschlange. Lass den Pool höchstens 10 Aufgaben in der Warteschlange warten. Was passiert mit der elften? (Channel.CreateBounded oder SemaphoreSlim)
  • Fortschritt. Gib nach jeder fertigen Rechnung „3 von 8 fertig“ aus, und am Ende noch einmal die geordnete Liste.
  • Ein Timeout. Eine Rechnung, die über 1 Sekunde dauert, gilt als gescheitert, ohne die anderen zu stoppen.

Test

Teste dich

1. Warum ein Pool aus ein paar Threads und nicht ein Thread für jedes hochgeladene Bild?
2. Ein Pool aus 3 Workern verkleinert 8 Bilder. Das dritte wird als Erstes fertig. Wo steht seine Größe, wenn du sie in Listenreihenfolge mit get, result, Promise.all oder sizes[i] liest?
3. Eine Aufgabe im Pool wirft eine Exception. Wann siehst du sie?
4. Der Server hat 8 Kerne. Jede E-Mail wartet 90 ms auf das Netzwerk und rechnet 10 ms. Etwa wie viele Worker halten die Kerne beschäftigt?
5. Wie viele Worker für Vorschaubilder, wenn die Arbeit nur rechnet und der Server 8 Kerne hat?

Erstellt von Ivo Gergov im Rahmen der Kampagne von Evo Tech für zugängliche Programmierausbildung ·