Wielowątkowość i programowanie równoległe w C# - Thread, ThreadPool, TPL, Parallel

W poprzednim wpisie Architektura pluginów w C# widziałeś, jak refleksja pozwala dynamicznie ładować i wykonywać obcy kod. Teraz robimy krok w bok i wkraczamy w temat, który dotyczy każdej aplikacji działającej na nowoczesnym sprzęcie: wielowątkowość.

Twój komputer ma najprawdopodobniej 8, 12, 16 albo więcej rdzeni CPU. Program napisany sekwencyjnie używa jednego. Pozostałe leżą bezczynnie, podczas gdy aplikacja czeka. Wielowątkowość to mechanizm, którym te leżące rdzenie zatrudniamy do pracy. W tej części poznasz trzy poziomy abstrakcji - od najniższej Thread, przez ThreadPool, aż po nowoczesny TPL (Task, Parallel.For, PLINQ). Plus pułapki, których w wielowątkowym kodzie pełno jest niemal wszędzie.

1. Proces, wątek, rdzeń #

Trzy pojęcia, które warto rozumieć precyzyjnie:

• Proces - izolowana jednostka uruchamiana przez system operacyjny. Ma własną przestrzeń adresową, otwarte pliki, swoje uchwyty. Twoja aplikacja MyApp.exe to jeden proces.
• Wątek - sekwencja instrukcji wykonywana w obrębie procesu. Każdy proces ma co najmniej jeden wątek (Main). Wątki w tym samym procesie dzielą pamięć - z czego biorą się wyścigi o dane.
• Rdzeń CPU - fizyczna jednostka wykonująca instrukcje. System operacyjny rozdziela wątki na dostępne rdzenie. 8 rdzeni może wykonywać 8 wątków naprawdę równolegle.

Współbieżność (concurrency) kontra równoległość (parallelism):

• Współbieżność - kilka zadań postępuje na zmianę, niekoniecznie w tym samym czasie. Jeden rdzeń też potrafi (system przełącza między wątkami).
• Równoległość - kilka zadań naprawdę wykonuje się jednocześnie. Wymaga wielu rdzeni.

Wielowątkowość daje współbieżność. Równoległość pojawia się, gdy wątki trafią na wolne rdzenie.

2. Klasa Thread - najniższy poziom #

Co to jest: System.Threading.Thread to bezpośredni wrapper na wątek systemu operacyjnego. Najstarsza i najniższa abstrakcja w .NET.

using System;
using System.Threading;

void Worker()
{
    Console.WriteLine($"Worker thread #{Thread.CurrentThread.ManagedThreadId}");
}

Thread t = new Thread(Worker);
t.Start();          // uruchom wątek
t.Join();           // czekaj, aż się zakończy
Console.WriteLine("Worker finished");

Foreground kontra background #

Każdy wątek jest albo foreground, albo background:

	Foreground (domyślnie nowy Thread)	Background (ThreadPool, Task.Run)
Czy utrzymuje proces	Tak - aplikacja nie skończy się, dopóki działają wątki foreground	Nie - background giną razem z foregroundem
Kiedy stosować	Krytyczne zadania, które muszą się skończyć	Praca pomocnicza, którą można przerwać

Ustawienie: t.IsBackground = true; przed Start().

Pułapka: koszt tworzenia wątku #

Każdy new Thread(...).Start() alokuje stos (~1 MB), uchwyt OS, struktury TLS. Dla pojedynczych długich zadań to nieistotne. Dla setek krótkich - katastrofa wydajnościowa. W nowym kodzie prawie nigdy nie tworzysz Thread ręcznie. Zamiast tego - Task.Run, który używa puli wątków.

3. ThreadPool - pula wątków roboczych #

Co to jest: System.Threading.ThreadPool to pula gotowych wątków zarządzanych przez .NET. Zamiast tworzyć nowy wątek dla każdego zadania, “wynajmujesz” jeden z puli; po zakończeniu wątek wraca do puli, gotowy dla następnego.

Dlaczego istnieje: cytat z dokumentacji Microsoftu: the thread pool enables you to use threads more efficiently by providing your application with a pool of worker threads that are managed by the system. Tworzenie wątku jest drogie - pula amortyzuje ten koszt.

ThreadPool.QueueUserWorkItem(state =>
{
    Console.WriteLine($"Hello from pool thread #{Thread.CurrentThread.ManagedThreadId}");
});

Co warto zapamiętać #

• Jeden pool na proces - wszystkie zadania pochodzące z Task, Timer, async I/O, QueueUserWorkItem korzystają tego samego pula.
• Wątki są zawsze background - aplikacja nie poczeka na ich zakończenie.
• Brak gwarancji co do liczby równoległości - runtime sam dobiera rozmiar puli na podstawie obciążenia i liczby rdzeni.
• Brak Join() - nie ma jak czekać bezpośrednio na konkretny work item z QueueUserWorkItem. W nowoczesnym kodzie używaj Task.Run (zwraca Task, na którym można czekać).

Pułapka: stan wątku w storage #

Cytat MS: When the thread pool reuses a thread, it does not clear the data in thread local storage or in fields that are marked with the ThreadStaticAttribute attribute. Jeśli zostawisz coś w [ThreadStatic] pole, następne work item używające tego samego wątku zobaczy stare dane. Zawsze czyść po sobie.

4. TPL - Task Parallel Library #

Co to jest: TPL (Task Parallel Library) to nowoczesny, preferowany sposób pisania kodu równoległego od .NET Framework 4. Wszystko żyje w przestrzeniach System.Threading i System.Threading.Tasks.

Dlaczego istnieje: cytat MS: The TPL dynamically scales the degree of concurrency to use all the available processors most efficiently. In addition, the TPL handles the partitioning of the work, the scheduling of threads on the ThreadPool, cancellation support, state management, and other low-level details. Krótko: TPL robi za Ciebie to, co kiedyś musiałeś robić ręcznie.

Task - asynchroniczna jednostka pracy #

using System.Threading.Tasks;

// Task<T> - praca zwracająca wartość
Task<int> task = Task.Run(() =>
{
    Thread.Sleep(100);   // symulacja długiej pracy
    return 42;
});

int result = task.Result;   // czekaj synchronicznie (blokujące)
Console.WriteLine(result);

Task.Run kolejkuje delegata do ThreadPool i zwraca uchwyt Task (lub Task<T>). Na zakończenie czekasz:

• await task - asynchronicznie (w metodzie async, bez blokady wątku)
• task.Result lub task.Wait() - synchronicznie (blokuje bieżący wątek)
• Task.WhenAll(tasks) - czeka na zakończenie wszystkich
• Task.WhenAny(tasks) - na pierwszy, który skończy

Task kontra Thread - kiedy które? #

	new Thread().Start()	Task.Run(...)
Koszt	wysoki - alokacja stosu, uchwyt OS	niski - pula
Idiomatyczne	dla długich, dedykowanych zadań	dla krótkich zadań
Zwraca	nic - własny mechanizm sygnalizacji	Task<T> z wynikiem
Anulowanie	ręczne (flaga, Thread.Abort jest deprecated)	wbudowane (CancellationToken)

W praktyce: w 95% przypadków Task.Run. Thread rezerwuj na sytuacje, gdy naprawdę potrzebujesz dedykowanego wątku.

5. Parallel.For i Parallel.ForEach - paralelizm danych #

Co to jest: klasa System.Threading.Tasks.Parallel daje równoległe wersje pętli for i foreach. Logikę piszesz prawie identycznie jak w sekwencyjnej pętli - TPL sam dzieli pracę na partycje i przydziela je wątkom z ThreadPool.

Dlaczego istnieje: dokumentacja MS pisze: Data parallelism refers to scenarios in which the same operation is performed concurrently on elements in a source collection or array. In data parallel operations, the source collection is partitioned so that multiple threads can operate on different segments concurrently.

// Sekwencyjnie
for (int i = 0; i < 1000; i++)
{
    Process(i);
}

// Równolegle - identyczna logika iteracji, TPL dzieli zakres na partycje
Parallel.For(0, 1000, i => Process(i));

// Tak samo dla kolekcji
Parallel.ForEach(items, item => Process(item));

Pułapka: współdzielony stan #

Najczęstszy błąd początkujących z Parallel.For:

int sum = 0;
Parallel.For(0, 1000, i =>
{
    sum += i;        // ŹLE - kilka wątków zwiększa naraz, inkrementy gubione
});
// sum jest losowa i mniejsza niż 499500

Rozwiązania w kolejności rosnącego wyrafinowania:

// 1. lock - prosty, ale wolny przy dużej liczbie iteracji
int sum = 0;
object gate = new();
Parallel.For(0, 1000, i =>
{
    lock (gate) { sum += i; }
});

// 2. Interlocked - atomowe operacje, znacznie szybsze niż lock
int sum = 0;
Parallel.For(0, 1000, i => Interlocked.Add(ref sum, i));

// 3. Lokalna agregacja per wątek - najszybsze przy dużej pracy
int sum = 0;
Parallel.For(0, 1000,
    () => 0,                               // initLocal: lokalna suma startuje od 0
    (i, state, local) => local + i,        // body: dodaj do lokalnej
    local => Interlocked.Add(ref sum, local) // finalize: doklej lokalną do globalnej
);

Wariant 3 unika synchronizacji prawie zupełnie - każdy wątek sumuje do swojego lokalnego akumulatora, dopiero na koniec dokleja wynik do globalnego.

Przerwanie iteracji #

Parallel.For ma przeciążenia z ParallelLoopState - pozwala “wcześniej wyjść”:

Parallel.For(0, 1_000_000, (i, state) =>
{
    if (FoundWhatINeeded(i))
    {
        state.Break();   // zatrzymaj iteracje większe niż i (już rozpoczęte dokończą)
        // state.Stop();  - jeszcze ostrzejsze: ASAP, bez gwarancji co do mniejszych indeksów
    }
});

6. PLINQ - paralelizm w stylu zapytań #

Pamiętasz LINQ? PLINQ (Parallel LINQ) to dosłownie LINQ + .AsParallel():

using System.Linq;

// LINQ sekwencyjny
var primes = Enumerable.Range(2, 100_000)
    .Where(IsPrime)
    .ToList();

// PLINQ - dokładnie to samo, ale wewnątrz partycjonowane na wątki
var primes = Enumerable.Range(2, 100_000)
    .AsParallel()
    .Where(IsPrime)
    .ToList();

PLINQ używa pod spodem tej samej infrastruktury (Task, ThreadPool, partitioners). Sprawdza się dla obliczeń intensywnych, gdy każdy element wymaga sporo pracy. Dla “mało roboty per element” - podobnie jak Parallel.For - daje narzut zamiast zysku.

7. Synchronizacja - lock, Interlocked, Monitor #

Gdy wątki muszą dzielić dane, potrzebujesz synchronizacji. Trzy najważniejsze narzędzia:

lock - wzajemne wykluczenie #

private readonly object _gate = new();

public void Withdraw(int amount)
{
    lock (_gate)
    {
        if (_balance >= amount)
        {
            _balance -= amount;
        }
    }
}

Tylko jeden wątek na raz wykonuje sekcję krytyczną dla danego _gate. Konwencja: blokuj zawsze na prywatnym, niemutowalnym polu (private readonly object). Nigdy na this, string, czy typeof(X) - inne kawałki kodu mogą blokować na tych samych obiektach i powstanie zakleszczenie albo niedeterminizm.

Interlocked - atomowe operacje dla prostych liczników #

private int _counter;

public void IncrementCounter()
{
    Interlocked.Increment(ref _counter);   // atomowy ++
}

Interlocked używa instrukcji CPU LOCK XADD (lub odpowiednika) - cała operacja “czytaj+zwiększ+zapisz” jest nierozdzielna. Dla prostych przypadków znacznie tańsze niż lock.

Inne użyteczne:

• Interlocked.Add(ref x, n) - atomowe dodanie
• Interlocked.Exchange(ref x, v) - atomowe przypisanie
• Interlocked.CompareExchange(ref x, newVal, comparand) - “jeśli x == comparand, ustaw na newVal” (CAS)

Pułapka: deadlock #

Dwa wątki, dwie blokady, różna kolejność:

// Wątek A:
lock (lockX) {
    lock (lockY) { ... }   // czeka na lockY trzymane przez B
}

// Wątek B:
lock (lockY) {
    lock (lockX) { ... }   // czeka na lockX trzymane przez A
}
// → wzajemne czekanie do końca świata

Zapobieganie: zawsze blokuj w tej samej kolejności (np. alfabetycznie po nazwie), używaj Monitor.TryEnter z timeoutem, albo - najlepiej - unikaj wielu blokad naraz przez wyższe abstrakcje (TPL, async/await, kolejki).

Cytat z dokumentacji MS: we recommend that you have a basic understanding of threading concepts, for example, locks, deadlocks, and race conditions, so that you can use the TPL effectively. TPL nie chroni przed wyścigami i zakleszczeniami - tylko upraszcza partycjonowanie pracy.

8. Kiedy NIE używać Parallel #

Najczęstszy mit wokół wielowątkowości: im więcej wątków, tym szybciej. To nieprawda.

Cytat z MS, dosłownie: if a loop performs only a small amount of work on each iteration, or it doesn’t run for many iterations, then the overhead of parallelization can cause the code to run more slowly.

Paralelizacja ma narzut:

• Partycjonowanie pracy
• Przekazanie do ThreadPool
• Synchronizacja na początku i końcu
• Cache invalidation między rdzeniami

Dla małego N lub krótkich iteracji ten narzut jest większy niż zysk. Zasada ogólna:

Sytuacja	Wybór
Pętla 5 elementów × prosta operacja	sekwencyjny for
Pętla 5_000_000 elementów × prosta operacja	sekwencyjny for zwykle szybszy (narzut paralelizacji > zysk)
Pętla 1_000 elementów × ciężka operacja (parsowanie, kryptografia, kompresja)	Parallel.For
Pętla z I/O (HTTP, dysk, baza)	async/await z Task.WhenAll zamiast Parallel

Zasada ogólna nadrzędna: zmierz. Benchmark (np. BenchmarkDotNet) zamiast intuicji.

I/O - inny problem niż CPU-bound #

Parallel.For jest projektowane dla CPU-bound (intensywne obliczenia). Dla operacji I/O-bound (czekanie na sieć, dysk, bazę danych) lepiej działa async/await z Task.WhenAll - jeden wątek może obsłużyć tysiące oczekujących operacji I/O. Wątek czekający na pakiet TCP marnuje 1 MB stosu, podczas gdy nic nie robi. Async to inaczej rozwiązany problem.

Podsumowanie tematu #

W tej części kursu poznałeś:

• Proces, wątek, rdzeń - proces izolowany, wątki współdzielą pamięć w procesie, rdzeń wykonuje instrukcje; współbieżność (concurrency) ≠ równoległość (parallelism)
• Thread - najniższy poziom; foreground (utrzymuje proces) kontra background (giną razem z foregroundem); drogi w tworzeniu
• ThreadPool - pula gotowych wątków, jeden na proces, wszystkie background; QueueUserWorkItem (rzadko już używane bezpośrednio)
• TPL: Task + Task.Run - nowoczesny standard; Task.Result/Wait/await; Task.WhenAll/WhenAny
• Parallel.For / Parallel.ForEach - partycjonowanie danych na wątki; pułapka współdzielonego stanu i trzy rozwiązania (lock, Interlocked, lokalna agregacja)
• PLINQ - .AsParallel() dla zapytań LINQ z dużą pracą per element
• Synchronizacja - lock (wzajemne wykluczenie), Interlocked (atomowe operacje), Monitor.TryEnter (z timeoutem)
• Pułapki - race condition (niedeterminizm), deadlock (wzajemne czekanie), narzut paralelizacji dla małego N (cytat MS)
• Kiedy nie używać Parallel - małe N, krótkie iteracje, I/O-bound (zamiast tego async/await)

W następnym wpisie Synchronizacja wątków w C# zejdziemy głębiej w to, co pojawiło się tu jako wzmianka - lock, Monitor.TryEnter, SemaphoreSlim z WaitAsync (jedyny mechanizm zgodny z async/await) oraz ReaderWriterLockSlim dla wzorca read-heavy.

Quiz i zadanie poniżej. Zadanie używa Parallel.For z bezpiecznym Interlocked.Increment - dokładnie tego wzorca, który spotkasz w produkcyjnym kodzie obliczeniowym.

using System;
using System.Threading;
using System.Threading.Tasks;

public static class PrimeCounter
{
    public static bool IsPrime(int n)
    {
        if (n < 2) return false;
        if (n == 2) return true;
        if (n % 2 == 0) return false;
        int limit = (int)Math.Sqrt(n);
        for (int i = 3; i <= limit; i += 2)
        {
            if (n % i == 0) return false;
        }
        return true;
    }

    public static int CountPrimesParallel(int from, int to)
    {
        int count = 0;
        Parallel.For(from, to, i =>
        {
            if (IsPrime(i))
            {
                Interlocked.Increment(ref count);
            }
        });
        return count;
    }
}

public class Program
{
    public static void Main()
    {
        int result = PrimeCounter.CountPrimesParallel(2, 101);
        Console.WriteLine($"Prime numbers in 2..100: {result}");
    }
}