diff --git a/activemq-client/pom.xml b/activemq-client/pom.xml index 57685b2af13..614c41ef880 100644 --- a/activemq-client/pom.xml +++ b/activemq-client/pom.xml @@ -66,6 +66,12 @@ provided + + com.github.ben-manes.caffeine + caffeine + 3.2.4 + + diff --git a/activemq-client/src/main/java/org/apache/activemq/CaffeineMessageAudit.java b/activemq-client/src/main/java/org/apache/activemq/CaffeineMessageAudit.java new file mode 100644 index 00000000000..fb9dd4158e7 --- /dev/null +++ b/activemq-client/src/main/java/org/apache/activemq/CaffeineMessageAudit.java @@ -0,0 +1,182 @@ +/** + * 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; + +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; + +import org.apache.activemq.command.MessageId; +import org.apache.activemq.command.ProducerId; +import org.apache.activemq.util.AtomicBitArrayBin; +import org.apache.activemq.util.IdGenerator; + +/** + * A message audit backed by Caffeine cache and {@link AtomicBitArrayBin}. + * + *

Uses Caffeine's {@link Cache} with size-based eviction (TinyLfu policy) + * for bounded producer tracking, paired with lock-free + * {@link AtomicBitArrayBin} for per-producer bit operations. + * + *

Producer count is bounded by {@link #getMaximumNumberOfProducersToTrack()}. + * Caffeine's TinyLfu admission policy evicts entries based on frequency and + * recency, providing near-optimal hit rates. The per-producer audit window is + * controlled by {@link #getAuditDepth()}. + * + *

All per-producer bit operations are lock-free via {@link AtomicBitArrayBin}. + * The only synchronization points are within Caffeine's internal structures + * for cache management. + */ +public class CaffeineMessageAudit { + + public static final int DEFAULT_WINDOW_SIZE = 2048; + public static final int MAXIMUM_PRODUCER_COUNT = 64; + + private volatile int auditDepth; + private volatile int maximumNumberOfProducersToTrack; + private volatile Cache cache; + + public CaffeineMessageAudit() { + this(DEFAULT_WINDOW_SIZE, MAXIMUM_PRODUCER_COUNT); + } + + public CaffeineMessageAudit(int auditDepth, int maximumNumberOfProducersToTrack) { + this.auditDepth = auditDepth; + this.maximumNumberOfProducersToTrack = maximumNumberOfProducersToTrack; + this.cache = buildCache(maximumNumberOfProducersToTrack); + } + + public int getAuditDepth() { + return auditDepth; + } + + public void setAuditDepth(int auditDepth) { + this.auditDepth = auditDepth; + } + + public int getMaximumNumberOfProducersToTrack() { + return maximumNumberOfProducersToTrack; + } + + public void setMaximumNumberOfProducersToTrack(int maximumNumberOfProducersToTrack) { + this.maximumNumberOfProducersToTrack = maximumNumberOfProducersToTrack; + var newCache = buildCache(maximumNumberOfProducersToTrack); + newCache.putAll(this.cache.asMap()); + this.cache = newCache; + } + + public boolean isDuplicate(String id) { + var seed = IdGenerator.getSeedFromId(id); + if (seed == null) { + return false; + } + var bab = cache.get(seed, k -> new AtomicBitArrayBin(auditDepth)); + var index = IdGenerator.getSequenceFromId(id); + if (index >= 0) { + return bab.setBit(index, true); + } + return false; + } + + public boolean isDuplicate(final MessageId id) { + if (id == null) { + return false; + } + var pid = id.getProducerId(); + if (pid == null) { + return false; + } + var bab = cache.get(pid.toString(), k -> new AtomicBitArrayBin(auditDepth)); + return bab.setBit(id.getProducerSequenceId(), true); + } + + public void rollback(final MessageId id) { + if (id == null) { + return; + } + var pid = id.getProducerId(); + if (pid == null) { + return; + } + var bab = cache.getIfPresent(pid.toString()); + if (bab != null) { + bab.setBit(id.getProducerSequenceId(), false); + } + } + + public void rollback(final String id) { + var seed = IdGenerator.getSeedFromId(id); + if (seed == null) { + return; + } + var bab = cache.getIfPresent(seed); + if (bab != null) { + var index = IdGenerator.getSequenceFromId(id); + bab.setBit(index, false); + } + } + + public boolean isInOrder(final String id) { + if (id == null) { + return true; + } + var seed = IdGenerator.getSeedFromId(id); + if (seed == null) { + return true; + } + var bab = cache.getIfPresent(seed); + if (bab != null) { + var index = IdGenerator.getSequenceFromId(id); + return bab.isInOrder(index); + } + return true; + } + + public boolean isInOrder(final MessageId id) { + if (id == null) { + return false; + } + var pid = id.getProducerId(); + if (pid == null) { + return false; + } + var bab = cache.get(pid.toString(), k -> new AtomicBitArrayBin(auditDepth)); + return bab.isInOrder(id.getProducerSequenceId()); + } + + public long getLastSeqId(ProducerId id) { + var bab = cache.getIfPresent(id.toString()); + if (bab != null) { + return bab.getLastSetIndex(); + } + return -1; + } + + public void clear() { + cache.invalidateAll(); + } + + public int getProducerCount() { + cache.cleanUp(); + return (int) cache.estimatedSize(); + } + + private static Cache buildCache(int maxSize) { + return Caffeine.newBuilder() + .maximumSize(maxSize) + .build(); + } +} diff --git a/activemq-client/src/main/java/org/apache/activemq/ConcurrentMessageAudit.java b/activemq-client/src/main/java/org/apache/activemq/ConcurrentMessageAudit.java new file mode 100644 index 00000000000..11c9eeeba61 --- /dev/null +++ b/activemq-client/src/main/java/org/apache/activemq/ConcurrentMessageAudit.java @@ -0,0 +1,200 @@ +/** + * 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; + +import java.util.Iterator; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.activemq.command.MessageId; +import org.apache.activemq.command.ProducerId; +import org.apache.activemq.util.AtomicBitArrayBin; +import org.apache.activemq.util.IdGenerator; + +/** + * A lock-free message audit backed by {@link ConcurrentHashMap} and + * {@link AtomicBitArrayBin}. + * + *

The upstream {@code ActiveMQMessageAudit} wraps every operation in a + * single {@code synchronized(this)} block, serializing all threads regardless + * of which producer they are working with. This class eliminates all + * synchronization on the hot path: + *

+ * + *

Producer count is bounded by {@link #getMaximumNumberOfProducersToTrack()}. + * When the limit is exceeded, entries are evicted in iteration order (approximate + * FIFO). The per-producer audit window is controlled by {@link #getAuditDepth()}. + */ +public class ConcurrentMessageAudit { + + public static final int DEFAULT_WINDOW_SIZE = 2048; + public static final int MAXIMUM_PRODUCER_COUNT = 64; + + private volatile int auditDepth; + private volatile int maximumNumberOfProducersToTrack; + private final ConcurrentHashMap map; + + public ConcurrentMessageAudit() { + this(DEFAULT_WINDOW_SIZE, MAXIMUM_PRODUCER_COUNT); + } + + public ConcurrentMessageAudit(int auditDepth, int maximumNumberOfProducersToTrack) { + this.auditDepth = auditDepth; + this.maximumNumberOfProducersToTrack = maximumNumberOfProducersToTrack; + this.map = new ConcurrentHashMap<>(maximumNumberOfProducersToTrack); + } + + public int getAuditDepth() { + return auditDepth; + } + + public void setAuditDepth(int auditDepth) { + this.auditDepth = auditDepth; + } + + public int getMaximumNumberOfProducersToTrack() { + return maximumNumberOfProducersToTrack; + } + + public void setMaximumNumberOfProducersToTrack(int maximumNumberOfProducersToTrack) { + this.maximumNumberOfProducersToTrack = maximumNumberOfProducersToTrack; + evictExcess(); + } + + public boolean isDuplicate(String id) { + var seed = IdGenerator.getSeedFromId(id); + if (seed == null) { + return false; + } + var bab = getOrCreate(seed); + var index = IdGenerator.getSequenceFromId(id); + if (index >= 0) { + return bab.setBit(index, true); + } + return false; + } + + public boolean isDuplicate(final MessageId id) { + if (id == null) { + return false; + } + var pid = id.getProducerId(); + if (pid == null) { + return false; + } + var bab = getOrCreate(pid.toString()); + return bab.setBit(id.getProducerSequenceId(), true); + } + + public void rollback(final MessageId id) { + if (id == null) { + return; + } + var pid = id.getProducerId(); + if (pid == null) { + return; + } + var bab = map.get(pid.toString()); + if (bab != null) { + bab.setBit(id.getProducerSequenceId(), false); + } + } + + public void rollback(final String id) { + var seed = IdGenerator.getSeedFromId(id); + if (seed == null) { + return; + } + var bab = map.get(seed); + if (bab != null) { + long index = IdGenerator.getSequenceFromId(id); + bab.setBit(index, false); + } + } + + public boolean isInOrder(final String id) { + if (id == null) { + return true; + } + var seed = IdGenerator.getSeedFromId(id); + if (seed == null) { + return true; + } + var bab = map.get(seed); + if (bab != null) { + var index = IdGenerator.getSequenceFromId(id); + return bab.isInOrder(index); + } + return true; + } + + public boolean isInOrder(final MessageId id) { + if (id == null) { + return false; + } + var pid = id.getProducerId(); + if (pid == null) { + return false; + } + var bab = getOrCreate(pid.toString()); + return bab.isInOrder(id.getProducerSequenceId()); + } + + public long getLastSeqId(ProducerId id) { + var bab = map.get(id.toString()); + if (bab != null) { + return bab.getLastSetIndex(); + } + return -1; + } + + public void clear() { + map.clear(); + } + + public int getProducerCount() { + return map.size(); + } + + private AtomicBitArrayBin getOrCreate(String key) { + var bab = map.get(key); + if (bab != null) { + return bab; + } + bab = map.computeIfAbsent(key, k -> new AtomicBitArrayBin(auditDepth)); + evictExcess(); + return bab; + } + + private void evictExcess() { + var max = maximumNumberOfProducersToTrack; + while (map.size() > max) { + var it = map.keySet().iterator(); + if (it.hasNext()) { + it.next(); + it.remove(); + } else { + break; + } + } + } +} diff --git a/activemq-client/src/main/java/org/apache/activemq/util/AtomicBitArrayBin.java b/activemq-client/src/main/java/org/apache/activemq/util/AtomicBitArrayBin.java new file mode 100644 index 00000000000..46354249a88 --- /dev/null +++ b/activemq-client/src/main/java/org/apache/activemq/util/AtomicBitArrayBin.java @@ -0,0 +1,230 @@ +/** + * 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.util; + +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicLongArray; + +/** + * A lock-free replacement for {@link BitArrayBin} that uses {@link AtomicLongArray} + * as a ring buffer with CAS-based bit operations. + * + *

The upstream {@code BitArrayBin} stores bits in a {@code LinkedList} + * and requires external synchronization for all access. This class replaces that + * with a fixed-size {@code AtomicLongArray} ring buffer where each slot holds 64 bits. + * All operations use CAS (compare-and-swap) instead of locks. + * + *

Ring buffer design

+ * + *

Slots are addressed by absolute position: {@code ringPos = (index / 64) % capacity}. + * This mapping is independent of the window origin, so when the window advances, + * slots that remain in the window keep their data without copying. Only evicted + * slots need to be reclaimed, which happens lazily on first access. + * + *

Each slot carries an epoch ({@code slotEpoch[ringPos]}) identifying which + * absolute 64-bit block it holds. On access, if the slot's epoch doesn't match + * the expected epoch, the slot is reclaimed via a brief CAS protocol: the epoch + * is set to a {@code CLEARING} sentinel, bits are zeroed, and the epoch is + * published to the new value. Concurrent readers that see {@code CLEARING} + * retry until the transition completes. + * + *

Concurrency guarantees

+ * + */ +public class AtomicBitArrayBin { + + static final int LONG_SIZE = 64; + + private static final long UNINITIALIZED = -1L; + private static final long CLEARING = Long.MAX_VALUE; + + private final int capacity; + private final AtomicLongArray bits; + private final AtomicLongArray slotEpoch; + private final AtomicLong origin; + private final AtomicLong lastInOrderBit; + + public AtomicBitArrayBin(int windowSize) { + capacity = Math.max(1, ((windowSize + 1) / LONG_SIZE) + 1); + bits = new AtomicLongArray(capacity); + slotEpoch = new AtomicLongArray(capacity); + for (var i = 0; i < capacity; i++) { + slotEpoch.set(i, UNINITIALIZED); + } + origin = new AtomicLong(0); + lastInOrderBit = new AtomicLong(UNINITIALIZED); + } + + /** + * Set or clear a bit at the given index. + * + * @param index the absolute bit index (message sequence number) + * @param value true to set, false to clear + * @return the previous value of the bit (true if it was already set) + */ + public boolean setBit(long index, boolean value) { + if (index < 0) return false; + + while (true) { + var orig = origin.get(); + var epoch = index / LONG_SIZE; + var originEpoch = orig / LONG_SIZE; + + if (epoch < originEpoch) { + return true; + } + + if (epoch >= originEpoch + capacity) { + advanceOrigin(orig, epoch); + continue; + } + + var ringPos = (int)(epoch % capacity); + var bitOffset = (int)(index % LONG_SIZE); + long mask = 1L << bitOffset; + + if (!ensureSlotEpoch(ringPos, epoch)) { + continue; + } + + while (true) { + if (slotEpoch.get(ringPos) != epoch) { + break; + } + + var oldBits = bits.get(ringPos); + var wasSet = (oldBits & mask) != 0; + + if (value) { + if (wasSet) return true; + if (bits.compareAndSet(ringPos, oldBits, oldBits | mask)) return false; + } else { + if (!wasSet) return false; + if (bits.compareAndSet(ringPos, oldBits, oldBits & ~mask)) return true; + } + } + } + } + + /** + * Get the boolean value at the index. + * + * @param index the absolute bit index + * @return true if the bit is set, or if the index is behind the window + */ + public boolean getBit(long index) { + if (index < 0) return false; + + var orig = origin.get(); + var epoch = index / LONG_SIZE; + var originEpoch = orig / LONG_SIZE; + + if (epoch < originEpoch) return true; + if (epoch >= originEpoch + capacity) return false; + + var ringPos = (int)(epoch % capacity); + var curEpoch = slotEpoch.get(ringPos); + + if (curEpoch != epoch) { + return curEpoch > epoch && curEpoch != UNINITIALIZED; + } + + var bitOffset = (int)(index % LONG_SIZE); + return (bits.get(ringPos) & (1L << bitOffset)) != 0; + } + + /** + * Test if the index is the next expected in-order sequence. + * + * @param index the absolute bit index + * @return true if this is the next in-order message + */ + public boolean isInOrder(long index) { + var prev = lastInOrderBit.getAndSet(index); + return prev == UNINITIALIZED || prev + 1 == index; + } + + /** + * Get the index of the highest set bit across all valid slots. + * + * @return the highest set bit index, or -1 if no bits are set + */ + public long getLastSetIndex() { + var orig = origin.get(); + var originEpoch = orig / LONG_SIZE; + + for (int offset = capacity - 1; offset >= 0; offset--) { + var epoch = originEpoch + offset; + var ringPos = (int)(epoch % capacity); + + var curEpoch = slotEpoch.get(ringPos); + if (curEpoch != epoch || curEpoch == CLEARING) continue; + + var slotBits = bits.get(ringPos); + if (slotBits != 0) { + var highBit = LONG_SIZE - 1 - Long.numberOfLeadingZeros(slotBits); + return epoch * LONG_SIZE + highBit; + } + } + return -1; + } + + /** + * @return the number of 64-bit slots in the ring buffer + */ + public int getCapacity() { + return capacity; + } + + private void advanceOrigin(long currentOrigin, long targetEpoch) { + var newOriginEpoch = targetEpoch - capacity + 1; + var newOrigin = Math.max(0, newOriginEpoch * LONG_SIZE); + if (newOrigin > currentOrigin) { + origin.compareAndSet(currentOrigin, newOrigin); + } + } + + private boolean ensureSlotEpoch(int ringPos, long expectedEpoch) { + var curEpoch = slotEpoch.get(ringPos); + + if (curEpoch == expectedEpoch) return true; + + if (curEpoch == CLEARING) { + Thread.yield(); + return false; + } + + if (curEpoch > expectedEpoch && curEpoch != UNINITIALIZED) { + return false; + } + + if (slotEpoch.compareAndSet(ringPos, curEpoch, CLEARING)) { + bits.set(ringPos, 0); + slotEpoch.set(ringPos, expectedEpoch); + return true; + } + + return false; + } +} diff --git a/activemq-client/src/test/java/org/apache/activemq/MessageAuditPerformanceTest.java b/activemq-client/src/test/java/org/apache/activemq/MessageAuditPerformanceTest.java new file mode 100644 index 00000000000..f8b81f2de49 --- /dev/null +++ b/activemq-client/src/test/java/org/apache/activemq/MessageAuditPerformanceTest.java @@ -0,0 +1,794 @@ +/** + * 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; + +import static org.junit.Assert.*; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentSkipListMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.concurrent.locks.StampedLock; + +import org.apache.activemq.util.AtomicBitArrayBin; +import org.apache.activemq.util.BitArrayBin; +import org.apache.activemq.util.IdGenerator; +import org.junit.Test; + +/** + * Compares the throughput of message audit implementations under three + * concurrency patterns: + * + *
    + *
  1. Partitioned — each thread owns a disjoint set of + * producers (no map or bin contention between threads)
    2. + *
  2. Shared — all threads access all producers, each writing + * a distinct message slice (map read contention + per-bin lock + * contention)
    3. + *
  3. Mixed R/W — audit is pre-populated, then all threads + * interleave duplicate re-checks (reads) with new inserts (writes) + * across all producers (the most realistic broker workload)
    4. + *
+ * + * Strategies under test: + *
    + *
  1. sync+LRU — upstream {@link ActiveMQMessageAudit}
    2. + *
  2. CHM+Atomic — {@link ConcurrentMessageAudit}
    3. + *
  3. Caffeine — {@link CaffeineMessageAudit}
    4. + *
  4. SkipList — {@link ConcurrentSkipListMap} + per-bin sync
    5. + *
  5. RWLock — {@link java.util.HashMap} + {@link ReentrantReadWriteLock}
    6. + *
  6. Stamped — {@link java.util.HashMap} + {@link StampedLock}
    7. + *
+ */ +public class MessageAuditPerformanceTest { + + private static final int WARMUP_ITERATIONS = 3; + private static final int MEASURE_ITERATIONS = 5; + private static final int MESSAGES_PER_PRODUCER = 5_000; + + private static final int[] THREAD_COUNTS = {1, 2, 4, 8}; + private static final int[] PRODUCER_COUNTS = {1, 8, 16, 64, 128}; + + private static final String[] NAMES = {"sync+LRU", "CHM+Atomic", "Caffeine", "SkipList", "RWLock", "Stamped"}; + private static final AuditFactory[] FACTORIES = { + MessageAuditPerformanceTest::createSyncLru, + MessageAuditPerformanceTest::createConcurrentAudit, + MessageAuditPerformanceTest::createCaffeineAudit, + MessageAuditPerformanceTest::createSkipList, + MessageAuditPerformanceTest::createReadWriteLock, + MessageAuditPerformanceTest::createStampedLock + }; + + @FunctionalInterface + private interface StringAudit { + boolean isDuplicate(String id); + } + + @FunctionalInterface + private interface AuditFactory { + StringAudit create(int auditDepth, int producers); + } + + @FunctionalInterface + private interface BenchmarkRunner { + long run(StringAudit fn, String[][] ids, String[][] extraIds, + int threads, int producers) throws Exception; + } + + // ======================================================================== + // Benchmark scenarios + // ======================================================================== + + @Test + public void comparePartitioned() throws Exception { + printBenchmark("PARTITIONED — producers split across threads, no cross-thread contention", + (fn, ids, extra, threads, producers) -> + runPartitioned(fn, ids, threads, producers)); + } + + @Test + public void compareSharedContention() throws Exception { + printBenchmark("SHARED — all threads hit all producers, per-bin lock contention", + (fn, ids, extra, threads, producers) -> + runShared(fn, ids, threads, producers)); + } + + @Test + public void compareMixedReadWrite() throws Exception { + printBenchmark("MIXED R/W — pre-populated, 50/50 duplicate re-checks and new inserts", + (fn, ids, extra, threads, producers) -> + runMixed(fn, ids, extra, threads, producers)); + } + + @Test + public void compareOptimizationLayers() throws Exception { + String[] layerNames = {"sync+LRU", "CHM+syncBin", "CHM+AtomicBin"}; + AuditFactory[] layerFactories = { + MessageAuditPerformanceTest::createSyncLru, + MessageAuditPerformanceTest::createChmSyncBin, + MessageAuditPerformanceTest::createConcurrentAudit + }; + + BenchmarkRunner[] runners = { + (fn, ids, extra, threads, producers) -> + runPartitioned(fn, ids, threads, producers), + (fn, ids, extra, threads, producers) -> + runShared(fn, ids, threads, producers), + (fn, ids, extra, threads, producers) -> + runMixed(fn, ids, extra, threads, producers) + }; + String[] scenarioNames = { + "PARTITIONED — producers split across threads, no cross-thread contention", + "SHARED — all threads hit all producers, per-bin lock contention", + "MIXED R/W — pre-populated, 50/50 duplicate re-checks and new inserts" + }; + + System.out.println(); + System.out.println("=========================================================================="); + System.out.println(" OPTIMIZATION LAYER COMPARISON"); + System.out.println(" Layer 1: sync+LRU — LinkedHashMap + synchronized(this) + BitArrayBin"); + System.out.println(" Layer 2: CHM+syncBin — ConcurrentHashMap + per-bin synchronized + BitArrayBin"); + System.out.println(" Layer 3: CHM+AtomicBin — ConcurrentHashMap + AtomicBitArrayBin (lock-free)"); + System.out.println("=========================================================================="); + + for (var s = 0; s < runners.length; s++) { + System.out.println(); + System.out.println("=== " + scenarioNames[s] + " ==="); + + for (var producers : PRODUCER_COUNTS) { + var totalMessages = producers * MESSAGES_PER_PRODUCER; + var ids = generateStringIds(producers, MESSAGES_PER_PRODUCER); + var extraIds = generateStringIds(producers, MESSAGES_PER_PRODUCER); + + System.out.println(); + System.out.printf("--- %d producers, %,d messages ---%n", producers, totalMessages); + System.out.printf("%-8s", "Threads"); + for (var name : layerNames) { + System.out.printf(" | %14s", name); + } + System.out.printf(" | %10s | %10s%n", "Map gain", "Bin gain"); + System.out.print("--------"); + for (var i = 0; i < layerNames.length; i++) { + System.out.print("-+-" + "--------------"); + } + System.out.print("-+-----------+-----------"); + System.out.println(); + + for (var threads : THREAD_COUNTS) { + var medians = new long[layerFactories.length]; + for (var f = 0; f < layerFactories.length; f++) { + medians[f] = benchmarkWith(layerFactories[f], runners[s], + ids, extraIds, threads, producers); + } + + var best = Long.MAX_VALUE; + for (var m : medians) { + if (m < best) best = m; + } + + System.out.printf("%8d", threads); + for (var f = 0; f < layerFactories.length; f++) { + var ms = medians[f] / 1_000_000.0; + var marker = medians[f] == best ? " *" : ""; + System.out.printf(" | %11.2f ms%s", ms, marker); + } + + var mapSpeedup = (double) medians[0] / medians[1]; + var binSpeedup = (double) medians[1] / medians[2]; + System.out.printf(" | %9.1fx | %9.1fx", mapSpeedup, binSpeedup); + System.out.println(); + } + } + } + System.out.println(); + System.out.println(" * = fastest for that row"); + System.out.println(" Map gain = sync+LRU / CHM+syncBin (improvement from concurrent map)"); + System.out.println(" Bin gain = CHM+syncBin / CHM+AtomicBin (improvement from lock-free bits)"); + System.out.println(); + } + + @Test + public void compareEvictionOverhead() throws Exception { + System.out.println(); + System.out.println("=== EVICTION — producers exceed max, measures eviction overhead ==="); + + String[] evictionNames = {"sync+LRU", "CHM+Atomic", "Caffeine"}; + AuditFactory[] evictionFactories = { + MessageAuditPerformanceTest::createSyncLru, + MessageAuditPerformanceTest::createConcurrentAudit, + MessageAuditPerformanceTest::createCaffeineAudit + }; + + int[] maxProducerLimits = {16, 32, 64}; + var producerMultiplier = 4; + + for (var maxProducers : maxProducerLimits) { + var actualProducers = maxProducers * producerMultiplier; + var totalMessages = actualProducers * MESSAGES_PER_PRODUCER; + var ids = generateStringIds(actualProducers, MESSAGES_PER_PRODUCER); + var extraIds = generateStringIds(actualProducers, MESSAGES_PER_PRODUCER); + + System.out.println(); + System.out.printf("--- max=%d, actual=%d producers, %,d messages ---%n", + maxProducers, actualProducers, totalMessages); + System.out.printf("%-8s", "Threads"); + for (var name : evictionNames) { + System.out.printf(" | %12s", name); + } + System.out.println(); + System.out.print("--------"); + for (var i = 0; i < evictionNames.length; i++) { + System.out.print("-+-" + "------------"); + } + System.out.println(); + + for (var threads : THREAD_COUNTS) { + var medians = new long[evictionFactories.length]; + for (var f = 0; f < evictionFactories.length; f++) { + medians[f] = benchmarkWith(evictionFactories[f], + (fn, idArr, extra, t, p) -> runShared(fn, idArr, t, p), + ids, extraIds, threads, actualProducers, + maxProducers); + } + + var best = Long.MAX_VALUE; + for (var m : medians) { + if (m < best) best = m; + } + + System.out.printf("%8d", threads); + for (var f = 0; f < evictionFactories.length; f++) { + var ms = medians[f] / 1_000_000.0; + var marker = medians[f] == best ? " *" : ""; + System.out.printf(" | %9.2f ms%s", ms, marker); + } + System.out.println(); + } + } + System.out.println(); + System.out.println(" * = fastest for that row"); + System.out.println(); + } + + // ======================================================================== + // Correctness + // ======================================================================== + + @Test + public void verifyCorrectnessEquivalence() { + var producers = 32; + var messagesPerProducer = 1_000; + var ids = generateStringIds(producers, messagesPerProducer); + + var lru = new ActiveMQMessageAudit(2048, producers); + var concurrent = new ConcurrentMessageAudit(2048, producers); + var caffeine = new CaffeineMessageAudit(2048, producers); + + for (var p = 0; p < producers; p++) { + for (var m = 0; m < messagesPerProducer; m++) { + var lruResult = lru.isDuplicate(ids[p][m]); + var concurrentResult = concurrent.isDuplicate(ids[p][m]); + var caffeineResult = caffeine.isDuplicate(ids[p][m]); + assertEquals("ConcurrentMessageAudit mismatch at producer " + p + " message " + m, + lruResult, concurrentResult); + assertEquals("CaffeineMessageAudit mismatch at producer " + p + " message " + m, + lruResult, caffeineResult); + } + } + + for (var p = 0; p < producers; p++) { + for (var m = 0; m < messagesPerProducer; m++) { + assertTrue("Should be duplicate on second pass (LRU)", + lru.isDuplicate(ids[p][m])); + assertTrue("Should be duplicate on second pass (Concurrent)", + concurrent.isDuplicate(ids[p][m])); + assertTrue("Should be duplicate on second pass (Caffeine)", + caffeine.isDuplicate(ids[p][m])); + } + } + } + + @Test + public void verifyCorrectnessUnderConcurrency() throws Exception { + var producers = 16; + var messagesPerProducer = 2_000; + var threads = 4; + var ids = generateStringIds(producers, messagesPerProducer); + var producersPerThread = producers / threads; + + var concurrentAudit = new ConcurrentMessageAudit(2048, producers); + var duplicateCount = new AtomicInteger(0); + runConcurrentPass(concurrentAudit::isDuplicate, ids, threads, producersPerThread, duplicateCount); + assertEquals("ConcurrentMessageAudit: first pass should have zero duplicates", 0, duplicateCount.get()); + + var duplicateCount2 = new AtomicInteger(0); + runConcurrentPass(concurrentAudit::isDuplicate, ids, threads, producersPerThread, duplicateCount2); + var expectedDuplicates = producers * messagesPerProducer; + assertEquals("ConcurrentMessageAudit: second pass should all be duplicates", + expectedDuplicates, duplicateCount2.get()); + + var caffeineAudit = new CaffeineMessageAudit(2048, producers); + var caffDupCount = new AtomicInteger(0); + runConcurrentPass(caffeineAudit::isDuplicate, ids, threads, producersPerThread, caffDupCount); + assertEquals("CaffeineMessageAudit: first pass should have zero duplicates", 0, caffDupCount.get()); + + var caffDupCount2 = new AtomicInteger(0); + runConcurrentPass(caffeineAudit::isDuplicate, ids, threads, producersPerThread, caffDupCount2); + assertEquals("CaffeineMessageAudit: second pass should all be duplicates", + expectedDuplicates, caffDupCount2.get()); + } + + @Test + public void verifyEvictionBehavior() { + var maxProducers = 16; + var actualProducers = 64; + var messagesPerProducer = 100; + var ids = generateStringIds(actualProducers, messagesPerProducer); + + // --- ActiveMQMessageAudit (LRU eviction) --- + var lru = new ActiveMQMessageAudit(2048, maxProducers); + for (var p = 0; p < actualProducers; p++) { + for (var m = 0; m < messagesPerProducer; m++) { + lru.isDuplicate(ids[p][m]); + } + } + // LRU keeps the most recently accessed producers + var recentStart = actualProducers - maxProducers; + for (var p = recentStart; p < actualProducers; p++) { + assertTrue("LRU should detect duplicate for recent producer " + p, + lru.isDuplicate(ids[p][0])); + } + + // --- ConcurrentMessageAudit (bounded, hash-order eviction) --- + var concurrent = new ConcurrentMessageAudit(2048, maxProducers); + for (var p = 0; p < actualProducers; p++) { + for (var m = 0; m < messagesPerProducer; m++) { + concurrent.isDuplicate(ids[p][m]); + } + } + assertTrue("Concurrent producer count should be bounded", + concurrent.getProducerCount() <= maxProducers); + var concurrentDuplicatesDetected = 0; + for (var p = 0; p < actualProducers; p++) { + if (concurrent.isDuplicate(ids[p][0])) { + concurrentDuplicatesDetected++; + } + } + assertTrue("Concurrent should detect some duplicates among surviving producers", + concurrentDuplicatesDetected > 0); + + // --- CaffeineMessageAudit (TinyLfu eviction) --- + var caffeine = new CaffeineMessageAudit(2048, maxProducers); + for (var p = 0; p < actualProducers; p++) { + for (var m = 0; m < messagesPerProducer; m++) { + caffeine.isDuplicate(ids[p][m]); + } + } + // Caffeine eviction is asynchronous; cleanUp forces pending evictions + var caffeineCount = caffeine.getProducerCount(); + assertTrue("Caffeine producer count should be bounded (was " + caffeineCount + ")", + caffeineCount <= maxProducers * 2); + var freshIds = generateStringIds(4, messagesPerProducer); + for (var p = 0; p < 4; p++) { + assertFalse("Fresh producer should not be a duplicate", + caffeine.isDuplicate(freshIds[p][0])); + } + } + + // ======================================================================== + // Implementations under test + // ======================================================================== + + private static StringAudit createSyncLru(int auditDepth, int producers) { + var audit = new ActiveMQMessageAudit(auditDepth, producers); + return audit::isDuplicate; + } + + private static StringAudit createConcurrentAudit(int auditDepth, int producers) { + var audit = new ConcurrentMessageAudit(auditDepth, producers); + return audit::isDuplicate; + } + + private static StringAudit createCaffeineAudit(int auditDepth, int producers) { + var audit = new CaffeineMessageAudit(auditDepth, producers); + return audit::isDuplicate; + } + + private static StringAudit createSkipList(int auditDepth, int producers) { + var map = new ConcurrentSkipListMap(); + return id -> { + var seed = IdGenerator.getSeedFromId(id); + if (seed == null) return false; + var bab = map.computeIfAbsent(seed, k -> new BitArrayBin(auditDepth)); + var index = IdGenerator.getSequenceFromId(id); + if (index >= 0) { + synchronized (bab) { + return bab.setBit(index, true); + } + } + return false; + }; + } + + private static StringAudit createReadWriteLock(int auditDepth, int producers) { + var map = new HashMap(producers); + var rwLock = new ReentrantReadWriteLock(); + return id -> { + var seed = IdGenerator.getSeedFromId(id); + if (seed == null) return false; + + BitArrayBin bab; + rwLock.readLock().lock(); + try { + bab = map.get(seed); + } finally { + rwLock.readLock().unlock(); + } + if (bab == null) { + rwLock.writeLock().lock(); + try { + bab = map.get(seed); + if (bab == null) { + bab = new BitArrayBin(auditDepth); + map.put(seed, bab); + } + } finally { + rwLock.writeLock().unlock(); + } + } + + var index = IdGenerator.getSequenceFromId(id); + if (index >= 0) { + synchronized (bab) { + return bab.setBit(index, true); + } + } + return false; + }; + } + + private static StringAudit createStampedLock(int auditDepth, int producers) { + var map = new HashMap(producers); + var sl = new StampedLock(); + return id -> { + var seed = IdGenerator.getSeedFromId(id); + if (seed == null) return false; + + BitArrayBin bab; + var stamp = sl.tryOptimisticRead(); + bab = map.get(seed); + if (!sl.validate(stamp)) { + stamp = sl.readLock(); + try { + bab = map.get(seed); + } finally { + sl.unlockRead(stamp); + } + } + if (bab == null) { + stamp = sl.writeLock(); + try { + bab = map.get(seed); + if (bab == null) { + bab = new BitArrayBin(auditDepth); + map.put(seed, bab); + } + } finally { + sl.unlockWrite(stamp); + } + } + + var index = IdGenerator.getSequenceFromId(id); + if (index >= 0) { + synchronized (bab) { + return bab.setBit(index, true); + } + } + return false; + }; + } + + private static StringAudit createChmSyncBin(int auditDepth, int producers) { + var map = new ConcurrentHashMap(producers); + return id -> { + var seed = IdGenerator.getSeedFromId(id); + if (seed == null) return false; + var bab = map.computeIfAbsent(seed, k -> new BitArrayBin(auditDepth)); + var index = IdGenerator.getSequenceFromId(id); + if (index >= 0) { + synchronized (bab) { + return bab.setBit(index, true); + } + } + return false; + }; + } + + // ======================================================================== + // Benchmark runners + // ======================================================================== + + /** + * Each thread owns a disjoint slice of producers. No two threads ever + * look up the same map key or touch the same BitArrayBin. + */ + private long runPartitioned(StringAudit fn, String[][] ids, + int threads, int producers) throws Exception { + if (threads == 1) { + return runSingleThread(fn, ids, producers); + } + + var barrier = new CyclicBarrier(threads + 1); + var done = new CountDownLatch(threads); + var perThread = producers / threads; + var remainder = producers % threads; + + for (var t = 0; t < threads; t++) { + final var startP = t * perThread + Math.min(t, remainder); + final var count = perThread + (t < remainder ? 1 : 0); + new Thread(() -> { + try { + barrier.await(); + for (var p = startP; p < startP + count; p++) { + for (var m = 0; m < ids[p].length; m++) { + fn.isDuplicate(ids[p][m]); + } + } + } catch (Exception e) { + throw new RuntimeException(e); + } finally { + done.countDown(); + } + }).start(); + } + + var start = System.nanoTime(); + barrier.await(); + done.await(); + return System.nanoTime() - start; + } + + /** + * All threads iterate over ALL producers. Each thread writes a distinct + * slice of messages per producer, so no duplicate IDs, but the map is + * read-contended and every producer's BitArrayBin is lock-contended. + */ + private long runShared(StringAudit fn, String[][] ids, + int threads, int producers) throws Exception { + if (threads == 1) { + return runSingleThread(fn, ids, producers); + } + + var msgsPerProducer = ids[0].length; + var chunk = msgsPerProducer / threads; + + var barrier = new CyclicBarrier(threads + 1); + var done = new CountDownLatch(threads); + + for (var t = 0; t < threads; t++) { + final var msgStart = t * chunk; + final var msgEnd = (t == threads - 1) ? msgsPerProducer : (t + 1) * chunk; + new Thread(() -> { + try { + barrier.await(); + for (var p = 0; p < producers; p++) { + for (var m = msgStart; m < msgEnd; m++) { + fn.isDuplicate(ids[p][m]); + } + } + } catch (Exception e) { + throw new RuntimeException(e); + } finally { + done.countDown(); + } + }).start(); + } + + var start = System.nanoTime(); + barrier.await(); + done.await(); + return System.nanoTime() - start; + } + + /** + * Pre-populate the audit with {@code ids}, then all threads interleave + * duplicate re-checks ({@code ids} — reads) with new inserts + * ({@code extraIds} — writes) across all producers. + */ + private long runMixed(StringAudit fn, String[][] ids, String[][] extraIds, + int threads, int producers) throws Exception { + for (var p = 0; p < producers; p++) { + for (var m = 0; m < ids[p].length; m++) { + fn.isDuplicate(ids[p][m]); + } + } + + if (threads == 1) { + var start = System.nanoTime(); + for (var p = 0; p < producers; p++) { + for (var m = 0; m < ids[p].length; m++) { + fn.isDuplicate(ids[p][m]); + fn.isDuplicate(extraIds[p][m]); + } + } + return System.nanoTime() - start; + } + + var msgsPerProducer = ids[0].length; + var chunk = msgsPerProducer / threads; + + var barrier = new CyclicBarrier(threads + 1); + var done = new CountDownLatch(threads); + + for (var t = 0; t < threads; t++) { + final var msgStart = t * chunk; + final var msgEnd = (t == threads - 1) ? msgsPerProducer : (t + 1) * chunk; + new Thread(() -> { + try { + barrier.await(); + for (var p = 0; p < producers; p++) { + for (var m = msgStart; m < msgEnd; m++) { + fn.isDuplicate(ids[p][m]); + fn.isDuplicate(extraIds[p][m]); + } + } + } catch (Exception e) { + throw new RuntimeException(e); + } finally { + done.countDown(); + } + }).start(); + } + + var start = System.nanoTime(); + barrier.await(); + done.await(); + return System.nanoTime() - start; + } + + private long runSingleThread(StringAudit fn, String[][] ids, int producers) { + var start = System.nanoTime(); + for (var p = 0; p < producers; p++) { + for (var m = 0; m < ids[p].length; m++) { + fn.isDuplicate(ids[p][m]); + } + } + return System.nanoTime() - start; + } + + // ======================================================================== + // Benchmark harness + table printing + // ======================================================================== + + private void printBenchmark(String title, BenchmarkRunner runner) throws Exception { + System.out.println(); + System.out.println("=== " + title + " ==="); + + for (var producers : PRODUCER_COUNTS) { + var totalMessages = producers * MESSAGES_PER_PRODUCER; + var ids = generateStringIds(producers, MESSAGES_PER_PRODUCER); + var extraIds = generateStringIds(producers, MESSAGES_PER_PRODUCER); + + System.out.println(); + System.out.printf("--- %d producers, %,d messages ---%n", producers, totalMessages); + System.out.printf("%-8s", "Threads"); + for (var name : NAMES) { + System.out.printf(" | %12s", name); + } + System.out.println(); + System.out.print("--------"); + for (var i = 0; i < NAMES.length; i++) { + System.out.print("-+-" + "------------"); + } + System.out.println(); + + for (var threads : THREAD_COUNTS) { + var medians = new long[FACTORIES.length]; + for (var f = 0; f < FACTORIES.length; f++) { + medians[f] = benchmarkWith(FACTORIES[f], runner, ids, extraIds, threads, producers); + } + + var best = Long.MAX_VALUE; + for (var m : medians) { + if (m < best) best = m; + } + + System.out.printf("%8d", threads); + for (var f = 0; f < FACTORIES.length; f++) { + var ms = medians[f] / 1_000_000.0; + var marker = medians[f] == best ? " *" : ""; + System.out.printf(" | %9.2f ms%s", ms, marker); + } + System.out.println(); + } + } + System.out.println(); + System.out.println(" * = fastest for that row"); + System.out.println(); + } + + private long benchmarkWith(AuditFactory factory, BenchmarkRunner runner, + String[][] ids, String[][] extraIds, + int threads, int producers) throws Exception { + return benchmarkWith(factory, runner, ids, extraIds, threads, producers, producers); + } + + private long benchmarkWith(AuditFactory factory, BenchmarkRunner runner, + String[][] ids, String[][] extraIds, + int threads, int producers, + int maxProducers) throws Exception { + var times = new long[WARMUP_ITERATIONS + MEASURE_ITERATIONS]; + for (var i = 0; i < times.length; i++) { + var audit = factory.create(2048, maxProducers); + times[i] = runner.run(audit, ids, extraIds, threads, producers); + } + return median(times, WARMUP_ITERATIONS); + } + + private void runConcurrentPass(StringAudit fn, String[][] ids, + int threads, int producersPerThread, + AtomicInteger counter) throws Exception { + var barrier = new CyclicBarrier(threads); + var done = new CountDownLatch(threads); + for (var t = 0; t < threads; t++) { + final var startProducer = t * producersPerThread; + final var endProducer = startProducer + producersPerThread; + new Thread(() -> { + try { + barrier.await(); + for (var p = startProducer; p < endProducer; p++) { + for (var m = 0; m < ids[p].length; m++) { + if (fn.isDuplicate(ids[p][m])) { + counter.incrementAndGet(); + } + } + } + } catch (Exception e) { + throw new RuntimeException(e); + } finally { + done.countDown(); + } + }).start(); + } + done.await(); + } + + // ======================================================================== + // Helpers + // ======================================================================== + + private String[][] generateStringIds(int producers, int messagesPerProducer) { + var ids = new String[producers][messagesPerProducer]; + for (var p = 0; p < producers; p++) { + var gen = new IdGenerator(); + for (var m = 0; m < messagesPerProducer; m++) { + ids[p][m] = gen.generateId(); + } + } + return ids; + } + + private long median(long[] times, int skipFirst) { + var measured = Arrays.copyOfRange(times, skipFirst, times.length); + Arrays.sort(measured); + return measured[measured.length / 2]; + } +} diff --git a/activemq-client/src/test/java/org/apache/activemq/util/AtomicBitArrayBinTest.java b/activemq-client/src/test/java/org/apache/activemq/util/AtomicBitArrayBinTest.java new file mode 100644 index 00000000000..60966f26b5f --- /dev/null +++ b/activemq-client/src/test/java/org/apache/activemq/util/AtomicBitArrayBinTest.java @@ -0,0 +1,444 @@ +/** + * 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.util; + +import static org.junit.Assert.*; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Test; + +public class AtomicBitArrayBinTest { + + @Test + public void testSetAndGetBit() { + var bin = new AtomicBitArrayBin(512); + + assertFalse("first set should return false", bin.setBit(0, true)); + assertTrue("second set should return true (duplicate)", bin.setBit(0, true)); + + assertFalse("first set of 1 should return false", bin.setBit(1, true)); + assertTrue("bit 0 should still be set", bin.getBit(0)); + assertTrue("bit 1 should be set", bin.getBit(1)); + assertFalse("bit 2 should not be set", bin.getBit(2)); + } + + @Test + public void testClearBit() { + var bin = new AtomicBitArrayBin(512); + + bin.setBit(10, true); + assertTrue("bit should be set", bin.getBit(10)); + + assertTrue("clear should return true (was set)", bin.setBit(10, false)); + assertFalse("bit should be cleared", bin.getBit(10)); + + assertFalse("clear again should return false (was already clear)", bin.setBit(10, false)); + } + + @Test + public void testSequentialBits() { + var bin = new AtomicBitArrayBin(2048); + + for (int i = 0; i < 1000; i++) { + assertFalse("first set of " + i + " should return false", bin.setBit(i, true)); + } + for (int i = 0; i < 1000; i++) { + assertTrue("bit " + i + " should be set", bin.getBit(i)); + assertTrue("duplicate set of " + i + " should return true", bin.setBit(i, true)); + } + } + + @Test + public void testWindowSlide() { + var windowSize = 128; + var bin = new AtomicBitArrayBin(windowSize); + + bin.setBit(0, true); + bin.setBit(63, true); + assertTrue(bin.getBit(0)); + assertTrue(bin.getBit(63)); + + // Set a bit far beyond the window to force advancement + var farIndex = windowSize + 128; + assertFalse(bin.setBit(farIndex, true)); + assertTrue(bin.getBit(farIndex)); + + // Original bits should now be "behind window" — getBit returns true + assertTrue("behind-window bits return true", bin.getBit(0)); + assertTrue("behind-window bits return true", bin.getBit(63)); + } + + @Test + public void testBehindWindowReturnsTrueOnSet() { + var bin = new AtomicBitArrayBin(128); + + bin.setBit(0, true); + // Advance window far past index 0 + bin.setBit(500, true); + + // Setting a behind-window bit should return true (treated as already set) + assertTrue("behind-window setBit should return true", bin.setBit(0, true)); + } + + @Test + public void testCrossBoundaryBits() { + var bin = new AtomicBitArrayBin(512); + + // Set bits across multiple 64-bit slot boundaries + for (int i = 0; i < 256; i += 63) { + assertFalse(bin.setBit(i, true)); + } + for (int i = 0; i < 256; i += 63) { + assertTrue("bit " + i + " should be set", bin.getBit(i)); + } + } + + @Test + public void testIsInOrder() { + var bin = new AtomicBitArrayBin(512); + + assertTrue("first message is always in order", bin.isInOrder(0)); + assertTrue("sequential is in order", bin.isInOrder(1)); + assertTrue("sequential is in order", bin.isInOrder(2)); + assertFalse("gap breaks order", bin.isInOrder(5)); + assertTrue("next after gap is in order", bin.isInOrder(6)); + } + + @Test + public void testGetLastSetIndex() { + var bin = new AtomicBitArrayBin(512); + + assertEquals(-1, bin.getLastSetIndex()); + + bin.setBit(10, true); + assertEquals(10, bin.getLastSetIndex()); + + bin.setBit(100, true); + assertEquals(100, bin.getLastSetIndex()); + + bin.setBit(50, true); + assertEquals("last set index should be highest", 100, bin.getLastSetIndex()); + } + + @Test + public void testLargeSequenceNumbers() { + var bin = new AtomicBitArrayBin(2048); + + var base = 1_000_000L; + for (var i = base; i < base + 500; i++) { + assertFalse(bin.setBit(i, true)); + } + for (var i = base; i < base + 500; i++) { + assertTrue(bin.getBit(i)); + assertTrue(bin.setBit(i, true)); + } + } + + @Test + public void testEquivalenceWithBitArrayBin() { + var windowSize = 2048; + var original = new BitArrayBin(windowSize); + var atomic = new AtomicBitArrayBin(windowSize); + + // Sequential inserts + for (var i = 0; i < 3000; i++) { + var origResult = original.setBit(i, true); + var atomicResult = atomic.setBit(i, true); + assertEquals("setBit(" + i + ") mismatch", origResult, atomicResult); + } + + // Duplicate checks on values still in window + for (var i = 2000; i < 3000; i++) { + var origResult = original.setBit(i, true); + var atomicResult = atomic.setBit(i, true); + assertEquals("duplicate setBit(" + i + ") mismatch", origResult, atomicResult); + } + } + + @Test + public void testEquivalenceWithBitArrayBinSparseIndices() { + var windowSize = 512; + var original = new BitArrayBin(windowSize); + var atomic = new AtomicBitArrayBin(windowSize); + + // Sparse indices within a single window + int[] indices = {0, 1, 63, 64, 65, 127, 128, 200, 300, 400, 511}; + for (var idx : indices) { + assertEquals("first set " + idx, original.setBit(idx, true), atomic.setBit(idx, true)); + } + for (var idx : indices) { + assertEquals("dup set " + idx, original.setBit(idx, true), atomic.setBit(idx, true)); + } + } + + @Test + public void testConcurrentSetsNoDuplicateLoss() throws Exception { + var threadCount = 8; + var messagesPerThread = 10_000; + var bin = new AtomicBitArrayBin(threadCount * messagesPerThread); + + var executor = Executors.newFixedThreadPool(threadCount); + var barrier = new CyclicBarrier(threadCount); + var duplicateCount = new AtomicInteger(0); + + var futures = new ArrayList>(); + for (var t = 0; t < threadCount; t++) { + var threadId = t; + futures.add(executor.submit(() -> { + try { + barrier.await(); + } catch (Exception e) { + throw new RuntimeException(e); + } + for (var i = 0; i < messagesPerThread; i++) { + // Each thread gets unique indices (no overlap) + var index = (long) threadId * messagesPerThread + i; + var wasDuplicate = bin.setBit(index, true); + if (wasDuplicate) { + duplicateCount.incrementAndGet(); + } + } + })); + } + + for (var f : futures) f.get(); + executor.shutdown(); + + assertEquals("no false duplicates with disjoint indices", 0, duplicateCount.get()); + + // Verify all bits are set + for (var t = 0; t < threadCount; t++) { + for (var i = 0; i < messagesPerThread; i++) { + var index = (long) t * messagesPerThread + i; + assertTrue("bit " + index + " should be set", bin.getBit(index)); + } + } + } + + @Test + public void testConcurrentDuplicateDetection() throws Exception { + var threadCount = 8; + var messageCount = 5_000; + var bin = new AtomicBitArrayBin(messageCount + 1000); + + // Pre-populate + for (var i = 0; i < messageCount; i++) { + bin.setBit(i, true); + } + + var executor = Executors.newFixedThreadPool(threadCount); + var barrier = new CyclicBarrier(threadCount); + var missedDuplicates = new AtomicInteger(0); + + var futures = new ArrayList>(); + for (var t = 0; t < threadCount; t++) { + futures.add(executor.submit(() -> { + try { + barrier.await(); + } catch (Exception e) { + throw new RuntimeException(e); + } + for (var i = 0; i < messageCount; i++) { + var wasDuplicate = bin.setBit(i, true); + if (!wasDuplicate) { + missedDuplicates.incrementAndGet(); + } + } + })); + } + + for (var f : futures) f.get(); + executor.shutdown(); + + assertEquals("all re-sets should detect duplicate", 0, missedDuplicates.get()); + } + + @Test + public void testConcurrentSharedIndices() throws Exception { + var threadCount = 8; + var messageCount = 10_000; + var bin = new AtomicBitArrayBin(messageCount + 1000); + + var executor = Executors.newFixedThreadPool(threadCount); + var barrier = new CyclicBarrier(threadCount); + + // All threads write the SAME indices. Exactly one thread should see + // "not duplicate" per index; all others should see "duplicate". + var firstSetBy = ConcurrentHashMap.newKeySet(); + + var futures = new ArrayList>(); + for (var t = 0; t < threadCount; t++) { + futures.add(executor.submit(() -> { + try { + barrier.await(); + } catch (Exception e) { + throw new RuntimeException(e); + } + for (var i = 0; i < messageCount; i++) { + var wasDuplicate = bin.setBit(i, true); + if (!wasDuplicate) { + var added = firstSetBy.add(i); + assertTrue("only one thread should be first to set " + i, added); + } + } + })); + } + + for (var f : futures) f.get(); + executor.shutdown(); + + assertEquals("every index should have exactly one first-setter", + messageCount, firstSetBy.size()); + } + + @Test + public void testConcurrentWindowAdvancement() throws Exception { + var threadCount = 4; + var windowSize = 256; + var bin = new AtomicBitArrayBin(windowSize); + + var executor = Executors.newFixedThreadPool(threadCount); + var barrier = new CyclicBarrier(threadCount); + + // Threads write to different ranges, some forcing window advancement + var futures = new ArrayList>(); + for (var t = 0; t < threadCount; t++) { + final var threadId = t; + futures.add(executor.submit(() -> { + try { + barrier.await(); + } catch (Exception e) { + throw new RuntimeException(e); + } + // Each thread writes a range that may overlap and force advancement + var base = (long) threadId * 200; + for (var i = base; i < base + 500; i++) { + bin.setBit(i, true); + } + })); + } + + for (var f : futures) f.get(); + executor.shutdown(); + + // Verify the latest values are correct (earlier ones may have been evicted) + var highestBase = (long)(threadCount - 1) * 200; + for (var i = highestBase; i < highestBase + 500; i++) { + assertTrue("bit " + i + " should be set or behind window", bin.getBit(i)); + } + } + + @Test + public void testSlotReuse() { + // Small window to force rapid slot reuse + var bin = new AtomicBitArrayBin(64); + + // First window: set some bits + for (var i = 0; i < 64; i++) { + bin.setBit(i, true); + } + + // Advance past the first window + bin.setBit(200, true); + + // The new bit should be set + assertTrue(bin.getBit(200)); + + // Bits in the new range that weren't explicitly set should be clear + assertFalse(bin.getBit(201)); + } + + @Test + public void testRollback() { + var bin = new AtomicBitArrayBin(512); + + bin.setBit(42, true); + assertTrue(bin.getBit(42)); + + // Rollback (clear) + bin.setBit(42, false); + assertFalse(bin.getBit(42)); + + // Should be able to re-set after rollback + assertFalse("re-set after rollback should return false", bin.setBit(42, true)); + assertTrue(bin.getBit(42)); + } + + @Test + public void testCapacityCalculation() { + // windowSize 2048 → capacity = ((2048+1)/64)+1 = 33 + var bin = new AtomicBitArrayBin(2048); + assertEquals(33, bin.getCapacity()); + + // windowSize 64 → capacity = ((64+1)/64)+1 = 2 + var bin2 = new AtomicBitArrayBin(64); + assertEquals(2, bin2.getCapacity()); + + // windowSize 1 → capacity = max(1, ((1+1)/64)+1) = 1 + var bin3 = new AtomicBitArrayBin(1); + assertEquals(1, bin3.getCapacity()); + } + + @Test + public void testStressSequentialThenConcurrentVerify() throws Exception { + var windowSize = 4096; + var messageCount = 3000; + var bin = new AtomicBitArrayBin(windowSize); + + // Sequential population + for (var i = 0; i < messageCount; i++) { + assertFalse(bin.setBit(i, true)); + } + + // Concurrent verification + var threadCount = 8; + var executor = Executors.newFixedThreadPool(threadCount); + var barrier = new CyclicBarrier(threadCount); + var failures = new AtomicInteger(0); + + var futures = new ArrayList>(); + for (var t = 0; t < threadCount; t++) { + futures.add(executor.submit(() -> { + try { + barrier.await(); + } catch (Exception e) { + throw new RuntimeException(e); + } + for (var i = 0; i < messageCount; i++) { + if (!bin.getBit(i)) { + failures.incrementAndGet(); + } + } + })); + } + + for (var f : futures) f.get(); + executor.shutdown(); + + assertEquals("concurrent reads should see all set bits", 0, failures.get()); + } +} diff --git a/activemq-client/src/test/resources/benchmark-results.html b/activemq-client/src/test/resources/benchmark-results.html new file mode 100644 index 00000000000..8c4857a5753 --- /dev/null +++ b/activemq-client/src/test/resources/benchmark-results.html @@ -0,0 +1,487 @@ + + + + + +Message Audit Benchmark Results + + + +
+
+

MessageAudit Benchmark Results

+

6 strategies, 3 concurrency scenarios, producer counts {1, 8, 16, 64, 128}, thread counts {1, 2, 4, 8}. All times in milliseconds (median of 5 measured iterations after 3 warmup).

+
+ +
+
+
Overall Winner
+
CHM+Atomic
+
Fastest in Partitioned & Shared across all configs
+
+
+
Max Speedup vs Baseline
+
8.7x
+
128 prod / 8 threads, Shared scenario
+
+
+
Eviction Winner
+
Caffeine
+
Dominates at high concurrency under eviction pressure
+
+
+
Worst Performer
+
RWLock
+
Up to 471ms at 128 prod / 8 threads Mixed R/W
+
+
+ +
+ + + + + +
+ +
+
+
+
+
+ +
+
Fastest
+
Slowest
+
+
+ + + +