Commit cefdd08c authored by Amitabha Roy's avatar Amitabha Roy
Browse files

Use matchIndex from RAFT paper to update commitIndex.

parent ce87cec8
......@@ -111,6 +111,10 @@ raft_node_t* raft_node_new(void* udata);
void raft_node_set_next_idx(raft_node_t* node, int nextIdx);
void raft_node_set_match_idx(raft_node_t* node, int matchIdx);
int raft_node_get_match_idx(raft_node_t* me_);
int raft_votes_is_majority(const int nnodes, const int nvotes);
#endif /* RAFT_PRIVATE_H_ */
......@@ -20,6 +20,7 @@ typedef struct
{
void* udata;
int next_idx;
int match_idx;
} raft_node_private_t;
raft_node_t* raft_node_new(void* udata)
......@@ -27,7 +28,8 @@ raft_node_t* raft_node_new(void* udata)
raft_node_private_t* me;
me = (raft_node_private_t*)calloc(1, sizeof(raft_node_private_t));
me->udata = udata;
me->next_idx = 1;
me->next_idx = 1;
me->match_idx = 0;
return (raft_node_t*)me;
}
......@@ -44,6 +46,20 @@ void raft_node_set_next_idx(raft_node_t* me_, int nextIdx)
me->next_idx = nextIdx < 1 ? 1 : nextIdx;
}
int raft_node_get_match_idx(raft_node_t* me_)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return me->match_idx;
}
void raft_node_set_match_idx(raft_node_t* me_, int matchIdx)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
me->match_idx = matchIdx;
}
void* raft_node_get_udata(raft_node_t* me_)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
......
......@@ -94,6 +94,7 @@ void raft_become_leader(raft_server_t* me_)
{
raft_node_t* p = raft_get_node(me_, i);
raft_node_set_next_idx(p, raft_get_current_idx(me_) + 1);
raft_node_set_match_idx(p, 0);
raft_send_appendentries(me_, i);
}
}
......@@ -191,25 +192,19 @@ int raft_recv_appendentries_response(raft_server_t* me_,
}
raft_node_set_next_idx(p, r->current_idx + 1);
raft_node_set_match_idx(p, r->current_idx);
/* Update commit idx */
int votes = 1; /* include me */
int i;
for (i = r->first_idx; i <= r->current_idx; i++)
log_mark_node_has_committed(me->log, i);
while (1)
{
raft_entry_t* e = raft_get_entry_from_idx(me_, me->last_applied_idx + 1);
/* majority has this */
if (e && me->num_nodes / 2 <= e->num_nodes)
{
if (-1 == raft_apply_entry(me_))
break;
}
else
break;
}
int point = r->current_idx;
for(i = 0; i < me->num_nodes; i++) /* Check others */
if(i != me->nodeid && raft_node_get_match_idx(p) >= point)
votes++;
if(votes > me->num_nodes/2 && raft_get_commit_idx(me_) < point)
raft_set_commit_idx(me_, point);
/* raft periodic applies committed entries lazily */
return 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