Commit d106d229 authored by Willem Thiart's avatar Willem Thiart
Browse files

Add RAFT_ERR constants

This change is backwards incompatible. The following functions return
different return codes:

* applylog
* log_offer
* raft_append_entry
* raft_apply_all
* raft_recv_entry
parent f1b9f401
......@@ -10,6 +10,10 @@
#ifndef RAFT_H_
#define RAFT_H_
#define RAFT_ERR_NOT_LEADER -2
#define RAFT_ERR_ONE_VOTING_CHANGE_ONLY -3
#define RAFT_ERR_SHUTDOWN -4
typedef enum {
RAFT_STATE_NONE,
RAFT_STATE_FOLLOWER,
......@@ -271,7 +275,9 @@ typedef struct
/** Callback for sending appendentries messages */
func_send_appendentries_f send_appendentries;
/** Callback for finite state machine application */
/** Callback for finite state machine application
* Return 0 on success.
* Return RAFT_ERR_SHUTDOWN if you want the server to shutdown. */
func_applylog_f applylog;
/** Callback for persisting vote data
......@@ -283,7 +289,9 @@ typedef struct
func_persist_int_f persist_term;
/** Callback for adding an entry to the log
* For safety reasons this callback MUST flush the change to disk. */
* For safety reasons this callback MUST flush the change to disk.
* Return 0 on success.
* Return RAFT_ERR_SHUTDOWN if you want the server to shutdown. */
func_logentry_event_f log_offer;
/** Callback for removing the oldest entry from the log
......@@ -452,7 +460,12 @@ int raft_recv_requestvote_response(raft_server_t* me,
* @param[in] node Index of the node who sent us this message
* @param[in] ety The entry message
* @param[out] r The resulting response
* @return 0 on success, -1 on failure */
* @return
* 0 on success;
* RAFT_ERR_NOT_LEADER server is not the leader;
* RAFT_ERR_SHUTDOWN server should be shutdown;
* RAFT_ERR_ONE_VOTING_CHANGE_ONLY there is a non-voting change inflight;
*/
int raft_recv_entry(raft_server_t* me,
msg_entry_t* ety,
msg_entry_response_t *r);
......@@ -593,7 +606,10 @@ void raft_set_commit_idx(raft_server_t* me, int commit_idx);
/** Add an entry to the server's log.
* This should be used to reload persistent state, ie. the commit log.
* @param[in] ety The entry to be appended */
* @param[in] ety The entry to be appended
* @return
* 0 on success;
* RAFT_ERR_SHUTDOWN server should shutdown */
int raft_append_entry(raft_server_t* me, raft_entry_t* ety);
/** Confirm if a msg_entry_response has been committed.
......@@ -630,8 +646,11 @@ void raft_node_set_voting(raft_node_t* node, int voting);
* @return 1 if this is a voting node. Otherwise 0. */
int raft_node_is_voting(raft_node_t* me_);
/** Apply all entries up to the commit index */
void raft_apply_all(raft_server_t* me_);
/** Apply all entries up to the commit index
* @return
* 0 on success;
* RAFT_ERR_SHUTDOWN when server should be shutdown */
int raft_apply_all(raft_server_t* me_);
/** Become leader
* WARNING: this is a dangerous function call. It could lead to your cluster
......
......@@ -105,11 +105,12 @@ int log_append_entry(log_t* me_, raft_entry_t* c)
__ensurecapacity(me);
if (me->cb && me->cb->log_offer)
me->cb->log_offer(me->raft, raft_get_udata(me->raft), c, me->back);
memcpy(&me->entries[me->back], c, sizeof(raft_entry_t));
me->count++;
me->back++;
if (me->cb && me->cb->log_offer)
return me->cb->log_offer(me->raft, raft_get_udata(me->raft), c, me->back - 1);
return 0;
}
......
......@@ -384,7 +384,12 @@ int raft_recv_appendentries(
int e = raft_append_entry(me_, &ae->entries[i]);
if (-1 == e)
goto fail_with_current_idx;
else if (RAFT_ERR_SHUTDOWN == e)
{
r->success = 0;
r->first_idx = 0;
return RAFT_ERR_SHUTDOWN;
}
r->current_idx = ae->prev_log_idx + 1 + i;
}
......@@ -543,10 +548,10 @@ int raft_recv_entry(raft_server_t* me_,
/* Only one voting cfg change at a time */
if (raft_entry_is_voting_cfg_change(e))
if (-1 != me->voting_cfg_change_log_idx)
return -1;
return RAFT_ERR_ONE_VOTING_CHANGE_ONLY;
if (!raft_is_leader(me_))
return -1;
return RAFT_ERR_NOT_LEADER;
__log(me_, NULL, "received entry t:%d id: %d idx: %d",
me->current_term, e->id, raft_get_current_idx(me_) + 1);
......@@ -624,16 +629,20 @@ int raft_apply_entry(raft_server_t* me_)
int log_idx = me->last_applied_idx + 1;
raft_entry_t* e = raft_get_entry_from_idx(me_, log_idx);
if (!e)
raft_entry_t* ety = raft_get_entry_from_idx(me_, log_idx);
if (!ety)
return -1;
__log(me_, NULL, "applying log: %d, id: %d size: %d",
me->last_applied_idx, e->id, e->data.len);
me->last_applied_idx, ety->id, ety->data.len);
me->last_applied_idx++;
if (me->cb.applylog)
me->cb.applylog(me_, me->udata, e);
{
int e = me->cb.applylog(me_, me->udata, ety);
if (RAFT_ERR_SHUTDOWN == e)
return RAFT_ERR_SHUTDOWN;
}
/* voting cfg change is now complete */
if (log_idx == me->voting_cfg_change_log_idx)
......@@ -796,10 +805,16 @@ int raft_msg_entry_response_committed(raft_server_t* me_,
return r->idx <= raft_get_commit_idx(me_);
}
void raft_apply_all(raft_server_t* me_)
int raft_apply_all(raft_server_t* me_)
{
while (raft_get_last_applied_idx(me_) < raft_get_commit_idx(me_))
raft_apply_entry(me_);
{
int e = raft_apply_entry(me_);
if (RAFT_ERR_SHUTDOWN == e)
return RAFT_ERR_SHUTDOWN;
}
return 0;
}
int raft_entry_is_voting_cfg_change(raft_entry_t* ety)
......
......@@ -166,29 +166,11 @@ void TestRaft_server_append_entry_means_entry_gets_current_term(CuTest* tc)
CuAssertTrue(tc, 1 == raft_get_current_idx(r));
}
static int __raft_logentry_offer(
raft_server_t* raft,
void *udata,
raft_entry_t *ety,
int ety_idx
)
{
ety->data.buf = udata;
return 0;
}
void TestRaft_server_append_entry_is_retrievable(CuTest * tc)
{
raft_cbs_t funcs = {
.log_offer = __raft_logentry_offer,
};
char *data = "xxx";
void *r = raft_new();
raft_set_state(r, RAFT_STATE_CANDIDATE);
raft_set_callbacks(r, &funcs, data);
raft_set_current_term(r, 5);
raft_entry_t ety = {};
ety.term = 1;
......@@ -407,7 +389,7 @@ void TestRaft_server_recv_entry_fails_if_there_is_already_a_voting_change(CuTest
CuAssertTrue(tc, 1 == raft_get_log_count(r));
ety.id = 2;
CuAssertTrue(tc, -1 == raft_recv_entry(r, &ety, &cr));
CuAssertTrue(tc, RAFT_ERR_ONE_VOTING_CHANGE_ONLY == raft_recv_entry(r, &ety, &cr));
CuAssertTrue(tc, 1 == raft_get_commit_idx(r));
}
......@@ -1655,7 +1637,7 @@ void TestRaft_non_leader_recv_entry_msg_fails(CuTest * tc)
/* receive entry */
int e = raft_recv_entry(r, &ety, &cr);
CuAssertTrue(tc, -1 == e);
CuAssertTrue(tc, RAFT_ERR_NOT_LEADER == e);
}
/* 5.3 */
......
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