Commit 475f897b authored by Willem Thiart's avatar Willem Thiart
Browse files

Added raft_msg_entry_response_committed

parent 5752034b
...@@ -11,7 +11,8 @@ ...@@ -11,7 +11,8 @@
#ifndef RAFT_H_ #ifndef RAFT_H_
#define RAFT_H_ #define RAFT_H_
typedef struct { typedef struct
{
void *buf; void *buf;
unsigned int len; unsigned int len;
...@@ -39,8 +40,11 @@ typedef struct ...@@ -39,8 +40,11 @@ typedef struct
/** the entry's unique ID */ /** the entry's unique ID */
unsigned int id; unsigned int id;
/** whether or not the entry was committed */ /** the entry's term */
int was_committed; int term;
/** the entry's index */
int idx;
} msg_entry_response_t; } msg_entry_response_t;
/** Vote request message. /** Vote request message.
...@@ -196,9 +200,9 @@ typedef int ( ...@@ -196,9 +200,9 @@ typedef int (
*func_applylog_f *func_applylog_f
) ( ) (
raft_server_t* raft, raft_server_t* raft,
void *udata, void *user_data,
const unsigned char *data, const unsigned char *log_data,
const int len const int log_len
); );
/** Callback for saving who we voted for to disk. /** Callback for saving who we voted for to disk.
...@@ -511,4 +515,9 @@ void raft_set_current_term(raft_server_t* me, const int term); ...@@ -511,4 +515,9 @@ void raft_set_current_term(raft_server_t* me, const int term);
* @param[in] ety The entry to be appended */ * @param[in] ety The entry to be appended */
int raft_append_entry(raft_server_t* me, raft_entry_t* ety); int raft_append_entry(raft_server_t* me, raft_entry_t* ety);
/** 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_,
const msg_entry_response_t* r);
#endif /* RAFT_H_ */ #endif /* RAFT_H_ */
...@@ -21,6 +21,9 @@ ...@@ -21,6 +21,9 @@
#include "raft_log.h" #include "raft_log.h"
#include "raft_private.h" #include "raft_private.h"
#define min(a, b) ((a) < (b) ? (a) : (b))
#define max(a, b) ((a) < (b) ? (b) : (a))
static void __log(raft_server_t *me_, const char *fmt, ...) static void __log(raft_server_t *me_, const char *fmt, ...)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
...@@ -181,7 +184,6 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -181,7 +184,6 @@ int raft_recv_appendentries_response(raft_server_t* me_,
/* If AppendEntries fails because of log inconsistency: /* If AppendEntries fails because of log inconsistency:
decrement nextIndex and retry (§5.3) */ decrement nextIndex and retry (§5.3) */
assert(0 <= raft_node_get_next_idx(p)); assert(0 <= raft_node_get_next_idx(p));
// TODO does this have test coverage?
// TODO can jump back to where node is different instead of iterating // TODO can jump back to where node is different instead of iterating
raft_node_set_next_idx(p, raft_node_get_next_idx(p) - 1); raft_node_set_next_idx(p, raft_node_get_next_idx(p) - 1);
...@@ -280,27 +282,26 @@ int raft_recv_appendentries( ...@@ -280,27 +282,26 @@ int raft_recv_appendentries(
r->success = 0; r->success = 0;
return 0; return 0;
} }
}
/* 3. If an existing entry conflicts with a new one (same index /* 3. If an existing entry conflicts with a new one (same index
but different terms), delete the existing entry and all that but different terms), delete the existing entry and all that
follow it (§5.3) */ follow it (§5.3) */
raft_entry_t* e2 = raft_get_entry_from_idx(me_, ae->prev_log_idx + 1); raft_entry_t* e2 = raft_get_entry_from_idx(me_, ae->prev_log_idx + 1);
if (e2) if (e2)
log_delete(me->log, ae->prev_log_idx + 1); {
log_delete(me->log, ae->prev_log_idx + 1);
raft_set_current_idx(me_, ae->prev_log_idx);
} }
/* 5. If leaderCommit > commitIndex, set commitIndex = /* 4. If leaderCommit > commitIndex, set commitIndex =
min(leaderCommit, last log index) */ min(leaderCommit, last log index) */
if (raft_get_commit_idx(me_) < ae->leader_commit) if (raft_get_commit_idx(me_) < ae->leader_commit)
{ {
raft_entry_t* e = log_peektail(me->log); int last_log_idx = max(raft_get_current_idx(me_) - 1, 1);
if (e) raft_set_commit_idx(me_, min(last_log_idx, ae->leader_commit));
{ while (0 == raft_apply_entry(me_))
int id = e->id < ae->leader_commit ? e->id : ae->leader_commit; ;
raft_set_commit_idx(me_, id);
while (0 == raft_apply_entry(me_))
;
}
} }
if (raft_is_candidate(me_)) if (raft_is_candidate(me_))
...@@ -407,8 +408,9 @@ int raft_recv_requestvote_response(raft_server_t* me_, int node, ...@@ -407,8 +408,9 @@ int raft_recv_requestvote_response(raft_server_t* me_, int node,
assert(node < me->num_nodes); assert(node < me->num_nodes);
// if (r->term != raft_get_current_term(me_)) // TODO: if invalid leader then stepdown
// return 0; // if (r->term != raft_get_current_term(me_))
// return 0;
if (1 == r->vote_granted) if (1 == r->vote_granted)
{ {
...@@ -438,13 +440,14 @@ int raft_recv_entry(raft_server_t* me_, int node, msg_entry_t* e, ...@@ -438,13 +440,14 @@ int raft_recv_entry(raft_server_t* me_, int node, msg_entry_t* e,
ety.data.len = e->data.len; ety.data.len = e->data.len;
ety.data.buf = malloc(e->data.len); ety.data.buf = malloc(e->data.len);
memcpy(ety.data.buf, e->data.buf, e->data.len); memcpy(ety.data.buf, e->data.buf, e->data.len);
int res = raft_append_entry(me_, &ety); raft_append_entry(me_, &ety);
for (i = 0; i < me->num_nodes; i++) for (i = 0; i < me->num_nodes; i++)
if (me->nodeid != i) if (me->nodeid != i)
raft_send_appendentries(me_, i); raft_send_appendentries(me_, i);
r->id = e->id; r->id = e->id;
r->was_committed = (0 == res); r->idx = me->current_idx;
r->term = me->current_term;
return 0; return 0;
} }
...@@ -614,3 +617,16 @@ void raft_vote(raft_server_t* me_, const int node) ...@@ -614,3 +617,16 @@ void raft_vote(raft_server_t* me_, const int node)
if (me->cb.persist_vote) if (me->cb.persist_vote)
me->cb.persist_vote(me_, me->udata, node); me->cb.persist_vote(me_, me->udata, node);
} }
int raft_msg_entry_response_committed(raft_server_t* me_,
const msg_entry_response_t* r)
{
raft_entry_t* ety = raft_get_entry_from_idx(me_, r->idx);
if (!ety)
return 0;
/* entry from another leader has invalidated this entry message */
if (r->term != ety->term)
return -1;
return r->idx <= raft_get_commit_idx(me_);
}
...@@ -1221,9 +1221,6 @@ void TestRaft_leader_responds_to_entry_msg_when_entry_is_committed(CuTest * tc) ...@@ -1221,9 +1221,6 @@ void TestRaft_leader_responds_to_entry_msg_when_entry_is_committed(CuTest * tc)
/* trigger response through commit */ /* trigger response through commit */
raft_apply_entry(r); raft_apply_entry(r);
/* leader sent response to entry message */
CuAssertTrue(tc, 1 == cr.was_committed);
} }
void TestRaft_non_leader_recv_entry_msg_fails(CuTest * tc) void TestRaft_non_leader_recv_entry_msg_fails(CuTest * tc)
...@@ -1538,6 +1535,79 @@ TestRaft_leader_recv_appendentries_response_increase_commit_idx_when_majority_ha ...@@ -1538,6 +1535,79 @@ TestRaft_leader_recv_appendentries_response_increase_commit_idx_when_majority_ha
CuAssertTrue(tc, 2 == raft_get_last_applied_idx(r)); CuAssertTrue(tc, 2 == raft_get_last_applied_idx(r));
} }
void TestRaft_leader_recv_entry_is_committed_returns_0_if_not_committed(CuTest * tc)
{
void *r = raft_new();
raft_add_peer(r, (void*)1, 1);
raft_add_peer(r, (void*)2, 0);
raft_set_state(r, RAFT_STATE_LEADER);
raft_set_current_term(r, 1);
raft_set_commit_idx(r, 0);
/* entry message */
msg_entry_t mety;
mety.id = 1;
mety.data.buf = "entry";
mety.data.len = strlen("entry");
/* receive entry */
msg_entry_response_t cr;
raft_recv_entry(r, 1, &mety, &cr);
CuAssertTrue(tc, 0 == raft_msg_entry_response_committed(r, &cr));
raft_set_commit_idx(r, 1);
CuAssertTrue(tc, 1 == raft_msg_entry_response_committed(r, &cr));
}
void TestRaft_leader_recv_entry_is_committed_returns_neg_1_if_invalidated(CuTest * tc)
{
void *r = raft_new();
raft_add_peer(r, (void*)1, 1);
raft_add_peer(r, (void*)2, 0);
raft_set_state(r, RAFT_STATE_LEADER);
raft_set_current_term(r, 1);
raft_set_commit_idx(r, 0);
/* entry message */
msg_entry_t mety;
mety.id = 1;
mety.data.buf = "entry";
mety.data.len = strlen("entry");
/* receive entry */
msg_entry_response_t cr;
raft_recv_entry(r, 1, &mety, &cr);
CuAssertTrue(tc, 0 == raft_msg_entry_response_committed(r, &cr));
CuAssertTrue(tc, cr.term == 1);
CuAssertTrue(tc, cr.idx == 1);
CuAssertTrue(tc, 1 == raft_get_current_idx(r));
CuAssertTrue(tc, 0 == raft_get_commit_idx(r));
/* append entry that invalidates entry message */
msg_appendentries_t ae;
memset(&ae, 0, sizeof(msg_appendentries_t));
ae.leader_commit = 1;
ae.term = 2;
ae.prev_log_idx = 0;
ae.prev_log_term = 0;
msg_appendentries_response_t aer;
msg_entry_t e[1];
memset(&e, 0, sizeof(msg_entry_t) * 1);
e[0].term = 2;
e[0].id = 999;
e[0].data.buf = "aaa";
e[0].data.len = 3;
ae.entries = e;
ae.n_entries = 1;
raft_recv_appendentries(r, 1, &ae, &aer);
CuAssertTrue(tc, 1 == aer.success);
CuAssertTrue(tc, 1 == raft_get_current_idx(r));
CuAssertTrue(tc, 1 == raft_get_commit_idx(r));
CuAssertTrue(tc, -1 == raft_msg_entry_response_committed(r, &cr));
}
void TestRaft_leader_recv_appendentries_response_failure_does_not_set_node_nextid_to_0( void TestRaft_leader_recv_appendentries_response_failure_does_not_set_node_nextid_to_0(
CuTest * tc) CuTest * tc)
{ {
......
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