Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

Konkurensi

AWS SDK for Rust Itu tidak memberikan kontrol konkurensi tetapi pengguna memiliki banyak opsi untuk mengimplementasikannya sendiri.

Ketentuan

Istilah yang terkait dengan subjek ini mudah membingungkan dan beberapa istilah telah menjadi sinonim meskipun awalnya mewakili konsep yang terpisah. Dalam panduan ini, kami akan mendefinisikan yang berikut:

Contoh sederhana

Contoh pertama kami adalah program sekuensial. Dalam contoh selanjutnya, kita akan mengubah kode ini menggunakan teknik konkurensi. Contoh selanjutnya menggunakan kembali build_client_and_list_objects_to_download() metode yang sama dan membuat perubahan di dalamnyamain().

Contoh tugas berikut adalah mengunduh semua file di bucket Amazon Simple Storage Service:

const EXAMPLE_BUCKET: &str = "<an-example-bucket>";

// This initialization function won't be reproduced in
// examples following this one, in order to save space.
async fn build_client_and_list_objects_to_download() {
    let cfg = aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await;
    let client = Client::new(&cfg);
    let objects_to_download: Vec<_> = client
        .list_objects_v2()
        .bucket(EXAMPLE_BUCKET)
        .send()
        .await
        .expect("listing objects succeeds")
        .contents()
        .into_iter()
        .flat_map(aws_sdk_s3::types::Object::key)
        .map(ToString::to_string)
        .collect();
         
    (client, objects_to_download)
}

#[tokio::main]
async fn main() {
    let (client, objects_to_download) =
        build_client_and_list_objects_to_download().await;
    
    for object in objects_to_download {
        let res = client
            .get_object()
            .key(&object)
            .bucket(EXAMPLE_BUCKET)
            .send()
            .await
            .expect("get_object succeeds");
        let body = res.body.collect().await.expect("reading body succeeds").into_bytes();
        std::fs::write(object, body).expect("write succeeds");
    }
}

Karena arsitektur komputer modern, kita dapat menulis ulang program ini menjadi jauh lebih efisien. Kita akan melakukannya dalam contoh selanjutnya, tetapi pertama-tama, mari kita pelajari beberapa konsep lagi.

Kepemilikan dan mutabilitas

Setiap nilai di Rust memiliki satu pemilik. Ketika pemilik keluar dari ruang lingkup, semua nilai yang dimilikinya juga akan dihapus. Pemilik dapat memberikan satu atau lebih referensi yang tidak dapat diubah ke nilai atau referensi tunggal yang dapat berubah. Kompiler Rust bertanggung jawab untuk memastikan bahwa tidak ada referensi yang hidup lebih lama dari pemiliknya.

Perencanaan dan desain tambahan diperlukan ketika beberapa tugas perlu mengakses sumber daya yang sama secara berubah-ubah. Dalam komputasi sekuensial, setiap tugas dapat mengakses sumber daya yang sama tanpa pertengkaran karena mereka berjalan satu demi satu secara berurutan. Namun, dalam komputasi bersamaan, tugas dapat berjalan dalam urutan apa pun, dan pada saat yang sama. Oleh karena itu, kita harus berbuat lebih banyak untuk membuktikan kepada kompiler bahwa beberapa referensi yang dapat berubah tidak mungkin (atau setidaknya macet jika memang terjadi).

Pustaka standar Rust menyediakan banyak alat untuk membantu kami mencapai hal ini. Untuk informasi lebih lanjut tentang topik ini, lihat Variabel dan Mutabilitas dan Memahami Kepemilikan dalam buku The Rust Programming Language.

Lebih banyak istilah!

Berikut ini adalah daftar “objek sinkronisasi”. Secara keseluruhan, mereka adalah alat yang diperlukan untuk meyakinkan kompiler bahwa program bersamaan kami tidak akan melanggar aturan kepemilikan.

Objek sinkronisasi perpustakaan standar:

Objek sinkronisasi Tokio:

Meskipun AWS SDKs dimaksudkan untuk menjadi async -runtime-agnostik, kami merekomendasikan penggunaan objek tokio sinkronisasi untuk kasus tertentu.

Menulis ulang contoh kita menjadi lebih efisien (konkurensi ulir tunggal)

Dalam contoh modifikasi berikut, kita gunakan futures_util::future::join_alluntuk menjalankan SEMUA get_object permintaan secara bersamaan.

#[tokio::main]
async fn main() {
    let (client, objects_to_download) =
        build_client_and_list_objects_to_download().await;
        
    let get_object_futures = objects_to_download.into_iter().map(|object| {
        let req = client
            .get_object()
            .key(&object)
            .bucket(EXAMPLE_BUCKET);

        async {
            let res = req
                .send()
                .await
                .expect("get_object succeeds");
            let body = res.body.collect().await.expect("body succeeds").into_bytes();
           // Note that we MUST use the async runtime's preferred way
           // of writing files. Otherwise, this call would block,
           // potentially causing a deadlock.
            tokio::fs::write(object, body).await.expect("write succeeds");
        }
    });

    futures_util::future::join_all(get_object_futures).await;
}

Ini adalah cara paling sederhana untuk mendapatkan keuntungan dari konkurensi, tetapi juga memiliki beberapa masalah yang mungkin tidak jelas pada pandangan pertama:

Untuk memperbaiki kedua masalah ini, kami harus membatasi jumlah permintaan yang kami kirim pada satu waktu. Kami akan melakukan ini dengan tokio semaphore:

const CONCURRENCY_LIMIT: usize = 50; 

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (client, objects_to_download) =
        build_client_and_list_objects_to_download().await;
    let concurrency_semaphore = Arc::new(Semaphore::new(CONCURRENCY_LIMIT));

    let get_object_futures = objects_to_download.into_iter().map(|object| {
        // Since each future needs to acquire a permit, we need to clone
        // the Arc'd semaphore before passing it in.
        let semaphore = concurrency_semaphore.clone();
        // We also need to clone the client so each task has its own handle.
        let client = client.clone();
        async move {
            let permit = semaphore
                .acquire()
                .await
                .expect("we'll get a permit if we wait long enough");
            let res = client
                .get_object()
                .key(&object)
                .bucket(EXAMPLE_BUCKET)
                .send()
                .await
                .expect("get_object succeeds");
            let body = res.body.collect().await.expect("body succeeds").into_bytes();
            tokio::fs::write(object, body).await.expect("write succeeds");
            std::mem::drop(permit);
        }
    });

    futures_util::future::join_all(get_object_futures).await;
}

Kami telah memperbaiki masalah penggunaan memori potensial dengan memindahkan pembuatan permintaan ke dalam async blok. Dengan cara ini, permintaan tidak akan dibuat sampai tiba waktunya untuk mengirimnya.

Kami juga telah memperbaiki masalah pengiriman terlalu banyak permintaan sekaligus dengan membatasi permintaan dalam penerbanganCONCURRENCY_LIMIT.

Menulis ulang contoh kita menjadi lebih efisien (konkurensi multi-utas)

Dalam dua contoh sebelumnya, kami melakukan permintaan kami secara bersamaan. Meskipun ini lebih efisien daripada menjalankannya secara sinkron, kita dapat membuat segalanya lebih efisien dengan menggunakan multi-threading. Untuk melakukan initokio, kita harus menelurkannya sebagai tugas terpisah.

// Set this based on the amount of cores your target machine has.
const THREADS: usize = 8; 

#[tokio::main(flavor = "multi_thread")]
async fn main() {
    let (client, objects_to_download) =
        build_client_and_list_objects_to_download().await;
    let concurrency_semaphore = Arc::new(Semaphore::new(THREADS));

    let get_object_task_handles = objects_to_download.into_iter().map(|object| {
        // Since each future needs to acquire a permit, we need to clone
        // the Arc'd semaphore before passing it in.
        let semaphore = concurrency_semaphore.clone();
        // We also need to clone the client so each task has its own handle.
        let client = client.clone();
        
        // Note this difference! We're using `tokio::task::spawn` to
        // immediately begin running these requests.
        tokio::task::spawn(async move {
            let permit = semaphore
                .acquire()
                .await
                .expect("we'll get a permit if we wait long enough");
            let res = client
                .get_object()
                .key(&object)
                .bucket(EXAMPLE_BUCKET)
                .send()
                .await
                .expect("get_object succeeds");
            let body = res.body.collect().await.expect("body succeeds").into_bytes();
            tokio::fs::write(object, body).await.expect("write succeeds");
            std::mem::drop(permit);
        })
    });

    futures_util::future::join_all(get_object_task_handles).await;
}

Membagi pekerjaan menjadi tugas bisa jadi rumit. Melakukan I/O (input/output) biasanya memblokir. Runtime mungkin berjuang untuk menyeimbangkan kebutuhan tugas yang berjalan lama dengan tugas-tugas jangka pendek. Apa pun runtime yang Anda pilih, pastikan untuk membaca rekomendasi mereka untuk cara paling efisien untuk membagi pekerjaan Anda menjadi tugas. Untuk rekomendasi tokio runtime, lihat Modul tokio::task.

Mendebug aplikasi multi-utas

Tugas yang berjalan secara bersamaan dapat dijalankan dalam urutan apa pun. Dengan demikian, log program bersamaan bisa sangat sulit dibaca. Di SDK untuk Rust, kami sarankan menggunakan sistem tracing logging. Itu dapat mengelompokkan log dengan tugas khusus mereka, tidak peduli kapan mereka berjalan. Untuk panduan, lihat Aktifkan pencatatan AWS SDK for Rust kode.

Alat yang sangat berguna untuk mengidentifikasi tugas yang telah terkunci adalah tokio-console, yang merupakan alat diagnostik dan debugging untuk program Rust asinkron. Dengan menginstrumentasi dan menjalankan program Anda, dan kemudian menjalankan tokio-console aplikasi, Anda dapat melihat tampilan langsung dari tugas yang dijalankan program Anda. Tampilan ini mencakup informasi bermanfaat seperti jumlah waktu yang dihabiskan tugas untuk menunggu untuk memperoleh sumber daya bersama atau berapa kali tugas tersebut telah disurvei.