Unverified Commit 897308f5 authored by Ozan Tezcan's avatar Ozan Tezcan Committed by GitHub
Browse files

Refactor msgid usage (#94)

parent b1a110d3
......@@ -102,17 +102,10 @@ struct raft_server {
* we're still the leader.
*/
raft_msg_id_t msg_id;
/*
* the maximum msg_id we've seen from our current leader. reset on term change
*/
raft_msg_id_t max_seen_msg_id;
raft_read_request_t *read_queue_head;
raft_read_request_t *read_queue_tail;
/* Do we need quorum ? e.g Leader received a read request, need quorum round
* before processing it */
int need_quorum_round;
raft_node_id_t node_transferring_leader_to; // the node we are targeting for leadership
long transfer_leader_time; // how long we should wait for leadership transfer to take, before aborting
int sent_timeout_now; // if we've already sent a leadership transfer signal
......@@ -174,9 +167,11 @@ int raft_is_single_node_voting_cluster(raft_server_t *me);
int raft_votes_is_majority(int nnodes, int nvotes);
void raft_node_set_last_ack(raft_node_t* me, raft_msg_id_t msgid, raft_term_t term);
void raft_node_set_match_msgid(raft_node_t *me, raft_msg_id_t msgid);
raft_msg_id_t raft_node_get_match_msgid(raft_node_t *me);
raft_msg_id_t raft_node_get_last_acked_msgid(raft_node_t* me);
void raft_node_set_next_msgid(raft_node_t *me, raft_msg_id_t msgid);
raft_msg_id_t raft_node_get_next_msgid(raft_node_t *me);
/* Heap functions */
extern void *(*raft_malloc)(size_t size);
......
......@@ -29,17 +29,16 @@ struct raft_node
raft_index_t next_idx;
raft_index_t match_idx;
raft_msg_id_t next_msgid;
raft_msg_id_t match_msgid;
raft_msg_id_t max_seen_msgid;
int flags;
raft_node_id_t id;
/* Next snapshot offset to send to this node */
raft_size_t snapshot_offset;
/* last AE heartbeat response received */
raft_term_t last_acked_term;
raft_msg_id_t last_acked_msgid;
raft_msg_id_t max_seen_msgid;
};
raft_node_t* raft_node_new(void* udata, raft_node_id_t id)
......@@ -84,6 +83,38 @@ void raft_node_set_match_idx(raft_node_t* me, raft_index_t idx)
me->match_idx = idx;
}
void raft_node_set_match_msgid(raft_node_t *me, raft_msg_id_t msgid)
{
me->match_msgid = msgid;
}
raft_msg_id_t raft_node_get_match_msgid(raft_node_t *me)
{
return me->match_msgid;
}
void raft_node_set_next_msgid(raft_node_t *me, raft_msg_id_t msgid)
{
me->next_msgid = msgid;
}
raft_msg_id_t raft_node_get_next_msgid(raft_node_t *me)
{
return me->next_msgid;
}
void raft_node_update_max_seen_msg_id(raft_node_t *me, raft_msg_id_t msg_id)
{
if (me->max_seen_msgid < msg_id) {
me->max_seen_msgid = msg_id;
}
}
raft_msg_id_t raft_node_get_max_seen_msg_id(raft_node_t *me)
{
return me->max_seen_msgid;
}
void* raft_node_get_udata(raft_node_t* me)
{
return me->udata;
......@@ -180,29 +211,6 @@ int raft_node_is_addition_committed(raft_node_t* me)
return (me->flags & RAFT_NODE_ADDITION_COMMITTED) != 0;
}
void raft_node_set_last_ack(raft_node_t* me, raft_msg_id_t msgid, raft_term_t term)
{
me->last_acked_msgid = msgid;
me->last_acked_term = term;
}
raft_msg_id_t raft_node_get_last_acked_msgid(raft_node_t* me)
{
return me->last_acked_msgid;
}
void raft_node_update_max_seen_msg_id(raft_node_t *me, raft_msg_id_t msg_id)
{
if (msg_id > me->max_seen_msgid) {
me->max_seen_msgid = msg_id;
}
}
raft_msg_id_t raft_node_get_max_seen_msg_id(raft_node_t *me)
{
return me->max_seen_msgid;
}
raft_size_t raft_node_get_snapshot_offset(raft_node_t *me)
{
return me->snapshot_offset;
......
......@@ -534,7 +534,7 @@ static raft_msg_id_t quorum_msg_id(raft_server_t* me)
if (me->node == node) {
msg_ids[num_voters++] = me->msg_id;
} else {
msg_ids[num_voters++] = raft_node_get_last_acked_msgid(node);
msg_ids[num_voters++] = raft_node_get_match_msgid(node);
}
}
......@@ -657,7 +657,7 @@ int raft_recv_appendentries_response(raft_server_t* me,
if (!raft_is_leader(me))
return RAFT_ERR_NOT_LEADER;
if (raft_node_get_last_acked_msgid(node) > r->msg_id) {
if (raft_node_get_match_msgid(node) > r->msg_id) {
// this was received out of order and is now irrelevant.
return 0;
}
......@@ -676,8 +676,9 @@ int raft_recv_appendentries_response(raft_server_t* me,
else if (me->current_term != r->term)
return 0;
// if we got here, it means that the follower has acked us as a leader, even if it cant accept the append_entry
raft_node_set_last_ack(node, r->msg_id, r->term);
/* If we got here, it means that the follower has acked us as a leader,
* even if it cant accept the append_entry */
raft_node_set_match_msgid(node, r->msg_id);
raft_index_t match_idx = raft_node_get_match_idx(node);
......@@ -762,15 +763,11 @@ int raft_recv_appendentries(raft_server_t* me,
{
int e = 0;
if (0 < ae->n_entries)
{
raft_log_node(
me, ae->leader_id,
"recvd appendentries li:%d t:%ld ci:%ld lc:%ld pli:%ld plt:%ld #%d",
ae->leader_id, ae->term, raft_get_current_idx(me),
ae->leader_commit, ae->prev_log_idx, ae->prev_log_term,
ae->n_entries);
}
raft_log_node(me, ae->leader_id,
"recv appendentries li:%d t:%ld ci:%ld lc:%ld pli:%ld plt:%ld msgid:%ld #%d",
ae->leader_id, ae->term, raft_get_current_idx(me),
ae->leader_commit, ae->prev_log_idx, ae->prev_log_term,
ae->msg_id, ae->n_entries);
r->msg_id = ae->msg_id;
r->success = 0;
......@@ -788,10 +785,6 @@ int raft_recv_appendentries(raft_server_t* me,
goto out;
}
if (node != NULL) {
raft_node_update_max_seen_msg_id(node, ae->msg_id);
}
/* update current leader because ae->term is up to date */
raft_accept_leader(me, ae->leader_id);
raft_reset_transfer_leader(me, 0);
......@@ -849,11 +842,13 @@ int raft_recv_appendentries(raft_server_t* me,
r->success = 1;
r->current_idx = ae->prev_log_idx;
/* synchronize msg_id to leader. not really needed for raft, but needed for virtraft for msg_id to be increasing
* cluster wide so that can verify read_queue correctness easily. Otherwise, it be fine for msg_id to be unique to
* each raft_server_t.
/* Synchronize msg_id to leader. Not really needed for raft, but needed for
* virtraft for msg_id to be increasing cluster wide so that can verify
* read_queue correctness easily. Otherwise, it'd be fine for msg_id to be
* unique to each raft_server_t.
*/
if (ae->msg_id > me->msg_id) {
if (me->msg_id < ae->msg_id) {
me->msg_id = ae->msg_id;
}
......@@ -916,6 +911,12 @@ int raft_recv_appendentries(raft_server_t* me,
}
out:
/* 'node' might be created after processing this message, fetching again. */
node = raft_get_node(me, ae->leader_id);
if (node != NULL) {
raft_node_update_max_seen_msg_id(node, ae->msg_id);
}
r->term = me->current_term;
if (0 == r->success)
r->current_idx = raft_get_current_idx(me);
......@@ -1320,10 +1321,6 @@ int raft_recv_snapshot(raft_server_t* me,
goto out;
}
if (node != NULL) {
raft_node_update_max_seen_msg_id(node, req->msg_id);
}
raft_accept_leader(me, req->leader_id);
raft_reset_transfer_leader(me, 0);
......@@ -1372,6 +1369,12 @@ success:
out:
resp->term = me->current_term;
/* 'node' might be created after processing this message, fetching again. */
node = raft_get_node(me, req->leader_id);
if (node != NULL) {
raft_node_update_max_seen_msg_id(node, req->msg_id);
}
return e;
}
......@@ -1392,7 +1395,7 @@ int raft_recv_snapshot_response(raft_server_t* me,
if (!raft_is_leader(me))
return RAFT_ERR_NOT_LEADER;
if (raft_node_get_last_acked_msgid(node) > r->msg_id) {
if (raft_node_get_match_msgid(node) > r->msg_id) {
return 0;
}
......@@ -1410,7 +1413,7 @@ int raft_recv_snapshot_response(raft_server_t* me,
return 0;
}
raft_node_set_last_ack(node, r->msg_id, r->term);
raft_node_set_match_msgid(node, r->msg_id);
if (!r->success) {
raft_node_set_snapshot_offset(node, r->offset);
......@@ -1497,6 +1500,7 @@ int raft_send_appendentries(raft_server_t* me, raft_node_t* node)
int res = me->cb.send_appendentries(me, me->udata, node, &ae);
if (!res) {
raft_node_set_next_idx(node, next_idx + ae.n_entries);
raft_node_set_next_msgid(node, ae.msg_id + 1);
}
raft_entry_release_list(ae.entries, ae.n_entries);
raft_free(ae.entries);
......@@ -1865,8 +1869,6 @@ int raft_queue_read_request(raft_server_t* me, raft_read_request_callback_f cb,
me->read_queue_tail->next = req;
me->read_queue_tail = req;
me->need_quorum_round = 1;
if (me->auto_flush)
return raft_flush(me, 0);
......@@ -2102,7 +2104,7 @@ int raft_set_auto_flush(raft_server_t* me, int flush)
int raft_flush(raft_server_t* me, raft_index_t sync_index)
{
if (!raft_is_leader(me)) {
return 0;
goto out;
}
if (sync_index > raft_node_get_match_idx(me->node)) {
......@@ -2114,20 +2116,20 @@ int raft_flush(raft_server_t* me, raft_index_t sync_index)
return e;
}
raft_msg_id_t last = me->read_queue_tail ? me->read_queue_tail->msg_id : 0;
for (int i = 0; i < me->num_nodes; i++)
{
if (me->node == me->nodes[i])
continue;
if (!me->need_quorum_round &&
if (raft_node_get_next_msgid(me->nodes[i]) >= last &&
raft_node_get_next_idx(me->nodes[i]) > raft_get_current_idx(me))
continue;
raft_send_appendentries(me, me->nodes[i]);
}
me->need_quorum_round = 0;
if (me->last_applied_idx < raft_get_commit_idx(me)) {
e = raft_apply_all(me);
if (e != 0) {
......@@ -2135,7 +2137,7 @@ int raft_flush(raft_server_t* me, raft_index_t sync_index)
}
}
out:
raft_process_read_queue(me);
return 0;
}
......@@ -90,7 +90,6 @@ int raft_set_current_term(raft_server_t* me, const raft_term_t term)
}
me->current_term = term;
me->voted_for = voted_for;
me->max_seen_msg_id = 0;
}
return 0;
}
......
......@@ -4056,7 +4056,7 @@ void TestRaft_quorum_msg_id_correctness(CuTest * tc)
CuAssertIntEquals(tc, 0, val);
// Second node acknowledges reaq req
raft_node_set_last_ack(raft_get_node(r, 2), 1, 1);
raft_node_set_match_msgid(raft_get_node(r, 2), 1);
raft_periodic(r, 200);
CuAssertIntEquals(tc, 1, val);
......@@ -4072,12 +4072,12 @@ void TestRaft_quorum_msg_id_correctness(CuTest * tc)
CuAssertIntEquals(tc, 0, val);
// Second node acknowledges read req,
raft_node_set_last_ack(raft_get_node(r, 2), 2, 1);
raft_node_set_match_msgid(raft_get_node(r, 2), 2);
raft_periodic(r, 200);
CuAssertIntEquals(tc, 0, val);
// Third node acknowledges read req
raft_node_set_last_ack(raft_get_node(r, 3), 2, 1);
raft_node_set_match_msgid(raft_get_node(r, 3), 2);
raft_periodic(r, 200);
CuAssertIntEquals(tc, 1, val);
}
......@@ -4186,7 +4186,7 @@ void TestRaft_leader_steps_down_if_there_is_no_quorum(CuTest * tc)
raft_periodic(r, quorum_timeout + 1);
CuAssertTrue(tc, !raft_is_leader(r));
raft_node_set_last_ack(raft_get_node(r, 2), 1, 1);
raft_node_set_match_msgid(raft_get_node(r, 2), 1);
raft_become_leader(r);
raft_periodic(r, 200);
CuAssertTrue(tc, raft_is_leader(r));
......@@ -4196,7 +4196,7 @@ void TestRaft_leader_steps_down_if_there_is_no_quorum(CuTest * tc)
CuAssertTrue(tc, raft_is_leader(r));
// If there is an ack from the follower, leader won't step down.
raft_node_set_last_ack(raft_get_node(r, 2), 2, 1);
raft_node_set_match_msgid(raft_get_node(r, 2), 2);
raft_periodic(r, quorum_timeout);
CuAssertTrue(tc, raft_is_leader(r));
......
......@@ -850,6 +850,7 @@ class RaftServer(object):
lib.raft_log_set_callbacks(lib.raft_get_log(self.raft), log_cbs, self.raft)
lib.raft_set_election_timeout(self.raft, 500)
lib.raft_set_auto_flush(self.raft, network.auto_flush)
lib.raft_set_log_enabled(self.raft, 1)
self.fsm_dict = {}
self.fsm_log = []
......
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