Method run_scenario

benchmarks/frameworks/common.py:256–297

Run a specific benchmark scenario.

(self, scenario: BenchmarkScenario)

        pass

    async def run_scenario(self, scenario: BenchmarkScenario) -> BenchmarkMetrics:
        """Run a specific benchmark scenario."""
        setup_start = time.perf_counter()
        await self.setup()
        setup_time = (time.perf_counter() - setup_start) * 1000

        metrics = None
        try:
            if scenario == BenchmarkScenario.SIMPLE_TASK:
                metrics = await self.run_simple_task()
            elif scenario == BenchmarkScenario.SEQUENTIAL_PIPELINE:
                metrics = await self.run_sequential_pipeline()
            elif scenario == BenchmarkScenario.PARALLEL_PIPELINE:
                metrics = await self.run_parallel_pipeline()
            elif scenario == BenchmarkScenario.COMPLEX_WORKFLOW:
                metrics = await self.run_complex_workflow()
            elif scenario == BenchmarkScenario.MEMORY_INTENSIVE:
                metrics = await self.run_memory_intensive()
            elif scenario == BenchmarkScenario.CONCURRENT_TASKS:
                metrics = await self.run_concurrent_tasks()
            else:
                raise ValueError(f"Unknown scenario: {scenario}")

            if metrics is not None:
                metrics.setup_time_ms = setup_time

        except Exception as e:
            # Create failure metrics if scenario fails
            metrics = BenchmarkMetrics()
            metrics.error_rate = 1.0
            metrics.setup_time_ms = setup_time
            # Re-raise the exception so the caller can handle it
            raise e

        finally:
            teardown_start = time.perf_counter()
            await self.teardown()
            teardown_time = (time.perf_counter() - teardown_start) * 1000
            if metrics is not None:
                metrics.teardown_time_ms = teardown_time

        return metrics


# Common test scenarios data