Unverified Commit ff0c70ae authored by Willem's avatar Willem Committed by GitHub
Browse files

Merge pull request #85 from yossigo/feat/cancel-snapshot

Add raft_cancel_snapshot().
parents cc90aca1 ca3c0b0e
...@@ -823,6 +823,16 @@ int raft_begin_snapshot(raft_server_t *me_); ...@@ -823,6 +823,16 @@ int raft_begin_snapshot(raft_server_t *me_);
**/ **/
int raft_end_snapshot(raft_server_t *me_); int raft_end_snapshot(raft_server_t *me_);
/** Cancel snapshotting.
*
* If an error occurs during snapshotting, this function can be called instead
* of raft_end_snapshot() to cancel the operation.
*
* The user MUST be sure the original snapshot is left untouched and remains
* usable.
*/
int raft_cancel_snapshot(raft_server_t *me_);
/** Get the entry index of the entry that was snapshotted /** Get the entry index of the entry that was snapshotted
**/ **/
raft_index_t raft_get_snapshot_entry_idx(raft_server_t *me_); raft_index_t raft_get_snapshot_entry_idx(raft_server_t *me_);
......
...@@ -77,6 +77,12 @@ typedef struct { ...@@ -77,6 +77,12 @@ typedef struct {
/* Last compacted snapshot */ /* Last compacted snapshot */
raft_index_t snapshot_last_idx; raft_index_t snapshot_last_idx;
raft_term_t snapshot_last_term; raft_term_t snapshot_last_term;
/* Previous index/term values stored during snapshot,
* which are restored if the operation is cancelled.
*/
raft_index_t saved_snapshot_last_idx;
raft_term_t saved_snapshot_last_term;
} raft_server_private_t; } raft_server_private_t;
int raft_election_start(raft_server_t* me); int raft_election_start(raft_server_t* me);
......
...@@ -1287,6 +1287,21 @@ int raft_begin_snapshot(raft_server_t *me_) ...@@ -1287,6 +1287,21 @@ int raft_begin_snapshot(raft_server_t *me_)
return 0; return 0;
} }
int raft_cancel_snapshot(raft_server_t *me_)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
if (!me->snapshot_in_progress)
return -1;
me->snapshot_last_idx = me->saved_snapshot_last_idx;
me->snapshot_last_term = me->saved_snapshot_last_term;
me->snapshot_in_progress = 0;
return 0;
}
int raft_end_snapshot(raft_server_t *me_) int raft_end_snapshot(raft_server_t *me_)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
...@@ -1335,8 +1350,6 @@ int raft_end_snapshot(raft_server_t *me_) ...@@ -1335,8 +1350,6 @@ int raft_end_snapshot(raft_server_t *me_)
} }
} }
assert(raft_get_log_count(me_) == 1);
return 0; return 0;
} }
......
...@@ -256,6 +256,8 @@ raft_term_t raft_get_snapshot_last_term(raft_server_t *me_) ...@@ -256,6 +256,8 @@ raft_term_t raft_get_snapshot_last_term(raft_server_t *me_)
void raft_set_snapshot_metadata(raft_server_t *me_, raft_term_t term, raft_index_t idx) void raft_set_snapshot_metadata(raft_server_t *me_, raft_term_t term, raft_index_t idx)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
me->saved_snapshot_last_term = me->snapshot_last_term;
me->saved_snapshot_last_idx = me->snapshot_last_idx;
me->snapshot_last_term = term; me->snapshot_last_term = term;
me->snapshot_last_idx = idx; me->snapshot_last_idx = idx;
} }
...@@ -644,3 +644,55 @@ void TestRaft_leader_sends_appendentries_with_correct_prev_log_idx_when_snapshot ...@@ -644,3 +644,55 @@ void TestRaft_leader_sends_appendentries_with_correct_prev_log_idx_when_snapshot
CuAssertIntEquals(tc, 2, ae->prev_log_term); CuAssertIntEquals(tc, 2, ae->prev_log_term);
CuAssertIntEquals(tc, 4, ae->prev_log_idx); CuAssertIntEquals(tc, 4, ae->prev_log_idx);
} }
void TestRaft_cancel_snapshot_restores_state(CuTest* tc)
{
raft_cbs_t funcs = {
.send_appendentries = __raft_send_appendentries,
};
void *r = raft_new();
raft_set_callbacks(r, &funcs, NULL);
msg_entry_response_t cr;
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
/* I am the leader */
raft_set_state(r, RAFT_STATE_LEADER);
CuAssertIntEquals(tc, 0, raft_get_log_count(r));
/* single entry */
msg_entry_t ety = {};
ety.id = 1;
ety.data.buf = "entry";
ety.data.len = strlen("entry");
raft_recv_entry(r, &ety, &cr);
ety.id = 2;
raft_recv_entry(r, &ety, &cr);
raft_set_commit_idx(r, 2);
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r));
CuAssertIntEquals(tc, 0, raft_end_snapshot(r));
/* more entries */
ety.id = 3;
raft_recv_entry(r, &ety, &cr);
ety.id = 4;
raft_recv_entry(r, &ety, &cr);
CuAssertIntEquals(tc, 2, raft_get_log_count(r));
CuAssertIntEquals(tc, 2, raft_get_snapshot_last_idx(r));
/* begin and cancel another snapshot */
raft_set_commit_idx(r, 4);
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r));
CuAssertIntEquals(tc, 1, raft_snapshot_is_in_progress(r));
CuAssertIntEquals(tc, 0, raft_cancel_snapshot(r));
/* snapshot no longer in progress, index must not have changed */
CuAssertIntEquals(tc, 0, raft_snapshot_is_in_progress(r));
CuAssertIntEquals(tc, 2, raft_get_snapshot_last_idx(r));
}
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