I want to execute three subflows concurrently. As ...
# ask-community
r
I want to execute three subflows concurrently. As soon as one of the subflows completes, you want to print its result and gracefully terminate the remaining running tasks in the Prefect flow, ensuring that there are no lingering tasks executing in the background. The goal is to achieve this without causing any errors in the execution flow. without using asynio.gather() this is my code is below and here i executed 3 subflows in 3 tasks without using gather(). i want return when i got result from any task and stop the remaining parallel tasks. Is it possible ? import asyncio from prefect import flow, task from prefect.task_runners import ConcurrentTaskRunner from prefect.exceptions import CancelledRun @flow async def subflow_1(): await asyncio.sleep(20) # Simulate work print("Subflow 1 completed") return "Result from subflow 1" @flow async def subflow_2(): await asyncio.sleep(30) # Simulate work print("Subflow 2 completed") return "Result from subflow 2" @flow async def subflow_3(): await asyncio.sleep(40) # Simulate work print("Subflow 3 completed") return "Result from subflow 3" @task async def task1(): return await subflow_1() # Run the subflow and return the result @task async def task2(): return await subflow_2() @task async def task3(): return await subflow_3() @flow(task_runner=ConcurrentTaskRunner()) async def main_flow(): # Submit tasks without awaiting them individually to ensure parallel execution task_1 = await task1.submit() task_2 = await task2.submit() task_3 = await task3.submit() if name == "main": asyncio.run(main_flow())