demux: add a on-disk cache

Somewhat similar to the old --cache-file, except for the demuxer cache.
Instead of keeping packet data in memory, it's written to disk and read
back when needed.

The idea is to reduce main memory usage, while allowing fast seeking in
large cached network streams (especially live streams). Keeping the
packet metadata on disk would be rather hard (would use mmap or so, or
rewrite the entire demux.c packet queue handling), and since it's
relatively small, just keep it in memory.

Also for simplicity, the disk cache is append-only. If you're watching
really long livestreams, and need pruning, you're probably out of luck.
This still could be improved by trying to free unused blocks with
fallocate(), but since we're writing multiple streams in an interleaved
manner, this is slightly hard.

Some rather gross ugliness in packet.h: we want to store the file
position of the cached data somewhere, but on 32 bit architectures, we
don't have any usable 64 bit members for this, just the buf/len fields,
which add up to 64 bit - so the shitty union aliases this memory.

Error paths untested. Side data (the complicated part of trying to
serialize ffmpeg packets) untested.

Stream recording had to be adjusted. Some minor details change due to
this, but probably nothing important.

The change in attempt_range_joining() is because packets in cache
have no valid len field. It was a useful check (heuristically
finding broken cases), but not a necessary one.

Various other approaches were tried. It would be interesting to list
them and to mention the pros and cons, but I don't feel like it.

6 files changed, 453 insertions, 39 deletions

diff --git a/demux/cache.c b/demux/cache.c
new file mode 100644
index 0000000000..4404c870de
--- /dev/null
+++ b/demux/cache.c
@@ -0,0 +1,323 @@
+/*
+ * This file is part of mpv.
+ *
+ * mpv is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * mpv is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with mpv.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include <errno.h>
+#include <fcntl.h>
+#include <stdlib.h>
+#include <sys/stat.h>
+#include <sys/types.h>
+#include <unistd.h>
+
+#include "cache.h"
+#include "common/msg.h"
+#include "common/av_common.h"
+#include "demux.h"
+#include "options/path.h"
+#include "options/m_config.h"
+#include "options/m_option.h"
+#include "osdep/io.h"
+
+struct demux_cache_opts {
+    char *cache_dir;
+    int unlink_files;
+};
+
+#define OPT_BASE_STRUCT struct demux_cache_opts
+
+const struct m_sub_options demux_cache_conf = {
+    .opts = (const struct m_option[]){
+        OPT_STRING("cache-dir", cache_dir, M_OPT_FILE),
+        OPT_CHOICE("cache-unlink-files", unlink_files, 0,
+                   ({"immediate", 2}, {"whendone", 1}, {"no", 0})),
+        {0}
+    },
+    .size = sizeof(struct demux_cache_opts),
+    .defaults = &(const struct demux_cache_opts){
+        .unlink_files = 2,
+    },
+};
+
+struct demux_cache {
+    struct mp_log *log;
+    struct demux_cache_opts *opts;
+
+    char *filename;
+    bool need_unlink;
+    int fd;
+    int64_t file_pos;
+    uint64_t file_size;
+};
+
+struct pkt_header {
+    uint32_t data_len;
+    uint32_t av_flags;
+    uint32_t num_sd;
+};
+
+struct sd_header {
+    uint32_t av_type;
+    uint32_t len;
+};
+
+static void cache_destroy(void *p)
+{
+    struct demux_cache *cache = p;
+
+    if (cache->fd >= 0)
+        close(cache->fd);
+
+    if (cache->need_unlink && cache->opts->unlink_files >= 1) {
+        if (unlink(cache->filename))
+            MP_ERR(cache, "Failed to delete cache temporary file.\n");
+    }
+}
+
+// Create a cache. This also initializes the cache file from the options. The
+// log parameter must stay valid until demux_cache is destroyed.
+// Free with talloc_free().
+struct demux_cache *demux_cache_create(struct mpv_global *global,
+                                       struct mp_log *log)
+{
+    struct demux_cache *cache = talloc_zero(NULL, struct demux_cache);
+    talloc_set_destructor(cache, cache_destroy);
+    cache->opts = mp_get_config_group(cache, global, &demux_cache_conf);
+    cache->log = log;
+    cache->fd = -1;
+
+    char *cache_dir = cache->opts->cache_dir;
+    if (!(cache_dir && cache_dir[0])) {
+        MP_ERR(cache, "No cache data directory supplied.\n");
+        goto fail;
+    }
+
+    cache->filename = mp_path_join(cache, cache_dir, "mpv-cache-XXXXXX.dat");
+    cache->fd = mp_mkostemps(cache->filename, 4, O_CLOEXEC);
+    if (cache->fd < 0) {
+        MP_ERR(cache, "Failed to create cache temporary file.\n");
+        goto fail;
+    }
+    cache->need_unlink = true;
+    if (cache->opts->unlink_files >= 2) {
+        if (unlink(cache->filename)) {
+            MP_ERR(cache, "Failed to unlink cache temporary file after creation.\n");
+        } else {
+            cache->need_unlink = false;
+        }
+    }
+
+    return cache;
+fail:
+    talloc_free(cache);
+    return NULL;
+}
+
+uint64_t demux_cache_get_size(struct demux_cache *cache)
+{
+    return cache->file_size;
+}
+
+static bool do_seek(struct demux_cache *cache, uint64_t pos)
+{
+    if (cache->file_pos == pos)
+        return true;
+
+    off_t res = lseek(cache->fd, pos, SEEK_SET);
+
+    if (res == (off_t)-1) {
+        MP_ERR(cache, "Failed to seek in cache file.\n");
+        cache->file_pos = -1;
+    } else {
+        cache->file_pos = res;
+    }
+
+    return cache->file_pos >= 0;
+}
+
+static bool write_raw(struct demux_cache *cache, void *ptr, size_t len)
+{
+    ssize_t res = write(cache->fd, ptr, len);
+
+    if (res < 0) {
+        MP_ERR(cache, "Failed to write to cache file: %s\n", mp_strerror(errno));
+        return false;
+    }
+
+    cache->file_pos += res;
+    cache->file_size = MPMAX(cache->file_size, cache->file_pos);
+
+    // Should never happen, unless the disk is full, or someone succeeded to
+    // trick us to write into a pipe or a socket.
+    if (res != len) {
+        MP_ERR(cache, "Could not write all data.\n");
+        return false;
+    }
+
+    return true;
+}
+
+static bool read_raw(struct demux_cache *cache, void *ptr, size_t len)
+{
+    ssize_t res = read(cache->fd, ptr, len);
+
+    if (res < 0) {
+        MP_ERR(cache, "Failed to read cache file: %s\n", mp_strerror(errno));
+        return false;
+    }
+
+    cache->file_pos += res;
+
+    // Should never happen, unless the file was cut short, or someone succeeded
+    // to rick us to write into a pipe or a socket.
+    if (res != len) {
+        MP_ERR(cache, "Could not read all data.\n");
+        return false;
+    }
+
+    return true;
+}
+
+// Serialize a packet to the cache file. Returns the packet position, which can
+// be passed to demux_cache_read() to read the packet again.
+// Returns a negative value on errors, i.e. writing the file failed.
+int64_t demux_cache_write(struct demux_cache *cache, struct demux_packet *dp)
+{
+    assert(dp->avpacket);
+
+    // AV_PKT_FLAG_TRUSTED usually means there are embedded pointers and such
+    // in the packet data. The pointer will become invalid if the packet is
+    // unreferenced.
+    if (dp->avpacket->flags & AV_PKT_FLAG_TRUSTED) {
+        MP_ERR(cache, "Cannot serialize this packet to cache file.\n");
+        return -1;
+    }
+
+    assert(!dp->is_cached);
+    assert(dp->len >= 0 && dp->len <= INT32_MAX);
+    assert(dp->avpacket->flags >= 0 && dp->avpacket->flags <= INT32_MAX);
+    assert(dp->avpacket->side_data_elems >= 0 &&
+           dp->avpacket->side_data_elems <= INT32_MAX);
+
+    if (!do_seek(cache, cache->file_size))
+        return -1;
+
+    uint64_t pos = cache->file_pos;
+
+    struct pkt_header hd = {
+        .data_len  = dp->len,
+        .av_flags = dp->avpacket->flags,
+        .num_sd = dp->avpacket->side_data_elems,
+    };
+
+    if (!write_raw(cache, &hd, sizeof(hd)))
+        goto fail;
+
+    if (!write_raw(cache, dp->buffer, dp->len))
+        goto fail;
+
+    // The handling of FFmpeg side data requires an extra long comment to
+    // explain why this code is fragile and insane.
+    // FFmpeg packet side data is per-packet out of band data, that contains
+    // further information for the decoder (extra metadata and such), which is
+    // not part of the codec itself and thus isn't contained in the packet
+    // payload. All types use a flat byte array. The format of this byte array
+    // is non-standard and FFmpeg-specific, and depends on the side data type
+    // field. The side data type is of course a FFmpeg ABI artifact.
+    // In some cases, the format is described as fixed byte layout. In others,
+    // it contains a struct, i.e. is bound to FFmpeg ABI. Some newer types make
+    // the format explicitly internal (and _not_ part of the ABI), and you need
+    // to use separate accessors to turn it into complex data structures.
+    // As of now, FFmpeg fortunately adheres to the idea that side data can not
+    // contain embedded pointers (due to API rules, but also because they forgot
+    // adding a refcount field, and can't change this until they break ABI).
+    // We rely on this. We hope that FFmpeg won't silently change their
+    // semantics, and add refcounting and embedded pointers. This way we can
+    // for example dump the data in a disk cache, even though we can't use the
+    // data from another process or if this process is restarted (unless we're
+    // absolutely sure the FFmpeg internals didn't change). The data has to be
+    // treated as a memory dump.
+    for (int n = 0; n < dp->avpacket->side_data_elems; n++) {
+        AVPacketSideData *sd = &dp->avpacket->side_data[n];
+
+        assert(sd->size >= 0 && sd->size <= INT32_MAX);
+        assert(sd->type >= 0 && sd->type <= INT32_MAX);
+
+        struct sd_header sd_hd = {
+            .av_type = sd->type,
+            .len = sd->size,
+        };
+
+        if (!write_raw(cache, &sd_hd, sizeof(sd_hd)))
+            goto fail;
+        if (!write_raw(cache, sd->data, sd->size))
+            goto fail;
+    }
+
+    return pos;
+
+fail:
+    // Reset file_size (try not to append crap forever).
+    do_seek(cache, pos);
+    cache->file_size = cache->file_pos;
+    return -1;
+}
+
+struct demux_packet *demux_cache_read(struct demux_cache *cache, uint64_t pos)
+{
+    if (!do_seek(cache, pos))
+        return NULL;
+
+    struct pkt_header hd;
+
+    if (!read_raw(cache, &hd, sizeof(hd)))
+        return NULL;
+
+    if (hd.data_len >= (size_t)-1)
+        return NULL;
+
+    struct demux_packet *dp = new_demux_packet(hd.data_len);
+    if (!dp)
+        goto fail;
+
+    if (!read_raw(cache, dp->buffer, dp->len))
+        goto fail;
+
+    dp->avpacket->flags = hd.av_flags;
+
+    for (uint32_t n = 0; n < hd.num_sd; n++) {
+        struct sd_header sd_hd;
+
+        if (!read_raw(cache, &sd_hd, sizeof(sd_hd)))
+            goto fail;
+
+        if (sd_hd.len > INT_MAX)
+            goto fail;
+
+        uint8_t *sd = av_packet_new_side_data(dp->avpacket, sd_hd.av_type,
+                                              sd_hd.len);
+        if (!sd)
+            goto fail;
+
+        if (!read_raw(cache, sd, sd_hd.len))
+            goto fail;
+    }
+
+    return dp;
+
+fail:
+    talloc_free(dp);
+    return NULL;
+}
diff --git a/demux/cache.h b/demux/cache.h
new file mode 100644
index 0000000000..95ea9649c0
--- /dev/null
+++ b/demux/cache.h
@@ -0,0 +1,16 @@
+#pragma once
+
+#include <stdint.h>
+
+struct demux_packet;
+struct mp_log;
+struct mpv_global;
+
+struct demux_cache;
+
+struct demux_cache *demux_cache_create(struct mpv_global *global,
+                                       struct mp_log *log);
+
+int64_t demux_cache_write(struct demux_cache *cache, struct demux_packet *pkt);
+struct demux_packet *demux_cache_read(struct demux_cache *cache, uint64_t pos);
+uint64_t demux_cache_get_size(struct demux_cache *cache);
diff --git a/demux/demux.c b/demux/demux.c
index 5fc81ad41e..b9101a3054 100644
--- a/demux/demux.c
+++ b/demux/demux.c
@@ -29,6 +29,7 @@
 #include <sys/types.h>
 #include <sys/stat.h>
 
+#include "cache.h"
 #include "config.h"
 #include "options/m_config.h"
 #include "options/m_option.h"
@@ -80,6 +81,7 @@ static const demuxer_desc_t *const demuxer_list[] = {
 
 struct demux_opts {
     int enable_cache;
+    int disk_cache;
     int64_t max_bytes;
     int64_t max_bytes_bw;
     double min_secs;
@@ -103,6 +105,7 @@ const struct m_sub_options demux_conf = {
     .opts = (const struct m_option[]){
         OPT_CHOICE("cache", enable_cache, 0,
                    ({"no", 0}, {"auto", -1}, {"yes", 1})),
+        OPT_FLAG("cache-on-disk", disk_cache, 0),
         OPT_DOUBLE("demuxer-readahead-secs", min_secs, M_OPT_MIN, .min = 0),
         // (The MAX_BYTES sizes may not be accurate because the max field is
         // of double type.)
@@ -179,6 +182,8 @@ struct demux_internal {
 
     int events;
 
+    struct demux_cache *cache;
+
     bool warned_queue_overflow;
     bool last_eof;              // last actual global EOF status
     bool eof;                   // whether we're in EOF state (reset for retry)
@@ -1039,6 +1044,10 @@ static void demux_shutdown(struct demux_internal *in)
 
     in->current_range = NULL;
     free_empty_cached_ranges(in);
+
+    talloc_free(in->cache);
+    in->cache = NULL;
+
     if (in->owns_stream)
         free_stream(demuxer->stream);
     demuxer->stream = NULL;
@@ -1612,9 +1621,10 @@ static void attempt_range_joining(struct demux_internal *in)
                     // in case pos/dts are not "correct" across the ranges (we
                     // never actually check that).
                     if (dp->dts != end->dts || dp->pos != end->pos ||
-                        dp->pts != end->pts || dp->len != end->len)
+                        dp->pts != end->pts)
                     {
-                        MP_WARN(in, "stream %d: weird demuxer behavior\n", n);
+                        MP_WARN(in,
+                            "stream %d: non-repeatable demuxer behavior\n", n);
                         goto failed;
                     }
 
@@ -1822,6 +1832,36 @@ static void adjust_seek_range_on_packet(struct demux_stream *ds,
     }
 }
 
+static void record_packet(struct demux_internal *in, struct demux_packet *dp)
+{
+    // (should preferably be outside of the lock)
+    if (in->enable_recording && !in->recorder &&
+        in->opts->record_file && in->opts->record_file[0])
+    {
+        // Later failures shouldn't make it retry and overwrite the previously
+        // recorded file.
+        in->enable_recording = false;
+
+        in->recorder =
+            mp_recorder_create(in->d_thread->global, in->opts->record_file,
+                               in->streams, in->num_streams);
+        if (!in->recorder)
+            MP_ERR(in, "Disabling recording.\n");
+    }
+
+    if (in->recorder) {
+        struct mp_recorder_sink *sink =
+            mp_recorder_get_sink(in->recorder, dp->stream);
+        if (sink) {
+            mp_recorder_feed_packet(sink, dp);
+        } else {
+            MP_ERR(in, "New stream appeared; stopping recording.\n");
+            mp_recorder_destroy(in->recorder);
+            in->recorder = NULL;
+        }
+    }
+}
+
 static void add_packet_locked(struct sh_stream *stream, demux_packet_t *dp)
 {
     struct demux_stream *ds = stream ? stream->ds : NULL;
@@ -1864,6 +1904,17 @@ static void add_packet_locked(struct sh_stream *stream, demux_packet_t *dp)
         return;
     }
 
+    record_packet(in, dp);
+
+    if (in->cache) {
+        int64_t pos = demux_cache_write(in->cache, dp);
+        if (pos >= 0) {
+            demux_packet_unref_contents(dp);
+            dp->is_cached = true;
+            dp->cached_data.pos = pos;
+        }
+    }
+
     queue->correct_pos &= dp->pos >= 0 && dp->pos > queue->last_pos;
     queue->correct_dts &= dp->dts != MP_NOPTS_VALUE && dp->dts > queue->last_dts;
     queue->last_pos = dp->pos;
@@ -1940,33 +1991,6 @@ static void add_packet_locked(struct sh_stream *stream, demux_packet_t *dp)
     if (!ds->reader_head)
         return;
 
-    // (should preferably be outside of the lock)
-    if (in->enable_recording && !in->recorder &&
-        in->opts->record_file && in->opts->record_file[0])
-    {
-        // Later failures shouldn't make it retry and overwrite the previously
-        // recorded file.
-        in->enable_recording = false;
-
-        in->recorder =
-            mp_recorder_create(in->d_thread->global, in->opts->record_file,
-                               in->streams, in->num_streams);
-        if (!in->recorder)
-            MP_ERR(in, "Disabling recording.\n");
-    }
-
-    if (in->recorder) {
-        struct mp_recorder_sink *sink =
-            mp_recorder_get_sink(in->recorder, dp->stream);
-        if (sink) {
-            mp_recorder_feed_packet(sink, dp);
-        } else {
-            MP_ERR(in, "New stream appeared; stopping recording.\n");
-            mp_recorder_destroy(in->recorder);
-            in->recorder = NULL;
-        }
-    }
-
     back_demux_see_packets(ds);
 
     wakeup_ds(ds);
@@ -2396,11 +2420,21 @@ static int dequeue_packet(struct demux_stream *ds, struct demux_packet **res)
     struct demux_packet *pkt = advance_reader_head(ds);
     assert(pkt);
 
-    // The returned packet is mutated etc. and will be owned by the user.
-    pkt = demux_copy_packet(pkt);
+    if (pkt->is_cached) {
+        assert(in->cache);
+        struct demux_packet *meta = pkt;
+        pkt = demux_cache_read(in->cache, pkt->cached_data.pos);
+        if (pkt) {
+            demux_packet_copy_attribs(pkt, meta);
+        } else {
+            MP_ERR(in, "Failed to retrieve packet from cache.\n");
+        }
+    } else {
+        // The returned packet is mutated etc. and will be owned by the user.
+        pkt = demux_copy_packet(pkt);
+    }
     if (!pkt)
-        abort();
-    pkt->next = NULL;
+        return 0;
 
     if (in->back_demuxing) {
         if (pkt->keyframe) {
@@ -3007,11 +3041,19 @@ static struct demuxer *open_given_type(struct mpv_global *global,
                     timeline_destroy(tl);
             }
         }
+
         if (!(params && params->is_top_level) || sub) {
             in->seekable_cache = false;
             in->min_secs = 0;
             in->max_bytes = 1;
         }
+
+        if (in->seekable_cache && opts->disk_cache) {
+            in->cache = demux_cache_create(global, log);
+            if (!in->cache)
+                MP_ERR(in, "Failed to create file cache.\n");
+        }
+
         switch_to_fresh_cache_range(in);
 
         demux_update(demuxer, MP_NOPTS_VALUE);
@@ -3803,6 +3845,7 @@ void demux_get_reader_state(struct demuxer *demuxer, struct demux_reader_state *
         .low_level_seeks = in->low_level_seeks,
         .ts_last = in->demux_ts,
         .bytes_per_second = in->bytes_per_second,
+        .file_cache_bytes = in->cache ? demux_cache_get_size(in->cache) : -1,
     };
     bool any_packets = false;
     for (int n = 0; n < in->num_streams; n++) {
diff --git a/demux/demux.h b/demux/demux.h
index 99acc1f015..14d145704a 100644
--- a/demux/demux.h
+++ b/demux/demux.h
@@ -44,6 +44,7 @@ struct demux_reader_state {
     double ts_end; // approx. timestamp of end of buffered range
     int64_t total_bytes;
     int64_t fw_bytes;
+    int64_t file_cache_bytes;
     double seeking; // current low level seek target, or NOPTS
     int low_level_seeks; // number of started low level seeks
     double ts_last; // approx. timestamp of demuxer position
diff --git a/demux/packet.c b/demux/packet.c
index 60c3e6aba0..32d799f9ce 100644
--- a/demux/packet.c
+++ b/demux/packet.c
@@ -31,10 +31,25 @@
 
 #include "packet.h"
 
+// Free any refcounted data dp holds (but don't free dp itself). This does not
+// care about pointers that are _not_ refcounted (like demux_packet.codec).
+// Normally, a user should use talloc_free(dp). This function is only for
+// annoyingly specific obscure use cases.
+void demux_packet_unref_contents(struct demux_packet *dp)
+{
+    if (dp->avpacket) {
+        av_packet_unref(dp->avpacket);
+        assert(!dp->is_cached);
+        dp->avpacket = NULL;
+        dp->buffer = NULL;
+        dp->len = 0;
+    }
+}
+
 static void packet_destroy(void *ptr)
 {
     struct demux_packet *dp = ptr;
-    av_packet_unref(dp->avpacket);
+    demux_packet_unref_contents(dp);
 }
 
 // This actually preserves only data and side data, not PTS/DTS/pos/etc.
@@ -161,8 +176,9 @@ struct demux_packet *demux_copy_packet(struct demux_packet *dp)
 size_t demux_packet_estimate_total_size(struct demux_packet *dp)
 {
     size_t size = ROUND_ALLOC(sizeof(struct demux_packet));
-    size += ROUND_ALLOC(dp->len);
     if (dp->avpacket) {
+        assert(!dp->is_cached);
+        size += ROUND_ALLOC(dp->len);
         size += ROUND_ALLOC(sizeof(AVPacket));
         size += ROUND_ALLOC(sizeof(AVBufferRef));
         size += 64; // upper bound estimate on sizeof(AVBuffer)
diff --git a/demux/packet.h b/demux/packet.h
index f4570004e8..cd1183d417 100644
--- a/demux/packet.h
+++ b/demux/packet.h
@@ -29,16 +29,29 @@ typedef struct demux_packet {
     double duration;
     int64_t pos;        // position in source file byte stream
 
-    unsigned char *buffer;
-    size_t len;
+    union {
+        // Normally valid for packets.
+        struct {
+            unsigned char *buffer;
+            size_t len;
+        };
+
+        // Used if is_cached==true, special uses only.
+        struct {
+            uint64_t pos;
+        } cached_data;
+    };
 
     int stream;         // source stream index (typically sh_stream.index)
 
     bool keyframe;
 
     // backward playback
-    bool back_restart;  // restart point (reverse and return previous frames)
-    bool back_preroll;  // initial discarded frame for smooth decoder reinit
+    bool back_restart : 1;  // restart point (reverse and return previous frames)
+    bool back_preroll : 1;  // initial discarded frame for smooth decoder reinit
+
+    // If true, cached_data is valid, while buffer/len are not.
+    bool is_cached : 1;
 
     // segmentation (ordered chapters, EDL)
     bool segmented;
@@ -68,4 +81,6 @@ int demux_packet_set_padding(struct demux_packet *dp, int start, int end);
 int demux_packet_add_blockadditional(struct demux_packet *dp, uint64_t id,
                                      void *data, size_t size);
 
+void demux_packet_unref_contents(struct demux_packet *dp);
+
 #endif /* MPLAYER_DEMUX_PACKET_H */