cp_library/perf/orchestrator.py

• View this file on GitHub
• Last update: 2025-08-24 20:27:05+09:00

Depends on

• cp_library/perf/interfaces.py

Required by

• cp_library/perf/benchmark.py
• perf/bit2.py
• perf/bit6.py
• perf/bool_list.py
• perf/bytearray_decode.py
• perf/csr.py
• perf/csr2.py
• perf/csr6.py
• perf/deque.py
• perf/edge_list.py
• perf/grid.py
• perf/heap_csr.py
• perf/heap_view.py
• perf/list2.py
• perf/list6.py
• perf/list_view.py
• perf/mlist.py
• perf/que.py
• perf/rank.py
• perf/segtree2.py
• perf/segtree6.py
• perf/stack_operations.py
• perf/view2.py
• perf/view6.py

Code

import cp_library.__header__
"""
Benchmark orchestrator implementation following Single Responsibility Principle.
"""

import sys
from typing import List
from .interfaces import (
    BenchmarkOrchestrator, BenchmarkRegistry, BenchmarkResult, 
    TimerInterface, OutputManager
)
import cp_library.perf.__header__


class BenchmarkOrchestratorImpl(BenchmarkOrchestrator):
    """Implementation of benchmark execution orchestration"""
    
    def __init__(self, 
                 registry: BenchmarkRegistry,
                 timer: TimerInterface,
                 output_manager: OutputManager = None):
        self.registry = registry
        self.timer = timer
        self.output_manager = output_manager
        self.results: List[BenchmarkResult] = []
    
    def run_benchmarks(self, operations: List[str], sizes: List[int]) -> List[BenchmarkResult]:
        """Execute benchmarks and return results"""
        self.results = []
        
        for operation in operations:
            for size in sizes:
                self._run_single(operation, size)
        
        if self.output_manager:
            self.output_manager.save_results(self.results, None)
        
        return self.results
    
    def _run_single(self, operation: str, size: int) -> None:
        """Run a single operation/size combination"""
        print(f"\nOperation: {operation}, Size: {size}")
        print("-" * 50)
        sys.stdout.flush()
        
        # Generate test data
        generator = self.registry.get_data_generator(operation)
        if not generator:
            raise ValueError(f"No data generator for operation: {operation}")
        
        test_data = generator.generate(size, operation)
        
        # Get implementations for this operation
        impls = self.registry.get_implementations(operation)
        if not impls:
            print(f"No implementations for operation: {operation}")
            return
        
        # Run reference implementation first (find first non-skipped implementation)
        expected_result = None
        ref_name = None
        for impl_name, impl_func in impls.items():
            setup_func = self.registry.get_setup(operation, impl_name)
            result, _ = self.timer.measure_time(impl_func, test_data, setup_func)
            if result is not None:
                expected_result = result
                ref_name = impl_name
                break
        
        # If all implementations return None, skip this operation/size combination
        if expected_result is None:
            print("  All implementations skipped")
            return
        
        # Get validator for this operation
        validator = self.registry.get_validator(operation)
        
        # Run all implementations
        for impl_name, impl_func in impls.items():
            try:
                setup_func = self.registry.get_setup(operation, impl_name)
                result, time_ms = self.timer.measure_time(impl_func, test_data, setup_func)
                
                # Check if implementation returned None to skip
                if result is None:
                    print(f"  {impl_name:<20} SKIPPED")
                    sys.stdout.flush()
                    continue
                
                correct = validator.validate(expected_result, result)
                
                # Store result
                benchmark_result = BenchmarkResult(
                    operation=operation,
                    size=size,
                    implementation=impl_name,
                    time_ms=time_ms,
                    correct=correct
                )
                self.results.append(benchmark_result)
                
                status = "OK" if correct else "FAIL"
                print(f"  {impl_name:<20} {time_ms:>8.3f} ms  {status}")
                sys.stdout.flush()
                
            except Exception as e:
                # Get a meaningful error message
                error_msg = str(e) if str(e) else f"{type(e).__name__}: {repr(e)}"
                
                benchmark_result = BenchmarkResult(
                    operation=operation,
                    size=size,
                    implementation=impl_name,
                    time_ms=float('inf'),
                    correct=False,
                    error=error_msg
                )
                self.results.append(benchmark_result)
                print(f"  {impl_name:<20} ERROR: {error_msg[:40]}")
                sys.stdout.flush()

Traceback (most recent call last):
  File "/opt/hostedtoolcache/PyPy/3.10.16/x64/lib/pypy3.10/site-packages/onlinejudge_verify/documentation/build.py", line 71, in _render_source_code_stat
    bundled_code = language.bundle(stat.path, basedir=basedir, options={'include_paths': [basedir]}).decode()
  File "/opt/hostedtoolcache/PyPy/3.10.16/x64/lib/pypy3.10/site-packages/onlinejudge_verify/languages/python.py", line 101, in bundle
    return bundler.update(path)
  File "/opt/hostedtoolcache/PyPy/3.10.16/x64/lib/pypy3.10/site-packages/onlinejudge_verify/languages/python_bundle.py", line 154, in update
    self.process_file(path)
  File "/opt/hostedtoolcache/PyPy/3.10.16/x64/lib/pypy3.10/site-packages/onlinejudge_verify/languages/python_bundle.py", line 24, in process_file
    self.bundled_code[file_path] = self.process_imports(tree, file_path, source, file_is_top_level)
  File "/opt/hostedtoolcache/PyPy/3.10.16/x64/lib/pypy3.10/site-packages/onlinejudge_verify/languages/python_bundle.py", line 102, in process_imports
    processor.visit(tree)
  File "/opt/hostedtoolcache/PyPy/3.10.16/x64/lib/pypy3.10/ast.py", line 418, in visit
    return visitor(node)
  File "/opt/hostedtoolcache/PyPy/3.10.16/x64/lib/pypy3.10/ast.py", line 426, in generic_visit
    self.visit(item)
  File "/opt/hostedtoolcache/PyPy/3.10.16/x64/lib/pypy3.10/ast.py", line 418, in visit
    return visitor(node)
  File "/opt/hostedtoolcache/PyPy/3.10.16/x64/lib/pypy3.10/site-packages/onlinejudge_verify/languages/python_bundle.py", line 64, in visit_ImportFrom
    raise NotImplementedError("Relative imports are not supported")
NotImplementedError: Relative imports are not supported

Back to top page