revert: "refactor(process): don't read from PTY master using uv_pipe_t" (#37401)

This reverts commit 75c8f75501.

(cherry picked from commit 9c15a382de)

src/nvim/event/defs.h

@@ -78,15 +78,12 @@ typedef void (*stream_close_cb)(Stream *stream, void *data);
 
 struct stream {
   bool closed;
-  bool use_poll;
   union {
     uv_pipe_t pipe;
     uv_tcp_t tcp;
     uv_idle_t idle;
 #ifdef MSWIN
     uv_tty_t tty;
-#else
-    uv_poll_t poll;
 #endif
   } uv;
   uv_stream_t *uvstream;  ///< NULL when the stream is a file

src/nvim/event/proc.c

@@ -45,16 +45,6 @@ int proc_spawn(Proc *proc, bool in, bool out, bool err)
   // forwarding stderr contradicts with processing it internally
   assert(!(err && proc->fwd_err));
 
-#ifdef MSWIN
-  const bool out_use_poll = false;
-#else
-  // Using uv_pipe_t to read from PTY master may drop data if the PTY process exits
-  // immediately after output, as libuv treats a partial read after POLLHUP as EOF,
-  // which isn't true for PTY master on Linux. Therefore use uv_poll_t instead. #3030
-  // Ref: https://github.com/libuv/libuv/issues/4992
-  const bool out_use_poll = proc->type == kProcTypePty;
-#endif
-
   if (in) {
     uv_pipe_init(&proc->loop->uv, &proc->in.uv.pipe, 0);
   } else {
@@ -62,9 +52,7 @@ int proc_spawn(Proc *proc, bool in, bool out, bool err)
   }
 
   if (out) {
-    if (!out_use_poll) {
+    uv_pipe_init(&proc->loop->uv, &proc->out.s.uv.pipe, 0);
-      uv_pipe_init(&proc->loop->uv, &proc->out.s.uv.pipe, 0);
-    }
   } else {
     proc->out.s.closed = true;
   }
@@ -94,7 +82,7 @@ int proc_spawn(Proc *proc, bool in, bool out, bool err)
     if (in) {
       uv_close((uv_handle_t *)&proc->in.uv.pipe, NULL);
     }
-    if (out && !out_use_poll) {
+    if (out) {
       uv_close((uv_handle_t *)&proc->out.s.uv.pipe, NULL);
     }
     if (err) {
@@ -112,29 +100,21 @@ int proc_spawn(Proc *proc, bool in, bool out, bool err)
   }
 
   if (in) {
-    stream_init(NULL, &proc->in, -1, false, (uv_stream_t *)&proc->in.uv.pipe);
+    stream_init(NULL, &proc->in, -1, (uv_stream_t *)&proc->in.uv.pipe);
     proc->in.internal_data = proc;
     proc->in.internal_close_cb = on_proc_stream_close;
     proc->refcount++;
   }
 
   if (out) {
-    if (out_use_poll) {
+    stream_init(NULL, &proc->out.s, -1, (uv_stream_t *)&proc->out.s.uv.pipe);
-#ifdef MSWIN
-      abort();
-#else
-      stream_init(proc->loop, &proc->out.s, ((PtyProc *)proc)->tty_fd, true, NULL);
-#endif
-    } else {
-      stream_init(NULL, &proc->out.s, -1, false, (uv_stream_t *)&proc->out.s.uv.pipe);
-    }
     proc->out.s.internal_data = proc;
     proc->out.s.internal_close_cb = on_proc_stream_close;
     proc->refcount++;
   }
 
   if (err) {
-    stream_init(NULL, &proc->err.s, -1, false, (uv_stream_t *)&proc->err.s.uv.pipe);
+    stream_init(NULL, &proc->err.s, -1, (uv_stream_t *)&proc->err.s.uv.pipe);
     proc->err.s.internal_data = proc;
     proc->err.s.internal_close_cb = on_proc_stream_close;
     proc->refcount++;

src/nvim/event/rstream.c

@@ -13,25 +13,19 @@
 #include "nvim/os/os_defs.h"
 #include "nvim/types_defs.h"
 
-#ifndef MSWIN
-# include <errno.h>
-
-# include "nvim/fileio.h"
-#endif
-
 #include "event/rstream.c.generated.h"
 
 void rstream_init_fd(Loop *loop, RStream *stream, int fd)
   FUNC_ATTR_NONNULL_ARG(1, 2)
 {
-  stream_init(loop, &stream->s, fd, false, NULL);
+  stream_init(loop, &stream->s, fd, NULL);
   rstream_init(stream);
 }
 
 void rstream_init_stream(RStream *stream, uv_stream_t *uvstream)
   FUNC_ATTR_NONNULL_ARG(1, 2)
 {
-  stream_init(NULL, &stream->s, -1, false, uvstream);
+  stream_init(NULL, &stream->s, -1, uvstream);
   rstream_init(stream);
 }
 
@@ -51,10 +45,6 @@ void rstream_start_inner(RStream *stream)
 {
   if (stream->s.uvstream) {
     uv_read_start(stream->s.uvstream, alloc_cb, read_cb);
-#ifndef MSWIN
-  } else if (stream->s.use_poll) {
-    uv_poll_start(&stream->s.uv.poll, UV_READABLE, poll_cb);
-#endif
   } else {
     uv_idle_start(&stream->s.uv.idle, fread_idle_cb);
   }
@@ -82,10 +72,6 @@ void rstream_stop_inner(RStream *stream)
 {
   if (stream->s.uvstream) {
     uv_read_stop(stream->s.uvstream);
-#ifndef MSWIN
-  } else if (stream->s.use_poll) {
-    uv_poll_stop(&stream->s.uv.poll);
-#endif
   } else {
     uv_idle_stop(&stream->s.uv.idle);
   }
@@ -185,46 +171,6 @@ static void fread_idle_cb(uv_idle_t *handle)
   invoke_read_cb(stream, false);
 }
 
-#ifndef MSWIN
-static void poll_cb(uv_poll_t *handle, int status, int events)
-{
-  RStream *stream = handle->data;
-
-  if (status < 0) {
-    ELOG("poll error on Stream (%p) with descriptor %d: %s (%s)", (void *)stream,
-         stream->s.fd, uv_err_name(status), os_strerror(status));
-    return;
-  }
-  if (!(events & UV_READABLE)) {
-    return;
-  }
-
-  ssize_t cnt = read_eintr(stream->s.fd, stream->write_pos, rstream_space(stream));
-
-  if (cnt < 0) {
-    if (errno == EAGAIN || errno == EWOULDBLOCK) {
-      return;
-    }
-    DLOG("closing Stream (%p) with descriptor %d: %s", (void *)stream,
-         stream->s.fd, strerror(errno));
-  }
-
-  if (cnt <= 0) {
-    // Read error or EOF, either way stop the stream and invoke the callback
-    // with eof == true
-    uv_poll_stop(handle);
-    invoke_read_cb(stream, true);
-    return;
-  }
-
-  // at this point we're sure that cnt is positive, no error occurred
-  size_t nread = (size_t)cnt;
-  stream->num_bytes += nread;
-  stream->write_pos += cnt;
-  invoke_read_cb(stream, false);
-}
-#endif
-
 static void read_event(void **argv)
 {
   RStream *stream = argv[0];

src/nvim/event/socket.c

@@ -239,7 +239,7 @@ int socket_watcher_accept(SocketWatcher *watcher, RStream *stream)
     return result;
   }
 
-  stream_init(NULL, &stream->s, -1, false, client);
+  stream_init(NULL, &stream->s, -1, client);
   return 0;
 }
 
@@ -332,7 +332,7 @@ tcp_retry:
   status = 1;
   LOOP_PROCESS_EVENTS_UNTIL(&main_loop, NULL, timeout, status != 1);
   if (status == 0) {
-    stream_init(NULL, &stream->s, -1, false, uv_stream);
+    stream_init(NULL, &stream->s, -1, uv_stream);
     assert(uv_stream->data != &closed);  // Should have been set by stream_init().
     success = true;
   } else {

src/nvim/event/stream.c

@@ -40,14 +40,11 @@ int stream_set_blocking(int fd, bool blocking)
   return retval;
 }
 
-void stream_init(Loop *loop, Stream *stream, int fd, bool poll, uv_stream_t *uvstream)
+void stream_init(Loop *loop, Stream *stream, int fd, uv_stream_t *uvstream)
   FUNC_ATTR_NONNULL_ARG(2)
 {
   // The underlying stream is either a file or an existing uv stream.
-  assert(uvstream == NULL ? fd >= 0 && loop != NULL : fd < 0 && loop == NULL && !poll);
+  assert(uvstream == NULL ? fd >= 0 && loop != NULL : fd < 0 && loop == NULL);
-#ifdef MSWIN
-  assert(!poll);
-#endif
   stream->uvstream = uvstream;
 
   if (fd >= 0) {
@@ -55,7 +52,6 @@ void stream_init(Loop *loop, Stream *stream, int fd, bool poll, uv_stream_t *uvs
     stream->fd = fd;
 
     if (type == UV_FILE) {
-      assert(!poll);
       // Non-blocking file reads are simulated with an idle handle that reads in
       // chunks of the ring buffer size, giving time for other events to be
       // processed between reads.
@@ -71,11 +67,6 @@ void stream_init(Loop *loop, Stream *stream, int fd, bool poll, uv_stream_t *uvs
         SetConsoleMode(stream->uv.tty.handle, dwMode);
       }
       stream->uvstream = (uv_stream_t *)&stream->uv.tty;
-#else
-    } else if (poll) {
-      uv_poll_init(&loop->uv, &stream->uv.poll, fd);
-      stream->uv.poll.data = stream;
-      stream->use_poll = true;
 #endif
     } else {
       assert(type == UV_NAMED_PIPE || type == UV_TTY);
@@ -134,8 +125,7 @@ void stream_close_handle(Stream *stream)
     }
     handle = (uv_handle_t *)stream->uvstream;
   } else {
-    // All members of the stream->uv union share the same address.
+    handle = (uv_handle_t *)&stream->uv.idle;
-    handle = (uv_handle_t *)&stream->uv;
   }
 
   assert(handle != NULL);

src/nvim/event/wstream.c

@@ -23,14 +23,14 @@ typedef struct {
 void wstream_init_fd(Loop *loop, Stream *stream, int fd, size_t maxmem)
   FUNC_ATTR_NONNULL_ARG(1) FUNC_ATTR_NONNULL_ARG(2)
 {
-  stream_init(loop, stream, fd, false, NULL);
+  stream_init(loop, stream, fd, NULL);
   wstream_init(stream, maxmem);
 }
 
 void wstream_init_stream(Stream *stream, uv_stream_t *uvstream, size_t maxmem)
   FUNC_ATTR_NONNULL_ARG(1) FUNC_ATTR_NONNULL_ARG(2)
 {
-  stream_init(NULL, stream, -1, false, uvstream);
+  stream_init(NULL, stream, -1, uvstream);
   wstream_init(stream, maxmem);
 }
 
@@ -68,7 +68,6 @@ int wstream_write(Stream *stream, WBuffer *buffer)
   FUNC_ATTR_NONNULL_ALL
 {
   assert(stream->maxmem);
-  assert(!stream->use_poll);
   // This should not be called after a stream was freed
   assert(!stream->closed);
 

src/nvim/os/pty_proc_unix.c

@@ -217,7 +217,10 @@ int pty_proc_spawn(PtyProc *ptyproc)
       && (status = set_duplicating_descriptor(master, &proc->in.uv.pipe))) {
     goto error;
   }
-  // The stream_init() call in proc_spawn() will initialize proc->out.s.uv.poll.
+  if (!proc->out.s.closed
+      && (status = set_duplicating_descriptor(master, &proc->out.s.uv.pipe))) {
+    goto error;
+  }
 
   ptyproc->tty_fd = master;
   proc->pid = pid;