Imagine you are developing a logging system where multiple threads record events as they happen. Each thread represents a separate component of your application, and every time an event occurs, the thread needs to increment a shared counter to keep track of the total number of events logged. If you allow all threads to access and update the counter without coordination, you risk introducing **race conditions**—where two or more threads read and write the counter at the same time, leading to incorrect totals. To prevent this, you need to ensure that only one thread can update the counter at any given moment, making your counter **thread-safe**.


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

class TestTask(unittest.TestCase):
    def test_increment_and_value(self):
        importlib.reload(user_code)
        counter = user_code.ThreadSafeCounter()
        for _ in range(5):
            counter.increment()
        val = counter.get_value()
        _dynamic_test(
            self,
            val == 5,
            "Counter increments and returns correct value",
            f"Expected value 5 after 5 increments, got {val}",
        )

    def test_thread_safety(self):
        importlib.reload(user_code)
        counter = user_code.ThreadSafeCounter()
        threads = []
        num_threads = 20
        increments_per_thread = 500
        for _ in range(num_threads):
            t = threading.Thread(target=user_code.worker, args=(counter, increments_per_thread))
            threads.append(t)
            t.start()
        for t in threads:
            t.join()
        val = counter.get_value()
        expected = num_threads * increments_per_thread
        _dynamic_test(
            self,
            val == expected,
            "ThreadSafeCounter is thread-safe for concurrent increments",
            f"Expected {expected}, got {val} after all threads finished",
        )

    def test_final_print_output(self):
        importlib.reload(user_code)
        f = io.StringIO()
        with redirect_stdout(f):
            importlib.reload(user_code)
        output = normalize_text(f.getvalue())
        expected = normalize_text(f"Final counter value: 10000")
        _dynamic_test(
            self,
            expected in output,
            f"Output includes line: {expected}",
            f"Expected output: '{expected}' not found in: '{output}'",
        )

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: Multithreaded Counter

Рішення