#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);
- close(source->fd);
+ safe_close(source->fd);
}
+
free(source->name);
free(source->buf);
iovw_free_contents(&source->iovw);
+
+ log_debug("Writer ref count %u", 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);
- if (source->buf)
- c = memchr(source->buf, '\n', source->filled);
+ while (true) {
+ if (source->buf) {
+ size_t start = MAX(source->scanned, source->offset);
- if (c != NULL)
- goto docopy;
+ c = memchr(source->buf + start, '\n',
+ source->filled - start);
+ if (c != NULL)
+ break;
+ }
- resize:
- if (source->fd < 0)
- /* we have to wait for some data to come to us */
- return -EWOULDBLOCK;
+ source->scanned = source->filled;
+ if (source->scanned >= DATA_SIZE_MAX) {
+ log_error("Entry is bigger than %u bytes.", DATA_SIZE_MAX);
+ return -E2BIG;
+ }
- if (source->size - source->filled < LINE_CHUNK) {
- // XXX: add check for maximum line length
+ if (source->passive_fd)
+ /* we have to wait for some data to come to us */
+ return -EWOULDBLOCK;
- if (!GREEDY_REALLOC(source->buf, source->size,
- source->filled + LINE_CHUNK))
- return log_oom();
- }
- assert(source->size - source->filled >= LINE_CHUNK);
-
- 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);
- return -errno;
- } else if (n == 0)
- return 0;
-
- c = memchr(source->buf + source->filled, '\n', n);
- source->filled += n;
-
- if (c == NULL)
- goto resize;
-
- docopy:
- *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);
+ if (source->size - source->filled < LINE_CHUNK &&
+ !realloc_buffer(source,
+ MIN(source->filled + LINE_CHUNK, ENTRY_SIZE_MAX)))
+ return log_oom();
+
+ assert(source->size - source->filled >= LINE_CHUNK ||
+ source->size == ENTRY_SIZE_MAX);
+
+ n = read(source->fd, source->buf + source->filled,
+ source->size - source->filled);
+ if (n < 0) {
+ if (errno != EAGAIN && errno != EWOULDBLOCK)
+ log_error_errno(errno, "read(%d, ..., %zd): %m", source->fd,
+ source->size - source->filled);
+ return -errno;
+ } else if (n == 0)
+ return 0;
+
+ source->filled += n;
}
- source->buf = newbuf;
- source->size = newsize;
- source->filled = remain;
+
+ *line = source->buf + source->offset;
+ *size = c + 1 - source->buf - source->offset;
+ source->offset += *size;
return 1;
}
assert(source);
assert(source->state != STATE_EOF);
- if (!GREEDY_REALLOC(source->buf, source->size,
- source->filled + size))
- return log_oom();
+ 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));
+ return -ENOMEM;
+ }
memcpy(source->buf + source->filled, data, size);
source->filled += 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->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,
+ log_error_errno(errno, "read(%d, ..., %zd): %m", source->fd,
source->size - source->filled);
return -errno;
} else if (n == 0)
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;
+ *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);
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;
}
static int get_data_newline(RemoteSource *source) {
int r;
- _cleanup_free_ char *data = NULL;
+ char *data;
assert(source);
assert(source->state == STATE_DATA_FINISH);
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
else
/* replace \n with = */
line[n-1] = '=';
- log_debug("Received: %.*s", (int) n, line);
+ log_trace("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;
}
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) {
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) {
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) {
}
}
-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 a full event from source@%p fd:%d (%s)",
+ source, source->fd, source->name);
if (!source->iovw.count) {
log_warning("Entry with no payload, skipping");
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;
}
~ianmdlvl / elogind.git / blobdiff