Commit bdd49a1e authored by Yossi Gottlieb's avatar Yossi Gottlieb
Browse files

Pluggable Log Implementation.

parent 863fb439
...@@ -54,7 +54,7 @@ static: $(OBJECTS) ...@@ -54,7 +54,7 @@ static: $(OBJECTS)
ar -r libraft.a $(OBJECTS) ar -r libraft.a $(OBJECTS)
.PHONY: tests .PHONY: tests
tests: src/raft_server.c src/raft_server_properties.c src/raft_log.c src/raft_node.c $(TEST_DIR)/main_test.c $(TEST_DIR)/test_server.c $(TEST_DIR)/test_node.c $(TEST_DIR)/test_log.c $(TEST_DIR)/test_snapshotting.c $(TEST_DIR)/test_scenario.c $(TEST_DIR)/mock_send_functions.c $(TEST_DIR)/CuTest.c $(LLQUEUE_DIR)/linked_list_queue.c tests: src/raft_server.c src/raft_server_properties.c src/raft_log.c src/raft_node.c $(TEST_DIR)/main_test.c $(TEST_DIR)/test_server.c $(TEST_DIR)/test_node.c $(TEST_DIR)/test_log.c $(TEST_DIR)/test_log_impl.c $(TEST_DIR)/test_snapshotting.c $(TEST_DIR)/test_scenario.c $(TEST_DIR)/mock_send_functions.c $(TEST_DIR)/CuTest.c $(LLQUEUE_DIR)/linked_list_queue.c
$(CC) $(CFLAGS) -o tests_main $^ $(CC) $(CFLAGS) -o tests_main $^
./tests_main ./tests_main
gcov raft_server.c gcov raft_server.c
......
...@@ -98,12 +98,19 @@ typedef struct ...@@ -98,12 +98,19 @@ typedef struct
raft_entry_id_t id; raft_entry_id_t id;
/** type of entry */ /** type of entry */
int type; short type;
/** number of references */
short refs;
/** private local data */ /** private local data */
void *user_data; void *user_data;
raft_entry_data_t data; /** data length */
unsigned int data_len;
/** data */
char data[0];
} raft_entry_t; } raft_entry_t;
/** Message sent from client to server. /** Message sent from client to server.
...@@ -178,8 +185,8 @@ typedef struct ...@@ -178,8 +185,8 @@ typedef struct
/** number of entries within this message */ /** number of entries within this message */
int n_entries; int n_entries;
/** array of entries within this message */ /** array of pointers to entries within this message */
msg_entry_t* entries; msg_entry_t** entries;
} msg_appendentries_t; } msg_appendentries_t;
/** Appendentries response message. /** Appendentries response message.
...@@ -392,30 +399,6 @@ typedef struct ...@@ -392,30 +399,6 @@ typedef struct
* disk atomically. */ * disk atomically. */
func_persist_term_f persist_term; func_persist_term_f persist_term;
/** Callback for adding an entry to the log
* For safety reasons this callback MUST flush the change to disk.
* Return 0 on success.
* Return RAFT_ERR_SHUTDOWN if you want the server to shutdown. */
func_logentry_event_f log_offer;
/** Callback for removing the oldest entry from the log
* For safety reasons this callback MUST flush the change to disk.
* @note If memory was malloc'd in log_offer then this should be the right
* time to free the memory. */
func_logentry_event_f log_poll;
/** Callback for removing the youngest entry from the log
* For safety reasons this callback MUST flush the change to disk.
* @note If memory was malloc'd in log_offer then this should be the right
* time to free the memory. */
func_logentry_event_f log_pop;
/** Callback called for every existing log entry when clearing the log.
* If memory was malloc'd in log_offer and the entry doesn't get a chance
* to go through log_poll or log_pop, this is the last chance to free it.
*/
func_logentry_event_f log_clear;
/** Callback for determining which node this configuration log entry /** Callback for determining which node this configuration log entry
* affects. This call only applies to configuration change log entries. * affects. This call only applies to configuration change log entries.
* @return the node ID of the node */ * @return the node ID of the node */
...@@ -431,6 +414,165 @@ typedef struct ...@@ -431,6 +414,165 @@ typedef struct
func_log_f log; func_log_f log;
} raft_cbs_t; } raft_cbs_t;
/** A generic notification callback used to allow Raft to notify caller
* on certain log operations.
*
* @param[in] arg Argument passed by Raft in the original call.
* @param[in] entry Entry for which notification is generated.
* @param[in] entry_idx Index of entry.
*
* The callback *must not* modify the entry or perform any preemptive
* log operation until it returns.
*/
typedef void (
*func_entry_notify_f
) (
void* arg,
raft_entry_t *entry,
raft_index_t entry_idx
);
/** Generic Raft Log implementation.
*
* This is an abstract interface that can be used to implement pluggable
* Raft Log implementations, unlike the built-in implementation which is
* more opinionated (e.g. is entirely in-memory, etc.).
*
* The log implementation is expected to be persistent, so it must avoid
* losing entries that have been appended to it.
*/
typedef struct raft_log_impl
{
/** Log implementation construction, called exactly once when Raft
* initializes.
*
* @param[in] raft The Raft server using the log.
* @param[in] arg User-specified initialization argument, as passed to
* raft_new().
* @return Initialized log handle. This handle is passed as 'log' on
* all subsequent calls.
*
* @note A common pattern may involve initializing the log engine
* in advance and passing a handle to it as arg. The init function
* can then simply return arg.
*/
void *(*init) (void *raft, void *arg);
/** Log implementation destruction, called exactly once when Raft
* shuts down.
*
* All memory and resources allocated since init() should be released.
*
* @param[in] log The log handle.
*/
void (*free) (void *log);
/** Reset log. All entries should be deleted, and the log is configured
* such that the next appended log entry would be assigned with the
* specified index.
*
* A log implementation that has been initialized for the first time and
* contains no persisted data should implicitly perform reset(1).
*
* A reset operation with a higher first_idx is expected when the log
* is compacted after a snapshot is taken. In this case the log
* implementation is expected to persist the index and term.
*
* @param[in] first_idx Index to assign to the first entry in the log.
* @param[in] term Term of last applied entry, if reset is called after
* a snapshot.
*/
void (*reset) (void *log, raft_index_t first_idx, raft_term_t term);
/** Append an entry to the log.
* @param[in] entry Entry to append.
* @return
* 0 on success;
* RAFT_ERR_SHUTDOWN server should shutdown;
* RAFT_ERR_NOMEM memory allocation failure.
*
* @note
* The passed raft_entry_t is expected to be allocated on the heap.
* The log implementation shall call raft_entry_hold() in order to
* maintain its reference count, and call raft_entry_release() when
* the entry is no longer needed.
*
* @todo
* 1. Batch append of multiple entries.
* 2. Consider an async option to make it possible to implement
* I/O in a background thread.
*/
int (*append) (void *log, raft_entry_t *entry);
/** Remove entries from the start of the log, as necessary when compacting
* the log and deleting the oldest entries.
*
* @param[in] first_idx Index of first entry to be left in log.
* @return
* 0 on success;
* -1 on error (e.g. log is empty).
*/
int (*poll) (void *log, raft_index_t first_idx);
/** Remove entries from the end of the log, as necessary when rolling back
* append operations that have not been committed.
*
* @param[in] from_idx Index of first entry to be removed. All entries
* starting from and including this index shall be removed.
* @param[in] cb Optional callback to execute for every removed entry.
* @param[in] cb_arg Argument to pass to callback.
* @return
* 0 on success;
* -1 on error.
*/
int (*pop) (void *log, raft_index_t from_idx, func_entry_notify_f cb, void *cb_arg);
/** Get a single entry from the log.
*
* @param[in] idx Index of entry to fetch.
* @return
* Pointer to entry on success;
* NULL if no entry in specified index.
*
* @note
* Returned entry memory is owned by the log and may be freed on next log
* operation, so caller must make a copy if it is going to be retained beyond
* this scope.
*/
raft_entry_t* (*get) (void *log, raft_index_t idx);
/** Get a batch of entries from the log.
*
* @param[in] idx Index of first entry to fetch.
* @param[in] entries_n Length of entries (max. entries to fetch).
* @param[out] entries An initialized array of raft_entry_t*.
* @return
* Number of entries fetched;
* -1 on error.
*/
int (*get_batch) (void *log, raft_index_t idx, int entries_n,
raft_entry_t **entries);
/** Get first entry's index.
* @return
* Index of first entry.
*/
raft_index_t (*first_idx) (void *log);
/** Get current (latest) entry's index.
* @return
* Index of latest entry.
*/
raft_index_t (*current_idx) (void *log);
/** Get number of entries in the log.
* @return
* Number of entries.
*/
raft_index_t (*count) (void *log);
} raft_log_impl_t;
typedef struct typedef struct
{ {
/** User data pointer for addressing. /** User data pointer for addressing.
...@@ -440,13 +582,22 @@ typedef struct ...@@ -440,13 +582,22 @@ typedef struct
void* udata_address; void* udata_address;
} raft_node_configuration_t; } raft_node_configuration_t;
/** Initialise a new Raft server. /** Initialise a new Raft server, using the in-memory log implementation.
* *
* Request timeout defaults to 200 milliseconds * Request timeout defaults to 200 milliseconds
* Election timeout defaults to 1000 milliseconds * Election timeout defaults to 1000 milliseconds
* *
* @return newly initialised Raft server */ * @return newly initialised Raft server */
raft_server_t* raft_new(void); raft_server_t* raft_new();
/** Initializes a new Raft server with a custom Raft Log implementation.
*
* @param[in] log_impl Callbacks structure for the Log implementation to use.
* @param[in] log_arg Argument to pass to Log implementation's init().
*
* @return newly initialised Raft server
*/
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 */
...@@ -873,7 +1024,7 @@ int raft_is_apply_allowed(raft_server_t* me_); ...@@ -873,7 +1024,7 @@ int raft_is_apply_allowed(raft_server_t* me_);
* This should be used for compacting logs. * This should be used for compacting logs.
* @return 0 on success * @return 0 on success
**/ **/
int raft_poll_entry(raft_server_t* me_, raft_entry_t **ety); int raft_poll_entry(raft_server_t* me_);
/** Get last applied entry /** Get last applied entry
**/ **/
...@@ -964,4 +1115,79 @@ void raft_node_set_addition_committed(raft_node_t* me_, int committed); ...@@ -964,4 +1115,79 @@ void raft_node_set_addition_committed(raft_node_t* me_, int committed);
* @return 1 if a voting change is in progress */ * @return 1 if a voting change is in progress */
int raft_voting_change_is_in_progress(raft_server_t* me_); int raft_voting_change_is_in_progress(raft_server_t* me_);
/** Get the log implementation handle in use.
*/
void *raft_get_log(raft_server_t* me_);
/** Backward compatible callbacks for log events, implemented by the
* default in-memory log implementation.
*
*/
typedef struct {
/** Callback for adding an entry to the log
* For safety reasons this callback MUST flush the change to disk.
* Return 0 on success.
* Return RAFT_ERR_SHUTDOWN if you want the server to shutdown. */
func_logentry_event_f log_offer;
/** Callback for removing the oldest entry from the log
* For safety reasons this callback MUST flush the change to disk.
* @note If memory was malloc'd in log_offer then this should be the right
* time to free the memory. */
func_logentry_event_f log_poll;
/** Callback for removing the youngest entry from the log
* For safety reasons this callback MUST flush the change to disk.
* @note If memory was malloc'd in log_offer then this should be the right
* time to free the memory. */
func_logentry_event_f log_pop;
/** Callback called for every existing log entry when clearing the log.
* If memory was malloc'd in log_offer and the entry doesn't get a chance
* to go through log_poll or log_pop, this is the last chance to free it.
*/
func_logentry_event_f log_clear;
} raft_log_cbs_t;
/** Allocate a new Raft Log entry.
*
* @param[in] data_len Length of user-supplied data for which additional
* memory should be allocated at the end of the entry.
* @returns
* Entry pointer (heap allocated).
*
* @note All raft_entry_t elements are reference counted and created with an
* initial refcount value of 1. Calling raft_entry_release() immediately would
* therefore result with deallocation.
*/
raft_entry_t *raft_entry_new(unsigned int data_len);
/** Hold the raft_entry_t, i.e. increment refcount by one.
*/
void raft_entry_hold(raft_entry_t *ety);
/** Release the raft_entry_t, i.e. decrement refcount by one and free
* if refcount reaches zero.
*/
void raft_entry_release(raft_entry_t *ety);
/** Iterate an array of raft_entry_t* and release each element.
*
* @param[in] ety_list A pointer to a raft_entry_t* array.
* @param[in] len Number of entries in the array.
*
* @note The array itself is not freed.
*/
void raft_entry_release_list(raft_entry_t **ety_list, size_t len);
/** Log Implementation callbacks structure for the default in-memory
* log implementation.
*/
extern const raft_log_impl_t raft_log_internal_impl;
void raft_handle_append_cfg_change(raft_server_t* me_, raft_entry_t* ety, raft_index_t idx);
#endif /* RAFT_H_ */ #endif /* RAFT_H_ */
...@@ -9,7 +9,7 @@ log_t* log_new(void); ...@@ -9,7 +9,7 @@ log_t* log_new(void);
log_t* log_alloc(raft_index_t initial_size); log_t* log_alloc(raft_index_t initial_size);
void log_set_callbacks(log_t* me_, raft_cbs_t* funcs, void* raft); void log_set_callbacks(log_t* me_, raft_log_cbs_t* funcs, void* raft);
void log_free(log_t* me_); void log_free(log_t* me_);
...@@ -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, int *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);
......
...@@ -19,6 +19,8 @@ enum { ...@@ -19,6 +19,8 @@ enum {
RAFT_NODE_STATUS_DISCONNECTING RAFT_NODE_STATUS_DISCONNECTING
}; };
struct raft_log_impl;
typedef struct { typedef struct {
/* Persistent state: */ /* Persistent state: */
...@@ -30,8 +32,9 @@ typedef struct { ...@@ -30,8 +32,9 @@ typedef struct {
* or Nil if it hasn't voted for any. */ * or Nil if it hasn't voted for any. */
raft_node_id_t voted_for; raft_node_id_t voted_for;
/* the log which is replicated */ /* log storage engine */
void* log; const struct raft_log_impl *log_impl;
void *log;
/* Volatile state: */ /* Volatile state: */
...@@ -135,9 +138,9 @@ void raft_node_set_has_sufficient_logs(raft_node_t* me_); ...@@ -135,9 +138,9 @@ void raft_node_set_has_sufficient_logs(raft_node_t* me_);
int raft_votes_is_majority(const int nnodes, const int nvotes); int raft_votes_is_majority(const int nnodes, const int nvotes);
void raft_offer_log(raft_server_t* me_, raft_entry_t* ety, const raft_index_t idx); void raft_handle_append_cfg_change(raft_server_t* me_, raft_entry_t* ety, const raft_index_t idx);
void raft_pop_log(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_);
......
...@@ -33,10 +33,11 @@ typedef struct ...@@ -33,10 +33,11 @@ typedef struct
/* we compact the log, and thus need to increment the Base Log Index */ /* we compact the log, and thus need to increment the Base Log Index */
raft_index_t base; raft_index_t base;
raft_entry_t* entries; raft_entry_t** entries;
/* callbacks */ /* callbacks */
raft_cbs_t *cb; raft_log_cbs_t cb;
void* raft; void* raft;
} log_private_t; } log_private_t;
...@@ -49,12 +50,12 @@ static int mod(raft_index_t a, raft_index_t b) ...@@ -49,12 +50,12 @@ static int mod(raft_index_t a, raft_index_t b)
static int __ensurecapacity(log_private_t * me) static int __ensurecapacity(log_private_t * me)
{ {
raft_index_t i, j; raft_index_t i, j;
raft_entry_t *temp; raft_entry_t **temp;
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_entry_t**)__raft_calloc(1, sizeof(raft_entry_t *) * me->size * 2);
if (!temp) if (!temp)
return RAFT_ERR_NOMEM; return RAFT_ERR_NOMEM;
...@@ -62,7 +63,7 @@ static int __ensurecapacity(log_private_t * me) ...@@ -62,7 +63,7 @@ static int __ensurecapacity(log_private_t * me)
{ {
if (j == me->size) if (j == me->size)
j = 0; j = 0;
memcpy(&temp[i], &me->entries[j], sizeof(raft_entry_t)); temp[i] = me->entries[j];
} }
/* clean up old entries */ /* clean up old entries */
...@@ -93,7 +94,7 @@ log_t* log_alloc(raft_index_t initial_size) ...@@ -93,7 +94,7 @@ log_t* log_alloc(raft_index_t initial_size)
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_entry_t**)__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;
...@@ -106,12 +107,12 @@ log_t* log_new(void) ...@@ -106,12 +107,12 @@ log_t* log_new(void)
return log_alloc(INITIAL_CAPACITY); return log_alloc(INITIAL_CAPACITY);
} }
void log_set_callbacks(log_t* me_, raft_cbs_t* funcs, void* raft) void log_set_callbacks(log_t* me_, raft_log_cbs_t* funcs, void* raft)
{ {
log_private_t* me = (log_private_t*)me_; log_private_t* me = (log_private_t*)me_;
me->raft = raft; me->raft = raft;
me->cb = funcs; me->cb = *funcs;
} }
void log_clear(log_t* me_) void log_clear(log_t* me_)
...@@ -128,13 +129,13 @@ void log_clear_entries(log_t* me_) ...@@ -128,13 +129,13 @@ void log_clear_entries(log_t* me_)
log_private_t* me = (log_private_t*)me_; log_private_t* me = (log_private_t*)me_;
raft_index_t i; raft_index_t i;
if (!me->count || !me->cb || !me->cb->log_clear) if (!me->count || !me->cb.log_clear)
return; return;
for (i = me->base; i <= me->base + me->count; i++) for (i = me->base; i <= me->base + me->count; i++)
{ {
me->cb->log_clear(me->raft, raft_get_udata(me->raft), me->cb.log_clear(me->raft, raft_get_udata(me->raft),
&me->entries[(me->front + i - me->base) % me->size], i); me->entries[(me->front + i - me->base) % me->size], i);
} }
} }
...@@ -149,15 +150,14 @@ int log_append_entry(log_t* me_, raft_entry_t* ety) ...@@ -149,15 +150,14 @@ int log_append_entry(log_t* me_, raft_entry_t* ety)
if (e != 0) if (e != 0)
return e; return e;
memcpy(&me->entries[me->back], ety, sizeof(raft_entry_t)); me->entries[me->back] = ety;
if (me->cb && me->cb->log_offer) if (me->cb.log_offer)
{ {
void* ud = raft_get_udata(me->raft); void* ud = raft_get_udata(me->raft);
e = me->cb->log_offer(me->raft, ud, &me->entries[me->back], idx); e = me->cb.log_offer(me->raft, ud, me->entries[me->back], idx);
if (0 != e) if (0 != e)
return e; return e;
raft_offer_log(me->raft, &me->entries[me->back], idx);
} }
me->count++; me->count++;
...@@ -167,7 +167,7 @@ int log_append_entry(log_t* me_, raft_entry_t* ety) ...@@ -167,7 +167,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, int *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;
...@@ -211,7 +211,7 @@ raft_entry_t* log_get_at_idx(log_t* me_, raft_index_t idx) ...@@ -211,7 +211,7 @@ raft_entry_t* log_get_at_idx(log_t* me_, raft_index_t idx)
idx -= 1; idx -= 1;
i = (me->front + idx - me->base) % me->size; i = (me->front + idx - me->base) % me->size;
return &me->entries[i]; return me->entries[i];
} }
raft_index_t log_count(log_t* me_) raft_index_t log_count(log_t* me_)
...@@ -219,7 +219,7 @@ raft_index_t log_count(log_t* me_) ...@@ -219,7 +219,7 @@ raft_index_t log_count(log_t* me_)
return ((log_private_t*)me_)->count; return ((log_private_t*)me_)->count;
} }
int log_delete(log_t* me_, raft_index_t idx) static int __log_delete(log_t* me_, raft_index_t idx, func_entry_notify_f cb, void *cb_arg)
{ {
log_private_t* me = (log_private_t*)me_; log_private_t* me = (log_private_t*)me_;
...@@ -234,20 +234,28 @@ int log_delete(log_t* me_, raft_index_t idx) ...@@ -234,20 +234,28 @@ int log_delete(log_t* me_, raft_index_t idx)
raft_index_t idx_tmp = me->base + me->count; raft_index_t idx_tmp = me->base + me->count;
raft_index_t back = mod(me->back - 1, me->size); raft_index_t back = mod(me->back - 1, me->size);
if (me->cb && me->cb->log_pop) if (me->cb.log_pop)
{ {
int e = me->cb->log_pop(me->raft, raft_get_udata(me->raft), int e = me->cb.log_pop(me->raft, raft_get_udata(me->raft),
&me->entries[back], idx_tmp); me->entries[back], idx_tmp);
if (0 != e) if (0 != e)
return e; return e;
} }
raft_pop_log(me->raft, &me->entries[back], idx_tmp);
if (cb)
cb(cb_arg, me->entries[back], idx_tmp);
me->back = back; me->back = back;
me->count--; me->count--;
} }
return 0; return 0;
} }
int log_delete(log_t* me_, raft_index_t idx)
{
return __log_delete(me_, idx, NULL, NULL);
}
int log_poll(log_t * me_, void** etyp) int log_poll(log_t * me_, void** etyp)
{ {
log_private_t* me = (log_private_t*)me_; log_private_t* me = (log_private_t*)me_;
...@@ -256,11 +264,11 @@ int log_poll(log_t * me_, void** etyp) ...@@ -256,11 +264,11 @@ int log_poll(log_t * me_, void** etyp)
if (0 == me->count) if (0 == me->count)
return -1; return -1;
const void *elem = &me->entries[me->front]; const void *elem = me->entries[me->front];
if (me->cb && me->cb->log_poll) if (me->cb.log_poll)
{ {
int e = me->cb->log_poll(me->raft, raft_get_udata(me->raft), int e = me->cb.log_poll(me->raft, raft_get_udata(me->raft),
&me->entries[me->front], idx); me->entries[me->front], idx);
if (0 != e) if (0 != e)
return e; return e;
} }
...@@ -281,9 +289,9 @@ raft_entry_t *log_peektail(log_t * me_) ...@@ -281,9 +289,9 @@ raft_entry_t *log_peektail(log_t * me_)
return NULL; return NULL;
if (0 == me->back) if (0 == me->back)
return &me->entries[me->size - 1]; return me->entries[me->size - 1];
else else
return &me->entries[me->back - 1]; return me->entries[me->back - 1];
} }
void log_empty(log_t * me_) void log_empty(log_t * me_)
...@@ -313,3 +321,113 @@ raft_index_t log_get_base(log_t* me_) ...@@ -313,3 +321,113 @@ raft_index_t log_get_base(log_t* me_)
{ {
return ((log_private_t*)me_)->base; return ((log_private_t*)me_)->base;
} }
/**
* The following functions wrap raft_log.c implementation to make it compatible
* with raft_log_impl_t binding.
*
* The rationale for doing this and not modifying raft_log.c directly is to
* leave test_log.c intact.
*
* @todo Ideally test_log.c should be implemented (and extended) to use
* raft_log_impl_t, so it can become a test harness for testing arbitrary
* log implementations.
*/
void *__log_init(void *raft, void *arg)
{
log_t *log = log_new();
if (arg) {
log_set_callbacks(log, arg, raft);
}
return log;
}
static void __log_free(void *log)
{
log_free(log);
}
static void __log_reset(void *log, raft_index_t first_idx, raft_term_t term)
{
log_clear_entries(log);
log_clear(log);
assert(first_idx >= 1);
((log_private_t*) log)->base = first_idx - 1;
}
static int __log_append(void *log, raft_entry_t *entry)
{
return log_append_entry(log, entry);
}
static raft_entry_t *__log_get(void *log, raft_index_t idx)
{
return log_get_at_idx(log, idx);
}
static int __log_get_batch(void *log, raft_index_t idx, int entries_n, raft_entry_t **entries)
{
int n, i;
raft_entry_t **r = log_get_from_idx(log, idx, &n);
if (!r || n < 1) {
return 0;
}
if (n > entries_n)
n = entries_n;
for (i = 0; i < n; i++) {
entries[i] = r[i];
}
return n;
}
static int __log_pop(void *log, raft_index_t from_idx, func_entry_notify_f cb, void *cb_arg)
{
return __log_delete(log, from_idx, cb, cb_arg);
}
static int __log_poll(void *log, raft_index_t first_idx)
{
while (log_get_base(log) + 1 < first_idx) {
raft_entry_t *ety;
int e = log_poll(log, (void **) &ety);
if (e < 0)
return e;
}
return 0;
}
static raft_index_t __log_first_idx(void *log)
{
return log_get_base(log) + 1;
}
static raft_index_t __log_current_idx(void *log)
{
return log_get_current_idx(log);
}
static raft_index_t __log_count(void *log)
{
return log_count(log);
}
const raft_log_impl_t raft_log_internal_impl = {
.init = __log_init,
.free = __log_free,
.reset = __log_reset,
.append = __log_append,
.poll = __log_poll,
.pop = __log_pop,
.get = __log_get,
.get_batch = __log_get_batch,
.first_idx = __log_first_idx,
.current_idx = __log_current_idx,
.count = __log_count
};
...@@ -66,7 +66,7 @@ void raft_randomize_election_timeout(raft_server_t* me_) ...@@ -66,7 +66,7 @@ void raft_randomize_election_timeout(raft_server_t* me_)
__log(me_, NULL, "randomize election timeout to %d", me->election_timeout_rand); __log(me_, NULL, "randomize election timeout to %d", me->election_timeout_rand);
} }
raft_server_t* raft_new(void) raft_server_t* raft_new_with_log(const raft_log_impl_t *log_impl, void *log_arg)
{ {
raft_server_private_t* me = raft_server_private_t* me =
(raft_server_private_t*)__raft_calloc(1, sizeof(raft_server_private_t)); (raft_server_private_t*)__raft_calloc(1, sizeof(raft_server_private_t));
...@@ -78,7 +78,8 @@ raft_server_t* raft_new(void) ...@@ -78,7 +78,8 @@ raft_server_t* raft_new(void)
me->request_timeout = 200; me->request_timeout = 200;
me->election_timeout = 1000; me->election_timeout = 1000;
raft_randomize_election_timeout((raft_server_t*)me); raft_randomize_election_timeout((raft_server_t*)me);
me->log = log_new(); me->log_impl = log_impl;
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;
...@@ -93,20 +94,26 @@ raft_server_t* raft_new(void) ...@@ -93,20 +94,26 @@ raft_server_t* raft_new(void)
return (raft_server_t*)me; return (raft_server_t*)me;
} }
extern raft_log_impl_t log_internal_impl;
raft_server_t* raft_new(void)
{
return raft_new_with_log(&raft_log_internal_impl, NULL);
}
void raft_set_callbacks(raft_server_t* me_, raft_cbs_t* funcs, void* udata) void raft_set_callbacks(raft_server_t* me_, raft_cbs_t* funcs, void* udata)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
memcpy(&me->cb, funcs, sizeof(raft_cbs_t)); memcpy(&me->cb, funcs, sizeof(raft_cbs_t));
me->udata = udata; me->udata = udata;
log_set_callbacks(me->log, &me->cb, me_);
} }
void raft_free(raft_server_t* me_) void raft_free(raft_server_t* me_)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
log_free(me->log); me->log_impl->free(me->log);
__raft_free(me_); __raft_free(me_);
} }
...@@ -125,8 +132,7 @@ void raft_clear(raft_server_t* me_) ...@@ -125,8 +132,7 @@ void raft_clear(raft_server_t* me_)
me->last_applied_idx = 0; me->last_applied_idx = 0;
me->num_nodes = 0; me->num_nodes = 0;
me->node = NULL; me->node = NULL;
log_clear_entries(me->log); me->log_impl->reset(me->log, 1, 1);
log_clear(me->log);
} }
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)
...@@ -138,7 +144,8 @@ int raft_delete_entry_from_idx(raft_server_t* me_, raft_index_t idx) ...@@ -138,7 +144,8 @@ int raft_delete_entry_from_idx(raft_server_t* me_, raft_index_t idx)
if (idx <= me->voting_cfg_change_log_idx) if (idx <= me->voting_cfg_change_log_idx)
me->voting_cfg_change_log_idx = -1; me->voting_cfg_change_log_idx = -1;
return log_delete(me->log, idx); return me->log_impl->pop(me->log, idx,
(func_entry_notify_f) raft_handle_remove_cfg_change, me_);
} }
int raft_election_start(raft_server_t* me_) int raft_election_start(raft_server_t* me_)
...@@ -262,7 +269,7 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period) ...@@ -262,7 +269,7 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period)
raft_entry_t* raft_get_entry_from_idx(raft_server_t* me_, raft_index_t etyidx) raft_entry_t* raft_get_entry_from_idx(raft_server_t* me_, raft_index_t etyidx)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
return log_get_at_idx(me->log, etyidx); return me->log_impl->get(me->log, etyidx);
} }
int raft_voting_change_is_in_progress(raft_server_t* me_) int raft_voting_change_is_in_progress(raft_server_t* me_)
...@@ -372,7 +379,7 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -372,7 +379,7 @@ int raft_recv_appendentries_response(raft_server_t* me_,
} }
/* Aggressively send remaining entries */ /* Aggressively send remaining entries */
if (raft_get_entry_from_idx(me_, raft_node_get_next_idx(node))) if (raft_node_get_next_idx(node) <= raft_get_current_idx(me_))
raft_send_appendentries(me_, node); raft_send_appendentries(me_, node);
/* periodic applies committed entries lazily */ /* periodic applies committed entries lazily */
...@@ -476,7 +483,7 @@ int raft_recv_appendentries( ...@@ -476,7 +483,7 @@ int raft_recv_appendentries(
int i; int i;
for (i = 0; i < ae->n_entries; i++) for (i = 0; i < ae->n_entries; i++)
{ {
raft_entry_t* ety = &ae->entries[i]; raft_entry_t* ety = ae->entries[i];
raft_index_t ety_index = ae->prev_log_idx + 1 + i; raft_index_t ety_index = ae->prev_log_idx + 1 + i;
raft_entry_t* existing_ety = raft_get_entry_from_idx(me_, ety_index); raft_entry_t* existing_ety = raft_get_entry_from_idx(me_, ety_index);
if (existing_ety && existing_ety->term != ety->term) if (existing_ety && existing_ety->term != ety->term)
...@@ -503,7 +510,7 @@ int raft_recv_appendentries( ...@@ -503,7 +510,7 @@ int raft_recv_appendentries(
/* Pick up remainder in case of mismatch or missing entry */ /* Pick up remainder in case of mismatch or missing entry */
for (; i < ae->n_entries; i++) for (; i < ae->n_entries; i++)
{ {
e = raft_append_entry(me_, &ae->entries[i]); e = raft_append_entry(me_, ae->entries[i]);
if (0 != e) if (0 != e)
goto out; goto out;
r->current_idx = ae->prev_log_idx + 1 + i; r->current_idx = ae->prev_log_idx + 1 + i;
...@@ -800,10 +807,18 @@ int raft_append_entry(raft_server_t* me_, raft_entry_t* ety) ...@@ -800,10 +807,18 @@ int raft_append_entry(raft_server_t* me_, raft_entry_t* ety)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
int e = me->log_impl->append(me->log, ety);
if (e < 0)
return e;
if (raft_entry_is_voting_cfg_change(ety)) if (raft_entry_is_voting_cfg_change(ety))
me->voting_cfg_change_log_idx = raft_get_current_idx(me_); me->voting_cfg_change_log_idx = raft_get_current_idx(me_) - 1;
if (raft_entry_is_cfg_change(ety)) {
raft_handle_append_cfg_change(me_, ety, raft_get_current_idx(me_));
}
return log_append_entry(me->log, ety); return 0;
} }
int raft_apply_entry(raft_server_t* me_) int raft_apply_entry(raft_server_t* me_)
...@@ -824,7 +839,7 @@ int raft_apply_entry(raft_server_t* me_) ...@@ -824,7 +839,7 @@ int raft_apply_entry(raft_server_t* me_)
return -1; return -1;
__log(me_, NULL, "applying log: %d, id: %d size: %d", __log(me_, NULL, "applying log: %d, id: %d size: %d",
log_idx, ety->id, ety->data.len); log_idx, ety->id, ety->data_len);
me->last_applied_idx++; me->last_applied_idx++;
if (me->cb.applylog) if (me->cb.applylog)
...@@ -842,7 +857,7 @@ int raft_apply_entry(raft_server_t* me_) ...@@ -842,7 +857,7 @@ int raft_apply_entry(raft_server_t* me_)
me->voting_cfg_change_log_idx = -1; me->voting_cfg_change_log_idx = -1;
if (!raft_entry_is_cfg_change(ety)) if (!raft_entry_is_cfg_change(ety))
return 0; goto exit;
raft_node_id_t node_id = me->cb.log_get_node_id(me_, raft_get_udata(me_), ety, log_idx); raft_node_id_t node_id = me->cb.log_get_node_id(me_, raft_get_udata(me_), ety, log_idx);
raft_node_t* node = raft_get_node(me_, node_id); raft_node_t* node = raft_get_node(me_, node_id);
...@@ -871,13 +886,31 @@ int raft_apply_entry(raft_server_t* me_) ...@@ -871,13 +886,31 @@ int raft_apply_entry(raft_server_t* me_)
break; break;
} }
exit:
return 0; return 0;
} }
raft_entry_t* raft_get_entries_from_idx(raft_server_t* me_, raft_index_t idx, int* n_etys) raft_entry_t** raft_get_entries_from_idx(raft_server_t* me_, raft_index_t idx, int* n_etys)
{ {
if (raft_get_current_idx(me_) < idx) {
*n_etys = 0;
return NULL;
}
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
return log_get_from_idx(me->log, idx, n_etys); int size = raft_get_current_idx(me_) - idx + 1;
raft_entry_t **e = __raft_malloc(size * sizeof(raft_entry_t*));
int n = me->log_impl->get_batch(me->log, idx, size, e);
if (n < 1) {
__raft_free(e);
*n_etys = 0;
return NULL;
}
*n_etys = n;
return e;
} }
int raft_send_appendentries(raft_server_t* me_, raft_node_t* node) int raft_send_appendentries(raft_server_t* me_, raft_node_t* node)
...@@ -926,15 +959,19 @@ int raft_send_appendentries(raft_server_t* me_, raft_node_t* node) ...@@ -926,15 +959,19 @@ int raft_send_appendentries(raft_server_t* me_, raft_node_t* node)
} }
} }
__log(me_, node, "sending appendentries node: ci:%d comi:%d t:%d lc:%d pli:%d plt:%d", __log(me_, node, "sending appendentries: ci:%lu comi:%lu t:%lu lc:%lu pli:%lu plt:%lu #%d",
raft_get_current_idx(me_), raft_get_current_idx(me_),
raft_get_commit_idx(me_), raft_get_commit_idx(me_),
ae.term, ae.term,
ae.leader_commit, ae.leader_commit,
ae.prev_log_idx, ae.prev_log_idx,
ae.prev_log_term); ae.prev_log_term,
ae.n_entries);
int res = me->cb.send_appendentries(me_, me->udata, node, &ae);
__raft_free(ae.entries);
return me->cb.send_appendentries(me_, me->udata, node, &ae); return res;
} }
int raft_send_appendentries_all(raft_server_t* me_) int raft_send_appendentries_all(raft_server_t* me_)
...@@ -1127,13 +1164,16 @@ int raft_entry_is_cfg_change(raft_entry_t* ety) ...@@ -1127,13 +1164,16 @@ int raft_entry_is_cfg_change(raft_entry_t* ety)
RAFT_LOGTYPE_REMOVE_NODE == ety->type); RAFT_LOGTYPE_REMOVE_NODE == ety->type);
} }
void raft_offer_log(raft_server_t* me_, raft_entry_t* ety, const raft_index_t idx) 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_; raft_server_private_t* me = (raft_server_private_t*)me_;
if (!raft_entry_is_cfg_change(ety)) if (!raft_entry_is_cfg_change(ety))
return; 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_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); raft_node_t* node = raft_get_node(me_, node_id);
int is_self = node_id == raft_get_nodeid(me_); int is_self = node_id == raft_get_nodeid(me_);
...@@ -1176,13 +1216,16 @@ void raft_offer_log(raft_server_t* me_, raft_entry_t* ety, const raft_index_t id ...@@ -1176,13 +1216,16 @@ void raft_offer_log(raft_server_t* me_, raft_entry_t* ety, const raft_index_t id
} }
} }
void raft_pop_log(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_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
if (!raft_entry_is_cfg_change(ety)) if (!raft_entry_is_cfg_change(ety))
return; 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_id_t node_id = me->cb.log_get_node_id(me_, raft_get_udata(me_), ety, idx);
switch (ety->type) switch (ety->type)
...@@ -1224,14 +1267,16 @@ void raft_pop_log(raft_server_t* me_, raft_entry_t* ety, const raft_index_t idx) ...@@ -1224,14 +1267,16 @@ void raft_pop_log(raft_server_t* me_, raft_entry_t* ety, const raft_index_t idx)
} }
} }
int raft_poll_entry(raft_server_t* me_, raft_entry_t **ety) 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_;
int e = log_poll(me->log, (void*)ety); /* We should never drop uncommitted entries */
assert(me->log_impl->first_idx(me->log) <= raft_get_commit_idx(me_));
int e = me->log_impl->poll(me->log, me->log_impl->first_idx(me->log));
if (e != 0) if (e != 0)
return e; return e;
assert(*ety != NULL);
return 0; return 0;
} }
...@@ -1240,7 +1285,10 @@ int raft_pop_entry(raft_server_t* me_) ...@@ -1240,7 +1285,10 @@ int raft_pop_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_;
return log_delete(me->log, log_get_current_idx(me->log)); raft_index_t cur_idx = me->log_impl->current_idx(me->log);
return me->log_impl->pop(me->log, cur_idx,
(func_entry_notify_f) raft_handle_remove_cfg_change, me_);
} }
raft_index_t raft_get_first_entry_idx(raft_server_t* me_) raft_index_t raft_get_first_entry_idx(raft_server_t* me_)
...@@ -1260,7 +1308,7 @@ raft_index_t raft_get_num_snapshottable_logs(raft_server_t *me_) ...@@ -1260,7 +1308,7 @@ raft_index_t raft_get_num_snapshottable_logs(raft_server_t *me_)
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
if (raft_get_log_count(me_) <= 1) if (raft_get_log_count(me_) <= 1)
return 0; return 0;
return raft_get_commit_idx(me_) - log_get_base(me->log); return raft_get_commit_idx(me_) - me->log_impl->first_idx(me->log) + 1;
} }
int raft_begin_snapshot(raft_server_t *me_, int flags) int raft_begin_snapshot(raft_server_t *me_, int flags)
...@@ -1290,7 +1338,7 @@ int raft_begin_snapshot(raft_server_t *me_, int flags) ...@@ -1290,7 +1338,7 @@ int raft_begin_snapshot(raft_server_t *me_, int flags)
me->snapshot_flags = flags; me->snapshot_flags = flags;
__log(me_, NULL, __log(me_, NULL,
"begin snapshot sli:%d slt:%d slogs:%d\n", "begin snapshot sli:%d slt:%d slogs:%d",
me->snapshot_last_idx, me->snapshot_last_idx,
me->snapshot_last_term, me->snapshot_last_term,
raft_get_num_snapshottable_logs(me_)); raft_get_num_snapshottable_logs(me_));
...@@ -1320,29 +1368,26 @@ int raft_end_snapshot(raft_server_t *me_) ...@@ -1320,29 +1368,26 @@ int raft_end_snapshot(raft_server_t *me_)
if (!me->snapshot_in_progress || me->snapshot_last_idx == 0) if (!me->snapshot_in_progress || me->snapshot_last_idx == 0)
return -1; return -1;
assert(raft_get_num_snapshottable_logs(me_) != 0); // TODO: What is the purpose of this assert? Looks wrong.
// assert(raft_get_num_snapshottable_logs(me_) != 0);
/* If needed, remove compacted logs */ /* If needed, remove compacted logs */
raft_index_t i = log_get_base(me->log) + 1, end = me->snapshot_last_idx; int e = me->log_impl->poll(me->log, me->snapshot_last_idx + 1);
for (; i <= end; i++)
{
raft_entry_t* _ety;
int e = raft_poll_entry(me_, &_ety);
if (e != 0) if (e != 0)
return -1; return -1;
}
me->snapshot_in_progress = 0; me->snapshot_in_progress = 0;
__log(me_, NULL, __log(me_, NULL,
"end snapshot base:%d commit-index:%d current-index:%d\n", "end snapshot base:%d commit-index:%d current-index:%d",
log_get_base(me->log), me->log_impl->first_idx(me->log) - 1,
raft_get_commit_idx(me_), raft_get_commit_idx(me_),
raft_get_current_idx(me_)); raft_get_current_idx(me_));
if (!raft_is_leader(me_)) if (!raft_is_leader(me_))
return 0; return 0;
int i;
for (i = 0; i < me->num_nodes; i++) for (i = 0; i < me->num_nodes; i++)
{ {
raft_node_t* node = me->nodes[i]; raft_node_t* node = me->nodes[i];
...@@ -1392,7 +1437,7 @@ int raft_begin_load_snapshot( ...@@ -1392,7 +1437,7 @@ int raft_begin_load_snapshot(
raft_set_state((raft_server_t*)me, RAFT_STATE_FOLLOWER); raft_set_state((raft_server_t*)me, RAFT_STATE_FOLLOWER);
me->current_leader = NULL; me->current_leader = NULL;
log_load_from_snapshot(me->log, last_included_index, last_included_term); me->log_impl->reset(me->log, last_included_index + 1, last_included_term);
if (raft_get_commit_idx(me_) < last_included_index) if (raft_get_commit_idx(me_) < last_included_index)
raft_set_commit_idx(me_, last_included_index); raft_set_commit_idx(me_, last_included_index);
...@@ -1417,7 +1462,7 @@ int raft_begin_load_snapshot( ...@@ -1417,7 +1462,7 @@ int raft_begin_load_snapshot(
me->num_nodes = 1; me->num_nodes = 1;
__log(me_, NULL, __log(me_, NULL,
"loaded snapshot sli:%d slt:%d slogs:%d\n", "loaded snapshot sli:%d slt:%d slogs:%d",
me->snapshot_last_idx, me->snapshot_last_idx,
me->snapshot_last_term, me->snapshot_last_term,
raft_get_num_snapshottable_logs(me_)); raft_get_num_snapshottable_logs(me_));
...@@ -1442,3 +1487,42 @@ int raft_end_load_snapshot(raft_server_t *me_) ...@@ -1442,3 +1487,42 @@ int raft_end_load_snapshot(raft_server_t *me_)
return 0; return 0;
} }
void *raft_get_log(raft_server_t *me_)
{
raft_server_private_t* me = (raft_server_private_t*)me_;
return me->log;
}
raft_entry_t *raft_entry_new(unsigned int data_len)
{
raft_entry_t *ety = __raft_calloc(1, sizeof(raft_entry_t) + data_len);
ety->data_len = data_len;
ety->refs = 1;
return ety;
}
void raft_entry_hold(raft_entry_t *ety)
{
ety->refs++;
}
void raft_entry_release(raft_entry_t *ety)
{
assert(ety->refs > 0);
ety->refs--;
if (!ety->refs) {
__raft_free(ety);
}
}
void raft_entry_release_list(raft_entry_t **ety_list, size_t len)
{
size_t i;
for (i = 0; i < len; i++) {
raft_entry_release(ety_list[i]);
}
}
...@@ -73,7 +73,7 @@ int raft_get_timeout_elapsed(raft_server_t* me_) ...@@ -73,7 +73,7 @@ int raft_get_timeout_elapsed(raft_server_t* me_)
raft_index_t raft_get_log_count(raft_server_t* me_) raft_index_t raft_get_log_count(raft_server_t* me_)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
return log_count(me->log); return me->log_impl->count(me->log);
} }
int raft_get_voted_for(raft_server_t* me_) int raft_get_voted_for(raft_server_t* me_)
...@@ -108,7 +108,7 @@ raft_term_t raft_get_current_term(raft_server_t* me_) ...@@ -108,7 +108,7 @@ raft_term_t raft_get_current_term(raft_server_t* me_)
raft_index_t raft_get_current_idx(raft_server_t* me_) raft_index_t raft_get_current_idx(raft_server_t* me_)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
return log_get_current_idx(me->log); return me->log_impl->current_idx(me->log);
} }
void raft_set_commit_idx(raft_server_t* me_, raft_index_t idx) void raft_set_commit_idx(raft_server_t* me_, raft_index_t idx)
...@@ -219,8 +219,10 @@ raft_term_t raft_get_last_log_term(raft_server_t* me_) ...@@ -219,8 +219,10 @@ raft_term_t raft_get_last_log_term(raft_server_t* me_)
if (0 < current_idx) if (0 < current_idx)
{ {
raft_entry_t* ety = raft_get_entry_from_idx(me_, current_idx); raft_entry_t* ety = raft_get_entry_from_idx(me_, current_idx);
if (ety) if (ety) {
return ety->term; raft_term_t term = ety->term;
return term;
}
} }
return 0; return 0;
} }
...@@ -246,7 +248,7 @@ raft_entry_t *raft_get_last_applied_entry(raft_server_t *me_) ...@@ -246,7 +248,7 @@ raft_entry_t *raft_get_last_applied_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_;
if (raft_get_last_applied_idx(me_) == 0) if (raft_get_last_applied_idx(me_) == 0)
return NULL; return NULL;
return log_get_at_idx(me->log, raft_get_last_applied_idx(me_)); return me->log_impl->get(me->log, raft_get_last_applied_idx(me_));
} }
raft_index_t raft_get_snapshot_last_idx(raft_server_t *me_) raft_index_t raft_get_snapshot_last_idx(raft_server_t *me_)
...@@ -262,6 +264,7 @@ raft_term_t raft_get_snapshot_last_term(raft_server_t *me_) ...@@ -262,6 +264,7 @@ raft_term_t raft_get_snapshot_last_term(raft_server_t *me_)
void raft_set_snapshot_metadata(raft_server_t *me_, raft_term_t term, raft_index_t idx) void raft_set_snapshot_metadata(raft_server_t *me_, raft_term_t term, raft_index_t idx)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
me->last_applied_idx = idx;
me->saved_snapshot_last_term = me->snapshot_last_term; me->saved_snapshot_last_term = me->snapshot_last_term;
me->saved_snapshot_last_idx = me->snapshot_last_idx; me->saved_snapshot_last_idx = me->snapshot_last_idx;
me->snapshot_last_term = term; me->snapshot_last_term = term;
......
#ifndef _HELPERS_H
#define _HELPERS_H
static raft_entry_t *__MAKE_ENTRY(int id, raft_term_t term, const char *data)
{
raft_entry_t *ety = raft_entry_new(data ? strlen(data) : 0);
ety->id = id;
ety->term = term;
if (data) {
memcpy(ety->data, data, strlen(data));
}
return ety;
}
static raft_entry_t **__MAKE_ENTRY_ARRAY(int id, raft_term_t term, const char *data)
{
raft_entry_t **array = calloc(1, sizeof(raft_entry_t *));
array[0] = __MAKE_ENTRY(id, term, data);
return array;
}
static raft_entry_t **__MAKE_ENTRY_ARRAY_SEQ_ID(int count, int start_id, raft_term_t term, const char *data)
{
raft_entry_t **array = calloc(count, sizeof(raft_entry_t *));
int i;
for (i = 0; i < count; i++) {
array[i] = __MAKE_ENTRY(start_id++, term, data);
}
return array;
}
static void __APPEND_ENTRY(void *r, int id, raft_term_t term, const char *data)
{
raft_entry_t *e = __MAKE_ENTRY(id, term, data);
raft_append_entry(r, e);
}
static void __APPEND_ENTRIES_SEQ_ID(void *r, int count, int id, raft_term_t term, const char *data)
{
int i;
for (i = 0; i < count; i++) {
raft_entry_t *e = __MAKE_ENTRY(id++, term, data);
raft_append_entry(r, e);
}
}
static void __APPEND_ENTRIES_SEQ_ID_TERM(void *r, int count, int id, raft_term_t term, const char *data)
{
int i;
for (i = 0; i < count; i++) {
raft_entry_t *e = __MAKE_ENTRY(id++, term++, data);
raft_append_entry(r, e);
}
}
#endif
...@@ -91,11 +91,19 @@ int sender_requestvote_response(raft_server_t* raft, ...@@ -91,11 +91,19 @@ int sender_requestvote_response(raft_server_t* raft,
int sender_appendentries(raft_server_t* raft, int sender_appendentries(raft_server_t* raft,
void* udata, raft_node_t* node, msg_appendentries_t* msg) void* udata, raft_node_t* node, msg_appendentries_t* msg)
{ {
msg_entry_t* entries = calloc(1, sizeof(msg_entry_t) * msg->n_entries); msg_entry_t** entries = calloc(1, sizeof(msg_entry_t *) * msg->n_entries);
memcpy(entries, msg->entries, sizeof(msg_entry_t) * msg->n_entries); int i;
for (i = 0; i < msg->n_entries; i++) {
entries[i] = msg->entries[i];
raft_entry_hold(entries[i]);
}
msg_entry_t** old_entries = msg->entries;
msg->entries = entries; msg->entries = entries;
return __append_msg(udata, msg, RAFT_MSG_APPENDENTRIES, sizeof(*msg), node, int ret = __append_msg(udata, msg, RAFT_MSG_APPENDENTRIES, sizeof(*msg), node,
raft); raft);
msg->entries = old_entries;
return ret;
} }
int sender_appendentries_response(raft_server_t* raft, int sender_appendentries_response(raft_server_t* raft,
......
...@@ -5,6 +5,26 @@ ffi = cffi.FFI() ...@@ -5,6 +5,26 @@ ffi = cffi.FFI()
ffi.set_source( ffi.set_source(
"tests", "tests",
""" """
#include "raft.h"
raft_entry_t *raft_entry_newdata(void *data) {
raft_entry_t *e = raft_entry_new(sizeof(void *));
*(void **) e->data = data;
e->refs = 100;
return e;
}
void *raft_entry_getdata(raft_entry_t *ety) {
return *(void **) ety->data;
}
raft_entry_t **raft_entry_array_deepcopy(raft_entry_t **src, int len) {
raft_entry_t **t = malloc(len * sizeof(raft_entry_t *));
int i;
for (i = 0; i < len; i++) {
int sz = sizeof(raft_entry_t) + src[i]->data_len;
t[i] = malloc(sz);
memcpy(t[i], src[i], sz);
}
return t;
}
""", """,
sources=""" sources="""
src/raft_log.c src/raft_log.c
...@@ -13,13 +33,13 @@ ffi.set_source( ...@@ -13,13 +33,13 @@ ffi.set_source(
src/raft_node.c src/raft_node.c
""".split(), """.split(),
include_dirs=["include"], include_dirs=["include"],
extra_compile_args=["-UNDEBUG"]
) )
library = ffi.compile() library = ffi.compile()
ffi = cffi.FFI() ffi = cffi.FFI()
lib = ffi.dlopen(library) lib = ffi.dlopen(library)
def load(fname): def load(fname):
return '\n'.join( return '\n'.join(
[line for line in subprocess.check_output( [line for line in subprocess.check_output(
...@@ -29,3 +49,7 @@ def load(fname): ...@@ -29,3 +49,7 @@ def load(fname):
ffi.cdef('void *malloc(size_t __size);') ffi.cdef('void *malloc(size_t __size);')
ffi.cdef(load('include/raft.h')) ffi.cdef(load('include/raft.h'))
ffi.cdef(load('include/raft_log.h')) ffi.cdef(load('include/raft_log.h'))
ffi.cdef('raft_entry_t *raft_entry_newdata(void *data);')
ffi.cdef('void *raft_entry_getdata(raft_entry_t *);')
ffi.cdef('raft_entry_t **raft_entry_array_deepcopy(raft_entry_t **src, int len);')
...@@ -12,6 +12,8 @@ ...@@ -12,6 +12,8 @@
#include "raft_log.h" #include "raft_log.h"
#include "raft_private.h" #include "raft_private.h"
#include "helpers.h"
static int __logentry_get_node_id( static int __logentry_get_node_id(
raft_server_t* raft, raft_server_t* raft,
void *udata, void *udata,
...@@ -57,15 +59,19 @@ static int __log_pop_failing( ...@@ -57,15 +59,19 @@ static int __log_pop_failing(
} }
raft_cbs_t funcs = { raft_cbs_t funcs = {
.log_pop = __log_pop,
.log_get_node_id = __logentry_get_node_id .log_get_node_id = __logentry_get_node_id
}; };
raft_log_cbs_t log_funcs = {
.log_pop = __log_pop
};
void* __set_up() void* __set_up()
{ {
void* queue = llqueue_new(); void* queue = llqueue_new();
void *r = raft_new(); void *r = raft_new();
raft_set_callbacks(r, &funcs, queue); raft_set_callbacks(r, &funcs, queue);
log_set_callbacks(raft_get_log(r), &log_funcs, r);
return r; return r;
} }
...@@ -89,7 +95,7 @@ void TestLog_append_is_not_empty(CuTest * tc) ...@@ -89,7 +95,7 @@ void TestLog_append_is_not_empty(CuTest * tc)
e.id = 1; e.id = 1;
l = log_new(); l = log_new();
raft_cbs_t funcs = { raft_log_cbs_t funcs = {
.log_offer = __log_offer .log_offer = __log_offer
}; };
log_set_callbacks(l, &funcs, r); log_set_callbacks(l, &funcs, r);
...@@ -143,13 +149,16 @@ void TestLog_delete(CuTest * tc) ...@@ -143,13 +149,16 @@ void TestLog_delete(CuTest * tc)
void* queue = llqueue_new(); void* queue = llqueue_new();
void *r = raft_new(); void *r = raft_new();
raft_cbs_t funcs = { raft_cbs_t funcs = {
.log_pop = __log_pop,
.log_get_node_id = __logentry_get_node_id .log_get_node_id = __logentry_get_node_id
}; };
raft_log_cbs_t log_funcs = {
.log_pop = __log_pop
};
raft_set_callbacks(r, &funcs, queue); raft_set_callbacks(r, &funcs, queue);
l = log_new(); l = log_new();
log_set_callbacks(l, &funcs, r); log_set_callbacks(l, &log_funcs, r);
memset(&e1, 0, sizeof(raft_entry_t)); memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t)); memset(&e2, 0, sizeof(raft_entry_t));
...@@ -184,9 +193,11 @@ void TestLog_delete_onwards(CuTest * tc) ...@@ -184,9 +193,11 @@ void TestLog_delete_onwards(CuTest * tc)
void* queue = llqueue_new(); void* queue = llqueue_new();
void *r = raft_new(); void *r = raft_new();
raft_cbs_t funcs = { raft_cbs_t funcs = {
.log_pop = __log_pop,
.log_get_node_id = __logentry_get_node_id .log_get_node_id = __logentry_get_node_id
}; };
raft_log_cbs_t log_funcs = {
.log_pop = __log_pop
};
raft_set_callbacks(r, &funcs, queue); raft_set_callbacks(r, &funcs, queue);
void *l; void *l;
...@@ -197,7 +208,7 @@ void TestLog_delete_onwards(CuTest * tc) ...@@ -197,7 +208,7 @@ void TestLog_delete_onwards(CuTest * tc)
memset(&e3, 0, sizeof(raft_entry_t)); memset(&e3, 0, sizeof(raft_entry_t));
l = log_new(); l = log_new();
log_set_callbacks(l, &funcs, r); log_set_callbacks(l, &log_funcs, r);
e1.id = 1; e1.id = 1;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e1)); CuAssertIntEquals(tc, 0, log_append_entry(l, &e1));
e2.id = 2; e2.id = 2;
...@@ -222,13 +233,15 @@ void TestLog_delete_handles_log_pop_failure(CuTest * tc) ...@@ -222,13 +233,15 @@ void TestLog_delete_handles_log_pop_failure(CuTest * tc)
void* queue = llqueue_new(); void* queue = llqueue_new();
void *r = raft_new(); void *r = raft_new();
raft_cbs_t funcs = { raft_cbs_t funcs = {
.log_pop = __log_pop_failing,
.log_get_node_id = __logentry_get_node_id .log_get_node_id = __logentry_get_node_id
}; };
raft_log_cbs_t log_funcs = {
.log_pop = __log_pop_failing
};
raft_set_callbacks(r, &funcs, queue); raft_set_callbacks(r, &funcs, queue);
l = log_new(); l = log_new();
log_set_callbacks(l, &funcs, r); log_set_callbacks(l, &log_funcs, r);
memset(&e1, 0, sizeof(raft_entry_t)); memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t)); memset(&e2, 0, sizeof(raft_entry_t));
...@@ -254,10 +267,13 @@ void TestLog_delete_fails_for_idx_zero(CuTest * tc) ...@@ -254,10 +267,13 @@ void TestLog_delete_fails_for_idx_zero(CuTest * tc)
void* queue = llqueue_new(); void* queue = llqueue_new();
void *r = raft_new(); void *r = raft_new();
raft_cbs_t funcs = { raft_cbs_t funcs = {
.log_pop = __log_pop,
.log_get_node_id = __logentry_get_node_id .log_get_node_id = __logentry_get_node_id
}; };
raft_log_cbs_t log_funcs = {
.log_pop = __log_pop
};
raft_set_callbacks(r, &funcs, queue); raft_set_callbacks(r, &funcs, queue);
log_set_callbacks(raft_get_log(r), &log_funcs, r);
void *l; void *l;
raft_entry_t e1, e2, e3, e4; raft_entry_t e1, e2, e3, e4;
...@@ -273,7 +289,7 @@ void TestLog_delete_fails_for_idx_zero(CuTest * tc) ...@@ -273,7 +289,7 @@ void TestLog_delete_fails_for_idx_zero(CuTest * tc)
e4.id = 4; e4.id = 4;
l = log_alloc(1); l = log_alloc(1);
log_set_callbacks(l, &funcs, r); log_set_callbacks(l, &log_funcs, r);
CuAssertIntEquals(tc, 0, log_append_entry(l, &e1)); CuAssertIntEquals(tc, 0, log_append_entry(l, &e1));
CuAssertIntEquals(tc, 0, log_append_entry(l, &e2)); CuAssertIntEquals(tc, 0, log_append_entry(l, &e2));
CuAssertIntEquals(tc, 0, log_append_entry(l, &e3)); CuAssertIntEquals(tc, 0, log_append_entry(l, &e3));
...@@ -286,16 +302,18 @@ void TestLog_poll(CuTest * tc) ...@@ -286,16 +302,18 @@ void TestLog_poll(CuTest * tc)
void* queue = llqueue_new(); void* queue = llqueue_new();
void *r = raft_new(); void *r = raft_new();
raft_cbs_t funcs = { raft_cbs_t funcs = {
.log_pop = __log_pop,
.log_get_node_id = __logentry_get_node_id .log_get_node_id = __logentry_get_node_id
}; };
raft_log_cbs_t log_funcs = {
.log_pop = __log_pop
};
raft_set_callbacks(r, &funcs, queue); raft_set_callbacks(r, &funcs, queue);
void *l; void *l;
raft_entry_t e1, e2, e3; raft_entry_t e1, e2, e3;
l = log_new(); l = log_new();
log_set_callbacks(l, &funcs, r); log_set_callbacks(l, &log_funcs, r);
memset(&e1, 0, sizeof(raft_entry_t)); memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t)); memset(&e2, 0, sizeof(raft_entry_t));
...@@ -428,25 +446,19 @@ void TestLog_front_pushes_across_boundary(CuTest * tc) ...@@ -428,25 +446,19 @@ void TestLog_front_pushes_across_boundary(CuTest * tc)
void* r = __set_up(); void* r = __set_up();
void *l; void *l;
raft_entry_t e1, e2, e3;
memset(&e1, 0, sizeof(raft_entry_t)); raft_entry_t *e1 = __MAKE_ENTRY(1, 1, "aa");
memset(&e2, 0, sizeof(raft_entry_t)); raft_entry_t *e2 = __MAKE_ENTRY(2, 1, "aa");
memset(&e3, 0, sizeof(raft_entry_t));
e1.id = 1;
e2.id = 2;
e3.id = 3;
l = log_alloc(1); l = log_alloc(1);
log_set_callbacks(l, &funcs, r); log_set_callbacks(l, &log_funcs, r);
raft_entry_t* ety; raft_entry_t* ety;
CuAssertIntEquals(tc, 0, log_append_entry(l, &e1)); CuAssertIntEquals(tc, 0, log_append_entry(l, e1));
CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0); CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0);
CuAssertIntEquals(tc, ety->id, 1); CuAssertIntEquals(tc, ety->id, 1);
CuAssertIntEquals(tc, 0, log_append_entry(l, &e2)); CuAssertIntEquals(tc, 0, log_append_entry(l, e2));
CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0); CuAssertIntEquals(tc, log_poll(l, (void*)&ety), 0);
CuAssertIntEquals(tc, ety->id, 2); CuAssertIntEquals(tc, ety->id, 2);
} }
...@@ -535,9 +547,11 @@ void TestLog_delete_after_polling_from_double_append(CuTest * tc) ...@@ -535,9 +547,11 @@ void TestLog_delete_after_polling_from_double_append(CuTest * tc)
void* queue = llqueue_new(); void* queue = llqueue_new();
void *r = raft_new(); void *r = raft_new();
raft_cbs_t funcs = { raft_cbs_t funcs = {
.log_pop = __log_pop,
.log_get_node_id = __logentry_get_node_id .log_get_node_id = __logentry_get_node_id
}; };
raft_log_cbs_t log_funcs = {
.log_pop = __log_pop
};
raft_set_callbacks(r, &funcs, queue); raft_set_callbacks(r, &funcs, queue);
void *l; void *l;
...@@ -554,7 +568,7 @@ void TestLog_delete_after_polling_from_double_append(CuTest * tc) ...@@ -554,7 +568,7 @@ void TestLog_delete_after_polling_from_double_append(CuTest * tc)
e4.id = 4; e4.id = 4;
l = log_alloc(1); l = log_alloc(1);
log_set_callbacks(l, &funcs, r); log_set_callbacks(l, &log_funcs, r);
raft_entry_t* ety; raft_entry_t* ety;
...@@ -582,9 +596,11 @@ void TestLog_get_from_idx_with_base_off_by_one(CuTest * tc) ...@@ -582,9 +596,11 @@ void TestLog_get_from_idx_with_base_off_by_one(CuTest * tc)
void* queue = llqueue_new(); void* queue = llqueue_new();
void *r = raft_new(); void *r = raft_new();
raft_cbs_t funcs = { raft_cbs_t funcs = {
.log_pop = __log_pop,
.log_get_node_id = __logentry_get_node_id .log_get_node_id = __logentry_get_node_id
}; };
raft_log_cbs_t log_funcs = {
.log_pop = __log_pop
};
raft_set_callbacks(r, &funcs, queue); raft_set_callbacks(r, &funcs, queue);
void *l; void *l;
...@@ -597,7 +613,7 @@ void TestLog_get_from_idx_with_base_off_by_one(CuTest * tc) ...@@ -597,7 +613,7 @@ void TestLog_get_from_idx_with_base_off_by_one(CuTest * tc)
e2.id = 2; e2.id = 2;
l = log_alloc(1); l = log_alloc(1);
log_set_callbacks(l, &funcs, r); log_set_callbacks(l, &log_funcs, r);
raft_entry_t* ety; raft_entry_t* ety;
...@@ -617,8 +633,9 @@ void TestLog_get_from_idx_with_base_off_by_one(CuTest * tc) ...@@ -617,8 +633,9 @@ void TestLog_get_from_idx_with_base_off_by_one(CuTest * tc)
CuAssertIntEquals(tc, n_etys, 0); CuAssertIntEquals(tc, n_etys, 0);
/* now get the correct index */ /* now get the correct index */
ety = log_get_from_idx(l, 2, &n_etys); raft_entry_t** e;
CuAssertPtrNotNull(tc, ety); e = log_get_from_idx(l, 2, &n_etys);
CuAssertPtrNotNull(tc, e);
CuAssertIntEquals(tc, n_etys, 1); CuAssertIntEquals(tc, n_etys, 1);
CuAssertIntEquals(tc, ety->id, 2); CuAssertIntEquals(tc, e[0]->id, 2);
} }
#include <stdbool.h>
#include <assert.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <stdint.h>
#include "CuTest.h"
#include "linked_list_queue.h"
#include "raft.h"
static int __logentry_get_node_id(
raft_server_t* raft,
void *udata,
raft_entry_t *ety,
raft_index_t ety_idx
)
{
return 0;
}
static int __log_offer(
raft_server_t* raft,
void *user_data,
raft_entry_t *entry,
raft_index_t entry_idx
)
{
CuAssertIntEquals((CuTest*)raft, 1, entry_idx);
return 0;
}
static int __log_pop(
raft_server_t* raft,
void *user_data,
raft_entry_t *entry,
raft_index_t entry_idx
)
{
raft_entry_t* copy = malloc(sizeof(*entry));
memcpy(copy, entry, sizeof(*entry));
llqueue_offer(user_data, copy);
return 0;
}
static int __log_pop_failing(
raft_server_t* raft,
void *user_data,
raft_entry_t *entry,
raft_index_t entry_idx
)
{
return -1;
}
static const raft_log_impl_t *impl = &raft_log_internal_impl;
void TestLogImpl_new_is_empty(CuTest * tc)
{
void *l;
l = impl->init(NULL, NULL);
CuAssertTrue(tc, 0 == impl->count(l));
CuAssertTrue(tc, 1 == impl->first_idx(l));
}
void TestLogImpl_append_is_not_empty(CuTest * tc)
{
void *l;
raft_entry_t e;
void *r = raft_new();
memset(&e, 0, sizeof(raft_entry_t));
e.id = 1;
l = impl->init(r, NULL);
CuAssertIntEquals(tc, 0, impl->append(l, &e));
CuAssertIntEquals(tc, 1, impl->count(l));
}
void TestLogImpl_get_at_idx(CuTest * tc)
{
void *l;
raft_entry_t e1, e2, e3;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
l = impl->init(NULL, NULL);
e1.id = 1;
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
e2.id = 2;
CuAssertIntEquals(tc, 0, impl->append(l, &e2));
e3.id = 3;
CuAssertIntEquals(tc, 0, impl->append(l, &e3));
CuAssertIntEquals(tc, 3, impl->count(l));
CuAssertIntEquals(tc, e1.id, impl->get(l, 1)->id);
CuAssertIntEquals(tc, e2.id, impl->get(l, 2)->id);
CuAssertIntEquals(tc, e3.id, impl->get(l, 3)->id);
}
void TestLogImpl_get_at_idx_returns_null_where_out_of_bounds(CuTest * tc)
{
void *l;
raft_entry_t e1;
memset(&e1, 0, sizeof(raft_entry_t));
l = impl->init(NULL, NULL);
CuAssertTrue(tc, NULL == impl->get(l, 0));
CuAssertTrue(tc, NULL == impl->get(l, 1));
e1.id = 1;
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
CuAssertTrue(tc, NULL == impl->get(l, 2));
}
static void event_entry_enqueue(void *arg, raft_entry_t *e, raft_index_t idx)
{
raft_entry_t* copy = malloc(sizeof(*e));
memcpy(copy, e, sizeof(*e));
llqueue_offer(arg, copy);
}
void TestLogImpl_pop(CuTest * tc)
{
void* queue = llqueue_new();
void *l;
raft_entry_t e1, e2, e3;
void *r = raft_new();
raft_cbs_t funcs = {
.log_get_node_id = __logentry_get_node_id
};
raft_set_callbacks(r, &funcs, NULL);
l = impl->init(r, NULL);
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
e1.id = 1;
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
e2.id = 2;
CuAssertIntEquals(tc, 0, impl->append(l, &e2));
e3.id = 3;
CuAssertIntEquals(tc, 0, impl->append(l, &e3));
CuAssertIntEquals(tc, 3, impl->count(l));
CuAssertIntEquals(tc, 3, impl->current_idx(l));
impl->pop(l, 3, event_entry_enqueue, queue);
CuAssertIntEquals(tc, 2, impl->count(l));
CuAssertIntEquals(tc, e3.id, ((raft_entry_t*)llqueue_poll(queue))->id);
CuAssertTrue(tc, NULL == impl->get(l, 3));
impl->pop(l, 2, NULL, NULL);
CuAssertIntEquals(tc, 1, impl->count(l));
CuAssertTrue(tc, NULL == impl->get(l, 2));
impl->pop(l, 1, NULL, NULL);
CuAssertIntEquals(tc, 0, impl->count(l));
CuAssertTrue(tc, NULL == impl->get(l, 1));
}
void TestLogImpl_pop_onwards(CuTest * tc)
{
void *r = raft_new();
void *l;
raft_entry_t e1, e2, e3;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
l = impl->init(r, NULL);
e1.id = 1;
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
e2.id = 2;
CuAssertIntEquals(tc, 0, impl->append(l, &e2));
e3.id = 3;
CuAssertIntEquals(tc, 0, impl->append(l, &e3));
CuAssertIntEquals(tc, 3, impl->count(l));
/* even 3 gets deleted */
impl->pop(l, 2, NULL, NULL);
CuAssertIntEquals(tc, 1, impl->count(l));
CuAssertIntEquals(tc, e1.id, impl->get(l, 1)->id);
CuAssertTrue(tc, NULL == impl->get(l, 2));
CuAssertTrue(tc, NULL == impl->get(l, 3));
}
void TestLogImpl_pop_fails_for_idx_zero(CuTest * tc)
{
void *r = raft_new();
void *l;
raft_entry_t e1, e2, e3, e4;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
memset(&e4, 0, sizeof(raft_entry_t));
e1.id = 1;
e2.id = 2;
e3.id = 3;
e4.id = 4;
l = impl->init(r, NULL);
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
CuAssertIntEquals(tc, 0, impl->append(l, &e2));
CuAssertIntEquals(tc, 0, impl->append(l, &e3));
CuAssertIntEquals(tc, 0, impl->append(l, &e4));
CuAssertIntEquals(tc, impl->pop(l, 0, NULL, NULL), -1);
}
void TestLogImpl_poll(CuTest * tc)
{
void *r = raft_new();
void *l;
raft_entry_t e1, e2, e3;
l = impl->init(r, NULL);
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
e1.id = 1;
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
CuAssertIntEquals(tc, 1, impl->current_idx(l));
e2.id = 2;
CuAssertIntEquals(tc, 0, impl->append(l, &e2));
CuAssertIntEquals(tc, 2, impl->current_idx(l));
e3.id = 3;
CuAssertIntEquals(tc, 0, impl->append(l, &e3));
CuAssertIntEquals(tc, 3, impl->count(l));
CuAssertIntEquals(tc, 3, impl->current_idx(l));
raft_entry_t *ety;
/* remove 1st */
ety = NULL;
CuAssertIntEquals(tc, impl->poll(l, 2), 0);
CuAssertIntEquals(tc, 2, impl->count(l));
CuAssertIntEquals(tc, 2, impl->first_idx(l));
CuAssertTrue(tc, NULL == impl->get(l, 1));
CuAssertTrue(tc, NULL != impl->get(l, 2));
CuAssertIntEquals(tc, impl->get(l, 2)->id, 2);
CuAssertTrue(tc, NULL != impl->get(l, 3));
CuAssertIntEquals(tc, impl->get(l, 3)->id, 3);
CuAssertIntEquals(tc, 3, impl->current_idx(l));
/* remove 2nd */
ety = NULL;
CuAssertIntEquals(tc, impl->poll(l, 3), 0);
CuAssertIntEquals(tc, 1, impl->count(l));
CuAssertTrue(tc, NULL == impl->get(l, 1));
CuAssertTrue(tc, NULL == impl->get(l, 2));
CuAssertTrue(tc, NULL != impl->get(l, 3));
CuAssertIntEquals(tc, impl->get(l, 3)->id, 3);
CuAssertIntEquals(tc, 3, impl->current_idx(l));
/* remove 3rd */
ety = NULL;
CuAssertIntEquals(tc, impl->poll(l, 4), 0);
CuAssertIntEquals(tc, 0, impl->count(l));
CuAssertTrue(tc, NULL == impl->get(l, 1));
CuAssertTrue(tc, NULL == impl->get(l, 2));
CuAssertTrue(tc, NULL == impl->get(l, 3));
CuAssertIntEquals(tc, 3, impl->current_idx(l));
}
void TestLogImpl_reset_after_compaction(CuTest * tc)
{
void *l;
l = impl->init(NULL, NULL);
CuAssertIntEquals(tc, 0, impl->current_idx(l));
/* Reset first_idx==1 indicates our next entry is going to be 10 */
impl->reset(l, 10, 1);
/* Current index is 9 (first_idx - 1), but log is empty */
CuAssertIntEquals(tc, 9, impl->current_idx(l));
CuAssertIntEquals(tc, 0, impl->count(l));
}
void TestLogImpl_load_from_snapshot_clears_log(CuTest * tc)
{
void *l;
raft_entry_t e1, e2, e3;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
l = impl->init(NULL, NULL);
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
CuAssertIntEquals(tc, 0, impl->append(l, &e2));
CuAssertIntEquals(tc, 2, impl->count(l));
CuAssertIntEquals(tc, 2, impl->current_idx(l));
impl->reset(l, 10, 1);
CuAssertIntEquals(tc, 0, impl->count(l));
CuAssertIntEquals(tc, 9, impl->current_idx(l));
}
void TestLogImpl_pop_after_polling(CuTest * tc)
{
void *l;
raft_entry_t e1, e2, e3, e4;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
memset(&e4, 0, sizeof(raft_entry_t));
e1.id = 1;
e2.id = 2;
e3.id = 3;
e4.id = 4;
l = impl->init(NULL, NULL);
/* append */
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
CuAssertIntEquals(tc, 1, impl->count(l));
CuAssertIntEquals(tc, 1, impl->first_idx(l));
CuAssertIntEquals(tc, 1, impl->current_idx(l));
/* poll */
CuAssertIntEquals(tc, impl->poll(l, 2), 0);
CuAssertIntEquals(tc, 0, impl->count(l));
CuAssertIntEquals(tc, 2, impl->first_idx(l));
CuAssertIntEquals(tc, 1, impl->current_idx(l));
/* append */
CuAssertIntEquals(tc, 0, impl->append(l, &e2));
CuAssertIntEquals(tc, 1, impl->count(l));
CuAssertIntEquals(tc, 2, impl->current_idx(l));
/* poll */
CuAssertIntEquals(tc, impl->pop(l, 1, NULL, NULL), 0);
CuAssertIntEquals(tc, 0, impl->count(l));
CuAssertIntEquals(tc, 1, impl->current_idx(l));
}
void TestLogImpl_pop_after_polling_from_double_append(CuTest * tc)
{
void *r = raft_new();
void *l;
raft_entry_t e1, e2, e3, e4;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
memset(&e3, 0, sizeof(raft_entry_t));
memset(&e4, 0, sizeof(raft_entry_t));
e1.id = 1;
e2.id = 2;
e3.id = 3;
e4.id = 4;
l = impl->init(r, NULL);
/* append append */
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
CuAssertIntEquals(tc, 0, impl->append(l, &e2));
CuAssertIntEquals(tc, 2, impl->count(l));
/* poll */
CuAssertIntEquals(tc, impl->poll(l, 2), 0);
CuAssertIntEquals(tc, impl->get(l, 2)->id, 2);
CuAssertIntEquals(tc, 1, impl->count(l));
/* append */
CuAssertIntEquals(tc, 0, impl->append(l, &e3));
CuAssertIntEquals(tc, 2, impl->count(l));
/* pop */
CuAssertIntEquals(tc, impl->pop(l, 1, NULL, NULL), 0);
CuAssertIntEquals(tc, 0, impl->count(l));
}
void TestLogImpl_get_from_idx_with_base_off_by_one(CuTest * tc)
{
void *r = raft_new();
void *l;
raft_entry_t e1, e2;
memset(&e1, 0, sizeof(raft_entry_t));
memset(&e2, 0, sizeof(raft_entry_t));
e1.id = 1;
e2.id = 2;
l = impl->init(r, NULL);
/* append append */
CuAssertIntEquals(tc, 0, impl->append(l, &e1));
CuAssertIntEquals(tc, 0, impl->append(l, &e2));
CuAssertIntEquals(tc, 2, impl->count(l));
/* poll */
CuAssertIntEquals(tc, impl->poll(l, 2), 0);
CuAssertIntEquals(tc, 1, impl->count(l));
/* get off-by-one index */
raft_entry_t *e[1];
CuAssertIntEquals(tc, impl->get_batch(l, 1, 1, e), 0);
/* now get the correct index */
CuAssertIntEquals(tc, impl->get_batch(l, 2, 1, e), 1);
CuAssertIntEquals(tc, e[0]->id, 2);
}
This diff is collapsed.
...@@ -11,6 +11,8 @@ ...@@ -11,6 +11,8 @@
#include "raft_private.h" #include "raft_private.h"
#include "mock_send_functions.h" #include "mock_send_functions.h"
#include "helpers.h"
static int __raft_persist_term( static int __raft_persist_term(
raft_server_t* raft, raft_server_t* raft,
void *udata, void *udata,
...@@ -108,15 +110,13 @@ void TestRaft_leader_begin_snapshot_fails_if_no_logs_to_compact(CuTest * tc) ...@@ -108,15 +110,13 @@ void TestRaft_leader_begin_snapshot_fails_if_no_logs_to_compact(CuTest * tc)
CuAssertIntEquals(tc, 0, raft_get_log_count(r)); CuAssertIntEquals(tc, 0, raft_get_log_count(r));
/* entry message */ /* entry message */
msg_entry_t ety = {}; msg_entry_t *ety = __MAKE_ENTRY(1, 1, "entry");
ety.id = 1;
ety.data.buf = "entry";
ety.data.len = strlen("entry");
/* receive entry */ /* receive entry */
raft_recv_entry(r, &ety, &cr); raft_recv_entry(r, ety, &cr);
ety.id = 2;
raft_recv_entry(r, &ety, &cr); ety = __MAKE_ENTRY(2, 1, "entry");
raft_recv_entry(r, ety, &cr);
CuAssertIntEquals(tc, 2, raft_get_log_count(r)); CuAssertIntEquals(tc, 2, raft_get_log_count(r));
CuAssertIntEquals(tc, -1, raft_begin_snapshot(r, 0)); CuAssertIntEquals(tc, -1, raft_begin_snapshot(r, 0));
...@@ -143,15 +143,11 @@ void TestRaft_leader_will_not_apply_entry_if_snapshot_is_in_progress(CuTest * tc ...@@ -143,15 +143,11 @@ void TestRaft_leader_will_not_apply_entry_if_snapshot_is_in_progress(CuTest * tc
CuAssertIntEquals(tc, 0, raft_get_log_count(r)); CuAssertIntEquals(tc, 0, raft_get_log_count(r));
/* entry message */ /* entry message */
msg_entry_t ety = {}; msg_entry_t *ety = __MAKE_ENTRY(1, 1, "entry");
ety.id = 1;
ety.data.buf = "entry";
ety.data.len = strlen("entry");
/* receive entry */ /* receive entry */
raft_recv_entry(r, &ety, &cr); raft_recv_entry(r, ety, &cr);
ety.id = 1; raft_recv_entry(r, ety, &cr);
raft_recv_entry(r, &ety, &cr);
raft_set_commit_idx(r, 1); raft_set_commit_idx(r, 1);
CuAssertIntEquals(tc, 2, raft_get_log_count(r)); CuAssertIntEquals(tc, 2, raft_get_log_count(r));
...@@ -199,13 +195,10 @@ void TestRaft_leader_snapshot_begin_fails_if_less_than_2_logs_to_compact(CuTest ...@@ -199,13 +195,10 @@ void TestRaft_leader_snapshot_begin_fails_if_less_than_2_logs_to_compact(CuTest
CuAssertIntEquals(tc, 0, raft_get_log_count(r)); CuAssertIntEquals(tc, 0, raft_get_log_count(r));
/* entry message */ /* entry message */
msg_entry_t ety = {}; msg_entry_t *ety = __MAKE_ENTRY(1, 1, "entry");
ety.id = 1;
ety.data.buf = "entry";
ety.data.len = strlen("entry");
/* receive entry */ /* receive entry */
raft_recv_entry(r, &ety, &cr); raft_recv_entry(r, ety, &cr);
raft_set_commit_idx(r, 1); raft_set_commit_idx(r, 1);
CuAssertIntEquals(tc, 1, raft_get_log_count(r)); CuAssertIntEquals(tc, 1, raft_get_log_count(r));
CuAssertIntEquals(tc, -1, raft_begin_snapshot(r, 0)); CuAssertIntEquals(tc, -1, raft_begin_snapshot(r, 0));
...@@ -232,27 +225,24 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc) ...@@ -232,27 +225,24 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc)
CuAssertIntEquals(tc, 0, raft_get_log_count(r)); CuAssertIntEquals(tc, 0, raft_get_log_count(r));
/* entry message */ /* entry message */
msg_entry_t ety = {}; msg_entry_t *ety = __MAKE_ENTRY(1, 1, "entry");
ety.id = 1;
ety.data.buf = "entry";
ety.data.len = strlen("entry");
/* receive entry */ /* receive entry */
raft_recv_entry(r, &ety, &cr); raft_recv_entry(r, ety, &cr);
ety.id = 2;
raft_recv_entry(r, &ety, &cr); ety = __MAKE_ENTRY(2, 1, "entry");
ety.id = 3; raft_recv_entry(r, ety, &cr);
raft_recv_entry(r, &ety, &cr); ety = __MAKE_ENTRY(3, 1, "entry");
raft_recv_entry(r, ety, &cr);
raft_set_commit_idx(r, 2); raft_set_commit_idx(r, 2);
CuAssertIntEquals(tc, 3, raft_get_log_count(r)); CuAssertIntEquals(tc, 3, raft_get_log_count(r));
CuAssertIntEquals(tc, 2, raft_get_num_snapshottable_logs(r)); CuAssertIntEquals(tc, 2, raft_get_num_snapshottable_logs(r));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0)); CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
raft_entry_t* _ety;
int i = raft_get_first_entry_idx(r); int i = raft_get_first_entry_idx(r);
for (; i < raft_get_commit_idx(r); i++) for (; i < raft_get_commit_idx(r); i++)
CuAssertIntEquals(tc, 0, raft_poll_entry(r, &_ety)); CuAssertIntEquals(tc, 0, raft_poll_entry(r));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0)); CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
CuAssertIntEquals(tc, 0, raft_end_snapshot(r)); CuAssertIntEquals(tc, 0, raft_end_snapshot(r));
...@@ -284,27 +274,23 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted2(CuTest * tc) ...@@ -284,27 +274,23 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted2(CuTest * tc)
CuAssertIntEquals(tc, 0, raft_get_log_count(r)); CuAssertIntEquals(tc, 0, raft_get_log_count(r));
/* entry message */ /* entry message */
msg_entry_t ety = {}; msg_entry_t *ety = __MAKE_ENTRY(1, 1, "entry");
ety.id = 1;
ety.data.buf = "entry";
ety.data.len = strlen("entry");
/* receive entry */ /* receive entry */
raft_recv_entry(r, &ety, &cr); raft_recv_entry(r, ety, &cr);
ety.id = 2; ety = __MAKE_ENTRY(2, 1, "entry");
raft_recv_entry(r, &ety, &cr); raft_recv_entry(r, ety, &cr);
ety.id = 3; ety = __MAKE_ENTRY(3, 1, "entry");
raft_recv_entry(r, &ety, &cr); raft_recv_entry(r, ety, &cr);
raft_set_commit_idx(r, 2); raft_set_commit_idx(r, 2);
CuAssertIntEquals(tc, 3, raft_get_log_count(r)); CuAssertIntEquals(tc, 3, raft_get_log_count(r));
CuAssertIntEquals(tc, 2, raft_get_num_snapshottable_logs(r)); CuAssertIntEquals(tc, 2, raft_get_num_snapshottable_logs(r));
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0)); CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
raft_entry_t* _ety;
int i = raft_get_first_entry_idx(r); int i = raft_get_first_entry_idx(r);
for (; i <= raft_get_commit_idx(r); i++) for (; i <= raft_get_commit_idx(r); i++)
CuAssertIntEquals(tc, 0, raft_poll_entry(r, &_ety)); CuAssertIntEquals(tc, 0, raft_poll_entry(r));
CuAssertIntEquals(tc, 0, raft_end_snapshot(r)); CuAssertIntEquals(tc, 0, raft_end_snapshot(r));
CuAssertIntEquals(tc, 0, raft_get_num_snapshottable_logs(r)); CuAssertIntEquals(tc, 0, raft_get_num_snapshottable_logs(r));
...@@ -333,15 +319,12 @@ void TestRaft_joinee_needs_to_get_snapshot(CuTest * tc) ...@@ -333,15 +319,12 @@ void TestRaft_joinee_needs_to_get_snapshot(CuTest * tc)
CuAssertIntEquals(tc, 0, raft_get_log_count(r)); CuAssertIntEquals(tc, 0, raft_get_log_count(r));
/* entry message */ /* entry message */
msg_entry_t ety = {}; msg_entry_t *ety = __MAKE_ENTRY(1, 1, "entry");
ety.id = 1;
ety.data.buf = "entry";
ety.data.len = strlen("entry");
/* receive entry */ /* receive entry */
raft_recv_entry(r, &ety, &cr); raft_recv_entry(r, ety, &cr);
ety.id = 2; ety = __MAKE_ENTRY(2, 1, "entry");
raft_recv_entry(r, &ety, &cr); raft_recv_entry(r, ety, &cr);
raft_set_commit_idx(r, 1); raft_set_commit_idx(r, 1);
CuAssertIntEquals(tc, 2, raft_get_log_count(r)); CuAssertIntEquals(tc, 2, raft_get_log_count(r));
CuAssertIntEquals(tc, 1, raft_get_num_snapshottable_logs(r)); CuAssertIntEquals(tc, 1, raft_get_num_snapshottable_logs(r));
...@@ -367,10 +350,7 @@ void TestRaft_follower_load_from_snapshot(CuTest * tc) ...@@ -367,10 +350,7 @@ void TestRaft_follower_load_from_snapshot(CuTest * tc)
CuAssertIntEquals(tc, 0, raft_get_log_count(r)); CuAssertIntEquals(tc, 0, raft_get_log_count(r));
/* entry message */ /* entry message */
msg_entry_t ety = {}; msg_entry_t *ety = __MAKE_ENTRY(1, 1, "entry");
ety.id = 1;
ety.data.buf = "entry";
ety.data.len = strlen("entry");
CuAssertIntEquals(tc, 0, raft_get_log_count(r)); CuAssertIntEquals(tc, 0, raft_get_log_count(r));
CuAssertIntEquals(tc, 0, raft_begin_load_snapshot(r, 5, 5)); CuAssertIntEquals(tc, 0, raft_begin_load_snapshot(r, 5, 5));
...@@ -383,10 +363,10 @@ void TestRaft_follower_load_from_snapshot(CuTest * tc) ...@@ -383,10 +363,10 @@ void TestRaft_follower_load_from_snapshot(CuTest * tc)
CuAssertIntEquals(tc, 0, raft_periodic(r, 1000)); CuAssertIntEquals(tc, 0, raft_periodic(r, 1000));
/* current idx means snapshot was unnecessary */ /* current idx means snapshot was unnecessary */
ety.id = 2; ety = __MAKE_ENTRY(2, 1, "entry");
raft_append_entry(r, &ety); raft_append_entry(r, ety);
ety.id = 3; ety = __MAKE_ENTRY(3, 1, "entry");
raft_append_entry(r, &ety); raft_append_entry(r, ety);
raft_set_commit_idx(r, 7); raft_set_commit_idx(r, 7);
CuAssertIntEquals(tc, -1, raft_begin_load_snapshot(r, 6, 5)); CuAssertIntEquals(tc, -1, raft_begin_load_snapshot(r, 6, 5));
CuAssertIntEquals(tc, 7, raft_get_commit_idx(r)); CuAssertIntEquals(tc, 7, raft_get_commit_idx(r));
...@@ -422,12 +402,6 @@ void TestRaft_follower_load_from_snapshot_fails_if_already_loaded(CuTest * tc) ...@@ -422,12 +402,6 @@ void TestRaft_follower_load_from_snapshot_fails_if_already_loaded(CuTest * tc)
raft_set_state(r, RAFT_STATE_FOLLOWER); raft_set_state(r, RAFT_STATE_FOLLOWER);
CuAssertIntEquals(tc, 0, raft_get_log_count(r)); CuAssertIntEquals(tc, 0, raft_get_log_count(r));
/* entry message */
msg_entry_t ety = {};
ety.id = 1;
ety.data.buf = "entry";
ety.data.len = strlen("entry");
CuAssertIntEquals(tc, 0, raft_get_log_count(r)); CuAssertIntEquals(tc, 0, raft_get_log_count(r));
CuAssertIntEquals(tc, 0, raft_begin_load_snapshot(r, 5, 5)); CuAssertIntEquals(tc, 0, raft_begin_load_snapshot(r, 5, 5));
CuAssertIntEquals(tc, 0, raft_end_load_snapshot(r)); CuAssertIntEquals(tc, 0, raft_end_load_snapshot(r));
...@@ -454,21 +428,9 @@ void TestRaft_follower_load_from_snapshot_does_not_break_cluster_safety(CuTest * ...@@ -454,21 +428,9 @@ void TestRaft_follower_load_from_snapshot_does_not_break_cluster_safety(CuTest *
CuAssertIntEquals(tc, 0, raft_get_log_count(r)); CuAssertIntEquals(tc, 0, raft_get_log_count(r));
/* entry message */ /* entry message */
msg_entry_t ety = {}; __APPEND_ENTRY(r, 1, 1, "entry");
ety.id = 1; __APPEND_ENTRY(r, 2, 1, "entry");
ety.data.buf = "entry"; __APPEND_ENTRY(r, 3, 1, "entry");
ety.data.len = strlen("entry");
raft_append_entry(r, &ety);
ety.id = 2;
ety.data.buf = "entry";
ety.data.len = strlen("entry");
raft_append_entry(r, &ety);
ety.id = 3;
ety.data.buf = "entry";
ety.data.len = strlen("entry");
raft_append_entry(r, &ety);
CuAssertIntEquals(tc, -1, raft_begin_load_snapshot(r, 2, 2)); CuAssertIntEquals(tc, -1, raft_begin_load_snapshot(r, 2, 2));
} }
...@@ -489,12 +451,6 @@ void TestRaft_follower_load_from_snapshot_fails_if_log_is_newer(CuTest * tc) ...@@ -489,12 +451,6 @@ void TestRaft_follower_load_from_snapshot_fails_if_log_is_newer(CuTest * tc)
raft_set_last_applied_idx(r, 5); raft_set_last_applied_idx(r, 5);
/* entry message */
msg_entry_t ety = {};
ety.id = 1;
ety.data.buf = "entry";
ety.data.len = strlen("entry");
CuAssertIntEquals(tc, -1, raft_begin_load_snapshot(r, 2, 2)); CuAssertIntEquals(tc, -1, raft_begin_load_snapshot(r, 2, 2));
} }
...@@ -515,27 +471,7 @@ void TestRaft_leader_sends_appendentries_when_node_next_index_was_compacted(CuTe ...@@ -515,27 +471,7 @@ void TestRaft_leader_sends_appendentries_when_node_next_index_was_compacted(CuTe
raft_add_node(r, NULL, 3, 0); raft_add_node(r, NULL, 3, 0);
/* entry 1 */ /* entry 1 */
char *str = "aaa"; __APPEND_ENTRIES_SEQ_ID(r, 3, 1, 1, "aaa");
raft_entry_t ety = {};
ety.term = 1;
ety.id = 1;
ety.data.buf = str;
ety.data.len = 3;
raft_append_entry(r, &ety);
/* entry 2 */
ety.term = 1;
ety.id = 2;
ety.data.buf = str;
ety.data.len = 3;
raft_append_entry(r, &ety);
/* entry 3 */
ety.term = 1;
ety.id = 3;
ety.data.buf = str;
ety.data.len = 3;
raft_append_entry(r, &ety);
CuAssertIntEquals(tc, 3, raft_get_current_idx(r)); CuAssertIntEquals(tc, 3, raft_get_current_idx(r));
/* compact entry 1 & 2 */ /* compact entry 1 & 2 */
...@@ -577,23 +513,20 @@ void TestRaft_recv_entry_fails_if_snapshot_in_progress(CuTest* tc) ...@@ -577,23 +513,20 @@ void TestRaft_recv_entry_fails_if_snapshot_in_progress(CuTest* tc)
CuAssertIntEquals(tc, 0, raft_get_log_count(r)); CuAssertIntEquals(tc, 0, raft_get_log_count(r));
/* entry message */ /* entry message */
msg_entry_t ety = {}; msg_entry_t *ety = __MAKE_ENTRY(1, 1, "entry");
ety.id = 1;
ety.data.buf = "entry";
ety.data.len = strlen("entry");
/* receive entry */ /* receive entry */
raft_recv_entry(r, &ety, &cr); raft_recv_entry(r, ety, &cr);
ety.id = 2; ety = __MAKE_ENTRY(2, 1, "entry");
raft_recv_entry(r, &ety, &cr); raft_recv_entry(r, ety, &cr);
CuAssertIntEquals(tc, 2, raft_get_log_count(r)); CuAssertIntEquals(tc, 2, raft_get_log_count(r));
raft_set_commit_idx(r, 1); raft_set_commit_idx(r, 1);
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0)); CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
ety.id = 3; ety = __MAKE_ENTRY(3, 1, "entry");
ety.type = RAFT_LOGTYPE_ADD_NODE; ety->type = RAFT_LOGTYPE_ADD_NODE;
CuAssertIntEquals(tc, RAFT_ERR_SNAPSHOT_IN_PROGRESS, raft_recv_entry(r, &ety, &cr)); CuAssertIntEquals(tc, RAFT_ERR_SNAPSHOT_IN_PROGRESS, raft_recv_entry(r, ety, &cr));
} }
void TestRaft_recv_entry_succeeds_if_snapshot_nonblocking_apply_is_set(CuTest* tc) void TestRaft_recv_entry_succeeds_if_snapshot_nonblocking_apply_is_set(CuTest* tc)
...@@ -615,23 +548,20 @@ void TestRaft_recv_entry_succeeds_if_snapshot_nonblocking_apply_is_set(CuTest* t ...@@ -615,23 +548,20 @@ void TestRaft_recv_entry_succeeds_if_snapshot_nonblocking_apply_is_set(CuTest* t
CuAssertIntEquals(tc, 0, raft_get_log_count(r)); CuAssertIntEquals(tc, 0, raft_get_log_count(r));
/* entry message */ /* entry message */
msg_entry_t ety = {}; msg_entry_t *ety = __MAKE_ENTRY(1, 1, "entry");
ety.id = 1;
ety.data.buf = "entry";
ety.data.len = strlen("entry");
/* receive entry */ /* receive entry */
raft_recv_entry(r, &ety, &cr); raft_recv_entry(r, ety, &cr);
ety.id = 2; ety = __MAKE_ENTRY(2, 1, "entry");
raft_recv_entry(r, &ety, &cr); raft_recv_entry(r, ety, &cr);
CuAssertIntEquals(tc, 2, raft_get_log_count(r)); CuAssertIntEquals(tc, 2, raft_get_log_count(r));
raft_set_commit_idx(r, 1); raft_set_commit_idx(r, 1);
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, RAFT_SNAPSHOT_NONBLOCKING_APPLY)); CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, RAFT_SNAPSHOT_NONBLOCKING_APPLY));
ety.id = 3; ety = __MAKE_ENTRY(3, 1, "entry");
ety.type = RAFT_LOGTYPE_ADD_NODE; ety->type = RAFT_LOGTYPE_ADD_NODE;
CuAssertIntEquals(tc, 0, raft_recv_entry(r, &ety, &cr)); CuAssertIntEquals(tc, 0, raft_recv_entry(r, ety, &cr));
} }
...@@ -659,11 +589,7 @@ void TestRaft_follower_recv_appendentries_is_successful_when_previous_log_idx_eq ...@@ -659,11 +589,7 @@ void TestRaft_follower_recv_appendentries_is_successful_when_previous_log_idx_eq
ae.prev_log_idx = 2; ae.prev_log_idx = 2;
ae.prev_log_term = 2; ae.prev_log_term = 2;
/* include entries */ /* include entries */
msg_entry_t e[5]; ae.entries = __MAKE_ENTRY_ARRAY(3, 3, NULL);
memset(&e, 0, sizeof(msg_entry_t) * 4);
e[0].term = 3;
e[0].id = 3;
ae.entries = e;
ae.n_entries = 1; ae.n_entries = 1;
CuAssertIntEquals(tc, 0, raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer)); CuAssertIntEquals(tc, 0, raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer));
CuAssertIntEquals(tc, 1, aer.success); CuAssertIntEquals(tc, 1, aer.success);
...@@ -723,24 +649,21 @@ void TestRaft_cancel_snapshot_restores_state(CuTest* tc) ...@@ -723,24 +649,21 @@ void TestRaft_cancel_snapshot_restores_state(CuTest* tc)
CuAssertIntEquals(tc, 0, raft_get_log_count(r)); CuAssertIntEquals(tc, 0, raft_get_log_count(r));
/* single entry */ /* single entry */
msg_entry_t ety = {}; msg_entry_t *ety = __MAKE_ENTRY(1, 1, "entry");
ety.id = 1; raft_recv_entry(r, ety, &cr);
ety.data.buf = "entry";
ety.data.len = strlen("entry"); ety = __MAKE_ENTRY(2, 1, "entry");
raft_recv_entry(r, &ety, &cr); raft_recv_entry(r, ety, &cr);
ety.id = 2;
raft_recv_entry(r, &ety, &cr);
raft_set_commit_idx(r, 2); raft_set_commit_idx(r, 2);
CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0)); CuAssertIntEquals(tc, 0, raft_begin_snapshot(r, 0));
CuAssertIntEquals(tc, 0, raft_end_snapshot(r)); CuAssertIntEquals(tc, 0, raft_end_snapshot(r));
/* more entries */ /* more entries */
ety.id = 3; ety = __MAKE_ENTRY(3, 1, "entry");
raft_recv_entry(r, &ety, &cr); raft_recv_entry(r, ety, &cr);
ety.id = 4; ety = __MAKE_ENTRY(4, 1, "entry");
raft_recv_entry(r, &ety, &cr); raft_recv_entry(r, ety, &cr);
CuAssertIntEquals(tc, 2, raft_get_log_count(r)); CuAssertIntEquals(tc, 2, raft_get_log_count(r));
CuAssertIntEquals(tc, 2, raft_get_snapshot_last_idx(r)); CuAssertIntEquals(tc, 2, raft_get_snapshot_last_idx(r));
...@@ -777,26 +700,13 @@ void TestRaft_leader_sends_snapshot_if_log_was_compacted(CuTest* tc) ...@@ -777,26 +700,13 @@ void TestRaft_leader_sends_snapshot_if_log_was_compacted(CuTest* tc)
/* entry 1 */ /* entry 1 */
char *str = "aaa"; char *str = "aaa";
raft_entry_t ety = {}; __APPEND_ENTRY(r, 1, 1, str);
ety.term = 1;
ety.id = 1;
ety.data.buf = str;
ety.data.len = 3;
raft_append_entry(r, &ety);
/* entry 2 */ /* entry 2 */
ety.term = 1; __APPEND_ENTRY(r, 2, 1, str);
ety.id = 2;
ety.data.buf = str;
ety.data.len = 3;
raft_append_entry(r, &ety);
/* entry 3 */ /* entry 3 */
ety.term = 1; __APPEND_ENTRY(r, 3, 1, str);
ety.id = 3;
ety.data.buf = str;
ety.data.len = 3;
raft_append_entry(r, &ety);
CuAssertIntEquals(tc, 3, raft_get_current_idx(r)); CuAssertIntEquals(tc, 3, raft_get_current_idx(r));
/* compact entry 1 & 2 */ /* compact entry 1 & 2 */
......
...@@ -204,7 +204,7 @@ def raft_logentry_pop(raft, udata, ety, ety_idx): ...@@ -204,7 +204,7 @@ def raft_logentry_pop(raft, udata, ety, ety_idx):
def raft_logentry_get_node_id(raft, udata, ety, ety_idx): def raft_logentry_get_node_id(raft, udata, ety, ety_idx):
change_entry = ffi.from_handle(ety.data.buf) change_entry = ffi.from_handle(lib.raft_entry_getdata(ety))
assert isinstance(change_entry, ChangeRaftEntry) assert isinstance(change_entry, ChangeRaftEntry)
return change_entry.node_id return change_entry.node_id
...@@ -264,6 +264,15 @@ class Network(object): ...@@ -264,6 +264,15 @@ class Network(object):
self.num_compactions = 0 self.num_compactions = 0
self.latest_applied_log_idx = 0 self.latest_applied_log_idx = 0
def add_entry(self, etype, data):
data_handle = ffi.new_handle(data)
ety = lib.raft_entry_newdata(data_handle)
ety.term = 0
ety.id = self.new_entry_id()
ety.type = etype
self.entries.append((ety, data_handle, data))
return ety
def add_server(self, server): def add_server(self, server):
self.server_id += 1 self.server_id += 1
server.id = self.server_id server.id = self.server_id
...@@ -274,14 +283,8 @@ class Network(object): ...@@ -274,14 +283,8 @@ class Network(object):
def push_set_entry(self, k, v): def push_set_entry(self, k, v):
for sv in self.active_servers: for sv in self.active_servers:
if lib.raft_is_leader(sv.raft): if lib.raft_is_leader(sv.raft):
ety = ffi.new('msg_entry_t*') ety = self.add_entry(lib.RAFT_LOGTYPE_NORMAL,
ety.term = 0 SetRaftEntry(k, v))
ety.id = self.new_entry_id()
ety.type = lib.RAFT_LOGTYPE_NORMAL
change = ffi.new_handle(SetRaftEntry(k, v))
ety.data.buf = change
ety.data.len = ffi.sizeof(ety.data.buf)
self.entries.append((ety, change))
e = sv.recv_entry(ety) e = sv.recv_entry(ety)
assert e == 0 assert e == 0
break break
...@@ -371,9 +374,7 @@ class Network(object): ...@@ -371,9 +374,7 @@ class Network(object):
ffi.memmove(new_msg, msg, msg_size) ffi.memmove(new_msg, msg, msg_size)
if msg_type == 'msg_appendentries_t *': if msg_type == 'msg_appendentries_t *':
size_of_entries = ffi.sizeof(ffi.getctype('msg_entry_t')) * new_msg.n_entries new_msg.entries = lib.raft_entry_array_deepcopy(msg.entries, msg.n_entries)
new_msg.entries = ffi.cast('msg_entry_t*', lib.malloc(size_of_entries))
ffi.memmove(new_msg.entries, msg.entries, size_of_entries)
self.messages.append(Message(new_msg, sendor, sendee)) self.messages.append(Message(new_msg, sendor, sendee))
...@@ -487,14 +488,8 @@ class Network(object): ...@@ -487,14 +488,8 @@ class Network(object):
lib.raft_become_leader(server.raft) lib.raft_become_leader(server.raft)
# Configuration change entry to bootstrap other nodes # Configuration change entry to bootstrap other nodes
ety = ffi.new('msg_entry_t*') ety = self.add_entry(lib.RAFT_LOGTYPE_ADD_NODE,
ety.term = 0 ChangeRaftEntry(server.id))
ety.type = lib.RAFT_LOGTYPE_ADD_NODE
ety.id = self.new_entry_id()
change = ffi.new_handle(ChangeRaftEntry(server.id))
ety.data.buf = change
ety.data.len = ffi.sizeof(ety.data.buf)
self.entries.append((ety, change))
e = server.recv_entry(ety) e = server.recv_entry(ety)
assert e == 0 assert e == 0
...@@ -534,17 +529,10 @@ class Network(object): ...@@ -534,17 +529,10 @@ class Network(object):
server = RaftServer(self) server = RaftServer(self)
# Create a new configuration entry to be processed by the leader # Create a new configuration entry to be processed by the leader
ety = ffi.new('msg_entry_t*') ety = self.add_entry(lib.RAFT_LOGTYPE_ADD_NONVOTING_NODE,
ety.term = 0 ChangeRaftEntry(server.id))
ety.id = self.new_entry_id()
ety.type = lib.RAFT_LOGTYPE_ADD_NONVOTING_NODE
change = ffi.new_handle(ChangeRaftEntry(server.id))
ety.data.buf = change
ety.data.len = ffi.sizeof(ety.data.buf)
assert(lib.raft_entry_is_cfg_change(ety)) assert(lib.raft_entry_is_cfg_change(ety))
self.entries.append((ety, change))
e = leader.recv_entry(ety) e = leader.recv_entry(ety)
if 0 != e: if 0 != e:
logger.error(err2str(e)) logger.error(err2str(e))
...@@ -586,18 +574,11 @@ class Network(object): ...@@ -586,18 +574,11 @@ class Network(object):
return return
# Create a new configuration entry to be processed by the leader # Create a new configuration entry to be processed by the leader
ety = ffi.new('msg_entry_t*') ety = self.add_entry(lib.RAFT_LOGTYPE_DEMOTE_NODE,
ety.term = 0 ChangeRaftEntry(server.id))
ety.id = self.new_entry_id()
assert server.connection_status == NODE_CONNECTED assert server.connection_status == NODE_CONNECTED
ety.type = lib.RAFT_LOGTYPE_DEMOTE_NODE
change = ffi.new_handle(ChangeRaftEntry(server.id))
ety.data.buf = change
ety.data.len = ffi.sizeof(ety.data.buf)
assert(lib.raft_entry_is_cfg_change(ety)) assert(lib.raft_entry_is_cfg_change(ety))
self.entries.append((ety, change))
e = leader.recv_entry(ety) e = leader.recv_entry(ety)
if 0 != e: if 0 != e:
logger.error(err2str(e)) logger.error(err2str(e))
...@@ -659,16 +640,18 @@ class RaftServer(object): ...@@ -659,16 +640,18 @@ class RaftServer(object):
cbs.applylog = self.raft_applylog cbs.applylog = self.raft_applylog
cbs.persist_vote = self.raft_persist_vote cbs.persist_vote = self.raft_persist_vote
cbs.persist_term = self.raft_persist_term cbs.persist_term = self.raft_persist_term
cbs.log_offer = self.raft_logentry_offer
cbs.log_poll = self.raft_logentry_poll
cbs.log_pop = self.raft_logentry_pop
cbs.log_get_node_id = self.raft_logentry_get_node_id cbs.log_get_node_id = self.raft_logentry_get_node_id
cbs.node_has_sufficient_logs = self.raft_node_has_sufficient_logs cbs.node_has_sufficient_logs = self.raft_node_has_sufficient_logs
cbs.notify_membership_event = self.raft_notify_membership_event cbs.notify_membership_event = self.raft_notify_membership_event
cbs.log = self.raft_log cbs.log = self.raft_log
lib.raft_set_callbacks(self.raft, cbs, self.udata) log_cbs = ffi.new('raft_log_cbs_t*')
log_cbs.log_offer = self.raft_logentry_offer
log_cbs.log_poll = self.raft_logentry_poll
log_cbs.log_pop = self.raft_logentry_pop
lib.raft_set_callbacks(self.raft, cbs, self.udata)
lib.log_set_callbacks(lib.raft_get_log(self.raft), log_cbs, self.raft)
lib.raft_set_election_timeout(self.raft, 500) lib.raft_set_election_timeout(self.raft, 500)
self.fsm_dict = {} self.fsm_dict = {}
...@@ -827,7 +810,7 @@ class RaftServer(object): ...@@ -827,7 +810,7 @@ class RaftServer(object):
if e is not None: if e is not None:
return e return e
change = ffi.from_handle(ety.data.buf) change = ffi.from_handle(lib.raft_entry_getdata(ety))
if ety.type == lib.RAFT_LOGTYPE_NORMAL: if ety.type == lib.RAFT_LOGTYPE_NORMAL:
self.fsm_dict[change.key] = change.val self.fsm_dict[change.key] = change.val
...@@ -839,12 +822,9 @@ class RaftServer(object): ...@@ -839,12 +822,9 @@ class RaftServer(object):
# Follow up by removing the node by receiving new entry # Follow up by removing the node by receiving new entry
elif lib.raft_is_leader(self.raft): elif lib.raft_is_leader(self.raft):
new_ety = ffi.new('msg_entry_t*') new_ety = self.network.add_entry(
new_ety.term = 0 lib.RAFT_LOGTYPE_REMOVE_NODE,
new_ety.id = self.network.new_entry_id() change)
new_ety.type = lib.RAFT_LOGTYPE_REMOVE_NODE
new_ety.data.buf = ety.data.buf
new_ety.data.len = ffi.sizeof(ety.data.buf)
assert(lib.raft_entry_is_cfg_change(new_ety)) assert(lib.raft_entry_is_cfg_change(new_ety))
e = self.recv_entry(new_ety) e = self.recv_entry(new_ety)
assert e == 0 assert e == 0
...@@ -1035,7 +1015,7 @@ class RaftServer(object): ...@@ -1035,7 +1015,7 @@ class RaftServer(object):
self.fsm_log.pop() self.fsm_log.pop()
self.network.log_pops += 1 self.network.log_pops += 1
change = ffi.from_handle(ety.data.buf) change = ffi.from_handle(lib.raft_entry_getdata(ety))
if ety.type == lib.RAFT_LOGTYPE_DEMOTE_NODE: if ety.type == lib.RAFT_LOGTYPE_DEMOTE_NODE:
pass pass
...@@ -1057,14 +1037,9 @@ class RaftServer(object): ...@@ -1057,14 +1037,9 @@ class RaftServer(object):
def node_has_sufficient_entries(self, node): def node_has_sufficient_entries(self, node):
assert(not lib.raft_node_is_voting(node)) assert(not lib.raft_node_is_voting(node))
ety = ffi.new('msg_entry_t*') ety = self.network.add_entry(
ety.term = 0 lib.RAFT_LOGTYPE_ADD_NODE,
ety.id = self.network.new_entry_id() ChangeRaftEntry(lib.raft_node_get_id(node)))
ety.type = lib.RAFT_LOGTYPE_ADD_NODE
change = ffi.new_handle(ChangeRaftEntry(lib.raft_node_get_id(node)))
ety.data.buf = change
ety.data.len = ffi.sizeof(ety.data.buf)
self.network.entries.append((ety, change))
assert(lib.raft_entry_is_cfg_change(ety)) assert(lib.raft_entry_is_cfg_change(ety))
# FIXME: leak # FIXME: leak
e = self.recv_entry(ety) e = self.recv_entry(ety)
......
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