Как реализовать метод Thread.ResetAbort в Python - PullRequest
Новая
       42

Как реализовать метод Thread.ResetAbort в Python

0 голосов
/ 14 мая 2019

После реализации метода Thread.abort в Python до меня дошло, что метод Thread.reset_abort также может быть полезен.Оба вдохновлены Thread объектами, найденными в C #.Хотя создание исключения, которое может автоматически повторно вызывать себя, еще не решено, отмена запланированного исключения все еще может быть полезна.

Вопрос Как автоматически повторно вызвать исключение после того, как обработчики уже было предложено решить проблему X, но этот вопрос касается проблемы Y. Если поток запланирован для создания исключения, все еще может быть возможность отменить его.Однако в API Python нет функции 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 a 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)

цель - отменить исключение из потока.Если когда-либо станет возможным получить исключение для автоматического повторного повышения самого себя после того, как оно будет обработано, то также будет полезна возможность отмены его способности автоматического броска.Сейчас ни одна из этих функций не доступна.

1 Ответ

0 голосов
/ 14 мая 2019

Функция 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 a 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.

...