Rerun failed fragile test

The patch is a bit dirty: tasks to rerun are added after 'stop worker'
marker, so old worker is stopped and the new one is spawned to handle
those tasks. See new FIXME comment for more information.

How to verify:

 | # Ensure a test is marked as fragile.
 | $ grep tarantoolctl.test.lua test/app-tap/suite.ini
 | fragile = tarantoolctl.test.lua     ; gh-5059
 |
 | # Let the test fail randomly.
 | $ git diff
 | --- a/test/app-tap/tarantoolctl.test.lua
 | +++ b/test/app-tap/tarantoolctl.test.lua
 | @@ -643,4 +643,24 @@ test:test('filter_xlog', function(test)
 |      end
 |  end)
 |
 | -os.exit(test:check() == true and 0 or -1)
 | +local function string_to_unsigned(str)
 | +    local byte_list = {string.byte(str, 1, #str)}
 | +    local res = 0
 | +    for i = 1, #byte_list do
 | +        res = bit.lshift(res, 8)
 | +        res = bit.bor(res, byte_list[i])
 | +    end
 | +    -- fix bitop's numbers range: signed 32-bit to unsigned 32-bit
 | +    -- http://bitop.luajit.org/semantics.html
 | +    res = (res < 0) and -res + 0x7fffffff or res
 | +    return res
 | +end
 | +
 | +local RND_SEED_LEN = 4
 | +local fh = require('fio').open('/dev/urandom', {'O_RDONLY'})
 | +local seed_raw = fh:read(RND_SEED_LEN)
 | +fh:close()
 | +local seed = string_to_unsigned(seed_raw)
 | +math.randomseed(seed)
 | +
 | +os.exit(test:check() == true and math.random(0, 1) or 1)
 |
 | # Run several times to observe actual behaviour.
 | $ (cd test && ./test-run.py app-tap/tarantoolctl.test.lua)

Part of tarantool/tarantool#5050

Author: Totktonada

dispatcher.py

@@ -130,13 +130,15 @@ def init_listeners(self):
             not args.valgrind
         watch_fail = not lib.Options().args.is_force
 
+        rerun_watcher = listeners.RerunWatcher(self.can_add_task, self.add_task)
         log_output_watcher = listeners.LogOutputWatcher()
         self.statistics = listeners.StatisticsWatcher(
             log_output_watcher.get_logfile)
         self.artifacts = listeners.ArtifactsWatcher(
             log_output_watcher.get_logfile)
         output_watcher = listeners.OutputWatcher()
-        self.listeners = [self.statistics, log_output_watcher, output_watcher, self.artifacts]
+        self.listeners = [rerun_watcher, self.statistics, log_output_watcher,
+                          output_watcher, self.artifacts]
         if watch_fail:
             self.fail_watcher = listeners.FailWatcher(
                 self.terminate_all_workers)
@@ -171,6 +173,8 @@ def find_nonempty_task_queue_disp(self):
         for task_queue_disp in task_queue_disps_rnd:
             if not task_queue_disp.is_parallel:
                 continue
+            # Don't add a worker when a task list is exhausted.
+            # Some tasks may be running at the moment.
             if task_queue_disp.done:
                 continue
             return task_queue_disp
@@ -179,7 +183,18 @@ def find_nonempty_task_queue_disp(self):
         for task_queue_disp in task_queue_disps_rnd:
             if len(task_queue_disp.worker_ids) > 0:
                 continue
-            if task_queue_disp.done:
+            # Allow to add a worker even if there was a point when
+            # all tasks were taken. It is possible that a task
+            # will be added afterwards.
+            #
+            # FIXME: It is a kind of hack. A worker may read
+            # 'stop worker' marker and stop, but than we'll start
+            # a new worker, which will handle reruns. We should
+            # keep a worker running while there is a possibility
+            # that a task will be added again to the queue.
+            #
+            # NB: Don't forget to handle 'not is_force' case.
+            if not task_queue_disp.undone_tasks():
                 continue
             return task_queue_disp
         return None
@@ -234,6 +249,20 @@ def del_worker(self, worker_id):
                 self.processes.remove(process)
                 break
 
+    def can_add_task(self, worker_id):
+        """Whether given task group (determined by the worker id)
+           allows to add a task dynamically (during tasks
+           processing).
+
+           It is prerequisite to call <add_task>().
+        """
+        task_queue_disp = self.get_task_queue_disp(worker_id)
+        return task_queue_disp.can_add_task()
+
+    def add_task(self, worker_id, task_id):
+        task_queue_disp = self.get_task_queue_disp(worker_id)
+        task_queue_disp.add_task(task_id)
+
     def mark_task_done(self, worker_id, task_id):
         task_queue_disp = self.get_task_queue_disp(worker_id)
         task_queue_disp.mark_task_done(task_id)
@@ -277,7 +306,8 @@ def wait(self):
 
             objs = self.invoke_listeners(inputs, ready_inputs)
             for obj in objs:
-                if isinstance(obj, WorkerTaskResult):
+                if isinstance(obj, WorkerTaskResult) and \
+                        obj.short_status != 'transient fail':
                     self.mark_task_done(obj.worker_id, obj.task_id)
                 elif isinstance(obj, WorkerDone):
                     self.del_worker(obj.worker_id)
@@ -409,6 +439,12 @@ def del_worker(self, worker_id):
         # with add-del workers
         self.done = True
 
+    def can_add_task(self):
+        return self.key.endswith('_fragile')
+
+    def add_task(self, task_id):
+        self.task_queue.put(task_id)
+
     def mark_task_done(self, task_id):
         self.done_task_ids.add(task_id)
 

listeners.py

@@ -28,6 +28,43 @@ def process_timeout(self, delta_seconds):
         pass
 
 
+class RerunWatcher(BaseWatcher):
+    def __init__(self, can_add_task, add_task):
+        self._can_add_task = can_add_task
+        self._add_task = add_task
+        self._fail_counters = dict()
+        self._fail_limit = 3
+
+    def _count_fail(self, obj):
+        if obj.task_id not in self._fail_counters:
+            self._fail_counters[obj.task_id] = 0
+        self._fail_counters[obj.task_id] += 1
+
+    def _can_rerun(self, task_id):
+        return self._fail_counters[task_id] < self._fail_limit
+
+    def process_result(self, obj):
+        if not isinstance(obj, WorkerTaskResult):
+            return
+
+        if obj.short_status != 'fail':
+            return
+
+        if not self._can_add_task(obj.worker_id):
+            return
+
+        self._count_fail(obj)
+
+        if self._can_rerun(obj.task_id):
+            color_stdout('Schedule task %s to run again\n' % str(obj.task_id),
+                         schema='test_var')
+            obj.short_status = 'transient fail'
+            self._add_task(obj.worker_id, obj.task_id)
+        else:
+            color_stdout('Task %s reaches fail limit (%d)\n' %
+                         (str(obj.task_id), self._fail_limit), schema='test_fail')
+
+
 class StatisticsWatcher(BaseWatcher):
     def __init__(self, get_logfile):
         self.stats = dict()