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

Merge pull request #86 from yossigo/feat/snapshot-nonblocking-apply

Allow entries to apply during snapshot
parents ff0c70ae ed56e88d
...@@ -39,6 +39,9 @@ typedef enum { ...@@ -39,6 +39,9 @@ typedef enum {
RAFT_STATE_LEADER RAFT_STATE_LEADER
} raft_state_e; } raft_state_e;
/** Allow entries to apply while taking a snapshot */
#define RAFT_SNAPSHOT_NONBLOCKING_APPLY 1
typedef enum { typedef enum {
/** /**
* Regular log type. * Regular log type.
...@@ -802,10 +805,14 @@ int raft_entry_is_cfg_change(raft_entry_t* ety); ...@@ -802,10 +805,14 @@ int raft_entry_is_cfg_change(raft_entry_t* ety);
* - not apply log entries * - not apply log entries
* - not start elections * - not start elections
* *
* If the RAFT_SNAPSHOT_NONBLOCKING_APPLY flag is specified, log entries will
* be applied during snapshot. The FSM must isolate the snapshot state and
* guarantee these changes do not affect it.
*
* @return 0 on success * @return 0 on success
* *
**/ **/
int raft_begin_snapshot(raft_server_t *me_); int raft_begin_snapshot(raft_server_t *me_, int flags);
/** Stop snapshotting. /** Stop snapshotting.
* *
...@@ -841,6 +848,11 @@ raft_index_t raft_get_snapshot_entry_idx(raft_server_t *me_); ...@@ -841,6 +848,11 @@ raft_index_t raft_get_snapshot_entry_idx(raft_server_t *me_);
**/ **/
int raft_snapshot_is_in_progress(raft_server_t *me_); int raft_snapshot_is_in_progress(raft_server_t *me_);
/** Check if entries can be applied now (no snapshot in progress, or
* RAFT_SNAPSHOT_NONBLOCKING_APPLY specified).
**/
int raft_is_apply_allowed(raft_server_t* me_);
/** Remove the first log entry. /** Remove the first log entry.
* This should be used for compacting logs. * This should be used for compacting logs.
* @return 0 on success * @return 0 on success
......
...@@ -73,6 +73,7 @@ typedef struct { ...@@ -73,6 +73,7 @@ typedef struct {
int connected; int connected;
int snapshot_in_progress; int snapshot_in_progress;
int snapshot_flags;
/* Last compacted snapshot */ /* Last compacted snapshot */
raft_index_t snapshot_last_idx; raft_index_t snapshot_last_idx;
......
...@@ -248,7 +248,7 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period) ...@@ -248,7 +248,7 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period)
} }
if (me->last_applied_idx < raft_get_commit_idx(me_) && if (me->last_applied_idx < raft_get_commit_idx(me_) &&
!raft_snapshot_is_in_progress(me_)) raft_is_apply_allowed(me_))
{ {
int e = raft_apply_all(me_); int e = raft_apply_all(me_);
if (0 != e) if (0 != e)
...@@ -728,7 +728,7 @@ int raft_recv_entry(raft_server_t* me_, ...@@ -728,7 +728,7 @@ int raft_recv_entry(raft_server_t* me_,
/* Multi-threading: need to fail here because user might be /* Multi-threading: need to fail here because user might be
* snapshotting membership settings. */ * snapshotting membership settings. */
if (raft_snapshot_is_in_progress(me_)) if (!raft_is_apply_allowed(me_))
return RAFT_ERR_SNAPSHOT_IN_PROGRESS; return RAFT_ERR_SNAPSHOT_IN_PROGRESS;
} }
...@@ -810,7 +810,7 @@ int raft_apply_entry(raft_server_t* me_) ...@@ -810,7 +810,7 @@ int raft_apply_entry(raft_server_t* me_)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
if (raft_snapshot_is_in_progress(me_)) if (!raft_is_apply_allowed(me_))
return -1; return -1;
/* Don't apply after the commit_idx */ /* Don't apply after the commit_idx */
...@@ -1096,7 +1096,7 @@ int raft_msg_entry_response_committed(raft_server_t* me_, ...@@ -1096,7 +1096,7 @@ int raft_msg_entry_response_committed(raft_server_t* me_,
int raft_apply_all(raft_server_t* me_) int raft_apply_all(raft_server_t* me_)
{ {
if (raft_snapshot_is_in_progress(me_)) if (!raft_is_apply_allowed(me_))
return 0; return 0;
while (raft_get_last_applied_idx(me_) < raft_get_commit_idx(me_)) while (raft_get_last_applied_idx(me_) < raft_get_commit_idx(me_))
...@@ -1253,7 +1253,7 @@ raft_index_t raft_get_num_snapshottable_logs(raft_server_t *me_) ...@@ -1253,7 +1253,7 @@ raft_index_t raft_get_num_snapshottable_logs(raft_server_t *me_)
return raft_get_commit_idx(me_) - log_get_base(me->log); return raft_get_commit_idx(me_) - log_get_base(me->log);
} }
int raft_begin_snapshot(raft_server_t *me_) int raft_begin_snapshot(raft_server_t *me_, int flags)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
...@@ -1277,6 +1277,7 @@ int raft_begin_snapshot(raft_server_t *me_) ...@@ -1277,6 +1277,7 @@ int raft_begin_snapshot(raft_server_t *me_)
raft_set_snapshot_metadata(me_, ety->term, snapshot_target); raft_set_snapshot_metadata(me_, ety->term, snapshot_target);
me->snapshot_in_progress = 1; me->snapshot_in_progress = 1;
me->snapshot_flags = flags;
__log(me_, NULL, __log(me_, NULL,
"begin snapshot sli:%d slt:%d slogs:%d\n", "begin snapshot sli:%d slt:%d slogs:%d\n",
......
...@@ -235,6 +235,12 @@ int raft_snapshot_is_in_progress(raft_server_t *me_) ...@@ -235,6 +235,12 @@ int raft_snapshot_is_in_progress(raft_server_t *me_)
return ((raft_server_private_t*)me_)->snapshot_in_progress; return ((raft_server_private_t*)me_)->snapshot_in_progress;
} }
int raft_is_apply_allowed(raft_server_t* me_)
{
return (!raft_snapshot_is_in_progress(me_) ||
(((raft_server_private_t*)me_)->snapshot_flags & RAFT_SNAPSHOT_NONBLOCKING_APPLY));
}
raft_entry_t *raft_get_last_applied_entry(raft_server_t *me_) raft_entry_t *raft_get_last_applied_entry(raft_server_t *me_)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
......
...@@ -108,10 +108,10 @@ void TestRaft_leader_begin_snapshot_fails_if_no_logs_to_compact(CuTest * tc) ...@@ -108,10 +108,10 @@ void TestRaft_leader_begin_snapshot_fails_if_no_logs_to_compact(CuTest * tc)
ety.id = 2; ety.id = 2;
raft_recv_entry(r, &ety, &cr); raft_recv_entry(r, &ety, &cr);
CuAssertIntEquals(tc, 2, raft_get_log_count(r)); CuAssertIntEquals(tc, 2, raft_get_log_count(r));
CuAssertIntEquals(tc, -1, raft_begin_snapshot(r)); CuAssertIntEquals(tc, -1, raft_begin_snapshot(r, 0));
raft_set_commit_idx(r, 1); raft_set_commit_idx(r, 1);
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r)); CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
} }
void TestRaft_leader_will_not_apply_entry_if_snapshot_is_in_progress(CuTest * tc) void TestRaft_leader_will_not_apply_entry_if_snapshot_is_in_progress(CuTest * tc)
...@@ -145,7 +145,7 @@ void TestRaft_leader_will_not_apply_entry_if_snapshot_is_in_progress(CuTest * tc ...@@ -145,7 +145,7 @@ void TestRaft_leader_will_not_apply_entry_if_snapshot_is_in_progress(CuTest * tc
raft_set_commit_idx(r, 1); raft_set_commit_idx(r, 1);
CuAssertIntEquals(tc, 2, raft_get_log_count(r)); CuAssertIntEquals(tc, 2, raft_get_log_count(r));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r)); CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
CuAssertIntEquals(tc, 1, raft_get_last_applied_idx(r)); CuAssertIntEquals(tc, 1, raft_get_last_applied_idx(r));
raft_set_commit_idx(r, 2); raft_set_commit_idx(r, 2);
CuAssertIntEquals(tc, -1, raft_apply_entry(r)); CuAssertIntEquals(tc, -1, raft_apply_entry(r));
...@@ -198,7 +198,7 @@ void TestRaft_leader_snapshot_begin_fails_if_less_than_2_logs_to_compact(CuTest ...@@ -198,7 +198,7 @@ void TestRaft_leader_snapshot_begin_fails_if_less_than_2_logs_to_compact(CuTest
raft_recv_entry(r, &ety, &cr); raft_recv_entry(r, &ety, &cr);
raft_set_commit_idx(r, 1); raft_set_commit_idx(r, 1);
CuAssertIntEquals(tc, 1, raft_get_log_count(r)); CuAssertIntEquals(tc, 1, raft_get_log_count(r));
CuAssertIntEquals(tc, -1, raft_begin_snapshot(r)); CuAssertIntEquals(tc, -1, raft_begin_snapshot(r, 0));
} }
void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc) void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc)
...@@ -235,7 +235,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc) ...@@ -235,7 +235,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc)
CuAssertIntEquals(tc, 2, raft_get_log_count(r)); CuAssertIntEquals(tc, 2, raft_get_log_count(r));
CuAssertIntEquals(tc, 1, raft_get_num_snapshottable_logs(r)); CuAssertIntEquals(tc, 1, raft_get_num_snapshottable_logs(r));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r)); CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
raft_entry_t* _ety; raft_entry_t* _ety;
int i = raft_get_first_entry_idx(r); int i = raft_get_first_entry_idx(r);
...@@ -286,7 +286,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted2(CuTest * tc) ...@@ -286,7 +286,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted2(CuTest * tc)
CuAssertIntEquals(tc, 3, raft_get_log_count(r)); CuAssertIntEquals(tc, 3, raft_get_log_count(r));
CuAssertIntEquals(tc, 2, raft_get_num_snapshottable_logs(r)); CuAssertIntEquals(tc, 2, raft_get_num_snapshottable_logs(r));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r)); CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
raft_entry_t* _ety; raft_entry_t* _ety;
int i = raft_get_first_entry_idx(r); int i = raft_get_first_entry_idx(r);
...@@ -333,7 +333,7 @@ void TestRaft_joinee_needs_to_get_snapshot(CuTest * tc) ...@@ -333,7 +333,7 @@ void TestRaft_joinee_needs_to_get_snapshot(CuTest * tc)
CuAssertIntEquals(tc, 2, raft_get_log_count(r)); CuAssertIntEquals(tc, 2, raft_get_log_count(r));
CuAssertIntEquals(tc, 1, raft_get_num_snapshottable_logs(r)); CuAssertIntEquals(tc, 1, raft_get_num_snapshottable_logs(r));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r)); CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
CuAssertIntEquals(tc, 1, raft_get_last_applied_idx(r)); CuAssertIntEquals(tc, 1, raft_get_last_applied_idx(r));
CuAssertIntEquals(tc, -1, raft_apply_entry(r)); CuAssertIntEquals(tc, -1, raft_apply_entry(r));
CuAssertIntEquals(tc, 1, raft_get_last_applied_idx(r)); CuAssertIntEquals(tc, 1, raft_get_last_applied_idx(r));
...@@ -572,13 +572,52 @@ void TestRaft_recv_entry_fails_if_snapshot_in_progress(CuTest* tc) ...@@ -572,13 +572,52 @@ void TestRaft_recv_entry_fails_if_snapshot_in_progress(CuTest* tc)
CuAssertIntEquals(tc, 2, raft_get_log_count(r)); CuAssertIntEquals(tc, 2, raft_get_log_count(r));
raft_set_commit_idx(r, 1); raft_set_commit_idx(r, 1);
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r)); CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
ety.id = 3; ety.id = 3;
ety.type = RAFT_LOGTYPE_ADD_NODE; ety.type = RAFT_LOGTYPE_ADD_NODE;
CuAssertIntEquals(tc, RAFT_ERR_SNAPSHOT_IN_PROGRESS, raft_recv_entry(r, &ety, &cr)); CuAssertIntEquals(tc, RAFT_ERR_SNAPSHOT_IN_PROGRESS, raft_recv_entry(r, &ety, &cr));
} }
void TestRaft_recv_entry_succeeds_if_snapshot_nonblocking_apply_is_set(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));
/* entry message */
msg_entry_t ety = {};
ety.id = 1;
ety.data.buf = "entry";
ety.data.len = strlen("entry");
/* receive entry */
raft_recv_entry(r, &ety, &cr);
ety.id = 2;
raft_recv_entry(r, &ety, &cr);
CuAssertIntEquals(tc, 2, raft_get_log_count(r));
raft_set_commit_idx(r, 1);
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, RAFT_SNAPSHOT_NONBLOCKING_APPLY));
ety.id = 3;
ety.type = RAFT_LOGTYPE_ADD_NODE;
CuAssertIntEquals(tc, 0, raft_recv_entry(r, &ety, &cr));
}
void TestRaft_follower_recv_appendentries_is_successful_when_previous_log_idx_equals_snapshot_last_idx( void TestRaft_follower_recv_appendentries_is_successful_when_previous_log_idx_equals_snapshot_last_idx(
CuTest * tc) CuTest * tc)
{ {
......
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