Skip to content

Commit f846035

Browse files
AlexProgrammerDEonebeastchris
AlexProgrammerDE
and
onebeastchris
authored
Split incoming/outgoing packet registry, transition protocol states correctly (#841)
* Initial code changes * Make it compile * Small inlining * Make less detectable by anticheats and fix keepalive during configuration * Fix keepalive edge case * Properly switch inbound protocol in server listener * Add flow control * Make toggling automatic keepalive work in another way * Remove ping pong packets again * Address review * Handle keepalive in configuration * Only spawn keepalive after login is acknowledged * Prevent very unlikely race conditions with keepalive being switched during a task * Add debug log for packet serialization and state switching * Add one more debug print * Update protocol/src/main/java/org/geysermc/mcprotocollib/network/Session.java Co-authored-by: chris <[email protected]> * Update protocol/src/main/java/org/geysermc/mcprotocollib/protocol/MinecraftProtocol.java Co-authored-by: chris <[email protected]> * Update protocol/src/main/java/org/geysermc/mcprotocollib/protocol/MinecraftProtocol.java Co-authored-by: chris <[email protected]> * Mark packet as nonnull * Fix outbound writing race conditions * Ensure packets are always sent on the event loop This replicates the same approach Mojang uses in their networking code. * Reduce log verbosity * Put errors into debug * Update protocol/src/main/java/org/geysermc/mcprotocollib/network/tcp/TcpClientSession.java Co-authored-by: chris <[email protected]> * Add comment to always running in event loop * Handle auto read earlier to prevent race conditions * Make instance dynamic * Revert "Make instance dynamic" This reverts commit 7f8affb. * Make flush packet priority * Do not hide original line that is the cause of the exception * Cancel packet using exception rather than return * Properly iterate through parents * Set log level to debug for unit tests * Revert "Properly iterate through parents" This reverts commit 4e2b64d. * Revert "Cancel packet using exception rather than return" This reverts commit 6507e77. * Add write length filter * Reuse bytebuf for fake flush to avoid unnecessary allocations * Make tests happy * Remake dropping packets * Update protocol/src/main/java/org/geysermc/mcprotocollib/network/tcp/TcpServer.java Co-authored-by: chris <[email protected]> * Fix space * Rename to flush packet * Add mojmap reference * Share keepalive code * Fix compilation * Revert a tiny bit closer to vanilla * Inline lambda * Inherit annotation * Inherit annotation 2 * Use checkerframework annotation * Fixup grammar slightly * Add reset states method * Add log marker for packet logging --------- Co-authored-by: chris <[email protected]>
1 parent b2c9268 commit f846035

File tree

22 files changed

+447
-160
lines changed

22 files changed

+447
-160
lines changed

‎example/src/main/java/org/geysermc/mcprotocollib/network/example/TestProtocol.java‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -82,7 +82,12 @@ public void newServerSession(Server server, Session session) {
8282
}
8383

8484
@Override
85-
public PacketRegistry getPacketRegistry() {
85+
public PacketRegistry getInboundPacketRegistry() {
86+
return registry;
87+
}
88+
89+
@Override
90+
public PacketRegistry getOutboundPacketRegistry() {
8691
return registry;
8792
}
8893
}

‎example/src/main/java/org/geysermc/mcprotocollib/protocol/example/MinecraftProtocolTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -135,7 +135,7 @@ public void packetReceived(Session session, Packet packet) {
135135
@Override
136136
public void sessionRemoved(SessionRemovedEvent event) {
137137
MinecraftProtocol protocol = (MinecraftProtocol) event.getSession().getPacketProtocol();
138-
if (protocol.getState() == ProtocolState.GAME) {
138+
if (protocol.getOutboundState() == ProtocolState.GAME) {
139139
log.info("Closing server.");
140140
event.getServer().close(false);
141141
}

‎protocol/src/main/java/org/geysermc/mcprotocollib/network/Session.java‎

Lines changed: 57 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package org.geysermc.mcprotocollib.network;
22

3+
import io.netty.channel.Channel;
34
import net.kyori.adventure.text.Component;
45
import org.checkerframework.checker.nullness.qual.NonNull;
56
import org.checkerframework.checker.nullness.qual.Nullable;
@@ -10,6 +11,7 @@
1011
import org.geysermc.mcprotocollib.network.event.session.SessionListener;
1112
import org.geysermc.mcprotocollib.network.packet.Packet;
1213
import org.geysermc.mcprotocollib.network.packet.PacketProtocol;
14+
import org.geysermc.mcprotocollib.network.tcp.FlushHandler;
1315

1416
import java.net.SocketAddress;
1517
import java.util.List;
@@ -212,7 +214,17 @@ public interface Session {
212214
*
213215
* @param packet Packet to send.
214216
*/
215-
void send(Packet packet);
217+
default void send(@NonNull Packet packet) {
218+
this.send(packet, null);
219+
}
220+
221+
/**
222+
* Sends a packet and runs the specified callback when the packet has been sent.
223+
*
224+
* @param packet Packet to send.
225+
* @param onSent Callback to run when the packet has been sent.
226+
*/
227+
void send(@NonNull Packet packet, @Nullable Runnable onSent);
216228

217229
/**
218230
* Disconnects the session.
@@ -255,4 +267,48 @@ default void disconnect(@NonNull Component reason) {
255267
* @param cause Throwable responsible for disconnecting.
256268
*/
257269
void disconnect(@NonNull Component reason, @Nullable Throwable cause);
270+
271+
/**
272+
* Auto read in netty means that the server is automatically reading from the channel.
273+
* Turning it off means that we won't get more packets being decoded until we turn it back on.
274+
* We use this to hold off on reading packets until we are ready to process them.
275+
* For example this is used for switching inbound states with {@link #switchInboundState(Runnable)}.
276+
*
277+
* @param autoRead Whether to enable auto read.
278+
* Default is true.
279+
*/
280+
void setAutoRead(boolean autoRead);
281+
282+
/**
283+
* Returns the underlying netty channel of this session.
284+
*
285+
* @return The netty channel
286+
*/
287+
Channel getChannel();
288+
289+
/**
290+
* Changes the inbound state of the session and then re-enables auto read.
291+
* This is used after a terminal packet was handled and the session is ready to receive more packets in the new state.
292+
*
293+
* @param switcher The runnable that switches the inbound state.
294+
*/
295+
default void switchInboundState(Runnable switcher) {
296+
switcher.run();
297+
298+
// We switched to the new inbound state
299+
// we can start reading again
300+
setAutoRead(true);
301+
}
302+
303+
/**
304+
* Flushes all packets that are due to be sent and changes the outbound state of the session.
305+
* This makes sure no other threads have scheduled packets to be sent.
306+
*
307+
* @param switcher The runnable that switches the outbound state.
308+
*/
309+
default void switchOutboundState(Runnable switcher) {
310+
getChannel().writeAndFlush(FlushHandler.FLUSH_PACKET).syncUninterruptibly();
311+
312+
switcher.run();
313+
}
258314
}

‎protocol/src/main/java/org/geysermc/mcprotocollib/network/packet/Packet.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package org.geysermc.mcprotocollib.network.packet;
22

33
import io.netty.buffer.ByteBuf;
4+
import org.geysermc.mcprotocollib.network.Session;
45

56
/**
67
* A network packet. Any given packet must have a constructor that takes in a {@link ByteBuf}.
@@ -17,4 +18,14 @@ public interface Packet {
1718
default boolean isPriority() {
1819
return false;
1920
}
21+
22+
/**
23+
* Returns whether the packet is terminal. If true, this should be the last packet sent inside a protocol state.
24+
* Subsequently, {@link Session#setAutoRead(boolean)} should be disabled when a terminal packet is received, until the session has switched into a new state and is ready to receive more packets.
25+
*
26+
* @return Whether the packet is terminal.
27+
*/
28+
default boolean isTerminal() {
29+
return false;
30+
}
2031
}

‎protocol/src/main/java/org/geysermc/mcprotocollib/network/packet/PacketProtocol.java‎

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -49,9 +49,16 @@ public abstract class PacketProtocol {
4949
public abstract void newServerSession(Server server, Session session);
5050

5151
/**
52-
* Gets the packet registry for this protocol.
52+
* Gets the inbound packet registry for this protocol.
5353
*
54-
* @return The protocol's packet registry.
54+
* @return The protocol's inbound packet registry.
5555
*/
56-
public abstract PacketRegistry getPacketRegistry();
56+
public abstract PacketRegistry getInboundPacketRegistry();
57+
58+
/**
59+
* Gets the outbound packet registry for this protocol.
60+
*
61+
* @return The protocol's outbound packet registry.
62+
*/
63+
public abstract PacketRegistry getOutboundPacketRegistry();
5764
}

‎protocol/src/main/java/org/geysermc/mcprotocollib/network/tcp/FlushHandler.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
package org.geysermc.mcprotocollib.network.tcp;
2+
3+
import io.netty.channel.ChannelHandlerContext;
4+
import io.netty.channel.ChannelOutboundHandlerAdapter;
5+
import io.netty.channel.ChannelPromise;
6+
7+
/**
8+
* Sending a {@link FlushPacket} will ensure all before were sent.
9+
* This handler makes sure it's dropped before it reaches the encoder.
10+
* This logic is similar to the Minecraft UnconfiguredPipelineHandler.OutboundConfigurationTask.
11+
*/
12+
public class FlushHandler extends ChannelOutboundHandlerAdapter {
13+
public static final FlushPacket FLUSH_PACKET = new FlushPacket();
14+
15+
@Override
16+
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
17+
if (msg == FLUSH_PACKET) {
18+
promise.setSuccess();
19+
} else {
20+
super.write(ctx, msg, promise);
21+
}
22+
}
23+
24+
public static class FlushPacket {
25+
private FlushPacket() {
26+
}
27+
}
28+
}

‎protocol/src/main/java/org/geysermc/mcprotocollib/network/tcp/TcpClientSession.java‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import io.netty.resolver.dns.DnsNameResolver;
2828
import io.netty.resolver.dns.DnsNameResolverBuilder;
2929
import io.netty.util.concurrent.DefaultThreadFactory;
30+
import org.checkerframework.checker.nullness.qual.NonNull;
3031
import org.geysermc.mcprotocollib.network.BuiltinFlags;
3132
import org.geysermc.mcprotocollib.network.ProxyInfo;
3233
import org.geysermc.mcprotocollib.network.codec.PacketCodecHelper;
@@ -100,7 +101,7 @@ public void connect(boolean wait, boolean transferring) {
100101
.localAddress(bindAddress, bindPort)
101102
.handler(new ChannelInitializer<>() {
102103
@Override
103-
public void initChannel(Channel channel) {
104+
public void initChannel(@NonNull Channel channel) {
104105
PacketProtocol protocol = getPacketProtocol();
105106
protocol.newClientSession(TcpClientSession.this, transferring);
106107

@@ -117,7 +118,9 @@ public void initChannel(Channel channel) {
117118
pipeline.addLast("sizer", new TcpPacketSizer(protocol.getPacketHeader(), getCodecHelper()));
118119
pipeline.addLast("compression", new TcpPacketCompression(getCodecHelper()));
119120

121+
pipeline.addLast("flow-control", new TcpFlowControlHandler());
120122
pipeline.addLast("codec", new TcpPacketCodec(TcpClientSession.this, true));
123+
pipeline.addLast("flush-handler", new FlushHandler());
121124
pipeline.addLast("manager", TcpClientSession.this);
122125
}
123126
});
@@ -246,9 +249,7 @@ private void initializeHAProxySupport(Channel channel) {
246249
HAProxyProtocolVersion.V2, HAProxyCommand.PROXY, proxiedProtocol,
247250
clientAddress.getAddress().getHostAddress(), remoteAddress.getAddress().getHostAddress(),
248251
clientAddress.getPort(), remoteAddress.getPort()
249-
)).addListener(future -> {
250-
channel.pipeline().remove("proxy-protocol-encoder");
251-
});
252+
)).addListener(future -> channel.pipeline().remove("proxy-protocol-encoder"));
252253
}
253254

254255
private static void createTcpEventLoopGroup() {

‎protocol/src/main/java/org/geysermc/mcprotocollib/network/tcp/TcpFlowControlHandler.java‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
package org.geysermc.mcprotocollib.network.tcp;
2+
3+
import io.netty.channel.ChannelHandlerContext;
4+
import io.netty.handler.flow.FlowControlHandler;
5+
6+
/**
7+
* A flow control handler for TCP connections.
8+
* When auto-read is disabled, this will halt decoding of packets until auto-read is re-enabled.
9+
* This is needed because auto-read still allows packets to be decoded, even if the channel is not reading anymore from the network.
10+
* This can happen when the channel already read a packet, but the packet is not yet decoded.
11+
* This will halt all decoding until the channel is ready to process more packets.
12+
*/
13+
public class TcpFlowControlHandler extends FlowControlHandler {
14+
@Override
15+
public void read(ChannelHandlerContext ctx) throws Exception {
16+
if (ctx.channel().config().isAutoRead()) {
17+
super.read(ctx);
18+
}
19+
}
20+
}

‎protocol/src/main/java/org/geysermc/mcprotocollib/network/tcp/TcpPacketCodec.java‎

Lines changed: 51 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -2,17 +2,27 @@
22

33
import io.netty.buffer.ByteBuf;
44
import io.netty.channel.ChannelHandlerContext;
5-
import io.netty.handler.codec.ByteToMessageCodec;
5+
import io.netty.handler.codec.DecoderException;
6+
import io.netty.handler.codec.EncoderException;
7+
import io.netty.handler.codec.MessageToMessageCodec;
68
import org.geysermc.mcprotocollib.network.Session;
79
import org.geysermc.mcprotocollib.network.codec.PacketCodecHelper;
810
import org.geysermc.mcprotocollib.network.codec.PacketDefinition;
911
import org.geysermc.mcprotocollib.network.event.session.PacketErrorEvent;
1012
import org.geysermc.mcprotocollib.network.packet.Packet;
1113
import org.geysermc.mcprotocollib.network.packet.PacketProtocol;
14+
import org.geysermc.mcprotocollib.network.packet.PacketRegistry;
15+
import org.slf4j.Logger;
16+
import org.slf4j.LoggerFactory;
17+
import org.slf4j.Marker;
18+
import org.slf4j.MarkerFactory;
1219

1320
import java.util.List;
1421

15-
public class TcpPacketCodec extends ByteToMessageCodec<Packet> {
22+
public class TcpPacketCodec extends MessageToMessageCodec<ByteBuf, Packet> {
23+
private static final Marker marker = MarkerFactory.getMarker("packet_logging");
24+
private static final Logger log = LoggerFactory.getLogger(TcpPacketCodec.class);
25+
1626
private final Session session;
1727
private final boolean client;
1828

@@ -23,57 +33,87 @@ public TcpPacketCodec(Session session, boolean client) {
2333

2434
@SuppressWarnings({"rawtypes", "unchecked"})
2535
@Override
26-
public void encode(ChannelHandlerContext ctx, Packet packet, ByteBuf buf) {
27-
int initial = buf.writerIndex();
36+
public void encode(ChannelHandlerContext ctx, Packet packet, List<Object> out) {
37+
if (log.isTraceEnabled()) {
38+
log.trace(marker, "Encoding packet: {}", packet.getClass().getSimpleName());
39+
}
2840

2941
PacketProtocol packetProtocol = this.session.getPacketProtocol();
42+
PacketRegistry packetRegistry = packetProtocol.getOutboundPacketRegistry();
3043
PacketCodecHelper codecHelper = this.session.getCodecHelper();
3144
try {
32-
int packetId = this.client ? packetProtocol.getPacketRegistry().getServerboundId(packet) : packetProtocol.getPacketRegistry().getClientboundId(packet);
33-
PacketDefinition definition = this.client ? packetProtocol.getPacketRegistry().getServerboundDefinition(packetId) : packetProtocol.getPacketRegistry().getClientboundDefinition(packetId);
45+
int packetId = this.client ? packetRegistry.getServerboundId(packet) : packetRegistry.getClientboundId(packet);
46+
PacketDefinition definition = this.client ? packetRegistry.getServerboundDefinition(packetId) : packetRegistry.getClientboundDefinition(packetId);
3447

48+
ByteBuf buf = ctx.alloc().buffer();
3549
packetProtocol.getPacketHeader().writePacketId(buf, codecHelper, packetId);
3650
definition.getSerializer().serialize(buf, codecHelper, packet);
51+
52+
out.add(buf);
53+
54+
if (log.isDebugEnabled()) {
55+
log.debug(marker, "Encoded packet {} ({})", packet.getClass().getSimpleName(), packetId);
56+
}
3757
} catch (Throwable t) {
38-
// Reset writer index to make sure incomplete data is not written out.
39-
buf.writerIndex(initial);
58+
log.debug(marker, "Error encoding packet", t);
4059

4160
PacketErrorEvent e = new PacketErrorEvent(this.session, t);
4261
this.session.callEvent(e);
4362
if (!e.shouldSuppress()) {
44-
throw t;
63+
throw new EncoderException(t);
4564
}
4665
}
4766
}
4867

4968
@Override
5069
protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> out) {
70+
// Vanilla also checks for 0 length
71+
if (buf.readableBytes() == 0) {
72+
return;
73+
}
74+
5175
int initial = buf.readerIndex();
5276

5377
PacketProtocol packetProtocol = this.session.getPacketProtocol();
78+
PacketRegistry packetRegistry = packetProtocol.getInboundPacketRegistry();
5479
PacketCodecHelper codecHelper = this.session.getCodecHelper();
80+
Packet packet = null;
5581
try {
5682
int id = packetProtocol.getPacketHeader().readPacketId(buf, codecHelper);
5783
if (id == -1) {
5884
buf.readerIndex(initial);
5985
return;
6086
}
6187

62-
Packet packet = this.client ? packetProtocol.getPacketRegistry().createClientboundPacket(id, buf, codecHelper) : packetProtocol.getPacketRegistry().createServerboundPacket(id, buf, codecHelper);
88+
log.trace(marker, "Decoding packet with id: {}", id);
89+
90+
packet = this.client ? packetRegistry.createClientboundPacket(id, buf, codecHelper) : packetRegistry.createServerboundPacket(id, buf, codecHelper);
6391

6492
if (buf.readableBytes() > 0) {
6593
throw new IllegalStateException("Packet \"" + packet.getClass().getSimpleName() + "\" not fully read.");
6694
}
6795

6896
out.add(packet);
97+
98+
if (log.isDebugEnabled()) {
99+
log.debug(marker, "Decoded packet {} ({})", packet.getClass().getSimpleName(), id);
100+
}
69101
} catch (Throwable t) {
102+
log.debug(marker, "Error decoding packet", t);
103+
70104
// Advance buffer to end to make sure remaining data in this packet is skipped.
71105
buf.readerIndex(buf.readerIndex() + buf.readableBytes());
72106

73107
PacketErrorEvent e = new PacketErrorEvent(this.session, t);
74108
this.session.callEvent(e);
75109
if (!e.shouldSuppress()) {
76-
throw t;
110+
throw new DecoderException(t);
111+
}
112+
} finally {
113+
if (packet != null && packet.isTerminal()) {
114+
// Next packets are in a different protocol state, so we must
115+
// disable auto-read to prevent reading wrong packets.
116+
session.setAutoRead(false);
77117
}
78118
}
79119
}

‎protocol/src/main/java/org/geysermc/mcprotocollib/network/tcp/TcpServer.java‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
import io.netty.channel.EventLoopGroup;
1010
import io.netty.handler.timeout.ReadTimeoutHandler;
1111
import io.netty.handler.timeout.WriteTimeoutHandler;
12+
import org.checkerframework.checker.nullness.qual.NonNull;
1213
import org.geysermc.mcprotocollib.network.AbstractServer;
1314
import org.geysermc.mcprotocollib.network.BuiltinFlags;
1415
import org.geysermc.mcprotocollib.network.helper.TransportHelper;
@@ -52,7 +53,7 @@ public void bindImpl(boolean wait, final Runnable callback) {
5253
.localAddress(this.getHost(), this.getPort())
5354
.childHandler(new ChannelInitializer<>() {
5455
@Override
55-
public void initChannel(Channel channel) {
56+
public void initChannel(@NonNull Channel channel) {
5657
InetSocketAddress address = (InetSocketAddress) channel.remoteAddress();
5758
PacketProtocol protocol = createPacketProtocol();
5859

@@ -68,7 +69,9 @@ public void initChannel(Channel channel) {
6869
pipeline.addLast("sizer", new TcpPacketSizer(protocol.getPacketHeader(), session.getCodecHelper()));
6970
pipeline.addLast("compression", new TcpPacketCompression(session.getCodecHelper()));
7071

72+
pipeline.addLast("flow-control", new TcpFlowControlHandler());
7173
pipeline.addLast("codec", new TcpPacketCodec(session, false));
74+
pipeline.addLast("flush-handler", new FlushHandler());
7275
pipeline.addLast("manager", session);
7376
}
7477
});

0 commit comments

Comments
 (0)