refactor(pm): use libdeflater for gzip decompression

crates/pm/Cargo.toml

@@ -11,8 +11,8 @@ path = "src/main.rs"
 
 [dependencies]
 anyhow = { workspace = true }
-async-compression = { version = "0.3", features = ["gzip", "tokio"] }
 atty = "0.2"
+bytes = "1.11.0"
 chrono = { version = "0.4", features = ["serde"] }
 clap = { workspace = true }
 colored = "2.1"
@@ -25,6 +25,7 @@ futures = { workspace = true }
 glob = "0.3.1"
 indicatif = "0.17.8"
 libc = "0.2"
+libdeflater = "1.25"
 once_cell = "1.19"
 owo-colors = { workspace = true }
 petgraph = "0.6"
@@ -44,8 +45,7 @@ tempfile = "3.15.0"
 term_size = "0.3"
 tokio-fs-ext = { workspace = true }
 tokio-retry = "0.3"
-tokio-tar = "0.3.1"
-tokio-util = { version = "0.7", features = ["io"] }
+tokio-util = { version = "0.7" }
 toml = { workspace = true }
 tracing = { workspace = true }
 tracing-appender = "0.2"

crates/pm/src/main.rs

@@ -250,25 +250,9 @@ enum Commands {
     },
 }
 
-fn main() {
-    let parallelism = std::thread::available_parallelism()
-        .map(|n| n.get())
-        .unwrap_or(4);
-    // Minimum 12 blocking threads for CI environments with fewer cores
-    let blocking_threads = (parallelism * 2).max(12);
-
-    let result = tokio::runtime::Builder::new_multi_thread()
-        .enable_all()
-        .worker_threads(parallelism)
-        .max_blocking_threads(blocking_threads)
-        .thread_name("utoo-worker")
-        .on_thread_stop(|| {})
-        .on_thread_park(|| {})
-        .build()
-        .expect("failed to build tokio runtime")
-        .block_on(async_main());
-
-    if let Err(e) = result {
+#[tokio::main]
+async fn main() {
+    if let Err(e) = async_main().await {
         tracing::error!("{:#}", e);
         if let Some(log_path) = get_log_file_path() {
             eprintln!("Full logs saved to: {}", log_path.display());

crates/pm/src/service/install.rs

@@ -18,6 +18,7 @@ use crate::model::package::PackageInfo;
 use crate::service::rebuild::RebuildService;
 use crate::util::cache::get_cache_dir;
 use crate::util::cloner::clone_package;
+use crate::util::config::get_manifests_concurrency_limit;
 use crate::util::downloader::download;
 use crate::util::linker::link;
 use crate::util::logger::{PROGRESS_BAR, finish_progress_bar, log_progress, start_progress_bar};
@@ -26,8 +27,6 @@ use utoo_ruborist::compat::{is_cpu_compatible, is_os_compatible};
 
 use super::binary::update_package_binary;
 
-static CONCURRENT_LIMIT: usize = 40;
-
 /// Check if a package should be omitted based on omit config
 fn should_omit_package(package: &Package, omit: &std::collections::HashSet<OmitType>) -> bool {
     if omit.is_empty() {
@@ -274,7 +273,7 @@ pub async fn install_packages(
     groups: &HashMap<usize, Vec<(std::string::String, Package)>>,
     cache_dir: &Path,
     cwd: &Path,
-    semaphore: Arc<Semaphore>,
+    download_semaphore: Arc<Semaphore>,
 ) -> Result<()> {
     // clean unused deps
     clean_deps(groups, cwd).await?;
@@ -342,18 +341,19 @@ pub async fn install_packages(
                     let cache_flag_path = cache_dir.join(format!("{name}/{version}/_resolved"));
                     let cwd_clone = cwd.to_path_buf();
                     let should_resolve = !crate::fs::try_exists(&cache_flag_path).await?;
-                    let semaphore = Arc::clone(&semaphore);
+                    let download_semaphore = Arc::clone(&download_semaphore);
 
                     // Check if this is an optional dependency
                     let is_optional =
                         package.optional == Some(true) || package.dev_optional == Some(true);
 
                     let task = tokio::spawn(async move {
-                        let _permit = semaphore
-                            .acquire()
-                            .await
-                            .expect("semaphore should not be closed");
                         if should_resolve {
+                            // Limit download concurrency, but not clone
+                            let _permit = download_semaphore
+                                .acquire()
+                                .await
+                                .expect("download semaphore should not be closed");
                             tracing::debug!("Downloading {path} to {name}");
                             match download(&resolved, &cache_path).await {
                                 Ok(_) => {
@@ -497,11 +497,11 @@ impl InstallService {
             PROGRESS_BAR.set_length(package_lock.packages.len() as u64);
         }
 
-        // Set concurrent limit for package installation
-        tracing::debug!("Setting concurrent limit to {CONCURRENT_LIMIT}");
-        let semaphore = Arc::new(Semaphore::new(CONCURRENT_LIMIT));
+        // Use same concurrency limit as manifests for download, clone is unlimited
+        let download_limit = get_manifests_concurrency_limit().await;
+        let download_semaphore = Arc::new(Semaphore::new(download_limit));
 
-        install_packages(&groups, &cache_dir, root_path, semaphore)
+        install_packages(&groups, &cache_dir, root_path, download_semaphore)
             .await
             .context("Failed to install packages")?;