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

Simplify read request processing (#105)

Simplify read request processing
parent a103739b
......@@ -1538,7 +1538,7 @@ extern const raft_log_impl_t raft_log_internal_impl;
void raft_handle_append_cfg_change(raft_server_t* me, raft_entry_t* ety, raft_index_t idx);
int raft_queue_read_request(raft_server_t* me, raft_read_request_callback_f cb, void *cb_arg);
int raft_recv_read_request(raft_server_t* me, raft_read_request_callback_f cb, void *cb_arg);
/** Attempt to process read queue.
*/
......@@ -1591,7 +1591,7 @@ raft_index_t raft_get_num_snapshottable_logs(raft_server_t* me);
* HandleNetworkOperations();
*
* for (int i = 0; i < new_readreq_count; i++)
* raft_queue_read_request(raft, read_requests[i]);
* raft_recv_read_request(raft, read_requests[i]);
*
* for (int i = 0; i < new_requests_count; i++)
* raft_recv_entry(raft, new_requests[i]);
......
......@@ -48,6 +48,9 @@ struct raft_server {
/* idx of highest log entry applied to state machine */
raft_index_t last_applied_idx;
/* term of the highest log entry applied to the state machine */
raft_term_t last_applied_term;
/* follower/leader/candidate indicator */
raft_state_e state;
......
......@@ -161,6 +161,7 @@ void raft_clear(raft_server_t* me)
me->leader_id = RAFT_NODE_ID_NONE;
me->commit_idx = 0;
me->last_applied_idx = 0;
me->last_applied_term = 0;
me->num_nodes = 0;
me->node = NULL;
me->log_impl->reset(me->log, 1, 1);
......@@ -387,54 +388,52 @@ void raft_accept_leader(raft_server_t* me, raft_node_id_t leader)
int raft_become_leader(raft_server_t* me)
{
int i;
raft_log(me, "becoming leader term:%ld", raft_get_current_term(me));
raft_log(me, "becoming leader term: %ld", me->current_term);
raft_index_t next_idx = raft_get_current_idx(me) + 1;
raft_entry_t *noop = raft_entry_new(0);
noop->term = raft_get_current_term(me);
noop->type = RAFT_LOGTYPE_NO_OP;
if (raft_get_current_term(me) > 1) {
raft_entry_t *noop = raft_entry_new(0);
noop->term = raft_get_current_term(me);
noop->type = RAFT_LOGTYPE_NO_OP;
int e = raft_append_entry(me, noop);
raft_entry_release(noop);
if (e != 0) {
return e;
}
int e = raft_append_entry(me, noop);
raft_entry_release(noop);
if (0 != e)
return e;
e = me->log_impl->sync(me->log);
if (e != 0) {
return e;
}
e = me->log_impl->sync(me->log);
if (0 != e)
return e;
raft_index_t current_idx = raft_get_current_idx(me);
raft_node_set_match_idx(me->node, raft_get_current_idx(me));
me->next_sync_index = raft_get_current_idx(me) + 1;
raft_node_set_match_idx(me->node, current_idx);
me->next_sync_index = current_idx + 1;
// Commit noop immediately if this is a single node cluster
if (raft_is_single_node_voting_cluster(me)) {
raft_set_commit_idx(me, raft_get_current_idx(me));
}
/* Commit noop immediately if this is a single node cluster. */
if (raft_is_single_node_voting_cluster(me)) {
raft_set_commit_idx(me, current_idx);
}
raft_set_state(me, RAFT_STATE_LEADER);
raft_update_quorum_meta(me, me->msg_id);
raft_clear_incoming_snapshot(me, 0);
me->timeout_elapsed = 0;
raft_reset_transfer_leader(me, 0);
me->timeout_elapsed = 0;
if (me->cb.notify_state_event)
me->cb.notify_state_event(me, raft_get_udata(me), RAFT_STATE_LEADER);
if (me->cb.notify_state_event) {
me->cb.notify_state_event(me, me->udata, RAFT_STATE_LEADER);
}
for (i = 0; i < me->num_nodes; i++)
{
raft_node_t* node = me->nodes[i];
for (int i = 0; i < me->num_nodes; i++) {
raft_node_t *node = me->nodes[i];
if (me->node == node)
if (me->node == node) {
continue;
}
raft_node_set_snapshot_offset(node, 0);
raft_node_set_next_idx(node, next_idx);
raft_node_set_next_idx(node, current_idx);
raft_node_set_match_idx(node, 0);
raft_send_appendentries(me, node);
}
......@@ -1168,19 +1167,18 @@ int raft_apply_entry(raft_server_t* me)
if (me->last_applied_idx == raft_get_commit_idx(me))
return -1;
raft_index_t log_idx = me->last_applied_idx + 1;
raft_index_t idx = me->last_applied_idx + 1;
raft_entry_t* ety = raft_get_entry_from_idx(me, log_idx);
raft_entry_t* ety = raft_get_entry_from_idx(me, idx);
if (!ety)
return -1;
raft_log(me, "applying log: %ld, id: %d size: %d",
log_idx, ety->id, ety->data_len);
idx, ety->id, ety->data_len);
me->last_applied_idx++;
if (me->cb.applylog)
{
int e = me->cb.applylog(me, me->udata, ety, me->last_applied_idx);
int e = me->cb.applylog(me, me->udata, ety, idx);
assert(e == 0 || e == RAFT_ERR_SHUTDOWN);
if (RAFT_ERR_SHUTDOWN == e) {
raft_entry_release(ety);
......@@ -1188,13 +1186,16 @@ int raft_apply_entry(raft_server_t* me)
}
}
if (log_idx == me->voting_cfg_change_log_idx)
me->last_applied_idx = idx;
me->last_applied_term = ety->term;
if (idx == me->voting_cfg_change_log_idx)
me->voting_cfg_change_log_idx = -1;
if (!raft_entry_is_cfg_change(ety))
goto exit;
raft_node_id_t node_id = me->cb.get_node_id(me, raft_get_udata(me), ety, log_idx);
raft_node_id_t node_id = me->cb.get_node_id(me, me->udata, ety, idx);
raft_node_t* node = raft_get_node(me, node_id);
switch (ety->type) {
......@@ -1671,35 +1672,25 @@ raft_index_t raft_get_num_snapshottable_logs(raft_server_t *me)
int raft_begin_snapshot(raft_server_t *me)
{
if (raft_get_num_snapshottable_logs(me) == 0)
raft_index_t entry_count = raft_get_num_snapshottable_logs(me);
if (entry_count == 0) {
return -1;
raft_index_t snapshot_target = raft_get_commit_idx(me);
if (!snapshot_target)
return -1;
raft_entry_t* ety = raft_get_entry_from_idx(me, snapshot_target);
if (!ety)
return -1;
raft_term_t ety_term = ety->term;
raft_entry_release(ety);
}
/* we need to get all the way to the commit idx */
int e = raft_apply_all(me);
if (e != 0)
if (e != 0) {
return e;
}
assert(raft_get_commit_idx(me) == raft_get_last_applied_idx(me));
assert(me->commit_idx == me->last_applied_idx);
me->snapshot_in_progress = 1;
me->next_snapshot_last_idx = snapshot_target;
me->next_snapshot_last_term = ety_term;
me->next_snapshot_last_idx = me->last_applied_idx;
me->next_snapshot_last_term = me->last_applied_term;
raft_log(me,
"begin snapshot sli:%ld slt:%ld slogs:%ld",
me->snapshot_last_idx,
me->snapshot_last_term,
raft_get_num_snapshottable_logs(me));
raft_log(me, "begin snapshot lai:%ld lat:%ld ec:%ld",
me->last_applied_idx, me->last_applied_term, entry_count);
return 0;
}
......@@ -1793,6 +1784,7 @@ int raft_begin_load_snapshot(
if (raft_get_commit_idx(me) < last_included_index)
raft_set_commit_idx(me, last_included_index);
me->last_applied_term = last_included_term;
me->last_applied_idx = last_included_index;
me->next_snapshot_last_term = last_included_term;
me->next_snapshot_last_idx = last_included_index;
......@@ -1889,12 +1881,16 @@ void raft_entry_release_list(raft_entry_t **ety_list, size_t len)
raft_free(ety_list);
}
int raft_queue_read_request(raft_server_t* me, raft_read_request_callback_f cb, void *cb_arg)
int raft_recv_read_request(raft_server_t* me, raft_read_request_callback_f cb, void *cb_arg)
{
raft_read_request_t *req = raft_malloc(sizeof(raft_read_request_t));
if (!raft_is_leader(me)) {
return RAFT_ERR_NOT_LEADER;
}
raft_read_request_t *req = raft_malloc(sizeof(*req));
req->read_idx = raft_get_current_idx(me);
req->read_term = raft_get_current_term(me);
req->read_term = me->current_term;
req->msg_id = ++me->msg_id;
req->cb = cb;
req->cb_arg = cb_arg;
......@@ -1933,60 +1929,32 @@ static void pop_read_queue(raft_server_t *me, int can_read)
void raft_process_read_queue(raft_server_t* me)
{
if (!me->read_queue_head)
if (!me->read_queue_head) {
return;
}
/* As a follower we drop all queued read requests */
if (raft_is_follower(me)) {
/* If not the leader, we drop all queued read requests */
if (!raft_is_leader(me)) {
while (me->read_queue_head) {
pop_read_queue(me, 0);
}
return;
}
/* As a leader we can process requests that fulfill these conditions:
* 1) Heartbeat acknowledged by majority
* 2) We're on the same term (note: is this needed or over cautious?)
* 1) Applied NOOP entry from this term.
* 2) Heartbeat acknowledged by majority.
* 3) State machine has advanced enough.
*/
if (!raft_is_leader(me))
if (me->last_applied_term < me->current_term) {
return;
/* Quickly bail if nothing to do */
if (!me->read_queue_head)
return;
if (raft_get_num_voting_nodes(me) > 1) {
raft_entry_t *ety = raft_get_entry_from_idx(me, raft_get_commit_idx(me));
if (!ety)
return;
raft_term_t ety_term = ety->term;
raft_entry_release(ety);
/* Don't read if we did not commit an entry this term yet!
* A new term has a NO_OP committed so if everything is well
* we can except that to happen.
*/
if (ety_term < me->current_term)
return;
}
raft_msg_id_t last_acked_msgid = quorum_msg_id(me);
raft_index_t last_applied_idx = me->last_applied_idx;
/* Special case: the log's first index is 1, so we need to account
* for that in case we read before anything was ever committed.
*
* Note that this also implies a single node because adding nodes would
* bump the log and commit index.
*/
if (!me->commit_idx && !me->last_applied_idx && raft_get_current_idx(me) == 1)
last_applied_idx = 1;
while (me->read_queue_head &&
me->read_queue_head->msg_id <= last_acked_msgid &&
me->read_queue_head->read_idx <= last_applied_idx) {
pop_read_queue(me, me->read_queue_head->read_term == me->current_term);
me->read_queue_head->msg_id <= last_acked_msgid &&
me->read_queue_head->read_idx <= me->last_applied_idx) {
pop_read_queue(me, 1);
}
}
......
......@@ -234,6 +234,7 @@ raft_term_t raft_get_snapshot_last_term(raft_server_t *me)
void raft_set_snapshot_metadata(raft_server_t *me, raft_term_t term, raft_index_t idx)
{
me->last_applied_term = term;
me->last_applied_idx = idx;
me->snapshot_last_term = term;
me->snapshot_last_idx = idx;
......
......@@ -463,7 +463,7 @@ void TestRaft_user_applylog_error_propogates_to_periodic(
/* let time lapse */
CuAssertIntEquals(tc, RAFT_ERR_SHUTDOWN, raft_periodic(r, 1));
CuAssertIntEquals(tc, 1, raft_get_last_applied_idx(r));
CuAssertIntEquals(tc, 0, raft_get_last_applied_idx(r));
}
void TestRaft_server_apply_entry_increments_last_applied_idx(CuTest* tc)
......@@ -584,7 +584,7 @@ void TestRaft_server_recv_entry_auto_commits_if_we_are_the_only_node(CuTest * tc
raft_add_node(r, NULL, 1, 1);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_become_leader(r);
CuAssertTrue(tc, 0 == raft_get_commit_idx(r));
CuAssertTrue(tc, 1 == raft_get_commit_idx(r));
/* entry message */
raft_entry_req_t *ety = __MAKE_ENTRY(1, 1, "entry");
......@@ -592,8 +592,8 @@ void TestRaft_server_recv_entry_auto_commits_if_we_are_the_only_node(CuTest * tc
/* receive entry */
raft_entry_resp_t cr;
raft_recv_entry(r, ety, &cr);
CuAssertTrue(tc, 1 == raft_get_log_count(r));
CuAssertTrue(tc, 1 == raft_get_commit_idx(r));
CuAssertTrue(tc, 2 == raft_get_log_count(r));
CuAssertTrue(tc, 2 == raft_get_commit_idx(r));
}
void TestRaft_server_recv_entry_fails_if_there_is_already_a_voting_change(CuTest * tc)
......@@ -603,7 +603,7 @@ void TestRaft_server_recv_entry_fails_if_there_is_already_a_voting_change(CuTest
raft_config(r, 1, RAFT_CONFIG_AUTO_FLUSH, 0);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_become_leader(r);
CuAssertTrue(tc, 0 == raft_get_commit_idx(r));
CuAssertTrue(tc, 1 == raft_get_commit_idx(r));
/* entry message */
raft_entry_req_t *ety = __MAKE_ENTRY(1, 1, "entry");
......@@ -612,12 +612,12 @@ void TestRaft_server_recv_entry_fails_if_there_is_already_a_voting_change(CuTest
/* receive entry */
raft_entry_resp_t cr;
CuAssertTrue(tc, 0 == raft_recv_entry(r, ety, &cr));
CuAssertTrue(tc, 1 == raft_get_log_count(r));
CuAssertTrue(tc, 2 == raft_get_log_count(r));
raft_entry_t *ety2 = __MAKE_ENTRY(2, 1, "entry");
ety2->type = RAFT_LOGTYPE_ADD_NODE;
CuAssertTrue(tc, RAFT_ERR_ONE_VOTING_CHANGE_ONLY == raft_recv_entry(r, ety2, &cr));
CuAssertTrue(tc, 0 == raft_get_commit_idx(r));
CuAssertTrue(tc, 1 == raft_get_commit_idx(r));
}
void TestRaft_server_cfg_sets_num_nodes(CuTest * tc)
......@@ -2517,6 +2517,10 @@ void TestRaft_non_leader_recv_entry_msg_fails(CuTest * tc)
/* receive entry */
int e = raft_recv_entry(r, ety, &cr);
CuAssertTrue(tc, RAFT_ERR_NOT_LEADER == e);
/* receive read request */
e = raft_recv_read_request(r, NULL, NULL);
CuAssertTrue(tc, RAFT_ERR_NOT_LEADER == e);
}
/* 5.3 */
......@@ -3895,45 +3899,41 @@ void TestRaft_read_action_callback(
{
void *r = raft_new();
struct read_request_arg ra = { 0 };
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_add_node(r, NULL, 3, 0);
raft_set_current_term(r, 1);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_become_leader(r);
__RAFT_APPEND_ENTRY(r, 1, 1, "aaa");
raft_set_commit_idx(r, 1);
raft_queue_read_request(r, __read_request_callback, &ra);
raft_recv_read_request(r, __read_request_callback, &ra);
/* not acked yet */
raft_periodic(r, 1);
CuAssertIntEquals(tc, 0, ra.calls);
/* acked by node 2 - enough for quorum */
raft_appendentries_resp_t aer = { .msg_id = 1, .term = 1, .success = 1, .current_idx = 1 };
raft_appendentries_resp_t aer = { .msg_id = 1, .term = 1, .success = 1, .current_idx = 2 };
CuAssertIntEquals(tc, 0, raft_recv_appendentries_response(r, raft_get_node(r, 2), &aer));
raft_periodic(r, 1);
CuAssertIntEquals(tc, 1, ra.calls);
CuAssertIntEquals(tc, 1, ra.last_cb_safe);
/* make sure read request is called only once */
raft_periodic(r, 1);
CuAssertIntEquals(tc, 1, ra.calls);
/* entry 2 */
__RAFT_APPEND_ENTRY(r, 2, 1, "aaa");
ra.calls = 0;
raft_queue_read_request(r, __read_request_callback, &ra);
raft_recv_read_request(r, __read_request_callback, &ra);
/* election started, nothing should be read */
raft_become_candidate(r);
raft_periodic(r, 1);
CuAssertIntEquals(tc, 0, ra.calls);
CuAssertIntEquals(tc, 0, ra.last_cb_safe);
CuAssertIntEquals(tc, 1, ra.calls);
/* we win, but for safety we will not process read requests
* from past terms */
......@@ -3941,21 +3941,23 @@ void TestRaft_read_action_callback(
aer.msg_id = 2;
aer.term = 3;
CuAssertIntEquals(tc, 0, raft_recv_appendentries_response(r, raft_get_node(r, 2), &aer));
raft_periodic(r, 1);
CuAssertIntEquals(tc, 1, ra.calls);
CuAssertIntEquals(tc, 0, ra.last_cb_safe);
/* entry 3 */
__RAFT_APPEND_ENTRY(r, 3, 1, "aaa");
ra.calls = 0;
raft_queue_read_request(r, __read_request_callback, &ra);
int err = raft_recv_read_request(r, __read_request_callback, &ra);
CuAssertIntEquals(tc, RAFT_ERR_NOT_LEADER, err);
raft_become_leader(r);
err = raft_recv_read_request(r, __read_request_callback, &ra);
CuAssertIntEquals(tc, 0, err);
/* elections again, we lose */
raft_become_candidate(r);
raft_become_follower(r);
/* queued read should fire back with can_read==false */
raft_periodic(r, 1);
CuAssertIntEquals(tc, 1, ra.calls);
......@@ -3978,7 +3980,7 @@ void TestRaft_single_node_commits_noop(CuTest * tc)
raft_set_current_term(r, 2);
raft_set_commit_idx(r, 0);
raft_periodic(r, 500);
raft_queue_read_request(r, single_node_commits_noop_cb, &str);
raft_recv_read_request(r, single_node_commits_noop_cb, &str);
raft_periodic(r, 500);
CuAssertIntEquals(tc, 1, raft_get_commit_idx(r));
......@@ -4066,10 +4068,10 @@ void TestRaft_quorum_msg_id_correctness(CuTest * tc)
raft_become_leader(r);
__RAFT_APPEND_ENTRY(r, 1, 1, "aaa");
raft_set_commit_idx(r, 1);
raft_set_commit_idx(r, 2);
raft_periodic(r, 100);
raft_queue_read_request(r, quorum_msg_id_correctness_cb, &val);
raft_recv_read_request(r, quorum_msg_id_correctness_cb, &val);
raft_periodic(r, 200);
// Read request is pending as it requires two acks
......@@ -4085,7 +4087,7 @@ void TestRaft_quorum_msg_id_correctness(CuTest * tc)
raft_add_node(r, NULL, 3, 0);
raft_add_node(r, NULL, 4, 0);
raft_periodic(r, 100);
raft_queue_read_request(r, quorum_msg_id_correctness_cb, &val);
raft_recv_read_request(r, quorum_msg_id_correctness_cb, &val);
raft_periodic(r, 200);
// Read request is pending as it requires three acks
......@@ -4719,7 +4721,7 @@ void TestRaft_flush_sends_msg(CuTest *tc)
raft_set_current_term(r, 1);
raft_become_leader(r);
raft_queue_read_request(r, NULL, NULL);
raft_recv_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. */
......
......@@ -413,7 +413,7 @@ class Network(object):
if lib.raft_is_leader(sv.raft):
msg_id = lib.raft_get_msg_id(sv.raft) + 1
arg = sv.id * net.rqm + msg_id
lib.raft_queue_read_request(sv.raft, sv.handle_read_queue, ffi.cast("void *", arg))
lib.raft_recv_read_request(sv.raft, sv.handle_read_queue, ffi.cast("void *", arg))
def id2server(self, id):
for server in self.servers:
......@@ -666,7 +666,7 @@ class Network(object):
e = server.recv_entry(ety)
assert e == 0
lib.raft_set_commit_idx(server.raft, 1)
lib.raft_set_commit_idx(server.raft, 2)
e = lib.raft_apply_all(server.raft)
assert e == 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