230. Multiprocessing Pipelines

1. Basic Queue Communication Example

A simple example of how to pass data between processes using a queue.

import multiprocessing

def producer(q):
    for i in range(5):
        print(f"Producer: {i}")
        q.put(i)

def consumer(q):
    while True:
        item = q.get()
        if item is None:  # Exit condition
            break
        print(f"Consumer: {item}")

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

    p1.join()
    q.put(None)  # Signal consumer to exit
    p2.join()

Explanation:

• The producer puts data into the queue, and the consumer retrieves it.

• The queue is shared between processes for communication.

---

2. Using Pipe for Simple Communication

A simple producer-consumer pipeline using a pipe for two-way communication.

import multiprocessing

def producer(pipe):
    for i in range(5):
        print(f"Producer: {i}")
        pipe.send(i)

def consumer(pipe):
    while True:
        item = pipe.recv()
        print(f"Consumer: {item}")

if __name__ == "__main__":
    parent_conn, child_conn = multiprocessing.Pipe()
    p1 = multiprocessing.Process(target=producer, args=(parent_conn,))
    p2 = multiprocessing.Process(target=consumer, args=(child_conn,))

    p1.start()
    p2.start()

    p1.join()
    p2.terminate()

Explanation:

• The Pipe provides a unidirectional channel for communication between two processes.

• The producer sends data via the pipe, and the consumer receives it.

---

3. Multiprocessing with Queue and Multiple Consumers

Using a queue with multiple consumers to handle messages concurrently.

import multiprocessing

def consumer(q, id):
    while True:
        item = q.get()
        if item == "STOP":
            break
        print(f"Consumer {id}: {item}")

if __name__ == "__main__":
    q = multiprocessing.Queue()
    
    # Start multiple consumers
    consumers = [multiprocessing.Process(target=consumer, args=(q, i)) for i in range(3)]
    for c in consumers:
        c.start()

    # Producer
    for i in range(5):
        q.put(i)

    # Stop consumers
    for c in consumers:
        q.put("STOP")

    for c in consumers:
        c.join()

Explanation:

• Multiple consumers can process items concurrently from a shared queue.

• A "STOP" signal is used to terminate the consumers.

---

4. Multiprocessing Pipeline with Function Chaining

Building a pipeline where the output of one process becomes the input of the next.

import multiprocessing

def task1(pipe_out):
    pipe_out.send("Hello")
    pipe_out.send("World")

def task2(pipe_in, pipe_out):
    while True:
        msg = pipe_in.recv()
        if msg == "END":
            break
        print(f"Task 2 received: {msg}")
        pipe_out.send(f"Processed {msg}")

def task3(pipe_in):
    while True:
        msg = pipe_in.recv()
        if msg == "END":
            break
        print(f"Task 3 processed: {msg}")

if __name__ == "__main__":
    pipe1_out, pipe1_in = multiprocessing.Pipe()
    pipe2_out, pipe2_in = multiprocessing.Pipe()

    p1 = multiprocessing.Process(target=task1, args=(pipe1_out,))
    p2 = multiprocessing.Process(target=task2, args=(pipe1_in, pipe2_out))
    p3 = multiprocessing.Process(target=task3, args=(pipe2_in,))

    p1.start()
    p2.start()
    p3.start()

    p1.join()
    pipe1_out.send("END")  # End signal for task2
    p2.join()
    pipe2_out.send("END")  # End signal for task3
    p3.join()

Explanation:

• This demonstrates chaining multiple processes using pipes, where each task sends its output to the next.

---

5. Pipe and Queue for Error Handling

Using a pipe and a queue to send errors and data between processes.

import multiprocessing
import time

def task_with_error(queue, pipe):
    try:
        for i in range(5):
            if i == 3:
                raise ValueError("An error occurred in task")
            queue.put(i)
            time.sleep(1)
    except Exception as e:
        pipe.send(str(e))

def handle_errors(pipe):
    error_message = pipe.recv()
    print(f"Error received: {error_message}")

if __name__ == "__main__":
    queue = multiprocessing.Queue()
    pipe = multiprocessing.Pipe()

    p1 = multiprocessing.Process(target=task_with_error, args=(queue, pipe))
    p2 = multiprocessing.Process(target=handle_errors, args=(pipe,))

    p1.start()
    p2.start()

    p1.join()
    p2.join()

Explanation:

• If an error occurs in one process, it’s communicated to another process via the pipe.

---

6. Using Queue for Worker Pool

Managing a pool of worker processes that process tasks using a queue.

import multiprocessing
import time

def worker(queue):
    while True:
        item = queue.get()
        if item is None:
            break
        print(f"Processing item: {item}")
        time.sleep(1)

if __name__ == "__main__":
    queue = multiprocessing.Queue()
    workers = [multiprocessing.Process(target=worker, args=(queue,)) for _ in range(3)]
    
    for w in workers:
        w.start()

    for item in range(5):
        queue.put(item)

    for _ in workers:
        queue.put(None)  # Signal to stop workers

    for w in workers:
        w.join()

Explanation:

• A pool of worker processes consumes tasks from the queue and processes them concurrently.

• Workers exit when receiving a None value.

---

7. Multiprocessing with Shared Memory

Using shared memory for communication between processes.

import multiprocessing

def worker(shared_data):
    shared_data[0] += 1
    print(f"Shared data: {shared_data[0]}")

if __name__ == "__main__":
    shared_data = multiprocessing.Array('i', [0])  # Shared memory array
    workers = [multiprocessing.Process(target=worker, args=(shared_data,)) for _ in range(5)]
    
    for w in workers:
        w.start()

    for w in workers:
        w.join()

    print(f"Final shared data: {shared_data[0]}")

Explanation:

• Processes share memory using multiprocessing.Array, allowing them to modify the shared data.

---

8. Using Queue for Task Distribution

Distribute tasks to worker processes using a queue.

import multiprocessing

def process_task(queue):
    while True:
        task = queue.get()
        if task == "STOP":
            break
        print(f"Processing: {task}")

if __name__ == "__main__":
    queue = multiprocessing.Queue()
    workers = [multiprocessing.Process(target=process_task, args=(queue,)) for _ in range(3)]
    
    for w in workers:
        w.start()

    tasks = ["task1", "task2", "task3", "task4", "task5"]
    for task in tasks:
        queue.put(task)

    for _ in workers:
        queue.put("STOP")  # Stop signal

    for w in workers:
        w.join()

Explanation:

• A set of workers retrieves tasks from the queue and processes them.

• The STOP signal terminates the workers.

---

9. Pipe and Queue for Synchronization

Using both a pipe and a queue to synchronize tasks and communicate status.

import multiprocessing

def task1(pipe):
    pipe.send("Task 1 completed")

def task2(pipe):
    status = pipe.recv()
    print(f"Received from Task 1: {status}")

if __name__ == "__main__":
    pipe1, pipe2 = multiprocessing.Pipe()
    
    p1 = multiprocessing.Process(target=task1, args=(pipe1,))
    p2 = multiprocessing.Process(target=task2, args=(pipe2,))
    
    p1.start()
    p2.start()

    p1.join()
    p2.join()

Explanation:

• The pipe is used to communicate the status of one task to another, demonstrating synchronization.

---

10. Using Queue to Implement a Producer-Consumer Pipeline

Implementing a classic producer-consumer pipeline using queues.

import multiprocessing
import time

def producer(queue):
    for i in range(5):
        print(f"Producer: {i}")
        queue.put(i)
        time.sleep(1)

def consumer(queue):
    while True:
        item = queue.get()
        if item == "STOP":
            break
        print(f"Consumer: {item}")

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()
    queue.put("STOP")
    p2.join()

Explanation:

• A producer produces tasks and puts them into a queue.

• A consumer retrieves tasks from the queue and processes them. The consumer stops when receiving a "STOP" signal.

---