test/test/autotest_runner.py

   1 # SPDX-License-Identifier: BSD-3-Clause
   2 # Copyright(c) 2010-2014 Intel Corporation
   3
   4 # The main logic behind running autotests in parallel
   5
   6 from __future__ import print_function
   7 import StringIO
   8 import csv
   9 from multiprocessing import Pool, Queue
  10 import pexpect
  11 import re
  12 import subprocess
  13 import sys
  14 import time
  15 import glob
  16 import os
  17
  18 # wait for prompt
  19 def wait_prompt(child):
  20     try:
  21         child.sendline()
  22         result = child.expect(["RTE>>", pexpect.TIMEOUT, pexpect.EOF],
  23                               timeout=120)
  24     except:
  25         return False
  26     if result == 0:
  27         return True
  28     else:
  29         return False
  30
  31
  32 # get all valid NUMA nodes
  33 def get_numa_nodes():
  34     return [
  35         int(
  36             re.match(r"node(\d+)", os.path.basename(node))
  37             .group(1)
  38         )
  39         for node in glob.glob("/sys/devices/system/node/node*")
  40     ]
  41
  42
  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)[0]
  47     cpu_name = os.path.basename(cpu_path)
  48     m = re.match(r"cpu(\d+)", cpu_name)
  49     return int(m.group(1))
  50
  51
  52 pool_child = None  # per-process child
  53
  54
  55 # we initialize each worker with a queue because we need per-pool unique
  56 # command-line arguments, but we cannot do different arguments in an initializer
  57 # because the API doesn't allow per-worker initializer arguments. so, instead,
  58 # we will initialize with a shared queue, and dequeue command-line arguments
  59 # from this queue
  60 def pool_init(queue, result_queue):
  61     global pool_child
  62
  63     cmdline, prefix = queue.get()
  64     start_time = time.time()
  65     name = ("Start %s" % prefix) if prefix != "" else "Start"
  66
  67     # use default prefix if no prefix was specified
  68     prefix_cmdline = "--file-prefix=%s" % prefix if prefix != "" else ""
  69
  70     # append prefix to cmdline
  71     cmdline = "%s %s" % (cmdline, prefix_cmdline)
  72
  73     # prepare logging of init
  74     startuplog = StringIO.StringIO()
  75
  76     # run test app
  77     try:
  78
  79         print("\n%s %s\n" % ("=" * 20, prefix), file=startuplog)
  80         print("\ncmdline=%s" % cmdline, file=startuplog)
  81
  82         pool_child = pexpect.spawn(cmdline, logfile=startuplog)
  83
  84         # wait for target to boot
  85         if not wait_prompt(pool_child):
  86             pool_child.close()
  87
  88             result = tuple((-1,
  89                             "Fail [No prompt]",
  90                             name,
  91                             time.time() - start_time,
  92                             startuplog.getvalue(),
  93                             None))
  94             pool_child = None
  95         else:
  96             result = tuple((0,
  97                             "Success",
  98                             name,
  99                             time.time() - start_time,
 100                             startuplog.getvalue(),
 101                             None))
 102     except:
 103         result = tuple((-1,
 104                         "Fail [Can't run]",
 105                         name,
 106                         time.time() - start_time,
 107                         startuplog.getvalue(),
 108                         None))
 109         pool_child = None
 110
 111     result_queue.put(result)
 112
 113
 114 # run a test
 115 # each result tuple in results list consists of:
 116 #   result value (0 or -1)
 117 #   result string
 118 #   test name
 119 #   total test run time (double)
 120 #   raw test log
 121 #   test report (if not available, should be None)
 122 #
 123 # this function needs to be outside AutotestRunner class because otherwise Pool
 124 # won't work (or rather it will require quite a bit of effort to make it work).
 125 def run_test(target, test):
 126     global pool_child
 127
 128     if pool_child is None:
 129         return -1, "Fail [No test process]", test["Name"], 0, "", None
 130
 131     # create log buffer for each test
 132     # in multiprocessing environment, the logging would be
 133     # interleaved and will create a mess, hence the buffering
 134     logfile = StringIO.StringIO()
 135     pool_child.logfile = logfile
 136
 137     # make a note when the test started
 138     start_time = time.time()
 139
 140     try:
 141         # print test name to log buffer
 142         print("\n%s %s\n" % ("-" * 20, test["Name"]), file=logfile)
 143
 144         # run test function associated with the test
 145         result = test["Func"](pool_child, test["Command"])
 146
 147         # make a note when the test was finished
 148         end_time = time.time()
 149
 150         log = logfile.getvalue()
 151
 152         # append test data to the result tuple
 153         result += (test["Name"], end_time - start_time, log)
 154
 155         # call report function, if any defined, and supply it with
 156         # target and complete log for test run
 157         if test["Report"]:
 158             report = test["Report"](target, log)
 159
 160             # append report to results tuple
 161             result += (report,)
 162         else:
 163             # report is None
 164             result += (None,)
 165     except:
 166         # make a note when the test crashed
 167         end_time = time.time()
 168
 169         # mark test as failed
 170         result = (-1, "Fail [Crash]", test["Name"],
 171                   end_time - start_time, logfile.getvalue(), None)
 172
 173     # return test results
 174     return result
 175
 176
 177 # class representing an instance of autotests run
 178 class AutotestRunner:
 179     cmdline = ""
 180     parallel_test_groups = []
 181     non_parallel_test_groups = []
 182     logfile = None
 183     csvwriter = None
 184     target = ""
 185     start = None
 186     n_tests = 0
 187     fails = 0
 188     log_buffers = []
 189     blacklist = []
 190     whitelist = []
 191
 192     def __init__(self, cmdline, target, blacklist, whitelist, n_processes):
 193         self.cmdline = cmdline
 194         self.target = target
 195         self.binary = cmdline.split()[0]
 196         self.blacklist = blacklist
 197         self.whitelist = whitelist
 198         self.skipped = []
 199         self.parallel_tests = []
 200         self.non_parallel_tests = []
 201         self.n_processes = n_processes
 202         self.active_processes = 0
 203
 204         # log file filename
 205         logfile = "%s.log" % target
 206         csvfile = "%s.csv" % target
 207
 208         self.logfile = open(logfile, "w")
 209         csvfile = open(csvfile, "w")
 210         self.csvwriter = csv.writer(csvfile)
 211
 212         # prepare results table
 213         self.csvwriter.writerow(["test_name", "test_result", "result_str"])
 214
 215     # set up cmdline string
 216     def __get_cmdline(self, cpu_nr):
 217         cmdline = ("taskset -c %i " % cpu_nr) + self.cmdline
 218
 219         return cmdline
 220
 221     def __process_result(self, result):
 222
 223         # unpack result tuple
 224         test_result, result_str, test_name, \
 225             test_time, log, report = result
 226
 227         # get total run time
 228         cur_time = time.time()
 229         total_time = int(cur_time - self.start)
 230
 231         # print results, test run time and total time since start
 232         result = ("%s:" % test_name).ljust(30)
 233         result += result_str.ljust(29)
 234         result += "[%02dm %02ds]" % (test_time / 60, test_time % 60)
 235
 236         # don't print out total time every line, it's the same anyway
 237         print(result + "[%02dm %02ds]" % (total_time / 60, total_time % 60))
 238
 239         # if test failed and it wasn't a "start" test
 240         if test_result < 0:
 241             self.fails += 1
 242
 243         # collect logs
 244         self.log_buffers.append(log)
 245
 246         # create report if it exists
 247         if report:
 248             try:
 249                 f = open("%s_%s_report.rst" %
 250                          (self.target, test_name), "w")
 251             except IOError:
 252                 print("Report for %s could not be created!" % test_name)
 253             else:
 254                 with f:
 255                     f.write(report)
 256
 257         # write test result to CSV file
 258         self.csvwriter.writerow([test_name, test_result, result_str])
 259
 260     # this function checks individual test and decides if this test should be in
 261     # the group by comparing it against  whitelist/blacklist. it also checks if
 262     # the test is compiled into the binary, and marks it as skipped if necessary
 263     def __filter_test(self, test):
 264         test_cmd = test["Command"]
 265         test_id = test_cmd
 266
 267         # dump tests are specified in full e.g. "Dump_mempool"
 268         if "_autotest" in test_id:
 269             test_id = test_id[:-len("_autotest")]
 270
 271         # filter out blacklisted/whitelisted tests
 272         if self.blacklist and test_id in self.blacklist:
 273             return False
 274         if self.whitelist and test_id not in self.whitelist:
 275             return False
 276
 277         # if test wasn't compiled in, remove it as well
 278
 279         # parse the binary for available test commands
 280         stripped = 'not stripped' not in \
 281                    subprocess.check_output(['file', self.binary])
 282         if not stripped:
 283             symbols = subprocess.check_output(['nm',
 284                                                self.binary]).decode('utf-8')
 285             avail_cmds = re.findall('test_register_(\w+)', symbols)
 286
 287             if test_cmd not in avail_cmds:
 288                 # notify user
 289                 result = 0, "Skipped [Not compiled]", test_id, 0, "", None
 290                 self.skipped.append(tuple(result))
 291                 return False
 292
 293         return True
 294
 295     def __run_test_group(self, test_group, worker_cmdlines):
 296         group_queue = Queue()
 297         init_result_queue = Queue()
 298         for proc, cmdline in enumerate(worker_cmdlines):
 299             prefix = "test%i" % proc if len(worker_cmdlines) > 1 else ""
 300             group_queue.put(tuple((cmdline, prefix)))
 301
 302         # create a pool of worker threads
 303         # we will initialize child in the initializer, and we don't need to
 304         # close the child because when the pool worker gets destroyed, child
 305         # closes the process
 306         pool = Pool(processes=len(worker_cmdlines),
 307                     initializer=pool_init,
 308                     initargs=(group_queue, init_result_queue))
 309
 310         results = []
 311
 312         # process all initialization results
 313         for _ in range(len(worker_cmdlines)):
 314             self.__process_result(init_result_queue.get())
 315
 316         # run all tests asynchronously
 317         for test in test_group:
 318             result = pool.apply_async(run_test, (self.target, test))
 319             results.append(result)
 320
 321         # tell the pool to stop all processes once done
 322         pool.close()
 323
 324         # iterate while we have group execution results to get
 325         while len(results) > 0:
 326             # iterate over a copy to be able to safely delete results
 327             # this iterates over a list of group results
 328             for async_result in results[:]:
 329                 # if the thread hasn't finished yet, continue
 330                 if not async_result.ready():
 331                     continue
 332
 333                 res = async_result.get()
 334
 335                 self.__process_result(res)
 336
 337                 # remove result from results list once we're done with it
 338                 results.remove(async_result)
 339
 340     # iterate over test groups and run tests associated with them
 341     def run_all_tests(self):
 342         # filter groups
 343         self.parallel_tests = list(
 344             filter(self.__filter_test,
 345                    self.parallel_tests)
 346         )
 347         self.non_parallel_tests = list(
 348             filter(self.__filter_test,
 349                    self.non_parallel_tests)
 350         )
 351
 352         parallel_cmdlines = []
 353         # FreeBSD doesn't have NUMA support
 354         numa_nodes = get_numa_nodes()
 355         if len(numa_nodes) > 0:
 356             for proc in range(self.n_processes):
 357                 # spread cpu affinity between NUMA nodes to have less chance of
 358                 # running out of memory while running multiple test apps in
 359                 # parallel. to do that, alternate between NUMA nodes in a round
 360                 # robin fashion, and pick an arbitrary CPU from that node to
 361                 # taskset our execution to
 362                 numa_node = numa_nodes[self.active_processes % len(numa_nodes)]
 363                 cpu_nr = first_cpu_on_node(numa_node)
 364                 parallel_cmdlines += [self.__get_cmdline(cpu_nr)]
 365                 # increase number of active processes so that the next cmdline
 366                 # gets a different NUMA node
 367                 self.active_processes += 1
 368         else:
 369             parallel_cmdlines = [self.cmdline] * self.n_processes
 370
 371         print("Running tests with %d workers" % self.n_processes)
 372
 373         # create table header
 374         print("")
 375         print("Test name".ljust(30) + "Test result".ljust(29) +
 376               "Test".center(9) + "Total".center(9))
 377         print("=" * 80)
 378
 379         if len(self.skipped):
 380             print("Skipped autotests:")
 381
 382             # print out any skipped tests
 383             for result in self.skipped:
 384                 # unpack result tuple
 385                 test_result, result_str, test_name, _, _, _ = result
 386                 self.csvwriter.writerow([test_name, test_result, result_str])
 387
 388                 t = ("%s:" % test_name).ljust(30)
 389                 t += result_str.ljust(29)
 390                 t += "[00m 00s]"
 391
 392                 print(t)
 393
 394         # make a note of tests start time
 395         self.start = time.time()
 396
 397         # whatever happens, try to save as much logs as possible
 398         try:
 399             if len(self.parallel_tests) > 0:
 400                 print("Parallel autotests:")
 401                 self.__run_test_group(self.parallel_tests, parallel_cmdlines)
 402
 403             if len(self.non_parallel_tests) > 0:
 404                 print("Non-parallel autotests:")
 405                 self.__run_test_group(self.non_parallel_tests, [self.cmdline])
 406
 407             # get total run time
 408             cur_time = time.time()
 409             total_time = int(cur_time - self.start)
 410
 411             # print out summary
 412             print("=" * 80)
 413             print("Total run time: %02dm %02ds" % (total_time / 60,
 414                                                    total_time % 60))
 415             if self.fails != 0:
 416                 print("Number of failed tests: %s" % str(self.fails))
 417
 418             # write summary to logfile
 419             self.logfile.write("Summary\n")
 420             self.logfile.write("Target: ".ljust(15) + "%s\n" % self.target)
 421             self.logfile.write("Tests: ".ljust(15) + "%i\n" % self.n_tests)
 422             self.logfile.write("Failed tests: ".ljust(
 423                 15) + "%i\n" % self.fails)
 424         except:
 425             print("Exception occurred")
 426             print(sys.exc_info())
 427             self.fails = 1
 428
 429         # drop logs from all executions to a logfile
 430         for buf in self.log_buffers:
 431             self.logfile.write(buf.replace("\r", ""))
 432
 433         return self.fails