Как ограничить время выполнения функции python

Асинхронный менеджер timeout() модуля asyncio в Python

Асинхронный диспетчер контекста для ограничения времени ожидания

Синтаксис:

import asyncio # Новое в Python 3.11. async with asyncio.timeout(delay): await long_running_task() 

Параметры:

• delay — число секунд ожидания. Может быть None или float / int .

Возвращаемое значение:

• Асинхронный менеджер контекста asyncio.Timeout .

Описание:

Функция timeout() модуля asyncio представляет собой асинхронный диспетчер контекста, который можно использовать для ограничения времени, затрачиваемого на ожидание чего-либо.

Аргумент delay может быть None или числом секунд ожидания ( float или int ). Если задержка равна None, ограничение по времени применяться не будет; это может быть полезно, если задержка неизвестна при создании менеджера контекста.

Новое в Python 3.11.

Также обратите внимание на функцию asyncio.timeout_at(when) (новая в Python 3.11), которая создает асинхронный контекстный менеджер аналогично asyncio.timeout() , за исключением того, что аргумент when — это абсолютное время ожидания, или None .

В любом случае диспетчер контекста может быть перепланирован после создания с помощью метода Timeout.reschedule() .

async def main(): async with asyncio.timeout(10): await long_running_task() 

Если для выполнения сопрограммы long_running_task() требуется более 10 секунд, то диспетчер контекста отменит текущую задачу и обработает полученную ошибку asyncio.CancelledError , преобразуя ее в ошибку asyncio.TimeoutError , которую можно перехватить и обработать.

Примечание. Контекстный менеджер asyncio.timeout() преобразует asyncio.CancelledError в asyncio.TimeoutError , что означает, что asyncio.TimeoutError может быть перехвачен только за пределами контекстного менеджера.

Пример отлова asyncio.TimeoutError :

async def main(): try: async with asyncio.timeout(10): await long_running_task() except TimeoutError: print("Время длительной операции истекло.") print("print будет выполняться независимо.") 

Менеджер контекста, созданный asyncio.timeout() , может быть перенесен на другой крайний срок и проверен.

Методы асинхронного менеджера контекста asyncio.Timeout .

Timeout.when() :

Метод Timeout.when() возвращает текущий крайний срок или None , если текущий крайний срок не установлен.

Крайний срок — это число float , соответствующее времени, возвращаемому функцией loop.time() .

Timeout.reschedule(when) :

Метод Timeout.reschedule() изменяет время срабатывания тайм-аута.

• Если when имеет значение None , то любой текущий крайний срок будет удален, и диспетчер контекста будет ждать неопределенное время.
• Если это значение float , то оно устанавливается как новый крайний срок.

Если значение when уже прошло, то тайм-аут сработает на следующей итерации цикла событий.

Timeout.expired() :

Метод Timeout.expired() возвращает True , если контекстный менеджер превысил свой крайний срок (т.е. время ожидания закончилось/истекло).

Общий пример использования asyncio.timeout() .

async def main(): try: # не знаем тайм-аута при запуске, поэтому передаем `None`. async with asyncio.timeout(None) as cm: # известен тайм-аут, поэтому переназначаем его. new_deadline = asyncio.get_running_loop().time() + 10 cm.reschedule(new_deadline) await long_running_task() except TimeoutError: pass if cm.expired: print("Похоже, вовремя не закончили.") 

Менеджеры контекста тайм-аута могут быть безопасно вложены друг в друга.

• КРАТКИЙ ОБЗОР МАТЕРИАЛА.
• Сопрограммы и механизмы их запуска модулем asyncio
• Что такое аwaitable объект модуля asyncio
• Функция run() модуля asyncio
• Менеджер контекста Runner() модуля asyncio
• Функция create_task() модуля asyncio
• Группы задач TaskGroup() модуля asyncio
• Класс Task() модуля asyncio
• Функция sleep() модуля asyncio
• Функция gather() модуля asyncio
• Функция shield() модуля asyncio
• Асинхронный менеджер timeout() модуля asyncio
• Асинхронный менеджер timeout_at() модуля asyncio
• Функция wait_for() модуля asyncio
• Функция as_completed() модуля asyncio
• Функция wait() модуля asyncio
• Функция to_thread() модуля asyncio
• Функция run_coroutine_threadsafe() модуля asyncio
• Функции current_task() и all_tasks() модуля asyncio
• Использование очереди asyncio.Queue
• Примитивы синхронизации задач в asyncio
• Запуск внешних программ из кода asyncio
• Работа с сетевыми соединениями модуля asyncio
• Объект Future модуля asyncio Python и связанные функции
• Создание и получение текущего цикла событий, модуль asyncio
• Создание, запуск и остановка цикла событий модуля asyncio
• Планирование обратных вызовов из цикла событий asyncio
• Создание Future и Task из цикла событий asyncio
• Немедленное выполнение задач модулем asyncio
• Создание пулов потоков и процессов из цикла событий asyncio
• Создание TCP, UDP и Unix соединений из цикла событий asyncio
• Создание сетевых серверов из цикла событий asyncio
• Создание субпроцесса из цикла событий asyncio
• Работа с сокетами напрямую из цикла событий asyncio
• Передача файлов из цикла событий asyncio
• Наблюдение за дескрипторами файлов из цикла событий asyncio
• DNS запросы из цикла событий asyncio
• Сигналы Unix в циклах событий asyncio
• Параллелизм и многопоточность в цикле событий asyncio
• Объекты Transport и Protocol в цикле событий asyncio
• Включение режима отладки в asyncio
• Обработка исключений в цикле событий модуля asyncio
• Исключения модуля asyncio Python

Как ограничить время выполнения функции Python windows?

Использую eval() для вычисления мат. выражений. Если передать в нее что то типо 2**100**100, то, естественно, всё зависнет. Хочу ограничить время выполнения например пятью секундами.
Модуль signal не работает нормально на винде, с multiprocessing, threading, asyncio у меня как-то тоже реализовать не вышло. Буду очень благодарен за помощь.

• Вопрос задан вчера
• 44 просмотра

Комментировать
Решения вопроса 0
Ответы на вопрос 1

Dimonchik @dimonchik2013
non progredi est regredi

нужен не multiprocessing а subprocess
гугл subprocess popen timeout и вот это во все
это по-красивому

ну и классика — kill/pskill и т.п., отлавливание через OS и гроханье

Как ограничить время работы сценария?

Как ограничить время работы сценария? Например я поставил чтобы через 5 минут программа закрылась, она работает и как только 5 минут пройдет программа закроется. Как это можно реализовать? В windows.

Отслеживать
52.2k 11 11 золотых знаков 108 108 серебряных знаков 312 312 бронзовых знаков
задан 8 авг 2018 в 8:25
Alex Firsov Alex Firsov
551 6 6 серебряных знаков 20 20 бронзовых знаков
связанный вопрос Как прерывать функцию через определенное время при отсутствии голоса или шума?
8 авг 2018 в 17:50

3 ответа 3

Сортировка: Сброс на вариант по умолчанию

Код сценария выполняется в функции run . Функция join ждет указанное количество времени, блокируя дальнейшее выполнение кода в главном потоке. После окончания указанного времени скрипт напечатает в консоль сообщение и закроется.

Параметр daemon=True указывает, что поток при завершении главного потока также завершится:

import time import threading def run(): i = 1 # Бесконечный цикл while True: print i i += 1 time.sleep(1) if __name__ == '__main__': thread = threading.Thread(target=run) thread.daemon = True thread.start() thread.join(5 * 60) print 'Quit!' 

Python code samples /
Полезные кусочки
кода на Python

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

1. Сколько попыток необходимо совершить? 2, 5 или 20?
2. Хотим ли мы делать временные паузы между попытками? Если да, то какова длина этих пауз? Длина представляет собой неизменное значение (например, между всеми попытками мы ждем 5 секунд) или ее величина растет с увеличением числа попыток? То есть например, мы можем сначала делать попытки чаще, но потом все реже, чтобы не выполнять лишнюю работу и не закидывать бесполезными запросами еле живой сервер).
3. Что является условием завершения наших попыток? Самое первое и желаемое — попытка проходит успешно. Но что, если код продолжает падать, а положительного результата все нет? Необходимо решить, когда остановить попытки. Здесь два основных варианта: 1) ограничить их количество, 2) ограничить время — то есть задать timeout, после которого мы останавливаем ретраи.
4. Что делать, если все попытки оказались неуспешными? В большинстве случаев мы захотим упасть с ошибкой, но возможны и варианты, когда проблему можно проигнорировать и ограничиться выводом варнинга в консоль нашего приложения.

Retry method (Python)

 import time # Максимальное количество попыток RETRIES = 3 # Промежуток времени (60 сек), в течение которого мы будем ретраиться TIMEOUT = 60 # Частота, c которой будем ретраиться PERIOD = 5 # Функция-декоратор, которая модифицирует переданный ей метод # В нашем случае модификация заключается в повторении вызовов переданного метода def retry(max_retries, timeout, period): def outer(func): def inner(*args, **kwargs): # Задаем время, когда необходимо остановить попытки end_time = time.time() + timeout # Создаем счетчик, с каждой попыткой будем уменьшать его значение на 1 retries = max_retries # Бесконечно крутимся в цикле до тех пор, пока не случится одно из событий: # 1) метод func() выполнится успешно # 2) закончится отведенное на ретраи время # 3) исчерпаются попытки while True: try: return func(*args, **kwargs) except Exception as e: print(f'') if time.time() > end_time: raise 'Timeout has expired!' if retries == 1: raise e else: retries -= 1 print(f"Attempts left: ") print(f"Sleeping seconds . " ) time.sleep(period) return inner return outer # Протестируем наш декоратор на примере падающей функции @retry(RETRIES, TIMEOUT, PERIOD) def send_request(): raise Exception('Code block has failed. This is expected.') # Тесты send_request() 

1. Было падение c эксепшеном, переданным через параметр exception.
2. Было падение с любым типом эксепшена, но в тексте ошибки содержится нужная фраза (передается через параметр error_msg).

Modified retry (Python)

 def modified_retry(max_retries, timeout, exception=None, error_msg=None, period=1): def outer(func): def inner(*args, **kwargs): end_time = time.time() + timeout retries = max_retries while retries: try: return func(*args, **kwargs) except Exception as e: if (e and e == exception) or (error_msg and error_msg in str(e)): log.error(f'ERROR: ') if time.time() > end_time: log.error(f'Failed to successfully execute cmd during sec') raise e if retries == 1: log.error(f'Failed to successfully execute cmd with attempts') raise e retries -= 1 time.sleep(period) else: raise e return inner return outer # Будем делать повторные попытки, только если упали с TimeoutExpired @retry(5, 60, exception=subprocess.TimeoutExpired) def send_request(): raise TimeoutExpired('Operation timed out. This is expected.') # Тест send_request()