From 75c8f755015e4724ba45bd84137d41efa40a3025 Mon Sep 17 00:00:00 2001 From: zeertzjq Date: Fri, 9 Jan 2026 10:45:44 +0800 Subject: [PATCH] refactor(process): don't read from PTY master using uv_pipe_t 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. This commit can be reverted if libuv/libuv#4992 is fixed for uv_pipe_t. --- src/nvim/event/defs.h | 3 ++ src/nvim/event/proc.c | 30 +++++++++++++++---- src/nvim/event/rstream.c | 58 +++++++++++++++++++++++++++++++++++-- src/nvim/event/socket.c | 4 +-- src/nvim/event/stream.c | 16 ++++++++-- src/nvim/event/wstream.c | 5 ++-- src/nvim/os/pty_proc_unix.c | 5 +--- 7 files changed, 103 insertions(+), 18 deletions(-) diff --git a/src/nvim/event/defs.h b/src/nvim/event/defs.h index 20724f9263..3309c934b5 100644 --- a/src/nvim/event/defs.h +++ b/src/nvim/event/defs.h @@ -78,12 +78,15 @@ 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 diff --git a/src/nvim/event/proc.c b/src/nvim/event/proc.c index cec16be752..01fe62d132 100644 --- a/src/nvim/event/proc.c +++ b/src/nvim/event/proc.c @@ -46,6 +46,16 @@ 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 { @@ -53,7 +63,9 @@ int proc_spawn(Proc *proc, bool in, bool out, bool err) } if (out) { - uv_pipe_init(&proc->loop->uv, &proc->out.s.uv.pipe, 0); + if (!out_use_poll) { + uv_pipe_init(&proc->loop->uv, &proc->out.s.uv.pipe, 0); + } } else { proc->out.s.closed = true; } @@ -83,7 +95,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) { + if (out && !out_use_poll) { uv_close((uv_handle_t *)&proc->out.s.uv.pipe, NULL); } if (err) { @@ -101,21 +113,29 @@ int proc_spawn(Proc *proc, bool in, bool out, bool err) } if (in) { - stream_init(NULL, &proc->in, -1, (uv_stream_t *)&proc->in.uv.pipe); + stream_init(NULL, &proc->in, -1, false, (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) { - stream_init(NULL, &proc->out.s, -1, (uv_stream_t *)&proc->out.s.uv.pipe); + if (out_use_poll) { +#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, (uv_stream_t *)&proc->err.s.uv.pipe); + stream_init(NULL, &proc->err.s, -1, false, (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++; diff --git a/src/nvim/event/rstream.c b/src/nvim/event/rstream.c index 8222d256d0..3bea2db613 100644 --- a/src/nvim/event/rstream.c +++ b/src/nvim/event/rstream.c @@ -13,19 +13,25 @@ #include "nvim/os/os_defs.h" #include "nvim/types_defs.h" +#ifndef MSWIN +# include + +# 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, NULL); + stream_init(loop, &stream->s, fd, false, 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, uvstream); + stream_init(NULL, &stream->s, -1, false, uvstream); rstream_init(stream); } @@ -45,6 +51,10 @@ 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); } @@ -72,6 +82,10 @@ 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); } @@ -171,6 +185,46 @@ 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]; diff --git a/src/nvim/event/socket.c b/src/nvim/event/socket.c index 4cc7470115..be20a249f7 100644 --- a/src/nvim/event/socket.c +++ b/src/nvim/event/socket.c @@ -154,7 +154,7 @@ int socket_watcher_accept(SocketWatcher *watcher, RStream *stream) return result; } - stream_init(NULL, &stream->s, -1, client); + stream_init(NULL, &stream->s, -1, false, client); return 0; } @@ -244,7 +244,7 @@ tcp_retry: status = 1; LOOP_PROCESS_EVENTS_UNTIL(&main_loop, NULL, timeout, status != 1); if (status == 0) { - stream_init(NULL, &stream->s, -1, uv_stream); + stream_init(NULL, &stream->s, -1, false, uv_stream); success = true; } else { if (!uv_is_closing((uv_handle_t *)uv_stream)) { diff --git a/src/nvim/event/stream.c b/src/nvim/event/stream.c index 5bb88b91b8..538373e1ad 100644 --- a/src/nvim/event/stream.c +++ b/src/nvim/event/stream.c @@ -40,11 +40,14 @@ int stream_set_blocking(int fd, bool blocking) return retval; } -void stream_init(Loop *loop, Stream *stream, int fd, uv_stream_t *uvstream) +void stream_init(Loop *loop, Stream *stream, int fd, bool poll, 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); + assert(uvstream == NULL ? fd >= 0 && loop != NULL : fd < 0 && loop == NULL && !poll); +#ifdef MSWIN + assert(!poll); +#endif stream->uvstream = uvstream; if (fd >= 0) { @@ -52,6 +55,7 @@ void stream_init(Loop *loop, Stream *stream, int fd, uv_stream_t *uvstream) 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. @@ -67,6 +71,11 @@ void stream_init(Loop *loop, Stream *stream, int fd, uv_stream_t *uvstream) 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); @@ -125,7 +134,8 @@ void stream_close_handle(Stream *stream) } handle = (uv_handle_t *)stream->uvstream; } else { - handle = (uv_handle_t *)&stream->uv.idle; + // All members of the stream->uv union share the same address. + handle = (uv_handle_t *)&stream->uv; } assert(handle != NULL); diff --git a/src/nvim/event/wstream.c b/src/nvim/event/wstream.c index 34da0167dc..40fea25959 100644 --- a/src/nvim/event/wstream.c +++ b/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, NULL); + stream_init(loop, stream, fd, false, 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, uvstream); + stream_init(NULL, stream, -1, false, uvstream); wstream_init(stream, maxmem); } @@ -68,6 +68,7 @@ bool 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); diff --git a/src/nvim/os/pty_proc_unix.c b/src/nvim/os/pty_proc_unix.c index 0c12410f49..1b5f520b5c 100644 --- a/src/nvim/os/pty_proc_unix.c +++ b/src/nvim/os/pty_proc_unix.c @@ -214,10 +214,7 @@ int pty_proc_spawn(PtyProc *ptyproc) && (status = set_duplicating_descriptor(master, &proc->in.uv.pipe))) { goto error; } - if (!proc->out.s.closed - && (status = set_duplicating_descriptor(master, &proc->out.s.uv.pipe))) { - goto error; - } + // The stream_init() call in proc_spawn() will initialize proc->out.s.uv.poll. ptyproc->tty_fd = master; proc->pid = pid;