Commit b7324a87 authored by Willem Thiart's avatar Willem Thiart
Browse files

Recv appendentries_response next_idx jumps back

To speed up deciding on the correct next_idx, the leader will now use
the current_idx provided by the peer node in deciding the next_idx.

Also, the next_idx is set to the leader's current_idx.
parent 0baa4b07
...@@ -89,13 +89,14 @@ void raft_become_leader(raft_server_t* me_) ...@@ -89,13 +89,14 @@ void raft_become_leader(raft_server_t* me_)
__log(me_, "becoming leader"); __log(me_, "becoming leader");
raft_set_state(me_, RAFT_STATE_LEADER); raft_set_state(me_, RAFT_STATE_LEADER);
for (i = 0; i < me->num_nodes; i++) for (i = 0; i < me->num_nodes; i++)
{ {
if (me->nodeid != i) if (me->nodeid != i)
{ {
raft_node_t* p = raft_get_node(me_, i); raft_node_t* node = raft_get_node(me_, i);
raft_node_set_next_idx(p, raft_get_current_idx(me_) + 1); raft_node_set_next_idx(node, raft_get_current_idx(me_));
raft_node_set_match_idx(p, 0); raft_node_set_match_idx(node, 0);
raft_send_appendentries(me_, i); raft_send_appendentries(me_, i);
} }
} }
...@@ -161,19 +162,19 @@ raft_entry_t* raft_get_entry_from_idx(raft_server_t* me_, int etyidx) ...@@ -161,19 +162,19 @@ raft_entry_t* raft_get_entry_from_idx(raft_server_t* me_, int etyidx)
} }
int raft_recv_appendentries_response(raft_server_t* me_, int raft_recv_appendentries_response(raft_server_t* me_,
int node, msg_appendentries_response_t* r) int node_idx, msg_appendentries_response_t* r)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
__log(me_, __log(me_,
"received appendentries response node: %d %s cidx: %d 1stidx: %d", "received appendentries response node: %d %s cidx: %d 1stidx: %d",
node, r->success == 1 ? "success" : "fail", r->current_idx, node_idx, r->success == 1 ? "success" : "fail", r->current_idx,
r->first_idx); r->first_idx);
if (!raft_is_leader(me_)) if (!raft_is_leader(me_))
return -1; return -1;
raft_node_t* p = raft_get_node(me_, node); raft_node_t* node = raft_get_node(me_, node_idx);
/* If response contains term T > currentTerm: set currentTerm = T /* If response contains term T > currentTerm: set currentTerm = T
and convert to follower (§5.3) */ and convert to follower (§5.3) */
...@@ -188,35 +189,35 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -188,35 +189,35 @@ int raft_recv_appendentries_response(raft_server_t* me_,
{ {
/* If AppendEntries fails because of log inconsistency: /* If AppendEntries fails because of log inconsistency:
decrement nextIndex and retry (§5.3) */ decrement nextIndex and retry (§5.3) */
assert(0 <= raft_node_get_next_idx(p)); assert(0 <= raft_node_get_next_idx(node));
// TODO can jump back to where node is different instead of iterating // TODO can jump back to where node is different instead of iterating
raft_node_set_next_idx(p, raft_node_get_next_idx(p) - 1); raft_node_set_next_idx(node, r->current_idx + 1);
/* retry */ /* retry */
raft_send_appendentries(me_, node); raft_send_appendentries(me_, node_idx);
return 0; return 0;
} }
/* response to a repeat transmission -- ignore */ /* response to a repeat transmission -- ignore */
if (raft_node_get_match_idx(p) == r->current_idx) if (raft_node_get_match_idx(node) == r->current_idx)
return 0; return 0;
raft_node_set_next_idx(p, r->current_idx + 1); raft_node_set_next_idx(node, r->current_idx + 1);
raft_node_set_match_idx(p, r->current_idx); raft_node_set_match_idx(node, r->current_idx);
/* Update commit idx */ /* Update commit idx */
int votes = 1; /* include me */ int votes = 1; /* include me */
int point = r->current_idx; int point = r->current_idx;
int i; int i;
for (i = 0; i < me->num_nodes; i++) /* Check others */ for (i = 0; i < me->num_nodes; i++) /* Check others */
if (i != me->nodeid && point <= raft_node_get_match_idx(p)) if (i != me->nodeid && point <= raft_node_get_match_idx(node))
votes++; votes++;
if (me->num_nodes / 2 < votes && raft_get_commit_idx(me_) < point) if (me->num_nodes / 2 < votes && raft_get_commit_idx(me_) < point)
raft_set_commit_idx(me_, point); raft_set_commit_idx(me_, point);
/* Aggressively send remaining entries */ /* Aggressively send remaining entries */
if (raft_get_entry_from_idx(me_, raft_node_get_next_idx(p))) if (raft_get_entry_from_idx(me_, raft_node_get_next_idx(node)))
raft_send_appendentries(me_, node); raft_send_appendentries(me_, node_idx);
/* periodic applies committed entries lazily */ /* periodic applies committed entries lazily */
...@@ -253,8 +254,7 @@ int raft_recv_appendentries( ...@@ -253,8 +254,7 @@ int raft_recv_appendentries(
if (ae->term < me->current_term) if (ae->term < me->current_term)
{ {
__log(me_, "AE term is less than current term"); __log(me_, "AE term is less than current term");
r->success = 0; goto fail;
return 0;
} }
/* Not the first appendentries we've received */ /* Not the first appendentries we've received */
...@@ -266,8 +266,7 @@ int raft_recv_appendentries( ...@@ -266,8 +266,7 @@ int raft_recv_appendentries(
if (!e) if (!e)
{ {
__log(me_, "AE no log at prev_idx %d", ae->prev_log_idx); __log(me_, "AE no log at prev_idx %d", ae->prev_log_idx);
r->success = 0; goto fail;
return 0;
} }
/* 2. Reply false if log doesn't contain an entry at prevLogIndex /* 2. Reply false if log doesn't contain an entry at prevLogIndex
...@@ -276,8 +275,7 @@ int raft_recv_appendentries( ...@@ -276,8 +275,7 @@ int raft_recv_appendentries(
{ {
__log(me_, "AE term doesn't match prev_idx (ie. %d vs %d)", __log(me_, "AE term doesn't match prev_idx (ie. %d vs %d)",
e->term, ae->prev_log_term); e->term, ae->prev_log_term);
r->success = 0; goto fail;
return 0;
} }
} }
...@@ -329,8 +327,9 @@ int raft_recv_appendentries( ...@@ -329,8 +327,9 @@ int raft_recv_appendentries(
return 0; return 0;
fail: fail:
__log(me_, "AE failure; couldn't append entry");
r->success = 0; r->success = 0;
r->current_idx = raft_get_current_idx(me_);
r->first_idx = 0;
return -1; return -1;
} }
...@@ -445,8 +444,7 @@ int raft_recv_entry(raft_server_t* me_, int node, msg_entry_t* e, ...@@ -445,8 +444,7 @@ int raft_recv_entry(raft_server_t* me_, int node, msg_entry_t* e,
if (me->nodeid != i) if (me->nodeid != i)
{ {
int next_idx = raft_node_get_next_idx(raft_get_node(me_, i)); int next_idx = raft_node_get_next_idx(raft_get_node(me_, i));
int last_log_idx = raft_get_current_idx(me_); if (next_idx == raft_get_current_idx(me_))
if (next_idx == last_log_idx)
raft_send_appendentries(me_, i); raft_send_appendentries(me_, i);
} }
...@@ -502,14 +500,14 @@ int raft_apply_entry(raft_server_t* me_) ...@@ -502,14 +500,14 @@ int raft_apply_entry(raft_server_t* me_)
return 0; return 0;
} }
void raft_send_appendentries(raft_server_t* me_, int node) void raft_send_appendentries(raft_server_t* me_, int node_idx)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
if (!(me->cb.send_appendentries)) if (!(me->cb.send_appendentries))
return; return;
raft_node_t* p = raft_get_node(me_, node); raft_node_t* node = raft_get_node(me_, node_idx);
msg_appendentries_t ae; msg_appendentries_t ae;
ae.term = me->current_term; ae.term = me->current_term;
...@@ -519,7 +517,7 @@ void raft_send_appendentries(raft_server_t* me_, int node) ...@@ -519,7 +517,7 @@ void raft_send_appendentries(raft_server_t* me_, int node)
ae.n_entries = 0; ae.n_entries = 0;
ae.entries = NULL; ae.entries = NULL;
int next_idx = raft_node_get_next_idx(p); int next_idx = raft_node_get_next_idx(node);
msg_entry_t mety; msg_entry_t mety;
...@@ -545,13 +543,13 @@ void raft_send_appendentries(raft_server_t* me_, int node) ...@@ -545,13 +543,13 @@ void raft_send_appendentries(raft_server_t* me_, int node)
} }
__log(me_, "sending appendentries node: %d, %d %d %d %d", __log(me_, "sending appendentries node: %d, %d %d %d %d",
node, node_idx,
ae.term, ae.term,
ae.leader_commit, ae.leader_commit,
ae.prev_log_idx, ae.prev_log_idx,
ae.prev_log_term); ae.prev_log_term);
me->cb.send_appendentries(me_, me->udata, node, &ae); me->cb.send_appendentries(me_, me->udata, node_idx, &ae);
} }
void raft_send_appendentries_all(raft_server_t* me_) void raft_send_appendentries_all(raft_server_t* me_)
......
...@@ -952,6 +952,44 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit( ...@@ -952,6 +952,44 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit(
CuAssertIntEquals(tc, 3, raft_get_commit_idx(r)); CuAssertIntEquals(tc, 3, raft_get_commit_idx(r));
} }
void TestRaft_follower_recv_appendentries_failure_includes_current_idx(
CuTest * tc)
{
void *r = raft_new();
raft_add_node(r, (void*)1, 1);
raft_add_node(r, (void*)2, 0);
raft_set_current_term(r, 1);
raft_entry_t ety;
ety.data.buf = "aaa";
ety.data.len = 3;
ety.id = 1;
ety.term = 1;
raft_append_entry(r, &ety);
/* receive an appendentry with commit */
msg_appendentries_t ae;
memset(&ae, 0, sizeof(msg_appendentries_t));
/* lower term means failure */
ae.term = 0;
ae.prev_log_term = 0;
ae.prev_log_idx = 0;
ae.leader_commit = 0;
msg_appendentries_response_t aer;
raft_recv_appendentries(r, 1, &ae, &aer);
CuAssertTrue(tc, 0 == aer.success);
CuAssertIntEquals(tc, 1, aer.current_idx);
/* try again with a higher current_idx */
memset(&aer, 0, sizeof(aer));
ety.id = 2;
raft_append_entry(r, &ety);
raft_recv_appendentries(r, 1, &ae, &aer);
CuAssertTrue(tc, 0 == aer.success);
CuAssertIntEquals(tc, 2, aer.current_idx);
}
void TestRaft_follower_becomes_candidate_when_election_timeout_occurs( void TestRaft_follower_becomes_candidate_when_election_timeout_occurs(
CuTest * tc) CuTest * tc)
{ {
...@@ -1705,6 +1743,73 @@ void TestRaft_leader_recv_appendentries_response_increase_commit_idx_when_majori ...@@ -1705,6 +1743,73 @@ void TestRaft_leader_recv_appendentries_response_increase_commit_idx_when_majori
CuAssertIntEquals(tc, 2, raft_get_last_applied_idx(r)); CuAssertIntEquals(tc, 2, raft_get_last_applied_idx(r));
} }
void TestRaft_leader_recv_appendentries_response_jumps_to_lower_next_idx(
CuTest * tc)
{
raft_cbs_t funcs = {
.send_appendentries = sender_appendentries,
.log = NULL
};
msg_appendentries_response_t aer;
void *sender = sender_new(NULL);
void *r = raft_new();
raft_add_node(r, (void*)1, 1);
raft_add_node(r, (void*)2, 0);
raft_set_callbacks(r, &funcs, sender);
raft_set_current_term(r, 2);
raft_set_commit_idx(r, 0);
/* append entries - we need two */
raft_entry_t ety;
ety.term = 1;
ety.id = 1;
ety.data.buf = "aaaa";
ety.data.len = 4;
raft_append_entry(r, &ety);
ety.term = 2;
ety.id = 2;
raft_append_entry(r, &ety);
ety.term = 3;
ety.id = 3;
raft_append_entry(r, &ety);
ety.term = 4;
ety.id = 4;
raft_append_entry(r, &ety);
msg_appendentries_t* ae;
/* become leader sets next_idx to current_idx */
raft_become_leader(r);
raft_node_t* node = raft_get_node(r, 1);
CuAssertIntEquals(tc, 4, raft_node_get_next_idx(node));
CuAssertTrue(tc, NULL != (ae = sender_poll_msg_data(sender)));
/* FIRST entry log application */
/* send appendentries -
* server will be waiting for response */
raft_send_appendentries(r, 1);
CuAssertTrue(tc, NULL != (ae = sender_poll_msg_data(sender)));
CuAssertIntEquals(tc, 3, ae->prev_log_term);
CuAssertIntEquals(tc, 3, ae->prev_log_idx);
/* receive mock success responses */
memset(&aer, 0, sizeof(msg_appendentries_response_t));
aer.term = 1;
aer.success = 0;
aer.current_idx = 1;
raft_recv_appendentries_response(r, 1, &aer);
CuAssertIntEquals(tc, 2, raft_node_get_next_idx(node));
/* see if new appendentries have appropriate values */
CuAssertTrue(tc, NULL != (ae = sender_poll_msg_data(sender)));
CuAssertIntEquals(tc, 1, ae->prev_log_term);
CuAssertIntEquals(tc, 1, ae->prev_log_idx);
CuAssertTrue(tc, NULL == sender_poll_msg_data(sender));
}
void TestRaft_leader_recv_appendentries_response_retry_only_if_leader(CuTest * tc) void TestRaft_leader_recv_appendentries_response_retry_only_if_leader(CuTest * tc)
{ {
raft_cbs_t funcs = { raft_cbs_t funcs = {
......
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