This is an automated email from the ASF dual-hosted git repository.
tabish pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
The following commit(s) were added to refs/heads/main by this push:
new c7d0ba7e35 ARTEMIS-5383 performance issues with
HierarchicalRepositoryChangeListener
c7d0ba7e35 is described below
commit c7d0ba7e35ccf4bbdda60a5e59c7a37a752d6dd5
Author: Justin Bertram <[email protected]>
AuthorDate: Fri Apr 4 23:56:59 2025 -0500
ARTEMIS-5383 performance issues with HierarchicalRepositoryChangeListener
This commit improves performance specifically for removing listeners
from o.a.a.a.c.s.i.HierarchicalObjectRepository. It was previously using
an ArrayList and the cost to remove an Object is O(n) on average
whereas the same operation on a HashSet is O(1) on average assuming good
hash values. Furthermore, the list was protected using a
Switching to a ConcurrentHashSet required a few changes to the listeners
themselves to ensure they have good hashCodes, etc.
This commit relies on existing tests for validation.
---
.../core/postoffice/impl/PostOfficeImpl.java | 28 ++++++-----
.../artemis/core/server/impl/QueueImpl.java | 39 ++++++---------
.../impl/HierarchicalObjectRepository.java | 38 +++++----------
.../NamedHierarchicalRepositoryChangeListener.java | 56 ++++++++++++++++++++++
4 files changed, 99 insertions(+), 62 deletions(-)
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
index 533c0eca44..c442501bed 100644
---
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
@@ -95,6 +95,7 @@ import
org.apache.activemq.artemis.core.server.management.NotificationListener;
import org.apache.activemq.artemis.core.server.mirror.MirrorController;
import org.apache.activemq.artemis.core.settings.HierarchicalRepository;
import
org.apache.activemq.artemis.core.settings.HierarchicalRepositoryChangeListener;
+import
org.apache.activemq.artemis.core.settings.impl.NamedHierarchicalRepositoryChangeListener;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.core.transaction.Transaction;
import org.apache.activemq.artemis.core.transaction.TransactionOperation;
@@ -573,18 +574,21 @@ public class PostOfficeImpl implements PostOffice,
NotificationListener, Binding
}
private void registerRepositoryListenerForRetroactiveAddress(SimpleString
name) {
- HierarchicalRepositoryChangeListener repositoryChangeListener = () -> {
- String prefix = server.getInternalNamingPrefix();
- String delimiter =
server.getConfiguration().getWildcardConfiguration().getDelimiterString();
- String address =
ResourceNames.decomposeRetroactiveResourceAddressName(prefix, delimiter,
name.toString());
- AddressSettings settings =
addressSettingsRepository.getMatch(address);
- Queue internalAnycastQueue =
server.locateQueue(ResourceNames.getRetroactiveResourceQueueName(prefix,
delimiter, SimpleString.of(address), RoutingType.ANYCAST));
- if (internalAnycastQueue != null &&
internalAnycastQueue.getRingSize() != settings.getRetroactiveMessageCount()) {
-
internalAnycastQueue.setRingSize(settings.getRetroactiveMessageCount());
- }
- Queue internalMulticastQueue =
server.locateQueue(ResourceNames.getRetroactiveResourceQueueName(prefix,
delimiter, SimpleString.of(address), RoutingType.MULTICAST));
- if (internalMulticastQueue != null &&
internalMulticastQueue.getRingSize() != settings.getRetroactiveMessageCount()) {
-
internalMulticastQueue.setRingSize(settings.getRetroactiveMessageCount());
+ HierarchicalRepositoryChangeListener repositoryChangeListener = new
NamedHierarchicalRepositoryChangeListener(name) {
+ @Override
+ public void onChange() {
+ String prefix = server.getInternalNamingPrefix();
+ String delimiter =
server.getConfiguration().getWildcardConfiguration().getDelimiterString();
+ String address =
ResourceNames.decomposeRetroactiveResourceAddressName(prefix, delimiter,
name.toString());
+ AddressSettings settings =
addressSettingsRepository.getMatch(address);
+ Queue internalAnycastQueue =
server.locateQueue(ResourceNames.getRetroactiveResourceQueueName(prefix,
delimiter, SimpleString.of(address), RoutingType.ANYCAST));
+ if (internalAnycastQueue != null &&
internalAnycastQueue.getRingSize() != settings.getRetroactiveMessageCount()) {
+
internalAnycastQueue.setRingSize(settings.getRetroactiveMessageCount());
+ }
+ Queue internalMulticastQueue =
server.locateQueue(ResourceNames.getRetroactiveResourceQueueName(prefix,
delimiter, SimpleString.of(address), RoutingType.MULTICAST));
+ if (internalMulticastQueue != null &&
internalMulticastQueue.getRingSize() != settings.getRetroactiveMessageCount()) {
+
internalMulticastQueue.setRingSize(settings.getRetroactiveMessageCount());
+ }
}
};
addressSettingsRepository.registerListener(repositoryChangeListener);
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
index 71943f4468..3707c0d5cb 100644
---
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
@@ -97,6 +97,7 @@ import
org.apache.activemq.artemis.core.server.management.Notification;
import org.apache.activemq.artemis.core.settings.HierarchicalRepository;
import
org.apache.activemq.artemis.core.settings.HierarchicalRepositoryChangeListener;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import
org.apache.activemq.artemis.core.settings.impl.NamedHierarchicalRepositoryChangeListener;
import org.apache.activemq.artemis.core.settings.impl.SlowConsumerPolicy;
import
org.apache.activemq.artemis.core.settings.impl.SlowConsumerThresholdMeasurementUnit;
import org.apache.activemq.artemis.core.transaction.Transaction;
@@ -290,7 +291,9 @@ public class QueueImpl extends CriticalComponentImpl
implements Queue {
private volatile boolean supportsDirectDeliver = false;
- private AddressSettingsRepositoryListener addressSettingsRepositoryListener;
+ private HierarchicalRepository<AddressSettings> addressSettingsRepository;
+
+ private HierarchicalRepositoryChangeListener
addressSettingsRepositoryListener;
private final ReusableLatch deliveriesInTransit = new ReusableLatch(0);
@@ -415,8 +418,16 @@ public class QueueImpl extends CriticalComponentImpl
implements Queue {
scheduledDeliveryHandler = new
ScheduledDeliveryHandlerImpl(scheduledExecutor, this);
if (addressSettingsRepository != null) {
- addressSettingsRepositoryListener = new
AddressSettingsRepositoryListener(addressSettingsRepository);
-
addressSettingsRepository.registerListener(addressSettingsRepositoryListener);
+ this.addressSettingsRepository = addressSettingsRepository;
+ addressSettingsRepositoryListener = new
NamedHierarchicalRepositoryChangeListener(queueConfiguration.getName()) {
+ @Override
+ public void onChange() {
+ cachedAddressSettings =
QueueImpl.this.addressSettingsRepository.getMatch(getAddressSettingsMatch());
+ checkDeadLetterAddressAndExpiryAddress();
+ configureSlowConsumerReaper();
+ }
+ };
+
this.addressSettingsRepository.registerListener(addressSettingsRepositoryListener);
this.cachedAddressSettings =
addressSettingsRepository.getMatch(getAddressSettingsMatch());
} else {
this.cachedAddressSettings = new AddressSettings();
@@ -1082,7 +1093,7 @@ public class QueueImpl extends CriticalComponentImpl
implements Queue {
public void close() throws Exception {
getExecutor().execute(this::cancelRedistributor);
- addressSettingsRepositoryListener.close();
+
addressSettingsRepository.unRegisterListener(addressSettingsRepositoryListener);
}
@Override
@@ -4484,26 +4495,6 @@ public class QueueImpl extends CriticalComponentImpl
implements Queue {
}
}
- private class AddressSettingsRepositoryListener implements
HierarchicalRepositoryChangeListener {
-
- HierarchicalRepository<AddressSettings> addressSettingsRepository;
-
- AddressSettingsRepositoryListener(HierarchicalRepository
addressSettingsRepository) {
- this.addressSettingsRepository = addressSettingsRepository;
- }
-
- @Override
- public void onChange() {
- cachedAddressSettings =
addressSettingsRepository.getMatch(getAddressSettingsMatch());
- checkDeadLetterAddressAndExpiryAddress();
- configureSlowConsumerReaper();
- }
-
- public void close() {
- addressSettingsRepository.unRegisterListener(this);
- }
- }
-
private String getAddressSettingsMatch() {
return
((ActiveMQServerImpl)server).getRuntimeTempQueueNamespace(queueConfiguration.isTemporary())
+ queueConfiguration.getAddress().toString();
}
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/HierarchicalObjectRepository.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/HierarchicalObjectRepository.java
index 07fdaeb101..17cec4f08a 100644
---
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/HierarchicalObjectRepository.java
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/HierarchicalObjectRepository.java
@@ -36,6 +36,7 @@ import
org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
import org.apache.activemq.artemis.core.settings.HierarchicalRepository;
import
org.apache.activemq.artemis.core.settings.HierarchicalRepositoryChangeListener;
import org.apache.activemq.artemis.core.settings.Mergeable;
+import org.apache.activemq.artemis.utils.collections.ConcurrentHashSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -102,7 +103,7 @@ public class HierarchicalObjectRepository<T> implements
HierarchicalRepository<T
/**
* any registered listeners, these get fired on changes to the repository
*/
- private final List<HierarchicalRepositoryChangeListener> listeners = new
ArrayList<>();
+ private final ConcurrentHashSet<HierarchicalRepositoryChangeListener>
listeners = new ConcurrentHashSet<>();
public HierarchicalObjectRepository() {
this(null);
@@ -319,25 +320,15 @@ public class HierarchicalObjectRepository<T> implements
HierarchicalRepository<T
@Override
public void registerListener(final HierarchicalRepositoryChangeListener
listener) {
- lock.writeLock().lock();
- try {
- listeners.add(listener);
- if (listenersEnabled) {
- listener.onChange();
- }
- } finally {
- lock.writeLock().unlock();
+ listeners.add(listener);
+ if (listenersEnabled) {
+ listener.onChange();
}
}
@Override
public void unRegisterListener(final HierarchicalRepositoryChangeListener
listener) {
- lock.writeLock().lock();
- try {
- listeners.remove(listener);
- } finally {
- lock.writeLock().unlock();
- }
+ listeners.remove(listener);
}
/**
@@ -405,19 +396,14 @@ public class HierarchicalObjectRepository<T> implements
HierarchicalRepository<T
}
private void onChange() {
- lock.readLock().lock();
- try {
- if (listenersEnabled) {
- for (HierarchicalRepositoryChangeListener listener : listeners) {
- try {
- listener.onChange();
- } catch (Throwable e) {
- ActiveMQServerLogger.LOGGER.errorCallingRepoListener(e);
- }
+ if (listenersEnabled) {
+ for (HierarchicalRepositoryChangeListener listener : listeners) {
+ try {
+ listener.onChange();
+ } catch (Throwable e) {
+ ActiveMQServerLogger.LOGGER.errorCallingRepoListener(e);
}
}
- } finally {
- lock.readLock().unlock();
}
}
diff --git
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/NamedHierarchicalRepositoryChangeListener.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/NamedHierarchicalRepositoryChangeListener.java
new file mode 100644
index 0000000000..7d1b265543
--- /dev/null
+++
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/settings/impl/NamedHierarchicalRepositoryChangeListener.java
@@ -0,0 +1,56 @@
+/*
+ * 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.core.settings.impl;
+
+import java.util.Objects;
+
+import org.apache.activemq.artemis.api.core.SimpleString;
+import
org.apache.activemq.artemis.core.settings.HierarchicalRepositoryChangeListener;
+
+public abstract class NamedHierarchicalRepositoryChangeListener implements
HierarchicalRepositoryChangeListener {
+
+ private SimpleString name;
+
+ public NamedHierarchicalRepositoryChangeListener(SimpleString name) {
+ this.name = Objects.requireNonNull(name);
+ }
+
+ SimpleString getName() {
+ return this.name;
+ }
+
+ @Override
+ public int hashCode() {
+ return name.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ NamedHierarchicalRepositoryChangeListener that =
(NamedHierarchicalRepositoryChangeListener) o;
+
+ if (!Objects.equals(name, that.name)) {
+ return false;
+ } else {
+ return true;
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
For further information, visit: https://activemq.apache.org/contact
