Merge pull request #2076 from CastagnaIT/wait_segments

Rework wait for segment code

Author: CastagnaIT

src/AdaptiveByteStream.cpp

@@ -59,7 +59,7 @@ AP4_Result CAdaptiveByteStream::GetSize(AP4_LargeSize& size)
 
 bool CAdaptiveByteStream::waitingForSegment() const
 {
-  return m_adStream->waitingForSegment();
+  return m_adStream->IsWaitingForSegment();
 }
 
 void CAdaptiveByteStream::FixateInitialization(bool on)

src/Session.cpp

@@ -812,7 +812,7 @@ bool SESSION::CSession::GetNextSample(ISampleReader*& sampleReader)
         {
           if (!res || streamReader->DTSorPTSManifest() < res->GetReader()->DTSorPTSManifest())
           {
-            if (stream->m_adStream.waitingForSegment())
+            if (stream->m_adStream.OnSampleRequested())
             {
               waiting = stream.get();
             }
@@ -828,6 +828,8 @@ bool SESSION::CSession::GetNextSample(ISampleReader*& sampleReader)
 
   if (waiting)
   {
+    // Prevents CPU usage spikes due to continuous Kodi VP requests
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
     return true;
   }
   else if (res)

src/common/AdaptiveStream.cpp

@@ -71,6 +71,7 @@ void adaptive::AdaptiveStream::Reset()
   segment_read_pos_ = 0;
   currentPTSOffset_ = 0;
   absolutePTSOffset_ = 0;
+  m_isWaitingForSegment = false;
 }
 
 bool adaptive::AdaptiveStream::Download(const DownloadInfo& downloadInfo,
@@ -969,15 +970,15 @@ bool adaptive::AdaptiveStream::ensureSegment()
     }
     else if (!m_tree->IsLastSegment(current_period_, current_rep_, current_rep_->current_segment_))
     {
-      if (m_segBuffers.IsEmpty() && !current_rep_->IsWaitForSegment())
+      if (m_segBuffers.IsEmpty() && !m_isWaitingForSegment)
       {
-        current_rep_->SetIsWaitForSegment(true);
-        LOG::LogF(LOGDEBUG, "[AS-%u] Begin WaitForSegment stream rep. id \"%s\" period id \"%s\"",
+        m_isWaitingForSegment = true;
+        LOG::LogF(LOGDEBUG, "[AS-%u] Begin WaitForSegment (buffer is empty) stream rep. id \"%s\" period id \"%s\"",
                   clsId, current_rep_->GetId().c_str(), current_period_->GetId().c_str());
       }
       return false;
     }
-    else if (current_rep_->IsWaitForSegment() &&
+    else if (m_isWaitingForSegment &&
              (m_tree->HasManifestUpdates() || m_tree->HasManifestUpdatesSegs()))
     {
       return false;
@@ -1115,7 +1116,7 @@ bool adaptive::AdaptiveStream::seek(uint64_t const pos, bool& isEos)
     if (ensureSegment())
       return true;
 
-    if (!current_rep_->IsWaitForSegment())
+    if (!m_isWaitingForSegment)
       isEos = true;
 
     return false;
@@ -1254,21 +1255,49 @@ bool adaptive::AdaptiveStream::seek_time(double seek_seconds)
   return true;
 }
 
-bool adaptive::AdaptiveStream::waitingForSegment() const
+bool adaptive::AdaptiveStream::OnSampleRequested()
 {
-  if (m_tree->IsLive() && thread_data_->State() == THREADDATA::ThState::RUNNING)
+  if (!m_tree->IsLive() || !thread_data_)
+    return false;
+
+  //! @todo: its missing an appropriate buffering management, example:
+  //!        if the bandwidth is too low stuttering may occur due to constant buffering
+  //!        it should wait that buffer if full before to resume the playback,
+  //!        it should also help to manage adaptive streaming (CRepresentationChooserDefault)
+  //!        the mechanism to auto-increase/decrease stream quality based on the actual network conditions
+
+  const auto now = std::chrono::steady_clock::now();
+  const long long elapsedMs =
+      std::chrono::duration_cast<std::chrono::milliseconds>(now - m_tsOnSampleRequest).count();
+
+  // Sample are requested in a very fast way, limit the requests to a minimum interval of 500 ms
+  if (elapsedMs >= 500)
   {
+    m_tsOnSampleRequest = now;
     std::lock_guard<adaptive::AdaptiveTree::TreeUpdateThread> lckUpdTree(m_tree->GetTreeUpdMutex());
 
     // Some manifests require segments to be generated and managed by the client
     // so they are not provided by the server through periodic manifest updates
     m_tree->InsertLiveSegment(current_period_, current_adp_, current_rep_);
 
-    // Although IsWaitForSegment may be true, do not anticipate the wait for segments
-    // if there are still segments in the buffer that can be read and/or downloaded
-    return current_rep_ && current_rep_->IsWaitForSegment() && m_segBuffers.IsEmpty();
+    if (m_isWaitingForSegment)
+    {
+      if (thread_data_->State() != THREADDATA::ThState::RUNNING)
+        return false;
+
+      // Unlocks buffering wait, if a segment exists and the buffer level is full
+      m_isWaitingForSegment = !current_rep_->GetNextSegment() ||
+                              m_segBuffers.GetSizeDownloaded() < m_segBuffers.GetSize();
+
+      if (!m_isWaitingForSegment)
+      {
+        LOG::LogF(LOGDEBUG, "[AS-%u] End WaitForSegment stream rep. id \"%s\" period id \"%s\"",
+                  clsId, current_rep_->GetId().c_str(), current_period_->GetId().c_str());
+      }
+    }
   }
-  return false;
+
+  return m_isWaitingForSegment;
 }
 
 void adaptive::AdaptiveStream::FixateInitialization(bool on)
@@ -1355,6 +1384,8 @@ void adaptive::AdaptiveStream::Stop()
   // otherwise if read some segments may invalidate this change
   if (current_rep_)
     current_rep_->SetIsEnabled(false);
+
+  m_isWaitingForSegment = false;
 }
 
 void adaptive::AdaptiveStream::clear()

src/common/AdaptiveStream.h

@@ -125,7 +125,9 @@ enum class EVENT_TYPE
 
     uint64_t GetCurrentPTSOffset() const { return currentPTSOffset_; }
     uint64_t GetAbsolutePTSOffset() const { return absolutePTSOffset_; }
-    bool waitingForSegment() const;
+
+    bool OnSampleRequested();
+
     void FixateInitialization(bool on);
     void AllowBufferQueue(bool isAllowed) { m_isAllowedBufferQueue = isAllowed; }
     void SetSegmentFileOffset(uint64_t offset) { m_segmentFileOffset = offset; };
@@ -140,6 +142,8 @@ enum class EVENT_TYPE
     */
     bool IsRequiredCreateMovieAtom();
 
+    bool IsWaitingForSegment() const { return m_isWaitingForSegment; }
+
   protected:
     virtual bool parseIndexRange(PLAYLIST::CRepresentation* rep,
                                  const std::vector<uint8_t>& buffer);
@@ -162,6 +166,9 @@ enum class EVENT_TYPE
     std::string m_streamParams;
     std::map<std::string, std::string> m_streamHeaders;
 
+    bool m_isWaitingForSegment{false};
+    mutable std::chrono::time_point<std::chrono::steady_clock> m_tsOnSampleRequest{};
+
    /*!
     * \brief Download a file, the representation chooser will be updated with current download speed.
     * \param downloadInfo The info about the file to download

src/common/Representation.h

@@ -145,9 +145,6 @@ class ATTR_DLL_LOCAL CRepresentation : public CCommonSegAttribs, public CCommonA
   bool IsEnabled() const { return m_isEnabled; }
   void SetIsEnabled(bool isEnabled) { m_isEnabled = isEnabled; }
 
-  bool IsWaitForSegment() const { return m_isWaitForSegment; }
-  void SetIsWaitForSegment(bool isWaitForSegment) { m_isWaitForSegment = isWaitForSegment; }
-
   // Define if it is a dummy representation for audio stream, that is embedded on the video stream
   bool IsIncludedStream() const { return m_isIncludedStream; }
   void SetIsIncludedStream(bool isIncludedStream) { m_isIncludedStream = isIncludedStream; }
@@ -229,7 +226,6 @@ class ATTR_DLL_LOCAL CRepresentation : public CCommonSegAttribs, public CCommonA
   bool m_isSubtitleFileStream{false};
 
   bool m_isEnabled{false};
-  bool m_isWaitForSegment{false};
 
   bool m_isIncludedStream{false};
   std::vector<DRM::DRMInfo> m_drmInfo;

src/common/SegmentBuffer.cpp

@@ -144,6 +144,21 @@ const size_t ADP::CSegmentBuffers::GetSize() const
   return m_buffers.size();
 }
 
+const size_t ADP::CSegmentBuffers::GetSizeDownloaded() const
+{
+  std::lock_guard<std::mutex> lock(m_mutex);
+  size_t size{0};
+  for (const auto& segBuffer : m_buffers)
+  {
+    if (segBuffer.State() == BufferState::DOWNLOADED ||
+        segBuffer.State() == BufferState::DOWNLOADING)
+      size++;
+    else
+      break;
+  }
+  return size;
+}
+
 const bool ADP::CSegmentBuffers::IsEmpty() const
 {
   std::lock_guard<std::mutex> lock(m_mutex);

src/common/SegmentBuffer.h

@@ -139,6 +139,11 @@ class ATTR_DLL_LOCAL CSegmentBuffers
   */
   const size_t GetSize() const;
 
+  /*!
+  * \brief Get the current number of elements contained in the segment buffers container that are in download or downloaded.
+  */
+  const size_t GetSizeDownloaded() const;
+
  /*!
   * \brief Determines whether the segment buffers container is empty.
   * \return True if empty, otherwise false.

src/parser/DASHTree.cpp

@@ -1738,12 +1738,6 @@ void adaptive::CDashTree::OnUpdateSegments()
               LOG::LogF(LOGDEBUG, "MPD update - Done (repr. id \"%s\", period id \"%s\")",
                         updRepr->GetId().c_str(), period->GetId().c_str());
             }
-
-            if (repr->IsWaitForSegment() && repr->GetNextSegment())
-            {
-              repr->SetIsWaitForSegment(false);
-              LOG::LogF(LOGDEBUG, "End WaitForSegment repr. id %s", repr->GetId().c_str());
-            }
           }
         }
       }
@@ -1875,14 +1869,6 @@ void adaptive::CDashTree::InsertLiveSegment(PLAYLIST::CPeriod* currPeriod,
                                             PLAYLIST::CAdaptationSet* currAdpSet,
                                             PLAYLIST::CRepresentation* currRepr)
 {
-  // InsertLiveSegment is called very frequently, since this implementation update all manifest data
-  // you do not need to have all these callbacks in such a short time
-  auto now = std::chrono::steady_clock::now();
-  if (now - m_insertLiveSegUpdate < std::chrono::milliseconds(500))
-    return;
-
-  m_insertLiveSegUpdate = now;
-
   // This code will keep updated the timeline on all representations
   // (across all periods) with current now time (TSB included),
   // exclusive case of representations having SegmentTemplate without defined timeline
@@ -1952,12 +1938,6 @@ void adaptive::CDashTree::InsertLiveSegment(PLAYLIST::CPeriod* currPeriod,
             period->SetTlDuration(tlDuration);
           }
         }
-
-        if (rep->IsWaitForSegment())
-        {
-          rep->SetIsWaitForSegment(false);
-          LOG::LogF(LOGDEBUG, "End WaitForSegment repr. id %s", rep->GetId().c_str());
-        }
       }
     }
   }

src/parser/DASHTree.h

@@ -126,9 +126,6 @@ class ATTR_DLL_LOCAL CDashTree : public adaptive::AdaptiveTree
   uint64_t m_minimumUpdatePeriod{PLAYLIST::NO_VALUE}; // in seconds, NO_VALUE if not set
   std::optional<int64_t> m_clockOffset; // Clock offset in ms, based on UTCTiming element
 
-  // The time point when the last live "insert segment update" has been called
-  mutable std::chrono::time_point<std::chrono::steady_clock> m_insertLiveSegUpdate{};
-
   std::string m_locationUrl; // Redirect manifest updates
 };
 } // namespace adaptive

src/parser/HLSTree.cpp

@@ -933,15 +933,6 @@ void adaptive::CHLSTree::PrepareSegments(PLAYLIST::CPeriod* period,
       rep->current_segment_.reset();
     }
   }
-
-  //! @todo: m_currentPeriod != m_periods.back().get() condition should be removed from here
-  //! this is done on AdaptiveStream::ensureSegment on IsLastSegment check
-  if (rep->IsWaitForSegment() &&
-      (rep->GetNextSegment() || m_currentPeriod != m_periods.back().get()))
-  {
-    LOG::LogF(LOGDEBUG, "End WaitForSegment stream id \"%s\"", rep->GetId().c_str());
-    rep->SetIsWaitForSegment(false);
-  }
 }
 
 void adaptive::CHLSTree::OnDataArrived(uint64_t segNum,