diff options
author | Zbigniew Jędrzejewski-Szmek <zbyszek@in.waw.pl> | 2016-11-05 22:40:54 -0400 |
---|---|---|
committer | Zbigniew Jędrzejewski-Szmek <zbyszek@in.waw.pl> | 2017-02-14 23:56:48 -0500 |
commit | b18453eda671277e44d4097e6f33b609d14a2d73 (patch) | |
tree | 42662b757c754301eeab668075a78cdecb7fda78 /src/journal-remote | |
parent | 988e89ee3b95c3ab2f47bc95c4eef0325e4636b7 (diff) |
Move export format parsing from src/journal-remote/ to src/basic/
No functional change.
Diffstat (limited to 'src/journal-remote')
-rw-r--r-- | src/journal-remote/journal-remote-parse.c | 431 | ||||
-rw-r--r-- | src/journal-remote/journal-remote-parse.h | 35 | ||||
-rw-r--r-- | src/journal-remote/journal-remote-write.c | 33 | ||||
-rw-r--r-- | src/journal-remote/journal-remote-write.h | 12 | ||||
-rw-r--r-- | src/journal-remote/journal-remote.c | 21 |
5 files changed, 26 insertions, 506 deletions
diff --git a/src/journal-remote/journal-remote-parse.c b/src/journal-remote/journal-remote-parse.c index 9ba9ee3fc0..79afe6604c 100644 --- a/src/journal-remote/journal-remote-parse.c +++ b/src/journal-remote/journal-remote-parse.c @@ -24,20 +24,11 @@ #include "parse-util.h" #include "string-util.h" -#define LINE_CHUNK 8*1024u - void source_free(RemoteSource *source) { if (!source) return; - 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); + journal_importer_cleanup(&source->importer); log_debug("Writer ref count %i", source->writer->n_ref); writer_unref(source->writer); @@ -65,442 +56,44 @@ RemoteSource* source_new(int fd, bool passive_fd, char *name, Writer *writer) { if (!source) return NULL; - source->fd = fd; - source->passive_fd = passive_fd; - source->name = name; + source->importer.fd = fd; + source->importer.passive_fd = passive_fd; + source->importer.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; - char *c = NULL; - - 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); - - for (;;) { - 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) { - log_error("Entry is bigger than %u bytes.", DATA_SIZE_MAX); - return -E2BIG; - } - - if (source->passive_fd) - /* we have to wait for some data to come to us */ - return -EAGAIN; - - /* We know that source->filled is at most DATA_SIZE_MAX, so if - we reallocate it, we'll increase the size at least a bit. */ - assert_cc(DATA_SIZE_MAX < ENTRY_SIZE_MAX); - if (source->size - source->filled < LINE_CHUNK && - !realloc_buffer(source, MIN(source->filled + LINE_CHUNK, ENTRY_SIZE_MAX))) - return log_oom(); - - assert(source->buf); - 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) - log_error_errno(errno, "read(%d, ..., %zu): %m", - source->fd, - source->size - source->filled); - return -errno; - } else if (n == 0) - return 0; - - source->filled += n; - } - - *line = source->buf + source->offset; - *size = c + 1 - source->buf - source->offset; - source->offset += *size; - - return 1; -} - -int push_data(RemoteSource *source, const char *data, size_t size) { - assert(source); - assert(source->state != STATE_EOF); - - 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; - - return 0; -} - -static int fill_fixed_size(RemoteSource *source, void **data, size_t size) { - - 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 - source->offset < size) { - int n; - - if (source->passive_fd) - /* we have to wait for some data to come to us */ - return -EAGAIN; - - 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) - log_error_errno(errno, "read(%d, ..., %zu): %m", source->fd, - source->size - source->filled); - return -errno; - } else if (n == 0) - return 0; - - source->filled += n; - } - - *data = source->buf + source->offset; - source->offset += size; - - return 1; -} - -static int get_data_size(RemoteSource *source) { - int r; - void *data; - - assert(source); - assert(source->state == STATE_DATA_START); - assert(source->data_size == 0); - - r = fill_fixed_size(source, &data, sizeof(uint64_t)); - if (r <= 0) - return r; - - source->data_size = le64toh( *(uint64_t *) data ); - if (source->data_size > DATA_SIZE_MAX) { - log_error("Stream declares field with size %zu > DATA_SIZE_MAX = %u", - source->data_size, DATA_SIZE_MAX); - return -EINVAL; - } - if (source->data_size == 0) - log_warning("Binary field with zero length"); - - return 1; -} - -static int get_data_data(RemoteSource *source, void **data) { - int r; - - assert(source); - assert(data); - assert(source->state == STATE_DATA); - - r = fill_fixed_size(source, data, source->data_size); - if (r <= 0) - return r; - - return 1; -} - -static int get_data_newline(RemoteSource *source) { - int r; - char *data; - - assert(source); - assert(source->state == STATE_DATA_FINISH); - - r = fill_fixed_size(source, (void**) &data, 1); - if (r <= 0) - return r; - - assert(data); - if (*data != '\n') { - log_error("expected newline, got '%c'", *data); - return -EINVAL; - } - - return 1; -} - -static int process_dunder(RemoteSource *source, char *line, size_t n) { - const char *timestamp; - int r; - - assert(line); - assert(n > 0); - assert(line[n-1] == '\n'); - - /* XXX: is it worth to support timestamps in extended format? - * We don't produce them, but who knows... */ - - timestamp = startswith(line, "__CURSOR="); - if (timestamp) - /* ignore __CURSOR */ - return 1; - - timestamp = startswith(line, "__REALTIME_TIMESTAMP="); - if (timestamp) { - long long unsigned x; - line[n-1] = '\0'; - r = safe_atollu(timestamp, &x); - if (r < 0) - log_warning("Failed to parse __REALTIME_TIMESTAMP: '%s'", timestamp); - else - source->ts.realtime = x; - return r < 0 ? r : 1; - } - - timestamp = startswith(line, "__MONOTONIC_TIMESTAMP="); - if (timestamp) { - long long unsigned x; - line[n-1] = '\0'; - r = safe_atollu(timestamp, &x); - if (r < 0) - log_warning("Failed to parse __MONOTONIC_TIMESTAMP: '%s'", timestamp); - else - source->ts.monotonic = x; - return r < 0 ? r : 1; - } - - timestamp = startswith(line, "__"); - if (timestamp) { - log_notice("Unknown dunder line %s", line); - return 1; - } - - /* no dunder */ - return 0; -} - -static int process_data(RemoteSource *source) { - int r; - - switch(source->state) { - case STATE_LINE: { - char *line, *sep; - size_t n = 0; - - assert(source->data_size == 0); - - r = get_line(source, &line, &n); - if (r < 0) - return r; - if (r == 0) { - source->state = STATE_EOF; - return r; - } - assert(n > 0); - assert(line[n-1] == '\n'); - - if (n == 1) { - log_trace("Received empty line, event is ready"); - return 1; - } - - r = process_dunder(source, line, n); - if (r != 0) - return r < 0 ? r : 0; - - /* MESSAGE=xxx\n - or - COREDUMP\n - LLLLLLLL0011223344...\n - */ - sep = memchr(line, '=', n); - if (sep) { - /* chomp newline */ - n--; - - r = iovw_put(&source->iovw, line, n); - if (r < 0) - return r; - } else { - /* replace \n with = */ - line[n-1] = '='; - - source->field_len = n; - source->state = STATE_DATA_START; - - /* we cannot put the field in iovec until we have all data */ - } - - log_trace("Received: %.*s (%s)", (int) n, line, sep ? "text" : "binary"); - - return 0; /* continue */ - } - - case STATE_DATA_START: - assert(source->data_size == 0); - - r = get_data_size(source); - // log_debug("get_data_size() -> %d", r); - if (r < 0) - return r; - if (r == 0) { - source->state = STATE_EOF; - return 0; - } - - source->state = source->data_size > 0 ? - STATE_DATA : STATE_DATA_FINISH; - - return 0; /* continue */ - - 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); - if (r < 0) - return r; - if (r == 0) { - source->state = STATE_EOF; - return 0; - } - - assert(data); - - 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; - - return 0; /* continue */ - } - - case STATE_DATA_FINISH: - r = get_data_newline(source); - // log_debug("get_data_newline() -> %d", r); - if (r < 0) - return r; - if (r == 0) { - source->state = STATE_EOF; - return 0; - } - - source->data_size = 0; - source->state = STATE_LINE; - - return 0; /* continue */ - default: - assert_not_reached("wtf?"); - } -} - int process_source(RemoteSource *source, bool compress, bool seal) { - size_t remain, target; int r; assert(source); assert(source->writer); - r = process_data(source); + r = journal_importer_process_data(&source->importer); if (r <= 0) return r; /* We have a full event */ log_trace("Received full event from source@%p fd:%d (%s)", - source, source->fd, source->name); + source, source->importer.fd, source->importer.name); - if (!source->iovw.count) { + if (source->importer.iovw.count == 0) { log_warning("Entry with no payload, skipping"); goto freeing; } - assert(source->iovw.iovec); - assert(source->iovw.count); + assert(source->importer.iovw.iovec); - r = writer_write(source->writer, &source->iovw, &source->ts, compress, seal); + r = writer_write(source->writer, &source->importer.iovw, &source->importer.ts, compress, seal); if (r < 0) log_error_errno(r, "Failed to write entry of %zu bytes: %m", - iovw_size(&source->iovw)); + iovw_size(&source->importer.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 && source->filled < 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; - } - } - + journal_importer_drop_iovw(&source->importer); return r; } diff --git a/src/journal-remote/journal-remote-parse.h b/src/journal-remote/journal-remote-parse.h index 1740a21f92..e3632528cf 100644 --- a/src/journal-remote/journal-remote-parse.h +++ b/src/journal-remote/journal-remote-parse.h @@ -21,34 +21,11 @@ #include "sd-event.h" +#include "journal-importer.h" #include "journal-remote-write.h" -typedef enum { - STATE_LINE = 0, /* waiting to read, or reading line */ - STATE_DATA_START, /* reading binary data header */ - STATE_DATA, /* reading binary data */ - STATE_DATA_FINISH, /* expecting newline */ - STATE_EOF, /* done */ -} source_state; - typedef struct RemoteSource { - char *name; - int fd; - bool passive_fd; - - char *buf; - size_t size; /* total size of the buffer */ - size_t offset; /* offset to the beginning of live data in the buffer */ - size_t scanned; /* number of bytes since the beginning of data without a newline */ - size_t filled; /* total number of bytes in the buffer */ - - size_t field_len; /* used for binary fields: the field name length */ - size_t data_size; /* and the size of the binary data chunk being processed */ - - struct iovec_wrapper iovw; - - source_state state; - dual_timestamp ts; + JournalImporter importer; Writer *writer; @@ -57,13 +34,5 @@ typedef struct RemoteSource { } RemoteSource; RemoteSource* source_new(int fd, bool passive_fd, char *name, Writer *writer); - -static inline size_t source_non_empty(RemoteSource *source) { - assert(source); - - return source->filled; -} - void source_free(RemoteSource *source); -int push_data(RemoteSource *source, const char *data, size_t size); int process_source(RemoteSource *source, bool compress, bool seal); diff --git a/src/journal-remote/journal-remote-write.c b/src/journal-remote/journal-remote-write.c index 8729372aa3..734cad333f 100644 --- a/src/journal-remote/journal-remote-write.c +++ b/src/journal-remote/journal-remote-write.c @@ -20,39 +20,6 @@ #include "alloc-util.h" #include "journal-remote.h" -int iovw_put(struct iovec_wrapper *iovw, void* data, size_t len) { - if (!GREEDY_REALLOC(iovw->iovec, iovw->size_bytes, iovw->count + 1)) - return log_oom(); - - iovw->iovec[iovw->count++] = (struct iovec) {data, len}; - return 0; -} - -void iovw_free_contents(struct iovec_wrapper *iovw) { - iovw->iovec = mfree(iovw->iovec); - iovw->size_bytes = iovw->count = 0; -} - -size_t iovw_size(struct iovec_wrapper *iovw) { - size_t n = 0, i; - - for (i = 0; i < iovw->count; i++) - n += iovw->iovec[i].iov_len; - - return n; -} - -void iovw_rebase(struct iovec_wrapper *iovw, char *old, char *new) { - size_t i; - - for (i = 0; i < iovw->count; i++) - iovw->iovec[i].iov_base = (char*) iovw->iovec[i].iov_base - old + new; -} - -/********************************************************************** - ********************************************************************** - **********************************************************************/ - static int do_rotate(JournalFile **f, bool compress, bool seal) { int r = journal_file_rotate(f, compress, seal, NULL); if (r < 0) { diff --git a/src/journal-remote/journal-remote-write.h b/src/journal-remote/journal-remote-write.h index 53ba45fc04..e04af54e55 100644 --- a/src/journal-remote/journal-remote-write.h +++ b/src/journal-remote/journal-remote-write.h @@ -20,20 +20,10 @@ ***/ #include "journal-file.h" +#include "journal-importer.h" typedef struct RemoteServer RemoteServer; -struct iovec_wrapper { - struct iovec *iovec; - size_t size_bytes; - size_t count; -}; - -int iovw_put(struct iovec_wrapper *iovw, void* data, size_t len); -void iovw_free_contents(struct iovec_wrapper *iovw); -size_t iovw_size(struct iovec_wrapper *iovw); -void iovw_rebase(struct iovec_wrapper *iovw, char *old, char *new); - typedef struct Writer { JournalFile *journal; JournalMetrics metrics; diff --git a/src/journal-remote/journal-remote.c b/src/journal-remote/journal-remote.c index d0d8d936e3..202a5a3f97 100644 --- a/src/journal-remote/journal-remote.c +++ b/src/journal-remote/journal-remote.c @@ -512,7 +512,8 @@ static int process_http_upload( if (*upload_data_size) { log_trace("Received %zu bytes", *upload_data_size); - r = push_data(source, upload_data, *upload_data_size); + r = journal_importer_push_data(&source->importer, + upload_data, *upload_data_size); if (r < 0) return mhd_respond_oom(connection); @@ -542,7 +543,7 @@ static int process_http_upload( /* The upload is finished */ - remaining = source_non_empty(source); + remaining = journal_importer_bytes_remaining(&source->importer); if (remaining > 0) { log_warning("Premature EOF byte. %zu bytes lost.", remaining); return mhd_respondf(connection, @@ -1036,19 +1037,19 @@ static int handle_raw_source(sd_event_source *event, assert(fd >= 0 && fd < (ssize_t) s->sources_size); source = s->sources[fd]; - assert(source->fd == fd); + assert(source->importer.fd == fd); r = process_source(source, arg_compress, arg_seal); - if (source->state == STATE_EOF) { + if (journal_importer_eof(&source->importer)) { size_t remaining; - log_debug("EOF reached with source fd:%d (%s)", - source->fd, source->name); + log_debug("EOF reached with source %s (fd=%d)", + source->importer.name, source->importer.fd); - remaining = source_non_empty(source); + remaining = journal_importer_bytes_remaining(&source->importer); if (remaining > 0) log_notice("Premature EOF. %zu bytes lost.", remaining); - remove_source(s, source->fd); + remove_source(s, source->importer.fd); log_debug("%zu active sources remaining", s->active); return 0; } else if (r == -E2BIG) { @@ -1072,7 +1073,7 @@ static int dispatch_raw_source_until_block(sd_event_source *event, /* Make sure event stays around even if source is destroyed */ sd_event_source_ref(event); - r = handle_raw_source(event, source->fd, EPOLLIN, server); + r = handle_raw_source(event, source->importer.fd, EPOLLIN, server); if (r != 1) /* No more data for now */ sd_event_source_set_enabled(event, SD_EVENT_OFF); @@ -1105,7 +1106,7 @@ static int dispatch_blocking_source_event(sd_event_source *event, void *userdata) { RemoteSource *source = userdata; - return handle_raw_source(event, source->fd, EPOLLIN, server); + return handle_raw_source(event, source->importer.fd, EPOLLIN, server); } static int accept_connection(const char* type, int fd, |