Commit 466c4c76 authored by willem's avatar willem
Browse files

Refinements to raft_log

parent a5670fa4
...@@ -32,7 +32,8 @@ typedef struct ...@@ -32,7 +32,8 @@ typedef struct
/* position of the queue */ /* position of the queue */
int front, back; int front, back;
int base_log_index; /* we compact the log, and thus need to increment the base idx */
int base_log_idx;
raft_entry_t* entries; raft_entry_t* entries;
} log_private_t; } log_private_t;
...@@ -120,9 +121,12 @@ raft_entry_t* log_get_from_idx(log_t* me_, int idx) ...@@ -120,9 +121,12 @@ raft_entry_t* log_get_from_idx(log_t* me_, int idx)
log_private_t* me = (void*)me_; log_private_t* me = (void*)me_;
int i; int i;
if (me->base_log_idx + me->count < idx || idx < me->base_log_idx)
return NULL;
/* idx starts at 1 */ /* idx starts at 1 */
idx -= 1; idx -= 1;
i = (me->front + idx - me->base_log_index) % me->size; i = (me->front + idx - me->base_log_idx) % me->size;
return &me->entries[i]; return &me->entries[i];
} }
...@@ -141,7 +145,7 @@ void log_delete(log_t* me_, int idx) ...@@ -141,7 +145,7 @@ void log_delete(log_t* me_, int idx)
/* idx starts at 1 */ /* idx starts at 1 */
idx -= 1; idx -= 1;
idx -= me->base_log_index; idx -= me->base_log_idx;
for (end = log_count(me_); idx<end; idx++) for (end = log_count(me_); idx<end; idx++)
{ {
...@@ -179,7 +183,7 @@ void *log_poll(log_t * me_) ...@@ -179,7 +183,7 @@ void *log_poll(log_t * me_)
elem = &me->entries[me->front]; elem = &me->entries[me->front];
me->front++; me->front++;
me->count--; me->count--;
me->base_log_index++; me->base_log_idx++;
return (void *) elem; return (void *) elem;
} }
......
...@@ -295,6 +295,7 @@ void raft_become_leader(raft_server_t* me_) ...@@ -295,6 +295,7 @@ void raft_become_leader(raft_server_t* me_)
int i; int i;
raft_set_state(me_,RAFT_STATE_LEADER); raft_set_state(me_,RAFT_STATE_LEADER);
for (i=0; i<me->npeers; i++) for (i=0; i<me->npeers; i++)
{ {
raft_peer_t* p = raft_get_peer(me_, i); raft_peer_t* p = raft_get_peer(me_, i);
...@@ -338,15 +339,21 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period) ...@@ -338,15 +339,21 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period)
{ {
raft_server_private_t* me = (void*)me_; raft_server_private_t* me = (void*)me_;
if (raft_get_last_applied_idx(me_) < raft_get_commit_idx(me_)) switch (me->state)
{ {
if (0 == raft_apply_entry(me_)) case RAFT_STATE_FOLLOWER:
return 0;
}
me->timeout_elapased += msec_since_last_period; if (raft_get_last_applied_idx(me_) < raft_get_commit_idx(me_))
if (me->election_timeout < me->timeout_elapased) {
raft_election_start(me_); if (0 == raft_apply_entry(me_))
return 0;
}
me->timeout_elapased += msec_since_last_period;
if (me->election_timeout < me->timeout_elapased)
raft_election_start(me_);
break;
}
return 1; return 1;
} }
...@@ -367,7 +374,7 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -367,7 +374,7 @@ int raft_recv_appendentries_response(raft_server_t* me_,
p = raft_get_peer(me_, peer); p = raft_get_peer(me_, peer);
if (aer->success) if (1 == aer->success)
{ {
int i; int i;
...@@ -378,7 +385,7 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -378,7 +385,7 @@ int raft_recv_appendentries_response(raft_server_t* me_,
{ {
raft_entry_t* e; raft_entry_t* e;
e = log_get_from_idx(me->log, me->commit_idx); e = log_get_from_idx(me->log, me->last_applied_idx + 1);
/* majority has this */ /* majority has this */
if (e && me->npeers / 2 <= e->npeers) if (e && me->npeers / 2 <= e->npeers)
...@@ -393,8 +400,10 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -393,8 +400,10 @@ int raft_recv_appendentries_response(raft_server_t* me_,
} }
else else
{ {
// TODO does this have coverage? /* If AppendEntries fails because of log inconsistency:
decrement nextIndex and retry (5.3) */
assert(0 <= raft_peer_get_next_idx(p)); assert(0 <= raft_peer_get_next_idx(p));
// TODO does this have test coverage?
// TODO can jump back to where peer is different instead of iterating // TODO can jump back to where peer is different instead of iterating
raft_peer_set_next_idx(p, raft_peer_get_next_idx(p)-1); raft_peer_set_next_idx(p, raft_peer_get_next_idx(p)-1);
raft_send_appendentries(me_, peer); raft_send_appendentries(me_, peer);
...@@ -461,7 +470,7 @@ int raft_recv_appendentries( ...@@ -461,7 +470,7 @@ int raft_recv_appendentries(
else else
{ {
__log(me_, NULL, "AE no log at prev_idx"); __log(me_, NULL, "AE no log at prev_idx");
assert(0); //assert(0);
} }
} }
...@@ -648,7 +657,7 @@ int raft_append_entry(raft_server_t* me_, raft_entry_t* c) ...@@ -648,7 +657,7 @@ int raft_append_entry(raft_server_t* me_, raft_entry_t* c)
} }
/** /**
* Apply entry at lastApplied + 1 * Apply entry at lastApplied + 1. Entry becomes 'committed'.
* @return 1 if entry committed, 0 otherwise */ * @return 1 if entry committed, 0 otherwise */
int raft_apply_entry(raft_server_t* me_) int raft_apply_entry(raft_server_t* me_)
{ {
...@@ -682,13 +691,16 @@ void raft_send_appendentries(raft_server_t* me_, int peer) ...@@ -682,13 +691,16 @@ void raft_send_appendentries(raft_server_t* me_, int peer)
(void*)&ae, sizeof(msg_appendentries_t)); (void*)&ae, sizeof(msg_appendentries_t));
} }
/**
* Set configuration
* @param peers Array of peers, end of array is marked by NULL entry */
void raft_set_configuration(raft_server_t* me_, void raft_set_configuration(raft_server_t* me_,
raft_peer_configuration_t* peers) raft_peer_configuration_t* peers)
{ {
raft_server_private_t* me = (void*)me_; raft_server_private_t* me = (void*)me_;
int npeers; int npeers;
/* TODO: one allocation only please */ /* TODO: one memory allocation only please */
for (npeers=0; peers->udata_address; peers++) for (npeers=0; peers->udata_address; peers++)
{ {
npeers++; npeers++;
......
...@@ -48,6 +48,17 @@ void TestLog_get_at_idx(CuTest * tc) ...@@ -48,6 +48,17 @@ void TestLog_get_at_idx(CuTest * tc)
CuAssertTrue(tc, e2.id == log_get_from_idx(l,2)->id); CuAssertTrue(tc, e2.id == log_get_from_idx(l,2)->id);
} }
void TestLog_get_at_idx_returns_null_where_out_of_bounds(CuTest * tc)
{
void *l;
raft_entry_t e1, e2, e3;
l = log_new();
e1.id = 1;
CuAssertTrue(tc, 1 == log_append_entry(l, &e1));
CuAssertTrue(tc, NULL == log_get_from_idx(l,2));
}
void TestLog_delete(CuTest * tc) void TestLog_delete(CuTest * tc)
{ {
void *l; void *l;
......
...@@ -1645,7 +1645,7 @@ void T_estRaft_leader_doesnt_append_entry_if_unique_id_is_duplicate(CuTest * tc) ...@@ -1645,7 +1645,7 @@ void T_estRaft_leader_doesnt_append_entry_if_unique_id_is_duplicate(CuTest * tc)
} }
#endif #endif
void TestRaft_leader_increase_appliedno_when_majority_have_entry_and_atleast_one_newer_entry(CuTest * tc) void TestRaft_leader_increase_commit_idx_when_majority_have_entry_and_atleast_one_newer_entry(CuTest * tc)
{ {
void *r; void *r;
void *sender; void *sender;
...@@ -1660,37 +1660,66 @@ void TestRaft_leader_increase_appliedno_when_majority_have_entry_and_atleast_one ...@@ -1660,37 +1660,66 @@ void TestRaft_leader_increase_appliedno_when_majority_have_entry_and_atleast_one
{(-1),(void*)1}, {(-1),(void*)1},
{(-1),(void*)2}, {(-1),(void*)2},
{(-1),NULL}}; {(-1),NULL}};
sender = sender_new(); sender = sender_new();
r = raft_new(); r = raft_new();
raft_set_configuration(r,cfg); raft_set_configuration(r,cfg);
raft_set_callbacks(r,&funcs,sender);
/* I'm the leader */ /* I'm the leader */
raft_set_state(r,RAFT_STATE_LEADER); raft_set_state(r,RAFT_STATE_LEADER);
raft_set_current_term(r,1);
raft_set_current_idx(r,0);
raft_set_commit_idx(r,0);
/* the last applied idx will became 1, and then 2 */
raft_set_last_applied_idx(r,0);
/* the last applied idx will became 5 */ /* append entries */
raft_set_current_term(r,5); raft_entry_t ety;
raft_set_current_idx(r,5); ety.term = 1;
raft_set_commit_idx(r,5); ety.id = 1;
raft_set_last_applied_idx(r,4); ety.data = "aaaa";
raft_set_callbacks(r,&funcs,sender); ety.len = 4;
raft_append_entry(r, &ety);
ety.id = 2;
raft_append_entry(r, &ety);
memset(&aer,0,sizeof(msg_appendentries_response_t));
/* server will be waiting for response */ /* FIRST entry log application */
/* send appendentries -
* server will be waiting for response */
raft_send_appendentries(r, 1); raft_send_appendentries(r, 1);
raft_send_appendentries(r, 2); raft_send_appendentries(r, 2);
/* receive mock success responses */
aer.term = 1;
aer.success = 1;
aer.current_idx = 1;
aer.first_idx = 1;
//raft_recv_appendentries_response(r,1,&aer);
//raft_recv_appendentries_response(r,2,&aer);
/* leader will now have majority followers who have appended this log */
printf("last applied idx: %d\n", raft_get_last_applied_idx(r));
printf("commit idx: %d\n", raft_get_commit_idx(r));
CuAssertTrue(tc, 1 == raft_get_commit_idx(r));
CuAssertTrue(tc, 1 == raft_get_last_applied_idx(r));
/* respond that we have the appendentries */ /* SECOND entry log application */
memset(&aer,0,sizeof(msg_appendentries_response_t)); /* send appendentries -
* server will be waiting for response */
raft_send_appendentries(r, 1);
raft_send_appendentries(r, 2);
/* receive mock success responses */
aer.term = 5; aer.term = 5;
aer.success = 1; aer.success = 1;
aer.current_idx = 5; aer.current_idx = 2;
aer.first_idx = 4; aer.first_idx = 2;
/* announce to leader that the majority have appended this log */
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 */
printf("last applied idx: %d\n", raft_get_last_applied_idx(r)); printf("last applied idx: %d\n", raft_get_last_applied_idx(r));
CuAssertTrue(tc, 5 == raft_get_last_applied_idx(r)); printf("commit idx: %d\n", raft_get_commit_idx(r));
CuAssertTrue(tc, 2 == raft_get_commit_idx(r));
CuAssertTrue(tc, 2 == raft_get_last_applied_idx(r));
} }
void TestRaft_leader_steps_down_if_received_appendentries_is_newer_than_itself(CuTest * tc) void TestRaft_leader_steps_down_if_received_appendentries_is_newer_than_itself(CuTest * tc)
......
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