Unverified Commit e1569508 authored by Hanna Fadida's avatar Hanna Fadida Committed by GitHub
Browse files

clean up add node flows (#147)

parent 4bb63026
...@@ -957,7 +957,7 @@ void raft_set_callbacks(raft_server_t* me, raft_cbs_t* funcs, void* user_data); ...@@ -957,7 +957,7 @@ void raft_set_callbacks(raft_server_t* me, raft_cbs_t* funcs, void* user_data);
* @param[in] is_self Set to 1 if this "node" is this server * @param[in] is_self Set to 1 if this "node" is this server
* @return * @return
* node if it was successfully added; * node if it was successfully added;
* NULL if the node already exists */ * NULL if a voting node already exists */
raft_node_t* raft_add_node(raft_server_t* me, void* user_data, raft_node_id_t id, int is_self); raft_node_t* raft_add_node(raft_server_t* me, void* user_data, raft_node_id_t id, int is_self);
/** Add a node which does not participate in voting. /** Add a node which does not participate in voting.
......
...@@ -85,7 +85,7 @@ struct raft_server { ...@@ -85,7 +85,7 @@ struct raft_server {
/* my node ID */ /* my node ID */
raft_node_t* node; raft_node_t* node;
/* the log which has a voting cfg change, otherwise -1 */ /* the log index which has a voting cfg change, otherwise -1 */
raft_index_t voting_cfg_change_log_idx; raft_index_t voting_cfg_change_log_idx;
int snapshot_in_progress; int snapshot_in_progress;
...@@ -160,7 +160,7 @@ void raft_set_last_applied_idx(raft_server_t *me, raft_index_t idx); ...@@ -160,7 +160,7 @@ void raft_set_last_applied_idx(raft_server_t *me, raft_index_t idx);
void raft_set_state(raft_server_t *me, int state); void raft_set_state(raft_server_t *me, int state);
raft_node_t *raft_node_new(void *udata, raft_node_id_t id); raft_node_t *raft_node_new(void *udata, raft_node_id_t id, int voting);
void raft_node_free(raft_node_t *node); void raft_node_free(raft_node_t *node);
...@@ -172,7 +172,7 @@ raft_index_t raft_node_get_next_idx(raft_node_t *node); ...@@ -172,7 +172,7 @@ raft_index_t raft_node_get_next_idx(raft_node_t *node);
void raft_node_clear_flags(raft_node_t *node); void raft_node_clear_flags(raft_node_t *node);
void raft_node_vote_for_me(raft_node_t *node, int vote); void raft_node_set_voted_for_me(raft_node_t *node, int vote);
int raft_node_has_vote_for_me(raft_node_t *node); int raft_node_has_vote_for_me(raft_node_t *node);
......
...@@ -37,7 +37,7 @@ struct raft_node { ...@@ -37,7 +37,7 @@ struct raft_node {
raft_size_t snapshot_offset; raft_size_t snapshot_offset;
}; };
raft_node_t *raft_node_new(void *udata, raft_node_id_t id) raft_node_t *raft_node_new(void *udata, raft_node_id_t id, int voting)
{ {
raft_node_t *me; raft_node_t *me;
...@@ -50,7 +50,7 @@ raft_node_t *raft_node_new(void *udata, raft_node_id_t id) ...@@ -50,7 +50,7 @@ raft_node_t *raft_node_new(void *udata, raft_node_id_t id)
me->next_idx = 1; me->next_idx = 1;
me->match_idx = 0; me->match_idx = 0;
me->id = id; me->id = id;
me->flags = RAFT_NODE_VOTING; raft_node_set_voting(me, voting);
return me; return me;
} }
...@@ -128,7 +128,7 @@ void raft_node_clear_flags(raft_node_t *node) ...@@ -128,7 +128,7 @@ void raft_node_clear_flags(raft_node_t *node)
node->flags = 0; node->flags = 0;
} }
void raft_node_vote_for_me(raft_node_t *node, const int vote) void raft_node_set_voted_for_me(raft_node_t *node, int vote)
{ {
if (vote) { if (vote) {
node->flags |= RAFT_NODE_VOTED_FOR_ME; node->flags |= RAFT_NODE_VOTED_FOR_ME;
......
...@@ -110,6 +110,7 @@ raft_server_t* raft_new_with_log(const raft_log_impl_t *log_impl, void *log_arg) ...@@ -110,6 +110,7 @@ raft_server_t* raft_new_with_log(const raft_log_impl_t *log_impl, void *log_arg)
me->node_transferring_leader_to = RAFT_NODE_ID_NONE; me->node_transferring_leader_to = RAFT_NODE_ID_NONE;
me->auto_flush = 1; me->auto_flush = 1;
me->exec_deadline = LLONG_MAX; me->exec_deadline = LLONG_MAX;
me->msg_id = 0;
raft_update_quorum_meta(me, me->msg_id); raft_update_quorum_meta(me, me->msg_id);
...@@ -174,7 +175,7 @@ void raft_clear(raft_server_t* me) ...@@ -174,7 +175,7 @@ void raft_clear(raft_server_t* me)
me->log_impl->reset(me->log, 1, 1); me->log_impl->reset(me->log, 1, 1);
} }
raft_node_t* raft_add_node_internal(raft_server_t* me, raft_entry_t *ety, void* udata, raft_node_id_t id, int is_self) raft_node_t* raft_add_node_internal(raft_server_t *me, raft_entry_t *ety, void *udata, raft_node_id_t id, int is_self, int voting)
{ {
if (id == RAFT_NODE_ID_NONE) if (id == RAFT_NODE_ID_NONE)
return NULL; return NULL;
...@@ -183,17 +184,18 @@ raft_node_t* raft_add_node_internal(raft_server_t* me, raft_entry_t *ety, void* ...@@ -183,17 +184,18 @@ raft_node_t* raft_add_node_internal(raft_server_t* me, raft_entry_t *ety, void*
raft_node_t* node = raft_get_node(me, id); raft_node_t* node = raft_get_node(me, id);
if (node) if (node)
{ {
if (!raft_node_is_voting(node)) /* node can be promoted only from a non voting state into a voting one */
if (!raft_node_is_voting(node) && voting)
{ {
raft_node_set_voting(node, 1); raft_node_set_voting(node, 1);
return node; return node;
} }
else else
/* we shouldn't add a node twice */ /* we shouldn't add a node with the same voting mode twice */
return NULL; return NULL;
} }
node = raft_node_new(udata, id); node = raft_node_new(udata, id, voting);
if (!node) if (!node)
return NULL; return NULL;
...@@ -208,8 +210,6 @@ raft_node_t* raft_add_node_internal(raft_server_t* me, raft_entry_t *ety, void* ...@@ -208,8 +210,6 @@ raft_node_t* raft_add_node_internal(raft_server_t* me, raft_entry_t *ety, void*
if (is_self) if (is_self)
me->node = me->nodes[me->num_nodes - 1]; me->node = me->nodes[me->num_nodes - 1];
node = me->nodes[me->num_nodes - 1];
if (me->cb.notify_membership_event) { if (me->cb.notify_membership_event) {
me->cb.notify_membership_event(me, me->udata, node, ety, me->cb.notify_membership_event(me, me->udata, node, ety,
RAFT_MEMBERSHIP_ADD); RAFT_MEMBERSHIP_ADD);
...@@ -218,27 +218,14 @@ raft_node_t* raft_add_node_internal(raft_server_t* me, raft_entry_t *ety, void* ...@@ -218,27 +218,14 @@ raft_node_t* raft_add_node_internal(raft_server_t* me, raft_entry_t *ety, void*
return node; return node;
} }
static raft_node_t* raft_add_non_voting_node_internal(raft_server_t* me, raft_entry_t *ety, void* udata, raft_node_id_t id, int is_self)
{
if (raft_get_node(me, id))
return NULL;
raft_node_t* node = raft_add_node_internal(me, ety, udata, id, is_self);
if (!node)
return NULL;
raft_node_set_voting(node, 0);
return node;
}
raft_node_t* raft_add_node(raft_server_t* me, void* udata, raft_node_id_t id, int is_self) raft_node_t* raft_add_node(raft_server_t* me, void* udata, raft_node_id_t id, int is_self)
{ {
return raft_add_node_internal(me, NULL, udata, id, is_self); return raft_add_node_internal(me, NULL, udata, id, is_self, 1);
} }
raft_node_t* raft_add_non_voting_node(raft_server_t* me, void* udata, raft_node_id_t id, int is_self) raft_node_t* raft_add_non_voting_node(raft_server_t* me, void* udata, raft_node_id_t id, int is_self)
{ {
return raft_add_non_voting_node_internal(me, NULL, udata, id, is_self); return raft_add_node_internal(me, NULL, udata, id, is_self, 0);
} }
void raft_remove_node(raft_server_t* me, raft_node_t* node) void raft_remove_node(raft_server_t* me, raft_node_t* node)
...@@ -287,27 +274,29 @@ void raft_handle_append_cfg_change(raft_server_t* me, raft_entry_t* ety, raft_in ...@@ -287,27 +274,29 @@ void raft_handle_append_cfg_change(raft_server_t* me, raft_entry_t* ety, raft_in
switch (ety->type) switch (ety->type)
{ {
case RAFT_LOGTYPE_ADD_NODE:
node = raft_add_node_internal(me, ety, NULL, node_id, is_self, 1);
assert(node);
assert(raft_node_is_voting(node));
break;
case RAFT_LOGTYPE_ADD_NONVOTING_NODE: case RAFT_LOGTYPE_ADD_NONVOTING_NODE:
if (!is_self) if (!is_self)
{ {
assert(!node || !raft_node_is_voting(node));
if (node && !raft_node_is_active(node)) if (node && !raft_node_is_active(node))
{ {
raft_node_set_active(node, 1); raft_node_set_active(node, 1);
} }
else if (!node) else if (!node)
{ {
node = raft_add_non_voting_node_internal(me, ety, NULL, node_id, is_self); node = raft_add_node_internal(me, ety, NULL, node_id, is_self, 0);
assert(node); assert(node);
assert(!raft_node_is_voting(node));
} }
} }
break; break;
case RAFT_LOGTYPE_ADD_NODE:
node = raft_add_node_internal(me, ety, NULL, node_id, is_self);
assert(node);
assert(raft_node_is_voting(node));
break;
case RAFT_LOGTYPE_REMOVE_NODE: case RAFT_LOGTYPE_REMOVE_NODE:
if (node) { if (node) {
raft_node_set_active(node, 0); raft_node_set_active(node, 0);
...@@ -332,8 +321,8 @@ void raft_handle_remove_cfg_change(raft_server_t* me, raft_entry_t* ety, const r ...@@ -332,8 +321,8 @@ void raft_handle_remove_cfg_change(raft_server_t* me, raft_entry_t* ety, const r
switch (ety->type) switch (ety->type)
{ {
case RAFT_LOGTYPE_REMOVE_NODE: case RAFT_LOGTYPE_ADD_NODE:
raft_node_set_active(node, 1); raft_node_set_voting(node, 0);
break; break;
case RAFT_LOGTYPE_ADD_NONVOTING_NODE: case RAFT_LOGTYPE_ADD_NONVOTING_NODE:
...@@ -341,8 +330,10 @@ void raft_handle_remove_cfg_change(raft_server_t* me, raft_entry_t* ety, const r ...@@ -341,8 +330,10 @@ void raft_handle_remove_cfg_change(raft_server_t* me, raft_entry_t* ety, const r
raft_remove_node(me, node); raft_remove_node(me, node);
break; break;
case RAFT_LOGTYPE_ADD_NODE: case RAFT_LOGTYPE_REMOVE_NODE:
raft_node_set_voting(node, 0); if (node) {
raft_node_set_active(node, 1);
}
break; break;
default: default:
...@@ -351,6 +342,36 @@ void raft_handle_remove_cfg_change(raft_server_t* me, raft_entry_t* ety, const r ...@@ -351,6 +342,36 @@ void raft_handle_remove_cfg_change(raft_server_t* me, raft_entry_t* ety, const r
} }
} }
void raft_handle_apply_cfg_change(raft_server_t* me, raft_entry_t* ety, raft_index_t idx)
{
if (!raft_entry_is_cfg_change(ety))
return;
if (!me->cb.get_node_id)
return;
raft_node_id_t node_id = me->cb.get_node_id(me, me->udata, ety, idx);
raft_node_t* node = raft_get_node(me, node_id);
switch (ety->type) {
case RAFT_LOGTYPE_ADD_NODE:
raft_node_set_addition_committed(node, 1);
raft_node_set_voting_committed(node, 1);
raft_node_set_has_sufficient_logs(node, 1);
break;
case RAFT_LOGTYPE_ADD_NONVOTING_NODE:
raft_node_set_addition_committed(node, 1);
break;
case RAFT_LOGTYPE_REMOVE_NODE:
if (node) {
raft_remove_node(me, node);
}
break;
default:
break;
}
}
int raft_delete_entry_from_idx(raft_server_t *me, raft_index_t idx) int raft_delete_entry_from_idx(raft_server_t *me, raft_index_t idx)
{ {
assert(me->commit_idx < idx); assert(me->commit_idx < idx);
...@@ -475,7 +496,7 @@ int raft_become_leader(raft_server_t *me) ...@@ -475,7 +496,7 @@ int raft_become_leader(raft_server_t *me)
int raft_become_precandidate(raft_server_t *me) int raft_become_precandidate(raft_server_t *me)
{ {
for (int 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_set_voted_for_me(me->nodes[i], 0);
} }
raft_set_state(me, RAFT_STATE_PRECANDIDATE); raft_set_state(me, RAFT_STATE_PRECANDIDATE);
...@@ -505,7 +526,7 @@ int raft_become_candidate(raft_server_t *me) ...@@ -505,7 +526,7 @@ int raft_become_candidate(raft_server_t *me)
} }
for (int 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_set_voted_for_me(me->nodes[i], 0);
} }
if (raft_node_is_voting(me->node)) { if (raft_node_is_voting(me->node)) {
...@@ -1110,7 +1131,7 @@ int raft_recv_requestvote_response(raft_server_t *me, ...@@ -1110,7 +1131,7 @@ int raft_recv_requestvote_response(raft_server_t *me,
if (resp->vote_granted) { if (resp->vote_granted) {
if (node) { if (node) {
raft_node_vote_for_me(node, 1); raft_node_set_voted_for_me(node, 1);
} }
int votes = raft_get_nvotes_for_me(me); int votes = raft_get_nvotes_for_me(me);
...@@ -1269,31 +1290,8 @@ int raft_apply_entry(raft_server_t* me) ...@@ -1269,31 +1290,8 @@ int raft_apply_entry(raft_server_t* me)
if (idx == me->voting_cfg_change_log_idx) if (idx == me->voting_cfg_change_log_idx)
me->voting_cfg_change_log_idx = -1; me->voting_cfg_change_log_idx = -1;
if (!raft_entry_is_cfg_change(ety)) if (raft_entry_is_cfg_change(ety))
goto exit; raft_handle_apply_cfg_change(me, ety, idx);
raft_node_id_t node_id = me->cb.get_node_id(me, me->udata, ety, idx);
raft_node_t* node = raft_get_node(me, node_id);
switch (ety->type) {
case RAFT_LOGTYPE_ADD_NODE:
raft_node_set_addition_committed(node, 1);
raft_node_set_voting_committed(node, 1);
raft_node_set_has_sufficient_logs(node, 1);
break;
case RAFT_LOGTYPE_ADD_NONVOTING_NODE:
raft_node_set_addition_committed(node, 1);
break;
case RAFT_LOGTYPE_REMOVE_NODE:
if (node) {
raft_remove_node(me, node);
}
break;
default:
break;
}
exit:
raft_entry_release(ety); raft_entry_release(ety);
return 0; return 0;
...@@ -1730,10 +1728,7 @@ raft_index_t raft_get_first_entry_idx(raft_server_t* me) ...@@ -1730,10 +1728,7 @@ raft_index_t raft_get_first_entry_idx(raft_server_t* me)
{ {
assert(0 < raft_get_current_idx(me)); assert(0 < raft_get_current_idx(me));
if (me->snapshot_last_idx == 0) return me->log_impl->first_idx(me->log);
return 1;
return me->snapshot_last_idx;
} }
raft_index_t raft_get_num_snapshottable_logs(raft_server_t *me) raft_index_t raft_get_num_snapshottable_logs(raft_server_t *me)
......
...@@ -10,15 +10,9 @@ ...@@ -10,15 +10,9 @@
#include "raft_log.h" #include "raft_log.h"
#include "raft_private.h" #include "raft_private.h"
void TestRaft_is_voting_by_default(CuTest * tc)
{
raft_node_t *p = raft_node_new((void*)1, 1);
CuAssertTrue(tc, raft_node_is_voting(p));
}
void TestRaft_node_set_nextIdx(CuTest * tc) void TestRaft_node_set_nextIdx(CuTest * tc)
{ {
raft_node_t *p = raft_node_new((void*)1, 1); raft_node_t *p = raft_node_new((void *) 1, 1, 1);
raft_node_set_next_idx(p, 3); raft_node_set_next_idx(p, 3);
CuAssertTrue(tc, 3 == raft_node_get_next_idx(p)); CuAssertTrue(tc, 3 == raft_node_get_next_idx(p));
} }
...@@ -29,7 +23,6 @@ int main(void) ...@@ -29,7 +23,6 @@ int main(void)
CuString *output = CuStringNew(); CuString *output = CuStringNew();
CuSuite* suite = CuSuiteNew(); CuSuite* suite = CuSuiteNew();
SUITE_ADD_TEST(suite, TestRaft_is_voting_by_default);
SUITE_ADD_TEST(suite, TestRaft_node_set_nextIdx); SUITE_ADD_TEST(suite, TestRaft_node_set_nextIdx);
CuSuiteRun(suite); CuSuiteRun(suite);
......
...@@ -3488,6 +3488,7 @@ void TestRaft_leader_recv_appendentries_response_increment_idx_of_node( ...@@ -3488,6 +3488,7 @@ void TestRaft_leader_recv_appendentries_response_increment_idx_of_node(
/* receive mock success responses */ /* receive mock success responses */
raft_appendentries_resp_t aer; raft_appendentries_resp_t aer;
memset(&aer, 0, sizeof(raft_appendentries_resp_t));
aer.term = 1; aer.term = 1;
aer.success = 1; aer.success = 1;
aer.current_idx = 0; aer.current_idx = 0;
......
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