From 493fb150a043ef1582461a2424e6828e1ba87894 Mon Sep 17 00:00:00 2001 From: donggyu Date: Thu, 20 Aug 2026 17:07:05 +0900 Subject: [PATCH] Fix busy-spin in SubscriberInputStream await() LockSupport.park() can return without resume() having set the parked thread reference to READY, either spuriously or when the thread is interrupted. The reference then still points to the current thread, so the compareAndSet in the await() loop can never succeed again and the loop neither parks nor exits, spinning on CPU until the upstream emits the next signal. On virtual threads this is particularly harmful since a spinning thread never yields its carrier. Once as many requests spin as there are carrier threads, unrelated virtual thread tasks are no longer scheduled. The interrupt case is reachable through Spring MVC async request handling, where both a timeout and a client disconnect cancel the task with interruption. Park again when the reference is still owned by the current thread after a spurious wakeup. On interruption, clear the reference and propagate the cancellation, unless data has arrived concurrently, in which case the data is delivered first and the interrupt status is preserved for the next call. Closes gh-37159 Signed-off-by: donggyu --- .../core/io/buffer/SubscriberInputStream.java | 15 ++++++- .../http/client/SubscriberInputStream.java | 15 ++++++- .../client/SubscriberInputStreamTests.java | 41 +++++++++++++++++++ 3 files changed, 69 insertions(+), 2 deletions(-) diff --git a/spring-core/src/main/java/org/springframework/core/io/buffer/SubscriberInputStream.java b/spring-core/src/main/java/org/springframework/core/io/buffer/SubscriberInputStream.java index 65ac925b3949..f8597de99366 100644 --- a/spring-core/src/main/java/org/springframework/core/io/buffer/SubscriberInputStream.java +++ b/spring-core/src/main/java/org/springframework/core/io/buffer/SubscriberInputStream.java @@ -391,10 +391,23 @@ private void await() { throw new IllegalStateException("Only one (Virtual)Thread can await!"); } - if (this.parkedThread.compareAndSet( null, toUnpark)) { + if (this.parkedThread.compareAndSet(null, toUnpark) || this.parkedThread.get() == toUnpark) { LockSupport.park(); // we don't just break here because park() can wake up spuriously // if we got a proper resume, get() == READY and the loop will quit above + // if we woke up spuriously, the reference is still ours and we park again + } + + // park() also returns on interruption, without a resume() having set READY, + // in which case the loop could otherwise neither park again nor exit. + // Honor the cancellation, unless data arrived concurrently: then deliver it + // first and leave the thread interrupted for the next await() call. + if (Thread.interrupted()) { + if (this.parkedThread.get() != READY) { + this.parkedThread.lazySet(null); + throw new IllegalStateException("Interrupted while awaiting data"); + } + Thread.currentThread().interrupt(); } } // clear the resume indicator so that the next await call will park without a resume() diff --git a/spring-web/src/main/java/org/springframework/http/client/SubscriberInputStream.java b/spring-web/src/main/java/org/springframework/http/client/SubscriberInputStream.java index 9967b2a1d953..23ba8e3c07fb 100644 --- a/spring-web/src/main/java/org/springframework/http/client/SubscriberInputStream.java +++ b/spring-web/src/main/java/org/springframework/http/client/SubscriberInputStream.java @@ -420,10 +420,23 @@ private void await() { throw new IllegalStateException("Only one (Virtual)Thread can await!"); } - if (this.parkedThread.compareAndSet( null, toUnpark)) { + if (this.parkedThread.compareAndSet(null, toUnpark) || this.parkedThread.get() == toUnpark) { LockSupport.park(); // we don't just break here because park() can wake up spuriously // if we got a proper resume, get() == READY and the loop will quit above + // if we woke up spuriously, the reference is still ours and we park again + } + + // park() also returns on interruption, without a resume() having set READY, + // in which case the loop could otherwise neither park again nor exit. + // Honor the cancellation, unless data arrived concurrently: then deliver it + // first and leave the thread interrupted for the next await() call. + if (Thread.interrupted()) { + if (this.parkedThread.get() != READY) { + this.parkedThread.lazySet(null); + throw new IllegalStateException("Interrupted while awaiting data"); + } + Thread.currentThread().interrupt(); } } // clear the resume indicator so that the next await call will park without a resume() diff --git a/spring-web/src/test/java/org/springframework/http/client/SubscriberInputStreamTests.java b/spring-web/src/test/java/org/springframework/http/client/SubscriberInputStreamTests.java index e4924e03f2f2..e7d75b8d89e7 100644 --- a/spring-web/src/test/java/org/springframework/http/client/SubscriberInputStreamTests.java +++ b/spring-web/src/test/java/org/springframework/http/client/SubscriberInputStreamTests.java @@ -24,6 +24,7 @@ import java.util.concurrent.Executor; import java.util.concurrent.Executors; import java.util.concurrent.Flow; +import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Test; @@ -233,4 +234,44 @@ void mapperThrowsException() throws InterruptedException { assertThat(savedEx).hasMessage("boom"); } + @Test // gh-37159 + void interruptWhileAwaitingData() throws InterruptedException { + CountDownLatch reading = new CountDownLatch(1); + AtomicReference savedEx = new AtomicReference<>(); + + // A publisher that never emits, so that read() parks in await() + Flow.Publisher publisher = subscriber -> subscriber.onSubscribe(new Flow.Subscription() { + @Override + public void request(long n) { + } + @Override + public void cancel() { + } + }); + + Thread reader = new Thread(() -> { + try (SubscriberInputStream is = new SubscriberInputStream<>(s -> s, s -> {}, 1)) { + publisher.subscribe(is); + reading.countDown(); + is.read(); + } + catch (Throwable ex) { + savedEx.set(ex); + } + }); + reader.start(); + + reading.await(); + for (int i = 0; i < 100 && reader.getState() != Thread.State.WAITING; i++) { + Thread.sleep(20); + } + assertThat(reader.getState()).isEqualTo(Thread.State.WAITING); + + reader.interrupt(); + reader.join(5000); + + assertThat(reader.isAlive()).as("read() did not return after interrupt").isFalse(); + assertThat(savedEx.get()).hasMessage("Interrupted while awaiting data"); + } + }