Установка таймаута на выполнение функции в Python

Быстрый ответ

Чтобы установить ограничение по времени на выполнение функции, используйте модуль signal. Важную роль в нём играет функция alarm. Создайте обработчик сигнала, который будет генерировать исключение TimeoutError, когда время истечёт:

import signal

def raise_timeout(signum, frame):
    raise TimeoutError  # Генерируем исключение TimeoutError

signal.signal(signal.SIGALRM, raise_timeout)  # Назначаем обработчик

def my_func():
    # Ваша функция

signal.alarm(5)  # Устанавливаем ограничение по времени в 5 секунд для my_func!
try:
    my_func()
finally:
    signal.alarm(0)  # Отменяем сигнал.

Этот подход подходит для Unix-подобных систем. Если вам нужно обеспечить работу таймаута на различных платформах, рассмотрите модули concurrent.futures или multiprocessing.

Путешествие по миру таймаутов: разные подходы

Если подход с сигналами не подходит, например, если программа работает в среде Windows или использует многопоточность, вам помогут модули multiprocessing и threading.

Многопроцессорность: эффективность без лишних слов

Многопроцессорность оказывается полезной для ресурсоемких задач и позволяет контролировать время их выполнения:

from multiprocessing import Process

def my_func():
    # Здесь будет выполняться ресурсоемкая задача

p = Process(target=my_func)
p.start()
p.join(timeout=5)  # Ожидаем выполнение функции не более 5 секунд, затем происходит прерывание

if p.is_alive():
    p.terminate()  # Если выполнение задачи не закончилось, мы её прерываем

Этот подход подходит для разных ОС.

Пул потоков: эффективное управление

Для операций ввода/вывода или простых задач лучше использовать пул потоков ThreadPool, который умело управляет многозадачностью:

from multiprocessing.pool import ThreadPool

pool = ThreadPool(processes=1)  # Создаём пул из одного потока
async_result = pool.apply_async(my_func)  # Применяем асинхронный вызов функции

try:
    output = async_result.get(timeout=5)  # Ожидаем результата не более 5 секунд, после чего происходит прерывание
except TimeoutError:
    print("Функция не успела выполниться за отведённое время!")

Декораторы: управление временем исполнения

Хотите установить ограничение по времени на выполнение функции с помощью декоратора? Воспользуйтесь следующим кодом:

from functools import wraps
import errno
import os
from threading import Timer

def timeout(seconds=10, error_message=os.strerror(errno.ETIME)):
    def decorator(func):
        @wraps(func)
        def _handle_timeout(*args, **kwargs):
            def _raise_timeout():
                raise TimeoutError
            timer = Timer(seconds, _raise_timeout)
            timer.start()
            try:
                result = func(*args, **kwargs)
            finally:
                timer.cancel()
            return result
        return _handle_timeout
    return decorator

@timeout(5)
def my_func():
    # Ваша функция, время выполнения которой ограничено

Декоратор timeout устанавливает ограничение по времени для каждого метода.

Визуализация

Применение таймаута к вызову функции можно представить следующей аналогией:

Функцию рассматриваем как поезд (🚂), а таймаут является временем отправления (⏰).

| Ситуация              | Визуализация                       |
| --------------------- | ---------------------------------- |
| Функция начала работу вовремя | 🚂 ==🕒==> Выполнение функции закончено в установленное время |
| Превышение времени выполнения | 🚂 ==(⏰!)==> Вызов функции прерван            |

Таймаут (⏰) действует как регулятор, гарантирующий выполнение функции в заданном временном интервале. Если время исполнения функции превышено, вызов функции прерывается.

Работа с таймаутами: специальные случаи и возможные проблемы

При работе с таймаутами имеются нюансы и могут возникнуть ошибки. Не существует универсального решения!

Многопоточность и отзывчивость: ищем баланс

В многопоточных программах управление таймаутами требует особого внимания. Используйте флаги управления и обрабатывайте исключения, чтобы избежать блокировки потоков:

def my_func(stop_event):
    while not stop_event.is_set():
        # Код, управляющийся таймаутом

Совместимость с разными версиями Python: уделяем внимание деталям

Убедитесь, что выбранный подход совместим с разными версиями Python. При работе с ранними версиями языка проявите особую осторожность при использовании модулей multiprocessing или threading.

Действия после истечения таймаута: всегда есть план «Б»

Если основная функция не успела выполниться за заданное время, попробуйте использовать резервный сценарий:

try:
    my_func()
except TimeoutError:
    alternative_func()  # Запускаем резервный сценарий

Неожиданное прерывание: заботимся о завершении

Прерывание потоков или процессов может привести к незавершенным операциям. Убедитесь, что вы правильно обрабатываете оставшиеся ресурсы и данные.

Полезные материалы

1. signal — Установка обработчиков асинхронных событий – Документация Python 3.12.1 – Управление таймаутами с помощью модуля signal по официальной документации Python.
2. threading — Параллельные вычисления на основе потоков — Документация Python 3.12.1 – Использование класса threading.Timer для настройки таймаутов.
3. concurrent.futures — Асинхронный запуск задач — Документация Python 3.12.1 – Применение асинхронных таймаутов с использованием concurrent.futures.
4. Multiprocessing — Параллельные вычисления на основе процессов — Документация Python 3.12.1 – Особенности установки многопроцессорных таймаутов.
5. Использование декоратора таймаута Python для загрузки в S3 – SaltyCrane Blog – Примеры применения декораторов для управления таймаутами.
6. Ускоряем Python-программу с нвое параллелизма – Real Python – Руководство по параллельным вычислениям в Python и управлению таймаутами с помощью concurrent.futures.