Unverified Commit 59770f6c authored by Ozan Tezcan's avatar Ozan Tezcan Committed by GitHub
Browse files

Fix: read operations will count acks correctly (#34)

parent 04cdff03
...@@ -1582,34 +1582,41 @@ void raft_entry_release_list(raft_entry_t **ety_list, size_t len) ...@@ -1582,34 +1582,41 @@ void raft_entry_release_list(raft_entry_t **ety_list, size_t len)
static int msgid_cmp(const void *a, const void *b) static int msgid_cmp(const void *a, const void *b)
{ {
return *(raft_msg_id_t *) a - *(raft_msg_id_t *) b; raft_msg_id_t va = *((raft_msg_id_t*) a);
raft_msg_id_t vb = *((raft_msg_id_t*) b);
return va > vb ? -1 : 1;
} }
raft_msg_id_t quorum_msg_id(raft_server_t* me_) raft_msg_id_t quorum_msg_id(raft_server_t* me_)
{ {
raft_server_private_t* me = (raft_server_private_t*) me_; raft_server_private_t* me = (raft_server_private_t*) me_;
int i;
raft_msg_id_t msg_ids[me->num_nodes]; raft_msg_id_t msg_ids[me->num_nodes];
int msg_ids_count = 0; int num_voters = 0;
for (i = 0; i < me->num_nodes; i++) { for (int i = 0; i < me->num_nodes; i++) {
raft_node_t* node = me->nodes[i]; raft_node_t* node = me->nodes[i];
if (!raft_node_is_voting(node)) if (!raft_node_is_voting(node))
continue; continue;
if (me->node == node) { if (me->node == node) {
msg_ids[msg_ids_count++] = me->msg_id; msg_ids[num_voters++] = me->msg_id;
} else { } else {
msg_ids[msg_ids_count++] = raft_node_get_last_acked_msgid(node); msg_ids[num_voters++] = raft_node_get_last_acked_msgid(node);
} }
} }
qsort(msg_ids, msg_ids_count, sizeof(raft_msg_id_t), msgid_cmp); assert(num_voters == raft_get_num_voting_nodes(me_));
int num_voting_nodes = raft_get_num_voting_nodes(me_);
assert(num_voting_nodes == msg_ids_count); /**
return msg_ids[num_voting_nodes - (num_voting_nodes / 2 + num_voting_nodes % 2)]; * Sort the acknowledged msg_ids in the descending order and return
* the median value. Median value means it's the highest msg_id
* acknowledged by the majority.
*/
qsort(msg_ids, num_voters, sizeof(raft_msg_id_t), msgid_cmp);
return msg_ids[num_voters / 2];
} }
void raft_queue_read_request(raft_server_t* me_, func_read_request_callback_f cb, void *cb_arg) void raft_queue_read_request(raft_server_t* me_, func_read_request_callback_f cb, void *cb_arg)
......
...@@ -60,7 +60,7 @@ int raft_get_num_voting_nodes(raft_server_t* me_) ...@@ -60,7 +60,7 @@ int raft_get_num_voting_nodes(raft_server_t* me_)
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
int i, num = 0; int i, num = 0;
for (i = 0; i < me->num_nodes; i++) for (i = 0; i < me->num_nodes; i++)
if (raft_node_is_active(me->nodes[i]) && raft_node_is_voting(me->nodes[i])) if (raft_node_is_voting(me->nodes[i]))
num++; num++;
return num; return num;
} }
......
...@@ -3838,6 +3838,60 @@ void TestRaft_single_node_commits_noop(CuTest * tc) ...@@ -3838,6 +3838,60 @@ void TestRaft_single_node_commits_noop(CuTest * tc)
CuAssertStrEquals(tc, "success", str); CuAssertStrEquals(tc, "success", str);
} }
void quorum_msg_id_correctness_cb(void* arg, int can_read)
{
(void) can_read;
(*(int*) arg)++;
}
void TestRaft_quorum_msg_id_correctness(CuTest * tc)
{
int val = 0;
void *r = raft_new();
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_set_current_term(r, 1);
raft_become_leader(r);
__RAFT_APPEND_ENTRY(r, 1, 1, "aaa");
raft_set_commit_idx(r, 1);
raft_periodic(r, 1000);
raft_queue_read_request(r, quorum_msg_id_correctness_cb, &val);
raft_periodic(r, 2000);
// Read request is pending as it requires two acks
CuAssertIntEquals(tc, 0, val);
// Second node acknowledges reaq req
raft_node_set_last_ack(raft_get_node(r, 2), 1, 1);
raft_periodic(r, 2000);
CuAssertIntEquals(tc, 1, val);
val = 0;
raft_add_node(r, NULL, 3, 0);
raft_add_node(r, NULL, 4, 0);
raft_periodic(r, 1000);
raft_queue_read_request(r, quorum_msg_id_correctness_cb, &val);
raft_periodic(r, 2000);
// Read request is pending as it requires three acks
CuAssertIntEquals(tc, 0, val);
// Second node acknowledges reaq req,
raft_node_set_last_ack(raft_get_node(r, 2), 2, 1);
raft_periodic(r, 2000);
CuAssertIntEquals(tc, 0, val);
// Third node acknowledges reaq req
raft_node_set_last_ack(raft_get_node(r, 3), 2, 1);
raft_periodic(r, 2000);
CuAssertIntEquals(tc, 1, val);
}
int main(void) int main(void)
{ {
CuString *output = CuStringNew(); CuString *output = CuStringNew();
...@@ -3962,6 +4016,7 @@ int main(void) ...@@ -3962,6 +4016,7 @@ int main(void)
SUITE_ADD_TEST(suite, TestRaft_leader_recv_appendentries_response_set_has_sufficient_logs_after_voting_committed); SUITE_ADD_TEST(suite, TestRaft_leader_recv_appendentries_response_set_has_sufficient_logs_after_voting_committed);
SUITE_ADD_TEST(suite, TestRaft_read_action_callback); SUITE_ADD_TEST(suite, TestRaft_read_action_callback);
SUITE_ADD_TEST(suite, TestRaft_single_node_commits_noop); SUITE_ADD_TEST(suite, TestRaft_single_node_commits_noop);
SUITE_ADD_TEST(suite, TestRaft_quorum_msg_id_correctness);
CuSuiteRun(suite); CuSuiteRun(suite);
CuSuiteDetails(suite, output); CuSuiteDetails(suite, output);
......
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