Skip to content

Commit

Permalink
Move membership changes inhouse
Browse files Browse the repository at this point in the history
  • Loading branch information
@willemt
willemt committed Jun 10, 2016
1 parent a833429 commit 2964ce9
Show file tree
Hide file tree
Showing 4 changed files with 100 additions and 63 deletions.
2 changes: 2 additions & 0 deletions deps/raft/raft_log.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,7 @@ int log_append_entry(log_t* me_, raft_entry_t* c)
{
void* ud = raft_get_udata(me->raft);
e = me->cb->log_offer(me->raft, ud, c, me->back);
raft_offer_log(me->raft, c, me->back);
if (e == RAFT_ERR_SHUTDOWN)
return e;
}
Expand Down Expand Up @@ -184,6 +185,7 @@ void log_delete(log_t* me_, int idx)
if (me->cb && me->cb->log_pop)
me->cb->log_pop(me->raft, raft_get_udata(me->raft),
&me->entries[me->back - 1], me->back);
raft_pop_log(me->raft, &me->entries[me->back - 1], me->back);
me->back--;
me->count--;
}
Expand Down
4 changes: 4 additions & 0 deletions deps/raft/raft_private.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -126,4 +126,8 @@ int raft_node_has_sufficient_logs(raft_node_t* me_);

int raft_votes_is_majority(const int nnodes, const int nvotes);

void raft_pop_log(raft_server_t* me_, raft_entry_t* ety, const int idx);

void raft_offer_log(raft_server_t* me_, raft_entry_t* ety, const int idx);

#endif /* RAFT_PRIVATE_H_ */
91 changes: 91 additions & 0 deletions deps/raft/raft_server.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1102,3 +1102,94 @@ int raft_recv_entries_response(raft_server_t* me_,
}
return 0;
}

void raft_offer_log(raft_server_t* me_, raft_entry_t* ety, const int idx)
{
raft_server_private_t* me = (raft_server_private_t*)me_;

if (!raft_entry_is_cfg_change(ety))
return;

int node_id = me->cb.log_get_node_id(me_, raft_get_udata(me_), ety, idx);
raft_node_t* node = raft_get_node(me_, node_id);
int is_self = node_id == raft_get_nodeid(me_);

switch (ety->type)
{
case RAFT_LOGTYPE_ADD_NONVOTING_NODE:
if (!is_self)
{
raft_node_t* node = raft_add_non_voting_node(me_, NULL, node_id, is_self);
assert(node);
}
break;

case RAFT_LOGTYPE_ADD_NODE:
node = raft_add_node(me_, NULL, node_id, is_self);
assert(node);
assert(raft_node_is_voting(node));
break;

case RAFT_LOGTYPE_DEMOTE_NODE:
raft_node_set_voting(node, 0);
break;

case RAFT_LOGTYPE_REMOVE_NODE:
if (node)
raft_remove_node(me_, node);
break;

default:
assert(0);
}
}

void raft_pop_log(raft_server_t* me_, raft_entry_t* ety, const int idx)
{
raft_server_private_t* me = (raft_server_private_t*)me_;

if (!raft_entry_is_cfg_change(ety))
return;

int node_id = me->cb.log_get_node_id(me_, raft_get_udata(me_), ety, idx);

switch (ety->type)
{
case RAFT_LOGTYPE_DEMOTE_NODE:
{
raft_node_t* node = raft_get_node(me_, node_id);
raft_node_set_voting(node, 1);
}
break;

case RAFT_LOGTYPE_REMOVE_NODE:
{
int is_self = node_id == raft_get_nodeid(me_);
raft_node_t* node = raft_add_non_voting_node(me_, NULL, node_id, is_self);
assert(node);
}
break;

case RAFT_LOGTYPE_ADD_NONVOTING_NODE:
{
int is_self = node_id == raft_get_nodeid(me_);
raft_node_t* node = raft_get_node(me_, node_id);
raft_remove_node(me_, node);
if (is_self)
assert(0);
}
break;

case RAFT_LOGTYPE_ADD_NODE:
{
raft_node_t* node = raft_get_node(me_, node_id);
raft_node_set_voting(node, 0);
}
break;

default:
assert(0);
break;
}

}
66 changes: 3 additions & 63 deletions src/main.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -372,48 +372,6 @@ static int __raft_logentry_offer(
if (server)
server->total_offer_count += 1;

if (!raft_entry_is_cfg_change(ety))
return 0;

entry_cfg_change_t *chg = (void*)ety->data.buf;
raft_node_t* node = raft_get_node(r, chg->node_id);
int is_self = chg->node_id == raft_get_nodeid(r);

switch (ety->type)
{
case RAFT_LOGTYPE_ADD_NONVOTING_NODE:
/* printf("ADD %0.10d to %0.10d\n", chg->node_id, raft_get_nodeid(r)); */
if (!is_self)
{
raft_node_t* node = raft_add_non_voting_node(r, NULL, chg->node_id, is_self);
assert(node);
}
break;

case RAFT_LOGTYPE_ADD_NODE:
/* printf("ADDV voting %0.10d to %0.10d\n", chg->node_id, raft_get_nodeid(r)); */
node = raft_add_node(r, NULL, chg->node_id, is_self);
assert(node);
assert(raft_node_is_voting(node));
break;

case RAFT_LOGTYPE_DEMOTE_NODE:
raft_node_set_voting(node, 0);
break;

case RAFT_LOGTYPE_REMOVE_NODE:
printf("REM %0.10d from %0.10d\n", chg->node_id, raft_get_nodeid(r));
if (node)
{
printf("REM OK\n");
raft_remove_node(r, node);
}
break;

default:
assert(0);
}

return 0;
}

Expand Down Expand Up @@ -442,34 +400,23 @@ static int __raft_logentry_pop(
)
{
entry_cfg_change_t *chg = (void*)ety->data.buf;
system_t* sys = udata;

sys->log_pops += 1;

if (!raft_entry_is_cfg_change(ety))
return 0;

printf("POPPING %0.10d type: %d from %0.10d\n",
chg->node_id, ety->type, raft_get_nodeid(raft));

system_t* sys = udata;

server_t* sv = __get_server_from_nodeid(sys, chg->node_id);

sys->log_pops += 1;

switch (ety->type)
{
case RAFT_LOGTYPE_DEMOTE_NODE:
{
/* int is_self = chg->node_id == raft_get_nodeid(raft); */
raft_node_t* node = raft_get_node(raft, chg->node_id);
raft_node_set_voting(node, 1);
}
break;

case RAFT_LOGTYPE_REMOVE_NODE:
{
int is_self = chg->node_id == raft_get_nodeid(raft);
raft_node_t* node = raft_add_non_voting_node(raft, NULL, chg->node_id, is_self);
assert(node);
if (is_self)
sv->connected = NODE_CONNECTED;
}
Expand All @@ -478,21 +425,14 @@ static int __raft_logentry_pop(
case RAFT_LOGTYPE_ADD_NONVOTING_NODE:
{
int is_self = chg->node_id == raft_get_nodeid(raft);
raft_node_t* node = raft_get_node(raft, chg->node_id);
raft_remove_node(raft, node);
if (is_self)
{
assert(0);
sv->connected = NODE_DISCONNECTED;
}
}
break;

case RAFT_LOGTYPE_ADD_NODE:
{
int is_self = chg->node_id == raft_get_nodeid(raft);
raft_node_t* node = raft_get_node(raft, chg->node_id);
raft_node_set_voting(node, 0);
if (is_self)
{
assert(sv->connected == NODE_CONNECTING);
Expand Down

0 comments on commit 2964ce9

Please sign in to comment.