You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2020/01/17 19:31:59 UTC
[activemq-artemis] branch master updated (2f68f00 -> 3b61764)
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git.
from 2f68f00 This closes #2943
new 4cc6464 ARTEMIS-2602 Improve Journal loading heap usage
new b10d765 ARTEMIS-2602 load surviving records into SparseArrayLinkedList
new 3b61764 This closes #2944
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
artemis-commons/pom.xml | 6 +
.../artemis/utils/collections/LongHashSet.java | 622 ++++++++++++++++
.../utils/collections/SparseArrayLinkedList.java | 243 +++++++
.../artemis/utils/collections/LongHashSetTest.java | 778 +++++++++++++++++++++
.../collections/SparseArrayLinkedListTest.java | 158 +++++
.../jdbc/store/journal/JDBCJournalImpl.java | 12 +
.../activemq/artemis/core/journal/Journal.java | 20 +-
.../core/journal/impl/FileWrapperJournal.java | 3 +-
.../artemis/core/journal/impl/JournalImpl.java | 40 +-
.../journal/AbstractJournalStorageManager.java | 466 ++++++------
.../impl/journal/JournalStorageManager.java | 5 +-
.../journal/LargeMessageTXFailureCallback.java | 9 +-
.../core/replication/ReplicatedJournal.java | 3 +-
.../integration/replication/ReplicationTest.java | 3 +-
.../core/journal/impl/JournalImplTestBase.java | 2 +-
15 files changed, 2090 insertions(+), 280 deletions(-)
create mode 100644 artemis-commons/src/main/java/org/apache/activemq/artemis/utils/collections/LongHashSet.java
create mode 100644 artemis-commons/src/main/java/org/apache/activemq/artemis/utils/collections/SparseArrayLinkedList.java
create mode 100644 artemis-commons/src/test/java/org/apache/activemq/artemis/utils/collections/LongHashSetTest.java
create mode 100644 artemis-commons/src/test/java/org/apache/activemq/artemis/utils/collections/SparseArrayLinkedListTest.java
[activemq-artemis] 02/03: ARTEMIS-2602 load surviving records into
SparseArrayLinkedList
Posted by cl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit b10d7651394033f8688d1606b42451c1386815e7
Author: Francesco Nigro <ni...@gmail.com>
AuthorDate: Fri Jan 17 15:21:11 2020 +0100
ARTEMIS-2602 load surviving records into SparseArrayLinkedList
---
.../jdbc/store/journal/JDBCJournalImpl.java | 12 +
.../activemq/artemis/core/journal/Journal.java | 20 +-
.../core/journal/impl/FileWrapperJournal.java | 3 +-
.../artemis/core/journal/impl/JournalImpl.java | 4 +-
.../journal/AbstractJournalStorageManager.java | 466 ++++++++++-----------
.../impl/journal/JournalStorageManager.java | 5 +-
.../journal/LargeMessageTXFailureCallback.java | 9 +-
.../core/replication/ReplicatedJournal.java | 3 +-
.../integration/replication/ReplicationTest.java | 3 +-
.../core/journal/impl/JournalImplTestBase.java | 2 +-
10 files changed, 265 insertions(+), 262 deletions(-)
diff --git a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/journal/JDBCJournalImpl.java b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/journal/JDBCJournalImpl.java
index 079a457..0507a22 100644
--- a/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/journal/JDBCJournalImpl.java
+++ b/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/journal/JDBCJournalImpl.java
@@ -30,6 +30,7 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.ActiveMQExceptionType;
@@ -896,6 +897,17 @@ public class JDBCJournalImpl extends AbstractJDBCDriver implements Journal {
}
@Override
+ public synchronized JournalLoadInformation load(final Consumer<RecordInfo> onCommittedRecords,
+ final List<PreparedTransactionInfo> preparedTransactions,
+ final TransactionFailureCallback failureCallback,
+ final boolean fixBadTX) throws Exception {
+ final List<RecordInfo> committedRecords = new ArrayList<>();
+ final JournalLoadInformation journalLoadInformation = load(committedRecords, preparedTransactions, failureCallback, fixBadTX);
+ committedRecords.forEach(onCommittedRecords);
+ return journalLoadInformation;
+ }
+
+ @Override
public synchronized JournalLoadInformation load(final List<RecordInfo> committedRecords,
final List<PreparedTransactionInfo> preparedTransactions,
final TransactionFailureCallback failureCallback,
diff --git a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/Journal.java b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/Journal.java
index 6844614..2da0543 100644
--- a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/Journal.java
+++ b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/Journal.java
@@ -18,6 +18,7 @@ package org.apache.activemq.artemis.core.journal;
import java.util.List;
import java.util.Map;
+import java.util.function.Consumer;
import org.apache.activemq.artemis.core.io.SequentialFileFactory;
import org.apache.activemq.artemis.core.journal.impl.JournalFile;
@@ -195,12 +196,25 @@ public interface Journal extends ActiveMQComponent {
void lineUpContext(IOCompletion callback);
default JournalLoadInformation load(List<RecordInfo> committedRecords,
- List<PreparedTransactionInfo> preparedTransactions,
- TransactionFailureCallback transactionFailure) throws Exception {
+ List<PreparedTransactionInfo> preparedTransactions,
+ TransactionFailureCallback transactionFailure) throws Exception {
+ return load(committedRecords == null ? null : committedRecords::add, preparedTransactions, transactionFailure, true);
+ }
+
+ default JournalLoadInformation load(Consumer<RecordInfo> committedRecords,
+ List<PreparedTransactionInfo> preparedTransactions,
+ TransactionFailureCallback transactionFailure) throws Exception {
return load(committedRecords, preparedTransactions, transactionFailure, true);
}
- JournalLoadInformation load(List<RecordInfo> committedRecords,
+ default JournalLoadInformation load(List<RecordInfo> committedRecords,
+ List<PreparedTransactionInfo> preparedTransactions,
+ TransactionFailureCallback transactionFailure,
+ boolean fixBadTx) throws Exception {
+ return load(committedRecords == null ? null : committedRecords::add, preparedTransactions, transactionFailure, fixBadTx);
+ }
+
+ JournalLoadInformation load(Consumer<RecordInfo> committedRecords,
List<PreparedTransactionInfo> preparedTransactions,
TransactionFailureCallback transactionFailure,
boolean fixBadTx) throws Exception;
diff --git a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/FileWrapperJournal.java b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/FileWrapperJournal.java
index 6a5ebbd..71cbf5a 100644
--- a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/FileWrapperJournal.java
+++ b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/FileWrapperJournal.java
@@ -20,6 +20,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.ActiveMQUnsupportedPacketException;
@@ -267,7 +268,7 @@ public final class FileWrapperJournal extends JournalBase {
}
@Override
- public JournalLoadInformation load(List<RecordInfo> committedRecords,
+ public JournalLoadInformation load(Consumer<RecordInfo> committedRecords,
List<PreparedTransactionInfo> preparedTransactions,
TransactionFailureCallback transactionFailure,
boolean fixbadtx) throws Exception {
diff --git a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java
index 55e9be6..8c35131 100644
--- a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java
+++ b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java
@@ -1445,7 +1445,7 @@ public class JournalImpl extends JournalBase implements TestableJournal, Journal
* @see JournalImpl#load(LoaderCallback)
*/
@Override
- public synchronized JournalLoadInformation load(final List<RecordInfo> committedRecords,
+ public synchronized JournalLoadInformation load(final Consumer<RecordInfo> committedRecords,
final List<PreparedTransactionInfo> preparedTransactions,
final TransactionFailureCallback failureCallback,
final boolean fixBadTX) throws Exception {
@@ -1514,7 +1514,7 @@ public class JournalImpl extends JournalBase implements TestableJournal, Journal
final Consumer<RecordInfo> fillCommittedRecord = record -> {
if (!recordsToDelete.contains(record.id)) {
- committedRecords.add(record);
+ committedRecords.accept(record);
}
};
// it helps GC by cleaning up each SparseArray too
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
index 79bab76..d4c6d53 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
@@ -23,11 +23,7 @@ import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalR
import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.PAGE_CURSOR_COUNTER_VALUE;
import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.SET_SCHEDULED_DELIVERY_TIME;
-import java.io.File;
-import java.io.FileInputStream;
-import java.security.DigestInputStream;
import java.security.InvalidParameterException;
-import java.security.MessageDigest;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
@@ -111,10 +107,10 @@ import org.apache.activemq.artemis.core.transaction.Transaction;
import org.apache.activemq.artemis.core.transaction.TransactionPropertyIndexes;
import org.apache.activemq.artemis.core.transaction.impl.TransactionImpl;
import org.apache.activemq.artemis.spi.core.protocol.MessagePersister;
-import org.apache.activemq.artemis.utils.Base64;
import org.apache.activemq.artemis.utils.ExecutorFactory;
import org.apache.activemq.artemis.utils.IDGenerator;
import org.apache.activemq.artemis.utils.collections.ConcurrentLongHashMap;
+import org.apache.activemq.artemis.utils.collections.SparseArrayLinkedList;
import org.apache.activemq.artemis.utils.critical.CriticalAnalyzer;
import org.apache.activemq.artemis.utils.critical.CriticalComponentImpl;
import org.jboss.logging.Logger;
@@ -258,25 +254,6 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
OperationContextImpl.clearContext();
}
- public static String md5(File file) {
- try {
- byte[] buffer = new byte[1 << 4];
- MessageDigest md = MessageDigest.getInstance("MD5");
-
- byte[] digest;
- try (FileInputStream is = new FileInputStream(file);
- DigestInputStream is2 = new DigestInputStream(is, md)) {
- while (is2.read(buffer) > 0) {
- continue;
- }
- digest = md.digest();
- }
- return Base64.encodeBytes(digest);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
public IDGenerator getIDGenerator() {
return idGenerator;
}
@@ -853,17 +830,17 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
final Set<Pair<Long, Long>> pendingLargeMessages,
List<PageCountPending> pendingNonTXPageCounter,
final JournalLoader journalLoader) throws Exception {
- List<RecordInfo> records = new ArrayList<>();
+ SparseArrayLinkedList<RecordInfo> records = new SparseArrayLinkedList<>();
List<PreparedTransactionInfo> preparedTransactions = new ArrayList<>();
- Set<PageTransactionInfo> invalidPageTransactions = null;
+ Set<PageTransactionInfo> invalidPageTransactions = new HashSet<>();
Map<Long, Message> messages = new HashMap<>();
readLock();
try {
- JournalLoadInformation info = messageJournal.load(records, preparedTransactions, new LargeMessageTXFailureCallback(this, messages));
+ JournalLoadInformation info = messageJournal.load(records::add, preparedTransactions, new LargeMessageTXFailureCallback(this));
ArrayList<LargeServerMessage> largeMessages = new ArrayList<>();
@@ -871,317 +848,321 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
Map<Long, PageSubscription> pageSubscriptions = new HashMap<>();
- final int totalSize = records.size();
+ final long totalSize = records.size();
- for (int reccount = 0; reccount < totalSize; reccount++) {
- // It will show log.info only with large journals (more than 1 million records)
- if (reccount > 0 && reccount % 1000000 == 0) {
- long percent = (long) ((((double) reccount) / ((double) totalSize)) * 100f);
+ final class MutableLong {
- ActiveMQServerLogger.LOGGER.percentLoaded(percent);
- }
+ long value;
+ }
- RecordInfo record = records.get(reccount);
- byte[] data = record.data;
+ final MutableLong recordNumber = new MutableLong();
+ // This will free up memory sooner while reading the records
+ records.clear(record -> {
+ try {
+ // It will show log.info only with large journals (more than 1 million records)
+ if (recordNumber.value > 0 && recordNumber.value % 1000000 == 0) {
+ long percent = (long) ((((double) recordNumber.value) / ((double) totalSize)) * 100f);
- ActiveMQBuffer buff = ActiveMQBuffers.wrappedBuffer(data);
+ ActiveMQServerLogger.LOGGER.percentLoaded(percent);
+ }
+ recordNumber.value++;
- byte recordType = record.getUserRecordType();
+ byte[] data = record.data;
- switch (recordType) {
- case JournalRecordIds.ADD_LARGE_MESSAGE_PENDING: {
- PendingLargeMessageEncoding pending = new PendingLargeMessageEncoding();
+ ActiveMQBuffer buff = ActiveMQBuffers.wrappedBuffer(data);
+
+ byte recordType = record.getUserRecordType();
- pending.decode(buff);
+ switch (recordType) {
+ case JournalRecordIds.ADD_LARGE_MESSAGE_PENDING: {
+ PendingLargeMessageEncoding pending = new PendingLargeMessageEncoding();
- if (pendingLargeMessages != null) {
- // it could be null on tests, and we don't need anything on that case
- pendingLargeMessages.add(new Pair<>(record.id, pending.largeMessageID));
+ pending.decode(buff);
+
+ if (pendingLargeMessages != null) {
+ // it could be null on tests, and we don't need anything on that case
+ pendingLargeMessages.add(new Pair<>(record.id, pending.largeMessageID));
+ }
+ break;
}
- break;
- }
- case JournalRecordIds.ADD_LARGE_MESSAGE: {
- LargeServerMessage largeMessage = parseLargeMessage(messages, buff);
+ case JournalRecordIds.ADD_LARGE_MESSAGE: {
+ LargeServerMessage largeMessage = parseLargeMessage(buff);
- messages.put(record.id, largeMessage);
+ messages.put(record.id, largeMessage);
- largeMessages.add(largeMessage);
+ largeMessages.add(largeMessage);
- break;
- }
- case JournalRecordIds.ADD_MESSAGE: {
- throw new IllegalStateException("This is using old journal data, export your data and import at the correct version");
- }
+ break;
+ }
+ case JournalRecordIds.ADD_MESSAGE: {
+ throw new IllegalStateException("This is using old journal data, export your data and import at the correct version");
+ }
- case JournalRecordIds.ADD_MESSAGE_PROTOCOL: {
+ case JournalRecordIds.ADD_MESSAGE_PROTOCOL: {
- Message message = MessagePersister.getInstance().decode(buff, null);
+ Message message = MessagePersister.getInstance().decode(buff, null);
- messages.put(record.id, message);
+ messages.put(record.id, message);
- break;
- }
- case JournalRecordIds.ADD_REF: {
- long messageID = record.id;
+ break;
+ }
+ case JournalRecordIds.ADD_REF: {
+ long messageID = record.id;
- RefEncoding encoding = new RefEncoding();
+ RefEncoding encoding = new RefEncoding();
- encoding.decode(buff);
+ encoding.decode(buff);
- Map<Long, AddMessageRecord> queueMessages = queueMap.get(encoding.queueID);
+ Map<Long, AddMessageRecord> queueMessages = queueMap.get(encoding.queueID);
- if (queueMessages == null) {
- queueMessages = new LinkedHashMap<>();
+ if (queueMessages == null) {
+ queueMessages = new LinkedHashMap<>();
- queueMap.put(encoding.queueID, queueMessages);
- }
+ queueMap.put(encoding.queueID, queueMessages);
+ }
- Message message = messages.get(messageID);
+ Message message = messages.get(messageID);
- if (message == null) {
- ActiveMQServerLogger.LOGGER.cannotFindMessage(record.id);
- } else {
- queueMessages.put(messageID, new AddMessageRecord(message));
- }
+ if (message == null) {
+ ActiveMQServerLogger.LOGGER.cannotFindMessage(record.id);
+ } else {
+ queueMessages.put(messageID, new AddMessageRecord(message));
+ }
- break;
- }
- case JournalRecordIds.ACKNOWLEDGE_REF: {
- long messageID = record.id;
+ break;
+ }
+ case JournalRecordIds.ACKNOWLEDGE_REF: {
+ long messageID = record.id;
- RefEncoding encoding = new RefEncoding();
+ RefEncoding encoding = new RefEncoding();
- encoding.decode(buff);
+ encoding.decode(buff);
- Map<Long, AddMessageRecord> queueMessages = queueMap.get(encoding.queueID);
+ Map<Long, AddMessageRecord> queueMessages = queueMap.get(encoding.queueID);
- if (queueMessages == null) {
- ActiveMQServerLogger.LOGGER.journalCannotFindQueue(encoding.queueID, messageID);
- } else {
- AddMessageRecord rec = queueMessages.remove(messageID);
+ if (queueMessages == null) {
+ ActiveMQServerLogger.LOGGER.journalCannotFindQueue(encoding.queueID, messageID);
+ } else {
+ AddMessageRecord rec = queueMessages.remove(messageID);
- if (rec == null) {
- ActiveMQServerLogger.LOGGER.cannotFindMessage(messageID);
+ if (rec == null) {
+ ActiveMQServerLogger.LOGGER.cannotFindMessage(messageID);
+ }
}
- }
-
- break;
- }
- case JournalRecordIds.UPDATE_DELIVERY_COUNT: {
- long messageID = record.id;
- DeliveryCountUpdateEncoding encoding = new DeliveryCountUpdateEncoding();
+ break;
+ }
+ case JournalRecordIds.UPDATE_DELIVERY_COUNT: {
+ long messageID = record.id;
- encoding.decode(buff);
+ DeliveryCountUpdateEncoding encoding = new DeliveryCountUpdateEncoding();
- Map<Long, AddMessageRecord> queueMessages = queueMap.get(encoding.queueID);
+ encoding.decode(buff);
- if (queueMessages == null) {
- ActiveMQServerLogger.LOGGER.journalCannotFindQueueDelCount(encoding.queueID);
- } else {
- AddMessageRecord rec = queueMessages.get(messageID);
+ Map<Long, AddMessageRecord> queueMessages = queueMap.get(encoding.queueID);
- if (rec == null) {
- ActiveMQServerLogger.LOGGER.journalCannotFindMessageDelCount(messageID);
+ if (queueMessages == null) {
+ ActiveMQServerLogger.LOGGER.journalCannotFindQueueDelCount(encoding.queueID);
} else {
- rec.setDeliveryCount(encoding.count);
+ AddMessageRecord rec = queueMessages.get(messageID);
+
+ if (rec == null) {
+ ActiveMQServerLogger.LOGGER.journalCannotFindMessageDelCount(messageID);
+ } else {
+ rec.setDeliveryCount(encoding.count);
+ }
}
- }
- break;
- }
- case JournalRecordIds.PAGE_TRANSACTION: {
- PageTransactionInfo invalidPGTx = null;
- if (record.isUpdate) {
- PageUpdateTXEncoding pageUpdate = new PageUpdateTXEncoding();
+ break;
+ }
+ case JournalRecordIds.PAGE_TRANSACTION: {
+ PageTransactionInfo invalidPGTx = null;
+ if (record.isUpdate) {
+ PageUpdateTXEncoding pageUpdate = new PageUpdateTXEncoding();
- pageUpdate.decode(buff);
+ pageUpdate.decode(buff);
- PageTransactionInfo pageTX = pagingManager.getTransaction(pageUpdate.pageTX);
+ PageTransactionInfo pageTX = pagingManager.getTransaction(pageUpdate.pageTX);
- if (pageTX == null) {
- ActiveMQServerLogger.LOGGER.journalCannotFindPageTX(pageUpdate.pageTX);
- } else {
- if (!pageTX.onUpdate(pageUpdate.recods, null, null)) {
- invalidPGTx = pageTX;
+ if (pageTX == null) {
+ ActiveMQServerLogger.LOGGER.journalCannotFindPageTX(pageUpdate.pageTX);
+ } else {
+ if (!pageTX.onUpdate(pageUpdate.recods, null, null)) {
+ invalidPGTx = pageTX;
+ }
}
- }
- } else {
- PageTransactionInfoImpl pageTransactionInfo = new PageTransactionInfoImpl();
+ } else {
+ PageTransactionInfoImpl pageTransactionInfo = new PageTransactionInfoImpl();
- pageTransactionInfo.decode(buff);
+ pageTransactionInfo.decode(buff);
- pageTransactionInfo.setRecordID(record.id);
+ pageTransactionInfo.setRecordID(record.id);
- pagingManager.addTransaction(pageTransactionInfo);
+ pagingManager.addTransaction(pageTransactionInfo);
- if (!pageTransactionInfo.checkSize(null, null)) {
- invalidPGTx = pageTransactionInfo;
+ if (!pageTransactionInfo.checkSize(null, null)) {
+ invalidPGTx = pageTransactionInfo;
+ }
}
- }
- if (invalidPGTx != null) {
- if (invalidPageTransactions == null) {
- invalidPageTransactions = new HashSet<>();
+ if (invalidPGTx != null) {
+ invalidPageTransactions.add(invalidPGTx);
}
- invalidPageTransactions.add(invalidPGTx);
- }
- break;
- }
- case JournalRecordIds.SET_SCHEDULED_DELIVERY_TIME: {
- long messageID = record.id;
+ break;
+ }
+ case JournalRecordIds.SET_SCHEDULED_DELIVERY_TIME: {
+ long messageID = record.id;
- ScheduledDeliveryEncoding encoding = new ScheduledDeliveryEncoding();
+ ScheduledDeliveryEncoding encoding = new ScheduledDeliveryEncoding();
- encoding.decode(buff);
+ encoding.decode(buff);
- Map<Long, AddMessageRecord> queueMessages = queueMap.get(encoding.queueID);
+ Map<Long, AddMessageRecord> queueMessages = queueMap.get(encoding.queueID);
- if (queueMessages == null) {
- ActiveMQServerLogger.LOGGER.journalCannotFindQueueScheduled(encoding.queueID, messageID);
- } else {
+ if (queueMessages == null) {
+ ActiveMQServerLogger.LOGGER.journalCannotFindQueueScheduled(encoding.queueID, messageID);
+ } else {
- AddMessageRecord rec = queueMessages.get(messageID);
+ AddMessageRecord rec = queueMessages.get(messageID);
- if (rec == null) {
- ActiveMQServerLogger.LOGGER.cannotFindMessage(messageID);
- } else {
- rec.setScheduledDeliveryTime(encoding.scheduledDeliveryTime);
+ if (rec == null) {
+ ActiveMQServerLogger.LOGGER.cannotFindMessage(messageID);
+ } else {
+ rec.setScheduledDeliveryTime(encoding.scheduledDeliveryTime);
+ }
}
+
+ break;
}
+ case JournalRecordIds.DUPLICATE_ID: {
+ DuplicateIDEncoding encoding = new DuplicateIDEncoding();
- break;
- }
- case JournalRecordIds.DUPLICATE_ID: {
- DuplicateIDEncoding encoding = new DuplicateIDEncoding();
+ encoding.decode(buff);
- encoding.decode(buff);
+ List<Pair<byte[], Long>> ids = duplicateIDMap.get(encoding.address);
- List<Pair<byte[], Long>> ids = duplicateIDMap.get(encoding.address);
+ if (ids == null) {
+ ids = new ArrayList<>();
- if (ids == null) {
- ids = new ArrayList<>();
+ duplicateIDMap.put(encoding.address, ids);
+ }
- duplicateIDMap.put(encoding.address, ids);
+ ids.add(new Pair<>(encoding.duplID, record.id));
+
+ break;
}
+ case JournalRecordIds.HEURISTIC_COMPLETION: {
+ HeuristicCompletionEncoding encoding = new HeuristicCompletionEncoding();
+ encoding.decode(buff);
+ resourceManager.putHeuristicCompletion(record.id, encoding.xid, encoding.isCommit);
+ break;
+ }
+ case JournalRecordIds.ACKNOWLEDGE_CURSOR: {
+ CursorAckRecordEncoding encoding = new CursorAckRecordEncoding();
+ encoding.decode(buff);
- ids.add(new Pair<>(encoding.duplID, record.id));
+ encoding.position.setRecordID(record.id);
- break;
- }
- case JournalRecordIds.HEURISTIC_COMPLETION: {
- HeuristicCompletionEncoding encoding = new HeuristicCompletionEncoding();
- encoding.decode(buff);
- resourceManager.putHeuristicCompletion(record.id, encoding.xid, encoding.isCommit);
- break;
- }
- case JournalRecordIds.ACKNOWLEDGE_CURSOR: {
- CursorAckRecordEncoding encoding = new CursorAckRecordEncoding();
- encoding.decode(buff);
+ PageSubscription sub = locateSubscription(encoding.queueID, pageSubscriptions, queueInfos, pagingManager);
- encoding.position.setRecordID(record.id);
-
- PageSubscription sub = locateSubscription(encoding.queueID, pageSubscriptions, queueInfos, pagingManager);
+ if (sub != null) {
+ sub.reloadACK(encoding.position);
+ } else {
+ ActiveMQServerLogger.LOGGER.journalCannotFindQueueReloading(encoding.queueID);
+ messageJournal.appendDeleteRecord(record.id, false);
- if (sub != null) {
- sub.reloadACK(encoding.position);
- } else {
- ActiveMQServerLogger.LOGGER.journalCannotFindQueueReloading(encoding.queueID);
- messageJournal.appendDeleteRecord(record.id, false);
+ }
+ break;
}
+ case JournalRecordIds.PAGE_CURSOR_COUNTER_VALUE: {
+ PageCountRecord encoding = new PageCountRecord();
- break;
- }
- case JournalRecordIds.PAGE_CURSOR_COUNTER_VALUE: {
- PageCountRecord encoding = new PageCountRecord();
+ encoding.decode(buff);
- encoding.decode(buff);
+ PageSubscription sub = locateSubscription(encoding.getQueueID(), pageSubscriptions, queueInfos, pagingManager);
- PageSubscription sub = locateSubscription(encoding.getQueueID(), pageSubscriptions, queueInfos, pagingManager);
+ if (sub != null) {
+ sub.getCounter().loadValue(record.id, encoding.getValue(), encoding.getPersistentSize());
+ } else {
+ ActiveMQServerLogger.LOGGER.journalCannotFindQueueReloadingPage(encoding.getQueueID());
+ messageJournal.appendDeleteRecord(record.id, false);
+ }
- if (sub != null) {
- sub.getCounter().loadValue(record.id, encoding.getValue(), encoding.getPersistentSize());
- } else {
- ActiveMQServerLogger.LOGGER.journalCannotFindQueueReloadingPage(encoding.getQueueID());
- messageJournal.appendDeleteRecord(record.id, false);
+ break;
}
- break;
- }
+ case JournalRecordIds.PAGE_CURSOR_COUNTER_INC: {
+ PageCountRecordInc encoding = new PageCountRecordInc();
- case JournalRecordIds.PAGE_CURSOR_COUNTER_INC: {
- PageCountRecordInc encoding = new PageCountRecordInc();
+ encoding.decode(buff);
- encoding.decode(buff);
+ PageSubscription sub = locateSubscription(encoding.getQueueID(), pageSubscriptions, queueInfos, pagingManager);
- PageSubscription sub = locateSubscription(encoding.getQueueID(), pageSubscriptions, queueInfos, pagingManager);
+ if (sub != null) {
+ sub.getCounter().loadInc(record.id, encoding.getValue(), encoding.getPersistentSize());
+ } else {
+ ActiveMQServerLogger.LOGGER.journalCannotFindQueueReloadingPageCursor(encoding.getQueueID());
+ messageJournal.appendDeleteRecord(record.id, false);
+ }
- if (sub != null) {
- sub.getCounter().loadInc(record.id, encoding.getValue(), encoding.getPersistentSize());
- } else {
- ActiveMQServerLogger.LOGGER.journalCannotFindQueueReloadingPageCursor(encoding.getQueueID());
- messageJournal.appendDeleteRecord(record.id, false);
+ break;
}
- break;
- }
+ case JournalRecordIds.PAGE_CURSOR_COMPLETE: {
+ CursorAckRecordEncoding encoding = new CursorAckRecordEncoding();
+ encoding.decode(buff);
- case JournalRecordIds.PAGE_CURSOR_COMPLETE: {
- CursorAckRecordEncoding encoding = new CursorAckRecordEncoding();
- encoding.decode(buff);
+ encoding.position.setRecordID(record.id);
- encoding.position.setRecordID(record.id);
-
- PageSubscription sub = locateSubscription(encoding.queueID, pageSubscriptions, queueInfos, pagingManager);
+ PageSubscription sub = locateSubscription(encoding.queueID, pageSubscriptions, queueInfos, pagingManager);
- if (sub != null) {
- if (!sub.reloadPageCompletion(encoding.position)) {
- if (logger.isDebugEnabled()) {
- logger.debug("Complete page " + encoding.position.getPageNr() + " doesn't exist on page manager " + sub.getPagingStore().getAddress());
+ if (sub != null) {
+ if (!sub.reloadPageCompletion(encoding.position)) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Complete page " + encoding.position.getPageNr() + " doesn't exist on page manager " + sub.getPagingStore().getAddress());
+ }
+ messageJournal.appendDeleteRecord(record.id, false);
}
+ } else {
+ ActiveMQServerLogger.LOGGER.cantFindQueueOnPageComplete(encoding.queueID);
messageJournal.appendDeleteRecord(record.id, false);
}
- } else {
- ActiveMQServerLogger.LOGGER.cantFindQueueOnPageComplete(encoding.queueID);
- messageJournal.appendDeleteRecord(record.id, false);
+
+ break;
}
- break;
- }
+ case JournalRecordIds.PAGE_CURSOR_PENDING_COUNTER: {
- case JournalRecordIds.PAGE_CURSOR_PENDING_COUNTER: {
+ PageCountPendingImpl pendingCountEncoding = new PageCountPendingImpl();
- PageCountPendingImpl pendingCountEncoding = new PageCountPendingImpl();
+ pendingCountEncoding.decode(buff);
+ pendingCountEncoding.setID(record.id);
+ PageSubscription sub = locateSubscription(pendingCountEncoding.getQueueID(), pageSubscriptions, queueInfos, pagingManager);
+ if (sub != null) {
+ sub.notEmpty();
+ }
+ // This can be null on testcases not interested on this outcome
+ if (pendingNonTXPageCounter != null) {
+ pendingNonTXPageCounter.add(pendingCountEncoding);
+ }
- pendingCountEncoding.decode(buff);
- pendingCountEncoding.setID(record.id);
- PageSubscription sub = locateSubscription(pendingCountEncoding.getQueueID(), pageSubscriptions, queueInfos, pagingManager);
- if (sub != null) {
- sub.notEmpty();
- }
- // This can be null on testcases not interested on this outcome
- if (pendingNonTXPageCounter != null) {
- pendingNonTXPageCounter.add(pendingCountEncoding);
+ break;
}
- break;
- }
-
- default: {
- throw new IllegalStateException("Invalid record type " + recordType);
+ default: {
+ throw new IllegalStateException("Invalid record type " + recordType);
+ }
}
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
}
-
- // This will free up memory sooner. The record is not needed any more
- // and its byte array would consume memory during the load process even though it's not necessary any longer
- // what would delay processing time during load
- records.set(reccount, null);
- }
+ });
// Release the memory as soon as not needed any longer
- records.clear();
records = null;
journalLoader.handleAddMessage(queueMap);
@@ -1221,7 +1202,7 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
public void checkInvalidPageTransactions(PagingManager pagingManager,
Set<PageTransactionInfo> invalidPageTransactions) {
- if (invalidPageTransactions != null) {
+ if (invalidPageTransactions != null && !invalidPageTransactions.isEmpty()) {
for (PageTransactionInfo pginfo : invalidPageTransactions) {
pginfo.checkSize(this, pagingManager);
}
@@ -1713,8 +1694,7 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
}
}
- protected abstract LargeServerMessage parseLargeMessage(Map<Long, Message> messages,
- ActiveMQBuffer buff) throws Exception;
+ protected abstract LargeServerMessage parseLargeMessage(ActiveMQBuffer buff) throws Exception;
private void loadPreparedTransactions(final PostOffice postOffice,
final PagingManager pagingManager,
@@ -1750,7 +1730,7 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
switch (recordType) {
case JournalRecordIds.ADD_LARGE_MESSAGE: {
- messages.put(record.id, parseLargeMessage(messages, buff));
+ messages.put(record.id, parseLargeMessage(buff));
break;
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManager.java
index e5da1d9..e200ead 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManager.java
@@ -358,12 +358,11 @@ public class JournalStorageManager extends AbstractJournalStorageManager {
@Override
/**
- * @param messages
* @param buff
* @return
* @throws Exception
- */ protected LargeServerMessage parseLargeMessage(final Map<Long, Message> messages,
- final ActiveMQBuffer buff) throws Exception {
+ */
+ protected LargeServerMessage parseLargeMessage(final ActiveMQBuffer buff) throws Exception {
LargeServerMessage largeMessage = createLargeMessage();
LargeMessagePersister.getInstance().decode(buff, largeMessage);
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/LargeMessageTXFailureCallback.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/LargeMessageTXFailureCallback.java
index 33be342..0a92bd2 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/LargeMessageTXFailureCallback.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/LargeMessageTXFailureCallback.java
@@ -17,11 +17,9 @@
package org.apache.activemq.artemis.core.persistence.impl.journal;
import java.util.List;
-import java.util.Map;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
-import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.core.journal.RecordInfo;
import org.apache.activemq.artemis.core.journal.TransactionFailureCallback;
import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
@@ -32,13 +30,10 @@ import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalR
public class LargeMessageTXFailureCallback implements TransactionFailureCallback {
private AbstractJournalStorageManager journalStorageManager;
- private final Map<Long, Message> messages;
- public LargeMessageTXFailureCallback(AbstractJournalStorageManager journalStorageManager,
- final Map<Long, Message> messages) {
+ public LargeMessageTXFailureCallback(AbstractJournalStorageManager journalStorageManager) {
super();
this.journalStorageManager = journalStorageManager;
- this.messages = messages;
}
@Override
@@ -52,7 +47,7 @@ public class LargeMessageTXFailureCallback implements TransactionFailureCallback
ActiveMQBuffer buff = ActiveMQBuffers.wrappedBuffer(data);
try {
- LargeServerMessage serverMessage = journalStorageManager.parseLargeMessage(messages, buff);
+ LargeServerMessage serverMessage = journalStorageManager.parseLargeMessage(buff);
serverMessage.decrementDelayDeletionCount();
} catch (Exception e) {
ActiveMQServerLogger.LOGGER.journalError(e);
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicatedJournal.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicatedJournal.java
index 8a04cc1..b658218 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicatedJournal.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicatedJournal.java
@@ -18,6 +18,7 @@ package org.apache.activemq.artemis.core.replication;
import java.util.List;
import java.util.Map;
+import java.util.function.Consumer;
import org.apache.activemq.artemis.core.io.SequentialFileFactory;
import org.apache.activemq.artemis.core.journal.EncodingSupport;
@@ -424,7 +425,7 @@ public class ReplicatedJournal implements Journal {
* @see org.apache.activemq.artemis.core.journal.Journal#load(java.util.List, java.util.List, org.apache.activemq.artemis.core.journal.TransactionFailureCallback)
*/
@Override
- public JournalLoadInformation load(final List<RecordInfo> committedRecords,
+ public JournalLoadInformation load(final Consumer<RecordInfo> committedRecords,
final List<PreparedTransactionInfo> preparedTransactions,
final TransactionFailureCallback transactionFailure,
final boolean fixbadTX) throws Exception {
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/ReplicationTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/ReplicationTest.java
index 1c89425..4628403 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/ReplicationTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/ReplicationTest.java
@@ -29,6 +29,7 @@ import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
@@ -811,7 +812,7 @@ public final class ReplicationTest extends ActiveMQTestBase {
}
@Override
- public JournalLoadInformation load(final List<RecordInfo> committedRecords,
+ public JournalLoadInformation load(final Consumer<RecordInfo> committedRecords,
final List<PreparedTransactionInfo> preparedTransactions,
final TransactionFailureCallback transactionFailure,
final boolean fixbadtx) throws Exception {
diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/journal/impl/JournalImplTestBase.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/journal/impl/JournalImplTestBase.java
index 07a3800..e8a2012 100644
--- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/journal/impl/JournalImplTestBase.java
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/journal/impl/JournalImplTestBase.java
@@ -360,7 +360,7 @@ public abstract class JournalImplTestBase extends ActiveMQTestBase {
}
protected void load() throws Exception {
- journal.load(null, null, null);
+ journal.load((List<RecordInfo>) null, null, null);
}
protected void beforeJournalOperation() throws Exception {
[activemq-artemis] 03/03: This closes #2944
Posted by cl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit 3b6176438ebf9f206028bc596af665a4be8ae83f
Merge: 2f68f00 b10d765
Author: Clebert Suconic <cl...@apache.org>
AuthorDate: Fri Jan 17 14:31:52 2020 -0500
This closes #2944
artemis-commons/pom.xml | 6 +
.../artemis/utils/collections/LongHashSet.java | 622 ++++++++++++++++
.../utils/collections/SparseArrayLinkedList.java | 243 +++++++
.../artemis/utils/collections/LongHashSetTest.java | 778 +++++++++++++++++++++
.../collections/SparseArrayLinkedListTest.java | 158 +++++
.../jdbc/store/journal/JDBCJournalImpl.java | 12 +
.../activemq/artemis/core/journal/Journal.java | 20 +-
.../core/journal/impl/FileWrapperJournal.java | 3 +-
.../artemis/core/journal/impl/JournalImpl.java | 40 +-
.../journal/AbstractJournalStorageManager.java | 466 ++++++------
.../impl/journal/JournalStorageManager.java | 5 +-
.../journal/LargeMessageTXFailureCallback.java | 9 +-
.../core/replication/ReplicatedJournal.java | 3 +-
.../integration/replication/ReplicationTest.java | 3 +-
.../core/journal/impl/JournalImplTestBase.java | 2 +-
15 files changed, 2090 insertions(+), 280 deletions(-)
[activemq-artemis] 01/03: ARTEMIS-2602 Improve Journal loading heap
usage
Posted by cl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit 4cc6464ddd0f6e51eb4af129ec1dc9bfa4480041
Author: Francesco Nigro <ni...@gmail.com>
AuthorDate: Thu Jan 16 14:04:40 2020 +0100
ARTEMIS-2602 Improve Journal loading heap usage
---
artemis-commons/pom.xml | 6 +
.../artemis/utils/collections/LongHashSet.java | 622 ++++++++++++++++
.../utils/collections/SparseArrayLinkedList.java | 243 +++++++
.../artemis/utils/collections/LongHashSetTest.java | 778 +++++++++++++++++++++
.../collections/SparseArrayLinkedListTest.java | 158 +++++
.../artemis/core/journal/impl/JournalImpl.java | 36 +-
6 files changed, 1825 insertions(+), 18 deletions(-)
diff --git a/artemis-commons/pom.xml b/artemis-commons/pom.xml
index c945d9b..3b86192 100644
--- a/artemis-commons/pom.xml
+++ b/artemis-commons/pom.xml
@@ -81,6 +81,12 @@
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.hamcrest</groupId>
+ <artifactId>hamcrest</artifactId>
+ <version>${hamcrest.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/collections/LongHashSet.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/collections/LongHashSet.java
new file mode 100644
index 0000000..a86a7f2
--- /dev/null
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/collections/LongHashSet.java
@@ -0,0 +1,622 @@
+/*
+ * 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.utils.collections;
+
+import static io.netty.util.internal.MathUtil.findNextPositivePowerOfTwo;
+
+import java.io.Serializable;
+import java.lang.reflect.Array;
+import java.util.AbstractSet;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.Set;
+
+/**
+ * A hash set implementation of {@link Set<Long>} that uses open addressing values.
+ * To minimize the memory footprint, this class uses open addressing rather than chaining.
+ * Collisions are resolved using linear probing. Deletions implement compaction, so cost of
+ * remove can approach O(N) for full maps, which makes a small loadFactor recommended.
+ *
+ * The implementation is based on <a href=https://github.com/real-logic/agrona/blob/master/agrona/src/main/java/org/agrona/collections/IntHashSet.java>Agrona IntHashSet</a>
+ * but uses long primitive keys and a different {@link #MISSING_VALUE} to account for {@link Long#hashCode} being 0 for -1.
+ */
+public class LongHashSet extends AbstractSet<Long> implements Serializable {
+
+ /**
+ * The initial capacity used when none is specified in the constructor.
+ */
+ public static final int DEFAULT_INITIAL_CAPACITY = 8;
+ public static final float DEFAULT_LOAD_FACTOR = 0.5f;
+ static final long MISSING_VALUE = -2;
+
+ private boolean containsMissingValue;
+ private final float loadFactor;
+ private int resizeThreshold;
+ // NB: excludes missing value
+ private int sizeOfArrayValues;
+
+ private long[] values;
+
+ /**
+ * Construct a hash set with {@link #DEFAULT_INITIAL_CAPACITY} and {@link #DEFAULT_LOAD_FACTOR}.
+ */
+ public LongHashSet() {
+ this(DEFAULT_INITIAL_CAPACITY);
+ }
+
+ /**
+ * Construct a hash set with a proposed capacity and {@link #DEFAULT_LOAD_FACTOR}.
+ *
+ * @param proposedCapacity for the initial capacity of the set.
+ */
+ public LongHashSet(final int proposedCapacity) {
+ this(proposedCapacity, DEFAULT_LOAD_FACTOR);
+ }
+
+ private static int hashIndex(long value, int mask) {
+ return hashCode(value) & mask;
+ }
+
+ private static int hashCode(long value) {
+ long hash = value * 31;
+ hash = (int) hash ^ (int) (hash >>> 32);
+ return (int) hash;
+ }
+
+ /**
+ * Construct a hash set with a proposed initial capacity and load factor.
+ *
+ * @param proposedCapacity for the initial capacity of the set.
+ * @param loadFactor to be used for resizing.
+ */
+ public LongHashSet(final int proposedCapacity, final float loadFactor) {
+ if (loadFactor < 0.1f || loadFactor > 0.9f) {
+ throw new IllegalArgumentException("load factor must be in the range of 0.1 to 0.9: " + loadFactor);
+ }
+ this.loadFactor = loadFactor;
+ sizeOfArrayValues = 0;
+ final int capacity = findNextPositivePowerOfTwo(Math.max(DEFAULT_INITIAL_CAPACITY, proposedCapacity));
+ resizeThreshold = (int) (capacity * loadFactor);
+ values = new long[capacity];
+ Arrays.fill(values, MISSING_VALUE);
+ }
+
+ /**
+ * Get the load factor beyond which the set will increase size.
+ *
+ * @return load factor for when the set should increase size.
+ */
+ public float loadFactor() {
+ return loadFactor;
+ }
+
+ /**
+ * Get the total capacity for the set to which the load factor with be a fraction of.
+ *
+ * @return the total capacity for the set.
+ */
+ public int capacity() {
+ return values.length;
+ }
+
+ /**
+ * Get the actual threshold which when reached the map will resize.
+ * This is a function of the current capacity and load factor.
+ *
+ * @return the threshold when the map will resize.
+ */
+ public int resizeThreshold() {
+ return resizeThreshold;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean add(final Long value) {
+ return add(value.longValue());
+ }
+
+ /**
+ * Primitive specialised overload of {this#add(Long)}
+ *
+ * @param value the value to add
+ * @return true if the collection has changed, false otherwise
+ * @throws IllegalArgumentException if value is missingValue
+ */
+ public boolean add(final long value) {
+ if (value == MISSING_VALUE) {
+ final boolean previousContainsMissingValue = this.containsMissingValue;
+ containsMissingValue = true;
+ return !previousContainsMissingValue;
+ }
+
+ final long[] values = this.values;
+ final int mask = values.length - 1;
+ int index = hashIndex(value, mask);
+
+ while (values[index] != MISSING_VALUE) {
+ if (values[index] == value) {
+ return false;
+ }
+
+ index = next(index, mask);
+ }
+
+ values[index] = value;
+ sizeOfArrayValues++;
+
+ if (sizeOfArrayValues > resizeThreshold) {
+ increaseCapacity();
+ }
+
+ return true;
+ }
+
+ private void increaseCapacity() {
+ final int newCapacity = values.length * 2;
+ if (newCapacity < 0) {
+ throw new IllegalStateException("max capacity reached at size=" + size());
+ }
+
+ rehash(newCapacity);
+ }
+
+ private void rehash(final int newCapacity) {
+ final int capacity = newCapacity;
+ final int mask = newCapacity - 1;
+ resizeThreshold = (int) (newCapacity * loadFactor);
+
+ final long[] tempValues = new long[capacity];
+ Arrays.fill(tempValues, MISSING_VALUE);
+
+ for (final long value : values) {
+ if (value != MISSING_VALUE) {
+ int newHash = hashIndex(value, mask);
+ while (tempValues[newHash] != MISSING_VALUE) {
+ newHash = ++newHash & mask;
+ }
+
+ tempValues[newHash] = value;
+ }
+ }
+ values = tempValues;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean remove(final Object value) {
+ return value instanceof Long && remove(((Long) value).longValue());
+ }
+
+ /**
+ * An int specialised version of {this#remove(Object)}.
+ *
+ * @param value the value to remove
+ * @return true if the value was present, false otherwise
+ */
+ public boolean remove(final long value) {
+ if (value == MISSING_VALUE) {
+ final boolean previousContainsMissingValue = this.containsMissingValue;
+ containsMissingValue = false;
+ return previousContainsMissingValue;
+ }
+
+ final long[] values = this.values;
+ final int mask = values.length - 1;
+ int index = hashIndex(value, mask);
+
+ while (values[index] != MISSING_VALUE) {
+ if (values[index] == value) {
+ values[index] = MISSING_VALUE;
+ compactChain(index);
+ sizeOfArrayValues--;
+ return true;
+ }
+
+ index = next(index, mask);
+ }
+
+ return false;
+ }
+
+ private static int next(final int index, final int mask) {
+ return (index + 1) & mask;
+ }
+
+ @SuppressWarnings("FinalParameters")
+ private void compactChain(int deleteIndex) {
+ final long[] values = this.values;
+ final int mask = values.length - 1;
+
+ int index = deleteIndex;
+ while (true) {
+ index = next(index, mask);
+ if (values[index] == MISSING_VALUE) {
+ return;
+ }
+
+ final int hash = hashIndex(values[index], mask);
+
+ if ((index < hash && (hash <= deleteIndex || deleteIndex <= index)) || (hash <= deleteIndex && deleteIndex <= index)) {
+ values[deleteIndex] = values[index];
+
+ values[index] = MISSING_VALUE;
+ deleteIndex = index;
+ }
+ }
+ }
+
+ /**
+ * Compact the backing arrays by rehashing with a capacity just larger than current size
+ * and giving consideration to the load factor.
+ */
+ public void compact() {
+ final int idealCapacity = (int) Math.round(size() * (1.0 / loadFactor));
+ rehash(findNextPositivePowerOfTwo(Math.max(DEFAULT_INITIAL_CAPACITY, idealCapacity)));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean contains(final Object value) {
+ return value instanceof Long && contains(((Long) value).longValue());
+ }
+
+ /**
+ * Contains method that does not box values.
+ *
+ * @param value to be check for if the set contains it.
+ * @return true if the value is contained in the set otherwise false.
+ * @see Collection#contains(Object)
+ */
+ public boolean contains(final long value) {
+ if (value == MISSING_VALUE) {
+ return containsMissingValue;
+ }
+
+ final long[] values = this.values;
+ final int mask = values.length - 1;
+ int index = hashIndex(value, mask);
+
+ while (values[index] != MISSING_VALUE) {
+ if (values[index] == value) {
+ return true;
+ }
+
+ index = next(index, mask);
+ }
+
+ return false;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public int size() {
+ return sizeOfArrayValues + (containsMissingValue ? 1 : 0);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean isEmpty() {
+ return size() == 0;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void clear() {
+ if (size() > 0) {
+ Arrays.fill(values, MISSING_VALUE);
+ sizeOfArrayValues = 0;
+ containsMissingValue = false;
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean addAll(final Collection<? extends Long> coll) {
+ boolean added = false;
+
+ for (final Long value : coll) {
+ added |= add(value);
+ }
+
+ return added;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean removeAll(final Collection<?> coll) {
+ boolean removed = false;
+
+ for (final Object value : coll) {
+ removed |= remove(value);
+ }
+
+ return removed;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public LongIterator iterator() {
+ return new LongIterator().reset();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public String toString() {
+ final StringBuilder sb = new StringBuilder();
+ sb.append('{');
+
+ for (final long value : values) {
+ if (value != MISSING_VALUE) {
+ sb.append(value).append(", ");
+ }
+ }
+
+ if (containsMissingValue) {
+ sb.append(MISSING_VALUE).append(", ");
+ }
+
+ if (sb.length() > 1) {
+ sb.setLength(sb.length() - 2);
+ }
+
+ sb.append('}');
+
+ return sb.toString();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @SuppressWarnings("unchecked")
+ @Override
+ public <T> T[] toArray(final T[] a) {
+ final Class<?> componentType = a.getClass().getComponentType();
+ if (!componentType.isAssignableFrom(Long.class)) {
+ throw new ArrayStoreException("cannot store Longs in array of type " + componentType);
+ }
+
+ final int size = size();
+ final T[] arrayCopy = a.length >= size ? a : (T[]) Array.newInstance(componentType, size);
+ copyValues(arrayCopy);
+
+ return arrayCopy;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Object[] toArray() {
+ final Object[] arrayCopy = new Object[size()];
+ copyValues(arrayCopy);
+
+ return arrayCopy;
+ }
+
+ private void copyValues(final Object[] arrayCopy) {
+ int i = 0;
+ final long[] values = this.values;
+ for (final long value : values) {
+ if (MISSING_VALUE != value) {
+ arrayCopy[i++] = value;
+ }
+ }
+
+ if (containsMissingValue) {
+ arrayCopy[sizeOfArrayValues] = MISSING_VALUE;
+ }
+ }
+
+ /**
+ * LongHashSet specialised variant of {this#containsAll(Collection)}.
+ *
+ * @param other int hash set to compare against.
+ * @return true if every element in other is in this.
+ */
+ public boolean containsAll(final LongHashSet other) {
+ for (final long value : other.values) {
+ if (value != MISSING_VALUE && !contains(value)) {
+ return false;
+ }
+ }
+
+ return !other.containsMissingValue || this.containsMissingValue;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean equals(final Object other) {
+ if (other == this) {
+ return true;
+ }
+
+ if (other instanceof LongHashSet) {
+ final LongHashSet otherSet = (LongHashSet) other;
+
+ return otherSet.containsMissingValue == containsMissingValue && otherSet.sizeOfArrayValues == sizeOfArrayValues && containsAll(otherSet);
+ }
+
+ if (!(other instanceof Set)) {
+ return false;
+ }
+
+ final Set<?> c = (Set<?>) other;
+ if (c.size() != size()) {
+ return false;
+ }
+
+ try {
+ return containsAll(c);
+ } catch (final ClassCastException | NullPointerException ignore) {
+ return false;
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public int hashCode() {
+ int hashCode = 0;
+ for (final long value : values) {
+ if (value != MISSING_VALUE) {
+ hashCode += Long.hashCode(value);
+ }
+ }
+
+ if (containsMissingValue) {
+ // Account negative hashcode
+ final int code = Long.hashCode(MISSING_VALUE);
+ hashCode += code;
+ }
+
+ return hashCode;
+ }
+
+ /**
+ * Iterator which supports unboxed access to values.
+ */
+ public final class LongIterator implements Iterator<Long>, Serializable {
+
+ private int remaining;
+ private int positionCounter;
+ private int stopCounter;
+ private boolean isPositionValid = false;
+
+ LongIterator reset() {
+ remaining = size();
+
+ final long[] values = LongHashSet.this.values;
+ final int length = values.length;
+ int i = length;
+
+ if (values[length - 1] != LongHashSet.MISSING_VALUE) {
+ for (i = 0; i < length; i++) {
+ if (values[i] == LongHashSet.MISSING_VALUE) {
+ break;
+ }
+ }
+ }
+
+ stopCounter = i;
+ positionCounter = i + length;
+ isPositionValid = false;
+
+ return this;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return remaining > 0;
+ }
+
+ public int remaining() {
+ return remaining;
+ }
+
+ @Override
+ public Long next() {
+ return nextValue();
+ }
+
+ /**
+ * Strongly typed alternative of {@link Iterator#next()} to avoid boxing.
+ *
+ * @return the next int value.
+ */
+ public long nextValue() {
+ if (remaining == 1 && containsMissingValue) {
+ remaining = 0;
+ isPositionValid = true;
+
+ return LongHashSet.MISSING_VALUE;
+ }
+
+ findNext();
+
+ final long[] values = LongHashSet.this.values;
+
+ return values[position(values)];
+ }
+
+ @Override
+ public void remove() {
+ if (isPositionValid) {
+ if (0 == remaining && containsMissingValue) {
+ containsMissingValue = false;
+ } else {
+ final long[] values = LongHashSet.this.values;
+ final int position = position(values);
+ values[position] = MISSING_VALUE;
+ --sizeOfArrayValues;
+
+ compactChain(position);
+ }
+
+ isPositionValid = false;
+ } else {
+ throw new IllegalStateException();
+ }
+ }
+
+ private void findNext() {
+ final long[] values = LongHashSet.this.values;
+ final int mask = values.length - 1;
+ isPositionValid = true;
+
+ for (int i = positionCounter - 1; i >= stopCounter; i--) {
+ final int index = i & mask;
+ if (values[index] != LongHashSet.MISSING_VALUE) {
+ positionCounter = i;
+ --remaining;
+ return;
+ }
+ }
+
+ isPositionValid = false;
+ throw new NoSuchElementException();
+ }
+
+ private int position(final long[] values) {
+ return positionCounter & (values.length - 1);
+ }
+ }
+}
\ No newline at end of file
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/collections/SparseArrayLinkedList.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/collections/SparseArrayLinkedList.java
new file mode 100644
index 0000000..b797220
--- /dev/null
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/collections/SparseArrayLinkedList.java
@@ -0,0 +1,243 @@
+/*
+ * 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.utils.collections;
+
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+
+/**
+ * This list share the same motivation and structure of https://en.wikipedia.org/wiki/Unrolled_linked_list:
+ * it's a linked list of arrays/chunks of {@code T}.<br>
+ * Differently from an {@code UnrolledLinkedList} this list doesn't optimize addition and removal to achieve a balanced
+ * utilization among chunks ie a chunk is removed only if empty and chunks can't be merged.
+ * This list has been optimized for small-sized chunks (ideally <= 32 elements): this allow search/removal to
+ * be performed with a greedy approach despite a sparse chunk utilization (ie chunks contains few sparse elements)
+ */
+public final class SparseArrayLinkedList<T> {
+
+ // the whole chunk fit into 1 or 2 cache lines depending if JVM COOPS are used
+ private static final int SPARSE_ARRAY_DEFAULT_CAPACITY = 16;
+
+ private static final class SparseArray<T> {
+
+ private final Object[] elements;
+ private int size;
+ // index next to the last non null element
+ private int tail;
+
+ private SparseArray(int capacity) {
+ elements = new Object[capacity];
+ size = 0;
+ tail = 0;
+ }
+
+ private boolean add(T e) {
+ final int capacity = elements.length;
+ if (tail == capacity) {
+ return false;
+ }
+ elements[tail] = (T) e;
+ tail++;
+ size++;
+ return true;
+ }
+
+ private int remove(Predicate<? super T> filter) {
+ if (size == 0) {
+ // this shouldn't happen: the chunk should be removed if empty
+ return 0;
+ }
+ // this is allowed to make holes
+ // to save System::arrayCopy while removing elements
+ int removed = 0;
+ final Object[] elements = this.elements;
+ int visited = 0;
+ final int originalSize = size;
+ for (int i = 0, capacity = elements.length; i < capacity; i++) {
+ final T e = (T) elements[i];
+ if (e != null) {
+ if (filter.test(e)) {
+ elements[i] = null;
+ removed++;
+ } else {
+ // allows a weak form of compaction: incoming elements
+ // will be placed right after it
+ tail = i + 1;
+ }
+ visited++;
+ if (visited == originalSize) {
+ break;
+ }
+ }
+ }
+ size -= removed;
+ return removed;
+ }
+
+ public int clear(Consumer<? super T> consumer) {
+ final int originalSize = size;
+ if (originalSize == 0) {
+ return 0;
+ }
+ int visited = 0;
+ final Object[] elements = this.elements;
+ for (int i = 0, capacity = elements.length; i < capacity; i++) {
+ final T e = (T) elements[i];
+ if (e != null) {
+ if (consumer != null) {
+ consumer.accept(e);
+ }
+ elements[i] = null;
+ size--;
+ visited++;
+ if (visited == originalSize) {
+ break;
+ }
+ }
+ }
+ assert size == 0;
+ tail = 0;
+ return originalSize;
+ }
+
+ private int size() {
+ return size;
+ }
+ }
+
+ public static <T> long removeFromSparseArrayList(List<SparseArray<T>> sparseArrayList, Predicate<? super T> filter) {
+ if (filter == null) {
+ return 0;
+ }
+ long removed = 0;
+ Iterator<SparseArray<T>> iter = sparseArrayList.iterator();
+ while (iter.hasNext()) {
+ final SparseArray<T> sparseArray = iter.next();
+ final int justRemoved = sparseArray.remove(filter);
+ removed += justRemoved;
+ if (justRemoved > 0) {
+ // remove RecordInfo only if empty:
+ // it means that there is a chance of fragmentation
+ // proportional with the capacity of chunk
+ if (sparseArray.size() == 0) {
+ iter.remove();
+ }
+ }
+ }
+ return removed;
+ }
+
+ public static <T> void addToSparseArrayList(List<SparseArray<T>> sparseArrayList, T e, int sparseArrayCapacity) {
+ final int size = sparseArrayList.size();
+ // LinkedList::get(size-1) is fast as LinkedList::getLast
+ if (size == 0 || !sparseArrayList.get(size - 1).add(e)) {
+ final SparseArray<T> sparseArray = new SparseArray<>(sparseArrayCapacity);
+ sparseArray.add(e);
+ sparseArrayList.add(sparseArray);
+ }
+ }
+
+ public static <T> long clearSparseArrayList(List<SparseArray<T>> sparseArrayList, Consumer<? super T> consumer) {
+ final int size = sparseArrayList.size();
+ long count = 0;
+ if (size > 0) {
+ for (int i = 0; i < size; i++) {
+ // LinkedList::remove(0) is fast as LinkedList::getFirst
+ final SparseArray<T> removed = sparseArrayList.remove(0);
+ count += removed.clear(consumer);
+ }
+ }
+ return count;
+ }
+
+ private final LinkedList<SparseArray<T>> list;
+ private final int sparseArrayCapacity;
+ private long size;
+
+ public SparseArrayLinkedList() {
+ this(SPARSE_ARRAY_DEFAULT_CAPACITY);
+ }
+
+ public SparseArrayLinkedList(int sparseArrayCapacity) {
+ if (sparseArrayCapacity <= 0) {
+ throw new IllegalArgumentException("sparseArrayCapacity must be > 0");
+ }
+ list = new LinkedList<>();
+ size = 0;
+ this.sparseArrayCapacity = sparseArrayCapacity;
+ }
+
+ /**
+ * Appends {@code e} to the end of this list.
+ */
+ public void add(T e) {
+ Objects.requireNonNull(e, "e cannot be null");
+ addToSparseArrayList(list, e, sparseArrayCapacity);
+ size++;
+ }
+
+ /**
+ * Removes any element of the list matching the given predicate.
+ */
+ public long remove(Predicate<? super T> filter) {
+ if (size == 0) {
+ return 0;
+ }
+ final long removed = removeFromSparseArrayList(list, filter);
+ size -= removed;
+ assert size >= 0;
+ return removed;
+ }
+
+ /**
+ * Clear while consuming (using the given {@code consumer} all the elements of this list.
+ */
+ public long clear(Consumer<? super T> consumer) {
+ if (size == 0) {
+ return 0;
+ }
+ final long removed = clearSparseArrayList(list, consumer);
+ assert removed == size;
+ size = 0;
+ return removed;
+ }
+
+ /**
+ * Returns the number of elements of this list.
+ */
+ public long size() {
+ return size;
+ }
+
+ /**
+ * Returns the configured capacity of each sparse array/chunk.
+ */
+ public int sparseArrayCapacity() {
+ return sparseArrayCapacity;
+ }
+
+ /**
+ * Returns the number of sparse arrays/chunks of this list.
+ */
+ public int sparseArraysCount() {
+ return list.size();
+ }
+}
diff --git a/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/collections/LongHashSetTest.java b/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/collections/LongHashSetTest.java
new file mode 100644
index 0000000..09d18d0
--- /dev/null
+++ b/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/collections/LongHashSetTest.java
@@ -0,0 +1,778 @@
+/*
+ * 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.utils.collections;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.Random;
+import java.util.Set;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.arrayContainingInAnyOrder;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.core.IsEqual.equalTo;
+
+/**
+ * These tests are based on <a href="https://github.com/real-logic/agrona/blob/master/agrona/src/test/java/org/agrona/collections/IntHashSetTest.java">Agrona IntHashSetTest</a>
+ * to guarantee a similar coverage to what's provided for a similar collection.
+ */
+public class LongHashSetTest {
+
+ private static final int INITIAL_CAPACITY = 100;
+
+ private final LongHashSet testSet = new LongHashSet(INITIAL_CAPACITY);
+
+ @Test
+ public void initiallyContainsNoElements() {
+ for (long i = 0; i < 10_000; i++) {
+ Assert.assertFalse(testSet.contains(i));
+ }
+ }
+
+ @Test
+ public void initiallyContainsNoBoxedElements() {
+ for (long i = 0; i < 10_000; i++) {
+ Assert.assertFalse(testSet.contains(Long.valueOf(i)));
+ }
+ }
+
+ @Test
+ public void containsAddedElement() {
+ Assert.assertTrue(testSet.add(1L));
+
+ Assert.assertTrue(testSet.contains(1L));
+ }
+
+ @Test
+ public void addingAnElementTwiceDoesNothing() {
+ Assert.assertTrue(testSet.add(1L));
+
+ Assert.assertFalse(testSet.add(1L));
+ }
+
+ @Test
+ public void containsAddedBoxedElements() {
+ Assert.assertTrue(testSet.add(1L));
+ Assert.assertTrue(testSet.add(Long.valueOf(2L)));
+
+ Assert.assertTrue(testSet.contains(Long.valueOf(1L)));
+ Assert.assertTrue(testSet.contains(2L));
+ }
+
+ @Test
+ public void removingAnElementFromAnEmptyListDoesNothing() {
+ Assert.assertFalse(testSet.remove(0L));
+ }
+
+ @Test
+ public void removingAPresentElementRemovesIt() {
+ Assert.assertTrue(testSet.add(1L));
+
+ Assert.assertTrue(testSet.remove(1L));
+
+ Assert.assertFalse(testSet.contains(1L));
+ }
+
+ @Test
+ public void sizeIsInitiallyZero() {
+ Assert.assertEquals(0, testSet.size());
+ }
+
+ @Test
+ public void sizeIncrementsWithNumberOfAddedElements() {
+ addTwoElements(testSet);
+
+ Assert.assertEquals(2, testSet.size());
+ }
+
+ @Test
+ public void sizeContainsNumberOfNewElements() {
+ testSet.add(1L);
+ testSet.add(1L);
+
+ Assert.assertEquals(1, testSet.size());
+ }
+
+ @Test
+ public void iteratorsListElements() {
+ addTwoElements(testSet);
+
+ assertIteratorHasElements();
+ }
+
+ @Test
+ public void iteratorsStartFromTheBeginningEveryTime() {
+ iteratorsListElements();
+
+ assertIteratorHasElements();
+ }
+
+ @Test
+ public void iteratorsListElementsWithoutHasNext() {
+ addTwoElements(testSet);
+
+ assertIteratorHasElementsWithoutHasNext();
+ }
+
+ @Test
+ public void iteratorsStartFromTheBeginningEveryTimeWithoutHasNext() {
+ iteratorsListElementsWithoutHasNext();
+
+ assertIteratorHasElementsWithoutHasNext();
+ }
+
+ @Test(expected = NoSuchElementException.class)
+ public void iteratorsThrowNoSuchElementException() {
+ addTwoElements(testSet);
+
+ exhaustIterator();
+ }
+
+ @Test(expected = NoSuchElementException.class)
+ public void iteratorsThrowNoSuchElementExceptionFromTheBeginningEveryTime() {
+ addTwoElements(testSet);
+
+ try {
+ exhaustIterator();
+ } catch (final NoSuchElementException ignore) {
+ }
+
+ exhaustIterator();
+ }
+
+ @Test
+ public void iteratorHasNoElements() {
+ Assert.assertFalse(testSet.iterator().hasNext());
+ }
+
+ @Test(expected = NoSuchElementException.class)
+ public void iteratorThrowExceptionForEmptySet() {
+ testSet.iterator().next();
+ }
+
+ @Test
+ public void clearRemovesAllElementsOfTheSet() {
+ addTwoElements(testSet);
+
+ testSet.clear();
+
+ Assert.assertEquals(0, testSet.size());
+ Assert.assertFalse(testSet.contains(1L));
+ Assert.assertFalse(testSet.contains(1001L));
+ }
+
+ @Test
+ public void twoEmptySetsAreEqual() {
+ final LongHashSet other = new LongHashSet(100);
+ Assert.assertEquals(testSet, other);
+ }
+
+ @Test
+ public void setsWithTheSameValuesAreEqual() {
+ final LongHashSet other = new LongHashSet(100);
+
+ addTwoElements(testSet);
+ addTwoElements(other);
+
+ Assert.assertEquals(testSet, other);
+ }
+
+ @Test
+ public void setsWithTheDifferentSizesAreNotEqual() {
+ final LongHashSet other = new LongHashSet(100);
+
+ addTwoElements(testSet);
+
+ other.add(1001L);
+
+ Assert.assertNotEquals(testSet, other);
+ }
+
+ @Test
+ public void setsWithTheDifferentValuesAreNotEqual() {
+ final LongHashSet other = new LongHashSet(100);
+
+ addTwoElements(testSet);
+
+ other.add(2L);
+ other.add(1001L);
+
+ Assert.assertNotEquals(testSet, other);
+ }
+
+ @Test
+ public void twoEmptySetsHaveTheSameHashcode() {
+ Assert.assertEquals(testSet.hashCode(), new LongHashSet(100).hashCode());
+ }
+
+ @Test
+ public void setsWithTheSameValuesHaveTheSameHashcode() {
+ final LongHashSet other = new LongHashSet(100);
+
+ addTwoElements(testSet);
+
+ addTwoElements(other);
+
+ Assert.assertEquals(testSet.hashCode(), other.hashCode());
+ }
+
+ @Test
+ public void reducesSizeWhenElementRemoved() {
+ addTwoElements(testSet);
+
+ testSet.remove(1001L);
+
+ Assert.assertEquals(1, testSet.size());
+ }
+
+ @Test(expected = ArrayStoreException.class)
+ public void toArrayThrowsArrayStoreExceptionForWrongType() {
+ testSet.toArray(new String[1]);
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void toArrayThrowsNullPointerExceptionForNullArgument() {
+ final Long[] into = null;
+ testSet.toArray(into);
+ }
+
+ @Test
+ public void toArrayCopiesElementsIntoSufficientlySizedArray() {
+ addTwoElements(testSet);
+
+ final Long[] result = testSet.toArray(new Long[testSet.size()]);
+
+ assertArrayContainingElements(result);
+ }
+
+ @Test
+ public void toArrayCopiesElementsIntoNewArray() {
+ addTwoElements(testSet);
+
+ final Long[] result = testSet.toArray(new Long[testSet.size()]);
+
+ assertArrayContainingElements(result);
+ }
+
+ @Test
+ public void toArraySupportsEmptyCollection() {
+ final Long[] result = testSet.toArray(new Long[testSet.size()]);
+
+ Assert.assertArrayEquals(result, new Long[]{});
+ }
+
+ // Test case from usage bug.
+ @Test
+ public void chainCompactionShouldNotCauseElementsToBeMovedBeforeTheirHash() {
+ final LongHashSet requiredFields = new LongHashSet(14);
+
+ requiredFields.add(8L);
+ requiredFields.add(9L);
+ requiredFields.add(35L);
+ requiredFields.add(49L);
+ requiredFields.add(56L);
+
+ Assert.assertTrue("Failed to remove 8", requiredFields.remove(8L));
+ Assert.assertTrue("Failed to remove 9", requiredFields.remove(9L));
+
+ assertThat(requiredFields, containsInAnyOrder(35L, 49L, 56L));
+ }
+
+ @Test
+ public void shouldResizeWhenItHitsCapacity() {
+ for (long i = 0; i < 2 * INITIAL_CAPACITY; i++) {
+ Assert.assertTrue(testSet.add(i));
+ }
+
+ for (long i = 0; i < 2 * INITIAL_CAPACITY; i++) {
+ Assert.assertTrue(testSet.contains(i));
+ }
+ }
+
+ @Test
+ public void containsEmptySet() {
+ final LongHashSet other = new LongHashSet(100);
+
+ Assert.assertTrue(testSet.containsAll(other));
+ Assert.assertTrue(testSet.containsAll((Collection<?>) other));
+ }
+
+ @Test
+ public void containsSubset() {
+ addTwoElements(testSet);
+
+ final LongHashSet subset = new LongHashSet(100);
+
+ subset.add(1L);
+
+ Assert.assertTrue(testSet.containsAll(subset));
+ Assert.assertTrue(testSet.containsAll((Collection<?>) subset));
+ }
+
+ @Test
+ public void doesNotContainDisjointSet() {
+ addTwoElements(testSet);
+
+ final LongHashSet other = new LongHashSet(100);
+
+ other.add(1L);
+ other.add(1002L);
+
+ Assert.assertFalse(testSet.containsAll(other));
+ Assert.assertFalse(testSet.containsAll((Collection<?>) other));
+ }
+
+ @Test
+ public void doesNotContainSuperset() {
+ addTwoElements(testSet);
+
+ final LongHashSet superset = new LongHashSet(100);
+
+ addTwoElements(superset);
+ superset.add(15L);
+
+ Assert.assertFalse(testSet.containsAll(superset));
+ Assert.assertFalse(testSet.containsAll((Collection<?>) superset));
+ }
+
+ @Test
+ public void addingEmptySetDoesNothing() {
+ addTwoElements(testSet);
+
+ Assert.assertFalse(testSet.addAll(new LongHashSet(100)));
+ Assert.assertFalse(testSet.addAll(new HashSet<>()));
+ assertContainsElements(testSet);
+ }
+
+ @Test
+ public void addingSubsetDoesNothing() {
+ addTwoElements(testSet);
+
+ final LongHashSet subset = new LongHashSet(100);
+
+ subset.add(1L);
+
+ final HashSet<Long> subSetCollection = new HashSet<>(subset);
+
+ Assert.assertFalse(testSet.addAll(subset));
+ Assert.assertFalse(testSet.addAll(subSetCollection));
+ assertContainsElements(testSet);
+ }
+
+ @Test
+ public void addingEqualSetDoesNothing() {
+ addTwoElements(testSet);
+
+ final LongHashSet equal = new LongHashSet(100);
+
+ addTwoElements(equal);
+
+ final HashSet<Long> equalCollection = new HashSet<>(equal);
+
+ Assert.assertFalse(testSet.addAll(equal));
+ Assert.assertFalse(testSet.addAll(equalCollection));
+ assertContainsElements(testSet);
+ }
+
+ @Test
+ public void containsValuesAddedFromDisjointSetPrimitive() {
+ addTwoElements(testSet);
+
+ final LongHashSet disjoint = new LongHashSet(100);
+
+ disjoint.add(2L);
+ disjoint.add(1002L);
+
+ Assert.assertTrue(testSet.addAll(disjoint));
+ Assert.assertTrue(testSet.contains(1L));
+ Assert.assertTrue(testSet.contains(1001L));
+ Assert.assertTrue(testSet.containsAll(disjoint));
+ }
+
+ @Test
+ public void containsValuesAddedFromDisjointSet() {
+ addTwoElements(testSet);
+
+ final HashSet<Long> disjoint = new HashSet<>();
+
+ disjoint.add(2L);
+ disjoint.add(1002L);
+
+ Assert.assertTrue(testSet.addAll(disjoint));
+ Assert.assertTrue(testSet.contains(1L));
+ Assert.assertTrue(testSet.contains(1001L));
+ Assert.assertTrue(testSet.containsAll(disjoint));
+ }
+
+ @Test
+ public void containsValuesAddedFromIntersectingSetPrimitive() {
+ addTwoElements(testSet);
+
+ final LongHashSet intersecting = new LongHashSet(100);
+
+ intersecting.add(1L);
+ intersecting.add(1002L);
+
+ Assert.assertTrue(testSet.addAll(intersecting));
+ Assert.assertTrue(testSet.contains(1L));
+ Assert.assertTrue(testSet.contains(1001L));
+ Assert.assertTrue(testSet.containsAll(intersecting));
+ }
+
+ @Test
+ public void containsValuesAddedFromIntersectingSet() {
+ addTwoElements(testSet);
+
+ final HashSet<Long> intersecting = new HashSet<>();
+
+ intersecting.add(1L);
+ intersecting.add(1002L);
+
+ Assert.assertTrue(testSet.addAll(intersecting));
+ Assert.assertTrue(testSet.contains(1L));
+ Assert.assertTrue(testSet.contains(1001L));
+ Assert.assertTrue(testSet.containsAll(intersecting));
+ }
+
+ @Test
+ public void removingEmptySetDoesNothing() {
+ addTwoElements(testSet);
+
+ Assert.assertFalse(testSet.removeAll(new LongHashSet(100)));
+ Assert.assertFalse(testSet.removeAll(new HashSet<Long>()));
+ assertContainsElements(testSet);
+ }
+
+ @Test
+ public void removingDisjointSetDoesNothing() {
+ addTwoElements(testSet);
+
+ final LongHashSet disjoint = new LongHashSet(100);
+
+ disjoint.add(2L);
+ disjoint.add(1002L);
+
+ Assert.assertFalse(testSet.removeAll(disjoint));
+ Assert.assertFalse(testSet.removeAll(new HashSet<Long>()));
+ assertContainsElements(testSet);
+ }
+
+ @Test
+ public void doesNotContainRemovedIntersectingSetPrimitive() {
+ addTwoElements(testSet);
+
+ final LongHashSet intersecting = new LongHashSet(100);
+
+ intersecting.add(1L);
+ intersecting.add(1002L);
+
+ Assert.assertTrue(testSet.removeAll(intersecting));
+ Assert.assertTrue(testSet.contains(1001L));
+ Assert.assertFalse(testSet.containsAll(intersecting));
+ }
+
+ @Test
+ public void doesNotContainRemovedIntersectingSet() {
+ addTwoElements(testSet);
+
+ final HashSet<Long> intersecting = new HashSet<>();
+
+ intersecting.add(1L);
+ intersecting.add(1002L);
+
+ Assert.assertTrue(testSet.removeAll(intersecting));
+ Assert.assertTrue(testSet.contains(1001L));
+ Assert.assertFalse(testSet.containsAll(intersecting));
+ }
+
+ @Test
+ public void isEmptyAfterRemovingEqualSetPrimitive() {
+ addTwoElements(testSet);
+
+ final LongHashSet equal = new LongHashSet(100);
+
+ addTwoElements(equal);
+
+ Assert.assertTrue(testSet.removeAll(equal));
+ Assert.assertTrue(testSet.isEmpty());
+ }
+
+ @Test
+ public void isEmptyAfterRemovingEqualSet() {
+ addTwoElements(testSet);
+
+ final HashSet<Long> equal = new HashSet<>();
+
+ addTwoElements(equal);
+
+ Assert.assertTrue(testSet.removeAll(equal));
+ Assert.assertTrue(testSet.isEmpty());
+ }
+
+ @Test
+ public void removeElementsFromIterator() {
+ addTwoElements(testSet);
+
+ final LongHashSet.LongIterator iterator = testSet.iterator();
+ while (iterator.hasNext()) {
+ if (iterator.nextValue() == 1L) {
+ iterator.remove();
+ }
+ }
+
+ assertThat(testSet, contains(1001L));
+ assertThat(testSet, hasSize(1));
+ }
+
+ @Test
+ public void shouldNotContainMissingValueInitially() {
+ Assert.assertFalse(testSet.contains(LongHashSet.MISSING_VALUE));
+ }
+
+ @Test
+ public void shouldAllowMissingValue() {
+ Assert.assertTrue(testSet.add(LongHashSet.MISSING_VALUE));
+
+ Assert.assertTrue(testSet.contains(LongHashSet.MISSING_VALUE));
+
+ Assert.assertFalse(testSet.add(LongHashSet.MISSING_VALUE));
+ }
+
+ @Test
+ public void shouldAllowRemovalOfMissingValue() {
+ Assert.assertTrue(testSet.add(LongHashSet.MISSING_VALUE));
+
+ Assert.assertTrue(testSet.remove(LongHashSet.MISSING_VALUE));
+
+ Assert.assertFalse(testSet.contains(LongHashSet.MISSING_VALUE));
+
+ Assert.assertFalse(testSet.remove(LongHashSet.MISSING_VALUE));
+ }
+
+ @Test
+ public void sizeAccountsForMissingValue() {
+ testSet.add(1L);
+ testSet.add(LongHashSet.MISSING_VALUE);
+
+ Assert.assertEquals(2, testSet.size());
+ }
+
+ @Test
+ public void toArrayCopiesElementsIntoNewArrayIncludingMissingValue() {
+ addTwoElements(testSet);
+
+ testSet.add(LongHashSet.MISSING_VALUE);
+
+ final Long[] result = testSet.toArray(new Long[testSet.size()]);
+
+ assertThat(result, arrayContainingInAnyOrder(1L, 1001L, LongHashSet.MISSING_VALUE));
+ }
+
+ @Test
+ public void toObjectArrayCopiesElementsIntoNewArrayIncludingMissingValue() {
+ addTwoElements(testSet);
+
+ testSet.add(LongHashSet.MISSING_VALUE);
+
+ final Object[] result = testSet.toArray();
+
+ assertThat(result, arrayContainingInAnyOrder(1L, 1001L, LongHashSet.MISSING_VALUE));
+ }
+
+ @Test
+ public void equalsAccountsForMissingValue() {
+ addTwoElements(testSet);
+ testSet.add(LongHashSet.MISSING_VALUE);
+
+ final LongHashSet other = new LongHashSet(100);
+ addTwoElements(other);
+
+ Assert.assertNotEquals(testSet, other);
+
+ other.add(LongHashSet.MISSING_VALUE);
+ Assert.assertEquals(testSet, other);
+
+ testSet.remove(LongHashSet.MISSING_VALUE);
+
+ Assert.assertNotEquals(testSet, other);
+ }
+
+ @Test
+ public void consecutiveValuesShouldBeCorrectlyStored() {
+ for (long i = 0; i < 10_000; i++) {
+ testSet.add(i);
+ }
+
+ assertThat(testSet, hasSize(10_000));
+
+ int distinctElements = 0;
+ for (final long ignore : testSet) {
+ distinctElements++;
+ }
+
+ assertThat(distinctElements, is(10_000));
+ }
+
+ @Test
+ public void hashCodeAccountsForMissingValue() {
+ addTwoElements(testSet);
+ testSet.add(LongHashSet.MISSING_VALUE);
+
+ final LongHashSet other = new LongHashSet(100);
+ addTwoElements(other);
+
+ Assert.assertNotEquals(testSet.hashCode(), other.hashCode());
+
+ other.add(LongHashSet.MISSING_VALUE);
+ Assert.assertEquals(testSet.hashCode(), other.hashCode());
+
+ testSet.remove(LongHashSet.MISSING_VALUE);
+
+ Assert.assertNotEquals(testSet.hashCode(), other.hashCode());
+ }
+
+ @Test
+ public void iteratorAccountsForMissingValue() {
+ addTwoElements(testSet);
+ testSet.add(LongHashSet.MISSING_VALUE);
+
+ int missingValueCount = 0;
+ final LongHashSet.LongIterator iterator = testSet.iterator();
+ while (iterator.hasNext()) {
+ if (iterator.nextValue() == LongHashSet.MISSING_VALUE) {
+ missingValueCount++;
+ }
+ }
+
+ Assert.assertEquals(1, missingValueCount);
+ }
+
+ @Test
+ public void iteratorCanRemoveMissingValue() {
+ addTwoElements(testSet);
+ testSet.add(LongHashSet.MISSING_VALUE);
+
+ final LongHashSet.LongIterator iterator = testSet.iterator();
+ while (iterator.hasNext()) {
+ if (iterator.nextValue() == LongHashSet.MISSING_VALUE) {
+ iterator.remove();
+ }
+ }
+
+ Assert.assertFalse(testSet.contains(LongHashSet.MISSING_VALUE));
+ }
+
+ @Test
+ public void shouldGenerateStringRepresentation() {
+ final long[] testEntries = {3L, 1L, -2L, 19L, 7L, 11L, 12L, 7L};
+
+ for (final long testEntry : testEntries) {
+ testSet.add(testEntry);
+ }
+
+ final String mapAsAString = "{1, 19, 11, 7, 3, 12, -2}";
+ assertThat(testSet.toString(), equalTo(mapAsAString));
+ }
+
+ @Test
+ public void shouldRemoveMissingValueWhenCleared() {
+ Assert.assertTrue(testSet.add(LongHashSet.MISSING_VALUE));
+
+ testSet.clear();
+
+ Assert.assertFalse(testSet.contains(LongHashSet.MISSING_VALUE));
+ }
+
+ @Test
+ public void shouldHaveCompatibleEqualsAndHashcode() {
+ final HashSet<Long> compatibleSet = new HashSet<>();
+ final long seed = System.nanoTime();
+ final Random r = new Random(seed);
+ for (long i = 0; i < 1024; i++) {
+ final long value = r.nextLong();
+ compatibleSet.add(value);
+ testSet.add(value);
+ }
+
+ if (r.nextBoolean()) {
+ compatibleSet.add(LongHashSet.MISSING_VALUE);
+ testSet.add(LongHashSet.MISSING_VALUE);
+ }
+
+ Assert.assertEquals("Fail with seed:" + seed, testSet, compatibleSet);
+ Assert.assertEquals("Fail with seed:" + seed, compatibleSet, testSet);
+ Assert.assertEquals("Fail with seed:" + seed, compatibleSet.hashCode(), testSet.hashCode());
+ }
+
+ private static void addTwoElements(final LongHashSet obj) {
+ obj.add(1L);
+ obj.add(1001L);
+ }
+
+ private static void addTwoElements(final HashSet<Long> obj) {
+ obj.add(1L);
+ obj.add(1001L);
+ }
+
+ private void assertIteratorHasElements() {
+ final Iterator<Long> iter = testSet.iterator();
+
+ final Set<Long> values = new HashSet<>();
+
+ Assert.assertTrue(iter.hasNext());
+ values.add(iter.next());
+ Assert.assertTrue(iter.hasNext());
+ values.add(iter.next());
+ Assert.assertFalse(iter.hasNext());
+
+ assertContainsElements(values);
+ }
+
+ private void assertIteratorHasElementsWithoutHasNext() {
+ final Iterator<Long> iter = testSet.iterator();
+
+ final Set<Long> values = new HashSet<>();
+
+ values.add(iter.next());
+ values.add(iter.next());
+
+ assertContainsElements(values);
+ }
+
+ private static void assertArrayContainingElements(final Long[] result) {
+ assertThat(result, arrayContainingInAnyOrder(1L, 1001L));
+ }
+
+ private static void assertContainsElements(final Set<Long> other) {
+ assertThat(other, containsInAnyOrder(1L, 1001L));
+ }
+
+ private void exhaustIterator() {
+ final Iterator iterator = testSet.iterator();
+ iterator.next();
+ iterator.next();
+ iterator.next();
+ }
+}
diff --git a/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/collections/SparseArrayLinkedListTest.java b/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/collections/SparseArrayLinkedListTest.java
new file mode 100644
index 0000000..3ff5805
--- /dev/null
+++ b/artemis-commons/src/test/java/org/apache/activemq/artemis/utils/collections/SparseArrayLinkedListTest.java
@@ -0,0 +1,158 @@
+/*
+ *
+ * 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.utils.collections;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import static org.hamcrest.core.Is.is;
+
+public class SparseArrayLinkedListTest {
+
+ private static final int SPARSE_ARRAY_CAPACITY = 4;
+ private static final int ELEMENTS = SPARSE_ARRAY_CAPACITY * 4;
+
+ private final SparseArrayLinkedList<Integer> list;
+
+ public SparseArrayLinkedListTest() {
+ list = new SparseArrayLinkedList<>(SPARSE_ARRAY_CAPACITY);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void shouldFailToCreateZeroArrayCapacityCollection() {
+ new SparseArrayLinkedList<>(0);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void shouldFailToCreateNegativeArrayCapacityCollection() {
+ new SparseArrayLinkedList<>(-1);
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void shouldFailToAddNull() {
+ list.add(null);
+ }
+
+ @Test
+ public void shouldNumberOfElementsBeTheSameOfTheAddedElements() {
+ final int elements = ELEMENTS;
+ for (int i = 0; i < elements; i++) {
+ Assert.assertEquals(i, list.size());
+ list.add(i);
+ }
+ Assert.assertEquals(elements, list.size());
+ }
+
+ @Test
+ public void shouldClearConsumeElementsInOrder() {
+ final int elements = ELEMENTS;
+ Assert.assertEquals(0, list.clear(null));
+ final ArrayList<Integer> expected = new ArrayList<>(elements);
+ for (int i = 0; i < elements; i++) {
+ final Integer added = i;
+ list.add(added);
+ expected.add(added);
+ }
+ final List<Integer> removed = new ArrayList<>(elements);
+ Assert.assertEquals(elements, list.clear(removed::add));
+ Assert.assertEquals(0, list.sparseArraysCount());
+ Assert.assertEquals(0, list.size());
+ Assert.assertThat(removed, is(expected));
+ }
+
+ @Test
+ public void shouldRemoveMatchingElements() {
+ final int elements = ELEMENTS;
+ for (int i = 0; i < elements; i++) {
+ list.add(i);
+ }
+ Assert.assertEquals(1, list.remove(e -> e.intValue() == 0));
+ Assert.assertEquals(elements - 1, list.size());
+ Assert.assertEquals(0, list.remove(e -> e.intValue() == 0));
+ Assert.assertEquals(elements - 1, list.size());
+ Assert.assertEquals(elements - 1, list.remove(e -> true));
+ Assert.assertEquals(0, list.size());
+ Assert.assertEquals(0, list.remove(e -> true));
+ Assert.assertEquals(0, list.sparseArraysCount());
+ }
+
+ @Test
+ public void shouldRemoveDetachSparseArrays() {
+ final int elements = list.sparseArrayCapacity() * 3;
+ for (int i = 0; i < elements; i++) {
+ list.add(i);
+ }
+ // remove elements in the middle
+ final int startInclusiveMiddle = list.sparseArrayCapacity();
+ final int endNotInclusiveMiddle = startInclusiveMiddle + list.sparseArrayCapacity();
+ Assert.assertEquals(list.sparseArrayCapacity(),
+ list.remove(e -> e.intValue() >= startInclusiveMiddle && e.intValue() < endNotInclusiveMiddle));
+ Assert.assertEquals(2, list.sparseArraysCount());
+ // remove elements at the beginning
+ final int startInclusiveFirst = 0;
+ final int endNotInclusiveFirst = startInclusiveMiddle;
+ Assert.assertEquals(list.sparseArrayCapacity(),
+ list.remove(e -> e.intValue() >= startInclusiveFirst && e.intValue() < endNotInclusiveFirst));
+ Assert.assertEquals(1, list.sparseArraysCount());
+ // remove all elements at the end
+ final int startInclusiveLast = endNotInclusiveMiddle;
+ final int endNotInclusiveLast = elements;
+ Assert.assertEquals(list.sparseArrayCapacity(),
+ list.remove(e -> e.intValue() >= startInclusiveLast && e.intValue() < endNotInclusiveLast));
+ Assert.assertEquals(0, list.sparseArraysCount());
+ }
+
+ @Test
+ public void shouldAddAfterRemoveAtTheEndReusingTheAvailableSpace() {
+ final int elements = list.sparseArrayCapacity();
+ for (int i = 0; i < elements; i++) {
+ list.add(i);
+ }
+ Assert.assertEquals(1, list.sparseArraysCount());
+ // removing last element
+ Assert.assertEquals(1, list.remove(e -> e.intValue() == elements - 1));
+ list.add(elements - 1);
+ Assert.assertEquals(1, list.sparseArraysCount());
+ Assert.assertEquals(1, list.remove(e -> e.intValue() == 0));
+ list.add(elements);
+ Assert.assertEquals(2, list.sparseArraysCount());
+ }
+
+ @Test
+ public void shouldClearConsumeRemainingElementsInOrder() {
+ final int elements = ELEMENTS;
+ final Integer zero = 0;
+ list.add(zero);
+ for (int i = 1; i < elements; i++) {
+ list.add(i);
+ }
+ Assert.assertEquals(elements - 1, list.remove(e -> e != zero));
+ final ArrayList<Integer> remaining = new ArrayList<>();
+ Assert.assertEquals(1, list.clear(remaining::add));
+ Assert.assertEquals(0, list.size());
+ Assert.assertEquals(1, remaining.size());
+ Assert.assertEquals(zero, remaining.get(0));
+ }
+
+}
diff --git a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java
index faeb968..55e9be6 100644
--- a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java
+++ b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java
@@ -24,13 +24,9 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
import java.util.LinkedHashMap;
-import java.util.LinkedList;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
@@ -44,6 +40,8 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
@@ -82,6 +80,8 @@ import org.apache.activemq.artemis.utils.actors.OrderedExecutorFactory;
import org.apache.activemq.artemis.utils.collections.ConcurrentHashSet;
import org.apache.activemq.artemis.utils.collections.ConcurrentLongHashMap;
import org.apache.activemq.artemis.utils.collections.ConcurrentLongHashSet;
+import org.apache.activemq.artemis.utils.collections.LongHashSet;
+import org.apache.activemq.artemis.utils.collections.SparseArrayLinkedList;
import org.jboss.logging.Logger;
import static org.apache.activemq.artemis.core.journal.impl.Reclaimer.scan;
@@ -1449,9 +1449,11 @@ public class JournalImpl extends JournalBase implements TestableJournal, Journal
final List<PreparedTransactionInfo> preparedTransactions,
final TransactionFailureCallback failureCallback,
final boolean fixBadTX) throws Exception {
- final Set<Long> recordsToDelete = new HashSet<>();
+ final LongHashSet recordsToDelete = new LongHashSet(1024);
+ final Predicate<RecordInfo> toDeleteFilter = recordInfo -> recordsToDelete.contains(recordInfo.id);
// ArrayList was taking too long to delete elements on checkDeleteSize
- final List<RecordInfo> records = new LinkedList<>();
+ // and LinkedList<RecordInfo> creates too many nodes
+ final SparseArrayLinkedList<RecordInfo> records = new SparseArrayLinkedList<>();
final int DELETE_FLUSH = 20000;
@@ -1461,18 +1463,15 @@ public class JournalImpl extends JournalBase implements TestableJournal, Journal
private void checkDeleteSize() {
// HORNETQ-482 - Flush deletes only if memory is critical
if (recordsToDelete.size() > DELETE_FLUSH && runtime.freeMemory() < runtime.maxMemory() * 0.2) {
- logger.debug("Flushing deletes during loading, deleteCount = " + recordsToDelete.size());
+ if (logger.isDebugEnabled()) {
+ logger.debugf("Flushing deletes during loading, deleteCount = %d", recordsToDelete.size());
+ }
// Clean up when the list is too large, or it won't be possible to load large sets of files
// Done as part of JBMESSAGING-1678
- Iterator<RecordInfo> iter = records.iterator();
- while (iter.hasNext()) {
- RecordInfo record = iter.next();
-
- if (recordsToDelete.contains(record.id)) {
- iter.remove();
- }
+ final long removed = records.remove(toDeleteFilter);
+ if (logger.isDebugEnabled()) {
+ logger.debugf("Removed records during loading = %d", removed);
}
-
recordsToDelete.clear();
logger.debug("flush delete done");
@@ -1513,12 +1512,13 @@ public class JournalImpl extends JournalBase implements TestableJournal, Journal
}
}, fixBadTX, null);
- for (RecordInfo record : records) {
+ final Consumer<RecordInfo> fillCommittedRecord = record -> {
if (!recordsToDelete.contains(record.id)) {
committedRecords.add(record);
}
- }
-
+ };
+ // it helps GC by cleaning up each SparseArray too
+ records.clear(fillCommittedRecord);
return info;
}