prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  Celery - Queue object has no attribute  'all_tasks_done'
  Hi all, 
  
  I'm integrating Sentry on a project in python that uses Celery. I'm getting this error when shutting down the worker: 
  
  ```
  Error in atexit._run_exitfuncs:
  Traceback (most recent call last):
    File "/Users/jibanez/API/.conda/envs/cimrender/lib/python3.6/site-packages/sentry_sdk/worker.py", line 84, in flush
      self._wait_flush(timeout, callback)
    File "/Users/jibanez/API/.conda/envs/cimrender/lib/python3.6/site-packages/sentry_sdk/worker.py", line 90, in _wait_flush
      if not self._timed_queue_join(initial_timeout):
    File "/Users/jibanez/API/.conda/envs/cimrender/lib/python3.6/site-packages/sentry_sdk/worker.py", line 48, in _timed_queue_join
      queue.all_tasks_done.acquire()  # type: ignore
  AttributeError: 'Queue' object has no attribute 'all_tasks_done'
  ```
  
  I'm using: 
  - Python 3.6
  - Celery 4.3.0
  - OSX Mojave
  
  Any thoughts? 
  
  --- 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         queue.all_tasks_done.acquire()  # type: ignore
  49         try:
  50             while queue.unfinished_tasks:  # type: ignore
  51                 delay = deadline - time()
  52                 if delay <= 0:
  53                     return False
  54                 queue.all_tasks_done.wait(timeout=delay)  # type: ignore
  55             return True
  56         finally:
  57             queue.all_tasks_done.release()  # type: ignore
  58 
  59     def start(self):
  60         # type: () -> None
  61         with self._lock:
  62             if not self.is_alive:
  63                 self._thread = Thread(
  64                     target=self._target, name="raven-sentry.BackgroundWorker"
  65                 )
  66                 self._thread.setDaemon(True)
  67                 self._thread.start()
  68                 self._thread_for_pid = os.getpid()
  69 
  70     def kill(self):
  71         # type: () -> None
  72         logger.debug("background worker got kill request")
  73         with self._lock:
  74             if self._thread:
  75                 self._queue.put_nowait(_TERMINATOR)
  76                 self._thread = None
  77                 self._thread_for_pid = None
  78 
  79     def flush(self, timeout, callback=None):
  80         # type: (float, Optional[Any]) -> None
  81         logger.debug("background worker got flush request")
  82         with self._lock:
  83             if self.is_alive and timeout > 0.0:
  84                 self._wait_flush(timeout, callback)
  85         logger.debug("background worker flushed")
  86 
  87     def _wait_flush(self, timeout, callback):
  88         # type: (float, Optional[Any]) -> None
  89         initial_timeout = min(0.1, timeout)
  90         if not self._timed_queue_join(initial_timeout):
  91             pending = self._queue.qsize()
  92             logger.debug("%d event(s) pending on flush", pending)
  93             if callback is not None:
  94                 callback(pending, timeout)
  95             self._timed_queue_join(timeout - initial_timeout)
  96 
  97     def submit(self, callback):
  98         # type: (Callable[[], None]) -> None
  99         self._ensure_thread()
  100         self._queue.put_nowait(callback)
  101 
  102     def _target(self):
  103         # type: () -> None
  104         while True:
  105             callback = self._queue.get()
  106             try:
  107                 if callback is _TERMINATOR:
  108                     break
  109                 try:
  110                     callback()
  111                 except Exception:
  112                     logger.error("Failed processing job", exc_info=True)
  113             finally:
  114                 self._queue.task_done()
  115             sleep(0)
  116
  ```
  --- 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()
  
  ```
  
golden_diff: diff --git a/sentry_sdk/worker.py b/sentry_sdk/worker.py
  --- a/sentry_sdk/worker.py
  +++ b/sentry_sdk/worker.py
  @@ -45,16 +45,33 @@
           # type: (float) -> bool
           deadline = time() + timeout
           queue = self._queue
  -        queue.all_tasks_done.acquire()  # type: ignore
  +
  +        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
  +
           try:
               while queue.unfinished_tasks:  # type: ignore
                   delay = deadline - time()
                   if delay <= 0:
                       return False
  -                queue.all_tasks_done.wait(timeout=delay)  # type: ignore
  +                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)
  +
               return True
           finally:
  -            queue.all_tasks_done.release()  # type: ignore
  +            if real_all_tasks_done is not None:
  +                real_all_tasks_done.release()  # type: ignore
   
       def start(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@@ -45,16 +45,33 @@\n         # type: (float) -> bool\n         deadline = time() + timeout\n         queue = self._queue\n-        queue.all_tasks_done.acquire()  # type: ignore\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:  # type: ignore\n                 delay = deadline - time()\n                 if delay <= 0:\n                     return False\n-                queue.all_tasks_done.wait(timeout=delay)  # type: ignore\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-            queue.all_tasks_done.release()  # type: ignore\n+            if real_all_tasks_done is not None:\n+                real_all_tasks_done.release()  # type: ignore\n \n     def start(self):\n         # type: () -> None\n", "issue": "Celery - Queue object has no attribute  'all_tasks_done'\nHi all, \r\n\r\nI'm integrating Sentry on a project in python that uses Celery. I'm getting this error when shutting down the worker: \r\n\r\n```\r\nError in atexit._run_exitfuncs:\r\nTraceback (most recent call last):\r\n  File \"/Users/jibanez/API/.conda/envs/cimrender/lib/python3.6/site-packages/sentry_sdk/worker.py\", line 84, in flush\r\n    self._wait_flush(timeout, callback)\r\n  File \"/Users/jibanez/API/.conda/envs/cimrender/lib/python3.6/site-packages/sentry_sdk/worker.py\", line 90, in _wait_flush\r\n    if not self._timed_queue_join(initial_timeout):\r\n  File \"/Users/jibanez/API/.conda/envs/cimrender/lib/python3.6/site-packages/sentry_sdk/worker.py\", line 48, in _timed_queue_join\r\n    queue.all_tasks_done.acquire()  # type: ignore\r\nAttributeError: 'Queue' object has no attribute 'all_tasks_done'\r\n```\r\n\r\nI'm using: \r\n- Python 3.6\r\n- Celery 4.3.0\r\n- OSX Mojave\r\n\r\nAny thoughts? \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        queue.all_tasks_done.acquire()  # type: ignore\n        try:\n            while queue.unfinished_tasks:  # type: ignore\n                delay = deadline - time()\n                if delay <= 0:\n                    return False\n                queue.all_tasks_done.wait(timeout=delay)  # type: ignore\n            return True\n        finally:\n            queue.all_tasks_done.release()  # type: ignore\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(-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:  # type: ignore\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()  # type: ignore\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"}]}