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

Replaced node idx with node pointer

parent 41210d8e
...@@ -142,7 +142,7 @@ typedef int ( ...@@ -142,7 +142,7 @@ typedef int (
) ( ) (
raft_server_t* raft, raft_server_t* raft,
void *user_data, void *user_data,
int node, raft_node_t* node,
msg_requestvote_t* msg msg_requestvote_t* msg
); );
...@@ -157,7 +157,7 @@ typedef int ( ...@@ -157,7 +157,7 @@ typedef int (
) ( ) (
raft_server_t* raft, raft_server_t* raft,
void *user_data, void *user_data,
int node, raft_node_t* node,
msg_appendentries_t* msg msg_appendentries_t* msg
); );
...@@ -203,7 +203,7 @@ typedef int ( ...@@ -203,7 +203,7 @@ typedef int (
) ( ) (
raft_server_t* raft, raft_server_t* raft,
void *user_data, void *user_data,
const int voted_for int node
); );
/** Callback for saving log entry changes. /** Callback for saving log entry changes.
...@@ -300,17 +300,6 @@ void raft_free(raft_server_t* me); ...@@ -300,17 +300,6 @@ void raft_free(raft_server_t* me);
* @param[in] user_data "User data" - user's context that's included in a callback */ * @param[in] user_data "User data" - user's context that's included in a callback */
void raft_set_callbacks(raft_server_t* me, raft_cbs_t* funcs, void* user_data); void raft_set_callbacks(raft_server_t* me, raft_cbs_t* funcs, void* user_data);
/** Set configuration.
*
* @deprecated This function has been replaced by raft_add_node and
* raft_remove_node
*
* @param[in] nodes Array of nodes. End of array is marked by NULL entry
* @param[in] my_idx Index of the node that refers to this Raft server */
void raft_set_configuration(raft_server_t* me,
raft_node_configuration_t* nodes, int my_idx)
__attribute__ ((deprecated));
/** Add node. /** Add node.
* *
* @note This library does not yet support membership changes. * @note This library does not yet support membership changes.
...@@ -325,9 +314,11 @@ __attribute__ ((deprecated)); ...@@ -325,9 +314,11 @@ __attribute__ ((deprecated));
* Examples of what this could be: * Examples of what this could be:
* - void* pointing to implementor's networking data * - void* pointing to implementor's networking data
* - a (IP,Port) tuple * - a (IP,Port) tuple
* @param[in] id The integer ID of this node
* This is used for identifying clients across sessions.
* @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 0 on success; otherwise -1 */ * @return 0 on success; otherwise -1 */
int raft_add_node(raft_server_t* me, void* user_data, int is_self); raft_node_t* raft_add_node(raft_server_t* me, void* user_data, int id, int is_self);
#define raft_add_peer raft_add_node #define raft_add_peer raft_add_node
...@@ -364,7 +355,7 @@ int raft_periodic(raft_server_t* me, int msec_elapsed); ...@@ -364,7 +355,7 @@ int raft_periodic(raft_server_t* me, int msec_elapsed);
* @param[out] r The resulting response * @param[out] r The resulting response
* @return 0 on success */ * @return 0 on success */
int raft_recv_appendentries(raft_server_t* me, int raft_recv_appendentries(raft_server_t* me,
int node, raft_node_t* node,
msg_appendentries_t* ae, msg_appendentries_t* ae,
msg_appendentries_response_t *r); msg_appendentries_response_t *r);
...@@ -373,7 +364,7 @@ int raft_recv_appendentries(raft_server_t* me, ...@@ -373,7 +364,7 @@ int raft_recv_appendentries(raft_server_t* me,
* @param[in] r The appendentries response message * @param[in] r The appendentries response message
* @return 0 on success */ * @return 0 on success */
int raft_recv_appendentries_response(raft_server_t* me, int raft_recv_appendentries_response(raft_server_t* me,
int node, raft_node_t* node,
msg_appendentries_response_t* r); msg_appendentries_response_t* r);
/** Receive a requestvote message. /** Receive a requestvote message.
...@@ -382,7 +373,7 @@ int raft_recv_appendentries_response(raft_server_t* me, ...@@ -382,7 +373,7 @@ int raft_recv_appendentries_response(raft_server_t* me,
* @param[out] r The resulting response * @param[out] r The resulting response
* @return 0 on success */ * @return 0 on success */
int raft_recv_requestvote(raft_server_t* me, int raft_recv_requestvote(raft_server_t* me,
int node, raft_node_t* node,
msg_requestvote_t* vr, msg_requestvote_t* vr,
msg_requestvote_response_t *r); msg_requestvote_response_t *r);
...@@ -391,7 +382,7 @@ int raft_recv_requestvote(raft_server_t* me, ...@@ -391,7 +382,7 @@ int raft_recv_requestvote(raft_server_t* me,
* @param[in] r The requestvote response message * @param[in] r The requestvote response message
* @return 0 on success */ * @return 0 on success */
int raft_recv_requestvote_response(raft_server_t* me, int raft_recv_requestvote_response(raft_server_t* me,
int node, raft_node_t* node,
msg_requestvote_response_t* r); msg_requestvote_response_t* r);
/** Receive an entry message from the client. /** Receive an entry message from the client.
...@@ -419,7 +410,7 @@ int raft_recv_requestvote_response(raft_server_t* me, ...@@ -419,7 +410,7 @@ int raft_recv_requestvote_response(raft_server_t* me,
* @param[out] r The resulting response * @param[out] r The resulting response
* @return 0 on success, -1 on failure */ * @return 0 on success, -1 on failure */
int raft_recv_entry(raft_server_t* me, int raft_recv_entry(raft_server_t* me,
int node, raft_node_t* node,
msg_entry_t* ety, msg_entry_t* ety,
msg_entry_response_t *r); msg_entry_response_t *r);
...@@ -499,7 +490,7 @@ raft_entry_t* raft_get_entry_from_idx(raft_server_t* me, int idx); ...@@ -499,7 +490,7 @@ raft_entry_t* raft_get_entry_from_idx(raft_server_t* me, int idx);
/** /**
* @param[in] node The node's index * @param[in] node The node's index
* @return node pointed to by node index */ * @return node pointed to by node index */
raft_node_t* raft_get_node(raft_server_t *me, int node); raft_node_t* raft_get_node(raft_server_t* me_, const int id);
/** /**
* @return number of votes this server has received this election */ * @return number of votes this server has received this election */
...@@ -525,7 +516,7 @@ int raft_get_my_id(raft_server_t* me); ...@@ -525,7 +516,7 @@ int raft_get_my_id(raft_server_t* me);
/** Vote for a server. /** Vote for a server.
* This should be used to reload persistent state, ie. the voted-for field. * This should be used to reload persistent state, ie. the voted-for field.
* @param[in] node The server to vote for */ * @param[in] node The server to vote for */
void raft_vote(raft_server_t* me, const int node); void raft_vote(raft_server_t* me_, raft_node_t* node);
/** Set the current term. /** Set the current term.
* This should be used to reload persistent state, ie. the current_term field. * This should be used to reload persistent state, ie. the current_term field.
...@@ -542,4 +533,8 @@ int raft_append_entry(raft_server_t* me, raft_entry_t* ety); ...@@ -542,4 +533,8 @@ int raft_append_entry(raft_server_t* me, raft_entry_t* ety);
int raft_msg_entry_response_committed(raft_server_t* me_, int raft_msg_entry_response_committed(raft_server_t* me_,
const msg_entry_response_t* r); const msg_entry_response_t* r);
/** Get node's ID.
* @return ID of node */
int raft_node_get_id(raft_node_t* me_);
#endif /* RAFT_H_ */ #endif /* RAFT_H_ */
...@@ -28,7 +28,7 @@ typedef struct { ...@@ -28,7 +28,7 @@ typedef struct {
/* The candidate the server voted for in its current term, /* The candidate the server voted for in its current term,
* or Nil if it hasn't voted for any. */ * or Nil if it hasn't voted for any. */
int voted_for; raft_node_t* voted_for;
/* the log which is replicated */ /* the log which is replicated */
void* log; void* log;
...@@ -47,9 +47,6 @@ typedef struct { ...@@ -47,9 +47,6 @@ typedef struct {
/* amount of time left till timeout */ /* amount of time left till timeout */
int timeout_elapsed; int timeout_elapsed;
/* who has voted for me. This is an array with N = 'num_nodes' elements */
int *votes_for_me;
raft_node_t* nodes; raft_node_t* nodes;
int num_nodes; int num_nodes;
...@@ -58,14 +55,14 @@ typedef struct { ...@@ -58,14 +55,14 @@ typedef struct {
/* what this node thinks is the node ID of the current leader, or -1 if /* what this node thinks is the node ID of the current leader, or -1 if
* there isn't a known current leader. */ * there isn't a known current leader. */
int current_leader; raft_node_t* current_leader;
/* callbacks */ /* callbacks */
raft_cbs_t cb; raft_cbs_t cb;
void* udata; void* udata;
/* my node ID */ /* my node ID */
int nodeid; raft_node_t* node;
} raft_server_private_t; } raft_server_private_t;
void raft_election_start(raft_server_t* me); void raft_election_start(raft_server_t* me);
...@@ -76,15 +73,15 @@ void raft_become_candidate(raft_server_t* me); ...@@ -76,15 +73,15 @@ void raft_become_candidate(raft_server_t* me);
void raft_become_follower(raft_server_t* me); void raft_become_follower(raft_server_t* me);
void raft_vote(raft_server_t* me, int node); void raft_vote(raft_server_t* me, raft_node_t* node);
void raft_set_current_term(raft_server_t* me,int term); void raft_set_current_term(raft_server_t* me,int term);
/** /**
* @return 0 on error */ * @return 0 on error */
int raft_send_requestvote(raft_server_t* me, int node); int raft_send_requestvote(raft_server_t* me, raft_node_t* node);
void raft_send_appendentries(raft_server_t* me, int node); int raft_send_appendentries(raft_server_t* me, raft_node_t* node);
void raft_send_appendentries_all(raft_server_t* me_); void raft_send_appendentries_all(raft_server_t* me_);
...@@ -107,7 +104,7 @@ void raft_set_last_applied_idx(raft_server_t* me, int idx); ...@@ -107,7 +104,7 @@ void raft_set_last_applied_idx(raft_server_t* me, int idx);
void raft_set_state(raft_server_t* me_, int state); void raft_set_state(raft_server_t* me_, int state);
int raft_get_state(raft_server_t* me_); int raft_get_state(raft_server_t* me_);
raft_node_t* raft_node_new(void* udata); raft_node_t* raft_node_new(void* udata, int id);
void raft_node_set_next_idx(raft_node_t* node, int nextIdx); void raft_node_set_next_idx(raft_node_t* node, int nextIdx);
...@@ -115,6 +112,10 @@ void raft_node_set_match_idx(raft_node_t* node, int matchIdx); ...@@ -115,6 +112,10 @@ void raft_node_set_match_idx(raft_node_t* node, int matchIdx);
int raft_node_get_match_idx(raft_node_t* me_); int raft_node_get_match_idx(raft_node_t* me_);
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_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_ */
...@@ -19,17 +19,24 @@ ...@@ -19,17 +19,24 @@
typedef struct typedef struct
{ {
void* udata; void* udata;
int next_idx; int next_idx;
int match_idx; int match_idx;
/* 1 if node has voted for me */
int voted_for_me;
int id;
} raft_node_private_t; } raft_node_private_t;
raft_node_t* raft_node_new(void* udata) raft_node_t* raft_node_new(void* udata, int id)
{ {
raft_node_private_t* me; raft_node_private_t* me;
me = (raft_node_private_t*)calloc(1, sizeof(raft_node_private_t)); me = (raft_node_private_t*)calloc(1, sizeof(raft_node_private_t));
me->udata = udata; me->udata = udata;
me->next_idx = 1; me->next_idx = 1;
me->match_idx = 0; me->match_idx = 0;
me->id = id;
return (raft_node_t*)me; return (raft_node_t*)me;
} }
...@@ -46,7 +53,6 @@ void raft_node_set_next_idx(raft_node_t* me_, int nextIdx) ...@@ -46,7 +53,6 @@ void raft_node_set_next_idx(raft_node_t* me_, int nextIdx)
me->next_idx = nextIdx < 1 ? 1 : nextIdx; me->next_idx = nextIdx < 1 ? 1 : nextIdx;
} }
int raft_node_get_match_idx(raft_node_t* me_) int raft_node_get_match_idx(raft_node_t* me_)
{ {
raft_node_private_t* me = (raft_node_private_t*)me_; raft_node_private_t* me = (raft_node_private_t*)me_;
...@@ -59,7 +65,6 @@ void raft_node_set_match_idx(raft_node_t* me_, int matchIdx) ...@@ -59,7 +65,6 @@ void raft_node_set_match_idx(raft_node_t* me_, int matchIdx)
me->match_idx = matchIdx; me->match_idx = matchIdx;
} }
void* raft_node_get_udata(raft_node_t* me_) void* raft_node_get_udata(raft_node_t* me_)
{ {
raft_node_private_t* me = (raft_node_private_t*)me_; raft_node_private_t* me = (raft_node_private_t*)me_;
...@@ -71,3 +76,21 @@ void raft_node_set_udata(raft_node_t* me_, void* udata) ...@@ -71,3 +76,21 @@ void raft_node_set_udata(raft_node_t* me_, void* udata)
raft_node_private_t* me = (raft_node_private_t*)me_; raft_node_private_t* me = (raft_node_private_t*)me_;
me->udata = udata; me->udata = udata;
} }
void raft_node_vote_for_me(raft_node_t* me_, const int vote)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
me->voted_for_me = vote;
}
int raft_node_has_vote_for_me(raft_node_t* me_)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return me->voted_for_me;
}
int raft_node_get_id(raft_node_t* me_)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return me->id;
}
...@@ -44,13 +44,13 @@ raft_server_t* raft_new() ...@@ -44,13 +44,13 @@ raft_server_t* raft_new()
if (!me) if (!me)
return NULL; return NULL;
me->current_term = 0; me->current_term = 0;
me->voted_for = -1; me->voted_for = NULL;
me->timeout_elapsed = 0; me->timeout_elapsed = 0;
me->request_timeout = 200; me->request_timeout = 200;
me->election_timeout = 1000; me->election_timeout = 1000;
me->log = log_new(); me->log = log_new();
raft_set_state((raft_server_t*)me, RAFT_STATE_FOLLOWER); raft_set_state((raft_server_t*)me, RAFT_STATE_FOLLOWER);
me->current_leader = -1; me->current_leader = NULL;
return (raft_server_t*)me; return (raft_server_t*)me;
} }
...@@ -89,14 +89,15 @@ void raft_become_leader(raft_server_t* me_) ...@@ -89,14 +89,15 @@ void raft_become_leader(raft_server_t* me_)
__log(me_, "becoming leader"); __log(me_, "becoming leader");
raft_set_state(me_, RAFT_STATE_LEADER); raft_set_state(me_, RAFT_STATE_LEADER);
for (i = 0; i < me->num_nodes; i++) for (i = 0; i < me->num_nodes; i++)
{ {
if (me->nodeid != i) if (me->node != me->nodes[i])
{ {
raft_node_t* p = raft_get_node(me_, i); raft_node_t* node = me->nodes[i];
raft_node_set_next_idx(p, raft_get_current_idx(me_) + 1); raft_node_set_next_idx(node, raft_get_current_idx(me_) + 1);
raft_node_set_match_idx(p, 0); raft_node_set_match_idx(node, 0);
raft_send_appendentries(me_, i); raft_send_appendentries(me_, node);
} }
} }
} }
...@@ -108,25 +109,27 @@ void raft_become_candidate(raft_server_t* me_) ...@@ -108,25 +109,27 @@ void raft_become_candidate(raft_server_t* me_)
__log(me_, "becoming candidate"); __log(me_, "becoming candidate");
memset(me->votes_for_me, 0, sizeof(int) * me->num_nodes); for (i = 0; i < me->num_nodes; i++)
raft_node_vote_for_me(me->nodes[i], 0);
me->current_term += 1; me->current_term += 1;
raft_vote(me_, me->nodeid); raft_vote(me_, me->node);
me->current_leader = -1; me->current_leader = NULL;
raft_set_state(me_, RAFT_STATE_CANDIDATE); raft_set_state(me_, RAFT_STATE_CANDIDATE);
/* we need a random factor here to prevent simultaneous candidates */ /* we need a random factor here to prevent simultaneous candidates */
me->timeout_elapsed = rand() % me->election_timeout; me->timeout_elapsed = rand() % me->election_timeout;
for (i = 0; i < me->num_nodes; i++) for (i = 0; i < me->num_nodes; i++)
if (me->nodeid != i) if (me->node != me->nodes[i])
raft_send_requestvote(me_, i); raft_send_requestvote(me_, me->nodes[i]);
} }
void raft_become_follower(raft_server_t* me_) void raft_become_follower(raft_server_t* me_)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_;
__log(me_, "becoming follower"); __log(me_, "becoming follower");
raft_set_state(me_, RAFT_STATE_FOLLOWER); raft_set_state(me_, RAFT_STATE_FOLLOWER);
raft_vote(me_, -1); me->voted_for = NULL;
} }
int raft_periodic(raft_server_t* me_, int msec_since_last_period) int raft_periodic(raft_server_t* me_, int msec_since_last_period)
...@@ -162,10 +165,14 @@ raft_entry_t* raft_get_entry_from_idx(raft_server_t* me_, int etyidx) ...@@ -162,10 +165,14 @@ raft_entry_t* raft_get_entry_from_idx(raft_server_t* me_, int etyidx)
} }
int raft_recv_appendentries_response(raft_server_t* me_, int raft_recv_appendentries_response(raft_server_t* me_,
int node, msg_appendentries_response_t* r) raft_node_t* node,
msg_appendentries_response_t* r)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
assert(node);
assert(node != me->node);
__log(me_, __log(me_,
"received appendentries response node: %d %s cidx: %d 1stidx: %d", "received appendentries response node: %d %s cidx: %d 1stidx: %d",
node, r->success == 1 ? "success" : "fail", r->current_idx, node, r->success == 1 ? "success" : "fail", r->current_idx,
...@@ -176,36 +183,34 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -176,36 +183,34 @@ int raft_recv_appendentries_response(raft_server_t* me_,
if (!raft_is_leader(me_)) if (!raft_is_leader(me_))
return -1; return -1;
raft_node_t* p = raft_get_node(me_, node);
if (0 == r->success) if (0 == r->success)
{ {
/* If AppendEntries fails because of log inconsistency: /* If AppendEntries fails because of log inconsistency:
decrement nextIndex and retry (§5.3) */ decrement nextIndex and retry (§5.3) */
assert(0 <= raft_node_get_next_idx(p)); assert(0 <= raft_node_get_next_idx(node));
// TODO can jump back to where node is different instead of iterating // TODO can jump back to where node is different instead of iterating
raft_node_set_next_idx(p, raft_node_get_next_idx(p) - 1); raft_node_set_next_idx(node, raft_node_get_next_idx(node) - 1);
/* retry */ /* retry */
raft_send_appendentries(me_, node); raft_send_appendentries(me_, node);
return 0; return 0;
} }
raft_node_set_next_idx(p, r->current_idx + 1); raft_node_set_next_idx(node, r->current_idx + 1);
raft_node_set_match_idx(p, r->current_idx); raft_node_set_match_idx(node, r->current_idx);
/* 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;
int i; int i;
for (i = 0; i < me->num_nodes; i++) /* Check others */ for (i = 0; i < me->num_nodes; i++) /* Check others */
if (i != me->nodeid && point <= raft_node_get_match_idx(p)) if (me->node != me->nodes[i] && point <= raft_node_get_match_idx(me->nodes[i]))
votes++; votes++;
if (me->num_nodes / 2 < votes && raft_get_commit_idx(me_) < point) if (me->num_nodes / 2 < votes && raft_get_commit_idx(me_) < point)
raft_set_commit_idx(me_, point); raft_set_commit_idx(me_, point);
/* Aggressively send remaining entries */ /* Aggressively send remaining entries */
if (raft_get_entry_from_idx(me_, raft_node_get_next_idx(p))) if (raft_get_entry_from_idx(me_, raft_node_get_next_idx(node)))
raft_send_appendentries(me_, node); raft_send_appendentries(me_, node);
/* periodic applies committed entries lazily */ /* periodic applies committed entries lazily */
...@@ -215,13 +220,16 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -215,13 +220,16 @@ int raft_recv_appendentries_response(raft_server_t* me_,
int raft_recv_appendentries( int raft_recv_appendentries(
raft_server_t* me_, raft_server_t* me_,
const int node, raft_node_t* node,
msg_appendentries_t* ae, msg_appendentries_t* ae,
msg_appendentries_response_t *r msg_appendentries_response_t *r
) )
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
assert(node);
assert(node != me->node);
me->timeout_elapsed = 0; me->timeout_elapsed = 0;
if (0 < ae->n_entries) if (0 < ae->n_entries)
...@@ -330,7 +338,7 @@ static int __should_grant_vote(raft_server_private_t* me, msg_requestvote_t* vr) ...@@ -330,7 +338,7 @@ static int __should_grant_vote(raft_server_private_t* me, msg_requestvote_t* vr)
return 0; return 0;
/* we've already voted */ /* we've already voted */
if (-1 != me->voted_for) if (me->voted_for)
return 0; return 0;
/* we have a more up-to-date log */ /* we have a more up-to-date log */
...@@ -339,12 +347,17 @@ static int __should_grant_vote(raft_server_private_t* me, msg_requestvote_t* vr) ...@@ -339,12 +347,17 @@ static int __should_grant_vote(raft_server_private_t* me, msg_requestvote_t* vr)
return 1; return 1;
} }
int raft_recv_requestvote(raft_server_t* me_, int node, msg_requestvote_t* vr, int raft_recv_requestvote(raft_server_t* me_,
raft_node_t* node,
msg_requestvote_t* vr,
msg_requestvote_response_t *r) msg_requestvote_response_t *r)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
assert(node);
assert(node != me->node);
if (raft_get_current_term(me_) < vr->term) if (raft_get_current_term(me_) < vr->term)
raft_become_follower(me_); raft_become_follower(me_);
...@@ -358,7 +371,7 @@ int raft_recv_requestvote(raft_server_t* me_, int node, msg_requestvote_t* vr, ...@@ -358,7 +371,7 @@ int raft_recv_requestvote(raft_server_t* me_, int node, msg_requestvote_t* vr,
r->vote_granted = 1; r->vote_granted = 1;
/* there must be in an election. */ /* there must be in an election. */
me->current_leader = -1; me->current_leader = NULL;
} }
else else
r->vote_granted = 0; r->vote_granted = 0;
...@@ -378,26 +391,27 @@ int raft_votes_is_majority(const int num_nodes, const int nvotes) ...@@ -378,26 +391,27 @@ int raft_votes_is_majority(const int num_nodes, const int nvotes)
return half + 1 <= nvotes; return half + 1 <= nvotes;
} }
int raft_recv_requestvote_response(raft_server_t* me_, int node, int raft_recv_requestvote_response(raft_server_t* me_, raft_node_t* node,
msg_requestvote_response_t* r) msg_requestvote_response_t* r)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
assert(node);
assert(node != me->node);
__log(me_, "node responded to requestvote: %d status: %s", __log(me_, "node responded to requestvote: %d status: %s",
node, r->vote_granted == 1 ? "granted" : "not granted"); node, r->vote_granted == 1 ? "granted" : "not granted");
if (raft_is_leader(me_)) if (raft_is_leader(me_))
return 0; return 0;
assert(node < me->num_nodes);
// TODO: if invalid leader then stepdown // TODO: if invalid leader then stepdown
// if (r->term != raft_get_current_term(me_)) // if (r->term != raft_get_current_term(me_))
// return 0; // return 0;
if (1 == r->vote_granted) if (1 == r->vote_granted)
{ {
me->votes_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(me->num_nodes, votes))
raft_become_leader(me_); raft_become_leader(me_);
...@@ -406,7 +420,7 @@ int raft_recv_requestvote_response(raft_server_t* me_, int node, ...@@ -406,7 +420,7 @@ int raft_recv_requestvote_response(raft_server_t* me_, int node,
return 0; return 0;
} }
int raft_recv_entry(raft_server_t* me_, int node, msg_entry_t* e, int raft_recv_entry(raft_server_t* me_, raft_node_t* node, msg_entry_t* e,
msg_entry_response_t *r) msg_entry_response_t *r)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
...@@ -426,11 +440,12 @@ int raft_recv_entry(raft_server_t* me_, int node, msg_entry_t* e, ...@@ -426,11 +440,12 @@ int raft_recv_entry(raft_server_t* me_, int node, msg_entry_t* e,
/* Only send new entries. /* Only send new entries.
* Don't send the entry to peers who are behind, to prevent them from * Don't send the entry to peers who are behind, to prevent them from
* becomming congested. */ * becomming congested. */
if (me->nodeid != i) { if (me->node != me->nodes[i])
{
int next_idx = raft_node_get_next_idx(raft_get_node(me_, i)); int next_idx = raft_node_get_next_idx(raft_get_node(me_, i));
int last_log_idx = raft_get_current_idx(me_); int last_log_idx = raft_get_current_idx(me_);
if (next_idx == last_log_idx) if (next_idx == last_log_idx)
raft_send_appendentries(me_, i); raft_send_appendentries(me_, me->nodes[i]);
} }
/* if we're the only node, we can consider the entry committed */ /* if we're the only node, we can consider the entry committed */
...@@ -443,11 +458,14 @@ int raft_recv_entry(raft_server_t* me_, int node, msg_entry_t* e, ...@@ -443,11 +458,14 @@ int raft_recv_entry(raft_server_t* me_, int node, msg_entry_t* e,
return 0; return 0;
} }
int raft_send_requestvote(raft_server_t* me_, int node) int raft_send_requestvote(raft_server_t* me_, raft_node_t* node)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
msg_requestvote_t rv; msg_requestvote_t rv;
assert(node);
assert(node != me->node);
__log(me_, "sending requestvote to: %d", node); __log(me_, "sending requestvote to: %d", node);
rv.term = me->current_term; rv.term = me->current_term;
...@@ -485,14 +503,15 @@ int raft_apply_entry(raft_server_t* me_) ...@@ -485,14 +503,15 @@ int raft_apply_entry(raft_server_t* me_)
return 0; return 0;
} }
void raft_send_appendentries(raft_server_t* me_, int node) int raft_send_appendentries(raft_server_t* me_, raft_node_t* node)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
if (!(me->cb.send_appendentries)) assert(node);
return; assert(node != me->node);
raft_node_t* p = raft_get_node(me_, node); if (!(me->cb.send_appendentries))
return -1;
msg_appendentries_t ae; msg_appendentries_t ae;
ae.term = me->current_term; ae.term = me->current_term;
...@@ -502,7 +521,7 @@ void raft_send_appendentries(raft_server_t* me_, int node) ...@@ -502,7 +521,7 @@ void raft_send_appendentries(raft_server_t* me_, int node)
ae.n_entries = 0; ae.n_entries = 0;
ae.entries = NULL; ae.entries = NULL;
int next_idx = raft_node_get_next_idx(p); int next_idx = raft_node_get_next_idx(node);
msg_entry_t mety; msg_entry_t mety;
...@@ -535,6 +554,8 @@ void raft_send_appendentries(raft_server_t* me_, int node) ...@@ -535,6 +554,8 @@ void raft_send_appendentries(raft_server_t* me_, int node)
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);
return 0;
} }
void raft_send_appendentries_all(raft_server_t* me_) void raft_send_appendentries_all(raft_server_t* me_)
...@@ -544,48 +565,27 @@ void raft_send_appendentries_all(raft_server_t* me_) ...@@ -544,48 +565,27 @@ void raft_send_appendentries_all(raft_server_t* me_)
me->timeout_elapsed = 0; me->timeout_elapsed = 0;
for (i = 0; i < me->num_nodes; i++) for (i = 0; i < me->num_nodes; i++)
if (me->nodeid != i) if (me->node != me->nodes[i])
raft_send_appendentries(me_, i); raft_send_appendentries(me_, me->nodes[i]);
}
void raft_set_configuration(raft_server_t* me_,
raft_node_configuration_t* nodes, int my_idx)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
int num_nodes;
/* TODO: one memory allocation only please */
for (num_nodes = 0; nodes->udata_address; nodes++)
{
num_nodes++;
me->nodes =
(raft_node_t*)realloc(me->nodes, sizeof(raft_node_t*) * num_nodes);
me->num_nodes = num_nodes;
me->nodes[num_nodes - 1] = raft_node_new(nodes->udata_address);
}
me->votes_for_me = (int*)calloc(num_nodes, sizeof(int));
me->nodeid = my_idx;
} }
int raft_add_node(raft_server_t* me_, void* udata, int is_self) raft_node_t* raft_add_node(raft_server_t* me_, void* udata, int id, int is_self)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
/* TODO: does not yet support dynamic membership changes */ /* TODO: does not yet support dynamic membership changes */
if (me->current_term != 0 && me->timeout_elapsed != 0 && if (me->current_term != 0 && me->timeout_elapsed != 0 &&
me->election_timeout != 0) me->election_timeout != 0)
return -1; return NULL;
me->num_nodes++; me->num_nodes++;
me->nodes = me->nodes = (raft_node_t*)realloc(me->nodes, sizeof(raft_node_t*) * me->num_nodes);
(raft_node_t*)realloc(me->nodes, sizeof(raft_node_t*) * me->num_nodes); me->nodes[me->num_nodes - 1] = raft_node_new(udata, id);
me->nodes[me->num_nodes - 1] = raft_node_new(udata); assert(me->nodes[me->num_nodes - 1]);
me->votes_for_me =
(int*)realloc(me->votes_for_me, me->num_nodes * sizeof(int));
me->votes_for_me[me->num_nodes - 1] = 0;
if (is_self) if (is_self)
me->nodeid = me->num_nodes - 1; me->node = me->nodes[me->num_nodes - 1];
return 0;
return me->nodes[me->num_nodes - 1];
} }
int raft_get_nvotes_for_me(raft_server_t* me_) int raft_get_nvotes_for_me(raft_server_t* me_)
...@@ -594,22 +594,23 @@ int raft_get_nvotes_for_me(raft_server_t* me_) ...@@ -594,22 +594,23 @@ int raft_get_nvotes_for_me(raft_server_t* me_)
int i, votes; int i, votes;
for (i = 0, votes = 0; i < me->num_nodes; i++) for (i = 0, votes = 0; i < me->num_nodes; i++)
if (me->nodeid != i) if (me->node != me->nodes[i])
if (1 == me->votes_for_me[i]) if (raft_node_has_vote_for_me(me->nodes[i]))
votes += 1; votes += 1;
if (me->voted_for == me->nodeid) if (me->voted_for == me->node)
votes += 1; votes += 1;
return votes; return votes;
} }
void raft_vote(raft_server_t* me_, const int node) void raft_vote(raft_server_t* me_, raft_node_t* node)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
me->voted_for = node; me->voted_for = node;
if (me->cb.persist_vote) if (me->cb.persist_vote)
me->cb.persist_vote(me_, me->udata, node); me->cb.persist_vote(me_, me->udata, raft_node_get_id(node));
} }
int raft_msg_entry_response_committed(raft_server_t* me_, int raft_msg_entry_response_committed(raft_server_t* me_,
......
...@@ -34,7 +34,7 @@ void raft_set_request_timeout(raft_server_t* me_, int millisec) ...@@ -34,7 +34,7 @@ void raft_set_request_timeout(raft_server_t* me_, int millisec)
int raft_get_nodeid(raft_server_t* me_) int raft_get_nodeid(raft_server_t* me_)
{ {
return ((raft_server_private_t*)me_)->nodeid; return raft_node_get_id(((raft_server_private_t*)me_)->node);
} }
int raft_get_election_timeout(raft_server_t* me_) int raft_get_election_timeout(raft_server_t* me_)
...@@ -65,7 +65,10 @@ int raft_get_log_count(raft_server_t* me_) ...@@ -65,7 +65,10 @@ int raft_get_log_count(raft_server_t* me_)
int raft_get_voted_for(raft_server_t* me_) int raft_get_voted_for(raft_server_t* me_)
{ {
return ((raft_server_private_t*)me_)->voted_for; raft_server_private_t* me = (raft_server_private_t*)me_;
if (me->voted_for)
return raft_node_get_id(me->voted_for);
return -1;
} }
void raft_set_current_term(raft_server_t* me_, const int term) void raft_set_current_term(raft_server_t* me_, const int term)
...@@ -114,7 +117,7 @@ void raft_set_state(raft_server_t* me_, int state) ...@@ -114,7 +117,7 @@ void raft_set_state(raft_server_t* me_, int state)
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
/* if became the leader, then update the current leader entry */ /* if became the leader, then update the current leader entry */
if (state == RAFT_STATE_LEADER) if (state == RAFT_STATE_LEADER)
me->current_leader = me->nodeid; me->current_leader = me->node;
me->state = state; me->state = state;
} }
...@@ -126,16 +129,21 @@ int raft_get_state(raft_server_t* me_) ...@@ -126,16 +129,21 @@ int raft_get_state(raft_server_t* me_)
raft_node_t* raft_get_node(raft_server_t *me_, int nodeid) raft_node_t* raft_get_node(raft_server_t *me_, int nodeid)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
int i;
for (i=0; i<me->num_nodes; i++)
if (nodeid == raft_node_get_id(me->nodes[i]))
return me->nodes[i];
if (nodeid < 0 || me->num_nodes <= nodeid) return NULL;
return NULL;
return (raft_node_t*)me->nodes[nodeid];
} }
int raft_get_current_leader(raft_server_t* me_) int raft_get_current_leader(raft_server_t* me_)
{ {
raft_server_private_t* me = (void*)me_; raft_server_private_t* me = (void*)me_;
return me->current_leader; if (me->current_leader)
return raft_node_get_id(me->current_leader);
return -1;
} }
void* raft_get_udata(raft_server_t* me_) void* raft_get_udata(raft_server_t* me_)
......
...@@ -31,7 +31,7 @@ typedef struct ...@@ -31,7 +31,7 @@ typedef struct
/* what type of message is it? */ /* what type of message is it? */
int type; int type;
/* who sent this? */ /* who sent this? */
int sender; raft_node_t* sender;
} msg_t; } msg_t;
static sender_t** __senders = NULL; static sender_t** __senders = NULL;
...@@ -48,62 +48,68 @@ static int __append_msg( ...@@ -48,62 +48,68 @@ static int __append_msg(
void* data, void* data,
int type, int type,
int len, int len,
int peer, raft_node_t* node,
raft_server_t* raft raft_server_t* raft
) )
{ {
msg_t* m; msg_t* m = malloc(sizeof(msg_t));
m = malloc(sizeof(msg_t));
m->type = type; m->type = type;
m->len = len; m->len = len;
m->data = malloc(len); m->data = malloc(len);
m->sender = raft_get_nodeid(raft); m->sender = raft_get_node(raft, raft_get_nodeid(raft));
memcpy(m->data, data, len); memcpy(m->data, data, len);
llqueue_offer(me->outbox, m); llqueue_offer(me->outbox, m);
if (__nsenders > peer) /* give to peer */
llqueue_offer(__senders[peer]->inbox, m); sender_t* peer = raft_node_get_udata(node);
if (peer)
{
msg_t* m2 = malloc(sizeof(msg_t));
memcpy(m2, m, sizeof(msg_t));
m2->sender = raft_get_node(peer->raft, raft_get_nodeid(raft));
llqueue_offer(peer->inbox, m2);
}
return 1; return 1;
} }
int sender_requestvote(raft_server_t* raft, int sender_requestvote(raft_server_t* raft,
void* udata, int peer, msg_requestvote_t* msg) void* udata, raft_node_t* node, msg_requestvote_t* msg)
{ {
return __append_msg(udata, msg, RAFT_MSG_REQUESTVOTE, sizeof(*msg), peer, return __append_msg(udata, msg, RAFT_MSG_REQUESTVOTE, sizeof(*msg), node,
raft); raft);
} }
int sender_requestvote_response(raft_server_t* raft, int sender_requestvote_response(raft_server_t* raft,
void* udata, int peer, void* udata, raft_node_t* node,
msg_requestvote_response_t* msg) msg_requestvote_response_t* msg)
{ {
return __append_msg(udata, msg, RAFT_MSG_REQUESTVOTE_RESPONSE, sizeof(*msg), return __append_msg(udata, msg, RAFT_MSG_REQUESTVOTE_RESPONSE, sizeof(*msg),
peer, raft); node, raft);
} }
int sender_appendentries(raft_server_t* raft, int sender_appendentries(raft_server_t* raft,
void* udata, int peer, msg_appendentries_t* msg) void* udata, raft_node_t* node, msg_appendentries_t* msg)
{ {
msg_entry_t* entries = calloc(1, sizeof(msg_entry_t) * msg->n_entries); msg_entry_t* entries = calloc(1, sizeof(msg_entry_t) * msg->n_entries);
memcpy(entries, msg->entries, sizeof(msg_entry_t) * msg->n_entries); memcpy(entries, msg->entries, sizeof(msg_entry_t) * msg->n_entries);
msg->entries = entries; msg->entries = entries;
return __append_msg(udata, msg, RAFT_MSG_APPENDENTRIES, sizeof(*msg), peer, return __append_msg(udata, msg, RAFT_MSG_APPENDENTRIES, sizeof(*msg), node,
raft); raft);
} }
int sender_appendentries_response(raft_server_t* raft, int sender_appendentries_response(raft_server_t* raft,
void* udata, int peer, void* udata, raft_node_t* node,
msg_appendentries_response_t* msg) msg_appendentries_response_t* msg)
{ {
return __append_msg(udata, msg, RAFT_MSG_APPENDENTRIES_RESPONSE, return __append_msg(udata, msg, RAFT_MSG_APPENDENTRIES_RESPONSE,
sizeof(*msg), peer, raft); sizeof(*msg), node, raft);
} }
int sender_entries_response(raft_server_t* raft, int sender_entries_response(raft_server_t* raft,
void* udata, int peer, msg_entry_response_t* msg) void* udata, raft_node_t* node, msg_entry_response_t* msg)
{ {
return __append_msg(udata, msg, RAFT_MSG_ENTRY_RESPONSE, sizeof(*msg), peer, return __append_msg(udata, msg, RAFT_MSG_ENTRY_RESPONSE, sizeof(*msg), node,
raft); raft);
} }
......
...@@ -24,21 +24,21 @@ int sender_msgs_available(void* s); ...@@ -24,21 +24,21 @@ int sender_msgs_available(void* s);
void sender_set_raft(void* s, void* r); void sender_set_raft(void* s, void* r);
int sender_requestvote(raft_server_t* raft, int sender_requestvote(raft_server_t* raft,
void* udata, int peer, msg_requestvote_t* msg); void* udata, raft_node_t* node, msg_requestvote_t* msg);
int sender_requestvote_response(raft_server_t* raft, int sender_requestvote_response(raft_server_t* raft,
void* udata, int peer, msg_requestvote_response_t* msg); void* udata, raft_node_t* node, msg_requestvote_response_t* msg);
int sender_appendentries(raft_server_t* raft, int sender_appendentries(raft_server_t* raft,
void* udata, int peer, msg_appendentries_t* msg); void* udata, raft_node_t* node, msg_appendentries_t* msg);
int sender_appendentries_response(raft_server_t* raft, int sender_appendentries_response(raft_server_t* raft,
void* udata, int peer, msg_appendentries_response_t* msg); void* udata, raft_node_t* node, msg_appendentries_response_t* msg);
int sender_entries(raft_server_t* raft, int sender_entries(raft_server_t* raft,
void* udata, int peer, msg_entry_t* msg); void* udata, raft_node_t* node, msg_entry_t* msg);
int sender_entries_response(raft_server_t* raft, int sender_entries_response(raft_server_t* raft,
void* udata, int peer, msg_entry_response_t* msg); void* udata, raft_node_t* node, msg_entry_response_t* msg);
#endif /* MOCK_SEND_FUNCTIONS_H */ #endif /* MOCK_SEND_FUNCTIONS_H */
...@@ -14,8 +14,7 @@ void TestRaft_node_set_nextIdx(CuTest * tc) ...@@ -14,8 +14,7 @@ void TestRaft_node_set_nextIdx(CuTest * tc)
{ {
raft_node_t *p; raft_node_t *p;
p = raft_node_new((void*)1); p = raft_node_new((void*)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));
} }
...@@ -19,15 +19,17 @@ void TestRaft_scenario_leader_appears(CuTest * tc) ...@@ -19,15 +19,17 @@ void TestRaft_scenario_leader_appears(CuTest * tc)
senders_new(); senders_new();
for (j = 0; j < 3; j++)
sender[j] = sender_new((void*)j);
for (j = 0; j < 3; j++) for (j = 0; j < 3; j++)
{ {
r[j] = raft_new(); r[j] = raft_new();
sender[j] = sender_new((void*)j);
sender_set_raft(sender[j], r[j]); sender_set_raft(sender[j], r[j]);
raft_set_election_timeout(r[j], 500); raft_set_election_timeout(r[j], 500);
raft_add_node(r[j], (void*)1, j==0); raft_add_node(r[j], sender[0], 1, j==0);
raft_add_node(r[j], (void*)2, j==1); raft_add_node(r[j], sender[1], 2, j==1);
raft_add_node(r[j], (void*)3, j==2); raft_add_node(r[j], sender[2], 3, j==2);
raft_set_callbacks(r[j], raft_set_callbacks(r[j],
&((raft_cbs_t) { &((raft_cbs_t) {
.send_requestvote = sender_requestvote, .send_requestvote = sender_requestvote,
...@@ -62,4 +64,3 @@ one_more_time: ...@@ -62,4 +64,3 @@ one_more_time:
CuAssertTrue(tc, 0 != leaders); CuAssertTrue(tc, 0 != leaders);
CuAssertTrue(tc, 1 == leaders); CuAssertTrue(tc, 1 == leaders);
} }
This diff is collapsed.
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