[Messenger] Run the cycle collector only when the Worker is idle

TimWolla · nicolas-grekas · TimWolla · commit 9b908ca31086 · 2025-04-08T15:52:12.000+02:00
Co-authored-by: Nicolas Grekas &lt;nicolas.grekas@gmail.com&gt;
diff --git a/src/Symfony/Component/Messenger/EventListener/ResetMemoryUsageListener.php b/src/Symfony/Component/Messenger/EventListener/ResetMemoryUsageListener.php
@@ -12,35 +12,45 @@
 namespace Symfony\Component\Messenger\EventListener;
 
 use Symfony\Component\EventDispatcher\EventSubscriberInterface;
-use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
-use Symfony\Component\Messenger\Event\WorkerMessageHandledEvent;
 use Symfony\Component\Messenger\Event\WorkerMessageReceivedEvent;
+use Symfony\Component\Messenger\Event\WorkerRunningEvent;
 
 /**
  * @author Tim Düsterhus <tim@tideways-gmbh.com>
  */
 class ResetMemoryUsageListener implements EventSubscriberInterface
 {
+    /**
+     * Indicates whether the GC already ran during the current idle phase.
+     */
+    private bool $collected = false;
+
     public function resetBefore(WorkerMessageReceivedEvent $event): void
     {
         // Reset the peak memory usage for accurate measurement of the
         // memory usage on a per-message basis.
         memory_reset_peak_usage();
     }
 
-    public function collectAfter(WorkerMessageHandledEvent|WorkerMessageFailedEvent $event): void
+    public function collectAfter(WorkerRunningEvent $event): void
     {
-        // Run the cycle collector after handling a message to avoid it
-        // running while the message is reserved and in processing.
-        gc_collect_cycles();
+        // Try to avoid running the GC while a message is handled by running it once while the worker is idle
+        if ($event->isWorkerIdle()) {
+            if (!$this->collected) {
+                gc_collect_cycles();
+
+                $this->collected = true;
+            }
+        } else {
+            $this->collected = false;
+        }
     }
 
     public static function getSubscribedEvents(): array
     {
         return [
             WorkerMessageReceivedEvent::class => ['resetBefore', -1024],
-            WorkerMessageFailedEvent::class => ['collectAfter', -1024],
-            WorkerMessageHandledEvent::class => ['collectAfter', -1024],
+            WorkerRunningEvent::class => ['collectAfter', -1024],
         ];
     }
 }
diff --git a/src/Symfony/Component/Messenger/Tests/WorkerTest.php b/src/Symfony/Component/Messenger/Tests/WorkerTest.php
@@ -587,7 +587,7 @@ public function testFlushBatchOnStop()
         $this->assertSame($expectedMessages, $handler->processedMessages);
     }
 
-    public function testGcCollectCyclesIsCalledOnMessageHandle()
+    public function testGcCollectCyclesIsCalledOnIdleWorker()
     {
         $apiMessage = new DummyMessage('API');
 
@@ -597,14 +597,35 @@ public function testGcCollectCyclesIsCalledOnMessageHandle()
 
         $dispatcher = new EventDispatcher();
         $dispatcher->addSubscriber(new ResetMemoryUsageListener());
-        $dispatcher->addSubscriber(new StopWorkerOnMessageLimitListener(1));
+        $before = 0;
+        $dispatcher->addListener(WorkerRunningEvent::class, function (WorkerRunningEvent $event) use (&$before) {
+            static $i = 0;
+
+            $after = gc_status()['runs'];
+            if (0 === $i) {
+                $this->assertFalse($event->isWorkerIdle());
+                $this->assertSame(0, $after - $before);
+            } else if (1 === $i) {
+                $this->assertTrue($event->isWorkerIdle());
+                $this->assertSame(1, $after - $before);
+            } else if (3 === $i) {
+                // Wait a few idle phases before stopping.
+                $this->assertSame(1, $after - $before);
+                $event->getWorker()->stop();
+            }
+
+            $i++;
+        }, PHP_INT_MIN);
+
 
         $worker = new Worker(['transport' => $receiver], $bus, $dispatcher);
-        $worker->run();
 
-        $gcStatus = gc_status();
+        gc_collect_cycles();
+        $before = gc_status()['runs'];
 
-        $this->assertGreaterThan(0, $gcStatus['runs']);
+        $worker->run([
+            'sleep' => 0,
+        ]);
     }
 
     public function testMemoryUsageIsResetOnMessageHandle()
