Use a socket, not pipe, for STDIN emulation in fselect

Using a TCP connected socket allows us to use a single select call instead of
interrupting it every 125ms to poll the stdin-pipe.

nbase/nbase_misc.c

@@ -352,139 +352,83 @@ const char *tval_unit(const char *tspec) {
  * they were NULL or empty.  This only works for sockets and stdin; if
  * you have a descriptor referring to a normal open file in the set,
  * Windows will return WSAENOTSOCK. */
-int fselect(int s, fd_set *rmaster, fd_set *wmaster, fd_set *emaster, struct timeval *tv)
+int fselect(int s, fd_set *rset, fd_set *wset, fd_set *eset, struct timeval *tv)
 {
 #ifdef WIN32
-    static int stdin_thread_started = 0;
+    static int stdin_socket = INVALID_SOCKET;
     int fds_ready = 0;
     int iter = -1;
-    int do_select = 0;
-    struct timeval stv;
-    fd_set rset, wset, eset;
     int r_stdin = 0;
     int e_stdin = 0;
-    int stdin_ready = 0;
 
     /* Figure out whether there are any FDs in the sets, as @$@!$# Windows
        returns WSAINVAL (10022) if you call a select() with no FDs, even though
        the Linux man page says that doing so is a good, reasonably portable way
        to sleep with subsecond precision.  Sigh. */
-    if (rmaster != NULL) {
+    if (s == 0 || (
-      /* If stdin is requested, clear it and remember it. */
+      (rset == NULL || rset->fd_count == 0) &&
-      if (checked_fd_isset(STDIN_FILENO, rmaster)) {
+      (wset == NULL || wset->fd_count == 0) &&
-        r_stdin = 1;
+      (eset == NULL || eset->fd_count == 0)
-        checked_fd_clr(STDIN_FILENO, rmaster);
+      )) {
-      }
+      usleep(tv->tv_sec * 1000000 + tv->tv_usec);
-      /* If any are left, we'll do a select. Otherwise, it's a sleep. */
+      return 0;
-      do_select = do_select || rmaster->fd_count;
+    }
+
+    /* If stdin is requested, clear it and remember it. */
+    if (rset && checked_fd_isset(STDIN_FILENO, rset)) {
+      r_stdin = 1;
+      checked_fd_clr(STDIN_FILENO, rset);
     }
 
     /* Same thing with exceptions */
-    if (emaster != NULL) {
+    if (eset && checked_fd_isset(STDIN_FILENO, eset)) {
-      if (checked_fd_isset(STDIN_FILENO, emaster)) {
+      e_stdin = 1;
-        e_stdin = 1;
+      checked_fd_clr(STDIN_FILENO, eset);
-        checked_fd_clr(STDIN_FILENO, emaster);
-      }
-      do_select = do_select || emaster->fd_count;
     }
 
     /* stdin can't be written to, so ignore it. */
-    if (wmaster != NULL) {
+    if (wset) {
-      assert(!checked_fd_isset(STDIN_FILENO, wmaster));
+      assert(!checked_fd_isset(STDIN_FILENO, wset));
-      do_select = do_select || wmaster->fd_count;
     }
 
-    /* Handle the case where stdin is not in scope. */
+    if (r_stdin || e_stdin) {
-    if (!(r_stdin || e_stdin)) {
+
-        if (do_select) {
+      /* This is a hack for Windows, which doesn't allow select()ing on
-            /* Do a normal select. */
+       * non-sockets (like stdin). We launch a background thread that shuttles
-            return select(s, rmaster, wmaster, emaster, tv);
+       * STDIN across a connected socket so that we can watch that with
-        } else {
+       * select().
-            /* No file descriptors given. Just sleep. */
+       */
-            if (tv == NULL) {
+
-                /* Sleep forever. */
+      /* nbase_winunix.c has all the nasty details behind checking if
-                while (1)
+       * stdin has input. It involves a background thread, which we start
-                    sleep(10000);
+       * now if necessary. */
-            } else {
+      if (stdin_socket == INVALID_SOCKET) {
-                usleep(tv->tv_sec * 1000000UL + tv->tv_usec);
+        stdin_socket = win_stdin_start_thread();
-                return 0;
+        assert(stdin_socket != INVALID_SOCKET);
-            }
+      }
-        }
+      if (r_stdin) {
+        checked_fd_set(stdin_socket, rset);
+      }
+      if (e_stdin) {
+        checked_fd_set(stdin_socket, eset);
+      }
     }
 
-    /* This is a hack for Windows, which doesn't allow select()ing on
+    fds_ready = select(s, rset, wset, eset, tv);
-     * non-sockets (like stdin).  We remove stdin from the fd_set and
+    if (fds_ready > 0) {
-     * loop while select()ing on everything else, with a timeout of
+      if (r_stdin && checked_fd_isset(stdin_socket, rset)) {
-     * 125ms.  Then we check if stdin is ready and increment fds_ready
+          checked_fd_clr(stdin_socket, rset);
-     * and set stdin in rmaster if it looks good.  We just keep looping
+        checked_fd_set(STDIN_FILENO, rset);
-     * until we have something or it times out.
+      }
-     */
+      if (e_stdin && checked_fd_isset(stdin_socket, eset)) {
-
+          checked_fd_clr(stdin_socket, eset);
-    /* nbase_winunix.c has all the nasty details behind checking if
+        checked_fd_set(STDIN_FILENO, eset);
-     * stdin has input. It involves a background thread, which we start
+      }
-     * now if necessary. */
-    if (!stdin_thread_started) {
-        int ret = win_stdin_start_thread();
-        assert(ret != 0);
-        stdin_thread_started = 1;
     }
 
-    if (tv) {
-        int usecs = (tv->tv_sec * 1000000) + tv->tv_usec;
-
-        iter = usecs / 125000;
-
-        if (usecs % 125000)
-            iter++;
-    }
-
-    FD_ZERO(&rset);
-    FD_ZERO(&wset);
-    FD_ZERO(&eset);
-
-    while (!fds_ready && iter) {
-        stv.tv_sec = 0;
-        stv.tv_usec = 125000;
-
-        if (rmaster)
-            rset = *rmaster;
-        if (wmaster)
-            wset = *wmaster;
-        if (emaster)
-            eset = *emaster;
-
-        if(r_stdin) {
-            stdin_ready = win_stdin_ready();
-            if(stdin_ready)
-                stv.tv_usec = 0; /* get status but don't wait since stdin is ready */
-        }
-
-        fds_ready = 0;
-        /* selecting on anything other than stdin? */
-        if (do_select)
-            fds_ready = select(s, &rset, &wset, &eset, &stv);
-        else
-            usleep(stv.tv_sec * 1000000UL + stv.tv_usec);
-
-        if (fds_ready > -1 && stdin_ready) {
-            checked_fd_set(STDIN_FILENO, &rset);
-            fds_ready++;
-        }
-
-        if (tv)
-            iter--;
-    }
-
-    if (rmaster)
-        *rmaster = rset;
-    if (wmaster)
-        *wmaster = wset;
-    if (emaster)
-        *emaster = eset;
-
     return fds_ready;
 #else
-    return select(s, rmaster, wmaster, emaster, tv);
+    return select(s, rset, wset, eset, tv);
 #endif
 }
 

nbase/nbase_winunix.c

@@ -75,16 +75,13 @@ program is actively reading from stdin (which would normally mean blocking).
 
 The strategy is to create a background thread that constantly reads from stdin.
 The thread blocks while reading, which lets characters be echoed. The thread
-writes each block of data into an anonymous pipe. We juggle file descriptors and
+writes each block of data to a TCP socket. We juggle file descriptors and
 Windows file handles to make the rest of the program think that the other end of
-the pipe is stdin. Only the thread keeps a reference to the real stdin. Windows
+the TCP connection is stdin. Only the thread keeps a reference to the real stdin.
-has a PeekNamedPipe function that we use to check for input in the pipe without
+Since "stdin" is now a socket, it can be used with select and poll.
-blocking.
 
-Call win_stdin_start_thread to start the thread and win_stdin_ready for the
+Call win_stdin_start_thread to start the thread and get the stdin socket.
-non-blocking input check. Any other operations on stdin (read, scanf, etc.)
+Any other operations on stdin (read, scanf, etc.) should be transparent. Any
-should be transparent, except I noticed that eof(0) returns 1 when there is
-nothing in the pipe, but will return 0 again if more is written to the pipe. Any
 data buffered but not delivered to the program before starting the background
 thread may be lost when the thread is started.
 */
@@ -92,34 +89,56 @@ thread may be lost when the thread is started.
 /* The background thread that reads and buffers the true stdin. */
 static HANDLE stdin_thread = NULL;
 
+struct win_thread_data {
 /* This is a copy of the true stdin file handle before any redirection. It is
    read by the thread. */
-static HANDLE thread_stdin_handle = NULL;
+  HANDLE stdin_handle;
-/* The thread writes to this pipe and standard input is reassigned to be the
+/* This is the listen socket for the thread. It is closed after the first
-   read end of it. */
+   connection. */
-static HANDLE stdin_pipe_r = NULL, stdin_pipe_w = NULL;
+  int socket_l;
+};
 
-/* This is the thread that reads from the true stdin (thread_stdin_handle) and
+/* This is the thread that reads from the true stdin (tdata->stdin_handle) and
-   writes to stdin_pipe_w, which is reassigned to be the stdin that the rest of
+   writes to socket_w, which is connected to the replacement stdin that the rest of
    the program sees. Once started, it never finishes except in case of error.
-   win_stdin_start_thread is responsible for setting up thread_stdin_handle. */
+   win_stdin_start_thread is responsible for setting up tdata->stdin_handle. */
 static DWORD WINAPI win_stdin_thread_func(void *data) {
+    struct win_thread_data *tdata = (struct win_thread_data *)data;
     DWORD n, nwritten;
     char buffer[BUFSIZ];
+    SOCKET socket_w = accept(tdata->socket_l, NULL, NULL);
+    if (socket_w == INVALID_SOCKET) {
+        //fprintf(stderr, "accept error: %d\n", socket_errno());
+        goto ThreadCleanup;
+    }
+
+    closesocket(tdata->socket_l);
+    tdata->socket_l = INVALID_SOCKET;
+    if (SOCKET_ERROR == shutdown(socket_w, SD_RECEIVE))
+        goto ThreadCleanup;
 
     for (;;) {
-        if (ReadFile(thread_stdin_handle, buffer, sizeof(buffer), &n, NULL) == 0)
+        if (ReadFile(tdata->stdin_handle, buffer, sizeof(buffer), &n, NULL) == 0)
             break;
         if (n == -1 || n == 0)
             break;
 
-        if (WriteFile(stdin_pipe_w, buffer, n, &nwritten, NULL) == 0)
+        // In the future, we can use WSASend to take advantage of the OVERLAPPED socket for IOCP
+        nwritten = send(socket_w, buffer, n, 0);
+        if (nwritten == SOCKET_ERROR)
             break;
         if (nwritten != n)
             break;
     }
-    CloseHandle(thread_stdin_handle);
+ThreadCleanup:
-    CloseHandle(stdin_pipe_w);
+    CloseHandle(tdata->stdin_handle);
+    tdata->stdin_handle = NULL;
+    if (tdata->socket_l != INVALID_SOCKET) {
+        closesocket(tdata->socket_l);
+        tdata->socket_l = INVALID_SOCKET;
+    }
+    if (socket_w != INVALID_SOCKET)
+        closesocket(socket_w);
 
     return 0;
 }
@@ -140,76 +159,126 @@ static int _getmode(int fd)
 }
 
 /* Start the reader thread and do all the file handle/descriptor redirection.
-   Returns nonzero on success, zero on error. */
+   Returns the STDIN socket on success, INVALID_SOCKET on error. */
 int win_stdin_start_thread(void) {
     int stdin_fd;
     int stdin_fmode;
+    int rc = 0, socksize = 0;
+    struct win_thread_data *tdata = NULL;
+    SOCKADDR_IN selfaddr;
+    SOCKET socket_r = INVALID_SOCKET;
 
     assert(stdin_thread == NULL);
-    assert(stdin_pipe_r == NULL);
-    assert(stdin_pipe_w == NULL);
-    assert(thread_stdin_handle == NULL);
 
-    /* Create the pipe that win_stdin_thread_func writes to. We reassign the
+    do {
-       read end to be the new stdin that the rest of the program sees. */
+        // Prepare handles for thread
-    if (CreatePipe(&stdin_pipe_r, &stdin_pipe_w, NULL, 0) == 0)
+        tdata = (struct win_thread_data *)safe_zalloc(sizeof(struct win_thread_data));
-        return 0;
 
-    /* Make a copy of the stdin handle to be used by win_stdin_thread_func.  It
+        /* Create the listening socket for the thread. When it starts, it will
-       will remain a reference to the true stdin after we fake stdin to read
+         * accept our connection and begin writing STDIN data to the connection. */
-       from the pipe instead. */
+        tdata->socket_l = (int) inheritable_socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
-    if (DuplicateHandle(GetCurrentProcess(), GetStdHandle(STD_INPUT_HANDLE),
+        if (tdata->socket_l == -1) {
-                        GetCurrentProcess(), &thread_stdin_handle,
+            //fprintf(stderr, "socket error: %d", socket_errno());
-                        0, FALSE, DUPLICATE_SAME_ACCESS) == 0) {
+            break;
-        CloseHandle(stdin_pipe_r);
+        }
-        CloseHandle(stdin_pipe_w);
+        socksize = sizeof(selfaddr);
-        return 0;
+        memset(&selfaddr, 0, socksize);
+        selfaddr.sin_family = AF_INET;
+        selfaddr.sin_addr.S_un.S_addr = htonl(INADDR_LOOPBACK);
+        // Bind to any available loopback port
+        if (SOCKET_ERROR == bind(tdata->socket_l, (SOCKADDR*)&selfaddr, socksize)) {
+            //fprintf(stderr, "bind error: %d", socket_errno());
+            break;
+        }
+        // Get the address that was assigned by bind()
+        if (SOCKET_ERROR == getsockname(tdata->socket_l, (SOCKADDR*)&selfaddr, &socksize)) {
+            //fprintf(stderr, "getsockname error: %d", socket_errno());
+            break;
+        }
+        if (SOCKET_ERROR == listen(tdata->socket_l, 1)) {
+            //fprintf(stderr, "listen error: %d\n", socket_errno());
+            break;
+        }
+
+        /* Make a copy of the stdin handle to be used by win_stdin_thread_func.  It
+           will remain a reference to the true stdin after we fake stdin to read
+           from the socket instead. */
+        if (DuplicateHandle(GetCurrentProcess(), GetStdHandle(STD_INPUT_HANDLE),
+                    GetCurrentProcess(), &tdata->stdin_handle,
+                    0, FALSE, DUPLICATE_SAME_ACCESS) == 0) {
+            //fprintf(stderr, "DuplicateHandle error: %08x", GetLastError());
+            break;
+        }
+
+        /* Start up the thread. We don't bother keeping a reference to it
+           because it runs until program termination. From here on out all reads
+           from the stdin handle or file descriptor 0 will be reading from the
+           socket that is fed by the thread. */
+        stdin_thread = CreateThread(NULL, 0, win_stdin_thread_func, tdata, 0, NULL);
+        if (stdin_thread == NULL) {
+            //fprintf(stderr, "CreateThread error: %08x", GetLastError());
+            break;
+        }
+
+        // Connect to the thread and rearrange our own STDIN handles
+        socket_r = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, 0);
+        if (socket_r == INVALID_SOCKET) {
+            //fprintf(stderr, "socket error: %d", socket_errno());
+            break;
+        }
+        if (SOCKET_ERROR == connect(socket_r, (SOCKADDR*)&selfaddr, socksize)) {
+            //fprintf(stderr, "connect error: %d", socket_errno());
+            break;
+        }
+        if (SOCKET_ERROR == shutdown(socket_r, SD_SEND)) {
+            //fprintf(stderr, "shutdown error: %d", socket_errno());
+            break;
+        }
+
+        /* Set the stdin handle to read from the socket. */
+        if (SetStdHandle(STD_INPUT_HANDLE, (HANDLE) socket_r) == 0) {
+            //fprintf(stderr, "SetStdHandle error: %08x", GetLastError());
+            break;
+        }
+        /* Need to redirect file descriptor 0 also. _open_osfhandle makes a new file
+           descriptor from an existing handle. */
+        /* Remember the newline translation mode (_O_TEXT or _O_BINARY), and
+           restore it in the new file descriptor. */
+        stdin_fmode = _getmode(STDIN_FILENO);
+        stdin_fd = _open_osfhandle((intptr_t) GetStdHandle(STD_INPUT_HANDLE), _O_RDONLY | stdin_fmode);
+        if (stdin_fd == -1) {
+            break;
+        }
+        dup2(stdin_fd, STDIN_FILENO);
+
+        rc = 1;
+    } while (0);
+
+    if (rc != 1) {
+        if (socket_r != INVALID_SOCKET) {
+            if (GetStdHandle(STD_INPUT_HANDLE) == (HANDLE) socket_r &&
+                    tdata->stdin_handle) {
+                // restore STDIN
+                SetStdHandle(STD_INPUT_HANDLE, tdata->stdin_handle);
+                tdata->stdin_handle = NULL; // make sure we don't close it later!
+            }
+            closesocket(socket_r);
+        }
+        if (stdin_thread) {
+            TerminateThread(stdin_thread, 1);
+            stdin_thread = NULL;
+        }
+        if (tdata) {
+            if (tdata->stdin_handle)
+                CloseHandle(tdata->stdin_handle);
+            if (tdata->socket_l != INVALID_SOCKET)
+                closesocket(tdata->socket_l);
+            free(tdata);
+        }
+
+        return INVALID_SOCKET;
     }
+    assert(socket_r != INVALID_SOCKET);
 
-    /* Set the stdin handle to read from the pipe. */
+    return socket_r;
-    if (SetStdHandle(STD_INPUT_HANDLE, stdin_pipe_r) == 0) {
-        CloseHandle(stdin_pipe_r);
-        CloseHandle(stdin_pipe_w);
-        CloseHandle(thread_stdin_handle);
-        return 0;
-    }
-    /* Need to redirect file descriptor 0 also. _open_osfhandle makes a new file
-       descriptor from an existing handle. */
-    /* Remember the newline translation mode (_O_TEXT or _O_BINARY), and
-       restore it in the new file descriptor. */
-    stdin_fmode = _getmode(STDIN_FILENO);
-    stdin_fd = _open_osfhandle((intptr_t) GetStdHandle(STD_INPUT_HANDLE), _O_RDONLY | stdin_fmode);
-    if (stdin_fd == -1) {
-        CloseHandle(stdin_pipe_r);
-        CloseHandle(stdin_pipe_w);
-        CloseHandle(thread_stdin_handle);
-        return 0;
-    }
-    dup2(stdin_fd, STDIN_FILENO);
-
-    /* Finally, start up the thread. We don't bother keeping a reference to it
-       because it runs until program termination. From here on out all reads
-       from the stdin handle or file descriptor 0 will be reading from the
-       anonymous pipe that is fed by the thread. */
-    stdin_thread = CreateThread(NULL, 0, win_stdin_thread_func, NULL, 0, NULL);
-    if (stdin_thread == NULL) {
-        CloseHandle(stdin_pipe_r);
-        CloseHandle(stdin_pipe_w);
-        CloseHandle(thread_stdin_handle);
-        return 0;
-    }
-
-    return 1;
-}
-
-/* Check if input is available on stdin, once all the above has taken place. */
-int win_stdin_ready(void) {
-    DWORD n;
-
-    assert(stdin_pipe_r != NULL);
-
-    if (!PeekNamedPipe(stdin_pipe_r, NULL, 0, NULL, &n, NULL))
-        return 1;
-
-    return n > 0;
 }

nbase/nbase_winunix.h

@@ -192,6 +192,5 @@ Nbase that legitimately use ENOENT for file operations.
 typedef unsigned short u_short_t;
 
 int win_stdin_start_thread(void);
-int win_stdin_ready(void);
 
 #endif /* NBASE_WINUNIX_H */