We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
celery_worker pytest fixture timeouts since celery 5.0.3
Since the 5.0.3 release of celery, the `celery_worker` pytest fixture leads to a timeout when performing ping check.

The issue can be reproduced using this simple test file:
```python
pytest_plugins = ["celery.contrib.pytest"]

def test_create_task(celery_app, celery_worker):
    @celery_app.task
    def mul(x, y):
        return x * y

    assert mul.delay(4, 4).get(timeout=10) == 16
```

Below is the pytest output:
```
$ pytest -sv test_celery_worker.py 
============================================================================================== test session starts ===============================================================================================
platform linux -- Python 3.7.3, pytest-6.1.2, py-1.9.0, pluggy-0.13.1 -- /home/anlambert/.virtualenvs/swh/bin/python3
cachedir: .pytest_cache
hypothesis profile 'default' -> database=DirectoryBasedExampleDatabase('/home/anlambert/tmp/.hypothesis/examples')
rootdir: /home/anlambert/tmp
plugins: postgresql-2.5.2, asyncio-0.14.0, mock-3.3.1, cov-2.10.1, django-4.1.0, requests-mock-1.8.0, hypothesis-5.41.3, forked-1.3.0, swh.core-0.9.2.dev4+g6f9779f, flask-1.1.0, xdist-2.1.0, dash-1.17.0, swh.journal-0.5.2.dev1+g12b31a2
collected 1 item                                                                                                                                                                                                 

test_celery_worker.py::test_create_task ERROR

===================================================================================================== ERRORS =====================================================================================================
_______________________________________________________________________________________ ERROR at setup of test_create_task _______________________________________________________________________________________

request = <SubRequest 'celery_worker' for <Function test_create_task>>, celery_app = <Celery celery.tests at 0x7f99b4b91d30>, celery_includes = (), celery_worker_pool = 'solo', celery_worker_parameters = {}

    @pytest.fixture()
    def celery_worker(request,
                      celery_app,
                      celery_includes,
                      celery_worker_pool,
                      celery_worker_parameters):
        # type: (Any, Celery, Sequence[str], str, Any) -> WorkController
        """Fixture: Start worker in a thread, stop it when the test returns."""
        if not NO_WORKER:
            for module in celery_includes:
                celery_app.loader.import_task_module(module)
            with worker.start_worker(celery_app,
                                     pool=celery_worker_pool,
>                                    **celery_worker_parameters) as w:

../dev/celery/celery/contrib/pytest.py:196: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/usr/lib/python3.7/contextlib.py:112: in __enter__
    return next(self.gen)
../dev/celery/celery/contrib/testing/worker.py:82: in start_worker
    assert ping.delay().get(timeout=ping_task_timeout) == 'pong'
../dev/celery/celery/result.py:230: in get
    on_message=on_message,
../dev/celery/celery/backends/base.py:655: in wait_for_pending
    no_ack=no_ack,
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <celery.backends.cache.CacheBackend object at 0x7f99b411fb00>, task_id = '98b047a2-2027-453c-a317-eb31f44a2547', timeout = 10.0, interval = 0.5, no_ack = True, on_interval = <promise@0x7f99b4a2adf0>

    def wait_for(self, task_id,
                 timeout=None, interval=0.5, no_ack=True, on_interval=None):
        """Wait for task and return its result.
    
        If the task raises an exception, this exception
        will be re-raised by :func:`wait_for`.
    
        Raises:
            celery.exceptions.TimeoutError:
                If `timeout` is not :const:`None`, and the operation
                takes longer than `timeout` seconds.
        """
        self._ensure_not_eager()
    
        time_elapsed = 0.0
    
        while 1:
            meta = self.get_task_meta(task_id)
            if meta['status'] in states.READY_STATES:
                return meta
            if on_interval:
                on_interval()
            # avoid hammering the CPU checking status.
            time.sleep(interval)
            time_elapsed += interval
            if timeout and time_elapsed >= timeout:
>               raise TimeoutError('The operation timed out.')
E               celery.exceptions.TimeoutError: The operation timed out.

../dev/celery/celery/backends/base.py:687: TimeoutError
============================================================================================ short test summary info =============================================================================================
ERROR test_celery_worker.py::test_create_task - celery.exceptions.TimeoutError: The operation timed out.
=============================================================================================== 1 error in 10.41s ================================================================================================
```
After a quick `git bisect` session, I managed to identify the commit that introduced the issue: https://github.com/celery/celery/commit/e2031688284484d5b5a57ba29cd9cae2d9a81e39

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `celery/backends/cache.py`
Content:
```
1 """Memcached and in-memory cache result backend."""
2 from kombu.utils.encoding import bytes_to_str, ensure_bytes
3 from kombu.utils.objects import cached_property
4 
5 from celery.exceptions import ImproperlyConfigured
6 from celery.utils.functional import LRUCache
7 
8 from .base import KeyValueStoreBackend
9 
10 __all__ = ('CacheBackend',)
11 
12 _imp = [None]
13 
14 REQUIRES_BACKEND = """\
15 The Memcached backend requires either pylibmc or python-memcached.\
16 """
17 
18 UNKNOWN_BACKEND = """\
19 The cache backend {0!r} is unknown,
20 Please use one of the following backends instead: {1}\
21 """
22 
23 
24 def import_best_memcache():
25     if _imp[0] is None:
26         is_pylibmc, memcache_key_t = False, bytes_to_str
27         try:
28             import pylibmc as memcache
29             is_pylibmc = True
30         except ImportError:
31             try:
32                 import memcache  # noqa
33             except ImportError:
34                 raise ImproperlyConfigured(REQUIRES_BACKEND)
35         _imp[0] = (is_pylibmc, memcache, memcache_key_t)
36     return _imp[0]
37 
38 
39 def get_best_memcache(*args, **kwargs):
40     # pylint: disable=unpacking-non-sequence
41     #   This is most definitely a sequence, but pylint thinks it's not.
42     is_pylibmc, memcache, key_t = import_best_memcache()
43     Client = _Client = memcache.Client
44 
45     if not is_pylibmc:
46         def Client(*args, **kwargs):  # noqa
47             kwargs.pop('behaviors', None)
48             return _Client(*args, **kwargs)
49 
50     return Client, key_t
51 
52 
53 class DummyClient:
54 
55     def __init__(self, *args, **kwargs):
56         self.cache = LRUCache(limit=5000)
57 
58     def get(self, key, *args, **kwargs):
59         return self.cache.get(key)
60 
61     def get_multi(self, keys):
62         cache = self.cache
63         return {k: cache[k] for k in keys if k in cache}
64 
65     def set(self, key, value, *args, **kwargs):
66         self.cache[key] = value
67 
68     def delete(self, key, *args, **kwargs):
69         self.cache.pop(key, None)
70 
71     def incr(self, key, delta=1):
72         return self.cache.incr(key, delta)
73 
74     def touch(self, key, expire):
75         pass
76 
77 
78 backends = {
79     'memcache': get_best_memcache,
80     'memcached': get_best_memcache,
81     'pylibmc': get_best_memcache,
82     'memory': lambda: (DummyClient, ensure_bytes),
83 }
84 
85 
86 class CacheBackend(KeyValueStoreBackend):
87     """Cache result backend."""
88 
89     servers = None
90     supports_autoexpire = True
91     supports_native_join = True
92     implements_incr = True
93 
94     def __init__(self, app, expires=None, backend=None,
95                  options=None, url=None, **kwargs):
96         options = {} if not options else options
97         super().__init__(app, **kwargs)
98         self.url = url
99 
100         self.options = dict(self.app.conf.cache_backend_options,
101                             **options)
102 
103         self.backend = url or backend or self.app.conf.cache_backend
104         if self.backend:
105             self.backend, _, servers = self.backend.partition('://')
106             self.servers = servers.rstrip('/').split(';')
107         self.expires = self.prepare_expires(expires, type=int)
108         try:
109             self.Client, self.key_t = backends[self.backend]()
110         except KeyError:
111             raise ImproperlyConfigured(UNKNOWN_BACKEND.format(
112                 self.backend, ', '.join(backends)))
113         self._encode_prefixes()  # rencode the keyprefixes
114 
115     def get(self, key):
116         return self.client.get(key)
117 
118     def mget(self, keys):
119         return self.client.get_multi(keys)
120 
121     def set(self, key, value):
122         return self.client.set(key, value, self.expires)
123 
124     def delete(self, key):
125         return self.client.delete(key)
126 
127     def _apply_chord_incr(self, header_result, body, **kwargs):
128         chord_key = self.get_key_for_chord(header_result.id)
129         self.client.set(chord_key, 0, time=self.expires)
130         return super()._apply_chord_incr(
131             header_result, body, **kwargs)
132 
133     def incr(self, key):
134         return self.client.incr(key)
135 
136     def expire(self, key, value):
137         return self.client.touch(key, value)
138 
139     @cached_property
140     def client(self):
141         return self.Client(self.servers, **self.options)
142 
143     def __reduce__(self, args=(), kwargs=None):
144         kwargs = {} if not kwargs else kwargs
145         servers = ';'.join(self.servers)
146         backend = f'{self.backend}://{servers}/'
147         kwargs.update(
148             {'backend': backend,
149              'expires': self.expires,
150              'options': self.options})
151         return super().__reduce__(args, kwargs)
152 
153     def as_uri(self, *args, **kwargs):
154         """Return the backend as an URI.
155 
156         This properly handles the case of multiple servers.
157         """
158         servers = ';'.join(self.servers)
159         return f'{self.backend}://{servers}/'
160
```
--- 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/celery/backends/cache.py b/celery/backends/cache.py
--- a/celery/backends/cache.py
+++ b/celery/backends/cache.py
@@ -20,6 +20,10 @@
 Please use one of the following backends instead: {1}\
 """
 
+# Global shared in-memory cache for in-memory cache client
+# This is to share cache between threads
+_DUMMY_CLIENT_CACHE = LRUCache(limit=5000)
+
 
 def import_best_memcache():
     if _imp[0] is None:
@@ -53,7 +57,7 @@
 class DummyClient:
 
     def __init__(self, *args, **kwargs):
-        self.cache = LRUCache(limit=5000)
+        self.cache = _DUMMY_CLIENT_CACHE
 
     def get(self, key, *args, **kwargs):
         return self.cache.get(key)

{"golden_diff": "diff --git a/celery/backends/cache.py b/celery/backends/cache.py\n--- a/celery/backends/cache.py\n+++ b/celery/backends/cache.py\n@@ -20,6 +20,10 @@\n Please use one of the following backends instead: {1}\\\n \"\"\"\n \n+# Global shared in-memory cache for in-memory cache client\n+# This is to share cache between threads\n+_DUMMY_CLIENT_CACHE = LRUCache(limit=5000)\n+\n \n def import_best_memcache():\n     if _imp[0] is None:\n@@ -53,7 +57,7 @@\n class DummyClient:\n \n     def __init__(self, *args, **kwargs):\n-        self.cache = LRUCache(limit=5000)\n+        self.cache = _DUMMY_CLIENT_CACHE\n \n     def get(self, key, *args, **kwargs):\n         return self.cache.get(key)\n", "issue": "celery_worker pytest fixture timeouts since celery 5.0.3\nSince the 5.0.3 release of celery, the `celery_worker` pytest fixture leads to a timeout when performing ping check.\r\n\r\nThe issue can be reproduced using this simple test file:\r\n```python\r\npytest_plugins = [\"celery.contrib.pytest\"]\r\n\r\ndef test_create_task(celery_app, celery_worker):\r\n    @celery_app.task\r\n    def mul(x, y):\r\n        return x * y\r\n\r\n    assert mul.delay(4, 4).get(timeout=10) == 16\r\n```\r\n\r\nBelow is the pytest output:\r\n```\r\n$ pytest -sv test_celery_worker.py \r\n============================================================================================== test session starts ===============================================================================================\r\nplatform linux -- Python 3.7.3, pytest-6.1.2, py-1.9.0, pluggy-0.13.1 -- /home/anlambert/.virtualenvs/swh/bin/python3\r\ncachedir: .pytest_cache\r\nhypothesis profile 'default' -> database=DirectoryBasedExampleDatabase('/home/anlambert/tmp/.hypothesis/examples')\r\nrootdir: /home/anlambert/tmp\r\nplugins: postgresql-2.5.2, asyncio-0.14.0, mock-3.3.1, cov-2.10.1, django-4.1.0, requests-mock-1.8.0, hypothesis-5.41.3, forked-1.3.0, swh.core-0.9.2.dev4+g6f9779f, flask-1.1.0, xdist-2.1.0, dash-1.17.0, swh.journal-0.5.2.dev1+g12b31a2\r\ncollected 1 item                                                                                                                                                                                                 \r\n\r\ntest_celery_worker.py::test_create_task ERROR\r\n\r\n===================================================================================================== ERRORS =====================================================================================================\r\n_______________________________________________________________________________________ ERROR at setup of test_create_task _______________________________________________________________________________________\r\n\r\nrequest = <SubRequest 'celery_worker' for <Function test_create_task>>, celery_app = <Celery celery.tests at 0x7f99b4b91d30>, celery_includes = (), celery_worker_pool = 'solo', celery_worker_parameters = {}\r\n\r\n    @pytest.fixture()\r\n    def celery_worker(request,\r\n                      celery_app,\r\n                      celery_includes,\r\n                      celery_worker_pool,\r\n                      celery_worker_parameters):\r\n        # type: (Any, Celery, Sequence[str], str, Any) -> WorkController\r\n        \"\"\"Fixture: Start worker in a thread, stop it when the test returns.\"\"\"\r\n        if not NO_WORKER:\r\n            for module in celery_includes:\r\n                celery_app.loader.import_task_module(module)\r\n            with worker.start_worker(celery_app,\r\n                                     pool=celery_worker_pool,\r\n>                                    **celery_worker_parameters) as w:\r\n\r\n../dev/celery/celery/contrib/pytest.py:196: \r\n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ \r\n/usr/lib/python3.7/contextlib.py:112: in __enter__\r\n    return next(self.gen)\r\n../dev/celery/celery/contrib/testing/worker.py:82: in start_worker\r\n    assert ping.delay().get(timeout=ping_task_timeout) == 'pong'\r\n../dev/celery/celery/result.py:230: in get\r\n    on_message=on_message,\r\n../dev/celery/celery/backends/base.py:655: in wait_for_pending\r\n    no_ack=no_ack,\r\n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ \r\n\r\nself = <celery.backends.cache.CacheBackend object at 0x7f99b411fb00>, task_id = '98b047a2-2027-453c-a317-eb31f44a2547', timeout = 10.0, interval = 0.5, no_ack = True, on_interval = <promise@0x7f99b4a2adf0>\r\n\r\n    def wait_for(self, task_id,\r\n                 timeout=None, interval=0.5, no_ack=True, on_interval=None):\r\n        \"\"\"Wait for task and return its result.\r\n    \r\n        If the task raises an exception, this exception\r\n        will be re-raised by :func:`wait_for`.\r\n    \r\n        Raises:\r\n            celery.exceptions.TimeoutError:\r\n                If `timeout` is not :const:`None`, and the operation\r\n                takes longer than `timeout` seconds.\r\n        \"\"\"\r\n        self._ensure_not_eager()\r\n    \r\n        time_elapsed = 0.0\r\n    \r\n        while 1:\r\n            meta = self.get_task_meta(task_id)\r\n            if meta['status'] in states.READY_STATES:\r\n                return meta\r\n            if on_interval:\r\n                on_interval()\r\n            # avoid hammering the CPU checking status.\r\n            time.sleep(interval)\r\n            time_elapsed += interval\r\n            if timeout and time_elapsed >= timeout:\r\n>               raise TimeoutError('The operation timed out.')\r\nE               celery.exceptions.TimeoutError: The operation timed out.\r\n\r\n../dev/celery/celery/backends/base.py:687: TimeoutError\r\n============================================================================================ short test summary info =============================================================================================\r\nERROR test_celery_worker.py::test_create_task - celery.exceptions.TimeoutError: The operation timed out.\r\n=============================================================================================== 1 error in 10.41s ================================================================================================\r\n```\r\nAfter a quick `git bisect` session, I managed to identify the commit that introduced the issue: https://github.com/celery/celery/commit/e2031688284484d5b5a57ba29cd9cae2d9a81e39\n", "before_files": [{"content": "\"\"\"Memcached and in-memory cache result backend.\"\"\"\nfrom kombu.utils.encoding import bytes_to_str, ensure_bytes\nfrom kombu.utils.objects import cached_property\n\nfrom celery.exceptions import ImproperlyConfigured\nfrom celery.utils.functional import LRUCache\n\nfrom .base import KeyValueStoreBackend\n\n__all__ = ('CacheBackend',)\n\n_imp = [None]\n\nREQUIRES_BACKEND = \"\"\"\\\nThe Memcached backend requires either pylibmc or python-memcached.\\\n\"\"\"\n\nUNKNOWN_BACKEND = \"\"\"\\\nThe cache backend {0!r} is unknown,\nPlease use one of the following backends instead: {1}\\\n\"\"\"\n\n\ndef import_best_memcache():\n    if _imp[0] is None:\n        is_pylibmc, memcache_key_t = False, bytes_to_str\n        try:\n            import pylibmc as memcache\n            is_pylibmc = True\n        except ImportError:\n            try:\n                import memcache  # noqa\n            except ImportError:\n                raise ImproperlyConfigured(REQUIRES_BACKEND)\n        _imp[0] = (is_pylibmc, memcache, memcache_key_t)\n    return _imp[0]\n\n\ndef get_best_memcache(*args, **kwargs):\n    # pylint: disable=unpacking-non-sequence\n    #   This is most definitely a sequence, but pylint thinks it's not.\n    is_pylibmc, memcache, key_t = import_best_memcache()\n    Client = _Client = memcache.Client\n\n    if not is_pylibmc:\n        def Client(*args, **kwargs):  # noqa\n            kwargs.pop('behaviors', None)\n            return _Client(*args, **kwargs)\n\n    return Client, key_t\n\n\nclass DummyClient:\n\n    def __init__(self, *args, **kwargs):\n        self.cache = LRUCache(limit=5000)\n\n    def get(self, key, *args, **kwargs):\n        return self.cache.get(key)\n\n    def get_multi(self, keys):\n        cache = self.cache\n        return {k: cache[k] for k in keys if k in cache}\n\n    def set(self, key, value, *args, **kwargs):\n        self.cache[key] = value\n\n    def delete(self, key, *args, **kwargs):\n        self.cache.pop(key, None)\n\n    def incr(self, key, delta=1):\n        return self.cache.incr(key, delta)\n\n    def touch(self, key, expire):\n        pass\n\n\nbackends = {\n    'memcache': get_best_memcache,\n    'memcached': get_best_memcache,\n    'pylibmc': get_best_memcache,\n    'memory': lambda: (DummyClient, ensure_bytes),\n}\n\n\nclass CacheBackend(KeyValueStoreBackend):\n    \"\"\"Cache result backend.\"\"\"\n\n    servers = None\n    supports_autoexpire = True\n    supports_native_join = True\n    implements_incr = True\n\n    def __init__(self, app, expires=None, backend=None,\n                 options=None, url=None, **kwargs):\n        options = {} if not options else options\n        super().__init__(app, **kwargs)\n        self.url = url\n\n        self.options = dict(self.app.conf.cache_backend_options,\n                            **options)\n\n        self.backend = url or backend or self.app.conf.cache_backend\n        if self.backend:\n            self.backend, _, servers = self.backend.partition('://')\n            self.servers = servers.rstrip('/').split(';')\n        self.expires = self.prepare_expires(expires, type=int)\n        try:\n            self.Client, self.key_t = backends[self.backend]()\n        except KeyError:\n            raise ImproperlyConfigured(UNKNOWN_BACKEND.format(\n                self.backend, ', '.join(backends)))\n        self._encode_prefixes()  # rencode the keyprefixes\n\n    def get(self, key):\n        return self.client.get(key)\n\n    def mget(self, keys):\n        return self.client.get_multi(keys)\n\n    def set(self, key, value):\n        return self.client.set(key, value, self.expires)\n\n    def delete(self, key):\n        return self.client.delete(key)\n\n    def _apply_chord_incr(self, header_result, body, **kwargs):\n        chord_key = self.get_key_for_chord(header_result.id)\n        self.client.set(chord_key, 0, time=self.expires)\n        return super()._apply_chord_incr(\n            header_result, body, **kwargs)\n\n    def incr(self, key):\n        return self.client.incr(key)\n\n    def expire(self, key, value):\n        return self.client.touch(key, value)\n\n    @cached_property\n    def client(self):\n        return self.Client(self.servers, **self.options)\n\n    def __reduce__(self, args=(), kwargs=None):\n        kwargs = {} if not kwargs else kwargs\n        servers = ';'.join(self.servers)\n        backend = f'{self.backend}://{servers}/'\n        kwargs.update(\n            {'backend': backend,\n             'expires': self.expires,\n             'options': self.options})\n        return super().__reduce__(args, kwargs)\n\n    def as_uri(self, *args, **kwargs):\n        \"\"\"Return the backend as an URI.\n\n        This properly handles the case of multiple servers.\n        \"\"\"\n        servers = ';'.join(self.servers)\n        return f'{self.backend}://{servers}/'\n", "path": "celery/backends/cache.py"}], "after_files": [{"content": "\"\"\"Memcached and in-memory cache result backend.\"\"\"\nfrom kombu.utils.encoding import bytes_to_str, ensure_bytes\nfrom kombu.utils.objects import cached_property\n\nfrom celery.exceptions import ImproperlyConfigured\nfrom celery.utils.functional import LRUCache\n\nfrom .base import KeyValueStoreBackend\n\n__all__ = ('CacheBackend',)\n\n_imp = [None]\n\nREQUIRES_BACKEND = \"\"\"\\\nThe Memcached backend requires either pylibmc or python-memcached.\\\n\"\"\"\n\nUNKNOWN_BACKEND = \"\"\"\\\nThe cache backend {0!r} is unknown,\nPlease use one of the following backends instead: {1}\\\n\"\"\"\n\n# Global shared in-memory cache for in-memory cache client\n# This is to share cache between threads\n_DUMMY_CLIENT_CACHE = LRUCache(limit=5000)\n\n\ndef import_best_memcache():\n    if _imp[0] is None:\n        is_pylibmc, memcache_key_t = False, bytes_to_str\n        try:\n            import pylibmc as memcache\n            is_pylibmc = True\n        except ImportError:\n            try:\n                import memcache  # noqa\n            except ImportError:\n                raise ImproperlyConfigured(REQUIRES_BACKEND)\n        _imp[0] = (is_pylibmc, memcache, memcache_key_t)\n    return _imp[0]\n\n\ndef get_best_memcache(*args, **kwargs):\n    # pylint: disable=unpacking-non-sequence\n    #   This is most definitely a sequence, but pylint thinks it's not.\n    is_pylibmc, memcache, key_t = import_best_memcache()\n    Client = _Client = memcache.Client\n\n    if not is_pylibmc:\n        def Client(*args, **kwargs):  # noqa\n            kwargs.pop('behaviors', None)\n            return _Client(*args, **kwargs)\n\n    return Client, key_t\n\n\nclass DummyClient:\n\n    def __init__(self, *args, **kwargs):\n        self.cache = _DUMMY_CLIENT_CACHE\n\n    def get(self, key, *args, **kwargs):\n        return self.cache.get(key)\n\n    def get_multi(self, keys):\n        cache = self.cache\n        return {k: cache[k] for k in keys if k in cache}\n\n    def set(self, key, value, *args, **kwargs):\n        self.cache[key] = value\n\n    def delete(self, key, *args, **kwargs):\n        self.cache.pop(key, None)\n\n    def incr(self, key, delta=1):\n        return self.cache.incr(key, delta)\n\n    def touch(self, key, expire):\n        pass\n\n\nbackends = {\n    'memcache': get_best_memcache,\n    'memcached': get_best_memcache,\n    'pylibmc': get_best_memcache,\n    'memory': lambda: (DummyClient, ensure_bytes),\n}\n\n\nclass CacheBackend(KeyValueStoreBackend):\n    \"\"\"Cache result backend.\"\"\"\n\n    servers = None\n    supports_autoexpire = True\n    supports_native_join = True\n    implements_incr = True\n\n    def __init__(self, app, expires=None, backend=None,\n                 options=None, url=None, **kwargs):\n        options = {} if not options else options\n        super().__init__(app, **kwargs)\n        self.url = url\n\n        self.options = dict(self.app.conf.cache_backend_options,\n                            **options)\n\n        self.backend = url or backend or self.app.conf.cache_backend\n        if self.backend:\n            self.backend, _, servers = self.backend.partition('://')\n            self.servers = servers.rstrip('/').split(';')\n        self.expires = self.prepare_expires(expires, type=int)\n        try:\n            self.Client, self.key_t = backends[self.backend]()\n        except KeyError:\n            raise ImproperlyConfigured(UNKNOWN_BACKEND.format(\n                self.backend, ', '.join(backends)))\n        self._encode_prefixes()  # rencode the keyprefixes\n\n    def get(self, key):\n        return self.client.get(key)\n\n    def mget(self, keys):\n        return self.client.get_multi(keys)\n\n    def set(self, key, value):\n        return self.client.set(key, value, self.expires)\n\n    def delete(self, key):\n        return self.client.delete(key)\n\n    def _apply_chord_incr(self, header_result, body, **kwargs):\n        chord_key = self.get_key_for_chord(header_result.id)\n        self.client.set(chord_key, 0, time=self.expires)\n        return super()._apply_chord_incr(\n            header_result, body, **kwargs)\n\n    def incr(self, key):\n        return self.client.incr(key)\n\n    def expire(self, key, value):\n        return self.client.touch(key, value)\n\n    @cached_property\n    def client(self):\n        return self.Client(self.servers, **self.options)\n\n    def __reduce__(self, args=(), kwargs=None):\n        kwargs = {} if not kwargs else kwargs\n        servers = ';'.join(self.servers)\n        backend = f'{self.backend}://{servers}/'\n        kwargs.update(\n            {'backend': backend,\n             'expires': self.expires,\n             'options': self.options})\n        return super().__reduce__(args, kwargs)\n\n    def as_uri(self, *args, **kwargs):\n        \"\"\"Return the backend as an URI.\n\n        This properly handles the case of multiple servers.\n        \"\"\"\n        servers = ';'.join(self.servers)\n        return f'{self.backend}://{servers}/'\n", "path": "celery/backends/cache.py"}]}