Commit 5c3aa05f authored by willem's avatar willem
Browse files

Better send() callback

parent 01102556
...@@ -10,10 +10,12 @@ How does it work? ...@@ -10,10 +10,12 @@ How does it work?
----------------- -----------------
See raft.h for full documentation. See raft.h for full documentation.
A CRaft user needs to implement all the functions in raft_cbs_t.
Networking is out of scope for this project. The implementor will need to do all the plumbing. *Currently*, this is done by: Networking is out of scope for this project. The implementor will need to do all the plumbing. *Currently*, this is done by:
- Implementing all the callbacks within raft_cbs_t; and - Implementing all the callbacks within raft_cbs_t; and
- Calling raft_recv_.* functions with msg_.* message structs - Calling raft_recv_* functions with msg_* message structs
Dependencies Dependencies
------------ ------------
......
...@@ -89,22 +89,33 @@ typedef struct { ...@@ -89,22 +89,33 @@ typedef struct {
int first_idx; int first_idx;
} msg_appendentries_response_t; } msg_appendentries_response_t;
enum { typedef void* raft_server_t;
typedef void* raft_node_t;
typedef enum {
RAFT_MSG_REQUESTVOTE, RAFT_MSG_REQUESTVOTE,
RAFT_MSG_REQUESTVOTE_RESPONSE, RAFT_MSG_REQUESTVOTE_RESPONSE,
RAFT_MSG_APPENDENTRIES, RAFT_MSG_APPENDENTRIES,
RAFT_MSG_APPENDENTRIES_RESPONSE, RAFT_MSG_APPENDENTRIES_RESPONSE,
RAFT_MSG_ENTRY, RAFT_MSG_ENTRY,
RAFT_MSG_ENTRY_RESPONSE, RAFT_MSG_ENTRY_RESPONSE,
}; } raft_message_type_e;
/**
* @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 ( typedef int (
*func_send_f *func_send_f
) ( ) (
void *cb_ctx, raft_server_t* raft,
void *udata, void *udata,
int node, int node,
int msg_type, raft_message_type_e msg_type,
const unsigned char *send_data, const unsigned char *send_data,
const int len const int len
); );
...@@ -114,19 +125,24 @@ typedef int ( ...@@ -114,19 +125,24 @@ typedef int (
typedef void ( typedef void (
*func_log_f *func_log_f
) ( ) (
void *cb_ctx, raft_server_t* raft,
void *src, void *udata,
const char *buf, const char *buf,
... ...
); );
#endif #endif
/** /**
* Apply this log to the state macine */ * Apply this log to the state machine
* @param raft The Raft server making this callback
* @param udata User data that is passed from Raft server
* @param data Data to be applied to the log
* @param len Length in bytes of data to be applied
* @return 0 on error */
typedef int ( typedef int (
*func_applylog_f *func_applylog_f
) ( ) (
void *cb_ctx, raft_server_t* raft,
void *udata, void *udata,
const unsigned char *data, const unsigned char *data,
const int len const int len
...@@ -138,8 +154,6 @@ typedef struct { ...@@ -138,8 +154,6 @@ typedef struct {
func_applylog_f applylog; func_applylog_f applylog;
} raft_cbs_t; } raft_cbs_t;
typedef void* raft_server_t;
typedef void* raft_node_t;
typedef struct { typedef struct {
/* entry's term */ /* entry's term */
...@@ -169,10 +183,12 @@ raft_server_t* raft_new(); ...@@ -169,10 +183,12 @@ raft_server_t* raft_new();
void raft_free(raft_server_t* me_); void raft_free(raft_server_t* me_);
/** /**
* Set callbacks * Set callbacks.
* Callbacks need to be set by the user for CRaft to work.
*
* @param funcs Callbacks * @param funcs Callbacks
* @param cb_ctx The context that we include when making a callback */ * @param udata The context that we include when making a callback */
void raft_set_callbacks(raft_server_t* me, raft_cbs_t* funcs, void* cb_ctx); void raft_set_callbacks(raft_server_t* me, raft_cbs_t* funcs, void* udata);
/** /**
* Set configuration * Set configuration
...@@ -195,7 +211,7 @@ void raft_set_request_timeout(raft_server_t* me_, int msec); ...@@ -195,7 +211,7 @@ void raft_set_request_timeout(raft_server_t* me_, int msec);
/** /**
* Process events that are dependent on time passing * Process events that are dependent on time passing
* @param msec_elapsed Time in milliseconds since the last raft_periodic() call * @param msec_elapsed Time in milliseconds since the last call
* @return 0 on error */ * @return 0 on error */
int raft_periodic(raft_server_t* me, int msec_elapsed); int raft_periodic(raft_server_t* me, int msec_elapsed);
......
...@@ -57,7 +57,7 @@ typedef struct { ...@@ -57,7 +57,7 @@ typedef struct {
/* callbacks */ /* callbacks */
raft_cbs_t cb; raft_cbs_t cb;
void* cb_ctx; void* udata;
/* my node ID */ /* my node ID */
int nodeid; int nodeid;
......
...@@ -55,12 +55,12 @@ raft_server_t* raft_new() ...@@ -55,12 +55,12 @@ raft_server_t* raft_new()
} }
void raft_set_callbacks(raft_server_t* me_, void raft_set_callbacks(raft_server_t* me_,
raft_cbs_t* funcs, void* cb_ctx) raft_cbs_t* funcs, void* udata)
{ {
raft_server_private_t* me = (void*)me_; raft_server_private_t* me = (void*)me_;
memcpy(&me->cb, funcs, sizeof(raft_cbs_t)); memcpy(&me->cb, funcs, sizeof(raft_cbs_t));
me->cb_ctx = cb_ctx; me->udata = udata;
} }
void raft_free(raft_server_t* me_) void raft_free(raft_server_t* me_)
...@@ -344,7 +344,7 @@ int raft_recv_appendentries( ...@@ -344,7 +344,7 @@ int raft_recv_appendentries(
done: done:
if (me->cb.send) if (me->cb.send)
me->cb.send(me->cb_ctx, me, node, RAFT_MSG_APPENDENTRIES_RESPONSE, me->cb.send(me_, me->udata, node, RAFT_MSG_APPENDENTRIES_RESPONSE,
(void*)&r, sizeof(msg_appendentries_response_t)); (void*)&r, sizeof(msg_appendentries_response_t));
return 1; return 1;
} }
...@@ -378,7 +378,7 @@ int raft_recv_requestvote(raft_server_t* me_, int node, msg_requestvote_t* vr) ...@@ -378,7 +378,7 @@ int raft_recv_requestvote(raft_server_t* me_, int node, msg_requestvote_t* vr)
r.term = raft_get_current_term(me_); r.term = raft_get_current_term(me_);
if (me->cb.send) if (me->cb.send)
me->cb.send(me->cb_ctx, me, node, RAFT_MSG_REQUESTVOTE_RESPONSE, me->cb.send(me_, me->udata, node, RAFT_MSG_REQUESTVOTE_RESPONSE,
(void*)&r, sizeof(msg_requestvote_response_t)); (void*)&r, sizeof(msg_requestvote_response_t));
return 0; return 0;
...@@ -434,7 +434,7 @@ int raft_send_entry_response(raft_server_t* me_, ...@@ -434,7 +434,7 @@ int raft_send_entry_response(raft_server_t* me_,
res.id = etyid; res.id = etyid;
res.was_committed = was_committed; res.was_committed = was_committed;
if (me->cb.send) if (me->cb.send)
me->cb.send(me->cb_ctx, me, node, RAFT_MSG_ENTRY_RESPONSE, me->cb.send(me_, me->udata, node, RAFT_MSG_ENTRY_RESPONSE,
(void*)&res, sizeof(msg_entry_response_t)); (void*)&res, sizeof(msg_entry_response_t));
return 0; return 0;
} }
...@@ -471,7 +471,7 @@ int raft_send_requestvote(raft_server_t* me_, int node) ...@@ -471,7 +471,7 @@ int raft_send_requestvote(raft_server_t* me_, int node)
rv.term = me->current_term; rv.term = me->current_term;
rv.last_log_idx = raft_get_current_idx(me_); rv.last_log_idx = raft_get_current_idx(me_);
if (me->cb.send) if (me->cb.send)
me->cb.send(me->cb_ctx, me, node, RAFT_MSG_REQUESTVOTE, me->cb.send(me_, me->udata, node, RAFT_MSG_REQUESTVOTE,
(void*)&rv, sizeof(msg_requestvote_t)); (void*)&rv, sizeof(msg_requestvote_t));
return 1; return 1;
} }
...@@ -502,7 +502,7 @@ int raft_apply_entry(raft_server_t* me_) ...@@ -502,7 +502,7 @@ int raft_apply_entry(raft_server_t* me_)
if (me->commit_idx < me->last_applied_idx) if (me->commit_idx < me->last_applied_idx)
me->commit_idx = me->last_applied_idx; me->commit_idx = me->last_applied_idx;
if (me->cb.applylog) if (me->cb.applylog)
me->cb.applylog(me->cb_ctx, me, e->data, e->len); me->cb.applylog(me_, me->udata, e->data, e->len);
return 1; return 1;
} }
...@@ -524,7 +524,7 @@ void raft_send_appendentries(raft_server_t* me_, int node) ...@@ -524,7 +524,7 @@ void raft_send_appendentries(raft_server_t* me_, int node)
// TODO: // TODO:
ae.prev_log_idx = 0; ae.prev_log_idx = 0;
ae.n_entries = 0; ae.n_entries = 0;
me->cb.send(me->cb_ctx, me, node, RAFT_MSG_APPENDENTRIES, me->cb.send(me_, me->udata, node, RAFT_MSG_APPENDENTRIES,
(void*)&ae, sizeof(msg_appendentries_t)); (void*)&ae, sizeof(msg_appendentries_t));
} }
......
...@@ -40,17 +40,21 @@ void senders_new() ...@@ -40,17 +40,21 @@ void senders_new()
__nsenders = 0; __nsenders = 0;
} }
int sender_send(void* caller, void* udata, int peer, int type, int sender_send(raft_server_t* raft,
const unsigned char* data, int len) void* udata,
int peer,
raft_message_type_e type,
const unsigned char* data,
int len)
{ {
sender_t* me = caller; sender_t* me = udata;
msg_t* m; msg_t* m;
m = malloc(sizeof(msg_t)); m = malloc(sizeof(msg_t));
m->type = type; m->type = type;
m->len = len; m->len = len;
m->data = malloc(len); m->data = malloc(len);
m->sender = raft_get_nodeid(udata); 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);
......
int sender_send(void* caller, void* udata, int peer, int type, int sender_send(raft_server_t* raft,
const unsigned char* data, const int len); void* udata,
int peer,
raft_message_type_e type,
const unsigned char* data,
const int len);
void* sender_new(void* address); void* sender_new(void* address);
......
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