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
2. Đây là một sự cố.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í
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
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
4 từ luồng con: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ủ
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
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
6, xem Cython, Python và Bàn phímTô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
0Khi 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
8.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
1Đố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.