This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch 11.0.x in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit fcb46a0812949f9d680ff46e349c7677bc4e2cdb Author: opencode <[email protected]> AuthorDate: Fri Oct 9 15:34:14 2026 +0200 Run TcpFailureDetector liveness checks outside the membership monitor The TCP liveness checks performed by memberAdded, memberDisappeared and the heartbeat and MBean driven member checks each connect to the target member, for up to the configured connection timeout and read test timeout. All of these checks ran while holding the membership monitor, so a network outage serialized seconds of blocked TCP connects against every other thread that needed to read or update membership state, delaying legitimate member add and remove processing and amplifying flapping. The checks now collect the members to verify under the monitor, run the network I/O outside it, then re-acquire the monitor and apply the result only if the membership state still matches the decision. Member events forwarded to the rest of the channel are likewise sent outside the monitor, which also removes a deadlock risk between the monitor and application membership listeners that call back into the interceptor. Expire handling for remove suspects gained a null guard because the suspect entry can now be removed concurrently while a member is being checked. Add a unit test that records whether the monitor is held during every liveness check and drives the add, disappearance and both member check paths end to end. --- .../group/interceptors/TcpFailureDetector.java | 268 +++++++++++++++------ .../TestTcpFailureDetectorLivenessChecks.java | 204 ++++++++++++++++ webapps/docs/changelog.xml | 14 ++ 3 files changed, 408 insertions(+), 78 deletions(-) diff --git a/java/org/apache/catalina/tribes/group/interceptors/TcpFailureDetector.java b/java/org/apache/catalina/tribes/group/interceptors/TcpFailureDetector.java index 4f9a83bf88..161b482539 100644 --- a/java/org/apache/catalina/tribes/group/interceptors/TcpFailureDetector.java +++ b/java/org/apache/catalina/tribes/group/interceptors/TcpFailureDetector.java @@ -22,6 +22,7 @@ import java.net.InetSocketAddress; import java.net.NoRouteToHostException; import java.net.Socket; import java.net.SocketTimeoutException; +import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; @@ -158,19 +159,28 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai setupMembership(); } boolean notify = false; + boolean verify = false; synchronized (membership) { if (removeSuspects.containsKey(member)) { // previously marked suspect, system below picked up the member again removeSuspects.remove(member); } else if (membership.getMember(member) == null) { - // if we add it here, then add it upwards too - // check to see if it is alive - if (memberAlive(member)) { - membership.memberAlive(member); - addSuspects.remove(member); - notify = true; - } else { - if (member instanceof StaticMember) { + verify = true; + } + } + if (verify) { + // if we add it here, then add it upwards too + // check to see if it is alive, outside the membership monitor so + // that slow TCP connects do not block other membership operations + boolean alive = memberAlive(member); + synchronized (membership) { + // the state may have changed while the member was being checked + if (!removeSuspects.containsKey(member) && membership.getMember(member) == null) { + if (alive) { + membership.memberAlive(member); + addSuspects.remove(member); + notify = true; + } else if (member instanceof StaticMember) { addSuspects.put(member, Long.valueOf(System.currentTimeMillis())); } } @@ -200,29 +210,41 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai } super.memberDisappeared(member); } else { + boolean verify; + synchronized (membership) { + verify = membership.contains(member); + } + if (!verify) { + if (log.isInfoEnabled()) { + log.info(sm.getString("tcpFailureDetector.already.disappeared", member)); + } + return; + } boolean notify = false; + boolean applied = false; if (log.isInfoEnabled()) { log.info(sm.getString("tcpFailureDetector.memberDisappeared.verify", member)); } + // check to see if the member really is gone, outside the membership + // monitor so that slow TCP connects do not block other membership + // operations + boolean alive = memberAlive(member); synchronized (membership) { - if (!membership.contains(member)) { - if (log.isInfoEnabled()) { - log.info(sm.getString("tcpFailureDetector.already.disappeared", member)); + // the state may have changed while the member was being checked + if (membership.contains(member)) { + applied = true; + if (alive) { + // add the member as suspect + removeSuspects.put(member, Long.valueOf(System.currentTimeMillis())); + } else { + // not correct, we need to maintain the map + membership.removeMember(member); + removeSuspects.remove(member); + if (member instanceof StaticMember) { + addSuspects.put(member, Long.valueOf(System.currentTimeMillis())); + } + notify = true; } - return; - } - // check to see if the member really is gone - if (!memberAlive(member)) { - // not correct, we need to maintain the map - membership.removeMember(member); - removeSuspects.remove(member); - if (member instanceof StaticMember) { - addSuspects.put(member, Long.valueOf(System.currentTimeMillis())); - } - notify = true; - } else { - // add the member as suspect - removeSuspects.put(member, Long.valueOf(System.currentTimeMillis())); } } if (notify) { @@ -230,7 +252,7 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai log.info(sm.getString("tcpFailureDetector.member.disappeared", member)); } super.memberDisappeared(member); - } else { + } else if (applied) { if (log.isInfoEnabled()) { log.info(sm.getString("tcpFailureDetector.still.alive", member)); } @@ -279,12 +301,10 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai if (membership == null) { setupMembership(); } - synchronized (membership) { - if (!checkAll) { - performBasicCheck(); - } else { - performForcedCheck(); - } + if (!checkAll) { + performBasicCheck(); + } else { + performForcedCheck(); } } catch (Exception e) { log.warn(sm.getString("tcpFailureDetector.heartbeat.failed"), e); @@ -292,86 +312,158 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai } /** - * Performs a forced check of all cluster members. + * Performs a forced check of all cluster members. The TCP liveness checks run outside the + * membership monitor and the results are only applied after re-acquiring the monitor and + * re-checking that the current state still matches the decision. */ protected void performForcedCheck() { // update all alive times Member[] members = super.getMembers(); for (int i = 0; members != null && i < members.length; i++) { - if (memberAlive(members[i])) { - if (membership.memberAlive(members[i])) { - super.memberAdded(members[i]); - } - addSuspects.remove(members[i]); - } else { - if (membership.getMember(members[i]) != null) { - membership.removeMember(members[i]); - removeSuspects.remove(members[i]); - if (members[i] instanceof StaticMember) { - addSuspects.put(members[i], Long.valueOf(System.currentTimeMillis())); + Member member = members[i]; + boolean alive = memberAlive(member); + boolean added = false; + boolean disappeared = false; + synchronized (membership) { + if (alive) { + if (membership.memberAlive(member)) { + added = true; } - super.memberDisappeared(members[i]); - } - } // end if + addSuspects.remove(member); + } else { + if (membership.getMember(member) != null) { + membership.removeMember(member); + removeSuspects.remove(member); + if (member instanceof StaticMember) { + addSuspects.put(member, Long.valueOf(System.currentTimeMillis())); + } + disappeared = true; + } + } // end if + } + if (added) { + super.memberAdded(member); + } else if (disappeared) { + super.memberDisappeared(member); + } } // for } /** - * Performs a basic check of suspected members. + * Performs a basic check of suspected members. The TCP liveness checks run outside the + * membership monitor and the results are only applied after re-acquiring the monitor and + * re-checking that the current state still matches the decision. */ protected void performBasicCheck() { // update all alive times Member[] members = super.getMembers(); + ArrayList<Member> toCheck = new ArrayList<>(); for (int i = 0; members != null && i < members.length; i++) { - if (addSuspects.containsKey(members[i]) && membership.getMember(members[i]) == null) { - // avoid temporary adding member. - continue; - } - if (membership.memberAlive(members[i])) { - // we don't have this one in our membership, check to see if the member is alive - if (memberAlive(members[i])) { - log.warn(sm.getString("tcpFailureDetector.performBasicCheck.memberAdded", members[i])); - super.memberAdded(members[i]); - } else { - membership.removeMember(members[i]); + synchronized (membership) { + if (addSuspects.containsKey(members[i]) && membership.getMember(members[i]) == null) { + // avoid temporary adding member. + continue; + } + if (membership.memberAlive(members[i])) { + // we don't have this one in our membership, check to see if the member is alive + toCheck.add(members[i]); } // end if - } // end if + } } // for + // The liveness checks run outside the monitor, so a concurrent memberDisappeared can + // remove a member from the tracker between the two passes. The apply below then skips + // that member and the concurrent disappearance is the one that is reported. + for (Member member : toCheck) { + boolean alive = memberAlive(member); + boolean notify = false; + synchronized (membership) { + if (membership.getMember(member) != null) { + if (alive) { + notify = true; + } else { + membership.removeMember(member); + } // end if + } + } + if (notify) { + log.warn(sm.getString("tcpFailureDetector.performBasicCheck.memberAdded", member)); + super.memberAdded(member); + } + } + // check suspect members if they are still alive, // if not, simply issue the memberDisappeared message - Member[] keys = removeSuspects.keySet().toArray(new Member[0]); + Member[] keys; + synchronized (membership) { + keys = removeSuspects.keySet().toArray(new Member[0]); + } for (Member m : keys) { - if (membership.getMember(m) != null && (!memberAlive(m))) { - membership.removeMember(m); - if (m instanceof StaticMember) { - addSuspects.put(m, Long.valueOf(System.currentTimeMillis())); + boolean verify; + synchronized (membership) { + verify = membership.getMember(m) != null; + } + boolean disappeared = false; + if (verify) { + boolean alive = memberAlive(m); + if (!alive) { + synchronized (membership) { + // the state may have changed while the member was being checked + if (membership.getMember(m) != null) { + membership.removeMember(m); + if (m instanceof StaticMember) { + addSuspects.put(m, Long.valueOf(System.currentTimeMillis())); + } + removeSuspects.remove(m); + disappeared = true; + } + } + } else { + synchronized (membership) { + expireRemoveSuspect(m); + } } + } else { + synchronized (membership) { + expireRemoveSuspect(m); + } + } + if (disappeared) { super.memberDisappeared(m); - removeSuspects.remove(m); if (log.isInfoEnabled()) { log.info(sm.getString("tcpFailureDetector.suspectMember.dead", m)); } - } else { - if (removeSuspectsTimeout > 0) { - long timeNow = System.currentTimeMillis(); - int timeIdle = (int) ((timeNow - removeSuspects.get(m).longValue()) / 1000L); - if (timeIdle > removeSuspectsTimeout) { - removeSuspects.remove(m); // remove suspect member - } - } } } // check add suspects members if they are alive now, // if they are, simply issue the memberAdded message - keys = addSuspects.keySet().toArray(new Member[0]); + synchronized (membership) { + keys = addSuspects.keySet().toArray(new Member[0]); + } for (Member m : keys) { - if (membership.getMember(m) == null && (memberAlive(m))) { - membership.memberAlive(m); + boolean verify; + synchronized (membership) { + verify = membership.getMember(m) == null; + } + if (!verify) { + continue; + } + boolean alive = memberAlive(m); + boolean notify = false; + if (alive) { + synchronized (membership) { + // the state may have changed while the member was being checked + if (membership.getMember(m) == null) { + membership.memberAlive(m); + addSuspects.remove(m); + notify = true; + } + } + } + if (notify) { super.memberAdded(m); - addSuspects.remove(m); if (log.isInfoEnabled()) { log.info(sm.getString("tcpFailureDetector.suspectMember.alive", m)); } @@ -379,6 +471,26 @@ public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFai } } + /** + * Removes the member from the remove suspects map if it has been suspect for longer than the + * configured timeout. Must be called while holding the membership monitor. + * + * @param m the member to expire + */ + protected void expireRemoveSuspect(Member m) { + if (removeSuspectsTimeout > 0) { + Long suspectTime = removeSuspects.get(m); + // the member may have been removed from the map while it was being checked + if (suspectTime != null) { + long timeNow = System.currentTimeMillis(); + int timeIdle = (int) ((timeNow - suspectTime.longValue()) / 1000L); + if (timeIdle > removeSuspectsTimeout) { + removeSuspects.remove(m); // remove suspect member + } + } + } + } + /** * Initializes the membership tracker if not already set up. */ diff --git a/test/org/apache/catalina/tribes/group/interceptors/TestTcpFailureDetectorLivenessChecks.java b/test/org/apache/catalina/tribes/group/interceptors/TestTcpFailureDetectorLivenessChecks.java new file mode 100644 index 0000000000..676435c472 --- /dev/null +++ b/test/org/apache/catalina/tribes/group/interceptors/TestTcpFailureDetectorLivenessChecks.java @@ -0,0 +1,204 @@ +/* + * 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.group.interceptors; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import org.junit.Assert; +import org.junit.Test; + +import org.apache.catalina.tribes.Member; +import org.apache.catalina.tribes.group.ChannelInterceptorBase; +import org.apache.catalina.tribes.group.GroupChannel; +import org.apache.catalina.tribes.membership.StaticMember; + +/** + * Verifies that the TCP liveness checks of the TcpFailureDetector are performed outside the + * membership monitor, and that the check results are applied correctly once the monitor is + * re-acquired. + */ +public class TestTcpFailureDetectorLivenessChecks { + + /* + * Overrides memberAlive(Member) to record whether the membership monitor was held at the + * moment of the call and to return a configurable result, so the tests do not need real + * network I/O. + */ + private static class CheckRecordingDetector extends TcpFailureDetector { + + private final List<Boolean> monitorHeld = Collections.synchronizedList(new ArrayList<>()); + private volatile boolean alive = true; + + void setAlive(boolean value) { + alive = value; + } + + @Override + protected boolean memberAlive(Member mbr) { + monitorHeld.add(Boolean.valueOf(Thread.holdsLock(membership))); + return alive; + } + + void assertNoCheckRanUnderMonitor() { + synchronized (monitorHeld) { + Assert.assertFalse("No liveness check may run under the membership monitor", + monitorHeld.isEmpty()); + for (Boolean held : monitorHeld) { + Assert.assertFalse("A liveness check ran while holding the membership monitor", + held.booleanValue()); + } + } + } + + int checkCount() { + return monitorHeld.size(); + } + } + + // Downstream stand-in: provides the channel membership view the detector checks + private static class FakeMembershipChain extends ChannelInterceptorBase { + + volatile Member[] members = new Member[0]; + volatile Member localMember; + + @Override + public Member[] getMembers() { + return members; + } + + @Override + public Member getLocalMember(boolean incAlive) { + return localMember; + } + } + + // Upstream stand-in: records the member events the detector forwards + private static class RecordingInterceptor extends ChannelInterceptorBase { + + final List<Member> added = Collections.synchronizedList(new ArrayList<>()); + final List<Member> disappeared = Collections.synchronizedList(new ArrayList<>()); + + @Override + public void memberAdded(Member member) { + added.add(member); + } + + @Override + public void memberDisappeared(Member member) { + disappeared.add(member); + } + } + + private static CheckRecordingDetector createDetector(FakeMembershipChain chain, + RecordingInterceptor recorder) throws Exception { + CheckRecordingDetector detector = new CheckRecordingDetector(); + detector.setChannel(new GroupChannel()); + detector.setNext(chain); + detector.setPrevious(recorder); + chain.localMember = new StaticMember("localhost", 5000, -1); + return detector; + } + + @Test + public void testMemberAddedAndDisappearedChecksWithoutMonitor() throws Exception { + FakeMembershipChain chain = new FakeMembershipChain(); + RecordingInterceptor recorder = new RecordingInterceptor(); + CheckRecordingDetector detector = createDetector(chain, recorder); + StaticMember member = new StaticMember("localhost", 4000, -1); + + detector.setAlive(true); + detector.memberAdded(member); + Assert.assertNotNull("A verified live member must be added", detector.getMember(member)); + Assert.assertEquals("The add must be forwarded once", 1, recorder.added.size()); + + // A disappearance of a live member only marks it as suspect + detector.setAlive(true); + detector.memberDisappeared(member); + Assert.assertNotNull("A live member must not be removed", detector.getMember(member)); + Assert.assertTrue("A live member must not disappear upwards", recorder.disappeared.isEmpty()); + Assert.assertTrue("The live member must be a remove suspect", detector.removeSuspects.containsKey(member)); + + // A disappearance of a dead member removes it and notifies + detector.setAlive(false); + detector.memberDisappeared(member); + Assert.assertNull("A dead member must be removed", detector.getMember(member)); + Assert.assertEquals("The disappearance must be forwarded once", 1, recorder.disappeared.size()); + + detector.assertNoCheckRanUnderMonitor(); + } + + @Test + public void testBasicCheckWithoutMonitor() throws Exception { + FakeMembershipChain chain = new FakeMembershipChain(); + RecordingInterceptor recorder = new RecordingInterceptor(); + CheckRecordingDetector detector = createDetector(chain, recorder); + StaticMember member = new StaticMember("localhost", 4000, -1); + + // The channel knows the member but the local membership table does not: + // the basic check must verify and add it + chain.members = new Member[] { member }; + detector.setAlive(true); + detector.checkMembers(false); + Assert.assertNotNull("A verified live member must be added", detector.getMember(member)); + Assert.assertEquals("The add must be forwarded", 1, recorder.added.size()); + int afterFirst = detector.checkCount(); + + // A suspect member that is now dead must be removed and reported + detector.memberDisappeared(member); + Assert.assertNotNull("The live suspect must still be present", detector.getMember(member)); + detector.setAlive(false); + detector.checkMembers(false); + Assert.assertNull("The dead suspect must be removed", detector.getMember(member)); + Assert.assertEquals("The disappearance must be forwarded", 1, recorder.disappeared.size()); + Assert.assertTrue("The dead static member must be an add suspect", + detector.addSuspects.containsKey(member)); + + // An add suspect that is alive again must be re-added and reported + detector.setAlive(true); + detector.checkMembers(false); + Assert.assertNotNull("The recovered suspect must be re-added", detector.getMember(member)); + Assert.assertEquals("The re-add must be forwarded", 2, recorder.added.size()); + + Assert.assertTrue("No suspect bookkeeping may remain", + detector.removeSuspects.isEmpty() && detector.addSuspects.isEmpty()); + Assert.assertTrue("Each scenario must have run its checks", detector.checkCount() > afterFirst); + detector.assertNoCheckRanUnderMonitor(); + } + + @Test + public void testForcedCheckWithoutMonitor() throws Exception { + FakeMembershipChain chain = new FakeMembershipChain(); + RecordingInterceptor recorder = new RecordingInterceptor(); + CheckRecordingDetector detector = createDetector(chain, recorder); + StaticMember member = new StaticMember("localhost", 4000, -1); + + chain.members = new Member[] { member }; + detector.setAlive(true); + detector.checkMembers(true); + Assert.assertNotNull("A verified live member must be added", detector.getMember(member)); + Assert.assertEquals("The add must be forwarded", 1, recorder.added.size()); + + detector.setAlive(false); + detector.checkMembers(true); + Assert.assertNull("A dead member must be removed", detector.getMember(member)); + Assert.assertEquals("The disappearance must be forwarded", 1, recorder.disappeared.size()); + + detector.assertNoCheckRanUnderMonitor(); + } +} diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml index 6171cb548b..f682004697 100644 --- a/webapps/docs/changelog.xml +++ b/webapps/docs/changelog.xml @@ -2558,6 +2558,20 @@ within this timeout are excluded from the merged view and the exclusion is now logged. (remm) </fix> + <fix> + The <code>TcpFailureDetector</code> now performs its TCP + liveness checks outside the membership monitor, and only + applies the result after re-acquiring the monitor and + re-checking that the membership state has not changed in the + meantime. Previously, the network I/O, up to the configured + connection timeout for each member, was performed while + holding the monitor, which blocked every other membership + operation and update for seconds during a network outage. + Member events forwarded by the interceptor are also no longer + sent while holding the monitor, removing a potential deadlock + with membership listeners that call back into the + interceptor. (remm) + </fix> </changelog> </subsection> <subsection name="Other"> --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
