Unverified Commit 63b0b1c1 authored by Shaya Potter's avatar Shaya Potter Committed by GitHub
Browse files

redo/tweak transfer leader in libraft (#61)

this started off as a small tweak and grew a bit

1) new callback for transfer leader result, along with its own enum of states

2) change/remove raft_reset_transfer_leader() calls

only call in places where there's a new leader or timeout (i.e. failed to transfer within period).

so, we only reset in 3 places 1) recv_appendentries 2) raft_periodic (timeout) 3) raft_become_leader

2a) raft_become_leader also got refactored a little to only issue the normal cb.notify_state_event() callback after it has set its state to be leader (as before, it could fail after the callback was issued)

2b) in raft_periodic the raft_reset_transfer_leader() is pulled out the "if LEADER" block, as will lose leader when a vote is issued to it, we still want it to timeout if leadership isn't gained

3) raft_reset_transfer_leader() does the logic for determining success/failure now, but becaue timeout is its own result which can't be determined by just looking at leader/desired, add a new flag to it to note timeout state

4) leader stickiness in requestvote is only tested for prevote (as in a timeout now case, we want to drop the timeout now flag after prevotes are sent).  The idea is that we will only get to actual election if prevote passes.

4) a bunch of new tests

* fix issue with timeout flag to reset_transfer_leader

also fix test to catch the original error

* libraft/redisraft changes needed for e2e leader transfer

* make leader stickiness only for prevote
parent 4f94a895
......@@ -36,9 +36,14 @@ typedef enum {
RAFT_STATE_PRECANDIDATE,
RAFT_STATE_CANDIDATE,
RAFT_STATE_LEADER,
RAFT_STATE_LEADERSHIP_TRANSFER_FAILED
} raft_state_e;
typedef enum {
RAFT_STATE_LEADERSHIP_TRANSFER_TIMEOUT,
RAFT_STATE_LEADERSHIP_TRANSFER_UNEXPECTED_LEADER,
RAFT_STATE_LEADERSHIP_TRANSFER_EXPECTED_LEADER,
} raft_transfer_state_e;
/** Allow entries to apply while taking a snapshot */
#define RAFT_SNAPSHOT_NONBLOCKING_APPLY 1
#define RAFT_NODE_ID_NONE (-1)
......@@ -423,6 +428,22 @@ typedef void (
raft_state_e state
);
/** Call for being notified of leadership transfer events.
*
* Implementing this callback is optional
*
* @param[in] raft The Raft server making this callback
* @param[in] user_data User data that is passed from Raft server
* @param[in] state the leadership transfer result
*/
typedef void (
*func_transfer_event_f
) (
raft_server_t* raft,
void *user_data,
raft_transfer_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
......@@ -474,6 +495,9 @@ typedef struct
/** Callback for being notified of state changes (optional). */
func_state_event_f notify_state_event;
/** Callbakc for notified of transfer leadership events (optional) */
func_transfer_event_f notify_transfer_event;
/** Callback for catching debugging log messages
* This callback is optional */
func_log_f log;
......@@ -1285,9 +1309,6 @@ void raft_process_read_queue(raft_server_t* me_);
*/
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_);
......
......@@ -118,6 +118,7 @@ typedef struct {
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 sent_timeout_now; // if we've already sent a leadership transfer signal
int timeout_now;
} raft_server_private_t;
......@@ -184,5 +185,7 @@ raft_msg_id_t raft_node_get_max_seen_msg_id(raft_node_t *me_);
/* get the server's current msg_id */
raft_msg_id_t raft_get_msg_id(raft_server_t* me_);
/* attempt to abort the leadership transfer */
void raft_reset_transfer_leader(raft_server_t* me_, int timed_out);
#endif /* RAFT_PRIVATE_H_ */
......@@ -379,10 +379,6 @@ int raft_become_leader(raft_server_t* me_)
int i;
raft_log(me_, "becoming leader term:%ld", raft_get_current_term(me_));
if (me->cb.notify_state_event)
me->cb.notify_state_event(me_, raft_get_udata(me_), RAFT_STATE_LEADER);
raft_reset_transfer_leader(me_);
raft_index_t next_idx = raft_get_current_idx(me_) + 1;
......@@ -406,6 +402,11 @@ int raft_become_leader(raft_server_t* me_)
raft_update_quorum_meta(me_, me->msg_id);
me->timeout_elapsed = 0;
raft_reset_transfer_leader(me_, 0);
if (me->cb.notify_state_event)
me->cb.notify_state_event(me_, raft_get_udata(me_), RAFT_STATE_LEADER);
for (i = 0; i < me->num_nodes; i++)
{
raft_node_t* node = me->nodes[i];
......@@ -440,8 +441,9 @@ int raft_become_precandidate(raft_server_t* me_)
{
raft_node_t* node = me->nodes[i];
if (me->node != node && raft_node_is_voting(node))
if (me->node != node && raft_node_is_voting(node)) {
raft_send_requestvote(me_, node);
}
}
return 0;
......@@ -456,8 +458,6 @@ int raft_become_candidate(raft_server_t* me_)
if (me->cb.notify_state_event)
me->cb.notify_state_event(me_, raft_get_udata(me_), RAFT_STATE_CANDIDATE);
raft_reset_transfer_leader(me_);
int e = raft_set_current_term(me_, raft_get_current_term(me_) + 1);
if (0 != e)
return e;
......@@ -490,8 +490,6 @@ void raft_become_follower(raft_server_t* me_)
if (me->cb.notify_state_event)
me->cb.notify_state_event(me_, raft_get_udata(me_), RAFT_STATE_FOLLOWER);
raft_reset_transfer_leader(me_);
raft_set_state(me_, RAFT_STATE_FOLLOWER);
raft_randomize_election_timeout(me_);
me->timeout_elapsed = 0;
......@@ -557,6 +555,14 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period)
}
}
/* needs to be outside state check, as can become a followr and still timeout */
if (me->node_transferring_leader_to != RAFT_NODE_ID_NONE) {
me->transfer_leader_time -= msec_since_last_period;
if (me->transfer_leader_time < 0) {
raft_reset_transfer_leader(me_, 1);
}
}
if (me->state == RAFT_STATE_LEADER)
{
if (me->request_timeout <= me->timeout_elapsed)
......@@ -589,16 +595,7 @@ 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 || me->timeout_now) &&
/* Don't become the leader when building snapshots or bad things will
......@@ -608,7 +605,6 @@ 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;
}
......@@ -694,9 +690,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)
if (me->cb.send_timeoutnow && raft_get_transfer_leader(me_) == raft_node_get_id(node) && !me->sent_timeout_now
&& raft_get_current_idx(me_) == r->current_idx) {
me->cb.send_timeoutnow(me_, node);
me->sent_timeout_now = 1;
}
if (!raft_node_is_voting(node) &&
......@@ -806,6 +803,7 @@ int raft_recv_appendentries(
/* update current leader because ae->term is up to date */
me->leader_id = ae->leader_id;
raft_reset_transfer_leader(me_, 0);
me->timeout_elapsed = 0;
/* Not the first appendentries we've received */
......@@ -940,8 +938,9 @@ int raft_recv_requestvote(raft_server_t* me_,
r->request_term = vr->term;
r->vote_granted = 0;
/* Reject request if we have a leader */
if (!vr->transfer_leader &&
/* Reject request if we have a leader, during prevote */
if (vr->prevote &&
!vr->transfer_leader &&
me->leader_id != RAFT_NODE_ID_NONE &&
me->leader_id != vr->candidate_id &&
me->timeout_elapsed < me->election_timeout) {
......@@ -1157,7 +1156,7 @@ 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);
rv.transfer_leader = me->timeout_now;
if (me->cb.send_requestvote)
e = me->cb.send_requestvote(me_, me->udata, node, &rv);
......@@ -1830,6 +1829,10 @@ int raft_transfer_leader(raft_server_t* me_, raft_node_id_t node_id, long timeou
{
raft_server_private_t* me = (raft_server_private_t*) me_;
if (me->state != RAFT_STATE_LEADER) {
return RAFT_ERR_NOT_LEADER;
}
if (me->node_transferring_leader_to != RAFT_NODE_ID_NONE) {
return RAFT_ERR_LEADER_TRANSFER_IN_PROGRESS;
}
......@@ -1842,6 +1845,7 @@ int raft_transfer_leader(raft_server_t* me_, raft_node_id_t node_id, long timeou
if (me->cb.send_timeoutnow &&
raft_get_current_idx(me_) == raft_node_get_match_idx(target)) {
me->cb.send_timeoutnow(me_, target);
me->sent_timeout_now = 1;
}
me->node_transferring_leader_to = node_id;
......@@ -1853,3 +1857,29 @@ int raft_transfer_leader(raft_server_t* me_, raft_node_id_t node_id, long timeou
return 0;
}
/* 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_, int timed_out)
{
raft_server_private_t* me = (raft_server_private_t*) me_;
if (me->node_transferring_leader_to != RAFT_NODE_ID_NONE) {
if (me->cb.notify_transfer_event) {
raft_transfer_state_e state = RAFT_STATE_LEADERSHIP_TRANSFER_EXPECTED_LEADER;
if (me->node_transferring_leader_to != me->leader_id) {
state = RAFT_STATE_LEADERSHIP_TRANSFER_UNEXPECTED_LEADER;
}
if (timed_out) {
state = RAFT_STATE_LEADERSHIP_TRANSFER_TIMEOUT;
}
me->cb.notify_transfer_event(me_, raft_get_udata(me_), state);
}
me->node_transferring_leader_to = RAFT_NODE_ID_NONE;
me->transfer_leader_time = 0;
me->sent_timeout_now = 0;
}
}
\ No newline at end of file
......@@ -289,18 +289,6 @@ raft_msg_id_t raft_get_msg_id(raft_server_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
*/
......
......@@ -983,6 +983,7 @@ void TestRaft_server_recv_requestvote_ignore_if_master_is_fresh(CuTest * tc)
CuAssertTrue(tc, 1 == aer.success);
msg_requestvote_t rv = {
.prevote = 1,
.term = 2,
.candidate_id = 3,
.last_log_idx = 0,
......@@ -3975,6 +3976,7 @@ void TestRaft_server_recv_requestvote_with_transfer_node(CuTest * tc)
/* setup requestvote struct */
msg_requestvote_t rv = {
.prevote = 1,
.term = 2,
.candidate_id = 2,
.last_log_idx = 0,
......@@ -4110,14 +4112,16 @@ void TestRaft_callback_timeoutnow_at_set_if_up_to_date(CuTest *tc)
__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);
int ret = raft_transfer_leader(r, 2, 0);
CuAssertIntEquals(tc, 0, ret);
CuAssertTrue(tc, 0 == timeoutnow_sent);
raft_reset_transfer_leader(r);
raft_reset_transfer_leader(r, 0);
/* should trigger callback */
raft_node_set_match_idx(raft_get_node(r, 2), 2);
raft_transfer_leader(r, 2, 0);
ret = raft_transfer_leader(r, 2, 0);
CuAssertIntEquals(tc, 0, ret);
CuAssertTrue(tc, 1 == timeoutnow_sent);
}
......@@ -4143,7 +4147,8 @@ void TestRaft_callback_timeoutnow_at_send_appendentries_response_if_up_to_date(C
__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);
int ret = raft_transfer_leader(r, 2, 0);
CuAssertIntEquals(tc, 0, ret);
CuAssertTrue(tc, 0 == timeoutnow_sent);
msg_appendentries_response_t aer;
......@@ -4432,6 +4437,109 @@ void TestRaft_verify_append_entries_fields_are_set(CuTest * tc)
raft_send_appendentries_all(r);
}
void cb_notify_transfer_event(raft_server_t *raft, void *udata, raft_transfer_state_e state)
{
raft_transfer_state_e * data = udata;
*data = state;
}
void Test_reset_transfer_leader(CuTest *tc)
{
raft_transfer_state_e state;
raft_cbs_t funcs = {
.notify_transfer_event = cb_notify_transfer_event,
};
raft_server_t *r = raft_new();
raft_set_state(r, RAFT_STATE_LEADER);
raft_server_private_t * me = (raft_server_private_t *) r;
raft_set_callbacks(r, &funcs, &state);
raft_add_node(r, NULL, 100, 1);
raft_add_node(r, NULL, 2, 0);
int ret = raft_transfer_leader(r, 2, 0);
CuAssertIntEquals(tc, 0, ret);
raft_reset_transfer_leader(r, 0);
CuAssertIntEquals(tc, RAFT_STATE_LEADERSHIP_TRANSFER_UNEXPECTED_LEADER, state);
ret = raft_transfer_leader(r, 2, 0);
CuAssertIntEquals(tc, 0, ret);
me->leader_id = 2;
raft_reset_transfer_leader(r, 0);
CuAssertIntEquals(tc, RAFT_STATE_LEADERSHIP_TRANSFER_EXPECTED_LEADER, state);
/* tests timeout in general, so don't need a separate test for it */
me->leader_id = 1;
ret = raft_transfer_leader(r, 2, 1);
CuAssertIntEquals(tc, 0, ret);
raft_periodic(r, 2);
CuAssertIntEquals(tc, RAFT_STATE_LEADERSHIP_TRANSFER_TIMEOUT, state);
}
void Test_transfer_leader_success(CuTest *tc)
{
raft_transfer_state_e state = RAFT_STATE_LEADERSHIP_TRANSFER_TIMEOUT;
raft_cbs_t funcs = {
.notify_transfer_event = cb_notify_transfer_event,
};
raft_server_t *r = raft_new();
raft_set_callbacks(r, &funcs, &state);
raft_add_node(r, NULL, 100, 1);
raft_add_node(r, NULL, 2, 0);
raft_set_state(r, RAFT_STATE_LEADER);
int ret = raft_transfer_leader(r, 2, 0);
CuAssertIntEquals(tc, 0, ret);
msg_appendentries_t ae = { 0 };
msg_appendentries_response_t aer;
ae.term = 1;
ae.leader_id = 2;
raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer);
CuAssertTrue(tc, 1 == aer.success);
CuAssertIntEquals(tc, RAFT_STATE_LEADERSHIP_TRANSFER_EXPECTED_LEADER, state);
}
void Test_transfer_leader_unexpected(CuTest *tc)
{
raft_transfer_state_e state = RAFT_STATE_LEADERSHIP_TRANSFER_TIMEOUT;
raft_cbs_t funcs = {
.notify_transfer_event = cb_notify_transfer_event,
};
raft_server_t *r = raft_new();
raft_set_callbacks(r, &funcs, &state);
raft_add_node(r, NULL, 100, 1);
raft_add_node(r, NULL, 2, 0);
raft_add_node(r, NULL, 3, 0);
raft_set_state(r, RAFT_STATE_LEADER);
int ret = raft_transfer_leader(r, 3, 0);
CuAssertIntEquals(tc, 0, ret);
msg_appendentries_t ae = { 0 };
msg_appendentries_response_t aer;
ae.term = 1;
ae.leader_id = 2;
raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer);
CuAssertTrue(tc, 1 == aer.success);
CuAssertIntEquals(tc, RAFT_STATE_LEADERSHIP_TRANSFER_UNEXPECTED_LEADER, state);
}
void Test_transfer_leader_not_leader(CuTest *tc)
{
raft_server_t *r = raft_new();
raft_add_node(r, NULL, 100, 1);
raft_add_node(r, NULL, 2, 0);
raft_add_node(r, NULL, 3, 0);
int ret = raft_transfer_leader(r, 3, 0);
CuAssertIntEquals(tc, RAFT_ERR_NOT_LEADER, ret);
}
int main(void)
{
CuString *output = CuStringNew();
......@@ -4571,7 +4679,10 @@ int main(void)
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);
SUITE_ADD_TEST(suite, Test_reset_transfer_leader);
SUITE_ADD_TEST(suite, Test_transfer_leader_success);
SUITE_ADD_TEST(suite, Test_transfer_leader_unexpected);
SUITE_ADD_TEST(suite, Test_transfer_leader_not_leader);
CuSuiteRun(suite);
CuSuiteDetails(suite, output);
printf("%s\n", output->buffer);
......
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