Allow workers to cancel running task?

Currently, if a task is already running on a thread of a worker, it will always run to completion/exception even if the work is no longer required (e.g., the future was explicitly cancelled). We usually say "you cannot interrupt a python thread", so Dask has no option here. However, you can set the thread state to exception using ctypes.pythonapi.PyThreadState_SetAsyncExc. Yes, this is a C-api call, but it is public and stable. Is this worth pursuing? It would allow not only early cancellation of unneeded work, but "interrupts" and timeouts from the worker itself, perhaps even better debugging if it can break deadlocks with a useful traceback.

(Note that the setting of an exception only has effect at the next python bytecode instruction, so some long-running C-level function or even wait(big) will not raise immediately)

If there is a safe and effective way to cancel a thread then I would be thrilled to see it added to Dask

…

On Mon, Apr 12, 2021 at 10:47 AM Martin Durant ***@***.***> wrote: Currently, if a task is already running on a thread of a worker, it will always run to completion/exception even if the work is no longer required (e.g., the future was explicitly cancelled). We usually say "you cannot interrupt a python thread", so Dask has no option here. However, you can <https://blog.finxter.com/how-to-kill-a-thread-in-python/#Method_4_Using_ctypes_To_Raise_Exceptions_In_A_Thread> set the thread state to exception using ctypes.pythonapi.PyThreadState_SetAsyncExc. Yes, this is a C-api call, but it is public and stable. Is this worth pursuing? It would allow not only early cancellation of unneeded work, but "interrupts" and timeouts from the worker itself, perhaps even better debugging if it can break deadlocks with a useful traceback. (Note that the setting of an exception only has effect at the next python bytecode instruction, so some long-running C-level function or even wait(big) will not raise immediately) — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#4694>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACKZTB7NVEHGBXUKJRXRRDTIMIYTANCNFSM42ZRYKOQ> .

I'll make a PR with a version of the code linked above, so we can discuss it more there.

Yeah this would be nice to be able to do using a stable, public API

cc @pitrou who might be able to tell us why this would break things

…

On Mon, Apr 12, 2021 at 10:56 AM James Bourbeau ***@***.***> wrote: Yeah this would be nice to be able to do using a stable, public API — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#4694 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACKZTH4BCGUC6WNZYR54BLTIMJ3HANCNFSM42ZRYKOQ> .

I have a few concerns

If the thread spawns subthreads, what happens? Are they closed automagically as well?
What if the compute thread is supposed to close external resources, e.g. a connection, does it have time to do so, e.g. will the __exit__ of a ctx manager be executed? IIUC yes but we should confirm

If the thread spawns subthreads, what happens?

Whatever would happen on a normal exception, I think. So the working thread, if it was waiting on some other thread, would raise the exception normally. The main thread would not know that there was another thread around, so work may still happen there, unless the code in the task was well-written to pass an "I have reached an exception, please end" message to its child thread.

if the compute thread is supposed to close external resources, e.g. a connection, does it have time to do so

Yes, it should be the equivalent of meeting the exception at the current location in the code, so contexts would be respected

If that all works this way, that sounds great

Well, the problem is that the exception may interrupt your code at any point. For example if you have:

try:
    # some computation ...
finally:
    # on exit, finalize some IO etc.

then the exception could be raised in the middle of the finally block and some cleanup steps would be missed. It could even be raised inside a __exit__ for a context manager.

For a library like distributed that does a lot of cleanup in destructors/finalizers/etc., it sounds like a bad idea.

That kind of cleanup doesn't happen within a task execution, though, at least not by Dask - these threads only run the functions supplied by the user.

You are right, in general, that code associated with leaving a task would not run if the "kill" happened after areas where an exception might have been anticipated. There's probably no easy way to tell if this is the case for arbitrary user code.

You could do this the other way: expose an API for user code to know whether the current task should be abandoned, and let the user code call it as desired. That allows for safe cancellation. And/or expose something like a allow_async_cancellation parameter or function in a task.

Oh, and is it inside a free-running thread or a ThreadPoolExecutor thread?

You could do this the other way: expose an API for user code to know whether the current task should be abandoned, and let the user code call it as desired.

I think it's pretty unlikely we would get any buy-in on this; in fact, you can already do this by introspecting the worker from get_worker().

is it inside a free-running thread or a ThreadPoolExecutor thread?

The latter, as subclassed in distributed. The exception would be seceded threads (which are still launched the same way). If I remember, an actor thread runs on a executor with exactly one thread.

If your async exception is raised inside the ThreadPoolExecutor bookkeeping code, you might get a deadlock or other issues.

Does any of ThreadPoolExecutor's own code run within the threads it manages?

Well, yes, so that threads are reused. Spawning a new thread for each task would be wasteful.
https://github.com/python/cpython/blob/master/Lib/concurrent/futures/thread.py#L74-L101

Oh, right, I see it.
We can guard this by using our own exception class, right?

I'm not sure that would change anything. You could still interrupt the worker's task handling loop at any point.

I mean like this
(distributed/_concurrent_futures_thread.py)

def _worker(executor_reference, work_queue):
    try:
        while True:
            try:
                work_item = work_queue.get(block=True)
                if work_item is not None:
                    work_item.run()
                    # Delete references to object. See issue16284
                    del work_item
                    continue
                executor = executor_reference()
                # Exit if:
                #   - The interpreter is shutting down OR
                #   - The executor that owns the worker has been collected OR
                #   - The executor that owns the worker has been shutdown.
                if _shutdown or executor is None or executor._shutdown:
                    # Notice other workers
                    work_queue.put(None)
                    return
                del executor
            except WorkerThreadInterrupt:
                # thread interrupt outside of a task
                pass
    except BaseException:
        _base.LOGGER.critical("Exception in worker", exc_info=True)

Ah! Well, I suppose that could work...

Navigation Menu

Repository navigation

Allow workers to cancel running task? #4694

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Allow workers to cancel running task? #4694

Description

Activity

mrocklin commented on Apr 12, 2021

martindurant commented on Apr 12, 2021

jrbourbeau commented on Apr 12, 2021

mrocklin commented on Apr 12, 2021

fjetter commented on Apr 12, 2021

martindurant commented on Apr 12, 2021

fjetter commented on Apr 12, 2021

pitrou commented on Apr 12, 2021

martindurant commented on Apr 12, 2021

pitrou commented on Apr 12, 2021

pitrou commented on Apr 12, 2021

martindurant commented on Apr 12, 2021

pitrou commented on Apr 12, 2021

martindurant commented on Apr 12, 2021

pitrou commented on Apr 12, 2021

martindurant commented on Apr 12, 2021

pitrou commented on Apr 12, 2021

martindurant commented on Apr 12, 2021

pitrou commented on Apr 12, 2021

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions