Commit 37bbc66b authored by willem's avatar willem
Browse files

Refactorings

parent 466c4c76
...@@ -121,7 +121,7 @@ CuTest* CuTestNew(const char* name, TestFunction function) ...@@ -121,7 +121,7 @@ CuTest* CuTestNew(const char* name, TestFunction function)
void CuTestRun(CuTest* tc) void CuTestRun(CuTest* tc)
{ {
#if 0 /* debugging */ #if 1 /* debugging */
printf(" running %s\n", tc->name); printf(" running %s\n", tc->name);
#endif #endif
jmp_buf buf; jmp_buf buf;
......
...@@ -106,7 +106,7 @@ static void __log(raft_server_t *me_, void *src, const char *fmt, ...) ...@@ -106,7 +106,7 @@ static void __log(raft_server_t *me_, void *src, const char *fmt, ...)
va_start(args, fmt); va_start(args, fmt);
vsprintf(buf, fmt, args); vsprintf(buf, fmt, args);
//printf("%s\n", buf); printf("%s\n", buf);
//__FUNC_log(bto,src,buf); //__FUNC_log(bto,src,buf);
} }
...@@ -179,7 +179,11 @@ int raft_get_state(raft_server_t* me_) ...@@ -179,7 +179,11 @@ int raft_get_state(raft_server_t* me_)
raft_peer_t* raft_get_peer(raft_server_t *me_, int peerid) raft_peer_t* raft_get_peer(raft_server_t *me_, int peerid)
{ {
return ((raft_server_private_t*)me_)->peers[peerid]; raft_server_private_t* me = (void*)me_;
if (peerid < 0 || me->npeers <= peerid)
return NULL;
return me->peers[peerid];
} }
/** /**
...@@ -342,19 +346,18 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period) ...@@ -342,19 +346,18 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period)
switch (me->state) switch (me->state)
{ {
case RAFT_STATE_FOLLOWER: case RAFT_STATE_FOLLOWER:
if (me->last_applied_idx < me->commit_idx)
if (raft_get_last_applied_idx(me_) < raft_get_commit_idx(me_))
{ {
if (0 == raft_apply_entry(me_)) // if (0 == raft_apply_entry(me_))
return 0; // return 0;
} }
me->timeout_elapased += msec_since_last_period;
if (me->election_timeout < me->timeout_elapased)
raft_election_start(me_);
break; break;
} }
me->timeout_elapased += msec_since_last_period;
if (me->election_timeout < me->timeout_elapased)
raft_election_start(me_);
return 1; return 1;
} }
...@@ -387,10 +390,14 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -387,10 +390,14 @@ int raft_recv_appendentries_response(raft_server_t* me_,
e = log_get_from_idx(me->log, me->last_applied_idx + 1); e = log_get_from_idx(me->log, me->last_applied_idx + 1);
if (e)
printf("comparing %d to %d\n", me->npeers / 2, e->npeers);
/* majority has this */ /* majority has this */
if (e && me->npeers / 2 <= e->npeers) if (e && me->npeers / 2 <= e->npeers)
{ {
raft_apply_entry(me_); printf("applying entry\n");
if (0 == raft_apply_entry(me_)) break;
} }
else else
{ {
...@@ -470,6 +477,8 @@ int raft_recv_appendentries( ...@@ -470,6 +477,8 @@ int raft_recv_appendentries(
else else
{ {
__log(me_, NULL, "AE no log at prev_idx"); __log(me_, NULL, "AE no log at prev_idx");
r.success = 0;
goto done;
//assert(0); //assert(0);
} }
} }
...@@ -663,10 +672,12 @@ int raft_apply_entry(raft_server_t* me_) ...@@ -663,10 +672,12 @@ int raft_apply_entry(raft_server_t* me_)
{ {
raft_server_private_t* me = (void*)me_; raft_server_private_t* me = (void*)me_;
if (me->last_applied_idx == raft_get_commit_idx(me_)) if (!log_get_from_idx(me->log, me->commit_idx+1))
return 0; return 0;
me->last_applied_idx++; me->last_applied_idx++;
if (me->commit_idx < me->last_applied_idx)
me->commit_idx = me->last_applied_idx;
// TODO: callback for state machine application goes here // TODO: callback for state machine application goes here
......
...@@ -222,17 +222,54 @@ void TestRaft_server_entry_is_retrieveable_using_idx(CuTest* tc) ...@@ -222,17 +222,54 @@ void TestRaft_server_entry_is_retrieveable_using_idx(CuTest* tc)
CuAssertTrue(tc, !strncmp(ety_appended->data,str2,3)); CuAssertTrue(tc, !strncmp(ety_appended->data,str2,3));
} }
void TestRaft_server_wont_apply_entry_if_we_dont_have_entry_to_apply(CuTest* tc)
{
void *r;
raft_entry_t ety;
raft_entry_t *ety_appended;
char *str = "aaa";
r = raft_new();
raft_set_commit_idx(r,0);
raft_set_last_applied_idx(r, 0);
raft_apply_entry(r);
CuAssertTrue(tc, 0 == raft_get_last_applied_idx(r));
CuAssertTrue(tc, 0 == raft_get_commit_idx(r));
ety.term = 1;
ety.id = 1;
ety.data = str;
ety.len = 3;
raft_append_entry(r,&ety);
raft_apply_entry(r);
CuAssertTrue(tc, 1 == raft_get_last_applied_idx(r));
CuAssertTrue(tc, 1 == raft_get_commit_idx(r));
}
// If commitidx > lastApplied: increment lastApplied, apply // If commitidx > lastApplied: increment lastApplied, apply
// log[lastApplied] to state machine (5.3) // log[lastApplied] to state machine (5.3)
void TestRaft_server_increment_lastApplied_when_lastApplied_lt_commitidx(CuTest* tc) void TestRaft_server_increment_lastApplied_when_lastApplied_lt_commitidx(CuTest* tc)
{ {
void *r; void *r;
raft_entry_t ety;
r = raft_new(); r = raft_new();
raft_set_commit_idx(r,5); /* must be follower */
raft_set_last_applied_idx(r, 4); raft_set_state(r,RAFT_STATE_FOLLOWER);
raft_set_commit_idx(r,1);
raft_set_last_applied_idx(r, 0);
/* need at least one entry */
ety.term = 1;
ety.id = 1;
ety.data = "aaa";
ety.len = 3;
raft_append_entry(r,&ety);
/* let time lapse */
raft_periodic(r,1); raft_periodic(r,1);
CuAssertTrue(tc, 5 == raft_get_last_applied_idx(r)); CuAssertTrue(tc, 1 == raft_get_last_applied_idx(r));
} }
void TestRaft_server_apply_entry_increments_last_applied_idx(CuTest* tc) void TestRaft_server_apply_entry_increments_last_applied_idx(CuTest* tc)
...@@ -245,18 +282,21 @@ void TestRaft_server_apply_entry_increments_last_applied_idx(CuTest* tc) ...@@ -245,18 +282,21 @@ void TestRaft_server_apply_entry_increments_last_applied_idx(CuTest* tc)
ety.term = 1; ety.term = 1;
r = raft_new(); r = raft_new();
raft_set_commit_idx(r,5); raft_set_commit_idx(r,1);
raft_set_last_applied_idx(r, 4); raft_set_last_applied_idx(r, 0);
ety.id = 1; ety.id = 1;
ety.data = str; ety.data = str;
ety.len = 3; ety.len = 3;
raft_append_entry(r,&ety); raft_append_entry(r,&ety);
raft_apply_entry(r); raft_apply_entry(r);
CuAssertTrue(tc, 5 == raft_get_last_applied_idx(r)); CuAssertTrue(tc, 1 == raft_get_last_applied_idx(r));
} }
void TestRaft_server_apply_entry_wont_apply_if_lastapplied_equalto_commit_index(CuTest* tc) #if 0
/* Not incorrect assertion.
* The leader will always have its commit idx = last applied idx */
void T_estRaft_server_apply_entry_wont_apply_if_lastapplied_equalto_commit_index(CuTest* tc)
{ {
void *r; void *r;
raft_entry_t ety; raft_entry_t ety;
...@@ -276,6 +316,7 @@ void TestRaft_server_apply_entry_wont_apply_if_lastapplied_equalto_commit_index( ...@@ -276,6 +316,7 @@ void TestRaft_server_apply_entry_wont_apply_if_lastapplied_equalto_commit_index(
raft_apply_entry(r); raft_apply_entry(r);
CuAssertTrue(tc, 5 == raft_get_last_applied_idx(r)); CuAssertTrue(tc, 5 == raft_get_last_applied_idx(r));
} }
#endif
void TestRaft_server_periodic_elapses_election_timeout(CuTest * tc) void TestRaft_server_periodic_elapses_election_timeout(CuTest * tc)
{ {
...@@ -686,9 +727,11 @@ void TestRaft_follower_recv_appendentries_reply_false_if_doesnt_have_log_at_prev ...@@ -686,9 +727,11 @@ void TestRaft_follower_recv_appendentries_reply_false_if_doesnt_have_log_at_prev
raft_set_configuration(r,cfg); raft_set_configuration(r,cfg);
raft_set_callbacks(r,&funcs,sender); raft_set_callbacks(r,&funcs,sender);
raft_set_current_term(r,2); /* term is different from appendentries */
raft_set_commit_idx(r,1); raft_set_current_term(r, 2);
raft_set_commit_idx(r, 1);
raft_set_last_applied_idx(r, 1); raft_set_last_applied_idx(r, 1);
// TODO at log manually?
/* log idx that server doesn't have */ /* log idx that server doesn't have */
memset(&ae,0,sizeof(msg_appendentries_t)); memset(&ae,0,sizeof(msg_appendentries_t));
...@@ -1545,7 +1588,7 @@ void TestRaft_leader_sends_appendentries_with_NextIdx_when_PrevIdx_gt_NextIdx(Cu ...@@ -1545,7 +1588,7 @@ void TestRaft_leader_sends_appendentries_with_NextIdx_when_PrevIdx_gt_NextIdx(Cu
raft_peer_set_next_idx(p, 4); raft_peer_set_next_idx(p, 4);
/* receive appendentries messages */ /* receive appendentries messages */
raft_send_appendentries(r,1); raft_send_appendentries(r,0);
ae = sender_poll_msg(sender); ae = sender_poll_msg(sender);
CuAssertTrue(tc, NULL != ae); CuAssertTrue(tc, NULL != ae);
} }
...@@ -1578,7 +1621,7 @@ void TestRaft_leader_retries_appendentries_with_decremented_NextIdx_log_inconsis ...@@ -1578,7 +1621,7 @@ void TestRaft_leader_retries_appendentries_with_decremented_NextIdx_log_inconsis
raft_set_state(r,RAFT_STATE_LEADER); raft_set_state(r,RAFT_STATE_LEADER);
/* receive appendentries messages */ /* receive appendentries messages */
raft_send_appendentries(r,1); raft_send_appendentries(r,0);
ae = sender_poll_msg(sender); ae = sender_poll_msg(sender);
CuAssertTrue(tc, NULL != ae); CuAssertTrue(tc, NULL != ae);
} }
...@@ -1673,7 +1716,7 @@ void TestRaft_leader_increase_commit_idx_when_majority_have_entry_and_atleast_on ...@@ -1673,7 +1716,7 @@ void TestRaft_leader_increase_commit_idx_when_majority_have_entry_and_atleast_on
/* the last applied idx will became 1, and then 2 */ /* the last applied idx will became 1, and then 2 */
raft_set_last_applied_idx(r,0); raft_set_last_applied_idx(r,0);
/* append entries */ /* append entries - we need two */
raft_entry_t ety; raft_entry_t ety;
ety.term = 1; ety.term = 1;
ety.id = 1; ety.id = 1;
...@@ -1688,15 +1731,15 @@ void TestRaft_leader_increase_commit_idx_when_majority_have_entry_and_atleast_on ...@@ -1688,15 +1731,15 @@ void TestRaft_leader_increase_commit_idx_when_majority_have_entry_and_atleast_on
/* FIRST entry log application */ /* FIRST entry log application */
/* send appendentries - /* send appendentries -
* server will be waiting for response */ * server will be waiting for response */
raft_send_appendentries(r, 0);
raft_send_appendentries(r, 1); raft_send_appendentries(r, 1);
raft_send_appendentries(r, 2);
/* receive mock success responses */ /* receive mock success responses */
aer.term = 1; aer.term = 1;
aer.success = 1; aer.success = 1;
aer.current_idx = 1; aer.current_idx = 1;
aer.first_idx = 1; aer.first_idx = 1;
//raft_recv_appendentries_response(r,1,&aer); raft_recv_appendentries_response(r,1,&aer);
//raft_recv_appendentries_response(r,2,&aer); raft_recv_appendentries_response(r,2,&aer);
/* leader will now have majority followers who have appended this log */ /* leader will now have majority followers who have appended this log */
printf("last applied idx: %d\n", raft_get_last_applied_idx(r)); printf("last applied idx: %d\n", raft_get_last_applied_idx(r));
printf("commit idx: %d\n", raft_get_commit_idx(r)); printf("commit idx: %d\n", raft_get_commit_idx(r));
...@@ -1706,8 +1749,8 @@ void TestRaft_leader_increase_commit_idx_when_majority_have_entry_and_atleast_on ...@@ -1706,8 +1749,8 @@ void TestRaft_leader_increase_commit_idx_when_majority_have_entry_and_atleast_on
/* SECOND entry log application */ /* SECOND entry log application */
/* send appendentries - /* send appendentries -
* server will be waiting for response */ * server will be waiting for response */
raft_send_appendentries(r, 0);
raft_send_appendentries(r, 1); raft_send_appendentries(r, 1);
raft_send_appendentries(r, 2);
/* receive mock success responses */ /* receive mock success responses */
aer.term = 5; aer.term = 5;
aer.success = 1; aer.success = 1;
......
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