This is an automated email from the ASF dual-hosted git repository. markt-asf pushed a commit to branch 10.1.x in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 48e23084c03f590d3046cb1bd6b920ad28b3a6ee 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 7dfd9c28ed..acab2c8b12 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]
