process.c: Prevent data loss for process output streams

For a terminating process, it's output streams could be closed,
before all data is read.
This commit is contained in:
oni-link 2016-04-23 00:11:33 +02:00
parent 529e2ab178
commit f0967b0f4c
5 changed files with 45 additions and 5 deletions

View File

@ -333,9 +333,47 @@ static void process_close(Process *proc)
}
}
/// Flush output stream.
///
/// @param proc Process, for which an output stream should be flushed.
/// @param stream Stream to flush.
static void flush_stream(Process *proc, Stream *stream)
FUNC_ATTR_NONNULL_ARG(1)
{
if (!stream || stream->closed) {
return;
}
// Limit amount of data we accept after process terminated.
size_t max_bytes = stream->num_bytes + rbuffer_capacity(stream->buffer);
while (!stream->closed && stream->num_bytes < max_bytes) {
// Remember number of bytes before polling
size_t num_bytes = stream->num_bytes;
// Poll for data and process the generated events.
loop_poll_events(&loop, 0);
if (proc->events && !queue_empty(proc->events)) {
queue_process_events(proc->events);
}
// Stream can be closed if it is empty.
if (num_bytes == stream->num_bytes) {
break;
}
}
}
static void process_close_handles(void **argv)
{
Process *proc = argv[0];
// Did our process forked a child that keeps the output streams open?
if (!process_is_tearing_down) {
flush_stream(proc, proc->out);
flush_stream(proc, proc->err);
}
process_close_streams(proc);
process_close(proc);
}

View File

@ -100,6 +100,10 @@ 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:
@ -185,10 +189,6 @@ static void read_event(void **argv)
static void invoke_read_cb(Stream *stream, size_t count, bool eof)
{
if (stream->closed) {
return;
}
// Don't let the stream be closed before the event is processed.
stream->pending_reqs++;

View File

@ -71,6 +71,7 @@ void stream_init(Loop *loop, Stream *stream, int fd, uv_stream_t *uvstream,
stream->closed = false;
stream->buffer = NULL;
stream->events = NULL;
stream->num_bytes = 0;
}
void stream_close(Stream *stream, stream_close_cb on_stream_close)

View File

@ -49,6 +49,7 @@ struct stream {
size_t curmem;
size_t maxmem;
size_t pending_reqs;
size_t num_bytes;
void *data, *internal_data;
bool closed;
Queue *events;

View File

@ -15,7 +15,7 @@ RBuffer *rbuffer_new(size_t capacity)
FUNC_ATTR_WARN_UNUSED_RESULT FUNC_ATTR_NONNULL_RET
{
if (!capacity) {
capacity = 0xffff;
capacity = 0x10000;
}
RBuffer *rv = xmalloc(sizeof(RBuffer) + capacity);