Commit 387fb3e3 authored by Willem Thiart's avatar Willem Thiart
Browse files

Add node_has_sufficient_logs callback

parent 1410fc5e
...@@ -168,6 +168,19 @@ typedef int ( ...@@ -168,6 +168,19 @@ typedef int (
msg_appendentries_t* msg msg_appendentries_t* msg
); );
/** Callback for detecting when non-voting nodes have obtained enough logs.
* @param[in] raft The Raft server making this callback
* @param[in] user_data User data that is passed from Raft server
* @param[in] node The node
* @return 0 on success */
typedef void (
*func_node_has_sufficient_logs_f
) (
raft_server_t* raft,
void *user_data,
raft_node_t* node
);
#ifndef HAVE_FUNC_LOG #ifndef HAVE_FUNC_LOG
#define HAVE_FUNC_LOG #define HAVE_FUNC_LOG
/** Callback for providing debug logging information. /** Callback for providing debug logging information.
...@@ -293,6 +306,9 @@ typedef struct ...@@ -293,6 +306,9 @@ typedef struct
* time to free the memory. */ * time to free the memory. */
func_logentry_event_f log_pop; func_logentry_event_f log_pop;
/** Callback for detecting when a non-voting node has sufficient logs. */
func_node_has_sufficient_logs_f node_has_sufficient_logs;
/** Callback for catching debugging log messages /** Callback for catching debugging log messages
* This callback is optional */ * This callback is optional */
func_log_f log; func_log_f log;
......
...@@ -110,7 +110,7 @@ void raft_become_candidate(raft_server_t* me_) ...@@ -110,7 +110,7 @@ void raft_become_candidate(raft_server_t* me_)
__log(me_, NULL, "becoming candidate"); __log(me_, NULL, "becoming candidate");
raft_set_current_term(me_, raft_get_current_term(me_) + 1); raft_set_current_term(me_, raft_get_current_term(me_) + 1);
for (i = 0; i < me->num_nodes; i++) for (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_vote(me_, me->node); raft_vote(me_, me->node);
...@@ -221,6 +221,11 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -221,6 +221,11 @@ int raft_recv_appendentries_response(raft_server_t* me_,
raft_node_set_next_idx(node, r->current_idx + 1); raft_node_set_next_idx(node, r->current_idx + 1);
raft_node_set_match_idx(node, r->current_idx); raft_node_set_match_idx(node, r->current_idx);
if (!raft_node_is_voting(node) &&
raft_get_current_term(me_) <= r->current_idx + 1 &&
me->cb.node_has_sufficient_logs)
me->cb.node_has_sufficient_logs(me_, me->udata, node);
/* Update commit idx */ /* Update commit idx */
int votes = 1; /* include me */ int votes = 1; /* include me */
int point = r->current_idx; int point = r->current_idx;
...@@ -291,7 +296,7 @@ int raft_recv_appendentries( ...@@ -291,7 +296,7 @@ int raft_recv_appendentries(
{ {
/* 1. Reply false if term < currentTerm (§5.1) */ /* 1. Reply false if term < currentTerm (§5.1) */
__log(me_, node, "AE term %d is less than current term %d", __log(me_, node, "AE term %d is less than current term %d",
ae->term, me->current_term); ae->term, me->current_term);
goto fail_with_current_idx; goto fail_with_current_idx;
} }
...@@ -356,7 +361,7 @@ int raft_recv_appendentries( ...@@ -356,7 +361,7 @@ int raft_recv_appendentries(
for (; i < ae->n_entries; i++) for (; i < ae->n_entries; i++)
{ {
__log(me_, node, "appending ci:%d id:%d", __log(me_, node, "appending ci:%d id:%d",
raft_get_current_idx(me_), ae->entries[i].id); raft_get_current_idx(me_), ae->entries[i].id);
int e = raft_append_entry(me_, &ae->entries[i]); int e = raft_append_entry(me_, &ae->entries[i]);
if (-1 == e) if (-1 == e)
goto fail_with_current_idx; goto fail_with_current_idx;
...@@ -411,7 +416,7 @@ static int __should_grant_vote(raft_server_private_t* me, msg_requestvote_t* vr) ...@@ -411,7 +416,7 @@ static int __should_grant_vote(raft_server_private_t* me, msg_requestvote_t* vr)
return 0; return 0;
} }
int raft_recv_requestvote(raft_server_t* me_, int raft_recv_requestvote(raft_server_t* me_,
raft_node_t* node, raft_node_t* node,
msg_requestvote_t* vr, msg_requestvote_t* vr,
...@@ -435,7 +440,7 @@ int raft_recv_requestvote(raft_server_t* me_, ...@@ -435,7 +440,7 @@ int raft_recv_requestvote(raft_server_t* me_,
{ {
raft_entry_t* e = raft_get_entry_from_idx((void*)me, current_idx); raft_entry_t* e = raft_get_entry_from_idx((void*)me, current_idx);
__log(me_, node, "granting2: ct:%d et:%d llt:%d lli:%d ci:%d", __log(me_, node, "granting2: ct:%d et:%d llt:%d lli:%d ci:%d",
me->current_term, e->term, vr->last_log_term, vr->last_log_idx, current_idx); me->current_term, e->term, vr->last_log_term, vr->last_log_idx, current_idx);
} }
/* 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
...@@ -523,7 +528,7 @@ int raft_recv_entry(raft_server_t* me_, raft_node_t* node, msg_entry_t* e, ...@@ -523,7 +528,7 @@ int raft_recv_entry(raft_server_t* me_, raft_node_t* node, msg_entry_t* e,
return -1; return -1;
__log(me_, node, "received entry t:%d id: %d idx: %d", __log(me_, node, "received entry t:%d id: %d idx: %d",
me->current_term, e->id, raft_get_current_idx(me_) + 1, node); me->current_term, e->id, raft_get_current_idx(me_) + 1, node);
raft_entry_t ety; raft_entry_t ety;
ety.term = me->current_term; ety.term = me->current_term;
...@@ -591,7 +596,7 @@ int raft_apply_entry(raft_server_t* me_) ...@@ -591,7 +596,7 @@ int raft_apply_entry(raft_server_t* me_)
return -1; return -1;
__log(me_, NULL, "applying log: %d, id: %d size: %d", __log(me_, NULL, "applying log: %d, id: %d size: %d",
me->last_applied_idx, e->id, e->data.len); me->last_applied_idx, e->id, e->data.len);
me->last_applied_idx++; me->last_applied_idx++;
if (me->cb.applylog) if (me->cb.applylog)
...@@ -643,11 +648,11 @@ int raft_send_appendentries(raft_server_t* me_, raft_node_t* node) ...@@ -643,11 +648,11 @@ int raft_send_appendentries(raft_server_t* me_, raft_node_t* node)
} }
__log(me_, node, "sending appendentries node: ci:%d t:%d lc:%d pli:%d plt:%d", __log(me_, node, "sending appendentries node: ci:%d t:%d lc:%d pli:%d plt:%d",
raft_get_current_idx(me_), raft_get_current_idx(me_),
ae.term, ae.term,
ae.leader_commit, ae.leader_commit,
ae.prev_log_idx, ae.prev_log_idx,
ae.prev_log_term); ae.prev_log_term);
me->cb.send_appendentries(me_, me->udata, node, &ae); me->cb.send_appendentries(me_, me->udata, node, &ae);
......
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