5 # Test fio's io_uring_cmd ioengine with NVMe pass-through commands.
8 # see python3 nvmept.py --help
11 # python3 t/nvmept.py --dut /dev/ng0n1
12 # python3 t/nvmept.py --dut /dev/ng1n1 -f ./fio
22 from pathlib import Path
23 from fiotestlib import FioJobCmdTest, run_fio_tests
26 class PassThruTest(FioJobCmdTest):
28 NVMe pass-through test class. Check to make sure output for selected data
29 direction(s) is non-zero and that zero data appears for other directions.
32 def setup(self, parameters):
37 "--ioengine=io_uring_cmd",
41 "--iodepth_batch_complete=4",
42 f"--filename={self.fio_opts['filename']}",
43 f"--rw={self.fio_opts['rw']}",
44 f"--output={self.filenames['output']}",
45 f"--output-format={self.fio_opts['output-format']}",
47 for opt in ['fixedbufs', 'nonvectored', 'force_async', 'registerfiles',
48 'sqthread_poll', 'sqthread_poll_cpu', 'hipri', 'nowait',
49 'time_based', 'runtime', 'verify', 'io_size']:
50 if opt in self.fio_opts:
51 option = f"--{opt}={self.fio_opts[opt]}"
52 fio_args.append(option)
54 super().setup(fio_args)
57 def check_result(self):
58 super().check_result()
60 if 'rw' not in self.fio_opts:
66 job = self.json_data['jobs'][0]
68 if self.fio_opts['rw'] in ['read', 'randread']:
69 self.passed = self.check_all_ddirs(['read'], job)
70 elif self.fio_opts['rw'] in ['write', 'randwrite']:
71 if 'verify' not in self.fio_opts:
72 self.passed = self.check_all_ddirs(['write'], job)
74 self.passed = self.check_all_ddirs(['read', 'write'], job)
75 elif self.fio_opts['rw'] in ['trim', 'randtrim']:
76 self.passed = self.check_all_ddirs(['trim'], job)
77 elif self.fio_opts['rw'] in ['readwrite', 'randrw']:
78 self.passed = self.check_all_ddirs(['read', 'write'], job)
79 elif self.fio_opts['rw'] in ['trimwrite', 'randtrimwrite']:
80 self.passed = self.check_all_ddirs(['trim', 'write'], job)
82 print(f"Unhandled rw value {self.fio_opts['rw']}")
85 if job['iodepth_level']['8'] < 95:
86 print("Did not achieve requested iodepth")
97 "output-format": "json",
99 "test_class": PassThruTest,
107 "output-format": "json",
109 "test_class": PassThruTest,
117 "output-format": "json",
119 "test_class": PassThruTest,
127 "output-format": "json",
129 "test_class": PassThruTest,
137 "output-format": "json",
139 "test_class": PassThruTest,
147 "output-format": "json",
149 "test_class": PassThruTest,
155 "io_size": 1024*1024,
157 "output-format": "json",
159 "test_class": PassThruTest,
165 "io_size": 1024*1024,
167 "output-format": "json",
169 "test_class": PassThruTest,
177 "output-format": "json",
179 "test_class": PassThruTest,
187 "output-format": "json",
189 "test_class": PassThruTest,
197 "output-format": "json",
199 "test_class": PassThruTest,
204 "rw": 'randtrimwrite',
207 "output-format": "json",
209 "test_class": PassThruTest,
222 "output-format": "json",
224 "test_class": PassThruTest,
237 "output-format": "json",
239 "test_class": PassThruTest,
242 # We can't enable fixedbufs because for trim-only
243 # workloads fio actually does not allocate any buffers
254 "output-format": "json",
256 "test_class": PassThruTest,
261 """Parse command-line arguments."""
263 parser = argparse.ArgumentParser()
264 parser.add_argument('-f', '--fio', help='path to file executable (e.g., ./fio)')
265 parser.add_argument('-a', '--artifact-root', help='artifact root directory')
266 parser.add_argument('-s', '--skip', nargs='+', type=int,
267 help='list of test(s) to skip')
268 parser.add_argument('-o', '--run-only', nargs='+', type=int,
269 help='list of test(s) to run, skipping all others')
270 parser.add_argument('--dut', help='target NVMe character device to test '
271 '(e.g., /dev/ng0n1). WARNING: THIS IS A DESTRUCTIVE TEST', required=True)
272 args = parser.parse_args()
278 """Run tests using fio's io_uring_cmd ioengine to send NVMe pass through commands."""
282 artifact_root = args.artifact_root if args.artifact_root else \
283 f"nvmept-test-{time.strftime('%Y%m%d-%H%M%S')}"
284 os.mkdir(artifact_root)
285 print(f"Artifact directory is {artifact_root}")
288 fio_path = str(Path(args.fio).absolute())
291 print(f"fio path is {fio_path}")
293 for test in TEST_LIST:
294 test['fio_opts']['filename'] = args.dut
297 'fio_path': fio_path,
298 'fio_root': str(Path(__file__).absolute().parent.parent),
299 'artifact_root': artifact_root,
300 'basename': 'nvmept',
303 _, failed, _ = run_fio_tests(TEST_LIST, test_env, args)
307 if __name__ == '__main__':