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

Throttle applying entries and processing read requests (#106)

Throttle operations if it takes longer than request timeout
parent ed8b54c0
......@@ -667,6 +667,19 @@ typedef raft_index_t (
raft_entry_t **entries
);
/** Callback to retrieve monotonic timestamp in microseconds .
*
* @param[in] raft The Raft server making this callback
* @param[in] user_data User data that is passed from Raft server
* @return Timestamp in microseconds
*/
typedef raft_time_t (
*raft_timestamp_f
) (
raft_server_t *raft,
void *user_data
);
typedef struct
{
/** Callback for sending request vote messages */
......@@ -735,6 +748,9 @@ typedef struct
/** Callback for preparing entries to send in a raft_appendentries_req */
raft_get_entries_to_send_f get_entries_to_send;
/** Callback to retrieve monotonic timestamp in microseconds */
raft_timestamp_f timestamp;
} raft_cbs_t;
/** A generic notification callback used to allow Raft to notify caller
......@@ -986,12 +1002,11 @@ raft_node_t* raft_add_non_voting_node(raft_server_t* me, void* udata, raft_node_
void raft_remove_node(raft_server_t* me, raft_node_t* node);
/** Process events that are dependent on time passing.
* @param[in] msec_elapsed Time in milliseconds since the last call
* @return
* 0 on success;
* -1 on failure;
* RAFT_ERR_SHUTDOWN when server MUST shutdown */
int raft_periodic(raft_server_t* me, int msec_elapsed);
int raft_periodic(raft_server_t *me);
/** Receive an appendentries message.
*
......@@ -1142,7 +1157,7 @@ int raft_is_candidate(raft_server_t* me);
/**
* @return currently elapsed timeout in milliseconds */
int raft_get_timeout_elapsed(raft_server_t* me);
raft_time_t raft_get_timeout_elapsed(raft_server_t* me);
/**
* @return index of last applied entry */
......@@ -1540,10 +1555,6 @@ void raft_handle_append_cfg_change(raft_server_t* me, raft_entry_t* ety, raft_in
int raft_recv_read_request(raft_server_t* me, raft_read_request_callback_f cb, void *cb_arg);
/** Attempt to process read queue.
*/
void raft_process_read_queue(raft_server_t* me);
/** Invoke a leadership transfer to targeted node
*
* @param[in] node_id targeted node, RAFT_NODE_ID_NONE for automatic target
......@@ -1616,7 +1627,9 @@ raft_index_t raft_get_num_snapshottable_logs(raft_server_t* me);
*
* @param[in] sync_index Entry index of the last persisted entry. '0' to skip
* updating persisted index.
* @return 0 on success
* @return
* 0 on success
* RAFT_ERR_SHUTDOWN when server MUST shutdown
*/
int raft_flush(raft_server_t* me, raft_index_t sync_index);
......@@ -1647,22 +1660,23 @@ raft_index_t raft_get_index_to_sync(raft_server_t *me);
* disable-apply : Skip applying entries. Useful for testing.
*
*
* | Enum | Type | Valid values | Default value |
* | ----------------------------- | ---- | ---------------- | --------------- |
* | RAFT_CONFIG_ELECTION_TIMEOUT | int | Positive integer | 1000 millis |
* | RAFT_CONFIG_REQUEST_TIMEOUT | int | Positive integer | 200 millis |
* | RAFT_CONFIG_AUTO_FLUSH | int | 0 or 1 | 0 |
* | RAFT_CONFIG_LOG_ENABLED | int | 0 or 1 | 0 |
* | RAFT_CONFIG_NONBLOCKING_APPLY | int | 0 or 1 | 0 |
* | RAFT_CONFIG_DISABLE_APPLY | int | 0 or 1 | 0 |
* | Enum | Type | Valid values | Default |
* | ----------------------------- | ------------ | ------------ | ----------- |
* | RAFT_CONFIG_ELECTION_TIMEOUT | raft_time_t | > 0 | 1000 millis |
* | RAFT_CONFIG_REQUEST_TIMEOUT | raft_time_t | > 0 | 200 millis |
* | RAFT_CONFIG_AUTO_FLUSH | int | 0 or 1 | 0 |
* | RAFT_CONFIG_LOG_ENABLED | int | 0 or 1 | 0 |
* | RAFT_CONFIG_NONBLOCKING_APPLY | int | 0 or 1 | 0 |
* | RAFT_CONFIG_DISABLE_APPLY | int | 0 or 1 | 0 |
*
* Example:
*
* - Set
* raft_config(raft, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 4000);
* raft_time_t timeout = 4000;
* raft_config(raft, 1, RAFT_CONFIG_ELECTION_TIMEOUT, timeout);
*
* - Get
* int election_timeout;
* raft_time_t election_timeout;
* raft_config(raft, 0, RAFT_CONFIG_ELECTION_TIMEOUT, &election_timeout);
*
* @param set 1 to set the value, 0 to get the current value.
......@@ -1672,4 +1686,18 @@ raft_index_t raft_get_index_to_sync(raft_server_t *me);
*/
int raft_config(raft_server_t *me, int set, raft_config_e config, ...);
/** Returns non-zero if there are read requests or entries ready to be executed.
*
* If executing entries/read requests take longer than `request-timeout`,
* raft_flush() will return early. In that case, application should call
* raft_flush() again to continue operation later, preferably after processing
* messages from the network. That way, server can send/receive heartbeat
* messages and execute long running batch of operations without affecting
* cluster availability.
*
* @param[in] raft The Raft server
* @return 0 if there is no pending operations, non-zero otherwise.
*/
int raft_pending_operations(raft_server_t *me);
#endif /* RAFT_H_ */
......@@ -55,13 +55,22 @@ struct raft_server {
raft_state_e state;
/* amount of time left till timeout */
int timeout_elapsed;
raft_time_t timeout_elapsed;
/* timestamp in milliseconds */
raft_time_t timestamp;
/* deadline to stop executing operations */
raft_time_t exec_deadline;
/* non-zero if there are ready to be executed operations. */
int pending_operations;
raft_node_t** nodes;
int num_nodes;
/* timer interval to check if we still have quorum */
long quorum_timeout;
raft_time_t quorum_timeout;
/* latest quorum id for the previous quorum_timeout round */
raft_msg_id_t last_acked_msg_id;
......@@ -104,9 +113,9 @@ struct raft_server {
raft_read_request_t *read_queue_head;
raft_read_request_t *read_queue_tail;
raft_node_id_t node_transferring_leader_to; // the node we are targeting for leadership
long transfer_leader_time; // how long we should wait for leadership transfer to take, before aborting
int sent_timeout_now; // if we've already sent a leadership transfer signal
raft_node_id_t node_transferring_leader_to; /* Leader transfer target. */
raft_time_t transfer_leader_time; /* Leader transfer timeout. */
int sent_timeout_now; /* If we've already sent timeout_now message. */
/* Index of the log entry that need to be written to the disk. Only useful
......@@ -117,8 +126,8 @@ struct raft_server {
/* Configuration parameters */
int election_timeout; /* Timeout for a follower to start an election */
int request_timeout; /* Heartbeat timeout */
raft_time_t election_timeout; /* Timeout for a node to start an election */
raft_time_t request_timeout; /* Heartbeat timeout */
int nonblocking_apply; /* Apply entries even when snapshot is in progress */
int auto_flush; /* Automatically call raft_flush() */
int log_enabled; /* Enable library logs */
......@@ -194,4 +203,8 @@ 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);
int raft_periodic_internal(raft_server_t *me, raft_time_t milliseconds);
int raft_exec_operations(raft_server_t *me);
#endif /* RAFT_PRIVATE_H_ */
......@@ -35,4 +35,9 @@ typedef int raft_node_id_t;
*/
typedef unsigned long raft_msg_id_t;
/**
* Time type.
*/
typedef long long raft_time_t;
#endif /* RAFT_DEFS_H_ */
......@@ -14,6 +14,7 @@
#include <assert.h>
#include <stdint.h>
#include <stdarg.h>
#include <limits.h>
#include "raft.h"
#include "raft_private.h"
......@@ -112,6 +113,7 @@ raft_server_t* raft_new_with_log(const raft_log_impl_t *log_impl, void *log_arg)
me->election_timeout = 1000;
me->node_transferring_leader_to = RAFT_NODE_ID_NONE;
me->auto_flush = 1;
me->exec_deadline = LLONG_MAX;
raft_update_quorum_meta(me, me->msg_id);
......@@ -360,7 +362,7 @@ int raft_delete_entry_from_idx(raft_server_t* me, raft_index_t idx)
int raft_election_start(raft_server_t* me, int skip_precandidate)
{
raft_log(me,
"election starting: %d %d, term: %ld ci: %ld",
"election starting: %d %lld, term: %ld ci: %ld",
me->election_timeout_rand, me->timeout_elapsed, me->current_term,
raft_get_current_idx(me));
......@@ -549,8 +551,34 @@ static raft_msg_id_t quorum_msg_id(raft_server_t* me)
return msg_ids[num_voters / 2];
}
int raft_periodic(raft_server_t* me, int msec_since_last_period)
static raft_time_t raft_time_millis(raft_server_t *me)
{
return me->cb.timestamp ? me->cb.timestamp(me, me->udata) / 1000 : 0;
}
int raft_periodic(raft_server_t *me)
{
return raft_periodic_internal(me, -1);
}
int raft_periodic_internal(raft_server_t *me,
raft_time_t msec_since_last_period)
{
if (msec_since_last_period < 0) {
raft_time_t timestamp = raft_time_millis(me);
assert(timestamp >= me->timestamp);
/* If this is the first call, previous timestamp will be zero. In this
* case, we just assume `0` millisecond has passed. */
if (me->timestamp == 0) {
msec_since_last_period = 0;
} else {
msec_since_last_period = timestamp - me->timestamp;
}
me->timestamp = timestamp;
}
me->timeout_elapsed += msec_since_last_period;
/* Only one voting node means it's safe for us to become the leader */
......@@ -617,14 +645,7 @@ int raft_periodic(raft_server_t* me, int msec_since_last_period)
return e;
}
int e = raft_apply_all(me);
if (e != 0) {
return e;
}
raft_process_read_queue(me);
return 0;
return raft_exec_operations(me);
}
raft_entry_t* raft_get_entry_from_idx(raft_server_t* me, raft_index_t etyidx)
......@@ -1612,16 +1633,22 @@ int raft_msg_entry_response_committed(raft_server_t* me,
return r->idx <= raft_get_commit_idx(me);
}
int raft_apply_all(raft_server_t* me)
int raft_apply_all(raft_server_t *me)
{
if (!raft_is_apply_allowed(me))
if (!raft_is_apply_allowed(me)) {
return 0;
}
while (me->commit_idx > me->last_applied_idx) {
if (raft_time_millis(me) > me->exec_deadline) {
me->pending_operations = 1;
return 0;
}
while (me->commit_idx > me->last_applied_idx)
{
int e = raft_apply_entry(me);
if (0 != e)
if (e != 0) {
return e;
}
}
return 0;
......@@ -1927,7 +1954,7 @@ static void pop_read_queue(raft_server_t *me, int can_read)
raft_free(p);
}
void raft_process_read_queue(raft_server_t* me)
void raft_process_read_queue(raft_server_t *me)
{
if (!me->read_queue_head) {
return;
......@@ -1938,6 +1965,7 @@ void raft_process_read_queue(raft_server_t* me)
while (me->read_queue_head) {
pop_read_queue(me, 0);
}
return;
}
/* As a leader we can process requests that fulfill these conditions:
......@@ -1954,6 +1982,12 @@ void raft_process_read_queue(raft_server_t* me)
while (me->read_queue_head &&
me->read_queue_head->msg_id <= last_acked_msgid &&
me->read_queue_head->read_idx <= me->last_applied_idx) {
if (raft_time_millis(me) > me->exec_deadline) {
me->pending_operations = 1;
return;
}
pop_read_queue(me, 1);
}
}
......@@ -2124,14 +2158,8 @@ int raft_flush(raft_server_t* me, raft_index_t sync_index)
raft_send_appendentries(me, me->nodes[i]);
}
int e = raft_apply_all(me);
if (e != 0) {
return e;
}
out:
raft_process_read_queue(me);
return 0;
return raft_exec_operations(me);
}
int raft_config(raft_server_t *me, int set, raft_config_e config, ...)
......@@ -2144,18 +2172,18 @@ int raft_config(raft_server_t *me, int set, raft_config_e config, ...)
switch (config) {
case RAFT_CONFIG_ELECTION_TIMEOUT:
if (set) {
me->election_timeout = va_arg(va, int);
me->election_timeout = va_arg(va, raft_time_t);
raft_update_quorum_meta(me, me->last_acked_msg_id);
raft_randomize_election_timeout(me);
} else {
*(va_arg(va, int*)) = me->election_timeout;
*(va_arg(va, raft_time_t*)) = me->election_timeout;
}
break;
case RAFT_CONFIG_REQUEST_TIMEOUT:
if (set) {
me->request_timeout = va_arg(va, int);
me->request_timeout = va_arg(va, raft_time_t);
} else {
*(va_arg(va, int*)) = me->request_timeout;
*(va_arg(va, raft_time_t*)) = me->request_timeout;
}
break;
case RAFT_CONFIG_AUTO_FLUSH:
......@@ -2195,3 +2223,27 @@ int raft_config(raft_server_t *me, int set, raft_config_e config, ...)
return ret;
}
int raft_pending_operations(raft_server_t *me)
{
return me->pending_operations;
}
int raft_exec_operations(raft_server_t *me)
{
me->pending_operations = 0;
me->exec_deadline = raft_time_millis(me) + me->request_timeout;
int e = raft_apply_all(me);
if (e != 0) {
goto out;
}
raft_process_read_queue(me);
out:
/* Set deadline to MAX to prevent raft_apply_all() to return early if this
* function is called manually or inside raft_begin_snapshot() */
me->exec_deadline = LLONG_MAX;
return e;
}
......@@ -32,7 +32,7 @@ int raft_get_num_voting_nodes(raft_server_t* me)
return num;
}
int raft_get_timeout_elapsed(raft_server_t* me)
raft_time_t raft_get_timeout_elapsed(raft_server_t* me)
{
return me->timeout_elapsed;
}
......
......@@ -60,7 +60,7 @@ void TestRaft_scenario_leader_appears(CuTest * tc)
}
/* NOTE: important for 1st node to send vote request before others */
raft_periodic(r[0], 1000);
raft_periodic_internal(r[0], 1000);
for (i = 0; i < 20; i++)
{
......@@ -74,7 +74,7 @@ one_more_time:
goto one_more_time;
for (j = 0; j < 3; j++)
raft_periodic(r[j], 100);
raft_periodic_internal(r[j], 100);
}
int leaders = 0;
......
......@@ -265,7 +265,7 @@ void TestRaft_server_starts_with_election_timeout_of_1000ms(CuTest * tc)
{
void *r = raft_new();
int election_timeout;
raft_time_t election_timeout;
raft_config(r, 0, RAFT_CONFIG_ELECTION_TIMEOUT, &election_timeout);
CuAssertTrue(tc, 1000 == election_timeout);
......@@ -275,7 +275,7 @@ void TestRaft_server_starts_with_request_timeout_of_200ms(CuTest * tc)
{
void *r = raft_new();
int request_timeout;
raft_time_t request_timeout;
raft_config(r, 0, RAFT_CONFIG_REQUEST_TIMEOUT, &request_timeout);
CuAssertTrue(tc, 200 == request_timeout);
......@@ -435,7 +435,7 @@ void TestRaft_server_increment_lastApplied_when_lastApplied_lt_commitidx(
raft_set_commit_idx(r, 1);
/* let time lapse */
raft_periodic(r, 1);
raft_periodic_internal(r, 1);
CuAssertIntEquals(tc, 1, raft_get_last_applied_idx(r));
}
......@@ -462,7 +462,7 @@ void TestRaft_user_applylog_error_propogates_to_periodic(
raft_set_commit_idx(r, 1);
/* let time lapse */
CuAssertIntEquals(tc, RAFT_ERR_SHUTDOWN, raft_periodic(r, 1));
CuAssertIntEquals(tc, RAFT_ERR_SHUTDOWN, raft_periodic_internal(r, 1));
CuAssertIntEquals(tc, 0, raft_get_last_applied_idx(r));
}
......@@ -491,10 +491,10 @@ void TestRaft_server_periodic_elapses_election_timeout(CuTest * tc)
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
CuAssertTrue(tc, 0 == raft_get_timeout_elapsed(r));
raft_periodic(r, 0);
raft_periodic_internal(r, 0);
CuAssertTrue(tc, 0 == raft_get_timeout_elapsed(r));
raft_periodic(r, 100);
raft_periodic_internal(r, 100);
CuAssertTrue(tc, 100 == raft_get_timeout_elapsed(r));
}
......@@ -514,7 +514,7 @@ void TestRaft_server_election_timeout_does_not_promote_us_to_leader_if_there_is_
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
/* clock over (ie. 1000 + 1), causing new election */
raft_periodic(r, 1001);
raft_periodic_internal(r, 1001);
CuAssertTrue(tc, 0 == raft_is_leader(r));
}
......@@ -526,7 +526,7 @@ void TestRaft_server_election_timeout_does_not_promote_us_to_leader_if_we_are_no
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
/* clock over (ie. 1000 + 1), causing new election */
raft_periodic(r, 1001);
raft_periodic_internal(r, 1001);
CuAssertTrue(tc, 0 == raft_is_leader(r));
CuAssertTrue(tc, 0 == raft_get_current_term(r));
......@@ -540,7 +540,7 @@ void TestRaft_server_election_timeout_does_not_start_election_if_there_are_no_vo
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
/* clock over (ie. 1000 + 1), causing new election */
raft_periodic(r, 1001);
raft_periodic_internal(r, 1001);
CuAssertTrue(tc, 0 == raft_get_current_term(r));
}
......@@ -553,7 +553,7 @@ void TestRaft_server_election_timeout_does_promote_us_to_leader_if_there_is_only
raft_term_t old_term = raft_get_current_term(r);
/* clock over (ie. 1000 + 1), causing new election */
raft_periodic(r, 1001);
raft_periodic_internal(r, 1001);
CuAssertTrue(tc, 1 == raft_is_leader(r));
CuAssertTrue(tc, old_term + 1 == raft_get_current_term(r));
......@@ -573,7 +573,7 @@ void TestRaft_server_election_timeout_does_promote_us_to_leader_if_there_is_only
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
/* clock over (ie. 1000 + 1), causing new election */
raft_periodic(r, 1001);
raft_periodic_internal(r, 1001);
CuAssertTrue(tc, 1 == raft_is_leader(r));
}
......@@ -883,7 +883,7 @@ void TestRaft_server_recv_requestvote_reset_timeout(
raft_set_current_term(r, 1);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_periodic(r, 900);
raft_periodic_internal(r, 900);
memset(&rv, 0, sizeof(raft_requestvote_req_t));
rv.term = 2;
......@@ -1030,7 +1030,7 @@ void TestRaft_server_recv_requestvote_ignore_if_master_is_fresh(CuTest * tc)
CuAssertTrue(tc, 1 != rvr.vote_granted);
/* After election timeout passed, the same requestvote should be accepted */
raft_periodic(r, 1001);
raft_periodic_internal(r, 1001);
raft_recv_requestvote(r, raft_get_node(r, 3), &rv, &rvr);
CuAssertTrue(tc, 1 == rvr.vote_granted);
}
......@@ -1820,7 +1820,7 @@ void TestRaft_follower_becomes_precandidate_when_election_timeout_occurs(
raft_add_node(r, NULL, 2, 0);
/* max election timeout have passed */
raft_periodic(r, max_election_timeout(1000) + 1);
raft_periodic_internal(r, max_election_timeout(1000) + 1);
/* is a candidate now */
CuAssertTrue(tc, 1 == raft_is_precandidate(r));
......@@ -2057,7 +2057,7 @@ void TestRaft_follower_becoming_candidate_resets_election_timeout(CuTest * tc)
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
CuAssertTrue(tc, 0 == raft_get_timeout_elapsed(r));
raft_periodic(r, 900);
raft_periodic_internal(r, 900);
CuAssertTrue(tc, 900 == raft_get_timeout_elapsed(r));
raft_become_candidate(r);
......@@ -2080,7 +2080,7 @@ void TestRaft_follower_recv_appendentries_resets_election_timeout(
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 1);
raft_periodic(r, 900);
raft_periodic_internal(r, 900);
raft_appendentries_req_t ae;
raft_appendentries_resp_t aer;
......@@ -2148,7 +2148,7 @@ void TestRaft_candidate_election_timeout_and_no_leader_results_in_new_election(
CuAssertTrue(tc, 1 == raft_get_current_term(r));
/* clock over (ie. max election timeout + 1), does not increment term */
raft_periodic(r, max_election_timeout(1000) + 1);
raft_periodic_internal(r, max_election_timeout(1000) + 1);
CuAssertTrue(tc, 1 == raft_get_current_term(r));
CuAssertIntEquals(tc, RAFT_STATE_PRECANDIDATE, raft_get_state(r));
......@@ -2832,7 +2832,7 @@ void TestRaft_leader_recv_appendentries_response_increase_commit_idx_when_majori
raft_recv_appendentries_response(r, raft_get_node(r, 3), &aer);
/* leader will now have majority followers who have appended this log */
CuAssertIntEquals(tc, 1, raft_get_commit_idx(r));
raft_periodic(r, 1);
raft_periodic_internal(r, 1);
CuAssertIntEquals(tc, 1, raft_get_last_applied_idx(r));
/* SECOND entry log application */
......@@ -2849,7 +2849,7 @@ void TestRaft_leader_recv_appendentries_response_increase_commit_idx_when_majori
raft_recv_appendentries_response(r, raft_get_node(r, 3), &aer);
/* leader will now have majority followers who have appended this log */
CuAssertIntEquals(tc, 2, raft_get_commit_idx(r));
raft_periodic(r, 1);
raft_periodic_internal(r, 1);
CuAssertIntEquals(tc, 2, raft_get_last_applied_idx(r));
}
......@@ -2987,7 +2987,7 @@ void TestRaft_leader_recv_appendentries_response_increase_commit_idx_using_votin
raft_recv_appendentries_response(r, raft_get_node(r, 2), &aer);
CuAssertIntEquals(tc, 1, raft_get_commit_idx(r));
/* leader will now have majority followers who have appended this log */
raft_periodic(r, 1);
raft_periodic_internal(r, 1);
CuAssertIntEquals(tc, 1, raft_get_last_applied_idx(r));
}
......@@ -3089,7 +3089,7 @@ void TestRaft_leader_recv_appendentries_response_do_not_increase_commit_idx_beca
CuAssertIntEquals(tc, 0, raft_get_commit_idx(r));
raft_recv_appendentries_response(r, raft_get_node(r, 3), &aer);
CuAssertIntEquals(tc, 0, raft_get_commit_idx(r));
raft_periodic(r, 1);
raft_periodic_internal(r, 1);
CuAssertIntEquals(tc, 0, raft_get_last_applied_idx(r));
/* SECOND entry log application */
......@@ -3105,7 +3105,7 @@ void TestRaft_leader_recv_appendentries_response_do_not_increase_commit_idx_beca
CuAssertIntEquals(tc, 0, raft_get_commit_idx(r));
raft_recv_appendentries_response(r, raft_get_node(r, 3), &aer);
CuAssertIntEquals(tc, 0, raft_get_commit_idx(r));
raft_periodic(r, 1);
raft_periodic_internal(r, 1);
CuAssertIntEquals(tc, 0, raft_get_last_applied_idx(r));
/* THIRD entry log application */
......@@ -3120,7 +3120,7 @@ void TestRaft_leader_recv_appendentries_response_do_not_increase_commit_idx_beca
CuAssertIntEquals(tc, 0, raft_get_commit_idx(r));
raft_recv_appendentries_response(r, raft_get_node(r, 3), &aer);
CuAssertIntEquals(tc, 3, raft_get_commit_idx(r));
raft_periodic(r, 1);
raft_periodic_internal(r, 1);
CuAssertIntEquals(tc, 3, raft_get_last_applied_idx(r));
}
......@@ -3262,7 +3262,7 @@ void TestRaft_leader_recv_entry_resets_election_timeout(
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_set_state(r, RAFT_STATE_LEADER);
raft_periodic(r, 900);
raft_periodic_internal(r, 900);
/* entry message */
raft_entry_req_t *mety = __MAKE_ENTRY(1, 1, "entry");
......@@ -3665,7 +3665,7 @@ void TestRaft_leader_sends_empty_appendentries_every_request_timeout(
CuAssertTrue(tc, NULL == ae);
/* force request timeout */
raft_periodic(r, 501);
raft_periodic_internal(r, 501);
ae = sender_poll_msg_data(sender);
CuAssertTrue(tc, NULL != ae);
}
......@@ -3911,18 +3911,18 @@ void TestRaft_read_action_callback(
raft_recv_read_request(r, __read_request_callback, &ra);
/* not acked yet */
raft_periodic(r, 1);
raft_periodic_internal(r, 1);
CuAssertIntEquals(tc, 0, ra.calls);
/* acked by node 2 - enough for quorum */
raft_appendentries_resp_t aer = { .msg_id = 1, .term = 1, .success = 1, .current_idx = 2 };
CuAssertIntEquals(tc, 0, raft_recv_appendentries_response(r, raft_get_node(r, 2), &aer));
raft_periodic(r, 1);
raft_periodic_internal(r, 1);
CuAssertIntEquals(tc, 1, ra.calls);
CuAssertIntEquals(tc, 1, ra.last_cb_safe);
/* make sure read request is called only once */
raft_periodic(r, 1);
raft_periodic_internal(r, 1);
CuAssertIntEquals(tc, 1, ra.calls);
/* entry 2 */
__RAFT_APPEND_ENTRY(r, 2, 1, "aaa");
......@@ -3931,7 +3931,7 @@ void TestRaft_read_action_callback(
/* election started, nothing should be read */
raft_become_candidate(r);
raft_periodic(r, 1);
raft_periodic_internal(r, 1);
CuAssertIntEquals(tc, 0, ra.last_cb_safe);
CuAssertIntEquals(tc, 1, ra.calls);
......@@ -3941,7 +3941,7 @@ void TestRaft_read_action_callback(
aer.msg_id = 2;
aer.term = 3;
CuAssertIntEquals(tc, 0, raft_recv_appendentries_response(r, raft_get_node(r, 2), &aer));
raft_periodic(r, 1);
raft_periodic_internal(r, 1);
CuAssertIntEquals(tc, 1, ra.calls);
CuAssertIntEquals(tc, 0, ra.last_cb_safe);
/* entry 3 */
......@@ -3959,7 +3959,7 @@ void TestRaft_read_action_callback(
raft_become_candidate(r);
raft_become_follower(r);
/* queued read should fire back with can_read==false */
raft_periodic(r, 1);
raft_periodic_internal(r, 1);
CuAssertIntEquals(tc, 1, ra.calls);
CuAssertIntEquals(tc, 0, ra.last_cb_safe);
}
......@@ -3979,9 +3979,9 @@ void TestRaft_single_node_commits_noop(CuTest * tc)
raft_add_node(r, NULL, 1, 1);
raft_set_current_term(r, 2);
raft_set_commit_idx(r, 0);
raft_periodic(r, 500);
raft_periodic_internal(r, 500);
raft_recv_read_request(r, single_node_commits_noop_cb, &str);
raft_periodic(r, 500);
raft_periodic_internal(r, 500);
CuAssertIntEquals(tc, 1, raft_get_commit_idx(r));
CuAssertStrEquals(tc, "success", str);
......@@ -4070,37 +4070,37 @@ void TestRaft_quorum_msg_id_correctness(CuTest * tc)
__RAFT_APPEND_ENTRY(r, 1, 1, "aaa");
raft_set_commit_idx(r, 2);
raft_periodic(r, 100);
raft_periodic_internal(r, 100);
raft_recv_read_request(r, quorum_msg_id_correctness_cb, &val);
raft_periodic(r, 200);
raft_periodic_internal(r, 200);
// Read request is pending as it requires two acks
CuAssertIntEquals(tc, 0, val);
// Second node acknowledges reaq req
raft_node_set_match_msgid(raft_get_node(r, 2), 1);
raft_periodic(r, 200);
raft_periodic_internal(r, 200);
CuAssertIntEquals(tc, 1, val);
val = 0;
raft_add_node(r, NULL, 3, 0);
raft_add_node(r, NULL, 4, 0);
raft_periodic(r, 100);
raft_periodic_internal(r, 100);
raft_recv_read_request(r, quorum_msg_id_correctness_cb, &val);
raft_periodic(r, 200);
raft_periodic_internal(r, 200);
// Read request is pending as it requires three acks
CuAssertIntEquals(tc, 0, val);
// Second node acknowledges read req,
raft_node_set_match_msgid(raft_get_node(r, 2), 2);
raft_periodic(r, 200);
raft_periodic_internal(r, 200);
CuAssertIntEquals(tc, 0, val);
// Third node acknowledges read req
raft_node_set_match_msgid(raft_get_node(r, 3), 2);
raft_periodic(r, 200);
raft_periodic_internal(r, 200);
CuAssertIntEquals(tc, 1, val);
}
......@@ -4195,10 +4195,10 @@ void TestRaft_leader_steps_down_if_there_is_no_quorum(CuTest * tc)
raft_config(r, 1, RAFT_CONFIG_REQUEST_TIMEOUT, 1000);
// Quorum timeout is twice the election timeout.
int election_timeout;
raft_time_t election_timeout;
raft_config(r, 0, RAFT_CONFIG_ELECTION_TIMEOUT, &election_timeout);
int quorum_timeout = election_timeout * 2;
raft_time_t quorum_timeout = election_timeout * 2;
raft_become_leader(r);
......@@ -4206,33 +4206,33 @@ void TestRaft_leader_steps_down_if_there_is_no_quorum(CuTest * tc)
__RAFT_APPEND_ENTRY(r, 2, 1, "aaa");
raft_set_commit_idx(r, 2);
raft_periodic(r, 200);
raft_periodic_internal(r, 200);
CuAssertTrue(tc, raft_is_leader(r));
// Waiting more than quorum timeout will make leader step down.
raft_periodic(r, quorum_timeout + 1);
raft_periodic_internal(r, quorum_timeout + 1);
CuAssertTrue(tc, !raft_is_leader(r));
raft_node_set_match_msgid(raft_get_node(r, 2), 1);
raft_become_leader(r);
raft_periodic(r, 200);
raft_periodic_internal(r, 200);
CuAssertTrue(tc, raft_is_leader(r));
// Trigger new round of append entries
int request_timeout;
raft_time_t request_timeout;
raft_config(r, 0, RAFT_CONFIG_REQUEST_TIMEOUT, &request_timeout);
raft_periodic(r, request_timeout + 1);
raft_periodic_internal(r, request_timeout + 1);
CuAssertTrue(tc, raft_is_leader(r));
// If there is an ack from the follower, leader won't step down.
raft_node_set_match_msgid(raft_get_node(r, 2), 2);
raft_periodic(r, quorum_timeout);
raft_periodic_internal(r, quorum_timeout);
CuAssertTrue(tc, raft_is_leader(r));
// No ack along quorum_timeout, leader will step down.
raft_periodic(r, quorum_timeout + 1);
raft_periodic_internal(r, quorum_timeout + 1);
CuAssertTrue(tc, !raft_is_leader(r));
}
......@@ -4346,7 +4346,7 @@ void TestRaft_unknown_node_can_become_leader(CuTest * tc)
CuAssertTrue(tc, resp.success == 1);
// Validate added node is still the leader
raft_periodic(r, 1000);
raft_periodic_internal(r, 1000);
CuAssertIntEquals(tc, raft_get_leader_id(r), req_append.leader_id);
// Promote node from non-voting to voting
......@@ -4374,7 +4374,7 @@ void TestRaft_unknown_node_can_become_leader(CuTest * tc)
CuAssertIntEquals(tc, raft_get_leader_id(r), req_append.leader_id);
CuAssertTrue(tc, resp.success == 1);
raft_periodic(r, 1000);
raft_periodic_internal(r, 1000);
CuAssertIntEquals(tc, raft_get_leader_id(r), req_append.leader_id);
CuAssertTrue(tc, resp.success == 1);
......@@ -4413,7 +4413,7 @@ void TestRaft_removed_node_starts_election(CuTest * tc)
raft_become_follower(r);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_periodic(r, 2000);
raft_periodic_internal(r, 2000);
raft_become_candidate(r);
raft_requestvote_resp_t reqresp = {
......@@ -4504,7 +4504,7 @@ void Test_reset_transfer_leader(CuTest *tc)
r->leader_id = 1;
ret = raft_transfer_leader(r, 2, 1);
CuAssertIntEquals(tc, 0, ret);
raft_periodic(r, 2);
raft_periodic_internal(r, 2);
CuAssertIntEquals(tc, RAFT_LEADER_TRANSFER_TIMEOUT, state);
}
......@@ -4602,15 +4602,16 @@ void Test_transfer_automatic(CuTest *tc)
void TestRaft_config(CuTest *tc)
{
int val;
raft_time_t time;
raft_server_t *r = raft_new();
CuAssertIntEquals(tc, 0, raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 566));
CuAssertIntEquals(tc, 0, raft_config(r, 0, RAFT_CONFIG_ELECTION_TIMEOUT, &val));
CuAssertIntEquals(tc, 566, val);
CuAssertIntEquals(tc, 0, raft_config(r, 0, RAFT_CONFIG_ELECTION_TIMEOUT, &time));
CuAssertIntEquals(tc, 566, time);
CuAssertIntEquals(tc, 0, raft_config(r, 1, RAFT_CONFIG_REQUEST_TIMEOUT, 755));
CuAssertIntEquals(tc, 0, raft_config(r, 0, RAFT_CONFIG_REQUEST_TIMEOUT, &val));
CuAssertIntEquals(tc, 755, val);
CuAssertIntEquals(tc, 0, raft_config(r, 0, RAFT_CONFIG_REQUEST_TIMEOUT, &time));
CuAssertIntEquals(tc, 755, time);
CuAssertIntEquals(tc, 0, raft_config(r, 1, RAFT_CONFIG_AUTO_FLUSH, 8218318312));
CuAssertIntEquals(tc, 0, raft_config(r, 0, RAFT_CONFIG_AUTO_FLUSH, &val));
......@@ -4775,6 +4776,88 @@ void TestRaft_recv_appendentries_does_not_change_next_idx(CuTest *tc)
CuAssertIntEquals(tc, 21, raft_node_get_next_msgid(node));
}
raft_time_t timestamp(raft_server_t *raft, void *user_data)
{
raft_time_t *time = user_data;
*time += 10000;
return *time;
}
void TestRaft_apply_entry_timeout(CuTest *tc)
{
raft_time_t ts = 0;
raft_cbs_t funcs = {
.timestamp = timestamp
};
void *r = raft_new();
raft_add_node(r, NULL, 1, 1);
raft_set_callbacks(r, &funcs, &ts);
raft_set_current_term(r, 1);
raft_time_t timeout = 100;
raft_config(r, 1, RAFT_CONFIG_REQUEST_TIMEOUT, timeout);
__RAFT_APPEND_ENTRIES_SEQ_ID(r, 21, 0, 1, "");
raft_set_commit_idx(r, 21);
raft_exec_operations(r);
CuAssertIntEquals(tc, 1, raft_pending_operations(r));
CuAssertIntEquals(tc, 10, raft_get_last_applied_idx(r));
raft_exec_operations(r);
CuAssertIntEquals(tc, 1, raft_pending_operations(r));
CuAssertIntEquals(tc, 20, raft_get_last_applied_idx(r));
raft_exec_operations(r);
CuAssertIntEquals(tc, 0, raft_pending_operations(r));
CuAssertIntEquals(tc, 21, raft_get_last_applied_idx(r));
raft_exec_operations(r);
CuAssertIntEquals(tc, 0, raft_pending_operations(r));
CuAssertIntEquals(tc, 21, raft_get_last_applied_idx(r));
}
void read_request(void *arg, int can_read)
{
int *count = arg;
*count -= 1;
}
void TestRaft_apply_read_request_timeout(CuTest *tc)
{
raft_time_t ts = 0;
raft_cbs_t funcs = {
.timestamp = timestamp
};
void *r = raft_new();
raft_add_node(r, NULL, 1, 1);
raft_set_callbacks(r, &funcs, &ts);
raft_become_leader(r);
raft_set_commit_idx(r, 1);
raft_apply_all(r);
raft_config(r, 1, RAFT_CONFIG_AUTO_FLUSH, 0);
raft_time_t timeout = 100;
raft_config(r, 1, RAFT_CONFIG_REQUEST_TIMEOUT, timeout);
int remaining = 20;
for (int i = 0; i < remaining; i++) {
raft_recv_read_request(r, read_request, &remaining);
}
raft_exec_operations(r);
CuAssertIntEquals(tc, 1, raft_pending_operations(r));
CuAssertIntEquals(tc, 10, remaining);
raft_exec_operations(r);
CuAssertIntEquals(tc, 0, raft_pending_operations(r));
CuAssertIntEquals(tc, 0, remaining);
}
int main(void)
{
CuString *output = CuStringNew();
......@@ -4923,6 +5006,8 @@ int main(void)
SUITE_ADD_TEST(suite, TestRaft_limit_appendentries_size);
SUITE_ADD_TEST(suite, TestRaft_flush_sends_msg);
SUITE_ADD_TEST(suite, TestRaft_recv_appendentries_does_not_change_next_idx);
SUITE_ADD_TEST(suite, TestRaft_apply_entry_timeout);
SUITE_ADD_TEST(suite, TestRaft_apply_read_request_timeout);
CuSuiteRun(suite);
CuSuiteDetails(suite, output);
printf("%s\n", output->buffer);
......
......@@ -353,7 +353,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc)
CuAssertIntEquals(tc, 2, raft_get_commit_idx(r));
CuAssertIntEquals(tc, 2, raft_get_last_applied_idx(r));
CuAssertIntEquals(tc, 1, raft_get_last_log_term(r));
CuAssertIntEquals(tc, 0, raft_periodic(r, 1000));
CuAssertIntEquals(tc, 0, raft_periodic_internal(r, 1000));
/* the above test returns correct term as didn't snapshot all entries, makes sure works if we snapshot all */
......@@ -363,7 +363,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc)
raft_recv_entry(r, ety, &cr);
raft_set_commit_idx(r, 4);
CuAssertIntEquals(tc, 0, raft_periodic(r, 1000));
CuAssertIntEquals(tc, 0, raft_periodic_internal(r, 1000));
CuAssertIntEquals(tc, 2, raft_get_log_count(r));
CuAssertIntEquals(tc, 4, raft_get_commit_idx(r));
CuAssertIntEquals(tc, 3, r->log_impl->first_idx(r->log));
......@@ -416,7 +416,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted2(CuTest * tc)
CuAssertIntEquals(tc, 1, raft_get_log_count(r));
CuAssertIntEquals(tc, 2, raft_get_commit_idx(r));
CuAssertIntEquals(tc, 2, raft_get_last_applied_idx(r));
CuAssertIntEquals(tc, 0, raft_periodic(r, 1000));
CuAssertIntEquals(tc, 0, raft_periodic_internal(r, 1000));
}
void TestRaft_joinee_needs_to_get_snapshot(CuTest * tc)
......@@ -479,7 +479,7 @@ void TestRaft_follower_load_from_snapshot(CuTest * tc)
CuAssertIntEquals(tc, 5, raft_get_commit_idx(r));
CuAssertIntEquals(tc, 5, raft_get_last_applied_idx(r));
CuAssertIntEquals(tc, 0, raft_periodic(r, 1000));
CuAssertIntEquals(tc, 0, raft_periodic_internal(r, 1000));
/* current idx means snapshot was unnecessary */
ety = __MAKE_ENTRY(2, 1, "entry");
......
......@@ -925,7 +925,7 @@ class RaftServer(object):
if self.network.random.randint(1, 100000) < self.network.compaction_rate:
self.do_compaction()
e = lib.raft_periodic(self.raft, msec)
e = lib.raft_periodic_internal(self.raft, msec)
if lib.RAFT_ERR_SHUTDOWN == e:
self.shutdown()
......@@ -1091,6 +1091,9 @@ class RaftServer(object):
logger.debug('{} loading snapshot'.format(self))
leader = find_leader()
if not leader:
return 0
leader_snapshot = leader.snapshot_buf
# Copy received snapshot as our snapshot and clear the temp buf
......
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