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
4469906e
Commit
4469906e
authored
Dec 09, 2013
by
willem
Browse files
Improved raft_log implemenation
parent
aba00838
Changes
5
Hide whitespace changes
Inline
Side-by-side
raft.h
View file @
4469906e
...
...
@@ -127,7 +127,7 @@ raft_server_t* raft_new();
int
raft_get_voted_for
(
raft_server_t
*
me
);
void
raft_set_
external_function
s
(
raft_server_t
*
me
,
raft_external_functions_t
*
funcs
,
void
*
caller
);
void
raft_set_
callback
s
(
raft_server_t
*
me
,
raft_external_functions_t
*
funcs
,
void
*
caller
);
void
raft_election_start
(
raft_server_t
*
me
);
...
...
@@ -210,3 +210,5 @@ 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_entry_t
*
raft_get_entry_from_idx
(
raft_server_t
*
me_
,
int
idx
);
raft_log.c
View file @
4469906e
...
...
@@ -15,15 +15,28 @@
#include <stdio.h>
#include <assert.h>
#include "linked_list_hashmap.h"
//
#include "linked_list_hashmap.h"
#include "arrayqueue.h"
#include "raft.h"
#include "raft_log.h"
#define INITIAL_CAPACITY 10
#define in(x) ((raft_log_private_t*)x)
typedef
struct
{
arrayqueue_t
*
log_qu
;
hashmap_t
*
log_map
;
typedef
struct
{
/* size of array */
int
size
;
/* the amount of elements in the array */
int
count
;
/* position of the queue */
int
front
,
back
;
int
base_log_index
;
raft_entry_t
*
entries
;
}
raft_log_private_t
;
static
unsigned
long
__entry_hash
(
...
...
@@ -41,13 +54,43 @@ static long __entry_compare(
return
obj
-
other
;
}
static
void
__ensurecapacity
(
raft_log_private_t
*
me
)
{
int
i
,
j
;
raft_entry_t
*
temp
;
if
(
me
->
count
<
me
->
size
)
return
;
temp
=
calloc
(
1
,
sizeof
(
raft_entry_t
)
*
me
->
size
*
2
);
for
(
i
=
0
,
j
=
me
->
front
;
i
<
me
->
count
;
i
++
,
j
++
)
{
if
(
j
==
me
->
size
)
j
=
0
;
memcpy
(
&
temp
[
i
],
&
me
->
entries
[
j
],
sizeof
(
raft_entry_t
));
}
me
->
size
*=
2
;
me
->
entries
=
temp
;
me
->
front
=
0
;
me
->
back
=
me
->
count
;
/* clean up old entries */
free
(
me
->
entries
);
}
raft_log_t
*
raft_log_new
()
{
raft_log_private_t
*
me
;
me
=
calloc
(
1
,
sizeof
(
raft_log_private_t
));
me
->
log_qu
=
arrayqueue_new
();
me
->
log_map
=
hashmap_new
(
__entry_hash
,
__entry_compare
,
100
);
me
->
size
=
INITIAL_CAPACITY
;
me
->
count
=
0
;
me
->
back
=
in
(
me
)
->
front
=
0
;
me
->
entries
=
calloc
(
1
,
sizeof
(
raft_entry_t
)
*
me
->
size
);
return
(
void
*
)
me
;
}
...
...
@@ -63,24 +106,98 @@ int raft_log_append_entry(raft_log_t* me_, raft_entry_t* c)
if
(
0
==
c
->
id
)
return
0
;
if
(
hashmap_get
(
me
->
log_map
,
(
void
*
)
c
->
id
))
return
0
;
__ensurecapacity
(
me
);
// if (hashmap_get(me->log_map, (void*)c->id+1))
// 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
))
;
memcpy
(
&
me
->
entries
[
me
->
back
],
c
,
sizeof
(
raft_entry_t
)
);
me
->
count
++
;
me
->
back
++
;
return
1
;
}
raft_entry_t
*
raft_log_get_from_i
nde
x
(
raft_log_t
*
me_
,
int
idx
)
raft_entry_t
*
raft_log_get_from_i
d
x
(
raft_log_t
*
me_
,
int
idx
)
{
raft_log_private_t
*
me
=
(
void
*
)
me_
;
return
hashmap_get
(
me
->
log_map
,
(
void
*
)
idx
)
;
int
i
=
(
me
->
front
+
idx
-
me
->
base_log_index
)
%
me
->
size
;
return
&
me
->
entries
[
i
]
;
}
int
raft_log_count
(
raft_log_t
*
me_
)
{
raft_log_private_t
*
me
=
(
void
*
)
me_
;
return
arrayqueue_count
(
me
->
log_qu
)
;
return
me
->
count
;
}
/**
* Delete all logs from this log onwards */
void
raft_log_delete
(
raft_log_t
*
me_
,
int
idx
)
{
raft_log_private_t
*
me
=
(
void
*
)
me_
;
int
end
,
i
;
for
(
end
=
raft_log_count
(
me_
);
idx
<
end
;
idx
++
)
{
int
idx2
=
(
me
->
front
+
idx
-
me
->
base_log_index
)
%
me
->
size
;
me
->
back
--
;
me
->
count
--
;
}
#if 0
const void *elem;
if (arrayqueue_is_empty(me))
return NULL;
// __checkwrapping(me);
in(me)->back--;
in(me)->count--;
if (-1 == in(me)->back)
in(me)->back = in(me)->size;
elem = me->entries[in(me)->back];
return (void *) elem;
#endif
}
/**
* Remove oldest entry
* @return oldest entry */
void
*
raft_log_poll
(
raft_log_t
*
me_
)
{
raft_log_private_t
*
me
=
(
void
*
)
me_
;
const
void
*
elem
;
if
(
0
==
raft_log_count
(
me_
))
return
NULL
;
elem
=
&
me
->
entries
[
me
->
front
];
me
->
front
++
;
me
->
count
--
;
me
->
base_log_index
++
;
return
(
void
*
)
elem
;
}
/**
* Empty the queue. */
void
raft_log_empty
(
raft_log_t
*
me_
)
{
raft_log_private_t
*
me
=
(
void
*
)
me_
;
me
->
front
=
0
;
me
->
back
=
0
;
me
->
count
=
0
;
}
void
raft_log_free
(
raft_log_t
*
me_
)
{
raft_log_private_t
*
me
=
(
void
*
)
me_
;
free
(
me
->
entries
);
free
(
me
);
}
raft_log.h
View file @
4469906e
...
...
@@ -8,3 +8,4 @@ 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 @
4469906e
...
...
@@ -15,6 +15,9 @@
#include <stdio.h>
#include <assert.h>
/* for varags */
#include <stdarg.h>
#include "linked_list_hashmap.h"
#include "raft.h"
#include "raft_log.h"
...
...
@@ -71,7 +74,7 @@ typedef struct {
int
*
votes_for_me
;
}
raft_server_private_t
;
/* TODO: replace with better hash() */
static
unsigned
long
__peer_hash
(
const
void
*
obj
)
...
...
@@ -87,6 +90,18 @@ static long __peer_compare(
return
obj
-
other
;
}
static
void
__log
(
raft_server_t
*
me_
,
void
*
src
,
const
char
*
fmt
,
...)
{
raft_server_private_t
*
me
=
(
void
*
)
me_
;
char
buf
[
1024
];
va_list
args
;
va_start
(
args
,
fmt
);
vsprintf
(
buf
,
fmt
,
args
);
printf
(
"%s
\n
"
,
buf
);
//__FUNC_log(bto,src,buf);
}
raft_server_t
*
raft_new
()
{
raft_server_private_t
*
me
;
...
...
@@ -122,7 +137,7 @@ int raft_get_state(raft_server_t* me_)
return
me
->
state
;
}
void
raft_set_
external_function
s
(
raft_server_t
*
me_
,
raft_external_functions_t
*
funcs
,
void
*
caller
)
void
raft_set_
callback
s
(
raft_server_t
*
me_
,
raft_external_functions_t
*
funcs
,
void
*
caller
)
{
raft_server_private_t
*
me
=
(
void
*
)
me_
;
...
...
@@ -222,33 +237,93 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period)
return
0
;
}
raft_entry_t
*
raft_get_entry_from_idx
(
raft_server_t
*
me_
,
int
idx
)
{
raft_server_private_t
*
me
=
(
void
*
)
me_
;
return
raft_log_get_from_idx
(
me
->
log
,
idx
);
}
int
raft_recv_appendentries_response
(
raft_server_t
*
me_
,
int
peer
,
msg_appendentries_response_t
*
ae
)
{
return
0
;
}
/**
* Invoked by leader to replicate log entries (5.3); also used as heartbeat (5.2). */
int
raft_recv_appendentries
(
raft_server_t
*
me_
,
const
int
peer
,
msg_appendentries_t
*
ae
)
int
raft_recv_appendentries
(
raft_server_t
*
me_
,
const
int
peer
,
msg_appendentries_t
*
ae
)
{
raft_server_private_t
*
me
=
(
void
*
)
me_
;
msg_appendentries_response_t
r
;
r
.
term
=
raft_get_current_term
(
me_
);
if
(
ae
->
term
<
raft_get_current_term
(
me_
)
||
ae
->
prev_log_idx
<
raft_get_current_idx
(
me_
))
/* 1. Reply false if term < currentTerm (5.1) */
if
(
ae
->
term
<
raft_get_current_term
(
me_
))
{
__log
(
me_
,
NULL
,
"AE term is less than current term"
);
r
.
success
=
0
;
goto
done
;
}
if
(
raft_is_candidate
(
me_
))
#if 0
if (-1 != ae->prev_log_idx &&
ae->prev_log_idx < raft_get_current_idx(me_))
{
raft_become_follower
(
me_
);
__log(me_, NULL, "AE prev_idx is less than current idx");
r.success = 0;
goto done;
}
#endif
if
(
raft_get_current_term
(
me_
)
<
ae
->
term
)
if
(
-
1
!=
ae
->
prev_log_idx
)
{
raft_set_current_term
(
me_
,
ae
->
term
);
raft_entry_t
*
e
;
if
((
e
=
raft_get_entry_from_idx
(
me_
,
ae
->
prev_log_idx
)))
{
/* 2. Reply false if log doesnt contain an entry at prevLogIndex
whose term matches prevLogTerm (5.3) */
if
(
e
->
term
!=
ae
->
prev_log_term
)
{
__log
(
me_
,
NULL
,
"AE term doesn't match prev_idx"
);
r
.
success
=
0
;
goto
done
;
}
/* 3. If an existing entry conflicts with a new one (same index
but different terms), delete the existing entry and all that
follow it (5.3) */
raft_entry_t
*
e2
;
if
((
e2
=
raft_get_entry_from_idx
(
me_
,
ae
->
prev_log_idx
+
1
)))
{
raft_log_delete
(
me
->
log
,
ae
->
prev_log_idx
+
1
);
}
}
else
{
__log
(
me_
,
NULL
,
"AE no log at prev_idx"
);
assert
(
0
);
}
}
else
{
}
if
(
raft_is_candidate
(
me_
))
{
raft_become_follower
(
me_
);
}
raft_set_current_term
(
me_
,
ae
->
term
);
int
i
;
for
(
i
=
0
;
i
<
ae
->
n_entries
;
i
++
)
...
...
@@ -258,6 +333,7 @@ int raft_recv_appendentries(raft_server_t* me_, const int peer, msg_appendentrie
cmd
=
&
ae
->
entries
[
i
];
/* TODO: replace malloc with mempoll/arena */
c
=
malloc
(
sizeof
(
raft_entry_t
));
c
->
term
=
raft_get_current_term
(
me_
);
c
->
len
=
cmd
->
len
;
...
...
@@ -267,7 +343,7 @@ int raft_recv_appendentries(raft_server_t* me_, const int peer, msg_appendentrie
if
(
0
==
raft_append_entry
(
me_
,
c
))
{
/*
failure
, we
couldn't append
it for some reason */
__log
(
me_
,
NULL
,
"AE
failure
;
couldn't append
entry"
);
r
.
success
=
0
;
goto
done
;
}
...
...
@@ -276,7 +352,6 @@ int raft_recv_appendentries(raft_server_t* me_, const int peer, msg_appendentrie
r
.
success
=
1
;
done:
if
(
me
->
ext_func
&&
me
->
ext_func
->
send
)
me
->
ext_func
->
send
(
me
->
caller
,
NULL
,
peer
,
(
void
*
)
&
r
,
sizeof
(
msg_appendentries_response_t
));
...
...
@@ -284,17 +359,6 @@ done:
return
0
;
}
raft_entry_t
*
raft_get_entry_from_idx
(
raft_server_t
*
me_
,
int
idx
)
{
raft_server_private_t
*
me
=
(
void
*
)
me_
;
return
NULL
;
}
int
raft_recv_appendentries_response
(
raft_server_t
*
me_
,
int
peer
,
msg_appendentries_response_t
*
ae
)
{
return
0
;
}
int
raft_recv_requestvote
(
raft_server_t
*
me_
,
int
peer
,
msg_requestvote_t
*
vr
)
{
...
...
test_server.c
View file @
4469906e
...
...
@@ -215,9 +215,8 @@ void TestRaft_server_entry_is_retrieveable_using_idx(CuTest* tc)
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));
CuAssertTrue
(
tc
,
NULL
!=
(
ety_appended
=
raft_get_entry_from_idx
(
r
,
1
)));
CuAssertTrue
(
tc
,
!
strncmp
(
ety_appended
->
data
,
str2
,
3
));
}
// If commitidx > lastApplied: increment lastApplied, apply
...
...
@@ -414,7 +413,7 @@ void TestRaft_server_recv_requestvote_reply_false_if_term_less_than_current_term
r
=
raft_new
();
raft_set_configuration
(
r
,
cfg
);
sender
=
sender_new
();
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
raft_set_current_term
(
r
,
1
);
...
...
@@ -456,7 +455,7 @@ void TestRaft_server_dont_grant_vote_if_we_didnt_vote_for_this_candidate(
r
=
raft_new
();
raft_set_configuration
(
r
,
cfg
);
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
raft_set_current_term
(
r
,
1
);
raft_vote
(
r
,
1
);
raft_recv_requestvote
(
r
,
1
,
&
rv
);
...
...
@@ -497,7 +496,7 @@ void TestRaft_follower_recv_appendentries_reply_false_if_term_less_than_currentt
r
=
raft_new
();
raft_set_configuration
(
r
,
cfg
);
sender
=
sender_new
();
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
/* term is low */
memset
(
&
ae
,
0
,
sizeof
(
msg_appendentries_t
));
...
...
@@ -517,28 +516,41 @@ void TestRaft_follower_recv_appendentries_reply_false_if_term_less_than_currentt
void
TestRaft_follower_recv_appendentries_updates_currentterm_if_term_gt_currentterm
(
CuTest
*
tc
)
{
void
*
r
;
void
*
sender
;
msg_appendentries_t
ae
;
msg_appendentries_response_t
*
aer
;
raft_external_functions_t
funcs
=
{
.
send
=
sender_send
,
.
log
=
NULL
};
/* 2 peers */
raft_peer_configuration_t
cfg
[]
=
{
{(
-
1
),(
void
*
)
1
},
{(
-
1
),(
void
*
)
2
},
{(
-
1
),
NULL
}};
/* newer term */
memset
(
&
ae
,
0
,
sizeof
(
msg_appendentries_t
));
ae
.
term
=
2
;
r
=
raft_new
();
raft_set_configuration
(
r
,
cfg
);
sender
=
sender_new
();
raft_set_callbacks
(
r
,
&
funcs
,
sender
);
/* older currentterm */
raft_set_current_term
(
r
,
1
);
/* newer term for appendentry */
memset
(
&
ae
,
0
,
sizeof
(
msg_appendentries_t
));
/* no prev log idx */
ae
.
prev_log_idx
=
-
1
;
ae
.
term
=
2
;
/* appendentry has newer term, so we change our currentterm */
raft_recv_appendentries
(
r
,
1
,
&
ae
);
aer
=
sender_poll_msg
(
sender
);
CuAssertTrue
(
tc
,
NULL
!=
aer
);
CuAssertTrue
(
tc
,
1
==
aer
->
success
);
/* term has been updated */
CuAssertTrue
(
tc
,
2
==
raft_get_current_term
(
r
));
}
...
...
@@ -554,10 +566,6 @@ void TestRaft_follower_doesnt_log_after_appendentry_if_no_entries_are_specified(
{(
-
1
),(
void
*
)
2
},
{(
-
1
),
NULL
}};
/* appendentry */
memset
(
&
ae
,
0
,
sizeof
(
msg_appendentries_t
));
r
=
raft_new
();
raft_set_configuration
(
r
,
cfg
);
...
...
@@ -566,15 +574,31 @@ void TestRaft_follower_doesnt_log_after_appendentry_if_no_entries_are_specified(
/* log size s */
CuAssertTrue
(
tc
,
0
==
raft_get_log_count
(
r
));
/* receive an appendentry with commit */
memset
(
&
ae
,
0
,
sizeof
(
msg_appendentries_t
));
ae
.
term
=
1
;
ae
.
prev_log_term
=
1
;
ae
.
prev_log_idx
=
4
;
ae
.
leader_commit
=
5
;
ae
.
n_entries
=
0
;
raft_recv_appendentries
(
r
,
1
,
&
ae
);
CuAssertTrue
(
tc
,
1
==
raft_get_log_count
(
r
));
CuAssertTrue
(
tc
,
0
==
raft_get_log_count
(
r
));
}
void
TestRaft_follower_increases_log_after_appendentry
(
CuTest
*
tc
)
{
void
*
r
;
void
*
sender
;
msg_appendentries_t
ae
;
msg_entry_t
ety
;
msg_appendentries_response_t
*
aer
;
char
*
str
=
"aaa"
;
raft_external_functions_t
funcs
=
{
.
send
=
sender_send
,
.
log
=
NULL
};
/* 2 peers */
raft_peer_configuration_t
cfg
[]
=
{
...
...
@@ -583,18 +607,35 @@ void TestRaft_follower_increases_log_after_appendentry(CuTest * tc)
{(
-
1
),
NULL
}};
/* appendentry */
memset
(
&
ae
,
0
,
sizeof
(
msg_appendentries_t
));
r
=
raft_new
();
raft_set_configuration
(
r
,
cfg
);
sender
=
sender_new
();
raft_set_callbacks
(
r
,
&
funcs
,
sender
);
raft_set_state
(
r
,
RAFT_STATE_FOLLOWER
);
/* log size s */
CuAssertTrue
(
tc
,
0
==
raft_get_log_count
(
r
));
/* receive an appendentry with commit */
memset
(
&
ae
,
0
,
sizeof
(
msg_appendentries_t
));
ae
.
term
=
1
;
ae
.
prev_log_term
=
1
;
/* first appendentries msg */
ae
.
prev_log_idx
=
-
1
;
ae
.
leader_commit
=
5
;
/* include one entry */
memset
(
&
ety
,
0
,
sizeof
(
msg_entry_t
));
ety
.
data
=
str
;
ety
.
len
=
3
;
ety
.
id
=
1
;
ae
.
entries
=
&
ety
;
ae
.
n_entries
=
1
;
raft_recv_appendentries
(
r
,
1
,
&
ae
);
aer
=
sender_poll_msg
(
sender
);
CuAssertTrue
(
tc
,
NULL
!=
aer
);
CuAssertTrue
(
tc
,
1
==
aer
->
success
);
CuAssertTrue
(
tc
,
1
==
raft_get_log_count
(
r
));
}
...
...
@@ -604,6 +645,8 @@ void TestRaft_follower_recv_appendentries_reply_false_if_doesnt_have_log_at_prev
void
*
r
;
void
*
sender
;
void
*
msg
;
msg_entry_t
ety
;
char
*
str
=
"aaa"
;
raft_external_functions_t
funcs
=
{
.
send
=
sender_send
,
.
log
=
NULL
...
...
@@ -620,15 +663,25 @@ void TestRaft_follower_recv_appendentries_reply_false_if_doesnt_have_log_at_prev
sender
=
sender_new
();
r
=
raft_new
();
raft_set_configuration
(
r
,
cfg
);
raft_set_external_functions
(
r
,
&
funcs
,
sender
);
raft_set_callbacks
(
r
,
&
funcs
,
sender
);
raft_set_current_term
(
r
,
2
);
raft_set_commit_idx
(
r
,
1
);
raft_set_last_applied_idx
(
r
,
1
);
/* log idx that server doesn't have */
memset
(
&
ae
,
0
,
sizeof
(
msg_appendentries_t
));
ae
.
term
=
5
;
ae
.
prev_log_idx
=
5
;
/* current term is old */
raft_set_current_term
(
r
,
5
);
ae
.
term
=
2
;
ae
.
prev_log_idx
=
1
;
/* prev_log_term is less than current term (ie. 2) */
ae
.
prev_log_term
=
1
;
/* include one entry */
memset
(
&
ety
,
0
,
sizeof
(
msg_entry_t
));
ety
.
data
=
str
;
ety
.
len
=
3
;
ety
.
id
=
1
;
ae
.
entries
=
&
ety
;
ae
.
n_entries
=
1
;
/* trigger reply */
raft_recv_appendentries
(
r
,
1
,
&
ae
);
...
...
@@ -642,47 +695,81 @@ void TestRaft_follower_recv_appendentries_reply_false_if_doesnt_have_log_at_prev
/* 5.3 */
void
TestRaft_follower_recv_appendentries_delete_entries_if_conflict_with_new_entries
(
CuTest
*
tc
)
{
printf
(
"X
\n
"
);
void
*
r
;
void
*
sender
;
msg_appendentries_t
ae
;
msg_appendentries_response_t
*
aer
;
raft_entry_t
*
ety_appended
;
raft_external_functions_t
funcs
=
{
.
send
=
sender_send
,
.
log
=
NULL
};
/* 2 peers */
raft_peer_configuration_t
cfg
[]
=
{
{(
-
1
),(
void
*
)
1
},
{(
-
1
),(
void
*
)
2
},
{(
-
1
),
NULL
}};
r
=
raft_new
();
raft_set_configuration
(
r
,
cfg
);
sender
=
sender_new
();
raft_set_callbacks
(
r
,
&
funcs
,
sender
);
raft_set_current_term
(
r
,
2
);
/* first append entry */
memset
(
&
ae
,
0
,
sizeof
(
msg_appendentries_t
));
ae
.
term
=
1
;
ae
.
prev_log_idx
=
0
;
ae
.
prev_log_term
=
0
;
raft_set_current_term
(
r
,
1
);
raft_entry_t
ety
;
char
*
str
=
"111"
;
/* increase log size */
ety
.
data
=
str
;
char
*
str1
=
"111"
;
ety
.
data
=
str1
;
ety
.
len
=
3
;
ety
.
id
=
1
;
ety
.
term
=
1
;
raft_append_entry
(
r
,
&
ety
);
CuAssertTrue
(
tc
,
1
==
raft_get_log_count
(
r
));
/* this log will be overwritten by the appendentries below */
char
*
str2
=
"222"
;
ety
.
data
=
str2
;
ety
.
len
=
3
;
ety
.
id
=
2
;
ety
.
term
=
1
;
raft_append_entry
(
r
,
&
ety
);
CuAssertTrue
(
tc
,
2
==
raft_get_log_count
(
r
));
CuAssertTrue
(
tc
,
NULL
!=
(
ety_appended
=
raft_get_entry_from_idx
(
r
,
1
)));
CuAssertTrue
(
tc
,
!
strncmp
(
ety_appended
->
data
,
str2
,
3
));
/* pass a appendentry that is newer */
msg_entry_t
mety
;
memset
(
&
ae
,
0
,
sizeof
(
msg_appendentries_t
));
ae
.
term
=
2
;
ae
.
prev_log_idx
=
1
;
ae
.
prev_log_term
=
1
;
/* include one entry */
memset
(
&
ety
,
0
,
sizeof
(
msg_entry_t
));
char
*
str3
=
"333"
;
mety
.
data
=
str3
;
mety
.
len
=
3
;
mety
.
id
=
3
;
ae
.
entries
=
&
mety
;
ae
.
n_entries
=
1
;
raft_recv_appendentries
(
r
,
1
,
&
ae
);
CuAssertTrue
(
tc
,
0
==
raft_get_log_count
(
r
));
aer
=
sender_poll_msg
(
sender
);
CuAssertTrue
(
tc
,
NULL
!=
aer
);
CuAssertTrue
(
tc
,
1
==
aer
->
success
);
CuAssertTrue
(
tc
,
2
==
raft_get_log_count
(
r
));
CuAssertTrue
(
tc
,
NULL
!=
(
ety_appended
=
raft_get_entry_from_idx
(
r
,
1
)));
CuAssertTrue
(
tc
,
!
strncmp
(
ety_appended
->
data
,
str2
,
3
));
}
void
TestRaft_follower_recv_appendentries_add_new_entries_not_already_in_log
(
CuTest
*
tc
)
{
printf
(
"Y
\n
"
);
void
*
r
;
void
*
sender
;
raft_external_functions_t
funcs
=
{
...
...
@@ -708,7 +795,7 @@ void TestRaft_follower_recv_appendentries_add_new_entries_not_already_in_log(CuT
raft_set_configuration
(
r
,
cfg
);
raft_set_current_term
(
r
,
1
);
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
raft_recv_appendentries
(
r
,
1
,
&
ae
);
CuAssertTrue
(
tc
,
0
);
...
...
@@ -739,7 +826,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx(CuTest * t
sender
=
sender_new
();
r
=
raft_new
();
raft_set_configuration
(
r
,
cfg
);
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
/* receive an appendentry with commit */
memset
(
&
ae
,
0
,
sizeof
(
msg_appendentries_t
));
...
...
@@ -774,7 +861,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit(CuTest *
sender
=
sender_new
();
r
=
raft_new
();
raft_set_configuration
(
r
,
cfg
);
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
/* receive an appendentry with commit */
memset
(
&
ae
,
0
,
sizeof
(
msg_appendentries_t
));
...
...
@@ -892,7 +979,7 @@ void TestRaft_follower_dont_grant_vote_if_candidate_has_a_less_complete_log(CuTe
sender
=
sender_new
();
r
=
raft_new
();
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
raft_set_configuration
(
r
,
cfg
);
/* request vote */
...
...
@@ -984,7 +1071,7 @@ void TestRaft_follower_becoming_candidate_requests_votes_from_other_servers(CuTe
sender
=
sender_new
();
r
=
raft_new
();
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
raft_set_configuration
(
r
,
cfg
);
/* set term so we can check it gets included in the outbound message */
...
...
@@ -1030,7 +1117,7 @@ void TestRaft_candidate_election_timeout_and_no_leader_results_in_new_election(C
r
=
raft_new
();
raft_set_configuration
(
r
,
cfg
);
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
raft_set_election_timeout
(
r
,
1000
);
/* server wants to be leader, so becomes candidate */
...
...
@@ -1072,7 +1159,7 @@ void TestRaft_candidate_receives_majority_of_votes_becomes_leader(CuTest * tc)
r
=
raft_new
();
raft_set_configuration
(
r
,
cfg
);
CuAssertTrue
(
tc
,
5
==
raft_get_npeers
(
r
));
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
raft_become_candidate
(
r
);
CuAssertTrue
(
tc
,
1
==
raft_get_current_term
(
r
));
...
...
@@ -1116,7 +1203,7 @@ void TestRaft_candidate_will_not_respond_to_voterequest_if_it_has_already_voted(
sender
=
sender_new
();
r
=
raft_new
();
raft_set_configuration
(
r
,
cfg
);
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
raft_vote
(
r
,
1
);
...
...
@@ -1149,7 +1236,7 @@ void TestRaft_candidate_requestvote_includes_logidx(CuTest * tc)
raft_set_configuration
(
r
,
cfg
);
raft_set_state
(
r
,
RAFT_STATE_CANDIDATE
);
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
raft_set_current_term
(
r
,
5
);
raft_set_current_idx
(
r
,
3
);
raft_send_requestvote
(
r
,
1
);
...
...
@@ -1185,7 +1272,7 @@ void TestRaft_candidate_recv_appendentries_frm_leader_results_in_follower(CuTest
r
=
raft_new
();
raft_set_configuration
(
r
,
cfg
);
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
raft_set_state
(
r
,
RAFT_STATE_CANDIDATE
);
CuAssertTrue
(
tc
,
0
==
raft_is_follower
(
r
));
...
...
@@ -1215,7 +1302,7 @@ void TestRaft_candidate_recv_appendentries_frm_invalid_leader_doesnt_result_in_f
sender
=
sender_new
();
r
=
raft_new
();
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
raft_set_configuration
(
r
,
cfg
);
/* server's log is newer */
...
...
@@ -1267,7 +1354,7 @@ void TestRaft_leader_when_it_becomes_a_leader_sends_empty_appendentries(CuTest *
sender
=
sender_new
();
r
=
raft_new
();
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
raft_set_configuration
(
r
,
cfg
);
/* candidate to leader */
...
...
@@ -1300,7 +1387,7 @@ void TestRaft_leader_responds_to_entry_msg_when_entry_is_committed(CuTest * tc)
sender
=
sender_new
();
r
=
raft_new
();
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
raft_set_configuration
(
r
,
cfg
);
/* I am the leader */
...
...
@@ -1346,7 +1433,7 @@ void TestRaft_leader_sends_appendentries_with_NextIdx_when_PrevIdx_gt_NextIdx(Cu
sender
=
sender_new
();
r
=
raft_new
();
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
raft_set_configuration
(
r
,
cfg
);
/* i'm leader */
...
...
@@ -1379,7 +1466,7 @@ void TestRaft_leader_retries_appendentries_with_decremented_NextIdx_log_inconsis
sender
=
sender_new
();
r
=
raft_new
();
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
raft_set_configuration
(
r
,
cfg
);
/* i'm leader */
...
...
@@ -1477,7 +1564,7 @@ void TestRaft_leader_increase_commitno_when_majority_have_entry_and_atleast_one_
raft_set_current_term
(
r
,
5
);
raft_set_current_idx
(
r
,
5
);
raft_set_commit_idx
(
r
,
4
);
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
raft_send_appendentries
(
r
,
1
);
raft_send_appendentries
(
r
,
2
);
...
...
@@ -1522,7 +1609,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_idx
(
r
,
5
);
raft_set_
external_function
s
(
r
,
&
funcs
,
sender
);
raft_set_
callback
s
(
r
,
&
funcs
,
sender
);
raft_recv_appendentries
(
r
,
1
,
&
ae
);
CuAssertTrue
(
tc
,
1
==
raft_is_follower
(
r
));
...
...
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