concurrent.futures

Modul concurrent.futures menyediakan antarmuka tingkat tinggi untuk menjalankan tugas secara bersamaan di atas thread maupun proses. Objek executor mengurus pembuatan kumpulan pekerja, distribusi tugas, dan pengumpulan hasilnya, sedangkan objek future merepresentasikan hasil yang belum tersedia. Topik ini membahas map, submit dengan Future, pola as_completed, ProcessPoolExecutor, beserta kesalahan umum dalam pemakaiannya.

Executor sebagai antarmuka tunggal#

Modul ini berdiri di atas dua lapisan yang lebih rendah, yaitu threading dan multiprocessing, dan menyatukan keduanya di balik antarmuka yang sama. Pergantian model konkurensi cukup dilakukan dengan mengganti kelas executor, tanpa mengubah cara tugas diserahkan dan hasilnya diambil.

Kelas ThreadPoolExecutor menjalankan tugas pada kumpulan thread. Metode map menerima fungsi dan daftar masukan, mendistribusikannya ke pekerja, lalu mengembalikan hasil dalam urutan masukan yang sama.

from concurrent.futures import ThreadPoolExecutor
import time

def muat(id_halaman):
    time.sleep(0.02)
    return f"isi {id_halaman}"

with ThreadPoolExecutor(max_workers=3) as executor:
    hasil = executor.map(muat, [1, 2, 3, 4, 5, 6])
    for item in hasil:
        print(item)

isi 1
isi 2
isi 3
isi 4
isi 5
isi 6

Enam tugas tersebut dijalankan tiga pekerja secara tumpang-tindih, tetapi hasilnya tetap dicetak berurutan karena map menjaga urutan masukan. Perintah with menutup executor pada akhir blok, yaitu menunggu seluruh tugas selesai dan berhenti menerima tugas baru. Pekerjaan yang menunggu seperti time.sleep di contoh ini mewakili permintaan jaringan atau pembacaan berkas pada program nyata.

submit dan objek Future#

Metode submit menerima satu tugas dan segera mengembalikan objek Future tanpa menunggu tugasnya selesai. Objek tersebut mewakili hasil yang akan datang, dan metode result() menunggu sampai nilainya tersedia. Karena submit segera kembali, seluruh tugas dapat diserahkan lebih dahulu, lalu hasilnya dikumpulkan setelah itu.

from concurrent.futures import ThreadPoolExecutor
import time

def kuadrat(n):
    time.sleep(0.01 * (6 - n))
    return n * n

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = [executor.submit(kuadrat, n) for n in (1, 2, 3, 4, 5)]
    hasil = [f.result() for f in futures]
print(hasil)

[1, 4, 9, 16, 25]

List futures dibuat dalam satu putaran loop terlebih dahulu sehingga lima tugas terserahkan hampir bersamaan, dan baru kemudian result() dipanggil satu per satu. Keluarannya tetap [1, 4, 9, 16, 25] meskipun tugas dengan masukan kecil selesai lebih dulu, karena pengumpulan mengikuti urutan list, bukan urutan penyelesaian. Selain nilai, Future juga menyimpan error tugasnya, yang dilempar ulang oleh result() pada saat dipanggil.

as_completed dan urutan penyelesaian#

Fungsi as_completed menghasilkan objek Future satu per satu sesuai urutan penyelesaiannya, bukan urutan pengirimannya. Urutan itu berubah-ubah antar eksekusi, sehingga hasilnya tidak boleh dicetak mentah bila urutannya bermakna. Pola yang aman adalah menggunakannya untuk perhitungan yang tidak bergantung pada urutan, seperti akumulasi.

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def proses(n):
    time.sleep(0.01)
    return n * 10

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = {executor.submit(proses, n): n for n in range(1, 7)}
    total = 0
    for future in as_completed(futures):
        total += future.result()
    print(total)

210

Enam hasil yang datang dengan urutan acak dijumlahkan menjadi 210 setiap kali, karena penjumlahan tidak terpengaruh urutan. Apabila pasangan masukan dan hasil diperlukan, dictionary futures di contoh menyediakan peta dari objek Future menuju masukannya. Untuk hasil yang harus berurutan, map tetap menjadi pilihan yang paling ringkas.

ProcessPoolExecutor untuk beban CPU#

ProcessPoolExecutor memakai antarmuka yang sama persis, tetapi pekerjanya berupa proses terpisah yang masing-masing melewati GIL. Fungsi beserta argumennya dikirim ke proses pekerja melalui pickling, sehingga pembuatan prosesnya wajib berada di dalam pengaman __main__ sebagaimana modul multiprocessing.

from concurrent.futures import ProcessPoolExecutor

def jumlah_besar(n):
    return sum(range(n))

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=4) as executor:
        hasil = executor.map(jumlah_besar, [1_000_000, 2_000_000, 3_000_000])
    print(list(hasil))

[499999500000, 1999999000000, 4499998500000]

Tiga penjumlahan berukuran jutaan iterasi dijalankan empat proses secara paralel, dan hasilnya kembali berurutan tanpa pengurutan ulang. Pergantian dari versi thread hanya mengubah nama kelasnya, sedangkan kode tugas dan pengumpulan hasilnya tetap sama. Kesamaan inilah yang membuat modul ini menjadi pintu masuk yang disarankan sebelum turun ke threading atau multiprocessing secara langsung.

Kesalahan umum#

Kesalahan pertama adalah memanggil result() segera setelah setiap submit di dalam loop yang sama. Pemanggilan result() menunggu tugasnya selesai, sehingga tugas berikutnya baru diserahkan setelah tugas sebelumnya tuntas dan seluruh eksekusi berjalan berurutan.

import time
from concurrent.futures import ThreadPoolExecutor

def kerja():
    time.sleep(0.2)
    return 1

with ThreadPoolExecutor(max_workers=3) as executor:
    awal = time.perf_counter()
    hasil = [executor.submit(kerja).result() for _ in range(3)]
    durasi_hasil_langsung = time.perf_counter() - awal

    awal = time.perf_counter()
    futures = [executor.submit(kerja) for _ in range(3)]
    hasil = [f.result() for f in futures]
    durasi_dikumpulkan = time.perf_counter() - awal

print(hasil)
print(durasi_hasil_langsung > durasi_dikumpulkan)

[1, 1, 1]
True

Pola pertama memakan waktu sekitar tiga kali lipat pola kedua karena tugasnya berjalan satu per satu, sedangkan pola kedua menyerahkan ketiganya lebih dahulu sehingga selesai tumpang-tindih. Nilai True pada baris terakhir menegaskan perbedaan durasi tersebut pada setiap eksekusi. Perbaikannya selalu sama, yaitu memisahkan penyerahan tugas dari pengambilan hasilnya.

Kesalahan kedua adalah memakai executor setelah blok with-nya ditutup. Executor yang sudah dimatikan tidak lagi menerima tugas dan menolaknya dengan error yang jelas.

from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=2)
with executor:
    pass

try:
    executor.submit(lambda: None)
except RuntimeError as e:
    print(type(e).__name__, str(e))

RuntimeError cannot schedule new futures after shutdown

Pesan cannot schedule new futures after shutdown menunjukkan bahwa siklus hidup executor sudah berakhir. Seluruh tugas harus diserahkan di dalam blok with, atau pemanggilan shutdown() harus ditunda sampai semua tugas selesai diserahkan.

Kesalahan ketiga adalah mengirim fungsi lokal atau lambda ke ProcessPoolExecutor. Fungsi harus dapat dipickle agar bisa dikirim ke proses pekerja, sedangkan fungsi yang didefinisikan di dalam fungsi lain tidak memenuhi syarat itu.

from concurrent.futures import ProcessPoolExecutor

def pabrik():
    def kuadrat_lokal(n):
        return n * n
    return kuadrat_lokal

if __name__ == "__main__":
    fungsi = pabrik()
    with ProcessPoolExecutor(max_workers=1) as executor:
        future = executor.submit(fungsi, 4)
        try:
            print(future.result())
        except Exception as e:
            print(type(e).__name__, str(e))

AttributeError Can't get local object 'pabrik.<locals>.kuadrat_lokal'

Error muncul pada saat hasil diambil, karena pengiriman tugas ke proses pekerja berlangsung di latar belakang. Perbaikannya adalah mendefinisikan fungsi tugas pada tingkat modul, sehingga namanya dapat ditemukan kembali oleh mekanisme pickling. Lambda dan fungsi bersarang memiliki batasan yang sama.

Ringkasan#

• concurrent.futures menyatukan thread dan proses di balik antarmuka executor yang sama, sehingga pergantian model cukup dilakukan dengan mengganti kelasnya.
• executor.map menjaga urutan hasil sesuai urutan masukan, sedangkan submit mengembalikan Future yang nilainya diambil melalui result().
• as_completed menghasilkan future sesuai urutan penyelesaian yang berubah-ubah, sehingga hanya cocok untuk pemrosesan yang tidak tergantung urutan.
• ProcessPoolExecutor melewati GIL untuk beban CPU, dengan syarat fungsi dan argumennya dapat dipickle serta pembuatannya berada di dalam pengaman __main__.

Referensi#

• concurrent.futures — Launching parallel tasks — Python Software Foundation — Dokumentasi resmi modul beserta Executor, Future, dan kedua turunannya.
• Future Objects — Python Software Foundation — Rincian metode result, done, dan exception pada objek Future.
• ThreadPoolExecutor — Python Software Foundation — Referensi kelas ThreadPoolExecutor beserta aturan max_workers dan inisialisasinya.