From 3e6f753c981897960eac23dd104273945b6e47f8 Mon Sep 17 00:00:00 2001 From: Shane Freeder Date: Fri, 13 Nov 2026 15:24:13 +0000 Subject: [PATCH 1/2] Expose the bytes in the SCTP association's own queues and the send buffer that bounds them Syncs jni/libdatachannel to the commit that reports them. --- jni/libdatachannel | 2 +- jni/src/native_peer.cpp | 3 ++- native-test/NativeTransportProbe.java | 18 ++++++++++++-- .../tel/schich/libdatachannel/SctpStats.java | 24 +++++++++++++++---- 4 files changed, 38 insertions(+), 9 deletions(-) diff --git a/jni/libdatachannel b/jni/libdatachannel index 2e2d86a..85e6538 160000 --- a/jni/libdatachannel +++ b/jni/libdatachannel @@ -1 +1 @@ -Subproject commit 2e2d86acc27a4734db2f28f95d09c1ab7a83836f +Subproject commit 85e65386265f9883caedf3bc951b4f2dc5106330 diff --git a/jni/src/native_peer.cpp b/jni/src/native_peer.cpp index e7edf76..aff5e96 100644 --- a/jni/src/native_peer.cpp +++ b/jni/src/native_peer.cpp @@ -369,7 +369,8 @@ JNIEXPORT jobject JNICALL Java_tel_schich_libdatachannel_LibDataChannelNative_rt return call_tel_schich_libdatachannel_SctpStats_create(env, stats.rtt, stats.rto, static_cast(stats.congestionWindow), static_cast(stats.peerReceiveWindow), static_cast(stats.unackedChunks), static_cast(stats.pendingChunks), - static_cast(stats.dataTimeouts)); + static_cast(stats.dataTimeouts), static_cast(stats.sendQueueBytes), + static_cast(stats.sendBufferSize), static_cast(stats.receiveQueueBytes)); } JNIEXPORT jint JNICALL Java_tel_schich_libdatachannel_LibDataChannelNative_rtcSetLocalDescriptionWithIce( diff --git a/native-test/NativeTransportProbe.java b/native-test/NativeTransportProbe.java index c454b3b..1f794af 100644 --- a/native-test/NativeTransportProbe.java +++ b/native-test/NativeTransportProbe.java @@ -43,11 +43,13 @@ static String answer(String ufrag, String fingerprint) { "\r\na=sctp-port:5000\r\na=max-message-size:262144\r\na=candidate:1 1 UDP 2130706431 127.0.0.1 " + PORT + " typ host\r\na=end-of-candidates\r\n"; } + /** Above the 256 KiB max message size, which would otherwise raise it, and not the 1 MiB default. */ + static final int SEND_BUFFER_SIZE = 512 << 10; public static void main(String[] args) throws Exception { // Set before the first peer, the associations below have to run on it LibDataChannel.setSctpSettings(SctpSettings.builder().minRetransmitTimeoutMs(400) .initialRetransmitTimeoutMs(500).maxRetransmitTimeoutMs(3000).delayedSackTimeMs(200) - .maxRetransmitAttempts(5).build()); + .maxRetransmitAttempts(5).sendBufferSize(SEND_BUFFER_SIZE).build()); Path certificate = Path.of(args[0]), key = Path.of(args[1]); byte[] der; try (var input = Files.newInputStream(certificate)) { @@ -455,6 +457,7 @@ static void run(Path certificate, Path key, String fingerprint, int ufragLength, ArrayBlockingQueue arrivals = new ArrayBlockingQueue<>(4); AtomicInteger notifications = new AtomicInteger(), channelMask = new AtomicInteger(), callbackGuards = new AtomicInteger(); AtomicReference failure = new AtomicReference<>(); + AtomicReference hostOrdered = new AtomicReference<>(); CountDownLatch opened = new CountDownLatch(2), messages = new CountDownLatch(402), failed = new CountDownLatch(1); PeerConnection host = null; try (IceUdpMuxListener mux = new IceUdpMuxListener(LOOPBACK, PORT, Runnable::run, request -> { @@ -499,6 +502,7 @@ static void run(Path certificate, Path key, String fingerprint, int ufragLength, peer.onDataChannel.register((p, dc) -> { int bit = dc.label().equals("ordered") ? 1 : dc.label().equals("unordered") ? 2 : 0; channelMask.getAndUpdate(mask -> mask | bit); + if (bit == 1) hostOrdered.set(dc); dc.onMessage.register(DataChannelCallback.Message.handleBinary((d, data) -> { try { check(bit != 0 && data.remaining() == 2 && data.get() == 0 && data.get() == bit, "distinct channel payload"); messages.countDown(); } catch (Throwable error) { failure.set(error); } @@ -523,6 +527,16 @@ static void run(Path certificate, Path key, String fingerprint, int ufragLength, check(sctp.congestionWindow() > 0 && sctp.peerReceiveWindow() > 0 && !sctp.rto().isZero() && sctp.dataTimeouts() == 0, "SCTP stats describe a healthy loopback association"); check(sctp.rto().toMillis() >= 400, "SCTP settings raise the minimum RTO, got " + sctp.rto()); + check(sctp.sendBufferSize() == SEND_BUFFER_SIZE, "SCTP settings size the send buffer, got " + sctp.sendBufferSize()); + PeerConnection sender = host; + long mostQueued = 0; + // Delayed SACKs keep at least the first message queued + for (int i = 0; i < 32; i++) { + hostOrdered.get().sendMessage(ByteBuffer.allocateDirect(16 << 10)); + mostQueued = Math.max(mostQueued, sender.sctpStats().map(SctpStats::sendQueueBytes).orElse(0L)); + } + check(mostQueued >= 16 << 10, "SCTP send queue holds an unacknowledged message, got " + mostQueued); + await(() -> sender.sctpStats().map(SctpStats::sendQueueBytes).orElse(-1L) == 0, "SCTP send queue drains once acknowledged"); long[] stats = mux.stats(); check(notifications.get() == 1 && stats[5] == 1 && stats[2] == 1 && stats[3] == 1 && stats[0] > 10, "transport traffic stays native after one admission callback"); @@ -535,7 +549,7 @@ static void run(Path certificate, Path key, String fingerprint, int ufragLength, check(remote == CloseReason.RTC_CLOSE_REASON_REMOTE_SHUTDOWN || remote == CloseReason.RTC_CLOSE_REASON_REMOTE_ABORT, "a client close reads as remote on the host, got " + remote); System.out.println("native-transport PASS ufragChars=" + ufragLength + " admissionCallbacks=1 duplicates=" + stats[6] + - " datagrams=" + stats[0] + " channels=2 messages=402 " + sctp + " hostCloseReason=" + remote); + " datagrams=" + stats[0] + " channels=2 messages=402 " + sctp + " mostQueued=" + mostQueued + " hostCloseReason=" + remote); } } finally { if (host != null) check(host.closeAndAwait(Duration.ofSeconds(5)), "accepted peer native cleanup"); } } diff --git a/src/main/java/tel/schich/libdatachannel/SctpStats.java b/src/main/java/tel/schich/libdatachannel/SctpStats.java index 858189d..920609e 100644 --- a/src/main/java/tel/schich/libdatachannel/SctpStats.java +++ b/src/main/java/tel/schich/libdatachannel/SctpStats.java @@ -7,10 +7,12 @@ /** What a peer's SCTP association reported about itself when it was read. */ public final class SctpStats { private final Duration rtt, rto; - private final long congestionWindow, peerReceiveWindow, unackedChunks, pendingChunks, dataTimeouts; + private final long congestionWindow, peerReceiveWindow, unackedChunks, pendingChunks, dataTimeouts, + sendQueueBytes, sendBufferSize, receiveQueueBytes; private SctpStats(Duration rtt, Duration rto, long congestionWindow, long peerReceiveWindow, - long unackedChunks, long pendingChunks, long dataTimeouts) { + long unackedChunks, long pendingChunks, long dataTimeouts, long sendQueueBytes, + long sendBufferSize, long receiveQueueBytes) { this.rtt = rtt; this.rto = rto; this.congestionWindow = congestionWindow; @@ -18,13 +20,18 @@ private SctpStats(Duration rtt, Duration rto, long congestionWindow, long peerRe this.unackedChunks = unackedChunks; this.pendingChunks = pendingChunks; this.dataTimeouts = dataTimeouts; + this.sendQueueBytes = sendQueueBytes; + this.sendBufferSize = sendBufferSize; + this.receiveQueueBytes = receiveQueueBytes; } @JNIAccess static SctpStats create(int rttMillis, int rtoMillis, long congestionWindow, long peerReceiveWindow, - long unackedChunks, long pendingChunks, long dataTimeouts) { + long unackedChunks, long pendingChunks, long dataTimeouts, long sendQueueBytes, + long sendBufferSize, long receiveQueueBytes) { return new SctpStats(Duration.ofMillis(rttMillis), Duration.ofMillis(rtoMillis), congestionWindow, - peerReceiveWindow, unackedChunks, pendingChunks, dataTimeouts); + peerReceiveWindow, unackedChunks, pendingChunks, dataTimeouts, sendQueueBytes, sendBufferSize, + receiveQueueBytes); } /** Smoothed round trip time in whole milliseconds, zero until sampled or while under one. */ @@ -41,11 +48,18 @@ static SctpStats create(int rttMillis, int rtoMillis, long congestionWindow, lon public long pendingChunks() { return pendingChunks; } /** Retransmission timer expiries since the association started, each a stall while data is resent. */ public long dataTimeouts() { return dataTimeouts; } + /** Bytes accepted for sending and not acknowledged yet, sent or not, which the buffered amount does not count. */ + public long sendQueueBytes() { return sendQueueBytes; } + /** Bytes the send queue may hold before messages wait in the buffered amount instead. */ + public long sendBufferSize() { return sendBufferSize; } + /** Received bytes waiting for reassembly or ordering. */ + public long receiveQueueBytes() { return receiveQueueBytes; } @Override public String toString() { return "SctpStats{rtt=" + rtt.toMillis() + "ms, rto=" + rto.toMillis() + "ms, congestionWindow=" + congestionWindow + ", peerReceiveWindow=" + peerReceiveWindow + ", unackedChunks=" + unackedChunks + - ", pendingChunks=" + pendingChunks + ", dataTimeouts=" + dataTimeouts + "}"; + ", pendingChunks=" + pendingChunks + ", dataTimeouts=" + dataTimeouts + ", sendQueueBytes=" + + sendQueueBytes + ", sendBufferSize=" + sendBufferSize + ", receiveQueueBytes=" + receiveQueueBytes + "}"; } } From 130c1ebcd5a287ffcdd3cfa7abb10b9bf1da6447 Mon Sep 17 00:00:00 2001 From: Shane Freeder Date: Fri, 13 Nov 2026 15:24:13 +0000 Subject: [PATCH 2/2] Fetch jni/libdatachannel from electronicboy/libdatachannel until it lands upstream Drop before merging. --- .gitmodules | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.gitmodules b/.gitmodules index fe4a864..de4d01f 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,6 +1,6 @@ [submodule "jni/libdatachannel"] path = jni/libdatachannel - url = https://github.com/opencollab-incubator/libdatachannel.git + url = https://github.com/electronicboy/libdatachannel.git [submodule "jni/cmake-conan"] path = jni/cmake-conan url = https://github.com/conan-io/cmake-conan.git