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

Fix style (#128)

Fix style issues in raft_log.c raft_node.c and raft_server_properties.c
parent 358f8d41
......@@ -1103,11 +1103,11 @@ raft_node_id_t raft_get_nodeid(raft_server_t *me);
/**
* @return the server's node */
raft_node_t* raft_get_my_node(raft_server_t *me);
raft_node_t *raft_get_my_node(raft_server_t *me);
/**
* @return number of nodes that this server has */
int raft_get_num_nodes(raft_server_t* me);
int raft_get_num_nodes(raft_server_t *me);
/**
* @return number of voting nodes that this server has */
......@@ -1115,19 +1115,19 @@ int raft_get_num_voting_nodes(raft_server_t* me);
/**
* @return number of items within log */
raft_index_t raft_get_log_count(raft_server_t* me);
raft_index_t raft_get_log_count(raft_server_t *me);
/**
* @return current term */
raft_term_t raft_get_current_term(raft_server_t* me);
raft_term_t raft_get_current_term(raft_server_t *me);
/**
* @return current log index */
raft_index_t raft_get_current_idx(raft_server_t* me);
raft_index_t raft_get_current_idx(raft_server_t *me);
/**
* @return commit index */
raft_index_t raft_get_commit_idx(raft_server_t* me);
raft_index_t raft_get_commit_idx(raft_server_t *me);
/**
* @return 1 if follower; 0 otherwise */
......@@ -1135,46 +1135,46 @@ int raft_is_follower(raft_server_t* me);
/**
* @return 1 if leader; 0 otherwise */
int raft_is_leader(raft_server_t* me);
int raft_is_leader(raft_server_t *me);
/**
* @return 1 if precandidate; 0 otherwise */
int raft_is_precandidate(raft_server_t* me);
int raft_is_precandidate(raft_server_t *me);
/**
* @return 1 if candidate; 0 otherwise */
int raft_is_candidate(raft_server_t* me);
int raft_is_candidate(raft_server_t *me);
/**
* @return currently elapsed timeout in milliseconds */
raft_time_t raft_get_timeout_elapsed(raft_server_t* me);
raft_time_t raft_get_timeout_elapsed(raft_server_t *me);
/**
* @return index of last applied entry */
raft_index_t raft_get_last_applied_idx(raft_server_t* me);
raft_index_t raft_get_last_applied_idx(raft_server_t *me);
/**
* @return the node's next index */
raft_index_t raft_node_get_next_idx(raft_node_t* node);
raft_index_t raft_node_get_next_idx(raft_node_t *node);
/**
* @return this node's user data */
raft_index_t raft_node_get_match_idx(raft_node_t* me);
raft_index_t raft_node_get_match_idx(raft_node_t *node);
/**
* @return this node's user data */
void* raft_node_get_udata(raft_node_t* me);
void* raft_node_get_udata(raft_node_t *node);
/**
* Set this node's user data */
void raft_node_set_udata(raft_node_t* me, void* user_data);
void raft_node_set_udata(raft_node_t *node, void *user_data);
/**
* After sending the snapshot, user can set the next index for the node
*
* @param[in] node node
* @param[in] idx next entry index */
void raft_node_set_next_idx(raft_node_t* me, raft_index_t idx);
void raft_node_set_next_idx(raft_node_t *node, raft_index_t idx);
/**
* @param[in] idx The entry's index
......@@ -1192,13 +1192,13 @@ raft_entry_t** raft_get_entries_from_idx(raft_server_t* me,
/**
* @param[in] node The node's ID
* @return node pointed to by node ID */
raft_node_t* raft_get_node(raft_server_t* me, raft_node_id_t id);
raft_node_t *raft_get_node(raft_server_t *me, raft_node_id_t id);
/**
* Used for iterating through nodes
* @param[in] node The node's idx
* @return node pointed to by node idx */
raft_node_t* raft_get_node_from_idx(raft_server_t* me, raft_index_t idx);
raft_node_t *raft_get_node_from_idx(raft_server_t *me, raft_index_t idx);
/**
* @return number of votes this server has received this election */
......@@ -1211,17 +1211,17 @@ raft_node_id_t raft_get_voted_for(raft_server_t *me);
/** Get what this node thinks the node ID of the leader is.
* @return node of what this node thinks is the valid leader;
* RAFT_NODE_ID_NONE if there is no leader */
raft_node_id_t raft_get_leader_id(raft_server_t* me);
raft_node_id_t raft_get_leader_id(raft_server_t *me);
/** Get what this node thinks the node of the leader is.
* @return node of what this node thinks is the valid leader;
* NULL if there is no leader or
* if the leader is not part of the local configuration yet */
raft_node_t* raft_get_leader_node(raft_server_t* me);
raft_node_t *raft_get_leader_node(raft_server_t *me);
/**
* @return callback user data */
void* raft_get_udata(raft_server_t* me);
void *raft_get_udata(raft_server_t *me);
/** Vote for a server.
* This should be used to reload persistent state, ie. the voted-for field.
......@@ -1242,12 +1242,12 @@ int raft_vote_for_nodeid(raft_server_t* me, raft_node_id_t nodeid);
* @param[in] term The new current term
* @return
* 0 on success */
int raft_set_current_term(raft_server_t* me, raft_term_t term);
int raft_set_current_term(raft_server_t *me, raft_term_t term);
/** Set the commit idx.
* This should be used to reload persistent state, ie. the commit_idx field.
* @param[in] commit_idx The new commit index. */
void raft_set_commit_idx(raft_server_t* me, raft_index_t commit_idx);
void raft_set_commit_idx(raft_server_t *me, raft_index_t commit_idx);
/** Add an entry to the server's log.
* This should be used to reload persistent state, ie. the commit log.
......@@ -1272,31 +1272,31 @@ int raft_msg_entry_response_committed(raft_server_t* me,
/** Get node's ID.
* @return ID of node */
raft_node_id_t raft_node_get_id(raft_node_t* me);
raft_node_id_t raft_node_get_id(raft_node_t *node);
/** Tell if we are a leader, candidate or follower.
* @return get state of type raft_state_e. */
int raft_get_state(raft_server_t* me);
int raft_get_state(raft_server_t *me);
/* @return state string */
const char *raft_get_state_str(raft_server_t* me);
const char *raft_get_state_str(raft_server_t *me);
/** Get the most recent log's term
* @return the last log term */
raft_term_t raft_get_last_log_term(raft_server_t* me);
raft_term_t raft_get_last_log_term(raft_server_t *me);
/** Turn a node into a voting node.
* Voting nodes can take part in elections and in-regards to committing entries,
* are counted in majorities. */
void raft_node_set_voting(raft_node_t* node, int voting);
void raft_node_set_voting(raft_node_t *node, int voting);
/** Tell if a node is a voting node or not.
* @return 1 if this is a voting node. Otherwise 0. */
int raft_node_is_voting(raft_node_t* me);
int raft_node_is_voting(raft_node_t *node);
/** Check if a node has sufficient logs to be able to join the cluster.
**/
int raft_node_has_sufficient_logs(raft_node_t* me);
int raft_node_has_sufficient_logs(raft_node_t *node);
/** Apply all entries up to the commit index
* @return
......@@ -1371,7 +1371,7 @@ int raft_snapshot_is_in_progress(raft_server_t *me);
/** Check if entries can be applied now (no snapshot in progress, or
* RAFT_SNAPSHOT_NONBLOCKING_APPLY specified).
**/
int raft_is_apply_allowed(raft_server_t* me);
int raft_is_apply_allowed(raft_server_t *me);
/** Get last applied entry
**/
......@@ -1435,7 +1435,7 @@ raft_term_t raft_get_snapshot_last_term(raft_server_t *me);
* Active nodes could become voting nodes.
* This should be used for creating the membership snapshot.
**/
int raft_node_is_active(raft_node_t* me);
int raft_node_is_active(raft_node_t *node);
/** Make the node active.
*
......@@ -1444,17 +1444,17 @@ int raft_node_is_active(raft_node_t* me);
*
* @param[in] active Set a node as active if this is 1
**/
void raft_node_set_active(raft_node_t* me, int active);
void raft_node_set_active(raft_node_t *node, int active);
/** Check if a node's voting status has been committed.
* This should be used for creating the membership snapshot.
**/
int raft_node_is_voting_committed(raft_node_t* me);
int raft_node_is_voting_committed(raft_node_t *node);
/** Check if a node's membership to the cluster has been committed.
* This should be used for creating the membership snapshot.
**/
int raft_node_is_addition_committed(raft_node_t* me);
int raft_node_is_addition_committed(raft_node_t *node);
/**
* Register custom heap management functions, to be used if an alternative
......@@ -1468,17 +1468,17 @@ void raft_set_heap_functions(void *(*_malloc)(size_t),
/** Confirm that a node's voting status is final
* @param[in] node The node
* @param[in] voting Whether this node's voting status is committed or not */
void raft_node_set_voting_committed(raft_node_t* me, int voting);
void raft_node_set_voting_committed(raft_node_t *node, int voting);
/** Confirm that a node's voting status is final
* @param[in] node The node
* @param[in] committed Whether this node's membership is committed or not */
void raft_node_set_addition_committed(raft_node_t* me, int committed);
void raft_node_set_addition_committed(raft_node_t *node, int committed);
/** Check if a voting change is in progress
* @param[in] raft The Raft server
* @return 1 if a voting change is in progress */
int raft_voting_change_is_in_progress(raft_server_t* me);
int raft_voting_change_is_in_progress(raft_server_t *me);
/** Get the log implementation handle in use.
*/
......@@ -1575,7 +1575,7 @@ int raft_recv_read_request(raft_server_t* me, raft_read_request_callback_f cb, v
int raft_transfer_leader(raft_server_t* me, raft_node_id_t node_id, long timeout);
/* get the targeted node_id if a leadership transfer is in progress, or RAFT_NODE_ID_NONE if not */
raft_node_id_t raft_get_transfer_leader(raft_server_t* me);
raft_node_id_t raft_get_transfer_leader(raft_server_t *me);
/* cause this server to force an election. */
int raft_timeout_now(raft_server_t* me);
......
......@@ -5,56 +5,60 @@
typedef struct raft_log raft_log_t;
raft_log_t* raft_log_new(void);
raft_log_t *raft_log_new(void);
raft_log_t* raft_log_alloc(raft_index_t initial_size);
raft_log_t *raft_log_alloc(raft_index_t initial_size);
void raft_log_set_callbacks(raft_log_t* me, raft_log_cbs_t* funcs, void* raft);
void raft_log_set_callbacks(raft_log_t *me, raft_log_cbs_t *funcs, void *raft);
void raft_log_free(raft_log_t* me);
void raft_log_free(raft_log_t *me);
void raft_log_clear(raft_log_t* me);
void raft_log_clear(raft_log_t *me);
void raft_log_clear_entries(raft_log_t* me);
void raft_log_clear_entries(raft_log_t *me);
/**
* Add entry to log.
* Don't add entry if we've already added this entry (based off ID)
* Don't add entries with ID=0
* @return 0 if unsuccessful; 1 otherwise */
int raft_log_append_entry(raft_log_t* me, raft_entry_t* c);
int raft_log_append_entry(raft_log_t *me, raft_entry_t *c);
/**
* @return number of entries held within log */
raft_index_t raft_log_count(raft_log_t* me);
raft_index_t raft_log_count(raft_log_t *me);
/**
* Delete all logs from this log onwards */
int raft_log_delete(raft_log_t* me, raft_index_t idx);
int raft_log_delete(raft_log_t *me, raft_index_t idx);
/**
* Empty the queue. */
void raft_log_empty(raft_log_t * me);
void raft_log_empty(raft_log_t *me);
/**
* Remove oldest entry. Set *etyp to oldest entry on success. */
int raft_log_poll(raft_log_t * me, raft_entry_t **etyp);
int raft_log_poll(raft_log_t *me, raft_entry_t **etyp);
/** Get an array of entries from this index onwards.
* This is used for batching.
*/
raft_entry_t** raft_log_get_from_idx(raft_log_t* me, raft_index_t idx, long *n_etys);
raft_entry_t **raft_log_get_from_idx(raft_log_t *me,
raft_index_t idx,
raft_index_t *n_etys);
raft_entry_t* raft_log_get_at_idx(raft_log_t* me, raft_index_t idx);
raft_entry_t *raft_log_get_at_idx(raft_log_t *me, raft_index_t idx);
/**
* @return youngest entry */
raft_entry_t *raft_log_peektail(raft_log_t * me);
raft_entry_t *raft_log_peektail(raft_log_t *me);
raft_index_t raft_log_get_current_idx(raft_log_t* me);
raft_index_t raft_log_get_current_idx(raft_log_t *me);
int raft_log_load_from_snapshot(raft_log_t *me, raft_index_t idx, raft_term_t term);
int raft_log_load_from_snapshot(raft_log_t *me,
raft_index_t idx,
raft_term_t term);
raft_index_t raft_log_get_base(raft_log_t* me);
raft_index_t raft_log_get_base(raft_log_t *me);
#endif /* RAFT_LOG_H_ */
......@@ -156,33 +156,33 @@ int raft_send_appendentries_all(raft_server_t* me);
* @return 1 if entry committed, 0 otherwise */
int raft_apply_entry(raft_server_t* me);
void raft_set_last_applied_idx(raft_server_t* me, raft_index_t idx);
void raft_set_last_applied_idx(raft_server_t *me, raft_index_t idx);
void raft_set_state(raft_server_t* me, int state);
void raft_set_state(raft_server_t *me, int state);
raft_node_t* raft_node_new(void* udata, raft_node_id_t id);
raft_node_t *raft_node_new(void *udata, raft_node_id_t id);
void raft_node_free(raft_node_t* me);
void raft_node_free(raft_node_t *node);
void raft_node_set_match_idx(raft_node_t* node, raft_index_t idx);
void raft_node_set_match_idx(raft_node_t *node, raft_index_t idx);
void raft_node_clear_flags(raft_node_t *me);
void raft_node_clear_flags(raft_node_t *node);
void raft_node_vote_for_me(raft_node_t* me, int vote);
void raft_node_vote_for_me(raft_node_t *node, int vote);
int raft_node_has_vote_for_me(raft_node_t* me);
int raft_node_has_vote_for_me(raft_node_t *node);
void raft_node_set_has_sufficient_logs(raft_node_t *me, int has_sufficient_log);
void raft_node_set_has_sufficient_logs(raft_node_t *node, int sufficient_logs);
int raft_is_single_node_voting_cluster(raft_server_t *me);
int raft_votes_is_majority(int nnodes, int nvotes);
void raft_node_set_match_msgid(raft_node_t *me, raft_msg_id_t msgid);
raft_msg_id_t raft_node_get_match_msgid(raft_node_t *me);
void raft_node_set_match_msgid(raft_node_t *node, raft_msg_id_t msgid);
raft_msg_id_t raft_node_get_match_msgid(raft_node_t *node);
void raft_node_set_next_msgid(raft_node_t *me, raft_msg_id_t msgid);
raft_msg_id_t raft_node_get_next_msgid(raft_node_t *me);
void raft_node_set_next_msgid(raft_node_t *node, raft_msg_id_t msgid);
raft_msg_id_t raft_node_get_next_msgid(raft_node_t *node);
/* Heap functions */
extern void *(*raft_malloc)(size_t size);
......@@ -191,18 +191,18 @@ extern void *(*raft_realloc)(void *ptr, size_t size);
extern void (*raft_free)(void *ptr);
/* update the max_seen_msg_id for this node */
void raft_node_update_max_seen_msg_id(raft_node_t *me, raft_msg_id_t msg_id);
void raft_node_update_max_seen_msg_id(raft_node_t *node, raft_msg_id_t msg_id);
/* get the max message id this server has seen from its the specified node */
raft_msg_id_t raft_node_get_max_seen_msg_id(raft_node_t *me);
raft_msg_id_t raft_node_get_max_seen_msg_id(raft_node_t *node);
/* get the server's current msg_id */
raft_msg_id_t raft_get_msg_id(raft_server_t* me);
raft_msg_id_t raft_get_msg_id(raft_server_t *me);
/* attempt to abort the leadership transfer */
void raft_reset_transfer_leader(raft_server_t* me, int timed_out);
raft_size_t raft_node_get_snapshot_offset(raft_node_t *me);
raft_size_t raft_node_get_snapshot_offset(raft_node_t *node);
void raft_node_set_snapshot_offset(raft_node_t *me, raft_size_t snapshot_offset);
void raft_node_set_snapshot_offset(raft_node_t *node, raft_size_t offset);
int raft_periodic_internal(raft_server_t *me, raft_time_t milliseconds);
......
......@@ -17,8 +17,7 @@
#define INITIAL_CAPACITY 10
struct raft_log
{
struct raft_log {
/* size of array */
raft_index_t size;
......@@ -31,12 +30,12 @@ struct raft_log
/* we compact the log, and thus need to increment the Base Log Index */
raft_index_t base;
raft_entry_t** entries;
raft_entry_t **entries;
/* callbacks */
raft_log_cbs_t cb;
void* raft;
void *raft;
};
static raft_index_t mod(raft_index_t a, raft_index_t b)
......@@ -47,20 +46,21 @@ static raft_index_t mod(raft_index_t a, raft_index_t b)
static int ensure_capacity(raft_log_t *me)
{
raft_index_t i, j;
raft_entry_t **temp;
if (me->count < me->size)
if (me->count < me->size) {
return 0;
}
temp = raft_calloc(1, sizeof(raft_entry_t *) * me->size * 2);
if (!temp)
temp = raft_calloc(1, sizeof(*temp) * me->size * 2);
if (!temp) {
return RAFT_ERR_NOMEM;
}
for (i = 0, j = me->front; i < me->count; i++, j++)
{
if (j == me->size)
for (raft_index_t i = 0, j = me->front; i < me->count; i++, j++) {
if (j == me->size) {
j = 0;
}
temp[i] = me->entries[j];
}
......@@ -71,10 +71,13 @@ static int ensure_capacity(raft_log_t *me)
me->entries = temp;
me->front = 0;
me->back = me->count;
return 0;
}
int raft_log_load_from_snapshot(raft_log_t *me, raft_index_t idx, raft_term_t term)
int raft_log_load_from_snapshot(raft_log_t *me,
raft_index_t idx,
raft_term_t term)
{
(void) term;
......@@ -88,15 +91,19 @@ int raft_log_load_from_snapshot(raft_log_t *me, raft_index_t idx, raft_term_t te
raft_log_t *raft_log_alloc(raft_index_t initial_size)
{
raft_log_t *me = raft_calloc(1, sizeof(*me));
if (!me)
if (!me) {
return NULL;
}
me->size = initial_size;
raft_log_clear(me);
me->entries = raft_calloc(1, sizeof(raft_entry_t *) * me->size);
me->entries = raft_calloc(1, sizeof(*me->entries) * me->size);
if (!me->entries) {
raft_free(me);
return NULL;
}
return me;
}
......@@ -105,7 +112,7 @@ raft_log_t *raft_log_new(void)
return raft_log_alloc(INITIAL_CAPACITY);
}
void raft_log_set_callbacks(raft_log_t *me, raft_log_cbs_t* funcs, void* raft)
void raft_log_set_callbacks(raft_log_t *me, raft_log_cbs_t *funcs, void *raft)
{
me->raft = raft;
me->cb = *funcs;
......@@ -119,38 +126,37 @@ void raft_log_clear(raft_log_t *me)
me->base = 0;
}
void raft_log_clear_entries(raft_log_t* me)
void raft_log_clear_entries(raft_log_t *me)
{
raft_index_t i;
if (!me->count || !me->cb.log_clear)
if (!me->count || !me->cb.log_clear) {
return;
}
for (i = me->base; i <= me->base + me->count; i++)
{
for (raft_index_t i = me->base; i <= me->base + me->count; i++) {
me->cb.log_clear(me->raft, raft_get_udata(me->raft),
me->entries[(me->front + i - me->base) % me->size], i);
me->entries[(me->front + i - me->base) % me->size], i);
}
}
/** TODO: rename log_append */
int raft_log_append_entry(raft_log_t *me, raft_entry_t*c)
int raft_log_append_entry(raft_log_t *me, raft_entry_t *ety)
{
raft_index_t idx = me->base + me->count + 1;
int e;
raft_index_t idx = me->base + me->count + 1;
e = ensure_capacity(me);
if (e != 0)
if (e != 0) {
return e;
}
me->entries[me->back] = ety;
me->entries[me->back] = c;
if (me->cb.log_offer) {
void *udata = raft_get_udata(me->raft);
if (me->cb.log_offer)
{
void* ud = raft_get_udata(me->raft);
e = me->cb.log_offer(me->raft, ud, me->entries[me->back], idx);
if (0 != e)
e = me->cb.log_offer(me->raft, udata, me->entries[me->back], idx);
if (e != 0) {
return e;
}
}
me->count++;
......@@ -160,14 +166,15 @@ int raft_log_append_entry(raft_log_t *me, raft_entry_t*c)
return 0;
}
raft_entry_t**raft_log_get_from_idx(raft_log_t *me, raft_index_t idx, long *n_etys)
raft_entry_t **raft_log_get_from_idx(raft_log_t *me,
raft_index_t idx,
raft_index_t *n_etys)
{
raft_index_t i;
assert(0 <= idx - 1);
if (me->base + me->count < idx || idx <= me->base)
{
if (me->base + me->count < idx || idx <= me->base) {
*n_etys = 0;
return NULL;
}
......@@ -178,23 +185,22 @@ raft_entry_t**raft_log_get_from_idx(raft_log_t *me, raft_index_t idx, long *n_et
i = (me->front + idx - me->base) % me->size;
/* log entries until the end of the log */
if (i < me->back)
*n_etys = me->back - i;
else
*n_etys = me->size - i;
*n_etys = (i < me->back) ? (me->back - i) : (me->size - i);
return &me->entries[i];
}
raft_entry_t* raft_log_get_at_idx(raft_log_t *me, raft_index_t idx)
raft_entry_t *raft_log_get_at_idx(raft_log_t *me, raft_index_t idx)
{
raft_index_t i;
if (idx == 0)
if (idx == 0) {
return NULL;
}
if (me->base + me->count < idx || idx <= me->base)
if (me->base + me->count < idx || idx <= me->base) {
return NULL;
}
/* idx starts at 1 */
idx -= 1;
......@@ -208,35 +214,41 @@ 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,
raft_entry_notify_f cb,
void *cb_arg)
{
if (0 == idx)
if (idx == 0) {
return -1;
}
if (idx < me->base)
if (idx < me->base) {
idx = me->base;
}
for (; idx <= me->base + me->count && me->count;)
{
for (; idx <= me->base + me->count && me->count;) {
raft_index_t idx_tmp = me->base + me->count;
raft_index_t back = mod(me->back - 1, me->size);
if (me->cb.log_pop)
{
if (me->cb.log_pop) {
int e = me->cb.log_pop(me->raft, raft_get_udata(me->raft),
me->entries[back], idx_tmp);
if (0 != e)
if (e != 0) {
return e;
}
}
if (cb)
if (cb) {
cb(cb_arg, me->entries[back], idx_tmp);
}
raft_entry_release(me->entries[back]);
me->back = back;
me->count--;
}
return 0;
}
......@@ -249,16 +261,18 @@ int raft_log_poll(raft_log_t *me, raft_entry_t **etyp)
{
raft_index_t idx = me->base + 1;
if (0 == me->count)
if (me->count == 0) {
return -1;
}
raft_entry_t *elem = me->entries[me->front];
const void *elem = me->entries[me->front];
if (me->cb.log_poll)
{
if (me->cb.log_poll) {
int e = me->cb.log_poll(me->raft, raft_get_udata(me->raft),
me->entries[me->front], idx);
if (0 != e)
me->entries[me->front], idx);
if (e != 0) {
return e;
}
}
raft_entry_release(me->entries[me->front]);
......@@ -268,19 +282,20 @@ int raft_log_poll(raft_log_t *me, raft_entry_t **etyp)
me->count--;
me->base++;
*etyp = (void*)elem;
*etyp = elem;
return 0;
}
raft_entry_t *raft_log_peektail(raft_log_t *me)
{
if (0 == me->count)
if (me->count == 0) {
return NULL;
}
raft_index_t tail = (me->back == 0) ? (me->size - 1) : (me->back - 1);
if (0 == me->back)
return me->entries[me->size - 1];
else
return me->entries[me->back - 1];
return me->entries[tail];
}
void raft_log_empty(raft_log_t *me)
......@@ -352,8 +367,10 @@ static int log_append(void *log, raft_entry_t *entry)
static raft_entry_t *log_get(void *log, raft_index_t idx)
{
raft_entry_t *e = raft_log_get_at_idx(log, idx);
if (e != NULL)
if (e != NULL) {
raft_entry_hold(e);
}
return e;
}
......@@ -369,17 +386,20 @@ static raft_index_t log_get_batch(void *log,
return 0;
}
if (n > entries_n)
if (n > entries_n) {
n = entries_n;
}
for (raft_index_t i = 0; i < n; i++) {
entries[i] = r[i];
raft_entry_hold(entries[i]);
}
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,
raft_entry_notify_f cb, void *cb_arg)
{
return log_delete(log, from_idx, cb, cb_arg);
}
......@@ -389,9 +409,9 @@ static int log_poll(void *log, raft_index_t first_idx)
while (raft_log_get_base(log) + 1 < first_idx) {
raft_entry_t *ety;
int e = raft_log_poll(log, &ety);
if (e < 0)
if (e < 0) {
return e;
}
}
return 0;
......
......@@ -9,9 +9,6 @@
* @version 0.1
*/
#include <string.h>
#include <assert.h>
#include "raft.h"
#include "raft_private.h"
......@@ -22,9 +19,8 @@
#define RAFT_NODE_VOTING_COMMITTED (1 << 4)
#define RAFT_NODE_ADDITION_COMMITTED (1 << 5)
struct raft_node
{
void* udata;
struct raft_node {
void *udata;
raft_index_t next_idx;
raft_index_t match_idx;
......@@ -41,186 +37,193 @@ struct raft_node
raft_size_t snapshot_offset;
};
raft_node_t* raft_node_new(void* udata, raft_node_id_t id)
raft_node_t *raft_node_new(void *udata, raft_node_id_t id)
{
raft_node_t* me;
raft_node_t *me;
me = raft_calloc(1, sizeof(*me));
if (!me)
if (!me) {
return NULL;
}
me->udata = udata;
me->next_idx = 1;
me->match_idx = 0;
me->id = id;
me->flags = RAFT_NODE_VOTING;
return me;
}
void raft_node_free(raft_node_t* me)
void raft_node_free(raft_node_t *node)
{
raft_free(me);
raft_free(node);
}
raft_index_t raft_node_get_next_idx(raft_node_t* me)
raft_index_t raft_node_get_next_idx(raft_node_t *node)
{
return me->next_idx;
return node->next_idx;
}
void raft_node_set_next_idx(raft_node_t* me, raft_index_t idx)
void raft_node_set_next_idx(raft_node_t *node, raft_index_t idx)
{
/* log index begins at 1 */
me->next_idx = idx < 1 ? 1 : idx;
node->next_idx = idx < 1 ? 1 : idx;
}
raft_index_t raft_node_get_match_idx(raft_node_t* me)
raft_index_t raft_node_get_match_idx(raft_node_t *node)
{
return me->match_idx;
return node->match_idx;
}
void raft_node_set_match_idx(raft_node_t* me, raft_index_t idx)
void raft_node_set_match_idx(raft_node_t *node, raft_index_t idx)
{
me->match_idx = idx;
node->match_idx = idx;
}
void raft_node_set_match_msgid(raft_node_t *me, raft_msg_id_t msgid)
void raft_node_set_match_msgid(raft_node_t *node, raft_msg_id_t msgid)
{
me->match_msgid = msgid;
node->match_msgid = msgid;
}
raft_msg_id_t raft_node_get_match_msgid(raft_node_t *me)
raft_msg_id_t raft_node_get_match_msgid(raft_node_t *node)
{
return me->match_msgid;
return node->match_msgid;
}
void raft_node_set_next_msgid(raft_node_t *me, raft_msg_id_t msgid)
void raft_node_set_next_msgid(raft_node_t *node, raft_msg_id_t msgid)
{
me->next_msgid = msgid;
node->next_msgid = msgid;
}
raft_msg_id_t raft_node_get_next_msgid(raft_node_t *me)
raft_msg_id_t raft_node_get_next_msgid(raft_node_t *node)
{
return me->next_msgid;
return node->next_msgid;
}
void raft_node_update_max_seen_msg_id(raft_node_t *me, raft_msg_id_t msg_id)
void raft_node_update_max_seen_msg_id(raft_node_t *node, raft_msg_id_t msg_id)
{
if (me->max_seen_msgid < msg_id) {
me->max_seen_msgid = msg_id;
if (node->max_seen_msgid < msg_id) {
node->max_seen_msgid = msg_id;
}
}
raft_msg_id_t raft_node_get_max_seen_msg_id(raft_node_t *me)
raft_msg_id_t raft_node_get_max_seen_msg_id(raft_node_t *node)
{
return me->max_seen_msgid;
return node->max_seen_msgid;
}
void* raft_node_get_udata(raft_node_t* me)
void* raft_node_get_udata(raft_node_t *node)
{
return me->udata;
return node->udata;
}
void raft_node_set_udata(raft_node_t* me, void* udata)
void raft_node_set_udata(raft_node_t *node, void* udata)
{
me->udata = udata;
node->udata = udata;
}
void raft_node_clear_flags(raft_node_t *me)
void raft_node_clear_flags(raft_node_t *node)
{
me->flags = 0;
node->flags = 0;
}
void raft_node_vote_for_me(raft_node_t* me, const int vote)
void raft_node_vote_for_me(raft_node_t *node, const int vote)
{
if (vote)
me->flags |= RAFT_NODE_VOTED_FOR_ME;
else
me->flags &= ~RAFT_NODE_VOTED_FOR_ME;
if (vote) {
node->flags |= RAFT_NODE_VOTED_FOR_ME;
} else {
node->flags &= ~RAFT_NODE_VOTED_FOR_ME;
}
}
int raft_node_has_vote_for_me(raft_node_t* me)
int raft_node_has_vote_for_me(raft_node_t *node)
{
return (me->flags & RAFT_NODE_VOTED_FOR_ME) != 0;
return (node->flags & RAFT_NODE_VOTED_FOR_ME) != 0;
}
void raft_node_set_voting(raft_node_t *me, int voting)
void raft_node_set_voting(raft_node_t *node, int voting)
{
if (voting) {
me->flags |= RAFT_NODE_VOTING;
node->flags |= RAFT_NODE_VOTING;
} else {
me->flags &= ~RAFT_NODE_VOTING;
node->flags &= ~RAFT_NODE_VOTING;
}
}
int raft_node_is_voting(raft_node_t* me)
int raft_node_is_voting(raft_node_t *node)
{
return (me->flags & RAFT_NODE_VOTING && !(me->flags & RAFT_NODE_INACTIVE));
return (node->flags & RAFT_NODE_VOTING &&
!(node->flags & RAFT_NODE_INACTIVE));
}
int raft_node_has_sufficient_logs(raft_node_t* me)
int raft_node_has_sufficient_logs(raft_node_t *node)
{
return (me->flags & RAFT_NODE_HAS_SUFFICIENT_LOG) != 0;
return (node->flags & RAFT_NODE_HAS_SUFFICIENT_LOG) != 0;
}
void raft_node_set_has_sufficient_logs(raft_node_t *me, int has_sufficient_log)
void raft_node_set_has_sufficient_logs(raft_node_t *node, int sufficient_logs)
{
if (has_sufficient_log) {
me->flags |= RAFT_NODE_HAS_SUFFICIENT_LOG;
if (sufficient_logs) {
node->flags |= RAFT_NODE_HAS_SUFFICIENT_LOG;
} else {
me->flags &= ~RAFT_NODE_HAS_SUFFICIENT_LOG;
node->flags &= ~RAFT_NODE_HAS_SUFFICIENT_LOG;
}
}
void raft_node_set_active(raft_node_t* me, int active)
void raft_node_set_active(raft_node_t *node, int active)
{
if (!active)
me->flags |= RAFT_NODE_INACTIVE;
else
me->flags &= ~RAFT_NODE_INACTIVE;
if (!active) {
node->flags |= RAFT_NODE_INACTIVE;
} else {
node->flags &= ~RAFT_NODE_INACTIVE;
}
}
int raft_node_is_active(raft_node_t* me)
int raft_node_is_active(raft_node_t *node)
{
return (me->flags & RAFT_NODE_INACTIVE) == 0;
return (node->flags & RAFT_NODE_INACTIVE) == 0;
}
void raft_node_set_voting_committed(raft_node_t* me, int voting)
void raft_node_set_voting_committed(raft_node_t *node, int voting)
{
if (voting)
me->flags |= RAFT_NODE_VOTING_COMMITTED;
else
me->flags &= ~RAFT_NODE_VOTING_COMMITTED;
if (voting) {
node->flags |= RAFT_NODE_VOTING_COMMITTED;
} else {
node->flags &= ~RAFT_NODE_VOTING_COMMITTED;
}
}
int raft_node_is_voting_committed(raft_node_t* me)
int raft_node_is_voting_committed(raft_node_t *node)
{
return (me->flags & RAFT_NODE_VOTING_COMMITTED) != 0;
return (node->flags & RAFT_NODE_VOTING_COMMITTED) != 0;
}
raft_node_id_t raft_node_get_id(raft_node_t* me)
raft_node_id_t raft_node_get_id(raft_node_t *node)
{
return me != NULL ? me->id : RAFT_NODE_ID_NONE;
return node != NULL ? node->id : RAFT_NODE_ID_NONE;
}
void raft_node_set_addition_committed(raft_node_t* me, int committed)
void raft_node_set_addition_committed(raft_node_t *node, int committed)
{
if (committed)
me->flags |= RAFT_NODE_ADDITION_COMMITTED;
else
me->flags &= ~RAFT_NODE_ADDITION_COMMITTED;
if (committed) {
node->flags |= RAFT_NODE_ADDITION_COMMITTED;
} else {
node->flags &= ~RAFT_NODE_ADDITION_COMMITTED;
}
}
int raft_node_is_addition_committed(raft_node_t* me)
int raft_node_is_addition_committed(raft_node_t *node)
{
return (me->flags & RAFT_NODE_ADDITION_COMMITTED) != 0;
return (node->flags & RAFT_NODE_ADDITION_COMMITTED) != 0;
}
raft_size_t raft_node_get_snapshot_offset(raft_node_t *me)
raft_size_t raft_node_get_snapshot_offset(raft_node_t *node)
{
return me->snapshot_offset;
return node->snapshot_offset;
}
void raft_node_set_snapshot_offset(raft_node_t *me, raft_size_t snapshot_offset)
void raft_node_set_snapshot_offset(raft_node_t *node, raft_size_t offset)
{
me->snapshot_offset = snapshot_offset;
node->snapshot_offset = offset;
}
......@@ -18,26 +18,30 @@ raft_node_id_t raft_get_nodeid(raft_server_t *me)
return raft_node_get_id(me->node);
}
int raft_get_num_nodes(raft_server_t* me)
int raft_get_num_nodes(raft_server_t *me)
{
return me->num_nodes;
}
int raft_get_num_voting_nodes(raft_server_t* me)
int raft_get_num_voting_nodes(raft_server_t *me)
{
int i, num = 0;
for (i = 0; i < me->num_nodes; i++)
if (raft_node_is_voting(me->nodes[i]))
int num = 0;
for (int i = 0; i < me->num_nodes; i++) {
if (raft_node_is_voting(me->nodes[i])) {
num++;
}
}
return num;
}
raft_time_t raft_get_timeout_elapsed(raft_server_t* me)
raft_time_t raft_get_timeout_elapsed(raft_server_t *me)
{
return me->timeout_elapsed;
}
raft_index_t raft_get_log_count(raft_server_t* me)
raft_index_t raft_get_log_count(raft_server_t *me)
{
return me->log_impl->count(me->log);
}
......@@ -47,7 +51,7 @@ raft_node_id_t raft_get_voted_for(raft_server_t *me)
return me->voted_for;
}
int raft_set_current_term(raft_server_t* me, const raft_term_t term)
int raft_set_current_term(raft_server_t *me, const raft_term_t term)
{
if (me->current_term >= term) {
return 0;
......@@ -66,52 +70,52 @@ int raft_set_current_term(raft_server_t* me, const raft_term_t term)
return 0;
}
raft_term_t raft_get_current_term(raft_server_t* me)
raft_term_t raft_get_current_term(raft_server_t *me)
{
return me->current_term;
}
raft_index_t raft_get_current_idx(raft_server_t* me)
raft_index_t raft_get_current_idx(raft_server_t *me)
{
return me->log_impl->current_idx(me->log);
}
void raft_set_commit_idx(raft_server_t* me, raft_index_t idx)
void raft_set_commit_idx(raft_server_t *me, raft_index_t idx)
{
assert(me->commit_idx <= idx);
assert(idx <= raft_get_current_idx(me));
me->commit_idx = idx;
}
void raft_set_last_applied_idx(raft_server_t* me, raft_index_t idx)
void raft_set_last_applied_idx(raft_server_t *me, raft_index_t idx)
{
me->last_applied_idx = idx;
}
raft_index_t raft_get_last_applied_idx(raft_server_t* me)
raft_index_t raft_get_last_applied_idx(raft_server_t *me)
{
return me->last_applied_idx;
}
raft_index_t raft_get_commit_idx(raft_server_t* me)
raft_index_t raft_get_commit_idx(raft_server_t *me)
{
return me->commit_idx;
}
void raft_set_state(raft_server_t* me, int state)
void raft_set_state(raft_server_t *me, int state)
{
/* if became the leader, then update the current leader entry */
if (state == RAFT_STATE_LEADER)
if (state == RAFT_STATE_LEADER) {
me->leader_id = raft_node_get_id(me->node);
}
me->state = state;
}
int raft_get_state(raft_server_t* me)
int raft_get_state(raft_server_t *me)
{
return me->state;
}
const char *raft_get_state_str(raft_server_t* me)
const char *raft_get_state_str(raft_server_t *me)
{
switch (me->state) {
case RAFT_STATE_FOLLOWER:
......@@ -127,81 +131,79 @@ const char *raft_get_state_str(raft_server_t* me)
}
}
raft_node_t* raft_get_node(raft_server_t *me, raft_node_id_t nodeid)
raft_node_t *raft_get_node(raft_server_t *me, raft_node_id_t nodeid)
{
int i;
for (i = 0; i < me->num_nodes; i++)
if (nodeid == raft_node_get_id(me->nodes[i]))
for (int i = 0; i < me->num_nodes; i++) {
if (nodeid == raft_node_get_id(me->nodes[i])) {
return me->nodes[i];
}
}
return NULL;
}
raft_node_t* raft_get_my_node(raft_server_t *me)
raft_node_t *raft_get_my_node(raft_server_t *me)
{
int i;
for (i = 0; i < me->num_nodes; i++)
if (raft_get_nodeid(me) == raft_node_get_id(me->nodes[i]))
return me->nodes[i];
return NULL;
return me->node;
}
raft_node_t* raft_get_node_from_idx(raft_server_t* me, const raft_index_t idx)
raft_node_t *raft_get_node_from_idx(raft_server_t *me, const raft_index_t idx)
{
return me->nodes[idx];
}
raft_node_id_t raft_get_leader_id(raft_server_t* me)
raft_node_id_t raft_get_leader_id(raft_server_t *me)
{
return me->leader_id;
}
raft_node_t* raft_get_leader_node(raft_server_t* me)
raft_node_t *raft_get_leader_node(raft_server_t *me)
{
return raft_get_node(me, me->leader_id);
}
void* raft_get_udata(raft_server_t* me)
void *raft_get_udata(raft_server_t *me)
{
return me->udata;
}
int raft_is_follower(raft_server_t* me)
int raft_is_follower(raft_server_t *me)
{
return me->state == RAFT_STATE_FOLLOWER;
}
int raft_is_leader(raft_server_t* me)
int raft_is_leader(raft_server_t *me)
{
return me->state == RAFT_STATE_LEADER;
}
int raft_is_precandidate(raft_server_t* me)
int raft_is_precandidate(raft_server_t *me)
{
return me->state == RAFT_STATE_PRECANDIDATE;
}
int raft_is_candidate(raft_server_t* me)
int raft_is_candidate(raft_server_t *me)
{
return me->state == RAFT_STATE_CANDIDATE;
}
raft_term_t raft_get_last_log_term(raft_server_t* me)
raft_term_t raft_get_last_log_term(raft_server_t *me)
{
raft_index_t current_idx = raft_get_current_idx(me);
if (0 < current_idx)
{
raft_entry_t* ety = raft_get_entry_from_idx(me, current_idx);
if (ety) {
raft_term_t term = ety->term;
raft_entry_release(ety);
return term;
} else if (me->snapshot_last_idx == current_idx) {
return me->snapshot_last_term;
}
if (current_idx == 0) {
return 0;
}
if (current_idx == me->snapshot_last_idx) {
return me->snapshot_last_term;
}
raft_entry_t *ety = raft_get_entry_from_idx(me, current_idx);
if (ety) {
raft_term_t term = ety->term;
raft_entry_release(ety);
return term;
}
return 0;
......@@ -212,7 +214,7 @@ int raft_snapshot_is_in_progress(raft_server_t *me)
return me->snapshot_in_progress;
}
int raft_is_apply_allowed(raft_server_t* me)
int raft_is_apply_allowed(raft_server_t *me)
{
return !me->disable_apply &&
(!me->snapshot_in_progress || me->nonblocking_apply);
......@@ -238,18 +240,15 @@ raft_term_t raft_get_snapshot_last_term(raft_server_t *me)
int raft_is_single_node_voting_cluster(raft_server_t *me)
{
return (1 == raft_get_num_voting_nodes(me) && raft_node_is_voting(me->node));
return (raft_get_num_voting_nodes(me) == 1 && raft_node_is_voting(me->node));
}
raft_msg_id_t raft_get_msg_id(raft_server_t* me)
raft_msg_id_t raft_get_msg_id(raft_server_t *me)
{
return me->msg_id;
}
/* return the targeted node_id if we are in the middle of attempting a leadership transfer
* return RAFT_NODE_ID_NONE if no leadership transfer is in progress
*/
raft_node_id_t raft_get_transfer_leader(raft_server_t* me)
raft_node_id_t raft_get_transfer_leader(raft_server_t *me)
{
return me->node_transferring_leader_to;
}
......
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