Sharing expensive resources (objects) between run in the worker

[gigaverse-oz]

Hello,

We’re exploring Prefect for orchestrating our longer tasks. We currently use Kafka and are pleased with its capabilities, but we hope Prefect could provide additional features, such as:

1. Delayed job scheduling – ideally, scheduling with Prefect’s semantics.
2. Native support for long-running tasks (30s+), which Kafka doesn’t handle well.
3. Comprehensive observability – built-in tools for tracking history and providing real-time insights.

In addition, we have these critical requirements:

1. Low latency – some tasks need to start within 0.1 seconds.
2. Horizontal scaling – our microservices run on Kubernetes (EKS).
3. Resource sharing – between jobs, such as Kafka producers or ML models, which can have significant initialization times.

Our workloads are largely I/O-based, so concurrency is essential, especially for sharing expensive resources.

We reviewed Prefect’s task and flow concepts, and each has pros and cons but none fully meet our needs:

1. Tasks – offer low latency and resource sharing, but lack scheduling, REST invocation, and robust observability.
2. Flows/Deployments – support delayed tasks and observability, but separate processes hinder resource sharing, and we've noticed significant startup latency (8-15 seconds).

Are there additional features or approach we might have missed? Do you plan to support single-threaded (asyncio-based) flows for I/O-heavy workloads? We believe this could help with flow latency.

Thank you!

[desertaxle]

Thanks for opening a discussion @gigaverse-oz! Your analysis of what Prefect currently offers for your use case is quite exhaustive. I think a few minor changes could make Prefect suitable for your use case. Before brainstorming what those changes are, can you elaborate on this question?

Do you plan to support single-threaded (asyncio-based) flows for I/O-heavy workloads?

We currently support asynchronous flows, but is this question more related to the dispatch of flow runs for a deployment? Said another way, do you want a worker that manages all flow run executions via a single event loop?

[gigaverse-oz]

Thank you for the response, @desertaxle!

Yes, having a worker that manages all flow run executions via a single event loop would allow us to share singleton objects between flows, which is exactly what we’re aiming for.

Currently, we’ve observed that each flow execution runs in a new subprocess (we verified this by logging the PIDs for each flow), outside the main thread.

Our current workflow:

1. Launch workers using {flow_object}.serve().
2. Clients trigger flow runs on demand using deployments:

run = deployments.run_deployment(
    name=deployment_name,
    parameters={"flow_input": flow_input.model_dump()},
    scheduled_time=scheduled_time,
    timeout=0,  # Returns flow run metadata immediately. 
)

Thanks again for your help!

[rmnvncnt]

Hello, @gigaverse-oz ! I don't know if this could help, but I figured flow latency using .serve() could be somehow altered by reducing PREFECT_RUNNER_POLL_FREQUENCY to 1 second, for instance. In my case, most of the observed overhead was due to the runner being idle while a flow run request was sitting in the queue.

[desertaxle]

@gigaverse-oz, you are correct that each flow execution will happen in a separate subprocess. We do that to isolate each execution, but it sounds like that's the opposite of what you want! By subclassing the ' Runner ' class, we could prove that running all flows on a single event loop is possible. Do you have an example flow where we can see if singleton objects are being shared?

Also, @rmnvncnt's suggestion to increase the poll frequency will reduce latency when starting up flow runs, but the tradeoff is increased load on the server.

[gigaverse-oz]

@rmnvncnt and @desertaxle Thanks for the latency suggestion, I’ll test it out. How are tasks executed (when using .delay() function) — using a polling mechanism? We're seeing 0.1-0.3s latency there. Could we reuse this for flow activation?

@desertaxle Here’s a very basic example (please excuse the rough code!):

import os
from prefect import flow
from loguru import logger

class TestEntryPoint:
    def __init__(self):
        self._counter = 0

    def set_counter(self):
        self._counter+=1
    
    def get_counter(self):
        return self._counter
    
global_counter = TestEntryPoint()

@flow(log_prints=False)
def test_flow(task_name: str) -> str:
    global global_counter
    logger.info(f"Got task {task_name}. Process ID: {str(os.getpid())}. Counter={global_counter.get_counter()}")
    global_counter.set_counter()
    return "This is a test"

In practice, we'd use a get_instance function or singleton decorator to return the singleton object from another file instead of using a global variable.

[desertaxle]

Tasks run with .delay are executed via task workers, who receive task runs via a WebSocket connection. Flow workers use polling to receive runs, so there can be more latency. We want to add WebSocket support to flow workers, but it will require server-side changes. It might be a good project to partner on depending on your bandwidth and appetite.

[rmnvncnt]

Just out of curiosity @gigaverse-oz, how long does it take for a flow run to start on EKS compared to a flow deployed using .serve()? I'm in a bit of a similar situation where I want to use an API to run a flow and retrieve the result for that run quickly, so latency is also an issue. I'm not expecting 0.1 startup times, but the shorter the better.

[gigaverse-oz]

Hi @rmnvncnt,

If you're using the Prefect K8S worker, starting a flow on EKS usually takes around 30+ seconds since it needs to deploy a new k8s job and k8s pod, and this timing depends on your image size and worker startup time. We plan to use the .serve() function with pre-deployed Prefect Process workers that wait for tasks, while the K8S auto-scheduler scales them based on CPU/memory. In our tests (locally, with minikube, not the production EKS), this type of worker has a latency of about 3-10 seconds, though it might be a bit faster on EKS.

One challenge with auto-scaling is that it may terminate pods arbitrarily during scale-down, which can cause flows handled by those workers to fail. To handle this, we've developed our own framework for critical, long-running tasks (30-60 minutes) that can't be interrupted. We've managed to achieve start times as low as 1-2 seconds, with potential improvements to 0.1 seconds when there's a clear event trigger. The scale-down happens automatically when tasks finish, and pods are closed as they complete.

Please note that these numbers are based on my own experience, and the Prefect team may have more reliable benchmarks for production environments.

Hope this helps!

P.S.
If you don't need to schedule your "workload", you can simply use tasks and have Task Workers listening to it. I observed 0.1 seconds latency there (on local machine). But you don't get the full abilities flows and deployments give you.

[rmnvncnt]

Thanks a lot, this is very insightful! I'm currently running long running tasks on auto-scaling ECS-over-EC2. Scale-down has not been an issue (yet) but startup times are pretty long. I was actually considering EKS as an improvement, your use case is really interesting.

When you say that you scale up multiple .serve() processes, do you mean that you can have multiple .serve() workers referring to a same deployment? For instance, if you have flow.serve(name="deployemnt-1") on k8s job n°1 and k8s job n°2, both will pick up flow runs in the deployment-1 queue?

[desertaxle]

@rmnvncnt you can have multiple .serve processes for the same deployment, and they will all pull scheduled flow runs from that deployment. One way to do this is to build a Docker image with your .serve script inside and run multiple copies of that.

@gigaverse-oz did you build your framework for long-running tasks on Prefect? I'm wondering because I'd like to improve infrastructure management for served flows, and it sounds like I could learn something from your implementation.

[gigaverse-oz]

@desertaxle No, but I don't mind share the ideas and implementations, weaknesses with you. We can schedule a VC.

[rmnvncnt]

What I would do in a first (hacky) attempt, would be to have a trigger/automation watching the amount of late runs on a given deployment, then create a new container once a threshold is reached. Finally, it would kill all but one container when the queue is empty for some time.