Dask: handling unresponsive workers

[Thomas-Z]

I did not really have any success on stackoverflow so I'm trying here :)

When using Dask with SGE or PBS clusters I sometimes have workers becoming unresponsive.
These workers are highlighted in red in the dashboard Info section with their "Last seen" number constantly increasing.

I know this can happen if submitted tasks hold the GIL for too long but that's not the case here. I'm talking about workers for which something went wrong (probably unrelated to dask or the task itself).
They will not come back and are not detected as dead either.

The problem is that tasks submitted on these workers (they become unresponsive after receiving a task, maybe when loading the environment) never end and block everything.

Is there a setting allowing to "timeout" or "invalidate" a worker if it was unresponsive for a given amount of time ?

If not, is it possible and what would be the recommended way to manually do this invalidation and dispatch remaining tasks on other workers ?

Thanks in advance for any help regarding this issue.

[mrocklin]

I wonder if there is interesting information in the logs of these workers. Typically tools like SGE or PBS will store logs somewhere. Those logs might contain valuable clues about what is going wrong.

…

On Tue, Mar 23, 2021 at 4:39 AM Thomas Zilio ***@***.***> wrote: I did not really have any success on stackoverflow <https://stackoverflow.com/questions/66221545/dask-handling-unresponsive-workers> so I'm trying here :) When using Dask with SGE or PBS clusters I sometimes have workers becoming unresponsive. These workers are highlighted in red in the dashboard Info section with their "Last seen" number constantly increasing. I know this can happen if submitted tasks hold the GIL for too long but that's not the case here. I'm talking about workers for which something went wrong (probably unrelated to dask or the task itself). They will not come back and are not detected as dead either. The problem is that tasks submitted on these workers (they become unresponsive after receiving a task, maybe when loading the environment) never end and block everything. Is there a setting allowing to "timeout" or "invalidate" a worker if it was unresponsive for a given amount of time ? If not, is it possible and what would be the recommended way to manually do this invalidation and dispatch remaining tasks on other workers ? Thanks in advance for any help regarding this issue. — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#7451>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACKZTCRE3QTTU6AQ4U4YI3TFBOUTANCNFSM4ZUZXSWQ> .

[Thomas-Z]

That's one of the problems, absolutely nothing in the logs.

My understanding of what's happening is the following (I might be completely wrong, I'm guessing for the most part).

In the following screenshots all submitted tasks are similar only including very small objects.
As you can see, the deserialization part for the second task (and the followings ones) on the same worker is insignificant.

The problem was quite frequent before we made some optimisations to our imports and the dask dashboard would look something like this :

What the french comment is saying is :

Between 60 and 260 seconds for the workers who did respond + all those who did not respond.

The info panel would look like this one (not the same run as previously, we're on a LocalCluster here but it was similar) :

I think the deserialization part is actually the time required to load the environment and import the required classes.
We're working with conda environments installed on network shares and I suspect some kind of saturation effect when xx workers try to load the same environment at the same time.

We managed to reduce this effect by optimizing imports (i.e. we do not import half of the world just in case anymore) but it still occurs from time to time.

From the worker point of view, nothing is wrong. It's loading its environment bit by bit and it will come back in a few days if I'm patient enough. It has no reason to log, crash or do anything other than continue its very (very very) slow loading.

I don't think it has anything to do with dask (I've noticed similar saturation problem in other situations) but I would like to know if there is a way to invalidate/restart/kill these workers.
I think I can detect them using the "Last seen" property but I'm not sure what's the best way to handle them.

Writing these lines I suspect I might be able, once my list of "Last seen way to large" workers done to get the list of tasks assigned to them and manually redispatch them. Is it the right way to do this ?

Would it be possible to have some kind of scheduler setting automatically invalidating a worker if not seen for a specified amount of time ?

Thanks !

[jrbourbeau]

@Thomas-Z there's a distributed.scheduler.worker-ttl configuration value you can set to remove workers from the cluster after they haven't been heard from for a certain period of time. Something like, for example, dask.config.set({"distributed.scheduler.worker-ttl": "120min"})

[Thomas-Z]

And I think that's exactly the answer I was looking for !

Is there any reason other than a job not releasing the GIL that could prevent a worker from sending its heartbeat and would it be safe to set this value as low as 5min or less if we know our jobs profiles ?

Thanks and lot !