This would reorder all bound args. I think the order of bound args should not be changed. For example,

deftest_keep_original_order(shared_ray_instance):@ray.remoteclassActor:deffoo(self,input_data):returninput_data*2defbar(self,input_data):returninput_data/2defbaz(self,*inputs):print(inputs)returnsum(inputs)a=Actor.remote()withInputNode()asinp:x=a.foo.bind(inp)y=a.bar.bind(inp)z=a.baz.bind(1,2,3,x,y,inp)dag=MultiOutputNode([y,x,z])ray.get(dag.execute(4))

z should print 1, 2, 3, x, y, inp. But the args are reordered as: x, y, 1, 2, 3, inp.

x would get 8 andy would get 2. I would expect executing this dag gives me[2, 8] as the result, same as the order specified inMultiOutputNode.

Hi Andy I had a follow up question about this. I've reworked my solution and I've gotten the functions to execute in bind order without inherently reordering the arguments to a node. However, I've run against the design decision about the ordering in which results should be returned to the user.

withInputNode()asinp:x=a.foo.bind(inp)y=a.bar.bind(inp)z=a.baz.bind(1,2,3,y,inp,x)dag=MultiOutputNode([y,x,z])# Execute the DAG with a sample input and observe the orderprint("Starting DAG execution to test task order...")result=ray.get(dag.execute(4))print("DAG execution result:",result)

In the way I wrote my first attempt, when this executes result will get [8, 2, 20] because foo returns 8, bar returns 2 and baz returns 20. Is this fundamentally the wrong approach? Should we always return these values in the position that they were passed in? This makes the most intuitive sense to me from a user perspective, but if that's the case then I'm not sure how to write unit tests for the ordering of functions that are executing remotely? Thanks for your help!

Yes, we should keep the same order that the values are passed in.

For unit tests, you can check that the order is as expected by modifying state internal to the actor, like keeping a list of tasks that were called.

I believe the nodes here are only for the current DAG node's arguments, not the entire DAG? In that case, I'm not sure that topological sort of just the current args will necessarily work.

withInputNode()asinp:x1=a.foo.bind(inp)x2=a.bar.bind(inp)y1=b.baz.bind(x1)y2=c.baz.bind(x2)dag=MultiOutputNode([y2,y1])

In this case, y2 and y1 execute on different actors and they have the same bind index, so we may visit y2 then y1. Then since it does DFS, it should execute x2, y2, x1, y1. But we want to make sure to execute x1 before x2.

Topological sort can work, but you would need to modify how the recursion is done to not do DFS I believe. The simpler way is probably to add a control dependency during DAG declaration instead of execution time, by adding the previous DAG node submitted to the same actor to the nodes found byscanner.find_nodes.