In many real-world applications, you need to coordinate work between different parts of a system running in parallel.

A common pattern for this is the **producer-consumer scenario**:

- A **producer process** generates data (such as numbers) and puts them into a shared `Queue`;
- At the same time, a **consumer process** retrieves data from the queue and processes it, for example by squaring each number and printing the result.

This pattern helps **decouple data creation from data processing**, making your programs more efficient and responsive when working with multiple processes.


import unittest
import user_code
import ast
import re   
import unittest
import importlib
import io
import sys
import os
import re
from multiprocessing import Process, Queue
import time

class TestTask(unittest.TestCase):
    def test_producer_puts_all_numbers_and_done(self):
        import user_code
        importlib.reload(user_code)
        numbers = [10, 20, 30]
        queue = Queue()
        p = Process(target=user_code.producer, args=(numbers, queue))
        p.start()
        p.join()
        results = []
        while not queue.empty():
            results.append(queue.get())
        expected = [10, 20, 30, 'DONE']
        _dynamic_test(
            self,
            results == expected,
            f"Producer puts all numbers and 'DONE' marker in order",
            f"Expected queue items {expected}, got {results}",
        )

    def test_consumer_processes_numbers_and_stops_on_done(self):
        import user_code
        importlib.reload(user_code)
        queue = Queue()
        numbers = [3, 4]
        for n in numbers:
            queue.put(n)
        queue.put('DONE')
        # Consumer runs in a subprocess, so we need to capture its stdout using a pipe
        parent_conn, child_conn = os.pipe()
        def run_consumer_with_pipe(queue, fd):
            # Redirect stdout to pipe
            os.dup2(fd, 1)
            user_code.consumer(queue)
        c = Process(target=run_consumer_with_pipe, args=(queue, child_conn))
        c.start()
        os.close(child_conn)
        output = b''
        while True:
            chunk = os.read(parent_conn, 1024)
            if not chunk:
                break
            output += chunk
            if c.exitcode is not None:
                break
        c.join()
        os.close(parent_conn)
        output = normalize_text(output.decode())
        expected_lines = [f"Processed: {n*n}" for n in numbers]
        for line in expected_lines:
            _dynamic_test(
                self,
                normalize_text(line) in output,
                f"Output includes line: {line}",
                f"Expected output: '{line}' not found in: '{output}'",
            )
        _dynamic_test(
            self,
            'done' not in output,
            "Consumer does not process 'DONE' marker",
            f"Unexpected 'DONE' in output: {output}"
        )

    def test_integration_producer_consumer(self):
        import user_code
        importlib.reload(user_code)
        numbers = [2, 5]
        queue = Queue()
        parent_conn, child_conn = os.pipe()
        def run_consumer_with_pipe(queue, fd):
            os.dup2(fd, 1)
            user_code.consumer(queue)
        p = Process(target=user_code.producer, args=(numbers, queue))
        c = Process(target=run_consumer_with_pipe, args=(queue, child_conn))
        p.start()
        c.start()
        p.join()
        os.close(child_conn)
        output = b''
        while True:
            chunk = os.read(parent_conn, 1024)
            if not chunk:
                break
            output += chunk
            if c.exitcode is not None:
                break
        c.join()
        os.close(parent_conn)
        output = normalize_text(output.decode())
        expected_lines = [f"Processed: {n*n}" for n in numbers]
        for line in expected_lines:
            _dynamic_test(
                self,
                normalize_text(line) in output,
                f"Output includes line: {line}",
                f"Expected output: '{line}' not found in: '{output}'",
            )
        _dynamic_test(
            self,
            'done' not in output,
            "Consumer does not print 'DONE' marker",
            f"Unexpected 'DONE' in output: {output}"
        )

def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\s{2,}", " ", text)
    text = re.sub(r"\s*([,:?])\s*", r"\1 ", text)
    return text.strip()

def change_var(code: str, var_name: str, value: str) -> str:
    tree = ast.parse(code)
    lines = code.splitlines()
    for i, node in enumerate(tree.body):
        if isinstance(node, ast.Assign):
            for target in node.targets:
                if isinstance(target, ast.Name) and target.id == var_name:
                    start_line = node.lineno - 1
                    line = lines[start_line]
                    indent = ' ' * (len(line) - len(line.lstrip()))
                    lines[start_line] = f"{indent}{var_name} = {value}"
                    next_line = len(lines)
                    for next_node in tree.body[i+1:]:
                        if hasattr(next_node, 'lineno'):
                            next_line = next_node.lineno - 1
                            break
                    if next_line > start_line + 1:
                        lines[start_line+1:next_line] = []
                    
                    return '\\n'.join(lines)
    return code

if __name__ == "__main__":
    unittest.main()


test_main.py

A beginner-friendly course introducing the concepts, techniques, and practical applications of multithreading and multiprocessing in Python. Learn how to write concurrent programs, manage threads and processes, and solve real-world problems using parallel execution.

Explore the foundational concepts of concurrency, parallelism, and the differences between threads and processes in Python.

Learn how to create, start, and manage threads in Python, and understand thread synchronization.

Explore how to create and manage processes in Python, and understand inter-process communication.

Delve into advanced concurrency patterns, performance considerations, and best practices for writing robust concurrent code.

Challenge: Producer-Consumer with Processes

Solution

Challenge: Producer-Consumer with Processes

Solution