Wait for upstream object refs before starting task

[j-tr]

Closes #80

Ray has built-in functionality to wait for the execution of a remote function until all input object refs are available as long as these object refs are direct parameters of the function. If wrapped into other objects, this doesn't work.

Prefect features are located in call.keywords.parameters and therefore not directly passed to the remote function.

By extracting them in exchange_prefect_for_ray_future and explicitly passing them as args to the remote function, in addition to the wrapped parameters, the expected behavior is restored.

Example

Screenshots

Checklist

• References any related issue by including "Closes #" or "Closes ".
  • If no issue exists and your change is not a small fix, please create an issue first.
• Includes tests or only affects documentation.
• Passes pre-commit checks.
  • Run pre-commit install && pre-commit run --all locally for formatting and linting.
• Includes screenshots of documentation updates.
  • Run mkdocs serve view documentation locally.
• Summarizes PR's changes in CHANGELOG.md

[pcmoritz]

Thanks a lot for submitting this PR, it indeed looks like the right way to do this to me. The tasks will be submitted and the Ray scheduler will wait until they get executed until all args are available -- tasks do not get executed prematurely:

[j-tr]

@ahuang11 does this look good to you? what would be needed in order to merge this?

[zanieb]

Hey @j-tr — they don't maintain this repository anymore. @desertaxle and I will review.

[achordia20]

@madkinsz any ideas on when this fix might make it into a release? Not sure we can responsibly leverage ray with prefect without this so curious on the timeline.

[zanieb]

@achordia20 this is failing all of the test suites for what looks like an unrelated reason so it looks like some maintenance will need to be done first.

Edit: Investigating in #84

[zanieb]

Can we rename *args to *upstream_ray_obj_refs and add a note about what's going on here?

[zanieb]

We may be able to just extract upstream ray object refs in this step then do the exchange with the resolved objects in _run_prefect_task — seems feasible to do as a follow-up though.

[j-tr]

true. however, we need to exchange the prefect futures with something else as they are not serializable and cannot be passed to a ray remote function. I think we would need in any case some sort of placeholder object which is then exchanged by the resolved object in _run_prefect_task. but a ray object ref is already a good candidate for such a placeholder and saves us from yet another indirection.