Commit a0456901 authored by Willem Thiart's avatar Willem Thiart
Browse files

Added log entry callbacks

parent 50868621
...@@ -179,7 +179,7 @@ typedef int ( ...@@ -179,7 +179,7 @@ typedef int (
* @param voted_for The node we voted for * @param voted_for The node we voted for
* @return 0 on success */ * @return 0 on success */
typedef int ( typedef int (
*func_persist_vote_f *func_persist_int_f
) ( ) (
raft_server_t* raft, raft_server_t* raft,
void *udata, void *udata,
...@@ -187,42 +187,38 @@ typedef int ( ...@@ -187,42 +187,38 @@ typedef int (
); );
/** /**
* Save the term we are on to disk
* @param raft The Raft server making this callback * @param raft The Raft server making this callback
* @param udata User data that is passed from Raft server * @param udata User data that is passed from Raft server
* @param term Current term * @param entry The entry that the event is happening to
* @return 0 on success */ * @return 0 on success */
typedef int ( typedef int (
*func_persist_current_term_f *func_logentry_event_f
) ( ) (
raft_server_t* raft, raft_server_t* raft,
void *udata, void *udata,
const int current_term raft_entry_t *entry
);
/**
* Save log to disk
* @param raft The Raft server making this callback
* @param udata User data that is passed from Raft server
* @param term Current term
* @return 0 on success */
typedef int (
*func_persist_entry_f
) (
raft_server_t* raft,
void *udata,
const raft_entry_t entry
); );
typedef struct typedef struct
{ {
/* message sending */
func_send_requestvote_f send_requestvote; func_send_requestvote_f send_requestvote;
func_send_appendentries_f send_appendentries; func_send_appendentries_f send_appendentries;
func_log_f log;
/* finite state machine application */
func_applylog_f applylog; func_applylog_f applylog;
func_persist_vote_f persist_vote;
func_persist_current_term_f persist_current_term; /* persistence */
func_persist_entry_f persist_entry; func_persist_int_f persist_vote;
func_persist_int_f persist_term;
/* log entry persistence */
func_logentry_event_f log_offer;
func_logentry_event_f log_poll;
func_logentry_event_f log_pop;
/* debugging - optional */
func_log_f log;
} raft_cbs_t; } raft_cbs_t;
/** /**
...@@ -431,4 +427,18 @@ int raft_get_voted_for(raft_server_t* me); ...@@ -431,4 +427,18 @@ int raft_get_voted_for(raft_server_t* me);
* -1 if the leader is unknown */ * -1 if the leader is unknown */
int raft_get_current_leader(raft_server_t* me); int raft_get_current_leader(raft_server_t* me);
/**
* @return callback user data */
void* raft_get_udata(raft_server_t* me_);
/**
* Vote for a server
* @param node The server to vote for */
void raft_vote(raft_server_t* me_, const int node);
/**
* Set the current term
* @param term The new current term */
void raft_set_current_term(raft_server_t* me_, const int term);
#endif /* RAFT_H_ */ #endif /* RAFT_H_ */
...@@ -5,6 +5,8 @@ typedef void* log_t; ...@@ -5,6 +5,8 @@ typedef void* log_t;
log_t* log_new(); log_t* log_new();
void log_set_callbacks(log_t* me_, raft_cbs_t* funcs, void* raft);
void log_free(log_t* me_); void log_free(log_t* me_);
/** /**
......
...@@ -31,7 +31,7 @@ typedef struct { ...@@ -31,7 +31,7 @@ typedef struct {
int voted_for; int voted_for;
/* the log which is replicated */ /* the log which is replicated */
log_t* log; void* log;
/* Volatile state: */ /* Volatile state: */
......
...@@ -15,6 +15,7 @@ ...@@ -15,6 +15,7 @@
#include <assert.h> #include <assert.h>
#include "raft.h" #include "raft.h"
#include "raft_private.h"
#include "raft_log.h" #include "raft_log.h"
#define INITIAL_CAPACITY 10 #define INITIAL_CAPACITY 10
...@@ -35,6 +36,10 @@ typedef struct ...@@ -35,6 +36,10 @@ typedef struct
int base_log_idx; int base_log_idx;
raft_entry_t* entries; raft_entry_t* entries;
/* callbacks */
raft_cbs_t *cb;
void* raft;
} log_private_t; } log_private_t;
static void __ensurecapacity(log_private_t * me) static void __ensurecapacity(log_private_t * me)
...@@ -73,6 +78,14 @@ log_t* log_new() ...@@ -73,6 +78,14 @@ log_t* log_new()
return (log_t*)me; return (log_t*)me;
} }
void log_set_callbacks(log_t* me_, raft_cbs_t* funcs, void* raft)
{
log_private_t* me = (log_private_t*)me_;
me->raft = raft;
me->cb = funcs;
}
int log_append_entry(log_t* me_, raft_entry_t* c) int log_append_entry(log_t* me_, raft_entry_t* c)
{ {
log_private_t* me = (log_private_t*)me_; log_private_t* me = (log_private_t*)me_;
...@@ -86,6 +99,8 @@ int log_append_entry(log_t* me_, raft_entry_t* c) ...@@ -86,6 +99,8 @@ int log_append_entry(log_t* me_, raft_entry_t* c)
me->entries[me->back].num_nodes = 0; me->entries[me->back].num_nodes = 0;
me->count++; me->count++;
me->back++; me->back++;
if (me->cb && me->cb->log_offer)
me->cb->log_offer(me->raft, raft_get_udata(me->raft), c);
return 0; return 0;
} }
...@@ -122,6 +137,8 @@ void log_delete(log_t* me_, int idx) ...@@ -122,6 +137,8 @@ void log_delete(log_t* me_, int idx)
for (end = log_count(me_); idx < end; idx++) for (end = log_count(me_); idx < end; idx++)
{ {
if (me->cb && me->cb->log_pop)
me->cb->log_pop(me->raft, raft_get_udata(me->raft), &me->entries[me->back]);
me->back--; me->back--;
me->count--; me->count--;
} }
...@@ -135,6 +152,8 @@ void *log_poll(log_t * me_) ...@@ -135,6 +152,8 @@ void *log_poll(log_t * me_)
return NULL; return NULL;
const void *elem = &me->entries[me->front]; const void *elem = &me->entries[me->front];
if (me->cb && me->cb->log_poll)
me->cb->log_poll(me->raft, raft_get_udata(me->raft), &me->entries[me->front]);
me->front++; me->front++;
me->count--; me->count--;
me->base_log_idx++; me->base_log_idx++;
......
...@@ -51,13 +51,13 @@ raft_server_t* raft_new() ...@@ -51,13 +51,13 @@ raft_server_t* raft_new()
return (raft_server_t*)me; return (raft_server_t*)me;
} }
void raft_set_callbacks(raft_server_t* me_, void raft_set_callbacks(raft_server_t* me_, raft_cbs_t* funcs, void* udata)
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_)
...@@ -597,9 +597,11 @@ int raft_get_nvotes_for_me(raft_server_t* me_) ...@@ -597,9 +597,11 @@ int raft_get_nvotes_for_me(raft_server_t* me_)
return votes; return votes;
} }
void raft_vote(raft_server_t* me_, int node) void raft_vote(raft_server_t* me_, const int node)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
me->voted_for = node; me->voted_for = node;
if (me->cb.persist_vote)
me->cb.persist_vote(me_, me->udata, node);
} }
...@@ -68,10 +68,12 @@ int raft_get_voted_for(raft_server_t* me_) ...@@ -68,10 +68,12 @@ int raft_get_voted_for(raft_server_t* me_)
return ((raft_server_private_t*)me_)->voted_for; return ((raft_server_private_t*)me_)->voted_for;
} }
void raft_set_current_term(raft_server_t* me_, int term) void raft_set_current_term(raft_server_t* me_, const int term)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
me->current_term = term; me->current_term = term;
if (me->cb.persist_term)
me->cb.persist_term(me_, me->udata, term);
} }
int raft_get_current_term(raft_server_t* me_) int raft_get_current_term(raft_server_t* me_)
...@@ -146,6 +148,11 @@ int raft_get_current_leader(raft_server_t* me_) ...@@ -146,6 +148,11 @@ int raft_get_current_leader(raft_server_t* me_)
return me->current_leader; return me->current_leader;
} }
void* raft_get_udata(raft_server_t* me_)
{
return ((raft_server_private_t*)me_)->udata;
}
int raft_is_follower(raft_server_t* me_) int raft_is_follower(raft_server_t* me_)
{ {
return raft_get_state(me_) == RAFT_STATE_FOLLOWER; return raft_get_state(me_) == RAFT_STATE_FOLLOWER;
......
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