ffmpeg_demux: add option readrate_catchup

At present, if reading from a readrate-limited input is stalled,
then upon resumption, ffmpeg will read the input without any
throttle till the average readrate matches the specified readrate.

This new option allows to set a speed limit when reading is resumed
until the average readrate matches the primary readrate.

Fixes #11469

doc/ffmpeg.texi

@@ -2325,6 +2325,11 @@ Read input at native frame rate. This is equivalent to setting @code{-readrate 1
 @item -readrate_initial_burst @var{seconds}
 Set an initial read burst time, in seconds, after which @option{-re/-readrate}
 will be enforced.
+@item -readrate_catchup @var{speed} (@emph{input})
+If either the input or output is blocked leading to actual read speed falling behind the
+specified readrate, then this rate takes effect till the input catches up with the
+specified readrate. Must not be lower than the primary readrate.
+
 @item -vsync @var{parameter} (@emph{global})
 @itemx -fps_mode[:@var{stream_specifier}] @var{parameter} (@emph{output,per-stream})
 Set video sync method / framerate mode. vsync is applied to all output video streams

fftools/ffmpeg.h

@@ -163,6 +163,7 @@ typedef struct OptionsContext {
     int loop;
     int rate_emu;
     float readrate;
+    float readrate_catchup;
     double readrate_initial_burst;
     int accurate_seek;
     int thread_queue_size;

fftools/ffmpeg_demux.c

@@ -94,6 +94,12 @@ typedef struct DemuxStream {
     uint64_t                 nb_packets;
     // combined size of all the packets read
     uint64_t                 data_size;
+    // latest wallclock time at which packet reading resumed after a stall - used for readrate
+    int64_t                  resume_wc;
+    // timestamp of first packet sent after the latest stall - used for readrate
+    int64_t                  resume_pts;
+    // measure of how far behind packet reading is against spceified readrate
+    int64_t                  lag;
 } DemuxStream;
 
 typedef struct Demuxer {
@@ -127,6 +133,7 @@ typedef struct Demuxer {
 
     float                 readrate;
     double                readrate_initial_burst;
+    float                 readrate_catchup;
 
     Scheduler            *sch;
 
@@ -495,16 +502,42 @@ static void readrate_sleep(Demuxer *d)
                           (f->start_time_effective != AV_NOPTS_VALUE ? f->start_time_effective * !start_at_zero : 0) +
                           (f->start_time != AV_NOPTS_VALUE ? f->start_time : 0)
                          );
-    int64_t burst_until = AV_TIME_BASE * d->readrate_initial_burst;
+    int64_t initial_burst = AV_TIME_BASE * d->readrate_initial_burst;
+    int resume_warn;
+
     for (int i = 0; i < f->nb_streams; i++) {
         InputStream *ist = f->streams[i];
         DemuxStream  *ds = ds_from_ist(ist);
-        int64_t stream_ts_offset, pts, now;
+        int64_t stream_ts_offset, pts, now, wc_elapsed, elapsed, lag, max_pts, limit_pts;
+
+        if (ds->discard) continue;
+
         stream_ts_offset = FFMAX(ds->first_dts != AV_NOPTS_VALUE ? ds->first_dts : 0, file_start);
         pts = av_rescale(ds->dts, 1000000, AV_TIME_BASE);
-        now = (av_gettime_relative() - d->wallclock_start) * d->readrate + stream_ts_offset;
-        if (pts - burst_until > now)
-            av_usleep(pts - burst_until - now);
+        now = av_gettime_relative();
+        wc_elapsed = now - d->wallclock_start;
+        max_pts = stream_ts_offset + initial_burst + wc_elapsed * d->readrate;
+        lag = FFMAX(max_pts - pts, 0);
+        if ( (!ds->lag && lag > 0.3 * AV_TIME_BASE) || ( lag > ds->lag + 0.3 * AV_TIME_BASE) ) {
+            ds->lag = lag;
+            ds->resume_wc = now;
+            ds->resume_pts = pts;
+            av_log_once(ds, AV_LOG_WARNING, AV_LOG_DEBUG, &resume_warn,
+                        "Resumed reading at pts %0.3f with rate %0.3f after a lag of %0.3fs\n",
+                        (float)pts/AV_TIME_BASE, d->readrate_catchup, (float)lag/AV_TIME_BASE);
+        }
+        if (ds->lag && !lag)
+            ds->lag = ds->resume_wc = ds->resume_pts = 0;
+        if (ds->resume_wc) {
+            elapsed = now - ds->resume_wc;
+            limit_pts = ds->resume_pts + elapsed * d->readrate_catchup;
+        } else {
+            elapsed = wc_elapsed;
+            limit_pts = max_pts;
+        }
+
+        if (pts > limit_pts)
+            av_usleep(pts - limit_pts);
     }
 }
 
@@ -1859,9 +1892,22 @@ int ifile_open(const OptionsContext *o, const char *filename, Scheduler *sch)
                    d->readrate_initial_burst);
             return AVERROR(EINVAL);
         }
-    } else if (o->readrate_initial_burst) {
-        av_log(d, AV_LOG_WARNING, "Option -readrate_initial_burst ignored "
-               "since neither -readrate nor -re were given\n");
+        d->readrate_catchup = o->readrate_catchup ? o->readrate_catchup : d->readrate;
+        if (d->readrate_catchup < d->readrate) {
+            av_log(d, AV_LOG_ERROR,
+                   "Option -readrate_catchup is %0.3f; it must be at least equal to %0.3f.\n",
+                   d->readrate_catchup, d->readrate);
+            return AVERROR(EINVAL);
+        }
+    } else {
+        if (o->readrate_initial_burst) {
+            av_log(d, AV_LOG_WARNING, "Option -readrate_initial_burst ignored "
+                   "since neither -readrate nor -re were given\n");
+        }
+        if (o->readrate_catchup) {
+            av_log(d, AV_LOG_WARNING, "Option -readrate_catchup ignored "
+                   "since neither -readrate nor -re were given\n");
+        }
     }
 
     /* Add all the streams from the given input file to the demuxer */

fftools/ffmpeg_opt.c

@@ -1639,6 +1639,9 @@ const OptionDef options[] = {
     { "readrate_initial_burst", OPT_TYPE_DOUBLE, OPT_OFFSET | OPT_EXPERT | OPT_INPUT,
         { .off = OFFSET(readrate_initial_burst) },
         "The initial amount of input to burst read before imposing any readrate", "seconds" },
+    { "readrate_catchup",       OPT_TYPE_FLOAT, OPT_OFFSET | OPT_EXPERT | OPT_INPUT,
+        { .off = OFFSET(readrate_catchup) },
+        "Temporary readrate used to catch up if an input lags behind the specified readrate", "speed" },
     { "target",                 OPT_TYPE_FUNC, OPT_FUNC_ARG | OPT_PERFILE | OPT_EXPERT | OPT_OUTPUT,
         { .func_arg = opt_target },
         "specify target file type (\"vcd\", \"svcd\", \"dvd\", \"dv\" or \"dv50\" "