modified compute_fbank_librispeech.py to minimize resource utilization

egs/librispeech/ASR/local/compute_fbank_librispeech.py

@@ -69,10 +69,10 @@ def get_args():
     )
 
     parser.add_argument(
-    "--num-workers",
-    type=int,
-    default=15,
-    help="Number of worker processes for feature extraction.",
+        "--num-workers",
+        type=int,
+        default=15,
+        help="Number of worker processes for feature extraction.",
     )
 
     return parser.parse_args()
@@ -133,6 +133,7 @@ def compute_fbank_librispeech(
                 logging.info(f"{partition} already exists - skipping.")
                 continue
             logging.info(f"Processing {partition}")
+
             cut_set = CutSet.from_manifests(
                 recordings=m["recordings"],
                 supervisions=m["supervisions"],
@@ -142,20 +143,44 @@ def compute_fbank_librispeech(
                 if bpe_model:
                     cut_set = filter_cuts(cut_set, sp)
                 if perturb_speed:
-                    logging.info(f"Doing speed perturb")
+                    logging.info("Doing speed perturb")
                     cut_set = (
                         cut_set
                         + cut_set.perturb_speed(0.9)
                         + cut_set.perturb_speed(1.1)
                     )
-            cut_set = cut_set.compute_and_store_features(
-                extractor=extractor,
-                storage_path=f"{output_dir}/{prefix}_feats_{partition}",
-                # when an executor is specified, make more partitions
-                num_jobs=num_jobs if ex is None else min(num_jobs * 2, 20),
-                executor=ex,
-                storage_type=LilcomChunkyWriter,
-            )
+            # 修改這部分代碼
+            if ex is None:
+                # 為 None 的情況（本地執行）創建自定義進程池上下文
+                import multiprocessing as mp
+                from concurrent.futures import ProcessPoolExecutor
+
+                # 計算工作數
+                actual_jobs = (
+                    min(num_jobs * 2, 20) if "train" in partition else num_jobs
+                )
+
+                # 使用 forkserver 方法
+                ctx = mp.get_context("forkserver")
+                with ProcessPoolExecutor(
+                    max_workers=actual_jobs, mp_context=ctx
+                ) as local_executor:
+                    cut_set = cut_set.compute_and_store_features(
+                        extractor=extractor,
+                        storage_path=f"{output_dir}/{prefix}_feats_{partition}",
+                        executor=local_executor,
+                        storage_type=LilcomChunkyWriter,
+                    )
+            else:
+                # 分佈式環境，使用提供的執行器
+                cut_set = cut_set.compute_and_store_features(
+                    extractor=extractor,
+                    storage_path=f"{output_dir}/{prefix}_feats_{partition}",
+                    num_jobs=min(num_jobs * 2, 20),
+                    executor=ex,
+                    storage_type=LilcomChunkyWriter,
+                )
+
             cut_set.to_file(output_dir / cuts_filename)
 
 
@@ -169,5 +194,5 @@ if __name__ == "__main__":
         bpe_model=args.bpe_model,
         dataset=args.dataset,
         perturb_speed=args.perturb_speed,
-        num_workers=args.num_workers,  
+        num_workers=args.num_workers,
     )