Client released keys make running tasks get lost

[danlat]

In my use of dask, where I have some very long running tasks, I often encounter the following situation:

• submit long running tasks (as part of a compute graph or otherwise)
• an error occurs or I change my mind and cancel the future or release my Client
• the Scheduler transitions the tasks from processing->released->forgotten
• any task that was already executing are still running on the worker, but since the scheduler sees that worker as available, the scheduler assigns new work to that worker, even if there are other idle workers
• I do not see the tasks that are actually still executing in the dashboard or workers.html page, they are no where to be seen. Even the Call Stacks from the worker dashboard does not show the old cancelled task that is actually still running
• the cluster appears to have work, and some idle workers, yet appears hung since it is not proceeding

Since I run dask workers with num-threads==1, I can simply restart the workers that are currently processing tasks when the client chooses to cancel or disconnects from the scheduler, but when I tried to use a WorkerPlugin, I did not see any updates when the task transitioned from processing to released. Is this intentional?

[mrocklin]

Dask runs tasks in separate Python threads. Unfortunately, once a task has started the Python language doesn't give us any ability to stop that thread. I'm not sure how much we can do here.

when I tried to use a WorkerPlugin, I did not see any updates when the

task transitioned from processing to released. Is this intentional? Taking a very brief look at the Worker.transition and Worker._notify_transition methods it looks like they alert on every transition. If you think that there is a bug here then the next step would be to create a minimal reproducible example.

…

On Tue, Apr 7, 2020 at 11:08 AM Dan Latuszek ***@***.***> wrote: In my use of dask, where I have some very long running tasks, I often encounter the following situation: - submit long running tasks (as part of a compute graph or otherwise) - an error occurs or I change my mind and cancel the future or release my Client - the Scheduler transitions the tasks from processing->released->forgotten - any task that was already executing are still running on the worker, but since the scheduler sees that worker as available, the scheduler assigns new work to that worker, even if there are other idle workers - I do not see the tasks that are actually still executing in the dashboard or workers.html page, they are no where to be seen. Even the Call Stacks from the worker dashboard does not show the old cancelled task that is actually still running - the cluster appears to have work, and some idle workers, yet appears hung since it is not proceeding Since I run dask workers with num-threads==1, I can simply restart the workers that are currently processing tasks when the client chooses to cancel or disconnects from the scheduler, but when I tried to use a WorkerPlugin, I did not see any updates when the task transitioned from processing to released. Is this intentional? — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#3681>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACKZTBNDNKA4WKBXBYV633RLNTX5ANCNFSM4MDKK3NQ> .

[danlat]

Thanks for the quick reply @mrocklin. I've read your other replies on the topic of stopping python threads and completely understand the technical restrictions. I'm really just interested in figuring out how to identify which worker has unstopped, yet released/forgotten tasks that are still executing so I can restart them. Since, for my use case, I don't typically keep much data around in the cluster, losing a worker that's just busy executing a canceled task is no big deal.

My minimal example to reproduce:

def long_running():
    import time
    time.sleep(600)
f = client.submit(long_running, pure=False)

then either call f.cancel() or restart my ipython kernel, destroying the client. Here are logs from the latter including a quick SchedulerPlugin that prints out the transitions

distributed.scheduler - INFO - Remove client Client-41663e68-7901-11ea-900d-5b4d6c96a0c1                                                               
distributed.scheduler - DEBUG - Client Client-41663e68-7901-11ea-900d-5b4d6c96a0c1 releases keys: ['long_running-0a92dfee-cf4f-47a3-bd45-27e43db63339']
key: long_running-0a92dfee-cf4f-47a3-bd45-27e43db63339                                                                                                 
start: processing                                                                                                                                      
finish: released                                                                                                                                       
args: ()                                                                                                                                               
kwargs: {}                                                                                                                                             

Worker logs nothing and just keeps reporting the state of this key that the scheduler has already forgotten about.

distributed.worker - DEBUG - Execute key: long_running-0a92dfee-cf4f-47a3-bd45-27e43db63339 worker: tcp://127.0.0.1:37318
distributed.worker - DEBUG - Heartbeat: tcp://127.0.0.1:37318
distributed.worker - DEBUG - Heartbeat: tcp://127.0.0.1:37318
distributed.worker - DEBUG - future state: long_running-0a92dfee-cf4f-47a3-bd45-27e43db63339 - RUNNING
distributed.worker - DEBUG - Heartbeat: tcp://127.0.0.1:37318
distributed.worker - DEBUG - Heartbeat: tcp://127.0.0.1:37318
distributed.worker - DEBUG - future state: long_running-0a92dfee-cf4f-47a3-bd45-27e43db63339 - RUNNING
distributed.worker - DEBUG - Heartbeat: tcp://127.0.0.1:37318
distributed.worker - DEBUG - Heartbeat: tcp://127.0.0.1:37318
distributed.worker - DEBUG - future state: long_running-0a92dfee-cf4f-47a3-bd45-27e43db63339 - RUNNING
distributed.worker - DEBUG - Heartbeat: tcp://127.0.0.1:37318
distributed.worker - DEBUG - Heartbeat: tcp://127.0.0.1:37318

I'm going to dig further in to what the Scheduler does in this scenario. By the way, this is with a cluster started via dask-scheduler and dask-worker on the command line.

[mrocklin]

You probably want to look more at the Worker than the Scheduler in this case.

…

On Tue, Apr 7, 2020 at 12:14 PM Dan Latuszek ***@***.***> wrote: Thanks for the quick reply @mrocklin <https://github.com/mrocklin>. I've read your other replies on the topic of stopping python threads and completely understand the technical restrictions. I'm really just interested in figuring out how to identify which worker has unstopped, yet released/forgotten tasks that are still executing so I can restart them. Since, for my use case, I don't typically keep much data around in the cluster, losing a worker that's just busy executing a canceled task is no big deal. My minimal example to reproduce: def long_running(): import time time.sleep(600) f = client.submit(long_running, pure=False) then either call f.cancel() or restart my ipython kernel, destroying the client. Here are logs from the latter including a quick SchedulerPlugin that prints out the transitions distributed.scheduler - INFO - Remove client Client-41663e68-7901-11ea-900d-5b4d6c96a0c1 distributed.scheduler - DEBUG - Client Client-41663e68-7901-11ea-900d-5b4d6c96a0c1 releases keys: ['long_running-0a92dfee-cf4f-47a3-bd45-27e43db63339'] key: long_running-0a92dfee-cf4f-47a3-bd45-27e43db63339 start: processing finish: released args: () kwargs: {} Worker logs nothing and just keeps reporting the state of this key that the scheduler has already forgotten about. distributed.worker - DEBUG - Execute key: long_running-0a92dfee-cf4f-47a3-bd45-27e43db63339 worker: tcp://127.0.0.1:37318 distributed.worker - DEBUG - Heartbeat: tcp://127.0.0.1:37318 distributed.worker - DEBUG - Heartbeat: tcp://127.0.0.1:37318 distributed.worker - DEBUG - future state: long_running-0a92dfee-cf4f-47a3-bd45-27e43db63339 - RUNNING distributed.worker - DEBUG - Heartbeat: tcp://127.0.0.1:37318 distributed.worker - DEBUG - Heartbeat: tcp://127.0.0.1:37318 distributed.worker - DEBUG - future state: long_running-0a92dfee-cf4f-47a3-bd45-27e43db63339 - RUNNING distributed.worker - DEBUG - Heartbeat: tcp://127.0.0.1:37318 distributed.worker - DEBUG - Heartbeat: tcp://127.0.0.1:37318 distributed.worker - DEBUG - future state: long_running-0a92dfee-cf4f-47a3-bd45-27e43db63339 - RUNNING distributed.worker - DEBUG - Heartbeat: tcp://127.0.0.1:37318 distributed.worker - DEBUG - Heartbeat: tcp://127.0.0.1:37318 I'm going to dig further in to what the Scheduler does in this scenario. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#3681 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACKZTACV6JTSOKPI6TRC2DRLN3QLANCNFSM4MDKK3NQ> .

[danlat]

Looks like Worker.release_key does get called, which unfortunately does not call anything on the WorkerPlugins. I'm planning to monkeypatch release_key with my own method with something like this, unless you have a different suggestion:

def _patched_release_key(worker, key, *args, **kwargs):
    if worker.task_state(key) == "executing":
        worker.io_loop.ensure_future(worker.close_gracefully())
    return worker.release_key(key, *args, **kwargs)

class MyPlugin(WorkerPlugin):
    def setup(self, worker):
        worker.release_key = _patched_release_key

[danlat]

This seems to be accomplishing what I want. Despite that, do you have an opinion on whether the released key/task should be visible in the diagnostics dashboard or whether you think the released yet still executing task on a worker should steer work towards other idle workers?

def _new_release_key(worker, key, *args, **kwargs):
    if worker.task_state[key] == "executing":
        print("Worker releasing key that is executing, restarting worker...")
        worker.io_loop.add_callback(worker.close_gracefully)
    return worker.release_key(key, *args, **kwargs)

class MyPlugin(distributed.diagnostics.plugin.WorkerPlugin):
    def setup(self, worker):
        print("Patching release-task handler")
        worker.lifetime_restart = True
        worker.stream_handlers["release-task"] = partial(_new_release_key, worker, report=False)

[mrocklin]

I think that the question to ask here is if we want to add release_key to the WorkerPlugin API

…

On Tue, Apr 7, 2020 at 3:31 PM Dan Latuszek ***@***.***> wrote: This seems to be accomplishing what I want. Despite that, do you have an opinion on whether the released key/task should be visible in the diagnostics dashboard or whether you think the released yet still executing task on a worker should steer work towards other idle workers? def _new_release_key(worker, key, *args, **kwargs): if worker.task_state[key] == "executing": print("Worker releasing key that is executing, restarting worker...") worker.io_loop.add_callback(worker.close_gracefully) return worker.release_key(key, *args, **kwargs) class MyPlugin(distributed.diagnostics.plugin.WorkerPlugin): def setup(self, worker): print("Patching release-task handler") worker.lifetime_restart = True worker.stream_handlers["release-task"] = partial(_new_release_key, worker, report=False) — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#3681 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACKZTH4BJJL62DBIONY32LRLOSSLANCNFSM4MDKK3NQ> .

[danlat]

I decided to implement this as a monkeypatch inside the custom init dask_setup hook when starting dask workers, so that all our clients don't need to upload a WorkerPlugin.

I'm still curious what you think about this situation:

# Start dask cluster with 2 workers
longfut = client.submit(long_running) # lets say it gets sent to worker1
longfut.cancel()
quick1 = client.submit(fast_running) # sent to worker2
quick2 = client.submit(fast_running) # sent to worker1

# now calling quick2.result() will hang until long_running is finished executing, despite worker2 being idle
quick2.result()

[mrocklin]

That sounds suboptimal to me. I suspect that this is because Dask doesn't yet have an expectation of when longfut will finish, but I would need to explore the work stealing code to be sure. If this is something that you're looking to fix I'm all for it.

…

On Wed, Apr 8, 2020 at 7:51 AM Dan Latuszek ***@***.***> wrote: I decided to implement this as a monkeypatch inside the custom init dask_setup hook when starting dask workers, so that all our clients don't need to upload a WorkerPlugin. I'm still curious what you think about this situation: # Start dask cluster with 2 workers longfut = client.submit(long_running) # lets say it gets sent to worker1 longfut.cancel() quick1 = client.submit(fast_running) # sent to worker2 quick2 = client.submit(fast_running) # sent to worker1 # now calling quick2.result() will hang until long_running is finished executing, despite worker2 being idle quick2.result() — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#3681 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACKZTFBEZA33B4EBY7PNODRLSFNVANCNFSM4MDKK3NQ> .

[danlat]

If Dask knew that longfut was going to take over a threshold amount of time compared to the network transfer time to move quick2, would you expect the work stealing to move it? I was suspecting that quick2 wouldn't get stolen because worker1 thought it is in a state to execute quick2, despite longfut still executing.

I'll look through the work stealing code and see if I can come up with a PR. Again, thanks for the quick replies!

[mrocklin]

Oh yeah, that could also be the problem. Good thinking.

…

On Wed, Apr 8, 2020 at 9:30 AM Dan Latuszek ***@***.***> wrote: If Dask knew that longfut was going to take over a threshold amount of time compared to the network transfer time to move quick2, would you expect the work stealing to move it? I was suspecting that quick2 wouldn't get stolen because worker1 thought it is in a state to execute quick2, despite longfut still executing. I'll look through the work stealing code and see if I can come up with a PR. Again, thanks for the quick replies! — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#3681 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACKZTH7IITNNRP3RCKKWXDRLSRDDANCNFSM4MDKK3NQ> .