apply a couple network patches
This commit is contained in:
parent
f80338725e
commit
1a0280ccc0
666 changed files with 41 additions and 60 deletions
|
@ -1,359 +0,0 @@
|
|||
From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
|
||||
From: Aikar <aikar@aikar.co>
|
||||
Date: Wed, 6 May 2020 04:53:35 -0400
|
||||
Subject: [PATCH] Optimize Network Manager and add advanced packet support
|
||||
|
||||
Adds ability for 1 packet to bundle other packets to follow it
|
||||
Adds ability for a packet to delay sending more packets until a state is ready.
|
||||
|
||||
Removes synchronization from sending packets
|
||||
Removes processing packet queue off of main thread
|
||||
- for the few cases where it is allowed, order is not necessary nor
|
||||
should it even be happening concurrently in first place (handshaking/login/status)
|
||||
|
||||
Ensures packets sent asynchronously are dispatched on main thread
|
||||
|
||||
This helps ensure safety for ProtocolLib as packet listeners
|
||||
are commonly accessing world state. This will allow you to schedule
|
||||
a packet to be sent async, but itll be dispatched sync for packet
|
||||
listeners to process.
|
||||
|
||||
This should solve some deadlock risks
|
||||
|
||||
Also adds Netty Channel Flush Consolidation to reduce the amount of flushing
|
||||
|
||||
Also avoids spamming closed channel exception by rechecking closed state in dispatch
|
||||
and then catch exceptions and close if they fire.
|
||||
|
||||
Part of this commit was authored by: Spottedleaf, sandtechnology
|
||||
|
||||
diff --git a/src/main/java/net/minecraft/network/Connection.java b/src/main/java/net/minecraft/network/Connection.java
|
||||
index e1eea381a2ebcc32424d31c42a45287d2a1cb16a..ca9a36e357e73b1b518ebb5c937314269655de32 100644
|
||||
--- a/src/main/java/net/minecraft/network/Connection.java
|
||||
+++ b/src/main/java/net/minecraft/network/Connection.java
|
||||
@@ -118,6 +118,10 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
|
||||
public int protocolVersion;
|
||||
public java.net.InetSocketAddress virtualHost;
|
||||
private static boolean enableExplicitFlush = Boolean.getBoolean("paper.explicit-flush");
|
||||
+ // Optimize network
|
||||
+ public boolean isPending = true;
|
||||
+ public boolean queueImmunity = false;
|
||||
+ public ConnectionProtocol protocol;
|
||||
// Paper end
|
||||
|
||||
public Connection(PacketFlow side) {
|
||||
@@ -141,6 +145,7 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
|
||||
}
|
||||
|
||||
public void setProtocol(ConnectionProtocol state) {
|
||||
+ protocol = state; // Paper
|
||||
this.channel.attr(Connection.ATTRIBUTE_PROTOCOL).set(state);
|
||||
this.channel.attr(BundlerInfo.BUNDLER_PROVIDER).set(state);
|
||||
this.channel.config().setAutoRead(true);
|
||||
@@ -223,19 +228,88 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
|
||||
Validate.notNull(listener, "packetListener", new Object[0]);
|
||||
this.packetListener = listener;
|
||||
}
|
||||
+ // Paper start
|
||||
+ public @Nullable net.minecraft.server.level.ServerPlayer getPlayer() {
|
||||
+ if (packetListener instanceof net.minecraft.server.network.ServerGamePacketListenerImpl serverGamePacketListener) {
|
||||
+ return serverGamePacketListener.player;
|
||||
+ } else {
|
||||
+ return null;
|
||||
+ }
|
||||
+ }
|
||||
+ private static class InnerUtil { // Attempt to hide these methods from ProtocolLib so it doesn't accidently pick them up.
|
||||
+ private static java.util.List<Packet> buildExtraPackets(Packet packet) {
|
||||
+ java.util.List<Packet> extra = packet.getExtraPackets();
|
||||
+ if (extra == null || extra.isEmpty()) {
|
||||
+ return null;
|
||||
+ }
|
||||
+ java.util.List<Packet> ret = new java.util.ArrayList<>(1 + extra.size());
|
||||
+ buildExtraPackets0(extra, ret);
|
||||
+ return ret;
|
||||
+ }
|
||||
+
|
||||
+ private static void buildExtraPackets0(java.util.List<Packet> extraPackets, java.util.List<Packet> into) {
|
||||
+ for (Packet extra : extraPackets) {
|
||||
+ into.add(extra);
|
||||
+ java.util.List<Packet> extraExtra = extra.getExtraPackets();
|
||||
+ if (extraExtra != null && !extraExtra.isEmpty()) {
|
||||
+ buildExtraPackets0(extraExtra, into);
|
||||
+ }
|
||||
+ }
|
||||
+ }
|
||||
+ // Paper start
|
||||
+ private static boolean canSendImmediate(Connection networkManager, Packet<?> packet) {
|
||||
+ return networkManager.isPending || networkManager.protocol != ConnectionProtocol.PLAY ||
|
||||
+ packet instanceof net.minecraft.network.protocol.game.ClientboundKeepAlivePacket ||
|
||||
+ packet instanceof net.minecraft.network.protocol.game.ClientboundPlayerChatPacket ||
|
||||
+ packet instanceof net.minecraft.network.protocol.game.ClientboundSystemChatPacket ||
|
||||
+ packet instanceof net.minecraft.network.protocol.game.ClientboundCommandSuggestionsPacket ||
|
||||
+ packet instanceof net.minecraft.network.protocol.game.ClientboundSetTitleTextPacket ||
|
||||
+ packet instanceof net.minecraft.network.protocol.game.ClientboundSetSubtitleTextPacket ||
|
||||
+ packet instanceof net.minecraft.network.protocol.game.ClientboundSetActionBarTextPacket ||
|
||||
+ packet instanceof net.minecraft.network.protocol.game.ClientboundSetTitlesAnimationPacket ||
|
||||
+ packet instanceof net.minecraft.network.protocol.game.ClientboundClearTitlesPacket ||
|
||||
+ packet instanceof net.minecraft.network.protocol.game.ClientboundBossEventPacket;
|
||||
+ }
|
||||
+ // Paper end
|
||||
+ }
|
||||
+ // Paper end
|
||||
|
||||
public void send(Packet<?> packet) {
|
||||
this.send(packet, (PacketSendListener) null);
|
||||
}
|
||||
|
||||
public void send(Packet<?> packet, @Nullable PacketSendListener callbacks) {
|
||||
- if (this.isConnected()) {
|
||||
- this.flushQueue();
|
||||
+ // Paper start - handle oversized packets better
|
||||
+ boolean connected = this.isConnected();
|
||||
+ if (!connected && !preparing) {
|
||||
+ return; // Do nothing
|
||||
+ }
|
||||
+ packet.onPacketDispatch(getPlayer());
|
||||
+ if (connected && (InnerUtil.canSendImmediate(this, packet) || (
|
||||
+ io.papermc.paper.util.MCUtil.isMainThread() && packet.isReady() && this.queue.isEmpty() &&
|
||||
+ (packet.getExtraPackets() == null || packet.getExtraPackets().isEmpty())
|
||||
+ ))) {
|
||||
this.sendPacket(packet, callbacks);
|
||||
- } else {
|
||||
- this.queue.add(new Connection.PacketHolder(packet, callbacks));
|
||||
+ return;
|
||||
}
|
||||
+ // write the packets to the queue, then flush - antixray hooks there already
|
||||
+ java.util.List<Packet> extraPackets = InnerUtil.buildExtraPackets(packet);
|
||||
+ boolean hasExtraPackets = extraPackets != null && !extraPackets.isEmpty();
|
||||
+ if (!hasExtraPackets) {
|
||||
+ this.queue.add(new Connection.PacketHolder(packet, callbacks));
|
||||
+ } else {
|
||||
+ java.util.List<Connection.PacketHolder> packets = new java.util.ArrayList<>(1 + extraPackets.size());
|
||||
+ packets.add(new Connection.PacketHolder(packet, null)); // delay the future listener until the end of the extra packets
|
||||
|
||||
+ for (int i = 0, len = extraPackets.size(); i < len;) {
|
||||
+ Packet extra = extraPackets.get(i);
|
||||
+ boolean end = ++i == len;
|
||||
+ packets.add(new Connection.PacketHolder(extra, end ? callbacks : null)); // append listener to the end
|
||||
+ }
|
||||
+ this.queue.addAll(packets); // atomic
|
||||
+ }
|
||||
+ this.flushQueue();
|
||||
+ // Paper end
|
||||
}
|
||||
|
||||
private void sendPacket(Packet<?> packet, @Nullable PacketSendListener callbacks) {
|
||||
@@ -267,6 +341,15 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
|
||||
this.setProtocol(packetState);
|
||||
}
|
||||
|
||||
+ // Paper start
|
||||
+ net.minecraft.server.level.ServerPlayer player = getPlayer();
|
||||
+ if (!isConnected()) {
|
||||
+ packet.onPacketDispatchFinish(player, null);
|
||||
+ return;
|
||||
+ }
|
||||
+
|
||||
+ try {
|
||||
+ // Paper end
|
||||
ChannelFuture channelfuture = this.channel.writeAndFlush(packet);
|
||||
|
||||
if (callbacks != null) {
|
||||
@@ -285,28 +368,72 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
|
||||
|
||||
});
|
||||
}
|
||||
+ // Paper start
|
||||
+ if (packet.hasFinishListener()) {
|
||||
+ channelfuture.addListener((ChannelFutureListener) channelFuture -> packet.onPacketDispatchFinish(player, channelFuture));
|
||||
+ }
|
||||
+ // Paper end
|
||||
|
||||
channelfuture.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
|
||||
+ // Paper start
|
||||
+ } catch (Exception e) {
|
||||
+ LOGGER.error("NetworkException: " + player, e);
|
||||
+ disconnect(Component.translatable("disconnect.genericReason", "Internal Exception: " + e.getMessage()));
|
||||
+ packet.onPacketDispatchFinish(player, null);
|
||||
+ }
|
||||
+ // Paper end
|
||||
}
|
||||
|
||||
private ConnectionProtocol getCurrentProtocol() {
|
||||
return (ConnectionProtocol) this.channel.attr(Connection.ATTRIBUTE_PROTOCOL).get();
|
||||
}
|
||||
|
||||
- private void flushQueue() {
|
||||
+ // Paper start - rewrite this to be safer if ran off main thread
|
||||
+ private boolean flushQueue() { // void -> boolean
|
||||
+ if (!isConnected()) {
|
||||
+ return true;
|
||||
+ }
|
||||
+ if (io.papermc.paper.util.MCUtil.isMainThread()) {
|
||||
+ return processQueue();
|
||||
+ } else if (isPending) {
|
||||
+ // Should only happen during login/status stages
|
||||
+ synchronized (this.queue) {
|
||||
+ return this.processQueue();
|
||||
+ }
|
||||
+ }
|
||||
+ return false;
|
||||
+ }
|
||||
+ private boolean processQueue() {
|
||||
try { // Paper - add pending task queue
|
||||
- if (this.channel != null && this.channel.isOpen()) {
|
||||
- Queue queue = this.queue;
|
||||
+ if (this.queue.isEmpty()) return true;
|
||||
+ // If we are on main, we are safe here in that nothing else should be processing queue off main anymore
|
||||
+ // But if we are not on main due to login/status, the parent is synchronized on packetQueue
|
||||
+ java.util.Iterator<PacketHolder> iterator = this.queue.iterator();
|
||||
+ while (iterator.hasNext()) {
|
||||
+ PacketHolder queued = iterator.next(); // poll -> peek
|
||||
+
|
||||
+ // Fix NPE (Spigot bug caused by handleDisconnection())
|
||||
+ if (queued == null) {
|
||||
+ return true;
|
||||
+ }
|
||||
|
||||
- synchronized (this.queue) {
|
||||
- Connection.PacketHolder networkmanager_queuedpacket;
|
||||
+ // Paper start - checking isConsumed flag and skipping packet sending
|
||||
+ if (queued.isConsumed()) {
|
||||
+ continue;
|
||||
+ }
|
||||
+ // Paper end - checking isConsumed flag and skipping packet sending
|
||||
|
||||
- while ((networkmanager_queuedpacket = (Connection.PacketHolder) this.queue.poll()) != null) {
|
||||
- this.sendPacket(networkmanager_queuedpacket.packet, networkmanager_queuedpacket.listener);
|
||||
+ Packet<?> packet = queued.packet;
|
||||
+ if (!packet.isReady()) {
|
||||
+ return false;
|
||||
+ } else {
|
||||
+ iterator.remove();
|
||||
+ if (queued.tryMarkConsumed()) { // Paper - try to mark isConsumed flag for de-duplicating packet
|
||||
+ this.sendPacket(packet, queued.listener);
|
||||
}
|
||||
-
|
||||
}
|
||||
}
|
||||
+ return true;
|
||||
} finally { // Paper start - add pending task queue
|
||||
Runnable r;
|
||||
while ((r = this.pendingTasks.poll()) != null) {
|
||||
@@ -314,6 +441,7 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
|
||||
}
|
||||
} // Paper end - add pending task queue
|
||||
}
|
||||
+ // Paper end
|
||||
|
||||
public void tick() {
|
||||
this.flushQueue();
|
||||
@@ -350,9 +478,22 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
|
||||
return this.address;
|
||||
}
|
||||
|
||||
+ // Paper start
|
||||
+ public void clearPacketQueue() {
|
||||
+ net.minecraft.server.level.ServerPlayer player = getPlayer();
|
||||
+ queue.forEach(queuedPacket -> {
|
||||
+ Packet<?> packet = queuedPacket.packet;
|
||||
+ if (packet.hasFinishListener()) {
|
||||
+ packet.onPacketDispatchFinish(player, null);
|
||||
+ }
|
||||
+ });
|
||||
+ queue.clear();
|
||||
+ }
|
||||
+ // Paper end
|
||||
public void disconnect(Component disconnectReason) {
|
||||
// Spigot Start
|
||||
this.preparing = false;
|
||||
+ clearPacketQueue(); // Paper
|
||||
// Spigot End
|
||||
if (this.channel.isOpen()) {
|
||||
this.channel.close(); // We can't wait as this may be called from an event loop.
|
||||
@@ -481,7 +622,7 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
|
||||
public void handleDisconnection() {
|
||||
if (this.channel != null && !this.channel.isOpen()) {
|
||||
if (this.disconnectionHandled) {
|
||||
- Connection.LOGGER.warn("handleDisconnection() called twice");
|
||||
+ //Connection.LOGGER.warn("handleDisconnection() called twice"); // Paper - Do not log useless message
|
||||
} else {
|
||||
this.disconnectionHandled = true;
|
||||
if (this.getDisconnectedReason() != null) {
|
||||
@@ -489,7 +630,7 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
|
||||
} else if (this.getPacketListener() != null) {
|
||||
this.getPacketListener().onDisconnect(Component.translatable("multiplayer.disconnect.generic"));
|
||||
}
|
||||
- this.queue.clear(); // Free up packet queue.
|
||||
+ clearPacketQueue(); // Paper
|
||||
// Paper start - Add PlayerConnectionCloseEvent
|
||||
final PacketListener packetListener = this.getPacketListener();
|
||||
if (packetListener instanceof net.minecraft.server.network.ServerGamePacketListenerImpl) {
|
||||
@@ -529,6 +670,18 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
|
||||
@Nullable
|
||||
final PacketSendListener listener;
|
||||
|
||||
+ // Paper start - isConsumed flag for the connection
|
||||
+ private java.util.concurrent.atomic.AtomicBoolean isConsumed = new java.util.concurrent.atomic.AtomicBoolean(false);
|
||||
+
|
||||
+ public boolean tryMarkConsumed() {
|
||||
+ return isConsumed.compareAndSet(false, true);
|
||||
+ }
|
||||
+
|
||||
+ public boolean isConsumed() {
|
||||
+ return isConsumed.get();
|
||||
+ }
|
||||
+ // Paper end - isConsumed flag for the connection
|
||||
+
|
||||
public PacketHolder(Packet<?> packet, @Nullable PacketSendListener callbacks) {
|
||||
this.packet = packet;
|
||||
this.listener = callbacks;
|
||||
diff --git a/src/main/java/net/minecraft/network/protocol/Packet.java b/src/main/java/net/minecraft/network/protocol/Packet.java
|
||||
index 74bfe0d3942259c45702b099efdc4e101a4e3022..e8fcd56906d26f6dc87959e32c4c7c78cfea9658 100644
|
||||
--- a/src/main/java/net/minecraft/network/protocol/Packet.java
|
||||
+++ b/src/main/java/net/minecraft/network/protocol/Packet.java
|
||||
@@ -9,6 +9,19 @@ public interface Packet<T extends PacketListener> {
|
||||
void handle(T listener);
|
||||
|
||||
// Paper start
|
||||
+ /**
|
||||
+ * @param player Null if not at PLAY stage yet
|
||||
+ */
|
||||
+ default void onPacketDispatch(@javax.annotation.Nullable net.minecraft.server.level.ServerPlayer player) {}
|
||||
+
|
||||
+ /**
|
||||
+ * @param player Null if not at PLAY stage yet
|
||||
+ * @param future Can be null if packet was cancelled
|
||||
+ */
|
||||
+ default void onPacketDispatchFinish(@javax.annotation.Nullable net.minecraft.server.level.ServerPlayer player, @javax.annotation.Nullable io.netty.channel.ChannelFuture future) {}
|
||||
+ default boolean hasFinishListener() { return false; }
|
||||
+ default boolean isReady() { return true; }
|
||||
+ default java.util.List<Packet> getExtraPackets() { return null; }
|
||||
default boolean packetTooLarge(net.minecraft.network.Connection manager) {
|
||||
return false;
|
||||
}
|
||||
diff --git a/src/main/java/net/minecraft/server/network/ServerConnectionListener.java b/src/main/java/net/minecraft/server/network/ServerConnectionListener.java
|
||||
index 776528e50a5abc0e02d9de99231fb47352aa4f43..fbf375534e2b8bd6ef052c4625764f4f8feb2ed6 100644
|
||||
--- a/src/main/java/net/minecraft/server/network/ServerConnectionListener.java
|
||||
+++ b/src/main/java/net/minecraft/server/network/ServerConnectionListener.java
|
||||
@@ -62,10 +62,12 @@ public class ServerConnectionListener {
|
||||
final List<Connection> connections = Collections.synchronizedList(Lists.newArrayList());
|
||||
// Paper start - prevent blocking on adding a new network manager while the server is ticking
|
||||
private final java.util.Queue<Connection> pending = new java.util.concurrent.ConcurrentLinkedQueue<>();
|
||||
+ private static final boolean disableFlushConsolidation = Boolean.getBoolean("Paper.disableFlushConsolidate"); // Paper
|
||||
private final void addPending() {
|
||||
Connection manager = null;
|
||||
while ((manager = pending.poll()) != null) {
|
||||
connections.add(manager);
|
||||
+ manager.isPending = false;
|
||||
}
|
||||
}
|
||||
// Paper end
|
||||
@@ -100,6 +102,7 @@ public class ServerConnectionListener {
|
||||
;
|
||||
}
|
||||
|
||||
+ if (!disableFlushConsolidation) channel.pipeline().addFirst(new io.netty.handler.flush.FlushConsolidationHandler()); // Paper
|
||||
ChannelPipeline channelpipeline = channel.pipeline().addLast("timeout", new ReadTimeoutHandler(30)).addLast("legacy_query", new LegacyQueryHandler(ServerConnectionListener.this));
|
||||
|
||||
Connection.configureSerialization(channelpipeline, PacketFlow.SERVERBOUND);
|
|
@ -1,150 +0,0 @@
|
|||
From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
|
||||
From: Spottedleaf <Spottedleaf@users.noreply.github.com>
|
||||
Date: Sat, 4 Apr 2020 15:27:44 -0700
|
||||
Subject: [PATCH] Allow controlled flushing for network manager
|
||||
|
||||
Only make one flush call when emptying the packet queue too
|
||||
|
||||
This patch will be used to optimise out flush calls in later
|
||||
patches.
|
||||
|
||||
diff --git a/src/main/java/net/minecraft/network/Connection.java b/src/main/java/net/minecraft/network/Connection.java
|
||||
index 1404389f16cb6ed8bb1b2dad35e364b1428810a4..d7dd0b806d8d8e3f04f165a2b29097a5c1bcd199 100644
|
||||
--- a/src/main/java/net/minecraft/network/Connection.java
|
||||
+++ b/src/main/java/net/minecraft/network/Connection.java
|
||||
@@ -124,6 +124,39 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
|
||||
public ConnectionProtocol protocol;
|
||||
// Paper end
|
||||
|
||||
+ // Paper start - allow controlled flushing
|
||||
+ volatile boolean canFlush = true;
|
||||
+ private final java.util.concurrent.atomic.AtomicInteger packetWrites = new java.util.concurrent.atomic.AtomicInteger();
|
||||
+ private int flushPacketsStart;
|
||||
+ private final Object flushLock = new Object();
|
||||
+
|
||||
+ public void disableAutomaticFlush() {
|
||||
+ synchronized (this.flushLock) {
|
||||
+ this.flushPacketsStart = this.packetWrites.get(); // must be volatile and before canFlush = false
|
||||
+ this.canFlush = false;
|
||||
+ }
|
||||
+ }
|
||||
+
|
||||
+ public void enableAutomaticFlush() {
|
||||
+ synchronized (this.flushLock) {
|
||||
+ this.canFlush = true;
|
||||
+ if (this.packetWrites.get() != this.flushPacketsStart) { // must be after canFlush = true
|
||||
+ this.flush(); // only make the flush call if we need to
|
||||
+ }
|
||||
+ }
|
||||
+ }
|
||||
+
|
||||
+ private final void flush() {
|
||||
+ if (this.channel.eventLoop().inEventLoop()) {
|
||||
+ this.channel.flush();
|
||||
+ } else {
|
||||
+ this.channel.eventLoop().execute(() -> {
|
||||
+ this.channel.flush();
|
||||
+ });
|
||||
+ }
|
||||
+ }
|
||||
+ // Paper end - allow controlled flushing
|
||||
+
|
||||
public Connection(PacketFlow side) {
|
||||
this.receiving = side;
|
||||
}
|
||||
@@ -292,7 +325,7 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
|
||||
io.papermc.paper.util.MCUtil.isMainThread() && packet.isReady() && this.queue.isEmpty() &&
|
||||
(packet.getExtraPackets() == null || packet.getExtraPackets().isEmpty())
|
||||
))) {
|
||||
- this.sendPacket(packet, callbacks);
|
||||
+ this.sendPacket(packet, callbacks, null); // Paper
|
||||
return;
|
||||
}
|
||||
// write the packets to the queue, then flush - antixray hooks there already
|
||||
@@ -316,6 +349,14 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
|
||||
}
|
||||
|
||||
private void sendPacket(Packet<?> packet, @Nullable PacketSendListener callbacks) {
|
||||
+ // Paper start - add flush parameter
|
||||
+ this.sendPacket(packet, callbacks, Boolean.TRUE);
|
||||
+ }
|
||||
+ private void sendPacket(Packet<?> packet, @Nullable PacketSendListener callbacks, Boolean flushConditional) {
|
||||
+ this.packetWrites.getAndIncrement(); // must be befeore using canFlush
|
||||
+ boolean effectiveFlush = flushConditional == null ? this.canFlush : flushConditional.booleanValue();
|
||||
+ final boolean flush = effectiveFlush || packet instanceof net.minecraft.network.protocol.game.ClientboundKeepAlivePacket || packet instanceof ClientboundDisconnectPacket; // no delay for certain packets
|
||||
+ // Paper end - add flush parameter
|
||||
ConnectionProtocol enumprotocol = ConnectionProtocol.getProtocolForPacket(packet);
|
||||
ConnectionProtocol enumprotocol1 = this.getCurrentProtocol();
|
||||
|
||||
@@ -330,16 +371,21 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
|
||||
}
|
||||
|
||||
if (this.channel.eventLoop().inEventLoop()) {
|
||||
- this.doSendPacket(packet, callbacks, enumprotocol, enumprotocol1);
|
||||
+ this.doSendPacket(packet, callbacks, enumprotocol, enumprotocol1, flush); // Paper
|
||||
} else {
|
||||
this.channel.eventLoop().execute(() -> {
|
||||
- this.doSendPacket(packet, callbacks, enumprotocol, enumprotocol1);
|
||||
+ this.doSendPacket(packet, callbacks, enumprotocol, enumprotocol1, flush); // Paper
|
||||
});
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private void doSendPacket(Packet<?> packet, @Nullable PacketSendListener callbacks, ConnectionProtocol packetState, ConnectionProtocol currentState) {
|
||||
+ // Paper start - add flush parameter
|
||||
+ this.doSendPacket(packet, callbacks, packetState, currentState, true);
|
||||
+ }
|
||||
+ private void doSendPacket(Packet<?> packet, @Nullable PacketSendListener callbacks, ConnectionProtocol packetState, ConnectionProtocol currentState, boolean flush) {
|
||||
+ // Paper end - add flush parameter
|
||||
if (packetState != currentState) {
|
||||
this.setProtocol(packetState);
|
||||
}
|
||||
@@ -353,7 +399,7 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
|
||||
|
||||
try {
|
||||
// Paper end
|
||||
- ChannelFuture channelfuture = this.channel.writeAndFlush(packet);
|
||||
+ ChannelFuture channelfuture = flush ? this.channel.writeAndFlush(packet) : this.channel.write(packet); // Paper - add flush parameter
|
||||
|
||||
if (callbacks != null) {
|
||||
channelfuture.addListener((future) -> {
|
||||
@@ -409,6 +455,10 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
|
||||
private boolean processQueue() {
|
||||
try { // Paper - add pending task queue
|
||||
if (this.queue.isEmpty()) return true;
|
||||
+ // Paper start - make only one flush call per sendPacketQueue() call
|
||||
+ final boolean needsFlush = this.canFlush;
|
||||
+ boolean hasWrotePacket = false;
|
||||
+ // Paper end - make only one flush call per sendPacketQueue() call
|
||||
// If we are on main, we are safe here in that nothing else should be processing queue off main anymore
|
||||
// But if we are not on main due to login/status, the parent is synchronized on packetQueue
|
||||
java.util.Iterator<PacketHolder> iterator = this.queue.iterator();
|
||||
@@ -416,7 +466,7 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
|
||||
PacketHolder queued = iterator.next(); // poll -> peek
|
||||
|
||||
// Fix NPE (Spigot bug caused by handleDisconnection())
|
||||
- if (queued == null) {
|
||||
+ if (false && queued == null) { // Paper - diff on change, this logic is redundant: iterator guarantees ret of an element - on change, hook the flush logic here
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -428,11 +478,17 @@ public class Connection extends SimpleChannelInboundHandler<Packet<?>> {
|
||||
|
||||
Packet<?> packet = queued.packet;
|
||||
if (!packet.isReady()) {
|
||||
+ // Paper start - make only one flush call per sendPacketQueue() call
|
||||
+ if (hasWrotePacket && (needsFlush || this.canFlush)) {
|
||||
+ this.flush();
|
||||
+ }
|
||||
+ // Paper end - make only one flush call per sendPacketQueue() call
|
||||
return false;
|
||||
} else {
|
||||
iterator.remove();
|
||||
if (queued.tryMarkConsumed()) { // Paper - try to mark isConsumed flag for de-duplicating packet
|
||||
- this.sendPacket(packet, queued.listener);
|
||||
+ this.sendPacket(packet, queued.listener, (!iterator.hasNext() && (needsFlush || this.canFlush)) ? Boolean.TRUE : Boolean.FALSE); // Paper - make only one flush call per sendPacketQueue() call
|
||||
+ hasWrotePacket = true; // Paper - make only one flush call per sendPacketQueue() call
|
||||
}
|
||||
}
|
||||
}
|
Loading…
Add table
Add a link
Reference in a new issue