Process Pools

Membuat multiprocessing.Process satu per satu cepat menjadi melelahkan: untuk 100 tugas kita harus mengatur 100 object, 100 start(), dan 100 join() — belum lagi pembagian data dan pengumpulan hasil. multiprocessing.Pool menyelesaikan seluruh urusan itu: sekali kita buat, pool menyediakan sekelompok worker siap pakai yang membagi tugas secara otomatis. Pada Chapter ini kita akan memakai map, apply_async, imap_unordered, dan mempelajari disiplin close()/join().

Pool dan map: Membagi Tugas Otomatis #

Cara kerja pool sederhana: satu koordinator membagikan tugas kepada sejumlah worker, dan worker yang menganggur langsung mengambil tugas berikutnya:

from multiprocessing import Pool

def kuadrat(n):
    return n * n

if __name__ == "__main__":
    pool = Pool(processes=2)
    try:
        hasil = pool.map(kuadrat, range(10))
    finally:
        pool.close()
        pool.join()
    print(hasil)

Output yang Dihasilkan:

[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

Mari kita telaah apa yang terjadi pada baris-baris kode di atas:

1. Baris 7 (pool = Pool(processes=2)): dua process worker dibuat dan siap menerima tugas. Ukuran pool yang lazim adalah os.cpu_count(); lebih banyak worker daripada jumlah core hanya menambah biaya tanpa menambah kecepatan.
2. Baris 9 (pool.map(kuadrat, range(10))): map membagi range(10) kepada kedua worker, memanggil kuadrat untuk setiap elemen di process terpisah, lalu menunggu semua selesai dan mengumpulkan hasilnya dalam urutan asal input — sama seperti builtin map, hanya dieksekusi paralel.
3. Baris 10–11 (pool.close() dan pool.join()): close() menyatakan tidak ada tugas baru lagi; join() menunggu semua worker menyelesaikan tugasnya. Blok finally memastikan dua baris tersebut tetap berjalan meski terjadi exception.

Urutan hasil map selalu mengikuti urutan input, terlepas dari worker mana yang lebih dulu selesai. Inilah map pilihan utama untuk pola “satu tugas per elemen, kumpulkan semuanya”.

apply_async: Satu Tugas Tanpa Menunggu #

pool.map menunggu seluruh pekerjaan tuntas sebelum baris berikutnya berjalan. apply_async mengirim satu tugas dan langsung mengembalikan penampung hasil bernama AsyncResult; hasil diambil nanti lewat .get():

from multiprocessing import Pool

def kuadrat(n):
    return n * n

if __name__ == "__main__":
    pool = Pool(processes=2)
    try:
        tugas_a = pool.apply_async(kuadrat, (7,))
        tugas_b = pool.apply_async(kuadrat, (9,))
        print(f"Hasil A: {tugas_a.get()}")
        print(f"Hasil B: {tugas_b.get()}")
    finally:
        pool.close()
        pool.join()

Output yang Dihasilkan:

Hasil A: 49
Hasil B: 81

Antara apply_async dan .get(), program utama bebas melakukan hal lain — misalnya mengirim tugas berikutnya, seperti baris tugas_b di atas. Pemanggilan .get() sendiri bersifat blocking: baris print pertama menunggu hasil 49 tersedia baru mencetak. Jika pekerjaan bisa gagal, .get() juga menimbulkan kembali exception yang terjadi di worker, sehingga pemanggilan ini tempat yang tepat untuk blok try/except.

imap_unordered dan chunksize #

map menunggu semua selesai; imap_unordered mengalirkan setiap hasil begitu tersedia, dalam urutan penyelesaian — bukan urutan input. Demo berikut sengaja memberi tiap tugas durasi berbeda:

from multiprocessing import Pool
import time

def kuadrat(n):
    time.sleep((9 - n) * 0.02)  # simulasi tugas dengan durasi berbeda
    return n * n

if __name__ == "__main__":
    pool = Pool(processes=2)
    try:
        hasil = list(pool.imap_unordered(kuadrat, range(10), chunksize=1))
        print(f"Urutan asal : {hasil}")
        print(f"Urutan sort : {sorted(hasil)}")
    finally:
        pool.close()
        pool.join()

Output yang Dihasilkan (urutan asal dapat berbeda pada setiap eksekusi):

Urutan asal : [1, 0, 9, 4, 25, 16, 49, 36, 81, 64]
Urutan sort : [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

Kumpulan hasilnya sama, tetapi urutan kedatangannya mengikuti kecepatan penyelesaian: tugas n = 9 (tidur paling singkat) tiba lebih awal daripada n = 0. Pola ini pas untuk pemrosesan streaming — setiap hasil bisa langsung ditulis atau ditampilkan tanpa menunggu kawan-kawannya.

Argument chunksize=1 mengatur berapa elemen dikirim sekaligus ke satu worker:

Nilai default dipilih otomatis oleh Python dan sudah cukup baik untuk sebagian besar kasus; anggap chunksize sebagai tuas penyetel halus, bukan sesuatu yang wajib diatur.

Pool atau Process Manual #

Aturan praktisnya: jika tugas-tugas kita berbentuk “olah setiap elemen dari kumpulan data”, pool hampir selalu menang. Jika hanya ada dua-tiga pekerjaan besar dengan bentuk berbeda, Process manual lebih jelas. Dan untuk API yang lebih modern dengan penanganan exception serta context manager yang rapi, Chapter concurrent.futures memperkenalkan lapisan tingkat tinggi di atas kedua pola tersebut.