You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2019/07/22 14:55:29 UTC
[activemq-artemis] branch master updated: ARTEMIS-2394 Improve
scheduling sync for AMQPConnectionContext
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
The following commit(s) were added to refs/heads/master by this push:
new 022b589 ARTEMIS-2394 Improve scheduling sync for AMQPConnectionContext
new a08b2d5 This closes #2727
022b589 is described below
commit 022b5895ef451468631d71c2963e5c105d275d19
Author: brusdev <br...@gmail.com>
AuthorDate: Thu Jul 18 16:19:04 2019 +0200
ARTEMIS-2394 Improve scheduling sync for AMQPConnectionContext
Remove synchronized blocks using an AtomicReference.
---
.../amqp/logger/ActiveMQAMQPProtocolLogger.java | 51 ++++++++++++++
.../amqp/proton/AMQPConnectionContext.java | 79 +++++++++++++---------
2 files changed, 97 insertions(+), 33 deletions(-)
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/logger/ActiveMQAMQPProtocolLogger.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/logger/ActiveMQAMQPProtocolLogger.java
new file mode 100644
index 0000000..77cca96
--- /dev/null
+++ b/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();
+}
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPConnectionContext.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPConnectionContext.java
index 35ea4f5..e6ec486 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPConnectionContext.java
+++ b/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 ScheduledFuture scheduledFuture;
- private final Object schedulingLock = new Object();
+ private final ScheduleOperator scheduleOp = new ScheduleOperator(new ScheduleRunnable());
+ private final AtomicReference<Future<?>> scheduledFutureRef = new AtomicReference(VOID_FUTURE);
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) {
- isSchedulingCancelled = true;
-
- if (scheduledPool != null && scheduledPool instanceof ThreadPoolExecutor &&
- scheduledFuture != null && scheduledFuture instanceof Runnable) {
- if (!((ThreadPoolExecutor) scheduledPool).remove((Runnable) scheduledFuture) &&
- !scheduledFuture.isCancelled() && !scheduledFuture.isDone()) {
- log.warn("Scheduled task can't be removed from scheduledPool.");
- } else {
- scheduledFuture = null;
- }
+ Future<?> scheduledFuture = scheduledFutureRef.getAndSet(null);
+
+ if (scheduledPool instanceof ThreadPoolExecutor && scheduledFuture != null &&
+ scheduledFuture != VOID_FUTURE && scheduledFuture instanceof Runnable) {
+ if (!((ThreadPoolExecutor) scheduledPool).remove((Runnable) scheduledFuture) &&
+ !scheduledFuture.isCancelled() && !scheduledFuture.isDone()) {
+ ActiveMQAMQPProtocolLogger.LOGGER.cantRemovingScheduledTask();
}
}
@@ -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) {
- scheduledFuture = scheduledPool.schedule(new ScheduleRunnable(), (nextKeepAliveTime - TimeUnit.NANOSECONDS.toMillis(System.nanoTime())), TimeUnit.MILLISECONDS);
- }
+
+ if (nextKeepAliveTime != 0 && scheduledPool != null) {
+ 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 (!isSchedulingCancelled) {
- if (rescheduleAt == null) {
- // this mean tick could not acquire a lock, we will just retry in 10 milliseconds.
- scheduledFuture = scheduledPool.schedule(scheduleRunnable, 10, TimeUnit.MILLISECONDS);
- } else if (rescheduleAt != 0) {
- scheduledFuture = scheduledPool.schedule(scheduleRunnable, rescheduleAt - TimeUnit.NANOSECONDS.toMillis(System.nanoTime()), TimeUnit.MILLISECONDS);
- }
- }
+ if (rescheduleAt == null) {
+ // this mean tick could not acquire a lock, we will just retry in 10 milliseconds.
+ scheduleOp.setDelay(10);
+
+ scheduledFutureRef.getAndUpdate(scheduleOp);
+ } else if (rescheduleAt != 0) {
+ 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() {