Unverified Commit 575335c3 authored by Shaya Potter's avatar Shaya Potter Committed by GitHub
Browse files

Adding support for TimeoutNow RPC (#7)

this is the small modifications needed for raftlib to support timeout now rpc

1) add a flag to requestvote type to tell other nodes its overriding normal chec
2) new send_timeoutnow callback to send a timeout now rpc at right time
3) call said callback is we've marked the node as timeout now and in raft_appendentries_response raft_get_current_idx(me_) == r->current_idx (i.e. response says the node's idx is up to date with leader)
4) add helper functions to set/get/reset timeout flag on node structs for targeted node.

how it would be used by client

1) client of libraft will call raft_transfer_leader(raft_server_t* me_, raft_node_id_t node_id) to target the node they want to transfer to.

2) client of libraft will provide the send_timeoutnow() callback for the actual sending of the rpc

3) client of libraft will modify their existing notify_state_event() callback to observe the result of the timeout now operation.

if it receives a RAFT_STATE_LEADERSHIP_TRANSFER_FAILED, then the transfer failed

if it received a RAFT_STATE_FOLLOWER, then it inspects for the actual leader to see if its the expected one and can return that it was transferred to expected node or not

if it received a RAFT_STATE_CANDIDATE, then the transfer also failed, as the targeted node sent a request vote which removed leadership, but wasn't able to win the election.

With PreVote, this latter case should be very rare.
parent ce8b0f02
......@@ -20,6 +20,8 @@ typedef enum {
RAFT_ERR_NEEDS_SNAPSHOT=-6,
RAFT_ERR_SNAPSHOT_IN_PROGRESS=-7,
RAFT_ERR_SNAPSHOT_ALREADY_LOADED=-8,
RAFT_ERR_INVALID_NODEID=-9,
RAFT_ERR_LEADER_TRANSFER_IN_PROGRESS=-10,
RAFT_ERR_LAST=-100,
} raft_error_e;
......@@ -33,7 +35,8 @@ typedef enum {
RAFT_STATE_FOLLOWER,
RAFT_STATE_PRECANDIDATE,
RAFT_STATE_CANDIDATE,
RAFT_STATE_LEADER
RAFT_STATE_LEADER,
RAFT_STATE_LEADERSHIP_TRANSFER_FAILED
} raft_state_e;
/** Allow entries to apply while taking a snapshot */
......@@ -145,6 +148,9 @@ typedef struct
/** term of candidate's last log entry */
raft_term_t last_log_term;
/** tell nodes that they should allow a vote, even with a healthy leader */
int transfer_leader;
} msg_requestvote_t;
/** Vote request response message.
......@@ -417,6 +423,17 @@ typedef void (
raft_state_e state
);
/** Callback for sending TimeoutNow RPC messages
* @param[in] raft The Raft server making this callback
* @param[in] node The node that we are sending this message to
* @return 0 on success
*/
typedef int (
*func_send_timeoutnow_f
) (
raft_server_t* raft,
raft_node_t* node
);
typedef struct
{
......@@ -460,6 +477,9 @@ typedef struct
/** Callback for catching debugging log messages
* This callback is optional */
func_log_f log;
/** Callback for sending TimeoutNow RPC messages to nodes */
func_send_timeoutnow_f send_timeoutnow;
} raft_cbs_t;
/** A generic notification callback used to allow Raft to notify caller
......@@ -1258,4 +1278,20 @@ void raft_queue_read_request(raft_server_t* me_, func_read_request_callback_f cb
*/
void raft_process_read_queue(raft_server_t* me_);
/*
* invoke a leadership transfer to targeted node
* node_id = targeted node
* timeout = timeout in ms before this transfer is aborted. if 0, use default election timeout
*/
int raft_transfer_leader(raft_server_t* me_, raft_node_id_t node_id, long timeout);
/* attempt to abort the leadership transfer */
void raft_reset_transfer_leader(raft_server_t* me_);
/* get the targeted node_id if a leadership transfer is in progress, or RAFT_NODE_ID_NONE if not */
raft_node_id_t raft_get_transfer_leader(raft_server_t* me_);
/* cause this server to force an election on its next raft_periodic function call */
void raft_set_timeout_now(raft_server_t* me_);
#endif /* RAFT_H_ */
......@@ -115,6 +115,11 @@ typedef struct {
raft_msg_id_t max_seen_msg_id;
raft_read_request_t *read_queue_head;
raft_read_request_t *read_queue_tail;
raft_node_id_t node_transferring_leader_to; // the node we are targeting for leadership
long transfer_leader_time; // how long we should wait for leadership transfer to take, before aborting
int timeout_now;
} raft_server_private_t;
int raft_election_start(raft_server_t* me);
......
......@@ -103,10 +103,11 @@ raft_server_t* raft_new_with_log(const raft_log_impl_t *log_impl, void *log_arg)
me->timeout_elapsed = 0;
me->request_timeout = 200;
me->election_timeout = 1000;
me->node_transferring_leader_to = RAFT_NODE_ID_NONE;
raft_update_quorum_meta((raft_server_t*)me, me->msg_id);
raft_randomize_election_timeout((raft_server_t*)me);
raft_randomize_election_timeout((raft_server_t*)me);
me->log_impl = log_impl;
me->log = me->log_impl->init(me, log_arg);
if (!me->log) {
......@@ -378,6 +379,7 @@ int raft_become_leader(raft_server_t* me_)
int i;
raft_log(me_, "becoming leader term:%ld", raft_get_current_term(me_));
raft_reset_transfer_leader(me_);
if (me->cb.notify_state_event)
me->cb.notify_state_event(me_, raft_get_udata(me_), RAFT_STATE_LEADER);
......@@ -450,6 +452,8 @@ int raft_become_candidate(raft_server_t* me_)
int i;
raft_log(me_, "becoming candidate");
raft_reset_transfer_leader(me_);
if (me->cb.notify_state_event)
me->cb.notify_state_event(me_, raft_get_udata(me_), RAFT_STATE_CANDIDATE);
......@@ -482,6 +486,8 @@ void raft_become_follower(raft_server_t* me_)
raft_server_private_t* me = (raft_server_private_t*)me_;
raft_log(me_, "becoming follower");
raft_reset_transfer_leader(me_);
if (me->cb.notify_state_event)
me->cb.notify_state_event(me_, raft_get_udata(me_), RAFT_STATE_FOLLOWER);
......@@ -582,9 +588,18 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period)
}
raft_update_quorum_meta(me_, quorum_id);
}
if (me->node_transferring_leader_to != RAFT_NODE_ID_NONE) {
me->transfer_leader_time -= msec_since_last_period;
if (me->transfer_leader_time < 0) {
if (me->cb.notify_state_event)
me->cb.notify_state_event(me_, raft_get_udata(me_), RAFT_STATE_LEADERSHIP_TRANSFER_FAILED);
raft_reset_transfer_leader(me_);
}
}
}
else if (me->election_timeout_rand <= me->timeout_elapsed &&
else if ((me->election_timeout_rand <= me->timeout_elapsed || me->timeout_now) &&
/* Don't become the leader when building snapshots or bad things will
* happen when we get a client request */
!raft_snapshot_is_in_progress(me_))
......@@ -592,6 +607,8 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period)
int e = raft_election_start(me_);
if (0 != e)
return e;
me->timeout_now = 0;
}
if (me->last_applied_idx < raft_get_commit_idx(me_) &&
......@@ -676,6 +693,10 @@ int raft_recv_appendentries_response(raft_server_t* me_,
return 0;
}
if (me->cb.send_timeoutnow && raft_get_transfer_leader(me_) == raft_node_get_id(node)
&& raft_get_current_idx(me_) == r->current_idx) {
me->cb.send_timeoutnow(me_, node);
}
if (!raft_node_is_voting(node) &&
!raft_voting_change_is_in_progress(me_) &&
......@@ -919,7 +940,8 @@ int raft_recv_requestvote(raft_server_t* me_,
r->vote_granted = 0;
/* Reject request if we have a leader */
if (me->leader_id != RAFT_NODE_ID_NONE &&
if (!vr->transfer_leader &&
me->leader_id != RAFT_NODE_ID_NONE &&
me->leader_id != vr->candidate_id &&
me->timeout_elapsed < me->election_timeout) {
goto done;
......@@ -1066,6 +1088,9 @@ int raft_recv_entry(raft_server_t* me_,
if (!raft_is_leader(me_))
return RAFT_ERR_NOT_LEADER;
if (raft_get_transfer_leader(me_) != RAFT_NODE_ID_NONE)
return RAFT_ERR_LEADER_TRANSFER_IN_PROGRESS;
raft_log(me_, "received entry t:%ld id: %d idx: %ld",
me->current_term, ety->id, raft_get_current_idx(me_) + 1);
......@@ -1131,8 +1156,11 @@ int raft_send_requestvote(raft_server_t* me_, raft_node_t* node)
rv.last_log_idx = raft_get_current_idx(me_);
rv.last_log_term = raft_get_last_log_term(me_);
rv.candidate_id = raft_get_nodeid(me_);
rv.transfer_leader = (me->node_transferring_leader_to != RAFT_NODE_ID_NONE);
if (me->cb.send_requestvote)
e = me->cb.send_requestvote(me_, me->udata, node, &rv);
return e;
}
......@@ -1791,3 +1819,36 @@ void raft_process_read_queue(raft_server_t* me_)
pop_read_queue(me, me->read_queue_head->read_term == me->current_term);
}
}
/* invoke a leadership transfer
* node_id = targeted node we are transfering to
* timeout = how long this should be allowed to take in milliseconds, as calculated by calls to raft_periodic)
* return an error if leadership transfer is already in progress of the targeted node_id is unknown
*/
int raft_transfer_leader(raft_server_t* me_, raft_node_id_t node_id, long timeout)
{
raft_server_private_t* me = (raft_server_private_t*) me_;
if (me->node_transferring_leader_to != RAFT_NODE_ID_NONE) {
return RAFT_ERR_LEADER_TRANSFER_IN_PROGRESS;
}
raft_node_t * target = raft_get_node(me_, node_id);
if (target == NULL) {
return RAFT_ERR_INVALID_NODEID;
}
if (me->cb.send_timeoutnow &&
raft_get_current_idx(me_) == raft_node_get_match_idx(target)) {
me->cb.send_timeoutnow(me_, target);
}
me->node_transferring_leader_to = node_id;
if (timeout == 0) {
me->transfer_leader_time = me->election_timeout;
} else {
me->transfer_leader_time = timeout;
}
return 0;
}
......@@ -288,3 +288,33 @@ raft_msg_id_t raft_get_msg_id(raft_server_t* me_)
raft_server_private_t* me = (raft_server_private_t*)me_;
return me->msg_id;
}
/* Stop trying to transfer leader to a targeted node
* internally used because either we have timed out our attempt or because we are no longer the leader
* possible to be used by a client as well.
*/
void raft_reset_transfer_leader(raft_server_t* me_)
{
raft_server_private_t* me = (raft_server_private_t*) me_;
me->node_transferring_leader_to = RAFT_NODE_ID_NONE;
me->transfer_leader_time = 0;
}
/* return the targeted node_id if we are in the middle of attempting a leadership transfer
* return RAFT_NODE_ID_NONE if no leadership transfer is in progress
*/
raft_node_id_t raft_get_transfer_leader(raft_server_t* me_)
{
raft_server_private_t* me = (raft_server_private_t*) me_;
return me->node_transferring_leader_to;
}
/* Forces the raft server to invoke an election on the next raft_periodic function call */
void raft_set_timeout_now(raft_server_t* me_)
{
raft_server_private_t* me = (raft_server_private_t*) me_;
me->timeout_now = 1;
}
......@@ -3953,6 +3953,76 @@ void TestRaft_single_node_commits_noop(CuTest * tc)
CuAssertStrEquals(tc, "success", str);
}
/*
* tests:
* 1) if requestvote transfer_leader flag is not set, will not accept a request vote, as not timed out
* 2) if requestvote transfer_leader flag is set, will accept a request vote, even though not timed out
*/
void TestRaft_server_recv_requestvote_with_transfer_node(CuTest * tc)
{
raft_cbs_t funcs = { 0 };
/* Setup cluster */
raft_server_t *r = raft_new();
raft_set_callbacks(r, &funcs, NULL);
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_add_node(r, NULL, 3, 0);
raft_set_current_term(r, 1);
raft_set_election_timeout(r, 1000);
raft_set_state(r, RAFT_STATE_LEADER);
/* setup requestvote struct */
msg_requestvote_t rv = {
.term = 2,
.candidate_id = 2,
.last_log_idx = 0,
.last_log_term = 1,
.transfer_leader = 0,
};
msg_requestvote_response_t rvr;
/* test #1: try to request a vote without transfer leader flag */
raft_recv_requestvote(r, raft_get_node(r, 2), &rv, &rvr);
CuAssertTrue(tc, 1 != rvr.vote_granted);
/* test #2: try to request a vote with the transfer leader flag */
rv.transfer_leader = 1;
raft_recv_requestvote(r, raft_get_node(r, 2), &rv, &rvr);
CuAssertTrue(tc, 1 == rvr.vote_granted);
}
/*
* tests:
* 1) if we set timeout_now, then periodic function should always declare election, even when not timed out yet
*/
void TestRaft_targeted_node_becomes_candidate_when_before_real_timeout_occurs(CuTest * tc)
{
raft_cbs_t funcs = {
.persist_term = __raft_persist_term,
.persist_vote = __raft_persist_vote,
.send_requestvote = __raft_send_requestvote,
};
raft_server_t *r = raft_new();
raft_set_callbacks(r, &funcs, NULL);
raft_set_timeout_now(r);
/* 1 second election timeout */
raft_set_election_timeout(r, 1000);
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
/* i.e. not a timeout, only 1 out of 1000 */
raft_periodic(r, 1);
/* is a candidate now */
CuAssertTrue(tc, 1 == raft_is_precandidate(r));
}
void quorum_msg_id_correctness_cb(void* arg, int can_read)
{
(void) can_read;
......@@ -4009,6 +4079,81 @@ void TestRaft_quorum_msg_id_correctness(CuTest * tc)
CuAssertIntEquals(tc, 1, val);
}
int timeoutnow_sent = 0;
int __fake_timeoutnow(raft_server_t* raft, raft_node_t* node)
{
timeoutnow_sent = 1;
return 0;
}
void TestRaft_callback_timeoutnow_at_set_if_up_to_date(CuTest *tc)
{
timeoutnow_sent = 0;
raft_cbs_t funcs = {
.send_timeoutnow = __fake_timeoutnow,
};
raft_server_t *r = raft_new();
raft_set_callbacks(r, &funcs, NULL);
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_set_state(r, RAFT_STATE_LEADER);
raft_set_current_term(r, 2);
raft_set_commit_idx(r, 0);
raft_set_last_applied_idx(r, 0);
/* append entry to increment current_idx */
__RAFT_APPEND_ENTRIES_SEQ_ID(r, 2, 1, 1, "aaaa");
/* shouldn't trigger callback as match_idx isn't uptodate */
raft_transfer_leader(r, 2, 0);
CuAssertTrue(tc, 0 == timeoutnow_sent);
raft_reset_transfer_leader(r);
/* should trigger callback */
raft_node_set_match_idx(raft_get_node(r, 2), 2);
raft_transfer_leader(r, 2, 0);
CuAssertTrue(tc, 1 == timeoutnow_sent);
}
void TestRaft_callback_timeoutnow_at_send_appendentries_response_if_up_to_date(CuTest *tc)
{
timeoutnow_sent = 0;
raft_cbs_t funcs = {
.send_timeoutnow = __fake_timeoutnow,
};
raft_server_t *r = raft_new();
raft_set_callbacks(r, &funcs, NULL);
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_set_state(r, RAFT_STATE_LEADER);
raft_set_current_term(r, 2);
raft_set_commit_idx(r, 0);
raft_set_last_applied_idx(r, 0);
/* append entry to increment current_idx */
__RAFT_APPEND_ENTRIES_SEQ_ID(r, 2, 1, 1, "aaaa");
/* shouldn't trigger callback as match_idx isn't uptodate */
raft_transfer_leader(r, 2, 0);
CuAssertTrue(tc, 0 == timeoutnow_sent);
msg_appendentries_response_t aer;
aer.term = 2;
aer.success = 1;
aer.current_idx = 2;
raft_recv_appendentries_response(r, raft_get_node(r, 2), &aer);
CuAssertTrue(tc, 1 == timeoutnow_sent);
}
void TestRaft_leader_steps_down_if_there_is_no_quorum(CuTest * tc)
{
void *r = raft_new();
......@@ -4422,6 +4567,10 @@ int main(void)
SUITE_ADD_TEST(suite, TestRaft_unknown_node_can_become_leader);
SUITE_ADD_TEST(suite, TestRaft_removed_node_starts_election);
SUITE_ADD_TEST(suite, TestRaft_verify_append_entries_fields_are_set);
SUITE_ADD_TEST(suite, TestRaft_server_recv_requestvote_with_transfer_node);
SUITE_ADD_TEST(suite, TestRaft_targeted_node_becomes_candidate_when_before_real_timeout_occurs);
SUITE_ADD_TEST(suite, TestRaft_callback_timeoutnow_at_set_if_up_to_date);
SUITE_ADD_TEST(suite, TestRaft_callback_timeoutnow_at_send_appendentries_response_if_up_to_date);
CuSuiteRun(suite);
CuSuiteDetails(suite, output);
......
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