#include <sys/statvfs.h>
#include <fcntl.h>
#include <stddef.h>
-
-#ifdef HAVE_XATTR
-#include <attr/xattr.h>
-#endif
+#include <sys/xattr.h>
#include "journal-def.h"
#include "journal-file.h"
/* How many entries to keep in the entry array chain cache at max */
#define CHAIN_CACHE_MAX 20
-int journal_file_set_online(JournalFile *f) {
+/* How much to increase the journal file size at once each time we allocate something new. */
+#define FILE_SIZE_INCREASE (8ULL*1024ULL*1024ULL) /* 8MB */
+
+static int journal_file_set_online(JournalFile *f) {
assert(f);
if (!f->writable)
if (f->header)
munmap(f->header, PAGE_ALIGN(sizeof(Header)));
- if (f->fd >= 0)
- close_nointr_nofail(f->fd);
-
+ safe_close(f->fd);
free(f->path);
if (f->mmap)
hashmap_free_free(f->chain_cache);
-#ifdef HAVE_XZ
+#if defined(HAVE_XZ) || defined(HAVE_LZ4)
free(f->compress_buffer);
#endif
}
static int journal_file_init_header(JournalFile *f, JournalFile *template) {
- Header h;
+ Header h = {};
ssize_t k;
int r;
assert(f);
- zero(h);
memcpy(h.signature, HEADER_SIGNATURE, 8);
h.header_size = htole64(ALIGN64(sizeof(h)));
- h.incompatible_flags =
- htole32(f->compress ? HEADER_INCOMPATIBLE_COMPRESSED : 0);
+ h.incompatible_flags |= htole32(
+ f->compress_xz * HEADER_INCOMPATIBLE_COMPRESSED_XZ |
+ f->compress_lz4 * HEADER_INCOMPATIBLE_COMPRESSED_LZ4);
- h.compatible_flags =
- htole32(f->seal ? HEADER_COMPATIBLE_SEALED : 0);
+ h.compatible_flags = htole32(
+ f->seal * HEADER_COMPATIBLE_SEALED);
r = sd_id128_randomize(&h.file_id);
if (r < 0)
journal_file_set_online(f);
/* Sync the online state to disk */
- msync(f->header, PAGE_ALIGN(sizeof(Header)), MS_SYNC);
- fdatasync(f->fd);
+ fsync(f->fd);
return 0;
}
static int journal_file_verify_header(JournalFile *f) {
+ uint32_t flags;
+
assert(f);
if (memcmp(f->header->signature, HEADER_SIGNATURE, 8))
/* In both read and write mode we refuse to open files with
* incompatible flags we don't know */
-#ifdef HAVE_XZ
- if ((le32toh(f->header->incompatible_flags) & ~HEADER_INCOMPATIBLE_COMPRESSED) != 0)
+ flags = le32toh(f->header->incompatible_flags);
+ if (flags & ~HEADER_INCOMPATIBLE_SUPPORTED) {
+ if (flags & ~HEADER_INCOMPATIBLE_ANY)
+ log_debug("Journal file %s has unknown incompatible flags %"PRIx32,
+ f->path, flags & ~HEADER_INCOMPATIBLE_ANY);
+ flags = (flags & HEADER_INCOMPATIBLE_ANY) & ~HEADER_INCOMPATIBLE_SUPPORTED;
+ if (flags)
+ log_debug("Journal file %s uses incompatible flags %"PRIx32
+ " disabled at compilation time.", f->path, flags);
return -EPROTONOSUPPORT;
-#else
- if (f->header->incompatible_flags != 0)
- return -EPROTONOSUPPORT;
-#endif
+ }
/* When open for writing we refuse to open files with
* compatible flags, too */
- if (f->writable) {
-#ifdef HAVE_GCRYPT
- if ((le32toh(f->header->compatible_flags) & ~HEADER_COMPATIBLE_SEALED) != 0)
- return -EPROTONOSUPPORT;
-#else
- if (f->header->compatible_flags != 0)
- return -EPROTONOSUPPORT;
-#endif
+ flags = le32toh(f->header->compatible_flags);
+ if (f->writable && (flags & ~HEADER_COMPATIBLE_SUPPORTED)) {
+ if (flags & ~HEADER_COMPATIBLE_ANY)
+ log_debug("Journal file %s has unknown compatible flags %"PRIx32,
+ f->path, flags & ~HEADER_COMPATIBLE_ANY);
+ flags = (flags & HEADER_COMPATIBLE_ANY) & ~HEADER_COMPATIBLE_SUPPORTED;
+ if (flags)
+ log_debug("Journal file %s uses compatible flags %"PRIx32
+ " disabled at compilation time.", f->path, flags);
+ return -EPROTONOSUPPORT;
}
if (f->header->state >= _STATE_MAX)
!VALID64(le64toh(f->header->entry_array_offset)))
return -ENODATA;
- if (le64toh(f->header->data_hash_table_offset) < le64toh(f->header->header_size) ||
- le64toh(f->header->field_hash_table_offset) < le64toh(f->header->header_size) ||
- le64toh(f->header->tail_object_offset) < le64toh(f->header->header_size) ||
- le64toh(f->header->entry_array_offset) < le64toh(f->header->header_size))
- return -ENODATA;
-
if (f->writable) {
uint8_t state;
sd_id128_t machine_id;
}
}
- f->compress = JOURNAL_HEADER_COMPRESSED(f->header);
+ f->compress_xz = JOURNAL_HEADER_COMPRESSED_XZ(f->header);
+ f->compress_lz4 = JOURNAL_HEADER_COMPRESSED_LZ4(f->header);
f->seal = JOURNAL_HEADER_SEALED(f->header);
if (new_size <= old_size)
return 0;
- if (f->metrics.max_size > 0 &&
- new_size > f->metrics.max_size)
+ if (f->metrics.max_size > 0 && new_size > f->metrics.max_size)
return -E2BIG;
- if (new_size > f->metrics.min_size &&
- f->metrics.keep_free > 0) {
+ if (new_size > f->metrics.min_size && f->metrics.keep_free > 0) {
struct statvfs svfs;
if (fstatvfs(f->fd, &svfs) >= 0) {
}
}
+ /* Increase by larger blocks at once */
+ new_size = ((new_size+FILE_SIZE_INCREASE-1) / FILE_SIZE_INCREASE) * FILE_SIZE_INCREASE;
+ if (f->metrics.max_size > 0 && new_size > f->metrics.max_size)
+ new_size = f->metrics.max_size;
+
/* Note that the glibc fallocate() fallback is very
inefficient, hence we try to minimize the allocation area
as we can. */
static uint64_t minimum_header_size(Object *o) {
- static uint64_t table[] = {
+ static const uint64_t table[] = {
[OBJECT_DATA] = sizeof(DataObject),
[OBJECT_FIELD] = sizeof(FieldObject),
[OBJECT_ENTRY] = sizeof(EntryObject),
void *t;
Object *o;
uint64_t s;
- unsigned context;
assert(f);
assert(ret);
if (!VALID64(offset))
return -EFAULT;
- /* One context for each type, plus one catch-all for the rest */
- context = type > 0 && type < _OBJECT_TYPE_MAX ? type : 0;
- r = journal_file_move_to(f, context, false, offset, sizeof(ObjectHeader), &t);
+ r = journal_file_move_to(f, type_to_context(type), false, offset, sizeof(ObjectHeader), &t);
if (r < 0)
return r;
if (r < 0)
return r;
- memset(o->hash_table.items, 0, s);
+ memzero(o->hash_table.items, s);
f->header->data_hash_table_offset = htole64(p + offsetof(Object, hash_table.items));
f->header->data_hash_table_size = htole64(s);
if (r < 0)
return r;
- memset(o->hash_table.items, 0, s);
+ memzero(o->hash_table.items, s);
f->header->field_hash_table_offset = htole64(p + offsetof(Object, hash_table.items));
f->header->field_hash_table_size = htole64(s);
if (le64toh(o->data.hash) != hash)
goto next;
- if (o->object.flags & OBJECT_COMPRESSED) {
-#ifdef HAVE_XZ
+ if (o->object.flags & OBJECT_COMPRESSION_MASK) {
+#if defined(HAVE_XZ) || defined(HAVE_LZ4)
uint64_t l, rsize;
l = le64toh(o->object.size);
l -= offsetof(Object, data.payload);
- if (!uncompress_blob(o->data.payload, l, &f->compress_buffer, &f->compress_buffer_size, &rsize, 0))
- return -EBADMSG;
+ r = decompress_blob(o->object.flags & OBJECT_COMPRESSION_MASK,
+ o->data.payload, l, &f->compress_buffer, &f->compress_buffer_size, &rsize, 0);
+ if (r < 0)
+ return r;
if (rsize == size &&
memcmp(f->compress_buffer, data, size) == 0) {
#else
return -EPROTONOSUPPORT;
#endif
-
} else if (le64toh(o->object.size) == osize &&
memcmp(o->data.payload, data, size) == 0) {
uint64_t hash, p;
uint64_t osize;
Object *o;
- int r;
- bool compressed = false;
+ int r, compression = 0;
const void *eq;
assert(f);
o->data.hash = htole64(hash);
-#ifdef HAVE_XZ
- if (f->compress &&
+#if defined(HAVE_XZ) || defined(HAVE_LZ4)
+ if (f->compress_xz &&
size >= COMPRESSION_SIZE_THRESHOLD) {
uint64_t rsize;
- compressed = compress_blob(data, size, o->data.payload, &rsize);
+ compression = compress_blob(data, size, o->data.payload, &rsize);
- if (compressed) {
+ if (compression) {
o->object.size = htole64(offsetof(Object, data.payload) + rsize);
- o->object.flags |= OBJECT_COMPRESSED;
+ o->object.flags |= compression;
- log_debug("Compressed data object %"PRIu64" -> %"PRIu64, size, rsize);
+ log_debug("Compressed data object %"PRIu64" -> %"PRIu64" using %s",
+ size, rsize, object_compressed_to_string(compression));
}
}
#endif
- if (!compressed && size > 0)
+ if (!compression && size > 0)
memcpy(o->data.payload, data, size);
r = journal_file_link_data(f, o, p, hash);
if (r < 0)
return r;
- eq = memchr(data, '=', size);
+ if (!data)
+ eq = NULL;
+ else
+ eq = memchr(data, '=', size);
if (eq && eq > data) {
+ Object *fo = NULL;
uint64_t fp;
- Object *fo;
/* Create field object ... */
r = journal_file_append_field(f, data, (uint8_t*) eq - (uint8_t*) data, &fo, &fp);
/* Order by the position on disk, in order to improve seek
* times for rotating media. */
- qsort(items, n_iovec, sizeof(EntryItem), entry_item_cmp);
+ qsort_safe(items, n_iovec, sizeof(EntryItem), entry_item_cmp);
r = journal_file_append_entry_internal(f, ts, xor_hash, items, n_iovec, seqnum, ret, offset);
}
typedef struct ChainCacheItem {
- uint64_t first; /* the array at the begin of the chain */
+ uint64_t first; /* the array at the beginning of the chain */
uint64_t array; /* the cached array */
uint64_t begin; /* the first item in the cached array */
uint64_t total; /* the total number of items in all arrays before this one in the chain */
+ uint64_t last_index; /* the last index we looked at, to optimize locality when bisecting */
} ChainCacheItem;
static void chain_cache_put(
uint64_t first,
uint64_t array,
uint64_t begin,
- uint64_t total) {
+ uint64_t total,
+ uint64_t last_index) {
if (!ci) {
/* If the chain item to cache for this chain is the
ci->array = array;
ci->begin = begin;
ci->total = total;
+ ci->last_index = last_index;
}
-static int generic_array_get(JournalFile *f,
- uint64_t first,
- uint64_t i,
- Object **ret, uint64_t *offset) {
+static int generic_array_get(
+ JournalFile *f,
+ uint64_t first,
+ uint64_t i,
+ Object **ret, uint64_t *offset) {
Object *o;
uint64_t p = 0, a, t = 0;
found:
/* Let's cache this item for the next invocation */
- chain_cache_put(f->chain_cache, ci, first, a, o->entry_array.items[0], t);
+ chain_cache_put(f->chain_cache, ci, first, a, le64toh(o->entry_array.items[0]), t, i);
r = journal_file_move_to_object(f, OBJECT_ENTRY, p, &o);
if (r < 0)
return 1;
}
-static int generic_array_get_plus_one(JournalFile *f,
- uint64_t extra,
- uint64_t first,
- uint64_t i,
- Object **ret, uint64_t *offset) {
+static int generic_array_get_plus_one(
+ JournalFile *f,
+ uint64_t extra,
+ uint64_t first,
+ uint64_t i,
+ Object **ret, uint64_t *offset) {
Object *o;
TEST_RIGHT
};
-static int generic_array_bisect(JournalFile *f,
- uint64_t first,
- uint64_t n,
- uint64_t needle,
- int (*test_object)(JournalFile *f, uint64_t p, uint64_t needle),
- direction_t direction,
- Object **ret,
- uint64_t *offset,
- uint64_t *idx) {
-
- uint64_t a, p, t = 0, i = 0, last_p = 0;
+static int generic_array_bisect(
+ JournalFile *f,
+ uint64_t first,
+ uint64_t n,
+ uint64_t needle,
+ int (*test_object)(JournalFile *f, uint64_t p, uint64_t needle),
+ direction_t direction,
+ Object **ret,
+ uint64_t *offset,
+ uint64_t *idx) {
+
+ uint64_t a, p, t = 0, i = 0, last_p = 0, last_index = (uint64_t) -1;
bool subtract_one = false;
Object *o, *array = NULL;
int r;
return r;
if (r == TEST_LEFT) {
- /* OK, what we are looking for is right of th
+ /* OK, what we are looking for is right of the
* begin of this EntryArray, so let's jump
* straight to previously cached array in the
* chain */
a = ci->array;
n -= ci->total;
t = ci->total;
+ last_index = ci->last_index;
}
}
if (r == TEST_RIGHT) {
left = 0;
right -= 1;
+
+ if (last_index != (uint64_t) -1) {
+ assert(last_index <= right);
+
+ /* If we cached the last index we
+ * looked at, let's try to not to jump
+ * too wildly around and see if we can
+ * limit the range to look at early to
+ * the immediate neighbors of the last
+ * index we looked at. */
+
+ if (last_index > 0) {
+ uint64_t x = last_index - 1;
+
+ p = le64toh(array->entry_array.items[x]);
+ if (p <= 0)
+ return -EBADMSG;
+
+ r = test_object(f, p, needle);
+ if (r < 0)
+ return r;
+
+ if (r == TEST_FOUND)
+ r = direction == DIRECTION_DOWN ? TEST_RIGHT : TEST_LEFT;
+
+ if (r == TEST_RIGHT)
+ right = x;
+ else
+ left = x + 1;
+ }
+
+ if (last_index < right) {
+ uint64_t y = last_index + 1;
+
+ p = le64toh(array->entry_array.items[y]);
+ if (p <= 0)
+ return -EBADMSG;
+
+ r = test_object(f, p, needle);
+ if (r < 0)
+ return r;
+
+ if (r == TEST_FOUND)
+ r = direction == DIRECTION_DOWN ? TEST_RIGHT : TEST_LEFT;
+
+ if (r == TEST_RIGHT)
+ right = y;
+ else
+ left = y + 1;
+ }
+ }
+
for (;;) {
if (left == right) {
if (direction == DIRECTION_UP)
}
assert(left < right);
-
i = (left + right) / 2;
+
p = le64toh(array->entry_array.items[i]);
if (p <= 0)
return -EBADMSG;
n -= k;
t += k;
+ last_index = (uint64_t) -1;
a = le64toh(array->entry_array.next_entry_array_offset);
}
return 0;
/* Let's cache this item for the next invocation */
- chain_cache_put(f->chain_cache, ci, first, a, array->entry_array.items[0], t);
+ chain_cache_put(f->chain_cache, ci, first, a, le64toh(array->entry_array.items[0]), t, subtract_one ? (i > 0 ? i-1 : (uint64_t) -1) : i);
if (subtract_one && i == 0)
p = last_p;
return 1;
}
-static int generic_array_bisect_plus_one(JournalFile *f,
- uint64_t extra,
- uint64_t first,
- uint64_t n,
- uint64_t needle,
- int (*test_object)(JournalFile *f, uint64_t p, uint64_t needle),
- direction_t direction,
- Object **ret,
- uint64_t *offset,
- uint64_t *idx) {
+
+static int generic_array_bisect_plus_one(
+ JournalFile *f,
+ uint64_t extra,
+ uint64_t first,
+ uint64_t n,
+ uint64_t needle,
+ int (*test_object)(JournalFile *f, uint64_t p, uint64_t needle),
+ direction_t direction,
+ Object **ret,
+ uint64_t *offset,
+ uint64_t *idx) {
int r;
bool step_back = false;
direction_t direction,
Object **ret, uint64_t *offset) {
- uint64_t i, n;
+ uint64_t i, n, ofs;
int r;
assert(f);
}
/* And jump to it */
- return generic_array_get(f,
- le64toh(f->header->entry_array_offset),
- i,
- ret, offset);
+ r = generic_array_get(f,
+ le64toh(f->header->entry_array_offset),
+ i,
+ ret, &ofs);
+ if (r <= 0)
+ return r;
+
+ if (p > 0 &&
+ (direction == DIRECTION_DOWN ? ofs <= p : ofs >= p)) {
+ log_debug("%s: entry array corrupted at entry %"PRIu64,
+ f->path, i);
+ return -EBADMSG;
+ }
+
+ if (offset)
+ *offset = ofs;
+
+ return 1;
}
int journal_file_skip_entry(
z = q;
}
-
- return 0;
}
int journal_file_move_to_entry_by_seqnum_for_data(
break;
}
- if (o->object.flags & OBJECT_COMPRESSED)
- printf("Flags: COMPRESSED\n");
+ if (o->object.flags & OBJECT_COMPRESSION_MASK)
+ printf("Flags: %s\n",
+ object_compressed_to_string(o->object.flags & OBJECT_COMPRESSION_MASK));
if (p == le64toh(f->header->tail_object_offset))
p = 0;
"Sequential Number ID: %s\n"
"State: %s\n"
"Compatible Flags:%s%s\n"
- "Incompatible Flags:%s%s\n"
+ "Incompatible Flags:%s%s%s\n"
"Header size: %"PRIu64"\n"
"Arena size: %"PRIu64"\n"
"Data Hash Table Size: %"PRIu64"\n"
f->header->state == STATE_ONLINE ? "ONLINE" :
f->header->state == STATE_ARCHIVED ? "ARCHIVED" : "UNKNOWN",
JOURNAL_HEADER_SEALED(f->header) ? " SEALED" : "",
- (le32toh(f->header->compatible_flags) & ~HEADER_COMPATIBLE_SEALED) ? " ???" : "",
- JOURNAL_HEADER_COMPRESSED(f->header) ? " COMPRESSED" : "",
- (le32toh(f->header->incompatible_flags) & ~HEADER_INCOMPATIBLE_COMPRESSED) ? " ???" : "",
+ (le32toh(f->header->compatible_flags) & ~HEADER_COMPATIBLE_ANY) ? " ???" : "",
+ JOURNAL_HEADER_COMPRESSED_XZ(f->header) ? " COMPRESSED-XZ" : "",
+ JOURNAL_HEADER_COMPRESSED_LZ4(f->header) ? " COMPRESSED-LZ4" : "",
+ (le32toh(f->header->incompatible_flags) & ~HEADER_INCOMPATIBLE_ANY) ? " ???" : "",
le64toh(f->header->header_size),
le64toh(f->header->arena_size),
le64toh(f->header->data_hash_table_size) / sizeof(HashItem),
f->flags = flags;
f->prot = prot_from_flags(flags);
f->writable = (flags & O_ACCMODE) != O_RDONLY;
-#ifdef HAVE_XZ
- f->compress = compress;
+#if defined(HAVE_LZ4)
+ f->compress_lz4 = compress;
+#elif defined(HAVE_XZ)
+ f->compress_xz = compress;
#endif
#ifdef HAVE_GCRYPT
f->seal = seal;
}
if (f->last_stat.st_size == 0 && f->writable) {
-#ifdef HAVE_XATTR
uint64_t crtime;
/* Let's attach the creation time to the journal file,
crtime = htole64((uint64_t) now(CLOCK_REALTIME));
fsetxattr(f->fd, "user.crtime_usec", &crtime, sizeof(crtime), XATTR_CREATE);
-#endif
#ifdef HAVE_GCRYPT
/* Try to load the FSPRG state, and if we can't, then
/* The file is corrupted. Rotate it away and try it again (but only once) */
l = strlen(fname);
- if (asprintf(&p, "%.*s@%016llx-%016llx.journal~",
+ if (asprintf(&p, "%.*s@%016llx-%016" PRIx64 ".journal~",
(int) l - 8, fname,
(unsigned long long) now(CLOCK_REALTIME),
- random_ull()) < 0)
+ random_u64()) < 0)
return -ENOMEM;
r = rename(fname, p);
ts.monotonic = le64toh(o->entry.monotonic);
ts.realtime = le64toh(o->entry.realtime);
- if (to->tail_entry_monotonic_valid &&
- ts.monotonic < le64toh(to->header->tail_entry_monotonic))
- return -EINVAL;
-
n = journal_file_entry_n_items(o);
- items = alloca(sizeof(EntryItem) * n);
+ /* alloca() can't take 0, hence let's allocate at least one */
+ items = alloca(sizeof(EntryItem) * MAX(1u, n));
for (i = 0; i < n; i++) {
uint64_t l, h;
if ((uint64_t) t != l)
return -E2BIG;
- if (o->object.flags & OBJECT_COMPRESSED) {
-#ifdef HAVE_XZ
+ if (o->object.flags & OBJECT_COMPRESSION_MASK) {
+#if defined(HAVE_XZ) || defined(HAVE_LZ4)
uint64_t rsize;
- if (!uncompress_blob(o->data.payload, l, &from->compress_buffer, &from->compress_buffer_size, &rsize, 0))
- return -EBADMSG;
+ r = decompress_blob(o->object.flags & OBJECT_COMPRESSION_MASK,
+ o->data.payload, l, &from->compress_buffer, &from->compress_buffer_size, &rsize, 0);
+ if (r < 0)
+ return r;
data = from->compress_buffer;
l = rsize;