#20434

Committed 04 Sep 2026 05:20AM UTC coverage: 89.211% (+0.009%) from 89.202%

Build # #20434

Build Type

push

github

Committed by

web-flow

Commit Message

core: Reset configSelector on realChannel while entering IDLE state (#12832)

ManagedChannel will be stuck in IDLE state when xDS control plane
doesn't have a resource anymore.
The scenario is following:
1. Channel is open for a xds resource.
2. XdsNameResolver subscribes to the resource on xDS control plane. 
3. The resource is removed from xDS control plane for extended period of
time (unhealthy for more than the idle timeout on Channel)
4. Channel enters into TRANSIENT_FAILURE
5. The Idle timeout triggers and Channel shutdowns XdsNameResolver and
other resources. xDS watchers are removed.
6. The resource comes back online on xDS control plane.
7. A new GRPC call is executed targeting the channel.
8. The channel stays in IDLE state and reports:
"io.grpc.StatusRuntimeException: UNAVAILABLE: LDS resource xxxx does not
exist nodeID: yyyy" because realChannel.configSelector still points to
old state.

```    
public <ReqT, RespT> ClientCall<ReqT, RespT> newCall(
        MethodDescriptor<ReqT, RespT> method, CallOptions callOptions) {
      if (configSelector.get() != INITIAL_PENDING_SELECTOR) {
        return newClientCall(method, callOptions);
      }
...
```

Coverage Stats

38641 of 43314 relevant lines covered (89.21%)

0.89 hits per line

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

88.32

/../netty/src/main/java/io/grpc/netty/NettyServer.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 com.google.common.base.Preconditions.checkNotNull;
import static io.grpc.netty.NettyServerBuilder.MAX_CONNECTION_AGE_NANOS_DISABLED;
import static io.netty.channel.ChannelOption.ALLOCATOR;
import static io.netty.channel.ChannelOption.SO_KEEPALIVE;

import com.google.common.base.MoreObjects;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import io.grpc.Attributes;
import io.grpc.InternalChannelz;
import io.grpc.InternalChannelz.SocketStats;
import io.grpc.InternalInstrumented;
import io.grpc.InternalLogId;
import io.grpc.InternalWithLogId;
import io.grpc.MetricRecorder;
import io.grpc.ServerStreamTracer;
import io.grpc.internal.InternalServer;
import io.grpc.internal.ObjectPool;
import io.grpc.internal.ServerListener;
import io.grpc.internal.ServerTransportListener;
import io.grpc.internal.TransportTracer;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFactory;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPromise;
import io.netty.channel.EventLoop;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.ServerChannel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.ChannelGroupFuture;
import io.netty.channel.group.ChannelGroupFutureListener;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.AbstractReferenceCounted;
import io.netty.util.AsciiString;
import io.netty.util.ReferenceCounted;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import java.io.IOException;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Netty-based server implementation.
 */
class NettyServer implements InternalServer, InternalWithLogId {
  private static final Logger log = Logger.getLogger(InternalServer.class.getName());

  private final InternalLogId logId;
  private final List<? extends SocketAddress> addresses;
  private final ChannelFactory<? extends ServerChannel> channelFactory;
  private final Map<ChannelOption<?>, ?> channelOptions;
  private final Map<ChannelOption<?>, ?> childChannelOptions;
  private final ProtocolNegotiator protocolNegotiator;
  private final int maxStreamsPerConnection;
  private final ObjectPool<? extends EventLoopGroup> bossGroupPool;
  private final ObjectPool<? extends EventLoopGroup> workerGroupPool;
  private final boolean forceHeapBuffer;
  private EventLoopGroup bossGroup;
  private EventLoopGroup workerGroup;
  private ServerListener listener;
  private final ChannelGroup channelGroup;
  private final boolean autoFlowControl;
  private final int flowControlWindow;
  private final Set<AsciiString> neverIndexedMetadataKeys;
  private final int maxMessageSize;
  private final int maxHeaderListSize;
  private final int softLimitHeaderListSize;
  private MetricRecorder metricRecorder;
  private final long keepAliveTimeInNanos;
  private final long keepAliveTimeoutInNanos;
  private final long maxConnectionIdleInNanos;
  private final long maxConnectionAgeInNanos;
  private final long maxConnectionAgeGraceInNanos;
  private final boolean permitKeepAliveWithoutCalls;
  private final long permitKeepAliveTimeInNanos;
  private final int maxRstCount;
  private final long maxRstPeriodNanos;
  private final Attributes eagAttributes;
  private final ReferenceCounted sharedResourceReferenceCounter =
      new SharedResourceReferenceCounter();
  private final List<? extends ServerStreamTracer.Factory> streamTracerFactories;
  private final TransportTracer.Factory transportTracerFactory;
  private final InternalChannelz channelz;
  private volatile List<InternalInstrumented<SocketStats>> listenSocketStatsList =
      Collections.emptyList();
  private volatile boolean terminated;
  private final EventLoop bossExecutor;

  NettyServer(
      List<? extends SocketAddress> addresses,
      ChannelFactory<? extends ServerChannel> channelFactory,
      Map<ChannelOption<?>, ?> channelOptions,
      Map<ChannelOption<?>, ?> childChannelOptions,
      ObjectPool<? extends EventLoopGroup> bossGroupPool,
      ObjectPool<? extends EventLoopGroup> workerGroupPool,
      boolean forceHeapBuffer,
      ProtocolNegotiator protocolNegotiator,
      List<? extends ServerStreamTracer.Factory> streamTracerFactories,
      TransportTracer.Factory transportTracerFactory,
      int maxStreamsPerConnection,
      boolean autoFlowControl,
      int flowControlWindow,
      Set<AsciiString> neverIndexedMetadataKeys,
      int maxMessageSize,
      int maxHeaderListSize,
      int softLimitHeaderListSize,
      long keepAliveTimeInNanos,
      long keepAliveTimeoutInNanos,
      long maxConnectionIdleInNanos,
      long maxConnectionAgeInNanos, long maxConnectionAgeGraceInNanos,
      boolean permitKeepAliveWithoutCalls, long permitKeepAliveTimeInNanos,
      int maxRstCount, long maxRstPeriodNanos,
      Attributes eagAttributes, InternalChannelz channelz,
      MetricRecorder metricRecorder) {
    this.addresses = checkNotNull(addresses, "addresses");
    this.metricRecorder = metricRecorder;
    this.channelFactory = checkNotNull(channelFactory, "channelFactory");
    checkNotNull(channelOptions, "channelOptions");
    this.channelOptions = new HashMap<ChannelOption<?>, Object>(channelOptions);
    checkNotNull(childChannelOptions, "childChannelOptions");
    this.childChannelOptions = new HashMap<ChannelOption<?>, Object>(childChannelOptions);
    this.bossGroupPool = checkNotNull(bossGroupPool, "bossGroupPool");
    this.workerGroupPool = checkNotNull(workerGroupPool, "workerGroupPool");
    this.forceHeapBuffer = forceHeapBuffer;
    this.bossGroup = bossGroupPool.getObject();
    this.bossExecutor = bossGroup.next();
    this.channelGroup = new DefaultChannelGroup(this.bossExecutor);
    this.workerGroup = workerGroupPool.getObject();
    this.protocolNegotiator = checkNotNull(protocolNegotiator, "protocolNegotiator");
    this.streamTracerFactories = checkNotNull(streamTracerFactories, "streamTracerFactories");
    this.transportTracerFactory = transportTracerFactory;
    this.maxStreamsPerConnection = maxStreamsPerConnection;
    this.autoFlowControl = autoFlowControl;
    this.flowControlWindow = flowControlWindow;
    this.neverIndexedMetadataKeys = Collections.unmodifiableSet(
        new HashSet<>(checkNotNull(neverIndexedMetadataKeys, "neverIndexedMetadataKeys")));
    this.maxMessageSize = maxMessageSize;
    this.maxHeaderListSize = maxHeaderListSize;
    this.softLimitHeaderListSize = softLimitHeaderListSize;
    this.keepAliveTimeInNanos = keepAliveTimeInNanos;
    this.keepAliveTimeoutInNanos = keepAliveTimeoutInNanos;
    this.maxConnectionIdleInNanos = maxConnectionIdleInNanos;
    this.maxConnectionAgeInNanos = maxConnectionAgeInNanos;
    this.maxConnectionAgeGraceInNanos = maxConnectionAgeGraceInNanos;
    this.permitKeepAliveWithoutCalls = permitKeepAliveWithoutCalls;
    this.permitKeepAliveTimeInNanos = permitKeepAliveTimeInNanos;
    this.maxRstCount = maxRstCount;
    this.maxRstPeriodNanos = maxRstPeriodNanos;
    this.eagAttributes = checkNotNull(eagAttributes, "eagAttributes");
    this.channelz = Preconditions.checkNotNull(channelz);
    this.logId = InternalLogId.allocate(getClass(), addresses.isEmpty() ? "No address" :
        String.valueOf(addresses));
  }


  @Override
  public SocketAddress getListenSocketAddress() {
    Iterator<Channel> it = channelGroup.iterator();
    if (it.hasNext()) {
      return it.next().localAddress();
    } else {
      // server is not listening/bound yet, just return the original port.
      return addresses.isEmpty() ? null : addresses.get(0);
    }
  }

  @Override
  public List<SocketAddress> getListenSocketAddresses() {
    List<SocketAddress> listenSocketAddresses = new ArrayList<>();
    for (Channel c: channelGroup) {
      listenSocketAddresses.add(c.localAddress());
    }
    // server is not listening/bound yet, just return the original ports.
    if (listenSocketAddresses.isEmpty())  {
      listenSocketAddresses.addAll(addresses);
    }
    return listenSocketAddresses;
  }

  @Override
  public InternalInstrumented<SocketStats> getListenSocketStats() {
    List<InternalInstrumented<SocketStats>> savedListenSocketStatsList = listenSocketStatsList;
    return savedListenSocketStatsList.isEmpty() ? null : savedListenSocketStatsList.get(0);
  }

  @Override
  public List<InternalInstrumented<SocketStats>> getListenSocketStatsList() {
    return listenSocketStatsList;
  }

  @Override
  public void start(ServerListener serverListener) throws IOException {
    listener = checkNotNull(serverListener, "serverListener");

    final ServerBootstrap b = new ServerBootstrap();
    b.option(ALLOCATOR, Utils.getByteBufAllocator(forceHeapBuffer));
    b.childOption(ALLOCATOR, Utils.getByteBufAllocator(forceHeapBuffer));
    b.group(bossExecutor, workerGroup);
    b.channelFactory(channelFactory);
    // For non-socket based channel, the option will be ignored.
    b.childOption(SO_KEEPALIVE, true);

    if (channelOptions != null) {
      for (Map.Entry<ChannelOption<?>, ?> entry : channelOptions.entrySet()) {
        @SuppressWarnings("unchecked")
        ChannelOption<Object> key = (ChannelOption<Object>) entry.getKey();
        b.option(key, entry.getValue());
      }
    }

    if (childChannelOptions != null) {
      for (Map.Entry<ChannelOption<?>, ?> entry : childChannelOptions.entrySet()) {
        @SuppressWarnings("unchecked")
        ChannelOption<Object> key = (ChannelOption<Object>) entry.getKey();
        b.childOption(key, entry.getValue());
      }
    }

    b.childHandler(new ChannelInitializer<Channel>() {
      @Override
      public void initChannel(Channel ch) {

        ChannelPromise channelDone = ch.newPromise();

        long maxConnectionAgeInNanos = NettyServer.this.maxConnectionAgeInNanos;
        if (maxConnectionAgeInNanos != MAX_CONNECTION_AGE_NANOS_DISABLED) {
          // apply a random jitter of +/-10% to max connection age
          maxConnectionAgeInNanos =
              (long) ((.9D + Math.random() * .2D) * maxConnectionAgeInNanos);
        }

            NettyServerTransport transport =
                new NettyServerTransport(
                    ch,
                    channelDone,
                    protocolNegotiator,
                    streamTracerFactories,
                    transportTracerFactory.create(),
                    maxStreamsPerConnection,
                    autoFlowControl,
                    flowControlWindow,
                    neverIndexedMetadataKeys,
                    maxMessageSize,
                    maxHeaderListSize,
                    softLimitHeaderListSize,
                    keepAliveTimeInNanos,
                    keepAliveTimeoutInNanos,
                    maxConnectionIdleInNanos,
                    maxConnectionAgeInNanos,
                    maxConnectionAgeGraceInNanos,
                    permitKeepAliveWithoutCalls,
                    permitKeepAliveTimeInNanos,
                    maxRstCount,
                    maxRstPeriodNanos,
                    eagAttributes,
                    metricRecorder);
        ServerTransportListener transportListener;
        // This is to order callbacks on the listener, not to guard access to channel.
        synchronized (NettyServer.this) {
          if (terminated) {
            // Server already terminated.
            ch.close();
            return;
          }
          // `channel` shutdown can race with `ch` initialization, so this is only safe to increment
          // inside the lock.
          sharedResourceReferenceCounter.retain();
          transportListener = listener.transportCreated(transport);
        }

        /* Releases the event loop if the channel is "done", possibly due to the channel closing. */
        final class LoopReleaser implements ChannelFutureListener {
          private boolean done;

          @Override
          public void operationComplete(ChannelFuture future) throws Exception {
            if (!done) {
              done = true;
              sharedResourceReferenceCounter.release();
            }
          }
        }

        transport.start(transportListener);
        ChannelFutureListener loopReleaser = new LoopReleaser();
        channelDone.addListener(loopReleaser);
        ch.closeFuture().addListener(loopReleaser);
      }
    });
    Future<Map<ChannelFuture, SocketAddress>> bindCallFuture =
        bossExecutor.submit(
            new Callable<Map<ChannelFuture, SocketAddress>>() {
          @Override
          public Map<ChannelFuture, SocketAddress> call() {
            Map<ChannelFuture, SocketAddress> bindFutures = new HashMap<>();
            for (SocketAddress address: addresses) {
                ChannelFuture future = b.bind(address);
                channelGroup.add(future.channel());
                bindFutures.put(future, address);
            }
            return bindFutures;
          }
        }
    );
    Map<ChannelFuture, SocketAddress> channelFutures =
        bindCallFuture.awaitUninterruptibly().getNow();

    if (!bindCallFuture.isSuccess()) {
      channelGroup.close().awaitUninterruptibly();
      throw new IOException(String.format("Failed to bind to addresses %s",
          addresses), bindCallFuture.cause());
    }
    final List<InternalInstrumented<SocketStats>> socketStats = new ArrayList<>();
    for (Map.Entry<ChannelFuture, SocketAddress> entry: channelFutures.entrySet()) {
      // We'd love to observe interruption, but if interrupted we will need to close the channel,
      // which itself would need an await() to guarantee the port is not used when the method
      // returns. See #6850
      final ChannelFuture future = entry.getKey();
      if (!future.awaitUninterruptibly().isSuccess()) {
        channelGroup.close().awaitUninterruptibly();
        throw new IOException(String.format("Failed to bind to address %s",
            entry.getValue()), future.cause());
      }
      final InternalInstrumented<SocketStats> listenSocketStats =
          new ListenSocket(future.channel());
      channelz.addListenSocket(listenSocketStats);
      socketStats.add(listenSocketStats);
      future.channel().closeFuture().addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
          channelz.removeListenSocket(listenSocketStats);
        }
      });
    }
    listenSocketStatsList = Collections.unmodifiableList(socketStats);
  }

  @Override
  public void shutdown() {
    if (terminated) {
      return;
    }
    ChannelGroupFuture groupFuture = channelGroup.close()
        .addListener(new ChannelGroupFutureListener() {
            @Override
            public void operationComplete(ChannelGroupFuture future) throws Exception {
              if (!future.isSuccess()) {
                log.log(Level.WARNING, "Error closing server channel group", future.cause());
              }
              sharedResourceReferenceCounter.release();
              protocolNegotiator.close();
              listenSocketStatsList = Collections.emptyList();
              synchronized (NettyServer.this) {
                listener.serverShutdown();
                terminated = true;
              }
            }
        });
    try {
      groupFuture.await();
    } catch (InterruptedException e) {
      log.log(Level.FINE, "Interrupted while shutting down", e);
      Thread.currentThread().interrupt();
    }
  }

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

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

  class SharedResourceReferenceCounter extends AbstractReferenceCounted {
    @Override
    protected void deallocate() {
      try {
        if (bossGroup != null) {
          bossGroupPool.returnObject(bossGroup);
        }
      } finally {
        bossGroup = null;
        try {
          if (workerGroup != null) {
            workerGroupPool.returnObject(workerGroup);
          }
        } finally {
          workerGroup = null;
        }
      }
    }

    @Override
    public ReferenceCounted touch(Object hint) {
      return this;
    }
  }

  /**
   * A class that can answer channelz queries about the server listen sockets.
   */
  private static final class ListenSocket implements InternalInstrumented<SocketStats> {
    private final InternalLogId id;
    private final Channel ch;

    ListenSocket(Channel ch) {
      this.ch = ch;
      this.id = InternalLogId.allocate(getClass(), String.valueOf(ch.localAddress()));
    }

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

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

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

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 com.google.common.base.Preconditions.checkNotNull;
20	import static io.grpc.netty.NettyServerBuilder.MAX_CONNECTION_AGE_NANOS_DISABLED;
21	import static io.netty.channel.ChannelOption.ALLOCATOR;
22	import static io.netty.channel.ChannelOption.SO_KEEPALIVE;
23
24	import com.google.common.base.MoreObjects;
25	import com.google.common.base.Preconditions;
26	import com.google.common.util.concurrent.ListenableFuture;
27	import com.google.common.util.concurrent.SettableFuture;
28	import io.grpc.Attributes;
29	import io.grpc.InternalChannelz;
30	import io.grpc.InternalChannelz.SocketStats;
31	import io.grpc.InternalInstrumented;
32	import io.grpc.InternalLogId;
33	import io.grpc.InternalWithLogId;
34	import io.grpc.MetricRecorder;
35	import io.grpc.ServerStreamTracer;
36	import io.grpc.internal.InternalServer;
37	import io.grpc.internal.ObjectPool;
38	import io.grpc.internal.ServerListener;
39	import io.grpc.internal.ServerTransportListener;
40	import io.grpc.internal.TransportTracer;
41	import io.netty.bootstrap.ServerBootstrap;
42	import io.netty.channel.Channel;
43	import io.netty.channel.ChannelFactory;
44	import io.netty.channel.ChannelFuture;
45	import io.netty.channel.ChannelFutureListener;
46	import io.netty.channel.ChannelInitializer;
47	import io.netty.channel.ChannelOption;
48	import io.netty.channel.ChannelPromise;
49	import io.netty.channel.EventLoop;
50	import io.netty.channel.EventLoopGroup;
51	import io.netty.channel.ServerChannel;
52	import io.netty.channel.group.ChannelGroup;
53	import io.netty.channel.group.ChannelGroupFuture;
54	import io.netty.channel.group.ChannelGroupFutureListener;
55	import io.netty.channel.group.DefaultChannelGroup;
56	import io.netty.util.AbstractReferenceCounted;
57	import io.netty.util.AsciiString;
58	import io.netty.util.ReferenceCounted;
59	import io.netty.util.concurrent.Future;
60	import io.netty.util.concurrent.GenericFutureListener;
61	import java.io.IOException;
62	import java.net.SocketAddress;
63	import java.util.ArrayList;
64	import java.util.Collections;
65	import java.util.HashMap;
66	import java.util.HashSet;
67	import java.util.Iterator;
68	import java.util.List;
69	import java.util.Map;
70	import java.util.Set;
71	import java.util.concurrent.Callable;
72	import java.util.logging.Level;
73	import java.util.logging.Logger;
74
75	/**
76	* Netty-based server implementation.
77	*/
78	class NettyServer implements InternalServer, InternalWithLogId {
79	private static final Logger log = Logger.getLogger(InternalServer.class.getName());	1✔
80
81	private final InternalLogId logId;
82	private final List<? extends SocketAddress> addresses;
83	private final ChannelFactory<? extends ServerChannel> channelFactory;
84	private final Map<ChannelOption<?>, ?> channelOptions;
85	private final Map<ChannelOption<?>, ?> childChannelOptions;
86	private final ProtocolNegotiator protocolNegotiator;
87	private final int maxStreamsPerConnection;
88	private final ObjectPool<? extends EventLoopGroup> bossGroupPool;
89	private final ObjectPool<? extends EventLoopGroup> workerGroupPool;
90	private final boolean forceHeapBuffer;
91	private EventLoopGroup bossGroup;
92	private EventLoopGroup workerGroup;
93	private ServerListener listener;
94	private final ChannelGroup channelGroup;
95	private final boolean autoFlowControl;
96	private final int flowControlWindow;
97	private final Set<AsciiString> neverIndexedMetadataKeys;
98	private final int maxMessageSize;
99	private final int maxHeaderListSize;
100	private final int softLimitHeaderListSize;
101	private MetricRecorder metricRecorder;
102	private final long keepAliveTimeInNanos;
103	private final long keepAliveTimeoutInNanos;
104	private final long maxConnectionIdleInNanos;
105	private final long maxConnectionAgeInNanos;
106	private final long maxConnectionAgeGraceInNanos;
107	private final boolean permitKeepAliveWithoutCalls;
108	private final long permitKeepAliveTimeInNanos;
109	private final int maxRstCount;
110	private final long maxRstPeriodNanos;
111	private final Attributes eagAttributes;
112	private final ReferenceCounted sharedResourceReferenceCounter =	1✔
113	new SharedResourceReferenceCounter();
114	private final List<? extends ServerStreamTracer.Factory> streamTracerFactories;
115	private final TransportTracer.Factory transportTracerFactory;
116	private final InternalChannelz channelz;
117	private volatile List<InternalInstrumented<SocketStats>> listenSocketStatsList =	1✔
118	Collections.emptyList();	1✔
119	private volatile boolean terminated;
120	private final EventLoop bossExecutor;
121
122	NettyServer(
123	List<? extends SocketAddress> addresses,
124	ChannelFactory<? extends ServerChannel> channelFactory,
125	Map<ChannelOption<?>, ?> channelOptions,
126	Map<ChannelOption<?>, ?> childChannelOptions,
127	ObjectPool<? extends EventLoopGroup> bossGroupPool,
128	ObjectPool<? extends EventLoopGroup> workerGroupPool,
129	boolean forceHeapBuffer,
130	ProtocolNegotiator protocolNegotiator,
131	List<? extends ServerStreamTracer.Factory> streamTracerFactories,
132	TransportTracer.Factory transportTracerFactory,
133	int maxStreamsPerConnection,
134	boolean autoFlowControl,
135	int flowControlWindow,
136	Set<AsciiString> neverIndexedMetadataKeys,
137	int maxMessageSize,
138	int maxHeaderListSize,
139	int softLimitHeaderListSize,
140	long keepAliveTimeInNanos,
141	long keepAliveTimeoutInNanos,
142	long maxConnectionIdleInNanos,
143	long maxConnectionAgeInNanos, long maxConnectionAgeGraceInNanos,
144	boolean permitKeepAliveWithoutCalls, long permitKeepAliveTimeInNanos,
145	int maxRstCount, long maxRstPeriodNanos,
146	Attributes eagAttributes, InternalChannelz channelz,
147	MetricRecorder metricRecorder) {	1✔
148	this.addresses = checkNotNull(addresses, "addresses");	1✔
149	this.metricRecorder = metricRecorder;	1✔
150	this.channelFactory = checkNotNull(channelFactory, "channelFactory");	1✔
151	checkNotNull(channelOptions, "channelOptions");	1✔
152	this.channelOptions = new HashMap<ChannelOption<?>, Object>(channelOptions);	1✔
153	checkNotNull(childChannelOptions, "childChannelOptions");	1✔
154	this.childChannelOptions = new HashMap<ChannelOption<?>, Object>(childChannelOptions);	1✔
155	this.bossGroupPool = checkNotNull(bossGroupPool, "bossGroupPool");	1✔
156	this.workerGroupPool = checkNotNull(workerGroupPool, "workerGroupPool");	1✔
157	this.forceHeapBuffer = forceHeapBuffer;	1✔
158	this.bossGroup = bossGroupPool.getObject();	1✔
159	this.bossExecutor = bossGroup.next();	1✔
160	this.channelGroup = new DefaultChannelGroup(this.bossExecutor);	1✔
161	this.workerGroup = workerGroupPool.getObject();	1✔
162	this.protocolNegotiator = checkNotNull(protocolNegotiator, "protocolNegotiator");	1✔
163	this.streamTracerFactories = checkNotNull(streamTracerFactories, "streamTracerFactories");	1✔
164	this.transportTracerFactory = transportTracerFactory;	1✔
165	this.maxStreamsPerConnection = maxStreamsPerConnection;	1✔
166	this.autoFlowControl = autoFlowControl;	1✔
167	this.flowControlWindow = flowControlWindow;	1✔
168	this.neverIndexedMetadataKeys = Collections.unmodifiableSet(	1✔
169	new HashSet<>(checkNotNull(neverIndexedMetadataKeys, "neverIndexedMetadataKeys")));	1✔
170	this.maxMessageSize = maxMessageSize;	1✔
171	this.maxHeaderListSize = maxHeaderListSize;	1✔
172	this.softLimitHeaderListSize = softLimitHeaderListSize;	1✔
173	this.keepAliveTimeInNanos = keepAliveTimeInNanos;	1✔
174	this.keepAliveTimeoutInNanos = keepAliveTimeoutInNanos;	1✔
175	this.maxConnectionIdleInNanos = maxConnectionIdleInNanos;	1✔
176	this.maxConnectionAgeInNanos = maxConnectionAgeInNanos;	1✔
177	this.maxConnectionAgeGraceInNanos = maxConnectionAgeGraceInNanos;	1✔
178	this.permitKeepAliveWithoutCalls = permitKeepAliveWithoutCalls;	1✔
179	this.permitKeepAliveTimeInNanos = permitKeepAliveTimeInNanos;	1✔
180	this.maxRstCount = maxRstCount;	1✔
181	this.maxRstPeriodNanos = maxRstPeriodNanos;	1✔
182	this.eagAttributes = checkNotNull(eagAttributes, "eagAttributes");	1✔
183	this.channelz = Preconditions.checkNotNull(channelz);	1✔
184	this.logId = InternalLogId.allocate(getClass(), addresses.isEmpty() ? "No address" :	1✔
185	String.valueOf(addresses));	1✔
186	}	1✔
187
188
189	@Override
190	public SocketAddress getListenSocketAddress() {
191	Iterator<Channel> it = channelGroup.iterator();	1✔
192	if (it.hasNext()) {	1✔
193	return it.next().localAddress();	1✔
194	} else {
195	// server is not listening/bound yet, just return the original port.
196	return addresses.isEmpty() ? null : addresses.get(0);	1✔
197	}
198	}
199
200	@Override
201	public List<SocketAddress> getListenSocketAddresses() {
202	List<SocketAddress> listenSocketAddresses = new ArrayList<>();	1✔
203	for (Channel c: channelGroup) {	1✔
204	listenSocketAddresses.add(c.localAddress());	1✔
205	}	1✔
206	// server is not listening/bound yet, just return the original ports.
207	if (listenSocketAddresses.isEmpty()) {	1✔
208	listenSocketAddresses.addAll(addresses);	1✔
209	}
210	return listenSocketAddresses;	1✔
211	}
212
213	@Override
214	public InternalInstrumented<SocketStats> getListenSocketStats() {
215	List<InternalInstrumented<SocketStats>> savedListenSocketStatsList = listenSocketStatsList;	1✔
216	return savedListenSocketStatsList.isEmpty() ? null : savedListenSocketStatsList.get(0);	1✔
217	}
218
219	@Override
220	public List<InternalInstrumented<SocketStats>> getListenSocketStatsList() {
221	return listenSocketStatsList;	1✔
222	}
223
224	@Override
225	public void start(ServerListener serverListener) throws IOException {
226	listener = checkNotNull(serverListener, "serverListener");	1✔
227
228	final ServerBootstrap b = new ServerBootstrap();	1✔
229	b.option(ALLOCATOR, Utils.getByteBufAllocator(forceHeapBuffer));	1✔
230	b.childOption(ALLOCATOR, Utils.getByteBufAllocator(forceHeapBuffer));	1✔
231	b.group(bossExecutor, workerGroup);	1✔
232	b.channelFactory(channelFactory);	1✔
233	// For non-socket based channel, the option will be ignored.
234	b.childOption(SO_KEEPALIVE, true);	1✔
235
236	if (channelOptions != null) {	1✔
237	for (Map.Entry<ChannelOption<?>, ?> entry : channelOptions.entrySet()) {	1✔
238	@SuppressWarnings("unchecked")
239	ChannelOption<Object> key = (ChannelOption<Object>) entry.getKey();	×
240	b.option(key, entry.getValue());	×
241	}	×
242	}
243
244	if (childChannelOptions != null) {	1✔
245	for (Map.Entry<ChannelOption<?>, ?> entry : childChannelOptions.entrySet()) {	1✔
246	@SuppressWarnings("unchecked")
247	ChannelOption<Object> key = (ChannelOption<Object>) entry.getKey();	1✔
248	b.childOption(key, entry.getValue());	1✔
249	}	1✔
250	}
251
252	b.childHandler(new ChannelInitializer<Channel>() {	1✔
253	@Override
254	public void initChannel(Channel ch) {
255
256	ChannelPromise channelDone = ch.newPromise();	1✔
257
258	long maxConnectionAgeInNanos = NettyServer.this.maxConnectionAgeInNanos;	1✔
259	if (maxConnectionAgeInNanos != MAX_CONNECTION_AGE_NANOS_DISABLED) {	1✔
260	// apply a random jitter of +/-10% to max connection age
261	maxConnectionAgeInNanos =	1✔
262	(long) ((.9D + Math.random() * .2D) * maxConnectionAgeInNanos);	1✔
263	}
264
265	NettyServerTransport transport =	1✔
266	new NettyServerTransport(
267	ch,
268	channelDone,
269	protocolNegotiator,	1✔
270	streamTracerFactories,	1✔
271	transportTracerFactory.create(),	1✔
272	maxStreamsPerConnection,	1✔
273	autoFlowControl,	1✔
274	flowControlWindow,	1✔
275	neverIndexedMetadataKeys,	1✔
276	maxMessageSize,	1✔
277	maxHeaderListSize,	1✔
278	softLimitHeaderListSize,	1✔
279	keepAliveTimeInNanos,	1✔
280	keepAliveTimeoutInNanos,	1✔
281	maxConnectionIdleInNanos,	1✔
282	maxConnectionAgeInNanos,
283	maxConnectionAgeGraceInNanos,	1✔
284	permitKeepAliveWithoutCalls,	1✔
285	permitKeepAliveTimeInNanos,	1✔
286	maxRstCount,	1✔
287	maxRstPeriodNanos,	1✔
288	eagAttributes,	1✔
289	metricRecorder);	1✔
290	ServerTransportListener transportListener;
291	// This is to order callbacks on the listener, not to guard access to channel.
292	synchronized (NettyServer.this) {	1✔
293	if (terminated) {	1✔
294	// Server already terminated.
295	ch.close();	×
296	return;	×
297	}
298	// `channel` shutdown can race with `ch` initialization, so this is only safe to increment
299	// inside the lock.
300	sharedResourceReferenceCounter.retain();	1✔
301	transportListener = listener.transportCreated(transport);	1✔
302	}	1✔
303
304	/* Releases the event loop if the channel is "done", possibly due to the channel closing. */
305	final class LoopReleaser implements ChannelFutureListener {	1✔
306	private boolean done;
307
308	@Override
309	public void operationComplete(ChannelFuture future) throws Exception {
310	if (!done) {	1✔
311	done = true;	1✔
312	sharedResourceReferenceCounter.release();	1✔
313	}
314	}	1✔
315	}
316
317	transport.start(transportListener);	1✔
318	ChannelFutureListener loopReleaser = new LoopReleaser();	1✔
319	channelDone.addListener(loopReleaser);	1✔
320	ch.closeFuture().addListener(loopReleaser);	1✔
321	}	1✔
322	});
323	Future<Map<ChannelFuture, SocketAddress>> bindCallFuture =	1✔
324	bossExecutor.submit(	1✔
325	new Callable<Map<ChannelFuture, SocketAddress>>() {	1✔
326	@Override
327	public Map<ChannelFuture, SocketAddress> call() {
328	Map<ChannelFuture, SocketAddress> bindFutures = new HashMap<>();	1✔
329	for (SocketAddress address: addresses) {	1✔
330	ChannelFuture future = b.bind(address);	1✔
331	channelGroup.add(future.channel());	1✔
332	bindFutures.put(future, address);	1✔
333	}	1✔
334	return bindFutures;	1✔
335	}
336	}
337	);
338	Map<ChannelFuture, SocketAddress> channelFutures =	1✔
339	bindCallFuture.awaitUninterruptibly().getNow();	1✔
340
341	if (!bindCallFuture.isSuccess()) {	1✔
342	channelGroup.close().awaitUninterruptibly();	1✔
343	throw new IOException(String.format("Failed to bind to addresses %s",	1✔
344	addresses), bindCallFuture.cause());	1✔
345	}
346	final List<InternalInstrumented<SocketStats>> socketStats = new ArrayList<>();	1✔
347	for (Map.Entry<ChannelFuture, SocketAddress> entry: channelFutures.entrySet()) {	1✔
348	// We'd love to observe interruption, but if interrupted we will need to close the channel,
349	// which itself would need an await() to guarantee the port is not used when the method
350	// returns. See #6850
351	final ChannelFuture future = entry.getKey();	1✔
352	if (!future.awaitUninterruptibly().isSuccess()) {	1✔
353	channelGroup.close().awaitUninterruptibly();	1✔
354	throw new IOException(String.format("Failed to bind to address %s",	1✔
355	entry.getValue()), future.cause());	1✔
356	}
357	final InternalInstrumented<SocketStats> listenSocketStats =	1✔
358	new ListenSocket(future.channel());	1✔
359	channelz.addListenSocket(listenSocketStats);	1✔
360	socketStats.add(listenSocketStats);	1✔
361	future.channel().closeFuture().addListener(new ChannelFutureListener() {	1✔
362	@Override
363	public void operationComplete(ChannelFuture future) throws Exception {
364	channelz.removeListenSocket(listenSocketStats);	1✔
365	}	1✔
366	});
367	}	1✔
368	listenSocketStatsList = Collections.unmodifiableList(socketStats);	1✔
369	}	1✔
370
371	@Override
372	public void shutdown() {
373	if (terminated) {	1✔
374	return;	×
375	}
376	ChannelGroupFuture groupFuture = channelGroup.close()	1✔
377	.addListener(new ChannelGroupFutureListener() {	1✔
378	@Override
379	public void operationComplete(ChannelGroupFuture future) throws Exception {
380	if (!future.isSuccess()) {	1✔
381	log.log(Level.WARNING, "Error closing server channel group", future.cause());	×
382	}
383	sharedResourceReferenceCounter.release();	1✔
384	protocolNegotiator.close();	1✔
385	listenSocketStatsList = Collections.emptyList();	1✔
386	synchronized (NettyServer.this) {	1✔
387	listener.serverShutdown();	1✔
388	terminated = true;	1✔
389	}	1✔
390	}	1✔
391	});
392	try {
393	groupFuture.await();	1✔
394	} catch (InterruptedException e) {	×
395	log.log(Level.FINE, "Interrupted while shutting down", e);	×
396	Thread.currentThread().interrupt();	×
397	}	1✔
398	}	1✔
399
400	@Override
401	public InternalLogId getLogId() {
402	return logId;	×
403	}
404
405	@Override
406	public String toString() {
407	return MoreObjects.toStringHelper(this)	×
408	.add("logId", logId.getId())	×
409	.add("addresses", addresses)	×
410	.toString();	×
411	}
412
413	class SharedResourceReferenceCounter extends AbstractReferenceCounted {	1✔
414	@Override
415	protected void deallocate() {
416	try {
417	if (bossGroup != null) {	1✔
418	bossGroupPool.returnObject(bossGroup);	1✔
419	}
420	} finally {
421	bossGroup = null;	1✔
422	try {
423	if (workerGroup != null) {	1✔
424	workerGroupPool.returnObject(workerGroup);	1✔
425	}
426	} finally {
427	workerGroup = null;	1✔
428	}
429	}
430	}	1✔
431
432	@Override
433	public ReferenceCounted touch(Object hint) {
434	return this;	×
435	}
436	}
437
438	/**
439	* A class that can answer channelz queries about the server listen sockets.
440	*/
441	private static final class ListenSocket implements InternalInstrumented<SocketStats> {
442	private final InternalLogId id;
443	private final Channel ch;
444
445	ListenSocket(Channel ch) {	1✔
446	this.ch = ch;	1✔
447	this.id = InternalLogId.allocate(getClass(), String.valueOf(ch.localAddress()));	1✔
448	}	1✔
449
450	@Override
451	public ListenableFuture<SocketStats> getStats() {
452	final SettableFuture<SocketStats> ret = SettableFuture.create();	1✔
453	if (ch.eventLoop().inEventLoop()) {	1✔
454	// This is necessary, otherwise we will block forever if we get the future from inside
455	// the event loop.
456	ret.set(new SocketStats(	×
457	/data=/ null,
458	ch.localAddress(),	×
459	/remote=/ null,
460	Utils.getSocketOptions(ch),	×
461	/security=/ null));
462	return ret;	×
463	}
464	ch.eventLoop()	1✔
465	.submit(	1✔
466	new Runnable() {	1✔
467	@Override
468	public void run() {
469	ret.set(new SocketStats(	1✔
470	/data=/ null,
471	ch.localAddress(),	1✔
472	/remote=/ null,
473	Utils.getSocketOptions(ch),	1✔
474	/security=/ null));
475	}	1✔
476	})
477	.addListener(	1✔
478	new GenericFutureListener<Future<Object>>() {	1✔
479	@Override
480	public void operationComplete(Future<Object> future) throws Exception {
481	if (!future.isSuccess()) {	1✔
482	ret.setException(future.cause());	×
483	}
484	}	1✔
485	});
486	return ret;	1✔
487	}
488
489	@Override
490	public InternalLogId getLogId() {
491	return id;	1✔
492	}
493
494	@Override
495	public String toString() {
496	return MoreObjects.toStringHelper(this)	×
497	.add("logId", id.getId())	×
498	.add("channel", ch)	×
499	.toString();	×
500	}
501	}
502	}

grpc / grpc-java / #20434

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