Suppose you are working with a very large list of numbers and need to perform a computation on each element, such as **squaring every number**. Processing the entire list sequentially could be time-consuming, especially as the list grows. To speed up this operation, you can divide the list into segments and assign each segment to a separate process. Each process will work independently to compute the squares for its assigned part of the list, and then the results will be combined at the end. This approach allows you to fully utilize multiple CPU cores and significantly reduce the total processing time for large datasets.


import unittest
import user_code
import ast
import re   
import unittest
import importlib
import io
from contextlib import redirect_stdout

class TestTask(unittest.TestCase):
    def test_parallel_square_basic(self):
        import user_code
        importlib.reload(user_code)
        numbers = [1, 2, 3, 4, 5, 6, 7, 8]
        num_processes = 3
        result = user_code.parallel_square(numbers, num_processes)
        expected = [1, 4, 9, 16, 25, 36, 49, 64]
        _dynamic_test(
            self,
            isinstance(result, list) and result == expected,
            "parallel_square returns correct squared list for 8 numbers, 3 processes",
            f"Expected {expected}, got {result}",
        )

    def test_parallel_square_single_process(self):
        import user_code
        importlib.reload(user_code)
        numbers = [5, 10, 15]
        num_processes = 1
        result = user_code.parallel_square(numbers, num_processes)
        expected = [25, 100, 225]
        _dynamic_test(
            self,
            isinstance(result, list) and result == expected,
            "parallel_square returns correct result for single process",
            f"Expected {expected}, got {result}",
        )

    def test_parallel_square_more_processes_than_elements(self):
        import user_code
        importlib.reload(user_code)
        numbers = [2, 3]
        num_processes = 4
        result = user_code.parallel_square(numbers, num_processes)
        expected = [4, 9]
        _dynamic_test(
            self,
            isinstance(result, list) and result == expected,
            "parallel_square handles more processes than elements",
            f"Expected {expected}, got {result}",
        )

    def test_parallel_square_empty_list(self):
        import user_code
        importlib.reload(user_code)
        numbers = []
        num_processes = 2
        result = user_code.parallel_square(numbers, num_processes)
        expected = []
        _dynamic_test(
            self,
            isinstance(result, list) and result == expected,
            "parallel_square returns empty list for input []",
            f"Expected empty list, got {result}",
        )

    def test_parallel_square_large_numbers(self):
        import user_code
        importlib.reload(user_code)
        numbers = [100, 200, 300, 400]
        num_processes = 2
        result = user_code.parallel_square(numbers, num_processes)
        expected = [10000, 40000, 90000, 160000]
        _dynamic_test(
            self,
            isinstance(result, list) and result == expected,
            "parallel_square correctly squares large numbers",
            f"Expected {expected}, got {result}",
        )

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: Parallel Data Processing

解答