1# SPDX-License-Identifier: BSD-3-Clause 2# Copyright(c) 2010-2014 Intel Corporation 3# Copyright(c) 2022-2023 PANTHEON.tech s.r.o. 4# Copyright(c) 2022-2023 University of New Hampshire 5 6"""Various utility classes and functions. 7 8These are used in multiple modules across the framework. They're here because 9they provide some non-specific functionality, greatly simplify imports or just don't 10fit elsewhere. 11 12Attributes: 13 REGEX_FOR_PCI_ADDRESS: The regex representing a PCI address, e.g. ``0000:00:08.0``. 14""" 15 16import atexit 17import json 18import os 19import subprocess 20from enum import Enum 21from pathlib import Path 22from subprocess import SubprocessError 23 24from scapy.packet import Packet # type: ignore[import] 25 26from .exception import ConfigurationError 27 28REGEX_FOR_PCI_ADDRESS: str = "/[0-9a-fA-F]{4}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}.[0-9]{1}/" 29 30 31def expand_range(range_str: str) -> list[int]: 32 """Process `range_str` into a list of integers. 33 34 There are two possible formats of `range_str`: 35 36 * ``n`` - a single integer, 37 * ``n-m`` - a range of integers. 38 39 The returned range includes both ``n`` and ``m``. Empty string returns an empty list. 40 41 Args: 42 range_str: The range to expand. 43 44 Returns: 45 All the numbers from the range. 46 """ 47 expanded_range: list[int] = [] 48 if range_str: 49 range_boundaries = range_str.split("-") 50 # will throw an exception when items in range_boundaries can't be converted, 51 # serving as type check 52 expanded_range.extend(range(int(range_boundaries[0]), int(range_boundaries[-1]) + 1)) 53 54 return expanded_range 55 56 57def get_packet_summaries(packets: list[Packet]) -> str: 58 """Format a string summary from `packets`. 59 60 Args: 61 packets: The packets to format. 62 63 Returns: 64 The summary of `packets`. 65 """ 66 if len(packets) == 1: 67 packet_summaries = packets[0].summary() 68 else: 69 packet_summaries = json.dumps(list(map(lambda pkt: pkt.summary(), packets)), indent=4) 70 return f"Packet contents: \n{packet_summaries}" 71 72 73class StrEnum(Enum): 74 """Enum with members stored as strings.""" 75 76 @staticmethod 77 def _generate_next_value_(name: str, start: int, count: int, last_values: object) -> str: 78 return name 79 80 def __str__(self) -> str: 81 """The string representation is the name of the member.""" 82 return self.name 83 84 85class MesonArgs(object): 86 """Aggregate the arguments needed to build DPDK.""" 87 88 _default_library: str 89 90 def __init__(self, default_library: str | None = None, **dpdk_args: str | bool): 91 """Initialize the meson arguments. 92 93 Args: 94 default_library: The default library type, Meson supports ``shared``, ``static`` and 95 ``both``. Defaults to :data:`None`, in which case the argument won't be used. 96 dpdk_args: The arguments found in ``meson_options.txt`` in root DPDK directory. 97 Do not use ``-D`` with them. 98 99 Example: 100 :: 101 102 meson_args = MesonArgs(enable_kmods=True). 103 """ 104 self._default_library = f"--default-library={default_library}" if default_library else "" 105 self._dpdk_args = " ".join( 106 ( 107 f"-D{dpdk_arg_name}={dpdk_arg_value}" 108 for dpdk_arg_name, dpdk_arg_value in dpdk_args.items() 109 ) 110 ) 111 112 def __str__(self) -> str: 113 """The actual args.""" 114 return " ".join(f"{self._default_library} {self._dpdk_args}".split()) 115 116 117class _TarCompressionFormat(StrEnum): 118 """Compression formats that tar can use. 119 120 Enum names are the shell compression commands 121 and Enum values are the associated file extensions. 122 """ 123 124 gzip = "gz" 125 compress = "Z" 126 bzip2 = "bz2" 127 lzip = "lz" 128 lzma = "lzma" 129 lzop = "lzo" 130 xz = "xz" 131 zstd = "zst" 132 133 134class DPDKGitTarball(object): 135 """Compressed tarball of DPDK from the repository. 136 137 The class supports the :class:`os.PathLike` protocol, 138 which is used to get the Path of the tarball:: 139 140 from pathlib import Path 141 tarball = DPDKGitTarball("HEAD", "output") 142 tarball_path = Path(tarball) 143 """ 144 145 _git_ref: str 146 _tar_compression_format: _TarCompressionFormat 147 _tarball_dir: Path 148 _tarball_name: str 149 _tarball_path: Path | None 150 151 def __init__( 152 self, 153 git_ref: str, 154 output_dir: str, 155 tar_compression_format: _TarCompressionFormat = _TarCompressionFormat.xz, 156 ): 157 """Create the tarball during initialization. 158 159 The DPDK version is specified with `git_ref`. The tarball will be compressed with 160 `tar_compression_format`, which must be supported by the DTS execution environment. 161 The resulting tarball will be put into `output_dir`. 162 163 Args: 164 git_ref: A git commit ID, tag ID or tree ID. 165 output_dir: The directory where to put the resulting tarball. 166 tar_compression_format: The compression format to use. 167 """ 168 self._git_ref = git_ref 169 self._tar_compression_format = tar_compression_format 170 171 self._tarball_dir = Path(output_dir, "tarball") 172 173 self._get_commit_id() 174 self._create_tarball_dir() 175 176 self._tarball_name = ( 177 f"dpdk-tarball-{self._git_ref}.tar.{self._tar_compression_format.value}" 178 ) 179 self._tarball_path = self._check_tarball_path() 180 if not self._tarball_path: 181 self._create_tarball() 182 183 def _get_commit_id(self) -> None: 184 result = subprocess.run( 185 ["git", "rev-parse", "--verify", self._git_ref], 186 text=True, 187 capture_output=True, 188 ) 189 if result.returncode != 0: 190 raise ConfigurationError( 191 f"{self._git_ref} is neither a path to an existing DPDK " 192 "archive nor a valid git reference.\n" 193 f"Command: {result.args}\n" 194 f"Stdout: {result.stdout}\n" 195 f"Stderr: {result.stderr}" 196 ) 197 self._git_ref = result.stdout.strip() 198 199 def _create_tarball_dir(self) -> None: 200 os.makedirs(self._tarball_dir, exist_ok=True) 201 202 def _check_tarball_path(self) -> Path | None: 203 if self._tarball_name in os.listdir(self._tarball_dir): 204 return Path(self._tarball_dir, self._tarball_name) 205 return None 206 207 def _create_tarball(self) -> None: 208 self._tarball_path = Path(self._tarball_dir, self._tarball_name) 209 210 atexit.register(self._delete_tarball) 211 212 result = subprocess.run( 213 'git -C "$(git rev-parse --show-toplevel)" archive ' 214 f'{self._git_ref} --prefix="dpdk-tarball-{self._git_ref + os.sep}" | ' 215 f"{self._tar_compression_format} > {Path(self._tarball_path.absolute())}", 216 shell=True, 217 text=True, 218 capture_output=True, 219 ) 220 221 if result.returncode != 0: 222 raise SubprocessError( 223 f"Git archive creation failed with exit code {result.returncode}.\n" 224 f"Command: {result.args}\n" 225 f"Stdout: {result.stdout}\n" 226 f"Stderr: {result.stderr}" 227 ) 228 229 atexit.unregister(self._delete_tarball) 230 231 def _delete_tarball(self) -> None: 232 if self._tarball_path and os.path.exists(self._tarball_path): 233 os.remove(self._tarball_path) 234 235 def __fspath__(self) -> str: 236 """The os.PathLike protocol implementation.""" 237 return str(self._tarball_path) 238