Commit 65353d06 authored by Willem Thiart's avatar Willem Thiart
Browse files

Add node_has_sufficient_logs callback

parent c0616495
...@@ -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.
...@@ -277,6 +290,9 @@ typedef struct ...@@ -277,6 +290,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;
......
...@@ -90,6 +90,7 @@ int raft_apply_entry(raft_server_t* me_); ...@@ -90,6 +90,7 @@ int raft_apply_entry(raft_server_t* me_);
int raft_append_entry(raft_server_t* me_, raft_entry_t* c); int raft_append_entry(raft_server_t* me_, raft_entry_t* c);
void raft_set_commit_idx(raft_server_t* me, int commit_idx); void raft_set_commit_idx(raft_server_t* me, int commit_idx);
int raft_get_commit_idx(raft_server_t* me_); int raft_get_commit_idx(raft_server_t* me_);
void raft_set_last_applied_idx(raft_server_t* me, int idx); void raft_set_last_applied_idx(raft_server_t* me, int idx);
...@@ -110,6 +111,10 @@ void raft_node_vote_for_me(raft_node_t* me_, const int vote); ...@@ -110,6 +111,10 @@ void raft_node_vote_for_me(raft_node_t* me_, const int vote);
int raft_node_has_vote_for_me(raft_node_t* me_); int raft_node_has_vote_for_me(raft_node_t* me_);
void raft_node_set_has_sufficient_logs(raft_node_t* me_);
int raft_node_has_sufficient_logs(raft_node_t* me_);
int raft_votes_is_majority(const int nnodes, const int nvotes); int raft_votes_is_majority(const int nnodes, const int nvotes);
#endif /* RAFT_PRIVATE_H_ */ #endif /* RAFT_PRIVATE_H_ */
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#define RAFT_NODE_VOTED_FOR_ME 1 #define RAFT_NODE_VOTED_FOR_ME 1
#define RAFT_NODE_VOTING 1 << 1 #define RAFT_NODE_VOTING 1 << 1
#define RAFT_NODE_HAS_SUFFICIENT_LOG 1 << 2
typedef struct typedef struct
{ {
...@@ -111,6 +112,18 @@ int raft_node_is_voting(raft_node_t* me_) ...@@ -111,6 +112,18 @@ int raft_node_is_voting(raft_node_t* me_)
return (me->flags & RAFT_NODE_VOTING) != 0; return (me->flags & RAFT_NODE_VOTING) != 0;
} }
void raft_node_set_has_sufficient_logs(raft_node_t* me_)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
me->flags |= RAFT_NODE_HAS_SUFFICIENT_LOG;
}
int raft_node_has_sufficient_logs(raft_node_t* me_)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return (me->flags & RAFT_NODE_HAS_SUFFICIENT_LOG) != 0;
}
int raft_node_get_id(raft_node_t* me_) int raft_node_get_id(raft_node_t* me_)
{ {
raft_node_private_t* me = (raft_node_private_t*)me_; raft_node_private_t* me = (raft_node_private_t*)me_;
......
...@@ -220,6 +220,16 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -220,6 +220,16 @@ 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_idx(me_) <= r->current_idx + 1 &&
me->cb.node_has_sufficient_logs &&
0 == raft_node_has_sufficient_logs(node)
)
{
raft_node_set_has_sufficient_logs(node);
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;
......
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