Luigi: Support for resources for limiting parallelism on a per worker basis

[graingert]

It would be great to label tasks with "uses_lots_of_ram" or "uses_analytics_mysql_db" and then apply parallelism limits to those labels, so say only one RAM hungry task runs at once, or only 2 tasks that use the analytics db

[Tarrasch]

There is global resources, would you also like per-worker or per-host resource limitations?

[graingert]

@Tarrasch I'm looking for per-host. I didn't know about gloabl resources. Do you have a docs link?

[Tarrasch]

https://github.com/spotify/luigi/blob/0a3797911425f6f16812ac441f14c4e5e0a2e4c8/doc/configuration.rst#resources and https://github.com/spotify/luigi/blob/ddbf24d2e5d1e66989a2d2739ccd19bbedd01d1e/luigi/task.py#L122-L124 are some good pointers.

[davemt]

We were looking for something like this recently.

One thought I had is something like the following, which is more of a hack really (also not tested):

class MyTask(Task):
    resources = {'big_ram': 1}

    # likely put in a base class:
    def process_resources(self):
        return dict(self.resources, **{
            '{0}-{1}'.format(k, socket.gethostname()): v
            for k, v in self.resources.iteritems()
        })

Then on scheduler you'd set limits per host

[resources]
big_ram = 20
big_ram-some.hostname = 5
big_ram-another.hostname = 10

The alternative that would require an actual implementation could be sections like [resources_some.hostname] or [host_resources] (for global per-host limits) in scheduler configs, and then scheduler would keep track of resources used per host.

[erikbern]

Doesn't this work as it is?

[davemt]

Yeah, I was offering an approach to be used in place of implementing a new feature for the issue (again haven't actually tested, but I don't see why it wouldn't work)

[erikbern]

i'm closing this for now. seems like it's solvable using a workaround, and it's probably not too hard to implement a cleaner solution if someone wants to. but until then the workaround is fine

[nilgoyette]

"seems like it's solvable using a workaround"
Do you know there's a workaround and how to do it or do you think there's probably one? Because if you know how to do per-worker ressource management, I would be really interested!

[yaakov2]

@davemt -- Can you please confirm that I understand your workaround correctly?

Your code seems to rely on the fact that the same task may report different resource requirements for each worker. In your example the "same task" MyTask() requires {big_ram-localhost:1} when running on localhost but {big_ram-bigserver:1} when run on bigserver. If your central scheduler specifies a low resource limit for big_ram-localhost and a high resource limit for big_ram-bigserver, then it will allow more of these tasks to be executed on bigserver than on localhost.

In the documentation, I did not find the fact that the central scheduler will properly honor different requirements for the same task as they are reported by the different workers.

Can you confirm that this is how per-worker resource management works. If yes -- can this be clarified in the documentation?