Hi #CL09KU1K7. Our team would like to create a flow which spawns at least 100s of thousands sub-flows. Currently we are doing this inside of an

asyncio.gather(*[sub_flow(x) for x in params_set])

- type statement, and testing with a local self-hosted Prefect server. It's important for our use-case that developers can run the workflow themselves locally with a self-hosted server. What we observed for 10k subflows were http PoolTimeout errors. Following some tips on the forum, we increased the PREFECT_API_REQUEST_TIMEOUT and set PREFECT_API_ENABLE_HTTP2 to False. Now, we no longer see the PoolTimeout error, but it is a major issue how long it takes the server to register the sub-flows - first as task runs, then as subflow runs, and then to actually start running them. E.g. for 2.5k locations, it took 6 minutes to start actually executing the subflows, but for 5k it took 30 minutes, i.e. it doesn't increase linearly. As stated previously, we would like to support spawning at least hundreds of thousands of subflows. As this seems to be a limitation on the Prefect server side in terms of the number of incoming requests, I don't think there is much we can change in our code. What is the explanation for this? And is there any workaround we can perform without having to scale up the Prefect server with multiple API servers? It would be an issue for us in terms of developer experience if each developer has to set up a scaled server rather than just quick-starting Prefect locally. Would appreciate any advice, thank you!

Unless there's a particular reason you've chosen subflows in your design, I think tasks are a better fit for this kind of volume. Flow runs need to make requests to the server to update their state, because the server is responsible for enforcing a flow run's orchestration rules. This means every flow run effectively has to wait until the server gives it permission to proceed every time its state changes. In

prefect>=3.0.0

, task runs update their state and have their orchestration rules enforced directly in the client, and emit events which the server receives and uses to update reporting on the state of task runs. This means there's a much smaller chance a high volume of task runs is going to lead to network issues, since events are emitted over websockets.

@Kevin Grismore Thank you for your response! It immediately works much better with tasks, and the flow itself shows up as Completed in the Web GUI very quickly. It does take quite some time to update the task runs themselves in the web UI, which might be an observability issue for large workflows. What is the limiting factor? Is it network-related or rather due to my Postgres database? Are there any specific settings I can change to help here?

There are probably two relevant components that contribute to what you're seeing: 1. A flow run still shouldn't report as complete until all its enclosed task runs enter a terminal state. If you're using

.map()

then the returned object is a

list[PrefectFuture]

. A

PrefectFuture

is an extension of a regular Python

Future

in that it's a placeholder object for work that has been dispatched elsewhere (in this case a different thread) so that your code can proceed. You can call methods on it at any time later on to check on the state of that dispatched work (

future.state()

), wait for that work to finish as a blocking call (

future.wait()

), or wait for that work to finish as a blocking call and retrieve its return value (

future.result()

). We offer a utility function for waiting on the completion of a list of futures you can use,

from prefect.futures import wait

that can be used something like this:

futures = my_task.map(param_set)
wait(futures)

That'll keep your flow run running until all your tasks are complete. 2. Events are processed as a queue, so they're eventually consistent, but how long that queue takes to get through depends on a number of factors. I'm personally not that familiar with how we handle the events system in a minimal OSS setup since I spend most of my time on Prefect Cloud, but @Nate might be able to offer some insight here. One more thing to consider is what your tasks are really doing. Is it possible that they're doing some sync or CPU-bound work that's limiting the benefits of Python concurrency? Either way, a scale of hundreds of thousands of concurrent tasks may need some additional considerations on your server setup or how you choose to limit the concurrency of your work.

hi @Robyn H - i would recommend checking out https://docs.prefect.io/v3/advanced/self-hosted if you're going to want to throw a lot of work at an OSS setup as kevin mentioned, the (default in-memory) queue will become a bottleneck at scale, so if you're trying to achieve that scale on an OSS setup for whatever reason, you should probably use the redis messaging impl as outlined in that doc. you could have a reusable docker compose for folks to spin up as needed

Thank you both very much! I'll take a look at setting up my local server using Docker Compose and Redis. Just to clarify, I was waiting for the tasks to complete and could see in the worker logs that they completed, it's just that the list & state of task runs in the Web UI takes a long time to update after the workflow has completed.

that makes sense. if you’re doing work faster than the default messaging implementation can put through, then the server (and its clients like the UI) might appear to be lagging until those events are processed