mirror of
https://github.com/layue13/MoonBridge.git
synced 2026-10-25 18:46:18 +00:00
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 <noreply@anthropic.com>
This commit is contained in:
layue13andClaude Opus 5.5 committed
1 parent
6224d3deb4
commit
1decc44081
11 files changed
+1720
-53
No files matched your search
@@ -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(" ")
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
*
|
||||
* <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;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
*
|
||||
* <p>
|
||||
* 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.
|
||||
*
|
||||
* <p>
|
||||
* 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<Object> pending = new ArrayDeque<Object>();
|
||||
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());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
*
|
||||
* <p>
|
||||
* 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)));
|
||||
|
||||
@@ -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<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());
|
||||
}
|
||||
}
|
||||
@@ -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<Frame> drain(EmbeddedChannel ch) {
|
||||
ch.runPendingTasks();
|
||||
List<Frame> frames = new ArrayList<Frame>();
|
||||
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<Frame> 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<Frame> 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<Frame> 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<ChannelFuture> futures = new ArrayList<ChannelFuture>();
|
||||
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<Frame> 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<Frame> 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<Frame> frames = new ArrayList<Frame>();
|
||||
|
||||
@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<String> seenByNextTask = new AtomicReference<String>();
|
||||
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<ChannelFuture> writes = new ArrayList<ChannelFuture>();
|
||||
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<Frame> frames = ch.eventLoop()
|
||||
.submit(() -> new ArrayList<Frame>(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();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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").
|
||||
*
|
||||
* <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.
|
||||
*
|
||||
* <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).
|
||||
*/
|
||||
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<long[]>() {
|
||||
|
||||
@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<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);
|
||||
}
|
||||
}
|
||||
+2
-20
@@ -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();
|
||||
}
|
||||
}
|
||||
@@ -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<Integer> separate = Arrays.asList(306, 65536, 70000 - 65536);
|
||||
List<Integer> 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());
|
||||
|
||||
|
||||
@@ -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.
|
||||
Reference in new issue
Block a user