Merge branch 'dev/1.1.0' into experiment/io_uring

build.gradle

@@ -16,7 +16,7 @@ allprojects {
     apply plugin: "com.github.spotbugs"
 
     group 'com.velocitypowered'
-    version '1.1.3-SNAPSHOT'
+    version '1.1.4-SNAPSHOT'
 
     ext {
         // dependency versions
@@ -26,7 +26,7 @@ allprojects {
         junitVersion = '5.7.0'
         slf4jVersion = '1.7.30'
         log4jVersion = '2.13.3'
-        nettyVersion = '4.1.56.Final'
+        nettyVersion = '4.1.58.Final'
         guavaVersion = '25.1-jre'
         checkerFrameworkVersion = '3.6.1'
         configurateVersion = '3.7.1'

proxy/src/main/java/com/velocitypowered/proxy/command/builtin/VelocityCommand.java

@@ -209,7 +209,7 @@ public class VelocityCommand implements SimpleCommand {
           .append(Component.text(version.getVersion()).decoration(TextDecoration.BOLD, false))
           .build();
       TextComponent copyright = Component
-          .text("Copyright 2018-2020 " + version.getVendor() + ". " + version.getName()
+          .text("Copyright 2018-2021 " + version.getVendor() + ". " + version.getName()
               + " is freely licensed under the terms of the MIT License.");
       source.sendMessage(Identity.nil(), velocity);
       source.sendMessage(Identity.nil(), copyright);

proxy/src/main/java/com/velocitypowered/proxy/config/VelocityConfiguration.java

@@ -646,7 +646,11 @@ public class VelocityConfiguration implements ProxyConfig {
         this.loginRatelimit = config.getIntOrElse("login-ratelimit", 3000);
         this.connectionTimeout = config.getIntOrElse("connection-timeout", 5000);
         this.readTimeout = config.getIntOrElse("read-timeout", 30000);
+        if (config.contains("haproxy-protocol")) {
+          this.proxyProtocol = config.getOrElse("haproxy-protocol", false);
+        } else {
           this.proxyProtocol = config.getOrElse("proxy-protocol", false);
+        }
         this.tcpFastOpen = config.getOrElse("tcp-fast-open", false);
         this.bungeePluginMessageChannel = config.getOrElse("bungee-plugin-message-channel", true);
         this.showPingRequests = config.getOrElse("show-ping-requests", false);

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

@@ -195,6 +195,8 @@ public class MinecraftConnection extends ChannelInboundHandlerAdapter {
   public void write(Object msg) {
     if (channel.isActive()) {
       channel.writeAndFlush(msg, channel.voidPromise());
+    } else {
+      ReferenceCountUtil.release(msg);
     }
   }
 
@@ -205,6 +207,8 @@ public class MinecraftConnection extends ChannelInboundHandlerAdapter {
   public void delayedWrite(Object msg) {
     if (channel.isActive()) {
       channel.write(msg, channel.voidPromise());
+    } else {
+      ReferenceCountUtil.release(msg);
     }
   }
 

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

@@ -28,6 +28,7 @@ import com.velocitypowered.proxy.protocol.util.PluginMessageUtil;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
 import io.netty.handler.timeout.ReadTimeoutException;
 import java.util.Collection;
 import org.apache.logging.log4j.LogManager;
@@ -36,6 +37,8 @@ import org.apache.logging.log4j.Logger;
 public class BackendPlaySessionHandler implements MinecraftSessionHandler {
 
   private static final Logger logger = LogManager.getLogger(BackendPlaySessionHandler.class);
+  private static final boolean BACKPRESSURE_LOG = Boolean
+      .getBoolean("velocity.log-server-backpressure");
   private final VelocityServer server;
   private final VelocityServerConnection serverConn;
   private final ClientPlaySessionHandler playerSessionHandler;
@@ -284,4 +287,20 @@ public class BackendPlaySessionHandler implements MinecraftSessionHandler {
       }
     }
   }
+
+  @Override
+  public void writabilityChanged() {
+    Channel serverChan = serverConn.ensureConnected().getChannel();
+    boolean writable = serverChan.isWritable();
+
+    if (BACKPRESSURE_LOG) {
+      if (writable) {
+        logger.info("{} is not writable, not auto-reading player connection data", this.serverConn);
+      } else {
+        logger.info("{} is writable, will auto-read player connection data", this.serverConn);
+      }
+    }
+
+    playerConnection.setAutoReading(writable);
+  }
 }

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

@@ -236,45 +236,34 @@ public class BungeeCordMessageResponder {
     });
   }
 
-  private ByteBuf prepareForwardMessage(ByteBufDataInput in) {
-    String channel = in.readUTF();
-    short messageLength = in.readShort();
-
-    ByteBuf buf = Unpooled.buffer();
-    ByteBufDataOutput forwarded = new ByteBufDataOutput(buf);
-    forwarded.writeUTF(channel);
-    forwarded.writeShort(messageLength);
-    buf.writeBytes(in.unwrap().readSlice(messageLength));
-    return buf;
-  }
-
   private void processForwardToPlayer(ByteBufDataInput in) {
-    proxy.getPlayer(in.readUTF())
-        .flatMap(Player::getCurrentServer)
-        .ifPresent(server -> sendServerResponse(player, prepareForwardMessage(in)));
+    Optional<Player> player = proxy.getPlayer(in.readUTF());
+    if (player.isPresent()) {
+      ByteBuf toForward = in.unwrap().copy();
+      sendServerResponse((ConnectedPlayer) player.get(), toForward);
+    }
   }
 
   private void processForwardToServer(ByteBufDataInput in) {
     String target = in.readUTF();
-    ByteBuf toForward = prepareForwardMessage(in);
+    ByteBuf toForward = in.unwrap().copy();
     if (target.equals("ALL")) {
-      ByteBuf unreleasableForward = Unpooled.unreleasableBuffer(toForward);
       try {
         for (RegisteredServer rs : proxy.getAllServers()) {
-          ((VelocityRegisteredServer) rs).sendPluginMessage(LEGACY_CHANNEL, unreleasableForward);
+          ((VelocityRegisteredServer) rs).sendPluginMessage(LEGACY_CHANNEL,
+              toForward.retainedSlice());
         }
       } finally {
         toForward.release();
       }
     } else {
-      proxy.getServer(target).ifPresent(rs -> ((VelocityRegisteredServer) rs)
-          .sendPluginMessage(LEGACY_CHANNEL, toForward));
+      Optional<RegisteredServer> server = proxy.getServer(target);
+      if (server.isPresent()) {
+        ((VelocityRegisteredServer) server.get()).sendPluginMessage(LEGACY_CHANNEL, toForward);
+      } else {
+        toForward.release();
       }
     }
-
-  // Note: this method will always release the buffer!
-  private void sendResponseOnConnection(ByteBuf buf) {
-    sendServerResponse(this.player, buf);
   }
 
   static String getBungeeCordChannel(ProtocolVersion version) {
@@ -282,29 +271,17 @@ public class BungeeCordMessageResponder {
         : LEGACY_CHANNEL.getId();
   }
 
+  // Note: this method will always release the buffer!
+  private void sendResponseOnConnection(ByteBuf buf) {
+    sendServerResponse(this.player, buf);
+  }
+
   // Note: this method will always release the buffer!
   private static void sendServerResponse(ConnectedPlayer player, ByteBuf buf) {
     MinecraftConnection serverConnection = player.ensureAndGetCurrentServer().ensureConnected();
     String chan = getBungeeCordChannel(serverConnection.getProtocolVersion());
-
-    PluginMessage msg = null;
-    boolean released = false;
-
-    try {
-      VelocityServerConnection vsc = player.getConnectedServer();
-      if (vsc == null) {
-        return;
-      }
-
-      MinecraftConnection serverConn = vsc.ensureConnected();
-      msg = new PluginMessage(chan, buf);
-      serverConn.write(msg);
-      released = true;
-    } finally {
-      if (!released && msg != null) {
-        msg.release();
-      }
-    }
+    PluginMessage msg = new PluginMessage(chan, buf);
+    serverConnection.write(msg);
   }
 
   boolean process(PluginMessage message) {

proxy/src/main/java/com/velocitypowered/proxy/connection/client/ClientPlaySessionHandler.java

@@ -300,7 +300,7 @@ public class ClientPlaySessionHandler implements MinecraftSessionHandler {
     boolean writable = player.getConnection().getChannel().isWritable();
 
     if (!writable) {
-      // We might have packets queued for the server, so flush them now to free up memory.
+      // We might have packets queued from the server, so flush them now to free up memory.
       player.getConnection().flush();
     }
 

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

@@ -20,7 +20,7 @@ public class MinecraftCompressEncoder extends MessageToByteEncoder<ByteBuf> {
   @Override
   protected void encode(ChannelHandlerContext ctx, ByteBuf msg, ByteBuf out) throws Exception {
     int uncompressed = msg.readableBytes();
-    if (uncompressed <= threshold) {
+    if (uncompressed < threshold) {
       // Under the threshold, there is nothing to do.
       ProtocolUtils.writeVarInt(out, 0);
       out.writeBytes(msg);

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

@@ -5,7 +5,7 @@ import com.velocitypowered.api.network.ProtocolVersion;
 import com.velocitypowered.proxy.protocol.MinecraftPacket;
 import com.velocitypowered.proxy.protocol.ProtocolUtils;
 import com.velocitypowered.proxy.protocol.StateRegistry;
-import com.velocitypowered.proxy.util.except.QuietDecoderException;
+import com.velocitypowered.proxy.util.except.QuietRuntimeException;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
@@ -14,8 +14,8 @@ import io.netty.handler.codec.CorruptedFrameException;
 public class MinecraftDecoder extends ChannelInboundHandlerAdapter {
 
   public static final boolean DEBUG = Boolean.getBoolean("velocity.packet-decode-logging");
-  private static final QuietDecoderException DECODE_FAILED =
-      new QuietDecoderException("A packet did not decode successfully (invalid data). If you are a "
+  private static final QuietRuntimeException DECODE_FAILED =
+      new QuietRuntimeException("A packet did not decode successfully (invalid data). If you are a "
           + "developer, launch Velocity with -Dvelocity.packet-decode-logging=true to see more.");
 
   private final ProtocolUtils.Direction direction;

proxy/src/main/java/com/velocitypowered/proxy/scheduler/VelocityScheduler.java

@@ -7,6 +7,7 @@ import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Multimap;
 import com.google.common.collect.Multimaps;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.velocitypowered.api.plugin.PluginContainer;
 import com.velocitypowered.api.plugin.PluginManager;
 import com.velocitypowered.api.scheduler.ScheduledTask;
 import com.velocitypowered.api.scheduler.Scheduler;
@@ -183,8 +184,18 @@ public class VelocityScheduler implements Scheduler {
         currentTaskThread = Thread.currentThread();
         try {
           runnable.run();
-        } catch (Exception e) {
-          Log.logger.error("Exception in task {} by plugin {}", runnable, plugin, e);
+        } catch (Throwable e) {
+          //noinspection ConstantConditions
+          if (e instanceof InterruptedException) {
+            Thread.currentThread().interrupt();
+          } else {
+            String friendlyPluginName = pluginManager.fromInstance(plugin)
+                .map(container -> container.getDescription().getName()
+                      .orElse(container.getDescription().getId()))
+                .orElse("UNKNOWN");
+            Log.logger.error("Exception in task {} by plugin {}", runnable, friendlyPluginName,
+                e);
+          }
         } finally {
           if (repeat == 0) {
             onFinish();

proxy/src/main/java/com/velocitypowered/proxy/server/VelocityRegisteredServer.java

@@ -125,7 +125,9 @@ public class VelocityRegisteredServer implements RegisteredServer, ForwardingAud
   }
 
   /**
-   * Sends a plugin message to the server through this connection.
+   * Sends a plugin message to the server through this connection. The message will be released
+   * afterwards.
+   *
    * @param identifier the channel ID to use
    * @param data the data
    * @return whether or not the message was sent
@@ -133,11 +135,12 @@ public class VelocityRegisteredServer implements RegisteredServer, ForwardingAud
   public boolean sendPluginMessage(ChannelIdentifier identifier, ByteBuf data) {
     for (ConnectedPlayer player : players.values()) {
       VelocityServerConnection connection = player.getConnectedServer();
-      if (connection != null && connection.getServerInfo().equals(serverInfo)) {
+      if (connection != null && connection.getServer() == this) {
         return connection.sendPluginMessage(identifier, data);
       }
     }
 
+    data.release();
     return false;
   }
 

proxy/src/main/resources/default-velocity.toml

@@ -105,8 +105,9 @@ connection-timeout = 5000
 # Specify a read timeout for connections here. The default is 30 seconds.
 read-timeout = 30000
 
-# Enables compatibility with HAProxy.
-proxy-protocol = false
+# Enables compatibility with HAProxy's PROXY protocol. If you don't know what this is for, then
+# don't enable it.
+haproxy-protocol = false
 
 # Enables TCP fast open support on the proxy. Requires the proxy to run on Linux.
 tcp-fast-open = false