pulse_worker: reuse kombu Consumer (bug 1974350)

docker-compose.yaml

@@ -20,6 +20,7 @@ services:
     environment:
       - EMAIL='docker@sync'
       - RABBITMQ=true
+      - TRACEMALLOC=1
     depends_on:
       sync_init:
         condition: service_completed_successfully

git_hg_sync/application.py

@@ -3,6 +3,7 @@
 import os
 import signal
 import sys
+import tracemalloc
 from collections.abc import Sequence
 from types import FrameType
 
@@ -18,6 +19,10 @@
 
 
 class Application:
+    _event_count: int = 0
+    _prev_snapshot = None
+    _snapshot = None
+
     def __init__(
         self,
         worker: PulseWorker,
@@ -83,6 +88,7 @@ def _handle_push_event(self, push_event: Push) -> None:
         )
 
     def _handle_event(self, event: Event) -> None:
+        self._trace_malloc()
         if event.repo_url not in self._repo_synchronizers:
             logger.warning(f"Ignoring event for untracked repository: {event.repo_url}")
             return
@@ -91,3 +97,27 @@ def _handle_event(self, event: Event) -> None:
                 self._handle_push_event(event)
             case _:
                 raise NotImplementedError()
+
+    def _trace_malloc(self) -> None:
+        self._event_count += 1
+        if not self._event_count % 10:
+            self._event_count = 0
+            if tracemalloc.is_tracing():
+                self._snapshot = tracemalloc.take_snapshot()
+
+                for i, stat in enumerate(
+                    self._snapshot.statistics("lineno")[:25],
+                    1,
+                ):
+                    logger.info(
+                        f"tracemalloc absolute, {i}, {stat.traceback[0].filename}, {stat.traceback[0].lineno}, {stat.size / 1024}, {stat.count}"
+                    )
+                if self._prev_snapshot:
+                    for i, stat in enumerate(
+                        self._snapshot.compare_to(self._prev_snapshot, "lineno")[:25],
+                        1,
+                    ):
+                        logger.info(
+                            f"tracemalloc diff, {i}, {stat.traceback[0].filename}, {stat.traceback[0].lineno}, {stat.size_diff / 1024}, {stat.count_diff}"
+                        )
+                self._prev_snapshot = self._snapshot