From ee009432480e74f4862e181c28d56efb01b16d5b Mon Sep 17 00:00:00 2001 From: Simon Westphahl Date: Wed, 19 Feb 2025 14:12:24 +0100 Subject: [PATCH] Only notify pipelines once about semaphore release We've seen a large number of semaphore release events being processed each leading to multiple pipeline semaphore release events for the same pipeline. Since the pipeline semaphore release event is just a notification and doesn't carry any information it is enough to notify the pipeline once. To achieve this we track tenants that have been notified of a semaphore release and don't send additional pipeline semaphore release events in the same iteration of the tenant management event queue. Change-Id: Idedaf5496430a218a1c4b79fdc3b4c58824a7da9 --- zuul/scheduler.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/zuul/scheduler.py b/zuul/scheduler.py index 18888cefae..eef34f3139 100644 --- a/zuul/scheduler.py +++ b/zuul/scheduler.py @@ -2084,7 +2084,7 @@ class Scheduler(threading.Thread): pipeline.manager.removeItem(item) return - def _doSemaphoreReleaseEvent(self, event, tenant): + def _doSemaphoreReleaseEvent(self, event, tenant, notified): semaphore = tenant.layout.getSemaphore( self.abide, event.semaphore_name) if semaphore.global_scope: @@ -2093,11 +2093,14 @@ class Scheduler(threading.Thread): else: tenants = [tenant] for tenant in tenants: + if tenant.name in notified: + continue for pipeline_name in tenant.layout.pipelines.keys(): event = PipelineSemaphoreReleaseEvent() self.pipeline_management_events[ tenant.name][pipeline_name].put( event, needs_result=False) + notified.add(tenant.name) def _areAllBuildsComplete(self): self.log.debug("Checking if all builds are complete") @@ -2660,6 +2663,9 @@ class Scheduler(threading.Thread): " in tenant %s", tenant.name) def _process_tenant_management_queue(self, tenant): + # Set of tenant names that were notified of + # a semaphore release. + semaphore_notified = set() for event in self.management_events[tenant.name]: event_forwarded = False try: @@ -2670,7 +2676,8 @@ class Scheduler(threading.Thread): elif isinstance(event, (PromoteEvent, ChangeManagementEvent)): event_forwarded = self._forward_management_event(event) elif isinstance(event, SemaphoreReleaseEvent): - self._doSemaphoreReleaseEvent(event, tenant) + self._doSemaphoreReleaseEvent( + event, tenant, semaphore_notified) else: self.log.error("Unable to handle event %s for tenant %s", event, tenant.name) @@ -2797,7 +2804,7 @@ class Scheduler(threading.Thread): # MODEL_API <= 32 # Kept for backward compatibility; semaphore release events # are now processed in the management event queue. - self._doSemaphoreReleaseEvent(event, pipeline.tenant) + self._doSemaphoreReleaseEvent(event, pipeline.tenant, set()) else: self.log.error("Unable to handle event %s", event)