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

Fix next_idx and next_msgid usages (#104)

Fix next_idx and next_msgid usages
parent 6f03a5c7
......@@ -673,16 +673,11 @@ 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_match_msgid(node, r->msg_id);
raft_index_t match_idx = raft_node_get_match_idx(node);
if (0 == r->success)
{
/* Stale response -- ignore */
if (r->current_idx < match_idx)
if (r->current_idx < raft_node_get_match_idx(node))
return 0;
raft_index_t next = min(r->current_idx + 1, raft_get_current_idx(me));
......@@ -715,13 +710,17 @@ int raft_recv_appendentries_response(raft_server_t* me,
raft_node_set_has_sufficient_logs(node);
}
if (r->current_idx <= match_idx)
return 0;
raft_index_t match_idx = raft_node_get_match_idx(node);
if (r->current_idx > match_idx) {
raft_node_set_match_idx(node, r->current_idx);
}
assert(r->current_idx <= raft_get_current_idx(me));
raft_node_set_next_idx(node, r->current_idx + 1);
raft_node_set_match_idx(node, r->current_idx);
raft_msg_id_t match_msgid = raft_node_get_match_msgid(node);
if (r->msg_id > match_msgid) {
raft_node_set_match_msgid(node, r->msg_id);
}
assert(r->msg_id <= me->msg_id);
if (me->auto_flush)
return raft_flush(me, 0);
......@@ -2150,7 +2149,7 @@ int raft_flush(raft_server_t* me, raft_index_t sync_index)
if (me->node == me->nodes[i])
continue;
if (raft_node_get_next_msgid(me->nodes[i]) >= last &&
if (raft_node_get_next_msgid(me->nodes[i]) > last &&
raft_node_get_next_idx(me->nodes[i]) > raft_get_current_idx(me))
continue;
......
......@@ -4172,10 +4172,12 @@ void TestRaft_callback_timeoutnow_at_send_appendentries_response_if_up_to_date(C
CuAssertIntEquals(tc, 0, ret);
CuAssertTrue(tc, 0 == timeoutnow_sent);
raft_appendentries_resp_t aer;
aer.term = 2;
aer.success = 1;
aer.current_idx = 2;
raft_appendentries_resp_t aer = {
.term = 2,
.success = 1,
.current_idx = 2
};
raft_recv_appendentries_response(r, raft_get_node(r, 2), &aer);
CuAssertTrue(tc, 1 == timeoutnow_sent);
}
......@@ -4688,6 +4690,89 @@ void TestRaft_limit_appendentries_size(CuTest *tc)
CuAssertIntEquals(tc, 20, appendentries_msg_count);
}
static int cb_sendmsg(raft_server_t *raft, void *udata, raft_node_t *node,
raft_appendentries_req_t *msg)
{
CuTest *tc = udata;
int *appendentries_msg_count = raft_node_get_udata(node);
(*appendentries_msg_count)++;
return 0;
}
void TestRaft_flush_sends_msg(CuTest *tc)
{
raft_cbs_t funcs = {
.send_appendentries = cb_sendmsg,
};
int appendentries_msg_count = 0;
raft_server_t *r = raft_new();
raft_set_callbacks(r, &funcs, tc);
raft_config(r, 1, RAFT_CONFIG_AUTO_FLUSH, 0);
raft_add_node(r, NULL, 100, 1);
raft_node_t *node = raft_add_node(r, NULL, 2, 0);
raft_node_set_udata(node, &appendentries_msg_count);
raft_set_current_term(r, 1);
raft_become_leader(r);
raft_queue_read_request(r, NULL, NULL);
/* Verify that we send appendentries if the next msgid of a node equals
* to the last read request's msgid. */
raft_node_set_match_msgid(node, r->msg_id - 1);
raft_node_set_next_msgid(node, r->msg_id);
int msg_send = appendentries_msg_count;
raft_flush(r, 0);
CuAssertIntEquals(tc, msg_send + 1, appendentries_msg_count);
}
void TestRaft_recv_appendentries_does_not_change_next_idx(CuTest *tc)
{
raft_cbs_t funcs = {
.send_appendentries = cb_send_ae,
.get_entries_to_send = cb_get_entries_to_send
};
int appendentries_msg_count = 0;
raft_server_t *r = raft_new();
raft_set_callbacks(r, &funcs, tc);
raft_add_node(r, NULL, 100, 1);
raft_node_t *node = raft_add_node(r, NULL, 2, 0);
raft_node_set_udata(node, &appendentries_msg_count);
raft_set_current_term(r, 1);
/* Append 200 entries */
__RAFT_APPEND_ENTRIES_SEQ_ID(r, 200, 0, 1, "test");
/* 20 appendentries message will be sent. Each message will contain
* 10 entries. */
raft_send_appendentries_all(r);
CuAssertIntEquals(tc, 20, appendentries_msg_count);
CuAssertIntEquals(tc, 201, raft_node_get_next_idx(node));
CuAssertIntEquals(tc, 21, raft_node_get_next_msgid(node));
raft_appendentries_resp_t resp = {
.success = 1,
.current_idx = 20,
.msg_id = 5,
.term = 1
};
raft_recv_appendentries_response(r, node, &resp);
CuAssertIntEquals(tc, 201, raft_node_get_next_idx(node));
CuAssertIntEquals(tc, 21, raft_node_get_next_msgid(node));
}
int main(void)
{
CuString *output = CuStringNew();
......@@ -4834,6 +4919,8 @@ int main(void)
SUITE_ADD_TEST(suite, Test_transfer_automatic);
SUITE_ADD_TEST(suite, TestRaft_config);
SUITE_ADD_TEST(suite, TestRaft_limit_appendentries_size);
SUITE_ADD_TEST(suite, TestRaft_flush_sends_msg);
SUITE_ADD_TEST(suite, TestRaft_recv_appendentries_does_not_change_next_idx);
CuSuiteRun(suite);
CuSuiteDetails(suite, output);
printf("%s\n", output->buffer);
......
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