core: reduce extra onReady() from hedging.

larry-safran · Sep 21, 2024 · 6337327 · 6337327
1 parent 782a44a
commit 6337327
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 12 deletions.
diff --git a/core/src/main/java/io/grpc/internal/RetriableStream.java b/core/src/main/java/io/grpc/internal/RetriableStream.java
@@ -311,7 +311,7 @@ private void drain(Substream substream) {
         }
         if (index == savedState.buffer.size()) { // I'm drained
           state = savedState.substreamDrained(substream);
-          if (!isReady()) {
+          if (!substream.stream.isReady()) {
             return;
           }
           onReadyRunnable = new Runnable() {
@@ -1131,15 +1131,15 @@ public void run() {
 
     @Override
     public void onReady() {
-      // FIXME(#7089): hedging case is broken.
-      if (!isReady()) {
+      if (!substream.stream.isReady()
+          || (state.winningSubstream != null && state.winningSubstream != substream)) {
         return;
       }
       listenerSerializeExecutor.execute(
           new Runnable() {
             @Override
             public void run() {
-              if (!isClosed) {
+              if (!isClosed && substream.stream.isReady()) {
                 masterListener.onReady();
               }
             }

diff --git a/core/src/test/java/io/grpc/internal/RetriableStreamTest.java b/core/src/test/java/io/grpc/internal/RetriableStreamTest.java
@@ -2150,7 +2150,6 @@ public Void answer(InvocationOnMock in) {
     inOrder.verify(mockStream2).writeMessage(any(InputStream.class));
     inOrder.verify(mockStream2).flush(); // Memory leak workaround
     inOrder.verify(mockStream2).request(456);
-    inOrder.verify(mockStream1).isReady();
     inOrder.verify(mockStream2).isReady();
     inOrder.verifyNoMoreInteractions();
 
@@ -2200,8 +2199,8 @@ public Void answer(InvocationOnMock in) {
     inOrder.verify(mockStream3).flush(); // Memory leak workaround
     inOrder.verify(mockStream3).writeMessage(any(InputStream.class));
     inOrder.verify(mockStream3).flush(); // Memory leak workaround
-    inOrder.verify(mockStream1).isReady();
-    inOrder.verify(mockStream2).isReady();
+    inOrder.verify(mockStream1, never()).isReady();
+    inOrder.verify(mockStream2, never()).isReady();
     inOrder.verify(mockStream3).isReady();
     inOrder.verifyNoMoreInteractions();
 
@@ -2257,8 +2256,6 @@ public Void answer(InvocationOnMock in) {
       inOrder.verify(mockStream4).writeMessage(any(InputStream.class));
       inOrder.verify(mockStream4).flush(); // Memory leak workaround
     }
-    inOrder.verify(mockStream1).isReady();
-    inOrder.verify(mockStream2).isReady();
     inOrder.verify(mockStream4).isReady();
     inOrder.verifyNoMoreInteractions();
 
@@ -2701,7 +2698,6 @@ public void hedging_cancelled() {
     ArgumentCaptor<ClientStreamListener> sublistenerCaptor2 =
         ArgumentCaptor.forClass(ClientStreamListener.class);
     inOrder.verify(mockStream2).start(sublistenerCaptor2.capture());
-    inOrder.verify(mockStream1).isReady();
     inOrder.verify(mockStream2).isReady();
     inOrder.verifyNoMoreInteractions();
 
@@ -2744,7 +2740,6 @@ public void hedging_perRpcBufferLimitExceeded() {
     ArgumentCaptor<ClientStreamListener> sublistenerCaptor2 =
         ArgumentCaptor.forClass(ClientStreamListener.class);
     verify(mockStream2).start(sublistenerCaptor2.capture());
-    verify(mockStream1, times(2)).isReady();
     verify(mockStream2).isReady();
 
     ClientStreamTracer bufferSizeTracer2 = bufferSizeTracer;
@@ -2786,7 +2781,6 @@ public void hedging_channelBufferLimitExceeded() {
     ArgumentCaptor<ClientStreamListener> sublistenerCaptor2 =
         ArgumentCaptor.forClass(ClientStreamListener.class);
     verify(mockStream2).start(sublistenerCaptor2.capture());
-    verify(mockStream1, times(2)).isReady();
     verify(mockStream2).isReady();
 
     ClientStreamTracer bufferSizeTracer2 = bufferSizeTracer;