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

Made AppendEntries easier

parent 5e9e46ec
...@@ -155,20 +155,6 @@ typedef int ( ...@@ -155,20 +155,6 @@ typedef int (
msg_appendentries_t* msg msg_appendentries_t* msg
); );
/**
* @param raft The Raft server making this callback
* @param udata User data that is passed from Raft server
* @param node The peer's ID that we are sending this message to
* @return 0 on error */
typedef int (
*func_send_appendentries_response_f
) (
raft_server_t* raft,
void *udata,
int node,
msg_appendentries_response_t* msg
);
/** /**
* @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
...@@ -230,7 +216,6 @@ typedef struct ...@@ -230,7 +216,6 @@ typedef struct
func_send_requestvote_f send_requestvote; func_send_requestvote_f send_requestvote;
func_send_requestvote_response_f send_requestvote_response; func_send_requestvote_response_f send_requestvote_response;
func_send_appendentries_f send_appendentries; func_send_appendentries_f send_appendentries;
func_send_appendentries_response_f send_appendentries_response;
func_send_entries_f send_entries; func_send_entries_f send_entries;
func_send_entries_response_f send_entries_response; func_send_entries_response_f send_entries_response;
func_log_f log; func_log_f log;
...@@ -302,9 +287,11 @@ int raft_periodic(raft_server_t* me, int msec_elapsed); ...@@ -302,9 +287,11 @@ int raft_periodic(raft_server_t* me, int msec_elapsed);
* Receive an appendentries message * Receive an appendentries message
* @param node Index of the node who sent us this message * @param node Index of the node who sent us this message
* @param ae The appendentries message * @param ae The appendentries message
* @param[out] r The resulting response
* @return 0 on error */ * @return 0 on error */
int raft_recv_appendentries(raft_server_t* me, int node, int raft_recv_appendentries(raft_server_t* me, int node,
msg_appendentries_t* ae); msg_appendentries_t* ae,
msg_appendentries_response_t *r);
/** /**
* Receive a response from an appendentries message we sent * Receive a response from an appendentries message we sent
......
/** /**
* Copyright (c) 2013, Willem-Hendrik Thiart * Copyright (c) 2013, Willem-Hendrik Thiart
* Use of this source code is governed by a BSD-style license that can be * Use of this source code is governed by a BSD-style license that can be
...@@ -38,9 +37,7 @@ typedef struct ...@@ -38,9 +37,7 @@ typedef struct
raft_entry_t* entries; raft_entry_t* entries;
} log_private_t; } log_private_t;
static void __ensurecapacity( static void __ensurecapacity(log_private_t * me)
log_private_t * me
)
{ {
int i, j; int i, j;
raft_entry_t *temp; raft_entry_t *temp;
...@@ -68,9 +65,7 @@ static void __ensurecapacity( ...@@ -68,9 +65,7 @@ static void __ensurecapacity(
log_t* log_new() log_t* log_new()
{ {
log_private_t* me; log_private_t* me = calloc(1, sizeof(log_private_t));
me = calloc(1, sizeof(log_private_t));
me->size = INITIAL_CAPACITY; me->size = INITIAL_CAPACITY;
me->count = 0; me->count = 0;
me->back = in(me)->front = 0; me->back = in(me)->front = 0;
...@@ -113,8 +108,7 @@ raft_entry_t* log_get_from_idx(log_t* me_, int idx) ...@@ -113,8 +108,7 @@ raft_entry_t* log_get_from_idx(log_t* me_, int idx)
int log_count(log_t* me_) int log_count(log_t* me_)
{ {
log_private_t* me = (void*)me_; return ((log_private_t*)me_)->count;
return me->count;
} }
void log_delete(log_t* me_, int idx) void log_delete(log_t* me_, int idx)
...@@ -200,4 +194,3 @@ void log_mark_node_has_committed(log_t* me_, int idx) ...@@ -200,4 +194,3 @@ void log_mark_node_has_committed(log_t* me_, int idx)
if ((e = log_get_from_idx(me_, idx))) if ((e = log_get_from_idx(me_, idx)))
e->num_nodes += 1; e->num_nodes += 1;
} }
/** /**
* Copyright (c) 2013, Willem-Hendrik Thiart * Copyright (c) 2013, Willem-Hendrik Thiart
* Use of this source code is governed by a BSD-style license that can be * Use of this source code is governed by a BSD-style license that can be
* found in the LICENSE file. * found in the LICENSE file.
* *
* @file * @file
* @brief Representation of a peer * @brief Representation of a peer
...@@ -17,7 +16,8 @@ ...@@ -17,7 +16,8 @@
#include "raft.h" #include "raft.h"
typedef struct { typedef struct
{
void* udata; void* udata;
int next_idx; int next_idx;
} raft_node_private_t; } raft_node_private_t;
...@@ -25,7 +25,7 @@ typedef struct { ...@@ -25,7 +25,7 @@ typedef struct {
raft_node_t* raft_node_new(void* udata) raft_node_t* raft_node_new(void* udata)
{ {
raft_node_private_t* me; raft_node_private_t* me;
me = calloc(1,sizeof(raft_node_private_t)); me = calloc(1, sizeof(raft_node_private_t));
me->udata = udata; me->udata = udata;
return (void*)me; return (void*)me;
} }
......
/** /**
* Copyright (c) 2013, Willem-Hendrik Thiart * Copyright (c) 2013, Willem-Hendrik Thiart
* Use of this source code is governed by a BSD-style license that can be * Use of this source code is governed by a BSD-style license that can be
...@@ -22,7 +21,6 @@ ...@@ -22,7 +21,6 @@
#include "raft_log.h" #include "raft_log.h"
#include "raft_private.h" #include "raft_private.h"
static void __log(raft_server_t *me_, const char *fmt, ...) static void __log(raft_server_t *me_, const char *fmt, ...)
{ {
char buf[1024]; char buf[1024];
...@@ -223,16 +221,17 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -223,16 +221,17 @@ 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_,
const int node, const int node,
msg_appendentries_t* ae) msg_appendentries_t* ae,
msg_appendentries_response_t *r
)
{ {
raft_server_private_t* me = (void*)me_; raft_server_private_t* me = (void*)me_;
msg_appendentries_response_t r;
me->timeout_elapsed = 0; me->timeout_elapsed = 0;
__log(me_, "received appendentries from: %d", node); __log(me_, "received appendentries from: %d", node);
r.term = me->current_term; r->term = me->current_term;
/* we've found a leader who is legitimate */ /* we've found a leader who is legitimate */
if (raft_is_leader(me_) && me->current_term <= ae->term) if (raft_is_leader(me_) && me->current_term <= ae->term)
...@@ -242,8 +241,8 @@ int raft_recv_appendentries( ...@@ -242,8 +241,8 @@ int raft_recv_appendentries(
if (ae->term < me->current_term) if (ae->term < me->current_term)
{ {
__log(me_, "AE term is less than current term"); __log(me_, "AE term is less than current term");
r.success = 0; r->success = 0;
goto done; return 1;
} }
#if 0 #if 0
...@@ -251,8 +250,8 @@ int raft_recv_appendentries( ...@@ -251,8 +250,8 @@ int raft_recv_appendentries(
ae->prev_log_idx < raft_get_current_idx(me_)) ae->prev_log_idx < raft_get_current_idx(me_))
{ {
__log(me_, "AE prev_idx is less than current idx"); __log(me_, "AE prev_idx is less than current idx");
r.success = 0; r->success = 0;
goto done; return 1;
} }
#endif #endif
...@@ -263,13 +262,13 @@ int raft_recv_appendentries( ...@@ -263,13 +262,13 @@ int raft_recv_appendentries(
if ((e = raft_get_entry_from_idx(me_, ae->prev_log_idx))) if ((e = raft_get_entry_from_idx(me_, ae->prev_log_idx)))
{ {
/* 2. Reply false if log doesn’t contain an entry at prevLogIndex /* 2. Reply false if log doesn't contain an entry at prevLogIndex
whose term matches prevLogTerm (§5.3) */ whose term matches prevLogTerm (§5.3) */
if (e->term != ae->prev_log_term) if (e->term != ae->prev_log_term)
{ {
__log(me_, "AE term doesn't match prev_idx"); __log(me_, "AE term doesn't match prev_idx");
r.success = 0; r->success = 0;
goto done; return 1;
} }
/* 3. If an existing entry conflicts with a new one (same index /* 3. If an existing entry conflicts with a new one (same index
...@@ -282,9 +281,8 @@ int raft_recv_appendentries( ...@@ -282,9 +281,8 @@ int raft_recv_appendentries(
else else
{ {
__log(me_, "AE no log at prev_idx"); __log(me_, "AE no log at prev_idx");
r.success = 0; r->success = 0;
goto done; return 1;
//assert(0);
} }
} }
...@@ -328,18 +326,14 @@ int raft_recv_appendentries( ...@@ -328,18 +326,14 @@ int raft_recv_appendentries(
if (0 == raft_append_entry(me_, c)) if (0 == raft_append_entry(me_, c))
{ {
__log(me_, "AE failure; couldn't append entry"); __log(me_, "AE failure; couldn't append entry");
r.success = 0; r->success = 0;
goto done; return 1;
} }
} }
r.success = 1; r->success = 1;
r.current_idx = raft_get_current_idx(me_); r->current_idx = raft_get_current_idx(me_);
r.first_idx = ae->prev_log_idx + 1; r->first_idx = ae->prev_log_idx + 1;
done:
if (me->cb.send_appendentries_response)
me->cb.send_appendentries_response(me_, me->udata, node, &r);
return 1; return 1;
} }
...@@ -571,29 +565,3 @@ void raft_vote(raft_server_t* me_, int node) ...@@ -571,29 +565,3 @@ void raft_vote(raft_server_t* me_, int node)
raft_server_private_t* me = (void*)me_; raft_server_private_t* me = (void*)me_;
me->voted_for = node; me->voted_for = node;
} }
raft_node_t* raft_get_node(raft_server_t *me_, int nodeid)
{
raft_server_private_t* me = (void*)me_;
if (nodeid < 0 || me->num_nodes <= nodeid)
return NULL;
return me->nodes[nodeid];
}
int raft_is_follower(raft_server_t* me_)
{
return raft_get_state(me_) == RAFT_STATE_FOLLOWER;
}
int raft_is_leader(raft_server_t* me_)
{
return raft_get_state(me_) == RAFT_STATE_LEADER;
}
int raft_is_candidate(raft_server_t* me_)
{
return raft_get_state(me_) == RAFT_STATE_CANDIDATE;
}
/*--------------------------------------------------------------79-characters-*/
/** /**
* Copyright (c) 2013, Willem-Hendrik Thiart * Copyright (c) 2013, Willem-Hendrik Thiart
* Use of this source code is governed by a BSD-style license that can be * Use of this source code is governed by a BSD-style license that can be
* found in the LICENSE file. * found in the LICENSE file.
* *
* @file * @file
* @author Willem Thiart himself@willemthiart.com * @author Willem Thiart himself@willemthiart.com
...@@ -53,7 +52,6 @@ int raft_get_num_nodes(raft_server_t* me_) ...@@ -53,7 +52,6 @@ int raft_get_num_nodes(raft_server_t* me_)
return ((raft_server_private_t*)me_)->num_nodes; return ((raft_server_private_t*)me_)->num_nodes;
} }
int raft_get_timeout_elapsed(raft_server_t* me_) int raft_get_timeout_elapsed(raft_server_t* me_)
{ {
return ((raft_server_private_t*)me_)->timeout_elapsed; return ((raft_server_private_t*)me_)->timeout_elapsed;
...@@ -130,3 +128,26 @@ int raft_get_state(raft_server_t* me_) ...@@ -130,3 +128,26 @@ int raft_get_state(raft_server_t* me_)
return ((raft_server_private_t*)me_)->state; return ((raft_server_private_t*)me_)->state;
} }
raft_node_t* raft_get_node(raft_server_t *me_, int nodeid)
{
raft_server_private_t* me = (void*)me_;
if (nodeid < 0 || me->num_nodes <= nodeid)
return NULL;
return me->nodes[nodeid];
}
int raft_is_follower(raft_server_t* me_)
{
return raft_get_state(me_) == RAFT_STATE_FOLLOWER;
}
int raft_is_leader(raft_server_t* me_)
{
return raft_get_state(me_) == RAFT_STATE_LEADER;
}
int raft_is_candidate(raft_server_t* me_)
{
return raft_get_state(me_) == RAFT_STATE_CANDIDATE;
}
...@@ -16,13 +16,15 @@ ...@@ -16,13 +16,15 @@
#include "raft.h" #include "raft.h"
typedef struct { typedef struct
{
void* outbox; void* outbox;
void* inbox; void* inbox;
void* raft; void* raft;
} sender_t; } sender_t;
typedef struct { typedef struct
{
void* data; void* data;
int len; int len;
/* what type of message is it? */ /* what type of message is it? */
...@@ -40,43 +42,14 @@ void senders_new() ...@@ -40,43 +42,14 @@ void senders_new()
__nsenders = 0; __nsenders = 0;
} }
static int __append_msg(
#if 0
int sender_send(raft_server_t* raft,
void* udata,
int peer,
raft_message_type_e type,
const unsigned char* data,
int len)
{
sender_t* me = udata;
msg_t* m;
m = malloc(sizeof(msg_t));
m->type = type;
m->len = len;
m->data = malloc(len);
m->sender = raft_get_nodeid(raft);
memcpy(m->data,data,len);
llqueue_offer(me->outbox,m);
if (__nsenders > peer)
{
llqueue_offer(__senders[peer]->inbox, m);
}
return 0;
}
#endif
int __append_msg(
sender_t* me, sender_t* me,
void* data, void* data,
int type, int type,
int len, int len,
int peer, int peer,
raft_server_t* raft raft_server_t* raft
) )
{ {
msg_t* m; msg_t* m;
m = malloc(sizeof(msg_t)); m = malloc(sizeof(msg_t));
...@@ -84,71 +57,72 @@ int __append_msg( ...@@ -84,71 +57,72 @@ int __append_msg(
m->len = len; m->len = len;
m->data = malloc(len); m->data = malloc(len);
m->sender = raft_get_nodeid(raft); m->sender = raft_get_nodeid(raft);
memcpy(m->data,data,len); memcpy(m->data, data, len);
llqueue_offer(me->outbox,m); llqueue_offer(me->outbox, m);
if (__nsenders > peer) if (__nsenders > peer)
{
llqueue_offer(__senders[peer]->inbox, m); llqueue_offer(__senders[peer]->inbox, m);
}
return 1; return 1;
} }
int sender_requestvote(raft_server_t* raft, int sender_requestvote(raft_server_t* raft,
void* udata, int peer, msg_requestvote_t* msg) void* udata, int peer, msg_requestvote_t* msg)
{ {
return __append_msg(udata,msg,RAFT_MSG_REQUESTVOTE,sizeof(*msg),peer,raft); return __append_msg(udata, msg, RAFT_MSG_REQUESTVOTE, sizeof(*msg), peer,
raft);
} }
int sender_requestvote_response(raft_server_t* raft, int sender_requestvote_response(raft_server_t* raft,
void* udata, int peer, msg_requestvote_response_t* msg) void* udata, int peer,
msg_requestvote_response_t* msg)
{ {
return __append_msg(udata,msg,RAFT_MSG_REQUESTVOTE_RESPONSE,sizeof(*msg),peer,raft); return __append_msg(udata, msg, RAFT_MSG_REQUESTVOTE_RESPONSE, sizeof(*msg),
peer, raft);
} }
int sender_appendentries(raft_server_t* raft, int sender_appendentries(raft_server_t* raft,
void* udata, int peer, msg_appendentries_t* msg) void* udata, int peer, msg_appendentries_t* msg)
{ {
return __append_msg(udata,msg,RAFT_MSG_APPENDENTRIES,sizeof(*msg),peer,raft); return __append_msg(udata, msg, RAFT_MSG_APPENDENTRIES, sizeof(*msg), peer,
raft);
} }
int sender_appendentries_response(raft_server_t* raft, int sender_appendentries_response(raft_server_t* raft,
void* udata, int peer, msg_appendentries_response_t* msg) void* udata, int peer,
msg_appendentries_response_t* msg)
{ {
return __append_msg(udata,msg,RAFT_MSG_APPENDENTRIES_RESPONSE,sizeof(*msg),peer,raft); return __append_msg(udata, msg, RAFT_MSG_APPENDENTRIES_RESPONSE,
sizeof(*msg), peer, raft);
} }
int sender_entries(raft_server_t* raft, int sender_entries(raft_server_t* raft,
void* udata, int peer, msg_entry_t* msg) void* udata, int peer, msg_entry_t* msg)
{ {
return __append_msg(udata,msg,RAFT_MSG_ENTRY,sizeof(*msg),peer,raft); return __append_msg(udata, msg, RAFT_MSG_ENTRY, sizeof(*msg), peer, raft);
} }
int sender_entries_response(raft_server_t* raft, int sender_entries_response(raft_server_t* raft,
void* udata, int peer, msg_entry_response_t* msg) void* udata, int peer, msg_entry_response_t* msg)
{ {
return __append_msg(udata,msg,RAFT_MSG_ENTRY_RESPONSE,sizeof(*msg),peer,raft); return __append_msg(udata, msg, RAFT_MSG_ENTRY_RESPONSE, sizeof(*msg), peer,
raft);
} }
void* sender_new(void* address) void* sender_new(void* address)
{ {
sender_t* me; sender_t* me = malloc(sizeof(sender_t));
me = malloc(sizeof(sender_t));
me->outbox = llqueue_new(); me->outbox = llqueue_new();
me->inbox = llqueue_new(); me->inbox = llqueue_new();
__senders = realloc(__senders,sizeof(sender_t*) * (++__nsenders)); __senders = realloc(__senders, sizeof(sender_t*) * (++__nsenders));
__senders[__nsenders-1] = me; __senders[__nsenders - 1] = me;
return me; return me;
} }
void* sender_poll_msg_data(void* s) void* sender_poll_msg_data(void* s)
{ {
sender_t* me = s; sender_t* me = s;
msg_t* msg; msg_t* msg = llqueue_poll(me->outbox);
msg = llqueue_poll(me->outbox);
return NULL != msg ? msg->data : NULL; return NULL != msg ? msg->data : NULL;
} }
...@@ -174,26 +148,31 @@ void sender_poll_msgs(void* s) ...@@ -174,26 +148,31 @@ void sender_poll_msgs(void* s)
{ {
switch (m->type) switch (m->type)
{ {
case RAFT_MSG_APPENDENTRIES: case RAFT_MSG_APPENDENTRIES:
raft_recv_appendentries(me->raft, m->sender, m->data); {
break; msg_appendentries_response_t response;
case RAFT_MSG_APPENDENTRIES_RESPONSE: raft_recv_appendentries(me->raft, m->sender, m->data, &response);
raft_recv_appendentries_response(me->raft, m->sender, m->data); __append_msg(me, &response, RAFT_MSG_APPENDENTRIES_RESPONSE,
break; sizeof(response), m->sender, me->raft);
case RAFT_MSG_REQUESTVOTE: }
raft_recv_requestvote(me->raft, m->sender, m->data); break;
break; case RAFT_MSG_APPENDENTRIES_RESPONSE:
case RAFT_MSG_REQUESTVOTE_RESPONSE: raft_recv_appendentries_response(me->raft, m->sender, m->data);
raft_recv_requestvote_response(me->raft, m->sender, m->data); break;
break; case RAFT_MSG_REQUESTVOTE:
case RAFT_MSG_ENTRY: raft_recv_requestvote(me->raft, m->sender, m->data);
raft_recv_entry(me->raft, m->sender, m->data); break;
break; case RAFT_MSG_REQUESTVOTE_RESPONSE:
case RAFT_MSG_ENTRY_RESPONSE: raft_recv_requestvote_response(me->raft, m->sender, m->data);
//raft_recv_entry_response(me->raft, m->sender, m->data); break;
break; case RAFT_MSG_ENTRY:
raft_recv_entry(me->raft, m->sender, m->data);
break;
case RAFT_MSG_ENTRY_RESPONSE:
#if 0
raft_recv_entry_response(me->raft, m->sender, m->data);
#endif
break;
} }
} }
} }
#include <stdbool.h> #include <stdbool.h>
#include <assert.h> #include <assert.h>
#include <stdlib.h> #include <stdlib.h>
...@@ -14,56 +13,58 @@ ...@@ -14,56 +13,58 @@
void TestRaft_scenario_leader_appears(CuTest * tc) void TestRaft_scenario_leader_appears(CuTest * tc)
{ {
int i,j; int i, j;
raft_server_t *r[3]; raft_server_t *r[3];
void* sender[3]; void* sender[3];
raft_node_configuration_t cfg[] = { raft_node_configuration_t cfg[] = {
{(-1),(void*)1}, { (-1), (void*)1 },
{(-1),(void*)2}, { (-1), (void*)2 },
{(-1),(void*)3}, { (-1), (void*)3 },
{(-1),NULL}}; { (-1), NULL }
};
senders_new(); senders_new();
for (j=0;j<3;j++) for (j = 0; j < 3; j++)
{ {
r[j] = raft_new(); r[j] = raft_new();
sender[j] = sender_new(cfg[j].udata_address); sender[j] = sender_new(cfg[j].udata_address);
sender_set_raft(sender[j], r[j]); sender_set_raft(sender[j], r[j]);
raft_set_election_timeout(r[j], 500); raft_set_election_timeout(r[j], 500);
raft_set_configuration(r[j],cfg,j); raft_set_configuration(r[j], cfg, j);
raft_set_callbacks(r[j],&((raft_cbs_t) { raft_set_callbacks(r[j],
.send_requestvote = sender_requestvote, &((raft_cbs_t) {
.send_requestvote_response = sender_requestvote_response, .send_requestvote = sender_requestvote,
.send_appendentries = sender_appendentries, .send_requestvote_response =
.send_appendentries_response = sender_appendentries_response, sender_requestvote_response,
.send_entries = sender_entries, .send_appendentries =
.send_entries_response = sender_entries_response, sender_appendentries,
.log = NULL .send_entries = sender_entries,
}), sender[j]); .send_entries_response =
sender_entries_response,
.log = NULL
}), sender[j]);
} }
for (i=0;i<20;i++) for (i = 0; i < 20; i++)
{ {
one_more_time: one_more_time:
for (j=0;j<3;j++) for (j = 0; j < 3; j++)
sender_poll_msgs(sender[j]); sender_poll_msgs(sender[j]);
for (j=0;j<3;j++) for (j = 0; j < 3; j++)
if (sender_msgs_available(sender[j])) if (sender_msgs_available(sender[j]))
goto one_more_time; goto one_more_time;
for (j=0;j<3;j++) for (j = 0; j < 3; j++)
raft_periodic(r[j], 100); raft_periodic(r[j], 100);
} }
int leaders = 0; int leaders = 0;
for (j=0;j<3;j++) for (j = 0; j < 3; j++)
{
if (raft_is_leader(r[j])) if (raft_is_leader(r[j]))
leaders += 1; leaders += 1;
}
CuAssertTrue(tc, 0 != leaders); CuAssertTrue(tc, 0 != leaders);
CuAssertTrue(tc, 1 == leaders); CuAssertTrue(tc, 1 == leaders);
......
...@@ -531,11 +531,10 @@ TestRaft_follower_recv_appendentries_reply_false_if_term_less_than_currentterm( ...@@ -531,11 +531,10 @@ TestRaft_follower_recv_appendentries_reply_false_if_term_less_than_currentterm(
void *r; void *r;
void *sender; void *sender;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.send_appendentries_response = sender_appendentries_response,
.log = NULL .log = NULL
}; };
msg_appendentries_t ae; msg_appendentries_t ae;
msg_appendentries_response_t *aer; msg_appendentries_response_t aer;
/* 2 nodes */ /* 2 nodes */
raft_node_configuration_t cfg[] = { raft_node_configuration_t cfg[] = {
...@@ -556,12 +555,10 @@ TestRaft_follower_recv_appendentries_reply_false_if_term_less_than_currentterm( ...@@ -556,12 +555,10 @@ TestRaft_follower_recv_appendentries_reply_false_if_term_less_than_currentterm(
/* higher current term */ /* higher current term */
raft_set_current_term(r, 5); raft_set_current_term(r, 5);
raft_recv_appendentries(r, 1, &ae); raft_recv_appendentries(r, 1, &ae, &aer);
/* response is false */ /* response is false */
aer = sender_poll_msg_data(sender); CuAssertTrue(tc, 0 == aer.success);
CuAssertTrue(tc, NULL != aer);
CuAssertTrue(tc, 0 == aer->success);
} }
/* TODO: check if test case is needed */ /* TODO: check if test case is needed */
...@@ -572,10 +569,9 @@ TestRaft_follower_recv_appendentries_updates_currentterm_if_term_gt_currentterm( ...@@ -572,10 +569,9 @@ TestRaft_follower_recv_appendentries_updates_currentterm_if_term_gt_currentterm(
void *r; void *r;
void *sender; void *sender;
msg_appendentries_t ae; msg_appendentries_t ae;
msg_appendentries_response_t *aer; msg_appendentries_response_t aer;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.send_appendentries_response = sender_appendentries_response,
.log = NULL .log = NULL
}; };
...@@ -601,10 +597,8 @@ TestRaft_follower_recv_appendentries_updates_currentterm_if_term_gt_currentterm( ...@@ -601,10 +597,8 @@ TestRaft_follower_recv_appendentries_updates_currentterm_if_term_gt_currentterm(
ae.term = 2; ae.term = 2;
/* appendentry has newer term, so we change our currentterm */ /* appendentry has newer term, so we change our currentterm */
raft_recv_appendentries(r, 1, &ae); raft_recv_appendentries(r, 1, &ae, &aer);
aer = sender_poll_msg_data(sender); CuAssertTrue(tc, 1 == aer.success);
CuAssertTrue(tc, NULL != aer);
CuAssertTrue(tc, 1 == aer->success);
/* term has been updated */ /* term has been updated */
CuAssertTrue(tc, 2 == raft_get_current_term(r)); CuAssertTrue(tc, 2 == raft_get_current_term(r));
} }
...@@ -615,6 +609,7 @@ void TestRaft_follower_doesnt_log_after_appendentry_if_no_entries_are_specified( ...@@ -615,6 +609,7 @@ void TestRaft_follower_doesnt_log_after_appendentry_if_no_entries_are_specified(
void *r; void *r;
msg_appendentries_t ae; msg_appendentries_t ae;
msg_appendentries_response_t aer;
/* 2 nodes */ /* 2 nodes */
raft_node_configuration_t cfg[] = { raft_node_configuration_t cfg[] = {
...@@ -639,7 +634,7 @@ void TestRaft_follower_doesnt_log_after_appendentry_if_no_entries_are_specified( ...@@ -639,7 +634,7 @@ void TestRaft_follower_doesnt_log_after_appendentry_if_no_entries_are_specified(
ae.leader_commit = 5; ae.leader_commit = 5;
ae.n_entries = 0; ae.n_entries = 0;
raft_recv_appendentries(r, 1, &ae); raft_recv_appendentries(r, 1, &ae, &aer);
CuAssertTrue(tc, 0 == raft_get_log_count(r)); CuAssertTrue(tc, 0 == raft_get_log_count(r));
} }
...@@ -649,11 +644,10 @@ void TestRaft_follower_increases_log_after_appendentry(CuTest * tc) ...@@ -649,11 +644,10 @@ void TestRaft_follower_increases_log_after_appendentry(CuTest * tc)
void *sender; void *sender;
msg_appendentries_t ae; msg_appendentries_t ae;
msg_entry_t ety; msg_entry_t ety;
msg_appendentries_response_t *aer; msg_appendentries_response_t aer;
char *str = "aaa"; char *str = "aaa";
raft_cbs_t funcs = { raft_cbs_t funcs = {
.send_appendentries_response = sender_appendentries_response,
.log = NULL .log = NULL
}; };
...@@ -689,10 +683,8 @@ void TestRaft_follower_increases_log_after_appendentry(CuTest * tc) ...@@ -689,10 +683,8 @@ void TestRaft_follower_increases_log_after_appendentry(CuTest * tc)
ae.entries = &ety; ae.entries = &ety;
ae.n_entries = 1; ae.n_entries = 1;
raft_recv_appendentries(r, 1, &ae); raft_recv_appendentries(r, 1, &ae, &aer);
aer = sender_poll_msg_data(sender); CuAssertTrue(tc, 1 == aer.success);
CuAssertTrue(tc, NULL != aer);
CuAssertTrue(tc, 1 == aer->success);
CuAssertTrue(tc, 1 == raft_get_log_count(r)); CuAssertTrue(tc, 1 == raft_get_log_count(r));
} }
...@@ -706,7 +698,6 @@ TestRaft_follower_recv_appendentries_reply_false_if_doesnt_have_log_at_prev_log_ ...@@ -706,7 +698,6 @@ TestRaft_follower_recv_appendentries_reply_false_if_doesnt_have_log_at_prev_log_
msg_entry_t ety; msg_entry_t ety;
char *str = "aaa"; char *str = "aaa";
raft_cbs_t funcs = { raft_cbs_t funcs = {
.send_appendentries_response = sender_appendentries_response,
.log = NULL .log = NULL
}; };
/* 2 nodes */ /* 2 nodes */
...@@ -717,7 +708,7 @@ TestRaft_follower_recv_appendentries_reply_false_if_doesnt_have_log_at_prev_log_ ...@@ -717,7 +708,7 @@ TestRaft_follower_recv_appendentries_reply_false_if_doesnt_have_log_at_prev_log_
}; };
msg_appendentries_t ae; msg_appendentries_t ae;
msg_appendentries_response_t *aer; msg_appendentries_response_t aer;
sender = sender_new(NULL); sender = sender_new(NULL);
r = raft_new(); r = raft_new();
...@@ -745,12 +736,10 @@ TestRaft_follower_recv_appendentries_reply_false_if_doesnt_have_log_at_prev_log_ ...@@ -745,12 +736,10 @@ TestRaft_follower_recv_appendentries_reply_false_if_doesnt_have_log_at_prev_log_
ae.n_entries = 1; ae.n_entries = 1;
/* trigger reply */ /* trigger reply */
raft_recv_appendentries(r, 1, &ae); raft_recv_appendentries(r, 1, &ae, &aer);
aer = sender_poll_msg_data(sender);
/* reply is false */ /* reply is false */
CuAssertTrue(tc, NULL != aer); CuAssertTrue(tc, 0 == aer.success);
CuAssertTrue(tc, 0 == aer->success);
} }
/* 5.3 */ /* 5.3 */
...@@ -761,11 +750,10 @@ TestRaft_follower_recv_appendentries_delete_entries_if_conflict_with_new_entries ...@@ -761,11 +750,10 @@ TestRaft_follower_recv_appendentries_delete_entries_if_conflict_with_new_entries
void *r; void *r;
void *sender; void *sender;
msg_appendentries_t ae; msg_appendentries_t ae;
msg_appendentries_response_t *aer; msg_appendentries_response_t aer;
raft_entry_t *ety_appended; raft_entry_t *ety_appended;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.send_appendentries_response = sender_appendentries_response,
.log = NULL .log = NULL
}; };
...@@ -821,10 +809,8 @@ TestRaft_follower_recv_appendentries_delete_entries_if_conflict_with_new_entries ...@@ -821,10 +809,8 @@ TestRaft_follower_recv_appendentries_delete_entries_if_conflict_with_new_entries
ae.entries = &mety; ae.entries = &mety;
ae.n_entries = 1; ae.n_entries = 1;
raft_recv_appendentries(r, 1, &ae); raft_recv_appendentries(r, 1, &ae, &aer);
aer = sender_poll_msg_data(sender); CuAssertTrue(tc, 1 == aer.success);
CuAssertTrue(tc, NULL != aer);
CuAssertTrue(tc, 1 == aer->success);
CuAssertTrue(tc, 2 == raft_get_log_count(r)); CuAssertTrue(tc, 2 == raft_get_log_count(r));
CuAssertTrue(tc, NULL != (ety_appended = raft_get_entry_from_idx(r, 1))); CuAssertTrue(tc, NULL != (ety_appended = raft_get_entry_from_idx(r, 1)));
CuAssertTrue(tc, !strncmp(ety_appended->data, str1, 3)); CuAssertTrue(tc, !strncmp(ety_appended->data, str1, 3));
...@@ -836,7 +822,6 @@ void TestRaft_follower_recv_appendentries_add_new_entries_not_already_in_log( ...@@ -836,7 +822,6 @@ void TestRaft_follower_recv_appendentries_add_new_entries_not_already_in_log(
void *r; void *r;
void *sender; void *sender;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.send_appendentries_response = sender_appendentries_response,
.log = NULL .log = NULL
}; };
...@@ -855,6 +840,8 @@ void TestRaft_follower_recv_appendentries_add_new_entries_not_already_in_log( ...@@ -855,6 +840,8 @@ void TestRaft_follower_recv_appendentries_add_new_entries_not_already_in_log(
raft_set_callbacks(r, &funcs, sender); raft_set_callbacks(r, &funcs, sender);
msg_appendentries_t ae; msg_appendentries_t ae;
msg_appendentries_response_t aer;
memset(&ae, 0, sizeof(msg_appendentries_t)); memset(&ae, 0, sizeof(msg_appendentries_t));
ae.term = 1; ae.term = 1;
ae.prev_log_idx = 0; ae.prev_log_idx = 0;
...@@ -866,12 +853,9 @@ void TestRaft_follower_recv_appendentries_add_new_entries_not_already_in_log( ...@@ -866,12 +853,9 @@ void TestRaft_follower_recv_appendentries_add_new_entries_not_already_in_log(
e[1].id = 2; e[1].id = 2;
ae.entries = e; ae.entries = e;
ae.n_entries = 2; ae.n_entries = 2;
raft_recv_appendentries(r, 1, &ae); raft_recv_appendentries(r, 1, &ae, &aer);
msg_appendentries_response_t *aer; CuAssertTrue(tc, 1 == aer.success);
aer = sender_poll_msg_data(sender);
CuAssertTrue(tc, NULL != aer);
CuAssertTrue(tc, 1 == aer->success);
CuAssertTrue(tc, 2 == raft_get_log_count(r)); CuAssertTrue(tc, 2 == raft_get_log_count(r));
} }
...@@ -883,7 +867,6 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx( ...@@ -883,7 +867,6 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx(
void *r; void *r;
void *sender; void *sender;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.send_appendentries_response = sender_appendentries_response,
.log = NULL .log = NULL
}; };
...@@ -900,6 +883,8 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx( ...@@ -900,6 +883,8 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx(
raft_set_callbacks(r, &funcs, sender); raft_set_callbacks(r, &funcs, sender);
msg_appendentries_t ae; msg_appendentries_t ae;
msg_appendentries_response_t aer;
memset(&ae, 0, sizeof(msg_appendentries_t)); memset(&ae, 0, sizeof(msg_appendentries_t));
ae.term = 1; ae.term = 1;
ae.prev_log_idx = 0; ae.prev_log_idx = 0;
...@@ -913,7 +898,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx( ...@@ -913,7 +898,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx(
e[3].id = 4; e[3].id = 4;
ae.entries = e; ae.entries = e;
ae.n_entries = 4; ae.n_entries = 4;
raft_recv_appendentries(r, 1, &ae); raft_recv_appendentries(r, 1, &ae, &aer);
/* receive an appendentry with commit */ /* receive an appendentry with commit */
memset(&ae, 0, sizeof(msg_appendentries_t)); memset(&ae, 0, sizeof(msg_appendentries_t));
...@@ -922,12 +907,9 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx( ...@@ -922,12 +907,9 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx(
ae.prev_log_idx = 4; ae.prev_log_idx = 4;
ae.leader_commit = 5; ae.leader_commit = 5;
/* receipt of appendentries changes commit idx */ /* receipt of appendentries changes commit idx */
raft_recv_appendentries(r, 1, &ae); raft_recv_appendentries(r, 1, &ae, &aer);
msg_appendentries_response_t *aer; CuAssertTrue(tc, 1 == aer.success);
aer = sender_poll_msg_data(sender);
CuAssertTrue(tc, NULL != aer);
CuAssertTrue(tc, 1 == aer->success);
/* set to 4 because commitIDX is lower */ /* set to 4 because commitIDX is lower */
CuAssertTrue(tc, 4 == raft_get_commit_idx(r)); CuAssertTrue(tc, 4 == raft_get_commit_idx(r));
} }
...@@ -938,7 +920,6 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit( ...@@ -938,7 +920,6 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit(
void *r; void *r;
void *sender; void *sender;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.send_appendentries_response = sender_appendentries_response,
.log = NULL .log = NULL
}; };
...@@ -955,6 +936,8 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit( ...@@ -955,6 +936,8 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit(
raft_set_callbacks(r, &funcs, sender); raft_set_callbacks(r, &funcs, sender);
msg_appendentries_t ae; msg_appendentries_t ae;
msg_appendentries_response_t aer;
memset(&ae, 0, sizeof(msg_appendentries_t)); memset(&ae, 0, sizeof(msg_appendentries_t));
ae.term = 1; ae.term = 1;
ae.prev_log_idx = 0; ae.prev_log_idx = 0;
...@@ -968,7 +951,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit( ...@@ -968,7 +951,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit(
e[3].id = 4; e[3].id = 4;
ae.entries = e; ae.entries = e;
ae.n_entries = 4; ae.n_entries = 4;
raft_recv_appendentries(r, 1, &ae); raft_recv_appendentries(r, 1, &ae, &aer);
/* receive an appendentry with commit */ /* receive an appendentry with commit */
memset(&ae, 0, sizeof(msg_appendentries_t)); memset(&ae, 0, sizeof(msg_appendentries_t));
...@@ -977,12 +960,9 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit( ...@@ -977,12 +960,9 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit(
ae.prev_log_idx = 3; ae.prev_log_idx = 3;
ae.leader_commit = 3; ae.leader_commit = 3;
/* receipt of appendentries changes commit idx */ /* receipt of appendentries changes commit idx */
raft_recv_appendentries(r, 1, &ae); raft_recv_appendentries(r, 1, &ae, &aer);
msg_appendentries_response_t *aer; CuAssertTrue(tc, 1 == aer.success);
aer = sender_poll_msg_data(sender);
CuAssertTrue(tc, NULL != aer);
CuAssertTrue(tc, 1 == aer->success);
/* set to 3 because leaderCommit is lower */ /* set to 3 because leaderCommit is lower */
CuAssertTrue(tc, 3 == raft_get_commit_idx(r)); CuAssertTrue(tc, 3 == raft_get_commit_idx(r));
} }
...@@ -1122,9 +1102,11 @@ void TestRaft_follower_receiving_appendentries_resets_election_timeout( ...@@ -1122,9 +1102,11 @@ void TestRaft_follower_receiving_appendentries_resets_election_timeout(
raft_periodic(r, 900); raft_periodic(r, 900);
msg_appendentries_t ae; msg_appendentries_t ae;
msg_appendentries_response_t aer;
memset(&ae, 0, sizeof(msg_appendentries_t)); memset(&ae, 0, sizeof(msg_appendentries_t));
ae.term = 1; ae.term = 1;
raft_recv_appendentries(r, 1, &ae); raft_recv_appendentries(r, 1, &ae, &aer);
CuAssertTrue(tc, 0 == raft_get_timeout_elapsed(r)); CuAssertTrue(tc, 0 == raft_get_timeout_elapsed(r));
} }
...@@ -1358,9 +1340,11 @@ void TestRaft_candidate_recv_appendentries_frm_leader_results_in_follower( ...@@ -1358,9 +1340,11 @@ void TestRaft_candidate_recv_appendentries_frm_leader_results_in_follower(
/* receive recent appendentries */ /* receive recent appendentries */
msg_appendentries_t ae; msg_appendentries_t ae;
msg_appendentries_response_t aer;
memset(&ae, 0, sizeof(msg_appendentries_t)); memset(&ae, 0, sizeof(msg_appendentries_t));
ae.term = 1; ae.term = 1;
raft_recv_appendentries(r, 1, &ae); raft_recv_appendentries(r, 1, &ae, &aer);
CuAssertTrue(tc, 1 == raft_is_follower(r)); CuAssertTrue(tc, 1 == raft_is_follower(r));
} }
...@@ -1383,6 +1367,7 @@ TestRaft_candidate_recv_appendentries_frm_invalid_leader_doesnt_result_in_follow ...@@ -1383,6 +1367,7 @@ TestRaft_candidate_recv_appendentries_frm_invalid_leader_doesnt_result_in_follow
}; };
msg_appendentries_t ae; msg_appendentries_t ae;
msg_appendentries_response_t aer;
sender = sender_new(NULL); sender = sender_new(NULL);
r = raft_new(); r = raft_new();
...@@ -1404,7 +1389,7 @@ TestRaft_candidate_recv_appendentries_frm_invalid_leader_doesnt_result_in_follow ...@@ -1404,7 +1389,7 @@ TestRaft_candidate_recv_appendentries_frm_invalid_leader_doesnt_result_in_follow
ae.prev_log_term = 1; ae.prev_log_term = 1;
/* appendentry from invalid leader doesn't make candidate become follower */ /* appendentry from invalid leader doesn't make candidate become follower */
raft_recv_appendentries(r, 1, &ae); raft_recv_appendentries(r, 1, &ae, &aer);
CuAssertTrue(tc, 1 == raft_is_candidate(r)); CuAssertTrue(tc, 1 == raft_is_candidate(r));
} }
...@@ -1690,7 +1675,6 @@ TestRaft_leader_increase_commit_idx_when_majority_have_entry_and_atleast_one_new ...@@ -1690,7 +1675,6 @@ TestRaft_leader_increase_commit_idx_when_majority_have_entry_and_atleast_one_new
void *sender; void *sender;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.send_appendentries = sender_appendentries, .send_appendentries = sender_appendentries,
.send_appendentries_response = sender_appendentries_response,
.log = NULL .log = NULL
}; };
msg_appendentries_response_t aer; msg_appendentries_response_t aer;
...@@ -1777,6 +1761,7 @@ void TestRaft_leader_steps_down_if_received_appendentries_is_newer_than_itself( ...@@ -1777,6 +1761,7 @@ void TestRaft_leader_steps_down_if_received_appendentries_is_newer_than_itself(
}; };
msg_appendentries_t ae; msg_appendentries_t ae;
msg_appendentries_response_t aer;
sender = sender_new(NULL); sender = sender_new(NULL);
r = raft_new(); r = raft_new();
...@@ -1791,7 +1776,7 @@ void TestRaft_leader_steps_down_if_received_appendentries_is_newer_than_itself( ...@@ -1791,7 +1776,7 @@ void TestRaft_leader_steps_down_if_received_appendentries_is_newer_than_itself(
ae.term = 5; ae.term = 5;
ae.prev_log_idx = 6; ae.prev_log_idx = 6;
ae.prev_log_term = 5; ae.prev_log_term = 5;
raft_recv_appendentries(r, 1, &ae); raft_recv_appendentries(r, 1, &ae, &aer);
CuAssertTrue(tc, 1 == raft_is_follower(r)); CuAssertTrue(tc, 1 == raft_is_follower(r));
} }
......
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