feat: test playahead buffer test

Signed-off-by: deltag0 <ioan.1931@gmail.com>

Author: deltag0

src/lib/image/MovieFFMpeg/MovieFFMpeg.cpp

@@ -20,6 +20,7 @@
 #include <TwkUtil/PathConform.h>
 #include <TwkUtil/File.h>
 #include <TwkUtil/sgcHop.h>
+#include <stream/StreamPreloadPool.h>
 #include <boost/thread/lock_algorithms.hpp>
 #include <boost/thread/mutex.hpp>
 #include <cstddef>
@@ -66,6 +67,7 @@ extern "C"
 
 static ENVVAR_BOOL(evUseUploadedMovieForStreaming, "RV_SHOTGRID_USE_UPLOADED_MOVIE_FOR_STREAMING", false);
 static ENVVAR_STRING(evStreamCachePath, "RV_STREAM_CACHE_PATH", "/tmp");
+static ENVVAR_FLOAT(evPlayheadPrefetchSeconds, "RV_STREAM_PLAYHEAD_PREFETCH_SECONDS", 5.0f);
 
 namespace TwkMovie
 {
@@ -1337,6 +1339,36 @@ namespace TwkMovie
         }
     }
 
+    void MovieFFMpegReader::prefetchAtFrame(int frame)
+    {
+        const float windowSeconds = evPlayheadPrefetchSeconds.getValue();
+        if (!TwkUtil::pathIsURL(m_filename) || windowSeconds <= 0.0f || m_info.fps <= 0.0f)
+            return;
+
+        const double frameSeconds = std::max(0.0, double(frame - m_info.start) / double(m_info.fps));
+        const int64_t window = static_cast<int64_t>(frameSeconds / windowSeconds);
+        if (m_lastPrefetchWindow.exchange(window) == window)
+            return;
+
+        std::string url = m_filename;
+        if (evUseUploadedMovieForStreaming.getValue())
+            boost::replace_all(url, "#.mp4", "");
+        url = "shared:" + url;
+
+        StreamerPool::Options options;
+        options.emplace_back("cache_dir", evStreamCachePath.getValue());
+        options.emplace_back("seekable", "1");
+        options.emplace_back("reconnect", "1");
+        options.emplace_back("multiple_requests", "1");
+        for (const auto& [name, value] : m_request.parameters)
+        {
+            if (name == "cookies" || name == "headers")
+                options.emplace_back(name, value);
+        }
+
+        StreamerPool::getPool().enqueueWindow(url, options, frameSeconds, windowSeconds);
+    }
+
     bool MovieFFMpegReader::openAVFormat()
     {
         const bool filepathIsURL = TwkUtil::pathIsURL(m_filename);
@@ -2647,7 +2679,7 @@ namespace TwkMovie
             AVCodecContext* videoCodecContext = track->avCodecContext;
 
             // Tell RV to restrict caching to one thread
-            bool slowTrackRandomAccess = (codecHasSlowAccess(videoCodecContext->codec->name) || TwkUtil::pathIsURL(m_filename));
+            bool slowTrackRandomAccess = codecHasSlowAccess(videoCodecContext->codec->name);
             slowRandomAccess = slowTrackRandomAccess || slowRandomAccess;
 
             // Make sure the orientation/rotation matches for each track

src/lib/image/MovieFFMpeg/MovieFFMpeg/MovieFFMpeg.h

@@ -10,6 +10,7 @@
 #include <TwkMovie/MovieReader.h>
 #include <TwkMovie/MovieWriter.h>
 #include <TwkMovie/MovieIO.h>
+#include <atomic>
 #include <stdint.h>
 extern "C"
 {
@@ -158,6 +159,7 @@ namespace TwkMovie
         // Used for streamed media only to establish connections at the launch of RV for movie reader clones
         //
         void warmOpen() override;
+        void prefetchAtFrame(int frame) override;
 
         virtual bool canConvertAudioChannels() const;
         void close();
@@ -279,6 +281,7 @@ namespace TwkMovie
         bool m_multiTrackAudio;
         AudioState* m_audioState;
         bool m_cloning{false};
+        std::atomic<int64_t> m_lastPrefetchWindow{-1};
         bool m_mustReadFirstFrame{false};
         AVPixelFormat m_pxlFormatOnOpen{AV_PIX_FMT_NONE};
 

src/lib/image/TwkMovie/MovieReader.cpp

@@ -38,6 +38,8 @@ namespace TwkMovie
 
     void MovieReader::warmOpen() {}
 
+    void MovieReader::prefetchAtFrame(int) {}
+
     Movie* MovieReader::clone() const { return 0; }
 
     float MovieReader::scanProgress() const { return 1.0; }

src/lib/image/TwkMovie/TwkMovie/MovieReader.h

@@ -132,6 +132,8 @@ namespace TwkMovie
 
         virtual void warmOpen();
 
+        virtual void prefetchAtFrame(int frame);
+
         ///
         /// if an additional scanning pass is required after open this
         /// should return true from needsScan() while that is the case.

src/lib/image/stream/StreamPreloadPool.cpp

@@ -11,11 +11,15 @@
 
 #include <TwkUtil/EnvVar.h>
 
+#include <algorithm>
+
 extern "C"
 {
+#include <libavformat/avformat.h>
 #include <libavformat/avio.h>
 #include <libavutil/dict.h>
 #include <libavutil/error.h>
+#include <libavutil/mathematics.h>
 }
 
 // 32 KB
@@ -72,6 +76,35 @@ void StreamerPool::enqueue(const std::string& url, const Options& options)
     m_wake.notify_all();
 }
 
+void StreamerPool::enqueueWindow(const std::string& url, const Options& options, double startSeconds, double durationSeconds)
+{
+    {
+        std::lock_guard<std::mutex> lock(m_mutex);
+
+        if (m_stopped)
+        {
+            return;
+        }
+
+        m_queue.erase(std::remove_if(m_queue.begin(), m_queue.end(), [](const Job& job) { return job.window; }), m_queue.end());
+
+        Job job;
+        job.url = url;
+        job.options = options;
+        job.window = true;
+        job.startSeconds = startSeconds;
+        job.durationSeconds = durationSeconds;
+        m_queue.push_front(job);
+
+        if (!m_scheduler.joinable())
+        {
+            m_scheduler = std::thread(&StreamerPool::schedulerLoop, this);
+        }
+    }
+
+    m_wake.notify_all();
+}
+
 void StreamerPool::schedulerLoop()
 {
     for (;;)
@@ -82,18 +115,32 @@ void StreamerPool::schedulerLoop()
                     [this]
                     {
                         reapFinished();
-                        return m_stopped || (!m_queue.empty() && m_activeWorkers.size() < m_maxThreads);
+                        return m_stopped
+                               || (!m_queue.empty()
+                                   && (m_queue.front().window ? m_activeWindowWorkers == 0 : m_activeWorkers.size() < m_maxThreads));
                     });
 
         if (m_stopped)
         {
             return;
         }
 
-        while (!m_queue.empty() && m_activeWorkers.size() < m_maxThreads)
+        while (!m_queue.empty())
         {
+            if (m_queue.front().window)
+            {
+                if (m_activeWindowWorkers != 0)
+                    break;
+            }
+            else if (m_activeWorkers.size() >= m_maxThreads)
+            {
+                break;
+            }
+
             Job job = m_queue.front();
             m_queue.pop_front();
+            if (job.window)
+                ++m_activeWindowWorkers;
             m_activeWorkers.emplace_back(&StreamerPool::workerFunc, this, job);
         }
     }
@@ -120,10 +167,15 @@ void StreamerPool::reapFinished()
 
 void StreamerPool::workerFunc(Job job)
 {
-    download(job);
+    if (job.window)
+        downloadWindow(job);
+    else
+        download(job);
 
     {
         std::lock_guard<std::mutex> lock(m_mutex);
+        if (job.window)
+            --m_activeWindowWorkers;
         m_finished.insert(std::this_thread::get_id());
     }
 
@@ -175,6 +227,58 @@ void StreamerPool::download(const Job& job)
     avio_closep(&context);
 }
 
+void StreamerPool::downloadWindow(const Job& job)
+{
+    AVDictionary* options = nullptr;
+    for (const auto& [key, value] : job.options)
+    {
+        av_dict_set(&options, key.c_str(), value.c_str(), 0);
+    }
+
+    AVFormatContext* context = avformat_alloc_context();
+    if (context == nullptr)
+    {
+        av_dict_free(&options);
+        return;
+    }
+
+    context->interrupt_callback.callback = &StreamerPool::interruptCallback;
+    context->interrupt_callback.opaque = this;
+
+    if (avformat_open_input(&context, job.url.c_str(), nullptr, &options) < 0)
+    {
+        av_dict_free(&options);
+        avformat_free_context(context);
+        return;
+    }
+    av_dict_free(&options);
+
+    if (avformat_find_stream_info(context, nullptr) >= 0)
+    {
+        const int64_t mediaStart = context->start_time == AV_NOPTS_VALUE ? 0 : context->start_time;
+        const int64_t start = mediaStart + static_cast<int64_t>(job.startSeconds * AV_TIME_BASE);
+        const int64_t end = start + static_cast<int64_t>(job.durationSeconds * AV_TIME_BASE);
+
+        if (avformat_seek_file(context, -1, INT64_MIN, start, INT64_MAX, AVSEEK_FLAG_BACKWARD) >= 0)
+        {
+            AVPacket* packet = av_packet_alloc();
+            while (packet != nullptr && !m_abort.load() && av_read_frame(context, packet) >= 0)
+            {
+                const int64_t timestamp = packet->pts != AV_NOPTS_VALUE ? packet->pts : packet->dts;
+                const int64_t packetTime = timestamp == AV_NOPTS_VALUE
+                                               ? start
+                                               : av_rescale_q(timestamp, context->streams[packet->stream_index]->time_base, AV_TIME_BASE_Q);
+                av_packet_unref(packet);
+                if (packetTime >= end)
+                    break;
+            }
+            av_packet_free(&packet);
+        }
+    }
+
+    avformat_close_input(&context);
+}
+
 void StreamerPool::shutdown()
 {
     {

src/lib/image/stream/stream/StreamPreloadPool.h

@@ -48,6 +48,10 @@ class StreamerPool
     // Wakes
     void enqueue(const std::string& url, const Options& options);
 
+    // Prefetch a short demuxed window without decoding it. Window jobs use
+    // one reserved worker and supersede older queued window jobs.
+    void enqueueWindow(const std::string& url, const Options& options, double startSeconds, double durationSeconds);
+
     //
     //  Interrupt every in flight download, drop whatever is still queued
     //  and join the workers.
@@ -63,6 +67,9 @@ class StreamerPool
     {
         std::string url;
         Options options;
+        bool window = false;
+        double startSeconds = 0.0;
+        double durationSeconds = 0.0;
     };
 
     //
@@ -76,6 +83,7 @@ class StreamerPool
     // Handles FFMPEG API calls and fully downloads the raw media of the file
     //
     void download(const Job& job);
+    void downloadWindow(const Job& job);
 
     //
     // Checks which workers have finished, waits for the threads to finish for sure,
@@ -102,6 +110,7 @@ class StreamerPool
     std::thread m_scheduler;
     std::list<std::thread> m_activeWorkers;
     std::set<std::thread::id> m_finished;
+    size_t m_activeWindowWorkers = 0;
 
     std::deque<Job> m_queue; // pending jobs if threads maxed out
     bool m_stopped;          // scheduler finished

src/lib/ip/IPBaseNodes/FileSourceIPNode.cpp

@@ -58,7 +58,7 @@
     }
 
 static ENVVAR_BOOL(evIgnoreAudio, "RV_IGNORE_AUDIO", false);
-static ENVVAR_INT(evWarmCloneThreads, "RV_WARM_CLONE_THREADS", 4);
+static ENVVAR_INT(evWarmCloneThreads, "RV_WARM_CLONE_THREADS", 2);
 static ENVVAR_BOOL(evDebugCookies, "RV_DEBUG_FFMPEG_COOKIES", false);
 static ENVVAR_BOOL(evDebugHeaders, "RV_DEBUG_FFMPEG_HEADERS", false);
 
@@ -1066,6 +1066,12 @@ namespace IPCore
 
         PROFILE_SAMPLE(profile, ioEnd);
 
+        if (context.thread & DisplayThread)
+        {
+            if (MovieReader* reader = dynamic_cast<MovieReader*>(mov))
+                reader->prefetchAtFrame(request.frame);
+        }
+
         if (failed || empty)
         {
             Movie::IdentifierVector ids;