fix: Fix deadlock in transport due to GC running

sentry_sdk/_compat.py

@@ -19,7 +19,6 @@
     import urlparse  # noqa
 
     text_type = unicode  # noqa
-    import Queue as queue  # noqa
 
     string_types = (str, text_type)
     number_types = (int, long, float)  # noqa
@@ -37,7 +36,6 @@ def implements_str(cls):
 
 else:
     import urllib.parse as urlparse  # noqa
-    import queue  # noqa
 
     text_type = str
     string_types = (text_type,)  # type: Tuple[type]

sentry_sdk/_queue.py

@@ -0,0 +1,222 @@
+"""
+A fork of Python 3.6's stdlib queue with Lock swapped out for RLock to avoid a
+deadlock while garbage collecting.
+
+See 
+https://codewithoutrules.com/2017/08/16/concurrency-python/
+https://bugs.python.org/issue14976
+https://github.com/sqlalchemy/sqlalchemy/blob/4eb747b61f0c1b1c25bdee3856d7195d10a0c227/lib/sqlalchemy/queue.py#L1
+
+We also vendor the code to evade eventlet's broken monkeypatching, see
+https://github.com/getsentry/sentry-python/pull/484
+"""
+
+try:
+    import threading
+except ImportError:
+    import dummy_threading as threading
+from collections import deque
+from heapq import heappush, heappop
+from time import monotonic as time
+
+__all__ = ['Empty', 'Full', 'Queue', 'PriorityQueue', 'LifoQueue']
+
+class Empty(Exception):
+    'Exception raised by Queue.get(block=0)/get_nowait().'
+    pass
+
+class Full(Exception):
+    'Exception raised by Queue.put(block=0)/put_nowait().'
+    pass
+
+class Queue(object):
+    '''Create a queue object with a given maximum size.
+
+    If maxsize is <= 0, the queue size is infinite.
+    '''
+
+    def __init__(self, maxsize=0):
+        self.maxsize = maxsize
+        self._init(maxsize)
+
+        # mutex must be held whenever the queue is mutating.  All methods
+        # that acquire mutex must release it before returning.  mutex
+        # is shared between the three conditions, so acquiring and
+        # releasing the conditions also acquires and releases mutex.
+        self.mutex = threading.RLock()
+
+        # Notify not_empty whenever an item is added to the queue; a
+        # thread waiting to get is notified then.
+        self.not_empty = threading.Condition(self.mutex)
+
+        # Notify not_full whenever an item is removed from the queue;
+        # a thread waiting to put is notified then.
+        self.not_full = threading.Condition(self.mutex)
+
+        # Notify all_tasks_done whenever the number of unfinished tasks
+        # drops to zero; thread waiting to join() is notified to resume
+        self.all_tasks_done = threading.Condition(self.mutex)
+        self.unfinished_tasks = 0
+
+    def task_done(self):
+        '''Indicate that a formerly enqueued task is complete.
+
+        Used by Queue consumer threads.  For each get() used to fetch a task,
+        a subsequent call to task_done() tells the queue that the processing
+        on the task is complete.
+
+        If a join() is currently blocking, it will resume when all items
+        have been processed (meaning that a task_done() call was received
+        for every item that had been put() into the queue).
+
+        Raises a ValueError if called more times than there were items
+        placed in the queue.
+        '''
+        with self.all_tasks_done:
+            unfinished = self.unfinished_tasks - 1
+            if unfinished <= 0:
+                if unfinished < 0:
+                    raise ValueError('task_done() called too many times')
+                self.all_tasks_done.notify_all()
+            self.unfinished_tasks = unfinished
+
+    def join(self):
+        '''Blocks until all items in the Queue have been gotten and processed.
+
+        The count of unfinished tasks goes up whenever an item is added to the
+        queue. The count goes down whenever a consumer thread calls task_done()
+        to indicate the item was retrieved and all work on it is complete.
+
+        When the count of unfinished tasks drops to zero, join() unblocks.
+        '''
+        with self.all_tasks_done:
+            while self.unfinished_tasks:
+                self.all_tasks_done.wait()
+
+    def qsize(self):
+        '''Return the approximate size of the queue (not reliable!).'''
+        with self.mutex:
+            return self._qsize()
+
+    def empty(self):
+        '''Return True if the queue is empty, False otherwise (not reliable!).
+
+        This method is likely to be removed at some point.  Use qsize() == 0
+        as a direct substitute, but be aware that either approach risks a race
+        condition where a queue can grow before the result of empty() or
+        qsize() can be used.
+
+        To create code that needs to wait for all queued tasks to be
+        completed, the preferred technique is to use the join() method.
+        '''
+        with self.mutex:
+            return not self._qsize()
+
+    def full(self):
+        '''Return True if the queue is full, False otherwise (not reliable!).
+
+        This method is likely to be removed at some point.  Use qsize() >= n
+        as a direct substitute, but be aware that either approach risks a race
+        condition where a queue can shrink before the result of full() or
+        qsize() can be used.
+        '''
+        with self.mutex:
+            return 0 < self.maxsize <= self._qsize()
+
+    def put(self, item, block=True, timeout=None):
+        '''Put an item into the queue.
+
+        If optional args 'block' is true and 'timeout' is None (the default),
+        block if necessary until a free slot is available. If 'timeout' is
+        a non-negative number, it blocks at most 'timeout' seconds and raises
+        the Full exception if no free slot was available within that time.
+        Otherwise ('block' is false), put an item on the queue if a free slot
+        is immediately available, else raise the Full exception ('timeout'
+        is ignored in that case).
+        '''
+        with self.not_full:
+            if self.maxsize > 0:
+                if not block:
+                    if self._qsize() >= self.maxsize:
+                        raise Full()
+                elif timeout is None:
+                    while self._qsize() >= self.maxsize:
+                        self.not_full.wait()
+                elif timeout < 0:
+                    raise ValueError("'timeout' must be a non-negative number")
+                else:
+                    endtime = time() + timeout
+                    while self._qsize() >= self.maxsize:
+                        remaining = endtime - time()
+                        if remaining <= 0.0:
+                            raise Full
+                        self.not_full.wait(remaining)
+            self._put(item)
+            self.unfinished_tasks += 1
+            self.not_empty.notify()
+
+    def get(self, block=True, timeout=None):
+        '''Remove and return an item from the queue.
+
+        If optional args 'block' is true and 'timeout' is None (the default),
+        block if necessary until an item is available. If 'timeout' is
+        a non-negative number, it blocks at most 'timeout' seconds and raises
+        the Empty exception if no item was available within that time.
+        Otherwise ('block' is false), return an item if one is immediately
+        available, else raise the Empty exception ('timeout' is ignored
+        in that case).
+        '''
+        with self.not_empty:
+            if not block:
+                if not self._qsize():
+                    raise Empty()
+            elif timeout is None:
+                while not self._qsize():
+                    self.not_empty.wait()
+            elif timeout < 0:
+                raise ValueError("'timeout' must be a non-negative number")
+            else:
+                endtime = time() + timeout
+                while not self._qsize():
+                    remaining = endtime - time()
+                    if remaining <= 0.0:
+                        raise Empty()
+                    self.not_empty.wait(remaining)
+            item = self._get()
+            self.not_full.notify()
+            return item
+
+    def put_nowait(self, item):
+        '''Put an item into the queue without blocking.
+
+        Only enqueue the item if a free slot is immediately available.
+        Otherwise raise the Full exception.
+        '''
+        return self.put(item, block=False)
+
+    def get_nowait(self):
+        '''Remove and return an item from the queue without blocking.
+
+        Only get an item if one is immediately available. Otherwise
+        raise the Empty exception.
+        '''
+        return self.get(block=False)
+
+    # Override these methods to implement other queue organizations
+    # (e.g. stack or priority queue).
+    # These will only be called with appropriate locks held
+
+    # Initialize the queue representation
+    def _init(self, maxsize):
+        self.queue = deque()
+
+    def _qsize(self):
+        return len(self.queue)
+
+    # Put a new item in the queue
+    def _put(self, item):
+        self.queue.append(item)
+
+    # Get an item from the queue
+    def _get(self):
+        return self.queue.popleft()

sentry_sdk/worker.py

@@ -1,14 +1,14 @@
 import os
+import threading
 
-from threading import Thread, Lock
 from time import sleep, time
-from sentry_sdk._compat import queue, check_thread_support
+from sentry_sdk._compat import check_thread_support
+from sentry_sdk._queue import Queue, Full
 from sentry_sdk.utils import logger
 
 from sentry_sdk._types import MYPY
 
 if MYPY:
-    from queue import Queue
     from typing import Any
     from typing import Optional
     from typing import Callable
@@ -18,12 +18,12 @@
 
 
 class BackgroundWorker(object):
-    def __init__(self):
-        # type: () -> None
+    def __init__(self, queue_size=30):
+        # type: (int) -> None
         check_thread_support()
-        self._queue = queue.Queue(30)  # type: Queue[Any]
-        self._lock = Lock()
-        self._thread = None  # type: Optional[Thread]
+        self._queue = Queue(queue_size)  # type: Queue[Any]
+        self._lock = threading.Lock()
+        self._thread = None  # type: Optional[threading.Thread]
         self._thread_for_pid = None  # type: Optional[int]
 
     @property
@@ -45,38 +45,24 @@ def _timed_queue_join(self, timeout):
         deadline = time() + timeout
         queue = self._queue
 
-        real_all_tasks_done = getattr(
-            queue, "all_tasks_done", None
-        )  # type: Optional[Any]
-        if real_all_tasks_done is not None:
-            real_all_tasks_done.acquire()
-            all_tasks_done = real_all_tasks_done  # type: Optional[Any]
-        elif queue.__module__.startswith("eventlet."):
-            all_tasks_done = getattr(queue, "_cond", None)
-        else:
-            all_tasks_done = None
+        queue.all_tasks_done.acquire()
 
         try:
             while queue.unfinished_tasks:
                 delay = deadline - time()
                 if delay <= 0:
                     return False
-                if all_tasks_done is not None:
-                    all_tasks_done.wait(timeout=delay)
-                else:
-                    # worst case, we just poll the number of remaining tasks
-                    sleep(0.1)
+                queue.all_tasks_done.wait(timeout=delay)
 
             return True
         finally:
-            if real_all_tasks_done is not None:
-                real_all_tasks_done.release()
+            queue.all_tasks_done.release()
 
     def start(self):
         # type: () -> None
         with self._lock:
             if not self.is_alive:
-                self._thread = Thread(
+                self._thread = threading.Thread(
                     target=self._target, name="raven-sentry.BackgroundWorker"
                 )
                 self._thread.setDaemon(True)
@@ -94,7 +80,7 @@ def kill(self):
             if self._thread:
                 try:
                     self._queue.put_nowait(_TERMINATOR)
-                except queue.Full:
+                except Full:
                     logger.debug("background worker queue full, kill failed")
 
                 self._thread = None
@@ -123,7 +109,7 @@ def submit(self, callback):
         self._ensure_thread()
         try:
             self._queue.put_nowait(callback)
-        except queue.Full:
+        except Full:
             logger.debug("background worker queue full, dropping event")
 
     def _target(self):