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.

Author: mprokopchuk

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,20 +38,26 @@
 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 {
 
     protected final ConfigKey<Integer> _HostPingRetryCount = new ConfigKey<Integer>("Advanced", Integer.class, "host.ping.retry.count", "0",
             "Number of times retrying a host ping while waiting for check results", true);
     protected final ConfigKey<Integer> _HostPingRetryTimer = new ConfigKey<Integer>("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<ScheduledFuture<?>> _futures = new ArrayList<ScheduledFuture<?>>();
-    long _seq = 0;
-    LinkedList<Task> tasks = new LinkedList<Task>();
-    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<ScheduledFuture<?>> _futures = new ArrayList<ScheduledFuture<?>>();
+    // 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<Task> tasks = new LinkedList<Task>();
+    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,20 +140,27 @@ 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));
+                }
+            }
         }
     }
 
     @Override
     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;
         }

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;
+    }
 }