# SPDX-License-Identifier: GPL-2.0 import configparser import errno import os import sys import glob import argparse import platform import tempfile import logging import re import shutil import subprocess log = logging.getLogger('test') def data_equal(a, b): # Allow multiple values in assignment separated by '|' a_list = a.split('|') b_list = b.split('|') for a_item in a_list: for b_item in b_list: if (a_item == b_item): return True elif (a_item == '*') or (b_item == '*'): return True return False CORE_PMU_AVAILABLE = None CORE_PMU_COUNT = None def core_pmu_available(): """ Is there a core PMU on this machine? Without one hardware events aren't supported and get replaced by the software fallback, so there's no way to check the expectations that pin their legacy encodings. The answer can't change while the tests run, so probe the binding only once. """ core_pmu_probe() return CORE_PMU_AVAILABLE def core_pmu_probe(): """ Ask perf itself, through its python binding, to parse and open 'cycles' once: a parse failure or an ENOENT means no PMU can count it, any other error means we can't tell whether there is one, and the number of events it expanded to is the number of core PMUs, see parse_events_multi_pmu_add(), which is what core_pmu_count() asks for. """ global CORE_PMU_AVAILABLE, CORE_PMU_COUNT if CORE_PMU_AVAILABLE is not None: return try: import perf parse_events = perf.parse_events except (ImportError, AttributeError): # Nothing to ask about it, let the expectations speak. CORE_PMU_AVAILABLE = True return try: evlist = parse_events('cycles') except OSError: # No PMU that can count it, the same as the ENOENT below. CORE_PMU_AVAILABLE = False return CORE_PMU_COUNT = len(evlist) try: evlist.open() except OSError as err: # Anything else is a permission problem or alike, not our business here. CORE_PMU_AVAILABLE = err.errno != errno.ENOENT return evlist.close() CORE_PMU_AVAILABLE = True PERF_TYPE_HARDWARE = 0 PERF_TYPE_HW_CACHE = 3 PERF_SAMPLE_ID = 1 << 6 PERF_SAMPLE_IDENTIFIER = 1 << 16 PERF_TYPE_SOFTWARE = 1 PERF_COUNT_SW_DUMMY = 9 def is_dummy_tracking_event(event): """ Is this the dummy event perf opens to carry the tracking events (mmap, comm, task) when the requested ones can't, like when there is more than one core PMU and each core PMU's events run on only part of the CPUs? """ return (event.get('type') == str(PERF_TYPE_SOFTWARE) and event.get('config') == str(PERF_COUNT_SW_DUMMY)) # With more than one core PMU the tracking terms go to the dummy tracking event, # leaving the requested events without them, so they can't be matched there. TRACKING_TERMS = ['mmap', 'comm', 'task', 'mmap_data'] # The legacy event types, where perf may qualify 'config' with the core PMU # type in its top 32 bits when there is more than one core PMU. LEGACY_TYPES = (str(PERF_TYPE_HARDWARE), str(PERF_TYPE_HW_CACHE)) def mask_bits(value, bits): """ Take the bits in 'bits' out of each of the '|'-separated alternatives in 'value', keeping the rest of the expression, wildcards included. """ masked = [] for alternative in value.split('|'): try: alternative = str(int(alternative, 0) & ~bits) except ValueError: pass masked.append(alternative) return '|'.join(masked) def hide_core_pmu_type(value): """ The kernel can be told which core PMU a legacy type event is meant for by putting its type in the top 32 bits of 'config', which perf only does when there is more than one core PMU, so take those bits out to keep the expectations written for the single core PMU case matching on heterogeneous machines. """ return mask_bits(value, 0xffffffff00000000) # The two sample id bits mean the same thing here: how the samples of the # events in the evlist get associated with the event that produced them. SAMPLE_ID_BITS = PERF_SAMPLE_ID | PERF_SAMPLE_IDENTIFIER # No real sample_type bit is this high, so it can stand for "either of the # sample id bits" when comparing the two sides. SAMPLE_ID_EITHER = 1 << 40 PERF_FORMAT_GROUP = 1 << 3 def core_pmu_count(): """ How many core PMUs does perf see? An event with no PMU named in it becomes one event per core PMU, see parse_events_multi_pmu_add(), and that is the expansion that leaves the clones for the other core PMUs without any group members and that makes perf qualify 'config' with the type of the PMU it is for, so this is the number of events perf makes of 'cycles', from the probe core_pmu_available() shares. """ core_pmu_probe() return CORE_PMU_COUNT def single_core_pmu(): """ Does perf see only one core PMU? Asked for by the comparisons that need it, not at import time, as a run may never get to one. """ return core_pmu_count() == 1 def hide_group_format(expected, recorded): """ read_format's PERF_FORMAT_GROUP bit asks for the group members to be read along with the leader, which aside from this bit is verified with the fd/group_fd based group checking, and with more than one core PMU a group whose members are pinned to one of them, say 'default_core', leaves the clones for the other core PMUs without any members, so that bit isn't there: whether it is there is a consequence of how the evlist got expanded to more than one core PMU, not of what the expectations are about, so take it out on both sides. On a machine with a single core PMU there is nothing to expand, so the expectations keep verifying it. """ if single_core_pmu(): return expected, recorded return (mask_bits(expected, PERF_FORMAT_GROUP), mask_bits(recorded, PERF_FORMAT_GROUP)) def hide_sample_id(expected, recorded): """ perf asks for a way to tell the samples apart, PERF_SAMPLE_IDENTIFIER where available, PERF_SAMPLE_ID otherwise, see evlist__config(): whenever the evlist has more than one event, and when asked to outright, with --sample-identifier, by an auxtrace session, or by record__config() for a pipe run, none of which depends on the sample types, as those only choose between the two bits: with more than one core PMU the dummy tracking events make the sample types differ, so IDENTIFIER is asked for even for events whose expectations ask for ID, written on a machine with a single core PMU. So treat the two as the same thing: when the expectation asks for either of them, ask for "an id of some sort" on both sides, and when the expectation doesn't ask for any, take the bits out of what got recorded. """ try: asked = int(expected, 0) & SAMPLE_ID_BITS except ValueError: # The expectation is a wildcard or a set of alternatives. return expected, recorded try: recorded_id_bits = int(recorded, 0) & SAMPLE_ID_BITS except ValueError: # The recorded side is a wildcard or a set of alternatives. return expected, mask_bits(recorded, SAMPLE_ID_BITS) if not asked: if single_core_pmu(): return expected, recorded return expected, mask_bits(recorded, SAMPLE_ID_BITS) return (str(int(expected, 0) & ~SAMPLE_ID_BITS | SAMPLE_ID_EITHER), str(int(recorded, 0) & ~SAMPLE_ID_BITS | (SAMPLE_ID_EITHER if recorded_id_bits else 0))) class Fail(Exception): def __init__(self, test, msg): self.msg = msg self.test = test def getMsg(self): return '\'%s\' - %s' % (self.test.path, self.msg) class Notest(Exception): def __init__(self, test, reason): self.reason = reason self.test = test def getMsg(self): return '[%s] \'%s\'' % (self.reason, self.test.path) class Unsup(Exception): def __init__(self, test, ret, expected): self.test = test self.ret = ret self.expected = expected def getMsg(self): return '\'%s\' exit %d, expected %s, no attrs recorded to verify' % ( self.test.path, self.ret, self.expected) class Event(dict): terms = [ 'cpu', 'flags', 'type', 'size', 'config', 'sample_period', 'sample_type', 'read_format', 'disabled', 'inherit', 'pinned', 'exclusive', 'exclude_user', 'exclude_kernel', 'exclude_hv', 'exclude_idle', 'mmap', 'comm', 'freq', 'inherit_stat', 'enable_on_exec', 'task', 'watermark', 'precise_ip', 'mmap_data', 'sample_id_all', 'exclude_host', 'exclude_guest', 'exclude_callchain_kernel', 'exclude_callchain_user', 'wakeup_events', 'bp_type', 'config1', 'config2', 'branch_sample_type', 'sample_regs_user', 'sample_stack_user', ] def add(self, data): for key, val in data: log.debug(" %s = %s", key, val) self[key] = val def __init__(self, name, data, base): log.debug(" Event %s", name) self.name = name self.group = '' self.add(base) self.add(data) def hide_core_pmu_type_terms(self, other): """ Is this a legacy type event? Then the core PMU type that perf may have put in the top 32 bits of 'config' to tell the kernel which core PMU it wants isn't part of what we asked for, see hide_core_pmu_type(). """ for event in (self, other): for type in event.get('type', '').split('|'): if type in LEGACY_TYPES: return True return False def hide_core_pmu_type(self, term, value, other): if term == 'config' and self.hide_core_pmu_type_terms(other): return hide_core_pmu_type(value) return value def hide_event_count_dependent_terms(self, term, expected, recorded): if term == 'sample_type': return hide_sample_id(expected, recorded) if term == 'read_format': return hide_group_format(expected, recorded) return expected, recorded def recorded_side(self, other): """ Which one is what perf recorded? Both directions get compared, see Test.compare(), so this can't be taken from the arguments' order. """ if getattr(other, 'recorded', False): return other return self if getattr(self, 'recorded', False) else None def hide_tracking_via_dummy(self, term, other): """ The tracking terms are set on the tracking event, which is the dummy event when there is more than one core PMU, leaving the requested events without them, nothing to match on then. """ if term not in TRACKING_TERMS: return False for event in (self, other): if term in getattr(event, 'tracking_moved_to_dummy', ()): return True return False def equal(self, other): for t in Event.terms: log.debug(" [%s] %s %s", t, self[t], other[t]) if t not in self or t not in other: return False if self.hide_tracking_via_dummy(t, other): continue recorded = self.recorded_side(other) if recorded is self: expected, got_event = other, self else: expected, got_event = self, other exp = self.hide_core_pmu_type(t, expected[t], other) got = self.hide_core_pmu_type(t, got_event[t], other) if recorded is not None: exp, got = self.hide_event_count_dependent_terms(t, exp, got) if not data_equal(exp, got): return False return True def optional(self): if 'optional' in self and self['optional'] == '1': return True return False def diff(self, other): for t in Event.terms: if t not in self or t not in other: continue if self.hide_tracking_via_dummy(t, other): continue recorded = self.recorded_side(other) if recorded is self: expected, got_event = other, self else: expected, got_event = self, other exp = self.hide_core_pmu_type(t, expected[t], other) got = self.hide_core_pmu_type(t, got_event[t], other) if recorded is not None: exp, got = self.hide_event_count_dependent_terms(t, exp, got) if not data_equal(exp, got): log.warning("expected %s=%s, got %s", t, expected[t], got_event[t]) def parse_version(version): if not version: return None return [int(v) for v in version.split(".")[0:2]] def format_version(version): return ".".join(str(v) for v in version) # Test file description needs to have following sections: # [config] # - just single instance in file # - needs to specify: # 'command' - perf command name # 'args' - special command arguments # 'ret' - Skip test if Perf doesn't exit with this value (0 by default) # 'test_ret'- If set to 'true', fail test instead of skipping for 'ret' argument # 'arch' - architecture specific test (optional) # comma separated list, ! at the beginning # negates it. # 'requires' - Comma separated list of things the test needs, it is # skipped when any of them isn't there, currently: # core-pmu - the machine can count hardware events. # 'auxv' - Truthy statement that is evaled in the scope of the auxv map. When false, # the test is skipped. For example 'auxv["AT_HWCAP"] == 10'. (optional) # 'kernel_since' - Inclusive kernel version from which the test will start running. Only the # first two values are supported, for example "6.1" (optional) # 'kernel_until' - Exclusive kernel version from which the test will stop running. (optional) # [eventX:base] # - one or multiple instances in file # - expected values assignments class Test(object): def __init__(self, path, options): parser = configparser.ConfigParser() parser.read(path) log.warning("running '%s'", path) self.path = path self.test_dir = options.test_dir self.perf = options.perf self.command = parser.get('config', 'command') self.args = parser.get('config', 'args') try: self.ret = parser.get('config', 'ret') except Exception: self.ret = 0 self.test_ret = parser.getboolean('config', 'test_ret', fallback=False) try: self.arch = parser.get('config', 'arch') log.warning("test limitation '%s'", self.arch) except Exception: self.arch = '' self.auxv = parser.get('config', 'auxv', fallback=None) self.requires = parser.get('config', 'requires', fallback='') self.kernel_since = parse_version(parser.get('config', 'kernel_since', fallback=None)) self.kernel_until = parse_version(parser.get('config', 'kernel_until', fallback=None)) self.expect = {} self.result = {} log.debug(" loading expected events") self.load_events(path, self.expect) def is_event(self, name): if name.find("event") == -1: return False else: return True def skip_test_kernel_since(self): if not self.kernel_since: return None if self.kernel_since <= parse_version(platform.release()): return None return "kernel older than %s" % format_version(self.kernel_since) def skip_test_kernel_until(self): if not self.kernel_until: return None if parse_version(platform.release()) < self.kernel_until: return None return "kernel newer than %s" % format_version(self.kernel_until) def skip_test_auxv(self): def new_auxv(a, pattern): items = list(filter(None, pattern.split(a))) # AT_HWCAP is hex but doesn't have a prefix, so special case it if items[0] == "AT_HWCAP": value = int(items[-1], 16) else: try: value = int(items[-1], 0) except ValueError: value = items[-1] return (items[0], value) if not self.auxv: return None auxv = subprocess.check_output("LD_SHOW_AUXV=1 sleep 0", shell=True) \ .decode(sys.stdout.encoding) pattern = re.compile(r"[: ]+") auxv = dict([new_auxv(a, pattern) for a in auxv.splitlines()]) if eval(self.auxv): return None return "the auxv condition doesn't hold here" def skip_test_arch(self, myarch): # If architecture not set always run test if self.arch == '': return None # Allow multiple values in assignment separated by ',' arch_list = self.arch.split(',') # Handle negated list such as !s390x,ppc if arch_list[0][0] == '!': arch_list[0] = arch_list[0][1:] log.warning("excluded architecture list %s", arch_list) for arch_item in arch_list: if arch_item == myarch: return "arch = %s, machine = %s" % (self.arch, myarch) return None for arch_item in arch_list: if arch_item == myarch: return None return "arch = %s, machine = %s" % (self.arch, myarch) def skip_test_requires(self): for requirement in filter(None, map(str.strip, self.requires.split(','))): if requirement == 'core-pmu': if not core_pmu_available(): log.warning("test limitation 'requires %s'", requirement) return "needs a core PMU, none here" else: log.warning("unknown requirement '%s', ignored", requirement) return None def restore_sample_rate(self, value=10000): try: # Check value of sample_rate with open("/proc/sys/kernel/perf_event_max_sample_rate", "r", encoding="utf-8") as fIn: curr_value = fIn.readline() # If too low restore to reasonable value if not curr_value or int(curr_value) < int(value): with open("/proc/sys/kernel/perf_event_max_sample_rate", "w", encoding="utf-8") as fOut: fOut.write(str(value)) except IOError as e: log.warning("couldn't restore sample_rate value: I/O error %s", e) except ValueError as e: log.warning("couldn't restore sample_rate value: Value error %s", e) except TypeError as e: log.warning("couldn't restore sample_rate value: Type error %s", e) def load_events(self, path, events): parser_event = configparser.ConfigParser() parser_event.read(path) # The event record section header contains 'event' word, # optionaly followed by ':' allowing to load 'parent # event' first as a base for section in filter(self.is_event, parser_event.sections()): parser_items = parser_event.items(section) base_items = {} # Read parent event if there's any if (':' in section): base = section[section.index(':') + 1:] parser_base = configparser.ConfigParser() parser_base.read(self.test_dir + '/' + base) base_items = parser_base.items('event') e = Event(section, parser_items, base_items) events[section] = e def run_cmd(self, tempdir): _junk1, _junk2, _junk3, _junk4, myarch = (os.uname()) arch = self.skip_test_arch(myarch) if arch: log.debug(" detail: the test file has arch = %s, the machine reports %s", self.arch, myarch) raise Notest(self, arch) auxv = self.skip_test_auxv() if auxv: log.debug(" detail: the auxv condition %s doesn't hold, the bits come " "from the LD_SHOW_AUXV=1 output", self.auxv) raise Notest(self, auxv) kernel = self.skip_test_kernel_since() if kernel: log.debug(" detail: the test file's kernel_since = %s, this kernel is %s", format_version(self.kernel_since), platform.release()) raise Notest(self, kernel) kernel = self.skip_test_kernel_until() if kernel: log.debug(" detail: the test file's kernel_until = %s, this kernel is %s", format_version(self.kernel_until), platform.release()) raise Notest(self, kernel) requirement = self.skip_test_requires() if requirement: log.debug(" detail: requires core-pmu, nothing countable here, check " "kernel.perf_event_paranoid and the PMU nodes under " "/sys/bus/event_source/") raise Notest(self, requirement) self.restore_sample_rate() cmd = "PERF_TEST_ATTR=%s %s %s -o %s/perf.data %s" % (tempdir, self.perf, self.command, tempdir, self.args) ret = os.WEXITSTATUS(os.system(cmd)) log.info(" '%s' ret '%s', expected '%s'", cmd, str(ret), str(self.ret)) if not data_equal(str(ret), str(self.ret)): plain_cmd = "%s %s -o /dev/null %s" % ( self.perf, self.command, self.args.split('>/dev/null')[0].rstrip()) if self.test_ret: raise Fail(self, "Perf exit code failure") else: log.debug(" detail: '%s' exited with %d, the expectations want the " "workload's %s exit, so there are no attrs recorded to " "verify; this is normally a capability problem, run '%s' " "to see why", plain_cmd, ret, self.ret, plain_cmd) raise Unsup(self, ret, str(self.ret)) def drop_dummy_tracking_events(self): """ Leave out the dummy event perf itself opened, see is_dummy_tracking_event(), unless the expectations ask for one. """ if any(is_dummy_tracking_event(e) for e in self.expect.values()): return if not any(is_dummy_tracking_event(e) for e in self.result.values()): return moved = set() for name, event in list(self.result.items()): if is_dummy_tracking_event(event): log.debug(" dropping dummy tracking event [%s]", name) moved.update(t for t in TRACKING_TERMS if event.get(t) == '1') del self.result[name] # The tracking terms the dropped dummies carried, flag the remaining # events so that the match leaves only those out. for event in self.result.values(): event.tracking_moved_to_dummy = moved def drop_other_core_pmu_events(self, devices='/sys/bus/event_source/devices'): """ Leave out the events of the other core PMUs, which have their own encodings, when the expectations name one of the core PMUs. """ asked = set() for event in self.expect.values(): types = event.get('type', '').split('|') if '*' in types: return asked.update(types) try: names = os.listdir(devices) except OSError: return core = set() for name in names: path = os.path.join(devices, name) has_cpus = os.path.exists(os.path.join(path, 'cpus')) if name not in ('cpu', 'cpum_cf') and not has_cpus: continue try: with open(os.path.join(path, 'type'), encoding='utf-8') as f: core.add(f.read().strip()) except OSError: pass if not core & asked: return others = core - asked for name, event in list(self.result.items()): if event.get('type') in others: log.debug(" dropping other core PMU event [%s]", name) del self.result[name] def compare(self, expect, result, strict_groups=True): match = {} log.debug(" compare") # For each expected event find all matching # events in result. Fail if there's not any. for exp_name, exp_event in expect.items(): exp_list = [] res_event = {} log.debug(" matching [%s]", exp_name) for res_name, res_event in result.items(): log.debug(" to [%s]", res_name) if (exp_event.equal(res_event)): exp_list.append(res_name) log.debug(" ->OK") else: log.debug(" ->FAIL") log.debug(" match: [%s] matches %s", exp_name, str(exp_list)) # we did not any matching event - fail if not exp_list: if exp_event.optional(): log.debug(" %s does not match, but is optional", exp_name) else: if not res_event: log.debug(" res_event is empty") else: exp_event.diff(res_event) raise Fail(self, 'match failure') match[exp_name] = exp_list # Check that each expected group is matched in the result; an event # opened in a group may match an expectation that doesn't declare # one, as those only verify the event attributes. for exp_name, exp_event in expect.items(): group = exp_event.group if (group == ''): continue for res_name in match[exp_name]: res_group = result[res_name].group if res_group == '' and not strict_groups: continue if res_group not in match[group]: raise Fail(self, 'group failure') log.debug(" group: [%s] matches group leader %s", exp_name, str(match[group])) log.debug(" matched") def resolve_groups(self, events): for name, event in events.items(): group_fd = event['group_fd'] if group_fd == '-1': continue for iname, ievent in events.items(): if (ievent['fd'] == group_fd): event.group = iname log.debug('[%s] has group leader [%s]', name, iname) break def run(self): tempdir = tempfile.mkdtemp() try: # run the test script self.run_cmd(tempdir) # load events expectation for the test log.debug(" loading result events") for f in glob.glob(tempdir + '/event*'): self.load_events(f, self.result) for event in self.result.values(): event.recorded = True self.drop_dummy_tracking_events() self.drop_other_core_pmu_events() # resolve group_fd to event names self.resolve_groups(self.expect) self.resolve_groups(self.result) # do the expectation - results matching - both ways self.compare(self.expect, self.result) self.compare(self.result, self.expect, strict_groups=False) finally: # cleanup shutil.rmtree(tempdir) def run_tests(options): for f in glob.glob(options.test_dir + '/' + options.test): try: Test(f, options).run() except Unsup as obj: log.warning("unsupp %s", obj.getMsg()) except Notest as obj: log.warning("skipped %s", obj.getMsg()) def setup_log(verbose): global log level = logging.CRITICAL if verbose == 1: level = logging.WARNING if verbose == 2: level = logging.INFO if verbose >= 3: level = logging.DEBUG log = logging.getLogger('test') log.setLevel(level) ch = logging.StreamHandler() ch.setLevel(level) formatter = logging.Formatter('%(message)s') ch.setFormatter(formatter) log.addHandler(ch) def main(): parser = argparse.ArgumentParser() parser.add_argument("-t", "--test", action="store", dest="test", help="single test") parser.add_argument("-d", "--test-dir", action="store", dest="test_dir", help="tests dir") parser.add_argument("-p", "--perf", action="store", dest="perf", help="perf binary") parser.add_argument("-v", "--verbose", default=0, action="count", dest="verbose", help="verbose level") options = parser.parse_args() setup_log(options.verbose) if not options.test_dir: print('FAILED no -d option specified') sys.exit(-1) if not options.test: options.test = 'test*' try: run_tests(options) except Fail as obj: print("FAILED %s" % obj.getMsg()) sys.exit(-1) sys.exit(0) if __name__ == '__main__': main()