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

Question

[Space Around]

Как реализовать двунаправленный обмен сообщениями между родительским и дочерним процессом в библиотеке multiprocessing на python?
В документации написано, что duplex должен быть равен True для двунаправленного обмена, а кода как это должно выглядеть нету. Предполагаю, что необходимо в функцию foo добавить обработчик сообщений как в конце кода, вот только у меня как-то криво работает, код не завершает работу.

Код однонаправленного обмена:

import time, random
from multiprocessing import Process, Pipe, current_process
from multiprocessing.connection import wait

def foo(w):
    for i in range(10):
        w.send((i, current_process().name))
    w.close()

if __name__ == '__main__':
    readers = []

    for i in range(4):
        r, w = Pipe(duplex=False)
        readers.append(r)
        p = Process(target=foo, args=(w,))
        p.start()
        w.close()

    while readers:
        for r in wait(readers):
            try:
                msg = r.recv()
            except EOFError:
                readers.remove(r)
            else:
                print(msg)

Код моей реализации двунаправленного обмена

import time, random
from multiprocessing import Process, Pipe, current_process
from multiprocessing.connection import wait

def foo(w, i):
    readers2 = []
    readers2.append(w)
    w.send((i, "Child"))
    while readers2:
        for r2 in wait(readers2):
            try:
                msg = r2.recv()
            except EOFError:
                readers2.remove(r2)
            else:
                print("Child: " + msg)
    w.close()


if __name__ == '__main__':
    readers = []

    for i in range(1):
        r, w = Pipe(duplex=True)
        readers.append(r)
        p = Process(target=foo, args=(w,i))
        p.start()

        w.close()

    while readers:
        for r in wait(readers):
            try:
                msg = r.recv()
                r.send("test_1")
                r.send("test_2")
            except EOFError:
                readers.remove(r)
            else:
                print(msg)
    r.close()

Answer 1

[javedimka]

Просто сделай две очереди. Зачем всё усложнять.

Comments

[Space Around]

javedimka, можно ссылку на пример?

[Space Around]

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

from multiprocessing import Process, Pipe, current_process
from multiprocessing.connection import wait

def foo(w, i):
    print("w -> 1234")
    w.send("1234")

    readers = []

    readers.append(w)

    while readers:
        for r in wait(readers):
            try:
                msg = r.recv()
            except EOFError:
                print("w: error")
                readers.remove(r)
            else:

                if(msg == "1234"):
                    w.send("4321")
                    print("1234 -> w -> 4321")

                if(msg == "4321"):
                    print("4321 -> w")
                    print("w: close")


if __name__ == '__main__':

    readers = []
    r, w = Pipe(duplex=True)
    readers.append(r)
    p = Process(target=foo, args=(w, 1))
    p.start()
    w.close()

    while readers:
        for r in wait(readers):
            try:
                msg = r.recv()
                if(msg == "1234"):
                    r.send("1234")
                    print("1234 -> r -> 1234")

                if(msg == "4321"):
                    r.send("r: 4321")
                    print("4321 -> r -> 4321")

            except EOFError:
                print("r: error")
                readers.remove(r)

[javedimka]

Space Around,
*Возможны дедлоки, как обходить указано в документации

from multiprocessing import Process, Queue

def power_of_two(in_q, out_q):
 while True:
  val = in_q.get()
  out_q.put(val**2)

power_of_two_incoming_queue = Queue()
power_of_two_outgoing_queue = Queue()

calculator = Process(target=power_of_two, args=(power_of_two_outgoing_queue, power_of_two_incoming_queue))
calculator.start()
...

>>> power_of_two_outgoing_queue.put(2)
>>> power_of_two_incoming_queue.get()
4
>>> power_of_two_outgoing_queue.put(3)
>>> power_of_two_incoming_queue.get()
9