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

Simplify entry removal (#130)

Simplied pop callback.

Previously, libraft was passing a callback(raft_entry_notify_f) to the pop callback.
The application was supposed to call that callback for each removed entry.

Libraft can loop over the entries and call the required functions itself.
So, we can delete raft_entry_notify_f from the pop callback just to
simplify API.

In addition to that, this PR removes raft_pop_entry() function which doesn't have
a use-case, as far as I can see.
We can use raft_delete_entry_from_idx(me, raft_get_current_idx()) instead if required.
parent 2279a960
......@@ -726,24 +726,6 @@ typedef struct
raft_timestamp_f timestamp;
} raft_cbs_t;
/** A generic notification callback used to allow Raft to notify caller
* on certain log operations.
*
* @param[in] arg Argument passed by Raft in the original call.
* @param[in] entry Entry for which notification is generated.
* @param[in] entry_idx Index of entry.
*
* The callback *must not* modify the entry or perform any preemptive
* log operation until it returns.
*/
typedef void (
*raft_entry_notify_f
) (
void* arg,
raft_entry_t *entry,
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.
......@@ -852,13 +834,11 @@ typedef struct raft_log_impl
*
* @param[in] from_idx Index of first entry to be removed. All entries
* starting from and including this index shall be removed.
* @param[in] cb Optional callback to execute for every removed entry.
* @param[in] cb_arg Argument to pass to callback.
* @return
* 0 on success;
* -1 on error.
*/
int (*pop) (void *log, raft_index_t from_idx, raft_entry_notify_f cb, void *cb_arg);
int (*pop) (void *log, raft_index_t from_idx);
/** Get a single entry from the log.
*
......@@ -1262,13 +1242,6 @@ void raft_set_commit_idx(raft_server_t *me, raft_index_t commit_idx);
* RAFT_ERR_NOMEM memory allocation failure */
int raft_append_entry(raft_server_t* me, raft_entry_t* ety);
/** Remove the last entry from the server's log.
* This should only be used when reloading persistent state from an append log
* store, where removed entries are still in the log but followed by a pop
* action.
*/
int raft_pop_entry(raft_server_t* me);
/** Confirm if a msg_entry_response has been committed.
* @param[in] r The response we want to check */
int raft_msg_entry_response_committed(raft_server_t* me,
......
......@@ -214,10 +214,7 @@ raft_index_t raft_log_count(raft_log_t *me)
return me->count;
}
static int log_delete(raft_log_t *me,
raft_index_t idx,
raft_entry_notify_f cb,
void *cb_arg)
static int log_delete(raft_log_t *me, raft_index_t idx)
{
if (idx == 0) {
return -1;
......@@ -239,10 +236,6 @@ static int log_delete(raft_log_t *me,
}
}
if (cb) {
cb(cb_arg, me->entries[back], idx_tmp);
}
raft_entry_release(me->entries[back]);
me->back = back;
......@@ -254,7 +247,7 @@ static int log_delete(raft_log_t *me,
int raft_log_delete(raft_log_t *me, raft_index_t idx)
{
return log_delete(me, idx, NULL, NULL);
return log_delete(me, idx);
}
int raft_log_poll(raft_log_t *me, raft_entry_t **etyp)
......@@ -398,10 +391,9 @@ static raft_index_t log_get_batch(void *log,
return n;
}
static int log_pop(void *log, raft_index_t from_idx,
raft_entry_notify_f cb, void *cb_arg)
static int log_pop(void *log, raft_index_t from_idx)
{
return log_delete(log, from_idx, cb, cb_arg);
return log_delete(log, from_idx);
}
static int log_poll(void *log, raft_index_t first_idx)
......
......@@ -355,17 +355,41 @@ void raft_handle_remove_cfg_change(raft_server_t* me, raft_entry_t* ety, const r
}
}
int raft_delete_entry_from_idx(raft_server_t* me, raft_index_t idx)
int raft_delete_entry_from_idx(raft_server_t *me, raft_index_t idx)
{
assert(me->commit_idx < idx);
if (idx <= me->voting_cfg_change_log_idx)
me->voting_cfg_change_log_idx = -1;
if (me->log_impl->count(me->log) == 0) {
return 0;
}
int e = me->log_impl->pop(me->log, idx,
(raft_entry_notify_f) raft_handle_remove_cfg_change, me);
if (e != 0)
return e;
raft_index_t first = me->log_impl->first_idx(me->log);
raft_index_t last = me->log_impl->current_idx(me->log);
if (idx < first || idx > last) {
return 0;
}
/* Delete entries starting from the last one. */
while (last >= idx) {
raft_entry_t *ety = raft_get_entry_from_idx(me, last);
int e = me->log_impl->pop(me->log, last);
if (e != 0) {
me->log_impl->sync(me->log);
raft_entry_release(ety);
return e;
}
raft_handle_remove_cfg_change(me, ety, last);
raft_entry_release(ety);
if (me->voting_cfg_change_log_idx == last) {
me->voting_cfg_change_log_idx = -1;
}
last--;
}
return me->log_impl->sync(me->log);
}
......@@ -1693,18 +1717,6 @@ int raft_entry_is_cfg_change(raft_entry_t* ety)
RAFT_LOGTYPE_REMOVE_NODE == ety->type);
}
int raft_pop_entry(raft_server_t* me)
{
raft_index_t cur_idx = me->log_impl->current_idx(me->log);
int e = me->log_impl->pop(me->log, cur_idx,
(raft_entry_notify_f) raft_handle_remove_cfg_change, me);
if (e != 0)
return e;
return me->log_impl->sync(me->log);
}
raft_index_t raft_get_first_entry_idx(raft_server_t* me)
{
assert(0 < raft_get_current_idx(me));
......
......@@ -141,16 +141,17 @@ void TestLogImpl_pop(CuTest * tc)
CuAssertIntEquals(tc, 3, impl->count(l));
CuAssertIntEquals(tc, 3, impl->current_idx(l));
impl->pop(l, 3, event_entry_enqueue, queue);
event_entry_enqueue(queue, impl->get(l, 3), 3);
impl->pop(l, 3);
CuAssertIntEquals(tc, 2, impl->count(l));
CuAssertIntEquals(tc, 3, ((raft_entry_t*)llqueue_poll(queue))->id);
CuAssertTrue(tc, NULL == impl->get(l, 3));
impl->pop(l, 2, NULL, NULL);
impl->pop(l, 2);
CuAssertIntEquals(tc, 1, impl->count(l));
CuAssertTrue(tc, NULL == impl->get(l, 2));
impl->pop(l, 1, NULL, NULL);
impl->pop(l, 1);
CuAssertIntEquals(tc, 0, impl->count(l));
CuAssertTrue(tc, NULL == impl->get(l, 1));
}
......@@ -166,7 +167,7 @@ void TestLogImpl_pop_onwards(CuTest * tc)
CuAssertIntEquals(tc, 3, impl->count(l));
/* even 3 gets deleted */
impl->pop(l, 2, NULL, NULL);
impl->pop(l, 2);
CuAssertIntEquals(tc, 1, impl->count(l));
CuAssertIntEquals(tc, 1, impl->get(l, 1)->id);
CuAssertTrue(tc, NULL == impl->get(l, 2));
......@@ -182,7 +183,7 @@ void TestLogImpl_pop_fails_for_idx_zero(CuTest * tc)
l = impl->init(r, NULL);
__LOGIMPL_APPEND_ENTRIES_SEQ_ID(l, 4, 1, 0, NULL);
CuAssertIntEquals(tc, impl->pop(l, 0, NULL, NULL), -1);
CuAssertIntEquals(tc, impl->pop(l, 0), -1);
}
void TestLogImpl_poll(CuTest * tc)
......@@ -290,7 +291,7 @@ void TestLogImpl_pop_after_polling(CuTest * tc)
CuAssertIntEquals(tc, 2, impl->current_idx(l));
/* poll */
CuAssertIntEquals(tc, impl->pop(l, 1, NULL, NULL), 0);
CuAssertIntEquals(tc, impl->pop(l, 1), 0);
CuAssertIntEquals(tc, 0, impl->count(l));
CuAssertIntEquals(tc, 1, impl->current_idx(l));
}
......@@ -316,7 +317,7 @@ void TestLogImpl_pop_after_polling_from_double_append(CuTest * tc)
CuAssertIntEquals(tc, 2, impl->count(l));
/* pop */
CuAssertIntEquals(tc, impl->pop(l, 1, NULL, NULL), 0);
CuAssertIntEquals(tc, impl->pop(l, 1), 0);
CuAssertIntEquals(tc, 0, impl->count(l));
}
......
......@@ -4931,6 +4931,60 @@ void TestRaft_rebuild_config_after_restart(CuTest *tc)
}
}
void TestRaft_delete_configuration_change_entries(CuTest *tc)
{
/* Delete configuration change entries and verify the configuration is
* rolled back correctly. */
raft_cbs_t funcs = {
.get_node_id = __raft_get_node_id
};
void *r = raft_new();
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_set_callbacks(r, &funcs, NULL);
raft_set_current_term(r, 1);
raft_set_state(r, RAFT_STATE_LEADER);
/* If there is no entry, delete should return immediately. */
CuAssertIntEquals(tc, 0, raft_delete_entry_from_idx(r, 2));
/* Add the non-voting node. */
raft_entry_t *ety = __MAKE_ENTRY(1, 1, "3");
ety->type = RAFT_LOGTYPE_ADD_NONVOTING_NODE;
CuAssertIntEquals(tc, 0, raft_recv_entry(r, ety, NULL));
/* If there idx is out of bounds, delete should return immediately. */
CuAssertIntEquals(tc, 0, raft_delete_entry_from_idx(r, 2));
/* Append a removal log entry for the non-voting node we just added. */
ety = __MAKE_ENTRY(1, 1, "3");
ety->type = RAFT_LOGTYPE_REMOVE_NODE;
CuAssertIntEquals(tc, 0, raft_recv_entry(r, ety, NULL));
/* Another configuration change entry should be rejected. */
ety = __MAKE_ENTRY(1, 1, "4");
ety->type = RAFT_LOGTYPE_ADD_NODE;
int err = raft_recv_entry(r, ety, NULL);
CuAssertIntEquals(tc, RAFT_ERR_ONE_VOTING_CHANGE_ONLY, err);
/* Add some random entries. */
__RAFT_APPEND_ENTRIES_SEQ_ID(r, 10, 1, 1, "data");
CuAssertIntEquals(tc, 12, raft_get_log_count(r));
/* Delete the removal log entry and others after it. */
CuAssertIntEquals(tc, 0, raft_delete_entry_from_idx(r, 2));
CuAssertIntEquals(tc, 1, raft_get_log_count(r));
CuAssertIntEquals(tc, 1, raft_node_is_active(raft_get_node(r, 3)));
CuAssertIntEquals(tc, 0, raft_node_is_voting(raft_get_node(r, 3)));
/* Configuration change entry should be accepted now. */
CuAssertIntEquals(tc, 0, raft_recv_entry(r, ety, NULL));
}
int main(void)
{
CuString *output = CuStringNew();
......@@ -5083,6 +5137,7 @@ int main(void)
SUITE_ADD_TEST(suite, TestRaft_apply_read_request_timeout);
SUITE_ADD_TEST(suite, TestRaft_test_metadata_on_restart);
SUITE_ADD_TEST(suite, TestRaft_rebuild_config_after_restart);
SUITE_ADD_TEST(suite, TestRaft_delete_configuration_change_entries);
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