From 6561adfec74a1aa1ca1a1123ae800483350b5700 Mon Sep 17 00:00:00 2001 From: pedroSG94 Date: Wed, 29 Jul 2026 21:03:02 +0200 Subject: [PATCH 1/2] adding poolbuffer --- .../main/java/com/pedro/common/BufferPool.kt | 101 ++++++++++++ .../main/java/com/pedro/common/Extensions.kt | 10 +- .../com/pedro/common/StreamBlockingQueue.kt | 16 +- .../java/com/pedro/common/base/BaseSender.kt | 65 ++++++-- .../java/com/pedro/common/BufferPoolTest.kt | 116 ++++++++++++++ .../com/pedro/common/base/BaseSenderTest.kt | 149 ++++++++++++++++++ .../java/com/pedro/rtmp/rtmp/RtmpClient.kt | 5 +- .../java/com/pedro/rtmp/rtmp/RtmpSender.kt | 38 ++--- .../java/com/pedro/rtsp/rtsp/RtspClient.kt | 5 +- .../java/com/pedro/rtsp/rtsp/RtspSender.kt | 50 +++--- .../main/java/com/pedro/srt/srt/SrtClient.kt | 5 +- .../main/java/com/pedro/srt/srt/SrtSender.kt | 18 +-- .../java/com/pedro/srt/srt/SrtSenderTest.kt | 30 +++- udp/src/main/java/com/pedro/udp/UdpClient.kt | 4 +- udp/src/main/java/com/pedro/udp/UdpSender.kt | 17 +- .../main/java/com/pedro/whip/WhipClient.kt | 5 +- .../main/java/com/pedro/whip/WhipSender.kt | 50 +++--- 17 files changed, 561 insertions(+), 123 deletions(-) create mode 100644 common/src/main/java/com/pedro/common/BufferPool.kt create mode 100644 common/src/test/java/com/pedro/common/BufferPoolTest.kt create mode 100644 common/src/test/java/com/pedro/common/base/BaseSenderTest.kt diff --git a/common/src/main/java/com/pedro/common/BufferPool.kt b/common/src/main/java/com/pedro/common/BufferPool.kt new file mode 100644 index 000000000..42cbe5f89 --- /dev/null +++ b/common/src/main/java/com/pedro/common/BufferPool.kt @@ -0,0 +1,101 @@ +/* + * 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.common + +/** + * Created by pedro on 29/7/26. + * + * Pool of reusable ByteArray to avoid allocating one per encoded frame. + * + * An encoded frame must be copied out of the MediaCodec output buffer before giving it back to + * the codec, so the copy is unavoidable, but the allocation is not. At 6Mbps a stream allocates + * around 750KB/s of short lived arrays, and all of it ends in the GC. + * + * Buffers are grouped in power of two size classes so a frame gets the smallest available array + * able to contain it. Arrays are handed out dirty (never zeroed) because the caller always + * writes the whole frame and reads it back limited to the frame size. + * + * Thread safe: acquired from the encoder thread and released from the sender thread. + */ +class BufferPool( + private val maxBuffersPerSizeClass: Int = 8, + private val maxRetainedBytes: Long = 16 * 1024 * 1024 +) { + + companion object { + //1KB, audio frames are smaller but a smaller class is not worth the fragmentation + private const val MIN_SIZE_CLASS = 10 + //4MB, bigger frames are allocated and discarded as before + private const val MAX_SIZE_CLASS = 22 + } + + private val buckets = arrayOfNulls>(MAX_SIZE_CLASS - MIN_SIZE_CLASS + 1) + private val lock = Any() + private var retainedBytes = 0L + + /** + * Return an array of at least [minSize] bytes. The content is undefined. + */ + fun acquire(minSize: Int): ByteArray { + val sizeClass = sizeClassOf(minSize) + if (sizeClass > MAX_SIZE_CLASS) return ByteArray(minSize) + synchronized(lock) { + val buffer = buckets[sizeClass - MIN_SIZE_CLASS]?.removeLastOrNull() + if (buffer != null) { + retainedBytes -= buffer.size + return buffer + } + } + return ByteArray(1 shl sizeClass) + } + + /** + * Give an array back to the pool. Arrays not created by [acquire], and arrays already released, + * are ignored, so the same array can never be handed out to two users at once. + */ + fun release(buffer: ByteArray) { + val size = buffer.size + if (Integer.bitCount(size) != 1) return + val sizeClass = 31 - Integer.numberOfLeadingZeros(size) + if (sizeClass !in MIN_SIZE_CLASS..MAX_SIZE_CLASS) return + synchronized(lock) { + if (retainedBytes + size > maxRetainedBytes) return + val bucket = buckets[sizeClass - MIN_SIZE_CLASS] + ?: ArrayDeque().also { buckets[sizeClass - MIN_SIZE_CLASS] = it } + if (bucket.size >= maxBuffersPerSizeClass) return + //a double release would put the same array twice in the bucket and corrupt one of its users + if (bucket.any { it === buffer }) return + bucket.addLast(buffer) + retainedBytes += size + } + } + + fun clear() { + synchronized(lock) { + buckets.fill(null) + retainedBytes = 0 + } + } + + fun getRetainedBytes(): Long = synchronized(lock) { retainedBytes } + + //ceil(log2(size)) clamped to the smallest size class + private fun sizeClassOf(size: Int): Int { + val value = maxOf(size, 1 shl MIN_SIZE_CLASS) + return 32 - Integer.numberOfLeadingZeros(value - 1) + } +} diff --git a/common/src/main/java/com/pedro/common/Extensions.kt b/common/src/main/java/com/pedro/common/Extensions.kt index c95210b80..9078e4a3a 100644 --- a/common/src/main/java/com/pedro/common/Extensions.kt +++ b/common/src/main/java/com/pedro/common/Extensions.kt @@ -357,4 +357,12 @@ fun ByteBuffer.removeHeader(): ByteBuffer { fun ByteArray.writeUInt32(offset: Int, value: Int) { value.toUInt32().copyInto(this, offset) -} \ No newline at end of file +} + +fun ByteBuffer.clone(data: ByteArray): ByteBuffer { + val length = limit() + val source = duplicate() + source.position(0) + source.get(data, 0, length) + return ByteBuffer.wrap(data, 0, length).slice() +} diff --git a/common/src/main/java/com/pedro/common/StreamBlockingQueue.kt b/common/src/main/java/com/pedro/common/StreamBlockingQueue.kt index adc8a083f..00709eef3 100644 --- a/common/src/main/java/com/pedro/common/StreamBlockingQueue.kt +++ b/common/src/main/java/com/pedro/common/StreamBlockingQueue.kt @@ -46,9 +46,19 @@ class StreamBlockingQueue(var capacity: Int) { cacheQueue.drainTo(destiny.cacheQueue) } - fun clear() { - queue.clear() - cacheQueue.clear() + /** + * @param onRemove called for every discarded frame, used to recycle pooled buffers. + */ + fun clear(onRemove: ((MediaFrame) -> Unit)? = null) { + if (onRemove != null) { + val removed = mutableListOf() + queue.drainTo(removed) + cacheQueue.drainTo(removed) + removed.forEach(onRemove) + } else { + queue.clear() + cacheQueue.clear() + } startTs = 0L cacheTimeFilled.set(false) } diff --git a/common/src/main/java/com/pedro/common/base/BaseSender.kt b/common/src/main/java/com/pedro/common/base/BaseSender.kt index 99d3b1a95..1b6696185 100644 --- a/common/src/main/java/com/pedro/common/base/BaseSender.kt +++ b/common/src/main/java/com/pedro/common/base/BaseSender.kt @@ -2,8 +2,10 @@ package com.pedro.common.base import android.util.Log import com.pedro.common.BitrateManager +import com.pedro.common.BufferPool import com.pedro.common.ConnectChecker import com.pedro.common.StreamBlockingQueue +import com.pedro.common.clone import com.pedro.common.frame.MediaFrame import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers @@ -13,6 +15,7 @@ import kotlinx.coroutines.cancelAndJoin import kotlinx.coroutines.delay import kotlinx.coroutines.isActive import kotlinx.coroutines.launch +import kotlinx.coroutines.runInterruptible import java.nio.ByteBuffer import java.util.concurrent.atomic.AtomicLong @@ -24,7 +27,8 @@ abstract class BaseSender( @Volatile protected var running = false - protected val queue = StreamBlockingQueue(400) + private val queue = StreamBlockingQueue(400) + private val bufferPool = BufferPool() protected val audioFramesSent = AtomicLong(0) protected val videoFramesSent = AtomicLong(0) @@ -44,24 +48,54 @@ abstract class BaseSender( protected abstract suspend fun onRun() protected abstract suspend fun stopImp(clear: Boolean = true) - fun sendMediaFrame(mediaFrame: MediaFrame) { - if (running && !queue.trySend(mediaFrame)) { - when (mediaFrame.type) { - MediaFrame.Type.VIDEO -> { - Log.i(TAG, "Video frame discarded") - droppedVideoFrames.incrementAndGet() - } - MediaFrame.Type.AUDIO -> { - Log.i(TAG, "Audio frame discarded") - droppedAudioFrames.incrementAndGet() - } + fun sendMediaFrame(buffer: ByteBuffer, info: MediaFrame.Info, type: MediaFrame.Type) { + if (!running) return + val data = bufferPool.acquire(buffer.limit()) + val mediaFrame = MediaFrame(buffer.clone(data), info, type) + if (!queue.trySend(mediaFrame)) { + countDiscardedFrame(type) + recycleFrame(mediaFrame) + } + } + + /** + * Take the next frame from the queue, hand it to [consume] and recycle its buffer once + * consumed. Senders must read frames only through this method, the buffer is reused right + * after [consume] returns. + */ + protected suspend fun consumeFrame(consume: suspend (MediaFrame) -> Unit) { + val mediaFrame = runInterruptible { queue.take() } + try { + consume(mediaFrame) + } finally { + recycleFrame(mediaFrame) + } + } + + /** + * Every frame here was built by [sendMediaFrame], so its data is a slice of a pooled array + * and [java.nio.ByteBuffer.array] gives it back. + */ + private fun recycleFrame(mediaFrame: MediaFrame) { + if (mediaFrame.data.hasArray()) bufferPool.release(mediaFrame.data.array()) + } + + private fun countDiscardedFrame(type: MediaFrame.Type) { + when (type) { + MediaFrame.Type.VIDEO -> { + Log.i(TAG, "Video frame discarded") + droppedVideoFrames.incrementAndGet() + } + MediaFrame.Type.AUDIO -> { + Log.i(TAG, "Audio frame discarded") + droppedAudioFrames.incrementAndGet() } } } fun start() { bitrateManager.reset() - queue.clear() + queue.clear { recycleFrame(it) } running = true job = scope.launch { val bitrateTask = async { @@ -86,7 +120,8 @@ abstract class BaseSender( resetBytesSend() job?.cancelAndJoin() job = null - queue.clear() + queue.clear { recycleFrame(it) } + bufferPool.clear() } @Throws(IllegalArgumentException::class) @@ -110,7 +145,7 @@ abstract class BaseSender( fun getItemsInCache(): Int = queue.getSize() fun clearCache() { - queue.clear() + queue.clear { recycleFrame(it) } } fun getSentAudioFrames(): Long = audioFramesSent.get() diff --git a/common/src/test/java/com/pedro/common/BufferPoolTest.kt b/common/src/test/java/com/pedro/common/BufferPoolTest.kt new file mode 100644 index 000000000..30c77b18a --- /dev/null +++ b/common/src/test/java/com/pedro/common/BufferPoolTest.kt @@ -0,0 +1,116 @@ +package com.pedro.common + +import org.junit.Assert.assertArrayEquals +import org.junit.Assert.assertEquals +import org.junit.Assert.assertNotSame +import org.junit.Assert.assertSame +import org.junit.Assert.assertTrue +import org.junit.Test + +class BufferPoolTest { + + @Test + fun `GIVEN a size WHEN acquire THEN return a power of two array able to contain it`() { + val pool = BufferPool() + val sizes = listOf(0, 1, 500, 1024, 1025, 4096, 100000) + sizes.forEach { size -> + val buffer = pool.acquire(size) + assertTrue("$size does not fit in ${buffer.size}", buffer.size >= size) + assertEquals("${buffer.size} is not a power of two", 1, Integer.bitCount(buffer.size)) + assertTrue("${buffer.size} is smaller than the min size class", buffer.size >= 1024) + } + } + + @Test + fun `GIVEN a released buffer WHEN acquire the same size class THEN reuse the same instance`() { + val pool = BufferPool() + val buffer = pool.acquire(3000) + assertEquals(4096, buffer.size) + pool.release(buffer) + //any size inside the same class must get the very same array back + assertSame(buffer, pool.acquire(2049)) + } + + @Test + fun `GIVEN an empty pool WHEN acquire twice without releasing THEN return different instances`() { + val pool = BufferPool() + val first = pool.acquire(1024) + val second = pool.acquire(1024) + assertNotSame(first, second) + } + + @Test + fun `GIVEN a foreign array WHEN release THEN ignore it`() { + val pool = BufferPool() + //not a power of two so it was never created by acquire + val foreign = ByteArray(3000) + pool.release(foreign) + assertEquals(0, pool.getRetainedBytes()) + assertNotSame(foreign, pool.acquire(3000)) + } + + @Test + fun `GIVEN a size over the max class WHEN acquire and release THEN allocate exact and do not retain`() { + val pool = BufferPool() + val size = 8 * 1024 * 1024 + val buffer = pool.acquire(size) + assertEquals(size, buffer.size) + pool.release(buffer) + assertEquals(0, pool.getRetainedBytes()) + } + + @Test + fun `GIVEN more buffers than the limit WHEN release THEN retain only the allowed ones`() { + val pool = BufferPool(maxBuffersPerSizeClass = 2) + val buffers = List(5) { pool.acquire(1024) } + buffers.forEach { pool.release(it) } + assertEquals(2048, pool.getRetainedBytes()) + } + + @Test + fun `GIVEN a retained bytes limit WHEN release THEN stop retaining once reached`() { + val pool = BufferPool(maxRetainedBytes = 4096) + val buffers = List(8) { pool.acquire(1024) } + buffers.forEach { pool.release(it) } + assertEquals(4096, pool.getRetainedBytes()) + } + + @Test + fun `GIVEN the same buffer WHEN release twice THEN retain it only once`() { + val pool = BufferPool() + val buffer = pool.acquire(1024) + pool.release(buffer) + pool.release(buffer) + assertEquals(1024, pool.getRetainedBytes()) + //the second acquire must not get the very same array than the first one + assertSame(buffer, pool.acquire(1024)) + assertNotSame(buffer, pool.acquire(1024)) + } + + @Test + fun `GIVEN retained buffers WHEN clear THEN drop all of them`() { + val pool = BufferPool() + val buffer = pool.acquire(1024) + pool.release(buffer) + assertEquals(1024, pool.getRetainedBytes()) + pool.clear() + assertEquals(0, pool.getRetainedBytes()) + assertNotSame(buffer, pool.acquire(1024)) + } + + @Test + fun `GIVEN a dirty reused buffer WHEN write a shorter frame THEN only the frame bytes are read`() { + val pool = BufferPool() + val first = pool.acquire(1024) + //fill it with garbage the next user must not see + first.fill(0xFF.toByte()) + pool.release(first) + + val frame = byteArrayOf(1, 2, 3, 4, 5) + val reused = pool.acquire(frame.size) + assertSame(first, reused) + frame.copyInto(reused) + //a consumer limited to the frame size never observes the stale bytes + assertArrayEquals(frame, reused.copyOfRange(0, frame.size)) + } +} diff --git a/common/src/test/java/com/pedro/common/base/BaseSenderTest.kt b/common/src/test/java/com/pedro/common/base/BaseSenderTest.kt new file mode 100644 index 000000000..f9ee0c981 --- /dev/null +++ b/common/src/test/java/com/pedro/common/base/BaseSenderTest.kt @@ -0,0 +1,149 @@ +package com.pedro.common.base + +import com.pedro.common.ConnectChecker +import com.pedro.common.frame.MediaFrame +import com.pedro.common.removeInfo +import kotlinx.coroutines.runBlocking +import org.junit.Assert.assertArrayEquals +import org.junit.Assert.assertEquals +import org.junit.Assert.assertSame +import org.junit.Test +import org.mockito.Mockito +import java.nio.ByteBuffer + +/** + * The sender copies each frame into a pooled array bigger than the frame itself. These tests lock + * the shape of the resulting buffer so the packetizers keep behaving like they did with a copy. + */ +class BaseSenderTest { + + private class FakeSender: BaseSender(Mockito.mock(ConnectChecker::class.java), "FakeSender") { + override fun setVideoInfo(sps: ByteBuffer, pps: ByteBuffer?, vps: ByteBuffer?) {} + override fun setAudioInfo(sampleRate: Int, isStereo: Boolean) {} + override suspend fun onRun() {} + override suspend fun stopImp(clear: Boolean) {} + + //consumeFrame is the only way senders read frames, use it to inspect what was enqueued + suspend fun consumeOne(block: (MediaFrame) -> Unit) = consumeFrame { block(it) } + fun forceRunning() { running = true } + } + + private fun ByteBuffer.readAll(): ByteArray = ByteArray(remaining()).also { duplicate().get(it) } + + @Test + fun `GIVEN a frame WHEN send THEN the buffer capacity is the frame size not the pooled size`() = runBlocking { + val sender = FakeSender() + sender.forceRunning() + val frame = byteArrayOf(0, 0, 0, 1, 101, 1, 2, 3, 4, 5) + + sender.sendMediaFrame(ByteBuffer.wrap(frame), MediaFrame.Info(0, frame.size, 0, true), MediaFrame.Type.VIDEO) + + sender.consumeOne { mediaFrame -> + val data = mediaFrame.data + //the pooled array is 1024 bytes, nothing beyond the frame may be reachable + assertEquals(frame.size, data.capacity()) + assertEquals(0, data.position()) + assertEquals(frame.size, data.limit()) + assertArrayEquals(frame, data.readAll()) + } + } + + @Test + fun `GIVEN a frame WHEN removeInfo THEN return the same bytes than the source`() = runBlocking { + val sender = FakeSender() + sender.forceRunning() + val frame = byteArrayOf(0, 0, 0, 1, 101, 9, 8, 7) + + sender.sendMediaFrame(ByteBuffer.wrap(frame), MediaFrame.Info(0, frame.size, 0, true), MediaFrame.Type.VIDEO) + + sender.consumeOne { mediaFrame -> + val fixedBuffer = mediaFrame.data.removeInfo(mediaFrame.info) + assertArrayEquals(frame, fixedBuffer.readAll()) + } + } + + @Test + fun `GIVEN a buffer with offset WHEN removeInfo THEN the frame is not shifted twice`() = runBlocking { + val sender = FakeSender() + sender.forceRunning() + val prefix = byteArrayOf(0x55, 0x55, 0x55, 0x55) + val frame = byteArrayOf(0, 0, 0, 1, 101, 9, 8, 7) + //MediaCodec hands the buffer positioned at the offset reported in the info + val buffer = ByteBuffer.wrap(prefix.plus(frame)) + buffer.position(prefix.size) + + sender.sendMediaFrame(buffer, MediaFrame.Info(prefix.size, frame.size, 0, true), MediaFrame.Type.VIDEO) + + sender.consumeOne { mediaFrame -> + //the copy keeps the absolute offsets, so the whole buffer is still there + assertEquals(prefix.size + frame.size, mediaFrame.data.capacity()) + assertEquals(0, mediaFrame.data.position()) + //removeInfo skips the prefix once, not twice + val fixedBuffer = mediaFrame.data.removeInfo(mediaFrame.info) + assertArrayEquals(frame, fixedBuffer.readAll()) + } + } + + @Test + fun `GIVEN a buffer with offset WHEN send THEN produce the same bytes than a plain copy`() = runBlocking { + val sender = FakeSender() + sender.forceRunning() + val source = byteArrayOf(0x55, 0x55, 0x55, 0x55, 0, 0, 0, 1, 101, 9, 8, 7) + val buffer = ByteBuffer.wrap(source) + buffer.position(4) + + sender.sendMediaFrame(buffer, MediaFrame.Info(4, source.size - 4, 0, true), MediaFrame.Type.VIDEO) + + sender.consumeOne { mediaFrame -> + //the old path was ByteBuffer.wrap(toByteArray()), copying from absolute 0 + val expected = ByteBuffer.wrap(source.copyOf()) + val pooled = mediaFrame.data + assertEquals(expected.capacity(), pooled.capacity()) + assertEquals(expected.position(), pooled.position()) + assertEquals(expected.limit(), pooled.limit()) + assertArrayEquals(source, pooled.readAll()) + } + } + + @Test + fun `GIVEN a recycled dirty buffer WHEN send a smaller frame THEN stale bytes are unreachable`() = runBlocking { + val sender = FakeSender() + sender.forceRunning() + //first frame fills a 1024 pooled array with a byte we must never see again + val big = ByteArray(600) { 0xFF.toByte() } + sender.sendMediaFrame(ByteBuffer.wrap(big), MediaFrame.Info(0, big.size, 0, true), MediaFrame.Type.VIDEO) + var firstArray: ByteArray? = null + //consuming it recycles the pooled array + sender.consumeOne { firstArray = it.data.array() } + + val small = byteArrayOf(1, 2, 3) + sender.sendMediaFrame(ByteBuffer.wrap(small), MediaFrame.Info(0, small.size, 0, false), MediaFrame.Type.VIDEO) + + sender.consumeOne { mediaFrame -> + //same underlying array reused, but capped to the new frame + assertSame(firstArray, mediaFrame.data.array()) + assertEquals(small.size, mediaFrame.data.capacity()) + assertArrayEquals(small, mediaFrame.data.readAll()) + } + } + + @Test + fun `GIVEN an info bigger than the frame WHEN removeInfo THEN do not expose stale bytes`() = runBlocking { + val sender = FakeSender() + sender.forceRunning() + val big = ByteArray(600) { 0xFF.toByte() } + sender.sendMediaFrame(ByteBuffer.wrap(big), MediaFrame.Info(0, big.size, 0, true), MediaFrame.Type.VIDEO) + sender.consumeOne { } + + val small = byteArrayOf(1, 2, 3) + //a malformed info asking for more bytes than the frame has + sender.sendMediaFrame(ByteBuffer.wrap(small), MediaFrame.Info(0, 500, 0, false), MediaFrame.Type.VIDEO) + + sender.consumeOne { mediaFrame -> + val fixedBuffer = mediaFrame.data.removeInfo(mediaFrame.info) + //truncated to the real frame instead of reading into the previous one + assertEquals(small.size, fixedBuffer.remaining()) + assertArrayEquals(small, fixedBuffer.readAll()) + } + } +} diff --git a/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt b/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt index bcbf53c4d..6dd90d553 100644 --- a/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt +++ b/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpClient.kt @@ -24,7 +24,6 @@ import com.pedro.common.ConnectionFailed import com.pedro.common.TimeUtils import com.pedro.common.UrlParser import com.pedro.common.VideoCodec -import com.pedro.common.clone import com.pedro.common.frame.MediaFrame import com.pedro.common.onMainThread import com.pedro.common.socket.base.SocketType @@ -599,13 +598,13 @@ class RtmpClient(private val connectChecker: ConnectChecker) { fun sendVideo(videoBuffer: ByteBuffer, info: MediaCodec.BufferInfo) { if (!commandsManager.videoDisabled) { - rtmpSender.sendMediaFrame(MediaFrame(videoBuffer.clone(), info.toMediaFrameInfo(), MediaFrame.Type.VIDEO)) + rtmpSender.sendMediaFrame(videoBuffer, info.toMediaFrameInfo(), MediaFrame.Type.VIDEO) } } fun sendAudio(audioBuffer: ByteBuffer, info: MediaCodec.BufferInfo) { if (!commandsManager.audioDisabled) { - rtmpSender.sendMediaFrame(MediaFrame(audioBuffer.clone(), info.toMediaFrameInfo(), MediaFrame.Type.AUDIO)) + rtmpSender.sendMediaFrame(audioBuffer, info.toMediaFrameInfo(), MediaFrame.Type.AUDIO) } } diff --git a/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpSender.kt b/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpSender.kt index 16303ba5a..5e942dd46 100644 --- a/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpSender.kt +++ b/rtmp/src/main/java/com/pedro/rtmp/rtmp/RtmpSender.kt @@ -37,7 +37,6 @@ import com.pedro.rtmp.flv.video.packet.Vp8Packet import com.pedro.rtmp.flv.video.packet.Vp9Packet import com.pedro.rtmp.utils.socket.RtmpSocket import kotlinx.coroutines.isActive -import kotlinx.coroutines.runInterruptible import java.nio.ByteBuffer /** @@ -81,28 +80,29 @@ class RtmpSender( override suspend fun onRun() { while (scope.isActive && running) { val error = runCatching { - val mediaFrame = runInterruptible { queue.take() } - getFlvPacket(mediaFrame) { flvPacket -> - var size = 0L - if (flvPacket.type == FlvType.VIDEO) { - videoFramesSent.incrementAndGet() - socket?.let { socket -> - size = commandsManager.sendVideoPacket(flvPacket, socket).toLong() - if (isEnableLogs) { - Log.i(TAG, "wrote Video packet, size $size") + consumeFrame { mediaFrame -> + getFlvPacket(mediaFrame) { flvPacket -> + var size = 0L + if (flvPacket.type == FlvType.VIDEO) { + videoFramesSent.incrementAndGet() + socket?.let { socket -> + size = commandsManager.sendVideoPacket(flvPacket, socket).toLong() + if (isEnableLogs) { + Log.i(TAG, "wrote Video packet, size $size") + } } - } - } else { - audioFramesSent.incrementAndGet() - socket?.let { socket -> - size = commandsManager.sendAudioPacket(flvPacket, socket).toLong() - if (isEnableLogs) { - Log.i(TAG, "wrote Audio packet, size $size") + } else { + audioFramesSent.incrementAndGet() + socket?.let { socket -> + size = commandsManager.sendAudioPacket(flvPacket, socket).toLong() + if (isEnableLogs) { + Log.i(TAG, "wrote Audio packet, size $size") + } } } + bytesSend.addAndGet(size) + bytesSendPerSecond.addAndGet(size) } - bytesSend.addAndGet(size) - bytesSendPerSecond.addAndGet(size) } }.exceptionOrNull() if (error != null) { diff --git a/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspClient.kt b/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspClient.kt index 291104a9d..d5c3faf29 100644 --- a/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspClient.kt +++ b/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspClient.kt @@ -23,7 +23,6 @@ import com.pedro.common.ConnectChecker import com.pedro.common.ConnectionFailed import com.pedro.common.UrlParser import com.pedro.common.VideoCodec -import com.pedro.common.clone import com.pedro.common.frame.MediaFrame import com.pedro.common.onMainThread import com.pedro.common.socket.base.SocketType @@ -464,13 +463,13 @@ class RtspClient(private val connectChecker: ConnectChecker) { fun sendVideo(videoBuffer: ByteBuffer, info: MediaCodec.BufferInfo) { if (!commandsManager.videoDisabled) { - rtspSender.sendMediaFrame(MediaFrame(videoBuffer.clone(), info.toMediaFrameInfo(), MediaFrame.Type.VIDEO)) + rtspSender.sendMediaFrame(videoBuffer, info.toMediaFrameInfo(), MediaFrame.Type.VIDEO) } } fun sendAudio(audioBuffer: ByteBuffer, info: MediaCodec.BufferInfo) { if (!commandsManager.audioDisabled) { - rtspSender.sendMediaFrame(MediaFrame(audioBuffer.clone(), info.toMediaFrameInfo(), MediaFrame.Type.AUDIO)) + rtspSender.sendMediaFrame(audioBuffer, info.toMediaFrameInfo(), MediaFrame.Type.AUDIO) } } diff --git a/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspSender.kt b/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspSender.kt index 16b4bbe0d..6507afa79 100644 --- a/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspSender.kt +++ b/rtsp/src/main/java/com/pedro/rtsp/rtsp/RtspSender.kt @@ -41,7 +41,6 @@ import com.pedro.rtsp.rtp.sockets.RtpSocketTcp import com.pedro.rtsp.rtsp.commands.CommandsManager import com.pedro.rtsp.utils.RtpConstants import kotlinx.coroutines.isActive -import kotlinx.coroutines.runInterruptible import java.io.IOException import java.nio.ByteBuffer import java.util.Random @@ -109,32 +108,33 @@ class RtspSender( val isTcp = rtpSocket is RtpSocketTcp while (scope.isActive && running) { val error = runCatching { - val mediaFrame = runInterruptible { queue.take() } - getRtpPackets(mediaFrame) { rtpFrames -> - var size = 0L - var isVideo = false - rtpFrames.forEach { rtpFrame -> - rtpSocket?.sendFrame(rtpFrame) - //4 is tcp header length - val packetSize = (if (isTcp) rtpFrame.length + 4 else rtpFrame.length).toLong() - bytesSend.addAndGet(packetSize) - bytesSendPerSecond.addAndGet(packetSize) - size += packetSize - isVideo = rtpFrame.isVideoFrame(commandsManager.rtpTracks.trackVideo) - if (isVideo) videoFramesSent.incrementAndGet() - else audioFramesSent.incrementAndGet() - if (baseSenderReport?.update(rtpFrame) == true) { + consumeFrame { mediaFrame -> + getRtpPackets(mediaFrame) { rtpFrames -> + var size = 0L + var isVideo = false + rtpFrames.forEach { rtpFrame -> + rtpSocket?.sendFrame(rtpFrame) //4 is tcp header length - val reportSize = (if (isTcp) RtpConstants.REPORT_PACKET_LENGTH + 4 else RtpConstants.REPORT_PACKET_LENGTH).toLong() - bytesSend.addAndGet(reportSize) - bytesSendPerSecond.addAndGet(reportSize) - if (isEnableLogs) Log.i(TAG, "wrote report") + val packetSize = (if (isTcp) rtpFrame.length + 4 else rtpFrame.length).toLong() + bytesSend.addAndGet(packetSize) + bytesSendPerSecond.addAndGet(packetSize) + size += packetSize + isVideo = rtpFrame.isVideoFrame(commandsManager.rtpTracks.trackVideo) + if (isVideo) videoFramesSent.incrementAndGet() + else audioFramesSent.incrementAndGet() + if (baseSenderReport?.update(rtpFrame) == true) { + //4 is tcp header length + val reportSize = (if (isTcp) RtpConstants.REPORT_PACKET_LENGTH + 4 else RtpConstants.REPORT_PACKET_LENGTH).toLong() + bytesSend.addAndGet(reportSize) + bytesSendPerSecond.addAndGet(reportSize) + if (isEnableLogs) Log.i(TAG, "wrote report") + } + } + rtpSocket?.flush() + if (isEnableLogs) { + val type = if (isVideo) "Video" else "Audio" + Log.i(TAG, "wrote $type packet, size $size") } - } - rtpSocket?.flush() - if (isEnableLogs) { - val type = if (isVideo) "Video" else "Audio" - Log.i(TAG, "wrote $type packet, size $size") } } }.exceptionOrNull() 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 ffffb8570..9af1938ed 100644 --- a/srt/src/main/java/com/pedro/srt/srt/SrtClient.kt +++ b/srt/src/main/java/com/pedro/srt/srt/SrtClient.kt @@ -23,7 +23,6 @@ import com.pedro.common.ConnectChecker import com.pedro.common.ConnectionFailed import com.pedro.common.UrlParser import com.pedro.common.VideoCodec -import com.pedro.common.clone import com.pedro.common.frame.MediaFrame import com.pedro.common.onMainThread import com.pedro.common.socket.base.SocketType @@ -428,13 +427,13 @@ class SrtClient(private val connectChecker: ConnectChecker) { fun sendVideo(videoBuffer: ByteBuffer, info: MediaCodec.BufferInfo) { if (!commandsManager.videoDisabled) { - srtSender.sendMediaFrame(MediaFrame(videoBuffer.clone(), info.toMediaFrameInfo(), MediaFrame.Type.VIDEO)) + srtSender.sendMediaFrame(videoBuffer, info.toMediaFrameInfo(), MediaFrame.Type.VIDEO) } } fun sendAudio(audioBuffer: ByteBuffer, info: MediaCodec.BufferInfo) { if (!commandsManager.audioDisabled) { - srtSender.sendMediaFrame(MediaFrame(audioBuffer.clone(), info.toMediaFrameInfo(), MediaFrame.Type.AUDIO)) + srtSender.sendMediaFrame(audioBuffer, info.toMediaFrameInfo(), MediaFrame.Type.AUDIO) } } diff --git a/srt/src/main/java/com/pedro/srt/srt/SrtSender.kt b/srt/src/main/java/com/pedro/srt/srt/SrtSender.kt index a430e83b9..ce6907e68 100644 --- a/srt/src/main/java/com/pedro/srt/srt/SrtSender.kt +++ b/srt/src/main/java/com/pedro/srt/srt/SrtSender.kt @@ -41,7 +41,6 @@ import com.pedro.srt.utils.SrtSocket import com.pedro.srt.utils.chunkPackets import com.pedro.srt.utils.toCodec import kotlinx.coroutines.isActive -import kotlinx.coroutines.runInterruptible import java.nio.ByteBuffer /** @@ -143,14 +142,15 @@ class SrtSender( sendPackets(psiPacketsConfig, MpegType.PSI) while (scope.isActive && running) { val error = runCatching { - val mediaFrame = runInterruptible { queue.take() } - getMpegTsPackets(mediaFrame) { mpegTsPackets -> - val isKey = mpegTsPackets[0].isKey - val psiPackets = psiManager.checkSendInfo(isKey, mpegTsPacketizer, chunkSize) - val bytesPsi = sendPackets(psiPackets, MpegType.PSI) - val bytes = sendPackets(mpegTsPackets, mpegTsPackets[0].type) - bytesSend.addAndGet(bytesPsi + bytes) - bytesSendPerSecond.addAndGet(bytesPsi + bytes) + consumeFrame { mediaFrame -> + getMpegTsPackets(mediaFrame) { mpegTsPackets -> + val isKey = mpegTsPackets[0].isKey + val psiPackets = psiManager.checkSendInfo(isKey, mpegTsPacketizer, chunkSize) + val bytesPsi = sendPackets(psiPackets, MpegType.PSI) + val bytes = sendPackets(mpegTsPackets, mpegTsPackets[0].type) + bytesSend.addAndGet(bytesPsi + bytes) + bytesSendPerSecond.addAndGet(bytesPsi + bytes) + } } }.exceptionOrNull() if (error != null) { diff --git a/srt/src/test/java/com/pedro/srt/srt/SrtSenderTest.kt b/srt/src/test/java/com/pedro/srt/srt/SrtSenderTest.kt index b27b5998d..bdb9e5b81 100644 --- a/srt/src/test/java/com/pedro/srt/srt/SrtSenderTest.kt +++ b/srt/src/test/java/com/pedro/srt/srt/SrtSenderTest.kt @@ -62,10 +62,32 @@ class SrtSenderTest { val videoData = ByteBuffer.wrap(header.plus(ByteArray(300) { 0x00 })) val audioData = ByteBuffer.wrap(ByteArray(256) { 0x00 }) - val videoFrame = MediaFrame(videoData, MediaFrame.Info(0, videoData.remaining(), 0, true), MediaFrame.Type.VIDEO) - val audioFrame = MediaFrame(audioData, MediaFrame.Info(0, audioData.remaining(), 0, false), MediaFrame.Type.AUDIO) - srtSender.sendMediaFrame(videoFrame) - srtSender.sendMediaFrame(audioFrame) + srtSender.sendMediaFrame(videoData, MediaFrame.Info(0, videoData.remaining(), 0, true), MediaFrame.Type.VIDEO) + srtSender.sendMediaFrame(audioData, MediaFrame.Info(0, audioData.remaining(), 0, false), MediaFrame.Type.AUDIO) + latch.await(1000, TimeUnit.MILLISECONDS) + srtSender.stop() + + assertEquals(1692, output.toByteArray().size) + } + + @Test + fun `GIVEN pooled mediaFrames WHEN send to sender THEN write the same packets than a copy`() = runTest { + latch = CountDownLatch(7) //writeData must be called 4 times + val srtSender = SrtSender(connectChecker, commandsManager) + srtSender.setAudioInfo(44100, true) + val sps = ByteBuffer.wrap(byteArrayOf(0, 0, 0, 1, 103, 100, 0, 30, -84, -76, 15, 2, -115, 53, 2, 2, 2, 7, -117, 23, 8)) + val pps = ByteBuffer.wrap(byteArrayOf(0, 0, 0, 1, 104, -18, 13, -117)) + srtSender.setVideoInfo(sps, pps, null) + srtSender.socket = socket + srtSender.start() + + val header = byteArrayOf(0x00, 0x00, 0x00, 0x01, 0x05) + val videoData = ByteBuffer.wrap(header.plus(ByteArray(300) { 0x00 })) + val audioData = ByteBuffer.wrap(ByteArray(256) { 0x00 }) + + //the sender copies into a pooled buffer bigger than the frame, the output must not change + srtSender.sendMediaFrame(videoData, MediaFrame.Info(0, videoData.remaining(), 0, true), MediaFrame.Type.VIDEO) + srtSender.sendMediaFrame(audioData, MediaFrame.Info(0, audioData.remaining(), 0, false), MediaFrame.Type.AUDIO) latch.await(1000, TimeUnit.MILLISECONDS) srtSender.stop() diff --git a/udp/src/main/java/com/pedro/udp/UdpClient.kt b/udp/src/main/java/com/pedro/udp/UdpClient.kt index f6d3be602..d9151f62b 100644 --- a/udp/src/main/java/com/pedro/udp/UdpClient.kt +++ b/udp/src/main/java/com/pedro/udp/UdpClient.kt @@ -245,13 +245,13 @@ class UdpClient(private val connectChecker: ConnectChecker) { fun sendVideo(videoBuffer: ByteBuffer, info: MediaCodec.BufferInfo) { if (!commandManager.videoDisabled) { - udpSender.sendMediaFrame(MediaFrame(videoBuffer.clone(), info.toMediaFrameInfo(), MediaFrame.Type.VIDEO)) + udpSender.sendMediaFrame(videoBuffer, info.toMediaFrameInfo(), MediaFrame.Type.VIDEO) } } fun sendAudio(audioBuffer: ByteBuffer, info: MediaCodec.BufferInfo) { if (!commandManager.audioDisabled) { - udpSender.sendMediaFrame(MediaFrame(audioBuffer.clone(), info.toMediaFrameInfo(), MediaFrame.Type.AUDIO)) + udpSender.sendMediaFrame(audioBuffer, info.toMediaFrameInfo(), MediaFrame.Type.AUDIO) } } diff --git a/udp/src/main/java/com/pedro/udp/UdpSender.kt b/udp/src/main/java/com/pedro/udp/UdpSender.kt index d333389c2..c28b19a00 100644 --- a/udp/src/main/java/com/pedro/udp/UdpSender.kt +++ b/udp/src/main/java/com/pedro/udp/UdpSender.kt @@ -125,14 +125,15 @@ class UdpSender( sendPackets(psiPacketsConfig, MpegType.PSI) while (scope.isActive && running) { val error = runCatching { - val mediaFrame = runInterruptible { queue.take() } - getMpegTsPackets(mediaFrame) { mpegTsPackets -> - val isKey = mpegTsPackets[0].isKey - val psiPackets = psiManager.checkSendInfo(isKey, mpegTsPacketizer, chunkSize) - val bytesPsi = sendPackets(psiPackets, MpegType.PSI) - val bytes = sendPackets(mpegTsPackets, mpegTsPackets[0].type) - bytesSend.addAndGet(bytesPsi + bytes) - bytesSendPerSecond.addAndGet(bytesPsi + bytes) + consumeFrame { mediaFrame -> + getMpegTsPackets(mediaFrame) { mpegTsPackets -> + val isKey = mpegTsPackets[0].isKey + val psiPackets = psiManager.checkSendInfo(isKey, mpegTsPacketizer, chunkSize) + val bytesPsi = sendPackets(psiPackets, MpegType.PSI) + val bytes = sendPackets(mpegTsPackets, mpegTsPackets[0].type) + bytesSend.addAndGet(bytesPsi + bytes) + bytesSendPerSecond.addAndGet(bytesPsi + bytes) + } } }.exceptionOrNull() if (error != null) { diff --git a/whip/src/main/java/com/pedro/whip/WhipClient.kt b/whip/src/main/java/com/pedro/whip/WhipClient.kt index 48a1fa4a2..9b0b97a5d 100644 --- a/whip/src/main/java/com/pedro/whip/WhipClient.kt +++ b/whip/src/main/java/com/pedro/whip/WhipClient.kt @@ -6,7 +6,6 @@ import com.pedro.common.AudioCodec import com.pedro.common.ConnectChecker import com.pedro.common.UrlParser import com.pedro.common.VideoCodec -import com.pedro.common.clone import com.pedro.common.frame.MediaFrame import com.pedro.common.onMainThread import com.pedro.common.socket.base.SocketType @@ -449,13 +448,13 @@ class WhipClient(private val connectChecker: ConnectChecker) { fun sendVideo(videoBuffer: ByteBuffer, info: MediaCodec.BufferInfo) { if (!commandsManager.videoDisabled) { - whipSender.sendMediaFrame(MediaFrame(videoBuffer.clone(), info.toMediaFrameInfo(), MediaFrame.Type.VIDEO)) + whipSender.sendMediaFrame(videoBuffer, info.toMediaFrameInfo(), MediaFrame.Type.VIDEO) } } fun sendAudio(audioBuffer: ByteBuffer, info: MediaCodec.BufferInfo) { if (!commandsManager.audioDisabled) { - whipSender.sendMediaFrame(MediaFrame(audioBuffer.clone(), info.toMediaFrameInfo(), MediaFrame.Type.AUDIO)) + whipSender.sendMediaFrame(audioBuffer, info.toMediaFrameInfo(), MediaFrame.Type.AUDIO) } } diff --git a/whip/src/main/java/com/pedro/whip/WhipSender.kt b/whip/src/main/java/com/pedro/whip/WhipSender.kt index e3c325d16..8eef0ca2f 100644 --- a/whip/src/main/java/com/pedro/whip/WhipSender.kt +++ b/whip/src/main/java/com/pedro/whip/WhipSender.kt @@ -26,7 +26,6 @@ import com.pedro.rtsp.utils.CryptoProperties import com.pedro.rtsp.utils.RtpConstants import com.pedro.whip.webrtc.CommandsManager import kotlinx.coroutines.isActive -import kotlinx.coroutines.runInterruptible import java.io.IOException import java.nio.ByteBuffer @@ -85,32 +84,33 @@ class WhipSender( val isTcp = rtpSocket is RtpSocketTcp while (scope.isActive && running) { val error = runCatching { - val mediaFrame = runInterruptible { queue.take() } - getRtpPackets(mediaFrame) { rtpFrames -> - var size = 0L - var isVideo = false - rtpFrames.forEach { rtpFrame -> - rtpSocket?.sendFrame(rtpFrame) - //4 is tcp header length - val packetSize = (if (isTcp) rtpFrame.length + 4 else rtpFrame.length).toLong() - bytesSend.addAndGet(packetSize) - bytesSendPerSecond.addAndGet(packetSize) - size += packetSize - isVideo = rtpFrame.isVideoFrame(commandsManager.rtpTracks.trackVideo) - if (isVideo) videoFramesSent.incrementAndGet() - else audioFramesSent.incrementAndGet() - if (baseSenderReport?.update(rtpFrame) == true) { + consumeFrame { mediaFrame -> + getRtpPackets(mediaFrame) { rtpFrames -> + var size = 0L + var isVideo = false + rtpFrames.forEach { rtpFrame -> + rtpSocket?.sendFrame(rtpFrame) //4 is tcp header length - val reportSize = (if (isTcp) RtpConstants.REPORT_PACKET_LENGTH + 4 else RtpConstants.REPORT_PACKET_LENGTH).toLong() - bytesSend.addAndGet(reportSize) - bytesSendPerSecond.addAndGet(reportSize) - if (isEnableLogs) Log.i(TAG, "wrote report") + val packetSize = (if (isTcp) rtpFrame.length + 4 else rtpFrame.length).toLong() + bytesSend.addAndGet(packetSize) + bytesSendPerSecond.addAndGet(packetSize) + size += packetSize + isVideo = rtpFrame.isVideoFrame(commandsManager.rtpTracks.trackVideo) + if (isVideo) videoFramesSent.incrementAndGet() + else audioFramesSent.incrementAndGet() + if (baseSenderReport?.update(rtpFrame) == true) { + //4 is tcp header length + val reportSize = (if (isTcp) RtpConstants.REPORT_PACKET_LENGTH + 4 else RtpConstants.REPORT_PACKET_LENGTH).toLong() + bytesSend.addAndGet(reportSize) + bytesSendPerSecond.addAndGet(reportSize) + if (isEnableLogs) Log.i(TAG, "wrote report") + } + } + rtpSocket?.flush() + if (isEnableLogs) { + val type = if (isVideo) "Video" else "Audio" + Log.i(TAG, "wrote $type packet, size $size") } - } - rtpSocket?.flush() - if (isEnableLogs) { - val type = if (isVideo) "Video" else "Audio" - Log.i(TAG, "wrote $type packet, size $size") } } }.exceptionOrNull() From 0cf2ec988ce65a8fe108ea16a343967d389613bb Mon Sep 17 00:00:00 2001 From: pedroSG94 Date: Wed, 29 Jul 2026 21:17:41 +0200 Subject: [PATCH 2/2] adding poolbuffer to record --- .../main/java/com/pedro/common/BufferPool.kt | 5 +- .../java/com/pedro/common/base/BaseSender.kt | 42 +++---- library/build.gradle.kts | 2 + .../recording/AsyncBaseRecordController.kt | 15 ++- .../AsyncBaseRecordControllerTest.kt | 109 ++++++++++++++++++ 5 files changed, 143 insertions(+), 30 deletions(-) create mode 100644 library/src/test/java/com/pedro/library/base/recording/AsyncBaseRecordControllerTest.kt diff --git a/common/src/main/java/com/pedro/common/BufferPool.kt b/common/src/main/java/com/pedro/common/BufferPool.kt index 42cbe5f89..395315db4 100644 --- a/common/src/main/java/com/pedro/common/BufferPool.kt +++ b/common/src/main/java/com/pedro/common/BufferPool.kt @@ -63,6 +63,10 @@ class BufferPool( return ByteArray(1 shl sizeClass) } + fun release(buffer: java.nio.ByteBuffer) { + if (buffer.hasArray()) release(buffer.array()) + } + /** * Give an array back to the pool. Arrays not created by [acquire], and arrays already released, * are ignored, so the same array can never be handed out to two users at once. @@ -93,7 +97,6 @@ class BufferPool( fun getRetainedBytes(): Long = synchronized(lock) { retainedBytes } - //ceil(log2(size)) clamped to the smallest size class private fun sizeClassOf(size: Int): Int { val value = maxOf(size, 1 shl MIN_SIZE_CLASS) return 32 - Integer.numberOfLeadingZeros(value - 1) diff --git a/common/src/main/java/com/pedro/common/base/BaseSender.kt b/common/src/main/java/com/pedro/common/base/BaseSender.kt index 1b6696185..2f2f64d18 100644 --- a/common/src/main/java/com/pedro/common/base/BaseSender.kt +++ b/common/src/main/java/com/pedro/common/base/BaseSender.kt @@ -53,8 +53,17 @@ abstract class BaseSender( val data = bufferPool.acquire(buffer.limit()) val mediaFrame = MediaFrame(buffer.clone(data), info, type) if (!queue.trySend(mediaFrame)) { - countDiscardedFrame(type) - recycleFrame(mediaFrame) + bufferPool.release(mediaFrame.data) + when (type) { + MediaFrame.Type.VIDEO -> { + Log.i(TAG, "Video frame discarded") + droppedVideoFrames.incrementAndGet() + } + MediaFrame.Type.AUDIO -> { + Log.i(TAG, "Audio frame discarded") + droppedAudioFrames.incrementAndGet() + } + } } } @@ -68,34 +77,13 @@ abstract class BaseSender( try { consume(mediaFrame) } finally { - recycleFrame(mediaFrame) - } - } - - /** - * Every frame here was built by [sendMediaFrame], so its data is a slice of a pooled array - * and [java.nio.ByteBuffer.array] gives it back. - */ - private fun recycleFrame(mediaFrame: MediaFrame) { - if (mediaFrame.data.hasArray()) bufferPool.release(mediaFrame.data.array()) - } - - private fun countDiscardedFrame(type: MediaFrame.Type) { - when (type) { - MediaFrame.Type.VIDEO -> { - Log.i(TAG, "Video frame discarded") - droppedVideoFrames.incrementAndGet() - } - MediaFrame.Type.AUDIO -> { - Log.i(TAG, "Audio frame discarded") - droppedAudioFrames.incrementAndGet() - } + bufferPool.release(mediaFrame.data) } } fun start() { bitrateManager.reset() - queue.clear { recycleFrame(it) } + queue.clear { bufferPool.release(it.data) } running = true job = scope.launch { val bitrateTask = async { @@ -120,7 +108,7 @@ abstract class BaseSender( resetBytesSend() job?.cancelAndJoin() job = null - queue.clear { recycleFrame(it) } + queue.clear { bufferPool.release(it.data) } bufferPool.clear() } @@ -145,7 +133,7 @@ abstract class BaseSender( fun getItemsInCache(): Int = queue.getSize() fun clearCache() { - queue.clear { recycleFrame(it) } + queue.clear { bufferPool.release(it.data) } } fun getSentAudioFrames(): Long = audioFramesSent.get() diff --git a/library/build.gradle.kts b/library/build.gradle.kts index aa61faa8c..ba51498fa 100644 --- a/library/build.gradle.kts +++ b/library/build.gradle.kts @@ -47,4 +47,6 @@ dependencies { api(project(":udp")) api(project(":whip")) api(project(":common")) + testImplementation(libs.kotlinx.coroutines.test) + testImplementation(libs.junit) } diff --git a/library/src/main/java/com/pedro/library/base/recording/AsyncBaseRecordController.kt b/library/src/main/java/com/pedro/library/base/recording/AsyncBaseRecordController.kt index dcb549f52..229a9b2a4 100644 --- a/library/src/main/java/com/pedro/library/base/recording/AsyncBaseRecordController.kt +++ b/library/src/main/java/com/pedro/library/base/recording/AsyncBaseRecordController.kt @@ -18,6 +18,7 @@ package com.pedro.library.base.recording import android.media.MediaCodec import com.pedro.common.AudioCodec import com.pedro.common.BitrateManager +import com.pedro.common.BufferPool import com.pedro.common.TimeUtils.getCurrentTimeMicro import com.pedro.common.VideoCodec import com.pedro.common.clone @@ -62,6 +63,7 @@ abstract class AsyncBaseRecordController : RecordController { private val scope = CoroutineScope(Dispatchers.IO) private var muxerChannel: Channel? = null private var muxerJob: Job? = null + private val bufferPool = BufferPool() override fun setRequestKeyFrame(requestKeyFrame: RequestKeyFrame?) { this.myRequestKeyFrame = requestKeyFrame @@ -135,9 +137,11 @@ abstract class AsyncBaseRecordController : RecordController { private fun sendFrame(buffer: ByteBuffer, info: MediaCodec.BufferInfo, type: MediaFrame.Type) { if (recordStatus == RecordController.Status.STOPPED) return + val channel = muxerChannel ?: return val frameInfo = info.toMediaFrameInfo() val i = updateFormat(frameInfo) - muxerChannel?.trySend(MediaFrame(buffer.clone(), i, type)) + val frame = MediaFrame(buffer.clone(bufferPool.acquire(buffer.limit())), i, type) + if (channel.trySend(frame).isFailure) bufferPool.release(frame.data) } override fun startRecord( @@ -176,7 +180,13 @@ abstract class AsyncBaseRecordController : RecordController { muxerChannel = Channel(CAPACITY) muxerJob = scope.launch { val channel = muxerChannel ?: return@launch - for (frame in channel) onWriteFrame(frame) + for (frame in channel) { + try { + onWriteFrame(frame) + } finally { + bufferPool.release(frame.data) + } + } } this.tracks = tracks this.listener = listener @@ -194,6 +204,7 @@ abstract class AsyncBaseRecordController : RecordController { muxerChannel = null muxerJob?.cancel() runBlocking { muxerJob?.join() } + bufferPool.clear() recordStatus = RecordController.Status.STOPPED clearTimestamp() myRequestKeyFrame = null diff --git a/library/src/test/java/com/pedro/library/base/recording/AsyncBaseRecordControllerTest.kt b/library/src/test/java/com/pedro/library/base/recording/AsyncBaseRecordControllerTest.kt new file mode 100644 index 000000000..1a02fb10b --- /dev/null +++ b/library/src/test/java/com/pedro/library/base/recording/AsyncBaseRecordControllerTest.kt @@ -0,0 +1,109 @@ +package com.pedro.library.base.recording + +import android.media.MediaCodec +import android.media.MediaFormat +import com.pedro.common.frame.MediaFrame +import org.junit.Assert.assertArrayEquals +import org.junit.Assert.assertEquals +import org.junit.Assert.assertSame +import org.junit.Test +import java.io.FileDescriptor +import java.nio.ByteBuffer +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit + +/** + * The controller copies each frame into a pooled buffer before handing it to the muxer. These + * tests lock the shape of that buffer and that it goes back to the pool once written. + */ +class AsyncBaseRecordControllerTest { + + private class FakeRecordController: AsyncBaseRecordController() { + val written = mutableListOf>() + var latch = CountDownLatch(1) + + override fun startRecordImp(fd: FileDescriptor, listener: RecordController.Listener?, tracks: RecordController.RecordTracks) {} + override fun startRecordImp(path: String, listener: RecordController.Listener?, tracks: RecordController.RecordTracks) {} + override fun stopRecordImp() {} + override fun setVideoFormat(videoFormat: MediaFormat) {} + override fun setAudioFormat(audioFormat: MediaFormat) {} + override fun resetFormats() {} + + override suspend fun onWriteFrame(frame: MediaFrame) { + //keep the array identity and the capacity, the buffer itself is recycled right after + written.add(frame.data.array() to frame.data.capacity()) + latch.countDown() + } + + fun start() { + recordStatus = RecordController.Status.RECORDING + startRecord("fake", null, RecordController.RecordTracks.ALL) + recordStatus = RecordController.Status.RECORDING + } + + fun awaitFrames(count: Int) { + latch.await(1000, TimeUnit.MILLISECONDS) + latch = CountDownLatch(count) + } + } + + private fun bufferInfoOf(size: Int) = MediaCodec.BufferInfo().apply { + offset = 0 + this.size = size + presentationTimeUs = 0 + flags = MediaCodec.BUFFER_FLAG_KEY_FRAME + } + + @Test + fun `GIVEN a frame WHEN record THEN the muxer gets a buffer capped to the frame size`() { + val controller = FakeRecordController() + controller.latch = CountDownLatch(1) + controller.start() + + val frame = ByteArray(300) { it.toByte() } + controller.recordVideo(ByteBuffer.wrap(frame), bufferInfoOf(frame.size)) + controller.awaitFrames(1) + controller.stopRecord() + + assertEquals(1, controller.written.size) + //the pooled array is 1024 bytes, the muxer must only reach the frame + assertEquals(frame.size, controller.written[0].second) + } + + @Test + fun `GIVEN two frames WHEN record THEN the second reuses the array of the first`() { + val controller = FakeRecordController() + controller.latch = CountDownLatch(1) + controller.start() + + val first = ByteArray(600) { 0xFF.toByte() } + controller.recordVideo(ByteBuffer.wrap(first), bufferInfoOf(first.size)) + controller.awaitFrames(1) + + val second = ByteArray(300) { it.toByte() } + controller.recordVideo(ByteBuffer.wrap(second), bufferInfoOf(second.size)) + controller.awaitFrames(1) + controller.stopRecord() + + assertEquals(2, controller.written.size) + //writing the first one recycled its array, so the second must get the very same one + assertSame(controller.written[0].first, controller.written[1].first) + //but capped to its own size, the 0xFF left behind are unreachable + assertEquals(second.size, controller.written[1].second) + } + + @Test + fun `GIVEN a frame WHEN record THEN the muxer receives the same bytes`() { + val controller = FakeRecordController() + controller.latch = CountDownLatch(1) + controller.start() + + val frame = byteArrayOf(0, 0, 0, 1, 101, 9, 8, 7) + controller.recordVideo(ByteBuffer.wrap(frame), bufferInfoOf(frame.size)) + controller.awaitFrames(1) + controller.stopRecord() + + val (array, capacity) = controller.written[0] + assertArrayEquals(frame, array.copyOfRange(0, capacity)) + } +}