Chapter std::async menutup rantai future dengan satu catatan penting: setiap panggilan ber-policy launch::async berpotensi membuka thread baru, dan thread itu dibongkar lagi setelah hasilnya diambil. Untuk lima tugas satuan, biaya tersebut masih wajar; untuk ratusan tugas kecil yang datang terus-menerus, membuka-menutup thread berulang kali menjadi pemborosan yang terasa. Pola thread pool menjawabnya dengan keputusan sederhana: buka sekelompok pekerja sekali di awal, taruh setiap tugas ke sebuah antrian, dan biarkan pekerja mengambilnya sendiri selama program hidup. Seluruh bahannya sudah kita pegang — mutex dan lock_guard dari chapter std::mutex, condition_variable dari chapter Condition Variable, serta std::future dari dua chapter sebelumnya — sehingga kita bisa langsung merakit pool mini yang benar-benar diuji di bawah ini.

Anatomi Thread Pool #

Thread pool tersusun dari tiga komponen yang saling menopang:

  • Antrian tugas — sebuah std::queue yang tiap elemennya callable tanpa argumen. Antrian ini ditulis dan diambil dari thread yang berbeda, sehingga seluruh aksesnya dilindungi mutex.
  • Worker loop — fungsi yang dijalankan setiap thread pekerja sepanjang hidupnya: tidur menunggu di condition variable, mengambil satu tugas, mengeksekusinya, lalu kembali tidur.
  • Shutdown bersih — satu flag berhenti yang ikut dibaca predicate wait(). Saat flag diaktifkan, pekerja menuntaskan sisa antrian lalu keluar dari loop, dan pemilik pool melakukan join() terhadap semua pekerja.

Pekerja tidak perlu tahu siapa pemesan tugas dan apa makna hasilnya — pekerja hanya mengeksekusi callable yang diambilnya. Konsekuensinya, elemen antrian kita bertipe std::move_only_function<void()>: wrapper callable dari C++23 yang boleh dipindah tetapi tidak harus bisa disalin. Pilihan ini penting karena tugas yang akan kita antriarkan membawa std::packaged_task, dan objek tersebut move-only.

Program Utama: Pool Mini #

Program berikut merakit ketiga komponen di atas menjadi satu class ThreadPool, mengirim enam tugas kuadrat, lalu mencetak hasilnya dalam urutan pengiriman:

cpp 
#include <condition_variable>
#include <functional>
#include <future>
#include <iostream>
#include <mutex>
#include <queue>
#include <string>
#include <thread>
#include <utility>
#include <vector>

// Thread pool mini: sekelompok pekerja mengambil tugas dari satu antrian.
class ThreadPool {
public:
    explicit ThreadPool(unsigned jumlahPekerja) {
        for (unsigned i = 0; i < jumlahPekerja; ++i)
            pekerja_.emplace_back([this] { loopPekerja(); });
    }

    ~ThreadPool() {
        {   // sinyal berhenti; antrian yang tersisa tetap dituntaskan
            std::lock_guard<std::mutex> kunci(mutex_);
            berhenti_ = true;
        }
        cv_.notify_all();
        for (std::thread& p : pekerja_) p.join();
    }

    // Kirim tugas ke antrian; hasilnya diambil lewat std::future.
    template <typename F>
    auto jalankan(F tugas) -> std::future<decltype(tugas())> {
        using Hasil = decltype(tugas());
        std::packaged_task<Hasil()> paket(std::move(tugas));
        std::future<Hasil> hasil = paket.get_future();
        {
            std::lock_guard<std::mutex> kunci(mutex_);
            antrian_.push([paket = std::move(paket)] mutable { paket(); });
        }
        cv_.notify_one();
        return hasil;
    }

private:
    // Worker loop: tidur di condition_variable sampai ada tugas atau berhenti.
    void loopPekerja() {
        for (;;) {
            std::move_only_function<void()> tugas;
            {
                std::unique_lock<std::mutex> kunci(mutex_);
                cv_.wait(kunci, [this] { return berhenti_ || !antrian_.empty(); });
                if (berhenti_ && antrian_.empty()) return;  // shutdown bersih
                tugas = std::move(antrian_.front());
                antrian_.pop();
            }
            tugas();  // tugas dieksekusi DI LUAR lock
        }
    }

    std::vector<std::thread> pekerja_;
    std::queue<std::move_only_function<void()>> antrian_;
    std::mutex mutex_;
    std::condition_variable cv_;
    bool berhenti_ = false;
};

int main() {
    std::vector<std::string> hasil;
    {
        ThreadPool pool(3);
        std::vector<std::future<std::string>> tiket;
        for (int id = 1; id <= 6; ++id) {
            tiket.push_back(pool.jalankan([id] -> std::string {
                return "tugas #" + std::to_string(id)
                     + " -> kuadrat " + std::to_string(id * id);
            }));
        }
        for (std::future<std::string>& t : tiket) hasil.push_back(t.get());
    }  // destruktor pool: sinyal berhenti + join semua pekerja

    std::cout << "hasil tugas (urutan pengiriman):\n";
    for (const std::string& baris : hasil) std::cout << baris << "\n";
    std::cout << "pool dimatikan: semua pekerja sudah di-join\n";
    return 0;
}

Output yang Dihasilkan:

text 
hasil tugas (urutan pengiriman):
tugas #1 -> kuadrat 1
tugas #2 -> kuadrat 4
tugas #3 -> kuadrat 9
tugas #4 -> kuadrat 16
tugas #5 -> kuadrat 25
tugas #6 -> kuadrat 36
pool dimatikan: semua pekerja sudah di-join

Keluaran selalu sama di setiap eksekusi — bukan kebetulan, melainkan desain yang akan kita bedah di akhir bagian ini. Mari kita telaah baris demi baris:

  1. Baris 15-17 (constructor): constructor langsung membuka jumlahPekerja thread, dan setiap pekerja menjalankan loopPekerja() milik objek yang sama karena lambda menangkap this. Angka 3 di main() kita buat tetap agar keluaran bisa dijamin — pembahasannya ada di bagian ukuran pool.
  2. Baris 31-34 (menyiapkan packaged_task): decltype(tugas()) meminta compiler menyimpulkan tipe hasil callable, sehingga jalankan() otomatis mengembalikan std::future yang tepat. std::packaged_task<Hasil()> membungkus tugasnya, dan get_future() memberi pemanggil tiket menuju shared state tempat hasil akan disimpan.
  3. Baris 35-39 (masuk antrian lalu notify): lambda di baris 37 membawa packaged_task ke dalam antrian; kata mutable wajib karena operator() milik packaged_task tidak const. Pengiriman dilakukan di bawah lock_guard, dan notify_one() pada baris 39 membangunkan satu pekerja yang sedang tidur.
  4. Baris 50 (predicate wait): pekerja tidur sampai berhenti_ || !antrian_.empty() bernilai true. Predicate menangkal lost wakeup — sinyal yang datang sebelum pekerja sempat tidur tetap terlihat saat wait() memeriksa predicate-nya.
  5. Baris 51-54 (mengambil satu tugas): bila berhenti_ sudah aktif DAN antrian kosong, pekerja keluar dari loop. Selain itu, tugas terdepan dipindah keluar antrian — masih di dalam lock, karena front() dan pop() harus selesai tanpa diselingi pekerja lain.
  6. Baris 55 (eksekusi di luar lock): scope lock sudah ditutup sebelum tugas() dipanggil, sehingga tugas yang lama selesainya tidak membuat pekerja lain menunggu mutex. Tanpa pemisahan ini, pool kehilangan paralelismenya dan berubah menjadi eksekutor sekuensial.
  7. Baris 20-27 (destruktor: shutdown bersih): flag berhenti_ diaktifkan di bawah lock, notify_all() membangunkan semua pekerja tidur, lalu join() menunggu masing-masing benar-benar berakhir. Predicate wait() menjamin pekerja yang masih menyisakan antrian menuntaskannya dulu, sehingga tidak ada tugas yang terlewat saat pool dibongkar.
  8. Baris 66-82 (kumpulkan dulu, cetak belakangan): get() pada baris 77 memblokir sampai tugas tiket itu selesai, dan tiket-tiket disimpan di vektor sesuai urutan pengiriman. Pencetakan baru berjalan setelah scope ditutup — saat semua tugas selesai dan semua pekerja sudah di-join.

Ukuran Pool dan hardware_concurrency #

Berapa pekerja yang ideal? Standar menyediakan titik awal berupa std::thread::hardware_concurrency() dari header <thread> — fungsi yang melaporkan perkiraan jumlah thread yang bisa dijalankan benar-benar bersamaan oleh implementasi. Di lingkungan praktikum ini fungsi tersebut melaporkan 2; di komputer pembaca nilainya bisa 4, 8, atau angka lain. Nilai yang perlu diwaspadai adalah 0: standar menyatakan 0 berarti informasi tidak tersedia, sehingga pool harus punya nilai cadangan sebelum dipakai.

cpp Idiom memilih ukuran pool 
unsigned nInti = std::thread::hardware_concurrency();  // bisa 0 bila tidak diketahui
unsigned ukuran = (nInti == 0) ? 2u : nInti;

Angka yang sehat bergantung pada bentuk pekerjaannya. Tugas yang sepanjang hidupnya menghabiskan CPU — kompresi, encoding, perhitungan numerik — bekerja paling efisien saat jumlah pekerja sekitar jumlah inti; lebih dari itu hanya menambah antrean penjadwal. Tugas yang sering tidur menunggu jaringan atau disk boleh diberi pekerja lebih banyak, karena pekerja yang menunggu tidak memerebutkan CPU. Pool 3 pekerja pada program utama di mesin 2 inti berarti sedikit oversubscription — untuk tugas sekecil kuadrat, perbedaannya tidak terasa, dan program tetap benar.

Thread Pool dan std::future #

jalankan() kita adalah jembatan antara dua dunia yang sudah kita kenal: sisi antrian berbicara dalam callable tanpa argumen, sisi pemanggil berbicara dalam std::future. Jembatannya adalah std::packaged_task — mekanisme yang sama dengan std::async pada chapter lalu: hasil maupun exception tugas disimpan di shared state, dan get() mengembalikan nilai atau melempar ulang exception-nya ke pemanggil. Karena itu pool kita gratis mendapat pengantaran exception: ganti salah satu tugas dengan versi yang melempar std::runtime_error, dan pemanggil cukup menangkapnya di sekitar get() persis seperti pada program std::async.

Bedanya terletak pada siapa yang membuka thread. std::async dengan policy async menitipkan keputusan itu ke implementasi dan berpotensi membuka thread baru untuk tiap panggilan; pool kita membuka thread tepat sebanyak yang kita minta, lalu memakainya ulang untuk semua tugas. Kepemilikan thread juga jelas: ThreadPool adalah objek RAII yang destructor-nya melakukan join(), sehingga lupa menutup pool tidak berujung pada std::terminate seperti kasus std::thread yang dibiarkan tanpa join pada chapter Siklus Hidup Thread.

Satu batasan pool sederhana ini perlu dicatat: pekerja hanya bisa berhenti di antara tugas. Sinyal berhenti_ baru diperiksa saat pekerja kembali ke wait(), sehingga tugas yang berjalan panjang tetap harus diselesaikan sampai tuntas. C++20 menyediakan jawaban untuk penghentian di tengah jalan — std::jthread dengan std::stop_token yang bisa diteruskan ke dalam tugas — dan mekanisme itulah topik chapter berikutnya.

Referensi: cppreference — Thread support library , cppreference — std::move_only_function , cppreference — std::thread::hardware_concurrency .