diff --git a/library/src/main/java/com/pedro/library/util/streamclient/SrtStreamClient.kt b/library/src/main/java/com/pedro/library/util/streamclient/SrtStreamClient.kt index 6d9944191..ec65582d9 100644 --- a/library/src/main/java/com/pedro/library/util/streamclient/SrtStreamClient.kt +++ b/library/src/main/java/com/pedro/library/util/streamclient/SrtStreamClient.kt @@ -183,4 +183,17 @@ class SrtStreamClient( * Packets lost reported by NAK command. Increment each time a NAK is received. */ fun getPacketsLost() = srtClient.packetsLost + + /** + * Unique lost sequence numbers reported by NAK. Each sequence is counted once. + */ + fun getPacketsLostUnique() = srtClient.packetsLostUnique + + /** + * Max retransmit bandwidth as a percentage of the estimated media rate (libsrt SRTO_OHEADBW). + * Default 25. Values <= 0 disable the limit. + */ + fun setRetransmitOverhead(percent: Int) { + srtClient.setRetransmitOverhead(percent) + } } \ No newline at end of file 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 312541c27..5afa10909 100644 --- a/srt/src/main/java/com/pedro/srt/srt/CommandsManager.kt +++ b/srt/src/main/java/com/pedro/srt/srt/CommandsManager.kt @@ -36,12 +36,14 @@ 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 /** * Created by pedro on 23/8/23. */ -class CommandsManager { +class CommandsManager(private val timeProvider: () -> Long = { TimeUtils.getCurrentTimeMicro() }) { private val TAG = "CommandsManager" //used for packet lost @@ -58,12 +60,26 @@ 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 + private var rtt = 0 + 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 + fun setPassphrase(passphrase: String, type: EncryptionType) { encryptor = if (passphrase.isEmpty() || type == EncryptionType.NONE) null else EncryptionUtil(type, passphrase) } @@ -79,12 +95,19 @@ class CommandsManager { fun encryptionEnabled() = encryptor != null fun loadStartTs() { - startTS = TimeUtils.getCurrentTimeMicro() + startTS = timeProvider() localSocketId = generateSocketId() } fun getTs(): Int { - return (TimeUtils.getCurrentTimeMicro() - startTS).toInt() + return (timeProvider() - startTS).toInt() + } + + suspend fun updateRtt(rtt: Int, rttVariance: Int) { + writeSync.withLock { + this.rtt = rtt + this.rttVariance = rttVariance + } } @Throws(IOException::class) @@ -127,6 +150,7 @@ class CommandsManager { sequenceNumber++ packetHandlingQueue.add(dataPacket) dropTooLatePackets(dataPacket.ts) + trackMediaBytes(packet.buffer.size, timeProvider()) dataPacket.write() socket?.write(dataPacket) return dataPacket.getSize() @@ -134,18 +158,45 @@ class CommandsManager { } @Throws(IOException::class) - suspend fun reSendPackets(lostRanges: List>, socket: SrtSocket?) { + suspend fun reSendPackets(lostRanges: List>, socket: SrtSocket?): Int { writeSync.withLock { - val dataPackets = packetHandlingQueue.filter { packet -> - lostRanges.any { (min, max) -> - ((packet.sequenceNumber - min) and 0x7FFFFFFF) <= ((max - min) and 0x7FFFFFFF) - } + val unlimited = retransmitOverheadPercent <= 0 + val nowTs = getTs() + val nowUs = timeProvider() + if (!unlimited) { + refillRetransmitTokens(nowUs) } - dataPackets.forEach { packet -> + val latencyUs = latency * 1000 + val minResendInterval = if (unlimited) 0 else { + min(max(rtt + 4 * rttVariance, MIN_RESEND_INTERVAL_US), latencyUs / 4) + } + var newlyReported = 0 + 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 (budgetExhausted) continue + val packetSize = dataPacketWireSize(packet) + if (retransmitTokens < packetSize) { + budgetExhausted = true + continue + } + retransmitTokens -= packetSize + } packet.retransmitted = true packet.write() socket?.write(packet) + packet.lastSentTs = nowTs } + return newlyReported } } @@ -159,6 +210,63 @@ class CommandsManager { } } + 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 + if (elapsed >= MEDIA_RATE_WINDOW_US) { + val rate = mediaWindowBytes.toDouble() * MEDIA_RATE_WINDOW_US / elapsed + mediaBytesPerSecond = if (mediaBytesPerSecond == 0.0) 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 { + // 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 { + 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 + } + 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 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 } @@ -200,6 +308,14 @@ class CommandsManager { startTS = 0L host = "" packetHandlingQueue.clear() + rtt = 0 + rttVariance = 0 + mediaBytesPerSecond = 0.0 + mediaWindowStartUs = 0L + mediaWindowBytes = 0L + retransmitTokens = 0.0 + lastTokenRefillUs = 0L + bucketInitialized = false } private fun generateInitialSequence(): Int { @@ -209,4 +325,14 @@ class CommandsManager { private fun generateSocketId(): Int { return Random.nextInt(1, Int.MAX_VALUE) } -} \ No newline at end of file + + 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 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/SrtClient.kt b/srt/src/main/java/com/pedro/srt/srt/SrtClient.kt index 81ccf7d2c..a645213ab 100644 --- a/srt/src/main/java/com/pedro/srt/srt/SrtClient.kt +++ b/srt/src/main/java/com/pedro/srt/srt/SrtClient.kt @@ -107,6 +107,11 @@ class SrtClient(private val connectChecker: ConnectChecker) { private set var packetsLost = 0 private set + /** + * Unique lost sequence numbers reported by NAK. Each sequence is counted once. + */ + var packetsLostUnique = 0 + private set var socketType = SocketType.JAVA var socketTimeout = StreamSocket.DEFAULT_TIMEOUT @@ -132,6 +137,14 @@ class SrtClient(private val connectChecker: ConnectChecker) { commandsManager.latency = latency } + /** + * Max retransmit bandwidth as a percentage of the estimated media rate (libsrt SRTO_OHEADBW). + * Default 25. Values <= 0 disable the limit. + */ + fun setRetransmitOverhead(percent: Int) { + commandsManager.retransmitOverheadPercent = percent + } + fun setDelay(millis: Long) { srtSender.setDelay(millis) } @@ -326,6 +339,7 @@ class SrtClient(private val connectChecker: ConnectChecker) { commandsManager.reset() rtt = 0 packetsLost = 0 + packetsLostUnique = 0 job?.cancelAndJoin() job = null scope.cancel() @@ -405,6 +419,7 @@ class SrtClient(private val connectChecker: ConnectChecker) { commandsManager.updateHandlingQueue(lastPacketSequence) if (ackSequence != 0) { rtt = srtPacket.rtt + commandsManager.updateRtt(srtPacket.rtt, srtPacket.rttVariance) commandsManager.writeAck2(ackSequence, socket) } } @@ -412,7 +427,7 @@ class SrtClient(private val connectChecker: ConnectChecker) { //packet lost reported, we should resend it val lostRanges = srtPacket.getNakRanges() this.packetsLost += srtPacket.getLostCount() - commandsManager.reSendPackets(lostRanges, socket) + packetsLostUnique += commandsManager.reSendPackets(lostRanges, socket) } is Shutdown -> { onMainThread { 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 a3a7110ec..cdb992025 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 @@ -41,6 +41,9 @@ class DataPacket( var payload: ByteArray = byteArrayOf() ): SrtPacket() { + var lastSentTs: Int = 0 + var nakReported: Boolean = false + fun write() { resetBuffer() val headerData = (PacketType.DATA.value shl 31) or (sequenceNumber and 0x7FFFFFFF) diff --git a/srt/src/test/java/com/pedro/srt/srt/CommandsManagerTest.kt b/srt/src/test/java/com/pedro/srt/srt/CommandsManagerTest.kt new file mode 100644 index 000000000..f4f02c503 --- /dev/null +++ b/srt/src/test/java/com/pedro/srt/srt/CommandsManagerTest.kt @@ -0,0 +1,309 @@ +/* + * 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.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.Assert.assertEquals +import org.junit.Before +import org.junit.Test +import org.junit.runner.RunWith +import org.mockito.Mock +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 CommandsManagerTest { + + @Mock + lateinit var socket: SrtSocket + + private var nowUs = 1_000_000L + + @Before + fun setup() { + nowUs = 1_000_000L + } + + private fun createManager(): CommandsManager { + return CommandsManager { nowUs } + } + + private suspend fun sendPacket( + manager: CommandsManager, + payloadSize: Int = 100 + ): Int { + val packet = MpegTsPacket( + buffer = ByteArray(payloadSize), + type = MpegType.VIDEO, + packetPosition = PacketPosition.SINGLE, + isKey = false + ) + return manager.writeData(packet, socket) + } + + private suspend fun establishMediaRate(manager: CommandsManager, bytesPerSecond: Int) { + // 11 equal chunks over a 1 s trackMediaBytes window -> measured rate ~= bytesPerSecond + 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 = createManager() + 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 = createManager() + 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 + 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: 96 + 8_000 * 50_000/1_000_000 = 496 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 = createManager() + 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 = createManager() + 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 overhead disabled WHEN repeated NAKs received THEN resend without limits`() = runTest { + val manager = createManager() + manager.loadStartTs() + manager.retransmitOverheadPercent = 0 + + val seq = manager.sequenceNumber + sendPacket(manager) + clearInvocations(socket) + + manager.reSendPackets(listOf(seq to seq), socket) + manager.reSendPackets(listOf(seq to seq), socket) + verify(socket, times(2)).write(any()) + } + + @Test + fun `GIVEN sequence wrap at max value WHEN NAK spans wrap THEN resend wrapped packets`() = runTest { + val manager = createManager() + 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 = createManager() + 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 = createManager() + 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 + 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 = createManager() + 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 + val startSeq = manager.sequenceNumber + repeat(18) { + sendPacket(manager, 400) + } + sendPacket(manager, 484) + 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()) + } + + @Test + fun `GIVEN burst capacity on healthy link WHEN NAK spans short media window THEN resend all immediately`() = runTest { + val manager = createManager() + 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 + 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 = createManager() + 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()) + } +}