Commit 6bf3f5f1 authored by Li Wei's avatar Li Wei Committed by Willem Thiart
Browse files

DAOS-333 server: Handle memory allocation errors



Check and handle memory allocation errors by reporting them back to
users via NULL or RAFT_ERR_NOMEM return values.
Signed-off-by: default avatarLi Wei <wei.g.li@intel.com>
parent 920d574a
...@@ -13,6 +13,7 @@ ...@@ -13,6 +13,7 @@
#define RAFT_ERR_NOT_LEADER -2 #define RAFT_ERR_NOT_LEADER -2
#define RAFT_ERR_ONE_VOTING_CHANGE_ONLY -3 #define RAFT_ERR_ONE_VOTING_CHANGE_ONLY -3
#define RAFT_ERR_SHUTDOWN -4 #define RAFT_ERR_SHUTDOWN -4
#define RAFT_ERR_NOMEM -5
#define RAFT_REQUESTVOTE_ERR_GRANTED 1 #define RAFT_REQUESTVOTE_ERR_GRANTED 1
#define RAFT_REQUESTVOTE_ERR_NOT_GRANTED 0 #define RAFT_REQUESTVOTE_ERR_NOT_GRANTED 0
...@@ -478,6 +479,7 @@ int raft_recv_requestvote_response(raft_server_t* me, ...@@ -478,6 +479,7 @@ int raft_recv_requestvote_response(raft_server_t* me,
* RAFT_ERR_NOT_LEADER server is not the leader; * RAFT_ERR_NOT_LEADER server is not the leader;
* RAFT_ERR_SHUTDOWN server should be shutdown; * RAFT_ERR_SHUTDOWN server should be shutdown;
* RAFT_ERR_ONE_VOTING_CHANGE_ONLY there is a non-voting change inflight; * RAFT_ERR_ONE_VOTING_CHANGE_ONLY there is a non-voting change inflight;
* RAFT_ERR_NOMEM memory allocation failure
*/ */
int raft_recv_entry(raft_server_t* me, int raft_recv_entry(raft_server_t* me,
msg_entry_t* ety, msg_entry_t* ety,
...@@ -622,7 +624,8 @@ void raft_set_commit_idx(raft_server_t* me, int commit_idx); ...@@ -622,7 +624,8 @@ void raft_set_commit_idx(raft_server_t* me, int commit_idx);
* @param[in] ety The entry to be appended * @param[in] ety The entry to be appended
* @return * @return
* 0 on success; * 0 on success;
* RAFT_ERR_SHUTDOWN server should shutdown */ * RAFT_ERR_SHUTDOWN server should shutdown
* RAFT_ERR_NOMEM memory allocation failure */
int raft_append_entry(raft_server_t* me, raft_entry_t* ety); int raft_append_entry(raft_server_t* me, raft_entry_t* ety);
/** Confirm if a msg_entry_response has been committed. /** Confirm if a msg_entry_response has been committed.
......
...@@ -107,6 +107,8 @@ int raft_get_state(raft_server_t* me_); ...@@ -107,6 +107,8 @@ int raft_get_state(raft_server_t* me_);
raft_node_t* raft_node_new(void* udata, int id); raft_node_t* raft_node_new(void* udata, int id);
void raft_node_free(raft_node_t* me_);
void raft_node_set_next_idx(raft_node_t* node, int nextIdx); void raft_node_set_next_idx(raft_node_t* node, int nextIdx);
void raft_node_set_match_idx(raft_node_t* node, int matchIdx); void raft_node_set_match_idx(raft_node_t* node, int matchIdx);
......
...@@ -41,15 +41,17 @@ typedef struct ...@@ -41,15 +41,17 @@ typedef struct
void* raft; void* raft;
} log_private_t; } log_private_t;
static void __ensurecapacity(log_private_t * me) static int __ensurecapacity(log_private_t * me)
{ {
int i, j; int i, j;
raft_entry_t *temp; raft_entry_t *temp;
if (me->count < me->size) if (me->count < me->size)
return; return 0;
temp = (raft_entry_t*)calloc(1, sizeof(raft_entry_t) * me->size * 2); temp = (raft_entry_t*)calloc(1, sizeof(raft_entry_t) * me->size * 2);
if (!temp)
return RAFT_ERR_NOMEM;
for (i = 0, j = me->front; i < me->count; i++, j++) for (i = 0, j = me->front; i < me->count; i++, j++)
{ {
...@@ -65,6 +67,7 @@ static void __ensurecapacity(log_private_t * me) ...@@ -65,6 +67,7 @@ static void __ensurecapacity(log_private_t * me)
me->entries = temp; me->entries = temp;
me->front = 0; me->front = 0;
me->back = me->count; me->back = me->count;
return 0;
} }
log_t* log_new() log_t* log_new()
...@@ -76,6 +79,10 @@ log_t* log_new() ...@@ -76,6 +79,10 @@ log_t* log_new()
me->count = 0; me->count = 0;
me->back = in(me)->front = 0; me->back = in(me)->front = 0;
me->entries = (raft_entry_t*)calloc(1, sizeof(raft_entry_t) * me->size); me->entries = (raft_entry_t*)calloc(1, sizeof(raft_entry_t) * me->size);
if (!me->entries) {
free(me);
return NULL;
}
return (log_t*)me; return (log_t*)me;
} }
...@@ -99,9 +106,11 @@ void log_clear(log_t* me_) ...@@ -99,9 +106,11 @@ void log_clear(log_t* me_)
int log_append_entry(log_t* me_, raft_entry_t* c) int log_append_entry(log_t* me_, raft_entry_t* c)
{ {
log_private_t* me = (log_private_t*)me_; log_private_t* me = (log_private_t*)me_;
int e = 0; int e;
__ensurecapacity(me); e = __ensurecapacity(me);
if (e != 0)
return e;
memcpy(&me->entries[me->back], c, sizeof(raft_entry_t)); memcpy(&me->entries[me->back], c, sizeof(raft_entry_t));
......
...@@ -46,6 +46,11 @@ raft_node_t* raft_node_new(void* udata, int id) ...@@ -46,6 +46,11 @@ raft_node_t* raft_node_new(void* udata, int id)
return (raft_node_t*)me; return (raft_node_t*)me;
} }
void raft_node_free(raft_node_t* me_)
{
free(me_);
}
int raft_node_get_next_idx(raft_node_t* me_) int raft_node_get_next_idx(raft_node_t* me_)
{ {
raft_node_private_t* me = (raft_node_private_t*)me_; raft_node_private_t* me = (raft_node_private_t*)me_;
......
...@@ -53,6 +53,10 @@ raft_server_t* raft_new() ...@@ -53,6 +53,10 @@ raft_server_t* raft_new()
me->request_timeout = 200; me->request_timeout = 200;
me->election_timeout = 1000; me->election_timeout = 1000;
me->log = log_new(); me->log = log_new();
if (!me->log) {
free(me);
return NULL;
}
me->voting_cfg_change_log_idx = -1; me->voting_cfg_change_log_idx = -1;
raft_set_state((raft_server_t*)me, RAFT_STATE_FOLLOWER); raft_set_state((raft_server_t*)me, RAFT_STATE_FOLLOWER);
me->current_leader = NULL; me->current_leader = NULL;
...@@ -316,6 +320,7 @@ int raft_recv_appendentries( ...@@ -316,6 +320,7 @@ int raft_recv_appendentries(
) )
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
int e = 0;
me->timeout_elapsed = 0; me->timeout_elapsed = 0;
...@@ -352,9 +357,9 @@ int raft_recv_appendentries( ...@@ -352,9 +357,9 @@ int raft_recv_appendentries(
/* NOTE: the log starts at 1 */ /* NOTE: the log starts at 1 */
if (0 < ae->prev_log_idx) if (0 < ae->prev_log_idx)
{ {
raft_entry_t* e = raft_get_entry_from_idx(me_, ae->prev_log_idx); raft_entry_t* ety = raft_get_entry_from_idx(me_, ae->prev_log_idx);
if (!e) if (!ety)
{ {
__log(me_, node, "AE no log at prev_idx %d", ae->prev_log_idx); __log(me_, node, "AE no log at prev_idx %d", ae->prev_log_idx);
goto fail_with_current_idx; goto fail_with_current_idx;
...@@ -365,10 +370,10 @@ int raft_recv_appendentries( ...@@ -365,10 +370,10 @@ int raft_recv_appendentries(
if (raft_get_current_idx(me_) < ae->prev_log_idx) if (raft_get_current_idx(me_) < ae->prev_log_idx)
goto fail_with_current_idx; goto fail_with_current_idx;
if (e->term != ae->prev_log_term) if (ety->term != ae->prev_log_term)
{ {
__log(me_, node, "AE term doesn't match prev_term (ie. %d vs %d) ci:%d pli:%d", __log(me_, node, "AE term doesn't match prev_term (ie. %d vs %d) ci:%d pli:%d",
e->term, ae->prev_log_term, raft_get_current_idx(me_), ae->prev_log_idx); ety->term, ae->prev_log_term, raft_get_current_idx(me_), ae->prev_log_idx);
/* Delete all the following log entries because they don't match */ /* Delete all the following log entries because they don't match */
raft_delete_entry_from_idx(me_, ae->prev_log_idx); raft_delete_entry_from_idx(me_, ae->prev_log_idx);
r->current_idx = ae->prev_log_idx - 1; r->current_idx = ae->prev_log_idx - 1;
...@@ -410,15 +415,17 @@ int raft_recv_appendentries( ...@@ -410,15 +415,17 @@ int raft_recv_appendentries(
/* Pick up remainder in case of mismatch or missing entry */ /* Pick up remainder in case of mismatch or missing entry */
for (; i < ae->n_entries; i++) for (; i < ae->n_entries; i++)
{ {
int e = raft_append_entry(me_, &ae->entries[i]); e = raft_append_entry(me_, &ae->entries[i]);
if (-1 == e) if (RAFT_ERR_SHUTDOWN == e)
goto fail_with_current_idx;
else if (RAFT_ERR_SHUTDOWN == e)
{ {
r->success = 0; r->success = 0;
r->first_idx = 0; r->first_idx = 0;
return RAFT_ERR_SHUTDOWN; return RAFT_ERR_SHUTDOWN;
} }
else if (0 != e)
{
goto fail_with_current_idx;
}
r->current_idx = ae->prev_log_idx + 1 + i; r->current_idx = ae->prev_log_idx + 1 + i;
} }
...@@ -442,7 +449,7 @@ fail_with_current_idx: ...@@ -442,7 +449,7 @@ fail_with_current_idx:
fail: fail:
r->success = 0; r->success = 0;
r->first_idx = 0; r->first_idx = 0;
return -1; return e;
} }
int raft_already_voted(raft_server_t* me_) int raft_already_voted(raft_server_t* me_)
...@@ -475,11 +482,11 @@ static int __should_grant_vote(raft_server_private_t* me, msg_requestvote_t* vr) ...@@ -475,11 +482,11 @@ static int __should_grant_vote(raft_server_private_t* me, msg_requestvote_t* vr)
if (0 == current_idx) if (0 == current_idx)
return 1; return 1;
raft_entry_t* e = raft_get_entry_from_idx((void*)me, current_idx); raft_entry_t* ety = raft_get_entry_from_idx((void*)me, current_idx);
if (e->term < vr->last_log_term) if (ety->term < vr->last_log_term)
return 1; return 1;
if (vr->last_log_term == e->term && current_idx <= vr->last_log_idx) if (vr->last_log_term == ety->term && current_idx <= vr->last_log_idx)
return 1; return 1;
return 0; return 0;
...@@ -610,14 +617,14 @@ int raft_recv_requestvote_response(raft_server_t* me_, ...@@ -610,14 +617,14 @@ int raft_recv_requestvote_response(raft_server_t* me_,
} }
int raft_recv_entry(raft_server_t* me_, int raft_recv_entry(raft_server_t* me_,
msg_entry_t* e, msg_entry_t* ety,
msg_entry_response_t *r) msg_entry_response_t *r)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
int i; int i;
/* Only one voting cfg change at a time */ /* Only one voting cfg change at a time */
if (raft_entry_is_voting_cfg_change(e)) if (raft_entry_is_voting_cfg_change(ety))
if (raft_voting_change_is_in_progress(me_)) if (raft_voting_change_is_in_progress(me_))
return RAFT_ERR_ONE_VOTING_CHANGE_ONLY; return RAFT_ERR_ONE_VOTING_CHANGE_ONLY;
...@@ -625,10 +632,12 @@ int raft_recv_entry(raft_server_t* me_, ...@@ -625,10 +632,12 @@ int raft_recv_entry(raft_server_t* me_,
return RAFT_ERR_NOT_LEADER; return RAFT_ERR_NOT_LEADER;
__log(me_, NULL, "received entry t:%d id: %d idx: %d", __log(me_, NULL, "received entry t:%d id: %d idx: %d",
me->current_term, e->id, raft_get_current_idx(me_) + 1); me->current_term, ety->id, raft_get_current_idx(me_) + 1);
e->term = me->current_term; ety->term = me->current_term;
raft_append_entry(me_, e); int e = raft_append_entry(me_, ety);
if (0 != e)
return e;
for (i = 0; i < me->num_nodes; i++) for (i = 0; i < me->num_nodes; i++)
{ {
if (me->node == me->nodes[i] || !me->nodes[i] || if (me->node == me->nodes[i] || !me->nodes[i] ||
...@@ -647,11 +656,11 @@ int raft_recv_entry(raft_server_t* me_, ...@@ -647,11 +656,11 @@ int raft_recv_entry(raft_server_t* me_,
if (1 == raft_get_num_voting_nodes(me_)) if (1 == raft_get_num_voting_nodes(me_))
raft_set_commit_idx(me_, raft_get_current_idx(me_)); raft_set_commit_idx(me_, raft_get_current_idx(me_));
r->id = e->id; r->id = ety->id;
r->idx = raft_get_current_idx(me_); r->idx = raft_get_current_idx(me_);
r->term = me->current_term; r->term = me->current_term;
if (raft_entry_is_voting_cfg_change(e)) if (raft_entry_is_voting_cfg_change(ety))
me->voting_cfg_change_log_idx = raft_get_current_idx(me_); me->voting_cfg_change_log_idx = raft_get_current_idx(me_);
return 0; return 0;
...@@ -805,10 +814,18 @@ raft_node_t* raft_add_node(raft_server_t* me_, void* udata, int id, int is_self) ...@@ -805,10 +814,18 @@ raft_node_t* raft_add_node(raft_server_t* me_, void* udata, int id, int is_self)
return NULL; return NULL;
} }
node = raft_node_new(udata, id);
if (!node)
return NULL;
me->num_nodes++; me->num_nodes++;
me->nodes = (raft_node_t*)realloc(me->nodes, sizeof(void*) * me->num_nodes); void* p = realloc(me->nodes, sizeof(void*) * me->num_nodes);
me->nodes[me->num_nodes - 1] = raft_node_new(udata, id); if (!p) {
assert(me->nodes[me->num_nodes - 1]); me->num_nodes--;
raft_node_free(node);
return NULL;
}
me->nodes = p;
me->nodes[me->num_nodes - 1] = node;
if (is_self) if (is_self)
me->node = me->nodes[me->num_nodes - 1]; me->node = me->nodes[me->num_nodes - 1];
...@@ -832,29 +849,20 @@ void raft_remove_node(raft_server_t* me_, raft_node_t* node) ...@@ -832,29 +849,20 @@ void raft_remove_node(raft_server_t* me_, raft_node_t* node)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
raft_node_t* new_array, *new_nodes;
new_array = (raft_node_t*)calloc((me->num_nodes - 1), sizeof(void*));
new_nodes = new_array;
int i, found = 0; int i, found = 0;
for (i = 0; i<me->num_nodes; i++) for (i = 0; i<me->num_nodes; i++)
{ {
if (me->nodes[i] == node) if (me->nodes[i] == node)
{ {
found = 1; found = 1;
continue; break;
} }
*new_nodes = me->nodes[i];
new_nodes++;
} }
assert(found); assert(found);
memmove(&me->nodes[i], &me->nodes[i + 1], sizeof(*me->nodes) * (me->num_nodes - i - 1));
me->num_nodes--; me->num_nodes--;
free(me->nodes);
me->nodes = new_array;
free(node); raft_node_free(node);
} }
int raft_get_nvotes_for_me(raft_server_t* me_) int raft_get_nvotes_for_me(raft_server_t* me_)
......
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