new threading

Author: syphar

src/index.rs

@@ -1,12 +1,10 @@
-use crate::{
-    error::Result,
-    utils::{report_error, spawn_blocking},
-};
-use anyhow::Context;
+use crate::{error::Result, utils::report_error};
+use anyhow::{Context as _, anyhow};
 use crates_index_diff::{Change, gix, index::diff::Order};
 use std::{
     path::{Path, PathBuf},
     sync::{Arc, Mutex, atomic::AtomicBool},
+    thread,
 };
 use tokio::process::Command;
 
@@ -15,7 +13,7 @@ pub struct Index {
     path: PathBuf,
     repository_url: Option<String>,
     // NOTE: we can use a sync mutex here, because we're only locking it
-    // inside spawn_blocking calls, so the mutex lock won't ever be held
+    // inside handle_in_thread calls, so the mutex lock won't ever be held
     // across await-points.
     #[allow(clippy::disallowed_types)]
     index: Arc<Mutex<crates_index_diff::Index>>,
@@ -34,20 +32,17 @@ impl Index {
             .map(|url| crates_index_diff::index::CloneOptions { url: url.clone() })
             .unwrap_or_default();
 
-        let index = spawn_blocking({
-            let path = path.clone();
-            move || {
-                Ok(Arc::new(Mutex::new(
-                    #[allow(clippy::disallowed_types)]
-                    crates_index_diff::Index::from_path_or_cloned_with_options(
-                        &path,
-                        gix::progress::Discard,
-                        &AtomicBool::default(),
-                        clone_options,
-                    )
-                    .context("initialising registry index repository")?,
-                )))
-            }
+        let index = handle_in_thread(|| {
+            Ok(Arc::new(Mutex::new(
+                #[allow(clippy::disallowed_types)]
+                crates_index_diff::Index::from_path_or_cloned_with_options(
+                    &path,
+                    gix::progress::Discard,
+                    &AtomicBool::default(),
+                    clone_options,
+                )
+                .context("initialising registry index repository")?,
+            )))
         })
         .await?;
 
@@ -80,18 +75,11 @@ impl Index {
         &self,
         order: Order,
     ) -> Result<(Vec<Change>, gix::hash::ObjectId)> {
-        spawn_blocking({
-            let index = self.index.clone();
-            move || {
-                let index = index.lock().unwrap();
-                index
-                    .peek_changes_with_options(
-                        gix::progress::Discard,
-                        &AtomicBool::default(),
-                        order,
-                    )
-                    .map_err(Into::into)
-            }
+        handle_in_thread(|| {
+            let index = self.index.lock().unwrap();
+            index
+                .peek_changes_with_options(gix::progress::Discard, &AtomicBool::default(), order)
+                .map_err(Into::into)
         })
         .await
     }
@@ -106,12 +94,9 @@ impl Index {
     }
 
     pub async fn set_last_seen_reference(&self, to: gix::hash::ObjectId) -> Result<()> {
-        spawn_blocking({
-            let index = self.index.clone();
-            move || {
-                let index = index.lock().unwrap();
-                index.set_last_seen_reference(to).map_err(Into::into)
-            }
+        handle_in_thread(|| {
+            let index = self.index.lock().unwrap();
+            index.set_last_seen_reference(to).map_err(Into::into)
         })
         .await
     }
@@ -126,6 +111,27 @@ impl Index {
     }
 }
 
+/// Move the execution of a blocking function into a separate, new thread.
+///
+/// Only for long-running / expensive operations that would block the async runtime or its
+/// blocking workerpool.
+async fn handle_in_thread<R, F>(f: F) -> Result<R>
+where
+    F: FnOnce() -> Result<R> + Send,
+    R: Send,
+{
+    thread::scope(|scope| {
+        let handle = thread::Builder::new()
+            .name("crates-index-diff".into())
+            .spawn_scoped(scope, move || f())
+            .context("couldn't spawn worker thread")?;
+
+        handle
+            .join()
+            .map_err(|_| anyhow!("failed to join worker thread"))?
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;