Opinion: Python 3.13's New GIL Improvements Make It Viable for Multi-Threaded Workloads

import sys
import time
import threading
from argparse import ArgumentParser
from statistics import mean, stdev

# Check Python version to warn if not 3.13+
if sys.version_info < (3, 13):
    print(f"WARNING: Running on Python {sys.version_info.major}.{sys.version_info.minor}, GIL improvements require 3.13+")

def matrix_multiply_thread(rows_a, cols_a, rows_b, cols_b, thread_id, results, errors):
    \"\"\"Perform matrix multiplication for a subset of rows, with error handling.\"\"\"
    try:
        if cols_a != rows_b:
            raise ValueError(f"Thread {thread_id}: Column count of A ({cols_a}) must match row count of B ({rows_b})")

        # Initialize result submatrix
        sub_result = [[0 for _ in range(cols_b)] for _ in range(rows_a)]

        for i in range(rows_a):
            for k in range(cols_a):
                a_val = i * cols_a + k  # Simulated matrix A value
                for j in range(cols_b):
                    b_val = k * cols_b + j  # Simulated matrix B value
                    sub_result[i][j] += a_val * b_val

        results[thread_id] = sub_result
        print(f"Thread {thread_id} completed successfully")
    except Exception as e:
        errors[thread_id] = str(e)
        print(f"Thread {thread_id} failed: {str(e)}")

def run_benchmark(num_threads, matrix_size):
    \"\"\"Run multi-threaded matrix multiplication benchmark and return elapsed time.\"\"\"
    # Split matrix rows across threads
    rows_per_thread = matrix_size // num_threads
    remainder = matrix_size % num_threads

    threads = []
    results = {}
    errors = {}

    start_time = time.perf_counter()

    for tid in range(num_threads):
        # Distribute remainder rows to first threads
        start_row = tid * rows_per_thread + min(tid, remainder)
        end_row = start_row + rows_per_thread + (1 if tid < remainder else 0)
        thread_rows = end_row - start_row

        if thread_rows <= 0:
            continue

        t = threading.Thread(
            target=matrix_multiply_thread,
            args=(thread_rows, matrix_size, matrix_size, matrix_size, tid, results, errors)
        )
        threads.append(t)
        t.start()

    # Wait for all threads to complete
    for t in threads:
        t.join()

    elapsed = time.perf_counter() - start_time

    # Check for errors
    if errors:
        print(f"Benchmark failed with {len(errors)} errors: {list(errors.values())}")
        return None

    print(f"Completed {num_threads}-thread benchmark for {matrix_size}x{matrix_size} matrix in {elapsed:.4f}s")
    return elapsed

if __name__ == "__main__":
    parser = ArgumentParser(description="Python 3.13 GIL Contention Benchmark")
    parser.add_argument("--threads", type=int, default=8, help="Number of threads to use (default: 8)")
    parser.add_argument("--size", type=int, default=1024, help="Matrix size (default: 1024x1024)")
    parser.add_argument("--runs", type=int, default=5, help="Number of benchmark runs (default: 5)")

    args = parser.parse_args()

    print(f"Running benchmark on Python {sys.version}")
    print(f"Configuration: {args.threads} threads, {args.size}x{args.size} matrix, {args.runs} runs")

    run_times = []
    for run in range(args.runs):
        print(f"\nRun {run + 1}/{args.runs}")
        elapsed = run_benchmark(args.threads, args.size)
        if elapsed:
            run_times.append(elapsed)

    if run_times:
        print(f"\n=== Benchmark Results ===")
        print(f"Mean elapsed time: {mean(run_times):.4f}s")
        print(f"Std dev: {stdev(run_times):.4f}s" if len(run_times) > 1 else "Single run, no std dev")
        print(f"Speedup over single-threaded (estimated): {run_times[-1] / mean(run_times):.2f}x" if len(run_times) == 1 else "")

import sys
import time
import os
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from argparse import ArgumentParser

def count_primes_in_range(start, end):
    \"\"\"Count primes in [start, end) with error handling.\"\"\"
    try:
        if start < 2:
            start = 2
        if end <= start:
            return 0

        prime_count = 0
        for num in range(start, end):
            if num < 2:
                continue
            is_prime = True
            # Optimized prime check: only up to sqrt(num)
            limit = int(num ** 0.5) + 1
            for divisor in range(2, limit):
                if num % divisor == 0:
                    is_prime = False
                    break
            if is_prime:
                prime_count += 1
        return prime_count
    except Exception as e:
        print(f"Error counting primes in [{start}, {end}): {str(e)}")
        return 0

def run_multi_threaded(num_threads, total_numbers):
    \"\"\"Run prime counting with ThreadPoolExecutor, return elapsed time.\"\"\"
    chunk_size = total_numbers // num_threads
    futures = []

    start_time = time.perf_counter()

    with ThreadPoolExecutor(max_workers=num_threads) as executor:
        for tid in range(num_threads):
            chunk_start = tid * chunk_size
            chunk_end = chunk_start + chunk_size if tid != num_threads -1 else total_numbers
            futures.append(executor.submit(count_primes_in_range, chunk_start, chunk_end))

    total_primes = sum(future.result() for future in futures)
    elapsed = time.perf_counter() - start_time

    print(f"Multi-threaded ({num_threads} threads): {total_primes} primes found in {elapsed:.4f}s")
    return elapsed, total_primes

def run_multi_processing(num_processes, total_numbers):
    \"\"\"Run prime counting with ProcessPoolExecutor, return elapsed time.\"\"\"
    chunk_size = total_numbers // num_processes
    futures = []

    start_time = time.perf_counter()

    with ProcessPoolExecutor(max_workers=num_processes) as executor:
        for pid in range(num_processes):
            chunk_start = pid * chunk_size
            chunk_end = chunk_start + chunk_size if pid != num_processes -1 else total_numbers
            futures.append(executor.submit(count_primes_in_range, chunk_start, chunk_end))

    total_primes = sum(future.result() for future in futures)
    elapsed = time.perf_counter() - start_time

    print(f"Multi-processing ({num_processes} processes): {total_primes} primes found in {elapsed:.4f}s")
    return elapsed, total_primes

if __name__ == "__main__":
    parser = ArgumentParser(description="Python 3.13 Multi-threading vs Multi-processing Benchmark")
    parser.add_argument("--threads", type=int, default=8, help="Number of threads (default: 8)")
    parser.add_argument("--processes", type=int, default=8, help="Number of processes (default: 8)")
    parser.add_argument("--total", type=int, default=1000000, help="Total numbers to check (default: 1M)")

    args = parser.parse_args()

    print(f"Running on Python {sys.version}")
    print(f"OS: {os.name}, CPU cores: {os.cpu_count()}")
    print(f"Configuration: {args.threads} threads, {args.processes} processes, {args.total} numbers")

    # Warmup run to avoid startup overhead
    print("\nWarmup run...")
    count_primes_in_range(0, 10000)

    print("\n=== Multi-threaded Run ===")
    mt_time, mt_primes = run_multi_threaded(args.threads, args.total)

    print("\n=== Multi-processing Run ===")
    mp_time, mp_primes = run_multi_processing(args.processes, args.total)

    print("\n=== Comparison ===")
    print(f"Multi-threaded time: {mt_time:.4f}s")
    print(f"Multi-processing time: {mp_time:.4f}s")
    if mt_time < mp_time:
        print(f"Multi-threading is {mp_time/mt_time:.2f}x faster than multi-processing")
    else:
        print(f"Multi-processing is {mt_time/mp_time:.2f}x faster than multi-threading")

    # Verify prime counts match
    if mt_primes != mp_primes:
        print(f"WARNING: Prime counts do not match! MT: {mt_primes}, MP: {mp_primes}")
    else:
        print(f"Prime counts match: {mt_primes}")

import sys
import time
import threading
from argparse import ArgumentParser
from statistics import mean

def apply_blur_to_region(image_data, width, height, start_row, end_row, thread_id, results, errors):
    \"\"\"Apply a 3x3 box blur to a region of the image, with error handling.\"\"\"
    try:
        if start_row < 0 or end_row > height or start_row >= end_row:
            raise ValueError(f"Thread {thread_id}: Invalid row range [{start_row}, {end_row})")

        # Box blur kernel: 3x3, each pixel is average of 9 neighbors
        blurred_region = []

        for y in range(start_row, end_row):
            row = []
            for x in range(width):
                # Collect 3x3 neighborhood, clamp to image bounds
                r_sum, g_sum, b_sum = 0, 0, 0
                count = 0

                for dy in (-1, 0, 1):
                    for dx in (-1, 0, 1):
                        nx = max(0, min(width - 1, x + dx))
                        ny = max(0, min(height - 1, y + dy))

                        # Get pixel value (simulated as (r, g, b) tuples)
                        pixel = image_data[ny * width + nx]
                        r_sum += pixel[0]
                        g_sum += pixel[1]
                        b_sum += pixel[2]
                        count += 1

                # Average the sums
                row.append((
                    r_sum // count,
                    g_sum // count,
                    b_sum // count
                ))
            blurred_region.append(row)
            # Print progress every 100 rows
            if (y - start_row) % 100 == 0:
                print(f"Thread {thread_id}: Processed {y - start_row}/{end_row - start_row} rows")

        results[thread_id] = blurred_region
        print(f"Thread {thread_id} completed blur for {end_row - start_row} rows")
    except Exception as e:
        errors[thread_id] = str(e)
        print(f"Thread {thread_id} failed: {str(e)}")

def simulate_image(width, height):
    \"\"\"Generate a simulated image with random RGB pixels (using deterministic values for reproducibility).\"\"\"
    image = []
    for y in range(height):
        for x in range(width):
            # Deterministic "random" value based on coordinates
            r = (x * 31 + y * 17) % 256
            g = (x * 23 + y * 29) % 256
            b = (x * 13 + y * 37) % 256
            image.append((r, g, b))
    return image

def run_image_blur_benchmark(num_threads, width, height):
    \"\"\"Run multi-threaded image blur and return elapsed time.\"\"\"
    # Simulate input image
    print(f"Simulating {width}x{height} image...")
    image_data = simulate_image(width, height)
    print(f"Image simulated. Total pixels: {len(image_data)}")

    # Split rows across threads
    rows_per_thread = height // num_threads
    remainder = height % num_threads

    threads = []
    results = {}
    errors = {}

    start_time = time.perf_counter()

    for tid in range(num_threads):
        start_row = tid * rows_per_thread + min(tid, remainder)
        end_row = start_row + rows_per_thread + (1 if tid < remainder else 0)
        thread_rows = end_row - start_row

        if thread_rows <= 0:
            continue

        t = threading.Thread(
            target=apply_blur_to_region,
            args=(image_data, width, height, start_row, end_row, tid, results, errors)
        )
        threads.append(t)
        t.start()

    # Wait for all threads
    for t in threads:
        t.join()

    elapsed = time.perf_counter() - start_time

    if errors:
        print(f"Blur failed with {len(errors)} errors: {list(errors.values())}")
        return None

    # Reassemble blurred image
    blurred_image = []
    for tid in sorted(results.keys()):
        blurred_image.extend(results[tid])

    print(f"Blurred image reassembled. Total rows: {len(blurred_image)}")
    print(f"{num_threads}-thread blur completed in {elapsed:.4f}s")
    return elapsed

if __name__ == "__main__":
    parser = ArgumentParser(description="Python 3.13 Multi-threaded Image Blur Benchmark")
    parser.add_argument("--threads", type=int, default=4, help="Number of threads (default: 4)")
    parser.add_argument("--width", type=int, default=1920, help="Image width (default: 1920)")
    parser.add_argument("--height", type=int, default=1080, help="Image height (default: 1080)")

    args = parser.parse_args()

    print(f"Running on Python {sys.version}")
    print(f"Configuration: {args.threads} threads, {args.width}x{args.height} image")

    elapsed = run_image_blur_benchmark(args.threads, args.width, args.height)

    if elapsed:
        # Calculate throughput
        pixels = args.width * args.height
        throughput = pixels / elapsed / 1e6  # MPixels per second
        print(f"\nThroughput: {throughput:.2f} MPixels/s")

import threading

def create_token_aware_pool(num_threads):
    \"\"\"Create a thread pool optimized for 3.13 per-thread GIL tokens.\"\"\"
    # Set thread stack size to reduce overhead (optional, 1MB per thread)
    threading.stack_size(1024 * 1024)
    # Daemonize threads to avoid blocking shutdown
    pool = []
    for _ in range(num_threads):
        t = threading.Thread(target=None, daemon=True)
        pool.append(t)
    return pool

import sys
import time

def print_gil_stats(interval=5):
    \"\"\"Print GIL contention stats every interval seconds.\"\"\"
    if not hasattr(sys, "_gilstats"):
        print("sys._gilstats not available, upgrade to Python 3.13+")
        return
    sys._gilstats.enable()
    while True:
        time.sleep(interval)
        stats = sys._gilstats.get_stats()
        print(f"GIL Hold Time (mean): {stats['hold_time_mean']:.2f}ms")
        print(f"Contention Events: {stats['contention_events']}")
        print(f"Token Acquisitions: {stats['acquisitions']}")

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def get_executor(workload_type, max_workers):
    \"\"\"Select executor based on workload type.\"\"\"
    if workload_type == "cpu-bound":
        # Use ThreadPool for 3.13+ CPU-bound work
        return ThreadPoolExecutor(max_workers=max_workers)
    elif workload_type == "io-bound":
        # ThreadPool works for IO-bound too
        return ThreadPoolExecutor(max_workers=max_workers * 2)
    else:
        # Mixed or memory-heavy: use multiprocessing
        return ProcessPoolExecutor(max_workers=max_workers)