player: add prefetching of the next playlist entry

Since for mpv CLI, the player state is a singleton, full prefetching is
a bit tricky. We do it only on the demuxer layer.

The implementation reuses the old "open thread". This means there is
significant potential for regressions even if the new option is not
used. This is made worse by the fact that I barely tested this code.

The generic mpctx_run_reentrant() wrapper is also removed - this was its
only user, and its remains become part of the new implementation.

8 files changed, 165 insertions, 91 deletions

diff --git a/DOCS/man/options.rst b/DOCS/man/options.rst
index a812fb088c..079672533f 100644
--- a/DOCS/man/options.rst
+++ b/DOCS/man/options.rst
@@ -2737,6 +2737,27 @@ Demuxer
     (This value tends to be fuzzy, because many file formats don't store linear
     timestamps.)
 
+``--prefetch-playlist=<yes|no>``
+    Prefetch next playlist entry while playback of the current entry is ending
+    (default: no). This merely opens the URL of the next playlist entry as soon
+    as the current URL is fully read.
+
+    This does **not** work with URLs resolved by the ``youtube-dl`` wrapper,
+    and it won't.
+
+    This does not affect HLS (``.m3u8`` URLs) - HLS prefetching depends on the
+    demuxer cache settings and is on by default.
+
+    This can give subtly wrong results if per-file options are used, or if
+    options are changed in the time window between prefetching start and next
+    file played.
+
+    This can occasionally make wrong prefetching decisions. For example, it
+    can't predict whether you go backwards in the playlist, and assumes you
+    won't edit the playlist.
+
+    Highly experimental.
+
 ``--force-seekable=<yes|no>``
     If the player thinks that the media is not seekable (e.g. playing from a
     pipe, or it's an http stream with a server that doesn't support range
diff --git a/options/options.c b/options/options.c
index bc9ddf8614..6afe611c6d 100644
--- a/options/options.c
+++ b/options/options.c
@@ -383,6 +383,7 @@ const m_option_t mp_opts[] = {
     OPT_STRING("audio-demuxer", audio_demuxer_name, 0),
     OPT_STRING("sub-demuxer", sub_demuxer_name, 0),
     OPT_FLAG("demuxer-thread", demuxer_thread, 0),
+    OPT_FLAG("prefetch-playlist", prefetch_open, 0),
     OPT_FLAG("cache-pause", cache_pausing, 0),
 
     OPT_DOUBLE("mf-fps", mf_fps, 0),
diff --git a/options/options.h b/options/options.h
index 1d79d55c31..d3b0ace24f 100644
--- a/options/options.h
+++ b/options/options.h
@@ -219,6 +219,7 @@ typedef struct MPOpts {
     char **audio_files;
     char *demuxer_name;
     int demuxer_thread;
+    int prefetch_open;
     char *audio_demuxer_name;
     char *sub_demuxer_name;
 
diff --git a/player/command.c b/player/command.c
index 1beca6158a..7c3e269e9c 100644
--- a/player/command.c
+++ b/player/command.c
@@ -4959,7 +4959,7 @@ int run_command(struct MPContext *mpctx, struct mp_cmd *cmd, struct mpv_node *re
         int dir = cmd->id == MP_CMD_PLAYLIST_PREV ? -1 : +1;
         int force = cmd->args[0].v.i;
 
-        struct playlist_entry *e = mp_next_file(mpctx, dir, force);
+        struct playlist_entry *e = mp_next_file(mpctx, dir, force, true);
         if (!e && !force)
             return -1;
         mp_set_playlist_entry(mpctx, e);
diff --git a/player/core.h b/player/core.h
index ad61500741..4767830bc9 100644
--- a/player/core.h
+++ b/player/core.h
@@ -435,6 +435,21 @@ typedef struct MPContext {
 
     // --- The following fields are protected by lock
     struct mp_cancel *demuxer_cancel; // cancel handle for MPContext.demuxer
+
+    // --- Owned by MPContext
+    pthread_t open_thread;
+    bool open_active; // open_thread is a valid thread handle, all setup
+    atomic_bool open_done;
+    // --- All fields below are immutable while open_active is true.
+    //     Otherwise, they're owned by MPContext.
+    struct mp_cancel *open_cancel;
+    char *open_url;
+    char *open_format;
+    int open_url_flags;
+    // --- All fields below are owned by open_thread, unless open_done was set
+    //     to true.
+    struct demuxer *open_res_demuxer;
+    int open_res_error;
 } MPContext;
 
 // audio.c
@@ -480,7 +495,7 @@ struct track *mp_track_by_tid(struct MPContext *mpctx, enum stream_type type,
 void add_demuxer_tracks(struct MPContext *mpctx, struct demuxer *demuxer);
 bool mp_remove_track(struct MPContext *mpctx, struct track *track);
 struct playlist_entry *mp_next_file(struct MPContext *mpctx, int direction,
-                                    bool force);
+                                    bool force, bool mutate);
 void mp_set_playlist_entry(struct MPContext *mpctx, struct playlist_entry *e);
 void mp_play_files(struct MPContext *mpctx);
 void update_demuxer_properties(struct MPContext *mpctx);
@@ -490,6 +505,7 @@ void prepare_playlist(struct MPContext *mpctx, struct playlist *pl);
 void autoload_external_files(struct MPContext *mpctx);
 struct track *select_default_track(struct MPContext *mpctx, int order,
                                    enum stream_type type);
+void prefetch_next(struct MPContext *mpctx);
 
 // main.c
 int mp_initialize(struct MPContext *mpctx, char **argv);
@@ -508,8 +524,6 @@ void update_vo_playback_state(struct MPContext *mpctx);
 void update_window_title(struct MPContext *mpctx, bool force);
 void error_on_track(struct MPContext *mpctx, struct track *track);
 int stream_dump(struct MPContext *mpctx, const char *source_filename);
-int mpctx_run_reentrant(struct MPContext *mpctx, void (*thread_fn)(void *arg),
-                        void *thread_arg);
 double get_track_seek_offset(struct MPContext *mpctx, struct track *track);
 
 // osd.c
diff --git a/player/loadfile.c b/player/loadfile.c
index e59acbc8d4..8023428e75 100644
--- a/player/loadfile.c
+++ b/player/loadfile.c
@@ -790,61 +790,141 @@ static void load_per_file_options(m_config_t *conf,
     }
 }
 
-struct demux_open_args {
-    int stream_flags;
-    char *url;
-    struct mpv_global *global;
-    struct mp_cancel *cancel;
-    struct mp_log *log;
-    // results
-    struct demuxer *demux;
-    int err;
-};
-
-static void open_demux_thread(void *pctx)
+static void *open_demux_thread(void *ctx)
 {
-    struct demux_open_args *args = pctx;
-    struct mpv_global *global = args->global;
+    struct MPContext *mpctx = ctx;
+
     struct demuxer_params p = {
-        .force_format = global->opts->demuxer_name,
+        .force_format = mpctx->open_format,
         .allow_capture = true,
-        .stream_flags = args->stream_flags,
+        .stream_flags = mpctx->open_url_flags,
     };
-    args->demux = demux_open_url(args->url, &p, args->cancel, global);
-    if (!args->demux) {
+    mpctx->open_res_demuxer =
+        demux_open_url(mpctx->open_url, &p, mpctx->open_cancel, mpctx->global);
+
+    if (mpctx->open_res_demuxer) {
+        MP_VERBOSE(mpctx, "Opening done: %s\n", mpctx->open_url);
+    } else {
+        MP_VERBOSE(mpctx, "Opening failed or was aborted: %s\n", mpctx->open_url);
+
         if (p.demuxer_failed) {
-            args->err = MPV_ERROR_UNKNOWN_FORMAT;
+            mpctx->open_res_error = MPV_ERROR_UNKNOWN_FORMAT;
         } else {
-            args->err = MPV_ERROR_LOADING_FAILED;
+            mpctx->open_res_error = MPV_ERROR_LOADING_FAILED;
         }
     }
+
+    atomic_store(&mpctx->open_done, true);
+    mp_wakeup_core(mpctx);
+    return NULL;
+}
+
+static void cancel_open(struct MPContext *mpctx)
+{
+    if (mpctx->open_cancel)
+        mp_cancel_trigger(mpctx->open_cancel);
+
+    if (mpctx->open_active)
+        pthread_join(mpctx->open_thread, NULL);
+    mpctx->open_active = false;
+
+    TA_FREEP(&mpctx->open_cancel);
+    TA_FREEP(&mpctx->open_url);
+    TA_FREEP(&mpctx->open_format);
+
+    if (mpctx->open_res_demuxer)
+        free_demuxer_and_stream(mpctx->open_res_demuxer);
+    mpctx->open_res_demuxer = NULL;
+
+    atomic_store(&mpctx->open_done, false);
+}
+
+// Setup all the field to open this url, and make sure a thread is running.
+static void start_open(struct MPContext *mpctx, char *url, int url_flags)
+{
+    cancel_open(mpctx);
+
+    assert(!mpctx->open_active);
+    assert(!mpctx->open_cancel);
+    assert(!mpctx->open_res_demuxer);
+    assert(!atomic_load(&mpctx->open_done));
+
+    mpctx->open_cancel = mp_cancel_new(NULL);
+    mpctx->open_url = talloc_strdup(NULL, url);
+    mpctx->open_format = talloc_strdup(NULL, mpctx->opts->demuxer_name);
+    mpctx->open_url_flags = url_flags;
+    if (mpctx->opts->load_unsafe_playlists)
+        mpctx->open_url_flags = 0;
+
+    if (pthread_create(&mpctx->open_thread, NULL, open_demux_thread, mpctx)) {
+        cancel_open(mpctx);
+        return;
+    }
+
+    mpctx->open_active = true;
 }
 
 static void open_demux_reentrant(struct MPContext *mpctx)
 {
-    struct demux_open_args args = {
-        .global = mpctx->global,
-        .cancel = mp_cancel_new(NULL),
-        .log = mpctx->log,
-        .stream_flags = mpctx->playing->stream_flags,
-        .url = talloc_strdup(NULL, mpctx->stream_open_filename),
-    };
+    char *url = mpctx->stream_open_filename;
+
+    if (mpctx->open_active) {
+        bool done = atomic_load(&mpctx->open_done);
+        bool failed = done && !mpctx->open_res_demuxer;
+        bool correct_url = strcmp(mpctx->open_url, url) == 0;
+
+        if (correct_url && !failed) {
+            MP_VERBOSE(mpctx, "Using prefetched/prefetching URL.\n");
+        } else if (correct_url && failed) {
+            MP_VERBOSE(mpctx, "Prefetched URL failed, retrying.\n");
+            cancel_open(mpctx);
+        } else {
+            if (!done)
+                MP_VERBOSE(mpctx, "Aborting onging prefetch of wrong URL.\n");
+            cancel_open(mpctx);
+        }
+    }
+
+    if (!mpctx->open_active)
+        start_open(mpctx, url, mpctx->playing->stream_flags);
+
+    // User abort should cancel the opener now.
     pthread_mutex_lock(&mpctx->lock);
-    mpctx->demuxer_cancel = args.cancel;
+    mpctx->demuxer_cancel = mpctx->open_cancel;
     pthread_mutex_unlock(&mpctx->lock);
-    if (mpctx->opts->load_unsafe_playlists)
-        args.stream_flags = 0;
-    mpctx_run_reentrant(mpctx, open_demux_thread, &args);
-    if (args.demux) {
-        mpctx->demuxer = args.demux;
+
+    while (!atomic_load(&mpctx->open_done)) {
+        mp_idle(mpctx);
+
+        if (mpctx->stop_play)
+            mp_abort_playback_async(mpctx);
+    }
+
+    if (mpctx->open_res_demuxer) {
+        assert(mpctx->demuxer_cancel == mpctx->open_cancel);
+        mpctx->demuxer = mpctx->open_res_demuxer;
+        mpctx->open_res_demuxer = NULL;
+        mpctx->open_cancel = NULL;
     } else {
-        mpctx->error_playing = args.err;
+        mpctx->error_playing = mpctx->open_res_error;
         pthread_mutex_lock(&mpctx->lock);
-        talloc_free(mpctx->demuxer_cancel);
         mpctx->demuxer_cancel = NULL;
         pthread_mutex_unlock(&mpctx->lock);
     }
-    talloc_free(args.url);
+
+    cancel_open(mpctx); // cleanup
+}
+
+void prefetch_next(struct MPContext *mpctx)
+{
+    if (!mpctx->opts->prefetch_open)
+        return;
+
+    struct playlist_entry *new_entry = mp_next_file(mpctx, +1, false, false);
+    if (new_entry && !mpctx->open_active && new_entry->filename) {
+        MP_VERBOSE(mpctx, "Prefetching: %s\n", new_entry->filename);
+        start_open(mpctx, new_entry->filename, new_entry->stream_flags);
+    }
 }
 
 static bool init_complex_filters(struct MPContext *mpctx)
@@ -1279,8 +1359,9 @@ terminate_playback:
 // it can have side-effects and mutate mpctx.
 //  direction: -1 (previous) or +1 (next)
 //  force: if true, don't skip playlist entries marked as failed
+//  mutate: if true, change loop counters
 struct playlist_entry *mp_next_file(struct MPContext *mpctx, int direction,
-                                    bool force)
+                                    bool force, bool mutate)
 {
     struct playlist_entry *next = playlist_get_next(mpctx->playlist, direction);
     if (next && direction < 0 && !force) {
@@ -1340,7 +1421,7 @@ void mp_play_files(struct MPContext *mpctx)
         if (mpctx->stop_play == PT_NEXT_ENTRY || mpctx->stop_play == PT_ERROR ||
             mpctx->stop_play == AT_END_OF_FILE || !mpctx->stop_play)
         {
-            new_entry = mp_next_file(mpctx, +1, false);
+            new_entry = mp_next_file(mpctx, +1, false, true);
         }
 
         mpctx->playlist->current = new_entry;
@@ -1350,6 +1431,8 @@ void mp_play_files(struct MPContext *mpctx)
         if (!mpctx->playlist->current && mpctx->opts->player_idle_mode < 2)
             break;
     }
+
+    cancel_open(mpctx);
 }
 
 // Abort current playback and set the given entry to play next.
diff --git a/player/misc.c b/player/misc.c
index 032342e84a..2c160aef73 100644
--- a/player/misc.c
+++ b/player/misc.c
@@ -17,7 +17,6 @@
 
 #include <stddef.h>
 #include <stdbool.h>
-#include <pthread.h>
 #include <assert.h>
 
 #include "config.h"
@@ -251,51 +250,3 @@ void merge_playlist_files(struct playlist *pl)
     playlist_add_file(pl, edl);
     talloc_free(edl);
 }
-
-struct wrapper_args {
-    struct MPContext *mpctx;
-    void (*thread_fn)(void *);
-    void *thread_arg;
-    pthread_mutex_t mutex;
-    bool done;
-};
-
-static void *thread_wrapper(void *pctx)
-{
-    struct wrapper_args *args = pctx;
-    mpthread_set_name("opener");
-    args->thread_fn(args->thread_arg);
-    pthread_mutex_lock(&args->mutex);
-    args->done = true;
-    pthread_mutex_unlock(&args->mutex);
-    mp_wakeup_core(args->mpctx); // this interrupts mp_idle()
-    return NULL;
-}
-
-// Run the thread_fn in a new thread. Wait until the thread returns, but while
-// waiting, process input and input commands.
-int mpctx_run_reentrant(struct MPContext *mpctx, void (*thread_fn)(void *arg),
-                        void *thread_arg)
-{
-    struct wrapper_args args = {mpctx, thread_fn, thread_arg};
-    pthread_mutex_init(&args.mutex, NULL);
-    bool success = false;
-    pthread_t thread;
-    if (pthread_create(&thread, NULL, thread_wrapper, &args))
-        goto done;
-    while (!success) {
-        mp_idle(mpctx);
-
-        if (mpctx->stop_play)
-            mp_abort_playback_async(mpctx);
-
-        pthread_mutex_lock(&args.mutex);
-        success |= args.done;
-        pthread_mutex_unlock(&args.mutex);
-    }
-    pthread_join(thread, NULL);
-done:
-    pthread_mutex_destroy(&args.mutex);
-    mp_wakeup_core(mpctx); // avoid lost wakeups during waiting
-    return success ? 0 : -1;
-}
diff --git a/player/playloop.c b/player/playloop.c
index 9db9396f95..232a75f814 100644
--- a/player/playloop.c
+++ b/player/playloop.c
@@ -663,6 +663,9 @@ static void handle_pause_on_low_cache(struct MPContext *mpctx)
         force_update = true;
     }
 
+    if (s.eof && !busy)
+        prefetch_next(mpctx);
+
     if (force_update)
         mp_notify(mpctx, MP_EVENT_CACHE_UPDATE, NULL);
 }