Unverified Commit 64747dee authored by Shaya Potter's avatar Shaya Potter Committed by GitHub
Browse files

include the prev_log_idx in the appendentry response (#45)

the problem we have is that 2 appendentries can be in process in parallel

Howver, the response handler doesn't expect this, it will reset the next_idx in both, but currently resets next_idx based on next_idx's value, not the old value that was the basis of prev_log_idx.

Solution: return prev_log_idx in the response

Problem with solution: prev_log_idx is not always based on next_idx from node, but can be based on a snapshot idx if it no longer exists. I haven't coded this yet, but perhaps can determine if the prev_log_idx that is returned is the snapshot_idx and if so, then dont increment by 1. Its possible we could pass even another flag into the appendentries struct to say if its a snapshot, but that's ugly.

In addition, we increment msg_id on each appendentries so that if a response comes back out of order and we have already accepted the later response, we just ignore this response.

In addition, on appendentry response, even if the appendentry wasn't a success, as long as its the same term as us (i.e. we are still the leader) then for leader/quorum purposes, that is sufficient, so move up set_last_ack to right after term test.
parent d8799dd9
...@@ -209,6 +209,9 @@ typedef struct ...@@ -209,6 +209,9 @@ typedef struct
/** If success, this is the highest log IDX we've received and appended to /** If success, this is the highest log IDX we've received and appended to
* our log; otherwise, this is the our currentIndex */ * our log; otherwise, this is the our currentIndex */
raft_index_t current_idx; raft_index_t current_idx;
/** what the leader thouught was the index of the log just before the newest_entry */
raft_index_t prev_log_idx;
} msg_appendentries_response_t; } msg_appendentries_response_t;
typedef void* raft_server_t; typedef void* raft_server_t;
......
...@@ -600,6 +600,11 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -600,6 +600,11 @@ int raft_recv_appendentries_response(raft_server_t* me_,
if (!raft_is_leader(me_)) if (!raft_is_leader(me_))
return RAFT_ERR_NOT_LEADER; return RAFT_ERR_NOT_LEADER;
if (raft_node_get_last_acked_msgid(node) > r->msg_id) {
// this was received out of order and is now irrelevant.
return 0;
}
/* 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) */
if (me->current_term < r->term) if (me->current_term < r->term)
...@@ -614,13 +619,16 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -614,13 +619,16 @@ int raft_recv_appendentries_response(raft_server_t* me_,
else if (me->current_term != r->term) else if (me->current_term != r->term)
return 0; return 0;
// if we got here, it means that the follower has acked us as a leader, even if it cant accept the append_entry
raft_node_set_last_ack(node, r->msg_id, r->term);
raft_index_t match_idx = raft_node_get_match_idx(node); raft_index_t match_idx = raft_node_get_match_idx(node);
if (0 == r->success) if (0 == r->success)
{ {
/* 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) */
raft_index_t next_idx = raft_node_get_next_idx(node); raft_index_t next_idx = r->prev_log_idx + 1;
assert(0 < next_idx); assert(0 < next_idx);
/* Stale response -- ignore */ /* Stale response -- ignore */
if (r->current_idx < match_idx) if (r->current_idx < match_idx)
...@@ -650,8 +658,6 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -650,8 +658,6 @@ int raft_recv_appendentries_response(raft_server_t* me_,
raft_node_set_has_sufficient_logs(node); raft_node_set_has_sufficient_logs(node);
} }
raft_node_set_last_ack(node, r->msg_id, r->term);
if (r->current_idx <= match_idx) if (r->current_idx <= match_idx)
return 0; return 0;
...@@ -715,6 +721,7 @@ int raft_recv_appendentries( ...@@ -715,6 +721,7 @@ int raft_recv_appendentries(
ae->n_entries); ae->n_entries);
r->msg_id = ae->msg_id; r->msg_id = ae->msg_id;
r->prev_log_idx = ae->prev_log_idx;
r->success = 0; r->success = 0;
if (raft_is_candidate(me_) && me->current_term == ae->term) if (raft_is_candidate(me_) && me->current_term == ae->term)
...@@ -1245,7 +1252,7 @@ int raft_send_appendentries(raft_server_t* me_, raft_node_t* node) ...@@ -1245,7 +1252,7 @@ int raft_send_appendentries(raft_server_t* me_, raft_node_t* node)
msg_appendentries_t ae = { msg_appendentries_t ae = {
.term = me->current_term, .term = me->current_term,
.leader_commit = raft_get_commit_idx(me_), .leader_commit = raft_get_commit_idx(me_),
.msg_id = me->msg_id, .msg_id = ++me->msg_id,
}; };
ae.entries = raft_get_entries_from_idx(me_, next_idx, &ae.n_entries); ae.entries = raft_get_entries_from_idx(me_, next_idx, &ae.n_entries);
......
...@@ -2977,6 +2977,7 @@ void TestRaft_leader_recv_appendentries_response_jumps_to_lower_next_idx( ...@@ -2977,6 +2977,7 @@ void TestRaft_leader_recv_appendentries_response_jumps_to_lower_next_idx(
aer.term = 2; aer.term = 2;
aer.success = 0; aer.success = 0;
aer.current_idx = 1; aer.current_idx = 1;
aer.prev_log_idx = ae->prev_log_idx;
raft_recv_appendentries_response(r, raft_get_node(r, 2), &aer); raft_recv_appendentries_response(r, raft_get_node(r, 2), &aer);
CuAssertIntEquals(tc, 2, raft_node_get_next_idx(node)); CuAssertIntEquals(tc, 2, raft_node_get_next_idx(node));
...@@ -3031,6 +3032,7 @@ void TestRaft_leader_recv_appendentries_response_decrements_to_lower_next_idx( ...@@ -3031,6 +3032,7 @@ void TestRaft_leader_recv_appendentries_response_decrements_to_lower_next_idx(
aer.term = 2; aer.term = 2;
aer.success = 0; aer.success = 0;
aer.current_idx = 4; aer.current_idx = 4;
aer.prev_log_idx = ae->prev_log_idx;
raft_recv_appendentries_response(r, raft_get_node(r, 2), &aer); raft_recv_appendentries_response(r, raft_get_node(r, 2), &aer);
CuAssertIntEquals(tc, 4, raft_node_get_next_idx(node)); CuAssertIntEquals(tc, 4, raft_node_get_next_idx(node));
...@@ -3044,6 +3046,7 @@ void TestRaft_leader_recv_appendentries_response_decrements_to_lower_next_idx( ...@@ -3044,6 +3046,7 @@ void TestRaft_leader_recv_appendentries_response_decrements_to_lower_next_idx(
aer.term = 2; aer.term = 2;
aer.success = 0; aer.success = 0;
aer.current_idx = 4; aer.current_idx = 4;
aer.prev_log_idx = ae->prev_log_idx;
raft_recv_appendentries_response(r, raft_get_node(r, 2), &aer); raft_recv_appendentries_response(r, raft_get_node(r, 2), &aer);
CuAssertIntEquals(tc, 3, raft_node_get_next_idx(node)); CuAssertIntEquals(tc, 3, raft_node_get_next_idx(node));
......
...@@ -777,6 +777,7 @@ void TestRaft_leader_sends_snapshot_if_log_was_compacted(CuTest* tc) ...@@ -777,6 +777,7 @@ void TestRaft_leader_sends_snapshot_if_log_was_compacted(CuTest* tc)
aer.term = 1; aer.term = 1;
aer.success = 0; aer.success = 0;
aer.current_idx = 1; aer.current_idx = 1;
aer.prev_log_idx = raft_node_get_next_idx(node) - 1;
raft_recv_appendentries_response(r, node, &aer); raft_recv_appendentries_response(r, node, &aer);
CuAssertIntEquals(tc, 1, send_snapshot_count); CuAssertIntEquals(tc, 1, send_snapshot_count);
......
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