[Prefork] Reset celery.worker.state after fork

celery/concurrency/prefork.py

@@ -79,6 +79,8 @@ def process_initializer(app, hostname):
     for name, task in items(app.tasks):
         task.__trace__ = build_tracer(name, task, app.loader, hostname,
                                       app=app)
+    from celery.worker import state as worker_state
+    worker_state.reset_state()
     signals.worker_process_init.send(sender=None)
 
 

celery/worker/state.py

@@ -64,6 +64,14 @@ should_stop = False
 should_terminate = False
 
 
+def reset_state():
+    reserved_requests.clear()
+    active_requests.clear()
+    total_count.clear()
+    all_total_count[:] = [0]
+    revoked.clear()
+
+
 def maybe_shutdown():
     if should_stop:
         raise WorkerShutdown()