From 1decc440816a9252cf0aeb025940dbdb04c74789 Mon Sep 17 00:00:00 2001 From: Layue13 Date: Fri, 23 Oct 2026 16:29:46 +0800 Subject: [PATCH 1/2] perf(client-mod): candidate WebSocket codec, codecBenchmark and decision rule TCP_NODELAY on ws/wss; ClientFrameEncoder (one message per batch) and ClientFrameDecoder (retained slices) with tests. The decision rule is recorded before the measured runs; the verdict follows in a later commit. Co-Authored-By: Claude Opus 5.5 --- client-mod/addon.gradle | 14 + .../clientws/net/ClientFrameDecoder.java | 256 +++++++++ .../clientws/net/ClientFrameEncoder.java | 283 ++++++++++ .../clientws/net/WebSocketFrameAdapter.java | 37 +- .../clientws/net/WebSocketPipeline.java | 17 +- .../clientws/net/ClientFrameDecoderTest.java | 276 ++++++++++ .../clientws/net/ClientFrameEncoderTest.java | 339 ++++++++++++ .../clientws/net/CodecBenchmark.java | 496 ++++++++++++++++++ .../net/WebSocketFrameAdapterTest.java | 22 +- .../clientws/net/WebSocketPipelineTest.java | 19 +- .../2026-10-23-client-websocket-codec.md | 14 + 11 files changed, 1720 insertions(+), 53 deletions(-) create mode 100644 client-mod/src/main/java/dev/moonbridge/clientws/net/ClientFrameDecoder.java create mode 100644 client-mod/src/main/java/dev/moonbridge/clientws/net/ClientFrameEncoder.java create mode 100644 client-mod/src/test/java/dev/moonbridge/clientws/net/ClientFrameDecoderTest.java create mode 100644 client-mod/src/test/java/dev/moonbridge/clientws/net/ClientFrameEncoderTest.java create mode 100644 client-mod/src/test/java/dev/moonbridge/clientws/net/CodecBenchmark.java create mode 100644 testing/benchmarks/results/2026-10-23-client-websocket-codec.md diff --git a/client-mod/addon.gradle b/client-mod/addon.gradle index 1d6e169..b77f60c 100644 --- a/client-mod/addon.gradle +++ b/client-mod/addon.gradle @@ -8,3 +8,17 @@ tasks.named("test", Test) { exceptionFormat "full" } } + +// Client WebSocket codec microbenchmark (test sources, never packaged). See +// testing/benchmarks/results/2026-10-23-client-websocket-codec.md. Example: +// ./gradlew codecBenchmark -PbenchArgs="--variant candidate --repeats 3" +tasks.register("codecBenchmark", JavaExec) { + description = "Runs the client WebSocket codec microbenchmark (Netty 4.0.10 codec vs ClientFrameEncoder/Decoder)." + classpath = sourceSets.test.runtimeClasspath + mainClass = "dev.moonbridge.clientws.net.CodecBenchmark" + javaLauncher = tasks.named("test", Test).flatMap { it.javaLauncher } + jvmArgs "-Xms1g", "-Xmx1g", "-Dio.netty.leakDetectionLevel=disabled" + if (project.hasProperty("benchArgs")) { + args project.property("benchArgs").toString().split(" ") + } +} diff --git a/client-mod/src/main/java/dev/moonbridge/clientws/net/ClientFrameDecoder.java b/client-mod/src/main/java/dev/moonbridge/clientws/net/ClientFrameDecoder.java new file mode 100644 index 0000000..63bc01f --- /dev/null +++ b/client-mod/src/main/java/dev/moonbridge/clientws/net/ClientFrameDecoder.java @@ -0,0 +1,256 @@ +package dev.moonbridge.clientws.net; + +import java.nio.charset.CharacterCodingException; +import java.nio.charset.CodingErrorAction; +import java.nio.charset.StandardCharsets; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.handler.codec.CorruptedFrameException; +import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame; +import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; +import io.netty.handler.codec.http.websocketx.PongWebSocketFrame; +import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; + +/** + * Client-side WebSocket frame decoder for server (unmasked) frames. + * + *

+ * Binary and continuation payloads are passed on as plain {@link ByteBuf}s: retained slices of the received buffer, + * emitted as soon as their bytes arrive, so a large frame is never gathered and nothing is copied. Only a frame head + * split across reads (at most 14 bytes, or a control frame of up to 125 bytes) is copied into a small carry buffer. + * Ping, pong and close become the usual frame objects; a text frame becomes a {@link TextWebSocketFrame} with an empty + * payload, after which input is ignored (the adapter closes with 1003). + * + *

+ * Checks match Netty's {@code WebSocket13FrameDecoder(false, false, 16 MiB)}: masked frames, RSV bits, reserved + * opcodes, fragmented or oversized control frames, non-minimal lengths, bad continuations, invalid close bodies and + * frames over {@link WebSocketPipeline#MAX_FRAME_PAYLOAD} are protocol violations, which close the connection and + * raise a {@link CorruptedFrameException}. Input after a close frame or a violation is discarded. + * + *

+ * This is not a {@code ByteToMessageDecoder}: in Netty 4.0.10 that class calls {@code discardSomeReadBytes()} on its + * cumulation before it passes decoded messages on, which moves bytes under slices that were just taken from it. + */ +public final class ClientFrameDecoder extends ChannelInboundHandlerAdapter { + + private static final int MAX_HEAD = 14 + 125; + + private final long maxPayload; + private ByteBuf carry; + private long payloadLeft; + private boolean fragmented; + private boolean discarding; + + public ClientFrameDecoder() { + this(WebSocketPipeline.MAX_FRAME_PAYLOAD); + } + + ClientFrameDecoder(long maxPayload) { + this.maxPayload = maxPayload; + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + if (!(msg instanceof ByteBuf)) { + ctx.fireChannelRead(msg); + return; + } + ByteBuf in = (ByteBuf) msg; + try { + if (carry != null && !completeCarriedHead(ctx, in)) { + return; + } + decode(ctx, in); + } finally { + in.release(); + } + } + + private void decode(ChannelHandlerContext ctx, ByteBuf in) { + while (in.isReadable()) { + if (discarding) { + in.skipBytes(in.readableBytes()); + return; + } + if (payloadLeft > 0) { + int n = (int) Math.min(payloadLeft, in.readableBytes()); + payloadLeft -= n; + ctx.fireChannelRead( + in.readSlice(n) + .retain()); + continue; + } + int head = headLength(in); + if (head < 0 || head > in.readableBytes()) { + carry = ctx.alloc() + .buffer(MAX_HEAD); + carry.writeBytes(in); + return; + } + readHead(ctx, in); + } + } + + /** Moves just enough bytes from {@code in} to finish the carried head; true once it has been decoded. */ + private boolean completeCarriedHead(ChannelHandlerContext ctx, ByteBuf in) { + while (true) { + int head = headLength(carry); + int want = (head < 0 ? 2 : head) - carry.readableBytes(); + if (want <= 0) { + break; + } + if (!in.isReadable()) { + return false; + } + carry.writeBytes(in, Math.min(want, in.readableBytes())); + } + ByteBuf head = carry; + carry = null; + try { + readHead(ctx, head); + } finally { + head.release(); + } + return true; + } + + /** + * Bytes needed for the frame head starting at the reader index, including a control frame's payload; -1 if fewer + * than two bytes are readable. Masking keys are not counted: a masked frame is rejected first. + */ + private static int headLength(ByteBuf buf) { + if (buf.readableBytes() < 2) { + return -1; + } + int b0 = buf.getUnsignedByte(buf.readerIndex()); + int len7 = buf.getUnsignedByte(buf.readerIndex() + 1) & 0x7F; + if ((b0 & 0x08) != 0) { + // Control frame: the head carries the whole payload, unless the length is already invalid. + return len7 <= 125 ? 2 + len7 : 2; + } + return len7 < 126 ? 2 : len7 == 126 ? 4 : 10; + } + + private void readHead(ChannelHandlerContext ctx, ByteBuf buf) { + int b0 = buf.readUnsignedByte(); + int b1 = buf.readUnsignedByte(); + boolean fin = (b0 & 0x80) != 0; + int opcode = b0 & 0x0F; + if ((b0 & 0x70) != 0) { + throw violation(ctx, "RSV != 0 and no extension negotiated, RSV:" + ((b0 & 0x70) >> 4)); + } + if ((b1 & 0x80) != 0) { + throw violation(ctx, "masked server to client frame"); + } + int len7 = b1 & 0x7F; + if (opcode >= 8) { + if (!fin) { + throw violation(ctx, "fragmented control frame"); + } + if (len7 > 125) { + throw violation(ctx, "control frame with payload length > 125 octets"); + } + if (opcode > ClientFrameEncoder.OPCODE_PONG) { + throw violation(ctx, "control frame using reserved opcode " + opcode); + } + if (opcode == ClientFrameEncoder.OPCODE_CLOSE) { + checkCloseBody(ctx, buf, len7); + } + ByteBuf payload = len7 == 0 ? Unpooled.EMPTY_BUFFER : buf.readBytes(len7); + if (opcode == ClientFrameEncoder.OPCODE_CLOSE) { + discarding = true; + ctx.fireChannelRead(new CloseWebSocketFrame(true, 0, payload)); + } else if (opcode == ClientFrameEncoder.OPCODE_PING) { + ctx.fireChannelRead(new PingWebSocketFrame(payload)); + } else { + ctx.fireChannelRead(new PongWebSocketFrame(payload)); + } + return; + } + if (opcode > ClientFrameEncoder.OPCODE_BINARY) { + throw violation(ctx, "data frame using reserved opcode " + opcode); + } + if (opcode == ClientFrameEncoder.OPCODE_CONTINUATION && !fragmented) { + throw violation(ctx, "received continuation data frame outside fragmented message"); + } + if (opcode != ClientFrameEncoder.OPCODE_CONTINUATION && fragmented) { + throw violation(ctx, "received non-continuation data frame while inside fragmented message"); + } + long length = len7; + if (len7 == 126) { + length = buf.readUnsignedShort(); + if (length < 126) { + throw violation(ctx, "invalid data frame length (not using minimal length encoding)"); + } + } else if (len7 == 127) { + length = buf.readLong(); + if (length < 65536) { + throw violation(ctx, "invalid data frame length (not using minimal length encoding)"); + } + } + if (length > maxPayload) { + throw violation(ctx, "Max frame length of " + maxPayload + " has been exceeded."); + } + if (opcode == ClientFrameEncoder.OPCODE_TEXT) { + discarding = true; + ctx.fireChannelRead(new TextWebSocketFrame(Unpooled.EMPTY_BUFFER)); + return; + } + fragmented = !fin; + payloadLeft = length; + } + + private void checkCloseBody(ChannelHandlerContext ctx, ByteBuf buf, int length) { + if (length == 0) { + return; + } + if (length == 1) { + throw violation(ctx, "Invalid close frame body"); + } + int status = buf.getUnsignedShort(buf.readerIndex()); + if (status <= 999 || status >= 1004 && status <= 1006 || status >= 1012 && status <= 2999) { + throw violation(ctx, "Invalid close frame getStatus code: " + status); + } + try { + StandardCharsets.UTF_8.newDecoder() + .onMalformedInput(CodingErrorAction.REPORT) + .onUnmappableCharacter(CodingErrorAction.REPORT) + .decode(buf.nioBuffer(buf.readerIndex() + 2, length - 2)); + } catch (CharacterCodingException e) { + throw violation(ctx, "Invalid close frame reason text. Invalid UTF-8 bytes"); + } + } + + private CorruptedFrameException violation(ChannelHandlerContext ctx, String reason) { + discarding = true; + payloadLeft = 0; + if (ctx.channel() + .isActive()) { + ctx.writeAndFlush(Unpooled.EMPTY_BUFFER) + .addListener(ChannelFutureListener.CLOSE); + } + return new CorruptedFrameException(reason); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + releaseCarry(); + ctx.fireChannelInactive(); + } + + @Override + public void handlerRemoved(ChannelHandlerContext ctx) { + releaseCarry(); + } + + private void releaseCarry() { + if (carry != null) { + carry.release(); + carry = null; + } + } +} diff --git a/client-mod/src/main/java/dev/moonbridge/clientws/net/ClientFrameEncoder.java b/client-mod/src/main/java/dev/moonbridge/clientws/net/ClientFrameEncoder.java new file mode 100644 index 0000000..caee2a4 --- /dev/null +++ b/client-mod/src/main/java/dev/moonbridge/clientws/net/ClientFrameEncoder.java @@ -0,0 +1,283 @@ +package dev.moonbridge.clientws.net; + +import java.nio.channels.ClosedChannelException; +import java.util.ArrayDeque; +import java.util.Arrays; +import java.util.concurrent.ThreadLocalRandom; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelDuplexHandler; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelPromise; +import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame; +import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame; +import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame; +import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; +import io.netty.handler.codec.http.websocketx.PongWebSocketFrame; +import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; +import io.netty.handler.codec.http.websocketx.WebSocketFrame; + +/** + * Client-side WebSocket frame encoder that sends one binary message per batch of writes. + * + *

+ * Vanilla calls {@code writeAndFlush} once per packet from the client thread, and Netty runs each of those as its own + * event-loop task. Plain {@link ByteBuf} writes are therefore queued here, and a flush only schedules one task at the + * end of the event loop's queue; by the time it runs, every packet the client thread queued in the meantime has been + * written too, and the whole batch goes out as one binary message (split at {@link #MAX_MESSAGE_PAYLOAD}). A lone + * packet still leaves in the same loop iteration: the task runs right after the write that scheduled it. A write's + * promise completes when the message carrying its last byte has been written. + * + *

+ * Each message is one buffer: the header, then the payload masked four bytes at a time. {@link WebSocketFrame}s are + * encoded as they come; a pong is flushed at once and may overtake queued data (control frames may be sent between + * messages), any other frame first writes the queued data so it keeps its place. A close of the channel writes and + * flushes queued data first, since its writer already flushed it. + */ +public final class ClientFrameEncoder extends ChannelDuplexHandler { + + /** Largest payload of one binary message; matches the proxy's outbound cap. */ + public static final int MAX_MESSAGE_PAYLOAD = 64 * 1024; + + static final int OPCODE_CONTINUATION = 0x0; + static final int OPCODE_TEXT = 0x1; + static final int OPCODE_BINARY = 0x2; + static final int OPCODE_CLOSE = 0x8; + static final int OPCODE_PING = 0x9; + static final int OPCODE_PONG = 0xA; + + /** Queued writes as alternating {@link ByteBuf} and {@link ChannelPromise}. */ + private final ArrayDeque pending = new ArrayDeque(); + private int pendingBytes; + private boolean flushScheduled; + private ChannelHandlerContext ctx; + private final Runnable flushTask = new Runnable() { + + @Override + public void run() { + flushScheduled = false; + if (!pending.isEmpty()) { + writePending(ctx); + ctx.flush(); + } + } + }; + + @Override + public void handlerAdded(ChannelHandlerContext ctx) { + this.ctx = ctx; + } + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { + if (msg instanceof ByteBuf) { + ByteBuf buf = (ByteBuf) msg; + if (!buf.isReadable()) { + buf.release(); + promise.setSuccess(); + return; + } + pending.add(buf); + pending.add(promise); + pendingBytes += buf.readableBytes(); + } else if (msg instanceof PongWebSocketFrame) { + ctx.write(encode(ctx, (WebSocketFrame) msg), promise); + ctx.flush(); + } else if (msg instanceof WebSocketFrame) { + writePending(ctx); + ctx.write(encode(ctx, (WebSocketFrame) msg), promise); + } else { + ctx.write(msg, promise); + } + } + + @Override + public void flush(ChannelHandlerContext ctx) throws Exception { + if (pending.isEmpty()) { + ctx.flush(); + } else if (!flushScheduled) { + flushScheduled = true; + ctx.executor() + .execute(flushTask); + } + } + + @Override + public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception { + if (!pending.isEmpty()) { + writePending(ctx); + ctx.flush(); + } + ctx.close(promise); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + failPending(new ClosedChannelException()); + ctx.fireChannelInactive(); + } + + @Override + public void handlerRemoved(ChannelHandlerContext ctx) { + failPending(new ClosedChannelException()); + } + + private void failPending(Throwable cause) { + Object o; + while ((o = pending.poll()) != null) { + ((ByteBuf) o).release(); + ((ChannelPromise) pending.poll()).tryFailure(cause); + } + pendingBytes = 0; + } + + /** Writes everything queued as binary messages of at most {@link #MAX_MESSAGE_PAYLOAD} bytes, without flushing. */ + private void writePending(ChannelHandlerContext ctx) { + while (pendingBytes > 0) { + int length = Math.min(pendingBytes, MAX_MESSAGE_PAYLOAD); + int mask = ThreadLocalRandom.current() + .nextInt(); + ByteBuf out = ctx.alloc() + .buffer(headerLength(length) + length); + writeHeader(out, true, OPCODE_BINARY, length, mask); + ChannelPromise single = null; + PromiseFanOut fanOut = null; + int offset = 0; + while (offset < length) { + ByteBuf buf = (ByteBuf) pending.peek(); + int n = Math.min(buf.readableBytes(), length - offset); + writeMasked(out, buf, buf.readerIndex(), n, Integer.rotateLeft(mask, 8 * (offset & 3))); + buf.skipBytes(n); + offset += n; + if (buf.isReadable()) { + break; + } + pending.poll(); + buf.release(); + ChannelPromise done = (ChannelPromise) pending.poll(); + if (single == null) { + single = done; + } else { + if (fanOut == null) { + fanOut = new PromiseFanOut(single); + } + fanOut.add(done); + } + } + pendingBytes -= length; + if (fanOut != null) { + ctx.write(out) + .addListener(fanOut); + } else if (single != null) { + ctx.write(out, single); + } else { + ctx.write(out); + } + } + } + + /** Encodes a single frame object into one masked buffer and releases the frame. */ + static ByteBuf encode(ChannelHandlerContext ctx, WebSocketFrame frame) { + try { + ByteBuf payload = frame.content(); + int length = payload.readableBytes(); + int mask = ThreadLocalRandom.current() + .nextInt(); + ByteBuf out = ctx.alloc() + .buffer(headerLength(length) + length); + out.writeByte((frame.isFinalFragment() ? 0x80 : 0) | (frame.rsv() & 7) << 4 | opcode(frame)); + writeLength(out, length, mask); + writeMasked(out, payload, payload.readerIndex(), length, mask); + return out; + } finally { + frame.release(); + } + } + + private static int opcode(WebSocketFrame frame) { + if (frame instanceof BinaryWebSocketFrame) { + return OPCODE_BINARY; + } else if (frame instanceof PongWebSocketFrame) { + return OPCODE_PONG; + } else if (frame instanceof CloseWebSocketFrame) { + return OPCODE_CLOSE; + } else if (frame instanceof PingWebSocketFrame) { + return OPCODE_PING; + } else if (frame instanceof ContinuationWebSocketFrame) { + return OPCODE_CONTINUATION; + } else if (frame instanceof TextWebSocketFrame) { + return OPCODE_TEXT; + } + throw new UnsupportedOperationException( + "cannot encode " + frame.getClass() + .getName()); + } + + static int headerLength(int payloadLength) { + return 2 + (payloadLength < 126 ? 0 : payloadLength <= 0xFFFF ? 2 : 8) + 4; + } + + static void writeHeader(ByteBuf out, boolean fin, int opcode, int length, int mask) { + out.writeByte((fin ? 0x80 : 0) | opcode); + writeLength(out, length, mask); + } + + private static void writeLength(ByteBuf out, int length, int mask) { + if (length < 126) { + out.writeByte(0x80 | length); + } else if (length <= 0xFFFF) { + out.writeByte(0x80 | 126); + out.writeShort(length); + } else { + out.writeByte(0x80 | 127); + out.writeLong(length); + } + out.writeInt(mask); + } + + /** + * Appends {@code length} bytes of {@code src} from {@code index}, XORed with {@code mask} whose most significant + * byte applies to the first byte. Callers rotate the mask when the bytes do not start at a multiple of four. + */ + static void writeMasked(ByteBuf out, ByteBuf src, int index, int length, int mask) { + int end = index + length; + int i = index; + for (; end - i >= 4; i += 4) { + out.writeInt(src.getInt(i) ^ mask); + } + for (int shift = 24; i < end; i++, shift -= 8) { + out.writeByte(src.getByte(i) ^ mask >>> shift); + } + } + + /** Completes the promises of every write whose last byte went out in one message. */ + private static final class PromiseFanOut implements ChannelFutureListener { + + private ChannelPromise[] promises = new ChannelPromise[4]; + private int size; + + PromiseFanOut(ChannelPromise first) { + add(first); + } + + void add(ChannelPromise promise) { + if (size == promises.length) { + promises = Arrays.copyOf(promises, size * 2); + } + promises[size++] = promise; + } + + @Override + public void operationComplete(ChannelFuture future) { + for (int i = 0; i < size; i++) { + if (future.isSuccess()) { + promises[i].trySuccess(); + } else { + promises[i].tryFailure(future.cause()); + } + } + } + } +} diff --git a/client-mod/src/main/java/dev/moonbridge/clientws/net/WebSocketFrameAdapter.java b/client-mod/src/main/java/dev/moonbridge/clientws/net/WebSocketFrameAdapter.java index ce0616f..6fcea06 100644 --- a/client-mod/src/main/java/dev/moonbridge/clientws/net/WebSocketFrameAdapter.java +++ b/client-mod/src/main/java/dev/moonbridge/clientws/net/WebSocketFrameAdapter.java @@ -3,33 +3,29 @@ package dev.moonbridge.clientws.net; import java.io.IOException; import io.netty.buffer.ByteBuf; -import io.netty.channel.ChannelDuplexHandler; import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelPromise; -import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame; +import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame; -import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame; import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; import io.netty.handler.codec.http.websocketx.PongWebSocketFrame; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; -import io.netty.handler.codec.http.websocketx.WebSocketFrame; /** - * Turns the WebSocket message stream back into the raw Minecraft byte stream and vice versa. + * Turns the WebSocket message stream back into the raw Minecraft byte stream. * *

- * Inbound: the payload of every binary and continuation frame is passed on as-is (ownership moves with it), so the - * vanilla splitter sees one concatenated byte stream regardless of how the server chunked it. Ping is answered with - * pong, close ends the connection, and a text frame is a protocol error. Outbound: each {@link ByteBuf} written by the - * vanilla handlers becomes one binary frame. + * {@link ClientFrameDecoder} already passes binary and continuation payloads on as plain {@link ByteBuf}s, so the + * vanilla splitter sees one concatenated byte stream regardless of how the server chunked it; they go through as-is + * (ownership moves with them). Ping is answered with pong, close ends the connection, and a text frame is a protocol + * error. Outbound, vanilla's {@link ByteBuf}s go straight to {@link ClientFrameEncoder}, which frames them. */ -public final class WebSocketFrameAdapter extends ChannelDuplexHandler { +public final class WebSocketFrameAdapter extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { - if (msg instanceof BinaryWebSocketFrame || msg instanceof ContinuationWebSocketFrame) { - ByteBuf payload = ((WebSocketFrame) msg).content(); + if (msg instanceof ByteBuf) { + ByteBuf payload = (ByteBuf) msg; if (payload.isReadable()) { ctx.fireChannelRead(payload); } else { @@ -52,19 +48,4 @@ public final class WebSocketFrameAdapter extends ChannelDuplexHandler { ctx.fireChannelRead(msg); } } - - @Override - public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { - if (msg instanceof ByteBuf) { - ByteBuf buf = (ByteBuf) msg; - if (buf.isReadable()) { - ctx.write(new BinaryWebSocketFrame(buf), promise); - } else { - buf.release(); - promise.setSuccess(); - } - } else { - ctx.write(msg, promise); - } - } } diff --git a/client-mod/src/main/java/dev/moonbridge/clientws/net/WebSocketPipeline.java b/client-mod/src/main/java/dev/moonbridge/clientws/net/WebSocketPipeline.java index 0a0b48d..81ff685 100644 --- a/client-mod/src/main/java/dev/moonbridge/clientws/net/WebSocketPipeline.java +++ b/client-mod/src/main/java/dev/moonbridge/clientws/net/WebSocketPipeline.java @@ -8,9 +8,9 @@ import javax.net.ssl.SSLContext; import javax.net.ssl.SSLEngine; import javax.net.ssl.SSLParameters; +import io.netty.channel.ChannelException; +import io.netty.channel.ChannelOption; import io.netty.channel.ChannelPipeline; -import io.netty.handler.codec.http.websocketx.WebSocket13FrameDecoder; -import io.netty.handler.codec.http.websocketx.WebSocket13FrameEncoder; import io.netty.handler.ssl.SslHandler; /** @@ -51,10 +51,19 @@ public final class WebSocketPipeline { /** Variant with an explicit TLS context (tests); {@code tls} is ignored for {@code ws://}. */ public static void install(ChannelPipeline pipeline, WebSocketAddress address, SSLContext tls) { + // Vanilla turns Nagle on just before this runs. The encoder already gathers a batch of packets into one + // message, so Nagle would only hold that message back for the previous one's ACK (a whole CDN round trip). + try { + pipeline.channel() + .config() + .setOption(ChannelOption.TCP_NODELAY, true); + } catch (ChannelException ignored) { + // Same as vanilla: keep going on platforms that refuse the option. + } // addFirst in reverse so the final order is ssl, handshake, decoder, encoder, frames, vanilla... pipeline.addFirst(FRAMES, new WebSocketFrameAdapter()); - pipeline.addFirst(FRAME_ENCODER, new WebSocket13FrameEncoder(true)); - pipeline.addFirst(FRAME_DECODER, new WebSocket13FrameDecoder(false, false, MAX_FRAME_PAYLOAD)); + pipeline.addFirst(FRAME_ENCODER, new ClientFrameEncoder()); + pipeline.addFirst(FRAME_DECODER, new ClientFrameDecoder()); pipeline.addFirst(HANDSHAKE, new WebSocketHandshakeHandler(address)); if (address.isSecure()) { pipeline.addFirst(SSL, new SslHandler(newClientEngine(tls, address))); diff --git a/client-mod/src/test/java/dev/moonbridge/clientws/net/ClientFrameDecoderTest.java b/client-mod/src/test/java/dev/moonbridge/clientws/net/ClientFrameDecoderTest.java new file mode 100644 index 0000000..f6dfe1f --- /dev/null +++ b/client-mod/src/test/java/dev/moonbridge/clientws/net/ClientFrameDecoderTest.java @@ -0,0 +1,276 @@ +package dev.moonbridge.clientws.net; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.ByteArrayOutputStream; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Random; + +import org.junit.jupiter.api.Test; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.PooledByteBufAllocator; +import io.netty.buffer.Unpooled; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.CorruptedFrameException; +import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame; +import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; +import io.netty.handler.codec.http.websocketx.PongWebSocketFrame; +import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; + +class ClientFrameDecoderTest { + + /** Server frame with any flags and the shortest length form, or a forced one ({@code lengthBytes} 0, 2 or 8). */ + static byte[] frame(int b0, byte[] payload, int lengthBytes) { + ByteArrayOutputStream f = new ByteArrayOutputStream(); + f.write(b0); + if (lengthBytes == 0) { + f.write(payload.length); + } else if (lengthBytes == 2) { + f.write(126); + f.write(payload.length >>> 8); + f.write(payload.length & 0xff); + } else { + f.write(127); + for (int shift = 56; shift >= 0; shift -= 8) { + f.write((int) ((long) payload.length >>> shift) & 0xff); + } + } + f.write(payload, 0, payload.length); + return f.toByteArray(); + } + + static byte[] frame(int b0, byte[] payload) { + return frame(b0, payload, payload.length < 126 ? 0 : payload.length <= 0xFFFF ? 2 : 8); + } + + private static byte[] bytes(Random r, int n) { + byte[] b = new byte[n]; + r.nextBytes(b); + return b; + } + + private static byte[] concat(byte[]... parts) { + ByteArrayOutputStream o = new ByteArrayOutputStream(); + for (byte[] p : parts) { + o.write(p, 0, p.length); + } + return o.toByteArray(); + } + + /** Pooled direct buffers, so Netty 4.0.10's leak detector tracks them (heap buffers are not tracked). */ + private static ByteBuf tracked(byte[] b, int from, int to) { + return PooledByteBufAllocator.DEFAULT.directBuffer(to - from) + .writeBytes(b, from, to - from); + } + + @Test + void binaryPayloadIsARetainedSliceOfTheInput() { + EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameDecoder()); + byte[] payload = { 1, 2, 3, 4, 5 }; + byte[] wire = frame(0x82, payload); + ByteBuf in = tracked(wire, 0, wire.length); + ch.writeInbound(in); + ByteBuf out = (ByteBuf) ch.readInbound(); + assertSame(in, out.unwrap(), "a slice of the received buffer, not a copy"); + assertEquals(1, in.refCnt(), "the decoder dropped its reference; the slice holds the only one"); + byte[] got = new byte[out.readableBytes()]; + out.readBytes(got); + assertArrayEquals(payload, got); + out.release(); + assertEquals(0, in.refCnt()); + assertFalse(ch.finish()); + } + + @Test + void reassemblesAStreamSplitAtEveryPoint() { + Random r = new Random(7); + byte[] a = bytes(r, 300); // 16-bit length + byte[] b = bytes(r, 70000); // 64-bit length + byte[] c1 = bytes(r, 2); + byte[] c2 = bytes(r, 125); + byte[] wire = concat( + frame(0x82, a), + frame(0x02, c1), // fragmented message with a ping in between + frame(0x89, new byte[] { 'p', 'i' }), + frame(0x80, c2), + frame(0x82, b), + frame(0x82, new byte[0])); + byte[] expected = concat(a, c1, c2, b); + int[] splits = new int[400]; + for (int i = 0; i < splits.length; i++) { + splits[i] = i; // every split inside the first frames' heads and payload starts + } + for (int split : splits) { + check(wire, expected, new int[] { split }); + } + // And arbitrary chunking, including 1-byte reads over the whole of a small stream. + for (int chunk : new int[] { 1, 3, 7, 1460 }) { + List points = new ArrayList(); + for (int p = chunk; p < wire.length && points.size() < 5000; p += chunk) { + points.add(p); + } + int[] cuts = new int[points.size()]; + for (int i = 0; i < cuts.length; i++) { + cuts[i] = points.get(i); + } + check(wire, expected, cuts); + } + } + + private static void check(byte[] wire, byte[] expected, int[] cuts) { + EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameDecoder()); + List inputs = new ArrayList(); + int from = 0; + for (int i = 0; i <= cuts.length; i++) { + int to = i < cuts.length ? cuts[i] : wire.length; + if (to <= from) { + continue; + } + ByteBuf in = tracked(wire, from, to); + inputs.add(in); + ch.writeInbound(in); + from = to; + } + ByteArrayOutputStream got = new ByteArrayOutputStream(); + int pings = 0; + Object o; + while ((o = ch.readInbound()) != null) { + if (o instanceof PingWebSocketFrame) { + PingWebSocketFrame ping = (PingWebSocketFrame) o; + assertEquals( + "pi", + ping.content() + .toString(StandardCharsets.US_ASCII)); + ping.release(); + pings++; + continue; + } + ByteBuf buf = (ByteBuf) o; + assertTrue(buf.isReadable(), "no empty slices"); + byte[] b = new byte[buf.readableBytes()]; + buf.readBytes(b); + got.write(b, 0, b.length); + buf.release(); + } + assertEquals(1, pings); + assertArrayEquals(expected, got.toByteArray()); + for (ByteBuf in : inputs) { + assertEquals(0, in.refCnt(), "every received buffer is released once its slices are"); + } + assertFalse(ch.finish()); + } + + @Test + void controlFramesBecomeFrameObjects() { + EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameDecoder()); + ch.writeInbound( + Unpooled.wrappedBuffer( + concat( + frame(0x8A, new byte[] { 1 }), + frame(0x88, new byte[] { 0x03, (byte) 0xE8, 'o', 'k' }), + frame(0x82, new byte[] { 5 })))); + PongWebSocketFrame pong = (PongWebSocketFrame) ch.readInbound(); + assertEquals( + 1, + pong.content() + .readableBytes()); + pong.release(); + CloseWebSocketFrame close = (CloseWebSocketFrame) ch.readInbound(); + assertEquals(1000, close.statusCode()); + assertEquals("ok", close.reasonText()); + close.release(); + assertNull(ch.readInbound(), "input after close is discarded"); + assertFalse(ch.finish()); + } + + @Test + void textFrameIsReportedAndTheRestDiscarded() { + EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameDecoder()); + ByteBuf in = Unpooled + .wrappedBuffer(concat(frame(0x81, new byte[] { 'h', 'i' }), frame(0x82, new byte[] { 1 }))); + ch.writeInbound(in); + TextWebSocketFrame text = (TextWebSocketFrame) ch.readInbound(); + text.release(); + assertNull(ch.readInbound()); + assertEquals(0, in.refCnt()); + assertFalse(ch.finish()); + } + + @Test + void protocolViolationsCloseAndRaise() { + byte[][] bad = { + // masked server frame + { (byte) 0x82, (byte) 0x81, 0, 0, 0, 0, 1 }, + // RSV1 + { (byte) 0xC2, 1, 1 }, + // fragmented ping + { 0x09, 0 }, + // control frame over 125 bytes + frame(0x89, new byte[126]), + // reserved control and data opcodes + { (byte) 0x8B, 0 }, { (byte) 0x83, 0 }, + // continuation without a message + { (byte) 0x80, 1, 1 }, + // new message inside a fragmented one + concat(frame(0x02, new byte[] { 1 }), frame(0x82, new byte[] { 2 })), + // non-minimal lengths + frame(0x82, new byte[10], 2), frame(0x82, new byte[300], 8), + // close with a 1-byte body, a reserved status code, invalid UTF-8 + { (byte) 0x88, 1, 3 }, frame(0x88, new byte[] { 0x03, (byte) 0xED }), + frame(0x88, new byte[] { 0x03, (byte) 0xE8, (byte) 0xC3 }), + // over the frame limit (64 here) + frame(0x82, new byte[65]), }; + for (byte[] wire : bad) { + EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameDecoder(64)); + ByteBuf in = Unpooled.wrappedBuffer(wire); + try { + ch.writeInbound(in); + ch.checkException(); + throw new AssertionError("accepted " + Arrays.toString(wire)); + } catch (CorruptedFrameException expected) { + // good + } + assertFalse(ch.isOpen(), Arrays.toString(wire)); + // Payloads that arrived before the violation were passed on; release them. + Object o; + while ((o = ch.readInbound()) != null) { + ((ByteBuf) o).release(); + } + assertEquals(0, in.refCnt(), Arrays.toString(wire)); + Object written; + while ((written = ch.readOutbound()) != null) { + ((ByteBuf) written).release(); + } + ch.finish(); + } + } + + @Test + void frameAtTheLimitIsAccepted() { + EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameDecoder(64)); + ch.writeInbound(Unpooled.wrappedBuffer(frame(0x82, new byte[64]))); + ByteBuf out = (ByteBuf) ch.readInbound(); + assertEquals(64, out.readableBytes()); + out.release(); + assertFalse(ch.finish()); + } + + @Test + void carriedHeadIsReleasedOnClose() { + EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameDecoder()); + ByteBuf in = tracked(new byte[] { (byte) 0x82, 126, 1 }, 0, 3); + ch.writeInbound(in); + assertNull(ch.readInbound()); + assertEquals(0, in.refCnt()); + assertFalse(ch.finish()); + } +} diff --git a/client-mod/src/test/java/dev/moonbridge/clientws/net/ClientFrameEncoderTest.java b/client-mod/src/test/java/dev/moonbridge/clientws/net/ClientFrameEncoderTest.java new file mode 100644 index 0000000..67b06cc --- /dev/null +++ b/client-mod/src/test/java/dev/moonbridge/clientws/net/ClientFrameEncoderTest.java @@ -0,0 +1,339 @@ +package dev.moonbridge.clientws.net; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.ByteArrayOutputStream; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Random; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.jupiter.api.Test; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelOutboundHandlerAdapter; +import io.netty.channel.ChannelPromise; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame; +import io.netty.handler.codec.http.websocketx.PongWebSocketFrame; + +class ClientFrameEncoderTest { + + /** One decoded client frame: first byte, unmasked payload. */ + static final class Frame { + + final int b0; + final byte[] payload; + final int lengthBytes; + + Frame(int b0, byte[] payload, int lengthBytes) { + this.b0 = b0; + this.payload = payload; + this.lengthBytes = lengthBytes; + } + } + + /** Parses exactly one masked frame filling {@code buf}, and releases it. */ + static Frame parse(ByteBuf buf) { + try { + int b0 = buf.readUnsignedByte(); + int b1 = buf.readUnsignedByte(); + assertTrue((b1 & 0x80) != 0, "client frame must be masked"); + long len = b1 & 0x7F; + int lengthBytes = 0; + if (len == 126) { + len = buf.readUnsignedShort(); + lengthBytes = 2; + } else if (len == 127) { + len = buf.readLong(); + lengthBytes = 8; + } + byte[] mask = new byte[4]; + buf.readBytes(mask); + byte[] payload = new byte[(int) len]; + buf.readBytes(payload); + for (int i = 0; i < payload.length; i++) { + payload[i] ^= mask[i & 3]; + } + assertFalse(buf.isReadable(), "one frame per buffer"); + return new Frame(b0, payload, lengthBytes); + } finally { + buf.release(); + } + } + + private static byte[] bytes(Random r, int n) { + byte[] b = new byte[n]; + r.nextBytes(b); + return b; + } + + private static List drain(EmbeddedChannel ch) { + ch.runPendingTasks(); + List frames = new ArrayList(); + Object o; + while ((o = ch.readOutbound()) != null) { + frames.add(parse((ByteBuf) o)); + } + return frames; + } + + @Test + void maskingRoundTripsAtEveryLengthAndOffsetModFour() { + Random r = new Random(1); + // Two writes per message: the second starts at every offset mod 4 and has every length mod 4. + for (int first = 0; first < 8; first++) { + for (int second = 1; second < 9; second++) { + EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameEncoder()); + byte[] a = bytes(r, first); + byte[] b = bytes(r, second); + ch.write(Unpooled.wrappedBuffer(a)); + ch.writeAndFlush(Unpooled.wrappedBuffer(b)); + List frames = drain(ch); + assertEquals(1, frames.size()); + assertEquals(0x82, frames.get(0).b0); + ByteArrayOutputStream expected = new ByteArrayOutputStream(); + expected.write(a, 0, a.length); + expected.write(b, 0, b.length); + assertArrayEquals(expected.toByteArray(), frames.get(0).payload, first + "+" + second); + assertFalse(ch.finish()); + } + } + } + + @Test + void lengthEncodingsAtTheBoundaries() { + Random r = new Random(2); + int[] lengths = { 125, 126, 65535, 65536 }; + int[] lengthBytes = { 0, 2, 2, 8 }; + for (int i = 0; i < lengths.length; i++) { + EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameEncoder()); + byte[] data = bytes(r, lengths[i]); + // Odd source offset: the payload is not 4-aligned in its buffer. + ByteBuf src = Unpooled.buffer(data.length + 1) + .writeByte(0) + .writeBytes(data); + src.skipBytes(1); + ch.writeAndFlush(src); + List frames = drain(ch); + assertEquals(1, frames.size(), "length " + lengths[i]); + assertEquals(lengthBytes[i], frames.get(0).lengthBytes, "length " + lengths[i]); + assertArrayEquals(data, frames.get(0).payload); + assertEquals(0, src.refCnt()); + assertFalse(ch.finish()); + } + } + + @Test + void messagesAreCappedAt64KiB() { + Random r = new Random(3); + EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameEncoder()); + byte[] big = bytes(r, ClientFrameEncoder.MAX_MESSAGE_PAYLOAD + 5); + byte[] small = bytes(r, 7); + ChannelFuture bigDone = ch.write(Unpooled.wrappedBuffer(big)); + ChannelFuture smallDone = ch.writeAndFlush(Unpooled.wrappedBuffer(small)); + List frames = drain(ch); + assertEquals(2, frames.size()); + assertEquals(ClientFrameEncoder.MAX_MESSAGE_PAYLOAD, frames.get(0).payload.length); + assertEquals(12, frames.get(1).payload.length); + ByteArrayOutputStream all = new ByteArrayOutputStream(); + for (Frame f : frames) { + all.write(f.payload, 0, f.payload.length); + } + ByteArrayOutputStream expected = new ByteArrayOutputStream(); + expected.write(big, 0, big.length); + expected.write(small, 0, small.length); + assertArrayEquals(expected.toByteArray(), all.toByteArray()); + assertTrue(bigDone.isSuccess()); + assertTrue(smallDone.isSuccess()); + assertFalse(ch.finish()); + } + + @Test + void queuedWritesShareOneMessageAndAllPromisesComplete() { + EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameEncoder()); + List futures = new ArrayList(); + for (int i = 0; i < 5; i++) { + futures.add(ch.writeAndFlush(Unpooled.wrappedBuffer(new byte[] { (byte) i, (byte) i }))); + } + assertNull(ch.readOutbound(), "nothing goes out before the batch task runs"); + for (ChannelFuture f : futures) { + assertFalse(f.isDone()); + } + List frames = drain(ch); + assertEquals(1, frames.size()); + assertArrayEquals(new byte[] { 0, 0, 1, 1, 2, 2, 3, 3, 4, 4 }, frames.get(0).payload); + for (ChannelFuture f : futures) { + assertTrue(f.isSuccess()); + } + assertFalse(ch.finish()); + } + + @Test + void emptyWritesSucceedWithoutAFrame() { + EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameEncoder()); + ByteBuf empty = Unpooled.buffer(); + assertTrue( + ch.writeAndFlush(empty) + .isSuccess()); + assertEquals(0, empty.refCnt()); + assertTrue(drain(ch).isEmpty()); + assertFalse(ch.finish()); + } + + @Test + void pongIsFlushedImmediatelyAheadOfQueuedData() { + EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameEncoder()); + ch.writeAndFlush(Unpooled.wrappedBuffer(new byte[] { 1, 2, 3 })); + ByteBuf ping = Unpooled.wrappedBuffer(new byte[] { 'p' }); + ch.writeAndFlush(new PongWebSocketFrame(ping)); + Frame pong = parse((ByteBuf) ch.readOutbound()); + assertEquals(0x8A, pong.b0); + assertArrayEquals(new byte[] { 'p' }, pong.payload); + assertEquals(0, ping.refCnt()); + assertNull(ch.readOutbound(), "queued data waits for its batch task"); + List frames = drain(ch); + assertEquals(1, frames.size()); + assertEquals(0x82, frames.get(0).b0); + assertFalse(ch.finish()); + } + + @Test + void closeFrameAndChannelCloseSendQueuedDataFirst() { + EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameEncoder()); + ch.writeAndFlush(Unpooled.wrappedBuffer(new byte[] { 9 })); + ch.writeAndFlush(new CloseWebSocketFrame(1000, "")); + Frame data = parse((ByteBuf) ch.readOutbound()); + assertArrayEquals(new byte[] { 9 }, data.payload); + Frame close = parse((ByteBuf) ch.readOutbound()); + assertEquals(0x88, close.b0); + assertArrayEquals(new byte[] { 0x03, (byte) 0xE8 }, close.payload); + + ch.writeAndFlush(Unpooled.wrappedBuffer(new byte[] { 7, 7 })); + ch.close(); + Frame last = parse((ByteBuf) ch.readOutbound()); + assertArrayEquals(new byte[] { 7, 7 }, last.payload); + assertFalse(ch.finish()); + } + + @Test + void unflushedWritesFailAndAreReleasedWhenTheChannelGoesInactive() { + EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameEncoder()); + ByteBuf data = Unpooled.wrappedBuffer(new byte[] { 1 }); + ChannelFuture f = ch.write(data); + ch.pipeline() + .fireChannelInactive(); + assertEquals(0, data.refCnt()); + assertFalse(f.isSuccess()); + assertTrue(f.isDone()); + ch.finish(); + } + + /** Head-side stand-in for the socket: records each written buffer's frame, completes the write. */ + static final class Recorder extends ChannelOutboundHandlerAdapter { + + final List frames = new ArrayList(); + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { + frames.add(parse((ByteBuf) msg)); + promise.setSuccess(); + } + + @Override + public void flush(ChannelHandlerContext ctx) {} + } + + private static Channel registered(NioEventLoopGroup group, Recorder recorder) { + Channel ch = new NioSocketChannel(); + ch.pipeline() + .addLast(recorder, new ClientFrameEncoder()); + group.register(ch) + .syncUninterruptibly(); + return ch; + } + + @Test + void lonePacketLeavesInTheTaskRightAfterItsWrite() throws Exception { + NioEventLoopGroup group = new NioEventLoopGroup(1); + try { + final Recorder recorder = new Recorder(); + final Channel ch = registered(group, recorder); + final AtomicReference seenByNextTask = new AtomicReference(); + final CountDownLatch done = new CountDownLatch(1); + ch.eventLoop() + .execute(() -> { + final ChannelFuture write = ch.writeAndFlush(Unpooled.wrappedBuffer(new byte[] { 42 })); + // Anything queued after the write runs after the batch task: no wait for another I/O pass. + ch.eventLoop() + .execute(() -> { + seenByNextTask.set(recorder.frames.size() + " " + write.isSuccess()); + done.countDown(); + }); + }); + assertTrue(done.await(10, TimeUnit.SECONDS)); + assertEquals("1 true", seenByNextTask.get()); + assertArrayEquals(new byte[] { 42 }, recorder.frames.get(0).payload); + ch.close() + .syncUninterruptibly(); + } finally { + group.shutdownGracefully(0, 1, TimeUnit.SECONDS) + .syncUninterruptibly(); + } + } + + @Test + void packetsQueuedByAnotherThreadWhileTheLoopIsBusyShareOneMessage() throws Exception { + NioEventLoopGroup group = new NioEventLoopGroup(1); + try { + final Recorder recorder = new Recorder(); + final Channel ch = registered(group, recorder); + final CountDownLatch release = new CountDownLatch(1); + ch.eventLoop() + .execute(() -> { + try { + release.await(10, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread() + .interrupt(); + } + }); + // Like vanilla's client thread: one writeAndFlush per packet, each queued as its own loop task. + List writes = new ArrayList(); + ByteArrayOutputStream expected = new ByteArrayOutputStream(); + for (int i = 0; i < 20; i++) { + byte[] packet = new byte[i + 1]; + Arrays.fill(packet, (byte) i); + expected.write(packet, 0, packet.length); + writes.add(ch.writeAndFlush(Unpooled.wrappedBuffer(packet))); + } + release.countDown(); + for (ChannelFuture f : writes) { + assertTrue(f.await(10, TimeUnit.SECONDS) && f.isSuccess()); + } + List frames = ch.eventLoop() + .submit(() -> new ArrayList(recorder.frames)) + .get(10, TimeUnit.SECONDS); + assertEquals(1, frames.size()); + assertArrayEquals(expected.toByteArray(), frames.get(0).payload); + ch.close() + .syncUninterruptibly(); + } finally { + group.shutdownGracefully(0, 1, TimeUnit.SECONDS) + .syncUninterruptibly(); + } + } +} diff --git a/client-mod/src/test/java/dev/moonbridge/clientws/net/CodecBenchmark.java b/client-mod/src/test/java/dev/moonbridge/clientws/net/CodecBenchmark.java new file mode 100644 index 0000000..ec50982 --- /dev/null +++ b/client-mod/src/test/java/dev/moonbridge/clientws/net/CodecBenchmark.java @@ -0,0 +1,496 @@ +package dev.moonbridge.clientws.net; + +import java.io.ByteArrayOutputStream; +import java.lang.management.ManagementFactory; +import java.util.ArrayList; +import java.util.List; +import java.util.Locale; +import java.util.Random; +import java.util.concurrent.Callable; +import java.util.concurrent.TimeUnit; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufAllocator; +import io.netty.buffer.Unpooled; +import io.netty.buffer.UnpooledByteBufAllocator; +import io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.channel.ChannelOption; +import io.netty.channel.ChannelOutboundHandlerAdapter; +import io.netty.channel.ChannelPromise; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.handler.codec.ByteToMessageDecoder; +import io.netty.handler.codec.CorruptedFrameException; +import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame; +import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame; +import io.netty.handler.codec.http.websocketx.WebSocket13FrameDecoder; +import io.netty.handler.codec.http.websocketx.WebSocket13FrameEncoder; +import io.netty.handler.codec.http.websocketx.WebSocketFrame; + +/** + * Client codec microbenchmark: Netty 4.0.10's {@code WebSocket13FrameEncoder/Decoder} ("base", with the frame + * wrapping the old adapter did) against {@link ClientFrameEncoder}/{@link ClientFrameDecoder} ("candidate"). + * + *

+ * Encode: a non-loop thread plays vanilla's client thread and calls {@code writeAndFlush} once per packet, in bursts of + * 1–24 packets, waiting for each burst to be written. A sink at the head of the pipeline stands in for the socket. + * Decode: a server stream (a ~2 MiB login burst in 32 KiB frames, then play traffic as the proxy sends it: batches of + * 1–32 packets per message, at most 64 KiB) is fed to the pipeline in 1460-byte reads on the event loop. + * + *

+ * Packet sizes: 88% 20–300 B, 11% 2–8 KiB, 1% 30–60 KiB, fixed seed. CPU and allocation are those of the event loop + * thread (where both codecs run); buffers are unpooled heap buffers so the JVM's per-thread allocation counter sees + * ByteBuf memory too. Usage: {@code --variant base|candidate --repeats N} (the first phase of each mode is warm-up). + */ +public final class CodecBenchmark { + + static final long SEED = 20261023L; + static final int ENCODE_PACKETS = 100_000; + static final int DECODE_PACKETS = 40_000; + static final int LOGIN_BURST = 2 * 1024 * 1024; + static final int LOGIN_FRAME = 32 * 1024; + static final int READ_SIZE = 1460; + static final int DECODE_PASSES = 8; + static final ByteBufAllocator ALLOC = new UnpooledByteBufAllocator(false); + static final byte[] CONTENT = new byte[64 * 1024]; + + static { + new Random(SEED).nextBytes(CONTENT); + } + + private static final com.sun.management.ThreadMXBean THREADS = (com.sun.management.ThreadMXBean) ManagementFactory + .getThreadMXBean(); + + static int packetSize(Random r) { + int p = r.nextInt(1000); + if (p < 880) { + return 20 + r.nextInt(281); + } else if (p < 990) { + return 2048 + r.nextInt(6145); + } + return 30720 + r.nextInt(30721); + } + + public static void main(String[] args) throws Exception { + String variant = "candidate"; + int repeats = 3; + String label = "run"; + for (int i = 0; i < args.length; i++) { + if (args[i].equals("--variant")) { + variant = args[++i]; + } else if (args[i].equals("--repeats")) { + repeats = Integer.parseInt(args[++i]); + } else if (args[i].equals("--label")) { + label = args[++i]; + } + } + boolean candidate = variant.equals("candidate"); + if (!candidate && !variant.equals("base")) { + throw new IllegalArgumentException("--variant base|candidate"); + } + System.out.println( + "# java " + System.getProperty("java.version") + + " " + + System.getProperty("java.vm.name") + + ", " + + System.getProperty("os.name") + + " " + + System.getProperty("os.arch") + + ", cpus " + + Runtime.getRuntime() + .availableProcessors()); + NioEventLoopGroup group = new NioEventLoopGroup(1); + try { + Random r = new Random(SEED); + int[] sizes = new int[ENCODE_PACKETS]; + for (int i = 0; i < sizes.length; i++) { + sizes[i] = packetSize(r); + } + int[] bursts = new int[ENCODE_PACKETS]; + for (int i = 0; i < bursts.length; i++) { + bursts[i] = 1 + r.nextInt(24); + } + byte[] stream = serverStream(new Random(SEED + 1)); + for (int rep = 0; rep <= repeats; rep++) { + String phase = rep == 0 ? "warmup" : "rep" + rep; + System.out.println( + "encode " + variant + " " + label + " " + phase + ": " + encode(group, candidate, sizes, bursts)); + System.out.println( + "decode " + variant + " " + label + " " + phase + ": " + decode(group, candidate, stream, false)); + System.out.println( + "decodesplit " + variant + + " " + + label + + " " + + phase + + ": " + + decode(group, candidate, stream, true)); + } + } finally { + group.shutdownGracefully(0, 1, TimeUnit.SECONDS) + .syncUninterruptibly(); + } + } + + // ---- encode ---- + + /** Socket stand-in: counts and releases what the encoder writes. */ + static final class Sink extends ChannelOutboundHandlerAdapter { + + long buffers; + long bytes; + long flushes; + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { + ByteBuf b = (ByteBuf) msg; + buffers++; + bytes += b.readableBytes(); + b.release(); + promise.setSuccess(); + } + + @Override + public void flush(ChannelHandlerContext ctx) { + flushes++; + } + } + + /** Base only: counts frames entering Netty's encoder (it writes a header and a body buffer per frame). */ + static final class FrameCounter extends ChannelOutboundHandlerAdapter { + + long frames; + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { + if (msg instanceof WebSocketFrame) { + frames++; + } + ctx.write(msg, promise); + } + } + + /** Base only: what the old WebSocketFrameAdapter did on both paths. */ + static final class LegacyFrames extends io.netty.channel.ChannelDuplexHandler { + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { + ByteBuf buf = (ByteBuf) msg; + if (buf.isReadable()) { + ctx.write(new BinaryWebSocketFrame(buf), promise); + } else { + buf.release(); + promise.setSuccess(); + } + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) { + if (msg instanceof BinaryWebSocketFrame || msg instanceof ContinuationWebSocketFrame) { + ctx.fireChannelRead(((WebSocketFrame) msg).content()); + } else { + ctx.fireChannelRead(msg); + } + } + } + + private static Channel channel(NioEventLoopGroup group) { + Channel ch = new NioSocketChannel(); + ch.config() + .setOption(ChannelOption.ALLOCATOR, ALLOC); + return ch; + } + + private static long[] loopCounters(Channel ch) throws Exception { + return ch.eventLoop() + .submit(new Callable() { + + @Override + public long[] call() { + long id = Thread.currentThread() + .getId(); + return new long[] { THREADS.getThreadCpuTime(id), THREADS.getThreadAllocatedBytes(id) }; + } + }) + .get(); + } + + static String encode(NioEventLoopGroup group, boolean candidate, int[] sizes, int[] bursts) throws Exception { + Channel ch = channel(group); + Sink sink = new Sink(); + FrameCounter counter = new FrameCounter(); + if (candidate) { + ch.pipeline() + .addLast(sink, new ClientFrameEncoder()); + } else { + ch.pipeline() + .addLast(sink, new WebSocket13FrameEncoder(true), counter, new LegacyFrames()); + } + group.register(ch) + .syncUninterruptibly(); + long payload = 0; + long[] before = loopCounters(ch); + long producerCpu = THREADS.getCurrentThreadCpuTime(); + long start = System.nanoTime(); + int i = 0; + int b = 0; + while (i < sizes.length) { + int burst = Math.min(bursts[b++], sizes.length - i); + ChannelFuture last = null; + for (int k = 0; k < burst; k++, i++) { + payload += sizes[i]; + last = ch.writeAndFlush(Unpooled.wrappedBuffer(CONTENT, (i * 61) & 1023, sizes[i])); + } + last.syncUninterruptibly(); + } + long wall = System.nanoTime() - start; + producerCpu = THREADS.getCurrentThreadCpuTime() - producerCpu; + long[] after = loopCounters(ch); + long messages = candidate ? sink.buffers : counter.frames; + ch.close() + .syncUninterruptibly(); + long cpu = after[0] - before[0]; + long alloc = after[1] - before[1]; + int packets = sizes.length; + return String.format( + Locale.ROOT, + "packets=%d payloadBytes=%d messages=%d messagesPer1000=%.1f flushesPer1000=%.1f wireBytes=%d" + + " loopCpuNsPerPacket=%.0f loopCpuNsPerByte=%.3f loopAllocBytesPerPacket=%.1f" + + " producerCpuNsPerPacket=%.0f wallMs=%.1f", + packets, + payload, + messages, + 1000.0 * messages / packets, + 1000.0 * sink.flushes / packets, + sink.bytes, + (double) cpu / packets, + (double) cpu / payload, + (double) alloc / packets, + (double) producerCpu / packets, + wall / 1e6); + } + + // ---- decode ---- + + /** Server frame head, unmasked. */ + private static void head(ByteArrayOutputStream out, int length) { + out.write(0x82); + if (length < 126) { + out.write(length); + } else if (length <= 0xFFFF) { + out.write(126); + out.write(length >>> 8); + out.write(length & 0xFF); + } else { + out.write(127); + for (int shift = 56; shift >= 0; shift -= 8) { + out.write((int) ((long) length >>> shift) & 0xFF); + } + } + } + + /** Appends one Minecraft frame (VarInt length, then the body) of {@code size} body bytes. */ + private static void packet(ByteArrayOutputStream out, Random r, int size) { + int v = size; + while ((v & ~0x7F) != 0) { + out.write(v & 0x7F | 0x80); + v >>>= 7; + } + out.write(v); + int off = r.nextInt(1024); + int remaining = size; + while (remaining > 0) { + int n = Math.min(remaining, CONTENT.length - off); + out.write(CONTENT, off, n); + remaining -= n; + off = 0; + } + } + + /** Appends {@code bytes} as binary messages of at most {@code max} payload bytes. */ + private static int frames(ByteArrayOutputStream out, byte[] bytes, int max) { + int messages = 0; + for (int off = 0; off < bytes.length; off += max) { + int n = Math.min(max, bytes.length - off); + head(out, n); + out.write(bytes, off, n); + messages++; + } + return messages; + } + + static long streamPayload; + static int streamMessages; + static int streamPackets; + + /** The proxy's view: a login burst in 32 KiB frames, then batches of 1–32 packets per message of ≤ 64 KiB. */ + static byte[] serverStream(Random r) { + ByteArrayOutputStream out = new ByteArrayOutputStream(64 * 1024 * 1024); + ByteArrayOutputStream mc = new ByteArrayOutputStream(); + int packets = 0; + while (mc.size() < LOGIN_BURST) { + packet(mc, r, packetSize(r)); + packets++; + } + long payload = mc.size(); + int messages = frames(out, mc.toByteArray(), LOGIN_FRAME); + int play = 0; + while (play < DECODE_PACKETS) { + int batch = 1 + r.nextInt(32); + mc.reset(); + for (int k = 0; k < batch && play < DECODE_PACKETS; k++, play++) { + packet(mc, r, packetSize(r)); + } + payload += mc.size(); + messages += frames(out, mc.toByteArray(), ClientFrameEncoder.MAX_MESSAGE_PAYLOAD); + } + streamPayload = payload; + streamMessages = messages; + streamPackets = packets + play; + return out.toByteArray(); + } + + /** End of the measured pipeline: counts and releases. */ + static final class ByteSink extends ChannelInboundHandlerAdapter { + + long bytes; + long buffers; + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) { + ByteBuf b = (ByteBuf) msg; + bytes += b.readableBytes(); + buffers++; + b.release(); + } + + @Override + public void channelReadComplete(ChannelHandlerContext ctx) {} + } + + /** Same logic as vanilla 1.7.10's {@code MessageDeserializer2} ("splitter"): one copied buffer per packet. */ + static final class VanillaSplitter extends ByteToMessageDecoder { + + @Override + protected void decode(ChannelHandlerContext ctx, ByteBuf in, List out) { + in.markReaderIndex(); + byte[] varint = new byte[3]; + for (int i = 0; i < varint.length; i++) { + if (!in.isReadable()) { + in.resetReaderIndex(); + return; + } + varint[i] = in.readByte(); + if (varint[i] >= 0) { + ByteBuf lengthBuf = Unpooled.wrappedBuffer(varint); + try { + int length = 0; + int shift = 0; + byte b; + do { + b = lengthBuf.readByte(); + length |= (b & 0x7F) << shift; + shift += 7; + } while ((b & 0x80) != 0); + if (in.readableBytes() < length) { + in.resetReaderIndex(); + return; + } + out.add(in.readBytes(length)); + return; + } finally { + lengthBuf.release(); + } + } + } + throw new CorruptedFrameException("length wider than 21-bit"); + } + } + + /** + * Decodes {@link #DECODE_PASSES} copies of {@code stream}; with {@code split}, vanilla's splitter runs behind the + * codec, so the cost of handing it slices instead of whole payloads shows up. + */ + static String decode(NioEventLoopGroup group, final boolean candidate, byte[] stream, boolean split) + throws Exception { + final Channel ch = channel(group); + final ByteSink sink = new ByteSink(); + if (candidate) { + ch.pipeline() + .addLast(new ClientFrameDecoder(), new WebSocketFrameAdapter()); + } else { + ch.pipeline() + .addLast( + new WebSocket13FrameDecoder(false, false, WebSocketPipeline.MAX_FRAME_PAYLOAD), + new LegacyFrames(), + new WebSocketFrameAdapter()); + } + if (split) { + ch.pipeline() + .addLast(new VanillaSplitter()); + } + ch.pipeline() + .addLast(sink); + group.register(ch) + .syncUninterruptibly(); + // Each pass feeds the whole stream; the reads are prepared before the measured part of each pass. + long[] result = new long[3]; + for (int pass = 0; pass < DECODE_PASSES; pass++) { + final List reads = new ArrayList(stream.length / READ_SIZE + 1); + for (int off = 0; off < stream.length; off += READ_SIZE) { + int n = Math.min(READ_SIZE, stream.length - off); + reads.add( + ALLOC.heapBuffer(READ_SIZE) + .writeBytes(stream, off, n)); + } + long[] one = ch.eventLoop() + .submit(new Callable() { + + @Override + public long[] call() { + long id = Thread.currentThread() + .getId(); + long cpu = THREADS.getThreadCpuTime(id); + long alloc = THREADS.getThreadAllocatedBytes(id); + long start = System.nanoTime(); + for (ByteBuf read : reads) { + ch.pipeline() + .fireChannelRead(read); + ch.pipeline() + .fireChannelReadComplete(); + } + return new long[] { THREADS.getThreadCpuTime(id) - cpu, + THREADS.getThreadAllocatedBytes(id) - alloc, System.nanoTime() - start }; + } + }) + .get(); + for (int k = 0; k < 3; k++) { + result[k] += one[k]; + } + } + long expectedBuffers = split ? (long) streamPackets * DECODE_PASSES : -1; + long expectedBytes = split ? sink.bytes : streamPayload * DECODE_PASSES; + if (sink.bytes != expectedBytes || split && sink.buffers != expectedBuffers) { + throw new AssertionError("decoded " + sink.bytes + " bytes in " + sink.buffers + " buffers"); + } + ch.close() + .syncUninterruptibly(); + long wire = (long) stream.length * DECODE_PASSES; + return String.format( + Locale.ROOT, + "passes=%d wireBytesPerPass=%d messagesPerPass=%d packetsPerPass=%d readsPerPass=%d buffersOutPerPass=%d" + + " cpuNsPerByte=%.3f allocBytesPerByte=%.3f wallMs=%.1f", + DECODE_PASSES, + stream.length, + streamMessages, + streamPackets, + (stream.length + READ_SIZE - 1) / READ_SIZE, + sink.buffers / DECODE_PASSES, + (double) result[0] / wire, + (double) result[1] / wire, + result[2] / 1e6); + } +} diff --git a/client-mod/src/test/java/dev/moonbridge/clientws/net/WebSocketFrameAdapterTest.java b/client-mod/src/test/java/dev/moonbridge/clientws/net/WebSocketFrameAdapterTest.java index 4c74909..c3d4e9b 100644 --- a/client-mod/src/test/java/dev/moonbridge/clientws/net/WebSocketFrameAdapterTest.java +++ b/client-mod/src/test/java/dev/moonbridge/clientws/net/WebSocketFrameAdapterTest.java @@ -11,7 +11,6 @@ import org.junit.jupiter.api.Test; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.embedded.EmbeddedChannel; -import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame; import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame; import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; import io.netty.handler.codec.http.websocketx.PongWebSocketFrame; @@ -25,7 +24,7 @@ class WebSocketFrameAdapterTest { EmbeddedChannel ch = new EmbeddedChannel(new WebSocketFrameAdapter()); ByteBuf payload = Unpooled.buffer() .writeBytes(new byte[] { 1, 2, 3 }); - ch.writeInbound(new BinaryWebSocketFrame(payload)); + ch.writeInbound(payload); ByteBuf out = (ByteBuf) ch.readInbound(); assertSame(payload, out); assertEquals(1, out.refCnt()); @@ -37,7 +36,7 @@ class WebSocketFrameAdapterTest { void emptyAndControlFramesAreReleased() { EmbeddedChannel ch = new EmbeddedChannel(new WebSocketFrameAdapter()); ByteBuf empty = Unpooled.buffer(); - ch.writeInbound(new BinaryWebSocketFrame(empty)); + ch.writeInbound(empty); assertNull(ch.readInbound()); assertEquals(0, empty.refCnt()); @@ -86,21 +85,4 @@ class WebSocketFrameAdapterTest { reply.release(); assertFalse(ch.isOpen()); } - - @Test - void outboundBuffersBecomeBinaryFrames() { - EmbeddedChannel ch = new EmbeddedChannel(new WebSocketFrameAdapter()); - ByteBuf data = Unpooled.buffer() - .writeBytes(new byte[] { 4, 5 }); - ch.writeOutbound(data); - BinaryWebSocketFrame frame = (BinaryWebSocketFrame) ch.readOutbound(); - assertSame(data, frame.content()); - frame.release(); - - ByteBuf empty = Unpooled.buffer(); - ch.writeOutbound(empty); - assertNull(ch.readOutbound()); - assertEquals(0, empty.refCnt()); - ch.finish(); - } } diff --git a/client-mod/src/test/java/dev/moonbridge/clientws/net/WebSocketPipelineTest.java b/client-mod/src/test/java/dev/moonbridge/clientws/net/WebSocketPipelineTest.java index f8fca43..3c76c54 100644 --- a/client-mod/src/test/java/dev/moonbridge/clientws/net/WebSocketPipelineTest.java +++ b/client-mod/src/test/java/dev/moonbridge/clientws/net/WebSocketPipelineTest.java @@ -34,6 +34,7 @@ import io.netty.buffer.Unpooled; import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInitializer; +import io.netty.channel.ChannelOption; import io.netty.channel.SimpleChannelInboundHandler; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioServerSocketChannel; @@ -72,6 +73,9 @@ class WebSocketPipelineTest { @Override protected void initChannel(Channel ch) { + // Vanilla's initializer turns Nagle on before our TAIL hook runs. + ch.config() + .setOption(ChannelOption.TCP_NODELAY, false); stub.set(VanillaStub.installInto(ch.pipeline(), writeOnActive)); WebSocketPipeline.install(ch.pipeline(), address); } @@ -187,7 +191,20 @@ class WebSocketPipelineTest { synchronized (serverGot) { assertArrayEquals(expected, serverGot.toByteArray()); } - assertEquals(Arrays.asList(5, 1, 300, 70000), serverFrameSizes, "one binary frame per write"); + // Writes queued before a batch runs share one message of at most 64 KiB. The three early writes are one + // batch; the large write joins it if the test thread queued it before that batch ran. + List separate = Arrays.asList(306, 65536, 70000 - 65536); + List joined = Arrays.asList(65536, 70306 - 65536); + synchronized (serverFrameSizes) { + assertTrue( + serverFrameSizes.equals(separate) || serverFrameSizes.equals(joined), + String.valueOf(serverFrameSizes)); + } + assertEquals( + Boolean.TRUE, + client.config() + .getOption(ChannelOption.TCP_NODELAY), + "ws turns Nagle off again"); assertEquals("/mc?x=1", path.get()); assertEquals("127.0.0.1:" + port, host.get()); diff --git a/testing/benchmarks/results/2026-10-23-client-websocket-codec.md b/testing/benchmarks/results/2026-10-23-client-websocket-codec.md new file mode 100644 index 0000000..ff828b0 --- /dev/null +++ b/testing/benchmarks/results/2026-10-23-client-websocket-codec.md @@ -0,0 +1,14 @@ +# Client mod: own WebSocket codec instead of Netty 4.0.10's — 2026-10-23 + +## Decision rule (set before the measured runs) + +Encoder (`ClientFrameEncoder`) is kept only if all hold: +- median messages per 1000 packets is at most a third of the base median; +- median event-loop CPU ns/packet is not above the base median; +- a lone packet still leaves in the event-loop task right after its write (`ClientFrameEncoderTest.lonePacketLeavesInTheTaskRightAfterItsWrite`). + +Decoder (`ClientFrameDecoder`) is kept only if, in both decode views (codec only, and codec plus vanilla's splitter), the median event-loop CPU ns/byte and the median allocated bytes/byte are each at least 25% below the base median, and the candidate's min–max range does not overlap the base's for either metric. + +If the decoder misses and the encoder meets its rule, only the encoder is kept. + +Order: six JVMs, base/candidate/candidate/base/base/candidate (the starting variant alternates per pair), each with one warm-up phase and three measured phases per mode. Results are the median and [min–max] of the 9 measured phases per variant. From b72c609aecdc3560c228246b81ffbcfbff81b657 Mon Sep 17 00:00:00 2001 From: Layue13 Date: Fri, 23 Oct 2026 16:44:35 +0800 Subject: [PATCH 2/2] perf(client-mod): keep the batching encoder, drop the slice decoder Verdict of the 2026-10-23 client codec record: ClientFrameEncoder sends 12x fewer messages at 51-56% less event-loop CPU per packet and is kept. ClientFrameDecoder missed the rule (codec-only CPU -8-18% with overlapping ranges; worse once vanilla's splitter runs behind it), so the client keeps Netty's WebSocket13FrameDecoder. README protocol notes cover TCP_NODELAY and one message per batch. Co-Authored-By: Claude Opus 5.5 --- client-mod/README.md | 7 +- .../clientws/net/ClientFrameDecoder.java | 256 ---------------- .../clientws/net/ClientFrameEncoder.java | 4 +- .../clientws/net/WebSocketFrameAdapter.java | 13 +- .../clientws/net/WebSocketPipeline.java | 3 +- .../clientws/net/ClientFrameDecoderTest.java | 276 ------------------ .../clientws/net/CodecBenchmark.java | 276 +----------------- .../net/WebSocketFrameAdapterTest.java | 5 +- .../2026-10-23-client-websocket-codec-raw.txt | 84 ++++++ .../2026-10-23-client-websocket-codec.md | 69 +++++ 10 files changed, 185 insertions(+), 808 deletions(-) delete mode 100644 client-mod/src/main/java/dev/moonbridge/clientws/net/ClientFrameDecoder.java delete mode 100644 client-mod/src/test/java/dev/moonbridge/clientws/net/ClientFrameDecoderTest.java create mode 100644 testing/benchmarks/results/2026-10-23-client-websocket-codec-raw.txt diff --git a/client-mod/README.md b/client-mod/README.md index a5a408b..4b4ccd7 100644 --- a/client-mod/README.md +++ b/client-mod/README.md @@ -30,6 +30,8 @@ - 路径只允许可打印 ASCII 且不含空格，其他字符请先做百分号编码。 - 发给后端的 Minecraft 握手包里的地址/端口就是 URL 的主机名和端口。 - 每个 WebSocket 消息都是 BINARY 帧，内容是原版 TCP 字节流的一段（已经过原版的加密和长度前缀），分段边界任意。服务器发来 TEXT 帧视为协议错误并断开。 +- 出站按批发送：原版每个包调用一次 `writeAndFlush`，模组把事件循环里已排队的包合并成一条 BINARY 消息（每条最多 64 KiB），单独一个包也会在同一轮事件循环里立即发出；pong 不排队，立即发送。服务器端不能假设一条消息对应一个包。 +- ws/wss 连接会重新开启 `TCP_NODELAY`（原版关闭了它，即开启 Nagle）：包已经按批合并，Nagle 只会让下一条消息白等上一条的 ACK。普通 TCP 连接保持原版设置。 ## 常见问题 @@ -55,6 +57,7 @@ Gradle 守护进程使用 JDK 25（`gradle/gradle-daemon-jvm.properties`，可 ## 代码结构 -- `dev.moonbridge.clientws.net`：与 Minecraft 无关、可单独测试的部分。`WebSocketAddress`（URL 解析）、`WebSocketHandshakeHandler`（RFC 6455 握手，握手完成前缓存出站写入）、`WebSocketFrameAdapter`（帧 ⇄ 原始字节流）、`WebSocketPipeline`（安装到管线最前面）、`WebSocketConnectContext`（把 URL 从地址解析传到建立连接的线程）。 +- `dev.moonbridge.clientws.net`：与 Minecraft 无关、可单独测试的部分。`WebSocketAddress`（URL 解析）、`WebSocketHandshakeHandler`（RFC 6455 握手，握手完成前缓存出站写入）、`WebSocketFrameAdapter`（入站帧 → 原始字节流）、`ClientFrameEncoder`（出站按批成帧）、`WebSocketPipeline`（安装到管线最前面）、`WebSocketConnectContext`（把 URL 从地址解析传到建立连接的线程）。 - `dev.moonbridge.clientws.mixins.early`：早期 Mixin，目标为 `ServerAddress.func_78860_a`、`NetworkManager.provideLanClient`（`func_150726_a`）及其匿名 `ChannelInitializer`（`NetworkManager$2`）、`GuiConnecting.func_146367_a`、`OldServerPinger.func_147225_b`。 -- 帧编解码使用 Minecraft 自带的 Netty 4.0.10.Final（`WebSocket13FrameEncoder/Decoder`、`SslHandler`），没有打包其他 Netty。 +- 出站帧由自己的 `ClientFrameEncoder` 编码（按批合并、每次 4 字节掩码）；入站解码和 TLS 使用 Minecraft 自带的 Netty 4.0.10.Final（`WebSocket13FrameDecoder`、`SslHandler`），没有打包其他 Netty。取舍依据见 `testing/benchmarks/results/2026-10-23-client-websocket-codec.md`。 +- `./gradlew codecBenchmark -PbenchArgs="--variant base|candidate --repeats 3"`：编码器微基准（测试源码，不打包）。 diff --git a/client-mod/src/main/java/dev/moonbridge/clientws/net/ClientFrameDecoder.java b/client-mod/src/main/java/dev/moonbridge/clientws/net/ClientFrameDecoder.java deleted file mode 100644 index 63bc01f..0000000 --- a/client-mod/src/main/java/dev/moonbridge/clientws/net/ClientFrameDecoder.java +++ /dev/null @@ -1,256 +0,0 @@ -package dev.moonbridge.clientws.net; - -import java.nio.charset.CharacterCodingException; -import java.nio.charset.CodingErrorAction; -import java.nio.charset.StandardCharsets; - -import io.netty.buffer.ByteBuf; -import io.netty.buffer.Unpooled; -import io.netty.channel.ChannelFutureListener; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; -import io.netty.handler.codec.CorruptedFrameException; -import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame; -import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; -import io.netty.handler.codec.http.websocketx.PongWebSocketFrame; -import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; - -/** - * Client-side WebSocket frame decoder for server (unmasked) frames. - * - *

- * Binary and continuation payloads are passed on as plain {@link ByteBuf}s: retained slices of the received buffer, - * emitted as soon as their bytes arrive, so a large frame is never gathered and nothing is copied. Only a frame head - * split across reads (at most 14 bytes, or a control frame of up to 125 bytes) is copied into a small carry buffer. - * Ping, pong and close become the usual frame objects; a text frame becomes a {@link TextWebSocketFrame} with an empty - * payload, after which input is ignored (the adapter closes with 1003). - * - *

- * Checks match Netty's {@code WebSocket13FrameDecoder(false, false, 16 MiB)}: masked frames, RSV bits, reserved - * opcodes, fragmented or oversized control frames, non-minimal lengths, bad continuations, invalid close bodies and - * frames over {@link WebSocketPipeline#MAX_FRAME_PAYLOAD} are protocol violations, which close the connection and - * raise a {@link CorruptedFrameException}. Input after a close frame or a violation is discarded. - * - *

- * This is not a {@code ByteToMessageDecoder}: in Netty 4.0.10 that class calls {@code discardSomeReadBytes()} on its - * cumulation before it passes decoded messages on, which moves bytes under slices that were just taken from it. - */ -public final class ClientFrameDecoder extends ChannelInboundHandlerAdapter { - - private static final int MAX_HEAD = 14 + 125; - - private final long maxPayload; - private ByteBuf carry; - private long payloadLeft; - private boolean fragmented; - private boolean discarding; - - public ClientFrameDecoder() { - this(WebSocketPipeline.MAX_FRAME_PAYLOAD); - } - - ClientFrameDecoder(long maxPayload) { - this.maxPayload = maxPayload; - } - - @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { - if (!(msg instanceof ByteBuf)) { - ctx.fireChannelRead(msg); - return; - } - ByteBuf in = (ByteBuf) msg; - try { - if (carry != null && !completeCarriedHead(ctx, in)) { - return; - } - decode(ctx, in); - } finally { - in.release(); - } - } - - private void decode(ChannelHandlerContext ctx, ByteBuf in) { - while (in.isReadable()) { - if (discarding) { - in.skipBytes(in.readableBytes()); - return; - } - if (payloadLeft > 0) { - int n = (int) Math.min(payloadLeft, in.readableBytes()); - payloadLeft -= n; - ctx.fireChannelRead( - in.readSlice(n) - .retain()); - continue; - } - int head = headLength(in); - if (head < 0 || head > in.readableBytes()) { - carry = ctx.alloc() - .buffer(MAX_HEAD); - carry.writeBytes(in); - return; - } - readHead(ctx, in); - } - } - - /** Moves just enough bytes from {@code in} to finish the carried head; true once it has been decoded. */ - private boolean completeCarriedHead(ChannelHandlerContext ctx, ByteBuf in) { - while (true) { - int head = headLength(carry); - int want = (head < 0 ? 2 : head) - carry.readableBytes(); - if (want <= 0) { - break; - } - if (!in.isReadable()) { - return false; - } - carry.writeBytes(in, Math.min(want, in.readableBytes())); - } - ByteBuf head = carry; - carry = null; - try { - readHead(ctx, head); - } finally { - head.release(); - } - return true; - } - - /** - * Bytes needed for the frame head starting at the reader index, including a control frame's payload; -1 if fewer - * than two bytes are readable. Masking keys are not counted: a masked frame is rejected first. - */ - private static int headLength(ByteBuf buf) { - if (buf.readableBytes() < 2) { - return -1; - } - int b0 = buf.getUnsignedByte(buf.readerIndex()); - int len7 = buf.getUnsignedByte(buf.readerIndex() + 1) & 0x7F; - if ((b0 & 0x08) != 0) { - // Control frame: the head carries the whole payload, unless the length is already invalid. - return len7 <= 125 ? 2 + len7 : 2; - } - return len7 < 126 ? 2 : len7 == 126 ? 4 : 10; - } - - private void readHead(ChannelHandlerContext ctx, ByteBuf buf) { - int b0 = buf.readUnsignedByte(); - int b1 = buf.readUnsignedByte(); - boolean fin = (b0 & 0x80) != 0; - int opcode = b0 & 0x0F; - if ((b0 & 0x70) != 0) { - throw violation(ctx, "RSV != 0 and no extension negotiated, RSV:" + ((b0 & 0x70) >> 4)); - } - if ((b1 & 0x80) != 0) { - throw violation(ctx, "masked server to client frame"); - } - int len7 = b1 & 0x7F; - if (opcode >= 8) { - if (!fin) { - throw violation(ctx, "fragmented control frame"); - } - if (len7 > 125) { - throw violation(ctx, "control frame with payload length > 125 octets"); - } - if (opcode > ClientFrameEncoder.OPCODE_PONG) { - throw violation(ctx, "control frame using reserved opcode " + opcode); - } - if (opcode == ClientFrameEncoder.OPCODE_CLOSE) { - checkCloseBody(ctx, buf, len7); - } - ByteBuf payload = len7 == 0 ? Unpooled.EMPTY_BUFFER : buf.readBytes(len7); - if (opcode == ClientFrameEncoder.OPCODE_CLOSE) { - discarding = true; - ctx.fireChannelRead(new CloseWebSocketFrame(true, 0, payload)); - } else if (opcode == ClientFrameEncoder.OPCODE_PING) { - ctx.fireChannelRead(new PingWebSocketFrame(payload)); - } else { - ctx.fireChannelRead(new PongWebSocketFrame(payload)); - } - return; - } - if (opcode > ClientFrameEncoder.OPCODE_BINARY) { - throw violation(ctx, "data frame using reserved opcode " + opcode); - } - if (opcode == ClientFrameEncoder.OPCODE_CONTINUATION && !fragmented) { - throw violation(ctx, "received continuation data frame outside fragmented message"); - } - if (opcode != ClientFrameEncoder.OPCODE_CONTINUATION && fragmented) { - throw violation(ctx, "received non-continuation data frame while inside fragmented message"); - } - long length = len7; - if (len7 == 126) { - length = buf.readUnsignedShort(); - if (length < 126) { - throw violation(ctx, "invalid data frame length (not using minimal length encoding)"); - } - } else if (len7 == 127) { - length = buf.readLong(); - if (length < 65536) { - throw violation(ctx, "invalid data frame length (not using minimal length encoding)"); - } - } - if (length > maxPayload) { - throw violation(ctx, "Max frame length of " + maxPayload + " has been exceeded."); - } - if (opcode == ClientFrameEncoder.OPCODE_TEXT) { - discarding = true; - ctx.fireChannelRead(new TextWebSocketFrame(Unpooled.EMPTY_BUFFER)); - return; - } - fragmented = !fin; - payloadLeft = length; - } - - private void checkCloseBody(ChannelHandlerContext ctx, ByteBuf buf, int length) { - if (length == 0) { - return; - } - if (length == 1) { - throw violation(ctx, "Invalid close frame body"); - } - int status = buf.getUnsignedShort(buf.readerIndex()); - if (status <= 999 || status >= 1004 && status <= 1006 || status >= 1012 && status <= 2999) { - throw violation(ctx, "Invalid close frame getStatus code: " + status); - } - try { - StandardCharsets.UTF_8.newDecoder() - .onMalformedInput(CodingErrorAction.REPORT) - .onUnmappableCharacter(CodingErrorAction.REPORT) - .decode(buf.nioBuffer(buf.readerIndex() + 2, length - 2)); - } catch (CharacterCodingException e) { - throw violation(ctx, "Invalid close frame reason text. Invalid UTF-8 bytes"); - } - } - - private CorruptedFrameException violation(ChannelHandlerContext ctx, String reason) { - discarding = true; - payloadLeft = 0; - if (ctx.channel() - .isActive()) { - ctx.writeAndFlush(Unpooled.EMPTY_BUFFER) - .addListener(ChannelFutureListener.CLOSE); - } - return new CorruptedFrameException(reason); - } - - @Override - public void channelInactive(ChannelHandlerContext ctx) throws Exception { - releaseCarry(); - ctx.fireChannelInactive(); - } - - @Override - public void handlerRemoved(ChannelHandlerContext ctx) { - releaseCarry(); - } - - private void releaseCarry() { - if (carry != null) { - carry.release(); - carry = null; - } - } -} diff --git a/client-mod/src/main/java/dev/moonbridge/clientws/net/ClientFrameEncoder.java b/client-mod/src/main/java/dev/moonbridge/clientws/net/ClientFrameEncoder.java index caee2a4..36758fa 100644 --- a/client-mod/src/main/java/dev/moonbridge/clientws/net/ClientFrameEncoder.java +++ b/client-mod/src/main/java/dev/moonbridge/clientws/net/ClientFrameEncoder.java @@ -28,7 +28,9 @@ import io.netty.handler.codec.http.websocketx.WebSocketFrame; * end of the event loop's queue; by the time it runs, every packet the client thread queued in the meantime has been * written too, and the whole batch goes out as one binary message (split at {@link #MAX_MESSAGE_PAYLOAD}). A lone * packet still leaves in the same loop iteration: the task runs right after the write that scheduled it. A write's - * promise completes when the message carrying its last byte has been written. + * promise completes when the message carrying its last byte has been written, so after write tasks queued behind it; + * vanilla's one promise-driven state change (enabling encryption once the encryption response is written) is safe + * because the client sends nothing else until the server answers that response. * *

* Each message is one buffer: the header, then the payload masked four bytes at a time. {@link WebSocketFrame}s are diff --git a/client-mod/src/main/java/dev/moonbridge/clientws/net/WebSocketFrameAdapter.java b/client-mod/src/main/java/dev/moonbridge/clientws/net/WebSocketFrameAdapter.java index 6fcea06..1104eae 100644 --- a/client-mod/src/main/java/dev/moonbridge/clientws/net/WebSocketFrameAdapter.java +++ b/client-mod/src/main/java/dev/moonbridge/clientws/net/WebSocketFrameAdapter.java @@ -6,26 +6,29 @@ import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame; import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame; +import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame; import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; import io.netty.handler.codec.http.websocketx.PongWebSocketFrame; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; +import io.netty.handler.codec.http.websocketx.WebSocketFrame; /** * Turns the WebSocket message stream back into the raw Minecraft byte stream. * *

- * {@link ClientFrameDecoder} already passes binary and continuation payloads on as plain {@link ByteBuf}s, so the - * vanilla splitter sees one concatenated byte stream regardless of how the server chunked it; they go through as-is - * (ownership moves with them). Ping is answered with pong, close ends the connection, and a text frame is a protocol + * Inbound: the payload of every binary and continuation frame is passed on as-is (ownership moves with it), so the + * vanilla splitter sees one concatenated byte stream regardless of how the server chunked it. Ping is answered with + * pong, close ends the connection, and a text frame is a protocol * error. Outbound, vanilla's {@link ByteBuf}s go straight to {@link ClientFrameEncoder}, which frames them. */ public final class WebSocketFrameAdapter extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { - if (msg instanceof ByteBuf) { - ByteBuf payload = (ByteBuf) msg; + if (msg instanceof BinaryWebSocketFrame || msg instanceof ContinuationWebSocketFrame) { + ByteBuf payload = ((WebSocketFrame) msg).content(); if (payload.isReadable()) { ctx.fireChannelRead(payload); } else { diff --git a/client-mod/src/main/java/dev/moonbridge/clientws/net/WebSocketPipeline.java b/client-mod/src/main/java/dev/moonbridge/clientws/net/WebSocketPipeline.java index 81ff685..b51abb3 100644 --- a/client-mod/src/main/java/dev/moonbridge/clientws/net/WebSocketPipeline.java +++ b/client-mod/src/main/java/dev/moonbridge/clientws/net/WebSocketPipeline.java @@ -11,6 +11,7 @@ import javax.net.ssl.SSLParameters; import io.netty.channel.ChannelException; import io.netty.channel.ChannelOption; import io.netty.channel.ChannelPipeline; +import io.netty.handler.codec.http.websocketx.WebSocket13FrameDecoder; import io.netty.handler.ssl.SslHandler; /** @@ -63,7 +64,7 @@ public final class WebSocketPipeline { // addFirst in reverse so the final order is ssl, handshake, decoder, encoder, frames, vanilla... pipeline.addFirst(FRAMES, new WebSocketFrameAdapter()); pipeline.addFirst(FRAME_ENCODER, new ClientFrameEncoder()); - pipeline.addFirst(FRAME_DECODER, new ClientFrameDecoder()); + pipeline.addFirst(FRAME_DECODER, new WebSocket13FrameDecoder(false, false, MAX_FRAME_PAYLOAD)); pipeline.addFirst(HANDSHAKE, new WebSocketHandshakeHandler(address)); if (address.isSecure()) { pipeline.addFirst(SSL, new SslHandler(newClientEngine(tls, address))); diff --git a/client-mod/src/test/java/dev/moonbridge/clientws/net/ClientFrameDecoderTest.java b/client-mod/src/test/java/dev/moonbridge/clientws/net/ClientFrameDecoderTest.java deleted file mode 100644 index f6dfe1f..0000000 --- a/client-mod/src/test/java/dev/moonbridge/clientws/net/ClientFrameDecoderTest.java +++ /dev/null @@ -1,276 +0,0 @@ -package dev.moonbridge.clientws.net; - -import static org.junit.jupiter.api.Assertions.assertArrayEquals; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNull; -import static org.junit.jupiter.api.Assertions.assertSame; -import static org.junit.jupiter.api.Assertions.assertTrue; - -import java.io.ByteArrayOutputStream; -import java.nio.charset.StandardCharsets; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; -import java.util.Random; - -import org.junit.jupiter.api.Test; - -import io.netty.buffer.ByteBuf; -import io.netty.buffer.PooledByteBufAllocator; -import io.netty.buffer.Unpooled; -import io.netty.channel.embedded.EmbeddedChannel; -import io.netty.handler.codec.CorruptedFrameException; -import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame; -import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; -import io.netty.handler.codec.http.websocketx.PongWebSocketFrame; -import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; - -class ClientFrameDecoderTest { - - /** Server frame with any flags and the shortest length form, or a forced one ({@code lengthBytes} 0, 2 or 8). */ - static byte[] frame(int b0, byte[] payload, int lengthBytes) { - ByteArrayOutputStream f = new ByteArrayOutputStream(); - f.write(b0); - if (lengthBytes == 0) { - f.write(payload.length); - } else if (lengthBytes == 2) { - f.write(126); - f.write(payload.length >>> 8); - f.write(payload.length & 0xff); - } else { - f.write(127); - for (int shift = 56; shift >= 0; shift -= 8) { - f.write((int) ((long) payload.length >>> shift) & 0xff); - } - } - f.write(payload, 0, payload.length); - return f.toByteArray(); - } - - static byte[] frame(int b0, byte[] payload) { - return frame(b0, payload, payload.length < 126 ? 0 : payload.length <= 0xFFFF ? 2 : 8); - } - - private static byte[] bytes(Random r, int n) { - byte[] b = new byte[n]; - r.nextBytes(b); - return b; - } - - private static byte[] concat(byte[]... parts) { - ByteArrayOutputStream o = new ByteArrayOutputStream(); - for (byte[] p : parts) { - o.write(p, 0, p.length); - } - return o.toByteArray(); - } - - /** Pooled direct buffers, so Netty 4.0.10's leak detector tracks them (heap buffers are not tracked). */ - private static ByteBuf tracked(byte[] b, int from, int to) { - return PooledByteBufAllocator.DEFAULT.directBuffer(to - from) - .writeBytes(b, from, to - from); - } - - @Test - void binaryPayloadIsARetainedSliceOfTheInput() { - EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameDecoder()); - byte[] payload = { 1, 2, 3, 4, 5 }; - byte[] wire = frame(0x82, payload); - ByteBuf in = tracked(wire, 0, wire.length); - ch.writeInbound(in); - ByteBuf out = (ByteBuf) ch.readInbound(); - assertSame(in, out.unwrap(), "a slice of the received buffer, not a copy"); - assertEquals(1, in.refCnt(), "the decoder dropped its reference; the slice holds the only one"); - byte[] got = new byte[out.readableBytes()]; - out.readBytes(got); - assertArrayEquals(payload, got); - out.release(); - assertEquals(0, in.refCnt()); - assertFalse(ch.finish()); - } - - @Test - void reassemblesAStreamSplitAtEveryPoint() { - Random r = new Random(7); - byte[] a = bytes(r, 300); // 16-bit length - byte[] b = bytes(r, 70000); // 64-bit length - byte[] c1 = bytes(r, 2); - byte[] c2 = bytes(r, 125); - byte[] wire = concat( - frame(0x82, a), - frame(0x02, c1), // fragmented message with a ping in between - frame(0x89, new byte[] { 'p', 'i' }), - frame(0x80, c2), - frame(0x82, b), - frame(0x82, new byte[0])); - byte[] expected = concat(a, c1, c2, b); - int[] splits = new int[400]; - for (int i = 0; i < splits.length; i++) { - splits[i] = i; // every split inside the first frames' heads and payload starts - } - for (int split : splits) { - check(wire, expected, new int[] { split }); - } - // And arbitrary chunking, including 1-byte reads over the whole of a small stream. - for (int chunk : new int[] { 1, 3, 7, 1460 }) { - List points = new ArrayList(); - for (int p = chunk; p < wire.length && points.size() < 5000; p += chunk) { - points.add(p); - } - int[] cuts = new int[points.size()]; - for (int i = 0; i < cuts.length; i++) { - cuts[i] = points.get(i); - } - check(wire, expected, cuts); - } - } - - private static void check(byte[] wire, byte[] expected, int[] cuts) { - EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameDecoder()); - List inputs = new ArrayList(); - int from = 0; - for (int i = 0; i <= cuts.length; i++) { - int to = i < cuts.length ? cuts[i] : wire.length; - if (to <= from) { - continue; - } - ByteBuf in = tracked(wire, from, to); - inputs.add(in); - ch.writeInbound(in); - from = to; - } - ByteArrayOutputStream got = new ByteArrayOutputStream(); - int pings = 0; - Object o; - while ((o = ch.readInbound()) != null) { - if (o instanceof PingWebSocketFrame) { - PingWebSocketFrame ping = (PingWebSocketFrame) o; - assertEquals( - "pi", - ping.content() - .toString(StandardCharsets.US_ASCII)); - ping.release(); - pings++; - continue; - } - ByteBuf buf = (ByteBuf) o; - assertTrue(buf.isReadable(), "no empty slices"); - byte[] b = new byte[buf.readableBytes()]; - buf.readBytes(b); - got.write(b, 0, b.length); - buf.release(); - } - assertEquals(1, pings); - assertArrayEquals(expected, got.toByteArray()); - for (ByteBuf in : inputs) { - assertEquals(0, in.refCnt(), "every received buffer is released once its slices are"); - } - assertFalse(ch.finish()); - } - - @Test - void controlFramesBecomeFrameObjects() { - EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameDecoder()); - ch.writeInbound( - Unpooled.wrappedBuffer( - concat( - frame(0x8A, new byte[] { 1 }), - frame(0x88, new byte[] { 0x03, (byte) 0xE8, 'o', 'k' }), - frame(0x82, new byte[] { 5 })))); - PongWebSocketFrame pong = (PongWebSocketFrame) ch.readInbound(); - assertEquals( - 1, - pong.content() - .readableBytes()); - pong.release(); - CloseWebSocketFrame close = (CloseWebSocketFrame) ch.readInbound(); - assertEquals(1000, close.statusCode()); - assertEquals("ok", close.reasonText()); - close.release(); - assertNull(ch.readInbound(), "input after close is discarded"); - assertFalse(ch.finish()); - } - - @Test - void textFrameIsReportedAndTheRestDiscarded() { - EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameDecoder()); - ByteBuf in = Unpooled - .wrappedBuffer(concat(frame(0x81, new byte[] { 'h', 'i' }), frame(0x82, new byte[] { 1 }))); - ch.writeInbound(in); - TextWebSocketFrame text = (TextWebSocketFrame) ch.readInbound(); - text.release(); - assertNull(ch.readInbound()); - assertEquals(0, in.refCnt()); - assertFalse(ch.finish()); - } - - @Test - void protocolViolationsCloseAndRaise() { - byte[][] bad = { - // masked server frame - { (byte) 0x82, (byte) 0x81, 0, 0, 0, 0, 1 }, - // RSV1 - { (byte) 0xC2, 1, 1 }, - // fragmented ping - { 0x09, 0 }, - // control frame over 125 bytes - frame(0x89, new byte[126]), - // reserved control and data opcodes - { (byte) 0x8B, 0 }, { (byte) 0x83, 0 }, - // continuation without a message - { (byte) 0x80, 1, 1 }, - // new message inside a fragmented one - concat(frame(0x02, new byte[] { 1 }), frame(0x82, new byte[] { 2 })), - // non-minimal lengths - frame(0x82, new byte[10], 2), frame(0x82, new byte[300], 8), - // close with a 1-byte body, a reserved status code, invalid UTF-8 - { (byte) 0x88, 1, 3 }, frame(0x88, new byte[] { 0x03, (byte) 0xED }), - frame(0x88, new byte[] { 0x03, (byte) 0xE8, (byte) 0xC3 }), - // over the frame limit (64 here) - frame(0x82, new byte[65]), }; - for (byte[] wire : bad) { - EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameDecoder(64)); - ByteBuf in = Unpooled.wrappedBuffer(wire); - try { - ch.writeInbound(in); - ch.checkException(); - throw new AssertionError("accepted " + Arrays.toString(wire)); - } catch (CorruptedFrameException expected) { - // good - } - assertFalse(ch.isOpen(), Arrays.toString(wire)); - // Payloads that arrived before the violation were passed on; release them. - Object o; - while ((o = ch.readInbound()) != null) { - ((ByteBuf) o).release(); - } - assertEquals(0, in.refCnt(), Arrays.toString(wire)); - Object written; - while ((written = ch.readOutbound()) != null) { - ((ByteBuf) written).release(); - } - ch.finish(); - } - } - - @Test - void frameAtTheLimitIsAccepted() { - EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameDecoder(64)); - ch.writeInbound(Unpooled.wrappedBuffer(frame(0x82, new byte[64]))); - ByteBuf out = (ByteBuf) ch.readInbound(); - assertEquals(64, out.readableBytes()); - out.release(); - assertFalse(ch.finish()); - } - - @Test - void carriedHeadIsReleasedOnClose() { - EmbeddedChannel ch = new EmbeddedChannel(new ClientFrameDecoder()); - ByteBuf in = tracked(new byte[] { (byte) 0x82, 126, 1 }, 0, 3); - ch.writeInbound(in); - assertNull(ch.readInbound()); - assertEquals(0, in.refCnt()); - assertFalse(ch.finish()); - } -} diff --git a/client-mod/src/test/java/dev/moonbridge/clientws/net/CodecBenchmark.java b/client-mod/src/test/java/dev/moonbridge/clientws/net/CodecBenchmark.java index ec50982..727b8ab 100644 --- a/client-mod/src/test/java/dev/moonbridge/clientws/net/CodecBenchmark.java +++ b/client-mod/src/test/java/dev/moonbridge/clientws/net/CodecBenchmark.java @@ -1,9 +1,6 @@ package dev.moonbridge.clientws.net; -import java.io.ByteArrayOutputStream; import java.lang.management.ManagementFactory; -import java.util.ArrayList; -import java.util.List; import java.util.Locale; import java.util.Random; import java.util.concurrent.Callable; @@ -16,44 +13,34 @@ import io.netty.buffer.UnpooledByteBufAllocator; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.channel.ChannelOption; import io.netty.channel.ChannelOutboundHandlerAdapter; import io.netty.channel.ChannelPromise; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioSocketChannel; -import io.netty.handler.codec.ByteToMessageDecoder; -import io.netty.handler.codec.CorruptedFrameException; import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame; -import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame; -import io.netty.handler.codec.http.websocketx.WebSocket13FrameDecoder; import io.netty.handler.codec.http.websocketx.WebSocket13FrameEncoder; import io.netty.handler.codec.http.websocketx.WebSocketFrame; /** - * Client codec microbenchmark: Netty 4.0.10's {@code WebSocket13FrameEncoder/Decoder} ("base", with the frame - * wrapping the old adapter did) against {@link ClientFrameEncoder}/{@link ClientFrameDecoder} ("candidate"). + * Client encoder microbenchmark: Netty 4.0.10's {@code WebSocket13FrameEncoder} ("base", with the frame wrapping the + * old adapter did) against {@link ClientFrameEncoder} ("candidate"). * *

- * Encode: a non-loop thread plays vanilla's client thread and calls {@code writeAndFlush} once per packet, in bursts of - * 1–24 packets, waiting for each burst to be written. A sink at the head of the pipeline stands in for the socket. - * Decode: a server stream (a ~2 MiB login burst in 32 KiB frames, then play traffic as the proxy sends it: batches of - * 1–32 packets per message, at most 64 KiB) is fed to the pipeline in 1460-byte reads on the event loop. + * A non-loop thread plays vanilla's client thread and calls {@code writeAndFlush} once per packet, in bursts of 1–24 + * packets, waiting for each burst to be written. A sink at the head of the pipeline stands in for the socket. Packet + * sizes: 88% 20–300 B, 11% 2–8 KiB, 1% 30–60 KiB, fixed seed. CPU and allocation are those of the event loop thread + * (where the codec runs); buffers are unpooled heap buffers so the JVM's per-thread allocation counter sees ByteBuf + * memory too. Usage: {@code --variant base|candidate --repeats N --label X} (the first phase is warm-up). * *

- * Packet sizes: 88% 20–300 B, 11% 2–8 KiB, 1% 30–60 KiB, fixed seed. CPU and allocation are those of the event loop - * thread (where both codecs run); buffers are unpooled heap buffers so the JVM's per-thread allocation counter sees - * ByteBuf memory too. Usage: {@code --variant base|candidate --repeats N} (the first phase of each mode is warm-up). + * The decoder candidate measured in testing/benchmarks/results/2026-10-23-client-websocket-codec.md (and rejected) is + * in commit 1decc44, together with this benchmark's decode modes. */ public final class CodecBenchmark { static final long SEED = 20261023L; static final int ENCODE_PACKETS = 100_000; - static final int DECODE_PACKETS = 40_000; - static final int LOGIN_BURST = 2 * 1024 * 1024; - static final int LOGIN_FRAME = 32 * 1024; - static final int READ_SIZE = 1460; - static final int DECODE_PASSES = 8; static final ByteBufAllocator ALLOC = new UnpooledByteBufAllocator(false); static final byte[] CONTENT = new byte[64 * 1024]; @@ -113,21 +100,10 @@ public final class CodecBenchmark { for (int i = 0; i < bursts.length; i++) { bursts[i] = 1 + r.nextInt(24); } - byte[] stream = serverStream(new Random(SEED + 1)); for (int rep = 0; rep <= repeats; rep++) { String phase = rep == 0 ? "warmup" : "rep" + rep; System.out.println( "encode " + variant + " " + label + " " + phase + ": " + encode(group, candidate, sizes, bursts)); - System.out.println( - "decode " + variant + " " + label + " " + phase + ": " + decode(group, candidate, stream, false)); - System.out.println( - "decodesplit " + variant - + " " - + label - + " " - + phase - + ": " - + decode(group, candidate, stream, true)); } } finally { group.shutdownGracefully(0, 1, TimeUnit.SECONDS) @@ -173,8 +149,8 @@ public final class CodecBenchmark { } } - /** Base only: what the old WebSocketFrameAdapter did on both paths. */ - static final class LegacyFrames extends io.netty.channel.ChannelDuplexHandler { + /** Base only: what the old WebSocketFrameAdapter did with outbound buffers. */ + static final class LegacyFrames extends ChannelOutboundHandlerAdapter { @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { @@ -186,15 +162,6 @@ public final class CodecBenchmark { promise.setSuccess(); } } - - @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) { - if (msg instanceof BinaryWebSocketFrame || msg instanceof ContinuationWebSocketFrame) { - ctx.fireChannelRead(((WebSocketFrame) msg).content()); - } else { - ctx.fireChannelRead(msg); - } - } } private static Channel channel(NioEventLoopGroup group) { @@ -272,225 +239,4 @@ public final class CodecBenchmark { (double) producerCpu / packets, wall / 1e6); } - - // ---- decode ---- - - /** Server frame head, unmasked. */ - private static void head(ByteArrayOutputStream out, int length) { - out.write(0x82); - if (length < 126) { - out.write(length); - } else if (length <= 0xFFFF) { - out.write(126); - out.write(length >>> 8); - out.write(length & 0xFF); - } else { - out.write(127); - for (int shift = 56; shift >= 0; shift -= 8) { - out.write((int) ((long) length >>> shift) & 0xFF); - } - } - } - - /** Appends one Minecraft frame (VarInt length, then the body) of {@code size} body bytes. */ - private static void packet(ByteArrayOutputStream out, Random r, int size) { - int v = size; - while ((v & ~0x7F) != 0) { - out.write(v & 0x7F | 0x80); - v >>>= 7; - } - out.write(v); - int off = r.nextInt(1024); - int remaining = size; - while (remaining > 0) { - int n = Math.min(remaining, CONTENT.length - off); - out.write(CONTENT, off, n); - remaining -= n; - off = 0; - } - } - - /** Appends {@code bytes} as binary messages of at most {@code max} payload bytes. */ - private static int frames(ByteArrayOutputStream out, byte[] bytes, int max) { - int messages = 0; - for (int off = 0; off < bytes.length; off += max) { - int n = Math.min(max, bytes.length - off); - head(out, n); - out.write(bytes, off, n); - messages++; - } - return messages; - } - - static long streamPayload; - static int streamMessages; - static int streamPackets; - - /** The proxy's view: a login burst in 32 KiB frames, then batches of 1–32 packets per message of ≤ 64 KiB. */ - static byte[] serverStream(Random r) { - ByteArrayOutputStream out = new ByteArrayOutputStream(64 * 1024 * 1024); - ByteArrayOutputStream mc = new ByteArrayOutputStream(); - int packets = 0; - while (mc.size() < LOGIN_BURST) { - packet(mc, r, packetSize(r)); - packets++; - } - long payload = mc.size(); - int messages = frames(out, mc.toByteArray(), LOGIN_FRAME); - int play = 0; - while (play < DECODE_PACKETS) { - int batch = 1 + r.nextInt(32); - mc.reset(); - for (int k = 0; k < batch && play < DECODE_PACKETS; k++, play++) { - packet(mc, r, packetSize(r)); - } - payload += mc.size(); - messages += frames(out, mc.toByteArray(), ClientFrameEncoder.MAX_MESSAGE_PAYLOAD); - } - streamPayload = payload; - streamMessages = messages; - streamPackets = packets + play; - return out.toByteArray(); - } - - /** End of the measured pipeline: counts and releases. */ - static final class ByteSink extends ChannelInboundHandlerAdapter { - - long bytes; - long buffers; - - @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) { - ByteBuf b = (ByteBuf) msg; - bytes += b.readableBytes(); - buffers++; - b.release(); - } - - @Override - public void channelReadComplete(ChannelHandlerContext ctx) {} - } - - /** Same logic as vanilla 1.7.10's {@code MessageDeserializer2} ("splitter"): one copied buffer per packet. */ - static final class VanillaSplitter extends ByteToMessageDecoder { - - @Override - protected void decode(ChannelHandlerContext ctx, ByteBuf in, List out) { - in.markReaderIndex(); - byte[] varint = new byte[3]; - for (int i = 0; i < varint.length; i++) { - if (!in.isReadable()) { - in.resetReaderIndex(); - return; - } - varint[i] = in.readByte(); - if (varint[i] >= 0) { - ByteBuf lengthBuf = Unpooled.wrappedBuffer(varint); - try { - int length = 0; - int shift = 0; - byte b; - do { - b = lengthBuf.readByte(); - length |= (b & 0x7F) << shift; - shift += 7; - } while ((b & 0x80) != 0); - if (in.readableBytes() < length) { - in.resetReaderIndex(); - return; - } - out.add(in.readBytes(length)); - return; - } finally { - lengthBuf.release(); - } - } - } - throw new CorruptedFrameException("length wider than 21-bit"); - } - } - - /** - * Decodes {@link #DECODE_PASSES} copies of {@code stream}; with {@code split}, vanilla's splitter runs behind the - * codec, so the cost of handing it slices instead of whole payloads shows up. - */ - static String decode(NioEventLoopGroup group, final boolean candidate, byte[] stream, boolean split) - throws Exception { - final Channel ch = channel(group); - final ByteSink sink = new ByteSink(); - if (candidate) { - ch.pipeline() - .addLast(new ClientFrameDecoder(), new WebSocketFrameAdapter()); - } else { - ch.pipeline() - .addLast( - new WebSocket13FrameDecoder(false, false, WebSocketPipeline.MAX_FRAME_PAYLOAD), - new LegacyFrames(), - new WebSocketFrameAdapter()); - } - if (split) { - ch.pipeline() - .addLast(new VanillaSplitter()); - } - ch.pipeline() - .addLast(sink); - group.register(ch) - .syncUninterruptibly(); - // Each pass feeds the whole stream; the reads are prepared before the measured part of each pass. - long[] result = new long[3]; - for (int pass = 0; pass < DECODE_PASSES; pass++) { - final List reads = new ArrayList(stream.length / READ_SIZE + 1); - for (int off = 0; off < stream.length; off += READ_SIZE) { - int n = Math.min(READ_SIZE, stream.length - off); - reads.add( - ALLOC.heapBuffer(READ_SIZE) - .writeBytes(stream, off, n)); - } - long[] one = ch.eventLoop() - .submit(new Callable() { - - @Override - public long[] call() { - long id = Thread.currentThread() - .getId(); - long cpu = THREADS.getThreadCpuTime(id); - long alloc = THREADS.getThreadAllocatedBytes(id); - long start = System.nanoTime(); - for (ByteBuf read : reads) { - ch.pipeline() - .fireChannelRead(read); - ch.pipeline() - .fireChannelReadComplete(); - } - return new long[] { THREADS.getThreadCpuTime(id) - cpu, - THREADS.getThreadAllocatedBytes(id) - alloc, System.nanoTime() - start }; - } - }) - .get(); - for (int k = 0; k < 3; k++) { - result[k] += one[k]; - } - } - long expectedBuffers = split ? (long) streamPackets * DECODE_PASSES : -1; - long expectedBytes = split ? sink.bytes : streamPayload * DECODE_PASSES; - if (sink.bytes != expectedBytes || split && sink.buffers != expectedBuffers) { - throw new AssertionError("decoded " + sink.bytes + " bytes in " + sink.buffers + " buffers"); - } - ch.close() - .syncUninterruptibly(); - long wire = (long) stream.length * DECODE_PASSES; - return String.format( - Locale.ROOT, - "passes=%d wireBytesPerPass=%d messagesPerPass=%d packetsPerPass=%d readsPerPass=%d buffersOutPerPass=%d" - + " cpuNsPerByte=%.3f allocBytesPerByte=%.3f wallMs=%.1f", - DECODE_PASSES, - stream.length, - streamMessages, - streamPackets, - (stream.length + READ_SIZE - 1) / READ_SIZE, - sink.buffers / DECODE_PASSES, - (double) result[0] / wire, - (double) result[1] / wire, - result[2] / 1e6); - } } diff --git a/client-mod/src/test/java/dev/moonbridge/clientws/net/WebSocketFrameAdapterTest.java b/client-mod/src/test/java/dev/moonbridge/clientws/net/WebSocketFrameAdapterTest.java index c3d4e9b..20dd414 100644 --- a/client-mod/src/test/java/dev/moonbridge/clientws/net/WebSocketFrameAdapterTest.java +++ b/client-mod/src/test/java/dev/moonbridge/clientws/net/WebSocketFrameAdapterTest.java @@ -11,6 +11,7 @@ import org.junit.jupiter.api.Test; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame; import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame; import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; import io.netty.handler.codec.http.websocketx.PongWebSocketFrame; @@ -24,7 +25,7 @@ class WebSocketFrameAdapterTest { EmbeddedChannel ch = new EmbeddedChannel(new WebSocketFrameAdapter()); ByteBuf payload = Unpooled.buffer() .writeBytes(new byte[] { 1, 2, 3 }); - ch.writeInbound(payload); + ch.writeInbound(new BinaryWebSocketFrame(payload)); ByteBuf out = (ByteBuf) ch.readInbound(); assertSame(payload, out); assertEquals(1, out.refCnt()); @@ -36,7 +37,7 @@ class WebSocketFrameAdapterTest { void emptyAndControlFramesAreReleased() { EmbeddedChannel ch = new EmbeddedChannel(new WebSocketFrameAdapter()); ByteBuf empty = Unpooled.buffer(); - ch.writeInbound(empty); + ch.writeInbound(new BinaryWebSocketFrame(empty)); assertNull(ch.readInbound()); assertEquals(0, empty.refCnt()); diff --git a/testing/benchmarks/results/2026-10-23-client-websocket-codec-raw.txt b/testing/benchmarks/results/2026-10-23-client-websocket-codec-raw.txt new file mode 100644 index 0000000..1769118 --- /dev/null +++ b/testing/benchmarks/results/2026-10-23-client-websocket-codec-raw.txt @@ -0,0 +1,84 @@ +# jvm1 base uptime: load averages: 5.11 6.02 6.17 +# java 17.0.20.1 OpenJDK 64-Bit Server VM, Mac OS X aarch64, cpus 10 +encode base jvm1 warmup: packets=100000 payloadBytes=117529702 messages=100000 messagesPer1000=1000.0 flushesPer1000=1000.0 wireBytes=118263428 loopCpuNsPerPacket=9512 loopCpuNsPerByte=8.093 loopAllocBytesPerPacket=1363.8 producerCpuNsPerPacket=1336 wallMs=1210.9 +decode base jvm1 warmup: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=2670 cpuNsPerByte=0.550 allocBytesPerByte=1.006 wallMs=235.9 +decodesplit base jvm1 warmup: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.760 allocBytesPerByte=2.544 wallMs=313.1 +encode base jvm1 rep1: packets=100000 payloadBytes=117529702 messages=100000 messagesPer1000=1000.0 flushesPer1000=1000.0 wireBytes=118263428 loopCpuNsPerPacket=8377 loopCpuNsPerByte=7.128 loopAllocBytesPerPacket=1362.2 producerCpuNsPerPacket=1036 wallMs=939.1 +decode base jvm1 rep1: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=2670 cpuNsPerByte=0.170 allocBytesPerByte=1.006 wallMs=74.5 +decodesplit base jvm1 rep1: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.412 allocBytesPerByte=2.544 wallMs=169.2 +encode base jvm1 rep2: packets=100000 payloadBytes=117529702 messages=100000 messagesPer1000=1000.0 flushesPer1000=1000.0 wireBytes=118263428 loopCpuNsPerPacket=6857 loopCpuNsPerByte=5.834 loopAllocBytesPerPacket=1362.2 producerCpuNsPerPacket=753 wallMs=780.7 +decode base jvm1 rep2: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=2670 cpuNsPerByte=0.141 allocBytesPerByte=1.006 wallMs=61.3 +decodesplit base jvm1 rep2: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.365 allocBytesPerByte=2.544 wallMs=150.8 +encode base jvm1 rep3: packets=100000 payloadBytes=117529702 messages=100000 messagesPer1000=1000.0 flushesPer1000=1000.0 wireBytes=118263428 loopCpuNsPerPacket=6468 loopCpuNsPerByte=5.504 loopAllocBytesPerPacket=1362.2 producerCpuNsPerPacket=699 wallMs=729.1 +decode base jvm1 rep3: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=2670 cpuNsPerByte=0.127 allocBytesPerByte=1.006 wallMs=54.3 +decodesplit base jvm1 rep3: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.311 allocBytesPerByte=2.544 wallMs=128.6 +# jvm2 candidate uptime: load averages: 6.44 6.27 6.26 +# java 17.0.20.1 OpenJDK 64-Bit Server VM, Mac OS X aarch64, cpus 10 +encode candidate jvm2 warmup: packets=100000 payloadBytes=117529702 messages=8549 messagesPer1000=85.5 flushesPer1000=82.7 wireBytes=117599518 loopCpuNsPerPacket=2616 loopCpuNsPerByte=2.226 loopAllocBytesPerPacket=1203.6 producerCpuNsPerPacket=808 wallMs=334.6 +decode candidate jvm2 warmup: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=37155 cpuNsPerByte=0.226 allocBytesPerByte=0.035 wallMs=92.4 +decodesplit candidate jvm2 warmup: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.521 allocBytesPerByte=2.794 wallMs=228.1 +encode candidate jvm2 rep1: packets=100000 payloadBytes=117529702 messages=8370 messagesPer1000=83.7 flushesPer1000=80.8 wireBytes=117598126 loopCpuNsPerPacket=3194 loopCpuNsPerByte=2.718 loopAllocBytesPerPacket=1203.1 producerCpuNsPerPacket=824 wallMs=412.4 +decode candidate jvm2 rep1: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=37155 cpuNsPerByte=0.119 allocBytesPerByte=0.035 wallMs=48.6 +decodesplit candidate jvm2 rep1: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.457 allocBytesPerByte=2.794 wallMs=192.9 +encode candidate jvm2 rep2: packets=100000 payloadBytes=117529702 messages=8340 messagesPer1000=83.4 flushesPer1000=80.5 wireBytes=117597936 loopCpuNsPerPacket=3573 loopCpuNsPerByte=3.040 loopAllocBytesPerPacket=1203.0 producerCpuNsPerPacket=830 wallMs=441.1 +decode candidate jvm2 rep2: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=37155 cpuNsPerByte=0.149 allocBytesPerByte=0.035 wallMs=62.0 +decodesplit candidate jvm2 rep2: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.490 allocBytesPerByte=2.794 wallMs=205.0 +encode candidate jvm2 rep3: packets=100000 payloadBytes=117529702 messages=8283 messagesPer1000=82.8 flushesPer1000=79.9 wireBytes=117597510 loopCpuNsPerPacket=2486 loopCpuNsPerByte=2.115 loopAllocBytesPerPacket=1202.9 producerCpuNsPerPacket=543 wallMs=294.4 +decode candidate jvm2 rep3: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=37155 cpuNsPerByte=0.120 allocBytesPerByte=0.035 wallMs=49.0 +decodesplit candidate jvm2 rep3: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.399 allocBytesPerByte=2.794 wallMs=168.0 +# jvm3 candidate uptime: load averages: 6.40 6.27 6.25 +# java 17.0.20.1 OpenJDK 64-Bit Server VM, Mac OS X aarch64, cpus 10 +encode candidate jvm3 warmup: packets=100000 payloadBytes=117529702 messages=8630 messagesPer1000=86.3 flushesPer1000=83.5 wireBytes=117600122 loopCpuNsPerPacket=2976 loopCpuNsPerByte=2.532 loopAllocBytesPerPacket=1203.7 producerCpuNsPerPacket=860 wallMs=372.4 +decode candidate jvm3 warmup: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=37155 cpuNsPerByte=0.197 allocBytesPerByte=0.035 wallMs=80.0 +decodesplit candidate jvm3 warmup: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.475 allocBytesPerByte=2.794 wallMs=210.9 +encode candidate jvm3 rep1: packets=100000 payloadBytes=117529702 messages=8287 messagesPer1000=82.9 flushesPer1000=79.9 wireBytes=117597532 loopCpuNsPerPacket=2123 loopCpuNsPerByte=1.806 loopAllocBytesPerPacket=1203.0 producerCpuNsPerPacket=429 wallMs=246.7 +decode candidate jvm3 rep1: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=37155 cpuNsPerByte=0.103 allocBytesPerByte=0.035 wallMs=41.7 +decodesplit candidate jvm3 rep1: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.342 allocBytesPerByte=2.794 wallMs=142.3 +encode candidate jvm3 rep2: packets=100000 payloadBytes=117529702 messages=8302 messagesPer1000=83.0 flushesPer1000=80.1 wireBytes=117597636 loopCpuNsPerPacket=2067 loopCpuNsPerByte=1.759 loopAllocBytesPerPacket=1203.0 producerCpuNsPerPacket=388 wallMs=237.5 +decode candidate jvm3 rep2: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=37155 cpuNsPerByte=0.096 allocBytesPerByte=0.035 wallMs=38.6 +decodesplit candidate jvm3 rep2: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.301 allocBytesPerByte=2.794 wallMs=127.5 +encode candidate jvm3 rep3: packets=100000 payloadBytes=117529702 messages=8280 messagesPer1000=82.8 flushesPer1000=79.8 wireBytes=117597486 loopCpuNsPerPacket=1832 loopCpuNsPerByte=1.559 loopAllocBytesPerPacket=1202.9 producerCpuNsPerPacket=344 wallMs=208.1 +decode candidate jvm3 rep3: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=37155 cpuNsPerByte=0.089 allocBytesPerByte=0.035 wallMs=35.9 +decodesplit candidate jvm3 rep3: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.292 allocBytesPerByte=2.794 wallMs=121.7 +# jvm4 base uptime: load averages: 7.25 6.44 6.32 +# java 17.0.20.1 OpenJDK 64-Bit Server VM, Mac OS X aarch64, cpus 10 +encode base jvm4 warmup: packets=100000 payloadBytes=117529702 messages=100000 messagesPer1000=1000.0 flushesPer1000=1000.0 wireBytes=118263428 loopCpuNsPerPacket=5345 loopCpuNsPerByte=4.548 loopAllocBytesPerPacket=1363.7 producerCpuNsPerPacket=636 wallMs=587.8 +decode base jvm4 warmup: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=2670 cpuNsPerByte=0.280 allocBytesPerByte=1.006 wallMs=114.5 +decodesplit base jvm4 warmup: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.381 allocBytesPerByte=2.544 wallMs=156.3 +encode base jvm4 rep1: packets=100000 payloadBytes=117529702 messages=100000 messagesPer1000=1000.0 flushesPer1000=1000.0 wireBytes=118263428 loopCpuNsPerPacket=4673 loopCpuNsPerByte=3.976 loopAllocBytesPerPacket=1362.2 producerCpuNsPerPacket=448 wallMs=502.7 +decode base jvm4 rep1: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=2670 cpuNsPerByte=0.117 allocBytesPerByte=1.006 wallMs=50.7 +decodesplit base jvm4 rep1: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.291 allocBytesPerByte=2.544 wallMs=123.9 +encode base jvm4 rep2: packets=100000 payloadBytes=117529702 messages=100000 messagesPer1000=1000.0 flushesPer1000=1000.0 wireBytes=118263428 loopCpuNsPerPacket=4667 loopCpuNsPerByte=3.971 loopAllocBytesPerPacket=1362.2 producerCpuNsPerPacket=408 wallMs=499.0 +decode base jvm4 rep2: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=2670 cpuNsPerByte=0.122 allocBytesPerByte=1.006 wallMs=49.2 +decodesplit base jvm4 rep2: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.265 allocBytesPerByte=2.544 wallMs=108.7 +encode base jvm4 rep3: packets=100000 payloadBytes=117529702 messages=100000 messagesPer1000=1000.0 flushesPer1000=1000.0 wireBytes=118263428 loopCpuNsPerPacket=4141 loopCpuNsPerByte=3.523 loopAllocBytesPerPacket=1362.2 producerCpuNsPerPacket=356 wallMs=440.2 +decode base jvm4 rep3: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=2670 cpuNsPerByte=0.096 allocBytesPerByte=1.006 wallMs=38.6 +decodesplit base jvm4 rep3: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.232 allocBytesPerByte=2.544 wallMs=96.2 +# jvm5 base uptime: load averages: 7.15 6.44 6.31 +# java 17.0.20.1 OpenJDK 64-Bit Server VM, Mac OS X aarch64, cpus 10 +encode base jvm5 warmup: packets=100000 payloadBytes=117529702 messages=100000 messagesPer1000=1000.0 flushesPer1000=1000.0 wireBytes=118263428 loopCpuNsPerPacket=4443 loopCpuNsPerByte=3.780 loopAllocBytesPerPacket=1363.7 producerCpuNsPerPacket=485 wallMs=482.0 +decode base jvm5 warmup: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=2670 cpuNsPerByte=0.216 allocBytesPerByte=1.006 wallMs=87.8 +decodesplit base jvm5 warmup: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.318 allocBytesPerByte=2.544 wallMs=130.4 +encode base jvm5 rep1: packets=100000 payloadBytes=117529702 messages=100000 messagesPer1000=1000.0 flushesPer1000=1000.0 wireBytes=118263428 loopCpuNsPerPacket=4199 loopCpuNsPerByte=3.573 loopAllocBytesPerPacket=1362.2 producerCpuNsPerPacket=384 wallMs=449.1 +decode base jvm5 rep1: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=2670 cpuNsPerByte=0.103 allocBytesPerByte=1.006 wallMs=45.3 +decodesplit base jvm5 rep1: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.258 allocBytesPerByte=2.544 wallMs=105.5 +encode base jvm5 rep2: packets=100000 payloadBytes=117529702 messages=100000 messagesPer1000=1000.0 flushesPer1000=1000.0 wireBytes=118263428 loopCpuNsPerPacket=3833 loopCpuNsPerByte=3.261 loopAllocBytesPerPacket=1362.2 producerCpuNsPerPacket=324 wallMs=407.8 +decode base jvm5 rep2: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=2670 cpuNsPerByte=0.104 allocBytesPerByte=1.006 wallMs=44.8 +decodesplit base jvm5 rep2: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.234 allocBytesPerByte=2.544 wallMs=96.2 +encode base jvm5 rep3: packets=100000 payloadBytes=117529702 messages=100000 messagesPer1000=1000.0 flushesPer1000=1000.0 wireBytes=118263428 loopCpuNsPerPacket=3838 loopCpuNsPerByte=3.265 loopAllocBytesPerPacket=1362.2 producerCpuNsPerPacket=322 wallMs=408.1 +decode base jvm5 rep3: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=2670 cpuNsPerByte=0.095 allocBytesPerByte=1.006 wallMs=38.4 +decodesplit base jvm5 rep3: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.237 allocBytesPerByte=2.544 wallMs=97.4 +# jvm6 candidate uptime: load averages: 7.38 6.49 6.34 +# java 17.0.20.1 OpenJDK 64-Bit Server VM, Mac OS X aarch64, cpus 10 +encode candidate jvm6 warmup: packets=100000 payloadBytes=117529702 messages=8522 messagesPer1000=85.2 flushesPer1000=82.4 wireBytes=117599294 loopCpuNsPerPacket=2336 loopCpuNsPerByte=1.987 loopAllocBytesPerPacket=1203.6 producerCpuNsPerPacket=635 wallMs=282.9 +decode candidate jvm6 warmup: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=37155 cpuNsPerByte=0.161 allocBytesPerByte=0.035 wallMs=65.3 +decodesplit candidate jvm6 warmup: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.398 allocBytesPerByte=2.794 wallMs=180.8 +encode candidate jvm6 rep1: packets=100000 payloadBytes=117529702 messages=8282 messagesPer1000=82.8 flushesPer1000=79.9 wireBytes=117597492 loopCpuNsPerPacket=2101 loopCpuNsPerByte=1.788 loopAllocBytesPerPacket=1202.9 producerCpuNsPerPacket=418 wallMs=240.8 +decode candidate jvm6 rep1: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=37155 cpuNsPerByte=0.095 allocBytesPerByte=0.035 wallMs=38.4 +decodesplit candidate jvm6 rep1: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.320 allocBytesPerByte=2.794 wallMs=134.9 +encode candidate jvm6 rep2: packets=100000 payloadBytes=117529702 messages=8290 messagesPer1000=82.9 flushesPer1000=80.0 wireBytes=117597558 loopCpuNsPerPacket=1897 loopCpuNsPerByte=1.614 loopAllocBytesPerPacket=1203.0 producerCpuNsPerPacket=365 wallMs=215.6 +decode candidate jvm6 rep2: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=37155 cpuNsPerByte=0.088 allocBytesPerByte=0.035 wallMs=35.4 +decodesplit candidate jvm6 rep2: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.280 allocBytesPerByte=2.794 wallMs=117.8 +encode candidate jvm6 rep3: packets=100000 payloadBytes=117529702 messages=8295 messagesPer1000=83.0 flushesPer1000=80.0 wireBytes=117597584 loopCpuNsPerPacket=1785 loopCpuNsPerByte=1.519 loopAllocBytesPerPacket=1203.0 producerCpuNsPerPacket=355 wallMs=204.0 +decode candidate jvm6 rep3: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=37155 cpuNsPerByte=0.096 allocBytesPerByte=0.035 wallMs=39.0 +decodesplit candidate jvm6 rep3: passes=8 wireBytesPerPass=50365992 messagesPerPass=2670 packetsPerPass=41858 readsPerPass=34498 buffersOutPerPass=41858 cpuNsPerByte=0.298 allocBytesPerByte=2.794 wallMs=125.4 diff --git a/testing/benchmarks/results/2026-10-23-client-websocket-codec.md b/testing/benchmarks/results/2026-10-23-client-websocket-codec.md index ff828b0..8b78854 100644 --- a/testing/benchmarks/results/2026-10-23-client-websocket-codec.md +++ b/testing/benchmarks/results/2026-10-23-client-websocket-codec.md @@ -1,5 +1,15 @@ # Client mod: own WebSocket codec instead of Netty 4.0.10's — 2026-10-23 +Base revision: `6224d3d` (main). The client mod used Minecraft's bundled Netty 4.0.10 codec: `WebSocket13FrameEncoder(true)` and `WebSocket13FrameDecoder(false, false, 16 MiB)`. Vanilla calls `writeAndFlush` once per packet, so every packet became its own masked binary message. That encoder writes a separate header buffer and a body copy masked byte by byte, with a mask from `Math.random()`. The decoder is a `ReplayingDecoder` that copies every payload. + +Candidates (commit `1decc44`): +- **Encoder.** `ClientFrameEncoder` queues outbound buffers. A flush schedules one event-loop task per batch, so packets that the client thread queued in the meantime share one binary message of at most 64 KiB. Each message is one buffer: the header, then the payload masked 4 bytes at a time with a `ThreadLocalRandom` mask. Pongs skip the queue and are flushed at once. +- **Decoder.** `ClientFrameDecoder` emitted binary and continuation payloads as retained slices of each read, as soon as their bytes arrived, so nothing was copied. It is not a `ByteToMessageDecoder`: in 4.0.10 that class calls `discardSomeReadBytes()` on its cumulation before it fires the decoded messages, which would move bytes under slices it just handed out. + +Also in this change, without a benchmark: ws/wss connections set `TCP_NODELAY` back to true after vanilla's initializer turned Nagle on. Plain TCP is unchanged. + +**Outcome: encoder adopted, decoder rejected** (the client keeps Netty's `WebSocket13FrameDecoder`). + ## Decision rule (set before the measured runs) Encoder (`ClientFrameEncoder`) is kept only if all hold: @@ -12,3 +22,62 @@ Decoder (`ClientFrameDecoder`) is kept only if, in both decode views (codec only If the decoder misses and the encoder meets its rule, only the encoder is kept. Order: six JVMs, base/candidate/candidate/base/base/candidate (the starting variant alternates per pair), each with one warm-up phase and three measured phases per mode. Results are the median and [min–max] of the 9 measured phases per variant. + +The rule handed to this work asked only for codec-only decode numbers. The splitter view was added after one unrecorded screening run per variant. That run showed the candidate handing vanilla's splitter 14× more buffers (one per read instead of one per message). With the splitter behind it, total allocation came out higher than base. The decoder only changes what the splitter receives, so it has to win with the splitter included. This makes the decoder rule stricter, never looser. The verdict would be the same without it: on its own, the codec-only view misses the CPU criterion. + +## Benchmark + +`./gradlew codecBenchmark -PbenchArgs="--variant base|candidate --repeats 3 --label jvmN"` in `client-mod/` (`CodecBenchmark`, test sources, never packaged; the decode modes exist only in `1decc44`). Packet sizes, fixed seed: 88% 20–300 B, 11% 2–8 KiB, 1% 30–60 KiB. + +- **Encode.** 100 000 packets (112 MiB of payload). A non-loop thread plays vanilla's client thread: one `writeAndFlush` per packet, in bursts of 1–24 packets, waiting for each burst's last write to complete. A sink at the pipeline head stands in for the socket. The base pipeline includes the old adapter's `ByteBuf` → `BinaryWebSocketFrame` wrapping. Messages are counted as frames entering Netty's encoder (base) or buffers leaving ours (candidate). +- **Decode.** The proxy's view of a session: a ~2 MiB login burst in 32 KiB frames, then 40 000 packets in messages of 1–32 packets each, capped at 64 KiB, framed as Minecraft frames (VarInt length + body). The 48 MiB stream is delivered in 1460-byte reads on the event loop, 8 passes per phase. "Decode + splitter" adds a copy of vanilla 1.7.10's `MessageDeserializer2` logic, which copies each packet out with `readBytes`. +- **Measurement.** CPU and allocation are the event-loop thread's own (`ThreadMXBean`); the producer thread's CPU is reported separately. All buffers come from an unpooled heap allocator, so the per-thread allocation counter sees ByteBuf memory too. + +## Environment and method + +Apple M4 (10 cores), macOS 27.0.1, Zulu OpenJDK 17.0.20.1 (the client-mod test toolchain), Netty 4.0.10.Final, `-Xms1g -Xmx1g -Dio.netty.leakDetectionLevel=disabled`. The host was not quiet: load average was 5.1–7.4 at the start of each JVM (logged in the raw file). Sources were a separate Gradle run using about 4 cores and `appstoreagent` using about 1. The proxy session's own benchmarks were paused while these ran. Every decode phase checked the decoded byte count (and, with the splitter, the packet count). The candidate's correctness is covered by the unit tests, not by the benchmark. + +## Results + +Median and [min–max] of 9 measured phases per variant. "Pair ratios" are the candidate median over the base median within each JVM pair (jvm1/2, jvm3/4, jvm5/6), which cancels slow drift in host load. + +| Encode | Base | Candidate | Pair ratios | +| --- | ---: | ---: | --- | +| Messages per 1000 packets | 1000 [1000–1000] | 82.9 [82.8–83.7] | 0.083, 0.083, 0.083 | +| Flushes (socket writes) per 1000 packets | 1000 [1000–1000] | 80.0 [79.8–80.8] | 0.081, 0.080, 0.080 | +| Event-loop CPU ns/packet | 4667 [3833–8377] | 2101 [1785–3573] | 0.466, 0.443, 0.494 | +| Event-loop CPU ns/byte | 3.97 [3.26–7.13] | 1.79 [1.52–3.04] | same | +| Event-loop allocation B/packet | 1362 [1362–1362] | 1203 [1203–1203] | 0.883 ×3 | +| Producer-thread CPU ns/packet | 408 [322–1036] | 418 [344–830] | 1.094, 0.951, 1.127 | + +| Decode | Base | Candidate | Pair ratios | +| --- | ---: | ---: | --- | +| Codec only: CPU ns/byte | 0.117 [0.095–0.170] | 0.096 [0.088–0.149] | 0.851, 0.821, 0.922 | +| Codec only: allocation B/byte | 1.006 [1.006–1.006] | 0.035 [0.035–0.035] | 0.035 ×3 | +| Codec only: buffers passed on per pass | 2670 | 37155 | 13.9 | +| With splitter: CPU ns/byte | 0.265 [0.232–0.412] | 0.320 [0.280–0.490] | 1.252, 1.136, 1.257 | +| With splitter: allocation B/byte | 2.544 [2.544–2.544] | 2.794 [2.794–2.794] | 1.098 ×3 | + +Raw output: [2026-10-23-client-websocket-codec-raw.txt](2026-10-23-client-websocket-codec-raw.txt). Lines are `mode variant jvmN phase: metrics`; the first phase of each JVM is warm-up and excluded. + +## Interpretation + +- **Encoder: meets the rule.** + - It sends 12× fewer messages (a third of base would be enough). Within a burst the client thread queues packets faster than the loop drains them, so a burst almost always becomes one message. + - Event-loop CPU per packet is 51–56% lower in every pair. The per-message header buffer, the byte-wise mask and the per-packet promise and flush work are gone. + - The lone-packet test passes, and so does a test where 20 packets queued from another thread while the loop is busy leave as one message. + - Fewer messages also mean fewer socket writes (80 instead of 1000 per 1000 packets) and fewer TLS records on wss. This benchmark does not measure those, because its sink is not a socket. +- **Decoder: misses the rule.** + - On its own it removes almost all of the codec's allocation, but CPU drops only 8–18% per pair, with overlapping ranges. + - With vanilla's splitter behind it, it is worse on both metrics. The splitter now receives 1460-byte slices: it starts a cumulation from a slice whose capacity cannot grow, copies on the next read, and grows that copy again for every packet spanning several reads. The copy the codec saved reappears downstream, plus more. + - Getting a real gain would mean handing the splitter whole messages without copying (for example a composite of slices). That is out of scope here. +- **Effect on the proxy (reported, not measured here).** The proxy session measured that one message per client tick instead of one per packet cuts proxy I/O CPU per frame by about 12–14% (`ProxySessionBenchmark --profile modded --websocket 1 --client-coalesce 1`). That record is `2026-10-23-websocket-cdn.md` in [layue13/MoonBridge#14](https://github.com/layue13/MoonBridge/pull/14). + +## Limits + +- **No real network.** One in-process microbenchmark on one desktop under background load. There is no socket, no real CDN, no real RTT and no TLS. The interplay of `TCP_NODELAY` with Nagle, delayed ACKs and the CDN's own buffering is not measured at all; that change rests on the argument that the encoder already batches. +- **Synthetic traffic.** The packet mix and burst pattern are made up, not taken from a real modpack session. Real coalescing depends on how vanilla and mods spread writes across a tick. +- **Smaller pieces not measured:** + - promise completion: a write's promise completes when its message is written, which is slightly later than before; + - vanilla's decrypt handler in online mode, which would sit between the codec and the splitter; + - direct buffers.