prompt: You will be provided with a partial code base and an issue statement explaining a problem to resolve.
  <issue>
  Memory leak introduced in 0.12.3
  Dear maintainers,
  
  I recently noticed that my application containers started going out of memory fairly quickly. Upon investigation, I found that recent upgrade of sentry-sdk was causing this.
  
  Tested with a few versions of sentry-sdk using Python 3.7.6:
  sentry-sdk==0.12.2 -> memory usage stays at ~70 MiB, does not raise
  
  sentry-sdk==0.12.3
  sentry-sdk==0.13.5
  sentry-sdk @ master
  -> memory usage starts from ~70MiB and raises steadily until reaching memory limit and crashes container.
  
  Please see if you can reproduce this behaviour. If not, I'll try to put up a minimal reproducible example, but this should be fairly simple to see. My application does not do anything special really, just receives some millions of messages over MQ and out of those, some few thousands end up getting logged to Sentry due to exceptions occurring. This is using pretty much the default Sentry config, no integrations.
  
  Thanks for your help!
  
  </issue>
  <code>
  [start of sentry_sdk/worker.py]
  1 import os
  2 
  3 from threading import Thread, Lock
  4 from time import sleep, time
  5 from sentry_sdk._compat import queue, check_thread_support
  6 from sentry_sdk.utils import logger
  7 
  8 
  9 from sentry_sdk._types import MYPY
  10 
  11 if MYPY:
  12     from queue import Queue
  13     from typing import Any
  14     from typing import Optional
  15     from typing import Callable
  16 
  17 
  18 _TERMINATOR = object()
  19 
  20 
  21 class BackgroundWorker(object):
  22     def __init__(self):
  23         # type: () -> None
  24         check_thread_support()
  25         self._queue = queue.Queue(-1)  # type: Queue[Any]
  26         self._lock = Lock()
  27         self._thread = None  # type: Optional[Thread]
  28         self._thread_for_pid = None  # type: Optional[int]
  29 
  30     @property
  31     def is_alive(self):
  32         # type: () -> bool
  33         if self._thread_for_pid != os.getpid():
  34             return False
  35         if not self._thread:
  36             return False
  37         return self._thread.is_alive()
  38 
  39     def _ensure_thread(self):
  40         # type: () -> None
  41         if not self.is_alive:
  42             self.start()
  43 
  44     def _timed_queue_join(self, timeout):
  45         # type: (float) -> bool
  46         deadline = time() + timeout
  47         queue = self._queue
  48 
  49         real_all_tasks_done = getattr(
  50             queue, "all_tasks_done", None
  51         )  # type: Optional[Any]
  52         if real_all_tasks_done is not None:
  53             real_all_tasks_done.acquire()
  54             all_tasks_done = real_all_tasks_done  # type: Optional[Any]
  55         elif queue.__module__.startswith("eventlet."):
  56             all_tasks_done = getattr(queue, "_cond", None)
  57         else:
  58             all_tasks_done = None
  59 
  60         try:
  61             while queue.unfinished_tasks:
  62                 delay = deadline - time()
  63                 if delay <= 0:
  64                     return False
  65                 if all_tasks_done is not None:
  66                     all_tasks_done.wait(timeout=delay)
  67                 else:
  68                     # worst case, we just poll the number of remaining tasks
  69                     sleep(0.1)
  70 
  71             return True
  72         finally:
  73             if real_all_tasks_done is not None:
  74                 real_all_tasks_done.release()
  75 
  76     def start(self):
  77         # type: () -> None
  78         with self._lock:
  79             if not self.is_alive:
  80                 self._thread = Thread(
  81                     target=self._target, name="raven-sentry.BackgroundWorker"
  82                 )
  83                 self._thread.setDaemon(True)
  84                 self._thread.start()
  85                 self._thread_for_pid = os.getpid()
  86 
  87     def kill(self):
  88         # type: () -> None
  89         logger.debug("background worker got kill request")
  90         with self._lock:
  91             if self._thread:
  92                 self._queue.put_nowait(_TERMINATOR)
  93                 self._thread = None
  94                 self._thread_for_pid = None
  95 
  96     def flush(self, timeout, callback=None):
  97         # type: (float, Optional[Any]) -> None
  98         logger.debug("background worker got flush request")
  99         with self._lock:
  100             if self.is_alive and timeout > 0.0:
  101                 self._wait_flush(timeout, callback)
  102         logger.debug("background worker flushed")
  103 
  104     def _wait_flush(self, timeout, callback):
  105         # type: (float, Optional[Any]) -> None
  106         initial_timeout = min(0.1, timeout)
  107         if not self._timed_queue_join(initial_timeout):
  108             pending = self._queue.qsize()
  109             logger.debug("%d event(s) pending on flush", pending)
  110             if callback is not None:
  111                 callback(pending, timeout)
  112             self._timed_queue_join(timeout - initial_timeout)
  113 
  114     def submit(self, callback):
  115         # type: (Callable[[], None]) -> None
  116         self._ensure_thread()
  117         self._queue.put_nowait(callback)
  118 
  119     def _target(self):
  120         # type: () -> None
  121         while True:
  122             callback = self._queue.get()
  123             try:
  124                 if callback is _TERMINATOR:
  125                     break
  126                 try:
  127                     callback()
  128                 except Exception:
  129                     logger.error("Failed processing job", exc_info=True)
  130             finally:
  131                 self._queue.task_done()
  132             sleep(0)
  133
  [end of sentry_sdk/worker.py]
  </code>
  
  I need you to solve this issue by generating a single patch file that I can apply directly to this repository using git apply. Please respond with a single patch file in the following format.
  <patch>
  diff --git a/file.py b/file.py
  --- a/file.py
  +++ b/file.py
  @@ -1,27 +1,35 @@
   def euclidean(a, b):
  -    while b:
  -        a, b = b, a % b
  -    return a
  +    if b == 0:
  +        return a
  +    return euclidean(b, a % b)
   
   
   def bresenham(x0, y0, x1, y1):
       points = []
       dx = abs(x1 - x0)
       dy = abs(y1 - y0)
  -    sx = 1 if x0 < x1 else -1
  -    sy = 1 if y0 < y1 else -1
  -    err = dx - dy
  +    x, y = x0, y0
  +    sx = -1 if x0 > x1 else 1
  +    sy = -1 if y0 > y1 else 1
   
  -    while True:
  -        points.append((x0, y0))
  -        if x0 == x1 and y0 == y1:
  -            break
  -        e2 = 2 * err
  -        if e2 > -dy:
  -            err -= dy
  -            x0 += sx
  -        if e2 < dx:
  -            err += dx
  -            y0 += sy
  +    if dx > dy:
  +        err = dx / 2.0
  +        while x != x1:
  +            points.append((x, y))
  +            err -= dy
  +            if err < 0:
  +                y += sy
  +                err += dx
  +            x += sx
  +    else:
  +        err = dy / 2.0
  +        while y != y1:
  +            points.append((x, y))
  +            err -= dx
  +            if err < 0:
  +                x += sx
  +                err += dy
  +            y += sy
  +
  +    points.append((x, y))
       return points
  
  </patch>
golden_diff: diff --git a/sentry_sdk/worker.py b/sentry_sdk/worker.py
  --- a/sentry_sdk/worker.py
  +++ b/sentry_sdk/worker.py
  @@ -22,7 +22,7 @@
       def __init__(self):
           # type: () -> None
           check_thread_support()
  -        self._queue = queue.Queue(-1)  # type: Queue[Any]
  +        self._queue = queue.Queue(30)  # type: Queue[Any]
           self._lock = Lock()
           self._thread = None  # type: Optional[Thread]
           self._thread_for_pid = None  # type: Optional[int]
  @@ -86,10 +86,18 @@
   
       def kill(self):
           # type: () -> None
  +        """
  +        Kill worker thread. Returns immediately. Not useful for
  +        waiting on shutdown for events, use `flush` for that.
  +        """
           logger.debug("background worker got kill request")
           with self._lock:
               if self._thread:
  -                self._queue.put_nowait(_TERMINATOR)
  +                try:
  +                    self._queue.put_nowait(_TERMINATOR)
  +                except queue.Full:
  +                    logger.debug("background worker queue full, kill failed")
  +
                   self._thread = None
                   self._thread_for_pid = None
   
  @@ -114,7 +122,10 @@
       def submit(self, callback):
           # type: (Callable[[], None]) -> None
           self._ensure_thread()
  -        self._queue.put_nowait(callback)
  +        try:
  +            self._queue.put_nowait(callback)
  +        except queue.Full:
  +            logger.debug("background worker queue full, dropping event")
   
       def _target(self):
           # type: () -> None
  
verification_info: {"golden_diff": "diff --git a/sentry_sdk/worker.py b/sentry_sdk/worker.py\n--- a/sentry_sdk/worker.py\n+++ b/sentry_sdk/worker.py\n@@ -22,7 +22,7 @@\n     def __init__(self):\n         # type: () -> None\n         check_thread_support()\n-        self._queue = queue.Queue(-1)  # type: Queue[Any]\n+        self._queue = queue.Queue(30)  # type: Queue[Any]\n         self._lock = Lock()\n         self._thread = None  # type: Optional[Thread]\n         self._thread_for_pid = None  # type: Optional[int]\n@@ -86,10 +86,18 @@\n \n     def kill(self):\n         # type: () -> None\n+        \"\"\"\n+        Kill worker thread. Returns immediately. Not useful for\n+        waiting on shutdown for events, use `flush` for that.\n+        \"\"\"\n         logger.debug(\"background worker got kill request\")\n         with self._lock:\n             if self._thread:\n-                self._queue.put_nowait(_TERMINATOR)\n+                try:\n+                    self._queue.put_nowait(_TERMINATOR)\n+                except queue.Full:\n+                    logger.debug(\"background worker queue full, kill failed\")\n+\n                 self._thread = None\n                 self._thread_for_pid = None\n \n@@ -114,7 +122,10 @@\n     def submit(self, callback):\n         # type: (Callable[[], None]) -> None\n         self._ensure_thread()\n-        self._queue.put_nowait(callback)\n+        try:\n+            self._queue.put_nowait(callback)\n+        except queue.Full:\n+            logger.debug(\"background worker queue full, dropping event\")\n \n     def _target(self):\n         # type: () -> None\n", "issue": "Memory leak introduced in 0.12.3\nDear maintainers,\r\n\r\nI recently noticed that my application containers started going out of memory fairly quickly. Upon investigation, I found that recent upgrade of sentry-sdk was causing this.\r\n\r\nTested with a few versions of sentry-sdk using Python 3.7.6:\r\nsentry-sdk==0.12.2 -> memory usage stays at ~70 MiB, does not raise\r\n\r\nsentry-sdk==0.12.3\r\nsentry-sdk==0.13.5\r\nsentry-sdk @ master\r\n-> memory usage starts from ~70MiB and raises steadily until reaching memory limit and crashes container.\r\n\r\nPlease see if you can reproduce this behaviour. If not, I'll try to put up a minimal reproducible example, but this should be fairly simple to see. My application does not do anything special really, just receives some millions of messages over MQ and out of those, some few thousands end up getting logged to Sentry due to exceptions occurring. This is using pretty much the default Sentry config, no integrations.\r\n\r\nThanks for your help!\n", "before_files": [{"content": "import os\n\nfrom threading import Thread, Lock\nfrom time import sleep, time\nfrom sentry_sdk._compat import queue, check_thread_support\nfrom sentry_sdk.utils import logger\n\n\nfrom sentry_sdk._types import MYPY\n\nif MYPY:\n    from queue import Queue\n    from typing import Any\n    from typing import Optional\n    from typing import Callable\n\n\n_TERMINATOR = object()\n\n\nclass BackgroundWorker(object):\n    def __init__(self):\n        # type: () -> None\n        check_thread_support()\n        self._queue = queue.Queue(-1)  # type: Queue[Any]\n        self._lock = Lock()\n        self._thread = None  # type: Optional[Thread]\n        self._thread_for_pid = None  # type: Optional[int]\n\n    @property\n    def is_alive(self):\n        # type: () -> bool\n        if self._thread_for_pid != os.getpid():\n            return False\n        if not self._thread:\n            return False\n        return self._thread.is_alive()\n\n    def _ensure_thread(self):\n        # type: () -> None\n        if not self.is_alive:\n            self.start()\n\n    def _timed_queue_join(self, timeout):\n        # type: (float) -> bool\n        deadline = time() + timeout\n        queue = self._queue\n\n        real_all_tasks_done = getattr(\n            queue, \"all_tasks_done\", None\n        )  # type: Optional[Any]\n        if real_all_tasks_done is not None:\n            real_all_tasks_done.acquire()\n            all_tasks_done = real_all_tasks_done  # type: Optional[Any]\n        elif queue.__module__.startswith(\"eventlet.\"):\n            all_tasks_done = getattr(queue, \"_cond\", None)\n        else:\n            all_tasks_done = None\n\n        try:\n            while queue.unfinished_tasks:\n                delay = deadline - time()\n                if delay <= 0:\n                    return False\n                if all_tasks_done is not None:\n                    all_tasks_done.wait(timeout=delay)\n                else:\n                    # worst case, we just poll the number of remaining tasks\n                    sleep(0.1)\n\n            return True\n        finally:\n            if real_all_tasks_done is not None:\n                real_all_tasks_done.release()\n\n    def start(self):\n        # type: () -> None\n        with self._lock:\n            if not self.is_alive:\n                self._thread = Thread(\n                    target=self._target, name=\"raven-sentry.BackgroundWorker\"\n                )\n                self._thread.setDaemon(True)\n                self._thread.start()\n                self._thread_for_pid = os.getpid()\n\n    def kill(self):\n        # type: () -> None\n        logger.debug(\"background worker got kill request\")\n        with self._lock:\n            if self._thread:\n                self._queue.put_nowait(_TERMINATOR)\n                self._thread = None\n                self._thread_for_pid = None\n\n    def flush(self, timeout, callback=None):\n        # type: (float, Optional[Any]) -> None\n        logger.debug(\"background worker got flush request\")\n        with self._lock:\n            if self.is_alive and timeout > 0.0:\n                self._wait_flush(timeout, callback)\n        logger.debug(\"background worker flushed\")\n\n    def _wait_flush(self, timeout, callback):\n        # type: (float, Optional[Any]) -> None\n        initial_timeout = min(0.1, timeout)\n        if not self._timed_queue_join(initial_timeout):\n            pending = self._queue.qsize()\n            logger.debug(\"%d event(s) pending on flush\", pending)\n            if callback is not None:\n                callback(pending, timeout)\n            self._timed_queue_join(timeout - initial_timeout)\n\n    def submit(self, callback):\n        # type: (Callable[[], None]) -> None\n        self._ensure_thread()\n        self._queue.put_nowait(callback)\n\n    def _target(self):\n        # type: () -> None\n        while True:\n            callback = self._queue.get()\n            try:\n                if callback is _TERMINATOR:\n                    break\n                try:\n                    callback()\n                except Exception:\n                    logger.error(\"Failed processing job\", exc_info=True)\n            finally:\n                self._queue.task_done()\n            sleep(0)\n", "path": "sentry_sdk/worker.py"}]}