diff --git a/activemq-client/pom.xml b/activemq-client/pom.xml index 57685b2af13..eee41c9322a 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 + + @@ -206,6 +212,7 @@ !org.apache.activemq*, !com.google.errorprone.annotations, !com.google.errorprone.annotations.concurrent, + com.github.benmanes.caffeine.*;resolution:="optional", com.thoughtworks.xstream.*;resolution:="optional", javax.jmdns.*;resolution:="optional", * 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..007b740897e --- /dev/null +++ b/activemq-client/src/main/java/org/apache/activemq/CaffeineMessageAudit.java @@ -0,0 +1,199 @@ +/** + * 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 final 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; + // Resize in place: rebuilding and swapping the cache would lose + // entries inserted concurrently with the copy + cache.policy().eviction() + .ifPresent(eviction -> eviction.setMaximum(maximumNumberOfProducersToTrack)); + } + + public boolean isDuplicate(String id) { + var seed = IdGenerator.getSeedFromId(id); + if (seed == null) { + return false; + } + var index = IdGenerator.getSequenceFromId(id); + if (index < 0) { + return false; + } + return markSeen(seed, index); + } + + public boolean isDuplicate(final MessageId id) { + if (id == null) { + return false; + } + var pid = id.getProducerId(); + if (pid == null) { + return false; + } + return markSeen(pid.toString(), id.getProducerSequenceId()); + } + + /** + * Record the index against the bin for the key. If the bin is evicted + * between lookup and the bit write, the write lands in an orphaned bin + * and the next occurrence of the id is not detected as a duplicate. + * This residual race is accepted: eviction only fires under producer + * oversubscription, membership re-check-and-retry in that regime was + * measured at a 2-5x throughput cost while the recreated entry is + * immediately evicted again, and the failure direction (duplicate + * redelivery) is tolerable under at-least-once delivery. + */ + private boolean markSeen(String key, long index) { + return cache.get(key, k -> new AtomicBitArrayBin(auditDepth)).setBit(index, true); + } + + public void rollback(final MessageId id) { + if (id == null) { + return; + } + var pid = id.getProducerId(); + if (pid == null) { + return; + } + var seqId = id.getProducerSequenceId(); + cache.asMap().computeIfPresent(pid.toString(), (k, bab) -> { + bab.setBit(seqId, false); + return bab; + }); + } + + public void rollback(final String id) { + var seed = IdGenerator.getSeedFromId(id); + if (seed == null) { + return; + } + var index = IdGenerator.getSequenceFromId(id); + if (index < 0) { + return; + } + cache.asMap().computeIfPresent(seed, (k, bab) -> { + bab.setBit(index, false); + return bab; + }); + } + + 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..7df758e33de --- /dev/null +++ b/activemq-client/src/main/java/org/apache/activemq/ConcurrentMessageAudit.java @@ -0,0 +1,222 @@ +/** + * 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 index = IdGenerator.getSequenceFromId(id); + if (index < 0) { + return false; + } + return markSeen(seed, index); + } + + public boolean isDuplicate(final MessageId id) { + if (id == null) { + return false; + } + var pid = id.getProducerId(); + if (pid == null) { + return false; + } + return markSeen(pid.toString(), id.getProducerSequenceId()); + } + + /** + * Record the index against the bin for the key. If the bin is evicted + * between lookup and the bit write, the write lands in an orphaned bin + * and the next occurrence of the id is not detected as a duplicate. + * This residual race is accepted: eviction only fires under producer + * oversubscription, membership re-check-and-retry in that regime was + * measured at a 2-5x throughput cost while the recreated entry is + * immediately evicted again, and the failure direction (duplicate + * redelivery) is tolerable under at-least-once delivery. + */ + private boolean markSeen(String key, long index) { + return getOrCreate(key).setBit(index, true); + } + + public void rollback(final MessageId id) { + if (id == null) { + return; + } + var pid = id.getProducerId(); + if (pid == null) { + return; + } + var seqId = id.getProducerSequenceId(); + map.computeIfPresent(pid.toString(), (k, bab) -> { + bab.setBit(seqId, false); + return bab; + }); + } + + public void rollback(final String id) { + var seed = IdGenerator.getSeedFromId(id); + if (seed == null) { + return; + } + var index = IdGenerator.getSequenceFromId(id); + if (index < 0) { + return; + } + map.computeIfPresent(seed, (k, bab) -> { + bab.setBit(index, false); + return bab; + }); + } + + 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; + } + + /** + * Trim the map to the configured maximum. Eviction follows CHM iteration + * order (approximate FIFO, not LRU). Concurrent callers may transiently + * remove more entries than strictly required; the bound is approximate + * by design and self-corrects on subsequent inserts. + */ + 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..281994fe7d7 --- /dev/null +++ b/activemq-client/src/main/java/org/apache/activemq/util/AtomicBitArrayBin.java @@ -0,0 +1,258 @@ +/** + * 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.AtomicReferenceArray; + +/** + * A lock-free replacement for {@link BitArrayBin} that uses a ring buffer of + * immutable-epoch slot holders 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 ring of {@link Slot} objects 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}. + * Each slot object carries a fixed epoch identifying which absolute 64-bit block + * it holds. The window covers exactly {@code capacity} consecutive epochs + * {@code [originEpoch, originEpoch + capacity)}, so two distinct in-window epochs + * never share a ring position. + * + *

When the window advances, an evicted slot is recycled by CAS-replacing the + * whole slot object with a fresh one for the new epoch. Because the epoch is + * immutable and the bits live inside the slot object, a writer that captured the + * old slot can only ever mutate the orphaned object — it can never corrupt the + * new epoch's bits. Every mutation is therefore linearizable: it either lands in + * the live slot, or in an orphan, which is equivalent to the operation having + * completed just before the window advanced. + * + *

Concurrency guarantees

+ * + * + *

Behind-window semantics

+ * + *

Matching {@link BitArrayBin}: {@code setBit} for an index behind the window + * is a no-op returning {@code false} (the message is accepted rather than + * reported as a duplicate, favoring at-least-once delivery over message loss), + * while {@code getBit} returns {@code true} (the range is assumed seen). + */ +public class AtomicBitArrayBin { + + static final int LONG_SIZE = 64; + + private static final long UNINITIALIZED = -1L; + + private final int capacity; + private final AtomicReferenceArray slots; + private final AtomicLong origin; + private final AtomicLong lastInOrderBit; + + /** + * One 64-bit block of the ring. The epoch is fixed at construction; only + * the bits mutate. Recycling replaces the whole object with a single + * reference CAS, so the epoch/bits pair a reader observes is always + * internally consistent. + */ + private static final class Slot { + final long epoch; + final AtomicLong bits = new AtomicLong(); + + Slot(long epoch) { + this.epoch = epoch; + } + } + + public AtomicBitArrayBin(int windowSize) { + capacity = Math.max(1, ((windowSize + 1) / LONG_SIZE) + 1); + slots = new AtomicReferenceArray<>(capacity); + 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); + * always false for an index behind the window (no-op) + */ + 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 false; + } + + if (epoch >= originEpoch + capacity) { + advanceOrigin(orig, epoch); + continue; + } + + var slot = slotFor(epoch); + if (slot == null) { + // Recycled to a newer epoch, which proves this epoch is now + // behind the window: re-read the origin and reclassify + continue; + } + + var mask = 1L << (int)(index % LONG_SIZE); + while (true) { + var oldBits = slot.bits.get(); + var wasSet = (oldBits & mask) != 0; + + if (value) { + if (wasSet) return true; + if (slot.bits.compareAndSet(oldBits, oldBits | mask)) return false; + } else { + if (!wasSet) return false; + if (slot.bits.compareAndSet(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 slot = slots.get((int)(epoch % capacity)); + if (slot == null) { + return false; + } + if (slot.epoch != epoch) { + // Newer epoch means this one was evicted (behind-window semantics); + // an older epoch means this block was never written + return slot.epoch > epoch; + } + return (slot.bits.get() & (1L << (int)(index % LONG_SIZE))) != 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; always false for a + * negative index (which also leaves order tracking untouched) + */ + public boolean isInOrder(long index) { + if (index < 0) { + return false; + } + var prev = lastInOrderBit.getAndSet(index); + return prev == UNINITIALIZED || prev + 1 == index; + } + + /** + * Get the index of the highest set bit across all valid slots. + * + *

Snapshot semantics: concurrent writers may set higher bits while the + * scan runs; the result reflects some recent consistent state. + * + * @return the highest set bit index, or -1 if no bits are set + */ + public long getLastSetIndex() { + var originEpoch = origin.get() / LONG_SIZE; + + for (var offset = capacity - 1; offset >= 0; offset--) { + var epoch = originEpoch + offset; + var slot = slots.get((int)(epoch % capacity)); + if (slot == null || slot.epoch != epoch) { + continue; + } + var slotBits = slot.bits.get(); + 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); + } + } + + /** + * Resolve the live slot for an epoch, installing a fresh one if the ring + * position holds an older epoch (or none). Installed epochs per position + * are strictly increasing, so the loop terminates: every CAS failure means + * another thread installed a newer slot. + * + * @return the slot owning this epoch, or null if the position was already + * recycled to a newer epoch (proving this epoch is behind the window) + */ + private Slot slotFor(long epoch) { + var ringPos = (int)(epoch % capacity); + while (true) { + var slot = slots.get(ringPos); + if (slot != null) { + if (slot.epoch == epoch) { + return slot; + } + if (slot.epoch > epoch) { + return null; + } + } + var fresh = new Slot(epoch); + if (slots.compareAndSet(ringPos, slot, fresh)) { + return fresh; + } + } + } +} 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..9558d25fb5b --- /dev/null +++ b/activemq-client/src/test/java/org/apache/activemq/MessageAuditPerformanceTest.java @@ -0,0 +1,845 @@ +/** + * 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.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.concurrent.locks.StampedLock; +import java.util.function.IntConsumer; + +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])); + } + } + + @Test + public void verifySetMaximumDuringTrafficPreservesState() throws Exception { + var producers = 8; + var messagesPerProducer = 2_000; + + var concurrent = new ConcurrentMessageAudit(2048, 64); + runSetMaxDuringTraffic("ConcurrentMessageAudit", concurrent::isDuplicate, + concurrent::setMaximumNumberOfProducersToTrack, + generateStringIds(producers, messagesPerProducer)); + + var caffeine = new CaffeineMessageAudit(2048, 64); + runSetMaxDuringTraffic("CaffeineMessageAudit", caffeine::isDuplicate, + caffeine::setMaximumNumberOfProducersToTrack, + generateStringIds(producers, messagesPerProducer)); + } + + /** + * Churns setMaximumNumberOfProducersToTrack between 32 and 64 while ids + * are inserted. With only 8 producers no eviction is ever legitimate, so + * every id must be detected as a duplicate afterwards — any miss means + * the resize lost audit state. + */ + private void runSetMaxDuringTraffic(String label, StringAudit fn, IntConsumer setMax, + String[][] ids) throws Exception { + var producers = ids.length; + var threads = 2; + var stop = new AtomicBoolean(false); + var resizer = new Thread(() -> { + var flip = false; + while (!stop.get()) { + setMax.accept(flip ? 32 : 64); + flip = !flip; + } + }); + resizer.start(); + try { + var firstPass = new AtomicInteger(); + runConcurrentPass(fn, ids, threads, producers / threads, firstPass); + } finally { + stop.set(true); + resizer.join(); + } + + var duplicates = new AtomicInteger(); + runConcurrentPass(fn, ids, threads, producers / threads, duplicates); + assertEquals(label + ": resize during traffic must not lose audit state", + producers * ids[0].length, duplicates.get()); + } + + // ======================================================================== + // 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..62f12032e2a --- /dev/null +++ b/activemq-client/src/test/java/org/apache/activemq/util/AtomicBitArrayBinTest.java @@ -0,0 +1,671 @@ +/** + * 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.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + +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 testBehindWindowSetIsNoOpAndGetReturnsTrue() { + var bin = new AtomicBitArrayBin(128); + + bin.setBit(0, true); + // Advance window far past index 0 + bin.setBit(500, true); + + // Behind-window setBit is a no-op returning false (BitArrayBin parity): + // a late message is accepted rather than dropped as a duplicate + assertFalse("behind-window setBit should be a no-op returning false", + bin.setBit(0, true)); + + // Behind-window getBit still reports the range as seen + assertTrue("behind-window getBit should return true", bin.getBit(0)); + } + + @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 testBehindWindowSetBitMatchesBitArrayBin() { + var windowSize = 128; + var original = new BitArrayBin(windowSize); + var atomic = new AtomicBitArrayBin(windowSize); + + assertEquals(original.setBit(0, true), atomic.setBit(0, true)); + // Advance both windows far past index 0 (both land on origin 320) + assertEquals(original.setBit(500, true), atomic.setBit(500, true)); + + // Behind-window semantics must match BitArrayBin: setBit is a no-op + // returning false. Reporting "duplicate" here would drop a legitimate + // late message. + assertEquals("behind-window setBit(true) parity", + original.setBit(0, true), atomic.setBit(0, true)); + assertEquals("behind-window setBit(false) parity", + original.setBit(0, false), atomic.setBit(0, false)); + assertEquals("behind-window getBit parity", + original.getBit(0), atomic.getBit(0)); + } + + @Test + public void testRollbackThenResendAfterWindowJump() { + var bin = new AtomicBitArrayBin(128); + + assertFalse("first delivery is not a duplicate", bin.setBit(100, true)); + + // An unrelated large sequence advances the window past index 100 + bin.setBit(10_000, true); + + // Rollback of the failed delivery: behind window, no-op + bin.setBit(100, false); + + // The resend must NOT be reported as a duplicate — that would make + // the rolled-back message undeliverable forever + assertFalse("resend after rollback must not be dropped as duplicate", + bin.setBit(100, true)); + } + + @Test + public void testIsInOrderNegativeIndexDoesNotPoison() { + var bin = new AtomicBitArrayBin(512); + + assertTrue(bin.isInOrder(0)); + assertTrue(bin.isInOrder(1)); + + // -1 is what IdGenerator.getSequenceFromId returns for unparseable ids. + // It must not be reported in-order and must not reset order tracking + // to the initial "anything is in order" state. + assertFalse("negative index is never in order", bin.isInOrder(-1)); + assertFalse("gap after invalid index must not be in order", bin.isInOrder(50)); + assertTrue("sequence resumes after the gap", bin.isInOrder(51)); + } + + /** + * Cross-epoch ABA plant detector. + * + *

capacity=2 ring: epochs f and f-2 share a slot. Writers hammer the + * trailing in-window epoch (frontier-1) at offsets 0..15 while the jumper + * advances the window one epoch per round. A writer that is mid-setBit + * when its slot is recycled must NOT leave a bit in the new epoch. + * + *

Writers never target the leading epoch f, so any bit in offsets 0..15 + * of a freshly recycled leading epoch is a planted bit — a future message + * at that index would be falsely reported duplicate and dropped. + */ + @Test(timeout = 60_000) + public void testNoCrossEpochBitPlantDuringRecycle() throws Exception { + var bin = new AtomicBitArrayBin(64); + assertEquals("test requires a 2-slot ring", 2, bin.getCapacity()); + + var rounds = 50_000; + var writerThreads = 2; + var frontier = new AtomicLong(1); + var stop = new AtomicBoolean(false); + var executor = Executors.newFixedThreadPool(writerThreads); + + var writers = new ArrayList>(); + for (var t = 0; t < writerThreads; t++) { + writers.add(executor.submit(() -> { + while (!stop.get()) { + var trailing = frontier.get() - 1; + var base = trailing * 64; + for (var o = 0; o < 16; o++) { + bin.setBit(base + o, true); + } + } + })); + } + + var plantedAt = -1L; + try { + for (var f = 2L; f <= rounds; f++) { + // Recycles the slot previously holding epoch f-2 while + // writers may be mid-flight on trailing epochs + bin.setBit(f * 64 + 63, true); + + // Probe the freshly recycled epoch before writers can + // legitimately reach it (they only ever target frontier-1 < f) + for (var scan = 0; scan < 3 && plantedAt < 0; scan++) { + for (var o = 0; o < 16; o++) { + if (bin.getBit(f * 64 + o)) { + plantedAt = f * 64 + o; + break; + } + } + Thread.onSpinWait(); + } + if (plantedAt >= 0) { + break; + } + frontier.set(f); + } + } finally { + stop.set(true); + for (var w : writers) w.get(); + executor.shutdown(); + } + + assertEquals("bit planted into recycled epoch (cross-epoch ABA) at index " + + plantedAt, -1L, plantedAt); + } + + /** + * Fabricated-index detector for getLastSetIndex. + * + *

Every written index is logged before the write. A concurrent scanner + * asserts every getLastSetIndex() result was actually written. A torn + * epoch/bits read across a slot recycle reconstructs an index (old epoch + * × new bits) that exists in neither. + */ + @Test(timeout = 60_000) + public void testGetLastSetIndexNeverFabricatesIndex() throws Exception { + var bin = new AtomicBitArrayBin(64); + assertEquals("test requires a 2-slot ring", 2, bin.getCapacity()); + + var rounds = 50_000; + var frontier = new AtomicLong(1); + var stop = new AtomicBoolean(false); + var writeLog = ConcurrentHashMap.newKeySet(); + var executor = Executors.newFixedThreadPool(2); + + var fabricated = new AtomicLong(-1); + + // Writer: hammers the trailing epoch, offset encodes the epoch (e % 16) + var writer = executor.submit(() -> { + while (!stop.get()) { + var trailing = frontier.get() - 1; + var index = trailing * 64 + (trailing % 16); + writeLog.add(index); + bin.setBit(index, true); + } + }); + + // Scanner: every result of getLastSetIndex must have been written + var scanner = executor.submit(() -> { + while (!stop.get()) { + var last = bin.getLastSetIndex(); + if (last >= 0 && !writeLog.contains(last)) { + fabricated.compareAndSet(-1, last); + return; + } + } + }); + + try { + // Jumper: offset also encodes the epoch (32 + f % 16), so an old + // epoch paired with new bits yields an unlogged index + for (var f = 2L; f <= rounds && fabricated.get() < 0; f++) { + var index = f * 64 + 32 + (f % 16); + writeLog.add(index); + bin.setBit(index, true); + frontier.set(f); + } + } finally { + stop.set(true); + writer.get(); + scanner.get(); + executor.shutdown(); + } + + assertEquals("getLastSetIndex returned an index that was never written: " + + fabricated.get(), -1L, fabricated.get()); + } + + @Test(timeout = 30_000) + public void testCapacityOneConcurrentChurn() throws Exception { + // Degenerate single-slot ring: every 64-index step recycles the slot + var bin = new AtomicBitArrayBin(1); + assertEquals(1, bin.getCapacity()); + + var threadCount = 2; + var perThread = 20_000; + var executor = Executors.newFixedThreadPool(threadCount); + var barrier = new CyclicBarrier(threadCount); + + 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); + } + for (var i = 0; i < perThread; i++) { + bin.setBit((long) i * threadCount + threadId, true); + } + })); + } + + for (var f : futures) f.get(); + executor.shutdown(); + + // Terminates without hang (timeout) and the ring is still coherent + var last = bin.getLastSetIndex(); + assertTrue("last set index should be near the top of the range, was " + last, + last >= (long) (perThread - 64) * threadCount - 64); + } + + @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..d092f4c7d09 --- /dev/null +++ b/activemq-client/src/test/resources/benchmark-results.html @@ -0,0 +1,503 @@ + + + + + + +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). Measured after the thread-safety review hardening (immutable slot holders in AtomicBitArrayBin).

+
+ +
+
+
Overall Winner
+
CHM+Atomic
+
Fastest in Partitioned & Shared across all configs
+
+
+
Max Speedup vs Baseline
+
9.0x
+
128 prod / 8 threads, Shared scenario
+
+
+
Eviction Winner
+
Caffeine
+
Dominates at high concurrency under eviction pressure
+
+
+
Worst Performer
+
RWLock
+
Up to 409ms at 128 prod / 8 threads Mixed R/W
+
+
+ +
+ + + + + +
+ +
+
+
+
+
+ +
+
Fastest
+
Slowest
+
+
+ + + + diff --git a/activemq-osgi/pom.xml b/activemq-osgi/pom.xml index eb70100180f..ab28c07aed0 100644 --- a/activemq-osgi/pom.xml +++ b/activemq-osgi/pom.xml @@ -181,6 +181,7 @@ javax.resource*;resolution:=optional, javax.servlet*;resolution:=optional, jakarta.servlet*;resolution:=optional, + com.github.benmanes.caffeine*;resolution:=optional, com.thoughtworks.xstream*;resolution:=optional, org.apache.camel*;resolution:=optional, org.apache.camel.spring.handler;resolution:=optional,