aboutsummaryrefslogtreecommitdiff
path: root/src/nvim/event
diff options
context:
space:
mode:
Diffstat (limited to 'src/nvim/event')
-rw-r--r--src/nvim/event/libuv_process.c12
-rw-r--r--src/nvim/event/process.c123
-rw-r--r--src/nvim/event/process.h16
-rw-r--r--src/nvim/event/rstream.c28
-rw-r--r--src/nvim/event/stream.c7
-rw-r--r--src/nvim/event/stream.h8
6 files changed, 95 insertions, 99 deletions
diff --git a/src/nvim/event/libuv_process.c b/src/nvim/event/libuv_process.c
index 758b35796e..c101cb1bb9 100644
--- a/src/nvim/event/libuv_process.c
+++ b/src/nvim/event/libuv_process.c
@@ -46,22 +46,22 @@ int libuv_process_spawn(LibuvProcess *uvproc)
uvproc->uvstdio[2].flags = UV_IGNORE;
uvproc->uv.data = proc;
- if (proc->in) {
+ if (!proc->in.closed) {
uvproc->uvstdio[0].flags = UV_CREATE_PIPE | UV_READABLE_PIPE;
uvproc->uvstdio[0].data.stream = STRUCT_CAST(uv_stream_t,
- &proc->in->uv.pipe);
+ &proc->in.uv.pipe);
}
- if (proc->out) {
+ if (!proc->out.closed) {
uvproc->uvstdio[1].flags = UV_CREATE_PIPE | UV_WRITABLE_PIPE;
uvproc->uvstdio[1].data.stream = STRUCT_CAST(uv_stream_t,
- &proc->out->uv.pipe);
+ &proc->out.uv.pipe);
}
- if (proc->err) {
+ if (!proc->err.closed) {
uvproc->uvstdio[2].flags = UV_CREATE_PIPE | UV_WRITABLE_PIPE;
uvproc->uvstdio[2].data.stream = STRUCT_CAST(uv_stream_t,
- &proc->err->uv.pipe);
+ &proc->err.uv.pipe);
}
int status;
diff --git a/src/nvim/event/process.c b/src/nvim/event/process.c
index 41e793500a..4eb2dd0baf 100644
--- a/src/nvim/event/process.c
+++ b/src/nvim/event/process.c
@@ -25,28 +25,28 @@
// For pty processes SIGTERM is sent first (in case SIGHUP was not enough).
#define KILL_TIMEOUT_MS 2000
-#define CLOSE_PROC_STREAM(proc, stream) \
- do { \
- if (proc->stream && !proc->stream->closed) { \
- stream_close(proc->stream, NULL, NULL); \
- } \
- } while (0)
-
static bool process_is_tearing_down = false;
/// @returns zero on success, or negative error code
-int process_spawn(Process *proc) FUNC_ATTR_NONNULL_ALL
+int process_spawn(Process *proc, bool in, bool out, bool err)
+ FUNC_ATTR_NONNULL_ALL
{
- if (proc->in) {
- uv_pipe_init(&proc->loop->uv, &proc->in->uv.pipe, 0);
+ if (in) {
+ uv_pipe_init(&proc->loop->uv, &proc->in.uv.pipe, 0);
+ } else {
+ proc->in.closed = true;
}
- if (proc->out) {
- uv_pipe_init(&proc->loop->uv, &proc->out->uv.pipe, 0);
+ if (out) {
+ uv_pipe_init(&proc->loop->uv, &proc->out.uv.pipe, 0);
+ } else {
+ proc->out.closed = true;
}
- if (proc->err) {
- uv_pipe_init(&proc->loop->uv, &proc->err->uv.pipe, 0);
+ if (err) {
+ uv_pipe_init(&proc->loop->uv, &proc->err.uv.pipe, 0);
+ } else {
+ proc->err.closed = true;
}
int status;
@@ -62,14 +62,14 @@ int process_spawn(Process *proc) FUNC_ATTR_NONNULL_ALL
}
if (status) {
- if (proc->in) {
- uv_close((uv_handle_t *)&proc->in->uv.pipe, NULL);
+ if (in) {
+ uv_close((uv_handle_t *)&proc->in.uv.pipe, NULL);
}
- if (proc->out) {
- uv_close((uv_handle_t *)&proc->out->uv.pipe, NULL);
+ if (out) {
+ uv_close((uv_handle_t *)&proc->out.uv.pipe, NULL);
}
- if (proc->err) {
- uv_close((uv_handle_t *)&proc->err->uv.pipe, NULL);
+ if (err) {
+ uv_close((uv_handle_t *)&proc->err.uv.pipe, NULL);
}
if (proc->type == kProcessTypeUv) {
@@ -82,30 +82,27 @@ int process_spawn(Process *proc) FUNC_ATTR_NONNULL_ALL
return status;
}
- if (proc->in) {
- stream_init(NULL, proc->in, -1,
- STRUCT_CAST(uv_stream_t, &proc->in->uv.pipe));
- proc->in->events = proc->events;
- proc->in->internal_data = proc;
- proc->in->internal_close_cb = on_process_stream_close;
+ if (in) {
+ stream_init(NULL, &proc->in, -1,
+ STRUCT_CAST(uv_stream_t, &proc->in.uv.pipe));
+ proc->in.internal_data = proc;
+ proc->in.internal_close_cb = on_process_stream_close;
proc->refcount++;
}
- if (proc->out) {
- stream_init(NULL, proc->out, -1,
- STRUCT_CAST(uv_stream_t, &proc->out->uv.pipe));
- proc->out->events = proc->events;
- proc->out->internal_data = proc;
- proc->out->internal_close_cb = on_process_stream_close;
+ if (out) {
+ stream_init(NULL, &proc->out, -1,
+ STRUCT_CAST(uv_stream_t, &proc->out.uv.pipe));
+ proc->out.internal_data = proc;
+ proc->out.internal_close_cb = on_process_stream_close;
proc->refcount++;
}
- if (proc->err) {
- stream_init(NULL, proc->err, -1,
- STRUCT_CAST(uv_stream_t, &proc->err->uv.pipe));
- proc->err->events = proc->events;
- proc->err->internal_data = proc;
- proc->err->internal_close_cb = on_process_stream_close;
+ if (err) {
+ stream_init(NULL, &proc->err, -1,
+ STRUCT_CAST(uv_stream_t, &proc->err.uv.pipe));
+ proc->err.internal_data = proc;
+ proc->err.internal_close_cb = on_process_stream_close;
proc->refcount++;
}
@@ -136,27 +133,11 @@ void process_teardown(Loop *loop) FUNC_ATTR_NONNULL_ALL
pty_process_teardown(loop);
}
-// Wrappers around `stream_close` that protect against double-closing.
void process_close_streams(Process *proc) FUNC_ATTR_NONNULL_ALL
{
- process_close_in(proc);
- process_close_out(proc);
- process_close_err(proc);
-}
-
-void process_close_in(Process *proc) FUNC_ATTR_NONNULL_ALL
-{
- CLOSE_PROC_STREAM(proc, in);
-}
-
-void process_close_out(Process *proc) FUNC_ATTR_NONNULL_ALL
-{
- CLOSE_PROC_STREAM(proc, out);
-}
-
-void process_close_err(Process *proc) FUNC_ATTR_NONNULL_ALL
-{
- CLOSE_PROC_STREAM(proc, err);
+ stream_may_close(&proc->in);
+ stream_may_close(&proc->out);
+ stream_may_close(&proc->err);
}
/// Synchronously wait for a process to finish
@@ -164,16 +145,15 @@ void process_close_err(Process *proc) FUNC_ATTR_NONNULL_ALL
/// @param process Process instance
/// @param ms Time in milliseconds to wait for the process.
/// 0 for no wait. -1 to wait until the process quits.
-/// @return Exit code of the process.
+/// @return Exit code of the process. proc->status will have the same value.
/// -1 if the timeout expired while the process is still running.
/// -2 if the user interruped the wait.
int process_wait(Process *proc, int ms, MultiQueue *events)
FUNC_ATTR_NONNULL_ARG(1)
{
- int status = -1; // default
bool interrupted = false;
if (!proc->refcount) {
- status = proc->status;
+ int status = proc->status;
LOOP_PROCESS_EVENTS(proc->loop, proc->events, 0);
return status;
}
@@ -209,7 +189,9 @@ int process_wait(Process *proc, int ms, MultiQueue *events)
if (proc->refcount == 1) {
// Job exited, collect status and manually invoke close_cb to free the job
// resources
- status = interrupted ? -2 : proc->status;
+ if (interrupted) {
+ proc->status = -2;
+ }
decref(proc);
if (events) {
// the decref call created an exit event, process it now
@@ -219,7 +201,7 @@ int process_wait(Process *proc, int ms, MultiQueue *events)
proc->refcount--;
}
- return status;
+ return proc->status;
}
/// Ask a process to terminate and eventually kill if it doesn't respond
@@ -233,8 +215,9 @@ void process_stop(Process *proc) FUNC_ATTR_NONNULL_ALL
switch (proc->type) {
case kProcessTypeUv:
// Close the process's stdin. If the process doesn't close its own
- // stdout/stderr, they will be closed when it exits (voluntarily or not).
- process_close_in(proc);
+ // stdout/stderr, they will be closed when it exits(possibly due to being
+ // terminated after a timeout)
+ stream_may_close(&proc->in);
ILOG("Sending SIGTERM to pid %d", proc->pid);
uv_kill(proc->pid, SIGTERM);
break;
@@ -375,15 +358,15 @@ static void flush_stream(Process *proc, Stream *stream)
// Poll for data and process the generated events.
loop_poll_events(proc->loop, 0);
- if (proc->events) {
- multiqueue_process_events(proc->events);
+ if (stream->events) {
+ multiqueue_process_events(stream->events);
}
// Stream can be closed if it is empty.
if (num_bytes == stream->num_bytes) {
- if (stream->read_cb) {
+ if (stream->read_cb && !stream->did_eof) {
// Stream callback could miss EOF handling if a child keeps the stream
- // open.
+ // open. But only send EOF if we haven't already.
stream->read_cb(stream, stream->buffer, 0, stream->cb_data, true);
}
break;
@@ -395,8 +378,8 @@ static void process_close_handles(void **argv)
{
Process *proc = argv[0];
- flush_stream(proc, proc->out);
- flush_stream(proc, proc->err);
+ flush_stream(proc, &proc->out);
+ flush_stream(proc, &proc->err);
process_close_streams(proc);
process_close(proc);
diff --git a/src/nvim/event/process.h b/src/nvim/event/process.h
index 5c00e8e7ec..033ce3604b 100644
--- a/src/nvim/event/process.h
+++ b/src/nvim/event/process.h
@@ -23,13 +23,14 @@ struct process {
uint64_t stopped_time;
const char *cwd;
char **argv;
- Stream *in, *out, *err;
+ Stream in, out, err;
process_exit_cb cb;
internal_process_cb internal_exit_cb, internal_close_cb;
bool closed, detach;
MultiQueue *events;
};
+
static inline Process process_init(Loop *loop, ProcessType type, void *data)
{
return (Process) {
@@ -38,14 +39,14 @@ static inline Process process_init(Loop *loop, ProcessType type, void *data)
.loop = loop,
.events = NULL,
.pid = 0,
- .status = 0,
+ .status = -1,
.refcount = 0,
.stopped_time = 0,
.cwd = NULL,
.argv = NULL,
- .in = NULL,
- .out = NULL,
- .err = NULL,
+ .in = { .closed = false },
+ .out = { .closed = false },
+ .err = { .closed = false },
.cb = NULL,
.closed = false,
.internal_close_cb = NULL,
@@ -54,6 +55,11 @@ static inline Process process_init(Loop *loop, ProcessType type, void *data)
};
}
+static inline bool process_is_stopped(Process *proc)
+{
+ return proc->stopped_time != 0;
+}
+
#ifdef INCLUDE_GENERATED_DECLARATIONS
# include "event/process.h.generated.h"
#endif
diff --git a/src/nvim/event/rstream.c b/src/nvim/event/rstream.c
index 2c4db08b30..e0500ba828 100644
--- a/src/nvim/event/rstream.c
+++ b/src/nvim/event/rstream.c
@@ -105,20 +105,20 @@ static void read_cb(uv_stream_t *uvstream, ssize_t cnt, const uv_buf_t *buf)
{
Stream *stream = uvstream->data;
- if (cnt > 0) {
- stream->num_bytes += (size_t)cnt;
- }
-
if (cnt <= 0) {
- if (cnt != UV_ENOBUFS
- // cnt == 0 means libuv asked for a buffer and decided it wasn't needed:
- // http://docs.libuv.org/en/latest/stream.html#c.uv_read_start.
- //
- // We don't need to do anything with the RBuffer because the next call
- // to `alloc_cb` will return the same unused pointer(`rbuffer_produced`
- // won't be called)
- && cnt != 0) {
- DLOG("closing Stream: %p: %s (%s)", stream,
+ // cnt == 0 means libuv asked for a buffer and decided it wasn't needed:
+ // http://docs.libuv.org/en/latest/stream.html#c.uv_read_start.
+ //
+ // We don't need to do anything with the RBuffer because the next call
+ // to `alloc_cb` will return the same unused pointer(`rbuffer_produced`
+ // won't be called)
+ if (cnt == UV_ENOBUFS || cnt == 0) {
+ return;
+ } else if (cnt == UV_EOF && uvstream->type == UV_TTY) {
+ // The TTY driver might signal TTY without closing the stream
+ invoke_read_cb(stream, 0, true);
+ } else {
+ DLOG("Closing Stream (%p): %s (%s)", stream,
uv_err_name((int)cnt), os_strerror((int)cnt));
// Read error or EOF, either way stop the stream and invoke the callback
// with eof == true
@@ -130,6 +130,7 @@ static void read_cb(uv_stream_t *uvstream, ssize_t cnt, const uv_buf_t *buf)
// at this point we're sure that cnt is positive, no error occurred
size_t nread = (size_t)cnt;
+ stream->num_bytes += nread;
// Data was already written, so all we need is to update 'wpos' to reflect
// the space actually used in the buffer.
rbuffer_produced(stream->buffer, nread);
@@ -187,6 +188,7 @@ static void read_event(void **argv)
if (stream->read_cb) {
size_t count = (uintptr_t)argv[1];
bool eof = (uintptr_t)argv[2];
+ stream->did_eof = eof;
stream->read_cb(stream, stream->buffer, count, stream->cb_data, eof);
}
stream->pending_reqs--;
diff --git a/src/nvim/event/stream.c b/src/nvim/event/stream.c
index 7c865bfe1e..ba25b76ec7 100644
--- a/src/nvim/event/stream.c
+++ b/src/nvim/event/stream.c
@@ -92,6 +92,13 @@ void stream_close(Stream *stream, stream_close_cb on_stream_close, void *data)
}
}
+void stream_may_close(Stream *stream)
+{
+ if (!stream->closed) {
+ stream_close(stream, NULL, NULL);
+ }
+}
+
void stream_close_handle(Stream *stream)
FUNC_ATTR_NONNULL_ALL
{
diff --git a/src/nvim/event/stream.h b/src/nvim/event/stream.h
index d27497e4a4..e713323f5c 100644
--- a/src/nvim/event/stream.h
+++ b/src/nvim/event/stream.h
@@ -14,10 +14,7 @@ typedef struct stream Stream;
///
/// @param stream The Stream instance
/// @param rbuffer The associated RBuffer instance
-/// @param count Number of bytes to read. This must be respected if keeping
-/// the order of events is a requirement. This is because events
-/// may be queued and only processed later when more data is copied
-/// into to the buffer, so one read may starve another.
+/// @param count Number of bytes that was read.
/// @param data User-defined data
/// @param eof If the stream reached EOF.
typedef void (*stream_read_cb)(Stream *stream, RBuffer *buf, size_t count,
@@ -33,6 +30,8 @@ typedef void (*stream_write_cb)(Stream *stream, void *data, int status);
typedef void (*stream_close_cb)(Stream *stream, void *data);
struct stream {
+ bool closed;
+ bool did_eof;
union {
uv_pipe_t pipe;
uv_tcp_t tcp;
@@ -52,7 +51,6 @@ struct stream {
size_t maxmem;
size_t pending_reqs;
size_t num_bytes;
- bool closed;
MultiQueue *events;
};
generated by cgit (git 2.34.1) at 2025-08-03 23:46:57 +0000