Process Communication

Introduction

Inter-process communication (IPC) allows processes to exchange data. Python supports pipes, shared memory, and managers for IPC.

Pipes

import multiprocessing

def worker(pipe):
    data = pipe.recv()
    pipe.send(data.upper())

if __name__ == "__main__":
    parent_conn, child_conn = multiprocessing.Pipe()
    
    p = multiprocessing.Process(target=worker, args=(child_conn,))
    p.start()
    
    parent_conn.send("hello")
    response = parent_conn.recv()
    print(response)
    
    p.join()

Shared Memory

import multiprocessing

def worker(shared_num, shared_array):
    shared_num.value += 1
    shared_array[0] = 99

if __name__ == "__main__":
    # Value - single value
    num = multiprocessing.Value("i", 0)
    
    # Array - typed array
    arr = multiprocessing.Array("i", [1, 2, 3, 4, 5])
    
    processes = [multiprocessing.Process(target=worker, args=(num, arr)) for _ in range(4)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    
    print(f"Result: {num.value}")  # 4
    print(f"Array: {list(arr)}")   # [99, 2, 3, 4, 5]

Manager for Complex Data

import multiprocessing

def worker(manager_dict, manager_list):
    manager_dict["count"] = manager_dict.get("count", 0) + 1
    manager_list.append(manager_dict["count"])

if __name__ == "__main__":
    with multiprocessing.Manager() as manager:
        d = manager.dict()
        l = manager.list()
        
        with multiprocessing.Pool(4) as pool:
            pool.starmap(worker, [(d, l)] * 10)
        
        print(dict(d))
        print(list(l))

Queue for Messages

import multiprocessing

def producer(queue):
    for i in range(10):
        queue.put({"type": "data", "value": i})
    queue.put(None)

def consumer(queue):
    while True:
        msg = queue.get()
        if msg is None:
            break
        print(f"Received: {msg}")

if __name__ == "__main__":
    queue = multiprocessing.Queue()
    p1 = multiprocessing.Process(target=producer, args=(queue,))
    p2 = multiprocessing.Process(target=consumer, args=(queue,))
    
    p1.start()
    p2.start()
    p1.join()
    p2.join()

Socket-Based IPC

import socket

def server():
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.bind(("localhost", 9999))
    sock.listen(1)
    conn, _ = sock.accept()
    data = conn.recv(1024)
    conn.send(b"ACK")
    conn.close()
    sock.close()

def client():
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect(("localhost", 9999))
    sock.send(b"Hello")
    print(sock.recv(1024))
    sock.close()

Practice Problems

1. Use Pipe for bidirectional communication
2. Share data with Value and Array
3. Create Manager for dictionaries
4. Implement Queue-based messaging
5. Build socket-based IPC