"""Utility functions related to devices on the platform"""
import os
import re
import logging
from perfmon.common.utils.execute_cmd import execute_cmd
from perfmon.common.utils.parsing import get_parser
_log = logging.getLogger(__name__)
# Infiniband metrics API directory
IB_API_DIR = '/sys/class/infiniband'
# RAPL powercap API directory
RAPL_API_DIR = '/sys/class/powercap/intel-rapl'
# Maximum number of RAPL domains
# cores, gpu, pkg, ram, psys
NUM_RAPL_DOMAINS = 5
# pylint: disable=E0401,W0201,C0301
[docs]def ibstat_ports():
"""This function returns Infiniband ports if present
Returns:
dict: A dict with IB port names and numbers
"""
# A dict with port name as key and port number as value
ports = {}
# Execute ibstat command to get IB ports
ibstat = execute_cmd('ibstat')
if ibstat != 'not_available':
# If command fails, execute_cmd returns not_available
ibstat = ibstat.splitlines()
for index, line in enumerate(ibstat):
line = line.strip()
# Get port name
match = re.match(r"CA '(.*)'", line)
if match:
name = match.group(1)
# Get port number
number = re.match(r'Port ([0-9]+)\:', ibstat[index + 7].strip()).group(1)
state = ibstat[index + 8].split(':')[1].strip()
# Check the state of the port
port_state = re.match(r'Active', state)
if port_state:
ports[name] = number
# If IB drivers are not installed (for example Omnipath), we can directly check the
# IB API directory
if ibstat == 'not_available' and os.path.exists(IB_API_DIR):
for name in os.listdir(IB_API_DIR):
number = os.listdir(os.path.join(IB_API_DIR, name, 'ports'))[0]
ports[name] = str(number)
_log.debug('Infiniband ports detected are %s', ','.join(ports.keys()))
return ports
[docs]def get_rapl_devices():
"""This function gets all the packages, core, uncore and dram device available within RAPL
powercap interface
Returns:
dict: A dict with package names and paths
"""
# Get output from lscpu command
cmd_out = execute_cmd('lscpu')
# Get parser function
parse_lscpu = get_parser(cmd_out)
# Get number of sockets present in the processor
num_sockets = int(parse_lscpu(r'Socket\(s\)'))
# Dict with package and domain names and paths
rapl_domains = {}
# Check available devices for each socket
for i_soc in range(num_sockets):
package_path = os.path.join(RAPL_API_DIR, 'intel-rapl:{}'.format(i_soc))
if os.path.exists(package_path):
with open(os.path.join(package_path, 'name'), 'r') as pkg_name:
package_name = pkg_name.readline().rstrip('\n')
package_num = re.search(r'package-([0-9]+)', package_name).group(1)
package_energy_path = os.path.join(package_path, 'energy_uj')
rapl_domains[package_name] = package_energy_path
for i_dom in range(NUM_RAPL_DOMAINS):
domain_path = os.path.join(package_path, 'intel-rapl:{}:{}'.format(i_soc, i_dom))
if os.path.exists(domain_path):
with open(os.path.join(domain_path, 'name'), 'r') as dom_name:
domain_name = '-'.join([dom_name.readline().rstrip('\n'), package_num])
domain_energy_path = os.path.join(domain_path, 'energy_uj')
rapl_domains[domain_name] = domain_energy_path
_log.debug('RAPL domains detected are %s', ','.join(rapl_domains.keys()))
return rapl_domains