Как отслеживать асинхронные результаты, возвращаемые из многопроцессорного пула

Ответ 1

Не видя фактического кода, я могу ответить только в общих чертах. Но есть два общих решения.

Во-первых, вместо использования callback и игнорирования AsyncResult s сохраните их в какой-то коллекции. Тогда вы можете просто использовать эту коллекцию. Например, если вы хотите иметь возможность искать результаты для функции, используя эту функцию в качестве ключа, просто создайте dict с помощью функций:

def in_parallel(funcs):
    results = {}
    pool = mp.Pool()
    for func in funcs:
        results[func] = pool.apply_async(func)
    pool.close()
    pool.join()
    return {func: result.get() for func, result in results.items()}

В качестве альтернативы вы можете изменить функцию обратного вызова, чтобы сохранить результаты в своей коллекции по ключу. Например:

def in_parallel(funcs):
    results = {}
    pool = mp.Pool()
    for func in funcs:
        def callback(result, func=func):
            results[func] = result
        pool.apply_async(func, callback=callback)
    pool.close()
    pool.join()
    return results

Я использую эту функцию как ключ. Но вместо этого вы хотите использовать индекс, это так же просто. Любое значение, которое у вас есть, можно использовать в качестве ключа.

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

pool = mp.Pool()
results = list(pool.imap_unordered(foo_pool, range(10)))
pool.close()
pool.join()

И затем, если вы хотите, чтобы результаты были в исходном порядке, а не в произвольном порядке, вы можете просто переключиться на imap или map. Итак:

pool = mp.Pool()
results = pool.map(foo_pool, range(10))
pool.close()
pool.join()

Если вам понадобится нечто подобное, но слишком сложное, чтобы вписаться в парадигму map, concurrent.futures, вероятно, сделает вашу жизнь проще, чем multiprocessing. Если вы находитесь на Python 2.x, вам нужно будет установить backport. Но тогда вы можете делать то, что гораздо сложнее сделать с AsyncResult или callback (или map), например, составление целого ряда будущих фьючерсов в одно большое будущее. См. Примеры в связанных документах.

Последнее примечание:

Важные моменты, которые нужно подчеркнуть, это то, что я не могу изменить существующие функции...

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

def number_and_square(x):
    return x, square(x)

И теперь я могу просто apply_async(number_and_square) вместо просто square и получить результаты, которые я хочу.

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