binder: Deliver closed() to streams started after shutdown

BinderClientTransport.newStream() registered a new ClientInbound in
ongoingCalls right away but the Inbound only learned about its
ClientStreamListener later, in ClientStream.start(). If the transport
terminated in between (binderDied(), onUnbound(), a peer's
SHUTDOWN_TRANSPORT or shutdownNow(), all of which close ongoing calls
asynchronously), closeAbnormal() found an UNINITIALIZED Inbound with no
listener or StatsTraceContext to report to. start() then attached the
listener to an Inbound that was already closed and the RPC hung forever
with its tracers never closed.

Registering at newStream() was also a leak waiting to happen: gRPC core
can abandon a ClientStream it got from newStream() without ever calling
start() or cancel() on it (DelayedStream.setStream() silently drops the
real stream if the DelayedStream was cancelled in the meantime). Such a
stream stayed in ongoingCalls and counted as in-use forever, so a
graceful shutdown() of that transport could never terminate.

Make newStream() do allocations only and move registration into
start(), as OkHttp and Netty do. The new BinderClientTransport
.startStream() helper holds the transport lock across both registration
and Inbound.init() so, relative to shutdownInternal()'s snapshot of
ongoingCalls, there are only two possible outcomes: either shutdown ran
first, in which case start() reports closed(shutdownStatus, MISCARRIED)
without registering anything (mirroring OkHttp's streamReadyToStart()
and allowing a transparent retry on a fresh transport), or we
registered an Inbound that already has a listener for the eventual
close to be delivered to. The listener callback itself runs with no
locks held. The miscarried stream's Outbound is put into CLOSED so that
the writeMessage()/halfClose() ClientCalls issues right after start()
can't send anything for a call the server must never see.

Two behaviour changes follow, both consistent with OkHttp/Netty:
- transportInUse(true) is reported at start() rather than newStream().
- A graceful shutdown() between newStream() and start() no longer waits
  for the unstarted stream, whose start() then miscarries.

Since nothing can reach an Inbound that was never registered, the only
way a client Inbound can still be UNINITIALIZED when cancel() runs is
if cancel() came before start(). ClientStream#cancel says that's not
allowed, gRPC core's paths honour it (ClientCallImpl arms Context and
deadline cancellation only after start(), see issue #1343) and other
transports NPE if it happens, so cancel() now throws
IllegalStateException instead of silently closing a stream nobody is
listening to. A cancel() after a miscarried start() remains a legal
no-op. unregisterInbound() only decrements the in-use count for an
Inbound that was actually registered.

Author: jdcormie

binder/src/main/java/io/grpc/binder/internal/BinderClientTransport.java

@@ -49,6 +49,8 @@
 import io.grpc.binder.InboundParcelablePolicy;
 import io.grpc.binder.SecurityPolicy;
 import io.grpc.internal.ClientStream;
+import io.grpc.internal.ClientStreamListener;
+import io.grpc.internal.ClientStreamListener.RpcProgress;
 import io.grpc.internal.ClientTransportFactory.ClientTransportOptions;
 import io.grpc.internal.ConnectionClientTransport;
 import io.grpc.internal.FailingClientStream;
@@ -251,6 +253,7 @@ public synchronized ClientStream newStream(
           tracers);
     }
 
+    // Just allocations here. The new call is registered when it's start()ed, in startStream().
     int callId = latestCallId++;
     if (latestCallId == LAST_CALL_ID) {
       latestCallId = FIRST_CALL_ID;
@@ -260,15 +263,6 @@ public synchronized ClientStream newStream(
     Inbound.ClientInbound inbound =
         new Inbound.ClientInbound(
             this, attributes, callId, GrpcUtil.shouldBeCountedForInUse(callOptions));
-    if (ongoingCalls.putIfAbsent(callId, inbound) != null) {
-      Status failure = Status.INTERNAL.withDescription("Clashing call IDs");
-      shutdownInternal(failure, true);
-      return newFailingClientStream(failure, attributes, headers, tracers);
-    }
-
-    if (inbound.countsForInUse() && numInUseStreams.getAndIncrement() == 0) {
-      clientTransportListener.transportInUse(true);
-    }
     Outbound.ClientOutbound outbound =
         new Outbound.ClientOutbound(this, callId, method, headers, statsTraceContext);
     if (method.getType().clientSendsOneMessage()) {
@@ -278,12 +272,95 @@ public synchronized ClientStream newStream(
     }
   }
 
+  /**
+   * Registers a call allocated by {@link #newStream} and connects it to its listener.
+   *
+   * <p>If this transport has already shut down, the call isn't registered at all and 'listener' is
+   * instead told that the stream was closed with {@link RpcProgress#MISCARRIED}, which is accurate:
+   * nothing has been sent on its behalf yet, so the server never saw it and the RPC can be
+   * transparently retried on another transport.
+   *
+   * <p>Takes the transport lock, so must not be called with any Inbound or Outbound lock held.
+   *
+   * @return true iff the call was registered and {@code inbound} initialized, in which case the
+   *     caller may proceed to use the stream normally. False iff the stream is now closed and
+   *     'listener' has already been told so, in which case the caller must not call it again.
+   */
+  boolean startStream(
+      Inbound.ClientInbound inbound,
+      Outbound.ClientOutbound outbound,
+      ClientStreamListener listener) {
+    Status failure = registerCallAndInitInbound(inbound, outbound, listener);
+    if (failure != null) {
+      // Nobody else can reach a never-registered Inbound so we can finish up without our own lock,
+      // like every other close. MISCARRIED promises the server never saw this call, yet a closed
+      // stream may still be written to and half-closed (the listener's state has no bearing on
+      // what a caller may do; see ClientCall). So first make sure nothing can ever go on the wire
+      // for this call, then tell the listener.
+      synchronized (outbound) {
+        outbound.close();
+      }
+      synchronized (inbound) {
+        inbound.closeMiscarried(failure);
+      }
+      return false;
+    }
+    return true;
+  }
+
+  /**
+   * Registers 'inbound' as an ongoing call, unless we've shut down, and either way initializes it
+   * with its Outbound and listener.
+   *
+   * @return null iff 'inbound' is now registered, otherwise the reason it couldn't be
+   */
+  @Nullable
+  private synchronized Status registerCallAndInitInbound(
+      Inbound.ClientInbound inbound,
+      Outbound.ClientOutbound outbound,
+      ClientStreamListener listener) {
+    Status streamFailure = null; // Why 'inbound' can't start, if it can't.
+    Status transportFailure = null; // Why this whole transport must terminate, if it must.
+    boolean nowInUse = false;
+    // Holding our lock across both registration and init() means that shutdownInternal() either
+    // runs before us (and the stream miscarries) or after us (and finds an Inbound with a listener
+    // to report the close to). Holding the Inbound's lock too means nothing can touch it in
+    // between, once it's visible in ongoingCalls.
+    synchronized (inbound) {
+      if (isShutdown()) {
+        streamFailure = shutdownStatus;
+      } else {
+        Inbound<?, ?> existing = ongoingCalls.putIfAbsent(inbound.callId, inbound);
+        if (existing == inbound) {
+          throw new IllegalStateException("start() called more than once");
+        } else if (existing != null) {
+          transportFailure = Status.INTERNAL.withDescription("Clashing call IDs");
+          streamFailure = transportFailure;
+        } else {
+          nowInUse = inbound.countsForInUse() && numInUseStreams.getAndIncrement() == 0;
+        }
+      }
+      inbound.init(outbound, listener);
+    }
+    // Callbacks into gRPC (via clientTransportListener) happen under our lock, as they always
+    // have, but not under the Inbound's.
+    if (transportFailure != null) {
+      shutdownInternal(transportFailure, true);
+    } else if (nowInUse) {
+      clientTransportListener.transportInUse(true);
+    }
+    return streamFailure;
+  }
+
   @Override
-  protected void unregisterInbound(Inbound<?, ?> inbound) {
-    if (inbound.countsForInUse() && numInUseStreams.decrementAndGet() == 0) {
+  protected boolean unregisterInbound(Inbound<?, ?> inbound) {
+    boolean removed = super.unregisterInbound(inbound);
+    // Only ever registered Inbounds counted towards in-use, and forced termination may have already
+    // unregistered everything (see notifyTerminated() for the corresponding transportInUse(false)).
+    if (removed && inbound.countsForInUse() && numInUseStreams.decrementAndGet() == 0) {
       clientTransportListener.transportInUse(false);
     }
-    super.unregisterInbound(inbound);
+    return removed;
   }
 
   @Override

binder/src/main/java/io/grpc/binder/internal/BinderTransport.java

@@ -396,11 +396,21 @@ protected synchronized void sendPing(int id) throws StatusException {
     }
   }
 
-  protected void unregisterInbound(Inbound<?, ?> inbound) {
-    unregisterCall(inbound.callId);
+  /**
+   * Removes 'inbound' from the set of ongoing calls, if it's there.
+   *
+   * @return true iff 'inbound' was registered (and now isn't)
+   */
+  protected boolean unregisterInbound(Inbound<?, ?> inbound) {
+    return unregisterCall(inbound.callId);
   }
 
-  final void unregisterCall(int callId) {
+  /**
+   * Removes the call identified by 'callId' from the set of ongoing calls, if it's there.
+   *
+   * @return true iff 'callId' was registered (and now isn't)
+   */
+  final boolean unregisterCall(int callId) {
     boolean removed = (ongoingCalls.remove(callId) != null);
     if (removed && ongoingCalls.isEmpty()) {
       // If inState(SHUTDOWN), we'd need to terminate (as the last call to finish). But we can't
@@ -416,6 +426,7 @@ final void unregisterCall(int callId) {
             }
           });
     }
+    return removed;
   }
 
   final void sendTransaction(int callId, ParcelHolder parcel) throws StatusException {

binder/src/main/java/io/grpc/binder/internal/Inbound.java

@@ -154,12 +154,11 @@ private Inbound(T transport, Attributes attributes, int callId) {
 
   @GuardedBy("this")
   final void init(Outbound outbound, L listener) {
+    checkState(isUninitialized(), "already initialized");
     this.outbound = outbound;
     this.statsTraceContext = outbound.getStatsTraceContext();
     this.listener = listener;
-    if (!isClosed()) {
-      onDeliveryState(State.INITIALIZED);
-    }
+    onDeliveryState(State.INITIALIZED);
   }
 
   final void unregister() {
@@ -184,6 +183,12 @@ protected final boolean isClosed() {
     return deliveryState == State.CLOSED;
   }
 
+  /** Returns true iff our delivery state is still {@link State#UNINITIALIZED}. */
+  @GuardedBy("this")
+  final boolean isUninitialized() {
+    return deliveryState == State.UNINITIALIZED;
+  }
+
   @GuardedBy("this")
   private final boolean messageAvailable() {
     return firstMessage != null || nextCompleteMessageEnd > 0;
@@ -616,6 +621,34 @@ protected void deliverSuffix() {
     protected void deliverCloseAbnormal(Status status) {
       listener.closed(status, RpcProgress.PROCESSED, new Metadata());
     }
+
+    /**
+     * Handles {@link io.grpc.internal.ClientStream#cancel}, which "may only be called after
+     * start()".
+     *
+     * <p>Nothing can reach us before we're registered, which happens at start(), so still being
+     * {@link State#UNINITIALIZED} here can only mean that contract was broken.
+     */
+    @GuardedBy("this")
+    void cancel(Status status) {
+      checkState(!isUninitialized(), "cancel() before start()");
+      closeOnCancel(status);
+    }
+
+    /**
+     * Closes a stream that couldn't be registered because its transport had already shut down.
+     *
+     * <p>Nothing has been sent on the stream's behalf, so this is reported as {@link
+     * RpcProgress#MISCARRIED}. And since the stream was never registered, there's no out-of-band
+     * close to send and nothing to unregister().
+     */
+    @GuardedBy("this")
+    void closeMiscarried(Status status) {
+      checkState(!isClosed(), "already closed");
+      onDeliveryState(State.CLOSED);
+      statsTraceContext.streamClosed(status);
+      listener.closed(status, RpcProgress.MISCARRIED, new Metadata());
+    }
   }
 
   // ======================================

binder/src/main/java/io/grpc/binder/internal/MultiMessageClientStream.java

@@ -59,8 +59,8 @@ final class MultiMessageClientStream implements ClientStream {
 
   @Override
   public void start(ClientStreamListener listener) {
-    synchronized (inbound) {
-      inbound.init(outbound, listener);
+    if (!inbound.transport.startStream(inbound, outbound, listener)) {
+      return; // Already closed as MISCARRIED and 'listener' has been told.
     }
     if (outbound.isReady()) {
       listener.onReady();
@@ -120,7 +120,7 @@ public void halfClose() {
   @Override
   public void cancel(Status status) {
     synchronized (inbound) {
-      inbound.closeOnCancel(status);
+      inbound.cancel(status);
     }
   }
 

binder/src/main/java/io/grpc/binder/internal/Outbound.java

@@ -211,12 +211,24 @@ final void send() throws StatusException {
         sendInternal();
       } catch (StatusException se) {
         // Ensure we don't send anything else and rethrow.
-        onOutboundState(State.CLOSED);
+        close();
         throw se;
       }
     }
   }
 
+  /**
+   * Moves to {@link State#CLOSED} so that nothing (more) is ever sent.
+   *
+   * <p>Anything subsequently handed to this Outbound is accepted but never sent. Besides the send
+   * failure case, this is for a stream that closed before anything was sent on its behalf, where
+   * sending anything now would be for a call the server must never hear about.
+   */
+  @GuardedBy("this")
+  final void close() {
+    onOutboundState(State.CLOSED);
+  }
+
   @GuardedBy("this")
   @SuppressWarnings("fallthrough")
   protected final void sendInternal() throws StatusException {

binder/src/main/java/io/grpc/binder/internal/SingleMessageClientStream.java

@@ -63,8 +63,8 @@ final class SingleMessageClientStream implements ClientStream {
 
   @Override
   public void start(ClientStreamListener listener) {
-    synchronized (inbound) {
-      inbound.init(outbound, listener);
+    if (!inbound.transport.startStream(inbound, outbound, listener)) {
+      return; // Already closed as MISCARRIED and 'listener' has been told.
     }
     if (outbound.isReady()) {
       listener.onReady();
@@ -114,7 +114,7 @@ public void halfClose() {
   @Override
   public void cancel(Status status) {
     synchronized (inbound) {
-      inbound.closeOnCancel(status);
+      inbound.cancel(status);
     }
   }