From 538bd9b42dabf1145cf7ab77f32347d516b01e88 Mon Sep 17 00:00:00 2001
From: Lennart Poettering <lennart@poettering.net>
Date: Tue, 12 May 2020 18:56:34 +0200
Subject: [PATCH] journald: rework pid change handling
Let's introduce an explicit line ending marker for line endings due to
pid change.
Let's also make sure we don't get confused with buffer management.
Fixes: #15654
(cherry picked from commit 45ba1ea5e9264d385fa565328fe957ef1d78caa1)
Resolves: #2029426
---
src/journal/journald-stream.c | 99 +++++++++++++++++++++++------------
1 file changed, 66 insertions(+), 33 deletions(-)
diff --git a/src/journal/journald-stream.c b/src/journal/journald-stream.c
index ab1a855943..5be5b0939c 100644
--- a/src/journal/journald-stream.c
+++ b/src/journal/journald-stream.c
@@ -54,6 +54,7 @@ typedef enum LineBreak {
LINE_BREAK_NUL,
LINE_BREAK_LINE_MAX,
LINE_BREAK_EOF,
+ LINE_BREAK_PID_CHANGE,
_LINE_BREAK_MAX,
_LINE_BREAK_INVALID = -1,
} LineBreak;
@@ -310,6 +311,7 @@ static int stdout_stream_log(
[LINE_BREAK_NUL] = "_LINE_BREAK=nul",
[LINE_BREAK_LINE_MAX] = "_LINE_BREAK=line-max",
[LINE_BREAK_EOF] = "_LINE_BREAK=eof",
+ [LINE_BREAK_PID_CHANGE] = "_LINE_BREAK=pid-change",
};
const char *c = line_break_field_table[line_break];
@@ -431,21 +433,43 @@ static int stdout_stream_line(StdoutStream *s, char *p, LineBreak line_break) {
assert_not_reached("Unknown stream state");
}
-static int stdout_stream_scan(StdoutStream *s, bool force_flush) {
- char *p;
- size_t remaining;
+static int stdout_stream_found(
+ StdoutStream *s,
+ char *p,
+ size_t l,
+ LineBreak line_break) {
+
+ char saved;
int r;
assert(s);
+ assert(p);
+
+ /* Let's NUL terminate the specified buffer for this call, and revert back afterwards */
+ saved = p[l];
+ p[l] = 0;
+ r = stdout_stream_line(s, p, line_break);
+ p[l] = saved;
- p = s->buffer;
- remaining = s->length;
+ return r;
+}
+
+static int stdout_stream_scan(
+ StdoutStream *s,
+ char *p,
+ size_t remaining,
+ LineBreak force_flush,
+ size_t *ret_consumed) {
- /* XXX: This function does nothing if (s->length == 0) */
+ size_t consumed = 0;
+ int r;
+
+ assert(s);
+ assert(p);
for (;;) {
LineBreak line_break;
- size_t skip;
+ size_t skip, found;
char *end1, *end2;
end1 = memchr(p, '\n', remaining);
@@ -453,43 +477,40 @@ static int stdout_stream_scan(StdoutStream *s, bool force_flush) {
if (end2) {
/* We found a NUL terminator */
- skip = end2 - p + 1;
+ found = end2 - p;
+ skip = found + 1;
line_break = LINE_BREAK_NUL;
} else if (end1) {
/* We found a \n terminator */
- *end1 = 0;
- skip = end1 - p + 1;
+ found = end1 - p;
+ skip = found + 1;
line_break = LINE_BREAK_NEWLINE;
} else if (remaining >= s->server->line_max) {
/* Force a line break after the maximum line length */
- *(p + s->server->line_max) = 0;
- skip = remaining;
+ found = skip = s->server->line_max;
line_break = LINE_BREAK_LINE_MAX;
} else
break;
- r = stdout_stream_line(s, p, line_break);
+ r = stdout_stream_found(s, p, found, line_break);
if (r < 0)
return r;
- remaining -= skip;
p += skip;
+ consumed += skip;
+ remaining -= skip;
}
- if (force_flush && remaining > 0) {
- p[remaining] = 0;
- r = stdout_stream_line(s, p, LINE_BREAK_EOF);
+ if (force_flush >= 0 && remaining > 0) {
+ r = stdout_stream_found(s, p, remaining, force_flush);
if (r < 0)
return r;
- p += remaining;
- remaining = 0;
+ consumed += remaining;
}
- if (p > s->buffer) {
- memmove(s->buffer, p, remaining);
- s->length = remaining;
- }
+ if (ret_consumed)
+ *ret_consumed = consumed;
return 0;
}
@@ -497,11 +518,12 @@ static int stdout_stream_scan(StdoutStream *s, bool force_flush) {
static int stdout_stream_process(sd_event_source *es, int fd, uint32_t revents, void *userdata) {
uint8_t buf[CMSG_SPACE(sizeof(struct ucred))];
StdoutStream *s = userdata;
+ size_t limit, consumed;
struct ucred *ucred = NULL;
struct cmsghdr *cmsg;
struct iovec iovec;
- size_t limit;
ssize_t l;
+ char *p;
int r;
struct msghdr msghdr = {
@@ -529,7 +551,7 @@ static int stdout_stream_process(sd_event_source *es, int fd, uint32_t revents,
/* Try to make use of the allocated buffer in full, but never read more than the configured line size. Also,
* always leave room for a terminating NUL we might need to add. */
limit = MIN(s->allocated - 1, s->server->line_max);
-
+ assert(s->length <= limit);
iovec = IOVEC_MAKE(s->buffer + s->length, limit - s->length);
l = recvmsg(s->fd, &msghdr, MSG_DONTWAIT|MSG_CMSG_CLOEXEC);
@@ -543,7 +565,7 @@ static int stdout_stream_process(sd_event_source *es, int fd, uint32_t revents,
cmsg_close_all(&msghdr);
if (l == 0) {
- stdout_stream_scan(s, true);
+ (void) stdout_stream_scan(s, s->buffer, s->length, /* force_flush = */ LINE_BREAK_EOF, NULL);
goto terminate;
}
@@ -562,22 +584,33 @@ static int stdout_stream_process(sd_event_source *es, int fd, uint32_t revents,
* in the meantime.
*/
if (ucred && ucred->pid != s->ucred.pid) {
- /* force out any previously half-written lines from a
- * different process, before we switch to the new ucred
- * structure for everything we just added */
- r = stdout_stream_scan(s, true);
+ /* Force out any previously half-written lines from a different process, before we switch to
+ * the new ucred structure for everything we just added */
+ r = stdout_stream_scan(s, s->buffer, s->length, /* force_flush = */ LINE_BREAK_PID_CHANGE, NULL);
if (r < 0)
goto terminate;
- s->ucred = *ucred;
s->context = client_context_release(s->server, s->context);
+
+ p = s->buffer + s->length;
+ } else {
+ p = s->buffer;
+ l += s->length;
}
- s->length += l;
- r = stdout_stream_scan(s, false);
+ /* Always copy in the new credentials */
+ if (ucred)
+ s->ucred = *ucred;
+
+ r = stdout_stream_scan(s, p, l, _LINE_BREAK_INVALID, &consumed);
if (r < 0)
goto terminate;
+ /* Move what wasn't consumed to the front of the buffer */
+ assert(consumed <= (size_t) l);
+ s->length = l - consumed;
+ memmove(s->buffer, p + consumed, s->length);
+
return 1;
terminate: