Google은 AI 기술을 사용하여 콘텐츠를 사용자의 기본 언어로 번역합니다. AI 번역에는 오류가 있을 수 있습니다.

Cloud Storage 객체 다운로드 속도 향상

Cloud Storage에서 단일 연결을 통해 대형 객체를 순차적으로 다운로드하면 사용 가능한 네트워크 대역폭이 사용되지 않을 수 있습니다. 객체를 스트라이프라고 하는 바이트 범위 세그먼트로 나누고 이러한 세그먼트를 동시에 읽으면 Rust용 Cloud 클라이언트 라이브러리를 사용하여 대형 객체의 다운로드 처리량을 개선할 수 있습니다.

이 문서에서는 객체 구성을 사용하여 대규모 테스트 객체를 시드하고, ReadRange::segment를 사용하여 바이트 범위를 동시에 다운로드하고, 고정 오프셋에서 로컬 파일에 스트라이프를 작성하는 방법을 보여줍니다. 클라이언트 라이브러리를 처음 사용하는 경우 먼저 Rust에서 Cloud Storage 사용을 참조하세요.

시작하기 전에

이 문서의 코드를 실행하기 전에 다음 기본 요건을 완료하세요.

결제가 사용 설정된 Google Cloud 프로젝트를 만들거나 선택합니다.
프로젝트에 전용 테스트 Cloud Storage 버킷을 만듭니다.
환경에 애플리케이션 기본 사용자 인증 정보를 구성합니다.
주 구성원에게 버킷의 객체를 읽고, 쓰고, 작성하고, 삭제할 수 있는 권한이 있는 Identity and Access Management(IAM) 역할(예: 스토리지 관리자(roles/storage.admin))을 부여합니다.

종속 항목

Cloud Storage 클라이언트 라이브러리 및 지원 크레이트를 Cargo.toml 파일에 추가합니다.

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

대규모 테스트 객체 만들기

네트워크를 통해 기가바이트 단위의 데이터를 업로드하지 않고 스트라이프 다운로드를 테스트하려면 작은 시드 객체를 업로드하고 Cloud Storage compose 작업을 사용하여 서버 측에서 복사본을 연결합니다. projects/_/buckets/BUCKET_ID 형식으로 client, control, bucket_name를 허용하는 비동기 seed 함수 내에서 다음 단계를 완료합니다.

작성 요청에서 기존 객체를 참조하려면 SourceObject를 가져옵니다.
```
use google_cloud_storage::model::compose_object_request::SourceObject;
```

1MiB.txt라는 이름의 1MiB 시드 객체를 업로드합니다.

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
);

32MiB.txt라는 32MiB 객체를 만들려면 서버에서 1MiB.txt의 사본 32개를 연결합니다.

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)
);

1GiB.txt이라는 1GiB 객체를 만들려면 32MiB.txt의 사본 32개를 연결합니다.

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)
);

1GiB.txt(2GiB.txt, 4GiB.txt, 8GiB.txt, 16GiB.txt, 32GiB.txt)에서 2의 거듭제곱 테스트 객체를 빌드하려면 작성 작업을 반복합니다.

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)
    );
}

동시 스트라이프 다운로드 조정

객체를 동시에 다운로드하려면 크기와 세대를 쿼리하고 로컬 대상 파일을 미리 할당하고 스트라이프당 하나의 future를 생성합니다. client, control, bucket_name, object_name, stripe_size, destination를 허용하는 download 함수 내에서 다음 단계를 완료합니다.

StorageControl::get_object를 사용하여 객체 메타데이터를 가져와 총 객체 크기를 검사하고 후속 읽기를 단일 세대로 잠급니다.
```
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)?;
```

stripe_size를 limit로 사용하여 각 스트라이프 오프셋을 계산하고 세그먼트당 write_stripe future를 만들어 객체 크기가 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,
    ))
}

futures::future::join_all를 사용하여 모든 스트라이프 Future를 동시에 실행하고 오류를 전파합니다.
```
futures::future::join_all(stripes)
    .await
    .into_iter()
    .collect::<anyhow::Result<Vec<_>>>()?;
```

개별 스트라이프 다운로드 및 쓰기

각 write_stripe 태스크는 Cloud Storage에서 단일 바이트 범위를 읽고 공유 로컬 파일 처리의 해당 오프셋에 수신된 청크를 씁니다. write_stripe를 구현하려면 다음 단계를 완료하세요.

ReadRange를 가져오고 write_stripe 함수를 정의합니다.

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;

Storage::read_object를 사용하여 ReadObject 요청 빌더를 만들고 다음 단계에서 생성 및 읽기 범위 옵션을 구성한 후 요청을 전송합니다.
```
let mut reader = client
    .read_object(&metadata.bucket, &metadata.name)
    .send()
    .await?;
```
모든 스트라이프를 동일한 객체 세대에 고정하고 다른 프로세스가 객체를 덮어쓰는 경우 일관되지 않은 읽기를 방지하려면 send를 호출하기 전에 요청 빌더에서 set_generation를 호출합니다.
```
.set_generation(metadata.generation)
```
스트라이프의 바이트 범위로 읽기를 제한하려면 요청 빌더에서 ReadRange::segment(offset, limit)를 사용하여 set_read_range를 호출합니다.
```
.set_read_range(ReadRange::segment(offset, limit))
```

리더에서 청크를 스트리밍하고 각 청크를 tokio::task::spawn_blocking 내부의 current_pos에 있는 대상 파일에 씁니다. Rust의 표준 라이브러리에는 크로스 플랫폼 위치 쓰기 메서드가 없으므로 이 예시에서는 Unix에서 FileExt::write_all_at를, Windows에서 FileExt::seek_write를 사용합니다.

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;
}

성능에 대한 고려사항

스트라이프 다운로드의 유효 처리량은 여러 요인에 따라 결정됩니다.

로컬 디스크 속도 느린 로컬 스토리지에 청크를 쓰는 것이 병목 현상이 될 수 있습니다. 처리량이 높은 로컬 SSD 스토리지를 사용하면 디스크 I/O를 네트워크 읽기보다 앞서 유지할 수 있습니다.
VM CPU 용량. Cloud Storage는 전송 중 데이터인 객체 데이터를 암호화하므로 클라이언트는 수신하는 모든 청크를 복호화합니다. VM에 스트라이프 간에 동시 TLS 복호화를 처리할 수 있는 충분한 vCPU가 있는지 확인합니다.
버킷 및 VM 위치 리전 간 다운로드로 인해 광역 통신망 지연 시간이 발생하고 처리량이 제한됩니다. 가능하면 버킷과 동일한 리전에 VM을 배치합니다.

stripe_size를 조정하여 매우 작은 나머지 스트라이프를 남기거나 너무 많은 작은 범위 요청을 만들지 않도록 실험할 수도 있습니다.

코드 완성

다음 예시에서는 seed, download, write_stripe 함수를 완전한 모듈로 결합합니다.

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(())
}

1MiB stripe_size를 사용하여 32MiB 객체에서 download를 실행하면 출력은 다음과 비슷하며 다운로드된 크기, 경과 시간, 스트라이프 수, 유효 대역폭을 보고합니다.

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

삭제

seed에서 만든 테스트 객체에 대해 Google Cloud 계정에 요금이 청구되지 않도록 하려면 완료 후 테스트 객체를 삭제하고 테스트 버킷을 삭제하세요. 로컬 대상 파일도 삭제합니다.