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
This commit is contained in:
+10
-3
@@ -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)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user