Commit 164beb15 authored by Willem Thiart's avatar Willem Thiart
Browse files

Made receive VoteRequest easier

parent 1b2a7e22
...@@ -127,20 +127,6 @@ typedef int ( ...@@ -127,20 +127,6 @@ typedef int (
msg_requestvote_t* msg msg_requestvote_t* msg
); );
/**
* @param raft The Raft server making this callback
* @param udata User data that is passed from Raft server
* @param node The peer's ID that we are sending this message to
* @return 0 on error */
typedef int (
*func_send_requestvote_response_f
) (
raft_server_t* raft,
void *udata,
int node,
msg_requestvote_response_t* msg
);
/** /**
* @param raft The Raft server making this callback * @param raft The Raft server making this callback
* @param udata User data that is passed from Raft server * @param udata User data that is passed from Raft server
...@@ -214,7 +200,6 @@ typedef int ( ...@@ -214,7 +200,6 @@ typedef int (
typedef struct typedef struct
{ {
func_send_requestvote_f send_requestvote; func_send_requestvote_f send_requestvote;
func_send_requestvote_response_f send_requestvote_response;
func_send_appendentries_f send_appendentries; func_send_appendentries_f send_appendentries;
func_send_entries_f send_entries; func_send_entries_f send_entries;
func_send_entries_response_f send_entries_response; func_send_entries_response_f send_entries_response;
...@@ -285,6 +270,7 @@ int raft_periodic(raft_server_t* me, int msec_elapsed); ...@@ -285,6 +270,7 @@ int raft_periodic(raft_server_t* me, int msec_elapsed);
/** /**
* Receive an appendentries message * Receive an appendentries message
* This function should block if it needs to append the message.
* @param node Index of the node who sent us this message * @param node Index of the node who sent us this message
* @param ae The appendentries message * @param ae The appendentries message
* @param[out] r The resulting response * @param[out] r The resulting response
...@@ -304,9 +290,11 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -304,9 +290,11 @@ int raft_recv_appendentries_response(raft_server_t* me_,
* Receive a requestvote message * Receive a requestvote message
* @param node Index of the node who sent us this message * @param node Index of the node who sent us this message
* @param vr The requestvote message * @param vr The requestvote message
* @param[out] r The resulting response
* @return 0 on error */ * @return 0 on error */
int raft_recv_requestvote(raft_server_t* me, int node, int raft_recv_requestvote(raft_server_t* me, int node,
msg_requestvote_t* vr); msg_requestvote_t* vr,
msg_requestvote_response_t *r);
/** /**
* Receive a response from a requestvote message we sent * Receive a response from a requestvote message we sent
...@@ -320,6 +308,7 @@ int raft_recv_requestvote_response(raft_server_t* me, int node, ...@@ -320,6 +308,7 @@ int raft_recv_requestvote_response(raft_server_t* me, int node,
* Receive an entry message from client. * Receive an entry message from client.
* Append the entry to the log * Append the entry to the log
* Send appendentries to followers * Send appendentries to followers
* This function should block if it needs to append the message.
* @param node Index of the node who sent us this message * @param node Index of the node who sent us this message
* @param e The entry message */ * @param e The entry message */
int raft_recv_entry(raft_server_t* me, int node, msg_entry_t* e); int raft_recv_entry(raft_server_t* me, int node, msg_entry_t* e);
......
...@@ -337,10 +337,10 @@ int raft_recv_appendentries( ...@@ -337,10 +337,10 @@ int raft_recv_appendentries(
return 1; return 1;
} }
int raft_recv_requestvote(raft_server_t* me_, int node, msg_requestvote_t* vr) int raft_recv_requestvote(raft_server_t* me_, int node, msg_requestvote_t* vr,
msg_requestvote_response_t *r)
{ {
raft_server_private_t* me = (void*)me_; raft_server_private_t* me = (void*)me_;
msg_requestvote_response_t r;
if (raft_get_current_term(me_) < vr->term) if (raft_get_current_term(me_) < vr->term)
me->voted_for = -1; me->voted_for = -1;
...@@ -350,20 +350,17 @@ int raft_recv_requestvote(raft_server_t* me_, int node, msg_requestvote_t* vr) ...@@ -350,20 +350,17 @@ int raft_recv_requestvote(raft_server_t* me_, int node, msg_requestvote_t* vr)
-1 != me->voted_for || -1 != me->voted_for ||
/* we have a more up-to-date log */ /* we have a more up-to-date log */
vr->last_log_idx < me->current_idx) vr->last_log_idx < me->current_idx)
r.vote_granted = 0; r->vote_granted = 0;
else else
{ {
raft_vote(me_, node); raft_vote(me_, node);
r.vote_granted = 1; r->vote_granted = 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");
r.term = raft_get_current_term(me_);
if (me->cb.send_requestvote_response)
me->cb.send_requestvote_response(me_, me->udata, node, &r);
r->term = raft_get_current_term(me_);
return 0; return 0;
} }
......
...@@ -160,8 +160,13 @@ void sender_poll_msgs(void* s) ...@@ -160,8 +160,13 @@ void sender_poll_msgs(void* s)
raft_recv_appendentries_response(me->raft, m->sender, m->data); raft_recv_appendentries_response(me->raft, m->sender, m->data);
break; break;
case RAFT_MSG_REQUESTVOTE: case RAFT_MSG_REQUESTVOTE:
raft_recv_requestvote(me->raft, m->sender, m->data); {
break; msg_requestvote_response_t response;
raft_recv_requestvote(me->raft, m->sender, m->data, &response);
__append_msg(me, &response, RAFT_MSG_REQUESTVOTE_RESPONSE,
sizeof(response), m->sender, me->raft);
}
break;
case RAFT_MSG_REQUESTVOTE_RESPONSE: case RAFT_MSG_REQUESTVOTE_RESPONSE:
raft_recv_requestvote_response(me->raft, m->sender, m->data); raft_recv_requestvote_response(me->raft, m->sender, m->data);
break; break;
......
...@@ -35,10 +35,7 @@ void TestRaft_scenario_leader_appears(CuTest * tc) ...@@ -35,10 +35,7 @@ void TestRaft_scenario_leader_appears(CuTest * tc)
raft_set_callbacks(r[j], raft_set_callbacks(r[j],
&((raft_cbs_t) { &((raft_cbs_t) {
.send_requestvote = sender_requestvote, .send_requestvote = sender_requestvote,
.send_requestvote_response = .send_appendentries = sender_appendentries,
sender_requestvote_response,
.send_appendentries =
sender_appendentries,
.send_entries = sender_entries, .send_entries = sender_entries,
.send_entries_response = .send_entries_response =
sender_entries_response, sender_entries_response,
......
...@@ -431,11 +431,10 @@ void TestRaft_server_recv_requestvote_reply_false_if_term_less_than_current_term ...@@ -431,11 +431,10 @@ void TestRaft_server_recv_requestvote_reply_false_if_term_less_than_current_term
void *r; void *r;
void *sender; void *sender;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.send_requestvote_response = sender_requestvote_response, .log = NULL
.log = NULL
}; };
msg_requestvote_t rv; msg_requestvote_t rv;
msg_requestvote_response_t *rvr; msg_requestvote_response_t rvr;
/* 2 nodes */ /* 2 nodes */
raft_node_configuration_t cfg[] = { raft_node_configuration_t cfg[] = {
...@@ -453,11 +452,9 @@ void TestRaft_server_recv_requestvote_reply_false_if_term_less_than_current_term ...@@ -453,11 +452,9 @@ void TestRaft_server_recv_requestvote_reply_false_if_term_less_than_current_term
/* term is less than current term */ /* term is less than current term */
memset(&rv, 0, sizeof(msg_requestvote_t)); memset(&rv, 0, sizeof(msg_requestvote_t));
rv.term = 1; rv.term = 1;
raft_recv_requestvote(r, 1, &rv); raft_recv_requestvote(r, 1, &rv, &rvr);
rvr = sender_poll_msg_data(sender); CuAssertTrue(tc, 0 == rvr.vote_granted);
CuAssertTrue(tc, NULL != rvr);
CuAssertTrue(tc, 0 == rvr->vote_granted);
} }
/* If votedFor is null or candidateId, and candidate's log is at /* If votedFor is null or candidateId, and candidate's log is at
...@@ -469,11 +466,10 @@ void TestRaft_server_dont_grant_vote_if_we_didnt_vote_for_this_candidate( ...@@ -469,11 +466,10 @@ void TestRaft_server_dont_grant_vote_if_we_didnt_vote_for_this_candidate(
void *r; void *r;
void *sender; void *sender;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.send_requestvote_response = sender_requestvote_response, .log = NULL
.log = NULL
}; };
msg_requestvote_t rv; msg_requestvote_t rv;
msg_requestvote_response_t *rvr; msg_requestvote_response_t rvr;
/* 2 nodes */ /* 2 nodes */
raft_node_configuration_t cfg[] = { raft_node_configuration_t cfg[] = {
...@@ -495,11 +491,8 @@ void TestRaft_server_dont_grant_vote_if_we_didnt_vote_for_this_candidate( ...@@ -495,11 +491,8 @@ void TestRaft_server_dont_grant_vote_if_we_didnt_vote_for_this_candidate(
rv.candidate_id = 1; rv.candidate_id = 1;
rv.last_log_idx = 0; rv.last_log_idx = 0;
rv.last_log_term = 1; rv.last_log_term = 1;
raft_recv_requestvote(r, 1, &rv); raft_recv_requestvote(r, 1, &rv, &rvr);
CuAssertTrue(tc, 0 == rvr.vote_granted);
rvr = sender_poll_msg_data(sender);
CuAssertTrue(tc, NULL != rvr);
CuAssertTrue(tc, 0 == rvr->vote_granted);
} }
void TestRaft_follower_becomes_follower_is_follower(CuTest * tc) void TestRaft_follower_becomes_follower_is_follower(CuTest * tc)
...@@ -531,7 +524,7 @@ TestRaft_follower_recv_appendentries_reply_false_if_term_less_than_currentterm( ...@@ -531,7 +524,7 @@ TestRaft_follower_recv_appendentries_reply_false_if_term_less_than_currentterm(
void *r; void *r;
void *sender; void *sender;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.log = NULL .log = NULL
}; };
msg_appendentries_t ae; msg_appendentries_t ae;
msg_appendentries_response_t aer; msg_appendentries_response_t aer;
...@@ -572,7 +565,7 @@ TestRaft_follower_recv_appendentries_updates_currentterm_if_term_gt_currentterm( ...@@ -572,7 +565,7 @@ TestRaft_follower_recv_appendentries_updates_currentterm_if_term_gt_currentterm(
msg_appendentries_response_t aer; msg_appendentries_response_t aer;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.log = NULL .log = NULL
}; };
/* 2 nodes */ /* 2 nodes */
...@@ -648,7 +641,7 @@ void TestRaft_follower_increases_log_after_appendentry(CuTest * tc) ...@@ -648,7 +641,7 @@ void TestRaft_follower_increases_log_after_appendentry(CuTest * tc)
char *str = "aaa"; char *str = "aaa";
raft_cbs_t funcs = { raft_cbs_t funcs = {
.log = NULL .log = NULL
}; };
/* 2 nodes */ /* 2 nodes */
...@@ -698,7 +691,7 @@ TestRaft_follower_recv_appendentries_reply_false_if_doesnt_have_log_at_prev_log_ ...@@ -698,7 +691,7 @@ TestRaft_follower_recv_appendentries_reply_false_if_doesnt_have_log_at_prev_log_
msg_entry_t ety; msg_entry_t ety;
char *str = "aaa"; char *str = "aaa";
raft_cbs_t funcs = { raft_cbs_t funcs = {
.log = NULL .log = NULL
}; };
/* 2 nodes */ /* 2 nodes */
raft_node_configuration_t cfg[] = { raft_node_configuration_t cfg[] = {
...@@ -754,7 +747,7 @@ TestRaft_follower_recv_appendentries_delete_entries_if_conflict_with_new_entries ...@@ -754,7 +747,7 @@ TestRaft_follower_recv_appendentries_delete_entries_if_conflict_with_new_entries
raft_entry_t *ety_appended; raft_entry_t *ety_appended;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.log = NULL .log = NULL
}; };
/* 2 nodes */ /* 2 nodes */
...@@ -822,7 +815,7 @@ void TestRaft_follower_recv_appendentries_add_new_entries_not_already_in_log( ...@@ -822,7 +815,7 @@ void TestRaft_follower_recv_appendentries_add_new_entries_not_already_in_log(
void *r; void *r;
void *sender; void *sender;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.log = NULL .log = NULL
}; };
/* 2 nodes */ /* 2 nodes */
...@@ -867,7 +860,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx( ...@@ -867,7 +860,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx(
void *r; void *r;
void *sender; void *sender;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.log = NULL .log = NULL
}; };
/* 2 nodes */ /* 2 nodes */
...@@ -920,7 +913,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit( ...@@ -920,7 +913,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit(
void *r; void *r;
void *sender; void *sender;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.log = NULL .log = NULL
}; };
/* 2 nodes */ /* 2 nodes */
...@@ -1000,8 +993,7 @@ void TestRaft_follower_dont_grant_vote_if_candidate_has_a_less_complete_log( ...@@ -1000,8 +993,7 @@ void TestRaft_follower_dont_grant_vote_if_candidate_has_a_less_complete_log(
void *r; void *r;
void *sender; void *sender;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.send_requestvote_response = sender_requestvote_response, .log = NULL
.log = NULL
}; };
/* 2 nodes */ /* 2 nodes */
...@@ -1013,7 +1005,7 @@ void TestRaft_follower_dont_grant_vote_if_candidate_has_a_less_complete_log( ...@@ -1013,7 +1005,7 @@ void TestRaft_follower_dont_grant_vote_if_candidate_has_a_less_complete_log(
msg_requestvote_t rv; msg_requestvote_t rv;
msg_requestvote_response_t *rvr; msg_requestvote_response_t rvr;
sender = sender_new(NULL); sender = sender_new(NULL);
r = raft_new(); r = raft_new();
...@@ -1033,10 +1025,8 @@ void TestRaft_follower_dont_grant_vote_if_candidate_has_a_less_complete_log( ...@@ -1033,10 +1025,8 @@ void TestRaft_follower_dont_grant_vote_if_candidate_has_a_less_complete_log(
raft_set_current_idx(r, 2); raft_set_current_idx(r, 2);
/* vote not granted */ /* vote not granted */
raft_recv_requestvote(r, 1, &rv); raft_recv_requestvote(r, 1, &rv, &rvr);
rvr = sender_poll_msg_data(sender); CuAssertTrue(tc, 0 == rvr.vote_granted);
CuAssertTrue(tc, NULL != rvr);
CuAssertTrue(tc, 0 == rvr->vote_granted);
} }
void TestRaft_candidate_becomes_candidate_is_candidate(CuTest * tc) void TestRaft_candidate_becomes_candidate_is_candidate(CuTest * tc)
...@@ -1249,8 +1239,7 @@ void TestRaft_candidate_will_not_respond_to_voterequest_if_it_has_already_voted( ...@@ -1249,8 +1239,7 @@ void TestRaft_candidate_will_not_respond_to_voterequest_if_it_has_already_voted(
void *r; void *r;
void *sender; void *sender;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.send_requestvote_response = sender_requestvote_response, .log = NULL
.log = NULL
}; };
/* 2 nodes */ /* 2 nodes */
...@@ -1260,8 +1249,8 @@ void TestRaft_candidate_will_not_respond_to_voterequest_if_it_has_already_voted( ...@@ -1260,8 +1249,8 @@ void TestRaft_candidate_will_not_respond_to_voterequest_if_it_has_already_voted(
{ (-1), NULL } { (-1), NULL }
}; };
msg_requestvote_response_t* rvr;
msg_requestvote_t rv; msg_requestvote_t rv;
msg_requestvote_response_t rvr;
sender = sender_new(NULL); sender = sender_new(NULL);
r = raft_new(); r = raft_new();
...@@ -1271,11 +1260,10 @@ void TestRaft_candidate_will_not_respond_to_voterequest_if_it_has_already_voted( ...@@ -1271,11 +1260,10 @@ void TestRaft_candidate_will_not_respond_to_voterequest_if_it_has_already_voted(
raft_vote(r, 0); raft_vote(r, 0);
memset(&rv, 0, sizeof(msg_requestvote_t)); memset(&rv, 0, sizeof(msg_requestvote_t));
raft_recv_requestvote(r, 1, &rv); raft_recv_requestvote(r, 1, &rv, &rvr);
/* we've vote already, so won't respond with a vote granted... */ /* we've vote already, so won't respond with a vote granted... */
rvr = sender_poll_msg_data(sender); CuAssertTrue(tc, 0 == rvr.vote_granted);
CuAssertTrue(tc, 0 == rvr->vote_granted);
} }
/* Candidate 5.2 */ /* Candidate 5.2 */
......
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