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

Delete private_t typedefs (#93)

Delete private typedefs, use namespace raft_ for exported symbols
parent 07190dc1
......@@ -287,8 +287,8 @@ typedef struct
raft_index_t current_idx;
} msg_appendentries_response_t;
typedef void* raft_server_t;
typedef void* raft_node_t;
typedef struct raft_server raft_server_t;
typedef struct raft_node raft_node_t;
/** Callback for sending request vote messages.
* @param[in] raft The Raft server making this callback
......@@ -903,7 +903,7 @@ raft_server_t* raft_new_with_log(const raft_log_impl_t *log_impl, void *log_arg)
/** De-initialise Raft server.
* Frees all memory */
void raft_destroy(raft_server_t* me_);
void raft_destroy(raft_server_t* me);
/** De-initialise Raft server. */
void raft_clear(raft_server_t* me);
......@@ -941,11 +941,11 @@ raft_node_t* raft_add_node(raft_server_t* me, void* user_data, raft_node_id_t id
* @return
* node if it was successfully added;
* NULL if the node already exists */
raft_node_t* raft_add_non_voting_node(raft_server_t* me_, void* udata, raft_node_id_t id, int is_self);
raft_node_t* raft_add_non_voting_node(raft_server_t* me, void* udata, raft_node_id_t id, int is_self);
/** Remove node.
* @param node The node to be removed. */
void raft_remove_node(raft_server_t* me_, raft_node_t* node);
void raft_remove_node(raft_server_t* me, raft_node_t* node);
/** Set election timeout.
* The amount of time that needs to elapse before we assume the leader is down
......@@ -959,7 +959,7 @@ void raft_set_request_timeout(raft_server_t* me, int msec);
/** Enable/disable library log.
* @param enable 0 to disable*/
void raft_set_log_enabled(raft_server_t* me_, int enable);
void raft_set_log_enabled(raft_server_t* me, int enable);
/** Process events that are dependent on time passing.
* @param[in] msec_elapsed Time in milliseconds since the last call
......@@ -1004,7 +1004,7 @@ int raft_recv_appendentries_response(raft_server_t* me,
* @param[out] resp The resulting response
* @return
* 0 on success */
int raft_recv_snapshot(raft_server_t* me_,
int raft_recv_snapshot(raft_server_t* me,
raft_node_t* node,
msg_snapshot_t *req,
msg_snapshot_response_t *resp);
......@@ -1016,7 +1016,7 @@ int raft_recv_snapshot(raft_server_t* me_,
* 0 on success;
* -1 on error;
* RAFT_ERR_NOT_LEADER server is not the leader */
int raft_recv_snapshot_response(raft_server_t* me_,
int raft_recv_snapshot_response(raft_server_t* me,
raft_node_t* node,
msg_snapshot_response_t *r);
......@@ -1074,7 +1074,7 @@ int 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 currently configured election timeout in milliseconds */
......@@ -1086,7 +1086,7 @@ int raft_get_num_nodes(raft_server_t* me);
/**
* @return number of voting nodes that this server has */
int raft_get_num_voting_nodes(raft_server_t* me_);
int raft_get_num_voting_nodes(raft_server_t* me);
/**
* @return number of items within log */
......@@ -1102,7 +1102,7 @@ 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 */
......@@ -1163,13 +1163,13 @@ raft_entry_t* raft_get_entry_from_idx(raft_server_t* me, raft_index_t idx);
/**
* @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 */
......@@ -1182,13 +1182,13 @@ int 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 */
......@@ -1199,14 +1199,14 @@ void* raft_get_udata(raft_server_t* me);
* @param[in] node The server to vote for
* @return
* 0 on success */
int raft_vote(raft_server_t* me_, raft_node_t* node);
int raft_vote(raft_server_t* me, raft_node_t* node);
/** Vote for a server.
* This should be used to reload persistent state, ie. the voted-for field.
* @param[in] nodeid The server to vote for by nodeid
* @return
* 0 on success */
int raft_vote_for_nodeid(raft_server_t* me_, raft_node_id_t nodeid);
int raft_vote_for_nodeid(raft_server_t* me, raft_node_id_t nodeid);
/** Set the current term.
* This should be used to reload persistent state, ie. the current_term field.
......@@ -1238,20 +1238,20 @@ int raft_pop_entry(raft_server_t* me);
/** 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_,
int raft_msg_entry_response_committed(raft_server_t* me,
const msg_entry_response_t* r);
/** 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* me);
/** 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);
/** 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,
......@@ -1260,17 +1260,17 @@ 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* me);
/** 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* me);
/** Apply all entries up to the commit index
* @return
* 0 on success;
* RAFT_ERR_SHUTDOWN when server MUST shutdown */
int raft_apply_all(raft_server_t* me_);
int raft_apply_all(raft_server_t* me);
/** Become leader
* WARNING: this is a dangerous function call. It could lead to your cluster
......@@ -1304,7 +1304,7 @@ int raft_entry_is_cfg_change(raft_entry_t* ety);
* @return 0 on success
*
**/
int raft_begin_snapshot(raft_server_t *me_, int flags);
int raft_begin_snapshot(raft_server_t *me, int flags);
/** Stop snapshotting.
*
......@@ -1320,7 +1320,7 @@ int raft_begin_snapshot(raft_server_t *me_, int flags);
* 0 on success
* -1 on failure
**/
int raft_end_snapshot(raft_server_t *me_);
int raft_end_snapshot(raft_server_t *me);
/** Cancel snapshotting.
*
......@@ -1330,22 +1330,22 @@ int raft_end_snapshot(raft_server_t *me_);
* The user MUST be sure the original snapshot is left untouched and remains
* usable.
*/
int raft_cancel_snapshot(raft_server_t *me_);
int raft_cancel_snapshot(raft_server_t *me);
/** Check is a snapshot is in progress
**/
int raft_snapshot_is_in_progress(raft_server_t *me_);
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
**/
raft_entry_t *raft_get_last_applied_entry(raft_server_t *me_);
raft_entry_t *raft_get_last_applied_entry(raft_server_t *me);
raft_index_t raft_get_first_entry_idx(raft_server_t* me_);
raft_index_t raft_get_first_entry_idx(raft_server_t* me);
/** Start loading snapshot
*
......@@ -1363,7 +1363,7 @@ raft_index_t raft_get_first_entry_idx(raft_server_t* me_);
* -1 on failure
* RAFT_ERR_SNAPSHOT_ALREADY_LOADED
**/
int raft_begin_load_snapshot(raft_server_t *me_,
int raft_begin_load_snapshot(raft_server_t *me,
raft_term_t last_included_term,
raft_index_t last_included_index);
......@@ -1373,19 +1373,19 @@ int raft_begin_load_snapshot(raft_server_t *me_,
* 0 on success
* -1 on failure
**/
int raft_end_load_snapshot(raft_server_t *me_);
int raft_end_load_snapshot(raft_server_t *me);
raft_index_t raft_get_snapshot_last_idx(raft_server_t *me_);
raft_index_t raft_get_snapshot_last_idx(raft_server_t *me);
raft_term_t raft_get_snapshot_last_term(raft_server_t *me_);
raft_term_t raft_get_snapshot_last_term(raft_server_t *me);
void raft_set_snapshot_metadata(raft_server_t *me_, raft_term_t term, raft_index_t idx);
void raft_set_snapshot_metadata(raft_server_t *me, raft_term_t term, raft_index_t idx);
/** Check if a node is active.
* 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* me);
/** Make the node active.
*
......@@ -1394,17 +1394,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* me, 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* me);
/** 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* me);
/**
* Register custom heap management functions, to be used if an alternative
......@@ -1418,21 +1418,21 @@ 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* me, 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* me, 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.
*/
void *raft_get_log(raft_server_t* me_);
void *raft_get_log(raft_server_t* me);
/** Backward compatible callbacks for log events, implemented by the
* default in-memory log implementation.
......@@ -1509,28 +1509,28 @@ void raft_entry_release_list(raft_entry_t **ety_list, size_t len);
*/
extern const raft_log_impl_t raft_log_internal_impl;
void raft_handle_append_cfg_change(raft_server_t* me_, raft_entry_t* ety, raft_index_t idx);
void raft_handle_append_cfg_change(raft_server_t* me, raft_entry_t* ety, raft_index_t idx);
int raft_queue_read_request(raft_server_t* me_, func_read_request_callback_f cb, void *cb_arg);
int raft_queue_read_request(raft_server_t* me, func_read_request_callback_f cb, void *cb_arg);
/** Attempt to process read queue.
*/
void raft_process_read_queue(raft_server_t* me_);
void raft_process_read_queue(raft_server_t* me);
/*
* invoke a leadership transfer to targeted node
* node_id = targeted node
* timeout = timeout in ms before this transfer is aborted. if 0, use default election timeout
*/
int raft_transfer_leader(raft_server_t* me_, raft_node_id_t node_id, long timeout);
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_);
int raft_timeout_now(raft_server_t* me);
raft_index_t raft_get_num_snapshottable_logs(raft_server_t* me_);
raft_index_t raft_get_num_snapshottable_logs(raft_server_t* me);
/** Disable auto flush mode. Default is enabled.
*
......
......@@ -3,58 +3,58 @@
#include "raft_types.h"
typedef void* log_t;
typedef struct raft_log raft_log_t;
log_t* log_new(void);
raft_log_t* raft_log_new(void);
log_t* log_alloc(raft_index_t initial_size);
raft_log_t* raft_log_alloc(raft_index_t initial_size);
void log_set_callbacks(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 log_free(log_t* me_);
void raft_log_free(raft_log_t* me);
void log_clear(log_t* me_);
void raft_log_clear(raft_log_t* me);
void log_clear_entries(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 log_append_entry(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 log_count(log_t* me_);
raft_index_t raft_log_count(raft_log_t* me);
/**
* Delete all logs from this log onwards */
int log_delete(log_t* me_, raft_index_t idx);
int raft_log_delete(raft_log_t* me, raft_index_t idx);
/**
* Empty the queue. */
void log_empty(log_t * me_);
void raft_log_empty(raft_log_t * me);
/**
* Remove oldest entry. Set *etyp to oldest entry on success. */
int log_poll(log_t * me_, void** 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** log_get_from_idx(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, long *n_etys);
raft_entry_t* log_get_at_idx(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 *log_peektail(log_t * me_);
raft_entry_t *raft_log_peektail(raft_log_t * me);
raft_index_t log_get_current_idx(log_t* me_);
raft_index_t raft_log_get_current_idx(raft_log_t* me);
int log_load_from_snapshot(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 log_get_base(log_t* me_);
raft_index_t raft_log_get_base(raft_log_t* me);
#endif /* RAFT_LOG_H_ */
......@@ -25,7 +25,7 @@ typedef struct raft_read_request {
struct raft_read_request *next;
} raft_read_request_t;
typedef struct {
struct raft_server {
/* Persistent state: */
/* the server's best guess of what the current term is
......@@ -54,7 +54,7 @@ typedef struct {
/* amount of time left till timeout */
int timeout_elapsed;
raft_node_t* nodes;
raft_node_t** nodes;
int num_nodes;
int election_timeout;
......@@ -129,7 +129,7 @@ typedef struct {
raft_index_t next_sync_index;
int log_enabled;
} raft_server_private_t;
};
int raft_election_start(raft_server_t* me, int skip_precandidate);
......@@ -137,9 +137,9 @@ int raft_become_candidate(raft_server_t* me);
int raft_become_precandidate(raft_server_t* me);
void raft_randomize_election_timeout(raft_server_t* me_);
void raft_randomize_election_timeout(raft_server_t* me);
void raft_update_quorum_meta(raft_server_t* me_, raft_msg_id_t id);
void raft_update_quorum_meta(raft_server_t* me, raft_msg_id_t id);
/**
* @return 0 on error */
......@@ -147,36 +147,36 @@ int raft_send_requestvote(raft_server_t* me, raft_node_t* node);
int raft_send_appendentries(raft_server_t* me, raft_node_t* node);
int raft_send_appendentries_all(raft_server_t* me_);
int raft_send_appendentries_all(raft_server_t* me);
/**
* Apply entry at lastApplied + 1. Entry becomes 'committed'.
* @return 1 if entry committed, 0 otherwise */
int raft_apply_entry(raft_server_t* me_);
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_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);
void raft_node_free(raft_node_t* me_);
void raft_node_free(raft_node_t* me);
void raft_node_set_match_idx(raft_node_t* node, raft_index_t idx);
void raft_node_vote_for_me(raft_node_t* me_, int vote);
void raft_node_vote_for_me(raft_node_t* me, int vote);
int raft_node_has_vote_for_me(raft_node_t* me_);
int raft_node_has_vote_for_me(raft_node_t* me);
void raft_node_set_has_sufficient_logs(raft_node_t* me_);
void raft_node_set_has_sufficient_logs(raft_node_t* me);
int raft_is_single_node_voting_cluster(raft_server_t *me_);
int raft_is_single_node_voting_cluster(raft_server_t *me);
int raft_votes_is_majority(int nnodes, int nvotes);
void raft_node_set_last_ack(raft_node_t* me_, raft_msg_id_t msgid, raft_term_t term);
void raft_node_set_last_ack(raft_node_t* me, raft_msg_id_t msgid, raft_term_t term);
raft_msg_id_t raft_node_get_last_acked_msgid(raft_node_t* me_);
raft_msg_id_t raft_node_get_last_acked_msgid(raft_node_t* me);
/* Heap functions */
extern void *(*raft_malloc)(size_t size);
......@@ -185,17 +185,17 @@ 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 *me, 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 *me);
/* 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);
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 *me);
void raft_node_set_snapshot_offset(raft_node_t *me_, raft_size_t snapshot_offset);
void raft_node_set_snapshot_offset(raft_node_t *me, raft_size_t snapshot_offset);
#endif /* RAFT_PRIVATE_H_ */
......@@ -17,7 +17,7 @@
#define INITIAL_CAPACITY 10
typedef struct
struct raft_log
{
/* size of array */
raft_index_t size;
......@@ -37,15 +37,15 @@ typedef struct
raft_log_cbs_t cb;
void* raft;
} log_private_t;
};
static int mod(raft_index_t a, raft_index_t b)
static raft_index_t mod(raft_index_t a, raft_index_t b)
{
int r = a % b;
raft_index_t r = a % b;
return r < 0 ? r + b : r;
}
static int __ensurecapacity(log_private_t * me)
static int ensure_capacity(raft_log_t *me)
{
raft_index_t i, j;
raft_entry_t **temp;
......@@ -74,58 +74,53 @@ static int __ensurecapacity(log_private_t * me)
return 0;
}
int log_load_from_snapshot(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;
log_private_t* me = (log_private_t*)me_;
log_clear_entries(me_);
log_clear(me_);
raft_log_clear_entries(me);
raft_log_clear(me);
me->base = idx;
return 0;
}
log_t* log_alloc(raft_index_t initial_size)
raft_log_t *raft_log_alloc(raft_index_t initial_size)
{
log_private_t* me = raft_calloc(1, sizeof(log_private_t));
raft_log_t *me = raft_calloc(1, sizeof(*me));
if (!me)
return NULL;
me->size = initial_size;
log_clear((log_t*)me);
raft_log_clear(me);
me->entries = raft_calloc(1, sizeof(raft_entry_t *) * me->size);
if (!me->entries) {
raft_free(me);
return NULL;
}
return (log_t*)me;
return me;
}
log_t* log_new(void)
raft_log_t *raft_log_new(void)
{
return log_alloc(INITIAL_CAPACITY);
return raft_log_alloc(INITIAL_CAPACITY);
}
void log_set_callbacks(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)
{
log_private_t* me = (log_private_t*)me_;
me->raft = raft;
me->cb = *funcs;
}
void log_clear(log_t* me_)
void raft_log_clear(raft_log_t *me)
{
log_private_t* me = (log_private_t*)me_;
me->count = 0;
me->back = 0;
me->front = 0;
me->base = 0;
}
void log_clear_entries(log_t* me_)
void raft_log_clear_entries(raft_log_t* me)
{
log_private_t* me = (log_private_t*)me_;
raft_index_t i;
if (!me->count || !me->cb.log_clear)
......@@ -139,17 +134,16 @@ void log_clear_entries(log_t* me_)
}
/** TODO: rename log_append */
int log_append_entry(log_t* me_, raft_entry_t* ety)
int raft_log_append_entry(raft_log_t *me, raft_entry_t*c)
{
log_private_t* me = (log_private_t*)me_;
raft_index_t idx = me->base + me->count + 1;
int e;
e = __ensurecapacity(me);
e = ensure_capacity(me);
if (e != 0)
return e;
me->entries[me->back] = ety;
me->entries[me->back] = c;
if (me->cb.log_offer)
{
......@@ -166,9 +160,8 @@ int log_append_entry(log_t* me_, raft_entry_t* ety)
return 0;
}
raft_entry_t** log_get_from_idx(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, long *n_etys)
{
log_private_t* me = (log_private_t*)me_;
raft_index_t i;
assert(0 <= idx - 1);
......@@ -193,9 +186,8 @@ raft_entry_t** log_get_from_idx(log_t* me_, raft_index_t idx, long *n_etys)
return &me->entries[i];
}
raft_entry_t* log_get_at_idx(log_t* me_, raft_index_t idx)
raft_entry_t* raft_log_get_at_idx(raft_log_t *me, raft_index_t idx)
{
log_private_t* me = (log_private_t*)me_;
raft_index_t i;
if (idx == 0)
......@@ -211,15 +203,13 @@ raft_entry_t* log_get_at_idx(log_t* me_, raft_index_t idx)
return me->entries[i];
}
raft_index_t log_count(log_t* me_)
raft_index_t raft_log_count(raft_log_t *me)
{
return ((log_private_t*)me_)->count;
return me->count;
}
static int __log_delete(log_t* me_, raft_index_t idx, func_entry_notify_f cb, void *cb_arg)
static int log_delete(raft_log_t* me, raft_index_t idx, func_entry_notify_f cb, void *cb_arg)
{
log_private_t* me = (log_private_t*)me_;
if (0 == idx)
return -1;
......@@ -250,14 +240,13 @@ static int __log_delete(log_t* me_, raft_index_t idx, func_entry_notify_f cb, vo
return 0;
}
int log_delete(log_t* me_, raft_index_t idx)
int raft_log_delete(raft_log_t *me, raft_index_t idx)
{
return __log_delete(me_, idx, NULL, NULL);
return log_delete(me, idx, NULL, NULL);
}
int log_poll(log_t * me_, void** etyp)
int raft_log_poll(raft_log_t *me, raft_entry_t **etyp)
{
log_private_t* me = (log_private_t*)me_;
raft_index_t idx = me->base + 1;
if (0 == me->count)
......@@ -283,10 +272,8 @@ int log_poll(log_t * me_, void** etyp)
return 0;
}
raft_entry_t *log_peektail(log_t * me_)
raft_entry_t *raft_log_peektail(raft_log_t *me)
{
log_private_t* me = (log_private_t*)me_;
if (0 == me->count)
return NULL;
......@@ -296,32 +283,27 @@ raft_entry_t *log_peektail(log_t * me_)
return me->entries[me->back - 1];
}
void log_empty(log_t * me_)
void raft_log_empty(raft_log_t *me)
{
log_private_t* me = (log_private_t*)me_;
me->front = 0;
me->back = 0;
me->count = 0;
}
void log_free(log_t * me_)
void raft_log_free(raft_log_t *me)
{
log_private_t* me = (log_private_t*)me_;
raft_free(me->entries);
raft_free(me);
}
raft_index_t log_get_current_idx(log_t* me_)
raft_index_t raft_log_get_current_idx(raft_log_t *me)
{
log_private_t* me = (log_private_t*)me_;
return log_count(me_) + me->base;
return raft_log_count(me) + me->base;
}
raft_index_t log_get_base(log_t* me_)
raft_index_t raft_log_get_base(raft_log_t *me)
{
return ((log_private_t*)me_)->base;
return me->base;
}
/**
......@@ -336,49 +318,49 @@ raft_index_t log_get_base(log_t* me_)
* log implementations.
*/
void *__log_init(void *raft, void *arg)
static void *log_init(void *raft, void *arg)
{
log_t *log = log_new();
raft_log_t *log = raft_log_new();
if (arg) {
log_set_callbacks(log, arg, raft);
raft_log_set_callbacks(log, arg, raft);
}
return log;
}
static void __log_free(void *log)
static void log_free(void *log)
{
log_free(log);
raft_log_free(log);
}
static void __log_reset(void *log, raft_index_t first_idx, raft_term_t term)
static void log_reset(void *log, raft_index_t first_idx, raft_term_t term)
{
(void) term;
log_clear_entries(log);
log_clear(log);
raft_log_clear_entries(log);
raft_log_clear(log);
assert(first_idx >= 1);
((log_private_t*) log)->base = first_idx - 1;
((raft_log_t *) log)->base = first_idx - 1;
}
static int __log_append(void *log, raft_entry_t *entry)
static int log_append(void *log, raft_entry_t *entry)
{
raft_entry_hold(entry);
return log_append_entry(log, entry);
return raft_log_append_entry(log, entry);
}
static raft_entry_t *__log_get(void *log, raft_index_t idx)
static raft_entry_t *log_get(void *log, raft_index_t idx)
{
raft_entry_t *e = log_get_at_idx(log, idx);
raft_entry_t *e = raft_log_get_at_idx(log, idx);
if (e != NULL)
raft_entry_hold(e);
return e;
}
static int __log_get_batch(void *log, raft_index_t idx, int entries_n, raft_entry_t **entries)
static int log_get_batch(void *log, raft_index_t idx, int entries_n, raft_entry_t **entries)
{
long n, i;
raft_entry_t **r = log_get_from_idx(log, idx, &n);
raft_entry_t **r = raft_log_get_from_idx(log, idx, &n);
if (!r || n < 1) {
return 0;
......@@ -394,16 +376,16 @@ static int __log_get_batch(void *log, raft_index_t idx, int entries_n, raft_entr
return (int) n;
}
static int __log_pop(void *log, raft_index_t from_idx, func_entry_notify_f cb, void *cb_arg)
static int log_pop(void *log, raft_index_t from_idx, func_entry_notify_f cb, void *cb_arg)
{
return __log_delete(log, from_idx, cb, cb_arg);
return log_delete(log, from_idx, cb, cb_arg);
}
static int __log_poll(void *log, raft_index_t first_idx)
static int log_poll(void *log, raft_index_t first_idx)
{
while (log_get_base(log) + 1 < first_idx) {
while (raft_log_get_base(log) + 1 < first_idx) {
raft_entry_t *ety;
int e = log_poll(log, (void **) &ety);
int e = raft_log_poll(log, &ety);
if (e < 0)
return e;
......@@ -412,38 +394,38 @@ static int __log_poll(void *log, raft_index_t first_idx)
return 0;
}
static raft_index_t __log_first_idx(void *log)
static raft_index_t log_first_idx(void *log)
{
return log_get_base(log) + 1;
return raft_log_get_base(log) + 1;
}
static raft_index_t __log_current_idx(void *log)
static raft_index_t log_current_idx(void *log)
{
return log_get_current_idx(log);
return raft_log_get_current_idx(log);
}
static raft_index_t __log_count(void *log)
static raft_index_t log_count(void *log)
{
return log_count(log);
return raft_log_count(log);
}
static int __log_sync(void *log)
static int log_sync(void *log)
{
(void) log;
return 0;
}
const raft_log_impl_t raft_log_internal_impl = {
.init = __log_init,
.free = __log_free,
.reset = __log_reset,
.append = __log_append,
.poll = __log_poll,
.pop = __log_pop,
.get = __log_get,
.get_batch = __log_get_batch,
.first_idx = __log_first_idx,
.current_idx = __log_current_idx,
.count = __log_count,
.sync = __log_sync
.init = log_init,
.free = log_free,
.reset = log_reset,
.append = log_append,
.poll = log_poll,
.pop = log_pop,
.get = log_get,
.get_batch = log_get_batch,
.first_idx = log_first_idx,
.current_idx = log_current_idx,
.count = log_count,
.sync = log_sync
};
......@@ -22,7 +22,7 @@
#define RAFT_NODE_VOTING_COMMITTED (1 << 4)
#define RAFT_NODE_ADDITION_COMMITTED (1 << 5)
typedef struct
struct raft_node
{
void* udata;
......@@ -40,13 +40,13 @@ typedef struct
raft_term_t last_acked_term;
raft_msg_id_t last_acked_msgid;
raft_msg_id_t max_seen_msgid;
} raft_node_private_t;
};
raft_node_t* raft_node_new(void* udata, raft_node_id_t id)
{
raft_node_private_t* me;
raft_node_t* me;
me = raft_calloc(1, sizeof(raft_node_private_t));
me = raft_calloc(1, sizeof(*me));
if (!me)
return NULL;
......@@ -55,185 +55,160 @@ raft_node_t* raft_node_new(void* udata, raft_node_id_t id)
me->match_idx = 0;
me->id = id;
me->flags = RAFT_NODE_VOTING;
return (raft_node_t*)me;
return me;
}
void raft_node_free(raft_node_t* me_)
void raft_node_free(raft_node_t* me)
{
raft_free(me_);
raft_free(me);
}
raft_index_t raft_node_get_next_idx(raft_node_t* me_)
raft_index_t raft_node_get_next_idx(raft_node_t* me)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return me->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* me, raft_index_t idx)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
/* log index begins at 1 */
me->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* me)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return me->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* me, raft_index_t idx)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
me->match_idx = idx;
}
void* raft_node_get_udata(raft_node_t* me_)
void* raft_node_get_udata(raft_node_t* me)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return me->udata;
}
void raft_node_set_udata(raft_node_t* me_, void* udata)
void raft_node_set_udata(raft_node_t* me, void* udata)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
me->udata = udata;
}
void raft_node_vote_for_me(raft_node_t* me_, const int vote)
void raft_node_vote_for_me(raft_node_t* me, const int vote)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
if (vote)
me->flags |= RAFT_NODE_VOTED_FOR_ME;
else
me->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* me)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return (me->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* me, int voting)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
if (voting)
{
assert(!raft_node_is_voting(me_));
assert(!raft_node_is_voting(me));
me->flags |= RAFT_NODE_VOTING;
}
else
{
assert(raft_node_is_voting(me_));
assert(raft_node_is_voting(me));
me->flags &= ~RAFT_NODE_VOTING;
}
}
int raft_node_is_voting(raft_node_t* me_)
int raft_node_is_voting(raft_node_t* me)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return (me->flags & RAFT_NODE_VOTING && !(me->flags & RAFT_NODE_INACTIVE));
}
int raft_node_has_sufficient_logs(raft_node_t* me_)
int raft_node_has_sufficient_logs(raft_node_t* me)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return (me->flags & RAFT_NODE_HAS_SUFFICIENT_LOG) != 0;
}
void raft_node_set_has_sufficient_logs(raft_node_t* me_)
void raft_node_set_has_sufficient_logs(raft_node_t* me)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
me->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* me, int active)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
if (!active)
me->flags |= RAFT_NODE_INACTIVE;
else
me->flags &= ~RAFT_NODE_INACTIVE;
}
int raft_node_is_active(raft_node_t* me_)
int raft_node_is_active(raft_node_t* me)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return (me->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* me, int voting)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
if (voting)
me->flags |= RAFT_NODE_VOTING_COMMITTED;
else
me->flags &= ~RAFT_NODE_VOTING_COMMITTED;
}
int raft_node_is_voting_committed(raft_node_t* me_)
int raft_node_is_voting_committed(raft_node_t* me)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return (me->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* me)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return me != NULL ? me->id : -1;
}
void raft_node_set_addition_committed(raft_node_t* me_, int committed)
void raft_node_set_addition_committed(raft_node_t* me, int committed)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
if (committed)
me->flags |= RAFT_NODE_ADDITION_COMMITTED;
else
me->flags &= ~RAFT_NODE_ADDITION_COMMITTED;
}
int raft_node_is_addition_committed(raft_node_t* me_)
int raft_node_is_addition_committed(raft_node_t* me)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return (me->flags & RAFT_NODE_ADDITION_COMMITTED) != 0;
}
void raft_node_set_last_ack(raft_node_t* me_, raft_msg_id_t msgid, raft_term_t term)
void raft_node_set_last_ack(raft_node_t* me, raft_msg_id_t msgid, raft_term_t term)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
me->last_acked_msgid = msgid;
me->last_acked_term = term;
}
raft_msg_id_t raft_node_get_last_acked_msgid(raft_node_t* me_)
raft_msg_id_t raft_node_get_last_acked_msgid(raft_node_t* me)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return me->last_acked_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 *me, raft_msg_id_t msg_id)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
if (msg_id > me->max_seen_msgid) {
me->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 *me)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return me->max_seen_msgid;
}
raft_size_t raft_node_get_snapshot_offset(raft_node_t *me_)
raft_size_t raft_node_get_snapshot_offset(raft_node_t *me)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return me->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 *me, raft_size_t snapshot_offset)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
me->snapshot_offset = snapshot_offset;
}
This diff is collapsed.
......@@ -13,54 +13,48 @@
#include "raft.h"
#include "raft_private.h"
void raft_set_election_timeout(raft_server_t* me_, int millisec)
void raft_set_election_timeout(raft_server_t* me, int millisec)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
me->election_timeout = millisec;
raft_update_quorum_meta(me_, me->last_acked_msg_id);
raft_randomize_election_timeout(me_);
raft_update_quorum_meta(me, me->last_acked_msg_id);
raft_randomize_election_timeout(me);
}
void raft_set_request_timeout(raft_server_t* me_, int millisec)
void raft_set_request_timeout(raft_server_t* me, int millisec)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
me->request_timeout = millisec;
}
void raft_set_log_enabled(raft_server_t* me_, int enable)
void raft_set_log_enabled(raft_server_t* me, int enable)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
me->log_enabled = enable;
}
raft_node_id_t raft_get_nodeid(raft_server_t* me_)
raft_node_id_t raft_get_nodeid(raft_server_t* me)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
if (!me->node)
return -1;
return raft_node_get_id(me->node);
}
int raft_get_election_timeout(raft_server_t* me_)
int raft_get_election_timeout(raft_server_t* me)
{
return ((raft_server_private_t*)me_)->election_timeout;
return me->election_timeout;
}
int raft_get_request_timeout(raft_server_t* me_)
int raft_get_request_timeout(raft_server_t* me)
{
return ((raft_server_private_t*)me_)->request_timeout;
return me->request_timeout;
}
int raft_get_num_nodes(raft_server_t* me_)
int raft_get_num_nodes(raft_server_t* me)
{
return ((raft_server_private_t*)me_)->num_nodes;
return me->num_nodes;
}
int raft_get_num_voting_nodes(raft_server_t* me_)
int raft_get_num_voting_nodes(raft_server_t* me)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
int i, num = 0;
for (i = 0; i < me->num_nodes; i++)
if (raft_node_is_voting(me->nodes[i]))
......@@ -68,32 +62,29 @@ int raft_get_num_voting_nodes(raft_server_t* me_)
return num;
}
int raft_get_timeout_elapsed(raft_server_t* me_)
int raft_get_timeout_elapsed(raft_server_t* me)
{
return ((raft_server_private_t*)me_)->timeout_elapsed;
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)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
return me->log_impl->count(me->log);
}
int raft_get_voted_for(raft_server_t* me_)
int raft_get_voted_for(raft_server_t* me)
{
raft_server_private_t* me = (raft_server_private_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)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
if (me->current_term < term)
{
int voted_for = -1;
if (me->cb.persist_term)
{
int e = me->cb.persist_term(me_, me->udata, term, voted_for);
int e = me->cb.persist_term(me, me->udata, term, voted_for);
if (0 != e)
return e;
}
......@@ -104,58 +95,53 @@ 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 ((raft_server_private_t*)me_)->current_term;
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)
{
raft_server_private_t* me = (raft_server_private_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)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
assert(me->commit_idx <= idx);
assert(idx <= raft_get_current_idx(me_));
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)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
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 ((raft_server_private_t*)me_)->last_applied_idx;
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 ((raft_server_private_t*)me_)->commit_idx;
return me->commit_idx;
}
void raft_set_state(raft_server_t* me_, int state)
void raft_set_state(raft_server_t* me, int state)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
/* if became the leader, then update the current leader entry */
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 ((raft_server_private_t*)me_)->state;
return me->state;
}
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)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
int i;
for (i = 0; i < me->num_nodes; i++)
......@@ -165,136 +151,125 @@ raft_node_t* raft_get_node(raft_server_t *me_, raft_node_id_t nodeid)
return NULL;
}
raft_node_t* raft_get_my_node(raft_server_t *me_)
raft_node_t* raft_get_my_node(raft_server_t *me)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
int i;
for (i = 0; i < me->num_nodes; i++)
if (raft_get_nodeid(me_) == raft_node_get_id(me->nodes[i]))
if (raft_get_nodeid(me) == raft_node_get_id(me->nodes[i]))
return me->nodes[i];
return NULL;
}
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)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
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)
{
raft_server_private_t* me = (void*)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)
{
raft_server_private_t* me = (void*)me_;
return raft_get_node(me_, me->leader_id);
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 ((raft_server_private_t*)me_)->udata;
return me->udata;
}
int raft_is_follower(raft_server_t* me_)
int raft_is_follower(raft_server_t* me)
{
return raft_get_state(me_) == RAFT_STATE_FOLLOWER;
return raft_get_state(me) == RAFT_STATE_FOLLOWER;
}
int raft_is_leader(raft_server_t* me_)
int raft_is_leader(raft_server_t* me)
{
return raft_get_state(me_) == RAFT_STATE_LEADER;
return raft_get_state(me) == RAFT_STATE_LEADER;
}
int raft_is_precandidate(raft_server_t* me_)
int raft_is_precandidate(raft_server_t* me)
{
return raft_get_state(me_) == RAFT_STATE_PRECANDIDATE;
return raft_get_state(me) == RAFT_STATE_PRECANDIDATE;
}
int raft_is_candidate(raft_server_t* me_)
int raft_is_candidate(raft_server_t* me)
{
return raft_get_state(me_) == RAFT_STATE_CANDIDATE;
return raft_get_state(me) == 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_);
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);
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 (raft_get_snapshot_last_idx(me_) == current_idx) {
return raft_get_snapshot_last_term(me_);
} else if (raft_get_snapshot_last_idx(me) == current_idx) {
return raft_get_snapshot_last_term(me);
}
}
return 0;
}
int raft_snapshot_is_in_progress(raft_server_t *me_)
int raft_snapshot_is_in_progress(raft_server_t *me)
{
return ((raft_server_private_t*)me_)->snapshot_in_progress;
return me->snapshot_in_progress;
}
int raft_is_apply_allowed(raft_server_t* me_)
int raft_is_apply_allowed(raft_server_t* me)
{
return (!raft_snapshot_is_in_progress(me_) ||
(((raft_server_private_t*)me_)->snapshot_flags & RAFT_SNAPSHOT_NONBLOCKING_APPLY));
return (!raft_snapshot_is_in_progress(me) ||
(me->snapshot_flags & RAFT_SNAPSHOT_NONBLOCKING_APPLY));
}
raft_entry_t *raft_get_last_applied_entry(raft_server_t *me_)
raft_entry_t *raft_get_last_applied_entry(raft_server_t *me)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
if (raft_get_last_applied_idx(me_) == 0)
if (raft_get_last_applied_idx(me) == 0)
return NULL;
return me->log_impl->get(me->log, raft_get_last_applied_idx(me_));
return me->log_impl->get(me->log, raft_get_last_applied_idx(me));
}
raft_index_t raft_get_snapshot_last_idx(raft_server_t *me_)
raft_index_t raft_get_snapshot_last_idx(raft_server_t *me)
{
return ((raft_server_private_t*)me_)->snapshot_last_idx;
return me->snapshot_last_idx;
}
raft_term_t raft_get_snapshot_last_term(raft_server_t *me_)
raft_term_t raft_get_snapshot_last_term(raft_server_t *me)
{
return ((raft_server_private_t*)me_)->snapshot_last_term;
return me->snapshot_last_term;
}
void raft_set_snapshot_metadata(raft_server_t *me_, raft_term_t term, raft_index_t idx)
void raft_set_snapshot_metadata(raft_server_t *me, raft_term_t term, raft_index_t idx)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
me->last_applied_idx = idx;
me->snapshot_last_term = term;
me->snapshot_last_idx = idx;
}
int raft_is_single_node_voting_cluster(raft_server_t *me_)
int raft_is_single_node_voting_cluster(raft_server_t *me)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
return (1 == raft_get_num_voting_nodes(me_) && raft_node_is_voting(me->node));
return (1 == raft_get_num_voting_nodes(me) && 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)
{
raft_server_private_t* me = (raft_server_private_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)
{
raft_server_private_t* me = (raft_server_private_t*) me_;
return me->node_transferring_leader_to;
}
......@@ -47,7 +47,7 @@ class CoreTestCase(unittest.TestCase):
r = self.r.lib
unique_id = 1
l = r.log_alloc(1)
l = r.raft_log_alloc(1)
log = Log()
......@@ -57,32 +57,31 @@ class CoreTestCase(unittest.TestCase):
entry.id = unique_id
unique_id += 1
ret = r.log_append_entry(l, entry)
ret = r.raft_log_append_entry(l, entry)
assert ret == 0
log.append(entry)
elif cmd == 'poll':
entry_ptr = self.r.ffi.new('void**')
entry_ptr = self.r.ffi.new('raft_entry_t**')
if log.entries:
ret = r.log_poll(l, entry_ptr)
ret = r.raft_log_poll(l, entry_ptr)
assert ret == 0
ety_expected = log.poll()
ety_actual = self.r.ffi.cast('raft_entry_t**', entry_ptr)[0]
assert ety_actual.id == ety_expected.id
assert entry_ptr[0].id == ety_expected.id
elif isinstance(cmd, int):
if log.entries:
log.delete(cmd)
ret = r.log_delete(l, cmd)
ret = r.raft_log_delete(l, cmd)
assert ret == 0
else:
assert False
self.assertEqual(r.log_count(l), log.count())
self.assertEqual(r.raft_log_count(l), log.count())
if __name__ == '__main__':
......
......@@ -18,7 +18,7 @@ static void __LOG_APPEND_ENTRY(void *l, int id, raft_term_t term, const char *da
{
raft_entry_t *e = __MAKE_ENTRY(id, term, data);
raft_entry_hold(e); /* need an extra ref because tests assume it lives on */
log_append_entry(l, e);
raft_log_append_entry(l, e);
}
static void __LOG_APPEND_ENTRIES_SEQ_ID(void *l, int count, int id, raft_term_t term, const char *data)
......@@ -27,7 +27,7 @@ static void __LOG_APPEND_ENTRIES_SEQ_ID(void *l, int count, int id, raft_term_t
for (i = 0; i < count; i++) {
raft_entry_t *e = __MAKE_ENTRY(id++, term, data);
raft_entry_hold(e); /* need an extra ref because tests assume it lives on */
log_append_entry(l, e);
raft_log_append_entry(l, e);
}
}
......@@ -87,7 +87,7 @@ void* __set_up()
void* queue = llqueue_new();
void *r = raft_new();
raft_set_callbacks(r, &funcs, queue);
log_set_callbacks(raft_get_log(r), &log_funcs, r);
raft_log_set_callbacks(raft_get_log(r), &log_funcs, r);
return r;
}
......@@ -95,8 +95,8 @@ void TestLog_new_is_empty(CuTest * tc)
{
void *l;
l = log_new();
CuAssertTrue(tc, 0 == log_count(l));
l = raft_log_new();
CuAssertTrue(tc, 0 == raft_log_count(l));
}
void TestLog_append_is_not_empty(CuTest * tc)
......@@ -105,38 +105,38 @@ void TestLog_append_is_not_empty(CuTest * tc)
void *r = raft_new();
l = log_new();
l = raft_log_new();
raft_log_cbs_t funcs = {
.log_offer = __log_offer
};
log_set_callbacks(l, &funcs, r);
raft_log_set_callbacks(l, &funcs, r);
__LOG_APPEND_ENTRY(l, 1, 0, NULL);
CuAssertIntEquals(tc, 1, log_count(l));
CuAssertIntEquals(tc, 1, raft_log_count(l));
}
void TestLog_get_at_idx(CuTest * tc)
{
void *l;
l = log_new();
l = raft_log_new();
__LOG_APPEND_ENTRIES_SEQ_ID(l, 3, 1, 0, NULL);
CuAssertIntEquals(tc, 3, log_count(l));
CuAssertIntEquals(tc, 3, raft_log_count(l));
CuAssertIntEquals(tc, 1, log_get_at_idx(l, 1)->id);
CuAssertIntEquals(tc, 2, log_get_at_idx(l, 2)->id);
CuAssertIntEquals(tc, 3, log_get_at_idx(l, 3)->id);
CuAssertIntEquals(tc, 1, raft_log_get_at_idx(l, 1)->id);
CuAssertIntEquals(tc, 2, raft_log_get_at_idx(l, 2)->id);
CuAssertIntEquals(tc, 3, raft_log_get_at_idx(l, 3)->id);
}
void TestLog_get_at_idx_returns_null_where_out_of_bounds(CuTest * tc)
{
void *l;
l = log_new();
CuAssertTrue(tc, NULL == log_get_at_idx(l, 0));
CuAssertTrue(tc, NULL == log_get_at_idx(l, 1));
l = raft_log_new();
CuAssertTrue(tc, NULL == raft_log_get_at_idx(l, 0));
CuAssertTrue(tc, NULL == raft_log_get_at_idx(l, 1));
__LOG_APPEND_ENTRY(l, 1, 0, NULL);
CuAssertTrue(tc, NULL == log_get_at_idx(l, 2));
CuAssertTrue(tc, NULL == raft_log_get_at_idx(l, 2));
}
void TestLog_delete(CuTest * tc)
......@@ -154,26 +154,26 @@ void TestLog_delete(CuTest * tc)
raft_set_callbacks(r, &funcs, queue);
l = log_new();
log_set_callbacks(l, &log_funcs, r);
l = raft_log_new();
raft_log_set_callbacks(l, &log_funcs, r);
__LOG_APPEND_ENTRIES_SEQ_ID(l, 3, 1, 0, NULL);
CuAssertIntEquals(tc, 3, log_count(l));
CuAssertIntEquals(tc, 3, log_get_current_idx(l));
CuAssertIntEquals(tc, 3, raft_log_count(l));
CuAssertIntEquals(tc, 3, raft_log_get_current_idx(l));
log_delete(l, 3);
CuAssertIntEquals(tc, 2, log_count(l));
raft_log_delete(l, 3);
CuAssertIntEquals(tc, 2, raft_log_count(l));
CuAssertIntEquals(tc, 3, ((raft_entry_t*)llqueue_poll(queue))->id);
CuAssertIntEquals(tc, 2, log_count(l));
CuAssertTrue(tc, NULL == log_get_at_idx(l, 3));
CuAssertIntEquals(tc, 2, raft_log_count(l));
CuAssertTrue(tc, NULL == raft_log_get_at_idx(l, 3));
log_delete(l, 2);
CuAssertIntEquals(tc, 1, log_count(l));
CuAssertTrue(tc, NULL == log_get_at_idx(l, 2));
raft_log_delete(l, 2);
CuAssertIntEquals(tc, 1, raft_log_count(l));
CuAssertTrue(tc, NULL == raft_log_get_at_idx(l, 2));
log_delete(l, 1);
CuAssertIntEquals(tc, 0, log_count(l));
CuAssertTrue(tc, NULL == log_get_at_idx(l, 1));
raft_log_delete(l, 1);
CuAssertIntEquals(tc, 0, raft_log_count(l));
CuAssertTrue(tc, NULL == raft_log_get_at_idx(l, 1));
}
void TestLog_delete_onwards(CuTest * tc)
......@@ -189,18 +189,18 @@ void TestLog_delete_onwards(CuTest * tc)
void *l;
l = log_new();
log_set_callbacks(l, &log_funcs, r);
l = raft_log_new();
raft_log_set_callbacks(l, &log_funcs, r);
__LOG_APPEND_ENTRIES_SEQ_ID(l, 3, 1, 0, NULL);
CuAssertIntEquals(tc, 3, log_count(l));
CuAssertIntEquals(tc, 3, raft_log_count(l));
/* even 3 gets deleted */
log_delete(l, 2);
CuAssertIntEquals(tc, 1, log_count(l));
CuAssertIntEquals(tc, 1, log_get_at_idx(l, 1)->id);
CuAssertTrue(tc, NULL == log_get_at_idx(l, 2));
CuAssertTrue(tc, NULL == log_get_at_idx(l, 3));
raft_log_delete(l, 2);
CuAssertIntEquals(tc, 1, raft_log_count(l));
CuAssertIntEquals(tc, 1, raft_log_get_at_idx(l, 1)->id);
CuAssertTrue(tc, NULL == raft_log_get_at_idx(l, 2));
CuAssertTrue(tc, NULL == raft_log_get_at_idx(l, 3));
}
void TestLog_delete_handles_log_pop_failure(CuTest * tc)
......@@ -216,17 +216,17 @@ void TestLog_delete_handles_log_pop_failure(CuTest * tc)
};
raft_set_callbacks(r, &funcs, queue);
l = log_new();
log_set_callbacks(l, &log_funcs, r);
l = raft_log_new();
raft_log_set_callbacks(l, &log_funcs, r);
__LOG_APPEND_ENTRIES_SEQ_ID(l, 3, 1, 0, NULL);
CuAssertIntEquals(tc, 3, log_count(l));
CuAssertIntEquals(tc, 3, log_get_current_idx(l));
CuAssertIntEquals(tc, 3, raft_log_count(l));
CuAssertIntEquals(tc, 3, raft_log_get_current_idx(l));
CuAssertIntEquals(tc, -1, log_delete(l, 3));
CuAssertIntEquals(tc, 3, log_count(l));
CuAssertIntEquals(tc, 3, log_count(l));
CuAssertIntEquals(tc, 3, ((raft_entry_t*)log_peektail(l))->id);
CuAssertIntEquals(tc, -1, raft_log_delete(l, 3));
CuAssertIntEquals(tc, 3, raft_log_count(l));
CuAssertIntEquals(tc, 3, raft_log_count(l));
CuAssertIntEquals(tc, 3, ((raft_entry_t*) raft_log_peektail(l))->id);
}
void TestLog_delete_fails_for_idx_zero(CuTest * tc)
......@@ -239,14 +239,14 @@ void TestLog_delete_fails_for_idx_zero(CuTest * tc)
.log_pop = __log_pop
};
raft_set_callbacks(r, &funcs, queue);
log_set_callbacks(raft_get_log(r), &log_funcs, r);
raft_log_set_callbacks(raft_get_log(r), &log_funcs, r);
void *l;
l = log_alloc(1);
log_set_callbacks(l, &log_funcs, r);
l = raft_log_alloc(1);
raft_log_set_callbacks(l, &log_funcs, r);
__LOG_APPEND_ENTRIES_SEQ_ID(l, 4, 1, 0, NULL);
CuAssertIntEquals(tc, log_delete(l, 0), -1);
CuAssertIntEquals(tc, raft_log_delete(l, 0), -1);
}
void TestLog_poll(CuTest * tc)
......@@ -262,65 +262,65 @@ void TestLog_poll(CuTest * tc)
void *l;
l = log_new();
log_set_callbacks(l, &log_funcs, r);
l = raft_log_new();
raft_log_set_callbacks(l, &log_funcs, r);
__LOG_APPEND_ENTRY(l, 1, 0, NULL);
CuAssertIntEquals(tc, 1, log_get_current_idx(l));
CuAssertIntEquals(tc, 1, raft_log_get_current_idx(l));
__LOG_APPEND_ENTRY(l, 2, 0, NULL);
CuAssertIntEquals(tc, 2, log_get_current_idx(l));
CuAssertIntEquals(tc, 2, raft_log_get_current_idx(l));
__LOG_APPEND_ENTRY(l, 3, 0, NULL);
CuAssertIntEquals(tc, 3, log_count(l));
CuAssertIntEquals(tc, 3, log_get_current_idx(l));
CuAssertIntEquals(tc, 3, raft_log_count(l));
CuAssertIntEquals(tc, 3, raft_log_get_current_idx(l));
raft_entry_t *ety;
/* remove 1st */
ety = NULL;
CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0);
CuAssertIntEquals(tc, raft_log_poll(l, (void *) &ety), 0);
CuAssertTrue(tc, NULL != ety);
CuAssertIntEquals(tc, 2, log_count(l));
CuAssertIntEquals(tc, 2, raft_log_count(l));
CuAssertIntEquals(tc, ety->id, 1);
CuAssertIntEquals(tc, 1, log_get_base(l));
CuAssertTrue(tc, NULL == log_get_at_idx(l, 1));
CuAssertTrue(tc, NULL != log_get_at_idx(l, 2));
CuAssertTrue(tc, NULL != log_get_at_idx(l, 3));
CuAssertIntEquals(tc, 3, log_get_current_idx(l));
CuAssertIntEquals(tc, 1, raft_log_get_base(l));
CuAssertTrue(tc, NULL == raft_log_get_at_idx(l, 1));
CuAssertTrue(tc, NULL != raft_log_get_at_idx(l, 2));
CuAssertTrue(tc, NULL != raft_log_get_at_idx(l, 3));
CuAssertIntEquals(tc, 3, raft_log_get_current_idx(l));
/* remove 2nd */
ety = NULL;
CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0);
CuAssertIntEquals(tc, raft_log_poll(l, (void *) &ety), 0);
CuAssertTrue(tc, NULL != ety);
CuAssertIntEquals(tc, 1, log_count(l));
CuAssertIntEquals(tc, 1, raft_log_count(l));
CuAssertIntEquals(tc, ety->id, 2);
CuAssertTrue(tc, NULL == log_get_at_idx(l, 1));
CuAssertTrue(tc, NULL == log_get_at_idx(l, 2));
CuAssertTrue(tc, NULL != log_get_at_idx(l, 3));
CuAssertIntEquals(tc, 3, log_get_current_idx(l));
CuAssertTrue(tc, NULL == raft_log_get_at_idx(l, 1));
CuAssertTrue(tc, NULL == raft_log_get_at_idx(l, 2));
CuAssertTrue(tc, NULL != raft_log_get_at_idx(l, 3));
CuAssertIntEquals(tc, 3, raft_log_get_current_idx(l));
/* remove 3rd */
ety = NULL;
CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0);
CuAssertIntEquals(tc, raft_log_poll(l, (void *) &ety), 0);
CuAssertTrue(tc, NULL != ety);
CuAssertIntEquals(tc, 0, log_count(l));
CuAssertIntEquals(tc, 0, raft_log_count(l));
CuAssertIntEquals(tc, ety->id, 3);
CuAssertTrue(tc, NULL == log_get_at_idx(l, 1));
CuAssertTrue(tc, NULL == log_get_at_idx(l, 2));
CuAssertTrue(tc, NULL == log_get_at_idx(l, 3));
CuAssertIntEquals(tc, 3, log_get_current_idx(l));
CuAssertTrue(tc, NULL == raft_log_get_at_idx(l, 1));
CuAssertTrue(tc, NULL == raft_log_get_at_idx(l, 2));
CuAssertTrue(tc, NULL == raft_log_get_at_idx(l, 3));
CuAssertIntEquals(tc, 3, raft_log_get_current_idx(l));
}
void TestLog_peektail(CuTest * tc)
{
void *l;
l = log_new();
l = raft_log_new();
__LOG_APPEND_ENTRIES_SEQ_ID(l, 3, 1, 0, NULL);
CuAssertIntEquals(tc, 3, log_count(l));
CuAssertIntEquals(tc, 3, log_peektail(l)->id);
CuAssertIntEquals(tc, 3, raft_log_count(l));
CuAssertIntEquals(tc, 3, raft_log_peektail(l)->id);
}
#if 0
......@@ -342,26 +342,26 @@ void TestLog_load_from_snapshot(CuTest * tc)
{
void *l;
l = log_new();
CuAssertIntEquals(tc, 0, log_get_current_idx(l));
CuAssertIntEquals(tc, 0, log_load_from_snapshot(l, 10, 5));
CuAssertIntEquals(tc, 10, log_get_current_idx(l));
CuAssertIntEquals(tc, 0, log_count(l));
l = raft_log_new();
CuAssertIntEquals(tc, 0, raft_log_get_current_idx(l));
CuAssertIntEquals(tc, 0, raft_log_load_from_snapshot(l, 10, 5));
CuAssertIntEquals(tc, 10, raft_log_get_current_idx(l));
CuAssertIntEquals(tc, 0, raft_log_count(l));
}
void TestLog_load_from_snapshot_clears_log(CuTest * tc)
{
void *l;
l = log_new();
l = raft_log_new();
__LOG_APPEND_ENTRIES_SEQ_ID(l, 2, 1, 0, NULL);
CuAssertIntEquals(tc, 2, log_count(l));
CuAssertIntEquals(tc, 2, log_get_current_idx(l));
CuAssertIntEquals(tc, 2, raft_log_count(l));
CuAssertIntEquals(tc, 2, raft_log_get_current_idx(l));
CuAssertIntEquals(tc, 0, log_load_from_snapshot(l, 10, 5));
CuAssertIntEquals(tc, 0, log_count(l));
CuAssertIntEquals(tc, 10, log_get_current_idx(l));
CuAssertIntEquals(tc, 0, raft_log_load_from_snapshot(l, 10, 5));
CuAssertIntEquals(tc, 0, raft_log_count(l));
CuAssertIntEquals(tc, 10, raft_log_get_current_idx(l));
}
void TestLog_front_pushes_across_boundary(CuTest * tc)
......@@ -370,16 +370,16 @@ void TestLog_front_pushes_across_boundary(CuTest * tc)
void *l;
l = log_alloc(1);
log_set_callbacks(l, &log_funcs, r);
l = raft_log_alloc(1);
raft_log_set_callbacks(l, &log_funcs, r);
raft_entry_t* ety;
__LOG_APPEND_ENTRY(l, 1, 0, NULL);
CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0);
CuAssertIntEquals(tc, raft_log_poll(l, (void *) &ety), 0);
CuAssertIntEquals(tc, ety->id, 1);
__LOG_APPEND_ENTRY(l, 2, 0, NULL);
CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0);
CuAssertIntEquals(tc, raft_log_poll(l, (void *) &ety), 0);
CuAssertIntEquals(tc, ety->id, 2);
}
......@@ -387,7 +387,7 @@ void TestLog_front_and_back_pushed_across_boundary_with_enlargement_required(CuT
{
void *l;
l = log_alloc(1);
l = raft_log_alloc(1);
raft_entry_t* ety;
......@@ -395,14 +395,14 @@ void TestLog_front_and_back_pushed_across_boundary_with_enlargement_required(CuT
__LOG_APPEND_ENTRY(l, 1, 0, NULL);
/* poll */
CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0);
CuAssertIntEquals(tc, raft_log_poll(l, (void *) &ety), 0);
CuAssertIntEquals(tc, ety->id, 1);
/* append */
__LOG_APPEND_ENTRY(l, 2, 0, NULL);
/* poll */
CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0);
CuAssertIntEquals(tc, raft_log_poll(l, (void *) &ety), 0);
CuAssertIntEquals(tc, ety->id, 2);
/* append append */
......@@ -410,7 +410,7 @@ void TestLog_front_and_back_pushed_across_boundary_with_enlargement_required(CuT
__LOG_APPEND_ENTRY(l, 4, 0, NULL);
/* poll */
CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0);
CuAssertIntEquals(tc, raft_log_poll(l, (void *) &ety), 0);
CuAssertIntEquals(tc, ety->id, 3);
}
......@@ -418,26 +418,26 @@ void TestLog_delete_after_polling(CuTest * tc)
{
void *l;
l = log_alloc(1);
l = raft_log_alloc(1);
raft_entry_t* ety;
/* append */
__LOG_APPEND_ENTRY(l, 1, 0, NULL);
CuAssertIntEquals(tc, 1, log_count(l));
CuAssertIntEquals(tc, 1, raft_log_count(l));
/* poll */
CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0);
CuAssertIntEquals(tc, raft_log_poll(l, (void *) &ety), 0);
CuAssertIntEquals(tc, ety->id, 1);
CuAssertIntEquals(tc, 0, log_count(l));
CuAssertIntEquals(tc, 0, raft_log_count(l));
/* append */
__LOG_APPEND_ENTRY(l, 2, 0, NULL);
CuAssertIntEquals(tc, 1, log_count(l));
CuAssertIntEquals(tc, 1, raft_log_count(l));
/* poll */
CuAssertIntEquals(tc, log_delete(l, 1), 0);
CuAssertIntEquals(tc, 0, log_count(l));
CuAssertIntEquals(tc, raft_log_delete(l, 1), 0);
CuAssertIntEquals(tc, 0, raft_log_count(l));
}
void TestLog_delete_after_polling_from_double_append(CuTest * tc)
......@@ -453,27 +453,27 @@ void TestLog_delete_after_polling_from_double_append(CuTest * tc)
void *l;
l = log_alloc(1);
log_set_callbacks(l, &log_funcs, r);
l = raft_log_alloc(1);
raft_log_set_callbacks(l, &log_funcs, r);
raft_entry_t* ety;
/* append append */
__LOG_APPEND_ENTRIES_SEQ_ID(l, 2, 1, 0, NULL);
CuAssertIntEquals(tc, 2, log_count(l));
CuAssertIntEquals(tc, 2, raft_log_count(l));
/* poll */
CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0);
CuAssertIntEquals(tc, raft_log_poll(l, (void *) &ety), 0);
CuAssertIntEquals(tc, ety->id, 1);
CuAssertIntEquals(tc, 1, log_count(l));
CuAssertIntEquals(tc, 1, raft_log_count(l));
/* append */
__LOG_APPEND_ENTRY(l, 3, 0, NULL);
CuAssertIntEquals(tc, 2, log_count(l));
CuAssertIntEquals(tc, 2, raft_log_count(l));
/* poll */
CuAssertIntEquals(tc, log_delete(l, 1), 0);
CuAssertIntEquals(tc, 0, log_count(l));
CuAssertIntEquals(tc, raft_log_delete(l, 1), 0);
CuAssertIntEquals(tc, 0, raft_log_count(l));
}
void TestLog_get_from_idx_with_base_off_by_one(CuTest * tc)
......@@ -489,28 +489,28 @@ void TestLog_get_from_idx_with_base_off_by_one(CuTest * tc)
void *l;
l = log_alloc(1);
log_set_callbacks(l, &log_funcs, r);
l = raft_log_alloc(1);
raft_log_set_callbacks(l, &log_funcs, r);
raft_entry_t* ety;
/* append append */
__LOG_APPEND_ENTRIES_SEQ_ID(l, 2, 1, 0, NULL);
CuAssertIntEquals(tc, 2, log_count(l));
CuAssertIntEquals(tc, 2, raft_log_count(l));
/* poll */
CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0);
CuAssertIntEquals(tc, raft_log_poll(l, (void *) &ety), 0);
CuAssertIntEquals(tc, ety->id, 1);
CuAssertIntEquals(tc, 1, log_count(l));
CuAssertIntEquals(tc, 1, raft_log_count(l));
/* get off-by-one index */
long n_etys;
CuAssertPtrEquals(tc, log_get_from_idx(l, 1, &n_etys), NULL);
CuAssertPtrEquals(tc, raft_log_get_from_idx(l, 1, &n_etys), NULL);
CuAssertIntEquals(tc, n_etys, 0);
/* now get the correct index */
raft_entry_t** e;
e = log_get_from_idx(l, 2, &n_etys);
e = raft_log_get_from_idx(l, 2, &n_etys);
CuAssertPtrNotNull(tc, e);
CuAssertIntEquals(tc, n_etys, 1);
CuAssertIntEquals(tc, e[0]->id, 2);
......
......@@ -1585,7 +1585,7 @@ void TestRaft_follower_recv_appendentries_partial_failures(
void *r = raft_new();
__raft_error_t error = {};
raft_set_callbacks(r, &funcs, &error);
log_set_callbacks(raft_get_log(r), &log_funcs, r);
raft_log_set_callbacks(raft_get_log(r), &log_funcs, r);
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
......@@ -3761,7 +3761,7 @@ void TestRaft_leader_recv_entry_add_nonvoting_node_remove_and_revert(CuTest *tc)
int has_sufficient_logs_flag = 0;
raft_set_callbacks(r, &funcs, &has_sufficient_logs_flag);
log_set_callbacks(raft_get_log(r), &log_funcs, r);
raft_log_set_callbacks(raft_get_log(r), &log_funcs, r);
/* I'm the leader */
raft_set_state(r, RAFT_STATE_LEADER);
......@@ -3807,7 +3807,7 @@ void TestRaft_leader_recv_appendentries_response_set_has_sufficient_logs_after_v
int has_sufficient_logs_flag = 0;
raft_set_callbacks(r, &funcs, &has_sufficient_logs_flag);
log_set_callbacks(raft_get_log(r), &log_funcs, r);
raft_log_set_callbacks(raft_get_log(r), &log_funcs, r);
/* I'm the leader */
raft_set_state(r, RAFT_STATE_LEADER);
......@@ -4452,7 +4452,6 @@ void Test_reset_transfer_leader(CuTest *tc)
};
raft_server_t *r = raft_new();
raft_set_state(r, RAFT_STATE_LEADER);
raft_server_private_t * me = (raft_server_private_t *) r;
raft_set_callbacks(r, &funcs, &state);
......@@ -4465,12 +4464,12 @@ void Test_reset_transfer_leader(CuTest *tc)
ret = raft_transfer_leader(r, 2, 0);
CuAssertIntEquals(tc, 0, ret);
me->leader_id = 2;
r->leader_id = 2;
raft_reset_transfer_leader(r, 0);
CuAssertIntEquals(tc, RAFT_LEADER_TRANSFER_EXPECTED_LEADER, state);
/* tests timeout in general, so don't need a separate test for it */
me->leader_id = 1;
r->leader_id = 1;
ret = raft_transfer_leader(r, 2, 1);
CuAssertIntEquals(tc, 0, ret);
raft_periodic(r, 2);
......
......@@ -319,7 +319,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc)
.send_appendentries = __raft_send_appendentries,
};
void *r = raft_new();
raft_server_t *r = raft_new();
raft_set_callbacks(r, &funcs, NULL);
msg_entry_response_t cr;
......@@ -366,8 +366,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc)
CuAssertIntEquals(tc, 0, raft_periodic(r, 1000));
CuAssertIntEquals(tc, 2, raft_get_log_count(r));
CuAssertIntEquals(tc, 4, raft_get_commit_idx(r));
raft_server_private_t *r_p = (raft_server_private_t *) r;
CuAssertIntEquals(tc, 3, r_p->log_impl->first_idx(r_p->log));
CuAssertIntEquals(tc, 3, r->log_impl->first_idx(r->log));
CuAssertIntEquals(tc, 2, raft_get_num_snapshottable_logs(r));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
CuAssertIntEquals(tc, 0, raft_end_snapshot(r));
......
......@@ -847,7 +847,7 @@ class RaftServer(object):
log_cbs.log_pop = self.raft_logentry_pop
lib.raft_set_callbacks(self.raft, cbs, self.udata)
lib.log_set_callbacks(lib.raft_get_log(self.raft), log_cbs, self.raft)
lib.raft_log_set_callbacks(lib.raft_get_log(self.raft), log_cbs, self.raft)
lib.raft_set_election_timeout(self.raft, 500)
lib.raft_set_auto_flush(self.raft, network.auto_flush)
......
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