check_concurrency_pcr_math.py

"""
check_concurrency_pcr.py
pcr: pre calculated randoms

Compare runtimes for a monte carlo simulation of pi by using a plain
python function, multiple threads, processes and subinterpreters (when
supported).

Random numbers are calculated before the monte carlo simulation starts
and provided as arguments to calculate_pi, resulting in an overhead for
inter-process and subinterpreter communication.

The calculation of the random numbers in python 3.13t starts to take a
long time if the list exceeds about 2e6 tuples of random data.
"""

import argparse
import itertools
import math
import sys
import time

from random import random
from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import ProcessPoolExecutor


def _has_isolated_interpreters():
    try:
        return sys.implementation.supports_isolated_interpreters
    except AttributeError:
        return False

        
HAS_ISOLATED_INTERPRETERS = _has_isolated_interpreters()
if HAS_ISOLATED_INTERPRETERS:
    from concurrent.futures import InterpreterPoolExecutor

    
def calculate_pi(random_numbers):
    inside = 0
    for x, y in random_numbers:
        if math.sqrt(x*x + y*y) < 1:
            inside += 1
    pi = 4 * inside / len(random_numbers)
    return pi
        

class ConcurrencyRunner:

    def __init__(self, runs, tasks):
        self.runs = runs
        self.tasks = tasks
        self.random_numbers = [(random(), random()) for _ in range(self.runs)]
                
    def calculate_pi_by_executor(self, executor):
        with executor(max_workers=self.tasks) as e:
            items_per_task = len(self.random_numbers) // self.tasks
            args = list(itertools.batched(self.random_numbers, items_per_task))
            args = args[:self.tasks]
            results = e.map(calculate_pi, args)  # potential expensive ipc
            result = sum(results) / self.tasks
        return result
    
    def calculate_pi_by_threads(self, *_):
        return self.calculate_pi_by_executor(ThreadPoolExecutor)

    def calculate_pi_by_processes(self, *_):
        return self.calculate_pi_by_executor(ProcessPoolExecutor)

    def calculate_pi_by_subinterpreters(self, *_):
        return self.calculate_pi_by_executor(InterpreterPoolExecutor)

    def get_runtime(self, function):
        start = time.perf_counter()
        result = function(self.random_numbers)
        stop = time.perf_counter()
        delta = stop - start
        return delta, result
        
    def run(self):

        def print_header():
            fn = "function name"
            rt = "runtime"
            fc = "factor"
            rs = "result"
            header = f"{fn:32}{rt:>10}{fc:>10}{rs:>10}"
            print(header)
            print("-" * len(header))
            
        functions = [
            calculate_pi,
            self.calculate_pi_by_threads,
            self.calculate_pi_by_processes
        ]
        if HAS_ISOLATED_INTERPRETERS:
            functions.append(self.calculate_pi_by_subinterpreters)
            
        print_header()
        reference_runtime = None
        for function in functions:
            runtime, result = self.get_runtime(function)
            if reference_runtime is None:
                reference_runtime = runtime
            factor = runtime / reference_runtime
            print(f"{function.__name__:32}{runtime:10.5f}{factor:10.2f}{result:10.4f}")
        print("\n")  # nicer output to stdout
    
    
def main(args):
    runs = int(args.runs)
    tasks = args.tasks
    print()
    print(f"Running version     : {sys.version}")
    print(f"Running script      : {sys.argv[0]}")
    print(f"Number of iterations: {runs:<6.3g}")
    print(f"Concurrency level   : {tasks}\n")
    print("create random numbers ... ", end="", flush=True)
    start = time.perf_counter()
    runner = ConcurrencyRunner(runs=runs, tasks=tasks)
    stop = time.perf_counter()
    print(f"duration: {stop-start:6.2g} sec\n")
    runner.run()
    
    
def parse():
    parser = argparse.ArgumentParser()
    parser.add_argument("-r", dest="runs", default=1e6, type=float, help="number of runs")
    parser.add_argument("-t", dest="tasks", default=1, type=int, help="number of concurrent tasks")
    args = parser.parse_args()
    return args
    
    
if __name__ == "__main__":
    main(parse())