Unverified Commit 7be57f36 authored by Ozan Tezcan's avatar Ozan Tezcan Committed by GitHub
Browse files

Refactor raft_recv_appendentries() (#148)

Refactored raft_recv_appendentries() just to simplify the flow.

This PR should not introduce any behavioral change. Only exception is
previously, some tests were manually creating an append request with
prev_log_idx:0 and prev_log_term:1. This is not possible and it
cannot happen without manually creating append requests. Term of the
entry at index 0 cannot be 1. (it is zero by definition). Fixed these tests
as part of this PR.
parent be217147
...@@ -841,7 +841,7 @@ int raft_recv_appendentries(raft_server_t *me, ...@@ -841,7 +841,7 @@ int raft_recv_appendentries(raft_server_t *me,
raft_appendentries_req_t *req, raft_appendentries_req_t *req,
raft_appendentries_resp_t *resp) raft_appendentries_resp_t *resp)
{ {
int e = 0; int e;
raft_log(me, "%d <-- %d, recv appendentries_req " raft_log(me, "%d <-- %d, recv appendentries_req "
"lead:%d, id:%ld, t:%ld, pli:%ld, plt:%ld, lc:%ld ent:%ld", "lead:%d, id:%ld, t:%ld, pli:%ld, plt:%ld, lc:%ld ent:%ld",
...@@ -868,31 +868,25 @@ int raft_recv_appendentries(raft_server_t *me, ...@@ -868,31 +868,25 @@ int raft_recv_appendentries(raft_server_t *me,
raft_accept_leader(me, req->leader_id); raft_accept_leader(me, req->leader_id);
raft_reset_transfer_leader(me, 0); raft_reset_transfer_leader(me, 0);
/* Not the first appendentries we've received */ if (req->prev_log_idx == me->snapshot_last_idx) {
/* NOTE: the log starts at 1 */ if (req->prev_log_term != me->snapshot_last_term) {
if (req->prev_log_idx > 0) { /* Should never happen; something is seriously wrong! */
raft_entry_t *ety = raft_get_entry_from_idx(me, req->prev_log_idx); raft_log(me, "AE prev_log_term:%ld and snapshot term:%ld conflict",
req->prev_log_term, me->snapshot_last_term);
/* Is a snapshot */
if (req->prev_log_idx == me->snapshot_last_idx) {
if (me->snapshot_last_term != req->prev_log_term) {
/* Should never happen; something is seriously wrong! */
raft_log(me, "AE prev_log_term:%ld conflicts with snapshot term:%ld",
req->prev_log_term, me->snapshot_last_term);
e = RAFT_ERR_SHUTDOWN; e = RAFT_ERR_SHUTDOWN;
if (ety) { goto out;
raft_entry_release(ety);
}
goto out;
}
} }
} else {
/* 2. Reply false if log doesn't contain an entry at prevLogIndex /* 2. Reply false if log doesn't contain an entry at prevLogIndex
whose term matches prevLogTerm (§5.3) */ whose term matches prevLogTerm (§5.3) */
else if (!ety) { raft_entry_t *ety = raft_get_entry_from_idx(me, req->prev_log_idx);
if (!ety) {
raft_log(me, "AE no log at prev_log_idx:%ld", req->prev_log_idx); raft_log(me, "AE no log at prev_log_idx:%ld", req->prev_log_idx);
goto out; goto out;
} else if (ety->term != req->prev_log_term) { }
if (ety->term != req->prev_log_term) {
raft_log(me, "AE prev_log_term:%ld doesn't match entry term:%ld", raft_log(me, "AE prev_log_term:%ld doesn't match entry term:%ld",
req->prev_log_term, ety->term); req->prev_log_term, ety->term);
...@@ -905,15 +899,14 @@ int raft_recv_appendentries(raft_server_t *me, ...@@ -905,15 +899,14 @@ int raft_recv_appendentries(raft_server_t *me,
raft_entry_release(ety); raft_entry_release(ety);
goto out; goto out;
} }
/* Delete all the following log entries because they don't match */ /* Delete all the following log entries because they don't match */
e = raft_delete_entry_from_idx(me, req->prev_log_idx); e = raft_delete_entry_from_idx(me, req->prev_log_idx);
raft_entry_release(ety); raft_entry_release(ety);
goto out; goto out;
} }
if (ety) { raft_entry_release(ety);
raft_entry_release(ety);
}
} }
resp->success = 1; resp->success = 1;
...@@ -937,12 +930,14 @@ int raft_recv_appendentries(raft_server_t *me, ...@@ -937,12 +930,14 @@ int raft_recv_appendentries(raft_server_t *me,
raft_index_t ety_index = req->prev_log_idx + 1 + i; raft_index_t ety_index = req->prev_log_idx + 1 + i;
raft_entry_t *existing_ety = raft_get_entry_from_idx(me, ety_index); raft_entry_t *existing_ety = raft_get_entry_from_idx(me, ety_index);
raft_term_t existing_term = existing_ety ? existing_ety->term : 0; if (!existing_ety) {
if (existing_ety) { break;
raft_entry_release(existing_ety);
} }
if (existing_ety && existing_term != ety->term) { raft_term_t existing_term = existing_ety->term;
raft_entry_release(existing_ety);
if (ety->term != existing_term) {
if (ety_index <= me->commit_idx) { if (ety_index <= me->commit_idx) {
/* Should never happen; something is seriously wrong! */ /* Should never happen; something is seriously wrong! */
raft_log(me, "AE entry index:%ld is less than commit idx:%ld", raft_log(me, "AE entry index:%ld is less than commit idx:%ld",
...@@ -956,8 +951,6 @@ int raft_recv_appendentries(raft_server_t *me, ...@@ -956,8 +951,6 @@ int raft_recv_appendentries(raft_server_t *me,
goto out; goto out;
} }
break; break;
} else if (!existing_ety) {
break;
} }
resp->current_idx = ety_index; resp->current_idx = ety_index;
} }
......
...@@ -1249,7 +1249,7 @@ void TestRaft_follower_recv_appendentries_increases_log(CuTest * tc) ...@@ -1249,7 +1249,7 @@ void TestRaft_follower_recv_appendentries_increases_log(CuTest * tc)
/* receive an appendentry with commit */ /* receive an appendentry with commit */
memset(&ae, 0, sizeof(raft_appendentries_req_t)); memset(&ae, 0, sizeof(raft_appendentries_req_t));
ae.term = 3; ae.term = 3;
ae.prev_log_term = 1; ae.prev_log_term = 0;
/* first appendentries msg */ /* first appendentries msg */
ae.prev_log_idx = 0; ae.prev_log_idx = 0;
ae.leader_commit = 5; ae.leader_commit = 5;
...@@ -1481,7 +1481,7 @@ void TestRaft_follower_recv_appendentries_add_new_entries_not_already_in_log( ...@@ -1481,7 +1481,7 @@ void TestRaft_follower_recv_appendentries_add_new_entries_not_already_in_log(
memset(&ae, 0, sizeof(raft_appendentries_req_t)); memset(&ae, 0, sizeof(raft_appendentries_req_t));
ae.term = 1; ae.term = 1;
ae.prev_log_idx = 0; ae.prev_log_idx = 0;
ae.prev_log_term = 1; ae.prev_log_term = 0;
/* include entries */ /* include entries */
ae.entries = __MAKE_ENTRY_ARRAY_SEQ_ID(2, 1, 1, "aaa"); ae.entries = __MAKE_ENTRY_ARRAY_SEQ_ID(2, 1, 1, "aaa");
ae.n_entries = 2; ae.n_entries = 2;
...@@ -1511,7 +1511,7 @@ void TestRaft_follower_recv_appendentries_does_not_add_dupe_entries_already_in_l ...@@ -1511,7 +1511,7 @@ void TestRaft_follower_recv_appendentries_does_not_add_dupe_entries_already_in_l
memset(&ae, 0, sizeof(raft_appendentries_req_t)); memset(&ae, 0, sizeof(raft_appendentries_req_t));
ae.term = 1; ae.term = 1;
ae.prev_log_idx = 0; ae.prev_log_idx = 0;
ae.prev_log_term = 1; ae.prev_log_term = 0;
/* include 1 entry */ /* include 1 entry */
ae.entries = __MAKE_ENTRY_ARRAY(1, 0, "aaa"); ae.entries = __MAKE_ENTRY_ARRAY(1, 0, "aaa");
ae.n_entries = 1; ae.n_entries = 1;
...@@ -1662,7 +1662,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx( ...@@ -1662,7 +1662,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx(
memset(&ae, 0, sizeof(raft_appendentries_req_t)); memset(&ae, 0, sizeof(raft_appendentries_req_t));
ae.term = 1; ae.term = 1;
ae.prev_log_idx = 0; ae.prev_log_idx = 0;
ae.prev_log_term = 1; ae.prev_log_term = 0;
/* include entries */ /* include entries */
raft_entry_req_t e[4] = { raft_entry_req_t e[4] = {
{ .id = 1, .term = 1 }, { .id = 1, .term = 1 },
...@@ -1710,7 +1710,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit( ...@@ -1710,7 +1710,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit(
memset(&ae, 0, sizeof(raft_appendentries_req_t)); memset(&ae, 0, sizeof(raft_appendentries_req_t));
ae.term = 1; ae.term = 1;
ae.prev_log_idx = 0; ae.prev_log_idx = 0;
ae.prev_log_term = 1; ae.prev_log_term = 0;
/* include entries */ /* include entries */
raft_entry_req_t e[4] = { raft_entry_req_t e[4] = {
{ .id = 1, .term = 1 }, { .id = 1, .term = 1 },
...@@ -1867,7 +1867,7 @@ void TestRaft_follower_recv_appendentries_heartbeat_does_not_overwrite_logs( ...@@ -1867,7 +1867,7 @@ void TestRaft_follower_recv_appendentries_heartbeat_does_not_overwrite_logs(
memset(&ae, 0, sizeof(raft_appendentries_req_t)); memset(&ae, 0, sizeof(raft_appendentries_req_t));
ae.term = 1; ae.term = 1;
ae.prev_log_idx = 0; ae.prev_log_idx = 0;
ae.prev_log_term = 1; ae.prev_log_term = 0;
/* include entries */ /* include entries */
ae.entries = __MAKE_ENTRY_ARRAY(1, 1, "aaa"); ae.entries = __MAKE_ENTRY_ARRAY(1, 1, "aaa");
ae.n_entries = 1; ae.n_entries = 1;
...@@ -1918,7 +1918,7 @@ void TestRaft_follower_recv_appendentries_does_not_deleted_commited_entries( ...@@ -1918,7 +1918,7 @@ void TestRaft_follower_recv_appendentries_does_not_deleted_commited_entries(
memset(&ae, 0, sizeof(raft_appendentries_req_t)); memset(&ae, 0, sizeof(raft_appendentries_req_t));
ae.term = 1; ae.term = 1;
ae.prev_log_idx = 0; ae.prev_log_idx = 0;
ae.prev_log_term = 1; ae.prev_log_term = 0;
/* include entries */ /* include entries */
ae.entries = __MAKE_ENTRY_ARRAY(1, 1, "aaa"); ae.entries = __MAKE_ENTRY_ARRAY(1, 1, "aaa");
ae.n_entries = 1; ae.n_entries = 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