Multiprosessering i Python gjør det mulig å kjøre flere prosesser parallelt, hver med sin egen Python-tolk og minneområde. Denne tilnærmingen er spesielt nyttig for CPU-krevende oppgaver, hvor du ønsker å utnytte flere CPU-kjerner for å utføre beregninger raskere. Ved å bruke multiprosessering kan du forbedre ytelsen betydelig for operasjoner som krever tung beregning, som databehandling, vitenskapelige kalkulasjoner eller simuleringer.

Klassen `multiprocessing.Manager` gir en måte å opprette delte objekter – som lister og ordbøker – som trygt kan nås og endres av flere prosesser. Når du trenger å samle resultater eller koordinere data mellom separate `Process`-instanser, bør du alltid bruke en `Manager` for å opprette disse delte objektene. Vanlige Python-lister og ordbøker deles ikke mellom prosesser, så endringer gjort i én prosess vil ikke være synlige for andre. Med en `Manager`-liste kan hver prosess legge til eller endre data, og alle endringer vil være synlige for hver prosess som bruker det delte objektet.

Merk

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

Denne koden viser hvordan man bruker `multiprocessing.Manager` for å opprette en delt liste som flere prosesser kan få tilgang til og endre. Hver prosess beregner kvadratet av et tall og legger resultatet sitt til den delte listen. Ved å bruke en `Manager`-liste sikrer du at alle prosesser trygt kan dele og samle data, noe som gjør det mulig å hente resultater fra separate prosesser som kjører parallelt.

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

Du vil tilegne deg praktiske ferdigheter i bruk av Python-moduler og import for effektiv kodehåndtering. Behersk avanserte teknikker for feilhåndtering for mer pålitelige applikasjoner, automatiser filoperasjoner med filhåndtering, og lær robuste teststrategier med Pytest og Unittest for å sikre kodeintegritet.

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.

Utfordring: Den Parallelle Kvadratsamleren

Løsning