In modern Python programming, choosing between **threads** and **processes** is a crucial design decision when building concurrent or parallel systems. Each approach has distinct strengths: **threads** shine in I/O-bound scenarios where tasks spend time waiting for input or output operations, while **processes** are ideal for CPU-bound workloads that benefit from true parallel execution. Consider the following real-world programming tasks and think about which concurrency model—**threads** or **processes**—would be most appropriate for each:

- Downloading multiple web pages simultaneously;
- Processing a large collection of high-resolution images;
- Reading and writing records to a shared database;
- Performing complex numerical computations on large datasets;
- Monitoring multiple directories for file changes;
- Running several independent machine learning model training jobs;
- Logging sensor data from several networked devices in real time;
- Rendering frames for an animated video;
- Scraping data from multiple websites at once;
- Compressing a batch of files into separate archives.


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

class TestTask(unittest.TestCase):
    def test_task1(self):
        import user_code
        importlib.reload(user_code)
        expected = "thread"
        actual = getattr(user_code, 'task1', None)
        _dynamic_test(
            self,
            actual == expected,
            "Correct classification for task 1.",
            f"Expected {expected}, got {actual}"
        )

    def test_task2(self):
        import user_code
        importlib.reload(user_code)
        expected = "process"
        actual = getattr(user_code, 'task2', None)
        _dynamic_test(
            self,
            actual == expected,
            "Correct classification for task 2.",
            f"Expected {expected}, got {actual}"
        )

    def test_task3(self):
        import user_code
        importlib.reload(user_code)
        expected = "thread"
        actual = getattr(user_code, 'task3', None)
        _dynamic_test(
            self,
            actual == expected,
            "Correct classification for task 3.",
            f"Expected {expected}, got {actual}"
        )

    def test_task4(self):
        import user_code
        importlib.reload(user_code)
        expected = "process"
        actual = getattr(user_code, 'task4', None)
        _dynamic_test(
            self,
            actual == expected,
            "Correct classification for task 4.",
            f"Expected {expected}, got {actual}"
        )

    def test_task5(self):
        import user_code
        importlib.reload(user_code)
        expected = "thread"
        actual = getattr(user_code, 'task5', None)
        _dynamic_test(
            self,
            actual == expected,
            "Correct classification for task 5.",
            f"Expected {expected}, got {actual}"
        )

    def test_task6(self):
        import user_code
        importlib.reload(user_code)
        expected = "process"
        actual = getattr(user_code, 'task6', None)
        _dynamic_test(
            self,
            actual == expected,
            "Correct classification for task 6.",
            f"Expected {expected}, got {actual}"
        )

    def test_task7(self):
        import user_code
        importlib.reload(user_code)
        expected = "thread"
        actual = getattr(user_code, 'task7', None)
        _dynamic_test(
            self,
            actual == expected,
            "Correct classification for task 7.",
            f"Expected {expected}, got {actual}"
        )

    def test_task8(self):
        import user_code
        importlib.reload(user_code)
        expected = "process"
        actual = getattr(user_code, 'task8', None)
        _dynamic_test(
            self,
            actual == expected,
            "Correct classification for task 8.",
            f"Expected {expected}, got {actual}"
        )

    def test_task9(self):
        import user_code
        importlib.reload(user_code)
        expected = "thread"
        actual = getattr(user_code, 'task9', None)
        _dynamic_test(
            self,
            actual == expected,
            "Correct classification for task 9.",
            f"Expected {expected}, got {actual}"
        )

    def test_task10(self):
        import user_code
        importlib.reload(user_code)
        expected = "process"
        actual = getattr(user_code, 'task10', None)
        _dynamic_test(
            self,
            actual == expected,
            "Correct classification for task 10.",
            f"Expected {expected}, got {actual}"
        )

# _dynamic_test, normalize_text, change_var provided by the platform

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

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: Classify Thread vs. Process Use Cases

Lösung