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

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

btw0 14 дек. 2008, в 18:03
Поделиться
Возможная копия Timeout при вызове функции
SmartManoj 08 май 2019, в 13:07

Поделиться:
multithreading

9 ответов

Лучший ответ

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

Итак, ваш код будет выглядеть примерно так.

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 14 дек. 2008, в 18:40
Поделиться
Что если я использую будильник для чего-то еще? 🙂
mat 21 дек. 2008, в 20:30
Сигналы и нити плохо смешиваются.
Miles 11 июль 2009, в 22:23

Да, наверное стоит выбрать один или другой.

Расчёт времени выполнения программы на python #short

Я сам парень процесса / сигнала. После просмотра blip.tv/file/2232410 я все меньше доверяю модели потоков Python.

rik.the.vik 14 июль 2009, в 16:45
Кроме того, это не отключает будильник впоследствии
Casebash 19 окт. 2009, в 12:14
для чего это сообщение?
user2372074 05 нояб. 2014, в 17:24
jfs 20 янв.

2016, в 08:37
iuradz 04 май 2016, в 08:17

У меня есть функция, которая просто спит (10), и я установил тайм-аут на 5 сек. Это не прерывает мою функцию сна.

drdot 06 июнь 2016, в 05:03
signal.alarm и signal.SIGALRM доступны только в Unix.
Logan 31 дек. 2017, в 01:39
Показать ещё 7 комментариев
Josh Lee 02 март 2009, в 03:27
Поделиться
try: yield n finally: signal.alarm(0)
jfs 19 окт. 2009, в 17:09
Чтобы быть очень суетливым, относится ли сигнал.тревога (секунды) к попытке?
Casebash 28 окт.

2009, в 11:00
Ну, это не задокументировано как поднятие каких-либо исключений, так что, вероятно, нет.
Josh Lee 28 окт. 2009, в 13:21
вместо последнего «Тайм-аут!» вы, вероятно, намеревались написать сообщение
Olga 21 дек.

2013, в 15:43

Я попытался реализовать этот метод в Python 3.4, и столкнулся с ошибкой, с которой мне трудно справиться: AttributeError: exit Функция обратного отслеживания обеспечивает только это: Файл «», строка 1, в . исправить эту проблему?

David Scott IV 05 июль 2016, в 15:29
Показать ещё 3 комментария

Здесь используется способ 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

Ariel Cabib 30 окт. 2014, в 22:15
Поделиться

Управление временем на python | Модуль time python (питон)

Отлично работает на моем ноутбуке с Linux, не работает на Windows и почти не работает на OSX. Не твоя вина, просто замечательный мир программирования.

Ameo 20 май 2016, в 19:04
Ariel Cabib 04 сен. 2016, в 09:52

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

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

Если ваше исключение создано здесь(), временный файл никогда не будет удален.

Решение здесь заключается в том, что асинхронные исключения переносятся до тех пор, пока код не будет находиться внутри кода обработки исключений (исключающий или окончательный блок), но Python этого не делает.

Обратите внимание, что это не прерывает ничего при выполнении собственного кода; он только прервет его, когда функция вернется, поэтому это может не помочь этому конкретному случаю. (Сам SIGALRM может прервать вызов, который блокирует — но код сокета обычно просто повторяет попытку после EINTR.)

Выполнение этого с помощью потоков — лучшая идея, поскольку она более портативна, чем сигналы. Поскольку вы начинаете рабочий поток и блокируете его до тех пор, пока он не завершится, нет никаких обычных concurrency проблем. К сожалению, не существует возможности для асинхронного предоставления исключения в другой поток в Python (это могут использовать другие API-интерфейсы потоков). Он также будет иметь ту же проблему с отправкой исключения во время обработчика исключений и требует того же исправления.

Glenn Maynard 11 июль 2009, в 22:09
Поделиться

нативный код может вызывать PyErr_CheckSignals() перед перезапуском в EINTR, чтобы разрешить запуск обработчика сигналов Python.

jfs 22 фев. 2016, в 11:44

Я предпочитаю подход контекстного менеджера, потому что он позволяет выполнять несколько операторов python внутри оператора with time_limit . Поскольку в системе Windows нет SIGALARM , более переносимый и, возможно, более простой метод может использовать Timer

Ключевым моментом здесь является использование _thread.interrupt_main для прерывания основного потока из потока таймера. Одно из предостережений состоит в том, что основная нить не всегда реагирует на KeyboardInterrupt , поднятый Timer быстро. Например, time.sleep() вызывает системную функцию, поэтому KeyboardInterrupt будет обрабатываться после вызова sleep .

user2283347 06 июнь 2016, в 03:07
Поделиться

Это кажется самым чистым. Есть ли способ отличить прерывание от прерывания клавиатуры, отправленного пользователем?

Lycan 13 июль 2017, в 22:17

Кроме того, это уничтожит все потоки и не подходит как таймер для уничтожения свободных потоков в многопоточной операции.

Alexander McFarlane 18 янв. 2018, в 16:15

Поскольку модуль thread был переименован в _thread в Python 3, для совместимости с Python 2 вам необходимо сделать: import sys if sys.version_info[0] < 3: from thread import interrupt_main else: from _thread import interrupt_main а затем: timer = threading.Timer(seconds, lambda: interrupt_main())

Roland Pihlakas 19 май 2018, в 18:03
Это на самом деле сработало
dipl0 26 июнь 2018, в 14:24
Показать ещё 2 комментария

Единственным «безопасным» способом сделать это на любом языке является использование вторичного процесса для выполнения этого тайм-аута, в противном случае вам нужно создать свой код таким образом, чтобы он безопасно выполнял время самостоятельно, например, путем проверки времени, прошедшего в цикле или аналогичном. Если изменение метода не является вариантом, поток будет недостаточным.

Почему? Потому что вы рискуете оставить вещи в плохом состоянии, когда вы это делаете. Если поток просто убит средним методом, удерживаемые блокировки и т.д. Будут просто удерживаться и не могут быть освобождены.

Итак, посмотрите на процесс, не смотрите на поток.

Lasse Vågsæther Karlsen 14 дек. 2008, в 18:00
Поделиться

Вам не нужно использовать потоки. Вы можете использовать другой процесс для выполнения блокировки, например, возможно, используя модуль subprocess. Если вы хотите разделить структуры данных между различными частями вашей программы, то Twisted — отличная библиотека для того, чтобы дать вам контроль над этим, и я ‘d рекомендовать его, если вы заботитесь о блокировке и ожидаете, что эта проблема будет много. Плохая новость с Twisted — вам нужно переписать свой код, чтобы избежать блокировки, и есть справедливая кривая обучения.

Вы можете использовать потоки, чтобы избежать блокировки, но я рассматривал бы это как последнее средство, поскольку оно предоставляет вам целый мир боли. Прочтите хорошую книгу на concurrency, прежде чем даже подумать об использовании потоков в производстве, например. Жан Бэкон «Параллельные системы». Я работаю с кучей людей, которые действительно охлаждают высокопроизводительные материалы с помощью потоков, и мы не вводим темы в проекты, если они нам действительно не нужны.

Dickon Reed 14 дек. 2008, в 19:07
Поделиться

Здесь функция тайм-аута, я думаю, что нашел через google, и это работает для меня.

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
monkut 15 дек. 2008, в 05:10
Поделиться

Почему обе части блока возврата if / else возвращают it.result?
NorthIsUp 13 сен. 2010, в 20:58

Это было мое редактирование для конкретной проблемы, с которой я имел дело. Я не помню почему.

В любом случае, оригинал вернул «default», если it.isAlive ().

monkut 16 сен. 2010, в 01:22

Но если кто-то заинтересован в том, чтобы это реализовано как декоратор, вот альтернатива.

Вот как это будет выглядеть:

И это модуль timeout.py :

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

Обратите внимание, что даже если выбрано TimeoutError , декорированный метод будет продолжать работать в другом потоке. Если вы также хотите, чтобы этот поток был «остановлен», см.: Есть ли способ убить поток в Python?

Seba 27 янв. 2016, в 14:30
Поделиться

1 — класс InterruptableThread без метода прерывания. 2 — даже после окончания выполнения поток все еще выполняет функцию

Sergey11g 23 окт. 2017, в 12:32

Ещё вопросы

Источник: overcoder.net

Класс Timer() модуля threading в Python.

Класс Timer() модуля threading создает таймер, который будет запускать функцию function с аргументами args и ключевыми аргументами kwargs по прошествии интервала interval секунд.

Если args равен None (по умолчанию), то будет использоваться пустой список. Если kwargs равен None (по умолчанию), то будет использоваться пустой словарь.

Этот класс представляет действие, которое следует запускать только по прошествии определенного времени — таймер. Таймер является подклассом threading.Thread() и, как таковой, также служит примером создания пользовательских потоков.

Таймеры запускаются, как и потоки, путем вызова их метода Timer.start() , унаследованного от класса threading.Thread() .

Таймер можно остановить до того, как его действие начнется, вызвав метод Timer.cancel() . Интервал, который таймер будет ожидать перед выполнением своего действия, может не совпадать с интервалом, указанным пользователем.

def hello(): print(«hello, world») t = Timer(30.0, hello) t.start() # через 30 секунд будет напечатано «hello, world».

Методы объекта Timer .

Объект Timer дополнительно определяет один метод. Остальные методы он наследует от класса threading.Thread() .

Timer.cancel() :

Метод Timer.cancel() останавливает таймер и отменяет выполнение действия таймера. Метод будет работать только в том случае, если таймер все еще находится в стадии ожидания.

Сработавший таймер остановить нельзя.

Примеры создания и запуска потоков-таймеров.

Таймер начинает свою работу после задержки и может быть отменен в любой момент в течение этого времени задержки.

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

Так как первый таймер это не поток демона, то он присоединяется неявно, когда основной поток завершится, по этому delayed() сообщение не напечатает.

import threading, time def delayed(): th_name = threading.current_thread().name print(f’Th:th_name> Worker запущен’) # Создание и запуск потоков таймеров t1 = threading.Timer(0.3, delayed) t1.name = ‘Timer-1’ t2 = threading.Timer(0.3, delayed) t2.name = ‘Timer-2’ print(‘Запуск таймеров’) t1.start() t2.start() print(f’Ожидание перед завершением t2.name>’) time.sleep(0.2) print(f’Завершение t2.name>’) t2.cancel() print(‘Выполнено’) # Запуск таймеров # Ожидание перед завершением Timer-2 # Завершение Timer-2 # Выполнено # Th:Timer-1 Worker запущен

• КРАТКИЙ ОБЗОР МАТЕРИАЛА.
• Получение общих сведений о потоках, модуль threading.
• Класс Thread() модуля threading.
• Класс local() модуля threading.
• Класс Event() модуля threading.
• Класс Lock() модуля threading.
• Класс RLock() модуля threading.
• Класс Condition() модуля threading.
• Класс Semaphore() модуля threading.
• Класс Timer() модуля threading.
• Класс Barrier() модуля threading.
• Протокол управления контекстом в модуле threading.
• Трассировка и профилирование потоков модулем threading.
• Как перезапускать потоки?

Источник: docs-python.ru

Python-сообщество

• Начало
• » Python для новичков
• » Как ограничить время выполнения скрипта?

#1 Окт. 22, 2010 13:01:52

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

Добрый День!
Столкнулся с такой ситуацией.

for row in cursor.fetchall():
os.system(«python generate.py » + row[0])
time.sleep(1)
try:
cursor1.execute(«»»UPDATE links SET created = 1 WHERE domain = %s»»», (row[0]) );
db1.commit()
except:
db1.rollback()

В строке os.system(“python generate.py ” + row)
запускается generate.py, который открывает оконное приложение (gtk, gtkmozembed) и выполняет различные действия.
Иногда этот скрипт отрабатывает слишком долго.
1) Как сделать так, чтобы скрипт при превышении допустим 30 секунд переходил на следующую итерацию?
2) Как можно в оконном приложении (в generate.py) отключить появление алертов? (предупреждающие сообщения с кнопкой OK)

Источник: python.su