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