Commit 816e8dd1 authored by Willem Thiart's avatar Willem Thiart
Browse files

Added leader_commit to append entries

parent 1ac903b0
...@@ -107,6 +107,20 @@ typedef enum ...@@ -107,6 +107,20 @@ typedef enum
RAFT_MSG_ENTRY_RESPONSE, RAFT_MSG_ENTRY_RESPONSE,
} raft_message_type_e; } raft_message_type_e;
typedef struct
{
/* entry's term */
unsigned int term;
/* the entry's unique ID */
unsigned int id;
/* entry data */
unsigned char* data;
/* length of entry data */
unsigned int len;
/* number of nodes that have this entry */
unsigned int num_nodes;
} raft_entry_t;
/** /**
* @param raft The Raft server making this callback * @param raft The Raft server making this callback
* @param udata User data that is passed from Raft server * @param udata User data that is passed from Raft server
...@@ -163,28 +177,59 @@ typedef int ( ...@@ -163,28 +177,59 @@ typedef int (
const int len const int len
); );
/**
* Save who we voted for to disk
* @param raft The Raft server making this callback
* @param udata User data that is passed from Raft server
* @param voted_for The node we voted for
* @return 0 on success */
typedef int (
*func_persist_vote_f
) (
raft_server_t* raft,
void *udata,
const int voted_for
);
/**
* Save the term we are on to disk
* @param raft The Raft server making this callback
* @param udata User data that is passed from Raft server
* @param term Current term
* @return 0 on success */
typedef int (
*func_persist_current_term_f
) (
raft_server_t* raft,
void *udata,
const int current_term
);
/**
* Save log to disk
* @param raft The Raft server making this callback
* @param udata User data that is passed from Raft server
* @param term Current term
* @return 0 on success */
typedef int (
*func_persist_entry_f
) (
raft_server_t* raft,
void *udata,
const raft_entry_t entry
);
typedef struct typedef struct
{ {
func_send_requestvote_f send_requestvote; func_send_requestvote_f send_requestvote;
func_send_appendentries_f send_appendentries; func_send_appendentries_f send_appendentries;
func_log_f log; func_log_f log;
func_applylog_f applylog; func_applylog_f applylog;
func_persist_vote_f persist_vote;
func_persist_current_term_f persist_current_term;
func_persist_entry_f persist_entry;
} raft_cbs_t; } raft_cbs_t;
typedef struct
{
/* entry's term */
unsigned int term;
/* the entry's unique ID */
unsigned int id;
/* entry data */
unsigned char* data;
/* length of entry data */
unsigned int len;
/* number of nodes that have this entry */
unsigned int num_nodes;
} raft_entry_t;
/** /**
* Initialise a new Raft server * Initialise a new Raft server
* *
......
...@@ -23,12 +23,16 @@ ...@@ -23,12 +23,16 @@
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_;
char buf[1024]; char buf[1024];
va_list args; va_list args;
va_start(args, fmt); va_start(args, fmt);
vsprintf(buf, fmt, args); vsprintf(buf, fmt, args);
if (me->cb.log)
me->cb.log(me_, me->udata, buf);
#if 0 /* debugging */ #if 0 /* debugging */
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
printf("%d: %s\n", me->nodeid, buf); printf("%d: %s\n", me->nodeid, buf);
...@@ -477,6 +481,7 @@ void raft_send_appendentries(raft_server_t* me_, int node) ...@@ -477,6 +481,7 @@ void raft_send_appendentries(raft_server_t* me_, int node)
ae.term = me->current_term; ae.term = me->current_term;
ae.leader_id = me->nodeid; ae.leader_id = me->nodeid;
ae.prev_log_term = raft_node_get_next_idx(p); ae.prev_log_term = raft_node_get_next_idx(p);
ae.leader_commit = raft_get_commit_idx(me_);
// TODO: // TODO:
ae.prev_log_idx = 0; ae.prev_log_idx = 0;
ae.n_entries = 0; ae.n_entries = 0;
......
...@@ -1414,6 +1414,31 @@ void TestRaft_leader_sends_appendentries_with_NextIdx_when_PrevIdx_gt_NextIdx( ...@@ -1414,6 +1414,31 @@ void TestRaft_leader_sends_appendentries_with_NextIdx_when_PrevIdx_gt_NextIdx(
CuAssertTrue(tc, NULL != ae); CuAssertTrue(tc, NULL != ae);
} }
void TestRaft_leader_sends_appendentries_with_leader_commit(
CuTest * tc)
{
raft_cbs_t funcs = {
.send_appendentries = sender_appendentries,
.log = NULL
};
void *sender = sender_new(NULL);
void *r = raft_new();
raft_set_callbacks(r, &funcs, sender);
raft_add_peer(r, (void*)1, 1);
raft_add_peer(r, (void*)2, 0);
/* i'm leader */
raft_set_state(r, RAFT_STATE_LEADER);
raft_set_commit_idx(r, 10);
/* receive appendentries messages */
raft_send_appendentries(r, 0);
msg_appendentries_t* ae = sender_poll_msg_data(sender);
CuAssertTrue(tc, NULL != ae);
CuAssertTrue(tc, ae->leader_commit == 10);
}
/* 5.3 */ /* 5.3 */
void void
TestRaft_leader_retries_appendentries_with_decremented_NextIdx_log_inconsistency( TestRaft_leader_retries_appendentries_with_decremented_NextIdx_log_inconsistency(
......
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