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

Made EntryResponse easier

parent 164beb15
...@@ -155,20 +155,6 @@ typedef int ( ...@@ -155,20 +155,6 @@ typedef int (
msg_entry_t* msg msg_entry_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_entries_response_f
) (
raft_server_t* raft,
void *udata,
int node,
msg_entry_response_t* msg
);
#ifndef HAVE_FUNC_LOG #ifndef HAVE_FUNC_LOG
#define HAVE_FUNC_LOG #define HAVE_FUNC_LOG
typedef void ( typedef void (
...@@ -202,7 +188,6 @@ typedef struct ...@@ -202,7 +188,6 @@ typedef struct
func_send_requestvote_f send_requestvote; func_send_requestvote_f send_requestvote;
func_send_appendentries_f send_appendentries; func_send_appendentries_f send_appendentries;
func_send_entries_f send_entries; func_send_entries_f send_entries;
func_send_entries_response_f send_entries_response;
func_log_f log; func_log_f log;
func_applylog_f applylog; func_applylog_f applylog;
} raft_cbs_t; } raft_cbs_t;
...@@ -310,8 +295,10 @@ int raft_recv_requestvote_response(raft_server_t* me, int node, ...@@ -310,8 +295,10 @@ int raft_recv_requestvote_response(raft_server_t* me, int node,
* Send appendentries to followers * Send appendentries to followers
* This function should block if it needs to append the message. * This function should block if it needs to append the message.
* @param node Index of the node who sent us this message * @param node Index of the node who sent us this message
* @param[out] r The resulting response
* @param e The entry message */ * @param e The entry message */
int raft_recv_entry(raft_server_t* me, int node, msg_entry_t* e); int raft_recv_entry(raft_server_t* me, int node, msg_entry_t* e,
msg_entry_response_t *r);
/** /**
* @return the server's node ID */ * @return the server's node ID */
......
...@@ -403,22 +403,8 @@ int raft_recv_requestvote_response(raft_server_t* me_, int node, ...@@ -403,22 +403,8 @@ int raft_recv_requestvote_response(raft_server_t* me_, int node,
return 0; return 0;
} }
int raft_send_entry_response(raft_server_t* me_, int raft_recv_entry(raft_server_t* me_, int node, msg_entry_t* e,
int node, int etyid, int was_committed) msg_entry_response_t *r)
{
raft_server_private_t* me = (void*)me_;
msg_entry_response_t res;
__log(me_, "send entry response to: %d", node);
res.id = etyid;
res.was_committed = was_committed;
if (me->cb.send_entries_response)
me->cb.send_entries_response(me_, me->udata, node, &res);
return 0;
}
int raft_recv_entry(raft_server_t* me_, int node, msg_entry_t* e)
{ {
raft_server_private_t* me = (void*)me_; raft_server_private_t* me = (void*)me_;
raft_entry_t ety; raft_entry_t ety;
...@@ -431,13 +417,12 @@ int raft_recv_entry(raft_server_t* me_, int node, msg_entry_t* e) ...@@ -431,13 +417,12 @@ int raft_recv_entry(raft_server_t* me_, int node, msg_entry_t* e)
ety.data = e->data; ety.data = e->data;
ety.len = e->len; ety.len = e->len;
res = raft_append_entry(me_, &ety); res = raft_append_entry(me_, &ety);
raft_send_entry_response(me_, node, e->id, res);
for (i = 0; i < me->num_nodes; i++) for (i = 0; i < me->num_nodes; i++)
{ if (me->nodeid != i)
if (me->nodeid == i)
continue;
raft_send_appendentries(me_, i); raft_send_appendentries(me_, i);
}
r->id = e->id;
r->was_committed = res;
return 0; return 0;
} }
......
...@@ -171,8 +171,14 @@ void sender_poll_msgs(void* s) ...@@ -171,8 +171,14 @@ void sender_poll_msgs(void* s)
raft_recv_requestvote_response(me->raft, m->sender, m->data); raft_recv_requestvote_response(me->raft, m->sender, m->data);
break; break;
case RAFT_MSG_ENTRY: case RAFT_MSG_ENTRY:
raft_recv_entry(me->raft, m->sender, m->data); {
msg_entry_response_t response;
raft_recv_entry(me->raft, m->sender, m->data, &response);
__append_msg(me, &response, RAFT_MSG_ENTRY_RESPONSE,
sizeof(response), m->sender, me->raft);
}
break; break;
case RAFT_MSG_ENTRY_RESPONSE: case RAFT_MSG_ENTRY_RESPONSE:
#if 0 #if 0
raft_recv_entry_response(me->raft, m->sender, m->data); raft_recv_entry_response(me->raft, m->sender, m->data);
......
#include <stdbool.h> #include <stdbool.h>
#include <assert.h> #include <assert.h>
#include <stdlib.h> #include <stdlib.h>
...@@ -16,7 +15,7 @@ void TestRaft_node_set_nextIdx(CuTest * tc) ...@@ -16,7 +15,7 @@ void TestRaft_node_set_nextIdx(CuTest * tc)
raft_node_t *p; raft_node_t *p;
p = raft_node_new((void*)1); p = raft_node_new((void*)1);
raft_node_set_next_idx(p,3); raft_node_set_next_idx(p, 3);
CuAssertTrue(tc, 3 == raft_node_get_next_idx(p)); CuAssertTrue(tc, 3 == raft_node_get_next_idx(p));
} }
...@@ -37,8 +37,6 @@ void TestRaft_scenario_leader_appears(CuTest * tc) ...@@ -37,8 +37,6 @@ void TestRaft_scenario_leader_appears(CuTest * tc)
.send_requestvote = sender_requestvote, .send_requestvote = sender_requestvote,
.send_appendentries = sender_appendentries, .send_appendentries = sender_appendentries,
.send_entries = sender_entries, .send_entries = sender_entries,
.send_entries_response =
sender_entries_response,
.log = NULL .log = NULL
}), sender[j]); }), sender[j]);
} }
......
...@@ -1481,9 +1481,8 @@ void TestRaft_leader_when_it_becomes_a_leader_sends_empty_appendentries( ...@@ -1481,9 +1481,8 @@ void TestRaft_leader_when_it_becomes_a_leader_sends_empty_appendentries(
void TestRaft_leader_responds_to_entry_msg_when_entry_is_committed(CuTest * tc) void TestRaft_leader_responds_to_entry_msg_when_entry_is_committed(CuTest * tc)
{ {
void *r, *sender; void *r, *sender;
msg_entry_response_t *cr; msg_entry_response_t cr;
raft_cbs_t funcs = { raft_cbs_t funcs = {
.send_entries_response = sender_entries_response,
.log = NULL .log = NULL
}; };
...@@ -1510,15 +1509,14 @@ void TestRaft_leader_responds_to_entry_msg_when_entry_is_committed(CuTest * tc) ...@@ -1510,15 +1509,14 @@ void TestRaft_leader_responds_to_entry_msg_when_entry_is_committed(CuTest * tc)
ety.len = strlen("entry"); ety.len = strlen("entry");
/* receive entry */ /* receive entry */
raft_recv_entry(r, 1, &ety); raft_recv_entry(r, 1, &ety, &cr);
CuAssertTrue(tc, 1 == raft_get_log_count(r)); CuAssertTrue(tc, 1 == raft_get_log_count(r));
/* trigger response through commit */ /* trigger response through commit */
raft_apply_entry(r); raft_apply_entry(r);
/* leader sent response to entry message */ /* leader sent response to entry message */
cr = sender_poll_msg_data(sender); CuAssertTrue(tc, 1 == cr.was_committed);
CuAssertTrue(tc, NULL != cr);
} }
/* 5.3 */ /* 5.3 */
...@@ -1610,6 +1608,8 @@ void TestRaft_leader_append_entry_to_log_increases_idxno(CuTest * tc) ...@@ -1610,6 +1608,8 @@ void TestRaft_leader_append_entry_to_log_increases_idxno(CuTest * tc)
}; };
msg_entry_t ety; msg_entry_t ety;
msg_entry_response_t cr;
ety.id = 1; ety.id = 1;
ety.data = "entry"; ety.data = "entry";
ety.len = strlen("entry"); ety.len = strlen("entry");
...@@ -1619,7 +1619,7 @@ void TestRaft_leader_append_entry_to_log_increases_idxno(CuTest * tc) ...@@ -1619,7 +1619,7 @@ void TestRaft_leader_append_entry_to_log_increases_idxno(CuTest * tc)
raft_set_state(r, RAFT_STATE_LEADER); raft_set_state(r, RAFT_STATE_LEADER);
CuAssertTrue(tc, 0 == raft_get_log_count(r)); CuAssertTrue(tc, 0 == raft_get_log_count(r));
raft_recv_entry(r, 1, &ety); raft_recv_entry(r, 1, &ety, &cr);
CuAssertTrue(tc, 1 == raft_get_log_count(r)); CuAssertTrue(tc, 1 == raft_get_log_count(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