fluencelabs/redis
1
0
Fork 0
mirror of https://github.com/fluencelabs/redis synced 2025-03-26 12:21:04 +00:00
Code Issues Projects Releases Wiki Activity

Cluster Manager: improved cleanup/error handling in various functions

Browse Source
This commit is contained in:
artix 2018-02-14 17:54:46 +01:00
parent 2d677e2bf3
commit 307d995f75
1 changed files with 56 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
src/redis-cli.c
View File

@ -2220,7 +2220,7 @@ static int clusterManagerAddSlots(clusterManagerNode *node, char**err)
{ {
redisReply *reply = NULL; redisReply *reply = NULL;
void *_reply = NULL; void *_reply = NULL;
int is_err = 0; int is_err = 0, success = 1;
int argc; int argc;
sds *argv = NULL; sds *argv = NULL;
size_t *argvlen = NULL; size_t *argvlen = NULL;
@ -2235,39 +2235,44 @@ static int clusterManagerAddSlots(clusterManagerNode *node, char**err)
added++; added++;
} }
} }
if (!added) goto node_cmd_err; if (!added) {
success = 0;
goto cleanup;
}
argv = cliSplitArgs(cmd, &argc); argv = cliSplitArgs(cmd, &argc);
if (argc == 0 || argv == NULL) goto node_cmd_err; if (argc == 0 || argv == NULL) {
success = 0;
goto cleanup;
}
argvlen = zmalloc(argc*sizeof(size_t)); argvlen = zmalloc(argc*sizeof(size_t));
for (i = 0; i < argc; i++) for (i = 0; i < argc; i++)
argvlen[i] = sdslen(argv[i]); argvlen[i] = sdslen(argv[i]);
redisAppendCommandArgv(node->context,argc,(const char**)argv,argvlen); redisAppendCommandArgv(node->context,argc,(const char**)argv,argvlen);
if (redisGetReply(node->context, &_reply) != REDIS_OK) goto node_cmd_err; if (redisGetReply(node->context, &_reply) != REDIS_OK) {
success = 1;
goto cleanup;
}
reply = (redisReply*) _reply; reply = (redisReply*) _reply;
if (reply == NULL || (is_err = (reply->type == REDIS_REPLY_ERROR))) { if (reply == NULL || (is_err = (reply->type == REDIS_REPLY_ERROR))) {
if (is_err && err != NULL) { if (is_err && err != NULL) {
*err = zmalloc((reply->len + 1) * sizeof(char)); *err = zmalloc((reply->len + 1) * sizeof(char));
strcpy(*err, reply->str); strcpy(*err, reply->str);
} }
goto node_cmd_err; success = 0;
goto cleanup;
} }
sdsfree(cmd); cleanup:
zfree(argvlen);
sdsfreesplitres(argv,argc);
freeReplyObject(reply);
return 1;
node_cmd_err:
sdsfree(cmd); sdsfree(cmd);
zfree(argvlen); zfree(argvlen);
if (argv != NULL) sdsfreesplitres(argv,argc); if (argv != NULL) sdsfreesplitres(argv,argc);
if (reply != NULL) freeReplyObject(reply); if (reply != NULL) freeReplyObject(reply);
return 0; return success;
} }
static int clusterManagerFlushNodeConfig(clusterManagerNode *node, char **err) { static int clusterManagerFlushNodeConfig(clusterManagerNode *node, char **err) {
if (!node->dirty) return 0; if (!node->dirty) return 0;
redisReply *reply = NULL; redisReply *reply = NULL;
int is_err = 0; int is_err = 0, success = 1;
*err = NULL; *err = NULL;
if (node->replicate != NULL) { if (node->replicate != NULL) {
reply = CLUSTER_MANAGER_COMMAND(node, "CLUSTER REPLICATE %s", reply = CLUSTER_MANAGER_COMMAND(node, "CLUSTER REPLICATE %s",
@ -2277,18 +2282,20 @@ static int clusterManagerFlushNodeConfig(clusterManagerNode *node, char **err) {
*err = zmalloc((reply->len + 1) * sizeof(char)); *err = zmalloc((reply->len + 1) * sizeof(char));
strcpy(*err, reply->str); strcpy(*err, reply->str);
} }
goto node_cmd_err; success = 0;
goto cleanup;
} }
} else { } else {
int added = clusterManagerAddSlots(node, err); int added = clusterManagerAddSlots(node, err);
if (!added || *err != NULL) goto node_cmd_err; if (!added || *err != NULL) {
success = 0;
goto cleanup;
}
} }
node->dirty = 0; node->dirty = 0;
freeReplyObject(reply); cleanup:
return 1; if (reply != NULL) freeReplyObject(reply);
node_cmd_err: return success;
freeReplyObject(reply);
return 0;
} }
static void clusterManagerWaitForClusterJoin(void) { static void clusterManagerWaitForClusterJoin(void) {
@ -2305,14 +2312,15 @@ static int clusterManagerNodeLoadInfo(clusterManagerNode *node, int opts,
char **err) char **err)
{ {
redisReply *reply = CLUSTER_MANAGER_COMMAND(node, "CLUSTER NODES"); redisReply *reply = CLUSTER_MANAGER_COMMAND(node, "CLUSTER NODES");
int is_err = 0; int is_err = 0, success = 1;
*err = NULL; *err = NULL;
if (reply == NULL || (is_err = (reply->type == REDIS_REPLY_ERROR))) { if (reply == NULL || (is_err = (reply->type == REDIS_REPLY_ERROR))) {
if (is_err && err != NULL) { if (is_err && err != NULL) {
*err = zmalloc((reply->len + 1) * sizeof(char)); *err = zmalloc((reply->len + 1) * sizeof(char));
strcpy(*err, reply->str); strcpy(*err, reply->str);
} }
goto node_cmd_err; success = 0;
goto cleanup;
} }
int getfriends = (opts & CLUSTER_MANAGER_OPT_GETFRIENDS); int getfriends = (opts & CLUSTER_MANAGER_OPT_GETFRIENDS);
char *lines = reply->str, *p, *line; char *lines = reply->str, *p, *line;
@ -2340,7 +2348,10 @@ static int clusterManagerNodeLoadInfo(clusterManagerNode *node, int opts,
} }
if (i == 8) break; // Slots if (i == 8) break; // Slots
} }
if (!flags) goto node_cmd_err; if (!flags) {
success = 0;
goto cleanup;
}
int myself = (strstr(flags, "myself") != NULL); int myself = (strstr(flags, "myself") != NULL);
clusterManagerNode *currentNode = NULL; clusterManagerNode *currentNode = NULL;
if (myself) { if (myself) {
@ -2406,14 +2417,16 @@ static int clusterManagerNodeLoadInfo(clusterManagerNode *node, int opts,
if (addr == NULL) { if (addr == NULL) {
// TODO: find a better err message // TODO: find a better err message
fprintf(stderr, "Error: invalid CLUSTER NODES reply\n"); fprintf(stderr, "Error: invalid CLUSTER NODES reply\n");
goto node_cmd_err; success = 0;
goto cleanup;
} }
char *c = strrchr(addr, '@'); char *c = strrchr(addr, '@');
if (c != NULL) *c = '\0'; if (c != NULL) *c = '\0';
c = strrchr(addr, ':'); c = strrchr(addr, ':');
if (c == NULL) { if (c == NULL) {
fprintf(stderr, "Error: invalid CLUSTER NODES reply\n"); fprintf(stderr, "Error: invalid CLUSTER NODES reply\n");
goto node_cmd_err; success = 0;
goto cleanup;
} }
*c = '\0'; *c = '\0';
int port = atoi(++c); int port = atoi(++c);
@ -2445,11 +2458,9 @@ static int clusterManagerNodeLoadInfo(clusterManagerNode *node, int opts,
if (ping_recv != NULL) currentNode->ping_recv = atoll(ping_recv); if (ping_recv != NULL) currentNode->ping_recv = atoll(ping_recv);
if (!getfriends && myself) break; if (!getfriends && myself) break;
} }
freeReplyObject(reply); cleanup:
return 1; if (reply) freeReplyObject(reply);
node_cmd_err: return success;
freeReplyObject(reply);
return 0;
} }
/* Retrieves info about the cluster using argument 'node' as the starting /* Retrieves info about the cluster using argument 'node' as the starting
@ -2780,7 +2791,7 @@ cluster_manager_err:
static int clusterManagerCommandCreate(int argc, char **argv) { static int clusterManagerCommandCreate(int argc, char **argv) {
printf("Cluster Manager: Creating Cluster\n"); printf("Cluster Manager: Creating Cluster\n");
int i, j; int i, j, success = 1;
cluster_manager.nodes = listCreate(); cluster_manager.nodes = listCreate();
for (i = 0; i < argc; i++) { for (i = 0; i < argc; i++) {
char *addr = argv[i]; char *addr = argv[i];
@ -2974,7 +2985,8 @@ assign_replicas:
CLUSTER_MANAGER_PRINT_REPLY_ERROR(node, err); CLUSTER_MANAGER_PRINT_REPLY_ERROR(node, err);
zfree(err); zfree(err);
} }
goto cmd_err; success = 0;
goto cleanup;
} else if (err != NULL) zfree(err); } else if (err != NULL) zfree(err);
} }
printf(">>> Nodes configuration updated\n"); printf(">>> Nodes configuration updated\n");
@ -3010,7 +3022,10 @@ assign_replicas:
is_err = 1; is_err = 1;
fprintf(stderr, "Failed to send CLUSTER MEET command.\n"); fprintf(stderr, "Failed to send CLUSTER MEET command.\n");
} }
if (is_err) goto cmd_err; if (is_err) {
success = 0;
goto cleanup;
}
} }
// Give one second for the join to start, in order to avoid that // Give one second for the join to start, in order to avoid that
// waiting for cluster join will find all the nodes agree about // waiting for cluster join will find all the nodes agree about
@ -3029,7 +3044,8 @@ assign_replicas:
CLUSTER_MANAGER_PRINT_REPLY_ERROR(node, err); CLUSTER_MANAGER_PRINT_REPLY_ERROR(node, err);
zfree(err); zfree(err);
} }
goto cmd_err; success = 0;
goto cleanup;
} }
} }
// Reset Nodes // Reset Nodes
@ -3041,9 +3057,13 @@ assign_replicas:
else freeClusterManagerNode(node); else freeClusterManagerNode(node);
} }
listEmpty(cluster_manager.nodes); listEmpty(cluster_manager.nodes);
if (!clusterManagerLoadInfoFromNode(first_node, 0)) goto cmd_err; //TODO: msg? if (!clusterManagerLoadInfoFromNode(first_node, 0)) {
success = 0;
goto cleanup; //TODO: msg?
}
clusterManagerCheckCluster(0); clusterManagerCheckCluster(0);
} }
cleanup:
/* Free everything */ /* Free everything */
zfree(masters); zfree(masters);
zfree(ips); zfree(ips);
@ -3052,16 +3072,7 @@ assign_replicas:
CLUSTER_MANAGER_NODEARRAY_FREE(node_array); CLUSTER_MANAGER_NODEARRAY_FREE(node_array);
} }
zfree(ip_nodes); zfree(ip_nodes);
return 1; return success;
cmd_err:
zfree(masters);
zfree(ips);
for (i = 0; i < node_len; i++) {
clusterManagerNodeArray *node_array = ip_nodes + i;
CLUSTER_MANAGER_NODEARRAY_FREE(node_array);
}
zfree(ip_nodes);
return 0;
} }
static int clusterManagerCommandInfo(int argc, char **argv) { static int clusterManagerCommandInfo(int argc, char **argv) {