Commit 6a46f305 authored by Willem Thiart's avatar Willem Thiart
Browse files

Add term to msg_entry_t

parent 8405f29e
...@@ -46,6 +46,9 @@ typedef struct ...@@ -46,6 +46,9 @@ typedef struct
/* the entry's unique ID */ /* the entry's unique ID */
unsigned int id; unsigned int id;
/* the term the entry was created */
int term;
msg_entry_data_t data; msg_entry_data_t data;
} msg_entry_t; } msg_entry_t;
......
...@@ -313,7 +313,7 @@ int raft_recv_appendentries( ...@@ -313,7 +313,7 @@ int raft_recv_appendentries(
/* TODO: replace malloc with mempoll/arena */ /* TODO: replace malloc with mempoll/arena */
raft_entry_t* c = (raft_entry_t*)malloc(sizeof(raft_entry_t)); raft_entry_t* c = (raft_entry_t*)malloc(sizeof(raft_entry_t));
c->term = me->current_term; c->term = cmd->term;
c->len = cmd->data.len; c->len = cmd->data.len;
c->id = cmd->id; c->id = cmd->id;
c->data = (unsigned char*)malloc(cmd->data.len); c->data = (unsigned char*)malloc(cmd->data.len);
...@@ -509,17 +509,14 @@ void raft_send_appendentries(raft_server_t* me_, int node) ...@@ -509,17 +509,14 @@ void raft_send_appendentries(raft_server_t* me_, int node)
msg_entry_t mety; msg_entry_t mety;
if (1 < next_idx) if (0 < next_idx)
{ {
/* previous log is the log just before the new logs */ raft_entry_t* ety = raft_get_entry_from_idx(me_, next_idx);
raft_entry_t* ety = raft_get_entry_from_idx(me_, next_idx - 1);
if (ety) if (ety)
{ {
ae.prev_log_idx = next_idx - 1;
ae.prev_log_term = ety->term;
if (me->commit_idx < next_idx) if (me->commit_idx < next_idx)
{ {
mety.term = ety->term;
mety.id = ety->id; mety.id = ety->id;
mety.data.len = ety->len; mety.data.len = ety->len;
mety.data.buf = ety->data; mety.data.buf = ety->data;
...@@ -530,6 +527,15 @@ void raft_send_appendentries(raft_server_t* me_, int node) ...@@ -530,6 +527,15 @@ void raft_send_appendentries(raft_server_t* me_, int node)
} }
} }
/* previous log is the log just before the new logs */
if (1 < next_idx)
{
raft_entry_t* prev_ety = raft_get_entry_from_idx(me_, next_idx - 1);
ae.prev_log_idx = next_idx - 1;
if (prev_ety)
ae.prev_log_term = prev_ety->term;
}
__log(me_, "sending appendentries node: %d, %d %d %d %d", __log(me_, "sending appendentries node: %d, %d %d %d %d",
node, node,
ae.term, ae.term,
......
...@@ -85,6 +85,9 @@ int sender_requestvote_response(raft_server_t* raft, ...@@ -85,6 +85,9 @@ int sender_requestvote_response(raft_server_t* 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)
{ {
msg_entry_t* entries = calloc(1, sizeof(msg_entry_t) * msg->n_entries);
memcpy(entries, msg->entries, sizeof(msg_entry_t) * msg->n_entries);
msg->entries = entries;
return __append_msg(udata, msg, RAFT_MSG_APPENDENTRIES, sizeof(*msg), peer, return __append_msg(udata, msg, RAFT_MSG_APPENDENTRIES, sizeof(*msg), peer,
raft); raft);
} }
......
...@@ -654,7 +654,7 @@ void TestRaft_follower_recv_appendentries_increases_log(CuTest * tc) ...@@ -654,7 +654,7 @@ void TestRaft_follower_recv_appendentries_increases_log(CuTest * tc)
/* receive an appendentry with commit */ /* receive an appendentry with commit */
memset(&ae, 0, sizeof(msg_appendentries_t)); memset(&ae, 0, sizeof(msg_appendentries_t));
ae.term = 1; ae.term = 3;
ae.prev_log_term = 1; ae.prev_log_term = 1;
/* first appendentries msg */ /* first appendentries msg */
ae.prev_log_idx = 0; ae.prev_log_idx = 0;
...@@ -664,12 +664,16 @@ void TestRaft_follower_recv_appendentries_increases_log(CuTest * tc) ...@@ -664,12 +664,16 @@ void TestRaft_follower_recv_appendentries_increases_log(CuTest * tc)
ety.data.buf = str; ety.data.buf = str;
ety.data.len = 3; ety.data.len = 3;
ety.id = 1; ety.id = 1;
/* check that old terms are passed onto the log */
ety.term = 2;
ae.entries = &ety; ae.entries = &ety;
ae.n_entries = 1; ae.n_entries = 1;
raft_recv_appendentries(r, 1, &ae, &aer); raft_recv_appendentries(r, 1, &ae, &aer);
CuAssertTrue(tc, 1 == aer.success); CuAssertTrue(tc, 1 == aer.success);
CuAssertTrue(tc, 1 == raft_get_log_count(r)); CuAssertTrue(tc, 1 == raft_get_log_count(r));
raft_entry_t* log = raft_get_entry_from_idx(r, 1);
CuAssertTrue(tc, 2 == log->term);
} }
/* 5.3 */ /* 5.3 */
...@@ -830,17 +834,10 @@ void TestRaft_follower_recv_appendentries_add_new_entries_not_already_in_log( ...@@ -830,17 +834,10 @@ void TestRaft_follower_recv_appendentries_add_new_entries_not_already_in_log(
void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx( void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx(
CuTest * tc) CuTest * tc)
{ {
void *r; sender_new(NULL);
void *sender; void *r = raft_new();
raft_cbs_t funcs = {
.log = NULL
};
sender = sender_new(NULL);
r = raft_new();
raft_add_peer(r, (void*)1, 1); raft_add_peer(r, (void*)1, 1);
raft_add_peer(r, (void*)2, 0); raft_add_peer(r, (void*)2, 0);
raft_set_callbacks(r, &funcs, sender);
msg_appendentries_t ae; msg_appendentries_t ae;
msg_appendentries_response_t aer; msg_appendentries_response_t aer;
...@@ -852,9 +849,13 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx( ...@@ -852,9 +849,13 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx(
/* include entries */ /* include entries */
msg_entry_t e[4]; msg_entry_t e[4];
memset(&e, 0, sizeof(msg_entry_t) * 4); memset(&e, 0, sizeof(msg_entry_t) * 4);
e[0].term = 1;
e[0].id = 1; e[0].id = 1;
e[1].term = 1;
e[1].id = 2; e[1].id = 2;
e[2].term = 1;
e[2].id = 3; e[2].id = 3;
e[3].term = 1;
e[3].id = 4; e[3].id = 4;
ae.entries = e; ae.entries = e;
ae.n_entries = 4; ae.n_entries = 4;
...@@ -899,9 +900,13 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit( ...@@ -899,9 +900,13 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit(
/* include entries */ /* include entries */
msg_entry_t e[4]; msg_entry_t e[4];
memset(&e, 0, sizeof(msg_entry_t) * 4); memset(&e, 0, sizeof(msg_entry_t) * 4);
e[0].term = 1;
e[0].id = 1; e[0].id = 1;
e[1].term = 1;
e[1].id = 2; e[1].id = 2;
e[2].term = 1;
e[2].id = 3; e[2].id = 3;
e[3].term = 1;
e[3].id = 4; e[3].id = 4;
ae.entries = e; ae.entries = e;
ae.n_entries = 4; ae.n_entries = 4;
...@@ -1510,36 +1515,41 @@ void TestRaft_leader_sends_appendentries_with_prevLogIdx( ...@@ -1510,36 +1515,41 @@ void TestRaft_leader_sends_appendentries_with_prevLogIdx(
void *sender = sender_new(NULL); void *sender = sender_new(NULL);
void *r = raft_new(); void *r = raft_new();
raft_set_callbacks(r, &funcs, sender); raft_set_callbacks(r, &funcs, sender);
raft_add_peer(r, (void*)1, 1); raft_add_peer(r, (void*)1, 1); /* me */
raft_add_peer(r, (void*)2, 0); raft_add_peer(r, (void*)2, 0);
/* i'm leader */ /* i'm leader */
raft_set_state(r, RAFT_STATE_LEADER); raft_set_state(r, RAFT_STATE_LEADER);
/* receive appendentries messages */ /* receive appendentries messages */
raft_send_appendentries(r, 0); raft_send_appendentries(r, 1);
msg_appendentries_t* ae = sender_poll_msg_data(sender); msg_appendentries_t* ae = sender_poll_msg_data(sender);
CuAssertTrue(tc, NULL != ae); CuAssertTrue(tc, NULL != ae);
CuAssertTrue(tc, ae->prev_log_idx == 0); CuAssertTrue(tc, ae->prev_log_idx == 0);
/* add an entry */ raft_node_t* n = raft_get_node(r, 1);
/* add 1 entry */
/* receive appendentries messages */ /* receive appendentries messages */
raft_entry_t ety; raft_entry_t ety;
ety.term = 1; ety.term = 2;
ety.id = 100; ety.id = 100;
ety.len = 4; ety.len = 4;
ety.data = (unsigned char*)"aaa"; ety.data = (unsigned char*)"aaa";
raft_append_entry(r, &ety); raft_append_entry(r, &ety);
raft_send_appendentries(r, 0); raft_node_set_next_idx(n, 1);
raft_send_appendentries(r, 1);
ae = sender_poll_msg_data(sender); ae = sender_poll_msg_data(sender);
CuAssertTrue(tc, NULL != ae); CuAssertTrue(tc, NULL != ae);
CuAssertTrue(tc, ae->prev_log_idx == 0); CuAssertTrue(tc, ae->prev_log_idx == 0);
CuAssertTrue(tc, ae->n_entries == 1);
CuAssertTrue(tc, ae->entries[0].id == 100);
CuAssertTrue(tc, ae->entries[0].term == 2);
/* set next_idx */ /* set next_idx */
/* receive appendentries messages */ /* receive appendentries messages */
raft_node_t* n = raft_get_node(r, 0);
raft_node_set_next_idx(n, 2); raft_node_set_next_idx(n, 2);
raft_send_appendentries(r, 0); raft_send_appendentries(r, 1);
ae = sender_poll_msg_data(sender); ae = sender_poll_msg_data(sender);
CuAssertTrue(tc, NULL != ae); CuAssertTrue(tc, NULL != ae);
CuAssertTrue(tc, ae->prev_log_idx == 1); CuAssertTrue(tc, ae->prev_log_idx == 1);
......
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