KAL
Menu

Streams Paralel

Menilai kapan streams paralel membantu dan kapan justru merugikan, dibuktikan dengan pengukuran.

Daftar isi

Stream paralel adalah mode eksekusi yang membagi pekerjaan pipeline ke beberapa thread sekaligus, diaktifkan dengan satu pemanggilan parallel() atau parallelStream(). Penulisannya murah, tetapi penilaiannya tidak, karena pembagian itu menimbulkan biaya yang tidak selalu tertutup oleh keuntungannya. Topik ini membahas cara kerja pembagiannya, membuktikan bahwa pekerjaan benar-benar terbagi ke beberapa thread, lalu menimbang dengan pengukuran kapan stream paralel membantu dan kapan justru merugikan.

Cara kerja pembagian pekerjaan#

Ketika sebuah stream diparalelkan, sumber datanya dipecah menjadi bagian-bagian oleh mekanisme spliterator, lalu tiap bagian diproses oleh thread yang berbeda sebelum hasilnya digabung kembali. Thread yang dipakai berasal dari ForkJoinPool.commonPool(), yaitu kumpulan thread bersama milik JVM dengan tingkat paralelisme bawaan sebesar jumlah inti prosesor dikurangi satu, dan thread pemanggil ikut bekerja pula. Karena penggabungan inilah kontrak operasi menjadi lebih ketat: lambda harus tanpa keadaan luar dan penggabungnya harus asosiatif agar hasilnya sama untuk susunan pemrosesan apa pun.

Bukti paling dasar bahwa paralel tidak merusak hasil adalah kesamaan jawaban antara eksekusi sekuensial dan paralel, sekaligus kontrol urutan lewat forEachOrdered. Program berikut menjumlahkan dua juta bilangan lewat dua jalur itu, lalu mencetak isi daftar kecil yang diolah secara paralel tetapi tetap pada urutan sumbernya.

java DemoParalelBenar.java 
import java.util.List;
import java.util.stream.LongStream;

public class DemoParalelBenar {

    public static void main(String[] args) {
        long jumlahParalel = LongStream.rangeClosed(1, 2_000_000)
                .parallel()
                .sum();
        long jumlahBiasa = LongStream.rangeClosed(1, 2_000_000)
                .sum();

        System.out.println("Jumlah paralel   : " + jumlahParalel);
        System.out.println("Jumlah sekuensial: " + jumlahBiasa);
        System.out.println("Hasilnya sama    : " + (jumlahParalel == jumlahBiasa));

        System.out.println("forEachOrdered:");
        List.of("sabun", "sikat", "handuk").parallelStream()
                .forEachOrdered(nama -> System.out.println("  " + nama));
    }
}
text Keluaran 
Jumlah paralel   : 2000001000000
Jumlah sekuensial: 2000001000000
Hasilnya sama    : true
forEachOrdered:
  sabun
  sikat
  handuk

Hasil penjumlahannya identik karena sum memenuhi kontrak penggabungan yang asosiatif. forEachOrdered menahan setiap elemen sampai gilirannya tiba, sehingga urutan sumber tetap terjaga meskipun pekerjaannya paralel. Tanpa forEachOrdered, bentuk forEach biasa pada stream paralel bebas mencetak dengan urutan apa pun yang kebetulan selesai lebih dulu. Kontrak semacam inilah yang membedakan kode paralel yang benar dari kode yang hanya kebetulan benar.

Bukti kerja terbagi ke beberapa thread#

Pernyataan “pekerjaan terbagi” dapat dibuktikan langsung dengan mencatat nama thread yang menyentuh elemen. Karena thread yang bersamaan menulis ke satu wadah berbahaya, wadah pencatatnya memakai ConcurrentHashMap.newKeySet() yang aman dari akses bersamaan. Program berikut menjalankan penjumlahan besar secara paralel lalu mencetak siapa saja thread yang ikut memprosesnya.

java SiapaBekerja.java 
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.LongStream;

public class SiapaBekerja {

    public static void main(String[] args) {
        Set<String> pelaksana = ConcurrentHashMap.newKeySet();

        LongStream.rangeClosed(1, 20_000_000).parallel()
                .forEach(n -> pelaksana.add(Thread.currentThread().getName()));

        System.out.println("Jumlah thread yang memproses: " + pelaksana.size());
        pelaksana.stream().sorted().forEach(t -> System.out.println("  " + t));
    }
}
text Keluaran 
Jumlah thread yang memproses: 2
  ForkJoinPool.commonPool-worker-1
  main

Keluaran tersebut diperoleh pada mesin dengan dua inti, sehingga tingkat paralelisme kumpulan thread bawaannya satu dan thread main ikut bekerja sebagai pelaksana kedua. Pada mesin dengan inti lebih banyak, daftar ini memuat nama worker tambahan sesuai jumlah intinya. Intinya, satu pemanggilan parallel() benar-benar memobilisasi lebih dari satu thread, dan wadah pencatatnya pun harus aman dari akses bersamaan sebagaimana ditunjukkan contohnya.

Kapan paralel membantu dan merugikan#

Keputusan memakai stream paralel sebaiknya berdiri di atas pengukuran, dan pengukuran yang jujur memerlukan pemanasan JIT karena eksekusi awal jauh lebih lambat daripada kondisi stabilnya. Program berikut mengukur penjumlahan 50 juta bilangan pada kedua mode setelah tiga putaran pemanasan, dan angkanya memang berbeda-beda antar mesin bahkan antar eksekusi, sehingga berkasnya sengaja tidak disertai keluaran baku di sini. Jalankan sendiri berkasnya dengan perintah java UkuranWaktu.java untuk melihat angka di lingkungan masing-masing.

java UkuranWaktu.java 
// tidak-dijalankan
import java.util.stream.LongStream;

public class UkuranWaktu {

    static long jumlahSekuensial(long batas) {
        return LongStream.rangeClosed(1, batas).sum();
    }

    static long jumlahParalel(long batas) {
        return LongStream.rangeClosed(1, batas).parallel().sum();
    }

    public static void main(String[] args) {
        long batas = 50_000_000;

        // pemanasan JIT agar pengukuran tidak mencatat fase penafsiran awal
        for (int i = 0; i < 3; i++) {
            jumlahSekuensial(batas);
            jumlahParalel(batas);
        }

        long mulai = System.nanoTime();
        long total1 = jumlahSekuensial(batas);
        long waktuSekuensial = System.nanoTime() - mulai;

        mulai = System.nanoTime();
        long total2 = jumlahParalel(batas);
        long waktuParalel = System.nanoTime() - mulai;

        System.out.println("Sekuensial : " + waktuSekuensial / 1_000_000 + " ms");
        System.out.println("Paralel    : " + waktuParalel / 1_000_000 + " ms");
        System.out.println("Total sama : " + (total1 == total2));
    }
}

Pola yang lazim muncul dari pengukuran semacam ini konsisten dengan teorinya. Untuk penjumlahan dengan elemen berbiaya sangat murah, keuntungan paralelnya kecil dan pada data yang lebih sedikit paralel justru kalah karena biaya pemecahan dan penggabungannya melebihi pekerjaan yang dihemat. Keuntungan baru terasa ketika data cukup besar, pekerjaan per elemen cukup berat, dan sumber mudah dipecah seperti ArrayList atau rentang bilangan, sedangkan LinkedList dan Stream.iterate sulit dipecah sehingga paralelisasinya sering merugikan. Adapun pekerjaan yang menunggu jaringan atau berkas bukan wilayah stream paralel, karena thread yang menunggu tidak menghitung apa pun dan kumpulan thread bawaannya dipakai bersama oleh seluruh aplikasi. Bila pengukuran performa menjadi pekerjaan rutin, pustaka JMH yang dirilis proyek OpenJDK menyediakan perangkat pengukuran mikro yang jauh lebih dapat diandalkan daripada System.nanoTime manual.

Kesalahan umum#

  • Menambahkan hasil ke ArrayList lewat forEach paralel. Beberapa thread menyentuh daftar yang sama tanpa pengaman, sehingga isinya kurang atau program melempar exception. Gunakan collect(Collectors.toList()) yang menangani penggabungan hasil antar-bagian dengan benar.
  • Memaksakan paralel untuk data kecil atau operasi murah. Biaya pemecahan, penjadwalan thread, dan penggabungan melebihi pekerjaan yang dihemat, sehingga waktunya justru bertambah. Ukur dahulu, dan biarkan pipeline kecil tetap sekuensial.
  • Memakai sumber yang sulit dipecah. LinkedList dan Stream.iterate hanya mampu dipecah satu per satu, sehingga biaya pembagiannya mendominasi. Gunakan ArrayList, array, atau rentang bilangan sebagai sumber bila paralel dikehendaki.
  • Mengira urutan tetap terjaga. forEach pada stream paralel mencetak dengan urutan penyelesaian, bukan urutan sumber. Pakai forEachOrdered bila urutan pertemuan menjadi bagian dari kebutuhan.

Ringkasan#

  • parallel() memecah sumber data lewat spliterator dan membagi pekerjaannya ke thread ForkJoinPool.commonPool() beserta thread pemanggil.
  • Hasil paralel menjamin kebenarannya hanya bila lambda tanpa keadaan luar dan penggabungannya asosiatif, sebagaimana sum pada contoh.
  • Pengukuran dengan pemanasan JIT menunjukkan pola yang konsisten, yaitu paralel menguntungkan untuk data besar, pekerjaan per elemen berat, dan sumber yang mudah dipecah.
  • forEachOrdered menjaga urutan pertemuan, sedangkan forEach paralel bebas urutan dan dilarang menulis ke koleksi biasa.

Referensi#