Skip to content

Commit 13d2c60

Browse files
mattrpavmattrpav
committed
[#2187] Thread safety fixes
1 parent 1aa22d1 commit 13d2c60

8 files changed

Lines changed: 519 additions & 172 deletions

File tree

‎activemq-client/pom.xml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -212,6 +212,7 @@
212212
!org.apache.activemq*,
213213
!com.google.errorprone.annotations,
214214
!com.google.errorprone.annotations.concurrent,
215+
com.github.benmanes.caffeine.*;resolution:="optional",
215216
com.thoughtworks.xstream.*;resolution:="optional",
216217
javax.jmdns.*;resolution:="optional",
217218
*

‎activemq-client/src/main/java/org/apache/activemq/CaffeineMessageAudit.java‎

Lines changed: 35 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,7 @@ public class CaffeineMessageAudit {
4747

4848
private volatile int auditDepth;
4949
private volatile int maximumNumberOfProducersToTrack;
50-
private volatile Cache<String, AtomicBitArrayBin> cache;
50+
private final Cache<String, AtomicBitArrayBin> cache;
5151

5252
public CaffeineMessageAudit() {
5353
this(DEFAULT_WINDOW_SIZE, MAXIMUM_PRODUCER_COUNT);
@@ -73,22 +73,22 @@ public int getMaximumNumberOfProducersToTrack() {
7373

7474
public void setMaximumNumberOfProducersToTrack(int maximumNumberOfProducersToTrack) {
7575
this.maximumNumberOfProducersToTrack = maximumNumberOfProducersToTrack;
76-
var newCache = buildCache(maximumNumberOfProducersToTrack);
77-
newCache.putAll(this.cache.asMap());
78-
this.cache = newCache;
76+
// Resize in place: rebuilding and swapping the cache would lose
77+
// entries inserted concurrently with the copy
78+
cache.policy().eviction()
79+
.ifPresent(eviction -> eviction.setMaximum(maximumNumberOfProducersToTrack));
7980
}
8081

8182
public boolean isDuplicate(String id) {
8283
var seed = IdGenerator.getSeedFromId(id);
8384
if (seed == null) {
8485
return false;
8586
}
86-
var bab = cache.get(seed, k -> new AtomicBitArrayBin(auditDepth));
8787
var index = IdGenerator.getSequenceFromId(id);
88-
if (index >= 0) {
89-
return bab.setBit(index, true);
88+
if (index < 0) {
89+
return false;
9090
}
91-
return false;
91+
return markSeen(seed, index);
9292
}
9393

9494
public boolean isDuplicate(final MessageId id) {
@@ -99,8 +99,21 @@ public boolean isDuplicate(final MessageId id) {
9999
if (pid == null) {
100100
return false;
101101
}
102-
var bab = cache.get(pid.toString(), k -> new AtomicBitArrayBin(auditDepth));
103-
return bab.setBit(id.getProducerSequenceId(), true);
102+
return markSeen(pid.toString(), id.getProducerSequenceId());
103+
}
104+
105+
/**
106+
* Record the index against the bin for the key. If the bin is evicted
107+
* between lookup and the bit write, the write lands in an orphaned bin
108+
* and the next occurrence of the id is not detected as a duplicate.
109+
* This residual race is accepted: eviction only fires under producer
110+
* oversubscription, membership re-check-and-retry in that regime was
111+
* measured at a 2-5x throughput cost while the recreated entry is
112+
* immediately evicted again, and the failure direction (duplicate
113+
* redelivery) is tolerable under at-least-once delivery.
114+
*/
115+
private boolean markSeen(String key, long index) {
116+
return cache.get(key, k -> new AtomicBitArrayBin(auditDepth)).setBit(index, true);
104117
}
105118

106119
public void rollback(final MessageId id) {
@@ -111,22 +124,26 @@ public void rollback(final MessageId id) {
111124
if (pid == null) {
112125
return;
113126
}
114-
var bab = cache.getIfPresent(pid.toString());
115-
if (bab != null) {
116-
bab.setBit(id.getProducerSequenceId(), false);
117-
}
127+
var seqId = id.getProducerSequenceId();
128+
cache.asMap().computeIfPresent(pid.toString(), (k, bab) -> {
129+
bab.setBit(seqId, false);
130+
return bab;
131+
});
118132
}
119133

120134
public void rollback(final String id) {
121135
var seed = IdGenerator.getSeedFromId(id);
122136
if (seed == null) {
123137
return;
124138
}
125-
var bab = cache.getIfPresent(seed);
126-
if (bab != null) {
127-
var index = IdGenerator.getSequenceFromId(id);
128-
bab.setBit(index, false);
139+
var index = IdGenerator.getSequenceFromId(id);
140+
if (index < 0) {
141+
return;
129142
}
143+
cache.asMap().computeIfPresent(seed, (k, bab) -> {
144+
bab.setBit(index, false);
145+
return bab;
146+
});
130147
}
131148

132149
public boolean isInOrder(final String id) {

‎activemq-client/src/main/java/org/apache/activemq/ConcurrentMessageAudit.java‎

Lines changed: 36 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -85,12 +85,11 @@ public boolean isDuplicate(String id) {
8585
if (seed == null) {
8686
return false;
8787
}
88-
var bab = getOrCreate(seed);
8988
var index = IdGenerator.getSequenceFromId(id);
90-
if (index >= 0) {
91-
return bab.setBit(index, true);
89+
if (index < 0) {
90+
return false;
9291
}
93-
return false;
92+
return markSeen(seed, index);
9493
}
9594

9695
public boolean isDuplicate(final MessageId id) {
@@ -101,8 +100,21 @@ public boolean isDuplicate(final MessageId id) {
101100
if (pid == null) {
102101
return false;
103102
}
104-
var bab = getOrCreate(pid.toString());
105-
return bab.setBit(id.getProducerSequenceId(), true);
103+
return markSeen(pid.toString(), id.getProducerSequenceId());
104+
}
105+
106+
/**
107+
* Record the index against the bin for the key. If the bin is evicted
108+
* between lookup and the bit write, the write lands in an orphaned bin
109+
* and the next occurrence of the id is not detected as a duplicate.
110+
* This residual race is accepted: eviction only fires under producer
111+
* oversubscription, membership re-check-and-retry in that regime was
112+
* measured at a 2-5x throughput cost while the recreated entry is
113+
* immediately evicted again, and the failure direction (duplicate
114+
* redelivery) is tolerable under at-least-once delivery.
115+
*/
116+
private boolean markSeen(String key, long index) {
117+
return getOrCreate(key).setBit(index, true);
106118
}
107119

108120
public void rollback(final MessageId id) {
@@ -113,22 +125,26 @@ public void rollback(final MessageId id) {
113125
if (pid == null) {
114126
return;
115127
}
116-
var bab = map.get(pid.toString());
117-
if (bab != null) {
118-
bab.setBit(id.getProducerSequenceId(), false);
119-
}
128+
var seqId = id.getProducerSequenceId();
129+
map.computeIfPresent(pid.toString(), (k, bab) -> {
130+
bab.setBit(seqId, false);
131+
return bab;
132+
});
120133
}
121134

122135
public void rollback(final String id) {
123136
var seed = IdGenerator.getSeedFromId(id);
124137
if (seed == null) {
125138
return;
126139
}
127-
var bab = map.get(seed);
128-
if (bab != null) {
129-
long index = IdGenerator.getSequenceFromId(id);
130-
bab.setBit(index, false);
140+
var index = IdGenerator.getSequenceFromId(id);
141+
if (index < 0) {
142+
return;
131143
}
144+
map.computeIfPresent(seed, (k, bab) -> {
145+
bab.setBit(index, false);
146+
return bab;
147+
});
132148
}
133149

134150
public boolean isInOrder(final String id) {
@@ -185,6 +201,12 @@ private AtomicBitArrayBin getOrCreate(String key) {
185201
return bab;
186202
}
187203

204+
/**
205+
* Trim the map to the configured maximum. Eviction follows CHM iteration
206+
* order (approximate FIFO, not LRU). Concurrent callers may transiently
207+
* remove more entries than strictly required; the bound is approximate
208+
* by design and self-corrects on subsequent inserts.
209+
*/
188210
private void evictExcess() {
189211
var max = maximumNumberOfProducersToTrack;
190212
while (map.size() > max) {

0 commit comments

Comments
 (0)