How to implement Thread.ResetAbort method in Python

86 Views Asked by At

After implementing a Thread.abort method in Python, it has come to my attention that a Thread.reset_abort method might also be useful. Both are inspired by Thread objects found in C#. While creating an exception that can automatically re-raise itself has yet to be solved, cancelling a scheduled exception may still be helpful.

The question How to automatically re-raise exception after handlers has already been asked to solve problem X, but this question is about problem Y. If a thread is scheduled to raise an exception, it may still be possible to cancel it. However, there is no PyThreadState_GetAsyncExc function in Python's API to see if an exception has been set.

Here is the code so far for raising exceptions in other threads:

#! /usr/bin/env python3
import _thread
import ctypes as _ctypes
import threading as _threading

_PyThreadState_SetAsyncExc = _ctypes.pythonapi.PyThreadState_SetAsyncExc
# noinspection SpellCheckingInspection
_PyThreadState_SetAsyncExc.argtypes = _ctypes.c_ulong, _ctypes.py_object
_PyThreadState_SetAsyncExc.restype = _ctypes.c_int

# noinspection PyUnreachableCode
if __debug__:
    # noinspection PyShadowingBuiltins
    def _set_async_exc(id, exc):
        if not isinstance(id, int):
            raise TypeError(f'{id!r} not an int instance')
        if not isinstance(exc, type):
            raise TypeError(f'{exc!r} not a type instance')
        if not issubclass(exc, BaseException):
            raise SystemError(f'{exc!r} not a BaseException subclass')
        return _PyThreadState_SetAsyncExc(id, exc)
else:
    _set_async_exc = _PyThreadState_SetAsyncExc


# noinspection PyShadowingBuiltins
def set_async_exc(id, exc, *args):
    if args:
        class StateInfo(exc):
            def __init__(self):
                super().__init__(*args)

        return _set_async_exc(id, StateInfo)
    return _set_async_exc(id, exc)


def interrupt(ident=None):
    if ident is None:
        _thread.interrupt_main()
    else:
        set_async_exc(ident, KeyboardInterrupt)


# noinspection PyShadowingBuiltins
def exit(ident=None):
    if ident is None:
        _thread.exit()
    else:
        set_async_exc(ident, SystemExit)


class ThreadAbortException(SystemExit):
    pass


class Thread(_threading.Thread):
    def set_async_exc(self, exc, *args):
        return set_async_exc(self.ident, exc, *args)

    def interrupt(self):
        self.set_async_exc(KeyboardInterrupt)

    def exit(self):
        self.set_async_exc(SystemExit)

    def abort(self, *args):
        self.set_async_exc(ThreadAbortException, *args)

The goal is to cancel an exception from being raised on a thread. If it ever becomes possible to get an exception to automatically re-raise itself after it has been handled, being able to cancel its auto-throw ability would also be helpful. Right now, neither of these features are available.

1

There are 1 best solutions below

0
Noctis Skytower On

The PyThreadState_SetAsyncExc function has a special capability that can be utilized to cancel raising exceptions asynchronously on other threads. You just need to pass a special value to the exc argument. The documentation has this to say:

If exc is NULL, the pending exception (if any) for the thread is cleared.

Though this is not the same as cancelling an exception from automatically raising itself after it has already been caught and handled, it still allows the possibility of "resetting an abort" that has been scheduled already. You just need to change some code:

_NULL = _ctypes.py_object()


def _set_async_exc(id, exc):
    if not isinstance(id, int):
        raise TypeError(f'{id!r} not an int instance')
    if exc is not _NULL:
        if not isinstance(exc, type):
            raise TypeError(f'{exc!r} not a type instance')
        if not issubclass(exc, BaseException):
            raise SystemError(f'{exc!r} not a BaseException subclass')
    return _PyThreadState_SetAsyncExc(id, exc)


class Thread(_threading.Thread):
    def reset_abort(self):
        self.set_async_exc(_NULL)

The changes and added code implements a Thread.reset_abort method that cancels any exception that has been scheduled to be raised (not just ThreadAbortException). If an exception class is ever implemented that can automatically raise itself, it will probably require its own reset_abort method that prevents such behavior after it has been called. While not implementing the exact functionality provided in C#, this solution probably provides the closest provision for what is currently possible using Python.