Improve sync times (#185)

* Improve tracing output

* Improve sync times

This is a bit uglier than I would like, but unfortunately we can't ship
off the async task join to a tokio task as
we spawn so many tasks that tokio won't schedule it. So instead we spawn
the rayon scope on a separate thread first. This cuts sync times by
~25+%

* Add sanity check

This test is kind of redundant to diff_cargo, but I was using it to
sanity check that we are 100% not giving cargo any work to do.

* Fix lint

* Ignore when using git

* Update CHANGELOG

Author: Jake-Shadle

CHANGELOG.md

@@ -8,6 +8,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 <!-- next-header -->
 ## [Unreleased] - ReleaseDate
+### Fixed
+- [PR#185](https://github.com/EmbarkStudios/cargo-fetcher/pull/185) significantly improved the speed of the sync subcommand.
+
 ## [0.14.3] - 2023-08-15
 ### Fixed
 - [PR#184](https://github.com/EmbarkStudios/cargo-fetcher/pull/184) fixed submodule checkout even better this time. For real this time.

src/lib.rs

@@ -112,8 +112,8 @@ impl<'a> fmt::Display for CloudId<'a> {
 
 #[allow(dead_code)]
 pub struct GcsLocation<'a> {
-    bucket: &'a str,
-    prefix: &'a str,
+    pub bucket: &'a str,
+    pub prefix: &'a str,
 }
 
 #[allow(dead_code)]

src/sync.rs

@@ -108,7 +108,7 @@ pub async fn registry_index(
     Ok(())
 }
 
-#[tracing::instrument(skip(pkg))]
+#[tracing::instrument(level = "debug", skip_all, fields(name = krate.name, version = krate.version, rev = %rev.id))]
 fn sync_git(
     db_dir: &Path,
     co_dir: &Path,
@@ -126,7 +126,13 @@ fn sync_git(
     let crate::git::GitPackage { db, checkout } = pkg;
 
     let unpack_path = db_path.clone();
-    util::unpack_tar(db, util::Encoding::Zstd, &unpack_path)?;
+    let compressed = db.len();
+    let uncompressed = util::unpack_tar(db, util::Encoding::Zstd, &unpack_path)?;
+    debug!(
+        compressed = compressed,
+        uncompressed = uncompressed,
+        "unpacked db dir"
+    );
 
     let co_path = co_dir.join(format!("{}/{}", krate.local_id(), rev.short()));
 
@@ -143,7 +149,13 @@ fn sync_git(
     // otherwise do a checkout
     match checkout {
         Some(checkout) => {
-            util::unpack_tar(checkout, util::Encoding::Zstd, &co_path)?;
+            let compressed = checkout.len();
+            let uncompressed = util::unpack_tar(checkout, util::Encoding::Zstd, &co_path)?;
+            debug!(
+                compressed = compressed,
+                uncompressed = uncompressed,
+                "unpacked checkout dir"
+            );
         }
         None => {
             // Do a checkout of the bare clone if we didn't/couldn't unpack the
@@ -158,7 +170,7 @@ fn sync_git(
     Ok(())
 }
 
-#[tracing::instrument(level = "debug", skip(data))]
+#[tracing::instrument(level = "debug", skip_all, fields(name = krate.name, version = krate.version))]
 fn sync_package(
     cache_dir: &Path,
     src_dir: &Path,
@@ -185,7 +197,7 @@ fn sync_package(
             f.write_all(&pack_data)?;
             f.sync_all()?;
 
-            debug!(bytes = pack_data.len(), path = ?packed_path, "wrote pack file to disk");
+            debug!(bytes = pack_data.len(), "wrote pack file to disk");
             Ok(())
         },
         || -> anyhow::Result<()> {
@@ -410,15 +422,71 @@ pub async fn crates(ctx: &crate::Ctx) -> anyhow::Result<Summary> {
         });
     }
 
-    let summary = std::sync::Mutex::new(Summary {
+    let summary = std::sync::Arc::new(std::sync::Mutex::new(Summary {
         total_bytes: 0,
         bad: 0,
         good: 0,
-    });
+    }));
+
+    let (tx, rx) = crossbeam_channel::unbounded::<(Krate, Pkg)>();
+    let fs_thread = {
+        let summary = summary.clone();
+        let root_dir = root_dir.clone();
+
+        std::thread::spawn(move || {
+            let db_dir = &git_db_dir;
+            let co_dir = &git_co_dir;
+            let root_dir = &root_dir;
+            let summary = &summary;
+            rayon::scope(|s| {
+                while let Ok((krate, pkg)) = rx.recv() {
+                    s.spawn(move |_s| {
+                        let synced = match (&krate.source, pkg) {
+                            (Source::Registry(rs), Pkg::Registry(krate_data)) => {
+                                let len = krate_data.len();
+                                let (cache_dir, src_dir) = rs.registry.sync_dirs(root_dir);
+                                if let Err(err) = sync_package(
+                                    &cache_dir, &src_dir, &krate, krate_data, &rs.chksum,
+                                ) {
+                                    error!(krate = %krate, "failed to splat package: {err:#}");
+                                    None
+                                } else {
+                                    Some(len)
+                                }
+                            }
+                            (Source::Git(gs), Pkg::Git(pkg)) => {
+                                let mut len = pkg.db.len();
+
+                                if let Some(co) = &pkg.checkout {
+                                    len += co.len();
+                                }
+
+                                match sync_git(db_dir, co_dir, &krate, pkg, &gs.rev) {
+                                    Ok(_) => Some(len),
+                                    Err(err) => {
+                                        error!(krate = %krate, "failed to splat git repo: {err:#}");
+                                        None
+                                    }
+                                }
+                            }
+                            _ => unreachable!(),
+                        };
+
+                        let mut sum = summary.lock().unwrap();
+                        if let Some(synced) = synced {
+                            sum.good += 1;
+                            sum.total_bytes += synced;
+                        } else {
+                            sum.bad += 1;
+                        }
+                    });
+                }
+            });
+        })
+    };
 
     // As each remote I/O op completes, pass it off to the thread pool to do
     // the more CPU intensive work of decompression, etc
-    let (tx, rx) = crossbeam_channel::unbounded();
     while let Some(res) = tasks.join_next().await {
         let Ok(res) = res else { continue; };
 
@@ -429,59 +497,13 @@ pub async fn crates(ctx: &crate::Ctx) -> anyhow::Result<Summary> {
         }
     }
 
-    // Need to drop the sender otherwise we'll deadlock waiting for the scope
-    // to finish on the receiver
+    // Drop the sender otherwise we'll deadlock
     drop(tx);
 
-    {
-        let db_dir = &git_db_dir;
-        let co_dir = &git_co_dir;
-        let summary = &summary;
-        rayon::scope(|s| {
-            while let Ok((krate, pkg)) = rx.recv() {
-                s.spawn(move |_s| {
-                    let synced = match (&krate.source, pkg) {
-                        (Source::Registry(rs), Pkg::Registry(krate_data)) => {
-                            let len = krate_data.len();
-                            let (cache_dir, src_dir) = rs.registry.sync_dirs(root_dir);
-                            if let Err(err) =
-                                sync_package(&cache_dir, &src_dir, &krate, krate_data, &rs.chksum)
-                            {
-                                error!(krate = %krate, "failed to splat package: {err:#}");
-                                None
-                            } else {
-                                Some(len)
-                            }
-                        }
-                        (Source::Git(gs), Pkg::Git(pkg)) => {
-                            let mut len = pkg.db.len();
-
-                            if let Some(co) = &pkg.checkout {
-                                len += co.len();
-                            }
-
-                            match sync_git(db_dir, co_dir, &krate, pkg, &gs.rev) {
-                                Ok(_) => Some(len),
-                                Err(err) => {
-                                    error!("failed to splat git repo: {err:#}");
-                                    None
-                                }
-                            }
-                        }
-                        _ => unreachable!(),
-                    };
-
-                    let mut sum = summary.lock().unwrap();
-                    if let Some(synced) = synced {
-                        sum.good += 1;
-                        sum.total_bytes += synced;
-                    } else {
-                        sum.bad += 1;
-                    }
-                });
-            }
-        });
-    }
+    fs_thread.join().expect("failed to join thread");
 
-    Ok(summary.into_inner().unwrap())
+    Ok(std::sync::Arc::into_inner(summary)
+        .unwrap()
+        .into_inner()
+        .unwrap())
 }

src/util.rs

@@ -58,22 +58,32 @@ use bytes::Bytes;
 use std::io;
 
 #[tracing::instrument(level = "debug")]
-pub(crate) fn unpack_tar(buffer: Bytes, encoding: Encoding, dir: &Path) -> anyhow::Result<()> {
+pub(crate) fn unpack_tar(buffer: Bytes, encoding: Encoding, dir: &Path) -> anyhow::Result<u64> {
+    struct DecoderWrapper<'z, R: io::Read + io::BufRead> {
+        /// The total bytes read from the compressed stream
+        total: u64,
+        inner: Decoder<'z, R>,
+    }
+
     #[allow(clippy::large_enum_variant)]
     enum Decoder<'z, R: io::Read + io::BufRead> {
         Gzip(flate2::read::GzDecoder<R>),
         Zstd(zstd::Decoder<'z, R>),
     }
 
-    impl<'z, R> io::Read for Decoder<'z, R>
+    impl<'z, R> io::Read for DecoderWrapper<'z, R>
     where
         R: io::Read + io::BufRead,
     {
         fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-            match self {
-                Self::Gzip(gz) => gz.read(buf),
-                Self::Zstd(zstd) => zstd.read(buf),
-            }
+            let read = match &mut self.inner {
+                Decoder::Gzip(gz) => gz.read(buf),
+                Decoder::Zstd(zstd) => zstd.read(buf),
+            };
+
+            let read = read?;
+            self.total += read as u64;
+            Ok(read)
         }
     }
 
@@ -90,7 +100,10 @@ pub(crate) fn unpack_tar(buffer: Bytes, encoding: Encoding, dir: &Path) -> anyho
         Encoding::Zstd => Decoder::Zstd(zstd::Decoder::new(buf_reader)?),
     };
 
-    let mut archive_reader = tar::Archive::new(decoder);
+    let mut archive_reader = tar::Archive::new(DecoderWrapper {
+        total: 0,
+        inner: decoder,
+    });
 
     #[cfg(unix)]
     #[allow(clippy::unnecessary_cast)]
@@ -125,7 +138,7 @@ pub(crate) fn unpack_tar(buffer: Bytes, encoding: Encoding, dir: &Path) -> anyho
         return Err(e).context("failed to unpack");
     }
 
-    Ok(())
+    Ok(archive_reader.into_inner().total)
 }
 
 #[tracing::instrument(level = "debug")]

tests/diff_cargo.rs

@@ -200,3 +200,67 @@ async fn diff_cargo() {
         assert_diff(fetcher_root, cargo_home);
     }
 }
+
+/// Validates that a cargo sync following a fetcher sync should do nothing
+#[tokio::test]
+async fn nothing_to_do() {
+    if std::env::var("CARGO_FETCHER_CRATES_IO_PROTOCOL")
+        .ok()
+        .as_deref()
+        == Some("git")
+    {
+        // Git registry is too unstable for diffing as the index changes too often
+        return;
+    }
+
+    util::hook_logger();
+
+    let sync_dir = util::tempdir();
+    let fs_root = util::tempdir();
+
+    {
+        let (the_krates, registries) = cf::cargo::read_lock_files(
+            vec!["tests/full/Cargo.lock".into()],
+            vec![util::crates_io_registry()],
+        )
+        .unwrap();
+
+        let mut fs_ctx = util::fs_ctx(fs_root.pb(), registries);
+        fs_ctx.krates = the_krates;
+        fs_ctx.root_dir = sync_dir.pb();
+
+        let registry_sets = fs_ctx.registry_sets();
+        let the_registry = fs_ctx.registries[0].clone();
+
+        cf::mirror::registry_indices(&fs_ctx, std::time::Duration::new(10, 0), registry_sets).await;
+        cf::mirror::crates(&fs_ctx)
+            .await
+            .expect("failed to mirror crates");
+
+        fs_ctx.prep_sync_dirs().expect("create base dirs");
+        cf::sync::crates(&fs_ctx).await.expect("synced crates");
+        cf::sync::registry_index(&fs_ctx.root_dir, fs_ctx.backend.clone(), the_registry)
+            .await
+            .expect("failed to sync index");
+    }
+
+    let output = std::process::Command::new("cargo")
+        .env("CARGO_HOME", sync_dir.path())
+        .args([
+            "fetch",
+            "--locked",
+            "--manifest-path",
+            "tests/full/Cargo.toml",
+        ])
+        .stdout(std::process::Stdio::piped())
+        .stderr(std::process::Stdio::piped())
+        .output()
+        .unwrap();
+
+    let stdout = String::from_utf8(output.stdout).unwrap();
+    let stderr = String::from_utf8(output.stderr).unwrap();
+
+    if !stdout.is_empty() || !stderr.is_empty() {
+        panic!("expected no output from cargo, got:\nstdout:\n{stdout}\nstderr:{stderr}\n");
+    }
+}