#19108

Committed 21 Mar 2024 10:37PM UTC coverage: 88.277% (-0.002%) from 88.279%

Build # #19108

Build Type

push

github

Committed by

web-flow

Commit Message

Allow configuration of the queued byte threshold at which a Stream is considered not ready (#10977)

* Allow the queued byte threshold for a Stream to be ready to be configurable

- on clients this is exposed by setting a CallOption
- on servers this is configured by calling a method on ServerCall or ServerStreamListener

Run Details

31190 of 35332 relevant lines covered (88.28%)

0.88 hits per line

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

95.21

/../netty/src/main/java/io/grpc/netty/NettyClientTransport.java

/*
 * Copyright 2014 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.netty;

import static io.grpc.internal.GrpcUtil.KEEPALIVE_TIME_NANOS_DISABLED;
import static io.netty.channel.ChannelOption.ALLOCATOR;
import static io.netty.channel.ChannelOption.SO_KEEPALIVE;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import com.google.common.base.Preconditions;
import com.google.common.base.Ticker;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import io.grpc.Attributes;
import io.grpc.CallOptions;
import io.grpc.ChannelLogger;
import io.grpc.ClientStreamTracer;
import io.grpc.InternalChannelz.SocketStats;
import io.grpc.InternalLogId;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.Status;
import io.grpc.internal.ClientStream;
import io.grpc.internal.ConnectionClientTransport;
import io.grpc.internal.FailingClientStream;
import io.grpc.internal.GrpcUtil;
import io.grpc.internal.Http2Ping;
import io.grpc.internal.KeepAliveManager;
import io.grpc.internal.KeepAliveManager.ClientKeepAlivePinger;
import io.grpc.internal.StatsTraceContext;
import io.grpc.internal.TransportTracer;
import io.grpc.netty.NettyChannelBuilder.LocalSocketPicker;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFactory;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoop;
import io.netty.channel.EventLoopGroup;
import io.netty.handler.codec.http2.StreamBufferingEncoder.Http2ChannelClosedException;
import io.netty.util.AsciiString;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import java.net.SocketAddress;
import java.nio.channels.ClosedChannelException;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;

/**
 * A Netty-based {@link ConnectionClientTransport} implementation.
 */
class NettyClientTransport implements ConnectionClientTransport {

  private final InternalLogId logId;
  private final Map<ChannelOption<?>, ?> channelOptions;
  private final SocketAddress remoteAddress;
  private final ChannelFactory<? extends Channel> channelFactory;
  private final EventLoopGroup group;
  private final ProtocolNegotiator negotiator;
  private final String authorityString;
  private final AsciiString authority;
  private final AsciiString userAgent;
  private final boolean autoFlowControl;
  private final int flowControlWindow;
  private final int maxMessageSize;
  private final int maxHeaderListSize;
  private KeepAliveManager keepAliveManager;
  private final long keepAliveTimeNanos;
  private final long keepAliveTimeoutNanos;
  private final boolean keepAliveWithoutCalls;
  private final AsciiString negotiationScheme;
  private final Runnable tooManyPingsRunnable;
  private NettyClientHandler handler;
  // We should not send on the channel until negotiation completes. This is a hard requirement
  // by SslHandler but is appropriate for HTTP/1.1 Upgrade as well.
  private Channel channel;
  /** If {@link #start} has been called, non-{@code null} if channel is {@code null}. */
  private Status statusExplainingWhyTheChannelIsNull;
  /** Since not thread-safe, may only be used from event loop. */
  private ClientTransportLifecycleManager lifecycleManager;
  /** Since not thread-safe, may only be used from event loop. */
  private final TransportTracer transportTracer;
  private final Attributes eagAttributes;
  private final LocalSocketPicker localSocketPicker;
  private final ChannelLogger channelLogger;
  private final boolean useGetForSafeMethods;
  private final Ticker ticker;

  NettyClientTransport(
      SocketAddress address, ChannelFactory<? extends Channel> channelFactory,
      Map<ChannelOption<?>, ?> channelOptions, EventLoopGroup group,
      ProtocolNegotiator negotiator, boolean autoFlowControl, int flowControlWindow,
      int maxMessageSize, int maxHeaderListSize,
      long keepAliveTimeNanos, long keepAliveTimeoutNanos,
      boolean keepAliveWithoutCalls, String authority, @Nullable String userAgent,
      Runnable tooManyPingsRunnable, TransportTracer transportTracer, Attributes eagAttributes,
      LocalSocketPicker localSocketPicker, ChannelLogger channelLogger,
      boolean useGetForSafeMethods, Ticker ticker) {

    this.negotiator = Preconditions.checkNotNull(negotiator, "negotiator");
    this.negotiationScheme = this.negotiator.scheme();
    this.remoteAddress = Preconditions.checkNotNull(address, "address");
    this.group = Preconditions.checkNotNull(group, "group");
    this.channelFactory = channelFactory;
    this.channelOptions = Preconditions.checkNotNull(channelOptions, "channelOptions");
    this.autoFlowControl = autoFlowControl;
    this.flowControlWindow = flowControlWindow;
    this.maxMessageSize = maxMessageSize;
    this.maxHeaderListSize = maxHeaderListSize;
    this.keepAliveTimeNanos = keepAliveTimeNanos;
    this.keepAliveTimeoutNanos = keepAliveTimeoutNanos;
    this.keepAliveWithoutCalls = keepAliveWithoutCalls;
    this.authorityString = authority;
    this.authority = new AsciiString(authority);
    this.userAgent = new AsciiString(GrpcUtil.getGrpcUserAgent("netty", userAgent));
    this.tooManyPingsRunnable =
        Preconditions.checkNotNull(tooManyPingsRunnable, "tooManyPingsRunnable");
    this.transportTracer = Preconditions.checkNotNull(transportTracer, "transportTracer");
    this.eagAttributes = Preconditions.checkNotNull(eagAttributes, "eagAttributes");
    this.localSocketPicker = Preconditions.checkNotNull(localSocketPicker, "localSocketPicker");
    this.logId = InternalLogId.allocate(getClass(), remoteAddress.toString());
    this.channelLogger = Preconditions.checkNotNull(channelLogger, "channelLogger");
    this.useGetForSafeMethods = useGetForSafeMethods;
    this.ticker = Preconditions.checkNotNull(ticker, "ticker");
  }

  @Override
  public void ping(final PingCallback callback, final Executor executor) {
    if (channel == null) {
      executor.execute(new Runnable() {
        @Override
        public void run() {
          callback.onFailure(statusExplainingWhyTheChannelIsNull.asException());
        }
      });
      return;
    }
    // The promise and listener always succeed in NettyClientHandler. So this listener handles the
    // error case, when the channel is closed and the NettyClientHandler no longer in the pipeline.
    ChannelFutureListener failureListener = new ChannelFutureListener() {
      @Override
      public void operationComplete(ChannelFuture future) throws Exception {
        if (!future.isSuccess()) {
          Status s = statusFromFailedFuture(future);
          Http2Ping.notifyFailed(callback, executor, s.asException());
        }
      }
    };
    // Write the command requesting the ping
    handler.getWriteQueue().enqueue(new SendPingCommand(callback, executor), true)
        .addListener(failureListener);
  }

  @Override
  public ClientStream newStream(
      MethodDescriptor<?, ?> method, Metadata headers, CallOptions callOptions,
      ClientStreamTracer[] tracers) {
    Preconditions.checkNotNull(method, "method");
    Preconditions.checkNotNull(headers, "headers");
    if (channel == null) {
      return new FailingClientStream(statusExplainingWhyTheChannelIsNull, tracers);
    }
    StatsTraceContext statsTraceCtx =
        StatsTraceContext.newClientContext(tracers, getAttributes(), headers);
    return new NettyClientStream(
        new NettyClientStream.TransportState(
            handler,
            channel.eventLoop(),
            maxMessageSize,
            statsTraceCtx,
            transportTracer,
            method.getFullMethodName(),
            callOptions) {
          @Override
          protected Status statusFromFailedFuture(ChannelFuture f) {
            return NettyClientTransport.this.statusFromFailedFuture(f);
          }
        },
        method,
        headers,
        channel,
        authority,
        negotiationScheme,
        userAgent,
        statsTraceCtx,
        transportTracer,
        callOptions,
        useGetForSafeMethods);
  }

  @SuppressWarnings("unchecked")
  @Override
  public Runnable start(Listener transportListener) {
    lifecycleManager = new ClientTransportLifecycleManager(
        Preconditions.checkNotNull(transportListener, "listener"));
    EventLoop eventLoop = group.next();
    if (keepAliveTimeNanos != KEEPALIVE_TIME_NANOS_DISABLED) {
      keepAliveManager = new KeepAliveManager(
          new ClientKeepAlivePinger(this), eventLoop, keepAliveTimeNanos, keepAliveTimeoutNanos,
          keepAliveWithoutCalls);
    }

    handler = NettyClientHandler.newHandler(
        lifecycleManager,
        keepAliveManager,
        autoFlowControl,
        flowControlWindow,
        maxHeaderListSize,
        GrpcUtil.STOPWATCH_SUPPLIER,
        tooManyPingsRunnable,
        transportTracer,
        eagAttributes,
        authorityString,
        channelLogger,
        ticker);

    ChannelHandler negotiationHandler = negotiator.newHandler(handler);

    Bootstrap b = new Bootstrap();
    b.option(ALLOCATOR, Utils.getByteBufAllocator(false));
    b.group(eventLoop);
    b.channelFactory(channelFactory);
    // For non-socket based channel, the option will be ignored.
    b.option(SO_KEEPALIVE, true);
    // For non-epoll based channel, the option will be ignored.
    if (keepAliveTimeNanos != KEEPALIVE_TIME_NANOS_DISABLED) {
      ChannelOption<Integer> tcpUserTimeout = Utils.maybeGetTcpUserTimeoutOption();
      if (tcpUserTimeout != null) {
        b.option(tcpUserTimeout, (int) TimeUnit.NANOSECONDS.toMillis(keepAliveTimeoutNanos));
      }
    }
    for (Map.Entry<ChannelOption<?>, ?> entry : channelOptions.entrySet()) {
      // Every entry in the map is obtained from
      // NettyChannelBuilder#withOption(ChannelOption<T> option, T value)
      // so it is safe to pass the key-value pair to b.option().
      b.option((ChannelOption<Object>) entry.getKey(), entry.getValue());
    }

    ChannelHandler bufferingHandler = new WriteBufferingAndExceptionHandler(negotiationHandler);

    /*
     * We don't use a ChannelInitializer in the client bootstrap because its "initChannel" method
     * is executed in the event loop and we need this handler to be in the pipeline immediately so
     * that it may begin buffering writes.
     */
    b.handler(bufferingHandler);
    ChannelFuture regFuture = b.register();
    if (regFuture.isDone() && !regFuture.isSuccess()) {
      channel = null;
      // Initialization has failed badly. All new streams should be made to fail.
      Throwable t = regFuture.cause();
      if (t == null) {
        t = new IllegalStateException("Channel is null, but future doesn't have a cause");
      }
      statusExplainingWhyTheChannelIsNull = Utils.statusFromThrowable(t);
      // Use a Runnable since lifecycleManager calls transportListener
      return new Runnable() {
        @Override
        public void run() {
          // NOTICE: we not are calling lifecycleManager from the event loop. But there isn't really
          // an event loop in this case, so nothing should be accessing the lifecycleManager. We
          // could use GlobalEventExecutor (which is what regFuture would use for notifying
          // listeners in this case), but avoiding on-demand thread creation in an error case seems
          // a good idea and is probably clearer threading.
          lifecycleManager.notifyTerminated(statusExplainingWhyTheChannelIsNull);
        }
      };
    }
    channel = regFuture.channel();
    // Start the write queue as soon as the channel is constructed
    handler.startWriteQueue(channel);
    // This write will have no effect, yet it will only complete once the negotiationHandler
    // flushes any pending writes. We need it to be staged *before* the `connect` so that
    // the channel can't have been closed yet, removing all handlers. This write will sit in the
    // AbstractBufferingHandler's buffer, and will either be flushed on a successful connection,
    // or failed if the connection fails.
    channel.writeAndFlush(NettyClientHandler.NOOP_MESSAGE).addListener(new ChannelFutureListener() {
      @Override
      public void operationComplete(ChannelFuture future) throws Exception {
        if (!future.isSuccess()) {
          // Need to notify of this failure, because NettyClientHandler may not have been added to
          // the pipeline before the error occurred.
          lifecycleManager.notifyTerminated(Utils.statusFromThrowable(future.cause()));
        }
      }
    });
    // Start the connection operation to the server.
    SocketAddress localAddress =
        localSocketPicker.createSocketAddress(remoteAddress, eagAttributes);
    if (localAddress != null) {
      channel.connect(remoteAddress, localAddress);
    } else {
      channel.connect(remoteAddress);
    }

    if (keepAliveManager != null) {
      keepAliveManager.onTransportStarted();
    }

    return null;
  }

  @Override
  public void shutdown(Status reason) {
    // start() could have failed
    if (channel == null) {
      return;
    }
    // Notifying of termination is automatically done when the channel closes.
    if (channel.isOpen()) {
      handler.getWriteQueue().enqueue(new GracefulCloseCommand(reason), true);
    }
  }

  @Override
  public void shutdownNow(final Status reason) {
    // Notifying of termination is automatically done when the channel closes.
    if (channel != null && channel.isOpen()) {
      handler.getWriteQueue().enqueue(new Runnable() {
        @Override
        public void run() {
          lifecycleManager.notifyShutdown(reason);
          channel.write(new ForcefulCloseCommand(reason));
        }
      }, true);
    }
  }

  @Override
  public String toString() {
    return MoreObjects.toStringHelper(this)
        .add("logId", logId.getId())
        .add("remoteAddress", remoteAddress)
        .add("channel", channel)
        .toString();
  }

  @Override
  public InternalLogId getLogId() {
    return logId;
  }

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

  @Override
  public ListenableFuture<SocketStats> getStats() {
    final SettableFuture<SocketStats> result = SettableFuture.create();
    if (channel.eventLoop().inEventLoop()) {
      // This is necessary, otherwise we will block forever if we get the future from inside
      // the event loop.
      result.set(getStatsHelper(channel));
      return result;
    }
    channel.eventLoop().submit(
        new Runnable() {
          @Override
          public void run() {
            result.set(getStatsHelper(channel));
          }
        })
        .addListener(
            new GenericFutureListener<Future<Object>>() {
              @Override
              public void operationComplete(Future<Object> future) throws Exception {
                if (!future.isSuccess()) {
                  result.setException(future.cause());
                }
              }
            });
    return result;
  }

  private SocketStats getStatsHelper(Channel ch) {
    assert ch.eventLoop().inEventLoop();
    return new SocketStats(
        transportTracer.getStats(),
        channel.localAddress(),
        channel.remoteAddress(),
        Utils.getSocketOptions(ch),
        handler == null ? null : handler.getSecurityInfo());
  }

  @VisibleForTesting
  Channel channel() {
    return channel;
  }

  @VisibleForTesting
  KeepAliveManager keepAliveManager() {
    return keepAliveManager;
  }

  /**
   * Convert ChannelFuture.cause() to a Status, taking into account that all handlers are removed
   * from the pipeline when the channel is closed. Since handlers are removed, you may get an
   * unhelpful exception like ClosedChannelException.
   *
   * <p>This method must only be called on the event loop.
   */
  private Status statusFromFailedFuture(ChannelFuture f) {
    Throwable t = f.cause();
    if (t instanceof ClosedChannelException
        // Exception thrown by the StreamBufferingEncoder if the channel is closed while there
        // are still streams buffered. This exception is not helpful. Replace it by the real
        // cause of the shutdown (if available).
        || t instanceof Http2ChannelClosedException) {
      Status shutdownStatus = lifecycleManager.getShutdownStatus();
      if (shutdownStatus == null) {
        return Status.UNKNOWN.withDescription("Channel closed but for unknown reason")
            .withCause(new ClosedChannelException().initCause(t));
      }
      return shutdownStatus;
    }
    return Utils.statusFromThrowable(t);
  }
}

1	/*
2	* Copyright 2014 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.netty;
18
19	import static io.grpc.internal.GrpcUtil.KEEPALIVE_TIME_NANOS_DISABLED;
20	import static io.netty.channel.ChannelOption.ALLOCATOR;
21	import static io.netty.channel.ChannelOption.SO_KEEPALIVE;
22
23	import com.google.common.annotations.VisibleForTesting;
24	import com.google.common.base.MoreObjects;
25	import com.google.common.base.Preconditions;
26	import com.google.common.base.Ticker;
27	import com.google.common.util.concurrent.ListenableFuture;
28	import com.google.common.util.concurrent.SettableFuture;
29	import io.grpc.Attributes;
30	import io.grpc.CallOptions;
31	import io.grpc.ChannelLogger;
32	import io.grpc.ClientStreamTracer;
33	import io.grpc.InternalChannelz.SocketStats;
34	import io.grpc.InternalLogId;
35	import io.grpc.Metadata;
36	import io.grpc.MethodDescriptor;
37	import io.grpc.Status;
38	import io.grpc.internal.ClientStream;
39	import io.grpc.internal.ConnectionClientTransport;
40	import io.grpc.internal.FailingClientStream;
41	import io.grpc.internal.GrpcUtil;
42	import io.grpc.internal.Http2Ping;
43	import io.grpc.internal.KeepAliveManager;
44	import io.grpc.internal.KeepAliveManager.ClientKeepAlivePinger;
45	import io.grpc.internal.StatsTraceContext;
46	import io.grpc.internal.TransportTracer;
47	import io.grpc.netty.NettyChannelBuilder.LocalSocketPicker;
48	import io.netty.bootstrap.Bootstrap;
49	import io.netty.channel.Channel;
50	import io.netty.channel.ChannelFactory;
51	import io.netty.channel.ChannelFuture;
52	import io.netty.channel.ChannelFutureListener;
53	import io.netty.channel.ChannelHandler;
54	import io.netty.channel.ChannelOption;
55	import io.netty.channel.EventLoop;
56	import io.netty.channel.EventLoopGroup;
57	import io.netty.handler.codec.http2.StreamBufferingEncoder.Http2ChannelClosedException;
58	import io.netty.util.AsciiString;
59	import io.netty.util.concurrent.Future;
60	import io.netty.util.concurrent.GenericFutureListener;
61	import java.net.SocketAddress;
62	import java.nio.channels.ClosedChannelException;
63	import java.util.Map;
64	import java.util.concurrent.Executor;
65	import java.util.concurrent.TimeUnit;
66	import javax.annotation.Nullable;
67
68	/**
69	* A Netty-based {@link ConnectionClientTransport} implementation.
70	*/
71	class NettyClientTransport implements ConnectionClientTransport {	1✔
72
73	private final InternalLogId logId;
74	private final Map<ChannelOption<?>, ?> channelOptions;
75	private final SocketAddress remoteAddress;
76	private final ChannelFactory<? extends Channel> channelFactory;
77	private final EventLoopGroup group;
78	private final ProtocolNegotiator negotiator;
79	private final String authorityString;
80	private final AsciiString authority;
81	private final AsciiString userAgent;
82	private final boolean autoFlowControl;
83	private final int flowControlWindow;
84	private final int maxMessageSize;
85	private final int maxHeaderListSize;
86	private KeepAliveManager keepAliveManager;
87	private final long keepAliveTimeNanos;
88	private final long keepAliveTimeoutNanos;
89	private final boolean keepAliveWithoutCalls;
90	private final AsciiString negotiationScheme;
91	private final Runnable tooManyPingsRunnable;
92	private NettyClientHandler handler;
93	// We should not send on the channel until negotiation completes. This is a hard requirement
94	// by SslHandler but is appropriate for HTTP/1.1 Upgrade as well.
95	private Channel channel;
96	/** If {@link #start} has been called, non-{@code null} if channel is {@code null}. */
97	private Status statusExplainingWhyTheChannelIsNull;
98	/** Since not thread-safe, may only be used from event loop. */
99	private ClientTransportLifecycleManager lifecycleManager;
100	/** Since not thread-safe, may only be used from event loop. */
101	private final TransportTracer transportTracer;
102	private final Attributes eagAttributes;
103	private final LocalSocketPicker localSocketPicker;
104	private final ChannelLogger channelLogger;
105	private final boolean useGetForSafeMethods;
106	private final Ticker ticker;
107
108	NettyClientTransport(
109	SocketAddress address, ChannelFactory<? extends Channel> channelFactory,
110	Map<ChannelOption<?>, ?> channelOptions, EventLoopGroup group,
111	ProtocolNegotiator negotiator, boolean autoFlowControl, int flowControlWindow,
112	int maxMessageSize, int maxHeaderListSize,
113	long keepAliveTimeNanos, long keepAliveTimeoutNanos,
114	boolean keepAliveWithoutCalls, String authority, @Nullable String userAgent,
115	Runnable tooManyPingsRunnable, TransportTracer transportTracer, Attributes eagAttributes,
116	LocalSocketPicker localSocketPicker, ChannelLogger channelLogger,
117	boolean useGetForSafeMethods, Ticker ticker) {	1✔
118
119	this.negotiator = Preconditions.checkNotNull(negotiator, "negotiator");	1✔
120	this.negotiationScheme = this.negotiator.scheme();	1✔
121	this.remoteAddress = Preconditions.checkNotNull(address, "address");	1✔
122	this.group = Preconditions.checkNotNull(group, "group");	1✔
123	this.channelFactory = channelFactory;	1✔
124	this.channelOptions = Preconditions.checkNotNull(channelOptions, "channelOptions");	1✔
125	this.autoFlowControl = autoFlowControl;	1✔
126	this.flowControlWindow = flowControlWindow;	1✔
127	this.maxMessageSize = maxMessageSize;	1✔
128	this.maxHeaderListSize = maxHeaderListSize;	1✔
129	this.keepAliveTimeNanos = keepAliveTimeNanos;	1✔
130	this.keepAliveTimeoutNanos = keepAliveTimeoutNanos;	1✔
131	this.keepAliveWithoutCalls = keepAliveWithoutCalls;	1✔
132	this.authorityString = authority;	1✔
133	this.authority = new AsciiString(authority);	1✔
134	this.userAgent = new AsciiString(GrpcUtil.getGrpcUserAgent("netty", userAgent));	1✔
135	this.tooManyPingsRunnable =	1✔
136	Preconditions.checkNotNull(tooManyPingsRunnable, "tooManyPingsRunnable");	1✔
137	this.transportTracer = Preconditions.checkNotNull(transportTracer, "transportTracer");	1✔
138	this.eagAttributes = Preconditions.checkNotNull(eagAttributes, "eagAttributes");	1✔
139	this.localSocketPicker = Preconditions.checkNotNull(localSocketPicker, "localSocketPicker");	1✔
140	this.logId = InternalLogId.allocate(getClass(), remoteAddress.toString());	1✔
141	this.channelLogger = Preconditions.checkNotNull(channelLogger, "channelLogger");	1✔
142	this.useGetForSafeMethods = useGetForSafeMethods;	1✔
143	this.ticker = Preconditions.checkNotNull(ticker, "ticker");	1✔
144	}	1✔
145
146	@Override
147	public void ping(final PingCallback callback, final Executor executor) {
148	if (channel == null) {	1✔
149	executor.execute(new Runnable() {	1✔
150	@Override
151	public void run() {
152	callback.onFailure(statusExplainingWhyTheChannelIsNull.asException());	1✔
153	}	1✔
154	});
155	return;	1✔
156	}
157	// The promise and listener always succeed in NettyClientHandler. So this listener handles the
158	// error case, when the channel is closed and the NettyClientHandler no longer in the pipeline.
159	ChannelFutureListener failureListener = new ChannelFutureListener() {	1✔
160	@Override
161	public void operationComplete(ChannelFuture future) throws Exception {
162	if (!future.isSuccess()) {	1✔
163	Status s = statusFromFailedFuture(future);	1✔
164	Http2Ping.notifyFailed(callback, executor, s.asException());	1✔
165	}
166	}	1✔
167	};
168	// Write the command requesting the ping
169	handler.getWriteQueue().enqueue(new SendPingCommand(callback, executor), true)	1✔
170	.addListener(failureListener);	1✔
171	}	1✔
172
173	@Override
174	public ClientStream newStream(
175	MethodDescriptor<?, ?> method, Metadata headers, CallOptions callOptions,
176	ClientStreamTracer[] tracers) {
177	Preconditions.checkNotNull(method, "method");	1✔
178	Preconditions.checkNotNull(headers, "headers");	1✔
179	if (channel == null) {	1✔
180	return new FailingClientStream(statusExplainingWhyTheChannelIsNull, tracers);	1✔
181	}
182	StatsTraceContext statsTraceCtx =	1✔
183	StatsTraceContext.newClientContext(tracers, getAttributes(), headers);	1✔
184	return new NettyClientStream(	1✔
185	new NettyClientStream.TransportState(
186	handler,
187	channel.eventLoop(),	1✔
188	maxMessageSize,
189	statsTraceCtx,
190	transportTracer,
191	method.getFullMethodName(),	1✔
192	callOptions) {	1✔
193	@Override
194	protected Status statusFromFailedFuture(ChannelFuture f) {
195	return NettyClientTransport.this.statusFromFailedFuture(f);	1✔
196	}
197	},
198	method,
199	headers,
200	channel,
201	authority,
202	negotiationScheme,
203	userAgent,
204	statsTraceCtx,
205	transportTracer,
206	callOptions,
207	useGetForSafeMethods);
208	}
209
210	@SuppressWarnings("unchecked")
211	@Override
212	public Runnable start(Listener transportListener) {
213	lifecycleManager = new ClientTransportLifecycleManager(	1✔
214	Preconditions.checkNotNull(transportListener, "listener"));	1✔
215	EventLoop eventLoop = group.next();	1✔
216	if (keepAliveTimeNanos != KEEPALIVE_TIME_NANOS_DISABLED) {	1✔
217	keepAliveManager = new KeepAliveManager(	1✔
218	new ClientKeepAlivePinger(this), eventLoop, keepAliveTimeNanos, keepAliveTimeoutNanos,
219	keepAliveWithoutCalls);
220	}
221
222	handler = NettyClientHandler.newHandler(	1✔
223	lifecycleManager,
224	keepAliveManager,
225	autoFlowControl,
226	flowControlWindow,
227	maxHeaderListSize,
228	GrpcUtil.STOPWATCH_SUPPLIER,
229	tooManyPingsRunnable,
230	transportTracer,
231	eagAttributes,
232	authorityString,
233	channelLogger,
234	ticker);
235
236	ChannelHandler negotiationHandler = negotiator.newHandler(handler);	1✔
237
238	Bootstrap b = new Bootstrap();	1✔
239	b.option(ALLOCATOR, Utils.getByteBufAllocator(false));	1✔
240	b.group(eventLoop);	1✔
241	b.channelFactory(channelFactory);	1✔
242	// For non-socket based channel, the option will be ignored.
243	b.option(SO_KEEPALIVE, true);	1✔
244	// For non-epoll based channel, the option will be ignored.
245	if (keepAliveTimeNanos != KEEPALIVE_TIME_NANOS_DISABLED) {	1✔
246	ChannelOption<Integer> tcpUserTimeout = Utils.maybeGetTcpUserTimeoutOption();	1✔
247	if (tcpUserTimeout != null) {	1✔
248	b.option(tcpUserTimeout, (int) TimeUnit.NANOSECONDS.toMillis(keepAliveTimeoutNanos));	1✔
249	}
250	}
251	for (Map.Entry<ChannelOption<?>, ?> entry : channelOptions.entrySet()) {	1✔
252	// Every entry in the map is obtained from
253	// NettyChannelBuilder#withOption(ChannelOption<T> option, T value)
254	// so it is safe to pass the key-value pair to b.option().
255	b.option((ChannelOption<Object>) entry.getKey(), entry.getValue());	1✔
256	}	1✔
257
258	ChannelHandler bufferingHandler = new WriteBufferingAndExceptionHandler(negotiationHandler);	1✔
259
260	/*
261	* We don't use a ChannelInitializer in the client bootstrap because its "initChannel" method
262	* is executed in the event loop and we need this handler to be in the pipeline immediately so
263	* that it may begin buffering writes.
264	*/
265	b.handler(bufferingHandler);	1✔
266	ChannelFuture regFuture = b.register();	1✔
267	if (regFuture.isDone() && !regFuture.isSuccess()) {	1✔
268	channel = null;	1✔
269	// Initialization has failed badly. All new streams should be made to fail.
270	Throwable t = regFuture.cause();	1✔
271	if (t == null) {	1✔
272	t = new IllegalStateException("Channel is null, but future doesn't have a cause");	×
273	}
274	statusExplainingWhyTheChannelIsNull = Utils.statusFromThrowable(t);	1✔
275	// Use a Runnable since lifecycleManager calls transportListener
276	return new Runnable() {	1✔
277	@Override
278	public void run() {
279	// NOTICE: we not are calling lifecycleManager from the event loop. But there isn't really
280	// an event loop in this case, so nothing should be accessing the lifecycleManager. We
281	// could use GlobalEventExecutor (which is what regFuture would use for notifying
282	// listeners in this case), but avoiding on-demand thread creation in an error case seems
283	// a good idea and is probably clearer threading.
284	lifecycleManager.notifyTerminated(statusExplainingWhyTheChannelIsNull);	1✔
285	}	1✔
286	};
287	}
288	channel = regFuture.channel();	1✔
289	// Start the write queue as soon as the channel is constructed
290	handler.startWriteQueue(channel);	1✔
291	// This write will have no effect, yet it will only complete once the negotiationHandler
292	// flushes any pending writes. We need it to be staged before the `connect` so that
293	// the channel can't have been closed yet, removing all handlers. This write will sit in the
294	// AbstractBufferingHandler's buffer, and will either be flushed on a successful connection,
295	// or failed if the connection fails.
296	channel.writeAndFlush(NettyClientHandler.NOOP_MESSAGE).addListener(new ChannelFutureListener() {	1✔
297	@Override
298	public void operationComplete(ChannelFuture future) throws Exception {
299	if (!future.isSuccess()) {	1✔
300	// Need to notify of this failure, because NettyClientHandler may not have been added to
301	// the pipeline before the error occurred.
302	lifecycleManager.notifyTerminated(Utils.statusFromThrowable(future.cause()));	1✔
303	}
304	}	1✔
305	});
306	// Start the connection operation to the server.
307	SocketAddress localAddress =	1✔
308	localSocketPicker.createSocketAddress(remoteAddress, eagAttributes);	1✔
309	if (localAddress != null) {	1✔
310	channel.connect(remoteAddress, localAddress);	×
311	} else {
312	channel.connect(remoteAddress);	1✔
313	}
314
315	if (keepAliveManager != null) {	1✔
316	keepAliveManager.onTransportStarted();	1✔
317	}
318
319	return null;	1✔
320	}
321
322	@Override
323	public void shutdown(Status reason) {
324	// start() could have failed
325	if (channel == null) {	1✔
326	return;	1✔
327	}
328	// Notifying of termination is automatically done when the channel closes.
329	if (channel.isOpen()) {	1✔
330	handler.getWriteQueue().enqueue(new GracefulCloseCommand(reason), true);	1✔
331	}
332	}	1✔
333
334	@Override
335	public void shutdownNow(final Status reason) {
336	// Notifying of termination is automatically done when the channel closes.
337	if (channel != null && channel.isOpen()) {	1✔
338	handler.getWriteQueue().enqueue(new Runnable() {	1✔
339	@Override
340	public void run() {
341	lifecycleManager.notifyShutdown(reason);	1✔
342	channel.write(new ForcefulCloseCommand(reason));	1✔
343	}	1✔
344	}, true);
345	}
346	}	1✔
347
348	@Override
349	public String toString() {
350	return MoreObjects.toStringHelper(this)	1✔
351	.add("logId", logId.getId())	1✔
352	.add("remoteAddress", remoteAddress)	1✔
353	.add("channel", channel)	1✔
354	.toString();	1✔
355	}
356
357	@Override
358	public InternalLogId getLogId() {
359	return logId;	1✔
360	}
361
362	@Override
363	public Attributes getAttributes() {
364	return handler.getAttributes();	1✔
365	}
366
367	@Override
368	public ListenableFuture<SocketStats> getStats() {
369	final SettableFuture<SocketStats> result = SettableFuture.create();	1✔
370	if (channel.eventLoop().inEventLoop()) {	1✔
371	// This is necessary, otherwise we will block forever if we get the future from inside
372	// the event loop.
373	result.set(getStatsHelper(channel));	×
374	return result;	×
375	}
376	channel.eventLoop().submit(	1✔
377	new Runnable() {	1✔
378	@Override
379	public void run() {
380	result.set(getStatsHelper(channel));	1✔
381	}	1✔
382	})
383	.addListener(	1✔
384	new GenericFutureListener<Future<Object>>() {	1✔
385	@Override
386	public void operationComplete(Future<Object> future) throws Exception {
387	if (!future.isSuccess()) {	1✔
388	result.setException(future.cause());	×
389	}
390	}	1✔
391	});
392	return result;	1✔
393	}
394
395	private SocketStats getStatsHelper(Channel ch) {
396	assert ch.eventLoop().inEventLoop();	1✔
397	return new SocketStats(	1✔
398	transportTracer.getStats(),	1✔
399	channel.localAddress(),	1✔
400	channel.remoteAddress(),	1✔
401	Utils.getSocketOptions(ch),	1✔
402	handler == null ? null : handler.getSecurityInfo());	1✔
403	}
404
405	@VisibleForTesting
406	Channel channel() {
407	return channel;	1✔
408	}
409
410	@VisibleForTesting
411	KeepAliveManager keepAliveManager() {
412	return keepAliveManager;	1✔
413	}
414
415	/**
416	* Convert ChannelFuture.cause() to a Status, taking into account that all handlers are removed
417	* from the pipeline when the channel is closed. Since handlers are removed, you may get an
418	* unhelpful exception like ClosedChannelException.
419	*
420	* <p>This method must only be called on the event loop.
421	*/
422	private Status statusFromFailedFuture(ChannelFuture f) {
423	Throwable t = f.cause();	1✔
424	if (t instanceof ClosedChannelException	1✔
425	// Exception thrown by the StreamBufferingEncoder if the channel is closed while there
426	// are still streams buffered. This exception is not helpful. Replace it by the real
427	// cause of the shutdown (if available).
428	\|\| t instanceof Http2ChannelClosedException) {
429	Status shutdownStatus = lifecycleManager.getShutdownStatus();	1✔
430	if (shutdownStatus == null) {	1✔
431	return Status.UNKNOWN.withDescription("Channel closed but for unknown reason")	×
432	.withCause(new ClosedChannelException().initCause(t));	×
433	}
434	return shutdownStatus;	1✔
435	}
436	return Utils.statusFromThrowable(t);	1✔
437	}
438	}

grpc / grpc-java / #19108

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