1

Есть функция. Нужно отслеживать время выполнения функции в реальном времени. То есть: выполняется функция и как только время выполнение функции превышает, скажем, 4 секунды, выполнение функции прекращалось

1

1 ответ 1

2

нормально прервать можно только процесс 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 сек

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.