Unverified Commit ba31349d authored by Ozan Tezcan's avatar Ozan Tezcan Committed by GitHub
Browse files

Code refactor (#39)

code refactor
parent 676c73d2
...@@ -30,7 +30,7 @@ typedef enum { ...@@ -30,7 +30,7 @@ typedef enum {
#define RAFT_REQUESTVOTE_ERR_GRANTED 1 #define RAFT_REQUESTVOTE_ERR_GRANTED 1
#define RAFT_REQUESTVOTE_ERR_NOT_GRANTED 0 #define RAFT_REQUESTVOTE_ERR_NOT_GRANTED 0
#define RAFT_REQUESTVOTE_ERR_UNKNOWN_NODE -1 #define RAFT_REQUESTVOTE_ERR_UNKNOWN_NODE (-1)
typedef enum { typedef enum {
RAFT_STATE_NONE, RAFT_STATE_NONE,
...@@ -628,7 +628,7 @@ raft_server_t* raft_new_with_log(const raft_log_impl_t *log_impl, void *log_arg) ...@@ -628,7 +628,7 @@ raft_server_t* raft_new_with_log(const raft_log_impl_t *log_impl, void *log_arg)
/** De-initialise Raft server. /** De-initialise Raft server.
* Frees all memory */ * Frees all memory */
void raft_free(raft_server_t* me); void raft_destroy(raft_server_t* me_);
/** De-initialise Raft server. */ /** De-initialise Raft server. */
void raft_clear(raft_server_t* me); void raft_clear(raft_server_t* me);
...@@ -660,8 +660,6 @@ void raft_set_callbacks(raft_server_t* me, raft_cbs_t* funcs, void* user_data); ...@@ -660,8 +660,6 @@ void raft_set_callbacks(raft_server_t* me, raft_cbs_t* funcs, void* user_data);
* NULL if the node already exists */ * NULL if the node already exists */
raft_node_t* raft_add_node(raft_server_t* me, void* user_data, raft_node_id_t id, int is_self); raft_node_t* raft_add_node(raft_server_t* me, void* user_data, raft_node_id_t id, int is_self);
#define raft_add_peer raft_add_node
/** Add a node which does not participate in voting. /** Add a node which does not participate in voting.
* If a node already exists the call will fail. * If a node already exists the call will fail.
* Parameters are identical to raft_add_node * Parameters are identical to raft_add_node
...@@ -846,6 +844,13 @@ void* raft_node_get_udata(raft_node_t* me); ...@@ -846,6 +844,13 @@ void* raft_node_get_udata(raft_node_t* me);
* Set this node's user data */ * Set this node's user data */
void raft_node_set_udata(raft_node_t* me, void* user_data); void raft_node_set_udata(raft_node_t* me, void* user_data);
/**
* After sending the snapshot, user can set the next index for the node
*
* @param[in] node node
* @param[in] idx next entry index */
void raft_node_set_next_idx(raft_node_t* me, raft_index_t idx);
/** /**
* @param[in] idx The entry's index * @param[in] idx The entry's index
* @return entry from index */ * @return entry from index */
...@@ -854,13 +859,13 @@ raft_entry_t* raft_get_entry_from_idx(raft_server_t* me, raft_index_t idx); ...@@ -854,13 +859,13 @@ raft_entry_t* raft_get_entry_from_idx(raft_server_t* me, raft_index_t idx);
/** /**
* @param[in] node The node's ID * @param[in] node The node's ID
* @return node pointed to by node ID */ * @return node pointed to by node ID */
raft_node_t* raft_get_node(raft_server_t* me_, const raft_node_id_t id); raft_node_t* raft_get_node(raft_server_t* me_, raft_node_id_t id);
/** /**
* Used for iterating through nodes * Used for iterating through nodes
* @param[in] node The node's idx * @param[in] node The node's idx
* @return node pointed to by node idx */ * @return node pointed to by node idx */
raft_node_t* raft_get_node_from_idx(raft_server_t* me_, const raft_index_t idx); raft_node_t* raft_get_node_from_idx(raft_server_t* me_, raft_index_t idx);
/** /**
* @return number of votes this server has received this election */ * @return number of votes this server has received this election */
...@@ -896,14 +901,14 @@ int raft_vote(raft_server_t* me_, raft_node_t* node); ...@@ -896,14 +901,14 @@ int raft_vote(raft_server_t* me_, raft_node_t* node);
* @param[in] nodeid The server to vote for by nodeid * @param[in] nodeid The server to vote for by nodeid
* @return * @return
* 0 on success */ * 0 on success */
int raft_vote_for_nodeid(raft_server_t* me_, const raft_node_id_t nodeid); int raft_vote_for_nodeid(raft_server_t* me_, raft_node_id_t nodeid);
/** Set the current term. /** Set the current term.
* This should be used to reload persistent state, ie. the current_term field. * This should be used to reload persistent state, ie. the current_term field.
* @param[in] term The new current term * @param[in] term The new current term
* @return * @return
* 0 on success */ * 0 on success */
int raft_set_current_term(raft_server_t* me, const raft_term_t term); int raft_set_current_term(raft_server_t* me, raft_term_t term);
/** Set the commit idx. /** Set the commit idx.
* This should be used to reload persistent state, ie. the commit_idx field. * This should be used to reload persistent state, ie. the commit_idx field.
......
...@@ -43,7 +43,7 @@ int log_poll(log_t * me_, void** etyp); ...@@ -43,7 +43,7 @@ int log_poll(log_t * me_, void** etyp);
/** Get an array of entries from this index onwards. /** Get an array of entries from this index onwards.
* This is used for batching. * This is used for batching.
*/ */
raft_entry_t** log_get_from_idx(log_t* me_, raft_index_t idx, int *n_etys); raft_entry_t** log_get_from_idx(log_t* me_, raft_index_t idx, long *n_etys);
raft_entry_t* log_get_at_idx(log_t* me_, raft_index_t idx); raft_entry_t* log_get_at_idx(log_t* me_, raft_index_t idx);
......
...@@ -134,29 +134,17 @@ int raft_send_appendentries_all(raft_server_t* me_); ...@@ -134,29 +134,17 @@ int raft_send_appendentries_all(raft_server_t* me_);
* @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_);
/**
* Appends entry using the current term.
* Note: we make the assumption that current term is up-to-date
* @return 0 if unsuccessful */
int raft_append_entry(raft_server_t* me_, raft_entry_t* c);
void raft_set_last_applied_idx(raft_server_t* me, raft_index_t idx); void raft_set_last_applied_idx(raft_server_t* me, raft_index_t idx);
void raft_set_state(raft_server_t* me_, int state); void raft_set_state(raft_server_t* me_, int state);
int raft_get_state(raft_server_t* me_);
raft_node_t* raft_node_new(void* udata, raft_node_id_t id); raft_node_t* raft_node_new(void* udata, raft_node_id_t id);
void raft_node_free(raft_node_t* me_); void raft_node_free(raft_node_t* me_);
void raft_node_set_next_idx(raft_node_t* node, raft_index_t nextIdx); void raft_node_set_match_idx(raft_node_t* node, raft_index_t idx);
void raft_node_set_match_idx(raft_node_t* node, raft_index_t matchIdx); void raft_node_vote_for_me(raft_node_t* me_, int vote);
raft_index_t raft_node_get_match_idx(raft_node_t* me_);
void raft_node_vote_for_me(raft_node_t* me_, const int vote);
int raft_node_has_vote_for_me(raft_node_t* me_); int raft_node_has_vote_for_me(raft_node_t* me_);
...@@ -164,32 +152,18 @@ void raft_node_set_has_sufficient_logs(raft_node_t* me_); ...@@ -164,32 +152,18 @@ void raft_node_set_has_sufficient_logs(raft_node_t* me_);
int raft_is_single_node_voting_cluster(raft_server_t *me_); int raft_is_single_node_voting_cluster(raft_server_t *me_);
int raft_votes_is_majority(const int nnodes, const int nvotes); int raft_votes_is_majority(int nnodes, int nvotes);
void raft_handle_append_cfg_change(raft_server_t* me_, raft_entry_t* ety, const raft_index_t idx);
void raft_handle_remove_cfg_change(raft_server_t* me_, raft_entry_t* ety, const raft_index_t idx);
raft_index_t raft_get_num_snapshottable_logs(raft_server_t* me_); raft_index_t raft_get_num_snapshottable_logs(raft_server_t* me_);
int raft_node_is_active(raft_node_t* me_);
void raft_node_set_voting_committed(raft_node_t* me_, int voting);
void raft_node_set_addition_committed(raft_node_t* me_, int committed);
void raft_node_set_last_ack(raft_node_t* me_, raft_msg_id_t msgid, raft_term_t term); void raft_node_set_last_ack(raft_node_t* me_, raft_msg_id_t msgid, raft_term_t term);
raft_msg_id_t raft_node_get_last_acked_msgid(raft_node_t* me_); raft_msg_id_t raft_node_get_last_acked_msgid(raft_node_t* me_);
raft_term_t raft_node_get_last_acked_term(raft_node_t* me_);
void raft_process_read_queue(raft_server_t* me_);
/* Heap functions */ /* Heap functions */
extern void *(*__raft_malloc)(size_t size); extern void *(*raft_malloc)(size_t size);
extern void *(*__raft_calloc)(size_t nmemb, size_t size); extern void *(*raft_calloc)(size_t nmemb, size_t size);
extern void *(*__raft_realloc)(void *ptr, size_t size); extern void *(*raft_realloc)(void *ptr, size_t size);
extern void (*__raft_free)(void *ptr); extern void (*raft_free)(void *ptr);
#endif /* RAFT_PRIVATE_H_ */ #endif /* RAFT_PRIVATE_H_ */
...@@ -8,9 +8,7 @@ ...@@ -8,9 +8,7 @@
* @author Willem Thiart himself@willemthiart.com * @author Willem Thiart himself@willemthiart.com
*/ */
#include <stdlib.h>
#include <string.h> #include <string.h>
#include <stdio.h>
#include <assert.h> #include <assert.h>
#include "raft.h" #include "raft.h"
...@@ -55,7 +53,7 @@ static int __ensurecapacity(log_private_t * me) ...@@ -55,7 +53,7 @@ static int __ensurecapacity(log_private_t * me)
if (me->count < me->size) if (me->count < me->size)
return 0; return 0;
temp = (raft_entry_t**)__raft_calloc(1, sizeof(raft_entry_t *) * me->size * 2); temp = raft_calloc(1, sizeof(raft_entry_t *) * me->size * 2);
if (!temp) if (!temp)
return RAFT_ERR_NOMEM; return RAFT_ERR_NOMEM;
...@@ -67,7 +65,7 @@ static int __ensurecapacity(log_private_t * me) ...@@ -67,7 +65,7 @@ static int __ensurecapacity(log_private_t * me)
} }
/* clean up old entries */ /* clean up old entries */
__raft_free(me->entries); raft_free(me->entries);
me->size *= 2; me->size *= 2;
me->entries = temp; me->entries = temp;
...@@ -89,14 +87,14 @@ int log_load_from_snapshot(log_t *me_, raft_index_t idx, raft_term_t term) ...@@ -89,14 +87,14 @@ int log_load_from_snapshot(log_t *me_, raft_index_t idx, raft_term_t term)
log_t* log_alloc(raft_index_t initial_size) log_t* log_alloc(raft_index_t initial_size)
{ {
log_private_t* me = (log_private_t*)__raft_calloc(1, sizeof(log_private_t)); log_private_t* me = raft_calloc(1, sizeof(log_private_t));
if (!me) if (!me)
return NULL; return NULL;
me->size = initial_size; me->size = initial_size;
log_clear((log_t*)me); log_clear((log_t*)me);
me->entries = (raft_entry_t**)__raft_calloc(1, sizeof(raft_entry_t *) * me->size); me->entries = raft_calloc(1, sizeof(raft_entry_t *) * me->size);
if (!me->entries) { if (!me->entries) {
__raft_free(me); raft_free(me);
return NULL; return NULL;
} }
return (log_t*)me; return (log_t*)me;
...@@ -167,7 +165,7 @@ int log_append_entry(log_t* me_, raft_entry_t* ety) ...@@ -167,7 +165,7 @@ int log_append_entry(log_t* me_, raft_entry_t* ety)
return 0; return 0;
} }
raft_entry_t** log_get_from_idx(log_t* me_, raft_index_t idx, int *n_etys) raft_entry_t** log_get_from_idx(log_t* me_, raft_index_t idx, long *n_etys)
{ {
log_private_t* me = (log_private_t*)me_; log_private_t* me = (log_private_t*)me_;
raft_index_t i; raft_index_t i;
...@@ -185,14 +183,12 @@ raft_entry_t** log_get_from_idx(log_t* me_, raft_index_t idx, int *n_etys) ...@@ -185,14 +183,12 @@ raft_entry_t** log_get_from_idx(log_t* me_, raft_index_t idx, int *n_etys)
i = (me->front + idx - me->base) % me->size; i = (me->front + idx - me->base) % me->size;
int logs_till_end_of_log; /* log entries until the end of the log */
if (i < me->back) if (i < me->back)
logs_till_end_of_log = me->back - i; *n_etys = me->back - i;
else else
logs_till_end_of_log = me->size - i; *n_etys = me->size - i;
*n_etys = logs_till_end_of_log;
return &me->entries[i]; return &me->entries[i];
} }
...@@ -312,8 +308,8 @@ void log_free(log_t * me_) ...@@ -312,8 +308,8 @@ void log_free(log_t * me_)
{ {
log_private_t* me = (log_private_t*)me_; log_private_t* me = (log_private_t*)me_;
__raft_free(me->entries); raft_free(me->entries);
__raft_free(me); raft_free(me);
} }
raft_index_t log_get_current_idx(log_t* me_) raft_index_t log_get_current_idx(log_t* me_)
...@@ -378,7 +374,7 @@ static raft_entry_t *__log_get(void *log, raft_index_t idx) ...@@ -378,7 +374,7 @@ static raft_entry_t *__log_get(void *log, raft_index_t idx)
static int __log_get_batch(void *log, raft_index_t idx, int entries_n, raft_entry_t **entries) static int __log_get_batch(void *log, raft_index_t idx, int entries_n, raft_entry_t **entries)
{ {
int n, i; long n, i;
raft_entry_t **r = log_get_from_idx(log, idx, &n); raft_entry_t **r = log_get_from_idx(log, idx, &n);
if (!r || n < 1) { if (!r || n < 1) {
...@@ -392,7 +388,7 @@ static int __log_get_batch(void *log, raft_index_t idx, int entries_n, raft_entr ...@@ -392,7 +388,7 @@ static int __log_get_batch(void *log, raft_index_t idx, int entries_n, raft_entr
entries[i] = r[i]; entries[i] = r[i];
raft_entry_hold(entries[i]); raft_entry_hold(entries[i]);
} }
return n; return (int) n;
} }
static int __log_pop(void *log, raft_index_t from_idx, func_entry_notify_f cb, void *cb_arg) static int __log_pop(void *log, raft_index_t from_idx, func_entry_notify_f cb, void *cb_arg)
......
...@@ -9,9 +9,7 @@ ...@@ -9,9 +9,7 @@
* @version 0.1 * @version 0.1
*/ */
#include <stdlib.h>
#include <string.h> #include <string.h>
#include <stdio.h>
#include <assert.h> #include <assert.h>
#include "raft.h" #include "raft.h"
...@@ -43,9 +41,11 @@ typedef struct ...@@ -43,9 +41,11 @@ typedef struct
raft_node_t* raft_node_new(void* udata, raft_node_id_t id) raft_node_t* raft_node_new(void* udata, raft_node_id_t id)
{ {
raft_node_private_t* me; raft_node_private_t* me;
me = (raft_node_private_t*)__raft_calloc(1, sizeof(raft_node_private_t));
me = raft_calloc(1, sizeof(raft_node_private_t));
if (!me) if (!me)
return NULL; return NULL;
me->udata = udata; me->udata = udata;
me->next_idx = 1; me->next_idx = 1;
me->match_idx = 0; me->match_idx = 0;
...@@ -56,7 +56,7 @@ raft_node_t* raft_node_new(void* udata, raft_node_id_t id) ...@@ -56,7 +56,7 @@ raft_node_t* raft_node_new(void* udata, raft_node_id_t id)
void raft_node_free(raft_node_t* me_) void raft_node_free(raft_node_t* me_)
{ {
__raft_free(me_); raft_free(me_);
} }
raft_index_t raft_node_get_next_idx(raft_node_t* me_) raft_index_t raft_node_get_next_idx(raft_node_t* me_)
...@@ -65,11 +65,11 @@ raft_index_t raft_node_get_next_idx(raft_node_t* me_) ...@@ -65,11 +65,11 @@ raft_index_t raft_node_get_next_idx(raft_node_t* me_)
return me->next_idx; return me->next_idx;
} }
void raft_node_set_next_idx(raft_node_t* me_, raft_index_t nextIdx) void raft_node_set_next_idx(raft_node_t* me_, raft_index_t idx)
{ {
raft_node_private_t* me = (raft_node_private_t*)me_; raft_node_private_t* me = (raft_node_private_t*)me_;
/* log index begins at 1 */ /* log index begins at 1 */
me->next_idx = nextIdx < 1 ? 1 : nextIdx; me->next_idx = idx < 1 ? 1 : idx;
} }
raft_index_t raft_node_get_match_idx(raft_node_t* me_) raft_index_t raft_node_get_match_idx(raft_node_t* me_)
...@@ -78,10 +78,10 @@ raft_index_t raft_node_get_match_idx(raft_node_t* me_) ...@@ -78,10 +78,10 @@ raft_index_t raft_node_get_match_idx(raft_node_t* me_)
return me->match_idx; return me->match_idx;
} }
void raft_node_set_match_idx(raft_node_t* me_, raft_index_t matchIdx) void raft_node_set_match_idx(raft_node_t* me_, raft_index_t idx)
{ {
raft_node_private_t* me = (raft_node_private_t*)me_; raft_node_private_t* me = (raft_node_private_t*)me_;
me->match_idx = matchIdx; me->match_idx = idx;
} }
void* raft_node_get_udata(raft_node_t* me_) void* raft_node_get_udata(raft_node_t* me_)
...@@ -207,9 +207,3 @@ raft_msg_id_t raft_node_get_last_acked_msgid(raft_node_t* me_) ...@@ -207,9 +207,3 @@ raft_msg_id_t raft_node_get_last_acked_msgid(raft_node_t* me_)
raft_node_private_t* me = (raft_node_private_t*)me_; raft_node_private_t* me = (raft_node_private_t*)me_;
return me->last_acked_msgid; return me->last_acked_msgid;
} }
raft_term_t raft_node_get_last_acked_term(raft_node_t* me_)
{
raft_node_private_t* me = (raft_node_private_t*)me_;
return me->last_acked_term;
}
...@@ -19,27 +19,31 @@ ...@@ -19,27 +19,31 @@
#include "raft_private.h" #include "raft_private.h"
#ifndef min #ifndef min
#define min(a, b) ((a) < (b) ? (a) : (b)) #define min(a, b) ((a) < (b) ? (a) : (b))
#endif #endif
#ifndef max #ifndef max
#define max(a, b) ((a) < (b) ? (b) : (a)) #define max(a, b) ((a) < (b) ? (b) : (a))
#endif #endif
void *(*__raft_malloc)(size_t) = malloc; #ifndef __GNUC__
void *(*__raft_calloc)(size_t, size_t) = calloc; #define __attribute__(a)
void *(*__raft_realloc)(void *, size_t) = realloc; #endif
void (*__raft_free)(void *) = free;
void *(*raft_malloc)(size_t) = malloc;
void *(*raft_calloc)(size_t, size_t) = calloc;
void *(*raft_realloc)(void *, size_t) = realloc;
void (*raft_free)(void *) = free;
void raft_set_heap_functions(void *(*_malloc)(size_t), void raft_set_heap_functions(void *(*_malloc)(size_t),
void *(*_calloc)(size_t, size_t), void *(*_calloc)(size_t, size_t),
void *(*_realloc)(void *, size_t), void *(*_realloc)(void *, size_t),
void (*_free)(void *)) void (*_free)(void *))
{ {
__raft_malloc = _malloc; raft_malloc = _malloc;
__raft_calloc = _calloc; raft_calloc = _calloc;
__raft_realloc = _realloc; raft_realloc = _realloc;
__raft_free = _free; raft_free = _free;
} }
static void raft_log(raft_server_t *me_, raft_node_t* node, const char *fmt, ...) __attribute__ ((format (printf, 3, 4))); static void raft_log(raft_server_t *me_, raft_node_t* node, const char *fmt, ...) __attribute__ ((format (printf, 3, 4)));
...@@ -83,7 +87,7 @@ void raft_update_quorum_meta(raft_server_t* me_, raft_msg_id_t id) ...@@ -83,7 +87,7 @@ void raft_update_quorum_meta(raft_server_t* me_, raft_msg_id_t id)
raft_server_t* raft_new_with_log(const raft_log_impl_t *log_impl, void *log_arg) raft_server_t* raft_new_with_log(const raft_log_impl_t *log_impl, void *log_arg)
{ {
raft_server_private_t* me = __raft_calloc(1, sizeof(raft_server_private_t)); raft_server_private_t* me = raft_calloc(1, sizeof(raft_server_private_t));
if (!me) if (!me)
return NULL; return NULL;
...@@ -99,7 +103,7 @@ raft_server_t* raft_new_with_log(const raft_log_impl_t *log_impl, void *log_arg) ...@@ -99,7 +103,7 @@ raft_server_t* raft_new_with_log(const raft_log_impl_t *log_impl, void *log_arg)
me->log_impl = log_impl; me->log_impl = log_impl;
me->log = me->log_impl->init(me, log_arg); me->log = me->log_impl->init(me, log_arg);
if (!me->log) { if (!me->log) {
__raft_free(me); raft_free(me);
return NULL; return NULL;
} }
...@@ -113,8 +117,6 @@ raft_server_t* raft_new_with_log(const raft_log_impl_t *log_impl, void *log_arg) ...@@ -113,8 +117,6 @@ raft_server_t* raft_new_with_log(const raft_log_impl_t *log_impl, void *log_arg)
return (raft_server_t*)me; return (raft_server_t*)me;
} }
extern raft_log_impl_t log_internal_impl;
raft_server_t* raft_new(void) raft_server_t* raft_new(void)
{ {
return raft_new_with_log(&raft_log_internal_impl, NULL); return raft_new_with_log(&raft_log_internal_impl, NULL);
...@@ -128,12 +130,12 @@ void raft_set_callbacks(raft_server_t* me_, raft_cbs_t* funcs, void* udata) ...@@ -128,12 +130,12 @@ void raft_set_callbacks(raft_server_t* me_, raft_cbs_t* funcs, void* udata)
me->udata = udata; me->udata = udata;
} }
void raft_free(raft_server_t* me_) void raft_destroy(raft_server_t* me_)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
me->log_impl->free(me->log); me->log_impl->free(me->log);
__raft_free(me_); raft_free(me_);
} }
void raft_clear(raft_server_t* me_) void raft_clear(raft_server_t* me_)
...@@ -154,6 +156,185 @@ void raft_clear(raft_server_t* me_) ...@@ -154,6 +156,185 @@ void raft_clear(raft_server_t* me_)
me->log_impl->reset(me->log, 1, 1); me->log_impl->reset(me->log, 1, 1);
} }
raft_node_t* raft_add_node_internal(raft_server_t* me_, raft_entry_t *ety, void* udata, raft_node_id_t id, int is_self)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
/* set to voting if node already exists */
raft_node_t* node = raft_get_node(me_, id);
if (node)
{
if (!raft_node_is_voting(node))
{
raft_node_set_voting(node, 1);
return node;
}
else
/* we shouldn't add a node twice */
return NULL;
}
node = raft_node_new(udata, id);
if (!node)
return NULL;
void* p = raft_realloc(me->nodes, sizeof(void*) * (me->num_nodes + 1));
if (!p) {
raft_node_free(node);
return NULL;
}
me->num_nodes++;
me->nodes = p;
me->nodes[me->num_nodes - 1] = node;
if (is_self)
me->node = me->nodes[me->num_nodes - 1];
node = me->nodes[me->num_nodes - 1];
if (me->cb.notify_membership_event)
me->cb.notify_membership_event(me_, raft_get_udata(me_), node, ety, RAFT_MEMBERSHIP_ADD);
return node;
}
static raft_node_t* raft_add_non_voting_node_internal(raft_server_t* me_, raft_entry_t *ety, void* udata, raft_node_id_t id, int is_self)
{
if (raft_get_node(me_, id))
return NULL;
raft_node_t* node = raft_add_node_internal(me_, ety, udata, id, is_self);
if (!node)
return NULL;
raft_node_set_voting(node, 0);
return node;
}
raft_node_t* raft_add_node(raft_server_t* me_, void* udata, raft_node_id_t id, int is_self)
{
return raft_add_node_internal(me_, NULL, udata, id, is_self);
}
raft_node_t* raft_add_non_voting_node(raft_server_t* me_, void* udata, raft_node_id_t id, int is_self)
{
return raft_add_non_voting_node_internal(me_, NULL, udata, id, is_self);
}
void raft_remove_node(raft_server_t* me_, raft_node_t* node)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
if (me->cb.notify_membership_event)
me->cb.notify_membership_event(me_, raft_get_udata(me_), node, NULL, RAFT_MEMBERSHIP_REMOVE);
assert(node);
int i, found = 0;
for (i = 0; i < me->num_nodes; i++)
{
if (me->nodes[i] == node)
{
found = 1;
break;
}
}
assert(found);
memmove(&me->nodes[i], &me->nodes[i + 1], sizeof(*me->nodes) * (me->num_nodes - i - 1));
me->num_nodes--;
/* if we are removing the current leader, have to reset it, as this data is now stale */
raft_node_t *current_leader_node = raft_get_current_leader_node(me_);
if (node == current_leader_node) {
me->current_leader = NULL;
}
raft_node_free(node);
}
void raft_handle_append_cfg_change(raft_server_t* me_, raft_entry_t* ety, raft_index_t idx)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
if (!raft_entry_is_cfg_change(ety))
return;
if (!me->cb.log_get_node_id)
return;
void* udata = raft_get_udata(me_);
raft_node_id_t node_id = me->cb.log_get_node_id(me_, udata, ety, idx);
raft_node_t* node = raft_get_node(me_, node_id);
int is_self = node_id == raft_get_nodeid(me_);
switch (ety->type)
{
case RAFT_LOGTYPE_ADD_NONVOTING_NODE:
if (!is_self)
{
if (node && !raft_node_is_active(node))
{
raft_node_set_active(node, 1);
}
else if (!node)
{
node = raft_add_non_voting_node_internal(me_, ety, NULL, node_id, is_self);
assert(node);
}
}
break;
case RAFT_LOGTYPE_ADD_NODE:
node = raft_add_node_internal(me_, ety, NULL, node_id, is_self);
assert(node);
assert(raft_node_is_voting(node));
break;
case RAFT_LOGTYPE_REMOVE_NODE:
if (node) {
raft_node_set_active(node, 0);
}
break;
default:
assert(0);
}
}
void raft_handle_remove_cfg_change(raft_server_t* me_, raft_entry_t* ety, const raft_index_t idx)
{
raft_server_private_t* me = (raft_server_private_t*) me_;
if (!raft_entry_is_cfg_change(ety))
return;
if (!me->cb.log_get_node_id)
return;
void* udata = raft_get_udata(me_);
raft_node_id_t node_id = me->cb.log_get_node_id(me_, udata, ety, idx);
raft_node_t* node = raft_get_node(me_, node_id);
switch (ety->type)
{
case RAFT_LOGTYPE_REMOVE_NODE:
raft_node_set_active(node, 1);
break;
case RAFT_LOGTYPE_ADD_NONVOTING_NODE:
assert(node_id != raft_get_nodeid(me_));
raft_remove_node(me_, node);
break;
case RAFT_LOGTYPE_ADD_NODE:
raft_node_set_voting(node, 0);
break;
default:
assert(0);
break;
}
}
int raft_delete_entry_from_idx(raft_server_t* me_, raft_index_t idx) int raft_delete_entry_from_idx(raft_server_t* me_, raft_index_t idx)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
...@@ -489,10 +670,10 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -489,10 +670,10 @@ int raft_recv_appendentries_response(raft_server_t* me_,
int votes = raft_node_is_voting(me->node) ? 1 : 0; int votes = raft_node_is_voting(me->node) ? 1 : 0;
for (int i = 0; i < me->num_nodes; i++) for (int i = 0; i < me->num_nodes; i++)
{ {
raft_node_t* node = me->nodes[i]; raft_node_t* follower = me->nodes[i];
if (me->node != node && if (me->node != follower &&
raft_node_is_voting(node) && raft_node_is_voting(follower) &&
point <= raft_node_get_match_idx(node)) point <= raft_node_get_match_idx(follower))
{ {
votes++; votes++;
} }
...@@ -675,7 +856,7 @@ int raft_already_voted(raft_server_t* me_) ...@@ -675,7 +856,7 @@ int raft_already_voted(raft_server_t* me_)
return ((raft_server_private_t*)me_)->voted_for != -1; return ((raft_server_private_t*)me_)->voted_for != -1;
} }
static int __should_grant_vote(raft_server_t* me_, msg_requestvote_t* vr) static int raft_should_grant_vote(raft_server_t* me_, msg_requestvote_t* vr)
{ {
if (!raft_node_is_voting(raft_get_my_node(me_))) if (!raft_node_is_voting(raft_get_my_node(me_)))
return 0; return 0;
...@@ -735,7 +916,7 @@ int raft_recv_requestvote(raft_server_t* me_, ...@@ -735,7 +916,7 @@ int raft_recv_requestvote(raft_server_t* me_,
me->current_leader = NULL; me->current_leader = NULL;
} }
if (__should_grant_vote(me_, vr)) if (raft_should_grant_vote(me_, vr))
{ {
/* It shouldn't be possible for a leader or candidate to grant a vote /* It shouldn't be possible for a leader or candidate to grant a vote
* Both states would have voted for themselves */ * Both states would have voted for themselves */
...@@ -749,7 +930,6 @@ int raft_recv_requestvote(raft_server_t* me_, ...@@ -749,7 +930,6 @@ int raft_recv_requestvote(raft_server_t* me_,
/* must be in an election. */ /* must be in an election. */
me->current_leader = NULL; me->current_leader = NULL;
me->timeout_elapsed = 0; me->timeout_elapsed = 0;
} }
else else
...@@ -796,11 +976,12 @@ int raft_recv_requestvote_response(raft_server_t* me_, ...@@ -796,11 +976,12 @@ int raft_recv_requestvote_response(raft_server_t* me_,
r->vote_granted == 1 ? "granted" : r->vote_granted == 1 ? "granted" :
r->vote_granted == 0 ? "not granted" : "unknown"); r->vote_granted == 0 ? "not granted" : "unknown");
if (!raft_is_candidate(me_)) if (!raft_is_candidate(me_) || raft_get_current_term(me_) > r->term)
{ {
return 0; return 0;
} }
else if (raft_get_current_term(me_) < r->term)
if (raft_get_current_term(me_) < r->term)
{ {
int e = raft_set_current_term(me_, r->term); int e = raft_set_current_term(me_, r->term);
if (0 != e) if (0 != e)
...@@ -809,13 +990,6 @@ int raft_recv_requestvote_response(raft_server_t* me_, ...@@ -809,13 +990,6 @@ int raft_recv_requestvote_response(raft_server_t* me_,
me->current_leader = NULL; me->current_leader = NULL;
return 0; return 0;
} }
else if (raft_get_current_term(me_) != r->term)
{
/* The node who voted for us would have obtained our term.
* Therefore this is an old message we should ignore.
* This happens if the network is pretty choppy. */
return 0;
}
raft_log(me_, node, "node responded to requestvote status:%s ct:%ld rt:%ld", raft_log(me_, node, "node responded to requestvote status:%s ct:%ld rt:%ld",
r->vote_granted == 1 ? "granted" : r->vote_granted == 1 ? "granted" :
...@@ -1029,12 +1203,12 @@ raft_entry_t** raft_get_entries_from_idx(raft_server_t* me_, raft_index_t idx, i ...@@ -1029,12 +1203,12 @@ raft_entry_t** raft_get_entries_from_idx(raft_server_t* me_, raft_index_t idx, i
} }
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
int size = raft_get_current_idx(me_) - idx + 1; raft_index_t size = raft_get_current_idx(me_) - idx + 1;
raft_entry_t **e = __raft_malloc(size * sizeof(raft_entry_t*)); raft_entry_t **e = raft_malloc(size * sizeof(raft_entry_t*));
int n = me->log_impl->get_batch(me->log, idx, size, e); int n = me->log_impl->get_batch(me->log, idx, (int) size, e);
if (n < 1) { if (n < 1) {
__raft_free(e); raft_free(e);
*n_etys = 0; *n_etys = 0;
return NULL; return NULL;
} }
...@@ -1103,7 +1277,7 @@ int raft_send_appendentries(raft_server_t* me_, raft_node_t* node) ...@@ -1103,7 +1277,7 @@ int raft_send_appendentries(raft_server_t* me_, raft_node_t* node)
int res = me->cb.send_appendentries(me_, me->udata, node, &ae); int res = me->cb.send_appendentries(me_, me->udata, node, &ae);
raft_entry_release_list(ae.entries, ae.n_entries); raft_entry_release_list(ae.entries, ae.n_entries);
__raft_free(ae.entries); raft_free(ae.entries);
return res; return res;
} }
...@@ -1128,100 +1302,6 @@ int raft_send_appendentries_all(raft_server_t* me_) ...@@ -1128,100 +1302,6 @@ int raft_send_appendentries_all(raft_server_t* me_)
return ret; return ret;
} }
raft_node_t* raft_add_node_internal(raft_server_t* me_, raft_entry_t *ety, void* udata, raft_node_id_t id, int is_self)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
/* set to voting if node already exists */
raft_node_t* node = raft_get_node(me_, id);
if (node)
{
if (!raft_node_is_voting(node))
{
raft_node_set_voting(node, 1);
return node;
}
else
/* we shouldn't add a node twice */
return NULL;
}
node = raft_node_new(udata, id);
if (!node)
return NULL;
void* p = __raft_realloc(me->nodes, sizeof(void*) * (me->num_nodes + 1));
if (!p) {
raft_node_free(node);
return NULL;
}
me->num_nodes++;
me->nodes = p;
me->nodes[me->num_nodes - 1] = node;
if (is_self)
me->node = me->nodes[me->num_nodes - 1];
node = me->nodes[me->num_nodes - 1];
if (me->cb.notify_membership_event)
me->cb.notify_membership_event(me_, raft_get_udata(me_), node, ety, RAFT_MEMBERSHIP_ADD);
return node;
}
raft_node_t* raft_add_node(raft_server_t* me_, void* udata, raft_node_id_t id, int is_self)
{
return raft_add_node_internal(me_, NULL, udata, id, is_self);
}
static raft_node_t* raft_add_non_voting_node_internal(raft_server_t* me_, raft_entry_t *ety, void* udata, raft_node_id_t id, int is_self)
{
if (raft_get_node(me_, id))
return NULL;
raft_node_t* node = raft_add_node_internal(me_, ety, udata, id, is_self);
if (!node)
return NULL;
raft_node_set_voting(node, 0);
return node;
}
raft_node_t* raft_add_non_voting_node(raft_server_t* me_, void* udata, raft_node_id_t id, int is_self)
{
return raft_add_non_voting_node_internal(me_, NULL, udata, id, is_self);
}
void raft_remove_node(raft_server_t* me_, raft_node_t* node)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
if (me->cb.notify_membership_event)
me->cb.notify_membership_event(me_, raft_get_udata(me_), node, NULL, RAFT_MEMBERSHIP_REMOVE);
assert(node);
int i, found = 0;
for (i = 0; i < me->num_nodes; i++)
{
if (me->nodes[i] == node)
{
found = 1;
break;
}
}
assert(found);
memmove(&me->nodes[i], &me->nodes[i + 1], sizeof(*me->nodes) * (me->num_nodes - i - 1));
me->num_nodes--;
/* if we are removing the current leader, have to reset it, as this data is now stale */
raft_node_t *current_leader_node = raft_get_current_leader_node(me_);
if (node == current_leader_node) {
me->current_leader = NULL;
}
raft_node_free(node);
}
int raft_get_nvotes_for_me(raft_server_t* me_) int raft_get_nvotes_for_me(raft_server_t* me_)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
...@@ -1305,99 +1385,6 @@ int raft_entry_is_cfg_change(raft_entry_t* ety) ...@@ -1305,99 +1385,6 @@ int raft_entry_is_cfg_change(raft_entry_t* ety)
RAFT_LOGTYPE_REMOVE_NODE == ety->type); RAFT_LOGTYPE_REMOVE_NODE == ety->type);
} }
void raft_handle_append_cfg_change(raft_server_t* me_, raft_entry_t* ety, raft_index_t idx)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
if (!raft_entry_is_cfg_change(ety))
return;
if (!me->cb.log_get_node_id)
return;
raft_node_id_t node_id = me->cb.log_get_node_id(me_, raft_get_udata(me_), ety, idx);
raft_node_t* node = raft_get_node(me_, node_id);
int is_self = node_id == raft_get_nodeid(me_);
switch (ety->type)
{
case RAFT_LOGTYPE_ADD_NONVOTING_NODE:
if (!is_self)
{
if (node && !raft_node_is_active(node))
{
raft_node_set_active(node, 1);
}
else if (!node)
{
node = raft_add_non_voting_node_internal(me_, ety, NULL, node_id, is_self);
assert(node);
}
}
break;
case RAFT_LOGTYPE_ADD_NODE:
node = raft_add_node_internal(me_, ety, NULL, node_id, is_self);
assert(node);
assert(raft_node_is_voting(node));
break;
case RAFT_LOGTYPE_REMOVE_NODE:
if (node) {
raft_node_set_active(node, 0);
}
break;
default:
assert(0);
}
}
void raft_handle_remove_cfg_change(raft_server_t* me_, raft_entry_t* ety, const raft_index_t idx)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
if (!raft_entry_is_cfg_change(ety))
return;
if (!me->cb.log_get_node_id)
return;
raft_node_id_t node_id = me->cb.log_get_node_id(me_, raft_get_udata(me_), ety, idx);
switch (ety->type)
{
case RAFT_LOGTYPE_REMOVE_NODE:
{
raft_node_t* node = raft_get_node(me_, node_id);
raft_node_set_active(node, 1);
}
break;
case RAFT_LOGTYPE_ADD_NONVOTING_NODE:
{
int is_self = node_id == raft_get_nodeid(me_);
raft_node_t* node = raft_get_node(me_, node_id);
raft_remove_node(me_, node);
if (is_self)
assert(0);
}
break;
case RAFT_LOGTYPE_ADD_NODE:
{
raft_node_t* node = raft_get_node(me_, node_id);
raft_node_set_voting(node, 0);
}
break;
default:
assert(0);
break;
}
}
int raft_poll_entry(raft_server_t* me_) int raft_poll_entry(raft_server_t* me_)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
...@@ -1450,7 +1437,7 @@ int raft_begin_snapshot(raft_server_t *me_, int flags) ...@@ -1450,7 +1437,7 @@ int raft_begin_snapshot(raft_server_t *me_, int flags)
return -1; return -1;
raft_index_t snapshot_target = raft_get_commit_idx(me_); raft_index_t snapshot_target = raft_get_commit_idx(me_);
if (!snapshot_target || snapshot_target == 0) if (!snapshot_target)
return -1; return -1;
raft_entry_t* ety = raft_get_entry_from_idx(me_, snapshot_target); raft_entry_t* ety = raft_get_entry_from_idx(me_, snapshot_target);
...@@ -1632,7 +1619,7 @@ void *raft_get_log(raft_server_t *me_) ...@@ -1632,7 +1619,7 @@ void *raft_get_log(raft_server_t *me_)
raft_entry_t *raft_entry_new(unsigned int data_len) raft_entry_t *raft_entry_new(unsigned int data_len)
{ {
raft_entry_t *ety = __raft_calloc(1, sizeof(raft_entry_t) + data_len); raft_entry_t *ety = raft_calloc(1, sizeof(raft_entry_t) + data_len);
ety->data_len = data_len; ety->data_len = data_len;
ety->refs = 1; ety->refs = 1;
...@@ -1654,7 +1641,7 @@ void raft_entry_release(raft_entry_t *ety) ...@@ -1654,7 +1641,7 @@ void raft_entry_release(raft_entry_t *ety)
if (ety->free_func) { if (ety->free_func) {
ety->free_func(ety); ety->free_func(ety);
} else { } else {
__raft_free(ety); raft_free(ety);
} }
} }
} }
...@@ -1672,7 +1659,7 @@ void raft_queue_read_request(raft_server_t* me_, func_read_request_callback_f cb ...@@ -1672,7 +1659,7 @@ void raft_queue_read_request(raft_server_t* me_, func_read_request_callback_f cb
{ {
raft_server_private_t* me = (raft_server_private_t*) me_; raft_server_private_t* me = (raft_server_private_t*) me_;
raft_read_request_t *req = __raft_malloc(sizeof(raft_read_request_t)); raft_read_request_t *req = raft_malloc(sizeof(raft_read_request_t));
req->read_idx = raft_get_current_idx(me_); req->read_idx = raft_get_current_idx(me_);
req->read_term = raft_get_current_term(me_); req->read_term = raft_get_current_term(me_);
...@@ -1706,7 +1693,7 @@ static void pop_read_queue(raft_server_private_t *me, int can_read) ...@@ -1706,7 +1693,7 @@ static void pop_read_queue(raft_server_private_t *me, int can_read)
me->read_queue_tail = NULL; me->read_queue_tail = NULL;
} }
__raft_free(p); raft_free(p);
} }
void raft_process_read_queue(raft_server_t* me_) void raft_process_read_queue(raft_server_t* me_)
......
...@@ -511,7 +511,7 @@ void TestLog_get_from_idx_with_base_off_by_one(CuTest * tc) ...@@ -511,7 +511,7 @@ void TestLog_get_from_idx_with_base_off_by_one(CuTest * tc)
CuAssertIntEquals(tc, 1, log_count(l)); CuAssertIntEquals(tc, 1, log_count(l));
/* get off-by-one index */ /* get off-by-one index */
int n_etys; long n_etys;
CuAssertPtrEquals(tc, log_get_from_idx(l, 1, &n_etys), NULL); CuAssertPtrEquals(tc, log_get_from_idx(l, 1, &n_etys), NULL);
CuAssertIntEquals(tc, n_etys, 0); CuAssertIntEquals(tc, n_etys, 0);
......
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