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 List, Optional, Sequence, Tuple, Type, TypeVar 11 12from crossbench.parse import ObjectParser 13from crossbench.runner.run import Run 14from crossbench.stories.story import Story 15 16PressBenchmarkStoryT = TypeVar( 17 "PressBenchmarkStoryT", bound="PressBenchmarkStory") 18 19 20class PressBenchmarkStory(Story, metaclass=abc.ABCMeta): 21 NAME: str = "" 22 URL: str = "" 23 URL_OFFICIAL: str = "" 24 URL_LOCAL: str = "" 25 SUBSTORIES: Tuple[str, ...] = () 26 27 @classmethod 28 def all_story_names(cls) -> Tuple[str, ...]: 29 assert cls.SUBSTORIES 30 return cls.SUBSTORIES 31 32 @classmethod 33 def default_story_names(cls) -> Tuple[str, ...]: 34 """Override this method to use a subset of all_story_names as default 35 selection if no story names are provided.""" 36 return cls.all_story_names() 37 38 @classmethod 39 def all(cls: Type[PressBenchmarkStoryT], 40 separate: bool = False, 41 url: Optional[str] = None, 42 **kwargs) -> List[PressBenchmarkStoryT]: 43 return cls.from_names(cls.all_story_names(), separate, url, **kwargs) 44 45 @classmethod 46 def default(cls: Type[PressBenchmarkStoryT], 47 separate: bool = False, 48 url: Optional[str] = None, 49 **kwargs) -> List[PressBenchmarkStoryT]: 50 return cls.from_names(cls.default_story_names(), separate, url, **kwargs) 51 52 @classmethod 53 def from_names(cls: Type[PressBenchmarkStoryT], 54 substories: Sequence[str], 55 separate: bool = False, 56 url: Optional[str] = None, 57 **kwargs) -> List[PressBenchmarkStoryT]: 58 if not substories: 59 raise ValueError("No substories provided") 60 if separate: 61 return [ 62 cls( # pytype: disable=not-instantiable 63 url=url, 64 substories=[substory], 65 **kwargs) for substory in substories 66 ] 67 return [ 68 cls( # pytype: disable=not-instantiable 69 url=url, 70 substories=substories, 71 **kwargs) 72 ] 73 74 def __init__(self, 75 *args, 76 substories: Sequence[str] = (), 77 duration: Optional[float] = None, 78 url: Optional[str] = None, 79 **kwargs) -> None: 80 cls = self.__class__ 81 assert self.SUBSTORIES, f"{cls}.SUBSTORIES is not set." 82 assert self.NAME is not None, f"{cls}.NAME is not set." 83 self._verify_url(self.URL, "URL") 84 self._verify_url(self.URL_OFFICIAL, "URL_OFFICIAL") 85 self._verify_url(self.URL_LOCAL, "URL_LOCAL") 86 assert substories, f"No substories provided for {cls}" 87 self._substories: Sequence[str] = substories 88 self._verify_substories() 89 kwargs["name"] = self._get_unique_name() 90 kwargs["duration"] = duration or self._get_initial_duration() 91 super().__init__(*args, **kwargs) 92 # If the _custom_url is empty, we generate a matching URL when the 93 # local file server is used. 94 self._custom_url: Optional[str] = url 95 96 def _get_unique_name(self) -> str: 97 substories_set = set(self._substories) 98 if substories_set == set(self.default_story_names()): 99 return self.NAME 100 if substories_set == set(self.all_story_names()): 101 name = "all" 102 else: 103 name = "_".join(self._substories) 104 if len(name) > 220: 105 # Crop the name and add some random hash bits 106 name = name[:220] + hex(hash(name))[2:10] 107 return name 108 109 def _get_initial_duration(self) -> dt.timedelta: 110 # Fixed delay for startup costs 111 startup_delay = dt.timedelta(seconds=2) 112 # Add some slack due to different story lengths 113 story_factor = 0.5 + 1.1 * len(self._substories) 114 return startup_delay + (story_factor * self.substory_duration) 115 116 def get_run_url(self, run: Run) -> str: 117 if self._custom_url: 118 # TODO: check that we have a live network / url host matches network host 119 return self._custom_url 120 network = run.browser_session.network 121 # Create a matching URL for a local file server. 122 if network.is_local_file_server and network.http_port: 123 return f"http://{network.host}:{network.http_port}" 124 # Return default URL in case of live network. 125 return self.url 126 127 @property 128 def substories(self) -> List[str]: 129 return list(self._substories) 130 131 @property 132 def has_default_substories(self) -> bool: 133 return tuple(self.substories) == self.default_story_names() 134 135 @property 136 def fast_duration(self) -> dt.timedelta: 137 """Expected benchmark duration on fast machines. 138 Keep this low enough to not have to wait needlessly at the end of a 139 benchmark. 140 """ 141 return self.duration / 3 142 143 @property 144 def slow_duration(self) -> dt.timedelta: 145 """Max duration that covers run-times on slow machines and/or 146 debug-mode browsers. 147 Making this number too large might cause needless wait times on broken 148 browsers/benchmarks. 149 """ 150 return dt.timedelta(seconds=15) + self.duration * 5 151 152 @property 153 @abc.abstractmethod 154 def substory_duration(self) -> dt.timedelta: 155 pass 156 157 @property 158 def url(self) -> str: 159 return self._custom_url or self.URL 160 161 def _verify_url(self, url: str, property_name: str) -> None: 162 cls = self.__class__ 163 assert url is not None, f"{cls}.{property_name} is not set." 164 165 def _verify_substories(self) -> None: 166 ObjectParser.unique_sequence(self._substories, "substories", ValueError) 167 if self._substories == self.SUBSTORIES: 168 return 169 for substory in self._substories: 170 assert substory in self.SUBSTORIES, (f"Unknown {self.NAME} substory %s" % 171 substory) 172 173 def log_run_details(self, run: Run) -> None: 174 super().log_run_details(run) 175 self.log_run_test_url(run) 176 177 def log_run_test_url(self, run: Run): 178 del run 179 logging.info("STORY PUBLIC TEST URL: %s", self.URL) 180