diff options
Diffstat (limited to 'prototypes/connections')
| -rw-r--r-- | prototypes/connections/store-tree.c | 392 | ||||
| -rw-r--r-- | prototypes/connections/test-tree-store.c | 173 |
2 files changed, 450 insertions, 115 deletions
diff --git a/prototypes/connections/store-tree.c b/prototypes/connections/store-tree.c index 3163265..a5d24c2 100644 --- a/prototypes/connections/store-tree.c +++ b/prototypes/connections/store-tree.c @@ -3,8 +3,9 @@ * path it changes, so inserting or deleting in a large document costs O(log n) * nodes instead of a whole-document copy. Unchanged extents are shared by old * and new versions. The document catalog is the same structure holding fixed - * records sorted by key. Two alternating head slots publish a commit only after - * its nodes are durable; readers take no lock and map one immutable prefix. */ + * records sorted by key. A commit appends its nodes and writes a head naming + * them and their digest into the older of two slots, then syncs once; readers + * take no lock and map one immutable prefix. */ #define _POSIX_C_SOURCE 200809L #ifdef __APPLE__ #define _DARWIN_C_SOURCE @@ -23,9 +24,13 @@ #include <unistd.h> enum { FREE = 0xffff }; -enum { SLOT = 64, DATA = 2 * SLOT, LEAF_MAX = 4096, LEAF_MIN = 1024, +/* Each head slot has a block of its own: a torn write of one never touches + * the other. */ +enum { SLOT = 128, BLOCK = 4096, DATA = 2 * BLOCK, LEAF_MAX = 4096, LEAF_MIN = 1024, FAN_MAX = 32, FAN_MIN = 8, RECORD = 96, KEY = 64, HEIGHT_MAX = 12, BUFFER = 65536 }; +/* Written zeros kept after the published end (see prepare_tail). */ +enum { TAIL_MIN = 65536, TAIL_STEP = 262144 }; #ifdef PURE_STORE_TEST enum { COMPACT_SLACK = 65536 }; #else @@ -34,7 +39,7 @@ enum { COMPACT_SLACK = 1048576 }; #define LEAF_BIT (UINT64_C(1) << 63) #define MEMORY_BIT (UINT64_C(1) << 62) #define LOCATION (MEMORY_BIT - 1) -static const unsigned char head_magic[8] = {'P','U','R','E','T','R','E','1'}; +static const unsigned char head_magic[8] = {'P','U','R','E','T','R','E','2'}; static const unsigned char branch_magic[4] = {'P','T','B','1'}; static const char tree_name[] = "pure.tree"; /* Threads of one process take turns as the writer: the file lock arbitrates @@ -64,12 +69,18 @@ typedef struct { size_t count, capacity, free_head, free_count; /* free_head is slot + 1 */ size_t bytes; /* payload held by arena nodes, including unreachable ones */ } Tree; -typedef struct { uint64_t seq, end, catalog, catalog_bytes, live; } Head; +/* A commit's bytes are start..end (none for a file synced whole before it was + * published); check is their digest. */ +typedef struct { uint64_t seq, end, catalog, catalog_bytes, live, start, check; } Head; +/* The xxHash64 construction without a seed: four multiply-rotate lanes over + * 32-byte stripes, several GB/s. An all-zero Digest is an empty one. */ +typedef struct { uint64_t lane[4], length; unsigned char held[32]; size_t count; } Digest; typedef struct { int fd; uint64_t at; /* file offset of the next byte */ size_t used; bool failed; + Digest digest; /* of every byte written */ unsigned char buffer[BUFFER]; } Writer; typedef struct { PureSpan part[3]; size_t index, offset; } Source; @@ -111,19 +122,100 @@ static void encode64(unsigned char *bytes, uint64_t value) { for (unsigned i = 0; i < 8; ++i) bytes[i] = (unsigned char)(value >> (8 * i)); } -/* Detects a torn head slot; it is not a defence against a hostile file. */ -static uint64_t checksum(const unsigned char *bytes, size_t length) { - uint64_t hash = UINT64_C(14695981039346656037); - for (size_t i = 0; i < length; ++i) hash = (hash ^ bytes[i]) * UINT64_C(1099511628211); - return hash; +static uint64_t load64(const unsigned char *bytes) { + uint64_t value; + memcpy(&value, bytes, 8); +#if defined(__BYTE_ORDER__) && __BYTE_ORDER__ == __ORDER_BIG_ENDIAN__ + value = __builtin_bswap64(value); +#endif + return value; +} + +/* Digests find torn and missing writes; they are not a defence against a + * hostile file. */ +static const uint64_t P1 = UINT64_C(0x9E3779B185EBCA87), P2 = UINT64_C(0xC2B2AE3D27D4EB4F), + P3 = UINT64_C(0x165667B19E3779F9), P4 = UINT64_C(0x85EBCA77C2B2AE63), + P5 = UINT64_C(0x27D4EB2F165667C5); + +static uint64_t rotate(uint64_t value, unsigned bits) { return value << bits | value >> (64 - bits); } +static uint64_t mix(uint64_t lane, uint64_t word) { return rotate(lane + word * P2, 31) * P1; } + +static void stripes(Digest *digest, const unsigned char *bytes, size_t count) { + uint64_t a = digest->lane[0], b = digest->lane[1], c = digest->lane[2], d = digest->lane[3]; + for (; count; --count, bytes += 32) { + a = mix(a, load64(bytes)); + b = mix(b, load64(bytes + 8)); + c = mix(c, load64(bytes + 16)); + d = mix(d, load64(bytes + 24)); + } + digest->lane[0] = a, digest->lane[1] = b, digest->lane[2] = c, digest->lane[3] = d; +} + +static void digest_add(Digest *digest, const unsigned char *bytes, size_t length) { + digest->length += length; + if (digest->count) { + size_t take = 32 - digest->count < length ? 32 - digest->count : length; + memcpy(digest->held + digest->count, bytes, take); + digest->count += take; + bytes += take; + length -= take; + if (digest->count < 32) return; + stripes(digest, digest->held, 1); + digest->count = 0; + } + stripes(digest, bytes, length / 32); + digest->count = length % 32; + if (digest->count) memcpy(digest->held, bytes + length - digest->count, digest->count); } +static uint64_t digest_end(const Digest *digest) { + const uint64_t *lane = digest->lane; + uint64_t hash = rotate(lane[0], 1) + rotate(lane[1], 7) + rotate(lane[2], 12) + rotate(lane[3], 18); + for (unsigned i = 0; i < 4; ++i) hash = (hash ^ mix(0, lane[i])) * P1 + P4; + hash += digest->length; + size_t i = 0; + for (; i + 8 <= digest->count; i += 8) hash = rotate(hash ^ mix(0, load64(digest->held + i)), 27) * P1 + P4; + for (; i < digest->count; ++i) hash = rotate(hash ^ digest->held[i] * P5, 11) * P1; + hash ^= hash >> 33; + hash *= P2; + hash ^= hash >> 29; + hash *= P3; + return hash ^ hash >> 32; +} + +static uint64_t digest_of(const unsigned char *bytes, size_t length) { + Digest digest = {0}; + digest_add(&digest, bytes, length); + return digest_end(&digest); +} + +#ifdef PURE_STORE_TEST +static unsigned long syncs; +unsigned long pure_tree_test_syncs(void) { return __atomic_load_n(&syncs, __ATOMIC_RELAXED); } +#define COUNT_SYNC() __atomic_add_fetch(&syncs, 1, __ATOMIC_RELAXED) +#else +#define COUNT_SYNC() ((void)0) +#endif + static bool sync_fd(int fd) { int rc; + COUNT_SYNC(); do { rc = fsync(fd); } while (rc && errno == EINTR); return rc == 0; } +/* Data and the size needed to read it back, without timestamps. */ +static bool sync_data(int fd) { +#ifdef __APPLE__ + return sync_fd(fd); +#else + int rc; + COUNT_SYNC(); + do { rc = fdatasync(fd); } while (rc && errno == EINTR); + return rc == 0; +#endif +} + static bool write_at(int fd, const unsigned char *bytes, size_t length, uint64_t offset) { while (length) { ssize_t n = pwrite(fd, bytes, length, (off_t)offset); @@ -156,35 +248,130 @@ static void encode_head(unsigned char out[SLOT], const Head *head) { encode64(out + 24, head->catalog); encode64(out + 32, head->catalog_bytes); encode64(out + 40, head->live); - encode64(out + 56, checksum(out, 56)); + encode64(out + 48, head->start); + encode64(out + 56, head->check); + encode64(out + SLOT - 8, digest_of(out, SLOT - 8)); } static bool decode_head(const unsigned char in[SLOT], Head *head) { - if (memcmp(in, head_magic, sizeof(head_magic)) || decode64(in + 56) != checksum(in, 56)) + if (memcmp(in, head_magic, sizeof(head_magic)) || decode64(in + SLOT - 8) != digest_of(in, SLOT - 8)) return false; - *head = (Head){decode64(in + 8), decode64(in + 16), decode64(in + 24), - decode64(in + 32), decode64(in + 40)}; - return head->seq && head->end >= DATA && head->end <= (uint64_t)INT64_MAX && - head->end <= SIZE_MAX && head->catalog_bytes % RECORD == 0 && + *head = (Head){decode64(in + 8), decode64(in + 16), decode64(in + 24), decode64(in + 32), + decode64(in + 40), decode64(in + 48), decode64(in + 56)}; + return head->seq && head->end <= (uint64_t)INT64_MAX && head->end <= SIZE_MAX && + head->start >= DATA && head->start <= head->end && head->catalog_bytes % RECORD == 0 && !head->catalog == !head->catalog_bytes; } -/* The newer valid slot wins; a torn newer slot leaves the previous commit. */ -static bool read_head(int fd, Head *out) { - unsigned char slots[2 * SLOT]; - Head a, b; - if (!read_at(fd, slots, sizeof(slots), 0)) return false; - bool first = decode_head(slots, &a), second = decode_head(slots + SLOT, &b); - if (!first && !second) return false; - *out = first && (!second || a.seq > b.seq) ? a : b; - return true; +/* ---- Trusted heads -------------------------------------------------------- + * One sync makes a commit's bytes and its head durable together, so a crash + * during it can leave the head on the device without all of the bytes: a + * head counts only once its digest matches them. The older slot needs no + * such luck. A writer builds only on a durable head, so the newer head was + * written after the older one's commit was complete on the device. + * The process remembers the heads it published or checked, so it reads a + * commit's bytes at most once, and which of them it made durable. */ +enum { UNKNOWN, BROKEN, CHECKED, DURABLE, PUBLISHING }; +enum { KNOWN = 32 }; +typedef struct { dev_t device; ino_t inode; uint64_t seq, check; int state; } Known; +static struct { + pthread_mutex_t lock; + Known heads[KNOWN]; + unsigned next; + Known publishing; /* written here, not durable yet */ +} known = {.lock = PTHREAD_MUTEX_INITIALIZER}; + +static bool names(const Known *known_head, const struct stat *file, const Head *head) { + return known_head->state && known_head->device == file->st_dev && known_head->inode == file->st_ino && + known_head->seq == head->seq && known_head->check == head->check; +} + +static int recall(const struct stat *file, const Head *head) { + int state = UNKNOWN; + pthread_mutex_lock(&known.lock); + if (names(&known.publishing, file, head)) state = PUBLISHING; + for (unsigned i = 0; i < KNOWN && state == UNKNOWN; ++i) + if (names(&known.heads[i], file, head)) state = known.heads[i].state; + pthread_mutex_unlock(&known.lock); + return state; +} + +/* PUBLISHING marks the one head being made durable; any other state ends + * that mark (UNKNOWN only ends it). */ +static void remember(const struct stat *file, const Head *head, int state) { + Known entry = {file->st_dev, file->st_ino, head->seq, head->check, state}; + pthread_mutex_lock(&known.lock); + if (state == PUBLISHING) known.publishing = entry; + else if (names(&known.publishing, file, head)) known.publishing.state = UNKNOWN; + if (state != PUBLISHING && state != UNKNOWN) { + Known *at = NULL; + for (unsigned i = 0; i < KNOWN && !at; ++i) + if (names(&known.heads[i], file, head)) at = &known.heads[i]; + if (!at) known.heads[known.next++ % KNOWN] = entry; + else if (at->state != DURABLE) *at = entry; /* a durable head stays durable */ + } + pthread_mutex_unlock(&known.lock); } -static bool map_tree(int fd, const Head *head, Tree *tree) { - struct stat info; +#ifdef PURE_STORE_TEST +/* Test-only: forget every head, as a new process after a crash would. */ +void pure_tree_test_forget(void) { + pthread_mutex_lock(&known.lock); + memset(known.heads, 0, sizeof(known.heads)); + known.publishing.state = UNKNOWN; + pthread_mutex_unlock(&known.lock); +} +#endif + +/* Whether a head's commit has all of its bytes. */ +static bool whole(int fd, const Head *head) { + unsigned char buffer[16384]; + Digest digest = {0}; + for (uint64_t at = head->start; at < head->end;) { + size_t n = head->end - at < sizeof(buffer) ? (size_t)(head->end - at) : sizeof(buffer); + if (!read_at(fd, buffer, n, at)) return false; + digest_add(&digest, buffer, n); + at += n; + } + return digest_end(&digest) == head->check; +} + +/* The newest head that counts. A reader passes over the head this process is + * still making durable, so the process never answers with a commit that a + * crash could take back; a writer never meets one (it is the only writer). */ +static bool pick(int fd, const struct stat *file, const unsigned char *slots[2], bool writer, Head *out) { + Head heads[2]; + bool valid[2]; + for (unsigned i = 0; i < 2; ++i) + valid[i] = decode_head(slots[i], &heads[i]) && heads[i].end <= (uint64_t)file->st_size; + unsigned newest = valid[1] && (!valid[0] || heads[1].seq > heads[0].seq); + for (unsigned turn = 0; turn < 2; ++turn) { + unsigned slot = turn ? !newest : newest; + const Head *head = &heads[slot]; + if (!valid[slot]) continue; + int state = recall(file, head); + if (state == BROKEN || (state == PUBLISHING && !writer)) continue; + if (state == UNKNOWN) { + bool counts = whole(fd, head); + remember(file, head, counts ? CHECKED : BROKEN); + if (!counts) continue; + } + *out = *head; + return true; + } + return false; +} + +static bool read_head(int fd, struct stat *file, bool writer, Head *out) { + unsigned char slots[BLOCK + SLOT]; + return !fstat(fd, file) && S_ISREG(file->st_mode) && file->st_size >= 0 && + read_at(fd, slots, sizeof(slots), 0) && + pick(fd, file, (const unsigned char *[2]){slots, slots + BLOCK}, writer, out); +} + +static bool map_tree(int fd, const struct stat *file, const Head *head, Tree *tree) { *tree = (Tree){0}; - if (fstat(fd, &info) || !S_ISREG(info.st_mode) || info.st_size < 0 || - (uint64_t)info.st_size < head->end) return false; + if ((uint64_t)file->st_size < head->end) return false; void *map = mmap(NULL, (size_t)head->end, PROT_READ, MAP_SHARED, fd, 0); if (map == MAP_FAILED) return false; *tree = (Tree){.map = map, .end = head->end}; @@ -534,13 +721,29 @@ static bool rebuild(Tree *tree, Entry root, uint64_t offset, uint64_t remove, } static bool flush(Writer *writer) { - if (writer->used && !writer->failed && - !write_at(writer->fd, writer->buffer, writer->used, writer->at - writer->used)) - writer->failed = true; + if (writer->used && !writer->failed) { + digest_add(&writer->digest, writer->buffer, writer->used); + if (!write_at(writer->fd, writer->buffer, writer->used, writer->at - writer->used)) writer->failed = true; + } writer->used = 0; return !writer->failed; } +/* Zeros from..to; a failure only costs speed. */ +static void zero_fill(int fd, uint64_t from, uint64_t to) { + static const unsigned char zeros[BUFFER]; + while (from < to) { + size_t n = to - from < BUFFER ? (size_t)(to - from) : BUFFER; + if (!write_at(fd, zeros, n, from)) return; + from += n; + } +} + +/* Where the written zeros after a published end should reach. */ +static uint64_t tail_end(uint64_t end) { + return end + ((end - DATA) / 8 > TAIL_STEP ? (end - DATA) / 8 : TAIL_STEP); +} + static void put_bytes(Writer *writer, const unsigned char *bytes, size_t length) { while (length && !writer->failed) { if (writer->used == BUFFER && !flush(writer)) return; @@ -647,23 +850,29 @@ static int create_temporary(int directory, char name[96]) { return -1; } -/* Publish a fresh file with both slots pointing at head. */ -static bool install(int directory, char temporary[96], int fd, const Head *head) { - unsigned char slots[2 * SLOT]; - encode_head(slots, head); - memcpy(slots + SLOT, slots, SLOT); - bool ok = write_at(fd, slots, sizeof(slots), 0) && sync_fd(fd); +/* Publish a fresh file with both slots pointing at head. It is synced whole + * before it is renamed into place, so its head names no commit bytes. */ +static bool install(int directory, char temporary[96], int fd, Head *head) { + unsigned char slot[SLOT]; + struct stat file; + head->start = head->end; + head->check = digest_end(&(Digest){0}); + encode_head(slot, head); + zero_fill(fd, head->end, tail_end(head->end)); + bool ok = write_at(fd, slot, SLOT, 0) && write_at(fd, slot, SLOT, BLOCK) && sync_fd(fd) && !fstat(fd, &file); if (close(fd)) ok = false; ok = ok && !renameat(directory, temporary, directory, tree_name); if (ok) temporary[0] = 0; - return ok && sync_fd(directory); + ok = ok && sync_fd(directory); + if (ok) remember(&file, head, DURABLE); + return ok; } /* Rewrite only reachable nodes. Open readers keep their mapping of the * previous file. A failure leaves the published file untouched. */ -static void compact(int directory, int source, const Head *head) { +static void compact(int directory, int source, const struct stat *file, const Head *head) { Tree tree; - if (!map_tree(source, head, &tree)) return; + if (!map_tree(source, file, head, &tree)) return; char temporary[96]; int fd = create_temporary(directory, temporary); Writer *writer = fd >= 0 ? malloc(sizeof(*writer)) : NULL; @@ -686,7 +895,7 @@ static void compact(int directory, int source, const Head *head) { ok = ok && rebuild(&tree, (Entry){0}, 0, 0, (PureSpan){records, (size_t)(count * RECORD)}, &fresh) && write_root(&tree, writer, &fresh, false) && flush(writer); if (ok) { - Head compacted = {head->seq + 1, writer->at, fresh.ref, fresh.bytes, writer->at - DATA}; + Head compacted = {head->seq + 1, writer->at, fresh.ref, fresh.bytes, writer->at - DATA, 0, 0}; ok = install(directory, temporary, fd, &compacted); fd = -1; } @@ -716,9 +925,10 @@ PureResult pure_store_init(const char *path) { bool ok = lock >= 0 && fstat(lock, &info) == 0 && S_ISREG(info.st_mode) && lock_writer(lock, true); if (ok) { Head head; + struct stat file; int fd = openat(directory, tree_name, O_RDONLY | O_NOFOLLOW | O_CLOEXEC); if (fd >= 0) { - ok = read_head(fd, &head); /* An unreadable file is never replaced. */ + ok = read_head(fd, &file, true, &head); /* An unreadable file is never replaced. */ close(fd); } else if (errno != ENOENT) ok = false; else { @@ -767,13 +977,14 @@ PureResult pure_store_select(const char *path, const char *key, size_t offset, if (fd < 0) return PURE_STORE_ERROR; Tree tree = {0}; Head head; + struct stat file; Entry document; uint64_t index; int64_t revision; bool found; unsigned height; PureResult result = PURE_STORE_ERROR; - if (!read_head(fd, &head) || !map_tree(fd, &head, &tree) || + if (!read_head(fd, &file, false, &head) || !map_tree(fd, &file, &head, &tree) || !find(&tree, (Entry){head.catalog, head.catalog_bytes}, key, &index, &found, &document, &revision)) goto finish; if (!found) { result = PURE_NOT_FOUND; goto finish; } @@ -838,6 +1049,16 @@ typedef struct { bool turn; /* holds this process's writer turn */ } Session; +/* Keep written zeros after the published end, where commits and their + * unpublished leftovers go. A sync of blocks the file already has writes + * only them; a new size or new blocks add a journal commit (on an SD card + * 2.4 ms against 6 ms per commit). The zeros become durable with the next + * commit's sync. */ +static void prepare_tail(Session *session) { + uint64_t size = (uint64_t)session->file.st_size, end = session->head.end; + if (size < end + TAIL_MIN) zero_fill(session->fd, size > end ? size : end, tail_end(end)); +} + static void close_session(Session *session) { release_tree(&session->tree); if (session->fd >= 0) close(session->fd); @@ -859,19 +1080,22 @@ static PureResult open_session(const char *path, Session *session) { if (!lock_writer(session->lock, false)) return errno == EACCES || errno == EAGAIN ? PURE_BUSY : PURE_STORE_ERROR; session->fd = openat(session->directory, tree_name, O_RDWR | O_NOFOLLOW | O_CLOEXEC); - if (session->fd < 0 || !read_head(session->fd, &session->head) || - !map_tree(session->fd, &session->head, &session->tree) || fstat(session->fd, &session->file)) + if (session->fd < 0 || !read_head(session->fd, &session->file, true, &session->head)) return PURE_STORE_ERROR; - /* Bytes after the published head belong to a commit nobody acknowledged. */ - if ((uint64_t)session->file.st_size > session->head.end && - ftruncate(session->fd, (off_t)session->head.end)) return PURE_STORE_ERROR; - return PURE_OK; + /* A head this process did not make durable may be the last write of a + * writer that died before its sync: make it durable before building on it. */ + if (recall(&session->file, &session->head) != DURABLE) { + if (!sync_data(session->fd)) return PURE_STORE_ERROR; + remember(&session->file, &session->head, DURABLE); + } + prepare_tail(session); + return map_tree(session->fd, &session->file, &session->head, &session->tree) ? PURE_OK : PURE_STORE_ERROR; } -/* The documents' new nodes are already appended at writer. Record each in - * the catalog (revisions come in as the base and leave as the new one), - * then make the commit durable before its head becomes visible: every - * document becomes visible at once, or none does. */ +/* The documents' new nodes are already written at writer. Record each in + * the catalog (revisions come in as the base and leave as the new one), then + * write the head and sync once: every document becomes visible at once, or + * none does. Readers of this process see the head only once it is durable. */ #ifdef PURE_STORE_TEST static unsigned long publications; unsigned long pure_tree_test_publications(void) { return publications; } @@ -900,19 +1124,21 @@ static PureResult publish(Session *session, Writer *writer, size_t count, const catalog = updated; } written = written && write_root(&session->tree, writer, &catalog, false) && flush(writer); - if (!written || fault(PURE_AFTER_HEADER) || fault(PURE_BEFORE_FILE_SYNC) || - !sync_fd(session->fd) || fault(PURE_BEFORE_RENAME)) { - if (ftruncate(session->fd, (off_t)session->head.end) == 0) sync_fd(session->fd); + /* Unpublished bytes stay in the tail; the next commit overwrites them. */ + if (!written || fault(PURE_AFTER_HEADER) || fault(PURE_BEFORE_FILE_SYNC) || fault(PURE_BEFORE_RENAME)) return PURE_STORE_ERROR; - } - Head published = {session->head.seq + 1, writer->at, catalog.ref, catalog.bytes, session->head.live}; + Head published = {session->head.seq + 1, writer->at, catalog.ref, catalog.bytes, session->head.live, + session->head.end, digest_end(&writer->digest)}; unsigned char slot[SLOT]; encode_head(slot, &published); for (size_t i = 0; i < count; ++i) ++revisions[i]; /* Past this point the head may already be visible. */ - if (!write_at(session->fd, slot, SLOT, (published.seq % 2) * SLOT) || - fault(PURE_BEFORE_DIR_SYNC) || !sync_fd(session->fd)) return PURE_UNCERTAIN; - if (published.end - DATA > 4 * published.live + COMPACT_SLACK) - compact(session->directory, session->fd, &published); + remember(&session->file, &published, PUBLISHING); + bool durable = write_at(session->fd, slot, SLOT, (published.seq % 2) * BLOCK) && + !fault(PURE_BEFORE_DIR_SYNC) && sync_data(session->fd); + remember(&session->file, &published, durable ? DURABLE : CHECKED); + if (!durable) return PURE_UNCERTAIN; + if (published.end - DATA > 4 * published.live + COMPACT_SLACK && !fstat(session->fd, &session->file)) + compact(session->directory, session->fd, &session->file, &published); return PURE_OK; } @@ -942,10 +1168,7 @@ static PureResult mutate(const char *path, const char *key, int64_t expected, if (!writer || !rebuild(&session.tree, splice ? document : (Entry){0}, splice ? offset : 0, splice ? remove : 0, body, &next)) goto finish; *writer = (Writer){.fd = session.fd, .at = session.head.end}; - if (!write_root(&session.tree, writer, &next, false)) { - if (ftruncate(session.fd, (off_t)session.head.end) == 0) sync_fd(session.fd); - goto finish; - } + if (!write_root(&session.tree, writer, &next, false)) goto finish; result = publish(&session, writer, 1, &key, &next, revision); finish: free(writer); @@ -964,8 +1187,9 @@ PureResult pure_store_keys(const char *path, bool (*visit)(void *, const char *k if (fd < 0) return PURE_STORE_ERROR; Tree tree = {0}; Head head; + struct stat file; PureResult result = PURE_STORE_ERROR; - if (read_head(fd, &head) && map_tree(fd, &head, &tree)) { + if (read_head(fd, &file, false, &head) && map_tree(fd, &file, &head, &tree)) { Entry catalog = {head.catalog, head.catalog_bytes}, document; unsigned char record[RECORD]; char key[KEY + 1] = {0}; @@ -1023,7 +1247,7 @@ static PureResult load_draft(PureDraft *draft) { uint64_t index; int64_t revision; bool found; - bool ok = !fstat(fd, &info) && read_head(fd, &head) && map_tree(fd, &head, &tree) && + bool ok = read_head(fd, &info, false, &head) && map_tree(fd, &info, &head, &tree) && find(&tree, (Entry){head.catalog, head.catalog_bytes}, draft->key, &index, &found, &document, &revision); close(fd); @@ -1069,8 +1293,9 @@ void pure_draft_close(PureDraft *draft) { free(draft); } -/* Bring the handle to the newest published version: the same file and head - * cost two system calls; a new commit maps again, a compaction reopens. */ +/* Bring the handle to the newest published version: an unchanged store costs + * one system call (the mapping shows the head slots); a new commit maps + * again, a compaction reopens. */ static PureResult refresh(PureStore *store) { struct stat info; if (fstatat(store->directory, tree_name, &info, AT_SYMLINK_NOFOLLOW) || !S_ISREG(info.st_mode)) @@ -1084,13 +1309,23 @@ static PureResult refresh(PureStore *store) { store->inode = info.st_ino; store->head = (Head){0}; } + unsigned char slots[BLOCK + SLOT]; + if (store->head.seq) { + /* A newer head lands in the other slot first; a torn or older one there + * cannot be newer. */ + const unsigned char *map = store->mapping->map; + if (decode64(map + (store->head.seq + 1) % 2 * BLOCK + 8) <= store->head.seq) return PURE_OK; + memcpy(slots, map, SLOT); + memcpy(slots + BLOCK, map + BLOCK, SLOT); + } else if (!read_at(store->fd, slots, sizeof(slots), 0)) return PURE_STORE_ERROR; Head head; - if (!read_head(store->fd, &head)) return PURE_STORE_ERROR; - if (store->mapping && head.seq == store->head.seq && head.end == store->head.end) return PURE_OK; + if (!pick(store->fd, &info, (const unsigned char *[2]){slots, slots + BLOCK}, false, &head)) + return PURE_STORE_ERROR; + if (store->mapping && head.seq == store->head.seq && head.check == store->head.check) return PURE_OK; Tree tree; Mapping *mapping = malloc(sizeof(*mapping)); if (!mapping) return PURE_ALLOCATION; - if (!map_tree(store->fd, &head, &tree)) { free(mapping); return PURE_STORE_ERROR; } + if (!map_tree(store->fd, &info, &head, &tree)) { free(mapping); return PURE_STORE_ERROR; } *mapping = (Mapping){tree.map, tree.end, 1}; release_mapping(store->mapping); store->mapping = mapping; @@ -1358,11 +1593,7 @@ static void commit_group(const char *path, Waiter *first) { bool relocate = session.file.st_dev != drafts[i]->device || session.file.st_ino != drafts[i]->inode; written = write_root(&drafts[i]->tree, writer, &roots[i], relocate); } - if (written) result = publish(&session, writer, count, keys, roots, revisions); - else { - if (ftruncate(session.fd, (off_t)session.head.end) == 0) sync_fd(session.fd); - result = PURE_STORE_ERROR; - } + result = written ? publish(&session, writer, count, keys, roots, revisions) : PURE_STORE_ERROR; } } /* Rebase while still holding the writer lock; the arenas are no longer @@ -1507,12 +1738,13 @@ bool pure_tree_test_shape(const char *path, const char *key, unsigned *height, if (directory >= 0) close(directory); Tree tree = {0}; Head head; + struct stat file; Entry document; uint64_t index; int64_t revision; bool found = false; *leaves = *small = 0; - bool ok = fd >= 0 && read_head(fd, &head) && map_tree(fd, &head, &tree) && + bool ok = fd >= 0 && read_head(fd, &file, false, &head) && map_tree(fd, &file, &head, &tree) && find(&tree, (Entry){head.catalog, head.catalog_bytes}, key, &index, &found, &document, &revision) && found && root_height(&tree, document, height) && (!document.bytes || shape(&tree, document, *height, true, leaves, small)); diff --git a/prototypes/connections/test-tree-store.c b/prototypes/connections/test-tree-store.c index 9cb0701..bdb6573 100644 --- a/prototypes/connections/test-tree-store.c +++ b/prototypes/connections/test-tree-store.c @@ -18,12 +18,17 @@ static int fail_stage = -1; static int crash; static int slow_once; /* the next publication pauses before its data sync */ +static int held_sync; /* 1: the next publication waits after its head write; 2: it is waiting */ static void pause_ms(long ms) { struct timespec wait = {ms / 1000, (ms % 1000) * 1000000}; nanosleep(&wait, NULL); } int pure_store_test_fault(PureStoreStage stage) { if (stage == PURE_BEFORE_FILE_SYNC && __atomic_exchange_n(&slow_once, 0, __ATOMIC_SEQ_CST)) pause_ms(100); + int one = 1; + if (stage == PURE_BEFORE_DIR_SYNC && + __atomic_compare_exchange_n(&held_sync, &one, 2, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) + while (__atomic_load_n(&held_sync, __ATOMIC_SEQ_CST) == 2) pause_ms(1); if ((int)stage != fail_stage) return 0; if (crash) _exit(77); return 1; @@ -33,8 +38,12 @@ bool pure_tree_test_shape(const char *path, const char *key, unsigned *height, void pure_draft_test_arena(const PureDraft *draft, size_t *nodes, size_t *bytes); unsigned long pure_tree_test_publications(void); +unsigned long pure_tree_test_syncs(void); +void pure_tree_test_forget(void); -enum { KEYS = 4, CAPACITY = 1 << 20 }; +/* The file: head slots in blocks 0 and 1 (sequence at +8, end at +16, + * catalog at +24), commits from DATA_START. */ +enum { KEYS = 4, CAPACITY = 1 << 20, SLOT_BLOCK = 4096, DATA_START = 8192 }; typedef struct { unsigned char *bytes; size_t size; int64_t revision; } Oracle; static uint32_t random_value(uint32_t *state) { @@ -60,6 +69,22 @@ static uint64_t file_size(const char *path) { return (uint64_t)info.st_size; } +/* Where the newest commit ends. */ +static uint64_t published_end(const char *path, const char *key) { + unsigned height; + uint64_t leaves, small, end; + CHECK(pure_tree_test_shape(path, key, &height, &leaves, &small, &end)); + return end; +} + +static uint64_t slot_field(int fd, unsigned slot, unsigned offset) { + unsigned char bytes[8]; + CHECK(pread(fd, bytes, 8, (off_t)(slot * SLOT_BLOCK + offset)) == 8); + uint64_t value = 0; + for (unsigned i = 0; i < 8; ++i) value |= (uint64_t)bytes[i] << (8 * i); + return value; +} + static void fill(unsigned char *bytes, size_t size, uint32_t *state) { for (size_t i = 0; i < size; ++i) bytes[i] = (unsigned char)random_value(state); } @@ -148,7 +173,7 @@ static void amplification(const char *path) { const size_t edits[][3] = {{size / 2, 0, 1}, {size / 3, 1, 0}, {size - 1, 1, 1}, {0, 0, 1}}; uint64_t worst = 0; for (size_t i = 0; i < sizeof(edits) / sizeof(*edits); ++i) { - uint64_t before = file_size(path); + uint64_t before = published_end(path, "large"); size_t offset = edits[i][0], remove = edits[i][1], count = edits[i][2]; CHECK(pure_store_splice(path, "large", o.revision, offset, remove, (PureSpan){&byte, count}, PURE_OBJECT_LIMIT, &revision) == PURE_OK); @@ -156,7 +181,7 @@ static void amplification(const char *path) { memcpy(bytes + offset, &byte, count); o.size = o.size - remove + count; o.revision = revision; - uint64_t after = file_size(path); + uint64_t after = published_end(path, "large"); if (after > before && after - before > worst) worst = after - before; expect(path, "large", &o); } @@ -176,12 +201,13 @@ static void faults(const char *path) { int64_t revision; CHECK(pure_store_put(path, "fault", 0, (PureSpan){text, 10}, &revision) == PURE_OK); Oracle o = {text, 10, revision}; - uint64_t size = file_size(path); + uint64_t size = file_size(path), end = published_end(path, "fault"); for (fail_stage = PURE_AFTER_HEADER; fail_stage <= PURE_BEFORE_RENAME; ++fail_stage) { CHECK(pure_store_splice(path, "fault", o.revision, 1, 3, (PureSpan){text, 5}, PURE_OBJECT_LIMIT, &revision) == PURE_STORE_ERROR); expect(path, "fault", &o); - CHECK(file_size(path) == size); + /* The bytes stayed unpublished, in the written zeros after the end. */ + CHECK(published_end(path, "fault") == end && file_size(path) == size); } fail_stage = PURE_BEFORE_DIR_SYNC; CHECK(pure_store_splice(path, "fault", o.revision, 10, 0, (PureSpan){text, 2}, @@ -194,16 +220,16 @@ static void faults(const char *path) { } /* A process that dies at each publication stage leaves either the previous - * or the next complete version, and its unpublished bytes are discarded. */ + * or the next complete version; the next commit overwrites its unpublished + * bytes. A head it wrote but did not sync is made durable before the next + * writer builds on it. */ static void crashes(const char *path) { unsigned char text[] = "crash-base"; int64_t revision; CHECK(pure_store_put(path, "crash", 0, (PureSpan){text, 10}, &revision) == PURE_OK); Oracle o = {text, 10, revision}; for (int stage = PURE_AFTER_HEADER; stage <= PURE_BEFORE_DIR_SYNC; ++stage) { - unsigned height; - uint64_t leaves, small, before, end; - CHECK(pure_tree_test_shape(path, "crash", &height, &leaves, &small, &before)); + uint64_t before = published_end(path, "crash"); pid_t child = fork(); CHECK(child >= 0); if (!child) { @@ -217,19 +243,20 @@ static void crashes(const char *path) { CHECK(waitpid(child, &status, 0) == child && WIFEXITED(status) && WEXITSTATUS(status) == 77); if (stage == PURE_BEFORE_DIR_SYNC) { text[0] = 'C'; ++o.revision; } expect(path, "crash", &o); - uint64_t garbage = file_size(path); + uint64_t published = published_end(path, "crash"); + CHECK(stage == PURE_BEFORE_DIR_SYNC ? published > before : published == before); + unsigned long syncs = pure_tree_test_syncs(); CHECK(pure_store_splice(path, "crash", o.revision, 10, 0, (PureSpan){(const unsigned char *)"", 0}, PURE_OBJECT_LIMIT, &revision) == PURE_OK); o.revision = revision; - CHECK(pure_tree_test_shape(path, "crash", &height, &leaves, &small, &end)); - CHECK(end == file_size(path)); - /* Same-sized commit: equal growth only if the dead writer's tail was cut. */ - if (stage != PURE_BEFORE_DIR_SYNC) CHECK(garbage > before && end - before == garbage - before); + expect(path, "crash", &o); + CHECK(pure_tree_test_syncs() - syncs == (stage == PURE_BEFORE_DIR_SYNC ? 2u : 1u)); + CHECK(published_end(path, "crash") > published); } } /* A torn newest head slot exposes the previous commit, and the next writer - * overwrites that slot and the unpublished tail. */ + * overwrites that slot and the commit's bytes. */ static void torn_head(const char *path) { unsigned char first[] = "first", second[] = "second"; int64_t revision; @@ -239,12 +266,11 @@ static void torn_head(const char *path) { char name[512]; snprintf(name, sizeof(name), "%s/pure.tree", path); int fd = open(name, O_RDWR); - unsigned char slots[128]; - CHECK(fd >= 0 && pread(fd, slots, 128, 0) == 128); - uint64_t a = 0, b = 0; - for (unsigned i = 0; i < 8; ++i) { a |= (uint64_t)slots[8 + i] << (8 * i); b |= (uint64_t)slots[72 + i] << (8 * i); } - off_t newest = a > b ? 0 : 64; - unsigned char torn = slots[newest + 20] ^ 0x5a; + CHECK(fd >= 0); + off_t newest = slot_field(fd, 0, 8) > slot_field(fd, 1, 8) ? 0 : SLOT_BLOCK; + unsigned char torn; + CHECK(pread(fd, &torn, 1, newest + 20) == 1); + torn ^= 0x5a; CHECK(pwrite(fd, &torn, 1, newest + 20) == 1 && close(fd) == 0); expect(path, "torn", &o); CHECK(pure_store_put(path, "torn", o.revision + 1, (PureSpan){first, 1}, &revision) == PURE_CONFLICT); @@ -298,12 +324,12 @@ static void corruption(const char *path) { CHECK(pure_store_put(path, "damaged", 0, (PureSpan){bytes, 200000}, &revision) == PURE_OK); char name[512]; snprintf(name, sizeof(name), "%s/pure.tree", path); - uint64_t size = file_size(path); + uint64_t end = published_end(path, "damaged"); int fd = open(name, O_RDWR); CHECK(fd >= 0); unsigned errors = 0; for (unsigned i = 0; i < 3000; ++i) { - off_t at = (off_t)(128 + random_value(&state) % (size - 128)); + off_t at = (off_t)(DATA_START + random_value(&state) % (end - DATA_START)); unsigned char original, damaged; CHECK(pread(fd, &original, 1, at) == 1); damaged = original ^ (unsigned char)(1u << (random_value(&state) % 8)); @@ -345,17 +371,14 @@ static void revision_limit(const char *path) { char name[512]; snprintf(name, sizeof(name), "%s/pure.tree", path); int fd = open(name, O_RDWR); - unsigned char slots[128]; - CHECK(fd >= 0 && pread(fd, slots, 128, 0) == 128); - uint64_t seq[2] = {0, 0}, catalog = 0; - for (unsigned s = 0; s < 2; ++s) - for (unsigned i = 0; i < 8; ++i) seq[s] |= (uint64_t)slots[64 * s + 8 + i] << (8 * i); - unsigned newest = seq[0] > seq[1] ? 0 : 1; - for (unsigned i = 0; i < 8; ++i) catalog |= (uint64_t)slots[64 * newest + 24 + i] << (8 * i); + CHECK(fd >= 0); + uint64_t catalog = slot_field(fd, slot_field(fd, 0, 8) > slot_field(fd, 1, 8) ? 0 : 1, 24); CHECK(catalog >> 63); const unsigned char maximum[] = {255,255,255,255,255,255,255,127}; off_t record = (off_t)(catalog & ((UINT64_C(1) << 62) - 1)); CHECK(pwrite(fd, maximum, 8, record + 80) == 8 && close(fd) == 0); + /* The forged commit no longer matches its digest; this process wrote it + * and does not read it again, a new process would pass over it. */ PureStoredView view; CHECK(pure_store_get(path, "work", 1, &view) == PURE_OK && view.revision == INT64_MAX); pure_store_release(&view); @@ -483,15 +506,16 @@ static void typing(const char *path) { /* Reachable: 10,000 net typed bytes, the leaves they landed in and one * path. Unreachable copies are reclaimed past max(reachable, 32 KiB). */ CHECK(peak < 2 * (10000 + 4 * 4096 + 2048) + 32768 + 8192); - uint64_t before = file_size(path); + uint64_t before = published_end(path, "typed"); int64_t revision; CHECK(pure_draft_commit(draft, &revision) == PURE_OK); o.revision = revision; - CHECK(file_size(path) - before < 10000 + 16384); + uint64_t appended = published_end(path, "typed") - before; + CHECK(appended < 10000 + 16384); expect(path, "typed", &o); pure_draft_close(draft); printf("{\"typing_keystrokes\":30000,\"peak_arena_bytes\":%zu,\"commit_appended_bytes\":%llu}\n", - peak, (unsigned long long)(file_size(path) - before)); + peak, (unsigned long long)appended); free(o.bytes); free(scratch); } @@ -927,6 +951,84 @@ static void grouped(const char *path) { } } +/* A crash during a commit's only sync can leave its head on the device + * without all of its bytes. A process that starts after it passes over that + * head to the previous commit, and the next writer overwrites both. */ +static void lost_commit(const char *path) { + unsigned char first[] = "first", second[] = "second"; + int64_t revision; + CHECK(pure_store_put(path, "lost", 0, (PureSpan){first, 5}, &revision) == PURE_OK); + Oracle o = {first, 5, revision}; + uint64_t start = published_end(path, "lost"); + CHECK(pure_store_put(path, "lost", o.revision, (PureSpan){second, 6}, &revision) == PURE_OK); + uint64_t end = published_end(path, "lost"); + char name[512]; + snprintf(name, sizeof(name), "%s/pure.tree", path); + int fd = open(name, O_RDWR); + off_t at = (off_t)(start + (end - start) / 2); + unsigned char byte; + CHECK(fd >= 0 && pread(fd, &byte, 1, at) == 1); + byte ^= 0xff; + CHECK(pwrite(fd, &byte, 1, at) == 1 && close(fd) == 0); + pure_tree_test_forget(); + expect(path, "lost", &o); + PureStore *store; + PureDraft *draft; + CHECK(pure_store_open(path, &store) == PURE_OK && pure_store_draft(store, "lost", &draft) == PURE_OK); + CHECK(pure_draft_revision(draft) == o.revision); + pure_draft_close(draft); + CHECK(pure_store_put(path, "lost", o.revision + 1, (PureSpan){first, 1}, &revision) == PURE_CONFLICT); + CHECK(pure_store_put(path, "lost", o.revision, (PureSpan){second, 3}, &revision) == PURE_OK); + o = (Oracle){second, 3, revision}; + CHECK(published_end(path, "lost") < end); /* it started where the lost commit did */ + CHECK(pure_store_draft(store, "lost", &draft) == PURE_OK && pure_draft_revision(draft) == o.revision); + pure_draft_close(draft); + pure_store_close(store); + pure_tree_test_forget(); + expect(path, "lost", &o); +} + +/* While a commit's sync is pending, this process keeps reading the previous + * version: a crash could still take the new one back. */ +static void *commit_held(void *context) { + int64_t revision; + CHECK(pure_store_put(context, "pending", 1, (PureSpan){(const unsigned char *)"new", 3}, &revision) == PURE_OK); + return NULL; +} + +static void pending_head(const char *path) { + int64_t revision; + CHECK(pure_store_put(path, "pending", 0, (PureSpan){(const unsigned char *)"old", 3}, &revision) == PURE_OK); + PureStore *store; + PureDraft *draft; + CHECK(pure_store_open(path, &store) == PURE_OK); + __atomic_store_n(&held_sync, 1, __ATOMIC_SEQ_CST); + pthread_t writer; + CHECK(pthread_create(&writer, NULL, commit_held, (void *)path) == 0); + while (__atomic_load_n(&held_sync, __ATOMIC_SEQ_CST) != 2) pause_ms(1); + CHECK(pure_store_draft(store, "pending", &draft) == PURE_OK && pure_draft_revision(draft) == 1); + pure_draft_close(draft); + expect(path, "pending", &(Oracle){(unsigned char *)"old", 3, 1}); + __atomic_store_n(&held_sync, 0, __ATOMIC_SEQ_CST); + CHECK(pthread_join(writer, NULL) == 0); + CHECK(pure_store_draft(store, "pending", &draft) == PURE_OK && pure_draft_revision(draft) == 2); + pure_draft_close(draft); + pure_store_close(store); + expect(path, "pending", &(Oracle){(unsigned char *)"new", 3, 2}); +} + +/* A commit costs one sync, and commits into the written zeros after the end + * leave the file size alone (no journal commit for a new size). */ +static void one_sync(const char *path) { + int64_t revision; + CHECK(pure_store_put(path, "sync", 0, (PureSpan){(const unsigned char *)"0", 1}, &revision) == PURE_OK); + uint64_t size = file_size(path); + unsigned long syncs = pure_tree_test_syncs(); + for (unsigned i = 0; i < 20; ++i) + CHECK(pure_store_put(path, "sync", revision, (PureSpan){(const unsigned char *)"x", 1}, &revision) == PURE_OK); + CHECK(pure_tree_test_syncs() - syncs == 20 && file_size(path) == size); +} + static void clean(const char *path) { DIR *entries = opendir(path); CHECK(entries); @@ -943,7 +1045,8 @@ int main(void) { void (*checks[])(const char *) = {oracle, amplification, faults, crashes, torn_head, isolation, corruption, many_keys, revision_limit, keys, drafts, typing, draft_conflict, draft_relocation, draft_faults, - draft_versions, anonymous_draft, batch, store_handle, threads, grouped}; + draft_versions, anonymous_draft, batch, store_handle, threads, grouped, + lost_commit, pending_head, one_sync}; for (size_t i = 0; i < sizeof(checks) / sizeof(*checks); ++i) { char path[] = "/tmp/pure-tree-check-XXXXXX"; CHECK(mkdtemp(path) && pure_store_init(path) == PURE_OK); @@ -954,6 +1057,6 @@ int main(void) { puts("{\"tree_store\":\"random-oracle-balance-held-views-one-byte-amplification-faults-" "process-crash-torn-head-lockless-isolation-compaction-corruption-1024-keys-revision-limit-key-listing-" "draft-oracle-typing-arena-conflict-relocation-commit-faults-held-versions-anonymous-" - "atomic-batch-runtime-keys-store-handle-threads-group-commit\"}"); + "atomic-batch-runtime-keys-store-handle-threads-group-commit-lost-commit-pending-head-one-sync\"}"); return 0; } |