This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch 10.1.x in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit f384f6681e55221dc1ac1657fd9e8af21199c809 Author: opencode <[email protected]> AuthorDate: Fri Oct 9 15:59:12 2026 +0200 Publish the SimpleCoordinator view safely The view field was written by the membership callback threads while the accessors read it from arbitrary threads without any happens-before edge, so readers could observe a null or stale view indefinitely. The comparison, publication and view change notification in installViewWhenStable were also unsynchronized, allowing two concurrent installs of the same view to both notify. Make the field volatile so the lock-free readers see installed views, and guard the comparison, the publication and the viewChange callback with a monitor so that installing the same view notifies exactly once and notifications observe the views in install order. The stability wait remains outside the monitor, and the callback contract is documented. Add a unit test covering the installed view and accessors and a concurrent install of the same view. --- .../group/interceptors/SimpleCoordinator.java | 35 ++++--- .../group/interceptors/TestSimpleCoordinator.java | 111 +++++++++++++++++++++ 2 files changed, 134 insertions(+), 12 deletions(-) diff --git a/java/org/apache/catalina/tribes/group/interceptors/SimpleCoordinator.java b/java/org/apache/catalina/tribes/group/interceptors/SimpleCoordinator.java index b94d2ea9f3..4083793fea 100644 --- a/java/org/apache/catalina/tribes/group/interceptors/SimpleCoordinator.java +++ b/java/org/apache/catalina/tribes/group/interceptors/SimpleCoordinator.java @@ -37,7 +37,7 @@ public class SimpleCoordinator extends ChannelInterceptorBase { // NO-OP } - private Member[] view; + private volatile Member[] view; private final AtomicBoolean membershipChanged = new AtomicBoolean(); @@ -60,7 +60,9 @@ public class SimpleCoordinator extends ChannelInterceptorBase { } /** - * Override to receive view changes. + * Override to receive view changes. The callback is invoked while holding the monitor of + * this interceptor, so implementations must not block for long or acquire external or + * channel locks. The view array must not be modified by the callback. * * @param view The members array */ @@ -90,15 +92,22 @@ public class SimpleCoordinator extends ChannelInterceptorBase { } final Member[] members = getMembers(); - final Member[] view = new Member[members.length + 1]; - System.arraycopy(members, 0, view, 0, members.length); - view[members.length] = getLocalMember(false); - Arrays.sort(view, AbsoluteOrder.comp); - if (Arrays.equals(view, this.view)) { - return; + final Member[] newView = new Member[members.length + 1]; + System.arraycopy(members, 0, newView, 0, members.length); + newView[members.length] = getLocalMember(false); + Arrays.sort(newView, AbsoluteOrder.comp); + // The view is published from the membership callback threads while the + // accessors below are called from arbitrary threads. The lock keeps the + // comparison, the publication and the view change notification atomic, + // so that concurrent installs of the same view only notify once and + // viewChange observers see the views in install order + synchronized (this) { + if (Arrays.equals(view, newView)) { + return; + } + view = newView; + viewChange(newView); } - this.view = view; - viewChange(view); } @Override @@ -121,7 +130,8 @@ public class SimpleCoordinator extends ChannelInterceptorBase { * @return the coordinator member, or {@code null} if no view is established */ public Member getCoordinator() { - return view == null ? null : view[0]; + final Member[] current = view; + return current == null ? null : current[0]; } /** @@ -130,7 +140,8 @@ public class SimpleCoordinator extends ChannelInterceptorBase { * @return {@code true} if this member is the coordinator */ public boolean isCoordinator() { - return view != null && getLocalMember(false).equals(getCoordinator()); + final Member[] current = view; + return current != null && getLocalMember(false).equals(current[0]); } } diff --git a/test/org/apache/catalina/tribes/group/interceptors/TestSimpleCoordinator.java b/test/org/apache/catalina/tribes/group/interceptors/TestSimpleCoordinator.java new file mode 100644 index 0000000000..015381c943 --- /dev/null +++ b/test/org/apache/catalina/tribes/group/interceptors/TestSimpleCoordinator.java @@ -0,0 +1,111 @@ +/* + * 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.membership.MemberImpl; + +public class TestSimpleCoordinator { + + private static class RecordingCoordinator extends SimpleCoordinator { + + private final List<Member[]> views = Collections.synchronizedList(new ArrayList<>()); + + @Override + protected void viewChange(Member[] view) { + views.add(view); + } + } + + // Downstream stand-in: the dynamically discovered membership + private static class DynamicMembershipChain extends ChannelInterceptorBase { + + private volatile Member[] members = new Member[0]; + private volatile Member localMember; + + @Override + public Member[] getMembers() { + return members; + } + + @Override + public Member getLocalMember(boolean incAlive) { + return localMember; + } + } + + private static RecordingCoordinator createCoordinator(Member local) { + RecordingCoordinator coordinator = new RecordingCoordinator(); + DynamicMembershipChain chain = new DynamicMembershipChain(); + chain.localMember = local; + coordinator.setNext(chain); + return coordinator; + } + + @Test + public void testViewInstallAndAccessors() throws Exception { + MemberImpl local = new MemberImpl("localhost", 4000, -1); + MemberImpl remote = new MemberImpl("localhost", 4001, -1); + RecordingCoordinator coordinator = createCoordinator(local); + DynamicMembershipChain chain = (DynamicMembershipChain) coordinator.getNext(); + chain.members = new Member[] { remote }; + + Assert.assertNull("No view before the first install", coordinator.getView()); + + coordinator.memberAdded(remote); + + Member[] view = coordinator.getView(); + Assert.assertNotNull("The view must be installed", view); + Assert.assertEquals("The view holds the remote member and the local member", 2, view.length); + Assert.assertSame(view[0], coordinator.getCoordinator()); + Assert.assertEquals("isCoordinator must agree with the view", + local.equals(view[0]), coordinator.isCoordinator()); + Assert.assertEquals("A single view change must be reported", 1, coordinator.views.size()); + Assert.assertSame(view, coordinator.views.get(0)); + } + + @Test + public void testConcurrentInstallsOfSameViewNotifyOnce() throws Exception { + MemberImpl local = new MemberImpl("localhost", 4000, -1); + MemberImpl remote = new MemberImpl("localhost", 4001, -1); + RecordingCoordinator coordinator = createCoordinator(local); + DynamicMembershipChain chain = (DynamicMembershipChain) coordinator.getNext(); + chain.members = new Member[] { remote }; + + // Two membership callbacks on different threads computing the same view + Thread first = new Thread(() -> coordinator.memberAdded(remote)); + Thread second = new Thread(() -> coordinator.memberAdded(remote)); + first.start(); + second.start(); + first.join(60000); + second.join(60000); + Assert.assertFalse("First install thread did not finish", first.isAlive()); + Assert.assertFalse("Second install thread did not finish", second.isAlive()); + + Assert.assertEquals("The view must be installed", 2, coordinator.getView().length); + Assert.assertEquals("Installing the same view concurrently must notify once", + 1, coordinator.views.size()); + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
