stream: redo playback abort handling

This mechanism originates from MPlayer's way of dealing with blocking
network, but it's still useful. On opening and closing, mpv waits for
network synchronously, and also some obscure commands and use-cases can
lead to such blocking. In these situations, the stream is asynchronously
forced to stop by "interrupting" it.

The old design interrupting I/O was a bit broken: polling with a
callback, instead of actively interrupting it. Change the direction of
this. There is no callback anymore, and the player calls
mp_cancel_trigger() to force the stream to return.

libavformat (via stream_lavf.c) has the old broken design, and fixing it
would require fixing libavformat, which won't happen so quickly. So we
have to keep that part. But everything above the stream layer is
prepared for a better design, and more sophisticated methods than
mp_cancel_test() could be easily introduced.

There's still one problem: commands are still run in the central
playback loop, which we assume can block on I/O in the worst case.
That's not a problem yet, because we simply mark some commands as being
able to stop playback of the current file ("quit" etc.), so input.c
could abort playback as soon as such a command is queued. But there are
also commands abort playback only conditionally, and the logic for that
is in the playback core and thus "unreachable". For example,
"playlist_next" aborts playback only if there's a next file. We don't
want it to always abort playback.

As a quite ugly hack, abort playback only if at least 2 abort commands
are queued - this pretty much happens only if the core is frozen and
doesn't react to input.

1 files changed, 43 insertions, 16 deletions

diff --git a/stream/stream.c b/stream/stream.c
index e2f71ae58f..a8162ffcb3 100644
--- a/stream/stream.c
+++ b/stream/stream.c
@@ -32,6 +32,7 @@
 
 #include <libavutil/common.h>
 #include "osdep/mpbswap.h"
+#include "osdep/atomics.h"
 
 #include "talloc.h"
 
@@ -252,8 +253,8 @@ static const char *match_proto(const char *url, const char *proto)
 }
 
 static int open_internal(const stream_info_t *sinfo, struct stream *underlying,
-                         const char *url, int flags, struct mpv_global *global,
-                         struct stream **ret)
+                         const char *url, int flags, struct mp_cancel *c,
+                         struct mpv_global *global, struct stream **ret)
 {
     if (sinfo->stream_filter != !!underlying)
         return STREAM_NO_MATCH;
@@ -281,6 +282,7 @@ static int open_internal(const stream_info_t *sinfo, struct stream *underlying,
     s->log = mp_log_new(s, global->log, sinfo->name);
     s->info = sinfo;
     s->opts = global->opts;
+    s->cancel = c;
     s->global = global;
     s->url = talloc_strdup(s, url);
     s->path = talloc_strdup(s, path);
@@ -335,7 +337,8 @@ static int open_internal(const stream_info_t *sinfo, struct stream *underlying,
     return STREAM_OK;
 }
 
-struct stream *stream_create(const char *url, int flags, struct mpv_global *global)
+struct stream *stream_create(const char *url, int flags,
+                             struct mp_cancel *c, struct mpv_global *global)
 {
     struct mp_log *log = mp_log_new(NULL, global->log, "!stream");
     struct stream *s = NULL;
@@ -344,7 +347,7 @@ struct stream *stream_create(const char *url, int flags, struct mpv_global *glob
     // Open stream proper
     bool unsafe = false;
     for (int i = 0; stream_list[i]; i++) {
-        int r = open_internal(stream_list[i], NULL, url, flags, global, &s);
+        int r = open_internal(stream_list[i], NULL, url, flags, c, global, &s);
         if (r == STREAM_OK)
             break;
         if (r == STREAM_NO_MATCH || r == STREAM_UNSUPPORTED)
@@ -375,7 +378,7 @@ struct stream *stream_create(const char *url, int flags, struct mpv_global *glob
     for (;;) {
         struct stream *new = NULL;
         for (int i = 0; stream_list[i]; i++) {
-            int r = open_internal(stream_list[i], s, s->url, flags, global, &new);
+            int r = open_internal(stream_list[i], s, s->url, flags, c, global, &new);
             if (r == STREAM_OK)
                 break;
         }
@@ -391,12 +394,12 @@ done:
 
 struct stream *stream_open(const char *filename, struct mpv_global *global)
 {
-    return stream_create(filename, STREAM_READ, global);
+    return stream_create(filename, STREAM_READ, NULL, global);
 }
 
 stream_t *open_output_stream(const char *filename, struct mpv_global *global)
 {
-    return stream_create(filename, STREAM_WRITE, global);
+    return stream_create(filename, STREAM_WRITE, NULL, global);
 }
 
 static int stream_reconnect(stream_t *s)
@@ -407,7 +410,7 @@ static int stream_reconnect(stream_t *s)
         return 0;
     if (!s->seekable)
         return 0;
-    if (stream_check_interrupt(s))
+    if (mp_cancel_test(s->cancel))
         return 0;
     int64_t pos = s->pos;
     int sleep_ms = 5;
@@ -419,7 +422,7 @@ static int stream_reconnect(stream_t *s)
             sleep_ms = MPMIN(sleep_ms * 2, RECONNECT_SLEEP_MAX_MS);
         }
 
-        if (stream_check_interrupt(s))
+        if (mp_cancel_test(s->cancel))
             return 0;
 
         s->eof = 1;
@@ -766,13 +769,6 @@ void free_stream(stream_t *s)
     talloc_free(s);
 }
 
-bool stream_check_interrupt(struct stream *s)
-{
-    if (!s->global || !s->global->stream_interrupt_cb)
-        return false;
-    return s->global->stream_interrupt_cb(s->global->stream_interrupt_cb_ctx);
-}
-
 stream_t *open_memory_stream(void *data, int len)
 {
     assert(len >= 0);
@@ -802,6 +798,7 @@ static stream_t *open_cache(stream_t *orig, const char *name)
     cache->streaming = orig->streaming,
     cache->is_network = orig->is_network;
     cache->opts = orig->opts;
+    cache->cancel = orig->cancel;
     cache->global = orig->global;
 
     cache->log = mp_log_new(cache, cache->global->log, name);
@@ -989,6 +986,36 @@ struct bstr stream_read_complete(struct stream *s, void *talloc_ctx,
     return (struct bstr){buf, total_read};
 }
 
+struct mp_cancel {
+    atomic_bool triggered;
+};
+
+struct mp_cancel *mp_cancel_new(void *talloc_ctx)
+{
+    struct mp_cancel *c = talloc_ptrtype(talloc_ctx, c);
+    *c = (struct mp_cancel){.triggered = ATOMIC_VAR_INIT(false)};
+    return c;
+}
+
+// Request abort.
+void mp_cancel_trigger(struct mp_cancel *c)
+{
+    c->triggered = true;
+}
+
+// Restore original state. (Allows reusing a mp_cancel.)
+void mp_cancel_reset(struct mp_cancel *c)
+{
+    c->triggered = false;
+}
+
+// Return whether the caller should abort.
+// For convenience, c==NULL is allowed.
+bool mp_cancel_test(struct mp_cancel *c)
+{
+    return c ? c->triggered : false;
+}
+
 void stream_print_proto_list(struct mp_log *log)
 {
     int count = 0;