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
Willemt Raft
Commits
aba00838
Commit
aba00838
authored
Oct 24, 2013
by
willem
Browse files
Semantics: command -> entry
parent
79033d25
Changes
5
Hide whitespace changes
Inline
Side-by-side
raft.h
View file @
aba00838
...
...
@@ -21,33 +21,33 @@ typedef struct {
/* candidateId candidate requesting vote */
int
candidate_id
;
/* i
nde
x of candidate's last log entry */
int
last_log_i
nde
x
;
/* i
d
x of candidate's last log entry */
int
last_log_i
d
x
;
/* term of candidate's last log entry */
int
last_log_term
;
}
msg_requestvote_t
;
/**
* every server must send
command
s in this format */
* every server must send
entry
s in this format */
typedef
struct
{
unsigned
int
id
;
unsigned
char
*
data
;
unsigned
int
len
;
}
msg_
command
_t
;
}
msg_
entry
_t
;
typedef
struct
{
unsigned
int
term
;
unsigned
int
id
;
unsigned
char
*
data
;
unsigned
int
len
;
}
raft_
command
_t
;
}
raft_
entry
_t
;
typedef
struct
{
unsigned
int
id
;
int
was_committed
;
}
msg_
command
_response_t
;
}
msg_
entry
_response_t
;
typedef
struct
{
/* currentTerm, for candidate to update itself */
...
...
@@ -60,10 +60,10 @@ typedef struct {
typedef
struct
{
int
term
;
int
leader_id
;
int
prev_log_i
nde
x
;
int
prev_log_i
d
x
;
int
prev_log_term
;
int
n_entries
;
msg_
command
_t
*
entries
;
msg_
entry
_t
*
entries
;
int
leader_commit
;
}
msg_appendentries_t
;
...
...
@@ -72,7 +72,7 @@ typedef struct {
int
term
;
/* success true if follower contained entry matching
* prevLog
Inde
x and prevLogTerm */
* prevLog
id
x and prevLogTerm */
int
success
;
}
msg_appendentries_response_t
;
...
...
@@ -145,7 +145,7 @@ int raft_recv_requestvote(raft_server_t* me, int peer, msg_requestvote_t* vr);
int
raft_recv_requestvote_response
(
raft_server_t
*
me
,
int
peer
,
msg_requestvote_response_t
*
r
);
void
raft_execute_
command
(
raft_server_t
*
me
);
void
raft_execute_
entry
(
raft_server_t
*
me
);
void
raft_set_election_timeout
(
raft_server_t
*
me
,
int
millisec
);
...
...
@@ -159,19 +159,19 @@ int raft_remove_peer(raft_server_t* me, int peer);
int
raft_get_num_peers
(
raft_server_t
*
me
);
int
raft_recv_
command
(
raft_server_t
*
me
,
int
peer
,
msg_
command
_t
*
cmd
);
int
raft_recv_
entry
(
raft_server_t
*
me
,
int
peer
,
msg_
entry
_t
*
cmd
);
int
raft_get_log_count
(
raft_server_t
*
me
);
void
raft_set_current_term
(
raft_server_t
*
me
,
int
term
);
void
raft_set_current_i
nde
x
(
raft_server_t
*
me
,
int
idx
);
void
raft_set_current_i
d
x
(
raft_server_t
*
me
,
int
idx
);
int
raft_get_current_term
(
raft_server_t
*
me
);
void
raft_set_current_i
nde
x
(
raft_server_t
*
me
,
int
idx
);
void
raft_set_current_i
d
x
(
raft_server_t
*
me
,
int
idx
);
int
raft_get_current_i
nde
x
(
raft_server_t
*
me
);
int
raft_get_current_i
d
x
(
raft_server_t
*
me
);
int
raft_is_follower
(
raft_server_t
*
me
);
...
...
@@ -183,30 +183,30 @@ int raft_send_requestvote(raft_server_t* me, int peer);
void
raft_send_appendentries
(
raft_server_t
*
me
,
int
peer
);
int
raft_append_command
(
raft_server_t
*
me_
,
raft_command_t
*
c
);
void
raft_commit_command
(
raft_server_t
*
me
,
int
logIndex
);
int
raft_append_entry
(
raft_server_t
*
me_
,
raft_entry_t
*
c
);
int
raft_get_timeout_elapsed
(
raft_server_t
*
me
);
void
raft_set_commit_i
nde
x
(
raft_server_t
*
me
,
int
commit_idx
);
void
raft_set_commit_i
d
x
(
raft_server_t
*
me
,
int
commit_idx
);
void
raft_set_last_applied_i
nde
x
(
raft_server_t
*
me
,
int
idx
);
void
raft_set_last_applied_i
d
x
(
raft_server_t
*
me
,
int
idx
);
void
raft_set_state
(
raft_server_t
*
me_
,
int
state
);
void
raft_set_request_timeout
(
raft_server_t
*
me_
,
int
millisec
);
int
raft_get_last_applied_i
nde
x
(
raft_server_t
*
me
);
int
raft_get_last_applied_i
d
x
(
raft_server_t
*
me
);
raft_peer_t
*
raft_peer_new
(
void
*
udata
);
int
raft_peer_is_leader
(
raft_peer_t
*
peer
);
int
raft_peer_get_next_i
nde
x
(
raft_peer_t
*
peer
);
int
raft_peer_get_next_i
d
x
(
raft_peer_t
*
peer
);
void
raft_peer_set_next_i
nde
x
(
raft_peer_t
*
peer
,
int
nextIdx
);
void
raft_peer_set_next_i
d
x
(
raft_peer_t
*
peer
,
int
nextIdx
);
void
raft_set_configuration
(
raft_server_t
*
me_
,
raft_peer_configuration_t
*
peers
);
int
raft_votes_is_majority
(
const
int
npeers
,
const
int
nvotes
);
void
raft_apply_entry
(
raft_server_t
*
me_
);
raft_log.c
View file @
aba00838
...
...
@@ -5,7 +5,7 @@
* found in the LICENSE file.
*
* @file
* @brief ADT for managing Raft log entries (aka
command
s)
* @brief ADT for managing Raft log entries (aka
entry
s)
* @author Willem Thiart himself@willemthiart.com
* @version 0.1
*/
...
...
@@ -22,7 +22,8 @@
typedef
struct
{
arrayqueue_t
*
log
;
arrayqueue_t
*
log_qu
;
hashmap_t
*
log_map
;
}
raft_log_private_t
;
static
unsigned
long
__entry_hash
(
...
...
@@ -45,19 +46,41 @@ raft_log_t* raft_log_new()
raft_log_private_t
*
me
;
me
=
calloc
(
1
,
sizeof
(
raft_log_private_t
));
me
->
log
=
arrayqueue_new
();
//
me->log_map = hashmap_new(__entry_hash, __entry_compare, 100);
me
->
log
_qu
=
arrayqueue_new
();
me
->
log_map
=
hashmap_new
(
__entry_hash
,
__entry_compare
,
100
);
return
(
void
*
)
me
;
}
void
raft_log_append_entry
(
raft_log_t
*
me_
,
raft_command_t
*
c
)
/**
* Add entry to log.
* Don't add entry if we've already added this entry (based off ID)
* Don't add entries with ID=0
* @return 0 if unsucessful; 1 otherwise */
int
raft_log_append_entry
(
raft_log_t
*
me_
,
raft_entry_t
*
c
)
{
raft_log_private_t
*
me
=
(
void
*
)
me_
;
arrayqueue_offer
(
me
->
log
,
c
);
if
(
0
==
c
->
id
)
return
0
;
if
(
hashmap_get
(
me
->
log_map
,
(
void
*
)
c
->
id
))
return
0
;
arrayqueue_offer
(
me
->
log_qu
,
c
);
hashmap_put
(
me
->
log_map
,
(
void
*
)
c
->
id
,
c
);
assert
(
arrayqueue_count
(
me
->
log_qu
)
==
hashmap_count
(
me
->
log_map
));
return
1
;
}
raft_entry_t
*
raft_log_get_from_index
(
raft_log_t
*
me_
,
int
idx
)
{
raft_log_private_t
*
me
=
(
void
*
)
me_
;
return
hashmap_get
(
me
->
log_map
,
(
void
*
)
idx
);
}
int
raft_log_count
(
raft_log_t
*
me_
)
{
raft_log_private_t
*
me
=
(
void
*
)
me_
;
return
arrayqueue_count
(
me
->
log
);
return
arrayqueue_count
(
me
->
log
_qu
);
}
raft_log.h
View file @
aba00838
...
...
@@ -3,6 +3,8 @@ typedef void* raft_log_t;
raft_log_t
*
raft_log_new
();
void
raft_log_append_entry
(
raft_log_t
*
me_
,
raft_
command
_t
*
c
);
int
raft_log_append_entry
(
raft_log_t
*
me_
,
raft_
entry
_t
*
c
);
int
raft_log_count
(
raft_log_t
*
me_
);
raft_entry_t
*
raft_log_get_from_idx
(
raft_log_t
*
me_
,
int
idx
);
raft_server.c
View file @
aba00838
...
...
@@ -36,11 +36,11 @@ typedef struct {
/* Volatile state: */
/*
Inde
x of highest log entry known to be committed */
int
commit_i
nde
x
;
/*
id
x of highest log entry known to be committed */
int
commit_i
d
x
;
/*
Inde
x of highest log entry applied to state machine */
int
last_applied_i
nde
x
;
/*
id
x of highest log entry applied to state machine */
int
last_applied_i
d
x
;
/* follower/leader/candidate indicator */
int
state
;
...
...
@@ -52,7 +52,7 @@ typedef struct {
void
*
caller
;
/* indicates size of log */
int
current_i
nde
x
;
int
current_i
d
x
;
/* amount of time left till timeout */
int
timeout_elapased
;
...
...
@@ -94,7 +94,7 @@ raft_server_t* raft_new()
if
(
!
(
me
=
calloc
(
1
,
sizeof
(
raft_server_private_t
))))
return
NULL
;
me
->
voted_for
=
-
1
;
me
->
current_i
nde
x
=
0
;
me
->
current_i
d
x
=
0
;
me
->
timeout_elapased
=
0
;
me
->
log
=
raft_log_new
();
raft_set_state
((
void
*
)
me
,
RAFT_STATE_FOLLOWER
);
...
...
@@ -167,7 +167,7 @@ void raft_become_candidate(raft_server_t* me_)
rv
.
term
=
me
->
current_term
;
rv
.
candidate_id
=
0
;
//me->current_term;
// rv.last_log_i
nde
x = me->log
// rv.last_log_i
d
x = me->log
if
(
me
->
ext_func
&&
me
->
ext_func
->
send
)
me
->
ext_func
->
send
(
me
->
caller
,
NULL
,
ii
,
(
void
*
)
&
rv
,
sizeof
(
msg_requestvote_t
));
...
...
@@ -207,6 +207,11 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period)
{
raft_server_private_t
*
me
=
(
void
*
)
me_
;
if
(
raft_get_last_applied_idx
(
me_
)
<
raft_get_commit_idx
(
me_
))
{
raft_apply_entry
(
me_
);
}
me
->
timeout_elapased
+=
msec_since_last_period
;
if
(
me
->
election_timeout
<
me
->
timeout_elapased
)
...
...
@@ -228,7 +233,7 @@ int raft_recv_appendentries(raft_server_t* me_, const int peer, msg_appendentrie
if
(
ae
->
term
<
raft_get_current_term
(
me_
)
||
ae
->
prev_log_i
nde
x
<
raft_get_current_i
nde
x
(
me_
))
ae
->
prev_log_i
d
x
<
raft_get_current_i
d
x
(
me_
))
{
r
.
success
=
0
;
goto
done
;
...
...
@@ -248,12 +253,12 @@ int raft_recv_appendentries(raft_server_t* me_, const int peer, msg_appendentrie
for
(
i
=
0
;
i
<
ae
->
n_entries
;
i
++
)
{
msg_
command
_t
*
cmd
;
raft_
command
_t
*
c
;
msg_
entry
_t
*
cmd
;
raft_
entry
_t
*
c
;
cmd
=
&
ae
->
entries
[
i
];
c
=
malloc
(
sizeof
(
raft_
command
_t
));
c
=
malloc
(
sizeof
(
raft_
entry
_t
));
c
->
term
=
raft_get_current_term
(
me_
);
c
->
len
=
cmd
->
len
;
c
->
id
=
cmd
->
id
;
...
...
@@ -279,7 +284,7 @@ done:
return
0
;
}
raft_
command
_t
*
raft_get_
command
_from_i
nde
x
(
raft_server_t
*
me_
,
int
idx
)
raft_
entry
_t
*
raft_get_
entry
_from_i
d
x
(
raft_server_t
*
me_
,
int
idx
)
{
raft_server_private_t
*
me
=
(
void
*
)
me_
;
...
...
@@ -358,7 +363,7 @@ int raft_recv_requestvote_response(raft_server_t* me_, int peer, msg_requestvote
return
0
;
}
void
raft_execute_
command
(
raft_server_t
*
me_
)
void
raft_execute_
entry
(
raft_server_t
*
me_
)
{
}
...
...
@@ -419,7 +424,7 @@ int raft_get_num_peers(raft_server_t* me_)
return
0
;
}
int
raft_recv_
command
(
raft_server_t
*
me_
,
int
peer
,
msg_
command
_t
*
cmd
)
int
raft_recv_
entry
(
raft_server_t
*
me_
,
int
peer
,
msg_
entry
_t
*
cmd
)
{
return
0
;
}
...
...
@@ -450,16 +455,16 @@ int raft_get_current_term(raft_server_t* me_)
return
me
->
current_term
;
}
void
raft_set_current_i
nde
x
(
raft_server_t
*
me_
,
int
idx
)
void
raft_set_current_i
d
x
(
raft_server_t
*
me_
,
int
idx
)
{
raft_server_private_t
*
me
=
(
void
*
)
me_
;
me
->
current_i
nde
x
=
idx
;
me
->
current_i
d
x
=
idx
;
}
int
raft_get_current_i
nde
x
(
raft_server_t
*
me_
)
int
raft_get_current_i
d
x
(
raft_server_t
*
me_
)
{
raft_server_private_t
*
me
=
(
void
*
)
me_
;
return
me
->
current_i
nde
x
;
return
me
->
current_i
d
x
;
}
int
raft_is_follower
(
raft_server_t
*
me_
)
...
...
@@ -488,7 +493,7 @@ int raft_send_requestvote(raft_server_t* me_, int peer)
msg_requestvote_t
rv
;
rv
.
term
=
raft_get_current_term
(
me_
);
rv
.
last_log_i
nde
x
=
raft_get_current_i
nde
x
(
me_
);
rv
.
last_log_i
d
x
=
raft_get_current_i
d
x
(
me_
);
if
(
me
->
ext_func
&&
me
->
ext_func
->
send
)
me
->
ext_func
->
send
(
me
->
caller
,
NULL
,
peer
,
(
void
*
)
&
rv
,
sizeof
(
msg_requestvote_t
));
...
...
@@ -496,42 +501,60 @@ int raft_send_requestvote(raft_server_t* me_, int peer)
}
/**
* Appends
command
using the current term.
* Appends
entry
using the current term.
* Note: we make the assumption that current term is up-to-date
* @return 0 if unsuccessful */
int
raft_append_entry
(
raft_server_t
*
me_
,
raft_
command
_t
*
c
)
int
raft_append_entry
(
raft_server_t
*
me_
,
raft_
entry
_t
*
c
)
{
raft_server_private_t
*
me
=
(
void
*
)
me_
;
raft_log_append_entry
(
me
->
log
,
c
);
me
->
current_index
+=
1
;
return
1
;
if
(
1
==
raft_log_append_entry
(
me
->
log
,
c
))
{
me
->
current_idx
+=
1
;
return
1
;
}
else
{
return
0
;
}
}
void
raft_set_commit_i
nde
x
(
raft_server_t
*
me_
,
int
commit_
idx
)
void
raft_set_commit_i
d
x
(
raft_server_t
*
me_
,
int
idx
)
{
raft_server_private_t
*
me
=
(
void
*
)
me_
;
me
->
commit_idx
=
idx
;
}
void
raft_set_last_applied_i
nde
x
(
raft_server_t
*
me_
,
int
idx
)
void
raft_set_last_applied_i
d
x
(
raft_server_t
*
me_
,
int
idx
)
{
raft_server_private_t
*
me
=
(
void
*
)
me_
;
me
->
last_applied_i
nde
x
=
idx
;
me
->
last_applied_i
d
x
=
idx
;
}
int
raft_get_last_applied_i
nde
x
(
raft_server_t
*
me_
)
int
raft_get_last_applied_i
d
x
(
raft_server_t
*
me_
)
{
raft_server_private_t
*
me
=
(
void
*
)
me_
;
return
me
->
last_applied_i
nde
x
;
return
me
->
last_applied_i
d
x
;
}
int
raft_get_commit_i
nde
x
(
raft_server_t
*
me_
)
int
raft_get_commit_i
d
x
(
raft_server_t
*
me_
)
{
return
0
;
raft_server_private_t
*
me
=
(
void
*
)
me_
;
return
me
->
commit_idx
;
}
void
raft_commit_command
(
raft_server_t
*
me_
,
int
logIndex
)
/**
* Apply entry at lastApplied + 1 */
void
raft_apply_entry
(
raft_server_t
*
me_
)
{
raft_server_private_t
*
me
=
(
void
*
)
me_
;
if
(
me
->
last_applied_idx
==
raft_get_commit_idx
(
me_
))
return
;
me
->
last_applied_idx
++
;
// TODO: callback for state machine application goes here
}
void
raft_send_appendentries
(
raft_server_t
*
me_
,
int
peer
)
...
...
test_server.c
View file @
aba00838
...
...
@@ -26,7 +26,7 @@ void TestRaft_server_idx_starts_at_1(CuTest * tc)
void
*
r
;
r
=
raft_new
();
CuAssertTrue
(
tc
,
1
==
raft_get_current_i
nde
x
(
r
));
CuAssertTrue
(
tc
,
1
==
raft_get_current_i
d
x
(
r
));
}
void
TestRaft_server_set_currentterm_sets_term
(
CuTest
*
tc
)
...
...
@@ -117,102 +117,162 @@ void TestRaft_server_starts_with_request_timeout_of_500ms(CuTest * tc)
CuAssertTrue
(
tc
,
500
==
raft_get_request_timeout
(
r
));
}
void
TestRaft_server_
command_append_increases_logindex
(
CuTest
*
tc
)
void
TestRaft_server_
entry_append_cant_append_if_id_is_zero
(
CuTest
*
tc
)
{
void
*
r
;
raft_
command_t
cmd
;
raft_
entry_t
ety
;
char
*
str
=
"aaa"
;
cmd
.
data
=
str
;
cmd
.
len
=
3
;
cmd
.
id
=
0
;
cmd
.
term
=
1
;
ety
.
data
=
str
;
ety
.
len
=
3
;
ety
.
id
=
0
;
ety
.
term
=
1
;
r
=
raft_new
();
CuAssertTrue
(
tc
,
0
==
raft_get_current_i
nde
x
(
r
));
raft_append_entry
(
r
,
&
cmd
);
CuAssertTrue
(
tc
,
1
==
raft_get_current_i
nde
x
(
r
));
CuAssertTrue
(
tc
,
0
==
raft_get_current_i
d
x
(
r
));
raft_append_entry
(
r
,
&
ety
);
CuAssertTrue
(
tc
,
0
==
raft_get_current_i
d
x
(
r
));
}
void
TestRaft_server_append_
command_means_command_gets_current_term
(
CuTest
*
tc
)
void
TestRaft_server_
entry_
append_
increases_logidx
(
CuTest
*
tc
)
{
void
*
r
;
raft_
command_t
cmd
;
raft_
entry_t
ety
;
char
*
str
=
"aaa"
;
cmd
.
data
=
str
;
cmd
.
len
=
3
;
cmd
.
id
=
0
;
cmd
.
term
=
1
;
ety
.
data
=
str
;
ety
.
len
=
3
;
ety
.
id
=
1
;
ety
.
term
=
1
;
r
=
raft_new
();
CuAssertTrue
(
tc
,
0
==
raft_get_current_i
nde
x
(
r
));
raft_append_entry
(
r
,
&
cmd
);
CuAssertTrue
(
tc
,
1
==
raft_get_current_i
nde
x
(
r
));
CuAssertTrue
(
tc
,
0
==
raft_get_current_i
d
x
(
r
));
raft_append_entry
(
r
,
&
ety
);
CuAssertTrue
(
tc
,
1
==
raft_get_current_i
d
x
(
r
));
}
void
TestRaft_server_append_
command_not_sucessful_if_command_with_id_already_appended
(
CuTest
*
tc
)
void
TestRaft_server_append_
entry_means_entry_gets_current_term
(
CuTest
*
tc
)
{
void
*
r
;
raft_
command_t
cmd
;
raft_
entry_t
ety
;
char
*
str
=
"aaa"
;
cmd
.
data
=
str
;
cmd
.
len
=
3
;
cmd
.
id
=
0
;
cmd
.
term
=
1
;
ety
.
data
=
str
;
ety
.
len
=
3
;
ety
.
id
=
1
;
ety
.
term
=
1
;
r
=
raft_new
();
CuAssertTrue
(
tc
,
0
==
raft_get_current_index
(
r
));
raft_append_entry
(
r
,
&
cmd
);
raft_append_entry
(
r
,
&
cmd
);
CuAssertTrue
(
tc
,
1
==
raft_get_current_index
(
r
));
CuAssertTrue
(
tc
,
0
==
raft_get_current_idx
(
r
));
raft_append_entry
(
r
,
&
ety
);
CuAssertTrue
(
tc
,
1
==
raft_get_current_idx
(
r
));
}
void
TestRaft_server_append_entry_not_sucessful_if_entry_with_id_already_appended
(
CuTest
*
tc
)
{
void
*
r
;
raft_entry_t
ety
;
char
*
str
=
"aaa"
;
ety
.
data
=
str
;
ety
.
len
=
3
;
ety
.
id
=
1
;
ety
.
term
=
1
;
r
=
raft_new
();
CuAssertTrue
(
tc
,
0
==
raft_get_current_idx
(
r
));
raft_append_entry
(
r
,
&
ety
);
raft_append_entry
(
r
,
&
ety
);
CuAssertTrue
(
tc
,
1
==
raft_get_current_idx
(
r
));
/* different ID so we can be successful */
cmd
.
id
=
1
;
raft_append_entry
(
r
,
&
cmd
);
CuAssertTrue
(
tc
,
2
==
raft_get_current_i
nde
x
(
r
));
ety
.
id
=
2
;
raft_append_entry
(
r
,
&
ety
);
CuAssertTrue
(
tc
,
2
==
raft_get_current_i
d
x
(
r
));
}
void
TestRaft_server_
command
_is_retrieveable_using_i
nde
x
(
CuTest
*
tc
)
void
TestRaft_server_
entry
_is_retrieveable_using_i
d
x
(
CuTest
*
tc
)
{
void
*
r
;
raft_
command_t
cmd
;
raft_
command
_t
*
cmd
_appended
;
raft_
entry_t
ety
;
raft_
entry
_t
*
ety
_appended
;
char
*
str
=
"aaa"
;
char
*
str2
=
"bbb"
;
cmd
.
term
=
1
;
ety
.
term
=
1
;
r
=
raft_new
();
cmd
.
id
=
0
;
cmd
.
data
=
str
;
cmd
.
len
=
3
;
raft_append_entry
(
r
,
&
cmd
);
ety
.
id
=
1
;
ety
.
data
=
str
;
ety
.
len
=
3
;
raft_append_entry
(
r
,
&
ety
);
/* different ID so we can be successful */
cmd
.
id
=
1
;
cmd
.
data
=
str2
;
cmd
.
len
=
3
;
raft_append_entry
(
r
,
&
cmd
);
//CuAssertTrue(tc, (cmd_appended = raft_get_command_from_index(r,1)));
//CuAssertTrue(tc, !strncmp(cmd_appended->data,str2,3));
ety
.
id
=
2
;
ety
.
data
=
str2
;
ety
.
len
=
3
;
raft_append_entry
(
r
,
&
ety
);
// ety_appended = raft_log_get_from_idx(r, int idx)
//CuAssertTrue(tc, (ety_appended = raft_get_entry_from_idx(r,1)));
//CuAssertTrue(tc, !strncmp(ety_appended->data,str2,3));
}
// If commit
Inde
x > lastApplied: increment lastApplied, apply
// If commit
id
x > lastApplied: increment lastApplied, apply
// log[lastApplied] to state machine (5.3)
void
TestRaft_server_increment_lastApplied_when_lastApplied_lt_commit
Inde
x
(
CuTest
*
tc
)
void
TestRaft_server_increment_lastApplied_when_lastApplied_lt_commit
id
x
(
CuTest
*
tc
)
{
void
*
r
;
r
=
raft_new
();
raft_set_commit_i
nde
x
(
r
,
5
);
raft_set_last_applied_i
nde
x
(
r
,
4
);
raft_set_commit_i
d
x
(
r
,
5
);
raft_set_last_applied_i
d
x
(
r
,
4
);
raft_periodic
(
r
,
1
);
CuAssertTrue
(
tc
,
5
==
raft_get_last_applied_index
(
r
));
CuAssertTrue
(
tc
,
5
==
raft_get_last_applied_idx
(
r
));
}
void
TestRaft_server_apply_entry_increments_last_applied_idx
(
CuTest
*
tc
)
{
void
*
r
;
raft_entry_t
ety
;
raft_entry_t
*
ety_appended
;
char
*
str
=
"aaa"
;
ety
.
term
=
1
;
r
=
raft_new
();
raft_set_commit_idx
(
r
,
5
);
raft_set_last_applied_idx
(
r
,
4
);
ety
.
id
=
1
;
ety
.
data
=
str
;
ety
.
len
=
3
;
raft_append_entry
(
r
,
&
ety
);
raft_apply_entry
(
r
);
CuAssertTrue
(
tc
,
5
==
raft_get_last_applied_idx
(
r
));
}
void
TestRaft_server_apply_entry_wont_apply_if_lastapplied_equalto_commit_index
(
CuTest
*
tc
)
{
void
*
r
;
raft_entry_t
ety
;
raft_entry_t
*
ety_appended
;
char
*
str
=
"aaa"
;
ety
.
term
=
1
;
r
=
raft_new
();
raft_set_commit_idx
(
r
,
5
);
raft_set_last_applied_idx
(
r
,
5
);
ety
.
id
=
1
;
ety
.
data
=
str
;
ety
.
len
=
3
;
raft_append_entry
(
r
,
&
ety
);
raft_apply_entry
(
r
);
CuAssertTrue
(
tc
,
5
==
raft_get_last_applied_idx
(
r
));
}
void
TestRaft_server_periodic_elapses_election_timeout
(
CuTest
*
tc
)
...
...
@@ -539,7 +599,7 @@ void TestRaft_follower_increases_log_after_appendentry(CuTest * tc)
}
/* 5.3 */
void
TestRaft_follower_recv_appendentries_reply_false_if_doesnt_have_log_at_prev_log_i
nde
x_which_matches_prev_log_term
(
CuTest
*
tc
)
void
TestRaft_follower_recv_appendentries_reply_false_if_doesnt_have_log_at_prev_log_i
d
x_which_matches_prev_log_term
(
CuTest
*
tc
)
{
void
*
r
;
void
*
sender
;
...
...
@@ -562,10 +622,10 @@ void TestRaft_follower_recv_appendentries_reply_false_if_doesnt_have_log_at_prev
raft_set_configuration
(
r
,
cfg
);
raft_set_external_functions
(
r
,
&
funcs
,
sender
);
/* log i
nde
x that server doesn't have */
/* log i
d
x that server doesn't have */
memset
(
&
ae
,
0
,
sizeof
(
msg_appendentries_t
));
ae
.
term
=
5
;
ae
.
prev_log_i
nde
x
=
5
;
ae
.
prev_log_i
d
x
=
5
;
/* current term is old */
raft_set_current_term
(
r
,
5
);
...
...
@@ -599,18 +659,18 @@ void TestRaft_follower_recv_appendentries_delete_entries_if_conflict_with_new_en
/* first append entry */
memset
(
&
ae
,
0
,
sizeof
(
msg_appendentries_t
));
ae
.
term
=
1
;
ae
.
prev_log_i
nde
x
=
0
;
ae
.
prev_log_i
d
x
=
0
;
ae
.
prev_log_term
=
0
;
raft_
command_t
cmd
;
raft_
entry_t
ety
;
char
*
str
=
"111"
;
/* increase log size */
cmd
.
data
=
str
;
cmd
.
len
=
3
;
cmd
.
id
=
0
;
cmd
.
term
=
1
;
raft_append_entry
(
r
,
&
cmd
);
ety
.
data
=
str
;
ety
.
len
=
3
;
ety
.
id
=
1
;
ety
.
term
=
1
;
raft_append_entry
(
r
,
&
ety
);
CuAssertTrue
(
tc
,
1
==
raft_get_log_count
(
r
));
/* pass a appendentry that is newer */
...
...
@@ -658,9 +718,9 @@ void TestRaft_follower_recv_appendentries_add_new_entries_not_already_in_log(CuT
// CuAssertTrue(tc, 1 == sender_msg_is_false(msg));
}
//If leaderCommit > commit
Inde
x, set commit
Inde
x =
//min(leaderCommit, last log i
nde
x)
void
TestRaft_follower_recv_appendentries_set_commiti
nde
x_to_prevLogIdx
(
CuTest
*
tc
)
//If leaderCommit > commit
id
x, set commit
id
x =
//min(leaderCommit, last log i
d
x)
void
TestRaft_follower_recv_appendentries_set_commiti
d
x_to_prevLogIdx
(
CuTest
*
tc
)
{
void
*
r
;
void
*
sender
;
...
...
@@ -685,16 +745,16 @@ void TestRaft_follower_recv_appendentries_set_commitindex_to_prevLogIdx(CuTest *
memset
(
&
ae
,
0
,
sizeof
(
msg_appendentries_t
));
ae
.
term
=
1
;
ae
.
prev_log_term
=
1
;
ae
.
prev_log_i
nde
x
=
4
;
ae
.
prev_log_i
d
x
=
4
;
ae
.
leader_commit
=
5
;
/* receipt of appendentries changes commit i
nde
x */
/* receipt of appendentries changes commit i
d
x */
raft_recv_appendentries
(
r
,
1
,
&
ae
);
/* set to 4 because prevLogIdx is lower */
CuAssertTrue
(
tc
,
4
==
raft_get_commit_i
nde
x
(
r
));
CuAssertTrue
(
tc
,
4
==
raft_get_commit_i
d
x
(
r
));
}
void
TestRaft_follower_recv_appendentries_set_commiti
nde
x_to_LeaderCommit
(
CuTest
*
tc
)
void
TestRaft_follower_recv_appendentries_set_commiti
d
x_to_LeaderCommit
(
CuTest
*
tc
)
{
void
*
r
;
void
*
sender
;
...
...
@@ -720,13 +780,13 @@ void TestRaft_follower_recv_appendentries_set_commitindex_to_LeaderCommit(CuTest
memset
(
&
ae
,
0
,
sizeof
(
msg_appendentries_t
));
ae
.
term
=
1
;
ae
.
prev_log_term
=
1
;
ae
.
prev_log_i
nde
x
=
4
;
ae
.
prev_log_i
d
x
=
4
;
ae
.
leader_commit
=
3
;
/* receipt of appendentries changes commit i
nde
x */
/* receipt of appendentries changes commit i
d
x */
raft_recv_appendentries
(
r
,
1
,
&
ae
);
/* set to 3 because leaderCommit is lower */
CuAssertTrue
(
tc
,
3
==
raft_get_commit_i
nde
x
(
r
));
CuAssertTrue
(
tc
,
3
==
raft_get_commit_i
d
x
(
r
));
}
void
TestRaft_follower_rejects_appendentries_if_idx_and_term_dont_match_preceding_ones
(
CuTest
*
tc
)
...
...
@@ -750,7 +810,7 @@ void TestRaft_follower_rejects_appendentries_if_idx_and_term_dont_match_precedin
/* first append entry */
memset
(
&
ae
,
0
,
sizeof
(
msg_appendentries_t
));
ae
.
term
=
1
;
ae
.
prev_log_i
nde
x
=
0
;
ae
.
prev_log_i
d
x
=
0
;
ae
.
prev_log_term
=
0
;
CuAssertTrue
(
tc
,
0
==
raft_get_log_count
(
r
));
...
...
@@ -760,7 +820,7 @@ void TestRaft_follower_rejects_appendentries_if_idx_and_term_dont_match_precedin
}
#if 0
void T_estRaft_follower_resends_
command
_if_request_from_leader_timesout(CuTest * tc)
void T_estRaft_follower_resends_
entry
_if_request_from_leader_timesout(CuTest * tc)
{
void *r;
...
...
@@ -840,12 +900,12 @@ void TestRaft_follower_dont_grant_vote_if_candidate_has_a_less_complete_log(CuTe
memset
(
&
rv
,
0
,
sizeof
(
msg_requestvote_t
));
rv
.
term
=
1
;
rv
.
candidate_id
=
0
;
rv
.
last_log_i
nde
x
=
1
;
rv
.
last_log_i
d
x
=
1
;
rv
.
last_log_term
=
1
;
/* server's term and i
nde
x are more up-to-date */
/* server's term and i
d
x are more up-to-date */
raft_set_current_term
(
r
,
1
);
raft_set_current_i
nde
x
(
r
,
2
);
raft_set_current_i
d
x
(
r
,
2
);
/* vote not granted */
raft_recv_requestvote
(
r
,
1
,
&
rv
);
...
...
@@ -929,7 +989,7 @@ void TestRaft_follower_becoming_candidate_requests_votes_from_other_servers(CuTe
/* set term so we can check it gets included in the outbound message */
raft_set_current_term
(
r
,
2
);
raft_set_current_i
nde
x
(
r
,
5
);
raft_set_current_i
d
x
(
r
,
5
);
/* becoming candidate triggers vote requests */
raft_become_candidate
(
r
);
...
...
@@ -1068,7 +1128,7 @@ void TestRaft_candidate_will_not_respond_to_voterequest_if_it_has_already_voted(
}
/* Candidate 5.2 */
void
TestRaft_candidate_requestvote_includes_log
Inde
x
(
CuTest
*
tc
)
void
TestRaft_candidate_requestvote_includes_log
id
x
(
CuTest
*
tc
)
{
void
*
r
;
void
*
sender
;
...
...
@@ -1091,12 +1151,12 @@ void TestRaft_candidate_requestvote_includes_logIndex(CuTest * tc)
raft_set_external_functions
(
r
,
&
funcs
,
sender
);
raft_set_current_term
(
r
,
5
);
raft_set_current_i
nde
x
(
r
,
3
);
raft_set_current_i
d
x
(
r
,
3
);
raft_send_requestvote
(
r
,
1
);
rv
=
sender_poll_msg
(
sender
);
CuAssertTrue
(
tc
,
NULL
!=
rv
);
CuAssertTrue
(
tc
,
3
==
rv
->
last_log_i
nde
x
);
CuAssertTrue
(
tc
,
3
==
rv
->
last_log_i
d
x
);
CuAssertTrue
(
tc
,
5
==
rv
->
term
);
}
...
...
@@ -1160,7 +1220,7 @@ void TestRaft_candidate_recv_appendentries_frm_invalid_leader_doesnt_result_in_f
/* server's log is newer */
raft_set_current_term
(
r
,
1
);
raft_set_current_i
nde
x
(
r
,
2
);
raft_set_current_i
d
x
(
r
,
2
);
/* is a candidate */
raft_set_state
(
r
,
RAFT_STATE_CANDIDATE
);
...
...
@@ -1169,7 +1229,7 @@ void TestRaft_candidate_recv_appendentries_frm_invalid_leader_doesnt_result_in_f
/* invalid leader determined by "leaders" old log */
memset
(
&
ae
,
0
,
sizeof
(
msg_appendentries_t
));
ae
.
term
=
1
;
ae
.
prev_log_i
nde
x
=
1
;
ae
.
prev_log_i
d
x
=
1
;
ae
.
prev_log_term
=
1
;
/* appendentry from invalid leader doesn't make candidate become follower */
...
...
@@ -1222,10 +1282,10 @@ void TestRaft_leader_when_it_becomes_a_leader_sends_empty_appendentries(CuTest *
}
/* 5.2 */
void
TestRaft_leader_responds_to_
command
_msg_when_
command
_is_committed
(
CuTest
*
tc
)
void
TestRaft_leader_responds_to_
entry
_msg_when_
entry
_is_committed
(
CuTest
*
tc
)
{
void
*
r
,
*
sender
;
msg_
command
_response_t
*
cr
;
msg_
entry
_response_t
*
cr
;
raft_external_functions_t
funcs
=
{
.
send
=
sender_send
,
.
log
=
NULL
...
...
@@ -1247,20 +1307,20 @@ void TestRaft_leader_responds_to_command_msg_when_command_is_committed(CuTest *
raft_set_state
(
r
,
RAFT_STATE_LEADER
);
CuAssertTrue
(
tc
,
0
==
raft_get_log_count
(
r
));
/*
command
message */
msg_
command_t
cmd
;
cmd
.
id
=
1
;
cmd
.
data
=
"
command
"
;
cmd
.
len
=
strlen
(
"
command
"
);
/*
entry
message */
msg_
entry_t
ety
;
ety
.
id
=
1
;
ety
.
data
=
"
entry
"
;
ety
.
len
=
strlen
(
"
entry
"
);
/* receive
command
*/
raft_recv_
command
(
r
,
1
,
&
cmd
);
/* receive
entry
*/
raft_recv_
entry
(
r
,
1
,
&
ety
);
CuAssertTrue
(
tc
,
1
==
raft_get_log_count
(
r
));
/* trigger response through commit */
raft_
commit_command
(
r
,
1
);
raft_
apply_entry
(
r
);
/* leader sent response to
command
message */
/* leader sent response to
entry
message */
cr
=
sender_poll_msg
(
sender
);
CuAssertTrue
(
tc
,
NULL
!=
cr
);
}
...
...
@@ -1332,12 +1392,12 @@ void TestRaft_leader_retries_appendentries_with_decremented_NextIdx_log_inconsis
}
/*
If there exists an N such that N > commit
Inde
x, a majority
of match
Inde
x[i] = N, and log[N].term == currentTerm:
set commit
Inde
x = N (5.2, 5.4).
If there exists an N such that N > commit
id
x, a majority
of match
id
x[i] = N, and log[N].term == currentTerm:
set commit
id
x = N (5.2, 5.4).
*/
void
TestRaft_leader_append_
command
_to_log_increases_idxno
(
CuTest
*
tc
)
void
TestRaft_leader_append_
entry
_to_log_increases_idxno
(
CuTest
*
tc
)
{
void
*
r
;
...
...
@@ -1348,21 +1408,21 @@ void TestRaft_leader_append_command_to_log_increases_idxno(CuTest * tc)
{(
-
1
),
NULL
}};
msg_
command_t
cmd
;
cmd
.
id
=
1
;
cmd
.
data
=
"
command
"
;
cmd
.
len
=
strlen
(
"
command
"
);
msg_
entry_t
ety
;
ety
.
id
=
1
;
ety
.
data
=
"
entry
"
;
ety
.
len
=
strlen
(
"
entry
"
);
r
=
raft_new
();
raft_set_configuration
(
r
,
cfg
);
raft_set_state
(
r
,
RAFT_STATE_LEADER
);
CuAssertTrue
(
tc
,
0
==
raft_get_log_count
(
r
));
raft_recv_
command
(
r
,
1
,
&
cmd
);
raft_recv_
entry
(
r
,
1
,
&
ety
);
CuAssertTrue
(
tc
,
1
==
raft_get_log_count
(
r
));
}
void
TestRaft_leader_doesnt_append_
command
_if_unique_id_is_duplicate
(
CuTest
*
tc
)
void
TestRaft_leader_doesnt_append_
entry
_if_unique_id_is_duplicate
(
CuTest
*
tc
)
{
void
*
r
;
...
...
@@ -1372,10 +1432,10 @@ void TestRaft_leader_doesnt_append_command_if_unique_id_is_duplicate(CuTest * tc
{(
-
1
),(
void
*
)
2
},
{(
-
1
),
NULL
}};
msg_
command_t
cmd
;
cmd
.
id
=
1
;
cmd
.
data
=
"
command
"
;
cmd
.
len
=
strlen
(
"
command
"
);
msg_
entry_t
ety
;
ety
.
id
=
1
;
ety
.
data
=
"
entry
"
;
ety
.
len
=
strlen
(
"
entry
"
);
r
=
raft_new
();
raft_set_configuration
(
r
,
cfg
);
...
...
@@ -1383,10 +1443,10 @@ void TestRaft_leader_doesnt_append_command_if_unique_id_is_duplicate(CuTest * tc
raft_set_state
(
r
,
RAFT_STATE_LEADER
);
CuAssertTrue
(
tc
,
0
==
raft_get_log_count
(
r
));
raft_recv_
command
(
r
,
1
,
&
cmd
);
raft_recv_
entry
(
r
,
1
,
&
ety
);
CuAssertTrue
(
tc
,
1
==
raft_get_log_count
(
r
));
raft_recv_
command
(
r
,
1
,
&
cmd
);
raft_recv_
entry
(
r
,
1
,
&
ety
);
CuAssertTrue
(
tc
,
1
==
raft_get_log_count
(
r
));
}
...
...
@@ -1413,10 +1473,10 @@ void TestRaft_leader_increase_commitno_when_majority_have_entry_and_atleast_one_
/* I'm the leader */
raft_set_state
(
r
,
RAFT_STATE_LEADER
);
/* the commit i
nde
x will became 5 */
/* the commit i
d
x will became 5 */
raft_set_current_term
(
r
,
5
);
raft_set_current_i
nde
x
(
r
,
5
);
raft_set_commit_i
nde
x
(
r
,
4
);
raft_set_current_i
d
x
(
r
,
5
);
raft_set_commit_i
d
x
(
r
,
4
);
raft_set_external_functions
(
r
,
&
funcs
,
sender
);
raft_send_appendentries
(
r
,
1
);
...
...
@@ -1430,7 +1490,7 @@ void TestRaft_leader_increase_commitno_when_majority_have_entry_and_atleast_one_
/* announce to leader that the majority have appended this log */
raft_recv_appendentries_response
(
r
,
1
,
&
aer
);
raft_recv_appendentries_response
(
r
,
2
,
&
aer
);
CuAssertTrue
(
tc
,
5
==
raft_get_commit_i
nde
x
(
r
));
CuAssertTrue
(
tc
,
5
==
raft_get_commit_i
d
x
(
r
));
}
void
TestRaft_leader_steps_down_if_received_appendentries_is_newer_than_itself
(
CuTest
*
tc
)
...
...
@@ -1452,7 +1512,7 @@ void TestRaft_leader_steps_down_if_received_appendentries_is_newer_than_itself(C
memset
(
&
ae
,
0
,
sizeof
(
msg_appendentries_t
));
ae
.
term
=
5
;
ae
.
prev_log_i
nde
x
=
6
;
ae
.
prev_log_i
d
x
=
6
;
ae
.
prev_log_term
=
5
;
sender
=
sender_new
();
...
...
@@ -1461,7 +1521,7 @@ void TestRaft_leader_steps_down_if_received_appendentries_is_newer_than_itself(C
raft_set_state
(
r
,
RAFT_STATE_LEADER
);
raft_set_current_term
(
r
,
5
);
raft_set_current_i
nde
x
(
r
,
5
);
raft_set_current_i
d
x
(
r
,
5
);
raft_set_external_functions
(
r
,
&
funcs
,
sender
);
raft_recv_appendentries
(
r
,
1
,
&
ae
);
...
...
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