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