Commit ec4f4305 authored by Willem Thiart's avatar Willem Thiart
Browse files

raft_recv_requestvote can now return RAFT_ERR_SHUTDOWN

parent 95a5781f
...@@ -14,6 +14,10 @@ ...@@ -14,6 +14,10 @@
#define RAFT_ERR_ONE_VOTING_CHANGE_ONLY -3 #define RAFT_ERR_ONE_VOTING_CHANGE_ONLY -3
#define RAFT_ERR_SHUTDOWN -4 #define RAFT_ERR_SHUTDOWN -4
#define RAFT_REQUESTVOTE_ERR_GRANTED 1
#define RAFT_REQUESTVOTE_ERR_NOT_GRANTED 0
#define RAFT_REQUESTVOTE_ERR_UNKNOWN_NODE -1
typedef enum { typedef enum {
RAFT_STATE_NONE, RAFT_STATE_NONE,
RAFT_STATE_FOLLOWER, RAFT_STATE_FOLLOWER,
...@@ -296,6 +300,11 @@ typedef struct ...@@ -296,6 +300,11 @@ 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 determining which node this configuration log entry
* affects. This call only applies to configuration change log entries.
* @return the node ID of the node */
func_logentry_event_f log_get_node_id;
/** Callback for detecting when a non-voting node has sufficient logs. */ /** Callback for detecting when a non-voting node has sufficient logs. */
func_node_has_sufficient_logs_f node_has_sufficient_logs; func_node_has_sufficient_logs_f node_has_sufficient_logs;
...@@ -433,7 +442,9 @@ int raft_recv_requestvote(raft_server_t* me, ...@@ -433,7 +442,9 @@ int raft_recv_requestvote(raft_server_t* me,
/** 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] r The requestvote response message * @param[in] r The requestvote response message
* @return 0 on success */ * @return
* 0 on success;
* RAFT_ERR_SHUTDOWN server should be 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,
msg_requestvote_response_t* r); msg_requestvote_response_t* r);
......
#ifndef RAFT_PRIVATE_H_ #ifndef RAFT_PRIVATE_H_
#define RAFT_PRIVATE_H_ #define RAFT_PRIVATE_H_
enum {
RAFT_NODE_STATUS_DISCONNECTED,
RAFT_NODE_STATUS_CONNECTED,
RAFT_NODE_STATUS_CONNECTING,
RAFT_NODE_STATUS_DISCONNECTING
};
/** /**
* Copyright (c) 2013, Willem-Hendrik Thiart * Copyright (c) 2013, Willem-Hendrik Thiart
* Use of this source code is governed by a BSD-style license that can be * Use of this source code is governed by a BSD-style license that can be
...@@ -57,6 +64,10 @@ typedef struct { ...@@ -57,6 +64,10 @@ typedef struct {
/* the log which has a voting cfg change, otherwise -1 */ /* the log which has a voting cfg change, otherwise -1 */
int voting_cfg_change_log_idx; int voting_cfg_change_log_idx;
/* our membership with the cluster is confirmed (ie. configuration log was
* committed) */
int connected;
} raft_server_private_t; } raft_server_private_t;
void raft_election_start(raft_server_t* me); void raft_election_start(raft_server_t* me);
......
...@@ -516,9 +516,23 @@ int raft_recv_requestvote(raft_server_t* me_, ...@@ -516,9 +516,23 @@ int raft_recv_requestvote(raft_server_t* me_,
me->timeout_elapsed = 0; me->timeout_elapsed = 0;
} }
else
{
/* It's possible the candidate node has been removed from the cluster but
* hasn't received the appendentries that confirms the removal. Therefore
* the node is partitioned and still thinks its part of the cluster. It
* will eventually send a requestvote. This is error response tells the
* node that it might be removed. */
if (!node)
{
r->vote_granted = RAFT_REQUESTVOTE_ERR_UNKNOWN_NODE;
goto done;
}
else else
r->vote_granted = 0; r->vote_granted = 0;
}
done:
__log(me_, node, "node requested vote: %d replying: %s", __log(me_, node, "node requested vote: %d replying: %s",
node, node,
r->vote_granted == 1 ? "granted" : r->vote_granted == 1 ? "granted" :
...@@ -570,13 +584,27 @@ int raft_recv_requestvote_response(raft_server_t* me_, ...@@ -570,13 +584,27 @@ int raft_recv_requestvote_response(raft_server_t* me_,
me->current_term, me->current_term,
r->term); r->term);
if (1 == r->vote_granted) switch (r->vote_granted)
{ {
case RAFT_REQUESTVOTE_ERR_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_);
if (raft_votes_is_majority(me->num_nodes, votes)) if (raft_votes_is_majority(raft_get_num_voting_nodes(me_), votes))
raft_become_leader(me_); raft_become_leader(me_);
break;
case RAFT_REQUESTVOTE_ERR_NOT_GRANTED:
break;
case RAFT_REQUESTVOTE_ERR_UNKNOWN_NODE:
if (raft_node_is_voting(raft_get_my_node(me_)) &&
me->connected == RAFT_NODE_STATUS_DISCONNECTING)
return RAFT_ERR_SHUTDOWN;
break;
default:
assert(0);
} }
return 0; return 0;
...@@ -688,6 +716,15 @@ int raft_apply_entry(raft_server_t* me_) ...@@ -688,6 +716,15 @@ int raft_apply_entry(raft_server_t* me_)
return RAFT_ERR_SHUTDOWN; return RAFT_ERR_SHUTDOWN;
} }
/* Membership Change: confirm connection with cluster */
if (RAFT_LOGTYPE_ADD_NODE == ety->type)
{
int node_id = me->cb.log_get_node_id(me_, raft_get_udata(me_), ety, log_idx);
raft_node_set_has_sufficient_logs(raft_get_node(me_, node_id));
if (node_id == raft_get_nodeid(me_))
me->connected = RAFT_NODE_STATUS_CONNECTED;
}
/* voting cfg change is now complete */ /* voting cfg change is now complete */
if (log_idx == me->voting_cfg_change_log_idx) if (log_idx == me->voting_cfg_change_log_idx)
me->voting_cfg_change_log_idx = -1; me->voting_cfg_change_log_idx = -1;
......
...@@ -217,3 +217,8 @@ int raft_get_last_log_term(raft_server_t* me_) ...@@ -217,3 +217,8 @@ int raft_get_last_log_term(raft_server_t* me_)
} }
return 0; return 0;
} }
int raft_is_connected(raft_server_t* me_)
{
return ((raft_server_private_t*)me_)->connected;
}
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