Python에서 함수 호출의 실행 시간을 제한하는 방법

Python에서 함수 호출의 실행 시간을 제한하는 방법

---

내 코드에 소켓 관련 함수 호출이 있습니다. 해당 함수는 다른 모듈에서 가져 왔으므로 제어 할 수 없습니다. 문제는 가끔 몇 시간 동안 차단된다는 것입니다. 이는 완전히 수용 할 수 없습니다. 내 코드에서 함수 실행 시간을 어떻게 제한 할 수 있습니까? 솔루션이 다른 스레드를 사용해야한다고 생각합니다.

---

크로스 플랫폼이 얼마나 될지 잘 모르겠지만 신호와 알람을 사용하는 것이 이것을 보는 좋은 방법 일 수 있습니다. 약간의 작업을 통해이를 완전히 일반화하고 어떤 상황에서도 사용할 수 있습니다.

http://docs.python.org/library/signal.html

따라서 코드는 다음과 같이 보일 것입니다.

import signal

def signal_handler(signum, frame):
    raise Exception("Timed out!")

signal.signal(signal.SIGALRM, signal_handler)
signal.alarm(10)   # Ten seconds
try:
    long_function_call()
except Exception, msg:
    print "Timed out!"

---

@ rik.the.vik의 대답에 대한 개선은 with문 을 사용하여 시간 제한 함수에 구문 설탕을 제공하는 것입니다.

import signal
from contextlib import contextmanager

class TimeoutException(Exception): pass

@contextmanager
def time_limit(seconds):
    def signal_handler(signum, frame):
        raise TimeoutException("Timed out!")
    signal.signal(signal.SIGALRM, signal_handler)
    signal.alarm(seconds)
    try:
        yield
    finally:
        signal.alarm(0)


try:
    with time_limit(10):
        long_function_call()
except TimeoutException as e:
    print("Timed out!")

---

다음은 함수의 실행 시간을 제한하는 Linux / OSX 방법입니다. 이것은 스레드를 사용하고 싶지 않고 프로그램이 함수가 종료되거나 시간 제한이 만료 될 때까지 기다리기를 원하는 경우입니다.

from multiprocessing import Process
from time import sleep

def f(time):
    sleep(time)


def run_with_limited_time(func, args, kwargs, time):
    """Runs a function with time limit

    :param func: The function to run
    :param args: The functions args, given as tuple
    :param kwargs: The functions keywords, given as dict
    :param time: The time limit in seconds
    :return: True if the function ended successfully. False if it was terminated.
    """
    p = Process(target=func, args=args, kwargs=kwargs)
    p.start()
    p.join(time)
    if p.is_alive():
        p.terminate()
        return False

    return True


if __name__ == '__main__':
    print run_with_limited_time(f, (1.5, ), {}, 2.5) # True
    print run_with_limited_time(f, (3.5, ), {}, 2.5) # False

---

문 내에서 여러 파이썬 문을 실행할 수 있기 때문에 컨텍스트 관리자 접근 방식을 선호합니다 with time_limit. Windows 시스템에는.이 없기 때문에 SIGALARM더 이식 가능하고 아마도 더 간단한 방법은Timer

from contextlib import contextmanager
import threading
import _thread

class TimeoutException(Exception):
    def __init__(self, msg=''):
        self.msg = msg

@contextmanager
def time_limit(seconds, msg=''):
    timer = threading.Timer(seconds, lambda: _thread.interrupt_main())
    timer.start()
    try:
        yield
    except KeyboardInterrupt:
        raise TimeoutException("Timed out for operation {}".format(msg))
    finally:
        # if the action ends in specified time, timer is canceled
        timer.cancel()

import time
# ends after 5 seconds
with time_limit(5, 'sleep'):
    for i in range(10):
        time.sleep(1)

# this will actually end after 10 seconds
with time_limit(5, 'sleep'):
    time.sleep(10)

여기서 핵심 기술 _thread.interrupt_main은 타이머 스레드에서 주 스레드를 인터럽트하는 데 사용하는 것입니다 . 한 가지주의 할 점은 주 스레드가 항상 빠르게 KeyboardInterrupt제기 된 스레드에 응답하지 않는다는 것 Timer입니다. 예를 들어, time.sleep()시스템 함수를 호출하여 a KeyboardInterrupt가 sleep호출 후에 처리됩니다 .

---

신호 처리기 내에서이 작업을 수행하는 것은 위험합니다. 예외가 발생했을 때 예외 처리기 내부에있을 수 있으며 문제가 발생한 상태로 남겨 둘 수 있습니다. 예를 들면

def function_with_enforced_timeout():
  f = open_temporary_file()
  try:
   ...
  finally:
   here()
   unlink(f.filename)

여기 ()에서 예외가 발생하면 임시 파일이 삭제되지 않습니다.

여기서 해결책은 코드가 예외 처리 코드 (except 또는 finally 블록) 내에 있지 않을 때까지 비동기 예외를 연기하는 것입니다. 그러나 Python은 그렇게하지 않습니다.

이것은 네이티브 코드를 실행하는 동안 어떤 것도 중단하지 않습니다. 함수가 반환 될 때만 중단되므로이 특정 경우에는 도움이되지 않을 수 있습니다. (SIGALRM 자체는 차단중인 호출을 중단 할 수 있지만 소켓 코드는 일반적으로 EINTR 후에 단순히 재 시도합니다.)

스레드로이 작업을 수행하는 것이 신호보다 이식성이 더 높기 때문에 더 나은 생각입니다. 작업자 스레드를 시작하고 완료 될 때까지 차단하므로 일반적인 동시성 문제는 없습니다. 불행히도 Python의 다른 스레드에 비동기 적으로 예외를 전달할 수있는 방법은 없습니다 (다른 스레드 API가이를 수행 할 수 있음). 또한 예외 처리기 중에 예외를 보내는 것과 동일한 문제가 발생하며 동일한 수정이 필요합니다.

---

스레드를 사용할 필요가 없습니다. 다른 프로세스를 사용하여 차단 작업을 수행 할 수 있습니다 (예 : 하위 프로세스 모듈 사용). 프로그램의 서로 다른 부분간에 데이터 구조를 공유하고 싶다면 Twisted 가이를 제어 할 수있는 훌륭한 라이브러리이며, 차단에 관심이 있고이 문제가 많이 발생할 것으로 예상되는 경우 권장합니다. Twisted의 나쁜 소식은 블로킹을 피하기 위해 코드를 다시 작성해야하고 공정한 학습 곡선이 있다는 것입니다.

You can use threads to avoid blocking, but I'd regard this as a last resort, since it exposes you to a whole world of pain. Read a good book on concurrency before even thinking about using threads in production, e.g. Jean Bacon's "Concurrent Systems". I work with a bunch of people who do really cool high performance stuff with threads, and we don't introduce threads into projects unless we really need them.

---

The only "safe" way to do this, in any language, is to use a secondary process to do that timeout-thing, otherwise you need to build your code in such a way that it will time out safely by itself, for instance by checking the time elapsed in a loop or similar. If changing the method isn't an option, a thread will not suffice.

Why? Because you're risking leaving things in a bad state when you do. If the thread is simply killed mid-method, locks being held, etc. will just be held, and cannot be released.

So look at the process way, do not look at the thread way.

---

Here's a timeout function I think I found via google and it works for me.

From: http://code.activestate.com/recipes/473878/

def timeout(func, args=(), kwargs={}, timeout_duration=1, default=None):
    '''This function will spwan a thread and run the given function using the args, kwargs and 
    return the given default value if the timeout_duration is exceeded 
    ''' 
    import threading
    class InterruptableThread(threading.Thread):
        def __init__(self):
            threading.Thread.__init__(self)
            self.result = default
        def run(self):
            try:
                self.result = func(*args, **kwargs)
            except:
                self.result = default
    it = InterruptableThread()
    it.start()
    it.join(timeout_duration)
    if it.isAlive():
        return it.result
    else:
        return it.result   

---

I would usually prefer using a contextmanager as suggested by @josh-lee

But in case someone is interested in having this implemented as a decorator, here's an alternative.

Here's how it would look like:

import time
from timeout import timeout

class Test(object):
    @timeout(2)
    def test_a(self, foo, bar):
        print foo
        time.sleep(1)
        print bar
        return 'A Done'

    @timeout(2)
    def test_b(self, foo, bar):
        print foo
        time.sleep(3)
        print bar
        return 'B Done'

t = Test()
print t.test_a('python', 'rocks')
print t.test_b('timing', 'out')

And this is the timeout.py module:

import threading

class TimeoutError(Exception):
    pass

class InterruptableThread(threading.Thread):
    def __init__(self, func, *args, **kwargs):
        threading.Thread.__init__(self)
        self._func = func
        self._args = args
        self._kwargs = kwargs
        self._result = None

    def run(self):
        self._result = self._func(*self._args, **self._kwargs)

    @property
    def result(self):
        return self._result


class timeout(object):
    def __init__(self, sec):
        self._sec = sec

    def __call__(self, f):
        def wrapped_f(*args, **kwargs):
            it = InterruptableThread(f, *args, **kwargs)
            it.start()
            it.join(self._sec)
            if not it.is_alive():
                return it.result
            raise TimeoutError('execution expired')
        return wrapped_f

The output:

python
rocks
A Done
timing
Traceback (most recent call last):
  ...
timeout.TimeoutError: execution expired
out

Notice that even if the TimeoutError is thrown, the decorated method will continue to run in a different thread. If you would also want this thread to be "stopped" see: Is there any way to kill a Thread in Python?

참고URL : https://stackoverflow.com/questions/366682/how-to-limit-execution-time-of-a-function-call-in-python