האצת ההורדות של אובייקטים ב-Cloud Storage

אם מורידים אובייקט גדול ב-Cloud Storage באופן סדרתי דרך חיבור יחיד, יכול להיות שרוחב הפס הזמין ברשת לא ינוצל. כדי לשפר את קצב ההורדה של אובייקטים גדולים באמצעות ספריות הלקוח של Cloud ל-Rust, אפשר לחלק אובייקט לפלחים של טווחי בייטים שנקראים פסים (stripes) ולקרוא את הפלחים האלה בו-זמנית.

במאמר הזה נסביר איך להפיץ אובייקטים גדולים לבדיקה באמצעות הרכבת אובייקטים, להוריד טווחים של בייטים בו-זמנית באמצעות ReadRange::segment ולכתוב פסים לקובץ מקומי בהיסטים קבועים. אם אתם משתמשים חדשים בספריית הלקוח, קודם קראו את המאמר שימוש ב-Cloud Storage עם Rust.

לפני שמתחילים

לפני שמריצים את הקוד במסמך הזה, צריך להשלים את הדרישות המוקדמות הבאות:

• יוצרים או בוחרים פרויקט ב-Google Cloud שהחיוב מופעל בו.
• יוצרים קטגוריה של Cloud Storage לבדיקה ייעודית בפרויקט.
• מגדירים את Application Default Credentials בסביבה שלכם.
• מקצים לחשבון המשתמש תפקיד בניהול זהויות והרשאות גישה (IAM) עם הרשאה לקרוא, לכתוב, ליצור ולמחוק אובייקטים בקטגוריה, כמו אדמין לניהול אחסון (roles/storage.admin).

הערה: הפעלת שלב ההזנה המלא יוצרת כמה אובייקטים לבדיקה, כל אחד בנפח של עד 32GiB (יותר מ-63GiB בסך הכול). כדי להימנע מחיובים שוטפים על אחסון, מוחקים את אובייקטי הבדיקה ואת קובץ הפלט המקומי מיד בסיום הבדיקה.

יחסי תלות

מוסיפים את ספריית הלקוח של Cloud Storage ואת ה-crates התומכים לקובץ Cargo.toml:

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

יצירת אובייקטים גדולים לבדיקה

כדי לבדוק הורדות מחולקות בלי להעלות גיגה-בייטים של נתונים ברשת, מעלים אובייקט seed קטן ומשתמשים בפעולת ה-compose של Cloud Storage כדי לשרשר עותקים בצד השרת. בתוך פונקציה אסינכרונית של seed שמקבלת את client,‏ control ו-bucket_name בפורמט projects/_/buckets/BUCKET_ID, מבצעים את הפעולות הבאות:

1. כדי להפנות לאובייקטים קיימים בבקשות הרכבה, מייבאים את SourceObject:

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

2. מעלים אובייקט ראשוני בנפח 1 MiB בשם 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. כדי ליצור אובייקט בגודל 32 MiB בשם 32MiB.txt, משרשרים 32 עותקים של 1MiB.txt בשרת:

   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. כדי ליצור אובייקט בנפח 1 GiB בשם 1GiB.txt, משרשרים 32 עותקים של 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. כדי ליצור אובייקטים לבדיקה בחזקות של שתיים מ-1GiB.txt (2GiB.txt, 4GiB.txt, 8GiB.txt, 16GiB.txt ו-32GiB.txt), חוזרים על פעולת ההרכבה:

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

תיאום הורדות מקבילות של רצועות

כדי להוריד אובייקט במקביל, שולחים שאילתה לגבי הגודל והגנרציה שלו, מקצים מראש את קובץ היעד המקומי ויוצרים עתיד אחד לכל פס. בתוך פונקציית download שמקבלת את client,‏ control,‏ bucket_name,‏ object_name,‏ stripe_size ו-destination, מבצעים את הפעולות הבאות:

1. אחזור המטא-נתונים של האובייקט באמצעות StorageControl::get_object כדי לבדוק את הגודל הכולל של האובייקט ולנעול את הקריאות הבאות לדור יחיד:

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

2. יוצרים את קובץ היעד ומקצים לו מראש את הגודל המדויק של האובייקט, כדי שמשימות ה-worker יוכלו לכתוב בהיסטים עצמאיים של בייטים:

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

3. מחשבים את ההיסט של כל פס באמצעות stripe_size בתור limit ויוצרים write_stripe עתידי לכל פלח, ומוסיפים פס חלקי בסוף אם גודל האובייקט לא מתחלק באופן שווה ב-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. מריצים את כל ה-futures של Stripe בו-זמנית עם futures::future::join_all ומפיצים את כל השגיאות:

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

הורדה וכתיבה של פס בודד

כל משימת write_stripe קוראת טווח בייטים יחיד מ-Cloud Storage וכותבת את הנתונים שהתקבלו במיקום המתאים ב-file handle המקומי המשותף. כדי להטמיע את write_stripe, מבצעים את השלבים הבאים:

1. מייבאים את 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;

2. יוצרים כלי ליצירת בקשות ReadObject באמצעות Storage::read_object, ושולחים את הבקשה אחרי שמגדירים את אפשרויות היצירה וטווח הקריאה בשלבים הבאים:

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

3. כדי להצמיד את כל הפסים לאותו דור של אובייקטים ולמנוע קריאות לא עקביות אם תהליך אחר מחליף את האובייקט, צריך להפעיל את set_generation בבונה הבקשות לפני שמפעילים את send:

   .set_generation(metadata.generation)

4. כדי להגביל את הקריאה לטווח הבייטים של ה-stripe, צריך להפעיל את set_read_range עם ReadRange::segment(offset, limit) בבונה הבקשות:

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

5. מעבירים בסטרימינג קטעים מהקורא וכותבים כל קטע לקובץ היעד ב-current_pos בתוך tokio::task::spawn_blocking. בספרייה הרגילה של Rust אין שיטת כתיבה לפי מיקום שמתאימה בפלטפורמות שונות, אז בדוגמה הזו משתמשים ב-FileExt::write_all_at ב-Unix וב-FileExt::seek_write ב-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;
   }

שיקולי ביצועים

יש כמה גורמים שקובעים את קצב העברת הנתונים האפקטיבי של הורדות מחולקות:

• מהירות הדיסק המקומי. כתיבת מקטעים לאחסון מקומי איטי יכולה להפוך לצוואר בקבוק. שימוש באחסון SSD מקומי עם תפוקה גבוהה עוזר לשמור על קצב קלט/פלט של הדיסק שגבוה מקצב הקריאה ברשת.
• קיבולת המעבד (CPU) של המכונה הווירטואלית. ‫Cloud Storage מצפין את נתוני האובייקטים במעבר, כך שהלקוח מפענח כל קבוצת נתונים שהוא מקבל. מוודאים שבמכונה הווירטואלית יש מספיק מעבדי vCPU כדי לטפל בפענוח TLS בו-זמנית בין פסים.
• המיקום של הקטגוריה והמכונה הווירטואלית. הורדה בין אזורים גורמת לזמן אחזור ברשת WAN ומגבילה את קצב הנתונים. אם אפשר, כדאי למקם את המכונה הווירטואלית באותו אזור שבו נמצאת הקטגוריה.

אתם יכולים גם להתנסות בכוונון של 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(())
}

כשמריצים את download על אובייקט בגודל 32MiB עם stripe_size בגודל 1MiB, הפלט דומה לפלט הבא, שמדווח על גודל ההורדה, הזמן שחלף, מספר הפסים ורוחב הפס בפועל:

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

הסרת המשאבים

כדי להימנע מצבירת חיובים בחשבון Google Cloud על אובייקטים לבדיקה שנוצרו על ידי seed, בסיום הבדיקה מחקו את האובייקטים לבדיקה ומחקו את הקטגוריה לבדיקה. צריך להסיר גם את קובץ היעד המקומי.