Operator concurrency

[dr3s]

I'm not sure if I just don't understand the API but I'm finding I can only use the opendal backblaze operator for one upload at a time.

This what I'm putting into my app state:

    let builder = B2::default()
        .root("/")
        .application_key_id(id.as_ref())
        .application_key(key.as_ref())
        .bucket("mybucket")
        .bucket_id("adafdasdassdfas");

    let op: Operator = Operator::new(builder)?.finish();

The operator gets cloned on every request but if I have two concurrent requests, they block on writing. What am I missing?

operator.write(file_path, bytes.clone()).await?;

EDIT: updated write call

[Xuanwo]

Hi, would you like to share the entire code? Operator::write is an async operation, so you will need to call like this to make it work:

operator.write(file_path, bytes.clone()).await?;

[dr3s]

hey thanks for the quick reply. yeah I have the await. It just was cut off in my question.

pub async fn upload_image(state: &AppState, id: &str, file_path: &str, file: Bytes) -> Result<Bytes> {
    let cursor = Cursor::new(&file);
    
    // Create JPEG decoder
    let mut decoder = JpegDecoder::new(cursor)?;
    
    // Read EXIF from decoder's metadata
    let orientation = decoder.orientation()?;

    // Create DynamicImage from decoder
    let mut img = DynamicImage::from_decoder(decoder)?;
    
    // Apply rotation based on EXIF orientation
    img = match orientation {
        Orientation::FlipHorizontal => img.fliph(),
        Orientation::Rotate180 => img.rotate180(),
        Orientation::FlipVertical => img.flipv(),
        Orientation::Rotate90FlipH => img.fliph().rotate90(),
        Orientation::Rotate90 => img.rotate90(),
        Orientation::Rotate270FlipH => img.fliph().rotate270(),
        Orientation::Rotate270 => img.rotate270(),
        _ => img  // orientation 1 or unknown
    };

    // Pre-allocate buffer for the processed image
    let mut jpeg_bytes = Vec::with_capacity(file.len());
    
    {
        let mut cursor = Cursor::new(&mut jpeg_bytes);
        // Resize the corrected image
        img.resize(800, 600, image::imageops::FilterType::Lanczos3)
           .write_to(&mut cursor, image::ImageFormat::Jpeg)?;
    }

    let bytes = Bytes::from(jpeg_bytes);

    println!("Uploading {} to Blob storage", id);
    state.storage.write(file_path, bytes.clone()).await?;
    
    println!("Uploaded {}", id);
    
    Ok(bytes)
}

[Xuanwo]

Thank you for sharing the code. We don't have internal locks for b2, so it shouldn't be related to the operator.

I have two questions:

• It seems image handling is CPU-intensive work. How long will it take? If it takes more than 20ms, it's better to move it outside the async function block to avoid blocking the entire async runtime. Please try using spawn_blocking instead.
• How does upload_image been used outside?

[dr3s]

basically only one request/thread is ever able to enter that block:

println!("Uploading {} to Blob storage", id);
    state.storage.write(file_path, bytes.clone()).await?;
    
    println!("Uploaded {}", id);

[dr3s]

Hmm, seems I need to read about rust async runtime. The image handling is very fast but I'll look into it. It's being called in an axum request handler. I had some expectation that it was thread based, so I guess I might just have the wrong expectation. I'll use spawn blocking as you suggest.

Thanks for confirming that the operator clone is fine and it should be able to use the B2 writer concurrently.