CELERY_BENCH envvar now enables benchmarking of time spent processing 1000 tasks

celery/worker/state.py

@@ -1,3 +1,4 @@
+import os
 import platform
 import shelve
 
@@ -49,6 +50,33 @@ def task_ready(request):
     reserved_requests.discard(request)
 
 
+if os.environ.get("CELERY_BENCH"):
+    from time import time
+
+    all_count = 0
+    bench_start = None
+    bench_every = int(os.environ.get("CELERY_BENCH_EVERY", 1000))
+    __reserved = task_reserved
+    __ready = task_ready
+
+    def task_reserved(request):
+        global bench_start
+        if bench_start is None:
+            bench_start = time()
+        return __reserved(request)
+
+    def task_ready(request):
+        global all_count, bench_start
+        all_count += 1
+        if not all_count % bench_every:
+            print("* Time spent processing %s tasks (since first "
+                    "task received): ~%.4fs\n" % (
+                bench_every, time() - bench_start))
+            bench_start = None
+
+        return __ready(request)
+
+
 class Persistent(object):
     storage = shelve
     _is_open = False