Hi,
I’m looking to see if I can change the assigned worker resources at runtime. So I can dynamically scale the number of workers that can handle handle certain tasks as workers go on/offline.
My thoughts were to currently do this by using a Plugin, either on the scheduler or workers. There does seem to be a Worker.set_resources
function, though in that context, I don’t have access to the total number of other workers.
Making a SchedulerPlugin seems like a better place to do this. However, it doesn’t seem as though you can change the workers resources safely from here:
# scheduler_setup.py
class MyPlugin(SchedulerPlugin):
def add_worker(self, scheduler=None, worker=None, **kwargs):
state = scheduler.workers[worker]
state.resources["new_resource"] = 1000
print("Total workers:", len(scheduler.workers))
# main.py
def main():
client = Client(asynchronous=False, scheduler_kwargs={"preload": "scheduler_setup"})
sum_ = da.arange(10, chunks=2).sum()
result = client.compute(sum_, resources={'new_resource': 1}).result()
print(result)
client.close()
if __name__ == '__main__':
main()
This method will just hang, as the changes to resources from the scheduler don’t get populated out. Maybe there’s another function for doing this that I’m unaware of, or a way to get scheduler info from the WorkerPlugin instead.
Any thoughts on doing something like this?