Speed up Cloud Storage object downloads

Downloading a large object in Cloud Storage sequentially over a single connection can leave available network bandwidth unused. By dividing an object into byte-range segments called stripes and reading those segments concurrently, you can improve download throughput for large objects using the Cloud Client Libraries for Rust.

This document shows you how to seed large test objects using object composition, download byte ranges concurrently with ReadRange::segment, and write stripes to a local file at fixed offsets. If you're new to the client library, see Using Cloud Storage with Rust first.

Before you begin

Before running the code in this document, complete the following prerequisites:

Dependencies

Add the Cloud Storage client library and supporting crates to your Cargo.toml file:

cargo add google-cloud-storage google-cloud-gax anyhow bytes futures
cargo add tokio --features macros,rt-multi-thread

Create large test objects

To test striped downloads without uploading gigabytes of data over the network, upload a small seed object and use the Cloud Storage compose operation to concatenate copies server-side. Inside an asynchronous seed function that accepts client, control, and bucket_name in the format projects/_/buckets/BUCKET_ID, complete the following steps:

  1. To reference existing objects in compose requests, import SourceObject: 

    use google_cloud_storage::model::compose_object_request::SourceObject;

  2. Upload a 1 MiB seed object named 1MiB.txt: 

    let buffer = String::from_iter(('a'..='z').cycle().take(1024 * 1024));
let seed = client
    .write_object(bucket_name, "1MiB.txt", bytes::Bytes::from_owner(buffer))
    .set_if_generation_match(0)
    .send_unbuffered()
    .await?;
println!(
    "Uploaded object {}, size={}KiB",
    seed.name,
    seed.size / 1024
);

  3. To create a 32 MiB object named 32MiB.txt, concatenate 32 copies of 1MiB.txt on the server: 

    let seed_32 = control
    .compose_object()
    .set_destination(Object::new().set_bucket(bucket_name).set_name("32MiB.txt"))
    .set_source_objects((0..32).map(|_| {
        SourceObject::new()
            .set_name(&seed.name)
            .set_generation(seed.generation)
    }))
    .set_if_generation_match(0)
    .with_idempotency(true)
    .send()
    .await?;
println!(
    "Created object {}, size={}MiB",
    seed.name,
    seed.size / (1024 * 1024)
);

  4. To create a 1 GiB object named 1GiB.txt, concatenate 32 copies of 32MiB.txt: 

    let seed_1024 = control
    .compose_object()
    .set_destination(Object::new().set_bucket(bucket_name).set_name("1GiB.txt"))
    .set_source_objects((0..32).map(|_| {
        SourceObject::new()
            .set_name(&seed_32.name)
            .set_generation(seed_32.generation)
    }))
    .set_if_generation_match(0)
    .with_idempotency(true)
    .send()
    .await?;
println!(
    "Created object {}, size={}MiB",
    seed.name,
    seed.size / (1024 * 1024)
);

  5. To build power-of-two test objects from 1GiB.txt (2GiB.txt, 4GiB.txt, 8GiB.txt, 16GiB.txt, and 32GiB.txt), repeat the compose operation: 

    for s in [2, 4, 8, 16, 32] {
    let name = format!("{s}GiB.txt");
    let target = control
        .compose_object()
        .set_destination(Object::new().set_bucket(bucket_name).set_name(&name))
        .set_source_objects((0..s).map(|_| {
            SourceObject::new()
                .set_name(&seed_1024.name)
                .set_generation(seed_1024.generation)
        }))
        .set_if_generation_match(0)
        .with_idempotency(true)
        .send()
        .await?;
    println!(
        "Created object {} size={} MiB",
        target.name,
        target.size / (1024 * 1024)
    );
}

Coordinate concurrent stripe downloads

To download an object in parallel, query its size and generation, preallocate the local destination file, and spawn one future per stripe. Inside a download function that accepts client, control, bucket_name, object_name, stripe_size, and destination, complete the following steps:

  1. Fetch the object metadata using StorageControl::get_object so that you can inspect the total object size and lock subsequent reads to a single generation: 

    let metadata = control
    .get_object()
    .set_bucket(bucket_name)
    .set_object(object_name)
    .send()
    .await?;

  2. Create the destination file and preallocate it to the exact object size so that worker tasks can write at independent byte offsets: 

    let file = Arc::new(std::fs::File::create(destination)?);
file.set_len(metadata.size as u64)?;

  3. Calculate each stripe offset using stripe_size as limit and create a write_stripe future per segment, appending a final partial stripe when the object size isn't evenly divisible by limit: 

    let size = metadata.size as u64;
let limit = stripe_size as u64;
let count = size / limit;
let mut stripes = (0..count)
    .map(|i| write_stripe(client.clone(), &file, i * limit, limit, &metadata))
    .collect::<Vec<_>>();
if !size.is_multiple_of(limit) {
    stripes.push(write_stripe(
        client.clone(),
        &file,
        count * limit,
        limit,
        &metadata,
    ))
}

  4. Run all stripe futures concurrently with futures::future::join_all and propagate any errors: 

    futures::future::join_all(stripes)
    .await
    .into_iter()
    .collect::<anyhow::Result<Vec<_>>>()?;

Download and write an individual stripe

Each write_stripe task reads a single byte range from Cloud Storage and writes the received chunks at the corresponding offset in the shared local file handle. To implement write_stripe, complete the following steps:

  1. Import ReadRange and define the write_stripe function: 

    async fn write_stripe(
    client: Storage,
    file: &Arc<std::fs::File>,
    offset: u64,
    limit: u64,
    metadata: &Object,
) -> anyhow::Result<()> {
    use google_cloud_storage::model_ext::ReadRange;

  2. Create a ReadObject request builder with Storage::read_object, and send the request after configuring the generation and read-range options in the following steps: 

    let mut reader = client
    .read_object(&metadata.bucket, &metadata.name)
    .send()
    .await?;

  3. To pin all stripes to the same object generation and prevent inconsistent reads if another process overwrites the object, call set_generation on the request builder before calling send: 

    .set_generation(metadata.generation)

  4. To restrict the read to the stripe's byte range, call set_read_range with ReadRange::segment(offset, limit) on the request builder: 

    .set_read_range(ReadRange::segment(offset, limit))

  5. Stream chunks from the reader and write each chunk to the destination file at current_pos inside tokio::task::spawn_blocking. Because Rust's standard library lacks a cross-platform positional write method, this example uses FileExt::write_all_at on Unix and FileExt::seek_write on Windows: 

    let mut current_pos = offset;
while let Some(b) = reader.next().await.transpose()? {
    let chunk_len = b.len() as u64;
    let handle = file.clone();
    #[cfg(unix)]
    tokio::task::spawn_blocking(move || {
        use std::os::unix::fs::FileExt;
        handle.write_all_at(&b, current_pos)
    })
    .await??;
    #[cfg(windows)]
    tokio::task::spawn_blocking(move || {
        use std::os::windows::fs::FileExt;
        handle.seek_write(&b, current_pos)
    })
    .await??;
    current_pos += chunk_len;
}

Performance considerations

Several factors determine the effective throughput of striped downloads:

  • Local disk speed. Writing chunks to slow local storage can become the bottleneck. Using high-throughput local SSD storage helps keep disk I/O ahead of network reads.
  • VM CPU capacity. Cloud Storage encrypts object data in transit, so the client decrypts every chunk it receives. Ensure your VM has enough vCPUs to handle concurrent TLS decryption across stripes.
  • Bucket and VM location. Downloading across regions introduces wide-area network latency and limits throughput. Colocate your VM in the same region as your bucket when possible.

You can also experiment with tuning stripe_size to avoid leaving a very small remainder stripe or creating too many tiny range requests.

Complete code

The following example combines the seed, download, and write_stripe functions into a complete module:

use std::sync::Arc;

use google_cloud_gax::options::RequestOptionsBuilder;
use google_cloud_storage::client::Storage;
use google_cloud_storage::client::StorageControl;
use google_cloud_storage::model::Object;

async fn seed(client: Storage, control: StorageControl, bucket_name: &str) -> anyhow::Result<()> {
    use google_cloud_storage::model::compose_object_request::SourceObject;

    let buffer = String::from_iter(('a'..='z').cycle().take(1024 * 1024));
    let seed = client
        .write_object(bucket_name, "1MiB.txt", bytes::Bytes::from_owner(buffer))
        .set_if_generation_match(0)
        .send_unbuffered()
        .await?;
    println!(
        "Uploaded object {}, size={}KiB",
        seed.name,
        seed.size / 1024
    );

    let seed_32 = control
        .compose_object()
        .set_destination(Object::new().set_bucket(bucket_name).set_name("32MiB.txt"))
        .set_source_objects((0..32).map(|_| {
            SourceObject::new()
                .set_name(&seed.name)
                .set_generation(seed.generation)
        }))
        .set_if_generation_match(0)
        .with_idempotency(true)
        .send()
        .await?;
    println!(
        "Created object {}, size={}MiB",
        seed.name,
        seed.size / (1024 * 1024)
    );

    let seed_1024 = control
        .compose_object()
        .set_destination(Object::new().set_bucket(bucket_name).set_name("1GiB.txt"))
        .set_source_objects((0..32).map(|_| {
            SourceObject::new()
                .set_name(&seed_32.name)
                .set_generation(seed_32.generation)
        }))
        .set_if_generation_match(0)
        .with_idempotency(true)
        .send()
        .await?;
    println!(
        "Created object {}, size={}MiB",
        seed.name,
        seed.size / (1024 * 1024)
    );

    for s in [2, 4, 8, 16, 32] {
        let name = format!("{s}GiB.txt");
        let target = control
            .compose_object()
            .set_destination(Object::new().set_bucket(bucket_name).set_name(&name))
            .set_source_objects((0..s).map(|_| {
                SourceObject::new()
                    .set_name(&seed_1024.name)
                    .set_generation(seed_1024.generation)
            }))
            .set_if_generation_match(0)
            .with_idempotency(true)
            .send()
            .await?;
        println!(
            "Created object {} size={} MiB",
            target.name,
            target.size / (1024 * 1024)
        );
    }

    Ok(())
}

async fn download(
    client: Storage,
    control: StorageControl,
    bucket_name: &str,
    object_name: &str,
    stripe_size: usize,
    destination: &str,
) -> anyhow::Result<()> {
    let metadata = control
        .get_object()
        .set_bucket(bucket_name)
        .set_object(object_name)
        .send()
        .await?;

    let file = Arc::new(std::fs::File::create(destination)?);
    file.set_len(metadata.size as u64)?;
    let start = std::time::Instant::now();

    let size = metadata.size as u64;
    let limit = stripe_size as u64;
    let count = size / limit;
    let mut stripes = (0..count)
        .map(|i| write_stripe(client.clone(), &file, i * limit, limit, &metadata))
        .collect::<Vec<_>>();
    if !size.is_multiple_of(limit) {
        stripes.push(write_stripe(
            client.clone(),
            &file,
            count * limit,
            limit,
            &metadata,
        ))
    }

    futures::future::join_all(stripes)
        .await
        .into_iter()
        .collect::<anyhow::Result<Vec<_>>>()?;

    let elapsed = start.elapsed();
    let mib = metadata.size as f64 / (1024.0 * 1024.0);
    let bw = mib / elapsed.as_secs_f64();
    println!(
        "Completed {mib:.2} MiB download in {elapsed:?}, using {count} stripes, effective bandwidth = {bw:.2} MiB/s"
    );

    Ok(())
}

async fn write_stripe(
    client: Storage,
    file: &Arc<std::fs::File>,
    offset: u64,
    limit: u64,
    metadata: &Object,
) -> anyhow::Result<()> {
    use google_cloud_storage::model_ext::ReadRange;
    let mut reader = client
        .read_object(&metadata.bucket, &metadata.name)
        .set_generation(metadata.generation)
        .set_read_range(ReadRange::segment(offset, limit))
        .send()
        .await?;
    let mut current_pos = offset;
    while let Some(b) = reader.next().await.transpose()? {
        let chunk_len = b.len() as u64;
        let handle = file.clone();
        #[cfg(unix)]
        tokio::task::spawn_blocking(move || {
            use std::os::unix::fs::FileExt;
            handle.write_all_at(&b, current_pos)
        })
        .await??;
        #[cfg(windows)]
        tokio::task::spawn_blocking(move || {
            use std::os::windows::fs::FileExt;
            handle.seek_write(&b, current_pos)
        })
        .await??;
        current_pos += chunk_len;
    }
    Ok(())
}

When you run download on a 32 MiB object with a 1 MiB stripe_size, the output is similar to the following, reporting the downloaded size, elapsed time, stripe count, and effective bandwidth:

Completed 32.00 MiB download in 412.5ms, using 32 stripes, effective bandwidth = 77.58 MiB/s

Clean up

To avoid incurring charges to your Google Cloud account for the test objects created by seed, delete the test objects and delete your test bucket when you're finished. Also remove the local destination file.