Konkurensi dengan async

Program yang menunggu banyak sumber luar, misalnya berkas jaringan atau layanan lain, membuang sebagian besar waktunya apabila menunggu satu per satu. Pustaka async menjalankan pekerjaan tersebut pada thread ringan yang berjalan bersama, dengan dua jaminan yang membuatnya nyaman digunakan, yaitu hasil setiap thread dikumpulkan sebagai nilai IO biasa dan kegagalan satu thread tidak hilang tanpa jejak. Topik ini membahas dasar-dasar async, kombinator race dan concurrently, serta pemetaannya atas list.

Pustaka async dipasang sekali melalui Cabal sebelum contoh-contohnya dijalankan, dan setelahnya modul Control.Concurrent.Async dapat diimpor dari program mana pun.

# tidak-dijalankan
# pasang pustaka async ke lingkungan paket pengguna
cabal install --lib async

Thread dan nilai aksinya#

Fungsi async :: IO a -> IO (Async a) memulai pekerjaan pada thread baru dan langsung mengembalikan tanda (handle) berupa Async a, sedangkan fungsi wait :: Async a -> IO a menunggu pekerjaan itu selesai dan menyampaikan hasilnya. Penggunaan paling dasarnya berupa memulai dua aksi lalu menunggu keduanya, dan kombinator concurrently merangkum pola itu menjadi satu pemanggilan.

-- dua aksi berjalan bersama dan hasilnya dikumpulkan
import Control.Concurrent.Async (concurrently)

main :: IO ()
main = do
  (angka, teks) <- concurrently (return (2 + 3)) (return "lima")
  print angka
  putStrLn teks

5
lima

Kembalian concurrently berupa pasangan, yaitu hasil aksi kiri dan hasil aksi kanan, sehingga pengumpulan dua pekerjaan tidak memerlukan pengelolaan thread secara manual. Pada contoh yang tampak di sini, kedua aksi berupa return yang selesai seketika, sedangkan aksi yang sesungguhnya menunggu sumber luar memperoleh manfaat penuh karena menunggunya tumpang-tindih alih-alih berurutan. Pembagian hasilnya juga mengikuti tipe secara ketat, karena aksi kiri dan kanan boleh bertipe berbeda dan pasangan hasilnya mengikuti keduanya.

Jaminan kedua pustaka ini menyangkut kegagalan. Apabila salah satu aksi yang dijalankan concurrently atau race mengalami exception, kombinator tersebut menyebarkan exception itu ke pemanggil alih-alih membiarkan thread yang gagal menghilang senyap, sehingga penanganan error program utama tetap bekerja sebagaimana biasa. Perilaku ini menghilangkan kelas bug yang lazim pada pengelolaan thread manual, yaitu pekerjaan yang gagal tanpa seorang pun mengetahuinya.

Race dan pembatalan#

Kombinator kedua bernama race, yang menjalankan dua aksi bersama dan mengembalikan hasil aksi yang selesai lebih dahulu, sedangkan aksi yang tertinggal dibatalkan. Kembalinya berupa Either, yaitu Left untuk pemenang kiri dan Right untuk pemenang kanan, sehingga identitas pemenangnya tetap terbaca.

-- pemenang lomba ditentukan jeda tercepat
import Control.Concurrent.Async
import Control.Concurrent (threadDelay)

pebalap :: Int -> String -> IO String
pebalap jeda nama = do
  threadDelay jeda
  return nama

main :: IO ()
main = do
  hasil <- race (pebalap 2000000 "kancil") (pebalap 1000000 "kelinci")
  print hasil

Right "kelinci"

Fungsi threadDelay menahan thread yang memanggilnya selama jumlah mikrodetik yang diberikan, sehingga aksi kelinci selesai pada jeda satu detik dan aksi kancil yang menunggu dua detik dibatalkan sebelum sempat mengembalikan hasilnya. Pembatalan pada pustaka ini bukan pengabaian, melainkan pengiriman exception pembatalan ke thread yang tertinggal, sehingga sumber daya yang dibuka aksi tersebut tetap dilepaskan melalui penanganan exceptionnya. Pola race menjadi dasar batas waktu, yaitu menjalankan pekerjaan bersama aksi yang hanya menunggu, dan pemenangnya menentukan hasil atau kegagalan batas waktu tersebut.

Pemetaan atas list#

Kebutuhan yang paling lazim berupa menjalankan aksi yang sama atas banyak masukan, dan mapConcurrently mengerjakannya dengan menjaga urutan hasil sesuai urutan masukan, bukan urutan keselesaiannya.

-- pemetaan aksi atas list dengan mapConcurrently
import Control.Concurrent.Async (mapConcurrently)
import Control.Concurrent (threadDelay)

unduh :: Int -> IO Int
unduh n = do
  threadDelay (n * 10000)
  return (n * n)

main :: IO ()
main = do
  hasil <- mapConcurrently unduh [3, 1, 2]
  print hasil

[9,1,4]

Aksi untuk masukan 3 membutuhkan waktu terlama, tetapi hasilnya tetap menempati posisi pertama karena kembalian mengikuti urutan list, sifat yang membuat kode pemanggil tidak perlu menyusun ulang hasilnya. Tiga masukan pada contoh ini berjalan bersama sebagai tiga thread ringan, dan jumlah thread semacam itu murah pada runtime GHC, sehingga pemetaan paralel atas puluhan pekerjaan terikat IO merupakan penggunaan yang wajar.

Varian lainnya berupa forConcurrently, yaitu mapConcurrently dengan urutan argumen yang terbalik sehingga aksinya ditulis setelah listnya, bentuk yang lebih nyaman digunakan bersama lambda. Seluruh kombinator pada pustaka ini mengikuti satu kesepakatan, yaitu menerima aksi IO biasa dan mengembalikan hasilnya sebagai nilai IO biasa, sehingga penggunaan konkurensi tidak mengubah bentuk tipe program yang memakainya. Perpindahan dari versi berurutan ke versi konkuren pun semudah mengganti mapM menjadi mapConcurrently pada tipe yang sama.

Kesalahan umum#

Kesalahan pertama adalah menunggu tanda thread secara manual dalam urutan yang keliru. Program yang memulai aksi lambat lebih dahulu lalu menunggunya sebelum memulai aksi cepat kehilangan seluruh manfaat konkurensinya, dan perbaikannya berupa memulai kedua aksi terlebih dahulu, bentuk yang tepat dirangkum oleh concurrently dan race.

Kesalahan kedua berupa mengharapkan urutan penyelesaian pada kembalian mapConcurrently. Urutan hasilnya mengikuti urutan masukan sesuai jaminan tipenya, sehingga efek samping di dalam aksi mungkin selesai lebih dahulu untuk masukan kedua, dan pemisahan yang tepat berupa hasil yang dikumpulkan sebagai nilai, bukan pencetakan di dalam thread. Kesalahan ketiga berupa menggunakan konkurensi untuk pekerjaan yang murni menghitung, karena thread semacam ini dirancang untuk menunggu sumber luar, sedangkan perhitungan berat membutuhkan mekanisme paralelisme yang berbeda.

Ringkasan#

• async memulai aksi pada thread ringan dan wait mengumpulkan hasilnya, sedangkan concurrently dan race merangkum pola ganda yang paling lazim.
• race mengembalikan Either berisi pemenang dan membatalkan aksi yang tertinggal melalui exception, bentuk dasar batas waktu.
• mapConcurrently menjaga urutan hasil sesuai urutan masukan, dan jumlah aksi yang berjalan bersama dibatasi melalui semaphore apabila ukuran listnya tidak terkendali.
• Konkurensi ini menguntungkan pekerjaan yang terikat IO, bukan perhitungan murni yang berat.

Referensi#

• Control.Concurrent.Async — Hackage — Dokumentasi async, race, concurrently, dan mapConcurrently.
• async package — Hackage — Halaman paket async beserta catatan desain pembatalannya.
• Control.Concurrent — Hackage — Dokumentasi threadDelay dan thread ringan pada pustaka standar.