1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
|
#!/usr/bin/env python3
"""
fiotestcommon.py
This contains constant definitions, helpers, and a Requirements class that can
be used to help with running fio tests.
"""
import os
import locale
import logging
import platform
import subprocess
import multiprocessing
SUCCESS_DEFAULT = {
'zero_return': True,
'stderr_empty': True,
'timeout': 600,
}
SUCCESS_LONG = {
'zero_return': True,
'stderr_empty': True,
'timeout': 3600,
}
SUCCESS_NONZERO = {
'zero_return': False,
'stderr_empty': False,
'timeout': 600,
}
SUCCESS_STDERR = {
'zero_return': True,
'stderr_empty': False,
'timeout': 600,
}
def get_file(filename):
"""Safely read a file."""
file_data = ''
success = True
try:
with open(filename, "r", encoding=locale.getpreferredencoding()) as output_file:
file_data = output_file.read()
except OSError:
success = False
return file_data, success
class Requirements():
"""Requirements consists of multiple run environment characteristics.
These are to determine if a particular test can be run"""
_linux = False
_libaio = False
_io_uring = False
_zbd = False
_root = False
_zoned_nullb = False
_not_macos = False
_not_windows = False
_unittests = False
_cpucount4 = False
_nvmecdev = False
def __init__(self, fio_root, args):
Requirements._not_macos = platform.system() != "Darwin"
Requirements._not_windows = platform.system() != "Windows"
Requirements._linux = platform.system() == "Linux"
if Requirements._linux:
config_file = os.path.join(fio_root, "config-host.h")
contents, success = get_file(config_file)
if not success:
print(f"Unable to open {config_file} to check requirements")
Requirements._zbd = True
else:
Requirements._zbd = "CONFIG_HAS_BLKZONED" in contents
Requirements._libaio = "CONFIG_LIBAIO" in contents
contents, success = get_file("/proc/kallsyms")
if not success:
print("Unable to open '/proc/kallsyms' to probe for io_uring support")
else:
Requirements._io_uring = "io_uring_setup" in contents
Requirements._root = os.geteuid() == 0
if Requirements._zbd and Requirements._root:
try:
subprocess.run(["modprobe", "null_blk"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
if os.path.exists("/sys/module/null_blk/parameters/zoned"):
Requirements._zoned_nullb = True
except Exception:
pass
if platform.system() == "Windows":
utest_exe = "unittest.exe"
else:
utest_exe = "unittest"
unittest_path = os.path.join(fio_root, "unittests", utest_exe)
Requirements._unittests = os.path.exists(unittest_path)
Requirements._cpucount4 = multiprocessing.cpu_count() >= 4
Requirements._nvmecdev = args.nvmecdev if hasattr(args, 'nvmecdev') else False
req_list = [
Requirements.linux,
Requirements.libaio,
Requirements.io_uring,
Requirements.zbd,
Requirements.root,
Requirements.zoned_nullb,
Requirements.not_macos,
Requirements.not_windows,
Requirements.unittests,
Requirements.cpucount4,
Requirements.nvmecdev,
]
for req in req_list:
value, desc = req()
logging.debug("Requirements: Requirement '%s' met? %s", desc, value)
@classmethod
def linux(cls):
"""Are we running on Linux?"""
return Requirements._linux, "Linux required"
@classmethod
def libaio(cls):
"""Is libaio available?"""
return Requirements._libaio, "libaio required"
@classmethod
def io_uring(cls):
"""Is io_uring available?"""
return Requirements._io_uring, "io_uring required"
@classmethod
def zbd(cls):
"""Is ZBD support available?"""
return Requirements._zbd, "Zoned block device support required"
@classmethod
def root(cls):
"""Are we running as root?"""
return Requirements._root, "root required"
@classmethod
def zoned_nullb(cls):
"""Are zoned null block devices available?"""
return Requirements._zoned_nullb, "Zoned null block device support required"
@classmethod
def not_macos(cls):
"""Are we running on a platform other than macOS?"""
return Requirements._not_macos, "platform other than macOS required"
@classmethod
def not_windows(cls):
"""Are we running on a platform other than Windws?"""
return Requirements._not_windows, "platform other than Windows required"
@classmethod
def unittests(cls):
"""Were unittests built?"""
return Requirements._unittests, "Unittests support required"
@classmethod
def cpucount4(cls):
"""Do we have at least 4 CPUs?"""
return Requirements._cpucount4, "4+ CPUs required"
@classmethod
def nvmecdev(cls):
"""Do we have an NVMe character device to test?"""
return Requirements._nvmecdev, "NVMe character device test target required"
|
