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

Add support to limit appendentries message size (#101)

parent bc37037b
......@@ -264,7 +264,7 @@ typedef struct
raft_index_t leader_commit;
/** number of entries within this message */
int n_entries;
raft_index_t n_entries;
/** array of pointers to entries within this message */
raft_entry_req_t** entries;
......@@ -638,6 +638,35 @@ typedef int (
raft_node_t* node
);
/** Callback for fetching entries to send in a appendentries message.
*
* This callback is useful when you want to limit appendentries message size.
* Application is supposed to fill the `entries` array by using
* raft_get_entry_from_idx() and raft_get_entries_from_idx() functions. If the
* application wants to limit the appendentries message size, it can fill the
* array partially. As this callback is inside a loop, the remaining entries
* will be fetched and sent as another append entries message in the next
* callback.
*
* @param[in] raft The Raft server making this callback.
* @param[in] user_data User data that is passed from Raft server.
* @param[in] node The node that we are sending this message to.
* @param[in] idx Index of first entry to fetch.
* @param[in] entries_n Length of entries (max. entries to fetch).
* @param[out] entries An initialized array of raft_entry_t*.
* @return Number of entries fetched
*/
typedef raft_index_t (
*raft_get_entries_to_send_f
) (
raft_server_t *raft,
void *user_data,
raft_node_t *node,
raft_index_t idx,
raft_index_t entries_n,
raft_entry_t **entries
);
typedef struct
{
/** Callback for sending request vote messages */
......@@ -703,6 +732,9 @@ typedef struct
/** Callback for deciding whether to send raft_appendentries_req to a node. */
raft_backpressure_f backpressure;
/** Callback for preparing entries to send in a raft_appendentries_req */
raft_get_entries_to_send_f get_entries_to_send;
} raft_cbs_t;
/** A generic notification callback used to allow Raft to notify caller
......@@ -861,8 +893,8 @@ typedef struct raft_log_impl
* Caller must use raft_entry_release_list() when no longer requiring
* the returned entries.
*/
int (*get_batch) (void *log, raft_index_t idx, int entries_n,
raft_entry_t **entries);
raft_index_t (*get_batch) (void *log, raft_index_t idx,
raft_index_t entries_n, raft_entry_t **entries);
/** Get first entry's index.
* @return
......@@ -1144,6 +1176,14 @@ void raft_node_set_next_idx(raft_node_t* me, raft_index_t idx);
* @return entry from index */
raft_entry_t* raft_get_entry_from_idx(raft_server_t* me, raft_index_t idx);
/**
* @param[in] idx The entry's index
* @param[out] n_etys Number of returned entries
* @return entry batch from index. Caller must use raft_entry_release_list(). */
raft_entry_t** raft_get_entries_from_idx(raft_server_t* me,
raft_index_t idx,
raft_index_t* n_etys);
/**
* @param[in] node The node's ID
* @return node pointed to by node ID */
......
......@@ -357,9 +357,12 @@ static raft_entry_t *log_get(void *log, raft_index_t idx)
return e;
}
static int log_get_batch(void *log, raft_index_t idx, int entries_n, raft_entry_t **entries)
static raft_index_t log_get_batch(void *log,
raft_index_t idx,
raft_index_t entries_n,
raft_entry_t **entries)
{
long n, i;
raft_index_t n;
raft_entry_t **r = raft_log_get_from_idx(log, idx, &n);
if (!r || n < 1) {
......@@ -369,11 +372,11 @@ static int log_get_batch(void *log, raft_index_t idx, int entries_n, raft_entry_
if (n > entries_n)
n = entries_n;
for (i = 0; i < n; i++) {
for (raft_index_t i = 0; i < n; i++) {
entries[i] = r[i];
raft_entry_hold(entries[i]);
}
return (int) n;
return n;
}
static int log_pop(void *log, raft_index_t from_idx, raft_entry_notify_f cb, void *cb_arg)
......
......@@ -761,7 +761,7 @@ int raft_recv_appendentries(raft_server_t* me,
int e = 0;
raft_log_node(me, ae->leader_id,
"recv appendentries li:%d t:%ld ci:%ld lc:%ld pli:%ld plt:%ld msgid:%ld #%d",
"recv appendentries li:%d t:%ld ci:%ld lc:%ld pli:%ld plt:%ld msgid:%ld #%ld",
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);
......@@ -1222,20 +1222,57 @@ exit:
return 0;
}
raft_entry_t** raft_get_entries_from_idx(raft_server_t* me, raft_index_t idx, int* n_etys)
static raft_entry_t **raft_get_entries(raft_server_t *me,
raft_node_t *node,
raft_index_t idx,
raft_index_t *n_etys)
{
const int max_entries_in_append_req = 64 * 1024;
raft_index_t n;
/* If callback is not implemented, fetch entries from log implementation. */
if (!me->cb.get_entries_to_send) {
return raft_get_entries_from_idx(me, idx, n_etys);
}
*n_etys = 0;
if (raft_get_current_idx(me) < idx) {
return NULL;
}
raft_index_t size = raft_get_current_idx(me) - idx + 1;
if (size > max_entries_in_append_req) {
size = max_entries_in_append_req;
}
raft_entry_t **e = raft_malloc(size * sizeof(*e));
n = me->cb.get_entries_to_send(me, me->udata, node, idx, size, e);
if (n < 0) {
n = 0;
}
*n_etys = n;
return e;
}
raft_entry_t **raft_get_entries_from_idx(raft_server_t *me,
raft_index_t idx,
raft_index_t *n_etys)
{
*n_etys = 0;
if (raft_get_current_idx(me) < idx) {
*n_etys = 0;
return NULL;
}
raft_index_t size = raft_get_current_idx(me) - idx + 1;
raft_entry_t **e = raft_malloc(size * sizeof(raft_entry_t*));
int n = me->log_impl->get_batch(me->log, idx, (int) size, e);
raft_index_t n = me->log_impl->get_batch(me->log, idx, size, e);
if (n < 1) {
raft_free(e);
*n_etys = 0;
return NULL;
}
......@@ -1429,81 +1466,81 @@ int raft_recv_snapshot_response(raft_server_t* me,
return 0;
}
raft_term_t raft_get_previous_log_term(raft_server_t *me, raft_index_t idx)
{
assert(idx > 0);
raft_entry_t *ety = raft_get_entry_from_idx(me, idx - 1);
if (!ety) {
return me->snapshot_last_term;
}
raft_term_t term = ety->term;
raft_entry_release(ety);
return term;
}
int raft_send_appendentries(raft_server_t* me, raft_node_t* node)
{
assert(node);
assert(node != me->node);
assert(raft_node_get_next_idx(node) != 0);
int e;
if (!raft_node_is_active(node)) {
return 0;
}
raft_index_t next_idx = raft_node_get_next_idx(node);
/* figure out if the client needs a snapshot sent */
if (me->snapshot_last_idx > 0 && next_idx <= me->snapshot_last_idx)
{
if (me->snapshot_last_idx >= raft_node_get_next_idx(node)) {
return raft_send_snapshot(me, node);
}
if (!me->cb.send_appendentries)
if (!me->cb.send_appendentries) {
return -1;
}
if (me->cb.backpressure) {
if (me->cb.backpressure(me, me->udata, node) != 0) {
return 0;
do {
if (me->cb.backpressure) {
if (me->cb.backpressure(me, me->udata, node) != 0) {
return 0;
}
}
}
raft_appendentries_req_t ae = {
.term = me->current_term,
.leader_id = raft_get_nodeid(me),
.leader_commit = raft_get_commit_idx(me),
.msg_id = ++me->msg_id,
};
raft_index_t next_idx = raft_node_get_next_idx(node);
ae.entries = raft_get_entries_from_idx(me, next_idx, &ae.n_entries);
assert((!ae.entries && 0 == ae.n_entries) ||
(ae.entries && 0 < ae.n_entries));
raft_appendentries_req_t ae = {
.term = me->current_term,
.leader_id = raft_get_nodeid(me),
.leader_commit = raft_get_commit_idx(me),
.msg_id = ++me->msg_id,
.prev_log_idx = next_idx - 1,
.prev_log_term = raft_get_previous_log_term(me, next_idx),
.entries = raft_get_entries(me, node, next_idx, &ae.n_entries)
};
/* previous log is the log just before the new logs */
if (next_idx > 1)
{
raft_entry_t* prev_ety = raft_get_entry_from_idx(me, next_idx - 1);
if (!prev_ety)
{
ae.prev_log_idx = me->snapshot_last_idx;
ae.prev_log_term = me->snapshot_last_term;
}
else
{
ae.prev_log_idx = next_idx - 1;
ae.prev_log_term = prev_ety->term;
raft_entry_release(prev_ety);
}
}
raft_log_node(me, raft_node_get_id(node),
"send ae: t:%lu lc:%lu pli:%lu plt:%lu msgid:%lu #%ld",
ae.term,
ae.leader_commit,
ae.prev_log_idx,
ae.prev_log_term,
ae.msg_id,
ae.n_entries);
e = me->cb.send_appendentries(me, me->udata, node, &ae);
raft_log_node(me,
raft_node_get_id(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_commit_idx(me),
ae.term,
ae.leader_commit,
ae.prev_log_idx,
ae.prev_log_term,
ae.msg_id,
ae.n_entries);
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);
raft_entry_release_list(ae.entries, ae.n_entries);
return res;
if (e != 0) {
return e;
}
} while (raft_node_get_next_idx(node) <= raft_get_current_idx(me));
return 0;
}
int raft_send_appendentries_all(raft_server_t* me)
......@@ -1842,11 +1879,15 @@ void raft_entry_release(raft_entry_t *ety)
void raft_entry_release_list(raft_entry_t **ety_list, size_t len)
{
size_t i;
if (!ety_list) {
return;
}
for (i = 0; i < len; i++) {
for (size_t i = 0; i < len; i++) {
raft_entry_release(ety_list[i]);
}
raft_free(ety_list);
}
int raft_queue_read_request(raft_server_t* me, raft_read_request_callback_f cb, void *cb_arg)
......
......@@ -4633,6 +4633,61 @@ void TestRaft_config(CuTest *tc)
CuAssertIntEquals(tc, 1, val);
}
static int cb_send_ae(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)++;
CuAssertIntEquals(tc, 10, msg->n_entries);
return 0;
}
static raft_index_t cb_get_entries_to_send(raft_server_t *raft,
void *user_data,
raft_node_t *node,
raft_index_t idx,
raft_index_t entries_n,
raft_entry_t **entries)
{
const int count = 10;
/* Fill with 10 entries */
for (int i = 0; i < count; i++) {
entries[i] = raft_get_entry_from_idx(raft, idx + i);
}
return count;
}
void TestRaft_limit_appendentries_size(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);
}
int main(void)
{
CuString *output = CuStringNew();
......@@ -4778,6 +4833,7 @@ int main(void)
SUITE_ADD_TEST(suite, Test_transfer_leader_not_leader);
SUITE_ADD_TEST(suite, Test_transfer_automatic);
SUITE_ADD_TEST(suite, TestRaft_config);
SUITE_ADD_TEST(suite, TestRaft_limit_appendentries_size);
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