#20551

Committed 03 Nov 2026 03:24AM UTC coverage: 89.449% (+0.007%) from 89.442%

Build # #20551

Build Type

push

github

Committed by

web-flow

Commit Message

xds: Fix intermittent failure in RingHashLoadBalancerTest.allSubchannelsInTransientFailure (#13130)

Deliver TRANSIENT_FAILURE in deterministic server order using
subchannelList instead of iterating over subchannels.values() (HashMap),
and expect 2 connections.

Why this became flaky with the new PickFirst LB:
Old PickFirst unconditionally invoked subchannel.requestConnection() on
each call without checking connection state, so triggering proactive
connections twice always produced 2 connection requests regardless of
iteration order (either [A, B] or [A, A]).

New PickFirst deduplicates connection requests if a subchannel is
already in CONNECTING state. When iterating over a HashMap, if the same
subchannel was triggered twice, New PickFirst deduplicated it to 1
connection, leading to the assumption in #13094 that New PickFirst would
always yield 1 connection. However, if two distinct subchannels were
triggered (e.g. server1 then server2), New PickFirst does not
deduplicate and yields 2 connections, causing verifyConnection(1) to
fail intermittently.

Fix this by iterating deterministically in server order, which reliably
triggers two distinct subchannels (server1 and server2) in both old and
new PickFirst.

Failure test run in fusion2/d9446b33-dc0e-4dee-b901-ea9b3b57f3c6

Coverage Stats

39413 of 44062 relevant lines covered (89.45%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.0

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

/*
 * Copyright 2015 The gRPC Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.grpc.internal;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
import static io.grpc.internal.GrpcAttributes.ATTR_SECURITY_LEVEL;
import static io.grpc.internal.GrpcUtil.ACCEPT_ENCODING_SPLITTER;
import static io.grpc.internal.GrpcUtil.CONTENT_LENGTH_KEY;
import static io.grpc.internal.GrpcUtil.MESSAGE_ACCEPT_ENCODING_KEY;
import static io.grpc.internal.GrpcUtil.MESSAGE_ENCODING_KEY;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Throwables;
import com.google.common.util.concurrent.MoreExecutors;
import io.grpc.Attributes;
import io.grpc.Codec;
import io.grpc.Compressor;
import io.grpc.CompressorRegistry;
import io.grpc.Context;
import io.grpc.DecompressorRegistry;
import io.grpc.InternalDecompressorRegistry;
import io.grpc.InternalStatus;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.SecurityLevel;
import io.grpc.ServerCall;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.perfmark.PerfMark;
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;

final class ServerCallImpl<ReqT, RespT> extends ServerCall<ReqT, RespT> {

  private static final Logger log = Logger.getLogger(ServerCallImpl.class.getName());

  @VisibleForTesting
  static final String TOO_MANY_RESPONSES = "Too many responses";
  @VisibleForTesting
  static final String MISSING_RESPONSE = "Completed without a response";

  private final ServerStream stream;
  private final MethodDescriptor<ReqT, RespT> method;
  private final Tag tag;
  private final Context.CancellableContext context;
  private final byte[] messageAcceptEncoding;
  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;
  private boolean sendHeadersCalled;
  private boolean closeCalled;
  private Compressor compressor;
  private boolean messageSent;

  ServerCallImpl(ServerStream stream, MethodDescriptor<ReqT, RespT> method,
      Metadata inboundHeaders, Context.CancellableContext context,
      DecompressorRegistry decompressorRegistry, CompressorRegistry compressorRegistry,
      CallTracer serverCallTracer, Tag tag) {
    this.stream = stream;
    this.method = method;
    this.context = context;
    this.messageAcceptEncoding = inboundHeaders.get(MESSAGE_ACCEPT_ENCODING_KEY);
    this.decompressorRegistry = decompressorRegistry;
    this.compressorRegistry = compressorRegistry;
    this.serverCallTracer = serverCallTracer;
    this.serverCallTracer.reportCallStarted();
    this.tag = tag;
  }

  @Override
  public void request(int numMessages) {
    try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.request")) {
      PerfMark.attachTag(tag);
      if (method.getType().clientSendsOneMessage() && numMessages == 1) {
        numMessages = 2;
      }
      stream.request(numMessages);
    }
  }

  @Override
  public void sendHeaders(Metadata headers) {
    try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.sendHeaders")) {
      PerfMark.attachTag(tag);
      sendHeadersInternal(headers);
    }
  }

  private void sendHeadersInternal(Metadata headers) {
    checkState(!sendHeadersCalled, "sendHeaders has already been called");
    checkState(!closeCalled, "call is closed");

    headers.discardAll(CONTENT_LENGTH_KEY);
    headers.discardAll(MESSAGE_ENCODING_KEY);
    if (compressor == null) {
      compressor = Codec.Identity.NONE;
    } else {
      if (messageAcceptEncoding != null) {
        // TODO(carl-mastrangelo): remove the string allocation.
        if (!GrpcUtil.iterableContains(
            ACCEPT_ENCODING_SPLITTER.split(new String(messageAcceptEncoding, GrpcUtil.US_ASCII)),
            compressor.getMessageEncoding())) {
          // resort to using no compression.
          compressor = Codec.Identity.NONE;
        }
      } else {
        compressor = Codec.Identity.NONE;
      }
    }

    // Always put compressor, even if it's identity.
    headers.put(MESSAGE_ENCODING_KEY, compressor.getMessageEncoding());

    stream.setCompressor(compressor);

    headers.discardAll(MESSAGE_ACCEPT_ENCODING_KEY);
    byte[] advertisedEncodings =
        InternalDecompressorRegistry.getRawAdvertisedMessageEncodings(decompressorRegistry);
    if (advertisedEncodings.length != 0) {
      headers.put(MESSAGE_ACCEPT_ENCODING_KEY, advertisedEncodings);
    }

    // Don't check if sendMessage has been called, since it requires that sendHeaders was already
    // called.
    sendHeadersCalled = true;
    stream.writeHeaders(headers, !getMethodDescriptor().getType().serverSendsOneMessage());
  }

  @Override
  public void sendMessage(RespT message) {
    try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.sendMessage")) {
      PerfMark.attachTag(tag);
      sendMessageInternal(message);
    }
  }

  private void sendMessageInternal(RespT message) {
    checkState(sendHeadersCalled, "sendHeaders has not been called");
    checkState(!closeCalled, "call is closed");

    if (method.getType().serverSendsOneMessage() && messageSent) {
      handleInternalError(Status.INTERNAL.withDescription(TOO_MANY_RESPONSES).asRuntimeException());
      return;
    }

    messageSent = true;
    try {
      InputStream resp = method.streamResponse(message);
      stream.writeMessage(resp);
      if (!getMethodDescriptor().getType().serverSendsOneMessage()) {
        stream.flush();
      }
    } catch (RuntimeException e) {
      handleInternalError(e);
    } catch (Error e) {
      close(
          Status.CANCELLED.withDescription("Server sendMessage() failed with Error"),
          new Metadata());
      throw e;
    }
  }

  @Override
  public void setMessageCompression(boolean enable) {
    stream.setMessageCompression(enable);
  }

  @Override
  public void setOnReadyThreshold(int numBytes) {
    stream.setOnReadyThreshold(numBytes);
  }

  @Override
  public void setCompression(String compressorName) {
    // Added here to give a better error message.
    checkState(!sendHeadersCalled, "sendHeaders has been called");

    compressor = compressorRegistry.lookupCompressor(compressorName);
    checkArgument(compressor != null, "Unable to find compressor by name %s", compressorName);
  }

  @Override
  public boolean isReady() {
    if (closeCalled) {
      return false;
    }
    return stream.isReady();
  }

  @Override
  public void close(Status status, Metadata trailers) {
    try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.close")) {
      PerfMark.attachTag(tag);
      closeInternal(status, trailers);
    }
  }

  private void closeInternal(Status status, Metadata trailers) {
    checkState(!closeCalled, "call already closed");
    try {
      closeCalled = true;

      if (status.isOk() && method.getType().serverSendsOneMessage() && !messageSent) {
        handleInternalError(Status.INTERNAL.withDescription(MISSING_RESPONSE).asRuntimeException());
        return;
      }

      stream.close(status, trailers);
    } finally {
      reportCallEnded(status.isOk());
    }
  }

  private void reportCallEnded(boolean success) {
    if (callEndReported.compareAndSet(false, true)) {
      serverCallTracer.reportCallEnded(success);
    }
  }

  @Override
  public boolean isCancelled() {
    return cancelled;
  }

  ServerStreamListener newServerStreamListener(ServerCall.Listener<ReqT> listener) {
    return new ServerStreamListenerImpl<>(this, listener, context);
  }

  @Override
  public Attributes getAttributes() {
    return stream.getAttributes();
  }

  @Override
  public String getAuthority() {
    return stream.getAuthority();
  }

  @Override
  public MethodDescriptor<ReqT, RespT> getMethodDescriptor() {
    return method;
  }

  @Override
  public void triggerEvent(Object event) {
    try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.triggerEvent")) {
      PerfMark.attachTag(tag);
      stream.triggerEvent(event);
    }
  }

  @Override
  public SecurityLevel getSecurityLevel() {
    final Attributes attributes = getAttributes();
    if (attributes == null) {
      return super.getSecurityLevel();
    }
    final SecurityLevel securityLevel = attributes.get(ATTR_SECURITY_LEVEL);
    return securityLevel == null ? super.getSecurityLevel() : securityLevel;
  }

  /**
   * Close the {@link ServerStream} because an internal error occurred. Allow the application to
   * run until completion, but silently ignore interactions with the {@link ServerStream} from now
   * on.
   */
  private void handleInternalError(Throwable internalError) {
    log.log(Level.WARNING, "Cancelling the stream because of internal error", internalError);
    Status status = (internalError instanceof StatusRuntimeException)
        ? ((StatusRuntimeException) internalError).getStatus()
        : Status.INTERNAL.withCause(internalError)
            .withDescription("Internal error so cancelling stream.");
    stream.cancel(status);
    reportCallEnded(false); // error so always false
  }

  /**
   * All of these callbacks are assumed to called on an application thread, and the caller is
   * responsible for handling thrown exceptions.
   */
  @VisibleForTesting
  static final class ServerStreamListenerImpl<ReqT> implements ServerStreamListener {
    private final ServerCallImpl<ReqT, ?> call;
    private final ServerCall.Listener<ReqT> listener;
    private final Context.CancellableContext context;
    private InputStream delayedMessage;

    public ServerStreamListenerImpl(
        ServerCallImpl<ReqT, ?> call, ServerCall.Listener<ReqT> listener,
        Context.CancellableContext context) {
      this.call = checkNotNull(call, "call");
      this.listener = checkNotNull(listener, "listener must not be null");
      this.context = checkNotNull(context, "context");
      // Wire ourselves up so that if the context is cancelled, our flag call.cancelled also
      // reflects the new state. Use a DirectExecutor so that it happens in the same thread
      // as the caller of {@link Context#cancel}.
      this.context.addListener(
          new Context.CancellationListener() {
            @Override
            public void cancelled(Context context) {
              // If the context has a cancellation cause then something exceptional happened
              // and we should also mark the call as cancelled.
              if (context.cancellationCause() != null) {
                ServerStreamListenerImpl.this.call.cancelled = true;
              }
            }
          },
          MoreExecutors.directExecutor());
    }

    @Override
    public void messagesAvailable(MessageProducer producer) {
      try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.messagesAvailable")) {
        PerfMark.attachTag(call.tag);
        messagesAvailableInternal(producer);
      }
    }

    @SuppressWarnings("Finally") // The code avoids suppressing the exception thrown from try
    private void messagesAvailableInternal(final MessageProducer producer) {
      if (call.cancelled) {
        GrpcUtil.closeQuietly(producer);
        return;
      }

      InputStream message;
      try {
        while ((message = producer.next()) != null) {
          // TODO: Consider forcing this check to be done in the transport (MessageDeframer)
          // https://github.com/grpc/grpc-java/pull/13004/changes#r3939373996
          if (call.method.getType().clientSendsOneMessage()) {
            if (delayedMessage != null) {
              GrpcUtil.closeQuietly(message);
              call.stream.cancel(Status.INTERNAL.withDescription("Too many requests"));
              GrpcUtil.closeQuietly(delayedMessage);
              delayedMessage = null;
              call.cancelled = true;
              return;
            }
            delayedMessage = message;
          } else {
            try {
              listener.onMessage(call.method.parseRequest(message));
            } catch (Throwable t) {
              GrpcUtil.closeQuietly(message);
              throw t;
            }
            message.close();
          }
        }
      } catch (Throwable t) {
        GrpcUtil.closeQuietly(producer);
        Throwables.throwIfUnchecked(t);
        throw new RuntimeException(t);
      }
    }

    @Override
    public void halfClosed() {
      try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.halfClosed")) {
        PerfMark.attachTag(call.tag);
        if (call.cancelled) {
          return;
        }

        if (delayedMessage != null) {
          InputStream message = delayedMessage;
          delayedMessage = null;
          try {
            listener.onMessage(call.method.parseRequest(message));
          } catch (Throwable t) {
            GrpcUtil.closeQuietly(message);
            Throwables.throwIfUnchecked(t);
            throw new RuntimeException(t);
          }
          GrpcUtil.closeQuietly(message);
        }

        listener.onHalfClose();
      }
    }

    @Override
    public void closed(Status status) {
      try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.closed")) {
        PerfMark.attachTag(call.tag);
        closedInternal(status);
      }
    }

    private void closedInternal(Status status) {
      if (delayedMessage != null) {
        GrpcUtil.closeQuietly(delayedMessage);
        delayedMessage = null;
      }
      Throwable cancelCause = null;
      try {
        if (status.isOk()) {
          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.
          cancelCause = InternalStatus.asRuntimeExceptionWithoutStacktrace(
              Status.CANCELLED.withDescription("RPC cancelled"), null);
        }
      } finally {
        // Cancel context after delivering RPC closure notification to allow the application to
        // clean up and update any state based on whether onComplete or onCancel was called.
        context.cancel(cancelCause);
      }
    }

    @Override
    public void onReady() {
      try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.onReady")) {
        PerfMark.attachTag(call.tag);
        if (call.cancelled) {
          return;
        }
        listener.onReady();
      }
    }

    @Override
    public void triggerEvent(Object event) {
      try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.triggerEvent")) {
        PerfMark.attachTag(call.tag);
        if (call.cancelled) {
          return;
        }
        listener.onEvent(event);
      }
    }
  }
}

1	/*
2	* Copyright 2015 The gRPC Authors
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16
17	package io.grpc.internal;
18
19	import static com.google.common.base.Preconditions.checkArgument;
20	import static com.google.common.base.Preconditions.checkNotNull;
21	import static com.google.common.base.Preconditions.checkState;
22	import static io.grpc.internal.GrpcAttributes.ATTR_SECURITY_LEVEL;
23	import static io.grpc.internal.GrpcUtil.ACCEPT_ENCODING_SPLITTER;
24	import static io.grpc.internal.GrpcUtil.CONTENT_LENGTH_KEY;
25	import static io.grpc.internal.GrpcUtil.MESSAGE_ACCEPT_ENCODING_KEY;
26	import static io.grpc.internal.GrpcUtil.MESSAGE_ENCODING_KEY;
27
28	import com.google.common.annotations.VisibleForTesting;
29	import com.google.common.base.Throwables;
30	import com.google.common.util.concurrent.MoreExecutors;
31	import io.grpc.Attributes;
32	import io.grpc.Codec;
33	import io.grpc.Compressor;
34	import io.grpc.CompressorRegistry;
35	import io.grpc.Context;
36	import io.grpc.DecompressorRegistry;
37	import io.grpc.InternalDecompressorRegistry;
38	import io.grpc.InternalStatus;
39	import io.grpc.Metadata;
40	import io.grpc.MethodDescriptor;
41	import io.grpc.SecurityLevel;
42	import io.grpc.ServerCall;
43	import io.grpc.Status;
44	import io.grpc.StatusRuntimeException;
45	import io.perfmark.PerfMark;
46	import io.perfmark.Tag;
47	import io.perfmark.TaskCloseable;
48	import java.io.InputStream;
49	import java.util.concurrent.atomic.AtomicBoolean;
50	import java.util.logging.Level;
51	import java.util.logging.Logger;
52
53	final class ServerCallImpl<ReqT, RespT> extends ServerCall<ReqT, RespT> {
54
55	private static final Logger log = Logger.getLogger(ServerCallImpl.class.getName());	1 ✔
56
57	@VisibleForTesting
58	static final String TOO_MANY_RESPONSES = "Too many responses";
59	@VisibleForTesting
60	static final String MISSING_RESPONSE = "Completed without a response";
61
62	private final ServerStream stream;
63	private final MethodDescriptor<ReqT, RespT> method;
64	private final Tag tag;
65	private final Context.CancellableContext context;
66	private final byte[] messageAcceptEncoding;
67	private final DecompressorRegistry decompressorRegistry;
68	private final CompressorRegistry compressorRegistry;
69	private CallTracer serverCallTracer;
70	// The application's close() and the transport's closed() may race; only the first is counted.
71	private final AtomicBoolean callEndReported = new AtomicBoolean();	1 ✔
72
73	// state
74	private volatile boolean cancelled;
75	private boolean sendHeadersCalled;
76	private boolean closeCalled;
77	private Compressor compressor;
78	private boolean messageSent;
79
80	ServerCallImpl(ServerStream stream, MethodDescriptor<ReqT, RespT> method,
81	Metadata inboundHeaders, Context.CancellableContext context,
82	DecompressorRegistry decompressorRegistry, CompressorRegistry compressorRegistry,
83	CallTracer serverCallTracer, Tag tag) {	1 ✔
84	this.stream = stream;	1 ✔
85	this.method = method;	1 ✔
86	this.context = context;	1 ✔
87	this.messageAcceptEncoding = inboundHeaders.get(MESSAGE_ACCEPT_ENCODING_KEY);	1 ✔
88	this.decompressorRegistry = decompressorRegistry;	1 ✔
89	this.compressorRegistry = compressorRegistry;	1 ✔
90	this.serverCallTracer = serverCallTracer;	1 ✔
91	this.serverCallTracer.reportCallStarted();	1 ✔
92	this.tag = tag;	1 ✔
93	}	1 ✔
94
95	@Override
96	public void request(int numMessages) {
97	try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.request")) {	1 ✔
98	PerfMark.attachTag(tag);	1 ✔
99	if (method.getType().clientSendsOneMessage() && numMessages == 1) {	1 ✔
100	numMessages = 2;	1 ✔
101	}
102	stream.request(numMessages);	1 ✔
103	}
104	}	1 ✔
105
106	@Override
107	public void sendHeaders(Metadata headers) {
108	try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.sendHeaders")) {	1 ✔
109	PerfMark.attachTag(tag);	1 ✔
110	sendHeadersInternal(headers);	1 ✔
111	}
112	}	1 ✔
113
114	private void sendHeadersInternal(Metadata headers) {
115	checkState(!sendHeadersCalled, "sendHeaders has already been called");	1 ✔
116	checkState(!closeCalled, "call is closed");	1 ✔
117
118	headers.discardAll(CONTENT_LENGTH_KEY);	1 ✔
119	headers.discardAll(MESSAGE_ENCODING_KEY);	1 ✔
120	if (compressor == null) {	1 ✔
121	compressor = Codec.Identity.NONE;	1 ✔
122	} else {
123	if (messageAcceptEncoding != null) {	1 ✔
124	// TODO(carl-mastrangelo): remove the string allocation.
125	if (!GrpcUtil.iterableContains(	1 ✔
126	ACCEPT_ENCODING_SPLITTER.split(new String(messageAcceptEncoding, GrpcUtil.US_ASCII)),	1 ✔
127	compressor.getMessageEncoding())) {	1 ✔
128	// resort to using no compression.
129	compressor = Codec.Identity.NONE;	1 ✔
130	}
131	} else {
132	compressor = Codec.Identity.NONE;	1 ✔
133	}
134	}
135
136	// Always put compressor, even if it's identity.
137	headers.put(MESSAGE_ENCODING_KEY, compressor.getMessageEncoding());	1 ✔
138
139	stream.setCompressor(compressor);	1 ✔
140
141	headers.discardAll(MESSAGE_ACCEPT_ENCODING_KEY);	1 ✔
142	byte[] advertisedEncodings =	1 ✔
143	InternalDecompressorRegistry.getRawAdvertisedMessageEncodings(decompressorRegistry);	1 ✔
144	if (advertisedEncodings.length != 0) {	1 ✔
145	headers.put(MESSAGE_ACCEPT_ENCODING_KEY, advertisedEncodings);	1 ✔
146	}
147
148	// Don't check if sendMessage has been called, since it requires that sendHeaders was already
149	// called.
150	sendHeadersCalled = true;	1 ✔
151	stream.writeHeaders(headers, !getMethodDescriptor().getType().serverSendsOneMessage());	1 ✔
152	}	1 ✔
153
154	@Override
155	public void sendMessage(RespT message) {
156	try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.sendMessage")) {	1 ✔
157	PerfMark.attachTag(tag);	1 ✔
158	sendMessageInternal(message);	1 ✔
159	}
160	}	1 ✔
161
162	private void sendMessageInternal(RespT message) {
163	checkState(sendHeadersCalled, "sendHeaders has not been called");	1 ✔
164	checkState(!closeCalled, "call is closed");	1 ✔
165
166	if (method.getType().serverSendsOneMessage() && messageSent) {	1 ✔
167	handleInternalError(Status.INTERNAL.withDescription(TOO_MANY_RESPONSES).asRuntimeException());	1 ✔
168	return;	1 ✔
169	}
170
171	messageSent = true;	1 ✔
172	try {
173	InputStream resp = method.streamResponse(message);	1 ✔
174	stream.writeMessage(resp);	1 ✔
175	if (!getMethodDescriptor().getType().serverSendsOneMessage()) {	1 ✔
176	stream.flush();	1 ✔
177	}
178	} catch (RuntimeException e) {	1 ✔
179	handleInternalError(e);	1 ✔
180	} catch (Error e) {	×
181	close(	×
182	Status.CANCELLED.withDescription("Server sendMessage() failed with Error"),	×
183	new Metadata());
184	throw e;	×
185	}	1 ✔
186	}	1 ✔
187
188	@Override
189	public void setMessageCompression(boolean enable) {
190	stream.setMessageCompression(enable);	1 ✔
191	}	1 ✔
192
193	@Override
194	public void setOnReadyThreshold(int numBytes) {
195	stream.setOnReadyThreshold(numBytes);	1 ✔
196	}	1 ✔
197
198	@Override
199	public void setCompression(String compressorName) {
200	// Added here to give a better error message.
201	checkState(!sendHeadersCalled, "sendHeaders has been called");	1 ✔
202
203	compressor = compressorRegistry.lookupCompressor(compressorName);	1 ✔
204	checkArgument(compressor != null, "Unable to find compressor by name %s", compressorName);	1 ✔
205	}	1 ✔
206
207	@Override
208	public boolean isReady() {
209	if (closeCalled) {	1 ✔
210	return false;	1 ✔
211	}
212	return stream.isReady();	1 ✔
213	}
214
215	@Override
216	public void close(Status status, Metadata trailers) {
217	try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.close")) {	1 ✔
218	PerfMark.attachTag(tag);	1 ✔
219	closeInternal(status, trailers);	1 ✔
220	}
221	}	1 ✔
222
223	private void closeInternal(Status status, Metadata trailers) {
224	checkState(!closeCalled, "call already closed");	1 ✔
225	try {
226	closeCalled = true;	1 ✔
227
228	if (status.isOk() && method.getType().serverSendsOneMessage() && !messageSent) {	1 ✔
229	handleInternalError(Status.INTERNAL.withDescription(MISSING_RESPONSE).asRuntimeException());	1 ✔
230	return;	1 ✔
231	}
232
233	stream.close(status, trailers);	1 ✔
234	} finally {
235	reportCallEnded(status.isOk());	1 ✔
236	}
237	}	1 ✔
238
239	private void reportCallEnded(boolean success) {
240	if (callEndReported.compareAndSet(false, true)) {	1 ✔
241	serverCallTracer.reportCallEnded(success);	1 ✔
242	}
243	}	1 ✔
244
245	@Override
246	public boolean isCancelled() {
247	return cancelled;	1 ✔
248	}
249
250	ServerStreamListener newServerStreamListener(ServerCall.Listener<ReqT> listener) {
251	return new ServerStreamListenerImpl<>(this, listener, context);	1 ✔
252	}
253
254	@Override
255	public Attributes getAttributes() {
256	return stream.getAttributes();	1 ✔
257	}
258
259	@Override
260	public String getAuthority() {
261	return stream.getAuthority();	1 ✔
262	}
263
264	@Override
265	public MethodDescriptor<ReqT, RespT> getMethodDescriptor() {
266	return method;	1 ✔
267	}
268
269	@Override
270	public void triggerEvent(Object event) {
271	try (TaskCloseable ignore = PerfMark.traceTask("ServerCall.triggerEvent")) {	1 ✔
272	PerfMark.attachTag(tag);	1 ✔
273	stream.triggerEvent(event);	1 ✔
274	}
275	}	1 ✔
276
277	@Override
278	public SecurityLevel getSecurityLevel() {
279	final Attributes attributes = getAttributes();	1 ✔
280	if (attributes == null) {	1 ✔
281	return super.getSecurityLevel();	1 ✔
282	}
283	final SecurityLevel securityLevel = attributes.get(ATTR_SECURITY_LEVEL);	1 ✔
284	return securityLevel == null ? super.getSecurityLevel() : securityLevel;	1 ✔
285	}
286
287	/**
288	* Close the {@link ServerStream} because an internal error occurred. Allow the application to
289	* run until completion, but silently ignore interactions with the {@link ServerStream} from now
290	* on.
291	*/
292	private void handleInternalError(Throwable internalError) {
293	log.log(Level.WARNING, "Cancelling the stream because of internal error", internalError);	1 ✔
294	Status status = (internalError instanceof StatusRuntimeException)	1 ✔
295	? ((StatusRuntimeException) internalError).getStatus()	1 ✔
296	: Status.INTERNAL.withCause(internalError)	1 ✔
297	.withDescription("Internal error so cancelling stream.");	1 ✔
298	stream.cancel(status);	1 ✔
299	reportCallEnded(false); // error so always false	1 ✔
300	}	1 ✔
301
302	/**
303	* All of these callbacks are assumed to called on an application thread, and the caller is
304	* responsible for handling thrown exceptions.
305	*/
306	@VisibleForTesting
307	static final class ServerStreamListenerImpl<ReqT> implements ServerStreamListener {
308	private final ServerCallImpl<ReqT, ?> call;
309	private final ServerCall.Listener<ReqT> listener;
310	private final Context.CancellableContext context;
311	private InputStream delayedMessage;
312
313	public ServerStreamListenerImpl(
314	ServerCallImpl<ReqT, ?> call, ServerCall.Listener<ReqT> listener,
315	Context.CancellableContext context) {	1 ✔
316	this.call = checkNotNull(call, "call");	1 ✔
317	this.listener = checkNotNull(listener, "listener must not be null");	1 ✔
318	this.context = checkNotNull(context, "context");	1 ✔
319	// Wire ourselves up so that if the context is cancelled, our flag call.cancelled also
320	// reflects the new state. Use a DirectExecutor so that it happens in the same thread
321	// as the caller of {@link Context#cancel}.
322	this.context.addListener(	1 ✔
323	new Context.CancellationListener() {	1 ✔
324	@Override
325	public void cancelled(Context context) {
326	// If the context has a cancellation cause then something exceptional happened
327	// and we should also mark the call as cancelled.
328	if (context.cancellationCause() != null) {	1 ✔
329	ServerStreamListenerImpl.this.call.cancelled = true;	1 ✔
330	}
331	}	1 ✔
332	},
333	MoreExecutors.directExecutor());	1 ✔
334	}	1 ✔
335
336	@Override
337	public void messagesAvailable(MessageProducer producer) {
338	try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.messagesAvailable")) {	1 ✔
339	PerfMark.attachTag(call.tag);	1 ✔
340	messagesAvailableInternal(producer);	1 ✔
341	}
342	}	1 ✔
343
344	@SuppressWarnings("Finally") // The code avoids suppressing the exception thrown from try
345	private void messagesAvailableInternal(final MessageProducer producer) {
346	if (call.cancelled) {	1 ✔
347	GrpcUtil.closeQuietly(producer);	×
348	return;	×
349	}
350
351	InputStream message;
352	try {
353	while ((message = producer.next()) != null) {	1 ✔
354	// TODO: Consider forcing this check to be done in the transport (MessageDeframer)
355	// https://github.com/grpc/grpc-java/pull/13004/changes#r3939373996
356	if (call.method.getType().clientSendsOneMessage()) {	1 ✔
357	if (delayedMessage != null) {	1 ✔
358	GrpcUtil.closeQuietly(message);	1 ✔
359	call.stream.cancel(Status.INTERNAL.withDescription("Too many requests"));	1 ✔
360	GrpcUtil.closeQuietly(delayedMessage);	1 ✔
361	delayedMessage = null;	1 ✔
362	call.cancelled = true;	1 ✔
363	return;	1 ✔
364	}
365	delayedMessage = message;	1 ✔
366	} else {
367	try {
368	listener.onMessage(call.method.parseRequest(message));	1 ✔
369	} catch (Throwable t) {	1 ✔
370	GrpcUtil.closeQuietly(message);	1 ✔
371	throw t;	1 ✔
372	}	1 ✔
373	message.close();	1 ✔
374	}
375	}
376	} catch (Throwable t) {	1 ✔
377	GrpcUtil.closeQuietly(producer);	1 ✔
378	Throwables.throwIfUnchecked(t);	×
379	throw new RuntimeException(t);	×
380	}	1 ✔
381	}	1 ✔
382
383	@Override
384	public void halfClosed() {
385	try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.halfClosed")) {	1 ✔
386	PerfMark.attachTag(call.tag);	1 ✔
387	if (call.cancelled) {	1 ✔
388	return;	1 ✔
389	}
390
391	if (delayedMessage != null) {	1 ✔
392	InputStream message = delayedMessage;	1 ✔
393	delayedMessage = null;	1 ✔
394	try {
395	listener.onMessage(call.method.parseRequest(message));	1 ✔
396	} catch (Throwable t) {	1 ✔
397	GrpcUtil.closeQuietly(message);	1 ✔
398	Throwables.throwIfUnchecked(t);	×
399	throw new RuntimeException(t);	×
400	}	1 ✔
401	GrpcUtil.closeQuietly(message);	1 ✔
402	}
403
404	listener.onHalfClose();	1 ✔
405	}
406	}	1 ✔
407
408	@Override
409	public void closed(Status status) {
410	try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.closed")) {	1 ✔
411	PerfMark.attachTag(call.tag);	1 ✔
412	closedInternal(status);	1 ✔
413	}
414	}	1 ✔
415
416	private void closedInternal(Status status) {
417	if (delayedMessage != null) {	1 ✔
418	GrpcUtil.closeQuietly(delayedMessage);	1 ✔
419	delayedMessage = null;	1 ✔
420	}
421	Throwable cancelCause = null;	1 ✔
422	try {
423	if (status.isOk()) {	1 ✔
424	listener.onComplete();	1 ✔
425	} else {
426	call.cancelled = true;	1 ✔
427	call.reportCallEnded(false);	1 ✔
428	listener.onCancel();	1 ✔
429	// The status will not have a cause in all failure scenarios, but we want to make sure
430	// we always cancel the context with one to keep the context cancelled state consistent.
431	cancelCause = InternalStatus.asRuntimeExceptionWithoutStacktrace(	1 ✔
432	Status.CANCELLED.withDescription("RPC cancelled"), null);	1 ✔
433	}
434	} finally {
435	// Cancel context after delivering RPC closure notification to allow the application to
436	// clean up and update any state based on whether onComplete or onCancel was called.
437	context.cancel(cancelCause);	1 ✔
438	}
439	}	1 ✔
440
441	@Override
442	public void onReady() {
443	try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.onReady")) {	1 ✔
444	PerfMark.attachTag(call.tag);	1 ✔
445	if (call.cancelled) {	1 ✔
446	return;	1 ✔
447	}
448	listener.onReady();	1 ✔
449	}
450	}	1 ✔
451
452	@Override
453	public void triggerEvent(Object event) {
454	try (TaskCloseable ignore = PerfMark.traceTask("ServerStreamListener.triggerEvent")) {	1 ✔
455	PerfMark.attachTag(call.tag);	1 ✔
456	if (call.cancelled) {	1 ✔
457	return;	1 ✔
458	}
459	listener.onEvent(event);	1 ✔
460	}
461	}	1 ✔
462	}
463	}

grpc / grpc-java / #20551

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous