Apa itu thread pool?

Thread pool adalah kumpulan worker thread yang dibuat sekali di awal dan tetap hidup, mengambil task dari sebuah antrian satu per satu, alih-alih membuat thread baru untuk setiap task. Ini menggabungkan hampir semua yang sudah dibahas di seri ini: mutex untuk melindungi antrian, condition variable untuk membangunkan worker saat ada task baru, dan struktur data queue.

Ilustrasi

Tanpa thread pool: tiap task bikin thread baru, jalan, lalu destroy
  Task1 -> create -> jalan -> join -> destroy
  Task2 -> create -> jalan -> join -> destroy
  (overhead create/destroy tiap kali)
 
Dengan thread pool: worker tetap hidup, ambil task dari queue
  Task1, Task2, Task3 -> [Queue] -> Worker1, Worker2, Worker3

Kenapa diperlukan

Membuat thread itu tidak gratis — ada overhead alokasi stack, registrasi ke kernel scheduler, dan cleanup resource saat selesai. Kalau aplikasi harus menangani ribuan task kecil, misalnya request HTTP, membuat satu thread baru per task akan menghabiskan banyak waktu hanya untuk overhead create/destroy, bukan untuk kerjaan sebenarnya.

Thread pool menyelesaikan ini dengan membuat worker thread sekali di awal. Mereka hidup terus mengambil task baru dari antrian sampai ada sinyal shutdown.

Implementasi

Struktur proyek

.
├── main.c
├── threadpool.c
└── threadpool.h

Step 1: interface thread pool

#ifndef THREADPOOL_H
#define THREADPOOL_H
 
typedef void (*task_fn)(void *arg);
 
void pool_init(int jumlah_worker);
void pool_submit(task_fn fn, void *arg);
void pool_shutdown(void);
 
#endif

Step 2: pool dengan task queue

Task queue dilindungi mutex, worker menunggu task baru lewat condition variable — pola yang sama seperti buffer producer-consumer sebelumnya, digeneralisasi untuk task apapun.

#include "threadpool.h"
#include <pthread.h>
#include <stdlib.h>
 
typedef struct task {
    task_fn fn;
    void *arg;
    struct task *next;
} task_t;
 
static pthread_t *workers;
static int num_workers;
 
static task_t *head = NULL, *tail = NULL;
static pthread_mutex_t queue_mutex = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t queue_cond = PTHREAD_COND_INITIALIZER;
static int shutdown_flag = 0;
 
static void *worker_loop(void *arg)
{
    while (1) {
        pthread_mutex_lock(&queue_mutex);
 
        while (head == NULL && !shutdown_flag) {
            pthread_cond_wait(&queue_cond, &queue_mutex);
        }
 
        if (shutdown_flag && head == NULL) {
            pthread_mutex_unlock(&queue_mutex);
            break;
        }
 
        task_t *task = head;
        head = head->next;
        if (head == NULL) tail = NULL;
 
        pthread_mutex_unlock(&queue_mutex);
 
        task->fn(task->arg); // dijalankan di luar lock
        free(task);
    }
    return NULL;
}
 
void pool_init(int jumlah_worker)
{
    num_workers = jumlah_worker;
    workers = malloc(sizeof(pthread_t) * jumlah_worker);
    for (int i = 0; i < jumlah_worker; i++) {
        pthread_create(&workers[i], NULL, worker_loop, NULL);
    }
}
 
void pool_submit(task_fn fn, void *arg)
{
    task_t *task = malloc(sizeof(task_t));
    task->fn = fn;
    task->arg = arg;
    task->next = NULL;
 
    pthread_mutex_lock(&queue_mutex);
    if (tail == NULL) {
        head = tail = task;
    } else {
        tail->next = task;
        tail = task;
    }
    pthread_cond_signal(&queue_cond);
    pthread_mutex_unlock(&queue_mutex);
}
 
void pool_shutdown(void)
{
    pthread_mutex_lock(&queue_mutex);
    shutdown_flag = 1;
    pthread_cond_broadcast(&queue_cond); // bangunkan semua worker, bukan cuma satu
    pthread_mutex_unlock(&queue_mutex);
 
    for (int i = 0; i < num_workers; i++) {
        pthread_join(workers[i], NULL);
    }
    free(workers);
}

task->fn(task->arg) dipanggil setelah pthread_mutex_unlock. Kalau task dijalankan sambil masih memegang lock, seluruh pool jadi serial dan kehilangan tujuan paralelisme.

Step 3: kirim beberapa task

#include "threadpool.h"
#include <stdio.h>
#include <unistd.h>
 
void cetak_task(void *arg)
{
    int id = *(int *)arg;
    printf("Task %d mulai dikerjakan\n", id);
    usleep(300000);
    printf("Task %d selesai\n", id);
    free(arg);
}
 
int main(void)
{
    pool_init(3);
 
    for (int i = 0; i < 6; i++) {
        int *id = malloc(sizeof(int));
        *id = i + 1;
        pool_submit(cetak_task, id);
    }
 
    pool_shutdown();
    return 0;
}

Penjelasan

pool_submit membungkus fungsi dan argumennya jadi satu task_t, menambahkannya ke akhir antrian, lalu memanggil pthread_cond_signal untuk membangunkan satu worker yang sedang menunggu.

worker_loop membuat worker tidur, bukan busy-wait, kalau antrian kosong dan belum ada sinyal shutdown. Begitu ada task baru atau shutdown di-set, worker bangun dan cek ulang kondisinya — pola while yang sama seperti condition variable sebelumnya.

pool_shutdown men-set flag shutdown lalu broadcast, bukan signal, supaya semua worker terbangun dan mengecek flag-nya, bukan cuma satu. Setelah itu pthread_join menunggu semua worker benar-benar selesai sebelum program keluar.

Timeline eksekusi

Time Worker1 Worker2 Worker3 Queue
t0 tidur tidur tidur [1,2,3,4,5,6]
t1 ambil task1 ambil task2 ambil task3 [4,5,6]
t2 kerja kerja kerja [4,5,6]
t3 selesai task1, ambil task4 selesai task2, ambil task5 selesai task3, ambil task6 []
t4 kerja kerja kerja []
t5 selesai, tidur (tunggu shutdown) selesai, tidur selesai, tidur []

Kompilasi

gcc -o threadpool main.c threadpool.c -lpthread
./threadpool

Output (urutan antar task bisa bervariasi karena 3 worker paralel):

Task 1 mulai dikerjakan
Task 2 mulai dikerjakan
Task 3 mulai dikerjakan
Task 1 selesai
Task 4 mulai dikerjakan
...

Best practices

Jalankan task di luar lock, karena kalau tidak, seluruh pool jadi serial. Shutdown pakai broadcast, bukan signal, supaya semua worker terbangun untuk mengecek flag. Pertimbangkan bounded queue — implementasi di atas pakai linked list tanpa batas ukuran, dan kalau submit jauh lebih cepat dari worker bisa proses, memory bisa membengkak. Versi produksi biasanya kasih kapasitas maksimum, mirip bounded buffer di post condition variable.

Ini contoh minimal untuk belajar, belum ada penanganan error dari task atau dynamic resizing jumlah worker — jangan langsung dipakai di production tanpa pengembangan lebih lanjut. Jumlah worker idealnya disesuaikan, mendekati jumlah core CPU untuk task CPU-bound, atau lebih banyak untuk task I/O-bound.

Kesimpulan

Thread pool menggabungkan mutex, condition variable, dan queue jadi satu pola yang sangat umum dipakai di sistem nyata, seperti web server dan task scheduler1. Ini juga bukti bagus bahwa primitif-primitif konkurensi yang sudah dibahas satu-satu di seri ini bisa digabung jadi sesuatu yang jauh lebih berguna.

Footnotes

  1. Library thread pool produksi, seperti di banyak runtime bahasa modern, biasanya menambahkan fitur seperti work stealing — worker yang nganggur mencuri task dari worker lain yang antriannya masih panjang — untuk load balancing yang lebih baik. Di luar cakupan implementasi dasar ini. ↩