2 # SPDX-License-Identifier: BSD-3-Clause
3 # Copyright(c) 2010-2014 Intel Corporation
5 # The main logic behind running autotests in parallel
9 from multiprocessing import Pool, Queue
19 def wait_prompt(child):
22 result = child.expect(["RTE>>", pexpect.TIMEOUT, pexpect.EOF],
32 # get all valid NUMA nodes
36 re.match(r"node(\d+)", os.path.basename(node))
39 for node in glob.glob("/sys/devices/system/node/node*")
43 # find first (or any, really) CPU on a particular node, will be used to spread
44 # processes around NUMA nodes to avoid exhausting memory on particular node
45 def first_cpu_on_node(node_nr):
46 cpu_path = glob.glob("/sys/devices/system/node/node%d/cpu*" % node_nr)
47 r = re.compile(r"cpu(\d+)")
48 cpu_name = filter(None,
50 map(os.path.basename, cpu_path)
53 return int(next(cpu_name).group(1))
55 pool_child = None # per-process child
58 # we initialize each worker with a queue because we need per-pool unique
59 # command-line arguments, but we cannot do different arguments in an initializer
60 # because the API doesn't allow per-worker initializer arguments. so, instead,
61 # we will initialize with a shared queue, and dequeue command-line arguments
63 def pool_init(queue, result_queue):
66 cmdline, prefix = queue.get()
67 start_time = time.time()
68 name = ("Start %s" % prefix) if prefix != "" else "Start"
70 # use default prefix if no prefix was specified
71 prefix_cmdline = "--file-prefix=%s" % prefix if prefix != "" else ""
73 # append prefix to cmdline
74 cmdline = "%s %s" % (cmdline, prefix_cmdline)
76 # prepare logging of init
77 startuplog = io.StringIO()
82 print("\n%s %s\n" % ("=" * 20, prefix), file=startuplog)
83 print("\ncmdline=%s" % cmdline, file=startuplog)
85 pool_child = pexpect.spawn(cmdline, logfile=startuplog, encoding='utf-8')
86 # wait for target to boot
87 if not wait_prompt(pool_child):
93 time.time() - start_time,
94 startuplog.getvalue(),
101 time.time() - start_time,
102 startuplog.getvalue(),
108 time.time() - start_time,
109 startuplog.getvalue(),
113 result_queue.put(result)
117 # each result tuple in results list consists of:
118 # result value (0 or -1)
121 # total test run time (double)
123 # test report (if not available, should be None)
125 # this function needs to be outside AutotestRunner class because otherwise Pool
126 # won't work (or rather it will require quite a bit of effort to make it work).
127 def run_test(target, test):
130 if pool_child is None:
131 return -1, "Fail [No test process]", test["Name"], 0, "", None
133 # create log buffer for each test
134 # in multiprocessing environment, the logging would be
135 # interleaved and will create a mess, hence the buffering
136 logfile = io.StringIO()
137 pool_child.logfile = logfile
139 # make a note when the test started
140 start_time = time.time()
143 # print test name to log buffer
144 print("\n%s %s\n" % ("-" * 20, test["Name"]), file=logfile)
146 # run test function associated with the test
147 result = test["Func"](pool_child, test["Command"])
149 # make a note when the test was finished
150 end_time = time.time()
152 log = logfile.getvalue()
154 # append test data to the result tuple
155 result += (test["Name"], end_time - start_time, log)
157 # call report function, if any defined, and supply it with
158 # target and complete log for test run
160 report = test["Report"](target, log)
162 # append report to results tuple
168 # make a note when the test crashed
169 end_time = time.time()
171 # mark test as failed
172 result = (-1, "Fail [Crash]", test["Name"],
173 end_time - start_time, logfile.getvalue(), None)
175 # return test results
179 # class representing an instance of autotests run
180 class AutotestRunner:
182 parallel_test_groups = []
183 non_parallel_test_groups = []
194 def __init__(self, cmdline, target, blocklist, allowlist, n_processes):
195 self.cmdline = cmdline
197 self.blocklist = blocklist
198 self.allowlist = allowlist
200 self.parallel_tests = []
201 self.non_parallel_tests = []
202 self.n_processes = n_processes
203 self.active_processes = 0
205 # parse the binary for available test commands
206 binary = cmdline.split()[0]
207 stripped = 'not stripped' not in \
208 subprocess.check_output(['file', binary]).decode()
210 symbols = subprocess.check_output(['nm', binary]).decode()
211 self.avail_cmds = re.findall('test_register_(\w+)', symbols)
213 self.avail_cmds = None
216 logfile = "%s.log" % target
217 csvfile = "%s.csv" % target
219 self.logfile = open(logfile, "w")
220 csvfile = open(csvfile, "w")
221 self.csvwriter = csv.writer(csvfile)
223 # prepare results table
224 self.csvwriter.writerow(["test_name", "test_result", "result_str"])
226 # set up cmdline string
227 def __get_cmdline(self, cpu_nr):
228 cmdline = ("taskset -c %i " % cpu_nr) + self.cmdline
232 def __process_result(self, result):
234 # unpack result tuple
235 test_result, result_str, test_name, \
236 test_time, log, report = result
239 cur_time = time.time()
240 total_time = int(cur_time - self.start)
242 # print results, test run time and total time since start
243 result = ("%s:" % test_name).ljust(30)
244 result += result_str.ljust(29)
245 result += "[%02dm %02ds]" % (test_time / 60, test_time % 60)
247 # don't print out total time every line, it's the same anyway
248 print(result + "[%02dm %02ds]" % (total_time / 60, total_time % 60))
250 # if test failed and it wasn't a "start" test
255 self.log_buffers.append(log)
257 # create report if it exists
260 f = open("%s_%s_report.rst" %
261 (self.target, test_name), "w")
263 print("Report for %s could not be created!" % test_name)
268 # write test result to CSV file
269 self.csvwriter.writerow([test_name, test_result, result_str])
271 # this function checks individual test and decides if this test should be in
272 # the group by comparing it against allowlist/blocklist. it also checks if
273 # the test is compiled into the binary, and marks it as skipped if necessary
274 def __filter_test(self, test):
275 test_cmd = test["Command"]
278 # dump tests are specified in full e.g. "Dump_mempool"
279 if "_autotest" in test_id:
280 test_id = test_id[:-len("_autotest")]
282 # filter out blocked/allowed tests
283 if self.blocklist and test_id in self.blocklist:
285 if self.allowlist and test_id not in self.allowlist:
288 # if test wasn't compiled in, remove it as well
289 if self.avail_cmds and test_cmd not in self.avail_cmds:
290 result = 0, "Skipped [Not compiled]", test_id, 0, "", None
291 self.skipped.append(tuple(result))
296 def __run_test_group(self, test_group, worker_cmdlines):
297 group_queue = Queue()
298 init_result_queue = Queue()
299 for proc, cmdline in enumerate(worker_cmdlines):
300 prefix = "test%i" % proc if len(worker_cmdlines) > 1 else ""
301 group_queue.put(tuple((cmdline, prefix)))
303 # create a pool of worker threads
304 # we will initialize child in the initializer, and we don't need to
305 # close the child because when the pool worker gets destroyed, child
307 pool = Pool(processes=len(worker_cmdlines),
308 initializer=pool_init,
309 initargs=(group_queue, init_result_queue))
313 # process all initialization results
314 for _ in range(len(worker_cmdlines)):
315 self.__process_result(init_result_queue.get())
317 # run all tests asynchronously
318 for test in test_group:
319 result = pool.apply_async(run_test, (self.target, test))
320 results.append(result)
322 # tell the pool to stop all processes once done
325 # iterate while we have group execution results to get
326 while len(results) > 0:
327 # iterate over a copy to be able to safely delete results
328 # this iterates over a list of group results
329 for async_result in results[:]:
330 # if the thread hasn't finished yet, continue
331 if not async_result.ready():
334 res = async_result.get()
336 self.__process_result(res)
338 # remove result from results list once we're done with it
339 results.remove(async_result)
341 # iterate over test groups and run tests associated with them
342 def run_all_tests(self):
344 self.parallel_tests = list(
345 filter(self.__filter_test,
348 self.non_parallel_tests = list(
349 filter(self.__filter_test,
350 self.non_parallel_tests)
353 parallel_cmdlines = []
354 # FreeBSD doesn't have NUMA support
355 numa_nodes = get_numa_nodes()
356 if len(numa_nodes) > 0:
357 for proc in range(self.n_processes):
358 # spread cpu affinity between NUMA nodes to have less chance of
359 # running out of memory while running multiple test apps in
360 # parallel. to do that, alternate between NUMA nodes in a round
361 # robin fashion, and pick an arbitrary CPU from that node to
362 # taskset our execution to
363 numa_node = numa_nodes[self.active_processes % len(numa_nodes)]
364 cpu_nr = first_cpu_on_node(numa_node)
365 parallel_cmdlines += [self.__get_cmdline(cpu_nr)]
366 # increase number of active processes so that the next cmdline
367 # gets a different NUMA node
368 self.active_processes += 1
370 parallel_cmdlines = [self.cmdline] * self.n_processes
372 print("Running tests with %d workers" % self.n_processes)
374 # create table header
376 print("Test name".ljust(30) + "Test result".ljust(29) +
377 "Test".center(9) + "Total".center(9))
380 if len(self.skipped):
381 print("Skipped autotests:")
383 # print out any skipped tests
384 for result in self.skipped:
385 # unpack result tuple
386 test_result, result_str, test_name, _, _, _ = result
387 self.csvwriter.writerow([test_name, test_result, result_str])
389 t = ("%s:" % test_name).ljust(30)
390 t += result_str.ljust(29)
395 # make a note of tests start time
396 self.start = time.time()
398 # whatever happens, try to save as much logs as possible
400 if len(self.parallel_tests) > 0:
401 print("Parallel autotests:")
402 self.__run_test_group(self.parallel_tests, parallel_cmdlines)
404 if len(self.non_parallel_tests) > 0:
405 print("Non-parallel autotests:")
406 self.__run_test_group(self.non_parallel_tests, [self.cmdline])
409 cur_time = time.time()
410 total_time = int(cur_time - self.start)
414 print("Total run time: %02dm %02ds" % (total_time / 60,
417 print("Number of failed tests: %s" % str(self.fails))
419 # write summary to logfile
420 self.logfile.write("Summary\n")
421 self.logfile.write("Target: ".ljust(15) + "%s\n" % self.target)
422 self.logfile.write("Tests: ".ljust(15) + "%i\n" % self.n_tests)
423 self.logfile.write("Failed tests: ".ljust(
424 15) + "%i\n" % self.fails)
426 print("Exception occurred")
427 print(sys.exc_info())
430 # drop logs from all executions to a logfile
431 for buf in self.log_buffers:
432 self.logfile.write(buf.replace("\r", ""))