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

Question

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

helpers.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)

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

How to try a function with a timeout exception in python

python

multiprocessing

flask

python-multiprocessing