如何在 Python 中实现 Thread.ResetAbort 方法

Question

在 Python 中实现 Thread.abort 方法后，我注意到 Thread.reset_abort 方法也可能有用。两者都受到 C# 中的 Thread 对象的启发。虽然创建一个可以自动重新引发的异常还有待解决，但取消计划的异常可能仍然有帮助。

How to automatically re-raise exception after handlers 的问题已经被要求解决问题 X，但是这个问题是关于问题 Y。如果一个线程被安排引发异常，它仍然可以取消它。但是，Python 的 API 中没有 PyThreadState_GetAsyncExc 函数来查看是否设置了异常。

这是目前在其他线程中引发异常的代码：

#! /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)

目标是取消在线程上引发的异常。如果有可能在处理后自动重新引发异常，那么能够取消其自动投掷能力也将有所帮助。目前，这些功能均不可用。

Answer 1

PyThreadState_SetAsyncExc 函数具有特殊功能，可用于取消在其他线程上异步引发异常。您只需将一个特殊值传递给 exc 参数。文档有这样的说法：

如果 exc 为 NULL，则线程的未决异常（如果有）被清除。

虽然这与取消已经捕获和处理后自动引发的异常不同，但它仍然允许“重置中止”已经安排好的可能性。您只需要更改一些代码：

_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)

更改和添加的代码实现了一个Thread.reset_abort 方法，该方法取消了任何计划引发的异常（不仅仅是ThreadAbortException）。如果曾经实现了一个可以自动引发自身的异常类，那么它可能需要自己的reset_abort 方法来防止在调用它之后发生这种行为。虽然没有实现 C# 中提供的确切功能，但该解决方案可能为当前使用 Python 提供的功能提供了最接近的规定。