From 474ed25f08121c743fb476dd6fd470989abc9b8b Mon Sep 17 00:00:00 2001 From: Cole Miller Date: Wed, 18 Sep 2024 13:32:48 -0400 Subject: [PATCH] Remove remaining traces of local_data I've preserved is_local since it's still potentially useful. Signed-off-by: Cole Miller --- src/leader.c | 10 --------- src/raft.h | 30 ------------------------- src/raft/client.c | 5 ++--- src/raft/fixture.c | 2 +- src/raft/log.c | 8 ++----- src/raft/log.h | 2 -- src/raft/replication.c | 2 +- src/raft/start.c | 2 +- src/raft/uv_append.c | 2 +- src/raft/uv_encoding.c | 24 +++++--------------- src/raft/uv_encoding.h | 12 ++++------ src/raft/uv_recv.c | 3 +-- src/raft/uv_segment.c | 21 ++++++++--------- test/raft/integration/test_apply.c | 4 ++-- test/raft/integration/test_fixture.c | 2 +- test/raft/integration/test_membership.c | 2 +- test/raft/lib/cluster.h | 2 +- test/raft/unit/test_log.c | 10 ++++----- 18 files changed, 38 insertions(+), 105 deletions(-) diff --git a/src/leader.c b/src/leader.c index d0d8043ef..4f1e4b5e4 100644 --- a/src/leader.c +++ b/src/leader.c @@ -224,11 +224,7 @@ static void leaderMaybeCheckpointLegacy(struct leader *l) tracef("raft_malloc - no mem"); goto err_after_buf_alloc; } -#ifdef USE_SYSTEM_RAFT rv = raft_apply(l->raft, apply, &buf, 1, leaderCheckpointApplyCb); -#else - rv = raft_apply(l->raft, apply, &buf, NULL, 1, leaderCheckpointApplyCb); -#endif if (rv != 0) { tracef("raft_apply failed %d", rv); raft_free(apply); @@ -336,13 +332,7 @@ static int leaderApplyFrames(struct exec *req, apply->type = COMMAND_FRAMES; idSet(apply->req.req_id, req->id); -#ifdef USE_SYSTEM_RAFT rv = raft_apply(l->raft, &apply->req, &buf, 1, leaderApplyFramesCb); -#else - /* TODO actual WAL slice goes here */ - struct raft_entry_local_data local_data = {}; - rv = raft_apply(l->raft, &apply->req, &buf, &local_data, 1, leaderApplyFramesCb); -#endif if (rv != 0) { tracef("raft apply failed %d", rv); goto err_after_command_encode; diff --git a/src/raft.h b/src/raft.h index 76bea3500..8d0156689 100644 --- a/src/raft.h +++ b/src/raft.h @@ -198,28 +198,6 @@ enum { RAFT_CHANGE /* Raft configuration change. */ }; -/** - * A small fixed-size inline buffer that stores extra data for a raft_entry - * that is different for each node in the cluster. - * - * A leader initializes the local data for an entry before passing it into - * raft_apply. This local data is stored in the volatile raft log and also - * in the persistent raft log on the leader. AppendEntries messages sent by - * the leader never contain the local data for entries. - * - * When a follower accepts an AppendEntries request, it invokes a callback - * provided by the FSM to fill out the local data for each new entry before - * appending the entries to its log (volatile and persistent). This local - * data doesn't have to be the same as the local data that the leader computed. - * - * When starting up, a raft node reads the local data for each entry for its - * persistent log as part of populating the volatile log. - */ -struct raft_entry_local_data { - /* Must be the only member of this struct. */ - uint8_t buf[16]; -}; - /** * A single entry in the raft log. * @@ -250,12 +228,6 @@ struct raft_entry_local_data { * message or in the persistent log. This field can be used by the FSM's `apply` * callback to handle a COMMAND entry differently depending on whether it * originated locally. - * - * Note: The @local_data and @is_local fields do not exist when we use an external - * libraft, because the last separate release of libraft predates their addition. - * The ifdef at the very top of this file ensures that we use the system raft headers - * when we build against an external libraft, so there will be no ABI mismatch as - * a result of incompatible struct layouts. */ struct raft_entry { @@ -263,7 +235,6 @@ struct raft_entry unsigned short type; /* Type (FSM command, barrier, config change). */ bool is_local; /* Placed here so it goes in the padding after @type. */ struct raft_buffer buf; /* Entry data. */ - struct raft_entry_local_data local_data; void *batch; /* Batch that buf's memory points to, if any. */ }; @@ -1244,7 +1215,6 @@ struct raft_apply RAFT_API int raft_apply(struct raft *r, struct raft_apply *req, const struct raft_buffer bufs[], - const struct raft_entry_local_data local_data[], const unsigned n, raft_apply_cb cb); diff --git a/src/raft/client.c b/src/raft/client.c index 8111df2b6..3fe61a19e 100644 --- a/src/raft/client.c +++ b/src/raft/client.c @@ -14,7 +14,6 @@ int raft_apply(struct raft *r, struct raft_apply *req, const struct raft_buffer bufs[], - const struct raft_entry_local_data local_data[], const unsigned n, raft_apply_cb cb) { @@ -42,7 +41,7 @@ int raft_apply(struct raft *r, req->cb = cb; /* Append the new entries to the log. */ - rv = logAppendCommands(r->log, r->current_term, bufs, local_data, n); + rv = logAppendCommands(r->log, r->current_term, bufs, n); if (rv != 0) { goto err; } @@ -91,7 +90,7 @@ int raft_barrier(struct raft *r, struct raft_barrier *req, raft_barrier_cb cb) req->index = index; req->cb = cb; - rv = logAppend(r->log, r->current_term, RAFT_BARRIER, buf, (struct raft_entry_local_data){}, true, NULL); + rv = logAppend(r->log, r->current_term, RAFT_BARRIER, buf, true, NULL); if (rv != 0) { goto err_after_buf_alloc; } diff --git a/src/raft/fixture.c b/src/raft/fixture.c index fe1772373..b8216434d 100644 --- a/src/raft/fixture.c +++ b/src/raft/fixture.c @@ -1317,7 +1317,7 @@ static void copyLeaderLog(struct raft_fixture *f) assert(buf.base != NULL); memcpy(buf.base, entry->buf.base, buf.len); /* FIXME(cole) what to do here for is_local? */ - rv = logAppend(f->log, entry->term, entry->type, buf, (struct raft_entry_local_data){}, false, NULL); + rv = logAppend(f->log, entry->term, entry->type, buf, false, NULL); assert(rv == 0); } logRelease(raft->log, 1, entries, n); diff --git a/src/raft/log.c b/src/raft/log.c index dc8013fa2..1e347de8c 100644 --- a/src/raft/log.c +++ b/src/raft/log.c @@ -546,7 +546,6 @@ int logAppend(struct raft_log *l, const raft_term term, const unsigned short type, struct raft_buffer buf, - struct raft_entry_local_data local_data, bool is_local, void *batch) { @@ -576,7 +575,6 @@ int logAppend(struct raft_log *l, entry->type = type; entry->buf = buf; entry->batch = batch; - entry->local_data = local_data; entry->is_local = is_local; l->back += 1; @@ -588,7 +586,6 @@ int logAppend(struct raft_log *l, int logAppendCommands(struct raft_log *l, const raft_term term, const struct raft_buffer bufs[], - const struct raft_entry_local_data local_data[], const unsigned n) { unsigned i; @@ -600,8 +597,7 @@ int logAppendCommands(struct raft_log *l, assert(n > 0); for (i = 0; i < n; i++) { - struct raft_entry_local_data loc = (local_data != NULL) ? local_data[i] : (struct raft_entry_local_data){}; - rv = logAppend(l, term, RAFT_COMMAND, bufs[i], loc, true, NULL); + rv = logAppend(l, term, RAFT_COMMAND, bufs[i], true, NULL); if (rv != 0) { return rv; } @@ -628,7 +624,7 @@ int logAppendConfiguration(struct raft_log *l, } /* Append the new entry to the log. */ - rv = logAppend(l, term, RAFT_CHANGE, buf, (struct raft_entry_local_data){}, true, NULL); + rv = logAppend(l, term, RAFT_CHANGE, buf, true, NULL); if (rv != 0) { goto err_after_encode; } diff --git a/src/raft/log.h b/src/raft/log.h index 1be9cdfbc..137ef02ae 100644 --- a/src/raft/log.h +++ b/src/raft/log.h @@ -114,7 +114,6 @@ int logAppend(struct raft_log *l, raft_term term, unsigned short type, struct raft_buffer buf, - struct raft_entry_local_data local_data, bool is_local, void *batch); @@ -122,7 +121,6 @@ int logAppend(struct raft_log *l, int logAppendCommands(struct raft_log *l, const raft_term term, const struct raft_buffer bufs[], - const struct raft_entry_local_data local_data[], const unsigned n); /* Convenience to encode and append a single #RAFT_CHANGE entry. */ diff --git a/src/raft/replication.c b/src/raft/replication.c index 20f92d566..e263f5369 100644 --- a/src/raft/replication.c +++ b/src/raft/replication.c @@ -1235,7 +1235,7 @@ int replicationAppend(struct raft *r, goto err_after_request_alloc; } - rv = logAppend(r->log, copy.term, copy.type, copy.buf, (struct raft_entry_local_data){}, false, NULL); + rv = logAppend(r->log, copy.term, copy.type, copy.buf, false, NULL); if (rv != 0) { goto err_after_request_alloc; } diff --git a/src/raft/start.c b/src/raft/start.c index 45cab67fd..ee96f5812 100644 --- a/src/raft/start.c +++ b/src/raft/start.c @@ -72,7 +72,7 @@ static int restoreEntries(struct raft *r, for (i = 0; i < n; i++) { struct raft_entry *entry = &entries[i]; rv = logAppend(r->log, entry->term, entry->type, entry->buf, - entry->local_data, entry->is_local, entry->batch); + entry->is_local, entry->batch); if (rv != 0) { goto err; } diff --git a/src/raft/uv_append.c b/src/raft/uv_append.c index 1544c274e..42756c451 100644 --- a/src/raft/uv_append.c +++ b/src/raft/uv_append.c @@ -597,7 +597,7 @@ static size_t uvAppendSize(struct uvAppend *a) { size_t size = sizeof(uint32_t) * 2; /* CRC checksums */ unsigned i; - size += uvSizeofBatchHeader(a->n, true); /* Batch header */ + size += uvSizeofBatchHeader(a->n); /* Batch header */ for (i = 0; i < a->n; i++) { /* Entries data */ size += bytePad64(a->entries[i].buf.len); } diff --git a/src/raft/uv_encoding.c b/src/raft/uv_encoding.c index be3a4e978..c43a86189 100644 --- a/src/raft/uv_encoding.c +++ b/src/raft/uv_encoding.c @@ -86,11 +86,10 @@ static size_t sizeofTimeoutNow(void) sizeof(uint64_t) /* Last log term. */; } -size_t uvSizeofBatchHeader(size_t n, bool with_local_data) +size_t uvSizeofBatchHeader(size_t n) { size_t res = 8 + /* Number of entries in the batch, little endian */ 16 * n; /* One header per entry */; - (void)with_local_data; return res; } @@ -139,7 +138,7 @@ static void encodeAppendEntries(const struct raft_append_entries *p, void *buf) bytePut64(&cursor, p->prev_log_term); /* Previous term. */ bytePut64(&cursor, p->leader_commit); /* Commit index. */ - uvEncodeBatchHeader(p->entries, p->n_entries, cursor, false /* no local data */); + uvEncodeBatchHeader(p->entries, p->n_entries, cursor); } static void encodeAppendEntriesResult( @@ -297,8 +296,7 @@ int uvEncodeMessage(const struct raft_message *message, void uvEncodeBatchHeader(const struct raft_entry *entries, unsigned n, - void *buf, - bool with_local_data) + void *buf) { unsigned i; void *cursor = buf; @@ -306,8 +304,6 @@ void uvEncodeBatchHeader(const struct raft_entry *entries, /* Number of entries in the batch, little endian */ bytePut64(&cursor, n); - (void)with_local_data; - for (i = 0; i < n; i++) { const struct raft_entry *entry = &entries[i]; @@ -368,8 +364,7 @@ static void decodeRequestVoteResult(const uv_buf_t *buf, int uvDecodeBatchHeader(const void *batch, struct raft_entry **entries, - unsigned *n, - uint64_t *local_data_size) + unsigned *n) { const void *cursor = batch; size_t i; @@ -382,8 +377,6 @@ int uvDecodeBatchHeader(const void *batch, return 0; } - (void)local_data_size; - *entries = raft_malloc(*n * sizeof **entries); if (*entries == NULL) { @@ -438,7 +431,7 @@ static int decodeAppendEntries(const uv_buf_t *buf, args->prev_log_term = byteGet64(&cursor); args->leader_commit = byteGet64(&cursor); - rv = uvDecodeBatchHeader(cursor, &args->entries, &args->n_entries, false); + rv = uvDecodeBatchHeader(cursor, &args->entries, &args->n_entries); if (rv != 0) { return rv; } @@ -560,8 +553,7 @@ int uvDecodeMessage(uint16_t type, int uvDecodeEntriesBatch(uint8_t *batch, size_t offset, struct raft_entry *entries, - unsigned n, - uint64_t local_data_size) + unsigned n) { uint8_t *cursor; @@ -581,10 +573,6 @@ int uvDecodeEntriesBatch(uint8_t *batch, } entry->is_local = false; - - entry->local_data = (struct raft_entry_local_data){}; - assert(local_data_size <= sizeof(entry->local_data.buf)); - assert(local_data_size % 8 == 0); } return 0; } diff --git a/src/raft/uv_encoding.h b/src/raft/uv_encoding.h index 3152ee646..d48ce825d 100644 --- a/src/raft/uv_encoding.h +++ b/src/raft/uv_encoding.h @@ -21,14 +21,12 @@ int uvDecodeMessage(uint16_t type, int uvDecodeBatchHeader(const void *batch, struct raft_entry **entries, - unsigned *n, - uint64_t *local_data_size); + unsigned *n); int uvDecodeEntriesBatch(uint8_t *batch, size_t offset, struct raft_entry *entries, - unsigned n, - uint64_t local_data_size); + unsigned n); /** * The layout of the memory pointed at by a @batch pointer is the following: @@ -47,17 +45,15 @@ int uvDecodeEntriesBatch(uint8_t *batch, * [1 byte ] Message type (Either RAFT_COMMAND or RAFT_CHANGE) * [3 bytes] Currently unused. * [4 bytes] Size of the log entry data, little endian. - * [8 bytes] Size of the local buffer, little endian. * * A payload data section for an entry is simply a sequence of bytes of * arbitrary lengths, possibly padded with extra bytes to reach 8-byte boundary * (which means that all entry data pointers are 8-byte aligned). */ -size_t uvSizeofBatchHeader(size_t n, bool with_local_data); +size_t uvSizeofBatchHeader(size_t n); void uvEncodeBatchHeader(const struct raft_entry *entries, unsigned n, - void *buf, - bool with_local_data); + void *buf); #endif /* UV_ENCODING_H_ */ diff --git a/src/raft/uv_recv.c b/src/raft/uv_recv.c index 01e4ffc73..22be2e284 100644 --- a/src/raft/uv_recv.c +++ b/src/raft/uv_recv.c @@ -294,8 +294,7 @@ static void uvServerReadCb(uv_stream_t *stream, payload.base, 0, s->message.append_entries.entries, s->message.append_entries - .n_entries, - false); + .n_entries); break; case RAFT_IO_INSTALL_SNAPSHOT: s->message.install_snapshot.data.base = diff --git a/src/raft/uv_segment.c b/src/raft/uv_segment.c index 3821c882a..8d91ff667 100644 --- a/src/raft/uv_segment.c +++ b/src/raft/uv_segment.c @@ -266,11 +266,11 @@ static int uvLoadEntriesBatch(struct uv *uv, /* Consume the batch header, excluding the first 8 bytes containing the * number of entries, which we have already read. */ - header.len = uvSizeofBatchHeader(n, true); + header.len = uvSizeofBatchHeader(n); header.base = batch; rv = uvConsumeContent(content, offset, - uvSizeofBatchHeader(n, true) - sizeof(uint64_t), NULL, + uvSizeofBatchHeader(n) - sizeof(uint64_t), NULL, errmsg); if (rv != 0) { ErrMsgTransfer(errmsg, uv->io->errmsg, "read header"); @@ -288,8 +288,7 @@ static int uvLoadEntriesBatch(struct uv *uv, } /* Decode the batch header, allocating the entries array. */ - uint64_t local_data_size = 0; - rv = uvDecodeBatchHeader(header.base, entries, n_entries, &local_data_size); + rv = uvDecodeBatchHeader(header.base, entries, n_entries); if (rv != 0) { goto err; } @@ -321,7 +320,7 @@ static int uvLoadEntriesBatch(struct uv *uv, } rv = uvDecodeEntriesBatch(content->base, *offset - data.len, *entries, - *n_entries, local_data_size); + *n_entries); if (rv != 0) { goto err_after_header_decode; } @@ -745,7 +744,7 @@ int uvSegmentBufferAppend(struct uvSegmentBuffer *b, int rv; size = sizeof(uint32_t) * 2; /* CRC checksums */ - size += uvSizeofBatchHeader(n_entries, true); /* Batch header */ + size += uvSizeofBatchHeader(n_entries); /* Batch header */ for (i = 0; i < n_entries; i++) { /* Entries data */ size += bytePad64(entries[i].buf.len); } @@ -764,9 +763,9 @@ int uvSegmentBufferAppend(struct uvSegmentBuffer *b, /* Batch header */ header = cursor; - uvEncodeBatchHeader(entries, n_entries, cursor, true /* encode local data */); - crc1 = byteCrc32(header, uvSizeofBatchHeader(n_entries, true), 0); - cursor = (uint8_t *)cursor + uvSizeofBatchHeader(n_entries, true); + uvEncodeBatchHeader(entries, n_entries, cursor); + crc1 = byteCrc32(header, uvSizeofBatchHeader(n_entries), 0); + cursor = (uint8_t *)cursor + uvSizeofBatchHeader(n_entries); /* Batch data */ crc2 = 0; @@ -776,8 +775,6 @@ int uvSegmentBufferAppend(struct uvSegmentBuffer *b, memcpy(cursor, entry->buf.base, entry->buf.len); crc2 = byteCrc32(cursor, entry->buf.len, crc2); cursor = (uint8_t *)cursor + entry->buf.len; - static_assert(sizeof(entry->local_data.buf) % sizeof(uint64_t) == 0, - "bad size for entry local data"); } bytePut32(&crc1_p, crc1); @@ -1018,7 +1015,7 @@ static int uvWriteClosedSegment(struct uv *uv, * block */ cap = uv->block_size - (sizeof(uint64_t) /* Format version */ + - sizeof(uint64_t) /* Checksums */ + uvSizeofBatchHeader(1, true /* include local bufs */)); + sizeof(uint64_t) /* Checksums */ + uvSizeofBatchHeader(1)); if (conf->len > cap) { return RAFT_TOOBIG; } diff --git a/test/raft/integration/test_apply.c b/test/raft/integration/test_apply.c index 370ed09eb..650df5a93 100644 --- a/test/raft/integration/test_apply.c +++ b/test/raft/integration/test_apply.c @@ -71,7 +71,7 @@ static bool applyCbHasFired(struct raft_fixture *f, void *arg) int _rv; \ FsmEncodeSetX(N, &_buf); \ _req.data = &_result; \ - _rv = raft_apply(CLUSTER_RAFT(I), &_req, &_buf, NULL, 1, applyCbAssertResult); \ + _rv = raft_apply(CLUSTER_RAFT(I), &_req, &_buf, 1, applyCbAssertResult); \ munit_assert_int(_rv, ==, 0); /* Expect the apply callback to fire with the given status. */ @@ -96,7 +96,7 @@ static bool applyCbHasFired(struct raft_fixture *f, void *arg) struct raft_apply _req; \ int _rv; \ FsmEncodeSetX(123, &_buf); \ - _rv = raft_apply(CLUSTER_RAFT(I), &_req, &_buf, NULL, 1, NULL); \ + _rv = raft_apply(CLUSTER_RAFT(I), &_req, &_buf, 1, NULL); \ munit_assert_int(_rv, ==, RV); \ munit_assert_string_equal(CLUSTER_ERRMSG(I), ERRMSG); \ raft_free(_buf.base); \ diff --git a/test/raft/integration/test_fixture.c b/test/raft/integration/test_fixture.c index 419ddac9b..c693ea273 100644 --- a/test/raft/integration/test_fixture.c +++ b/test/raft/integration/test_fixture.c @@ -86,7 +86,7 @@ static void tearDown(void *data) struct raft_buffer buf; \ int rc; \ FsmEncodeAddX(1, &buf); \ - rc = raft_apply(GET(I), REQ, &buf, NULL, 1, NULL); \ + rc = raft_apply(GET(I), REQ, &buf, 1, NULL); \ munit_assert_int(rc, ==, 0); \ } #define STEP_UNTIL_APPLIED(INDEX) \ diff --git a/test/raft/integration/test_membership.c b/test/raft/integration/test_membership.c index 1edd2ac16..53d43aea9 100644 --- a/test/raft/integration/test_membership.c +++ b/test/raft/integration/test_membership.c @@ -90,7 +90,7 @@ struct result int _rv; \ FsmEncodeSetX(123, &_buf); \ _req.data = &_result; \ - _rv = raft_apply(CLUSTER_RAFT(I), &_req, &_buf, NULL, 1, NULL); \ + _rv = raft_apply(CLUSTER_RAFT(I), &_req, &_buf, 1, NULL); \ munit_assert_int(_rv, ==, 0); /****************************************************************************** diff --git a/test/raft/lib/cluster.h b/test/raft/lib/cluster.h index 8bd8d56a4..9d157f455 100644 --- a/test/raft/lib/cluster.h +++ b/test/raft/lib/cluster.h @@ -276,7 +276,7 @@ int rv_; \ FsmEncodeAddX(VALUE, &buf_); \ raft_ = raft_fixture_get(&f->cluster, I); \ - rv_ = raft_apply(raft_, REQ, &buf_, NULL, 1, CB); \ + rv_ = raft_apply(raft_, REQ, &buf_, 1, CB); \ munit_assert_int(rv_, ==, 0); \ } diff --git a/test/raft/unit/test_log.c b/test/raft/unit/test_log.c index f32bf2b0d..b5ed8945c 100644 --- a/test/raft/unit/test_log.c +++ b/test/raft/unit/test_log.c @@ -36,7 +36,7 @@ struct fixture buf_.base = raft_malloc(8); \ buf_.len = 8; \ strcpy(buf_.base, "hello"); \ - rv_ = logAppend(f->log, TERM, RAFT_COMMAND, buf_, (struct raft_entry_local_data){}, true, NULL); \ + rv_ = logAppend(f->log, TERM, RAFT_COMMAND, buf_, true, NULL); \ munit_assert_int(rv_, ==, 0); \ } @@ -56,7 +56,7 @@ struct fixture int rv_; \ buf_.base = raft_malloc(8); \ buf_.len = 8; \ - rv_ = logAppend(f->log, TERM, RAFT_COMMAND, buf_, (struct raft_entry_local_data){}, true, NULL); \ + rv_ = logAppend(f->log, TERM, RAFT_COMMAND, buf_, true, NULL); \ munit_assert_int(rv_, ==, RV); \ raft_free(buf_.base); \ } @@ -77,7 +77,7 @@ struct fixture buf.base = (uint8_t *)batch + offset; \ buf.len = 8; \ *(uint64_t *)buf.base = i * 1000; \ - rv = logAppend(f->log, 1, RAFT_COMMAND, buf, (struct raft_entry_local_data){}, true, batch); \ + rv = logAppend(f->log, 1, RAFT_COMMAND, buf, true, batch); \ munit_assert_int(rv, ==, 0); \ offset += 8; \ } \ @@ -609,7 +609,7 @@ TEST(logAppend, oom, setUp, tearDown, 0, logAppendOom) buf.base = NULL; buf.len = 0; HeapFaultEnable(&f->heap); - rv = logAppend(f->log, 1, RAFT_COMMAND, buf, (struct raft_entry_local_data){}, true, NULL); + rv = logAppend(f->log, 1, RAFT_COMMAND, buf, true, NULL); munit_assert_int(rv, ==, RAFT_NOMEM); return MUNIT_OK; } @@ -1032,7 +1032,7 @@ TEST(logTruncate, acquiredOom, setUp, tearDown, 0, logTruncateAcquiredOom) HeapFaultEnable(&f->heap); - rv = logAppend(f->log, 2, RAFT_COMMAND, buf, (struct raft_entry_local_data){}, true, NULL); + rv = logAppend(f->log, 2, RAFT_COMMAND, buf, true, NULL); munit_assert_int(rv, ==, RAFT_NOMEM); RELEASE(2);