Dasar-dasar threading

Thread adalah alur eksekusi yang berjalan di dalam satu proses dan berbagi memori dengan thread lain di proses yang sama. Modul threading menyediakan cara membuat thread agar pekerjaan yang menunggu luaran eksternal, seperti permintaan jaringan atau pembacaan berkas, dapat berjalan secara tumpang-tindih. Topik ini membahas pembuatan thread, pengumpulan hasil secara aman, race condition yang muncul ketika memori bersama diubah tanpa pengaman, serta perbaikannya menggunakan lock sebagai kunci bersama.

Membuat dan menjalankan thread#

Kelas threading.Thread menerima fungsi melalui parameter target beserta argumennya melalui args, dan metode start() memulai alur eksekusi baru tanpa menghentikan thread utama. Metode join() menahan thread pemanggil sampai thread yang bersangkutan selesai. Contoh berikut menjalankan satu thread sederhana dan memperlihatkan urutan keluarannya.

import threading
import time

def kerja(indeks):
    time.sleep(0.02)
    print(f"tugas {indeks} selesai")

thread = threading.Thread(target=kerja, args=(1,))
thread.start()
print("utama menunggu")
thread.join()
print("utama lanjut")

utama menunggu
tugas 1 selesai
utama lanjut

Baris utama menunggu dicetak lebih dahulu karena start() segera kembali, sedangkan thread baru membutuhkan waktu 20 milidetik untuk menyelesaikan tugasnya. Metode join() kemudian menahan thread utama hingga cetakan tugas 1 selesai muncul, sehingga baris utama lanjut selalu menjadi cetakan terakhir. Tanpa join(), urutan dua cetakan terakhir tidak dapat dijamin.

Biasanya satu program menjalankan lebih dari satu thread. Karena urutan penyelesaian antar thread tidak dapat dipastikan, hasil pekerjaan dikumpulkan pada struktur data yang diberi indeks, lalu dicetak setelah seluruh thread selesai. Pola ini menjaga keluaran program tetap sama meskipun waktu penyelesaian setiap thread berbeda-beda.

import threading
import time

hasil = [None] * 4

def kerja(indeks):
    time.sleep(0.01 * (4 - indeks))
    hasil[indeks] = indeks * indeks

threads = []
for i in range(4):
    t = threading.Thread(target=kerja, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(hasil)

[0, 1, 4, 9]

Jeda setiap thread sengaja dibuat menurun, sehingga thread dengan indeks terbesar selesai lebih dulu, tetapi hal itu tidak memengaruhi hasil karena setiap thread menulis pada indeksnya sendiri. Daftar hasil tetap terisi pada posisinya masing-masing, dan cetakan akhirnya berurutan. Nilai balik fungsi target tidak digunakan Python, sehingga komunikasi hasil harus melalui struktur data bersama seperti list di atas.

Race condition pada memori bersama#

Semua thread dalam satu proses membaca dan menulis memori yang sama, dan kebebasan inilah sumber race condition, yaitu kondisi ketika hasil akhir bergantung pada urutan antar thread yang tidak dapat dipastikan. Operasi penambahan seperti counter += 1 terlihat sebagai satu langkah, padahal interpreter menjalankannya sebagai tiga langkah, yaitu membaca nilai, menambahkannya, lalu menulis kembali. Apabila dua thread membaca nilai yang sama sebelum keduanya menulis, satu kenaikan hilang.

Diagram berikut memperlihatkan kejadian tersebut pada counter yang dibaca dan ditulis dua thread secara bersisian.

Contoh berikut sengaja membiarkan dua thread menaikkan counter yang sama tanpa pengaman. Potongan ini tidak dijalankan di sini karena keluarannya berubah-ubah antar eksekusi.

# tidak-dijalankan
import threading

counter = 0

def naikkan():
    global counter
    for _ in range(100000):
        counter += 1

t1 = threading.Thread(target=naikkan)
t2 = threading.Thread(target=naikkan)
t1.start()
t2.start()
t1.join()
t2.join()
print(counter)

Nilai yang benar seharusnya 200000, tetapi keluaran nyata program tersebut biasanya lebih kecil dan berbeda pada setiap eksekusi, sehingga percobaan berulang memperlihatkan angka yang berbeda-beda. Pola seperti ini sulit dideteksi saat pengujian karena kegagalannya jarang terjadi dan tidak dapat direproduksi secara pasti. Kode yang tampak benar pada satu mesin dapat gagal pada mesin lain yang penjadwalannya berbeda.

Mengamankan memori bersama dengan Lock#

Objek threading.Lock adalah kunci yang hanya boleh dipegang satu thread pada satu waktu. Metode acquire() mengunci, dan thread lain yang berusaha mengunci akan menunggu sampai pemiliknya memanggil release(). Bentuk with lock lebih disarankan karena pelepasan kunci dijamin berlangsung meskipun terjadi error di tengah blok.

import threading

counter = 0
lock = threading.Lock()

def naikkan():
    global counter
    for _ in range(100000):
        with lock:
            counter += 1

t1 = threading.Thread(target=naikkan)
t2 = threading.Thread(target=naikkan)
t1.start()
t2.start()
t1.join()
t2.join()
print(counter)

200000

Dengan kunci, tiga langkah membaca, menambah, dan menulis tidak dapat diselingi thread lain, sehingga hasilnya selalu 200000 pada setiap eksekusi. Kunci memang menambah biaya waktu karena thread saling menunggu, tetapi biaya tersebut jauh lebih murah daripada hasil perhitungan yang salah. Aturan praktisnya, setiap memori bersama yang dibaca lalu ditulis oleh lebih dari satu thread perlu dilindungi lock.

Kesalahan umum#

Kesalahan pertama adalah memanggil run() dengan harapan thread baru berjalan. Metode run() hanya menjalankan fungsi target pada thread pemanggil, sehingga program berjalan berurutan seperti tanpa thread sama sekali. Metode yang tepat untuk memulai alur eksekusi baru adalah start().

import threading
import time

def kerja():
    time.sleep(0.01)
    print("kerja selesai")

t = threading.Thread(target=kerja)
t.run()
print("baris ini baru muncul setelahnya")

kerja selesai
baris ini baru muncul setelahnya

Cetakan baris ini baru muncul setelahnya baru muncul setelah kerja selesai karena seluruh pekerjaan berlangsung pada thread utama. Apabila start() yang dipanggil, baris ini baru muncul setelahnya justru dicetak lebih dahulu karena thread baru membutuhkan waktu untuk menyelesaikan jedaannya. Pemanggilan join() tetap diperlukan bila urutan cetakan ingin dipertahankan.

Kesalahan kedua adalah membaca hasil bersama sebelum join() dipanggil. Thread utama bergerak jauh lebih cepat daripada thread yang sedang menunggu, sehingga struktur hasil masih kosong ketika dibaca.

import threading
import time

hasil = []

def kerja():
    time.sleep(0.05)
    hasil.append(42)

t = threading.Thread(target=kerja)
t.start()
print(hasil)
t.join()
print(hasil)

[]
[42]

Cetakan pertama menunjukkan daftar yang masih kosong karena tugas tersebut baru selesai 50 milidetik kemudian. Setelah join(), isi daftar sudah lengkap. Pola yang aman adalah mengumpulkan seluruh thread pada sebuah list, memanggil join() untuk semuanya, dan baru menggunakan hasilnya setelah itu.

Ringkasan#

• threading.Thread(target=..., args=...) memulai alur eksekusi baru melalui start(), dan join() menunggu thread tersebut selesai.
• Urutan penyelesaian antar thread tidak dapat dipastikan, sehingga hasil dikumpulkan pada struktur berindeks dan dicetak setelah seluruh join() tuntas.
• Operasi counter += 1 terdiri atas tiga langkah yang dapat diselingi thread lain, sehingga memori bersama tanpa pengaman mengalami race condition dan kehilangan kenaikan.
• threading.Lock memastikan hanya satu thread yang menjalankan bagian kritisnya, dengan bentuk with lock sebagai cara paling aman melepas kuncinya.

Referensi#

• threading — Thread-based parallelism — Python Software Foundation — Dokumentasi resmi modul threading, termasuk kelas Thread beserta perilaku start dan join.
• Lock objects — Python Software Foundation — Rincian acquire, release, dan bentuk with pada objek Lock.