Commit 33367594 authored by willem's avatar willem
Browse files

Better documentation

parent e95a1a87
...@@ -8,26 +8,24 @@ ...@@ -8,26 +8,24 @@
* @version 0.1 * @version 0.1
*/ */
enum {
RAFT_STATE_NONE,
RAFT_STATE_FOLLOWER,
RAFT_STATE_CANDIDATE,
RAFT_STATE_LEADER
};
typedef struct { typedef struct {
/* So that we can tell which nodes were removed/added */ /** The ID that this node used to have.
* So that we can tell which nodes were removed/added when the
* configuration changes */
int old_id; int old_id;
/* User data pointer for addressing.
* ie. This is most likely a (IP,Port) tuple */ /** User data pointer for addressing.
* Examples of what this could be:
* - void* pointing to implementor's networking data
* - a (IP,Port) tuple */
void* udata_address; void* udata_address;
} raft_node_configuration_t; } raft_node_configuration_t;
typedef struct { typedef struct {
/* term candidate's term */ /* candidate's term */
int term; int term;
/* candidateId candidate requesting vote */ /* candidate requesting vote */
int candidate_id; int candidate_id;
/* idx of candidate's last log entry */ /* idx of candidate's last log entry */
...@@ -37,25 +35,22 @@ typedef struct { ...@@ -37,25 +35,22 @@ typedef struct {
int last_log_term; int last_log_term;
} msg_requestvote_t; } msg_requestvote_t;
/**
* every server must send entrys in this format */
typedef struct { typedef struct {
/* the entry's unique ID */
unsigned int id; unsigned int id;
/* entry data */
unsigned char* data; unsigned char* data;
/* length of entry data */
unsigned int len; unsigned int len;
} msg_entry_t; } msg_entry_t;
typedef struct { typedef struct {
unsigned int term; /* the entry's unique ID */
unsigned int id; unsigned int id;
unsigned char* data;
unsigned int len;
/* number of nodes that have this entry */
unsigned int nnodes;
} raft_entry_t;
typedef struct { /* whether or not the entry was committed */
unsigned int id;
int was_committed; int was_committed;
} msg_entry_response_t; } msg_entry_response_t;
...@@ -85,12 +80,12 @@ typedef struct { ...@@ -85,12 +80,12 @@ typedef struct {
* prevLogidx and prevLogTerm */ * prevLogidx and prevLogTerm */
int success; int success;
/* Non Raft fields */ /* Non-Raft fields follow: */
/* Having the following fields allows us to do less book keeping in /* Having the following fields allows us to do less book keeping in
* regards to full fledged RPC */ * regards to full fledged RPC */
/* This is the highest log IDX we've received and appended to our log */ /* This is the highest log IDX we've received and appended to our log */
int current_idx; int current_idx;
/* The first idx that we received within the append entries message */ /* The first idx that we received within the appendentries message */
int first_idx; int first_idx;
} msg_appendentries_response_t; } msg_appendentries_response_t;
...@@ -106,10 +101,10 @@ enum { ...@@ -106,10 +101,10 @@ enum {
typedef int ( typedef int (
*func_send_f *func_send_f
) ( ) (
void *caller, void *cb_ctx,
void *udata, void *udata,
int node, int node,
int type, int msg_type,
const unsigned char *send_data, const unsigned char *send_data,
const int len const int len
); );
...@@ -119,17 +114,19 @@ typedef int ( ...@@ -119,17 +114,19 @@ typedef int (
typedef void ( typedef void (
*func_log_f *func_log_f
) ( ) (
void *udata, void *cb_ctx,
void *src, void *src,
const char *buf, const char *buf,
... ...
); );
#endif #endif
/**
* Apply this log to the state macine */
typedef int ( typedef int (
*func_applylog_f *func_applylog_f
) ( ) (
void *caller, void *cb_ctx,
void *udata, void *udata,
const unsigned char *data, const unsigned char *data,
const int len const int len
...@@ -144,6 +141,19 @@ typedef struct { ...@@ -144,6 +141,19 @@ typedef struct {
typedef void* raft_server_t; typedef void* raft_server_t;
typedef void* raft_node_t; typedef void* raft_node_t;
typedef struct {
/* entry's term */
unsigned int term;
/* the entry's unique ID */
unsigned int id;
/* entry data */
unsigned char* data;
/* length of entry data */
unsigned int len;
/* number of nodes that have this entry */
unsigned int nnodes;
} raft_entry_t;
/** /**
* Initialise a new raft server * Initialise a new raft server
* Election timeout defaults to 1000 milliseconds */ * Election timeout defaults to 1000 milliseconds */
...@@ -162,16 +172,18 @@ void raft_set_callbacks(raft_server_t* me, raft_cbs_t* funcs, void* cb_ctx); ...@@ -162,16 +172,18 @@ void raft_set_callbacks(raft_server_t* me, raft_cbs_t* funcs, void* cb_ctx);
/** /**
* Set configuration * Set configuration
* @param nodes Array of nodes, end of array is marked by NULL entry * @param nodes Array of nodes, end of array is marked by NULL entry
* @param me_idx Which node is myself */ * @param my_idx Which node is myself */
void raft_set_configuration(raft_server_t* me_, void raft_set_configuration(raft_server_t* me_,
raft_node_configuration_t* nodes, int me_idx); raft_node_configuration_t* nodes, int my_idx);
/** /**
* Set election timeout */ * Set election timeout
* @param millisec Election timeout in milliseconds */
void raft_set_election_timeout(raft_server_t* me, int millisec); void raft_set_election_timeout(raft_server_t* me, int millisec);
/** /**
* Set request timeout */ * Set request timeout in milliseconds
* @param millisec Request timeout in milliseconds */
void raft_set_request_timeout(raft_server_t* me_, int millisec); void raft_set_request_timeout(raft_server_t* me_, int millisec);
/** /**
...@@ -211,11 +223,19 @@ int raft_recv_requestvote_response(raft_server_t* me, int node, ...@@ -211,11 +223,19 @@ int raft_recv_requestvote_response(raft_server_t* me, int node,
msg_requestvote_response_t* r); msg_requestvote_response_t* r);
/** /**
* Receive an ENTRY message. * Receive an entry message from client.
* Append entry to log * Append the entry to the log
* Send APPENDENTRIES to followers */ * Send appendentries to followers
int raft_recv_entry(raft_server_t* me, int node, msg_entry_t* cmd); * @param node The node this response was sent by
* @param e The entry message */
int raft_recv_entry(raft_server_t* me, int node, msg_entry_t* e);
/**
* @return the server's node ID */
int raft_get_nodeid(raft_server_t* me_);
/**
* @return currently configured election timeout in milliseconds */
int raft_get_election_timeout(raft_server_t* me); int raft_get_election_timeout(raft_server_t* me);
/** /**
...@@ -226,29 +246,52 @@ int raft_get_num_nodes(raft_server_t* me); ...@@ -226,29 +246,52 @@ int raft_get_num_nodes(raft_server_t* me);
* @return number of items within log */ * @return number of items within log */
int raft_get_log_count(raft_server_t* me); int raft_get_log_count(raft_server_t* me);
/**
* @return current term */
int raft_get_current_term(raft_server_t* me); int raft_get_current_term(raft_server_t* me);
/**
* @return current log index */
int raft_get_current_idx(raft_server_t* me); int raft_get_current_idx(raft_server_t* me);
/**
* @return 1 if follower; 0 otherwise */
int raft_is_follower(raft_server_t* me); int raft_is_follower(raft_server_t* me);
/**
* @return 1 if leader; 0 otherwise */
int raft_is_leader(raft_server_t* me); int raft_is_leader(raft_server_t* me);
/**
* @return 1 if candidate; 0 otherwise */
int raft_is_candidate(raft_server_t* me); int raft_is_candidate(raft_server_t* me);
/**
* @return currently elapsed timeout in milliseconds */
int raft_get_timeout_elapsed(raft_server_t* me); int raft_get_timeout_elapsed(raft_server_t* me);
/**
* @return index of last applied entry */
int raft_get_last_applied_idx(raft_server_t* me); int raft_get_last_applied_idx(raft_server_t* me);
/**
* @return 1 if node is leader; 0 otherwise */
int raft_node_is_leader(raft_node_t* node); int raft_node_is_leader(raft_node_t* node);
/**
* @return the node's next index */
int raft_node_get_next_idx(raft_node_t* node); int raft_node_get_next_idx(raft_node_t* node);
/**
* @param idx The entry's index
* @return entry from index */
raft_entry_t* raft_get_entry_from_idx(raft_server_t* me_, int idx); raft_entry_t* raft_get_entry_from_idx(raft_server_t* me_, int idx);
raft_node_t* raft_get_node(raft_server_t *me_, int nodeid); /**
* @param node The node's index
int raft_get_nodeid(raft_server_t* me_); * @return node pointed to by node index
*/
raft_node_t* raft_get_node(raft_server_t *me_, int node);
/** /**
* @return number of votes this server has received this election */ * @return number of votes this server has received this election */
......
...@@ -47,3 +47,9 @@ void raft_node_set_next_idx(raft_node_t* me_, int nextIdx) ...@@ -47,3 +47,9 @@ void raft_node_set_next_idx(raft_node_t* me_, int nextIdx)
raft_node_private_t* me = (void*)me_; raft_node_private_t* me = (void*)me_;
me->next_idx = nextIdx; me->next_idx = nextIdx;
} }
void* raft_node_get_udata(raft_node_t* me_)
{
raft_node_private_t* me = (void*)me_;
return me->udata;
}
...@@ -8,7 +8,13 @@ ...@@ -8,7 +8,13 @@
* @version 0.1 * @version 0.1
*/ */
/* These are here for testing purposes */ enum {
RAFT_STATE_NONE,
RAFT_STATE_FOLLOWER,
RAFT_STATE_CANDIDATE,
RAFT_STATE_LEADER
};
void raft_election_start(raft_server_t* me); void raft_election_start(raft_server_t* me);
......
...@@ -510,7 +510,7 @@ int raft_send_entry_response(raft_server_t* me_, ...@@ -510,7 +510,7 @@ int raft_send_entry_response(raft_server_t* me_,
return 0; return 0;
} }
int raft_recv_entry(raft_server_t* me_, int node, msg_entry_t* cmd) int raft_recv_entry(raft_server_t* me_, int node, msg_entry_t* e)
{ {
raft_server_private_t* me = (void*)me_; raft_server_private_t* me = (void*)me_;
raft_entry_t ety; raft_entry_t ety;
...@@ -519,11 +519,11 @@ int raft_recv_entry(raft_server_t* me_, int node, msg_entry_t* cmd) ...@@ -519,11 +519,11 @@ int raft_recv_entry(raft_server_t* me_, int node, msg_entry_t* cmd)
__log(me_, NULL, "received entry from: %d", node); __log(me_, NULL, "received entry from: %d", node);
ety.term = me->current_term; ety.term = me->current_term;
ety.id = cmd->id; ety.id = e->id;
ety.data = cmd->data; ety.data = e->data;
ety.len = cmd->len; ety.len = e->len;
res = raft_append_entry(me_, &ety); res = raft_append_entry(me_, &ety);
raft_send_entry_response(me_, node, cmd->id, res); raft_send_entry_response(me_, node, e->id, res);
for (i=0; i<me->nnodes; i++) for (i=0; i<me->nnodes; i++)
{ {
if (me->nodeid == i) continue; if (me->nodeid == i) continue;
...@@ -562,8 +562,9 @@ int raft_append_entry(raft_server_t* me_, raft_entry_t* c) ...@@ -562,8 +562,9 @@ int raft_append_entry(raft_server_t* me_, raft_entry_t* c)
int raft_apply_entry(raft_server_t* me_) int raft_apply_entry(raft_server_t* me_)
{ {
raft_server_private_t* me = (void*)me_; raft_server_private_t* me = (void*)me_;
raft_entry_t* e;
if (!log_get_from_idx(me->log, me->last_applied_idx+1)) if (!(e = log_get_from_idx(me->log, me->last_applied_idx+1)))
return 0; return 0;
__log(me_, NULL, "applying log: %d", me->last_applied_idx); __log(me_, NULL, "applying log: %d", me->last_applied_idx);
...@@ -571,9 +572,8 @@ int raft_apply_entry(raft_server_t* me_) ...@@ -571,9 +572,8 @@ int raft_apply_entry(raft_server_t* me_)
me->last_applied_idx++; me->last_applied_idx++;
if (me->commit_idx < me->last_applied_idx) if (me->commit_idx < me->last_applied_idx)
me->commit_idx = me->last_applied_idx; me->commit_idx = me->last_applied_idx;
if (me->cb.applylog)
// TODO: callback for state machine application goes here me->cb.applylog(me->cb_ctx, me, e->data, e->len);
return 1; return 1;
} }
...@@ -612,7 +612,7 @@ void raft_send_appendentries_all(raft_server_t* me_) ...@@ -612,7 +612,7 @@ void raft_send_appendentries_all(raft_server_t* me_)
} }
void raft_set_configuration(raft_server_t* me_, void raft_set_configuration(raft_server_t* me_,
raft_node_configuration_t* nodes, int me_idx) raft_node_configuration_t* nodes, int my_idx)
{ {
raft_server_private_t* me = (void*)me_; raft_server_private_t* me = (void*)me_;
int nnodes; int nnodes;
...@@ -623,10 +623,10 @@ void raft_set_configuration(raft_server_t* me_, ...@@ -623,10 +623,10 @@ void raft_set_configuration(raft_server_t* me_,
nnodes++; nnodes++;
me->nodes = realloc(me->nodes,sizeof(raft_node_t*) * nnodes); me->nodes = realloc(me->nodes,sizeof(raft_node_t*) * nnodes);
me->nnodes = nnodes; me->nnodes = nnodes;
me->nodes[nnodes-1] = raft_node_new(nodes); me->nodes[nnodes-1] = raft_node_new(nodes->udata_address);
} }
me->votes_for_me = calloc(nnodes, sizeof(int)); me->votes_for_me = calloc(nnodes, sizeof(int));
me->nodeid = me_idx; me->nodeid = my_idx;
} }
int raft_get_nvotes_for_me(raft_server_t* me_) int raft_get_nvotes_for_me(raft_server_t* me_)
...@@ -647,6 +647,12 @@ int raft_get_nvotes_for_me(raft_server_t* me_) ...@@ -647,6 +647,12 @@ int raft_get_nvotes_for_me(raft_server_t* me_)
return votes; return votes;
} }
void raft_vote(raft_server_t* me_, int node)
{
raft_server_private_t* me = (void*)me_;
me->voted_for = node;
}
void raft_set_election_timeout(raft_server_t* me_, int millisec) void raft_set_election_timeout(raft_server_t* me_, int millisec)
{ {
raft_server_private_t* me = (void*)me_; raft_server_private_t* me = (void*)me_;
...@@ -674,12 +680,6 @@ int raft_get_request_timeout(raft_server_t* me_) ...@@ -674,12 +680,6 @@ int raft_get_request_timeout(raft_server_t* me_)
return ((raft_server_private_t*)me_)->request_timeout; return ((raft_server_private_t*)me_)->request_timeout;
} }
void raft_vote(raft_server_t* me_, int node)
{
raft_server_private_t* me = (void*)me_;
me->voted_for = node;
}
int raft_get_num_nodes(raft_server_t* me_) int raft_get_num_nodes(raft_server_t* me_)
{ {
return ((raft_server_private_t*)me_)->nnodes; return ((raft_server_private_t*)me_)->nnodes;
......
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