Scheduling based on labels

[mitar]

Maybe scheduling could also work on custom labels. For example, I could run a worker saying that it runs on a machine with datasets_available and then I could have a job which would require that it is scheduled only on a machine/worker with datasets_available label.

So initially we thought of using a distributed file system mounted on every worker's machine and then have workers access datasets as needed. But that would require scheduling to know if a dataset is already moved to a worker by the distributed file system or not (to schedule jobs close to data, if possible). But I think it could be simpler if only one machine has datasets and then jobs read it as needed and store them into object store and then Ray can transport those objects around as needed. We are working anyway with datasets which can fit into memory, so it seems this is the best approach anyway. And if we get to the phase of larger datasets, locally serializing them from object store to a cache on local drive seems even much better than whole dataset parsing from scratch. (Although, it depends on compression; uncompressed images can be much larger than compressed ones.)

[atumanov]

Hi @mitar. Thanks for the suggestion. What you are referring to is an attribute-based hard constraint. Internally, our scheduling algorithms (both local and global) already support attribute-based constraints, but adding new types of attributes, at the moment, requires adding an entry to the enum type and recompiling. So label support is not user-ready in its generality. Once we add a configuration manager (like etcd), we'll look into adding more general support for labels/attributes at that time. If you'd like a specific use case supported in the meantime, we could walk through what that entails.

[mitar]

In my case, I know in advance which labels I would need. So I could deploy all Ray instances with some command line switch configuring those.

In fact, for now, I really just imagine one label: does the worker have access to datasets or not (so is a volume with datasets mounted into its container).

[robertnishihara]

@atumanov, it sounds like @mitar's use case might require some extension to the current approach.

For example, if a task requires a GPU, then it "uses up" that GPU for the duration of the task, and other tasks won't be assigned the same GPU resource.

However, if multiple tasks require a dataset, then they can all simultaneously use the dataset.

I suppose this could be done in the current framework by saying that the dataset resource is infinite, but it feels like a slightly different concept.

[atumanov]

@robertnishihara , here's a copy paste from our scheduler design doc:

Internally, we aim to support arbitrary string-valued keys with numeric (double-precision) values. In Python, attributes may be a set of predefined RAY constants exposed to the user. Attributes may be static or dynamic. Static attribute examples: total_gpus, public_ip, num_cores, num_disks. Dynamic attribute examples: local scheduler load information.

Static resources were meant to be non-consumables, while dynamic resources -- consumables. And yes, one way to approximate it for simplicity is to have non-consumables have infinite capacity.

[mitar]

Is there any workaround for now? What would be plan to implement this?

Does Ray remember on which workers a particular task failed? Then I could just be failing read dataset tasks until they get scheduled on a system with data. :-)

[mitar]

@atumanov, any progress with this?

[robertnishihara]

Should have an experimental version soon in #761 (but there will definitely be API changes down the road).

[mitar]

Superb. API changes are not an issue for us. :-)

(Hm, you do not have changelog anywhere?)

[robertnishihara]

Great! Yeah, we don't have a changelog yet.

[robertnishihara]

What do you guys think of the following specific API? Any number of custom labels could be used, and arbitrary strings could be used.

Starting Ray from the command line.

# The --resources argument is a JSON encoded dictionary.
ray start --head --resources="{'CPU': 64, 'GPU': 4, 'CustomLabel': 1}"

Starting Ray from Python.

ray.init(resources={"CPU": 64, "GPU": 4, "CustomLabel": 1})

Defining remote functions and actors.

@ray.remote(resources={"GPU": 1, "CustomLabel": 1})
def f():
    ....

@ray.remote(resources={"GPU": 1, "CustomLabel": 1})
class Actor(object):
    ...

@atumanov @mitar

[robertnishihara]

One extension that may be required in the future is to specify different "types" of resources. E.g., CPUs and GPUs behave slightly differently. Not totally sure how this would fit into the above proposal.