We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN 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!

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `sentry_sdk/worker.py`
Content:
```
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 FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

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

{"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"}], "after_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(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\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        \"\"\"\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                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    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        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        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"}]}