Lit: Implement RWLock to serialize retrying flaky test.

[chapuni]

Summary

To stabilize flaky tests, I thought they may be serialized after 2nd attempt.

I added rwlock object in litConfig and I know it is in wrong place. Any better idea?

libcxx/trunk/utils/libcxx/test/format.py

def _evaluate_pass_test(self, test, tmpBase, lit_config,
             # Right now we just mark all of the .dat files in the same
             # directory as dependencies, but it's likely less than that. We
             # should add a `// FILE-DEP: foo.dat` to each test to track this.
             data_files = [os.path.join(local_cwd, f)
                           for f in os.listdir(local_cwd) if f.endswith('.dat')]
             is_flaky = self._get_parser('FLAKY_TEST.', parsers).getValue()
             max_retry = 3 if is_flaky else 1
             for retry_count in range(max_retry):
+                lit_config._retry_rwlock.acquire(retry_count)
                 cmd, out, err, rc = self.executor.run(exec_path, [exec_path],
                                                       local_cwd, data_files,
                                                       env)
+                lit_config._retry_rwlock.release(retry_count)
                 if rc == 0:
                     res = lit.Test.PASS if retry_count == 0 else lit.Test.FLAKYPASS
                     return res, ''
                 elif rc != 0 and retry_count + 1 == max_retry:
                     report = libcxx.util.makeReport(cmd, out, err, rc)
                     report = "Compiled With: %s\n%s" % (compile_cmd, report)
                     report += "Compiled test failed unexpectedly!"
                     return lit.Test.FAIL, report

llvm/trunk/utils/lit/lit/TestRunner.py

substitutions += getDefaultSubstitutions(test, tmpDir, tmpBase,
                                              normalize_slashes=useExternalSh)
     script = applySubstitutions(script, substitutions)
 
     # Re-run failed tests up to test_retry_attempts times.
     attempts = 1
     if hasattr(test.config, 'test_retry_attempts'):
         attempts += test.config.test_retry_attempts
     for i in range(attempts):
+        if hasattr(litConfig, '_retry_rwlock'):
+            litConfig._retry_rwlock.acquire(i)
         res = _runShTest(test, litConfig, useExternalSh, script, tmpBase)
+        if hasattr(litConfig, '_retry_rwlock'):
+            litConfig._retry_rwlock.release(i)
         if res.code != Test.FAIL:
             break
     # If we had to run the test more than once, count it as a flaky pass. These
     # will be printed separately in the test summary.
     if i > 0 and res.code == Test.PASS:
         res.code = Test.FLAKYPASS
     return res

llvm/trunk/utils/lit/lit/run.py

 
 try:
     import win32api
 except ImportError:
     win32api = None
 
 import multiprocessing
 import lit.Test
+import lit.util
 
 def abort_now():
     """Abort the current process without doing any exception teardown"""
     sys.stdout.flush()
     if win32api:
         win32api.TerminateProcess(win32api.GetCurrentProcess(), 3)
     else:
         os.kill(0, 9)
def execute_tests(self, display, jobs, max_time=None):
                 abort_now()
                 return True
             win32api.SetConsoleCtrlHandler(console_ctrl_handler, True)
 
         # Save the display object on the runner so that we can update it from
         # our task completion callback.
         self.display = display
 
+        # FIXME: It might be in wrong place.
+        self.lit_config._retry_rwlock = lit.util.RWLock()
+
         self.failure_count = 0
         self.hit_max_failures = False
         if self.lit_config.singleProcess:
             global child_lit_config
             child_lit_config = self.lit_config
             for test_index, test in enumerate(self.tests):
                 result = worker_run_one_test(test_index, test)
                 self.consume_test_result(result)

llvm/trunk/utils/lit/lit/util.py

 import errno
 import itertools
 import math
+import multiprocessing
 import numbers
 import os
 import platform
 import signal
 import subprocess
 import sys
 import threading
 
try:
         for child in children_iterator:
             try:
                 child.kill()
             except psutil.NoSuchProcess:
                 pass
         psutilProc.kill()
     except psutil.NoSuchProcess:
         pass
+
+class RWLock:
+    """
+    Provide RWLock
+    """
+    def __init__(self):
+        lock =  multiprocessing.RLock()
+        self._lock = lock
+        self._rcond = multiprocessing.Condition(lock)
+        self._wcond = multiprocessing.Condition(lock)
+        self._rval  = multiprocessing.Value('i', 0)
+        self._wval  = multiprocessing.Value('i', 0)
+
+    def read_acquire(self):
+        self._lock.acquire()
+        while self._wval.value != 0:
+            self._rcond.wait()
+        self._rval.value += 1
+        self._lock.release()
+
+    def write_acquire(self):
+        self._lock.acquire()
+        while self._rval.value > 0:
+            self._wcond.wait()
+        assert self._wval.value == 0
+        self._wval.value = 1
+
+    def read_release(self):
+        self._lock.acquire()
+        rval = self._rval.value - 1
+        self._rval.value = rval
+        if rval == 0:
+            self._wcond.notify()
+        self._lock.release()
+
+    def write_release(self):
+        self._wval.value = 0
+        self._wcond.notify()
+        self._rcond.notify_all()
+        self._lock.release()
+
+    def acquire(self, count, exclusive_since=1):
+        if count < exclusive_since:
+            self.read_acquire()
+        else:
+            self.write_acquire()
+
+    def release(self, count, exclusive_since=1):
+        if count < exclusive_since:
+            self.read_release()
+        else:
+            self.write_release()