check_concurrency_ppcr.py

"""
check_concurrency_ppcr.py
ppcr: partial 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).

The random numbers are pre-calculated in the threads or processes, so
there is no expensive ipc overhead by transfering pre-calculated numbers
from the main thread.

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 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(runs):
    random_numbers = [(random(), random()) for _ in range(runs)]
    inside = 0
    for x, y in random_numbers:
        if 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
            
    def calculate_pi_by_executor(self, executor):
        with executor(max_workers=self.tasks) as e:
            runs = self.runs // self.tasks
            args = [runs] * self.tasks
            results = e.map(calculate_pi, args)  # less 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.runs)
        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")
    runner = ConcurrencyRunner(runs=runs, tasks=tasks)
    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())