Merge pull request #27 from nuclearcat/move-to-async

Improve performance and concurrent operations handling

Author: nuclearcat

src/azure.rs

@@ -600,19 +600,15 @@ async fn azure_list_files(directory: String) -> Vec<String> {
 /// Implement Driver trait for AzureDriver
 #[async_trait]
 impl super::Driver for AzureDriver {
-    fn write_file(
+    async fn write_file(
         &self,
         filename: String,
         data: Vec<u8>,
         cont_type: String,
         owner_email: Option<String>,
     ) -> String {
         let filenameret = filename.clone();
-        /* Call async write_file_to_blob use tokio::task::block_in_place */
-        tokio::task::block_in_place(|| {
-            let rt = tokio::runtime::Runtime::new().unwrap();
-            rt.block_on(write_file_to_blob(filename, data, cont_type, owner_email));
-        });
+        write_file_to_blob(filename, data, cont_type, owner_email).await;
         filenameret
     }
     async fn write_file_streaming(
@@ -625,35 +621,21 @@ impl super::Driver for AzureDriver {
         let (_status, size) = write_file_to_blob_streaming(filename.clone(), data, cont_type, owner_email).await;
         (filename, size)
     }
-    fn tag_file(
+    async fn tag_file(
         &self,
         filename: String,
         user_tags: Vec<(String, String)>,
     ) -> Result<String, String> {
-        let rt = tokio::runtime::Runtime::new().unwrap();
-        let ret = rt.block_on(azure_set_filename_tags(filename, user_tags));
-        ret
+        azure_set_filename_tags(filename, user_tags).await
     }
-    fn get_file(&self, filename: String) -> ReceivedFile {
-        /* Call async get_file_from_blob use tokio::task::block_in_place */
-        let mut received_file = ReceivedFile {
-            original_filename: "".to_string(),
-            cached_filename: "".to_string(),
-            headers: HeaderMap::new(),
-            valid: false,
-        };
-        tokio::task::block_in_place(|| {
-            let rt = tokio::runtime::Runtime::new().unwrap();
-            received_file = rt.block_on(get_file_from_blob(filename));
-        });
-        received_file
+    async fn get_file(&self, filename: String) -> ReceivedFile {
+        get_file_from_blob(filename).await
     }
-    fn list_files(&self, directory: String) -> Vec<String> {
-        let mut ret = Vec::new();
-        tokio::task::block_in_place(|| {
-            let rt = tokio::runtime::Runtime::new().unwrap();
-            ret = rt.block_on(azure_list_files(directory));
-        });
-        ret
+    // Disabled: listing files on Azure Blob Storage is extremely slow due to
+    // the flat namespace requiring enumeration of all blobs with prefix filtering.
+    // For large containers this can take minutes and time out.
+    // The HTTP handler (ax_list_files) returns 403 Forbidden for Azure backends.
+    async fn list_files(&self, _directory: String) -> Vec<String> {
+        Vec::new()
     }
 }

src/local.rs

@@ -326,19 +326,26 @@ fn set_tags_for_local_file(
 /// Implement Driver trait for LocalDriver
 #[async_trait]
 impl super::Driver for LocalDriver {
-    fn write_file(
+    async fn write_file(
         &self,
         filename: String,
         data: Vec<u8>,
         cont_type: String,
         owner_email: Option<String>,
     ) -> String {
-        match write_file_to_local(filename.clone(), data, cont_type, owner_email) {
-            Ok(_) => filename,
-            Err(e) => {
+        let fname = filename.clone();
+        match tokio::task::spawn_blocking(move || {
+            write_file_to_local(fname, data, cont_type, owner_email)
+        }).await {
+            Ok(Ok(_)) => filename,
+            Ok(Err(e)) => {
                 eprintln!("Local storage write error: {}", e);
                 String::new()
             }
+            Err(e) => {
+                eprintln!("Local storage write task panicked: {}", e);
+                String::new()
+            }
         }
     }
 
@@ -358,19 +365,36 @@ impl super::Driver for LocalDriver {
         }
     }
 
-    fn get_file(&self, filename: String) -> ReceivedFile {
-        get_file_from_local(filename)
+    async fn get_file(&self, filename: String) -> ReceivedFile {
+        tokio::task::spawn_blocking(move || get_file_from_local(filename))
+            .await
+            .unwrap_or_else(|e| {
+                eprintln!("Local storage get_file task panicked: {}", e);
+                ReceivedFile {
+                    original_filename: String::new(),
+                    cached_filename: String::new(),
+                    headers: HeaderMap::new(),
+                    valid: false,
+                }
+            })
     }
 
-    fn tag_file(
+    async fn tag_file(
         &self,
         filename: String,
         user_tags: Vec<(String, String)>,
     ) -> Result<String, String> {
-        set_tags_for_local_file(filename, user_tags)
+        tokio::task::spawn_blocking(move || set_tags_for_local_file(filename, user_tags))
+            .await
+            .unwrap_or_else(|e| Err(format!("Local storage tag_file task panicked: {}", e)))
     }
 
-    fn list_files(&self, directory: String) -> Vec<String> {
-        list_files_in_local(directory)
+    async fn list_files(&self, directory: String) -> Vec<String> {
+        tokio::task::spawn_blocking(move || list_files_in_local(directory))
+            .await
+            .unwrap_or_else(|e| {
+                eprintln!("Local storage list_files task panicked: {}", e);
+                Vec::new()
+            })
     }
 }

src/main.rs

@@ -192,7 +192,7 @@ impl<'a> tokio::io::AsyncRead for FieldStream<'a> {
 
 #[async_trait]
 trait Driver: Send + Sync {
-    fn write_file(
+    async fn write_file(
         &self,
         filename: String,
         data: Vec<u8>,
@@ -206,13 +206,13 @@ trait Driver: Send + Sync {
         cont_type: String,
         owner_email: Option<String>,
     ) -> (String, usize);
-    fn get_file(&self, filename: String) -> ReceivedFile;
-    fn tag_file(
+    async fn get_file(&self, filename: String) -> ReceivedFile;
+    async fn tag_file(
         &self,
         filename: String,
         user_tags: Vec<(String, String)>,
     ) -> Result<String, String>;
-    fn list_files(&self, directory: String) -> Vec<String>;
+    async fn list_files(&self, directory: String) -> Vec<String>;
 }
 
 fn init_driver(driver_type: &str) -> Box<dyn Driver> {
@@ -984,7 +984,7 @@ async fn ax_get_file(
         };
 
     // IMPORTANT! Headers in cache must be stored in lowercase
-    let received_file = driver_get_file(filepath.clone());
+    let received_file = driver_get_file(filepath.clone()).await;
 
     // Release the semaphore now that the file is resolved
     drop(_permit);
@@ -1147,21 +1147,21 @@ async fn ax_get_file(
     }
 }
 
-fn driver_get_file(filepath: String) -> ReceivedFile {
+async fn driver_get_file(filepath: String) -> ReceivedFile {
     let driver_name = get_driver_type();
     let driver = init_driver(&driver_name);
-    driver.get_file(filepath)
+    driver.get_file(filepath).await
 }
 
-fn write_file_driver(
+async fn write_file_driver(
     filename: String,
     data: Vec<u8>,
     cont_type: String,
     owner_email: Option<String>,
 ) -> String {
     let driver_name = get_driver_type();
     let driver = init_driver(&driver_name);
-    driver.write_file(filename, data, cont_type, owner_email);
+    driver.write_file(filename, data, cont_type, owner_email).await;
     "".to_string()
 }
 
@@ -1222,8 +1222,13 @@ fn verify_auth_hdr(headers: &HeaderMap) -> Result<String, Option<String>> {
 
 async fn ax_list_files() -> (StatusCode, String) {
     let driver_name = get_driver_type();
+    // Listing files is disabled for Azure backend because it is too slow
+    // (flat blob namespace requires enumerating all blobs with prefix filtering).
+    if driver_name == "azure" {
+        return (StatusCode::FORBIDDEN, "Listing files is disabled for Azure storage backend".to_string());
+    }
     let driver = init_driver(&driver_name);
-    let files = driver.list_files("/".to_string());
+    let files = driver.list_files("/".to_string()).await;
     // generate nice list of files, with one file per line
     let files_str = files.join("\n");
     (StatusCode::OK, files_str)