player: use better way to wait for input and dispatching commands

Instead of using input_ctx for waiting, use the dispatch queue directly.
One big change is that the dispatch queue will just process commands
that come in (e.g. from client API) without returning. This should
reduce unnecessary playloop excutions (which is good since the playloop
got a bit fat from rechecking a lot of conditions every iteration).

Since this doesn't force a new playloop iteration on every access, this
has to be enforced manually in some cases.

Normal input (via terminal or VO window) still wakes up the playloop
every time, though that's not too important. It makes testing this
harder, though. If there are missing wakeup calls, it will be noticed
only when using the client API in some form.

At this point we could probably use a normal lock instead of the
dispatch queue stuff.

9 files changed, 80 insertions, 57 deletions

diff --git a/input/input.c b/input/input.c
index f820b8b174..90cd439779 100644
--- a/input/input.c
+++ b/input/input.c
@@ -22,6 +22,7 @@
 #include <stdio.h>
 #include <stdbool.h>
 #include <unistd.h>
+#include <math.h>
 #include <errno.h>
 #include <sys/types.h>
 #include <sys/stat.h>
@@ -34,7 +35,6 @@
 #include <libavutil/common.h>
 
 #include "osdep/io.h"
-#include "osdep/semaphore.h"
 #include "misc/rendezvous.h"
 
 #include "input.h"
@@ -96,7 +96,6 @@ struct cmd_queue {
 
 struct input_ctx {
     pthread_mutex_t mutex;
-    sem_t wakeup;
     struct mp_log *log;
     struct mpv_global *global;
     struct m_config_cache *opts_cache;
@@ -147,6 +146,9 @@ struct input_ctx {
     struct cmd_queue cmd_queue;
 
     struct mp_cancel *cancel;
+
+    void (*wakeup_cb)(void *ctx);
+    void *wakeup_ctx;
 };
 
 static int parse_config(struct input_ctx *ictx, bool builtin, bstr data,
@@ -846,33 +848,18 @@ static mp_cmd_t *check_autorepeat(struct input_ctx *ictx)
     return NULL;
 }
 
-void mp_input_wait(struct input_ctx *ictx, double seconds)
+double mp_input_get_delay(struct input_ctx *ictx)
 {
     input_lock(ictx);
+    double seconds = INFINITY;
     adjust_max_wait_time(ictx, &seconds);
     input_unlock(ictx);
-    while (sem_trywait(&ictx->wakeup) == 0)
-        seconds = -1;
-    if (seconds > 0) {
-        MP_STATS(ictx, "start sleep");
-        struct timespec ts =
-            mp_time_us_to_timespec(mp_add_timeout(mp_time_us(), seconds));
-        sem_timedwait(&ictx->wakeup, &ts);
-        MP_STATS(ictx, "end sleep");
-    }
-}
-
-void mp_input_wakeup_nolock(struct input_ctx *ictx)
-{
-    // Some audio APIs discourage use of locking in their audio callback,
-    // and these audio callbacks happen to call mp_input_wakeup_nolock()
-    // when new data is needed. This is why we use semaphores here.
-    sem_post(&ictx->wakeup);
+    return seconds;
 }
 
 void mp_input_wakeup(struct input_ctx *ictx)
 {
-    mp_input_wakeup_nolock(ictx);
+    ictx->wakeup_cb(ictx->wakeup_ctx);
 }
 
 mp_cmd_t *mp_input_read_cmd(struct input_ctx *ictx)
@@ -1196,7 +1183,9 @@ done:
     return r;
 }
 
-struct input_ctx *mp_input_init(struct mpv_global *global)
+struct input_ctx *mp_input_init(struct mpv_global *global,
+                                void (*wakeup_cb)(void *ctx),
+                                void *wakeup_ctx)
 {
 
     struct input_ctx *ictx = talloc_ptrtype(NULL, ictx);
@@ -1206,15 +1195,12 @@ struct input_ctx *mp_input_init(struct mpv_global *global)
         .log = mp_log_new(ictx, global->log, "input"),
         .mouse_section = "default",
         .opts_cache = m_config_cache_alloc(ictx, global, &input_config),
+        .wakeup_cb = wakeup_cb,
+        .wakeup_ctx = wakeup_ctx,
     };
 
     ictx->opts = ictx->opts_cache->opts;
 
-    if (sem_init(&ictx->wakeup, 0, 0)) {
-        MP_FATAL(ictx, "mpv doesn't work on systems without POSIX semaphores.\n");
-        abort();
-    }
-
     mpthread_mutex_init_recursive(&ictx->mutex);
 
     // Setup default section, so that it does nothing.
@@ -1308,7 +1294,6 @@ void mp_input_uninit(struct input_ctx *ictx)
     clear_queue(&ictx->cmd_queue);
     talloc_free(ictx->current_down_cmd);
     pthread_mutex_destroy(&ictx->mutex);
-    sem_destroy(&ictx->wakeup);
     talloc_free(ictx);
 }
 
diff --git a/input/input.h b/input/input.h
index a5710b6065..cc523efc62 100644
--- a/input/input.h
+++ b/input/input.h
@@ -219,22 +219,22 @@ bool mp_input_test_dragging(struct input_ctx *ictx, int x, int y);
 
 // Initialize the input system
 struct mpv_global;
-struct input_ctx *mp_input_init(struct mpv_global *global);
+struct input_ctx *mp_input_init(struct mpv_global *global,
+                                void (*wakeup_cb)(void *ctx),
+                                void *wakeup_ctx);
 
 // Load config, options, and devices.
 void mp_input_load(struct input_ctx *ictx);
 
 void mp_input_uninit(struct input_ctx *ictx);
 
-// Sleep for the given amount of seconds, until mp_input_wakeup() is called,
-// or new input arrives. seconds<=0 returns immediately.
-void mp_input_wait(struct input_ctx *ictx, double seconds);
+// Return number of seconds until the next autorepeat event will be generated.
+// Returns INFINITY if no autorepeated key is active.
+double mp_input_get_delay(struct input_ctx *ictx);
 
 // Wake up sleeping input loop from another thread.
 void mp_input_wakeup(struct input_ctx *ictx);
 
-void mp_input_wakeup_nolock(struct input_ctx *ictx);
-
 // Used to asynchronously abort playback. Needed because the core still can
 // block on network in some situations.
 struct mp_cancel;
diff --git a/misc/dispatch.c b/misc/dispatch.c
index 4f1cca03ec..dca39e5fe1 100644
--- a/misc/dispatch.c
+++ b/misc/dispatch.c
@@ -30,6 +30,8 @@ struct mp_dispatch_queue {
     pthread_cond_t cond;
     void (*wakeup_fn)(void *wakeup_ctx);
     void *wakeup_ctx;
+    // Make mp_dispatch_queue_process() exit if it's idle.
+    bool interrupted;
     // The target thread is blocked by mp_dispatch_queue_process(). Note that
     // mp_dispatch_lock() can set this from true to false to keep the thread
     // blocked (this stops if from processing other dispatch items, and from
@@ -185,10 +187,11 @@ void mp_dispatch_run(struct mp_dispatch_queue *queue,
 // function can be much higher if the suspending/locking functions are used, or
 // if executing the dispatch items takes time. On the other hand, this function
 // can return much earlier than the timeout due to sporadic wakeups.
-// It is also guaranteed that if at least one queue item was processed, the
-// function will return as soon as possible, ignoring the timeout. This
-// simplifies users, such as re-checking conditions before waiting. (It will
-// still process the remaining queue items, and wait for unsuspend.)
+// Note that this will strictly return only after:
+//      - timeout has passed,
+//      - all queue items were processed,
+//      - the possibly acquired lock has been released
+// It's possible to cancel the timeout by calling mp_dispatch_interrupt().
 void mp_dispatch_queue_process(struct mp_dispatch_queue *queue, double timeout)
 {
     int64_t wait = timeout > 0 ? mp_add_timeout(mp_time_us(), timeout) : 0;
@@ -243,18 +246,33 @@ void mp_dispatch_queue_process(struct mp_dispatch_queue *queue, double timeout)
             } else {
                 item->completed = true;
             }
-        } else if (wait > 0) {
+        } else if (wait > 0 && !queue->interrupted) {
             struct timespec ts = mp_time_us_to_timespec(wait);
-            pthread_cond_timedwait(&queue->cond, &queue->lock, &ts);
+            if (pthread_cond_timedwait(&queue->cond, &queue->lock, &ts))
+                break;
         } else {
             break;
         }
-        wait = 0;
     }
     queue->idling = false;
     assert(!frame.locked);
     assert(queue->frame == &frame);
     queue->frame = frame.prev;
+    queue->interrupted = false;
+    pthread_mutex_unlock(&queue->lock);
+}
+
+// If the queue is inside of mp_dispatch_queue_process(), make it return as
+// soon as all work items have been run, without waiting for the timeout. This
+// does not make it return early if it's blocked by a mp_dispatch_lock().
+// If mp_dispatch_queue_process() is called in a reentrant way (including the
+// case where another thread calls mp_dispatch_lock() and then
+// mp_dispatch_queue_process()), this affects only the "topmost" invocation.
+void mp_dispatch_interrupt(struct mp_dispatch_queue *queue)
+{
+    pthread_mutex_lock(&queue->lock);
+    queue->interrupted = true;
+    pthread_cond_broadcast(&queue->cond);
     pthread_mutex_unlock(&queue->lock);
 }
 
diff --git a/misc/dispatch.h b/misc/dispatch.h
index 7a0d037cab..a762e47cd2 100644
--- a/misc/dispatch.h
+++ b/misc/dispatch.h
@@ -15,6 +15,7 @@ void mp_dispatch_enqueue_autofree(struct mp_dispatch_queue *queue,
 void mp_dispatch_run(struct mp_dispatch_queue *queue,
                      mp_dispatch_fn fn, void *fn_data);
 void mp_dispatch_queue_process(struct mp_dispatch_queue *queue, double timeout);
+void mp_dispatch_interrupt(struct mp_dispatch_queue *queue);
 void mp_dispatch_lock(struct mp_dispatch_queue *queue);
 void mp_dispatch_unlock(struct mp_dispatch_queue *queue);
 
diff --git a/player/client.c b/player/client.c
index 44732ed04f..bed21565a7 100644
--- a/player/client.c
+++ b/player/client.c
@@ -356,6 +356,7 @@ void mpv_resume(mpv_handle *ctx)
         mp_dispatch_lock(ctx->mpctx->dispatch);
         ctx->mpctx->suspend_count--;
         mp_dispatch_unlock(ctx->mpctx->dispatch);
+        mp_dispatch_interrupt(ctx->mpctx->dispatch);
     }
 }
 
diff --git a/player/core.h b/player/core.h
index c39f316276..0f1c8c7e17 100644
--- a/player/core.h
+++ b/player/core.h
@@ -21,6 +21,8 @@
 #include <stdbool.h>
 #include <pthread.h>
 
+#include "osdep/atomic.h"
+
 #include "libmpv/client.h"
 
 #include "common/common.h"
@@ -233,6 +235,7 @@ typedef struct MPContext {
     struct mp_client_api *clients;
     struct mp_dispatch_queue *dispatch;
     struct mp_cancel *playback_abort;
+    bool in_dispatch;
 
     struct mp_log *statusline;
     struct osd_state *osd;
@@ -510,7 +513,7 @@ void get_current_osd_sym(struct MPContext *mpctx, char *buf, size_t buf_size);
 void set_osd_bar_chapters(struct MPContext *mpctx, int type);
 
 // playloop.c
-void mp_wait_events(struct MPContext *mpctx, double sleeptime);
+void mp_wait_events(struct MPContext *mpctx);
 void mp_set_timeout(struct MPContext *mpctx, double sleeptime);
 void mp_wakeup_core(struct MPContext *mpctx);
 void mp_wakeup_core_cb(void *ctx);
diff --git a/player/main.c b/player/main.c
index 7d2aa399ea..a6e82287d7 100644
--- a/player/main.c
+++ b/player/main.c
@@ -143,7 +143,7 @@ static void shutdown_clients(struct MPContext *mpctx)
     while (mpctx->clients && mp_clients_num(mpctx)) {
         mp_client_broadcast_event(mpctx, MPV_EVENT_SHUTDOWN, NULL);
         mp_dispatch_queue_process(mpctx->dispatch, 0);
-        mp_wait_events(mpctx, 10000);
+        mp_wait_events(mpctx);
     }
 }
 
@@ -344,7 +344,7 @@ struct MPContext *mp_create(void)
 
     mpctx->global->opts = mpctx->opts;
 
-    mpctx->input = mp_input_init(mpctx->global);
+    mpctx->input = mp_input_init(mpctx->global, mp_wakeup_core_cb, mpctx);
     screenshot_init(mpctx);
     command_init(mpctx);
     init_libav(mpctx->global);
diff --git a/player/playloop.c b/player/playloop.c
index a8d2d1a695..c2795e0d40 100644
--- a/player/playloop.c
+++ b/player/playloop.c
@@ -51,10 +51,24 @@
 #include "command.h"
 
 // Wait until mp_wakeup_core() is called, since the last time
-// mp_wait_events() was called. (But see mp_process_input().)
-void mp_wait_events(struct MPContext *mpctx, double sleeptime)
+// mp_wait_events() was called.
+void mp_wait_events(struct MPContext *mpctx)
 {
-    mp_input_wait(mpctx->input, sleeptime);
+    if (mpctx->sleeptime > 0)
+        MP_STATS(mpctx, "start sleep");
+
+    mpctx->in_dispatch = true;
+
+    mp_dispatch_queue_process(mpctx->dispatch, mpctx->sleeptime);
+
+    while (mpctx->suspend_count)
+        mp_dispatch_queue_process(mpctx->dispatch, 100);
+
+    mpctx->in_dispatch = false;
+    mpctx->sleeptime = INFINITY;
+
+    if (mpctx->sleeptime > 0)
+        MP_STATS(mpctx, "end sleep");
 }
 
 // Set the timeout used when the playloop goes to sleep. This means the
@@ -63,6 +77,10 @@ void mp_wait_events(struct MPContext *mpctx, double sleeptime)
 void mp_set_timeout(struct MPContext *mpctx, double sleeptime)
 {
     mpctx->sleeptime = MPMIN(mpctx->sleeptime, sleeptime);
+
+    // Can't adjust timeout if called from mp_dispatch_queue_process().
+    if (mpctx->in_dispatch && isfinite(sleeptime))
+        mp_wakeup_core(mpctx);
 }
 
 // Cause the playloop to run. This can be called from any thread. If called
@@ -70,7 +88,7 @@ void mp_set_timeout(struct MPContext *mpctx, double sleeptime)
 // of going to sleep in the next mp_wait_events().
 void mp_wakeup_core(struct MPContext *mpctx)
 {
-    mp_input_wakeup(mpctx->input);
+    mp_dispatch_interrupt(mpctx->dispatch);
 }
 
 // Opaque callback variant of mp_wakeup_core().
@@ -84,17 +102,14 @@ void mp_wakeup_core_cb(void *ctx)
 // API threads. This also resets the "wakeup" flag used with mp_wait_events().
 void mp_process_input(struct MPContext *mpctx)
 {
-    mp_dispatch_queue_process(mpctx->dispatch, 0);
     for (;;) {
         mp_cmd_t *cmd = mp_input_read_cmd(mpctx->input);
         if (!cmd)
             break;
         run_command(mpctx, cmd, NULL);
         mp_cmd_free(cmd);
-        mp_dispatch_queue_process(mpctx->dispatch, 0);
     }
-    while (mpctx->suspend_count)
-        mp_dispatch_queue_process(mpctx->dispatch, 100);
+    mp_set_timeout(mpctx, mp_input_get_delay(mpctx->input));
 }
 
 double get_relative_time(struct MPContext *mpctx)
@@ -1045,7 +1060,7 @@ void run_playloop(struct MPContext *mpctx)
 
     if (mpctx->lavfi) {
         if (lavfi_process(mpctx->lavfi))
-            mpctx->sleeptime = 0;
+            mp_wakeup_core(mpctx);
         if (lavfi_has_failed(mpctx->lavfi))
             mpctx->stop_play = AT_END_OF_FILE;
     }
@@ -1079,8 +1094,7 @@ void run_playloop(struct MPContext *mpctx)
 
     handle_osd_redraw(mpctx);
 
-    mp_wait_events(mpctx, mpctx->sleeptime);
-    mpctx->sleeptime = 1e9; // infinite for all practical purposes
+    mp_wait_events(mpctx);
 
     handle_pause_on_low_cache(mpctx);
 
@@ -1096,8 +1110,7 @@ void run_playloop(struct MPContext *mpctx)
 void mp_idle(struct MPContext *mpctx)
 {
     handle_dummy_ticks(mpctx);
-    mp_wait_events(mpctx, mpctx->sleeptime);
-    mpctx->sleeptime = 100.0;
+    mp_wait_events(mpctx);
     mp_process_input(mpctx);
     handle_command_updates(mpctx);
     handle_cursor_autohide(mpctx);
diff --git a/video/out/w32_common.c b/video/out/w32_common.c
index 0dff2643b1..64c3e700aa 100644
--- a/video/out/w32_common.c
+++ b/video/out/w32_common.c
@@ -1621,6 +1621,8 @@ static void do_terminate(void *ptr)
 
     if (!w32->destroyed)
         DestroyWindow(w32->window);
+
+    mp_dispatch_interrupt(w32->dispatch);
 }
 
 void vo_w32_uninit(struct vo *vo)