demux, command: add a third stream recording mechanism

That's right, and it's probably not the end of it. I'll just claim that
I have no idea how to create a proper user interface for this, so I'm
creating multiple partially-orthogonal, of which some may work better in
each of its special use cases.

Until now, there was --record-file. You get relatively good control
about what is muxed, and it can use the cache. But it sucks that it's
bound to playback. If you pause while it's set, muxing stops. If you
seek while it's set, the output will be sort-of trashed, and that's by
design.

Then --stream-record was added. This is a bit better (especially for
live streams), but you can't really control well when muxing stops or
ends. In particular, it can't use the cache (it just dumps whatever the
underlying demuxer returns).

Today, the idea is that the user should just be able to select a time
range to dump to a file, and it should not affected by the user seeking
around in the cache. In addition, the stream may still be running, so
there's some need to continue dumping, even if it's redundant to
--stream-record.

One notable thing is that it uses the async command shit. Not sure
whether this is a good idea. Maybe not, but whatever. Also, a user can
always use the "async" prefix to pretend it doesn't.

Much of this was barely tested (especially the reinterleaving crap),
let's just hope it mostly works. I'm sure you can tolerate the one or
other crash?

8 files changed, 383 insertions, 7 deletions

diff --git a/DOCS/man/input.rst b/DOCS/man/input.rst
index 4944c878ff..7c40a5cd43 100644
--- a/DOCS/man/input.rst
+++ b/DOCS/man/input.rst
@@ -920,6 +920,60 @@ Input Commands that are Possibly Subject to Change
         equivalent is ``--glsl-shaders-append=file.glsl`` or alternatively
         ``--glsl-shader=file.glsl``.
 
+``dump-cache <start> <end> <filename>``
+    Dump the current cache to the given filename. The ``<filename>`` file is
+    overwritten if it already exists. ``<start>`` and ``<end>`` give the
+    time range of what to dump. If no data is cached at the given time range,
+    nothing may be dumped (creating a file with no packets).
+
+    Dumping a larger part of the cache will freeze the player. No effort was
+    made to fix this, as this feature was meant mostly for creating small
+    excerpts.
+
+    If ``<filename>`` is an empty string, an ongoing ``dump-cache`` is stopped.
+
+    If ``<end>`` is ``no``, then continuous dumping is enabled. Then, after
+    dumping the existing parts of the cache, anything read from network is
+    appended to the cache as well. This behaves similar to ``--stream-record``
+    (although it does not conflict with that option, and they can be both active
+    at the same time).
+
+    If the ``<end>`` time is after the cache, the command will _not_ wait and
+    write newly received data to it.
+
+    The end of the resulting file may be slightly damaged or incomplete at the
+    end. (Not enough effort was made to ensure that the end lines up properly.)
+
+    Note that this command will finish only once dumping ends. That means it
+    works similar to the ``screenshot`` command, just that it can block much
+    longer. If continuous dumping is used, the command will not finish until
+    playback is stopped, an error happens, another ``dump-cache`` command is
+    run, or an API like ``mp.abort_async_command`` was called to explicitly stop
+    the command. See `Synchronous vs. Asynchronous`_.
+
+    .. note::
+
+        This was mostly created for network streams. For local files, there may
+        be much better methods to create excerpts and such. There are tons of
+        much more user-friendly Lua scripts, that will reencode parts of a file
+        by spawning a separate instance of ``ffmpeg``. With network streams,
+        this is not that easily possible, as the stream would have to be
+        downloaded again. Even if ``--stream-record`` is used to record the
+        stream to the local filesystem, there may be problems, because the
+        recorded file is still written to.
+
+    This command is experimental, and all details about it may change in the
+    future.
+
+``ab-loop-dump-cache <filename>``
+    Essentially calls ``dump-cache`` with the current AB-loop points as
+    arguments. Like ``dump-cache``, this will overwrite the file at
+    ``<filename>``. Likewise, if the B point is set to ``no``, it will enter
+    continuous dumping after the existing cache was dumped.
+
+    The author reserves the right to remove this command if enough motivation
+    is found to move this functionality to a trivial Lua script.
+
 
 Undocumented commands: ``ao-reload`` (experimental/internal).
 
@@ -1087,7 +1141,8 @@ default now, and ``async`` changes behavior only in the ways mentioned above.
 
 Currently the following commands have different waiting characteristics with
 sync vs. async: sub-add, audio-add, sub-reload, audio-reload,
-rescan-external-files, screenshot, screenshot-to-file.
+rescan-external-files, screenshot, screenshot-to-file, dump-cache,
+ab-loop-dump-cache.
 
 Input Sections
 --------------
diff --git a/DOCS/man/options.rst b/DOCS/man/options.rst
index ea47fc79b0..b590526f3e 100644
--- a/DOCS/man/options.rst
+++ b/DOCS/man/options.rst
@@ -5880,10 +5880,8 @@ Miscellaneous
 
     If this is set at runtime, the old file is closed, and the new file is
     opened. Note that this will write only data that is appended at the end of
-    the cache, and the already cached data cannot be written. (A fix for that
-    would be a command that dumps the cache using a given time range, possibly
-    with the option to be open-ended, which would continue to write data
-    appended to the cache. Such a command doesn't exist yet.)
+    the cache, and the already cached data cannot be written. You can try the
+    ``dump-cache`` command as an alternative.
 
 ``--lavfi-complex=<string>``
     Set a "complex" libavfilter filter, which means a single filter graph can
diff --git a/demux/demux.c b/demux/demux.c
index 6c03c71930..19a57a7c32 100644
--- a/demux/demux.c
+++ b/demux/demux.c
@@ -260,6 +260,9 @@ struct demux_internal {
     bool force_metadata_update;
     int cached_metadata_index;  // speed up repeated lookups
 
+    struct mp_recorder *dumper;
+    int dumper_status;
+
     // -- Access from demuxer thread only
     bool enable_recording;
     struct mp_recorder *recorder;
@@ -380,6 +383,10 @@ struct demux_stream {
     bool attached_picture_added;
     bool need_wakeup;       // call wakeup_cb on next reader_head state change
 
+    // For demux_internal.dumper. Currently, this is used only temporarily
+    // during blocking dumping.
+    struct demux_packet *dump_pos;
+
     // for refresh seeks: pos/dts of last packet returned to reader
     int64_t last_ret_pos;
     double last_ret_dts;
@@ -430,6 +437,7 @@ static void find_backward_restart_pos(struct demux_stream *ds);
 static struct demux_packet *find_seek_target(struct demux_queue *queue,
                                              double pts, int flags);
 static void prune_old_packets(struct demux_internal *in);
+static void dumper_close(struct demux_internal *in);
 
 static uint64_t get_foward_buffered_bytes(struct demux_stream *ds)
 {
@@ -1060,6 +1068,8 @@ static void demux_shutdown(struct demux_internal *in)
         in->recorder = NULL;
     }
 
+    dumper_close(in);
+
     if (demuxer->desc->close)
         demuxer->desc->close(in->d_thread);
     demuxer->priv = NULL;
@@ -1892,6 +1902,20 @@ static void adjust_seek_range_on_packet(struct demux_stream *ds,
     }
 }
 
+static void write_dump_packet(struct demux_internal *in, struct demux_packet *dp)
+{
+    assert(in->dumper);
+    assert(in->dumper_status == CONTROL_TRUE);
+
+    struct mp_recorder_sink *sink = mp_recorder_get_sink(in->dumper, dp->stream);
+    if (sink) {
+        mp_recorder_feed_packet(sink, dp);
+    } else {
+        MP_ERR(in, "New stream appeared; stopping recording.\n");
+        in->dumper_status = CONTROL_ERROR;
+    }
+}
+
 static void record_packet(struct demux_internal *in, struct demux_packet *dp)
 {
     // (should preferably be outside of the lock)
@@ -1920,6 +1944,9 @@ static void record_packet(struct demux_internal *in, struct demux_packet *dp)
             in->recorder = NULL;
         }
     }
+
+    if (in->dumper_status == CONTROL_OK)
+        write_dump_packet(in, dp);
 }
 
 static void add_packet_locked(struct sh_stream *stream, demux_packet_t *dp)
@@ -3970,6 +3997,183 @@ static void update_cache(struct demux_internal *in)
     pthread_mutex_unlock(&in->lock);
 }
 
+static void dumper_close(struct demux_internal *in)
+{
+    if (in->dumper)
+        mp_recorder_destroy(in->dumper);
+    in->dumper = NULL;
+    if (in->dumper_status == CONTROL_TRUE)
+        in->dumper_status = CONTROL_FALSE; // make abort equal to success
+}
+
+static int range_time_compare(const void *p1, const void *p2)
+{
+    struct demux_cached_range *r1 = (void *)p1;
+    struct demux_cached_range *r2 = (void *)p2;
+
+    if (r1->seek_start == r2->seek_start)
+        return 0;
+    return r1->seek_start < r2->seek_start ? -1 : 1;
+}
+
+static void dump_cache(struct demux_internal *in, double start, double end)
+{
+    in->dumper_status = in->dumper ? CONTROL_TRUE : CONTROL_ERROR;
+    if (!in->dumper)
+        return;
+
+    // (only in pathological cases there might be more ranges than allowed)
+    struct demux_cached_range *ranges[MAX_SEEK_RANGES];
+    int num_ranges = 0;
+    for (int n = 0; n < MPMIN(MP_ARRAY_SIZE(ranges), in->num_ranges); n++)
+        ranges[num_ranges++] = in->ranges[n];
+    qsort(ranges, num_ranges, sizeof(ranges[0]), range_time_compare);
+
+    for (int n = 0; n < num_ranges; n++) {
+        struct demux_cached_range *r = ranges[n];
+        if (r->seek_start == MP_NOPTS_VALUE)
+            continue;
+        if (r->seek_end <= start)
+            continue;
+        if (end != MP_NOPTS_VALUE && r->seek_start >= end)
+            continue;
+
+        mp_recorder_mark_discontinuity(in->dumper);
+
+        double pts = start;
+        int flags = 0;
+        adjust_cache_seek_target(in, r, &pts, &flags);
+
+        for (int i = 0; i < r->num_streams; i++) {
+            struct demux_queue *q = r->streams[i];
+            struct demux_stream *ds = q->ds;
+
+            ds->dump_pos = find_seek_target(q, pts, flags);
+        }
+
+        // We need to reinterleave the separate streams somehow, which makes
+        // everything more complex.
+        while (1) {
+            struct demux_packet *next = NULL;
+            double next_dts = MP_NOPTS_VALUE;
+
+            for (int i = 0; i < r->num_streams; i++) {
+                struct demux_stream *ds = r->streams[i]->ds;
+                struct demux_packet *dp = ds->dump_pos;
+
+                if (!dp)
+                    continue;
+                assert(dp->stream == ds->index);
+
+                double pdts = MP_PTS_OR_DEF(dp->dts, dp->pts);
+
+                // Check for stream EOF. Note that we don't try to EOF
+                // streams at the same point (e.g. video can take longer
+                // to finish than audio, so the output file will have no
+                // audio for the last part of the video). Too much effort.
+                if (pdts != MP_NOPTS_VALUE && end != MP_NOPTS_VALUE &&
+                    pdts >= end && dp->keyframe)
+                {
+                    ds->dump_pos = NULL;
+                    continue;
+                }
+
+                if (pdts == MP_NOPTS_VALUE || next_dts == MP_NOPTS_VALUE ||
+                    pdts < next_dts)
+                {
+                    next_dts = pdts;
+                    next = dp;
+                }
+            }
+
+            if (!next)
+                break;
+
+            struct demux_stream *ds = in->streams[next->stream]->ds;
+            ds->dump_pos = next->next;
+
+            struct demux_packet *dp = read_packet_from_cache(in, next);
+            if (!dp) {
+                in->dumper_status = CONTROL_ERROR;
+                break;
+            }
+
+            write_dump_packet(in, dp);
+
+            talloc_free(dp);
+        }
+
+        if (in->dumper_status != CONTROL_OK)
+            break;
+    }
+
+    // (strictly speaking unnecessary; for clarity)
+    for (int n = 0; n < in->num_streams; n++)
+        in->streams[n]->ds->dump_pos = NULL;
+
+    // If dumping (in end==NOPTS mode) doesn't continue at the range that
+    // was written last, we have a discontinuity.
+    if (num_ranges && ranges[num_ranges - 1] != in->current_range)
+        mp_recorder_mark_discontinuity(in->dumper);
+
+    // end=NOPTS means the demuxer output continues to be written to the
+    // dump file.
+    if (end != MP_NOPTS_VALUE || in->dumper_status != CONTROL_OK)
+        dumper_close(in);
+}
+
+// Set the current cache dumping mode. There is only at most 1 dump process
+// active, so calling this aborts the previous dumping. Passing file==NULL
+// stops dumping.
+// This is synchronous with demux_cache_dump_get_status() (i.e. starting or
+// aborting is not asynchronous). On status change, the demuxer wakeup callback
+// is invoked (except for this call).
+// Returns whether dumping was logically started.
+bool demux_cache_dump_set(struct demuxer *demuxer, double start, double end,
+                          char *file)
+{
+    struct demux_internal *in = demuxer->in;
+    assert(demuxer == in->d_user);
+
+    bool res = false;
+
+    pthread_mutex_lock(&in->lock);
+
+    start = MP_ADD_PTS(start, -in->ts_offset);
+    end = MP_ADD_PTS(end, -in->ts_offset);
+
+    dumper_close(in);
+
+    if (file && file[0] && start != MP_NOPTS_VALUE) {
+        res = true;
+
+        in->dumper = mp_recorder_create(in->d_thread->global, file,
+                                        in->streams, in->num_streams);
+
+        // This is not asynchronous and will freeze the shit for a while if the
+        // user is unlucky. It could be moved to a thread with some effort.
+        // General idea: iterate over all cache ranges, dump what intersects.
+        // After that, and if the user requested it, make it dump all newly
+        // received packets, even if it's awkward (consider the case if the
+        // current range is not the last range).
+        dump_cache(in, start, end);
+    }
+
+    pthread_mutex_unlock(&in->lock);
+
+    return res;
+}
+
+// Returns one of CONTROL_*. CONTROL_TRUE means dumping is in progress.
+int demux_cache_dump_get_status(struct demuxer *demuxer)
+{
+    struct demux_internal *in = demuxer->in;
+    pthread_mutex_lock(&in->lock);
+    int status = in->dumper_status;
+    pthread_mutex_unlock(&in->lock);
+    return status;
+}
+
 // Used by demuxers to report the amount of transferred bytes. This is for
 // streams which circumvent demuxer->stream (stream statistics are handled by
 // demux.c itself).
diff --git a/demux/demux.h b/demux/demux.h
index 14d145704a..3716f28b87 100644
--- a/demux/demux.h
+++ b/demux/demux.h
@@ -292,6 +292,10 @@ void demux_close_stream(struct demuxer *demuxer);
 void demux_metadata_changed(demuxer_t *demuxer);
 void demux_update(demuxer_t *demuxer, double playback_pts);
 
+bool demux_cache_dump_set(struct demuxer *demuxer, double start, double end,
+                          char *file);
+int demux_cache_dump_get_status(struct demuxer *demuxer);
+
 bool demux_is_network_cached(demuxer_t *demuxer);
 
 void demux_report_unbuffered_read_bytes(struct demuxer *demuxer, int64_t new);
diff --git a/player/command.c b/player/command.c
index 2472aab5e9..a2bdb19585 100644
--- a/player/command.c
+++ b/player/command.c
@@ -63,6 +63,7 @@
 #include "misc/dispatch.h"
 #include "misc/node.h"
 #include "misc/thread_pool.h"
+#include "misc/thread_tools.h"
 
 #include "osdep/io.h"
 #include "osdep/subprocess.h"
@@ -104,6 +105,8 @@ struct command_ctx {
     char *cur_ipc_input;
 
     int silence_option_deprecations;
+
+    struct mp_cmd_ctx *cache_dump_cmd; // in progress cache dumping
 };
 
 struct overlay {
@@ -5560,6 +5563,91 @@ static void cmd_load_script(void *p)
         cmd->success = false;
 }
 
+static void cache_dump_poll(struct MPContext *mpctx)
+{
+    struct command_ctx *ctx = mpctx->command_ctx;
+    struct mp_cmd_ctx *cmd = ctx->cache_dump_cmd;
+
+    if (!cmd)
+        return;
+
+    // Can't close demuxer without stopping dumping.
+    assert(mpctx->demuxer);
+
+    if (mp_cancel_test(cmd->abort->cancel)) {
+        // Synchronous abort. In particular, the dump command shall not report
+        // completion to the user before the dump target file was closed.
+        demux_cache_dump_set(mpctx->demuxer, 0, 0, NULL);
+        assert(demux_cache_dump_get_status(mpctx->demuxer) <= 0);
+    }
+
+    int status = demux_cache_dump_get_status(mpctx->demuxer);
+    if (status <= 0) {
+        if (status < 0) {
+            mp_cmd_msg(cmd, MSGL_ERR, "Cache dumping stopped due to error.");
+            cmd->success = false;
+        } else {
+            mp_cmd_msg(cmd, MSGL_INFO, "Cache dumping successfully ended.");
+            cmd->success = true;
+        }
+        ctx->cache_dump_cmd = NULL;
+        mp_cmd_ctx_complete(cmd);
+    }
+}
+
+void mp_abort_cache_dumping(struct MPContext *mpctx)
+{
+    struct command_ctx *ctx = mpctx->command_ctx;
+
+    if (ctx->cache_dump_cmd)
+        mp_cancel_trigger(ctx->cache_dump_cmd->abort->cancel);
+    cache_dump_poll(mpctx);
+    assert(!ctx->cache_dump_cmd); // synchronous abort, must have worked
+}
+
+static void run_dump_cmd(struct mp_cmd_ctx *cmd, double start, double end,
+                         char *filename)
+{
+    struct MPContext *mpctx = cmd->mpctx;
+    struct command_ctx *ctx = mpctx->command_ctx;
+
+    mp_abort_cache_dumping(mpctx);
+
+    if (!mpctx->demuxer) {
+        mp_cmd_msg(cmd, MSGL_ERR, "No demuxer open.");
+        cmd->success = false;
+        mp_cmd_ctx_complete(cmd);
+        return;
+    }
+
+    if (!demux_cache_dump_set(mpctx->demuxer, start, end, filename)) {
+        mp_cmd_msg(cmd, MSGL_INFO, "Cache dumping stopped.");
+        mp_cmd_ctx_complete(cmd);
+        return;
+    }
+
+    mp_cmd_msg(cmd, MSGL_INFO, "Cache dumping started.");
+
+    ctx->cache_dump_cmd = cmd;
+    cache_dump_poll(mpctx);
+}
+
+static void cmd_dump_cache(void *p)
+{
+    struct mp_cmd_ctx *cmd = p;
+
+    run_dump_cmd(cmd, cmd->args[0].v.d, cmd->args[1].v.d, cmd->args[2].v.s);
+}
+
+static void cmd_dump_cache_ab(void *p)
+{
+    struct mp_cmd_ctx *cmd = p;
+    struct MPContext *mpctx = cmd->mpctx;
+
+    run_dump_cmd(cmd, mpctx->opts->ab_loop[0], mpctx->opts->ab_loop[1],
+                 cmd->args[0].v.s);
+}
+
 /* This array defines all known commands.
  * The first field the command name used in libmpv and input.conf.
  * The second field is the handler function (see mp_cmd_def.handler and
@@ -5884,6 +5972,20 @@ const struct mp_cmd_def mp_cmds[] = {
 
     { "load-script", cmd_load_script, {OPT_STRING("filename", v.s, 0)} },
 
+    { "dump-cache", cmd_dump_cache, { OPT_TIME("start", v.d, 0,
+                                               .min = MP_NOPTS_VALUE),
+                                      OPT_TIME("end", v.d, 0,
+                                               .min = MP_NOPTS_VALUE),
+                                      OPT_STRING("filename", v.s, 0) },
+        .exec_async = true,
+        .can_abort = true,
+    },
+
+    { "ab-loop-dump-cache", cmd_dump_cache_ab, { OPT_STRING("filename", v.s, 0) },
+        .exec_async = true,
+        .can_abort = true,
+    },
+
     {0}
 };
 
@@ -5892,8 +5994,13 @@ const struct mp_cmd_def mp_cmds[] = {
 
 void command_uninit(struct MPContext *mpctx)
 {
+    struct command_ctx *ctx = mpctx->command_ctx;
+
+    assert(!ctx->cache_dump_cmd); // closing the demuxer must have aborted it
+
     overlay_uninit(mpctx);
-    ao_hotplug_destroy(mpctx->command_ctx->hotplug);
+    ao_hotplug_destroy(ctx->hotplug);
+
     talloc_free(mpctx->command_ctx);
     mpctx->command_ctx = NULL;
 }
@@ -5967,6 +6074,9 @@ void handle_command_updates(struct MPContext *mpctx)
     // to recheck the state. Then the client(s) will read the property.
     if (ctx->hotplug && ao_hotplug_check_update(ctx->hotplug))
         mp_notify_property(mpctx, "audio-device-list");
+
+    // Depends on polling demuxer wakeup callback notifications.
+    cache_dump_poll(mpctx);
 }
 
 void mp_notify(struct MPContext *mpctx, int event, void *arg)
diff --git a/player/command.h b/player/command.h
index 4cc774c6e8..3952ce37e7 100644
--- a/player/command.h
+++ b/player/command.h
@@ -110,4 +110,6 @@ void mp_hook_add(struct MPContext *mpctx, const char *client, const char *name,
 
 void mark_seek(struct MPContext *mpctx);
 
+void mp_abort_cache_dumping(struct MPContext *mpctx);
+
 #endif /* MPLAYER_COMMAND_H */
diff --git a/player/core.h b/player/core.h
index bf4b62fa56..893a4cf3e8 100644
--- a/player/core.h
+++ b/player/core.h
@@ -468,7 +468,8 @@ typedef struct MPContext {
 struct mp_abort_entry {
     // General conditions.
     bool coupled_to_playback;   // trigger when playback is terminated
-    // Actual trigger to abort the work.
+    // Actual trigger to abort the work. Pointer immutable, owner may access
+    // without holding the abort_lock.
     struct mp_cancel *cancel;
     // For client API.
     struct mpv_handle *client;  // non-NULL if done by a client API user
diff --git a/player/loadfile.c b/player/loadfile.c
index af84a82c99..ec84e8a003 100644
--- a/player/loadfile.c
+++ b/player/loadfile.c
@@ -201,6 +201,8 @@ static void uninit_demuxer(struct MPContext *mpctx)
     mpctx->chapters = NULL;
     mpctx->num_chapters = 0;
 
+    mp_abort_cache_dumping(mpctx);
+
     struct demuxer **demuxers = NULL;
     int num_demuxers = 0;