input: rework event reading and command queuing

Rework much of the logic related to reading from event sources and
queuing commands. The two biggest architecture changes are:
- The code buffering keycodes in mp_fifo.c is gone. Instead key input
  is now immediately fed to input.c and interpreted as commands, and
  then the commands are buffered instead.
- mp_input_get_cmd() now always tries to read every available event
  from every event source and convert them to (buffered) commands.
  Before it would only process new events until one new command became
  available.

Some relevant behavior changes:
- Before commands could be lost when stream code called
  mp_input_check_interrupt() which read commands (to see if they were
  of types that triggered aborts during slow IO tasks) and then threw
  them away. This was especially an issue if cache was enabled and slow
  to read. Fixed - now it's possible to check whether there are queued
  commands which will abort playback of the current file without
  throwing other commands away.
- mp_input_check_interrupt() now prints a message if it returns
  true. This is especially useful because the failures caused by
  aborted stream reads can trigger error messages from other code that
  was doing the read; the new message makes it more obvious what the
  cause of the subsequent error messages is.
- It's now possible to again avoid making stdin non-blocking (which
  caused some issues) without reintroducing extra latency. The change
  will be done in a subsequent commit.
- Event sources that do not support select() should now have somewhat
  lower latency in certain situations as they will be checked both
  before and after select()/sleep in input reading; before the sleep
  always happened first even if such sources already had queued
  input. Before the key fifo was also handled in this manner (first
  key triggered select, but if multiple were read then rest could be
  delayed; however in most cases this didn't add latency in practice
  as after central code started doing command handling it queried for
  further commands with a max sleep time of 0).
- Key fifo limiting is more accurate now: it now counts actual
  commands intead of keycodes, and all queued keys are read
  immediately from input devices so they can be counted correctly.
- Since keypresses are now interpreted immediately, commands which
  change keybindings will no longer affect following keypresses that
  have already been read before the command is executed. This should
  not be an issue in practice with current keybinding behavior.

1 files changed, 206 insertions, 150 deletions

diff --git a/input/input.c b/input/input.c
index b8ff15a0c7..fe7acbb3e7 100644
--- a/input/input.c
+++ b/input/input.c
@@ -29,6 +29,7 @@
 #include <sys/time.h>
 #include <fcntl.h>
 #include <ctype.h>
+#include <assert.h>
 
 #include "input.h"
 #include "mp_fifo.h"
@@ -550,8 +551,6 @@ static const struct cmd_bind def_cmd_binds[] = {
 #define MP_MAX_CMD_FD 10
 #endif
 
-#define CMD_QUEUE_SIZE 100
-
 struct input_fd {
     int fd;
     union {
@@ -582,6 +581,13 @@ struct cmd_bind_section {
     struct cmd_bind_section *next;
 };
 
+struct cmd_queue {
+    struct mp_cmd *first;
+    struct mp_cmd *last;
+    int num_cmds;
+    int num_abort_cmds;
+};
+
 struct input_ctx {
     // Autorepeat stuff
     short ar_state;
@@ -590,6 +596,9 @@ struct input_ctx {
     // Autorepeat config
     unsigned int ar_delay;
     unsigned int ar_rate;
+    // Maximum number of queued commands from keypresses (limit to avoid
+    // repeated slow commands piling up)
+    int key_fifo_size;
 
     // these are the keys currently down
     int key_down[MP_MAX_KEY_DOWN];
@@ -605,14 +614,18 @@ struct input_ctx {
     struct cmd_bind *cmd_binds;
     struct cmd_bind *cmd_binds_default;
 
+    // Used to track whether we managed to read something while checking
+    // events sources. If yes, the sources may have more queued.
+    bool got_new_events;
+
     struct input_fd key_fds[MP_MAX_KEY_FD];
     unsigned int num_key_fd;
 
     struct input_fd cmd_fds[MP_MAX_CMD_FD];
     unsigned int num_cmd_fd;
 
-    mp_cmd_t *cmd_queue[CMD_QUEUE_SIZE];
-    unsigned int cmd_queue_length, cmd_queue_start, cmd_queue_end;
+    struct cmd_queue key_cmd_queue;
+    struct cmd_queue control_cmd_queue;
 };
 
 
@@ -685,6 +698,46 @@ static char *get_key_combo_name(int *keys, int max)
     return ret;
 }
 
+static bool is_abort_cmd(int cmd_id)
+{
+    switch (cmd_id) {
+    case MP_CMD_QUIT:
+    case MP_CMD_PLAY_TREE_STEP:
+    case MP_CMD_PLAY_TREE_UP_STEP:
+    case MP_CMD_PLAY_ALT_SRC_STEP:
+        return true;
+    }
+    return false;
+}
+
+static void queue_pop(struct cmd_queue *queue)
+{
+    assert(queue->num_cmds > 0);
+    struct mp_cmd *cmd = queue->first;
+    queue->first = cmd->queue_next;
+    queue->num_cmds--;
+    queue->num_abort_cmds -= is_abort_cmd(cmd->id);
+}
+
+static void queue_add(struct cmd_queue *queue, struct mp_cmd *cmd,
+                      bool at_head)
+{
+    if (!queue->num_cmds) {
+        queue->first = cmd;
+        queue->last = cmd;
+    } else if (at_head) {
+        queue->first->queue_prev = cmd;
+        cmd->queue_next = queue->first;
+        queue->first = cmd;
+    } else {
+        queue->last->queue_next = cmd;
+        cmd->queue_prev = queue->last;
+        queue->last = cmd;
+    }
+    queue->num_cmds++;
+    queue->num_abort_cmds += is_abort_cmd(cmd->id);
+}
+
 int mp_input_add_cmd_fd(struct input_ctx *ictx, int fd, int select,
                         int read_func(int fd, char *dest, int size),
                         int close_func(int fd))
@@ -1261,193 +1314,199 @@ static mp_cmd_t *check_autorepeat(struct input_ctx *ictx)
     return NULL;
 }
 
+void mp_input_feed_key(struct input_ctx *ictx, int code)
+{
+    ictx->got_new_events = true;
+    if (code == MP_INPUT_RELEASE_ALL) {
+        memset(ictx->key_down, 0, sizeof(ictx->key_down));
+        ictx->num_key_down = 0;
+        ictx->last_key_down = 0;
+        return;
+    }
+    struct mp_cmd *cmd = interpret_key(ictx, code);
+    if (!cmd)
+        return;
+    struct cmd_queue *queue = &ictx->key_cmd_queue;
+    if (queue->num_cmds >= ictx->key_fifo_size &&
+            (!is_abort_cmd(cmd->id) || queue->num_abort_cmds))
+        return;
+    queue_add(queue, cmd, false);
+}
+
+static void read_cmd_fd(struct input_ctx *ictx, struct input_fd *cmd_fd)
+{
+    int r;
+    char *text;
+    while ((r = read_cmd(cmd_fd, &text)) >= 0) {
+        ictx->got_new_events = true;
+        struct mp_cmd *cmd = mp_input_parse_cmd(text);
+        talloc_free(text);
+        if (cmd)
+            queue_add(&ictx->control_cmd_queue, cmd, false);
+        if (!cmd_fd->got_cmd)
+            return;
+    }
+    if (r == MP_INPUT_ERROR)
+        mp_tmsg(MSGT_INPUT, MSGL_ERR, "Error on command file descriptor %d\n",
+                cmd_fd->fd);
+    else if (r == MP_INPUT_DEAD)
+        cmd_fd->dead = true;
+}
+
+static void read_key_fd(struct input_ctx *ictx, struct input_fd *key_fd)
+{
+    int code = key_fd->read_func.key(key_fd->ctx, key_fd->fd);
+    if (code >= 0 || code == MP_INPUT_RELEASE_ALL) {
+        mp_input_feed_key(ictx, code);
+        return;
+    }
+
+    if (code == MP_INPUT_ERROR)
+        mp_tmsg(MSGT_INPUT, MSGL_ERR,
+                "Error on key input file descriptor %d\n", key_fd->fd);
+    else if (code == MP_INPUT_DEAD) {
+        mp_tmsg(MSGT_INPUT, MSGL_ERR,
+                "Dead key input on file descriptor %d\n", key_fd->fd);
+        key_fd->dead = true;
+    }
+}
 
 /**
  * \param time time to wait at most for an event in milliseconds
  */
-static mp_cmd_t *read_events(struct input_ctx *ictx, int time)
+static void read_events(struct input_ctx *ictx, int time)
 {
-    int i;
-    int got_cmd = 0;
+    ictx->got_new_events = false;
     struct input_fd *key_fds = ictx->key_fds;
     struct input_fd *cmd_fds = ictx->cmd_fds;
-    for (i = 0; i < ictx->num_key_fd; i++)
+    for (int i = 0; i < ictx->num_key_fd; i++)
         if (key_fds[i].dead) {
             mp_input_rm_key_fd(ictx, key_fds[i].fd);
             i--;
-        }
-    for (i = 0; i < ictx->num_cmd_fd; i++)
+        } else if (time && key_fds[i].no_select)
+            read_key_fd(ictx, &key_fds[i]);
+    for (int i = 0; i < ictx->num_cmd_fd; i++)
         if (cmd_fds[i].dead || cmd_fds[i].eof) {
             mp_input_rm_cmd_fd(ictx, cmd_fds[i].fd);
             i--;
-        } else if (cmd_fds[i].got_cmd)
-            got_cmd = 1;
+        } else if (time && cmd_fds[i].no_select)
+            read_cmd_fd(ictx, &cmd_fds[i]);
+    if (ictx->got_new_events)
+        time = 0;
 #ifdef HAVE_POSIX_SELECT
     fd_set fds;
     FD_ZERO(&fds);
-    if (!got_cmd) {
-        int max_fd = 0;
-        for (i = 0; i < ictx->num_key_fd; i++) {
-            if (key_fds[i].no_select)
-                continue;
-            if (key_fds[i].fd > max_fd)
-                max_fd = key_fds[i].fd;
-            FD_SET(key_fds[i].fd, &fds);
-        }
-        for (i = 0; i < ictx->num_cmd_fd; i++) {
-            if (cmd_fds[i].no_select)
-                continue;
-            if (cmd_fds[i].fd > max_fd)
-                max_fd = cmd_fds[i].fd;
-            FD_SET(cmd_fds[i].fd, &fds);
-        }
-        struct timeval tv, *time_val;
-        if (time >= 0) {
-            tv.tv_sec = time / 1000;
-            tv.tv_usec = (time % 1000) * 1000;
-            time_val = &tv;
-        } else
-            time_val = NULL;
-        if (select(max_fd + 1, &fds, NULL, NULL, time_val) < 0) {
-            if (errno != EINTR)
-                mp_tmsg(MSGT_INPUT, MSGL_ERR, "Select error: %s\n",
-                        strerror(errno));
-            FD_ZERO(&fds);
-        }
+    int max_fd = 0;
+    for (int i = 0; i < ictx->num_key_fd; i++) {
+        if (key_fds[i].no_select)
+            continue;
+        if (key_fds[i].fd > max_fd)
+            max_fd = key_fds[i].fd;
+        FD_SET(key_fds[i].fd, &fds);
+    }
+    for (int i = 0; i < ictx->num_cmd_fd; i++) {
+        if (cmd_fds[i].no_select)
+            continue;
+        if (cmd_fds[i].fd > max_fd)
+            max_fd = cmd_fds[i].fd;
+        FD_SET(cmd_fds[i].fd, &fds);
+    }
+    struct timeval tv, *time_val;
+    if (time >= 0) {
+        tv.tv_sec = time / 1000;
+        tv.tv_usec = (time % 1000) * 1000;
+        time_val = &tv;
+    } else
+        time_val = NULL;
+    if (select(max_fd + 1, &fds, NULL, NULL, time_val) < 0) {
+        if (errno != EINTR)
+            mp_tmsg(MSGT_INPUT, MSGL_ERR, "Select error: %s\n",
+                    strerror(errno));
+        FD_ZERO(&fds);
     }
 #else
-    if (!got_cmd && time)
+    if (time)
         usec_sleep(time * 1000);
 #endif
 
 
-    for (i = 0; i < ictx->num_key_fd; i++) {
+    for (int i = 0; i < ictx->num_key_fd; i++) {
 #ifdef HAVE_POSIX_SELECT
         if (!key_fds[i].no_select && !FD_ISSET(key_fds[i].fd, &fds))
             continue;
 #endif
-
-        int code;
-        while (1) {
-            code = key_fds[i].read_func.key(key_fds[i].ctx, key_fds[i].fd);
-            if (code < 0) {
-                if (code == MP_INPUT_RELEASE_ALL) {
-                    memset(ictx->key_down, 0, sizeof(ictx->key_down));
-                    ictx->num_key_down = 0;
-                    ictx->last_key_down = 0;
-                    continue;
-                }
-                break;
-            }
-            mp_cmd_t *ret = interpret_key(ictx, code);
-            if (ret)
-                return ret;
-        }
-        if (code == MP_INPUT_ERROR)
-            mp_tmsg(MSGT_INPUT, MSGL_ERR, "Error on key input "
-                    "file descriptor %d\n", key_fds[i].fd);
-        else if (code == MP_INPUT_DEAD) {
-            mp_tmsg(MSGT_INPUT, MSGL_ERR, "Dead key input on "
-                    "file descriptor %d\n", key_fds[i].fd);
-            key_fds[i].dead = 1;
-        }
+        read_key_fd(ictx, &key_fds[i]);
     }
-    mp_cmd_t *autorepeat_cmd = check_autorepeat(ictx);
-    if (autorepeat_cmd)
-        return autorepeat_cmd;
 
-    for (i = 0; i < ictx->num_cmd_fd; i++) {
+    for (int i = 0; i < ictx->num_cmd_fd; i++) {
 #ifdef HAVE_POSIX_SELECT
-        if (!cmd_fds[i].no_select && !FD_ISSET(cmd_fds[i].fd, &fds) &&
-            !cmd_fds[i].got_cmd)
+        if (!cmd_fds[i].no_select && !FD_ISSET(cmd_fds[i].fd, &fds))
             continue;
 #endif
-        char *cmd;
-        int r;
-        while ((r = read_cmd(&cmd_fds[i], &cmd)) >= 0) {
-            mp_cmd_t *ret = mp_input_parse_cmd(cmd);
-            talloc_free(cmd);
-            if (ret)
-                return ret;
-        }
-        if (r == MP_INPUT_ERROR)
-            mp_tmsg(MSGT_INPUT, MSGL_ERR, "Error on command "
-                    "file descriptor %d\n", cmd_fds[i].fd);
-        else if (r == MP_INPUT_DEAD)
-            cmd_fds[i].dead = 1;
+        read_cmd_fd(ictx, &cmd_fds[i]);
     }
-
-    return NULL;
 }
 
+/* To support blocking file descriptors we don't loop the read over
+ * every source until it's known to be empty. Instead we use this wrapper
+ * to run select() again.
+ */
+static void read_all_events(struct input_ctx *ictx, int time)
+{
+    while (1) {
+        read_events(ictx, time);
+        if (!ictx->got_new_events)
+            return;
+        time = 0;
+    }
+}
 
 int mp_input_queue_cmd(struct input_ctx *ictx, mp_cmd_t *cmd)
 {
-    if (!cmd || ictx->cmd_queue_length >= CMD_QUEUE_SIZE)
+    ictx->got_new_events = true;
+    if (!cmd)
         return 0;
-    ictx->cmd_queue[ictx->cmd_queue_end] = cmd;
-    ictx->cmd_queue_end = (ictx->cmd_queue_end + 1) % CMD_QUEUE_SIZE;
-    ictx->cmd_queue_length++;
+    queue_add(&ictx->control_cmd_queue, cmd, true);
     return 1;
 }
 
-static mp_cmd_t *get_queued_cmd(struct input_ctx *ictx, int peek_only)
-{
-    mp_cmd_t *ret;
-
-    if (ictx->cmd_queue_length == 0)
-        return NULL;
-
-    ret = ictx->cmd_queue[ictx->cmd_queue_start];
-
-    if (!peek_only) {
-        ictx->cmd_queue_length--;
-        ictx->cmd_queue_start = (ictx->cmd_queue_start + 1) % CMD_QUEUE_SIZE;
-    }
-
-    return ret;
-}
-
 /**
  * \param peek_only when set, the returned command stays in the queue.
  * Do not free the returned cmd whe you set this!
  */
 mp_cmd_t *mp_input_get_cmd(struct input_ctx *ictx, int time, int peek_only)
 {
-    mp_cmd_t *ret = NULL;
-    struct cmd_filter *cf;
-    int from_queue;
-
     if (async_quit_request)
         return mp_input_parse_cmd("quit 1");
-    while (1) {
-        from_queue = 1;
-        ret = get_queued_cmd(ictx, peek_only);
-        if (ret)
-            break;
-        from_queue = 0;
-        ret = read_events(ictx, time);
-        if (!ret) {
-            from_queue = 1;
-            ret = get_queued_cmd(ictx, peek_only);
-        }
-        break;
-    }
-    if (!ret)
-        return NULL;
 
-    for (cf = cmd_filters; cf; cf = cf->next) {
+    if (ictx->control_cmd_queue.num_cmds || ictx->key_cmd_queue.num_cmds)
+        time = 0;
+    read_all_events(ictx, time);
+    struct mp_cmd *ret;
+    struct cmd_queue *queue = &ictx->control_cmd_queue;
+    if (!queue->num_cmds)
+        queue = &ictx->key_cmd_queue;
+    if (!queue->num_cmds) {
+        queue = NULL;
+        ret = check_autorepeat(ictx);
+        if (!ret)
+            return NULL;
+    } else
+        ret = queue->first;
+
+    for (struct cmd_filter *cf = cmd_filters; cf; cf = cf->next) {
         if (cf->filter(ret, cf->ctx)) {
-            if (peek_only && from_queue)
-                // The filter ate the cmd, so we remove it from queue
-                ret = get_queued_cmd(ictx, 0);
+            // The filter ate the cmd, so remove it from the queue
+            if (queue)
+                queue_pop(queue);
             mp_cmd_free(ret);
-            return NULL;
+            // Retry with next command
+            return mp_input_get_cmd(ictx, 0, peek_only);
         }
     }
 
-    if (!from_queue && peek_only)
-        mp_input_queue_cmd(ictx, ret);
+    if (!peek_only && queue)
+        queue_pop(queue);
 
     return ret;
 }
@@ -1752,6 +1811,7 @@ struct input_ctx *mp_input_init(struct input_conf *input_conf)
 {
     struct input_ctx *ictx = talloc_ptrtype(NULL, ictx);
     *ictx = (struct input_ctx){
+        .key_fifo_size = input_conf->key_fifo_size,
         .ar_state = -1,
         .ar_delay = input_conf->ar_delay,
         .ar_rate = input_conf->ar_rate,
@@ -1917,19 +1977,15 @@ static int print_cmd_list(m_option_t *cfg)
  */
 int mp_input_check_interrupt(struct input_ctx *ictx, int time)
 {
-    mp_cmd_t *cmd;
-    if ((cmd = mp_input_get_cmd(ictx, time, 1)) == NULL)
-        return 0;
-    switch (cmd->id) {
-    case MP_CMD_QUIT:
-    case MP_CMD_PLAY_TREE_STEP:
-    case MP_CMD_PLAY_TREE_UP_STEP:
-    case MP_CMD_PLAY_ALT_SRC_STEP:
-        // The cmd will be executed when we are back in the main loop
-        return 1;
+    for (int i = 0; ; i++) {
+        if (async_quit_request || ictx->key_cmd_queue.num_abort_cmds ||
+                ictx->control_cmd_queue.num_abort_cmds) {
+            mp_tmsg(MSGT_INPUT, MSGL_WARN, "Received command to move to "
+                   "another file. Aborting current processing.\n");
+            return true;
+        }
+        if (i)
+            return false;
+        read_all_events(ictx, time);
     }
-    // remove the cmd from the queue
-    cmd = mp_input_get_cmd(ictx, time, 0);
-    mp_cmd_free(cmd);
-    return 0;
 }