utils/lit/lit/run.py

   1 import os
   2 import threading
   3 import time
   4 import traceback
   5
   6 try:
   7     import win32api
   8 except ImportError:
   9     win32api = None
  10
  11 import lit.Test
  12
  13 ###
  14 # Test Execution Implementation
  15
  16 class TestProvider(object):
  17     def __init__(self, tests, max_time):
  18         self.max_time = max_time
  19         self.iter = iter(range(len(tests)))
  20         self.lock = threading.Lock()
  21         self.start_time = time.time()
  22         self.canceled = False
  23
  24     def cancel(self):
  25         self.lock.acquire()
  26         self.canceled = True
  27         self.lock.release()
  28
  29     def get(self):
  30         # Check if we have run out of time.
  31         if self.max_time is not None:
  32             if time.time() - self.start_time > self.max_time:
  33                 return None
  34
  35         # Otherwise take the next test.
  36         self.lock.acquire()
  37         if self.canceled:
  38           self.lock.release()
  39           return None
  40         for item in self.iter:
  41             break
  42         else:
  43             item = None
  44         self.lock.release()
  45         return item
  46
  47 class Tester(object):
  48     def __init__(self, run_instance, provider, consumer):
  49         self.run_instance = run_instance
  50         self.provider = provider
  51         self.consumer = consumer
  52
  53     def run(self):
  54         while 1:
  55             item = self.provider.get()
  56             if item is None:
  57                 break
  58             self.run_test(item)
  59         self.consumer.task_finished()
  60
  61     def run_test(self, test_index):
  62         test = self.run_instance.tests[test_index]
  63         try:
  64             self.run_instance.execute_test(test)
  65         except KeyboardInterrupt:
  66             # This is a sad hack. Unfortunately subprocess goes
  67             # bonkers with ctrl-c and we start forking merrily.
  68             print('\nCtrl-C detected, goodbye.')
  69             os.kill(0,9)
  70         self.consumer.update(test_index, test)
  71
  72 class ThreadResultsConsumer(object):
  73     def __init__(self, display):
  74         self.display = display
  75         self.lock = threading.Lock()
  76
  77     def update(self, test_index, test):
  78         self.lock.acquire()
  79         try:
  80             self.display.update(test)
  81         finally:
  82             self.lock.release()
  83
  84     def task_finished(self):
  85         pass
  86
  87     def handle_results(self):
  88         pass
  89
  90 def run_one_tester(run, provider, display):
  91     tester = Tester(run, provider, display)
  92     tester.run()
  93
  94 ###
  95
  96 class Run(object):
  97     """
  98     This class represents a concrete, configured testing run.
  99     """
 100
 101     def __init__(self, lit_config, tests):
 102         self.lit_config = lit_config
 103         self.tests = tests
 104
 105     def execute_test(self, test):
 106         result = None
 107         start_time = time.time()
 108         try:
 109             result = test.config.test_format.execute(test, self.lit_config)
 110
 111             # Support deprecated result from execute() which returned the result
 112             # code and additional output as a tuple.
 113             if isinstance(result, tuple):
 114                 code, output = result
 115                 result = lit.Test.Result(code, output)
 116             elif not isinstance(result, lit.Test.Result):
 117                 raise ValueError("unexpected result from test execution")
 118         except KeyboardInterrupt:
 119             raise
 120         except:
 121             if self.lit_config.debug:
 122                 raise
 123             output = 'Exception during script execution:\n'
 124             output += traceback.format_exc()
 125             output += '\n'
 126             result = lit.Test.Result(lit.Test.UNRESOLVED, output)
 127         result.elapsed = time.time() - start_time
 128
 129         test.setResult(result)
 130
 131     def execute_tests(self, display, jobs, max_time=None):
 132         """
 133         execute_tests(display, jobs, [max_time])
 134
 135         Execute each of the tests in the run, using up to jobs number of
 136         parallel tasks, and inform the display of each individual result. The
 137         provided tests should be a subset of the tests available in this run
 138         object.
 139
 140         If max_time is non-None, it should be a time in seconds after which to
 141         stop executing tests.
 142
 143         The display object will have its update method called with each test as
 144         it is completed. The calls are guaranteed to be locked with respect to
 145         one another, but are *not* guaranteed to be called on the same thread as
 146         this method was invoked on.
 147
 148         Upon completion, each test in the run will have its result
 149         computed. Tests which were not actually executed (for any reason) will
 150         be given an UNRESOLVED result.
 151         """
 152
 153         # Create the test provider object.
 154         provider = TestProvider(self.tests, max_time)
 155
 156         # Install a console-control signal handler on Windows.
 157         if win32api is not None:
 158             def console_ctrl_handler(type):
 159                 provider.cancel()
 160                 return True
 161             win32api.SetConsoleCtrlHandler(console_ctrl_handler, True)
 162
 163         # Actually execute the tests.
 164         self._execute_tests_with_provider(provider, display, jobs)
 165
 166         # Update results for any tests which weren't run.
 167         for test in self.tests:
 168             if test.result is None:
 169                 test.setResult(lit.Test.Result(lit.Test.UNRESOLVED, '', 0.0))
 170
 171     def _execute_tests_with_provider(self, provider, display, jobs):
 172         consumer = ThreadResultsConsumer(display)
 173
 174         # If only using one testing thread, don't use tasks at all; this lets us
 175         # profile, among other things.
 176         if jobs == 1:
 177             run_one_tester(self, provider, consumer)
 178             return
 179
 180         # Start all of the tasks.
 181         tasks = [threading.Thread(target=run_one_tester,
 182                                   args=(self, provider, consumer))
 183                  for i in range(jobs)]
 184         for t in tasks:
 185             t.start()
 186
 187         # Allow the consumer to handle results, if necessary.
 188         consumer.handle_results()
 189
 190         # Wait for all the tasks to complete.
 191         for t in tasks:
 192             t.join()