[Driver/Java] Fix a race condition when there are only spies and spies simulate connection, i.e. adding/removing spies mutated the array that the sender thread was iterating over. Now conductor computes max spy position while processing stream counters and share it via a volatile long field.

vyazelenko · vyazelenko · commit 1c06aad8fac8 · 2026-06-30T22:20:40.000+02:00
(cherry picked from commit 1dde9e3)
diff --git a/aeron-driver/src/main/java/io/aeron/driver/NetworkPublication.java b/aeron-driver/src/main/java/io/aeron/driver/NetworkPublication.java
@@ -38,6 +38,8 @@
 import org.agrona.concurrent.status.Position;
 import org.agrona.concurrent.status.ReadablePosition;
 
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
@@ -79,12 +81,27 @@ class NetworkPublicationPadding1
 class NetworkPublicationConductorFields extends NetworkPublicationPadding1
 {
     static final ReadablePosition[] EMPTY_POSITIONS = new ReadablePosition[0];
+    static final VarHandle MAX_SPY_POSITION_MH;
+
+    static
+    {
+        try
+        {
+            MAX_SPY_POSITION_MH = MethodHandles.lookup()
+                .findVarHandle(NetworkPublicationConductorFields.class, "maxSpyPosition", long.class);
+        }
+        catch (final ReflectiveOperationException ex)
+        {
+            throw new ExceptionInInitializerError(ex);
+        }
+    }
 
     long cleanPosition;
     long timeOfLastActivityNs;
     long lastSenderPosition;
     int refCount = 0;
     ReadablePosition[] spyPositions = EMPTY_POSITIONS;
+    volatile long maxSpyPosition;
     final ArrayList<UntetheredSubscription> untetheredSubscriptions = new ArrayList<>();
 }
 
@@ -677,7 +694,7 @@ int send(final long nowNs)
 
             if (spiesSimulateConnection && hasSpies && !hasReceivers)
             {
-                final long newSenderPosition = maxSpyPosition(senderPosition);
+                final long newSenderPosition = Math.max((long)MAX_SPY_POSITION_MH.getAcquire(this), senderPosition);
                 this.senderPosition.setRelease(newSenderPosition);
                 senderLimit.setRelease(flowControl.onIdle(nowNs, newSenderPosition, newSenderPosition, isEndOfStream));
             }
@@ -779,10 +796,14 @@ int updatePublisherPositionAndLimit()
             if (hasSubscribers())
             {
                 long minConsumerPosition = senderPosition;
+                long maxConsumerPosition = senderPosition;
                 for (final ReadablePosition spyPosition : spyPositions)
                 {
-                    minConsumerPosition = Math.min(minConsumerPosition, spyPosition.getVolatile());
+                    final long position = spyPosition.getAcquire();
+                    minConsumerPosition = Math.min(minConsumerPosition, position);
+                    maxConsumerPosition = Math.max(maxConsumerPosition, position);
                 }
+                MAX_SPY_POSITION_MH.setRelease(this, maxConsumerPosition);
 
                 final long newLimitPosition = minConsumerPosition + termWindowLength;
                 if (newLimitPosition > publisherLimit.get())
@@ -1029,18 +1050,6 @@ private boolean spiesFinishedConsuming(final DriverConductor conductor, final lo
         return true;
     }
 
-    private long maxSpyPosition(final long senderPosition)
-    {
-        long position = senderPosition;
-
-        for (final ReadablePosition spyPosition : spyPositions)
-        {
-            position = Math.max(position, spyPosition.getVolatile());
-        }
-
-        return position;
-    }
-
     private void updateConnectedState(final boolean newConnectedState)
     {
         if (newConnectedState != isConnected)
diff --git a/aeron-system-tests/src/test/java/io/aeron/SpySimulatedConnectionTest.java b/aeron-system-tests/src/test/java/io/aeron/SpySimulatedConnectionTest.java
@@ -29,6 +29,7 @@
 import io.aeron.test.driver.TestMediaDriver;
 import org.agrona.CloseHelper;
 import org.agrona.collections.MutableInteger;
+import org.agrona.concurrent.NoOpIdleStrategy;
 import org.agrona.concurrent.UnsafeBuffer;
 import org.agrona.concurrent.status.CountersReader;
 import org.junit.jupiter.api.AfterEach;
@@ -85,14 +86,17 @@ private static List<String> channels()
     @RegisterExtension
     final SystemTestWatcher watcher = new SystemTestWatcher();
 
-    private void launch()
+    private void launch(final ThreadingMode threadingMode)
     {
         driverContext
             .aeronDirectoryName(CommonContext.generateRandomDirName())
             .publicationTermBufferLength(TERM_BUFFER_LENGTH)
             .errorHandler(Tests::onError)
             .dirDeleteOnStart(true)
-            .threadingMode(ThreadingMode.SHARED);
+            .threadingMode(threadingMode)
+            .senderIdleStrategy(NoOpIdleStrategy.INSTANCE)
+            .receiverIdleStrategy(NoOpIdleStrategy.INSTANCE)
+            .conductorIdleStrategy(NoOpIdleStrategy.INSTANCE);
 
         driver = TestMediaDriver.launch(driverContext, watcher);
         watcher.dataCollector().add(driver.context().aeronDirectory());
@@ -111,7 +115,7 @@ void after()
     @InterruptAfter(10)
     void shouldNotSimulateConnectionWhenNotConfigured(final String channel)
     {
-        launch();
+        launch(ThreadingMode.SHARED);
 
         spy = client.addSubscription(spyForChannel(channel), STREAM_ID);
         publication = client.addPublication(channel, STREAM_ID);
@@ -126,7 +130,7 @@ void shouldNotSimulateConnectionWhenNotConfigured(final String channel)
     @InterruptAfter(10)
     void shouldSimulateConnectionWhenOnChannel(final String channel)
     {
-        launch();
+        launch(ThreadingMode.SHARED);
 
         spy = client.addSubscription(spyForChannel(channel), STREAM_ID);
         publication = client.addPublication(channel + "|ssc=true", STREAM_ID);
@@ -147,7 +151,7 @@ void shouldSimulateConnectionWithNoNetworkSubscriptions(final String channel)
             .timerIntervalNs(TimeUnit.MILLISECONDS.toNanos(100))
             .spiesSimulateConnection(true);
 
-        launch();
+        launch(ThreadingMode.SHARED);
 
         spy = client.addSubscription(spyForChannel(channel), STREAM_ID);
         publication = client.addPublication(channel, STREAM_ID);
@@ -198,7 +202,7 @@ void shouldSimulateConnectionWithSlowNetworkSubscription(final String channel)
             .timerIntervalNs(TimeUnit.MILLISECONDS.toNanos(100))
             .spiesSimulateConnection(true);
 
-        launch();
+        launch(ThreadingMode.SHARED);
 
         spy = client.addSubscription(spyForChannel(channel), STREAM_ID);
         subscription = client.addSubscription(channel, STREAM_ID);
@@ -247,7 +251,7 @@ void shouldSimulateConnectionWithLeavingNetworkSubscription(final String channel
             .timerIntervalNs(TimeUnit.MILLISECONDS.toNanos(100))
             .spiesSimulateConnection(true);
 
-        launch();
+        launch(ThreadingMode.SHARED);
 
         spy = client.addSubscription(spyForChannel(channel), STREAM_ID);
         subscription = client.addSubscription(channel, STREAM_ID);
@@ -290,7 +294,7 @@ else if (!isSubscriptionClosed)
     @InterruptAfter(10)
     void shouldNotHaveShortSendsWithMDCPublication()
     {
-        launch();
+        launch(ThreadingMode.SHARED);
 
         final String channel = "aeron:udp?control=localhost:20000|control-mode=dynamic";
 
@@ -323,7 +327,7 @@ void shouldNotHaveShortSendsWithMDCPublication()
     void shouldNotChangeConnectionStatusOnPublicationIfNormalSubscriberGoesAway(final String channel)
         throws InterruptedException
     {
-        launch();
+        launch(ThreadingMode.SHARED);
 
         spy = client.addSubscription(spyForChannel(channel), STREAM_ID);
         subscription = client.addSubscription(channel, STREAM_ID);
@@ -377,6 +381,64 @@ void shouldNotChangeConnectionStatusOnPublicationIfNormalSubscriberGoesAway(fina
         assertTrue(spy.isConnected());
     }
 
+    @ParameterizedTest
+    @MethodSource("channels")
+    @InterruptAfter(10)
+    void addingAndRemovingSpiesIsThreadSafe(final String channel) throws InterruptedException
+    {
+        launch(ThreadingMode.DEDICATED);
+
+        final String spyChannel = spyForChannel(channel);
+        spy = client.addSubscription(spyChannel, STREAM_ID);
+        publication = client.addPublication(channel + "|ssc=true", STREAM_ID);
+        Tests.awaitConnected(publication);
+        Tests.awaitConnected(spy);
+
+        final AtomicBoolean running = new AtomicBoolean(true);
+        final Thread sender = new Thread(() ->
+        {
+            while (running.get())
+            {
+                if (publication.offer(buffer) < 0)
+                {
+                    Tests.yield();
+                }
+            }
+        });
+        sender.start();
+
+        final long[] spyRegistrationIds = new long[100];
+        for (int i = 0; i < spyRegistrationIds.length; i++)
+        {
+            spyRegistrationIds[i] = client.asyncAddSubscription(spyChannel, STREAM_ID);
+        }
+
+        final Subscription[] spies = new Subscription[spyRegistrationIds.length];
+
+        int count;
+        do
+        {
+            count = 0;
+            for (int i = 0; i < spyRegistrationIds.length; i++)
+            {
+                if (null == spies[i])
+                {
+                    spies[i] = client.getSubscription(spyRegistrationIds[i]);
+                }
+                else
+                {
+                    count++;
+                }
+            }
+        }
+        while (spyRegistrationIds.length != count);
+
+        CloseHelper.closeAll(spies);
+
+        running.set(false);
+        sender.join();
+    }
+
     private void waitUntilFullConnectivity()
     {
         while (!spy.isConnected() || !subscription.isConnected() || !publication.isConnected())
diff --git a/aeron-test-support/src/main/java/io/aeron/test/driver/CTestMediaDriver.java b/aeron-test-support/src/main/java/io/aeron/test/driver/CTestMediaDriver.java
@@ -28,6 +28,7 @@
 import org.agrona.collections.MutableInteger;
 import org.agrona.collections.Object2ObjectHashMap;
 import org.agrona.concurrent.AgentInvoker;
+import org.agrona.concurrent.IdleStrategy;
 import org.agrona.concurrent.status.CountersReader;
 
 import java.io.File;
@@ -36,6 +37,7 @@
 import java.nio.file.Files;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
 
 public final class CTestMediaDriver implements TestMediaDriver
 {
@@ -239,6 +241,11 @@ public static CTestMediaDriver launch(
         {
             environment.put("AERON_THREADING_MODE", context.threadingMode().name());
         }
+        setIdleStrategy(environment, "AERON_CONDUCTOR_IDLE_STRATEGY", context::conductorIdleStrategy);
+        setIdleStrategy(environment, "AERON_SENDER_IDLE_STRATEGY", context::senderIdleStrategy);
+        setIdleStrategy(environment, "AERON_RECEIVER_IDLE_STRATEGY", context::receiverIdleStrategy);
+        setIdleStrategy(environment, "AERON_SHARED_IDLE_STRATEGY", context::sharedIdleStrategy);
+        setIdleStrategy(environment, "AERON_SHAREDNETWORK_IDLE_STRATEGY", context::sharedNetworkIdleStrategy);
         environment.put("AERON_TIMER_INTERVAL", String.valueOf(context.timerIntervalNs()));
         environment.put(
             "AERON_UNTETHERED_WINDOW_LIMIT_TIMEOUT", String.valueOf(context.untetheredWindowLimitTimeoutNs()));
@@ -345,6 +352,18 @@ public static CTestMediaDriver launch(
         }
     }
 
+    private static void setIdleStrategy(
+        final HashMap<String, String> environment,
+        final String envKey,
+        final Supplier<IdleStrategy> idleStrategySupplier)
+    {
+        final IdleStrategy idleStrategy = idleStrategySupplier.get();
+        if (null != idleStrategy)
+        {
+            environment.put(envKey, idleStrategy.alias());
+        }
+    }
+
     public MediaDriver.Context context()
     {
         return context;
