Interprocess Communication — Pipes, Message Queues, Shared Memory & Sockets

import os
import sys

def pipe_demo():
    r, w = os.pipe()  # Create pipe: r is read end, w is write end
    pid = os.fork()

    if pid == 0:  # Child process
        os.close(w)  # Close write end
        data = os.read(r, 1024)
        print(f'Child received: {data.decode()!r}')
        os.close(r)
        sys.exit(0)
    else:  # Parent process
        os.close(r)  # Close read end
        message = b'Hello from parent!'
        os.write(w, message)
        os.close(w)
        os.waitpid(pid, 0)
        print('Parent: message sent')

if __name__ == '__main__':
    pipe_demo()

import os
import time
import threading

def fifo_writer(path='/tmp/myfifo'):
    if not os.path.exists(path):
        os.mkfifo(path)
    with open(path, 'w') as f:
        for i in range(5):
            msg = f'Message {i}\n'
            f.write(msg)
            f.flush()
            print(f'Writer sent: {msg.strip()}')
            time.sleep(0.5)
    os.unlink(path)

def fifo_reader(path='/tmp/myfifo'):
    while os.path.exists(path):
        with open(path, 'r') as f:
            data = f.read()
            if data:
                print(f'Reader got: {data.strip()}')
        time.sleep(0.1)

writer = threading.Thread(target=fifo_writer)
reader = threading.Thread(target=fifo_reader)
reader.start()
time.sleep(0.2)
writer.start()
writer.join()
reader.join(timeout=1)

import sysv_ipc  # Requires: pip install sysv_ipc
import os

def message_queue_demo():
    try:
        # Create a message queue with key 1234
        mq = sysv_ipc.MessageQueue(1234, sysv_ipc.IPC_CREAT)

        # Writer
        mq.send(b'Hello via message queue!', type=1)
        mq.send(b'High priority message', type=2)
        print('Sent 2 messages')

        # Reader
        msg1 = mq.receive(type=1)
        print(f'Received type=1: {msg1[0].decode()!r}')

        msg2 = mq.receive(type=2)
        print(f'Received type=2: {msg2[0].decode()!r}')

        # Clean up
        mq.remove()
    except ImportError:
        print('sysv_ipc not installed — showing code only')

if __name__ == '__main__':
    message_queue_demo()

import multiprocessing
import time

def worker_process(name, shared_dict):
    print(f'{name} sees: {shared_dict}')
    shared_dict[f'message_from_{name}'] = f'Hello from {name}!'
    print(f'{name} wrote to shared memory')

if __name__ == '__main__':
    manager = multiprocessing.Manager()
    shared_data = manager.dict()

    p1 = multiprocessing.Process(
        target=worker_process, args=('Process A', shared_data))
    p2 = multiprocessing.Process(
        target=worker_process, args=('Process B', shared_data))

    p1.start()
    p2.start()
    p1.join()
    p2.join()

    print(f'\nFinal shared data: {dict(shared_data)}')

import mmap
import os
import time

def mmap_shared_demo():
    # Create a shared file
    with open('/dev/shm/mmap_test', 'wb') as f:
        f.write(b'\x00' * 4096)

    # Map into memory
    with open('/dev/shm/mmap_test', 'r+b') as f:
        mm = mmap.mmap(f.fileno(), 0)

        # Write to shared memory
        mm[0:12] = b'Hello shared'
        print(f'Wrote: {mm[:12].decode()!r}')

        # Another process could read this
        mm.close()

    os.unlink('/dev/shm/mmap_test')

mmap_shared_demo()

import multiprocessing
import time

def worker(pid, sema, counter):
    with sema:
        print(f'Worker {pid}: acquired semaphore')
        counter.value += 1
        time.sleep(0.5)
        print(f'Worker {pid}: releasing semaphore')

if __name__ == '__main__':
    sema = multiprocessing.Semaphore(2)  # Allow 2 concurrent workers
    counter = multiprocessing.Value('i', 0)

    workers = [
        multiprocessing.Process(target=worker, args=(i, sema, counter))
        for i in range(5)
    ]

    for w in workers:
        w.start()

    for w in workers:
        w.join()

    print(f'All workers done. Counter: {counter.value}')

import signal
import time

def handler(signum, frame):
    print(f'Received signal {signum}! Cleaning up...')

def signal_demo():
    signal.signal(signal.SIGUSR1, handler)
    signal.signal(signal.SIGTERM, handler)

    print(f'PID: {os.getpid()}')
    print('Send a signal with: kill -USR1 <pid>')

    # Simulate waiting
    try:
        time.sleep(10)
    except KeyboardInterrupt:
        print('Got Ctrl+C')

if __name__ == '__main__':
    import os as _os
    os = _os
    signal_demo()

import socket
import os
import threading
import time

def socket_server(path='/tmp/ipc_socket'):
    if os.path.exists(path):
        os.unlink(path)

    server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    server.bind(path)
    server.listen(1)
    print(f'Server listening on {path}')

    conn, _ = server.accept()
    data = conn.recv(1024)
    print(f'Server received: {data.decode()!r}')
    conn.send(b'Hello from server!')
    conn.close()
    os.unlink(path)

def socket_client(path='/tmp/ipc_socket'):
    time.sleep(0.2)  # Wait for server
    client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    client.connect(path)
    client.send(b'Hello from client!')
    response = client.recv(1024)
    print(f'Client received: {response.decode()!r}')
    client.close()

server = threading.Thread(target=socket_server)
client = threading.Thread(target=socket_client)
server.start()
client.start()
server.join()
client.join()

import time
import os
import sys

def benchmark_ipc(iterations=10000):
    # Benchmark 1: Pipe
    r, w = os.pipe()
    start = time.time()

    for i in range(iterations):
        os.write(w, b'x')
        os.read(r, 1)

    pipe_time = time.time() - start
    os.close(r)
    os.close(w)
    print(f'Pipe:        {iterations} transfers in {pipe_time:.3f}s')

    # Benchmark 2: Unix socket (loopback)
    import socket
    import threading

    def bench_socket():
        server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server_sock.bind(('127.0.0.1', 0))
        port = server_sock.getsockname()[1]
        server_sock.listen(1)

        client_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        client_sock.connect(('127.0.0.1', port))
        conn, _ = server_sock.accept()

        start = time.time()
        for _ in range(iterations):
            client_sock.send(b'x')
            conn.recv(1)

        socket_time = time.time() - start
        print(f'Socket:      {iterations} transfers in {socket_time:.3f}s')
        client_sock.close()
        conn.close()
        server_sock.close()

    bench_socket()

if __name__ == '__main__':
    benchmark_ipc(10000)