Skip to content
GitLab
Menu
Projects
Groups
Snippets
Loading...
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
Menu
Open sidebar
ruanhaishen
RedisLabs Raft
Commits
8ee3f264
Unverified
Commit
8ee3f264
authored
Mar 30, 2022
by
Ozan Tezcan
Committed by
GitHub
Mar 30, 2022
Browse files
Add raft_ prefix for exported symbols (#96)
parent
80849148
Changes
9
Expand all
Hide whitespace changes
Inline
Side-by-side
include/raft.h
View file @
8ee3f264
...
...
@@ -119,7 +119,21 @@ typedef struct raft_entry
/** Message sent from client to server.
* The client sends this message to a server with the intention of having it
* applied to the FSM. */
typedef
raft_entry_t
msg_entry_t
;
typedef
raft_entry_t
raft_entry_req_t
;
/** Entry message response.
* Indicates to client if entry was committed or not. */
typedef
struct
{
/** the entry's unique ID */
raft_entry_id_t
id
;
/** the entry's term */
raft_term_t
term
;
/** the entry's index */
raft_index_t
idx
;
}
raft_entry_resp_t
;
typedef
struct
{
...
...
@@ -136,20 +150,6 @@ typedef struct
int
last_chunk
;
}
raft_snapshot_chunk_t
;
/** Entry message response.
* Indicates to client if entry was committed or not. */
typedef
struct
{
/** the entry's unique ID */
raft_entry_id_t
id
;
/** the entry's term */
raft_term_t
term
;
/** the entry's index */
raft_index_t
idx
;
}
msg_entry_response_t
;
/** Vote request message.
* Sent to nodes when a server wants to become leader.
* This message could force a leader/candidate to become a follower. */
...
...
@@ -169,7 +169,7 @@ typedef struct
/** term of candidate's last log entry */
raft_term_t
last_log_term
;
}
msg
_requestvote_t
;
}
raft
_requestvote_
req_
t
;
/** Vote request response message.
* Indicates if node has accepted the server's vote request. */
...
...
@@ -186,7 +186,7 @@ typedef struct
/** true means candidate received vote */
int
vote_granted
;
}
msg
_requestvote_resp
onse
_t
;
}
raft
_requestvote_resp_t
;
typedef
struct
{
...
...
@@ -209,7 +209,7 @@ typedef struct
/** snapshot chunk **/
raft_snapshot_chunk_t
chunk
;
}
msg
_snapshot_t
;
}
raft
_snapshot_
req_
t
;
typedef
struct
{
...
...
@@ -227,7 +227,7 @@ typedef struct
/** 1 if this is a response to the final chunk */
int
last_chunk
;
}
msg
_snapshot_resp
onse
_t
;
}
raft
_snapshot_resp_t
;
/** Appendentries message.
* This message is used to tell nodes if it's safe to apply entries to the FSM.
...
...
@@ -261,8 +261,8 @@ typedef struct
int
n_entries
;
/** array of pointers to entries within this message */
msg
_entry_t
**
entries
;
}
msg
_appendentries_t
;
raft
_entry_
req_
t
**
entries
;
}
raft
_appendentries_
req_
t
;
/** Appendentries response message.
* Can be sent without any entries as a keep alive message.
...
...
@@ -285,7 +285,7 @@ typedef struct
/** If success, this is the highest log IDX we've received and appended to
* our log; otherwise, this is the our currentIndex */
raft_index_t
current_idx
;
}
msg
_appendentries_resp
onse
_t
;
}
raft
_appendentries_resp_t
;
typedef
struct
raft_server
raft_server_t
;
typedef
struct
raft_node
raft_node_t
;
...
...
@@ -297,12 +297,12 @@ typedef struct raft_node raft_node_t;
* @param[in] msg The request vote message to be sent
* @return 0 on success */
typedef
int
(
*
func
_send_requestvote_f
*
raft
_send_requestvote_f
)
(
raft_server_t
*
raft
,
void
*
user_data
,
raft_node_t
*
node
,
msg
_requestvote_t
*
msg
raft
_requestvote_
req_
t
*
msg
);
/** Callback for sending append entries messages.
...
...
@@ -312,12 +312,12 @@ typedef int (
* @param[in] msg The appendentries message to be sent
* @return 0 on success */
typedef
int
(
*
func
_send_appendentries_f
*
raft
_send_appendentries_f
)
(
raft_server_t
*
raft
,
void
*
user_data
,
raft_node_t
*
node
,
msg
_appendentries_t
*
msg
raft
_appendentries_
req_
t
*
msg
);
/** Callback for sending snapshot messages.
...
...
@@ -327,12 +327,12 @@ typedef int (
* @param[in] msg Snapshot msg
**/
typedef
int
(
*
func
_send_snapshot_f
*
raft
_send_snapshot_f
)
(
raft_server_t
*
raft
,
void
*
user_data
,
raft_node_t
*
node
,
msg
_snapshot_t
*
msg
raft
_snapshot_
req_
t
*
msg
);
/** Callback for loading the received snapshot. User should load snapshot using
...
...
@@ -366,7 +366,7 @@ typedef int (
* @param[in] snapshot_term Received snapshot term
* @return 0 on success */
typedef
int
(
*
func
_load_snapshot_f
*
raft
_load_snapshot_f
)
(
raft_server_t
*
raft
,
void
*
user_data
,
...
...
@@ -386,7 +386,7 @@ typedef int (
* @param[in] chunk Snapshot chunk
* @return 0 on success */
typedef
int
(
*
func
_get_snapshot_chunk_f
*
raft
_get_snapshot_chunk_f
)
(
raft_server_t
*
raft
,
void
*
user_data
,
...
...
@@ -403,7 +403,7 @@ typedef int (
* @param[in] chunk Snapshot chunk
* @return 0 on success */
typedef
int
(
*
func
_store_snapshot_chunk_f
*
raft
_store_snapshot_chunk_f
)
(
raft_server_t
*
raft
,
void
*
user_data
,
...
...
@@ -420,7 +420,7 @@ typedef int (
* @param[in] user_data User data that is passed from Raft server
* @return 0 on success */
typedef
int
(
*
func
_clear_snapshot_f
*
raft
_clear_snapshot_f
)
(
raft_server_t
*
raft
,
void
*
user_data
...
...
@@ -433,7 +433,7 @@ typedef int (
* @param[in] node The node
* @return 0 does not want to be notified again; otherwise -1 */
typedef
int
(
*
func
_node_has_sufficient_logs_f
*
raft
_node_has_sufficient_logs_f
)
(
raft_server_t
*
raft
,
void
*
user_data
,
...
...
@@ -450,7 +450,7 @@ typedef int (
* @param[in] user_data User data that is passed from Raft server
* @param[in] buf The buffer that was logged */
typedef
void
(
*
func
_log_f
*
raft
_log_f
)
(
raft_server_t
*
raft
,
raft_node_id_t
node_id
,
...
...
@@ -466,7 +466,7 @@ typedef void (
* @param[in] vote The node we voted for
* @return 0 on success */
typedef
int
(
*
func
_persist_vote_f
*
raft
_persist_vote_f
)
(
raft_server_t
*
raft
,
void
*
user_data
,
...
...
@@ -482,7 +482,7 @@ typedef int (
* @param[in] vote The node value dictating we haven't voted for anybody
* @return 0 on success */
typedef
int
(
*
func
_persist_term_f
*
raft
_persist_term_f
)
(
raft_server_t
*
raft
,
void
*
user_data
,
...
...
@@ -509,7 +509,7 @@ typedef int (
* @return 0 on success
* */
typedef
int
(
*
func
_logentry_event_f
*
raft
_logentry_event_f
)
(
raft_server_t
*
raft
,
void
*
user_data
,
...
...
@@ -528,7 +528,7 @@ typedef int (
* @return the node ID of the node
* */
typedef
raft_node_id_t
(
*
func
_get_node_id_f
*
raft
_get_node_id_f
)
(
raft_server_t
*
raft
,
void
*
user_data
,
...
...
@@ -548,7 +548,7 @@ typedef raft_node_id_t (
* @param[in] entry The entry that was the trigger for the event. Could be NULL.
* @param[in] type The type of membership change */
typedef
void
(
*
func
_membership_event_f
*
raft
_membership_event_f
)
(
raft_server_t
*
raft
,
void
*
user_data
,
...
...
@@ -565,7 +565,7 @@ typedef void (
* @param[in] user_data User data that is passed from Raft server
* @param[in] state The new cluster state. */
typedef
void
(
*
func
_state_event_f
*
raft
_state_event_f
)
(
raft_server_t
*
raft
,
void
*
user_data
,
...
...
@@ -581,7 +581,7 @@ typedef void (
* @param[in] result the leadership transfer result
*/
typedef
void
(
*
func
_transfer_event_f
*
raft
_transfer_event_f
)
(
raft_server_t
*
raft
,
void
*
user_data
,
...
...
@@ -594,19 +594,19 @@ typedef void (
* @return 0 on success
*/
typedef
int
(
*
func
_send_timeoutnow_f
*
raft
_send_timeoutnow_f
)
(
raft_server_t
*
raft
,
raft_node_t
*
node
);
/** Callback to skip sending
msg
_appendentries to the node
/** Callback to skip sending
raft
_appendentries
_req
to the node
*
* Implementing this callback is optional
*
* If there are already pending appendentries messages in flight, you may want
* to skip sending more until you receive response for the previous ones.
* If the node is a slow consumer and you create
msg
_appendentries for each
* If the node is a slow consumer and you create
raft
_appendentries
_req
for each
* batch of new entries received, it may cause out of memory.
*
* Also, this way you can do batching. If new entries are received with an
...
...
@@ -620,12 +620,12 @@ typedef int (
* performance.
*
* @param[in] raft The Raft server making this callback
* @param[in] node The node that we are about to send
msg
_appendentries to
* @param[in] node The node that we are about to send
raft
_appendentries
_req
to
* @return 0 to send message
* Any other value to skip sending message
*/
typedef
int
(
*
func
_backpressure_f
*
raft
_backpressure_f
)
(
raft_server_t
*
raft
,
void
*
user_data
,
...
...
@@ -635,68 +635,68 @@ typedef int (
typedef
struct
{
/** Callback for sending request vote messages */
func
_send_requestvote_f
send_requestvote
;
raft
_send_requestvote_f
send_requestvote
;
/** Callback for sending appendentries messages */
func
_send_appendentries_f
send_appendentries
;
raft
_send_appendentries_f
send_appendentries
;
/** Callback for sending snapshot messages */
func
_send_snapshot_f
send_snapshot
;
raft
_send_snapshot_f
send_snapshot
;
/** Callback for loading snapshot. This will be called when we complete
* receiving snapshot from the leader */
func
_load_snapshot_f
load_snapshot
;
raft
_load_snapshot_f
load_snapshot
;
/** Callback to get a chunk of the snapshot file */
func
_get_snapshot_chunk_f
get_snapshot_chunk
;
raft
_get_snapshot_chunk_f
get_snapshot_chunk
;
/** Callback to store a chunk of the snapshot */
func
_store_snapshot_chunk_f
store_snapshot_chunk
;
raft
_store_snapshot_chunk_f
store_snapshot_chunk
;
/** Callback to dismiss temporary file which is used for incoming
* snapshot chunks */
func
_clear_snapshot_f
clear_snapshot
;
raft
_clear_snapshot_f
clear_snapshot
;
/** Callback for finite state machine application
* Return 0 on success.
* Return RAFT_ERR_SHUTDOWN if you want the server to shutdown. */
func
_logentry_event_f
applylog
;
raft
_logentry_event_f
applylog
;
/** Callback for persisting vote data
* For safety reasons this callback MUST flush the change to disk. */
func
_persist_vote_f
persist_vote
;
raft
_persist_vote_f
persist_vote
;
/** Callback for persisting term (and nil vote) data
* For safety reasons this callback MUST flush the term and vote changes to
* disk atomically. */
func
_persist_term_f
persist_term
;
raft
_persist_term_f
persist_term
;
/** Callback for determining which node this configuration log entry
* affects. This call only applies to configuration change log entries.
* @return the node ID of the node */
func
_get_node_id_f
get_node_id
;
raft
_get_node_id_f
get_node_id
;
/** Callback for detecting when a non-voting node has sufficient logs. */
func
_node_has_sufficient_logs_f
node_has_sufficient_logs
;
raft
_node_has_sufficient_logs_f
node_has_sufficient_logs
;
/** Callback for being notified of membership changes (optional). */
func
_membership_event_f
notify_membership_event
;
raft
_membership_event_f
notify_membership_event
;
/** Callback for being notified of state changes (optional). */
func
_state_event_f
notify_state_event
;
raft
_state_event_f
notify_state_event
;
/** Callbakc for notified of transfer leadership events (optional) */
func
_transfer_event_f
notify_transfer_event
;
raft
_transfer_event_f
notify_transfer_event
;
/** Callback for catching debugging log messages
* This callback is optional */
func
_log_f
log
;
raft
_log_f
log
;
/** Callback for sending TimeoutNow RPC messages to nodes */
func
_send_timeoutnow_f
send_timeoutnow
;
raft
_send_timeoutnow_f
send_timeoutnow
;
/** Callback for deciding whether to send
msg
_appendentries to a node. */
func
_backpressure_f
backpressure
;
/** Callback for deciding whether to send
raft
_appendentries
_req
to a node. */
raft
_backpressure_f
backpressure
;
}
raft_cbs_t
;
/** A generic notification callback used to allow Raft to notify caller
...
...
@@ -710,7 +710,7 @@ typedef struct
* log operation until it returns.
*/
typedef
void
(
*
func
_entry_notify_f
*
raft
_entry_notify_f
)
(
void
*
arg
,
raft_entry_t
*
entry
,
...
...
@@ -725,7 +725,7 @@ typedef void (
* arriving to a non leader.
*/
typedef
void
(
*
func
_read_request_callback_f
*
raft
_read_request_callback_f
)
(
void
*
arg
,
int
can_read
...
...
@@ -827,7 +827,7 @@ typedef struct raft_log_impl
* 0 on success;
* -1 on error.
*/
int
(
*
pop
)
(
void
*
log
,
raft_index_t
from_idx
,
func
_entry_notify_f
cb
,
void
*
cb_arg
);
int
(
*
pop
)
(
void
*
log
,
raft_index_t
from_idx
,
raft
_entry_notify_f
cb
,
void
*
cb_arg
);
/** Get a single entry from the log.
*
...
...
@@ -977,26 +977,26 @@ int raft_periodic(raft_server_t* me, int msec_elapsed);
* included entries.
*
* @param[in] node The node who sent us this message
* @param[in]
ae
The appendentries message
* @param[out] r The resulting response
* @param[in]
req
The appendentries message
* @param[out] r
esp
The resulting response
* @return
* 0 on success
* */
int
raft_recv_appendentries
(
raft_server_t
*
me
,
raft_node_t
*
node
,
msg
_appendentries_
t
*
ae
,
msg
_appendentries_resp
onse_t
*
r
);
raft
_appendentries_
req_t
*
req
,
raft
_appendentries_resp
_t
*
resp
);
/** Receive a response from an appendentries message we sent.
* @param[in] node The node who sent us this message
* @param[in] r The appendentries response message
* @param[in] r
esp
The appendentries response message
* @return
* 0 on success;
* -1 on error;
* RAFT_ERR_NOT_LEADER server is not the leader */
int
raft_recv_appendentries_response
(
raft_server_t
*
me
,
raft_node_t
*
node
,
msg
_appendentries_resp
onse
_t
*
r
);
raft
_appendentries_resp_t
*
r
esp
);
/** Receive a snapshot message.
* @param[in] node The node who sent us this message
...
...
@@ -1006,39 +1006,39 @@ int raft_recv_appendentries_response(raft_server_t* me,
* 0 on success */
int
raft_recv_snapshot
(
raft_server_t
*
me
,
raft_node_t
*
node
,
msg
_snapshot_t
*
req
,
msg
_snapshot_resp
onse
_t
*
resp
);
raft
_snapshot_
req_
t
*
req
,
raft
_snapshot_resp_t
*
resp
);
/** Receive a response from a snapshot message we sent.
* @param[in] node The node who sent us this message
* @param[in] r The snapshot response message
* @param[in] r
esp
The snapshot response message
* @return
* 0 on success;
* -1 on error;
* RAFT_ERR_NOT_LEADER server is not the leader */
int
raft_recv_snapshot_response
(
raft_server_t
*
me
,
raft_node_t
*
node
,
msg
_snapshot_resp
onse
_t
*
r
);
raft
_snapshot_resp_t
*
r
esp
);
/** Receive a requestvote message.
* @param[in] node The node who sent us this message
* @param[in]
v
r The requestvote message
* @param[out] r The resulting response
* @param[in] r
eq
The requestvote message
* @param[out] r
esp
The resulting response
* @return 0 on success */
int
raft_recv_requestvote
(
raft_server_t
*
me
,
raft_node_t
*
node
,
msg
_requestvote_
t
*
vr
,
msg
_requestvote_resp
onse_t
*
r
);
raft
_requestvote_
req_t
*
req
,
raft
_requestvote_resp
_t
*
resp
);
/** Receive a response from a requestvote message we sent.
* @param[in] node The node this response was sent by
* @param[in] r The requestvote response message
* @param[in] r
eq
The requestvote response message
* @return
* 0 on success;
* RAFT_ERR_SHUTDOWN server MUST shutdown; */
int
raft_recv_requestvote_response
(
raft_server_t
*
me
,
raft_node_t
*
node
,
msg
_requestvote_resp
onse
_t
*
r
);
raft
_requestvote_resp_t
*
r
eq
);
/** Receive an entry message from the client.
*
...
...
@@ -1055,8 +1055,8 @@ int raft_recv_requestvote_response(raft_server_t* me,
* </ul>
*
* @param[in] node The node who sent us this message
* @param[in]
ety
The entry message
* @param[out] r The resulting response
* @param[in]
req
The entry message
* @param[out] r
esp
The resulting response
* @return
* 0 on success;
* RAFT_ERR_NOT_LEADER server is not the leader;
...
...
@@ -1065,8 +1065,8 @@ int raft_recv_requestvote_response(raft_server_t* me,
* RAFT_ERR_NOMEM memory allocation failure
*/
int
raft_recv_entry
(
raft_server_t
*
me
,
msg
_entry_
t
*
ety
,
msg
_entry_resp
onse
_t
*
r
);
raft
_entry_
req_t
*
req
,
raft
_entry_resp_t
*
r
esp
);
/**
* @return server's node ID; -1 if it doesn't know what it is */
...
...
@@ -1239,7 +1239,7 @@ 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
,
const
msg
_entry_resp
onse
_t
*
r
);
const
raft
_entry_resp_t
*
r
);
/** Get node's ID.
* @return ID of node */
...
...
@@ -1445,7 +1445,7 @@ typedef struct {
* Return 0 on success.
* Return RAFT_ERR_SHUTDOWN if you want the server to shutdown. */
func
_logentry_event_f
log_offer
;
raft
_logentry_event_f
log_offer
;
/** Callback for removing the oldest entry from the log
* For safety reasons this callback MUST flush the change to disk.
...
...
@@ -1454,7 +1454,7 @@ typedef struct {
* desired after the callback returns, raft_entry_hold() should be
* used.
*/
func
_logentry_event_f
log_poll
;
raft
_logentry_event_f
log_poll
;
/** Callback for removing the youngest entry from the log
* For safety reasons this callback MUST flush the change to disk.
...
...
@@ -1463,13 +1463,13 @@ typedef struct {
* desired after the callback returns, raft_entry_hold() should be
* used.
*/
func
_logentry_event_f
log_pop
;
raft
_logentry_event_f
log_pop
;
/** Callback called for every existing log entry when clearing the log.
* If memory was malloc'd in log_offer and the entry doesn't get a chance
* to go through log_poll or log_pop, this is the last chance to free it.
*/
func
_logentry_event_f
log_clear
;
raft
_logentry_event_f
log_clear
;
}
raft_log_cbs_t
;
/** Allocate a new Raft Log entry.
...
...
@@ -1511,7 +1511,7 @@ 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
);
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
,
raft
_read_request_callback_f
cb
,
void
*
cb_arg
);
/** Attempt to process read queue.
*/
...
...
include/raft_private.h
View file @
8ee3f264
...
...
@@ -19,7 +19,7 @@ typedef struct raft_read_request {
raft_term_t
read_term
;
raft_msg_id_t
msg_id
;
func
_read_request_callback_f
cb
;
raft
_read_request_callback_f
cb
;
void
*
cb_arg
;
struct
raft_read_request
*
next
;
...
...
src/raft_log.c
View file @
8ee3f264
...
...
@@ -208,7 +208,7 @@ 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
,
func
_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
)
return
-
1
;
...
...
@@ -376,7 +376,7 @@ static int log_get_batch(void *log, raft_index_t idx, int entries_n, raft_entry_
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
,
raft
_entry_notify_f
cb
,
void
*
cb_arg
)
{
return
log_delete
(
log
,
from_idx
,
cb
,
cb_arg
);
}
...
...
src/raft_server.c
View file @
8ee3f264
...
...
@@ -349,7 +349,7 @@ int raft_delete_entry_from_idx(raft_server_t* me, raft_index_t idx)
me
->
voting_cfg_change_log_idx
=
-
1
;
int
e
=
me
->
log_impl
->
pop
(
me
->
log
,
idx
,
(
func
_entry_notify_f
)
raft_handle_remove_cfg_change
,
me
);
(
raft
_entry_notify_f
)
raft_handle_remove_cfg_change
,
me
);
if
(
e
!=
0
)
return
e
;
...
...
@@ -643,7 +643,7 @@ int raft_voting_change_is_in_progress(raft_server_t* me)
int
raft_recv_appendentries_response
(
raft_server_t
*
me
,
raft_node_t
*
node
,
msg
_appendentries_resp
onse
_t
*
r
)
raft
_appendentries_resp_t
*
r
)
{
raft_log_node
(
me
,
raft_node_get_id
(
node
),
"received appendentries response %s ci:%ld rci:%ld msgid:%lu"
,
...
...
@@ -755,12 +755,10 @@ static int raft_receive_term(raft_server_t* me, raft_term_t term)
return
0
;
}
int
raft_recv_appendentries
(
raft_server_t
*
me
,
raft_node_t
*
node
,
msg_appendentries_t
*
ae
,
msg_appendentries_response_t
*
r
)
int
raft_recv_appendentries
(
raft_server_t
*
me
,
raft_node_t
*
node
,
raft_appendentries_req_t
*
ae
,
raft_appendentries_resp_t
*
r
)
{
int
e
=
0
;
...
...
@@ -926,8 +924,8 @@ out:
int
raft_recv_requestvote
(
raft_server_t
*
me
,
raft_node_t
*
node
,
msg
_requestvote_t
*
vr
,
msg
_requestvote_resp
onse_t
*
r
)
raft
_requestvote_
req_
t
*
vr
,
raft
_requestvote_resp
_t
*
r
)
{
(
void
)
node
;
int
e
=
0
;
...
...
@@ -1011,7 +1009,7 @@ int raft_votes_is_majority(const int num_nodes, const int nvotes)
int
raft_recv_requestvote_response
(
raft_server_t
*
me
,
raft_node_t
*
node
,
msg
_requestvote_resp
onse
_t
*
r
)
raft
_requestvote_resp_t
*
r
)
{
raft_log_node
(
me
,
raft_node_get_id
(
node
),
"node responded to requestvote status:%s pv:%d rt:%ld ct:%ld rt:%ld"
,
...
...
@@ -1062,8 +1060,8 @@ int raft_recv_requestvote_response(raft_server_t* me,
}
int
raft_recv_entry
(
raft_server_t
*
me
,
msg
_entry_t
*
ety
,
msg
_entry_resp
onse_t
*
r
)
raft
_entry_
req_
t
*
ety
,
raft
_entry_resp
_t
*
r
)
{
if
(
raft_entry_is_voting_cfg_change
(
ety
))
{
...
...
@@ -1108,7 +1106,7 @@ int raft_recv_entry(raft_server_t* me,
int
raft_send_requestvote
(
raft_server_t
*
me
,
raft_node_t
*
node
)
{
msg
_requestvote_t
rv
;
raft
_requestvote_
req_
t
rv
;
int
e
=
0
;
assert
(
node
);
...
...
@@ -1254,7 +1252,7 @@ int raft_send_snapshot(raft_server_t* me, raft_node_t* node)
while
(
1
)
{
raft_size_t
offset
=
raft_node_get_snapshot_offset
(
node
);
msg
_snapshot_t
msg
=
{
raft
_snapshot_
req_
t
msg
=
{
.
leader_id
=
raft_get_nodeid
(
me
),
.
snapshot_index
=
me
->
snapshot_last_idx
,
.
snapshot_term
=
me
->
snapshot_last_term
,
...
...
@@ -1287,8 +1285,8 @@ int raft_send_snapshot(raft_server_t* me, raft_node_t* node)
int
raft_recv_snapshot
(
raft_server_t
*
me
,
raft_node_t
*
node
,
msg
_snapshot_t
*
req
,
msg
_snapshot_resp
onse
_t
*
resp
)
raft
_snapshot_
req_
t
*
req
,
raft
_snapshot_resp_t
*
resp
)
{
int
e
=
0
;
...
...
@@ -1379,7 +1377,7 @@ out:
int
raft_recv_snapshot_response
(
raft_server_t
*
me
,
raft_node_t
*
node
,
msg
_snapshot_resp
onse
_t
*
r
)
raft
_snapshot_resp_t
*
r
)
{
raft_log_node
(
me
,
raft_node_get_id
(
node
),
"recv snapshot response: ci:%lu comi:%lu mi:%lu t:%ld o:%llu s:%d lc:%d"
,
...
...
@@ -1456,7 +1454,7 @@ int raft_send_appendentries(raft_server_t* me, raft_node_t* node)
}
}
msg
_appendentries_t
ae
=
{
raft
_appendentries_
req_
t
ae
=
{
.
term
=
me
->
current_term
,
.
leader_id
=
raft_get_nodeid
(
me
),
.
leader_commit
=
raft_get_commit_idx
(
me
),
...
...
@@ -1561,7 +1559,7 @@ int raft_vote_for_nodeid(raft_server_t* me, const raft_node_id_t nodeid)
}
int
raft_msg_entry_response_committed
(
raft_server_t
*
me
,
const
msg
_entry_resp
onse
_t
*
r
)
const
raft
_entry_resp_t
*
r
)
{
raft_entry_t
*
ety
=
raft_get_entry_from_idx
(
me
,
r
->
idx
);
if
(
!
ety
)
...
...
@@ -1609,7 +1607,7 @@ int raft_pop_entry(raft_server_t* me)
raft_index_t
cur_idx
=
me
->
log_impl
->
current_idx
(
me
->
log
);
int
e
=
me
->
log_impl
->
pop
(
me
->
log
,
cur_idx
,
(
func
_entry_notify_f
)
raft_handle_remove_cfg_change
,
me
);
(
raft
_entry_notify_f
)
raft_handle_remove_cfg_change
,
me
);
if
(
e
!=
0
)
return
e
;
...
...
@@ -1850,7 +1848,7 @@ void raft_entry_release_list(raft_entry_t **ety_list, size_t len)
}
}
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
,
raft
_read_request_callback_f
cb
,
void
*
cb_arg
)
{
raft_read_request_t
*
req
=
raft_malloc
(
sizeof
(
raft_read_request_t
));
...
...
tests/mock_send_functions.c
View file @
8ee3f264
...
...
@@ -74,33 +74,35 @@ static int __append_msg(
}
int
sender_requestvote
(
raft_server_t
*
raft
,
void
*
udata
,
raft_node_t
*
node
,
msg_requestvote_t
*
msg
)
void
*
udata
,
raft_node_t
*
node
,
raft_requestvote_req_t
*
msg
)
{
return
__append_msg
(
udata
,
msg
,
RAFT_
MSG_
REQUESTVOTE
,
sizeof
(
*
msg
),
node
,
return
__append_msg
(
udata
,
msg
,
RAFT_REQUESTVOTE
_REQ
,
sizeof
(
*
msg
),
node
,
raft
);
}
int
sender_requestvote_response
(
raft_server_t
*
raft
,
void
*
udata
,
raft_node_t
*
node
,
msg
_requestvote_resp
onse
_t
*
msg
)
raft
_requestvote_resp_t
*
msg
)
{
return
__append_msg
(
udata
,
msg
,
RAFT_
MSG_
REQUESTVOTE_RESP
ONSE
,
sizeof
(
*
msg
),
return
__append_msg
(
udata
,
msg
,
RAFT_REQUESTVOTE_RESP
,
sizeof
(
*
msg
),
node
,
raft
);
}
int
sender_appendentries
(
raft_server_t
*
raft
,
void
*
udata
,
raft_node_t
*
node
,
msg_appendentries_t
*
msg
)
void
*
udata
,
raft_node_t
*
node
,
raft_appendentries_req_t
*
msg
)
{
msg
_entry_t
**
entries
=
calloc
(
1
,
sizeof
(
msg
_entry_t
*
)
*
msg
->
n_entries
);
raft
_entry_
req_
t
**
entries
=
calloc
(
1
,
sizeof
(
raft
_entry_
req_
t
*
)
*
msg
->
n_entries
);
int
i
;
for
(
i
=
0
;
i
<
msg
->
n_entries
;
i
++
)
{
entries
[
i
]
=
msg
->
entries
[
i
];
raft_entry_hold
(
entries
[
i
]);
}
msg
_entry_t
**
old_entries
=
msg
->
entries
;
raft
_entry_
req_
t
**
old_entries
=
msg
->
entries
;
msg
->
entries
=
entries
;
int
ret
=
__append_msg
(
udata
,
msg
,
RAFT_
MSG_
APPENDENTRIES
,
sizeof
(
*
msg
),
node
,
int
ret
=
__append_msg
(
udata
,
msg
,
RAFT_APPENDENTRIES
_REQ
,
sizeof
(
*
msg
),
node
,
raft
);
msg
->
entries
=
old_entries
;
return
ret
;
...
...
@@ -108,17 +110,17 @@ int sender_appendentries(raft_server_t* raft,
int
sender_appendentries_response
(
raft_server_t
*
raft
,
void
*
udata
,
raft_node_t
*
node
,
msg
_appendentries_resp
onse
_t
*
msg
)
raft
_appendentries_resp_t
*
msg
)
{
return
__append_msg
(
udata
,
msg
,
RAFT_
MSG_
APPENDENTRIES_RESP
ONSE
,
return
__append_msg
(
udata
,
msg
,
RAFT_APPENDENTRIES_RESP
,
sizeof
(
*
msg
),
node
,
raft
);
}
int
sender_entries_response
(
raft_server_t
*
raft
,
void
*
udata
,
raft_node_t
*
node
,
msg_entry_response_t
*
msg
)
void
*
udata
,
raft_node_t
*
node
,
raft_entry_resp_t
*
msg
)
{
return
__append_msg
(
udata
,
msg
,
RAFT_MSG_ENTRY_RESPONSE
,
sizeof
(
*
msg
),
node
,
raft
);
return
__append_msg
(
udata
,
msg
,
RAFT_ENTRY_RESP
,
sizeof
(
*
msg
),
node
,
raft
);
}
void
*
sender_new
(
void
*
address
)
...
...
@@ -160,38 +162,38 @@ void sender_poll_msgs(void* s)
{
switch
(
m
->
type
)
{
case
RAFT_
MSG_
APPENDENTRIES
:
case
RAFT_APPENDENTRIES
_REQ
:
{
msg
_appendentries_resp
onse
_t
response
;
raft
_appendentries_resp_t
response
;
raft_recv_appendentries
(
me
->
raft
,
m
->
sender
,
m
->
data
,
&
response
);
__append_msg
(
me
,
&
response
,
RAFT_
MSG_
APPENDENTRIES_RESP
ONSE
,
__append_msg
(
me
,
&
response
,
RAFT_APPENDENTRIES_RESP
,
sizeof
(
response
),
m
->
sender
,
me
->
raft
);
}
break
;
case
RAFT_
MSG_
APPENDENTRIES_RESP
ONSE
:
case
RAFT_APPENDENTRIES_RESP
:
raft_recv_appendentries_response
(
me
->
raft
,
m
->
sender
,
m
->
data
);
break
;
case
RAFT_
MSG_
REQUESTVOTE
:
case
RAFT_REQUESTVOTE
_REQ
:
{
msg
_requestvote_resp
onse
_t
response
;
raft
_requestvote_resp_t
response
;
raft_recv_requestvote
(
me
->
raft
,
m
->
sender
,
m
->
data
,
&
response
);
__append_msg
(
me
,
&
response
,
RAFT_
MSG_
REQUESTVOTE_RESP
ONSE
,
__append_msg
(
me
,
&
response
,
RAFT_REQUESTVOTE_RESP
,
sizeof
(
response
),
m
->
sender
,
me
->
raft
);
}
break
;
case
RAFT_
MSG_
REQUESTVOTE_RESP
ONSE
:
case
RAFT_REQUESTVOTE_RESP
:
raft_recv_requestvote_response
(
me
->
raft
,
m
->
sender
,
m
->
data
);
break
;
case
RAFT_
MSG_
ENTRY
:
case
RAFT_ENTRY
_REQ
:
{
msg
_entry_resp
onse
_t
response
;
raft
_entry_resp_t
response
;
raft_recv_entry
(
me
->
raft
,
m
->
data
,
&
response
);
__append_msg
(
me
,
&
response
,
RAFT_
MSG_
ENTRY_RESP
ONSE
,
__append_msg
(
me
,
&
response
,
RAFT_ENTRY_RESP
,
sizeof
(
response
),
m
->
sender
,
me
->
raft
);
}
break
;
case
RAFT_
MSG_
ENTRY_RESP
ONSE
:
case
RAFT_ENTRY_RESP
:
#if 0
raft_recv_entry_response(me->raft, m->sender, m->data);
#endif
...
...
tests/mock_send_functions.h
View file @
8ee3f264
...
...
@@ -3,12 +3,12 @@
typedef
enum
{
RAFT_
MSG_
REQUESTVOTE
,
RAFT_
MSG_
REQUESTVOTE_RESP
ONSE
,
RAFT_
MSG_
APPENDENTRIES
,
RAFT_
MSG_
APPENDENTRIES_RESP
ONSE
,
RAFT_
MSG_
ENTRY
,
RAFT_
MSG_
ENTRY_RESP
ONSE
,
RAFT_REQUESTVOTE
_REQ
,
RAFT_REQUESTVOTE_RESP
,
RAFT_APPENDENTRIES
_REQ
,
RAFT_APPENDENTRIES_RESP
,
RAFT_ENTRY
_REQ
,
RAFT_ENTRY_RESP
,
}
raft_message_type_e
;
void
senders_new
();
...
...
@@ -24,21 +24,21 @@ int sender_msgs_available(void* s);
void
sender_set_raft
(
void
*
s
,
void
*
r
);
int
sender_requestvote
(
raft_server_t
*
raft
,
void
*
udata
,
raft_node_t
*
node
,
msg
_requestvote_t
*
msg
);
void
*
udata
,
raft_node_t
*
node
,
raft
_requestvote_
req_
t
*
msg
);
int
sender_requestvote_response
(
raft_server_t
*
raft
,
void
*
udata
,
raft_node_t
*
node
,
msg
_requestvote_resp
onse
_t
*
msg
);
void
*
udata
,
raft_node_t
*
node
,
raft
_requestvote_resp_t
*
msg
);
int
sender_appendentries
(
raft_server_t
*
raft
,
void
*
udata
,
raft_node_t
*
node
,
msg
_appendentries_t
*
msg
);
void
*
udata
,
raft_node_t
*
node
,
raft
_appendentries_
req_
t
*
msg
);
int
sender_appendentries_response
(
raft_server_t
*
raft
,
void
*
udata
,
raft_node_t
*
node
,
msg
_appendentries_resp
onse
_t
*
msg
);
void
*
udata
,
raft_node_t
*
node
,
raft
_appendentries_resp_t
*
msg
);
int
sender_entries
(
raft_server_t
*
raft
,
void
*
udata
,
raft_node_t
*
node
,
msg
_entry_t
*
msg
);
void
*
udata
,
raft_node_t
*
node
,
raft
_entry_
req_
t
*
msg
);
int
sender_entries_response
(
raft_server_t
*
raft
,
void
*
udata
,
raft_node_t
*
node
,
msg
_entry_resp
onse
_t
*
msg
);
void
*
udata
,
raft_node_t
*
node
,
raft
_entry_resp_t
*
msg
);
#endif
/* MOCK_SEND_FUNCTIONS_H */
tests/test_server.c
View file @
8ee3f264
This diff is collapsed.
Click to expand it.
tests/test_snapshotting.c
View file @
8ee3f264
...
...
@@ -45,7 +45,7 @@ static int __raft_applylog(
static
int
__raft_send_requestvote
(
raft_server_t
*
raft
,
void
*
udata
,
raft_node_t
*
node
,
msg
_requestvote_t
*
msg
)
raft
_requestvote_
req_
t
*
msg
)
{
return
0
;
}
...
...
@@ -53,7 +53,7 @@ static int __raft_send_requestvote(raft_server_t* raft,
static
int
__raft_send_appendentries
(
raft_server_t
*
raft
,
void
*
udata
,
raft_node_t
*
node
,
msg
_appendentries_t
*
msg
)
raft
_appendentries_
req_
t
*
msg
)
{
return
0
;
}
...
...
@@ -61,16 +61,16 @@ static int __raft_send_appendentries(raft_server_t* raft,
static
int
__raft_send_appendentries_capture
(
raft_server_t
*
raft
,
void
*
udata
,
raft_node_t
*
node
,
msg
_appendentries_t
*
msg
)
raft
_appendentries_
req_
t
*
msg
)
{
*
((
msg
_appendentries_t
*
)
udata
)
=
*
msg
;
*
((
raft
_appendentries_
req_
t
*
)
udata
)
=
*
msg
;
return
0
;
}
static
int
__raft_send_snapshot_increment
(
raft_server_t
*
raft
,
void
*
udata
,
raft_node_t
*
node
,
msg
_snapshot_t
*
msg
)
raft
_snapshot_
req_
t
*
msg
)
{
int
*
counter
=
udata
;
...
...
@@ -122,7 +122,7 @@ struct test_data
static
int
test_send_snapshot_increment
(
raft_server_t
*
raft
,
void
*
udata
,
raft_node_t
*
node
,
msg
_snapshot_t
*
msg
)
raft
_snapshot_
req_
t
*
msg
)
{
struct
test_data
*
t
=
udata
;
...
...
@@ -208,7 +208,7 @@ void TestRaft_leader_begin_snapshot_fails_if_no_logs_to_compact(CuTest * tc)
void
*
r
=
raft_new
();
raft_set_callbacks
(
r
,
&
funcs
,
NULL
);
msg
_entry_resp
onse
_t
cr
;
raft
_entry_resp_t
cr
;
raft_add_node
(
r
,
NULL
,
1
,
1
);
raft_add_node
(
r
,
NULL
,
2
,
0
);
...
...
@@ -218,7 +218,7 @@ void TestRaft_leader_begin_snapshot_fails_if_no_logs_to_compact(CuTest * tc)
CuAssertIntEquals
(
tc
,
0
,
raft_get_log_count
(
r
));
/* entry message */
msg
_entry_t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
raft
_entry_
req_
t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
/* receive entry */
raft_recv_entry
(
r
,
ety
,
&
cr
);
...
...
@@ -241,7 +241,7 @@ void TestRaft_leader_will_not_apply_entry_if_snapshot_is_in_progress(CuTest * tc
void
*
r
=
raft_new
();
raft_set_callbacks
(
r
,
&
funcs
,
NULL
);
msg
_entry_resp
onse
_t
cr
;
raft
_entry_resp_t
cr
;
raft_add_node
(
r
,
NULL
,
1
,
1
);
raft_add_node
(
r
,
NULL
,
2
,
0
);
...
...
@@ -251,7 +251,7 @@ void TestRaft_leader_will_not_apply_entry_if_snapshot_is_in_progress(CuTest * tc
CuAssertIntEquals
(
tc
,
0
,
raft_get_log_count
(
r
));
/* entry message */
msg
_entry_t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
raft
_entry_
req_
t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
/* receive entry */
raft_recv_entry
(
r
,
ety
,
&
cr
);
...
...
@@ -293,7 +293,7 @@ void TestRaft_leader_snapshot_begin_fails_if_less_than_2_logs_to_compact(CuTest
void
*
r
=
raft_new
();
raft_set_callbacks
(
r
,
&
funcs
,
NULL
);
msg
_entry_resp
onse
_t
cr
;
raft
_entry_resp_t
cr
;
raft_add_node
(
r
,
NULL
,
1
,
1
);
raft_add_node
(
r
,
NULL
,
2
,
0
);
...
...
@@ -303,7 +303,7 @@ void TestRaft_leader_snapshot_begin_fails_if_less_than_2_logs_to_compact(CuTest
CuAssertIntEquals
(
tc
,
0
,
raft_get_log_count
(
r
));
/* entry message */
msg
_entry_t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
raft
_entry_
req_
t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
/* receive entry */
raft_recv_entry
(
r
,
ety
,
&
cr
);
...
...
@@ -322,7 +322,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc)
raft_server_t
*
r
=
raft_new
();
raft_set_callbacks
(
r
,
&
funcs
,
NULL
);
msg
_entry_resp
onse
_t
cr
;
raft
_entry_resp_t
cr
;
raft_add_node
(
r
,
NULL
,
1
,
1
);
raft_add_node
(
r
,
NULL
,
2
,
0
);
...
...
@@ -333,7 +333,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc)
CuAssertIntEquals
(
tc
,
0
,
raft_get_log_count
(
r
));
/* entry message */
msg
_entry_t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
raft
_entry_
req_
t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
/* receive entry */
raft_recv_entry
(
r
,
ety
,
&
cr
);
...
...
@@ -387,7 +387,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted2(CuTest * tc)
void
*
r
=
raft_new
();
raft_set_callbacks
(
r
,
&
funcs
,
NULL
);
msg
_entry_resp
onse
_t
cr
;
raft
_entry_resp_t
cr
;
raft_add_node
(
r
,
NULL
,
1
,
1
);
raft_add_node
(
r
,
NULL
,
2
,
0
);
...
...
@@ -398,7 +398,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted2(CuTest * tc)
CuAssertIntEquals
(
tc
,
0
,
raft_get_log_count
(
r
));
/* entry message */
msg
_entry_t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
raft
_entry_
req_
t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
/* receive entry */
raft_recv_entry
(
r
,
ety
,
&
cr
);
...
...
@@ -428,7 +428,7 @@ void TestRaft_joinee_needs_to_get_snapshot(CuTest * tc)
void
*
r
=
raft_new
();
raft_set_callbacks
(
r
,
&
funcs
,
NULL
);
msg
_entry_resp
onse
_t
cr
;
raft
_entry_resp_t
cr
;
raft_add_node
(
r
,
NULL
,
1
,
1
);
raft_add_node
(
r
,
NULL
,
2
,
0
);
...
...
@@ -438,7 +438,7 @@ void TestRaft_joinee_needs_to_get_snapshot(CuTest * tc)
CuAssertIntEquals
(
tc
,
0
,
raft_get_log_count
(
r
));
/* entry message */
msg
_entry_t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
raft
_entry_
req_
t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
/* receive entry */
raft_recv_entry
(
r
,
ety
,
&
cr
);
...
...
@@ -469,7 +469,7 @@ void TestRaft_follower_load_from_snapshot(CuTest * tc)
CuAssertIntEquals
(
tc
,
0
,
raft_get_log_count
(
r
));
/* entry message */
msg
_entry_t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
raft
_entry_
req_
t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
CuAssertIntEquals
(
tc
,
0
,
raft_get_log_count
(
r
));
CuAssertIntEquals
(
tc
,
0
,
raft_begin_load_snapshot
(
r
,
5
,
5
));
...
...
@@ -623,7 +623,7 @@ void TestRaft_leader_sends_snapshot_when_node_next_index_was_compacted(CuTest* t
.
send_appendentries
=
__raft_send_appendentries_capture
,
};
msg
_appendentries_t
ae
;
raft
_appendentries_
req_
t
ae
;
raft_set_callbacks
(
r
,
&
funcs
,
&
ae
);
/* node wants an entry just one after the snapshot index */
...
...
@@ -644,7 +644,7 @@ void TestRaft_recv_entry_fails_if_snapshot_in_progress(CuTest* tc)
void
*
r
=
raft_new
();
raft_set_callbacks
(
r
,
&
funcs
,
NULL
);
msg
_entry_resp
onse
_t
cr
;
raft
_entry_resp_t
cr
;
raft_add_node
(
r
,
NULL
,
1
,
1
);
raft_add_node
(
r
,
NULL
,
2
,
0
);
...
...
@@ -654,7 +654,7 @@ void TestRaft_recv_entry_fails_if_snapshot_in_progress(CuTest* tc)
CuAssertIntEquals
(
tc
,
0
,
raft_get_log_count
(
r
));
/* entry message */
msg
_entry_t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
raft
_entry_
req_
t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
/* receive entry */
raft_recv_entry
(
r
,
ety
,
&
cr
);
...
...
@@ -679,7 +679,7 @@ void TestRaft_recv_entry_succeeds_if_snapshot_nonblocking_apply_is_set(CuTest* t
void
*
r
=
raft_new
();
raft_set_callbacks
(
r
,
&
funcs
,
NULL
);
msg
_entry_resp
onse
_t
cr
;
raft
_entry_resp_t
cr
;
raft_add_node
(
r
,
NULL
,
1
,
1
);
raft_add_node
(
r
,
NULL
,
2
,
0
);
...
...
@@ -689,7 +689,7 @@ void TestRaft_recv_entry_succeeds_if_snapshot_nonblocking_apply_is_set(CuTest* t
CuAssertIntEquals
(
tc
,
0
,
raft_get_log_count
(
r
));
/* entry message */
msg
_entry_t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
raft
_entry_
req_
t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
/* receive entry */
raft_recv_entry
(
r
,
ety
,
&
cr
);
...
...
@@ -722,10 +722,10 @@ void TestRaft_follower_recv_appendentries_is_successful_when_previous_log_idx_eq
CuAssertIntEquals
(
tc
,
0
,
raft_begin_load_snapshot
(
r
,
2
,
2
));
CuAssertIntEquals
(
tc
,
0
,
raft_end_load_snapshot
(
r
));
msg
_appendentries_t
ae
;
msg
_appendentries_resp
onse
_t
aer
;
raft
_appendentries_
req_
t
ae
;
raft
_appendentries_resp_t
aer
;
memset
(
&
ae
,
0
,
sizeof
(
msg
_appendentries_t
));
memset
(
&
ae
,
0
,
sizeof
(
raft
_appendentries_
req_
t
));
ae
.
term
=
3
;
ae
.
prev_log_idx
=
2
;
ae
.
prev_log_term
=
2
;
...
...
@@ -765,7 +765,7 @@ void TestRaft_leader_sends_appendentries_with_correct_prev_log_idx_when_snapshot
/* receive appendentries messages */
raft_send_appendentries
(
r
,
p
);
msg
_appendentries_t
*
ae
=
sender_poll_msg_data
(
sender
);
raft
_appendentries_
req_
t
*
ae
=
sender_poll_msg_data
(
sender
);
CuAssertTrue
(
tc
,
NULL
!=
ae
);
CuAssertIntEquals
(
tc
,
2
,
ae
->
prev_log_term
);
CuAssertIntEquals
(
tc
,
4
,
ae
->
prev_log_idx
);
...
...
@@ -780,7 +780,7 @@ void TestRaft_cancel_snapshot_restores_state(CuTest* tc)
void
*
r
=
raft_new
();
raft_set_callbacks
(
r
,
&
funcs
,
NULL
);
msg
_entry_resp
onse
_t
cr
;
raft
_entry_resp_t
cr
;
raft_add_node
(
r
,
NULL
,
1
,
1
);
raft_add_node
(
r
,
NULL
,
2
,
0
);
...
...
@@ -790,7 +790,7 @@ void TestRaft_cancel_snapshot_restores_state(CuTest* tc)
CuAssertIntEquals
(
tc
,
0
,
raft_get_log_count
(
r
));
/* single entry */
msg
_entry_t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
raft
_entry_
req_
t
*
ety
=
__MAKE_ENTRY
(
1
,
1
,
"entry"
);
raft_recv_entry
(
r
,
ety
,
&
cr
);
ety
=
__MAKE_ENTRY
(
2
,
1
,
"entry"
);
...
...
@@ -869,7 +869,7 @@ void TestRaft_leader_sends_snapshot_if_log_was_compacted(CuTest* tc)
raft_node_set_match_idx
(
node
,
1
);
raft_node_set_next_idx
(
node
,
2
);
msg
_appendentries_resp
onse
_t
aer
;
raft
_appendentries_resp_t
aer
;
aer
.
term
=
1
;
aer
.
success
=
0
;
aer
.
current_idx
=
1
;
...
...
@@ -895,7 +895,7 @@ void TestRaft_clear_snapshot_on_leader_change(CuTest * tc)
raft_set_callbacks
(
r
,
&
funcs
,
&
data
);
raft_add_node
(
r
,
NULL
,
1
,
1
);
msg
_snapshot_t
msg
=
{
raft
_snapshot_
req_
t
msg
=
{
.
leader_id
=
2
,
.
snapshot_index
=
1
,
.
snapshot_term
=
1
,
...
...
@@ -905,7 +905,7 @@ void TestRaft_clear_snapshot_on_leader_change(CuTest * tc)
.
chunk
.
last_chunk
=
0
,
};
msg
_snapshot_resp
onse
_t
resp
=
{
0
};
raft
_snapshot_resp_t
resp
=
{
0
};
raft_recv_snapshot
(
r
,
NULL
,
&
msg
,
&
resp
);
CuAssertIntEquals
(
tc
,
1
,
data
.
store_chunk
);
...
...
@@ -935,7 +935,7 @@ void TestRaft_reject_wrong_offset(CuTest * tc)
raft_set_callbacks
(
r
,
&
funcs
,
&
data
);
raft_add_node
(
r
,
NULL
,
1
,
1
);
msg
_snapshot_t
msg
=
{
raft
_snapshot_
req_
t
msg
=
{
.
leader_id
=
2
,
.
snapshot_index
=
1
,
.
snapshot_term
=
1
,
...
...
@@ -946,7 +946,7 @@ void TestRaft_reject_wrong_offset(CuTest * tc)
.
chunk
.
last_chunk
=
0
,
};
msg
_snapshot_resp
onse
_t
resp
=
{
0
};
raft
_snapshot_resp_t
resp
=
{
0
};
raft_recv_snapshot
(
r
,
NULL
,
&
msg
,
&
resp
);
CuAssertIntEquals
(
tc
,
1
,
data
.
store_chunk
);
...
...
@@ -975,7 +975,7 @@ void TestRaft_set_last_chunk_on_duplicate(CuTest * tc)
raft_set_callbacks
(
r
,
&
funcs
,
&
data
);
raft_add_node
(
r
,
NULL
,
1
,
1
);
msg
_snapshot_t
msg
=
{
raft
_snapshot_
req_
t
msg
=
{
.
term
=
1
,
.
leader_id
=
2
,
.
snapshot_index
=
5
,
...
...
@@ -987,13 +987,13 @@ void TestRaft_set_last_chunk_on_duplicate(CuTest * tc)
.
chunk
.
last_chunk
=
1
,
};
msg
_snapshot_resp
onse
_t
resp
=
{
0
};
raft
_snapshot_resp_t
resp
=
{
0
};
raft_recv_snapshot
(
r
,
NULL
,
&
msg
,
&
resp
);
CuAssertIntEquals
(
tc
,
1
,
data
.
store_chunk
);
CuAssertIntEquals
(
tc
,
1
,
data
.
load
);
msg
_snapshot_t
msg2
=
{
raft
_snapshot_
req_
t
msg2
=
{
.
term
=
1
,
.
leader_id
=
2
,
.
snapshot_index
=
4
,
...
...
tests/virtraft2.py
View file @
8ee3f264
...
...
@@ -530,7 +530,7 @@ class Network(object):
new_msg
=
ffi
.
cast
(
ffi
.
typeof
(
msg
),
lib
.
malloc
(
msg_size
))
ffi
.
memmove
(
new_msg
,
msg
,
msg_size
)
if
msg_type
==
'
msg
_appendentries_t *'
:
if
msg_type
==
'
raft
_appendentries_
req_
t *'
:
new_msg
.
entries
=
lib
.
raft_entry_array_deepcopy
(
msg
.
entries
,
msg
.
n_entries
)
self
.
messages
.
append
(
Message
(
new_msg
,
sendor
,
sendee
))
...
...
@@ -540,9 +540,9 @@ class Network(object):
logger
.
debug
(
f
"poll_message:
{
msg
.
sendor
.
id
}
->
{
msg
.
sendee
.
id
}
(
{
msg_type
}
"
)
if
msg_type
==
'
msg
_appendentries_t *'
:
if
msg_type
==
'
raft
_appendentries_
req_
t *'
:
node
=
lib
.
raft_get_node
(
msg
.
sendee
.
raft
,
msg
.
sendor
.
id
)
response
=
ffi
.
new
(
'
msg
_appendentries_resp
onse
_t*'
)
response
=
ffi
.
new
(
'
raft
_appendentries_resp_t*'
)
e
=
lib
.
raft_recv_appendentries
(
msg
.
sendee
.
raft
,
node
,
msg
.
data
,
response
)
if
lib
.
RAFT_ERR_SHUTDOWN
==
e
:
logger
.
error
(
'Catastrophic'
)
...
...
@@ -553,27 +553,27 @@ class Network(object):
else
:
self
.
enqueue_msg
(
response
,
msg
.
sendee
,
msg
.
sendor
)
elif
msg_type
==
'
msg
_appendentries_resp
onse
_t *'
:
elif
msg_type
==
'
raft
_appendentries_resp_t *'
:
node
=
lib
.
raft_get_node
(
msg
.
sendee
.
raft
,
msg
.
sendor
.
id
)
lib
.
raft_recv_appendentries_response
(
msg
.
sendee
.
raft
,
node
,
msg
.
data
)
elif
msg_type
==
'
msg
_snapshot_t *'
:
response
=
ffi
.
new
(
'
msg
_snapshot_resp
onse
_t *'
)
elif
msg_type
==
'
raft
_snapshot_
req_
t *'
:
response
=
ffi
.
new
(
'
raft
_snapshot_resp_t *'
)
node
=
lib
.
raft_get_node
(
msg
.
sendee
.
raft
,
msg
.
sendor
.
id
)
lib
.
raft_recv_snapshot
(
msg
.
sendee
.
raft
,
node
,
msg
.
data
,
response
)
self
.
enqueue_msg
(
response
,
msg
.
sendee
,
msg
.
sendor
)
elif
msg_type
==
'
msg
_snapshot_resp
onse
_t *'
:
elif
msg_type
==
'
raft
_snapshot_resp_t *'
:
node
=
lib
.
raft_get_node
(
msg
.
sendee
.
raft
,
msg
.
sendor
.
id
)
lib
.
raft_recv_snapshot_response
(
msg
.
sendee
.
raft
,
node
,
msg
.
data
)
elif
msg_type
==
'
msg
_requestvote_t *'
:
response
=
ffi
.
new
(
'
msg
_requestvote_resp
onse
_t*'
)
elif
msg_type
==
'
raft
_requestvote_
req_
t *'
:
response
=
ffi
.
new
(
'
raft
_requestvote_resp_t*'
)
node
=
lib
.
raft_get_node
(
msg
.
sendee
.
raft
,
msg
.
sendor
.
id
)
lib
.
raft_recv_requestvote
(
msg
.
sendee
.
raft
,
node
,
msg
.
data
,
response
)
self
.
enqueue_msg
(
response
,
msg
.
sendee
,
msg
.
sendor
)
elif
msg_type
==
'
msg
_requestvote_resp
onse
_t *'
:
elif
msg_type
==
'
raft
_requestvote_resp_t *'
:
node
=
lib
.
raft_get_node
(
msg
.
sendee
.
raft
,
msg
.
sendor
.
id
)
e
=
lib
.
raft_recv_requestvote_response
(
msg
.
sendee
.
raft
,
node
,
msg
.
data
)
if
lib
.
RAFT_ERR_SHUTDOWN
==
e
:
...
...
@@ -941,9 +941,9 @@ class RaftServer(object):
self
.
network
=
network
def
load_callbacks
(
self
):
self
.
raft_send_requestvote
=
ffi
.
callback
(
"int(raft_server_t*, void*, raft_node_t*,
msg
_requestvote_t*)"
,
raft_send_requestvote
)
self
.
raft_send_appendentries
=
ffi
.
callback
(
"int(raft_server_t*, void*, raft_node_t*,
msg
_appendentries_t*)"
,
raft_send_appendentries
)
self
.
raft_send_snapshot
=
ffi
.
callback
(
"int(raft_server_t*, void* , raft_node_t*,
msg
_snapshot_t*)"
,
raft_send_snapshot
)
self
.
raft_send_requestvote
=
ffi
.
callback
(
"int(raft_server_t*, void*, raft_node_t*,
raft
_requestvote_
req_
t*)"
,
raft_send_requestvote
)
self
.
raft_send_appendentries
=
ffi
.
callback
(
"int(raft_server_t*, void*, raft_node_t*,
raft
_appendentries_
req_
t*)"
,
raft_send_appendentries
)
self
.
raft_send_snapshot
=
ffi
.
callback
(
"int(raft_server_t*, void* , raft_node_t*,
raft
_snapshot_
req_
t*)"
,
raft_send_snapshot
)
self
.
raft_load_snapshot
=
ffi
.
callback
(
"int(raft_server_t*, void*, raft_index_t, raft_term_t)"
,
raft_load_snapshot
)
self
.
raft_clear_snapshot
=
ffi
.
callback
(
"int(raft_server_t*, void*)"
,
raft_clear_snapshot
)
self
.
raft_get_snapshot_chunk
=
ffi
.
callback
(
"int(raft_server_t*, void*, raft_node_t*, raft_size_t offset, raft_snapshot_chunk_t*)"
,
raft_get_snapshot_chunk
)
...
...
@@ -962,7 +962,7 @@ class RaftServer(object):
def
recv_entry
(
self
,
ety
):
# FIXME: leak
response
=
ffi
.
new
(
'
msg
_entry_resp
onse
_t*'
)
response
=
ffi
.
new
(
'
raft
_entry_resp_t*'
)
return
lib
.
raft_recv_entry
(
self
.
raft
,
ety
,
response
)
def
get_entry
(
self
,
idx
):
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
.
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment