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

Add raft_config() function for configuration (#102)

Add raft_config() function for configuration
parent c4e2dab1
......@@ -13,17 +13,17 @@
#include "raft_types.h"
typedef enum {
RAFT_ERR_NOT_LEADER=-2,
RAFT_ERR_ONE_VOTING_CHANGE_ONLY=-3,
RAFT_ERR_SHUTDOWN=-4,
RAFT_ERR_NOMEM=-5,
RAFT_ERR_SNAPSHOT_IN_PROGRESS=-6,
RAFT_ERR_SNAPSHOT_ALREADY_LOADED=-7,
RAFT_ERR_INVALID_NODEID=-8,
RAFT_ERR_LEADER_TRANSFER_IN_PROGRESS=-9,
RAFT_ERR_DONE=-10,
RAFT_ERR_STALE_TERM=-11,
RAFT_ERR_LAST=-100,
RAFT_ERR_NOT_LEADER = -2,
RAFT_ERR_ONE_VOTING_CHANGE_ONLY = -3,
RAFT_ERR_SHUTDOWN = -4,
RAFT_ERR_NOMEM = -5,
RAFT_ERR_SNAPSHOT_IN_PROGRESS = -6,
RAFT_ERR_SNAPSHOT_ALREADY_LOADED = -7,
RAFT_ERR_INVALID_NODEID = -8,
RAFT_ERR_LEADER_TRANSFER_IN_PROGRESS = -9,
RAFT_ERR_DONE = -10,
RAFT_ERR_STALE_TERM = -11,
RAFT_ERR_NOTFOUND = -12
} raft_error_e;
typedef enum {
......@@ -44,8 +44,15 @@ typedef enum {
RAFT_LEADER_TRANSFER_EXPECTED_LEADER,
} raft_leader_transfer_e;
/** Allow entries to apply while taking a snapshot */
#define RAFT_SNAPSHOT_NONBLOCKING_APPLY 1
typedef enum {
RAFT_CONFIG_ELECTION_TIMEOUT = 1,
RAFT_CONFIG_REQUEST_TIMEOUT,
RAFT_CONFIG_AUTO_FLUSH,
RAFT_CONFIG_LOG_ENABLED,
RAFT_CONFIG_NONBLOCKING_APPLY,
RAFT_CONFIG_DISABLE_APPLY,
} raft_config_e;
#define RAFT_NODE_ID_NONE (-1)
typedef enum {
......@@ -946,20 +953,6 @@ raft_node_t* raft_add_non_voting_node(raft_server_t* me, void* udata, raft_node_
* @param node The node to be removed. */
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
* @param[in] msec Election timeout in milliseconds */
void raft_set_election_timeout(raft_server_t* me, int msec);
/** Set request timeout in milliseconds.
* The amount of time before we resend an appendentries message
* @param[in] msec Request timeout in milliseconds */
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);
/** Process events that are dependent on time passing.
* @param[in] msec_elapsed Time in milliseconds since the last call
* @return
......@@ -1075,10 +1068,6 @@ int raft_get_nodeid(raft_server_t* me);
* @return the server's node */
raft_node_t* raft_get_my_node(raft_server_t *me);
/**
* @return currently configured election timeout in milliseconds */
int raft_get_election_timeout(raft_server_t* me);
/**
* @return number of nodes that this server has */
int raft_get_num_nodes(raft_server_t* me);
......@@ -1123,10 +1112,6 @@ int raft_is_candidate(raft_server_t* me);
* @return currently elapsed timeout in milliseconds */
int raft_get_timeout_elapsed(raft_server_t* me);
/**
* @return request timeout in milliseconds */
int raft_get_request_timeout(raft_server_t* me);
/**
* @return index of last applied entry */
raft_index_t raft_get_last_applied_idx(raft_server_t* me);
......@@ -1299,14 +1284,14 @@ int raft_entry_is_cfg_change(raft_entry_t* ety);
* - not apply log entries
* - not start elections
*
* If the RAFT_SNAPSHOT_NONBLOCKING_APPLY flag is specified, log entries will
* be applied during snapshot. The FSM must isolate the snapshot state and
* If RAFT_CONFIG_NONBLOCKING_APPLY config is set via raft_config(), log entries
* will be applied during snapshot. The FSM must isolate the snapshot state and
* guarantee these changes do not affect it.
*
* @return 0 on success
*
**/
int raft_begin_snapshot(raft_server_t *me, int flags);
int raft_begin_snapshot(raft_server_t *me);
/** Stop snapshotting.
*
......@@ -1538,21 +1523,29 @@ int raft_timeout_now(raft_server_t* me);
raft_index_t raft_get_num_snapshottable_logs(raft_server_t* me);
/** Disable auto flush mode. Default is enabled.
*
* In auto flush mode, after each raft_recv_entry() call, raft_log_impl_t's
* sync() is called to verify entry is persisted. Also, appendentries messages
* are sent for the entry immediately. It's easy to use library in this mode but
* to achieve better performance, we need batching. We can write entries to disk
* in another thread and send a single appendentries message for multiple
* entries. To do that, we disable auto flush mode. Once we do that, library
* user must check the newest log index by calling raft_get_index_to_sync() and
* verify new entries upto that index is written to the disk, probably in
* another thread. Also, users should call raft_flush() often to update
* persisted log index and to send new appendentries messages.
/**
* Library can be used in two modes:
*
* - Auto flush enabled: This is the default mode. In auto flush mode, after
* each raft_recv_entry() call, raft_log_impl_t's sync() is called to verify
* entry is persisted. Also, appendentries messages are sent for the new entry
* immediately. Easier to use the library in this mode but performance will be
* limited as there is no batching.
*
* - Auto flush disabled: To achieve better performance, we need batching.
* We can write entries to disk in another thread and send a single
* appendentries message for multiple entries. To do that, we disable auto flush
* mode. Once we do that, the library user must check the newest log index by
* calling raft_get_index_to_sync() and verify new entries up to that index are
* written to the disk, probably in another thread. Also, users should call
* raft_flush() often to update persisted log index and send new appendentries
* messages.
*
* Example:
*
* To disable auto flush mode:
* raft_config(raft, 1, RAFT_CONFIG_AUTO_FLUSH, 0);
*
* void server_loop() {
* while (1) {
* HandleNetworkOperations();
......@@ -1575,13 +1568,17 @@ raft_index_t raft_get_num_snapshottable_logs(raft_server_t* me);
* }
* }
*
* raft_flush() is no-op if node is follower.
* -------------------------------------------------------------------------
* raft_flush():
* - Updates persisted index and commit index.
* - Applies entries.
* - Sends messages(e.g appendentries) to the followers.
*
* @param[in] raft The Raft server
* @param[in] flush 1 to enable, 0 to disable
* @param[in] sync_index Entry index of the last persisted entry. '0' to skip
* updating persisted index.
* @return 0 on success
*/
int raft_set_auto_flush(raft_server_t* me, int flush);
int raft_flush(raft_server_t* me, raft_index_t sync_index);
/** Returns the latest entry index that needs to be written to the disk.
*
......@@ -1598,15 +1595,41 @@ int raft_set_auto_flush(raft_server_t* me, int flush);
*/
raft_index_t raft_get_index_to_sync(raft_server_t *me);
/** Update persisted index, send messages(e.g appendentries) to the followers.
/** Set or get config
*
* raft_flush() is no-op if node is follower.
* Valid configurations:
*
* @param[in] raft The Raft server
* @param[in] sync_index Entry index of the last persisted entry. '0' to skip
* updating persisted index.
* @return 0 on success
* election-timeout : Amount of time before node assumes leader is down.
* request-timeout : Heartbeat interval.
* auto-flush : See raft_flush().
* log-enabled : Enable / disable library logs.
* non-blocking-apply : See raft_begin_snapshot().
* 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 |
*
* Example:
*
* - Set
* raft_config(raft, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 4000);
*
* - Get
* int 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.
* @param config Config enum.
* @param ... Value to set or destination variable to get the config.
* @return 0 on success, RAFT_ERR_NOTFOUND if config is missing.
*/
int raft_flush(raft_server_t* me, raft_index_t sync_index);
int raft_config(raft_server_t *me, int set, raft_config_e config, ...);
#endif /* RAFT_H_ */
......@@ -57,10 +57,6 @@ struct raft_server {
raft_node_t** nodes;
int num_nodes;
int election_timeout;
int election_timeout_rand;
int request_timeout;
/* timer interval to check if we still have quorum */
long quorum_timeout;
......@@ -82,7 +78,6 @@ struct raft_server {
raft_index_t voting_cfg_change_log_idx;
int snapshot_in_progress;
int snapshot_flags;
/* Last compacted snapshot */
raft_index_t snapshot_last_idx;
......@@ -110,18 +105,21 @@ struct raft_server {
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
/* If this config is off (equals zero), user must call raft_flush()
* manually. It will trigger sending appendreqs, applying entries etc.
* Useful for batching, e.g after many raft_recv_entry() calls,
* one raft_flush() call will trigger sending appendreq for the latest
* entries. */
int auto_flush;
/* Index of the log entry that need to be written to the disk. Only useful
* when auto flush is disabled. */
raft_index_t next_sync_index;
int log_enabled;
int election_timeout_rand;
/* Configuration parameters */
int election_timeout; /* Timeout for a follower to start an election */
int 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 */
int disable_apply; /* Do not apply entries, useful for testing */
};
int raft_election_start(raft_server_t* me, int skip_precandidate);
......
......@@ -1633,7 +1633,7 @@ raft_index_t raft_get_num_snapshottable_logs(raft_server_t *me)
return raft_get_commit_idx(me) - me->log_impl->first_idx(me->log) + 1;
}
int raft_begin_snapshot(raft_server_t *me, int flags)
int raft_begin_snapshot(raft_server_t *me)
{
if (raft_get_num_snapshottable_logs(me) == 0)
return -1;
......@@ -1658,7 +1658,6 @@ int raft_begin_snapshot(raft_server_t *me, int flags)
me->snapshot_in_progress = 1;
me->next_snapshot_last_idx = snapshot_target;
me->next_snapshot_last_term = ety_term;
me->snapshot_flags = flags;
raft_log(me,
"begin snapshot sli:%ld slt:%ld slogs:%ld",
......@@ -2091,12 +2090,6 @@ raft_index_t raft_get_index_to_sync(raft_server_t *me)
return idx;
}
int raft_set_auto_flush(raft_server_t* me, int flush)
{
me->auto_flush = flush ? 1 : 0;
return 0;
}
int raft_flush(raft_server_t* me, raft_index_t sync_index)
{
if (!raft_is_leader(me)) {
......@@ -2132,3 +2125,65 @@ out:
raft_process_read_queue(me);
return 0;
}
int raft_config(raft_server_t *me, int set, raft_config_e config, ...)
{
int ret = 0;
va_list va;
va_start(va, config);
switch (config) {
case RAFT_CONFIG_ELECTION_TIMEOUT:
if (set) {
me->election_timeout = va_arg(va, int);
raft_update_quorum_meta(me, me->last_acked_msg_id);
raft_randomize_election_timeout(me);
} else {
*(va_arg(va, int*)) = me->election_timeout;
}
break;
case RAFT_CONFIG_REQUEST_TIMEOUT:
if (set) {
me->request_timeout = va_arg(va, int);
} else {
*(va_arg(va, int*)) = me->request_timeout;
}
break;
case RAFT_CONFIG_AUTO_FLUSH:
if (set) {
me->auto_flush = (va_arg(va, int)) ? 1 : 0;
} else {
*(va_arg(va, int*)) = me->auto_flush;
}
break;
case RAFT_CONFIG_LOG_ENABLED:
if (set) {
me->log_enabled = (va_arg(va, int)) ? 1 : 0;
} else {
*(va_arg(va, int*)) = me->log_enabled;
}
break;
case RAFT_CONFIG_NONBLOCKING_APPLY:
if (set) {
me->nonblocking_apply = (va_arg(va, int)) ? 1 : 0;
} else {
*(va_arg(va, int*)) = me->nonblocking_apply;
}
break;
case RAFT_CONFIG_DISABLE_APPLY:
if (set) {
me->disable_apply = (va_arg(va, int)) ? 1 : 0;
} else {
*(va_arg(va, int*)) = me->disable_apply;
}
break;
default:
ret = RAFT_ERR_NOTFOUND;
break;
}
va_end(va);
return ret;
}
......@@ -13,39 +13,11 @@
#include "raft.h"
#include "raft_private.h"
void raft_set_election_timeout(raft_server_t* me, int millisec)
{
me->election_timeout = millisec;
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)
{
me->request_timeout = millisec;
}
void raft_set_log_enabled(raft_server_t* me, int enable)
{
me->log_enabled = enable;
}
raft_node_id_t raft_get_nodeid(raft_server_t* me)
{
return raft_node_get_id(me->node);
}
int raft_get_election_timeout(raft_server_t* me)
{
return me->election_timeout;
}
int raft_get_request_timeout(raft_server_t* me)
{
return me->request_timeout;
}
int raft_get_num_nodes(raft_server_t* me)
{
return me->num_nodes;
......@@ -239,8 +211,8 @@ int raft_snapshot_is_in_progress(raft_server_t *me)
int raft_is_apply_allowed(raft_server_t* me)
{
return (!raft_snapshot_is_in_progress(me) ||
(me->snapshot_flags & RAFT_SNAPSHOT_NONBLOCKING_APPLY));
return !me->disable_apply &&
(!raft_snapshot_is_in_progress(me) || me->nonblocking_apply);
}
raft_entry_t *raft_get_last_applied_entry(raft_server_t *me)
......
......@@ -45,7 +45,7 @@ void TestRaft_scenario_leader_appears(CuTest * tc)
{
r[j] = raft_new();
sender_set_raft(sender[j], r[j]);
raft_set_election_timeout(r[j], 500);
raft_config(r[j], 1, RAFT_CONFIG_ELECTION_TIMEOUT, 500);
raft_add_node(r[j], sender[0], 1, j==0);
raft_add_node(r[j], sender[1], 2, j==1);
raft_add_node(r[j], sender[2], 3, j==2);
......
......@@ -264,13 +264,21 @@ void TestRaft_server_starts_as_follower(CuTest * tc)
void TestRaft_server_starts_with_election_timeout_of_1000ms(CuTest * tc)
{
void *r = raft_new();
CuAssertTrue(tc, 1000 == raft_get_election_timeout(r));
int election_timeout;
raft_config(r, 0, RAFT_CONFIG_ELECTION_TIMEOUT, &election_timeout);
CuAssertTrue(tc, 1000 == election_timeout);
}
void TestRaft_server_starts_with_request_timeout_of_200ms(CuTest * tc)
{
void *r = raft_new();
CuAssertTrue(tc, 200 == raft_get_request_timeout(r));
int request_timeout;
raft_config(r, 0, RAFT_CONFIG_REQUEST_TIMEOUT, &request_timeout);
CuAssertTrue(tc, 200 == request_timeout);
}
void TestRaft_server_entry_append_increases_logidx(CuTest* tc)
......@@ -480,7 +488,7 @@ void TestRaft_server_periodic_elapses_election_timeout(CuTest * tc)
{
void *r = raft_new();
/* we don't want to set the timeout to zero */
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
CuAssertTrue(tc, 0 == raft_get_timeout_elapsed(r));
raft_periodic(r, 0);
......@@ -503,7 +511,7 @@ void TestRaft_server_election_timeout_does_not_promote_us_to_leader_if_there_is_
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
/* clock over (ie. 1000 + 1), causing new election */
raft_periodic(r, 1001);
......@@ -515,7 +523,7 @@ void TestRaft_server_election_timeout_does_not_promote_us_to_leader_if_we_are_no
{
void *r = raft_new();
raft_add_non_voting_node(r, NULL, 1, 1);
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
/* clock over (ie. 1000 + 1), causing new election */
raft_periodic(r, 1001);
......@@ -529,7 +537,7 @@ void TestRaft_server_election_timeout_does_not_start_election_if_there_are_no_vo
void *r = raft_new();
raft_add_non_voting_node(r, NULL, 1, 1);
raft_add_non_voting_node(r, NULL, 2, 0);
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
/* clock over (ie. 1000 + 1), causing new election */
raft_periodic(r, 1001);
......@@ -541,7 +549,7 @@ void TestRaft_server_election_timeout_does_promote_us_to_leader_if_there_is_only
{
void *r = raft_new();
raft_add_node(r, NULL, 1, 1);
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_term_t old_term = raft_get_current_term(r);
/* clock over (ie. 1000 + 1), causing new election */
......@@ -562,7 +570,7 @@ void TestRaft_server_election_timeout_does_promote_us_to_leader_if_there_is_only
raft_add_node(r, NULL, 1, 1);
raft_add_non_voting_node(r, NULL, 2, 0);
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
/* clock over (ie. 1000 + 1), causing new election */
raft_periodic(r, 1001);
......@@ -574,7 +582,7 @@ void TestRaft_server_recv_entry_auto_commits_if_we_are_the_only_node(CuTest * tc
{
void *r = raft_new();
raft_add_node(r, NULL, 1, 1);
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_become_leader(r);
CuAssertTrue(tc, 0 == raft_get_commit_idx(r));
......@@ -591,9 +599,9 @@ void TestRaft_server_recv_entry_auto_commits_if_we_are_the_only_node(CuTest * tc
void TestRaft_server_recv_entry_fails_if_there_is_already_a_voting_change(CuTest * tc)
{
void *r = raft_new();
raft_set_auto_flush(r, 0);
raft_add_node(r, NULL, 1, 1);
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_AUTO_FLUSH, 0);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_become_leader(r);
CuAssertTrue(tc, 0 == raft_get_commit_idx(r));
......@@ -874,7 +882,7 @@ void TestRaft_server_recv_requestvote_reset_timeout(
raft_add_node(r, NULL, 2, 0);
raft_set_current_term(r, 1);
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_periodic(r, 900);
memset(&rv, 0, sizeof(raft_requestvote_req_t));
......@@ -1001,7 +1009,7 @@ void TestRaft_server_recv_requestvote_ignore_if_master_is_fresh(CuTest * tc)
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_set_current_term(r, 1);
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_appendentries_req_t ae = { 0 };
raft_appendentries_resp_t aer;
......@@ -1034,7 +1042,7 @@ void TestRaft_server_recv_prevote_ignore_if_candidate(CuTest * tc)
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_set_current_term(r, 1);
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_become_candidate(r);
......@@ -1057,7 +1065,7 @@ void TestRaft_server_recv_reqvote_ignore_if_not_candidate(CuTest * tc)
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_set_current_term(r, 1);
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_become_precandidate(r);
......@@ -1079,7 +1087,7 @@ void TestRaft_server_recv_reqvote_always_update_term(CuTest * tc)
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_set_current_term(r, 1);
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_become_precandidate(r);
......@@ -1806,7 +1814,7 @@ void TestRaft_follower_becomes_precandidate_when_election_timeout_occurs(
raft_set_callbacks(r, &funcs, NULL);
/* 1 second election timeout */
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
......@@ -2046,7 +2054,7 @@ void TestRaft_follower_becoming_candidate_resets_election_timeout(CuTest * tc)
raft_add_node(r, NULL, 2, 0);
raft_set_callbacks(r, &funcs, NULL);
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
CuAssertTrue(tc, 0 == raft_get_timeout_elapsed(r));
raft_periodic(r, 900);
......@@ -2068,7 +2076,7 @@ void TestRaft_follower_recv_appendentries_resets_election_timeout(
void *r = raft_new();
raft_set_callbacks(r, &funcs, NULL);
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 1);
......@@ -2133,7 +2141,7 @@ void TestRaft_candidate_election_timeout_and_no_leader_results_in_new_election(
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
/* server wants to be leader, so becomes candidate */
raft_become_candidate(r);
......@@ -3247,7 +3255,7 @@ void TestRaft_leader_recv_entry_resets_election_timeout(
{
void *r = raft_new();
raft_add_node(r, NULL, 1, 1);
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_set_state(r, RAFT_STATE_LEADER);
raft_periodic(r, 900);
......@@ -3634,8 +3642,8 @@ void TestRaft_leader_sends_empty_appendentries_every_request_timeout(
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_add_node(r, NULL, 3, 0);
raft_set_election_timeout(r, 1000);
raft_set_request_timeout(r, 500);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_config(r, 1, RAFT_CONFIG_REQUEST_TIMEOUT, 500);
CuAssertTrue(tc, 0 == raft_get_timeout_elapsed(r));
/* candidate to leader */
......@@ -3678,8 +3686,8 @@ void TestRaft_leader_recv_requestvote_responds_without_granting(CuTest * tc)
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_add_node(r, NULL, 3, 0);
raft_set_election_timeout(r, 1000);
raft_set_request_timeout(r, 500);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_config(r, 1, RAFT_CONFIG_REQUEST_TIMEOUT, 500);
CuAssertTrue(tc, 0 == raft_get_timeout_elapsed(r));
raft_election_start(r, 0);
......@@ -3731,8 +3739,8 @@ void T_estRaft_leader_recv_requestvote_responds_with_granting_if_term_is_higher(
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_add_node(r, NULL, 3, 0);
raft_set_election_timeout(r, 1000);
raft_set_request_timeout(r, 500);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_config(r, 1, RAFT_CONFIG_REQUEST_TIMEOUT, 500);
CuAssertTrue(tc, 0 == raft_get_timeout_elapsed(r));
raft_election_start(r);
......@@ -3893,7 +3901,7 @@ void TestRaft_read_action_callback(
raft_add_node(r, NULL, 3, 0);
raft_set_current_term(r, 1);
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_become_leader(r);
__RAFT_APPEND_ENTRY(r, 1, 1, "aaa");
......@@ -3994,7 +4002,7 @@ void TestRaft_server_recv_requestvote_with_transfer_node(CuTest * tc)
raft_add_node(r, NULL, 2, 0);
raft_add_node(r, NULL, 3, 0);
raft_set_current_term(r, 1);
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_set_state(r, RAFT_STATE_LEADER);
/* setup requestvote struct */
......@@ -4053,8 +4061,8 @@ void TestRaft_quorum_msg_id_correctness(CuTest * tc)
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_set_current_term(r, 1);
raft_set_election_timeout(r, 10000);
raft_set_request_timeout(r, 10000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 10000);
raft_config(r, 1, RAFT_CONFIG_REQUEST_TIMEOUT, 10000);
raft_become_leader(r);
__RAFT_APPEND_ENTRY(r, 1, 1, "aaa");
......@@ -4179,11 +4187,14 @@ void TestRaft_leader_steps_down_if_there_is_no_quorum(CuTest * tc)
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_set_current_term(r, 1);
raft_set_election_timeout(r, 1000);
raft_set_request_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_config(r, 1, RAFT_CONFIG_REQUEST_TIMEOUT, 1000);
// Quorum timeout is twice the election timeout.
int quorum_timeout = raft_get_election_timeout(r) * 2;
int election_timeout;
raft_config(r, 0, RAFT_CONFIG_ELECTION_TIMEOUT, &election_timeout);
int quorum_timeout = election_timeout * 2;
raft_become_leader(r);
......@@ -4204,7 +4215,11 @@ void TestRaft_leader_steps_down_if_there_is_no_quorum(CuTest * tc)
CuAssertTrue(tc, raft_is_leader(r));
// Trigger new round of append entries
raft_periodic(r, raft_get_request_timeout(r) + 1);
int request_timeout;
raft_config(r, 0, RAFT_CONFIG_REQUEST_TIMEOUT, &request_timeout);
raft_periodic(r, request_timeout + 1);
CuAssertTrue(tc, raft_is_leader(r));
// If there is an ack from the follower, leader won't step down.
......@@ -4393,7 +4408,7 @@ void TestRaft_removed_node_starts_election(CuTest * tc)
CuAssertIntEquals(r, raft_get_num_nodes(r), 2);
raft_become_follower(r);
raft_set_election_timeout(r, 1000);
raft_config(r, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000);
raft_periodic(r, 2000);
raft_become_candidate(r);
......@@ -4580,6 +4595,44 @@ void Test_transfer_automatic(CuTest *tc)
CuAssertIntEquals(tc, 3, raft_get_transfer_leader(r));
}
void TestRaft_config(CuTest *tc)
{
int val;
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, 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, 1, RAFT_CONFIG_AUTO_FLUSH, 8218318312));
CuAssertIntEquals(tc, 0, raft_config(r, 0, RAFT_CONFIG_AUTO_FLUSH, &val));
CuAssertIntEquals(tc, 1, val);
CuAssertIntEquals(tc, 0, raft_config(r, 1, RAFT_CONFIG_LOG_ENABLED, 1));
CuAssertIntEquals(tc, 0, raft_config(r, 0, RAFT_CONFIG_LOG_ENABLED, &val));
CuAssertIntEquals(tc, 1, val);
CuAssertIntEquals(tc, 0, raft_config(r, 1, RAFT_CONFIG_NONBLOCKING_APPLY, 1));
CuAssertIntEquals(tc, 0, raft_config(r, 0, RAFT_CONFIG_NONBLOCKING_APPLY, &val));
CuAssertIntEquals(tc, 1, val);
CuAssertIntEquals(tc, 0, raft_config(r, 1, RAFT_CONFIG_DISABLE_APPLY, 1));
CuAssertIntEquals(tc, 0, raft_config(r, 0, RAFT_CONFIG_DISABLE_APPLY, &val));
CuAssertIntEquals(tc, 1, val);
CuAssertIntEquals(tc, RAFT_ERR_NOTFOUND, raft_config(r, 1, -1, 1));
CuAssertIntEquals(tc, RAFT_ERR_NOTFOUND, raft_config(r, 0, -1, &val));
long long tmp;
CuAssertIntEquals(tc, 0, raft_config(r, 1, RAFT_CONFIG_DISABLE_APPLY, 1));
CuAssertIntEquals(tc, 0, raft_config(r, 0, RAFT_CONFIG_DISABLE_APPLY, &tmp));
CuAssertIntEquals(tc, 1, val);
}
int main(void)
{
CuString *output = CuStringNew();
......@@ -4724,6 +4777,7 @@ int main(void)
SUITE_ADD_TEST(suite, Test_transfer_leader_unexpected);
SUITE_ADD_TEST(suite, Test_transfer_leader_not_leader);
SUITE_ADD_TEST(suite, Test_transfer_automatic);
SUITE_ADD_TEST(suite, TestRaft_config);
CuSuiteRun(suite);
CuSuiteDetails(suite, output);
printf("%s\n", output->buffer);
......
......@@ -226,10 +226,10 @@ void TestRaft_leader_begin_snapshot_fails_if_no_logs_to_compact(CuTest * tc)
ety = __MAKE_ENTRY(2, 1, "entry");
raft_recv_entry(r, ety, &cr);
CuAssertIntEquals(tc, 2, raft_get_log_count(r));
CuAssertIntEquals(tc, -1, raft_begin_snapshot(r, 0));
CuAssertIntEquals(tc, -1, raft_begin_snapshot(r));
raft_set_commit_idx(r, 1);
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r));
}
void TestRaft_leader_will_not_apply_entry_if_snapshot_is_in_progress(CuTest * tc)
......@@ -259,7 +259,7 @@ void TestRaft_leader_will_not_apply_entry_if_snapshot_is_in_progress(CuTest * tc
raft_set_commit_idx(r, 1);
CuAssertIntEquals(tc, 2, raft_get_log_count(r));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r));
CuAssertIntEquals(tc, 1, raft_get_last_applied_idx(r));
raft_set_commit_idx(r, 2);
CuAssertIntEquals(tc, -1, raft_apply_entry(r));
......@@ -309,7 +309,7 @@ void TestRaft_leader_snapshot_begin_fails_if_less_than_2_logs_to_compact(CuTest
raft_recv_entry(r, ety, &cr);
raft_set_commit_idx(r, 1);
CuAssertIntEquals(tc, 1, raft_get_log_count(r));
CuAssertIntEquals(tc, -1, raft_begin_snapshot(r, 0));
CuAssertIntEquals(tc, -1, raft_begin_snapshot(r));
}
void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc)
......@@ -346,7 +346,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc)
CuAssertIntEquals(tc, 3, raft_get_log_count(r));
CuAssertIntEquals(tc, 2, raft_get_num_snapshottable_logs(r));
CuAssertIntEquals(tc, 1, raft_get_last_log_term(r));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r));
CuAssertIntEquals(tc, 0, raft_end_snapshot(r));
CuAssertIntEquals(tc, 0, raft_get_num_snapshottable_logs(r));
CuAssertIntEquals(tc, 1, raft_get_log_count(r));
......@@ -368,7 +368,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc)
CuAssertIntEquals(tc, 4, raft_get_commit_idx(r));
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_begin_snapshot(r));
CuAssertIntEquals(tc, 0, raft_end_snapshot(r));
CuAssertIntEquals(tc, 0, raft_get_num_snapshottable_logs(r));
CuAssertIntEquals(tc, 0, raft_get_log_count(r));
......@@ -410,7 +410,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted2(CuTest * tc)
CuAssertIntEquals(tc, 3, raft_get_log_count(r));
CuAssertIntEquals(tc, 2, raft_get_num_snapshottable_logs(r));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r));
CuAssertIntEquals(tc, 0, raft_end_snapshot(r));
CuAssertIntEquals(tc, 0, raft_get_num_snapshottable_logs(r));
CuAssertIntEquals(tc, 1, raft_get_log_count(r));
......@@ -448,7 +448,7 @@ void TestRaft_joinee_needs_to_get_snapshot(CuTest * tc)
CuAssertIntEquals(tc, 2, raft_get_log_count(r));
CuAssertIntEquals(tc, 1, raft_get_num_snapshottable_logs(r));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r));
CuAssertIntEquals(tc, 1, raft_get_last_applied_idx(r));
CuAssertIntEquals(tc, -1, raft_apply_entry(r));
CuAssertIntEquals(tc, 1, raft_get_last_applied_idx(r));
......@@ -663,7 +663,7 @@ void TestRaft_recv_entry_fails_if_snapshot_in_progress(CuTest* tc)
CuAssertIntEquals(tc, 2, raft_get_log_count(r));
raft_set_commit_idx(r, 1);
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r));
ety = __MAKE_ENTRY(3, 1, "entry");
ety->type = RAFT_LOGTYPE_ADD_NODE;
......@@ -698,7 +698,9 @@ void TestRaft_recv_entry_succeeds_if_snapshot_nonblocking_apply_is_set(CuTest* t
CuAssertIntEquals(tc, 2, raft_get_log_count(r));
raft_set_commit_idx(r, 1);
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, RAFT_SNAPSHOT_NONBLOCKING_APPLY));
raft_config(r, 1, RAFT_CONFIG_NONBLOCKING_APPLY, 1);
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r));
ety = __MAKE_ENTRY(3, 1, "entry");
ety->type = RAFT_LOGTYPE_ADD_NODE;
......@@ -797,7 +799,7 @@ void TestRaft_cancel_snapshot_restores_state(CuTest* tc)
raft_recv_entry(r, ety, &cr);
raft_set_commit_idx(r, 2);
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r));
CuAssertIntEquals(tc, 0, raft_end_snapshot(r));
/* more entries */
......@@ -810,7 +812,7 @@ void TestRaft_cancel_snapshot_restores_state(CuTest* tc)
/* begin and cancel another snapshot */
raft_set_commit_idx(r, 4);
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r));
CuAssertIntEquals(tc, 1, raft_snapshot_is_in_progress(r));
CuAssertIntEquals(tc, 0, raft_cancel_snapshot(r));
......@@ -855,7 +857,7 @@ void TestRaft_leader_sends_snapshot_if_log_was_compacted(CuTest* tc)
/* compact entry 1 & 2 */
raft_set_commit_idx(r, 2);
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r));
CuAssertIntEquals(tc, 0, raft_end_snapshot(r));
CuAssertIntEquals(tc, 1, raft_get_log_count(r));
......
......@@ -848,9 +848,13 @@ class RaftServer(object):
lib.raft_set_callbacks(self.raft, cbs, self.udata)
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)
lib.raft_set_log_enabled(self.raft, 1)
lib.raft_config(self.raft, 1, lib.RAFT_CONFIG_ELECTION_TIMEOUT,
ffi.cast("int", 500))
lib.raft_config(self.raft, 1, lib.RAFT_CONFIG_LOG_ENABLED,
ffi.cast("int", 1))
lib.raft_config(self.raft, 1, lib.RAFT_CONFIG_AUTO_FLUSH,
ffi.cast("int", network.auto_flush))
self.fsm_dict = {}
self.fsm_log = []
......@@ -891,7 +895,7 @@ class RaftServer(object):
# logger.warning('{} snapshotting'.format(self))
# entries_before = lib.raft_get_log_count(self.raft)
e = lib.raft_begin_snapshot(self.raft, 0)
e = lib.raft_begin_snapshot(self.raft)
if e != 0:
return
......
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