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

Refactor debug logging (#127)

Refactor debug logging and change log callback

e.g:
19 --> 13, sent snapshot_resp id:27712, t:24, s:1, o:4096, lc:0
14 --> 17, sent requestvote_req pv:1, t:25, ci:14, lli:6848, llt:23
14 <-- 17, recv requestvote_resp pv:1, t:25, rt:24, vg:0

Breaking changes: 

raft_log_f() callback will not pass node_id_t anymore.
Sender and receiver will be part of the log line as you can see above.
parent c0270e61
...@@ -441,15 +441,12 @@ typedef int ( ...@@ -441,15 +441,12 @@ typedef int (
/** Callback for providing debug logging information. /** Callback for providing debug logging information.
* This callback is optional * This callback is optional
* @param[in] raft The Raft server making this callback * @param[in] raft The Raft server making this callback
* @param[in] node_id The node id that is the subject of this log. If log is not
* related to a node, this could be 'RAFT_NODE_ID_NONE'.
* @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] buf The buffer that was logged */ * @param[in] buf The buffer that was logged */
typedef void ( typedef void (
*raft_log_f *raft_log_f
) ( ) (
raft_server_t* raft, raft_server_t* raft,
raft_node_id_t node_id,
void *user_data, void *user_data,
const char *buf const char *buf
); );
...@@ -991,10 +988,10 @@ int raft_periodic(raft_server_t *me); ...@@ -991,10 +988,10 @@ int raft_periodic(raft_server_t *me);
* @return * @return
* 0 on success * 0 on success
* */ * */
int raft_recv_appendentries(raft_server_t* me, int raft_recv_appendentries(raft_server_t *me,
raft_node_t* node, raft_node_t *node,
raft_appendentries_req_t* req, raft_appendentries_req_t *req,
raft_appendentries_resp_t* resp); raft_appendentries_resp_t *resp);
/** Receive a response from an appendentries message we sent. /** Receive a response from an appendentries message we sent.
* @param[in] node The node who sent us this message * @param[in] node The node who sent us this message
...@@ -1003,9 +1000,9 @@ int raft_recv_appendentries(raft_server_t* me, ...@@ -1003,9 +1000,9 @@ int raft_recv_appendentries(raft_server_t* me,
* 0 on success; * 0 on success;
* -1 on error; * -1 on error;
* RAFT_ERR_NOT_LEADER server is not the leader */ * RAFT_ERR_NOT_LEADER server is not the leader */
int raft_recv_appendentries_response(raft_server_t* me, int raft_recv_appendentries_response(raft_server_t *me,
raft_node_t* node, raft_node_t *node,
raft_appendentries_resp_t* resp); raft_appendentries_resp_t *resp);
/** Receive a snapshot message. /** Receive a snapshot message.
* @param[in] node The node who sent us this message * @param[in] node The node who sent us this message
...@@ -1025,8 +1022,8 @@ int raft_recv_snapshot(raft_server_t* me, ...@@ -1025,8 +1022,8 @@ int raft_recv_snapshot(raft_server_t* me,
* 0 on success; * 0 on success;
* -1 on error; * -1 on error;
* RAFT_ERR_NOT_LEADER server is not the leader */ * RAFT_ERR_NOT_LEADER server is not the leader */
int raft_recv_snapshot_response(raft_server_t* me, int raft_recv_snapshot_response(raft_server_t *me,
raft_node_t* node, raft_node_t *node,
raft_snapshot_resp_t *resp); raft_snapshot_resp_t *resp);
/** Receive a requestvote message. /** Receive a requestvote message.
...@@ -1034,20 +1031,20 @@ int raft_recv_snapshot_response(raft_server_t* me, ...@@ -1034,20 +1031,20 @@ int raft_recv_snapshot_response(raft_server_t* me,
* @param[in] req The requestvote message * @param[in] req The requestvote message
* @param[out] resp The resulting response * @param[out] resp The resulting response
* @return 0 on success */ * @return 0 on success */
int raft_recv_requestvote(raft_server_t* me, int raft_recv_requestvote(raft_server_t *me,
raft_node_t* node, raft_node_t *node,
raft_requestvote_req_t* req, raft_requestvote_req_t *req,
raft_requestvote_resp_t* resp); raft_requestvote_resp_t *resp);
/** Receive a response from a requestvote message we sent. /** Receive a response from a requestvote message we sent.
* @param[in] node The node this response was sent by * @param[in] node The node this response was sent by
* @param[in] req The requestvote response message * @param[in] resp The requestvote response message
* @return * @return
* 0 on success; * 0 on success;
* RAFT_ERR_SHUTDOWN server MUST shutdown; */ * RAFT_ERR_SHUTDOWN server MUST shutdown; */
int raft_recv_requestvote_response(raft_server_t* me, int raft_recv_requestvote_response(raft_server_t *me,
raft_node_t* node, raft_node_t *node,
raft_requestvote_resp_t* req); raft_requestvote_resp_t *resp);
/** Receive an entry message from the client. /** Receive an entry message from the client.
* *
...@@ -1073,7 +1070,7 @@ int raft_recv_requestvote_response(raft_server_t* me, ...@@ -1073,7 +1070,7 @@ int raft_recv_requestvote_response(raft_server_t* me,
* RAFT_ERR_ONE_VOTING_CHANGE_ONLY there is a non-voting change inflight; * RAFT_ERR_ONE_VOTING_CHANGE_ONLY there is a non-voting change inflight;
* RAFT_ERR_NOMEM memory allocation failure * RAFT_ERR_NOMEM memory allocation failure
*/ */
int raft_recv_entry(raft_server_t* me, int raft_recv_entry(raft_server_t *me,
raft_entry_req_t *req, raft_entry_req_t *req,
raft_entry_resp_t *resp); raft_entry_resp_t *resp);
...@@ -1284,11 +1281,11 @@ int raft_apply_all(raft_server_t* me); ...@@ -1284,11 +1281,11 @@ 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. */
int 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. */
void raft_become_follower(raft_server_t* me); void raft_become_follower(raft_server_t *me);
/** Determine if entry is voting configuration change. /** Determine if entry is voting configuration change.
* @param[in] ety The entry to query. * @param[in] ety The entry to query.
......
...@@ -133,13 +133,13 @@ struct raft_server { ...@@ -133,13 +133,13 @@ struct raft_server {
int disable_apply; /* Do not apply entries, useful for testing */ int disable_apply; /* Do not apply entries, useful for testing */
}; };
int raft_election_start(raft_server_t* me, int skip_precandidate); 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);
int raft_become_precandidate(raft_server_t* me); int raft_become_precandidate(raft_server_t *me);
void raft_randomize_election_timeout(raft_server_t* me); void raft_randomize_election_timeout(raft_server_t *me);
void raft_update_quorum_meta(raft_server_t* me, raft_msg_id_t id); void raft_update_quorum_meta(raft_server_t* me, raft_msg_id_t id);
......
...@@ -47,16 +47,12 @@ void raft_set_heap_functions(void *(*_malloc)(size_t), ...@@ -47,16 +47,12 @@ void raft_set_heap_functions(void *(*_malloc)(size_t),
raft_free = _free; raft_free = _free;
} }
static void raft_log_node(raft_server_t *me, __attribute__ ((format (printf, 2, 3)))
raft_node_id_t id, static void raft_log(raft_server_t *me, const char *fmt, ...)
const char *fmt, ...) __attribute__ ((format (printf, 3, 4)));
static void raft_log_node(raft_server_t *me,
raft_node_id_t id,
const char *fmt, ...)
{ {
if (!me->log_enabled || me->cb.log == NULL) if (!me->log_enabled || me->cb.log == NULL) {
return; return;
}
char buf[1024]; char buf[1024];
va_list args; va_list args;
...@@ -68,16 +64,16 @@ static void raft_log_node(raft_server_t *me, ...@@ -68,16 +64,16 @@ static void raft_log_node(raft_server_t *me,
} }
va_end(args); va_end(args);
me->cb.log(me, id, me->udata, buf); me->cb.log(me, me->udata, buf);
} }
#define raft_log(me, ...) (raft_log_node(me, RAFT_NODE_ID_NONE, __VA_ARGS__)) void raft_randomize_election_timeout(raft_server_t *me)
void raft_randomize_election_timeout(raft_server_t* me)
{ {
raft_time_t random = (rand() % me->election_timeout);
/* [election_timeout, 2 * election_timeout) */ /* [election_timeout, 2 * election_timeout) */
me->election_timeout_rand = me->election_timeout + rand() % me->election_timeout; me->election_timeout_rand = me->election_timeout + random;
raft_log(me, "randomize election timeout to %lld", me->election_timeout_rand); raft_log(me, "randomized election timeout:%lld", me->election_timeout_rand);
} }
void raft_update_quorum_meta(raft_server_t* me, raft_msg_id_t id) void raft_update_quorum_meta(raft_server_t* me, raft_msg_id_t id)
...@@ -394,12 +390,10 @@ int raft_delete_entry_from_idx(raft_server_t *me, raft_index_t idx) ...@@ -394,12 +390,10 @@ 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 skip_precandidate) int raft_election_start(raft_server_t *me, int skip_precandidate)
{ {
raft_log(me, raft_log(me, "election starting, timeout_elapsed:%lld, term:%ld",
"election starting: %lld %lld, term: %ld ci: %ld", me->timeout_elapsed, me->current_term);
me->election_timeout_rand, me->timeout_elapsed, me->current_term,
raft_get_current_idx(me));
me->leader_id = RAFT_NODE_ID_NONE; me->leader_id = RAFT_NODE_ID_NONE;
me->timeout_elapsed = 0; me->timeout_elapsed = 0;
...@@ -423,10 +417,8 @@ void raft_accept_leader(raft_server_t* me, raft_node_id_t leader) ...@@ -423,10 +417,8 @@ void raft_accept_leader(raft_server_t* me, raft_node_id_t leader)
me->leader_id = leader; me->leader_id = leader;
} }
int raft_become_leader(raft_server_t* me) int raft_become_leader(raft_server_t *me)
{ {
raft_log(me, "becoming leader term: %ld", me->current_term);
raft_entry_t *noop = raft_entry_new(0); raft_entry_t *noop = raft_entry_new(0);
noop->term = me->current_term; noop->term = me->current_term;
noop->type = RAFT_LOGTYPE_NO_OP; noop->type = RAFT_LOGTYPE_NO_OP;
...@@ -458,10 +450,6 @@ int raft_become_leader(raft_server_t* me) ...@@ -458,10 +450,6 @@ int raft_become_leader(raft_server_t* me)
raft_reset_transfer_leader(me, 0); raft_reset_transfer_leader(me, 0);
me->timeout_elapsed = 0; me->timeout_elapsed = 0;
if (me->cb.notify_state_event) {
me->cb.notify_state_event(me, me->udata, RAFT_STATE_LEADER);
}
for (int i = 0; i < me->num_nodes; i++) { for (int i = 0; i < me->num_nodes; i++) {
raft_node_t *node = me->nodes[i]; raft_node_t *node = me->nodes[i];
...@@ -475,81 +463,88 @@ int raft_become_leader(raft_server_t* me) ...@@ -475,81 +463,88 @@ int raft_become_leader(raft_server_t* me)
raft_send_appendentries(me, node); raft_send_appendentries(me, node);
} }
if (me->cb.notify_state_event) {
me->cb.notify_state_event(me, me->udata, RAFT_STATE_LEADER);
}
raft_log(me, "become leader, term:%ld", me->current_term);
return 0; return 0;
} }
int raft_become_precandidate(raft_server_t* me) int raft_become_precandidate(raft_server_t *me)
{ {
raft_log(me, for (int i = 0; i < me->num_nodes; i++) {
"becoming pre-candidate, next term : %ld", me->current_term + 1);
if (me->cb.notify_state_event)
me->cb.notify_state_event(me, me->udata, RAFT_STATE_PRECANDIDATE);
for (int i = 0; i < me->num_nodes; i++)
raft_node_vote_for_me(me->nodes[i], 0); raft_node_vote_for_me(me->nodes[i], 0);
}
raft_set_state(me, RAFT_STATE_PRECANDIDATE); raft_set_state(me, RAFT_STATE_PRECANDIDATE);
for (int i = 0; i < me->num_nodes; i++) for (int i = 0; i < me->num_nodes; i++) {
{ raft_node_t *node = me->nodes[i];
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); raft_send_requestvote(me, node);
} }
} }
if (me->cb.notify_state_event) {
me->cb.notify_state_event(me, me->udata, RAFT_STATE_PRECANDIDATE);
}
raft_log(me, "become pre-candidate, term:%ld", me->current_term);
return 0; return 0;
} }
int raft_become_candidate(raft_server_t* me) int raft_become_candidate(raft_server_t *me)
{ {
int i;
raft_log(me, "becoming candidate");
if (me->cb.notify_state_event) {
me->cb.notify_state_event(me, me->udata, RAFT_STATE_CANDIDATE);
}
int e = raft_set_current_term(me, me->current_term + 1); int e = raft_set_current_term(me, me->current_term + 1);
if (0 != e) if (e != 0) {
return e; return e;
for (i = 0; i < me->num_nodes; i++) }
for (int i = 0; i < me->num_nodes; i++) {
raft_node_vote_for_me(me->nodes[i], 0); raft_node_vote_for_me(me->nodes[i], 0);
}
if (raft_node_is_voting(me->node)) if (raft_node_is_voting(me->node)) {
raft_vote(me, me->node); raft_vote(me, me->node);
}
me->leader_id = RAFT_NODE_ID_NONE; me->leader_id = RAFT_NODE_ID_NONE;
raft_set_state(me, RAFT_STATE_CANDIDATE); raft_set_state(me, RAFT_STATE_CANDIDATE);
for (i = 0; i < me->num_nodes; i++) for (int i = 0; i < me->num_nodes; i++) {
{ raft_node_t *node = me->nodes[i];
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); raft_send_requestvote(me, node);
} }
} }
if (me->cb.notify_state_event) {
me->cb.notify_state_event(me, me->udata, RAFT_STATE_CANDIDATE);
}
raft_log(me, "become candidate, term:%ld", me->current_term);
return 0; return 0;
} }
void raft_become_follower(raft_server_t* me) void raft_become_follower(raft_server_t* me)
{ {
raft_log(me, "becoming follower");
if (me->cb.notify_state_event) {
me->cb.notify_state_event(me, me->udata, RAFT_STATE_FOLLOWER);
}
raft_set_state(me, RAFT_STATE_FOLLOWER); raft_set_state(me, RAFT_STATE_FOLLOWER);
raft_randomize_election_timeout(me); raft_randomize_election_timeout(me);
raft_clear_incoming_snapshot(me, 0); raft_clear_incoming_snapshot(me, 0);
me->timeout_elapsed = 0; me->timeout_elapsed = 0;
me->leader_id = RAFT_NODE_ID_NONE; me->leader_id = RAFT_NODE_ID_NONE;
if (me->cb.notify_state_event) {
me->cb.notify_state_event(me, me->udata, RAFT_STATE_FOLLOWER);
}
raft_log(me, "become follower, term:%ld", me->current_term);
} }
static int msgid_cmp(const void *a, const void *b) static int msgid_cmp(const void *a, const void *b)
...@@ -701,49 +696,48 @@ int raft_voting_change_is_in_progress(raft_server_t* me) ...@@ -701,49 +696,48 @@ int raft_voting_change_is_in_progress(raft_server_t* me)
return me->voting_cfg_change_log_idx != -1; return me->voting_cfg_change_log_idx != -1;
} }
int raft_recv_appendentries_response(raft_server_t* me, int raft_recv_appendentries_response(raft_server_t *me,
raft_node_t* node, raft_node_t *node,
raft_appendentries_resp_t* r) raft_appendentries_resp_t *resp)
{ {
raft_log_node(me, raft_node_get_id(node), raft_log(me, "%d <-- %d, recv appendentries_resp "
"received appendentries response %s ci:%ld rci:%ld msgid:%lu", "id:%lu, t:%ld, s:%d, ci:%ld",
r->success == 1 ? "SUCCESS" : "fail", raft_get_nodeid(me), raft_node_get_id(node),
raft_get_current_idx(me), resp->msg_id, resp->term, resp->success, resp->current_idx);
r->current_idx, r->msg_id);
if (!node) if (!node) {
return -1; return -1;
}
if (!raft_is_leader(me)) if (!raft_is_leader(me)) {
return RAFT_ERR_NOT_LEADER; return RAFT_ERR_NOT_LEADER;
}
if (raft_node_get_match_msgid(node) > r->msg_id) { if (resp->msg_id < raft_node_get_match_msgid(node) ||
// this was received out of order and is now irrelevant. resp->term < me->current_term) {
return 0; return 0;
} }
/* If response contains term T > currentTerm: set currentTerm = T /* If response contains term T > currentTerm: set currentTerm = T
and convert to follower (§5.3) */ and convert to follower (§5.3) */
if (me->current_term < r->term) if (resp->term > me->current_term) {
{ int e = raft_set_current_term(me, resp->term);
int e = raft_set_current_term(me, r->term); if (e != 0) {
if (0 != e)
return e; return e;
raft_become_follower(me); }
raft_become_follower(me);
return 0; return 0;
} }
else if (me->current_term != r->term)
return 0;
if (0 == r->success) if (resp->success == 0) {
{
/* Stale response -- ignore */ /* Stale response -- ignore */
if (r->current_idx < raft_node_get_match_idx(node)) if (resp->current_idx < raft_node_get_match_idx(node)) {
return 0; return 0;
}
raft_index_t next = min(r->current_idx + 1, raft_get_current_idx(me)); raft_index_t current_idx = raft_get_current_idx(me);
raft_index_t next = min(resp->current_idx + 1, current_idx);
assert(0 < next); assert(0 < next);
raft_node_set_next_idx(node, next); raft_node_set_next_idx(node, next);
...@@ -753,41 +747,46 @@ int raft_recv_appendentries_response(raft_server_t* me, ...@@ -753,41 +747,46 @@ int raft_recv_appendentries_response(raft_server_t* me,
return 0; return 0;
} }
if (me->cb.send_timeoutnow && raft_get_transfer_leader(me) == raft_node_get_id(node) && !me->sent_timeout_now if (raft_get_transfer_leader(me) == raft_node_get_id(node) &&
&& raft_get_current_idx(me) == r->current_idx) { raft_get_current_idx(me) == resp->current_idx) {
me->cb.send_timeoutnow(me, me->udata, node); me->cb.send_timeoutnow(me, me->udata, node);
me->sent_timeout_now = 1; me->sent_timeout_now = 1;
raft_log(me, "%d --> %d, sent timeout_now", raft_get_nodeid(me),
raft_node_get_id(node));
} }
if (!raft_node_is_voting(node) && if (!raft_node_is_voting(node) &&
!raft_voting_change_is_in_progress(me) && !raft_voting_change_is_in_progress(me) &&
raft_get_current_idx(me) <= r->current_idx + 1 && raft_get_current_idx(me) <= resp->current_idx + 1 &&
!raft_node_is_voting_committed(node) && !raft_node_is_voting_committed(node) &&
raft_node_is_addition_committed(node) && raft_node_is_addition_committed(node) &&
me->cb.node_has_sufficient_logs && raft_node_has_sufficient_logs(node) == 0) {
0 == raft_node_has_sufficient_logs(node)
) if (me->cb.node_has_sufficient_logs) {
{
int e = me->cb.node_has_sufficient_logs(me, me->udata, node); int e = me->cb.node_has_sufficient_logs(me, me->udata, node);
if (e == 0) { if (e == 0) {
raft_node_set_has_sufficient_logs(node, 1); raft_node_set_has_sufficient_logs(node, 1);
} }
} }
}
raft_index_t match_idx = raft_node_get_match_idx(node); raft_index_t match_idx = raft_node_get_match_idx(node);
if (r->current_idx > match_idx) { if (resp->current_idx > match_idx) {
raft_node_set_match_idx(node, r->current_idx); raft_node_set_match_idx(node, resp->current_idx);
} }
assert(r->current_idx <= raft_get_current_idx(me)); assert(resp->current_idx <= raft_get_current_idx(me));
raft_msg_id_t match_msgid = raft_node_get_match_msgid(node); raft_msg_id_t match_msgid = raft_node_get_match_msgid(node);
if (r->msg_id > match_msgid) { if (resp->msg_id > match_msgid) {
raft_node_set_match_msgid(node, r->msg_id); raft_node_set_match_msgid(node, resp->msg_id);
} }
assert(r->msg_id <= me->msg_id); assert(resp->msg_id <= me->msg_id);
if (me->auto_flush) if (me->auto_flush) {
return raft_flush(me, 0); return raft_flush(me, 0);
}
return 0; return 0;
} }
...@@ -816,29 +815,28 @@ static int raft_receive_term(raft_server_t* me, raft_term_t term) ...@@ -816,29 +815,28 @@ static int raft_receive_term(raft_server_t* me, raft_term_t term)
return 0; return 0;
} }
int raft_recv_appendentries(raft_server_t* me, int raft_recv_appendentries(raft_server_t *me,
raft_node_t* node, raft_node_t *node,
raft_appendentries_req_t* ae, raft_appendentries_req_t *req,
raft_appendentries_resp_t* r) raft_appendentries_resp_t *resp)
{ {
int e = 0; int e = 0;
raft_log_node(me, ae->leader_id, raft_log(me, "%d <-- %d, recv appendentries_req "
"recv appendentries li:%d t:%ld ci:%ld lc:%ld pli:%ld plt:%ld msgid:%ld #%ld", "lead:%d, id:%ld, t:%ld, pli:%ld, plt:%ld, lc:%ld ent:%ld",
ae->leader_id, ae->term, raft_get_current_idx(me), raft_get_nodeid(me), raft_node_get_id(node),
ae->leader_commit, ae->prev_log_idx, ae->prev_log_term, req->leader_id, req->msg_id, req->term, req->prev_log_idx,
ae->msg_id, ae->n_entries); req->prev_log_term, req->leader_commit, req->n_entries);
r->msg_id = ae->msg_id; resp->msg_id = req->msg_id;
r->success = 0; resp->success = 0;
e = raft_receive_term(me, ae->term); e = raft_receive_term(me, req->term);
if (e != 0) { if (e != 0) {
if (e == RAFT_ERR_STALE_TERM) { if (e == RAFT_ERR_STALE_TERM) {
/* 1. Reply false if term < currentTerm (§5.1) */ /* 1. Reply false if term < currentTerm (§5.1) */
raft_log_node(me, ae->leader_id, raft_log(me, "AE term:%ld is less than current term:%ld",
"AE term %ld is less than current term %ld", ae->term, req->term, me->current_term);
me->current_term);
e = 0; e = 0;
} }
...@@ -846,181 +844,189 @@ int raft_recv_appendentries(raft_server_t* me, ...@@ -846,181 +844,189 @@ int raft_recv_appendentries(raft_server_t* me,
} }
/* update current leader because ae->term is up to date */ /* update current leader because ae->term is up to date */
raft_accept_leader(me, ae->leader_id); raft_accept_leader(me, req->leader_id);
raft_reset_transfer_leader(me, 0); raft_reset_transfer_leader(me, 0);
/* Not the first appendentries we've received */ /* Not the first appendentries we've received */
/* NOTE: the log starts at 1 */ /* NOTE: the log starts at 1 */
if (0 < ae->prev_log_idx) if (req->prev_log_idx > 0) {
{ raft_entry_t *ety = raft_get_entry_from_idx(me, req->prev_log_idx);
raft_entry_t* ety = raft_get_entry_from_idx(me, ae->prev_log_idx);
/* Is a snapshot */ /* Is a snapshot */
if (ae->prev_log_idx == me->snapshot_last_idx) if (req->prev_log_idx == me->snapshot_last_idx) {
{ if (me->snapshot_last_term != req->prev_log_term) {
if (me->snapshot_last_term != ae->prev_log_term)
{
/* Should never happen; something is seriously wrong! */ /* Should never happen; something is seriously wrong! */
raft_log_node(me, ae->leader_id, raft_log(me, "AE prev_log_term:%ld conflicts with snapshot term:%ld",
"Snapshot AE prev conflicts with committed entry"); req->prev_log_term, me->snapshot_last_term);
e = RAFT_ERR_SHUTDOWN; e = RAFT_ERR_SHUTDOWN;
if (ety) if (ety) {
raft_entry_release(ety); raft_entry_release(ety);
}
goto out; goto out;
} }
} }
/* 2. Reply false if log doesn't contain an entry at prevLogIndex /* 2. Reply false if log doesn't contain an entry at prevLogIndex
whose term matches prevLogTerm (§5.3) */ whose term matches prevLogTerm (§5.3) */
else if (!ety) else if (!ety) {
{ raft_log(me, "AE no log at prev_log_idx:%ld", req->prev_log_idx);
raft_log_node(me, ae->leader_id,
"AE no log at prev_idx %ld", ae->prev_log_idx);
goto out; goto out;
} } else if (ety->term != req->prev_log_term) {
else if (ety->term != ae->prev_log_term) raft_log(me, "AE prev_log_term:%ld doesn't match entry term:%ld",
{ req->prev_log_term, ety->term);
raft_log_node(me, ae->leader_id, "AE term doesn't match prev_term (ie. %ld vs %ld) ci:%ld comi:%ld lcomi:%ld pli:%ld",
ety->term, ae->prev_log_term, raft_get_current_idx(me), if (req->prev_log_idx <= me->commit_idx) {
me->commit_idx, ae->leader_commit, ae->prev_log_idx);
if (ae->prev_log_idx <= me->commit_idx)
{
/* Should never happen; something is seriously wrong! */ /* Should never happen; something is seriously wrong! */
raft_log_node(me, ae->leader_id, raft_log(me, "AE prev_log_idx:%ld is less than commit idx:%ld",
"AE prev conflicts with committed entry"); req->prev_log_idx, me->commit_idx);
e = RAFT_ERR_SHUTDOWN; e = RAFT_ERR_SHUTDOWN;
raft_entry_release(ety); raft_entry_release(ety);
goto out; goto out;
} }
/* Delete all the following log entries because they don't match */ /* Delete all the following log entries because they don't match */
e = raft_delete_entry_from_idx(me, ae->prev_log_idx); e = raft_delete_entry_from_idx(me, req->prev_log_idx);
raft_entry_release(ety); raft_entry_release(ety);
goto out; goto out;
} }
if (ety)
if (ety) {
raft_entry_release(ety); raft_entry_release(ety);
} }
}
r->success = 1; resp->success = 1;
r->current_idx = ae->prev_log_idx; resp->current_idx = req->prev_log_idx;
/* Synchronize msg_id to leader. Not really needed for raft, but needed for /* Synchronize msg_id to leader. Not really needed for raft, but needed for
* virtraft for msg_id to be increasing cluster wide so that can verify * virtraft for msg_id to be increasing cluster wide so that can verify
* read_queue correctness easily. Otherwise, it'd be fine for msg_id to be * read_queue correctness easily. Otherwise, it'd be fine for msg_id to be
* unique to each raft_server_t. * unique to each raft_server_t.
*/ */
if (me->msg_id < ae->msg_id) { if (me->msg_id < req->msg_id) {
me->msg_id = ae->msg_id; me->msg_id = req->msg_id;
} }
/* 3. If an existing entry conflicts with a new one (same index /* 3. If an existing entry conflicts with a new one (same index
but different terms), delete the existing entry and all that but different terms), delete the existing entry and all that
follow it (§5.3) */ follow it (§5.3) */
int i; int i;
for (i = 0; i < ae->n_entries; i++) for (i = 0; i < req->n_entries; i++) {
{ raft_entry_t *ety = req->entries[i];
raft_entry_t* ety = ae->entries[i]; raft_index_t ety_index = req->prev_log_idx + 1 + i;
raft_index_t ety_index = ae->prev_log_idx + 1 + i;
raft_entry_t* existing_ety = raft_get_entry_from_idx(me, ety_index); raft_entry_t *existing_ety = raft_get_entry_from_idx(me, ety_index);
raft_term_t existing_term = existing_ety ? existing_ety->term : 0; raft_term_t existing_term = existing_ety ? existing_ety->term : 0;
if (existing_ety) if (existing_ety) {
raft_entry_release(existing_ety); raft_entry_release(existing_ety);
}
if (existing_ety && existing_term != ety->term) if (existing_ety && existing_term != ety->term) {
{ if (ety_index <= me->commit_idx) {
if (ety_index <= me->commit_idx)
{
/* Should never happen; something is seriously wrong! */ /* Should never happen; something is seriously wrong! */
raft_log_node(me, ae->leader_id, "AE entry conflicts with committed entry ci:%ld comi:%ld lcomi:%ld pli:%ld", raft_log(me, "AE entry index:%ld is less than commit idx:%ld",
raft_get_current_idx(me), me->commit_idx, ety_index, me->commit_idx);
ae->leader_commit, ae->prev_log_idx);
e = RAFT_ERR_SHUTDOWN; e = RAFT_ERR_SHUTDOWN;
goto out; goto out;
} }
e = raft_delete_entry_from_idx(me, ety_index); e = raft_delete_entry_from_idx(me, ety_index);
if (0 != e) if (e != 0) {
goto out; goto out;
break;
} }
else if (!existing_ety)
break; break;
r->current_idx = ety_index; } else if (!existing_ety) {
break;
}
resp->current_idx = ety_index;
} }
/* Pick up remainder in case of mismatch or missing entry */ /* Pick up remainder in case of mismatch or missing entry */
for (; i < ae->n_entries; i++) for (; i < req->n_entries; i++) {
{ e = raft_append_entry(me, req->entries[i]);
e = raft_append_entry(me, ae->entries[i]); if (e != 0) {
if (0 != e)
goto out; goto out;
r->current_idx = ae->prev_log_idx + 1 + i; }
resp->current_idx = req->prev_log_idx + 1 + i;
} }
if (ae->n_entries > 0) { if (req->n_entries > 0) {
e = me->log_impl->sync(me->log); e = me->log_impl->sync(me->log);
if (0 != e) if (e != 0) {
goto out; goto out;
} }
}
/* 4. If leaderCommit > commitIndex, set commitIndex = /* 4. If leaderCommit > commitIndex, set commitIndex =
min(leaderCommit, index of most recent entry) */ min(leaderCommit, index of most recent entry) */
if (me->commit_idx < ae->leader_commit) if (me->commit_idx < req->leader_commit) {
{
raft_index_t last_log_idx = max(raft_get_current_idx(me), 1); raft_index_t last_log_idx = max(raft_get_current_idx(me), 1);
raft_set_commit_idx(me, min(last_log_idx, ae->leader_commit)); raft_set_commit_idx(me, min(last_log_idx, req->leader_commit));
} }
out: out:
/* 'node' might be created after processing this message, fetching again. */ /* 'node' might be created after processing this message, fetching again. */
node = raft_get_node(me, ae->leader_id); node = raft_get_node(me, req->leader_id);
if (node != NULL) { if (node != NULL) {
raft_node_update_max_seen_msg_id(node, ae->msg_id); raft_node_update_max_seen_msg_id(node, req->msg_id);
} }
r->term = me->current_term; resp->term = me->current_term;
if (0 == r->success) if (resp->success == 0) {
r->current_idx = raft_get_current_idx(me); resp->current_idx = raft_get_current_idx(me);
}
raft_log(me, "%d --> %d, sent appendentries_resp "
"id:%lu, t:%ld, s:%d, ci:%ld",
raft_get_nodeid(me), raft_node_get_id(node),
resp->msg_id, resp->term, resp->success, resp->current_idx);
return e; return e;
} }
int raft_recv_requestvote(raft_server_t* me, int raft_recv_requestvote(raft_server_t *me,
raft_node_t* node, raft_node_t *node,
raft_requestvote_req_t* vr, raft_requestvote_req_t *req,
raft_requestvote_resp_t* r) raft_requestvote_resp_t *resp)
{ {
(void) node; (void) node;
int e = 0; int e = 0;
r->prevote = vr->prevote; raft_log(me, "%d <-- %d, recv requestvote_req "
r->request_term = vr->term; "pv:%d, t:%ld, ci:%d, lli:%ld, llt:%ld",
r->vote_granted = 0; raft_get_nodeid(me), raft_node_get_id(node),
req->prevote, req->term, req->candidate_id, req->last_log_idx,
req->last_log_term);
resp->prevote = req->prevote;
resp->request_term = req->term;
resp->vote_granted = 0;
/* Reject request if we have a leader, during prevote */ /* Reject request if we have a leader, during prevote */
if (vr->prevote && if (req->prevote &&
me->leader_id != RAFT_NODE_ID_NONE && me->leader_id != RAFT_NODE_ID_NONE &&
me->leader_id != vr->candidate_id && me->leader_id != req->candidate_id &&
me->timeout_elapsed < me->election_timeout) { me->timeout_elapsed < me->election_timeout) {
goto done; goto done;
} }
/* Update the term only if this is not a prevote request */ /* Update the term only if this is not a prevote request */
if (!vr->prevote && me->current_term < vr->term) if (!req->prevote && me->current_term < req->term) {
{ e = raft_set_current_term(me, req->term);
e = raft_set_current_term(me, vr->term); if (e != 0) {
if (0 != e) {
goto done; goto done;
} }
raft_become_follower(me); raft_become_follower(me);
} }
if (me->current_term > vr->term) { if (me->current_term > req->term) {
goto done; goto done;
} }
if (me->current_term == vr->term) { if (me->current_term == req->term) {
/* If already voted for some other node for this term, return failure */ /* If already voted for some other node for this term, return failure */
if (me->voted_for != RAFT_NODE_ID_NONE && if (me->voted_for != RAFT_NODE_ID_NONE &&
me->voted_for != vr->candidate_id) { me->voted_for != req->candidate_id) {
goto done; goto done;
} }
} }
...@@ -1029,22 +1035,22 @@ int raft_recv_requestvote(raft_server_t* me, ...@@ -1029,22 +1035,22 @@ int raft_recv_requestvote(raft_server_t* me,
raft_index_t current_idx = raft_get_current_idx(me); raft_index_t current_idx = raft_get_current_idx(me);
raft_term_t ety_term = raft_get_last_log_term(me); raft_term_t ety_term = raft_get_last_log_term(me);
if (vr->last_log_term < ety_term || if (req->last_log_term < ety_term ||
(vr->last_log_term == ety_term && vr->last_log_idx < current_idx)) { (req->last_log_term == ety_term && req->last_log_idx < current_idx)) {
goto done; goto done;
} }
r->vote_granted = 1; resp->vote_granted = 1;
if (!vr->prevote) if (!req->prevote) {
{
/* It shouldn't be possible for a leader or candidate to grant a vote /* It shouldn't be possible for a leader or candidate to grant a vote
* Both states would have voted for themselves */ * Both states would have voted for themselves */
assert(!(raft_is_leader(me) || raft_is_candidate(me))); assert(!(raft_is_leader(me) || raft_is_candidate(me)));
e = raft_vote_for_nodeid(me, vr->candidate_id); e = raft_vote_for_nodeid(me, req->candidate_id);
if (0 != e) if (e != 0) {
r->vote_granted = 0; resp->vote_granted = 0;
}
/* must be in an election. */ /* must be in an election. */
me->leader_id = RAFT_NODE_ID_NONE; me->leader_id = RAFT_NODE_ID_NONE;
...@@ -1052,14 +1058,13 @@ int raft_recv_requestvote(raft_server_t* me, ...@@ -1052,14 +1058,13 @@ int raft_recv_requestvote(raft_server_t* me,
} }
done: done:
raft_log_node(me, vr->candidate_id, "node requested vote: %d, t:%ld, pv:%d replying: %s", resp->term = me->current_term;
vr->candidate_id,
vr->term, raft_log(me, "%d --> %d, sent requestvote_resp "
vr->prevote, "pv:%d, t:%ld, rt:%ld, vg:%d",
r->vote_granted == 1 ? "granted" : raft_get_nodeid(me), raft_node_get_id(node), resp->prevote,
r->vote_granted == 0 ? "not granted" : "unknown"); resp->request_term, resp->term, resp->vote_granted);
r->term = me->current_term;
return e; return e;
} }
...@@ -1071,43 +1076,42 @@ int raft_votes_is_majority(const int num_nodes, const int nvotes) ...@@ -1071,43 +1076,42 @@ int raft_votes_is_majority(const int num_nodes, const int nvotes)
return half + 1 <= nvotes; return half + 1 <= nvotes;
} }
int raft_recv_requestvote_response(raft_server_t* me, int raft_recv_requestvote_response(raft_server_t *me,
raft_node_t* node, raft_node_t *node,
raft_requestvote_resp_t* r) raft_requestvote_resp_t *resp)
{ {
raft_log_node(me, raft_node_get_id(node), raft_log(me, "%d <-- %d, recv requestvote_resp "
"node responded to requestvote status:%s pv:%d rt:%ld ct:%ld rt:%ld", "pv:%d, t:%ld, rt:%ld, vg:%d",
r->vote_granted == 1 ? "granted" : raft_get_nodeid(me), raft_node_get_id(node),
r->vote_granted == 0 ? "not granted" : "unknown", resp->prevote, resp->request_term, resp->term, resp->vote_granted);
r->prevote,
r->request_term,
me->current_term,
r->term);
if (r->term > me->current_term) if (resp->term > me->current_term) {
{ int e = raft_set_current_term(me, resp->term);
int e = raft_set_current_term(me, r->term); if (e != 0) {
if (0 != e)
return e; return e;
raft_become_follower(me); }
raft_become_follower(me);
return 0; return 0;
} }
if (r->prevote) { if (resp->prevote) {
/* Validate prevote is not stale */ /* Validate prevote is not stale */
if (!raft_is_precandidate(me) || r->request_term != me->current_term + 1) if (!raft_is_precandidate(me) ||
resp->request_term != me->current_term + 1) {
return 0; return 0;
}
} else { } else {
/* Validate reqvote is not stale */ /* Validate reqvote is not stale */
if (!raft_is_candidate(me) || r->request_term != me->current_term) if (!raft_is_candidate(me) || resp->request_term != me->current_term) {
return 0; return 0;
} }
}
if (r->vote_granted) if (resp->vote_granted) {
{ 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);
int nodes = raft_get_num_voting_nodes(me); int nodes = raft_get_num_voting_nodes(me);
...@@ -1115,43 +1119,46 @@ int raft_recv_requestvote_response(raft_server_t* me, ...@@ -1115,43 +1119,46 @@ int raft_recv_requestvote_response(raft_server_t* me,
if (raft_votes_is_majority(nodes, votes)) { if (raft_votes_is_majority(nodes, votes)) {
int e = raft_is_precandidate(me) ? raft_become_candidate(me) : int e = raft_is_precandidate(me) ? raft_become_candidate(me) :
raft_become_leader(me); raft_become_leader(me);
if (0 != e) if (e != 0) {
return e; return e;
} }
} }
}
return 0; return 0;
} }
int raft_recv_entry(raft_server_t* me, int raft_recv_entry(raft_server_t *me,
raft_entry_req_t* ety, raft_entry_req_t *ety,
raft_entry_resp_t* r) raft_entry_resp_t *r)
{ {
if (raft_entry_is_voting_cfg_change(ety)) if (raft_entry_is_voting_cfg_change(ety)) {
{
/* Only one voting cfg change at a time */ /* Only one voting cfg change at a time */
if (raft_voting_change_is_in_progress(me)) if (raft_voting_change_is_in_progress(me)) {
return RAFT_ERR_ONE_VOTING_CHANGE_ONLY; return RAFT_ERR_ONE_VOTING_CHANGE_ONLY;
}
/* Multi-threading: need to fail here because user might be /* Multi-threading: need to fail here because user might be
* snapshotting membership settings. */ * snapshotting membership settings. */
if (!raft_is_apply_allowed(me)) if (!raft_is_apply_allowed(me)) {
return RAFT_ERR_SNAPSHOT_IN_PROGRESS; return RAFT_ERR_SNAPSHOT_IN_PROGRESS;
} }
}
if (!raft_is_leader(me)) if (!raft_is_leader(me)) {
return RAFT_ERR_NOT_LEADER; return RAFT_ERR_NOT_LEADER;
}
if (raft_get_transfer_leader(me) != RAFT_NODE_ID_NONE) if (raft_get_transfer_leader(me) != RAFT_NODE_ID_NONE) {
return RAFT_ERR_LEADER_TRANSFER_IN_PROGRESS; 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);
ety->term = me->current_term; ety->term = me->current_term;
int e = raft_append_entry(me, ety); int e = raft_append_entry(me, ety);
if (0 != e) if (e != 0) {
return e; return e;
}
if (r) { if (r) {
r->id = ety->id; r->id = ety->id;
...@@ -1159,45 +1166,50 @@ int raft_recv_entry(raft_server_t* me, ...@@ -1159,45 +1166,50 @@ int raft_recv_entry(raft_server_t* me,
r->term = me->current_term; r->term = me->current_term;
} }
raft_log(me, "recv entry t:%ld, id:%d, type:%d, len:%d. assigned idx:%ld",
ety->term, ety->id, ety->type, ety->data_len,
raft_get_current_idx(me));
if (me->auto_flush) { if (me->auto_flush) {
e = me->log_impl->sync(me->log); e = me->log_impl->sync(me->log);
if (0 != e) if (e != 0) {
return e; return e;
}
return raft_flush(me, raft_get_current_idx(me)); return raft_flush(me, raft_get_current_idx(me));
} }
return 0; return 0;
} }
int raft_send_requestvote(raft_server_t* me, raft_node_t* node) int raft_send_requestvote(raft_server_t *me, raft_node_t *node)
{ {
raft_requestvote_req_t rv;
int e = 0; int e = 0;
raft_requestvote_req_t req;
assert(node); assert(node);
assert(node != me->node); assert(node != me->node);
raft_node_id_t id = raft_node_get_id(node); if (raft_is_precandidate(me)) {
raft_log_node(me, id, "sending requestvote to: %d", id); req.prevote = 1;
req.term = me->current_term + 1;
if (raft_is_precandidate(me)) } else {
{ req.prevote = 0;
rv.prevote = 1; req.term = me->current_term;
rv.term = me->current_term + 1;
}
else
{
rv.prevote = 0;
rv.term = me->current_term;
} }
rv.last_log_idx = raft_get_current_idx(me); req.last_log_idx = raft_get_current_idx(me);
rv.last_log_term = raft_get_last_log_term(me); req.last_log_term = raft_get_last_log_term(me);
rv.candidate_id = raft_get_nodeid(me); req.candidate_id = raft_get_nodeid(me);
if (me->cb.send_requestvote) raft_log(me, "%d --> %d, sent requestvote_req "
e = me->cb.send_requestvote(me, me->udata, node, &rv); "pv:%d, t:%ld, ci:%d, lli:%ld, llt:%ld",
raft_get_nodeid(me), raft_node_get_id(node),
req.prevote, req.term, req.candidate_id, req.last_log_idx,
req.last_log_term);
if (me->cb.send_requestvote) {
e = me->cb.send_requestvote(me, me->udata, node, &req);
}
return e; return e;
} }
...@@ -1345,15 +1357,16 @@ raft_entry_t **raft_get_entries_from_idx(raft_server_t *me, ...@@ -1345,15 +1357,16 @@ raft_entry_t **raft_get_entries_from_idx(raft_server_t *me,
return e; return e;
} }
int raft_send_snapshot(raft_server_t* me, raft_node_t* node) int raft_send_snapshot(raft_server_t *me, raft_node_t *node)
{ {
if (!me->cb.send_snapshot) if (!me->cb.send_snapshot) {
return 0; return 0;
}
while (1) { while (1) {
raft_size_t offset = raft_node_get_snapshot_offset(node); raft_size_t offset = raft_node_get_snapshot_offset(node);
raft_snapshot_req_t msg = { raft_snapshot_req_t req = {
.leader_id = raft_get_nodeid(me), .leader_id = raft_get_nodeid(me),
.snapshot_index = me->snapshot_last_idx, .snapshot_index = me->snapshot_last_idx,
.snapshot_term = me->snapshot_last_term, .snapshot_term = me->snapshot_last_term,
...@@ -1362,14 +1375,21 @@ int raft_send_snapshot(raft_server_t* me, raft_node_t* node) ...@@ -1362,14 +1375,21 @@ int raft_send_snapshot(raft_server_t* me, raft_node_t* node)
.chunk.offset = offset .chunk.offset = offset
}; };
raft_snapshot_chunk_t *chunk = &msg.chunk; raft_snapshot_chunk_t *chunk = &req.chunk;
int e = me->cb.get_snapshot_chunk(me, me->udata, node, offset, chunk); int e = me->cb.get_snapshot_chunk(me, me->udata, node, offset, chunk);
if (e != 0) { if (e != 0) {
return (e != RAFT_ERR_DONE) ? e : 0; return (e != RAFT_ERR_DONE) ? e : 0;
} }
e = me->cb.send_snapshot(me, me->udata, node, &msg); raft_log(me, "%d --> %d, sent snapshot_req "
"t:%ld, li:%d, id:%ld, si:%ld, st:%ld, o:%llu len:%llu lc:%d",
raft_get_nodeid(me), raft_node_get_id(node),
req.term, req.leader_id, req.msg_id, req.snapshot_index,
req.snapshot_term, req.chunk.offset, req.chunk.len,
req.chunk.last_chunk);
e = me->cb.send_snapshot(me, me->udata, node, &req);
if (e != 0) { if (e != 0) {
return e; return e;
} }
...@@ -1384,25 +1404,19 @@ int raft_send_snapshot(raft_server_t* me, raft_node_t* node) ...@@ -1384,25 +1404,19 @@ int raft_send_snapshot(raft_server_t* me, raft_node_t* node)
} }
} }
int raft_recv_snapshot(raft_server_t* me, int raft_recv_snapshot(raft_server_t *me,
raft_node_t* node, raft_node_t *node,
raft_snapshot_req_t *req, raft_snapshot_req_t *req,
raft_snapshot_resp_t *resp) raft_snapshot_resp_t *resp)
{ {
int e = 0; int e;
raft_log_node(me, raft_node_get_id(node), raft_log(me, "%d <-- %d, recv snapshot_req "
"recv snapshot: ci:%lu comi:%lu t:%lu li:%d mi:%lu si:%lu st:%lu o:%llu, lc:%d, len:%llu", "t:%ld, lead:%d, id:%ld, si:%ld, st:%ld, o:%llu len:%llu lc:%d",
raft_get_current_idx(me), raft_get_nodeid(me), raft_node_get_id(node),
me->commit_idx, req->term, req->leader_id, req->msg_id, req->snapshot_index,
req->term, req->snapshot_term, req->chunk.offset, req->chunk.len,
req->leader_id, req->chunk.last_chunk);
req->msg_id,
req->snapshot_index,
req->snapshot_term,
req->chunk.offset,
req->chunk.last_chunk,
req->chunk.len);
resp->msg_id = req->msg_id; resp->msg_id = req->msg_id;
resp->last_chunk = req->chunk.last_chunk; resp->last_chunk = req->chunk.last_chunk;
...@@ -1412,8 +1426,7 @@ int raft_recv_snapshot(raft_server_t* me, ...@@ -1412,8 +1426,7 @@ int raft_recv_snapshot(raft_server_t* me,
e = raft_receive_term(me, req->term); e = raft_receive_term(me, req->term);
if (e != 0) { if (e != 0) {
if (e == RAFT_ERR_STALE_TERM) { if (e == RAFT_ERR_STALE_TERM) {
raft_log_node(me, req->leader_id, raft_log(me, "Snapshot req term:%ld is less than current term:%ld",
"Snapshot req term %ld is less than current term %ld",
req->term, me->current_term); req->term, me->current_term);
e = 0; e = 0;
} }
...@@ -1475,58 +1488,56 @@ out: ...@@ -1475,58 +1488,56 @@ out:
raft_node_update_max_seen_msg_id(node, req->msg_id); raft_node_update_max_seen_msg_id(node, req->msg_id);
} }
raft_log(me, "%d --> %d, sent snapshot_resp "
"id:%lu, t:%ld, s:%d, o:%llu, lc:%d",
raft_get_nodeid(me), raft_node_get_id(node), resp->msg_id,
resp->term, resp->success, resp->offset, resp->last_chunk);
return e; return e;
} }
int raft_recv_snapshot_response(raft_server_t* me, int raft_recv_snapshot_response(raft_server_t *me,
raft_node_t* node, raft_node_t *node,
raft_snapshot_resp_t *r) raft_snapshot_resp_t *resp)
{ {
raft_log_node(me, raft_node_get_id(node), raft_log(me, "%d <-- %d, recv snapshot_resp "
"recv snapshot response: ci:%lu comi:%lu mi:%lu t:%ld o:%llu s:%d lc:%d", "id:%lu, t:%ld, s:%d, o:%llu, lc:%d",
raft_get_current_idx(me), raft_get_nodeid(me), raft_node_get_id(node), resp->msg_id,
me->commit_idx, resp->term, resp->success, resp->offset, resp->last_chunk);
r->msg_id,
r->term,
r->offset,
r->success,
r->last_chunk);
if (!raft_is_leader(me)) if (!raft_is_leader(me)) {
return RAFT_ERR_NOT_LEADER; return RAFT_ERR_NOT_LEADER;
}
if (raft_node_get_match_msgid(node) > r->msg_id) { if (resp->term < me->current_term ||
resp->msg_id < raft_node_get_match_msgid(node) ) {
return 0; return 0;
} }
if (me->current_term < r->term) if (resp->term > me->current_term) {
{ int e = raft_set_current_term(me, resp->term);
int e = raft_set_current_term(me, r->term); if (e != 0) {
if (0 != e)
return e; return e;
}
raft_become_follower(me); raft_become_follower(me);
return 0; return 0;
} }
else if (me->current_term != r->term)
{
return 0;
}
raft_node_set_match_msgid(node, r->msg_id); raft_node_set_match_msgid(node, resp->msg_id);
if (!r->success) { if (!resp->success) {
raft_node_set_snapshot_offset(node, r->offset); raft_node_set_snapshot_offset(node, resp->offset);
} }
if (r->success && r->last_chunk) { if (resp->success && resp->last_chunk) {
raft_node_set_snapshot_offset(node, 0); raft_node_set_snapshot_offset(node, 0);
raft_node_set_next_idx(node, max(me->snapshot_last_idx + 1, raft_node_set_next_idx(node, max(me->snapshot_last_idx + 1,
raft_node_get_next_idx(node))); raft_node_get_next_idx(node)));
} }
if (me->auto_flush) if (me->auto_flush) {
return raft_flush(me, 0); return raft_flush(me, 0);
}
return 0; return 0;
} }
...@@ -1546,7 +1557,7 @@ raft_term_t raft_get_previous_log_term(raft_server_t *me, raft_index_t idx) ...@@ -1546,7 +1557,7 @@ raft_term_t raft_get_previous_log_term(raft_server_t *me, raft_index_t idx)
return term; return term;
} }
int raft_send_appendentries(raft_server_t* me, raft_node_t* node) int raft_send_appendentries(raft_server_t *me, raft_node_t *node)
{ {
assert(node != me->node); assert(node != me->node);
assert(raft_node_get_next_idx(node) != 0); assert(raft_node_get_next_idx(node) != 0);
...@@ -1574,30 +1585,27 @@ int raft_send_appendentries(raft_server_t* me, raft_node_t* node) ...@@ -1574,30 +1585,27 @@ int raft_send_appendentries(raft_server_t* me, raft_node_t* node)
raft_index_t next_idx = raft_node_get_next_idx(node); raft_index_t next_idx = raft_node_get_next_idx(node);
raft_appendentries_req_t ae = { raft_appendentries_req_t req = {
.term = me->current_term, .term = me->current_term,
.leader_id = raft_get_nodeid(me), .leader_id = raft_get_nodeid(me),
.leader_commit = me->commit_idx, .leader_commit = me->commit_idx,
.msg_id = ++me->msg_id, .msg_id = ++me->msg_id,
.prev_log_idx = next_idx - 1, .prev_log_idx = next_idx - 1,
.prev_log_term = raft_get_previous_log_term(me, next_idx), .prev_log_term = raft_get_previous_log_term(me, next_idx),
.entries = raft_get_entries(me, node, next_idx, &ae.n_entries) .entries = raft_get_entries(me, node, next_idx, &req.n_entries)
}; };
raft_log_node(me, raft_node_get_id(node), raft_log(me, "%d --> %d, sent appendentries_req "
"send ae: t:%lu lc:%lu pli:%lu plt:%lu msgid:%lu #%ld", "lead:%d, id:%ld, t:%ld, pli:%ld, plt:%ld, lc:%ld ent:%ld",
ae.term, raft_get_nodeid(me), raft_node_get_id(node),
ae.leader_commit, req.leader_id, req.msg_id, req.term, req.prev_log_idx,
ae.prev_log_idx, req.prev_log_term, req.leader_commit, req.n_entries);
ae.prev_log_term,
ae.msg_id,
ae.n_entries);
e = me->cb.send_appendentries(me, me->udata, node, &ae); e = me->cb.send_appendentries(me, me->udata, node, &req);
raft_node_set_next_idx(node, next_idx + ae.n_entries); raft_node_set_next_idx(node, next_idx + req.n_entries);
raft_node_set_next_msgid(node, ae.msg_id + 1); raft_node_set_next_msgid(node, req.msg_id + 1);
raft_entry_release_list(ae.entries, ae.n_entries); raft_entry_release_list(req.entries, req.n_entries);
if (e != 0) { if (e != 0) {
return e; return e;
...@@ -1817,10 +1825,8 @@ int raft_end_snapshot(raft_server_t *me) ...@@ -1817,10 +1825,8 @@ int raft_end_snapshot(raft_server_t *me)
me->snapshot_in_progress = 0; me->snapshot_in_progress = 0;
raft_log(me, raft_log(me,"end snapshot, base:%ld, commit-index:%ld, current-index:%ld",
"end snapshot base:%ld commit-index:%ld current-index:%ld", me->log_impl->first_idx(me->log) - 1, me->commit_idx,
me->log_impl->first_idx(me->log) - 1,
me->commit_idx,
raft_get_current_idx(me)); raft_get_current_idx(me));
if (!raft_is_leader(me)) if (!raft_is_leader(me))
...@@ -2081,10 +2087,12 @@ int raft_transfer_leader(raft_server_t* me, raft_node_id_t node_id, long timeout ...@@ -2081,10 +2087,12 @@ int raft_transfer_leader(raft_server_t* me, raft_node_id_t node_id, long timeout
return RAFT_ERR_INVALID_NODEID; return RAFT_ERR_INVALID_NODEID;
} }
if (me->cb.send_timeoutnow && if (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, me->udata, target); me->cb.send_timeoutnow(me, me->udata, target);
me->sent_timeout_now = 1; me->sent_timeout_now = 1;
raft_log(me, "%d --> %d, sent timeout_now", raft_get_nodeid(me),
raft_node_get_id(target));
} }
me->node_transferring_leader_to = raft_node_get_id(target); me->node_transferring_leader_to = raft_node_get_id(target);
......
...@@ -4064,7 +4064,7 @@ void TestRaft_quorum_msg_id_correctness(CuTest * tc) ...@@ -4064,7 +4064,7 @@ void TestRaft_quorum_msg_id_correctness(CuTest * tc)
int timeoutnow_sent = 0; int timeoutnow_sent = 0;
int __fake_timeoutnow(raft_server_t* raft, void *udata, raft_node_t* node) int cb_timeoutnow(raft_server_t* raft, void *udata, raft_node_t* node)
{ {
timeoutnow_sent = 1; timeoutnow_sent = 1;
...@@ -4076,7 +4076,7 @@ void TestRaft_callback_timeoutnow_at_set_if_up_to_date(CuTest *tc) ...@@ -4076,7 +4076,7 @@ void TestRaft_callback_timeoutnow_at_set_if_up_to_date(CuTest *tc)
timeoutnow_sent = 0; timeoutnow_sent = 0;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.send_timeoutnow = __fake_timeoutnow, .send_timeoutnow = cb_timeoutnow,
}; };
raft_server_t *r = raft_new(); raft_server_t *r = raft_new();
...@@ -4111,7 +4111,7 @@ void TestRaft_callback_timeoutnow_at_send_appendentries_response_if_up_to_date(C ...@@ -4111,7 +4111,7 @@ void TestRaft_callback_timeoutnow_at_send_appendentries_response_if_up_to_date(C
timeoutnow_sent = 0; timeoutnow_sent = 0;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.send_timeoutnow = __fake_timeoutnow, .send_timeoutnow = cb_timeoutnow,
}; };
raft_server_t *r = raft_new(); raft_server_t *r = raft_new();
...@@ -4439,6 +4439,7 @@ void Test_reset_transfer_leader(CuTest *tc) ...@@ -4439,6 +4439,7 @@ void Test_reset_transfer_leader(CuTest *tc)
raft_leader_transfer_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,
.send_timeoutnow = cb_timeoutnow,
}; };
raft_server_t *r = raft_new(); raft_server_t *r = raft_new();
raft_set_state(r, RAFT_STATE_LEADER); raft_set_state(r, RAFT_STATE_LEADER);
...@@ -4471,6 +4472,7 @@ void Test_transfer_leader_success(CuTest *tc) ...@@ -4471,6 +4472,7 @@ void Test_transfer_leader_success(CuTest *tc)
raft_leader_transfer_e state = RAFT_LEADER_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,
.send_timeoutnow = cb_timeoutnow
}; };
raft_server_t *r = raft_new(); raft_server_t *r = raft_new();
raft_set_callbacks(r, &funcs, &state); raft_set_callbacks(r, &funcs, &state);
...@@ -4496,6 +4498,7 @@ void Test_transfer_leader_unexpected(CuTest *tc) ...@@ -4496,6 +4498,7 @@ void Test_transfer_leader_unexpected(CuTest *tc)
raft_leader_transfer_e state = RAFT_LEADER_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,
.send_timeoutnow = cb_timeoutnow
}; };
raft_server_t *r = raft_new(); raft_server_t *r = raft_new();
raft_set_callbacks(r, &funcs, &state); raft_set_callbacks(r, &funcs, &state);
......
...@@ -331,13 +331,12 @@ def handle_read_queue(arg, can_read): ...@@ -331,13 +331,12 @@ def handle_read_queue(arg, can_read):
logger.debug(f"ignoring read_request {val}") logger.debug(f"ignoring read_request {val}")
def raft_log(raft, node, udata, buf): def raft_log(raft, udata, buf):
server = ffi.from_handle(lib.raft_get_udata(raft)) server = ffi.from_handle(lib.raft_get_udata(raft))
# if server.id in [1] or (node and node.id in [1]): # if server.id in [1] or (node and node.id in [1]):
logger.info('{0}> {1}:{2}: {3}'.format( logger.info('{0}> {1}: {2}'.format(
server.network.iteration, server.network.iteration,
server.id, server.id,
node,
ffi.string(buf).decode('utf8'), ffi.string(buf).decode('utf8'),
)) ))
...@@ -956,7 +955,7 @@ class RaftServer(object): ...@@ -956,7 +955,7 @@ class RaftServer(object):
self.raft_get_node_id = ffi.callback("int(raft_server_t*, void*, raft_entry_t*, raft_index_t)", raft_get_node_id) self.raft_get_node_id = ffi.callback("int(raft_server_t*, void*, raft_entry_t*, raft_index_t)", raft_get_node_id)
self.raft_node_has_sufficient_logs = ffi.callback("int(raft_server_t* raft, void *user_data, raft_node_t* node)", raft_node_has_sufficient_logs) self.raft_node_has_sufficient_logs = ffi.callback("int(raft_server_t* raft, void *user_data, raft_node_t* node)", raft_node_has_sufficient_logs)
self.raft_notify_membership_event = ffi.callback("void(raft_server_t* raft, void *user_data, raft_node_t* node, raft_entry_t* ety, raft_membership_e)", raft_notify_membership_event) self.raft_notify_membership_event = ffi.callback("void(raft_server_t* raft, void *user_data, raft_node_t* node, raft_entry_t* ety, raft_membership_e)", raft_notify_membership_event)
self.raft_log = ffi.callback("void(raft_server_t*, raft_node_id_t, void*, const char* buf)", raft_log) self.raft_log = ffi.callback("void(raft_server_t*, void*, const char* buf)", raft_log)
self.handle_read_queue = ffi.callback("void(void *arg, int can_read)", handle_read_queue) self.handle_read_queue = ffi.callback("void(void *arg, int can_read)", handle_read_queue)
def recv_entry(self, ety): def recv_entry(self, ety):
......
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