check pending stream completion at delayed transport lifecycle #7720

YifeiZhuang · 2020-12-11T00:20:14Z

Problem

The issue was that delayedClientTransport reprocess() schedules work createRealStream() but does not track whether it is committed, i.e. the real stream done set to the delayed stream which completes the delegation. Because it does not track the completion, shutdown prematurely goes through and then executor was destroyed.
The inflight work that has been scheduled happen to use the same executor, causing RejectedExecutionException

 t1-Thread0:delayedTransport.reprocess()                            t3-Thread1:createRealStream()`RejectedExecutionException`        
                                          t2-Thread0:shutdownNow()

Reproduce

The problem can be reproduced by putting a pause at createRealStream(), and then run a client call.

grpc-java/core/src/main/java/io/grpc/internal/DelayedStream.java

Line 140 in ddaf1c8

private void drainPendingCalls() {

  private void drainPendingCalls() {
      Thread.sleep(3000);
 ...

core/src/main/java/io/grpc/internal/DelayedStream.java

core/src/main/java/io/grpc/internal/DelayedClientTransport.java

dapengzhang0 · 2020-12-14T18:21:36Z

core/src/main/java/io/grpc/internal/DelayedClientTransport.java

@@ -66,6 +66,23 @@
  @Nonnull
  @GuardedBy("lock")
  private Collection<PendingStream> pendingStreams = new LinkedHashSet<>();
+  @GuardedBy("lock")
+  private Collection<PendingStream> toCheckCompletionStreams = new LinkedHashSet<>();
+  private Runnable pollForStreamTransferCompletion = new Runnable() {


Make it final?

dapengzhang0 · 2020-12-14T18:46:12Z

core/src/main/java/io/grpc/internal/DelayedStream.java

@@ -132,6 +135,24 @@ final void setStream(ClientStream stream) {
    drainPendingCalls();
  }

+  protected boolean isStreamTransferCompleted() {
+    return realStreamStarted.getCount() == 0;


Note that getCount() is typically used for debugging and testing purposes. https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/CountDownLatch.html#getCount()

Better avoid it in the main source if possible.

dapengzhang0 · 2020-12-14T19:05:57Z

core/src/main/java/io/grpc/internal/DelayedClientTransport.java

        return;
      }
-      pendingStreams.removeAll(toRemove);
+      pendingStreams.removeAll(newlyCreated);
+      toCheckCompletionStreams.addAll(newlyCreated);


I would prefer making toCheckCompletionStreams be either 'non-final and immutable' or 'final and mutable', but not 'non-final' and 'mutable'.

dapengzhang0 · 2020-12-14T19:11:11Z

core/src/main/java/io/grpc/internal/DelayedClientTransport.java

+  @VisibleForTesting
+  final int getUncommittedStreamsCount() {
+    synchronized (lock) {
+      return toCheckCompletionStreams.size();


What about using the same name for toCheckCompletionStreams and uncommittedStreams?

dapengzhang0 · 2020-12-14T19:14:04Z

core/src/main/java/io/grpc/internal/DelayedClientTransport.java

+    public void run() {
+      ArrayList<PendingStream> savedToCheckCompletionStreams;
+      synchronized (lock) {
+        savedToCheckCompletionStreams = new ArrayList<>(toCheckCompletionStreams);


You don't need copy the list, just saving the reference should be sufficient.

dapengzhang0 · 2020-12-14T20:25:39Z

core/src/main/java/io/grpc/internal/DelayedClientTransport.java

    synchronized (lock) {
      lastPicker = picker;
      lastPickerVersion++;
-      if (picker == null || !hasPendingStreams()) {
+      if ((picker == null || !hasPendingStreams()) && !hasUncommittedStreams()) {


I think changing this line is not necessary. Regardless of hasUncommittedStreams, if return then stream.createRealStream() or drain() will not be called and will not cause trouble.

This line is because it needs to take care another case: when shutdown() is called but there are still uncommittedStreams, the shutdown path would return, then it will never shutdown. It relied on reprocess() to trigger the termination. (Similar to cancel() takes care of the last item and then trigger termination callback). This is different in shudownNow() which would take care of waiting uncommitted streams and then finalize termination.

hen shutdown() is called but there are still uncommittedStreams, the shutdown path would return, then it will never shutdown. It relied on reprocess() to trigger the termination.

uncommittedStreams can be considered as existing RPCs, so shutdown() should not terminate them. uncommittedStreams will complete transfer by themselves, and they don't rely on a second reprocess().

shutdown() never terminates then existing RPCs. Do you mean we await() during shutdown()?
It seems it's hard to avoid that. Say, if there are both pendingStreams and uncommittedStreams when shutdown is called, so shutdown has to return. Then during next reprocess(), the newly created stream has not been drained , which would cause pendingStreams empty but uncommittedstream still has items in it, and we need a way to drain it. It looks reprocess() is that place, it would trigger in the next call or after idle timer.

I think we should await() during shutdown(), because there is no "next" reprocess() after shutdown, so we can not rely on reprocess() for existing uncommittedStreams.

Another way is introducing an abstract method DelayedStream.onTransferComplete(), and implementing DelayedClientTransport.PendingStream.onTransferComplete() to managed decrement of `uncommittedStreams.

Thanks. Indeed it appears that idle timeout is permanently cancelled immediately after shutdown is called so there is no next reprocess(). It looks like just awaitTransferCompletion() during shutdown is not enough, moreover, I believe generally we are not supposed to await() during shutdown.

This reverts commit b1f553a.

This reverts commit eb18861.

This reverts commit 1fe3bce.

This reverts commit ac99972.

voidzcy

LGTM.

voidzcy · 2020-12-16T19:03:40Z

core/src/main/java/io/grpc/internal/TransferableClientStream.java

+   * Provides the place to define actions at the point when transfer is done.
+   * Call this method to trigger those transfer completion activities. No-op by default.
+   */
+  public void onTransferComplete() {


nit: public redundant.

core/src/main/java/io/grpc/internal/TransferableClientStream.java

core/src/main/java/io/grpc/internal/DelayedClientTransport.java

dapengzhang0 · 2020-12-17T07:49:49Z

core/src/main/java/io/grpc/internal/DelayedClientTransport.java

-            }
-          }
+        boolean justRemovedAnElement = pendingStreams.remove(this);
+        if (!hasPendingStreams() && justRemovedAnElement && reportTransportTerminated != null) {


No need to check reportTransportTerminated != null after the change. Previously if reportTransportTerminated == null, pendingStream is guaranteed reset to empty and justRemovedAnElement will never happen.

I was thinking that in time order: 1. super.cancel() 2.onTransferComplete(), reportTransportTerminated=null 3. this lock block might reportNotInUse, which won't happen previously.

I still think the check for reportTransportTerminated != null is unnecessary and the (!hasPendingStreams() && justRemovedAnElement) is the canonical invariant for reportTransportNotInUse. But ether way seems working anyway.

…le (#7720)" This reverts commit 9085012.

…le (#7720)" (#7744) This reverts commit 9085012. Alternative solution: alternative #7743

…#7720) add onTransferComplete() at delayedStream and wait for all pending streams to complete transfer when shutting down delayedClientTransport

…le (grpc#7720)" (grpc#7744) This reverts commit 9085012. Alternative solution: alternative grpc#7743

YifeiZhuang added 2 commits December 10, 2020 12:23

check pending stream completion at delayed transport lifecycle

ac99972

rename

1fe3bce

YifeiZhuang requested a review from dapengzhang0 December 11, 2020 00:20

YifeiZhuang changed the title ~~Add pollForTransferCompletion in delayedStream~~ check pending stream completion at delayed transport lifecycle Dec 11, 2020

YifeiZhuang requested a review from voidzcy December 11, 2020 00:23

voidzcy reviewed Dec 11, 2020

View reviewed changes

fix travis test

eb18861

dapengzhang0 reviewed Dec 14, 2020

View reviewed changes

YifeiZhuang added 6 commits December 14, 2020 13:06

Fix syntax: immutability and started flag

b1f553a

Revert "Fix syntax: immutability and started flag"

81ab463

This reverts commit b1f553a.

Revert "fix travis test"

193eb48

This reverts commit eb18861.

Revert "rename"

9304375

This reverts commit 1fe3bce.

Revert "check pending stream completion at delayed transport lifecycle"

56374dd

This reverts commit ac99972.

transfer complete abstract class

5024290

voidzcy approved these changes Dec 16, 2020

View reviewed changes

dapengzhang0 reviewed Dec 17, 2020

View reviewed changes

YifeiZhuang added 2 commits December 17, 2020 10:07

make delayedStream abstract class and fix thread safetiness

7022f5e

non-final ApplierDelayedStream inner class

cf2d51b

dapengzhang0 approved these changes Dec 21, 2020

View reviewed changes

YifeiZhuang merged commit 9085012 into grpc:master Dec 21, 2020

YifeiZhuang added a commit that referenced this pull request Dec 21, 2020

Revert " check pending stream completion at delayed transport lifecyc…

0f0c0f7

…le (#7720)" This reverts commit 9085012.

YifeiZhuang mentioned this pull request Dec 21, 2020

Revert " check pending stream completion at delayed transport lifecycle" #7744

Merged

ejona86 mentioned this pull request Dec 21, 2020

DelayedClientTransport can trigger RejectedExecutionException #6283

Closed

YifeiZhuang added a commit that referenced this pull request Dec 21, 2020

Revert " check pending stream completion at delayed transport lifecyc…

a67d816

…le (#7720)" (#7744) This reverts commit 9085012. Alternative solution: alternative #7743

dfawley pushed a commit to dfawley/grpc-java that referenced this pull request Jan 15, 2021

Revert " check pending stream completion at delayed transport lifecyc…

bab60ee

…le (grpc#7720)" (grpc#7744) This reverts commit 9085012. Alternative solution: alternative grpc#7743

YifeiZhuang deleted the delayed-stream branch February 16, 2021 05:41

github-actions bot locked as resolved and limited conversation to collaborators Jun 3, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

check pending stream completion at delayed transport lifecycle #7720

check pending stream completion at delayed transport lifecycle #7720

YifeiZhuang commented Dec 11, 2020

dapengzhang0 Dec 14, 2020

dapengzhang0 Dec 14, 2020

dapengzhang0 Dec 14, 2020

dapengzhang0 Dec 14, 2020

dapengzhang0 Dec 14, 2020

dapengzhang0 Dec 14, 2020

YifeiZhuang Dec 14, 2020

dapengzhang0 Dec 14, 2020

YifeiZhuang Dec 14, 2020

dapengzhang0 Dec 15, 2020

YifeiZhuang Dec 15, 2020

voidzcy left a comment

voidzcy Dec 16, 2020

dapengzhang0 Dec 17, 2020

YifeiZhuang Dec 17, 2020

dapengzhang0 Dec 21, 2020

check pending stream completion at delayed transport lifecycle #7720

check pending stream completion at delayed transport lifecycle #7720

Conversation

YifeiZhuang commented Dec 11, 2020

Problem

Reproduce

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

voidzcy left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment