Как передать переменные инициализации для конкретного работника в рабочий пул, если аргументы содержат очередь?

Question

Я заинтересован в создании пула работников, the_pool, используя multiprocessing.Pool, который использует связь Queue. Однако у каждого работника есть аргумент role, который является уникальным для этого работника и должен быть предоставлен при инициализации работника. Это ограничение навязывается API, с которым я взаимодействую, и поэтому его нельзя обойти. Если бы я не нуждался в очереди, я мог бы просто перебрать список role значений и вызывать apply_async, например, так:

[the_pool.apply_async(worker_main, role) for role in roles] 

К сожалению, Queue объект может быть передан только в бассейнах во время бассейна конкретизации, как:

the_pool = multiprocessing.Pool(3, worker_main, (the_queue,)) 

Попытка передать Queue через аргументы apply_async вызывает ошибку во время выполнения. В следующем примере, адаптированном из this question, мы пытаемся создать пул из трех рабочих. Но пример не работает, потому что нет способа получить элемент ролей от roles до initargs для пула.

import os 
import time 
import multiprocessing 

# A dummy function representing some fixed functionality. 
def do_something(x): 
    print('I got a thing:', x) 

# A main function, run by our workers. (Remove role arg for working example) 
def worker_main(queue, role): 

    print('The worker at', os.getpid(), 'has role', role, ' and is initialized.') 

    # Use role in some way. (Comment out for working example) 
    do_something(role) 

    while True: 
     # Block until something is in the queue. 
     item = queue.get(True) 
     print(item) 
     time.sleep(0.5) 

if __name__ == '__main__': 

    # Define some roles for our workers. 
    roles = [1, 2, 3] 

    # Instantiate a Queue for communication. 
    the_queue = multiprocessing.Queue() 

    # Build a Pool of workers, each running worker_main. 
    # PROBLEM: Next line breaks - how do I pass one element of roles to each worker? 
    the_pool = multiprocessing.Pool(3, worker_main, (the_queue,)) 

    # Iterate, sending data via the Queue. 
    [the_queue.put('Insert useful message here') for _ in range(5)] 

    worker_pool.close() 
    worker_pool.join() 
    time.sleep(10) 

Один тривиальных Обходной включать в initargs, который служит только для связи роли каждого работника второго Queue и блокировать выполнение рабочих до тех пор, пока не получит свою роль с помощью этой очереди. Это, однако, вводит дополнительную очередь, которая не должна быть необходимой. Соответствующая документация - here. Руководство и советы очень приветствуются.

Answer 1

Почему бы не использовать две рабочие функции, одну для инициализации? Как:

def worker_init(q): 
    global queue 
    queue = q 

def worker_main(role): 
    # use the global `queue` freely here 

Инициализация является такой же, как и то, что вы показали, за исключением вызова worker_init:

the_pool = multiprocessing.Pool(3, worker_init, (the_queue,)) 

инициализация выполняется только один раз в рабочий процесс, и каждый процесс повторяется, пока Pool завершается. Для того, чтобы получить работу, делать то, что вы хотите сделать:

[the_pool.apply_async(worker_main, role) for role in roles] 

Там нет необходимости передавать the_queue тоже - каждый рабочий процесс уже узнал об этом во время инициализации.

Answer 2

Вы можете просто создать очередь с ролями:

import os 
import time 
import multiprocessing 

# A dummy function representing some fixed functionality. 
def do_something(x): 
    print('I got a thing:', x) 

# A main function, run by our workers. (Remove role arg for working example) 
def worker_main(queue, roles): 
    role = roles.get() 
    print('The worker at', os.getpid(), 'has role', role, ' and is initialized.') 

    # Use role in some way. (Comment out for working example) 
    do_something(role) 

    while True: 
     # Block until something is in the queue. 
     item = queue.get(True) 
     print(item) 
     time.sleep(0.5) 

if __name__ == '__main__': 

    # Define some roles for our workers. 
    roles = [1, 2, 3] 

    # Instantiate a Queue for communication. 
    the_queue = multiprocessing.Queue() 
    roles_queue = multiprocessing.Queue() 
    for role in roles: 
     roles_queue.put(role) 


    # Build a Pool of workers, each running worker_main. 
    # PROBLEM: Next line breaks - how do I pass one element of roles to each worker? 
    the_pool = multiprocessing.Pool(3, worker_main, (the_queue, roles_queue)) 

    # Iterate, sending data via the Queue. 
    [the_queue.put('Insert useful message here') for _ in range(5)] 

    worker_pool.close() 
    worker_pool.join() 
    time.sleep(10)