함수를 어떻게 호출하거나 5 초 이상 걸리면 스크립트가 취소하도록 랩핑하는 방법은 무엇입니까?

나는 이 질문 / 문제를 해결 하는 요점 을 데코레이터와threading.Timer . 여기에 고장이 있습니다.

호환성을위한 가져 오기 및 설정

Python 2 및 3으로 테스트되었으며 Unix / Linux 및 Windows에서도 작동합니다.

먼저 수입품. 다음은 Python 버전에 관계없이 코드 일관성을 유지하려는 시도입니다.

from __future__ import print_function
import sys
import threading
from time import sleep
try:
    import thread
except ImportError:
    import _thread as thread

버전 독립적 코드 사용 :

try:
    range, _print = xrange, print
    def print(*args, **kwargs):
        flush = kwargs.pop('flush', False)
        _print(*args, **kwargs)
        if flush:
            kwargs.get('file', sys.stdout).flush()
except NameError:
    pass

이제 표준 라이브러리에서 기능을 가져 왔습니다.

exit_after 데코레이터

다음으로 main()자식 스레드에서 종료하는 함수가 필요합니다 .

def quit_function(fn_name):
    # print to stderr, unbuffered in Python 2.
    print('{0} took too long'.format(fn_name), file=sys.stderr)
    sys.stderr.flush() # Python 3 stderr is likely buffered.
    thread.interrupt_main() # raises KeyboardInterrupt

그리고 여기 데코레이터 자체가 있습니다.

def exit_after(s):
    '''
    use as decorator to exit process if
    function takes longer than s seconds
    '''
    def outer(fn):
        def inner(*args, **kwargs):
            timer = threading.Timer(s, quit_function, args=[fn.__name__])
            timer.start()
            try:
                result = fn(*args, **kwargs)
            finally:
                timer.cancel()
            return result
        return inner
    return outer

용법

다음은 5 초 후에 종료에 대한 귀하의 질문에 직접 답변하는 사용법입니다!

@exit_after(5)
def countdown(n):
    print('countdown started', flush=True)
    for i in range(n, -1, -1):
        print(i, end=', ', flush=True)
        sleep(1)
    print('countdown finished')

데모:

>>> countdown(3)
countdown started
3, 2, 1, 0, countdown finished
>>> countdown(10)
countdown started
10, 9, 8, 7, 6, countdown took too long
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 11, in inner
  File "<stdin>", line 6, in countdown
KeyboardInterrupt

두 번째 함수 호출이 완료되지 않고 대신 프로세스가 역 추적로 종료됩니다!

KeyboardInterrupt 잠자는 실을 항상 멈추지는 않습니다

Windows의 Python 2에서 키보드 인터럽트로 인해 절전 모드가 항상 중단되는 것은 아닙니다.

@exit_after(1)
def sleep10():
    sleep(10)
    print('slept 10 seconds')

>>> sleep10()
sleep10 took too long         # Note that it hangs here about 9 more seconds
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 11, in inner
  File "<stdin>", line 3, in sleep10
KeyboardInterrupt

명시 적으로 확인하지 않는 한 확장에서 실행되는 코드를 중단하지 않을 수도 PyErr_CheckSignals()있습니다. Cython, Python 및 KeyboardInterrupt를 참조하십시오.

어쨌든 스레드가 1 초 이상 잠자 지 않도록해야합니다. 프로세서 시간이 짧습니다.

함수를 어떻게 호출합니까? 또는 5 초 이상 걸리면 스크립트가 취소하고 다른 작업을 수행하기 위해 무엇을 랩핑 합니까?

그것을 잡아서 다른 것을하기 위해 KeyboardInterrupt를 잡을 수 있습니다.

>>> try:
...     countdown(10)
... except KeyboardInterrupt:
...     print('do something else')
...
countdown started
10, 9, 8, 7, 6, countdown took too long
do something else

답변

def timeout(func, args=(), kwargs={}, timeout_duration=1, default=None):
    import signal

    class TimeoutError(Exception):
        pass

    def handler(signum, frame):
        raise TimeoutError()

    # set the timeout handler
    signal.signal(signal.SIGALRM, handler)
    signal.alarm(timeout_duration)
    try:
        result = func(*args, **kwargs)
    except TimeoutError as exc:
        result = default
    finally:
        signal.alarm(0)

    return result

import multiprocessing.pool
import functools

def timeout(max_timeout):
    """Timeout decorator, parameter in seconds."""
    def timeout_decorator(item):
        """Wrap the original function."""
        @functools.wraps(item)
        def func_wrapper(*args, **kwargs):
            """Closure for function."""
            pool = multiprocessing.pool.ThreadPool(processes=1)
            async_result = pool.apply_async(item, args, kwargs)
            # raises a TimeoutError if execution exceeds max_timeout
            return async_result.get(max_timeout)
        return func_wrapper
    return timeout_decorator

@timeout(5.0)  # if execution takes longer than 5 seconds, raise a TimeoutError
def test_base_regression(self):
    ...

from concurrent.futures import ProcessPoolExecutor

# Warning: this does not terminate function if timeout
def timeout_five(fnc, *args, **kwargs):
    with ProcessPoolExecutor() as p:
        f = p.submit(fnc, *args, **kwargs)
        return f.result(timeout=5)