This is an automated email from the ASF dual-hosted git repository.
markt-asf pushed a commit to branch 9.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
The following commit(s) were added to refs/heads/9.0.x by this push:
new 6203929ca5 Review "Scope the map member & state locks of
AbstractReplicatedMap..."
6203929ca5 is described below
commit 6203929ca5c68f20cfd6ca0d8b196bf65a247c0f
Author: Mark Thomas <[email protected]>
AuthorDate: Fri Oct 23 11:07:57 2026 +0100
Review "Scope the map member & state locks of AbstractReplicatedMap..."
Fix concurrency regressions identified by GPT-5.6
Co-authored by: CoPilot / GPT-5.6 Sol
---
.../tribes/tipis/AbstractReplicatedMap.java | 95 ++++++++++++----
.../catalina/tribes/tipis/ReplicatedMap.java | 11 +-
.../tribes/tipis/TestAbstractReplicatedMap.java | 123 +++++++++++++++++++++
3 files changed, 199 insertions(+), 30 deletions(-)
diff --git a/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
b/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
index 89680c2b55..3a41090122 100644
--- a/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
+++ b/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java
@@ -125,6 +125,10 @@ public abstract class AbstractReplicatedMap<K, V>
* A list of members in our map
*/
protected final transient HashMap<Member,Long> mapMembers = new
HashMap<>();
+ /**
+ * Members removed by ping expiry before their disappearance is processed.
+ */
+ private final transient Set<Member> membersPendingDisappearance = new
LinkedHashSet<>();
/**
* Our default send options
*/
@@ -380,10 +384,20 @@ public abstract class AbstractReplicatedMap<K, V>
synchronized (mapMembers) {
Long access = mapMembers.get(member);
expired = access != null && (System.currentTimeMillis() -
access.longValue()) > timeout;
+ if (expired) {
+ mapMembers.remove(member);
+ membersPendingDisappearance.add(member);
+ }
}
if (expired) {
log.warn(sm.getString("abstractReplicatedMap.ping.timeout",
member, mapname));
- memberDisappeared(member);
+ try {
+ memberDisappeared(member);
+ } finally {
+ synchronized (mapMembers) {
+ membersPendingDisappearance.remove(member);
+ }
+ }
}
}
}
@@ -721,22 +735,22 @@ public abstract class AbstractReplicatedMap<K, V>
// state transfer request
if (mapmsg.getMsgType() == MapMessage.MSG_STATE || mapmsg.getMsgType()
== MapMessage.MSG_STATE_COPY) {
boolean copy = (mapmsg.getMsgType() == MapMessage.MSG_STATE_COPY);
- List<MapEntry<K,V>> snapshot = new ArrayList<>();
+ List<MapState<K,V>> snapshot = new ArrayList<>();
synchronized (stateMutex) { // scan under the lock, so that a
concurrent state operation cannot overlap it
for (Entry<K,MapEntry<K,V>> e : innerMap.entrySet()) {
MapEntry<K,V> entry = innerMap.get(e.getKey());
if (entry != null && entry.isSerializable()) {
- snapshot.add(entry);
+ snapshot.add(new MapState<>(entry));
}
}
} // synchronized
// Serialize the snapshot outside of the state lock, as
serializing the whole
// map content blocks every other state operation for a long time
otherwise
ArrayList<MapMessage> list = new ArrayList<>(snapshot.size());
- for (MapEntry<K,V> entry : snapshot) {
+ for (MapState<K,V> entry : snapshot) {
MapMessage me = new MapMessage(mapContextName, copy ?
MapMessage.MSG_COPY : MapMessage.MSG_PROXY,
- false, (Serializable) entry.getKey(), copy ?
(Serializable) entry.getValue() : null, null,
- entry.getPrimary(), entry.getBackupNodes());
+ false, (Serializable) entry.key, copy ? (Serializable)
entry.value : null, null, entry.primary,
+ entry.backupNodes);
list.add(me);
}
mapmsg.setValue(list);
@@ -992,21 +1006,40 @@ public abstract class AbstractReplicatedMap<K, V>
// blocks until the backup acknowledges, while session operations
and the state
// transfer of joining members need that lock
for (MapEntry<K,V> entry : entriesWithoutBackup) {
- if (!entry.isPrimary() || (entry.getBackupNodes() != null &&
entry.getBackupNodes().length > 0)) {
- // The entry found its backup in the meantime
- continue;
+ synchronized (entry) {
+ if (!entry.isPrimary() || (entry.getBackupNodes() != null
&& entry.getBackupNodes().length > 0)) {
+ // The entry found its backup in the meantime
+ continue;
+ }
+ try {
+ Member[] backup = publishEntryInfo(entry.getKey(),
entry.getValue());
+ entry.setBackupNodes(backup);
+ entry.setPrimary(channel.getLocalMember(false));
+ } catch (ChannelException x) {
+
log.error(sm.getString("abstractReplicatedMap.unableSelect.backup"), x);
+ } // catch
}
- try {
- Member[] backup = publishEntryInfo(entry.getKey(),
entry.getValue());
- entry.setBackupNodes(backup);
- entry.setPrimary(channel.getLocalMember(false));
- } catch (ChannelException x) {
-
log.error(sm.getString("abstractReplicatedMap.unableSelect.backup"), x);
- } // catch
} // for
} // end if
}
+ /**
+ * Removes a map member unless it was already removed by ping expiry.
+ *
+ * @param member The member to remove
+ *
+ * @return {@code true} if the disappearance needs to be processed
+ */
+ protected boolean removeMapMember(Member member) {
+ synchronized (mapMembers) {
+ if (membersPendingDisappearance.remove(member)) {
+ // A ping response may have re-added the member after the
expiry removed it
+ return !mapMembers.containsKey(member);
+ }
+ return mapMembers.remove(member) != null;
+ }
+ }
+
/**
* Checks if a member is in the given set.
*
@@ -1071,14 +1104,11 @@ public abstract class AbstractReplicatedMap<K, V>
*/
@Override
public void memberDisappeared(Member member) {
- synchronized (mapMembers) {
- boolean removed = (mapMembers.remove(member) != null);
- if (!removed) {
- if (log.isDebugEnabled()) {
-
log.debug(sm.getString("replicatedMap.member.disappeared.unknown", member));
- }
- return; // the member was not part of our map.
+ if (!removeMapMember(member)) {
+ if (log.isDebugEnabled()) {
+
log.debug(sm.getString("replicatedMap.member.disappeared.unknown", member));
}
+ return; // the member was not part of our map.
}
if (log.isInfoEnabled()) {
log.info(sm.getString("replicatedMap.member.disappeared", member));
@@ -1601,6 +1631,25 @@ public abstract class AbstractReplicatedMap<K, V>
}
+ //
------------------------------------------------------------------------------
+ // Map state snapshot class
+ //
------------------------------------------------------------------------------
+ private static final class MapState<K, V> {
+
+ private final K key;
+ private final V value;
+ private final Member primary;
+ private final Member[] backupNodes;
+
+ private MapState(MapEntry<K,V> entry) {
+ key = entry.getKey();
+ value = entry.getValue();
+ primary = entry.getPrimary();
+ Member[] nodes = entry.getBackupNodes();
+ backupNodes = nodes == null ? null : nodes.clone();
+ }
+ }
+
//
------------------------------------------------------------------------------
// Map Entry class
//
------------------------------------------------------------------------------
diff --git a/java/org/apache/catalina/tribes/tipis/ReplicatedMap.java
b/java/org/apache/catalina/tribes/tipis/ReplicatedMap.java
index dbdb44d837..df77d045b6 100644
--- a/java/org/apache/catalina/tribes/tipis/ReplicatedMap.java
+++ b/java/org/apache/catalina/tribes/tipis/ReplicatedMap.java
@@ -184,14 +184,11 @@ public class ReplicatedMap<K, V> extends
AbstractReplicatedMap<K,V> {
@Override
public void memberDisappeared(Member member) {
Log log = getLog();
- synchronized (mapMembers) {
- boolean removed = (mapMembers.remove(member) != null);
- if (!removed) {
- if (log.isDebugEnabled()) {
-
log.debug(sm.getString("replicatedMap.member.disappeared.unknown", member));
- }
- return; // the member was not part of our map.
+ if (!removeMapMember(member)) {
+ if (log.isDebugEnabled()) {
+
log.debug(sm.getString("replicatedMap.member.disappeared.unknown", member));
}
+ return; // the member was not part of our map.
}
if (log.isInfoEnabled()) {
log.info(sm.getString("replicatedMap.member.disappeared", member));
diff --git
a/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java
b/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java
index 02a57d9c4f..6b4b84da75 100644
--- a/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java
+++ b/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java
@@ -21,6 +21,8 @@ import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.Map.Entry;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -131,6 +133,7 @@ public class TestAbstractReplicatedMap {
public void memberDisappeared(Member disappearedMember) {
disappeared.incrementAndGet();
lockAvailable.set(monitorAvailable(mapMembers, 2000));
+ super.memberDisappeared(disappearedMember);
}
};
@@ -143,6 +146,28 @@ public class TestAbstractReplicatedMap {
Assert.assertTrue("The ping must expire members outside the map
members lock", lockAvailable.get());
}
+ @Test
+ public void testPingDoesNotRemoveRefreshedMember() throws Exception {
+ Member member = new MemberImpl("localhost", 4014, -1);
+
+ LazyReplicatedMap<String,String> map = new LazyReplicatedMap<>((key,
value) -> {
+ // Not made primary here
+ }, new TestChannel(), 1000, "TestAbstractReplicatedMap", null) {
+
+ @Override
+ public void memberDisappeared(Member disappearedMember) {
+ memberAlive(disappearedMember);
+ super.memberDisappeared(disappearedMember);
+ }
+ };
+
+ map.memberAlive(member);
+ Thread.sleep(10);
+ map.ping(0);
+
+ Assert.assertTrue(map.inSet(member, map.getMapMembers()));
+ }
+
/*
* Adding a member selects backups for the primary entries that have none,
which blocks until the selected backups
* acknowledge. That must not run while holding the state lock, since the
state transfer of joining members needs
@@ -177,6 +202,60 @@ public class TestAbstractReplicatedMap {
Assert.assertTrue("The member addition must publish backups outside
the state lock", lockAvailable.get());
}
+ @Test
+ public void testConcurrentMemberAddsPublishBackupOnce() throws Exception {
+ CountDownLatch firstPublishStarted = new CountDownLatch(1);
+ CountDownLatch releaseFirstPublish = new CountDownLatch(1);
+ AtomicBoolean probeOnPublish = new AtomicBoolean();
+ AtomicInteger published = new AtomicInteger();
+ Member backup = new MemberImpl("localhost", 4015, -1);
+
+ LazyReplicatedMap<String,String> map = new LazyReplicatedMap<>((key,
value) -> {
+ // Not made primary here
+ }, new TestChannel(), 1000, "TestAbstractReplicatedMap", null) {
+
+ @Override
+ protected Member[] publishEntryInfo(Object key, Object value) {
+ if (!probeOnPublish.get()) {
+ return new Member[0];
+ }
+ if (published.incrementAndGet() == 1) {
+ firstPublishStarted.countDown();
+ try {
+ releaseFirstPublish.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ Assert.fail(e.getMessage());
+ }
+ }
+ return new Member[] { backup };
+ }
+ };
+
+ map.put("key", "value");
+ probeOnPublish.set(true);
+ Member firstMember = new MemberImpl("localhost", 4016, -1);
+ Member secondMember = new MemberImpl("localhost", 4017, -1);
+ Thread first = new Thread(() -> map.mapMemberAdded(firstMember));
+ Thread second = new Thread(() -> map.mapMemberAdded(secondMember));
+ first.setDaemon(true);
+ second.setDaemon(true);
+ first.start();
+ Assert.assertTrue(firstPublishStarted.await(5, TimeUnit.SECONDS));
+ second.start();
+ try {
+ second.join(1000);
+ } finally {
+ releaseFirstPublish.countDown();
+ }
+ first.join(5000);
+ second.join(5000);
+
+ Assert.assertFalse(first.isAlive());
+ Assert.assertFalse(second.isAlive());
+ Assert.assertEquals(1, published.get());
+ }
+
/*
* Answering a state request serializes the whole map content, which must
not happen while holding the state lock,
* since it blocks the state transfer of other joining members for the
duration of that serialization.
@@ -217,6 +296,50 @@ public class TestAbstractReplicatedMap {
Assert.assertEquals(MapMessage.MSG_COPY, ((MapMessage) ((ArrayList<?>)
list).get(0)).getMsgType());
}
+ @Test
+ public void testStateRequestUsesStableEntrySnapshot() throws Exception {
+ CountDownLatch release = new CountDownLatch(1);
+ BlockingValue blockingValue = new BlockingValue(release);
+ LazyReplicatedMap<String,Serializable> map = new
LazyReplicatedMap<>((key, value) -> {
+ // Not made primary here
+ }, new TestChannel(), 1000, "TestAbstractReplicatedMap", null);
+ map.put("first", "value");
+ map.put("second", "old");
+
+
Iterator<Entry<String,AbstractReplicatedMap.MapEntry<String,Serializable>>>
entries =
+ map.innerMap.entrySet().iterator();
+ Entry<String,AbstractReplicatedMap.MapEntry<String,Serializable>>
first = entries.next();
+ Entry<String,AbstractReplicatedMap.MapEntry<String,Serializable>>
second = entries.next();
+ first.getValue().setValue(blockingValue);
+ Serializable originalSecondValue = second.getValue().getValue();
+
+ MapMessage stateRequest = new
MapMessage("TestAbstractReplicatedMap".getBytes(StandardCharsets.UTF_8),
+ MapMessage.MSG_STATE_COPY, false, null, null, null, null,
null);
+ AtomicReference<Serializable> reply = new AtomicReference<>();
+ Member sender = new MemberImpl("localhost", 4018, -1);
+ Thread replyThread = new Thread(() ->
reply.set(map.replyRequest(stateRequest, sender)));
+ replyThread.setDaemon(true);
+ replyThread.start();
+
+ try {
+ Assert.assertTrue(blockingValue.serializationStarted.await(5,
TimeUnit.SECONDS));
+ synchronized (map.stateMutex) {
+ second.getValue().setValue("new");
+ }
+ } finally {
+ release.countDown();
+ }
+ replyThread.join(5000);
+
+ Assert.assertFalse(replyThread.isAlive());
+ MapMessage replyMessage = (MapMessage) reply.get();
+ @SuppressWarnings("unchecked")
+ ArrayList<MapMessage> state = (ArrayList<MapMessage>)
replyMessage.getValue();
+ MapMessage secondMessage =
+ state.stream().filter(message ->
second.getKey().equals(message.getKey())).findFirst().orElseThrow();
+ Assert.assertEquals(originalSecondValue, secondMessage.getValue());
+ }
+
/*
* The put silently accepts keys that cannot be replicated, so the remove
must not fail on such a key either, but
* just skip the notification to the other members. The removal previously
threw an unchecked exception from the
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
