[BUG Fix] Launching dependent LocalPipelineExecutors with skip_completed=False lead to waiting

src/datatrove/executor/base.py

@@ -134,17 +134,20 @@ def mark_rank_as_completed(self, rank: int):
         """
         self.logging_dir.open(f"completions/{rank:05d}", "w").close()
 
-    def get_incomplete_ranks(self, ranks=None) -> list[int]:
+    def get_incomplete_ranks(self, ranks=None, skip_completed=None) -> list[int]:
         """
             Gets a full list of ranks that are still incomplete.
             Usually faster than calling `is_rank_completed` for each task.
+            `skip_completed` can be used to override the internal attribute.
         Returns: list of ranks that are incomplete
 
         """
+        if skip_completed is None:
+            skip_completed = self.skip_completed
         completed = set(self.logging_dir.list_files("completions"))
         return list(
             filter(
-                lambda rank: not self.skip_completed or f"completions/{rank:05d}" not in completed,
+                lambda rank: not skip_completed or f"completions/{rank:05d}" not in completed,
                 ranks if ranks is not None else range(self.world_size),
             )
         )

src/datatrove/executor/local.py

@@ -98,7 +98,7 @@ def run(self):
             if not self.depends._launched:
                 logger.info(f'Launching dependency job "{self.depends}"')
                 self.depends.run()
-            while (incomplete := len(self.depends.get_incomplete_ranks())) > 0:
+            while (incomplete := len(self.depends.get_incomplete_ranks(skip_completed=True))) > 0:  # set skip_completed=True to get *real* incomplete task count
                 logger.info(f"Dependency job still has {incomplete}/{self.depends.world_size} tasks. Waiting...")
                 time.sleep(2 * 60)