libuv: initial support

2025-12-22 20:04:19 +01:00 · 2019-11-27 22:34:23 +08:00
parent 706479909a
commit 59601329ae
5 changed files with 138 additions and 103 deletions
--- a/.github/workflows/backend.yml
+++ b/.github/workflows/backend.yml
@@ -20,7 +20,7 @@ jobs:
      - name: install apt packages
        run: |
          sudo apt-get update
-          sudo apt-get install build-essential cmake libjson-c-dev libssl-dev
+          sudo apt-get install build-essential cmake libjson-c-dev libssl-dev libuv1-dev
      - name: compile libwebsockets-${{ matrix.lws-version }} from source
        env:
          LWS_VERSION: ${{ matrix.lws-version }}
@@ -28,7 +28,7 @@ jobs:
          cd $(mktemp -d)
          curl -sLo- https://github.com/warmcat/libwebsockets/archive/v${LWS_VERSION}.tar.gz | tar xz
          cd libwebsockets-${LWS_VERSION}
-          cmake -DLWS_UNIX_SOCK=ON -DLWS_IPV6=ON -DLWS_WITHOUT_TESTAPPS=ON -DCMAKE_BUILD_TYPE=RELEASE .
+          cmake -DLWS_WITH_LIBUV=ON -DLWS_UNIX_SOCK=ON -DLWS_IPV6=ON -DLWS_WITHOUT_TESTAPPS=ON -DCMAKE_BUILD_TYPE=RELEASE .
          make && sudo make install
      - uses: actions/checkout@v1
      - name: build ttyd
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -59,6 +59,14 @@ include(FindPackageHandleStandardArgs)
 find_package_handle_standard_args(JSON-C DEFAULT_MSG JSON-C_LIBRARY JSON-C_INCLUDE_DIR)
 mark_as_advanced(JSON-C_INCLUDE_DIR JSON-C_LIBRARY)

+pkg_check_modules(LIBUV REQUIRED libuv>=1.0.0)
+find_path(LIBUV_INCLUDE_DIR NAMES uv.h
+        HINTS ${LIBUV_INCLUDEDIRS} ${LIBUV_INCLUDEDIR})
+find_library(LIBUV_LIBRARY NAMES uv libuv
+        HINTS ${LIBUV_LIBRARY_DIRS} ${LIBUV_LIBDIR})
+find_package_handle_standard_args(LIBUV DEFAULT_MSG LIBUV_LIBRARY LIBUV_INCLUDE_DIR)
+mark_as_advanced(LIBUV_INCLUDE_DIR LIBUV_LIBRARY)
+
 find_program(CMAKE_XXD NAMES xxd)
 add_custom_command(OUTPUT html.h
        COMMAND ${CMAKE_XXD} -i index.html html.h
@@ -66,8 +74,8 @@ add_custom_command(OUTPUT html.h
        COMMENT "Generating html.h from index.html")
 list(APPEND SOURCE_FILES html.h)

-set(INCLUDE_DIRS ${OPENSSL_INCLUDE_DIR} ${LIBWEBSOCKETS_INCLUDE_DIR} ${JSON-C_INCLUDE_DIR})
-set(LINK_LIBS ${OPENSSL_LIBRARIES} ${LIBWEBSOCKETS_LIBRARIES} ${JSON-C_LIBRARY})
+set(INCLUDE_DIRS ${OPENSSL_INCLUDE_DIR} ${LIBWEBSOCKETS_INCLUDE_DIR} ${JSON-C_INCLUDE_DIR} ${LIBUV_INCLUDE_DIR})
+set(LINK_LIBS ${OPENSSL_LIBRARIES} ${LIBWEBSOCKETS_LIBRARIES} ${JSON-C_LIBRARY} ${LIBUV_LIBRARY})

 if(APPLE)
    # required for the new homebrew version of libwebsockets
--- a/src/protocol.c
+++ b/src/protocol.c
@@ -140,22 +140,51 @@ tty_client_destroy(struct tty_client *client) {
    close(client->pty);

 cleanup:
+    uv_read_stop((uv_stream_t *) &client->pipe);
+
    // free the buffer
    if (client->buffer != NULL)
        free(client->buffer);
+    if (client->pty_buffer != NULL)
+        free(client->pty_buffer);

    for (int i = 0; i < client->argc; i++) {
        free(client->args[i]);
    }

-#if LWS_LIBRARY_VERSION_NUMBER >= 3002000
-    lws_sul_schedule(context, 0, &client->sul_stagger, NULL, LWS_SET_TIMER_USEC_CANCEL);
-#endif
-
    // remove from client list
    tty_client_remove(client);
 }

+void
+alloc_cb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
+    buf->base = xmalloc(suggested_size);
+    buf->len = suggested_size;
+}
+
+void
+read_cb(uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf) {
+    struct tty_client *client = (struct tty_client *) stream->data;
+    client->pty_len = nread;
+
+    if (nread <= 0) {
+        if (nread == UV_ENOBUFS || nread == 0)
+            return;
+        if (nread == UV_EOF)
+            client->pty_len = 0;
+        else
+            lwsl_err("read_cb: %s\n", uv_err_name(nread));
+        client->pty_buffer = NULL;
+    } else {
+        client->pty_buffer = xmalloc(LWS_PRE + 1 + (size_t ) nread);
+        memcpy(client->pty_buffer + LWS_PRE + 1, buf->base, (size_t ) nread);
+    }
+    free(buf->base);
+
+    lws_callback_on_writable(client->wsi);
+    uv_read_stop(stream);
+}
+
 int
 spawn_process(struct tty_client *client) {
    // append url args to arguments
@@ -199,33 +228,14 @@ spawn_process(struct tty_client *client) {
    if (client->size.ws_row > 0 && client->size.ws_col > 0)
        ioctl(client->pty, TIOCSWINSZ, &client->size);

+    client->pipe.data = client;
+    uv_pipe_open(&client->pipe, pty);
+
+    lws_callback_on_writable(client->wsi);
+
    return 0;
 }

-void
-tty_client_poll(struct tty_client *client) {
-    if (client->pid <= 0 || client->state == STATE_READY) return;
-
-    if (!client->running) {
-        memset(client->pty_buffer, 0, sizeof(client->pty_buffer));
-        client->pty_len = client->exit_status;
-        client->state = STATE_READY;
-        return;
-    }
-
-    fd_set des_set;
-    FD_ZERO (&des_set);
-    FD_SET (client->pty, &des_set);
-    struct timeval tv = { 0, 5000 }; // 5ms
-    if (select(client->pty + 1, &des_set, NULL, NULL, &tv) <= 0) return;
-
-    if (FD_ISSET (client->pty, &des_set)) {
-        memset(client->pty_buffer, 0, sizeof(client->pty_buffer));
-        client->pty_len = read(client->pty, client->pty_buffer + LWS_PRE + 1, BUF_SIZE);
-        client->state = STATE_READY;
-    }
-}
-
 int
 callback_tty(struct lws *wsi, enum lws_callback_reasons reason,
             void *user, void *in, size_t len) {
@@ -261,10 +271,11 @@ callback_tty(struct lws *wsi, enum lws_callback_reasons reason,
            client->authenticated = false;
            client->wsi = wsi;
            client->buffer = NULL;
-            client->state = STATE_INIT;
            client->pty_len = 0;
            client->argc = 0;

+            uv_pipe_init(server->loop, &client->pipe, 0);
+
            if (server->url_arg) {
                while (lws_hdr_copy_fragment(wsi, buf, sizeof(buf), WSI_TOKEN_HTTP_URI_ARGS, n++) > 0) {
                    if (strncmp(buf, "arg=", 4) == 0) {
@@ -293,6 +304,7 @@ callback_tty(struct lws *wsi, enum lws_callback_reasons reason,
            if (!client->initialized) {
                if (client->initial_cmd_index == sizeof(initial_cmds)) {
                    client->initialized = true;
+                    uv_read_start((uv_stream_t *)& client->pipe, alloc_cb, read_cb);
                    break;
                }
                if (send_initial_message(wsi, client->initial_cmd_index) < 0) {
@@ -304,9 +316,6 @@ callback_tty(struct lws *wsi, enum lws_callback_reasons reason,
                lws_callback_on_writable(wsi);
                break;
            }
-            if (client->state != STATE_READY) {
-                break;
-            }

            // read error or client exited, close connection
            if (client->pty_len == 0) {
@@ -318,12 +327,17 @@ callback_tty(struct lws *wsi, enum lws_callback_reasons reason,
                return -1;
            }

+            if (client->pty_buffer == NULL)
+                break;
+
            client->pty_buffer[LWS_PRE] = OUTPUT;
            n = (size_t) (client->pty_len + 1);
            if (lws_write(wsi, (unsigned char *) client->pty_buffer + LWS_PRE, n, LWS_WRITE_BINARY) < n) {
-                lwsl_err("write data to WS\n");
+                lwsl_err("write OUTPUT to WS\n");
            }
-            client->state = STATE_DONE;
+            free(client->pty_buffer);
+            client->pty_buffer = NULL;
+            uv_read_start((uv_stream_t *)& client->pipe, alloc_cb, read_cb);
            break;

        case LWS_CALLBACK_RECEIVE:
@@ -356,9 +370,10 @@ callback_tty(struct lws *wsi, enum lws_callback_reasons reason,
                        break;
                    if (server->readonly)
                        return 0;
-                    if (write(client->pty, client->buffer + 1, client->len - 1) == -1) {
-                        lwsl_err("write INPUT to pty: %d (%s)\n", errno, strerror(errno));
-                        lws_close_reason(wsi, LWS_CLOSE_STATUS_UNEXPECTED_CONDITION, NULL, 0);
+                    uv_buf_t b = { client->buffer + 1, client->len - 1 };
+                    int err = uv_try_write((uv_stream_t *) &client->pipe, &b, 1);
+                    if (err < 0) {
+                        lwsl_err("uv_try_write: %s\n", uv_err_name(err));
                        return -1;
                    }
                    break;
@@ -401,8 +416,8 @@ callback_tty(struct lws *wsi, enum lws_callback_reasons reason,
            break;

        case LWS_CALLBACK_CLOSED:
-            tty_client_destroy(client);
            lwsl_notice("WS closed from %s, clients: %d\n", client->address, server->client_count);
+            tty_client_destroy(client);
            if (server->once && server->client_count == 0) {
                lwsl_notice("exiting due to the --once option.\n");
                force_exit = true;
--- a/src/server.c
+++ b/src/server.c
@@ -140,6 +140,9 @@ tty_server_new(int argc, char **argv, int start) {
    }
    *ptr = '\0'; // null terminator

+    ts->loop = xmalloc(sizeof *ts->loop);
+    uv_loop_init(ts->loop);
+
    return ts;
 }

@@ -164,39 +167,54 @@ tty_server_free(struct tty_server *ts) {
            unlink(ts->socket_path);
        }
    }
+    uv_loop_close(ts->loop);
+    free(ts->loop);
    free(ts);
 }

 void
-sig_handler(int sig) {
+signal_cb(uv_signal_t *watcher, int signum) {
+    char sig_name[20];
+    pid_t pid;
+    int status;
+
+    switch (watcher->signum) {
+        case SIGINT:
+        case SIGTERM:
+            get_sig_name(watcher->signum, sig_name, sizeof(sig_name));
+            lwsl_notice("received signal: %s (%d), exiting...\n", sig_name, watcher->signum);
+            break;
+        case SIGCHLD:
+            status = wait_proc(-1, &pid);
+            if (pid > 0) {
+                lwsl_notice("process exited with code %d, pid: %d\n", status, pid);
+                struct tty_client *iterator;
+                LIST_FOREACH(iterator, &server->clients, list) {
+                    if (iterator->pid == pid) {
+                        iterator->running = false;
+                        iterator->exit_status = status;
+                        break;
+                    }
+                }
+            }
+            return;
+        default:
+            signal(SIGABRT, SIG_DFL);
+            abort();
+    }
+
    if (force_exit)
        exit(EXIT_FAILURE);
-
-    char sig_name[20];
-    get_sig_name(sig, sig_name, sizeof(sig_name));
-    lwsl_notice("received signal: %s (%d), exiting...\n", sig_name, sig);
    force_exit = true;
    lws_cancel_service(context);
+#if LWS_LIBRARY_VERSION_MAJOR < 3
+    lws_libuv_stop(context);
+#else
+    uv_stop(server->loop);
+#endif
    lwsl_notice("send ^C to force exit.\n");
 }

-void
-sigchld_handler() {
-    pid_t pid;
-    int status = wait_proc(-1, &pid);
-    if (pid > 0) {
-        lwsl_notice("process exited with code %d, pid: %d\n", status, pid);
-        struct tty_client *iterator;
-        LIST_FOREACH(iterator, &server->clients, list) {
-            if (iterator->pid == pid) {
-                iterator->running = false;
-                iterator->exit_status = status;
-                break;
-            }
-        }
-    }
-}
-
 int
 calc_command_start(int argc, char **argv) {
    // make a copy of argc and argv
@@ -235,15 +253,6 @@ calc_command_start(int argc, char **argv) {
    return start;
 }

-#if LWS_LIBRARY_VERSION_NUMBER >= 3002000
-void
-stagger_callback(lws_sorted_usec_list_t *sul) {
-    struct tty_client *client = lws_container_of(sul, struct tty_client, sul_stagger);
-
-    lws_callback_on_writable(client->wsi);
-}
-#endif
-
 int
 main(int argc, char **argv) {
    if (argc == 1) {
@@ -264,7 +273,7 @@ main(int argc, char **argv) {
    info.gid = -1;
    info.uid = -1;
    info.max_http_header_pool = 16;
-    info.options = LWS_SERVER_OPTION_VALIDATE_UTF8 | LWS_SERVER_OPTION_DISABLE_IPV6;
+    info.options = LWS_SERVER_OPTION_LIBUV | LWS_SERVER_OPTION_VALIDATE_UTF8 | LWS_SERVER_OPTION_DISABLE_IPV6;
    info.extensions = extensions;
    info.max_http_header_data = 20480;

@@ -474,9 +483,11 @@ main(int argc, char **argv) {
        lwsl_notice("  custom index.html: %s\n", server->index);
    }

-    signal(SIGINT, sig_handler);  // ^C
-    signal(SIGTERM, sig_handler); // kill
-    signal(SIGCHLD, sigchld_handler);
+#if LWS_LIBRARY_VERSION_MAJOR >= 3
+    void *foreign_loops[1];
+    foreign_loops[0] = server->loop;
+    info.foreign_loops = foreign_loops;
+#endif

    context = lws_create_context(&info);
    if (context == NULL) {
@@ -490,21 +501,30 @@ main(int argc, char **argv) {
        open_uri(url);
    }

-    while (!force_exit) {
-        if (!LIST_EMPTY(&server->clients)) {
-            struct tty_client *client;
-            LIST_FOREACH(client, &server->clients, list) {
-                tty_client_poll(client);
-#if LWS_LIBRARY_VERSION_NUMBER >= 3002000
-                lws_sul_schedule(context, 0, &client->sul_stagger, stagger_callback, 0);
-#else
-                lws_callback_on_writable(client->wsi);
-#endif
-            }
-        }
-        lws_service(context, 10);
+#if LWS_LIBRARY_VERSION_MAJOR >= 3
+    int sig_nums[] = {SIGINT, SIGTERM, SIGCHLD};
+    int ns = sizeof(sig_nums)/sizeof(sig_nums[0]);
+    uv_signal_t signals[ns];
+    for (int i = 0; i < ns; i++) {
+        uv_signal_init(server->loop, &signals[i]);
+        uv_signal_start(&signals[i], signal_cb, sig_nums[i]);
    }

+    lws_service(context, 0);
+
+    for (int i = 0; i < ns; i++) {
+        uv_signal_stop(&signals[i]);
+    }
+#else
+#if LWS_LIBRARY_VERSION_MAJOR < 2
+    lws_uv_initloop(context, server->loop, signal_cb, 0);
+#else
+    lws_uv_sigint_cfg(context, 1, signal_cb);
+    lws_uv_initloop(context, server->loop, 0);
+#endif
+    lws_libuv_run(context, 0);
+#endif
+
    lws_context_destroy(context);

    // cleanup
--- a/src/server.h
+++ b/src/server.h
@@ -1,6 +1,7 @@
 #include <stdbool.h>
 #include <sys/ioctl.h>
 #include <sys/queue.h>
+#include <uv.h>

 // client message
 #define INPUT '0'
@@ -15,16 +16,10 @@
 // websocket url path
 #define WS_PATH "/ws"

-#define BUF_SIZE 32768 // 32K
-
 extern volatile bool force_exit;
 extern struct lws_context *context;
 extern struct tty_server *server;

-enum pty_state {
-    STATE_INIT, STATE_READY, STATE_DONE
-};
-
 struct tty_client {
    bool running;
    bool initialized;
@@ -35,9 +30,6 @@ struct tty_client {
    int argc;

    struct lws *wsi;
-#if LWS_LIBRARY_VERSION_NUMBER >= 3002000
-    lws_sorted_usec_list_t sul_stagger;
-#endif
    struct winsize size;
    char *buffer;
    size_t len;
@@ -45,10 +37,11 @@ struct tty_client {
    int pid;
    int pty;
    int exit_status;
-    enum pty_state state;
-    char pty_buffer[LWS_PRE + 1 + BUF_SIZE];
+    char *pty_buffer;
    ssize_t pty_len;

+    uv_pipe_t pipe;
+
    LIST_ENTRY(tty_client) list;
 };

@@ -77,6 +70,7 @@ struct tty_server {
    bool once;                                // whether accept only one client and exit on disconnection
    char socket_path[255];                    // UNIX domain socket path
    char terminal_type[30];                   // terminal type to report
+    uv_loop_t *loop;                          // the libuv event loop
 };

 extern int
@@ -85,5 +79,3 @@ callback_http(struct lws *wsi, enum lws_callback_reasons reason, void *user, voi
 extern int
 callback_tty(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len);

-extern void
-tty_client_poll(struct tty_client *client);