This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 87c5bf8ff0ec94ec787f1f3d5aa77f99d8707154 Author: opencode <[email protected]> AuthorDate: Sun Oct 11 13:36:44 2026 +0200 Scope the map member and state locks of AbstractReplicatedMap to the map operations The expiry handling of the ping ran the relocation of the entries of a timed out member while holding the map members lock. That relocation sends to the remaining members and blocks until they acknowledge, so one slow or dead member stalled every session operation, since every put, remove and get reads the map members lock. Likewise, the backup selection for a newly added member performed its blocking publications while holding the state lock, and the state transfer paths held the state lock across the full (de)serialization of the map content: the deserialization of a received full state, and the serialization of every serializable entry when answering a state request for a joining member. The locks are now held only briefly. The ping collects the expired members under the lock and then re-checks each one before handling the disappearance outside of the lock. The member addition collects the primary entries that have no backup under the state lock and publishes them outside of it, re-checking that the entry still needs a backup. The state transfer and the state request only scan the map under the state lock, and deserialize or serialize outside of it. Add unit tests that verify that the other thread can enter the monitor while the expiry, the backup publication and the state serialization run. --- .../tribes/tipis/AbstractReplicatedMap.java | 92 ++++++--- .../tribes/tipis/TestAbstractReplicatedMap.java | 214 +++++++++++++++++++++ webapps/docs/changelog.xml | 12 ++ 3 files changed, 293 insertions(+), 25 deletions(-) diff --git a/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java b/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java index 46380ba8b2..51dd9981f9 100644 --- a/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java +++ b/java/org/apache/catalina/tribes/tipis/AbstractReplicatedMap.java @@ -363,17 +363,30 @@ public abstract class AbstractReplicatedMap<K, V> } } // update our map of members, expire some if we didn't receive a ping back + List<Member> expiredMembers = new ArrayList<>(); synchronized (mapMembers) { Member[] members = mapMembers.keySet().toArray(new Member[0]); long now = System.currentTimeMillis(); for (Member member : members) { long access = mapMembers.get(member).longValue(); if ((now - access) > timeout) { - log.warn(sm.getString("abstractReplicatedMap.ping.timeout", member, mapname)); - memberDisappeared(member); + expiredMembers.add(member); } } } // synch + // The expiry handling relocates entries, which sends to other members and may block + // until they acknowledge, so it must not run while holding the map members lock + for (Member member : expiredMembers) { + boolean expired; + synchronized (mapMembers) { + Long access = mapMembers.get(member); + expired = access != null && (System.currentTimeMillis() - access.longValue()) > timeout; + } + if (expired) { + log.warn(sm.getString("abstractReplicatedMap.ping.timeout", member, mapname)); + memberDisappeared(member); + } + } } /** @@ -625,12 +638,26 @@ public abstract class AbstractReplicatedMap<K, V> Response[] resp = rpcChannel.send(new Member[] { backup }, msg, RpcChannel.FIRST_REPLY, channelSendOptions, rpcTimeout); if (resp.length > 0) { + msg = (MapMessage) resp[0].getMessage(); + msg.deserialize(getExternalLoaders()); + ArrayList<?> list = (ArrayList<?>) msg.getValue(); + // Deserialize the state messages outside of the state lock below, so that + // it is held only for the application of the entries to the map + List<Serializable> messages = new ArrayList<>(list.size()); + for (Object o : list) { + if (o instanceof MapMessage mapMessage) { + try { + mapMessage.deserialize(getExternalLoaders()); + } catch (ClassNotFoundException | IOException x) { + log.error(sm.getString("abstractReplicatedMap.unable.deserialize.MapMessage"), x); + continue; + } + } + messages.add((Serializable) o); + } synchronized (stateMutex) { - msg = (MapMessage) resp[0].getMessage(); - msg.deserialize(getExternalLoaders()); - ArrayList<?> list = (ArrayList<?>) msg.getValue(); - for (Object o : list) { - messageReceived((Serializable) o, resp[0].getSource()); + for (Serializable message : messages) { + messageReceived(message, resp[0].getSource()); } // for } stateTransferred = true; @@ -683,23 +710,27 @@ public abstract class AbstractReplicatedMap<K, V> // state transfer request if (mapmsg.getMsgType() == MapMessage.MSG_STATE || mapmsg.getMsgType() == MapMessage.MSG_STATE_COPY) { - synchronized (stateMutex) { // make sure we don't do two things at the same time - ArrayList<MapMessage> list = new ArrayList<>(); + boolean copy = (mapmsg.getMsgType() == MapMessage.MSG_STATE_COPY); + List<MapEntry<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()) { - boolean copy = (mapmsg.getMsgType() == MapMessage.MSG_STATE_COPY); - 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()); - list.add(me); + snapshot.add(entry); } } - mapmsg.setValue(list); - return mapmsg; - } // 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) { + 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()); + list.add(me); + } + mapmsg.setValue(list); + return mapmsg; } // ping @@ -930,6 +961,7 @@ public abstract class AbstractReplicatedMap<K, V> } } if (memberAdded) { + List<MapEntry<K,V>> entriesWithoutBackup = new ArrayList<>(); synchronized (stateMutex) { for (Entry<K,MapEntry<K,V>> e : innerMap.entrySet()) { MapEntry<K,V> entry = innerMap.get(e.getKey()); @@ -937,16 +969,26 @@ public abstract class AbstractReplicatedMap<K, V> continue; } if (entry.isPrimary() && (entry.getBackupNodes() == null || entry.getBackupNodes().length == 0)) { - 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 + entriesWithoutBackup.add(entry); } // end if } // while } // synchronized + // Select and publish the backups outside of the state lock, as the replication + // 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; + } + 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 } diff --git a/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java b/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java new file mode 100644 index 0000000000..83d933246b --- /dev/null +++ b/test/org/apache/catalina/tribes/tipis/TestAbstractReplicatedMap.java @@ -0,0 +1,214 @@ +/* + * 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.catalina.tribes.tipis; + +import java.io.IOException; +import java.io.ObjectOutputStream; +import java.io.Serial; +import java.io.Serializable; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.Assert; +import org.junit.Test; + +import org.apache.catalina.tribes.Member; +import org.apache.catalina.tribes.group.GroupChannel; +import org.apache.catalina.tribes.membership.MemberImpl; +import org.apache.catalina.tribes.tipis.AbstractReplicatedMap.MapMessage; + +public class TestAbstractReplicatedMap { + + /* + * The channel resolves any looked up member directly, so that a member can be added to the map + * members without starting a channel. + */ + private static final class TestChannel extends GroupChannel { + + @Override + public Member getMember(Member mbr) { + return mbr; + } + } + + /* + * A value whose first serialization blocks until it is released, to simulate the long lasting + * serialization of a full map content. + */ + private static final class BlockingValue implements Serializable { + + @Serial + private static final long serialVersionUID = 1L; + + private final transient CountDownLatch release; + private final transient CountDownLatch serializationStarted = new CountDownLatch(1); + private transient boolean serializedOnce = false; + + BlockingValue(CountDownLatch release) { + this.release = release; + } + + private void writeObject(ObjectOutputStream out) throws IOException { + out.defaultWriteObject(); + if (!serializedOnce) { + serializedOnce = true; + serializationStarted.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + } + } + } + + /* + * Whether another thread could enter the given monitor while the current thread stays inside + * the operation that is expected to run outside of that monitor. + */ + private static boolean monitorAvailable(Object monitor, long timeoutMs) { + CountDownLatch acquired = new CountDownLatch(1); + Thread probe = new Thread(() -> { + synchronized (monitor) { + acquired.countDown(); + } + }); + probe.setDaemon(true); + probe.start(); + try { + return acquired.await(timeoutMs, TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + } + + /* + * The expiry handling of the ping aborts the entries of a timed out member, which sends to the + * remaining members and may block until they acknowledge. It must not run while holding the + * map members lock, since every session operation needs that lock. + */ + @Test + public void testPingExpiresOutsideMapMembersLock() throws Exception { + Member member = new MemberImpl("localhost", 4011, -1); + AtomicInteger disappeared = new AtomicInteger(); + AtomicBoolean lockAvailable = new AtomicBoolean(); + + LazyReplicatedMap<String,String> map = new LazyReplicatedMap<String,String>( + (key, value) -> { + // Not made primary here + }, new TestChannel(), 1000, "TestAbstractReplicatedMap", null) { + + @Override + public void memberDisappeared(Member disappearedMember) { + disappeared.incrementAndGet(); + lockAvailable.set(monitorAvailable(mapMembers, 2000)); + } + }; + + map.memberAlive(member); + Thread.sleep(10); + // A timeout of zero expires every member that was not refreshed just now + map.ping(0); + + Assert.assertEquals(1, disappeared.get()); + Assert.assertTrue("The ping must expire members outside the map members lock", lockAvailable.get()); + } + + /* + * 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 that lock. + */ + @Test + public void testMapMemberAddedPublishesOutsideStateLock() throws Exception { + AtomicBoolean probeOnPublish = new AtomicBoolean(); + AtomicBoolean lockAvailable = new AtomicBoolean(); + AtomicInteger published = new AtomicInteger(); + + LazyReplicatedMap<String,String> map = new LazyReplicatedMap<String,String>( + (key, value) -> { + // Not made primary here + }, new TestChannel(), 1000, "TestAbstractReplicatedMap", null) { + + @Override + protected Member[] publishEntryInfo(Object key, Object value) { + if (probeOnPublish.get()) { + lockAvailable.set(monitorAvailable(stateMutex, 2000)); + } + published.incrementAndGet(); + return new Member[0]; + } + }; + + // Stores a primary entry without a backup, without probing the publication it attempts + map.put("key", "value"); + probeOnPublish.set(true); + map.mapMemberAdded(new MemberImpl("localhost", 4012, -1)); + + Assert.assertEquals(2, published.get()); + Assert.assertTrue("The member addition must publish backups outside the state lock", lockAvailable.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. + */ + @Test + public void testStateRequestSerializesOutsideStateLock() throws Exception { + CountDownLatch release = new CountDownLatch(1); + BlockingValue value = new BlockingValue(release); + + LazyReplicatedMap<String,BlockingValue> map = new LazyReplicatedMap<String,BlockingValue>( + (key, val) -> { + // Not made primary here + }, new TestChannel(), 1000, "TestAbstractReplicatedMap", null); + map.put("key", value); + + 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", 4013, -1); + Thread replyThread = new Thread(() -> reply.set(map.replyRequest(stateRequest, sender))); + replyThread.setDaemon(true); + replyThread.start(); + + // Once the value serialization has started, the map scan is complete + try { + Assert.assertTrue("The state request must serialize the map content", + value.serializationStarted.await(5, TimeUnit.SECONDS)); + boolean available = monitorAvailable(map.stateMutex, 2000); + Assert.assertTrue("The state request must serialize outside the state lock", available); + } finally { + release.countDown(); + } + replyThread.join(5000); + Assert.assertTrue(reply.get() instanceof MapMessage); + Object list = ((MapMessage) reply.get()).getValue(); + Assert.assertTrue(list instanceof ArrayList); + Assert.assertEquals(1, ((ArrayList<?>) list).size()); + Assert.assertTrue(((ArrayList<?>) list).get(0) instanceof MapMessage); + Assert.assertEquals(MapMessage.MSG_COPY, ((MapMessage) ((ArrayList<?>) list).get(0)).getMsgType()); + } +} diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml index c9aaba337a..04f3720e3b 100644 --- a/webapps/docs/changelog.xml +++ b/webapps/docs/changelog.xml @@ -552,6 +552,18 @@ further confirmation ever followed, so a single moment of network reordering cost one lost cluster message. (remm) </fix> + <fix> + The <code>AbstractReplicatedMap</code> now holds the map + members lock and the state lock only briefly. The entry + relocation triggered by the ping expiry runs outside the map + members lock, the backup selection for a newly added member + outside the state lock, and the (de)serialization of the map + content for a member joining or transferring its state + likewise outside the state lock. Previously, a slow send to a + dead member or the serialization of a large map blocked every + session operation and the state transfer of joining members + for the duration of that work. (remm) + </fix> </changelog> </subsection> <subsection name="jdbc-pool"> --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
