Python跨平台函数执行时间限制

2018年3月21日 0条评论 770次阅读 1人点赞

项目中用到的一个小功能，但是在网上没有找到相关的任何比较好的想法/实现。

自己想到一种Python跨平台实现函数执行时间限制的思路。

网上的方法比较多的是利用信号量 signal 设置一个定时器，超时之后执行回调函数，引发 TimeoutError，退出执行。

这种方法易于实现，并且占用资源少，只需要设置一个 signal.alarm 的闹钟即可。

但是缺点也显而易见：在 Windows平台并不能够使用。

分析 signal 的这种方式，我有了一种灵感：让触发退出的回调函数与任务函数 “抢占执行” —— 谁先执行完，就返回谁的值：在计时线程返回之后，不管任务子线程的执行结果，直接返回错误/返回指定值。

threading 库中有一个 Timer 函数，我们用它来计时；在整个过程中，需要同步线程内部的值到父线程/主进程中 —— 最好的选择个人认为还是队列。这里我们不单单将闭包中的队列变量用来同步 return 的值，还可以利用队列一个很好的特性 —— 锁，在任务线程和计时线程均开始运行之后，可以用队列的 get 锁把父线程/主进程阻塞掉，这样既能保证很低的系统占用，又可以实现回收线程返回值的目标（但我这里单独给了一个锁变量控制），一举两得。

下面贴代码：

# author : guiqiqi187@gmail.com
# @basic.tlimit(limit_time_int) : result
# 通过该装饰器可以限制一个函数的运行时间
#
# 思路：
# 主线程中初始化一个锁变量并立刻将其acquire
# 开启一个定时器线程, 在指定时长之后release锁变量
# 开启一个任务线程, 当任务函数执行完之后也同样release锁变量
# 		这个任务线程将执行的函数返回值放入结果队列
# 此时程序中有两个线程在同时执行(其中的定时器线程暂时sleep)
# 将主线程挂起(通过尝试acquire锁)
# 当acquire到锁之后说明两个线程中有一个结束了
#		这时检查结果队列是否为空, 如果为空说明时间到了
#		之后在主线程抛出TimoutError, 交由上级处理
def tlimit(limit, *default):
	def wrapper(function):
		def proWarp(*args, **kwargs):
			result = queue.Queue()
			mutex = threading.Lock()
			mutex.acquire()
			def t_error() : mutex.release()
			def process(*args, **kwargs):
				result.put(function(*args, **kwargs))
				if mutex.locked() : mutex.release()
			timer = threading.Timer(limit, t_error)
			thread = threading.Thread(target = process,
				args = args, kwargs = kwargs)
			thread.setDaemon(True)
			timer.start()
			thread.start()
			if mutex.acquire():
				timer.cancel()
				if not result.empty():
					return result.get()
				else:
					if default:
						return default[0]
					raise TimeoutError
		return proWarp
	return wrapper

# author : guiqiqi187@gmail.com

# @basic.tlimit(limit_time_int) : result

# 通过该装饰器可以限制一个函数的运行时间

# 思路：

# 主线程中初始化一个锁变量并立刻将其acquire

# 开启一个定时器线程, 在指定时长之后release锁变量

# 开启一个任务线程, 当任务函数执行完之后也同样release锁变量

# 这个任务线程将执行的函数返回值放入结果队列

# 此时程序中有两个线程在同时执行(其中的定时器线程暂时sleep)

# 将主线程挂起(通过尝试acquire锁)

# 当acquire到锁之后说明两个线程中有一个结束了

# 这时检查结果队列是否为空, 如果为空说明时间到了

# 之后在主线程抛出TimoutError, 交由上级处理

def tlimit(limit, *default):

def wrapper(function):

def proWarp(*args, **kwargs):

result = queue.Queue()

mutex = threading.Lock()

mutex.acquire()

def t_error() : mutex.release()

def process(*args, **kwargs):

result.put(function(*args, **kwargs))

if mutex.locked() : mutex.release()

timer = threading.Timer(limit, t_error)

thread = threading.Thread(target = process,

args = args, kwargs = kwargs)

thread.setDaemon(True)

timer.start()

thread.start()

if mutex.acquire():

timer.cancel()

if not result.empty():

return result.get()

else:

if default:

return default[0]

raise TimeoutError

return proWarp

return wrapper

如果不想让超时之后引发 TimeoutError，请给装饰器传入 default参数，作为超时之后的默认返回值。

另外需要注意的一点时：任务函数并不会随着装饰器函数的退出而结束，因为在这里父线程是主进程，虽然设置了 Daemon ，但是装饰起函数退出后，主进程仍在执行，便不会退出任务线程（如果一定要退出任务线程，请考虑将整个过程单独出来成为一个线程，但是这样回更多的占用资源，并且需要再次同步父线程和主进程之间的返回值，比较麻烦。）

这样就可以实现跨平台的函数执行时间限制。