[RFC] runtime_env for actors and tasks

[edoakes]

There's been a lot of user questions and discussion around how to specify dependencies for a given job, actor, or task. There are a few different general use cases here:

1. Users want to run tasks/actors that require different or conflicting Python dependencies as part of one Ray application.
2. Users want to use Docker containers to manage dependencies (not just Python) for different tasks and actors as part of one Ray application.
3. Users want to distribute local Python modules and files to their workers/actors for rapid iteration/development.
4. Users want to easily install new Python packages as part of their development workflow (e.g., change library versions without restarting the cluster).

This proposal is to introduce a new runtime_env API that enables all of these use cases and can generalize to future worker environment-related demands.

The runtime_env will be a dictionary that can be passed as an option to actor/task creation:

f.options(runtime_env=env).remote()
Actor.options(runtime_env=env).remote()

This dictionary will include the following arguments:

• container_image (str): Require a given (Docker) container image. The image must have the same version of Ray installed.
• conda_env (str): Activates a named conda environment that the worker will run in. The environment must already exist on the node.
• files (Path): Project files and local modules to unpack in the working directory of the task/actor.
• (possible future extension) python_requirements (Union[File, List[str]]): List of Python requirements or a requirements.txt file to use to dynamically create a new conda environment.

These options should cover the all known dependency management use cases listed above.

Misc semantics:

• Any downstream tasks/actors will by default inherit the runtime_env of their parent.
• The runtime_env needs to be able to be specified on an individual actor and task basis, but for convenience it should also be able to be set in the JobConfig as a default for all tasks/actors spawned by the driver.

At this point, this RFC is primarily about the use cases and interface, not the implementation of each runtime_env option. Please comment if you believe there is a use case not covered, the UX could be improved, etc.

[richardliaw]

A couple questions --

• Where must the files be if they are file system path? Will it be only on the head node or must it exist on any arbitrary node?
• With python_requirements, we will probably also want to introduce a 'pip args' to specify a distro pull build from?

cc @tgaddair if you have any thoughts, as this might simplify some deployment work.

[ericl]

Where must the files be if they are file system path? Will it be only on the head node or must it exist on any arbitrary node?

They can be anywhere, Ray core will zip up and unpack it on the workers automatically in the worker working directory.

With python_requirements, we will probably also want to introduce a 'pip args' to specify a distro pull build from?

Yeah, probably need to scope this out one more.

[rkooo567]

This looks like a really nice improvement! For the possible serialization issues among actors and tasks, is it supposed to be just handled by users?

cc @valiantljk. Please comment if you have any feedback!

[ericl]

Perhaps we can consider cross-env method calls to be similar to cross-lang calls? https://docs.ray.io/en/master/cross-language.html

The restrictions appear to be:

1. Only primitive types are supported: float, bytes, str, ActorHandle, and arrays of such.
2. Serialization is with msgpack.

We could potentially use the same msgpack interface and extend it to support more types (e.g., numpy arrays, ObjectRef, Dicts).

Other options might include:

• Going back to Arrow-based serialization.

[edoakes]

I think the simplest solution for now will be to restrict cross-env calls to the same Ray and Python minor versions. Primitive types should then work as expected and it will be up to users to ensure they're using the same application-level library versions if they're passing non-primitive types (e.g., numpy arrays).

Arrow might indeed be a good long-term solution here.

[valiantljk]

I think this will be very useful. I'm not clear about how to support the container runtime_env though.
In my mind, it's a bit weird that the the actor or task itself should bother with the container runtime, e.g., which image to be based on. If the interface is designed in this way, then during actor creation or execution, there should be some runtime check, i.e., comparing the runtime_env with the actual container runtime.
Would it be nicer if the runtime_env is supported out of the scope of the actor creation?

But as @edoakes stated this is just for interface design. I'd like to stay tuned for the design and implementation.

[edoakes]

@valiantljk good points. We've discussed more about the implementation about the container_image env as that's actually what prompted this discussion initially. Here's a link to a document that goes more in-depth on it:
https://docs.google.com/document/d/1MbsjSye2KgYuLPPUWziU8iPTBCC_0h9EOrUIX4LFC2M/edit#

The TL;DR is that for now this will be tightly integrated with the autoscaler. The autoscaler will add new node types to satisfy scheduling constraints specified by the container_image requirement. In the future we may be able to support per-worker containers, which would allow us to remove this tight coupling of the container image and the node.

[raulchen]

Allowing different environments for different jobs makes a lot of senses.
But I'm not clear in which case we would need to use different environments for tasks/actors in the same job. Why can't just start multiple jobs?
I could imagine many potential issues:

• cross-env call could be error-prone.
• if we initialize the env unpon task/actor submission. that could be very slow.
• hard for debugging when something goes wrong as there are many different envs in a job.

[ericl]

We have a number of users asking for this, to summarize the main use cases:

• In microservice and model serving workloads, developers want to run different versions of services and models within the same serving job topology.
• In distributed ML pipelines, ML engineers want to stitch together operators written by different teams that may have different library and code dependencies.

Currently these users are not able to use Ray at all (they consider runtime env support as a hard blocker).

re: issues, the plan is as follows:

cross-env call could be error-prone.

We plan to define a restricted x-env call format (similar to xlang). Ideally this would be unified with xlang calls.

if we initialize the env unpon task/actor submission. that could be very slow.

It's no slower than launching a new job. The env can be cached on the node, similar to the job.