tl_cpu.py

# Copyright (c) 2012-2020, Intel Corporation
# Author: Andi Kleen
#
# This program is free software; you can redistribute it and/or modify it
# under the terms and conditions of the GNU General Public License,
# version 2, as published by the Free Software Foundation.
#
# This program is distributed in the hope it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
# more details.
#
# toplev CPU detection

#
# Environment variables to override (mainly for testing):
# FORCECPU=cpu     Force CPU type (e.g. skl). Also --force-cpu
# FORCECOUNTERS=n  Force number of generic counters
# FORCEHT=0/1      Force SMT mode
# HYPERVISOR=0/1   Force hypervisor mode (also --force-hypervisor)
# CPUINFO=file     Read cpu information from file instead of /proc/cpuinfo. Also --force-cpuinfo
# REDUCED_COUNTERS=n Reduce counters by n
#

from collections import defaultdict, Counter
import os
import re
import glob
import sys
if sys.version_info.major == 3:
    import typing # noqa
    from typing import Set, List, Dict, Tuple # noqa

modelid_map = {
    (0x8e, ): "KBLR",
    (0x9e, ): "CFL",
}

cpus_8gpc = set(["icl", "tgl", "icx", "spr", "sprmax", "srf"]) # XXX handle hybrid ADL

def num_offline_cpus():
    cpus = glob.glob("/sys/devices/system/cpu/cpu[0-9]*/online")
    offline = 0
    for fn in cpus:
        with open(fn, "r") as f:
            if int(f.read()) == 0:
                offline += 1
    return offline

def reduced_counters():
    rc = os.getenv("REDUCED_COUNTERS")
    if rc:
        return int(rc)
    val = 1
    fn = "/sys/devices/cpu/allow_tsx_force_abort"
    if os.path.exists(fn):
        with open(fn, "r") as f:
            val = int(f.read())
    return 1 if val == 0 else 0

class Env(object):
    def __init__(self):
        self.forcecpu = os.getenv("FORCECPU")
        self.forcecounters = os.getenv("FORCECOUNTERS")
        self.forceht = os.getenv("FORCEHT")
        self.hypervisor = os.getenv("HYPERVISOR")
        self.cpuinfo = os.getenv("CPUINFO")

class CPU(object):
    """Detect the CPU."""
    # overrides for easy regression tests
    def force_cpu(self, known_cpus):
        force = self.env.forcecpu
        if not force:
            return False
        self.cpu = ""
        for i in known_cpus:
            if force == i[0]:
                self.cpu = i[0]
                break
        if self.cpu is None:
            print("Unknown FORCECPU ",force)
        return True

    def force_ht(self):
        ht = self.env.forceht
        if ht:
            self.ht = True if int(ht) else False
            return True
        return False

    def __init__(self, known_cpus=(), nocheck=False, env=Env()):
        self.vendor = ""
        self.env = env
        self.model = 0
        self.cpu = ""
        self.realcpu = "simple"
        self.ht = False
        self.counters = {} # type: Dict[str,int]
        self.has_tsx = False
        self.hypervisor = False
        self.force_hypervisor = False
        if self.env.hypervisor:
            self.hypervisor = True
            self.force_hypervisor = True
        self.freq = 0.0
        self.threads = 0
        forced_cpu = self.force_cpu(known_cpus)
        forced_ht = self.force_ht()
        cores = Counter() # type: typing.Counter[Tuple[int,int]]
        sockets = Counter() # type: typing.Counter[int]
        self.coreids = defaultdict(list)
        self.cputocore = {}
        self.cputothread = {}
        self.sockettocpus = defaultdict(list)
        self.cputosocket = {}
        self.allcpus = []
        self.step = 0
        self.name = ""
        cpuinfo = self.env.cpuinfo
        if cpuinfo is None:
            cpuinfo = "/proc/cpuinfo"
        with open(cpuinfo, "r") as f:
            seen = set()
            for l in f:
                n = l.split()
                if len(n) < 3:
                    continue
                if n[0] == 'processor':
                    seen.add("processor")
                    cpunum = int(n[2])
                    self.allcpus.append(cpunum)
                elif (n[0], n[2]) == ("vendor_id", "GenuineIntel"):
                    self.vendor = n[2]
                    seen.add("vendor_id")
                elif (len(n) > 3 and
                        (n[0], n[1], n[3]) == ("cpu", "family", "6")):
                    seen.add("cpu family")
                elif (n[0], n[1]) == ("model", ":"):
                    seen.add("model")
                    self.model = int(n[2])
                elif (n[0], n[1]) == ("model", "name"):
                    seen.add("model name")
                    m = re.search(r"@ (\d+\.\d+)GHz", l)
                    if m:
                        self.freq = float(m.group(1))
                    self.name = " ".join(n[3:])
                elif (n[0], n[1]) == ("physical", "id"):
                    physid = int(n[3])
                    sockets[physid] += 1
                    self.sockettocpus[physid].append(cpunum)
                    self.cputosocket[cpunum] = physid
                elif (n[0], n[1]) == ("core", "id"):
                    coreid = int(n[3])
                    key = (physid, coreid,)
                    cores[key] += 1
                    self.threads = max(self.threads, cores[key])
                    if self.threads > 1 and not forced_ht:
                        self.ht = True
                    self.coreids[key].append(cpunum)
                    self.cputocore[cpunum] = key
                    self.cputothread[cpunum] = self.coreids[key].index(cpunum)
                elif n[0] == "flags":
                    seen.add("flags")
                    self.has_tsx = "rtm" in n
                    if "hypervisor" in n:
                        self.hypervisor = True
                elif n[0] == "stepping":
                    seen.add("stepping")
                    self.step = int(n[2])
        if len(seen) >= 7:
            for i in known_cpus:
                if self.model in i[1] or (self.model, self.step) in i[1]:
                    self.realcpu = i[0]
                    if not forced_cpu:
                        self.cpu = i[0]
                    break
        self.force_counters()
        self.limit4_counters = { "cpu": "none" }
        self.standard_counters = { "cpu": tuple(("0,1,2,3",)) }
        if self.cpu.startswith("adl") or self.cpu.startswith("mtl"):
            atom_counters = 6 if self.cpu.startswith("adl") else 8
            newcounters = {
                "cpu_core": 8,
                "cpu": 4,
                "cpu_atom": atom_counters,
            }
            self.standard_counters = {
                "cpu_core": ("0,1,2,3,4,5,6,7", "0,1,2,3", ),
                "cpu": ("0,1,2,3,4,5,6,7", "0,1,2,3", ),
                "cpu_atom": ("0,1,2,3,4,5", ) if atom_counters == 6 else ("0,1,2,3,4,5,6,7,", )
            }
            self.limit4_counters = { "cpu_core": "0,1,2,3", "cpu_atom": "none",
                                     "cpu": "0,1,2,3" }
        elif self.cpu == "slm":
            newcounters = { "cpu": 2 }
            self.standard_counters = { "cpu": ("0,1",) }
        # when running in a hypervisor always assume worst case HT in on
        # also when CPUs are offline assume SMT is on
        elif self.ht or self.hypervisor or (num_offline_cpus() > 0 and not nocheck) or self.cpu in cpus_8gpc:
            if self.cpu in cpus_8gpc or (self.cpu == "simple" and self.realcpu in cpus_8gpc):
                newcounters = {"cpu": 8 }
                self.standard_counters = { "cpu": ("0,1,2,3,4,5,6,7", "0,1,2,3", ) }
                self.limit4_counters = { "cpu": "0,1,2,3" }
            else:
                newcounters = { "cpu": 4 }
        elif self.cpu == "ehl":
            newcounters = { "cpu": 4 }
        else:
            newcounters = { "cpu": 8 }
        if not self.counters:
            self.counters = newcounters
        if not nocheck and not self.env.forcecounters:
            for j in ("cpu", "cpu_core", "cpu_atom"):
                if j in self.counters:
                    self.counters[j] -= reduced_counters()
        if self.cpu in cpus_8gpc:
            self.standard_counters = { "cpu": ("0,1,2,3,4,5,6,7", "0,1,2,3", ) }
            self.limit4_counters = { "cpu": "0,1,2,3" }

        if self.cpu == "spr" and "Max" in self.name:
            self.cpu = "sprmax"

        try:
            self.pmu_name = open("/sys/devices/cpu/caps/pmu_name").read().strip()
        except IOError:
            self.pmu_name = ""

        self.sockets = len(sockets.keys())
        self.modelid = None
        mid = (self.model,)
        self.true_name = self.cpu
        if mid in modelid_map:
            self.modelid = modelid_map[mid]
            self.true_name = self.modelid.lower()
        # XXX match steppings here too

    def force_counters(self):
        cnt = self.env.forcecounters
        if cnt:
            cntn = int(cnt)
            if self.realcpu == "adl" or self.realcpu == "mtl":
                self.counters = {
                    "cpu_core": cntn,
                    "cpu_atom": cntn,
                    "cpu": cntn }
            else:
                self.counters = { "cpu": cntn }