1. Idea producer–consumer i role
Jeśli spojrzeć na rzeczywiste programy, bardzo często jeden fragment kodu „produkuje pracę”, a inny ją „przetwarza”. Na przykład użytkownik naciska przyciski (produkuje zdarzenia), a handler zdarzeń musi je obsłużyć; sieć dostarcza pakiety, a logika biznesowa je przetwarza; logger otrzymuje komunikaty, a potem zapisuje je do pliku. W takich sytuacjach chcemy, żeby producent nie czekał na przetworzenie, a konsument nie wykonywał pętli „gdzie jest moja praca?” i nie zużywał CPU, jakby mu za to płacono.
Właśnie tutaj pojawia się model producer–consumer. Producer (producent) wkłada „zadania” do kolejki, consumer (konsument) je pobiera i wykonuje. Pomiędzy nimi jest kolejka zadań — wspólny zasób, który chronimy mutex-em i „ożywiamy” condition_variable, aby consumer mógł spać, dopóki pracy nie ma.
Trzy role: producer, kolejka i consumer
Aby nie zamieniać wielowątkowości w magię i rytuały, wygodnie jest mentalnie podzielić system na trzy części. Producer — to ten, kto dodaje elementy do kolejki. Consumer — to ten, kto wyciąga elementy i wykonuje pracę. Kolejka — to „bufor” między nimi i jednocześnie źródło prawdy: czy jest praca teraz, czy nie. Powiadomienia (notify_*) — to nie „praca”, a jedynie delikatne szturchnięcie: „obudź się i sprawdź stan”.
Mała tabelka, żeby utrwalić role:
| Rola | Co robi | Z czym pracuje | Ryzyko przy błędnym działaniu |
|---|---|---|---|
| Producer | |
|
utrata powiadomienia / nieobudzenie consumera |
| Kolejka | przechowuje zadania | |
warunki wyścigu, uszkodzenie struktury, zablokowanie |
| Consumer | oczekuje i |
|
busy-wait, odczyt pustej kolejki, deadlock |
Schematycznie wygląda to tak:
flowchart LR
P[Producer<br/>wkłada zadania] -->|push pod mutex| Q[Kolejka zadań<br/>wspólny stan]
Q -->|pop pod mutex| C[Consumer / Worker<br/>przetwarza]
P -->|notify_one| CV[condition_variable]
CV -->|budzi| C
Kluczowa myśl: consumer oczekuje na stan kolejki, a nie na samo notify (np. !q.empty()). Notify — to tylko sygnał „warunek mógł się zmienić”.
Kolejka zadań jako chroniony stan
Bardzo kuszące jest napisanie czegoś w rodzaju „sprawdzimy q.empty(), jeśli pusto — poczekamy, jeśli nie — pobierzemy”. Na jednym wątku to OK. W wielowątkowości to staje się scenariuszem „spojrzałem — było pusto, mrugnąłem — i wszystko się zmieniło”. Dlatego w producer–consumer obowiązuje żelazna zasada: wszystkie operacje na kolejce — tylko pod tym samym mutex. Nawet empty() — to odczyt stanu współdzielonego, a odczyt bez dyscypliny nie jest „bezpieczniejszy” od zapisu.
Zazwyczaj chronimy takie operacje:
- Producer: lock → q.push(x) → unlock → notify_one
- Consumer: lock → wait(dopóki !q.empty()) → pobierz element → q.pop() → unlock → przetwórz element
Zwróć uwagę na słowo „przetwórz”: przetwarzanie powinno być poza mutexem. Pod mutexem wykonujemy tylko minimum: sprawdzenie warunku, przeniesienie danych z kolejki do lokalnej zmiennej, aktualizację kolejki.
Aby poczuć, dlaczego empty() też nie wolno dotykać bez mutexa, oto mini-antyprzykład (nie rób tego w domu, nawet jeśli bardzo kusi):
#include <queue>
std::queue<int> q;
bool has_work_unsafe() {
return !q.empty(); // ZLE: w wielowątkowości tak nie wolno bez mutexa
}
Nawet jeśli „wydaje ci się”, że empty() to „tylko podgląd”, w wielowątkowości to i tak odczyt zasobu współdzielonego bez gwarancji spójności. Potem zwykle następuje klasyka: zobaczyłeś false (niepusto), sięgnąłeś po front(), a kolejka już jest pusta — i witaj, błąd logiczny.
2. Protokół: mutex, oczekiwanie i wykonanie
Granica sekcji krytycznej: „wziąłem zadanie — zwolniłem mutex”
Początkujący często popełniają jedną z dwóch skrajności. Pierwsza — trzymać mutex zawsze, „żeby było bezpiecznie”. Wtedy producer nie może nic dodać, dopóki consumer długo przetwarza zadanie, i cały sens wielowątkowości znika. Druga — trzymać mutex prawie nigdy i liczyć, że „jakoś to przejdzie”. Zwykle nie przechodzi.
Prawidłowy styl: pod mutex wykonujemy tylko „manipulacje kolejką”, a pracę wykonujemy po wyjściu z sekcji krytycznej.
Porównajmy dwa fragmenty. Najpierw zły (mutex trzymany za długo):
#include <mutex>
#include <queue>
#include <iostream>
std::mutex m;
std::queue<int> q;
void process_bad() {
std::lock_guard<std::mutex> lock(m);
int x = q.front();
q.pop();
// ZLE: pod mutexem wykonujemy długą pracę
std::cout << "Processing " << x << "\n";
}
Teraz dobry (mutex krótki, przetwarzanie na zewnątrz):
#include <mutex>
#include <queue>
#include <iostream>
std::mutex m;
std::queue<int> q;
void process_good() {
int x = 0;
{
std::lock_guard<std::mutex> lock(m);
x = q.front();
q.pop();
} // mutex zwolniony tutaj
std::cout << "Processing " << x << "\n"; // juz bez mutexa
}
Druga wersja jest lepsza, bo producer może nadal wkładać nowe zadania, podczas gdy consumer przetwarza stare. Mówiąc obrazowo: „szybko wziąłem zamówienie przy kasie, odszedłem przygotować kawę, nie blokując kolejki”.
Minimalna kolejka zadań: projektujemy TaskQueue
Teraz zbudujemy mały element, z którego łatwo postawić workera. Naszym celem jest klasa, która enkapsuluje trzy rzeczy: std::queue, std::mutex, std::condition_variable. Dzięki temu zasady protokołu nie rozmazują się po kodzie, lecz są w jednym miejscu. To nie „OOP dla OOP”, a sposób, by nie zapomnieć, gdzie są jakie mutexy i kto co komu musi zrobić.
Zaczniemy od modelu zadania. Dla przykładu zadanie niech będzie id i tekst, który worker „wykona”, wypisując go:
#include <string>
struct Task {
int id = 0;
std::string text;
};
Teraz szkic kolejki:
#include <condition_variable>
#include <mutex>
#include <queue>
class TaskQueue {
public:
void push(Task t);
Task pop_blocking();
private:
std::mutex m_;
std::condition_variable cv_;
std::queue<Task> q_;
};
Zaimplementujemy push. Zmieniamy stan (q_) pod mutex, a następnie powiadamiamy oczekującego consumera. Powiadomienie zwykle robi się po wyjściu z zakresu lock_guard, żeby nie budzić wątku „do środka mutexu”, ale w prostych scenariuszach powiadomienie pod mutexem nie łamie poprawności — jedynie może być mniej wydajne.
#include <utility> // std::move
void TaskQueue::push(Task t) {
{
std::lock_guard<std::mutex> lock(m_);
q_.push(std::move(t));
}
cv_.notify_one();
}
Teraz pop_blocking. Potrzebujemy std::unique_lock, bo condition_variable::wait musi umieć tymczasowo zwolnić mutex i ponownie go zająć po przebudzeniu. lock_guard tego nie potrafi — on „zajmuje i trzyma do końca zakresu”, bez możliwości.
Task TaskQueue::pop_blocking() {
std::unique_lock<std::mutex> lock(m_);
cv_.wait(lock, [&] {
return !q_.empty();
});
Task t = std::move(q_.front());
q_.pop();
return t;
}
Zwróć uwagę: zwracamy Task przez wartość. To wygodne i bezpieczne: pobraliśmy zadanie z kolejki pod mutexem, dalej worker operuje na własnej kopii/przesuniętym obiekcie. Nie zwracamy referencji do elementu kolejki, bo element znika po pop(), a „referencja do znikniętego” to kiepski prezent.
Worker: jak wygląda pętla consumera
Wątek-worker zwykle żyje w nieskończonej pętli: „czekam na zadanie → pobieram → wykonuję”. To podobne do kasjera, który siedzi i czeka na następnego klienta: nie biega co 10 ms z pytaniem „czy już?”, tylko spokojnie czeka. W kodzie rolę „spokojnego czekania” pełni condition_variable::wait.
Zrobimy funkcję workera. Aby nie projektować teraz pełnego protokołu zakończenia (to osobny temat), zastosujemy ograniczenie edukacyjne: worker przetworzy dokładnie N zadań i zakończy działanie. To nie „produkcja”, ale uczciwy sposób zbudować model bez zawieszeń.
#include <chrono>
#include <iostream>
#include <thread>
void worker_loop(TaskQueue& queue, int tasks_to_process) {
for (int i = 0; i < tasks_to_process; ++i) {
Task t = queue.pop_blocking(); // czekamy, az pojawi sie zadanie
// przetwarzanie poza kolejka: symulujemy prace
std::cout << "[worker] task #" << t.id << ": " << t.text << "\n";
// przyklad wyjscia: [worker] task #1: say hello
std::this_thread::sleep_for(std::chrono::milliseconds{50});
}
std::cout << "[worker] done\n"; // [worker] done
}
Ważne, że pop_blocking() blokuje wątek bez busy-wait. Worker nie „kręci się”, nie zjada CPU, tylko śpi, dopóki producer nie doda zadania i nie wywoła notify_one().
Producer: uruchamiamy workera i wrzucamy zadania do kolejki
Teraz złożymy małą aplikację konsolową. Niech wątek główny będzie producerem: tworzy zadania i wkłada je do kolejki. Drugi wątek — consumer: pobiera i wypisuje. To bardzo podobne do realnych scenariuszy: wątek główny „generuje zdarzenia”, worker je „przetwarza”.
Minimalne main() wygląda tak:
#include <iostream>
#include <thread>
#include <vector>
int main() {
TaskQueue queue;
const std::vector<Task> tasks = {
{1, "say hello"},
{2, "compute something (pretend)"},
{3, "write log line"},
{4, "send email (pretend)"},
{5, "cleanup"},
};
std::thread worker(worker_loop, std::ref(queue), (int)tasks.size());
for (const auto& t : tasks) {
std::cout << "[main] push task #" << t.id << "\n";
// przyklad wyjscia: [main] push task #1
queue.push(t);
}
worker.join();
std::cout << "[main] program finished\n"; // [main] program finished
}
Kilka detali do omówienia (bez krzyku). std::ref(queue) jest potrzebne, by przekazać kolejkę przez odniesienie; inaczej std::thread spróbuje skopiować queue, a nasz TaskQueue nie jest kopiowalny (zawiera mutex, a mutex — niekopiowalny).
Zwróć też uwagę na porządek: najpierw uruchamiamy workera, potem dodajemy zadania. Jeśli zrobisz odwrotnie, nic „logicznie” się nie zepsuje, ale sens demonstracji trochę ucierpi: worker i tak sobie poradzi, bo pop_blocking() poprawnie czeka na stan. To ważna cecha dobrego protokołu: powinien być poprawny niezależnie od tego, kto „zdążył pierwszy”.
Gdzie dokładnie trzymany jest mutex: „rentgen” protokołu
Kiedy zaczynasz, wielowątkowość wydaje się czarną skrzynką: „on czeka, potem się budzi, potem coś tam…”. Aby poczuć się pewniej, warto rozłożyć jedną iterację consumera na kroki i wyraźnie oznaczyć, gdzie trzymany jest mutex, a gdzie nie. Wtedy condition_variable przestaje wyglądać jak magia i staje się zwykłym protokołem.
Przedstawmy sekwencję zdarzeń przy jednym push i jednym pop:
sequenceDiagram
participant P as Producer (main)
participant M as Mutex
participant Q as Queue
participant CV as condition_variable
participant C as Consumer (worker)
C->>M: lock (unique_lock)
C->>CV: wait(lock, pred)<br/>mutex jest zwolniony wewnatrz wait
Note over C,CV: Watek worker spi
P->>M: lock (lock_guard)
P->>Q: push(task)
P->>M: unlock
P->>CV: notify_one()
CV-->>C: wake up
C->>M: lock (wraca z wait z zajetym mutexem)
C->>Q: pop(task)
C->>M: unlock
C->>C: process task (bez mutexa)
Jeśli streścić to w jednym zdaniu, dostaniemy niemal zen: pod mutexem czytamy i zmieniamy stan, a wait — to „zasnąć, tymczasowo zwolnić mutex i obudzić się z mutexem”.
Bardzo częsty błąd — przetwarzanie zadania pod mutexem. Diagnostyka prosta: jeśli widzisz, że worker trzyma mutex i jednocześnie robi sleep_for, wypisuje duże porcje danych, lub, o zgrozo, wykonuje operacje sieciowe, to znaczy, że kolejka stała się wąskim gardłem.
3. Typowe błędy
Błąd nr 1: „Sprawdzę empty() bez mutexa, to tylko podglad”.
Tak zwykle zaczynają się historie o rzadkich, trudnych do złapania błądach. W wielowątkowości „tylko podgląd” to już udział w wyścigu, jeśli inny wątek może jednocześnie modyfikować kolejkę. Prawidłowa dyscyplina: każde empty()/front()/pop()/push() wykonujemy pod tym samym mutex, bez wyjątków.
Błąd nr 2: consumer robi front()/pop() bez gwarancji !empty().
Nawet jeśli wydaje się, że „producer już na pewno włożył”, rzeczywisty scheduler nie podpisał kontraktu „działaj według twojej wygody”. Consumer musi czekać na stan „kolejka niepusta” i dopiero wtedy pobierać element. W poprawnym modelu wyraża się to przez cv.wait(lock, predicate) albo równoważną pętlę.
Błąd nr 3: trzymanie mutexa podczas przetwarzania zadania.
To zabija równoległość. W efekcie producer stoi w kolejce za workerem jak w urzędzie. Właściwy wzorzec: pod mutexem tylko „pobrać zadanie z kolejki”, potem zwolnić mutex i dopiero wtedy przetwarzać. Jeśli przetwarzanie jest długie — tym bardziej musi być na zewnątrz.
Błąd nr 4: wywołanie notify_one() „gdzieś później” albo zapomnienie o nim.
Jeśli producer zmienia stan tak, że predykat oczekiwania staje się prawdziwy (np. kolejka przestaje być pusta), ale nie wykonuje notify_*, consumer może zbyt długo spać. W dobrej implementacji push zawsze kończy się powiadomieniem. To nie „optymalizacja”, to element protokołu.
Błąd nr 5: zwracanie na zewnątrz referencji/wskaźnika do elementu kolejki.
Kolejka to kontener, a element w kontenerze żyje tylko dopóki go nie usuniesz. Jeśli pobrałeś front(), potem pop(), a na zewnątrz gdzieś trzymasz referencję do „tego frontu”, otrzymasz referencję do już nieistniejącego obiektu. Solidna praktyka: pobrać wartość (kopią lub std::move) do lokalnej zmiennej pod mutexem, a dalej pracować z tą lokalną zmienną.
GO TO FULL VERSION