more async points in leveldb handlers (#15)

Co-authored-by: ad <[email protected]>

Cargo.toml

@@ -11,6 +11,7 @@ bytes = "1.7.1"
 clap = { version = "4.5.17", features = ["derive"] }
 db-key = "0.1.0"
 env_logger = "0.11.5"
+futures = "0.3.30"
 gxhash = "3.4.1"
 leveldb = "0.8.6"
 log = "0.4.22"

src/main.rs

@@ -2,9 +2,13 @@ use std::{collections::HashSet, net::IpAddr, path::Path, str::FromStr, sync::Arc
 
 use bytes::Bytes;
 use clap::Parser;
+use futures::stream::{FuturesUnordered, StreamExt};
 use log::{debug, error};
 use rand::{seq::SliceRandom, SeedableRng};
-use tokio::sync::{RwLock, RwLockWriteGuard};
+use tokio::{
+    sync::{RwLock, RwLockWriteGuard},
+    task,
+};
 use warp::Filter;
 
 mod record;
@@ -214,7 +218,7 @@ async fn handle_put_record(
     let mut lock_keys = LevelDbKeyGuard::lock(&lock_keys, key.clone()).await;
     lock_keys.guard.insert(key.clone());
 
-    let record = match leveldb.get_record_or_default(&key) {
+    let record = match leveldb.get_record_or_default(&key).await {
         Ok(record) => record,
         Err(e) => {
             error!(
@@ -235,23 +239,32 @@ async fn handle_put_record(
 
     // TODO partNumber
     let replicas_volumes = record::get_volume(&key, volumes, replicas, subvolumes);
+    let mut futures = FuturesUnordered::new();
     for volume in replicas_volumes.clone() {
         let remote_replica_volume_path = record::get_remote_path(&key);
         let remote_url = format!("http://{}{}", volume, remote_replica_volume_path);
         // TODO is this value Bytes an efficient buffer?
         debug!("put_record key: {} remote_url: {}", key, remote_url);
-        match remote_put(&client, remote_url, &value).await {
+        let client_clone = client.clone();
+        let value_clone = value.clone();
+        futures.push(task::spawn(async move {
+            remote_put(client_clone, remote_url, value_clone).await
+        }));
+    }
+
+    while let Some(result) = futures.next().await {
+        match result {
             Ok(_) => (),
             Err(e) => {
                 error!(
-                    "put_record: failed to put record {} in remote replica volume {} with path {}: {}",
-                    key, volume, remote_replica_volume_path, e
+                    "put_record: failed to put record {} in remote replica: {}",
+                    key, e
                 );
 
                 // In case of error we want to mark the record as Deleted::Soft in the local leveldb
                 let record =
                     record::Record::new(record::Deleted::Soft, String::new(), replicas_volumes);
-                match leveldb.put_record(&key, record) {
+                match leveldb.put_record(&key, record).await {
                     Ok(_) => (),
                     Err(e) => {
                         error!("put_record: failed to put record {} in leveldb: {}", key, e);
@@ -269,13 +282,21 @@ async fn handle_put_record(
     }
 
     let value_md5_hash = if verify_checksums {
-        format!("{:x}", md5::compute(value))
+        match task::spawn_blocking(move || format!("{:x}", md5::compute(value))).await {
+            Ok(value_md5_hash) => value_md5_hash,
+            Err(e) => {
+                error!("put_record: failed to compute md5 hash for value: {}", e);
+                return Ok(warp::http::Response::builder()
+                    .status(warp::http::StatusCode::INTERNAL_SERVER_ERROR)
+                    .body(e.to_string()));
+            }
+        }
     } else {
         String::new()
     };
 
     let record = record::Record::new(record::Deleted::No, value_md5_hash, replicas_volumes);
-    match leveldb.put_record(&key, record) {
+    match leveldb.put_record(&key, record).await {
         Ok(_) => (),
         Err(e) => {
             error!(
@@ -294,9 +315,9 @@ async fn handle_put_record(
 }
 
 async fn remote_put(
-    client: &reqwest::Client,
+    client: reqwest::Client,
     remote_url: String,
-    value: &bytes::Bytes,
+    value: bytes::Bytes,
 ) -> anyhow::Result<()> {
     let res = client
         .put(remote_url.clone())
@@ -334,7 +355,7 @@ async fn handle_get_record(
     debug!("get_record: key: {}", key);
 
     let record = {
-        match leveldb.get_record_or_default(&key) {
+        match leveldb.get_record_or_default(&key).await {
             Ok(record) => record,
             Err(e) => {
                 error!(
@@ -440,7 +461,7 @@ async fn handle_delete_record(
     let mut lock_keys = LevelDbKeyGuard::lock(&lock_keys, key.clone()).await;
     lock_keys.guard.insert(key.clone());
 
-    let record = match leveldb.get_record_or_default(&key) {
+    let record = match leveldb.get_record_or_default(&key).await {
         Ok(record) => record,
         Err(e) => {
             error!(
@@ -467,7 +488,7 @@ async fn handle_delete_record(
         record.hash().to_string(),
         record.read_volumes().to_vec(),
     );
-    match leveldb.put_record(&key, deleted_record) {
+    match leveldb.put_record(&key, deleted_record).await {
         Ok(_) => (),
         Err(e) => {
             error!(

src/record.rs

@@ -93,7 +93,7 @@ impl LevelDb {
         Ok(Self { leveldb })
     }
 
-    pub(crate) fn put_record(&self, key: &str, record: Record) -> anyhow::Result<()> {
+    pub(crate) async fn put_record(&self, key: &str, record: Record) -> anyhow::Result<()> {
         let leveldb_key = leveldb_key_from_str(key);
         let write_options = leveldb::options::WriteOptions::new();
         self.leveldb
@@ -107,7 +107,7 @@ impl LevelDb {
         Ok(())
     }
 
-    pub(crate) fn get_record_or_default(&self, key: &str) -> anyhow::Result<Record> {
+    pub(crate) async fn get_record_or_default(&self, key: &str) -> anyhow::Result<Record> {
         let read_options = leveldb::options::ReadOptions::new();
         let leveldb_key = leveldb_key_from_str(key);