input: use libwaio for pipe input on Windows

Use libwaio to read from pipes (stdin or named pipes) on Windows. This
liberates us from nasty issues, such as pipes (as created by most
programs) not being possible to read in a non-blocking or event-driven
way. Although it would be possible to do that in a somewhat sane way
on Vista+, it's still not easy, and on XP it's especially hard. libwaio
handles these things for us.

Move pipe.c to pipe-unix.c, and remove Windows specific things. Also
adjust the input.c code to make this work cleanly.

DOCS/crosscompile-mingw.md

@@ -26,6 +26,15 @@ created scripts to help ease the process. These are the two recommended methods:
 Note that MinGW environments included in Linux distributions are often broken,
 outdated and useless, and usually don't use MinGW-w64.
 
+Additional dependencies
+-----------------------
+
+You need a pthread wrapper. It must be interoperable with native Windows
+threads. pthreads-win32 or MinGW pthreads might work.
+
+If you want ``--input-file=...`` to work, you need libwaio. It's available
+from: git://midipix.org/waio
+
 Example with MXE
 ----------------
 

input/input.c

@@ -1273,8 +1273,13 @@ struct input_ctx *mp_input_init(struct mpv_global *global)
 
     ictx->win_drag = global->opts->allow_win_drag;
 
-    if (input_conf->in_file && input_conf->in_file[0])
-        mp_input_add_pipe(ictx, input_conf->in_file);
+    if (input_conf->in_file && input_conf->in_file[0]) {
+#if !defined(__MINGW32__) || HAVE_WAIO
+        mp_input_pipe_add(ictx, input_conf->in_file);
+#else
+        MP_ERR(ictx, "Pipes not available.\n");
+#endif
+    }
 
     return ictx;
 }
@@ -1399,10 +1404,12 @@ void mp_input_src_kill(struct mp_input_src *src)
             MP_TARRAY_REMOVE_AT(ictx->sources, ictx->num_sources, n);
             input_unlock(ictx);
             write(src->in->wakeup[1], &(char){0}, 1);
-            if (src->close)
-                src->close(src);
+            if (src->cancel)
+                src->cancel(src);
             if (src->in->thread_running)
                 pthread_join(src->in->thread, NULL);
+            if (src->uninit)
+                src->uninit(src);
             talloc_free(src);
             return;
         }

input/input.h

@@ -91,34 +91,38 @@ struct mp_input_src {
 
     struct mp_input_src_internal *in;
 
-    // If not-NULL: called before destroying the input_src. Should close the
-    // underlying device, and free all memory.
-    void (*close)(struct mp_input_src *src);
+    // If not-NULL: called before destroying the input_src. Should unblock the
+    // reader loop, and make it exit. (Use with mp_input_add_thread_src().)
+    void (*cancel)(struct mp_input_src *src);
+    // Called after the reader thread returns, and cancel() won't be called
+    // again. This should make sure that nothing after this call accesses src.
+    void (*uninit)(struct mp_input_src *src);
 
     // For free use by the implementer.
     void *priv;
 };
 
-/* Add a new input source. The input code can create a new thread, which feeds
- * keys or commands to input_ctx. mp_input_src.close must be set.
- */
+// Add a new input source. The input code can create a new thread, which feeds
+// keys or commands to input_ctx. mp_input_src.uninit must be set.
+// mp_input_src_kill() must not be called by anything after init.
 struct mp_input_src *mp_input_add_src(struct input_ctx *ictx);
 
 // Add an input source that runs on a thread. The source is automatically
 // removed if the thread loop exits.
 //  ctx: this is passed to loop_fn.
 //  loop_fn: this is called once inside of a new thread, and should not return
-//      until all input is read, or src->close is called by another thread.
+//      until all input is read, or src->cancel is called by another thread.
 //      You must call mp_input_src_init_done(src) early during init to signal
-//      success (then src->close may be called at a later point); on failure,
+//      success (then src->cancel may be called at a later point); on failure,
 //      return from loop_fn immediately.
 // Returns >=0 on success, <0 on failure to allocate resources.
-// Do not set src->close after mp_input_src_init_done() has been called.
+// Do not set src->cancel after mp_input_src_init_done() has been called.
 int mp_input_add_thread_src(struct input_ctx *ictx, void *ctx,
     void (*loop_fn)(struct mp_input_src *src, void *ctx));
 
 // Signal successful init.
 // Must be called on the same thread as loop_fn (see mp_input_add_thread_src()).
+// Set src->cancel and src->uninit (if needed) before calling this.
 void mp_input_src_init_done(struct mp_input_src *src);
 
 // Currently only with mp_input_add_thread_src().
@@ -241,7 +245,7 @@ bool mp_input_use_alt_gr(struct input_ctx *ictx);
 void mp_input_run_cmd(struct input_ctx *ictx, int def_flags, const char **cmd,
                       const char *location);
 
-void mp_input_add_pipe(struct input_ctx *ictx, const char *filename);
+void mp_input_pipe_add(struct input_ctx *ictx, const char *filename);
 void mp_input_joystick_add(struct input_ctx *ictx, char *dev);
 void mp_input_lirc_add(struct input_ctx *ictx, char *lirc_configfile);
 

input/pipe-unix.c

@@ -0,0 +1,59 @@
+#include <stdio.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <unistd.h>
+
+#include <poll.h>
+
+#include "common/msg.h"
+#include "osdep/io.h"
+#include "input.h"
+#include "cmd_parse.h"
+
+static void read_pipe_thread(struct mp_input_src *src, void *param)
+{
+    void *tmp = talloc_new(NULL);
+    char *filename = talloc_strdup(tmp, param); // param deallocates after init
+    int wakeup_fd = mp_input_src_get_wakeup_fd(src);
+    int fd = -1;
+
+    struct mp_log *log = src->log;
+
+    int mode = O_RDONLY;
+    // Use RDWR for FIFOs to ensure they stay open over multiple accesses.
+    struct stat st;
+    if (stat(filename, &st) == 0 && S_ISFIFO(st.st_mode))
+        mode = O_RDWR;
+    fd = open(filename, mode);
+    if (fd < 0) {
+        mp_err(log, "Can't open %s.\n", filename);
+        goto done;
+    }
+
+    mp_input_src_init_done(src);
+
+    while (1) {
+        struct pollfd fds[2] = {
+            { .fd = fd, .events = POLLIN },
+            { .fd = wakeup_fd, .events = POLLIN },
+        };
+        poll(fds, 2, -1);
+        if (!(fds[0].revents & POLLIN))
+            break;
+        char buffer[128];
+        int r = read(fd, buffer, sizeof(buffer));
+        if (r <= 0)
+            break;
+        mp_input_src_feed_cmd_text(src, buffer, r);
+    }
+
+done:
+    close(fd);
+    talloc_free(tmp);
+}
+
+void mp_input_pipe_add(struct input_ctx *ictx, const char *filename)
+{
+    mp_input_add_thread_src(ictx, (void *)filename, read_pipe_thread);
+}

input/pipe-win32.c

@@ -0,0 +1,91 @@
+#include <pthread.h>
+#include <stdio.h>
+#include <windows.h>
+#include <io.h>
+
+#include <stdint.h>
+#include <waio/waio.h>
+
+#include "common/msg.h"
+#include "osdep/io.h"
+#include "input.h"
+
+static void request_cancel(struct mp_input_src *src)
+{
+    HANDLE terminate = src->priv;
+
+    MP_VERBOSE(src, "Exiting...\n");
+    SetEvent(terminate);
+}
+
+static void uninit(struct mp_input_src *src)
+{
+    HANDLE terminate = src->priv;
+
+    CloseHandle(terminate);
+    MP_VERBOSE(src, "Exited.\n");
+}
+
+static void read_pipe_thread(struct mp_input_src *src, void *param)
+{
+    char *filename = talloc_strdup(src, param);
+
+    struct waio_cx_interface *waio = NULL;
+    int mode = O_RDONLY;
+    int fd = -1;
+    bool close_fd = true;
+    if (strcmp(filename, "/dev/stdin") == 0) { // for symmetry with unix
+        fd = STDIN_FILENO;
+        close_fd = false;
+    }
+    if (fd < 0)
+        fd = open(filename, mode);
+    if (fd < 0) {
+        MP_ERR(src, "Can't open %s.\n", filename);
+        goto done;
+    }
+
+    waio = waio_alloc((void *)_get_osfhandle(fd), 0, NULL, NULL);
+    if (!waio) {
+        MP_ERR(src, "Can't initialize win32 file reader.\n");
+        goto done;
+    }
+
+    HANDLE terminate = CreateEvent(NULL, TRUE, FALSE, NULL);
+    if (!terminate)
+        goto done;
+
+    src->priv = terminate;
+    src->cancel = request_cancel;
+    src->uninit = uninit;
+    mp_input_src_init_done(src);
+
+    char buffer[128];
+    struct waio_aiocb cb = {
+        .aio_buf = buffer,
+        .aio_nbytes = sizeof(buffer),
+        .hsignal = terminate,
+    };
+    while (1) {
+        if (waio_read(waio, &cb)) {
+            MP_ERR(src, "Read operation failed.\n");
+            break;
+        }
+        if (waio_suspend(waio, (const struct waio_aiocb *[]){&cb}, 1, NULL))
+            break;
+        ssize_t r = waio_return(waio, &cb);
+        if (r <= 0)
+            break; // EOF or error
+        mp_input_src_feed_cmd_text(src, buffer, r);
+    }
+
+done:
+    waio_free(waio);
+    if (close_fd)
+        close(fd);
+}
+
+void mp_input_pipe_add(struct input_ctx *ictx, const char *filename)
+{
+    mp_input_add_thread_src(ictx, (void *)filename, read_pipe_thread);
+}

input/pipe.c

@@ -1,133 +0,0 @@
-#include <pthread.h>
-#include <stdio.h>
-#include <sys/types.h>
-#include <sys/stat.h>
-#include <fcntl.h>
-#include <unistd.h>
-
-#ifndef __MINGW32__
-#include <poll.h>
-#endif
-
-#include "common/msg.h"
-#include "misc/bstr.h"
-#include "osdep/io.h"
-#include "input.h"
-#include "cmd_parse.h"
-
-static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
-
-struct priv {
-    struct mp_log *log;
-    char *filename;
-    struct mp_input_src *src;
-    int wakeup_pipe[2];
-};
-
-static void *reader_thread(void *ctx)
-{
-    struct priv *p = ctx;
-    pthread_detach(pthread_self());
-
-    int mode = O_RDONLY;
-#ifndef __MINGW32__
-    // Use RDWR for FIFOs to ensure they stay open over multiple accesses.
-    // Note that on Windows due to how the API works, using RDONLY should
-    // be ok.
-    struct stat st;
-    if (stat(p->filename, &st) == 0 && S_ISFIFO(st.st_mode))
-        mode = O_RDWR;
-#endif
-    int fd = -1;
-    bool close_fd = true;
-    if (strcmp(p->filename, "/dev/stdin") == 0) { // mainly for win32
-        fd = STDIN_FILENO;
-        close_fd = false;
-    }
-    if (fd < 0)
-        fd = open(p->filename, mode);
-    if (fd < 0) {
-        MP_ERR(p, "Can't open %s.\n", p->filename);
-        goto done;
-    }
-
-    while (1) {
-#ifndef __MINGW32__
-        struct pollfd fds[2] = {
-            { .fd = fd, .events = POLLIN },
-            { .fd = p->wakeup_pipe[0], .events = POLLIN },
-        };
-        poll(fds, 2, -1);
-        if (!(fds[0].revents & POLLIN))
-            break;
-#endif
-        char buffer[128];
-        int r = read(fd, buffer, sizeof(buffer));
-        if (r <= 0)
-            break;
-
-        pthread_mutex_lock(&lock);
-        if (!p->src) {
-            pthread_mutex_unlock(&lock);
-            break;
-        }
-        mp_input_src_feed_cmd_text(p->src, buffer, r);
-        pthread_mutex_unlock(&lock);
-    }
-
-    if (close_fd)
-        close(fd);
-
-done:
-    pthread_mutex_lock(&lock);
-    if (p->src)
-        p->src->priv = NULL;
-    pthread_mutex_unlock(&lock);
-    close(p->wakeup_pipe[0]);
-    close(p->wakeup_pipe[1]);
-    talloc_free(p);
-    return NULL;
-}
-
-static void close_pipe(struct mp_input_src *src)
-{
-    pthread_mutex_lock(&lock);
-    struct priv *p = src->priv;
-    // Windows pipe have a severe problem: they can't be made non-blocking (not
-    // after creation), and you can't wait on them. The only things that work
-    // are cancellation (Vista+, broken in wine) or forceful thread termination.
-    // So don't bother with "correct" termination, and just abandon the reader
-    // thread.
-    // On Unix, we interrupt it using the wakeup pipe.
-    if (p) {
-#ifndef __MINGW32__
-        write(p->wakeup_pipe[1], &(char){0}, 1);
-#endif
-        p->src = NULL;
-    }
-    pthread_mutex_unlock(&lock);
-}
-
-void mp_input_add_pipe(struct input_ctx *ictx, const char *filename)
-{
-    struct mp_input_src *src = mp_input_add_src(ictx);
-    if (!src)
-        return;
-
-    struct priv *p = talloc_zero(NULL, struct priv);
-    src->priv = p;
-    p->filename = talloc_strdup(p, filename);
-    p->src = src;
-    p->log = mp_log_new(p, src->log, NULL);
-    mp_make_wakeup_pipe(p->wakeup_pipe);
-
-    pthread_t thread;
-    if (pthread_create(&thread, NULL, reader_thread, p)) {
-        close(p->wakeup_pipe[0]);
-        close(p->wakeup_pipe[1]);
-        talloc_free(p);
-        mp_input_src_kill(src);
-    } else {
-        src->close = close_pipe;
-    }
-}

old-configure

@@ -961,6 +961,7 @@ cat > $TMPC << EOF
 #define HAVE_SYS_MMAN_H 1
 #define HAVE_NANOSLEEP 1
 #define HAVE_SDL1 0
+#define HAVE_WAIO 0
 
 #define DEFAULT_CDROM_DEVICE "/dev/cdrom"
 #define DEFAULT_DVD_DEVICE   "/dev/dvd"

old-makefile

@@ -176,7 +176,7 @@ SOURCES = audio/audio.c \
           input/event.c \
           input/input.c \
           input/keycodes.c \
-          input/pipe.c \
+          input/pipe-unix.c \
           misc/bstr.c \
           misc/charset_conv.c \
           misc/dispatch.c \

wscript

@@ -103,6 +103,11 @@ main_dependencies = [
         'name': 'sys-mman-h',
         'desc': 'mman.h',
         'func': check_statement('sys/mman.h', 'mmap(0, 0, 0, 0, 0, 0)')
+    }, {
+        'name': 'mingw',
+        'desc': 'MinGW',
+        'deps': [ 'os-win32' ],
+        'func': check_statement('stddef.h', 'int x = __MINGW32__')
     }, {
         'name': 'pthreads',
         'desc': 'POSIX threads',
@@ -159,6 +164,12 @@ iconv support use --disable-iconv.",
         'desc': 'w32 priority API',
         'deps_any': [ 'os-win32', 'os-cygwin'],
         'func': check_true
+    }, {
+        'name': '--waio',
+        'desc': 'libwaio for win32',
+        'deps': [ 'os-win32', 'mingw' ],
+        'func': check_libs(['waio'],
+                    check_statement('waio/waio.h', 'waio_alloc(0, 0, 0, 0)')),
     }, {
         'name': 'videoio',
         'desc': 'videoio.h',

wscript_build.py

@@ -190,7 +190,8 @@ def build(ctx):
         ( "input/event.c" ),
         ( "input/input.c" ),
         ( "input/keycodes.c" ),
-        ( "input/pipe.c" ),
+        ( "input/pipe-unix.c",                   "!mingw" ),
+        ( "input/pipe-win32.c",                  "waio" ),
         ( "input/joystick.c",                    "joystick" ),
         ( "input/lirc.c",                        "lirc" ),