ARTEMIS-2394 Improve scheduling sync for AMQPConnectionContext

Remove synchronized blocks using an AtomicReference.

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/logger/ActiveMQAMQPProtocolLogger.java

@@ -0,0 +1,51 @@
+/*
+ * 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.activemq.artemis.protocol.amqp.logger;
+
+import org.jboss.logging.BasicLogger;
+import org.jboss.logging.Logger;
+import org.jboss.logging.annotations.LogMessage;
+import org.jboss.logging.annotations.Message;
+import org.jboss.logging.annotations.MessageLogger;
+
+/**
+ * Logger Code 11
+ *
+ * each message id must be 6 digits long starting with 33, the 3rd digit donates the level so
+ *
+ * INF0  1
+ * WARN  2
+ * DEBUG 3
+ * ERROR 4
+ * TRACE 5
+ * FATAL 6
+ *
+ * so an INFO message would be 111000 to 111999
+ */
+
+@MessageLogger(projectCode = "AMQ")
+public interface ActiveMQAMQPProtocolLogger extends BasicLogger {
+
+   /**
+    * The default logger.
+    */
+   ActiveMQAMQPProtocolLogger LOGGER = Logger.getMessageLogger(ActiveMQAMQPProtocolLogger.class, ActiveMQAMQPProtocolLogger.class.getPackage().getName());
+
+   @LogMessage(level = Logger.Level.WARN)
+   @Message(id = 111000, value = "Scheduled task can't be removed from scheduledPool.", format = Message.Format.MESSAGE_FORMAT)
+   void cantRemovingScheduledTask();
+}

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPConnectionContext.java

@@ -22,10 +22,13 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.UnaryOperator;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.EventLoop;
@@ -35,6 +38,7 @@ import org.apache.activemq.artemis.protocol.amqp.broker.AMQPConnectionCallback;
 import org.apache.activemq.artemis.protocol.amqp.broker.AMQPSessionCallback;
 import org.apache.activemq.artemis.protocol.amqp.broker.ProtonProtocolManager;
 import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPException;
+import org.apache.activemq.artemis.protocol.amqp.logger.ActiveMQAMQPProtocolLogger;
 import org.apache.activemq.artemis.protocol.amqp.proton.handler.EventHandler;
 import org.apache.activemq.artemis.protocol.amqp.proton.handler.ExecutorNettyAdapter;
 import org.apache.activemq.artemis.protocol.amqp.proton.handler.ExtCapability;
@@ -71,6 +75,7 @@ public class AMQPConnectionContext extends ProtonInitializable implements EventH
 
    public static final Symbol CONNECTION_OPEN_FAILED = Symbol.valueOf("amqp:connection-establishment-failed");
    public static final String AMQP_CONTAINER_ID = "amqp-container-id";
+   private static final FutureTask<Void> VOID_FUTURE = new FutureTask<>(() -> { }, null);
 
    protected final ProtonHandler handler;
 
@@ -87,9 +92,8 @@ public class AMQPConnectionContext extends ProtonInitializable implements EventH
 
    private final boolean useCoreSubscriptionNaming;
 
-   private boolean isSchedulingCancelled;
+   private final ScheduleOperator scheduleOp = new ScheduleOperator(new ScheduleRunnable());
-   private ScheduledFuture scheduledFuture;
+   private final AtomicReference<Future<?>> scheduledFutureRef = new AtomicReference(VOID_FUTURE);
-   private final Object schedulingLock = new Object();
 
    public AMQPConnectionContext(ProtonProtocolManager protocolManager,
                                 AMQPConnectionCallback connectionSP,
@@ -117,8 +121,6 @@ public class AMQPConnectionContext extends ProtonInitializable implements EventH
          this.connectionProperties.putAll(connectionProperties);
       }
 
-      this.scheduledFuture = null;
-      this.isSchedulingCancelled = false;
       this.scheduledPool = scheduledPool;
       connectionCallback.setConnection(this);
       EventLoop nettyExecutor;
@@ -191,17 +193,13 @@ public class AMQPConnectionContext extends ProtonInitializable implements EventH
    }
 
    public void close(ErrorCondition errorCondition) {
-      synchronized (schedulingLock) {
+      Future<?> scheduledFuture = scheduledFutureRef.getAndSet(null);
-         isSchedulingCancelled = true;
 
-         if (scheduledPool != null && scheduledPool instanceof ThreadPoolExecutor &&
+      if (scheduledPool instanceof ThreadPoolExecutor && scheduledFuture != null &&
-            scheduledFuture != null && scheduledFuture instanceof Runnable) {
+         scheduledFuture != VOID_FUTURE && scheduledFuture instanceof Runnable) {
-            if (!((ThreadPoolExecutor) scheduledPool).remove((Runnable) scheduledFuture) &&
+         if (!((ThreadPoolExecutor) scheduledPool).remove((Runnable) scheduledFuture) &&
-               !scheduledFuture.isCancelled() && !scheduledFuture.isDone()) {
+            !scheduledFuture.isCancelled() && !scheduledFuture.isDone()) {
-               log.warn("Scheduled task can't be removed from scheduledPool.");
+            ActiveMQAMQPProtocolLogger.LOGGER.cantRemovingScheduledTask();
-            } else {
-               scheduledFuture = null;
-            }
          }
       }
 
@@ -416,42 +414,57 @@ public class AMQPConnectionContext extends ProtonInitializable implements EventH
        * */
       if (connection.getRemoteProperties() == null || !connection.getRemoteProperties().containsKey(CONNECTION_OPEN_FAILED)) {
          long nextKeepAliveTime = handler.tick(true);
-         synchronized (schedulingLock) {
+
-            if (nextKeepAliveTime != 0 && scheduledPool != null && !isSchedulingCancelled) {
+         if (nextKeepAliveTime != 0 && scheduledPool != null) {
-               scheduledFuture = scheduledPool.schedule(new ScheduleRunnable(), (nextKeepAliveTime - TimeUnit.NANOSECONDS.toMillis(System.nanoTime())), TimeUnit.MILLISECONDS);
+            scheduleOp.setDelay(nextKeepAliveTime - TimeUnit.NANOSECONDS.toMillis(System.nanoTime()));
-            }
+
+            scheduledFutureRef.getAndUpdate(scheduleOp);
          }
       }
    }
 
-   class TickerRunnable implements Runnable {
+   class ScheduleOperator implements UnaryOperator<Future<?>> {
 
+      private long delay;
       final ScheduleRunnable scheduleRunnable;
 
-      TickerRunnable(ScheduleRunnable scheduleRunnable) {
+      ScheduleOperator(ScheduleRunnable scheduleRunnable) {
          this.scheduleRunnable = scheduleRunnable;
       }
 
+      @Override
+      public Future<?> apply(Future<?> future) {
+         return (future != null) ? scheduledPool.schedule(scheduleRunnable, delay, TimeUnit.MILLISECONDS) : null;
+      }
+
+      public void setDelay(long delay) {
+         this.delay = delay;
+      }
+   }
+
+
+   class TickerRunnable implements Runnable {
+
       @Override
       public void run() {
          Long rescheduleAt = handler.tick(false);
 
-         synchronized (schedulingLock) {
+         if (rescheduleAt == null) {
-            if (!isSchedulingCancelled) {
+            // this mean tick could not acquire a lock, we will just retry in 10 milliseconds.
-               if (rescheduleAt == null) {
+            scheduleOp.setDelay(10);
-                  // this mean tick could not acquire a lock, we will just retry in 10 milliseconds.
+
-                  scheduledFuture = scheduledPool.schedule(scheduleRunnable, 10, TimeUnit.MILLISECONDS);
+            scheduledFutureRef.getAndUpdate(scheduleOp);
-               } else if (rescheduleAt != 0) {
+         } else if (rescheduleAt != 0) {
-                  scheduledFuture = scheduledPool.schedule(scheduleRunnable, rescheduleAt - TimeUnit.NANOSECONDS.toMillis(System.nanoTime()), TimeUnit.MILLISECONDS);
+            scheduleOp.setDelay(rescheduleAt - TimeUnit.NANOSECONDS.toMillis(System.nanoTime()));
-               }
+
-            }
+            scheduledFutureRef.getAndUpdate(scheduleOp);
          }
       }
    }
 
    class ScheduleRunnable implements Runnable {
 
-      TickerRunnable tickerRunnable = new TickerRunnable(this);
+      final TickerRunnable tickerRunnable = new TickerRunnable();
 
       @Override
       public void run() {