Commit 5d7915f3 authored by Yossi Gottlieb's avatar Yossi Gottlieb
Browse files

Add RAFT_LOGTYPE_NO_OP.

The no-op entry gets committed as soon as a new leader is elected, in
order to establish the commit index and prevent stale reads.
parent c3064082
...@@ -78,6 +78,11 @@ typedef enum { ...@@ -78,6 +78,11 @@ typedef enum {
* Users can piggyback the entry mechanism by specifying log types that * Users can piggyback the entry mechanism by specifying log types that
* are higher than RAFT_LOGTYPE_NUM. * are higher than RAFT_LOGTYPE_NUM.
*/ */
RAFT_LOGTYPE_NO_OP,
/**
* A no-op entry appended automatically when a leader begins a new term,
* in order to determine the current commit index.
*/
RAFT_LOGTYPE_NUM=100, RAFT_LOGTYPE_NUM=100,
} raft_logtype_e; } raft_logtype_e;
...@@ -946,7 +951,7 @@ int raft_apply_all(raft_server_t* me_); ...@@ -946,7 +951,7 @@ int raft_apply_all(raft_server_t* me_);
/** Become leader /** Become leader
* WARNING: this is a dangerous function call. It could lead to your cluster * WARNING: this is a dangerous function call. It could lead to your cluster
* losing it's consensus guarantees. */ * losing it's consensus guarantees. */
void raft_become_leader(raft_server_t* me); int raft_become_leader(raft_server_t* me);
/** Become follower. This may be used to give up leadership. It does not change /** Become follower. This may be used to give up leadership. It does not change
* currentTerm. */ * currentTerm. */
......
...@@ -160,13 +160,25 @@ int raft_election_start(raft_server_t* me_) ...@@ -160,13 +160,25 @@ int raft_election_start(raft_server_t* me_)
return raft_become_candidate(me_); return raft_become_candidate(me_);
} }
void raft_become_leader(raft_server_t* me_) int raft_become_leader(raft_server_t* me_)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
int i; int i;
__log(me_, NULL, "becoming leader term:%d", raft_get_current_term(me_)); __log(me_, NULL, "becoming leader term:%d", raft_get_current_term(me_));
raft_index_t next_idx = raft_get_current_idx(me_) + 1;
if (raft_get_current_term(me_) > 1) {
raft_entry_t *noop = raft_entry_new(0);
noop->term = raft_get_current_term(me_);
noop->type = RAFT_LOGTYPE_NO_OP;
int e = raft_append_entry(me_, noop);
if (0 != e)
return e;
raft_entry_release(noop);
}
raft_set_state(me_, RAFT_STATE_LEADER); raft_set_state(me_, RAFT_STATE_LEADER);
me->timeout_elapsed = 0; me->timeout_elapsed = 0;
for (i = 0; i < me->num_nodes; i++) for (i = 0; i < me->num_nodes; i++)
...@@ -176,10 +188,12 @@ void raft_become_leader(raft_server_t* me_) ...@@ -176,10 +188,12 @@ void raft_become_leader(raft_server_t* me_)
if (me->node == node || !raft_node_is_active(node)) if (me->node == node || !raft_node_is_active(node))
continue; continue;
raft_node_set_next_idx(node, raft_get_current_idx(me_) + 1); raft_node_set_next_idx(node, next_idx);
raft_node_set_match_idx(node, 0); raft_node_set_match_idx(node, 0);
raft_send_appendentries(me_, node); raft_send_appendentries(me_, node);
} }
return 0;
} }
int raft_become_candidate(raft_server_t* me_) int raft_become_candidate(raft_server_t* me_)
...@@ -234,8 +248,11 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period) ...@@ -234,8 +248,11 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period)
/* Only one voting node means it's safe for us to become the leader */ /* Only one voting node means it's safe for us to become the leader */
if (1 == raft_get_num_voting_nodes(me_) && if (1 == raft_get_num_voting_nodes(me_) &&
raft_node_is_voting(raft_get_my_node((void*)me)) && raft_node_is_voting(raft_get_my_node((void*)me)) &&
!raft_is_leader(me_)) !raft_is_leader(me_)) {
raft_become_leader(me_); int e = raft_become_leader(me_);
if (e != 0)
return e;
}
if (me->state == RAFT_STATE_LEADER) if (me->state == RAFT_STATE_LEADER)
{ {
...@@ -718,8 +735,11 @@ int raft_recv_requestvote_response(raft_server_t* me_, ...@@ -718,8 +735,11 @@ int raft_recv_requestvote_response(raft_server_t* me_,
if (node) if (node)
raft_node_vote_for_me(node, 1); raft_node_vote_for_me(node, 1);
int votes = raft_get_nvotes_for_me(me_); int votes = raft_get_nvotes_for_me(me_);
if (raft_votes_is_majority(raft_get_num_voting_nodes(me_), votes)) if (raft_votes_is_majority(raft_get_num_voting_nodes(me_), votes)) {
raft_become_leader(me_); int e = raft_become_leader(me_);
if (0 != e)
return e;
}
break; break;
case RAFT_REQUESTVOTE_ERR_NOT_GRANTED: case RAFT_REQUESTVOTE_ERR_NOT_GRANTED:
...@@ -1660,6 +1680,22 @@ void raft_process_read_queue(raft_server_t* me_) ...@@ -1660,6 +1680,22 @@ void raft_process_read_queue(raft_server_t* me_)
if (!raft_is_leader(me_)) if (!raft_is_leader(me_))
return; return;
/* Quickly bail if nothing to do */
if (!me->read_queue_head)
return;
if (raft_get_num_voting_nodes(me_) > 1) {
raft_entry_t *ety = raft_get_entry_from_idx(me_, raft_get_commit_idx(me_));
if (!ety)
return;
raft_term_t ety_term = ety->term;
raft_entry_release(ety);
if (ety_term < me->current_term)
return;
}
raft_msg_id_t acked_msgid = quorum_msg_id(me_); raft_msg_id_t acked_msgid = quorum_msg_id(me_);
while (me->read_queue_head && while (me->read_queue_head &&
me->read_queue_head->msg_id <= acked_msgid && me->read_queue_head->msg_id <= acked_msgid &&
......
...@@ -3696,7 +3696,7 @@ void TestRaft_read_action_callback( ...@@ -3696,7 +3696,7 @@ void TestRaft_read_action_callback(
* from past terms */ * from past terms */
raft_become_leader(r); raft_become_leader(r);
aer.msg_id = 2; aer.msg_id = 2;
aer.term = 2; aer.term = 3;
CuAssertIntEquals(tc, 0, raft_recv_appendentries_response(r, raft_get_node(r, 2), &aer)); CuAssertIntEquals(tc, 0, raft_recv_appendentries_response(r, raft_get_node(r, 2), &aer));
raft_periodic(r, 1); raft_periodic(r, 1);
...@@ -3705,6 +3705,7 @@ void TestRaft_read_action_callback( ...@@ -3705,6 +3705,7 @@ void TestRaft_read_action_callback(
/* entry 3 */ /* entry 3 */
__RAFT_APPEND_ENTRY(r, 3, 1, "aaa"); __RAFT_APPEND_ENTRY(r, 3, 1, "aaa");
ra.calls = 0; ra.calls = 0;
raft_queue_read_request(r, __read_request_callback, &ra); raft_queue_read_request(r, __read_request_callback, &ra);
......
...@@ -90,6 +90,8 @@ def logtype2str(log_type): ...@@ -90,6 +90,8 @@ def logtype2str(log_type):
return 'add_nonvoting' return 'add_nonvoting'
elif log_type == lib.RAFT_LOGTYPE_ADD_NODE: elif log_type == lib.RAFT_LOGTYPE_ADD_NODE:
return 'add' return 'add'
elif log_type == lib.RAFT_LOGTYPE_NO_OP:
return 'no_op'
else: else:
return 'unknown' return 'unknown'
...@@ -812,6 +814,9 @@ class RaftServer(object): ...@@ -812,6 +814,9 @@ class RaftServer(object):
if e is not None: if e is not None:
return e return e
if ety.type == lib.RAFT_LOGTYPE_NO_OP:
return 0
change = ffi.from_handle(lib.raft_entry_getdata(ety)) change = ffi.from_handle(lib.raft_entry_getdata(ety))
if ety.type == lib.RAFT_LOGTYPE_NORMAL: if ety.type == lib.RAFT_LOGTYPE_NORMAL:
...@@ -965,11 +970,23 @@ class RaftServer(object): ...@@ -965,11 +970,23 @@ class RaftServer(object):
This is a virtraft specific check to make sure entry passing is This is a virtraft specific check to make sure entry passing is
working correctly. working correctly.
""" """
if ety.type == lib.RAFT_LOGTYPE_NO_OP:
return
ci = lib.raft_get_current_idx(self.raft) ci = lib.raft_get_current_idx(self.raft)
if 0 < ci and not lib.raft_get_snapshot_last_idx(self.raft) == ci: if 0 < ci and not lib.raft_get_snapshot_last_idx(self.raft) == ci:
try: try:
prev_ety = lib.raft_get_entry_from_idx(self.raft, ci) prev_ety = lib.raft_get_entry_from_idx(self.raft, ci)
assert prev_ety assert prev_ety
if prev_ety.type == lib.RAFT_LOGTYPE_NO_OP:
if lib.raft_get_snapshot_last_idx(self.raft) != ci - 1:
lib.raft_entry_release(prev_ety)
prev_ety = lib.raft_get_entry_from_idx(self.raft,
ci - 1)
assert prev_ety
else:
lib.raft_entry_release(prev_ety)
return
other_id = prev_ety.id other_id = prev_ety.id
assert other_id < ety.id assert other_id < ety.id
lib.raft_entry_release(prev_ety) lib.raft_entry_release(prev_ety)
...@@ -1017,6 +1034,9 @@ class RaftServer(object): ...@@ -1017,6 +1034,9 @@ class RaftServer(object):
self.fsm_log.pop() self.fsm_log.pop()
self.network.log_pops += 1 self.network.log_pops += 1
if ety.type == lib.RAFT_LOGTYPE_NO_OP:
return 0
change = ffi.from_handle(lib.raft_entry_getdata(ety)) change = ffi.from_handle(lib.raft_entry_getdata(ety))
if ety.type == lib.RAFT_LOGTYPE_DEMOTE_NODE: if ety.type == lib.RAFT_LOGTYPE_DEMOTE_NODE:
pass pass
......
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