Manage parallel execution for workflow nodes. Provides shared logic for parallel batches and serializes Human nodes when needed.
| 7 | |
| 8 | |
| 9 | class ParallelExecutor: |
| 10 | """Manage parallel execution for workflow nodes. |
| 11 | |
| 12 | Provides shared logic for parallel batches and serializes Human nodes when needed. |
| 13 | """ |
| 14 | |
| 15 | def __init__(self, log_manager: LogManager, nodes_dict: dict): |
| 16 | """Initialize the parallel executor. |
| 17 | |
| 18 | Args: |
| 19 | log_manager: Logger instance |
| 20 | nodes_dict: Mapping of ``node_id`` to ``Node`` |
| 21 | """ |
| 22 | self.log_manager = log_manager |
| 23 | self.nodes_dict = nodes_dict |
| 24 | |
| 25 | def execute_items_parallel( |
| 26 | self, |
| 27 | items: List[Any], |
| 28 | executor_func: Callable, |
| 29 | item_desc_func: Callable[[Any], str], |
| 30 | has_blocking_func: Callable[[Any], bool] | None = None, |
| 31 | ) -> None: |
| 32 | """Execute a list of items in parallel when possible. |
| 33 | |
| 34 | Args: |
| 35 | items: Items to execute |
| 36 | executor_func: Callable that executes a single item |
| 37 | item_desc_func: Callable for logging a human-readable description |
| 38 | has_blocking_func: Optional callable to decide if an item requires serialization |
| 39 | """ |
| 40 | blocking_items, parallel_items = self._partition_blocking_items(items, has_blocking_func) |
| 41 | |
| 42 | if parallel_items: |
| 43 | self._execute_parallel_batch(parallel_items, executor_func, item_desc_func) |
| 44 | |
| 45 | if blocking_items: |
| 46 | self._execute_sequential_batch(blocking_items, executor_func, item_desc_func) |
| 47 | |
| 48 | def execute_nodes_parallel( |
| 49 | self, |
| 50 | node_ids: List[str], |
| 51 | executor_func: Callable[[str], None] |
| 52 | ) -> None: |
| 53 | """Execute a list of nodes in parallel. |
| 54 | |
| 55 | Convenience wrapper around ``execute_items_parallel`` specialized for nodes. |
| 56 | |
| 57 | Args: |
| 58 | node_ids: List of node identifiers |
| 59 | executor_func: Callable that executes a single node |
| 60 | """ |
| 61 | def item_desc_func(node_id: str) -> str: |
| 62 | return f"node {node_id}" |
| 63 | |
| 64 | def has_blocking_func(node_id: str) -> bool: |
| 65 | return False |
| 66 |