Commit 1f02f374 authored by Willem Thiart's avatar Willem Thiart
Browse files

Flatbuffers transport

parent 296c7e0b
...@@ -9,6 +9,8 @@ SHELL = /bin/bash ...@@ -9,6 +9,8 @@ SHELL = /bin/bash
CFLAGS += -Iinclude -Werror -Werror=return-type -Werror=uninitialized -Wcast-align \ CFLAGS += -Iinclude -Werror -Werror=return-type -Werror=uninitialized -Wcast-align \
-Wno-pointer-sign -fno-omit-frame-pointer -fno-common -fsigned-char \ -Wno-pointer-sign -fno-omit-frame-pointer -fno-common -fsigned-char \
-Wunused-variable \ -Wunused-variable \
-D FLATCC_PORTABLE \
-I/Users/willem/contrib/flatcc/include \
$(GCOV_CCFLAGS) -I$(LLQUEUE_DIR) -Iinclude -g -O2 -fPIC $(GCOV_CCFLAGS) -I$(LLQUEUE_DIR) -Iinclude -g -O2 -fPIC
UNAME := $(shell uname) UNAME := $(shell uname)
...@@ -26,10 +28,20 @@ SHAREDFLAGS = -shared ...@@ -26,10 +28,20 @@ SHAREDFLAGS = -shared
SHAREDEXT = so SHAREDEXT = so
endif endif
OBJECTS = src/raft_server.o src/raft_server_properties.o src/raft_node.o src/raft_log.o OBJECTS = src/raft_server.o src/raft_server_properties.o src/raft_node.o src/raft_log.o
TRANSPORT_DIR = include/transport
TRANSPORT_OBJECTS = $(TRANSPORT_DIR)/core_builder.h $(TRANSPORT_DIR)/core_reader.h $(TRANSPORT_DIR)/core_verifier.h $(TRANSPORT_DIR)/flatbuffers_common_builder.h $(TRANSPORT_DIR)/flatbuffers_common_reader.h
all: static shared all: static shared
.PHONY: transport
transport:
mkdir -p $(TRANSPORT_DIR)
/Users/willem/contrib/flatcc/bin/flatcc -a -o $(TRANSPORT_DIR) src/transport/core.fbs
clinkedlistqueue: clinkedlistqueue:
mkdir -p $(LLQUEUE_DIR)/.git mkdir -p $(LLQUEUE_DIR)/.git
git --git-dir=$(LLQUEUE_DIR)/.git init git --git-dir=$(LLQUEUE_DIR)/.git init
...@@ -45,11 +57,11 @@ $(TEST_DIR)/main_test.c: ...@@ -45,11 +57,11 @@ $(TEST_DIR)/main_test.c:
cd $(TEST_DIR) && sh make-tests.sh "test_*.c" > main_test.c && cd .. cd $(TEST_DIR) && sh make-tests.sh "test_*.c" > main_test.c && cd ..
.PHONY: shared .PHONY: shared
shared: $(OBJECTS) shared: transport $(OBJECTS)
$(CC) $(OBJECTS) $(LDFLAGS) $(CFLAGS) -fPIC $(SHAREDFLAGS) -o libraft.$(SHAREDEXT) $(CC) $(OBJECTS) $(LDFLAGS) $(CFLAGS) -fPIC $(SHAREDFLAGS) -o libraft.$(SHAREDEXT)
.PHONY: static .PHONY: static
static: $(OBJECTS) static: transport $(OBJECTS)
ar -r libraft.a $(OBJECTS) ar -r libraft.a $(OBJECTS)
.PHONY: tests .PHONY: tests
...@@ -89,6 +101,7 @@ do_infer: ...@@ -89,6 +101,7 @@ do_infer:
clean: clean:
@rm -f $(TEST_DIR)/main_test.c *.o $(GCOV_OUTPUT); \ @rm -f $(TEST_DIR)/main_test.c *.o $(GCOV_OUTPUT); \
rm $(TRANSPORT_DIR)/*.h; \
if [ -f "libraft.$(SHAREDEXT)" ]; then rm libraft.$(SHAREDEXT); fi;\ if [ -f "libraft.$(SHAREDEXT)" ]; then rm libraft.$(SHAREDEXT); fi;\
if [ -f libraft.a ]; then rm libraft.a; fi;\ if [ -f libraft.a ]; then rm libraft.a; fi;\
if [ -f tests_main ]; then rm tests_main; fi; if [ -f tests_main ]; then rm tests_main; fi;
...@@ -126,7 +126,7 @@ We call ``raft_recv_entry`` when we want to append the entry to the log. ...@@ -126,7 +126,7 @@ We call ``raft_recv_entry`` when we want to append the entry to the log.
.. code-block:: c .. code-block:: c
msg_entry_response_t response; LogEntryResponseMsg_table_t response;
e = raft_recv_entry(raft, &entry, &response); e = raft_recv_entry(raft, &entry, &response);
You should populate the ``entry`` struct with the log entry the client has sent. After the call completes the ``response`` parameter is populated and can be used by the ``raft_msg_entry_response_committed`` function to check if the log entry has been committed or not. You should populate the ``entry`` struct with the log entry the client has sent. After the call completes the ``response`` parameter is populated and can be used by the ``raft_msg_entry_response_committed`` function to check if the log entry has been committed or not.
...@@ -141,7 +141,7 @@ The ``raft_recv_entry`` function does not block! This means you will need to imp ...@@ -141,7 +141,7 @@ The ``raft_recv_entry`` function does not block! This means you will need to imp
.. code-block:: c .. code-block:: c
msg_entry_response_t response; LogEntryResponseMsg_table_t response;
e = raft_recv_entry(sv->raft, &entry, &response); e = raft_recv_entry(sv->raft, &entry, &response);
if (0 != e) if (0 != e)
...@@ -242,7 +242,7 @@ The following callbacks MUST be implemented: ``send_requestvote``, ``send_append ...@@ -242,7 +242,7 @@ The following callbacks MUST be implemented: ``send_requestvote``, ``send_append
**send_requestvote()** **send_requestvote()**
For this callback we have to serialize a ``msg_requestvote_t`` struct, and then send it to the peer identified by ``node``. For this callback we have to serialize a ``RequestVoteMsg_t`` struct, and then send it to the peer identified by ``node``.
*Example from ticketd showing how the callback is implemented:* *Example from ticketd showing how the callback is implemented:*
...@@ -252,7 +252,7 @@ For this callback we have to serialize a ``msg_requestvote_t`` struct, and then ...@@ -252,7 +252,7 @@ For this callback we have to serialize a ``msg_requestvote_t`` struct, and then
raft_server_t* raft, raft_server_t* raft,
void *udata, void *udata,
raft_node_t* node, raft_node_t* node,
msg_requestvote_t* m RequestVoteMsg_t* m
) )
{ {
peer_connection_t* conn = raft_node_get_udata(node); peer_connection_t* conn = raft_node_get_udata(node);
...@@ -272,7 +272,7 @@ For this callback we have to serialize a ``msg_requestvote_t`` struct, and then ...@@ -272,7 +272,7 @@ For this callback we have to serialize a ``msg_requestvote_t`` struct, and then
**send_appendentries()** **send_appendentries()**
For this callback we have to serialize a ``msg_appendentries_t`` struct, and then send it to the peer identified by ``node``. This struct is more complicated to serialize because the ``m->entries`` array might be populated. For this callback we have to serialize a ``AppendEntriesMsg_t`` struct, and then send it to the peer identified by ``node``. This struct is more complicated to serialize because the ``m->entries`` array might be populated.
*Example from ticketd showing how the callback is implemented:* *Example from ticketd showing how the callback is implemented:*
...@@ -282,7 +282,7 @@ For this callback we have to serialize a ``msg_appendentries_t`` struct, and the ...@@ -282,7 +282,7 @@ For this callback we have to serialize a ``msg_appendentries_t`` struct, and the
raft_server_t* raft, raft_server_t* raft,
void *user_data, void *user_data,
raft_node_t* node, raft_node_t* node,
msg_appendentries_t* m AppendEntriesMsg_t* m
) )
{ {
uv_buf_t bufs[3]; uv_buf_t bufs[3];
...@@ -376,21 +376,21 @@ The table below shows the structs that you need to deserialize-to or deserialize ...@@ -376,21 +376,21 @@ The table below shows the structs that you need to deserialize-to or deserialize
+-------------------------+------------------------------+----------------------------------+ +-------------------------+------------------------------+----------------------------------+
| Message Type | Struct | Function | | Message Type | Struct | Function |
+-------------------------+------------------------------+----------------------------------+ +-------------------------+------------------------------+----------------------------------+
| Append Entries | msg_appendentries_t | raft_recv_appendentries | | Append Entries | AppendEntriesMsg_t | raft_recv_appendentries |
+-------------------------+------------------------------+----------------------------------+ +-------------------------+------------------------------+----------------------------------+
| Append Entries response | msg_appendentries_response_t | raft_recv_appendentries_response | | Append Entries response | AppendEntriesResponseMsg_t | raft_recv_appendentries_response |
+-------------------------+------------------------------+----------------------------------+ +-------------------------+------------------------------+----------------------------------+
| Request Vote | msg_requestvote_t | raft_recv_requestvote | | Request Vote | RequestVoteMsg_t | raft_recv_requestvote |
+-------------------------+------------------------------+----------------------------------+ +-------------------------+------------------------------+----------------------------------+
| Request Vote response | msg_requestvote_response_t | raft_recv_requestvote_response | | Request Vote response | RequestVoteResponseMsg_t | raft_recv_requestvote_response |
+-------------------------+------------------------------+----------------------------------+ +-------------------------+------------------------------+----------------------------------+
*Example of how we receive an Append Entries message, and reply to it:* *Example of how we receive an Append Entries message, and reply to it:*
.. code-block:: c .. code-block:: c
msg_appendentries_t ae; AppendEntriesMsg_t ae;
msg_appendentries_response_t response; AppendEntriesResponseMsg_t response;
char buf_in[1024], buf_out[1024]; char buf_in[1024], buf_out[1024];
size_t len_in, len_out; size_t len_in, len_out;
......
...@@ -23,6 +23,12 @@ ...@@ -23,6 +23,12 @@
#define RAFT_REQUESTVOTE_ERR_NOT_GRANTED 0 #define RAFT_REQUESTVOTE_ERR_NOT_GRANTED 0
#define RAFT_REQUESTVOTE_ERR_UNKNOWN_NODE -1 #define RAFT_REQUESTVOTE_ERR_UNKNOWN_NODE -1
// #include "transport/core_builder.h"
#include "transport/core_reader.h"
// #include "transport/core_verifier.h"
// #include "transport/flatbuffers_common_builder.h"
// #include "transport/flatbuffers_common_reader.h"
typedef enum { typedef enum {
RAFT_STATE_NONE, RAFT_STATE_NONE,
RAFT_STATE_FOLLOWER, RAFT_STATE_FOLLOWER,
...@@ -92,105 +98,6 @@ typedef struct ...@@ -92,105 +98,6 @@ typedef struct
raft_entry_data_t data; raft_entry_data_t data;
} raft_entry_t; } raft_entry_t;
/** Message sent from client to server.
* The client sends this message to a server with the intention of having it
* applied to the FSM. */
typedef raft_entry_t msg_entry_t;
/** Entry message response.
* Indicates to client if entry was committed or not. */
typedef struct
{
/** the entry's unique ID */
unsigned int id;
/** the entry's term */
int term;
/** the entry's index */
int idx;
} msg_entry_response_t;
/** Vote request message.
* Sent to nodes when a server wants to become leader.
* This message could force a leader/candidate to become a follower. */
typedef struct
{
/** currentTerm, to force other leader/candidate to step down */
int term;
/** candidate requesting vote */
int candidate_id;
/** index of candidate's last log entry */
int last_log_idx;
/** term of candidate's last log entry */
int last_log_term;
} msg_requestvote_t;
/** Vote request response message.
* Indicates if node has accepted the server's vote request. */
typedef struct
{
/** currentTerm, for candidate to update itself */
int term;
/** true means candidate received vote */
int vote_granted;
} msg_requestvote_response_t;
/** Appendentries message.
* This message is used to tell nodes if it's safe to apply entries to the FSM.
* Can be sent without any entries as a keep alive message.
* This message could force a leader/candidate to become a follower. */
typedef struct
{
/** currentTerm, to force other leader/candidate to step down */
int term;
/** the index of the log just before the newest entry for the node who
* receives this message */
int prev_log_idx;
/** the term of the log just before the newest entry for the node who
* receives this message */
int prev_log_term;
/** the index of the entry that has been appended to the majority of the
* cluster. Entries up to this index will be applied to the FSM */
int leader_commit;
/** number of entries within this message */
int n_entries;
/** array of entries within this message */
msg_entry_t* entries;
} msg_appendentries_t;
/** Appendentries response message.
* Can be sent without any entries as a keep alive message.
* This message could force a leader/candidate to become a follower. */
typedef struct
{
/** currentTerm, to force other leader/candidate to step down */
int term;
/** true if follower contained entry matching prevLogidx and prevLogTerm */
int success;
/* Non-Raft fields follow: */
/* Having the following fields allows us to do less book keeping in
* regards to full fledged RPC */
/** If success, this is the highest log IDX we've received and appended to
* our log; otherwise, this is the our currentIndex */
int current_idx;
/** The first idx that we received within the appendentries message */
int first_idx;
} msg_appendentries_response_t;
typedef void* raft_server_t; typedef void* raft_server_t;
typedef void* raft_node_t; typedef void* raft_node_t;
...@@ -206,7 +113,7 @@ typedef int ( ...@@ -206,7 +113,7 @@ typedef int (
raft_server_t* raft, raft_server_t* raft,
void *user_data, void *user_data,
raft_node_t* node, raft_node_t* node,
msg_requestvote_t* msg RequestVoteMsg_t* msg
); );
/** Callback for sending append entries messages. /** Callback for sending append entries messages.
...@@ -221,7 +128,7 @@ typedef int ( ...@@ -221,7 +128,7 @@ typedef int (
raft_server_t* raft, raft_server_t* raft,
void *user_data, void *user_data,
raft_node_t* node, raft_node_t* node,
msg_appendentries_t* msg AppendEntriesMsg_t* msg
); );
/** /**
...@@ -479,7 +386,7 @@ int raft_periodic(raft_server_t* me, int msec_elapsed); ...@@ -479,7 +386,7 @@ int raft_periodic(raft_server_t* me, int msec_elapsed);
* *
* The log_offer callback will be called. * The log_offer callback will be called.
* *
* @note The memory pointer (ie. raft_entry_data_t) for each msg_entry_t is * @note The memory pointer (ie. raft_entry_data_t) for each LogEntryMsg_table_t is
* copied directly. If the memory is temporary you MUST either make the * copied directly. If the memory is temporary you MUST either make the
* memory permanent (ie. via malloc) OR re-assign the memory within the * memory permanent (ie. via malloc) OR re-assign the memory within the
* log_offer callback. * log_offer callback.
...@@ -493,8 +400,8 @@ int raft_periodic(raft_server_t* me, int msec_elapsed); ...@@ -493,8 +400,8 @@ int raft_periodic(raft_server_t* me, int msec_elapsed);
* */ * */
int raft_recv_appendentries(raft_server_t* me, int raft_recv_appendentries(raft_server_t* me,
raft_node_t* node, raft_node_t* node,
msg_appendentries_t* ae, AppendEntriesMsg_t* ae,
msg_appendentries_response_t *r); AppendEntriesResponseMsg_t *r);
/** Receive a response from an appendentries message we sent. /** Receive a response from an appendentries message we sent.
* @param[in] node The node who sent us this message * @param[in] node The node who sent us this message
...@@ -505,7 +412,7 @@ int raft_recv_appendentries(raft_server_t* me, ...@@ -505,7 +412,7 @@ int raft_recv_appendentries(raft_server_t* me,
* RAFT_ERR_NOT_LEADER server is not the leader */ * RAFT_ERR_NOT_LEADER server is not the leader */
int raft_recv_appendentries_response(raft_server_t* me, int raft_recv_appendentries_response(raft_server_t* me,
raft_node_t* node, raft_node_t* node,
msg_appendentries_response_t* r); AppendEntriesResponseMsg_t* r);
/** Receive a requestvote message. /** Receive a requestvote message.
* @param[in] node The node who sent us this message * @param[in] node The node who sent us this message
...@@ -514,8 +421,8 @@ int raft_recv_appendentries_response(raft_server_t* me, ...@@ -514,8 +421,8 @@ int raft_recv_appendentries_response(raft_server_t* me,
* @return 0 on success */ * @return 0 on success */
int raft_recv_requestvote(raft_server_t* me, int raft_recv_requestvote(raft_server_t* me,
raft_node_t* node, raft_node_t* node,
msg_requestvote_t* vr, RequestVoteMsg_t* vr,
msg_requestvote_response_t *r); RequestVoteResponseMsg_t *r);
/** Receive a response from a requestvote message we sent. /** Receive a response from a requestvote message we sent.
* @param[in] node The node this response was sent by * @param[in] node The node this response was sent by
...@@ -525,7 +432,7 @@ int raft_recv_requestvote(raft_server_t* me, ...@@ -525,7 +432,7 @@ int raft_recv_requestvote(raft_server_t* me,
* RAFT_ERR_SHUTDOWN server MUST shutdown; */ * RAFT_ERR_SHUTDOWN server MUST shutdown; */
int raft_recv_requestvote_response(raft_server_t* me, int raft_recv_requestvote_response(raft_server_t* me,
raft_node_t* node, raft_node_t* node,
msg_requestvote_response_t* r); RequestVoteResponseMsg_t* r);
/** Receive an entry message from the client. /** Receive an entry message from the client.
* *
...@@ -537,7 +444,7 @@ int raft_recv_requestvote_response(raft_server_t* me, ...@@ -537,7 +444,7 @@ int raft_recv_requestvote_response(raft_server_t* me,
* *
* The log_offer callback will be called. * The log_offer callback will be called.
* *
* @note The memory pointer (ie. raft_entry_data_t) in msg_entry_t is * @note The memory pointer (ie. raft_entry_data_t) in LogEntryMsg_table_t is
* copied directly. If the memory is temporary you MUST either make the * copied directly. If the memory is temporary you MUST either make the
* memory permanent (ie. via malloc) OR re-assign the memory within the * memory permanent (ie. via malloc) OR re-assign the memory within the
* log_offer callback. * log_offer callback.
...@@ -558,8 +465,8 @@ int raft_recv_requestvote_response(raft_server_t* me, ...@@ -558,8 +465,8 @@ int raft_recv_requestvote_response(raft_server_t* me,
* RAFT_ERR_NOMEM memory allocation failure * RAFT_ERR_NOMEM memory allocation failure
*/ */
int raft_recv_entry(raft_server_t* me, int raft_recv_entry(raft_server_t* me,
msg_entry_t* ety, LogEntryMsg_table_t* ety,
msg_entry_response_t *r); LogEntryResponseMsg_table_t *r);
/** /**
* @return server's node ID; -1 if it doesn't know what it is */ * @return server's node ID; -1 if it doesn't know what it is */
...@@ -713,7 +620,7 @@ int raft_append_entry(raft_server_t* me, raft_entry_t* ety); ...@@ -713,7 +620,7 @@ int raft_append_entry(raft_server_t* me, raft_entry_t* ety);
/** Confirm if a msg_entry_response has been committed. /** Confirm if a msg_entry_response has been committed.
* @param[in] r The response we want to check */ * @param[in] r The response we want to check */
int raft_msg_entry_response_committed(raft_server_t* me_, int raft_msg_entry_response_committed(raft_server_t* me_,
const msg_entry_response_t* r); const LogEntryResponseMsg_table_t* r);
/** Get node's ID. /** Get node's ID.
* @return ID of node */ * @return ID of node */
......
...@@ -272,7 +272,7 @@ int raft_voting_change_is_in_progress(raft_server_t* me_) ...@@ -272,7 +272,7 @@ int raft_voting_change_is_in_progress(raft_server_t* me_)
int raft_recv_appendentries_response(raft_server_t* me_, int raft_recv_appendentries_response(raft_server_t* me_,
raft_node_t* node, raft_node_t* node,
msg_appendentries_response_t* r) AppendEntriesResponseMsg_t* r)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
...@@ -383,8 +383,8 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -383,8 +383,8 @@ int raft_recv_appendentries_response(raft_server_t* me_,
int raft_recv_appendentries( int raft_recv_appendentries(
raft_server_t* me_, raft_server_t* me_,
raft_node_t* node, raft_node_t* node,
msg_appendentries_t* ae, AppendEntriesMsg_t* ae,
msg_appendentries_response_t *r AppendEntriesResponseMsg_t *r
) )
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
...@@ -520,7 +520,7 @@ int raft_already_voted(raft_server_t* me_) ...@@ -520,7 +520,7 @@ int raft_already_voted(raft_server_t* me_)
return ((raft_server_private_t*)me_)->voted_for != -1; return ((raft_server_private_t*)me_)->voted_for != -1;
} }
static int __should_grant_vote(raft_server_private_t* me, msg_requestvote_t* vr) static int __should_grant_vote(raft_server_private_t* me, RequestVoteMsg_t* vr)
{ {
/* TODO: 4.2.3 Raft Dissertation: /* TODO: 4.2.3 Raft Dissertation:
* if a server receives a RequestVote request within the minimum election * if a server receives a RequestVote request within the minimum election
...@@ -557,8 +557,8 @@ static int __should_grant_vote(raft_server_private_t* me, msg_requestvote_t* vr) ...@@ -557,8 +557,8 @@ static int __should_grant_vote(raft_server_private_t* me, msg_requestvote_t* vr)
int raft_recv_requestvote(raft_server_t* me_, int raft_recv_requestvote(raft_server_t* me_,
raft_node_t* node, raft_node_t* node,
msg_requestvote_t* vr, RequestVoteMsg_t* vr,
msg_requestvote_response_t *r) RequestVoteResponseMsg_t *r)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
int e = 0; int e = 0;
...@@ -630,7 +630,7 @@ int raft_votes_is_majority(const int num_nodes, const int nvotes) ...@@ -630,7 +630,7 @@ int raft_votes_is_majority(const int num_nodes, const int nvotes)
int raft_recv_requestvote_response(raft_server_t* me_, int raft_recv_requestvote_response(raft_server_t* me_,
raft_node_t* node, raft_node_t* node,
msg_requestvote_response_t* r) RequestVoteResponseMsg_t* r)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
...@@ -692,8 +692,8 @@ int raft_recv_requestvote_response(raft_server_t* me_, ...@@ -692,8 +692,8 @@ int raft_recv_requestvote_response(raft_server_t* me_,
} }
int raft_recv_entry(raft_server_t* me_, int raft_recv_entry(raft_server_t* me_,
msg_entry_t* ety, LogEntryMsg_table_t* ety,
msg_entry_response_t *r) LogEntryResponseMsg_table_t *r)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
int i; int i;
...@@ -755,7 +755,7 @@ int raft_recv_entry(raft_server_t* me_, ...@@ -755,7 +755,7 @@ int raft_recv_entry(raft_server_t* me_,
int raft_send_requestvote(raft_server_t* me_, raft_node_t* node) int raft_send_requestvote(raft_server_t* me_, raft_node_t* node)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
msg_requestvote_t rv; RequestVoteMsg_t rv;
int e = 0; int e = 0;
assert(node); assert(node);
...@@ -863,7 +863,7 @@ int raft_send_appendentries(raft_server_t* me_, raft_node_t* node) ...@@ -863,7 +863,7 @@ int raft_send_appendentries(raft_server_t* me_, raft_node_t* node)
if (!(me->cb.send_appendentries)) if (!(me->cb.send_appendentries))
return -1; return -1;
msg_appendentries_t ae = {}; AppendEntriesMsg_t ae = {};
ae.term = me->current_term; ae.term = me->current_term;
ae.leader_commit = raft_get_commit_idx(me_); ae.leader_commit = raft_get_commit_idx(me_);
ae.prev_log_idx = 0; ae.prev_log_idx = 0;
...@@ -1035,7 +1035,7 @@ int raft_vote_for_nodeid(raft_server_t* me_, const int nodeid) ...@@ -1035,7 +1035,7 @@ int raft_vote_for_nodeid(raft_server_t* me_, const int nodeid)
} }
int raft_msg_entry_response_committed(raft_server_t* me_, int raft_msg_entry_response_committed(raft_server_t* me_,
const msg_entry_response_t* r) const LogEntryResponseMsg_table_t* r)
{ {
raft_entry_t* ety = raft_get_entry_from_idx(me_, r->idx); raft_entry_t* ety = raft_get_entry_from_idx(me_, r->idx);
if (!ety) if (!ety)
......
table LogEntryMsg
{
term: uint64;
id: uint64;
type: uint64;
data: [byte];
}
table LogEntryResponseMsg
{
term: uint64;
id: uint64;
idx: uint64;
}
struct RequestVoteMsg
{
term: uint64;
candidateId: uint64;
lastLogIdx: uint64;
lastLogTerm: uint64;
}
struct RequestVoteResponseMsg
{
term: uint64;
voteGranted: uint64;
}
table AppendEntriesMsg
{
term: uint64;
prevLogIdx: uint64;
prevLogTerm: uint64;
leaderCommit: uint64;
nEntries: uint64;
entries: [LogEntryMsg];
}
struct AppendEntriesResponseMsg
{
term: uint64;
success: uint64;
currentIdx: uint64;
firstIdx: uint64;
}
...@@ -74,7 +74,7 @@ static int __append_msg( ...@@ -74,7 +74,7 @@ static int __append_msg(
} }
int sender_requestvote(raft_server_t* raft, int sender_requestvote(raft_server_t* raft,
void* udata, raft_node_t* node, msg_requestvote_t* msg) void* udata, raft_node_t* node, RequestVoteMsg_t* msg)
{ {
return __append_msg(udata, msg, RAFT_MSG_REQUESTVOTE, sizeof(*msg), node, return __append_msg(udata, msg, RAFT_MSG_REQUESTVOTE, sizeof(*msg), node,
raft); raft);
...@@ -82,17 +82,17 @@ int sender_requestvote(raft_server_t* raft, ...@@ -82,17 +82,17 @@ int sender_requestvote(raft_server_t* raft,
int sender_requestvote_response(raft_server_t* raft, int sender_requestvote_response(raft_server_t* raft,
void* udata, raft_node_t* node, void* udata, raft_node_t* node,
msg_requestvote_response_t* msg) RequestVoteResponseMsg_t* msg)
{ {
return __append_msg(udata, msg, RAFT_MSG_REQUESTVOTE_RESPONSE, sizeof(*msg), return __append_msg(udata, msg, RAFT_MSG_REQUESTVOTE_RESPONSE, sizeof(*msg),
node, raft); node, 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, AppendEntriesMsg_t* msg)
{ {
msg_entry_t* entries = calloc(1, sizeof(msg_entry_t) * msg->n_entries); LogEntryMsg_table_t* entries = calloc(1, sizeof(LogEntryMsg_table_t) * msg->n_entries);
memcpy(entries, msg->entries, sizeof(msg_entry_t) * msg->n_entries); memcpy(entries, msg->entries, sizeof(LogEntryMsg_table_t) * msg->n_entries);
msg->entries = entries; msg->entries = entries;
return __append_msg(udata, msg, RAFT_MSG_APPENDENTRIES, sizeof(*msg), node, return __append_msg(udata, msg, RAFT_MSG_APPENDENTRIES, sizeof(*msg), node,
raft); raft);
...@@ -100,14 +100,14 @@ int sender_appendentries(raft_server_t* raft, ...@@ -100,14 +100,14 @@ int sender_appendentries(raft_server_t* raft,
int sender_appendentries_response(raft_server_t* raft, int sender_appendentries_response(raft_server_t* raft,
void* udata, raft_node_t* node, void* udata, raft_node_t* node,
msg_appendentries_response_t* msg) AppendEntriesResponseMsg_t* msg)
{ {
return __append_msg(udata, msg, RAFT_MSG_APPENDENTRIES_RESPONSE, return __append_msg(udata, msg, RAFT_MSG_APPENDENTRIES_RESPONSE,
sizeof(*msg), node, raft); sizeof(*msg), node, raft);
} }
int sender_entries_response(raft_server_t* raft, int sender_entries_response(raft_server_t* raft,
void* udata, raft_node_t* node, msg_entry_response_t* msg) void* udata, raft_node_t* node, LogEntryResponseMsg_table_t* msg)
{ {
return __append_msg(udata, msg, RAFT_MSG_ENTRY_RESPONSE, sizeof(*msg), node, return __append_msg(udata, msg, RAFT_MSG_ENTRY_RESPONSE, sizeof(*msg), node,
raft); raft);
...@@ -154,7 +154,7 @@ void sender_poll_msgs(void* s) ...@@ -154,7 +154,7 @@ void sender_poll_msgs(void* s)
{ {
case RAFT_MSG_APPENDENTRIES: case RAFT_MSG_APPENDENTRIES:
{ {
msg_appendentries_response_t response; AppendEntriesResponseMsg_t response;
raft_recv_appendentries(me->raft, m->sender, m->data, &response); raft_recv_appendentries(me->raft, m->sender, m->data, &response);
__append_msg(me, &response, RAFT_MSG_APPENDENTRIES_RESPONSE, __append_msg(me, &response, RAFT_MSG_APPENDENTRIES_RESPONSE,
sizeof(response), m->sender, me->raft); sizeof(response), m->sender, me->raft);
...@@ -165,7 +165,7 @@ void sender_poll_msgs(void* s) ...@@ -165,7 +165,7 @@ void sender_poll_msgs(void* s)
break; break;
case RAFT_MSG_REQUESTVOTE: case RAFT_MSG_REQUESTVOTE:
{ {
msg_requestvote_response_t response; RequestVoteResponseMsg_t response;
raft_recv_requestvote(me->raft, m->sender, m->data, &response); raft_recv_requestvote(me->raft, m->sender, m->data, &response);
__append_msg(me, &response, RAFT_MSG_REQUESTVOTE_RESPONSE, __append_msg(me, &response, RAFT_MSG_REQUESTVOTE_RESPONSE,
sizeof(response), m->sender, me->raft); sizeof(response), m->sender, me->raft);
...@@ -176,7 +176,7 @@ void sender_poll_msgs(void* s) ...@@ -176,7 +176,7 @@ void sender_poll_msgs(void* s)
break; break;
case RAFT_MSG_ENTRY: case RAFT_MSG_ENTRY:
{ {
msg_entry_response_t response; LogEntryResponseMsg_table_t response;
raft_recv_entry(me->raft, m->data, &response); raft_recv_entry(me->raft, m->data, &response);
__append_msg(me, &response, RAFT_MSG_ENTRY_RESPONSE, __append_msg(me, &response, RAFT_MSG_ENTRY_RESPONSE,
sizeof(response), m->sender, me->raft); sizeof(response), m->sender, me->raft);
......
...@@ -24,21 +24,21 @@ int sender_msgs_available(void* s); ...@@ -24,21 +24,21 @@ int sender_msgs_available(void* s);
void sender_set_raft(void* s, void* r); void sender_set_raft(void* s, void* r);
int sender_requestvote(raft_server_t* raft, int sender_requestvote(raft_server_t* raft,
void* udata, raft_node_t* node, msg_requestvote_t* msg); void* udata, raft_node_t* node, RequestVoteMsg_t* msg);
int sender_requestvote_response(raft_server_t* raft, int sender_requestvote_response(raft_server_t* raft,
void* udata, raft_node_t* node, msg_requestvote_response_t* msg); void* udata, raft_node_t* node, RequestVoteResponseMsg_t* msg);
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, AppendEntriesMsg_t* msg);
int sender_appendentries_response(raft_server_t* raft, int sender_appendentries_response(raft_server_t* raft,
void* udata, raft_node_t* node, msg_appendentries_response_t* msg); void* udata, raft_node_t* node, AppendEntriesResponseMsg_t* msg);
int sender_entries(raft_server_t* raft, int sender_entries(raft_server_t* raft,
void* udata, raft_node_t* node, msg_entry_t* msg); void* udata, raft_node_t* node, LogEntryMsg_table_t* msg);
int sender_entries_response(raft_server_t* raft, int sender_entries_response(raft_server_t* raft,
void* udata, raft_node_t* node, msg_entry_response_t* msg); void* udata, raft_node_t* node, LogEntryResponseMsg_table_t* msg);
#endif /* MOCK_SEND_FUNCTIONS_H */ #endif /* MOCK_SEND_FUNCTIONS_H */
This diff is collapsed.
...@@ -43,7 +43,7 @@ static int __raft_applylog( ...@@ -43,7 +43,7 @@ static int __raft_applylog(
static int __raft_send_requestvote(raft_server_t* raft, static int __raft_send_requestvote(raft_server_t* raft,
void* udata, void* udata,
raft_node_t* node, raft_node_t* node,
msg_requestvote_t* msg) RequestVoteMsg_t* msg)
{ {
return 0; return 0;
} }
...@@ -51,7 +51,7 @@ static int __raft_send_requestvote(raft_server_t* raft, ...@@ -51,7 +51,7 @@ static int __raft_send_requestvote(raft_server_t* raft,
static int __raft_send_appendentries(raft_server_t* raft, static int __raft_send_appendentries(raft_server_t* raft,
void* udata, void* udata,
raft_node_t* node, raft_node_t* node,
msg_appendentries_t* msg) AppendEntriesMsg_t* msg)
{ {
return 0; return 0;
} }
...@@ -59,10 +59,10 @@ static int __raft_send_appendentries(raft_server_t* raft, ...@@ -59,10 +59,10 @@ static int __raft_send_appendentries(raft_server_t* raft,
static int __raft_send_appendentries_capture(raft_server_t* raft, static int __raft_send_appendentries_capture(raft_server_t* raft,
void* udata, void* udata,
raft_node_t* node, raft_node_t* node,
msg_appendentries_t* msg) AppendEntriesMsg_t* msg)
{ {
msg_appendentries_t* msg_captured = (msg_appendentries_t*)udata; AppendEntriesMsg_t* msg_captured = (AppendEntriesMsg_t*)udata;
memcpy(msg_captured, msg, sizeof(msg_appendentries_t)); memcpy(msg_captured, msg, sizeof(AppendEntriesMsg_t));
return 0; return 0;
} }
...@@ -88,7 +88,7 @@ void TestRaft_leader_begin_snapshot_fails_if_no_logs_to_compact(CuTest * tc) ...@@ -88,7 +88,7 @@ void TestRaft_leader_begin_snapshot_fails_if_no_logs_to_compact(CuTest * tc)
void *r = raft_new(); void *r = raft_new();
raft_set_callbacks(r, &funcs, NULL); raft_set_callbacks(r, &funcs, NULL);
msg_entry_response_t cr; LogEntryResponseMsg_table_t cr;
raft_add_node(r, NULL, 1, 1); raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0); raft_add_node(r, NULL, 2, 0);
...@@ -98,7 +98,7 @@ void TestRaft_leader_begin_snapshot_fails_if_no_logs_to_compact(CuTest * tc) ...@@ -98,7 +98,7 @@ 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 = {}; LogEntryMsg_table_t ety = {};
ety.id = 1; ety.id = 1;
ety.data.buf = "entry"; ety.data.buf = "entry";
ety.data.len = strlen("entry"); ety.data.len = strlen("entry");
...@@ -123,7 +123,7 @@ void TestRaft_leader_will_not_apply_entry_if_snapshot_is_in_progress(CuTest * tc ...@@ -123,7 +123,7 @@ void TestRaft_leader_will_not_apply_entry_if_snapshot_is_in_progress(CuTest * tc
void *r = raft_new(); void *r = raft_new();
raft_set_callbacks(r, &funcs, NULL); raft_set_callbacks(r, &funcs, NULL);
msg_entry_response_t cr; LogEntryResponseMsg_table_t cr;
raft_add_node(r, NULL, 1, 1); raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0); raft_add_node(r, NULL, 2, 0);
...@@ -133,7 +133,7 @@ void TestRaft_leader_will_not_apply_entry_if_snapshot_is_in_progress(CuTest * tc ...@@ -133,7 +133,7 @@ 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 = {}; LogEntryMsg_table_t ety = {};
ety.id = 1; ety.id = 1;
ety.data.buf = "entry"; ety.data.buf = "entry";
ety.data.len = strlen("entry"); ety.data.len = strlen("entry");
...@@ -179,7 +179,7 @@ void TestRaft_leader_snapshot_begin_fails_if_less_than_2_logs_to_compact(CuTest ...@@ -179,7 +179,7 @@ void TestRaft_leader_snapshot_begin_fails_if_less_than_2_logs_to_compact(CuTest
void *r = raft_new(); void *r = raft_new();
raft_set_callbacks(r, &funcs, NULL); raft_set_callbacks(r, &funcs, NULL);
msg_entry_response_t cr; LogEntryResponseMsg_table_t cr;
raft_add_node(r, NULL, 1, 1); raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0); raft_add_node(r, NULL, 2, 0);
...@@ -189,7 +189,7 @@ void TestRaft_leader_snapshot_begin_fails_if_less_than_2_logs_to_compact(CuTest ...@@ -189,7 +189,7 @@ 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 = {}; LogEntryMsg_table_t ety = {};
ety.id = 1; ety.id = 1;
ety.data.buf = "entry"; ety.data.buf = "entry";
ety.data.len = strlen("entry"); ety.data.len = strlen("entry");
...@@ -211,7 +211,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc) ...@@ -211,7 +211,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc)
void *r = raft_new(); void *r = raft_new();
raft_set_callbacks(r, &funcs, NULL); raft_set_callbacks(r, &funcs, NULL);
msg_entry_response_t cr; LogEntryResponseMsg_table_t cr;
raft_add_node(r, NULL, 1, 1); raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0); raft_add_node(r, NULL, 2, 0);
...@@ -222,7 +222,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted(CuTest * tc) ...@@ -222,7 +222,7 @@ 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 = {}; LogEntryMsg_table_t ety = {};
ety.id = 1; ety.id = 1;
ety.data.buf = "entry"; ety.data.buf = "entry";
ety.data.len = strlen("entry"); ety.data.len = strlen("entry");
...@@ -260,7 +260,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted2(CuTest * tc) ...@@ -260,7 +260,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted2(CuTest * tc)
void *r = raft_new(); void *r = raft_new();
raft_set_callbacks(r, &funcs, NULL); raft_set_callbacks(r, &funcs, NULL);
msg_entry_response_t cr; LogEntryResponseMsg_table_t cr;
raft_add_node(r, NULL, 1, 1); raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0); raft_add_node(r, NULL, 2, 0);
...@@ -271,7 +271,7 @@ void TestRaft_leader_snapshot_end_succeeds_if_log_compacted2(CuTest * tc) ...@@ -271,7 +271,7 @@ 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 = {}; LogEntryMsg_table_t ety = {};
ety.id = 1; ety.id = 1;
ety.data.buf = "entry"; ety.data.buf = "entry";
ety.data.len = strlen("entry"); ety.data.len = strlen("entry");
...@@ -310,7 +310,7 @@ void TestRaft_joinee_needs_to_get_snapshot(CuTest * tc) ...@@ -310,7 +310,7 @@ void TestRaft_joinee_needs_to_get_snapshot(CuTest * tc)
void *r = raft_new(); void *r = raft_new();
raft_set_callbacks(r, &funcs, NULL); raft_set_callbacks(r, &funcs, NULL);
msg_entry_response_t cr; LogEntryResponseMsg_table_t cr;
raft_add_node(r, NULL, 1, 1); raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0); raft_add_node(r, NULL, 2, 0);
...@@ -320,7 +320,7 @@ void TestRaft_joinee_needs_to_get_snapshot(CuTest * tc) ...@@ -320,7 +320,7 @@ 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 = {}; LogEntryMsg_table_t ety = {};
ety.id = 1; ety.id = 1;
ety.data.buf = "entry"; ety.data.buf = "entry";
ety.data.len = strlen("entry"); ety.data.len = strlen("entry");
...@@ -354,7 +354,7 @@ void TestRaft_follower_load_from_snapshot(CuTest * tc) ...@@ -354,7 +354,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 = {}; LogEntryMsg_table_t ety = {};
ety.id = 1; ety.id = 1;
ety.data.buf = "entry"; ety.data.buf = "entry";
ety.data.len = strlen("entry"); ety.data.len = strlen("entry");
...@@ -410,7 +410,7 @@ void TestRaft_follower_load_from_snapshot_fails_if_already_loaded(CuTest * tc) ...@@ -410,7 +410,7 @@ void TestRaft_follower_load_from_snapshot_fails_if_already_loaded(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 = {}; LogEntryMsg_table_t ety = {};
ety.id = 1; ety.id = 1;
ety.data.buf = "entry"; ety.data.buf = "entry";
ety.data.len = strlen("entry"); ety.data.len = strlen("entry");
...@@ -441,7 +441,7 @@ void TestRaft_follower_load_from_snapshot_does_not_break_cluster_safety(CuTest * ...@@ -441,7 +441,7 @@ 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 = {}; LogEntryMsg_table_t ety = {};
ety.id = 1; ety.id = 1;
ety.data.buf = "entry"; ety.data.buf = "entry";
ety.data.len = strlen("entry"); ety.data.len = strlen("entry");
...@@ -477,7 +477,7 @@ void TestRaft_follower_load_from_snapshot_fails_if_log_is_newer(CuTest * tc) ...@@ -477,7 +477,7 @@ 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 */ /* entry message */
msg_entry_t ety = {}; LogEntryMsg_table_t ety = {};
ety.id = 1; ety.id = 1;
ety.data.buf = "entry"; ety.data.buf = "entry";
ety.data.len = strlen("entry"); ety.data.len = strlen("entry");
...@@ -491,7 +491,7 @@ void TestRaft_leader_sends_appendentries_when_node_next_index_was_compacted(CuTe ...@@ -491,7 +491,7 @@ void TestRaft_leader_sends_appendentries_when_node_next_index_was_compacted(CuTe
.send_appendentries = __raft_send_appendentries_capture, .send_appendentries = __raft_send_appendentries_capture,
}; };
msg_appendentries_t ae; AppendEntriesMsg_t ae;
void *r = raft_new(); void *r = raft_new();
raft_set_callbacks(r, &funcs, &ae); raft_set_callbacks(r, &funcs, &ae);
...@@ -550,7 +550,7 @@ void TestRaft_recv_entry_fails_if_snapshot_in_progress(CuTest* tc) ...@@ -550,7 +550,7 @@ void TestRaft_recv_entry_fails_if_snapshot_in_progress(CuTest* tc)
void *r = raft_new(); void *r = raft_new();
raft_set_callbacks(r, &funcs, NULL); raft_set_callbacks(r, &funcs, NULL);
msg_entry_response_t cr; LogEntryResponseMsg_table_t cr;
raft_add_node(r, NULL, 1, 1); raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0); raft_add_node(r, NULL, 2, 0);
...@@ -560,7 +560,7 @@ void TestRaft_recv_entry_fails_if_snapshot_in_progress(CuTest* tc) ...@@ -560,7 +560,7 @@ 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 = {}; LogEntryMsg_table_t ety = {};
ety.id = 1; ety.id = 1;
ety.data.buf = "entry"; ety.data.buf = "entry";
ety.data.len = strlen("entry"); ety.data.len = strlen("entry");
......
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