Słownik python kolejek między procesami

Question

To nie jest bardzo ważne, tylko głupi eksperyment. Chciałbym stworzyć własną wiadomość przekazującą. Chciałbym mieć słownik kolejek, gdzie każdy klucz jest PID procesu. Ponieważ chciałbym, aby procesy (utworzone przez Process()) wymieniały komunikaty wstawiając je w kolejce procesu, do którego chcą wysłać wiadomość (znając jej pid). Jest to kod głupie:

from multiprocessing import Process, Manager, Queue 
from os import getpid 
from time import sleep 

def begin(dic, manager, parentQ): 
    parentQ.put(getpid()) 
    dic[getpid()] = manager.Queue() 
    dic[getpid()].put("Something...") 

if __name__== '__main__': 
    manager = Manager() 
    dic = manager.dict() 
    parentQ = Queue() 

    p = Process(target = begin, args=(dic, manager, parentQ)) 
    p.start() 
    son = parentQ.get() 
    print son 
    sleep(2) 
    print dic[son].get() 

dic[getpid()] = manager.Queue(), to działa prawidłowo. Ale kiedy wykonać dic[son].put()/get() otrzymuję komunikat:

Process Process-2: 
Traceback (most recent call last): 
    File "/usr/lib/python2.7/multiprocessing/process.py", line 258, in _bootstrap 
    self.run() 
    File "/usr/lib/python2.7/multiprocessing/process.py", line 114, in run 
    self._target(*self._args, **self._kwargs) 
    File "mps.py", line 8, in begin 
    dic[getpid()].put("Something...") 
    File "<string>", line 2, in __getitem__ 
    File "/usr/lib/python2.7/multiprocessing/managers.py", line 773, in _callmethod 
    raise convert_to_error(kind, result) 
RemoteError: 
--------------------------------------------------------------------------- 
Unserializable message: ('#RETURN', <Queue.Queue instance at 0x8a92d0c>) 
--------------------------------------------------------------------------- 

Czy wiesz co to jest właściwy sposób to zrobić?

Answer 1

Wierzę, że Twój kod się nie powiódł, ponieważ kolejek nie można szeregować, tak jak mówi traceback. Obiekt multiprocessing.Manager() może utworzyć dla ciebie wspólny dokument bez problemu, tak jak tutaj, ale wartości zapisane w dyktafonie muszą być możliwe do serializacji (lub picklable in Pythonese). Jeśli jesteś porządku z podprocesów nie mając dostępu do kolejek nawzajem, to powinny pracować dla Ciebie:

from multiprocessing import Process, Manager, Queue 
from os import getpid 

number_of_subprocesses_i_want = 5 

def begin(myQ): 
    myQ.put("Something sentimental from your friend, PID {0}".format(getpid())) 
    return 

if __name__== '__main__': 
    queue_dic = {} 
    queue_manager = Manager() 

    process_list = [] 

    for i in xrange(number_of_subprocesses_i_want): 
     child_queue = queue_manager.Queue() 

     p = Process(target = begin, args=(child_queue,)) 
     p.start() 
     queue_dic[p.pid] = child_queue 
     process_list.append(p) 

    for p in process_list: 
     print(queue_dic[p.pid].get()) 
     p.join() 

Pozostawia cię ze słownikiem którego klucze są procesy potomne, a wartości są ich odpowiednie kolejki, które można wykorzystać z głównego procesu.

Nie sądzę, aby pierwotny cel był osiągalny w przypadku kolejek, ponieważ kolejki, które mają być używane w podprocesie, muszą być przekazywane do procesów, gdy są tworzone, aby można było uruchamiać więcej procesów, nie można podać istniejący dostęp do procesu do nowej kolejki.

Możliwym sposobem na komunikację między procesami byłoby mieć każdy dzielić pojedynczą kolejkę przekazać wiadomości z powrotem do głównego procesu w pakiecie z pewnego rodzaju nagłówka, tak jak w krotki:

(destination_pid, sender_pid, message) 

..i głownie przeczytaj destination_pid i wyślij bezpośrednio (sender_pid, message) do tej kolejki podprocesu. Oczywiście oznacza to, że potrzebujesz metody powiadamiania istniejących procesów, kiedy nowy proces jest dostępny do komunikowania się.