mandelbrot_threading.py

#!/usr/bin/env python3
# mandelbrot_threaded.py
import threading
import os


def mandelbrot_checksum_range(width, height, max_iter, xmin, xmax, ymin, ymax,
                               row_start, row_end, results, thread_id):
    """Calculate Mandelbrot checksum for a range of rows."""
    dx = (xmax - xmin) / (width - 1)
    dy = (ymax - ymin) / (height - 1)
    partial_total = 0

    for row in range(row_start, row_end):
        c_imag = ymin + row * dy
        for col in range(width):
            c_real = xmin + col * dx

            # z = x + yi (starts at 0), iterate z = z² + c
            x = 0.0
            y = 0.0
            iterations = 0

            while (x * x + y * y) <= 4.0 and iterations < max_iter:
                x, y = x * x - y * y + c_real, 2.0 * x * y + c_imag
                iterations += 1

            partial_total += iterations

    results[thread_id] = partial_total


def mandelbrot_checksum(width=1800, height=1200, max_iter=1200,
                        xmin=-2.0, xmax=1.0, ymin=-1.2, ymax=1.2,
                        num_threads=None):
    """Calculate Mandelbrot set using threading and return checksum."""
    if num_threads is None:
        num_threads = os.cpu_count()

    threads = []
    results = [0] * num_threads
    rows_per_thread = height // num_threads

    for i in range(num_threads):
        row_start = i * rows_per_thread
        row_end = height if i == num_threads - 1 else (i + 1) * rows_per_thread

        t = threading.Thread(
            target=mandelbrot_checksum_range,
            args=(width, height, max_iter, xmin, xmax, ymin, ymax,
                  row_start, row_end, results, i)
        )
        threads.append(t)
        t.start()

    for t in threads:
        t.join()

    return sum(results)


if __name__ == "__main__":
    cs = mandelbrot_checksum()
    print(f"checksum={cs}")