You are viewing a plain text version of this content. The canonical link for it is here.
Posted to gitbox@activemq.apache.org by "clebertsuconic (via GitHub)" <gi...@apache.org> on 2024/02/20 13:55:51 UTC
[PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
clebertsuconic opened a new pull request, #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827
As part of this PR, I am also including a little PWD command, that I used to test this manually.
I would have many terminal windows open, and not knowing which terminal was connected at which server's. So PWD helped me a lot.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499280271
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMapProvider.java:
##########
@@ -0,0 +1,86 @@
+/*
+ * 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.journal.collections;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.function.LongFunction;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import io.netty.util.collection.LongObjectHashMap;
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.IOCompletion;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.journal.RecordInfo;
+import org.apache.activemq.artemis.core.persistence.Persister;
+
+public class JournalHashMapProvider<K, V, C> {
+
+ final Journal journal;
+ final Persister<JournalHashMap.MapRecord<K, V>> persister;
+ final LongObjectHashMap<JournalHashMap<K, V, C>> journalMaps = new LongObjectHashMap<>();
+ final LongSupplier idSupplier;
+ final byte recordType;
+ final IOCriticalErrorListener ioExceptionListener;
+ final Supplier<IOCompletion> ioCompletionSupplier;
+ final LongFunction<C> contextProvider;
+
+ public JournalHashMapProvider(LongSupplier idSupplier, Journal journal, AbstractHashMapPersister persister, byte recordType, Supplier<IOCompletion> ioCompletionSupplier, LongFunction<C> contextProvider, IOCriticalErrorListener ioExceptionListener) {
+ this.idSupplier = idSupplier;
+ this.persister = persister;
+ this.journal = journal;
+ this.recordType = recordType;
+ this.ioExceptionListener = ioExceptionListener;
+ this.contextProvider = contextProvider;
+ this.ioCompletionSupplier = ioCompletionSupplier;
+ }
+
+ public List<JournalHashMap<K, V, C>> getMaps() {
+ ArrayList<JournalHashMap<K, V, C>> maps = new ArrayList<>();
+ journalMaps.values().forEach(maps::add);
+ return maps;
+ }
+
+ public void clear() {
+ journalMaps.clear();
+ }
+
+ public void reload(RecordInfo recordInfo) {
+ JournalHashMap.MapRecord<K, V> mapRecord = persister.decode(recordInfo.wrapData(), null, null);
+ getMap(mapRecord.collectionID, null).reload(mapRecord);
+ }
+
+ public Iterator<JournalHashMap<K, V, C>> iterMaps() {
+ return journalMaps.values().iterator();
+ }
+
+ public JournalHashMap<K, V, C> getMap(long collectionID, C context) {
+ JournalHashMap<K, V, C> journalHashMap = journalMaps.get(collectionID);
+ if (journalHashMap == null) {
+ journalHashMap = new JournalHashMap<>(collectionID, journal, idSupplier, persister, recordType, ioCompletionSupplier, contextProvider, ioExceptionListener).setContext(context);
+ journalMaps.put(collectionID, journalHashMap);
+ }
+ return journalHashMap;
+ }
Review Comment:
It is from a single thread, but I will add the synchronized.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499051782
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/AbstractHashMapPersister.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.journal.collections;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.core.persistence.CoreMessageObjectPools;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.utils.DataConstants;
+
+public abstract class AbstractHashMapPersister<K, V> implements Persister<JournalHashMap.MapRecord<K, V>> {
+
+ @Override
+ public byte getID() {
+ return 0;
+ }
+
+ @Override
+ public final int getEncodeSize(JournalHashMap.MapRecord<K, V> record) {
+ return DataConstants.SIZE_BYTE + // FILLER, could be used for versioning in the future
+ DataConstants.SIZE_LONG + // recordID
+ DataConstants.SIZE_LONG + // collectionID
+ getKeySize(record.key) +
+ getValueSize(record.value);
+ }
+
+ protected abstract int getKeySize(K key);
+
+ protected abstract void encodeKey(ActiveMQBuffer buffer, K key);
+
+ protected abstract K decodeKey(ActiveMQBuffer buffer);
+
+ protected abstract int getValueSize(V value);
+
+ protected abstract void encodeValue(ActiveMQBuffer buffer, V value);
+
+ protected abstract V decodeValue(ActiveMQBuffer buffer, K key);
+
+ @Override
+ public final void encode(ActiveMQBuffer buffer, JournalHashMap.MapRecord<K, V> record) {
+ buffer.writeByte((byte)0); // filler - could be used for versioning in the future.
+ buffer.writeLong(record.id);
+ buffer.writeLong(record.collectionID);
+ encodeKey(buffer, record.key);
+ encodeValue(buffer, record.value);
+ }
+
+ @Override
+ public final JournalHashMap.MapRecord<K, V> decode(ActiveMQBuffer buffer,
+ JournalHashMap.MapRecord<K, V> record,
+ CoreMessageObjectPools pool) {
+ buffer.readByte(); // filler - not used currently - just in case we ever need to version this wiring
+ long id = buffer.readLong();
+ long collectionID = buffer.readLong();
+ K key = decodeKey(buffer);
+ V value = decodeValue(buffer, key);
+
+ JournalHashMap.MapRecord<K, V> mapRecord = new JournalHashMap.MapRecord<>(collectionID, id, key, value);
Review Comment:
Is there a reason the persister favours the 'id' as the first arg, but the object representation favours the 'collectionId' as the first?
Would be nice if they were consistent unless there is a reason they shouldnt be?
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/AbstractHashMapPersister.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.journal.collections;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.core.persistence.CoreMessageObjectPools;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.utils.DataConstants;
+
+public abstract class AbstractHashMapPersister<K, V> implements Persister<JournalHashMap.MapRecord<K, V>> {
+
+ @Override
+ public byte getID() {
+ return 0;
+ }
+
+ @Override
+ public final int getEncodeSize(JournalHashMap.MapRecord<K, V> record) {
+ return DataConstants.SIZE_BYTE + // FILLER, could be used for versioning in the future
+ DataConstants.SIZE_LONG + // recordID
+ DataConstants.SIZE_LONG + // collectionID
+ getKeySize(record.key) +
+ getValueSize(record.value);
+ }
+
+ protected abstract int getKeySize(K key);
+
+ protected abstract void encodeKey(ActiveMQBuffer buffer, K key);
+
+ protected abstract K decodeKey(ActiveMQBuffer buffer);
+
+ protected abstract int getValueSize(V value);
+
+ protected abstract void encodeValue(ActiveMQBuffer buffer, V value);
+
+ protected abstract V decodeValue(ActiveMQBuffer buffer, K key);
+
+ @Override
+ public final void encode(ActiveMQBuffer buffer, JournalHashMap.MapRecord<K, V> record) {
+ buffer.writeByte((byte)0); // filler - could be used for versioning in the future.
+ buffer.writeLong(record.id);
+ buffer.writeLong(record.collectionID);
+ encodeKey(buffer, record.key);
+ encodeValue(buffer, record.value);
+ }
+
+ @Override
+ public final JournalHashMap.MapRecord<K, V> decode(ActiveMQBuffer buffer,
+ JournalHashMap.MapRecord<K, V> record,
+ CoreMessageObjectPools pool) {
+ buffer.readByte(); // filler - not used currently - just in case we ever need to version this wiring
Review Comment:
Could we just say its the version now, and make the code + comments (perhaps then deletable..) clearer for later?
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMap.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.journal.collections;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.LongFunction;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.IOCompletion;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JournalHashMap<K, V, C> implements Map<K, V> {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static class MapRecord<K, V> implements Entry<K, V> {
+ final long collectionID;
+ long id;
+ K key;
+ V value;
+
+ MapRecord(long collectionID, long id, K key, V value) {
+ this.collectionID = collectionID;
+ this.id = id;
+ this.key = key;
+ this.value = value;
+ }
+
+ @Override
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public V getValue() {
+ return value;
+ }
+
+ @Override
+ public V setValue(V value) {
+ V oldValue = this.value;
+ this.value = value;
+ return oldValue;
+ }
+
+ @Override
+ public String toString() {
+ return "MapRecord{" + "collectionID=" + collectionID + ", id=" + id + ", key=" + key + ", value=" + value + '}';
+ }
+ }
+
+ public JournalHashMap(long collectionId, Journal journal, LongSupplier idGenerator, Persister<MapRecord<K, V>> persister, byte recordType, Supplier<IOCompletion> completionSupplier, LongFunction<C> contextProvider, IOCriticalErrorListener ioExceptionListener) {
+ this.collectionId = collectionId;
+ this.journal = journal;
+ this.idGenerator = idGenerator;
+ this.persister = persister;
+ this.recordType = recordType;
+ this.exceptionListener = ioExceptionListener;
+ this.completionSupplier = completionSupplier;
+ this.contextProvider = contextProvider;
+ }
+
+ C context;
+
+ LongFunction<C> contextProvider;
+
+ private final Persister<MapRecord<K, V>> persister;
+
+ private final Journal journal;
+
+ private final long collectionId;
+
+ private final byte recordType;
+
+ private final LongSupplier idGenerator;
+
+ private final Supplier<IOCompletion> completionSupplier;
+
+ private final IOCriticalErrorListener exceptionListener;
+
+ private final Map<K, MapRecord<K, V>> map = new HashMap<>();
+
+ public long getCollectionId() {
+ return collectionId;
+ }
+
+ @Override
+ public synchronized int size() {
+ return map.size();
+ }
+
+ public C getContext() {
+ if (context == null && contextProvider != null) {
+ context = contextProvider.apply(this.collectionId);
+ }
+ return context;
+ }
+
+ public JournalHashMap<K, V, C> setContext(C context) {
+ this.context = context;
+ return this;
+ }
+
+ @Override
+ public synchronized boolean isEmpty() {
+ return map.isEmpty();
+ }
+
+ @Override
+ public synchronized boolean containsKey(Object key) {
+ return map.containsKey(key);
+ }
+
+ @Override
+ public synchronized boolean containsValue(Object value) {
+ for (Entry<K, MapRecord<K, V>> entry : map.entrySet()) {
+ if (value.equals(entry.getValue().value)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public synchronized V get(Object key) {
+ MapRecord<K, V> reccord = map.get(key);
+ if (reccord == null) {
+ return null;
+ } else {
+ return reccord.value;
+ }
+ }
+
+ /** This is to be called from a single thread during reload, no need to be synchronized */
+ public void reload(MapRecord<K, V> reloadValue) {
+ map.put(reloadValue.getKey(), reloadValue);
+ }
+
+ @Override
+ public synchronized V put(K key, V value) {
+ logger.debug("adding {} = {}", key, value);
+ long id = idGenerator.getAsLong();
+ MapRecord<K, V> record = new MapRecord(collectionId, id, key, value);
+ store(key, record);
+ MapRecord<K, V> oldRecord = map.put(key, record);
+
+ if (oldRecord != null) {
+ removed(oldRecord);
+ return oldRecord.value;
+ } else {
+ return null;
+ }
+
+ }
+
+ private synchronized void store(K key, MapRecord<K, V> record) {
+ try {
+ IOCompletion callback = null;
+ if (completionSupplier != null) {
+ callback = completionSupplier.get();
+ }
+
+ if (callback == null) {
+ journal.appendAddRecord(record.id, recordType, persister, record, false);
+ } else {
+ journal.appendAddRecord(record.id, recordType, persister, record, true, callback);
+ }
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord) {
Review Comment:
Typo, record. Also, add type info.
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMap.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.journal.collections;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.LongFunction;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.IOCompletion;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JournalHashMap<K, V, C> implements Map<K, V> {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static class MapRecord<K, V> implements Entry<K, V> {
+ final long collectionID;
+ long id;
+ K key;
+ V value;
+
+ MapRecord(long collectionID, long id, K key, V value) {
+ this.collectionID = collectionID;
+ this.id = id;
+ this.key = key;
+ this.value = value;
+ }
+
+ @Override
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public V getValue() {
+ return value;
+ }
+
+ @Override
+ public V setValue(V value) {
+ V oldValue = this.value;
+ this.value = value;
+ return oldValue;
+ }
+
+ @Override
+ public String toString() {
+ return "MapRecord{" + "collectionID=" + collectionID + ", id=" + id + ", key=" + key + ", value=" + value + '}';
+ }
+ }
+
+ public JournalHashMap(long collectionId, Journal journal, LongSupplier idGenerator, Persister<MapRecord<K, V>> persister, byte recordType, Supplier<IOCompletion> completionSupplier, LongFunction<C> contextProvider, IOCriticalErrorListener ioExceptionListener) {
+ this.collectionId = collectionId;
+ this.journal = journal;
+ this.idGenerator = idGenerator;
+ this.persister = persister;
+ this.recordType = recordType;
+ this.exceptionListener = ioExceptionListener;
+ this.completionSupplier = completionSupplier;
+ this.contextProvider = contextProvider;
+ }
+
+ C context;
+
+ LongFunction<C> contextProvider;
+
+ private final Persister<MapRecord<K, V>> persister;
+
+ private final Journal journal;
+
+ private final long collectionId;
+
+ private final byte recordType;
+
+ private final LongSupplier idGenerator;
+
+ private final Supplier<IOCompletion> completionSupplier;
+
+ private final IOCriticalErrorListener exceptionListener;
+
+ private final Map<K, MapRecord<K, V>> map = new HashMap<>();
+
+ public long getCollectionId() {
+ return collectionId;
+ }
+
+ @Override
+ public synchronized int size() {
+ return map.size();
+ }
+
+ public C getContext() {
+ if (context == null && contextProvider != null) {
+ context = contextProvider.apply(this.collectionId);
+ }
+ return context;
+ }
+
+ public JournalHashMap<K, V, C> setContext(C context) {
+ this.context = context;
+ return this;
+ }
+
+ @Override
+ public synchronized boolean isEmpty() {
+ return map.isEmpty();
+ }
+
+ @Override
+ public synchronized boolean containsKey(Object key) {
+ return map.containsKey(key);
+ }
+
+ @Override
+ public synchronized boolean containsValue(Object value) {
+ for (Entry<K, MapRecord<K, V>> entry : map.entrySet()) {
+ if (value.equals(entry.getValue().value)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public synchronized V get(Object key) {
+ MapRecord<K, V> reccord = map.get(key);
+ if (reccord == null) {
+ return null;
+ } else {
+ return reccord.value;
+ }
+ }
+
+ /** This is to be called from a single thread during reload, no need to be synchronized */
+ public void reload(MapRecord<K, V> reloadValue) {
+ map.put(reloadValue.getKey(), reloadValue);
+ }
+
+ @Override
+ public synchronized V put(K key, V value) {
+ logger.debug("adding {} = {}", key, value);
+ long id = idGenerator.getAsLong();
+ MapRecord<K, V> record = new MapRecord(collectionId, id, key, value);
+ store(key, record);
+ MapRecord<K, V> oldRecord = map.put(key, record);
+
+ if (oldRecord != null) {
+ removed(oldRecord);
+ return oldRecord.value;
+ } else {
+ return null;
+ }
+
+ }
+
+ private synchronized void store(K key, MapRecord<K, V> record) {
+ try {
+ IOCompletion callback = null;
+ if (completionSupplier != null) {
+ callback = completionSupplier.get();
+ }
+
+ if (callback == null) {
+ journal.appendAddRecord(record.id, recordType, persister, record, false);
+ } else {
+ journal.appendAddRecord(record.id, recordType, persister, record, true, callback);
+ }
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord) {
+ try {
+ journal.appendDeleteRecord(reccord.id, false);
+ } catch (Exception e) {
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord, long txid) {
+ try {
+ journal.appendDeleteRecordTransactional(txid, reccord.id);
+ } catch (Exception e) {
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ @Override
+ public synchronized V remove(Object key) {
+ MapRecord<K, V> record = map.remove(key);
+ this.removed(record);
+ return record.value;
+ }
+
+ /** This method will remove the element from the HashMap immediately however the record is still part of a transaction.
+ * This is not playing with rollbacks. So a rollback on the transaction wouldn't place the elements back.
+ * This is intended to make sure the operation would be atomic in case of a failure, while an appendRollback is not expected. */
+ public synchronized V remove(Object key, long transactionID) {
+ MapRecord<K, V> record = map.remove(key);
+ this.removed(record, transactionID);
+ return record.value;
+ }
+
+ @Override
+ public synchronized void putAll(Map<? extends K, ? extends V> m) {
+ m.forEach(this::put);
+ }
+
+ @Override
+ public synchronized void clear() {
+ map.values().forEach(v -> remove(v));
+ map.clear();
+ }
+
+ @Override
+ public synchronized Set<K> keySet() {
+ HashSet<K> keys = new HashSet(map.size());
+ map.values().forEach(v -> keys.add(v.key));
+ return keys;
+ }
Review Comment:
Pretty sure this violates the Map contract for keySet() that the Set is backed by the map contents.
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AmqpSupport.java:
##########
@@ -43,6 +43,8 @@ public class AmqpSupport {
public static final int AMQP_CREDITS_DEFAULT = 1000;
public static final int AMQP_LOW_CREDITS_DEFAULT = 300;
+ public static final int AMQP_MIRROR_ACK_RETRY_INTERVAL = 10_000;
+
Review Comment:
Is this used any more?
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AckManager.java:
##########
@@ -0,0 +1,513 @@
+/*
+ * 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.connect.mirror;
+
+import java.lang.invoke.MethodHandles;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.LongSupplier;
+
+import io.netty.util.collection.LongObjectHashMap;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.QueueConfiguration;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.journal.RecordInfo;
+import org.apache.activemq.artemis.core.journal.collections.JournalHashMap;
+import org.apache.activemq.artemis.core.journal.collections.JournalHashMapProvider;
+import org.apache.activemq.artemis.core.paging.PagingStore;
+import org.apache.activemq.artemis.core.paging.cursor.PageSubscription;
+import org.apache.activemq.artemis.core.paging.cursor.PagedReference;
+import org.apache.activemq.artemis.core.paging.impl.Page;
+import org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds;
+import org.apache.activemq.artemis.core.persistence.impl.journal.OperationContextImpl;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.AckRetry;
+import org.apache.activemq.artemis.core.server.ActiveMQComponent;
+import org.apache.activemq.artemis.core.server.ActiveMQScheduledComponent;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.MessageReference;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.core.server.RoutingContext;
+import org.apache.activemq.artemis.core.server.impl.AckReason;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.core.server.mirror.MirrorController;
+import org.apache.activemq.artemis.core.transaction.Transaction;
+import org.apache.activemq.artemis.core.transaction.impl.TransactionImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class AckManager implements ActiveMQComponent {
+
+ // we first retry on the queue a few tiems
+ public static final short MAX_QUEUE_ATTEMPT = Short.parseShort(System.getProperty(AckRetry.class.getName() + ".MAX_QUEUE_ATTEMPT", "5"));
Review Comment:
tiems -> times
I'd probably change/remove 'a few times' given its actually 5
Do we retry on the queue after trying paging?
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMap.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.journal.collections;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.LongFunction;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.IOCompletion;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JournalHashMap<K, V, C> implements Map<K, V> {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static class MapRecord<K, V> implements Entry<K, V> {
+ final long collectionID;
+ long id;
+ K key;
+ V value;
Review Comment:
Why is only the first one final, when apparently only the last one can be changed?
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMap.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.journal.collections;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.LongFunction;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.IOCompletion;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JournalHashMap<K, V, C> implements Map<K, V> {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static class MapRecord<K, V> implements Entry<K, V> {
+ final long collectionID;
+ long id;
+ K key;
+ V value;
+
+ MapRecord(long collectionID, long id, K key, V value) {
+ this.collectionID = collectionID;
+ this.id = id;
+ this.key = key;
+ this.value = value;
+ }
+
+ @Override
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public V getValue() {
+ return value;
+ }
+
+ @Override
+ public V setValue(V value) {
+ V oldValue = this.value;
+ this.value = value;
+ return oldValue;
+ }
+
+ @Override
+ public String toString() {
+ return "MapRecord{" + "collectionID=" + collectionID + ", id=" + id + ", key=" + key + ", value=" + value + '}';
+ }
+ }
+
+ public JournalHashMap(long collectionId, Journal journal, LongSupplier idGenerator, Persister<MapRecord<K, V>> persister, byte recordType, Supplier<IOCompletion> completionSupplier, LongFunction<C> contextProvider, IOCriticalErrorListener ioExceptionListener) {
+ this.collectionId = collectionId;
+ this.journal = journal;
+ this.idGenerator = idGenerator;
+ this.persister = persister;
+ this.recordType = recordType;
+ this.exceptionListener = ioExceptionListener;
+ this.completionSupplier = completionSupplier;
+ this.contextProvider = contextProvider;
+ }
+
+ C context;
+
+ LongFunction<C> contextProvider;
+
+ private final Persister<MapRecord<K, V>> persister;
+
+ private final Journal journal;
+
+ private final long collectionId;
+
+ private final byte recordType;
+
+ private final LongSupplier idGenerator;
+
+ private final Supplier<IOCompletion> completionSupplier;
+
+ private final IOCriticalErrorListener exceptionListener;
+
+ private final Map<K, MapRecord<K, V>> map = new HashMap<>();
+
+ public long getCollectionId() {
+ return collectionId;
+ }
+
+ @Override
+ public synchronized int size() {
+ return map.size();
+ }
+
+ public C getContext() {
+ if (context == null && contextProvider != null) {
+ context = contextProvider.apply(this.collectionId);
+ }
+ return context;
+ }
+
+ public JournalHashMap<K, V, C> setContext(C context) {
+ this.context = context;
+ return this;
+ }
+
+ @Override
+ public synchronized boolean isEmpty() {
+ return map.isEmpty();
+ }
+
+ @Override
+ public synchronized boolean containsKey(Object key) {
+ return map.containsKey(key);
+ }
+
+ @Override
+ public synchronized boolean containsValue(Object value) {
+ for (Entry<K, MapRecord<K, V>> entry : map.entrySet()) {
+ if (value.equals(entry.getValue().value)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public synchronized V get(Object key) {
+ MapRecord<K, V> reccord = map.get(key);
+ if (reccord == null) {
+ return null;
+ } else {
+ return reccord.value;
+ }
+ }
+
+ /** This is to be called from a single thread during reload, no need to be synchronized */
+ public void reload(MapRecord<K, V> reloadValue) {
+ map.put(reloadValue.getKey(), reloadValue);
+ }
+
+ @Override
+ public synchronized V put(K key, V value) {
+ logger.debug("adding {} = {}", key, value);
+ long id = idGenerator.getAsLong();
+ MapRecord<K, V> record = new MapRecord(collectionId, id, key, value);
+ store(key, record);
+ MapRecord<K, V> oldRecord = map.put(key, record);
+
+ if (oldRecord != null) {
+ removed(oldRecord);
+ return oldRecord.value;
+ } else {
+ return null;
+ }
+
+ }
+
+ private synchronized void store(K key, MapRecord<K, V> record) {
+ try {
+ IOCompletion callback = null;
+ if (completionSupplier != null) {
+ callback = completionSupplier.get();
+ }
+
+ if (callback == null) {
+ journal.appendAddRecord(record.id, recordType, persister, record, false);
+ } else {
+ journal.appendAddRecord(record.id, recordType, persister, record, true, callback);
+ }
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord) {
+ try {
+ journal.appendDeleteRecord(reccord.id, false);
+ } catch (Exception e) {
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord, long txid) {
Review Comment:
Same
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMap.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.journal.collections;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.LongFunction;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.IOCompletion;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JournalHashMap<K, V, C> implements Map<K, V> {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static class MapRecord<K, V> implements Entry<K, V> {
+ final long collectionID;
+ long id;
+ K key;
+ V value;
+
+ MapRecord(long collectionID, long id, K key, V value) {
+ this.collectionID = collectionID;
+ this.id = id;
+ this.key = key;
+ this.value = value;
+ }
+
+ @Override
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public V getValue() {
+ return value;
+ }
+
+ @Override
+ public V setValue(V value) {
+ V oldValue = this.value;
+ this.value = value;
+ return oldValue;
+ }
+
+ @Override
+ public String toString() {
+ return "MapRecord{" + "collectionID=" + collectionID + ", id=" + id + ", key=" + key + ", value=" + value + '}';
+ }
+ }
+
+ public JournalHashMap(long collectionId, Journal journal, LongSupplier idGenerator, Persister<MapRecord<K, V>> persister, byte recordType, Supplier<IOCompletion> completionSupplier, LongFunction<C> contextProvider, IOCriticalErrorListener ioExceptionListener) {
+ this.collectionId = collectionId;
+ this.journal = journal;
+ this.idGenerator = idGenerator;
+ this.persister = persister;
+ this.recordType = recordType;
+ this.exceptionListener = ioExceptionListener;
+ this.completionSupplier = completionSupplier;
+ this.contextProvider = contextProvider;
+ }
+
+ C context;
+
+ LongFunction<C> contextProvider;
+
+ private final Persister<MapRecord<K, V>> persister;
+
+ private final Journal journal;
+
+ private final long collectionId;
+
+ private final byte recordType;
+
+ private final LongSupplier idGenerator;
+
+ private final Supplier<IOCompletion> completionSupplier;
+
+ private final IOCriticalErrorListener exceptionListener;
+
+ private final Map<K, MapRecord<K, V>> map = new HashMap<>();
+
+ public long getCollectionId() {
+ return collectionId;
+ }
+
+ @Override
+ public synchronized int size() {
+ return map.size();
+ }
+
+ public C getContext() {
+ if (context == null && contextProvider != null) {
+ context = contextProvider.apply(this.collectionId);
+ }
+ return context;
+ }
+
+ public JournalHashMap<K, V, C> setContext(C context) {
+ this.context = context;
+ return this;
+ }
+
+ @Override
+ public synchronized boolean isEmpty() {
+ return map.isEmpty();
+ }
+
+ @Override
+ public synchronized boolean containsKey(Object key) {
+ return map.containsKey(key);
+ }
+
+ @Override
+ public synchronized boolean containsValue(Object value) {
+ for (Entry<K, MapRecord<K, V>> entry : map.entrySet()) {
+ if (value.equals(entry.getValue().value)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public synchronized V get(Object key) {
+ MapRecord<K, V> reccord = map.get(key);
+ if (reccord == null) {
+ return null;
+ } else {
+ return reccord.value;
+ }
+ }
+
+ /** This is to be called from a single thread during reload, no need to be synchronized */
+ public void reload(MapRecord<K, V> reloadValue) {
+ map.put(reloadValue.getKey(), reloadValue);
+ }
+
+ @Override
+ public synchronized V put(K key, V value) {
+ logger.debug("adding {} = {}", key, value);
+ long id = idGenerator.getAsLong();
+ MapRecord<K, V> record = new MapRecord(collectionId, id, key, value);
+ store(key, record);
+ MapRecord<K, V> oldRecord = map.put(key, record);
+
+ if (oldRecord != null) {
+ removed(oldRecord);
+ return oldRecord.value;
+ } else {
+ return null;
+ }
+
+ }
+
+ private synchronized void store(K key, MapRecord<K, V> record) {
+ try {
+ IOCompletion callback = null;
+ if (completionSupplier != null) {
+ callback = completionSupplier.get();
+ }
+
+ if (callback == null) {
+ journal.appendAddRecord(record.id, recordType, persister, record, false);
+ } else {
+ journal.appendAddRecord(record.id, recordType, persister, record, true, callback);
+ }
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord) {
+ try {
+ journal.appendDeleteRecord(reccord.id, false);
+ } catch (Exception e) {
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord, long txid) {
+ try {
+ journal.appendDeleteRecordTransactional(txid, reccord.id);
+ } catch (Exception e) {
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ @Override
+ public synchronized V remove(Object key) {
+ MapRecord<K, V> record = map.remove(key);
+ this.removed(record);
+ return record.value;
+ }
+
+ /** This method will remove the element from the HashMap immediately however the record is still part of a transaction.
+ * This is not playing with rollbacks. So a rollback on the transaction wouldn't place the elements back.
+ * This is intended to make sure the operation would be atomic in case of a failure, while an appendRollback is not expected. */
+ public synchronized V remove(Object key, long transactionID) {
+ MapRecord<K, V> record = map.remove(key);
+ this.removed(record, transactionID);
+ return record.value;
+ }
+
+ @Override
+ public synchronized void putAll(Map<? extends K, ? extends V> m) {
+ m.forEach(this::put);
+ }
+
+ @Override
+ public synchronized void clear() {
+ map.values().forEach(v -> remove(v));
+ map.clear();
+ }
+
+ @Override
+ public synchronized Set<K> keySet() {
+ HashSet<K> keys = new HashSet(map.size());
+ map.values().forEach(v -> keys.add(v.key));
+ return keys;
+ }
+
+ @Override
+ public synchronized Collection<V> values() {
+ ArrayList values = new ArrayList(map.size());
+ map.values().forEach(v -> values.add(v.value));
+ return values;
+ }
Review Comment:
Similarly violates the Map contract
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMap.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.journal.collections;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.LongFunction;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.IOCompletion;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JournalHashMap<K, V, C> implements Map<K, V> {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static class MapRecord<K, V> implements Entry<K, V> {
+ final long collectionID;
+ long id;
+ K key;
+ V value;
+
+ MapRecord(long collectionID, long id, K key, V value) {
+ this.collectionID = collectionID;
+ this.id = id;
+ this.key = key;
+ this.value = value;
+ }
+
+ @Override
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public V getValue() {
+ return value;
+ }
+
+ @Override
+ public V setValue(V value) {
+ V oldValue = this.value;
+ this.value = value;
+ return oldValue;
+ }
+
+ @Override
+ public String toString() {
+ return "MapRecord{" + "collectionID=" + collectionID + ", id=" + id + ", key=" + key + ", value=" + value + '}';
+ }
+ }
+
+ public JournalHashMap(long collectionId, Journal journal, LongSupplier idGenerator, Persister<MapRecord<K, V>> persister, byte recordType, Supplier<IOCompletion> completionSupplier, LongFunction<C> contextProvider, IOCriticalErrorListener ioExceptionListener) {
+ this.collectionId = collectionId;
+ this.journal = journal;
+ this.idGenerator = idGenerator;
+ this.persister = persister;
+ this.recordType = recordType;
+ this.exceptionListener = ioExceptionListener;
+ this.completionSupplier = completionSupplier;
+ this.contextProvider = contextProvider;
+ }
+
+ C context;
+
+ LongFunction<C> contextProvider;
+
+ private final Persister<MapRecord<K, V>> persister;
+
+ private final Journal journal;
+
+ private final long collectionId;
+
+ private final byte recordType;
+
+ private final LongSupplier idGenerator;
+
+ private final Supplier<IOCompletion> completionSupplier;
+
+ private final IOCriticalErrorListener exceptionListener;
+
+ private final Map<K, MapRecord<K, V>> map = new HashMap<>();
+
+ public long getCollectionId() {
+ return collectionId;
+ }
+
+ @Override
+ public synchronized int size() {
+ return map.size();
+ }
+
+ public C getContext() {
+ if (context == null && contextProvider != null) {
+ context = contextProvider.apply(this.collectionId);
+ }
+ return context;
+ }
+
+ public JournalHashMap<K, V, C> setContext(C context) {
+ this.context = context;
+ return this;
+ }
+
+ @Override
+ public synchronized boolean isEmpty() {
+ return map.isEmpty();
+ }
+
+ @Override
+ public synchronized boolean containsKey(Object key) {
+ return map.containsKey(key);
+ }
+
+ @Override
+ public synchronized boolean containsValue(Object value) {
+ for (Entry<K, MapRecord<K, V>> entry : map.entrySet()) {
+ if (value.equals(entry.getValue().value)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public synchronized V get(Object key) {
+ MapRecord<K, V> reccord = map.get(key);
+ if (reccord == null) {
+ return null;
+ } else {
+ return reccord.value;
+ }
+ }
+
+ /** This is to be called from a single thread during reload, no need to be synchronized */
+ public void reload(MapRecord<K, V> reloadValue) {
+ map.put(reloadValue.getKey(), reloadValue);
+ }
+
+ @Override
+ public synchronized V put(K key, V value) {
+ logger.debug("adding {} = {}", key, value);
+ long id = idGenerator.getAsLong();
+ MapRecord<K, V> record = new MapRecord(collectionId, id, key, value);
+ store(key, record);
+ MapRecord<K, V> oldRecord = map.put(key, record);
+
+ if (oldRecord != null) {
+ removed(oldRecord);
+ return oldRecord.value;
+ } else {
+ return null;
+ }
+
+ }
+
+ private synchronized void store(K key, MapRecord<K, V> record) {
+ try {
+ IOCompletion callback = null;
+ if (completionSupplier != null) {
+ callback = completionSupplier.get();
+ }
+
+ if (callback == null) {
+ journal.appendAddRecord(record.id, recordType, persister, record, false);
+ } else {
+ journal.appendAddRecord(record.id, recordType, persister, record, true, callback);
+ }
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord) {
+ try {
+ journal.appendDeleteRecord(reccord.id, false);
+ } catch (Exception e) {
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord, long txid) {
+ try {
+ journal.appendDeleteRecordTransactional(txid, reccord.id);
+ } catch (Exception e) {
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ @Override
+ public synchronized V remove(Object key) {
+ MapRecord<K, V> record = map.remove(key);
+ this.removed(record);
+ return record.value;
+ }
+
+ /** This method will remove the element from the HashMap immediately however the record is still part of a transaction.
+ * This is not playing with rollbacks. So a rollback on the transaction wouldn't place the elements back.
+ * This is intended to make sure the operation would be atomic in case of a failure, while an appendRollback is not expected. */
+ public synchronized V remove(Object key, long transactionID) {
+ MapRecord<K, V> record = map.remove(key);
+ this.removed(record, transactionID);
+ return record.value;
+ }
+
+ @Override
+ public synchronized void putAll(Map<? extends K, ? extends V> m) {
+ m.forEach(this::put);
+ }
+
+ @Override
+ public synchronized void clear() {
+ map.values().forEach(v -> remove(v));
+ map.clear();
+ }
+
+ @Override
+ public synchronized Set<K> keySet() {
+ HashSet<K> keys = new HashSet(map.size());
+ map.values().forEach(v -> keys.add(v.key));
+ return keys;
+ }
+
+ @Override
+ public synchronized Collection<V> values() {
+ ArrayList values = new ArrayList(map.size());
+ map.values().forEach(v -> values.add(v.value));
+ return values;
+ }
+
+ @Override
+ public synchronized Set<Entry<K, V>> entrySet() {
+ HashSet<Entry<K, V>> values = new HashSet<>();
+ map.values().forEach(values::add);
+ return values;
+ }
Review Comment:
Ditto
##########
artemis-commons/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQScheduledComponent.java:
##########
@@ -288,6 +298,15 @@ public synchronized void stop() {
}
+ @Override
+ public void run() {
+ if (scheduledTask == null) {
+ logger.warn("Task not defined on {}", this.getClass());
+ } else {
+ scheduledTask.run();
+ }
+ }
+
Review Comment:
I think it would probably be better to leave this class abstract and implement run in the new Component, as every other use of the class did and still does. This class already has too many different ways to use it from what I recall of previous looks. This basically just adds another.
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -424,6 +453,10 @@ public void preAcknowledge(final Transaction tx, final MessageReference ref, fin
MirrorController controllerInUse = getControllerInUse();
+ if (controllerInUse != null && !controllerInUse.isAllowACK()) {
Review Comment:
This comment relates to code just above this change, not the change it is against.
The trace log at the start of the method looks wrong, references postACKInternalMessage when thats a seperate method further up, this is actually preAcknowledge
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMap.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.journal.collections;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.LongFunction;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.IOCompletion;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JournalHashMap<K, V, C> implements Map<K, V> {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static class MapRecord<K, V> implements Entry<K, V> {
+ final long collectionID;
+ long id;
+ K key;
+ V value;
+
+ MapRecord(long collectionID, long id, K key, V value) {
+ this.collectionID = collectionID;
+ this.id = id;
+ this.key = key;
+ this.value = value;
+ }
+
+ @Override
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public V getValue() {
+ return value;
+ }
+
+ @Override
+ public V setValue(V value) {
+ V oldValue = this.value;
+ this.value = value;
+ return oldValue;
+ }
+
+ @Override
+ public String toString() {
+ return "MapRecord{" + "collectionID=" + collectionID + ", id=" + id + ", key=" + key + ", value=" + value + '}';
+ }
+ }
+
+ public JournalHashMap(long collectionId, Journal journal, LongSupplier idGenerator, Persister<MapRecord<K, V>> persister, byte recordType, Supplier<IOCompletion> completionSupplier, LongFunction<C> contextProvider, IOCriticalErrorListener ioExceptionListener) {
+ this.collectionId = collectionId;
+ this.journal = journal;
+ this.idGenerator = idGenerator;
+ this.persister = persister;
+ this.recordType = recordType;
+ this.exceptionListener = ioExceptionListener;
+ this.completionSupplier = completionSupplier;
+ this.contextProvider = contextProvider;
+ }
+
+ C context;
+
+ LongFunction<C> contextProvider;
+
+ private final Persister<MapRecord<K, V>> persister;
+
+ private final Journal journal;
+
+ private final long collectionId;
+
+ private final byte recordType;
+
+ private final LongSupplier idGenerator;
+
+ private final Supplier<IOCompletion> completionSupplier;
+
+ private final IOCriticalErrorListener exceptionListener;
+
+ private final Map<K, MapRecord<K, V>> map = new HashMap<>();
+
+ public long getCollectionId() {
+ return collectionId;
+ }
+
+ @Override
+ public synchronized int size() {
+ return map.size();
+ }
+
+ public C getContext() {
+ if (context == null && contextProvider != null) {
+ context = contextProvider.apply(this.collectionId);
+ }
+ return context;
+ }
+
+ public JournalHashMap<K, V, C> setContext(C context) {
+ this.context = context;
+ return this;
+ }
+
+ @Override
+ public synchronized boolean isEmpty() {
+ return map.isEmpty();
+ }
+
+ @Override
+ public synchronized boolean containsKey(Object key) {
+ return map.containsKey(key);
+ }
+
+ @Override
+ public synchronized boolean containsValue(Object value) {
+ for (Entry<K, MapRecord<K, V>> entry : map.entrySet()) {
+ if (value.equals(entry.getValue().value)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public synchronized V get(Object key) {
+ MapRecord<K, V> reccord = map.get(key);
+ if (reccord == null) {
+ return null;
+ } else {
+ return reccord.value;
+ }
+ }
+
+ /** This is to be called from a single thread during reload, no need to be synchronized */
+ public void reload(MapRecord<K, V> reloadValue) {
+ map.put(reloadValue.getKey(), reloadValue);
+ }
+
+ @Override
+ public synchronized V put(K key, V value) {
+ logger.debug("adding {} = {}", key, value);
+ long id = idGenerator.getAsLong();
+ MapRecord<K, V> record = new MapRecord(collectionId, id, key, value);
Review Comment:
Add diamond <> (or actual full type<K,V>) to suppress warnings. Here and lots of other places, it looks like.
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMap.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.journal.collections;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.LongFunction;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.IOCompletion;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JournalHashMap<K, V, C> implements Map<K, V> {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static class MapRecord<K, V> implements Entry<K, V> {
Review Comment:
This Entry should be implementing equals and hashCode, especially given it is actually being inserted into a HashMap
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMapProvider.java:
##########
@@ -0,0 +1,86 @@
+/*
+ * 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.journal.collections;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.function.LongFunction;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import io.netty.util.collection.LongObjectHashMap;
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.IOCompletion;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.journal.RecordInfo;
+import org.apache.activemq.artemis.core.persistence.Persister;
+
+public class JournalHashMapProvider<K, V, C> {
+
+ final Journal journal;
+ final Persister<JournalHashMap.MapRecord<K, V>> persister;
+ final LongObjectHashMap<JournalHashMap<K, V, C>> journalMaps = new LongObjectHashMap<>();
+ final LongSupplier idSupplier;
+ final byte recordType;
+ final IOCriticalErrorListener ioExceptionListener;
+ final Supplier<IOCompletion> ioCompletionSupplier;
+ final LongFunction<C> contextProvider;
+
+ public JournalHashMapProvider(LongSupplier idSupplier, Journal journal, AbstractHashMapPersister persister, byte recordType, Supplier<IOCompletion> ioCompletionSupplier, LongFunction<C> contextProvider, IOCriticalErrorListener ioExceptionListener) {
+ this.idSupplier = idSupplier;
+ this.persister = persister;
+ this.journal = journal;
+ this.recordType = recordType;
+ this.ioExceptionListener = ioExceptionListener;
+ this.contextProvider = contextProvider;
+ this.ioCompletionSupplier = ioCompletionSupplier;
+ }
+
+ public List<JournalHashMap<K, V, C>> getMaps() {
+ ArrayList<JournalHashMap<K, V, C>> maps = new ArrayList<>();
+ journalMaps.values().forEach(maps::add);
+ return maps;
+ }
+
+ public void clear() {
+ journalMaps.clear();
+ }
+
+ public void reload(RecordInfo recordInfo) {
+ JournalHashMap.MapRecord<K, V> mapRecord = persister.decode(recordInfo.wrapData(), null, null);
+ getMap(mapRecord.collectionID, null).reload(mapRecord);
+ }
+
+ public Iterator<JournalHashMap<K, V, C>> iterMaps() {
+ return journalMaps.values().iterator();
+ }
+
+ public JournalHashMap<K, V, C> getMap(long collectionID, C context) {
+ JournalHashMap<K, V, C> journalHashMap = journalMaps.get(collectionID);
+ if (journalHashMap == null) {
+ journalHashMap = new JournalHashMap<>(collectionID, journal, idSupplier, persister, recordType, ioCompletionSupplier, contextProvider, ioExceptionListener).setContext(context);
+ journalMaps.put(collectionID, journalHashMap);
+ }
+ return journalHashMap;
+ }
Review Comment:
Presumably this is only being used from a single thread?
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -617,4 +650,55 @@ public static void routeMirrorCommand(ActiveMQServer server, Message message, Tr
server.getPostOffice().route(message, ctx, false);
}
+ class PagedRouteContext implements RouteContextList {
+
+ private final List<Queue> listRepresentation;
+ private final List<Queue> emptyList = Collections.emptyList();
+
+ PagedRouteContext() {
+ listRepresentation = new ArrayList<>(1);
+ listRepresentation.add(snfQueue);
Review Comment:
If the snfQueue durability is essentially fixed, cant we pre-determine the answers/return values and simplify much of the method code?
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -281,6 +300,15 @@ public void sendMessage(Transaction tx, Message message, RoutingContext context)
return;
}
+ if (snfQueue.getPagingStore().page(message, tx, pagedRouteContext, this::copyMessageForPaging)) {
+ return;
+ }
+
+ if (message.isPaged()) {
+ // if the source was paged, we copy the message
+ message = copyMessageForPaging(message);
+ }
Review Comment:
Why?
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -281,6 +300,15 @@ public void sendMessage(Transaction tx, Message message, RoutingContext context)
return;
}
+ if (snfQueue.getPagingStore().page(message, tx, pagedRouteContext, this::copyMessageForPaging)) {
+ return;
+ }
Review Comment:
What is this doing? What did it before?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499234573
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -281,6 +300,15 @@ public void sendMessage(Transaction tx, Message message, RoutingContext context)
return;
}
+ if (snfQueue.getPagingStore().page(message, tx, pagedRouteContext, this::copyMessageForPaging)) {
+ return;
+ }
Review Comment:
Before the message would just pile up in the mememory, ignore paging and be ot of order.
if not paging ,this is NoOp. If it paged, I must not route.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499296963
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -424,6 +453,10 @@ public void preAcknowledge(final Transaction tx, final MessageReference ref, fin
MirrorController controllerInUse = getControllerInUse();
+ if (controllerInUse != null && !controllerInUse.isAllowACK()) {
Review Comment:
that comment stays where it is.. I'm adding an extra comment here.
This is to avoid re-acking the retry.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499410286
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMap.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.journal.collections;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.LongFunction;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.IOCompletion;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JournalHashMap<K, V, C> implements Map<K, V> {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static class MapRecord<K, V> implements Entry<K, V> {
+ final long collectionID;
+ long id;
+ K key;
+ V value;
+
+ MapRecord(long collectionID, long id, K key, V value) {
+ this.collectionID = collectionID;
+ this.id = id;
+ this.key = key;
+ this.value = value;
+ }
+
+ @Override
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public V getValue() {
+ return value;
+ }
+
+ @Override
+ public V setValue(V value) {
+ V oldValue = this.value;
+ this.value = value;
+ return oldValue;
+ }
+
+ @Override
+ public String toString() {
+ return "MapRecord{" + "collectionID=" + collectionID + ", id=" + id + ", key=" + key + ", value=" + value + '}';
+ }
+ }
+
+ public JournalHashMap(long collectionId, Journal journal, LongSupplier idGenerator, Persister<MapRecord<K, V>> persister, byte recordType, Supplier<IOCompletion> completionSupplier, LongFunction<C> contextProvider, IOCriticalErrorListener ioExceptionListener) {
+ this.collectionId = collectionId;
+ this.journal = journal;
+ this.idGenerator = idGenerator;
+ this.persister = persister;
+ this.recordType = recordType;
+ this.exceptionListener = ioExceptionListener;
+ this.completionSupplier = completionSupplier;
+ this.contextProvider = contextProvider;
+ }
+
+ C context;
+
+ LongFunction<C> contextProvider;
+
+ private final Persister<MapRecord<K, V>> persister;
+
+ private final Journal journal;
+
+ private final long collectionId;
+
+ private final byte recordType;
+
+ private final LongSupplier idGenerator;
+
+ private final Supplier<IOCompletion> completionSupplier;
+
+ private final IOCriticalErrorListener exceptionListener;
+
+ private final Map<K, MapRecord<K, V>> map = new HashMap<>();
+
+ public long getCollectionId() {
+ return collectionId;
+ }
+
+ @Override
+ public synchronized int size() {
+ return map.size();
+ }
+
+ public C getContext() {
+ if (context == null && contextProvider != null) {
+ context = contextProvider.apply(this.collectionId);
+ }
+ return context;
+ }
+
+ public JournalHashMap<K, V, C> setContext(C context) {
+ this.context = context;
+ return this;
+ }
+
+ @Override
+ public synchronized boolean isEmpty() {
+ return map.isEmpty();
+ }
+
+ @Override
+ public synchronized boolean containsKey(Object key) {
+ return map.containsKey(key);
+ }
+
+ @Override
+ public synchronized boolean containsValue(Object value) {
+ for (Entry<K, MapRecord<K, V>> entry : map.entrySet()) {
+ if (value.equals(entry.getValue().value)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public synchronized V get(Object key) {
+ MapRecord<K, V> reccord = map.get(key);
+ if (reccord == null) {
+ return null;
+ } else {
+ return reccord.value;
+ }
+ }
+
+ /** This is to be called from a single thread during reload, no need to be synchronized */
+ public void reload(MapRecord<K, V> reloadValue) {
+ map.put(reloadValue.getKey(), reloadValue);
+ }
+
+ @Override
+ public synchronized V put(K key, V value) {
+ logger.debug("adding {} = {}", key, value);
+ long id = idGenerator.getAsLong();
+ MapRecord<K, V> record = new MapRecord(collectionId, id, key, value);
+ store(key, record);
+ MapRecord<K, V> oldRecord = map.put(key, record);
+
+ if (oldRecord != null) {
+ removed(oldRecord);
+ return oldRecord.value;
+ } else {
+ return null;
+ }
+
+ }
+
+ private synchronized void store(K key, MapRecord<K, V> record) {
+ try {
+ IOCompletion callback = null;
+ if (completionSupplier != null) {
+ callback = completionSupplier.get();
+ }
+
+ if (callback == null) {
+ journal.appendAddRecord(record.id, recordType, persister, record, false);
+ } else {
+ journal.appendAddRecord(record.id, recordType, persister, record, true, callback);
+ }
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord) {
+ try {
+ journal.appendDeleteRecord(reccord.id, false);
+ } catch (Exception e) {
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord, long txid) {
+ try {
+ journal.appendDeleteRecordTransactional(txid, reccord.id);
+ } catch (Exception e) {
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ @Override
+ public synchronized V remove(Object key) {
+ MapRecord<K, V> record = map.remove(key);
+ this.removed(record);
+ return record.value;
+ }
+
+ /** This method will remove the element from the HashMap immediately however the record is still part of a transaction.
+ * This is not playing with rollbacks. So a rollback on the transaction wouldn't place the elements back.
+ * This is intended to make sure the operation would be atomic in case of a failure, while an appendRollback is not expected. */
+ public synchronized V remove(Object key, long transactionID) {
+ MapRecord<K, V> record = map.remove(key);
+ this.removed(record, transactionID);
+ return record.value;
+ }
+
+ @Override
+ public synchronized void putAll(Map<? extends K, ? extends V> m) {
+ m.forEach(this::put);
+ }
+
+ @Override
+ public synchronized void clear() {
+ map.values().forEach(v -> remove(v));
+ map.clear();
+ }
+
+ @Override
+ public synchronized Set<K> keySet() {
+ HashSet<K> keys = new HashSet(map.size());
+ map.values().forEach(v -> keys.add(v.key));
+ return keys;
+ }
Review Comment:
Not later on, if you expect it does what a Map should because its assigned as simply Map...then it turns out...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#issuecomment-1960397285
@gemmellr I will merge this now.. ping me if you find anything else later please.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499277765
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMap.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.journal.collections;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.LongFunction;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.IOCompletion;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JournalHashMap<K, V, C> implements Map<K, V> {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static class MapRecord<K, V> implements Entry<K, V> {
+ final long collectionID;
+ long id;
+ K key;
+ V value;
+
+ MapRecord(long collectionID, long id, K key, V value) {
+ this.collectionID = collectionID;
+ this.id = id;
+ this.key = key;
+ this.value = value;
+ }
+
+ @Override
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public V getValue() {
+ return value;
+ }
+
+ @Override
+ public V setValue(V value) {
+ V oldValue = this.value;
+ this.value = value;
+ return oldValue;
+ }
+
+ @Override
+ public String toString() {
+ return "MapRecord{" + "collectionID=" + collectionID + ", id=" + id + ", key=" + key + ", value=" + value + '}';
+ }
+ }
+
+ public JournalHashMap(long collectionId, Journal journal, LongSupplier idGenerator, Persister<MapRecord<K, V>> persister, byte recordType, Supplier<IOCompletion> completionSupplier, LongFunction<C> contextProvider, IOCriticalErrorListener ioExceptionListener) {
+ this.collectionId = collectionId;
+ this.journal = journal;
+ this.idGenerator = idGenerator;
+ this.persister = persister;
+ this.recordType = recordType;
+ this.exceptionListener = ioExceptionListener;
+ this.completionSupplier = completionSupplier;
+ this.contextProvider = contextProvider;
+ }
+
+ C context;
+
+ LongFunction<C> contextProvider;
+
+ private final Persister<MapRecord<K, V>> persister;
+
+ private final Journal journal;
+
+ private final long collectionId;
+
+ private final byte recordType;
+
+ private final LongSupplier idGenerator;
+
+ private final Supplier<IOCompletion> completionSupplier;
+
+ private final IOCriticalErrorListener exceptionListener;
+
+ private final Map<K, MapRecord<K, V>> map = new HashMap<>();
+
+ public long getCollectionId() {
+ return collectionId;
+ }
+
+ @Override
+ public synchronized int size() {
+ return map.size();
+ }
+
+ public C getContext() {
+ if (context == null && contextProvider != null) {
+ context = contextProvider.apply(this.collectionId);
+ }
+ return context;
+ }
+
+ public JournalHashMap<K, V, C> setContext(C context) {
+ this.context = context;
+ return this;
+ }
+
+ @Override
+ public synchronized boolean isEmpty() {
+ return map.isEmpty();
+ }
+
+ @Override
+ public synchronized boolean containsKey(Object key) {
+ return map.containsKey(key);
+ }
+
+ @Override
+ public synchronized boolean containsValue(Object value) {
+ for (Entry<K, MapRecord<K, V>> entry : map.entrySet()) {
+ if (value.equals(entry.getValue().value)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public synchronized V get(Object key) {
+ MapRecord<K, V> reccord = map.get(key);
+ if (reccord == null) {
+ return null;
+ } else {
+ return reccord.value;
+ }
+ }
+
+ /** This is to be called from a single thread during reload, no need to be synchronized */
+ public void reload(MapRecord<K, V> reloadValue) {
+ map.put(reloadValue.getKey(), reloadValue);
+ }
+
+ @Override
+ public synchronized V put(K key, V value) {
+ logger.debug("adding {} = {}", key, value);
+ long id = idGenerator.getAsLong();
+ MapRecord<K, V> record = new MapRecord(collectionId, id, key, value);
+ store(key, record);
+ MapRecord<K, V> oldRecord = map.put(key, record);
+
+ if (oldRecord != null) {
+ removed(oldRecord);
+ return oldRecord.value;
+ } else {
+ return null;
+ }
+
+ }
+
+ private synchronized void store(K key, MapRecord<K, V> record) {
+ try {
+ IOCompletion callback = null;
+ if (completionSupplier != null) {
+ callback = completionSupplier.get();
+ }
+
+ if (callback == null) {
+ journal.appendAddRecord(record.id, recordType, persister, record, false);
+ } else {
+ journal.appendAddRecord(record.id, recordType, persister, record, true, callback);
+ }
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord) {
+ try {
+ journal.appendDeleteRecord(reccord.id, false);
+ } catch (Exception e) {
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord, long txid) {
+ try {
+ journal.appendDeleteRecordTransactional(txid, reccord.id);
+ } catch (Exception e) {
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ @Override
+ public synchronized V remove(Object key) {
+ MapRecord<K, V> record = map.remove(key);
+ this.removed(record);
+ return record.value;
+ }
+
+ /** This method will remove the element from the HashMap immediately however the record is still part of a transaction.
+ * This is not playing with rollbacks. So a rollback on the transaction wouldn't place the elements back.
+ * This is intended to make sure the operation would be atomic in case of a failure, while an appendRollback is not expected. */
+ public synchronized V remove(Object key, long transactionID) {
+ MapRecord<K, V> record = map.remove(key);
+ this.removed(record, transactionID);
+ return record.value;
+ }
+
+ @Override
+ public synchronized void putAll(Map<? extends K, ? extends V> m) {
+ m.forEach(this::put);
+ }
+
+ @Override
+ public synchronized void clear() {
+ map.values().forEach(v -> remove(v));
+ map.clear();
+ }
+
+ @Override
+ public synchronized Set<K> keySet() {
+ HashSet<K> keys = new HashSet(map.size());
+ map.values().forEach(v -> keys.add(v.key));
+ return keys;
+ }
Review Comment:
I willl add a comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499399307
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMap.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.journal.collections;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.LongFunction;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.IOCompletion;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JournalHashMap<K, V, C> implements Map<K, V> {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static class MapRecord<K, V> implements Entry<K, V> {
+ final long collectionID;
+ long id;
+ K key;
+ V value;
+
+ MapRecord(long collectionID, long id, K key, V value) {
+ this.collectionID = collectionID;
+ this.id = id;
+ this.key = key;
+ this.value = value;
+ }
+
+ @Override
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public V getValue() {
+ return value;
+ }
+
+ @Override
+ public V setValue(V value) {
+ V oldValue = this.value;
+ this.value = value;
+ return oldValue;
+ }
+
+ @Override
+ public String toString() {
+ return "MapRecord{" + "collectionID=" + collectionID + ", id=" + id + ", key=" + key + ", value=" + value + '}';
+ }
+ }
+
+ public JournalHashMap(long collectionId, Journal journal, LongSupplier idGenerator, Persister<MapRecord<K, V>> persister, byte recordType, Supplier<IOCompletion> completionSupplier, LongFunction<C> contextProvider, IOCriticalErrorListener ioExceptionListener) {
+ this.collectionId = collectionId;
+ this.journal = journal;
+ this.idGenerator = idGenerator;
+ this.persister = persister;
+ this.recordType = recordType;
+ this.exceptionListener = ioExceptionListener;
+ this.completionSupplier = completionSupplier;
+ this.contextProvider = contextProvider;
+ }
+
+ C context;
+
+ LongFunction<C> contextProvider;
+
+ private final Persister<MapRecord<K, V>> persister;
+
+ private final Journal journal;
+
+ private final long collectionId;
+
+ private final byte recordType;
+
+ private final LongSupplier idGenerator;
+
+ private final Supplier<IOCompletion> completionSupplier;
+
+ private final IOCriticalErrorListener exceptionListener;
+
+ private final Map<K, MapRecord<K, V>> map = new HashMap<>();
+
+ public long getCollectionId() {
+ return collectionId;
+ }
+
+ @Override
+ public synchronized int size() {
+ return map.size();
+ }
+
+ public C getContext() {
+ if (context == null && contextProvider != null) {
+ context = contextProvider.apply(this.collectionId);
+ }
+ return context;
+ }
+
+ public JournalHashMap<K, V, C> setContext(C context) {
+ this.context = context;
+ return this;
+ }
+
+ @Override
+ public synchronized boolean isEmpty() {
+ return map.isEmpty();
+ }
+
+ @Override
+ public synchronized boolean containsKey(Object key) {
+ return map.containsKey(key);
+ }
+
+ @Override
+ public synchronized boolean containsValue(Object value) {
+ for (Entry<K, MapRecord<K, V>> entry : map.entrySet()) {
+ if (value.equals(entry.getValue().value)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public synchronized V get(Object key) {
+ MapRecord<K, V> reccord = map.get(key);
+ if (reccord == null) {
+ return null;
+ } else {
+ return reccord.value;
+ }
+ }
+
+ /** This is to be called from a single thread during reload, no need to be synchronized */
+ public void reload(MapRecord<K, V> reloadValue) {
+ map.put(reloadValue.getKey(), reloadValue);
+ }
+
+ @Override
+ public synchronized V put(K key, V value) {
+ logger.debug("adding {} = {}", key, value);
+ long id = idGenerator.getAsLong();
+ MapRecord<K, V> record = new MapRecord(collectionId, id, key, value);
+ store(key, record);
+ MapRecord<K, V> oldRecord = map.put(key, record);
+
+ if (oldRecord != null) {
+ removed(oldRecord);
+ return oldRecord.value;
+ } else {
+ return null;
+ }
+
+ }
+
+ private synchronized void store(K key, MapRecord<K, V> record) {
+ try {
+ IOCompletion callback = null;
+ if (completionSupplier != null) {
+ callback = completionSupplier.get();
+ }
+
+ if (callback == null) {
+ journal.appendAddRecord(record.id, recordType, persister, record, false);
+ } else {
+ journal.appendAddRecord(record.id, recordType, persister, record, true, callback);
+ }
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord) {
+ try {
+ journal.appendDeleteRecord(reccord.id, false);
+ } catch (Exception e) {
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord, long txid) {
+ try {
+ journal.appendDeleteRecordTransactional(txid, reccord.id);
+ } catch (Exception e) {
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ @Override
+ public synchronized V remove(Object key) {
+ MapRecord<K, V> record = map.remove(key);
+ this.removed(record);
+ return record.value;
+ }
+
+ /** This method will remove the element from the HashMap immediately however the record is still part of a transaction.
+ * This is not playing with rollbacks. So a rollback on the transaction wouldn't place the elements back.
+ * This is intended to make sure the operation would be atomic in case of a failure, while an appendRollback is not expected. */
+ public synchronized V remove(Object key, long transactionID) {
+ MapRecord<K, V> record = map.remove(key);
+ this.removed(record, transactionID);
+ return record.value;
+ }
+
+ @Override
+ public synchronized void putAll(Map<? extends K, ? extends V> m) {
+ m.forEach(this::put);
+ }
+
+ @Override
+ public synchronized void clear() {
+ map.values().forEach(v -> remove(v));
+ map.clear();
+ }
+
+ @Override
+ public synchronized Set<K> keySet() {
+ HashSet<K> keys = new HashSet(map.size());
+ map.values().forEach(v -> keys.add(v.key));
+ return keys;
+ }
Review Comment:
It makes it easier to understand though.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499236202
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AmqpSupport.java:
##########
@@ -43,6 +43,8 @@ public class AmqpSupport {
public static final int AMQP_CREDITS_DEFAULT = 1000;
public static final int AMQP_LOW_CREDITS_DEFAULT = 300;
+ public static final int AMQP_MIRROR_ACK_RETRY_INTERVAL = 10_000;
+
Review Comment:
nope.. will remove it.. thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499365864
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -424,6 +453,10 @@ public void preAcknowledge(final Transaction tx, final MessageReference ref, fin
MirrorController controllerInUse = getControllerInUse();
+ if (controllerInUse != null && !controllerInUse.isAllowACK()) {
Review Comment:
I dont really understand the reply. I was saying it seems weird for this "preAcknowledge" method to be logging about "postACKInternalMessage" just above this when that is a different method (and on the face of it, suggests a different pre vs post timing)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499381939
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AckManager.java:
##########
@@ -0,0 +1,513 @@
+/*
+ * 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.connect.mirror;
+
+import java.lang.invoke.MethodHandles;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.LongSupplier;
+
+import io.netty.util.collection.LongObjectHashMap;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.QueueConfiguration;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.journal.RecordInfo;
+import org.apache.activemq.artemis.core.journal.collections.JournalHashMap;
+import org.apache.activemq.artemis.core.journal.collections.JournalHashMapProvider;
+import org.apache.activemq.artemis.core.paging.PagingStore;
+import org.apache.activemq.artemis.core.paging.cursor.PageSubscription;
+import org.apache.activemq.artemis.core.paging.cursor.PagedReference;
+import org.apache.activemq.artemis.core.paging.impl.Page;
+import org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds;
+import org.apache.activemq.artemis.core.persistence.impl.journal.OperationContextImpl;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.AckRetry;
+import org.apache.activemq.artemis.core.server.ActiveMQComponent;
+import org.apache.activemq.artemis.core.server.ActiveMQScheduledComponent;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.MessageReference;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.core.server.RoutingContext;
+import org.apache.activemq.artemis.core.server.impl.AckReason;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.core.server.mirror.MirrorController;
+import org.apache.activemq.artemis.core.transaction.Transaction;
+import org.apache.activemq.artemis.core.transaction.impl.TransactionImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class AckManager implements ActiveMQComponent {
+
+ // we first retry on the queue a few tiems
+ public static final short MAX_QUEUE_ATTEMPT = Short.parseShort(System.getProperty(AckRetry.class.getName() + ".MAX_QUEUE_ATTEMPT", "5"));
Review Comment:
>> Do we retry on the queue after trying paging?
next time paging is retried, I will retry all the pending ACKs on the queue again, since I'm already there anyways.
I couldn't find a better name though other than MAX_QUEUE_ATTEMPT.
I actually used to name this MIN_QUEUE_ATTEMPT at some point.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499231125
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMap.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.journal.collections;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.LongFunction;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.IOCompletion;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JournalHashMap<K, V, C> implements Map<K, V> {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static class MapRecord<K, V> implements Entry<K, V> {
+ final long collectionID;
+ long id;
+ K key;
+ V value;
+
+ MapRecord(long collectionID, long id, K key, V value) {
+ this.collectionID = collectionID;
+ this.id = id;
+ this.key = key;
+ this.value = value;
+ }
+
+ @Override
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public V getValue() {
+ return value;
+ }
+
+ @Override
+ public V setValue(V value) {
+ V oldValue = this.value;
+ this.value = value;
+ return oldValue;
+ }
+
+ @Override
+ public String toString() {
+ return "MapRecord{" + "collectionID=" + collectionID + ", id=" + id + ", key=" + key + ", value=" + value + '}';
+ }
+ }
+
+ public JournalHashMap(long collectionId, Journal journal, LongSupplier idGenerator, Persister<MapRecord<K, V>> persister, byte recordType, Supplier<IOCompletion> completionSupplier, LongFunction<C> contextProvider, IOCriticalErrorListener ioExceptionListener) {
+ this.collectionId = collectionId;
+ this.journal = journal;
+ this.idGenerator = idGenerator;
+ this.persister = persister;
+ this.recordType = recordType;
+ this.exceptionListener = ioExceptionListener;
+ this.completionSupplier = completionSupplier;
+ this.contextProvider = contextProvider;
+ }
+
+ C context;
+
+ LongFunction<C> contextProvider;
+
+ private final Persister<MapRecord<K, V>> persister;
+
+ private final Journal journal;
+
+ private final long collectionId;
+
+ private final byte recordType;
+
+ private final LongSupplier idGenerator;
+
+ private final Supplier<IOCompletion> completionSupplier;
+
+ private final IOCriticalErrorListener exceptionListener;
+
+ private final Map<K, MapRecord<K, V>> map = new HashMap<>();
+
+ public long getCollectionId() {
+ return collectionId;
+ }
+
+ @Override
+ public synchronized int size() {
+ return map.size();
+ }
+
+ public C getContext() {
+ if (context == null && contextProvider != null) {
+ context = contextProvider.apply(this.collectionId);
+ }
+ return context;
+ }
+
+ public JournalHashMap<K, V, C> setContext(C context) {
+ this.context = context;
+ return this;
+ }
+
+ @Override
+ public synchronized boolean isEmpty() {
+ return map.isEmpty();
+ }
+
+ @Override
+ public synchronized boolean containsKey(Object key) {
+ return map.containsKey(key);
+ }
+
+ @Override
+ public synchronized boolean containsValue(Object value) {
+ for (Entry<K, MapRecord<K, V>> entry : map.entrySet()) {
+ if (value.equals(entry.getValue().value)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public synchronized V get(Object key) {
+ MapRecord<K, V> reccord = map.get(key);
+ if (reccord == null) {
+ return null;
+ } else {
+ return reccord.value;
+ }
+ }
+
+ /** This is to be called from a single thread during reload, no need to be synchronized */
+ public void reload(MapRecord<K, V> reloadValue) {
+ map.put(reloadValue.getKey(), reloadValue);
+ }
+
+ @Override
+ public synchronized V put(K key, V value) {
+ logger.debug("adding {} = {}", key, value);
+ long id = idGenerator.getAsLong();
+ MapRecord<K, V> record = new MapRecord(collectionId, id, key, value);
+ store(key, record);
+ MapRecord<K, V> oldRecord = map.put(key, record);
+
+ if (oldRecord != null) {
+ removed(oldRecord);
+ return oldRecord.value;
+ } else {
+ return null;
+ }
+
+ }
+
+ private synchronized void store(K key, MapRecord<K, V> record) {
+ try {
+ IOCompletion callback = null;
+ if (completionSupplier != null) {
+ callback = completionSupplier.get();
+ }
+
+ if (callback == null) {
+ journal.appendAddRecord(record.id, recordType, persister, record, false);
+ } else {
+ journal.appendAddRecord(record.id, recordType, persister, record, true, callback);
+ }
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord) {
+ try {
+ journal.appendDeleteRecord(reccord.id, false);
+ } catch (Exception e) {
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord, long txid) {
+ try {
+ journal.appendDeleteRecordTransactional(txid, reccord.id);
+ } catch (Exception e) {
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ @Override
+ public synchronized V remove(Object key) {
+ MapRecord<K, V> record = map.remove(key);
+ this.removed(record);
+ return record.value;
+ }
+
+ /** This method will remove the element from the HashMap immediately however the record is still part of a transaction.
+ * This is not playing with rollbacks. So a rollback on the transaction wouldn't place the elements back.
+ * This is intended to make sure the operation would be atomic in case of a failure, while an appendRollback is not expected. */
+ public synchronized V remove(Object key, long transactionID) {
+ MapRecord<K, V> record = map.remove(key);
+ this.removed(record, transactionID);
+ return record.value;
+ }
+
+ @Override
+ public synchronized void putAll(Map<? extends K, ? extends V> m) {
+ m.forEach(this::put);
+ }
+
+ @Override
+ public synchronized void clear() {
+ map.values().forEach(v -> remove(v));
+ map.clear();
+ }
+
+ @Override
+ public synchronized Set<K> keySet() {
+ HashSet<K> keys = new HashSet(map.size());
+ map.values().forEach(v -> keys.add(v.key));
+ return keys;
+ }
+
+ @Override
+ public synchronized Collection<V> values() {
+ ArrayList values = new ArrayList(map.size());
+ map.values().forEach(v -> values.add(v.value));
+ return values;
+ }
Review Comment:
I would rather break the contract and document it here and the other cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499356875
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -281,6 +300,15 @@ public void sendMessage(Transaction tx, Message message, RoutingContext context)
return;
}
+ if (snfQueue.getPagingStore().page(message, tx, pagedRouteContext, this::copyMessageForPaging)) {
+ return;
+ }
Review Comment:
ok...but if it is paging, it seems like this can throw, or delete the message, and other stuff...is that expected?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499402585
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AckManager.java:
##########
@@ -0,0 +1,513 @@
+/*
+ * 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.connect.mirror;
+
+import java.lang.invoke.MethodHandles;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.LongSupplier;
+
+import io.netty.util.collection.LongObjectHashMap;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.QueueConfiguration;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.journal.RecordInfo;
+import org.apache.activemq.artemis.core.journal.collections.JournalHashMap;
+import org.apache.activemq.artemis.core.journal.collections.JournalHashMapProvider;
+import org.apache.activemq.artemis.core.paging.PagingStore;
+import org.apache.activemq.artemis.core.paging.cursor.PageSubscription;
+import org.apache.activemq.artemis.core.paging.cursor.PagedReference;
+import org.apache.activemq.artemis.core.paging.impl.Page;
+import org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds;
+import org.apache.activemq.artemis.core.persistence.impl.journal.OperationContextImpl;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.AckRetry;
+import org.apache.activemq.artemis.core.server.ActiveMQComponent;
+import org.apache.activemq.artemis.core.server.ActiveMQScheduledComponent;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.MessageReference;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.core.server.RoutingContext;
+import org.apache.activemq.artemis.core.server.impl.AckReason;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.core.server.mirror.MirrorController;
+import org.apache.activemq.artemis.core.transaction.Transaction;
+import org.apache.activemq.artemis.core.transaction.impl.TransactionImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class AckManager implements ActiveMQComponent {
+
+ // we first retry on the queue a few tiems
+ public static final short MAX_QUEUE_ATTEMPT = Short.parseShort(System.getProperty(AckRetry.class.getName() + ".MAX_QUEUE_ATTEMPT", "5"));
Review Comment:
> > > Do we retry on the queue after trying paging?
>
> next time paging is retried, I will retry all the pending ACKs on the queue again, since I'm already there anyways.
>
What happens if we try the queue, but it isnt there so we try paging, but it isnt there either, then we try the queue again but it isnt there....we stop trying at that point right?
(e.g supposing it was actually in paging when we tried the queue...then was in the queue while we tried in paging...then vanishes due to e.g consumption or expiration or such like)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499518163
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -281,6 +300,15 @@ public void sendMessage(Transaction tx, Message message, RoutingContext context)
return;
}
+ if (snfQueue.getPagingStore().page(message, tx, pagedRouteContext, this::copyMessageForPaging)) {
+ return;
+ }
+
+ if (message.isPaged()) {
+ // if the source was paged, we copy the message
+ message = copyMessageForPaging(message);
+ }
Review Comment:
ok. I think the 2 if's should cover that quickly in the comments, its not obvious now so it definitely wont be later.
Also, thinking....Does any of this changing the ID stuff affect things in terms of acking later, or what happens when the destination broker is itself mirroring? E.g I see its setting an 'extra property'.. are those persisted? Or passed on if the message isnt reencoded? If the broker restarts before the message gets sent, does that change what the representation of this copied paged message actually is?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499447542
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -424,6 +453,10 @@ public void preAcknowledge(final Transaction tx, final MessageReference ref, fin
MirrorController controllerInUse = getControllerInUse();
+ if (controllerInUse != null && !controllerInUse.isAllowACK()) {
Review Comment:
Ohh... I see now.. when you said "this comment" I thought you were referring to this:
https://github.com/apache/activemq-artemis/blob/bc7e4639e0a4665c106f336fb837c33d2cbcd361/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java#L427
That log. was already there before, and I see it now.. I am changing it.
and I also made an improvement to the code there as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499992844
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -132,6 +138,15 @@ public AMQPMirrorControllerSource(ProtonProtocolManager protonProtocolManager, Q
this.acks = replicaConfig.isMessageAcknowledgements();
this.brokerConnection = brokerConnection;
this.sync = replicaConfig.isSync();
+ this.pagedRouteContext = new PagedRouteContext(snfQueue);
+
+ if (sync) {
+ logger.info("Mirror is configured to sync, so pageStore={} being enforced to BLOCK, and not page", snfQueue.getName());
Review Comment:
oops... almost merged this :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499266530
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/AbstractHashMapPersister.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.journal.collections;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.core.persistence.CoreMessageObjectPools;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.utils.DataConstants;
+
+public abstract class AbstractHashMapPersister<K, V> implements Persister<JournalHashMap.MapRecord<K, V>> {
+
+ @Override
+ public byte getID() {
+ return 0;
+ }
+
+ @Override
+ public final int getEncodeSize(JournalHashMap.MapRecord<K, V> record) {
+ return DataConstants.SIZE_BYTE + // FILLER, could be used for versioning in the future
+ DataConstants.SIZE_LONG + // recordID
+ DataConstants.SIZE_LONG + // collectionID
+ getKeySize(record.key) +
+ getValueSize(record.value);
+ }
+
+ protected abstract int getKeySize(K key);
+
+ protected abstract void encodeKey(ActiveMQBuffer buffer, K key);
+
+ protected abstract K decodeKey(ActiveMQBuffer buffer);
+
+ protected abstract int getValueSize(V value);
+
+ protected abstract void encodeValue(ActiveMQBuffer buffer, V value);
+
+ protected abstract V decodeValue(ActiveMQBuffer buffer, K key);
+
+ @Override
+ public final void encode(ActiveMQBuffer buffer, JournalHashMap.MapRecord<K, V> record) {
+ buffer.writeByte((byte)0); // filler - could be used for versioning in the future.
+ buffer.writeLong(record.id);
+ buffer.writeLong(record.collectionID);
+ encodeKey(buffer, record.key);
+ encodeValue(buffer, record.value);
+ }
+
+ @Override
+ public final JournalHashMap.MapRecord<K, V> decode(ActiveMQBuffer buffer,
+ JournalHashMap.MapRecord<K, V> record,
+ CoreMessageObjectPools pool) {
+ buffer.readByte(); // filler - not used currently - just in case we ever need to version this wiring
+ long id = buffer.readLong();
+ long collectionID = buffer.readLong();
+ K key = decodeKey(buffer);
+ V value = decodeValue(buffer, key);
+
+ JournalHashMap.MapRecord<K, V> mapRecord = new JournalHashMap.MapRecord<>(collectionID, id, key, value);
Review Comment:
all the other persisters have the firs few bytes as the recodID.
https://github.com/apache/activemq-artemis/blob/f790911c44c4fc830d577debb7163163679670c4/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/protocol/MessagePersister.java#L106
I will actually place the version bellow the ID to be consistent with the others.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499339024
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMap.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.journal.collections;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.LongFunction;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.IOCompletion;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JournalHashMap<K, V, C> implements Map<K, V> {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static class MapRecord<K, V> implements Entry<K, V> {
Review Comment:
The value here is the MapRecord... it can have them called for collisions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499377931
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -617,4 +650,55 @@ public static void routeMirrorCommand(ActiveMQServer server, Message message, Tr
server.getPostOffice().route(message, ctx, false);
}
+ class PagedRouteContext implements RouteContextList {
+
+ private final List<Queue> listRepresentation;
+ private final List<Queue> emptyList = Collections.emptyList();
+
+ PagedRouteContext() {
+ listRepresentation = new ArrayList<>(1);
+ listRepresentation.add(snfQueue);
Review Comment:
I thought it was simple enough.. but I will make a change here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#issuecomment-1964061636
I've spent a while more looking at this again and continue to find it fairly impenetrable, I'm surrendering on looking at it any further.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic merged PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499220752
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMap.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.journal.collections;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.LongFunction;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.IOCompletion;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JournalHashMap<K, V, C> implements Map<K, V> {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static class MapRecord<K, V> implements Entry<K, V> {
Review Comment:
only the value is being inserted. It will be dead code, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499323366
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/AbstractHashMapPersister.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.journal.collections;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.core.persistence.CoreMessageObjectPools;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.utils.DataConstants;
+
+public abstract class AbstractHashMapPersister<K, V> implements Persister<JournalHashMap.MapRecord<K, V>> {
+
+ @Override
+ public byte getID() {
+ return 0;
+ }
+
+ @Override
+ public final int getEncodeSize(JournalHashMap.MapRecord<K, V> record) {
+ return DataConstants.SIZE_BYTE + // FILLER, could be used for versioning in the future
+ DataConstants.SIZE_LONG + // recordID
+ DataConstants.SIZE_LONG + // collectionID
+ getKeySize(record.key) +
+ getValueSize(record.value);
+ }
+
+ protected abstract int getKeySize(K key);
+
+ protected abstract void encodeKey(ActiveMQBuffer buffer, K key);
+
+ protected abstract K decodeKey(ActiveMQBuffer buffer);
+
+ protected abstract int getValueSize(V value);
+
+ protected abstract void encodeValue(ActiveMQBuffer buffer, V value);
+
+ protected abstract V decodeValue(ActiveMQBuffer buffer, K key);
+
+ @Override
+ public final void encode(ActiveMQBuffer buffer, JournalHashMap.MapRecord<K, V> record) {
+ buffer.writeByte((byte)0); // filler - could be used for versioning in the future.
+ buffer.writeLong(record.id);
+ buffer.writeLong(record.collectionID);
+ encodeKey(buffer, record.key);
+ encodeValue(buffer, record.value);
+ }
+
+ @Override
+ public final JournalHashMap.MapRecord<K, V> decode(ActiveMQBuffer buffer,
+ JournalHashMap.MapRecord<K, V> record,
+ CoreMessageObjectPools pool) {
+ buffer.readByte(); // filler - not used currently - just in case we ever need to version this wiring
+ long id = buffer.readLong();
+ long collectionID = buffer.readLong();
+ K key = decodeKey(buffer);
+ V value = decodeValue(buffer, key);
+
+ JournalHashMap.MapRecord<K, V> mapRecord = new JournalHashMap.MapRecord<>(collectionID, id, key, value);
Review Comment:
Ok, well that maybe explains why the persister does what it does...but not why have the object representation does something slightly different? Would be more readable if they were ordered the same.
Placing the version second seems a bit weird. What if we want to change the ID type? Didnt you just mentioned the idea of doing that?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499449021
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -281,6 +300,15 @@ public void sendMessage(Transaction tx, Message message, RoutingContext context)
return;
}
+ if (snfQueue.getPagingStore().page(message, tx, pagedRouteContext, this::copyMessageForPaging)) {
+ return;
+ }
+
+ if (message.isPaged()) {
+ // if the source was paged, we copy the message
+ message = copyMessageForPaging(message);
+ }
Review Comment:
I - large messages
ii - the acks will now happen in different sources.. the message will be paged in the source queue... and not paged in the SNF, and with the same ID. so the message must be copied.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499459754
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMap.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.journal.collections;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.LongFunction;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.IOCompletion;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JournalHashMap<K, V, C> implements Map<K, V> {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static class MapRecord<K, V> implements Entry<K, V> {
+ final long collectionID;
+ long id;
+ K key;
+ V value;
+
+ MapRecord(long collectionID, long id, K key, V value) {
+ this.collectionID = collectionID;
+ this.id = id;
+ this.key = key;
+ this.value = value;
+ }
+
+ @Override
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public V getValue() {
+ return value;
+ }
+
+ @Override
+ public V setValue(V value) {
+ V oldValue = this.value;
+ this.value = value;
+ return oldValue;
+ }
+
+ @Override
+ public String toString() {
+ return "MapRecord{" + "collectionID=" + collectionID + ", id=" + id + ", key=" + key + ", value=" + value + '}';
+ }
+ }
+
+ public JournalHashMap(long collectionId, Journal journal, LongSupplier idGenerator, Persister<MapRecord<K, V>> persister, byte recordType, Supplier<IOCompletion> completionSupplier, LongFunction<C> contextProvider, IOCriticalErrorListener ioExceptionListener) {
+ this.collectionId = collectionId;
+ this.journal = journal;
+ this.idGenerator = idGenerator;
+ this.persister = persister;
+ this.recordType = recordType;
+ this.exceptionListener = ioExceptionListener;
+ this.completionSupplier = completionSupplier;
+ this.contextProvider = contextProvider;
+ }
+
+ C context;
+
+ LongFunction<C> contextProvider;
+
+ private final Persister<MapRecord<K, V>> persister;
+
+ private final Journal journal;
+
+ private final long collectionId;
+
+ private final byte recordType;
+
+ private final LongSupplier idGenerator;
+
+ private final Supplier<IOCompletion> completionSupplier;
+
+ private final IOCriticalErrorListener exceptionListener;
+
+ private final Map<K, MapRecord<K, V>> map = new HashMap<>();
+
+ public long getCollectionId() {
+ return collectionId;
+ }
+
+ @Override
+ public synchronized int size() {
+ return map.size();
+ }
+
+ public C getContext() {
+ if (context == null && contextProvider != null) {
+ context = contextProvider.apply(this.collectionId);
+ }
+ return context;
+ }
+
+ public JournalHashMap<K, V, C> setContext(C context) {
+ this.context = context;
+ return this;
+ }
+
+ @Override
+ public synchronized boolean isEmpty() {
+ return map.isEmpty();
+ }
+
+ @Override
+ public synchronized boolean containsKey(Object key) {
+ return map.containsKey(key);
+ }
+
+ @Override
+ public synchronized boolean containsValue(Object value) {
+ for (Entry<K, MapRecord<K, V>> entry : map.entrySet()) {
+ if (value.equals(entry.getValue().value)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public synchronized V get(Object key) {
+ MapRecord<K, V> reccord = map.get(key);
+ if (reccord == null) {
+ return null;
+ } else {
+ return reccord.value;
+ }
+ }
+
+ /** This is to be called from a single thread during reload, no need to be synchronized */
+ public void reload(MapRecord<K, V> reloadValue) {
+ map.put(reloadValue.getKey(), reloadValue);
+ }
+
+ @Override
+ public synchronized V put(K key, V value) {
+ logger.debug("adding {} = {}", key, value);
+ long id = idGenerator.getAsLong();
+ MapRecord<K, V> record = new MapRecord(collectionId, id, key, value);
+ store(key, record);
+ MapRecord<K, V> oldRecord = map.put(key, record);
+
+ if (oldRecord != null) {
+ removed(oldRecord);
+ return oldRecord.value;
+ } else {
+ return null;
+ }
+
+ }
+
+ private synchronized void store(K key, MapRecord<K, V> record) {
+ try {
+ IOCompletion callback = null;
+ if (completionSupplier != null) {
+ callback = completionSupplier.get();
+ }
+
+ if (callback == null) {
+ journal.appendAddRecord(record.id, recordType, persister, record, false);
+ } else {
+ journal.appendAddRecord(record.id, recordType, persister, record, true, callback);
+ }
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord) {
+ try {
+ journal.appendDeleteRecord(reccord.id, false);
+ } catch (Exception e) {
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord, long txid) {
+ try {
+ journal.appendDeleteRecordTransactional(txid, reccord.id);
+ } catch (Exception e) {
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ @Override
+ public synchronized V remove(Object key) {
+ MapRecord<K, V> record = map.remove(key);
+ this.removed(record);
+ return record.value;
+ }
+
+ /** This method will remove the element from the HashMap immediately however the record is still part of a transaction.
+ * This is not playing with rollbacks. So a rollback on the transaction wouldn't place the elements back.
+ * This is intended to make sure the operation would be atomic in case of a failure, while an appendRollback is not expected. */
+ public synchronized V remove(Object key, long transactionID) {
+ MapRecord<K, V> record = map.remove(key);
+ this.removed(record, transactionID);
+ return record.value;
+ }
+
+ @Override
+ public synchronized void putAll(Map<? extends K, ? extends V> m) {
+ m.forEach(this::put);
+ }
+
+ @Override
+ public synchronized void clear() {
+ map.values().forEach(v -> remove(v));
+ map.clear();
+ }
+
+ @Override
+ public synchronized Set<K> keySet() {
+ HashSet<K> keys = new HashSet(map.size());
+ map.values().forEach(v -> keys.add(v.key));
+ return keys;
+ }
Review Comment:
I will throw some exception such as notImplementedYet! I may do it properly later on.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499453054
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -281,6 +300,15 @@ public void sendMessage(Transaction tx, Message message, RoutingContext context)
return;
}
+ if (snfQueue.getPagingStore().page(message, tx, pagedRouteContext, this::copyMessageForPaging)) {
+ return;
+ }
+
+ if (message.isPaged()) {
+ // if the source was paged, we copy the message
+ message = copyMessageForPaging(message);
+ }
Review Comment:
also, when the message is not paged.. I can add a reference into two different queues.. However it's paged... it gets messy. so I have to switch the message here *only* if either is paged.
Notice that the copy if the target is paged will happen in a different place (inside the page method where I send a method reference).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1497556085
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AckManager.java:
##########
@@ -0,0 +1,479 @@
+/*
+ * 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.connect.mirror;
+
+import java.lang.invoke.MethodHandles;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.LongSupplier;
+
+import io.netty.util.collection.LongObjectHashMap;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.QueueConfiguration;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.journal.RecordInfo;
+import org.apache.activemq.artemis.core.journal.collections.JournalHashMap;
+import org.apache.activemq.artemis.core.journal.collections.JournalHashMapProvider;
+import org.apache.activemq.artemis.core.paging.PagingStore;
+import org.apache.activemq.artemis.core.paging.cursor.PageSubscription;
+import org.apache.activemq.artemis.core.paging.cursor.PagedReference;
+import org.apache.activemq.artemis.core.paging.impl.Page;
+import org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds;
+import org.apache.activemq.artemis.core.persistence.impl.journal.OperationContextImpl;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.AckRetry;
+import org.apache.activemq.artemis.core.postoffice.impl.LocalQueueBinding;
+import org.apache.activemq.artemis.core.server.ActiveMQComponent;
+import org.apache.activemq.artemis.core.server.ActiveMQScheduledComponent;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.MessageReference;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.core.server.RoutingContext;
+import org.apache.activemq.artemis.core.server.impl.AckReason;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.core.server.mirror.MirrorController;
+import org.apache.activemq.artemis.core.transaction.Transaction;
+import org.apache.activemq.artemis.core.transaction.impl.TransactionImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class AckManager implements ActiveMQComponent {
+
+ // we first retry on the queue a few tiems
+ public static final short MIN_QUEUE_ATTEMPT = 4;
Review Comment:
I will rename this to MAX_QUEUE_ATTEMPT
and If I'm able to add a configuration element for the AckManager, perhaps I could configure these two attributes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499382695
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AckManager.java:
##########
@@ -0,0 +1,513 @@
+/*
+ * 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.connect.mirror;
+
+import java.lang.invoke.MethodHandles;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.LongSupplier;
+
+import io.netty.util.collection.LongObjectHashMap;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.QueueConfiguration;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.journal.RecordInfo;
+import org.apache.activemq.artemis.core.journal.collections.JournalHashMap;
+import org.apache.activemq.artemis.core.journal.collections.JournalHashMapProvider;
+import org.apache.activemq.artemis.core.paging.PagingStore;
+import org.apache.activemq.artemis.core.paging.cursor.PageSubscription;
+import org.apache.activemq.artemis.core.paging.cursor.PagedReference;
+import org.apache.activemq.artemis.core.paging.impl.Page;
+import org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds;
+import org.apache.activemq.artemis.core.persistence.impl.journal.OperationContextImpl;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.AckRetry;
+import org.apache.activemq.artemis.core.server.ActiveMQComponent;
+import org.apache.activemq.artemis.core.server.ActiveMQScheduledComponent;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.MessageReference;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.core.server.RoutingContext;
+import org.apache.activemq.artemis.core.server.impl.AckReason;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.core.server.mirror.MirrorController;
+import org.apache.activemq.artemis.core.transaction.Transaction;
+import org.apache.activemq.artemis.core.transaction.impl.TransactionImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class AckManager implements ActiveMQComponent {
+
+ // we first retry on the queue a few tiems
+ public static final short MAX_QUEUE_ATTEMPT = Short.parseShort(System.getProperty(AckRetry.class.getName() + ".MAX_QUEUE_ATTEMPT", "5"));
Review Comment:
I will rename it back to MIN_QUEUE_ATTEMPTS (will include the S)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499405901
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AckManager.java:
##########
@@ -0,0 +1,513 @@
+/*
+ * 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.connect.mirror;
+
+import java.lang.invoke.MethodHandles;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.LongSupplier;
+
+import io.netty.util.collection.LongObjectHashMap;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.QueueConfiguration;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.journal.RecordInfo;
+import org.apache.activemq.artemis.core.journal.collections.JournalHashMap;
+import org.apache.activemq.artemis.core.journal.collections.JournalHashMapProvider;
+import org.apache.activemq.artemis.core.paging.PagingStore;
+import org.apache.activemq.artemis.core.paging.cursor.PageSubscription;
+import org.apache.activemq.artemis.core.paging.cursor.PagedReference;
+import org.apache.activemq.artemis.core.paging.impl.Page;
+import org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds;
+import org.apache.activemq.artemis.core.persistence.impl.journal.OperationContextImpl;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.AckRetry;
+import org.apache.activemq.artemis.core.server.ActiveMQComponent;
+import org.apache.activemq.artemis.core.server.ActiveMQScheduledComponent;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.MessageReference;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.core.server.RoutingContext;
+import org.apache.activemq.artemis.core.server.impl.AckReason;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
+import org.apache.activemq.artemis.core.server.mirror.MirrorController;
+import org.apache.activemq.artemis.core.transaction.Transaction;
+import org.apache.activemq.artemis.core.transaction.impl.TransactionImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class AckManager implements ActiveMQComponent {
+
+ // we first retry on the queue a few tiems
+ public static final short MAX_QUEUE_ATTEMPT = Short.parseShort(System.getProperty(AckRetry.class.getName() + ".MAX_QUEUE_ATTEMPT", "5"));
Review Comment:
> I will rename it back to MIN_QUEUE_ATTEMPTS (will include the S)
Maybe drop the MIN or MAX since its not really either? Just QUEUE_ATTEMPTS?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499474846
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -281,6 +300,15 @@ public void sendMessage(Transaction tx, Message message, RoutingContext context)
return;
}
+ if (snfQueue.getPagingStore().page(message, tx, pagedRouteContext, this::copyMessageForPaging)) {
+ return;
+ }
Review Comment:
If an user misconfigured to throw.. yes.. wrong things can happen.
I could log a warn.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499550612
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -281,6 +300,15 @@ public void sendMessage(Transaction tx, Message message, RoutingContext context)
return;
}
+ if (snfQueue.getPagingStore().page(message, tx, pagedRouteContext, this::copyMessageForPaging)) {
+ return;
+ }
+
+ if (message.isPaged()) {
+ // if the source was paged, we copy the message
+ message = copyMessageForPaging(message);
+ }
Review Comment:
I had to change the ID providers to return the ID from the broker property if being set. so.. on that case the ID comes into the right place.
I did a lot of testing to make sure I got this right... a lot of work into just that part of the acks from a different queue.. it wasn't easy to balance it out. :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499917788
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -281,6 +300,15 @@ public void sendMessage(Transaction tx, Message message, RoutingContext context)
return;
}
+ if (snfQueue.getPagingStore().page(message, tx, pagedRouteContext, this::copyMessageForPaging)) {
+ return;
+ }
Review Comment:
I'm adding an option to enforce a certain page policy. When sync = true, I'm enforcing BLOCK, and when sync = false, I'm enforcing PAGE. (No drop or fail ever allowed on mirroring).
And if sync we must not page otherwise the CompletionContext is lost through paging.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#issuecomment-1960404186
I will actually wait a few more days before merging this.. but I think it's ready now.. a lot of test done here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] ARTEMIS-4651 Performance improvements on Mirror and Paging [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #4827:
URL: https://github.com/apache/activemq-artemis/pull/4827#discussion_r1499340484
##########
artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMap.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * 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.journal.collections;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.function.LongFunction;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.IOCompletion;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JournalHashMap<K, V, C> implements Map<K, V> {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static class MapRecord<K, V> implements Entry<K, V> {
+ final long collectionID;
+ long id;
+ K key;
+ V value;
+
+ MapRecord(long collectionID, long id, K key, V value) {
+ this.collectionID = collectionID;
+ this.id = id;
+ this.key = key;
+ this.value = value;
+ }
+
+ @Override
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public V getValue() {
+ return value;
+ }
+
+ @Override
+ public V setValue(V value) {
+ V oldValue = this.value;
+ this.value = value;
+ return oldValue;
+ }
+
+ @Override
+ public String toString() {
+ return "MapRecord{" + "collectionID=" + collectionID + ", id=" + id + ", key=" + key + ", value=" + value + '}';
+ }
+ }
+
+ public JournalHashMap(long collectionId, Journal journal, LongSupplier idGenerator, Persister<MapRecord<K, V>> persister, byte recordType, Supplier<IOCompletion> completionSupplier, LongFunction<C> contextProvider, IOCriticalErrorListener ioExceptionListener) {
+ this.collectionId = collectionId;
+ this.journal = journal;
+ this.idGenerator = idGenerator;
+ this.persister = persister;
+ this.recordType = recordType;
+ this.exceptionListener = ioExceptionListener;
+ this.completionSupplier = completionSupplier;
+ this.contextProvider = contextProvider;
+ }
+
+ C context;
+
+ LongFunction<C> contextProvider;
+
+ private final Persister<MapRecord<K, V>> persister;
+
+ private final Journal journal;
+
+ private final long collectionId;
+
+ private final byte recordType;
+
+ private final LongSupplier idGenerator;
+
+ private final Supplier<IOCompletion> completionSupplier;
+
+ private final IOCriticalErrorListener exceptionListener;
+
+ private final Map<K, MapRecord<K, V>> map = new HashMap<>();
+
+ public long getCollectionId() {
+ return collectionId;
+ }
+
+ @Override
+ public synchronized int size() {
+ return map.size();
+ }
+
+ public C getContext() {
+ if (context == null && contextProvider != null) {
+ context = contextProvider.apply(this.collectionId);
+ }
+ return context;
+ }
+
+ public JournalHashMap<K, V, C> setContext(C context) {
+ this.context = context;
+ return this;
+ }
+
+ @Override
+ public synchronized boolean isEmpty() {
+ return map.isEmpty();
+ }
+
+ @Override
+ public synchronized boolean containsKey(Object key) {
+ return map.containsKey(key);
+ }
+
+ @Override
+ public synchronized boolean containsValue(Object value) {
+ for (Entry<K, MapRecord<K, V>> entry : map.entrySet()) {
+ if (value.equals(entry.getValue().value)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public synchronized V get(Object key) {
+ MapRecord<K, V> reccord = map.get(key);
+ if (reccord == null) {
+ return null;
+ } else {
+ return reccord.value;
+ }
+ }
+
+ /** This is to be called from a single thread during reload, no need to be synchronized */
+ public void reload(MapRecord<K, V> reloadValue) {
+ map.put(reloadValue.getKey(), reloadValue);
+ }
+
+ @Override
+ public synchronized V put(K key, V value) {
+ logger.debug("adding {} = {}", key, value);
+ long id = idGenerator.getAsLong();
+ MapRecord<K, V> record = new MapRecord(collectionId, id, key, value);
+ store(key, record);
+ MapRecord<K, V> oldRecord = map.put(key, record);
+
+ if (oldRecord != null) {
+ removed(oldRecord);
+ return oldRecord.value;
+ } else {
+ return null;
+ }
+
+ }
+
+ private synchronized void store(K key, MapRecord<K, V> record) {
+ try {
+ IOCompletion callback = null;
+ if (completionSupplier != null) {
+ callback = completionSupplier.get();
+ }
+
+ if (callback == null) {
+ journal.appendAddRecord(record.id, recordType, persister, record, false);
+ } else {
+ journal.appendAddRecord(record.id, recordType, persister, record, true, callback);
+ }
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord) {
+ try {
+ journal.appendDeleteRecord(reccord.id, false);
+ } catch (Exception e) {
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ // callers must be synchronized
+ private void removed(MapRecord reccord, long txid) {
+ try {
+ journal.appendDeleteRecordTransactional(txid, reccord.id);
+ } catch (Exception e) {
+ exceptionListener.onIOException(e, e.getMessage(), null);
+ }
+ }
+
+ @Override
+ public synchronized V remove(Object key) {
+ MapRecord<K, V> record = map.remove(key);
+ this.removed(record);
+ return record.value;
+ }
+
+ /** This method will remove the element from the HashMap immediately however the record is still part of a transaction.
+ * This is not playing with rollbacks. So a rollback on the transaction wouldn't place the elements back.
+ * This is intended to make sure the operation would be atomic in case of a failure, while an appendRollback is not expected. */
+ public synchronized V remove(Object key, long transactionID) {
+ MapRecord<K, V> record = map.remove(key);
+ this.removed(record, transactionID);
+ return record.value;
+ }
+
+ @Override
+ public synchronized void putAll(Map<? extends K, ? extends V> m) {
+ m.forEach(this::put);
+ }
+
+ @Override
+ public synchronized void clear() {
+ map.values().forEach(v -> remove(v));
+ map.clear();
+ }
+
+ @Override
+ public synchronized Set<K> keySet() {
+ HashSet<K> keys = new HashSet(map.size());
+ map.values().forEach(v -> keys.add(v.key));
+ return keys;
+ }
Review Comment:
What about just not implement Map, if not _actually_ implementing it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org