Il multiprocessing in Python consente di eseguire più processi in parallelo, ciascuno con il proprio interprete Python e spazio di memoria. Questo approccio è particolarmente utile per attività che richiedono molta CPU, dove si desidera sfruttare più core della CPU per eseguire calcoli più rapidamente. Utilizzando il multiprocessing, è possibile migliorare significativamente le prestazioni per operazioni che richiedono elaborazioni intensive, come l'elaborazione dei dati, i calcoli scientifici o le simulazioni.

La classe `multiprocessing.Manager` offre un modo per creare oggetti condivisi - come liste e dizionari - che possono essere accessibili e modificati in modo sicuro da più processi. Quando è necessario raccogliere risultati o coordinare dati tra diverse istanze di `Process`, utilizzare sempre un `Manager` per creare questi oggetti condivisi. Le normali liste e dizionari Python non sono condivisi tra i processi, quindi le modifiche apportate in un processo non saranno visibili agli altri. Con una lista gestita da `Manager`, ogni processo può aggiungere o modificare dati, e tutte le modifiche saranno visibili a ogni processo che utilizza quell'oggetto condiviso.

Nota

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))

Questo codice dimostra come utilizzare `multiprocessing.Manager` per creare una lista condivisa a cui più processi possono accedere e modificare. Ogni processo calcola il quadrato di un numero e aggiunge il risultato alla lista condivisa. Utilizzando una lista gestita da `Manager`, si garantisce che tutti i processi possano condividere e raccogliere dati in modo sicuro, rendendo possibile raccogliere i risultati da processi separati che vengono eseguiti in parallelo.

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

Acquisizione di competenze pratiche nella gestione dei moduli e delle importazioni in Python per una gestione efficiente del codice. Padronanza delle tecniche avanzate di gestione degli errori per applicazioni più affidabili, automazione delle operazioni sui file con la gestione dei file e apprendimento di strategie di testing robuste con Pytest e Unittest per garantire l'integrità del codice.

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.

Sfida: Il Collezionista Parallelo di Quadrati

Soluzione