[Workflows][DAG] Are node results/states cached?

Ray Core Ray Workflows
1

Consider the following example:


@ray.remote
def t1():
    print('Started t1')
    time.sleep(5)
    print('Finished t1')
    return ['t1']


@ray.remote
def t2(t1):
    return ['t2'] + t1


@ray.remote
def t3(t1):
    return ['t3'] + t1


def test_dep_tasks():
    with ray.init(address='auto'):
        _t1 = t1.bind()
        _t2 = t2.bind(_t1)
        _t3 = t3.bind(_t1)

        r1 = workflow.run_async(_t2)
        r2 = workflow.run_async(_t3)

        print(ray.get(r1))
        print(ray.get(r2))

Will t1 be called and executed twice in this case (as a dependency for both t2 and t3)? Or Ray knows that t1 is already being executed as a part of another DAG (t2) and the current DAG (t3) just waits for t1 to return? If not, how can I achieve this behaviour, without re-executing t1? Thanks!

2

Hey @dirtyValera , good question.

So from the example above, I think t1 will be executed twice.

Try running this:

import ray
from ray import workflow


@ray.remote
def t1():
    print("t1")
    return ["t1"]


@ray.remote
def t2(t1):
    print("t2")
    return ["t2"] + t1


@ray.remote
def t3(t1):
    print("t3")
    return ["t3"] + t1


_t1 = t1.bind()
_t2 = t2.bind(_t1)
_t3 = t3.bind(_t1)

workflow.run(_t2)
workflow.run(_t3)

However, I am not entirely sure if this is any native API from workflow that could skip execution of a node that’s overlapped with another DAG
cc @yic

Some random ideas, If you want to avoid executing some code, I wonder if you could use object store as a cache, so you could put the executed result into object store with ray.put and get from it the next time?

3

@dirtyValera t1 will be executed twice since they come from different workflows. There is no guarantee that t1 will always produce the same results.

For your case, you can try two things:

  • use a gather step and accept t2 and t3 as input and output a pair, you’ll get r1 and r2 in the end.
  • or use named task (Getting Started — Ray 2.3.0) You’ll be able to ray.get the specific step result.
1 Like