diff --git a/stub/src/main/java/io/grpc/stub/BlockingClientCall.java b/stub/src/main/java/io/grpc/stub/BlockingClientCall.java index 6a52ce50776..a68380dc2e5 100644 --- a/stub/src/main/java/io/grpc/stub/BlockingClientCall.java +++ b/stub/src/main/java/io/grpc/stub/BlockingClientCall.java @@ -241,7 +241,8 @@ void sendSingleRequest(ReqT request) { /** * Cancel stream and stop any further writes. Note that some reads that are in flight may still - * happen after the cancel. + * happen after the cancel. Reads that have been received but not yet delivered are processed + * when cancelling, releasing their resources. * * @param message if not {@code null}, will appear as the description of the CANCELLED status * @param cause if not {@code null}, will appear as the cause of the CANCELLED status @@ -249,6 +250,10 @@ void sendSingleRequest(ReqT request) { public void cancel(String message, Throwable cause) { writeClosed = true; call.cancel(message, cause); + // Reads queued by the transport hold buffers that are only released when their delivery task + // runs and observes the cancelled stream. Nothing else will drain the executor once the user + // abandons the call, so the queued tasks must be processed here. + executor.drain(); } /** @@ -278,6 +283,11 @@ Status getClosedStatus() { return (state == null) ? null : state.status; } + @VisibleForTesting + ThreadSafeThreadlessExecutor getExecutor() { + return executor; + } + /** * Check for whether some action is ready. * diff --git a/stub/src/test/java/io/grpc/stub/BlockingClientCallTest.java b/stub/src/test/java/io/grpc/stub/BlockingClientCallTest.java index e3a4f90e2c2..abb1e1661d7 100644 --- a/stub/src/test/java/io/grpc/stub/BlockingClientCallTest.java +++ b/stub/src/test/java/io/grpc/stub/BlockingClientCallTest.java @@ -221,6 +221,23 @@ public void testCancel() throws Exception { } + @Test + public void testCancelDrainsPendingReads() throws Exception { + biDiStream = ClientCalls.blockingBidiStreamingCall(channel, BIDI_STREAMING_METHOD, + CallOptions.DEFAULT); + + // The server delivering a message queues a read-delivery task in the call executor, which + // only runs when the user reads/writes or the call is cancelled + testMethod.sendValueToClient(60); + assertThat(biDiStream.getExecutor()).isNotEmpty(); + + biDiStream.cancel("done reading", null); + + // The queued read task must be processed by cancel() itself; when orphaned it holds the + // message's transport buffers, which leak (see #12355) + assertThat(biDiStream.getExecutor()).isEmpty(); + } + @Test public void testIsActivityReady() throws Exception { biDiStream = ClientCalls.blockingBidiStreamingCall(channel, BIDI_STREAMING_METHOD,