Multiprocessing in Python allows you to run multiple processes in parallel, each with its own Python interpreter and memory space. This approach is especially useful for CPU-bound tasks, where you want to leverage multiple CPU cores to perform computations faster. By using multiprocessing, you can significantly improve performance for operations that require heavy computation, such as data processing, scientific calculations, or simulations.

The `multiprocessing.Manager` class provides a way to create shared objects - such as lists and dictionaries - that can be safely accessed and modified by multiple processes. When you need to collect results or coordinate data between separate `Process` instances, always use a `Manager` to create these shared objects. Regular Python lists and dictionaries are not shared between processes, so changes made in one process will not be visible to others. With a `Manager` list, each process can append or modify data, and all changes will be visible to every process using that shared object.

Note

from multiprocessing import Process, Manager

def add_value(value, shared_list):
    shared_list.append(value)

if __name__ == "__main__":
    manager = Manager()
    shared_list = manager.list()
    processes = []
    values = [10, 20, 30, 40, 50]

    for v in values:
        p = Process(target=add_value, args=(v, shared_list))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

    print("Collected results:", list(shared_list))


This code demonstrates how to use `multiprocessing.Manager` to create a shared list that multiple processes can access and modify. Each process computes the square of a number and appends its result to the shared list. By using a `Manager` list, you ensure that all processes can safely share and collect data, making it possible to gather results from separate processes running in parallel.

import unittest
import user_code
import importlib
import ast
import re
import sys

class TestTask(unittest.TestCase):
    def test_compute_square_function(self):
        import user_code
        importlib.reload(user_code)
        func = getattr(user_code, 'compute_square', None)
        _dynamic_test(self, callable(func),
            'compute_square function is defined',
            'compute_square function is missing or not callable.'
        )
        if callable(func):
            results = []
            func(3, results)
            _dynamic_test(self, results == [9],
                'compute_square(3, results) appends 9',
                f'compute_square(3, results) should append 9, got {results}'
            )
            results = []
            func(0, results)
            _dynamic_test(self, results == [0],
                'compute_square(0, results) appends 0',
                f'compute_square(0, results) should append 0, got {results}'
            )
            results = []
            func(-2, results)
            _dynamic_test(self, results == [4],
                'compute_square(-2, results) appends 4',
                f'compute_square(-2, results) should append 4, got {results}'
            )

    def test_results_variable(self):
        import user_code
        importlib.reload(user_code)
        results = getattr(user_code, 'results', None)
        _dynamic_test(self,
            isinstance(results, list),
            'results is a list',
            f'results is not a list: {results}'
        )
        expected = set([1, 4, 9, 16, 25])
        if isinstance(results, list):
            _dynamic_test(self,
                set(results) == expected and len(results) == 5,
                f'results contains all squares: {expected}',
                f'Expected results: {expected}, got: {set(results)}'
            )

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()
    changed = False
    assign_nodes = [
        (i, node)
        for i, node in enumerate(tree.body)
        if isinstance(node, ast.Assign)
        and any(isinstance(target, ast.Name) and target.id == var_name for target in node.targets)
    ]
    if not assign_nodes:
        return code
    for i, node in reversed(assign_nodes):
        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] = []
        changed = True
    return '\\n'.join(lines) if changed else code

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


test_main.py

A deep dive into advanced structural programming concepts in Python, focusing on robust file handling, error management, concurrency, asynchronous programming and testing.

Master Python's exception system for robust error handling and debugging.

Explore advanced techniques for file operations and resource management using context managers.

Compare and apply Python's multithreading and multiprocessing modules for concurrent programming.

Master asynchronous programming in Python using the asyncio library for scalable I/O-bound applications.

Explore how to write, organize, and run tests using Python's unittest and pytest frameworks.

Challenge: The Parallel Square Collector

Oplossing