Unverified Commit 7d66a156 authored by Ozan Tezcan's avatar Ozan Tezcan Committed by GitHub
Browse files

Improve snapshot support (#121)

- raft_load_snapshot_f(): Changed order of arguments term and index
to be inline with raft_begin_load_snapshot().

- Refactored raft_begin_load_snapshot(). Now, inside this function, 
current server's node flags are cleared. Previously, application had to
take care of it. e.g Application had to clear inactive flag on a snapshot.

- Added raft_restore_snapshot() to let libraft know that application has
loaded a snapshot.
parent 5602930c
......@@ -25,7 +25,8 @@ typedef enum {
RAFT_ERR_LEADER_TRANSFER_IN_PROGRESS = -9,
RAFT_ERR_DONE = -10,
RAFT_ERR_STALE_TERM = -11,
RAFT_ERR_NOTFOUND = -12
RAFT_ERR_NOTFOUND = -12,
RAFT_ERR_MISUSE = -13,
} raft_error_e;
typedef enum {
......@@ -359,16 +360,16 @@ typedef int (
*
* @param[in] raft Raft server making this callback
* @param[in] user_data User data that is passed from Raft server
* @param[in] snapshot_index Received snapshot index
* @param[in] snapshot_term Received snapshot term
* @param[in] snapshot_index Received snapshot index
* @return 0 on success */
typedef int (
*raft_load_snapshot_f
) (
raft_server_t* raft,
void *user_data,
raft_index_t snapshot_index,
raft_term_t snapshot_term
raft_term_t snapshot_term,
raft_index_t snapshot_index
);
/** Callback to get a chunk from the snapshot file. This chunk will be sent
......@@ -1379,6 +1380,26 @@ raft_entry_t *raft_get_last_applied_entry(raft_server_t *me);
raft_index_t raft_get_first_entry_idx(raft_server_t* me);
/** Restore the snapshot after a restart.
*
* This function should be called on a restart just after application restores
* the snapshot and configures the nodes from the snapshot.
*
* Correct order to restore the state after a restart:
* 1- Restore the snapshot
* 2- Load log entries
* 3- Restore metadata
*
* @param[in] last_included_term Term of the last log of the snapshot
* @param[in] last_included_index Index of the last log of the snapshot
* @return
* 0 on success
* RAFT_ERR_MISUSE if the server has already started
*/
int raft_restore_snapshot(raft_server_t *me,
raft_term_t last_included_term,
raft_index_t last_included_index);
/** Start loading snapshot
*
* This is usually the result of a snapshot being loaded.
......@@ -1392,12 +1413,12 @@ raft_index_t raft_get_first_entry_idx(raft_server_t* me);
*
* @return
* 0 on success
* -1 on failure
* RAFT_ERR_MISUSE
* RAFT_ERR_SNAPSHOT_ALREADY_LOADED
**/
int raft_begin_load_snapshot(raft_server_t *me,
raft_term_t last_included_term,
raft_index_t last_included_index);
raft_term_t last_included_term,
raft_index_t last_included_index);
/** Stop loading snapshot.
*
......
......@@ -166,11 +166,13 @@ void raft_node_free(raft_node_t* me);
void raft_node_set_match_idx(raft_node_t* node, raft_index_t idx);
void raft_node_clear_flags(raft_node_t *me);
void raft_node_vote_for_me(raft_node_t* me, int vote);
int raft_node_has_vote_for_me(raft_node_t* me);
void raft_node_set_has_sufficient_logs(raft_node_t* me);
void raft_node_set_has_sufficient_logs(raft_node_t *me, int has_sufficient_log);
int raft_is_single_node_voting_cluster(raft_server_t *me);
......
......@@ -125,6 +125,11 @@ void raft_node_set_udata(raft_node_t* me, void* udata)
me->udata = udata;
}
void raft_node_clear_flags(raft_node_t *me)
{
me->flags = 0;
}
void raft_node_vote_for_me(raft_node_t* me, const int vote)
{
if (vote)
......@@ -138,16 +143,11 @@ int raft_node_has_vote_for_me(raft_node_t* me)
return (me->flags & RAFT_NODE_VOTED_FOR_ME) != 0;
}
void raft_node_set_voting(raft_node_t* me, int voting)
void raft_node_set_voting(raft_node_t *me, int voting)
{
if (voting)
{
assert(!raft_node_is_voting(me));
if (voting) {
me->flags |= RAFT_NODE_VOTING;
}
else
{
assert(raft_node_is_voting(me));
} else {
me->flags &= ~RAFT_NODE_VOTING;
}
}
......@@ -162,9 +162,13 @@ int raft_node_has_sufficient_logs(raft_node_t* me)
return (me->flags & RAFT_NODE_HAS_SUFFICIENT_LOG) != 0;
}
void raft_node_set_has_sufficient_logs(raft_node_t* me)
void raft_node_set_has_sufficient_logs(raft_node_t *me, int has_sufficient_log)
{
me->flags |= RAFT_NODE_HAS_SUFFICIENT_LOG;
if (has_sufficient_log) {
me->flags |= RAFT_NODE_HAS_SUFFICIENT_LOG;
} else {
me->flags &= ~RAFT_NODE_HAS_SUFFICIENT_LOG;
}
}
void raft_node_set_active(raft_node_t* me, int active)
......
......@@ -730,8 +730,9 @@ int raft_recv_appendentries_response(raft_server_t* me,
)
{
int e = me->cb.node_has_sufficient_logs(me, me->udata, node);
if (0 == e)
raft_node_set_has_sufficient_logs(node);
if (e == 0) {
raft_node_set_has_sufficient_logs(node, 1);
}
}
raft_index_t match_idx = raft_node_get_match_idx(node);
......@@ -1227,7 +1228,7 @@ int raft_apply_entry(raft_server_t* me)
case RAFT_LOGTYPE_ADD_NODE:
raft_node_set_addition_committed(node, 1);
raft_node_set_voting_committed(node, 1);
raft_node_set_has_sufficient_logs(node);
raft_node_set_has_sufficient_logs(node, 1);
break;
case RAFT_LOGTYPE_ADD_NONVOTING_NODE:
raft_node_set_addition_committed(node, 1);
......@@ -1416,8 +1417,8 @@ int raft_recv_snapshot(raft_server_t* me,
me->snapshot_recv_offset = req->chunk.offset + req->chunk.len;
if (req->chunk.last_chunk) {
e = me->cb.load_snapshot(me, me->udata,
req->snapshot_index, req->snapshot_term);
e = me->cb.load_snapshot(me, me->udata, req->snapshot_term,
req->snapshot_index);
if (e != 0) {
goto out;
}
......@@ -1701,6 +1702,39 @@ raft_index_t raft_get_num_snapshottable_logs(raft_server_t *me)
return raft_get_commit_idx(me) - me->log_impl->first_idx(me->log) + 1;
}
int raft_restore_snapshot(raft_server_t *me,
raft_term_t last_included_term,
raft_index_t last_included_index)
{
if (last_included_term < 0 || last_included_index < 0 ||
me->current_term != 0 || me->commit_idx != 0 ||
me->snapshot_last_term != 0 || me->snapshot_last_idx != 0 ||
me->last_applied_term != 0 || me->last_applied_idx != 0) {
return RAFT_ERR_MISUSE;
}
me->current_term = last_included_term;
me->commit_idx = last_included_index;
me->last_applied_term = last_included_term;
me->last_applied_idx = last_included_index;
me->snapshot_last_term = last_included_term;
me->snapshot_last_idx = last_included_index;
/* Adjust node flags as they are configured by `voting` data only. */
for (int i = 0; i < me->num_nodes; i++) {
int voting = raft_node_is_voting(me->nodes[i]);
raft_node_set_has_sufficient_logs(me->nodes[i], voting);
raft_node_set_voting_committed(me->nodes[i], voting);
raft_node_set_addition_committed(me->nodes[i], 1);
}
raft_log(me, "restored snapshot lii:%ld lit:%ld",
last_included_index, last_included_term);
return 0;
}
int raft_begin_snapshot(raft_server_t *me)
{
raft_index_t entry_count = raft_get_num_snapshottable_logs(me);
......@@ -1780,40 +1814,30 @@ int raft_end_snapshot(raft_server_t *me)
return 0;
}
int raft_begin_load_snapshot(
raft_server_t *me,
raft_term_t last_included_term,
raft_index_t last_included_index)
int raft_begin_load_snapshot(raft_server_t *me,
raft_term_t last_included_term,
raft_index_t last_included_index)
{
if (last_included_index == -1)
return -1;
if (last_included_index == 0 || last_included_term == 0)
return -1;
/* loading the snapshot will break cluster safety */
if (last_included_index < me->last_applied_idx)
return -1;
/* snapshot was unnecessary */
if (last_included_index < raft_get_current_idx(me))
return -1;
if (last_included_term <= 0 ||
last_included_index <= 0 ||
last_included_index < me->last_applied_idx ||
last_included_index < raft_get_current_idx(me)) {
return RAFT_ERR_MISUSE;
}
if (last_included_index <= me->snapshot_last_idx)
if (last_included_index <= me->snapshot_last_idx) {
return RAFT_ERR_SNAPSHOT_ALREADY_LOADED;
if (me->current_term < last_included_term) {
raft_set_current_term(me, last_included_term);
me->current_term = last_included_term;
}
raft_set_state(me, RAFT_STATE_FOLLOWER);
me->leader_id = RAFT_NODE_ID_NONE;
me->log_impl->reset(me->log, last_included_index + 1, last_included_term);
if (raft_get_commit_idx(me) < last_included_index)
if (last_included_index > me->commit_idx) {
raft_set_commit_idx(me, last_included_index);
}
if (last_included_term > me->current_term) {
raft_set_current_term(me, last_included_term);
}
me->last_applied_term = last_included_term;
me->last_applied_idx = last_included_index;
......@@ -1821,47 +1845,43 @@ int raft_begin_load_snapshot(
me->next_snapshot_last_idx = last_included_index;
/* remove all nodes but self */
int i, my_node_by_idx = 0;
for (i = 0; i < me->num_nodes; i++)
{
if (raft_get_nodeid(me) == raft_node_get_id(me->nodes[i]))
my_node_by_idx = i;
else {
raft_node_free(me->nodes[i]);
me->nodes[i] = NULL;
for (int i = 0; i < me->num_nodes; i++) {
if (me->nodes[i] == me->node) {
continue;
}
raft_node_free(me->nodes[i]);
}
/* this will be realloc'd by a raft_add_node */
me->nodes[0] = me->nodes[my_node_by_idx];
raft_node_clear_flags(me->node);
me->num_nodes = 1;
me->nodes = raft_realloc(me->nodes, sizeof(*me->nodes) * me->num_nodes);
me->nodes[0] = me->node;
raft_log(me,
"loaded snapshot sli:%ld slt:%ld slogs:%ld",
me->snapshot_last_idx,
me->snapshot_last_term,
raft_get_num_snapshottable_logs(me));
raft_log(me, "begin loading snapshot lii:%ld lit:%ld",
last_included_index, last_included_term);
return 0;
}
int raft_end_load_snapshot(raft_server_t *me)
{
int i;
me->snapshot_last_idx = me->next_snapshot_last_idx;
me->snapshot_last_term = me->next_snapshot_last_term;
/* Set nodes' voting status as committed */
for (i = 0; i < me->num_nodes; i++)
{
raft_node_t* node = me->nodes[i];
raft_node_set_voting_committed(node, raft_node_is_voting(node));
raft_node_set_addition_committed(node, 1);
if (raft_node_is_voting(node))
raft_node_set_has_sufficient_logs(node);
/* Adjust node flags as they are configured by `voting` data only. */
for (int i = 0; i < me->num_nodes; i++) {
int voting = raft_node_is_voting(me->nodes[i]);
raft_node_set_has_sufficient_logs(me->nodes[i], voting);
raft_node_set_voting_committed(me->nodes[i], voting);
raft_node_set_addition_committed(me->nodes[i], 1);
}
raft_log(me, "loaded snapshot sli:%ld slt:%ld",
me->snapshot_last_idx, me->snapshot_last_term);
return 0;
}
......
......@@ -174,8 +174,8 @@ static int test_clear_snapshot(raft_server_t* raft,
static int test_load_snapshot(raft_server_t* raft,
void *user_data,
raft_index_t snapshot_index,
raft_term_t snapshot_term)
raft_term_t snapshot_term,
raft_index_t snapshot_index)
{
struct test_data *t = user_data;
t->load++;
......@@ -487,7 +487,7 @@ void TestRaft_follower_load_from_snapshot(CuTest * tc)
ety = __MAKE_ENTRY(3, 1, "entry");
raft_append_entry(r, ety);
raft_set_commit_idx(r, 7);
CuAssertIntEquals(tc, -1, raft_begin_load_snapshot(r, 6, 5));
CuAssertIntEquals(tc, RAFT_ERR_MISUSE, raft_begin_load_snapshot(r, 6, 5));
CuAssertIntEquals(tc, 7, raft_get_commit_idx(r));
}
......@@ -504,7 +504,7 @@ void TestRaft_follower_load_from_snapshot_fails_if_term_is_0(CuTest * tc)
raft_set_state(r, RAFT_STATE_FOLLOWER);
CuAssertIntEquals(tc, 0, raft_get_log_count(r));
CuAssertIntEquals(tc, -1, raft_begin_load_snapshot(r, 0, 5));
CuAssertIntEquals(tc, RAFT_ERR_MISUSE, raft_begin_load_snapshot(r, 0, 5));
}
void TestRaft_follower_load_from_snapshot_fails_if_already_loaded(CuTest * tc)
......@@ -553,7 +553,7 @@ void TestRaft_follower_load_from_snapshot_does_not_break_cluster_safety(CuTest *
__RAFT_APPEND_ENTRY(r, 2, 1, "entry");
__RAFT_APPEND_ENTRY(r, 3, 1, "entry");
CuAssertIntEquals(tc, -1, raft_begin_load_snapshot(r, 2, 2));
CuAssertIntEquals(tc, RAFT_ERR_MISUSE, raft_begin_load_snapshot(r, 2, 2));
}
void TestRaft_follower_load_from_snapshot_fails_if_log_is_newer(CuTest * tc)
......@@ -572,7 +572,7 @@ void TestRaft_follower_load_from_snapshot_fails_if_log_is_newer(CuTest * tc)
raft_set_last_applied_idx(r, 5);
CuAssertIntEquals(tc, -1, raft_begin_load_snapshot(r, 2, 2));
CuAssertIntEquals(tc, RAFT_ERR_MISUSE, raft_begin_load_snapshot(r, 2, 2));
}
void TestRaft_leader_sends_snapshot_when_node_next_index_was_compacted(CuTest* tc)
......@@ -1012,6 +1012,23 @@ void TestRaft_set_last_chunk_on_duplicate(CuTest * tc)
CuAssertIntEquals(tc, 1, resp.last_chunk);
}
void TestRaft_restore_after_restart(CuTest * tc)
{
raft_server_t *r = raft_new();
raft_add_node(r, NULL, 1, 1);
/* Restore should fail if the term is not 0 */
raft_set_current_term(r, 10);
CuAssertIntEquals(tc, RAFT_ERR_MISUSE, raft_restore_snapshot(r, -1, 3));
CuAssertIntEquals(tc, RAFT_ERR_MISUSE, raft_restore_snapshot(r, 11, 5));
r = raft_new();
raft_add_node(r, NULL, 1, 1);
/* Restore should work on start-up */
CuAssertIntEquals(tc, 0, raft_restore_snapshot(r, 11, 5));
}
int main(void)
{
CuString *output = CuStringNew();
......@@ -1039,6 +1056,7 @@ int main(void)
SUITE_ADD_TEST(suite, TestRaft_clear_snapshot_on_leader_change);
SUITE_ADD_TEST(suite, TestRaft_reject_wrong_offset);
SUITE_ADD_TEST(suite, TestRaft_set_last_chunk_on_duplicate);
SUITE_ADD_TEST(suite, TestRaft_restore_after_restart);
CuSuiteRun(suite);
CuSuiteDetails(suite, output);
......
......@@ -189,8 +189,8 @@ def raft_send_snapshot(raft, udata, node, msg):
return 0
def raft_load_snapshot(raft, udata, index, term):
return ffi.from_handle(udata).load_snapshot(index, term)
def raft_load_snapshot(raft, udata, term, index):
return ffi.from_handle(udata).load_snapshot(term, index)
def raft_clear_snapshot(raft, udata):
......@@ -949,7 +949,7 @@ class RaftServer(object):
self.raft_send_requestvote = ffi.callback("int(raft_server_t*, void*, raft_node_t*, raft_requestvote_req_t*)", raft_send_requestvote)
self.raft_send_appendentries = ffi.callback("int(raft_server_t*, void*, raft_node_t*, raft_appendentries_req_t*)", raft_send_appendentries)
self.raft_send_snapshot = ffi.callback("int(raft_server_t*, void* , raft_node_t*, raft_snapshot_req_t*)", raft_send_snapshot)
self.raft_load_snapshot = ffi.callback("int(raft_server_t*, void*, raft_index_t, raft_term_t)", raft_load_snapshot)
self.raft_load_snapshot = ffi.callback("int(raft_server_t*, void*, raft_term_t, raft_index_t)", raft_load_snapshot)
self.raft_clear_snapshot = ffi.callback("int(raft_server_t*, void*)", raft_clear_snapshot)
self.raft_get_snapshot_chunk = ffi.callback("int(raft_server_t*, void*, raft_node_t*, raft_size_t offset, raft_snapshot_chunk_t*)", raft_get_snapshot_chunk)
self.raft_store_snapshot_chunk = ffi.callback("int(raft_server_t*, void*, raft_index_t index, raft_size_t offset, raft_snapshot_chunk_t*)", raft_store_snapshot_chunk)
......@@ -1087,7 +1087,7 @@ class RaftServer(object):
return 0
def load_snapshot(self, index, term):
def load_snapshot(self, term, index):
logger.debug('{} loading snapshot'.format(self))
leader = find_leader()
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment