Prefect-Ray spins up workers before the inputs for...
# prefect-community
Prefect-Ray spins up workers before the inputs for the worker tasks are ready / before the task's dependencies are complete: Hi folks, we're using Prefect and Ray to orchestrate distributed workflows. When we generate a future
x = task_1.submit()
, if we pass that future as an input to another task,
y = task_2.submit(x=x)
, Ray will spin up a worker for
with the requested resources, and this worker will wait on
. This is obviously undesirable! If we had
x1 = task1(); x2 = task2(x1=x1); x3=task3(x2=x2); ...
, all the later tasks in the pipeline would be wasting worker resources while waiting for the earlier workers to complete. We would like the worker for a task to only launch when the task inputs become ready. Is this known behavior / does anyone know how to solve this? Here's a working example:
Copy code
from prefect import flow, task
from prefect_ray import RayTaskRunner
from prefect_ray.context import remote_options
import time

def task_a():
    return 'a'

def task_b():
    return 'b'

def multi_test():
    with remote_options(num_cpus=1, num_gpus=1):
        x = task_a.submit()
    for i in range(10):
        with remote_options(num_cpus=1, num_gpus=1):

if __name__ == '__main__':
When run, we immediately get 11 workers created, per the image. CC @Muxin Fang @Haotian Li