From 4419f3e69ec815ec688ed8f3b55643cf70fb6a34 Mon Sep 17 00:00:00 2001 From: TuanNX Date: Tue, 6 Oct 2026 00:25:01 -0500 Subject: [PATCH 1/2] concurrent-api: reproduce lost and duplicated demand on resubscribe **Phase 1: test only. Expected red; production fix follows review.** #### Motivation Overlapping switches can lose outstanding demand or request it twice. Lost demand is the hypothesis investigated in apple/servicetalk#1672. #### Modifications - Stress 200,000 resubscriptions after source termination; require exactly 10 demand in every iteration. - Retain the latch-controlled duplicate-demand test and its serialized control. #### Result Local stress: **6 lost-demand cases (0), 10 duplicate-demand cases (20)**; the other 199,984 received 10. Counts vary with scheduling; this is not deterministic or an end-to-end proof of #1672. The latch test also reproduces **expected 1, actual 2**. Production code is unchanged. Validation: full concurrent-api suite has only these two regression failures; quality passes. A local serialized-order control passes all 200,000 iterations. --- .../api/SequentialSubscriptionTest.java | 111 ++++++++++++++++++ 1 file changed, 111 insertions(+) diff --git a/servicetalk-concurrent-api/src/test/java/io/servicetalk/concurrent/api/SequentialSubscriptionTest.java b/servicetalk-concurrent-api/src/test/java/io/servicetalk/concurrent/api/SequentialSubscriptionTest.java index 05fe964d15..0a88b2618b 100644 --- a/servicetalk-concurrent-api/src/test/java/io/servicetalk/concurrent/api/SequentialSubscriptionTest.java +++ b/servicetalk-concurrent-api/src/test/java/io/servicetalk/concurrent/api/SequentialSubscriptionTest.java @@ -25,9 +25,13 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; +import java.util.Map; +import java.util.TreeMap; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.CyclicBarrier; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; @@ -42,6 +46,7 @@ import static java.util.concurrent.Executors.newCachedThreadPool; import static java.util.concurrent.TimeUnit.SECONDS; import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.contains; import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.lessThanOrEqualTo; @@ -58,6 +63,8 @@ final class SequentialSubscriptionTest { private static final int ITERATIONS_FOR_CONCURRENT_TESTS = 500; + private static final int ITERATIONS_FOR_RACE_TESTS = 200_000; + private static final int EMIT_BATCH = 10; private SequentialSubscription s; private Subscription s1; private Subscription s2; @@ -237,6 +244,57 @@ void requestNLongMinAlwaysDirectedTowardSwitchedSubscription() throws Exception } } + @ParameterizedTest(name = "{displayName} [{index}]: overlap={0}") + @ValueSource(booleans = {false, true}) + void switchToAfterReentrantRequestTransfersOutstandingDemandOnce(boolean overlap) throws Exception { + final CountDownLatch requested = new CountDownLatch(1); + final CountDownLatch release = new CountDownLatch(1); + doAnswer(invocation -> { + s.itemReceived(); + s.request(1); + requested.countDown(); + assertThat("The old request was not released", release.await(DEFAULT_TIMEOUT_SECONDS, SECONDS), is(true)); + return null; + }).doNothing().when(s2).request(anyLong()); + s.request(1); + final CountingSubscription next = new CountingSubscription(); + final Future switching = executor.submit(() -> s.switchTo(s2)); + try { + assertThat("The old request was not reached", requested.await(DEFAULT_TIMEOUT_SECONDS, SECONDS), is(true)); + if (!overlap) { + release.countDown(); + switching.get(); + } + // The old source emitted its item; only its request callback may still be unwinding. + s.switchTo(next); + } finally { + release.countDown(); + switching.get(); + } + assertThat("Outstanding demand must be transferred once", next.requestedReceived(), is(1L)); + } + + @Test + @Timeout(60) // Deliberate stress loop; the latch does not force the handoff window. + void switchToWhileAnotherSwitchUnwindsRequestsOutstandingDemandFromNewSubscription() throws Exception { + final Map demands = new TreeMap<>(); + for (int i = 0; i < ITERATIONS_FOR_RACE_TESTS; ++i) { + final SequentialSubscription subscription = new SequentialSubscription(); + subscription.request(EMIT_BATCH); + final EmittingSubscription source = new EmittingSubscription(subscription, 2 * EMIT_BATCH, EMIT_BATCH); + final CountingSubscription next = new CountingSubscription(); + final Future resubscribe = executor.submit(() -> { + source.awaitTerminated(); + subscription.switchTo(next); + return null; + }); + subscription.switchTo(source); + resubscribe.get(); + demands.merge(next.requestedReceived(), 1, Integer::sum); + } + assertThat("Requested demand -> iterations: " + demands, demands.keySet(), contains((long) EMIT_BATCH)); + } + private void testConcurrentRequestEmitAndSwitch(int totalItems, int maxDeliveryPerSubscription) throws Exception { final SequentialSubscription subscription = new SequentialSubscription(); final CyclicBarrier allStarted = new CyclicBarrier(3); @@ -378,6 +436,59 @@ public String toString() { } } + /** + * Emits items synchronously from {@link #request(long)} like a {@link Publisher#range(int, int)} does. + */ + private static final class EmittingSubscription implements Subscription { + private final SequentialSubscription subscription; + private final int totalItems; + private final int requestMoreEvery; + private final CountDownLatch terminatedLatch = new CountDownLatch(1); + private long demand; + private int emitted; + private boolean emitting; + + EmittingSubscription(final SequentialSubscription subscription, final int totalItems, + final int requestMoreEvery) { + this.subscription = subscription; + this.totalItems = totalItems; + this.requestMoreEvery = requestMoreEvery; + } + + @Override + public void request(final long n) { + demand += n; + if (emitting) { + return; + } + emitting = true; + try { + while (demand > 0 && emitted < totalItems) { + --demand; + ++emitted; + subscription.itemReceived(); + if (emitted % requestMoreEvery == 0) { + subscription.request(requestMoreEvery); + } + } + } finally { + emitting = false; + } + if (emitted == totalItems) { + terminatedLatch.countDown(); + } + } + + @Override + public void cancel() { + } + + void awaitTerminated() throws InterruptedException { + assertThat("The emitting source did not terminate", + terminatedLatch.await(DEFAULT_TIMEOUT_SECONDS, SECONDS), is(true)); + } + } + private static Matcher matchValueOrCancelled( Matcher valueMatcher) { return new BaseMatcher() { From 506ce73e691f533bc13ef87fcd42622bc1f348fe Mon Sep 17 00:00:00 2001 From: TuanNX Date: Tue, 6 Oct 2026 13:44:49 -0500 Subject: [PATCH 2/2] concurrent-api: preserve demand across overlapping subscription switches #### Motivation A replacement arriving while a request callback unwinds can lose or duplicate outstanding demand. #### Modifications Publish replacements through an atomic mailbox and use the existing drain-lock helpers to serialize demand accounting and callbacks. Keep pending work visible through reentry and callback failures. Add terminal-action and demand-boundary coverage. #### Result The lost/duplicate-demand regressions pass, including 200,000 stress iterations. The concurrent-api suite and quality pass on JDK 17; focused tests pass on JDK 8. --- .../api/SequentialSubscription.java | 210 ++++++++---------- .../api/SequentialSubscriptionTest.java | 96 +++++++- 2 files changed, 182 insertions(+), 124 deletions(-) diff --git a/servicetalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/SequentialSubscription.java b/servicetalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/SequentialSubscription.java index cb33f5444f..246851064d 100644 --- a/servicetalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/SequentialSubscription.java +++ b/servicetalk-concurrent-api/src/main/java/io/servicetalk/concurrent/api/SequentialSubscription.java @@ -1,5 +1,5 @@ /* - * Copyright © 2018-2019 Apple Inc. and the ServiceTalk project authors + * Copyright © 2018-2026 Apple Inc. and the ServiceTalk project authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -19,10 +19,17 @@ import io.servicetalk.concurrent.PublisherSource.Subscription; import io.servicetalk.concurrent.internal.FlowControlUtils; +import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import java.util.concurrent.atomic.AtomicLongFieldUpdater; +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; +import javax.annotation.Nullable; +import static io.servicetalk.concurrent.internal.ConcurrentUtils.releaseLock; +import static io.servicetalk.concurrent.internal.ConcurrentUtils.tryAcquireLock; import static io.servicetalk.concurrent.internal.EmptySubscriptions.EMPTY_SUBSCRIPTION_NO_THROW; import static io.servicetalk.concurrent.internal.SubscriberUtils.isRequestNValid; +import static io.servicetalk.concurrent.internal.ThrowableUtils.catchUnexpected; +import static io.servicetalk.utils.internal.ThrowableUtils.throwException; import static java.util.Objects.requireNonNull; import static java.util.concurrent.atomic.AtomicLongFieldUpdater.newUpdater; @@ -39,18 +46,24 @@ * {@link #switchTo(Subscription)} will be cancelled. */ final class SequentialSubscription implements Subscription { - private static final long SWITCHING = -1; - private static final long REQUESTED = -2; - private static final long CANCELLED = -3; private static final AtomicLongFieldUpdater requestedUpdater = newUpdater(SequentialSubscription.class, "requested"); - private static final AtomicLongFieldUpdater sourceRequestedUpdater = - newUpdater(SequentialSubscription.class, "sourceRequested"); + private static final AtomicIntegerFieldUpdater emittingUpdater = + AtomicIntegerFieldUpdater.newUpdater(SequentialSubscription.class, "emitting"); + private static final AtomicReferenceFieldUpdater pendingSubscriptionUpdater = + AtomicReferenceFieldUpdater.newUpdater(SequentialSubscription.class, Subscription.class, + "pendingSubscription"); + // Only the drain owner changes the active subscription and its demand accounting. private Subscription subscription; + private long sourceRequested; + // Subscriber signals are serialized. Publishing the next subscription also publishes this count to the drain. private long sourceEmitted; private volatile long requested; - private volatile long sourceRequested; + private volatile int emitting; + private volatile boolean cancelled; + @Nullable + private volatile Subscription pendingSubscription; /** * New instance. @@ -70,142 +83,93 @@ final class SequentialSubscription implements Subscription { @Override public void request(long n) { - final long currRequested; if (isRequestNValid(n)) { - currRequested = requestedUpdater.accumulateAndGet(this, n, + requestedUpdater.accumulateAndGet(this, n, FlowControlUtils::addWithOverflowProtectionIfNotNegative); } else { - currRequested = sanitizeInvalidRequestN(n); - requested = currRequested; - } - - for (;;) { - final long currSourceRequested = sourceRequested; - if (currSourceRequested == CANCELLED) { - break; - } else if (currSourceRequested < 0) { - assert currSourceRequested == SWITCHING || currSourceRequested == REQUESTED; - if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, REQUESTED)) { - break; - } - } else { - // We must read the subscription BEFORE the CAS (which involves a read barrier). This ensures if the - // sourceRequested value is visible then the subscription (which may have been switched by - // another thread) is also visible. - final Subscription currSubscription = subscription; - if (isRequestNValid(currRequested)) { - // sourceRequested ...[delta]... requested - final long delta = currRequested - currSourceRequested; - if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, currSourceRequested + delta)) { - // sourceRequested is either monotonically increasing, or set to an invalid value - // (e.g. negative) if a Subscription switch is on going and atomically set to requestN to - // preserve the monotonic increasing property. If the CAS worked that means the value of - // subscription before will be visible if there was previously a switch. We also know there is - // no concurrent interaction on the subscription because currSourceRequested is known not to be - // SWITCHING, and the value would have increased. - if (delta != 0) { - currSubscription.request(delta); - } - break; - } - } else if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, CANCELLED)) { - currSubscription.request(currRequested); - break; - } - } + requested = sanitizeInvalidRequestN(n); } + drain(); } @Override public void cancel() { - final Subscription currSubscription = subscription; - final long currSourceRequested = sourceRequestedUpdater.getAndSet(this, CANCELLED); - // To avoid concurrent invocation with the switch thread we defer to that thread to cancel. - if (currSourceRequested >= 0) { - currSubscription.cancel(); - } + cancelled = true; + drain(); } /** - * Switches current {@link Subscription} to {@code next}. It is assumed the {@link Subscriber} - * associated with the previous {@link Subscription} will no longer call {@link #itemReceived()}. - *

- * Only can be called in the {@link Subscriber} thread! + * Switches current {@link Subscription} to {@code next}. The previous source must no longer call + * {@link #itemReceived()}. Subscriber signals must be serialized, but a previous request callback may still be + * unwinding on another thread. * @param next {@link Subscription} that should now be current. */ void switchTo(Subscription next) { - requireNonNull(next); - // No special concurrency considerations for sourceEmitted access is required in this method because we are - // on the Subscriber thread in this method. We want to track the effective source requested for the purposes of - // how much more request(n) is necessary below. - long effectiveSourceRequested = sourceEmitted; - for (;;) { - final long currSourceRequested = sourceRequested; - if (currSourceRequested == CANCELLED) { - final long currRequested = requested; - if (currRequested >= 0) { - next.cancel(); - } else { // invalid requestN is pending, deliver to each subscription. - next.request(currRequested); - } - break; - } else if (currSourceRequested == SWITCHING) { - // concurrency is not allowed on this method, but reentry is allowed. save next into subscription so - // we can use it when the stack unwinds to deliver demand to it. - subscription = next; - break; - } else if (sourceRequestedUpdater.compareAndSet(this, currSourceRequested, SWITCHING)) { - assert currSourceRequested >= 0 || currSourceRequested == REQUESTED; - final long currRequested = requested; - if (currRequested < 0) { // invalid requestN is pending. - sourceRequested = CANCELLED; - next.request(currRequested); - break; - } - - // sourceEmitted is stable here because we are on the Subscriber thread. We want to request the - // difference between total requested and what has been emitted from the new subscription. We also - // need to set the value of total requested below to make sure it is monotonically increasing. - // effectiveSourceRequested ...[delta]... requested - final long delta = currRequested - effectiveSourceRequested; - assert delta >= 0; - final Subscription beforeSubscription = subscription; - if (delta != 0) { - // There maybe concurrency with the Subscription thread, or synchronous delivery of data from - // request(n). In these cases we want to avoid "double request" from requested, so we track how much - // we have already requested and decrement it on future loop iterations. - effectiveSourceRequested = currRequested; - next.request(delta); - } - - final boolean reentry = beforeSubscription != subscription; - if (reentry) { - // subscription was overwritten higher in the stack to track the more recent value. overwrite next - // in our current stack frame to use the more recent value on the next loop iteration. - next = subscription; - - // There is a new subscription so we need to reset state for how much has been emitted, so we - // deliver demand to the more recent subscription on the next loop iteration. - effectiveSourceRequested = sourceEmitted; + final Subscription previous = pendingSubscriptionUpdater.getAndSet(this, requireNonNull(next)); + try { + // A source may terminate without demand before the drain reaches it. Only the latest source needs demand, + // but a displaced subscription must still observe a pending terminal action. + if (previous != null) { + if (cancelled) { + previous.cancel(); } else { - // Make the subscription visible before restoring the state of sourceRequested. If the Subscription - // thread observes the sourceRequested change it will also observe the subscription change. The - // Subscription thread also uses sourceRequested to make sure there is no concurrent invocation of - // the switched Subscription. - subscription = next; + final long n = requested; + if (n < 0) { + previous.request(n); + } } + } + } finally { + drain(); + } + } - // We want to set sourceRequested to currRequested because we have already requested the delta between - // the two above, and we want to ensure sourceRequested is always monotonically increasing - // (besides control values) to prevent the Subscription thread from requesting from an old subscription. - if (sourceRequestedUpdater.compareAndSet(this, SWITCHING, currRequested) && !reentry) { - break; + private void drain() { + Throwable delayedCause = null; + boolean tryAcquire = true; + while (tryAcquire && tryAcquireLock(emittingUpdater, this)) { + try { + final Subscription next = pendingSubscriptionUpdater.getAndSet(this, null); + if (cancelled) { + final Subscription current = subscription; + subscription = EMPTY_SUBSCRIPTION_NO_THROW; + try { + current.cancel(); + } finally { + if (next != null) { + next.cancel(); + } + } + } else { + if (next != null) { + subscription = next; + sourceRequested = sourceEmitted; + } + final long n = requested; + if (sourceRequested >= 0) { + if (n < 0) { + sourceRequested = n; + subscription.request(n); + } else { + final long delta = n - sourceRequested; + if (delta != 0) { + // Commit before the callback: reentrant request/switch calls only enqueue more work. + sourceRequested = n; + subscription.request(delta); + } + } + } } - // else the Subscription thread was active in the mean time, we need to process the pending event(s). - // if the state is cancelled the Subscription thread defers to this thread to do the cancel on the - // next loop invocation. or this method was invoked in a re-entry fashion and we need to loop again. + } catch (Throwable cause) { + delayedCause = catchUnexpected(delayedCause, cause); + } finally { + // A publisher racing with release either marks pending work or becomes the next drain owner. + tryAcquire = !releaseLock(emittingUpdater, this); } } + if (delayedCause != null) { + throwException(delayedCause); + } } /** diff --git a/servicetalk-concurrent-api/src/test/java/io/servicetalk/concurrent/api/SequentialSubscriptionTest.java b/servicetalk-concurrent-api/src/test/java/io/servicetalk/concurrent/api/SequentialSubscriptionTest.java index 0a88b2618b..c79a21c231 100644 --- a/servicetalk-concurrent-api/src/test/java/io/servicetalk/concurrent/api/SequentialSubscriptionTest.java +++ b/servicetalk-concurrent-api/src/test/java/io/servicetalk/concurrent/api/SequentialSubscriptionTest.java @@ -33,6 +33,7 @@ import java.util.TreeMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicLong; @@ -58,6 +59,7 @@ import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoMoreInteractions; @@ -81,7 +83,8 @@ void setUp() { @AfterEach void tearDown() throws Exception { executor.shutdownNow(); - executor.awaitTermination(DEFAULT_TIMEOUT_SECONDS, SECONDS); + assertThat("Executor did not terminate", + executor.awaitTermination(DEFAULT_TIMEOUT_SECONDS, SECONDS), is(true)); } @ParameterizedTest @@ -295,6 +298,97 @@ void switchToWhileAnotherSwitchUnwindsRequestsOutstandingDemandFromNewSubscripti assertThat("Requested demand -> iterations: " + demands, demands.keySet(), contains((long) EMIT_BATCH)); } + @Test + void switchToWithoutOutstandingDemandReceivesSubsequentRequest() { + s.request(1); + s.itemReceived(); + s.switchTo(s2); + verifyNoMoreInteractions(s2); + s.request(1); + verify(s2).request(1); + verify(s1).request(1); + verifyNoMoreInteractions(s1, s2); + } + + @Test + void switchToAfterRequestOverflowPreservesOutstandingDemand() { + s.request(MAX_VALUE - 1); + s.request(2); + s.itemReceived(); + s.switchTo(s2); + verify(s2).request(MAX_VALUE - 1); + verifyNoMoreInteractions(s2); + } + + @Test + void requestThrowsAfterSwitchStillDrainsPendingDemand() { + doAnswer(invocation -> { + s.itemReceived(); + s.switchTo(s2); + s.request(1); + throw DELIBERATE_EXCEPTION; + }).when(s1).request(anyLong()); + assertThrows(DeliberateException.class, () -> s.request(1)); + verify(s2).request(1); + s.request(1); + verify(s2, times(2)).request(1); + verifyNoMoreInteractions(s2); + } + + @ParameterizedTest(name = "{displayName} [{index}]: requestThrows={0}") + @ValueSource(booleans = {false, true}) + void cancelDuringRequestCancelsActiveAndPendingSubscriptions(boolean requestThrows) throws Exception { + final CountDownLatch requested = new CountDownLatch(1); + final CountDownLatch release = new CountDownLatch(1); + doAnswer(invocation -> { + requested.countDown(); + assertThat(release.await(DEFAULT_TIMEOUT_SECONDS, SECONDS), is(true)); + if (requestThrows) { + throw DELIBERATE_EXCEPTION; + } + return null; + }).when(s1).request(anyLong()); + final Subscription s3 = mock(Subscription.class); + final Future requesting = executor.submit(() -> s.request(1)); + try { + assertThat(requested.await(DEFAULT_TIMEOUT_SECONDS, SECONDS), is(true)); + s.switchTo(s2); + s.cancel(); + s.switchTo(s3); + } finally { + release.countDown(); + if (requestThrows) { + assertThat(assertThrows(ExecutionException.class, requesting::get).getCause(), + is(DELIBERATE_EXCEPTION)); + } else { + requesting.get(); + } + } + verify(s1).request(1); + verify(s1).cancel(); + verify(s2).cancel(); + verify(s3).cancel(); + verifyNoMoreInteractions(s1, s2, s3); + } + + @ParameterizedTest(name = "{displayName} [{index}]: n={0}") + @ValueSource(longs = {-1, 0, MIN_VALUE}) + void invalidRequestDuringSwitchReachesPendingSubscriptions(long n) { + final Subscription s3 = mock(Subscription.class); + doAnswer(invocation -> { + s.switchTo(s2); + s.request(n); + s.switchTo(s3); + return null; + }).doNothing().when(s1).request(anyLong()); + s.request(1); + final long expected = n == 0 ? MIN_VALUE : n; + verify(s3).request(expected); + verify(s2).request(expected); + s.request(1); + verifyNoMoreInteractions(s2, s3); + } + private void testConcurrentRequestEmitAndSwitch(int totalItems, int maxDeliveryPerSubscription) throws Exception { final SequentialSubscription subscription = new SequentialSubscription(); final CyclicBarrier allStarted = new CyclicBarrier(3);