mirror of
https://github.com/layue13/MoonBridge.git
synced 2026-10-25 18:46:18 +00:00
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 <noreply@anthropic.com>
This commit is contained in:
layue13andClaude Opus 5.5 committed
1 parent
1decc44081
commit
b72c609aec
10 files changed
+185
-808
No files matched your search
@@ -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"`：编码器微基准（测试源码，不打包）。
|
||||
@@ -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.
|
||||
*
|
||||
* <p>
|
||||
* 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).
|
||||
*
|
||||
* <p>
|
||||
* 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.
|
||||
*
|
||||
* <p>
|
||||
* 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
*
|
||||
* <p>
|
||||
* Each message is one buffer: the header, then the payload masked four bytes at a time. {@link WebSocketFrame}s are
|
||||
|
||||
@@ -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.
|
||||
*
|
||||
* <p>
|
||||
* {@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 {
|
||||
|
||||
@@ -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)));
|
||||
|
||||
@@ -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<Integer> points = new ArrayList<Integer>();
|
||||
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<ByteBuf> inputs = new ArrayList<ByteBuf>();
|
||||
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());
|
||||
}
|
||||
}
|
||||
@@ -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").
|
||||
*
|
||||
* <p>
|
||||
* 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).
|
||||
*
|
||||
* <p>
|
||||
* 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<Object> 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<ByteBuf> reads = new ArrayList<ByteBuf>(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<long[]>() {
|
||||
|
||||
@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);
|
||||
}
|
||||
}
|
||||
@@ -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());
|
||||
|
||||
|
||||
@@ -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
|
||||
@@ -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.
|
||||
Reference in new issue
Block a user