concurrent.futures

Baik threading maupun multiprocessing.Pool memaksa kita mengurus detail rendah: membuat object satu per satu, menjaga disiplin close()/join(), dan mengatur pengumpulan hasil secara manual. Module concurrent.futures (hadir sejak Python 3.2) menyembunyikan seluruh kerumitan itu di balik satu API tingkat tinggi: kirim pekerjaan, terima object Future, ambil hasilnya kapan pun kita butuh. Pada Chapter ini kita akan memakai ThreadPoolExecutor, ProcessPoolExecutor, dan berbagai cara mengelola Future.

Executor dan Future #

Executor adalah pengelola pool: ThreadPoolExecutor untuk thread, ProcessPoolExecutor untuk process. Method submit() mengirim satu function beserta argumentnya, lalu langsung mengembalikan object Future — janji tertulis bahwa suatu hasil akan tersedia:

from concurrent.futures import ThreadPoolExecutor
import time

def unduh_berkas(nama, detik):
    time.sleep(detik)
    return f"{nama} selesai diunduh"

with ThreadPoolExecutor(max_workers=2) as executor:
    tugas_pertama = executor.submit(unduh_berkas, "laporan.pdf", 0.3)
    tugas_kedua = executor.submit(unduh_berkas, "foto.jpg", 0.1)
    print(tugas_pertama.result())
    print(tugas_kedua.result())
    print(f"Tugas pertama selesai? {tugas_pertama.done()}")

Output yang Dihasilkan:

laporan.pdf selesai diunduh
foto.jpg selesai diunduh
Tugas pertama selesai? True

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

1. Baris 9 (with ThreadPoolExecutor(max_workers=2) ...): executor dibuat dengan dua thread worker. Ketika blok with selesai, executor otomatis menunggu semua tugas tuntas — kita tidak perlu close()/join() manual seperti pada multiprocessing.Pool.
2. Baris 10–11 (executor.submit(...)): kedua pemanggilan langsung kembali setelah mendaftarkan tugas; unduh_berkas berjalan di thread lain. Object Future yang dikembalikan menampung hasil saat tersedia.
3. Baris 12 (tugas_pertama.result()): result() bersifat blocking — baris ini menunggu hingga unduh_berkas("laporan.pdf", 0.3) selesai, baru mencetak hasilnya. Exception yang terjadi di dalam tugas juga muncul di titik ini, sehingga blok try/except cukup ditulis di sini.
4. Baris 14 (tugas_pertama.done()): method done() memeriksa apakah tugas telah selesai tanpa menunggu. Setelah result() kembali, nilainya pasti True.

as_completed: Memproses Hasil yang Tercepat #

Blok with menunggu semua tugas, tetapi sering kali kita ingin memproses setiap hasil begitu siap. Function as_completed() menerima kumpulan Future dan menghasilkan future yang selesai lebih dulu:

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def unduh_berkas(nama, detik):
    time.sleep(detik)
    return f"{nama} selesai ({detik} detik)"

daftar_unduhan = [
    ("laporan.pdf", 0.3),
    ("foto.jpg", 0.1),
    ("data.csv", 0.2),
]

with ThreadPoolExecutor(max_workers=3) as executor:
    daftar_tugas = {
        executor.submit(unduh_berkas, nama, detik): nama
        for nama, detik in daftar_unduhan
    }
    for tugas in as_completed(daftar_tugas):
        print(tugas.result())

Output yang Dihasilkan:

foto.jpg selesai (0.1 detik)
data.csv selesai (0.2 detik)
laporan.pdf selesai (0.3 detik)

Urutan pencetakan mengikuti urutan penyelesaian, bukan urutan pengiriman: foto.jpg (0,1 detik) tampil lebih dulu daripada laporan.pdf (0,3 detik). Dictionary comprehension memetakan setiap Future ke namanya sehingga kita bisa tahu tugas mana yang baru selesai lewat daftar_tugas[tugas] — pola yang sangat umum untuk progress reporting.

Timeout: Batas Waktu yang Jelas #

Pemanggil result() tanpa argument menunggu tanpa batas waktu. Dengan timeout, kita membatasi penantian dan menangkap TimeoutError bila batas terlampaui:

from concurrent.futures import ThreadPoolExecutor
import time

def tugas_lambat():
    time.sleep(1.0)
    return "Akhirnya selesai"

with ThreadPoolExecutor(max_workers=1) as executor:
    tugas = executor.submit(tugas_lambat)
    try:
        print(tugas.result(timeout=0.2))
    except TimeoutError:
        print("TimeoutError: tugas belum selesai dalam 0.2 detik")
    print(f"Sudah selesai? {tugas.done()}")

Output yang Dihasilkan:

TimeoutError: tugas belum selesai dalam 0.2 detik
Sudah selesai? False

Satu hal penting: TimeoutError tidak membatalkan tugas — tugas_lambat tetap berjalan di belakang, dan blok with akan tetap menunggunya saat keluar. Timeout di sini adalah batas kesabaran pemanggil, bukan perintah penghentian.

executor.map dan ProcessPoolExecutor #

Untuk pola “satu tugas per elemen”, executor.map memanggil function untuk setiap elemen dan menjaga urutan hasil — kebalikan dari imap_unordered yang kita coba pada Chapter Process Pools:

from concurrent.futures import ProcessPoolExecutor

def kuadrat(n):
    return n * n

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=2) as executor:
        hasil = list(executor.map(kuadrat, range(10)))
    print(hasil)

Output yang Dihasilkan:

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

Berbekal ProcessPoolExecutor, pekerjaan CPU-bound dipecah ke beberapa process tanpa kita menyentuh multiprocessing sama sekali — cukup ganti nama class executor, dan seluruh API (submit, map, as_completed) tetap sama. Tentu saja batasan dari Chapter Multiprocessing berlaku: function dan argument harus picklable, dan pelindung if __name__ == "__main__": tetap wajib.

Futures vs multiprocessing.Pool #

Bagi kode baru, concurrent.futures adalah pilihan utama: API-nya seragam untuk thread maupun process, penanganan exception-nya lebih rapi, dan blok with tidak meninggalkan pekerjaan tergantung. multiprocessing.Pool tetap berguna untuk kode lama serta kasus khusus seperti imap_unordered bawaan dan callback per tugas.