Skip to content

NetherNet transport improvements from production use #7

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
SendableMetatype wants to merge 7 commits into
base: master
Choose a base branch
from
Open

NetherNet transport improvements from production use #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
56f13ac
Stabilize NetherNet channels for long running servers
SendableMetatype Jul 2, 2026
a936e43
Make signaling WebSockets reconnectable in place
SendableMetatype Jul 2, 2026
2131d7a
Optimize the data channel path
SendableMetatype Jul 2, 2026
7af2f67
Offload connection setup, pool factories, real remote addresses
SendableMetatype Jul 2, 2026
79ed490
Scope pending RPC request failures to their own socket
SendableMetatype Jul 3, 2026
4ee49e6
Scope client handshake retries to their attempt
SendableMetatype Jul 3, 2026
3fbf5f1
Buffer remote candidates until the answer is applied
SendableMetatype Jul 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
249 changes: 205 additions & 44 deletions transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import java.nio.ByteBuffer;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;

public abstract class NetherNetChannel extends AbstractChannel {
private static final InternalLogger log = InternalLoggerFactory.getInstance(NetherNetChannel.class);
Expand All @@ -35,10 +36,38 @@ public abstract class NetherNetChannel extends AbstractChannel {
protected RTCDataChannel reliableChannel;
protected RTCDataChannel unreliableChannel;

// Mirrors the reliable channel's OPEN state, updated from the data channel
// observer. isActive() is called on every write and formerly crossed JNI
// into a blocking native call for it; the cached flag makes it free.
private volatile boolean reliableOpen;

// Reusable outbound fragment buffer. Confined to the event loop (doWrite),
// and safe to reuse across sends because the native layer copies the data
// out before sendAsync returns.
private ByteBuffer outboundScratch;

protected final Queue<Object> pendingWrites = new ConcurrentLinkedQueue<>();
private final AtomicBoolean channelActiveFired = new AtomicBoolean();

protected volatile boolean open = true;

// Reassembly state for inbound fragmented messages. assemblyBuf is written on
// the WebRTC callback thread (the data channel observer) and released on the
// event loop (doClose), so every access is guarded by assemblyLock together
// with the assemblyClosed flag, which orders the release strictly after the
// observer is detached and prevents a use-after-free.
private final Object assemblyLock = new Object();
private ByteBuf assemblyBuf;
private boolean assemblyClosed;
private int currentSegmentCount = -1;

// Maximum outbound SCTP message size in bytes. writeInternal must never
// fragment beyond what the remote advertised it can receive (the
// a=max-message-size attribute in its SDP); subclasses set this from the
// negotiated description. Defaults to the conservative MAX_SCTP_MESSAGE_SIZE
// until negotiation supplies the real value.
protected volatile int maxOutboundMessageSize = NetherNetConstants.MAX_SCTP_MESSAGE_SIZE;

protected NetherNetChannel(Channel parent, InetSocketAddress remote, InetSocketAddress local) {
super(parent);
this.remoteAddress = remote;
Expand All @@ -47,12 +76,25 @@ protected NetherNetChannel(Channel parent, InetSocketAddress remote, InetSocketA

public void setDataChannels(RTCDataChannel reliable, RTCDataChannel unreliable) {
this.reliableChannel = reliable;
// Seed the cached state once; the observer keeps it current from here.
this.reliableOpen = reliable.getState() == RTCDataChannelState.OPEN;
// The unreliable channel is intentionally stored but never observed or
// written to: this is a reliable-only implementation, which gives the
// ordered, lossless stream a Bedrock translation proxy needs. If it is
// ever wired up it must stay single-message only, because an unreliable
// channel can drop or reorder and would strand a multi-fragment message
// mid-reassembly: observe it but never run reassembly on it (every
// inbound message must arrive complete, header 0), and on send drop
// anything too large to fit one message rather than fragmenting it.
this.unreliableChannel = unreliable;

RTCDataChannelObserver observer = new RTCDataChannelObserver() {
private final ByteBuf assemblyBuf = config.getAllocator().buffer();
private int currentSegmentCount = -1;
synchronized (assemblyLock) {
if (assemblyBuf == null && !assemblyClosed) {
assemblyBuf = config.getAllocator().buffer();
}
}

RTCDataChannelObserver observer = new RTCDataChannelObserver() {
@Override
public void onBufferedAmountChange(long previousAmount) {
}
Expand All @@ -64,45 +106,53 @@ public void onStateChange() {

@Override
public void onMessage(RTCDataChannelBuffer buffer) {
// The ByteBuffer wraps native memory that is only valid for
// the duration of this callback, so every path below must copy
// exactly once into a netty buffer before handing off.
ByteBuffer data = buffer.data;
if (!data.hasRemaining())
return;

int segments = data.get() & 0xFF;

if (currentSegmentCount == -1) {
currentSegmentCount = segments;
} else {
if (segments != currentSegmentCount - 1) {
assemblyBuf.clear();
currentSegmentCount = -1;
synchronized (assemblyLock) {
if (assemblyClosed || assemblyBuf == null) {
return;
}
currentSegmentCount = segments;
}

if (data.hasRemaining()) {
byte[] payload = new byte[data.remaining()];
data.get(payload);
assemblyBuf.writeBytes(payload);
}

if (segments == 0) {
try {
if (assemblyBuf.isReadable()) {
ByteBuf packet = assemblyBuf.copy();
assemblyBuf.skipBytes(assemblyBuf.readableBytes());
// Fast path: complete single-segment message (the common
// case) skips the assembly buffer entirely.
if (currentSegmentCount == -1 && segments == 0) {
if (data.hasRemaining()) {
deliver(copyToBuf(data));
}
return;
}

eventLoop().execute(() -> {
pipeline().fireChannelRead(packet);
pipeline().fireChannelReadComplete();
});
if (currentSegmentCount == -1) {
currentSegmentCount = segments;
} else {
if (segments != currentSegmentCount - 1) {
assemblyBuf.clear();
currentSegmentCount = -1;
return;
}
} catch (Exception e) {
log.error("Error processing packet", e);
} finally {
assemblyBuf.clear();
currentSegmentCount = segments;
}

if (data.hasRemaining()) {
assemblyBuf.writeBytes(data);
}

if (segments == 0) {
currentSegmentCount = -1;
if (assemblyBuf.isReadable()) {
// Hand the assembled buffer off to the pipeline and
// start a fresh one instead of copying it again.
ByteBuf packet = assemblyBuf;
assemblyBuf = config.getAllocator().buffer();
deliver(packet);
}
}
}
}
Expand All @@ -115,17 +165,78 @@ public void onMessage(RTCDataChannelBuffer buffer) {
}
}

/**
* Copies the remaining bytes of the given callback-scoped buffer into a
* freshly allocated netty buffer.
*/
private ByteBuf copyToBuf(ByteBuffer data) {
ByteBuf packet = config.getAllocator().buffer(data.remaining());
packet.writeBytes(data);
return packet;
}

/**
* Fires a fully reassembled packet down the pipeline on the event loop.
* Releases the packet if the event loop rejects the task (shutdown race)
* so the buffer cannot leak.
*/
private void deliver(ByteBuf packet) {
try {
eventLoop().execute(() -> {
fireChannelActiveIfReady();
pipeline().fireChannelRead(packet);
pipeline().fireChannelReadComplete();
});
} catch (Exception e) {
packet.release();
}
}

/**
* Sets the maximum outbound SCTP message size, in bytes, for this channel.
* Should be the {@code a=max-message-size} the remote peer advertised in its
* SDP. Values too small to leave room for the fragment header are ignored.
*
* @param size the negotiated maximum message size
*/
public void setMaxOutboundMessageSize(int size) {
if (size > 1) {
this.maxOutboundMessageSize = size;
}
}

private void onDataChannelStateChange() {
RTCDataChannel channel = this.reliableChannel;
if (channel == null) {
return;
}
// One native call per state transition; every other isActive() check
// reads the cached flag.
RTCDataChannelState state = channel.getState();
reliableOpen = state == RTCDataChannelState.OPEN;

if (isActive()) {
if (!pendingWrites.isEmpty()) {
pipeline().fireChannelWritabilityChanged();
unsafe().flush();
}
} else if (reliableChannel.getState() == RTCDataChannelState.CLOSED) {
fireChannelActiveIfReady();
} else if (state == RTCDataChannelState.CLOSED) {
close();
}
}

protected void fireChannelActiveIfReady() {
if (!isRegistered() || !isActive()) {
return;
}

if (channelActiveFired.compareAndSet(false, true)) {
pipeline().fireChannelActive();
}

if (!pendingWrites.isEmpty()) {
pipeline().fireChannelWritabilityChanged();
unsafe().flush();
}
}

@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
if (!isActive()) {
Expand Down Expand Up @@ -160,29 +271,46 @@ private void writeInternal(Object msg) {

ByteBuf payload = (ByteBuf) msg;

ByteBuf framed = payload.retainedDuplicate();

int totalLength = framed.readableBytes();
int maxPayload = NetherNetConstants.MAX_SCTP_MESSAGE_SIZE - 1;
int maxPayload = maxOutboundMessageSize - 1;
int totalLength = payload.readableBytes();

int segments = (totalLength / maxPayload);
if (totalLength % maxPayload != 0)
segments++;

// Each fragment carries a one-byte countdown header, so a message can
// span at most 256 fragments. The negotiated max-message-size keeps this
// ceiling far out of reach, but guard it anyway: without the check an
// oversized message would wrap the header byte and silently corrupt the
// stream instead of failing.
if (segments > 256) {
pipeline().fireExceptionCaught(new IllegalStateException(
"Outbound message of " + totalLength + " bytes exceeds the maximum "
+ (256 * maxPayload) + " bytes addressable by the fragment header"));
return;
}

ByteBuf framed = payload.retainedDuplicate();
try {
// One reusable direct buffer for every fragment of every message:
// sendAsync copies the bytes out before returning, so the scratch
// can be refilled immediately, and unlike the blocking send it
// never stalls this event loop on the native network thread.
ByteBuffer chunk = outboundScratch(1 + maxPayload);

int offset = 0;
for (int i = 0; i < segments; i++) {
int remaining = segments - 1 - i;
int chunkSize = Math.min(maxPayload, framed.readableBytes() - offset);

ByteBuffer chunk = ByteBuffer.allocateDirect(1 + chunkSize);
chunk.clear();
chunk.limit(1 + chunkSize);
chunk.put((byte) remaining);

framed.getBytes(offset, chunk);
chunk.position(chunk.limit());
chunk.flip();

reliableChannel.send(new RTCDataChannelBuffer(chunk, true));
reliableChannel.sendAsync(new RTCDataChannelBuffer(chunk, true));
offset += chunkSize;
}
} catch (Exception e) {
Expand All @@ -192,8 +320,25 @@ private void writeInternal(Object msg) {
}
}

/**
* Returns the reusable outbound fragment buffer, growing it when the
* negotiated max message size demands more. Only used from doWrite on the
* event loop.
*/
private ByteBuffer outboundScratch(int capacity) {
ByteBuffer scratch = this.outboundScratch;
if (scratch == null || scratch.capacity() < capacity) {
scratch = ByteBuffer.allocateDirect(capacity);
this.outboundScratch = scratch;
}
return scratch;
}

@Override
protected void doRegister() throws Exception {
if (isActive()) {
channelActiveFired.set(true);
}
}

@Override
Expand All @@ -213,6 +358,8 @@ protected void doDisconnect() throws Exception {
@Override
protected void doClose() throws Exception {
this.open = false;
this.reliableOpen = false;
this.outboundScratch = null;

if (reliableChannel != null) {
reliableChannel.unregisterObserver();
Expand All @@ -226,6 +373,18 @@ protected void doClose() throws Exception {
peerConnection.close();
}

// Release the reassembly buffer now that the observer is unregistered.
// Guarded by assemblyLock with assemblyClosed so it cannot race, or be
// re-allocated by, an in-flight onMessage on the WebRTC callback thread.
synchronized (assemblyLock) {
assemblyClosed = true;
if (assemblyBuf != null) {
assemblyBuf.release();
assemblyBuf = null;
}
currentSegmentCount = -1;
}

Object msg;
while ((msg = pendingWrites.poll()) != null) {
ReferenceCountUtil.release(msg);
Expand Down Expand Up @@ -263,11 +422,13 @@ public boolean isOpen() {

@Override
public boolean isActive() {
return isOpen() && this.reliableChannel != null && this.reliableChannel.getState() == RTCDataChannelState.OPEN;
// Hot path: called around every write. Reads the observer-maintained
// flag instead of a blocking JNI call into the native network thread.
return isOpen() && this.reliableChannel != null && this.reliableOpen;
}

@Override
public ChannelMetadata metadata() {
return METADATA;
}
}
}
18 changes: 18 additions & 0 deletions ...t-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannelFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import io.netty.channel.Channel;
import io.netty.channel.ChannelFactory;

import java.util.List;
import java.util.function.Supplier;

public class NetherNetChannelFactory<T extends Channel> implements ChannelFactory<T> {
Expand All @@ -32,6 +33,23 @@ public static ChannelFactory<NetherNetServerChannel> server(PeerConnectionFactor
return new NetherNetChannelFactory<>(() -> new NetherNetServerChannel(factory, signaling));
}

/**
* Creates a NetherNet Server Channel Factory backed by a pool of
* PeerConnectionFactory instances. Each native factory carries exactly one
* network, worker, and signaling thread shared by every peer connection it
* creates, so a pool multiplies the threads available to the data plane:
* incoming connections are assigned round robin across the pool.
*
* @param factories The PeerConnectionFactory pool. Must not be empty. The
* resulting server channel takes ownership and disposes
* every factory on close.
* @param signaling The NetherNetServerSignaling instance for signaling.
* @return A ChannelFactory for NetherNetServerChannel.
*/
public static ChannelFactory<NetherNetServerChannel> server(List<PeerConnectionFactory> factories, NetherNetServerSignaling signaling) {
return new NetherNetChannelFactory<>(() -> new NetherNetServerChannel(factories, signaling));
}

/**
* Creates a NetherNet Client Channel Factory.
*
Expand Down
Loading