diff --git a/core/src/main/java/io/grpc/internal/ServerCallImpl.java b/core/src/main/java/io/grpc/internal/ServerCallImpl.java index 07aa558e1b7..17cb6973442 100644 --- a/core/src/main/java/io/grpc/internal/ServerCallImpl.java +++ b/core/src/main/java/io/grpc/internal/ServerCallImpl.java @@ -46,6 +46,7 @@ import io.perfmark.Tag; import io.perfmark.TaskCloseable; import java.io.InputStream; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.logging.Level; import java.util.logging.Logger; @@ -66,6 +67,8 @@ final class ServerCallImpl extends ServerCall { private final DecompressorRegistry decompressorRegistry; private final CompressorRegistry compressorRegistry; private CallTracer serverCallTracer; + // The application's close() and the transport's closed() may race; only the first is counted. + private final AtomicBoolean callEndReported = new AtomicBoolean(); // state private volatile boolean cancelled; @@ -229,7 +232,13 @@ private void closeInternal(Status status, Metadata trailers) { stream.close(status, trailers); } finally { - serverCallTracer.reportCallEnded(status.isOk()); + reportCallEnded(status.isOk()); + } + } + + private void reportCallEnded(boolean success) { + if (callEndReported.compareAndSet(false, true)) { + serverCallTracer.reportCallEnded(success); } } @@ -287,7 +296,7 @@ private void handleInternalError(Throwable internalError) { : Status.INTERNAL.withCause(internalError) .withDescription("Internal error so cancelling stream."); stream.cancel(status); - serverCallTracer.reportCallEnded(false); // error so always false + reportCallEnded(false); // error so always false } /** @@ -415,6 +424,7 @@ private void closedInternal(Status status) { listener.onComplete(); } else { call.cancelled = true; + call.reportCallEnded(false); listener.onCancel(); // The status will not have a cause in all failure scenarios, but we want to make sure // we always cancel the context with one to keep the context cancelled state consistent. diff --git a/core/src/test/java/io/grpc/internal/ServerCallImplTest.java b/core/src/test/java/io/grpc/internal/ServerCallImplTest.java index 33c9a019a08..f054302bb61 100644 --- a/core/src/test/java/io/grpc/internal/ServerCallImplTest.java +++ b/core/src/test/java/io/grpc/internal/ServerCallImplTest.java @@ -127,6 +127,57 @@ public void callTracer_failure() { callTracer0(Status.UNKNOWN); } + @Test + public void callTracer_clientCancelled_reportsCallFailed() { + ServerStreamListenerImpl streamListener = + new ServerCallImpl.ServerStreamListenerImpl<>(call, callListener, context); + + streamListener.closed(Status.CANCELLED); + + ServerStats.Builder afterBuilder = new ServerStats.Builder(); + serverCallTracer.updateBuilder(afterBuilder); + ServerStats after = afterBuilder.build(); + assertEquals(1, after.callsStarted); + assertEquals(0, after.callsSucceeded); + assertEquals(1, after.callsFailed); + } + + @Test + public void callTracer_clientCancelledThenClosed_reportsCallFailedOnce() { + ServerStreamListenerImpl streamListener = + new ServerCallImpl.ServerStreamListenerImpl<>(call, callListener, context); + + streamListener.closed(Status.CANCELLED); + call.sendHeaders(new Metadata()); + call.sendMessage(123L); + call.close(Status.OK, new Metadata()); + + ServerStats.Builder afterBuilder = new ServerStats.Builder(); + serverCallTracer.updateBuilder(afterBuilder); + ServerStats after = afterBuilder.build(); + assertEquals(1, after.callsStarted); + assertEquals(0, after.callsSucceeded); + assertEquals(1, after.callsFailed); + } + + @Test + public void callTracer_closedThenClientCancelled_reportsCallSucceededOnce() { + ServerStreamListenerImpl streamListener = + new ServerCallImpl.ServerStreamListenerImpl<>(call, callListener, context); + + call.sendHeaders(new Metadata()); + call.sendMessage(123L); + call.close(Status.OK, new Metadata()); + streamListener.closed(Status.CANCELLED); + + ServerStats.Builder afterBuilder = new ServerStats.Builder(); + serverCallTracer.updateBuilder(afterBuilder); + ServerStats after = afterBuilder.build(); + assertEquals(1, after.callsStarted); + assertEquals(1, after.callsSucceeded); + assertEquals(0, after.callsFailed); + } + private void callTracer0(Status status) { CallTracer tracer = CallTracer.getDefaultFactory().create(); ServerStats.Builder beforeBuilder = new ServerStats.Builder();