netty: support listening on multiple port

Author: carl-mastrangelo

core/src/main/java/io/grpc/InternalChannelz.java

@@ -311,7 +311,7 @@ public static final class Builder {
       private long callsSucceeded;
       private long callsFailed;
       private long lastCallStartedNanos;
-      public List<InternalInstrumented<SocketStats>> listenSockets = Collections.emptyList();
+      public List<InternalInstrumented<SocketStats>> listenSockets = new ArrayList<>();
 
       public Builder setCallsStarted(long callsStarted) {
         this.callsStarted = callsStarted;
@@ -334,10 +334,11 @@ public Builder setLastCallStartedNanos(long lastCallStartedNanos) {
       }
 
       /** Sets the listen sockets. */
-      public Builder setListenSockets(List<InternalInstrumented<SocketStats>> listenSockets) {
-        checkNotNull(listenSockets);
-        this.listenSockets = Collections.unmodifiableList(
-            new ArrayList<InternalInstrumented<SocketStats>>(listenSockets));
+      public Builder addListenSockets(List<InternalInstrumented<SocketStats>> listenSockets) {
+        checkNotNull(listenSockets, "listenSockets");
+        for (InternalInstrumented<SocketStats> ss : listenSockets) {
+          this.listenSockets.add(checkNotNull(ss, "null listen socket"));
+        }
         return this;
       }
 

core/src/main/java/io/grpc/inprocess/InProcessServerBuilder.java

@@ -27,6 +27,7 @@
 import io.grpc.internal.ObjectPool;
 import io.grpc.internal.SharedResourcePool;
 import java.io.File;
+import java.util.Collections;
 import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.ScheduledExecutorService;
@@ -145,9 +146,9 @@ public InProcessServerBuilder maxInboundMetadataSize(int bytes) {
   }
 
   @Override
-  protected InProcessServer buildTransportServer(
+  protected List<InProcessServer> buildTransportServers(
       List<? extends ServerStreamTracer.Factory> streamTracerFactories) {
-    return new InProcessServer(this, streamTracerFactories);
+    return Collections.singletonList(new InProcessServer(this, streamTracerFactories));
   }
 
   @Override

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

@@ -220,7 +220,7 @@ protected void setTracingEnabled(boolean value) {
   public final Server build() {
     ServerImpl server = new ServerImpl(
         this,
-        buildTransportServer(getTracerFactories()),
+        buildTransportServers(getTracerFactories()),
         Context.ROOT);
     for (InternalNotifyOnServerBuild notifyTarget : notifyOnBuildList) {
       notifyTarget.notifyOnBuild(server);
@@ -266,7 +266,7 @@ protected final TransportTracer.Factory getTransportTracerFactory() {
    *
    * @param streamTracerFactories an immutable list of stream tracer factories
    */
-  protected abstract io.grpc.internal.InternalServer buildTransportServer(
+  protected abstract List<? extends io.grpc.internal.InternalServer> buildTransportServers(
       List<? extends ServerStreamTracer.Factory> streamTracerFactories);
 
   private T thisT() {

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

@@ -101,12 +101,11 @@ public final class ServerImpl extends io.grpc.Server implements InternalInstrume
   @GuardedBy("lock") private boolean serverShutdownCallbackInvoked;
   @GuardedBy("lock") private boolean terminated;
   /** Service encapsulating something similar to an accept() socket. */
-  private final InternalServer transportServer;
+  private final List<? extends InternalServer> transportServers;
   private final Object lock = new Object();
-  @GuardedBy("lock") private boolean transportServerTerminated;
+  @GuardedBy("lock") private boolean transportServersTerminated;
   /** {@code transportServer} and services encapsulating something similar to a TCP connection. */
-  @GuardedBy("lock") private final Collection<ServerTransport> transports =
-      new HashSet<ServerTransport>();
+  @GuardedBy("lock") private final Collection<ServerTransport> transports = new HashSet<>();
 
   private final Context rootContext;
 
@@ -126,14 +125,18 @@ public final class ServerImpl extends io.grpc.Server implements InternalInstrume
    */
   ServerImpl(
       AbstractServerImplBuilder<?> builder,
-      InternalServer transportServer,
+      List<? extends InternalServer> transportServers,
       Context rootContext) {
     this.executorPool = Preconditions.checkNotNull(builder.executorPool, "executorPool");
     this.registry = Preconditions.checkNotNull(builder.registryBuilder.build(), "registryBuilder");
     this.fallbackRegistry =
         Preconditions.checkNotNull(builder.fallbackRegistry, "fallbackRegistry");
-    this.transportServer = Preconditions.checkNotNull(transportServer, "transportServer");
-    this.logId = InternalLogId.allocate("Server", String.valueOf(transportServer.getPort()));
+    Preconditions.checkNotNull(transportServers, "transportServers");
+    Preconditions.checkArgument(!transportServers.isEmpty(), "no servers provided");
+    this.transportServers = new ArrayList<>(transportServers);
+    // TODO(notcarl): concatenate all listening ports in the Log Id.
+    this.logId =
+        InternalLogId.allocate("Server", String.valueOf(transportServers.get(0).getPort()));
     // Fork from the passed in context so that it does not propagate cancellation, it only
     // inherits values.
     this.rootContext = Preconditions.checkNotNull(rootContext, "rootContext").fork();
@@ -163,8 +166,12 @@ public ServerImpl start() throws IOException {
     synchronized (lock) {
       checkState(!started, "Already started");
       checkState(!shutdown, "Shutting down");
-      // Start and wait for any port to actually be bound.
-      transportServer.start(new ServerListenerImpl());
+      // Start and wait for any ports to actually be bound.
+
+      for (InternalServer ts : transportServers) {
+        ServerListenerImpl listener = new ServerListenerImpl(ts);
+        ts.start(listener);
+      }
       executor = Preconditions.checkNotNull(executorPool.getObject(), "executor");
       started = true;
       return this;
@@ -176,7 +183,13 @@ public int getPort() {
     synchronized (lock) {
       checkState(started, "Not started");
       checkState(!terminated, "Already terminated");
-      return transportServer.getPort();
+      for (InternalServer ts : transportServers) {
+        int port = ts.getPort();
+        if (port != -1) {
+          return port;
+        }
+      }
+      return -1;
     }
   }
 
@@ -211,20 +224,22 @@ public List<ServerServiceDefinition> getMutableServices() {
    */
   @Override
   public ServerImpl shutdown() {
-    boolean shutdownTransportServer;
+    boolean shutdownTransportServers;
     synchronized (lock) {
       if (shutdown) {
         return this;
       }
       shutdown = true;
-      shutdownTransportServer = started;
-      if (!shutdownTransportServer) {
-        transportServerTerminated = true;
+      shutdownTransportServers = started;
+      if (!shutdownTransportServers) {
+        transportServersTerminated = true;
         checkForTermination();
       }
     }
-    if (shutdownTransportServer) {
-      transportServer.shutdown();
+    if (shutdownTransportServers) {
+      for (InternalServer ts : transportServers) {
+        ts.shutdown();
+      }
     }
     return this;
   }
@@ -311,7 +326,7 @@ private void transportClosed(ServerTransport transport) {
   /** Notify of complete shutdown if necessary. */
   private void checkForTermination() {
     synchronized (lock) {
-      if (shutdown && transports.isEmpty() && transportServerTerminated) {
+      if (shutdown && transports.isEmpty() && transportServersTerminated) {
         if (terminated) {
           throw new AssertionError("Server already terminated");
         }
@@ -327,6 +342,11 @@ private void checkForTermination() {
   }
 
   private final class ServerListenerImpl implements ServerListener {
+
+    ServerListenerImpl(InternalServer transportServer) {
+
+    }
+
     @Override
     public ServerTransportListener transportCreated(ServerTransport transport) {
       synchronized (lock) {
@@ -356,7 +376,7 @@ public void serverShutdown() {
         }
       }
       synchronized (lock) {
-        transportServerTerminated = true;
+        transportServersTerminated = true;
         checkForTermination();
       }
     }
@@ -577,9 +597,10 @@ public InternalLogId getLogId() {
 
   @Override
   public ListenableFuture<ServerStats> getStats() {
-    ServerStats.Builder builder
-        = new ServerStats.Builder()
-        .setListenSockets(transportServer.getListenSockets());
+    ServerStats.Builder builder = new ServerStats.Builder();
+    for (InternalServer ts : transportServers) {
+      builder.addListenSockets(ts.getListenSockets());
+    }
     serverCallTracer.updateBuilder(builder);
     SettableFuture<ServerStats> ret = SettableFuture.create();
     ret.set(builder.build());
@@ -590,7 +611,7 @@ public ListenableFuture<ServerStats> getStats() {
   public String toString() {
     return MoreObjects.toStringHelper(this)
         .add("logId", logId.getId())
-        .add("transportServer", transportServer)
+        .add("transportServers", transportServers)
         .toString();
   }
 

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

@@ -22,6 +22,7 @@
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertSame;
 
+import com.google.common.collect.Iterables;
 import io.grpc.ServerStreamTracer.Factory;
 import io.grpc.internal.FakeClock;
 import io.grpc.internal.ObjectPool;
@@ -54,7 +55,8 @@ public void generateName() {
   @Test
   public void scheduledExecutorService_default() {
     InProcessServerBuilder builder = InProcessServerBuilder.forName("foo");
-    InProcessServer server = builder.buildTransportServer(new ArrayList<Factory>());
+    InProcessServer server =
+        Iterables.getOnlyElement(builder.buildTransportServers(new ArrayList<Factory>()));
 
     ObjectPool<ScheduledExecutorService> scheduledExecutorServicePool =
         server.getScheduledExecutorServicePool();
@@ -78,7 +80,8 @@ public void scheduledExecutorService_custom() {
     InProcessServerBuilder builder1 = builder.scheduledExecutorService(scheduledExecutorService);
     assertSame(builder, builder1);
 
-    InProcessServer server = builder1.buildTransportServer(new ArrayList<Factory>());
+    InProcessServer server =
+        Iterables.getOnlyElement(builder1.buildTransportServers(new ArrayList<Factory>()));
     ObjectPool<ScheduledExecutorService> scheduledExecutorServicePool =
         server.getScheduledExecutorServicePool();
 

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

@@ -21,6 +21,7 @@
 import io.grpc.internal.InternalServer;
 import io.grpc.internal.ManagedClientTransport;
 import io.grpc.internal.testing.AbstractTransportTest;
+import java.util.Collections;
 import java.util.List;
 import org.junit.Ignore;
 import org.junit.Test;
@@ -35,16 +36,17 @@ public class InProcessTransportTest extends AbstractTransportTest {
   private static final String USER_AGENT = "a-testing-user-agent";
 
   @Override
-  protected InternalServer newServer(List<ServerStreamTracer.Factory> streamTracerFactories) {
+  protected List<? extends InternalServer> newServer(
+      List<ServerStreamTracer.Factory> streamTracerFactories) {
     InProcessServerBuilder builder = InProcessServerBuilder
         .forName(TRANSPORT_NAME)
         .maxInboundMetadataSize(GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE);
-    return new InProcessServer(builder, streamTracerFactories);
+    return Collections.singletonList(new InProcessServer(builder, streamTracerFactories));
   }
 
   @Override
-  protected InternalServer newServer(
-      InternalServer server, List<ServerStreamTracer.Factory> streamTracerFactories) {
+  protected List<? extends InternalServer> newServer(
+      int port, List<ServerStreamTracer.Factory> streamTracerFactories) {
     return newServer(streamTracerFactories);
   }
 

core/src/test/java/io/grpc/internal/AbstractServerImplBuilderTest.java

@@ -101,7 +101,7 @@ static class Builder extends AbstractServerImplBuilder<Builder> {
     }
 
     @Override
-    protected io.grpc.internal.InternalServer buildTransportServer(
+    protected List<io.grpc.internal.InternalServer> buildTransportServers(
         List<? extends ServerStreamTracer.Factory> streamTracerFactories) {
       throw new UnsupportedOperationException();
     }

core/src/test/java/io/grpc/internal/ServerImplTest.java

@@ -1326,7 +1326,7 @@ private void createServer() {
 
     builder.fallbackHandlerRegistry(fallbackRegistry);
     builder.executorPool = executorPool;
-    server = new ServerImpl(builder, transportServer, SERVER_CONTEXT);
+    server = new ServerImpl(builder, Collections.singletonList(transportServer), SERVER_CONTEXT);
   }
 
   private void verifyExecutorsAcquired() {
@@ -1411,7 +1411,7 @@ public ListenableFuture<SocketStats> getStats() {
   }
 
   private static class Builder extends AbstractServerImplBuilder<Builder> {
-    @Override protected InternalServer buildTransportServer(
+    @Override protected List<InternalServer> buildTransportServers(
         List<? extends ServerStreamTracer.Factory> streamTracerFactories) {
       throw new UnsupportedOperationException();
     }

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

@@ -39,6 +39,8 @@
 import java.io.InputStream;
 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;
@@ -65,7 +67,7 @@ public final class NettyServerBuilder extends AbstractServerImplBuilder<NettySer
   private static final long MIN_MAX_CONNECTION_AGE_NANO = TimeUnit.SECONDS.toNanos(1L);
   private static final long AS_LARGE_AS_INFINITE = TimeUnit.DAYS.toNanos(1000L);
 
-  private final SocketAddress address;
+  private final List<SocketAddress> listenAddresses = new ArrayList<>();
   private Class<? extends ServerChannel> channelType = NioServerSocketChannel.class;
   private final Map<ChannelOption<?>, Object> channelOptions = new HashMap<>();
   @Nullable
@@ -110,12 +112,12 @@ public static NettyServerBuilder forAddress(SocketAddress address) {
 
   @CheckReturnValue
   private NettyServerBuilder(int port) {
-    this.address = new InetSocketAddress(port);
+    this.listenAddresses.add(new InetSocketAddress(port));
   }
 
   @CheckReturnValue
   private NettyServerBuilder(SocketAddress address) {
-    this.address = address;
+    this.listenAddresses.add(address);
   }
 
   /**
@@ -436,22 +438,26 @@ public NettyServerBuilder permitKeepAliveWithoutCalls(boolean permit) {
 
   @Override
   @CheckReturnValue
-  protected NettyServer buildTransportServer(
+  protected List<NettyServer> buildTransportServers(
       List<? extends ServerStreamTracer.Factory> streamTracerFactories) {
     ProtocolNegotiator negotiator = protocolNegotiator;
     if (negotiator == null) {
       negotiator = sslContext != null ? ProtocolNegotiators.serverTls(sslContext) :
               ProtocolNegotiators.serverPlaintext();
     }
-
-    return new NettyServer(
-        address, channelType, channelOptions, bossEventLoopGroup, workerEventLoopGroup,
-        negotiator, streamTracerFactories, getTransportTracerFactory(),
-        maxConcurrentCallsPerConnection, flowControlWindow,
-        maxMessageSize, maxHeaderListSize, keepAliveTimeInNanos, keepAliveTimeoutInNanos,
-        maxConnectionIdleInNanos,
-        maxConnectionAgeInNanos, maxConnectionAgeGraceInNanos,
-        permitKeepAliveWithoutCalls, permitKeepAliveTimeInNanos, getChannelz());
+    List<NettyServer> transportServers = new ArrayList<>(listenAddresses.size());
+    for (SocketAddress listenAddress : listenAddresses) {
+      NettyServer transportServer = new NettyServer(
+          listenAddress, channelType, channelOptions, bossEventLoopGroup, workerEventLoopGroup,
+          negotiator, streamTracerFactories, getTransportTracerFactory(),
+          maxConcurrentCallsPerConnection, flowControlWindow,
+          maxMessageSize, maxHeaderListSize, keepAliveTimeInNanos, keepAliveTimeoutInNanos,
+          maxConnectionIdleInNanos,
+          maxConnectionAgeInNanos, maxConnectionAgeGraceInNanos,
+          permitKeepAliveWithoutCalls, permitKeepAliveTimeInNanos, getChannelz());
+      transportServers.add(transportServer);
+    }
+    return Collections.unmodifiableList(transportServers);
   }
 
   @Override

netty/src/test/java/io/grpc/netty/NettyTransportTest.java

@@ -55,23 +55,23 @@ public void releaseClientFactory() {
   }
 
   @Override
-  protected InternalServer newServer(List<ServerStreamTracer.Factory> streamTracerFactories) {
+  protected List<? extends InternalServer> newServer(
+      List<ServerStreamTracer.Factory> streamTracerFactories) {
     return NettyServerBuilder
         .forPort(0)
         .flowControlWindow(65 * 1024)
         .setTransportTracerFactory(fakeClockTransportTracer)
-        .buildTransportServer(streamTracerFactories);
+        .buildTransportServers(streamTracerFactories);
   }
 
   @Override
-  protected InternalServer newServer(
-      InternalServer server, List<ServerStreamTracer.Factory> streamTracerFactories) {
-    int port = server.getPort();
+  protected List<? extends InternalServer> newServer(
+      int port, List<ServerStreamTracer.Factory> streamTracerFactories) {
     return NettyServerBuilder
         .forPort(port)
         .flowControlWindow(65 * 1024)
         .setTransportTracerFactory(fakeClockTransportTracer)
-        .buildTransportServer(streamTracerFactories);
+        .buildTransportServers(streamTracerFactories);
   }
 
   @Override