Есть функция. Нужно отслеживать время выполнения функции в реальном времени. То есть: выполняется функция и как только время выполнение функции превышает, скажем, 4 секунды, выполнение функции прекращалось
1 ответ
нормально прервать можно только процесс multiprocessing/subprocess
import multiprocessing, subprocess, time
def target():
time.sleep(10)
print(target)
def process_with_timeout_terminate(timeout: float, func: 'callable', *args, **kwargs):
p = multiprocessing.Process(target=func, args=args, kwargs=kwargs)
p.start()
p.join(timeout) # Wait until child process terminates
if p.is_alive():
p.terminate() # Terminate process; sends SIGTERM signal or uses TerminateProcess()
def programma_with_timeout_terminate(timeout: float, programma: str, *args):
p = subprocess.Popen([programma, *args])
try:
p.wait(timeout) # Wait for child process to terminate; returns self.returncode.
except subprocess.TimeoutExpired:
p.terminate() # Terminates the process.
if __name__ == '__main__':
process_with_timeout_terminate(timeout=4, func=target)
programma_with_timeout_terminate(timeout=4, programma='notepad.exe')
Если же требуется прервать выполнение в главном процессе или в потоке, то прервать функцию можно только если она генератор, т.е. в процессе своего выполнения она что-то возвращает используя yield. В этом случае прервать ее можно, но только после выполнения любого из таких yield. Однако слудкет помнить, что время которое может пройти между срабатываниями yield, может быть и больше timeout, т.е. не факт что прерывание выполнения произойдет точно в то время, как вы хотите.
import time
def target():
while True:
time.sleep(1)
print(target)
yield
def generator_with_timeout_terminate(timeout: float, func: 'callable', *args, **kwargs):
t = time.monotonic()
for _ in func(*args, **kwargs):
if (time.monotonic() - t) >= timeout:
break
if __name__ == '__main__':
generator_with_timeout_terminate(timeout=4, func=target)
Если требуется просто получить время выполнения куска кода или функции
import contextlib, time, datetime
def func(s):
time.sleep(s)
@contextlib.contextmanager
def get_time():
dt = {'start': datetime.datetime.now(), }
try:
yield dt
except:
raise
finally:
dt['stop'] = datetime.datetime.now()
dt['execute'] = (dt['stop'] - dt['start']).total_seconds()
if __name__ == '__main__':
with get_time() as dt:
func(3)
func(2)
time.sleep(1)
print(dt) # {'execute': 6.001126б 'start': datetime.datetime(2020, 4, 30, 11, 33, 7, 521648), 'stop': datetime.datetime(2020, 4, 30, 11, 33, 13, 522774)}
t = time.monotonic()
func(3)
func(2)
time.sleep(1)
print(time.monotonic() - t, 'сек') # 6.0 сек