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"); + } + }