diff --git a/binder/src/main/java/io/grpc/binder/internal/BinderClientTransport.java b/binder/src/main/java/io/grpc/binder/internal/BinderClientTransport.java index 58e7d7e2b31..e06a83aa09b 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderClientTransport.java +++ b/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,30 +263,91 @@ public synchronized ClientStream newStream( Inbound.ClientInbound inbound = new Inbound.ClientInbound( this, attributes, callId, GrpcUtil.shouldBeCountedForInUse(callOptions)); - if (ongoingCalls.putIfAbsent(callId, inbound) != null) { + Outbound.ClientOutbound outbound = + new Outbound.ClientOutbound(this, callId, method, headers, statsTraceContext); + if (method.getType().clientSendsOneMessage()) { + return new SingleMessageClientStream(this, inbound, outbound, attributes); + } else { + return new MultiMessageClientStream(this, inbound, outbound, attributes); + } + } + + /** + * Starts a stream allocated by {@link #newStream}: connects {@code inbound} to its Outbound and + * listener and registers it as an ongoing call. + * + *

If this transport has already shut down, the call isn't registered at all and {@code + * 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. + * + *

Takes the transport lock, so must not be called with any Inbound or Outbound lock held. + * + * @return true iff the call was registered, in which case the caller may proceed to use the + * stream normally. False iff the stream is now closed and {@code listener} has already been + * told so, in which case the caller must not call it again. + */ + @CheckReturnValue + boolean startStream( + Inbound.ClientInbound inbound, + Outbound.ClientOutbound outbound, + ClientStreamListener listener) { + // Initializing first means that, from the moment it's visible in ongoingCalls, the Inbound has + // a listener to report any close to. + synchronized (inbound) { + inbound.init(outbound, listener); + } + Status failure = registerCall(inbound); + 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 {@code inbound} as an ongoing call, unless we've shut down. + * + * @return null iff {@code inbound} is now registered, otherwise the reason it couldn't be + */ + @Nullable + private synchronized Status registerCall(Inbound.ClientInbound inbound) { + // Registering under our lock means that, relative to shutdownInternal()'s snapshot of + // ongoingCalls, either shutdown ran first (and the stream miscarries) or we're in the snapshot + // (and the stream gets closed like any other). + if (isShutdown()) { + return shutdownStatus; + } + if (ongoingCalls.putIfAbsent(inbound.callId, inbound) != null) { Status failure = Status.INTERNAL.withDescription("Clashing call IDs"); shutdownInternal(failure, true); - return newFailingClientStream(failure, attributes, headers, tracers); + return failure; } - if (inbound.countsForInUse() && numInUseStreams.getAndIncrement() == 0) { clientTransportListener.transportInUse(true); } - Outbound.ClientOutbound outbound = - new Outbound.ClientOutbound(this, callId, method, headers, statsTraceContext); - if (method.getType().clientSendsOneMessage()) { - return new SingleMessageClientStream(inbound, outbound, attributes); - } else { - return new MultiMessageClientStream(inbound, outbound, attributes); - } + return null; } @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 diff --git a/binder/src/main/java/io/grpc/binder/internal/BinderTransport.java b/binder/src/main/java/io/grpc/binder/internal/BinderTransport.java index db06d3052ab..c5ccbff9f06 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderTransport.java +++ b/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 {@code inbound} from the set of ongoing calls, if it's there. + * + * @return true iff {@code 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 {@code callId} from the set of ongoing calls, if it's there. + * + * @return true iff {@code 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 { diff --git a/binder/src/main/java/io/grpc/binder/internal/Inbound.java b/binder/src/main/java/io/grpc/binder/internal/Inbound.java index bef25fb4de2..cdd61be6118 100644 --- a/binder/src/main/java/io/grpc/binder/internal/Inbound.java +++ b/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(), "init() in state %s", deliveryState); 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; @@ -285,7 +290,7 @@ final void closeAbnormal(Status status) { private final void closeAbnormal( Status outboundStatus, Status internalStatus, boolean isOobFromRemote) { if (!isClosed()) { - boolean wasInitialized = (deliveryState != State.UNINITIALIZED); + boolean wasInitialized = !isUninitialized(); onDeliveryState(State.CLOSED); if (wasInitialized) { statsTraceContext.streamClosed(internalStatus); @@ -308,8 +313,13 @@ final void onTransportReady() { Outbound outbound = null; StreamListener listener = null; synchronized (this) { + // closed() is always the listener's final callback. + if (!isClosed()) { + listener = this.listener; + } + // But poke the Outbound even if we're closed since a server stream's close could be sitting + // in it, blocked on flow control. outbound = this.outbound; - listener = this.listener; } if (listener != null) { listener.onReady(); @@ -616,6 +626,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()". + * + *

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. + * + *

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()); + } } // ====================================== diff --git a/binder/src/main/java/io/grpc/binder/internal/MultiMessageClientStream.java b/binder/src/main/java/io/grpc/binder/internal/MultiMessageClientStream.java index 9873adcb447..5b2a14753fe 100644 --- a/binder/src/main/java/io/grpc/binder/internal/MultiMessageClientStream.java +++ b/binder/src/main/java/io/grpc/binder/internal/MultiMessageClientStream.java @@ -46,12 +46,17 @@ */ final class MultiMessageClientStream implements ClientStream { + private final BinderClientTransport transport; private final Inbound.ClientInbound inbound; private final Outbound.ClientOutbound outbound; private final Attributes attributes; MultiMessageClientStream( - Inbound.ClientInbound inbound, Outbound.ClientOutbound outbound, Attributes attributes) { + BinderClientTransport transport, + Inbound.ClientInbound inbound, + Outbound.ClientOutbound outbound, + Attributes attributes) { + this.transport = transport; this.inbound = inbound; this.outbound = outbound; this.attributes = attributes; @@ -59,8 +64,8 @@ final class MultiMessageClientStream implements ClientStream { @Override public void start(ClientStreamListener listener) { - synchronized (inbound) { - inbound.init(outbound, listener); + if (!transport.startStream(inbound, outbound, listener)) { + return; // Already closed as MISCARRIED and 'listener' has been told. } if (outbound.isReady()) { listener.onReady(); @@ -120,7 +125,7 @@ public void halfClose() { @Override public void cancel(Status status) { synchronized (inbound) { - inbound.closeOnCancel(status); + inbound.cancel(status); } } diff --git a/binder/src/main/java/io/grpc/binder/internal/Outbound.java b/binder/src/main/java/io/grpc/binder/internal/Outbound.java index 7db5bf0fbe4..0de448bb986 100644 --- a/binder/src/main/java/io/grpc/binder/internal/Outbound.java +++ b/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. + * + *

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 { diff --git a/binder/src/main/java/io/grpc/binder/internal/SingleMessageClientStream.java b/binder/src/main/java/io/grpc/binder/internal/SingleMessageClientStream.java index 8a899d621a2..e6cf08ad518 100644 --- a/binder/src/main/java/io/grpc/binder/internal/SingleMessageClientStream.java +++ b/binder/src/main/java/io/grpc/binder/internal/SingleMessageClientStream.java @@ -47,6 +47,7 @@ */ final class SingleMessageClientStream implements ClientStream { + private final BinderClientTransport transport; private final Inbound.ClientInbound inbound; private final Outbound.ClientOutbound outbound; private final Attributes attributes; @@ -55,7 +56,11 @@ final class SingleMessageClientStream implements ClientStream { @Nullable private Deadline pendingDeadline; SingleMessageClientStream( - Inbound.ClientInbound inbound, Outbound.ClientOutbound outbound, Attributes attributes) { + BinderClientTransport transport, + Inbound.ClientInbound inbound, + Outbound.ClientOutbound outbound, + Attributes attributes) { + this.transport = transport; this.inbound = inbound; this.outbound = outbound; this.attributes = attributes; @@ -63,8 +68,8 @@ final class SingleMessageClientStream implements ClientStream { @Override public void start(ClientStreamListener listener) { - synchronized (inbound) { - inbound.init(outbound, listener); + if (!transport.startStream(inbound, outbound, listener)) { + return; // Already closed as MISCARRIED and 'listener' has been told. } if (outbound.isReady()) { listener.onReady(); @@ -114,7 +119,7 @@ public void halfClose() { @Override public void cancel(Status status) { synchronized (inbound) { - inbound.closeOnCancel(status); + inbound.cancel(status); } } diff --git a/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java b/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java index db9a9202702..8c5117f4668 100644 --- a/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java +++ b/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java @@ -19,10 +19,19 @@ import static com.google.common.truth.Truth.assertThat; import static io.grpc.StatusSubject.assertThat; import static io.grpc.binder.internal.TransactionBuilder.newOutOfBandCloseTxnBuilder; +import static io.grpc.binder.internal.TransactionBuilder.newShutdownTransportTxnBuilder; import static io.grpc.binder.internal.TransactionBuilder.newStreamTxnToClientBuilder; import static io.grpc.binder.internal.TransactionBuilder.utf8; import static io.grpc.binder.internal.TransactionUtils.FLAG_MESSAGE_DATA; +import static org.junit.Assert.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import static org.robolectric.Shadows.shadowOf; import android.content.ComponentName; @@ -38,14 +47,17 @@ import io.grpc.binder.AndroidComponentAddress; import io.grpc.internal.ClientStream; import io.grpc.internal.ClientStreamListener; +import io.grpc.internal.ClientStreamListener.RpcProgress; import io.grpc.internal.FixedObjectPool; import io.grpc.internal.GrpcUtil; import io.grpc.internal.ManagedClientTransport; import io.grpc.internal.StreamListener.MessageProducer; +import io.grpc.internal.testing.TestClientStreamTracer; import java.net.SocketAddress; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; +import org.mockito.InOrder; import org.robolectric.RobolectricTestRunner; /** @@ -398,6 +410,20 @@ public void transactionsIgnoredAfterClosed() throws Exception { assertThat(listener.getReadMessages()).isEmpty(); } + @Test + public void transportReadyAfterClosed_doesNotReportOnReady() throws Exception { + newStreamTxnToClientBuilder(nextTxIndex++).withPrefix(new Metadata()).dispatchTo(inbound); + newOutOfBandCloseTxnBuilder(Status.CANCELLED).dispatchTo(inbound); + drainExecutors(); + assertThat(listener.isClosed()).isTrue(); + + // Simulates BinderTransport.handleAcknowledgedBytes() picking this Inbound out of ongoingCalls + // just before a concurrent close (e.g. cancel()) unregistered it. FakeClientStreamListener + // throws if onReady() is invoked after closed(). + inbound.onTransportReady(); + drainExecutors(); + } + @Test public void allInOneUnaryTransaction() throws Exception { newStreamTxnToClientBuilder(0) @@ -489,25 +515,314 @@ public void clientCancelUnregisters() throws Exception { } @Test - public void abnormalCloseBeforeStreamStartDoesNotThrow() throws Exception { - transport.newStream( + public void newStream_doesNotRegisterOrCountAsInUseUntilStart() throws Exception { + ManagedClientTransport.Listener transportListener = mock(ManagedClientTransport.Listener.class); + BinderClientTransport transport = + new DummyClientTransportBuilder().setTransportListener(transportListener).build(); + ClientStream unstartedStream = newUnstartedStream(transport); + drainExecutors(); + + assertThat(transport.getOngoingCalls()).doesNotContainKey(BinderTransport.FIRST_CALL_ID); + verify(transportListener, never()).transportInUse(anyBoolean()); + + unstartedStream.start(new FakeClientStreamListener()); + drainExecutors(); + + assertThat(transport.getOngoingCalls()).containsKey(BinderTransport.FIRST_CALL_ID); + verify(transportListener).transportInUse(true); + } + + @Test + public void shutdownNowBeforeStart_closesListenerAndTracersUponStart() throws Exception { + TestClientStreamTracer tracer = new TestClientStreamTracer(); + ClientStream unstartedStream = newUnstartedStream(new ClientStreamTracer[] {tracer}); + + transport.shutdownNow(Status.UNAVAILABLE.withDescription("test shutdownNow")); + drainExecutors(); + assertThat(transport.getOngoingCalls()).isEmpty(); + assertThat(tracer.getStatus()).isNull(); + + FakeClientStreamListener unstartedListener = new FakeClientStreamListener(); + unstartedStream.start(unstartedListener); + drainExecutors(); + + assertThat(unstartedListener.isClosed()).isTrue(); + assertThat(unstartedListener.getClosedStatus()).hasCode(Status.Code.UNAVAILABLE); + assertThat(unstartedListener.getClosedStatus().getDescription()).isEqualTo("test shutdownNow"); + assertThat(unstartedListener.getClosedProgress()).isEqualTo(RpcProgress.MISCARRIED); + assertThat(unstartedListener.getClosedTrailers().keys()).isEmpty(); + assertThat(tracer.getStatus()).hasCode(Status.Code.UNAVAILABLE); + assertThat(tracer.getStatus().getDescription()).isEqualTo("test shutdownNow"); + assertThat(tracer.getOutboundHeaders()).isFalse(); + assertThat(transport.getOngoingCalls()).isEmpty(); + } + + @Test + public void binderDiedBeforeStart_closesListenerUponStart() throws Exception { + ClientStream unstartedStream = newUnstartedStream(new ClientStreamTracer[0]); + + transport.binderDied(); + drainExecutors(); + assertThat(transport.getOngoingCalls()).isEmpty(); + + FakeClientStreamListener unstartedListener = new FakeClientStreamListener(); + unstartedStream.start(unstartedListener); + drainExecutors(); + + assertThat(unstartedListener.isClosed()).isTrue(); + assertThat(unstartedListener.getClosedStatus()).hasCode(Status.Code.UNAVAILABLE); + assertThat(unstartedListener.getClosedStatus().getDescription()).contains("binderDied"); + assertThat(unstartedListener.getClosedProgress()).isEqualTo(RpcProgress.MISCARRIED); + } + + @Test + public void peerShutdownBeforeStart_closesListenerUponStart() throws Exception { + ClientStream unstartedStream = newUnstartedStream(new ClientStreamTracer[0]); + + newShutdownTransportTxnBuilder().dispatchTo(transport); + drainExecutors(); + assertThat(transport.getOngoingCalls()).isEmpty(); + + FakeClientStreamListener unstartedListener = new FakeClientStreamListener(); + unstartedStream.start(unstartedListener); + drainExecutors(); + + assertThat(unstartedListener.isClosed()).isTrue(); + assertThat(unstartedListener.getClosedStatus()).hasCode(Status.Code.UNAVAILABLE); + assertThat(unstartedListener.getClosedStatus().getDescription()) + .isEqualTo("transport shutdown by peer"); + assertThat(unstartedListener.getClosedProgress()).isEqualTo(RpcProgress.MISCARRIED); + } + + @Test + public void shutdownBeforeStart_closesListenerUponStart() throws Exception { + ClientStream unstartedStream = newUnstartedStream(new ClientStreamTracer[0]); + + // The stream from setUp() is still ongoing so this graceful shutdown won't terminate anything. + transport.shutdown(Status.UNAVAILABLE.withDescription("test shutdown")); + drainExecutors(); + assertThat(transport.getOngoingCalls()).containsKey(inbound.callId); + + FakeClientStreamListener unstartedListener = new FakeClientStreamListener(); + unstartedStream.start(unstartedListener); + drainExecutors(); + + assertThat(unstartedListener.isClosed()).isTrue(); + assertThat(unstartedListener.getClosedStatus()).hasCode(Status.Code.UNAVAILABLE); + assertThat(unstartedListener.getClosedStatus().getDescription()).isEqualTo("test shutdown"); + assertThat(unstartedListener.getClosedProgress()).isEqualTo(RpcProgress.MISCARRIED); + assertThat(transport.getOngoingCalls()).doesNotContainKey(inbound.callId + 1); + assertThat(transport.getOngoingCalls()).containsKey(inbound.callId); + } + + @Test + public void shutdownBeforeStart_doesNotWaitForUnstartedStream() throws Exception { + ManagedClientTransport.Listener transportListener = mock(ManagedClientTransport.Listener.class); + BinderClientTransport transport = + new DummyClientTransportBuilder().setTransportListener(transportListener).build(); + ClientStream unstartedStream = newUnstartedStream(transport); + + transport.shutdown(Status.UNAVAILABLE.withDescription("test shutdown")); + drainExecutors(); + + // Nothing was in flight as far as the transport is concerned, so it's already gone. + verify(transportListener).transportTerminated(); + verify(transportListener, never()).transportInUse(anyBoolean()); + + FakeClientStreamListener unstartedListener = new FakeClientStreamListener(); + unstartedStream.start(unstartedListener); + drainExecutors(); + + assertThat(unstartedListener.isClosed()).isTrue(); + assertThat(unstartedListener.getClosedStatus()).hasCode(Status.Code.UNAVAILABLE); + assertThat(unstartedListener.getClosedStatus().getDescription()).isEqualTo("test shutdown"); + assertThat(unstartedListener.getClosedProgress()).isEqualTo(RpcProgress.MISCARRIED); + verify(transportListener, never()).transportInUse(anyBoolean()); + } + + @Test + public void shutdownNowWithStartedAndUnstartedStreams_reportsNotInUseOnceBeforeTerminated() + throws Exception { + ManagedClientTransport.Listener transportListener = mock(ManagedClientTransport.Listener.class); + BinderClientTransport transport = + new DummyClientTransportBuilder().setTransportListener(transportListener).build(); + ClientStream startedStream = newUnstartedStream(transport); + FakeClientStreamListener startedListener = new FakeClientStreamListener(); + startedStream.start(startedListener); + ClientStream unstartedStream = newUnstartedStream(transport); + drainExecutors(); + verify(transportListener).transportInUse(true); + + transport.shutdownNow(Status.UNAVAILABLE.withDescription("test shutdownNow")); + drainExecutors(); + FakeClientStreamListener unstartedListener = new FakeClientStreamListener(); + unstartedStream.start(unstartedListener); + drainExecutors(); + + assertThat(startedListener.isClosed()).isTrue(); + assertThat(startedListener.getClosedProgress()).isEqualTo(RpcProgress.PROCESSED); + assertThat(unstartedListener.isClosed()).isTrue(); + assertThat(unstartedListener.getClosedProgress()).isEqualTo(RpcProgress.MISCARRIED); + // Only the started stream ever counted as in use, and the forced termination (not the + // never-registered stream's start()) is what un-counts it, exactly once, before terminating. + InOrder inOrder = inOrder(transportListener); + inOrder.verify(transportListener).transportInUse(true); + inOrder.verify(transportListener).transportInUse(false); + inOrder.verify(transportListener).transportTerminated(); + verify(transportListener, times(1)).transportInUse(false); + } + + @Test + public void shutdownWithStartedAndUnstartedStreams_terminatesWhenStartedStreamCompletes() + throws Exception { + ManagedClientTransport.Listener transportListener = mock(ManagedClientTransport.Listener.class); + BinderClientTransport transport = + new DummyClientTransportBuilder().setTransportListener(transportListener).build(); + ClientStream startedStream = newUnstartedStream(transport); + FakeClientStreamListener startedListener = new FakeClientStreamListener(); + startedStream.start(startedListener); + ClientStream unstartedStream = newUnstartedStream(transport); + drainExecutors(); + + transport.shutdown(Status.UNAVAILABLE.withDescription("test shutdown")); + drainExecutors(); + verify(transportListener).transportShutdown(any(), any()); + verify(transportListener, never()).transportTerminated(); + + // The unstarted stream miscarries but the started one keeps the transport alive. + FakeClientStreamListener unstartedListener = new FakeClientStreamListener(); + unstartedStream.start(unstartedListener); + drainExecutors(); + assertThat(unstartedListener.getClosedProgress()).isEqualTo(RpcProgress.MISCARRIED); + verify(transportListener, never()).transportTerminated(); + + // Until it completes. + Inbound.ClientInbound startedInbound = + (Inbound.ClientInbound) transport.getOngoingCalls().get(BinderTransport.FIRST_CALL_ID); + newStreamTxnToClientBuilder(0) + .withPrefix(new Metadata()) + .withSuffix(Status.OK, new Metadata()) + .dispatchTo(startedInbound); + drainExecutors(); + + assertThat(startedListener.isClosed()).isTrue(); + assertThat(startedListener.getClosedStatus()).isOk(); + InOrder inOrder = inOrder(transportListener); + inOrder.verify(transportListener).transportInUse(true); + inOrder.verify(transportListener).transportInUse(false); + inOrder.verify(transportListener).transportTerminated(); + verify(transportListener, times(1)).transportInUse(false); + } + + @Test + public void unaryStreamStartedAfterShutdownNow_closesListenerOnceAndSendsNothing() + throws Exception { + OneWayBinderProxy outgoingBinder = mock(OneWayBinderProxy.class); + BinderClientTransport transport = + new DummyClientTransportBuilder().setOutgoingBinder(outgoingBinder).build(); + MethodDescriptor unaryMethod = + newDummyMethodDescriptorBuilder().setType(MethodDescriptor.MethodType.UNARY).build(); + ClientStream unaryStream = + transport.newStream( + unaryMethod, new Metadata(), CallOptions.DEFAULT, new ClientStreamTracer[0]); + int callId = BinderTransport.FIRST_CALL_ID; + assertThat(transport.getOngoingCalls()).doesNotContainKey(callId); + verify(outgoingBinder, never()).transact(eq(callId), any()); // Nothing is sent until start(). + + transport.shutdownNow(Status.UNAVAILABLE.withDescription("test shutdownNow")); + drainExecutors(); + // The transport never heard of this call so shutdownNow() has nothing to close on its behalf. + verify(outgoingBinder, never()).transact(eq(callId), any()); + + // The ClientStream API allows all of this after a close (and it's how a unary RPC typically + // drives a stream, without waiting to see whether start() closed it). + FakeClientStreamListener unaryListener = new FakeClientStreamListener(); + unaryStream.start(unaryListener); + unaryStream.request(2); + unaryStream.writeMessage(unaryMethod.getRequestMarshaller().stream("request")); + unaryStream.halfClose(); + drainExecutors(); + + // FakeClientStreamListener throws if closed() is called twice or onReady() follows closed(). + assertThat(unaryListener.isClosed()).isTrue(); + assertThat(unaryListener.getClosedStatus()).hasCode(Status.Code.UNAVAILABLE); + assertThat(unaryListener.getClosedProgress()).isEqualTo(RpcProgress.MISCARRIED); + // None of start()/writeMessage()/halfClose() put anything on the wire for a call the server + // must never see. + verify(outgoingBinder, never()).transact(eq(callId), any()); + } + + @Test + public void cancelBeforeStart_throwsIllegalStateException() throws Exception { + // ClientStream#cancel() "may only be called after start()". gRPC core's normal paths honour + // that (ClientCallImpl arms Context/deadline cancellation only after start(); DelayedStream and + // RetriableStream start() a stream before they'd ever cancel() it) and other transports NPE if + // it happens. Fail loudly rather than inventing a close for a stream nobody is listening to. + OneWayBinderProxy outgoingBinder = mock(OneWayBinderProxy.class); + ManagedClientTransport.Listener transportListener = mock(ManagedClientTransport.Listener.class); + BinderClientTransport transport = + new DummyClientTransportBuilder() + .setOutgoingBinder(outgoingBinder) + .setTransportListener(transportListener) + .build(); + ClientStream unstartedStream = newUnstartedStream(transport); + int callId = BinderTransport.FIRST_CALL_ID; + + assertThrows(IllegalStateException.class, () -> unstartedStream.cancel(Status.CANCELLED)); + drainExecutors(); + + // The rejected cancel() left no trace. + assertThat(transport.getOngoingCalls()).doesNotContainKey(callId); + verify(outgoingBinder, never()).transact(eq(callId), any()); + verify(transportListener, never()).transportInUse(anyBoolean()); + } + + @Test + public void cancelAfterMiscarriedStart_isNoOp() throws Exception { + // Unlike cancel() *before* start(), cancel() after a start() that miscarried is perfectly + // legal: ClientCallImpl arms its deadline/Context cancellation only after start() returns. + OneWayBinderProxy outgoingBinder = mock(OneWayBinderProxy.class); + BinderClientTransport transport = + new DummyClientTransportBuilder().setOutgoingBinder(outgoingBinder).build(); + ClientStream unstartedStream = newUnstartedStream(transport); + int callId = BinderTransport.FIRST_CALL_ID; + + transport.shutdownNow(Status.UNAVAILABLE.withDescription("test shutdownNow")); + drainExecutors(); + FakeClientStreamListener unstartedListener = new FakeClientStreamListener(); + unstartedStream.start(unstartedListener); + drainExecutors(); + assertThat(unstartedListener.isClosed()).isTrue(); + assertThat(unstartedListener.getClosedProgress()).isEqualTo(RpcProgress.MISCARRIED); + + unstartedStream.cancel(Status.CANCELLED.withDescription("cancelled after miscarry")); + drainExecutors(); + + // FakeClientStreamListener throws if closed() is called twice. + assertThat(unstartedListener.getClosedStatus()).hasCode(Status.Code.UNAVAILABLE); + assertThat(transport.getOngoingCalls()).doesNotContainKey(callId); + // Nothing was ever sent for this call: no prefix, and no out-of-band close for the cancel. + verify(outgoingBinder, never()).transact(eq(callId), any()); + } + + private ClientStream newUnstartedStream(ClientStreamTracer[] tracers) { + return transport.newStream( + newDummyMethodDescriptorBuilder().build(), new Metadata(), CallOptions.DEFAULT, tracers); + } + + private static ClientStream newUnstartedStream(BinderClientTransport transport) { + return transport.newStream( newDummyMethodDescriptorBuilder().build(), new Metadata(), CallOptions.DEFAULT, new ClientStreamTracer[0]); - Inbound.ClientInbound unstartedInbound = - (Inbound.ClientInbound) transport.getOngoingCalls().get(inbound.callId + 1); - assertThat(transport.getOngoingCalls()).containsKey(unstartedInbound.callId); - - newOutOfBandCloseTxnBuilder(Status.CANCELLED.withDescription("remote abort")) - .dispatchTo(unstartedInbound); - - drainExecutors(); - assertThat(transport.getOngoingCalls()).doesNotContainKey(unstartedInbound.callId); } private static final class DummyClientTransportBuilder { private final BinderClientTransportFactory.Builder factoryBuilder; + private OneWayBinderProxy outgoingBinder = mock(OneWayBinderProxy.class); // Blackhole. + private ManagedClientTransport.Listener transportListener = + mock(ManagedClientTransport.Listener.class); DummyClientTransportBuilder() { MainThreadScheduledExecutorService mainThreadExecutor = @@ -519,6 +834,17 @@ private static final class DummyClientTransportBuilder { .setScheduledExecutorPool(new FixedObjectPool<>(mainThreadExecutor)); } + public DummyClientTransportBuilder setOutgoingBinder(OneWayBinderProxy outgoingBinder) { + this.outgoingBinder = outgoingBinder; + return this; + } + + public DummyClientTransportBuilder setTransportListener( + ManagedClientTransport.Listener transportListener) { + this.transportListener = transportListener; + return this; + } + public BinderClientTransport build() { SocketAddress serverAddress = AndroidComponentAddress.forComponent(new ComponentName("fake.pkg", "fake.cls")); @@ -529,10 +855,9 @@ public BinderClientTransport build() { .build(); // This hack lets us create a transport without the need for a real server for handshaking. synchronized (transport) { - // Blackhole Outbound. - transport.setOutgoingBinder(mock(OneWayBinderProxy.class)); + transport.setOutgoingBinder(outgoingBinder); } - Runnable unused = transport.start(mock(ManagedClientTransport.Listener.class)); + Runnable unused = transport.start(transportListener); synchronized (transport) { transport.setState(BinderTransport.TransportState.READY); } diff --git a/binder/src/test/java/io/grpc/binder/internal/RobolectricBinderTransportTest.java b/binder/src/test/java/io/grpc/binder/internal/RobolectricBinderTransportTest.java index 63c47bf4f19..a13e14f7a66 100644 --- a/binder/src/test/java/io/grpc/binder/internal/RobolectricBinderTransportTest.java +++ b/binder/src/test/java/io/grpc/binder/internal/RobolectricBinderTransportTest.java @@ -29,6 +29,7 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; import static org.junit.Assume.assumeTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -62,6 +63,7 @@ import io.grpc.binder.internal.SettableAsyncSecurityPolicy.AuthRequest; import io.grpc.internal.AbstractTransportTest; import io.grpc.internal.ClientStream; +import io.grpc.internal.ClientStreamListener.RpcProgress; import io.grpc.internal.ClientStreamListenerBase; import io.grpc.internal.ClientTransport; import io.grpc.internal.ClientTransportFactory.ClientTransportOptions; @@ -78,6 +80,7 @@ import java.util.concurrent.BlockingQueue; import java.util.concurrent.Executor; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicReference; import org.junit.Before; import org.junit.Ignore; import org.junit.Rule; @@ -420,6 +423,41 @@ public void clientReportsAuthzErrorToServer() throws Exception { assertThat(serverTransportListener.isTerminated()).isTrue(); } + @Test + public void shutdownNowBetweenNewStreamAndStart_closesStreamAsMiscarried() throws Exception { + server.start(serverListener); + client = newClientTransport(server); + startTransport(client, mockClientTransportListener); + + ClientStream stream = + client.newStream(methodDescriptor, new Metadata(), CallOptions.DEFAULT, tracers); + + Status shutdownReason = Status.UNAVAILABLE.withDescription("shutdownNow() before start()"); + client.shutdownNow(shutdownReason); + verify(mockClientTransportListener, timeout(TIMEOUT_MS)).transportTerminated(); + // An unstarted stream doesn't count as "in use". + verify(mockClientTransportListener, never()).transportInUse(anyBoolean()); + + AtomicReference closedProgress = new AtomicReference<>(); + ClientStreamListenerBase clientStreamListener = + new ClientStreamListenerBase() { + @Override + public void closed(Status status, RpcProgress rpcProgress, Metadata trailers) { + closedProgress.set(rpcProgress); + super.closed(status, rpcProgress, trailers); + } + }; + stream.start(clientStreamListener); + + assertThat(clientStreamListener.awaitClose(TIMEOUT_MS, MILLISECONDS)) + .isSameInstanceAs(shutdownReason); + assertThat(clientStreamListener.awaitTrailers(TIMEOUT_MS, MILLISECONDS)).isNotNull(); + assertThat(closedProgress.get()).isEqualTo(RpcProgress.MISCARRIED); + assertThat(clientStreamTracer1.await(TIMEOUT_MS, MILLISECONDS)).isTrue(); + assertThat(clientStreamTracer1.getStatus()).isSameInstanceAs(shutdownReason); + assertThat(clientStreamTracer1.getOutboundHeaders()).isFalse(); + } + @Test @Override // We don't quite pass the official/abstract version of this test yet because diff --git a/binder/src/testFixtures/java/io/grpc/binder/internal/TransactionBuilder.java b/binder/src/testFixtures/java/io/grpc/binder/internal/TransactionBuilder.java index 2dbe3373f63..f6c7f81ab72 100644 --- a/binder/src/testFixtures/java/io/grpc/binder/internal/TransactionBuilder.java +++ b/binder/src/testFixtures/java/io/grpc/binder/internal/TransactionBuilder.java @@ -54,6 +54,11 @@ public static OutOfBandCloseTxnBuilder newOutOfBandCloseTxnBuilder(Status status return new OutOfBandCloseTxnBuilder(status); } + /** Creates a builder for a transport-level SHUTDOWN_TRANSPORT control transaction. */ + public static ShutdownTransportTxnBuilder newShutdownTransportTxnBuilder() { + return new ShutdownTransportTxnBuilder(); + } + /** Functional interface for parcel consumers that may throw checked exceptions. */ @FunctionalInterface public interface ParcelConsumer { @@ -264,6 +269,33 @@ public void dispatchTo(ParcelConsumer consumer) throws Exception { } } + /** + * Builder for the SHUTDOWN_TRANSPORT control transaction a peer sends when it terminates. + * + *

Unlike stream transactions, this one is addressed to the transport as a whole rather than to + * a particular call, so use {@link #dispatchTo(BinderTransport)}. + */ + public static final class ShutdownTransportTxnBuilder extends TransactionBuilder { + private ShutdownTransportTxnBuilder() {} + + /** Dispatches the synthesized parcel to a transport, as if it came from the peer. */ + public void dispatchTo(BinderTransport transport) throws Exception { + dispatchTo(parcel -> transport.handleTransaction(BinderTransport.SHUTDOWN_TRANSPORT, parcel)); + } + + @Override + public void dispatchTo(ParcelConsumer consumer) throws Exception { + Parcel parcel = Parcel.obtain(); + try { + parcel.writeInt(0); // Flags. Always empty but present, mirroring what a real peer sends. + parcel.setDataPosition(0); + consumer.accept(parcel); + } finally { + parcel.recycle(); + } + } + } + /** * Encodes the given string to bytes using UTF-8. *