ipc: add Windows implementation with named pipes

This implements the JSON IPC protocol with named pipes, which are
probably the closest Windows equivalent to Unix domain sockets in terms
of functionality. Like with Unix sockets, this will allow mpv to listen
for IPC connections and handle multiple IPC clients at once. A few cross
platform libraries and frameworks (Qt, node.js) use named pipes for IPC
on Windows and Unix sockets on Linux and Unix, so hopefully this will
ease the creation of portable JSON IPC clients.

Unlike the Unix implementation, this doesn't share code with
--input-file, meaning --input-file on Windows won't understand JSON
commands (yet.) Sharing code and removing the separate implementation in
pipe-win32.c is definitely a possible future improvement.

1 files changed, 408 insertions, 0 deletions

diff --git a/input/ipc-win.c b/input/ipc-win.c
new file mode 100644
index 0000000000..3c5f3832aa
--- /dev/null
+++ b/input/ipc-win.c
@@ -0,0 +1,408 @@
+/*
+ * This file is part of mpv.
+ *
+ * mpv is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * mpv is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with mpv.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include <windows.h>
+
+#include "config.h"
+
+#include "osdep/io.h"
+#include "osdep/threads.h"
+#include "osdep/windows_utils.h"
+
+#include "common/common.h"
+#include "common/global.h"
+#include "common/msg.h"
+#include "input/input.h"
+#include "libmpv/client.h"
+#include "options/options.h"
+#include "player/client.h"
+
+struct mp_ipc_ctx {
+    struct mp_log *log;
+    struct mp_client_api *client_api;
+    const wchar_t *path;
+
+    pthread_t thread;
+    HANDLE death_event;
+};
+
+struct client_arg {
+    struct mp_log *log;
+    struct mpv_handle *client;
+
+    char *client_name;
+    HANDLE client_h;
+    bool writable;
+    OVERLAPPED write_ol;
+};
+
+static void wakeup_cb(void *d)
+{
+    HANDLE event = d;
+    SetEvent(event);
+}
+
+// Wrapper for ReadFile that treats ERROR_IO_PENDING as success
+static DWORD async_read(HANDLE file, void *buf, unsigned size, OVERLAPPED* ol)
+{
+    DWORD err = ReadFile(file, buf, size, NULL, ol) ? 0 : GetLastError();
+    return err == ERROR_IO_PENDING ? 0 : err;
+}
+
+// Wrapper for WriteFile that treats ERROR_IO_PENDING as success
+static DWORD async_write(HANDLE file, const void *buf, unsigned size, OVERLAPPED* ol)
+{
+    DWORD err = WriteFile(file, buf, size, NULL, ol) ? 0 : GetLastError();
+    return err == ERROR_IO_PENDING ? 0 : err;
+}
+
+static bool pipe_error_is_fatal(DWORD error)
+{
+    switch (error) {
+    case 0:
+    case ERROR_HANDLE_EOF:
+    case ERROR_BROKEN_PIPE:
+    case ERROR_PIPE_NOT_CONNECTED:
+    case ERROR_NO_DATA:
+        return false;
+    }
+    return true;
+}
+
+static DWORD ipc_write_str(struct client_arg *arg, const char *buf)
+{
+    DWORD error = 0;
+
+    if ((error = async_write(arg->client_h, buf, strlen(buf), &arg->write_ol)))
+        goto done;
+    if (!GetOverlappedResult(arg->client_h, &arg->write_ol, &(DWORD){0}, TRUE)) {
+        error = GetLastError();
+        goto done;
+    }
+
+done:
+    if (pipe_error_is_fatal(error)) {
+        MP_VERBOSE(arg, "Error writing to pipe: %s\n",
+            mp_HRESULT_to_str(HRESULT_FROM_WIN32(error)));
+    }
+
+    if (error)
+        arg->writable = false;
+    return error;
+}
+
+static void report_read_error(struct client_arg *arg, DWORD error)
+{
+    // Only report the error if it's not just due to the pipe closing
+    if (pipe_error_is_fatal(error)) {
+        MP_ERR(arg, "Error reading from pipe: %s\n",
+            mp_HRESULT_to_str(HRESULT_FROM_WIN32(error)));
+    } else {
+        MP_VERBOSE(arg, "Client disconnected\n");
+    }
+}
+
+static void *client_thread(void *p)
+{
+    pthread_detach(pthread_self());
+
+    struct client_arg *arg = p;
+    char buf[4096];
+    HANDLE wakeup_event = CreateEventW(NULL, TRUE, FALSE, NULL);
+    OVERLAPPED ol = { .hEvent = CreateEventW(NULL, TRUE, TRUE, NULL) };
+    bstr client_msg = { talloc_strdup(NULL, ""), 0 };
+    DWORD ioerr = 0;
+    DWORD r;
+
+    mpthread_set_name(arg->client_name);
+
+    arg->write_ol.hEvent = CreateEventW(NULL, TRUE, TRUE, NULL);
+    if (!wakeup_event || !ol.hEvent || !arg->write_ol.hEvent) {
+        MP_ERR(arg, "Couldn't create events\n");
+        goto done;
+    }
+
+    MP_VERBOSE(arg, "Client connected\n");
+
+    mpv_set_wakeup_callback(arg->client, wakeup_cb, wakeup_event);
+    mpv_suspend(arg->client);
+
+    // Do the first read operation on the pipe
+    if ((ioerr = async_read(arg->client_h, buf, 4096, &ol))) {
+        report_read_error(arg, ioerr);
+        goto done;
+    }
+
+    while (1) {
+        HANDLE handles[] = { wakeup_event, ol.hEvent };
+        int n = WaitForMultipleObjects(2, handles, FALSE, 0);
+        if (n == WAIT_TIMEOUT) {
+            mpv_resume(arg->client);
+            n = WaitForMultipleObjects(2, handles, FALSE, INFINITE);
+            mpv_suspend(arg->client);
+        }
+
+        switch (n) {
+        case WAIT_OBJECT_0: // wakeup_event
+            ResetEvent(wakeup_event);
+
+            while (1) {
+                mpv_event *event = mpv_wait_event(arg->client, 0);
+
+                if (event->event_id == MPV_EVENT_NONE)
+                    break;
+
+                if (event->event_id == MPV_EVENT_SHUTDOWN)
+                    goto done;
+
+                if (!arg->writable)
+                    continue;
+
+                char *event_msg = mp_json_encode_event(event);
+                if (!event_msg) {
+                    MP_ERR(arg, "Encoding error\n");
+                    goto done;
+                }
+
+                ipc_write_str(arg, event_msg);
+                talloc_free(event_msg);
+            }
+
+            break;
+        case WAIT_OBJECT_0 + 1: // ol.hEvent
+            // Complete the read operation on the pipe
+            if (!GetOverlappedResult(arg->client_h, &ol, &r, TRUE)) {
+                report_read_error(arg, GetLastError());
+                goto done;
+            }
+
+            bstr_xappend(NULL, &client_msg, (bstr){buf, r});
+            while (bstrchr(client_msg, '\n') != -1) {
+                char *reply_msg = mp_ipc_consume_next_command(arg->client,
+                    NULL, &client_msg);
+                if (reply_msg && arg->writable)
+                    ipc_write_str(arg, reply_msg);
+                talloc_free(reply_msg);
+            }
+
+            // Begin the next read operation on the pipe
+            if ((ioerr = async_read(arg->client_h, buf, 4096, &ol))) {
+                report_read_error(arg, ioerr);
+                goto done;
+            }
+            break;
+        default:
+            MP_ERR(arg, "WaitForMultipleObjects failed\n");
+            goto done;
+        }
+    }
+
+done:
+    if (client_msg.len > 0)
+        MP_WARN(arg, "Ignoring unterminated command on disconnect.\n");
+
+    if (CancelIoEx(arg->client_h, &ol) || GetLastError() != ERROR_NOT_FOUND)
+        GetOverlappedResult(arg->client_h, &ol, &(DWORD){0}, TRUE);
+    if (wakeup_event)
+        CloseHandle(wakeup_event);
+    if (ol.hEvent)
+        CloseHandle(ol.hEvent);
+    if (arg->write_ol.hEvent)
+        CloseHandle(arg->write_ol.hEvent);
+
+    CloseHandle(arg->client_h);
+    mpv_detach_destroy(arg->client);
+    talloc_free(arg);
+    return NULL;
+}
+
+static void ipc_start_client(struct mp_ipc_ctx *ctx, struct client_arg *client)
+{
+    client->client = mp_new_client(ctx->client_api, client->client_name),
+    client->log    = mp_client_get_log(client->client);
+
+    pthread_t client_thr;
+    if (pthread_create(&client_thr, NULL, client_thread, client)) {
+        mpv_detach_destroy(client->client);
+        CloseHandle(client->client_h);
+        talloc_free(client);
+    }
+}
+
+static void ipc_start_client_json(struct mp_ipc_ctx *ctx, int id, HANDLE h)
+{
+    struct client_arg *client = talloc_ptrtype(NULL, client);
+    *client = (struct client_arg){
+        .client_name = talloc_asprintf(client, "ipc-%d", id),
+        .client_h = h,
+        .writable = true,
+    };
+
+    ipc_start_client(ctx, client);
+}
+
+static void *ipc_thread(void *p)
+{
+    // Use PIPE_TYPE_MESSAGE | PIPE_READMODE_BYTE so message framing is
+    // maintained for message-mode clients, but byte-mode clients can still
+    // connect, send and receive data. This is the most compatible mode.
+    static const DWORD state =
+        PIPE_TYPE_MESSAGE | PIPE_READMODE_BYTE | PIPE_WAIT |
+        PIPE_REJECT_REMOTE_CLIENTS;
+    static const DWORD mode =
+        PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED;
+    static const DWORD bufsiz = 4096;
+
+    struct mp_ipc_ctx *arg = p;
+    HANDLE server = INVALID_HANDLE_VALUE;
+    HANDLE client = INVALID_HANDLE_VALUE;
+    int client_num = 0;
+
+    mpthread_set_name("ipc named pipe listener");
+    MP_VERBOSE(arg, "Starting IPC master\n");
+
+    OVERLAPPED ol = { .hEvent = CreateEventW(NULL, TRUE, TRUE, NULL) };
+    if (!ol.hEvent) {
+        MP_ERR(arg, "Couldn't create event");
+        goto done;
+    }
+
+    server = CreateNamedPipeW(arg->path, mode | FILE_FLAG_FIRST_PIPE_INSTANCE,
+        state, PIPE_UNLIMITED_INSTANCES, bufsiz, bufsiz, 0, NULL);
+    if (server == INVALID_HANDLE_VALUE) {
+        MP_ERR(arg, "Couldn't create first pipe instance: %s\n",
+            mp_LastError_to_str());
+        goto done;
+    }
+
+    while (1) {
+        DWORD err = ConnectNamedPipe(server, &ol) ? 0 : GetLastError();
+
+        if (err == ERROR_IO_PENDING) {
+            int n = WaitForMultipleObjects(2, (HANDLE[]) {
+                arg->death_event,
+                ol.hEvent,
+            }, FALSE, INFINITE) - WAIT_OBJECT_0;
+
+            switch (n) {
+            case 0:
+                // Stop waiting for new clients
+                CancelIo(server);
+                GetOverlappedResult(server, &ol, &(DWORD){0}, TRUE);
+                goto done;
+            case 1:
+                // Complete the ConnectNamedPipe request
+                err = GetOverlappedResult(server, &ol, &(DWORD){0}, TRUE)
+                    ? 0 : GetLastError();
+                break;
+            default:
+                MP_ERR(arg, "WaitForMultipleObjects failed\n");
+                goto done;
+            }
+        }
+
+        // ERROR_PIPE_CONNECTED is returned if a client connects before
+        // ConnectNamedPipe is called. ERROR_NO_DATA is returned if a client
+        // connects, (possibly) writes data and exits before ConnectNamedPipe
+        // is called. Both cases should be handled as normal connections.
+        if (err == ERROR_PIPE_CONNECTED || err == ERROR_NO_DATA)
+            err = 0;
+
+        if (err) {
+            MP_ERR(arg, "ConnectNamedPipe failed: %s\n",
+                mp_HRESULT_to_str(HRESULT_FROM_WIN32(err)));
+            goto done;
+        }
+
+        // Create the next pipe instance before the client thread to avoid the
+        // theoretical race condition where the client thread immediately
+        // closes the handle and there are no active instances of the pipe
+        client = server;
+        server = CreateNamedPipeW(arg->path, mode, state,
+            PIPE_UNLIMITED_INSTANCES, bufsiz, bufsiz, 0, NULL);
+        if (server == INVALID_HANDLE_VALUE) {
+            MP_ERR(arg, "Couldn't create additional pipe instance: %s\n",
+                mp_LastError_to_str());
+            goto done;
+        }
+
+        ipc_start_client_json(arg, client_num++, client);
+        client = NULL;
+    }
+
+done:
+    if (client != INVALID_HANDLE_VALUE)
+        CloseHandle(client);
+    if (server != INVALID_HANDLE_VALUE)
+        CloseHandle(server);
+    if (ol.hEvent)
+        CloseHandle(ol.hEvent);
+    return NULL;
+}
+
+struct mp_ipc_ctx *mp_init_ipc(struct mp_client_api *client_api,
+                               struct mpv_global *global)
+{
+    struct MPOpts *opts = global->opts;
+
+    struct mp_ipc_ctx *arg = talloc_ptrtype(NULL, arg);
+    *arg = (struct mp_ipc_ctx){
+        .log = mp_log_new(arg, global->log, "ipc"),
+        .client_api = client_api,
+    };
+
+    if (!opts->ipc_path || !*opts->ipc_path)
+        goto out;
+
+    // Ensure the path is a legal Win32 pipe name by prepending \\.\pipe\ if
+    // it's not already present. Qt's QLocalSocket uses the same logic, so
+    // cross-platform programs that use paths like /tmp/mpv-socket should just
+    // work. (Win32 converts this path to \Device\NamedPipe\tmp\mpv-socket)
+    if (!strncmp(opts->ipc_path, "\\\\.\\pipe\\", 9)) {
+        arg->path = mp_from_utf8(arg, opts->ipc_path);
+    } else {
+        char *path = talloc_asprintf(NULL, "\\\\.\\pipe\\%s", opts->ipc_path);
+        arg->path = mp_from_utf8(arg, path);
+        talloc_free(path);
+    }
+
+    if (!(arg->death_event = CreateEventW(NULL, TRUE, FALSE, NULL)))
+        goto out;
+
+    if (pthread_create(&arg->thread, NULL, ipc_thread, arg))
+        goto out;
+
+    return arg;
+
+out:
+    if (arg->death_event)
+        CloseHandle(arg->death_event);
+    talloc_free(arg);
+    return NULL;
+}
+
+void mp_uninit_ipc(struct mp_ipc_ctx *arg)
+{
+    if (!arg)
+        return;
+
+    SetEvent(arg->death_event);
+    pthread_join(arg->thread, NULL);
+
+    CloseHandle(arg->death_event);
+    talloc_free(arg);
+}