From ba53cba2dc380c0ed16094b48cdb80dbd46f383a Mon Sep 17 00:00:00 2001 From: mprokopchuk Date: Thu, 9 Apr 2026 22:29:55 +0200 Subject: [PATCH] Fix concurrent publish bottleneck and agent ping task leaks MessageBusBase: replaced exclusive Gate with ReadWriteLock so multiple publishers can run in parallel. Subscriber callbacks are now invoked outside the lock to prevent slow hypervisor callbacks from starving write lock holders (subscribe/unsubscribe). Under heavy reconnect load (pod restart with many hosts) the old design serialized all publish() callers through a single gate, blocking API threads and causing full management server unresponsiveness. DirectAgentAttache: fixed a race where PingTask could be scheduled after disconnect() had already cleared _futures, causing it to never be cancelled. Switched PingTask scheduling from scheduleAtFixedRate to scheduleWithFixedDelay so slow hypervisor responses do not pile up concurrent ping executions and exhaust the cron thread pool. --- .../agent/manager/DirectAgentAttache.java | 121 ++++--- .../agent/manager/DirectAgentAttacheTest.java | 47 ++- .../framework/messagebus/MessageBusBase.java | 302 +++++------------- .../messagebus/MessageBusBaseTest.java | 106 ++++++ 4 files changed, 318 insertions(+), 258 deletions(-) create mode 100644 framework/ipc/src/test/java/org/apache/cloudstack/framework/messagebus/MessageBusBaseTest.java diff --git a/engine/orchestration/src/main/java/com/cloud/agent/manager/DirectAgentAttache.java b/engine/orchestration/src/main/java/com/cloud/agent/manager/DirectAgentAttache.java index 906a06da7ddc..7abfd0eaddc8 100644 --- a/engine/orchestration/src/main/java/com/cloud/agent/manager/DirectAgentAttache.java +++ b/engine/orchestration/src/main/java/com/cloud/agent/manager/DirectAgentAttache.java @@ -22,9 +22,11 @@ import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import org.apache.cloudstack.framework.config.ConfigKey; import org.apache.cloudstack.managed.context.ManagedContextRunnable; +import org.apache.logging.log4j.ThreadContext; import com.cloud.agent.api.Answer; import com.cloud.agent.api.Command; @@ -36,7 +38,6 @@ import com.cloud.exception.AgentUnavailableException; import com.cloud.host.Status; import com.cloud.resource.ServerResource; -import org.apache.logging.log4j.ThreadContext; public class DirectAgentAttache extends AgentAttache { @@ -44,12 +45,19 @@ public class DirectAgentAttache extends AgentAttache { "Number of times retrying a host ping while waiting for check results", true); protected final ConfigKey _HostPingRetryTimer = new ConfigKey("Advanced", Integer.class, "host.ping.retry.timer", "5", "Interval to wait before retrying a host ping while waiting for check results", true); - ServerResource _resource; - List> _futures = new ArrayList>(); - long _seq = 0; - LinkedList tasks = new LinkedList(); - AtomicInteger _outstandingTaskCount; - AtomicInteger _outstandingCronTaskCount; + // volatile so that isClosed() and PingTask/CronTask can read it without a lock. + // All writes go through _futuresLock to stay atomic with futures management. + private volatile ServerResource _resource; + private final List> _futures = new ArrayList>(); + // Separate lock for futures and _resource state. We intentionally do NOT use + // synchronized(this) here because disconnect() calls resource.disconnected() which + // can be slow (hypervisor roundtrip). A dedicated lock keeps that slow call outside + // the critical section so process() and send() are not blocked by it. + private final Object _futuresLock = new Object(); + private final AtomicLong _seq = new AtomicLong(0); + private final LinkedList tasks = new LinkedList(); + private final AtomicInteger _outstandingTaskCount; + private final AtomicInteger _outstandingCronTaskCount; public DirectAgentAttache(AgentManagerImpl agentMgr, long id, String uuid,String name, ServerResource resource, boolean maintenance) { super(agentMgr, id, uuid, name, maintenance); @@ -62,15 +70,21 @@ public DirectAgentAttache(AgentManagerImpl agentMgr, long id, String uuid,String public void disconnect(Status state) { logger.debug("Processing disconnect [id: {}, uuid: {}, name: {}]", _id, _uuid, _name); - for (ScheduledFuture future : _futures) { - future.cancel(false); - } - - synchronized (this) { - if (_resource != null) { - _resource.disconnected(); - _resource = null; + // Capture the resource reference and null it out atomically with futures cleanup, + // so that process() and send() cannot sneak in a new scheduled task after we clear. + // We call resource.disconnected() outside the lock intentionally - it can be slow + // (calls into the hypervisor driver), and we don't want to hold _futuresLock during that. + ServerResource resource; + synchronized (_futuresLock) { + for (ScheduledFuture future : _futures) { + future.cancel(false); } + _futures.clear(); + resource = _resource; + _resource = null; + } + if (resource != null) { + resource.disconnected(); } } @@ -83,7 +97,7 @@ public boolean equals(Object obj) { } @Override - public synchronized boolean isClosed() { + public boolean isClosed() { return _resource == null; } @@ -96,7 +110,14 @@ public void send(Request req) throws AgentUnavailableException { if (answers != null && answers[0] instanceof StartupAnswer) { StartupAnswer startup = (StartupAnswer)answers[0]; int interval = startup.getPingInterval(); - _futures.add(_agentMgr.getCronJobPool().scheduleAtFixedRate(new PingTask(), interval, interval, TimeUnit.SECONDS)); + synchronized (_futuresLock) { + if (!isClosed()) { + // scheduleWithFixedDelay - next ping starts only after the previous one + // finishes. scheduleAtFixedRate would pile up concurrent pings if the + // hypervisor is slow, eventually exhausting the cron thread pool. + _futures.add(_agentMgr.getCronJobPool().scheduleWithFixedDelay(new PingTask(), interval, interval, TimeUnit.SECONDS)); + } + } } } else { Command[] cmds = req.getCommands(); @@ -105,7 +126,11 @@ public void send(Request req) throws AgentUnavailableException { scheduleFromQueue(); } else { CronCommand cmd = (CronCommand)cmds[0]; - _futures.add(_agentMgr.getCronJobPool().scheduleAtFixedRate(new CronTask(req), cmd.getInterval(), cmd.getInterval(), TimeUnit.SECONDS)); + synchronized (_futuresLock) { + if (!isClosed()) { + _futures.add(_agentMgr.getCronJobPool().scheduleAtFixedRate(new CronTask(req), cmd.getInterval(), cmd.getInterval(), TimeUnit.SECONDS)); + } + } } } } @@ -115,8 +140,17 @@ public void process(Answer[] answers) { if (answers != null && answers[0] instanceof StartupAnswer) { StartupAnswer startup = (StartupAnswer)answers[0]; int interval = startup.getPingInterval(); - logger.info("StartupAnswer received [id: {}, uuid: {}, name: {}, interval: {}]", startup.getHostId(), startup.getHostUuid(), startup.getHostName(), interval); - _futures.add(_agentMgr.getCronJobPool().scheduleAtFixedRate(new PingTask(), interval, interval, TimeUnit.SECONDS)); + logger.info(String.format( + "StartupAnswer received [id: %d, uuid: %s, name: %s, interval: %d]", + startup.getHostId(), startup.getHostUuid(), startup.getHostName(), interval)); + synchronized (_futuresLock) { + if (!isClosed()) { + // scheduleWithFixedDelay - next ping starts only after the previous one + // finishes. scheduleAtFixedRate would pile up concurrent pings if the + // hypervisor is slow, eventually exhausting the cron thread pool. + _futures.add(_agentMgr.getCronJobPool().scheduleWithFixedDelay(new PingTask(), interval, interval, TimeUnit.SECONDS)); + } + } } } @@ -124,11 +158,9 @@ public void process(Answer[] answers) { protected void finalize() throws Throwable { try { assert _resource == null : "Come on now....If you're going to dabble in agent code, you better know how to close out our resources. Ever considered why there's a method called disconnect()?"; - synchronized (this) { - if (_resource != null) { - logger.warn("Lost attache for [id: {}, uuid: {}, name: {}]", _id, _uuid, _name); - disconnect(Status.Alert); - } + if (_resource != null) { + logger.warn(String.format("Lost attache for [id: %d, uuid: %s, name: %s]", _id, _uuid, _name)); + disconnect(Status.Alert); } } finally { super.finalize(); @@ -143,8 +175,19 @@ private synchronized void scheduleFromQueue() { logger.trace("Agent attache [id: {}, uuid: {}, name: {}], task queue size={}, outstanding tasks={}", _id, _uuid, _name, tasks.size(), _outstandingTaskCount.get()); while (!tasks.isEmpty() && _outstandingTaskCount.get() < _agentMgr.getDirectAgentThreadCap()) { + Task task = tasks.removeFirst(); _outstandingTaskCount.incrementAndGet(); - _agentMgr.getDirectAgentPool().execute(tasks.remove()); + try { + _agentMgr.getDirectAgentPool().execute(task); + } catch (RuntimeException e) { + // The thread pool rejected the task (likely full or shutting down). + // Roll back: return the slot and put the task back at the head of the queue + // so it gets a chance to run on the next scheduleFromQueue() call. + _outstandingTaskCount.decrementAndGet(); + tasks.addFirst(task); + logger.warn("Failed to submit direct agent task, will retry on next schedule", e); + break; + } } } @@ -155,12 +198,12 @@ public int hashCode() { protected class PingTask extends ManagedContextRunnable { @Override - protected synchronized void runInContext() { + protected void runInContext() { try { - if (_outstandingCronTaskCount.incrementAndGet() >= _agentMgr.getDirectAgentThreadCap()) { - logger.warn( - "PingTask execution for direct attache [id: {}, uuid: {}, name: {}] has reached maximum outstanding limit({}), bailing out", - _id, _uuid, _name, _agentMgr.getDirectAgentThreadCap()); + if (_outstandingCronTaskCount.incrementAndGet() > _agentMgr.getDirectAgentThreadCap()) { + logger.warn(String.format( + "PingTask execution for direct attache [id: %d, uuid: %s, name: %s] has reached maximum outstanding limit(%d), bailing out", + _id, _uuid, _name, _agentMgr.getDirectAgentThreadCap())); return; } @@ -183,7 +226,7 @@ protected synchronized void runInContext() { ThreadContext.put("logcontextid", cmd.getContextParam("logid")); } logger.debug("Ping from [id: {}, uuid: {}, name: {}]", _id, _uuid, _name); - long seq = _seq++; + long seq = _seq.getAndIncrement(); logger.trace("SeqA {}-{}: {}", _id, seq, new Request(_id, -1, cmd, false).toString()); @@ -200,8 +243,7 @@ protected synchronized void runInContext() { } protected class CronTask extends ManagedContextRunnable { - Request _req; - + private final Request _req; public CronTask(Request req) { _req = req; } @@ -226,10 +268,10 @@ private void bailout() { protected void runInContext() { long seq = _req.getSequence(); try { - if (_outstandingCronTaskCount.incrementAndGet() >= _agentMgr.getDirectAgentThreadCap()) { - logger.warn( - "CronTask execution for direct attache [id: {}, uuid: {}, name: {}] has reached maximum outstanding limit({}), bailing out", - _id, _uuid, _name, _agentMgr.getDirectAgentThreadCap()); + if (_outstandingCronTaskCount.incrementAndGet() > _agentMgr.getDirectAgentThreadCap()) { + logger.warn(String.format( + "CronTask execution for direct attache [id: %d, uuid: %s, name: %s] has reached maximum outstanding limit(%d), bailing out", + _id, _uuid, _name, _agentMgr.getDirectAgentThreadCap())); bailout(); return; } @@ -282,8 +324,7 @@ protected void runInContext() { } protected class Task extends ManagedContextRunnable { - Request _req; - + private final Request _req; public Task(Request req) { _req = req; } diff --git a/engine/orchestration/src/test/java/com/cloud/agent/manager/DirectAgentAttacheTest.java b/engine/orchestration/src/test/java/com/cloud/agent/manager/DirectAgentAttacheTest.java index 65e31c271a42..d24f26249e4a 100644 --- a/engine/orchestration/src/test/java/com/cloud/agent/manager/DirectAgentAttacheTest.java +++ b/engine/orchestration/src/test/java/com/cloud/agent/manager/DirectAgentAttacheTest.java @@ -16,6 +16,15 @@ // under the License. package com.cloud.agent.manager; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; + +import java.util.UUID; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -24,10 +33,11 @@ import org.mockito.MockitoAnnotations; import org.mockito.junit.MockitoJUnitRunner; +import com.cloud.agent.api.Answer; +import com.cloud.agent.api.StartupAnswer; +import com.cloud.host.Status; import com.cloud.resource.ServerResource; -import java.util.UUID; - @RunWith(MockitoJUnitRunner.class) public class DirectAgentAttacheTest { @Mock @@ -36,6 +46,12 @@ public class DirectAgentAttacheTest { @Mock private ServerResource _resource; + @Mock + private ScheduledExecutorService _cronJobPool; + + @Mock + private ScheduledFuture _future; + long _id = 0L; String _uuid = UUID.randomUUID().toString(); @@ -55,4 +71,31 @@ public void testPingTask() throws Exception { pt.runInContext(); Mockito.verify(_resource, Mockito.times(1)).getCurrentStatus(_id); } + + @Test + public void testProcessSchedulesPingWhenConnected() { + Mockito.doReturn(_cronJobPool).when(_agentMgr).getCronJobPool(); + Mockito.doReturn(_future).when(_cronJobPool).scheduleWithFixedDelay(any(Runnable.class), anyLong(), anyLong(), any(TimeUnit.class)); + + directAgentAttache.process(new Answer[] {buildStartupAnswer()}); + + Mockito.verify(_cronJobPool, Mockito.times(1)).scheduleWithFixedDelay(any(Runnable.class), eq(60L), eq(60L), eq(TimeUnit.SECONDS)); + } + + @Test + public void testProcessDoesNotSchedulePingAfterDisconnect() { + // Once disconnect() has cleared the resource, a late StartupAnswer must not schedule + // a PingTask - otherwise the future would never be cancelled (the leak this fix targets). + directAgentAttache.disconnect(Status.Disconnected); + + directAgentAttache.process(new Answer[] {buildStartupAnswer()}); + + Mockito.verify(_cronJobPool, Mockito.never()).scheduleWithFixedDelay(any(Runnable.class), anyLong(), anyLong(), any(TimeUnit.class)); + } + + private StartupAnswer buildStartupAnswer() { + StartupAnswer startup = Mockito.mock(StartupAnswer.class); + Mockito.doReturn(60).when(startup).getPingInterval(); + return startup; + } } diff --git a/framework/ipc/src/main/java/org/apache/cloudstack/framework/messagebus/MessageBusBase.java b/framework/ipc/src/main/java/org/apache/cloudstack/framework/messagebus/MessageBusBase.java index 742fd90c33e0..eed368139e50 100644 --- a/framework/ipc/src/main/java/org/apache/cloudstack/framework/messagebus/MessageBusBase.java +++ b/framework/ipc/src/main/java/org/apache/cloudstack/framework/messagebus/MessageBusBase.java @@ -21,23 +21,45 @@ import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; - -import org.apache.logging.log4j.Logger; -import org.apache.logging.log4j.LogManager; +import java.util.Set; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; import org.apache.cloudstack.framework.serializer.MessageSerializer; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; import com.cloud.utils.db.TransactionLegacy; import com.cloud.utils.exception.CloudRuntimeException; +/** + * MessageBus implementation based on a hierarchical topic tree. + * + * Locking model: + * Multiple publishers can run concurrently - publish() takes a read lock, + * which only blocks when someone is modifying the subscriber tree (subscribe, + * unsubscribe, clearAll, prune). This means a pod restart with hundreds of + * hosts reconnecting and firing events simultaneously will not serialize + * through a single bottleneck. + * + * Subscriber callbacks are intentionally called OUTSIDE the read lock. + * Holding a lock during external callbacks is a classic source of production + * outages: a slow subscriber (DB call, GC pause, hypervisor roundtrip) would + * block ALL other publishers and eventually starve write lock holders + * (subscribe/unsubscribe). Instead we snapshot the subscriber list under + * the lock and release it before calling anyone. + */ public class MessageBusBase implements MessageBus { - private final Gate _gate; - private final List _pendingActions; + // Fair mode: a queued writer (subscribe/unsubscribe/prune/clearAll) is served ahead of + // newly arriving readers, so a steady stream of publish() read locks cannot starve writers. + private final ReadWriteLock _lock = new ReentrantReadWriteLock(true); private final SubscriptionNode _subscriberRoot; private MessageSerializer _messageSerializer; @@ -45,9 +67,6 @@ public class MessageBusBase implements MessageBus { protected Logger logger = LogManager.getLogger(getClass()); public MessageBusBase() { - _gate = new Gate(); - _pendingActions = new ArrayList(); - _subscriberRoot = new SubscriptionNode(null, "/", null); } @@ -65,82 +84,54 @@ public MessageSerializer getMessageSerializer() { public void subscribe(String subject, MessageSubscriber subscriber) { assert (subject != null); assert (subscriber != null); - if (_gate.enter()) { - if (logger.isTraceEnabled()) { - logger.trace("Enter gate in message bus subscribe"); - } - try { - SubscriptionNode current = locate(subject, null, true); - assert (current != null); - current.addSubscriber(subscriber); - } finally { - _gate.leave(); - } - } else { - synchronized (_pendingActions) { - _pendingActions.add(new ActionRecord(ActionType.Subscribe, subject, subscriber)); - } + _lock.writeLock().lock(); + try { + logger.trace("Acquired write lock in message bus subscribe"); + SubscriptionNode current = locate(subject, null, true); + assert (current != null); + current.addSubscriber(subscriber); + } finally { + _lock.writeLock().unlock(); } } @Override public void unsubscribe(String subject, MessageSubscriber subscriber) { - if (_gate.enter()) { - if (logger.isTraceEnabled()) { - logger.trace("Enter gate in message bus unsubscribe"); - } - try { - if (subject != null) { - SubscriptionNode current = locate(subject, null, false); - if (current != null) - current.removeSubscriber(subscriber, false); - } else { - _subscriberRoot.removeSubscriber(subscriber, true); - } - } finally { - _gate.leave(); - } - } else { - synchronized (_pendingActions) { - _pendingActions.add(new ActionRecord(ActionType.Unsubscribe, subject, subscriber)); + _lock.writeLock().lock(); + try { + logger.trace("Acquired write lock in message bus unsubscribe"); + if (subject != null) { + SubscriptionNode current = locate(subject, null, false); + if (current != null) + current.removeSubscriber(subscriber, false); + } else { + _subscriberRoot.removeSubscriber(subscriber, true); } + } finally { + _lock.writeLock().unlock(); } } @Override public void clearAll() { - if (_gate.enter()) { - if (logger.isTraceEnabled()) { - logger.trace("Enter gate in message bus clearAll"); - } - try { - _subscriberRoot.clearAll(); - doPrune(); - } finally { - _gate.leave(); - } - } else { - synchronized (_pendingActions) { - _pendingActions.add(new ActionRecord(ActionType.ClearAll, null, null)); - } + _lock.writeLock().lock(); + try { + logger.trace("Acquired write lock in message bus clearAll"); + _subscriberRoot.clearAll(); + doPrune(); + } finally { + _lock.writeLock().unlock(); } } @Override public void prune() { - if (_gate.enter()) { - if (logger.isTraceEnabled()) { - logger.trace("Enter gate in message bus prune"); - } - try { - doPrune(); - } finally { - _gate.leave(); - } - } else { - synchronized (_pendingActions) { - _pendingActions.add(new ActionRecord(ActionType.Prune, null, null)); - } + _lock.writeLock().lock(); + try { + logger.trace("Acquired write lock in message bus prune"); + doPrune(); + } finally { + _lock.writeLock().unlock(); } } @@ -163,66 +154,36 @@ private void doPrune() { @Override public void publish(String senderAddress, String subject, PublishScope scope, Object args) { // publish cannot be in DB transaction, which may hold DB lock too long, and we are guarding this here - if (!noDbTxn()){ + if (!noDbTxn()) { String errMsg = "NO EVENT PUBLISH CAN BE WRAPPED WITHIN DB TRANSACTION!"; logger.error(errMsg, new CloudRuntimeException(errMsg)); } - if (_gate.enter(true)) { - if (logger.isTraceEnabled()) { - logger.trace("Enter gate in message bus publish"); - } - try { - List chainFromTop = new ArrayList(); - SubscriptionNode current = locate(subject, chainFromTop, false); - - if (current != null) - current.notifySubscribers(senderAddress, subject, args); - - Collections.reverse(chainFromTop); - for (SubscriptionNode node : chainFromTop) - node.notifySubscribers(senderAddress, subject, args); - } finally { - _gate.leave(); - } + // Collect subscribers under read lock (fast - just tree traversal and list copy), + // then release the lock before calling any callbacks. + // LinkedHashSet deduplicates: a subscriber registered on both "Host" and "Host.123" + // gets exactly one callback when "Host.123" is published. + Set toNotify = new LinkedHashSet<>(); + _lock.readLock().lock(); + try { + logger.trace("Acquired read lock in message bus publish"); + List chainFromTop = new ArrayList<>(); + SubscriptionNode current = locate(subject, chainFromTop, false); + + if (current != null) + current.collectSubscribers(toNotify); + + Collections.reverse(chainFromTop); + for (SubscriptionNode node : chainFromTop) + node.collectSubscribers(toNotify); + } finally { + _lock.readLock().unlock(); } - } - - private void onGateOpen() { - synchronized (_pendingActions) { - ActionRecord record = null; - while (_pendingActions.size() > 0) { - record = _pendingActions.remove(0); - switch (record.getType()) { - case Subscribe: { - SubscriptionNode current = locate(record.getSubject(), null, true); - assert (current != null); - current.addSubscriber(record.getSubscriber()); - } - break; - - case Unsubscribe: - if (record.getSubject() != null) { - SubscriptionNode current = locate(record.getSubject(), null, false); - if (current != null) - current.removeSubscriber(record.getSubscriber(), false); - } else { - _subscriberRoot.removeSubscriber(record.getSubscriber(), true); - } - break; - - case ClearAll: - _subscriberRoot.clearAll(); - break; - - case Prune: - doPrune(); - break; - - default: - assert (false); - break; - } + for (MessageSubscriber subscriber : toNotify) { + try { + subscriber.onPublishMessage(senderAddress, subject, args); + } catch (Throwable t) { + logger.error("Subscriber threw an exception during publish of subject: " + subject + " scope: " + scope + " args: " + args, t); } } } @@ -272,95 +233,6 @@ private boolean noDbTxn() { // // Support inner classes // - private static enum ActionType { - Subscribe, Unsubscribe, ClearAll, Prune - } - - private static class ActionRecord { - private final ActionType _type; - private final String _subject; - private final MessageSubscriber _subscriber; - - public ActionRecord(ActionType type, String subject, MessageSubscriber subscriber) { - _type = type; - _subject = subject; - _subscriber = subscriber; - } - - public ActionType getType() { - return _type; - } - - public String getSubject() { - return _subject; - } - - public MessageSubscriber getSubscriber() { - return _subscriber; - } - } - - private class Gate { - private int _reentranceCount; - private Thread _gateOwner; - - public Gate() { - _reentranceCount = 0; - _gateOwner = null; - } - - public boolean enter() { - return enter(false); - } - - public boolean enter(boolean wait) { - while (true) { - synchronized (this) { - if (_reentranceCount == 0) { - assert (_gateOwner == null); - - _reentranceCount++; - _gateOwner = Thread.currentThread(); - return true; - } else { - if (wait) { - try { - wait(); - } catch (InterruptedException e) { - logger.debug("[ignored] interrupted while guarding re-entrance on message bus."); - } - } else { - break; - } - } - } - } - - return false; - } - - public void leave() { - synchronized (this) { - if (_reentranceCount > 0) { - try { - assert (_gateOwner == Thread.currentThread()); - - onGateOpen(); - } finally { - if (logger.isTraceEnabled()) { - logger.trace("Open gate of message bus"); - } - _reentranceCount--; - assert (_reentranceCount == 0); - _gateOwner = null; - - notifyAll(); - } - } - } - } - } - private static class SubscriptionNode { private final String _nodeKey; private final List _subscribers; @@ -437,10 +309,8 @@ public void prune(List trimNodes) { trimNodes.add(this); } - public void notifySubscribers(String senderAddress, String subject, Object args) { - for (MessageSubscriber subscriber : _subscribers) { - subscriber.onPublishMessage(senderAddress, subject, args); - } + public void collectSubscribers(Collection target) { + target.addAll(_subscribers); } public boolean isTrimmable() { diff --git a/framework/ipc/src/test/java/org/apache/cloudstack/framework/messagebus/MessageBusBaseTest.java b/framework/ipc/src/test/java/org/apache/cloudstack/framework/messagebus/MessageBusBaseTest.java new file mode 100644 index 000000000000..c13e813e3f5f --- /dev/null +++ b/framework/ipc/src/test/java/org/apache/cloudstack/framework/messagebus/MessageBusBaseTest.java @@ -0,0 +1,106 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.cloudstack.framework.messagebus; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import com.cloud.utils.db.TransactionLegacy; + +public class MessageBusBaseTest { + + private MessageBusBase bus; + private TransactionLegacy txn; + + @Before + public void setUp() { + bus = new MessageBusBase(); + // publish() calls noDbTxn() which needs a thread-local transaction to exist. + // open() only registers a lazy transaction; it does not open a DB connection. + txn = TransactionLegacy.open("MessageBusBaseTest"); + } + + @After + public void tearDown() { + if (txn != null) { + txn.close(); + } + } + + /** + * A subscriber registered on both an ancestor ("Host") and a descendant ("Host.123") + * must receive exactly one callback when the descendant topic is published, not one per + * matching node. This pins the LinkedHashSet dedup behavior in publish(). + */ + @Test + public void testSubscriberOnAncestorAndDescendantNotifiedOnce() { + CountingSubscriber subscriber = new CountingSubscriber(); + bus.subscribe("Host", subscriber); + bus.subscribe("Host.123", subscriber); + + bus.publish(null, "Host.123", PublishScope.LOCAL, null); + + Assert.assertEquals(1, subscriber.count); + } + + /** + * Distinct subscribers on ancestor and descendant are both notified, and the more specific + * (descendant) subscriber is notified before the ancestor one. + */ + @Test + public void testDescendantNotifiedBeforeAncestor() { + final List order = new ArrayList<>(); + MessageSubscriber ancestor = (sender, subject, args) -> order.add("ancestor"); + MessageSubscriber descendant = (sender, subject, args) -> order.add("descendant"); + bus.subscribe("Host", ancestor); + bus.subscribe("Host.123", descendant); + + bus.publish(null, "Host.123", PublishScope.LOCAL, null); + + Assert.assertEquals(List.of("descendant", "ancestor"), order); + } + + /** + * A subscriber on a sibling topic ("Host.456") must not be notified for "Host.123". + */ + @Test + public void testSiblingSubscriberNotNotified() { + CountingSubscriber sibling = new CountingSubscriber(); + bus.subscribe("Host.456", sibling); + + bus.publish(null, "Host.123", PublishScope.LOCAL, null); + + Assert.assertEquals(0, sibling.count); + } + + private static class CountingSubscriber implements MessageSubscriber { + int count = 0; + + @Override + public void onPublishMessage(String senderAddress, String subject, Object args) { + count++; + } + } +}