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 if 'rw' not in self.fio_opts:
64 job = self.json_data['jobs'][0]
66 if self.fio_opts['rw'] in ['read', 'randread']:
67 self.passed = self.check_all_ddirs(['read'], job)
68 elif self.fio_opts['rw'] in ['write', 'randwrite']:
69 if 'verify' not in self.fio_opts:
70 self.passed = self.check_all_ddirs(['write'], job)
72 self.passed = self.check_all_ddirs(['read', 'write'], job)
73 elif self.fio_opts['rw'] in ['trim', 'randtrim']:
74 self.passed = self.check_all_ddirs(['trim'], job)
75 elif self.fio_opts['rw'] in ['readwrite', 'randrw']:
76 self.passed = self.check_all_ddirs(['read', 'write'], job)
77 elif self.fio_opts['rw'] in ['trimwrite', 'randtrimwrite']:
78 self.passed = self.check_all_ddirs(['trim', 'write'], job)
80 print(f"Unhandled rw value {self.fio_opts['rw']}")
83 if job['iodepth_level']['8'] < 95:
84 print("Did not achieve requested iodepth")
95 "output-format": "json",
97 "test_class": PassThruTest,
105 "output-format": "json",
107 "test_class": PassThruTest,
115 "output-format": "json",
117 "test_class": PassThruTest,
125 "output-format": "json",
127 "test_class": PassThruTest,
135 "output-format": "json",
137 "test_class": PassThruTest,
145 "output-format": "json",
147 "test_class": PassThruTest,
153 "io_size": 1024*1024,
155 "output-format": "json",
157 "test_class": PassThruTest,
163 "io_size": 1024*1024,
165 "output-format": "json",
167 "test_class": PassThruTest,
175 "output-format": "json",
177 "test_class": PassThruTest,
185 "output-format": "json",
187 "test_class": PassThruTest,
195 "output-format": "json",
197 "test_class": PassThruTest,
202 "rw": 'randtrimwrite',
205 "output-format": "json",
207 "test_class": PassThruTest,
220 "output-format": "json",
222 "test_class": PassThruTest,
235 "output-format": "json",
237 "test_class": PassThruTest,
242 """Parse command-line arguments."""
244 parser = argparse.ArgumentParser()
245 parser.add_argument('-f', '--fio', help='path to file executable (e.g., ./fio)')
246 parser.add_argument('-a', '--artifact-root', help='artifact root directory')
247 parser.add_argument('-s', '--skip', nargs='+', type=int,
248 help='list of test(s) to skip')
249 parser.add_argument('-o', '--run-only', nargs='+', type=int,
250 help='list of test(s) to run, skipping all others')
251 parser.add_argument('--dut', help='target NVMe character device to test '
252 '(e.g., /dev/ng0n1). WARNING: THIS IS A DESTRUCTIVE TEST', required=True)
253 args = parser.parse_args()
259 """Run tests using fio's io_uring_cmd ioengine to send NVMe pass through commands."""
263 artifact_root = args.artifact_root if args.artifact_root else \
264 f"nvmept-test-{time.strftime('%Y%m%d-%H%M%S')}"
265 os.mkdir(artifact_root)
266 print(f"Artifact directory is {artifact_root}")
269 fio_path = str(Path(args.fio).absolute())
272 print(f"fio path is {fio_path}")
274 for test in TEST_LIST:
275 test['fio_opts']['filename'] = args.dut
278 'fio_path': fio_path,
279 'fio_root': str(Path(__file__).absolute().parent.parent),
280 'artifact_root': artifact_root,
281 'basename': 'readonly',
284 _, failed, _ = run_fio_tests(TEST_LIST, test_env, args)
288 if __name__ == '__main__':