Показывать индикатор выполнения для моего многопоточного процесса

У меня есть простое веб-приложение Flask, которое делает много HTTP-запросов внешней службе, когда пользователь нажимает кнопку. На стороне клиента у меня есть приложение angularjs.

Серверная часть кода выглядит так (используя multiprocessing.dummy):

worker = MyWorkerClass()
pool = Pool(processes=10)
result_objs = [pool.apply_async(worker.do_work, (q,))
                           for q in queries]
pool.close() # Close pool
pool.join()  # Wait for all task to finish
errors = not all(obj.successful() for obj in result_objs)
# extract result only from successful task
items = [obj.get() for obj in result_objs if obj.successful()]

Как вы можете видеть, я использую apply_async, потому что я хочу позже проверить каждую задачу и извлечь из них результат только в том случае, если задача не вызвала никаких исключений.

Я понял, что для того, чтобы показать индикатор выполнения на стороне клиента, мне нужно опубликовать где-то количество завершенных задач, чтобы я сделал простое представление:

@app.route('/api/v1.0/progress', methods=['GET'])
def view_progress():
    return jsonify(dict(progress=session['progress']))

Это покажет содержимое переменной сеанса. Теперь, во время процесса, мне нужно обновить эту переменную с количеством завершенных задач (общее количество задач для завершения является фиксированным и известным).

Любые идеи о том, как это сделать? Я работаю в правильном направлении?

Я видел похожие вопросы на SO как этот, но я не могу адаптировать ответ к моему делу.

Спасибо.

Ответ 1

Для межпроцессного взаимодействия вы можете использовать multiprocessiong.Queue, и ваши работники могут put_nowait привязывать к нему информацию о ходе выполнения, выполняя свою работу, Ваш основной процесс может обновлять все ваши данные view_progress, пока все результаты не будут готовы.

Немного похоже на пример использования очереди с несколькими настройками:

В писателях (рабочих) я использовал бы put_nowait вместо put, потому что работа важнее, чем ждать, чтобы сообщить, что вы работаете (но, возможно, вы судите об этом иначе и решаете, что информирование пользователя является частью задача и никогда не должна быть пропущена).

Пример только строк puts в очереди, я бы использовал collections.namedtuples для более структурированных сообщений. В задачах с несколькими шагами это позволяет вам повысить разрешение вашего отчета о проделанной работе и сообщить пользователю больше.

Ответ 2

В целом подход, который вы принимаете, в порядке, я делаю это аналогичным образом.

Чтобы рассчитать прогресс, вы можете использовать вспомогательную функцию, которая учитывает завершенные задачи:

def get_progress(result_objs):
    done = 0
    errors = 0
    for r in result_objs:
        if r.ready():
            done += 1
            if not r.successful():
                errors += 1
    return (done, errors)

Обратите внимание, что в качестве бонуса эта функция возвращает количество выполненных задач, заканчивающихся ошибками.

Большая проблема заключается в маршруте /api/v1.0/progress для поиска массива объектов AsyncResult.

К сожалению, объекты AsyncResult не могут быть сериализованы для сеанса, так что опция отсутствует. Если ваше приложение поддерживает один набор задач async за один раз, вы можете просто сохранить этот массив как глобальную переменную. Если вам нужно поддерживать несколько клиентов, каждый из которых имеет другой набор асинхронных задач, тогда вам нужно будет найти стратегию для хранения данных сеанса клиента на сервере.

Я реализовал решение для одного клиента в качестве быстрого теста. Мои функции просмотра выглядят следующим образом:

results = None

@app.route('/')
def index():
    global results
    results = [pool.apply_async(do_work) for n in range(20)]
    return render_template('index.html')

@app.route('/api/v1.0/progress')
def progress():
    global results
    total = len(results)
    done, errored = get_progress(results)
    return jsonify({'total': total, 'done': done, 'errored': errored})

Надеюсь, это поможет!

Ответ 3

Я думаю, вы сможете обновить количество завершенных задач, используя multiprocessing.Value и multiprocessing.Lock.

В главном коде используйте:

processes=multiprocessing.Value('i', 10)
lock=multiprocessing.Lock()

И затем, когда вы вызываете employee.dowork, передайте ему объект блокировки и значение:

worker.dowork(lock, processes)

В коде employee.dowork уменьшите "процессы" на один, когда код будет завершен:

lock.acquire()
processes.value-=1
lock.release()

Теперь "process.value" должно быть доступно из вашего основного кода и быть равно количеству оставшихся процессов. Удостоверьтесь, что вы приобрели блокировку до завершения процессов. Значение и затем отпустите блокировку.