Commit 30e4688b authored by Willem Thiart's avatar Willem Thiart
Browse files

Fix commit_idx being set by raft_apply_entry

parent 94048d4a
...@@ -288,10 +288,10 @@ int raft_recv_appendentries( ...@@ -288,10 +288,10 @@ int raft_recv_appendentries(
log_delete(me->log, ae->prev_log_idx + 1); log_delete(me->log, ae->prev_log_idx + 1);
/* 4. If leaderCommit > commitIndex, set commitIndex = /* 4. If leaderCommit > commitIndex, set commitIndex =
min(leaderCommit, last log index) */ min(leaderCommit, index of most recent entry) */
if (raft_get_commit_idx(me_) < ae->leader_commit) if (raft_get_commit_idx(me_) < ae->leader_commit)
{ {
int last_log_idx = max(raft_get_current_idx(me_) - 1, 1); int last_log_idx = max(raft_get_current_idx(me_), 1);
raft_set_commit_idx(me_, min(last_log_idx, ae->leader_commit)); raft_set_commit_idx(me_, min(last_log_idx, ae->leader_commit));
while (0 == raft_apply_entry(me_)) while (0 == raft_apply_entry(me_))
; ;
...@@ -469,6 +469,10 @@ int raft_apply_entry(raft_server_t* me_) ...@@ -469,6 +469,10 @@ 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_;
/* Don't apply after the commit_idx */
if (me->last_applied_idx == me->commit_idx)
return -1;
raft_entry_t* e = raft_get_entry_from_idx(me_, me->last_applied_idx + 1); raft_entry_t* e = raft_get_entry_from_idx(me_, me->last_applied_idx + 1);
if (!e) if (!e)
return -1; return -1;
...@@ -476,8 +480,6 @@ int raft_apply_entry(raft_server_t* me_) ...@@ -476,8 +480,6 @@ int raft_apply_entry(raft_server_t* me_)
__log(me_, "applying log: %d, size: %d", me->last_applied_idx, e->data.len); __log(me_, "applying log: %d, size: %d", me->last_applied_idx, e->data.len);
me->last_applied_idx++; me->last_applied_idx++;
if (me->commit_idx < me->last_applied_idx)
me->commit_idx = me->last_applied_idx;
if (me->cb.applylog) if (me->cb.applylog)
me->cb.applylog(me_, me->udata, e->data.buf, e->data.len); me->cb.applylog(me_, me->udata, e->data.buf, e->data.len);
return 0; return 0;
......
...@@ -230,10 +230,21 @@ void TestRaft_server_entry_is_retrieveable_using_idx(CuTest* tc) ...@@ -230,10 +230,21 @@ void TestRaft_server_entry_is_retrieveable_using_idx(CuTest* tc)
void TestRaft_server_wont_apply_entry_if_we_dont_have_entry_to_apply(CuTest* tc) void TestRaft_server_wont_apply_entry_if_we_dont_have_entry_to_apply(CuTest* tc)
{ {
raft_entry_t ety; void *r = raft_new();
char *str = "aaa"; raft_set_commit_idx(r, 0);
raft_set_last_applied_idx(r, 0);
raft_apply_entry(r);
CuAssertTrue(tc, 0 == raft_get_last_applied_idx(r));
CuAssertTrue(tc, 0 == raft_get_commit_idx(r));
}
void TestRaft_server_wont_apply_entry_if_there_isnt_a_majority(CuTest* tc)
{
void *r = raft_new(); void *r = raft_new();
raft_add_node(r, (void*)1, 1);
raft_add_node(r, (void*)2, 0);
raft_add_node(r, (void*)3, 0);
raft_set_commit_idx(r, 0); raft_set_commit_idx(r, 0);
raft_set_last_applied_idx(r, 0); raft_set_last_applied_idx(r, 0);
...@@ -241,14 +252,17 @@ void TestRaft_server_wont_apply_entry_if_we_dont_have_entry_to_apply(CuTest* tc) ...@@ -241,14 +252,17 @@ void TestRaft_server_wont_apply_entry_if_we_dont_have_entry_to_apply(CuTest* tc)
CuAssertTrue(tc, 0 == raft_get_last_applied_idx(r)); CuAssertTrue(tc, 0 == raft_get_last_applied_idx(r));
CuAssertTrue(tc, 0 == raft_get_commit_idx(r)); CuAssertTrue(tc, 0 == raft_get_commit_idx(r));
char *str = "aaa";
raft_entry_t ety;
ety.term = 1; ety.term = 1;
ety.id = 1; ety.id = 1;
ety.data.buf = str; ety.data.buf = str;
ety.data.len = 3; ety.data.len = 3;
raft_append_entry(r, &ety); raft_append_entry(r, &ety);
raft_apply_entry(r); raft_apply_entry(r);
CuAssertTrue(tc, 1 == raft_get_last_applied_idx(r)); /* Not allowed to be applied because we haven't confirmed a majority yet */
CuAssertTrue(tc, 1 == raft_get_commit_idx(r)); CuAssertTrue(tc, 0 == raft_get_last_applied_idx(r));
CuAssertTrue(tc, 0 == raft_get_commit_idx(r));
} }
/* If commitidx > lastApplied: increment lastApplied, apply log[lastApplied] /* If commitidx > lastApplied: increment lastApplied, apply log[lastApplied]
...@@ -826,7 +840,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx( ...@@ -826,7 +840,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx(
CuAssertTrue(tc, 1 == aer.success); CuAssertTrue(tc, 1 == aer.success);
/* set to 4 because commitIDX is lower */ /* set to 4 because commitIDX is lower */
CuAssertTrue(tc, 4 == raft_get_commit_idx(r)); CuAssertIntEquals(tc, 4, raft_get_commit_idx(r));
} }
void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit( void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit(
...@@ -869,7 +883,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit( ...@@ -869,7 +883,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit(
CuAssertTrue(tc, 1 == aer.success); CuAssertTrue(tc, 1 == aer.success);
/* set to 3 because leaderCommit is lower */ /* set to 3 because leaderCommit is lower */
CuAssertTrue(tc, 3 == raft_get_commit_idx(r)); CuAssertIntEquals(tc, 3, raft_get_commit_idx(r));
} }
void TestRaft_follower_becomes_candidate_when_election_timeout_occurs( void TestRaft_follower_becomes_candidate_when_election_timeout_occurs(
...@@ -1559,6 +1573,8 @@ TestRaft_leader_recv_appendentries_response_increase_commit_idx_when_majority_ha ...@@ -1559,6 +1573,8 @@ TestRaft_leader_recv_appendentries_response_increase_commit_idx_when_majority_ha
raft_append_entry(r, &ety); raft_append_entry(r, &ety);
ety.id = 2; ety.id = 2;
raft_append_entry(r, &ety); raft_append_entry(r, &ety);
ety.id = 3;
raft_append_entry(r, &ety);
memset(&aer, 0, sizeof(msg_appendentries_response_t)); memset(&aer, 0, sizeof(msg_appendentries_response_t));
...@@ -1624,7 +1640,6 @@ void TestRaft_leader_recv_appendentries_response_retry_only_if_leader(CuTest * t ...@@ -1624,7 +1640,6 @@ void TestRaft_leader_recv_appendentries_response_retry_only_if_leader(CuTest * t
ety.data.len = 4; ety.data.len = 4;
raft_append_entry(r, &ety); raft_append_entry(r, &ety);
raft_send_appendentries(r, 0); raft_send_appendentries(r, 0);
raft_send_appendentries(r, 1); raft_send_appendentries(r, 1);
......
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