Как передать ссылку на функцию, управляемую pool.map_async ()? - PullRequest
Новая
       15

Как передать ссылку на функцию, управляемую pool.map_async ()?

37 голосов
/ 10 июля 2010

Я хочу, чтобы длительный процесс возвращал свой прогресс через очередь (или что-то подобное), который я передам в диалоговое окно индикатора выполнения.Мне также нужен результат, когда процесс будет завершен.Тестовый пример здесь не удался с RuntimeError: Queue objects should only be shared between processes through inheritance. 

import multiprocessing, time

def task(args):
    count = args[0]
    queue = args[1]
    for i in xrange(count):
        queue.put("%d mississippi" % i)
    return "Done"

def main():
    q = multiprocessing.Queue()
    pool = multiprocessing.Pool()
    result = pool.map_async(task, [(x, q) for x in range(10)])
    time.sleep(1)
    while not q.empty():
        print q.get()
    print result.get()

if __name__ == "__main__":
    main()

. Мне удалось заставить это работать, используя отдельные объекты Process (где я am позволил передать ссылку Queue) но у меня нет пула для управления многими процессами, которые я хочу запустить.Любой совет для лучшего шаблона для этого?

Ответы [ 2 ]

44 голосов
/ 13 июля 2010

Кажется, работает следующий код: 

import multiprocessing, time

def task(args):
    count = args[0]
    queue = args[1]
    for i in xrange(count):
        queue.put("%d mississippi" % i)
    return "Done"


def main():
    manager = multiprocessing.Manager()
    q = manager.Queue()
    pool = multiprocessing.Pool()
    result = pool.map_async(task, [(x, q) for x in range(10)])
    time.sleep(1)
    while not q.empty():
        print q.get()
    print result.get()

if __name__ == "__main__":
    main()

Обратите внимание, что Очередь получена от manager.Queue (), а не от multiprocessing.Queue (). Спасибо, Алекс, за то, что указал мне в этом направлении.

8 голосов
/ 10 июля 2010

Создание q global работает ...: 

import multiprocessing, time

q = multiprocessing.Queue()

def task(count):
    for i in xrange(count):
        q.put("%d mississippi" % i)
    return "Done"

def main():
    pool = multiprocessing.Pool()
    result = pool.map_async(task, range(10))
    time.sleep(1)
    while not q.empty():
        print q.get()
    print result.get()

if __name__ == "__main__":
    main()

Если вам нужно несколько очередей, например, чтобы не перепутать ход различных процессов пула, глобальный списокочередей должно работать (конечно, каждый процесс должен будет знать, какой индекс в списке использовать, но это нормально для передачи в качестве аргумента; -).

...