Add support for epoll

build.gradle

@@ -22,6 +22,8 @@ dependencies {
     compile "io.netty:netty-codec:${nettyVersion}"
     compile "io.netty:netty-codec-http:${nettyVersion}"
     compile "io.netty:netty-handler:${nettyVersion}"
+    compile "io.netty:netty-transport-native-epoll:${nettyVersion}"
+    compile "io.netty:netty-transport-native-epoll:${nettyVersion}:linux-x86_64"
     compile 'net.kyori:text:1.12-1.5.0'
     testCompile 'org.junit.jupiter:junit-jupiter-api:5.3.0-M1'
     testCompile 'org.junit.jupiter:junit-jupiter-engine:5.3.0-M1'

src/main/java/com/velocitypowered/network/ConnectionManager.java

@@ -0,0 +1,148 @@
+package com.velocitypowered.network;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.velocitypowered.proxy.connection.MinecraftConnection;
+import com.velocitypowered.proxy.connection.client.HandshakeSessionHandler;
+import com.velocitypowered.proxy.protocol.ProtocolConstants;
+import com.velocitypowered.proxy.protocol.StateRegistry;
+import com.velocitypowered.proxy.protocol.netty.LegacyPingDecoder;
+import com.velocitypowered.proxy.protocol.netty.LegacyPingEncoder;
+import com.velocitypowered.proxy.protocol.netty.MinecraftDecoder;
+import com.velocitypowered.proxy.protocol.netty.MinecraftEncoder;
+import com.velocitypowered.proxy.protocol.netty.MinecraftVarintFrameDecoder;
+import com.velocitypowered.proxy.protocol.netty.MinecraftVarintLengthEncoder;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.epoll.Epoll;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.epoll.EpollServerSocketChannel;
+import io.netty.channel.epoll.EpollSocketChannel;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.ServerSocketChannel;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.timeout.ReadTimeoutHandler;
+
+import java.net.InetSocketAddress;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+import static com.velocitypowered.network.Connections.CLIENT_READ_TIMEOUT_SECONDS;
+import static com.velocitypowered.network.Connections.FRAME_DECODER;
+import static com.velocitypowered.network.Connections.FRAME_ENCODER;
+import static com.velocitypowered.network.Connections.LEGACY_PING_DECODER;
+import static com.velocitypowered.network.Connections.LEGACY_PING_ENCODER;
+import static com.velocitypowered.network.Connections.MINECRAFT_DECODER;
+import static com.velocitypowered.network.Connections.MINECRAFT_ENCODER;
+import static com.velocitypowered.network.Connections.READ_TIMEOUT;
+
+public final class ConnectionManager {
+    private static final String DISABLE_EPOLL_PROPERTY = "velocity.connection.disable-epoll";
+    private static final boolean DISABLE_EPOLL = Boolean.getBoolean(DISABLE_EPOLL_PROPERTY);
+    private final Set<Channel> endpoints = new HashSet<>();
+    private final Class<? extends ServerSocketChannel> serverSocketChannelClass;
+    private final Class<? extends SocketChannel> socketChannelClass;
+    private final EventLoopGroup bossGroup;
+    private final EventLoopGroup workerGroup;
+
+    public ConnectionManager() {
+        final boolean epoll = canUseEpoll();
+        if (epoll) {
+            this.serverSocketChannelClass = EpollServerSocketChannel.class;
+            this.socketChannelClass = EpollSocketChannel.class;
+            this.bossGroup = new EpollEventLoopGroup(0, createThreadFactory("Netty Epoll Boss #%d"));
+            this.workerGroup = new EpollEventLoopGroup(0, createThreadFactory("Netty Epoll Worker #%d"));
+        } else {
+            this.serverSocketChannelClass = NioServerSocketChannel.class;
+            this.socketChannelClass = NioSocketChannel.class;
+            this.bossGroup = new NioEventLoopGroup(0, createThreadFactory("Netty Nio Boss #%d"));
+            this.workerGroup = new NioEventLoopGroup(0, createThreadFactory("Netty Nio Worker #%d"));
+        }
+        this.logChannelInformation(epoll);
+    }
+
+    private void logChannelInformation(final boolean epoll) {
+        final StringBuilder sb = new StringBuilder();
+        sb.append("Channel type: ");
+        sb.append(epoll ? "epoll": "nio");
+        if(DISABLE_EPOLL) {
+            sb.append(String.format(" - epoll explicitly disabled using -D%s=true", DISABLE_EPOLL_PROPERTY));
+        }
+        System.out.println(sb.toString()); // TODO: move to logger
+    }
+
+    public void bind(final InetSocketAddress address) {
+        final ServerBootstrap bootstrap = new ServerBootstrap()
+                .channel(this.serverSocketChannelClass)
+                .group(this.bossGroup, this.workerGroup)
+                .childHandler(new ChannelInitializer<Channel>() {
+                    @Override
+                    protected void initChannel(final Channel ch) {
+                        ch.pipeline()
+                                .addLast(READ_TIMEOUT, new ReadTimeoutHandler(CLIENT_READ_TIMEOUT_SECONDS, TimeUnit.SECONDS))
+                                .addLast(LEGACY_PING_DECODER, new LegacyPingDecoder())
+                                .addLast(FRAME_DECODER, new MinecraftVarintFrameDecoder())
+                                .addLast(LEGACY_PING_ENCODER, LegacyPingEncoder.INSTANCE)
+                                .addLast(FRAME_ENCODER, MinecraftVarintLengthEncoder.INSTANCE)
+                                .addLast(MINECRAFT_DECODER, new MinecraftDecoder(ProtocolConstants.Direction.TO_SERVER))
+                                .addLast(MINECRAFT_ENCODER, new MinecraftEncoder(ProtocolConstants.Direction.TO_CLIENT));
+
+                        final MinecraftConnection connection = new MinecraftConnection(ch);
+                        connection.setState(StateRegistry.HANDSHAKE);
+                        connection.setSessionHandler(new HandshakeSessionHandler(connection));
+                        ch.pipeline().addLast(Connections.HANDLER, connection);
+                    }
+                })
+                .childOption(ChannelOption.TCP_NODELAY, true)
+                .childOption(ChannelOption.IP_TOS, 0x18)
+                .localAddress(address);
+        bootstrap.bind()
+                .addListener((ChannelFutureListener) future -> {
+                    final Channel channel = future.channel();
+                    if (future.isSuccess()) {
+                        this.endpoints.add(channel);
+                        System.out.println("Listening on " + channel.localAddress()); // TODO: move to logger
+                    } else {
+                        System.out.println("Can't bind to " + channel.localAddress()); // TODO: move to logger
+                        future.cause().printStackTrace();
+                    }
+                });
+    }
+
+    public Bootstrap createWorker() {
+        return new Bootstrap()
+                .channel(this.socketChannelClass)
+                .group(this.workerGroup);
+    }
+
+    public void shutdown() {
+        for (final Channel endpoint : this.endpoints) {
+            try {
+                System.out.println(String.format("Closing endpoint %s", endpoint.localAddress())); // TODO: move to logger
+                endpoint.close().sync();
+            } catch (final InterruptedException e) {
+                System.err.println("Interrupted whilst closing endpoint"); // TODO: move to logger
+                e.printStackTrace();
+            }
+        }
+    }
+
+    private static boolean canUseEpoll() {
+        return Epoll.isAvailable() && !DISABLE_EPOLL;
+    }
+
+    private static ThreadFactory createThreadFactory(final String nameFormat) {
+        return new ThreadFactoryBuilder()
+                .setNameFormat(nameFormat)
+                .setDaemon(true)
+                .build();
+    }
+}

src/main/java/com/velocitypowered/network/Connections.java

@@ -0,0 +1,19 @@
+package com.velocitypowered.network;
+
+public interface Connections {
+    String CIPHER_DECODER = "cipher-decoder";
+    String CIPHER_ENCODER = "cipher-encoder";
+    String COMPRESSION_DECODER = "compression-decoder";
+    String COMPRESSION_ENCODER = "compression-encoder";
+    String FRAME_DECODER = "frame-decoder";
+    String FRAME_ENCODER = "frame-encoder";
+    String HANDLER = "handler";
+    String LEGACY_PING_DECODER = "legacy-ping-decoder";
+    String LEGACY_PING_ENCODER = "legacy-ping-encoder";
+    String MINECRAFT_DECODER = "minecraft-decoder";
+    String MINECRAFT_ENCODER = "minecraft-encoder";
+    String READ_TIMEOUT = "read-timeout";
+
+    int CLIENT_READ_TIMEOUT_SECONDS = 30; // client -> proxy
+    int SERVER_READ_TIMEOUT_SECONDS = 30; // proxy -> server
+}

src/main/java/com/velocitypowered/proxy/Velocity.java

@@ -2,8 +2,10 @@ package com.velocitypowered.proxy;
 
 public class Velocity {
     public static void main(String... args) throws InterruptedException {
-        VelocityServer server = new VelocityServer();
+        final VelocityServer server = VelocityServer.getServer();
-        server.initialize();
+        server.start();
+
+        Runtime.getRuntime().addShutdownHook(new Thread(server::shutdown, "Shutdown thread"));
 
         Thread.currentThread().join();
     }

src/main/java/com/velocitypowered/proxy/VelocityServer.java

@@ -1,99 +1,53 @@
 package com.velocitypowered.proxy;
 
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
-import com.velocitypowered.proxy.connection.MinecraftConnection;
+import com.velocitypowered.network.ConnectionManager;
 import com.velocitypowered.proxy.connection.http.NettyHttpClient;
-import com.velocitypowered.proxy.protocol.StateRegistry;
+import com.velocitypowered.proxy.util.EncryptionUtils;
-import com.velocitypowered.proxy.protocol.netty.MinecraftPipelineUtils;
-import com.velocitypowered.proxy.connection.client.HandshakeSessionHandler;
-import com.velocitypowered.proxy.protocol.packets.EncryptionRequest;
 import io.netty.bootstrap.Bootstrap;
-import io.netty.bootstrap.ServerBootstrap;
-import io.netty.channel.*;
-import io.netty.channel.nio.NioEventLoopGroup;
-import io.netty.channel.socket.nio.NioServerSocketChannel;
-import io.netty.channel.socket.nio.NioSocketChannel;
 import net.kyori.text.Component;
 import net.kyori.text.serializer.GsonComponentSerializer;
 
+import java.net.InetSocketAddress;
 import java.security.KeyPair;
-import java.security.KeyPairGenerator;
-import java.security.NoSuchAlgorithmException;
 
 public class VelocityServer {
+    private static final VelocityServer INSTANCE = new VelocityServer();
     public static final Gson GSON = new GsonBuilder()
             .registerTypeHierarchyAdapter(Component.class, new GsonComponentSerializer())
             .create();
-    private static VelocityServer server;
 
-    private EventLoopGroup bossGroup;
+    private final ConnectionManager cm = new ConnectionManager();
-    private EventLoopGroup childGroup;
     private NettyHttpClient httpClient;
     private KeyPair serverKeyPair;
 
-    public VelocityServer() {
+    private VelocityServer() {
-
     }
 
     public static VelocityServer getServer() {
-        return server;
+        return INSTANCE;
     }
 
     public KeyPair getServerKeyPair() {
         return serverKeyPair;
     }
 
-    public void initialize() {
+    public void start() {
         // Create a key pair
-        try {
+        serverKeyPair = EncryptionUtils.createRsaKeyPair(1024);
-            KeyPairGenerator generator = KeyPairGenerator.getInstance("RSA");
-            generator.initialize(1024);
-            serverKeyPair = generator.generateKeyPair();
-        } catch (NoSuchAlgorithmException e) {
-            throw new RuntimeException("Unable to generate server encryption key", e);
-        }
 
-        // Start the listener
-        bossGroup = new NioEventLoopGroup(0, new ThreadFactoryBuilder().setDaemon(true).setNameFormat("Netty Boss Thread").build());
-        childGroup = new NioEventLoopGroup(0, new ThreadFactoryBuilder().setDaemon(true).setNameFormat("Netty I/O Thread #%d").build());
         httpClient = new NettyHttpClient(this);
-        server = this;
-        new ServerBootstrap()
-                .channel(NioServerSocketChannel.class)
-                .group(bossGroup, childGroup)
-                .childOption(ChannelOption.TCP_NODELAY, true)
-                .childOption(ChannelOption.IP_TOS, 0x18)
-                .childHandler(new ChannelInitializer<Channel>() {
-                    @Override
-                    protected void initChannel(Channel ch) throws Exception {
-                        MinecraftPipelineUtils.strapPipelineForProxy(ch);
 
-                        MinecraftConnection connection = new MinecraftConnection(ch);
+        this.cm.bind(new InetSocketAddress(26671));
-                        connection.setState(StateRegistry.HANDSHAKE);
-                        connection.setSessionHandler(new HandshakeSessionHandler(connection));
-                        ch.pipeline().addLast("handler", connection);
-                    }
-                })
-                .bind(26671)
-                .addListener(new ChannelFutureListener() {
-                    @Override
-                    public void operationComplete(ChannelFuture future) throws Exception {
-                        if (future.isSuccess()) {
-                            System.out.println("Listening on " + future.channel().localAddress());
-                        } else {
-                            System.out.println("Can't bind to " + future.channel().localAddress());
-                            future.cause().printStackTrace();
-                        }
-                    }
-                });
     }
 
     public Bootstrap initializeGenericBootstrap() {
-        return new Bootstrap()
+        return this.cm.createWorker();
-                .channel(NioSocketChannel.class)
+    }
-                .group(childGroup);
+
+    public void shutdown() {
+        this.cm.shutdown();
     }
 
     public NettyHttpClient getHttpClient() {

src/main/java/com/velocitypowered/proxy/connection/MinecraftConnection.java

@@ -20,7 +20,14 @@ import javax.crypto.spec.SecretKeySpec;
 
 import java.security.GeneralSecurityException;
 
-import static com.velocitypowered.proxy.protocol.netty.MinecraftPipelineUtils.*;
+import static com.velocitypowered.network.Connections.CIPHER_DECODER;
+import static com.velocitypowered.network.Connections.CIPHER_ENCODER;
+import static com.velocitypowered.network.Connections.COMPRESSION_DECODER;
+import static com.velocitypowered.network.Connections.COMPRESSION_ENCODER;
+import static com.velocitypowered.network.Connections.FRAME_DECODER;
+import static com.velocitypowered.network.Connections.FRAME_ENCODER;
+import static com.velocitypowered.network.Connections.MINECRAFT_DECODER;
+import static com.velocitypowered.network.Connections.MINECRAFT_ENCODER;
 
 /**
  * A utility class to make working with the pipeline a little less painful and transparently handles certain Minecraft
@@ -149,8 +156,8 @@ public class MinecraftConnection extends ChannelInboundHandlerAdapter {
 
     public void setCompressionThreshold(int threshold) {
         if (threshold == -1) {
-            channel.pipeline().remove("compress-decoder");
+            channel.pipeline().remove(COMPRESSION_DECODER);
-            channel.pipeline().remove("compress-encoder");
+            channel.pipeline().remove(COMPRESSION_ENCODER);
             return;
         }
 
@@ -158,8 +165,8 @@ public class MinecraftConnection extends ChannelInboundHandlerAdapter {
         MinecraftCompressEncoder encoder = new MinecraftCompressEncoder(threshold, compressor);
         MinecraftCompressDecoder decoder = new MinecraftCompressDecoder(threshold, compressor);
 
-        channel.pipeline().addBefore(MINECRAFT_DECODER, "compress-decoder", decoder);
+        channel.pipeline().addBefore(MINECRAFT_DECODER, COMPRESSION_DECODER, decoder);
-        channel.pipeline().addBefore(MINECRAFT_ENCODER, "compress-encoder", encoder);
+        channel.pipeline().addBefore(MINECRAFT_ENCODER, COMPRESSION_ENCODER, encoder);
     }
 
     public void enableEncryption(byte[] secret) throws GeneralSecurityException {
@@ -167,7 +174,7 @@ public class MinecraftConnection extends ChannelInboundHandlerAdapter {
 
         VelocityCipher decryptionCipher = new JavaVelocityCipher(false, key);
         VelocityCipher encryptionCipher = new JavaVelocityCipher(true, key);
-        channel.pipeline().addBefore(FRAME_DECODER, "cipher-decoder", new MinecraftCipherDecoder(decryptionCipher));
+        channel.pipeline().addBefore(FRAME_DECODER, CIPHER_DECODER, new MinecraftCipherDecoder(decryptionCipher));
-        channel.pipeline().addBefore(FRAME_ENCODER, "cipher-encoder", new MinecraftCipherEncoder(encryptionCipher));
+        channel.pipeline().addBefore(FRAME_ENCODER, CIPHER_ENCODER, new MinecraftCipherEncoder(encryptionCipher));
     }
 }

src/main/java/com/velocitypowered/proxy/connection/backend/ServerConnection.java

@@ -1,14 +1,29 @@
 package com.velocitypowered.proxy.connection.backend;
 
+import com.velocitypowered.proxy.protocol.ProtocolConstants;
+import com.velocitypowered.proxy.protocol.netty.MinecraftDecoder;
+import com.velocitypowered.proxy.protocol.netty.MinecraftEncoder;
+import com.velocitypowered.proxy.protocol.netty.MinecraftVarintFrameDecoder;
+import com.velocitypowered.proxy.protocol.netty.MinecraftVarintLengthEncoder;
 import com.velocitypowered.proxy.protocol.packets.Handshake;
 import com.velocitypowered.proxy.protocol.packets.ServerLogin;
 import com.velocitypowered.proxy.connection.MinecraftConnection;
 import com.velocitypowered.proxy.protocol.StateRegistry;
 import com.velocitypowered.proxy.data.ServerInfo;
-import com.velocitypowered.proxy.protocol.netty.MinecraftPipelineUtils;
 import com.velocitypowered.proxy.VelocityServer;
 import com.velocitypowered.proxy.connection.client.ConnectedPlayer;
 import io.netty.channel.*;
+import io.netty.handler.timeout.ReadTimeoutHandler;
+
+import java.util.concurrent.TimeUnit;
+
+import static com.velocitypowered.network.Connections.FRAME_DECODER;
+import static com.velocitypowered.network.Connections.FRAME_ENCODER;
+import static com.velocitypowered.network.Connections.HANDLER;
+import static com.velocitypowered.network.Connections.MINECRAFT_DECODER;
+import static com.velocitypowered.network.Connections.MINECRAFT_ENCODER;
+import static com.velocitypowered.network.Connections.READ_TIMEOUT;
+import static com.velocitypowered.network.Connections.SERVER_READ_TIMEOUT_SECONDS;
 
 public class ServerConnection {
     private final ServerInfo serverInfo;
@@ -27,12 +42,17 @@ public class ServerConnection {
                 .handler(new ChannelInitializer<Channel>() {
                     @Override
                     protected void initChannel(Channel ch) throws Exception {
-                        MinecraftPipelineUtils.strapPipelineForBackend(ch);
+                        ch.pipeline()
+                                .addLast(READ_TIMEOUT, new ReadTimeoutHandler(SERVER_READ_TIMEOUT_SECONDS, TimeUnit.SECONDS))
+                                .addLast(FRAME_DECODER, new MinecraftVarintFrameDecoder())
+                                .addLast(FRAME_ENCODER, MinecraftVarintLengthEncoder.INSTANCE)
+                                .addLast(MINECRAFT_DECODER, new MinecraftDecoder(ProtocolConstants.Direction.TO_CLIENT))
+                                .addLast(MINECRAFT_ENCODER, new MinecraftEncoder(ProtocolConstants.Direction.TO_SERVER));
 
                         MinecraftConnection connection = new MinecraftConnection(ch);
                         connection.setState(StateRegistry.HANDSHAKE);
                         connection.setSessionHandler(new LoginSessionHandler(ServerConnection.this));
-                        ch.pipeline().addLast("handler", connection);
+                        ch.pipeline().addLast(HANDLER, connection);
                     }
                 })
                 .connect(serverInfo.getAddress())

src/main/java/com/velocitypowered/proxy/protocol/netty/MinecraftPipelineUtils.java

@@ -1,37 +0,0 @@
-package com.velocitypowered.proxy.protocol.netty;
-
-import com.velocitypowered.proxy.protocol.ProtocolConstants;
-import io.netty.channel.Channel;
-import io.netty.handler.timeout.ReadTimeoutHandler;
-
-import java.util.concurrent.TimeUnit;
-
-public interface MinecraftPipelineUtils {
-    String FRAME_DECODER = "frame-decoder";
-    String FRAME_ENCODER = "frame-encoder";
-    String LEGACY_PING_DECODER = "legacy-ping-decoder";
-    String LEGACY_PING_ENCODER = "legacy-ping-encoder";
-    String MINECRAFT_DECODER = "minecraft-decoder";
-    String MINECRAFT_ENCODER = "minecraft-encoder";
-    String READ_TIMEOUT = "read-timeout";
-
-    static void strapPipelineForProxy(Channel ch) {
-        ch.pipeline()
-          .addLast(READ_TIMEOUT, new ReadTimeoutHandler(30, TimeUnit.SECONDS))
-          .addLast(LEGACY_PING_DECODER, new LegacyPingDecoder())
-          .addLast(FRAME_DECODER, new MinecraftVarintFrameDecoder())
-          .addLast(LEGACY_PING_ENCODER, LegacyPingEncoder.INSTANCE)
-          .addLast(FRAME_ENCODER, MinecraftVarintLengthEncoder.INSTANCE)
-          .addLast(MINECRAFT_DECODER, new MinecraftDecoder(ProtocolConstants.Direction.TO_SERVER))
-          .addLast(MINECRAFT_ENCODER, new MinecraftEncoder(ProtocolConstants.Direction.TO_CLIENT));
-    }
-
-    static void strapPipelineForBackend(Channel ch) {
-        ch.pipeline()
-          .addLast(READ_TIMEOUT, new ReadTimeoutHandler(30, TimeUnit.SECONDS))
-          .addLast(FRAME_DECODER, new MinecraftVarintFrameDecoder())
-          .addLast(FRAME_ENCODER, MinecraftVarintLengthEncoder.INSTANCE)
-          .addLast(MINECRAFT_DECODER, new MinecraftDecoder(ProtocolConstants.Direction.TO_CLIENT))
-          .addLast(MINECRAFT_ENCODER, new MinecraftEncoder(ProtocolConstants.Direction.TO_SERVER));
-    }
-}

src/main/java/com/velocitypowered/proxy/util/EncryptionUtils.java

@@ -4,7 +4,19 @@ import javax.crypto.Cipher;
 import java.math.BigInteger;
 import java.security.*;
 
-public enum EncryptionUtils { ;
+public enum EncryptionUtils {
+    ;
+
+    public static KeyPair createRsaKeyPair(final int keysize) {
+        try {
+            final KeyPairGenerator generator = KeyPairGenerator.getInstance("RSA");
+            generator.initialize(keysize);
+            return generator.generateKeyPair();
+        } catch (final NoSuchAlgorithmException e) {
+            throw new RuntimeException("Unable to generate RSA keypair", e);
+        }
+    }
+
     public static String twosComplementSha1Digest(byte[] digest) {
         return new BigInteger(digest).toString(16);
     }