Commit 2d680b42 authored by Yossi Gottlieb's avatar Yossi Gottlieb
Browse files

Raft entry memory safety documentation + fixes.

parent bdd49a1e
......@@ -101,7 +101,7 @@ typedef struct
short type;
/** number of references */
short refs;
unsigned short refs;
/** private local data */
void *user_data;
......@@ -339,11 +339,9 @@ typedef int (
* @param[in] raft The Raft server making this callback
* @param[in] user_data User data that is passed from Raft server
* @param[in] entry The entry that the event is happening to.
* For offering, polling, and popping, the user is allowed to change the
* memory pointed to in the raft_entry_data_t struct. This MUST be done if
* the memory is temporary.
* @param[in] entry_idx The entries index in the log
* @return 0 on success */
* @return 0 on success
* */
typedef int (
*func_logentry_event_f
) (
......@@ -407,6 +405,7 @@ typedef struct
/** Callback for detecting when a non-voting node has sufficient logs. */
func_node_has_sufficient_logs_f node_has_sufficient_logs;
/** Callback for being notified of membership changes (optional). */
func_membership_event_f notify_membership_event;
/** Callback for catching debugging log messages
......@@ -493,7 +492,9 @@ typedef struct raft_log_impl
* RAFT_ERR_NOMEM memory allocation failure.
*
* @note
* The passed raft_entry_t is expected to be allocated on the heap.
* The passed raft_entry_t is expected to be allocated by raft_entry_new().
* The caller is expected to call raft_entry_release() after the append.
*
* The log implementation shall call raft_entry_hold() in order to
* maintain its reference count, and call raft_entry_release() when
* the entry is no longer needed.
......@@ -536,9 +537,8 @@ typedef struct raft_log_impl
* NULL if no entry in specified index.
*
* @note
* Returned entry memory is owned by the log and may be freed on next log
* operation, so caller must make a copy if it is going to be retained beyond
* this scope.
* Caller must use raft_entry_release() when no longer requiring the
* entry.
*/
raft_entry_t* (*get) (void *log, raft_index_t idx);
......@@ -550,6 +550,10 @@ typedef struct raft_log_impl
* @return
* Number of entries fetched;
* -1 on error.
*
* @note
* 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);
......@@ -669,14 +673,8 @@ int raft_periodic(raft_server_t* me, int msec_elapsed);
*
* Will block (ie. by syncing to disk) if we need to append a message.
*
* Might call malloc once to increase the log entry array size.
*
* The log_offer callback will be called.
*
* @note The memory pointer (ie. raft_entry_data_t) for each msg_entry_t is
* copied directly. If the memory is temporary you MUST either make the
* memory permanent (ie. via malloc) OR re-assign the memory within the
* log_offer callback.
* The caller is responsible to call raft_entry_release() for all
* included entries.
*
* @param[in] node The node who sent us this message
* @param[in] ae The appendentries message
......@@ -727,14 +725,8 @@ int raft_recv_requestvote_response(raft_server_t* me,
*
* Will block (ie. by syncing to disk) if we need to append a message.
*
* Might call malloc once to increase the log entry array size.
*
* The log_offer callback will be called.
*
* @note The memory pointer (ie. raft_entry_data_t) in msg_entry_t is
* copied directly. If the memory is temporary you MUST either make the
* memory permanent (ie. via malloc) OR re-assign the memory within the
* log_offer callback.
* The caller is responsible to call raft_entry_release() following this
* call.
*
* Will fail:
* <ul>
......
......@@ -245,6 +245,8 @@ static int __log_delete(log_t* me_, raft_index_t idx, func_entry_notify_f cb, vo
if (cb)
cb(cb_arg, me->entries[back], idx_tmp);
raft_entry_release(me->entries[back]);
me->back = back;
me->count--;
}
......@@ -272,6 +274,9 @@ int log_poll(log_t * me_, void** etyp)
if (0 != e)
return e;
}
raft_entry_release(me->entries[me->front]);
me->front++;
me->front = me->front % me->size;
me->count--;
......@@ -359,12 +364,16 @@ static void __log_reset(void *log, raft_index_t first_idx, raft_term_t term)
static int __log_append(void *log, raft_entry_t *entry)
{
raft_entry_hold(entry);
return log_append_entry(log, entry);
}
static raft_entry_t *__log_get(void *log, raft_index_t idx)
{
return log_get_at_idx(log, idx);
raft_entry_t *e = log_get_at_idx(log, idx);
if (e != NULL)
raft_entry_hold(e);
return e;
}
static int __log_get_batch(void *log, raft_index_t idx, int entries_n, raft_entry_t **entries)
......@@ -381,6 +390,7 @@ static int __log_get_batch(void *log, raft_index_t idx, int entries_n, raft_entr
for (i = 0; i < n; i++) {
entries[i] = r[i];
raft_entry_hold(entries[i]);
}
return n;
}
......
......@@ -12,6 +12,7 @@
#include <string.h>
#include <stdio.h>
#include <assert.h>
#include <stdint.h>
/* for varags */
#include <stdarg.h>
......@@ -376,6 +377,8 @@ int raft_recv_appendentries_response(raft_server_t* me_,
if (raft_get_num_voting_nodes(me_) / 2 < votes)
raft_set_commit_idx(me_, point);
}
if (ety)
raft_entry_release(ety);
}
/* Aggressively send remaining entries */
......@@ -446,6 +449,8 @@ int raft_recv_appendentries(
/* Should never happen; something is seriously wrong! */
__log(me_, node, "Snapshot AE prev conflicts with committed entry");
e = RAFT_ERR_SHUTDOWN;
if (ety)
raft_entry_release(ety);
goto out;
}
}
......@@ -466,12 +471,16 @@ int raft_recv_appendentries(
/* Should never happen; something is seriously wrong! */
__log(me_, node, "AE prev conflicts with committed entry");
e = RAFT_ERR_SHUTDOWN;
raft_entry_release(ety);
goto out;
}
/* Delete all the following log entries because they don't match */
e = raft_delete_entry_from_idx(me_, ae->prev_log_idx);
raft_entry_release(ety);
goto out;
}
if (ety)
raft_entry_release(ety);
}
r->success = 1;
......@@ -485,8 +494,13 @@ int raft_recv_appendentries(
{
raft_entry_t* ety = ae->entries[i];
raft_index_t ety_index = ae->prev_log_idx + 1 + i;
raft_entry_t* existing_ety = raft_get_entry_from_idx(me_, ety_index);
if (existing_ety && existing_ety->term != ety->term)
raft_term_t existing_term = existing_ety ? existing_ety->term : 0;
if (existing_ety)
raft_entry_release(existing_ety);
if (existing_ety && existing_term != ety->term)
{
if (ety_index <= raft_get_commit_idx(me_))
{
......@@ -561,9 +575,10 @@ static int __should_grant_vote(raft_server_private_t* me, msg_requestvote_t* vr)
int ety_term;
// TODO: add test
if (ety)
if (ety) {
ety_term = ety->term;
else if (!ety && me->snapshot_last_idx == current_idx)
raft_entry_release(ety);
} else if (!ety && me->snapshot_last_idx == current_idx)
ety_term = me->snapshot_last_term;
else
return 0;
......@@ -845,8 +860,10 @@ int raft_apply_entry(raft_server_t* me_)
if (me->cb.applylog)
{
int e = me->cb.applylog(me_, me->udata, ety, me->last_applied_idx);
if (RAFT_ERR_SHUTDOWN == e)
if (RAFT_ERR_SHUTDOWN == e) {
raft_entry_release(ety);
return RAFT_ERR_SHUTDOWN;
}
}
/* voting cfg change is now complete.
......@@ -888,6 +905,7 @@ int raft_apply_entry(raft_server_t* me_)
exit:
raft_entry_release(ety);
return 0;
}
......@@ -956,6 +974,7 @@ int raft_send_appendentries(raft_server_t* me_, raft_node_t* node)
{
ae.prev_log_idx = next_idx - 1;
ae.prev_log_term = prev_ety->term;
raft_entry_release(prev_ety);
}
}
......@@ -969,6 +988,7 @@ int raft_send_appendentries(raft_server_t* me_, raft_node_t* node)
ae.n_entries);
int res = me->cb.send_appendentries(me_, me->udata, node, &ae);
raft_entry_release_list(ae.entries, ae.n_entries);
__raft_free(ae.entries);
return res;
......@@ -1505,6 +1525,7 @@ raft_entry_t *raft_entry_new(unsigned int data_len)
void raft_entry_hold(raft_entry_t *ety)
{
assert(ety->refs < UINT16_MAX);
ety->refs++;
}
......
......@@ -221,6 +221,7 @@ raft_term_t raft_get_last_log_term(raft_server_t* me_)
raft_entry_t* ety = raft_get_entry_from_idx(me_, current_idx);
if (ety) {
raft_term_t term = ety->term;
raft_entry_release(ety);
return term;
}
}
......
......@@ -32,13 +32,13 @@ static raft_entry_t **__MAKE_ENTRY_ARRAY_SEQ_ID(int count, int start_id, raft_te
return array;
}
static void __APPEND_ENTRY(void *r, int id, raft_term_t term, const char *data)
static void __RAFT_APPEND_ENTRY(void *r, int id, raft_term_t term, const char *data)
{
raft_entry_t *e = __MAKE_ENTRY(id, term, data);
raft_append_entry(r, e);
}
static void __APPEND_ENTRIES_SEQ_ID(void *r, int count, int id, raft_term_t term, const char *data)
static void __RAFT_APPEND_ENTRIES_SEQ_ID(void *r, int count, int id, raft_term_t term, const char *data)
{
int i;
for (i = 0; i < count; i++) {
......@@ -47,7 +47,7 @@ static void __APPEND_ENTRIES_SEQ_ID(void *r, int count, int id, raft_term_t term
}
}
static void __APPEND_ENTRIES_SEQ_ID_TERM(void *r, int count, int id, raft_term_t term, const char *data)
static void __RAFT_APPEND_ENTRIES_SEQ_ID_TERM(void *r, int count, int id, raft_term_t term, const char *data)
{
int i;
for (i = 0; i < count; i++) {
......@@ -56,5 +56,4 @@ static void __APPEND_ENTRIES_SEQ_ID_TERM(void *r, int count, int id, raft_term_t
}
}
#endif
......@@ -21,6 +21,7 @@ class Libraft(object):
src/raft_node.c
""".split(),
include_dirs=["include"],
extra_compile_args=["-UNDEBUG"]
)
library = ffi.compile()
......@@ -83,7 +84,7 @@ class CoreTestCase(unittest.TestCase):
for cmd in commands:
if cmd == 'append':
entry = self.r.ffi.new('raft_entry_t*')
entry = r.raft_entry_new(0)
entry.id = unique_id
unique_id += 1
......
......@@ -14,6 +14,21 @@
#include "helpers.h"
static void __LOG_APPEND_ENTRY(void *l, int id, raft_term_t term, const char *data)
{
raft_entry_t *e = __MAKE_ENTRY(id, term, data);
log_append_entry(l, e);
}
static void __LOG_APPEND_ENTRIES_SEQ_ID(void *l, int count, int id, raft_term_t term, const char *data)
{
int i;
for (i = 0; i < count; i++) {
raft_entry_t *e = __MAKE_ENTRY(id++, term, data);
log_append_entry(l, e);
}
}
static int __logentry_get_node_id(
raft_server_t* raft,
void *udata,
......@@ -86,65 +101,46 @@ void TestLog_new_is_empty(CuTest * tc)
void TestLog_append_is_not_empty(CuTest * tc)
{
void *l;
raft_entry_t e;
void *r = raft_new();
memset(&e, 0, sizeof(raft_entry_t));
e.id = 1;
l = log_new();
raft_log_cbs_t funcs = {
.log_offer = __log_offer
};
log_set_callbacks(l, &funcs, r);
CuAssertIntEquals(tc, 0, log_append_entry(l, &e));
__LOG_APPEND_ENTRY(l, 1, 0, NULL);
CuAssertIntEquals(tc, 1, log_count(l));
}
void TestLog_get_at_idx(CuTest * tc)
{
void *l;
raft_entry_t e1, e2, e3;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
l = log_new();
e1.id = 1;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e1));
e2.id = 2;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e2));
e3.id = 3;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e3));
__LOG_APPEND_ENTRIES_SEQ_ID(l, 3, 1, 0, NULL);
CuAssertIntEquals(tc, 3, log_count(l));
CuAssertIntEquals(tc, e1.id, log_get_at_idx(l, 1)->id);
CuAssertIntEquals(tc, e2.id, log_get_at_idx(l, 2)->id);
CuAssertIntEquals(tc, e3.id, log_get_at_idx(l, 3)->id);
CuAssertIntEquals(tc, 1, log_get_at_idx(l, 1)->id);
CuAssertIntEquals(tc, 2, log_get_at_idx(l, 2)->id);
CuAssertIntEquals(tc, 3, log_get_at_idx(l, 3)->id);
}
void TestLog_get_at_idx_returns_null_where_out_of_bounds(CuTest * tc)
{
void *l;
raft_entry_t e1;
memset(&e1, 0, sizeof(raft_entry_t));
l = log_new();
CuAssertTrue(tc, NULL == log_get_at_idx(l, 0));
CuAssertTrue(tc, NULL == log_get_at_idx(l, 1));
e1.id = 1;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e1));
__LOG_APPEND_ENTRY(l, 1, 0, NULL);
CuAssertTrue(tc, NULL == log_get_at_idx(l, 2));
}
void TestLog_delete(CuTest * tc)
{
void *l;
raft_entry_t e1, e2, e3;
void* queue = llqueue_new();
void *r = raft_new();
......@@ -160,22 +156,13 @@ void TestLog_delete(CuTest * tc)
l = log_new();
log_set_callbacks(l, &log_funcs, r);
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
e1.id = 1;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e1));
e2.id = 2;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e2));
e3.id = 3;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e3));
__LOG_APPEND_ENTRIES_SEQ_ID(l, 3, 1, 0, NULL);
CuAssertIntEquals(tc, 3, log_count(l));
CuAssertIntEquals(tc, 3, log_get_current_idx(l));
log_delete(l, 3);
CuAssertIntEquals(tc, 2, log_count(l));
CuAssertIntEquals(tc, e3.id, ((raft_entry_t*)llqueue_poll(queue))->id);
CuAssertIntEquals(tc, 3, ((raft_entry_t*)llqueue_poll(queue))->id);
CuAssertIntEquals(tc, 2, log_count(l));
CuAssertTrue(tc, NULL == log_get_at_idx(l, 3));
......@@ -201,26 +188,17 @@ void TestLog_delete_onwards(CuTest * tc)
raft_set_callbacks(r, &funcs, queue);
void *l;
raft_entry_t e1, e2, e3;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
l = log_new();
log_set_callbacks(l, &log_funcs, r);
e1.id = 1;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e1));
e2.id = 2;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e2));
e3.id = 3;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e3));
__LOG_APPEND_ENTRIES_SEQ_ID(l, 3, 1, 0, NULL);
CuAssertIntEquals(tc, 3, log_count(l));
/* even 3 gets deleted */
log_delete(l, 2);
CuAssertIntEquals(tc, 1, log_count(l));
CuAssertIntEquals(tc, e1.id, log_get_at_idx(l, 1)->id);
CuAssertIntEquals(tc, 1, log_get_at_idx(l, 1)->id);
CuAssertTrue(tc, NULL == log_get_at_idx(l, 2));
CuAssertTrue(tc, NULL == log_get_at_idx(l, 3));
}
......@@ -228,7 +206,6 @@ void TestLog_delete_onwards(CuTest * tc)
void TestLog_delete_handles_log_pop_failure(CuTest * tc)
{
void *l;
raft_entry_t e1, e2, e3;
void* queue = llqueue_new();
void *r = raft_new();
......@@ -243,23 +220,14 @@ void TestLog_delete_handles_log_pop_failure(CuTest * tc)
l = log_new();
log_set_callbacks(l, &log_funcs, r);
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
e1.id = 1;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e1));
e2.id = 2;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e2));
e3.id = 3;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e3));
__LOG_APPEND_ENTRIES_SEQ_ID(l, 3, 1, 0, NULL);
CuAssertIntEquals(tc, 3, log_count(l));
CuAssertIntEquals(tc, 3, log_get_current_idx(l));
CuAssertIntEquals(tc, -1, log_delete(l, 3));
CuAssertIntEquals(tc, 3, log_count(l));
CuAssertIntEquals(tc, 3, log_count(l));
CuAssertIntEquals(tc, e3.id, ((raft_entry_t*)log_peektail(l))->id);
CuAssertIntEquals(tc, 3, ((raft_entry_t*)log_peektail(l))->id);
}
void TestLog_delete_fails_for_idx_zero(CuTest * tc)
......@@ -276,24 +244,10 @@ void TestLog_delete_fails_for_idx_zero(CuTest * tc)
log_set_callbacks(raft_get_log(r), &log_funcs, r);
void *l;
raft_entry_t e1, e2, e3, e4;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
memset(&e4, 0, sizeof(raft_entry_t));
e1.id = 1;
e2.id = 2;
e3.id = 3;
e4.id = 4;
l = log_alloc(1);
log_set_callbacks(l, &log_funcs, r);
CuAssertIntEquals(tc, 0, log_append_entry(l, &e1));
CuAssertIntEquals(tc, 0, log_append_entry(l, &e2));
CuAssertIntEquals(tc, 0, log_append_entry(l, &e3));
CuAssertIntEquals(tc, 0, log_append_entry(l, &e4));
__LOG_APPEND_ENTRIES_SEQ_ID(l, 4, 1, 0, NULL);
CuAssertIntEquals(tc, log_delete(l, 0), -1);
}
......@@ -310,25 +264,17 @@ void TestLog_poll(CuTest * tc)
raft_set_callbacks(r, &funcs, queue);
void *l;
raft_entry_t e1, e2, e3;
l = log_new();
log_set_callbacks(l, &log_funcs, r);
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
e1.id = 1;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e1));
__LOG_APPEND_ENTRY(l, 1, 0, NULL);
CuAssertIntEquals(tc, 1, log_get_current_idx(l));
e2.id = 2;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e2));
__LOG_APPEND_ENTRY(l, 2, 0, NULL);
CuAssertIntEquals(tc, 2, log_get_current_idx(l));
e3.id = 3;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e3));
__LOG_APPEND_ENTRY(l, 3, 0, NULL);
CuAssertIntEquals(tc, 3, log_count(l));
CuAssertIntEquals(tc, 3, log_get_current_idx(l));
......@@ -372,21 +318,12 @@ void TestLog_poll(CuTest * tc)
void TestLog_peektail(CuTest * tc)
{
void *l;
raft_entry_t e1, e2, e3;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
l = log_new();
e1.id = 1;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e1));
e2.id = 2;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e2));
e3.id = 3;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e3));
__LOG_APPEND_ENTRIES_SEQ_ID(l, 3, 1, 0, NULL);
CuAssertIntEquals(tc, 3, log_count(l));
CuAssertIntEquals(tc, e3.id, log_peektail(l)->id);
CuAssertIntEquals(tc, 3, log_peektail(l)->id);
}
#if 0
......@@ -407,11 +344,6 @@ void T_estlog_cant_append_duplicates(CuTest * tc)
void TestLog_load_from_snapshot(CuTest * tc)
{
void *l;
raft_entry_t e1, e2, e3;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
l = log_new();
CuAssertIntEquals(tc, 0, log_get_current_idx(l));
......@@ -423,16 +355,10 @@ void TestLog_load_from_snapshot(CuTest * tc)
void TestLog_load_from_snapshot_clears_log(CuTest * tc)
{
void *l;
raft_entry_t e1, e2, e3;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
l = log_new();
CuAssertIntEquals(tc, 0, log_append_entry(l, &e1));
CuAssertIntEquals(tc, 0, log_append_entry(l, &e2));
__LOG_APPEND_ENTRIES_SEQ_ID(l, 2, 1, 0, NULL);
CuAssertIntEquals(tc, 2, log_count(l));
CuAssertIntEquals(tc, 2, log_get_current_idx(l));
......@@ -447,18 +373,15 @@ void TestLog_front_pushes_across_boundary(CuTest * tc)
void *l;
raft_entry_t *e1 = __MAKE_ENTRY(1, 1, "aa");
raft_entry_t *e2 = __MAKE_ENTRY(2, 1, "aa");
l = log_alloc(1);
log_set_callbacks(l, &log_funcs, r);
raft_entry_t* ety;
CuAssertIntEquals(tc, 0, log_append_entry(l, e1));
__LOG_APPEND_ENTRY(l, 1, 0, NULL);
CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0);
CuAssertIntEquals(tc, ety->id, 1);
CuAssertIntEquals(tc, 0, log_append_entry(l, e2));
__LOG_APPEND_ENTRY(l, 2, 0, NULL);
CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0);
CuAssertIntEquals(tc, ety->id, 2);
}
......@@ -466,39 +389,28 @@ void TestLog_front_pushes_across_boundary(CuTest * tc)
void TestLog_front_and_back_pushed_across_boundary_with_enlargement_required(CuTest * tc)
{
void *l;
raft_entry_t e1, e2, e3, e4;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
memset(&e4, 0, sizeof(raft_entry_t));
e1.id = 1;
e2.id = 2;
e3.id = 3;
e4.id = 4;
l = log_alloc(1);
raft_entry_t* ety;
/* append */
CuAssertIntEquals(tc, 0, log_append_entry(l, &e1));
__LOG_APPEND_ENTRY(l, 1, 0, NULL);
/* poll */
CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0);
CuAssertIntEquals(tc, ety->id, 1);
/* append */
CuAssertIntEquals(tc, 0, log_append_entry(l, &e2));
__LOG_APPEND_ENTRY(l, 2, 0, NULL);
/* poll */
CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0);
CuAssertIntEquals(tc, ety->id, 2);
/* append append */
CuAssertIntEquals(tc, 0, log_append_entry(l, &e3));
CuAssertIntEquals(tc, 0, log_append_entry(l, &e4));
__LOG_APPEND_ENTRY(l, 3, 0, NULL);
__LOG_APPEND_ENTRY(l, 4, 0, NULL);
/* poll */
CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0);
......@@ -508,24 +420,13 @@ void TestLog_front_and_back_pushed_across_boundary_with_enlargement_required(CuT
void TestLog_delete_after_polling(CuTest * tc)
{
void *l;
raft_entry_t e1, e2, e3, e4;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
memset(&e4, 0, sizeof(raft_entry_t));
e1.id = 1;
e2.id = 2;
e3.id = 3;
e4.id = 4;
l = log_alloc(1);
raft_entry_t* ety;
/* append */
CuAssertIntEquals(tc, 0, log_append_entry(l, &e1));
__LOG_APPEND_ENTRY(l, 1, 0, NULL);
CuAssertIntEquals(tc, 1, log_count(l));
/* poll */
......@@ -534,7 +435,7 @@ void TestLog_delete_after_polling(CuTest * tc)
CuAssertIntEquals(tc, 0, log_count(l));
/* append */
CuAssertIntEquals(tc, 0, log_append_entry(l, &e2));
__LOG_APPEND_ENTRY(l, 2, 0, NULL);
CuAssertIntEquals(tc, 1, log_count(l));
/* poll */
......@@ -555,17 +456,6 @@ void TestLog_delete_after_polling_from_double_append(CuTest * tc)
raft_set_callbacks(r, &funcs, queue);
void *l;
raft_entry_t e1, e2, e3, e4;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
memset(&e4, 0, sizeof(raft_entry_t));
e1.id = 1;
e2.id = 2;
e3.id = 3;
e4.id = 4;
l = log_alloc(1);
log_set_callbacks(l, &log_funcs, r);
......@@ -573,8 +463,7 @@ void TestLog_delete_after_polling_from_double_append(CuTest * tc)
raft_entry_t* ety;
/* append append */
CuAssertIntEquals(tc, 0, log_append_entry(l, &e1));
CuAssertIntEquals(tc, 0, log_append_entry(l, &e2));
__LOG_APPEND_ENTRIES_SEQ_ID(l, 2, 1, 0, NULL);
CuAssertIntEquals(tc, 2, log_count(l));
/* poll */
......@@ -583,7 +472,7 @@ void TestLog_delete_after_polling_from_double_append(CuTest * tc)
CuAssertIntEquals(tc, 1, log_count(l));
/* append */
CuAssertIntEquals(tc, 0, log_append_entry(l, &e3));
__LOG_APPEND_ENTRY(l, 3, 0, NULL);
CuAssertIntEquals(tc, 2, log_count(l));
/* poll */
......@@ -604,13 +493,6 @@ void TestLog_get_from_idx_with_base_off_by_one(CuTest * tc)
raft_set_callbacks(r, &funcs, queue);
void *l;
raft_entry_t e1, e2;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
e1.id = 1;
e2.id = 2;
l = log_alloc(1);
log_set_callbacks(l, &log_funcs, r);
......@@ -618,8 +500,7 @@ void TestLog_get_from_idx_with_base_off_by_one(CuTest * tc)
raft_entry_t* ety;
/* append append */
CuAssertIntEquals(tc, 0, log_append_entry(l, &e1));
CuAssertIntEquals(tc, 0, log_append_entry(l, &e2));
__LOG_APPEND_ENTRIES_SEQ_ID(l, 2, 1, 0, NULL);
CuAssertIntEquals(tc, 2, log_count(l));
/* poll */
......
......@@ -9,6 +9,7 @@
#include "linked_list_queue.h"
#include "raft.h"
#include "helpers.h"
static int __logentry_get_node_id(
raft_server_t* raft,
......@@ -56,6 +57,22 @@ static int __log_pop_failing(
static const raft_log_impl_t *impl = &raft_log_internal_impl;
static void __LOGIMPL_APPEND_ENTRY(void *l, int id, raft_term_t term, const char *data)
{
raft_entry_t *e = __MAKE_ENTRY(id, term, data);
impl->append(l, e);
}
static void __LOGIMPL_APPEND_ENTRIES_SEQ_ID(void *l, int count, int id, raft_term_t term, const char *data)
{
int i;
for (i = 0; i < count; i++) {
raft_entry_t *e = __MAKE_ENTRY(id++, term, data);
impl->append(l, e);
}
}
void TestLogImpl_new_is_empty(CuTest * tc)
{
void *l;
......@@ -68,54 +85,35 @@ void TestLogImpl_new_is_empty(CuTest * tc)
void TestLogImpl_append_is_not_empty(CuTest * tc)
{
void *l;
raft_entry_t e;
void *r = raft_new();
memset(&e, 0, sizeof(raft_entry_t));
e.id = 1;
l = impl->init(r, NULL);
CuAssertIntEquals(tc, 0, impl->append(l, &e));
__LOGIMPL_APPEND_ENTRY(l, 1, 0, NULL);
CuAssertIntEquals(tc, 1, impl->count(l));
}
void TestLogImpl_get_at_idx(CuTest * tc)
{
void *l;
raft_entry_t e1, e2, e3;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
l = impl->init(NULL, NULL);
e1.id = 1;
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
e2.id = 2;
CuAssertIntEquals(tc, 0, impl->append(l, &e2));
e3.id = 3;
CuAssertIntEquals(tc, 0, impl->append(l, &e3));
__LOGIMPL_APPEND_ENTRIES_SEQ_ID(l, 3, 1, 0, NULL);
CuAssertIntEquals(tc, 3, impl->count(l));
CuAssertIntEquals(tc, e1.id, impl->get(l, 1)->id);
CuAssertIntEquals(tc, e2.id, impl->get(l, 2)->id);
CuAssertIntEquals(tc, e3.id, impl->get(l, 3)->id);
CuAssertIntEquals(tc, 1, impl->get(l, 1)->id);
CuAssertIntEquals(tc, 2, impl->get(l, 2)->id);
CuAssertIntEquals(tc, 3, impl->get(l, 3)->id);
}
void TestLogImpl_get_at_idx_returns_null_where_out_of_bounds(CuTest * tc)
{
void *l;
raft_entry_t e1;
memset(&e1, 0, sizeof(raft_entry_t));
l = impl->init(NULL, NULL);
CuAssertTrue(tc, NULL == impl->get(l, 0));
CuAssertTrue(tc, NULL == impl->get(l, 1));
e1.id = 1;
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
__LOGIMPL_APPEND_ENTRY(l, 1, 0, NULL);
CuAssertTrue(tc, NULL == impl->get(l, 2));
}
......@@ -130,7 +128,6 @@ void TestLogImpl_pop(CuTest * tc)
{
void* queue = llqueue_new();
void *l;
raft_entry_t e1, e2, e3;
void *r = raft_new();
raft_cbs_t funcs = {
......@@ -140,23 +137,13 @@ void TestLogImpl_pop(CuTest * tc)
raft_set_callbacks(r, &funcs, NULL);
l = impl->init(r, NULL);
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
e1.id = 1;
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
e2.id = 2;
CuAssertIntEquals(tc, 0, impl->append(l, &e2));
e3.id = 3;
CuAssertIntEquals(tc, 0, impl->append(l, &e3));
__LOGIMPL_APPEND_ENTRIES_SEQ_ID(l, 3, 1, 0, NULL);
CuAssertIntEquals(tc, 3, impl->count(l));
CuAssertIntEquals(tc, 3, impl->current_idx(l));
impl->pop(l, 3, event_entry_enqueue, queue);
CuAssertIntEquals(tc, 2, impl->count(l));
CuAssertIntEquals(tc, e3.id, ((raft_entry_t*)llqueue_poll(queue))->id);
CuAssertIntEquals(tc, 3, ((raft_entry_t*)llqueue_poll(queue))->id);
CuAssertTrue(tc, NULL == impl->get(l, 3));
impl->pop(l, 2, NULL, NULL);
......@@ -173,25 +160,15 @@ void TestLogImpl_pop_onwards(CuTest * tc)
void *r = raft_new();
void *l;
raft_entry_t e1, e2, e3;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
l = impl->init(r, NULL);
e1.id = 1;
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
e2.id = 2;
CuAssertIntEquals(tc, 0, impl->append(l, &e2));
e3.id = 3;
CuAssertIntEquals(tc, 0, impl->append(l, &e3));
__LOGIMPL_APPEND_ENTRIES_SEQ_ID(l, 3, 1, 0, NULL);
CuAssertIntEquals(tc, 3, impl->count(l));
/* even 3 gets deleted */
impl->pop(l, 2, NULL, NULL);
CuAssertIntEquals(tc, 1, impl->count(l));
CuAssertIntEquals(tc, e1.id, impl->get(l, 1)->id);
CuAssertIntEquals(tc, 1, impl->get(l, 1)->id);
CuAssertTrue(tc, NULL == impl->get(l, 2));
CuAssertTrue(tc, NULL == impl->get(l, 3));
}
......@@ -202,23 +179,9 @@ void TestLogImpl_pop_fails_for_idx_zero(CuTest * tc)
void *r = raft_new();
void *l;
raft_entry_t e1, e2, e3, e4;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
memset(&e4, 0, sizeof(raft_entry_t));
e1.id = 1;
e2.id = 2;
e3.id = 3;
e4.id = 4;
l = impl->init(r, NULL);
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
CuAssertIntEquals(tc, 0, impl->append(l, &e2));
CuAssertIntEquals(tc, 0, impl->append(l, &e3));
CuAssertIntEquals(tc, 0, impl->append(l, &e4));
__LOGIMPL_APPEND_ENTRIES_SEQ_ID(l, 4, 1, 0, NULL);
CuAssertIntEquals(tc, impl->pop(l, 0, NULL, NULL), -1);
}
......@@ -227,24 +190,15 @@ void TestLogImpl_poll(CuTest * tc)
void *r = raft_new();
void *l;
raft_entry_t e1, e2, e3;
l = impl->init(r, NULL);
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
e1.id = 1;
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
__LOGIMPL_APPEND_ENTRY(l, 1, 0, NULL);
CuAssertIntEquals(tc, 1, impl->current_idx(l));
e2.id = 2;
CuAssertIntEquals(tc, 0, impl->append(l, &e2));
__LOGIMPL_APPEND_ENTRY(l, 2, 0, NULL);
CuAssertIntEquals(tc, 2, impl->current_idx(l));
e3.id = 3;
CuAssertIntEquals(tc, 0, impl->append(l, &e3));
__LOGIMPL_APPEND_ENTRY(l, 3, 0, NULL);
CuAssertIntEquals(tc, 3, impl->count(l));
CuAssertIntEquals(tc, 3, impl->current_idx(l));
......@@ -300,16 +254,10 @@ void TestLogImpl_reset_after_compaction(CuTest * tc)
void TestLogImpl_load_from_snapshot_clears_log(CuTest * tc)
{
void *l;
raft_entry_t e1, e2, e3;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
l = impl->init(NULL, NULL);
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
CuAssertIntEquals(tc, 0, impl->append(l, &e2));
__LOGIMPL_APPEND_ENTRIES_SEQ_ID(l, 2, 1, 0, NULL);
CuAssertIntEquals(tc, 2, impl->count(l));
CuAssertIntEquals(tc, 2, impl->current_idx(l));
......@@ -321,22 +269,11 @@ void TestLogImpl_load_from_snapshot_clears_log(CuTest * tc)
void TestLogImpl_pop_after_polling(CuTest * tc)
{
void *l;
raft_entry_t e1, e2, e3, e4;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
memset(&e4, 0, sizeof(raft_entry_t));
e1.id = 1;
e2.id = 2;
e3.id = 3;
e4.id = 4;
l = impl->init(NULL, NULL);
/* append */
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
__LOGIMPL_APPEND_ENTRY(l, 1, 0, NULL);
CuAssertIntEquals(tc, 1, impl->count(l));
CuAssertIntEquals(tc, 1, impl->first_idx(l));
CuAssertIntEquals(tc, 1, impl->current_idx(l));
......@@ -348,7 +285,7 @@ void TestLogImpl_pop_after_polling(CuTest * tc)
CuAssertIntEquals(tc, 1, impl->current_idx(l));
/* append */
CuAssertIntEquals(tc, 0, impl->append(l, &e2));
__LOGIMPL_APPEND_ENTRY(l, 2, 0, NULL);
CuAssertIntEquals(tc, 1, impl->count(l));
CuAssertIntEquals(tc, 2, impl->current_idx(l));
......@@ -362,23 +299,11 @@ void TestLogImpl_pop_after_polling_from_double_append(CuTest * tc)
{
void *r = raft_new();
void *l;
raft_entry_t e1, e2, e3, e4;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
memset(&e4, 0, sizeof(raft_entry_t));
e1.id = 1;
e2.id = 2;
e3.id = 3;
e4.id = 4;
l = impl->init(r, NULL);
/* append append */
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
CuAssertIntEquals(tc, 0, impl->append(l, &e2));
__LOGIMPL_APPEND_ENTRIES_SEQ_ID(l, 2, 1, 0, NULL);
CuAssertIntEquals(tc, 2, impl->count(l));
/* poll */
......@@ -387,7 +312,7 @@ void TestLogImpl_pop_after_polling_from_double_append(CuTest * tc)
CuAssertIntEquals(tc, 1, impl->count(l));
/* append */
CuAssertIntEquals(tc, 0, impl->append(l, &e3));
__LOGIMPL_APPEND_ENTRY(l, 3, 0, NULL);
CuAssertIntEquals(tc, 2, impl->count(l));
/* pop */
......@@ -400,19 +325,11 @@ void TestLogImpl_get_from_idx_with_base_off_by_one(CuTest * tc)
void *r = raft_new();
void *l;
raft_entry_t e1, e2;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
e1.id = 1;
e2.id = 2;
l = impl->init(r, NULL);
/* append append */
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
CuAssertIntEquals(tc, 0, impl->append(l, &e2));
__LOGIMPL_APPEND_ENTRIES_SEQ_ID(l, 2, 1, 0, NULL);
CuAssertIntEquals(tc, 2, impl->count(l));
/* poll */
......
......@@ -129,7 +129,7 @@ void TestRaft_server_idx_starts_at_1(CuTest * tc)
void *r = raft_new();
CuAssertTrue(tc, 0 == raft_get_current_idx(r));
__APPEND_ENTRY(r, 1, 1, "aaa");
__RAFT_APPEND_ENTRY(r, 1, 1, "aaa");
CuAssertTrue(tc, 1 == raft_get_current_idx(r));
}
......@@ -275,7 +275,7 @@ void TestRaft_server_append_entry_is_retrievable(CuTest * tc)
raft_set_state(r, RAFT_STATE_CANDIDATE);
raft_set_current_term(r, 5);
__APPEND_ENTRY(r, 100, 1, "aaa");
__RAFT_APPEND_ENTRY(r, 100, 1, "aaa");
raft_entry_t* kept = raft_get_entry_from_idx(r, 1);
CuAssertIntEquals(tc, 3, kept->data_len);
......@@ -330,8 +330,8 @@ void TestRaft_server_entry_is_retrieveable_using_idx(CuTest* tc)
void *r = raft_new();
__APPEND_ENTRY(r, 1, 1, str);
__APPEND_ENTRY(r, 2, 1, str2);
__RAFT_APPEND_ENTRY(r, 1, 1, str);
__RAFT_APPEND_ENTRY(r, 2, 1, str2);
CuAssertTrue(tc, NULL != (ety_appended = raft_get_entry_from_idx(r, 2)));
CuAssertTrue(tc, !strncmp(ety_appended->data, str2, 3));
......@@ -362,7 +362,7 @@ void TestRaft_server_wont_apply_entry_if_there_isnt_a_majority(CuTest* tc)
CuAssertTrue(tc, 0 == raft_get_commit_idx(r));
char *str = "aaa";
__APPEND_ENTRY(r, 1, 1, str);
__RAFT_APPEND_ENTRY(r, 1, 1, str);
raft_apply_entry(r);
/* Not allowed to be applied because we haven't confirmed a majority yet */
CuAssertTrue(tc, 0 == raft_get_last_applied_idx(r));
......@@ -388,7 +388,7 @@ void TestRaft_server_increment_lastApplied_when_lastApplied_lt_commitidx(
raft_set_last_applied_idx(r, 0);
/* need at least one entry */
__APPEND_ENTRY(r, 1, 1, "aaa");
__RAFT_APPEND_ENTRY(r, 1, 1, "aaa");
raft_set_commit_idx(r, 1);
......@@ -414,7 +414,7 @@ void TestRaft_user_applylog_error_propogates_to_periodic(
raft_set_last_applied_idx(r, 0);
/* need at least one entry */
__APPEND_ENTRY(r, 1, 1, "aaa");
__RAFT_APPEND_ENTRY(r, 1, 1, "aaa");
raft_set_commit_idx(r, 1);
......@@ -433,7 +433,7 @@ void TestRaft_server_apply_entry_increments_last_applied_idx(CuTest* tc)
raft_set_callbacks(r, &funcs, NULL);
raft_set_last_applied_idx(r, 0);
__APPEND_ENTRY(r, 1, 1, "aaa");
__RAFT_APPEND_ENTRY(r, 1, 1, "aaa");
raft_set_commit_idx(r, 1);
raft_apply_entry(r);
CuAssertTrue(tc, 1 == raft_get_last_applied_idx(r));
......@@ -1218,19 +1218,19 @@ static raft_entry_t* __create_mock_entries_for_conflict_tests(
/* increase log size */
char *str1 = strs[0];
__APPEND_ENTRY(r, 1, 1, str1);
__RAFT_APPEND_ENTRY(r, 1, 1, str1);
CuAssertTrue(tc, 1 == raft_get_log_count(r));
/* this log will be overwritten by a later appendentries */
char *str2 = strs[1];
__APPEND_ENTRY(r, 2, 1, str2);
__RAFT_APPEND_ENTRY(r, 2, 1, str2);
CuAssertTrue(tc, 2 == raft_get_log_count(r));
CuAssertTrue(tc, NULL != (ety_appended = raft_get_entry_from_idx(r, 2)));
CuAssertTrue(tc, !strncmp(ety_appended->data, str2, 3));
/* this log will be overwritten by a later appendentries */
char *str3 = strs[2];
__APPEND_ENTRY(r, 3, 1, str3);
__RAFT_APPEND_ENTRY(r, 3, 1, str3);
CuAssertTrue(tc, 3 == raft_get_log_count(r));
CuAssertTrue(tc, NULL != (ety_appended = raft_get_entry_from_idx(r, 3)));
CuAssertTrue(tc, !strncmp(ety_appended->data, str3, 3));
......@@ -1494,8 +1494,8 @@ void TestRaft_follower_recv_appendentries_partial_failures(
raft_set_current_term(r, 1);
/* Append entry 1 and 2 of term 1. */
__APPEND_ENTRY(r, 1, 1, "1aa");
__APPEND_ENTRY(r, 2, 1, "1bb");
__RAFT_APPEND_ENTRY(r, 1, 1, "1aa");
__RAFT_APPEND_ENTRY(r, 2, 1, "1bb");
CuAssertIntEquals(tc, 2, raft_get_current_idx(r));
msg_appendentries_t ae;
......@@ -1658,7 +1658,7 @@ void TestRaft_follower_recv_appendentries_failure_includes_current_idx(
raft_add_node(r, NULL, 2, 0);
raft_set_current_term(r, 1);
__APPEND_ENTRY(r, 1, 1, "aaa");
__RAFT_APPEND_ENTRY(r, 1, 1, "aaa");
/* receive an appendentry with commit */
msg_appendentries_t ae;
......@@ -1676,7 +1676,7 @@ void TestRaft_follower_recv_appendentries_failure_includes_current_idx(
/* try again with a higher current_idx */
memset(&aer, 0, sizeof(aer));
__APPEND_ENTRY(r, 2, 1, "aaa");
__RAFT_APPEND_ENTRY(r, 2, 1, "aaa");
raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer);
CuAssertTrue(tc, 0 == aer.success);
CuAssertIntEquals(tc, 2, aer.current_idx);
......@@ -1736,8 +1736,8 @@ void TestRaft_follower_dont_grant_vote_if_candidate_has_a_less_complete_log(
raft_set_current_term(r, 1);
/* server's idx are more up-to-date */
__APPEND_ENTRY(r, 100, 1, "aaa");
__APPEND_ENTRY(r, 101, 2, "aaa");
__RAFT_APPEND_ENTRY(r, 100, 1, "aaa");
__RAFT_APPEND_ENTRY(r, 101, 2, "aaa");
/* vote not granted */
raft_recv_requestvote(r, raft_get_node(r, 2), &rv, &rvr);
......@@ -2127,9 +2127,9 @@ void TestRaft_candidate_requestvote_includes_logidx(CuTest * tc)
raft_set_callbacks(r, &funcs, sender);
raft_set_current_term(r, 5);
/* 3 entries */
__APPEND_ENTRY(r, 100, 1, "aaa");
__APPEND_ENTRY(r, 101, 1, "aaa");
__APPEND_ENTRY(r, 102, 3, "aaa");
__RAFT_APPEND_ENTRY(r, 100, 1, "aaa");
__RAFT_APPEND_ENTRY(r, 101, 1, "aaa");
__RAFT_APPEND_ENTRY(r, 102, 3, "aaa");
raft_send_requestvote(r, raft_get_node(r, 2));
msg_requestvote_t* rv = sender_poll_msg_data(sender);
......@@ -2427,7 +2427,7 @@ void TestRaft_leader_sends_appendentries_with_leader_commit(
for (i=0; i<10; i++)
{
__APPEND_ENTRY(r, 1, 1, "aaa");
__RAFT_APPEND_ENTRY(r, 1, 1, "aaa");
}
raft_set_commit_idx(r, 10);
......@@ -2466,7 +2466,7 @@ void TestRaft_leader_sends_appendentries_with_prevLogIdx(
/* add 1 entry */
/* receive appendentries messages */
__APPEND_ENTRY(r, 100, 2, "aaa");
__RAFT_APPEND_ENTRY(r, 100, 2, "aaa");
raft_node_set_next_idx(n, 1);
raft_send_appendentries(r, raft_get_node(r, 2));
ae = sender_poll_msg_data(sender);
......@@ -2510,7 +2510,7 @@ void TestRaft_leader_sends_appendentries_when_node_has_next_idx_of_0(
/* receive appendentries messages */
raft_node_t* n = raft_get_node(r, 2);
raft_node_set_next_idx(n, 1);
__APPEND_ENTRY(r, 100, 1, "aaa");
__RAFT_APPEND_ENTRY(r, 100, 1, "aaa");
raft_send_appendentries(r, n);
ae = sender_poll_msg_data(sender);
CuAssertTrue(tc, NULL != ae);
......@@ -2626,7 +2626,7 @@ void TestRaft_leader_recv_appendentries_response_increase_commit_idx_when_majori
raft_set_last_applied_idx(r, 0);
/* append entries - we need two */
__APPEND_ENTRIES_SEQ_ID(r, 3, 1, 1, "aaa");
__RAFT_APPEND_ENTRIES_SEQ_ID(r, 3, 1, 1, "aaa");
memset(&aer, 0, sizeof(msg_appendentries_response_t));
......@@ -2696,7 +2696,7 @@ void TestRaft_leader_recv_appendentries_response_set_has_sufficient_logs_for_nod
raft_set_last_applied_idx(r, 0);
/* append entries - we need two */
__APPEND_ENTRIES_SEQ_ID(r, 3, 1, 1, "aaaa");
__RAFT_APPEND_ENTRIES_SEQ_ID(r, 3, 1, 1, "aaaa");
memset(&aer, 0, sizeof(msg_appendentries_response_t));
......@@ -2742,7 +2742,7 @@ void TestRaft_leader_recv_appendentries_response_increase_commit_idx_using_votin
raft_set_last_applied_idx(r, 0);
/* append entries - we need two */
__APPEND_ENTRY(r, 1, 1, "aaaa");
__RAFT_APPEND_ENTRY(r, 1, 1, "aaaa");
memset(&aer, 0, sizeof(msg_appendentries_response_t));
......@@ -2787,7 +2787,7 @@ void TestRaft_leader_recv_appendentries_response_duplicate_does_not_decrement_ma
raft_set_last_applied_idx(r, 0);
/* append entries - we need two */
__APPEND_ENTRIES_SEQ_ID(r, 3, 1, 1, "aaaa");
__RAFT_APPEND_ENTRIES_SEQ_ID(r, 3, 1, 1, "aaaa");
memset(&aer, 0, sizeof(msg_appendentries_response_t));
......@@ -2843,8 +2843,8 @@ void TestRaft_leader_recv_appendentries_response_do_not_increase_commit_idx_beca
raft_set_last_applied_idx(r, 0);
/* append entries - we need two */
__APPEND_ENTRIES_SEQ_ID(r, 2, 1, 1, "aaaa");
__APPEND_ENTRY(r, 3, 2, "aaaa");
__RAFT_APPEND_ENTRIES_SEQ_ID(r, 2, 1, 1, "aaaa");
__RAFT_APPEND_ENTRY(r, 3, 2, "aaaa");
memset(&aer, 0, sizeof(msg_appendentries_response_t));
......@@ -2919,7 +2919,7 @@ void TestRaft_leader_recv_appendentries_response_jumps_to_lower_next_idx(
raft_set_commit_idx(r, 0);
/* append entries */
__APPEND_ENTRIES_SEQ_ID_TERM(r, 4, 1, 1, "aaaa");
__RAFT_APPEND_ENTRIES_SEQ_ID_TERM(r, 4, 1, 1, "aaaa");
msg_appendentries_t* ae;
......@@ -2973,7 +2973,7 @@ void TestRaft_leader_recv_appendentries_response_decrements_to_lower_next_idx(
raft_set_commit_idx(r, 0);
/* append entries */
__APPEND_ENTRIES_SEQ_ID_TERM(r, 4, 1, 1, "aaaa");
__RAFT_APPEND_ENTRIES_SEQ_ID_TERM(r, 4, 1, 1, "aaaa");
msg_appendentries_t* ae;
......@@ -3042,7 +3042,7 @@ void TestRaft_leader_recv_appendentries_response_retry_only_if_leader(CuTest * t
raft_set_last_applied_idx(r, 0);
/* append entries - we need two */
__APPEND_ENTRY(r, 1, 1, "aaaa");
__RAFT_APPEND_ENTRY(r, 1, 1, "aaaa");
raft_send_appendentries(r, raft_get_node(r, 2));
raft_send_appendentries(r, raft_get_node(r, 3));
......@@ -3250,7 +3250,7 @@ void TestRaft_leader_recv_entry_does_not_send_new_appendentries_to_slow_nodes(Cu
raft_node_set_next_idx(raft_get_node(r, 2), 1);
/* append entries */
__APPEND_ENTRY(r, 1, 1, "aaaa");
__RAFT_APPEND_ENTRY(r, 1, 1, "aaaa");
/* entry message */
msg_entry_t *mety = __MAKE_ENTRY(1, 1, "entry");
......@@ -3286,7 +3286,7 @@ void TestRaft_leader_recv_appendentries_response_failure_does_not_set_node_nexti
raft_set_commit_idx(r, 0);
/* append entries */
__APPEND_ENTRY(r, 1, 1, "aaaa");
__RAFT_APPEND_ENTRY(r, 1, 1, "aaaa");
/* send appendentries -
* server will be waiting for response */
......
......@@ -428,9 +428,9 @@ void TestRaft_follower_load_from_snapshot_does_not_break_cluster_safety(CuTest *
CuAssertIntEquals(tc, 0, raft_get_log_count(r));
/* entry message */
__APPEND_ENTRY(r, 1, 1, "entry");
__APPEND_ENTRY(r, 2, 1, "entry");
__APPEND_ENTRY(r, 3, 1, "entry");
__RAFT_APPEND_ENTRY(r, 1, 1, "entry");
__RAFT_APPEND_ENTRY(r, 2, 1, "entry");
__RAFT_APPEND_ENTRY(r, 3, 1, "entry");
CuAssertIntEquals(tc, -1, raft_begin_load_snapshot(r, 2, 2));
}
......@@ -471,7 +471,7 @@ void TestRaft_leader_sends_appendentries_when_node_next_index_was_compacted(CuTe
raft_add_node(r, NULL, 3, 0);
/* entry 1 */
__APPEND_ENTRIES_SEQ_ID(r, 3, 1, 1, "aaa");
__RAFT_APPEND_ENTRIES_SEQ_ID(r, 3, 1, 1, "aaa");
CuAssertIntEquals(tc, 3, raft_get_current_idx(r));
/* compact entry 1 & 2 */
......@@ -700,13 +700,13 @@ void TestRaft_leader_sends_snapshot_if_log_was_compacted(CuTest* tc)
/* entry 1 */
char *str = "aaa";
__APPEND_ENTRY(r, 1, 1, str);
__RAFT_APPEND_ENTRY(r, 1, 1, str);
/* entry 2 */
__APPEND_ENTRY(r, 2, 1, str);
__RAFT_APPEND_ENTRY(r, 2, 1, str);
/* entry 3 */
__APPEND_ENTRY(r, 3, 1, str);
__RAFT_APPEND_ENTRY(r, 3, 1, str);
CuAssertIntEquals(tc, 3, raft_get_current_idx(r));
/* compact entry 1 & 2 */
......
......@@ -438,6 +438,7 @@ class Network(object):
ety = lib.raft_get_entry_from_idx(server.raft, ci)
try:
assert ety
lib.raft_entry_release(ety)
except Exception:
print('current idx ', ci)
print('count', lib.raft_get_log_count(server.raft))
......@@ -799,6 +800,7 @@ class RaftServer(object):
logger.error(self.debug_log())
logger.error(server.debug_log())
return lib.RAFT_ERR_SHUTDOWN
lib.raft_entry_release(their_log)
def entry_apply(self, ety, idx):
# collect stats
......@@ -971,6 +973,7 @@ class RaftServer(object):
assert prev_ety
other_id = prev_ety.id
assert other_id < ety.id
lib.raft_entry_release(prev_ety)
except Exception as e:
logger.error(other_id, ety.id)
self.abort_exception = e
......
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