fix: Flush behavior

untitaker · untitaker · commit 1ca25468b6dd · 2019-02-07T14:13:19.000+01:00
diff --git a/sentry_sdk/worker.py b/sentry_sdk/worker.py
@@ -27,20 +27,6 @@ def _ensure_thread(self):
         if not self.is_alive:
             self.start()
 
-    def _timed_queue_join(self, timeout):
-        deadline = time() + timeout
-        queue = self._queue
-        queue.all_tasks_done.acquire()
-        try:
-            while queue.unfinished_tasks:
-                delay = deadline - time()
-                if delay <= 0:
-                    return False
-                queue.all_tasks_done.wait(timeout=delay)
-            return True
-        finally:
-            queue.all_tasks_done.release()
-
     def start(self):
         with self._lock:
             if not self.is_alive:
@@ -67,13 +53,15 @@ def flush(self, timeout, callback=None):
         logger.debug("background worker flushed")
 
     def _wait_flush(self, timeout, callback):
-        initial_timeout = min(0.1, timeout)
-        if not self._timed_queue_join(initial_timeout):
-            pending = self._queue.qsize()
-            logger.debug("%d event(s) pending on flush", pending)
-            if callback is not None:
-                callback(pending, timeout)
-            self._timed_queue_join(timeout - initial_timeout)
+        event = threading.Event()
+        pending = self._queue.qsize()
+        self._queue.put_nowait(event.set)
+        logger.debug("%d event(s) pending on flush", pending)
+
+        if callback is not None:
+            callback(pending, timeout)
+
+        event.wait(timeout)
 
     def submit(self, callback):
         self._ensure_thread()
diff --git a/tests/test_worker.py b/tests/test_worker.py
@@ -0,0 +1,31 @@
+import threading
+
+import time
+
+from sentry_sdk.worker import BackgroundWorker
+
+
+def test_flush_starvation(request):
+    worker = BackgroundWorker()
+    request.addfinalizer(worker.kill)
+
+    def send():
+        time.sleep(1)
+
+    worker.submit(send)
+
+    def spam_queue():
+        for _ in range(10):
+            time.sleep(0.1)
+            worker.submit(send)
+
+    t = threading.Thread(target=spam_queue)
+    t.daemon = True
+    t.start()
+
+    before = time.time()
+    worker.flush(10)
+    after = time.time()
+    duration = after - before
+
+    assert duration < 2
