123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402 |
- #!/usr/bin/env python3
- # Copyright 2018 The Chromium Authors. All rights reserved.
- # Use of this source code is governed by a BSD-style license that can be
- # found in the LICENSE file.
- """generate_perf_report.py is to be used after comparative_tester.py has been
- executed and written some test data into the location specified by
- target_spec.py. It writes to results_dir and reads all present test info from
- raw_data_dir. Using this script should just be a matter of invoking it from
- chromium/src while raw test data exists in raw_data_dir."""
- import json
- import logging
- import math
- import os
- import sys
- from typing import List, Dict, Set, Tuple, Optional, Any, TypeVar, Callable
- import target_spec
- from test_results import (TargetResult, ReadTargetFromJson, TestResult,
- ResultLine)
- class LineStats(object):
- def __init__(self, desc: str, unit: str, time_avg: float, time_dev: float,
- cv: float, samples: int) -> None:
- """A corpus of stats about a particular line from a given test's output.
- Args:
- desc (str): Descriptive text of the line in question.
- unit (str): The units of measure that the line's result is in.
- time_avg (float): The average measurement.
- time_dev (float): The standard deviation of the measurement.
- cv (float): The coefficient of variance of the measure.
- samples (int): The number of samples that went into making this object.
- """
- self.desc = desc
- self.time_avg = time_avg
- self.time_dev = time_dev
- self.cv = cv
- self.unit = unit
- self.sample_num = samples
- def ToString(self) -> str:
- """Converts the line to a human-readable string."""
- if self.sample_num > 1:
- return "{}: {:.5f} σ={:.5f} {} with n={} cv={}".format(
- self.desc, self.time_avg, self.time_dev, self.unit, self.sample_num,
- self.cv)
- else:
- return "{}: {:.5f} with only one sample".format(self.desc, self.time_avg)
- def LineFromList(lines: List[ResultLine]) -> LineStats:
- """Takes a list of ResultLines and generates statistics for them.
- Args:
- lines (List[ResultLine]): The list of lines to generate stats for.
- Returns:
- LineStats: the representation of statistical data for the lines.
- """
- desc = lines[0].desc
- unit = lines[0].unit
- times = [line.meas for line in lines]
- avg, dev, cv = GenStats(times)
- return LineStats(desc, unit, avg, dev, cv, len(lines))
- class TestStats(object):
- def __init__(self, name: str, time_avg: float, time_dev: float, cv: float,
- samples: int, lines: List[LineStats]) -> None:
- """Represents a summary of relevant statistics for a list of tests.
- Args:
- name (str): The name of the test whose runs are being averaged.
- time_avg (float): The average time to execute the test.
- time_dev (float): The standard deviation in the mean.
- cv (float): The coefficient of variance of the population.
- samples (int): The number of samples in the population
- lines (List[LineStats]): The averaged list of all the lines of output that
- comprises this test.
- """
- self.name = name
- self.time_avg = time_avg
- self.time_dev = time_dev
- self.cv = cv
- self.sample_num = samples
- self.lines = lines
- def ToLines(self) -> List[str]:
- """The stats of this test, as well as its constituent LineStats, in a human-
- readable format.
- Returns:
- List[str]: The human-readable list of lines.
- """
- lines = []
- if self.sample_num > 1:
- lines.append("{}: {:.5f} σ={:.5f}ms with n={} cv={}".format(
- self.name, self.time_avg, self.time_dev, self.sample_num, self.cv))
- else:
- lines.append("{}: {:.5f} with only one sample".format(
- self.name, self.time_avg))
- for line in self.lines:
- lines.append(" {}".format(line.ToString()))
- return lines
- def TestFromList(tests: List[TestResult]) -> TestStats:
- """Coalesces a list of TestResults into a single TestStats object.
- Args:
- tests (List[TestResult]): The input sample of the tests.
- Returns:
- TestStats: A representation of the statistics of the tests.
- """
- name = tests[0].name
- avg, dev, cv = GenStats([test.time for test in tests])
- lines = {} # type: Dict[str, List[ResultLine]]
- for test in tests:
- assert test.name == name
- for line in test.lines:
- if not line.desc in lines:
- lines[line.desc] = [line]
- else:
- lines[line.desc].append(line)
- test_lines = []
- for _, line_list in lines.items():
- stat_line = LineFromList(line_list)
- if stat_line:
- test_lines.append(stat_line)
- return TestStats(name, avg, dev, cv, len(tests), test_lines)
- class TargetStats(object):
- def __init__(self, name: str, samples: int, tests: List[TestStats]) -> None:
- """A representation of the actual target that was built and run on the
- platforms multiple times to generate statistical data.
- Args:
- name (str): The name of the target that was built and run.
- samples (int): The number of times the tests were run.
- tests (List[TestStats]): The statistics of tests included in the target.
- """
- self.name = name
- self.sample_num = samples
- self.tests = tests
- def ToLines(self) -> List[str]:
- """Converts the entire target into a list of lines in human-readable format.
- Returns:
- List[str]: The human-readable test lines.
- """
- lines = []
- if self.sample_num > 1:
- lines.append("{}: ".format(self.name))
- else:
- lines.append("{}: with only one sample".format(self.name))
- for test in self.tests:
- for line in test.ToLines():
- lines.append(" {}".format(line))
- return lines
- def __format__(self, format_spec):
- return "\n".join(self.ToLines())
- def TargetFromList(results: List[TargetResult]) -> TargetStats:
- """Coalesces a list of TargetResults into a single collection of stats.
- Args:
- results (List[TargetResult]): The sampling of target executions to generate
- stats for.
- Returns:
- TargetStats: The body of stats for the sample given.
- """
- name = results[0].name
- sample_num = len(results)
- tests = {} # type: Dict[str, List[TestResult]]
- for result in results:
- assert result.name == name
- # This groups tests by name so that they can be considered independently,
- # so that in the event tests flake out, their average times can
- # still be accurately calculated
- for test in result.tests:
- if not test.name in tests.keys():
- tests[test.name] = [test]
- tests[test.name].append(test)
- test_stats = [TestFromList(test_list) for _, test_list in tests.items()]
- return TargetStats(name, sample_num, test_stats)
- def GenStats(corpus: List[float]) -> Tuple[float, float, float]:
- """Generates statistics from a list of values
- Args:
- corpus (List[float]): The set of data to generate statistics for.
- Returns:
- Tuple[float, float, float]: The mean, standard deviation, and coefficient of
- variation for the given sample data.
- """
- avg = sum(corpus) / len(corpus)
- adjusted_sum = 0.0
- for item in corpus:
- adjusted = item - avg
- adjusted_sum += adjusted * adjusted
- dev = math.sqrt(adjusted_sum / len(corpus))
- cv = dev / avg
- return avg, dev, cv
- def DirectoryStats(directory: str) -> List[TargetStats]:
- """Takes a path to directory, and uses JSON files in that directory to compile
- a list of statistical objects for each independent test target it can detect
- in the directory.
- Args:
- directory (str): The directory to scan for relevant JSONs
- Returns:
- List[TargetStats]: Each element in this list is one target, averaged up over
- all of its executions.
- """
- resultMap = {} # type: Dict[str, List[TargetResult]]
- for file in os.listdir(directory):
- results = ReadTargetFromJson("{}/{}".format(directory, file))
- if not results.name in resultMap.keys():
- resultMap[results.name] = [results]
- else:
- resultMap[results.name].append(results)
- targets = []
- for _, resultList in resultMap.items():
- targets.append(TargetFromList(resultList))
- return targets
- def CompareTargets(linux: TargetStats, fuchsia: TargetStats) -> Dict[str, Any]:
- """Compare takes a corpus of statistics from both Fuchsia and Linux, and then
- lines up the values, compares them to each other, and writes them into a
- dictionary that can be JSONified.
- """
- if linux and fuchsia:
- assert linux.name == fuchsia.name
- paired_tests = ZipListsByPredicate(linux.tests, fuchsia.tests,
- lambda test: test.name)
- paired_tests = MapDictValues(paired_tests, CompareTests)
- return {"name": linux.name, "tests": paired_tests}
- else:
- # One of them has to be non-null, by the way ZipListsByPredicate functions
- assert linux or fuchsia
- if linux:
- logging.error("Fuchsia was missing test target {}".format(linux.name))
- else:
- logging.error("Linux was missing test target {}".format(fuchsia.name))
- return None
- def CompareTests(linux: TestStats, fuchsia: TestStats) -> Dict[str, Any]:
- """As CompareTargets, but at the test level"""
- if not linux and not fuchsia:
- logging.error("Two null TestStats objects were passed to CompareTests.")
- return {}
- if not linux or not fuchsia:
- if linux:
- name = linux.name
- failing_os = "Fuchsia"
- else:
- name = fuchsia.name
- failing_os = "Linux"
- logging.error("%s failed to produce output for the test %s",
- failing_os, name)
- return {}
- assert linux.name == fuchsia.name
- paired_lines = ZipListsByPredicate(linux.lines, fuchsia.lines,
- lambda line: line.desc)
- paired_lines = MapDictValues(paired_lines, CompareLines)
- result = {"lines": paired_lines, "unit": "ms"} # type: Dict[str, Any]
- if linux:
- result["name"] = linux.name
- result["linux_avg"] = linux.time_avg
- result["linux_dev"] = linux.time_dev
- result["linux_cv"] = linux.cv
- if fuchsia == None:
- logging.warning("Fuchsia is missing test case {}".format(linux.name))
- else:
- result["name"] = fuchsia.name
- result["fuchsia_avg"] = fuchsia.time_avg
- result["fuchsia_dev"] = fuchsia.time_dev
- result["fuchsia_cv"] = fuchsia.cv
- return result
- def CompareLines(linux: LineStats, fuchsia: LineStats) -> Dict[str, Any]:
- """CompareLines wraps two LineStats objects up as a JSON-dumpable dict.
- It also logs a warning every time a line is given which can't be matched up.
- If both lines passed are None, or their units or descriptions are not the same
- (which should never happen) this function fails.
- """
- if linux != None and fuchsia != None:
- assert linux.desc == fuchsia.desc
- assert linux.unit == fuchsia.unit
- assert linux != None or fuchsia != None
- # ref_test is because we don't actually care which test we get the values
- # from, as long as we get values for the name and description
- ref_test = linux if linux else fuchsia
- result = {"desc": ref_test.desc, "unit": ref_test.unit}
- if fuchsia == None:
- logging.warning("Fuchsia is missing test line {}".format(linux.desc))
- else:
- result["fuchsia_avg"] = fuchsia.time_avg
- result["fuchsia_dev"] = fuchsia.time_dev
- result["fuchsia_cv"] = fuchsia.cv
- if linux:
- result["linux_avg"] = linux.time_avg
- result["linux_dev"] = linux.time_dev
- result["linux_cv"] = linux.cv
- return result
- T = TypeVar("T")
- R = TypeVar("R")
- def ZipListsByPredicate(left: List[T], right: List[T],
- pred: Callable[[T], R]) -> Dict[R, Tuple[T, T]]:
- """This function takes two lists, and a predicate. The predicate is applied to
- the values in both lists to obtain a keying value from them. Each item is then
- inserted into the returned dictionary using the obtained key. The predicate
- should not map multiple values from one list to the same key.
- """
- paired_items = {} # type: Dict [R, Tuple[T, T]]
- for item in left:
- key = pred(item)
- # the first list shouldn't cause any key collisions
- assert key not in paired_items.keys()
- paired_items[key] = item, None
- for item in right:
- key = pred(item)
- if key in paired_items.keys():
- # elem 1 of the tuple is always None if the key exists in the map
- prev, _ = paired_items[key]
- paired_items[key] = prev, item
- else:
- paired_items[key] = None, item
- return paired_items
- U = TypeVar("U")
- V = TypeVar("V")
- def MapDictValues(dct: Dict[T, Tuple[R, U]],
- predicate: Callable[[R, U], V]) -> Dict[T, V]:
- """This function applies the predicate to all the values in the dictionary,
- returning a new dictionary with the new values.
- """
- out_dict = {}
- for key, val in dct.items():
- out_dict[key] = predicate(*val)
- return out_dict
- def main():
- linux_avgs = DirectoryStats(target_spec.raw_linux_dir)
- fuchsia_avgs = DirectoryStats(target_spec.raw_fuchsia_dir)
- paired_targets = ZipListsByPredicate(linux_avgs, fuchsia_avgs,
- lambda target: target.name)
- for name, targets in paired_targets.items():
- comparison_dict = CompareTargets(*targets)
- if comparison_dict:
- with open("{}/{}.json".format(target_spec.results_dir, name),
- "w") as outfile:
- json.dump(comparison_dict, outfile, indent=2)
- if __name__ == "__main__":
- sys.exit(main())
|