azure: remove blocking and convert to proper async

Initially i misinterpret Azure as partially sync, but seems everything
we use async, so we dont need to use block_in_place() and we can stay
fully async.
But local driver (which is sync) was running directly on tokio's async worker threads,
and its recommended to put it on separate blocking pool via spawn_blocking.

Signed-off-by: Denys Fedoryshchenko <denys.f@collabora.com>

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,40 +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
     }
     // 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.
-    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
+    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()
 }
 
@@ -1228,7 +1228,7 @@ async fn ax_list_files() -> (StatusCode, String) {
         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)