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, 39 insertions, 427 deletions

diff --git a/input/ipc.c b/input/ipc.c
index c628fc9ea4..c7563e30d8 100644
--- a/input/ipc.c
+++ b/input/ipc.c
@@ -15,58 +15,16 @@
  * License along with mpv.  If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include <pthread.h>
-#include <assert.h>
-#include <errno.h>
-#include <unistd.h>
-
-#include <poll.h>
-#include <sys/types.h>
-#include <sys/socket.h>
-#include <sys/stat.h>
-#include <sys/un.h>
-
 #include "config.h"
 
-#include "osdep/io.h"
-#include "osdep/threads.h"
-
-#include "common/common.h"
-#include "common/global.h"
 #include "common/msg.h"
 #include "input/input.h"
-#include "libmpv/client.h"
-#include "misc/bstr.h"
 #include "misc/json.h"
 #include "options/m_option.h"
 #include "options/options.h"
 #include "options/path.h"
 #include "player/client.h"
 
-#ifndef MSG_NOSIGNAL
-#define MSG_NOSIGNAL 0
-#endif
-
-struct mp_ipc_ctx {
-    struct mp_log *log;
-    struct mp_client_api *client_api;
-    const char *path;
-
-    pthread_t thread;
-    int death_pipe[2];
-};
-
-struct client_arg {
-    struct mp_log *log;
-    struct mpv_handle *client;
-
-    char *client_name;
-    int client_fd;
-    bool close_client_fd;
-
-    bool writable;
-};
-
 static mpv_node *mpv_node_map_get(mpv_node *src, const char *key)
 {
     if (src->format != MPV_FORMAT_NODE_MAP)
@@ -219,7 +177,7 @@ static void mpv_event_to_node(void *ta_parent, mpv_event *event, mpv_node *dst)
     }
 }
 
-static char *json_encode_event(mpv_event *event)
+char *mp_json_encode_event(mpv_event *event)
 {
     void *ta_parent = talloc_new(NULL);
     mpv_node event_node = {.format = MPV_FORMAT_NODE_MAP, .u.list = NULL};
@@ -236,11 +194,12 @@ static char *json_encode_event(mpv_event *event)
 }
 
 // Function is allowed to modify src[n].
-static char *json_execute_command(struct client_arg *arg, void *ta_parent,
+static char *json_execute_command(struct mpv_handle *client, void *ta_parent,
                                   char *src)
 {
     int rc;
     const char *cmd = NULL;
+    struct mp_log *log = mp_client_get_log(client);
 
     mpv_node msg_node;
     mpv_node reply_node = {.format = MPV_FORMAT_NODE_MAP, .u.list = NULL};
@@ -248,7 +207,7 @@ static char *json_execute_command(struct client_arg *arg, void *ta_parent,
 
     rc = json_parse(ta_parent, &msg_node, &src, 3);
     if (rc < 0) {
-        MP_ERR(arg, "malformed JSON received\n");
+        mp_err(log, "malformed JSON received\n");
         rc = MPV_ERROR_INVALID_PARAMETER;
         goto error;
     }
@@ -278,11 +237,11 @@ static char *json_execute_command(struct client_arg *arg, void *ta_parent,
     cmd = cmd_str_node->u.string;
 
     if (!strcmp("client_name", cmd)) {
-        const char *client_name = mpv_client_name(arg->client);
+        const char *client_name = mpv_client_name(client);
         mpv_node_map_add_string(ta_parent, &reply_node, "data", client_name);
         rc = MPV_ERROR_SUCCESS;
     } else if (!strcmp("get_time_us", cmd)) {
-        int64_t time_us = mpv_get_time_us(arg->client);
+        int64_t time_us = mpv_get_time_us(client);
         mpv_node_map_add_int64(ta_parent, &reply_node, "data", time_us);
         rc = MPV_ERROR_SUCCESS;
     } else if (!strcmp("get_version", cmd)) {
@@ -302,7 +261,7 @@ static char *json_execute_command(struct client_arg *arg, void *ta_parent,
             goto error;
         }
 
-        rc = mpv_get_property(arg->client, cmd_node->u.list->values[1].u.string,
+        rc = mpv_get_property(client, cmd_node->u.list->values[1].u.string,
                               MPV_FORMAT_NODE, &result_node);
         if (rc >= 0) {
             mpv_node_map_add(ta_parent, &reply_node, "data", &result_node);
@@ -319,7 +278,7 @@ static char *json_execute_command(struct client_arg *arg, void *ta_parent,
             goto error;
         }
 
-        char *result = mpv_get_property_string(arg->client,
+        char *result = mpv_get_property_string(client,
                                         cmd_node->u.list->values[1].u.string);
         if (!result) {
             mpv_node_map_add_null(ta_parent, &reply_node, "data");
@@ -338,7 +297,7 @@ static char *json_execute_command(struct client_arg *arg, void *ta_parent,
             goto error;
         }
 
-        rc = mpv_set_property(arg->client, cmd_node->u.list->values[1].u.string,
+        rc = mpv_set_property(client, cmd_node->u.list->values[1].u.string,
                               MPV_FORMAT_NODE, &cmd_node->u.list->values[2]);
     } else if (!strcmp("set_property_string", cmd)) {
         if (cmd_node->u.list->num != 3) {
@@ -356,7 +315,7 @@ static char *json_execute_command(struct client_arg *arg, void *ta_parent,
             goto error;
         }
 
-        rc = mpv_set_property_string(arg->client,
+        rc = mpv_set_property_string(client,
                                      cmd_node->u.list->values[1].u.string,
                                      cmd_node->u.list->values[2].u.string);
     } else if (!strcmp("observe_property", cmd)) {
@@ -375,7 +334,7 @@ static char *json_execute_command(struct client_arg *arg, void *ta_parent,
             goto error;
         }
 
-        rc = mpv_observe_property(arg->client,
+        rc = mpv_observe_property(client,
                                   cmd_node->u.list->values[1].u.int64,
                                   cmd_node->u.list->values[2].u.string,
                                   MPV_FORMAT_NODE);
@@ -395,7 +354,7 @@ static char *json_execute_command(struct client_arg *arg, void *ta_parent,
             goto error;
         }
 
-        rc = mpv_observe_property(arg->client,
+        rc = mpv_observe_property(client,
                                   cmd_node->u.list->values[1].u.int64,
                                   cmd_node->u.list->values[2].u.string,
                                   MPV_FORMAT_STRING);
@@ -410,8 +369,8 @@ static char *json_execute_command(struct client_arg *arg, void *ta_parent,
             goto error;
         }
 
-        rc = mpv_unobserve_property(arg->client,
-                                  cmd_node->u.list->values[1].u.int64);
+        rc = mpv_unobserve_property(client,
+                                    cmd_node->u.list->values[1].u.int64);
     } else if (!strcmp("request_log_messages", cmd)) {
         if (cmd_node->u.list->num != 2) {
             rc = MPV_ERROR_INVALID_PARAMETER;
@@ -423,13 +382,13 @@ static char *json_execute_command(struct client_arg *arg, void *ta_parent,
             goto error;
         }
 
-        rc = mpv_request_log_messages(arg->client,
+        rc = mpv_request_log_messages(client,
                                       cmd_node->u.list->values[1].u.string);
     } else if (!strcmp("suspend", cmd)) {
-        mpv_suspend(arg->client);
+        mpv_suspend(client);
         rc = MPV_ERROR_SUCCESS;
     } else if (!strcmp("resume", cmd)) {
-        mpv_resume(arg->client);
+        mpv_resume(client);
         rc = MPV_ERROR_SUCCESS;
     } else if (!strcmp("enable_event", cmd) ||
                !strcmp("disable_event", cmd))
@@ -449,7 +408,7 @@ static char *json_execute_command(struct client_arg *arg, void *ta_parent,
         char *name = cmd_node->u.list->values[1].u.string;
         if (strcmp(name, "all") == 0) {
             for (int n = 0; n < 64; n++)
-                mpv_request_event(arg->client, n, enable);
+                mpv_request_event(client, n, enable);
             rc = MPV_ERROR_SUCCESS;
         } else {
             int event = -1;
@@ -462,12 +421,12 @@ static char *json_execute_command(struct client_arg *arg, void *ta_parent,
                 rc = MPV_ERROR_INVALID_PARAMETER;
                 goto error;
             }
-            rc = mpv_request_event(arg->client, event, enable);
+            rc = mpv_request_event(client, event, enable);
         }
     } else {
         mpv_node result_node;
 
-        rc = mpv_command_node(arg->client, cmd_node, &result_node);
+        rc = mpv_command_node(client, cmd_node, &result_node);
         if (rc >= 0)
             mpv_node_map_add(ta_parent, &reply_node, "data", &result_node);
     }
@@ -490,382 +449,35 @@ error:
     return output;
 }
 
-static char *text_execute_command(struct client_arg *arg, void *tmp, char *src)
+static char *text_execute_command(struct mpv_handle *client, void *tmp, char *src)
 {
-    mpv_command_string(arg->client, src);
+    mpv_command_string(client, src);
 
     return NULL;
 }
 
-static int ipc_write_str(struct client_arg *client, const char *buf)
+char *mp_ipc_consume_next_command(struct mpv_handle *client, void *ctx, bstr *buf)
 {
-    size_t count = strlen(buf);
-    while (count > 0) {
-        ssize_t rc = send(client->client_fd, buf, count, MSG_NOSIGNAL);
-        if (rc <= 0) {
-            if (rc == 0)
-                return -1;
-
-            if (errno == EBADF) {
-                client->writable = false;
-                return 0;
-            }
+    void *tmp = talloc_new(NULL);
 
-            if (errno == EINTR)
-                continue;
+    bstr rest;
+    bstr line = bstr_getline(*buf, &rest);
+    char *line0 = bstrto0(tmp, line);
+    talloc_steal(tmp, buf->start);
+    *buf = bstrdup(NULL, rest);
 
-            if (errno == EAGAIN)
-                return 0;
-
-            return rc;
-        }
+    json_skip_whitespace(&line0);
 
-        count -= rc;
-        buf   += rc;
-    }
-
-    return 0;
-}
-
-static void *client_thread(void *p)
-{
-    pthread_detach(pthread_self());
-
-    int rc;
-
-    struct client_arg *arg = p;
-    bstr client_msg = { talloc_strdup(NULL, ""), 0 };
-
-    mpthread_set_name(arg->client_name);
-
-    int pipe_fd = mpv_get_wakeup_pipe(arg->client);
-    if (pipe_fd < 0) {
-        MP_ERR(arg, "Could not get wakeup pipe\n");
-        goto done;
-    }
-
-    MP_VERBOSE(arg, "Client connected\n");
-
-    struct pollfd fds[2] = {
-        {.events = POLLIN, .fd = pipe_fd},
-        {.events = POLLIN, .fd = arg->client_fd},
-    };
-
-    fcntl(arg->client_fd, F_SETFL, fcntl(arg->client_fd, F_GETFL, 0) | O_NONBLOCK);
-    mpv_suspend(arg->client);
-
-    while (1) {
-        rc = poll(fds, 2, 0);
-        if (rc == 0) {
-            mpv_resume(arg->client);
-            rc = poll(fds, 2, -1);
-            mpv_suspend(arg->client);
-        }
-        if (rc < 0) {
-            MP_ERR(arg, "Poll error\n");
-            continue;
-        }
-
-        if (fds[0].revents & POLLIN) {
-            char discard[100];
-            read(pipe_fd, discard, sizeof(discard));
-
-            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 = json_encode_event(event);
-                if (!event_msg) {
-                    MP_ERR(arg, "Encoding error\n");
-                    goto done;
-                }
-
-                rc = ipc_write_str(arg, event_msg);
-                talloc_free(event_msg);
-                if (rc < 0) {
-                    MP_ERR(arg, "Write error (%s)\n", mp_strerror(errno));
-                    goto done;
-                }
-            }
-        }
-
-        if (fds[1].revents & (POLLIN | POLLHUP)) {
-            while (1) {
-                char buf[128];
-                bstr append = { buf, 0 };
-
-                ssize_t bytes = read(arg->client_fd, buf, sizeof(buf));
-                if (bytes < 0) {
-                    if (errno == EAGAIN)
-                        break;
-
-                    MP_ERR(arg, "Read error (%s)\n", mp_strerror(errno));
-                    goto done;
-                }
-
-                if (bytes == 0) {
-                    MP_VERBOSE(arg, "Client disconnected\n");
-                    goto done;
-                }
-
-                append.len = bytes;
-
-                bstr_xappend(NULL, &client_msg, append);
-
-                while (bstrchr(client_msg, '\n') != -1) {
-                    void *tmp = talloc_new(NULL);
-                    bstr rest;
-                    bstr line = bstr_getline(client_msg, &rest);
-                    char *line0 = bstrto0(tmp, line);
-                    talloc_steal(tmp, client_msg.start);
-                    client_msg = bstrdup(NULL, rest);
-
-                    json_skip_whitespace(&line0);
-
-                    char *reply_msg = NULL;
-                    if (line0[0] == '\0' || line0[0] == '#') {
-                        // skip
-                    } else if (line0[0] == '{') {
-                        reply_msg = json_execute_command(arg, tmp, line0);
-                    } else {
-                        reply_msg = text_execute_command(arg, tmp, line0);
-                    }
-
-                    if (reply_msg && arg->writable) {
-                        rc = ipc_write_str(arg, reply_msg);
-                        if (rc < 0) {
-                            MP_ERR(arg, "Write error (%s)\n", mp_strerror(errno));
-                            talloc_free(tmp);
-                            goto done;
-                        }
-                    }
-
-                    talloc_free(tmp);
-                }
-            }
-        }
-    }
-
-done:
-    if (client_msg.len > 0)
-        MP_WARN(arg, "Ignoring unterminated command on disconnect.\n");
-    talloc_free(client_msg.start);
-    if (arg->close_client_fd)
-        close(arg->client_fd);
-    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);
-        if (client->close_client_fd)
-            close(client->client_fd);
-        talloc_free(client);
-    }
-}
-
-static void ipc_start_client_json(struct mp_ipc_ctx *ctx, int id, int fd)
-{
-    struct client_arg *client = talloc_ptrtype(NULL, client);
-    *client = (struct client_arg){
-        .client_name = talloc_asprintf(client, "ipc-%d", id),
-        .client_fd   = fd,
-        .close_client_fd = true,
-
-        .writable = true,
-    };
-
-    ipc_start_client(ctx, client);
-}
-
-static void ipc_start_client_text(struct mp_ipc_ctx *ctx, const char *path)
-{
-    int mode = O_RDONLY;
-    int client_fd = -1;
-    bool close_client_fd = true;
-    bool writable = false;
-
-    if (strcmp(path, "/dev/stdin") == 0) { // for symmetry with Linux
-        client_fd = STDIN_FILENO;
-        close_client_fd = false;
-    } else if (strncmp(path, "fd://", 5) == 0) {
-        char *end = NULL;
-        client_fd = strtol(path + 5, &end, 0);
-        if (!end || end == path + 5 || end[0]) {
-            MP_ERR(ctx, "Invalid FD: %s\n", path);
-            return;
-        }
-        close_client_fd = false;
-        writable = true; // maybe
+    char *reply_msg = NULL;
+    if (line0[0] == '\0' || line0[0] == '#') {
+        // skip
+    } else if (line0[0] == '{') {
+        reply_msg = json_execute_command(client, tmp, line0);
     } else {
-        // Use RDWR for FIFOs to ensure they stay open over multiple accesses.
-        struct stat st;
-        if (stat(path, &st) == 0 && S_ISFIFO(st.st_mode))
-            mode = O_RDWR;
-        client_fd = open(path, mode);
+        reply_msg = text_execute_command(client, tmp, line0);
     }
-    if (client_fd < 0) {
-        MP_ERR(ctx, "Could not open '%s'\n", path);
-        return;
-    }
-
-    struct client_arg *client = talloc_ptrtype(NULL, client);
-    *client = (struct client_arg){
-        .client_name = "input-file",
-        .client_fd   = client_fd,
-        .close_client_fd = close_client_fd,
-        .writable = writable,
-    };
-
-    ipc_start_client(ctx, client);
-}
-
-static void *ipc_thread(void *p)
-{
-    int rc;
-
-    int ipc_fd;
-    struct sockaddr_un ipc_un = {0};
-
-    struct mp_ipc_ctx *arg = p;
-
-    mpthread_set_name("ipc socket listener");
-
-    MP_VERBOSE(arg, "Starting IPC master\n");
-
-    ipc_fd = socket(AF_UNIX, SOCK_STREAM, 0);
-    if (ipc_fd < 0) {
-        MP_ERR(arg, "Could not create IPC socket\n");
-        goto done;
-    }
-
-#if HAVE_FCHMOD
-    fchmod(ipc_fd, 0600);
-#endif
-
-    size_t path_len = strlen(arg->path);
-    if (path_len >= sizeof(ipc_un.sun_path) - 1) {
-        MP_ERR(arg, "Could not create IPC socket\n");
-        goto done;
-    }
-
-    ipc_un.sun_family = AF_UNIX,
-    strncpy(ipc_un.sun_path, arg->path, sizeof(ipc_un.sun_path));
-
-    unlink(ipc_un.sun_path);
-
-    if (ipc_un.sun_path[0] == '@') {
-        ipc_un.sun_path[0] = '\0';
-        path_len--;
-    }
-
-    size_t addr_len = offsetof(struct sockaddr_un, sun_path) + 1 + path_len;
-    rc = bind(ipc_fd, (struct sockaddr *) &ipc_un, addr_len);
-    if (rc < 0) {
-        MP_ERR(arg, "Could not bind IPC socket\n");
-        goto done;
-    }
-
-    rc = listen(ipc_fd, 10);
-    if (rc < 0) {
-        MP_ERR(arg, "Could not listen on IPC socket\n");
-        goto done;
-    }
-
-    int client_num = 0;
-
-    struct pollfd fds[2] = {
-        {.events = POLLIN, .fd = arg->death_pipe[0]},
-        {.events = POLLIN, .fd = ipc_fd},
-    };
-
-    while (1) {
-        rc = poll(fds, 2, -1);
-        if (rc < 0) {
-            MP_ERR(arg, "Poll error\n");
-            continue;
-        }
-
-        if (fds[0].revents & POLLIN)
-            goto done;
-
-        if (fds[1].revents & POLLIN) {
-            int client_fd = accept(ipc_fd, NULL, NULL);
-            if (client_fd < 0) {
-                MP_ERR(arg, "Could not accept IPC client\n");
-                goto done;
-            }
-
-            ipc_start_client_json(arg, client_num++, client_fd);
-        }
-    }
-
-done:
-    if (ipc_fd >= 0)
-        close(ipc_fd);
-
-    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,
-        .path       = mp_get_user_path(arg, global, opts->ipc_path),
-        .death_pipe = {-1, -1},
-    };
-    char *input_file = mp_get_user_path(arg, global, opts->input_file);
-
-    if (input_file && *input_file)
-        ipc_start_client_text(arg, input_file);
-
-    if (!opts->ipc_path || !*opts->ipc_path)
-        goto out;
-
-    if (mp_make_wakeup_pipe(arg->death_pipe) < 0)
-        goto out;
-
-    if (pthread_create(&arg->thread, NULL, ipc_thread, arg))
-        goto out;
-
-    return arg;
-
-out:
-    close(arg->death_pipe[0]);
-    close(arg->death_pipe[1]);
-    talloc_free(arg);
-    return NULL;
-}
-
-void mp_uninit_ipc(struct mp_ipc_ctx *arg)
-{
-    if (!arg)
-        return;
-
-    write(arg->death_pipe[1], &(char){0}, 1);
-    pthread_join(arg->thread, NULL);
 
-    close(arg->death_pipe[0]);
-    close(arg->death_pipe[1]);
-    talloc_free(arg);
+    talloc_steal(ctx, reply_msg);
+    talloc_free(tmp);
+    return reply_msg;
 }