Fix runner timeout

Decided to put thread instance to timeout_queue because
thread.ident in python is recycled. So this is the way to
check whether the thread alive or not.

Closes-bug: #1570798

Change-Id: I745030ec9942c37b47c6d53864ae5fc7ecae7ccd

rally-jobs/rally.yaml

@@ -526,6 +526,31 @@
       sla:
         failure_rate:
           max: 0
+    -
+      args:
+        sleep: 0.6
+      runner:
+        type: "rps"
+        rps: 2
+        times: 5
+        timeout: 1
+      context:
+        users:
+          tenants: 1
+          users_per_tenant: 1
+    -
+      args:
+        sleep: 0.6
+      runner:
+        type: "constant"
+        concurrency: 2
+        times: 4
+        timeout: 1
+      context:
+        users:
+          tenants: 1
+          users_per_tenant: 1
+
 
   Dummy.dummy_exception:
     -

rally/common/utils.py

@@ -530,16 +530,17 @@ def timeout_thread(queue):
         if not all_threads:
             timeout = None
         else:
-            thread_ident, deadline = all_threads[0]
+            thread, deadline = all_threads[0]
             timeout = deadline - time.time()
         try:
             next_thread = queue.get(timeout=timeout)
             all_threads.append(next_thread)
         except (moves.queue.Empty, ValueError):
             # NOTE(rvasilets) Empty means that timeout was occurred.
-            # ValueError means that timeout lower then 0.
-            LOG.info("Thread %s is timed out. Terminating." % thread_ident)
-            terminate_thread(thread_ident)
+            # ValueError means that timeout lower than 0.
+            if thread.isAlive():
+                LOG.info("Thread %s is timed out. Terminating." % thread.ident)
+                terminate_thread(thread.ident)
             all_threads.popleft()
 
         if next_thread == (None, None,):

rally/plugins/common/runners/constant.py

@@ -79,7 +79,7 @@ def _worker_process(queue, iteration_gen, timeout, concurrency, times,
 
         thread.start()
         if timeout:
-            timeout_queue.put((thread.ident, time.time() + timeout))
+            timeout_queue.put((thread, time.time() + timeout))
         pool.append(thread)
         alive_threads_in_pool += 1
 

rally/plugins/common/runners/rps.py

@@ -82,7 +82,7 @@ def _worker_process(queue, iteration_gen, timeout, rps, times,
         i += 1
         thread.start()
         if timeout:
-            timeout_queue.put((thread.ident, time.time() + timeout))
+            timeout_queue.put((thread, time.time() + timeout))
         pool.append(thread)
 
         time_gap = time.time() - start
@@ -102,8 +102,7 @@ def _worker_process(queue, iteration_gen, timeout, rps, times,
                 time.sleep(0.001)
 
     while pool:
-        thr = pool.popleft()
-        thr.join()
+        pool.popleft().join()
 
     if timeout:
         timeout_queue.put((None, None,))

tests/unit/common/test_utils.py

@@ -472,7 +472,7 @@ class TimeoutThreadTestCase(test.TestCase):
         )
         test_thread.start()
         start_time = time.time()
-        queue.put((test_thread.ident, start_time + 1))
+        queue.put((test_thread, start_time + 1))
         killer_thread.start()
         test_thread.join()
         end_time = time.time()