[Feature] Accelerate the comparison for batched

Author: jan-janssen

src/executorlib/standalone/batched.py

@@ -1,26 +1,24 @@
 from concurrent.futures import Future
 
 
-def batched_futures(lst: list[Future], skip_lst: list[list], n: int) -> list[list]:
+def batched_futures(lst: list[Future], skip_set: set, n: int) -> list[list]:
     """
     Batch n completed future objects. If the number of completed futures is smaller than n and the end of the batch is
-    not reached yet, then an empty list is returned. If n future objects are done, which are not included in the skip_lst
+    not reached yet, then an empty list is returned. If n future objects are done, which are not included in the skip_set
     then they are returned as batch.
 
     Args:
         lst (list): list of all future objects
-        skip_lst (list): list of previous batches of future objects
+        skip_set (set): flat set of individual results already assigned to previous batches
         n (int): batch size
 
     Returns:
         list: results of the batched futures
     """
-    skipped_elements_lst = [item for items in skip_lst for item in items]
-
     done_lst = []
-    n_expected = min(n, len(lst) - len(skipped_elements_lst))
+    n_expected = min(n, len(lst) - len(skip_set))
     for v in lst:
-        if v.done() and v.result() not in skipped_elements_lst:
+        if v.done() and v.result() not in skip_set:
             done_lst.append(v.result())
         if len(done_lst) == n_expected:
             return done_lst

src/executorlib/task_scheduler/interactive/dependency.py

@@ -346,7 +346,7 @@ def _update_waiting_task(
             done_lst = batched_futures(
                 lst=task_wait_dict["kwargs"]["lst"],
                 n=task_wait_dict["kwargs"]["n"],
-                skip_lst=[f.result() for f in task_wait_dict["kwargs"]["skip_lst"]],
+                skip_set={item for f in task_wait_dict["kwargs"]["skip_lst"] for item in f.result()},
             )
             if len(done_lst) == 0:
                 wait_tmp_lst.append(task_wait_dict)