Request the current cluster slots configuration by calling CLUSTER SLOTS * and atomically update the slots after a successful reply. */
| 1300 | /* Request the current cluster slots configuration by calling CLUSTER SLOTS |
| 1301 | * and atomically update the slots after a successful reply. */ |
| 1302 | static int fetchClusterSlotsConfiguration(client c) { |
| 1303 | UNUSED(c); |
| 1304 | int success = 1, is_fetching_slots = 0, last_update = 0; |
| 1305 | size_t i; |
| 1306 | atomicGet(config.slots_last_update, last_update); |
| 1307 | if (c->slots_last_update < last_update) { |
| 1308 | c->slots_last_update = last_update; |
| 1309 | return -1; |
| 1310 | } |
| 1311 | redisReply *reply = NULL; |
| 1312 | atomicGetIncr(config.is_fetching_slots, is_fetching_slots, 1); |
| 1313 | if (is_fetching_slots) return -1; //TODO: use other codes || errno ? |
| 1314 | atomicSet(config.is_fetching_slots, 1); |
| 1315 | printf("WARNING: Cluster slots configuration changed, fetching new one...\n"); |
| 1316 | const char *errmsg = "Failed to update cluster slots configuration"; |
| 1317 | static dictType dtype = { |
| 1318 | dictSdsHash, /* hash function */ |
| 1319 | NULL, /* key dup */ |
| 1320 | NULL, /* val dup */ |
| 1321 | dictSdsKeyCompare, /* key compare */ |
| 1322 | NULL, /* key destructor */ |
| 1323 | NULL, /* val destructor */ |
| 1324 | NULL /* allow to expand */ |
| 1325 | }; |
| 1326 | /* printf("[%d] fetchClusterSlotsConfiguration\n", c->thread_id); */ |
| 1327 | dict *masters = dictCreate(&dtype, NULL); |
| 1328 | redisContext *ctx = NULL; |
| 1329 | for (i = 0; i < (size_t) config.cluster_node_count; i++) { |
| 1330 | clusterNode *node = config.cluster_nodes[i]; |
| 1331 | assert(node->ip != NULL); |
| 1332 | assert(node->name != NULL); |
| 1333 | assert(node->port); |
| 1334 | /* Use first node as entry point to connect to. */ |
| 1335 | if (ctx == NULL) { |
| 1336 | ctx = getRedisContext(node->ip, node->port, NULL); |
| 1337 | if (!ctx) { |
| 1338 | success = 0; |
| 1339 | goto cleanup; |
| 1340 | } |
| 1341 | } |
| 1342 | if (node->updated_slots != NULL) |
| 1343 | zfree(node->updated_slots); |
| 1344 | node->updated_slots = NULL; |
| 1345 | node->updated_slots_count = 0; |
| 1346 | dictReplace(masters, node->name, node) ; |
| 1347 | } |
| 1348 | reply = (redisReply*)redisCommand(ctx, "CLUSTER SLOTS"); |
| 1349 | if (reply == NULL || reply->type == REDIS_REPLY_ERROR) { |
| 1350 | success = 0; |
| 1351 | if (reply) |
| 1352 | fprintf(stderr,"%s\nCLUSTER SLOTS ERROR: %s\n",errmsg,reply->str); |
| 1353 | goto cleanup; |
| 1354 | } |
| 1355 | assert(reply->type == REDIS_REPLY_ARRAY); |
| 1356 | for (i = 0; i < reply->elements; i++) { |
| 1357 | redisReply *r = reply->element[i]; |
| 1358 | assert(r->type == REDIS_REPLY_ARRAY); |
| 1359 | assert(r->elements >= 3); |
no test coverage detected