Performance improvements in play request and flush (#428)

media/client/ipc/include/MediaPipelineIpc.h

@@ -79,7 +79,7 @@ class MediaPipelineIpc : public IMediaPipelineIpc, public IpcModule
 
     bool setVideoWindow(uint32_t x, uint32_t y, uint32_t width, uint32_t height) override;
 
-    bool play() override;
+    bool play(bool &async) override;
 
     bool pause() override;
 

media/client/ipc/interface/IMediaPipelineIpc.h

@@ -126,9 +126,11 @@ class IMediaPipelineIpc
     /**
      * @brief Request play on the playback session.
      *
+     * @param[out] async     : True if play method call is asynchronous
+     *
      * @retval true on success.
      */
-    virtual bool play() = 0;
+    virtual bool play(bool &async) = 0;
 
     /**
      * @brief Request pause on the playback session.

media/client/ipc/source/MediaPipelineIpc.cpp

@@ -348,7 +348,7 @@ bool MediaPipelineIpc::setVideoWindow(uint32_t x, uint32_t y, uint32_t width, ui
     return true;
 }
 
-bool MediaPipelineIpc::play()
+bool MediaPipelineIpc::play(bool &async)
 {
     if (!reattachChannelIfRequired())
     {
@@ -375,6 +375,8 @@ bool MediaPipelineIpc::play()
         return false;
     }
 
+    async = response.async();
+
     return true;
 }
 

media/client/main/include/MediaPipeline.h

@@ -118,7 +118,7 @@ class MediaPipeline : public IMediaPipelineAndIControlClient, public IMediaPipel
 
     bool allSourcesAttached() override;
 
-    bool play() override;
+    bool play(bool &async) override;
 
     bool pause() override;
 

media/client/main/include/MediaPipelineProxy.h

@@ -52,7 +52,7 @@ class MediaPipelineProxy : public IMediaPipelineAndIControlClient
 
     bool allSourcesAttached() override { return m_mediaPipeline->allSourcesAttached(); }
 
-    bool play() override { return m_mediaPipeline->play(); }
+    bool play(bool &async) override { return m_mediaPipeline->play(async); }
 
     bool pause() override { return m_mediaPipeline->pause(); }
 

media/client/main/source/MediaPipeline.cpp

@@ -252,11 +252,11 @@ bool MediaPipeline::allSourcesAttached()
     return m_mediaPipelineIpc->allSourcesAttached();
 }
 
-bool MediaPipeline::play()
+bool MediaPipeline::play(bool &async)
 {
     RIALTO_CLIENT_LOG_DEBUG("entry:");
 
-    return m_mediaPipelineIpc->play();
+    return m_mediaPipelineIpc->play(async);
 }
 
 bool MediaPipeline::pause()

media/public/include/IMediaPipeline.h

@@ -1119,16 +1119,15 @@ class IMediaPipeline
     /**
      * @brief Starts playback of the media.
      *
-     * This method is considered to be asynchronous and MUST NOT block
-     * but should request playback and then return.
-     *
      * Once the backend is successfully playing it should notify the
      * media player client of playback state
      * IMediaPipelineClient::PlaybackState::PLAYING.
      *
+     * @param[out] async     : True if play method call is asynchronous
+     *
      * @retval true on success.
      */
-    virtual bool play() = 0;
+    virtual bool play(bool &async) = 0;
 
     /**
      * @brief Pauses playback of the media.

media/server/gstplayer/include/GstGenericPlayer.h

@@ -36,6 +36,7 @@
 #include "tasks/IGenericPlayerTaskFactory.h"
 #include "tasks/IPlayerTask.h"
 #include <IMediaPipeline.h>
+#include <atomic>
 #include <memory>
 #include <string>
 #include <utility>
@@ -108,7 +109,7 @@ class GstGenericPlayer : public IGstGenericPlayer, public IGstGenericPlayerPriva
     void attachSource(const std::unique_ptr<IMediaPipeline::MediaSource> &mediaSource) override;
     void removeSource(const MediaSourceType &mediaSourceType) override;
     void allSourcesAttached() override;
-    void play() override;
+    void play(bool &async) override;
     void pause() override;
     void stop() override;
     void attachSamples(const IMediaPipeline::MediaSegmentVector &mediaSegments) override;
@@ -168,7 +169,7 @@ class GstGenericPlayer : public IGstGenericPlayer, public IGstGenericPlayerPriva
     void updateVideoCaps(int32_t width, int32_t height, Fraction frameRate,
                          const std::shared_ptr<CodecData> &codecData) override;
     void addAudioClippingToBuffer(GstBuffer *buffer, uint64_t clippingStart, uint64_t clippingEnd) const override;
-    bool changePipelineState(GstState newState) override;
+    GstStateChangeReturn changePipelineState(GstState newState) override;
     int64_t getPosition(GstElement *element) override;
     void startPositionReportingAndCheckAudioUnderflowTimer() override;
     void stopPositionReportingAndCheckAudioUnderflowTimer() override;
@@ -427,6 +428,11 @@ class GstGenericPlayer : public IGstGenericPlayer, public IGstGenericPlayerPriva
      * @brief The postponed flush tasks
      */
     std::vector<std::pair<MediaSourceType, bool>> m_postponedFlushes{};
+
+    /**
+     * @brief The ongoing state change operations counter
+     */
+    std::atomic<uint32_t> m_ongoingStateChangesNumber{0};
 };
 
 } // namespace firebolt::rialto::server

media/server/gstplayer/include/IGstGenericPlayerPrivate.h

@@ -174,9 +174,9 @@ class IGstGenericPlayerPrivate
      *
      * @param[in] newState    : The desired state.
      *
-     * @retval true on success.
+     * @retval state change status
      */
-    virtual bool changePipelineState(GstState newState) = 0;
+    virtual GstStateChangeReturn changePipelineState(GstState newState) = 0;
 
     /**
      * @brief Gets the current position of the element

media/server/gstplayer/interface/IGstGenericPlayer.h

@@ -105,14 +105,15 @@ class IGstGenericPlayer
     /**
      * @brief Starts playback of the media.
      *
-     * This method is considered to be asynchronous and MUST NOT block
-     * but should request playback and then return.
-     *
      * Once the backend is successfully playing it should notify the
-     * media player client of playback state PlaybackState::PLAYING.
+     * media player client of playback state
+     * IMediaPipelineClient::PlaybackState::PLAYING.
      *
+     * @param[out] async     : True if play method call is asynchronous
+     *
+     * @retval true on success.
      */
-    virtual void play() = 0;
+    virtual void play(bool &async) = 0;
 
     /**
      * @brief Pauses playback of the media.

media/server/gstplayer/source/GstGenericPlayer.cpp

@@ -1186,16 +1186,32 @@ void GstGenericPlayer::cancelUnderflow(firebolt::rialto::MediaSourceType mediaSo
     }
 }
 
-void GstGenericPlayer::play()
+void GstGenericPlayer::play(bool &async)
 {
-    if (m_workerThread)
+    if (0 == m_ongoingStateChangesNumber)
     {
-        m_workerThread->enqueueTask(m_taskFactory->createPlay(*this));
+        // Operation called on main thread, because PAUSED->PLAYING change is synchronous and needs to be done fast.
+        //
+        // m_context.pipeline can be used, because it's modified only in GstGenericPlayer
+        // constructor and destructor. GstGenericPlayer is created/destructed on main thread, so we won't have a crash here.
+        ++m_ongoingStateChangesNumber;
+        async = (changePipelineState(GST_STATE_PLAYING) == GST_STATE_CHANGE_ASYNC);
+        RIALTO_SERVER_LOG_MIL("State change to PLAYING requested");
+    }
+    else
+    {
+        ++m_ongoingStateChangesNumber;
+        async = true;
+        if (m_workerThread)
+        {
+            m_workerThread->enqueueTask(m_taskFactory->createPlay(*this));
+        }
     }
 }
 
 void GstGenericPlayer::pause()
 {
+    ++m_ongoingStateChangesNumber;
     if (m_workerThread)
     {
         m_workerThread->enqueueTask(m_taskFactory->createPause(m_context, *this));
@@ -1204,29 +1220,32 @@ void GstGenericPlayer::pause()
 
 void GstGenericPlayer::stop()
 {
+    ++m_ongoingStateChangesNumber;
     if (m_workerThread)
     {
         m_workerThread->enqueueTask(m_taskFactory->createStop(m_context, *this));
     }
 }
 
-bool GstGenericPlayer::changePipelineState(GstState newState)
+GstStateChangeReturn GstGenericPlayer::changePipelineState(GstState newState)
 {
     if (!m_context.pipeline)
     {
         RIALTO_SERVER_LOG_ERROR("Change state failed - pipeline is nullptr");
         if (m_gstPlayerClient)
             m_gstPlayerClient->notifyPlaybackState(PlaybackState::FAILURE);
-        return false;
+        --m_ongoingStateChangesNumber;
+        return GST_STATE_CHANGE_FAILURE;
     }
-    if (m_gstWrapper->gstElementSetState(m_context.pipeline, newState) == GST_STATE_CHANGE_FAILURE)
+    const GstStateChangeReturn result{m_gstWrapper->gstElementSetState(m_context.pipeline, newState)};
+    if (result == GST_STATE_CHANGE_FAILURE)
     {
         RIALTO_SERVER_LOG_ERROR("Change state failed - Gstreamer returned an error");
         if (m_gstPlayerClient)
             m_gstPlayerClient->notifyPlaybackState(PlaybackState::FAILURE);
-        return false;
     }
-    return true;
+    --m_ongoingStateChangesNumber;
+    return result;
 }
 
 int64_t GstGenericPlayer::getPosition(GstElement *element)

media/server/ipc/source/MediaPipelineModuleService.cpp

@@ -566,11 +566,13 @@ void MediaPipelineModuleService::play(::google::protobuf::RpcController *control
                                       ::firebolt::rialto::PlayResponse *response, ::google::protobuf::Closure *done)
 {
     RIALTO_SERVER_LOG_DEBUG("entry:");
-    if (!m_mediaPipelineService.play(request->session_id()))
+    bool async{false};
+    if (!m_mediaPipelineService.play(request->session_id(), async))
     {
         RIALTO_SERVER_LOG_ERROR("Play failed");
         controller->SetFailed("Operation failed");
     }
+    response->set_async(async);
     done->Run();
 }
 

media/server/main/include/MediaPipelineServerInternal.h

@@ -99,7 +99,7 @@ class MediaPipelineServerInternal : public IMediaPipelineServerInternal, public
 
     bool allSourcesAttached() override;
 
-    bool play() override;
+    bool play(bool &async) override;
 
     bool pause() override;
 
@@ -346,9 +346,11 @@ class MediaPipelineServerInternal : public IMediaPipelineServerInternal, public
     /**
      * @brief Play internally, only to be called on the main thread.
      *
+     * @param[out] async     : True if play method call is asynchronous
+     *
      * @retval true on success.
      */
-    bool playInternal();
+    bool playInternal(bool &async);
 
     /**
      * @brief Pause internally, only to be called on the main thread.

media/server/main/source/MediaPipelineServerInternal.cpp

@@ -335,26 +335,26 @@ bool MediaPipelineServerInternal::allSourcesAttachedInternal()
     return true;
 }
 
-bool MediaPipelineServerInternal::play()
+bool MediaPipelineServerInternal::play(bool &async)
 {
     RIALTO_SERVER_LOG_DEBUG("entry:");
 
     bool result;
-    auto task = [&]() { result = playInternal(); };
+    auto task = [&]() { result = playInternal(async); };
 
-    m_mainThread->enqueueTaskAndWait(m_mainThreadClientId, task);
+    m_mainThread->enqueuePriorityTaskAndWait(m_mainThreadClientId, task);
     return result;
 }
 
-bool MediaPipelineServerInternal::playInternal()
+bool MediaPipelineServerInternal::playInternal(bool &async)
 {
     if (!m_gstPlayer)
     {
         RIALTO_SERVER_LOG_ERROR("Failed to play - Gstreamer player has not been loaded");
         return false;
     }
 
-    m_gstPlayer->play();
+    m_gstPlayer->play(async);
     return true;
 }
 

media/server/service/include/IMediaPipelineService.h

@@ -48,7 +48,7 @@ class IMediaPipelineService
     virtual bool attachSource(int sessionId, const std::unique_ptr<IMediaPipeline::MediaSource> &source) = 0;
     virtual bool removeSource(int sessionId, std::int32_t sourceId) = 0;
     virtual bool allSourcesAttached(int sessionId) = 0;
-    virtual bool play(int sessionId) = 0;
+    virtual bool play(int sessionId, bool &async) = 0;
     virtual bool pause(int sessionId) = 0;
     virtual bool stop(int sessionId) = 0;
     virtual bool setPlaybackRate(int sessionId, double rate) = 0;

media/server/service/source/MediaPipelineService.cpp

@@ -169,7 +169,7 @@ bool MediaPipelineService::allSourcesAttached(int sessionId)
     return mediaPipelineIter->second->allSourcesAttached();
 }
 
-bool MediaPipelineService::play(int sessionId)
+bool MediaPipelineService::play(int sessionId, bool &async)
 {
     RIALTO_SERVER_LOG_INFO("MediaPipelineService requested to play, session id: %d", sessionId);
 
@@ -180,7 +180,7 @@ bool MediaPipelineService::play(int sessionId)
         RIALTO_SERVER_LOG_ERROR("Session with id: %d does not exists", sessionId);
         return false;
     }
-    return mediaPipelineIter->second->play();
+    return mediaPipelineIter->second->play(async);
 }
 
 bool MediaPipelineService::pause(int sessionId)

media/server/service/source/MediaPipelineService.h

@@ -59,7 +59,7 @@ class MediaPipelineService : public IMediaPipelineService
     bool attachSource(int sessionId, const std::unique_ptr<IMediaPipeline::MediaSource> &source) override;
     bool removeSource(int sessionId, std::int32_t sourceId) override;
     bool allSourcesAttached(int sessionId) override;
-    bool play(int sessionId) override;
+    bool play(int sessionId, bool &async) override;
     bool pause(int sessionId) override;
     bool stop(int sessionId) override;
     bool setPlaybackRate(int sessionId, double rate) override;

proto/mediapipelinemodule.proto

@@ -261,7 +261,8 @@ message SetVideoWindowResponse {
  * @fn void play(int session_id)
  * @brief Starts playback on a session.
  *
- * @param[in]  session_id   The id of the A/V session.
+ * @param[in]   session_id  The id of the A/V session.
+ * @param[out]  async       True if play method call is asynchronous
  *
  * This method is asynchronous. Once the backend is successfully playing it will notify the media player client of
  * playback state.
@@ -271,6 +272,7 @@ message PlayRequest {
     optional int32 session_id = 1 [default = -1];
 }
 message PlayResponse {
+    optional bool async = 1 [default = true];
 }
 
 /**

tests/componenttests/client/tests/base/MediaPipelineTestMethods.cpp

@@ -892,7 +892,8 @@ void MediaPipelineTestMethods::shouldNotifyPlaybackStateFailure()
 
 void MediaPipelineTestMethods::playFailure()
 {
-    EXPECT_EQ(m_mediaPipeline->play(), false);
+    bool async{false};
+    EXPECT_EQ(m_mediaPipeline->play(async), false);
 }
 
 void MediaPipelineTestMethods::pauseFailure()
@@ -2033,7 +2034,8 @@ void MediaPipelineTestMethods::haveDataInternal(const std::unique_ptr<IMediaPipe
 
 void MediaPipelineTestMethods::playInternal(const std::unique_ptr<IMediaPipeline> &mediaPipeline, const bool status)
 {
-    EXPECT_EQ(mediaPipeline->play(), status);
+    bool async{false};
+    EXPECT_EQ(mediaPipeline->play(async), status);
 }
 
 void MediaPipelineTestMethods::sendNotifyPlaybackStateInternal(const int32_t sessionId, const PlaybackState &state)