Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add multi-database support to cluster mode #1671

Open
wants to merge 7 commits into
base: unstable
Choose a base branch
from
+660 −79
Open

Add multi-database support to cluster mode #1671

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
b5c933f
Add multi-database support to cluster mode
xbasel Jan 23, 2025
4d45a7e
Reformatting
xbasel Feb 5, 2025
a20c149
Typo
xbasel Feb 5, 2025
1c3ea2f
fix tests
xbasel Feb 10, 2025
0cf9b2d
fix test
xbasel Feb 10, 2025
9fff9f1
Correctly determine the DB of the key in getNodeByQuery
soloestoy Feb 13, 2025
0e29ea2
CLUSTER RESET should ensure primary node has no keys
soloestoy Feb 13, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 13 additions & 5 deletions src/cluster.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -814,8 +814,16 @@ static int shouldReturnTlsInfo(void) {
}
}

unsigned int countKeysInSlotForDb(unsigned int hashslot, serverDb *db) {
return kvstoreHashtableSize(db->keys, hashslot);
}

unsigned int countKeysInSlot(unsigned int slot) {
return kvstoreHashtableSize(server.db->keys, slot);
unsigned int result = 0;
for (int i = 0; i < server.dbnum; i++) {
result += countKeysInSlotForDb(slot, server.db + i);
}
return result;
}

void clusterCommandHelp(client *c) {
Expand Down Expand Up @@ -897,7 +905,7 @@ void clusterCommand(client *c) {
addReplyError(c, "Invalid slot");
return;
}
addReplyLongLong(c, countKeysInSlot(slot));
addReplyLongLong(c, countKeysInSlotForDb(slot, c->db));
} else if (!strcasecmp(c->argv[1]->ptr, "getkeysinslot") && c->argc == 4) {
/* CLUSTER GETKEYSINSLOT <slot> <count> */
long long maxkeys, slot;
Expand All @@ -909,11 +917,11 @@ void clusterCommand(client *c) {
return;
}

unsigned int keys_in_slot = countKeysInSlot(slot);
unsigned int keys_in_slot = countKeysInSlotForDb(slot, c->db);
unsigned int numkeys = maxkeys > keys_in_slot ? keys_in_slot : maxkeys;
addReplyArrayLen(c, numkeys);
kvstoreHashtableIterator *kvs_di = NULL;
kvs_di = kvstoreGetHashtableIterator(server.db->keys, slot, 0);
kvs_di = kvstoreGetHashtableIterator(c->db->keys, slot, 0);
for (unsigned int i = 0; i < numkeys; i++) {
void *next;
serverAssert(kvstoreHashtableIteratorNext(kvs_di, &next));
Expand Down Expand Up @@ -1102,7 +1110,7 @@ getNodeByQuery(client *c, struct serverCommand *cmd, robj **argv, int argc, int
* NODE <node-id>. */
int flags = LOOKUP_NOTOUCH | LOOKUP_NOSTATS | LOOKUP_NONOTIFY | LOOKUP_NOEXPIRE;
if ((migrating_slot || importing_slot) && !pubsubshard_included) {
if (lookupKeyReadWithFlags(&server.db[0], thiskey, flags) == NULL)
if (lookupKeyReadWithFlags(c->db, thiskey, flags) == NULL)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Here, I modified it to use c->db, so for most commands, the key it wants to access can be correctly located. However, some cross-DB commands, such as COPY, still require additional checks. The ultimate solution is atomic-slot-migration I believe. Once ATM is implemented, the TRYAGAIN issue will no longer occur.

missing_keys++;
else
existing_keys++;
Expand Down
1 change: 1 addition & 0 deletions src/cluster.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,7 @@ int detectAndUpdateCachedNodeHealth(void);
client *createCachedResponseClient(int resp);
void deleteCachedResponseClient(client *recording_client);
void clearCachedClusterSlotsResponse(void);
unsigned int countKeysInSlotForDb(unsigned int hashslot, serverDb *db);
unsigned int countKeysInSlot(unsigned int hashslot);
int getSlotOrReply(client *c, robj *o);

Expand Down
66 changes: 36 additions & 30 deletions src/cluster_legacy.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5794,11 +5794,6 @@ int verifyClusterConfigWithData(void) {
* completely depend on the replication stream. */
if (nodeIsReplica(myself)) return C_OK;

/* Make sure we only have keys in DB0. */
for (j = 1; j < server.dbnum; j++) {
if (kvstoreSize(server.db[j].keys)) return C_ERR;
}

/* Check that all the slots we see populated memory have a corresponding
* entry in the cluster table. Otherwise fix the table. */
for (j = 0; j < CLUSTER_SLOTS; j++) {
Expand Down Expand Up @@ -6430,29 +6425,31 @@ unsigned int delKeysInSlot(unsigned int hashslot) {
server.server_del_keys_in_slot = 1;
unsigned int j = 0;

kvstoreHashtableIterator *kvs_di = NULL;
void *next;
kvs_di = kvstoreGetHashtableIterator(server.db->keys, hashslot, HASHTABLE_ITER_SAFE);
while (kvstoreHashtableIteratorNext(kvs_di, &next)) {
robj *valkey = next;
enterExecutionUnit(1, 0);
sds sdskey = objectGetKey(valkey);
robj *key = createStringObject(sdskey, sdslen(sdskey));
dbDelete(&server.db[0], key);
propagateDeletion(&server.db[0], key, server.lazyfree_lazy_server_del);
signalModifiedKey(NULL, &server.db[0], key);
/* The keys are not actually logically deleted from the database, just moved to another node.
* The modules needs to know that these keys are no longer available locally, so just send the
* keyspace notification to the modules, but not to clients. */
moduleNotifyKeyspaceEvent(NOTIFY_GENERIC, "del", key, server.db[0].id);
exitExecutionUnit();
postExecutionUnitOperations();
decrRefCount(key);
j++;
server.dirty++;
for (int i = 0; i < server.dbnum; i++) {
kvstoreHashtableIterator *kvs_di = NULL;
void *next;
serverDb db = server.db[i];
kvs_di = kvstoreGetHashtableIterator(db.keys, hashslot, HASHTABLE_ITER_SAFE);
while (kvstoreHashtableIteratorNext(kvs_di, &next)) {
robj *valkey = next;
enterExecutionUnit(1, 0);
sds sdskey = objectGetKey(valkey);
robj *key = createStringObject(sdskey, sdslen(sdskey));
dbDelete(&db, key);
propagateDeletion(&db, key, server.lazyfree_lazy_server_del);
signalModifiedKey(NULL, &db, key);
/* The keys are not actually logically deleted from the database, just moved to another node.
* The modules needs to know that these keys are no longer available locally, so just send the
* keyspace notification to the modules, but not to clients. */
moduleNotifyKeyspaceEvent(NOTIFY_GENERIC, "del", key, db.id);
exitExecutionUnit();
postExecutionUnitOperations();
decrRefCount(key);
j++;
server.dirty++;
}
kvstoreReleaseHashtableIterator(kvs_di);
}
kvstoreReleaseHashtableIterator(kvs_di);

server.server_del_keys_in_slot = 0;
serverAssert(server.execution_nesting == 0);
return j;
Expand Down Expand Up @@ -6881,6 +6878,15 @@ void clusterCommandSetSlot(client *c) {
addReply(c, shared.ok);
}

int dbHasNoKeys(void) {
for (int i = 0; i < server.dbnum; i++) {
if (kvstoreSize(server.db[i].keys) != 0) {
return 0;
}
}
return 1;
}

int clusterCommandSpecial(client *c) {
if (!strcasecmp(c->argv[1]->ptr, "meet") && (c->argc == 4 || c->argc == 5)) {
/* CLUSTER MEET <ip> <port> [cport] */
Expand Down Expand Up @@ -6912,7 +6918,7 @@ int clusterCommandSpecial(client *c) {
}
} else if (!strcasecmp(c->argv[1]->ptr, "flushslots") && c->argc == 2) {
/* CLUSTER FLUSHSLOTS */
if (kvstoreSize(server.db[0].keys) != 0) {
if (!dbHasNoKeys()) {
addReplyError(c, "DB must be empty to perform CLUSTER FLUSHSLOTS.");
return 1;
}
Expand Down Expand Up @@ -7053,7 +7059,7 @@ int clusterCommandSpecial(client *c) {
/* If the instance is currently a primary, it should have no assigned
* slots nor keys to accept to replicate some other node.
* Replicas can switch to another primary without issues. */
if (clusterNodeIsPrimary(myself) && (myself->numslots != 0 || kvstoreSize(server.db[0].keys) != 0)) {
if (clusterNodeIsPrimary(myself) && (myself->numslots != 0 || !dbHasNoKeys())) {
addReplyError(c, "To set a master the node must be empty and "
"without assigned slots.");
return 1;
Expand Down Expand Up @@ -7187,7 +7193,7 @@ int clusterCommandSpecial(client *c) {

/* Replicas can be reset while containing data, but not primary nodes
* that must be empty. */
if (clusterNodeIsPrimary(myself) && kvstoreSize(c->db->keys) != 0) {
if (clusterNodeIsPrimary(myself) && !dbHasNoKeys()) {
addReplyError(c, "CLUSTER RESET can't be called with "
"master nodes containing keys");
return 1;
Expand Down
7 changes: 0 additions & 7 deletions src/config.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -607,13 +607,6 @@ void loadServerConfigFromString(char *config) {
goto loaderr;
}

/* in case cluster mode is enabled dbnum must be 1 */
if (server.cluster_enabled && server.dbnum > 1) {
serverLog(LL_WARNING, "WARNING: Changing databases number from %d to 1 since we are in cluster mode",
server.dbnum);
server.dbnum = 1;
}

/* To ensure backward compatibility and work while hz is out of range */
if (server.config_hz < CONFIG_MIN_HZ) server.config_hz = CONFIG_MIN_HZ;
if (server.config_hz > CONFIG_MAX_HZ) server.config_hz = CONFIG_MAX_HZ;
Expand Down
20 changes: 0 additions & 20 deletions src/db.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -860,10 +860,6 @@ void selectCommand(client *c) {

if (getIntFromObjectOrReply(c, c->argv[1], &id, NULL) != C_OK) return;

if (server.cluster_enabled && id != 0) {
addReplyError(c, "SELECT is not allowed in cluster mode");
return;
}
if (selectDb(c, id) == C_ERR) {
addReplyError(c, "DB index is out of range");
} else {
Expand Down Expand Up @@ -1429,11 +1425,6 @@ void moveCommand(client *c) {
int srcid, dbid;
long long expire;

if (server.cluster_enabled) {
addReplyError(c, "MOVE is not allowed in cluster mode");
return;
}

/* Obtain source and target DB pointers */
src = c->db;
srcid = c->db->id;
Expand Down Expand Up @@ -1518,11 +1509,6 @@ void copyCommand(client *c) {
}
}

if ((server.cluster_enabled == 1) && (srcid != 0 || dbid != 0)) {
addReplyError(c, "Copying to another database is not allowed in cluster mode");
return;
}

/* If the user select the same DB as
* the source DB and using newkey as the same key
* it is probably an error. */
Expand Down Expand Up @@ -1728,12 +1714,6 @@ void swapMainDbWithTempDb(serverDb *tempDb) {
void swapdbCommand(client *c) {
int id1, id2;

/* Not allowed in cluster mode: we have just DB 0 there. */
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would that be enough for swapdb to work in cluster mode? What will happen in setup with 2 shards, each responsible for half of slots in db's?

Copy link
Member Author

@xbasel xbasel Feb 11, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

With this implementation SWAPDB must be executed in all primary nodes. There are three options:

  1. Allow SWAPDB and shift responsibility to the user – Risky, non-atomic, can cause temporary inconsistency and data corruption. Needs strong warnings.
  2. Keep SWAPDB disabled in cluster mode – Safest, avoids inconsistency.
  3. Make SWAPDB cluster-wide and atomic or – Complex, unclear feasibility.

I think option 2 is the safest bet. @JoBeR007 wdyt?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is SWAPDB replicated as a single command? Then it's atomic.

If it's risky, it's risky in standslone mode with replicas too, right?

I think we can allow it. Swapping the data can only be done in some non-realtime workloads anyway I think.

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think risky because of replication and risky because of the need to execute SWAPDB on all primary nodes are unrelated just because as a user you can't control first, but user is the main risk in the second case.
I would keep SWAPDB disabled in cluster mode, if we decide to continue with this implementation

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In cluster mode, consistency is per slot.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is SWAPDB replicated as a single command? Then it's atomic.

If it's risky, it's risky in standslone mode with replicas too, right?

I think we can allow it. Swapping the data can only be done in some non-realtime workloads anyway I think.

I don’t think it’s very risky with standalone replicas. The only downside is if SWAPDB propagation to the replica takes time, a client might still access the wrong database. At least the client won’t be able to modify the wrong database, as they can only read.
In cluster mode, the same (logical) DB can be DB0 on one node and DB1 on another, but similar issues already exist today, FLUSHDB on one node doesn’t clear the entire DB since data exists in other slots/nodes. But as you said, consistency is per slot.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, FLUSHDB is very similar in this regard. If a failover happens just before this command has been propagated to replicas, it's a big thing, but it's no surprise I think. The client can use WAIT or check replication offset to make sure the FLUSHDB or SWAPDB was successful on the replicas.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Regarding this, I think it is not just an issue of Multi-database but is more related to atomic slot migration. If a shard is in a stable state (not undergoing slot migration), then flushdb/flushall/swapdb are safe. However, if slot migration is in progress, it might lead to data inconsistency.

I think this needs to be considered alongside atomic-slot-migration:

  1. During the ATM process, for slots being migrated, if we encounter flushall/flushdb, we can send a command like flushslot or flushslotall to the target shard
  2. As for swapdb, I recommend temporarily prohibiting execution during the ATM process

@PingXie @enjoy-binbin , please also take note of this.

if (server.cluster_enabled) {
addReplyError(c, "SWAPDB is not allowed in cluster mode");
return;
}

/* Get the two DBs indexes. */
if (getIntFromObjectOrReply(c, c->argv[1], &id1, "invalid first DB index") != C_OK) return;

Expand Down
2 changes: 1 addition & 1 deletion src/valkey-benchmark.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -707,7 +707,7 @@ static client createClient(char *cmd, size_t len, client from, int thread_id) {
* buffer with the SELECT command, that will be discarded the first
* time the replies are received, so if the client is reused the
* SELECT command will not be used again. */
if (config.conn_info.input_dbnum != 0 && !is_cluster_client) {
if (config.conn_info.input_dbnum) {
c->obuf = sdscatprintf(c->obuf, "*2\r\n$6\r\nSELECT\r\n$%d\r\n%s\r\n", (int)sdslen(config.input_dbnumstr),
config.input_dbnumstr);
c->prefix_pending++;
Expand Down
53 changes: 39 additions & 14 deletions src/valkey-cli.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1553,11 +1553,11 @@ static int cliAuth(redisContext *ctx, char *user, char *auth) {
}

/* Send SELECT input_dbnum to the server */
static int cliSelect(void) {
static int cliSelect(struct config *config, redisContext *ctx) {
redisReply *reply;
if (config.conn_info.input_dbnum == config.dbnum) return REDIS_OK;
if (config->conn_info.input_dbnum == config->dbnum) return REDIS_OK;

reply = redisCommand(context, "SELECT %d", config.conn_info.input_dbnum);
reply = redisCommand(ctx, "SELECT %d", config->conn_info.input_dbnum);
if (reply == NULL) {
fprintf(stderr, "\nI/O error\n");
return REDIS_ERR;
Expand All @@ -1566,9 +1566,9 @@ static int cliSelect(void) {
int result = REDIS_OK;
if (reply->type == REDIS_REPLY_ERROR) {
result = REDIS_ERR;
fprintf(stderr, "SELECT %d failed: %s\n", config.conn_info.input_dbnum, reply->str);
fprintf(stderr, "SELECT %d failed: %s\n", config->conn_info.input_dbnum, reply->str);
} else {
config.dbnum = config.conn_info.input_dbnum;
config->dbnum = config->conn_info.input_dbnum;
cliRefreshPrompt();
}
freeReplyObject(reply);
Expand Down Expand Up @@ -1667,7 +1667,7 @@ static int cliConnect(int flags) {

/* Do AUTH, select the right DB, switch to RESP3 if needed. */
if (cliAuth(context, config.conn_info.user, config.conn_info.auth) != REDIS_OK) return REDIS_ERR;
if (cliSelect() != REDIS_OK) return REDIS_ERR;
if (cliSelect(&config, context) != REDIS_OK) return REDIS_ERR;
if (cliSwitchProto() != REDIS_OK) return REDIS_ERR;
}

Expand Down Expand Up @@ -2448,7 +2448,7 @@ static int cliSendCommand(int argc, char **argv, long repeat) {
config.conn_info.input_dbnum = config.dbnum = atoi(argv[1]);
cliRefreshPrompt();
} else if (!strcasecmp(command, "auth") && (argc == 2 || argc == 3)) {
cliSelect();
cliSelect(&config, context);
} else if (!strcasecmp(command, "multi") && argc == 1 && config.last_cmd_type != REDIS_REPLY_ERROR) {
config.in_multi = 1;
config.pre_multi_dbnum = config.dbnum;
Expand Down Expand Up @@ -4789,8 +4789,10 @@ static redisReply *clusterManagerMigrateKeysInReply(clusterManagerNode *source,
argv_len = zcalloc(argc * sizeof(size_t));
char portstr[255];
char timeoutstr[255];
char dbnum[255];
snprintf(portstr, 10, "%d", target->port);
snprintf(timeoutstr, 10, "%d", timeout);
snprintf(dbnum, 10, "%d", config.dbnum);
argv[0] = "MIGRATE";
argv_len[0] = 7;
argv[1] = target->ip;
Expand All @@ -4799,8 +4801,8 @@ static redisReply *clusterManagerMigrateKeysInReply(clusterManagerNode *source,
argv_len[2] = strlen(portstr);
argv[3] = "";
argv_len[3] = 0;
argv[4] = "0";
argv_len[4] = 1;
argv[4] = dbnum;
argv_len[4] = strlen(dbnum);
argv[5] = timeoutstr;
argv_len[5] = strlen(timeoutstr);
if (replace) {
Expand Down Expand Up @@ -4852,6 +4854,8 @@ static redisReply *clusterManagerMigrateKeysInReply(clusterManagerNode *source,
return migrate_reply;
}

static int getDatabases(redisContext *ctx);

/* Migrate all keys in the given slot from source to target.*/
static int clusterManagerMigrateKeysInSlot(clusterManagerNode *source,
clusterManagerNode *target,
Expand All @@ -4863,15 +4867,29 @@ static int clusterManagerMigrateKeysInSlot(clusterManagerNode *source,
int success = 1;
int do_fix = config.cluster_manager_command.flags & CLUSTER_MANAGER_CMD_FLAG_FIX;
int do_replace = config.cluster_manager_command.flags & CLUSTER_MANAGER_CMD_FLAG_REPLACE;

int dbnum = getDatabases(source->context);
int orig_db = config.conn_info.input_dbnum;
config.conn_info.input_dbnum = 0;

while (1) {
if (config.conn_info.input_dbnum == dbnum) {
break;
}
if (cliSelect(&config, source->context) == REDIS_ERR) {
success = 0;
goto next;
}
char *dots = NULL;
redisReply *reply = NULL, *migrate_reply = NULL;
reply = CLUSTER_MANAGER_COMMAND(source,
"CLUSTER "
"GETKEYSINSLOT %d %d",
slot, pipeline);
success = (reply != NULL);
if (!success) return 0;
if (!success) {
goto next;
}
if (reply->type == REDIS_REPLY_ERROR) {
success = 0;
if (err != NULL) {
Expand All @@ -4885,7 +4903,9 @@ static int clusterManagerMigrateKeysInSlot(clusterManagerNode *source,
size_t count = reply->elements;
if (count == 0) {
freeReplyObject(reply);
break;
reply = NULL;
config.conn_info.input_dbnum++;
continue;
}
if (verbose) dots = zmalloc((count + 1) * sizeof(char));
/* Calling MIGRATE command. */
Expand Down Expand Up @@ -5009,8 +5029,13 @@ static int clusterManagerMigrateKeysInSlot(clusterManagerNode *source,
if (reply != NULL) freeReplyObject(reply);
if (migrate_reply != NULL) freeReplyObject(migrate_reply);
if (dots) zfree(dots);
reply = NULL;
migrate_reply = NULL;
dots = NULL;
if (!success) break;
}
config.conn_info.input_dbnum = orig_db;
cliSelect(&config, source->context);
return success;
}

Expand Down Expand Up @@ -8651,11 +8676,11 @@ static int getDbSize(void) {
return size;
}

static int getDatabases(void) {
static int getDatabases(redisContext *ctx) {
redisReply *reply;
int dbnum;

reply = redisCommand(context, "CONFIG GET databases");
reply = redisCommand(ctx, "CONFIG GET databases");

if (reply == NULL) {
fprintf(stderr, "\nI/O error\n");
Expand Down Expand Up @@ -9158,7 +9183,7 @@ void bytesToHuman(char *s, size_t size, long long n) {
static void statMode(void) {
redisReply *reply;
long aux, requests = 0;
int dbnum = getDatabases();
int dbnum = getDatabases(context);
int i = 0;

while (1) {
Expand Down
Loading
Loading