test: add parallel test for fanout cache

storage-node/src/storage_manager.rs

@@ -524,4 +524,97 @@ mod tests {
         assert!(result.is_ok());
         assert_eq!(consume_receiver(result.unwrap()).await, 2013);
     }
+
+    #[tokio::test]
+    async fn test_fanout_cach_parallel() {
+        let data_store_cache_num = 6;
+        let mut data_store_caches = Vec::new();
+        for _ in 0..data_store_cache_num {
+            let disk_cache = LruReplacer::new(1000000);
+            let mem_cache = LruReplacer::new(120000);
+
+            let tmp = tempfile::tempdir().unwrap();
+            let disk_cache_base_path = tmp.path().to_owned();
+
+            let data_store_cache = MemDiskStoreCache::new(
+                disk_cache,
+                disk_cache_base_path.display().to_string(),
+                Some(mem_cache),
+                Some(120000),
+            );
+            data_store_caches.push(data_store_cache);
+        }
+        let storage_manager = Arc::new(StorageManager::new(data_store_caches));
+
+        let request_path_bucket1 = "tests-parquet".to_string();
+        let request_path_keys1 = vec!["userdata2.parquet".to_string()];
+        let request_data1 = RequestParams::MockS3((request_path_bucket1, request_path_keys1));
+
+        let request_path_bucket2 = "tests-parquet".to_string();
+        let request_path_keys2 = vec!["userdata1.parquet".to_string()];
+        let request_data2 = RequestParams::MockS3((request_path_bucket2, request_path_keys2));
+
+        let request_path_bucket3 = "tests-text".to_string();
+        let request_path_keys3 = vec!["what-can-i-hold-you-with".to_string()];
+        let request_data3 = RequestParams::MockS3((request_path_bucket3, request_path_keys3));
+
+        let storage_manager_1 = storage_manager.clone();
+        let request_data1_1 = request_data1.clone();
+        let get_data_fut_1 =
+            tokio::spawn(async move { storage_manager_1.get_data(request_data1_1, true).await });
+
+        let storage_manager_2 = storage_manager.clone();
+        let request_data1_2 = request_data1.clone();
+        let get_data_fut_2 =
+            tokio::spawn(async move { storage_manager_2.get_data(request_data1_2, true).await });
+
+        let storage_manager_3 = storage_manager.clone();
+        let request_data3_3 = request_data3.clone();
+        let get_data_fut_3 =
+            tokio::spawn(async move { storage_manager_3.get_data(request_data3_3, true).await });
+
+        let storage_manager_4 = storage_manager.clone();
+        let request_data2_4 = request_data2.clone();
+        let get_data_fut_4 =
+            tokio::spawn(async move { storage_manager_4.get_data(request_data2_4, true).await });
+
+        let storage_manager_5 = storage_manager.clone();
+        let request_data2_5 = request_data2.clone();
+        let get_data_fut_5 =
+            tokio::spawn(async move { storage_manager_5.get_data(request_data2_5, true).await });
+
+        let storage_manager_6 = storage_manager.clone();
+        let request_data1_6 = request_data1.clone();
+        let get_data_fut_6 =
+            tokio::spawn(async move { storage_manager_6.get_data(request_data1_6, true).await });
+
+        let storage_manager_7 = storage_manager.clone();
+        let request_data3_7 = request_data3.clone();
+        let get_data_fut_7 =
+            tokio::spawn(async move { storage_manager_7.get_data(request_data3_7, true).await });
+
+        let result = join!(
+            get_data_fut_1,
+            get_data_fut_2,
+            get_data_fut_3,
+            get_data_fut_4,
+            get_data_fut_5,
+            get_data_fut_6,
+            get_data_fut_7
+        );
+        assert!(result.0.is_ok());
+        assert_eq!(consume_receiver(result.0.unwrap().unwrap()).await, 112193);
+        assert!(result.1.is_ok());
+        assert_eq!(consume_receiver(result.1.unwrap().unwrap()).await, 112193);
+        assert!(result.2.is_ok());
+        assert_eq!(consume_receiver(result.2.unwrap().unwrap()).await, 930);
+        assert!(result.3.is_ok());
+        assert_eq!(consume_receiver(result.3.unwrap().unwrap()).await, 113629);
+        assert!(result.4.is_ok());
+        assert_eq!(consume_receiver(result.4.unwrap().unwrap()).await, 113629);
+        assert!(result.5.is_ok());
+        assert_eq!(consume_receiver(result.5.unwrap().unwrap()).await, 112193);
+        assert!(result.6.is_ok());
+        assert_eq!(consume_receiver(result.6.unwrap().unwrap()).await, 930);
+    }
 }