Unverified Commit 07190dc1 authored by Ozan Tezcan's avatar Ozan Tezcan Committed by GitHub
Browse files

Make leader transfer target start an election immediately (#88)

parent 5614926e
...@@ -40,10 +40,10 @@ typedef enum { ...@@ -40,10 +40,10 @@ typedef enum {
} raft_state_e; } raft_state_e;
typedef enum { typedef enum {
RAFT_STATE_LEADERSHIP_TRANSFER_TIMEOUT, RAFT_LEADER_TRANSFER_TIMEOUT,
RAFT_STATE_LEADERSHIP_TRANSFER_UNEXPECTED_LEADER, RAFT_LEADER_TRANSFER_UNEXPECTED_LEADER,
RAFT_STATE_LEADERSHIP_TRANSFER_EXPECTED_LEADER, RAFT_LEADER_TRANSFER_EXPECTED_LEADER,
} raft_transfer_state_e; } raft_leader_transfer_e;
/** Allow entries to apply while taking a snapshot */ /** Allow entries to apply while taking a snapshot */
#define RAFT_SNAPSHOT_NONBLOCKING_APPLY 1 #define RAFT_SNAPSHOT_NONBLOCKING_APPLY 1
...@@ -169,9 +169,6 @@ typedef struct ...@@ -169,9 +169,6 @@ typedef struct
/** term of candidate's last log entry */ /** term of candidate's last log entry */
raft_term_t last_log_term; 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; } msg_requestvote_t;
/** Vote request response message. /** Vote request response message.
...@@ -581,15 +578,15 @@ typedef void ( ...@@ -581,15 +578,15 @@ typedef void (
* *
* @param[in] raft The Raft server making this callback * @param[in] raft The Raft server making this callback
* @param[in] user_data User data that is passed from Raft server * @param[in] user_data User data that is passed from Raft server
* @param[in] state the leadership transfer result * @param[in] result the leadership transfer result
*/ */
typedef void ( typedef void (
*func_transfer_event_f *func_transfer_event_f
) ( ) (
raft_server_t* raft, raft_server_t* raft,
void *user_data, void *user_data,
raft_transfer_state_e state raft_leader_transfer_e result
); );
/** Callback for sending TimeoutNow RPC messages /** Callback for sending TimeoutNow RPC messages
* @param[in] raft The Raft server making this callback * @param[in] raft The Raft server making this callback
...@@ -599,8 +596,8 @@ typedef void ( ...@@ -599,8 +596,8 @@ typedef void (
typedef int ( typedef int (
*func_send_timeoutnow_f *func_send_timeoutnow_f
) ( ) (
raft_server_t* raft, raft_server_t* raft,
raft_node_t* node raft_node_t* node
); );
/** Callback to skip sending msg_appendentries to the node /** Callback to skip sending msg_appendentries to the node
...@@ -1530,8 +1527,8 @@ int raft_transfer_leader(raft_server_t* me_, raft_node_id_t node_id, long timeou ...@@ -1530,8 +1527,8 @@ int raft_transfer_leader(raft_server_t* me_, raft_node_id_t node_id, long timeou
/* get the targeted node_id if a leadership transfer is in progress, or RAFT_NODE_ID_NONE if not */ /* 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_); 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 */ /* cause this server to force an election. */
void raft_set_timeout_now(raft_server_t* me_); int raft_timeout_now(raft_server_t* me_);
raft_index_t raft_get_num_snapshottable_logs(raft_server_t* me_); raft_index_t raft_get_num_snapshottable_logs(raft_server_t* me_);
......
...@@ -128,12 +128,10 @@ typedef struct { ...@@ -128,12 +128,10 @@ typedef struct {
* when auto flush is disabled. */ * when auto flush is disabled. */
raft_index_t next_sync_index; raft_index_t next_sync_index;
int timeout_now;
int log_enabled; int log_enabled;
} raft_server_private_t; } raft_server_private_t;
int raft_election_start(raft_server_t* me); int raft_election_start(raft_server_t* me, int skip_precandidate);
int raft_become_candidate(raft_server_t* me); int raft_become_candidate(raft_server_t* me);
......
...@@ -379,7 +379,7 @@ int raft_delete_entry_from_idx(raft_server_t* me_, raft_index_t idx) ...@@ -379,7 +379,7 @@ int raft_delete_entry_from_idx(raft_server_t* me_, raft_index_t idx)
return me->log_impl->sync(me->log); return me->log_impl->sync(me->log);
} }
int raft_election_start(raft_server_t* me_) int raft_election_start(raft_server_t* me_, int skip_precandidate)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
...@@ -392,7 +392,8 @@ int raft_election_start(raft_server_t* me_) ...@@ -392,7 +392,8 @@ int raft_election_start(raft_server_t* me_)
me->timeout_elapsed = 0; me->timeout_elapsed = 0;
raft_randomize_election_timeout(me_); raft_randomize_election_timeout(me_);
return raft_become_precandidate(me_); return skip_precandidate ? raft_become_candidate(me_):
raft_become_precandidate(me_);
} }
void raft_accept_leader(raft_server_t* me_, raft_node_id_t leader) void raft_accept_leader(raft_server_t* me_, raft_node_id_t leader)
...@@ -646,12 +647,11 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period) ...@@ -646,12 +647,11 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period)
raft_update_quorum_meta(me_, quorum_id); raft_update_quorum_meta(me_, quorum_id);
} }
} }
else if ((me->election_timeout_rand <= me->timeout_elapsed || me->timeout_now)) else if (me->election_timeout_rand <= me->timeout_elapsed)
{ {
int e = raft_election_start(me_); int e = raft_election_start(me_, 0);
if (0 != e) if (0 != e)
return e; return e;
me->timeout_now = 0;
} }
if (me->last_applied_idx < raft_get_commit_idx(me_) && if (me->last_applied_idx < raft_get_commit_idx(me_) &&
...@@ -980,7 +980,6 @@ int raft_recv_requestvote(raft_server_t* me_, ...@@ -980,7 +980,6 @@ int raft_recv_requestvote(raft_server_t* me_,
/* Reject request if we have a leader, during prevote */ /* Reject request if we have a leader, during prevote */
if (vr->prevote && if (vr->prevote &&
!vr->transfer_leader &&
me->leader_id != RAFT_NODE_ID_NONE && me->leader_id != RAFT_NODE_ID_NONE &&
me->leader_id != vr->candidate_id && me->leader_id != vr->candidate_id &&
me->timeout_elapsed < me->election_timeout) { me->timeout_elapsed < me->election_timeout) {
...@@ -1179,7 +1178,6 @@ int raft_send_requestvote(raft_server_t* me_, raft_node_t* node) ...@@ -1179,7 +1178,6 @@ int raft_send_requestvote(raft_server_t* me_, raft_node_t* node)
rv.last_log_idx = raft_get_current_idx(me_); rv.last_log_idx = raft_get_current_idx(me_);
rv.last_log_term = raft_get_last_log_term(me_); rv.last_log_term = raft_get_last_log_term(me_);
rv.candidate_id = raft_get_nodeid(me_); rv.candidate_id = raft_get_nodeid(me_);
rv.transfer_leader = me->timeout_now;
if (me->cb.send_requestvote) if (me->cb.send_requestvote)
e = me->cb.send_requestvote(me_, me->udata, node, &rv); e = me->cb.send_requestvote(me_, me->udata, node, &rv);
...@@ -2040,8 +2038,8 @@ void raft_process_read_queue(raft_server_t* me_) ...@@ -2040,8 +2038,8 @@ void raft_process_read_queue(raft_server_t* me_)
} }
/* invoke a leadership transfer /* invoke a leadership transfer
* node_id = targeted node we are transfering to * node_id = targeted node we are transferring to
* timeout = how long this should be allowed to take in milliseconds, as calculated by calls to raft_periodic) * 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 * 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) int raft_transfer_leader(raft_server_t* me_, raft_node_id_t node_id, long timeout)
...@@ -2056,51 +2054,70 @@ int raft_transfer_leader(raft_server_t* me_, raft_node_id_t node_id, long timeou ...@@ -2056,51 +2054,70 @@ int raft_transfer_leader(raft_server_t* me_, raft_node_id_t node_id, long timeou
return RAFT_ERR_LEADER_TRANSFER_IN_PROGRESS; return RAFT_ERR_LEADER_TRANSFER_IN_PROGRESS;
} }
raft_node_t * target = raft_get_node(me_, node_id); raft_node_t *target = raft_get_node(me_, node_id);
if (target == NULL) { if (target == NULL || target == me->node) {
return RAFT_ERR_INVALID_NODEID; return RAFT_ERR_INVALID_NODEID;
} }
if (me->cb.send_timeoutnow && if (me->cb.send_timeoutnow &&
raft_get_current_idx(me_) == raft_node_get_match_idx(target)) { raft_get_current_idx(me_) == raft_node_get_match_idx(target)) {
me->cb.send_timeoutnow(me_, target); me->cb.send_timeoutnow(me_, target);
me->sent_timeout_now = 1; me->sent_timeout_now = 1;
} }
me->node_transferring_leader_to = node_id; me->node_transferring_leader_to = node_id;
if (timeout == 0) { me->transfer_leader_time = (timeout != 0) ? timeout : me->election_timeout;
me->transfer_leader_time = me->election_timeout;
} else {
me->transfer_leader_time = timeout;
}
return 0; return 0;
} }
/* Forces the raft server to invoke an election as part of leader transfer
* operation. */
int raft_timeout_now(raft_server_t* me_)
{
if (raft_is_leader(me_)) {
return 0;
}
/* Starting an election but pre-candidate round will be skipped. This node
* becomes candidate immediately. It means the term will be incremented and
* appendentries RPCs from the prior leader will be rejected. Otherwise,
* in pre-candidate round, this node would have to accept RPCs from the
* prior leader and abort the election. Also, by skipping pre-candidate
* round, leader transfer operation can be completed faster and cluster will
* experience less downtime.
*/
return raft_election_start(me_, 1);
}
/* Stop trying to transfer leader to a targeted node /* 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 * 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. * possible to be used by a client as well.
*/ */
void raft_reset_transfer_leader(raft_server_t* me_, int timed_out) void raft_reset_transfer_leader(raft_server_t* me_, int timed_out)
{ {
raft_leader_transfer_e result;
raft_server_private_t* me = (raft_server_private_t*) me_; raft_server_private_t* me = (raft_server_private_t*) me_;
if (me->node_transferring_leader_to != RAFT_NODE_ID_NONE) { if (me->node_transferring_leader_to == RAFT_NODE_ID_NONE) {
if (me->cb.notify_transfer_event) { return;
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; if (timed_out) {
me->transfer_leader_time = 0; result = RAFT_LEADER_TRANSFER_TIMEOUT;
me->sent_timeout_now = 0; } else {
result = (me->node_transferring_leader_to == me->leader_id) ?
RAFT_LEADER_TRANSFER_EXPECTED_LEADER :
RAFT_LEADER_TRANSFER_UNEXPECTED_LEADER;
}
if (me->cb.notify_transfer_event) {
me->cb.notify_transfer_event(me_, me->udata, result);
} }
me->node_transferring_leader_to = RAFT_NODE_ID_NONE;
me->transfer_leader_time = 0;
me->sent_timeout_now = 0;
} }
static int index_cmp(const void *a, const void *b) static int index_cmp(const void *a, const void *b)
......
...@@ -298,10 +298,3 @@ raft_node_id_t raft_get_transfer_leader(raft_server_t* me_) ...@@ -298,10 +298,3 @@ raft_node_id_t raft_get_transfer_leader(raft_server_t* me_)
return me->node_transferring_leader_to; 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;
}
...@@ -221,7 +221,7 @@ void TestRaft_election_start_does_not_increment_term(CuTest * tc) ...@@ -221,7 +221,7 @@ void TestRaft_election_start_does_not_increment_term(CuTest * tc)
raft_add_node(r, NULL, 1, 1); raft_add_node(r, NULL, 1, 1);
raft_set_callbacks(r, &generic_funcs, NULL); raft_set_callbacks(r, &generic_funcs, NULL);
raft_set_current_term(r, 1); raft_set_current_term(r, 1);
raft_election_start(r); raft_election_start(r, 0);
CuAssertTrue(tc, 1 == raft_get_current_term(r)); CuAssertTrue(tc, 1 == raft_get_current_term(r));
} }
...@@ -3670,7 +3670,7 @@ void TestRaft_leader_recv_requestvote_responds_without_granting(CuTest * tc) ...@@ -3670,7 +3670,7 @@ void TestRaft_leader_recv_requestvote_responds_without_granting(CuTest * tc)
raft_set_request_timeout(r, 500); raft_set_request_timeout(r, 500);
CuAssertTrue(tc, 0 == raft_get_timeout_elapsed(r)); CuAssertTrue(tc, 0 == raft_get_timeout_elapsed(r));
raft_election_start(r); raft_election_start(r, 0);
msg_requestvote_response_t rvr = { msg_requestvote_response_t rvr = {
.prevote = 1, .prevote = 1,
...@@ -3967,8 +3967,8 @@ void TestRaft_single_node_commits_noop(CuTest * tc) ...@@ -3967,8 +3967,8 @@ void TestRaft_single_node_commits_noop(CuTest * tc)
/* /*
* tests: * tests:
* 1) if requestvote transfer_leader flag is not set, will not accept a request vote, as not timed out * 1) if requestvote prevote flag is 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 * 2) if requestvote prevote flag is not set, will accept a request vote, even though not timed out
*/ */
void TestRaft_server_recv_requestvote_with_transfer_node(CuTest * tc) void TestRaft_server_recv_requestvote_with_transfer_node(CuTest * tc)
{ {
...@@ -3991,17 +3991,16 @@ void TestRaft_server_recv_requestvote_with_transfer_node(CuTest * tc) ...@@ -3991,17 +3991,16 @@ void TestRaft_server_recv_requestvote_with_transfer_node(CuTest * tc)
.term = 2, .term = 2,
.candidate_id = 2, .candidate_id = 2,
.last_log_idx = 0, .last_log_idx = 0,
.last_log_term = 1, .last_log_term = 1
.transfer_leader = 0,
}; };
msg_requestvote_response_t rvr; msg_requestvote_response_t rvr;
/* test #1: try to request a vote without transfer leader flag */ /* test #1: try to request a vote with prevote flag */
raft_recv_requestvote(r, raft_get_node(r, 2), &rv, &rvr); raft_recv_requestvote(r, raft_get_node(r, 2), &rv, &rvr);
CuAssertTrue(tc, 1 != rvr.vote_granted); CuAssertTrue(tc, 1 != rvr.vote_granted);
/* test #2: try to request a vote with the transfer leader flag */ /* test #2: try to request a vote without the prevote flag */
rv.transfer_leader = 1; rv.prevote = 0;
raft_recv_requestvote(r, raft_get_node(r, 2), &rv, &rvr); raft_recv_requestvote(r, raft_get_node(r, 2), &rv, &rvr);
CuAssertTrue(tc, 1 == rvr.vote_granted); CuAssertTrue(tc, 1 == rvr.vote_granted);
} }
...@@ -4019,21 +4018,12 @@ void TestRaft_targeted_node_becomes_candidate_when_before_real_timeout_occurs(Cu ...@@ -4019,21 +4018,12 @@ void TestRaft_targeted_node_becomes_candidate_when_before_real_timeout_occurs(Cu
}; };
raft_server_t *r = raft_new(); 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, 1, 1);
raft_add_node(r, NULL, 2, 0); raft_add_node(r, NULL, 2, 0);
raft_set_callbacks(r, &funcs, NULL);
/* i.e. not a timeout, only 1 out of 1000 */ raft_timeout_now(r);
raft_periodic(r, 1); CuAssertTrue(tc, 1 == raft_is_candidate(r));
/* is a candidate now */
CuAssertTrue(tc, 1 == raft_is_precandidate(r));
} }
void quorum_msg_id_correctness_cb(void* arg, int can_read) void quorum_msg_id_correctness_cb(void* arg, int can_read)
...@@ -4448,15 +4438,15 @@ void TestRaft_verify_append_entries_fields_are_set(CuTest * tc) ...@@ -4448,15 +4438,15 @@ void TestRaft_verify_append_entries_fields_are_set(CuTest * tc)
raft_send_appendentries_all(r); raft_send_appendentries_all(r);
} }
void cb_notify_transfer_event(raft_server_t *raft, void *udata, raft_transfer_state_e state) void cb_notify_transfer_event(raft_server_t *raft, void *udata, raft_leader_transfer_e state)
{ {
raft_transfer_state_e * data = udata; raft_leader_transfer_e * data = udata;
*data = state; *data = state;
} }
void Test_reset_transfer_leader(CuTest *tc) void Test_reset_transfer_leader(CuTest *tc)
{ {
raft_transfer_state_e state; raft_leader_transfer_e state;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.notify_transfer_event = cb_notify_transfer_event, .notify_transfer_event = cb_notify_transfer_event,
}; };
...@@ -4471,25 +4461,25 @@ void Test_reset_transfer_leader(CuTest *tc) ...@@ -4471,25 +4461,25 @@ void Test_reset_transfer_leader(CuTest *tc)
int ret = raft_transfer_leader(r, 2, 0); int ret = raft_transfer_leader(r, 2, 0);
CuAssertIntEquals(tc, 0, ret); CuAssertIntEquals(tc, 0, ret);
raft_reset_transfer_leader(r, 0); raft_reset_transfer_leader(r, 0);
CuAssertIntEquals(tc, RAFT_STATE_LEADERSHIP_TRANSFER_UNEXPECTED_LEADER, state); CuAssertIntEquals(tc, RAFT_LEADER_TRANSFER_UNEXPECTED_LEADER, state);
ret = raft_transfer_leader(r, 2, 0); ret = raft_transfer_leader(r, 2, 0);
CuAssertIntEquals(tc, 0, ret); CuAssertIntEquals(tc, 0, ret);
me->leader_id = 2; me->leader_id = 2;
raft_reset_transfer_leader(r, 0); raft_reset_transfer_leader(r, 0);
CuAssertIntEquals(tc, RAFT_STATE_LEADERSHIP_TRANSFER_EXPECTED_LEADER, state); CuAssertIntEquals(tc, RAFT_LEADER_TRANSFER_EXPECTED_LEADER, state);
/* tests timeout in general, so don't need a separate test for it */ /* tests timeout in general, so don't need a separate test for it */
me->leader_id = 1; me->leader_id = 1;
ret = raft_transfer_leader(r, 2, 1); ret = raft_transfer_leader(r, 2, 1);
CuAssertIntEquals(tc, 0, ret); CuAssertIntEquals(tc, 0, ret);
raft_periodic(r, 2); raft_periodic(r, 2);
CuAssertIntEquals(tc, RAFT_STATE_LEADERSHIP_TRANSFER_TIMEOUT, state); CuAssertIntEquals(tc, RAFT_LEADER_TRANSFER_TIMEOUT, state);
} }
void Test_transfer_leader_success(CuTest *tc) void Test_transfer_leader_success(CuTest *tc)
{ {
raft_transfer_state_e state = RAFT_STATE_LEADERSHIP_TRANSFER_TIMEOUT; raft_leader_transfer_e state = RAFT_LEADER_TRANSFER_TIMEOUT;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.notify_transfer_event = cb_notify_transfer_event, .notify_transfer_event = cb_notify_transfer_event,
}; };
...@@ -4509,12 +4499,12 @@ void Test_transfer_leader_success(CuTest *tc) ...@@ -4509,12 +4499,12 @@ void Test_transfer_leader_success(CuTest *tc)
raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer); raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer);
CuAssertTrue(tc, 1 == aer.success); CuAssertTrue(tc, 1 == aer.success);
CuAssertIntEquals(tc, RAFT_STATE_LEADERSHIP_TRANSFER_EXPECTED_LEADER, state); CuAssertIntEquals(tc, RAFT_LEADER_TRANSFER_EXPECTED_LEADER, state);
} }
void Test_transfer_leader_unexpected(CuTest *tc) void Test_transfer_leader_unexpected(CuTest *tc)
{ {
raft_transfer_state_e state = RAFT_STATE_LEADERSHIP_TRANSFER_TIMEOUT; raft_leader_transfer_e state = RAFT_LEADER_TRANSFER_TIMEOUT;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.notify_transfer_event = cb_notify_transfer_event, .notify_transfer_event = cb_notify_transfer_event,
}; };
...@@ -4536,7 +4526,7 @@ void Test_transfer_leader_unexpected(CuTest *tc) ...@@ -4536,7 +4526,7 @@ void Test_transfer_leader_unexpected(CuTest *tc)
raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer); raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer);
CuAssertTrue(tc, 1 == aer.success); CuAssertTrue(tc, 1 == aer.success);
CuAssertIntEquals(tc, RAFT_STATE_LEADERSHIP_TRANSFER_UNEXPECTED_LEADER, state); CuAssertIntEquals(tc, RAFT_LEADER_TRANSFER_UNEXPECTED_LEADER, state);
} }
void Test_transfer_leader_not_leader(CuTest *tc) void Test_transfer_leader_not_leader(CuTest *tc)
......
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