X-Git-Url: https://www.chiark.greenend.org.uk/ucgi/~ianmdlvl/git?a=blobdiff_plain;f=src%2Fjournal-remote%2Fjournal-remote-parse.c;h=6c096de03aa30e110046f805ee3ec2bdc3e84b0b;hb=0e72da6fe8671d49b4d458519f5ac7600fd04f03;hp=a08ca2fdcbe29d5f94d87c9c52953dbdb2a75dca;hpb=8201af08fa09c2bd0f005fbe262f27e2c5bd2d86;p=elogind.git

diff --git a/src/journal-remote/journal-remote-parse.c b/src/journal-remote/journal-remote-parse.c
index a08ca2fdc..6c096de03 100644
--- a/src/journal-remote/journal-remote-parse.c
+++ b/src/journal-remote/journal-remote-parse.c
@@ -22,42 +22,86 @@
 #include "journal-remote-parse.h"
 #include "journald-native.h"
 
-#define LINE_CHUNK 1024u
+#define LINE_CHUNK 8*1024u
 
 void source_free(RemoteSource *source) {
         if (!source)
                 return;
 
-        if (source->fd >= 0) {
+        if (source->fd >= 0 && !source->passive_fd) {
                 log_debug("Closing fd:%d (%s)", source->fd, source->name);
                 safe_close(source->fd);
         }
+
         free(source->name);
         free(source->buf);
         iovw_free_contents(&source->iovw);
 
+        log_debug("Writer ref count %i", source->writer->n_ref);
+        writer_unref(source->writer);
+
         sd_event_source_unref(source->event);
 
         free(source);
 }
 
+/**
+ * Initialize zero-filled source with given values. On success, takes
+ * ownerhship of fd and writer, otherwise does not touch them.
+ */
+RemoteSource* source_new(int fd, bool passive_fd, char *name, Writer *writer) {
+
+        RemoteSource *source;
+
+        log_debug("Creating source for %sfd:%d (%s)",
+                  passive_fd ? "passive " : "", fd, name);
+
+        assert(fd >= 0);
+
+        source = new0(RemoteSource, 1);
+        if (!source)
+                return NULL;
+
+        source->fd = fd;
+        source->passive_fd = passive_fd;
+        source->name = name;
+        source->writer = writer;
+
+        return source;
+}
+
+static char* realloc_buffer(RemoteSource *source, size_t size) {
+        char *b, *old = source->buf;
+
+        b = GREEDY_REALLOC(source->buf, source->size, size);
+        if (!b)
+                return NULL;
+
+        iovw_rebase(&source->iovw, old, source->buf);
+
+        return b;
+}
+
 static int get_line(RemoteSource *source, char **line, size_t *size) {
-        ssize_t n, remain;
+        ssize_t n;
         char *c = NULL;
-        char *newbuf = NULL;
-        size_t newsize = 0;
 
         assert(source);
         assert(source->state == STATE_LINE);
+        assert(source->offset <= source->filled);
         assert(source->filled <= source->size);
         assert(source->buf == NULL || source->size > 0);
+        assert(source->fd >= 0);
 
         while (true) {
-                if (source->buf)
-                        c = memchr(source->buf + source->scanned, '\n',
-                                   source->filled - source->scanned);
-                if (c != NULL)
-                        break;
+                if (source->buf) {
+                        size_t start = MAX(source->scanned, source->offset);
+
+                        c = memchr(source->buf + start, '\n',
+                                   source->filled - start);
+                        if (c != NULL)
+                                break;
+                }
 
                 source->scanned = source->filled;
                 if (source->scanned >= DATA_SIZE_MAX) {
@@ -65,23 +109,24 @@ static int get_line(RemoteSource *source, char **line, size_t *size) {
                         return -E2BIG;
                 }
 
-                if (source->fd < 0)
+                if (source->passive_fd)
                         /* we have to wait for some data to come to us */
                         return -EWOULDBLOCK;
 
                 if (source->size - source->filled < LINE_CHUNK &&
-                    !GREEDY_REALLOC(source->buf, source->size,
-                                    MAX(source->filled + LINE_CHUNK, DATA_SIZE_MAX)))
+                    !realloc_buffer(source,
+                                    MIN(source->filled + LINE_CHUNK, ENTRY_SIZE_MAX)))
                                 return log_oom();
 
-                assert(source->size - source->filled >= LINE_CHUNK);
+                assert(source->size - source->filled >= LINE_CHUNK ||
+                       source->size == ENTRY_SIZE_MAX);
 
                 n = read(source->fd, source->buf + source->filled,
-                         MAX(source->size, DATA_SIZE_MAX) - source->filled);
+                         source->size - source->filled);
                 if (n < 0) {
                         if (errno != EAGAIN && errno != EWOULDBLOCK)
-                                log_error("read(%d, ..., %zd): %m", source->fd,
-                                          source->size - source->filled);
+                                log_error_errno(errno, "read(%d, ..., %zu): %m", source->fd,
+                                                source->size - source->filled);
                         return -errno;
                 } else if (n == 0)
                         return 0;
@@ -89,23 +134,9 @@ static int get_line(RemoteSource *source, char **line, size_t *size) {
                 source->filled += n;
         }
 
-        *line = source->buf;
-        *size = c + 1 - source->buf;
-
-        /* Check if something remains */
-        remain = source->buf + source->filled - c - 1;
-        assert(remain >= 0);
-        if (remain) {
-                newsize = MAX(remain, LINE_CHUNK);
-                newbuf = malloc(newsize);
-                if (!newbuf)
-                        return log_oom();
-                memcpy(newbuf, c + 1, remain);
-        }
-        source->buf = newbuf;
-        source->size = newsize;
-        source->filled = remain;
-        source->scanned = 0;
+        *line = source->buf + source->offset;
+        *size = c + 1 - source->buf - source->offset;
+        source->offset += *size;
 
         return 1;
 }
@@ -114,8 +145,7 @@ int push_data(RemoteSource *source, const char *data, size_t size) {
         assert(source);
         assert(source->state != STATE_EOF);
 
-        if (!GREEDY_REALLOC(source->buf, source->size,
-                            source->filled + size)) {
+        if (!realloc_buffer(source, source->filled + size)) {
                 log_error("Failed to store received data of size %zu "
                           "(in addition to existing %zu bytes with %zu filled): %s",
                           size, source->size, source->filled, strerror(ENOMEM));
@@ -129,35 +159,35 @@ int push_data(RemoteSource *source, const char *data, size_t size) {
 }
 
 static int fill_fixed_size(RemoteSource *source, void **data, size_t size) {
-        int n;
-        char *newbuf = NULL;
-        size_t newsize = 0, remain;
 
         assert(source);
         assert(source->state == STATE_DATA_START ||
                source->state == STATE_DATA ||
                source->state == STATE_DATA_FINISH);
         assert(size <= DATA_SIZE_MAX);
+        assert(source->offset <= source->filled);
         assert(source->filled <= source->size);
-        assert(source->scanned <= source->filled);
         assert(source->buf != NULL || source->size == 0);
         assert(source->buf == NULL || source->size > 0);
+        assert(source->fd >= 0);
         assert(data);
 
-        while(source->filled < size) {
-                if (source->fd < 0)
+        while (source->filled - source->offset < size) {
+                int n;
+
+                if (source->passive_fd)
                         /* we have to wait for some data to come to us */
                         return -EWOULDBLOCK;
 
-                if (!GREEDY_REALLOC(source->buf, source->size, size))
+                if (!realloc_buffer(source, source->offset + size))
                         return log_oom();
 
                 n = read(source->fd, source->buf + source->filled,
                          source->size - source->filled);
                 if (n < 0) {
                         if (errno != EAGAIN && errno != EWOULDBLOCK)
-                                log_error("read(%d, ..., %zd): %m", source->fd,
-                                          source->size - source->filled);
+                                log_error_errno(errno, "read(%d, ..., %zu): %m", source->fd,
+                                                source->size - source->filled);
                         return -errno;
                 } else if (n == 0)
                         return 0;
@@ -165,29 +195,15 @@ static int fill_fixed_size(RemoteSource *source, void **data, size_t size) {
                 source->filled += n;
         }
 
-        *data = source->buf;
-
-        /* Check if something remains */
-        assert(size <= source->filled);
-        remain = source->filled - size;
-        if (remain) {
-                newsize = MAX(remain, LINE_CHUNK);
-                newbuf = malloc(newsize);
-                if (!newbuf)
-                        return log_oom();
-                memcpy(newbuf, source->buf + size, remain);
-        }
-        source->buf = newbuf;
-        source->size = newsize;
-        source->filled = remain;
-        source->scanned = 0;
+        *data = source->buf + source->offset;
+        source->offset += size;
 
         return 1;
 }
 
 static int get_data_size(RemoteSource *source) {
         int r;
-        _cleanup_free_ void *data = NULL;
+        void *data;
 
         assert(source);
         assert(source->state == STATE_DATA_START);
@@ -199,7 +215,7 @@ static int get_data_size(RemoteSource *source) {
 
         source->data_size = le64toh( *(uint64_t *) data );
         if (source->data_size > DATA_SIZE_MAX) {
-                log_error("Stream declares field with size %zu > %u == DATA_SIZE_MAX",
+                log_error("Stream declares field with size %zu > DATA_SIZE_MAX = %u",
                           source->data_size, DATA_SIZE_MAX);
                 return -EINVAL;
         }
@@ -225,7 +241,7 @@ static int get_data_data(RemoteSource *source, void **data) {
 
 static int get_data_newline(RemoteSource *source) {
         int r;
-        _cleanup_free_ char *data = NULL;
+        char *data;
 
         assert(source);
         assert(source->state == STATE_DATA_FINISH);
@@ -314,16 +330,13 @@ int process_data(RemoteSource *source) {
                 assert(line[n-1] == '\n');
 
                 if (n == 1) {
-                        log_debug("Received empty line, event is ready");
-                        free(line);
+                        log_trace("Received empty line, event is ready");
                         return 1;
                 }
 
                 r = process_dunder(source, line, n);
-                if (r != 0) {
-                        free(line);
+                if (r != 0)
                         return r < 0 ? r : 0;
-                }
 
                 /* MESSAGE=xxx\n
                    or
@@ -331,23 +344,25 @@ int process_data(RemoteSource *source) {
                    LLLLLLLL0011223344...\n
                 */
                 sep = memchr(line, '=', n);
-                if (sep)
+                if (sep) {
                         /* chomp newline */
                         n--;
-                else
+
+                        r = iovw_put(&source->iovw, line, n);
+                        if (r < 0)
+                                return r;
+                } else {
                         /* replace \n with = */
                         line[n-1] = '=';
-                log_debug("Received: %.*s", (int) n, line);
 
-                r = iovw_put(&source->iovw, line, n);
-                if (r < 0) {
-                        log_error("Failed to put line in iovect");
-                        free(line);
-                        return r;
+                        source->field_len = n;
+                        source->state = STATE_DATA_START;
+
+                        /* we cannot put the field in iovec until we have all data */
                 }
 
-                if (!sep)
-                        source->state = STATE_DATA_START;
+                log_trace("Received: %.*s (%s)", (int) n, line, sep ? "text" : "binary");
+
                 return 0; /* continue */
         }
 
@@ -355,7 +370,7 @@ int process_data(RemoteSource *source) {
                 assert(source->data_size == 0);
 
                 r = get_data_size(source);
-                log_debug("get_data_size() -> %d", r);
+                // log_debug("get_data_size() -> %d", r);
                 if (r < 0)
                         return r;
                 if (r == 0) {
@@ -370,11 +385,12 @@ int process_data(RemoteSource *source) {
 
         case STATE_DATA: {
                 void *data;
+                char *field;
 
                 assert(source->data_size > 0);
 
                 r = get_data_data(source, &data);
-                log_debug("get_data_data() -> %d", r);
+                // log_debug("get_data_data() -> %d", r);
                 if (r < 0)
                         return r;
                 if (r == 0) {
@@ -384,11 +400,12 @@ int process_data(RemoteSource *source) {
 
                 assert(data);
 
-                r = iovw_put(&source->iovw, data, source->data_size);
-                if (r < 0) {
-                        log_error("failed to put binary buffer in iovect");
+                field = (char*) data - sizeof(uint64_t) - source->field_len;
+                memmove(field + sizeof(uint64_t), field, source->field_len);
+
+                r = iovw_put(&source->iovw, field + sizeof(uint64_t), source->field_len + source->data_size);
+                if (r < 0)
                         return r;
-                }
 
                 source->state = STATE_DATA_FINISH;
 
@@ -397,7 +414,7 @@ int process_data(RemoteSource *source) {
 
         case STATE_DATA_FINISH:
                 r = get_data_newline(source);
-                log_debug("get_data_newline() -> %d", r);
+                // log_debug("get_data_newline() -> %d", r);
                 if (r < 0)
                         return r;
                 if (r == 0) {
@@ -414,19 +431,20 @@ int process_data(RemoteSource *source) {
         }
 }
 
-int process_source(RemoteSource *source, Writer *writer, bool compress, bool seal) {
+int process_source(RemoteSource *source, bool compress, bool seal) {
+        size_t remain, target;
         int r;
 
         assert(source);
-        assert(writer);
+        assert(source->writer);
 
         r = process_data(source);
         if (r <= 0)
                 return r;
 
         /* We have a full event */
-        log_info("Received a full event from source@%p fd:%d (%s)",
-                 source, source->fd, source->name);
+        log_trace("Received full event from source@%p fd:%d (%s)",
+                  source, source->fd, source->name);
 
         if (!source->iovw.count) {
                 log_warning("Entry with no payload, skipping");
@@ -436,14 +454,45 @@ int process_source(RemoteSource *source, Writer *writer, bool compress, bool sea
         assert(source->iovw.iovec);
         assert(source->iovw.count);
 
-        r = writer_write(writer, &source->iovw, &source->ts, compress, seal);
+        r = writer_write(source->writer, &source->iovw, &source->ts, compress, seal);
         if (r < 0)
-                log_error("Failed to write entry of %zu bytes: %s",
-                          iovw_size(&source->iovw), strerror(-r));
+                log_error_errno(r, "Failed to write entry of %zu bytes: %m",
+                                iovw_size(&source->iovw));
         else
                 r = 1;
 
  freeing:
         iovw_free_contents(&source->iovw);
+
+        /* possibly reset buffer position */
+        remain = source->filled - source->offset;
+
+        if (remain == 0) /* no brainer */
+                source->offset = source->scanned = source->filled = 0;
+        else if (source->offset > source->size - source->filled &&
+                 source->offset > remain) {
+                memcpy(source->buf, source->buf + source->offset, remain);
+                source->offset = source->scanned = 0;
+                source->filled = remain;
+        }
+
+        target = source->size;
+        while (target > 16 * LINE_CHUNK && remain < target / 2)
+                target /= 2;
+        if (target < source->size) {
+                char *tmp;
+
+                tmp = realloc(source->buf, target);
+                if (!tmp)
+                        log_warning("Failed to reallocate buffer to (smaller) size %zu",
+                                    target);
+                else {
+                        log_debug("Reallocated buffer from %zu to %zu bytes",
+                                  source->size, target);
+                        source->buf = tmp;
+                        source->size = target;
+                }
+        }
+
         return r;
 }