netty: support listening on multiple port

Author: carl-mastrangelo

core/src/test/java/io/grpc/inprocess/InProcessTransportTest.java

@@ -44,7 +44,7 @@ protected InternalServer newServer(List<ServerStreamTracer.Factory> streamTracer
 
   @Override
   protected InternalServer newServer(
-      InternalServer server, List<ServerStreamTracer.Factory> streamTracerFactories) {
+      int port, List<ServerStreamTracer.Factory> streamTracerFactories) {
     return newServer(streamTracerFactories);
   }
 

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

@@ -23,7 +23,6 @@
 
 import com.google.common.base.MoreObjects;
 import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
 import io.grpc.InternalChannelz;
@@ -54,9 +53,14 @@
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import javax.annotation.Nullable;
@@ -68,7 +72,7 @@ class NettyServer implements InternalServer, InternalWithLogId {
   private static final Logger log = Logger.getLogger(InternalServer.class.getName());
 
   private final InternalLogId logId;
-  private final SocketAddress address;
+  private final List<SocketAddress> requestedListenAddresses;
   private final Class<? extends ServerChannel> channelType;
   private final Map<ChannelOption<?>, ?> channelOptions;
   private final ProtocolNegotiator protocolNegotiator;
@@ -78,7 +82,7 @@ class NettyServer implements InternalServer, InternalWithLogId {
   private EventLoopGroup bossGroup;
   private EventLoopGroup workerGroup;
   private ServerListener listener;
-  private Channel channel;
+  private List<Channel> parentChannels = new ArrayList<>();
   private final int flowControlWindow;
   private final int maxMessageSize;
   private final int maxHeaderListSize;
@@ -93,13 +97,13 @@ class NettyServer implements InternalServer, InternalWithLogId {
   private final List<? extends ServerStreamTracer.Factory> streamTracerFactories;
   private final TransportTracer.Factory transportTracerFactory;
   private final InternalChannelz channelz;
-  // Only modified in event loop but safe to read any time. Set at startup and unset at shutdown.
+  // Only modified in event loop but safe to read any time. Set at startup and cleared at shutdown.
   // In the future we may have >1 listen socket.
-  private volatile ImmutableList<InternalInstrumented<SocketStats>> listenSockets
-      = ImmutableList.of();
+  private final List<ListenSocket> listenSocketStats = new CopyOnWriteArrayList<>();
+  private final List<SocketAddress> boundListenAddresses = new CopyOnWriteArrayList<>();
 
   NettyServer(
-      SocketAddress address, Class<? extends ServerChannel> channelType,
+      List<SocketAddress> listenAddresses, Class<? extends ServerChannel> channelType,
       Map<ChannelOption<?>, ?> channelOptions,
       @Nullable EventLoopGroup bossGroup, @Nullable EventLoopGroup workerGroup,
       ProtocolNegotiator protocolNegotiator,
@@ -111,7 +115,8 @@ class NettyServer implements InternalServer, InternalWithLogId {
       long maxConnectionAgeInNanos, long maxConnectionAgeGraceInNanos,
       boolean permitKeepAliveWithoutCalls, long permitKeepAliveTimeInNanos,
       InternalChannelz channelz) {
-    this.address = address;
+    this.requestedListenAddresses =
+        Collections.unmodifiableList(checkNotNull(listenAddresses, "listenAddresses"));
     this.channelType = checkNotNull(channelType, "channelType");
     checkNotNull(channelOptions, "channelOptions");
     this.channelOptions = new HashMap<ChannelOption<?>, Object>(channelOptions);
@@ -134,63 +139,51 @@ class NettyServer implements InternalServer, InternalWithLogId {
     this.permitKeepAliveWithoutCalls = permitKeepAliveWithoutCalls;
     this.permitKeepAliveTimeInNanos = permitKeepAliveTimeInNanos;
     this.channelz = Preconditions.checkNotNull(channelz);
-    this.logId =
-        InternalLogId.allocate(getClass(), address != null ? address.toString() : "No address");
+    this.logId = InternalLogId.allocate(getClass(), requestedListenAddresses.toString());
   }
 
   @Override
   public int getPort() {
-    if (channel == null) {
-      return -1;
-    }
-    SocketAddress localAddr = channel.localAddress();
-    if (!(localAddr instanceof InetSocketAddress)) {
-      return -1;
+    for (SocketAddress boundAddress : boundListenAddresses) {
+      if (boundAddress instanceof InetSocketAddress) {
+        return ((InetSocketAddress) boundAddress).getPort();
+      }
     }
-    return ((InetSocketAddress) localAddr).getPort();
+    return -1;
+  }
+
+  public List<SocketAddress> getListenAddresses() {
+    return Collections.unmodifiableList(new ArrayList<>(boundListenAddresses));
   }
 
   @Override
   public List<InternalInstrumented<SocketStats>> getListenSockets() {
-    return listenSockets;
+    return Collections.unmodifiableList(
+        new ArrayList<InternalInstrumented<SocketStats>>(listenSocketStats));
   }
 
   @Override
   public void start(ServerListener serverListener) throws IOException {
     listener = checkNotNull(serverListener, "serverListener");
-
     // If using the shared groups, get references to them.
     allocateSharedGroups();
 
-    ServerBootstrap b = new ServerBootstrap();
-    b.group(bossGroup, workerGroup);
-    b.channel(channelType);
-    if (NioServerSocketChannel.class.isAssignableFrom(channelType)) {
-      b.option(SO_BACKLOG, 128);
-      b.childOption(SO_KEEPALIVE, true);
-    }
+    final class ServerInit extends ChannelInitializer<Channel> {
+      private final AtomicReference<Channel> parentChannelRef;
 
-    if (channelOptions != null) {
-      for (Map.Entry<ChannelOption<?>, ?> entry : channelOptions.entrySet()) {
-        @SuppressWarnings("unchecked")
-        ChannelOption<Object> key = (ChannelOption<Object>) entry.getKey();
-        b.childOption(key, entry.getValue());
+      ServerInit(AtomicReference<Channel> parentChannelRef) {
+        this.parentChannelRef = parentChannelRef;
       }
-    }
 
-    b.childHandler(new ChannelInitializer<Channel>() {
       @Override
-      public void initChannel(Channel ch) throws Exception {
-
+      protected void initChannel(Channel ch) throws Exception {
         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,
@@ -209,10 +202,12 @@ public void initChannel(Channel ch) throws Exception {
                 maxConnectionAgeGraceInNanos,
                 permitKeepAliveWithoutCalls,
                 permitKeepAliveTimeInNanos);
+
         ServerTransportListener transportListener;
         // This is to order callbacks on the listener, not to guard access to channel.
         synchronized (NettyServer.this) {
-          if (channel != null && !channel.isOpen()) {
+          Channel localParentChannel = parentChannelRef.get();
+          if (localParentChannel == null || !localParentChannel.isOpen()) {
             // Server already shutdown.
             ch.close();
             return;
@@ -223,76 +218,120 @@ public void initChannel(Channel ch) throws Exception {
           transportListener = listener.transportCreated(transport);
         }
 
-        /**
-         * Releases the event loop if the channel is "done", possibly due to the channel closing.
-         */
-        final class LoopReleaser implements ChannelFutureListener {
-          boolean done;
-
-          @Override
-          public void operationComplete(ChannelFuture future) throws Exception {
-            if (!done) {
-              done = true;
-              eventLoopReferenceCounter.release();
-            }
-          }
-        }
-
         transport.start(transportListener);
         ChannelFutureListener loopReleaser = new LoopReleaser();
         channelDone.addListener(loopReleaser);
         ch.closeFuture().addListener(loopReleaser);
       }
-    });
-    // Bind and start to accept incoming connections.
-    ChannelFuture future = b.bind(address);
-    try {
-      future.await();
-    } catch (InterruptedException ex) {
-      Thread.currentThread().interrupt();
-      throw new RuntimeException("Interrupted waiting for bind");
-    }
-    if (!future.isSuccess()) {
-      throw new IOException("Failed to bind", future.cause());
+
+      /**
+       * Releases the event loop if the channel is "done", possibly due to the channel closing.
+       */
+      final class LoopReleaser implements ChannelFutureListener {
+        boolean done;
+
+        @Override
+        public void operationComplete(ChannelFuture future) throws Exception {
+          if (!done) {
+            done = true;
+            eventLoopReferenceCounter.release();
+          }
+        }
+      }
     }
-    channel = future.channel();
-    Future<?> channelzFuture = channel.eventLoop().submit(new Runnable() {
-      @Override
-      public void run() {
-        InternalInstrumented<SocketStats> listenSocket = new ListenSocket(channel);
-        listenSockets = ImmutableList.of(listenSocket);
-        channelz.addListenSocket(listenSocket);
+
+    for (SocketAddress requestedListenAddress : requestedListenAddresses) {
+      ServerBootstrap b = new ServerBootstrap();
+      b.group(bossGroup, workerGroup);
+      b.channel(channelType);
+      if (NioServerSocketChannel.class.isAssignableFrom(channelType)) {
+        b.option(SO_BACKLOG, 128);
+        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.childOption(key, entry.getValue());
+        }
+      }
+      AtomicReference<Channel> parentChannelRef = new AtomicReference<>();
+      b.childHandler(new ServerInit(parentChannelRef));
+      // Bind and start to accept incoming connections.
+      ChannelFuture future = b.bind(requestedListenAddress);
+      try {
+        future.await();
+      } catch (InterruptedException ex) {
+        Thread.currentThread().interrupt();
+        throw new RuntimeException("Interrupted waiting for bind", ex);
+      }
+      if (!future.isSuccess()) {
+        throw new IOException("Failed to bind", future.cause());
+      }
+      final Channel channel = future.channel();
+      parentChannelRef.set(channel);
+      parentChannels.add(channel);
+      boundListenAddresses.add(channel.localAddress());
+      Future<?> channelzFuture = channel.eventLoop().submit(new Runnable() {
+        @Override
+        public void run() {
+          ListenSocket listenSocket = new ListenSocket(channel);
+          listenSocketStats.add(listenSocket);
+          channelz.addListenSocket(listenSocket);
+        }
+      });
+      try {
+        channelzFuture.await();
+      } catch (InterruptedException ex) {
+        throw new RuntimeException("Interrupted while registering listen socket to channelz", ex);
       }
-    });
-    try {
-      channelzFuture.await();
-    } catch (InterruptedException ex) {
-      throw new RuntimeException("Interrupted while registering listen socket to channelz", ex);
     }
   }
 
   @Override
   public void shutdown() {
-    if (channel == null || !channel.isOpen()) {
-      // Already closed.
+    List<Channel> localParentChannels = parentChannels;
+    final AtomicInteger remaining = new AtomicInteger(localParentChannels.size());
+    parentChannels = Collections.emptyList();
+    if (remaining.get() == 0) {
+      // Degenerate case, no sockets.
+      synchronized (NettyServer.this) {
+        listener.serverShutdown();
+      }
       return;
     }
-    channel.close().addListener(new ChannelFutureListener() {
-      @Override
-      public void operationComplete(ChannelFuture future) throws Exception {
-        if (!future.isSuccess()) {
-          log.log(Level.WARNING, "Error shutting down server", future.cause());
-        }
-        for (InternalInstrumented<SocketStats> listenSocket : listenSockets) {
-          channelz.removeListenSocket(listenSocket);
-        }
-        listenSockets = null;
-        synchronized (NettyServer.this) {
-          listener.serverShutdown();
-        }
-        eventLoopReferenceCounter.release();
+    for (final Channel localParentChannel : localParentChannels) {
+      if (!localParentChannel.isOpen()) {
+        // Already closed.
+        continue;
       }
-    });
+      final SocketAddress boundListenAddress = localParentChannel.localAddress();
+      localParentChannel.close().addListener(new ChannelFutureListener() {
+        @Override
+        public void operationComplete(ChannelFuture future) {
+          if (!future.isSuccess()) {
+            log.log(Level.WARNING, "Error shutting down server", future.cause());
+          }
+          boundListenAddresses.remove(boundListenAddress);
+          // This ends up being quadratic, but presumably the number of listen sockets is small.
+          // Also, it's probably not worth maintaining another map of channel to listenSocket.
+          for (ListenSocket listenSocket : listenSocketStats) {
+            // compare to localParentChannel instead of future.channel(), which may be null.
+            if (listenSocket.ch == localParentChannel) {
+              channelz.removeListenSocket(listenSocket);
+              break;
+            }
+          }
+          if (remaining.decrementAndGet() == 0) {
+            synchronized (NettyServer.this) {
+              listenSocketStats.clear();
+              listener.serverShutdown();
+            }
+          }
+        }
+      });
+    }
+    eventLoopReferenceCounter.release();
   }
 
   private void allocateSharedGroups() {
@@ -313,11 +352,11 @@ public InternalLogId getLogId() {
   public String toString() {
     return MoreObjects.toStringHelper(this)
         .add("logId", logId.getId())
-        .add("address", address)
+        .add("addresses", boundListenAddresses)
         .toString();
   }
 
-  class EventLoopReferenceCounter extends AbstractReferenceCounted {
+  final class EventLoopReferenceCounter extends AbstractReferenceCounted {
     @Override
     protected void deallocate() {
       try {