RDKEMW-15991 : gst_app_src_set_max_bytes to gst_app_src_set_max_buffers

CMakeLists.txt

@@ -98,6 +98,8 @@ if ( RIALTO_LOG_DEBUG_ENABLED )
     add_compile_definitions( RIALTO_LOG_DEBUG_ENABLED )
 endif()
 
+# add_compile_definitions( RIALTO_ENABLE_BUFFER_SIZE_LOGGING )
+
 #check if TextTrack plugin exists in Thunder
 find_path( TEXT_TRACK_INCLUDE_DIR NAMES WPEFramework/interfaces/ITextTrack.h )
 if (TEXT_TRACK_INCLUDE_DIR)

media/server/gstplayer/include/GstPlayerConfig.h

@@ -0,0 +1,119 @@
+/*
+ * If not stated otherwise in this file or this component's LICENSE file the
+ * following copyright and licenses apply:
+ *
+ * Copyright 2024 Sky UK
+ *
+ * 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.
+ */
+
+#ifndef FIREBOLT_RIALTO_SERVER_GST_PLAYER_CONFIG_H_
+#define FIREBOLT_RIALTO_SERVER_GST_PLAYER_CONFIG_H_
+
+/**
+ * @file GstPlayerConfig.h
+ *
+ * Configuration constants for GStreamer player buffer management.
+ *
+ * These constants control the appsrc queue limits for different media types.
+ * They support both byte-based limits (GStreamer < 1.20) and buffer-based
+ * limits (GStreamer >= 1.20).
+ *
+ * IMPORTANT: The buffer counts are based on estimated average buffer sizes.
+ * These estimates may need adjustment based on your actual content:
+ *   - Video buffers vary by codec, bitrate, GOP structure, resolution
+ *   - Audio buffers vary by codec, bitrate, sample rate, chunk duration
+ *   - Subtitle buffers vary by format complexity and styling
+ *
+ * To validate and tune these values:
+ *   1. Build with -DRIALTO_ENABLE_BUFFER_SIZE_LOGGING
+ *   2. Run typical playback scenarios
+ *   3. Analyze actual buffer sizes in logs
+ *   4. Adjust these constants if needed
+ */
+
+#include <stdint.h>
+
+namespace firebolt::rialto::server
+{
+/**
+ * @brief Video appsrc max queue size in bytes (for GStreamer < 1.20)
+ */
+constexpr uint32_t kVideoMaxBytes = 8 * 1024 * 1024; // 8 MB
+
+/**
+ * @brief Audio appsrc max queue size in bytes (for GStreamer < 1.20)
+ */
+constexpr uint32_t kAudioMaxBytes = 512 * 1024; // 512 KB
+
+/**
+ * @brief Subtitle appsrc max queue size in bytes (for GStreamer < 1.20)
+ */
+constexpr uint32_t kSubtitleMaxBytes = 256 * 1024; // 256 KB
+
+/**
+ * @brief WebAudio appsrc max queue size in bytes (for GStreamer < 1.20)
+ */
+constexpr uint32_t kWebAudioMaxBytes = 10 * 1024; // 10 KB
+
+/**
+ * @brief Video appsrc max queue size in buffer count (for GStreamer >= 1.20)
+ *
+ * Based on: 8MB ÷ ~200KB avg video buffer = 40 buffers
+ * Estimated avg buffer size: ~200 KB (HD video with mixed I/P/B frames)
+ *
+ * Adjust based on your content characteristics:
+ *   - Higher for low bitrate content (smaller buffers)
+ *   - Lower for high bitrate 4K content (larger buffers)
+ */
+constexpr uint32_t kVideoMaxBuffers = 40;
+
+/**
+ * @brief Audio appsrc max queue size in buffer count (for GStreamer >= 1.20)
+ *
+ * Based on: 512KB ÷ ~16KB avg audio buffer = 32 buffers
+ * Estimated avg buffer size: ~16 KB (compressed audio at 128-256 kbps)
+ *
+ * Adjust based on your audio format:
+ *   - Higher for lower bitrate audio (smaller buffers)
+ *   - Lower for high bitrate or lossless audio (larger buffers)
+ */
+constexpr uint32_t kAudioMaxBuffers = 32;
+
+/**
+ * @brief Subtitle appsrc max queue size in buffer count (for GStreamer >= 1.20)
+ *
+ * Based on: 256KB ÷ ~4KB avg subtitle buffer = 64 buffers
+ * Estimated avg buffer size: ~4 KB (text subtitles with minimal styling)
+ *
+ * Adjust based on subtitle format:
+ *   - Higher for simple text subtitles (smaller buffers)
+ *   - Lower for complex styled or image-based subtitles (larger buffers)
+ */
+constexpr uint32_t kSubtitleMaxBuffers = 64;
+
+/**
+ * @brief WebAudio appsrc max queue size in buffer count (for GStreamer >= 1.20)
+ *
+ * Based on: 10KB ÷ ~512 bytes avg buffer = 20 buffers
+ * Estimated avg buffer size: ~512 bytes (small chunks for low latency)
+ *
+ * Adjust based on latency requirements:
+ *   - Higher for lower latency (smaller chunk sizes)
+ *   - Lower if larger chunks acceptable
+ */
+constexpr uint32_t kWebAudioMaxBuffers = 20;
+
+} // namespace firebolt::rialto::server
+
+#endif // FIREBOLT_RIALTO_SERVER_GST_PLAYER_CONFIG_H_

media/server/gstplayer/include/WebAudioPlayerContext.h

@@ -28,11 +28,13 @@
 #include <memory>
 #include <mutex>
 
+#include "GstPlayerConfig.h"
 #include "IGstSrc.h"
 
 namespace firebolt::rialto::server
 {
-constexpr uint32_t kMaxWebAudioBytes{10 * 1024};
+// WebAudio buffer limits are now defined in GstPlayerConfig.h
+// Use kWebAudioMaxBytes and kWebAudioMaxBuffers constants
 
 struct WebAudioPlayerContext
 {

media/server/gstplayer/source/GstGenericPlayer.cpp

@@ -1327,6 +1327,11 @@ void GstGenericPlayer::attachData(const firebolt::rialto::MediaSourceType mediaT
 
         for (GstBuffer *buffer : streamInfo.buffers)
         {
+#ifdef RIALTO_ENABLE_BUFFER_SIZE_LOGGING
+            gsize bufferSize = gst_buffer_get_size(buffer);
+            RIALTO_SERVER_LOG_DEBUG("Pushing %s buffer of size: %zu bytes", common::convertMediaSourceType(mediaType),
+                                    bufferSize);
+#endif
             m_gstWrapper->gstAppSrcPushBuffer(GST_APP_SRC(streamInfo.appSrc), buffer);
         }
         streamInfo.buffers.clear();

media/server/gstplayer/source/GstSrc.cpp

@@ -19,6 +19,7 @@
 
 #include <stdexcept>
 
+#include "GstPlayerConfig.h"
 #include "GstSrc.h"
 #include "GstTextTrackSinkFactory.h"
 #include "RialtoServerLogging.h"
@@ -397,10 +398,34 @@ void GstSrc::setupAndAddAppSrc(IDecryptionService *decryptionService, GstElement
                               GST_APP_STREAM_TYPE_STREAM, "min-percent", 20, "handle-segment-change", TRUE, nullptr);
     m_gstWrapper->gstAppSrcSetCallbacks(GST_APP_SRC(streamInfo.appSrc), callbacks, userData, nullptr);
 
+    // Queue size limits: we support both byte-based (GStreamer < 1.20) and buffer-based (>= 1.20) limits.
+    // The buffer counts are computed to maintain similar memory footprint as the byte limits.
+    // To validate these assumptions, build with -DRIALTO_ENABLE_BUFFER_SIZE_LOGGING to log actual buffer sizes.
+    // See GstPlayerConfig.h to adjust these values.
+
+#if GST_CHECK_VERSION(1, 20, 0)
+    // Queue size in buffer counts (for GStreamer >= 1.20)
+    // See GstPlayerConfig.h for detailed documentation on these values.
+    const std::unordered_map<firebolt::rialto::MediaSourceType, uint32_t> queueBuffers =
+        {{firebolt::rialto::MediaSourceType::VIDEO, kVideoMaxBuffers},
+         {firebolt::rialto::MediaSourceType::AUDIO, kAudioMaxBuffers},
+         {firebolt::rialto::MediaSourceType::SUBTITLE, kSubtitleMaxBuffers}};
+
+    auto buffersIt = queueBuffers.find(type);
+    if (buffersIt != queueBuffers.end())
+    {
+        m_gstWrapper->gstAppSrcSetMaxBuffers(GST_APP_SRC(streamInfo.appSrc), buffersIt->second);
+    }
+    else
+    {
+        GST_WARNING_OBJECT(source, "Could not find max-buffers value for appsrc");
+    }
+#else
+    // Queue size in bytes (for GStreamer < 1.20)
     const std::unordered_map<firebolt::rialto::MediaSourceType, uint32_t> queueSize =
-        {{firebolt::rialto::MediaSourceType::VIDEO, 8 * 1024 * 1024},
-         {firebolt::rialto::MediaSourceType::AUDIO, 512 * 1024},
-         {firebolt::rialto::MediaSourceType::SUBTITLE, 256 * 1024}};
+        {{firebolt::rialto::MediaSourceType::VIDEO, kVideoMaxBytes},
+         {firebolt::rialto::MediaSourceType::AUDIO, kAudioMaxBytes},
+         {firebolt::rialto::MediaSourceType::SUBTITLE, kSubtitleMaxBytes}};
 
     auto sizeIt = queueSize.find(type);
     if (sizeIt != queueSize.end())
@@ -411,6 +436,7 @@ void GstSrc::setupAndAddAppSrc(IDecryptionService *decryptionService, GstElement
     {
         GST_WARNING_OBJECT(source, "Could not find max-bytes value for appsrc");
     }
+#endif
 
     m_gstWrapper->gstAppSrcSetStreamType(GST_APP_SRC(streamInfo.appSrc), GST_APP_STREAM_TYPE_SEEKABLE);
 

media/server/gstplayer/source/GstWebAudioPlayer.cpp

@@ -163,7 +163,16 @@ bool GstWebAudioPlayer::initWebAudioPipeline(const uint32_t priority)
         RIALTO_SERVER_LOG_ERROR("Failed to create the appsrc");
         return false;
     }
-    m_gstWrapper->gstAppSrcSetMaxBytes(GST_APP_SRC(m_context.source), kMaxWebAudioBytes);
+
+    // Set queue limits: Use buffer-based on GStreamer 1.20+, byte-based on older versions.
+    // See GstPlayerConfig.h to adjust these values.
+#if GST_CHECK_VERSION(1, 20, 0)
+    m_gstWrapper->gstAppSrcSetMaxBuffers(GST_APP_SRC(m_context.source), kWebAudioMaxBuffers);
+    // Also set max-bytes on 1.20+ to keep current-level-bytes bounded and prevent underflow in WriteBuffer
+    m_gstWrapper->gstAppSrcSetMaxBytes(GST_APP_SRC(m_context.source), kWebAudioMaxBytes);
+#else
+    m_gstWrapper->gstAppSrcSetMaxBytes(GST_APP_SRC(m_context.source), kWebAudioMaxBytes);
+#endif
     m_glibWrapper->gObjectSet(m_context.source, "format", GST_FORMAT_TIME, nullptr);
 
     // Perform sink specific initalisation

media/server/gstplayer/source/tasks/webAudio/WriteBuffer.cpp

@@ -44,7 +44,8 @@ void WriteBuffer::execute() const
 {
     RIALTO_SERVER_LOG_DEBUG("Executing WriteBuffer");
 
-    uint64_t freeBytes = kMaxWebAudioBytes - m_gstWrapper->gstAppSrcGetCurrentLevelBytes(GST_APP_SRC(m_context.source));
+    uint64_t currentLevelBytes = m_gstWrapper->gstAppSrcGetCurrentLevelBytes(GST_APP_SRC(m_context.source));
+    uint64_t freeBytes = (currentLevelBytes >= kWebAudioMaxBytes) ? 0 : (kWebAudioMaxBytes - currentLevelBytes);
     uint64_t maxBytesToWrite = std::min(freeBytes, m_mainLength + m_wrapLength);
     uint64_t bytesToWrite = maxBytesToWrite - (maxBytesToWrite % m_context.bytesPerSample);
     uint64_t bytesWritten = 0;
@@ -80,6 +81,11 @@ void WriteBuffer::execute() const
                                        bytesToWrite, bytesWritten);
             }
 
+#ifdef RIALTO_ENABLE_BUFFER_SIZE_LOGGING
+            gsize bufferSize = gst_buffer_get_size(gstBuffer);
+            RIALTO_SERVER_LOG_DEBUG("Pushing WebAudio buffer of size: %zu bytes", bufferSize);
+#endif
+
             if (GST_FLOW_OK != m_gstWrapper->gstAppSrcPushBuffer(GST_APP_SRC(m_context.source), gstBuffer))
             {
                 RIALTO_SERVER_LOG_ERROR("Failed to push the buffers to the appsrc");

tests/common/externalLibraryMocks/GstWrapperMock.h

@@ -63,6 +63,7 @@ class GstWrapperMock : public IGstWrapper
                 (GstAppSrc * appsrc, GstAppSrcCallbacks *callbacks, gpointer userData, GDestroyNotify notify),
                 (override));
     MOCK_METHOD(void, gstAppSrcSetMaxBytes, (GstAppSrc * appsrc, guint64 max), (override));
+    MOCK_METHOD(void, gstAppSrcSetMaxBuffers, (GstAppSrc * appsrc, guint max), (override));
     MOCK_METHOD(void, gstAppSrcSetStreamType, (GstAppSrc * appsrc, GstAppStreamType type), (override));
     MOCK_METHOD(gboolean, gstBinAdd, (GstBin * bin, GstElement *element), (override));
     MOCK_METHOD(void, gstBaseTransformSetInPlace, (GstBaseTransform * trans, gboolean in_place), (override));

tests/componenttests/server/fixtures/MediaPipelineTest.cpp

@@ -183,7 +183,13 @@ void MediaPipelineTest::willSetupAndAddSource(GstAppSrc *appSrc)
     EXPECT_CALL(*m_glibWrapperMock, gObjectSetStub(GST_ELEMENT(appSrc), StrEq("stream-type")));
     EXPECT_CALL(*m_glibWrapperMock, gObjectSetStub(GST_ELEMENT(appSrc), StrEq("min-percent")));
     EXPECT_CALL(*m_glibWrapperMock, gObjectSetStub(GST_ELEMENT(appSrc), StrEq("handle-segment-change")));
+#if GST_CHECK_VERSION(1, 20, 0)
+    // GStreamer 1.20+ uses buffer-based limits
+    const guint kMaxBuffers = ((appSrc == &m_audioAppSrc) ? 32 : 40);
+    EXPECT_CALL(*m_gstWrapperMock, gstAppSrcSetMaxBuffers(appSrc, kMaxBuffers));
+#else
     EXPECT_CALL(*m_gstWrapperMock, gstAppSrcSetMaxBytes(appSrc, kMaxBytes));
+#endif
     EXPECT_CALL(*m_gstWrapperMock, gstAppSrcSetStreamType(appSrc, GST_APP_STREAM_TYPE_SEEKABLE));
     EXPECT_CALL(*m_glibWrapperMock, gStrdupPrintfStub(_)).WillOnce(Return(m_sourceName.data())).RetiresOnSaturation();
     EXPECT_CALL(*m_gstWrapperMock, gstBinAdd(GST_BIN(&m_rialtoSource), GST_ELEMENT(appSrc)));

tests/componenttests/server/tests/mediaPipeline/DualVideoPlaybackTest.cpp

@@ -167,7 +167,11 @@ class DualVideoPlaybackTest : public MediaPipelineTest
         EXPECT_CALL(*m_glibWrapperMock, gObjectSetStub(GST_ELEMENT(&m_secondaryVideoAppSrc), StrEq("min-percent")));
         EXPECT_CALL(*m_glibWrapperMock,
                     gObjectSetStub(GST_ELEMENT(&m_secondaryVideoAppSrc), StrEq("handle-segment-change")));
+#if GST_CHECK_VERSION(1, 20, 0)
+        EXPECT_CALL(*m_gstWrapperMock, gstAppSrcSetMaxBuffers(&m_secondaryVideoAppSrc, 40));
+#else
         EXPECT_CALL(*m_gstWrapperMock, gstAppSrcSetMaxBytes(&m_secondaryVideoAppSrc, (8 * 1024 * 1024)));
+#endif
         EXPECT_CALL(*m_gstWrapperMock, gstAppSrcSetStreamType(&m_secondaryVideoAppSrc, GST_APP_STREAM_TYPE_SEEKABLE));
         EXPECT_CALL(*m_glibWrapperMock, gStrdupPrintfStub(_)).WillOnce(Return(m_sourceName.data())).RetiresOnSaturation();
         EXPECT_CALL(*m_gstWrapperMock,

tests/componenttests/server/tests/webAudio/WebAudioTestMethods.cpp

@@ -175,7 +175,11 @@ void WebAudioTestMethods::willCreateWebAudioPlayer()
 
     // EXPECTS coming from...
     //   GstWebAudioPlayerTestCommon::expectInitAppSrc()
+#if GST_CHECK_VERSION(1, 20, 0)
+    EXPECT_CALL(*m_gstWrapperMock, gstAppSrcSetMaxBuffers(&m_appSrc, 20));
+#else
     EXPECT_CALL(*m_gstWrapperMock, gstAppSrcSetMaxBytes(&m_appSrc, 10 * 1024));
+#endif
     EXPECT_CALL(*m_glibWrapperMock, gObjectSetStub(G_OBJECT(&m_appSrc), StrEq("format")));
 
     // EXPECTS coming from...

tests/unittests/media/server/gstplayer/rialtoSrc/AppSrcTest.cpp

@@ -114,7 +114,13 @@ class RialtoServerAppSrcGstSrcTest : public ::testing::Test
 
         EXPECT_CALL(*m_gstWrapperMock,
                     gstAppSrcSetCallbacks(GST_APP_SRC(m_streamInfo.appSrc), &m_callbacks, this, nullptr));
+#if GST_CHECK_VERSION(1, 20, 0)
+        // GStreamer 1.20+ uses buffer-based limits
+        guint maxBuffers = (max == 8 * 1024 * 1024) ? 40 : ((max == 512 * 1024) ? 32 : 64);
+        EXPECT_CALL(*m_gstWrapperMock, gstAppSrcSetMaxBuffers(GST_APP_SRC(m_streamInfo.appSrc), maxBuffers));
+#else
         EXPECT_CALL(*m_gstWrapperMock, gstAppSrcSetMaxBytes(GST_APP_SRC(m_streamInfo.appSrc), max));
+#endif
         EXPECT_CALL(*m_gstWrapperMock,
                     gstAppSrcSetStreamType(GST_APP_SRC(m_streamInfo.appSrc), GST_APP_STREAM_TYPE_SEEKABLE));
     }

tests/unittests/media/server/gstplayer/webAudioPlayer/common/GstWebAudioPlayerTestCommon.cpp

@@ -119,7 +119,11 @@ void GstWebAudioPlayerTestCommon::expectCreatePipeline()
 void GstWebAudioPlayerTestCommon::expectInitAppSrc()
 {
     EXPECT_CALL(*m_gstWrapperMock, gstElementFactoryMake(StrEq("appsrc"), StrEq("audsrc"))).WillOnce(Return(&m_appSrc));
+#if GST_CHECK_VERSION(1, 20, 0)
+    EXPECT_CALL(*m_gstWrapperMock, gstAppSrcSetMaxBuffers(GST_APP_SRC(&m_appSrc), 20));
+#else
     EXPECT_CALL(*m_gstWrapperMock, gstAppSrcSetMaxBytes(GST_APP_SRC(&m_appSrc), 10 * 1024));
+#endif
     EXPECT_CALL(*m_glibWrapperMock, gObjectSetStub(G_OBJECT(&m_appSrc), StrEq("format")));
 }
 

tests/unittests/media/server/gstplayer/webAudioPlayer/common/WebAudioTasksTestsBase.cpp

@@ -44,7 +44,7 @@ std::shared_ptr<WebAudioTasksTestsContext> testContext;
 constexpr double kVolume{0.7};
 constexpr uint64_t kChannelMask{5};
 constexpr uint32_t kBytesPerSample{4};
-constexpr uint32_t kMainLength = firebolt::rialto::server::kMaxWebAudioBytes - 40;
+constexpr uint32_t kMainLength = firebolt::rialto::server::kWebAudioMaxBytes - 40;
 constexpr uint32_t kWrapLength = 40;
 constexpr uint32_t kUnalignedWriteSize = kMainLength + kWrapLength - kBytesPerSample;
 const std::string kAudioMimeType{"audio/x-raw"};
@@ -386,7 +386,7 @@ void WebAudioTasksTestsBase::shouldNotWriteBufferIfPushBufferFails()
 void WebAudioTasksTestsBase::shouldNotWriteBufferIfBytesToWriteLessThanBytesPerSample()
 {
     EXPECT_CALL(*testContext->m_gstWrapper, gstAppSrcGetCurrentLevelBytes(GST_APP_SRC(&testContext->m_src)))
-        .WillOnce(Return(firebolt::rialto::server::kMaxWebAudioBytes - testContext->m_context.bytesPerSample + 1));
+        .WillOnce(Return(firebolt::rialto::server::kWebAudioMaxBytes - testContext->m_context.bytesPerSample + 1));
 }
 
 void WebAudioTasksTestsBase::shouldWriteBufferThatNotAlignedWithBytesPerSample()

wrappers/include/GstWrapper.h

@@ -157,6 +157,16 @@ class GstWrapper : public IGstWrapper
 
     void gstAppSrcSetMaxBytes(GstAppSrc *appsrc, guint64 max) override { gst_app_src_set_max_bytes(appsrc, max); }
 
+    void gstAppSrcSetMaxBuffers(GstAppSrc *appsrc, guint max) override
+    {
+#if GST_CHECK_VERSION(1, 20, 0)
+        gst_app_src_set_max_buffers(appsrc, max);
+#else
+        (void)appsrc;
+        (void)max;
+#endif
+    }
+
     void gstAppSrcSetStreamType(GstAppSrc *appsrc, GstAppStreamType type) override
     {
         gst_app_src_set_stream_type(appsrc, type);

wrappers/interface/IGstWrapper.h

@@ -339,6 +339,15 @@ class IGstWrapper
      */
     virtual void gstAppSrcSetMaxBytes(GstAppSrc *appsrc, guint64 max) = 0;
 
+    /**
+     * @brief Set the maximum buffers that can be set on the appsrc queue.
+     *        Requires GStreamer >= 1.20.
+     *
+     * @param[in] appsrc    : The appsrc.
+     * @param[in] max       : Max buffers to set.
+     */
+    virtual void gstAppSrcSetMaxBuffers(GstAppSrc *appsrc, guint max) = 0;
+
     /**
      * @brief Set the stream type on the appsrc.
      *