Hướng dẫn how do you timeout a function in python? - làm thế nào để bạn hết thời gian chờ một hàm trong python?

Làm cách nào để gọi chức năng hoặc tôi sẽ bọc nó trong đó để mất nhiều thời gian hơn 5 giây thì kịch bản hủy bỏ nó?

Tôi đã đăng một ý chính giải quyết câu hỏi/vấn đề này với một người trang trí và

try:
    range, _print = xrange, print
    def print(*args, **kwargs): 
        flush = kwargs.pop('flush', False)
        _print(*args, **kwargs)
        if flush:
            kwargs.get('file', sys.stdout).flush()            
except NameError:
    pass

Nhập khẩu và thiết lập để tương thích

Nó đã được thử nghiệm với Python 2 và 3. Nó cũng sẽ hoạt động trong Unix/Linux và Windows.

Đầu tiên là nhập khẩu. Những cố gắng giữ cho mã nhất quán bất kể phiên bản Python:

from __future__ import print_function
import sys
import threading
from time import sleep
try:
    import thread
except ImportError:
    import _thread as thread

Sử dụng mã độc lập phiên bản:

try:
    range, _print = xrange, print
    def print(*args, **kwargs): 
        flush = kwargs.pop('flush', False)
        _print(*args, **kwargs)
        if flush:
            kwargs.get('file', sys.stdout).flush()            
except NameError:
    pass

Bây giờ chúng tôi đã nhập chức năng của chúng tôi từ thư viện tiêu chuẩn.

try: range, _print = xrange, print def print(*args, **kwargs): flush = kwargs.pop('flush', False) _print(*args, **kwargs) if flush: kwargs.get('file', sys.stdout).flush() except NameError: pass 3 Trang trí

Tiếp theo, chúng ta cần một chức năng để chấm dứt

try:
    range, _print = xrange, print
    def print(*args, **kwargs): 
        flush = kwargs.pop('flush', False)
        _print(*args, **kwargs)
        if flush:
            kwargs.get('file', sys.stdout).flush()            
except NameError:
    pass

def quit_function(fn_name):
    # print to stderr, unbuffered in Python 2.
    print('{0} took too long'.format(fn_name), file=sys.stderr)
    sys.stderr.flush() # Python 3 stderr is likely buffered.
    thread.interrupt_main() # raises KeyboardInterrupt

Và đây là chính người trang trí:

def exit_after(s):
    '''
    use as decorator to exit process if 
    function takes longer than s seconds
    '''
    def outer(fn):
        def inner(*args, **kwargs):
            timer = threading.Timer(s, quit_function, args=[fn.__name__])
            timer.start()
            try:
                result = fn(*args, **kwargs)
            finally:
                timer.cancel()
            return result
        return inner
    return outer

Cách sử dụng

Và đây là cách sử dụng trực tiếp trả lời câu hỏi của bạn về việc thoát sau 5 giây!:

@exit_after(5)
def countdown(n):
    print('countdown started', flush=True)
    for i in range(n, -1, -1):
        print(i, end=', ', flush=True)
        sleep(1)
    print('countdown finished')

Demo:

>>> countdown(3)
countdown started
3, 2, 1, 0, countdown finished
>>> countdown(10)
countdown started
10, 9, 8, 7, 6, countdown took too long
Traceback (most recent call last):
  File "", line 1, in 
  File "", line 11, in inner
  File "", line 6, in countdown
KeyboardInterrupt

Cuộc gọi chức năng thứ hai sẽ không kết thúc, thay vào đó, quá trình sẽ thoát với một dấu vết!

try: range, _print = xrange, print def print(*args, **kwargs): flush = kwargs.pop('flush', False) _print(*args, **kwargs) if flush: kwargs.get('file', sys.stdout).flush() except NameError: pass 5 không phải lúc nào cũng dừng một sợi ngủ

Lưu ý rằng giấc ngủ sẽ không phải lúc nào cũng bị gián đoạn bởi ngắt bàn phím, trên Python 2 trên Windows, ví dụ:

@exit_after(1)
def sleep10():
    sleep(10)
    print('slept 10 seconds')

>>> sleep10()
sleep10 took too long         # Note that it hangs here about 9 more seconds
Traceback (most recent call last):
  File "", line 1, in 
  File "", line 11, in inner
  File "", line 3, in sleep10
KeyboardInterrupt

Nó cũng không có khả năng làm gián đoạn mã chạy trong các phần mở rộng trừ khi nó kiểm tra rõ ràng

try:
    range, _print = xrange, print
    def print(*args, **kwargs): 
        flush = kwargs.pop('flush', False)
        _print(*args, **kwargs)
        if flush:
            kwargs.get('file', sys.stdout).flush()            
except NameError:
    pass

Tôi sẽ tránh ngủ một chủ đề nhiều hơn một giây, trong mọi trường hợp - đó là EON trong bộ xử lý.

Làm cách nào để gọi chức năng hoặc tôi sẽ bọc nó trong đó để mất nhiều thời gian hơn 5 giây, tập lệnh hủy bỏ nó và làm gì khác?and does something else?

Để bắt nó và làm một cái gì đó khác, bạn có thể bắt gặp bàn phím.

>>> try:
...     countdown(10)
... except KeyboardInterrupt:
...     print('do something else')
... 
countdown started
10, 9, 8, 7, 6, countdown took too long
do something else

Thời gian chờ rất hữu ích khi bạn muốn giới hạn thời gian tối đa để gọi hàm hoặc chạy lệnh. Dưới đây là hai triển khai Python của tôi. (Tôi đã sử dụng Python 3.6.5).

Sử dụng đa bộ xử lý để hết thời gian một hàm Python

Sử dụng một quy trình để giữ thời gian/kiểm tra thời gian chờ và một quy trình khác để gọi chức năng Python này.

import time
from itertools import count
from multiprocessing import Processdef inc_forever():
    print('Starting function inc_forever()...')
    while True:
        time.sleep(1)
        print(next(counter))def return_zero():
    print('Starting function return_zero()...')
    return 0if __name__ == '__main__':
    # counter is an infinite iterator
    counter = count(0)    p1 = Process(target=inc_forever, name='Process_inc_forever')
    p2 = Process(target=return_zero, name='Process_return_zero')
    p1.start()
    p2.start()
    p1.join(timeout=5)
    p2.join(timeout=5)
    p1.terminate()
    p2.terminate()if p1.exitcode is None:
       print(f'Oops, {p1} timeouts!')if p2.exitcode == 0:
        print(f'{p2} is luck and finishes in 5 seconds!')

Đầu ra là như sau. Mã thoát cho biết chức năng là thời gian chờ (EXITCODE = none) hay hoàn thành.

Starting function inc_forever()...
Starting function return_zero()...
0
1
2
3
4
Oops,  timeouts!
 is luck and finishes in 5 seconds!

Sử dụng quy trình con để hết thời gian một lệnh bên ngoài

Kể từ Python 3.5, có một API tiện dụng và được đề xuất api trong mô-đun phụ, có hỗ trợ thời gian chờ tích hợp.

try:
    range, _print = xrange, print
    def print(*args, **kwargs): 
        flush = kwargs.pop('flush', False)
        _print(*args, **kwargs)
        if flush:
            kwargs.get('file', sys.stdout).flush()            
except NameError:
    pass

Khi hết thời gian, nó tăng ngoại lệ

try:
    range, _print = xrange, print
    def print(*args, **kwargs): 
        flush = kwargs.pop('flush', False)
        _print(*args, **kwargs)
        if flush:
            kwargs.get('file', sys.stdout).flush()            
except NameError:
    pass

try:
    range, _print = xrange, print
    def print(*args, **kwargs): 
        flush = kwargs.pop('flush', False)
        _print(*args, **kwargs)
        if flush:
            kwargs.get('file', sys.stdout).flush()            
except NameError:
    pass

Đối với Unix, có một cách khác để sử dụng mô-đun tín hiệu để báo hiệu một trình xử lý để tăng ngoại lệ sau 5 giây, nhưng tôi nghĩ rằng nó có một chút thấp và không đơn giản.

Làm thế nào để bạn làm một thời gian chờ trong Python?

Làm thế nào để bạn dừng một chức năng sau thời gian trong Python?

Làm thế nào để bạn làm cho Python đợi 5 giây?

Làm thế nào để bạn tăng thời gian chờ trong Python?