How to limit execution time of a function call? [duplicate]

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

Улучшением ответа @rik.the.vik было бы использование with оператора для предоставления функции timeout некоторого синтаксического сахара:

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!")

Я не уверен, насколько кроссплатформенным это может быть, но использование signals и alarm может быть хорошим способом взглянуть на это. Немного поработав, вы могли бы сделать это полностью универсальным и пригодным для использования в любой ситуации.

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!"

Вот способ ограничить время выполнения функции в 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

Я предпочитаю подход context manager, потому что он позволяет выполнять несколько операторов python в 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 вызова.