Is there a way to define a Worker Resource as infinite?

[aa1371]

Inspired by this issue in airflow for simulating airflow queues in the DaskExecutor using Worker Resources:
apache/airflow#16739

Is there a way to define an "infinite" worker resource? I think the answer is no, based on what I see regarding available_resources and resource_restrictions in worker.py. If that assumption is correct, is there any reason not to allow the definition of an "infinite" resource?

[quasiben]

What would it mean to to specify a worker as infinite ?

[aa1371]

Was thinking for things that aren't quantifiably limited. E.g. the existence of an env var, or simply to use the resource as a tag to control the subset of your cluster that a task can be submitted to, in a manner analogous to celery queues (not calling this feature a queue, but just mean that it can mimic this functionality). (see linked airflow issue above for more details on a potential use case)

In any case I'm realizing now that for cases like this the resource can likely just be set to a limit of 1 on the worker, it seems like if you have the following set up:

# 1 dask worker
$ dask-worker <address> --resources="TAG1=1"

# submit 2 tasks requesting the same resource
def task():
    time.sleep(10)

client.submit(task, pure=False, resources={'TAG1': 1})
client.submit(task, pure=False, resources={'TAG1': 1})

The scheduler is able to assign both tasks to the worker without issue. I assumed that if the requested resource on a worker was "depleted" from the workers POV, that the scheduler wouldn't schedule another task, requesting that resource, on that worker until the resource was "released". But after some quick testing seems that isn't the case

So my new understanding is that the resource limit is a per instance threshold, not a cumulative threshold, and for the use cases I mentioned in the first paragraph (resources as tags) it is sufficient to define a worker resource limit of 1 as long as you always submit the task with resources={resource: 1}. Is that about right?

[aa1371]

Ah never mind. I see now, it is cumulative, however, I usually start my workers with nthreads=1, so I don't ever have more than 1 task processing on a worker at one time. However, with more threads I see that a limit of 1 prevents multiple tasks from running in separate threads on a worker at once.

So in the general case an infinite limit (or more specifically a limit == nthreads on the worker) would still be desirable, although in my use case it wouldn't be necessary.

[mrocklin]

Setting it to 1e99 might be effectively the same, and is already implemented today

…

On Fri, Jul 2, 2021 at 9:58 AM aa1371 ***@***.***> wrote: Was thinking for things that aren't quantifiably limited. E.g. the existence of an env var, or simply to use the resource as a tag to control the subset of your cluster that a task can be submitted to, in a manner analogous to celery queues (not calling this feature a queue, but just mean that it can mimic this functionality). In any case I'm realizing now that for cases like this the resource can likely just be set to a limit of 1 on the worker, it seems like if you have the following set up: # 1 dask worker $ dask-worker <address> --resources="TAG1=1" # submit 2 tasks requesting the same resource def task(): time.sleep(10) client.submit(task, pure=False, resources={'TAG1': 1}) client.submit(task, pure=False, resources={'TAG1': 1}) The scheduler is able to assign both tasks to the worker without issue. I assumed that if the requested resource on a worker was "depleted" from the workers POV, that the scheduler wouldn't schedule another task, requesting that resource, on that worker. But after some quick testing seems that isn't the case So my new understanding is that the resource limit is a per instance threshold, not a cumulative threshold, and for the use cases I mentioned in the first paragraph (resources as tags) it is sufficient to define a worker resource limit of 1 as long as you always submit the task with resources={resource: 1}. Is that about right? — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#5010 (reply in thread)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACKZTC5CQCRX4G5QAG7QGTTVXV3XANCNFSM47T2JZNA> .

[aa1371]

Yup that was my original workaround. Any thoughts on my statement in the response above?

"
So my new understanding is that the resource limit is a per instance threshold, not a cumulative threshold, and for the use cases I mentioned in the first paragraph (resources as tags) it is sufficient to define a worker resource limit of 1 as long as you always submit the task with resources={resource: 1}. Is that about right?
"

[aa1371]

From my follow-up above:

"
Ah never mind. I see now, it is cumulative, however, I usually start my workers with nthreads=1, so I don't ever have more than 1 task processing on a worker at one time. However, with more threads I see that a limit of 1 prevents multiple tasks from running in separate threads on a worker at once.

So in the general case an infinite limit (or more specifically a limit == nthreads on the worker) would still be desirable, although in my use case it wouldn't be necessary.
"

[aa1371]

So I just got a pleasant surprise. Looks like infinite limit resources already are supported by worker resources.

$ dask-worker <address> --resources="TAG1=inf"

Works as expected since float('inf') is a valid float, and respects all the behavior I'd expect in the worker resources code, e.g. inf - 1 == inf, inf > 0, etc.

cc @jrbourbeau

[aa1371]

$ dask-worker <address> --resources="TAG1=inf"

Works as expected since float('inf') is a valid float, and respects all the behavior I'd expect in the worker resources code, e.g. inf - 1 == inf, inf > 0, etc.