Task run creation blocks the event loop about prefect HOT 2 OPEN

After looking at this issue with fresh eyes, I think everything is working as expected. The time increase is due to the overhead of calling the Prefect API for each task run, and the order of execution is the same if the @task decorator is or isn't there. I think it looks concurrent because the Python-only version is so much faster. I'm going to close this, but feel free to reopen or respond if there's something I missed!

from prefect.

Reopening because we've identified that the new engine creates task runs sequentially in async flows. See this example for reproduction:

import asyncio

from prefect import flow, task, get_run_logger


@task
async def waiter(n: int = 60):
    get_run_logger().info(f"Waiting for {n} seconds")
    await asyncio.sleep(n)


@flow
async def main(n: int = 100):
    waiters = [waiter() for _ in range(n)]
    await asyncio.gather(*waiters)


if __name__ == "__main__":
    asyncio.run(main())

This is caused by the new engine's use of SyncPrefectClient to send create task run API calls. The resolution of this issue will involve making those calls non-blocking in async contexts.

from prefect.