Commit bfff3ede authored by Matt Amos's avatar Matt Amos Committed by Willem Thiart
Browse files

Added function to track the leader's node ID

parent 816e8dd1
...@@ -424,4 +424,11 @@ int raft_get_nvotes_for_me(raft_server_t* me_); ...@@ -424,4 +424,11 @@ int raft_get_nvotes_for_me(raft_server_t* me_);
* @return node ID of who I voted for */ * @return node ID of who I voted for */
int raft_get_voted_for(raft_server_t* me); int raft_get_voted_for(raft_server_t* me);
/**
* Gets what this node thinks the node ID of the leader is.
* If this node doesn't think there's a leader, returns 0.
* @param current_leader Pointer to an integer to store the result.
* @return True if this node thinks there's a valid leader. */
int raft_get_current_leader(raft_server_t* me, int* current_leader);
#endif /* RAFT_H_ */ #endif /* RAFT_H_ */
...@@ -59,6 +59,10 @@ typedef struct { ...@@ -59,6 +59,10 @@ typedef struct {
int election_timeout; int election_timeout;
int request_timeout; int request_timeout;
/* what this node thinks is the node ID of the current leader, or -1 if
* there isn't a known current leader. */
int current_leader;
/* callbacks */ /* callbacks */
raft_cbs_t cb; raft_cbs_t cb;
void* udata; void* udata;
......
...@@ -53,6 +53,7 @@ raft_server_t* raft_new() ...@@ -53,6 +53,7 @@ raft_server_t* raft_new()
me->election_timeout = 1000; me->election_timeout = 1000;
me->log = log_new(); me->log = log_new();
raft_set_state((raft_server_t*)me, RAFT_STATE_FOLLOWER); raft_set_state((raft_server_t*)me, RAFT_STATE_FOLLOWER);
me->current_leader = -1;
return (raft_server_t*)me; return (raft_server_t*)me;
} }
...@@ -113,6 +114,7 @@ void raft_become_candidate(raft_server_t* me_) ...@@ -113,6 +114,7 @@ void raft_become_candidate(raft_server_t* me_)
memset(me->votes_for_me, 0, sizeof(int) * me->num_nodes); memset(me->votes_for_me, 0, sizeof(int) * me->num_nodes);
me->current_term += 1; me->current_term += 1;
raft_vote(me_, me->nodeid); raft_vote(me_, me->nodeid);
me->current_leader = -1;
raft_set_state(me_, RAFT_STATE_CANDIDATE); raft_set_state(me_, RAFT_STATE_CANDIDATE);
/* we need a random factor here to prevent simultaneous candidates */ /* we need a random factor here to prevent simultaneous candidates */
...@@ -306,6 +308,8 @@ int raft_recv_appendentries( ...@@ -306,6 +308,8 @@ int raft_recv_appendentries(
raft_become_follower(me_); raft_become_follower(me_);
raft_set_current_term(me_, ae->term); raft_set_current_term(me_, ae->term);
/* update current leader because we accepted appendentries from it */
me->current_leader = node;
int i; int i;
...@@ -355,6 +359,10 @@ int raft_recv_requestvote(raft_server_t* me_, int node, msg_requestvote_t* vr, ...@@ -355,6 +359,10 @@ int raft_recv_requestvote(raft_server_t* me_, int node, msg_requestvote_t* vr,
r->vote_granted = 1; r->vote_granted = 1;
} }
/* voted for someone, therefore must be in an election. */
if (0 <= me->voted_for)
me->current_leader = -1;
__log(me_, "node requested vote: %d replying: %s", __log(me_, "node requested vote: %d replying: %s",
node, r->vote_granted == 1 ? "granted" : "not granted"); node, r->vote_granted == 1 ? "granted" : "not granted");
...@@ -554,3 +562,19 @@ void raft_vote(raft_server_t* me_, int node) ...@@ -554,3 +562,19 @@ void raft_vote(raft_server_t* me_, int node)
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
me->voted_for = node; me->voted_for = node;
} }
int raft_get_current_leader(raft_server_t* me_, int* current_leader)
{
raft_server_private_t* me = (void*)me_;
/* use -1 as marker for unknown leader, because node IDs
* must be valid array indexes (see raft_get_node) */
if (me->current_leader < 0)
return 0;
else
{
if (current_leader != NULL)
*current_leader = me->current_leader;
return 1;
}
}
...@@ -120,6 +120,9 @@ int raft_get_commit_idx(raft_server_t* me_) ...@@ -120,6 +120,9 @@ int raft_get_commit_idx(raft_server_t* me_)
void raft_set_state(raft_server_t* me_, int state) void raft_set_state(raft_server_t* me_, int state)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
/* if became the leader, then update the current leader entry */
if (state == RAFT_STATE_LEADER)
me->current_leader = me->nodeid;
me->state = state; me->state = state;
} }
......
...@@ -515,13 +515,15 @@ TestRaft_follower_recv_appendentries_reply_false_if_term_less_than_currentterm( ...@@ -515,13 +515,15 @@ TestRaft_follower_recv_appendentries_reply_false_if_term_less_than_currentterm(
.log = NULL .log = NULL
}; };
msg_appendentries_t ae; msg_appendentries_t ae;
msg_appendentries_response_t aer; msg_appendentries_response_t aer, *aerr;
r = raft_new(); r = raft_new();
raft_add_peer(r, (void*)1, 1); raft_add_peer(r, (void*)1, 1);
raft_add_peer(r, (void*)2, 0); raft_add_peer(r, (void*)2, 0);
sender = sender_new(NULL); sender = sender_new(NULL);
raft_set_callbacks(r, &funcs, sender); raft_set_callbacks(r, &funcs, sender);
/* no leader known at this point */
CuAssertTrue(tc, 0 == raft_get_current_leader(r, NULL));
/* term is low */ /* term is low */
memset(&ae, 0, sizeof(msg_appendentries_t)); memset(&ae, 0, sizeof(msg_appendentries_t));
...@@ -531,8 +533,11 @@ TestRaft_follower_recv_appendentries_reply_false_if_term_less_than_currentterm( ...@@ -531,8 +533,11 @@ TestRaft_follower_recv_appendentries_reply_false_if_term_less_than_currentterm(
raft_set_current_term(r, 5); raft_set_current_term(r, 5);
raft_recv_appendentries(r, 1, &ae, &aer); raft_recv_appendentries(r, 1, &ae, &aer);
/* response is false */ aerr = sender_poll_msg_data(sender);
CuAssertTrue(tc, 0 == aer.success); CuAssertTrue(tc, NULL != aerr);
CuAssertTrue(tc, 0 == aerr->success);
/* rejected appendentries doesn't change the current leader. */
CuAssertTrue(tc, 0 == raft_get_current_leader(r, NULL));
} }
/* TODO: check if test case is needed */ /* TODO: check if test case is needed */
...@@ -544,6 +549,7 @@ TestRaft_follower_recv_appendentries_updates_currentterm_if_term_gt_currentterm( ...@@ -544,6 +549,7 @@ TestRaft_follower_recv_appendentries_updates_currentterm_if_term_gt_currentterm(
void *sender; void *sender;
msg_appendentries_t ae; msg_appendentries_t ae;
msg_appendentries_response_t aer; msg_appendentries_response_t aer;
int current_leader = 0;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.log = NULL .log = NULL
...@@ -557,6 +563,7 @@ TestRaft_follower_recv_appendentries_updates_currentterm_if_term_gt_currentterm( ...@@ -557,6 +563,7 @@ TestRaft_follower_recv_appendentries_updates_currentterm_if_term_gt_currentterm(
/* older currentterm */ /* older currentterm */
raft_set_current_term(r, 1); raft_set_current_term(r, 1);
CuAssertTrue(tc, 0 == raft_get_current_leader(r, NULL));
/* newer term for appendentry */ /* newer term for appendentry */
memset(&ae, 0, sizeof(msg_appendentries_t)); memset(&ae, 0, sizeof(msg_appendentries_t));
...@@ -569,6 +576,9 @@ TestRaft_follower_recv_appendentries_updates_currentterm_if_term_gt_currentterm( ...@@ -569,6 +576,9 @@ TestRaft_follower_recv_appendentries_updates_currentterm_if_term_gt_currentterm(
CuAssertTrue(tc, 1 == aer.success); CuAssertTrue(tc, 1 == aer.success);
/* term has been updated */ /* term has been updated */
CuAssertTrue(tc, 2 == raft_get_current_term(r)); CuAssertTrue(tc, 2 == raft_get_current_term(r));
/* and leader has been updated */
CuAssertTrue(tc, 1 == raft_get_current_leader(r, &current_leader));
CuAssertTrue(tc, 1 == current_leader);
} }
void TestRaft_follower_doesnt_log_after_appendentry_if_no_entries_are_specified( void TestRaft_follower_doesnt_log_after_appendentry_if_no_entries_are_specified(
...@@ -1198,6 +1208,7 @@ void TestRaft_candidate_recv_appendentries_frm_leader_results_in_follower( ...@@ -1198,6 +1208,7 @@ void TestRaft_candidate_recv_appendentries_frm_leader_results_in_follower(
{ {
void *r; void *r;
void *sender; void *sender;
int current_leader = 0;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.log = NULL .log = NULL
}; };
...@@ -1211,6 +1222,7 @@ void TestRaft_candidate_recv_appendentries_frm_leader_results_in_follower( ...@@ -1211,6 +1222,7 @@ void TestRaft_candidate_recv_appendentries_frm_leader_results_in_follower(
raft_set_state(r, RAFT_STATE_CANDIDATE); raft_set_state(r, RAFT_STATE_CANDIDATE);
CuAssertTrue(tc, 0 == raft_is_follower(r)); CuAssertTrue(tc, 0 == raft_is_follower(r));
CuAssertTrue(tc, 0 == raft_get_current_leader(r, NULL));
/* receive recent appendentries */ /* receive recent appendentries */
msg_appendentries_t ae; msg_appendentries_t ae;
...@@ -1220,6 +1232,9 @@ void TestRaft_candidate_recv_appendentries_frm_leader_results_in_follower( ...@@ -1220,6 +1232,9 @@ void TestRaft_candidate_recv_appendentries_frm_leader_results_in_follower(
ae.term = 1; ae.term = 1;
raft_recv_appendentries(r, 1, &ae, &aer); raft_recv_appendentries(r, 1, &ae, &aer);
CuAssertTrue(tc, 1 == raft_is_follower(r)); CuAssertTrue(tc, 1 == raft_is_follower(r));
/* after accepting a leader, it's available as the last known leader */
CuAssertTrue(tc, 1 == raft_get_current_leader(r, &current_leader));
CuAssertTrue(tc, 1 == current_leader);
} }
/* Candidate 5.2 */ /* Candidate 5.2 */
...@@ -1249,6 +1264,7 @@ TestRaft_candidate_recv_appendentries_frm_invalid_leader_doesnt_result_in_follow ...@@ -1249,6 +1264,7 @@ TestRaft_candidate_recv_appendentries_frm_invalid_leader_doesnt_result_in_follow
/* is a candidate */ /* is a candidate */
raft_set_state(r, RAFT_STATE_CANDIDATE); raft_set_state(r, RAFT_STATE_CANDIDATE);
CuAssertTrue(tc, 0 == raft_is_follower(r)); CuAssertTrue(tc, 0 == raft_is_follower(r));
CuAssertTrue(tc, 0 == raft_get_current_leader(r, NULL));
/* invalid leader determined by "leaders" old log */ /* invalid leader determined by "leaders" old log */
memset(&ae, 0, sizeof(msg_appendentries_t)); memset(&ae, 0, sizeof(msg_appendentries_t));
...@@ -1259,6 +1275,8 @@ TestRaft_candidate_recv_appendentries_frm_invalid_leader_doesnt_result_in_follow ...@@ -1259,6 +1275,8 @@ TestRaft_candidate_recv_appendentries_frm_invalid_leader_doesnt_result_in_follow
/* appendentry from invalid leader doesn't make candidate become follower */ /* appendentry from invalid leader doesn't make candidate become follower */
raft_recv_appendentries(r, 1, &ae, &aer); raft_recv_appendentries(r, 1, &ae, &aer);
CuAssertTrue(tc, 1 == raft_is_candidate(r)); CuAssertTrue(tc, 1 == raft_is_candidate(r));
/* should not have changed leader after rejecting appendentry */
CuAssertTrue(tc, 0 == raft_get_current_leader(r, NULL));
} }
void TestRaft_leader_becomes_leader_is_leader(CuTest * tc) void TestRaft_leader_becomes_leader_is_leader(CuTest * tc)
...@@ -1602,6 +1620,7 @@ void TestRaft_leader_steps_down_if_received_appendentries_is_newer_than_itself( ...@@ -1602,6 +1620,7 @@ void TestRaft_leader_steps_down_if_received_appendentries_is_newer_than_itself(
{ {
void *r; void *r;
void *sender; void *sender;
int current_leader = 0;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.log = NULL .log = NULL
}; };
...@@ -1618,6 +1637,10 @@ void TestRaft_leader_steps_down_if_received_appendentries_is_newer_than_itself( ...@@ -1618,6 +1637,10 @@ void TestRaft_leader_steps_down_if_received_appendentries_is_newer_than_itself(
raft_set_current_term(r, 5); raft_set_current_term(r, 5);
raft_set_current_idx(r, 5); raft_set_current_idx(r, 5);
raft_set_callbacks(r, &funcs, sender); raft_set_callbacks(r, &funcs, sender);
/* check that node 0 considers itself the leader */
CuAssertTrue(tc, 1 == raft_is_leader(r));
CuAssertTrue(tc, 1 == raft_get_current_leader(r, &current_leader));
CuAssertTrue(tc, 0 == current_leader);
memset(&ae, 0, sizeof(msg_appendentries_t)); memset(&ae, 0, sizeof(msg_appendentries_t));
ae.term = 5; ae.term = 5;
...@@ -1625,7 +1648,11 @@ void TestRaft_leader_steps_down_if_received_appendentries_is_newer_than_itself( ...@@ -1625,7 +1648,11 @@ void TestRaft_leader_steps_down_if_received_appendentries_is_newer_than_itself(
ae.prev_log_term = 5; ae.prev_log_term = 5;
raft_recv_appendentries(r, 1, &ae, &aer); raft_recv_appendentries(r, 1, &ae, &aer);
/* after more recent appendentries from node 1, node 0 should
* consider node 1 the leader. */
CuAssertTrue(tc, 1 == raft_is_follower(r)); CuAssertTrue(tc, 1 == raft_is_follower(r));
CuAssertTrue(tc, 1 == raft_get_current_leader(r, &current_leader));
CuAssertTrue(tc, 0 == current_leader);
} }
void TestRaft_leader_steps_down_if_received_appendentries_has_newer_term_than_itself( void TestRaft_leader_steps_down_if_received_appendentries_has_newer_term_than_itself(
......
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