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/01/17 15:26:01 UTC
[activemq-artemis] branch master updated: ARTEMIS-2212 Avoid using
CLQ on ServerConsumerImpl
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 11214f0 ARTEMIS-2212 Avoid using CLQ on ServerConsumerImpl
new 4179c44 This closes #2480
11214f0 is described below
commit 11214f0ca9ea8b5c58c7d3a9c46249cb95ac3cfc
Author: Francesco Nigro <ni...@gmail.com>
AuthorDate: Wed Dec 12 17:47:33 2018 +0100
ARTEMIS-2212 Avoid using CLQ on ServerConsumerImpl
It would deliver a better performance for the most
common operations eg offer, poll, iterations, size.
---
.../core/server/impl/ServerConsumerImpl.java | 102 ++++++++++++---------
1 file changed, 57 insertions(+), 45 deletions(-)
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
index 19e3956..18bb987 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
@@ -18,10 +18,12 @@ package org.apache.activemq.artemis.core.server.impl;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.ArrayDeque;
+import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
-import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -117,7 +119,10 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
@Override
public String debug() {
- return toString() + "::Delivering " + this.deliveringRefs.size();
+ String debug = toString() + "::Delivering ";
+ synchronized (lock) {
+ return debug + this.deliveringRefs.size();
+ }
}
/**
@@ -132,7 +137,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
private final StorageManager storageManager;
- protected final java.util.Deque<MessageReference> deliveringRefs = new ConcurrentLinkedDeque<>();
+ private final java.util.Deque<MessageReference> deliveringRefs = new ArrayDeque<>();
private final SessionCallback callback;
@@ -334,16 +339,20 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
@Override
public List<MessageReference> getDeliveringMessages() {
- List<MessageReference> refs = new LinkedList<>();
synchronized (lock) {
+ int expectedSize = 0;
List<MessageReference> refsOnConsumer = session.getInTXMessagesForConsumer(this.id);
if (refsOnConsumer != null) {
+ expectedSize = refsOnConsumer.size();
+ }
+ expectedSize += deliveringRefs.size();
+ final List<MessageReference> refs = new ArrayList<>(expectedSize);
+ if (refsOnConsumer != null) {
refs.addAll(refsOnConsumer);
}
refs.addAll(deliveringRefs);
+ return refs;
}
-
- return refs;
}
/** i
@@ -522,21 +531,16 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
removeItself();
- LinkedList<MessageReference> refs = cancelRefs(failed, false, null);
-
- Iterator<MessageReference> iter = refs.iterator();
+ List<MessageReference> refs = cancelRefs(failed, false, null);
Transaction tx = new TransactionImpl(storageManager);
- while (iter.hasNext()) {
- MessageReference ref = iter.next();
-
+ refs.forEach(ref -> {
if (logger.isTraceEnabled()) {
logger.trace("ServerConsumerImpl::" + this + " cancelling reference " + ref);
}
-
ref.getQueue().cancel(tx, ref, true);
- }
+ });
tx.rollback();
@@ -638,9 +642,9 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
}
@Override
- public LinkedList<MessageReference> cancelRefs(final boolean failed,
- final boolean lastConsumedAsDelivered,
- final Transaction tx) throws Exception {
+ public List<MessageReference> cancelRefs(final boolean failed,
+ final boolean lastConsumedAsDelivered,
+ final Transaction tx) throws Exception {
boolean performACK = lastConsumedAsDelivered;
try {
@@ -654,30 +658,30 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
largeMessageDeliverer = null;
}
- LinkedList<MessageReference> refs = new LinkedList<>();
-
synchronized (lock) {
- if (!deliveringRefs.isEmpty()) {
- for (MessageReference ref : deliveringRefs) {
- if (performACK) {
- ref.acknowledge(tx, this);
-
- performACK = false;
- } else {
- refs.add(ref);
- updateDeliveryCountForCanceledRef(ref, failed);
- }
+ if (deliveringRefs.isEmpty()) {
+ return Collections.emptyList();
+ }
+ final List<MessageReference> refs = new ArrayList<>(deliveringRefs.size());
+ MessageReference ref;
+ while ((ref = deliveringRefs.poll()) != null) {
+ if (performACK) {
+ ref.acknowledge(tx, this);
- if (logger.isTraceEnabled()) {
- logger.trace("ServerConsumerImpl::" + this + " Preparing Cancelling list for messageID = " + ref.getMessage().getMessageID() + ", ref = " + ref);
- }
+ performACK = false;
+ } else {
+ refs.add(ref);
+ updateDeliveryCountForCanceledRef(ref, failed);
}
- deliveringRefs.clear();
+ if (logger.isTraceEnabled()) {
+ logger.trace("ServerConsumerImpl::" + this + " Preparing Cancelling list for messageID = " + ref.getMessage().getMessageID() + ", ref = " + ref);
+ }
}
+
+ return refs;
}
- return refs;
}
protected void updateDeliveryCountForCanceledRef(MessageReference ref, boolean failed) {
@@ -1017,7 +1021,9 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
@Override
public synchronized void backToDelivering(MessageReference reference) {
- deliveringRefs.addFirst(reference);
+ synchronized (lock) {
+ deliveringRefs.addFirst(reference);
+ }
}
@Override
@@ -1038,24 +1044,30 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
if (deliveringRefs.peek().getMessage().getMessageID() == messageID) {
return deliveringRefs.poll();
}
+ //slow path in a separate method
+ return removeDeliveringRefById(messageID);
+ }
+ }
- Iterator<MessageReference> iter = deliveringRefs.iterator();
+ private MessageReference removeDeliveringRefById(long messageID) {
+ assert deliveringRefs.peek().getMessage().getMessageID() != messageID;
- MessageReference ref = null;
+ Iterator<MessageReference> iter = deliveringRefs.iterator();
- while (iter.hasNext()) {
- MessageReference theRef = iter.next();
+ MessageReference ref = null;
- if (theRef.getMessage().getMessageID() == messageID) {
- iter.remove();
+ while (iter.hasNext()) {
+ MessageReference theRef = iter.next();
- ref = theRef;
+ if (theRef.getMessage().getMessageID() == messageID) {
+ iter.remove();
- break;
- }
+ ref = theRef;
+
+ break;
}
- return ref;
}
+ return ref;
}
/**