from threading import Thread
import functools
def timeout(timeout):
def deco(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
res = [Exception('function [%s] timeout [%s seconds] exceeded!' % (func.__name__, timeout))]
def newFunc():
try:
res[0] = func(*args, **kwargs)
except Exception as e:
res[0] = e
t = Thread(target=newFunc)
t.daemon = True
try:
t.start()
t.join(timeout)
except Exception as je:
print ('error starting thread')
raise je
ret = res[0]
if isinstance(ret, BaseException):
raise ret
return ret
return wrapper
return deco
Sau đó, làm điều gì đó như thế này:
func = timeout(timeout=16)(MyModule.MyFunc)
try:
func()
except:
pass #handle errors here
Bạn có thể sử dụng máy trang trí này ở bất cứ nơi nào bạn cần với một cái gì đó như:
@timeout(60)
def f():
...
Hiệu suất không đồng bộ
Hiệu suất chức năng nhanh chóngShow
Tôi có chức năng trang trí sáng tạo hết thời gian sau đây:
Nội dung chính Show
Bộ trang trí thời gian chờ cho các chức năng đồng bộ và không đồng bộ.
Được viết bằng import inspect class withHostAndToken(object): __name__ = "withHostAndToken" __doc__ = "Get the Host and Token for the API call" def __init__(self, func): print "withHostAndToken: \t" + func.__name__ self.func = func self.HOST = '' self.TOKEN = '' def __call__(self,*args, **kwds): if self.HOST == '': self.HOST = "HOST" if self.TOKEN == '': self.TOKEN = "TOKEN" argsspec = inspect.getargspec(self.func) function_args = argsspec[0] if 'HOST' in function_args: if 'TOKEN' in function_args: return self.func(self.HOST , self.TOKEN , *args, **kwds) else: return self.func(self.HOST , *args, **kwds) elif 'TOKEN' in function_args: return self.func(self.TOKEN, *args, **kwds) 1 thuần túy và không có sự phụ thuộc nào ngoài các thư viện cơ sở.
Từ import inspect class withHostAndToken(object): __name__ = "withHostAndToken" __doc__ = "Get the Host and Token for the API call" def __init__(self, func): print "withHostAndToken: \t" + func.__name__ self.func = func self.HOST = '' self.TOKEN = '' def __call__(self,*args, **kwds): if self.HOST == '': self.HOST = "HOST" if self.TOKEN == '': self.TOKEN = "TOKEN" argsspec = inspect.getargspec(self.func) function_args = argsspec[0] if 'HOST' in function_args: if 'TOKEN' in function_args: return self.func(self.HOST , self.TOKEN , *args, **kwds) else: return self.func(self.HOST , *args, **kwds) elif 'TOKEN' in function_args: return self.func(self.TOKEN, *args, **kwds) 2:
Trường hợp cuối cùng này rất hữu ích nếu hàm import inspect class withHostAndToken(object): __name__ = "withHostAndToken" __doc__ = "Get the Host and Token for the API call" def __init__(self, func): print "withHostAndToken: \t" + func.__name__ self.func = func self.HOST = '' self.TOKEN = '' def __call__(self,*args, **kwds): if self.HOST == '': self.HOST = "HOST" if self.TOKEN == '': self.TOKEN = "TOKEN" argsspec = inspect.getargspec(self.func) function_args = argsspec[0] if 'HOST' in function_args: if 'TOKEN' in function_args: return self.func(self.HOST , self.TOKEN , *args, **kwds) else: return self.func(self.HOST , *args, **kwds) elif 'TOKEN' in function_args: return self.func(self.TOKEN, *args, **kwds) 3 đã có trong không gian tên của bạn.
: CẢNH BÁO: CẢNH BÁO: Khi một hàm hết thời gian, một ngoại lệ được nêu ra nhưng hủy bỏ không được đảm bảo. Người trang trí này chỉ thông báo cho người dùng khi đủ thời gian trôi qua kể từ khi gọi chức năng. Xử lý tình huống và đảm bảo hủy bỏ tùy thuộc vào người dùng.
Bộ trang trí thời gian chờ cho các chức năng đồng bộ và không đồng bộ. timeout: withHostAndToken timeout wrapper: withHostAndToken
Được viết bằng import inspect class withHostAndToken(object): __name__ = "withHostAndToken" __doc__ = "Get the Host and Token for the API call" def __init__(self, func): print "withHostAndToken: \t" + func.__name__ self.func = func self.HOST = '' self.TOKEN = '' def __call__(self,*args, **kwds): if self.HOST == '': self.HOST = "HOST" if self.TOKEN == '': self.TOKEN = "TOKEN" argsspec = inspect.getargspec(self.func) function_args = argsspec[0] if 'HOST' in function_args: if 'TOKEN' in function_args: return self.func(self.HOST , self.TOKEN , *args, **kwds) else: return self.func(self.HOST , *args, **kwds) elif 'TOKEN' in function_args: return self.func(self.TOKEN, *args, **kwds) 1 thuần túy và không có sự phụ thuộc nào ngoài các thư viện cơ sở.
Hiệu suất không đồng bộ
Tôi có chức năng trang trí sáng tạo hết thời gian sau đây:
Nội dung chính Show
Bộ trang trí thời gian chờ cho các chức năng đồng bộ và không đồng bộ.
Được viết bằng import inspect class withHostAndToken(object): __name__ = "withHostAndToken" __doc__ = "Get the Host and Token for the API call" def __init__(self, func): print "withHostAndToken: \t" + func.__name__ self.func = func self.HOST = '' self.TOKEN = '' def __call__(self,*args, **kwds): if self.HOST == '': self.HOST = "HOST" if self.TOKEN == '': self.TOKEN = "TOKEN" argsspec = inspect.getargspec(self.func) function_args = argsspec[0] if 'HOST' in function_args: if 'TOKEN' in function_args: return self.func(self.HOST , self.TOKEN , *args, **kwds) else: return self.func(self.HOST , *args, **kwds) elif 'TOKEN' in function_args: return self.func(self.TOKEN, *args, **kwds) 1 thuần túy và không có sự phụ thuộc nào ngoài các thư viện cơ sở.
pip install .
Từ import inspect class withHostAndToken(object): __name__ = "withHostAndToken" __doc__ = "Get the Host and Token for the API call" def __init__(self, func): print "withHostAndToken: \t" + func.__name__ self.func = func self.HOST = '' self.TOKEN = '' def __call__(self,*args, **kwds): if self.HOST == '': self.HOST = "HOST" if self.TOKEN == '': self.TOKEN = "TOKEN" argsspec = inspect.getargspec(self.func) function_args = argsspec[0] if 'HOST' in function_args: if 'TOKEN' in function_args: return self.func(self.HOST , self.TOKEN , *args, **kwds) else: return self.func(self.HOST , *args, **kwds) elif 'TOKEN' in function_args: return self.func(self.TOKEN, *args, **kwds) 2:
pip install timeout-function-decorator
Trường hợp cuối cùng này rất hữu ích nếu hàm import inspect class withHostAndToken(object): __name__ = "withHostAndToken" __doc__ = "Get the Host and Token for the API call" def __init__(self, func): print "withHostAndToken: \t" + func.__name__ self.func = func self.HOST = '' self.TOKEN = '' def __call__(self,*args, **kwds): if self.HOST == '': self.HOST = "HOST" if self.TOKEN == '': self.TOKEN = "TOKEN" argsspec = inspect.getargspec(self.func) function_args = argsspec[0] if 'HOST' in function_args: if 'TOKEN' in function_args: return self.func(self.HOST , self.TOKEN , *args, **kwds) else: return self.func(self.HOST , *args, **kwds) elif 'TOKEN' in function_args: return self.func(self.TOKEN, *args, **kwds) 3 đã có trong không gian tên của bạn.
func = timeout(timeout=16)(MyModule.MyFunc)
try:
func()
except:
pass #handle errors here
0
Nhập mô -đun
Trường hợp cuối cùng này rất hữu ích nếu hàm import inspect class withHostAndToken(object): __name__ = "withHostAndToken" __doc__ = "Get the Host and Token for the API call" def __init__(self, func): print "withHostAndToken: \t" + func.__name__ self.func = func self.HOST = '' self.TOKEN = '' def __call__(self,*args, **kwds): if self.HOST == '': self.HOST = "HOST" if self.TOKEN == '': self.TOKEN = "TOKEN" argsspec = inspect.getargspec(self.func) function_args = argsspec[0] if 'HOST' in function_args: if 'TOKEN' in function_args: return self.func(self.HOST , self.TOKEN , *args, **kwds) else: return self.func(self.HOST , *args, **kwds) elif 'TOKEN' in function_args: return self.func(self.TOKEN, *args, **kwds) 3 đã có trong không gian tên của bạn.
Cách sử dụng
func = timeout(timeout=16)(MyModule.MyFunc)
try:
func()
except:
pass #handle errors here
1
Sử dụng người trang trí đơn giản như:
Như bạn có thể nhận thấy, người trang trí yêu cầu dấu ngoặc ngay cả khi không có thông số nào được thông qua.
func = timeout(timeout=16)(MyModule.MyFunc)
try:
func()
except:
pass #handle errors here
2
Kết quả tương tự có thể thu được cho các chức năng không đồng bộ:
import inspect
class withHostAndToken(object):
__name__ = "withHostAndToken"
__doc__ = "Get the Host and Token for the API call"
def __init__(self, func):
print "withHostAndToken: \t" + func.__name__
self.func = func
self.HOST = ''
self.TOKEN = ''
def __call__(self,*args, **kwds):
if self.HOST == '':
self.HOST = "HOST"
if self.TOKEN == '':
self.TOKEN = "TOKEN"
argsspec = inspect.getargspec(self.func)
function_args = argsspec[0]
if 'HOST' in function_args:
if 'TOKEN' in function_args:
return self.func(self.HOST , self.TOKEN , *args, **kwds)
else:
return self.func(self.HOST , *args, **kwds)
elif 'TOKEN' in function_args:
return self.func(self.TOKEN, *args, **kwds)
0
Nếu bạn đã có chức năng
import inspect
class withHostAndToken(object):
__name__ = "withHostAndToken"
__doc__ = "Get the Host and Token for the API call"
def __init__(self, func):
print "withHostAndToken: \t" + func.__name__
self.func = func
self.HOST = ''
self.TOKEN = ''
def __call__(self,*args, **kwds):
if self.HOST == '':
self.HOST = "HOST"
if self.TOKEN == '':
self.TOKEN = "TOKEN"
argsspec = inspect.getargspec(self.func)
function_args = argsspec[0]
if 'HOST' in function_args:
if 'TOKEN' in function_args:
return self.func(self.HOST , self.TOKEN , *args, **kwds)
else:
return self.func(self.HOST , *args, **kwds)
elif 'TOKEN' in function_args:
return self.func(self.TOKEN, *args, **kwds)
3 trong không gian tên của mình, bạn có thể dễ dàng sử dụng bộ trang trí với không gian tên mô -đun:Warning: When a function times out, an exception is raised but cancellation is not guaranteed. This decorator only notifies the user when enough time has passed since a function call. Handling of the situation and ensuring cancellation is up to the user.Warning: When a function times out, an exception is raised but cancellation is not guaranteed. This decorator only notifies the user when enough time has passed since a function call. Handling of the situation and ensuring cancellation is up to the user.
: CẢNH BÁO: CẢNH BÁO: Khi một hàm hết thời gian, một ngoại lệ được nêu ra nhưng hủy bỏ không được đảm bảo. Người trang trí này chỉ thông báo cho người dùng khi đủ thời gian trôi qua kể từ khi gọi chức năng. Xử lý tình huống và đảm bảo hủy bỏ tùy thuộc vào người dùng.
Chữ ký
Nói chung, người trang trí chấp nhận hai tham số:
import inspect
class withHostAndToken(object):
__name__ = "withHostAndToken"
__doc__ = "Get the Host and Token for the API call"
def __init__(self, func):
print "withHostAndToken: \t" + func.__name__
self.func = func
self.HOST = ''
self.TOKEN = ''
def __call__(self,*args, **kwds):
if self.HOST == '':
self.HOST = "HOST"
if self.TOKEN == '':
self.TOKEN = "TOKEN"
argsspec = inspect.getargspec(self.func)
function_args = argsspec[0]
if 'HOST' in function_args:
if 'TOKEN' in function_args:
return self.func(self.HOST , self.TOKEN , *args, **kwds)
else:
return self.func(self.HOST , *args, **kwds)
elif 'TOKEN' in function_args:
return self.func(self.TOKEN, *args, **kwds)
5: A
import inspect
class withHostAndToken(object):
__name__ = "withHostAndToken"
__doc__ = "Get the Host and Token for the API call"
def __init__(self, func):
print "withHostAndToken: \t" + func.__name__
self.func = func
self.HOST = ''
self.TOKEN = ''
def __call__(self,*args, **kwds):
if self.HOST == '':
self.HOST = "HOST"
if self.TOKEN == '':
self.TOKEN = "TOKEN"
argsspec = inspect.getargspec(self.func)
function_args = argsspec[0]
if 'HOST' in function_args:
if 'TOKEN' in function_args:
return self.func(self.HOST , self.TOKEN , *args, **kwds)
else:
return self.func(self.HOST , *args, **kwds)
elif 'TOKEN' in function_args:
return self.func(self.TOKEN, *args, **kwds)
6 Chỉ định thời gian chờ tính bằng giây. Nếu
import inspect
class withHostAndToken(object):
__name__ = "withHostAndToken"
__doc__ = "Get the Host and Token for the API call"
def __init__(self, func):
print "withHostAndToken: \t" + func.__name__
self.func = func
self.HOST = ''
self.TOKEN = ''
def __call__(self,*args, **kwds):
if self.HOST == '':
self.HOST = "HOST"
if self.TOKEN == '':
self.TOKEN = "TOKEN"
argsspec = inspect.getargspec(self.func)
function_args = argsspec[0]
if 'HOST' in function_args:
if 'TOKEN' in function_args:
return self.func(self.HOST , self.TOKEN , *args, **kwds)
else:
return self.func(self.HOST , *args, **kwds)
elif 'TOKEN' in function_args:
return self.func(self.TOKEN, *args, **kwds)
7, chức năng sẽ không hết thời gian. Mặc định là
import inspect
class withHostAndToken(object):
__name__ = "withHostAndToken"
__doc__ = "Get the Host and Token for the API call"
def __init__(self, func):
print "withHostAndToken: \t" + func.__name__
self.func = func
self.HOST = ''
self.TOKEN = ''
def __call__(self,*args, **kwds):
if self.HOST == '':
self.HOST = "HOST"
if self.TOKEN == '':
self.TOKEN = "TOKEN"
argsspec = inspect.getargspec(self.func)
function_args = argsspec[0]
if 'HOST' in function_args:
if 'TOKEN' in function_args:
return self.func(self.HOST , self.TOKEN , *args, **kwds)
else:
return self.func(self.HOST , *args, **kwds)
elif 'TOKEN' in function_args:
return self.func(self.TOKEN, *args, **kwds)
7
import inspect
class withHostAndToken(object):
__name__ = "withHostAndToken"
__doc__ = "Get the Host and Token for the API call"
def __init__(self, func):
print "withHostAndToken: \t" + func.__name__
self.func = func
self.HOST = ''
self.TOKEN = ''
def __call__(self,*args, **kwds):
if self.HOST == '':
self.HOST = "HOST"
if self.TOKEN == '':
self.TOKEN = "TOKEN"
argsspec = inspect.getargspec(self.func)
function_args = argsspec[0]
if 'HOST' in function_args:
if 'TOKEN' in function_args:
return self.func(self.HOST , self.TOKEN , *args, **kwds)
else:
return self.func(self.HOST , *args, **kwds)
elif 'TOKEN' in function_args:
return self.func(self.TOKEN, *args, **kwds)
Vỏ bọc sử dụng asyncio và chủ đề để theo dõi thời gian chờ. Điều này thêm một chi phí không tầm thường trên các chức năng được bọc, đặc biệt là trên các chức năng đồng bộ vì chúng cần một luồng để được tạo.
Do đó, các trường hợp sử dụng phổ biến sẽ bao gồm các bộ thử nghiệm trong đó kiểm soát những gì xảy ra khi chúng hết thời gian là quan trọng, vì với các gói như
2, bạn không thể có các thử nghiệm thất bại với một ngoại lệ được chỉ định.
Mặt khác, các trường hợp sử dụng trong mã sản xuất sẽ được giới hạn trong các chức năng dài với tỷ lệ cuộc gọi thấp vì chi phí là tuyến tính với số lượng cuộc gọi nhưng không phụ thuộc vào kích thước của hàm hoặc thời gian thực hiện của nó.
Một cuốn sổ tay phác thảo tác động của chi phí đối với các hàm đồng bộ hóa và async có thể được tìm thấy ở đây.
Các lô từ máy tính xách tay được trình bày dưới đây để đánh giá hiệu suất nhanh.