`is_async=True` no longer works on Jupyter

[astrojuanlu]

Description

As per title

Steps to Reproduce

1. Follow the QA notes of Refactor _run_node_async to use asyncio #4562 (comment) with Kedro 0.19.12, run session.run("adata_processing", runner=SequentialRunner(is_async=True)) on Jupyter, everything works ✔
2. Install main branch
3. Run session.run("adata_processing", runner=SequentialRunner(is_async=True)) on Jupyter, it breaks 💥

Expected Result

Continues to work as it used to

Actual Result

[03/25/25 17:53:41] INFO     Kedro project kedro-async-test                                          [session.py](file:///Users/juan_cano/Projects/QuantumBlackLabs/tmp/kedro-async-test/.venv/lib/python3.13/site-packages/kedro/framework/session/session.py):[329](file:///Users/juan_cano/Projects/QuantumBlackLabs/tmp/kedro-async-test/.venv/lib/python3.13/site-packages/kedro/framework/session/session.py#329)

                    WARNING  /Users/juan_cano/Projects/QuantumBlackLabs/tmp/kedro-async-test/.venv/ [warnings.py](file:///Users/juan_cano/.local/share/uv/python/cpython-3.13.0-macos-aarch64-none/lib/python3.13/warnings.py):[110](file:///Users/juan_cano/.local/share/uv/python/cpython-3.13.0-macos-aarch64-none/lib/python3.13/warnings.py#110)
                             lib/python3.13/site-packages/kedro/io/data_catalog.py:165:                            
                             KedroDeprecationWarning: `DataCatalog` has been deprecated and will be                
                             replaced by `KedroDataCatalog`, in Kedro 1.0.0.Currently some                         
                             `KedroDataCatalog` APIs have been retained for compatibility with                     
                             `DataCatalog`, including the `datasets` property and the                              
                             `get_datasets`, `_get_datasets`, `add`,` list`, `add_feed_dict`, and                  
                             `shallow_copy` methods. These will be removed or replaced with updated                
                             alternatives in Kedro 1.0.0. For more details, refer to the                           
                             documentation:                                                                        
                             https://docs.kedro.org/en/stable/data/index.html#kedrodatacatalog-expe                
                             rimental-feature                                                                      
                               warnings.warn(                                                                      
                                                                                                                   

                    INFO     Kedro is sending anonymous usage data with the sole purpose of improving [plugin.py](file:///Users/juan_cano/Projects/QuantumBlackLabs/tmp/kedro-async-test/.venv/lib/python3.13/site-packages/kedro_telemetry/plugin.py):[233](file:///Users/juan_cano/Projects/QuantumBlackLabs/tmp/kedro-async-test/.venv/lib/python3.13/site-packages/kedro_telemetry/plugin.py#233)
                             the product. No personal data or IP addresses are stored on our side. If              
                             you want to opt out, set the `KEDRO_DISABLE_TELEMETRY` or `DO_NOT_TRACK`              
                             environment variables, or create a `.telemetry` file in the current                   
                             working directory with the contents `consent: false`. Read more at                    
                             https://docs.kedro.org/en/stable/configuration/telemetry.html                         

[03/25/25 17:53:42] INFO     Asynchronous mode is enabled for loading and saving data                 [runner.py](file:///Users/juan_cano/Projects/QuantumBlackLabs/tmp/kedro-async-test/.venv/lib/python3.13/site-packages/kedro/runner/runner.py):[125](file:///Users/juan_cano/Projects/QuantumBlackLabs/tmp/kedro-async-test/.venv/lib/python3.13/site-packages/kedro/runner/runner.py#125)

                    WARNING  No nodes ran. Repeat the previous command to attempt a new run.          [runner.py](file:///Users/juan_cano/Projects/QuantumBlackLabs/tmp/kedro-async-test/.venv/lib/python3.13/site-packages/kedro/runner/runner.py):[334](file:///Users/juan_cano/Projects/QuantumBlackLabs/tmp/kedro-async-test/.venv/lib/python3.13/site-packages/kedro/runner/runner.py#334)

╭─────────────────────────────── Traceback (most recent call last) ────────────────────────────────╮
│ in <module>:1                                                                                    │
│                                                                                                  │
│ ❱ 1 session.run("adata_processing", runner=SequentialRunner(is_async=True))                      │
│   2                                                                                              │
│                                                                                                  │
│ /Users/juan_cano/Projects/QuantumBlackLabs/tmp/kedro-async-test/.venv/lib/python3.13/site-packag │
│ es/kedro/framework/session/session.py:399 in run                                                 │
│                                                                                                  │
│   396 │   │   │   run_params=record_data, pipeline=filtered_pipeline, catalog=catalog            │
│   397 │   │   )                                                                                  │
│   398 │   │   try:                                                                               │
│ ❱ 399 │   │   │   run_result = runner.run(                                                       │
│   400 │   │   │   │   filtered_pipeline, catalog, hook_manager, session_id                       │
│   401 │   │   │   )                                                                              │
│   402 │   │   │   self._run_called = True                                                        │
│                                                                                                  │
│ /Users/juan_cano/Projects/QuantumBlackLabs/tmp/kedro-async-test/.venv/lib/python3.13/site-packag │
│ es/kedro/runner/runner.py:129 in run                                                             │
│                                                                                                  │
│   126 │   │   │   │   "Asynchronous mode is enabled for loading and saving data"                 │
│   127 │   │   │   )                                                                              │
│   128 │   │                                                                                      │
│ ❱ 129 │   │   self._run(pipeline, catalog, hook_or_null_manager, session_id)  # type: ignore[a   │
│   130 │   │                                                                                      │
│   131 │   │   self._logger.info("Pipeline execution completed successfully.")                    │
│   132                                                                                            │
│                                                                                                  │
│ /Users/juan_cano/Projects/QuantumBlackLabs/tmp/kedro-async-test/.venv/lib/python3.13/site-packag │
│ es/kedro/runner/sequential_runner.py:72 in _run                                                  │
│                                                                                                  │
│   69 │   │   │   │   "Using synchronous mode for loading and saving data. Use the --async fla    │
│   70 │   │   │   │   "for potential performance gains. https://docs.kedro.org/en/stable/nodes    │
│   71 │   │   │   )                                                                               │
│ ❱ 72 │   │   super()._run(                                                                       │
│   73 │   │   │   pipeline=pipeline,                                                              │
│   74 │   │   │   catalog=catalog,                                                                │
│   75 │   │   │   hook_manager=hook_manager,                                                      │
│                                                                                                  │
│ /Users/juan_cano/Projects/QuantumBlackLabs/tmp/kedro-async-test/.venv/lib/python3.13/site-packag │
│ es/kedro/runner/runner.py:239 in _run                                                            │
│                                                                                                  │
│   236 │   │   │   │   │   │   hook_manager=hook_manager,                                         │
│   237 │   │   │   │   │   │   is_async=self._is_async,                                           │
│   238 │   │   │   │   │   │   session_id=session_id,                                             │
│ ❱ 239 │   │   │   │   │   ).execute()                                                            │
│   240 │   │   │   │   │   done_nodes.add(node)                                                   │
│   241 │   │   │   │   except Exception:                                                          │
│   242 │   │   │   │   │   self._suggest_resume_scenario(pipeline, done_nodes, catalog)           │
│                                                                                                  │
│ /Users/juan_cano/Projects/QuantumBlackLabs/tmp/kedro-async-test/.venv/lib/python3.13/site-packag │
│ es/kedro/runner/task.py:80 in execute                                                            │
│                                                                                                  │
│    77 │   │   if self.is_async:                                                                  │
│    78 │   │   │   import asyncio                                                                 │
│    79 │   │   │                                                                                  │
│ ❱  80 │   │   │   node = asyncio.run(                                                            │
│    81 │   │   │   │   self._run_node_async(                                                      │
│    82 │   │   │   │   │   self.node,                                                             │
│    83 │   │   │   │   │   self.catalog,                                                          │
│                                                                                                  │
│ /Users/juan_cano/.local/share/uv/python/cpython-3.13.0-macos-aarch64-none/lib/python3.13/asyncio │
│ /runners.py:190 in run                                                                           │
│                                                                                                  │
│   187 │   """                                                                                    │
│   188 │   if events._get_running_loop() is not None:                                             │
│   189 │   │   # fail fast with short traceback                                                   │
│ ❱ 190 │   │   raise RuntimeError(                                                                │
│   191 │   │   │   "asyncio.run() cannot be called from a running event loop")                    │
│   192 │                                                                                          │
│   193 │   with Runner(debug=debug, loop_factory=loop_factory) as runner:                         │
╰──────────────────────────────────────────────────────────────────────────────────────────────────╯
RuntimeError: asyncio.run() cannot be called from a running event loop

Your Environment

• Kedro version used (pip show kedro or kedro -V):
• Python version used (python -V): tested with 3.13
• Operating system and version: macOS

[ravi-kumar-pilla]

I am able to reproduce this issue. Thanks for raising @astrojuanlu

[ravi-kumar-pilla]

Hi @astrojuanlu ,

I tried a fix suggested by some in this discussion. Tested it and it is working, but there are also comments around not using nested event loops. So not really sure on other possibilities. If we have anyone with prior experience on nested event loops, please let me know

cc: @noklam @ElenaKhaustova

Thank you

[astrojuanlu]

Would something like this work?

def execute(self):
    ...
    event_loop = asyncio.get_event_loop()
    if event_loop.is_running():
        event_loop.run_until_complete(self._run_node_async(...))
    else:
        asyncio.run(...)

? (Not tested)

[ravi-kumar-pilla]

Hi @astrojuanlu ,

Thank you for the suggestion. I tried this initially and it did not work. Today, I tried with other lower level apis like the one you mentioned and call_soon and call_soon_threadsafe. Though the call_* apis do not satisfy our use case of running async task, I experimented a bit.

I have mentioned the solutions + failed solution here - https://github.com/kedro-org/kedro/pull/4613/files#diff-5d09912373c4aa0659742d26f44d8a8c1fef10587455012006d8ece2cb0e179bR106

To summarize next steps based on my understanding -

1. It is hard and not advisable to have nested event loops - there is some good discussion here
2. If we go ahead with using asyncio, for the notebook usecase or FAST API or any case where we have already running event loop, we should either use third party library nest_asyncio or start a new thread.
3. nest_asyncio seems to be archived and not maintained - https://github.com/erdewit/nest_asyncio (Not recommended). Though the usage is clean, it seems to monkey patch lot of stuff and might have issues
4. Starting a new thread (Recommended) - This seems like the most viable solution with a slight overhead of spawning a thread.

Let me know your thoughts on this.

cc: @noklam

Thank you

[astrojuanlu]

Thanks a lot for the investigation @ravi-kumar-pilla. I wasn't sure if run_until_complete would work, it's been 4 years since I last touched asyncio...

It seems to me that adopting nest_asyncio can become a liability in terms of its maintenance (or lack thereof), the general recommendation against using it, and also because it's 1 more dependency, when we're trying to move in the other direction. So my recommendation would be to either start a new thread or roll the change back, so that we can preserve the user-facing behavior.

[noklam]

I think we may have accidentally let some regression went through. I just checked the benchmark the refactoring actually slow down thread runner too, I am not sure why this is the case.

https://kedro-org.github.io/kedro-benchmark-results/#benchmark_runner.RunnerTimeSuite.time_io_bound_runner

[ravi-kumar-pilla]

Hi @noklam , I don't think that spike is related to is_async refactor. I am not sure if I am understanding correctly but the spike commit seems - f6ecca50 which is a wider refactoring done on runners. Also the thread runner does not support async mode, so asyncio does not come into play. wdyt ?