We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
make REVOKES_MAX and REVOKE_EXPIRES configurable
Values of REVOKE_EXPIRES and REVOKES_MAX in worker/state.py  are hardcoded.

This should be configurable. Some of us really needed to change this.



--- 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/worker/state.py`
Content:
```
1 """Internal worker state (global).
2 
3 This includes the currently active and reserved tasks,
4 statistics, and revoked tasks.
5 """
6 import os
7 import platform
8 import shelve
9 import sys
10 import weakref
11 import zlib
12 from collections import Counter
13 
14 from kombu.serialization import pickle, pickle_protocol
15 from kombu.utils.objects import cached_property
16 
17 from celery import __version__
18 from celery.exceptions import WorkerShutdown, WorkerTerminate
19 from celery.utils.collections import LimitedSet
20 
21 __all__ = (
22     'SOFTWARE_INFO', 'reserved_requests', 'active_requests',
23     'total_count', 'revoked', 'task_reserved', 'maybe_shutdown',
24     'task_accepted', 'task_ready', 'Persistent',
25 )
26 
27 #: Worker software/platform information.
28 SOFTWARE_INFO = {
29     'sw_ident': 'py-celery',
30     'sw_ver': __version__,
31     'sw_sys': platform.system(),
32 }
33 
34 #: maximum number of revokes to keep in memory.
35 REVOKES_MAX = 50000
36 
37 #: maximum number of successful tasks to keep in memory.
38 SUCCESSFUL_MAX = 1000
39 
40 #: how many seconds a revoke will be active before
41 #: being expired when the max limit has been exceeded.
42 REVOKE_EXPIRES = 10800
43 
44 #: how many seconds a successful task will be cached in memory
45 #: before being expired when the max limit has been exceeded.
46 SUCCESSFUL_EXPIRES = 10800
47 
48 #: Mapping of reserved task_id->Request.
49 requests = {}
50 
51 #: set of all reserved :class:`~celery.worker.request.Request`'s.
52 reserved_requests = weakref.WeakSet()
53 
54 #: set of currently active :class:`~celery.worker.request.Request`'s.
55 active_requests = weakref.WeakSet()
56 
57 #: A limited set of successful :class:`~celery.worker.request.Request`'s.
58 successful_requests = LimitedSet(maxlen=SUCCESSFUL_MAX,
59                                  expires=SUCCESSFUL_EXPIRES)
60 
61 #: count of tasks accepted by the worker, sorted by type.
62 total_count = Counter()
63 
64 #: count of all tasks accepted by the worker
65 all_total_count = [0]
66 
67 #: the list of currently revoked tasks.  Persistent if ``statedb`` set.
68 revoked = LimitedSet(maxlen=REVOKES_MAX, expires=REVOKE_EXPIRES)
69 
70 should_stop = None
71 should_terminate = None
72 
73 
74 def reset_state():
75     requests.clear()
76     reserved_requests.clear()
77     active_requests.clear()
78     successful_requests.clear()
79     total_count.clear()
80     all_total_count[:] = [0]
81     revoked.clear()
82 
83 
84 def maybe_shutdown():
85     """Shutdown if flags have been set."""
86     if should_terminate is not None and should_terminate is not False:
87         raise WorkerTerminate(should_terminate)
88     elif should_stop is not None and should_stop is not False:
89         raise WorkerShutdown(should_stop)
90 
91 
92 def task_reserved(request,
93                   add_request=requests.__setitem__,
94                   add_reserved_request=reserved_requests.add):
95     """Update global state when a task has been reserved."""
96     add_request(request.id, request)
97     add_reserved_request(request)
98 
99 
100 def task_accepted(request,
101                   _all_total_count=None,
102                   add_active_request=active_requests.add,
103                   add_to_total_count=total_count.update):
104     """Update global state when a task has been accepted."""
105     if not _all_total_count:
106         _all_total_count = all_total_count
107     add_active_request(request)
108     add_to_total_count({request.name: 1})
109     all_total_count[0] += 1
110 
111 
112 def task_ready(request,
113                successful=False,
114                remove_request=requests.pop,
115                discard_active_request=active_requests.discard,
116                discard_reserved_request=reserved_requests.discard):
117     """Update global state when a task is ready."""
118     if successful:
119         successful_requests.add(request.id)
120 
121     remove_request(request.id, None)
122     discard_active_request(request)
123     discard_reserved_request(request)
124 
125 
126 C_BENCH = os.environ.get('C_BENCH') or os.environ.get('CELERY_BENCH')
127 C_BENCH_EVERY = int(os.environ.get('C_BENCH_EVERY') or
128                     os.environ.get('CELERY_BENCH_EVERY') or 1000)
129 if C_BENCH:  # pragma: no cover
130     import atexit
131     from time import monotonic
132 
133     from billiard.process import current_process
134 
135     from celery.utils.debug import memdump, sample_mem
136 
137     all_count = 0
138     bench_first = None
139     bench_start = None
140     bench_last = None
141     bench_every = C_BENCH_EVERY
142     bench_sample = []
143     __reserved = task_reserved
144     __ready = task_ready
145 
146     if current_process()._name == 'MainProcess':
147         @atexit.register
148         def on_shutdown():
149             if bench_first is not None and bench_last is not None:
150                 print('- Time spent in benchmark: {!r}'.format(
151                     bench_last - bench_first))
152                 print('- Avg: {}'.format(
153                     sum(bench_sample) / len(bench_sample)))
154                 memdump()
155 
156     def task_reserved(request):
157         """Called when a task is reserved by the worker."""
158         global bench_start
159         global bench_first
160         now = None
161         if bench_start is None:
162             bench_start = now = monotonic()
163         if bench_first is None:
164             bench_first = now
165 
166         return __reserved(request)
167 
168     def task_ready(request):
169         """Called when a task is completed."""
170         global all_count
171         global bench_start
172         global bench_last
173         all_count += 1
174         if not all_count % bench_every:
175             now = monotonic()
176             diff = now - bench_start
177             print('- Time spent processing {} tasks (since first '
178                   'task received): ~{:.4f}s\n'.format(bench_every, diff))
179             sys.stdout.flush()
180             bench_start = bench_last = now
181             bench_sample.append(diff)
182             sample_mem()
183         return __ready(request)
184 
185 
186 class Persistent:
187     """Stores worker state between restarts.
188 
189     This is the persistent data stored by the worker when
190     :option:`celery worker --statedb` is enabled.
191 
192     Currently only stores revoked task id's.
193     """
194 
195     storage = shelve
196     protocol = pickle_protocol
197     compress = zlib.compress
198     decompress = zlib.decompress
199     _is_open = False
200 
201     def __init__(self, state, filename, clock=None):
202         self.state = state
203         self.filename = filename
204         self.clock = clock
205         self.merge()
206 
207     def open(self):
208         return self.storage.open(
209             self.filename, protocol=self.protocol, writeback=True,
210         )
211 
212     def merge(self):
213         self._merge_with(self.db)
214 
215     def sync(self):
216         self._sync_with(self.db)
217         self.db.sync()
218 
219     def close(self):
220         if self._is_open:
221             self.db.close()
222             self._is_open = False
223 
224     def save(self):
225         self.sync()
226         self.close()
227 
228     def _merge_with(self, d):
229         self._merge_revoked(d)
230         self._merge_clock(d)
231         return d
232 
233     def _sync_with(self, d):
234         self._revoked_tasks.purge()
235         d.update({
236             '__proto__': 3,
237             'zrevoked': self.compress(self._dumps(self._revoked_tasks)),
238             'clock': self.clock.forward() if self.clock else 0,
239         })
240         return d
241 
242     def _merge_clock(self, d):
243         if self.clock:
244             d['clock'] = self.clock.adjust(d.get('clock') or 0)
245 
246     def _merge_revoked(self, d):
247         try:
248             self._merge_revoked_v3(d['zrevoked'])
249         except KeyError:
250             try:
251                 self._merge_revoked_v2(d.pop('revoked'))
252             except KeyError:
253                 pass
254         # purge expired items at boot
255         self._revoked_tasks.purge()
256 
257     def _merge_revoked_v3(self, zrevoked):
258         if zrevoked:
259             self._revoked_tasks.update(pickle.loads(self.decompress(zrevoked)))
260 
261     def _merge_revoked_v2(self, saved):
262         if not isinstance(saved, LimitedSet):
263             # (pre 3.0.18) used to be stored as a dict
264             return self._merge_revoked_v1(saved)
265         self._revoked_tasks.update(saved)
266 
267     def _merge_revoked_v1(self, saved):
268         add = self._revoked_tasks.add
269         for item in saved:
270             add(item)
271 
272     def _dumps(self, obj):
273         return pickle.dumps(obj, protocol=self.protocol)
274 
275     @property
276     def _revoked_tasks(self):
277         return self.state.revoked
278 
279     @cached_property
280     def db(self):
281         self._is_open = True
282         return self.open()
283
```
--- 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/worker/state.py b/celery/worker/state.py
--- a/celery/worker/state.py
+++ b/celery/worker/state.py
@@ -32,18 +32,18 @@
 }
 
 #: maximum number of revokes to keep in memory.
-REVOKES_MAX = 50000
+REVOKES_MAX = int(os.environ.get('CELERY_WORKER_REVOKES_MAX', 50000))
 
 #: maximum number of successful tasks to keep in memory.
-SUCCESSFUL_MAX = 1000
+SUCCESSFUL_MAX = int(os.environ.get('CELERY_WORKER_SUCCESSFUL_MAX', 1000))
 
 #: how many seconds a revoke will be active before
 #: being expired when the max limit has been exceeded.
-REVOKE_EXPIRES = 10800
+REVOKE_EXPIRES = float(os.environ.get('CELERY_WORKER_REVOKE_EXPIRES', 10800))
 
 #: how many seconds a successful task will be cached in memory
 #: before being expired when the max limit has been exceeded.
-SUCCESSFUL_EXPIRES = 10800
+SUCCESSFUL_EXPIRES = float(os.environ.get('CELERY_WORKER_SUCCESSFUL_EXPIRES', 10800))
 
 #: Mapping of reserved task_id->Request.
 requests = {}

{"golden_diff": "diff --git a/celery/worker/state.py b/celery/worker/state.py\n--- a/celery/worker/state.py\n+++ b/celery/worker/state.py\n@@ -32,18 +32,18 @@\n }\n \n #: maximum number of revokes to keep in memory.\n-REVOKES_MAX = 50000\n+REVOKES_MAX = int(os.environ.get('CELERY_WORKER_REVOKES_MAX', 50000))\n \n #: maximum number of successful tasks to keep in memory.\n-SUCCESSFUL_MAX = 1000\n+SUCCESSFUL_MAX = int(os.environ.get('CELERY_WORKER_SUCCESSFUL_MAX', 1000))\n \n #: how many seconds a revoke will be active before\n #: being expired when the max limit has been exceeded.\n-REVOKE_EXPIRES = 10800\n+REVOKE_EXPIRES = float(os.environ.get('CELERY_WORKER_REVOKE_EXPIRES', 10800))\n \n #: how many seconds a successful task will be cached in memory\n #: before being expired when the max limit has been exceeded.\n-SUCCESSFUL_EXPIRES = 10800\n+SUCCESSFUL_EXPIRES = float(os.environ.get('CELERY_WORKER_SUCCESSFUL_EXPIRES', 10800))\n \n #: Mapping of reserved task_id->Request.\n requests = {}\n", "issue": "make REVOKES_MAX and REVOKE_EXPIRES configurable\nValues of REVOKE_EXPIRES and REVOKES_MAX in worker/state.py  are hardcoded.\n\nThis should be configurable. Some of us really needed to change this.\n\n\n", "before_files": [{"content": "\"\"\"Internal worker state (global).\n\nThis includes the currently active and reserved tasks,\nstatistics, and revoked tasks.\n\"\"\"\nimport os\nimport platform\nimport shelve\nimport sys\nimport weakref\nimport zlib\nfrom collections import Counter\n\nfrom kombu.serialization import pickle, pickle_protocol\nfrom kombu.utils.objects import cached_property\n\nfrom celery import __version__\nfrom celery.exceptions import WorkerShutdown, WorkerTerminate\nfrom celery.utils.collections import LimitedSet\n\n__all__ = (\n    'SOFTWARE_INFO', 'reserved_requests', 'active_requests',\n    'total_count', 'revoked', 'task_reserved', 'maybe_shutdown',\n    'task_accepted', 'task_ready', 'Persistent',\n)\n\n#: Worker software/platform information.\nSOFTWARE_INFO = {\n    'sw_ident': 'py-celery',\n    'sw_ver': __version__,\n    'sw_sys': platform.system(),\n}\n\n#: maximum number of revokes to keep in memory.\nREVOKES_MAX = 50000\n\n#: maximum number of successful tasks to keep in memory.\nSUCCESSFUL_MAX = 1000\n\n#: how many seconds a revoke will be active before\n#: being expired when the max limit has been exceeded.\nREVOKE_EXPIRES = 10800\n\n#: how many seconds a successful task will be cached in memory\n#: before being expired when the max limit has been exceeded.\nSUCCESSFUL_EXPIRES = 10800\n\n#: Mapping of reserved task_id->Request.\nrequests = {}\n\n#: set of all reserved :class:`~celery.worker.request.Request`'s.\nreserved_requests = weakref.WeakSet()\n\n#: set of currently active :class:`~celery.worker.request.Request`'s.\nactive_requests = weakref.WeakSet()\n\n#: A limited set of successful :class:`~celery.worker.request.Request`'s.\nsuccessful_requests = LimitedSet(maxlen=SUCCESSFUL_MAX,\n                                 expires=SUCCESSFUL_EXPIRES)\n\n#: count of tasks accepted by the worker, sorted by type.\ntotal_count = Counter()\n\n#: count of all tasks accepted by the worker\nall_total_count = [0]\n\n#: the list of currently revoked tasks.  Persistent if ``statedb`` set.\nrevoked = LimitedSet(maxlen=REVOKES_MAX, expires=REVOKE_EXPIRES)\n\nshould_stop = None\nshould_terminate = None\n\n\ndef reset_state():\n    requests.clear()\n    reserved_requests.clear()\n    active_requests.clear()\n    successful_requests.clear()\n    total_count.clear()\n    all_total_count[:] = [0]\n    revoked.clear()\n\n\ndef maybe_shutdown():\n    \"\"\"Shutdown if flags have been set.\"\"\"\n    if should_terminate is not None and should_terminate is not False:\n        raise WorkerTerminate(should_terminate)\n    elif should_stop is not None and should_stop is not False:\n        raise WorkerShutdown(should_stop)\n\n\ndef task_reserved(request,\n                  add_request=requests.__setitem__,\n                  add_reserved_request=reserved_requests.add):\n    \"\"\"Update global state when a task has been reserved.\"\"\"\n    add_request(request.id, request)\n    add_reserved_request(request)\n\n\ndef task_accepted(request,\n                  _all_total_count=None,\n                  add_active_request=active_requests.add,\n                  add_to_total_count=total_count.update):\n    \"\"\"Update global state when a task has been accepted.\"\"\"\n    if not _all_total_count:\n        _all_total_count = all_total_count\n    add_active_request(request)\n    add_to_total_count({request.name: 1})\n    all_total_count[0] += 1\n\n\ndef task_ready(request,\n               successful=False,\n               remove_request=requests.pop,\n               discard_active_request=active_requests.discard,\n               discard_reserved_request=reserved_requests.discard):\n    \"\"\"Update global state when a task is ready.\"\"\"\n    if successful:\n        successful_requests.add(request.id)\n\n    remove_request(request.id, None)\n    discard_active_request(request)\n    discard_reserved_request(request)\n\n\nC_BENCH = os.environ.get('C_BENCH') or os.environ.get('CELERY_BENCH')\nC_BENCH_EVERY = int(os.environ.get('C_BENCH_EVERY') or\n                    os.environ.get('CELERY_BENCH_EVERY') or 1000)\nif C_BENCH:  # pragma: no cover\n    import atexit\n    from time import monotonic\n\n    from billiard.process import current_process\n\n    from celery.utils.debug import memdump, sample_mem\n\n    all_count = 0\n    bench_first = None\n    bench_start = None\n    bench_last = None\n    bench_every = C_BENCH_EVERY\n    bench_sample = []\n    __reserved = task_reserved\n    __ready = task_ready\n\n    if current_process()._name == 'MainProcess':\n        @atexit.register\n        def on_shutdown():\n            if bench_first is not None and bench_last is not None:\n                print('- Time spent in benchmark: {!r}'.format(\n                    bench_last - bench_first))\n                print('- Avg: {}'.format(\n                    sum(bench_sample) / len(bench_sample)))\n                memdump()\n\n    def task_reserved(request):\n        \"\"\"Called when a task is reserved by the worker.\"\"\"\n        global bench_start\n        global bench_first\n        now = None\n        if bench_start is None:\n            bench_start = now = monotonic()\n        if bench_first is None:\n            bench_first = now\n\n        return __reserved(request)\n\n    def task_ready(request):\n        \"\"\"Called when a task is completed.\"\"\"\n        global all_count\n        global bench_start\n        global bench_last\n        all_count += 1\n        if not all_count % bench_every:\n            now = monotonic()\n            diff = now - bench_start\n            print('- Time spent processing {} tasks (since first '\n                  'task received): ~{:.4f}s\\n'.format(bench_every, diff))\n            sys.stdout.flush()\n            bench_start = bench_last = now\n            bench_sample.append(diff)\n            sample_mem()\n        return __ready(request)\n\n\nclass Persistent:\n    \"\"\"Stores worker state between restarts.\n\n    This is the persistent data stored by the worker when\n    :option:`celery worker --statedb` is enabled.\n\n    Currently only stores revoked task id's.\n    \"\"\"\n\n    storage = shelve\n    protocol = pickle_protocol\n    compress = zlib.compress\n    decompress = zlib.decompress\n    _is_open = False\n\n    def __init__(self, state, filename, clock=None):\n        self.state = state\n        self.filename = filename\n        self.clock = clock\n        self.merge()\n\n    def open(self):\n        return self.storage.open(\n            self.filename, protocol=self.protocol, writeback=True,\n        )\n\n    def merge(self):\n        self._merge_with(self.db)\n\n    def sync(self):\n        self._sync_with(self.db)\n        self.db.sync()\n\n    def close(self):\n        if self._is_open:\n            self.db.close()\n            self._is_open = False\n\n    def save(self):\n        self.sync()\n        self.close()\n\n    def _merge_with(self, d):\n        self._merge_revoked(d)\n        self._merge_clock(d)\n        return d\n\n    def _sync_with(self, d):\n        self._revoked_tasks.purge()\n        d.update({\n            '__proto__': 3,\n            'zrevoked': self.compress(self._dumps(self._revoked_tasks)),\n            'clock': self.clock.forward() if self.clock else 0,\n        })\n        return d\n\n    def _merge_clock(self, d):\n        if self.clock:\n            d['clock'] = self.clock.adjust(d.get('clock') or 0)\n\n    def _merge_revoked(self, d):\n        try:\n            self._merge_revoked_v3(d['zrevoked'])\n        except KeyError:\n            try:\n                self._merge_revoked_v2(d.pop('revoked'))\n            except KeyError:\n                pass\n        # purge expired items at boot\n        self._revoked_tasks.purge()\n\n    def _merge_revoked_v3(self, zrevoked):\n        if zrevoked:\n            self._revoked_tasks.update(pickle.loads(self.decompress(zrevoked)))\n\n    def _merge_revoked_v2(self, saved):\n        if not isinstance(saved, LimitedSet):\n            # (pre 3.0.18) used to be stored as a dict\n            return self._merge_revoked_v1(saved)\n        self._revoked_tasks.update(saved)\n\n    def _merge_revoked_v1(self, saved):\n        add = self._revoked_tasks.add\n        for item in saved:\n            add(item)\n\n    def _dumps(self, obj):\n        return pickle.dumps(obj, protocol=self.protocol)\n\n    @property\n    def _revoked_tasks(self):\n        return self.state.revoked\n\n    @cached_property\n    def db(self):\n        self._is_open = True\n        return self.open()\n", "path": "celery/worker/state.py"}], "after_files": [{"content": "\"\"\"Internal worker state (global).\n\nThis includes the currently active and reserved tasks,\nstatistics, and revoked tasks.\n\"\"\"\nimport os\nimport platform\nimport shelve\nimport sys\nimport weakref\nimport zlib\nfrom collections import Counter\n\nfrom kombu.serialization import pickle, pickle_protocol\nfrom kombu.utils.objects import cached_property\n\nfrom celery import __version__\nfrom celery.exceptions import WorkerShutdown, WorkerTerminate\nfrom celery.utils.collections import LimitedSet\n\n__all__ = (\n    'SOFTWARE_INFO', 'reserved_requests', 'active_requests',\n    'total_count', 'revoked', 'task_reserved', 'maybe_shutdown',\n    'task_accepted', 'task_ready', 'Persistent',\n)\n\n#: Worker software/platform information.\nSOFTWARE_INFO = {\n    'sw_ident': 'py-celery',\n    'sw_ver': __version__,\n    'sw_sys': platform.system(),\n}\n\n#: maximum number of revokes to keep in memory.\nREVOKES_MAX = int(os.environ.get('CELERY_WORKER_REVOKES_MAX', 50000))\n\n#: maximum number of successful tasks to keep in memory.\nSUCCESSFUL_MAX = int(os.environ.get('CELERY_WORKER_SUCCESSFUL_MAX', 1000))\n\n#: how many seconds a revoke will be active before\n#: being expired when the max limit has been exceeded.\nREVOKE_EXPIRES = float(os.environ.get('CELERY_WORKER_REVOKE_EXPIRES', 10800))\n\n#: how many seconds a successful task will be cached in memory\n#: before being expired when the max limit has been exceeded.\nSUCCESSFUL_EXPIRES = float(os.environ.get('CELERY_WORKER_SUCCESSFUL_EXPIRES', 10800))\n\n#: Mapping of reserved task_id->Request.\nrequests = {}\n\n#: set of all reserved :class:`~celery.worker.request.Request`'s.\nreserved_requests = weakref.WeakSet()\n\n#: set of currently active :class:`~celery.worker.request.Request`'s.\nactive_requests = weakref.WeakSet()\n\n#: A limited set of successful :class:`~celery.worker.request.Request`'s.\nsuccessful_requests = LimitedSet(maxlen=SUCCESSFUL_MAX,\n                                 expires=SUCCESSFUL_EXPIRES)\n\n#: count of tasks accepted by the worker, sorted by type.\ntotal_count = Counter()\n\n#: count of all tasks accepted by the worker\nall_total_count = [0]\n\n#: the list of currently revoked tasks.  Persistent if ``statedb`` set.\nrevoked = LimitedSet(maxlen=REVOKES_MAX, expires=REVOKE_EXPIRES)\n\nshould_stop = None\nshould_terminate = None\n\n\ndef reset_state():\n    requests.clear()\n    reserved_requests.clear()\n    active_requests.clear()\n    successful_requests.clear()\n    total_count.clear()\n    all_total_count[:] = [0]\n    revoked.clear()\n\n\ndef maybe_shutdown():\n    \"\"\"Shutdown if flags have been set.\"\"\"\n    if should_terminate is not None and should_terminate is not False:\n        raise WorkerTerminate(should_terminate)\n    elif should_stop is not None and should_stop is not False:\n        raise WorkerShutdown(should_stop)\n\n\ndef task_reserved(request,\n                  add_request=requests.__setitem__,\n                  add_reserved_request=reserved_requests.add):\n    \"\"\"Update global state when a task has been reserved.\"\"\"\n    add_request(request.id, request)\n    add_reserved_request(request)\n\n\ndef task_accepted(request,\n                  _all_total_count=None,\n                  add_active_request=active_requests.add,\n                  add_to_total_count=total_count.update):\n    \"\"\"Update global state when a task has been accepted.\"\"\"\n    if not _all_total_count:\n        _all_total_count = all_total_count\n    add_active_request(request)\n    add_to_total_count({request.name: 1})\n    all_total_count[0] += 1\n\n\ndef task_ready(request,\n               successful=False,\n               remove_request=requests.pop,\n               discard_active_request=active_requests.discard,\n               discard_reserved_request=reserved_requests.discard):\n    \"\"\"Update global state when a task is ready.\"\"\"\n    if successful:\n        successful_requests.add(request.id)\n\n    remove_request(request.id, None)\n    discard_active_request(request)\n    discard_reserved_request(request)\n\n\nC_BENCH = os.environ.get('C_BENCH') or os.environ.get('CELERY_BENCH')\nC_BENCH_EVERY = int(os.environ.get('C_BENCH_EVERY') or\n                    os.environ.get('CELERY_BENCH_EVERY') or 1000)\nif C_BENCH:  # pragma: no cover\n    import atexit\n    from time import monotonic\n\n    from billiard.process import current_process\n\n    from celery.utils.debug import memdump, sample_mem\n\n    all_count = 0\n    bench_first = None\n    bench_start = None\n    bench_last = None\n    bench_every = C_BENCH_EVERY\n    bench_sample = []\n    __reserved = task_reserved\n    __ready = task_ready\n\n    if current_process()._name == 'MainProcess':\n        @atexit.register\n        def on_shutdown():\n            if bench_first is not None and bench_last is not None:\n                print('- Time spent in benchmark: {!r}'.format(\n                    bench_last - bench_first))\n                print('- Avg: {}'.format(\n                    sum(bench_sample) / len(bench_sample)))\n                memdump()\n\n    def task_reserved(request):\n        \"\"\"Called when a task is reserved by the worker.\"\"\"\n        global bench_start\n        global bench_first\n        now = None\n        if bench_start is None:\n            bench_start = now = monotonic()\n        if bench_first is None:\n            bench_first = now\n\n        return __reserved(request)\n\n    def task_ready(request):\n        \"\"\"Called when a task is completed.\"\"\"\n        global all_count\n        global bench_start\n        global bench_last\n        all_count += 1\n        if not all_count % bench_every:\n            now = monotonic()\n            diff = now - bench_start\n            print('- Time spent processing {} tasks (since first '\n                  'task received): ~{:.4f}s\\n'.format(bench_every, diff))\n            sys.stdout.flush()\n            bench_start = bench_last = now\n            bench_sample.append(diff)\n            sample_mem()\n        return __ready(request)\n\n\nclass Persistent:\n    \"\"\"Stores worker state between restarts.\n\n    This is the persistent data stored by the worker when\n    :option:`celery worker --statedb` is enabled.\n\n    Currently only stores revoked task id's.\n    \"\"\"\n\n    storage = shelve\n    protocol = pickle_protocol\n    compress = zlib.compress\n    decompress = zlib.decompress\n    _is_open = False\n\n    def __init__(self, state, filename, clock=None):\n        self.state = state\n        self.filename = filename\n        self.clock = clock\n        self.merge()\n\n    def open(self):\n        return self.storage.open(\n            self.filename, protocol=self.protocol, writeback=True,\n        )\n\n    def merge(self):\n        self._merge_with(self.db)\n\n    def sync(self):\n        self._sync_with(self.db)\n        self.db.sync()\n\n    def close(self):\n        if self._is_open:\n            self.db.close()\n            self._is_open = False\n\n    def save(self):\n        self.sync()\n        self.close()\n\n    def _merge_with(self, d):\n        self._merge_revoked(d)\n        self._merge_clock(d)\n        return d\n\n    def _sync_with(self, d):\n        self._revoked_tasks.purge()\n        d.update({\n            '__proto__': 3,\n            'zrevoked': self.compress(self._dumps(self._revoked_tasks)),\n            'clock': self.clock.forward() if self.clock else 0,\n        })\n        return d\n\n    def _merge_clock(self, d):\n        if self.clock:\n            d['clock'] = self.clock.adjust(d.get('clock') or 0)\n\n    def _merge_revoked(self, d):\n        try:\n            self._merge_revoked_v3(d['zrevoked'])\n        except KeyError:\n            try:\n                self._merge_revoked_v2(d.pop('revoked'))\n            except KeyError:\n                pass\n        # purge expired items at boot\n        self._revoked_tasks.purge()\n\n    def _merge_revoked_v3(self, zrevoked):\n        if zrevoked:\n            self._revoked_tasks.update(pickle.loads(self.decompress(zrevoked)))\n\n    def _merge_revoked_v2(self, saved):\n        if not isinstance(saved, LimitedSet):\n            # (pre 3.0.18) used to be stored as a dict\n            return self._merge_revoked_v1(saved)\n        self._revoked_tasks.update(saved)\n\n    def _merge_revoked_v1(self, saved):\n        add = self._revoked_tasks.add\n        for item in saved:\n            add(item)\n\n    def _dumps(self, obj):\n        return pickle.dumps(obj, protocol=self.protocol)\n\n    @property\n    def _revoked_tasks(self):\n        return self.state.revoked\n\n    @cached_property\n    def db(self):\n        self._is_open = True\n        return self.open()\n", "path": "celery/worker/state.py"}]}