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

Add raft_persist_metadata callback (#118)

Combine raft_persist_vote_f and raft_persist_term_f into 
a single callback: raft_persist_metadata_f.

Users should persist vote and term into a single metadata file.

After a restart, application should call raft_restore_metadata() 
to restore vote and term
parent e8791a27
...@@ -454,32 +454,18 @@ typedef void ( ...@@ -454,32 +454,18 @@ typedef void (
const char *buf const char *buf
); );
/** Callback for saving who we voted for to disk. /** Callback for saving current term and vote to the disk.
* For safety reasons this callback MUST flush the change to disk.
* @param[in] raft The Raft server making this callback
* @param[in] user_data User data that is passed from Raft server
* @param[in] vote The node we voted for
* @return 0 on success */
typedef int (
*raft_persist_vote_f
) (
raft_server_t* raft,
void *user_data,
raft_node_id_t vote
);
/** Callback for saving current term (and nil vote) to disk.
* For safety reasons this callback MUST flush the term and vote changes to * For safety reasons this callback MUST flush the term and vote changes to
* disk atomically. * disk atomically.
* @param[in] raft The Raft server making this callback * @param[in] raft The Raft server making this callback
* @param[in] user_data User data that is passed from Raft server * @param[in] user_data User data that is passed from Raft server
* @param[in] term Current term * @param[in] term Current term
* @param[in] vote The node value dictating we haven't voted for anybody * @param[in] vote Vote
* @return 0 on success */ * @return 0 on success */
typedef int ( typedef int (
*raft_persist_term_f *raft_persist_metadata_f
) ( ) (
raft_server_t* raft, raft_server_t *raft,
void *user_data, void *user_data,
raft_term_t term, raft_term_t term,
raft_node_id_t vote raft_node_id_t vote
...@@ -701,14 +687,10 @@ typedef struct ...@@ -701,14 +687,10 @@ typedef struct
* Return RAFT_ERR_SHUTDOWN if you want the server to shutdown. */ * Return RAFT_ERR_SHUTDOWN if you want the server to shutdown. */
raft_logentry_event_f applylog; raft_logentry_event_f applylog;
/** Callback for persisting vote data /** Callback for persisting term and vote data
* For safety reasons this callback MUST flush the change to disk. */
raft_persist_vote_f persist_vote;
/** Callback for persisting term (and nil vote) data
* For safety reasons this callback MUST flush the term and vote changes to * For safety reasons this callback MUST flush the term and vote changes to
* disk atomically. */ * disk atomically. */
raft_persist_term_f persist_term; raft_persist_metadata_f persist_metadata;
/** 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.
...@@ -953,6 +935,15 @@ void raft_destroy(raft_server_t* me); ...@@ -953,6 +935,15 @@ void raft_destroy(raft_server_t* me);
/** De-initialise Raft server. */ /** De-initialise Raft server. */
void raft_clear(raft_server_t* me); void raft_clear(raft_server_t* me);
/** Restore term and vote after reading the metadata file from the disk.
*
* On a restart, the application should set term and vote after reading metadata
* file from the disk. See `raft_persist_metadata_f`.
**/
int raft_restore_metadata(raft_server_t *me,
raft_term_t term,
raft_node_id_t vote);
/** Set callbacks and user data. /** Set callbacks and user data.
* *
* @param[in] funcs Callbacks * @param[in] funcs Callbacks
......
...@@ -140,6 +140,16 @@ raft_server_t* raft_new(void) ...@@ -140,6 +140,16 @@ raft_server_t* raft_new(void)
return raft_new_with_log(&raft_log_internal_impl, NULL); return raft_new_with_log(&raft_log_internal_impl, NULL);
} }
int raft_restore_metadata(raft_server_t *me,
raft_term_t term,
raft_node_id_t vote)
{
me->current_term = term;
me->voted_for = vote;
return 0;
}
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)
{ {
me->cb = *funcs; me->cb = *funcs;
...@@ -1614,12 +1624,17 @@ int raft_vote(raft_server_t* me, raft_node_t* node) ...@@ -1614,12 +1624,17 @@ int raft_vote(raft_server_t* me, raft_node_t* node)
int raft_vote_for_nodeid(raft_server_t* me, const raft_node_id_t nodeid) int raft_vote_for_nodeid(raft_server_t* me, const raft_node_id_t nodeid)
{ {
if (me->cb.persist_vote) { int e;
int e = me->cb.persist_vote(me, me->udata, nodeid);
if (0 != e) if (me->cb.persist_metadata) {
e = me->cb.persist_metadata(me, me->udata, me->current_term, nodeid);
if (e != 0) {
return e; return e;
}
} }
me->voted_for = nodeid; me->voted_for = nodeid;
return 0; return 0;
} }
......
...@@ -49,17 +49,20 @@ raft_node_id_t raft_get_voted_for(raft_server_t *me) ...@@ -49,17 +49,20 @@ raft_node_id_t raft_get_voted_for(raft_server_t *me)
int raft_set_current_term(raft_server_t* me, const raft_term_t term) int raft_set_current_term(raft_server_t* me, const raft_term_t term)
{ {
if (me->current_term < term) if (me->current_term >= term) {
{ return 0;
if (me->cb.persist_term) }
{
int e = me->cb.persist_term(me, me->udata, term, RAFT_NODE_ID_NONE); if (me->cb.persist_metadata) {
if (0 != e) int e = me->cb.persist_metadata(me, me->udata, term, RAFT_NODE_ID_NONE);
return e; if (e != 0) {
return e;
} }
me->current_term = term;
me->voted_for = RAFT_NODE_ID_NONE;
} }
me->current_term = term;
me->voted_for = RAFT_NODE_ID_NONE;
return 0; return 0;
} }
......
...@@ -11,20 +11,11 @@ ...@@ -11,20 +11,11 @@
#include "raft_private.h" #include "raft_private.h"
#include "mock_send_functions.h" #include "mock_send_functions.h"
static int __raft_persist_term( static int __raft_persist_metadata(
raft_server_t* raft, raft_server_t *raft,
void *udata, void *udata,
raft_term_t term, raft_term_t term,
int vote raft_node_id_t vote
)
{
return 0;
}
static int __raft_persist_vote(
raft_server_t* raft,
void *udata,
int vote
) )
{ {
return 0; return 0;
...@@ -53,8 +44,7 @@ void TestRaft_scenario_leader_appears(CuTest * tc) ...@@ -53,8 +44,7 @@ void TestRaft_scenario_leader_appears(CuTest * tc)
&((raft_cbs_t) { &((raft_cbs_t) {
.send_requestvote = sender_requestvote, .send_requestvote = sender_requestvote,
.send_appendentries = sender_appendentries, .send_appendentries = sender_appendentries,
.persist_term = __raft_persist_term, .persist_metadata = __raft_persist_metadata,
.persist_vote = __raft_persist_vote,
.log = NULL .log = NULL
}), sender[j]); }), sender[j]);
} }
......
This diff is collapsed.
...@@ -13,20 +13,11 @@ ...@@ -13,20 +13,11 @@
#include "helpers.h" #include "helpers.h"
static int __raft_persist_term( static int __raft_persist_metadata(
raft_server_t* raft, raft_server_t* raft,
void *udata, void *udata,
raft_term_t term, raft_term_t term,
int vote raft_node_id_t vote
)
{
return 0;
}
static int __raft_persist_vote(
raft_server_t* raft,
void *udata,
int vote
) )
{ {
return 0; return 0;
...@@ -186,11 +177,6 @@ static int test_load_snapshot(raft_server_t* raft, ...@@ -186,11 +177,6 @@ static int test_load_snapshot(raft_server_t* raft,
return 0; return 0;
} }
/* static raft_cbs_t generic_funcs = { */
/* .persist_term = __raft_persist_term, */
/* .persist_vote = __raft_persist_vote, */
/* }; */
static int max_election_timeout(int election_timeout) static int max_election_timeout(int election_timeout)
{ {
return 2 * election_timeout; return 2 * election_timeout;
...@@ -315,7 +301,7 @@ void TestRaft_leader_snapshot_begin_fails_if_less_than_2_logs_to_compact(CuTest ...@@ -315,7 +301,7 @@ void TestRaft_leader_snapshot_begin_fails_if_less_than_2_logs_to_compact(CuTest
void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc) void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc)
{ {
raft_cbs_t funcs = { raft_cbs_t funcs = {
.persist_term = __raft_persist_term, .persist_metadata = __raft_persist_metadata,
.send_appendentries = __raft_send_appendentries, .send_appendentries = __raft_send_appendentries,
}; };
...@@ -380,7 +366,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc) ...@@ -380,7 +366,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc)
void TestRaft_leader_snapshot_end_succeeds_if_log_compacted2(CuTest * tc) void TestRaft_leader_snapshot_end_succeeds_if_log_compacted2(CuTest * tc)
{ {
raft_cbs_t funcs = { raft_cbs_t funcs = {
.persist_term = __raft_persist_term, .persist_metadata = __raft_persist_metadata,
.send_appendentries = __raft_send_appendentries, .send_appendentries = __raft_send_appendentries,
}; };
...@@ -712,7 +698,7 @@ void TestRaft_follower_recv_appendentries_is_successful_when_previous_log_idx_eq ...@@ -712,7 +698,7 @@ void TestRaft_follower_recv_appendentries_is_successful_when_previous_log_idx_eq
CuTest * tc) CuTest * tc)
{ {
raft_cbs_t funcs = { raft_cbs_t funcs = {
.persist_term = __raft_persist_term, .persist_metadata = __raft_persist_metadata,
}; };
void *r = raft_new(); void *r = raft_new();
......
...@@ -233,12 +233,8 @@ def raft_applylog(raft, udata, ety, idx): ...@@ -233,12 +233,8 @@ def raft_applylog(raft, udata, ety, idx):
return lib.RAFT_ERR_SHUTDOWN return lib.RAFT_ERR_SHUTDOWN
def raft_persist_vote(raft, udata, voted_for): def raft_persist_metadata(raft, udata, term, vote):
return ffi.from_handle(udata).persist_vote(voted_for) return ffi.from_handle(udata).persist_metadata(term, vote)
def raft_persist_term(raft, udata, term, vote):
return ffi.from_handle(udata).persist_term(term, vote)
def raft_logentry_offer(raft, udata, ety, ety_idx): def raft_logentry_offer(raft, udata, ety, ety_idx):
...@@ -834,8 +830,7 @@ class RaftServer(object): ...@@ -834,8 +830,7 @@ class RaftServer(object):
cbs.get_snapshot_chunk = self.raft_get_snapshot_chunk cbs.get_snapshot_chunk = self.raft_get_snapshot_chunk
cbs.store_snapshot_chunk = self.raft_store_snapshot_chunk cbs.store_snapshot_chunk = self.raft_store_snapshot_chunk
cbs.applylog = self.raft_applylog cbs.applylog = self.raft_applylog
cbs.persist_vote = self.raft_persist_vote cbs.persist_metadata = self.raft_persist_metadata
cbs.persist_term = self.raft_persist_term
cbs.get_node_id = self.raft_get_node_id cbs.get_node_id = self.raft_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
...@@ -954,8 +949,7 @@ class RaftServer(object): ...@@ -954,8 +949,7 @@ class RaftServer(object):
self.raft_get_snapshot_chunk = ffi.callback("int(raft_server_t*, void*, raft_node_t*, raft_size_t offset, raft_snapshot_chunk_t*)", raft_get_snapshot_chunk) self.raft_get_snapshot_chunk = ffi.callback("int(raft_server_t*, void*, raft_node_t*, raft_size_t offset, raft_snapshot_chunk_t*)", raft_get_snapshot_chunk)
self.raft_store_snapshot_chunk = ffi.callback("int(raft_server_t*, void*, raft_index_t index, raft_size_t offset, raft_snapshot_chunk_t*)", raft_store_snapshot_chunk) self.raft_store_snapshot_chunk = ffi.callback("int(raft_server_t*, void*, raft_index_t index, raft_size_t offset, raft_snapshot_chunk_t*)", raft_store_snapshot_chunk)
self.raft_applylog = ffi.callback("int(raft_server_t*, void*, raft_entry_t*, raft_index_t)", raft_applylog) self.raft_applylog = ffi.callback("int(raft_server_t*, void*, raft_entry_t*, raft_index_t)", raft_applylog)
self.raft_persist_vote = ffi.callback("int(raft_server_t*, void*, raft_node_id_t)", raft_persist_vote) self.raft_persist_metadata = ffi.callback("int(raft_server_t*, void*, raft_term_t, raft_node_id_t)", raft_persist_metadata)
self.raft_persist_term = ffi.callback("int(raft_server_t*, void*, raft_term_t, raft_node_id_t)", raft_persist_term)
self.raft_logentry_offer = ffi.callback("int(raft_server_t*, void*, raft_entry_t*, raft_index_t)", raft_logentry_offer) self.raft_logentry_offer = ffi.callback("int(raft_server_t*, void*, raft_entry_t*, raft_index_t)", raft_logentry_offer)
self.raft_logentry_poll = ffi.callback("int(raft_server_t*, void*, raft_entry_t*, raft_index_t)", raft_logentry_poll) self.raft_logentry_poll = ffi.callback("int(raft_server_t*, void*, raft_entry_t*, raft_index_t)", raft_logentry_poll)
self.raft_logentry_pop = ffi.callback("int(raft_server_t*, void*, raft_entry_t*, raft_index_t)", raft_logentry_pop) self.raft_logentry_pop = ffi.callback("int(raft_server_t*, void*, raft_entry_t*, raft_index_t)", raft_logentry_pop)
...@@ -1168,11 +1162,7 @@ class RaftServer(object): ...@@ -1168,11 +1162,7 @@ class RaftServer(object):
# self, snapshot.last_term, snapshot.last_idx)) # self, snapshot.last_term, snapshot.last_idx))
return 0 return 0
def persist_vote(self, voted_for): def persist_metadata(self, term, vote):
# TODO: add disk simulation
return 0
def persist_term(self, term, vote):
# TODO: add disk simulation # TODO: add disk simulation
return 0 return 0
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment