• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2023 The Chromium Authors
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5from __future__ import annotations
6
7import abc
8import datetime as dt
9import logging
10from typing import TYPE_CHECKING, Any, Dict, Optional, Sequence, Tuple
11
12from crossbench import helper
13from crossbench.benchmarks.base import Benchmark
14from crossbench.cli.ui import timer
15from crossbench.parse import DurationParser
16from crossbench.stories.story import Story
17
18if TYPE_CHECKING:
19  import argparse
20
21  from crossbench.cli.parser import CrossBenchArgumentParser
22  from crossbench.runner.run import Run
23
24
25class ManualStory(Story, metaclass=abc.ABCMeta):
26
27  STORY_NAME = "manual"
28
29  def __init__(self, start_after: Optional[dt.timedelta],
30               run_for: Optional[dt.timedelta]):
31    self._start_after = start_after
32    self._run_for = run_for
33    duration = ((start_after or dt.timedelta()) +
34                (run_for or dt.timedelta(seconds=30)))
35    super().__init__(self.STORY_NAME, duration)
36
37  def setup(self, run: Run) -> None:
38    if self._start_after is None:
39      logging.info("-" * 80)
40      logging.critical("Press enter to start:")
41      input()
42    elif self._start_after.total_seconds():
43      logging.critical("-" * 80)
44      logging.critical(
45          "The browser has launched. Measurement will start in %s" +
46          " (or press enter to start immediately)", self._start_after)
47      helper.input_with_timeout(timeout=self._start_after)
48    logging.info("Starting Manual Benchmark...")
49
50  def run(self, run: Run) -> None:
51    with timer():
52      logging.info("-" * 80)
53      self._wait_for_input()
54      # Empty line to preserve timer output.
55      print()
56      logging.info("Stopping Manual Benchmark...")
57
58  def _wait_for_input(self) -> None:
59    if self._run_for is None:
60      logging.critical("Press enter to stop:")
61      try:
62        input()
63      except KeyboardInterrupt:
64        pass
65    else:
66      logging.critical(
67          "Measurement has started. The browser will close in %s" +
68          " (or press enter to close immediately)", self._run_for)
69      helper.input_with_timeout(timeout=self._run_for)
70
71
72  @classmethod
73  def all_story_names(cls) -> Tuple[str, ...]:
74    return (ManualStory.STORY_NAME,)
75
76
77class ManualBenchmark(Benchmark, metaclass=abc.ABCMeta):
78  """
79  Benchmark runner for the manual mode.
80
81  Just launches the browser and lets the user perform the desired interactions.
82  Optionally waits for |start_after| seconds, then runs measurements for
83  |run_for| seconds, then closes the browser.
84  """
85  NAME = "manual"
86  DEFAULT_STORY_CLS = ManualStory
87
88  def __init__(self, start_after: Optional[dt.timedelta],
89               run_for: Optional[dt.timedelta]) -> None:
90    super().__init__([ManualStory(start_after=start_after, run_for=run_for)])
91
92  @classmethod
93  def add_cli_parser(
94      cls, subparsers: argparse.ArgumentParser, aliases: Sequence[str] = ()
95  ) -> CrossBenchArgumentParser:
96    parser = super().add_cli_parser(subparsers, aliases)
97    parser.add_argument(
98        "--start-after",
99        help="How long to wait until measurement starts",
100        required=False,
101        type=DurationParser.positive_or_zero_duration)
102    parser.add_argument(
103        "--run-for",
104        "--stop-after",
105        "--duration",
106        help="How long to run measurement for",
107        required=False,
108        type=DurationParser.positive_duration)
109    return parser
110
111  @classmethod
112  def kwargs_from_cli(cls, args: argparse.Namespace) -> Dict[str, Any]:
113    kwargs = super().kwargs_from_cli(args)
114    kwargs["start_after"] = args.start_after
115    kwargs["run_for"] = args.run_for
116    return kwargs
117