Repository: kedro-org/kedro
File: kedro/runner/runner.py (Python)

Code before:
         inputs have already been checked and normalized by run().
         This contains the Common pipeline execution logic using an executor.

        Args:
            pipeline: The ``Pipeline`` to run.
            catalog: An implemented instance of ``CatalogProtocol`` from which to fetch data.
            hook_manager: The ``PluginManager`` to activate hooks.
            session_id: The id of the session.
        """

        nodes = pipeline.nodes

        self._validate_catalog(catalog, pipeline)
        self._validate_nodes(nodes)
        self._set_manager_datasets(catalog, pipeline)

        load_counts = Counter(chain.from_iterable(n.inputs for n in pipeline.nodes))
        node_dependencies = pipeline.node_dependencies
        todo_nodes = set(node_dependencies.keys())
        done_nodes: set[Node] = set()
        futures = set()
        done = None
        max_workers = self._get_required_workers_count(pipeline)

        pool = self._get_executor(max_workers)
        if pool is not None:
            with pool as executor:
                while True:
                    ready = {
                        n for n in todo_nodes if node_dependencies[n] <= done_nodes
                    }
                    todo_nodes -= ready
                    for node in ready:
                        task = Task(
                            node=node,
                            catalog=catalog,
                            hook_manager=hook_manager,
                            is_async=self._is_async,
                            session_id=session_id,
                        )
                        if isinstance(executor, ProcessPoolExecutor):
                            task.parallel = True
                        futures.add(executor.submit(task))
                    if not futures:
                        if todo_nodes:
                            self._raise_runtime_error(
                                todo_nodes, done_nodes, ready, done
                            )
                        break
                    done, futures = wait(futures, return_when=FIRST_COMPLETED)
                    for future in done:

Review comment (suggestion):
Not sure if it's related to what @merelcht is saying, but for the record I was skimming this code and thought "what if `pool is None`?" (there's no `else` branch here). Just a comment from the peanut gallery.

Code after:
         inputs have already been checked and normalized by run().
         This contains the Common pipeline execution logic using an executor.

        Args:
            pipeline: The ``Pipeline`` to run.
            catalog: An implemented instance of ``CatalogProtocol`` from which to fetch data.
            hook_manager: The ``PluginManager`` to activate hooks.
            session_id: The id of the session.
        """

        nodes = pipeline.nodes

        self._validate_catalog(catalog, pipeline)
        self._validate_nodes(nodes)
        self._set_manager_datasets(catalog, pipeline)

        load_counts = Counter(chain.from_iterable(n.inputs for n in pipeline.nodes))
        node_dependencies = pipeline.node_dependencies
        todo_nodes = set(node_dependencies.keys())
        done_nodes: set[Node] = set()
        futures = set()
        done = None
        max_workers = self._get_required_workers_count(pipeline)

        pool = self._get_executor(max_workers)
        if pool is None:
            for exec_index, node in enumerate(nodes):
                try:
                    Task(
                        node=node,
                        catalog=catalog,
                        hook_manager=hook_manager,
                        is_async=self._is_async,
                        session_id=session_id,
                    ).execute()
                    done_nodes.add(node)
                except Exception:
                    self._suggest_resume_scenario(pipeline, done_nodes, catalog)
                    raise
                self._logger.info("Completed node: %s", node.name)
                self._logger.info(
                    "Completed %d out of %d tasks", len(done_nodes), len(nodes)
                )
                self._release_datasets(node, catalog, load_counts, pipeline)

            return  # Exit early since everything runs sequentially

        with pool as executor:
            while True:
                ready = {n for n in todo_nodes if node_dependencies[n] <= done_nodes}
                todo_nodes -= ready