perf: persistent cache use multi-thread (#9010)

* perf: persistent cache use multi-thread

* fix: ci

Author: jerrykingxyz

crates/rspack_core/src/cache/persistent/occasion/make/module_graph.rs

@@ -132,9 +132,16 @@ pub async fn recovery_module_graph(
   let mut need_check_dep = vec![];
   let mut partial = ModuleGraphPartial::default();
   let mut mg = ModuleGraph::new(vec![], Some(&mut partial));
-  for (_, v) in storage.load(SCOPE).await? {
-    let mut node: Node =
-      from_bytes(&v, context).expect("unexpected module graph deserialize failed");
+  let nodes: Vec<_> = storage
+    .load(SCOPE)
+    .await?
+    .into_par_iter()
+    .map(|(_, v)| {
+      from_bytes::<Node, CacheableContext>(&v, context)
+        .expect("unexpected module graph deserialize failed")
+    })
+    .collect();
+  for mut node in nodes {
     for (dep, parent_block) in node.dependencies {
       mg.set_parents(
         *dep.id(),

crates/rspack_core/src/cache/persistent/snapshot/mod.rs

@@ -6,6 +6,7 @@ use std::{path::Path, sync::Arc};
 use rspack_cacheable::{from_bytes, to_bytes};
 use rspack_error::Result;
 use rspack_fs::ReadableFileSystem;
+use rspack_futures::FuturesResults;
 use rspack_paths::{ArcPath, AssertUtf8};
 use rustc_hash::FxHashSet as HashSet;
 
@@ -42,37 +43,38 @@ impl Snapshot {
   #[tracing::instrument("Cache::Snapshot::add", skip_all)]
   pub async fn add(&self, paths: impl Iterator<Item = &Path>) {
     let default_strategy = StrategyHelper::compile_time();
-    let mut helper = StrategyHelper::new(self.fs.clone());
-    // TODO use multi thread
+    let helper = StrategyHelper::new(self.fs.clone());
     // TODO merge package version file
-    for path in paths {
-      let utf8_path = path.assert_utf8();
-      // check path exists
-      if self.fs.metadata(utf8_path).is_err() {
-        continue;
-      }
-      // TODO directory path should check all sub file
-      let path_str = utf8_path.as_str();
-      if self.options.is_immutable_path(path_str) {
-        continue;
-      }
-      if self.options.is_managed_path(path_str) {
-        if let Some(v) = helper.package_version(path).await {
-          self.storage.set(
-            SCOPE,
-            path.as_os_str().as_encoded_bytes().to_vec(),
-            to_bytes::<_, ()>(&v, &()).expect("should to bytes success"),
-          );
-          continue;
+    let _ = paths
+      .map(|path| async {
+        let utf8_path = path.assert_utf8();
+        // check path exists
+        if self.fs.metadata(utf8_path).is_err() {
+          return;
         }
-      }
-      // compiler time
-      self.storage.set(
-        SCOPE,
-        path.as_os_str().as_encoded_bytes().to_vec(),
-        to_bytes::<_, ()>(&default_strategy, &()).expect("should to bytes success"),
-      );
-    }
+        // TODO directory path should check all sub file
+        let path_str = utf8_path.as_str();
+        if self.options.is_immutable_path(path_str) {
+          return;
+        }
+        if self.options.is_managed_path(path_str) {
+          if let Some(v) = helper.package_version(path).await {
+            self.storage.set(
+              SCOPE,
+              path.as_os_str().as_encoded_bytes().to_vec(),
+              to_bytes::<_, ()>(&v, &()).expect("should to bytes success"),
+            );
+            return;
+          }
+        }
+        // compiler time
+        self.storage.set(
+          SCOPE,
+          path.as_os_str().as_encoded_bytes().to_vec(),
+          to_bytes::<_, ()>(&default_strategy, &()).expect("should to bytes success"),
+        );
+      })
+      .collect::<FuturesResults<_>>();
   }
 
   pub fn remove(&self, paths: impl Iterator<Item = &Path>) {
@@ -85,16 +87,26 @@ impl Snapshot {
 
   #[tracing::instrument("Cache::Snapshot::calc_modified_path", skip_all)]
   pub async fn calc_modified_paths(&self) -> Result<(HashSet<ArcPath>, HashSet<ArcPath>)> {
-    let mut helper = StrategyHelper::new(self.fs.clone());
+    let helper = StrategyHelper::new(self.fs.clone());
+
+    let results = self
+      .storage
+      .load(SCOPE)
+      .await?
+      .iter()
+      .map(|(key, value)| async {
+        let path: ArcPath = Path::new(&*String::from_utf8_lossy(key)).into();
+        let strategy: Strategy =
+          from_bytes::<Strategy, ()>(value, &()).expect("should from bytes success");
+        let validate = helper.validate(&path, &strategy).await;
+        (path, validate)
+      })
+      .collect::<FuturesResults<_>>();
+
     let mut modified_path = HashSet::default();
     let mut deleted_path = HashSet::default();
-
-    // TODO use multi thread
-    for (key, value) in self.storage.load(SCOPE).await? {
-      let path: ArcPath = Path::new(&*String::from_utf8_lossy(&key)).into();
-      let strategy: Strategy =
-        from_bytes::<Strategy, ()>(&value, &()).expect("should from bytes success");
-      match helper.validate(&path, &strategy).await {
+    for (path, validate) in results.into_inner() {
+      match validate {
         ValidateResult::Modified => {
           modified_path.insert(path);
         }
@@ -123,7 +135,7 @@ mod tests {
     };
   }
 
-  #[tokio::test]
+  #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
   async fn should_snapshot_work() {
     let fs = Arc::new(MemoryFileSystem::default());
     let storage = Arc::new(MemoryStorage::default());

crates/rspack_core/src/cache/persistent/snapshot/strategy.rs

@@ -4,10 +4,10 @@ use std::{
   time::{SystemTime, UNIX_EPOCH},
 };
 
+use dashmap::DashMap;
 use rspack_cacheable::cacheable;
 use rspack_fs::ReadableFileSystem;
 use rspack_paths::{ArcPath, AssertUtf8};
-use rustc_hash::FxHashMap as HashMap;
 
 /// Snapshot check strategy
 #[cacheable]
@@ -38,7 +38,7 @@ pub enum ValidateResult {
 
 pub struct StrategyHelper {
   fs: Arc<dyn ReadableFileSystem>,
-  package_version_cache: HashMap<ArcPath, Option<String>>,
+  package_version_cache: DashMap<ArcPath, Option<String>>,
 }
 
 impl StrategyHelper {
@@ -60,7 +60,7 @@ impl StrategyHelper {
 
   /// get path file version in package.json
   #[async_recursion::async_recursion]
-  async fn package_version_with_cache(&mut self, path: &Path) -> Option<String> {
+  async fn package_version_with_cache(&self, path: &Path) -> Option<String> {
     if let Some(version) = self.package_version_cache.get(path) {
       return version.clone();
     }
@@ -99,15 +99,15 @@ impl StrategyHelper {
     Strategy::CompileTime(now)
   }
   /// get path file package version strategy
-  pub async fn package_version(&mut self, path: &Path) -> Option<Strategy> {
+  pub async fn package_version(&self, path: &Path) -> Option<Strategy> {
     self
       .package_version_with_cache(path)
       .await
       .map(Strategy::PackageVersion)
   }
 
   /// validate path file by target strategy
-  pub async fn validate(&mut self, path: &Path, strategy: &Strategy) -> ValidateResult {
+  pub async fn validate(&self, path: &Path, strategy: &Strategy) -> ValidateResult {
     match strategy {
       Strategy::PackageVersion(version) => {
         if let Some(ref cur_version) = self.package_version_with_cache(path).await {
@@ -172,7 +172,7 @@ mod tests {
     };
     assert!(time1 < time2);
 
-    let mut helper = StrategyHelper::new(fs.clone());
+    let helper = StrategyHelper::new(fs.clone());
     // modified_time
     assert_eq!(
       helper.modified_time(Path::new("/file1")).await,