Merge pull request #153 from splitio/bugfix/race_condition_on_init

fix race condition

Author: mredolatti

CHANGES.txt

@@ -1,3 +1,6 @@
+8.1.3 (Oct 4, 2019)
+ - Fixed race condition related to segment fetching and SDK_READY event
+
 8.1.2 (Jul 19, 2019)
  - Validated TLS support for redis connections
  - Fixed traffic type count issue

splitio/tasks/util/workerpool.py

@@ -63,15 +63,18 @@ def _wrapper(self, worker_number, func):
         while self._should_be_working[worker_number]:
             try:
                 message = self._incoming.get(True, 0.5)
-                self._incoming.task_done()
 
                 # For some reason message can be None in python2 implementation of queue.
                 # This method must be both ignored and acknowledged with .task_done()
                 # otherwise .join() will halt.
                 if message is None:
+                    self._incoming.task_done()
                     continue
 
+                # If the task is successfully executed, the ack is done AFTERWARDS,
+                # to avoid race conditions on SDK initialization.
                 ok = self._safe_run(func, message)  #pylint: disable=invalid-name
+                self._incoming.task_done()
                 if not ok:
                     self._logger.error(
                         ("Something went wrong during the execution, "

splitio/version.py

@@ -1 +1 @@
-__version__ = '8.1.2'
+__version__ = '8.1.3'

tests/tasks/util/test_workerpool.py

@@ -1,4 +1,5 @@
 """Workerpool test module."""
+# pylint: disable=no-self-use,too-few-public-methods,missing-docstring
 import time
 import threading
 from splitio.tasks.util import workerpool
@@ -10,44 +11,63 @@ class WorkerPoolTests(object):
     def test_normal_operation(self, mocker):
         """Test normal opeation works properly."""
         worker_func = mocker.Mock()
-        wp = workerpool.WorkerPool(10, worker_func)
-        wp.start()
-        for x in range(0, 100):
-            wp.submit_work(str(x))
+        wpool = workerpool.WorkerPool(10, worker_func)
+        wpool.start()
+        for num in range(0, 100):
+            wpool.submit_work(str(num))
 
         stop_event = threading.Event()
-        wp.stop(stop_event)
+        wpool.stop(stop_event)
         stop_event.wait(5)
         assert stop_event.is_set()
 
         calls = worker_func.mock_calls
-        for x in range(0, 100):
-            assert mocker.call(str(x)) in calls
+        for num in range(0, 100):
+            assert mocker.call(str(num)) in calls
 
-    def test_failure_in_message_doesnt_breal(self, mocker):
+    def test_fail_in_msg_doesnt_break(self):
         """Test that if a message cannot be parsed it is ignored and others are processed."""
-        class Worker:
+        class Worker(object):  #pylint: disable=
             def __init__(self):
-                self._worked = set()
+                self.worked = set()
 
-            def do_work(self, w):
-                if w == '55':
+            def do_work(self, work):
+                if work == '55':
                     raise Exception('something')
-                self._worked.add(w)
+                self.worked.add(work)
 
         worker = Worker()
-        wp = workerpool.WorkerPool(50, worker.do_work)
-        wp.start()
-        for x in range(0, 100):
-            wp.submit_work(str(x))
+        wpool = workerpool.WorkerPool(50, worker.do_work)
+        wpool.start()
+        for num in range(0, 100):
+            wpool.submit_work(str(num))
 
         stop_event = threading.Event()
-        wp.stop(stop_event)
+        wpool.stop(stop_event)
         stop_event.wait(5)
         assert stop_event.is_set()
 
-        for x in range(0, 100):
-            if x != 55:
-                assert str(x) in worker._worked
+        for num in range(0, 100):
+            if num != 55:
+                assert str(num) in worker.worked
             else:
-                assert str(x) not in worker._worked
+                assert str(num) not in worker.worked
+
+    def test_msg_acked_after_processed(self):
+        """Test that events are only set after all the work in the pipeline is done."""
+        class Worker(object):
+            def __init__(self):
+                self.worked = set()
+
+            def do_work(self, work):
+                self.worked.add(work)
+                time.sleep(0.02)  # will wait 2 seconds in total for 100 elements
+
+        worker = Worker()
+        wpool = workerpool.WorkerPool(50, worker.do_work)
+        wpool.start()
+        for num in range(0, 100):
+            wpool.submit_work(str(num))
+
+        wpool.wait_for_completion()
+        assert len(worker.worked) == 100