如何在python中尝试具有超时异常的函数

Question

我想在helpers.py中调用一个函数，如果它运行超过10秒，就会超时。我已经尝试过多处理来做到这一点，但我不知道如何从多处理中提取我的结果字典，我想知道你们是否都有一个好的方法来做这件事？

帮助者.py

def speed_test():
print('running speedtest')
servers = []
# If you want to test against a specific server
# servers = [1234]
threads = None
# If you want to use a single threaded test
# threads = 1
s = speedtest.Speedtest()
s.get_servers(servers)
s.get_best_server()
s.download(threads=threads)
s.upload(threads=threads)
results_dict = s.results.dict()
print('speedtest complete')
return results_dict

烧瓶路线

from app import helpers
@bp.route('/speed_test')
@login_required
def speed_test1():
if not current_user.admin:
flash('Page unacessible.')
return redirect(url_for('main.index'))
#This gives my results in a dictionary i can use just fine
#results = helpers.speed_test()
#This executes my function just fine, but I have no idea how to pull the dicitonary result from my function
results = multiprocessing.Process(target=helpers.speed_test)
results.start()
results.join(30)
if results.is_alive():
print('timeout kill')
results.kill()
results='timeout'
results.join()
#results.('i want my dictionary from the function return')
return render_template('test/speedtest.html', title='Speed Test', results=results)

Answer 1

虽然处理池通常用于并行运行多个作业，但它可能是最简单的"一举两得"；，它有一个机制来计时进程并从进程返回字典：

def speed_test(): # unchanged
# code omitted
results_dict = s.results.dict()
print('speedtest complete')
return results_dict

@bp.route('/speed_test')
@login_required
def speed_test1():
#code omitted
with multiprocessing.Pool(1) as pool: # create a pool size of 1
result = pool.apply_async(speed_test)
# the following will generate a multiprocessing.TimeoutError if speed_test takes longer than 10 seconds:
results_dict = result.get(10)