Hướng dẫn how do i limit a run time in python? - làm cách nào để giới hạn thời gian chạy trong python?

Một cải tiến trên @RIK.The.Vik của câu trả lời sẽ là sử dụng tuyên bố

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!"

4 để cung cấp chức năng hết thời gian một số đường cú pháp:

import signal
from contextlib import contextmanager

class TimeoutException(Exception): pass

@contextmanager
def time_limit(seconds):
    def signal_handler(signum, frame):
        raise TimeoutException("Timed out!")
    signal.signal(signal.SIGALRM, signal_handler)
    signal.alarm(seconds)
    try:
        yield
    finally:
        signal.alarm(0)


try:
    with time_limit(10):
        long_function_call()
except TimeoutException as e:
    print("Timed out!")

Đã trả lời ngày 2 tháng 3 năm 2009 lúc 3:14Mar 2, 2009 at 3:14

Josh Leejosh LeeJosh Lee

165K37 Huy hiệu vàng267 Huy hiệu bạc271 Huy hiệu Đồng37 gold badges267 silver badges271 bronze badges

7

Tôi không chắc điều này có thể làm thế nào, nhưng sử dụng tín hiệu và báo động có thể là một cách tốt để xem xét điều này. Với một công việc nhỏ, bạn cũng có thể làm điều này hoàn toàn chung chung và có thể sử dụng trong mọi tình huống.

http://docs.python.org/library/signal.html

Vì vậy, mã của bạn sẽ trông giống như thế này.

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!"

Đã trả lời ngày 14 tháng 12 năm 2008 lúc 17:27Dec 14, 2008 at 17:27

rik.the.vikrik.the.vikrik.the.vik

9096 Huy hiệu bạc6 Huy hiệu Đồng6 silver badges6 bronze badges

7

Đây là một cách Linux/OSX để hạn chế thời gian chạy của hàm. Điều này trong trường hợp bạn không muốn sử dụng các luồng và muốn chương trình của bạn đợi cho đến khi chức năng kết thúc hoặc giới hạn thời gian hết hạn.

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

Đã trả lời ngày 30 tháng 10 năm 2014 lúc 22:05Oct 30, 2014 at 22:05

Ariel Cabibariel CabibAriel Cabib

2.01218 Huy hiệu bạc14 Huy hiệu đồng18 silver badges14 bronze badges

3

Tôi thích cách tiếp cận trình quản lý ngữ cảnh vì nó cho phép thực hiện nhiều câu lệnh Python trong câu lệnh

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!"

5. Bởi vì hệ thống Windows không có

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!"

6, một phương pháp di động hơn và có lẽ đơn giản hơn có thể sử dụng

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!"

7

from contextlib import contextmanager
import threading
import _thread

class TimeoutException(Exception):
    def __init__(self, msg=''):
        self.msg = msg

@contextmanager
def time_limit(seconds, msg=''):
    timer = threading.Timer(seconds, lambda: _thread.interrupt_main())
    timer.start()
    try:
        yield
    except KeyboardInterrupt:
        raise TimeoutException("Timed out for operation {}".format(msg))
    finally:
        # if the action ends in specified time, timer is canceled
        timer.cancel()

import time
# ends after 5 seconds
with time_limit(5, 'sleep'):
    for i in range(10):
        time.sleep(1)

# this will actually end after 10 seconds
with time_limit(5, 'sleep'):
    time.sleep(10)

Kỹ thuật chính ở đây là việc sử dụng

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!"

8 để làm gián đoạn chủ đề chính từ luồng hẹn giờ. Một cảnh báo là chủ đề chính không phải lúc nào cũng phản ứng với

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!"

9 được tăng lên bởi

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!"

7 một cách nhanh chóng. Ví dụ:

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

1 gọi chức năng hệ thống để

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!"

9 sẽ được xử lý sau cuộc gọi

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

3.

Đã trả lời ngày 6 tháng 6 năm 2016 lúc 1:41Jun 6, 2016 at 1:41

user2283347user2283347user2283347

5607 Huy hiệu bạc11 Huy hiệu đồng7 silver badges11 bronze badges

3

Dưới đây: Một cách đơn giản để có được hiệu ứng mong muốn:

https://pypi.org/project/func-timeout

Điều này đã cứu mạng tôi.

Và bây giờ là một ví dụ về cách thức hoạt động: giả sử bạn có một danh sách lớn các mục được xử lý và bạn đang lặp lại chức năng của mình trên các mục đó. Tuy nhiên, vì một số lý do kỳ lạ, chức năng của bạn bị mắc kẹt trên mục N, mà không cần phải tăng ngoại lệ. Bạn cần phải xử lý các mặt hàng khác, càng tốt. Trong trường hợp này, bạn có thể đặt thời gian chờ để xử lý từng mục:

import time
import func_timeout


def my_function(n):
    """Sleep for n seconds and return n squared."""
    print(f'Processing {n}')
    time.sleep(n)
    return n**2


def main_controller(max_wait_time, all_data):
    """
    Feed my_function with a list of itens to process (all_data).

    However, if max_wait_time is exceeded, return the item and a fail info.
    """
    res = []
    for data in all_data:
        try:
            my_square = func_timeout.func_timeout(
                max_wait_time, my_function, args=[data]
                )
            res.append((my_square, 'processed'))
        except func_timeout.FunctionTimedOut:
            print('error')
            res.append((data, 'fail'))
            continue

    return res


timeout_time = 2.1  # my time limit
all_data = range(1, 10)  # the data to be processed

res = main_controller(timeout_time, all_data)
print(res)

Đã trả lời ngày 10 tháng 2 năm 2021 lúc 19:40Feb 10, 2021 at 19:40

Erickfiserickfiserickfis

93411 Huy hiệu bạc19 Huy hiệu đồng11 silver badges19 bronze badges

4

Làm điều này từ bên trong một người xử lý tín hiệu là nguy hiểm: bạn có thể ở trong một người xử lý ngoại lệ tại thời điểm ngoại lệ được nâng lên và để mọi thứ ở trạng thái bị hỏng. Ví dụ,

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

Nếu ngoại lệ của bạn được nâng lên ở đây (), tệp tạm thời sẽ không bao giờ bị xóa.

Giải pháp ở đây là cho các ngoại lệ không đồng bộ sẽ bị hoãn lại cho đến khi mã không nằm trong mã xử lý ngoại lệ (ngoại trừ hoặc cuối cùng là khối), nhưng Python không làm điều đó.

Lưu ý rằng điều này sẽ không làm gián đoạn bất cứ điều gì trong khi thực thi mã gốc; Nó sẽ chỉ làm gián đoạn nó khi hàm quay trở lại, vì vậy điều này có thể không giúp ích gì cho trường hợp cụ thể này. .

Làm điều này với các chủ đề là một ý tưởng tốt hơn, vì nó di động hơn tín hiệu. Vì bạn đang bắt đầu một chủ đề công nhân và chặn cho đến khi nó kết thúc, không có ai trong số những lo lắng đồng thời thông thường. Thật không may, không có cách nào để cung cấp một ngoại lệ không đồng bộ đến một chủ đề khác trong Python (API luồng khác có thể làm điều này). Nó cũng sẽ có cùng một vấn đề với việc gửi một ngoại lệ trong một trình xử lý ngoại lệ và yêu cầu sửa chữa tương tự.

Đã trả lời ngày 11 tháng 7 năm 2009 lúc 20:30Jul 11, 2009 at 20:30

Glenn Maynardglenn MaynardGlenn Maynard

54.4K10 Huy hiệu vàng117 Huy hiệu bạc131 Huy hiệu đồng10 gold badges117 silver badges131 bronze badges

1

Bạn không phải sử dụng các chủ đề. Bạn có thể sử dụng một quy trình khác để thực hiện công việc chặn, ví dụ, có thể sử dụng mô -đun phụ. Nếu bạn muốn chia sẻ cấu trúc dữ liệu giữa các phần khác nhau trong chương trình của bạn thì Twisted là một thư viện tuyệt vời để tự kiểm soát điều này và tôi khuyên bạn nên quan tâm đến việc chặn và mong đợi gặp rắc rối này rất nhiều. Tin xấu với Twisted là bạn phải viết lại mã của mình để tránh mọi chặn, và có một đường cong học tập công bằng.

Bạn có thể sử dụng các chủ đề để tránh chặn, nhưng tôi coi đây là phương sách cuối cùng, vì nó cho bạn tiếp xúc với cả một thế giới đau đớn. Đọc một cuốn sách hay về đồng thời trước khi nghĩ về việc sử dụng các chủ đề trong sản xuất, ví dụ: "Hệ thống đồng thời" của Jean Bacon. Tôi làm việc với một loạt những người thực hiện công cụ hiệu suất cao thực sự tuyệt vời với các chủ đề và chúng tôi không đưa chủ đề vào các dự án trừ khi chúng tôi thực sự cần chúng.

Đã trả lời ngày 14 tháng 12 năm 2008 lúc 17:13Dec 14, 2008 at 17:13

Dickon Reeddickon ReedDickon Reed

3.5152 Huy hiệu vàng23 Huy hiệu bạc25 Huy hiệu Đồng2 gold badges23 silver badges25 bronze badges

Cách "an toàn" duy nhất để thực hiện việc này, trong bất kỳ ngôn ngữ nào, là sử dụng quy trình thứ cấp để thực hiện thời gian chờ đó, nếu không bạn cần xây dựng mã của mình theo cách mà nó sẽ tự mình ra một cách an toàn, ví dụ như Kiểm tra thời gian trôi qua trong một vòng lặp hoặc tương tự. Nếu thay đổi phương thức không phải là một tùy chọn, một luồng sẽ không đủ.

Tại sao? Bởi vì bạn đang mạo hiểm để lại mọi thứ trong tình trạng tồi tệ khi bạn làm. Nếu chủ đề đơn giản bị giết giữa phương pháp, khóa được giữ, v.v ... sẽ được giữ và không thể được phát hành.

Vì vậy, hãy nhìn vào quá trình, không nhìn vào cách chủ đề.

Đã trả lời ngày 14 tháng 12 năm 2008 lúc 17:20Dec 14, 2008 at 17:20

Lasse V. Karlsenlasse V. KarlsenLasse V. Karlsen

372K98 Huy hiệu vàng623 Huy hiệu bạc809 Huy hiệu Đồng98 gold badges623 silver badges809 bronze badges

Tôi thường thích sử dụng trình quản lý bối cảnh theo đề xuất của @josh-le

Nhưng trong trường hợp ai đó quan tâm đến việc thực hiện điều này như một người trang trí, đây là một giải pháp thay thế.

Đây là cách nó sẽ trông như thế nào:

import time
from timeout import timeout

class Test(object):
    @timeout(2)
    def test_a(self, foo, bar):
        print foo
        time.sleep(1)
        print bar
        return 'A Done'

    @timeout(2)
    def test_b(self, foo, bar):
        print foo
        time.sleep(3)
        print bar
        return 'B Done'

t = Test()
print t.test_a('python', 'rocks')
print t.test_b('timing', 'out')

Và đây là mô -đun

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

4:

import threading

class TimeoutError(Exception):
    pass

class InterruptableThread(threading.Thread):
    def __init__(self, func, *args, **kwargs):
        threading.Thread.__init__(self)
        self._func = func
        self._args = args
        self._kwargs = kwargs
        self._result = None

    def run(self):
        self._result = self._func(*self._args, **self._kwargs)

    @property
    def result(self):
        return self._result


class timeout(object):
    def __init__(self, sec):
        self._sec = sec

    def __call__(self, f):
        def wrapped_f(*args, **kwargs):
            it = InterruptableThread(f, *args, **kwargs)
            it.start()
            it.join(self._sec)
            if not it.is_alive():
                return it.result
            raise TimeoutError('execution expired')
        return wrapped_f

Đầu ra:

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

Lưu ý rằng ngay cả khi

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

5 bị ném, phương pháp được trang trí sẽ tiếp tục chạy trong một luồng khác. Nếu bạn cũng muốn chủ đề này được "dừng", hãy xem: Có cách nào để giết một chủ đề trong Python không?

Đã trả lời ngày 27 tháng 1 năm 2016 lúc 13:38Jan 27, 2016 at 13:38

SebasebaSeba

5343 Huy hiệu bạc10 Huy hiệu Đồng3 silver badges10 bronze badges

3

Sử dụng máy trang trí đơn giản

Đây là phiên bản tôi đã thực hiện sau khi nghiên cứu câu trả lời ở trên. Khá thẳng về phía trước.

def function_timeout(seconds: int):
    """Wrapper of Decorator to pass arguments"""

    def decorator(func):
        @contextmanager
        def time_limit(seconds_):
            def signal_handler(signum, frame):  # noqa
                raise TimeoutException(f"Timed out in {seconds_} seconds!")

            signal.signal(signal.SIGALRM, signal_handler)
            signal.alarm(seconds_)
            try:
                yield
            finally:
                signal.alarm(0)

        @wraps(func)
        def wrapper(*args, **kwargs):
            with time_limit(seconds):
                return func(*args, **kwargs)

        return wrapper

    return decorator

Làm thế nào để sử dụng?

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!"

0

Tất nhiên, đừng quên nhập chức năng nếu nó nằm trong một tệp riêng biệt.

Đã trả lời ngày 21 tháng 4 lúc 18:47Apr 21 at 18:47

Ali Sajjadali SajjadAli Sajjad

2.60922 huy hiệu bạc34 huy hiệu đồng22 silver badges34 bronze badges

Đây là một chức năng thời gian chờ tôi nghĩ rằng tôi đã tìm thấy thông qua Google và nó hoạt động cho tôi.

Từ: http://code.activestate.com/recipes/473878/

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!"

1

Đã trả lời ngày 15 tháng 12 năm 2008 lúc 4:41Dec 15, 2008 at 4:41

Monkutmonkutmonkut

40.5K23 Huy hiệu vàng120 Huy hiệu bạc148 Huy hiệu đồng23 gold badges120 silver badges148 bronze badges

3

Phương pháp từ @user2283347 được kiểm tra hoạt động, nhưng chúng tôi muốn loại bỏ các tin nhắn theo dõi. Sử dụng thủ thuật vượt qua từ loại bỏ Traceback trong Python trên CTRL-C, mã đã được sửa đổi là:

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!"

2

Chúng ta có thể thấy rằng phương pháp này có thể cần rất lâu để làm gián đoạn tính toán, chúng tôi đã yêu cầu trong 5 giây, nhưng nó hoạt động trong 5 giờ.

Đã trả lời ngày 11 tháng 9 năm 2020 lúc 22:37Sep 11, 2020 at 22:37

FrankfrankFrank

4974 Huy hiệu bạc14 Huy hiệu đồng4 silver badges14 bronze badges

Mã này hoạt động cho Windows Server Datacenter 2016 với Python 3.7.3 và tôi đã không thử nghiệm trên UNIX, sau khi trộn một số câu trả lời từ Google và Stackoverflow, cuối cùng nó cũng hiệu quả với tôi như thế này:

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!"

3

Mã chính là từ liên kết này: Tạo hai quy trình con bằng Python (Windows)

Sau đó, tôi đã sử dụng

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

6 để giết quá trình trẻ em. Bạn có thể thấy rằng hàm F gọi 2 bản in, một lần sau 5 giây và một sau 10 giây. Tuy nhiên, với giấc ngủ 7 giây và chấm dứt (), nó không hiển thị bản in cuối cùng.

Nó làm việc cho tôi, hy vọng nó sẽ giúp!

Đã trả lời ngày 4 tháng 5 lúc 12:21May 4 at 12:21

Làm cách nào để giữ giới hạn thời gian trong Python?

Làm thế nào để bạn ngăn chặn một chức năng Python chạy?