Commit 72f03664 authored by willem's avatar willem
Browse files

Multiple send functions

parent 1d8c95e1
......@@ -105,23 +105,100 @@ typedef enum {
RAFT_MSG_ENTRY_RESPONSE,
} raft_message_type_e;
typedef int (
*func_send_f
) (
raft_server_t* raft,
void *udata,
int node,
raft_message_type_e msg_type,
const unsigned char *send_data,
const int len
);
/**
* @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
* @param msg_type ID of the message type
* @param send_data Data to be sent
* @param len Length in bytes of data to be sent
* @return 0 on error */
typedef int (
*func_send_f
*func_send_requestvote_f
) (
raft_server_t* raft,
void *udata,
int node,
raft_message_type_e msg_type,
const unsigned char *send_data,
const int len
msg_requestvote_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_requestvote_response_f
) (
raft_server_t* raft,
void *udata,
int node,
msg_requestvote_response_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_f
) (
raft_server_t* raft,
void *udata,
int node,
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 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_f
) (
raft_server_t* raft,
void *udata,
int node,
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
......@@ -153,12 +230,16 @@ typedef int (
);
typedef struct {
func_send_f send;
func_send_requestvote_f send_requestvote;
func_send_requestvote_response_f send_requestvote_response;
func_send_appendentries_f send_appendentries;
func_send_appendentries_response_f send_appendentries_response;
func_send_entries_f send_entries;
func_send_entries_response_f send_entries_response;
func_log_f log;
func_applylog_f applylog;
} raft_cbs_t;
typedef struct {
/* entry's term */
unsigned int term;
......
......@@ -343,9 +343,8 @@ int raft_recv_appendentries(
r.first_idx = ae->prev_log_idx + 1;
done:
if (me->cb.send)
me->cb.send(me_, me->udata, node, RAFT_MSG_APPENDENTRIES_RESPONSE,
(void*)&r, sizeof(msg_appendentries_response_t));
if (me->cb.send_appendentries_response)
me->cb.send_appendentries_response(me_, me->udata, node, &r);
return 1;
}
......@@ -377,9 +376,8 @@ int raft_recv_requestvote(raft_server_t* me_, int node, msg_requestvote_t* vr)
node, r.vote_granted == 1 ? "granted" : "not granted");
r.term = raft_get_current_term(me_);
if (me->cb.send)
me->cb.send(me_, me->udata, node, RAFT_MSG_REQUESTVOTE_RESPONSE,
(void*)&r, sizeof(msg_requestvote_response_t));
if (me->cb.send_requestvote_response)
me->cb.send_requestvote_response(me_, me->udata, node, &r);
return 0;
}
......@@ -433,9 +431,8 @@ int raft_send_entry_response(raft_server_t* me_,
res.id = etyid;
res.was_committed = was_committed;
if (me->cb.send)
me->cb.send(me_, me->udata, node, RAFT_MSG_ENTRY_RESPONSE,
(void*)&res, sizeof(msg_entry_response_t));
if (me->cb.send_entries_response)
me->cb.send_entries_response(me_, me->udata, node, &res);
return 0;
}
......@@ -470,9 +467,8 @@ int raft_send_requestvote(raft_server_t* me_, int node)
rv.term = me->current_term;
rv.last_log_idx = raft_get_current_idx(me_);
if (me->cb.send)
me->cb.send(me_, me->udata, node, RAFT_MSG_REQUESTVOTE,
(void*)&rv, sizeof(msg_requestvote_t));
if (me->cb.send_requestvote)
me->cb.send_requestvote(me_, me->udata, node, &rv);
return 1;
}
......@@ -512,7 +508,7 @@ void raft_send_appendentries(raft_server_t* me_, int node)
__log(me_, NULL, "sending appendentries to: %d", node);
if (!(me->cb.send))
if (!(me->cb.send_appendentries))
return;
msg_appendentries_t ae;
......@@ -524,8 +520,8 @@ void raft_send_appendentries(raft_server_t* me_, int node)
// TODO:
ae.prev_log_idx = 0;
ae.n_entries = 0;
me->cb.send(me_, me->udata, node, RAFT_MSG_APPENDENTRIES,
(void*)&ae, sizeof(msg_appendentries_t));
if (me->cb.send_appendentries)
me->cb.send_appendentries(me_, me->udata, node, &ae);
}
void raft_send_appendentries_all(raft_server_t* me_)
......
......@@ -40,6 +40,8 @@ void senders_new()
__nsenders = 0;
}
#if 0
int sender_send(raft_server_t* raft,
void* udata,
int peer,
......@@ -65,6 +67,69 @@ int sender_send(raft_server_t* raft,
return 0;
}
#endif
int __append_msg(
sender_t* me,
void* data,
int type,
int len,
int peer,
raft_server_t* raft
)
{
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 1;
}
int sender_requestvote(raft_server_t* raft,
void* udata, int peer, msg_requestvote_t* msg)
{
return __append_msg(udata,msg,RAFT_MSG_REQUESTVOTE,sizeof(*msg),peer,raft);
}
int sender_requestvote_response(raft_server_t* raft,
void* udata, int peer, msg_requestvote_response_t* msg)
{
return __append_msg(udata,msg,RAFT_MSG_REQUESTVOTE_RESPONSE,sizeof(*msg),peer,raft);
}
int sender_appendentries(raft_server_t* raft,
void* udata, int peer, msg_appendentries_t* msg)
{
return __append_msg(udata,msg,RAFT_MSG_APPENDENTRIES,sizeof(*msg),peer,raft);
}
int sender_appendentries_response(raft_server_t* raft,
void* udata, int peer, msg_appendentries_response_t* msg)
{
return __append_msg(udata,msg,RAFT_MSG_APPENDENTRIES_RESPONSE,sizeof(*msg),peer,raft);
}
int sender_entries(raft_server_t* raft,
void* udata, int peer, msg_entry_t* msg)
{
return __append_msg(udata,msg,RAFT_MSG_ENTRY,sizeof(*msg),peer,raft);
}
int sender_entries_response(raft_server_t* raft,
void* udata, int peer, msg_entry_response_t* msg)
{
return __append_msg(udata,msg,RAFT_MSG_ENTRY_RESPONSE,sizeof(*msg),peer,raft);
}
void* sender_new(void* address)
{
......
int sender_send(raft_server_t* raft,
void* udata,
int peer,
raft_message_type_e type,
const unsigned char* data,
const int len);
void* sender_new(void* address);
void* sender_poll_msg_data(void* s);
int sender_msgs_available(void* s);
int sender_requestvote(raft_server_t* raft,
void* udata, int peer, msg_requestvote_t* msg);
int sender_requestvote_response(raft_server_t* raft,
void* udata, int peer, msg_requestvote_response_t* msg);
int sender_appendentries(raft_server_t* raft,
void* udata, int peer, msg_appendentries_t* msg);
int sender_appendentries_response(raft_server_t* raft,
void* udata, int peer, msg_appendentries_response_t* msg);
int sender_entries(raft_server_t* raft,
void* udata, int peer, msg_entry_t* msg);
int sender_entries_response(raft_server_t* raft,
void* udata, int peer, msg_entry_response_t* msg);
......@@ -33,7 +33,12 @@ void TestRaft_scenario_leader_appears(CuTest * tc)
raft_set_election_timeout(r[j], 500);
raft_set_configuration(r[j],cfg,j);
raft_set_callbacks(r[j],&((raft_cbs_t) {
.send = sender_send,
.send_requestvote = sender_requestvote,
.send_requestvote_response = sender_requestvote_response,
.send_appendentries = sender_appendentries,
.send_appendentries_response = sender_appendentries_response,
.send_entries = sender_entries,
.send_entries_response = sender_entries_response,
.log = NULL
}), sender[j]);
}
......
......@@ -423,7 +423,7 @@ void TestRaft_server_recv_requestvote_reply_false_if_term_less_than_current_term
void *r;
void *sender;
raft_cbs_t funcs = {
.send = sender_send,
.send_requestvote_response = sender_requestvote_response,
.log = NULL
};
msg_requestvote_t rv;
......@@ -461,7 +461,7 @@ void TestRaft_server_dont_grant_vote_if_we_didnt_vote_for_this_candidate(
void *sender;
void *msg;
raft_cbs_t funcs = {
.send = sender_send,
.send_requestvote_response = sender_requestvote_response,
.log = NULL
};
msg_requestvote_t rv;
......@@ -520,7 +520,7 @@ void TestRaft_follower_recv_appendentries_reply_false_if_term_less_than_currentt
void *r;
void *sender;
raft_cbs_t funcs = {
.send = sender_send,
.send_appendentries_response = sender_appendentries_response,
.log = NULL
};
msg_appendentries_t ae;
......@@ -561,7 +561,7 @@ void TestRaft_follower_recv_appendentries_updates_currentterm_if_term_gt_current
msg_appendentries_response_t *aer;
raft_cbs_t funcs = {
.send = sender_send,
.send_appendentries_response = sender_appendentries_response,
.log = NULL
};
......@@ -636,7 +636,7 @@ void TestRaft_follower_increases_log_after_appendentry(CuTest * tc)
char *str = "aaa";
raft_cbs_t funcs = {
.send = sender_send,
.send_appendentries_response = sender_appendentries_response,
.log = NULL
};
......@@ -687,7 +687,7 @@ void TestRaft_follower_recv_appendentries_reply_false_if_doesnt_have_log_at_prev
msg_entry_t ety;
char *str = "aaa";
raft_cbs_t funcs = {
.send = sender_send,
.send_appendentries_response = sender_appendentries_response,
.log = NULL
};
/* 2 nodes */
......@@ -743,7 +743,7 @@ void TestRaft_follower_recv_appendentries_delete_entries_if_conflict_with_new_en
raft_entry_t *ety_appended;
raft_cbs_t funcs = {
.send = sender_send,
.send_appendentries_response = sender_appendentries_response,
.log = NULL
};
......@@ -812,7 +812,7 @@ void TestRaft_follower_recv_appendentries_add_new_entries_not_already_in_log(CuT
void *r;
void *sender;
raft_cbs_t funcs = {
.send = sender_send,
.send_appendentries_response = sender_appendentries_response,
.log = NULL
};
......@@ -857,7 +857,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_prevLogIdx(CuTest * t
void *r;
void *sender;
raft_cbs_t funcs = {
.send = sender_send,
.send_appendentries_response = sender_appendentries_response,
.log = NULL
};
......@@ -910,7 +910,7 @@ void TestRaft_follower_recv_appendentries_set_commitidx_to_LeaderCommit(CuTest *
void *r;
void *sender;
raft_cbs_t funcs = {
.send = sender_send,
.send_appendentries_response = sender_appendentries_response,
.log = NULL
};
......@@ -988,7 +988,7 @@ void TestRaft_follower_dont_grant_vote_if_candidate_has_a_less_complete_log(CuTe
void *r;
void *sender;
raft_cbs_t funcs = {
.send = sender_send,
.send_requestvote_response = sender_requestvote_response,
.log = NULL
};
......@@ -1100,7 +1100,7 @@ void TestRaft_follower_becoming_candidate_requests_votes_from_other_servers(CuTe
void *r;
void *sender;
raft_cbs_t funcs = {
.send = sender_send,
.send_requestvote = sender_requestvote,
.log = NULL
};
raft_node_configuration_t cfg[] = {
......@@ -1139,7 +1139,6 @@ void TestRaft_candidate_election_timeout_and_no_leader_results_in_new_election(C
void *r;
void *sender;
raft_cbs_t funcs = {
.send = sender_send,
.log = NULL
};
......@@ -1181,7 +1180,6 @@ void TestRaft_candidate_receives_majority_of_votes_becomes_leader(CuTest * tc)
void *r;
void *sender;
raft_cbs_t funcs = {
.send = sender_send,
.log = NULL
};
......@@ -1230,7 +1228,7 @@ void TestRaft_candidate_will_not_respond_to_voterequest_if_it_has_already_voted(
void *sender;
void *msg;
raft_cbs_t funcs = {
.send = sender_send,
.send_requestvote_response = sender_requestvote_response,
.log = NULL
};
......@@ -1264,7 +1262,7 @@ void TestRaft_candidate_requestvote_includes_logidx(CuTest * tc)
void *r;
void *sender;
raft_cbs_t funcs = {
.send = sender_send,
.send_requestvote = sender_requestvote,
.log = NULL
};
msg_requestvote_t* rv;
......@@ -1298,7 +1296,6 @@ void TestRaft_candidate_recv_appendentries_frm_leader_results_in_follower(CuTest
void *sender;
void *msg;
raft_cbs_t funcs = {
.send = sender_send,
.log = NULL
};
......@@ -1331,7 +1328,6 @@ void TestRaft_candidate_recv_appendentries_frm_invalid_leader_doesnt_result_in_f
void *r;
void *sender;
raft_cbs_t funcs = {
.send = sender_send,
.log = NULL
};
......@@ -1393,7 +1389,6 @@ void TestRaft_leader_when_becomes_leader_all_nodes_have_nextidx_equal_to_lastlog
void *r;
void *sender;
raft_cbs_t funcs = {
.send = sender_send,
.log = NULL
};
......@@ -1431,7 +1426,7 @@ void TestRaft_leader_when_it_becomes_a_leader_sends_empty_appendentries(CuTest *
void *r;
void *sender;
raft_cbs_t funcs = {
.send = sender_send,
.send_appendentries = sender_appendentries,
.log = NULL
};
......@@ -1466,7 +1461,7 @@ void TestRaft_leader_responds_to_entry_msg_when_entry_is_committed(CuTest * tc)
void *r, *sender;
msg_entry_response_t *cr;
raft_cbs_t funcs = {
.send = sender_send,
.send_entries_response = sender_entries_response,
.log = NULL
};
......@@ -1509,7 +1504,7 @@ void TestRaft_leader_sends_appendentries_with_NextIdx_when_PrevIdx_gt_NextIdx(Cu
void *r;
void *sender;
raft_cbs_t funcs = {
.send = sender_send,
.send_appendentries = sender_appendentries,
.log = NULL
};
......@@ -1519,7 +1514,6 @@ void TestRaft_leader_sends_appendentries_with_NextIdx_when_PrevIdx_gt_NextIdx(Cu
{(-1),(void*)2},
{(-1),NULL}};
msg_appendentries_t* ae;
sender = sender_new(NULL);
......@@ -1546,7 +1540,7 @@ void TestRaft_leader_retries_appendentries_with_decremented_NextIdx_log_inconsis
void *r;
void *sender;
raft_cbs_t funcs = {
.send = sender_send,
.send_appendentries = sender_appendentries,
.log = NULL
};
......@@ -1556,7 +1550,6 @@ void TestRaft_leader_retries_appendentries_with_decremented_NextIdx_log_inconsis
{(-1),(void*)2},
{(-1),NULL}};
msg_appendentries_t* ae;
sender = sender_new(NULL);
......@@ -1587,7 +1580,6 @@ void TestRaft_leader_append_entry_to_log_increases_idxno(CuTest * tc)
{(-1),(void*)2},
{(-1),NULL}};
msg_entry_t ety;
ety.id = 1;
ety.data = "entry";
......@@ -1638,7 +1630,8 @@ void TestRaft_leader_increase_commit_idx_when_majority_have_entry_and_atleast_on
void *r;
void *sender;
raft_cbs_t funcs = {
.send = sender_send,
.send_appendentries = sender_appendentries,
.send_appendentries_response = sender_appendentries_response,
.log = NULL
};
msg_appendentries_response_t aer;
......@@ -1712,7 +1705,6 @@ void TestRaft_leader_steps_down_if_received_appendentries_is_newer_than_itself(C
void *r;
void *sender;
raft_cbs_t funcs = {
.send = sender_send,
.log = NULL
};
......
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