Merge pull request #487 from imageworks/import-configuration

Import Configuration and Efficiency

Author: rydrman

Cargo.lock

@@ -17,6 +17,22 @@ pub struct Sync {
     /// already exists
     #[clap(long)]
     pub resync: bool,
+
+    /// The total number of manifests that can be synced concurrently
+    #[clap(
+        long,
+        env = "SPFS_SYNC_MAX_CONCURRENT_MANIFESTS",
+        default_value_t = spfs::sync::DEFAULT_MAX_CONCURRENT_MANIFESTS
+    )]
+    pub max_concurrent_manifests: usize,
+
+    /// The total number of file payloads that can be synced concurrently
+    #[clap(
+        long,
+        env = "SPFS_SYNC_MAX_CONCURRENT_PAYLOADS",
+        default_value_t = spfs::sync::DEFAULT_MAX_CONCURRENT_PAYLOADS
+    )]
+    pub max_concurrent_payloads: usize,
 }
 
 impl Sync {
@@ -36,6 +52,8 @@ impl Sync {
         };
         spfs::Syncer::new(src, dest)
             .with_policy(policy)
+            .with_max_concurrent_manifests(self.max_concurrent_manifests)
+            .with_max_concurrent_payloads(self.max_concurrent_payloads)
             .with_reporter(spfs::sync::ConsoleSyncReporter::default())
     }
 }

crates/spfs-cli/common/src/args.rs

@@ -3,6 +3,7 @@
 // https://github.com/imageworks/spk
 
 use std::collections::HashSet;
+use std::sync::Arc;
 
 use futures::stream::{FuturesUnordered, TryStreamExt};
 use once_cell::sync::OnceCell;
@@ -11,11 +12,22 @@ use tokio::sync::{RwLock, Semaphore};
 use crate::{encoding, prelude::*};
 use crate::{graph, storage, tracking, Error, Result};
 
+/// The default limit for concurrent manifest sync operations
+/// per-syncer if not otherwise specified using
+/// [`Syncer::with_max_concurrent_manifests`]
+pub const DEFAULT_MAX_CONCURRENT_MANIFESTS: usize = 100;
+
+/// The default limit for concurrent payload sync operations
+/// per-syncer if not otherwise specified using
+/// [`Syncer::with_max_concurrent_payloads`]
+pub const DEFAULT_MAX_CONCURRENT_PAYLOADS: usize = 100;
+
 #[cfg(test)]
 #[path = "./sync_test.rs"]
 mod sync_test;
 
 /// Methods for syncing data between repositories
+#[derive(Copy, Clone, Debug)]
 pub enum SyncPolicy {
     /// Starting at the top-most requested item, sync and
     /// descend only into objects that are missing in the
@@ -57,14 +69,16 @@ impl SyncPolicy {
 }
 
 /// Handles the syncing of data between repositories
+///
+/// The syncer can be cloned efficiently
 pub struct Syncer<'src, 'dst, Reporter: SyncReporter = SilentSyncReporter> {
     src: &'src storage::RepositoryHandle,
     dest: &'dst storage::RepositoryHandle,
-    reporter: Reporter,
+    reporter: Arc<Reporter>,
     policy: SyncPolicy,
-    manifest_semaphore: Semaphore,
-    payload_semaphore: Semaphore,
-    processed_digests: RwLock<HashSet<encoding::Digest>>,
+    manifest_semaphore: Arc<Semaphore>,
+    payload_semaphore: Arc<Semaphore>,
+    processed_digests: Arc<RwLock<HashSet<encoding::Digest>>>,
 }
 
 impl<'src, 'dst> Syncer<'src, 'dst> {
@@ -75,11 +89,11 @@ impl<'src, 'dst> Syncer<'src, 'dst> {
         Self {
             src,
             dest,
-            reporter: SilentSyncReporter::default(),
+            reporter: Arc::new(SilentSyncReporter::default()),
             policy: SyncPolicy::default(),
-            manifest_semaphore: Semaphore::new(100),
-            payload_semaphore: Semaphore::new(100),
-            processed_digests: RwLock::new(HashSet::new()),
+            manifest_semaphore: Arc::new(Semaphore::new(DEFAULT_MAX_CONCURRENT_MANIFESTS)),
+            payload_semaphore: Arc::new(Semaphore::new(DEFAULT_MAX_CONCURRENT_PAYLOADS)),
+            processed_digests: Arc::new(RwLock::new(HashSet::new())),
         }
     }
 }
@@ -88,6 +102,27 @@ impl<'src, 'dst, Reporter> Syncer<'src, 'dst, Reporter>
 where
     Reporter: SyncReporter,
 {
+    /// Creates a new syncer pulling from the provided source
+    ///
+    /// This new instance shares the same resource pool and cache
+    /// as the one that is was cloned from. This allows them to more
+    /// safely run concurrently, and to sync more efficiently
+    /// by avoiding re-syncing the same objects.
+    pub fn clone_with_source<'src2>(
+        &self,
+        source: &'src2 storage::RepositoryHandle,
+    ) -> Syncer<'src2, 'dst, Reporter> {
+        Syncer {
+            src: source,
+            dest: self.dest,
+            reporter: Arc::clone(&self.reporter),
+            policy: self.policy,
+            manifest_semaphore: Arc::clone(&self.manifest_semaphore),
+            payload_semaphore: Arc::clone(&self.payload_semaphore),
+            processed_digests: Arc::clone(&self.processed_digests),
+        }
+    }
+
     /// Specifies how the Syncer should deal with different types of data
     /// during the sync process, replacing any existing one.
     /// See [`SyncPolicy`].
@@ -101,28 +136,29 @@ where
     /// The possible total concurrent sync tasks will be the
     /// layer concurrency plus the payload concurrency.
     pub fn with_max_concurrent_manifests(mut self, concurrency: usize) -> Self {
-        self.manifest_semaphore = Semaphore::new(concurrency);
+        self.manifest_semaphore = Arc::new(Semaphore::new(concurrency));
         self
     }
 
     /// Set how many payloads/files can be processed at once.
     ///
     /// The possible total concurrent sync tasks will be the
     /// layer concurrency plus the payload concurrency.
-    pub fn with_max_payload_concurrency(mut self, concurrency: usize) -> Self {
-        self.payload_semaphore = Semaphore::new(concurrency);
+    pub fn with_max_concurrent_payloads(mut self, concurrency: usize) -> Self {
+        self.payload_semaphore = Arc::new(Semaphore::new(concurrency));
         self
     }
 
     /// Report progress to the given instance, replacing any existing one
-    pub fn with_reporter<R>(self, reporter: R) -> Syncer<'src, 'dst, R>
+    pub fn with_reporter<T, R>(self, reporter: T) -> Syncer<'src, 'dst, R>
     where
+        T: Into<Arc<R>>,
         R: SyncReporter,
     {
         Syncer {
             src: self.src,
             dest: self.dest,
-            reporter,
+            reporter: reporter.into(),
             policy: self.policy,
             manifest_semaphore: self.manifest_semaphore,
             payload_semaphore: self.payload_semaphore,

crates/spfs/src/sync.rs

@@ -5,70 +5,10 @@
 use rstest::rstest;
 use spk_schema::foundation::spec_ops::PackageOps;
 use spk_schema::recipe;
-use spk_storage::{export_package, fixtures::*, import_package};
+use spk_storage::{export_package, fixtures::*};
 
 use crate::{BinaryPackageBuilder, BuildSource};
 
-#[rstest]
-#[tokio::test]
-async fn test_archive_io() {
-    let rt = spfs_runtime().await;
-    let spec = recipe!(
-        {
-            "pkg": "spk-archive-test/0.0.1",
-            "build": {"script": "touch /spfs/file.txt"},
-        }
-    );
-    rt.tmprepo.publish_recipe(&spec).await.unwrap();
-    let (spec, _) = BinaryPackageBuilder::from_recipe(spec)
-        .with_source(BuildSource::LocalPath(".".into()))
-        .build_and_publish(&*rt.tmprepo)
-        .await
-        .unwrap();
-    let filename = rt.tmpdir.path().join("archive.spk");
-    filename.ensure();
-    export_package(spec.ident(), &filename)
-        .await
-        .expect("failed to export");
-    let mut actual = Vec::new();
-    let mut tarfile = tar::Archive::new(std::fs::File::open(&filename).unwrap());
-    for entry in tarfile.entries().unwrap() {
-        let filename = entry.unwrap().path().unwrap().to_string_lossy().to_string();
-        if filename.contains('/') && !filename.contains("tags") {
-            // ignore specific object data for this test
-            continue;
-        }
-        actual.push(filename);
-    }
-    actual.sort();
-    assert_eq!(
-        actual,
-        vec![
-            "VERSION".to_string(),
-            "objects".to_string(),
-            "payloads".to_string(),
-            "renders".to_string(),
-            "tags".to_string(),
-            "tags/spk".to_string(),
-            "tags/spk/pkg".to_string(),
-            "tags/spk/pkg/spk-archive-test".to_string(),
-            "tags/spk/pkg/spk-archive-test/0.0.1".to_string(),
-            "tags/spk/pkg/spk-archive-test/0.0.1/3I42H3S6".to_string(),
-            "tags/spk/pkg/spk-archive-test/0.0.1/3I42H3S6.tag".to_string(),
-            "tags/spk/pkg/spk-archive-test/0.0.1/3I42H3S6/build.tag".to_string(),
-            "tags/spk/pkg/spk-archive-test/0.0.1/3I42H3S6/run.tag".to_string(),
-            "tags/spk/spec".to_string(),
-            "tags/spk/spec/spk-archive-test".to_string(),
-            "tags/spk/spec/spk-archive-test/0.0.1".to_string(),
-            "tags/spk/spec/spk-archive-test/0.0.1.tag".to_string(),
-            "tags/spk/spec/spk-archive-test/0.0.1/3I42H3S6.tag".to_string(),
-        ]
-    );
-    import_package(&filename)
-        .await
-        .expect("failed to import package");
-}
-
 #[rstest]
 #[tokio::test]
 async fn test_archive_create_parents() {

crates/spk-build/src/archive_test.rs

@@ -9,7 +9,16 @@ anyhow = "1.0"
 async-trait = "0.1"
 clap = { version = "3.2", features = ["derive", "env"] }
 colored = "2.0.0"
+futures = "0.3.9"
+spfs = { path = "../../spfs" }
 spk-cli-common = { path = '../common' }
 spk-storage = { path = '../../spk-storage' }
 spk-schema = { path = '../../spk-schema' }
+spfs-cli-common = { path = '../../spfs-cli/common' }
 tracing = "0.1.35"
+tokio = "1.20"
+
+[dev-dependencies]
+tar = "0.4.3"
+spk-build = { path = '../../spk-build' }
+rstest = "0.15.0"

crates/spk-cli/group3/Cargo.toml

@@ -4,25 +4,51 @@
 
 use anyhow::{Context, Result};
 use clap::Args;
+use futures::TryStreamExt;
+
 use spk_cli_common::{CommandArgs, Run};
-use spk_storage::{self as storage};
+
+#[cfg(test)]
+#[path = "./cmd_import_test.rs"]
+mod cmd_import_test;
 
 /// Import a previously exported package/archive
 #[derive(Args)]
 pub struct Import {
+    #[clap(flatten)]
+    sync: spfs_cli_common::Sync,
+
     /// The archive to import from
-    #[clap(name = "FILE")]
+    #[clap(name = "FILE", required = true)]
     pub files: Vec<std::path::PathBuf>,
 }
 
 #[async_trait::async_trait]
 impl Run for Import {
     async fn run(&mut self) -> Result<i32> {
+        let mut summary = spfs::sync::SyncSummary::default();
+        let local_repo = spk_storage::local_repository().await?;
+        // src and dst are the same here which is useless, but we will
+        // be using this syncer to create more useful ones for each archive
+        let syncer = self.sync.get_syncer(&local_repo, &local_repo);
         for filename in self.files.iter() {
-            storage::import_package(filename)
+            let tar_repo = spfs::storage::tar::TarRepository::open(&filename).await?;
+            let tar_repo: spfs::storage::RepositoryHandle = tar_repo.into();
+            let env_spec = tar_repo
+                .iter_tags()
+                .map_ok(|(spec, _)| spec)
+                .try_collect()
+                .await
+                .context("Failed to collect tags from archive")?;
+            tracing::info!(archive = ?filename, "importing");
+            summary += syncer
+                .clone_with_source(&tar_repo)
+                .sync_env(env_spec)
                 .await
-                .context("Import failed")?;
+                .context("Failed to sync archived data")?
+                .summary();
         }
+        tracing::info!("{:#?}", summary);
         Ok(0)
     }
 }