#18733

pending completion

Build # #18733

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Upgrade dependencies post v1.57.x branch cut (#10359)

### Dependency updates

#### Update successfully to the latest
- [x] `androidx.core:core 1.10.0 -> 1.10.1`
- [x] `com.google.api.grpc:proto-google-common-protos 2.17.0 -> 2.22.0`
- [x] `com.google.cloud:google-cloud-logging 3.14.5 -> 3.15.5`
- [x] `com.google.truth:truth 1.0.1 -> 1.1.5`
- [x] `com.puppycrawl.tools:checkstyle 8.28 -> 10.12.1`
- [x] `org.robolectric:robolectric 4.9.2 -> 4.10.3`
- [x] Auto-value
   - [x] `com.google.auto.value:auto-value 1.10.1 -> 1.10.2`
   - [x] `com.google.auto.value:auto-value-annotations 1.10.1 -> 1.10.2`
- [x] Protobuf
   - [x] `com.google.protobuf:protobuf-java 3.22.3 -> 3.23.4`
   - [x] `com.google.protobuf:protobuf-java-util 3.22.3 -> 3.23.4`
   - [x] `com.google.protobuf:protobuf-javalite 3.22.3 -> 3.23.4`
   - [x] `com.google.protobuf:protoc 3.22.3 -> 3.23.4`
- [x] Errorprone
   - [x] `com.google.errorprone:error_prone_annotations 2.18.0 -> 2.20.0`
   - [x] `com.google.errorprone:error_prone_core 2.18.0 -> 2.20.0`
   - ~`libs.checkstylejava8 = com.puppycrawl.tools:checkstyle 9.3 -> 10.12.1`~ -- pinned to last version supporting java8, update not needed

#### Updated to non-latest
- [x] `com.squareup.okio:okio 1.17.5 ->` ~`3.4.0`~ `2.10.0` -- updating to 3.x failed due to them [introducing gradle multiplatform artifacts](https://github.com/square/okio/blob/master/CHANGELOG.md#version-320). Error in [the comment below](https://github.com/grpc/grpc-java/pull/10359#issuecomment-1632853307.).
- [x] `org.checkerframework:checker-qual 3.33.0 ->` ~`3.36.0`~ -- removed, no longer needed
- [x] Mockito - updated to `4.11.0`. Versions 5.x break some tests. Errors in [the comment below](https://github.com/grpc/grpc-java/pull/10359#issuecomment-1632834435).
   - `org.mockito:mockito-android 3.12.4 ->` ~`5.4.0`~ `4.11.0`
   - `org.mockito:mockito-core 3.12.4 ->` ~`5.4.0`~ `4.11.0`

#### Not updated
- Cronet -- upgrade failed, created a block... (continued)

Run Details

29160 of 33046 relevant lines covered (88.24%)

0.88 hits per line

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

88.67

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

grpc / grpc-java / #18733

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