Source code for mojo.jobManagement.management.cpuInfo

"""
CpuInfo
=======
CpuInfo supplies a class to hold the information of the cpu(s) of a workstation.
It owns methods to read the information from proc/cpuinfo and calculate more information from
the given.

Required classes and methods::

    import re
"""

import dataclasses
import re
import typing


@dataclasses.dataclass(frozen=True)
class CPUInfo:
    """wraps properties of /proc/cpuinfo. Such as number of physical/logical CPU cores and hyper-threading status.
    After initialization those properties are accessible as object attributes.
    This class works only for _symmetric_ multiprocessor systems running linux.

    Instance variables:
      numLogicalCPUs - the number of cores that linux takes for the number of cpus
      numCoresPerSocket - how many physical cores are on each socket
      numSiblingsPerSocket - how many (total) cores are on each socket
      numSiblingsPerCore - how many (total) cores are on each core
      hyperThreadingEnabled - determines whether hyperThreading is activated on the system
      numPhysicalCores - how many real (physical) cores are available on the system
      numCPUSockets - how many sockets (slots) are the cpus plugged into
    """
    numLogicalCPUs: int
    numCoresPerSocket: int
    numSiblingsPerSocket: int

    __environment_singleton: typing.ClassVar['CPUInfo'] = None

    @property
    def numSiblingsPerCore(self):
        return self.numSiblingsPerSocket // self.numCoresPerSocket

    @property
    def hyperThreadingEnabled(self):
        return self.numCoresPerSocket * 2 == self.numSiblingsPerSocket

    @property
    def numPhysicalCores(self):
        return self.numLogicalCPUs // self.numSiblingsPerCore

    @property
    def numCPUSockets(self):
        return self.numPhysicalCores // self.numCoresPerSocket

    @classmethod
    def from_proc_cpuinfo(cls) -> 'CPUInfo':
        if cls.__environment_singleton is None:
            with open("/proc/cpuinfo", "r") as handler:
                cpu_info = handler.read()

            section = cpu_info.split("\n\n")[0]  # assuming uniform architecture

            n_logical_cpus = len(re.findall("(processor)", cpu_info))

            cores = re.search(r"cpu cores\s*:\s*(\d+)", section)
            if cores:
                n_cores_per_socket = int(cores.groups()[0])
            else:
                n_cores_per_socket = 1

            siblings = re.search(r"siblings\s*:\s*(\d+)", section)
            if siblings:
                n_siblings_per_socket = int(siblings.groups()[0])
            else:
                n_siblings_per_socket = 1
            cls.__environment_singleton = CPUInfo(n_logical_cpus, n_cores_per_socket, n_siblings_per_socket)
        return cls.__environment_singleton