From a21814521056fd80e56297002255bae1ae5e0c18 Mon Sep 17 00:00:00 2001 From: pedroSG94 Date: Tue, 22 Sep 2026 19:18:54 +0200 Subject: [PATCH 1/4] remove redundant code --- .../java/com/pedro/encoder/audio/AudioEncoder.java | 6 +----- .../pedro/library/util/AacMuxerRecordController.kt | 14 ++++++-------- 2 files changed, 7 insertions(+), 13 deletions(-) diff --git a/encoder/src/main/java/com/pedro/encoder/audio/AudioEncoder.java b/encoder/src/main/java/com/pedro/encoder/audio/AudioEncoder.java index 78d2fb3f56..c99653e7e7 100644 --- a/encoder/src/main/java/com/pedro/encoder/audio/AudioEncoder.java +++ b/encoder/src/main/java/com/pedro/encoder/audio/AudioEncoder.java @@ -167,11 +167,7 @@ protected long calculatePts(Frame frame, long presentTimeUs) { @Override protected boolean checkBuffer(@NonNull ByteBuffer byteBuffer, @NonNull MediaCodec.BufferInfo bufferInfo) { - // Codec config (e.g. AAC AudioSpecificConfig) is signalled per protocol and via - // MediaFormat csd-0 for the muxer; do not forward it as an encoded audio frame. - if ((bufferInfo.flags & MediaCodec.BUFFER_FLAG_CODEC_CONFIG) != 0) { - return false; - } + if ((bufferInfo.flags & MediaCodec.BUFFER_FLAG_CODEC_CONFIG) != 0) return false; return checkValidTimeStamp(bufferInfo); } diff --git a/library/src/main/java/com/pedro/library/util/AacMuxerRecordController.kt b/library/src/main/java/com/pedro/library/util/AacMuxerRecordController.kt index a34fd1331e..f3bc923d4d 100644 --- a/library/src/main/java/com/pedro/library/util/AacMuxerRecordController.kt +++ b/library/src/main/java/com/pedro/library/util/AacMuxerRecordController.kt @@ -119,14 +119,12 @@ class AacMuxerRecordController : AsyncBaseRecordController() { private suspend fun write(byteBuffer: ByteBuffer, info: MediaCodec.BufferInfo) { try { - if ((info.flags and MediaCodec.BUFFER_FLAG_CODEC_CONFIG) != MediaCodec.BUFFER_FLAG_CODEC_CONFIG) { - val header = createAdtsHeader(2, info.size - info.offset, sampleRate, channels).array() - outputStream?.write(header) - val data = ByteArray(byteBuffer.remaining()) - byteBuffer.get(data) - outputStream?.write(data) - bitrateManager?.calculateBitrate(info.size * 8L) - } + val header = createAdtsHeader(2, info.size - info.offset, sampleRate, channels).array() + outputStream?.write(header) + val data = ByteArray(byteBuffer.remaining()) + byteBuffer.get(data) + outputStream?.write(data) + bitrateManager?.calculateBitrate(info.size * 8L) } catch (e: Exception) { listener?.onError(e) } From d26c52d27555882e3a165fd26994d24ed9c7bd02 Mon Sep 17 00:00:00 2001 From: pedroSG94 Date: Tue, 22 Sep 2026 20:10:56 +0200 Subject: [PATCH 2/4] refactoring srt commands --- .../java/com/pedro/srt/srt/CommandsManager.kt | 88 +++++---- .../com/pedro/srt/srt/packets/DataPacket.kt | 1 - .../com/pedro/srt/srt/CommandsManagerTest.kt | 179 +++++++++++++++--- 3 files changed, 203 insertions(+), 65 deletions(-) diff --git a/srt/src/main/java/com/pedro/srt/srt/CommandsManager.kt b/srt/src/main/java/com/pedro/srt/srt/CommandsManager.kt index 5afa109099..09e46efd1b 100644 --- a/srt/src/main/java/com/pedro/srt/srt/CommandsManager.kt +++ b/srt/src/main/java/com/pedro/srt/srt/CommandsManager.kt @@ -43,7 +43,7 @@ import kotlin.random.Random /** * Created by pedro on 23/8/23. */ -class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurrentTimeMicro() }) { +class CommandsManager { private val TAG = "CommandsManager" //used for packet lost @@ -71,7 +71,9 @@ class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurr var videoCodec = VideoCodec.H264 var audioCodec = AudioCodec.AAC + @Volatile private var rtt = 0 + @Volatile private var rttVariance = 0 private var mediaBytesPerSecond = 0.0 private var mediaWindowStartUs = 0L @@ -79,6 +81,9 @@ class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurr private var retransmitTokens = 0.0 private var lastTokenRefillUs = 0L private var bucketInitialized = false + //used for unique packet lost + private var lastNakSequence = 0 + private var lastNakInitialized = false fun setPassphrase(passphrase: String, type: EncryptionType) { encryptor = if (passphrase.isEmpty() || type == EncryptionType.NONE) null else EncryptionUtil(type, passphrase) @@ -95,19 +100,18 @@ class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurr fun encryptionEnabled() = encryptor != null fun loadStartTs() { - startTS = timeProvider() + startTS = TimeUtils.getCurrentTimeMicro() localSocketId = generateSocketId() } fun getTs(): Int { - return (timeProvider() - startTS).toInt() + return (TimeUtils.getCurrentTimeMicro() - startTS).toInt() } - suspend fun updateRtt(rtt: Int, rttVariance: Int) { - writeSync.withLock { - this.rtt = rtt - this.rttVariance = rttVariance - } + fun updateRtt(rtt: Int, rttVariance: Int) { + val maxUs = latency * 1000 + this.rtt = rtt.coerceIn(0, maxUs) + this.rttVariance = rttVariance.coerceIn(0, maxUs) } @Throws(IOException::class) @@ -150,8 +154,8 @@ class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurr sequenceNumber++ packetHandlingQueue.add(dataPacket) dropTooLatePackets(dataPacket.ts) - trackMediaBytes(packet.buffer.size, timeProvider()) dataPacket.write() + trackMediaBytes(dataPacket.getSize(), TimeUtils.getCurrentTimeMicro()) socket?.write(dataPacket) return dataPacket.getSize() } @@ -160,31 +164,22 @@ class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurr @Throws(IOException::class) suspend fun reSendPackets(lostRanges: List>, socket: SrtSocket?): Int { writeSync.withLock { - val unlimited = retransmitOverheadPercent <= 0 + val budgetEnabled = retransmitOverheadPercent > 0 val nowTs = getTs() - val nowUs = timeProvider() - if (!unlimited) { - refillRetransmitTokens(nowUs) - } + val nowUs = TimeUtils.getCurrentTimeMicro() + if (budgetEnabled) refillRetransmitTokens(nowUs) + val latencyUs = latency * 1000 - val minResendInterval = if (unlimited) 0 else { - min(max(rtt + 4 * rttVariance, MIN_RESEND_INTERVAL_US), latencyUs / 4) - } - var newlyReported = 0 + val minResendInterval = min(max(rtt + 4 * rttVariance, MIN_RESEND_INTERVAL_US), latencyUs / 4) + val newlyReported = countNewLostPackets(lostRanges) var budgetExhausted = false for (packet in packetHandlingQueue) { if (!isInLostRange(packet.sequenceNumber, lostRanges)) continue - if (!packet.nakReported) { - packet.nakReported = true - newlyReported++ - } - if (!unlimited) { - if ((nowTs - packet.ts + rtt / 2) >= latencyUs) continue - // The gate only suppresses repeated reports of a packet that was already retransmitted; - // the first NAK is always honored even when it arrives within minResendInterval of the original send. - if (packet.retransmitted && (nowTs - packet.lastSentTs) < minResendInterval) continue + if ((nowTs - packet.ts + rtt / 2) >= latencyUs) continue + if (packet.retransmitted && (nowTs - packet.lastSentTs) < minResendInterval) continue + if (budgetEnabled) { if (budgetExhausted) continue - val packetSize = dataPacketWireSize(packet) + val packetSize = packet.getSize() if (retransmitTokens < packetSize) { budgetExhausted = true continue @@ -210,6 +205,27 @@ class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurr } } + //a nak reports the same missing packets again until received, only sequences after the last + //reported one are a new lost. Counted from the ranges so packets already dropped are counted too + private fun countNewLostPackets(lostRanges: List>): Int { + var count = 0 + lostRanges.forEach { (from, to) -> + val size = ((to - from) and 0x7FFFFFFF) + 1 + if (!lastNakInitialized) { + lastNakInitialized = true + lastNakSequence = to + count += size + } else { + val ahead = (to - lastNakSequence) and 0x7FFFFFFF + if (ahead in 1 until 0x40000000) { + count += min(ahead, size) + lastNakSequence = to + } + } + } + return count + } + private fun isInLostRange(sequenceNumber: Int, lostRanges: List>): Boolean { return lostRanges.any { (min, max) -> ((sequenceNumber - min) and 0x7FFFFFFF) <= ((max - min) and 0x7FFFFFFF) @@ -220,9 +236,11 @@ class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurr if (mediaWindowStartUs == 0L) mediaWindowStartUs = nowUs mediaWindowBytes += bytes val elapsed = nowUs - mediaWindowStartUs - if (elapsed >= MEDIA_RATE_WINDOW_US) { + val window = if (mediaBytesPerSecond > 0.0) MEDIA_RATE_WINDOW_US else MEDIA_RATE_WARMUP_WINDOW_US + if (elapsed >= window) { val rate = mediaWindowBytes.toDouble() * MEDIA_RATE_WINDOW_US / elapsed - mediaBytesPerSecond = if (mediaBytesPerSecond == 0.0) rate else { + //follow an increase at once to not limit below the real media rate, average a decrease + mediaBytesPerSecond = if (rate > mediaBytesPerSecond) rate else { mediaBytesPerSecond * MEDIA_RATE_EWMA_OLD + rate * MEDIA_RATE_EWMA_NEW } mediaWindowStartUs = nowUs @@ -237,8 +255,6 @@ class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurr } private fun getRetransmitCapacity(rate: Double): Int { - // Allow an immediate burst up to half a second of media so short loss events - // (e.g. a brief link flap) are not retried one packet at a time on healthy links. val burstCapacity = if (mediaBytesPerSecond > 0.0) { (mediaBytesPerSecond * RETRANSMIT_BURST_WINDOW_US / MEDIA_RATE_WINDOW_US).toInt() } else { @@ -263,10 +279,6 @@ class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurr lastTokenRefillUs = nowUs } - private fun dataPacketWireSize(packet: DataPacket): Int { - return packet.payload.size + DATA_HEADER_SIZE - } - private fun dropTooLatePackets(nowTs: Int) { val thresholdUs = latency * 1000 val firstKept = packetHandlingQueue.indexOfFirst { (nowTs - it.ts) <= thresholdUs } @@ -316,6 +328,8 @@ class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurr retransmitTokens = 0.0 lastTokenRefillUs = 0L bucketInitialized = false + lastNakSequence = 0 + lastNakInitialized = false } private fun generateInitialSequence(): Int { @@ -327,10 +341,10 @@ class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurr } companion object { - private const val DATA_HEADER_SIZE = 16 private const val MIN_RETRANSMIT_BYTES_PER_SECOND = 8_000 private const val MIN_RESEND_INTERVAL_US = 20_000 private const val MEDIA_RATE_WINDOW_US = 1_000_000L + private const val MEDIA_RATE_WARMUP_WINDOW_US = 200_000L private const val RETRANSMIT_BURST_WINDOW_US = 500_000L private const val MEDIA_RATE_EWMA_OLD = 0.8 private const val MEDIA_RATE_EWMA_NEW = 0.2 diff --git a/srt/src/main/java/com/pedro/srt/srt/packets/DataPacket.kt b/srt/src/main/java/com/pedro/srt/srt/packets/DataPacket.kt index cdb992025b..b7f6db6c8d 100644 --- a/srt/src/main/java/com/pedro/srt/srt/packets/DataPacket.kt +++ b/srt/src/main/java/com/pedro/srt/srt/packets/DataPacket.kt @@ -42,7 +42,6 @@ class DataPacket( ): SrtPacket() { var lastSentTs: Int = 0 - var nakReported: Boolean = false fun write() { resetBuffer() diff --git a/srt/src/test/java/com/pedro/srt/srt/CommandsManagerTest.kt b/srt/src/test/java/com/pedro/srt/srt/CommandsManagerTest.kt index f4f02c503e..f1a5c72298 100644 --- a/srt/src/test/java/com/pedro/srt/srt/CommandsManagerTest.kt +++ b/srt/src/test/java/com/pedro/srt/srt/CommandsManagerTest.kt @@ -16,17 +16,21 @@ package com.pedro.srt.srt +import com.pedro.common.TimeUtils import com.pedro.srt.mpeg2ts.MpegTsPacket import com.pedro.srt.mpeg2ts.MpegType import com.pedro.srt.srt.packets.SrtPacket import com.pedro.srt.srt.packets.data.PacketPosition import com.pedro.srt.utils.SrtSocket import kotlinx.coroutines.test.runTest +import org.junit.After import org.junit.Assert.assertEquals import org.junit.Before import org.junit.Test import org.junit.runner.RunWith import org.mockito.Mock +import org.mockito.MockedStatic +import org.mockito.Mockito import org.mockito.Mockito.clearInvocations import org.mockito.Mockito.never import org.mockito.Mockito.times @@ -40,15 +44,19 @@ class CommandsManagerTest { @Mock lateinit var socket: SrtSocket + private lateinit var timeUtilsMocked: MockedStatic private var nowUs = 1_000_000L @Before fun setup() { nowUs = 1_000_000L + timeUtilsMocked = Mockito.mockStatic(TimeUtils::class.java) + timeUtilsMocked.`when`(TimeUtils::getCurrentTimeMicro).then { nowUs } } - private fun createManager(): CommandsManager { - return CommandsManager { nowUs } + @After + fun teardown() { + timeUtilsMocked.close() } private suspend fun sendPacket( @@ -65,7 +73,8 @@ class CommandsManagerTest { } private suspend fun establishMediaRate(manager: CommandsManager, bytesPerSecond: Int) { - // 11 equal chunks over a 1 s trackMediaBytes window -> measured rate ~= bytesPerSecond + // 11 equal chunks over a 1 s trackMediaBytes window. The rate is measured on the wire size, + // so it ends up as bytesPerSecond plus the 16 B header of each of the 11 packets val chunkSize = bytesPerSecond / 11 repeat(10) { sendPacket(manager, chunkSize) @@ -76,7 +85,7 @@ class CommandsManagerTest { @Test fun `GIVEN repeated NAKs for same packet WHEN within time gate THEN resend once and again after clock advance`() = runTest { - val manager = createManager() + val manager = CommandsManager() manager.loadStartTs() // latency 2000 ms: at +200 ms the packet is still inside the latency window // (200_000 + rtt/2 = 250_000 < 2_000_000); default 120 ms would mark it too late @@ -102,16 +111,17 @@ class CommandsManagerTest { @Test fun `GIVEN NAK range over retransmit budget WHEN tokens refill THEN resend oldest packets first`() = runTest { - val manager = createManager() + val manager = CommandsManager() manager.loadStartTs() manager.retransmitOverheadPercent = 1 manager.latency = 120 establishMediaRate(manager, 16_000) manager.updateRtt(50_000, 10_000) - // rate = max(16_000 * 1%, 8_000) = 8_000 B/s - // capacity = max(16_000 * 0.5, 8_000 * 120/1000, MTU=1500) = 8_000 B - // wire = 400 + 16 = 416 B; 19 * 416 = 7_904 fits, 20th needs 416 but only 96 B left + // media rate = 11 * (1454 + 16) = 16_170 B/s + // rate = max(16_170 * 1%, 8_000) = 8_000 B/s + // capacity = max(16_170 * 0.5, max(8_000 * 120/1000, MTU=1500)) = 8_085 B + // wire = 400 + 16 = 416 B; 19 * 416 = 7_904 fits, 20th needs 416 but only 181 B left val startSeq = manager.sequenceNumber repeat(20) { sendPacket(manager, 400) @@ -123,7 +133,7 @@ class CommandsManagerTest { verify(socket, times(19)).write(any()) nowUs += 50_000 - // refill: 96 + 8_000 * 50_000/1_000_000 = 496 B, enough for the remaining 416 B packet + // refill: 181 + 8_000 * 50_000/1_000_000 = 581 B, enough for the remaining 416 B packet // still in time: 50_000 + rtt/2 = 75_000 < 120_000 us; never retransmitted, so no time gate manager.reSendPackets(listOf(endSeq to endSeq), socket) verify(socket, times(20)).write(any()) @@ -131,7 +141,7 @@ class CommandsManagerTest { @Test fun `GIVEN packet near latency expiry WHEN NAK received THEN skip resend`() = runTest { - val manager = createManager() + val manager = CommandsManager() manager.loadStartTs() manager.latency = 1000 manager.updateRtt(100_000, 0) @@ -148,7 +158,7 @@ class CommandsManagerTest { @Test fun `GIVEN small loss on healthy link WHEN NAK received THEN resend all lost packets immediately`() = runTest { - val manager = createManager() + val manager = CommandsManager() manager.loadStartTs() manager.updateRtt(10_000, 2_000) @@ -166,23 +176,136 @@ class CommandsManagerTest { } @Test - fun `GIVEN retransmit overhead disabled WHEN repeated NAKs received THEN resend without limits`() = runTest { - val manager = createManager() + fun `GIVEN retransmit budget disabled WHEN NAK range over budget THEN resend every packet`() = runTest { + val manager = CommandsManager() manager.loadStartTs() manager.retransmitOverheadPercent = 0 + manager.latency = 120 + establishMediaRate(manager, 16_000) + manager.updateRtt(50_000, 10_000) + + // same scenario as the budget test, where the token bucket stops before the last packets + val startSeq = manager.sequenceNumber + repeat(20) { + sendPacket(manager, 400) + } + val endSeq = manager.sequenceNumber - 1 + clearInvocations(socket) + + assertEquals(20, manager.reSendPackets(listOf(startSeq to endSeq), socket)) + verify(socket, times(20)).write(any()) + } + + @Test + fun `GIVEN retransmit budget disabled WHEN repeated NAKs for same packet THEN time gate still applies`() = runTest { + val manager = CommandsManager() + manager.loadStartTs() + manager.retransmitOverheadPercent = 0 + manager.updateRtt(12_000, 0) + // minResendInterval = min(max(12_000, 20_000), 30_000) = 20_000 us val seq = manager.sequenceNumber sendPacket(manager) clearInvocations(socket) manager.reSendPackets(listOf(seq to seq), socket) + verify(socket, times(1)).write(any()) + + manager.reSendPackets(listOf(seq to seq), socket) + verify(socket, times(1)).write(any()) + + nowUs += 20_000 manager.reSendPackets(listOf(seq to seq), socket) verify(socket, times(2)).write(any()) } + @Test + fun `GIVEN loss right after stream start WHEN NAK received THEN media rate is already estimated`() = runTest { + val manager = CommandsManager() + manager.loadStartTs() + manager.latency = 120 + + // 200 ms of media is enough for a first estimation: 11 * (1316 + 16) B in 200 ms -> 73_260 B/s + repeat(10) { + sendPacket(manager, 1316) + } + nowUs += 200_000 + sendPacket(manager, 1316) + + // rate = max(73_260 * 25%, 8_000) = 18_315 B/s + // capacity = max(73_260 * 0.5, max(18_315 * 120/1000, MTU=1500)) = 36_630 B + // 10 * 1332 = 13_320 B fit. Waiting a full second the budget would be 1_500 B, a single packet + val startSeq = manager.sequenceNumber + repeat(10) { + sendPacket(manager, 1316) + } + val endSeq = manager.sequenceNumber - 1 + clearInvocations(socket) + + assertEquals(10, manager.reSendPackets(listOf(startSeq to endSeq), socket)) + verify(socket, times(10)).write(any()) + } + + @Test + fun `GIVEN NAK for a packet already dropped from the queue WHEN counting unique lost THEN count it`() = runTest { + val manager = CommandsManager() + manager.loadStartTs() + manager.latency = 120 + + val seq = manager.sequenceNumber + sendPacket(manager) + // at +200 ms the first packet is older than the latency window and dropTooLatePackets removes it + nowUs += 200_000 + sendPacket(manager) + clearInvocations(socket) + + // no longer in the queue so it can't be resent, but it is still a lost packet + assertEquals(1, manager.reSendPackets(listOf(seq to seq), socket)) + verify(socket, never()).write(any()) + } + + @Test + fun `GIVEN media rate increase WHEN next window is measured THEN budget follows it at once`() = runTest { + val manager = CommandsManager() + manager.loadStartTs() + manager.latency = 120 + establishMediaRate(manager, 16_000) + establishMediaRate(manager, 160_000) + + // media rate jumps from 16_170 to 160_171 B/s + // capacity = max(160_171 * 0.5, max(40_042 * 120/1000, MTU=1500)) = 80_085 B + // 40 * (1316 + 16) = 53_280 B fit. Averaged with the old rate it would be 22_485 B, only 16 packets + val startSeq = manager.sequenceNumber + repeat(40) { + sendPacket(manager, 1316) + } + val endSeq = manager.sequenceNumber - 1 + clearInvocations(socket) + + assertEquals(40, manager.reSendPackets(listOf(startSeq to endSeq), socket)) + verify(socket, times(40)).write(any()) + } + + @Test + fun `GIVEN ack reporting an out of range rtt WHEN NAK received THEN clamp it and keep resending`() = runTest { + val manager = CommandsManager() + manager.loadStartTs() + manager.latency = 120 + manager.updateRtt(Int.MAX_VALUE, Int.MAX_VALUE) + + // clamped to 120_000 us, so rtt / 2 leaves the packet inside the latency window. + // Without the clamp rtt / 2 alone is bigger than the window and nothing is ever resent + val seq = manager.sequenceNumber + sendPacket(manager) + clearInvocations(socket) + + assertEquals(1, manager.reSendPackets(listOf(seq to seq), socket)) + verify(socket, times(1)).write(any()) + } + @Test fun `GIVEN sequence wrap at max value WHEN NAK spans wrap THEN resend wrapped packets`() = runTest { - val manager = createManager() + val manager = CommandsManager() manager.loadStartTs() manager.retransmitOverheadPercent = 0 manager.sequenceNumber = 0x7FFFFFFE @@ -198,7 +321,7 @@ class CommandsManagerTest { @Test fun `GIVEN repeated NAK for same packet WHEN already reported THEN return zero newly reported`() = runTest { - val manager = createManager() + val manager = CommandsManager() manager.loadStartTs() manager.retransmitOverheadPercent = 0 @@ -211,14 +334,14 @@ class CommandsManagerTest { @Test fun `GIVEN active retransmit state WHEN reset called THEN allow immediate resend again`() = runTest { - val manager = createManager() + val manager = CommandsManager() manager.loadStartTs() manager.retransmitOverheadPercent = 1 manager.latency = 120 establishMediaRate(manager, 800_000) manager.updateRtt(50_000, 10_000) - // capacity = max(400_000, 960, MTU=1500) = 400_000 B; 20 * 516 = 10_320 B fits entirely + // capacity = max(800_173 * 0.5, max(960, MTU=1500)) = 400_086 B; 20 * 516 = 10_320 B fits entirely val startSeq = manager.sequenceNumber repeat(20) { sendPacket(manager, 500) @@ -243,39 +366,41 @@ class CommandsManagerTest { @Test fun `GIVEN budget exhausted on large packet WHEN smaller packet follows THEN skip both resends but report both`() = runTest { - val manager = createManager() + val manager = CommandsManager() manager.loadStartTs() manager.retransmitOverheadPercent = 1 manager.latency = 120 establishMediaRate(manager, 16_000) manager.updateRtt(50_000, 10_000) - // 20 packets in [startSeq..endSeq]: 18 * 416 = 7_488 B; +500 B (484 payload) = 7_988 B; 12 B left - // packet 20 wire = 68 + 16 = 84 B -> budgetExhausted; all 20 reported, only 19 retransmitted + // capacity = 8_085 B. 20 packets in [startSeq..endSeq]: 18 * 416 = 7_488 B, 597 B left + // packet 19 wire = 584 + 16 = 600 B -> budgetExhausted + // packet 20 wire = 68 + 16 = 84 B would still fit but is skipped too, all 20 are reported val startSeq = manager.sequenceNumber repeat(18) { sendPacket(manager, 400) } - sendPacket(manager, 484) + sendPacket(manager, 584) sendPacket(manager, 68) val endSeq = manager.sequenceNumber - 1 clearInvocations(socket) assertEquals(20, manager.reSendPackets(listOf(startSeq to endSeq), socket)) - verify(socket, times(19)).write(any()) + verify(socket, times(18)).write(any()) } @Test fun `GIVEN burst capacity on healthy link WHEN NAK spans short media window THEN resend all immediately`() = runTest { - val manager = createManager() + val manager = CommandsManager() manager.loadStartTs() manager.latency = 2000 establishMediaRate(manager, 50_000) manager.updateRtt(10_000, 2_000) - // rate = max(50_000 * 25%, 8_000) = 12_500 B/s - // capacity = max(50_000 * 0.5, 12_500 * 2000/1000, MTU=1500) = 25_000 B - // 200 * (100 + 16) = 23_200 B wire < 25_000 B + // media rate = 11 * (4545 + 16) = 50_171 B/s + // rate = max(50_171 * 25%, 8_000) = 12_542 B/s + // capacity = max(50_171 * 0.5, max(12_542 * 2000/1000, MTU=1500)) = 25_085 B + // 200 * (100 + 16) = 23_200 B wire < 25_085 B val startSeq = manager.sequenceNumber repeat(200) { sendPacket(manager, 100) @@ -289,7 +414,7 @@ class CommandsManagerTest { @Test fun `GIVEN first NAK within minResendInterval of original send WHEN second NAK follows quickly THEN honor first and suppress second`() = runTest { - val manager = createManager() + val manager = CommandsManager() manager.loadStartTs() manager.updateRtt(12_000, 0) // minResendInterval = min(max(12_000, 20_000), 30_000) = 20_000 us From c72bf1b1cdbf6cfaca83d3509c23b701114605ab Mon Sep 17 00:00:00 2001 From: pedroSG94 Date: Tue, 22 Sep 2026 20:26:13 +0200 Subject: [PATCH 3/4] move all retransmission to other class --- .../java/com/pedro/srt/srt/CommandsManager.kt | 171 +--------------- .../com/pedro/srt/srt/PacketRetransmission.kt | 188 ++++++++++++++++++ 2 files changed, 196 insertions(+), 163 deletions(-) create mode 100644 srt/src/main/java/com/pedro/srt/srt/PacketRetransmission.kt diff --git a/srt/src/main/java/com/pedro/srt/srt/CommandsManager.kt b/srt/src/main/java/com/pedro/srt/srt/CommandsManager.kt index 09e46efd1b..b742b3d6e9 100644 --- a/srt/src/main/java/com/pedro/srt/srt/CommandsManager.kt +++ b/srt/src/main/java/com/pedro/srt/srt/CommandsManager.kt @@ -36,8 +36,6 @@ import com.pedro.srt.utils.SrtSocket import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock import java.io.IOException -import kotlin.math.max -import kotlin.math.min import kotlin.random.Random /** @@ -46,8 +44,6 @@ import kotlin.random.Random class CommandsManager { private val TAG = "CommandsManager" - //used for packet lost - private val packetHandlingQueue = mutableListOf() var sequenceNumber: Int = generateInitialSequence() var messageNumber = 1 @@ -60,30 +56,12 @@ class CommandsManager { var videoDisabled = false var host = "" var latency = 120 //in millis - /** - * Max retransmit bandwidth as a percentage of the estimated media rate. - * Values <= 0 disable the limit (legacy behavior). - */ - var retransmitOverheadPercent: Int = 25 //Avoid write a packet in middle of other. private val writeSync = Mutex(locked = false) private var encryptor: EncryptionUtil? = null var videoCodec = VideoCodec.H264 var audioCodec = AudioCodec.AAC - - @Volatile - private var rtt = 0 - @Volatile - private var rttVariance = 0 - private var mediaBytesPerSecond = 0.0 - private var mediaWindowStartUs = 0L - private var mediaWindowBytes = 0L - private var retransmitTokens = 0.0 - private var lastTokenRefillUs = 0L - private var bucketInitialized = false - //used for unique packet lost - private var lastNakSequence = 0 - private var lastNakInitialized = false + private val packetRetransmission = PacketRetransmission() fun setPassphrase(passphrase: String, type: EncryptionType) { encryptor = if (passphrase.isEmpty() || type == EncryptionType.NONE) null else EncryptionUtil(type, passphrase) @@ -109,9 +87,7 @@ class CommandsManager { } fun updateRtt(rtt: Int, rttVariance: Int) { - val maxUs = latency * 1000 - this.rtt = rtt.coerceIn(0, maxUs) - this.rttVariance = rttVariance.coerceIn(0, maxUs) + packetRetransmission.updateRtt(rtt, rttVariance, latency) } @Throws(IOException::class) @@ -152,10 +128,10 @@ class CommandsManager { socketId = socketId ) sequenceNumber++ - packetHandlingQueue.add(dataPacket) - dropTooLatePackets(dataPacket.ts) + packetRetransmission.addPacket(dataPacket) + packetRetransmission.dropTooLatePackets(dataPacket.ts, latency) dataPacket.write() - trackMediaBytes(dataPacket.getSize(), TimeUtils.getCurrentTimeMicro()) + packetRetransmission.trackMediaBytes(dataPacket.getSize(), TimeUtils.getCurrentTimeMicro()) socket?.write(dataPacket) return dataPacket.getSize() } @@ -164,125 +140,14 @@ class CommandsManager { @Throws(IOException::class) suspend fun reSendPackets(lostRanges: List>, socket: SrtSocket?): Int { writeSync.withLock { - val budgetEnabled = retransmitOverheadPercent > 0 - val nowTs = getTs() - val nowUs = TimeUtils.getCurrentTimeMicro() - if (budgetEnabled) refillRetransmitTokens(nowUs) - - val latencyUs = latency * 1000 - val minResendInterval = min(max(rtt + 4 * rttVariance, MIN_RESEND_INTERVAL_US), latencyUs / 4) - val newlyReported = countNewLostPackets(lostRanges) - var budgetExhausted = false - for (packet in packetHandlingQueue) { - if (!isInLostRange(packet.sequenceNumber, lostRanges)) continue - if ((nowTs - packet.ts + rtt / 2) >= latencyUs) continue - if (packet.retransmitted && (nowTs - packet.lastSentTs) < minResendInterval) continue - if (budgetEnabled) { - if (budgetExhausted) continue - val packetSize = packet.getSize() - if (retransmitTokens < packetSize) { - budgetExhausted = true - continue - } - retransmitTokens -= packetSize - } - packet.retransmitted = true - packet.write() - socket?.write(packet) - packet.lastSentTs = nowTs - } - return newlyReported + return packetRetransmission.reSendPackets(lostRanges, socket, getTs(), latency, MTU) } } suspend fun updateHandlingQueue(lastPacketSequence: Int) { writeSync.withLock { - packetHandlingQueue.removeAll { - //discard confirmed packets - val diff = (lastPacketSequence - it.sequenceNumber) and 0x7FFFFFFF - diff in 1 until 0x40000000 - } - } - } - - //a nak reports the same missing packets again until received, only sequences after the last - //reported one are a new lost. Counted from the ranges so packets already dropped are counted too - private fun countNewLostPackets(lostRanges: List>): Int { - var count = 0 - lostRanges.forEach { (from, to) -> - val size = ((to - from) and 0x7FFFFFFF) + 1 - if (!lastNakInitialized) { - lastNakInitialized = true - lastNakSequence = to - count += size - } else { - val ahead = (to - lastNakSequence) and 0x7FFFFFFF - if (ahead in 1 until 0x40000000) { - count += min(ahead, size) - lastNakSequence = to - } - } - } - return count - } - - private fun isInLostRange(sequenceNumber: Int, lostRanges: List>): Boolean { - return lostRanges.any { (min, max) -> - ((sequenceNumber - min) and 0x7FFFFFFF) <= ((max - min) and 0x7FFFFFFF) - } - } - - private fun trackMediaBytes(bytes: Int, nowUs: Long) { - if (mediaWindowStartUs == 0L) mediaWindowStartUs = nowUs - mediaWindowBytes += bytes - val elapsed = nowUs - mediaWindowStartUs - val window = if (mediaBytesPerSecond > 0.0) MEDIA_RATE_WINDOW_US else MEDIA_RATE_WARMUP_WINDOW_US - if (elapsed >= window) { - val rate = mediaWindowBytes.toDouble() * MEDIA_RATE_WINDOW_US / elapsed - //follow an increase at once to not limit below the real media rate, average a decrease - mediaBytesPerSecond = if (rate > mediaBytesPerSecond) rate else { - mediaBytesPerSecond * MEDIA_RATE_EWMA_OLD + rate * MEDIA_RATE_EWMA_NEW - } - mediaWindowStartUs = nowUs - mediaWindowBytes = 0 - } - } - - private fun getRetransmitRate(): Double { - val percent = retransmitOverheadPercent - val mediaRate = if (mediaBytesPerSecond > 0.0) mediaBytesPerSecond else MIN_RETRANSMIT_BYTES_PER_SECOND.toDouble() - return max(mediaRate * percent / 100.0, MIN_RETRANSMIT_BYTES_PER_SECOND.toDouble()) - } - - private fun getRetransmitCapacity(rate: Double): Int { - val burstCapacity = if (mediaBytesPerSecond > 0.0) { - (mediaBytesPerSecond * RETRANSMIT_BURST_WINDOW_US / MEDIA_RATE_WINDOW_US).toInt() - } else { - 0 - } - return max(burstCapacity, max((rate * latency / 1000.0).toInt(), MTU)) - } - - private fun refillRetransmitTokens(nowUs: Long) { - if (!bucketInitialized) { - val rate = getRetransmitRate() - retransmitTokens = getRetransmitCapacity(rate).toDouble() - lastTokenRefillUs = nowUs - bucketInitialized = true - return + packetRetransmission.updateHandlingQueue(lastPacketSequence) } - val elapsedUs = nowUs - lastTokenRefillUs - if (elapsedUs <= 0) return - val rate = getRetransmitRate() - val capacity = getRetransmitCapacity(rate) - retransmitTokens = min(retransmitTokens + rate * elapsedUs / MEDIA_RATE_WINDOW_US, capacity.toDouble()) - lastTokenRefillUs = nowUs - } - - private fun dropTooLatePackets(nowTs: Int) { - val thresholdUs = latency * 1000 - val firstKept = packetHandlingQueue.indexOfFirst { (nowTs - it.ts) <= thresholdUs } - if (firstKept > 0) packetHandlingQueue.subList(0, firstKept).clear() } @Throws(IOException::class) @@ -319,17 +184,7 @@ class CommandsManager { socketId = 0 startTS = 0L host = "" - packetHandlingQueue.clear() - rtt = 0 - rttVariance = 0 - mediaBytesPerSecond = 0.0 - mediaWindowStartUs = 0L - mediaWindowBytes = 0L - retransmitTokens = 0.0 - lastTokenRefillUs = 0L - bucketInitialized = false - lastNakSequence = 0 - lastNakInitialized = false + packetRetransmission.reset() } private fun generateInitialSequence(): Int { @@ -339,14 +194,4 @@ class CommandsManager { private fun generateSocketId(): Int { return Random.nextInt(1, Int.MAX_VALUE) } - - companion object { - private const val MIN_RETRANSMIT_BYTES_PER_SECOND = 8_000 - private const val MIN_RESEND_INTERVAL_US = 20_000 - private const val MEDIA_RATE_WINDOW_US = 1_000_000L - private const val MEDIA_RATE_WARMUP_WINDOW_US = 200_000L - private const val RETRANSMIT_BURST_WINDOW_US = 500_000L - private const val MEDIA_RATE_EWMA_OLD = 0.8 - private const val MEDIA_RATE_EWMA_NEW = 0.2 - } } diff --git a/srt/src/main/java/com/pedro/srt/srt/PacketRetransmission.kt b/srt/src/main/java/com/pedro/srt/srt/PacketRetransmission.kt new file mode 100644 index 0000000000..8e65ec4334 --- /dev/null +++ b/srt/src/main/java/com/pedro/srt/srt/PacketRetransmission.kt @@ -0,0 +1,188 @@ +package com.pedro.srt.srt + +import com.pedro.common.TimeUtils +import com.pedro.srt.srt.packets.DataPacket +import com.pedro.srt.utils.SrtSocket +import java.io.IOException +import kotlin.math.max +import kotlin.math.min +import kotlin.text.clear + +class PacketRetransmission { + companion object { + private const val MIN_RETRANSMIT_BYTES_PER_SECOND = 8_000 + private const val MIN_RESEND_INTERVAL_US = 20_000 + private const val MEDIA_RATE_WINDOW_US = 1_000_000L + private const val MEDIA_RATE_WARMUP_WINDOW_US = 200_000L + private const val RETRANSMIT_BURST_WINDOW_US = 500_000L + private const val MEDIA_RATE_EWMA_OLD = 0.8 + private const val MEDIA_RATE_EWMA_NEW = 0.2 + } + + private val packetHandlingQueue = mutableListOf() + /** + * Max retransmit bandwidth as a percentage of the estimated media rate. + * Values <= 0 disable the limit (legacy behavior). + */ + var retransmitOverheadPercent: Int = 25 + @Volatile + private var rtt = 0 + @Volatile + private var rttVariance = 0 + private var mediaBytesPerSecond = 0.0 + private var mediaWindowStartUs = 0L + private var mediaWindowBytes = 0L + private var retransmitTokens = 0.0 + private var lastTokenRefillUs = 0L + private var bucketInitialized = false + //used for unique packet lost + private var lastNakSequence = 0 + private var lastNakInitialized = false + + fun addPacket(packet: DataPacket) { + packetHandlingQueue.add(packet) + } + + fun updateRtt(rtt: Int, rttVariance: Int, latency: Int) { + val maxUs = latency * 1000 + this.rtt = rtt.coerceIn(0, maxUs) + this.rttVariance = rttVariance.coerceIn(0, maxUs) + } + + fun reset() { + packetHandlingQueue.clear() + rtt = 0 + rttVariance = 0 + mediaBytesPerSecond = 0.0 + mediaWindowStartUs = 0L + mediaWindowBytes = 0L + retransmitTokens = 0.0 + lastTokenRefillUs = 0L + bucketInitialized = false + lastNakSequence = 0 + lastNakInitialized = false + } + + @Throws(IOException::class) + suspend fun reSendPackets( + lostRanges: List>, + socket: SrtSocket?, + nowTs: Int, + latency: Int, + mtu: Int + ): Int { + val budgetEnabled = retransmitOverheadPercent > 0 + val nowUs = TimeUtils.getCurrentTimeMicro() + if (budgetEnabled) refillRetransmitTokens(nowUs, latency, mtu) + + val latencyUs = latency * 1000 + val minResendInterval = min(max(rtt + 4 * rttVariance, MIN_RESEND_INTERVAL_US), latencyUs / 4) + val newlyReported = countNewLostPackets(lostRanges) + var budgetExhausted = false + for (packet in packetHandlingQueue) { + if (!isInLostRange(packet.sequenceNumber, lostRanges)) continue + if ((nowTs - packet.ts + rtt / 2) >= latencyUs) continue + if (packet.retransmitted && (nowTs - packet.lastSentTs) < minResendInterval) continue + if (budgetEnabled) { + if (budgetExhausted) continue + val packetSize = packet.getSize() + if (retransmitTokens < packetSize) { + budgetExhausted = true + continue + } + retransmitTokens -= packetSize + } + packet.retransmitted = true + packet.write() + socket?.write(packet) + packet.lastSentTs = nowTs + } + return newlyReported + } + + suspend fun updateHandlingQueue(lastPacketSequence: Int) { + packetHandlingQueue.removeAll { + //discard confirmed packets + val diff = (lastPacketSequence - it.sequenceNumber) and 0x7FFFFFFF + diff in 1 until 0x40000000 + } + } + + private fun countNewLostPackets(lostRanges: List>): Int { + var count = 0 + lostRanges.forEach { (from, to) -> + val size = ((to - from) and 0x7FFFFFFF) + 1 + if (!lastNakInitialized) { + lastNakInitialized = true + lastNakSequence = to + count += size + } else { + val ahead = (to - lastNakSequence) and 0x7FFFFFFF + if (ahead in 1 until 0x40000000) { + count += min(ahead, size) + lastNakSequence = to + } + } + } + return count + } + + private fun isInLostRange(sequenceNumber: Int, lostRanges: List>): Boolean { + return lostRanges.any { (min, max) -> + ((sequenceNumber - min) and 0x7FFFFFFF) <= ((max - min) and 0x7FFFFFFF) + } + } + + fun trackMediaBytes(bytes: Int, nowUs: Long) { + if (mediaWindowStartUs == 0L) mediaWindowStartUs = nowUs + mediaWindowBytes += bytes + val elapsed = nowUs - mediaWindowStartUs + val window = if (mediaBytesPerSecond > 0.0) MEDIA_RATE_WINDOW_US else MEDIA_RATE_WARMUP_WINDOW_US + if (elapsed >= window) { + val rate = mediaWindowBytes.toDouble() * MEDIA_RATE_WINDOW_US / elapsed + //follow an increase at once to not limit below the real media rate, average a decrease + mediaBytesPerSecond = if (rate > mediaBytesPerSecond) rate else { + mediaBytesPerSecond * MEDIA_RATE_EWMA_OLD + rate * MEDIA_RATE_EWMA_NEW + } + mediaWindowStartUs = nowUs + mediaWindowBytes = 0 + } + } + + private fun getRetransmitRate(): Double { + val percent = retransmitOverheadPercent + val mediaRate = if (mediaBytesPerSecond > 0.0) mediaBytesPerSecond else MIN_RETRANSMIT_BYTES_PER_SECOND.toDouble() + return max(mediaRate * percent / 100.0, MIN_RETRANSMIT_BYTES_PER_SECOND.toDouble()) + } + + private fun getRetransmitCapacity(rate: Double, latency: Int, mtu: Int): Int { + val burstCapacity = if (mediaBytesPerSecond > 0.0) { + (mediaBytesPerSecond * RETRANSMIT_BURST_WINDOW_US / MEDIA_RATE_WINDOW_US).toInt() + } else { + 0 + } + return max(burstCapacity, max((rate * latency / 1000.0).toInt(), mtu)) + } + + fun dropTooLatePackets(nowTs: Int, latency: Int) { + val thresholdUs = latency * 1000 + val firstKept = packetHandlingQueue.indexOfFirst { (nowTs - it.ts) <= thresholdUs } + if (firstKept > 0) packetHandlingQueue.subList(0, firstKept).clear() + } + + private fun refillRetransmitTokens(nowUs: Long, latency: Int, mtu: Int) { + if (!bucketInitialized) { + val rate = getRetransmitRate() + retransmitTokens = getRetransmitCapacity(rate, latency, mtu).toDouble() + lastTokenRefillUs = nowUs + bucketInitialized = true + return + } + val elapsedUs = nowUs - lastTokenRefillUs + if (elapsedUs <= 0) return + val rate = getRetransmitRate() + val capacity = getRetransmitCapacity(rate, latency, mtu) + retransmitTokens = min(retransmitTokens + rate * elapsedUs / MEDIA_RATE_WINDOW_US, capacity.toDouble()) + lastTokenRefillUs = nowUs + } +} \ No newline at end of file From 5f23924b15442f9399aaa2944a01060d7d994524 Mon Sep 17 00:00:00 2001 From: pedroSG94 Date: Tue, 22 Sep 2026 20:42:12 +0200 Subject: [PATCH 4/4] move all retransmission to other class 2 --- .../java/com/pedro/srt/srt/CommandsManager.kt | 4 + .../com/pedro/srt/srt/PacketRetransmission.kt | 19 +- .../main/java/com/pedro/srt/srt/SrtClient.kt | 2 +- .../com/pedro/srt/srt/CommandsManagerTest.kt | 351 +-------------- .../pedro/srt/srt/PacketRetransmissionTest.kt | 408 ++++++++++++++++++ 5 files changed, 439 insertions(+), 345 deletions(-) create mode 100644 srt/src/test/java/com/pedro/srt/srt/PacketRetransmissionTest.kt diff --git a/srt/src/main/java/com/pedro/srt/srt/CommandsManager.kt b/srt/src/main/java/com/pedro/srt/srt/CommandsManager.kt index b742b3d6e9..9a5522b12f 100644 --- a/srt/src/main/java/com/pedro/srt/srt/CommandsManager.kt +++ b/srt/src/main/java/com/pedro/srt/srt/CommandsManager.kt @@ -63,6 +63,10 @@ class CommandsManager { var audioCodec = AudioCodec.AAC private val packetRetransmission = PacketRetransmission() + fun setRetransmitOverhead(percent: Int) { + packetRetransmission.retransmitOverheadPercent = percent + } + fun setPassphrase(passphrase: String, type: EncryptionType) { encryptor = if (passphrase.isEmpty() || type == EncryptionType.NONE) null else EncryptionUtil(type, passphrase) } diff --git a/srt/src/main/java/com/pedro/srt/srt/PacketRetransmission.kt b/srt/src/main/java/com/pedro/srt/srt/PacketRetransmission.kt index 8e65ec4334..af0e7aed7a 100644 --- a/srt/src/main/java/com/pedro/srt/srt/PacketRetransmission.kt +++ b/srt/src/main/java/com/pedro/srt/srt/PacketRetransmission.kt @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2024 pedroSG94. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + package com.pedro.srt.srt import com.pedro.common.TimeUtils @@ -6,7 +22,6 @@ import com.pedro.srt.utils.SrtSocket import java.io.IOException import kotlin.math.max import kotlin.math.min -import kotlin.text.clear class PacketRetransmission { companion object { @@ -100,7 +115,7 @@ class PacketRetransmission { return newlyReported } - suspend fun updateHandlingQueue(lastPacketSequence: Int) { + fun updateHandlingQueue(lastPacketSequence: Int) { packetHandlingQueue.removeAll { //discard confirmed packets val diff = (lastPacketSequence - it.sequenceNumber) and 0x7FFFFFFF diff --git a/srt/src/main/java/com/pedro/srt/srt/SrtClient.kt b/srt/src/main/java/com/pedro/srt/srt/SrtClient.kt index a645213abb..53c4f0a0a5 100644 --- a/srt/src/main/java/com/pedro/srt/srt/SrtClient.kt +++ b/srt/src/main/java/com/pedro/srt/srt/SrtClient.kt @@ -142,7 +142,7 @@ class SrtClient(private val connectChecker: ConnectChecker) { * Default 25. Values <= 0 disable the limit. */ fun setRetransmitOverhead(percent: Int) { - commandsManager.retransmitOverheadPercent = percent + commandsManager.setRetransmitOverhead(percent) } fun setDelay(millis: Long) { diff --git a/srt/src/test/java/com/pedro/srt/srt/CommandsManagerTest.kt b/srt/src/test/java/com/pedro/srt/srt/CommandsManagerTest.kt index f1a5c72298..e5d7de49ed 100644 --- a/srt/src/test/java/com/pedro/srt/srt/CommandsManagerTest.kt +++ b/srt/src/test/java/com/pedro/srt/srt/CommandsManagerTest.kt @@ -32,12 +32,15 @@ import org.mockito.Mock import org.mockito.MockedStatic import org.mockito.Mockito import org.mockito.Mockito.clearInvocations -import org.mockito.Mockito.never import org.mockito.Mockito.times import org.mockito.Mockito.verify import org.mockito.junit.MockitoJUnitRunner import org.mockito.kotlin.any +/** + * The retransmission rules are covered by PacketRetransmissionTest, this checks the wiring + * CommandsManager does around them + */ @RunWith(MockitoJUnitRunner::class) class CommandsManagerTest { @@ -59,10 +62,7 @@ class CommandsManagerTest { timeUtilsMocked.close() } - private suspend fun sendPacket( - manager: CommandsManager, - payloadSize: Int = 100 - ): Int { + private suspend fun sendPacket(manager: CommandsManager, payloadSize: Int = 100): Int { val packet = MpegTsPacket( buffer = ByteArray(payloadSize), type = MpegType.VIDEO, @@ -72,155 +72,8 @@ class CommandsManagerTest { return manager.writeData(packet, socket) } - private suspend fun establishMediaRate(manager: CommandsManager, bytesPerSecond: Int) { - // 11 equal chunks over a 1 s trackMediaBytes window. The rate is measured on the wire size, - // so it ends up as bytesPerSecond plus the 16 B header of each of the 11 packets - val chunkSize = bytesPerSecond / 11 - repeat(10) { - sendPacket(manager, chunkSize) - } - nowUs += 1_000_000 - sendPacket(manager, chunkSize) - } - @Test - fun `GIVEN repeated NAKs for same packet WHEN within time gate THEN resend once and again after clock advance`() = runTest { - val manager = CommandsManager() - manager.loadStartTs() - // latency 2000 ms: at +200 ms the packet is still inside the latency window - // (200_000 + rtt/2 = 250_000 < 2_000_000); default 120 ms would mark it too late - manager.latency = 2000 - manager.updateRtt(100_000, 25_000) - - val seq = manager.sequenceNumber - sendPacket(manager) - clearInvocations(socket) - - manager.reSendPackets(listOf(seq to seq), socket) - // first NAK is honored immediately; minResendInterval = min(max(200_000, 20_000), 500_000) = 200_000 us - verify(socket, times(1)).write(any()) - - manager.reSendPackets(listOf(seq to seq), socket) - // second NAK within 200_000 us of the retransmit is suppressed - verify(socket, times(1)).write(any()) - - nowUs += 200_000 - manager.reSendPackets(listOf(seq to seq), socket) - verify(socket, times(2)).write(any()) - } - - @Test - fun `GIVEN NAK range over retransmit budget WHEN tokens refill THEN resend oldest packets first`() = runTest { - val manager = CommandsManager() - manager.loadStartTs() - manager.retransmitOverheadPercent = 1 - manager.latency = 120 - establishMediaRate(manager, 16_000) - manager.updateRtt(50_000, 10_000) - - // media rate = 11 * (1454 + 16) = 16_170 B/s - // rate = max(16_170 * 1%, 8_000) = 8_000 B/s - // capacity = max(16_170 * 0.5, max(8_000 * 120/1000, MTU=1500)) = 8_085 B - // wire = 400 + 16 = 416 B; 19 * 416 = 7_904 fits, 20th needs 416 but only 181 B left - val startSeq = manager.sequenceNumber - repeat(20) { - sendPacket(manager, 400) - } - val endSeq = manager.sequenceNumber - 1 - clearInvocations(socket) - - assertEquals(20, manager.reSendPackets(listOf(startSeq to endSeq), socket)) - verify(socket, times(19)).write(any()) - - nowUs += 50_000 - // refill: 181 + 8_000 * 50_000/1_000_000 = 581 B, enough for the remaining 416 B packet - // still in time: 50_000 + rtt/2 = 75_000 < 120_000 us; never retransmitted, so no time gate - manager.reSendPackets(listOf(endSeq to endSeq), socket) - verify(socket, times(20)).write(any()) - } - - @Test - fun `GIVEN packet near latency expiry WHEN NAK received THEN skip resend`() = runTest { - val manager = CommandsManager() - manager.loadStartTs() - manager.latency = 1000 - manager.updateRtt(100_000, 0) - - val seq = manager.sequenceNumber - sendPacket(manager) - clearInvocations(socket) - - // (960_000 + 50_000) >= 1_000_000 -> too late, no resend; newlyReported still 1 - nowUs += 960_000 - assertEquals(1, manager.reSendPackets(listOf(seq to seq), socket)) - verify(socket, never()).write(any()) - } - - @Test - fun `GIVEN small loss on healthy link WHEN NAK received THEN resend all lost packets immediately`() = runTest { - val manager = CommandsManager() - manager.loadStartTs() - manager.updateRtt(10_000, 2_000) - - // no media rate yet: rate = 8_000 B/s, capacity = max(0, 960, MTU=1500) = 1_500 B - // 3 * (100 + 16) = 348 B < 1_500 B - val startSeq = manager.sequenceNumber - repeat(3) { - sendPacket(manager, 100) - } - val endSeq = manager.sequenceNumber - 1 - clearInvocations(socket) - - assertEquals(3, manager.reSendPackets(listOf(startSeq to endSeq), socket)) - verify(socket, times(3)).write(any()) - } - - @Test - fun `GIVEN retransmit budget disabled WHEN NAK range over budget THEN resend every packet`() = runTest { - val manager = CommandsManager() - manager.loadStartTs() - manager.retransmitOverheadPercent = 0 - manager.latency = 120 - establishMediaRate(manager, 16_000) - manager.updateRtt(50_000, 10_000) - - // same scenario as the budget test, where the token bucket stops before the last packets - val startSeq = manager.sequenceNumber - repeat(20) { - sendPacket(manager, 400) - } - val endSeq = manager.sequenceNumber - 1 - clearInvocations(socket) - - assertEquals(20, manager.reSendPackets(listOf(startSeq to endSeq), socket)) - verify(socket, times(20)).write(any()) - } - - @Test - fun `GIVEN retransmit budget disabled WHEN repeated NAKs for same packet THEN time gate still applies`() = runTest { - val manager = CommandsManager() - manager.loadStartTs() - manager.retransmitOverheadPercent = 0 - manager.updateRtt(12_000, 0) - // minResendInterval = min(max(12_000, 20_000), 30_000) = 20_000 us - - val seq = manager.sequenceNumber - sendPacket(manager) - clearInvocations(socket) - - manager.reSendPackets(listOf(seq to seq), socket) - verify(socket, times(1)).write(any()) - - manager.reSendPackets(listOf(seq to seq), socket) - verify(socket, times(1)).write(any()) - - nowUs += 20_000 - manager.reSendPackets(listOf(seq to seq), socket) - verify(socket, times(2)).write(any()) - } - - @Test - fun `GIVEN loss right after stream start WHEN NAK received THEN media rate is already estimated`() = runTest { + fun `GIVEN packets sent with writeData WHEN NAK received THEN media rate feeds the retransmit budget`() = runTest { val manager = CommandsManager() manager.loadStartTs() manager.latency = 120 @@ -232,9 +85,9 @@ class CommandsManagerTest { nowUs += 200_000 sendPacket(manager, 1316) - // rate = max(73_260 * 25%, 8_000) = 18_315 B/s - // capacity = max(73_260 * 0.5, max(18_315 * 120/1000, MTU=1500)) = 36_630 B - // 10 * 1332 = 13_320 B fit. Waiting a full second the budget would be 1_500 B, a single packet + // capacity = max(73_260 * 0.5, max(18_315 * 120/1000, MTU=1500)) = 36_630 B, so 10 * 1332 = 13_320 B fit. + // Tracking the media bytes before serializing the packet would measure 0 on every packet and + // leave the budget at its minimum of 1_500 B, a single packet val startSeq = manager.sequenceNumber repeat(10) { sendPacket(manager, 1316) @@ -245,190 +98,4 @@ class CommandsManagerTest { assertEquals(10, manager.reSendPackets(listOf(startSeq to endSeq), socket)) verify(socket, times(10)).write(any()) } - - @Test - fun `GIVEN NAK for a packet already dropped from the queue WHEN counting unique lost THEN count it`() = runTest { - val manager = CommandsManager() - manager.loadStartTs() - manager.latency = 120 - - val seq = manager.sequenceNumber - sendPacket(manager) - // at +200 ms the first packet is older than the latency window and dropTooLatePackets removes it - nowUs += 200_000 - sendPacket(manager) - clearInvocations(socket) - - // no longer in the queue so it can't be resent, but it is still a lost packet - assertEquals(1, manager.reSendPackets(listOf(seq to seq), socket)) - verify(socket, never()).write(any()) - } - - @Test - fun `GIVEN media rate increase WHEN next window is measured THEN budget follows it at once`() = runTest { - val manager = CommandsManager() - manager.loadStartTs() - manager.latency = 120 - establishMediaRate(manager, 16_000) - establishMediaRate(manager, 160_000) - - // media rate jumps from 16_170 to 160_171 B/s - // capacity = max(160_171 * 0.5, max(40_042 * 120/1000, MTU=1500)) = 80_085 B - // 40 * (1316 + 16) = 53_280 B fit. Averaged with the old rate it would be 22_485 B, only 16 packets - val startSeq = manager.sequenceNumber - repeat(40) { - sendPacket(manager, 1316) - } - val endSeq = manager.sequenceNumber - 1 - clearInvocations(socket) - - assertEquals(40, manager.reSendPackets(listOf(startSeq to endSeq), socket)) - verify(socket, times(40)).write(any()) - } - - @Test - fun `GIVEN ack reporting an out of range rtt WHEN NAK received THEN clamp it and keep resending`() = runTest { - val manager = CommandsManager() - manager.loadStartTs() - manager.latency = 120 - manager.updateRtt(Int.MAX_VALUE, Int.MAX_VALUE) - - // clamped to 120_000 us, so rtt / 2 leaves the packet inside the latency window. - // Without the clamp rtt / 2 alone is bigger than the window and nothing is ever resent - val seq = manager.sequenceNumber - sendPacket(manager) - clearInvocations(socket) - - assertEquals(1, manager.reSendPackets(listOf(seq to seq), socket)) - verify(socket, times(1)).write(any()) - } - - @Test - fun `GIVEN sequence wrap at max value WHEN NAK spans wrap THEN resend wrapped packets`() = runTest { - val manager = CommandsManager() - manager.loadStartTs() - manager.retransmitOverheadPercent = 0 - manager.sequenceNumber = 0x7FFFFFFE - - sendPacket(manager) - sendPacket(manager) - sendPacket(manager) - clearInvocations(socket) - - manager.reSendPackets(listOf(0x7FFFFFFE to 0), socket) - verify(socket, times(3)).write(any()) - } - - @Test - fun `GIVEN repeated NAK for same packet WHEN already reported THEN return zero newly reported`() = runTest { - val manager = CommandsManager() - manager.loadStartTs() - manager.retransmitOverheadPercent = 0 - - val seq = manager.sequenceNumber - sendPacket(manager) - - assertEquals(1, manager.reSendPackets(listOf(seq to seq), socket)) - assertEquals(0, manager.reSendPackets(listOf(seq to seq), socket)) - } - - @Test - fun `GIVEN active retransmit state WHEN reset called THEN allow immediate resend again`() = runTest { - val manager = CommandsManager() - manager.loadStartTs() - manager.retransmitOverheadPercent = 1 - manager.latency = 120 - establishMediaRate(manager, 800_000) - manager.updateRtt(50_000, 10_000) - - // capacity = max(800_173 * 0.5, max(960, MTU=1500)) = 400_086 B; 20 * 516 = 10_320 B fits entirely - val startSeq = manager.sequenceNumber - repeat(20) { - sendPacket(manager, 500) - } - val endSeq = manager.sequenceNumber - 1 - manager.reSendPackets(listOf(startSeq to endSeq), socket) - - manager.reset() - manager.loadStartTs() - manager.retransmitOverheadPercent = 1 - manager.latency = 120 - establishMediaRate(manager, 800_000) - manager.updateRtt(50_000, 10_000) - - val resetSeq = manager.sequenceNumber - sendPacket(manager, 500) - clearInvocations(socket) - - assertEquals(1, manager.reSendPackets(listOf(resetSeq to resetSeq), socket)) - verify(socket, times(1)).write(any()) - } - - @Test - fun `GIVEN budget exhausted on large packet WHEN smaller packet follows THEN skip both resends but report both`() = runTest { - val manager = CommandsManager() - manager.loadStartTs() - manager.retransmitOverheadPercent = 1 - manager.latency = 120 - establishMediaRate(manager, 16_000) - manager.updateRtt(50_000, 10_000) - - // capacity = 8_085 B. 20 packets in [startSeq..endSeq]: 18 * 416 = 7_488 B, 597 B left - // packet 19 wire = 584 + 16 = 600 B -> budgetExhausted - // packet 20 wire = 68 + 16 = 84 B would still fit but is skipped too, all 20 are reported - val startSeq = manager.sequenceNumber - repeat(18) { - sendPacket(manager, 400) - } - sendPacket(manager, 584) - sendPacket(manager, 68) - val endSeq = manager.sequenceNumber - 1 - clearInvocations(socket) - - assertEquals(20, manager.reSendPackets(listOf(startSeq to endSeq), socket)) - verify(socket, times(18)).write(any()) - } - - @Test - fun `GIVEN burst capacity on healthy link WHEN NAK spans short media window THEN resend all immediately`() = runTest { - val manager = CommandsManager() - manager.loadStartTs() - manager.latency = 2000 - establishMediaRate(manager, 50_000) - manager.updateRtt(10_000, 2_000) - - // media rate = 11 * (4545 + 16) = 50_171 B/s - // rate = max(50_171 * 25%, 8_000) = 12_542 B/s - // capacity = max(50_171 * 0.5, max(12_542 * 2000/1000, MTU=1500)) = 25_085 B - // 200 * (100 + 16) = 23_200 B wire < 25_085 B - val startSeq = manager.sequenceNumber - repeat(200) { - sendPacket(manager, 100) - } - val endSeq = manager.sequenceNumber - 1 - clearInvocations(socket) - - assertEquals(200, manager.reSendPackets(listOf(startSeq to endSeq), socket)) - verify(socket, times(200)).write(any()) - } - - @Test - fun `GIVEN first NAK within minResendInterval of original send WHEN second NAK follows quickly THEN honor first and suppress second`() = runTest { - val manager = CommandsManager() - manager.loadStartTs() - manager.updateRtt(12_000, 0) - // minResendInterval = min(max(12_000, 20_000), 30_000) = 20_000 us - - val seq = manager.sequenceNumber - sendPacket(manager) - clearInvocations(socket) - - nowUs += 10_000 - manager.reSendPackets(listOf(seq to seq), socket) - verify(socket, times(1)).write(any()) - - nowUs += 5_000 - manager.reSendPackets(listOf(seq to seq), socket) - verify(socket, times(1)).write(any()) - } } diff --git a/srt/src/test/java/com/pedro/srt/srt/PacketRetransmissionTest.kt b/srt/src/test/java/com/pedro/srt/srt/PacketRetransmissionTest.kt new file mode 100644 index 0000000000..40a8e64407 --- /dev/null +++ b/srt/src/test/java/com/pedro/srt/srt/PacketRetransmissionTest.kt @@ -0,0 +1,408 @@ +/* + * Copyright (C) 2024 pedroSG94. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.pedro.srt.srt + +import com.pedro.common.TimeUtils +import com.pedro.srt.srt.packets.DataPacket +import com.pedro.srt.srt.packets.SrtPacket +import com.pedro.srt.srt.packets.data.PacketPosition +import com.pedro.srt.utils.Constants +import com.pedro.srt.utils.SrtSocket +import kotlinx.coroutines.test.runTest +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Before +import org.junit.Test +import org.junit.runner.RunWith +import org.mockito.Mock +import org.mockito.MockedStatic +import org.mockito.Mockito +import org.mockito.Mockito.clearInvocations +import org.mockito.Mockito.never +import org.mockito.Mockito.times +import org.mockito.Mockito.verify +import org.mockito.junit.MockitoJUnitRunner +import org.mockito.kotlin.any + +@RunWith(MockitoJUnitRunner::class) +class PacketRetransmissionTest { + + @Mock + lateinit var socket: SrtSocket + + private lateinit var timeUtilsMocked: MockedStatic + private var nowUs = START_US + private var sequenceNumber = 0 + //latency and mtu are owned by CommandsManager and passed in on each call + private var latency = 120 + + @Before + fun setup() { + nowUs = START_US + sequenceNumber = 0 + latency = 120 + timeUtilsMocked = Mockito.mockStatic(TimeUtils::class.java) + timeUtilsMocked.`when`(TimeUtils::getCurrentTimeMicro).then { nowUs } + } + + @After + fun teardown() { + timeUtilsMocked.close() + } + + private fun nowTs() = (nowUs - START_US).toInt() + + /** + * Same steps CommandsManager.writeData does after building the packet + */ + private fun sendPacket(retransmission: PacketRetransmission, payloadSize: Int = 100): Int { + if (sequenceNumber.toUInt() > 0x7FFFFFFFu) sequenceNumber = 0 + val packet = DataPacket( + sequenceNumber = sequenceNumber, + packetPosition = PacketPosition.SINGLE, + messageNumber = 1, + ts = nowTs(), + payload = ByteArray(payloadSize) + ) + val sequence = sequenceNumber + sequenceNumber++ + retransmission.addPacket(packet) + retransmission.dropTooLatePackets(packet.ts, latency) + packet.write() + retransmission.trackMediaBytes(packet.getSize(), nowUs) + return sequence + } + + private fun establishMediaRate(retransmission: PacketRetransmission, bytesPerSecond: Int) { + // 11 equal chunks over a 1 s trackMediaBytes window. The rate is measured on the wire size, + // so it ends up as bytesPerSecond plus the 16 B header of each of the 11 packets + val chunkSize = bytesPerSecond / 11 + repeat(10) { + sendPacket(retransmission, chunkSize) + } + nowUs += 1_000_000 + sendPacket(retransmission, chunkSize) + } + + private suspend fun reSendPackets(retransmission: PacketRetransmission, from: Int, to: Int = from): Int { + return retransmission.reSendPackets(listOf(from to to), socket, nowTs(), latency, Constants.MTU) + } + + @Test + fun `GIVEN repeated NAKs for same packet WHEN within time gate THEN resend once and again after clock advance`() = runTest { + val retransmission = PacketRetransmission() + // latency 2000 ms: at +200 ms the packet is still inside the latency window + // (200_000 + rtt/2 = 250_000 < 2_000_000); default 120 ms would mark it too late + latency = 2000 + retransmission.updateRtt(100_000, 25_000, latency) + + val seq = sendPacket(retransmission) + + reSendPackets(retransmission, seq) + // first NAK is honored immediately; minResendInterval = min(max(200_000, 20_000), 500_000) = 200_000 us + verify(socket, times(1)).write(any()) + + reSendPackets(retransmission, seq) + // second NAK within 200_000 us of the retransmit is suppressed + verify(socket, times(1)).write(any()) + + nowUs += 200_000 + reSendPackets(retransmission, seq) + verify(socket, times(2)).write(any()) + } + + @Test + fun `GIVEN NAK range over retransmit budget WHEN tokens refill THEN resend oldest packets first`() = runTest { + val retransmission = PacketRetransmission() + retransmission.retransmitOverheadPercent = 1 + establishMediaRate(retransmission, 16_000) + retransmission.updateRtt(50_000, 10_000, latency) + + // media rate = 11 * (1454 + 16) = 16_170 B/s + // rate = max(16_170 * 1%, 8_000) = 8_000 B/s + // capacity = max(16_170 * 0.5, max(8_000 * 120/1000, MTU=1500)) = 8_085 B + // wire = 400 + 16 = 416 B; 19 * 416 = 7_904 fits, 20th needs 416 but only 181 B left + val startSeq = sequenceNumber + repeat(20) { + sendPacket(retransmission, 400) + } + val endSeq = sequenceNumber - 1 + + assertEquals(20, reSendPackets(retransmission, startSeq, endSeq)) + verify(socket, times(19)).write(any()) + + nowUs += 50_000 + // refill: 181 + 8_000 * 50_000/1_000_000 = 581 B, enough for the remaining 416 B packet + // still in time: 50_000 + rtt/2 = 75_000 < 120_000 us; never retransmitted, so no time gate + reSendPackets(retransmission, endSeq) + verify(socket, times(20)).write(any()) + } + + @Test + fun `GIVEN packet near latency expiry WHEN NAK received THEN skip resend`() = runTest { + val retransmission = PacketRetransmission() + latency = 1000 + retransmission.updateRtt(100_000, 0, latency) + + val seq = sendPacket(retransmission) + + // (960_000 + 50_000) >= 1_000_000 -> too late, no resend; newlyReported still 1 + nowUs += 960_000 + assertEquals(1, reSendPackets(retransmission, seq)) + verify(socket, never()).write(any()) + } + + @Test + fun `GIVEN small loss on healthy link WHEN NAK received THEN resend all lost packets immediately`() = runTest { + val retransmission = PacketRetransmission() + retransmission.updateRtt(10_000, 2_000, latency) + + // no media rate yet: rate = 8_000 B/s, capacity = max(0, 960, MTU=1500) = 1_500 B + // 3 * (100 + 16) = 348 B < 1_500 B + val startSeq = sequenceNumber + repeat(3) { + sendPacket(retransmission, 100) + } + val endSeq = sequenceNumber - 1 + + assertEquals(3, reSendPackets(retransmission, startSeq, endSeq)) + verify(socket, times(3)).write(any()) + } + + @Test + fun `GIVEN retransmit budget disabled WHEN NAK range over budget THEN resend every packet`() = runTest { + val retransmission = PacketRetransmission() + retransmission.retransmitOverheadPercent = 0 + establishMediaRate(retransmission, 16_000) + retransmission.updateRtt(50_000, 10_000, latency) + + // same scenario as the budget test, where the token bucket stops before the last packets + val startSeq = sequenceNumber + repeat(20) { + sendPacket(retransmission, 400) + } + val endSeq = sequenceNumber - 1 + + assertEquals(20, reSendPackets(retransmission, startSeq, endSeq)) + verify(socket, times(20)).write(any()) + } + + @Test + fun `GIVEN retransmit budget disabled WHEN repeated NAKs for same packet THEN time gate still applies`() = runTest { + val retransmission = PacketRetransmission() + retransmission.retransmitOverheadPercent = 0 + retransmission.updateRtt(12_000, 0, latency) + // minResendInterval = min(max(12_000, 20_000), 30_000) = 20_000 us + + val seq = sendPacket(retransmission) + + reSendPackets(retransmission, seq) + verify(socket, times(1)).write(any()) + + reSendPackets(retransmission, seq) + verify(socket, times(1)).write(any()) + + nowUs += 20_000 + reSendPackets(retransmission, seq) + verify(socket, times(2)).write(any()) + } + + @Test + fun `GIVEN loss right after stream start WHEN NAK received THEN media rate is already estimated`() = runTest { + val retransmission = PacketRetransmission() + + // 200 ms of media is enough for a first estimation: 11 * (1316 + 16) B in 200 ms -> 73_260 B/s + repeat(10) { + sendPacket(retransmission, 1316) + } + nowUs += 200_000 + sendPacket(retransmission, 1316) + + // rate = max(73_260 * 25%, 8_000) = 18_315 B/s + // capacity = max(73_260 * 0.5, max(18_315 * 120/1000, MTU=1500)) = 36_630 B + // 10 * 1332 = 13_320 B fit. Waiting a full second the budget would be 1_500 B, a single packet + val startSeq = sequenceNumber + repeat(10) { + sendPacket(retransmission, 1316) + } + val endSeq = sequenceNumber - 1 + + assertEquals(10, reSendPackets(retransmission, startSeq, endSeq)) + verify(socket, times(10)).write(any()) + } + + @Test + fun `GIVEN NAK for a packet already dropped from the queue WHEN counting unique lost THEN count it`() = runTest { + val retransmission = PacketRetransmission() + + val seq = sendPacket(retransmission) + // at +200 ms the first packet is older than the latency window and dropTooLatePackets removes it + nowUs += 200_000 + sendPacket(retransmission) + + // no longer in the queue so it can't be resent, but it is still a lost packet + assertEquals(1, reSendPackets(retransmission, seq)) + verify(socket, never()).write(any()) + } + + @Test + fun `GIVEN media rate increase WHEN next window is measured THEN budget follows it at once`() = runTest { + val retransmission = PacketRetransmission() + establishMediaRate(retransmission, 16_000) + establishMediaRate(retransmission, 160_000) + + // media rate jumps from 16_170 to 160_171 B/s + // capacity = max(160_171 * 0.5, max(40_042 * 120/1000, MTU=1500)) = 80_085 B + // 40 * (1316 + 16) = 53_280 B fit. Averaged with the old rate it would be 22_485 B, only 16 packets + val startSeq = sequenceNumber + repeat(40) { + sendPacket(retransmission, 1316) + } + val endSeq = sequenceNumber - 1 + + assertEquals(40, reSendPackets(retransmission, startSeq, endSeq)) + verify(socket, times(40)).write(any()) + } + + @Test + fun `GIVEN ack reporting an out of range rtt WHEN NAK received THEN clamp it and keep resending`() = runTest { + val retransmission = PacketRetransmission() + retransmission.updateRtt(Int.MAX_VALUE, Int.MAX_VALUE, latency) + + // clamped to 120_000 us, so rtt / 2 leaves the packet inside the latency window. + // Without the clamp rtt / 2 alone is bigger than the window and nothing is ever resent + val seq = sendPacket(retransmission) + + assertEquals(1, reSendPackets(retransmission, seq)) + verify(socket, times(1)).write(any()) + } + + @Test + fun `GIVEN sequence wrap at max value WHEN NAK spans wrap THEN resend wrapped packets`() = runTest { + val retransmission = PacketRetransmission() + retransmission.retransmitOverheadPercent = 0 + sequenceNumber = 0x7FFFFFFE + + sendPacket(retransmission) + sendPacket(retransmission) + sendPacket(retransmission) + + reSendPackets(retransmission, 0x7FFFFFFE, 0) + verify(socket, times(3)).write(any()) + } + + @Test + fun `GIVEN repeated NAK for same packet WHEN already reported THEN return zero newly reported`() = runTest { + val retransmission = PacketRetransmission() + retransmission.retransmitOverheadPercent = 0 + + val seq = sendPacket(retransmission) + + assertEquals(1, reSendPackets(retransmission, seq)) + assertEquals(0, reSendPackets(retransmission, seq)) + } + + @Test + fun `GIVEN active retransmit state WHEN reset called THEN allow immediate resend again`() = runTest { + val retransmission = PacketRetransmission() + retransmission.retransmitOverheadPercent = 1 + establishMediaRate(retransmission, 800_000) + retransmission.updateRtt(50_000, 10_000, latency) + + // capacity = max(800_173 * 0.5, max(960, MTU=1500)) = 400_086 B; 20 * 516 = 10_320 B fits entirely + val startSeq = sequenceNumber + repeat(20) { + sendPacket(retransmission, 500) + } + val endSeq = sequenceNumber - 1 + reSendPackets(retransmission, startSeq, endSeq) + + retransmission.reset() + establishMediaRate(retransmission, 800_000) + retransmission.updateRtt(50_000, 10_000, latency) + + val resetSeq = sendPacket(retransmission, 500) + clearInvocations(socket) + + assertEquals(1, reSendPackets(retransmission, resetSeq)) + verify(socket, times(1)).write(any()) + } + + @Test + fun `GIVEN budget exhausted on large packet WHEN smaller packet follows THEN skip both resends but report both`() = runTest { + val retransmission = PacketRetransmission() + retransmission.retransmitOverheadPercent = 1 + establishMediaRate(retransmission, 16_000) + retransmission.updateRtt(50_000, 10_000, latency) + + // capacity = 8_085 B. 20 packets in [startSeq..endSeq]: 18 * 416 = 7_488 B, 597 B left + // packet 19 wire = 584 + 16 = 600 B -> budgetExhausted + // packet 20 wire = 68 + 16 = 84 B would still fit but is skipped too, all 20 are reported + val startSeq = sequenceNumber + repeat(18) { + sendPacket(retransmission, 400) + } + sendPacket(retransmission, 584) + sendPacket(retransmission, 68) + val endSeq = sequenceNumber - 1 + + assertEquals(20, reSendPackets(retransmission, startSeq, endSeq)) + verify(socket, times(18)).write(any()) + } + + @Test + fun `GIVEN burst capacity on healthy link WHEN NAK spans short media window THEN resend all immediately`() = runTest { + val retransmission = PacketRetransmission() + latency = 2000 + establishMediaRate(retransmission, 50_000) + retransmission.updateRtt(10_000, 2_000, latency) + + // media rate = 11 * (4545 + 16) = 50_171 B/s + // rate = max(50_171 * 25%, 8_000) = 12_542 B/s + // capacity = max(50_171 * 0.5, max(12_542 * 2000/1000, MTU=1500)) = 25_085 B + // 200 * (100 + 16) = 23_200 B wire < 25_085 B + val startSeq = sequenceNumber + repeat(200) { + sendPacket(retransmission, 100) + } + val endSeq = sequenceNumber - 1 + + assertEquals(200, reSendPackets(retransmission, startSeq, endSeq)) + verify(socket, times(200)).write(any()) + } + + @Test + fun `GIVEN first NAK within minResendInterval of original send WHEN second NAK follows quickly THEN honor first and suppress second`() = runTest { + val retransmission = PacketRetransmission() + retransmission.updateRtt(12_000, 0, latency) + // minResendInterval = min(max(12_000, 20_000), 30_000) = 20_000 us + + val seq = sendPacket(retransmission) + + nowUs += 10_000 + reSendPackets(retransmission, seq) + verify(socket, times(1)).write(any()) + + nowUs += 5_000 + reSendPackets(retransmission, seq) + verify(socket, times(1)).write(any()) + } + + private companion object { + const val START_US = 1_000_000L + } +}