moved failed property to each task

Author: chillaq

splitio/sync/segment.py

@@ -360,8 +360,7 @@ async def synchronize_segments(self, segment_names = None, dont_wait = False):
         jobs = await self._worker_pool.submit_work(segment_names)
         if (dont_wait):
             return True
-        await jobs.await_completion()
-        return not self._worker_pool.pop_failed()
+        return await jobs.await_completion()
 
     async def segment_exist_in_storage(self, segment_name):
         """

splitio/tasks/util/workerpool.py

@@ -152,7 +152,6 @@ def __init__(self, worker_count, worker_func):
         self._queue = asyncio.Queue()
         self._handler = worker_func
         self._aborted = False
-        self._failed = False
 
     async def _schedule_work(self):
         """wrap the message handler execution."""
@@ -173,7 +172,7 @@ async def _do_work(self, message):
         except Exception:
             _LOGGER.error("Something went wrong when processing message %s", message)
             _LOGGER.debug('Original traceback: ', exc_info=True)
-            self._failed = True
+            message._failed = True
         message._complete.set()
         self._semaphore.release() # signal worker is idle
 
@@ -204,17 +203,13 @@ async def stop(self, event=None):
         """abort all execution (except currently running handlers)."""
         await self._queue.put(self._abort)
 
-    def pop_failed(self):
-        old = self._failed
-        self._failed = False
-        return old
-
 
 class TaskCompletionWraper:
     """Task completion class"""
     def __init__(self, message):
         self._message = message
         self._complete = asyncio.Event()
+        self._failed = False
 
     async def await_completion(self):
         await self._complete.wait()
@@ -230,3 +225,7 @@ def __init__(self, tasks):
 
     async def await_completion(self):
         await asyncio.gather(*[task.await_completion() for task in self._tasks])
+        for task in self._tasks:
+            if task._failed:
+                return False
+        return True

tests/tasks/util/test_workerpool.py

@@ -94,11 +94,10 @@ async def worker_func(num):
             jobs.append(str(num))
 
         task = await wpool.submit_work(jobs)
-        await task.await_completion()
+        assert await task.await_completion()
         await wpool.stop()
         for num in range(0, 11):
             assert str(num) in calls
-        assert not wpool.pop_failed()
 
     @pytest.mark.asyncio
     async def test_fail_in_msg_doesnt_break(self):
@@ -115,11 +114,13 @@ async def do_work(self, work):
         worker = Worker()
         wpool = workerpool.WorkerPoolAsync(50, worker.do_work)
         wpool.start()
+        jobs = []
         for num in range(0, 100):
-            await wpool.submit_work([str(num)])
-        await asyncio.sleep(1)
+            jobs.append(str(num))
+        task = await wpool.submit_work(jobs)
+
+        assert not await task.await_completion()
         await wpool.stop()
-        assert wpool.pop_failed()
 
         for num in range(0, 100):
             if num != 55:
@@ -145,6 +146,6 @@ async def do_work(self, work):
         for num in range(0, 100):
             jobs.append(str(num))
         task = await wpool.submit_work(jobs)
-        await task.await_completion()
+        assert await task.await_completion()
         await wpool.stop()
         assert len(worker.worked) == 100