CSIT-1139: Implement parallel test execution

The implementation of parallel test execution in VPP Test Framework.
- VPPTestCase test methods are grouped together
- tests are running in separate processes
- VPP instances spawned by tests are assigned to different cores
- output from these processes is redirected through pipes and printed
out testcase by testcase
- TEST_JOBS env var is used to specify the number of parallel processes
- improved test summary
- a bit of code cleanup

Change-Id: I9ca93904d9fe2c3daf980500c64a8611838ae28c
Signed-off-by: juraj.linkes <juraj.linkes@pantheon.tech>

1 files changed, 631 insertions, 158 deletions

diff --git a/test/run_tests.py b/test/run_tests.py
index 02e4738184d..aac28d13ed9 100644
--- a/test/run_tests.py
+++ b/test/run_tests.py
@@ -7,10 +7,17 @@ import select
 import unittest
 import argparse
 import time
-from multiprocessing import Process, Pipe
-from framework import VppTestRunner
+import threading
+import signal
+import psutil
+from multiprocessing import Process, Pipe, cpu_count
+from multiprocessing.queues import Queue
+from multiprocessing.managers import BaseManager
+from framework import VppTestRunner, running_extended_tests, VppTestCase, \
+    get_testcase_doc_name, get_test_description
 from debug import spawn_gdb
-from log import global_logger
+from log import get_parallel_logger, double_line_delim, RED, YELLOW, GREEN, \
+    colorize
 from discover_tests import discover_tests
 from subprocess import check_output, CalledProcessError
 from util import check_core_path
@@ -20,29 +27,349 @@ from util import check_core_path
 # that child process is stuck (e.g. waiting for shm mutex, which will never
 # get unlocked) and kill the child
 core_timeout = 3
+min_req_shm = 536870912  # min 512MB shm required
+# 128MB per extra process
+shm_per_process = 134217728
 
 
-def test_runner_wrapper(suite, keep_alive_pipe, result_pipe, failed_pipe):
-    result = not VppTestRunner(
-        keep_alive_pipe=keep_alive_pipe,
-        failed_pipe=failed_pipe,
-        verbosity=verbose,
-        failfast=failfast).run(suite).wasSuccessful()
+class StreamQueue(Queue):
+    def write(self, msg):
+        self.put(msg)
+
+    def flush(self):
+        sys.__stdout__.flush()
+        sys.__stderr__.flush()
+
+    def fileno(self):
+        return self._writer.fileno()
+
+
+class StreamQueueManager(BaseManager):
+    pass
+
+
+StreamQueueManager.register('Queue', StreamQueue)
+
+
+def test_runner_wrapper(suite, keep_alive_pipe, result_pipe, stdouterr_queue,
+                        logger):
+    sys.stdout = stdouterr_queue
+    sys.stderr = stdouterr_queue
+    VppTestCase.logger = logger
+    result = VppTestRunner(keep_alive_pipe=keep_alive_pipe,
+                           descriptions=descriptions,
+                           verbosity=verbose,
+                           failfast=failfast).run(suite)
     result_pipe.send(result)
     result_pipe.close()
     keep_alive_pipe.close()
-    failed_pipe.close()
 
 
-class add_to_suite_callback:
-    def __init__(self, suite):
-        self.suite = suite
+class TestCaseWrapper(object):
+    def __init__(self, testcase_suite, manager):
+        self.keep_alive_parent_end, self.keep_alive_child_end = Pipe(
+            duplex=False)
+        self.result_parent_end, self.result_child_end = Pipe(duplex=False)
+        self.testcase_suite = testcase_suite
+        self.stdouterr_queue = manager.Queue()
+        self.logger = get_parallel_logger(self.stdouterr_queue)
+        self.child = Process(target=test_runner_wrapper,
+                             args=(testcase_suite, self.keep_alive_child_end,
+                                   self.result_child_end, self.stdouterr_queue,
+                                   self.logger)
+                             )
+        self.child.start()
+        self.pid = self.child.pid
+        self.last_test_temp_dir = None
+        self.last_test_vpp_binary = None
+        self.last_test = None
+        self.result = None
+        self.last_heard = time.time()
+        self.core_detected_at = None
+        self.failed_tests = []
+
+    def close_pipes(self):
+        self.keep_alive_child_end.close()
+        self.result_child_end.close()
+        self.keep_alive_parent_end.close()
+        self.result_parent_end.close()
+
+
+def stdouterr_reader_wrapper(unread_testcases, finished_unread_testcases,
+                             read_testcases):
+    read_testcase = None
+    while read_testcases.is_set() or len(unread_testcases) > 0:
+        if not read_testcase:
+            if len(finished_unread_testcases) > 0:
+                read_testcase = finished_unread_testcases.pop()
+                unread_testcases.remove(read_testcase)
+            elif len(unread_testcases) > 0:
+                read_testcase = unread_testcases.pop()
+        if read_testcase:
+            data = ''
+            while data is not None:
+                sys.stdout.write(data)
+                data = read_testcase.stdouterr_queue.get()
+
+            read_testcase.stdouterr_queue.close()
+            finished_unread_testcases.discard(read_testcase)
+            read_testcase = None
+
+
+def run_forked(testcases):
+    wrapped_testcase_suites = set()
+
+    # suites are unhashable, need to use list
+    results = []
+    debug_core = os.getenv("DEBUG", "").lower() == "core"
+    unread_testcases = set()
+    finished_unread_testcases = set()
+    manager = StreamQueueManager()
+    manager.start()
+    for i in range(concurrent_tests):
+        if len(testcases) > 0:
+            wrapped_testcase_suite = TestCaseWrapper(testcases.pop(0), manager)
+            wrapped_testcase_suites.add(wrapped_testcase_suite)
+            unread_testcases.add(wrapped_testcase_suite)
+            # time.sleep(1)
+        else:
+            break
+
+    read_from_testcases = threading.Event()
+    read_from_testcases.set()
+    stdouterr_thread = threading.Thread(target=stdouterr_reader_wrapper,
+                                        args=(unread_testcases,
+                                              finished_unread_testcases,
+                                              read_from_testcases))
+    stdouterr_thread.start()
+
+    while len(wrapped_testcase_suites) > 0:
+        finished_testcase_suites = set()
+        for wrapped_testcase_suite in wrapped_testcase_suites:
+            readable = select.select(
+                [wrapped_testcase_suite.keep_alive_parent_end.fileno(),
+                 wrapped_testcase_suite.result_parent_end.fileno()],
+                [], [], 1)[0]
+            if wrapped_testcase_suite.result_parent_end.fileno() in readable:
+                results.append(
+                    (wrapped_testcase_suite.testcase_suite,
+                     wrapped_testcase_suite.result_parent_end.recv()))
+                finished_testcase_suites.add(wrapped_testcase_suite)
+                continue
+
+            if wrapped_testcase_suite.keep_alive_parent_end.fileno() \
+                    in readable:
+                while wrapped_testcase_suite.keep_alive_parent_end.poll():
+                    wrapped_testcase_suite.last_test, \
+                        wrapped_testcase_suite.last_test_vpp_binary, \
+                        wrapped_testcase_suite.last_test_temp_dir, \
+                        wrapped_testcase_suite.vpp_pid = \
+                        wrapped_testcase_suite.keep_alive_parent_end.recv()
+                wrapped_testcase_suite.last_heard = time.time()
+
+            fail = False
+            if wrapped_testcase_suite.last_heard + test_timeout < time.time() \
+                    and not os.path.isfile(
+                                "%s/_core_handled" %
+                                wrapped_testcase_suite.last_test_temp_dir):
+                fail = True
+                wrapped_testcase_suite.logger.critical(
+                    "Timeout while waiting for child test "
+                    "runner process (last test running was "
+                    "`%s' in `%s')!" %
+                    (wrapped_testcase_suite.last_test,
+                     wrapped_testcase_suite.last_test_temp_dir))
+            elif not wrapped_testcase_suite.child.is_alive():
+                fail = True
+                wrapped_testcase_suite.logger.critical(
+                    "Child python process unexpectedly died "
+                    "(last test running was `%s' in `%s')!" %
+                    (wrapped_testcase_suite.last_test,
+                     wrapped_testcase_suite.last_test_temp_dir))
+            elif wrapped_testcase_suite.last_test_temp_dir and \
+                    wrapped_testcase_suite.last_test_vpp_binary:
+                core_path = "%s/core" % \
+                            wrapped_testcase_suite.last_test_temp_dir
+                if os.path.isfile(core_path):
+                    if wrapped_testcase_suite.core_detected_at is None:
+                        wrapped_testcase_suite.core_detected_at = time.time()
+                    elif wrapped_testcase_suite.core_detected_at + \
+                            core_timeout < time.time():
+                        if not os.path.isfile(
+                                        "%s/_core_handled" %
+                                        wrapped_testcase_suite.
+                                        last_test_temp_dir):
+                            wrapped_testcase_suite.logger.critical(
+                                "Child python process unresponsive and core-"
+                                "file exists in test temporary directory!")
+                            fail = True
+
+            if fail:
+                failed_dir = os.getenv('VPP_TEST_FAILED_DIR')
+                lttd = os.path.basename(
+                    wrapped_testcase_suite.last_test_temp_dir)
+                link_path = '%s%s-FAILED' % (failed_dir, lttd)
+                wrapped_testcase_suite.logger.error(
+                    "Creating a link to the failed test: %s -> %s" %
+                    (link_path, lttd))
+                if not os.path.exists(link_path):
+                    os.symlink(wrapped_testcase_suite.last_test_temp_dir,
+                               link_path)
+                api_post_mortem_path = "/tmp/api_post_mortem.%d" % \
+                                       wrapped_testcase_suite.vpp_pid
+                if os.path.isfile(api_post_mortem_path):
+                    wrapped_testcase_suite.logger.error(
+                        "Copying api_post_mortem.%d to %s" %
+                        (wrapped_testcase_suite.vpp_pid,
+                         wrapped_testcase_suite.last_test_temp_dir))
+                    shutil.copy2(api_post_mortem_path,
+                                 wrapped_testcase_suite.last_test_temp_dir)
+                if wrapped_testcase_suite.last_test_temp_dir and \
+                        wrapped_testcase_suite.last_test_vpp_binary:
+                    core_path = "%s/core" % \
+                                wrapped_testcase_suite.last_test_temp_dir
+                    if os.path.isfile(core_path):
+                        wrapped_testcase_suite.logger.error(
+                            "Core-file exists in test temporary directory: %s!"
+                            % core_path)
+                        check_core_path(wrapped_testcase_suite.logger,
+                                        core_path)
+                        wrapped_testcase_suite.logger.debug(
+                            "Running `file %s':" % core_path)
+                        try:
+                            info = check_output(["file", core_path])
+                            wrapped_testcase_suite.logger.debug(info)
+                        except CalledProcessError as e:
+                            wrapped_testcase_suite.logger.error(
+                                "Could not run `file' utility on core-file, "
+                                "rc=%s" % e.returncode)
+                            pass
+                        if debug_core:
+                            spawn_gdb(
+                                wrapped_testcase_suite.last_test_vpp_binary,
+                                core_path, wrapped_testcase_suite.logger)
+                wrapped_testcase_suite.child.terminate()
+                try:
+                    # terminating the child process tends to leave orphan
+                    # VPP process around
+                    os.kill(wrapped_testcase_suite.vpp_pid, signal.SIGTERM)
+                except OSError:
+                    # already dead
+                    pass
+                results.append((wrapped_testcase_suite.testcase_suite, None))
+                finished_testcase_suites.add(wrapped_testcase_suite)
+
+        for finished_testcase in finished_testcase_suites:
+            finished_testcase.child.join()
+            finished_testcase.close_pipes()
+            wrapped_testcase_suites.remove(finished_testcase)
+            finished_unread_testcases.add(finished_testcase)
+            finished_testcase.stdouterr_queue.put(None)
+            if len(testcases) > 0:
+                new_testcase = TestCaseWrapper(testcases.pop(0), manager)
+                wrapped_testcase_suites.add(new_testcase)
+                unread_testcases.add(new_testcase)
+
+    read_from_testcases.clear()
+    stdouterr_thread.join(test_timeout)
+    manager.shutdown()
+    return results
+
+
+class SplitToSuitesCallback:
+    def __init__(self, filter_callback):
+        self.suites = {}
+        self.suite_name = 'default'
+        self.filter_callback = filter_callback
+        self.filtered = unittest.TestSuite()
 
     def __call__(self, file_name, cls, method):
-        suite.addTest(cls(method))
+        test_method = cls(method)
+        if self.filter_callback(file_name, cls.__name__, method):
+            self.suite_name = file_name + cls.__name__
+            if self.suite_name not in self.suites:
+                self.suites[self.suite_name] = unittest.TestSuite()
+            self.suites[self.suite_name].addTest(test_method)
+
+        else:
+            self.filtered.addTest(test_method)
+
 
+test_option = "TEST"
 
-class Filter_by_class_list:
+
+def parse_test_option():
+    f = os.getenv(test_option, None)
+    filter_file_name = None
+    filter_class_name = None
+    filter_func_name = None
+    if f:
+        if '.' in f:
+            parts = f.split('.')
+            if len(parts) > 3:
+                raise Exception("Unrecognized %s option: %s" %
+                                (test_option, f))
+            if len(parts) > 2:
+                if parts[2] not in ('*', ''):
+                    filter_func_name = parts[2]
+            if parts[1] not in ('*', ''):
+                filter_class_name = parts[1]
+            if parts[0] not in ('*', ''):
+                if parts[0].startswith('test_'):
+                    filter_file_name = parts[0]
+                else:
+                    filter_file_name = 'test_%s' % parts[0]
+        else:
+            if f.startswith('test_'):
+                filter_file_name = f
+            else:
+                filter_file_name = 'test_%s' % f
+    if filter_file_name:
+        filter_file_name = '%s.py' % filter_file_name
+    return filter_file_name, filter_class_name, filter_func_name
+
+
+def filter_tests(tests, filter_cb):
+    result = unittest.suite.TestSuite()
+    for t in tests:
+        if isinstance(t, unittest.suite.TestSuite):
+            # this is a bunch of tests, recursively filter...
+            x = filter_tests(t, filter_cb)
+            if x.countTestCases() > 0:
+                result.addTest(x)
+        elif isinstance(t, unittest.TestCase):
+            # this is a single test
+            parts = t.id().split('.')
+            # t.id() for common cases like this:
+            # test_classifier.TestClassifier.test_acl_ip
+            # apply filtering only if it is so
+            if len(parts) == 3:
+                if not filter_cb(parts[0], parts[1], parts[2]):
+                    continue
+            result.addTest(t)
+        else:
+            # unexpected object, don't touch it
+            result.addTest(t)
+    return result
+
+
+class FilterByTestOption:
+    def __init__(self, filter_file_name, filter_class_name, filter_func_name):
+        self.filter_file_name = filter_file_name
+        self.filter_class_name = filter_class_name
+        self.filter_func_name = filter_func_name
+
+    def __call__(self, file_name, class_name, func_name):
+        if self.filter_file_name and file_name != self.filter_file_name:
+            return False
+        if self.filter_class_name and class_name != self.filter_class_name:
+            return False
+        if self.filter_func_name and func_name != self.filter_func_name:
+            return False
+        return True
+
+
+class FilterByClassList:
     def __init__(self, class_list):
         self.class_list = class_list
 
@@ -51,174 +378,320 @@ class Filter_by_class_list:
 
 
 def suite_from_failed(suite, failed):
-    filter_cb = Filter_by_class_list(failed)
-    suite = VppTestRunner.filter_tests(suite, filter_cb)
-    if 0 == suite.countTestCases():
-        raise Exception("Suite is empty after filtering out the failed tests!")
+    filter_cb = FilterByClassList(failed)
+    suite = filter_tests(suite, filter_cb)
     return suite
 
 
-def run_forked(suite):
-    keep_alive_parent_end, keep_alive_child_end = Pipe(duplex=False)
-    result_parent_end, result_child_end = Pipe(duplex=False)
-    failed_parent_end, failed_child_end = Pipe(duplex=False)
-
-    child = Process(target=test_runner_wrapper,
-                    args=(suite, keep_alive_child_end, result_child_end,
-                          failed_child_end))
-    child.start()
-    last_test_temp_dir = None
-    last_test_vpp_binary = None
-    last_test = None
-    result = None
-    failed = set()
-    last_heard = time.time()
-    core_detected_at = None
-    debug_core = os.getenv("DEBUG", "").lower() == "core"
-    while True:
-        readable = select.select([keep_alive_parent_end.fileno(),
-                                  result_parent_end.fileno(),
-                                  failed_parent_end.fileno(),
-                                  ],
-                                 [], [], 1)[0]
-        if result_parent_end.fileno() in readable:
-            result = result_parent_end.recv()
-            break
-        if keep_alive_parent_end.fileno() in readable:
-            while keep_alive_parent_end.poll():
-                last_test, last_test_vpp_binary,\
-                    last_test_temp_dir, vpp_pid = keep_alive_parent_end.recv()
-            last_heard = time.time()
-        if failed_parent_end.fileno() in readable:
-            while failed_parent_end.poll():
-                failed_test = failed_parent_end.recv()
-                failed.add(failed_test.__name__)
-            last_heard = time.time()
-        fail = False
-        if last_heard + test_timeout < time.time() and \
-                not os.path.isfile("%s/_core_handled" % last_test_temp_dir):
-            fail = True
-            global_logger.critical("Timeout while waiting for child test "
-                                   "runner process (last test running was "
-                                   "`%s' in `%s')!" %
-                                   (last_test, last_test_temp_dir))
-        elif not child.is_alive():
-            fail = True
-            global_logger.critical("Child python process unexpectedly died "
-                                   "(last test running was `%s' in `%s')!" %
-                                   (last_test, last_test_temp_dir))
-        elif last_test_temp_dir and last_test_vpp_binary:
-            core_path = "%s/core" % last_test_temp_dir
-            if os.path.isfile(core_path):
-                if core_detected_at is None:
-                    core_detected_at = time.time()
-                elif core_detected_at + core_timeout < time.time():
-                    if not os.path.isfile(
-                            "%s/_core_handled" % last_test_temp_dir):
-                        global_logger.critical(
-                            "Child python process unresponsive and core-file "
-                            "exists in test temporary directory!")
-                        fail = True
-
-        if fail:
-            failed_dir = os.getenv('VPP_TEST_FAILED_DIR')
-            lttd = last_test_temp_dir.split("/")[-1]
-            link_path = '%s%s-FAILED' % (failed_dir, lttd)
-            global_logger.error("Creating a link to the failed " +
-                                "test: %s -> %s" % (link_path, lttd))
-            try:
-                os.symlink(last_test_temp_dir, link_path)
-            except Exception:
-                pass
-            api_post_mortem_path = "/tmp/api_post_mortem.%d" % vpp_pid
-            if os.path.isfile(api_post_mortem_path):
-                global_logger.error("Copying api_post_mortem.%d to %s" %
-                                    (vpp_pid, last_test_temp_dir))
-                shutil.copy2(api_post_mortem_path, last_test_temp_dir)
-            if last_test_temp_dir and last_test_vpp_binary:
-                core_path = "%s/core" % last_test_temp_dir
-                if os.path.isfile(core_path):
-                    global_logger.error("Core-file exists in test temporary "
-                                        "directory: %s!" % core_path)
-                    check_core_path(global_logger, core_path)
-                    global_logger.debug("Running `file %s':" % core_path)
-                    try:
-                        info = check_output(["file", core_path])
-                        global_logger.debug(info)
-                    except CalledProcessError as e:
-                        global_logger.error(
-                            "Could not run `file' utility on core-file, "
-                            "rc=%s" % e.returncode)
-                        pass
-                    if debug_core:
-                        spawn_gdb(last_test_vpp_binary, core_path,
-                                  global_logger)
-            child.terminate()
-            result = -1
-            break
-    keep_alive_parent_end.close()
-    result_parent_end.close()
-    failed_parent_end.close()
-    return result, failed
+class NonPassedResults(dict):
+    def __init__(self):
+        super(NonPassedResults, self).__init__()
+        self.all_testcases = 0
+        self.results_per_suite = {}
+        self.failures_id = 'failures'
+        self.errors_id = 'errors'
+        self.crashes_id = 'crashes'
+        self.skipped_id = 'skipped'
+        self.expectedFailures_id = 'expectedFailures'
+        self.unexpectedSuccesses_id = 'unexpectedSuccesses'
+        self.rerun = []
+        self[self.failures_id] = 0
+        self[self.errors_id] = 0
+        self[self.crashes_id] = 0
+        self[self.skipped_id] = 0
+        self[self.expectedFailures_id] = 0
+        self[self.unexpectedSuccesses_id] = 0
+
+    def _add_result(self, test, result_id):
+        if isinstance(test, VppTestCase):
+            parts = test.id().split('.')
+            if len(parts) == 3:
+                tc_class = get_testcase_doc_name(test)
+                if tc_class not in self.results_per_suite:
+                    # failed, errored, skipped, expectedly failed,
+                    # unexpectedly passed
+                    self.results_per_suite[tc_class] = \
+                        {self.failures_id: [],
+                         self.errors_id: [],
+                         self.crashes_id: [],
+                         self.skipped_id: [],
+                         self.expectedFailures_id: [],
+                         self.unexpectedSuccesses_id: []}
+                self.results_per_suite[tc_class][result_id].append(test)
+                return True
+        return False
+
+    def add_results(self, testcases, testcase_result,
+                    duplicates=None):
+        for failed_testcase, _ in testcases:
+            if self._add_result(failed_testcase, testcase_result):
+                if duplicates:
+                    if failed_testcase not in duplicates:
+                        self[testcase_result] += 1
+                else:
+                    self[testcase_result] += 1
+
+    def add_result(self, testcase_suite, result):
+        retval = 0
+        self.all_testcases += testcase_suite.countTestCases()
+        if result:
+            # suite finished properly
+            if not result.wasSuccessful():
+                retval = 1
+
+            self.add_results(result.failures, self.failures_id)
+            self.add_results(result.errors, self.errors_id,
+                             result.failures + result.errors)
+            self.add_results(result.skipped, self.skipped_id)
+            self.add_results(result.expectedFailures,
+                             self.expectedFailures_id)
+            self.add_results(result.unexpectedSuccesses,
+                             self.unexpectedSuccesses_id)
+
+        else:
+            # suite crashed
+            retval = -1
+            self.add_results([(x, None) for x in testcase_suite],
+                             self.crashes_id)
+
+        if retval != 0:
+            if concurrent_tests == 1:
+                if result:
+                    rerun_classes = {x[0].__class__.__name__ for
+                                     x in result.errors}
+                    rerun_classes.update({x[0].__class__.__name__ for
+                                          x in result.failures})
+                    self.rerun.append(suite_from_failed(testcase_suite,
+                                                        rerun_classes))
+                else:
+                    self.rerun.append(testcase_suite)
+            else:
+                self.rerun.append(testcase_suite)
+
+        return retval
+
+    def print_results(self):
+        print('')
+        print(double_line_delim)
+        print('TEST RESULTS:')
+        print('        Executed tests: {}'.format(self.all_testcases))
+        print('          Passed tests: {}'.format(
+            colorize(str(self.all_testcases -
+                         self.all_nonpassed), GREEN)))
+        if self[self.failures_id] > 0:
+            print('          Failed tests: {}'.format(
+                colorize(str(self[self.failures_id]), RED)))
+        if self[self.errors_id] > 0:
+            print('         Errored tests: {}'.format(
+                colorize(str(self[self.errors_id]), RED)))
+        if self[self.crashes_id] > 0:
+            print('         Crashed tests: {}'.format(
+                colorize(str(self[self.crashes_id]), RED)))
+        if self[self.skipped_id] > 0:
+            print('         Skipped tests: {}'.format(
+                colorize(str(self[self.skipped_id]), YELLOW)))
+        if self[self.expectedFailures_id] > 0:
+            print('     Expected failures: {}'.format(
+                colorize(str(self[self.expectedFailures_id]), GREEN)))
+        if self[self.unexpectedSuccesses_id] > 0:
+            print('  Unexpected successes: {}'.format(
+                colorize(str(self[self.unexpectedSuccesses_id]), YELLOW)))
+
+        if self.all_failed > 0:
+            print('FAILED TESTS:')
+            for testcase_class, suite_results in \
+                    self.results_per_suite.items():
+                failed_testcases = suite_results[
+                    self.failures_id]
+                errored_testcases = suite_results[
+                    self.errors_id]
+                crashed_testcases = suite_results[
+                    self.crashes_id]
+                if len(failed_testcases) or len(errored_testcases) \
+                        or len(crashed_testcases):
+                    print('  Testcase name: {}'.format(
+                        colorize(testcase_class, RED)))
+                    for failed_test in failed_testcases:
+                        print('     FAILED: {}'.format(
+                            colorize(get_test_description(
+                                descriptions, failed_test), RED)))
+                    for failed_test in errored_testcases:
+                        print('    ERRORED: {}'.format(
+                            colorize(get_test_description(
+                                descriptions, failed_test), RED)))
+                    for failed_test in crashed_testcases:
+                        print('    CRASHED: {}'.format(
+                            colorize(get_test_description(
+                                descriptions, failed_test), RED)))
+
+        print(double_line_delim)
+        print('')
+
+    @property
+    def all_nonpassed(self):
+        return self[self.failures_id] + self[self.errors_id] + \
+               self[self.crashes_id] + self[self.skipped_id] + \
+               self[self.expectedFailures_id] + \
+               self[self.unexpectedSuccesses_id]
+
+    @property
+    def all_failed(self):
+        return self[self.failures_id] + self[self.errors_id] + \
+               self[self.crashes_id]
+
+
+def parse_results(results):
+    """
+    Prints the number of executed, passed, failed, errored, skipped,
+    expectedly failed and unexpectedly passed tests and details about
+    failed, errored, expectedly failed and unexpectedly passed tests.
+
+    Also returns any suites where any test failed.
+
+    :param results:
+    :return:
+    """
+
+    results_per_suite = NonPassedResults()
+    crashed = False
+    failed = False
+    for testcase_suite, result in results:
+        result_code = results_per_suite.add_result(testcase_suite, result)
+        if result_code == 1:
+            failed = True
+        elif result_code == -1:
+            crashed = True
+
+    results_per_suite.print_results()
+
+    if crashed:
+        return_code = -1
+    elif failed:
+        return_code = 1
+    else:
+        return_code = 0
+    return return_code, results_per_suite.rerun
+
+
+def parse_digit_env(env_var, default):
+    value = os.getenv(env_var, default)
+    if value != default:
+        if value.isdigit():
+            value = int(value)
+        else:
+            print('WARNING: unsupported value "%s" for env var "%s",'
+                  'defaulting to %s' % (value, env_var, default))
+            value = default
+    return value
 
 
 if __name__ == '__main__':
 
-    try:
-        verbose = int(os.getenv("V", 0))
-    except ValueError:
-        verbose = 0
+    verbose = parse_digit_env("V", 0)
+
+    test_timeout = parse_digit_env("TIMEOUT", 600)  # default = 10 minutes
+
+    retries = parse_digit_env("RETRIES", 0)
+
+    debug = os.getenv("DEBUG", "n").lower() in ["gdb", "gdbserver"]
+
+    step = os.getenv("STEP", "n").lower() in ("y", "yes", "1")
+
+    force_foreground = \
+        os.getenv("FORCE_FOREGROUND", "n").lower() in ("y", "yes", "1")
 
-    default_test_timeout = 600  # 10 minutes
-    try:
-        test_timeout = int(os.getenv("TIMEOUT", default_test_timeout))
-    except ValueError:
-        test_timeout = default_test_timeout
+    run_interactive = debug or step or force_foreground
 
-    debug = os.getenv("DEBUG")
+    test_jobs = os.getenv("TEST_JOBS", "1").lower()  # default = 1 process
+    if test_jobs == 'auto':
+        if run_interactive:
+            concurrent_tests = 1
+            print('Interactive mode required, running on one core')
+        else:
+            shm_free = psutil.disk_usage('/dev/shm').free
+            shm_max_processes = 1
+            if shm_free < min_req_shm:
+                raise Exception('Not enough free space in /dev/shm. Required '
+                                'free space is at least %sM.'
+                                % (min_req_shm >> 20))
+            else:
+                extra_shm = shm_free - min_req_shm
+                shm_max_processes += extra_shm / shm_per_process
+            concurrent_tests = max(cpu_count(), shm_max_processes)
+            print('Found enough resources to run tests with %s cores'
+                  % concurrent_tests)
+    elif test_jobs.isdigit():
+        concurrent_tests = int(test_jobs)
+    else:
+        concurrent_tests = 1
 
-    s = os.getenv("STEP", "n")
-    step = True if s.lower() in ("y", "yes", "1") else False
+    if run_interactive and concurrent_tests > 1:
+        raise NotImplementedError(
+            'Running tests interactively (DEBUG, STEP or FORCE_FOREGROUND is '
+            'set) in parallel (TEST_JOBS is more than 1) is not '
+            'supported')
 
     parser = argparse.ArgumentParser(description="VPP unit tests")
-    parser.add_argument("-f", "--failfast", action='count',
+    parser.add_argument("-f", "--failfast", action='store_true',
                         help="fast failure flag")
     parser.add_argument("-d", "--dir", action='append', type=str,
                         help="directory containing test files "
                              "(may be specified multiple times)")
     args = parser.parse_args()
-    failfast = True if args.failfast == 1 else False
+    failfast = args.failfast
+    descriptions = True
+
+    print("Running tests using custom test runner")  # debug message
+    filter_file, filter_class, filter_func = parse_test_option()
+
+    print("Active filters: file=%s, class=%s, function=%s" % (
+        filter_file, filter_class, filter_func))
 
-    suite = unittest.TestSuite()
-    cb = add_to_suite_callback(suite)
+    filter_cb = FilterByTestOption(filter_file, filter_class, filter_func)
+
+    cb = SplitToSuitesCallback(filter_cb)
     for d in args.dir:
         print("Adding tests from directory tree %s" % d)
         discover_tests(d, cb)
 
-    try:
-        retries = int(os.getenv("RETRIES", 0))
-    except ValueError:
-        retries = 0
+    # suites are not hashable, need to use list
+    suites = []
+    tests_amount = 0
+    for testcase_suite in cb.suites.values():
+        tests_amount += testcase_suite.countTestCases()
+        suites.append(testcase_suite)
+
+    if concurrent_tests == 1:
+        new_suite = unittest.TestSuite()
+        for suite in suites:
+            new_suite.addTest(suite)
+
+        suites = [new_suite]
+
+    print("%s out of %s tests match specified filters" % (
+        tests_amount, tests_amount + cb.filtered.countTestCases()))
+
+    if not running_extended_tests():
+        print("Not running extended tests (some tests will be skipped)")
 
-    try:
-        force_foreground = int(os.getenv("FORCE_FOREGROUND", 0))
-    except ValueError:
-        force_foreground = 0
     attempts = retries + 1
     if attempts > 1:
         print("Perform %s attempts to pass the suite..." % attempts)
-    if (debug is not None and debug.lower() in ["gdb", "gdbserver"]) or step\
-            or force_foreground:
-        # don't fork if requiring interactive terminal..
+
+    if run_interactive:
+        # don't fork if requiring interactive terminal
         sys.exit(not VppTestRunner(
-            verbosity=verbose, failfast=failfast).run(suite).wasSuccessful())
+            verbosity=verbose, failfast=failfast)
+                 .run(suites[0]).wasSuccessful())
     else:
-        while True:
-            result, failed = run_forked(suite)
-            attempts = attempts - 1
-            print("%s test(s) failed, %s attempt(s) left" %
-                  (len(failed), attempts))
-            if len(failed) > 0 and attempts > 0:
-                suite = suite_from_failed(suite, failed)
-                continue
-            sys.exit(result)
+        exit_code = 0
+        while len(suites) > 0 and attempts > 0:
+            tests_amount = sum([x.countTestCases() for x in suites])
+            results = run_forked(suites)
+            exit_code, suites = parse_results(results)
+            attempts -= 1
+            if exit_code == 0:
+                print('Test run was successful')
+            else:
+                print('%s attempt(s) left.' % attempts)
+        sys.exit(exit_code)