Каналы Pipe

Тип Pipe из пакета multiprocessing представляет канал между двумя процессами:

multiprocessing.Pipe(duplex=True)

При создании экземпляра канала pipe возвращаются два объекта соединения (conn1, conn2) в виде объекта Connection. conn1 и conn2 - это два конца соединения (фактически два процесса, которые обмениваются данными). Оба эти объекта позволяют отправлять и получать данные. Формально первый канал - conn1 является родительским, а второй - conn2 - дочерним.

Конструктор класса принимает необязательное числовое значение - параметр duplex. Если оно имеет значение True (по умолчанию), то канал является двунаправленным. Если duplex имеет значение False, то канал - однонаправленный: conn1 можно использовать только для получения сообщений, а conn2 - только для их отправки.

Для взаимодействия между двумя точками канала обмена для класса Connection доступно ряд методов:

• send(obj): отправляет объект на другой конец соединения. Для его получения следует использовать метод recv()

  При этом объект должен поддерживать сериализацию (быть "picklable"). При попытке отправить очень большие сериализованные объекты (примерно от 32 МБ и более, в зависимости от ОС) может возникнуть исключение ValueError.

• recv(): возвращает объект, отправленный с другого конца соединения с помощью метода send(). Блокирует выполнение до тех пор, пока не появятся данные для получения. Вызывает исключение EOFError, если данных для получения больше нет, а соединение на другом конце было закрыто.

• close(): закрывает соединение. Этот метод вызывается автоматически при сборке мусора для объекта соединения.

• poll([timeout]): возвращает признак того, доступны ли данные для чтения.

• send_bytes(buf[, offset[, size]]): отправляет байтовые данные из объекта, который поддерживает протокол буфера (bytes-like object). Если указан параметр offset, данные считываются из buf начиная с этой позиции. Если указан параметр size, из buf считывается именно такое количество байтов. При использовании очень больших буферов (примерно от 32 МБ и более, в зависимости от ОС) может возникнуть исключение ValueError.

• recv_bytes([maxlength]): возвращает сообщение из байтовых данных, отправленное с другого конца соединения. Блокирует выполнение до тех пор, пока не появятся данные для получения. Вызывает исключение EOFError, если данных для получения больше нет, а соединение на другом конце было закрыто.

  Если указан параметр maxlength и длина сообщения превышает это значение, вызывается исключение OSError, и чтение из соединения становится невозможным.

• recv_bytes_into(buf[, offset]): считывает в буфер buf сообщение из байтовых данных, отправленное с другого конца соединения, и возвращает количество байтов в сообщении. Блокирует выполнение до тех пор, пока не появятся данные для получения. Вызывает исключение EOFError, если данных для получения больше нет, а соединение на другом конце было закрыто.

Ключевыми здесь являются метод send(), который сериализует объект с помощью модуля pickle и отправляет его в канал, и метод recv(), который получае данные из канала и восстанавливает объект. Сначала рассмотрим упрощенный пример без процессов:

from multiprocessing import Pipe

a, b = Pipe()

a.send([1, "hello", True])      # отправляем в канал

data = b.recv()     # получаем данные из канала

print(data)     # [1, 'hello', True]

Здесь канал создается между двумя объектами - a и b. Объект a посылает в канал список из трех элементов, а объект b получает данные из канала. Аналогично можно отправлять и большее, количество данных:

from multiprocessing import Pipe

a, b = Pipe()

a.send([1, "hello", True])
a.send("World")

data = b.recv()
print(data)     # [1, "hello", True]
data = b.recv()
print(data)     # "World"

Теперь посмотрим, как сделать канал между двумя процессами:

from multiprocessing import Process, Pipe

# функция, которая запускается во втором процессе
def send_to_parent(conn):
    conn.send("Hello, parent")
    conn.close()    # закрываем канал

if __name__ == "__main__":
    parent_conn, child_conn = Pipe() 
    # второй процесс вызывает функцию send_to_parent
    child_process = Process(target=send_to_parent, args=(child_conn,))
    child_process.start()

    message = parent_conn.recv()    # получаем данные от второго процесса
    print("Child sent:", message)   # Child sent: Hello, parent
    child_process.join()

Здесь второй канал - child_process вызывает функцию send_to_parent и получает объект соединения child_conn. Через этот объект в функции send_to_parent отправляем данные в канал:

conn.send("Hello, parent")

На другом конце соединения мы можем получить эти данные из канала:

message = parent_conn.recv()

Обратите внимание, что данные в канале могут быть повреждены, если два процесса (или потока) попытаются одновременно читать из одного и того же конца канала или писать в него.

Дуплексные каналы

В примере выше данные передавались от одного процесса другому. Однако в реальности процессы могут обмениваться данными через каналы. Например:

from multiprocessing import Process, Pipe

# функция, которая запускается во втором процессе
def conn1_func(conn):
    n = 5
    conn.send(n)            # посылаем число
    square_n = conn.recv()  # получаем из квадрат числа
    conn.close()            # закрываем канал
    print(f"Квадрат числа {n} равен {square_n}")

def conn2_func(conn):
    n = conn.recv()     # получаем из канала число
    conn.send(n*n)      # посылаем обратно квадрат числа

if __name__ == "__main__":
    conn1, conn2 = Pipe() 

    # определяем и запускаем два процесса для каждой из сторон канала
    conn1_process = Process(target=conn1_func, args=(conn1,))
    conn2_process = Process(target=conn2_func, args=(conn2,))

    conn1_process.start()
    conn2_process.start()

    conn1_process.join()
    conn2_process.join()

Здесь создаем два процесса для каждого из концов канала - conn1_process и conn2_process. ПЕрвый процесс отправляет через канал второму число. Второй процесс вычисляет его квадрат и отправляет его обратно первому процессу. Первый процесс, получив результат вычислений, закрывает канал и выводит результат. Консольный вывод программы:

Квадрат числа 5 равен 25