Commit 84362fa7 authored by Yossi Gottlieb's avatar Yossi Gottlieb
Browse files

Read request handling support.

This makes it possible to request a consistent read without having to
commit it as a log operation.

For every read request an AE heartbeat is delivered and a quorum
response is expected.
parent b20cb85c
...@@ -163,6 +163,9 @@ typedef struct ...@@ -163,6 +163,9 @@ typedef struct
* This message could force a leader/candidate to become a follower. */ * This message could force a leader/candidate to become a follower. */
typedef struct typedef struct
{ {
/** id, to make it possible to associate responses with requests. */
raft_msg_id_t msg_id;
/** currentTerm, to force other leader/candidate to step down */ /** currentTerm, to force other leader/candidate to step down */
raft_term_t term; raft_term_t term;
...@@ -190,6 +193,9 @@ typedef struct ...@@ -190,6 +193,9 @@ typedef struct
* This message could force a leader/candidate to become a follower. */ * This message could force a leader/candidate to become a follower. */
typedef struct typedef struct
{ {
/** the msg_id this response refers to */
unsigned long msg_id;
/** currentTerm, to force other leader/candidate to step down */ /** currentTerm, to force other leader/candidate to step down */
raft_term_t term; raft_term_t term;
...@@ -424,6 +430,20 @@ typedef void ( ...@@ -424,6 +430,20 @@ typedef void (
raft_index_t entry_idx raft_index_t entry_idx
); );
/** A callback used to notify when queued read requests can be processed.
*
* @param[in] arg Argument passed in the original call.
* @param[in] can_read If non-zero, the read requests may be processed and
* returned to the user. Otherwise the request should be treated as if
* arriving to a non leader.
*/
typedef void (
*func_read_request_callback_f
) (
void *arg,
int can_read
);
/** Generic Raft Log implementation. /** Generic Raft Log implementation.
* *
* This is an abstract interface that can be used to implement pluggable * This is an abstract interface that can be used to implement pluggable
...@@ -1166,4 +1186,6 @@ extern const raft_log_impl_t raft_log_internal_impl; ...@@ -1166,4 +1186,6 @@ 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); void raft_handle_append_cfg_change(raft_server_t* me_, raft_entry_t* ety, raft_index_t idx);
void raft_queue_read_request(raft_server_t* me_, func_read_request_callback_f cb, void *cb_arg);
#endif /* RAFT_H_ */ #endif /* RAFT_H_ */
...@@ -21,6 +21,17 @@ enum { ...@@ -21,6 +21,17 @@ enum {
struct raft_log_impl; struct raft_log_impl;
typedef struct raft_read_request {
raft_index_t read_idx;
raft_term_t read_term;
raft_msg_id_t msg_id;
func_read_request_callback_f cb;
void *cb_arg;
struct raft_read_request *next;
} raft_read_request_t;
typedef struct { typedef struct {
/* Persistent state: */ /* Persistent state: */
...@@ -87,6 +98,13 @@ typedef struct { ...@@ -87,6 +98,13 @@ typedef struct {
*/ */
raft_index_t saved_snapshot_last_idx; raft_index_t saved_snapshot_last_idx;
raft_term_t saved_snapshot_last_term; raft_term_t saved_snapshot_last_term;
/* Read requests that await a network round trip to confirm
* we're still the leader.
*/
raft_msg_id_t msg_id;
raft_read_request_t *read_queue_head;
raft_read_request_t *read_queue_tail;
} raft_server_private_t; } raft_server_private_t;
int raft_election_start(raft_server_t* me); int raft_election_start(raft_server_t* me);
...@@ -150,6 +168,14 @@ void raft_node_set_voting_committed(raft_node_t* me_, int voting); ...@@ -150,6 +168,14 @@ void raft_node_set_voting_committed(raft_node_t* me_, int voting);
void raft_node_set_addition_committed(raft_node_t* me_, int committed); void raft_node_set_addition_committed(raft_node_t* me_, int committed);
void raft_node_set_last_ack(raft_node_t* me_, raft_msg_id_t msgid, raft_term_t term);
raft_msg_id_t raft_node_get_last_acked_msgid(raft_node_t* me_);
raft_term_t raft_node_get_last_acked_term(raft_node_t* me_);
void raft_process_read_queue(raft_server_t* me_);
/* Heap functions */ /* Heap functions */
extern void *(*__raft_malloc)(size_t size); extern void *(*__raft_malloc)(size_t size);
extern void *(*__raft_calloc)(size_t nmemb, size_t size); extern void *(*__raft_calloc)(size_t nmemb, size_t size);
......
...@@ -25,4 +25,9 @@ typedef long int raft_index_t; ...@@ -25,4 +25,9 @@ typedef long int raft_index_t;
*/ */
typedef int raft_node_id_t; typedef int raft_node_id_t;
/**
* Unique message identifier.
*/
typedef unsigned long raft_msg_id_t;
#endif /* RAFT_DEFS_H_ */ #endif /* RAFT_DEFS_H_ */
...@@ -34,6 +34,10 @@ typedef struct ...@@ -34,6 +34,10 @@ typedef struct
int flags; int flags;
raft_node_id_t id; raft_node_id_t id;
/* last AE heartbeat response received */
raft_term_t last_acked_term;
raft_msg_id_t last_acked_msgid;
} raft_node_private_t; } raft_node_private_t;
raft_node_t* raft_node_new(void* udata, raft_node_id_t id) raft_node_t* raft_node_new(void* udata, raft_node_id_t id)
...@@ -190,3 +194,22 @@ int raft_node_is_addition_committed(raft_node_t* me_) ...@@ -190,3 +194,22 @@ int raft_node_is_addition_committed(raft_node_t* me_)
raft_node_private_t* me = (raft_node_private_t*)me_; raft_node_private_t* me = (raft_node_private_t*)me_;
return (me->flags & RAFT_NODE_ADDITION_COMMITTED) != 0; 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)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
me->last_acked_msgid = msgid;
me->last_acked_term = term;
}
raft_msg_id_t raft_node_get_last_acked_msgid(raft_node_t* me_)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return me->last_acked_msgid;
}
raft_term_t raft_node_get_last_acked_term(raft_node_t* me_)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return me->last_acked_term;
}
...@@ -264,6 +264,8 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period) ...@@ -264,6 +264,8 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period)
return e; return e;
} }
raft_process_read_queue(me_);
return 0; return 0;
} }
...@@ -285,10 +287,10 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -285,10 +287,10 @@ int raft_recv_appendentries_response(raft_server_t* me_,
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
__log(me_, node, __log(me_, node,
"received appendentries response %s ci:%d rci:%d", "received appendentries response %s ci:%d rci:%d msgid:%lu",
r->success == 1 ? "SUCCESS" : "fail", r->success == 1 ? "SUCCESS" : "fail",
raft_get_current_idx(me_), raft_get_current_idx(me_),
r->current_idx); r->current_idx, r->msg_id);
if (!node) if (!node)
return -1; return -1;
...@@ -345,6 +347,8 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -345,6 +347,8 @@ int raft_recv_appendentries_response(raft_server_t* me_,
raft_node_set_has_sufficient_logs(node); raft_node_set_has_sufficient_logs(node);
} }
raft_node_set_last_ack(node, r->msg_id, r->term);
if (r->current_idx <= match_idx) if (r->current_idx <= match_idx)
return 0; return 0;
...@@ -408,6 +412,7 @@ int raft_recv_appendentries( ...@@ -408,6 +412,7 @@ int raft_recv_appendentries(
ae->prev_log_term, ae->prev_log_term,
ae->n_entries); ae->n_entries);
r->msg_id = ae->msg_id;
r->success = 0; r->success = 0;
if (raft_is_candidate(me_) && me->current_term == ae->term) if (raft_is_candidate(me_) && me->current_term == ae->term)
...@@ -838,7 +843,7 @@ int raft_apply_entry(raft_server_t* me_) ...@@ -838,7 +843,7 @@ int raft_apply_entry(raft_server_t* me_)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
if (!raft_is_apply_allowed(me_)) if (!raft_is_apply_allowed(me_))
return -1; return -1;
/* Don't apply after the commit_idx */ /* Don't apply after the commit_idx */
...@@ -944,6 +949,7 @@ int raft_send_appendentries(raft_server_t* me_, raft_node_t* node) ...@@ -944,6 +949,7 @@ int raft_send_appendentries(raft_server_t* me_, raft_node_t* node)
ae.leader_commit = raft_get_commit_idx(me_); ae.leader_commit = raft_get_commit_idx(me_);
ae.prev_log_idx = 0; ae.prev_log_idx = 0;
ae.prev_log_term = 0; ae.prev_log_term = 0;
ae.msg_id = me->msg_id;
raft_index_t next_idx = raft_node_get_next_idx(node); raft_index_t next_idx = raft_node_get_next_idx(node);
...@@ -976,13 +982,14 @@ int raft_send_appendentries(raft_server_t* me_, raft_node_t* node) ...@@ -976,13 +982,14 @@ int raft_send_appendentries(raft_server_t* me_, raft_node_t* node)
} }
} }
__log(me_, node, "sending appendentries: ci:%lu comi:%lu t:%lu lc:%lu pli:%lu plt:%lu #%d", __log(me_, node, "sending appendentries: ci:%lu comi:%lu t:%lu lc:%lu pli:%lu plt:%lu msgid:%lu #%d",
raft_get_current_idx(me_), raft_get_current_idx(me_),
raft_get_commit_idx(me_), raft_get_commit_idx(me_),
ae.term, ae.term,
ae.leader_commit, ae.leader_commit,
ae.prev_log_idx, ae.prev_log_idx,
ae.prev_log_term, ae.prev_log_term,
ae.msg_id,
ae.n_entries); ae.n_entries);
int res = me->cb.send_appendentries(me_, me->udata, node, &ae); int res = me->cb.send_appendentries(me_, me->udata, node, &ae);
...@@ -1553,3 +1560,105 @@ void raft_entry_release_list(raft_entry_t **ety_list, size_t len) ...@@ -1553,3 +1560,105 @@ void raft_entry_release_list(raft_entry_t **ety_list, size_t len)
raft_entry_release(ety_list[i]); raft_entry_release(ety_list[i]);
} }
} }
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 quorum_msg_id(raft_server_t* me_)
{
raft_server_private_t* me = (raft_server_private_t*) me_;
int i;
raft_msg_id_t msg_ids[me->num_nodes];
int msg_ids_count = 0;
for (i = 0; i < me->num_nodes; i++) {
raft_node_t* node = me->nodes[i];
if (me->node == node) {
msg_ids[msg_ids_count++] = me->msg_id;
continue;
}
if (!raft_node_is_active(node) || !raft_node_is_voting(node))
continue;
msg_ids[msg_ids_count++] = raft_node_get_last_acked_msgid(node);
}
qsort(msg_ids, msg_ids_count, sizeof(raft_msg_id_t), msgid_cmp);
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)];
}
void raft_queue_read_request(raft_server_t* me_, func_read_request_callback_f cb, void *cb_arg)
{
raft_server_private_t* me = (raft_server_private_t*) me_;
raft_read_request_t *req = __raft_malloc(sizeof(raft_read_request_t));
req->read_idx = raft_get_commit_idx(me_);
req->read_term = raft_get_current_term(me_);
req->msg_id = ++me->msg_id;
req->cb = cb;
req->cb_arg = cb_arg;
req->next = NULL;
if (!me->read_queue_head)
me->read_queue_head = req;
if (me->read_queue_tail)
me->read_queue_tail->next = req;
me->read_queue_tail = req;
raft_send_appendentries_all(me_);
}
static void pop_read_queue(raft_server_private_t *me, int can_read)
{
raft_read_request_t *p = me->read_queue_head;
p->cb(p->cb_arg, can_read);
/* remove entry and update head/tail */
if (p->next) {
me->read_queue_head = p->next;
if (!me->read_queue_head->next)
me->read_queue_tail = me->read_queue_head;
} else {
me->read_queue_head = NULL;
me->read_queue_tail = NULL;
}
__raft_free(p);
}
void raft_process_read_queue(raft_server_t* me_)
{
raft_server_private_t* me = (raft_server_private_t*) me_;
if (!me->read_queue_head)
return;
/* As a follower we drop all queued read requests */
if (raft_is_follower(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?)
*/
if (!raft_is_leader(me_))
return;
raft_msg_id_t acked_msgid = quorum_msg_id(me_);
while (me->read_queue_head &&
me->read_queue_head->msg_id <= acked_msgid &&
me->read_queue_head->read_idx <= me->last_applied_idx) {
pop_read_queue(me, me->read_queue_head->read_term == me->current_term);
}
}
...@@ -3634,3 +3634,86 @@ void TestRaft_leader_recv_appendentries_response_set_has_sufficient_logs_after_v ...@@ -3634,3 +3634,86 @@ void TestRaft_leader_recv_appendentries_response_set_has_sufficient_logs_after_v
raft_recv_appendentries_response(r, raft_get_node(r, 2), &aer); raft_recv_appendentries_response(r, raft_get_node(r, 2), &aer);
CuAssertIntEquals(tc, 2, has_sufficient_logs_flag); CuAssertIntEquals(tc, 2, has_sufficient_logs_flag);
} }
struct read_request_arg {
int calls;
int last_cb_safe;
};
static void __read_request_callback(void *arg, int safe)
{
struct read_request_arg *a = (struct read_request_arg *) arg;
a->calls++;
a->last_cb_safe = safe;
}
void TestRaft_read_action_callback(
CuTest * tc)
{
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_set_election_timeout(r, 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);
/* not acked yet */
raft_periodic(r, 1);
CuAssertIntEquals(tc, 0, ra.calls);
/* acked by node 2 - enough for quorum */
msg_appendentries_response_t aer = { .msg_id = 1, .term = 1, .success = 1, .current_idx = 1 };
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);
/* election started, nothing should be read */
raft_become_candidate(r);
raft_periodic(r, 1);
CuAssertIntEquals(tc, 0, ra.calls);
/* we win, but for safety we will not process read requests
* from past terms */
raft_become_leader(r);
aer.msg_id = 2;
aer.term = 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, 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);
/* 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);
CuAssertIntEquals(tc, 0, ra.last_cb_safe);
}
...@@ -877,7 +877,6 @@ class RaftServer(object): ...@@ -877,7 +877,6 @@ class RaftServer(object):
response.success = 1 response.success = 1
response.current_idx = snapshot.last_idx response.current_idx = snapshot.last_idx
response.term = lib.raft_get_current_term(self.raft) response.term = lib.raft_get_current_term(self.raft)
response.first_idx = response.current_idx
self.network.enqueue_msg(response, self, other) self.network.enqueue_msg(response, self, other)
node_id = lib.raft_get_nodeid(self.raft) node_id = lib.raft_get_nodeid(self.raft)
......
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