Commit a5670fa4 authored by willem's avatar willem
Browse files

Refactoring

parent 5bd3d1d2
......@@ -41,6 +41,8 @@ typedef struct {
unsigned int id;
unsigned char* data;
unsigned int len;
/* number of peers that have this entry */
unsigned int npeers;
} raft_entry_t;
typedef struct {
......@@ -74,11 +76,13 @@ typedef struct {
* prevLogidx and prevLogTerm */
int success;
/* NON-RAFT field */
/* Having the following field allows us to do less book keeping in
* regards to full fledged RPC. This is the highest log IDX we've
* received and appended to our log */
/* Non Raft fields */
/* Having the following fields allows us to do less book keeping in
* regards to full fledged RPC */
/* This is the highest log IDX we've received and appended to our log */
int current_idx;
/* The first idx that we received within the append entries message */
int first_idx;
} msg_appendentries_response_t;
typedef int (
......
......@@ -219,3 +219,7 @@ void log_free(log_t * me_)
free(me);
}
void log_mark_peer_has_committed(log_t* me_, int idx)
{
}
......@@ -10,3 +10,5 @@ int log_count(log_t* me_);
raft_entry_t* log_get_from_idx(log_t* me_, int idx);
raft_entry_t *log_peektail(log_t * me_);
void log_mark_peer_has_committed(log_t* me_, int idx);
......@@ -72,24 +72,24 @@ typedef struct {
/* follower/leader/candidate indicator */
int state;
/* callbacks */
raft_external_functions_t *cb;
void* cb_ctx;
/* indicates size of log */
/* most recently append idx, also indicates size of log */
int current_idx;
/* amount of time left till timeout */
int timeout_elapased;
/* who has voted for me. This is an array with N = 'npeers' elements */
int *votes_for_me;
raft_peer_t* peers;
int npeers;
int election_timeout;
int request_timeout;
/* who has voted for me */
int *votes_for_me;
/* callbacks */
raft_external_functions_t *cb;
void* cb_ctx;
union {
leader_t leader;
......@@ -116,6 +116,7 @@ raft_server_t* raft_new()
if (!(me = calloc(1,sizeof(raft_server_private_t))))
return NULL;
me->voted_for = -1;
me->current_idx = 1;
me->timeout_elapased = 0;
......@@ -128,6 +129,8 @@ raft_server_t* raft_new()
void raft_free(raft_server_t* me_)
{
raft_server_private_t* me = (void*)me_;
log_free(me->log);
free(me_);
}
......@@ -191,6 +194,84 @@ int raft_get_timeout_elapsed(raft_server_t* me_)
return ((raft_server_private_t*)me_)->timeout_elapased;
}
/**
* @return number of items within log */
int raft_get_log_count(raft_server_t* me_)
{
raft_server_private_t* me = (void*)me_;
return log_count(me->log);
}
int raft_get_voted_for(raft_server_t* me_)
{
return ((raft_server_private_t*)me_)->voted_for;
}
void raft_set_current_term(raft_server_t* me_, int term)
{
raft_server_private_t* me = (void*)me_;
me->current_term = term;
}
int raft_get_current_term(raft_server_t* me_)
{
return ((raft_server_private_t*)me_)->current_term;
}
void raft_set_current_idx(raft_server_t* me_, int idx)
{
raft_server_private_t* me = (void*)me_;
me->current_idx = idx;
}
int raft_get_current_idx(raft_server_t* me_)
{
return ((raft_server_private_t*)me_)->current_idx;
}
int raft_is_follower(raft_server_t* me_)
{
return raft_get_state(me_) == RAFT_STATE_FOLLOWER;
}
int raft_is_leader(raft_server_t* me_)
{
return raft_get_state(me_) == RAFT_STATE_LEADER;
}
int raft_is_candidate(raft_server_t* me_)
{
return raft_get_state(me_) == RAFT_STATE_CANDIDATE;
}
int raft_get_my_id(raft_server_t* me_)
{
// TODO
return 0;
}
void raft_set_commit_idx(raft_server_t* me_, int idx)
{
raft_server_private_t* me = (void*)me_;
me->commit_idx = idx;
}
void raft_set_last_applied_idx(raft_server_t* me_, int idx)
{
raft_server_private_t* me = (void*)me_;
me->last_applied_idx = idx;
}
int raft_get_last_applied_idx(raft_server_t* me_)
{
return ((raft_server_private_t*)me_)->last_applied_idx;
}
int raft_get_commit_idx(raft_server_t* me_)
{
return ((raft_server_private_t*)me_)->commit_idx;
}
void raft_set_callbacks(raft_server_t* me_, raft_external_functions_t* funcs, void* cb_ctx)
{
raft_server_private_t* me = (void*)me_;
......@@ -201,13 +282,13 @@ void raft_set_callbacks(raft_server_t* me_, raft_external_functions_t* funcs, vo
void raft_election_start(raft_server_t* me_)
{
raft_server_private_t* me = (void*)me_;
// TODO: randomise timeout
me->timeout_elapased = 0;
/* time to throw our hat in */
raft_become_candidate(me_);
}
/**
* Candidate transitions to leader. */
void raft_become_leader(raft_server_t* me_)
{
raft_server_private_t* me = (void*)me_;
......@@ -251,18 +332,23 @@ void raft_become_follower(raft_server_t* me_)
raft_set_state(me_,RAFT_STATE_FOLLOWER);
}
/**
* @return 0 on error */
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_);
{
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_);
return 0;
return 1;
}
raft_entry_t* raft_get_entry_from_idx(raft_server_t* me_, int etyidx)
......@@ -271,21 +357,54 @@ raft_entry_t* raft_get_entry_from_idx(raft_server_t* me_, int etyidx)
return log_get_from_idx(me->log, etyidx);
}
/**
* @return 0 on error */
int raft_recv_appendentries_response(raft_server_t* me_,
int peer, msg_appendentries_response_t* ae)
int peer, msg_appendentries_response_t* aer)
{
raft_server_private_t* me = (void*)me_;
raft_peer_t* p;
p = raft_get_peer(me_, peer);
if (ae->success)
if (aer->success)
{
int i;
for (i=aer->first_idx; i<aer->current_idx; i++)
log_mark_peer_has_committed(me->log, i);
while (1)
{
raft_entry_t* e;
e = log_get_from_idx(me->log, me->commit_idx);
/* majority has this */
if (e && me->npeers / 2 <= e->npeers)
{
raft_apply_entry(me_);
}
else
{
break;
}
}
}
else
{
// TODO does this have coverage?
assert(0 <= raft_peer_get_next_idx(p));
// 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_send_appendentries(me_, peer);
}
return 0;
return 1;
}
/**
* Invoked by leader to replicate log entries (5.3); also used as heartbeat (5.2). */
* @return 0 on error */
int raft_recv_appendentries(
raft_server_t* me_,
const int peer,
......@@ -314,6 +433,7 @@ int raft_recv_appendentries(
}
#endif
/* not the first appendentries we've received */
if (0 != ae->prev_log_idx)
{
raft_entry_t* e;
......@@ -366,7 +486,7 @@ int raft_recv_appendentries(
int i;
/* append entries */
/* append all entries to log */
for (i=0; i<ae->n_entries; i++)
{
msg_entry_t* cmd;
......@@ -391,12 +511,13 @@ int raft_recv_appendentries(
r.success = 1;
r.current_idx = raft_get_current_idx(me_);
r.first_idx = ae->prev_log_idx + 1;
done:
if (me->cb && me->cb->send)
me->cb->send(me->cb_ctx, NULL, peer, (void*)&r,
sizeof(msg_appendentries_response_t));
return 0;
return 1;
}
int raft_recv_requestvote(raft_server_t* me_, int peer, msg_requestvote_t* vr)
......@@ -461,7 +582,8 @@ int raft_recv_requestvote_response(raft_server_t* me_, int peer, msg_requestvote
return 0;
}
int raft_send_entry_response(raft_server_t* me_, int peerid, int etyid, int was_committed)
int raft_send_entry_response(raft_server_t* me_,
int peerid, int etyid, int was_committed)
{
raft_server_private_t* me = (void*)me_;
msg_entry_response_t res;
......@@ -475,12 +597,14 @@ int raft_send_entry_response(raft_server_t* me_, int peerid, int etyid, int was_
}
/**
* Receive an ENTRY message */
* Receive an ENTRY message.
* Append entry to log
* Send APPENDENTRIES to followers */
int raft_recv_entry(raft_server_t* me_, int peer, msg_entry_t* cmd)
{
raft_server_private_t* me = (void*)me_;
raft_entry_t ety;
int res;
int res, i;
ety.term = raft_get_current_term(me_);
ety.id = cmd->id;
......@@ -488,87 +612,13 @@ int raft_recv_entry(raft_server_t* me_, int peer, msg_entry_t* cmd)
ety.len = cmd->len;
res = raft_append_entry(me_, &ety);
raft_send_entry_response(me_, peer, cmd->id, res);
for (i=0; i<me->npeers; i++)
raft_send_appendentries(me_,i);
return 0;
}
/**
* @return number of items within log */
int raft_get_log_count(raft_server_t* me_)
{
raft_server_private_t* me = (void*)me_;
return log_count(me->log);
}
int raft_get_voted_for(raft_server_t* me_)
{
return ((raft_server_private_t*)me_)->voted_for;
}
void raft_set_current_term(raft_server_t* me_, int term)
{
raft_server_private_t* me = (void*)me_;
me->current_term = term;
}
int raft_get_current_term(raft_server_t* me_)
{
return ((raft_server_private_t*)me_)->current_term;
}
void raft_set_current_idx(raft_server_t* me_, int idx)
{
raft_server_private_t* me = (void*)me_;
me->current_idx = idx;
}
int raft_get_current_idx(raft_server_t* me_)
{
return ((raft_server_private_t*)me_)->current_idx;
}
int raft_is_follower(raft_server_t* me_)
{
return raft_get_state(me_) == RAFT_STATE_FOLLOWER;
}
int raft_is_leader(raft_server_t* me_)
{
return raft_get_state(me_) == RAFT_STATE_LEADER;
}
int raft_is_candidate(raft_server_t* me_)
{
return raft_get_state(me_) == RAFT_STATE_CANDIDATE;
}
int raft_get_my_id(raft_server_t* me_)
{
// TODO
return 0;
}
void raft_set_commit_idx(raft_server_t* me_, int idx)
{
raft_server_private_t* me = (void*)me_;
me->commit_idx = idx;
}
void raft_set_last_applied_idx(raft_server_t* me_, int idx)
{
raft_server_private_t* me = (void*)me_;
me->last_applied_idx = idx;
}
int raft_get_last_applied_idx(raft_server_t* me_)
{
return ((raft_server_private_t*)me_)->last_applied_idx;
}
int raft_get_commit_idx(raft_server_t* me_)
{
return ((raft_server_private_t*)me_)->commit_idx;
}
* @return 0 on error */
int raft_send_requestvote(raft_server_t* me_, int peer)
{
raft_server_private_t* me = (void*)me_;
......@@ -578,7 +628,7 @@ int raft_send_requestvote(raft_server_t* me_, int peer)
rv.last_log_idx = raft_get_current_idx(me_);
if (me->cb && me->cb->send)
me->cb->send(me->cb_ctx,NULL, peer, (void*)&rv, sizeof(msg_requestvote_t));
return 0;
return 1;
}
/**
......@@ -594,10 +644,7 @@ int raft_append_entry(raft_server_t* me_, raft_entry_t* c)
me->current_idx += 1;
return 1;
}
else
{
return 0;
}
return 0;
}
/**
......@@ -635,7 +682,8 @@ void raft_send_appendentries(raft_server_t* me_, int peer)
(void*)&ae, sizeof(msg_appendentries_t));
}
void raft_set_configuration(raft_server_t* me_, raft_peer_configuration_t* peers)
void raft_set_configuration(raft_server_t* me_,
raft_peer_configuration_t* peers)
{
raft_server_private_t* me = (void*)me_;
int npeers;
......
#include <stdbool.h>
#include <assert.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <stdint.h>
#include "CuTest.h"
#include "raft.h"
#include "raft_log.h"
void TestLog_new_is_empty(CuTest * tc)
{
void *l;
l = log_new();
CuAssertTrue(tc, 0 == log_count(l));
}
void TestLog_append_is_not_empty(CuTest * tc)
{
void *l;
raft_entry_t e;
e.id = 1;
l = log_new();
CuAssertTrue(tc, 1 == log_append_entry(l, &e));
CuAssertTrue(tc, 1 == log_count(l));
}
void TestLog_get_at_idx(CuTest * tc)
{
void *l;
raft_entry_t e1, e2, e3;
l = log_new();
e1.id = 1;
CuAssertTrue(tc, 1 == log_append_entry(l, &e1));
e2.id = 2;
CuAssertTrue(tc, 1 == log_append_entry(l, &e2));
e3.id = 3;
CuAssertTrue(tc, 1 == log_append_entry(l, &e3));
CuAssertTrue(tc, 3 == log_count(l));
CuAssertTrue(tc, 3 == log_count(l));
CuAssertTrue(tc, e2.id == log_get_from_idx(l,2)->id);
}
void TestLog_delete(CuTest * tc)
{
void *l;
raft_entry_t e1, e2, e3;
l = log_new();
e1.id = 1;
CuAssertTrue(tc, 1 == log_append_entry(l, &e1));
e2.id = 2;
CuAssertTrue(tc, 1 == log_append_entry(l, &e2));
e3.id = 3;
CuAssertTrue(tc, 1 == log_append_entry(l, &e3));
CuAssertTrue(tc, 3 == log_count(l));
log_delete(l,3);
CuAssertTrue(tc, 2 == log_count(l));
CuAssertTrue(tc, NULL == log_get_from_idx(l,3));
log_delete(l,2);
CuAssertTrue(tc, 1 == log_count(l));
CuAssertTrue(tc, NULL == log_get_from_idx(l,2));
log_delete(l,1);
CuAssertTrue(tc, 0 == log_count(l));
CuAssertTrue(tc, NULL == log_get_from_idx(l,1));
}
void TestLog_delete_onwards(CuTest * tc)
{
void *l;
raft_entry_t e1, e2, e3;
l = log_new();
e1.id = 1;
CuAssertTrue(tc, 1 == log_append_entry(l, &e1));
e2.id = 2;
CuAssertTrue(tc, 1 == log_append_entry(l, &e2));
e3.id = 3;
CuAssertTrue(tc, 1 == log_append_entry(l, &e3));
CuAssertTrue(tc, 3 == log_count(l));
/* even 3 gets deleted */
log_delete(l,2);
CuAssertTrue(tc, 1 == log_count(l));
CuAssertTrue(tc, e1.id == log_get_from_idx(l,1)->id);
CuAssertTrue(tc, NULL == log_get_from_idx(l,2));
CuAssertTrue(tc, NULL == log_get_from_idx(l,3));
}
void TestLog_peektail(CuTest * tc)
{
void *l;
raft_entry_t e1, e2, e3;
l = log_new();
e1.id = 1;
CuAssertTrue(tc, 1 == log_append_entry(l, &e1));
e2.id = 2;
CuAssertTrue(tc, 1 == log_append_entry(l, &e2));
e3.id = 3;
CuAssertTrue(tc, 1 == log_append_entry(l, &e3));
CuAssertTrue(tc, 3 == log_count(l));
CuAssertTrue(tc, e3.id == log_peektail(l)->id);
}
#if 0
// TODO: duplicate testing not implemented yet
void T_estlog_cant_append_duplicates(CuTest * tc)
{
void *l;
raft_entry_t e;
e.id = 1;
l = log_new();
CuAssertTrue(tc, 1 == log_append_entry(l, &e));
CuAssertTrue(tc, 1 == log_count(l));
}
#endif
......@@ -1645,7 +1645,7 @@ void T_estRaft_leader_doesnt_append_entry_if_unique_id_is_duplicate(CuTest * tc)
}
#endif
void TestRaft_leader_increase_commitno_when_majority_have_entry_and_atleast_one_newer_entry(CuTest * tc)
void TestRaft_leader_increase_appliedno_when_majority_have_entry_and_atleast_one_newer_entry(CuTest * tc)
{
void *r;
void *sender;
......@@ -1668,10 +1668,11 @@ 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 idx will became 5 */
/* the last applied idx will became 5 */
raft_set_current_term(r,5);
raft_set_current_idx(r,5);
raft_set_commit_idx(r,4);
raft_set_commit_idx(r,5);
raft_set_last_applied_idx(r,4);
raft_set_callbacks(r,&funcs,sender);
/* server will be waiting for response */
......@@ -1682,11 +1683,14 @@ void TestRaft_leader_increase_commitno_when_majority_have_entry_and_atleast_one_
memset(&aer,0,sizeof(msg_appendentries_response_t));
aer.term = 5;
aer.success = 1;
aer.current_idx = 5;
aer.first_idx = 4;
/* 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_idx(r));
printf("last applied idx: %d\n", raft_get_last_applied_idx(r));
CuAssertTrue(tc, 5 == raft_get_last_applied_idx(r));
}
void TestRaft_leader_steps_down_if_received_appendentries_is_newer_than_itself(CuTest * tc)
......@@ -1726,3 +1730,7 @@ void TestRaft_leader_steps_down_if_received_appendentries_is_newer_than_itself(C
/* TODO: If a server receives a request with a stale term number, it rejects the request. */
#if 0
void T_estRaft_leader_sends_appendentries_when_receive_entry_msg(CuTest * tc)
#endif
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