Добавить тайм-аут к функции Python под Linux/Mac

Python

Когда мы запрашиваем стороннюю библиотеку, используя сеть, такую ​​​​как запросы, мы видим, что у нее есть параметр с именемtimeout, что означает, что расчет начинается при отправке сетевого запроса.Если время ожидания превышено, а возврат не получен, создается исключение времени ожидания. (Конечно, есть особые случаи, когда время ожидания будет недействительным, см.Timeouts and cancellation for humans*Пример автора в этой статье мы не рассматриваем этот частный случай).

Но задумывались ли вы когда-нибудь, как установить тайм-аут для обычных функций? В частности, при выполнении некоторой обработки данных и кода, связанного с ИИ, функция может выполняться в течение длительного времени.Мы хотим понять, что, когда функция выполняется дольше определенного времени, об ошибке будет сообщено автоматически.

Например, есть такой сценарий, я написал функциюcalc_statistic(datas), который вычисляет значение на основе данных, переданных пользователем. Но если данные, переданные пользователем, очень велики, эта функция может работать долгое время. Я хочу, чтобы эта функция работала до 10 секунд. Если запуск не завершен в течение 10 секунд, выдается сообщение об ошибке. Как я должен это делать?

Если на вашем компьютере установлена ​​операционная система Linux или macOS, вы можете использовать сигнал для ее решения.

В статье, опубликованной несколько дней назад на официальном аккаунте, мы представили использование сигнала для приема сигнала прерывания клавиатуры.signal.SIGINT. Сегодня мы будем использоватьsignal.SIGALRM.

Во-первых, давайте посмотрим, как используется этот сигнал:

import time
import signal


def handler(signum, _):
    print('定时到！')
    raise Exception('定时到了！')

def clac_statistic(datas):
    time.sleep(100)
    

signal.signal(signal.SIGALRM, handler)
signal.alarm(5)
clac_statistic('xxx')

Эффект операции показан на следующем рисунке:

связывать первымsignal.SIGALRMсобытие дляhandlerфункцию, затем используйтеsignal.alarm(10)Отправить сигнал с 10-секундной задержкой. Через 10 секунд функцияhandlerработает. В функции возникло исключение, что привело к завершению программы.clac_statisticФункция должна была работать в течение 100 секунд, но остановилась через 10 секунд, тем самым реализуя функцию тайм-аута функции.

Основываясь на вышеизложенных принципах, мы реализуем декоратор, чтобы упростить настройку функции тайм-аута для разных функций:

import time
import signal


class FuncTimeoutException(Exception):
    pass

def handler(signum, _):
    raise FuncTimeoutException('函数定时到了！')

def func_timeout(times=0):
    def decorator(func):
        if not times:
            return func
        def wraps(*args, **kwargs):
            signal.alarm(times)
            result = func(*args, **kwargs)
            signal.alarm(0)  # 函数提前运行完成，取消信号
            return result
        return wraps
    return decorator

signal.signal(signal.SIGALRM, handler)

Мы пытаемся протестировать эту функцию декоратора тайм-аута. Первая тестовая функция времени работы меньше таймаута, программа работает без проблем:

Проверим случай, когда время работы функции превышает время ожидания:

нормальный бросокFuncTimeoutExceptionаномальный.

Тогда в реальном использовании мы можем использоватьtry...except FuncTimeoutExceptionПерехватите это исключение и реализуйте собственный поток обработки, например:

try:
    clac_statistic(100)
except FuncTimeException:
    print('该函数运行超时，运行自定义的处理流程')

Конечно, если вы хотите напрямую пропустить это исключение, нет проблем:

import contextlib:
with contextlib.supress(FuncTimeException):
    clac_statistic(100)

Категории