Offload gitoxide to a tokio blocking thread

Author: ColinKinloch

gitlab-runner/examples/demo-runner.rs

@@ -12,6 +12,7 @@ use gitlab_runner::{
 };
 use serde::Deserialize;
 use tokio::signal::unix::{signal, SignalKind};
+use tokio_util::sync::CancellationToken;
 use tracing::{debug, info};
 use tracing_subscriber::prelude::*;
 use url::Url;
@@ -141,13 +142,16 @@ impl Run {
                     };
                     let reference = p.next();
 
+                    let cancel_token = CancellationToken::new();
                     let repo_path = clone_git_repository(
                         self.job.build_dir(),
                         url,
                         reference,
                         std::iter::empty::<&str>(),
                         Some(1),
+                        cancel_token,
                     )
+                    .await
                     .map_err(|e| outputln!("Failed to checkout repo: {}", e.to_string()))?;
 
                     std::fs::rename(repo_path, path)
@@ -160,9 +164,11 @@ impl Run {
                         Some(path) => self.job.build_dir().join(path),
                         _ => self.job.build_dir().to_path_buf(),
                     };
+                    let cancel_token = CancellationToken::new();
                     let temp_repo_path = self
                         .job
-                        .clone_git_repository()
+                        .clone_git_repository(cancel_token)
+                        .await
                         .map_err(|e| outputln!("Failed to checkout repo: {}", e.to_string()))?;
 
                     outputln!("Checked out to {:?}", temp_repo_path);

gitlab-runner/src/client.rs

@@ -309,6 +309,10 @@ pub enum GitCheckoutError {
     GitWorktreeCheckout(#[from] gix::worktree::state::checkout::Error),
     #[error(transparent)]
     GitHeadPeel(#[from] gix::head::peel::Error),
+    #[error(transparent)]
+    ThreadJoinError(#[from] tokio::task::JoinError),
+    #[error("Checkout cancelled")]
+    Cancelled,
     #[error("Job does not allow fetch")]
     FetchNotAllowed,
     #[error("Failed to find commit")]

gitlab-runner/src/job.rs

@@ -10,6 +10,7 @@ use std::path::{Path, PathBuf};
 use std::sync::{Arc, Mutex};
 use tokio::io::AsyncWrite;
 use tokio_retry2::strategy::{jitter, FibonacciBackoff};
+use tokio_util::sync::CancellationToken;
 use tracing::info;
 
 use crate::client::Error as ClientError;
@@ -370,7 +371,10 @@ impl Job {
     ///
     /// This creates a new path for the repo as gitoxide deletes it on failure.
     #[allow(clippy::result_large_err)]
-    pub fn clone_git_repository(&self) -> Result<PathBuf, GitCheckoutError> {
+    pub async fn clone_git_repository(
+        &self,
+        cancel_token: CancellationToken,
+    ) -> Result<PathBuf, GitCheckoutError> {
         if !self.response.allow_git_fetch {
             return Err(GitCheckoutError::FetchNotAllowed);
         }
@@ -381,6 +385,8 @@ impl Job {
             Some(&self.response.git_info.sha),
             &self.response.git_info.refspecs,
             Some(self.response.git_info.depth),
+            cancel_token,
         )
+        .await
     }
 }

gitlab-runner/src/lib.rs

@@ -33,6 +33,8 @@ use std::fmt::Write;
 use std::num::NonZeroU32;
 use std::path::{Path, PathBuf};
 use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
 use tokio::time::{sleep, Duration};
 use tracing::warn;
 use url::Url;
@@ -460,27 +462,65 @@ impl Runner {
     }
 }
 
-// TODO: Should clone_git_repository be async
-//   gitoxide doesn't currently have async implemented for http backend
+// TODO: Should we re-export gix::progress to provide a means to monitor progress
 // TODO: Is "clippy::result_large_err" the best solution to GitCheckoutError
 
+/// Fetch and checkout a given worktree on a thread
+///
+/// See: [clone_git_repository_sync]
+pub async fn clone_git_repository(
+    parent_path: &Path,
+    repo_url: &str,
+    head_ref: Option<&str>,
+    refspecs: impl IntoIterator<Item = impl AsRef<str>>,
+    depth: Option<u32>,
+    cancel_token: CancellationToken,
+) -> Result<PathBuf, GitCheckoutError> {
+    let parent_path = parent_path.to_owned();
+    let repo_url = repo_url.to_owned();
+    let head_ref = head_ref.map(|a| a.to_owned());
+    let should_interrupt: Arc<AtomicBool> = Default::default();
+    let should_interrupt_cancel = should_interrupt.clone();
+    let refspecs: Vec<_> = refspecs
+        .into_iter()
+        .map(|s| s.as_ref().to_owned())
+        .collect();
+    // offload the clone operation to one of the tokio runtimes blocking threads
+    tokio::select! {
+        result = tokio::task::spawn_blocking(move || {
+            clone_git_repository_sync(
+                &parent_path,
+                &repo_url,
+                head_ref.as_deref(),
+                refspecs,
+                depth,
+                &should_interrupt,
+            )
+        }) => result?,
+        _ = cancel_token.cancelled() => {
+            should_interrupt_cancel.store(true, Ordering::SeqCst);
+            Err(GitCheckoutError::Cancelled)
+        }
+    }
+}
+
 /// Fetch and checkout a given worktree
 ///
 /// This creates a new path for the repo as gitoxide deletes it on failure.
 #[allow(clippy::result_large_err)]
-pub fn clone_git_repository(
+pub fn clone_git_repository_sync(
     parent_path: &Path,
     repo_url: &str,
     head_ref: Option<&str>,
     refspecs: impl IntoIterator<Item = impl AsRef<str>>,
     depth: Option<u32>,
+    should_interrupt: &AtomicBool,
 ) -> Result<PathBuf, GitCheckoutError> {
     let repo_dir = tempfile::Builder::new()
         .prefix("repo_")
         .tempdir_in(parent_path)?;
 
     // TODO: Should we expose the ability to interrupt / report progress
-    let should_interrupt = AtomicBool::new(false);
     let mut progress = progress::Discard;
 
     // TODO: Is Options::isolated correct here?
@@ -543,7 +583,7 @@ pub fn clone_git_repository(
     }
 
     let checkout_progress = progress.add_child("checkout".to_string());
-    let (checkout, _outcome) = fetch.fetch_then_checkout(checkout_progress, &should_interrupt)?;
+    let (checkout, _outcome) = fetch.fetch_then_checkout(checkout_progress, should_interrupt)?;
 
     let repo = checkout.persist();
 
@@ -596,7 +636,7 @@ pub fn clone_git_repository(
         repo.objects.clone().into_arc()?,
         &files_progress,
         &bytes_progress,
-        &should_interrupt,
+        should_interrupt,
         Default::default(),
     )?;