Commit 5752034b authored by Willem Thiart's avatar Willem Thiart
Browse files

Improved documentation

parent 8aeb24cf
...@@ -11,119 +11,140 @@ ...@@ -11,119 +11,140 @@
#ifndef RAFT_H_ #ifndef RAFT_H_
#define RAFT_H_ #define RAFT_H_
typedef struct
{
/** User data pointer for addressing.
* Examples of what this could be:
* - void* pointing to implementor's networking data
* - a (IP,Port) tuple */
void* udata_address;
} raft_node_configuration_t;
typedef struct
{
/* candidate's term */
int term;
/* candidate requesting vote */
int candidate_id;
/* idx of candidate's last log entry */
int last_log_idx;
/* term of candidate's last log entry */
int last_log_term;
} msg_requestvote_t;
typedef struct { typedef struct {
void *buf; void *buf;
unsigned int len; unsigned int len;
} msg_entry_data_t; } raft_entry_data_t;
/** Message sent from client to server.
* The client sends this message to a server with the intention of having it
* applied to the FSM. */
typedef struct typedef struct
{ {
/* the entry's unique ID */ /** the term the entry was created */
unsigned int id;
/* the term the entry was created */
int term; int term;
msg_entry_data_t data; /** the entry's unique ID
* The ID is used to prevent duplicate entries from being appended */
unsigned int id;
raft_entry_data_t data;
} msg_entry_t; } msg_entry_t;
/** Entry message response.
* Indicates to client if entry was committed or not. */
typedef struct 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 */ /** whether or not the entry was committed */
int was_committed; int was_committed;
} msg_entry_response_t; } msg_entry_response_t;
/** Vote request message.
* Sent to peers when a server wants to become leader.
* This message could force a leader/candidate to become a follower. */
typedef struct
{
/** currentTerm, to force other leader/candidate to step down */
int term;
/** candidate requesting vote */
int candidate_id;
/** index of candidate's last log entry */
int last_log_idx;
/** term of candidate's last log entry */
int last_log_term;
} msg_requestvote_t;
/** Vote request response message.
* Indicates if peer has accepted the server's vote request. */
typedef struct typedef struct
{ {
/* currentTerm, for candidate to update itself */ /** currentTerm, for candidate to update itself */
int term; int term;
/* true means candidate received vote */ /** true means candidate received vote */
int vote_granted; int vote_granted;
} msg_requestvote_response_t; } msg_requestvote_response_t;
/** Appendentries message.
* This message is used to tell peers if it's safe to apply entries to the FSM.
* Can be sent without any entries as a keep alive message.
* This message could force a leader/candidate to become a follower. */
typedef struct typedef struct
{ {
/** currentTerm, to force other leader/candidate to step down */
int term; int term;
int leader_id;
/** the index of the log just before the newest entry for the peer who
* receives this message */
int prev_log_idx; int prev_log_idx;
/** the term of the log just before the newest entry for the peer who
* receives this message */
int prev_log_term; int prev_log_term;
/** the index of the entry that has been appended to the majority of the
* cluster. Entries up to this index will be applied to the FSM */
int leader_commit; int leader_commit;
/** number of entries within this message */
int n_entries; int n_entries;
/** array of entries within this message */
msg_entry_t* entries; msg_entry_t* entries;
} msg_appendentries_t; } msg_appendentries_t;
/** Appendentries response message.
* Can be sent without any entries as a keep alive message.
* This message could force a leader/candidate to become a follower. */
typedef struct typedef struct
{ {
/* currentTerm, for leader to update itself */ /** currentTerm, to force other leader/candidate to step down */
int term; int term;
/* success true if follower contained entry matching /** true if follower contained entry matching prevLogidx and prevLogTerm */
* prevLogidx and prevLogTerm */
int success; int success;
/* Non-Raft fields follow: */ /* Non-Raft fields follow: */
/* Having the following fields allows us to do less book keeping in /* Having the following fields allows us to do less book keeping in
* regards to full fledged RPC */ * regards to full fledged RPC */
/* This is the highest log IDX we've received and appended to our log */
/** This is the highest log IDX we've received and appended to our log */
int current_idx; int current_idx;
/* The first idx that we received within the appendentries message */
/** The first idx that we received within the appendentries message */
int first_idx; int first_idx;
} msg_appendentries_response_t; } msg_appendentries_response_t;
typedef void* raft_server_t; typedef void* raft_server_t;
typedef void* raft_node_t; typedef void* raft_node_t;
/** Entry that is stored in the server's entry log. */
typedef struct typedef struct
{ {
/* entry's term */ /** the entry's term at the point it was created */
unsigned int term; unsigned int term;
/* the entry's unique ID */ /** the entry's unique ID */
unsigned int id; unsigned int id;
/* number of nodes that have this entry */ /** number of nodes that have this entry */
unsigned int num_nodes; unsigned int num_nodes;
/* length of entry data */ raft_entry_data_t data;
unsigned int len;
/* entry data */
unsigned char* data;
} raft_entry_t; } raft_entry_t;
/** /** Callback for sending request vote messages.
* @param raft The Raft server making this callback * @param[in] raft The Raft server making this callback
* @param udata User data that is passed from Raft server * @param[in] udata User data that is passed from Raft server
* @param node The peer's ID that we are sending this message to * @param[in] node The peer's ID that we are sending this message to
* @param[in] msg The request vote message to be sent
* @return 0 on success */ * @return 0 on success */
typedef int ( typedef int (
*func_send_requestvote_f *func_send_requestvote_f
...@@ -134,10 +155,11 @@ typedef int ( ...@@ -134,10 +155,11 @@ typedef int (
msg_requestvote_t* msg msg_requestvote_t* msg
); );
/** /** Callback for sending append entries messages.
* @param raft The Raft server making this callback * @param[in] raft The Raft server making this callback
* @param udata User data that is passed from Raft server * @param[in] udata User data that is passed from Raft server
* @param node The peer's ID that we are sending this message to * @param[in] node The peer's ID that we are sending this message to
* @param[in] msg The appendentries message to be sent
* @return 0 on success */ * @return 0 on success */
typedef int ( typedef int (
*func_send_appendentries_f *func_send_appendentries_f
...@@ -150,10 +172,11 @@ typedef int ( ...@@ -150,10 +172,11 @@ typedef int (
#ifndef HAVE_FUNC_LOG #ifndef HAVE_FUNC_LOG
#define HAVE_FUNC_LOG #define HAVE_FUNC_LOG
/** /** Callback for providing debug logging information.
* @param raft The Raft server making this callback * This callback is optional
* @param udata User data that is passed from Raft server * @param[in] raft The Raft server making this callback
* @param buf The buffer that was logged */ * @param[in] udata User data that is passed from Raft server
* @param[in] buf The buffer that was logged */
typedef void ( typedef void (
*func_log_f *func_log_f
) ( ) (
...@@ -163,12 +186,11 @@ typedef void ( ...@@ -163,12 +186,11 @@ typedef void (
); );
#endif #endif
/** /** Callback for applying this log entry to the state machine.
* Apply this log to the state machine * @param[in] raft The Raft server making this callback
* @param raft The Raft server making this callback * @param[in] udata User data that is passed from Raft server
* @param udata User data that is passed from Raft server * @param[in] data Data to be applied to the log
* @param data Data to be applied to the log * @param[in] len Length in bytes of data to be applied
* @param len Length in bytes of data to be applied
* @return 0 on success */ * @return 0 on success */
typedef int ( typedef int (
*func_applylog_f *func_applylog_f
...@@ -179,11 +201,11 @@ typedef int ( ...@@ -179,11 +201,11 @@ typedef int (
const int len const int len
); );
/** /** Callback for saving who we voted for to disk.
* Save who we voted for to disk * This callback MUST flush the change to disk.
* @param raft The Raft server making this callback * @param[in] raft The Raft server making this callback
* @param udata User data that is passed from Raft server * @param[in] udata User data that is passed from Raft server
* @param voted_for The node we voted for * @param[in] voted_for The node we voted for
* @return 0 on success */ * @return 0 on success */
typedef int ( typedef int (
*func_persist_int_f *func_persist_int_f
...@@ -193,11 +215,18 @@ typedef int ( ...@@ -193,11 +215,18 @@ typedef int (
const int voted_for const int voted_for
); );
/** /** Callback for saving log entry changes.
* @param raft The Raft server making this callback * This callback is used for:
* @param udata User data that is passed from Raft server * <ul>
* @param entry The entry that the event is happening to * <li>Adding entries to the log (ie. offer)
* @param entry_idx The entries index in the log * <li>Removing the first entry from the log (ie. polling)
* <li>Removing the last entry from the log (ie. popping)
* </ul>
* This callback MUST flush the change to disk.
* @param[in] raft The Raft server making this callback
* @param[in] udata User data that is passed from Raft server
* @param[in] entry The entry that the event is happening to
* @param[in] entry_idx The entries index in the log
* @return 0 on success */ * @return 0 on success */
typedef int ( typedef int (
*func_logentry_event_f *func_logentry_event_f
...@@ -210,28 +239,50 @@ typedef int ( ...@@ -210,28 +239,50 @@ typedef int (
typedef struct typedef struct
{ {
/* message sending */ /** Callback for sending request vote messages */
func_send_requestvote_f send_requestvote; func_send_requestvote_f send_requestvote;
/** Callback for sending appendentries messages */
func_send_appendentries_f send_appendentries; func_send_appendentries_f send_appendentries;
/* finite state machine application */ /** Callback for finite state machine application */
func_applylog_f applylog; func_applylog_f applylog;
/* persistence */ /** Callback for persisting vote data
* This callback MUST flush the change to disk. */
func_persist_int_f persist_vote; func_persist_int_f persist_vote;
/** Callback for persisting term data
* This callback MUST flush the change to disk. */
func_persist_int_f persist_term; func_persist_int_f persist_term;
/* log entry persistence */ /** Callback for adding an entry to the log
* This callback MUST flush the change to disk. */
func_logentry_event_f log_offer; func_logentry_event_f log_offer;
/** Callback for removing the oldest entry from the log
* This callback MUST flush the change to disk. */
func_logentry_event_f log_poll; func_logentry_event_f log_poll;
/** Callback for removing the youngest entry from the log
* This callback MUST flush the change to disk. */
func_logentry_event_f log_pop; func_logentry_event_f log_pop;
/* debugging - optional */ /** Callback for catching debugging log messages
* This callback is optional */
func_log_f log; func_log_f log;
} raft_cbs_t; } raft_cbs_t;
/** typedef struct
* Initialise a new Raft server {
/** User data pointer for addressing.
* Examples of what this could be:
* - void* pointing to implementor's networking data
* - a (IP,Port) tuple */
void* udata_address;
} raft_node_configuration_t;
/** Initialise a new Raft server.
* *
* Request timeout defaults to 200 milliseconds * Request timeout defaults to 200 milliseconds
* Election timeout defaults to 1000 milliseconds * Election timeout defaults to 1000 milliseconds
...@@ -239,118 +290,120 @@ typedef struct ...@@ -239,118 +290,120 @@ typedef struct
* @return newly initialised Raft server */ * @return newly initialised Raft server */
raft_server_t* raft_new(); raft_server_t* raft_new();
/** /** De-initialise Raft server.
* De-Initialise Raft server * Frees all memory */
* Free all memory */ void raft_free(raft_server_t* me);
void raft_free(raft_server_t* me_);
/** /** Set callbacks and user data.
* Set callbacks.
* Callbacks need to be set by the user for CRaft to work.
* *
* @param funcs Callbacks * @param[in] funcs Callbacks
* @param udata "User data" - user's context that's included in a callback */ * @param[in] udata "User data" - user's context that's included in a callback */
void raft_set_callbacks(raft_server_t* me, raft_cbs_t* funcs, void* udata); void raft_set_callbacks(raft_server_t* me, raft_cbs_t* funcs, void* udata);
/** /** Set configuration.
* Set configuration *
* @param nodes Array of nodes. End of array is marked by NULL entry * @deprecated This function has been replaced by raft_add_peer and
* @param my_idx Index of the node that refers to this Raft server */ * raft_remove peer
void raft_set_configuration(raft_server_t* me_, *
* @param[in] nodes Array of nodes. End of array is marked by NULL entry
* @param[in] my_idx Index of the node that refers to this Raft server */
void raft_set_configuration(raft_server_t* me,
raft_node_configuration_t* nodes, int my_idx) raft_node_configuration_t* nodes, int my_idx)
__attribute__ ((deprecated)); __attribute__ ((deprecated));
/** /** Add peer.
* Add peer
* *
* NOTE: This library does not yet support membership changes. * @note This library does not yet support membership changes.
* Once raft_periodic has been run this will fail. * Once raft_periodic has been run this will fail.
* *
* NOTE: The order this call is made is important. * @note The order this call is made is important.
* This call MUST be made in the same order as the other raft peers. * This call MUST be made in the same order as the other raft peers.
* This is because the node ID is assigned depending on when this call is made * This is because the node ID is assigned depending on when this call is made
* *
* @param udata The user data for the node. * @param[in] udata The user data for the node.
* This is obtained using raft_node_get_udata. * This is obtained using raft_node_get_udata.
* Examples of what this could be: * Examples of what this could be:
* - void* pointing to implementor's networking data * - void* pointing to implementor's networking data
* - a (IP,Port) tuple * - a (IP,Port) tuple
* @param is_self True if this "peer" is this server * @param[in] is_self Set to 1 if this "peer" is this server
* @return 0 on success; otherwise -1 */ * @return 0 on success; otherwise -1 */
int raft_add_peer(raft_server_t* me_, void* udata, int is_self); int raft_add_peer(raft_server_t* me, void* udata, int is_self);
/** /** Set election timeout.
* Set election timeout
* The amount of time that needs to elapse before we assume the leader is down * The amount of time that needs to elapse before we assume the leader is down
* @param msec Election timeout in milliseconds */ * @param[in] msec Election timeout in milliseconds */
void raft_set_election_timeout(raft_server_t* me, int msec); void raft_set_election_timeout(raft_server_t* me, int msec);
/** /** Set request timeout in milliseconds.
* Set request timeout in milliseconds
* The amount of time before we resend an appendentries message * The amount of time before we resend an appendentries message
* @param msec Request timeout in milliseconds */ * @param[in] msec Request timeout in milliseconds */
void raft_set_request_timeout(raft_server_t* me_, int msec); void raft_set_request_timeout(raft_server_t* me, int msec);
/** /** Process events that are dependent on time passing.
* Process events that are dependent on time passing * @param[in] msec_elapsed Time in milliseconds since the last call
* @param msec_elapsed Time in milliseconds since the last call
* @return 0 on success */ * @return 0 on success */
int raft_periodic(raft_server_t* me, int msec_elapsed); int raft_periodic(raft_server_t* me, int msec_elapsed);
/** /** Receive an appendentries message.
* Receive an appendentries message
* This function will block if it needs to append the message. * This function will block if it needs to append the message.
* @param node Index of the node who sent us this message * @param[in] node Index of the node who sent us this message
* @param ae The appendentries message * @param[in] ae The appendentries message
* @param[out] r The resulting response * @param[out] r The resulting response
* @return 0 on success */ * @return 0 on success */
int raft_recv_appendentries(raft_server_t* me, int node, int raft_recv_appendentries(raft_server_t* me,
int node,
msg_appendentries_t* ae, msg_appendentries_t* ae,
msg_appendentries_response_t *r); msg_appendentries_response_t *r);
/** /** Receive a response from an appendentries message we sent.
* Receive a response from an appendentries message we sent * @param[in] node Index of the node who sent us this message
* @param node Index of the node who sent us this message * @param[in] r The appendentries response message
* @param r The appendentries response message
* @return 0 on success */ * @return 0 on success */
int raft_recv_appendentries_response(raft_server_t* me_, int raft_recv_appendentries_response(raft_server_t* me,
int node, msg_appendentries_response_t* r); int node,
/** msg_appendentries_response_t* r);
* Receive a requestvote message
* @param node Index of the node who sent us this message /** Receive a requestvote message.
* @param vr The requestvote message * @param[in] node Index of the node who sent us this message
* @param[in] vr The requestvote message
* @param[out] r The resulting response * @param[out] r The resulting response
* @return 0 on success */ * @return 0 on success */
int raft_recv_requestvote(raft_server_t* me, int node, int raft_recv_requestvote(raft_server_t* me,
int node,
msg_requestvote_t* vr, msg_requestvote_t* vr,
msg_requestvote_response_t *r); msg_requestvote_response_t *r);
/** /** Receive a response from a requestvote message we sent.
* Receive a response from a requestvote message we sent * @param[in] node The node this response was sent by
* @param node Index of the node who sent us this message * @param[in] r The requestvote response message
* @param r The requestvote response message
* @param node The node this response was sent by
* @return 0 on success */ * @return 0 on success */
int raft_recv_requestvote_response(raft_server_t* me, int node, int raft_recv_requestvote_response(raft_server_t* me,
int node,
msg_requestvote_response_t* r); msg_requestvote_response_t* r);
/** /** Receive an entry message from client.
* Receive an entry message from client. *
* Append the entry to the log * Append the entry to the log and send appendentries to followers.
* Send appendentries to followers *
* This function will block if it needs to append the message. * This function will block if it needs to append the message.
*
* Will fail: * Will fail:
* - if the server is not the leader * <ul>
* @param node Index of the node who sent us this message * <li>if the server is not the leader
* </ul>
*
* @param[in] node Index of the node who sent us this message
* @param[in] ety The entry message
* @param[out] r The resulting response * @param[out] r The resulting response
* @param e The entry message
* @return 0 on success, -1 on failure */ * @return 0 on success, -1 on failure */
int raft_recv_entry(raft_server_t* me, int node, msg_entry_t* e, int raft_recv_entry(raft_server_t* me,
int node,
msg_entry_t* ety,
msg_entry_response_t *r); msg_entry_response_t *r);
/** /**
* @return the server's node ID */ * @return the server's node ID */
int raft_get_nodeid(raft_server_t* me_); int raft_get_nodeid(raft_server_t* me);
/** /**
* @return currently configured election timeout in milliseconds */ * @return currently configured election timeout in milliseconds */
...@@ -390,7 +443,7 @@ int raft_get_timeout_elapsed(raft_server_t* me); ...@@ -390,7 +443,7 @@ int raft_get_timeout_elapsed(raft_server_t* me);
/** /**
* @return request timeout in milliseconds */ * @return request timeout in milliseconds */
int raft_get_request_timeout(raft_server_t* me_); int raft_get_request_timeout(raft_server_t* me);
/** /**
* @return index of last applied entry */ * @return index of last applied entry */
...@@ -406,60 +459,56 @@ int raft_node_get_next_idx(raft_node_t* node); ...@@ -406,60 +459,56 @@ int raft_node_get_next_idx(raft_node_t* node);
/** /**
* @return this node's user data */ * @return this node's user data */
void* raft_node_get_udata(raft_node_t* me_); void* raft_node_get_udata(raft_node_t* me);
/** /**
* Set this node's udata */ * Set this node's udata */
void raft_node_set_udata(raft_node_t* me_, void* udata); void raft_node_set_udata(raft_node_t* me, void* udata);
/** /**
* @param idx The entry's index * @param[in] idx The entry's index
* @return entry from index */ * @return entry from index */
raft_entry_t* raft_get_entry_from_idx(raft_server_t* me_, int idx); raft_entry_t* raft_get_entry_from_idx(raft_server_t* me, int idx);
/** /**
* @param node The node's index * @param[in] node The node's index
* @return node pointed to by node index */ * @return node pointed to by node index */
raft_node_t* raft_get_node(raft_server_t *me_, int node); raft_node_t* raft_get_node(raft_server_t *me, int node);
/** /**
* @return number of votes this server has received this election */ * @return number of votes this server has received this election */
int raft_get_nvotes_for_me(raft_server_t* me_); int raft_get_nvotes_for_me(raft_server_t* me);
/** /**
* @return node ID of who I voted for */ * @return node ID of who I voted for */
int raft_get_voted_for(raft_server_t* me); int raft_get_voted_for(raft_server_t* me);
/** /** Get what this node thinks the node ID of the leader is.
* Gets what this node thinks the node ID of the leader is.
* @return node of what this node thinks is the valid leader; * @return node of what this node thinks is the valid leader;
* -1 if the leader is unknown */ * -1 if the leader is unknown */
int raft_get_current_leader(raft_server_t* me); int raft_get_current_leader(raft_server_t* me);
/** /**
* @return callback user data */ * @return callback user data */
void* raft_get_udata(raft_server_t* me_); void* raft_get_udata(raft_server_t* me);
/** /**
* @return this server's node ID */ * @return this server's node ID */
int raft_get_my_id(raft_server_t* me_); int raft_get_my_id(raft_server_t* me);
/** /** Vote for a server.
* Vote for a server
* This should be used to reload persistent state, ie. the voted-for field. * This should be used to reload persistent state, ie. the voted-for field.
* @param node The server to vote for */ * @param[in] node The server to vote for */
void raft_vote(raft_server_t* me_, const int node); void raft_vote(raft_server_t* me, const int node);
/** /** Set the current term.
* Set the current term
* This should be used to reload persistent state, ie. the current_term field. * This should be used to reload persistent state, ie. the current_term field.
* @param term The new current term */ * @param[in] term The new current term */
void raft_set_current_term(raft_server_t* me_, const int term); void raft_set_current_term(raft_server_t* me, const int term);
/** /** Add an entry to the server's log.
* Add an entry to the server's log
* This should be used to reload persistent state, ie. the commit log. * This should be used to reload persistent state, ie. the commit log.
* @parma 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);
#endif /* RAFT_H_ */ #endif /* RAFT_H_ */
...@@ -172,6 +172,8 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -172,6 +172,8 @@ int raft_recv_appendentries_response(raft_server_t* me_,
__log(me_, "received appendentries response node: %d %s cidx: %d 1stidx: %d", __log(me_, "received appendentries response node: %d %s cidx: %d 1stidx: %d",
node, r->success == 1 ? "success" : "fail", r->current_idx, r->first_idx); node, r->success == 1 ? "success" : "fail", r->current_idx, r->first_idx);
// TODO: should force invalid leaders to stepdown
raft_node_t* p = raft_get_node(me_, node); raft_node_t* p = raft_get_node(me_, node);
if (0 == r->success) if (0 == r->success)
...@@ -318,10 +320,10 @@ int raft_recv_appendentries( ...@@ -318,10 +320,10 @@ int raft_recv_appendentries(
/* TODO: replace malloc with mempoll/arena */ /* TODO: replace malloc with mempoll/arena */
raft_entry_t* c = (raft_entry_t*)malloc(sizeof(raft_entry_t)); raft_entry_t* c = (raft_entry_t*)malloc(sizeof(raft_entry_t));
c->term = cmd->term; c->term = cmd->term;
c->len = cmd->data.len; memcpy(&c->data, &cmd->data, sizeof(raft_entry_data_t));
c->data.buf = (unsigned char*)malloc(cmd->data.len);
memcpy(c->data.buf, cmd->data.buf, cmd->data.len);
c->id = cmd->id; c->id = cmd->id;
c->data = (unsigned char*)malloc(cmd->data.len);
memcpy(c->data, cmd->data.buf, cmd->data.len);
int e = raft_append_entry(me_, c); int e = raft_append_entry(me_, c);
if (-1 == e) if (-1 == e)
{ {
...@@ -433,9 +435,9 @@ int raft_recv_entry(raft_server_t* me_, int node, msg_entry_t* e, ...@@ -433,9 +435,9 @@ int raft_recv_entry(raft_server_t* me_, int node, msg_entry_t* e,
ety.term = me->current_term; ety.term = me->current_term;
ety.id = e->id; ety.id = e->id;
ety.len = e->data.len; ety.data.len = e->data.len;
ety.data = malloc(e->data.len); ety.data.buf = malloc(e->data.len);
memcpy(ety.data, e->data.buf, e->data.len); memcpy(ety.data.buf, e->data.buf, e->data.len);
int res = raft_append_entry(me_, &ety); int res = 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)
...@@ -482,13 +484,13 @@ int raft_apply_entry(raft_server_t* me_) ...@@ -482,13 +484,13 @@ int raft_apply_entry(raft_server_t* me_)
if (!e) if (!e)
return -1; return -1;
__log(me_, "applying log: %d, size: %d", me->last_applied_idx, e->len); __log(me_, "applying log: %d, size: %d", me->last_applied_idx, e->data.len);
me->last_applied_idx++; me->last_applied_idx++;
if (me->commit_idx < me->last_applied_idx) if (me->commit_idx < me->last_applied_idx)
me->commit_idx = me->last_applied_idx; me->commit_idx = me->last_applied_idx;
if (me->cb.applylog) if (me->cb.applylog)
me->cb.applylog(me_, me->udata, e->data, e->len); me->cb.applylog(me_, me->udata, e->data.buf, e->data.len);
return 0; return 0;
} }
...@@ -503,7 +505,6 @@ void raft_send_appendentries(raft_server_t* me_, int node) ...@@ -503,7 +505,6 @@ void raft_send_appendentries(raft_server_t* me_, int node)
msg_appendentries_t ae; msg_appendentries_t ae;
ae.term = me->current_term; ae.term = me->current_term;
ae.leader_id = me->nodeid;
ae.leader_commit = raft_get_commit_idx(me_); ae.leader_commit = raft_get_commit_idx(me_);
ae.prev_log_idx = 0; ae.prev_log_idx = 0;
ae.prev_log_term = 0; ae.prev_log_term = 0;
...@@ -519,8 +520,8 @@ void raft_send_appendentries(raft_server_t* me_, int node) ...@@ -519,8 +520,8 @@ void raft_send_appendentries(raft_server_t* me_, int node)
{ {
mety.term = ety->term; mety.term = ety->term;
mety.id = ety->id; mety.id = ety->id;
mety.data.len = ety->len; mety.data.len = ety->data.len;
mety.data.buf = ety->data; mety.data.buf = ety->data.buf;
ae.entries = &mety; ae.entries = &mety;
// TODO: we want to send more than 1 at a time // TODO: we want to send more than 1 at a time
ae.n_entries = 1; ae.n_entries = 1;
......
...@@ -27,8 +27,8 @@ void TestRaft_server_idx_starts_at_1(CuTest * tc) ...@@ -27,8 +27,8 @@ void TestRaft_server_idx_starts_at_1(CuTest * tc)
CuAssertTrue(tc, 0 == raft_get_current_idx(r)); CuAssertTrue(tc, 0 == raft_get_current_idx(r));
raft_entry_t ety; raft_entry_t ety;
ety.data = "aaa"; ety.data.buf = "aaa";
ety.len = 3; ety.data.len = 3;
ety.id = 1; ety.id = 1;
ety.term = 1; ety.term = 1;
raft_append_entry(r, &ety); raft_append_entry(r, &ety);
...@@ -95,8 +95,8 @@ void TestRaft_server_entry_append_cant_append_if_id_is_zero(CuTest* tc) ...@@ -95,8 +95,8 @@ void TestRaft_server_entry_append_cant_append_if_id_is_zero(CuTest* tc)
raft_entry_t ety; raft_entry_t ety;
char *str = "aaa"; char *str = "aaa";
ety.data = str; ety.data.buf = str;
ety.len = 3; ety.data.len = 3;
ety.id = 0; ety.id = 0;
ety.term = 1; ety.term = 1;
...@@ -111,8 +111,8 @@ void TestRaft_server_entry_append_increases_logidx(CuTest* tc) ...@@ -111,8 +111,8 @@ void TestRaft_server_entry_append_increases_logidx(CuTest* tc)
raft_entry_t ety; raft_entry_t ety;
char *str = "aaa"; char *str = "aaa";
ety.data = str; ety.data.buf = str;
ety.len = 3; ety.data.len = 3;
ety.id = 1; ety.id = 1;
ety.term = 1; ety.term = 1;
...@@ -127,8 +127,8 @@ void TestRaft_server_append_entry_means_entry_gets_current_term(CuTest* tc) ...@@ -127,8 +127,8 @@ void TestRaft_server_append_entry_means_entry_gets_current_term(CuTest* tc)
raft_entry_t ety; raft_entry_t ety;
char *str = "aaa"; char *str = "aaa";
ety.data = str; ety.data.buf = str;
ety.len = 3; ety.data.len = 3;
ety.id = 1; ety.id = 1;
ety.term = 1; ety.term = 1;
...@@ -148,8 +148,8 @@ T_estRaft_server_append_entry_not_sucessful_if_entry_with_id_already_appended( ...@@ -148,8 +148,8 @@ T_estRaft_server_append_entry_not_sucessful_if_entry_with_id_already_appended(
raft_entry_t ety; raft_entry_t ety;
char *str = "aaa"; char *str = "aaa";
ety.data = str; ety.data.buf = str;
ety.len = 3; ety.data.len = 3;
ety.id = 1; ety.id = 1;
ety.term = 1; ety.term = 1;
...@@ -178,19 +178,19 @@ void TestRaft_server_entry_is_retrieveable_using_idx(CuTest* tc) ...@@ -178,19 +178,19 @@ void TestRaft_server_entry_is_retrieveable_using_idx(CuTest* tc)
e1.term = 1; e1.term = 1;
e1.id = 1; e1.id = 1;
e1.data = str; e1.data.buf = str;
e1.len = 3; e1.data.len = 3;
raft_append_entry(r, &e1); raft_append_entry(r, &e1);
/* different ID so we can be successful */ /* different ID so we can be successful */
e2.term = 1; e2.term = 1;
e2.id = 2; e2.id = 2;
e2.data = str2; e2.data.buf = str2;
e2.len = 3; e2.data.len = 3;
raft_append_entry(r, &e2); raft_append_entry(r, &e2);
CuAssertTrue(tc, NULL != (ety_appended = raft_get_entry_from_idx(r, 2))); CuAssertTrue(tc, NULL != (ety_appended = raft_get_entry_from_idx(r, 2)));
CuAssertTrue(tc, !strncmp(ety_appended->data, str2, 3)); CuAssertTrue(tc, !strncmp(ety_appended->data.buf, str2, 3));
} }
void TestRaft_server_wont_apply_entry_if_we_dont_have_entry_to_apply(CuTest* tc) void TestRaft_server_wont_apply_entry_if_we_dont_have_entry_to_apply(CuTest* tc)
...@@ -208,8 +208,8 @@ void TestRaft_server_wont_apply_entry_if_we_dont_have_entry_to_apply(CuTest* tc) ...@@ -208,8 +208,8 @@ void TestRaft_server_wont_apply_entry_if_we_dont_have_entry_to_apply(CuTest* tc)
ety.term = 1; ety.term = 1;
ety.id = 1; ety.id = 1;
ety.data = str; ety.data.buf = str;
ety.len = 3; ety.data.len = 3;
raft_append_entry(r, &ety); raft_append_entry(r, &ety);
raft_apply_entry(r); raft_apply_entry(r);
CuAssertTrue(tc, 1 == raft_get_last_applied_idx(r)); CuAssertTrue(tc, 1 == raft_get_last_applied_idx(r));
...@@ -234,8 +234,8 @@ void TestRaft_server_increment_lastApplied_when_lastApplied_lt_commitidx( ...@@ -234,8 +234,8 @@ void TestRaft_server_increment_lastApplied_when_lastApplied_lt_commitidx(
/* need at least one entry */ /* need at least one entry */
ety.term = 1; ety.term = 1;
ety.id = 1; ety.id = 1;
ety.data = "aaa"; ety.data.buf = "aaa";
ety.len = 3; ety.data.len = 3;
raft_append_entry(r, &ety); raft_append_entry(r, &ety);
/* let time lapse */ /* let time lapse */
...@@ -256,8 +256,8 @@ void TestRaft_server_apply_entry_increments_last_applied_idx(CuTest* tc) ...@@ -256,8 +256,8 @@ void TestRaft_server_apply_entry_increments_last_applied_idx(CuTest* tc)
raft_set_last_applied_idx(r, 0); raft_set_last_applied_idx(r, 0);
ety.id = 1; ety.id = 1;
ety.data = str; ety.data.buf = str;
ety.len = 3; ety.data.len = 3;
raft_append_entry(r, &ety); raft_append_entry(r, &ety);
raft_apply_entry(r); raft_apply_entry(r);
CuAssertTrue(tc, 1 == raft_get_last_applied_idx(r)); CuAssertTrue(tc, 1 == raft_get_last_applied_idx(r));
...@@ -668,8 +668,8 @@ TestRaft_follower_recv_appendentries_delete_entries_if_conflict_with_new_entries ...@@ -668,8 +668,8 @@ TestRaft_follower_recv_appendentries_delete_entries_if_conflict_with_new_entries
/* increase log size */ /* increase log size */
char *str1 = "111"; char *str1 = "111";
ety.data = str1; ety.data.buf = str1;
ety.len = 3; ety.data.len = 3;
ety.id = 1; ety.id = 1;
ety.term = 1; ety.term = 1;
raft_append_entry(r, &ety); raft_append_entry(r, &ety);
...@@ -677,14 +677,14 @@ TestRaft_follower_recv_appendentries_delete_entries_if_conflict_with_new_entries ...@@ -677,14 +677,14 @@ TestRaft_follower_recv_appendentries_delete_entries_if_conflict_with_new_entries
/* this log will be overwritten by the appendentries below */ /* this log will be overwritten by the appendentries below */
char *str2 = "222"; char *str2 = "222";
ety.data = str2; ety.data.buf = str2;
ety.len = 3; ety.data.len = 3;
ety.id = 2; ety.id = 2;
ety.term = 1; ety.term = 1;
raft_append_entry(r, &ety); raft_append_entry(r, &ety);
CuAssertTrue(tc, 2 == raft_get_log_count(r)); CuAssertTrue(tc, 2 == raft_get_log_count(r));
CuAssertTrue(tc, NULL != (ety_appended = raft_get_entry_from_idx(r, 2))); CuAssertTrue(tc, NULL != (ety_appended = raft_get_entry_from_idx(r, 2)));
CuAssertTrue(tc, !strncmp(ety_appended->data, str2, 3)); CuAssertTrue(tc, !strncmp(ety_appended->data.buf, str2, 3));
/* pass a appendentry that is newer */ /* pass a appendentry that is newer */
msg_entry_t mety; msg_entry_t mety;
...@@ -706,7 +706,7 @@ TestRaft_follower_recv_appendentries_delete_entries_if_conflict_with_new_entries ...@@ -706,7 +706,7 @@ TestRaft_follower_recv_appendentries_delete_entries_if_conflict_with_new_entries
CuAssertTrue(tc, 1 == aer.success); CuAssertTrue(tc, 1 == aer.success);
CuAssertTrue(tc, 2 == raft_get_log_count(r)); CuAssertTrue(tc, 2 == raft_get_log_count(r));
CuAssertTrue(tc, NULL != (ety_appended = raft_get_entry_from_idx(r, 1))); CuAssertTrue(tc, NULL != (ety_appended = raft_get_entry_from_idx(r, 1)));
CuAssertTrue(tc, !strncmp(ety_appended->data, str1, 3)); CuAssertTrue(tc, !strncmp(ety_appended->data.buf, str1, 3));
} }
void TestRaft_follower_recv_appendentries_add_new_entries_not_already_in_log( void TestRaft_follower_recv_appendentries_add_new_entries_not_already_in_log(
...@@ -1330,8 +1330,8 @@ void TestRaft_leader_sends_appendentries_with_prevLogIdx( ...@@ -1330,8 +1330,8 @@ void TestRaft_leader_sends_appendentries_with_prevLogIdx(
raft_entry_t ety; raft_entry_t ety;
ety.term = 2; ety.term = 2;
ety.id = 100; ety.id = 100;
ety.len = 4; ety.data.len = 4;
ety.data = (unsigned char*)"aaa"; ety.data.buf = (unsigned char*)"aaa";
raft_append_entry(r, &ety); raft_append_entry(r, &ety);
raft_node_set_next_idx(n, 1); raft_node_set_next_idx(n, 1);
raft_send_appendentries(r, 1); raft_send_appendentries(r, 1);
...@@ -1379,8 +1379,8 @@ void TestRaft_leader_sends_appendentries_when_peer_has_next_idx_of_0( ...@@ -1379,8 +1379,8 @@ void TestRaft_leader_sends_appendentries_when_peer_has_next_idx_of_0(
raft_entry_t ety; raft_entry_t ety;
ety.term = 1; ety.term = 1;
ety.id = 100; ety.id = 100;
ety.len = 4; ety.data.len = 4;
ety.data = (unsigned char*)"aaa"; ety.data.buf = (unsigned char*)"aaa";
raft_append_entry(r, &ety); raft_append_entry(r, &ety);
raft_send_appendentries(r, 0); raft_send_appendentries(r, 0);
ae = sender_poll_msg_data(sender); ae = sender_poll_msg_data(sender);
...@@ -1452,7 +1452,7 @@ void T_estRaft_leader_doesnt_append_entry_if_unique_id_is_duplicate(CuTest * tc) ...@@ -1452,7 +1452,7 @@ void T_estRaft_leader_doesnt_append_entry_if_unique_id_is_duplicate(CuTest * tc)
msg_entry_t ety; msg_entry_t ety;
ety.id = 1; ety.id = 1;
ety.data = "entry"; ety.data = "entry";
ety.len = strlen("entry"); ety.data.len = strlen("entry");
r = raft_new(); r = raft_new();
raft_set_configuration(r, cfg, 0); raft_set_configuration(r, cfg, 0);
...@@ -1495,8 +1495,8 @@ TestRaft_leader_recv_appendentries_response_increase_commit_idx_when_majority_ha ...@@ -1495,8 +1495,8 @@ TestRaft_leader_recv_appendentries_response_increase_commit_idx_when_majority_ha
raft_entry_t ety; raft_entry_t ety;
ety.term = 1; ety.term = 1;
ety.id = 1; ety.id = 1;
ety.data = "aaaa"; ety.data.buf = "aaaa";
ety.len = 4; ety.data.len = 4;
raft_append_entry(r, &ety); raft_append_entry(r, &ety);
ety.id = 2; ety.id = 2;
raft_append_entry(r, &ety); raft_append_entry(r, &ety);
...@@ -1561,8 +1561,8 @@ void TestRaft_leader_recv_appendentries_response_failure_does_not_set_node_nexti ...@@ -1561,8 +1561,8 @@ void TestRaft_leader_recv_appendentries_response_failure_does_not_set_node_nexti
raft_entry_t ety; raft_entry_t ety;
ety.term = 1; ety.term = 1;
ety.id = 1; ety.id = 1;
ety.data = "aaaa"; ety.data.buf = "aaaa";
ety.len = 4; ety.data.len = 4;
raft_append_entry(r, &ety); raft_append_entry(r, &ety);
/* send appendentries - /* send appendentries -
......
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