Compare commits

..
Author SHA1 Message Date
Vivek Saini 330d57198e fix unused lambda capture 2022-01-04 18:00:45 -05:00
John Sully b4528b797d Remove double free in keydb-benchmark 2021-12-23 14:20:41 -05:00
John Sully e070651039 Fix deadlock issue #183 cause by lock inversion 2021-12-23 13:17:14 -05:00
John Sully 7a32ec39fd Apple build break fix 2021-12-23 13:12:59 -05:00
John Sully 50dfee2039 Fix module test break 2021-12-23 13:12:34 -05:00
Vivek Saini 4b6bdcbcce Merge pull request #375 from VivekSainiEQ/time_thread_backport
Backport the time thread code from the enterprise version of KeyDB.

Also fix an issue with clustering where connections are potentially written to after being closed.
2021-11-15 12:57:15 -05:00
benschermel 31c0e38d86 Merge pull request #345 from EQ-Alpha/pkg_update_tools
add keydb-diagnostic-tool to packaging
2021-11-10 22:16:09 -05:00
benschermel 8f5b94c75f Merge pull request #376 from EQ-Alpha/systemd_build
Systemd build
2021-11-10 22:14:24 -05:00
benschermel d1fc7b96ed fix typo 2021-11-11 02:36:20 +00:00
benschermel b253a7aa8d update license 2021-11-11 02:19:56 +00:00
benschermel 1625c74a36 Merge branch 'unstable' into systemd_build 2021-11-11 00:59:33 +00:00
Vivek Saini 3c716ee33d Security fixes (#373)
Fixing potential buffer overflows when using the bitop shift and cron commands, as well as comparing ACL user passwords in constant time.
2021-11-09 19:00:01 -05:00
VivekSainiEQ b03eab2a06 Backported time thread code from enterprise 2021-11-05 00:30:34 +00:00
VivekSainiEQ 6a0a005ff2 Check that cluster connection is valid before setting write handler 2021-11-02 17:43:17 +00:00
malavan 51216ee433 null check for delete override 2021-10-29 15:59:21 +00:00
benschermel d7bd4f7a1c update deb packaging to build with systemd flag 2021-09-08 02:28:17 +00:00
benschermel 45bb6e801f add keydb-diagnostic-tool to packaging 2021-08-16 22:10:31 +00:00
37 changed files with 390 additions and 826 deletions

No files matched your search

+1
View File
@@ -6,6 +6,7 @@ Build-Depends:
debhelper (>= 9~),
dpkg-dev (>= 1.17.5),
systemd,
libsystemd-dev <!nocheck>,
procps <!nocheck>,
build-essential <!nocheck>,
tcl <!nocheck>,
+85 -7
View File
@@ -1,16 +1,94 @@
Format: https://www.debian.org/doc/packaging-manuals/copyright-format/1.0/
Upstream-Contact: John Sully <john@eqalpha.com>
Upstream-Name: keydb-pro
Source: https://github.com/JohnSully/KeyDB-Pro
Upstream-Name: KeyDB
Source: https://github.com/EQ-Alpha/KeyDB
Files: *
Copyright: © 2006-2014 Salvatore Sanfilippo <antirez@gmail.com>
Copyright © 2019, John Sully
License: Proprietary
Copyright:
© 2006-2014 Salvatore Sanfilippo <antirez@gmail.com>
© 2019-2021 John Sully
© 2019-2021 EQ Alpha Technology Ltd.
License: BSD-3-clause
Files: debian/*
Copyright: © 2009 Chris Lamb <lamby@debian.org>
Files:
src/rio.*
src/t_zset.c
src/ziplist.h
src/intset.*
src/redis-check-aof.c
deps/hiredis/*
deps/linenoise/*
Copyright:
© 2009-2012 Pieter Noordhuis <pcnoordhuis@gmail.com>
© 2009-2012 Salvatore Sanfilippo <antirez@gmail.com>
License: BSD-3-clause
Files:
src/lzf.h
src/lzfP.h
src/lzf_d.c
src/lzf_c.c
Copyright:
© 2000-2007 Marc Alexander Lehmann <schmorp@schmorp.de>
© 2009-2012 Salvatore Sanfilippo <antirez@gmail.com>
License: BSD-2-clause
Files: src/setproctitle.c
Copyright:
© 2010 William Ahern
© 2013 Salvatore Sanfilippo
© 2013 Stam He
License: BSD-3-clause
Files: src/ae_evport.c
Copyright: © 2012 Joyent, Inc.
License: BSD-3-clause
Files: src/ae_kqueue.c
Copyright: © 2009 Harish Mallipeddi <harish.mallipeddi@gmail.com>
License: BSD-3-clause
Files: utils/install_server.sh
Copyright: © 2011 Dvir Volk <dvirsk@gmail.com>
License: BSD-3-clause
Files: deps/jemalloc/*
Copyright:
© 2002-2012 Jason Evans <jasone@canonware.com>
© 2007-2012 Mozilla Foundation
© 2009-2012 Facebook, Inc.
License: BSD-3-clause
Files: src/pqsort.*
Copyright: © 1992-1993 The Regents of the University of California
License: BSD-3-clause
Files: deps/lua/*
Copyright: © 1994-2012 Lua.org, PUC-Ri
License: MIT
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
.
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
Files: pkg/deb/debian/*
Copyright:
© 2009 Chris Lamb <lamby@debian.org>
© 2020-2021 EQ Alpha Technology Ltd. <support@eqalpha.com>
License: BSD-3-clause
License: BSD-2-clause
+1
View File
@@ -5,3 +5,4 @@ src/keydb-check-rdb /usr/bin
src/keydb-cli /usr/bin
src/keydb-server /usr/bin
src/keydb-sentinel /usr/bin
src/keydb-diagnostic-tool /usr/bin
+1
View File
@@ -3,6 +3,7 @@
include /usr/share/dpkg/buildflags.mk
export BUILD_TLS=yes
export USE_SYSTEMD=yes
export CFLAGS CPPFLAGS LDFLAGS
export DEB_BUILD_MAINT_OPTIONS = hardening=+all
export DEB_LDFLAGS_MAINT_APPEND = -ldl -latomic $(LUA_LDFLAGS)
+1
View File
@@ -6,6 +6,7 @@ Build-Depends:
debhelper (>= 9~),
dpkg-dev (>= 1.17.5),
systemd,
libsystemd-dev <!nocheck>,
procps <!nocheck>,
build-essential <!nocheck>,
tcl <!nocheck>,
+11 -6
View File
@@ -1,11 +1,14 @@
Format: https://www.debian.org/doc/packaging-manuals/copyright-format/1.0/
Upstream-Contact: John Sully <john@eqalpha.com>
Upstream-Name: keydb
Source: https://github.com/JohnSully/KeyDB
Upstream-Name: KeyDB
Source: https://github.com/EQ-Alpha/KeyDB
Files: *
Copyright: © 2006-2014 Salvatore Sanfilippo <antirez@gmail.com>
Copyright © 2019, John Sully
Copyright:
© 2006-2014 Salvatore Sanfilippo <antirez@gmail.com>
© 2019-2021 John Sully
© 2019-2021 EQ Alpha Technology Ltd.
License: BSD-3-clause
Files:
@@ -82,8 +85,10 @@ License: MIT
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
Files: debian/*
Copyright: © 2009 Chris Lamb <lamby@debian.org>
Files: pkg/deb/debian_dh9/*
Copyright:
© 2009 Chris Lamb <lamby@debian.org>
© 2020-2021 EQ Alpha Technology Ltd. <support@eqalpha.com>
License: BSD-3-clause
License: BSD-2-clause
+1
View File
@@ -5,3 +5,4 @@ src/keydb-check-aof /usr/bin
src/keydb-check-rdb /usr/bin
src/keydb-cli /usr/bin
src/keydb-sentinel /usr/bin
src/keydb-diagnostic-tool /usr/bin
+1
View File
@@ -9,6 +9,7 @@ include /usr/share/dpkg/buildflags.mk
#LUA_LDFLAGS = $(addprefix -llua5.1-,$(LUA_LIBS_DEBIAN)) $(addprefix ../deps/lua/src/,$(LUA_OBJECTS))
export BUILD_TLS=yes
export USE_SYSTEMD=yes
export CFLAGS CPPFLAGS LDFLAGS
export DEB_BUILD_MAINT_OPTIONS = hardening=+all
export DEB_LDFLAGS_MAINT_APPEND = -ldl -latomic $(LUA_LDFLAGS)
+7 -1
View File
@@ -25,7 +25,13 @@ mkdir -p $DIR/keydb_build/keydb_rpm/var/log/keydb
# move binaries to bin
rm $DIR/keydb_build/keydb_rpm/usr/bin/*
cp $DIR/../../src/keydb-* $DIR/keydb_build/keydb_rpm/usr/bin/
cp $DIR/../../src/keydb-server $DIR/keydb_build/keydb_rpm/usr/bin/
cp $DIR/../../src/keydb-sentinel $DIR/keydb_build/keydb_rpm/usr/bin/
cp $DIR/../../src/keydb-cli $DIR/keydb_build/keydb_rpm/usr/bin/
cp $DIR/../../src/keydb-benchmark $DIR/keydb_build/keydb_rpm/usr/bin/
cp $DIR/../../src/keydb-check-aof $DIR/keydb_build/keydb_rpm/usr/bin/
cp $DIR/../../src/keydb-check-rdb $DIR/keydb_build/keydb_rpm/usr/bin/
cp $DIR/../../src/keydb-diagnostic-tool $DIR/keydb_build/keydb_rpm/usr/bin/
# update spec file with build info
sed -i '2d' $DIR/keydb_build/keydb.spec
+9 -38
View File
@@ -33,7 +33,6 @@ extern "C" {
}
#include <fcntl.h>
#include <ctype.h>
#include <uuid/uuid.h>
/* =============================================================================
* Global state for ACLs
@@ -224,12 +223,18 @@ uint64_t ACLGetCommandCategoryFlagByName(const char *name) {
return 0; /* No match. */
}
/* Method for passwords/pattern comparison used for the user->passwords list
/* Method for pattern comparison used for the user->patterns list
* so that we can search for items with listSearchKey(). */
int ACLListMatchSds(void *a, void *b) {
return sdscmp((sds)a,(sds)b) == 0;
}
/* Method for password comparison used for the user->passwords list
* Like the above, but uses a time independant compare for security reasons */
int ACLListMatchSdsSecure(void *a, void* b) {
return time_independent_strcmp((sds)a,(sds)b) == 0;
}
/* Method to free list elements from ACL users password/patterns lists. */
void ACLListFreeSds(const void *item) {
sdsfree((sds)item);
@@ -249,19 +254,12 @@ user *ACLCreateUser(const char *name, size_t namelen) {
if (raxFind(Users,(unsigned char*)name,namelen) != raxNotFound) return NULL;
user *u = (user*)zmalloc(sizeof(*u), MALLOC_LOCAL);
u->name = sdsnewlen(name,namelen);
if (!strcasecmp(cserver.default_user_namespace, "::auto")) {
u->ns_name = generateAutoNamespaceName();
} else {
u->ns_name = sdsnew(cserver.default_user_namespace);
}
u->flags = USER_FLAG_DISABLED | g_pserver->acl_pubsub_default;
u->allowed_subcommands = NULL;
u->passwords = listCreate();
u->patterns = listCreate();
u->channels = listCreate();
listSetMatchMethod(u->passwords,ACLListMatchSds);
listSetMatchMethod(u->passwords,ACLListMatchSdsSecure);
listSetFreeMethod(u->passwords,ACLListFreeSds);
listSetDupMethod(u->passwords,ACLListDupSds);
listSetMatchMethod(u->patterns,ACLListMatchSds);
@@ -296,7 +294,6 @@ user *ACLCreateUnlinkedUser(void) {
* will not remove the user from the Users global radix tree. */
void ACLFreeUser(user *u) {
sdsfree(u->name);
sdsfree(u->ns_name);
listRelease(u->passwords);
listRelease(u->patterns);
listRelease(u->channels);
@@ -347,8 +344,6 @@ void ACLCopyUser(user *dst, user *src) {
memcpy(dst->allowed_commands,src->allowed_commands,
sizeof(dst->allowed_commands));
dst->flags = src->flags;
sdsfree(dst->ns_name);
dst->ns_name = sdsdup(src->ns_name);
ACLResetSubcommands(dst);
/* Copy the allowed subcommands array of array of SDS strings. */
if (src->allowed_subcommands) {
@@ -680,13 +675,6 @@ sds ACLDescribeUser(user *u) {
sds rules = ACLDescribeUserCommandRules(u);
res = sdscatsds(res,rules);
sdsfree(rules);
/* namespace */
if (strcmp(u->ns_name, cserver.default_user_namespace) != 0) {
res = sdscatlen(res, " ", 1);
res = sdscatsds(res, u->ns_name);
}
return res;
}
@@ -819,9 +807,6 @@ void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) {
* reset Performs the following actions: resetpass, resetkeys, off,
* -@all. The user returns to the same state it has immediately
* after its creation.
* ::<ns> Sets the namespace for the user.
* :: is the default namespace
* ::auto will generate a new namespace with a random uuid
*
* The 'op' string must be null terminated. The 'oplen' argument should
* specify the length of the 'op' string in case the caller requires to pass
@@ -1021,14 +1006,6 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
errno = ENOENT;
return C_ERR;
}
} else if (op[0] == ':' && op[1] == ':') {
if (!strcasecmp(op, "::auto")) {
u->ns_name = generateAutoNamespaceName();
} else {
sds ns = sdsnewlen(op,oplen);
sdsfree(u->ns_name);
u->ns_name = ns;
}
} else if (!strcasecmp(op,"reset")) {
serverAssert(ACLSetUser(u,"resetpass",-1) == C_OK);
serverAssert(ACLSetUser(u,"resetkeys",-1) == C_OK);
@@ -1143,8 +1120,6 @@ int ACLAuthenticateUser(client *c, robj *username, robj *password) {
if (ACLCheckUserCredentials(username,password) == C_OK) {
c->authenticated = 1;
c->user = ACLGetUserByName((sds)ptrFromObj(username),sdslen((sds)ptrFromObj(username)));
c->ns = getNamespace(c->user->ns_name);
selectDbNamespaced(c, c->db->mapped_id);
moduleNotifyUserChanged(c);
return C_OK;
} else {
@@ -2000,7 +1975,7 @@ void aclCommand(client *c) {
return;
}
addReplyMapLen(c,6);
addReplyMapLen(c,5);
/* Flags */
addReplyBulkCString(c,"flags");
@@ -2061,10 +2036,6 @@ void aclCommand(client *c) {
addReplyBulkCBuffer(c,thispat,sdslen(thispat));
}
}
/* namespace */
addReplyBulkCString(c,"namespace");
addReplyBulkCString(c, u->ns_name);
} else if ((!strcasecmp(sub,"list") || !strcasecmp(sub,"users")) &&
c->argc == 2)
{
+4 -19
View File
@@ -731,14 +731,10 @@ void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int a
/* The DB this command was targeting is not the same as the last command
* we appended. To issue a SELECT command is needed. */
if (dictid != -1 && dictid != g_pserver->aof_selected_db) {
char seldb[64], mappeddb[64];
if (dictid != g_pserver->aof_selected_db) {
char seldb[64];
snprintf(seldb,sizeof(seldb),"%d",dictid);
snprintf(mappeddb,sizeof(seldb),"%d",g_pserver->db[dictid].mapped_id);
buf = sdscatprintf(buf, "*4\r\n$8\r\nALLOCATE\r\n$%lu\r\n%s\r\n$%lu\r\n%s\r\n$%lu\r\n%s\r\n",
sdslen(g_pserver->db[dictid].ns->name), g_pserver->db[dictid].ns->name, strlen(seldb), seldb, strlen(mappeddb), mappeddb);
buf = sdscatprintf(buf,"*2\r\n$6\r\nSELECT\r\n$%lu\r\n%s\r\n",
(unsigned long)strlen(seldb),seldb);
g_pserver->aof_selected_db = dictid;
@@ -771,8 +767,7 @@ void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int a
struct client *createAOFClient(void) {
struct client *c =(client*) zmalloc(sizeof(*c), MALLOC_LOCAL);
c->ns = g_pserver->default_namespace;
selectDbNamespaced(c, 0);
selectDb(c,0);
c->id = CLIENT_ID_AOF; /* So modules can identify it's the AOF client. */
c->conn = NULL;
c->iel = IDX_EVENT_LOOP_MAIN;
@@ -1551,22 +1546,12 @@ int rewriteAppendOnlyFileRio(rio *aof) {
long long updated_time = 0;
for (j = 0; j < cserver.dbnum; j++) {
char allocatecmd[] = "*4\r\n$8\r\nALLOCATE\r\n";
char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n";
redisDb *db = g_pserver->db+j;
dict *d = db->dict;
if (dictSize(d) == 0) continue;
serverAssert(db->ns);
serverAssert(db->mapped_id >= 0);
di = dictGetSafeIterator(d);
/* ALLOCATE DB */
if (rioWrite(aof,allocatecmd,sizeof(allocatecmd)-1) == 0) goto werr;
if (rioWriteBulkString(aof,db->ns->name, sdslen(db->ns->name)) == 0) goto werr;
if (rioWriteBulkLongLong(aof,db->id) == 0) goto werr;
if (rioWriteBulkLongLong(aof,db->mapped_id) == 0) goto werr;
/* SELECT the new DB */
if (rioWrite(aof,selectcmd,sizeof(selectcmd)-1) == 0) goto werr;
if (rioWriteBulkLongLong(aof,j) == 0) goto werr;
+15 -6
View File
@@ -405,6 +405,10 @@ void printBits(unsigned char *p, unsigned long count) {
#define BITFIELDOP_SET 1
#define BITFIELDOP_INCRBY 2
/* Make sure we don't bit shift too many bits at a time.
* Even this amount is probably way too large. */
#define BITOP_SHIFT_MAX (1<<30)
/* This helper function used by GETBIT / SETBIT parses the bit offset argument
* making sure an error is returned if it is negative or if it overflows
* Redis 512 MB limit for the string value or more (server.proto_max_bulk_len).
@@ -642,6 +646,11 @@ void bitopCommand(client *c) {
addReplyError(c, "BITOP SHIFT's last parameter must be an integer");
return;
}
if (shift > BITOP_SHIFT_MAX) {
addReplyError(c, "BITOP SHIFT value is too large.");
return;
}
if (op == BITOP_RSHIFT)
shift = -shift;
}
@@ -682,7 +691,7 @@ void bitopCommand(client *c) {
if (fShiftOp)
{
long newlen = (long)maxlen + shift/CHAR_BIT;
long long newlen = (long long)maxlen + shift/CHAR_BIT;
if (shift > 0 && (shift % CHAR_BIT) != 0)
newlen++;
@@ -694,14 +703,14 @@ void bitopCommand(client *c) {
res = (unsigned char*) sdsnewlen(NULL,newlen);
if (shift >= 0)
{ // left shift
long byteoffset = shift/CHAR_BIT;
long long byteoffset = shift/CHAR_BIT;
memset(res, 0, byteoffset);
long srcLen = newlen - byteoffset - ((shift % CHAR_BIT) ? 1 : 0);
long long srcLen = newlen - byteoffset - ((shift % CHAR_BIT) ? 1 : 0);
// now the bitshift+copy
unsigned bitshift = shift % CHAR_BIT;
unsigned char carry = 0;
for (long iSrc = 0; iSrc < srcLen; ++iSrc)
for (long long iSrc = 0; iSrc < srcLen; ++iSrc)
{
res[byteoffset+iSrc] = (src[0][iSrc] << bitshift) | carry;
carry = src[0][iSrc] >> (CHAR_BIT - bitshift);
@@ -711,14 +720,14 @@ void bitopCommand(client *c) {
}
else
{ // right shift
long byteoffset = -shift/CHAR_BIT;
long long byteoffset = -shift/CHAR_BIT;
unsigned bitshift = -shift % CHAR_BIT;
if (bitshift)
++byteoffset;
res[0] = (src[0][byteoffset] << (CHAR_BIT-bitshift));
if (byteoffset > 0)
res[0] |= (src[0][byteoffset-1] >> bitshift);
for (long idx = 1; idx < newlen; ++idx)
for (long long idx = 1; idx < newlen; ++idx)
{
res[idx] = (src[0][byteoffset+idx] << (CHAR_BIT-bitshift)) | (src[0][byteoffset+idx-1] >> bitshift);
}
-1
View File
@@ -592,7 +592,6 @@ void handleClientsBlockedOnKeys(void) {
* lookup, invalidating the first one.
* See https://github.com/redis/redis/pull/6554. */
serverTL->fixed_time_expire++;
updateCachedTime(0);
/* Serve clients blocked on the key. */
robj *o = lookupKeyWrite(rl->db,rl->key);
+16 -24
View File
@@ -2196,32 +2196,24 @@ int clusterProcessPacket(clusterLink *link) {
}
} else if (type == CLUSTERMSG_TYPE_PUBLISH) {
robj *channel, *message;
sds ns_name;
redisNamespace *ns;
uint32_t ns_len, channel_len, message_len;
ns_len = ntohl(hdr->data.publish.msg.ns_len);
ns_name = sdsnewlen((char*)hdr->data.publish.msg.bulk_data,ns_len);
ns = getNamespace(ns_name);
uint32_t channel_len, message_len;
/* Don't bother creating useless objects if there are no
* Pub/Sub subscribers. */
if (!ns ||
dictSize(ns->pubsub_channels) ||
dictSize(ns->pubsub_patterns))
if (dictSize(g_pserver->pubsub_channels) ||
dictSize(g_pserver->pubsub_patterns))
{
channel_len = ntohl(hdr->data.publish.msg.channel_len);
message_len = ntohl(hdr->data.publish.msg.message_len);
channel = createStringObject(
(char*)hdr->data.publish.msg.bulk_data+ns_len,channel_len);
(char*)hdr->data.publish.msg.bulk_data,channel_len);
message = createStringObject(
(char*)hdr->data.publish.msg.bulk_data+ns_len+channel_len,
(char*)hdr->data.publish.msg.bulk_data+channel_len,
message_len);
pubsubPublishMessage(ns,channel,message);
pubsubPublishMessage(channel,message);
decrRefCount(channel);
decrRefCount(message);
}
sdsfree(ns_name);
} else if (type == CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST) {
if (!sender) return 1; /* We don't know that node. */
clusterSendFailoverAuthIfNeeded(sender,hdr);
@@ -2455,7 +2447,10 @@ void clusterSendMessage(clusterLink *link, unsigned char *msg, size_t msglen) {
if (sdslen(link->sndbuf) == 0 && msglen != 0)
{
aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, [link] {
connSetWriteHandlerWithBarrier(link->conn, clusterWriteHandler, 1);
/* The connection could be timed out before this posted function executes (thanks to TCP keepalive).
* So check that the connection is still there before setting the write handler, otherwise you segfault */
if (link->conn != nullptr)
connSetWriteHandlerWithBarrier(link->conn, clusterWriteHandler, 1);
});
}
@@ -2762,24 +2757,22 @@ void clusterBroadcastPong(int target) {
/* Send a PUBLISH message.
*
* If link is NULL, then the message is broadcasted to the whole cluster. */
void clusterSendPublish(redisNamespace *ns, clusterLink *link, robj *channel, robj *message) {
void clusterSendPublish(clusterLink *link, robj *channel, robj *message) {
unsigned char *payload;
clusterMsg buf[1];
clusterMsg *hdr = (clusterMsg*) buf;
uint32_t totlen;
uint32_t channel_len, message_len, ns_len;
uint32_t channel_len, message_len;
channel = getDecodedObject(channel);
message = getDecodedObject(message);
channel_len = sdslen(szFromObj(channel));
message_len = sdslen(szFromObj(message));
ns_len = sdslen(ns->name);
clusterBuildMessageHdr(hdr,CLUSTERMSG_TYPE_PUBLISH);
totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
totlen += sizeof(clusterMsgDataPublish) - 8 + channel_len + message_len + ns_len;
totlen += sizeof(clusterMsgDataPublish) - 8 + channel_len + message_len;
hdr->data.publish.msg.ns_len = htonl(ns_len);
hdr->data.publish.msg.channel_len = htonl(channel_len);
hdr->data.publish.msg.message_len = htonl(message_len);
hdr->totlen = htonl(totlen);
@@ -2792,8 +2785,7 @@ void clusterSendPublish(redisNamespace *ns, clusterLink *link, robj *channel, ro
memcpy(payload,hdr,sizeof(*hdr));
hdr = (clusterMsg*) payload;
}
memcpy(hdr->data.publish.msg.bulk_data,ns->name,ns_len);
memcpy(hdr->data.publish.msg.bulk_data+ns_len,ptrFromObj(channel),sdslen(szFromObj(channel)));
memcpy(hdr->data.publish.msg.bulk_data,ptrFromObj(channel),sdslen(szFromObj(channel)));
memcpy(hdr->data.publish.msg.bulk_data+sdslen(szFromObj(channel)),
ptrFromObj(message),sdslen(szFromObj(message)));
@@ -2899,8 +2891,8 @@ int clusterSendModuleMessageToTarget(const char *target, uint64_t module_id, uin
* cluster. In the future we'll try to get smarter and avoiding propagating those
* messages to hosts without receives for a given channel.
* -------------------------------------------------------------------------- */
void clusterPropagatePublish(redisNamespace *ns, robj *channel, robj *message) {
clusterSendPublish(ns, NULL, channel, message);
void clusterPropagatePublish(robj *channel, robj *message) {
clusterSendPublish(NULL, channel, message);
}
/* -----------------------------------------------------------------------------
-1
View File
@@ -209,7 +209,6 @@ typedef struct {
} clusterMsgDataFail;
typedef struct {
uint32_t ns_len;
uint32_t channel_len;
uint32_t message_len;
unsigned char bulk_data[8]; /* 8 bytes just as placeholder. */
+2 -7
View File
@@ -542,11 +542,7 @@ void loadServerConfigFromString(char *config) {
if (port < 0 || port > 65535 || *ptr != '\0') {
err= "Invalid master port"; goto loaderr;
}
redisMasterConnInfo *masterInfo = (redisMasterConnInfo*)zmalloc(sizeof(redisMasterConnInfo), MALLOC_LOCAL);
masterInfo->ip = sdsnew(argv[1]);
masterInfo->port = port;
listAddNodeTail(g_pserver->repl_init_masters, masterInfo);
replicationAddMaster(argv[1], port);
} else if (!strcasecmp(argv[0],"requirepass") && argc == 2) {
if (strlen(argv[1]) > CONFIG_AUTHPASS_MAX_LEN) {
err = "Password is longer than CONFIG_AUTHPASS_MAX_LEN";
@@ -2671,7 +2667,7 @@ standardConfig configs[] = {
createStringConfig("bgsave_cpulist", NULL, IMMUTABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->bgsave_cpulist, NULL, NULL, NULL),
createStringConfig("ignore-warnings", NULL, MODIFIABLE_CONFIG, ALLOW_EMPTY_STRING, g_pserver->ignore_warnings, "", NULL, NULL),
createStringConfig("proc-title-template", NULL, MODIFIABLE_CONFIG, ALLOW_EMPTY_STRING, cserver.proc_title_template, CONFIG_DEFAULT_PROC_TITLE_TEMPLATE, isValidProcTitleTemplate, updateProcTitleTemplate),
createStringConfig("acl-namespace-default", NULL, IMMUTABLE_CONFIG, ALLOW_EMPTY_STRING, cserver.default_user_namespace, "::", NULL, NULL), //TODO: validation
/* SDS Configs */
createSDSConfig("masterauth", NULL, MODIFIABLE_CONFIG | SENSITIVE_CONFIG, EMPTY_STRING_IS_NULL, cserver.default_masterauth, NULL, NULL, updateMasterAuthConfig),
createSDSConfig("requirepass", NULL, MODIFIABLE_CONFIG | SENSITIVE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->requirepass, NULL, NULL, updateRequirePass),
@@ -2689,7 +2685,6 @@ standardConfig configs[] = {
/* Integer configs */
createIntConfig("databases", NULL, IMMUTABLE_CONFIG, 1, INT_MAX, cserver.dbnum, 16, INTEGER_CONFIG, NULL, NULL),
createIntConfig("databases-per-namespace", NULL, IMMUTABLE_CONFIG, 1, INT_MAX, cserver.ns_dbnum, INT_MAX, INTEGER_CONFIG, NULL, NULL),
createIntConfig("port", NULL, MODIFIABLE_CONFIG, 0, 65535, g_pserver->port, 6379, INTEGER_CONFIG, NULL, updatePort), /* TCP port. */
createIntConfig("auto-aof-rewrite-percentage", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, g_pserver->aof_rewrite_perc, 100, INTEGER_CONFIG, NULL, NULL),
createIntConfig("cluster-replica-validity-factor", "cluster-slave-validity-factor", MODIFIABLE_CONFIG, 0, INT_MAX, g_pserver->cluster_slave_validity_factor, 10, INTEGER_CONFIG, NULL, NULL), /* Slave max data age factor. */
+1
View File
@@ -51,6 +51,7 @@ void cronCommand(client *c)
if (c->argc < (6 + numkeys)) {
addReplyError(c, "Missing arguments or numkeys is too big");
return;
}
}
+11 -146
View File
@@ -494,13 +494,6 @@ long long emptyDbStructure(redisDb *dbarray, int dbnum, int async,
}
/* Because all keys of database are removed, reset average ttl. */
dbarray[j].avg_ttl = 0;
if (dbnum == -1 && dbarray[j].ns && g_pserver->db == dbarray) {
dbarray[j].ns->db[dbarray[j].mapped_id] = nullptr;
dbarray[j].mapped_id = -1;
dbarray[j].ns = nullptr;
g_pserver->last_allocated_db = -1;
}
}
return removed;
@@ -650,88 +643,6 @@ int selectDb(client *c, int id) {
if (id < 0 || id >= cserver.dbnum)
return C_ERR;
c->db = &g_pserver->db[id];
c->ns = c->db->ns;
serverAssertWithInfo(NULL,NULL,c->db->mapped_id >= 0);
serverAssertWithInfo(NULL,NULL,c->db->ns);
return C_OK;
}
void mapDb(redisNamespace *ns, int global_db, int ns_db, int do_propagate) {
//serverLog(LL_WARNING, "about to map %s %d %d", ns->name, global_db, ns_db);
serverAssertWithInfo(NULL, NULL, global_db >= 0 && global_db < cserver.dbnum);
serverAssertWithInfo(NULL, NULL, ns_db >= 0 && ns_db < std::min(cserver.dbnum, cserver.ns_dbnum));
serverAssertWithInfo(NULL, NULL, !g_pserver->db[global_db].ns || (g_pserver->db[global_db].ns == ns && g_pserver->db[global_db].mapped_id == ns_db));
g_pserver->last_allocated_db = std::max(g_pserver->last_allocated_db, global_db);
g_pserver->db[global_db].ns = ns;
g_pserver->db[global_db].mapped_id = ns_db;
ns->db[ns_db] = &g_pserver->db[global_db];
if (do_propagate > 0) {
/* propagate allocation to make sure mapping is the same on all nodes */
robj *argv[4];
argv[0] = shared.allocate;
argv[1] = createObject(OBJ_STRING, sdsdup(ns->name));
argv[2] = createObject(OBJ_STRING, sdsfromlonglong(global_db));
argv[3] = createObject(OBJ_STRING, sdsfromlonglong(ns_db));
serverLog(LL_WARNING, "propagate %s %d %d", ns->name, global_db, ns_db);
if (do_propagate == 1)
propagate(cserver.allocateCommand, -1, argv, 4, PROPAGATE_AOF|PROPAGATE_REPL);
else
alsoPropagate(cserver.allocateCommand, -1, argv, 4, PROPAGATE_AOF|PROPAGATE_REPL);
decrRefCount(argv[1]);
decrRefCount(argv[2]);
decrRefCount(argv[3]);
}
}
void printDbMap() {
serverLog(LL_NOTICE, "------ database mapping ------");
for (int i=0; i<cserver.dbnum;i++) {
serverLog(LL_NOTICE, "%d => %d (%s) @ %d keys", i, g_pserver->db[i].mapped_id,
g_pserver->db[i].ns ? g_pserver->db[i].ns->name : "<none>", (int) dictSize(g_pserver->db[i].dict));
}
serverLog(LL_NOTICE, "------ default ns mapping ------");
for (int i=0; i<std::min(cserver.dbnum,cserver.ns_dbnum);i++) {
if (g_pserver->default_namespace->db[i]) {
serverLog(LL_NOTICE, "%d => %d (%s) @ %d keys", i, g_pserver->default_namespace->db[i]->mapped_id,
g_pserver->default_namespace->db[i]->ns ? g_pserver->default_namespace->db[i]->ns->name
: "<none>",
(int) dictSize(g_pserver->default_namespace->db[i]->dict));
}
}
serverLog(LL_NOTICE, "---- end database mapping ----");
}
redisDb *allocateDb(client *c, int id) {
if (id < 0 || id >= std::min(cserver.dbnum, cserver.ns_dbnum)) return nullptr;
if (!c->ns->db[id]) {
/* allocate database from global pool */
int found = 0;
for (int i=g_pserver->last_allocated_db+1;i<cserver.dbnum; i++) {
if (!g_pserver->db[i].ns) {
mapDb(c->ns, i, id, 1);
found=1;
break;
}
}
if (!found) {
serverLog(LL_WARNING, "We ran out of databases!!");
return nullptr;
}
}
return c->ns->db[id];
}
int selectDbNamespaced(client *c, int id) {
redisDb *db = allocateDb(c, id);
if (db == nullptr) return C_ERR;
c->db = db;
return C_OK;
}
@@ -805,17 +716,8 @@ int getFlushCommandFlags(client *c, int *flags) {
}
/* Flushes the whole server data set. */
void flushAllDataAndResetRDB(redisNamespace *ns, int flags) {
if (ns == NULL) {
g_pserver->dirty += emptyDb(-1, flags, NULL);
} else {
for (int i=0; i<std::min(cserver.dbnum, cserver.ns_dbnum); i++) {
if (ns->db[i]) {
g_pserver->dirty += emptyDb(ns->db[i]->id,flags,NULL);
}
}
}
void flushAllDataAndResetRDB(int flags) {
g_pserver->dirty += emptyDb(-1,flags,NULL);
if (g_pserver->child_type == CHILD_TYPE_RDB) killRDBChild();
if (g_pserver->saveparamslen > 0) {
/* Normally rdbSave() will reset dirty, but we don't want this here
@@ -863,7 +765,7 @@ void flushdbCommand(client *c) {
void flushallCommand(client *c) {
int flags;
if (getFlushCommandFlags(c,&flags) == C_ERR) return;
flushAllDataAndResetRDB(c->ns, flags);
flushAllDataAndResetRDB(flags);
addReply(c,shared.ok);
}
@@ -923,48 +825,13 @@ void selectCommand(client *c) {
addReplyError(c,"SELECT is not allowed in cluster mode");
return;
}
bool is_replication = ((c->flags & CLIENT_MASTER) || c->id == CLIENT_ID_AOF);
if (is_replication && selectDb(c, id) == C_ERR) {
addReplyError(c,"DB index is out of range");
} else if (!is_replication && selectDbNamespaced(c, id) == C_ERR) {
if (selectDb(c,id) == C_ERR) {
addReplyError(c,"DB index is out of range");
} else {
addReply(c,shared.ok);
}
}
void selectNsCommand(client *c) {
//TODO: validate namespace name
c->ns = getNamespace(szFromObj(c->argv[2]));
selectCommand(c);
}
void allocateCommand(client *c) {
int global_db, ns_db;
bool is_replication = ((c->flags & CLIENT_MASTER) || c->id == CLIENT_ID_AOF);
if (!is_replication) {
addReplyError(c,"ALLOCATE is only allowed for replication");
return;
}
//TODO: validate namespace name
redisNamespace *ns = getNamespace(szFromObj(c->argv[1]));
if (getIntFromObjectOrReply(c, c->argv[2], &global_db, NULL) != C_OK)
return;
if (getIntFromObjectOrReply(c, c->argv[3], &ns_db, NULL) != C_OK)
return;
serverLog(LL_WARNING, "allocate %s %d %d", ns->name, global_db, ns_db);
//TODO: error handling
mapDb(ns, global_db, ns_db, 0);
addReply(c,shared.ok);
}
void randomkeyCommand(client *c) {
robj *key;
@@ -1392,7 +1259,7 @@ void moveCommand(client *c) {
if (getIntFromObjectOrReply(c, c->argv[2], &dbid, NULL) != C_OK)
return;
if (selectDbNamespaced(c, dbid) == C_ERR) {
if (selectDb(c,dbid) == C_ERR) {
addReplyError(c,"DB index is out of range");
return;
}
@@ -1467,7 +1334,7 @@ void copyCommand(client *c) {
if (getIntFromObjectOrReply(c, c->argv[j+1], &dbid, NULL) != C_OK)
return;
if (selectDbNamespaced(c, dbid) == C_ERR) {
if (selectDb(c, dbid) == C_ERR) {
addReplyError(c,"DB index is out of range");
return;
}
@@ -1571,13 +1438,11 @@ void scanDatabaseForReadyLists(redisDb *db) {
*
* Returns C_ERR if at least one of the DB ids are out of range, otherwise
* C_OK is returned. */
int dbSwapDatabases(client *c, int id1, int id2) {
if (id1 < 0 || id1 >= std::min(cserver.dbnum, cserver.ns_dbnum) ||
id2 < 0 || id2 >= std::min(cserver.dbnum, cserver.ns_dbnum)) return C_ERR;
int dbSwapDatabases(int id1, int id2) {
if (id1 < 0 || id1 >= cserver.dbnum ||
id2 < 0 || id2 >= cserver.dbnum) return C_ERR;
if (id1 == id2) return C_OK;
redisDb *db1 = allocateDb(c, id1);
redisDb *db2 = allocateDb(c, id2);
redisDb *db1 = &g_pserver->db[id1], *db2 = &g_pserver->db[id2];
/* Swap hash tables. Note that we don't swap blocking_keys,
* ready_keys and watched_keys, since we want clients to
@@ -1627,7 +1492,7 @@ void swapdbCommand(client *c) {
return;
/* Swap... */
if (dbSwapDatabases(c, id1,id2) == C_ERR) {
if (dbSwapDatabases(id1,id2) == C_ERR) {
addReplyError(c,"DB index is out of range");
return;
} else {
+4 -8
View File
@@ -567,11 +567,9 @@ NULL
serverLog(LL_WARNING,"DB reloaded by DEBUG RELOAD");
addReply(c,shared.ok);
} else if (!strcasecmp(szFromObj(c->argv[1]),"loadaof")) {
serverLog(LL_WARNING,"Append Only File loaded by DEBUG LOADAOF started");
if (g_pserver->aof_state != AOF_OFF) flushAppendOnlyFile(1);
emptyDb(-1,EMPTYDB_NO_FLAGS,NULL);
protectClient(c);
int ret = loadAppendOnlyFile(g_pserver->aof_filename);
unprotectClient(c);
if (ret != C_OK) {
@@ -579,7 +577,7 @@ NULL
return;
}
g_pserver->dirty = 0; /* Prevent AOF / replication */
serverLog(LL_WARNING,"Append Only File loaded by DEBUG LOADAOF done");
serverLog(LL_WARNING,"Append Only File loaded by DEBUG LOADAOF");
addReply(c,shared.ok);
} else if (!strcasecmp(szFromObj(c->argv[1]),"object") && c->argc == 3) {
dictEntry *de;
@@ -842,18 +840,18 @@ NULL
sdsfree(stats);
return;
}
if (dbid < 0 || dbid >= std::min(cserver.dbnum, cserver.ns_dbnum)) {
if (dbid < 0 || dbid >= cserver.dbnum) {
sdsfree(stats);
addReplyError(c,"Out of range database");
return;
}
stats = sdscatprintf(stats,"[Dictionary HT]\n");
dictGetStats(buf,sizeof(buf),c->ns->db[dbid]->dict);
dictGetStats(buf,sizeof(buf),g_pserver->db[dbid].dict);
stats = sdscat(stats,buf);
stats = sdscatprintf(stats,"[Expires set]\n");
c->ns->db[dbid]->setexpire->getstats(buf, sizeof(buf));
g_pserver->db[dbid].setexpire->getstats(buf, sizeof(buf));
stats = sdscat(stats, buf);
addReplyVerbatim(c,stats,sdslen(stats),"txt");
@@ -974,7 +972,6 @@ void _serverAssertPrintClientInfo(const client *c) {
bugReportStart();
serverLog(LL_WARNING,"=== ASSERTION FAILED CLIENT CONTEXT ===");
serverLog(LL_WARNING,"client->id = %llu", (unsigned long long) c->id);
serverLog(LL_WARNING,"client->flags = %llu", (unsigned long long) c->flags);
serverLog(LL_WARNING,"client->conn = %s", connGetInfo(c->conn, conninfo, sizeof(conninfo)));
serverLog(LL_WARNING,"client->argc = %d", c->argc);
@@ -1038,7 +1035,6 @@ void _serverAssertPrintObject(robj_roptr o) {
void _serverAssertWithInfo(const client *c, robj_roptr o, const char *estr, const char *file, int line) {
if (c) _serverAssertPrintClientInfo(c);
if (o) _serverAssertPrintObject(o);
printDbMap();
_serverAssert(estr,file,line);
}
+2 -13
View File
@@ -981,19 +981,8 @@ long defragOtherGlobals() {
/* there are many more pointers to defrag (e.g. client argv, output / aof buffers, etc.
* but we assume most of these are short lived, we only need to defrag allocations
* that remain static for a long time */
redisNamespace *ns;
dictEntry *de;
dictIterator *di;
di = dictGetSafeIterator(g_pserver->namespaces);
while((de = dictNext(di)) != NULL) {
ns = (struct redisNamespace *) dictGetVal(de);
defragged += activeDefragSdsDict(ns->lua_scripts, DEFRAG_SDS_DICT_VAL_IS_STROB);
defragged += activeDefragSdsListAndDict(ns->repl_scriptcache_fifo, ns->repl_scriptcache_dict, DEFRAG_SDS_DICT_NO_VAL);
}
dictReleaseIterator(di);
defragged += activeDefragSdsDict(g_pserver->namespaces, DEFRAG_SDS_DICT_VAL_VOID_PTR);
defragged += activeDefragSdsDict(g_pserver->lua_scripts, DEFRAG_SDS_DICT_VAL_IS_STROB);
defragged += activeDefragSdsListAndDict(g_pserver->repl_scriptcache_fifo, g_pserver->repl_scriptcache_dict, DEFRAG_SDS_DICT_NO_VAL);
defragged += moduleDefragGlobals();
return defragged;
}
+13 -6
View File
@@ -2097,9 +2097,10 @@ int RM_GetClientInfoById(void *ci, uint64_t id) {
/* Publish a message to subscribers (see PUBLISH command). */
int RM_PublishMessage(RedisModuleCtx *ctx, RedisModuleString *channel, RedisModuleString *message) {
int receivers = pubsubPublishMessage(ctx->client->ns, channel, message);
UNUSED(ctx);
int receivers = pubsubPublishMessage(channel, message);
if (g_pserver->cluster_enabled)
clusterPropagatePublish(ctx->client->ns, channel, message);
clusterPropagatePublish(channel, message);
return receivers;
}
@@ -2292,7 +2293,7 @@ int RM_AvoidReplicaTraffic() {
* returns back to the original one, it should call RedisModule_GetSelectedDb()
* before in order to restore the old DB number before returning. */
int RM_SelectDb(RedisModuleCtx *ctx, int newid) {
int retval = selectDbNamespaced(ctx->client,newid);
int retval = selectDb(ctx->client,newid);
return (retval == C_OK) ? REDISMODULE_OK : REDISMODULE_ERR;
}
@@ -2512,7 +2513,7 @@ int RM_SetAbsExpire(RedisModuleKey *key, mstime_t expire) {
* When async is set to true, db contents will be freed by a background thread. */
void RM_ResetDataset(int restart_aof, int async) {
if (restart_aof && g_pserver->aof_state != AOF_OFF) stopAppendOnly();
flushAllDataAndResetRDB(NULL, async? EMPTYDB_ASYNC: EMPTYDB_NO_FLAGS);
flushAllDataAndResetRDB(async? EMPTYDB_ASYNC: EMPTYDB_NO_FLAGS);
if (g_pserver->aof_enabled && restart_aof) restartAOFAfterSYNC();
}
@@ -4214,7 +4215,6 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
/* We do not want to allow block, the module do not expect it */
c->flags |= CLIENT_DENY_BLOCKING;
c->db = ctx->client->db;
c->ns = ctx->client->ns;
c->argv = argv;
c->argc = argc;
if (ctx->module) ctx->module->in_call++;
@@ -4289,7 +4289,14 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
if (!(flags & REDISMODULE_ARGV_NO_REPLICAS))
call_flags |= CMD_CALL_PROPAGATE_REPL;
}
{
aeAcquireLock();
std::unique_lock<fastlock> ul(c->lock);
call(c,call_flags);
aeReleaseLock();
}
g_pserver->replication_allowed = prev_replication_allowed;
serverAssert((c->flags & CLIENT_BLOCKED) == 0);
@@ -8527,7 +8534,7 @@ void moduleFireServerEvent(uint64_t eid, int subid, void *data) {
* up to the specific event setup to change it when it makes
* sense. For instance for FLUSHDB events we select the correct
* DB automatically. */
selectDbNamespaced(ctx.client, 0);
selectDb(ctx.client, 0);
/* Event specific context and data pointer setup. */
if (eid == REDISMODULE_EVENT_CLIENT_CHANGE) {
+7 -15
View File
@@ -140,12 +140,12 @@ client *createClient(connection *conn, int iel) {
connSetPrivateData(conn, c);
}
selectDb(c,0);
uint64_t client_id;
client_id = g_pserver->next_client_id.fetch_add(1);
c->iel = iel;
fastlock_init(&c->lock, "client");
c->id = client_id;
c->ns = g_pserver->default_namespace;
c->resp = 2;
c->conn = conn;
c->name = NULL;
@@ -172,7 +172,6 @@ client *createClient(connection *conn, int iel) {
/* If the default user does not require authentication, the user is
* directly authenticated. */
clientSetDefaultAuth(c);
selectDbNamespaced(c,0);
c->replstate = REPL_STATE_NONE;
c->repl_put_online_on_ack = 0;
c->reploff = 0;
@@ -1322,7 +1321,7 @@ void acceptOnThread(connection *conn, int flags, char *cip)
szT = (char*)zmalloc(NET_IP_STR_LEN, MALLOC_LOCAL);
memcpy(szT, cip, NET_IP_STR_LEN);
}
int res = aePostFunction(g_pserver->rgthreadvar[ielTarget].el, [conn, flags, ielTarget, szT, fBootLoad] {
int res = aePostFunction(g_pserver->rgthreadvar[ielTarget].el, [conn, flags, ielTarget, szT] {
connMarshalThread(conn);
acceptCommonHandler(conn,flags,szT,ielTarget);
rgacceptsInFlight[ielTarget].fetch_sub(1, std::memory_order_relaxed);
@@ -1943,7 +1942,9 @@ void ProcessPendingAsyncWrites()
bool fResult = c->postFunction([](client *c) {
c->fPendingAsyncWriteHandler = false;
clientInstallWriteHandler(c);
c->lock.unlock();
handleClientsWithPendingWrites(c->iel, g_pserver->aof_state);
c->lock.lock();
}, false);
if (!fResult)
@@ -2733,7 +2734,7 @@ sds catClientInfoString(sds s, client *client) {
total_mem += zmalloc_size(client->argv);
return sdscatfmt(s,
"id=%U addr=%s laddr=%s %s name=%s age=%I idle=%I flags=%s db=%i(%i) sub=%i psub=%i multi=%i qbuf=%U qbuf-free=%U argv-mem=%U obl=%U oll=%U omem=%U tot-mem=%U events=%s cmd=%s user=%s redir=%I ns=%s",
"id=%U addr=%s laddr=%s %s name=%s age=%I idle=%I flags=%s db=%i sub=%i psub=%i multi=%i qbuf=%U qbuf-free=%U argv-mem=%U obl=%U oll=%U omem=%U tot-mem=%U events=%s cmd=%s user=%s redir=%I",
(unsigned long long) client->id,
getClientPeerId(client),
getClientSockname(client),
@@ -2742,7 +2743,6 @@ sds catClientInfoString(sds s, client *client) {
(long long)(g_pserver->unixtime - client->ctime),
(long long)(g_pserver->unixtime - client->lastinteraction),
flags,
client->db->mapped_id,
client->db->id,
(int) dictSize(client->pubsub_channels),
(int) listLength(client->pubsub_patterns),
@@ -2757,8 +2757,7 @@ sds catClientInfoString(sds s, client *client) {
events,
client->lastcmd ? client->lastcmd->name : "NULL",
client->user ? client->user->name : "(superuser)",
(client->flags & CLIENT_TRACKING) ? (long long) client->client_tracking_redirection : -1,
client->ns ? client->ns->name : "(unset)");
(client->flags & CLIENT_TRACKING) ? (long long) client->client_tracking_redirection : -1);
}
sds getAllClientsInfoString(int type) {
@@ -2839,7 +2838,7 @@ void resetCommand(client *c) {
}
if (c->flags & CLIENT_TRACKING) disableTracking(c);
selectDb(c,0);
c->resp = 2;
clientSetDefaultAuth(c);
@@ -2849,9 +2848,6 @@ void resetCommand(client *c) {
pubsubUnsubscribeAllChannels(c,0);
pubsubUnsubscribeAllPatterns(c,0);
c->ns = g_pserver->default_namespace;
selectDbNamespaced(c,0);
if (c->name) {
decrRefCount(c->name);
c->name = NULL;
@@ -3873,10 +3869,6 @@ void processEventsWhileBlocked(int iel) {
listNode *ln;
listRewind(g_pserver->clients, &li);
/* Update our cached time since it is used to create and update the last
* interaction time with clients and for other important things. */
updateCachedTime(0);
// All client locks must be acquired *after* the global lock is reacquired to prevent deadlocks
// so unlock here, and save them for reacquisition later
while ((ln = listNext(&li)) != nullptr)
+5 -2
View File
@@ -27,14 +27,17 @@ void *operator new(std::size_t size, const std::nothrow_t &) noexcept
return zmalloc(size, MALLOC_LOCAL);
}
//need to do null checks for delete since the compiler can optimize out null checks in zfree
void operator delete(void * p) noexcept
{
zfree(p);
if (p != nullptr)
zfree(p);
}
void operator delete(void *p, std::size_t) noexcept
{
zfree(p);
if (p != nullptr)
zfree(p);
}
#endif
+5 -6
View File
@@ -101,7 +101,6 @@ sds keyspaceEventsFlagsToString(int flags) {
void notifyKeyspaceEvent(int type, const char *event, robj *key, int dbid) {
sds chan;
robj *chanobj, *eventobj;
redisDb *db = g_pserver->db + dbid;
int len = -1;
char buf[24];
@@ -109,7 +108,7 @@ void notifyKeyspaceEvent(int type, const char *event, robj *key, int dbid) {
* This bypasses the notifications configuration, but the module engine
* will only call event subscribers if the event type matches the types
* they are interested in. */
moduleNotifyKeyspaceEvent(type, event, key, dbid);
moduleNotifyKeyspaceEvent(type, event, key, dbid);
/* If notifications for this class of events are off, return ASAP. */
if (!(g_pserver->notify_keyspace_events & type)) return;
@@ -119,24 +118,24 @@ void notifyKeyspaceEvent(int type, const char *event, robj *key, int dbid) {
/* __keyspace@<db>__:<key> <event> notifications. */
if (g_pserver->notify_keyspace_events & NOTIFY_KEYSPACE) {
chan = sdsnewlen("__keyspace@",11);
len = ll2string(buf,sizeof(buf), db->mapped_id);
len = ll2string(buf,sizeof(buf),dbid);
chan = sdscatlen(chan, buf, len);
chan = sdscatlen(chan, "__:", 3);
chan = sdscatsds(chan, szFromObj(key));
chanobj = createObject(OBJ_STRING, chan);
pubsubPublishMessage(db->ns, chanobj, eventobj);
pubsubPublishMessage(chanobj, eventobj);
decrRefCount(chanobj);
}
/* __keyevent@<db>__:<event> <key> notifications. */
if (g_pserver->notify_keyspace_events & NOTIFY_KEYEVENT) {
chan = sdsnewlen("__keyevent@",11);
if (len == -1) len = ll2string(buf,sizeof(buf),db->mapped_id);
if (len == -1) len = ll2string(buf,sizeof(buf),dbid);
chan = sdscatlen(chan, buf, len);
chan = sdscatlen(chan, "__:", 3);
chan = sdscatsds(chan, szFromObj(eventobj));
chanobj = createObject(OBJ_STRING, chan);
pubsubPublishMessage(db->ns, chanobj, key);
pubsubPublishMessage(chanobj, key);
decrRefCount(chanobj);
}
decrRefCount(eventobj);
+9 -21
View File
@@ -1112,26 +1112,15 @@ struct redisMemOverhead *getMemoryOverheadData(void) {
mh->aof_buffer = mem;
mem_total+=mem;
mem = 0;
redisNamespace *ns;
dictEntry *de;
dictIterator *di;
di = dictGetSafeIterator(g_pserver->namespaces);
while((de = dictNext(di)) != NULL) {
ns = (struct redisNamespace *) dictGetVal(de);
mem += ns->lua_scripts_mem;
mem += dictSize(ns->lua_scripts) * sizeof(dictEntry) +
dictSlots(ns->lua_scripts) * sizeof(dictEntry*);
mem += dictSize(ns->repl_scriptcache_dict) * sizeof(dictEntry) +
dictSlots(ns->repl_scriptcache_dict) * sizeof(dictEntry*);
if (listLength(ns->repl_scriptcache_fifo) > 0) {
mem += listLength(ns->repl_scriptcache_fifo) * (sizeof(listNode) +
sdsZmallocSize((sds)listNodeValue(listFirst(ns->repl_scriptcache_fifo))));
}
mem = g_pserver->lua_scripts_mem;
mem += dictSize(g_pserver->lua_scripts) * sizeof(dictEntry) +
dictSlots(g_pserver->lua_scripts) * sizeof(dictEntry*);
mem += dictSize(g_pserver->repl_scriptcache_dict) * sizeof(dictEntry) +
dictSlots(g_pserver->repl_scriptcache_dict) * sizeof(dictEntry*);
if (listLength(g_pserver->repl_scriptcache_fifo) > 0) {
mem += listLength(g_pserver->repl_scriptcache_fifo) * (sizeof(listNode) +
sdsZmallocSize((sds)listNodeValue(listFirst(g_pserver->repl_scriptcache_fifo))));
}
dictReleaseIterator(di);
mh->lua_caches = mem;
mem_total+=mem;
@@ -1245,8 +1234,7 @@ sds getMemoryDoctorReport(void) {
}
/* Too many scripts are cached? */
//TODO: all namespaces
if (dictSize(g_pserver->default_namespace->lua_scripts) > 1000) {
if (dictSize(g_pserver->lua_scripts) > 1000) {
many_scripts = 1;
num_reports++;
}
+16 -17
View File
@@ -145,10 +145,10 @@ int pubsubSubscribeChannel(client *c, robj *channel) {
retval = 1;
incrRefCount(channel);
/* Add the client to the channel -> list of clients hash table */
de = dictFind(c->ns->pubsub_channels,channel);
de = dictFind(g_pserver->pubsub_channels,channel);
if (de == NULL) {
clients = listCreate();
dictAdd(c->ns->pubsub_channels,channel,clients);
dictAdd(g_pserver->pubsub_channels,channel,clients);
incrRefCount(channel);
} else {
clients = (list*)dictGetVal(de);
@@ -174,7 +174,7 @@ int pubsubUnsubscribeChannel(client *c, robj *channel, int notify) {
if (dictDelete(c->pubsub_channels,channel) == DICT_OK) {
retval = 1;
/* Remove the client from the channel -> clients list hash table */
de = dictFind(c->ns->pubsub_channels,channel);
de = dictFind(g_pserver->pubsub_channels,channel);
serverAssertWithInfo(c,NULL,de != NULL);
clients = (list*)dictGetVal(de);
ln = listSearchKey(clients,c);
@@ -184,7 +184,7 @@ int pubsubUnsubscribeChannel(client *c, robj *channel, int notify) {
/* Free the list and associated hash entry at all if this was
* the latest client, so that it will be possible to abuse
* Redis PUBSUB creating millions of channels. */
dictDelete(c->ns->pubsub_channels,channel);
dictDelete(g_pserver->pubsub_channels,channel);
}
}
/* Notify the client */
@@ -205,10 +205,10 @@ int pubsubSubscribePattern(client *c, robj *pattern) {
listAddNodeTail(c->pubsub_patterns,pattern);
incrRefCount(pattern);
/* Add the client to the pattern -> list of clients hash table */
de = dictFind(c->ns->pubsub_patterns,pattern);
de = dictFind(g_pserver->pubsub_patterns,pattern);
if (de == NULL) {
clients = listCreate();
dictAdd(c->ns->pubsub_patterns,pattern,clients);
dictAdd(g_pserver->pubsub_patterns,pattern,clients);
incrRefCount(pattern);
} else {
clients = (list*)dictGetVal(de);
@@ -233,7 +233,7 @@ int pubsubUnsubscribePattern(client *c, robj *pattern, int notify) {
retval = 1;
listDelNode(c->pubsub_patterns,ln);
/* Remove the client from the pattern -> clients list hash table */
de = dictFind(c->ns->pubsub_patterns,pattern);
de = dictFind(g_pserver->pubsub_patterns,pattern);
serverAssertWithInfo(c,NULL,de != NULL);
clients = (list*)dictGetVal(de);
ln = listSearchKey(clients,c);
@@ -242,7 +242,7 @@ int pubsubUnsubscribePattern(client *c, robj *pattern, int notify) {
if (listLength(clients) == 0) {
/* Free the list and associated hash entry at all if this was
* the latest client. */
dictDelete(c->ns->pubsub_patterns,pattern);
dictDelete(g_pserver->pubsub_patterns,pattern);
}
}
/* Notify the client */
@@ -291,9 +291,8 @@ int pubsubUnsubscribeAllPatterns(client *c, int notify) {
}
/* Publish a message */
int pubsubPublishMessage(redisNamespace *ns, robj *channel, robj *message) {
int pubsubPublishMessage(robj *channel, robj *message) {
serverAssert(GlobalLocksAcquired());
serverAssert(ns);
int receivers = 0;
dictEntry *de;
dictIterator *di;
@@ -301,7 +300,7 @@ int pubsubPublishMessage(redisNamespace *ns, robj *channel, robj *message) {
listIter li;
/* Send to clients listening for that channel */
de = dictFind(ns->pubsub_channels,channel);
de = dictFind(g_pserver->pubsub_channels,channel);
if (de) {
list *list = reinterpret_cast<::list*>(dictGetVal(de));
listNode *ln;
@@ -321,7 +320,7 @@ int pubsubPublishMessage(redisNamespace *ns, robj *channel, robj *message) {
}
}
/* Send to clients listening to matching channels */
di = dictGetIterator(ns->pubsub_patterns);
di = dictGetIterator(g_pserver->pubsub_patterns);
if (di) {
channel = getDecodedObject(channel);
while((de = dictNext(di)) != NULL) {
@@ -424,9 +423,9 @@ void punsubscribeCommand(client *c) {
/* PUBLISH <channel> <message> */
void publishCommand(client *c) {
int receivers = pubsubPublishMessage(c->ns,c->argv[1],c->argv[2]);
int receivers = pubsubPublishMessage(c->argv[1],c->argv[2]);
if (g_pserver->cluster_enabled)
clusterPropagatePublish(c->ns,c->argv[1],c->argv[2]);
clusterPropagatePublish(c->argv[1],c->argv[2]);
else
forceCommandPropagation(c,PROPAGATE_REPL);
addReplyLongLong(c,receivers);
@@ -451,7 +450,7 @@ NULL
{
/* PUBSUB CHANNELS [<pattern>] */
sds pat = (c->argc == 2) ? NULL : szFromObj(c->argv[2]);
dictIterator *di = dictGetIterator(c->ns->pubsub_channels);
dictIterator *di = dictGetIterator(g_pserver->pubsub_channels);
dictEntry *de;
long mblen = 0;
void *replylen;
@@ -476,14 +475,14 @@ NULL
addReplyArrayLen(c,(c->argc-2)*2);
for (j = 2; j < c->argc; j++) {
list *l = (list*)dictFetchValue(c->ns->pubsub_channels,c->argv[j]);
list *l = (list*)dictFetchValue(g_pserver->pubsub_channels,c->argv[j]);
addReplyBulk(c,c->argv[j]);
addReplyLongLong(c,l ? listLength(l) : 0);
}
} else if (!strcasecmp(szFromObj(c->argv[1]),"numpat") && c->argc == 2) {
/* PUBSUB NUMPAT */
addReplyLongLong(c,dictSize(c->ns->pubsub_patterns));
addReplyLongLong(c,dictSize(g_pserver->pubsub_patterns));
} else {
addReplySubcommandSyntaxError(c);
}
+13 -56
View File
@@ -1283,7 +1283,6 @@ ssize_t rdbSaveSingleModuleAux(rio *rdb, int when, moduleType *mt) {
* error. */
int rdbSaveRio(rio *rdb, int *error, int rdbflags, rdbSaveInfo *rsi) {
dictIterator *di = NULL;
dictIterator *di_ns = NULL;
dictEntry *de;
char magic[10];
uint64_t cksum;
@@ -1303,8 +1302,7 @@ int rdbSaveRio(rio *rdb, int *error, int rdbflags, rdbSaveInfo *rsi) {
for (j = 0; j < cserver.dbnum; j++) {
redisDb *db = g_pserver->db+j;
dict *d = db->dict;
if (db->mapped_id < 0) continue;
serverAssert(db->ns);
if (dictSize(d) == 0) continue;
di = dictGetSafeIterator(d);
/* Write the SELECT DB opcode */
@@ -1318,13 +1316,10 @@ int rdbSaveRio(rio *rdb, int *error, int rdbflags, rdbSaveInfo *rsi) {
if (rdbSaveType(rdb,RDB_OPCODE_RESIZEDB) == -1) goto werr;
if (rdbSaveLen(rdb,db_size) == -1) goto werr;
if (rdbSaveLen(rdb,expires_size) == -1) goto werr;
if (rdbSaveAuxFieldStrStr(rdb,"keydb-namespace", db->ns->name) == -1) goto werr;
if (rdbSaveAuxFieldStrInt(rdb,"keydb-namespace-dbid", db->mapped_id) == -1) goto werr;
/* Iterate this DB writing every entry */
size_t ckeysExpired = 0;
while(db_size != 0 && (de = dictNext(di)) != NULL) {
while((de = dictNext(di)) != NULL) {
sds keystr = (sds)dictGetKey(de);
robj *o = (robj*)dictGetVal(de);
@@ -1354,27 +1349,15 @@ int rdbSaveRio(rio *rdb, int *error, int rdbflags, rdbSaveInfo *rsi) {
* the script cache as well: on successful PSYNC after a restart, we need
* to be able to process any EVALSHA inside the replication backlog the
* master will send us. */
if (rsi) {
redisNamespace *ns;
dictEntry *de_ns;
di_ns = dictGetSafeIterator(g_pserver->namespaces);
while((de_ns = dictNext(di_ns)) != NULL) {
ns = (struct redisNamespace *) dictGetVal(de_ns);
if (dictSize(ns->lua_scripts)) {
di = dictGetIterator(ns->lua_scripts);
while ((de = dictNext(di)) != NULL) {
robj *body = (robj *) dictGetVal(de);
if (rdbSaveAuxField(rdb, "keydb-namespace-lua", 19, ns->name, sdslen(ns->name)) == -1)
goto werr;
if (rdbSaveAuxField(rdb, "lua", 3, szFromObj(body), sdslen(szFromObj(body))) == -1)
goto werr;
}
dictReleaseIterator(di);
di = NULL; /* So that we don't release it again on error. */
}
if (rsi && dictSize(g_pserver->lua_scripts)) {
di = dictGetIterator(g_pserver->lua_scripts);
while((de = dictNext(di)) != NULL) {
robj *body = (robj*)dictGetVal(de);
if (rdbSaveAuxField(rdb,"lua",3,szFromObj(body),sdslen(szFromObj(body))) == -1)
goto werr;
}
dictReleaseIterator(di_ns);
di_ns = NULL;
dictReleaseIterator(di);
di = NULL; /* So that we don't release it again on error. */
}
if (rdbSaveModulesAux(rdb, REDISMODULE_AUX_AFTER_RDB) == -1) goto werr;
@@ -1392,7 +1375,6 @@ int rdbSaveRio(rio *rdb, int *error, int rdbflags, rdbSaveInfo *rsi) {
werr:
if (error) *error = errno;
if (di) dictReleaseIterator(di);
if (di_ns) dictReleaseIterator(di_ns);
return C_ERR;
}
@@ -2569,7 +2551,6 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
bool fLastKeyExpired = false;
int error;
long long empty_keys_skipped = 0, expired_keys_skipped = 0, keys_loaded = 0;
client *lua_client = nullptr;
rdb->update_cksum = rdbLoadProgressCallback;
rdb->max_processing_chunk = g_pserver->loading_process_events_interval_bytes;
@@ -2589,7 +2570,7 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
now = mstime();
lru_clock = LRU_CLOCK();
lua_client = createClient(nullptr, IDX_EVENT_LOOP_MAIN);
while(1) {
robj *val;
@@ -2637,7 +2618,6 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
exit(1);
}
db = g_pserver->db+dbid;
db->ns = g_pserver->default_namespace;
continue; /* Read next opcode. */
} else if (type == RDB_OPCODE_RESIZEDB) {
/* RESIZEDB: Hint about the size of the keys in the currently
@@ -2677,7 +2657,7 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
if (rsi) rsi->repl_offset = strtoll(szFromObj(auxval),NULL,10);
} else if (!strcasecmp(szFromObj(auxkey),"lua")) {
/* Load the script back in memory. */
if (luaCreateFunction(lua_client, lua_client->ns->lua, auxval) == NULL) {
if (luaCreateFunction(NULL,g_pserver->lua,auxval) == NULL) {
rdbReportCorruptRDB(
"Can't load Lua script from RDB file! "
"BODY: %s", (char*)ptrFromObj(auxval));
@@ -2730,25 +2710,6 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
decrRefCount(subexpireKey);
subexpireKey = nullptr;
}
} else if (!strcasecmp(szFromObj(auxkey), "keydb-namespace")) {
db->ns = getNamespace(szFromObj(auxval));
} else if (!strcasecmp(szFromObj(auxkey), "keydb-namespace-lua")) {
lua_client->ns = getNamespace(szFromObj(auxval));
} else if (!strcasecmp(szFromObj(auxkey), "keydb-namespace-dbid")) {
long long max_db = std::min(cserver.dbnum,cserver.ns_dbnum);
long long mapped_id = strtoll(szFromObj(auxval),NULL,10);
serverAssert(db->ns);
serverAssert(mapped_id >= 0);
if (mapped_id >= max_db) {
serverLog(LL_WARNING,
"FATAL: Data file was created with a KeyDB "
"server configured to handle more than %d "
"namespaced databases. Exiting\n", (int)max_db);
exit(1);
}
db->mapped_id = (int) mapped_id;
db->ns->db[db->mapped_id] = db;
} else {
/* We ignore fields we don't understand, as by AUX field
* contract. */
@@ -2925,8 +2886,6 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
subexpireKey = nullptr;
}
freeClient(lua_client);
/* Verify the checksum if RDB version is >= 5 */
if (rdbver >= 5) {
uint64_t cksum, expected = rdb->cksum;
@@ -2973,8 +2932,6 @@ eoferr:
subexpireKey = nullptr;
}
freeClient(lua_client);
serverLog(LL_WARNING,
"Short read or OOM loading DB. Unrecoverable error, aborting now.");
rdbReportReadError("Unexpected EOF reading RDB file");
-1
View File
@@ -1984,7 +1984,6 @@ int main(int argc, const char **argv) {
} while(config.loop);
zfree(data);
zfree(data);
if (config.redis_config != NULL) freeRedisConfig(config.redis_config);
return 0;
+19 -31
View File
@@ -3285,8 +3285,7 @@ void freeMasterInfo(redisMaster *mi)
zfree(mi->masteruser);
if (mi->repl_transfer_tmpfile)
zfree(mi->repl_transfer_tmpfile);
if (mi->staleKeyMap != nullptr)
delete mi->staleKeyMap;
delete mi->staleKeyMap;
if (mi->cached_master != nullptr)
freeClientAsync(mi->cached_master);
if (mi->master != nullptr)
@@ -3793,10 +3792,10 @@ void refreshGoodSlavesCount(void) {
*/
/* Initialize the script cache, only called at startup. */
void replicationScriptCacheInit(redisNamespace *ns) {
ns->repl_scriptcache_size = 10000;
ns->repl_scriptcache_dict = dictCreate(&replScriptCacheDictType,NULL);
ns->repl_scriptcache_fifo = listCreate();
void replicationScriptCacheInit(void) {
g_pserver->repl_scriptcache_size = 10000;
g_pserver->repl_scriptcache_dict = dictCreate(&replScriptCacheDictType,NULL);
g_pserver->repl_scriptcache_fifo = listCreate();
}
/* Empty the script cache. Should be called every time we are no longer sure
@@ -3811,49 +3810,38 @@ void replicationScriptCacheInit(redisNamespace *ns) {
* to reclaim otherwise unused memory.
*/
void replicationScriptCacheFlush(void) {
redisNamespace *ns;
dictEntry *de;
dictIterator *di;
di = dictGetSafeIterator(g_pserver->namespaces);
while((de = dictNext(di)) != NULL) {
ns = (struct redisNamespace *) dictGetVal(de);
if (listLength(ns->repl_scriptcache_fifo) != 0) {
dictEmpty(ns->repl_scriptcache_dict, NULL);
listRelease(ns->repl_scriptcache_fifo);
ns->repl_scriptcache_fifo = listCreate();
}
}
dictReleaseIterator(di);
dictEmpty(g_pserver->repl_scriptcache_dict,NULL);
listRelease(g_pserver->repl_scriptcache_fifo);
g_pserver->repl_scriptcache_fifo = listCreate();
}
/* Add an entry into the script cache, if we reach max number of entries the
* oldest is removed from the list. */
void replicationScriptCacheAdd(redisNamespace *ns, sds sha1) {
void replicationScriptCacheAdd(sds sha1) {
int retval;
sds key = sdsdup(sha1);
/* Evict oldest. */
if (listLength(ns->repl_scriptcache_fifo) == ns->repl_scriptcache_size)
if (listLength(g_pserver->repl_scriptcache_fifo) == g_pserver->repl_scriptcache_size)
{
listNode *ln = listLast(ns->repl_scriptcache_fifo);
listNode *ln = listLast(g_pserver->repl_scriptcache_fifo);
sds oldest = (sds)listNodeValue(ln);
retval = dictDelete(ns->repl_scriptcache_dict,oldest);
retval = dictDelete(g_pserver->repl_scriptcache_dict,oldest);
serverAssert(retval == DICT_OK);
listDelNode(ns->repl_scriptcache_fifo,ln);
listDelNode(g_pserver->repl_scriptcache_fifo,ln);
}
/* Add current. */
retval = dictAdd(ns->repl_scriptcache_dict,key,NULL);
listAddNodeHead(ns->repl_scriptcache_fifo,key);
retval = dictAdd(g_pserver->repl_scriptcache_dict,key,NULL);
listAddNodeHead(g_pserver->repl_scriptcache_fifo,key);
serverAssert(retval == DICT_OK);
}
/* Returns non-zero if the specified entry exists inside the cache, that is,
* if all the slaves are aware of this script SHA1. */
int replicationScriptCacheExists(redisNamespace *ns, sds sha1) {
return dictFind(ns->repl_scriptcache_dict,sha1) != NULL;
int replicationScriptCacheExists(sds sha1) {
return dictFind(g_pserver->repl_scriptcache_dict,sha1) != NULL;
}
/* ----------------------- SYNCHRONOUS REPLICATION --------------------------
@@ -4227,7 +4215,8 @@ void replicationCron(void) {
* free our Replication Script Cache as there is no need to propagate
* EVALSHA at all. */
if (listLength(g_pserver->slaves) == 0 &&
g_pserver->aof_state == AOF_OFF)
g_pserver->aof_state == AOF_OFF &&
listLength(g_pserver->repl_scriptcache_fifo) != 0)
{
replicationScriptCacheFlush();
}
@@ -4708,7 +4697,6 @@ void replicaReplayCommand(client *c)
cFake->lock.lock();
cFake->authenticated = c->authenticated;
cFake->user = c->user;
cFake->ns = c->ns;
cFake->querybuf = sdscatsds(cFake->querybuf,(sds)ptrFromObj(c->argv[2]));
selectDb(cFake, c->db->id);
auto ccmdPrev = serverTL->commandsExecuted;
+38 -41
View File
@@ -605,7 +605,6 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) {
c->argv = argv;
c->argc = argc;
c->user = g_pserver->lua_caller->user;
c->ns = g_pserver->lua_caller->ns;
/* Process module hooks */
moduleCallCommandFilters(c);
@@ -1145,31 +1144,31 @@ void scriptingEnableGlobalsProtection(lua_State *lua) {
* in order to reset the Lua scripting environment.
*
* However it is simpler to just call scriptingReset() that does just that. */
void scriptingSetup() {
for (int iel = 0; iel < cserver.cthreads; ++iel)
{
g_pserver->rgthreadvar[iel].lua_client = createClient(nullptr, iel);
g_pserver->rgthreadvar[iel].lua_client->flags |= CLIENT_LUA;
/* We do not want to allow blocking commands inside Lua */
g_pserver->rgthreadvar[iel].lua_client->flags |= CLIENT_DENY_BLOCKING;
}
g_pserver->lua_timedout = 0;
g_pserver->lua_caller = NULL;
g_pserver->lua_cur_script = NULL;
ldbInit();
}
void scriptingInit(redisNamespace *ns) {
void scriptingInit(int setup) {
lua_State *lua = lua_open();
if (setup) {
for (int iel = 0; iel < cserver.cthreads; ++iel)
{
g_pserver->rgthreadvar[iel].lua_client = createClient(nullptr, iel);
g_pserver->rgthreadvar[iel].lua_client->flags |= CLIENT_LUA;
/* We do not want to allow blocking commands inside Lua */
g_pserver->rgthreadvar[iel].lua_client->flags |= CLIENT_DENY_BLOCKING;
}
g_pserver->lua_timedout = 0;
g_pserver->lua_caller = NULL;
g_pserver->lua_cur_script = NULL;
ldbInit();
}
luaLoadLibraries(lua);
luaRemoveUnsupportedFunctions(lua);
/* Initialize a dictionary we use to map SHAs to scripts.
* This is useful for replication, as we need to replicate EVALSHA
* as EVAL, so we need to remember the associated script. */
ns->lua_scripts = dictCreate(&shaScriptObjectDictType,NULL);
ns->lua_scripts_mem = 0;
g_pserver->lua_scripts = dictCreate(&shaScriptObjectDictType,NULL);
g_pserver->lua_scripts_mem = 0;
/* Register the redis commands table and fields */
lua_newtable(lua);
@@ -1317,23 +1316,23 @@ void scriptingInit(redisNamespace *ns) {
* to global variables. */
scriptingEnableGlobalsProtection(lua);
ns->lua = lua;
g_pserver->lua = lua;
}
/* Release resources related to Lua scripting.
* This function is used in order to reset the scripting environment. */
void scriptingRelease(redisNamespace *ns, int async) {
void scriptingRelease(int async) {
if (async)
freeLuaScriptsAsync(ns->lua_scripts);
freeLuaScriptsAsync(g_pserver->lua_scripts);
else
dictRelease(ns->lua_scripts);
ns->lua_scripts_mem = 0;
lua_close(ns->lua);
dictRelease(g_pserver->lua_scripts);
g_pserver->lua_scripts_mem = 0;
lua_close(g_pserver->lua);
}
void scriptingReset(redisNamespace *ns, int async) {
scriptingRelease(ns, async);
scriptingInit(ns);
void scriptingReset(int async) {
scriptingRelease(async);
scriptingInit(0);
}
/* Set an array of Redis String Objects as a Lua array (table) stored into a
@@ -1422,7 +1421,7 @@ sds luaCreateFunction(client *c, lua_State *lua, robj *body) {
sha1hex(funcname+2,(char*)ptrFromObj(body),sdslen((sds)ptrFromObj(body)));
sds sha = sdsnewlen(funcname+2,40);
if ((de = dictFind(c->ns->lua_scripts,sha)) != NULL) {
if ((de = dictFind(g_pserver->lua_scripts,sha)) != NULL) {
sdsfree(sha);
return (sds)dictGetKey(de);
}
@@ -1460,9 +1459,9 @@ sds luaCreateFunction(client *c, lua_State *lua, robj *body) {
/* We also save a SHA1 -> Original script map in a dictionary
* so that we can replicate / write in the AOF all the
* EVALSHA commands as EVAL using the original script. */
int retval = dictAdd(c->ns->lua_scripts,sha,body);
int retval = dictAdd(g_pserver->lua_scripts,sha,body);
serverAssertWithInfo(c ? c : serverTL->lua_client,NULL,retval == DICT_OK);
c->ns->lua_scripts_mem += sdsZmallocSize(sha) + getStringObjectSdsUsedMemory(body);
g_pserver->lua_scripts_mem += sdsZmallocSize(sha) + getStringObjectSdsUsedMemory(body);
incrRefCount(body);
return sha;
}
@@ -1523,7 +1522,7 @@ void resetLuaClient(void) {
}
void evalGenericCommand(client *c, int evalsha) {
lua_State *lua = c->ns->lua;
lua_State *lua = g_pserver->lua;
char funcname[43];
long long numkeys;
long long initial_server_dirty = g_pserver->dirty;
@@ -1627,7 +1626,7 @@ void evalGenericCommand(client *c, int evalsha) {
lua_sethook(lua,luaMaskCountHook,LUA_MASKCOUNT,100000);
delhook = 1;
} else if (ldb.active) {
lua_sethook(c->ns->lua,luaLdbLineHook,LUA_MASKLINE|LUA_MASKCOUNT,100000);
lua_sethook(g_pserver->lua,luaLdbLineHook,LUA_MASKLINE|LUA_MASKCOUNT,100000);
delhook = 1;
}
@@ -1710,13 +1709,13 @@ void evalGenericCommand(client *c, int evalsha) {
* flush our cache of scripts that can be replicated as EVALSHA, while
* for AOF we need to do so every time we rewrite the AOF file. */
if (evalsha && !g_pserver->lua_replicate_commands) {
if (!replicationScriptCacheExists(c->ns, (sds)ptrFromObj(c->argv[1]))) {
if (!replicationScriptCacheExists((sds)ptrFromObj(c->argv[1]))) {
/* This script is not in our script cache, replicate it as
* EVAL, then add it into the script cache, as from now on
* slaves and AOF know about it. */
robj *script = (robj*)dictFetchValue(c->ns->lua_scripts,ptrFromObj(c->argv[1]));
robj *script = (robj*)dictFetchValue(g_pserver->lua_scripts,ptrFromObj(c->argv[1]));
replicationScriptCacheAdd(c->ns, (sds)ptrFromObj(c->argv[1]));
replicationScriptCacheAdd((sds)ptrFromObj(c->argv[1]));
serverAssertWithInfo(c,NULL,script != NULL);
/* If the script did not produce any changes in the dataset we want
@@ -1799,7 +1798,7 @@ NULL
addReplyError(c,"SCRIPT FLUSH only support SYNC|ASYNC option");
return;
}
scriptingReset(c->ns, async);
scriptingReset(async);
addReply(c,shared.ok);
replicationScriptCacheFlush();
g_pserver->dirty++; /* Propagating this command is a good idea. */
@@ -1808,13 +1807,13 @@ NULL
addReplyArrayLen(c, c->argc-2);
for (j = 2; j < c->argc; j++) {
if (dictFind(c->ns->lua_scripts,ptrFromObj(c->argv[j])))
if (dictFind(g_pserver->lua_scripts,ptrFromObj(c->argv[j])))
addReply(c,shared.cone);
else
addReply(c,shared.czero);
}
} else if (c->argc == 3 && !strcasecmp((const char*)ptrFromObj(c->argv[1]),"load")) {
sds sha = luaCreateFunction(c, c->ns->lua,c->argv[2]);
sds sha = luaCreateFunction(c,g_pserver->lua,c->argv[2]);
if (sha == NULL) return; /* The error was sent by luaCreateFunction(). */
addReplyBulkCBuffer(c,sha,40);
forceCommandPropagation(c,PROPAGATE_REPL|PROPAGATE_AOF);
@@ -1823,8 +1822,6 @@ NULL
addReplyError(c,"-NOTBUSY No scripts in execution right now.");
} else if (g_pserver->lua_caller->flags & CLIENT_MASTER) {
addReplyError(c,"-UNKILLABLE The busy script was sent by a master instance in the context of replication and cannot be killed.");
} else if (g_pserver->lua_caller->ns != c->ns) {
addReplyError(c,"-UNKILLABLE Script is not yours to kill");
} else if (g_pserver->lua_write_dirty) {
addReplyError(c,"-UNKILLABLE Sorry the script already executed write commands against the dataset. You can either wait the script termination or kill the server in a hard way using the SHUTDOWN NOSAVE command.");
} else {
+1 -1
View File
@@ -734,7 +734,7 @@ void sentinelEvent(int level, const char *type, sentinelRedisInstance *ri,
if (level != LL_DEBUG) {
channel = createStringObject(type,strlen(type));
payload = createStringObject(msg,strlen(msg));
pubsubPublishMessage(g_pserver->default_namespace, channel,payload);
pubsubPublishMessage(channel,payload);
decrRefCount(channel);
decrRefCount(payload);
}
+67 -182
View File
@@ -63,6 +63,7 @@
#include <algorithm>
#include <uuid/uuid.h>
#include <mutex>
#include <condition_variable>
#include "aelocker.h"
#include "motd.h"
#include "t_nhash.h"
@@ -94,6 +95,10 @@ struct redisServer server; /* Server global state */
redisServer *g_pserver = &GlobalHidden::server;
struct redisServerConst cserver;
__thread struct redisServerThreadVars *serverTL = NULL; // thread local server vars
std::mutex time_thread_mutex;
std::condition_variable time_thread_cv;
int sleeping_threads = 0;
void wakeTimeThread();
/* Our command table.
*
@@ -1169,15 +1174,7 @@ struct redisCommand redisCommandTable[] = {
{"failover",failoverCommand,-1,
"admin no-script ok-stale",
0,NULL,0,0,0,0,0,0},
{"allocate",allocateCommand,4,
"ok-loading fast ok-stale @keyspace",
0,NULL,0,0,0,0,0,0},
{"selectns",selectNsCommand,3,
"admin ok-loading fast ok-stale @keyspace",
0,NULL,0,0,0,0,0,0},
0,NULL,0,0,0,0,0,0}
};
/*============================ Utility functions ============================ */
@@ -1610,18 +1607,6 @@ dictType modulesDictType = {
NULL /* allow to expand */
};
/* Namespaces dictionary type. Keys are namespace name,
* values are pointer to redisNamespace struct. */
dictType namespaceDictType = {
dictSdsCaseHash, /* hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCaseCompare, /* key compare */
dictSdsDestructor, /* key destructor */
NULL, /* val destructor */
NULL /* allow to expand */
};
/* Migrate cache dict type. */
dictType migrateCacheDictType = {
dictSdsHash, /* hash function */
@@ -2027,7 +2012,7 @@ void databasesCron(bool fMainThread) {
* info or not using the 'update_daylight_info' argument. Normally we update
* such info only when calling this function from serverCron() but not when
* calling it from call(). */
void updateCachedTime(int update_daylight_info) {
void updateCachedTime() {
long long t = ustime();
__atomic_store(&g_pserver->ustime, &t, __ATOMIC_RELAXED);
t /= 1000;
@@ -2040,12 +2025,10 @@ void updateCachedTime(int update_daylight_info) {
* context is safe since we will never fork() while here, in the main
* thread. The logging function will call a thread safe version of
* localtime that has no locks. */
if (update_daylight_info) {
struct tm tm;
time_t ut = g_pserver->unixtime;
localtime_r(&ut,&tm);
__atomic_store(&g_pserver->daylight_active, &tm.tm_isdst, __ATOMIC_RELAXED);
}
struct tm tm;
time_t ut = g_pserver->unixtime;
localtime_r(&ut,&tm);
__atomic_store(&g_pserver->daylight_active, &tm.tm_isdst, __ATOMIC_RELAXED);
}
void checkChildrenDone(void) {
@@ -2124,17 +2107,8 @@ void cronUpdateMemoryStats() {
/* LUA memory isn't part of zmalloc_used, but it is part of the process RSS,
* so we must deduct it in order to be able to calculate correct
* "allocator fragmentation" ratio */
g_pserver->cron_malloc_stats.allocator_resident = g_pserver->cron_malloc_stats.process_rss;
redisNamespace *ns;
dictEntry *de;
dictIterator *di;
di = dictGetSafeIterator(g_pserver->namespaces);
while((de = dictNext(di)) != NULL) {
ns = (struct redisNamespace *) dictGetVal(de);
g_pserver->cron_malloc_stats.allocator_resident -= lua_gc(ns->lua,LUA_GCCOUNT,0)*1024LL;
}
dictReleaseIterator(di);
size_t lua_memory = lua_gc(g_pserver->lua,LUA_GCCOUNT,0)*1024LL;
g_pserver->cron_malloc_stats.allocator_resident = g_pserver->cron_malloc_stats.process_rss - lua_memory;
}
if (!g_pserver->cron_malloc_stats.allocator_active)
g_pserver->cron_malloc_stats.allocator_active = g_pserver->cron_malloc_stats.allocator_resident;
@@ -2181,9 +2155,6 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
* handler if we don't return here fast enough. */
if (g_pserver->watchdog_period) watchdogScheduleSignal(g_pserver->watchdog_period);
/* Update the time cache. */
updateCachedTime(1);
g_pserver->hz = g_pserver->config_hz;
/* Adapt the g_pserver->hz value to the number of configured clients. If we have
* many clients, we want to call serverCron() with an higher frequency. */
@@ -2441,7 +2412,6 @@ int serverCronLite(struct aeEventLoop *eventLoop, long long id, void *clientData
void blockingOperationStarts() {
if(!g_pserver->blocking_op_nesting++){
updateCachedTime(0);
g_pserver->blocked_last_cron = g_pserver->mstime;
}
}
@@ -2652,6 +2622,14 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
locker.disarm();
if (!fSentReplies)
handleClientsWithPendingWrites(iel, aof_state);
// Scope lock_guard
{
std::lock_guard<std::mutex> lock(time_thread_mutex);
sleeping_threads++;
serverAssert(sleeping_threads <= cserver.cthreads);
}
/* Determine whether the modules are enabled before sleeping, and use that result
both here, and after wakeup to avoid double acquire or release of the GIL */
serverTL->modulesEnabledThisAeLoop = !!moduleCount();
@@ -2671,6 +2649,7 @@ void afterSleep(struct aeEventLoop *eventLoop) {
Don't check here that modules are enabled, rather use the result from beforeSleep
Otherwise you may double acquire the GIL and cause deadlocks in the module */
if (!ProcessingEventsWhileBlocked) {
wakeTimeThread();
if (serverTL->modulesEnabledThisAeLoop) moduleAcquireGIL(TRUE /*fServerThread*/);
}
}
@@ -2762,7 +2741,6 @@ void createSharedObjects(void) {
"*2\r\n$6\r\nSELECT\r\n$%d\r\n%s\r\n",
dictid_len, dictid_str)));
}
shared.messagebulk = makeObjectShared("$7\r\nmessage\r\n",13);
shared.pmessagebulk = makeObjectShared("$8\r\npmessage\r\n",14);
shared.subscribebulk = makeObjectShared("$9\r\nsubscribe\r\n",15);
@@ -2771,7 +2749,6 @@ void createSharedObjects(void) {
shared.punsubscribebulk = makeObjectShared("$12\r\npunsubscribe\r\n",19);
/* Shared command names */
shared.allocate = makeObjectShared("ALLOCATE", 8);
shared.del = makeObjectShared("DEL",3);
shared.unlink = makeObjectShared("UNLINK",6);
shared.rpop = makeObjectShared("RPOP",4);
@@ -2870,7 +2847,7 @@ void initMasterInfo(redisMaster *master)
void initServerConfig(void) {
int j;
updateCachedTime(true);
updateCachedTime();
getRandomHexChars(g_pserver->runid,CONFIG_RUN_ID_SIZE);
g_pserver->runid[CONFIG_RUN_ID_SIZE] = '\0';
changeReplicationId();
@@ -2937,7 +2914,6 @@ void initServerConfig(void) {
/* Replication related */
g_pserver->masters = listCreate();
g_pserver->repl_init_masters = listCreate();
g_pserver->enable_multimaster = CONFIG_DEFAULT_ENABLE_MULTIMASTER;
g_pserver->repl_syncio_timeout = CONFIG_REPL_SYNCIO_TIMEOUT;
g_pserver->master_repl_offset = 0;
@@ -2994,7 +2970,6 @@ void initServerConfig(void) {
cserver.hdelCommand = lookupCommandByCString("hdel");
cserver.zremCommand = lookupCommandByCString("zrem");
cserver.lmoveCommand = lookupCommandByCString("lmove");
cserver.allocateCommand = lookupCommandByCString("allocate");
/* Debugging */
g_pserver->watchdog_period = 0;
@@ -3381,16 +3356,6 @@ void resetServerStats(void) {
g_pserver->stat_total_error_replies = 0;
g_pserver->stat_dump_payload_sanitizations = 0;
g_pserver->aof_delayed_fsync = 0;
redisNamespace *ns;
dictEntry *de;
dictIterator *di;
di = dictGetSafeIterator(g_pserver->namespaces);
while((de = dictNext(di)) != NULL) {
ns = (struct redisNamespace *) dictGetVal(de);
ns->stat_numcommands = 0;
}
dictReleaseIterator(di);
}
/* Make the thread killable at any time, so that kill threads functions
@@ -3515,47 +3480,13 @@ static void initServerThread(struct redisServerThreadVars *pvar, int fMain)
}
}
sds generateAutoNamespaceName() {
char uuid[37];
uuid_t uuid_bin;
uuid_generate(uuid_bin);
uuid_unparse(uuid_bin, uuid);
void initServer(void) {
signal(SIGHUP, SIG_IGN);
signal(SIGPIPE, SIG_IGN);
setupSignalHandlers();
makeThreadKillable();
//TODO: for readability we might want to add username
// but because we are operating on a fake user here
// the actual user name is not available.
sds auto_ns = sdsempty();
return sdscatfmt(auto_ns, "::%s", uuid);
}
redisNamespace *getNamespace(const char* name) {
sds ns_name = sdsnew(name);
redisNamespace *ns = (redisNamespace *) dictFetchValue(g_pserver->namespaces, ns_name);
if (ns == NULL) {
ns = (redisNamespace *)zmalloc(sizeof(redisNamespace), MALLOC_LOCAL);
ns = new(ns) redisNamespace;
ns->name = sdsnew(ns_name);
ns->pubsub_channels = dictCreate(&keylistDictType,NULL);
ns->pubsub_patterns = dictCreate(&keylistDictType,NULL);
ns->db = (redisDb**)zcalloc(sizeof(void *)*std::min(cserver.dbnum,cserver.ns_dbnum), MALLOC_LOCAL);
ns->stat_numcommands = 0;
ns->stat_commands = (redisNamespaceCommandStats*)zcalloc(sizeof(redisNamespaceCommandStats)*dictSize(g_pserver->commands), MALLOC_LOCAL);
resetCommandTableStatsNs(ns);
scriptingInit(ns);
replicationScriptCacheInit(ns);
dictAdd(g_pserver->namespaces, sdsnew(ns_name), ns);
}
sdsfree(ns_name);
return ns;
}
void initDb(void) {
g_pserver->namespaces = dictCreate(&namespaceDictType,NULL);
g_pserver->default_namespace = getNamespace("::");
g_pserver->db = (redisDb*)zmalloc(sizeof(redisDb)*cserver.dbnum, MALLOC_LOCAL);
g_pserver->last_allocated_db = -1;
/* Create the Redis databases, and initialize other internal state. */
for (int j = 0; j < cserver.dbnum; j++) {
@@ -3567,20 +3498,11 @@ void initDb(void) {
g_pserver->db[j].ready_keys = dictCreate(&objectKeyPointerValueDictType,NULL);
g_pserver->db[j].watched_keys = dictCreate(&keylistDictType,NULL);
g_pserver->db[j].id = j;
g_pserver->db[j].mapped_id = -1;
g_pserver->db[j].avg_ttl = 0;
g_pserver->db[j].last_expire_set = 0;
g_pserver->db[j].defrag_later = listCreate();
g_pserver->db[j].ns = nullptr;
listSetFreeMethod(g_pserver->db[j].defrag_later,(void (*)(const void*))sdsfree);
}
}
void initServer(void) {
signal(SIGHUP, SIG_IGN);
signal(SIGPIPE, SIG_IGN);
setupSignalHandlers();
makeThreadKillable();
/* Fixup Master Client Database */
listIter li;
@@ -3627,6 +3549,8 @@ void initServer(void) {
serverLog(LL_NOTICE, "monotonic clock: %s", clk_msg);
evictionPoolAlloc(); /* Initialize the LRU keys pool. */
g_pserver->pubsub_channels = dictCreate(&keylistDictType,NULL);
g_pserver->pubsub_patterns = dictCreate(&keylistDictType,NULL);
g_pserver->cronloops = 0;
g_pserver->propagate_in_transaction = 0;
g_pserver->client_pause_in_transaction = 0;
@@ -3709,7 +3633,8 @@ void initServer(void) {
uuid_generate((unsigned char*)cserver.uuid);
if (g_pserver->cluster_enabled) clusterInit();
scriptingSetup();
replicationScriptCacheInit();
scriptingInit(1);
slowlogInit();
latencyMonitorInit();
@@ -3821,7 +3746,6 @@ void populateCommandTable(void) {
void resetCommandTableStats(void) {
struct redisCommand *c;
redisNamespace *ns;
dictEntry *de;
dictIterator *di;
@@ -3835,28 +3759,6 @@ void resetCommandTableStats(void) {
}
dictReleaseIterator(di);
di = dictGetSafeIterator(g_pserver->namespaces);
while((de = dictNext(di)) != NULL) {
ns = (redisNamespace *) dictGetVal(de);
resetCommandTableStatsNs(ns);
}
dictReleaseIterator(di);
}
void resetCommandTableStatsNs(redisNamespace *ns) {
struct redisCommand *c;
dictEntry *de;
dictIterator *di;
di = dictGetSafeIterator(g_pserver->commands);
while((de = dictNext(di)) != NULL) {
c = (struct redisCommand *) dictGetVal(de);
ns->stat_commands[c->id].microseconds = 0;
ns->stat_commands[c->id].calls = 0;
ns->stat_commands[c->id].rejected_calls = 0;
ns->stat_commands[c->id].failed_calls = 0;
}
dictReleaseIterator(di);
}
static void zfree_noconst(void *p) {
@@ -4112,7 +4014,6 @@ void call(client *c, int flags) {
prev_err_count = g_pserver->stat_total_error_replies;
g_pserver->fixed_time_expire++;
updateCachedTime(0);
incrementMvccTstamp();
elapsedStart(&call_timer);
try {
@@ -4136,7 +4037,6 @@ void call(client *c, int flags) {
* the same error twice. */
if ((g_pserver->stat_total_error_replies - prev_err_count) > 0) {
real_cmd->failed_calls++;
c->ns->stat_commands[real_cmd->id].failed_calls++;
}
/* After executing command, we will close the client after writing entire
@@ -4196,8 +4096,6 @@ void call(client *c, int flags) {
if (flags & CMD_CALL_STATS) {
real_cmd->microseconds += duration;
real_cmd->calls++;
c->ns->stat_commands[real_cmd->id].microseconds += duration;
c->ns->stat_commands[real_cmd->id].calls++;
}
/* Propagate the command into the AOF and replication link */
@@ -4303,7 +4201,6 @@ void call(client *c, int flags) {
}
g_pserver->stat_numcommands++;
c->ns->stat_numcommands++;
serverTL->fixed_time_expire--;
prev_err_count = g_pserver->stat_total_error_replies;
@@ -5220,7 +5117,7 @@ sds genRedisInfoString(const char *section) {
size_t zmalloc_used = zmalloc_used_memory();
size_t total_system_mem = cserver.system_memory_size;
const char *evict_policy = evictPolicyToString();
long long memory_lua = g_pserver->default_namespace->lua ? (long long)lua_gc(g_pserver->default_namespace->lua,LUA_GCCOUNT,0)*1024 : 0; //TODO: all namespaces
long long memory_lua = g_pserver->lua ? (long long)lua_gc(g_pserver->lua,LUA_GCCOUNT,0)*1024 : 0;
struct redisMemOverhead *mh = getMemoryOverheadData();
/* Peak memory is updated from time to time by serverCron() so it
@@ -5302,7 +5199,7 @@ sds genRedisInfoString(const char *section) {
used_memory_lua_hmem,
(long long) mh->lua_caches,
used_memory_scripts_hmem,
dictSize(g_pserver->default_namespace->lua_scripts),
dictSize(g_pserver->lua_scripts),
g_pserver->maxmemory,
maxmemory_hmem,
evict_policy,
@@ -5519,8 +5416,8 @@ sds genRedisInfoString(const char *section) {
g_pserver->stat_evictedkeys,
g_pserver->stat_keyspace_hits,
g_pserver->stat_keyspace_misses,
dictSize(g_pserver->default_namespace->pubsub_channels), //TODO: all namespaces
dictSize(g_pserver->default_namespace->pubsub_patterns), //TODO: all namespaces
dictSize(g_pserver->pubsub_channels),
dictSize(g_pserver->pubsub_patterns),
g_pserver->stat_fork_time,
g_pserver->stat_total_forks,
dictSize(g_pserver->migrate_cached_sockets),
@@ -5767,31 +5664,6 @@ sds genRedisInfoString(const char *section) {
if (tmpsafe != NULL) zfree(tmpsafe);
}
dictReleaseIterator(di);
dictIterator *di_ns;
dictEntry *de_ns;
di_ns = dictGetSafeIterator(g_pserver->namespaces);
while((de_ns = dictNext(di_ns)) != NULL) {
redisNamespace *ns = (redisNamespace *) dictGetVal(de_ns);
if (sections++) info = sdscat(info, "\r\n");
info = sdscatprintf(info, "# Commandstats %s\r\n", ns->name);
di = dictGetSafeIterator(g_pserver->commands);
while ((de = dictNext(di)) != NULL) {
char *tmpsafe;
c = (struct redisCommand *) dictGetVal(de);
if (!c->calls && !c->failed_calls && !c->rejected_calls)
continue;
info = sdscatprintf(info,
"cmdstat_%s:calls=%lld,usec=%lld,usec_per_call=%.2f"
",rejected_calls=%lld,failed_calls=%lld\r\n",
getSafeInfoString(c->name, strlen(c->name), &tmpsafe), c->calls, c->microseconds,
(c->calls == 0) ? 0 : ((float) c->microseconds / c->calls),
c->rejected_calls, c->failed_calls);
if (tmpsafe != NULL) zfree(tmpsafe);
}
dictReleaseIterator(di);
}
dictReleaseIterator(di_ns);
}
/* Error statistics */
if (allsections || defsections || !strcasecmp(section,"errorstats")) {
@@ -6669,6 +6541,35 @@ void OnTerminate()
serverPanic("std::teminate() called");
}
void wakeTimeThread() {
updateCachedTime();
std::lock_guard<std::mutex> lock(time_thread_mutex);
if (sleeping_threads >= cserver.cthreads)
time_thread_cv.notify_one();
sleeping_threads--;
serverAssert(sleeping_threads >= 0);
}
void *timeThreadMain(void*) {
timespec delay;
delay.tv_sec = 0;
delay.tv_nsec = 100;
while (true) {
{
std::unique_lock<std::mutex> lock(time_thread_mutex);
if (sleeping_threads >= cserver.cthreads) {
time_thread_cv.wait(lock);
}
}
updateCachedTime();
#if defined(__APPLE__)
nanosleep(&delay, nullptr);
#else
clock_nanosleep(CLOCK_MONOTONIC, 0, &delay, NULL);
#endif
}
}
void *workerThreadMain(void *parg)
{
int iel = (int)((int64_t)parg);
@@ -6952,24 +6853,6 @@ int main(int argc, char **argv) {
validateConfiguration();
initDb();
if (listLength(g_pserver->repl_init_masters)) {
listIter li;
listNode *ln;
listRewind(g_pserver->repl_init_masters, &li);
while ((ln = listNext(&li)))
{
struct redisMasterConnInfo *mi = (struct redisMasterConnInfo*)listNodeValue(ln);
replicationAddMaster(mi->ip, mi->port);
sdsfree(mi->ip);
zfree(mi);
listDelNode(g_pserver->repl_init_masters, ln);
}
}
listRelease(g_pserver->repl_init_masters);
for (int iel = 0; iel < cserver.cthreads; ++iel)
{
initServerThread(g_pserver->rgthreadvar+iel, iel == IDX_EVENT_LOOP_MAIN);
@@ -7083,6 +6966,8 @@ int main(int argc, char **argv) {
setOOMScoreAdj(-1);
serverAssert(cserver.cthreads > 0 && cserver.cthreads <= MAX_EVENT_LOOPS);
pthread_create(&cserver.time_thread_id, nullptr, timeThreadMain, nullptr);
pthread_attr_t tattr;
pthread_attr_init(&tattr);
pthread_attr_setstacksize(&tattr, 1 << 23); // 8 MB
+22 -59
View File
@@ -970,8 +970,6 @@ typedef struct clientReplyBlock {
#endif
} clientReplyBlock;
struct redisNamespace;
/* Redis database representation. There are multiple databases identified
* by integers from 0 (the default database) up to the max configured
* database. The database number is the 'id' field in the structure. */
@@ -982,7 +980,6 @@ struct redisDb {
~redisDb();
redisNamespace *ns;
::dict *dict; /* The keyspace for this DB */
expireset *setexpire;
expireset::setiter expireitr;
@@ -990,33 +987,12 @@ struct redisDb {
::dict *blocking_keys; /* Keys with clients waiting for data (BLPOP)*/
::dict *ready_keys; /* Blocked keys that received a PUSH */
::dict *watched_keys; /* WATCHED keys for MULTI/EXEC CAS */
int id; /* Global database ID */
int mapped_id; /* Namespaced database ID */
int id; /* Database ID */
long long last_expire_set; /* when the last expire was set */
double avg_ttl; /* Average TTL, just for stats */
list *defrag_later; /* List of key names to attempt to defrag one by one, gradually. */
};
struct redisNamespaceCommandStats {
long long microseconds, calls, rejected_calls, failed_calls;
};
struct redisNamespace {
sds name;
redisDb **db;
::dict *pubsub_channels; /* Map channels to list of subscribed clients */
::dict *pubsub_patterns; /* A dict of pubsub_patterns */
::dict *lua_scripts; /* A dictionary of SHA1 -> Lua scripts */
lua_State *lua; /* The Lua interpreter. We use one per namespace */
unsigned long long lua_scripts_mem; /* Cached scripts' memory + oh */
::dict *repl_scriptcache_dict; /* SHA1 all slaves are aware of. */
list *repl_scriptcache_fifo; /* First in, first out LRU eviction. */
unsigned int repl_scriptcache_size; /* Max number of elements. */
redisNamespaceCommandStats *stat_commands;
long long stat_numcommands;
};
/* Declare database backup that include redis main DBs and slots to keys map.
* Definition is in db.c. We can't define it here since we define CLUSTER_SLOTS
* in cluster.h. */
@@ -1126,7 +1102,6 @@ typedef struct readyList {
typedef struct {
sds name; /* The username as an SDS string. */
sds ns_name; /* The namespace of this user as an SDS string. */
uint64_t flags; /* See USER_FLAG_* */
/* The bit in allowed_commands is set if this user has the right to
@@ -1166,7 +1141,6 @@ typedef struct client {
connection *conn;
int resp; /* RESP protocol version. Can be 2 or 3. */
redisDb *db; /* Pointer to currently SELECTed DB. */
redisNamespace *ns; /* Pointer to currently active namespace. */
robj *name; /* As set by CLIENT SETNAME. */
sds querybuf; /* Buffer we use to accumulate client queries. */
size_t qb_pos; /* The position we have read in querybuf. */
@@ -1312,7 +1286,7 @@ struct sharedObjectsStruct {
*time, *pxat, *px, *retrycount, *force, *justid,
*lastid, *ping, *replping, *setid, *keepttl, *load, *createconsumer,
*getack, *special_asterick, *special_equals, *default_username,
*hdel, *zrem, *mvccrestore, *pexpirememberat, *redacted, *allocate,
*hdel, *zrem, *mvccrestore, *pexpirememberat, *redacted,
*select[PROTO_SHARED_SELECT_CMDS],
*integers[OBJ_SHARED_INTEGERS],
*mbulkhdr[OBJ_SHARED_BULKHDR_LEN], /* "*<value>\r\n" */
@@ -1525,11 +1499,6 @@ struct redisServerThreadVars {
std::vector<client*> vecclientsProcess;
};
struct redisMasterConnInfo {
char *ip;
int port;
};
struct redisMaster {
char *masteruser; /* AUTH with this user and masterauth with master */
char *masterauth; /* AUTH with this password with master */
@@ -1567,6 +1536,7 @@ struct redisServerConst {
pid_t pid; /* Main process pid. */
time_t stat_starttime; /* Server start time */
pthread_t main_thread_id; /* Main thread id */
pthread_t time_thread_id;
char *configfile; /* Absolute config file path, or NULL */
char *executable; /* Absolute executable file path. */
char **exec_argv; /* Executable argv vector (copy). */
@@ -1582,12 +1552,11 @@ struct redisServerConst {
*zpopmaxCommand, *sremCommand, *execCommand,
*expireCommand, *pexpireCommand, *xclaimCommand,
*xgroupCommand, *rreplayCommand, *rpoplpushCommand,
*hdelCommand, *zremCommand, *lmoveCommand, *allocateCommand;
*hdelCommand, *zremCommand, *lmoveCommand;
/* Configuration */
char *default_masteruser; /* AUTH with this user and masterauth with master */
char *default_masterauth; /* AUTH with this password with master */
char *default_user_namespace; /* default user namespace */
int verbosity; /* Loglevel in keydb.conf */
int maxidletime; /* Client timeout in seconds */
int tcpkeepalive; /* Set SO_KEEPALIVE if non-zero. */
@@ -1603,7 +1572,6 @@ struct redisServerConst {
unsigned long active_defrag_max_scan_fields; /* maximum number of fields of set/hash/zset/list to process from within the main dict scan */
size_t client_max_querybuf_len; /* Limit for client query buffer length */
int dbnum; /* Total number of configured DBs */
int ns_dbnum; /* Maximum number of configured databases per namespace */
int supervised; /* 1 if supervised, 0 otherwise. */
int supervised_mode; /* See SUPERVISED_* */
int daemonize; /* True if running as a daemon */
@@ -1632,10 +1600,7 @@ struct redisServer {
mode_t umask; /* The umask value of the process on startup */
std::atomic<int> hz; /* serverCron() calls frequency in hertz */
int in_fork_child; /* indication that this is a fork child */
int last_allocated_db; /* store last allocated db, to speed up allocation */
redisDb *db;
redisNamespace *default_namespace;
::dict *namespaces;
::dict *commands; /* Command table */
::dict *orig_commands; /* Command table before command renaming. */
@@ -1870,8 +1835,7 @@ struct redisServer {
int repl_diskless_sync_delay; /* Delay to start a diskless repl BGSAVE. */
/* Replication (replica) */
list *masters;
list *repl_init_masters; /* temp variable to hold the master ip/port pairs to init after config parsing */
int enable_multimaster;
int enable_multimaster;
int repl_timeout; /* Timeout after N seconds of master idle */
int repl_syncio_timeout; /* Timeout for synchronous I/O calls */
int repl_disable_tcp_nodelay; /* Disable TCP_NODELAY after SYNC? */
@@ -1884,6 +1848,10 @@ struct redisServer {
int slave_announce_port; /* Give the master this listening port. */
char *slave_announce_ip; /* Give the master this ip address. */
int repl_slave_lazy_flush; /* Lazy FLUSHALL before loading DB? */
/* Replication script cache. */
::dict *repl_scriptcache_dict; /* SHA1 all slaves are aware of. */
list *repl_scriptcache_fifo; /* First in, first out LRU eviction. */
unsigned int repl_scriptcache_size; /* Max number of elements. */
/* Synchronous replication. */
list *clients_waiting_acks; /* Clients waiting in WAIT command. */
int get_ack_from_slaves; /* If true we send REPLCONF GETACK. */
@@ -1934,6 +1902,8 @@ struct redisServer {
size_t blocking_op_nesting; /* Nesting level of blocking operation, used to reset blocked_last_cron. */
long long blocked_last_cron; /* Indicate the mstime of the last time we did cron jobs from a blocking operation */
/* Pubsub */
dict *pubsub_channels; /* Map channels to list of subscribed clients */
dict *pubsub_patterns; /* A dict of pubsub_patterns */
int notify_keyspace_events; /* Events to propagate via Pub/Sub. This is an
xor of NOTIFY_... flags. */
/* Cluster */
@@ -1960,8 +1930,11 @@ struct redisServer {
is down? */
int cluster_config_file_lock_fd; /* cluster config fd, will be flock */
/* Scripting */
lua_State *lua; /* The Lua interpreter. We use just one for all clients */
client *lua_caller = nullptr; /* The client running EVAL right now, or NULL */
char* lua_cur_script = nullptr; /* SHA1 of the script currently running, or NULL */
::dict *lua_scripts; /* A dictionary of SHA1 -> Lua scripts */
unsigned long long lua_scripts_mem; /* Cached scripts' memory + oh */
mstime_t lua_time_limit; /* Script timeout in milliseconds */
monotime lua_time_start; /* monotonic timer to detect timed-out script */
mstime_t lua_time_snapshot; /* Snapshot of mstime when script is started */
@@ -2162,10 +2135,6 @@ extern dictType sdsReplyDictType;
* Functions prototypes
*----------------------------------------------------------------------------*/
/* Namespaces */
redisNamespace *getNamespace(const char* name);
sds generateAutoNamespaceName();
/* Modules */
void moduleInitModulesSystem(void);
void moduleInitModulesSystemLast(void);
@@ -2455,10 +2424,10 @@ void resizeReplicationBacklog(long long newsize);
struct redisMaster *replicationAddMaster(char *ip, int port);
void replicationUnsetMaster(struct redisMaster *mi);
void refreshGoodSlavesCount(void);
void replicationScriptCacheInit(redisNamespace *ns);
void replicationScriptCacheInit(void);
void replicationScriptCacheFlush(void);
void replicationScriptCacheAdd(redisNamespace *ns, sds sha1);
int replicationScriptCacheExists(redisNamespace *ns, sds sha1);
void replicationScriptCacheAdd(sds sha1);
int replicationScriptCacheExists(sds sha1);
void processClientsWaitingReplicas(void);
void unblockClientWaitingReplicas(client *c);
int replicationCountAcksByOffset(long long offset);
@@ -2665,12 +2634,11 @@ void updateDictResizePolicy(void);
int htNeedsResize(dict *dict);
void populateCommandTable(void);
void resetCommandTableStats(void);
void resetCommandTableStatsNs(redisNamespace *ns);
void resetErrorTableStats(void);
void adjustOpenFilesLimit(void);
void incrementErrorCount(const char *fullerr, size_t namelen);
void closeListeningSockets(int unlink_unix_socket);
void updateCachedTime(int update_daylight_info);
void updateCachedTime(void);
void resetServerStats(void);
void activeDefragCycle(void);
unsigned int getLRUClock(void);
@@ -2733,7 +2701,7 @@ int hashZiplistValidateIntegrity(unsigned char *zl, size_t size, int deep);
/* Pub / Sub */
int pubsubUnsubscribeAllChannels(client *c, int notify);
int pubsubUnsubscribeAllPatterns(client *c, int notify);
int pubsubPublishMessage(redisNamespace *ns, robj *channel, robj *message);
int pubsubPublishMessage(robj *channel, robj *message);
void addReplyPubsubMessage(client *c, robj *channel, robj *msg);
/* Keyspace events notification */
@@ -2793,7 +2761,7 @@ robj *dbUnshareStringValue(redisDb *db, robj *key, robj *o);
#define EMPTYDB_ASYNC (1<<0) /* Reclaim memory in another thread. */
long long emptyDb(int dbnum, int flags, void(callback)(void*));
long long emptyDbStructure(redisDb *dbarray, int dbnum, int async, void(callback)(void*));
void flushAllDataAndResetRDB(redisNamespace *ns, int flags);
void flushAllDataAndResetRDB(int flags);
long long dbTotalServerKeyCount();
dbBackup *backupDb(void);
void restoreDbBackup(dbBackup *buckup);
@@ -2801,8 +2769,6 @@ void discardDbBackup(dbBackup *buckup, int flags, void(callback)(void*));
int selectDb(client *c, int id);
int selectDbNamespaced(client *c, int id);
void printDbMap();
void signalModifiedKey(client *c, redisDb *db, robj *key);
void signalFlushedDb(int dbid, int async);
unsigned int getKeysInSlot(unsigned int hashslot, robj **keys, unsigned int count);
@@ -2842,7 +2808,7 @@ void clusterInit(void);
extern "C" unsigned short crc16(const char *buf, int len);
unsigned int keyHashSlot(char *key, int keylen);
void clusterCron(void);
void clusterPropagatePublish(redisNamespace *ns, robj *channel, robj *message);
void clusterPropagatePublish(robj *channel, robj *message);
void migrateCloseTimedoutSockets(void);
void clusterBeforeSleep(void);
int clusterSendModuleMessageToTarget(const char *target, uint64_t module_id, uint8_t type, unsigned char *payload, uint32_t len);
@@ -2864,8 +2830,7 @@ int redis_check_rdb_main(int argc, const char **argv, FILE *fp);
int redis_check_aof_main(int argc, char **argv);
/* Scripting */
void scriptingSetup();
void scriptingInit(redisNamespace *ns);
void scriptingInit(int setup);
int ldbRemoveChild(pid_t pid);
void ldbKillForkedSessions(void);
int ldbPendingChildren(void);
@@ -2958,8 +2923,6 @@ void incrbyCommand(client *c);
void decrbyCommand(client *c);
void incrbyfloatCommand(client *c);
void selectCommand(client *c);
void selectNsCommand(client *c);
void allocateCommand(client *c);
void swapdbCommand(client *c);
void randomkeyCommand(client *c);
void keysCommand(client *c);
-4
View File
@@ -1,4 +0,0 @@
user alice on ~* +@all -@dangerous -@admin >alice ::ns1
user bob on ~* +@all -@dangerous -@admin >bob ::ns2
user b2 on ~* +@all -@dangerous -@admin >b2 ::ns2
user default on nopass ~* +@all ::
+2 -4
View File
@@ -1,7 +1,7 @@
start_server {tags {"introspection"}} {
test {CLIENT LIST} {
r client list
} {*addr=*:* fd=* age=* idle=* flags=N db=9* sub=0 psub=0 multi=-1 qbuf=26 qbuf-free=* argv-mem=* obl=0 oll=0 omem=0 tot-mem=* events=r cmd=client* ns=*}
} {*addr=*:* fd=* age=* idle=* flags=N db=9 sub=0 psub=0 multi=-1 qbuf=26 qbuf-free=* argv-mem=* obl=0 oll=0 omem=0 tot-mem=* events=r cmd=client*}
test {CLIENT LIST with IDs} {
set myid [r client id]
@@ -11,7 +11,7 @@ start_server {tags {"introspection"}} {
test {CLIENT INFO} {
r client info
} {*addr=*:* fd=* age=* idle=* flags=N db=9* sub=0 psub=0 multi=-1 qbuf=26 qbuf-free=* argv-mem=* obl=0 oll=0 omem=0 tot-mem=* events=r cmd=client* ns=*}
} {*addr=*:* fd=* age=* idle=* flags=N db=9 sub=0 psub=0 multi=-1 qbuf=26 qbuf-free=* argv-mem=* obl=0 oll=0 omem=0 tot-mem=* events=r cmd=client*}
test {MONITOR can log executed commands} {
set rd [redis_deferring_client]
@@ -147,7 +147,6 @@ start_server {tags {"introspection"}} {
supervised
syslog-facility
databases
databases-per-namespace
io-threads
logfile
unixsocketperm
@@ -161,7 +160,6 @@ start_server {tags {"introspection"}} {
active-replica
bind
set-proc-title
acl-namespace-default
}
if {!$::tls} {
-101
View File
@@ -1,101 +0,0 @@
set server_path [tmpdir "server.namepsace"]
exec cp -f tests/assets/user-namespaces.acl $server_path
start_server [list overrides [list "dir" $server_path "aclfile" "user-namespaces.acl"]] {
# user alice on ~* +@all -@dangerous -@admin >alice ::ns1
# user bob on ~* +@all -@dangerous -@admin >bob ::ns2
# user b2 on ~* +@all -@dangerous -@admin >b2 ::ns2
# user default on nopass ~* +@all ::
test {::auto generates unique namespace} {
r ACL setuser auto1 on ~* +@all -@dangerous -@admin >auto ::auto
set ns1 [dict get [r ACL getuser auto1] namespace]
assert_match "::*-*-*-*" $ns1
r ACL setuser auto2 on ~* +@all -@dangerous -@admin >auto ::auto
set ns2 [dict get [r ACL getuser auto2] namespace]
assert_match "::*-*-*-*" $ns2
assert {$ns1 != $ns2}
}
test {alice can set keys in multiple databases} {
r AUTH alice alice
r SELECT 0
r SET name alice0
r SELECT 1
r SET name alice1
r SELECT 2
r SET name alice2
}
test {bob does not have acces to the latest db of alice} {
r AUTH bob bob
assert_equal "" [r GET name]
}
test {bob can set keys in multiple databases} {
r AUTH bob bob
r SELECT 0
r SET name bob0
r SELECT 1
r SET name bob1
r SELECT 2
r SET name bob2
}
test {alice's keys are not overwritten by bob} {
r AUTH alice alice
r SELECT 0
assert_equal "alice0" [r GET name]
r SELECT 1
assert_equal "alice1" [r GET name]
r SELECT 2
assert_equal "alice2" [r GET name]
}
test {bob's keys are not overwritten by alice} {
r AUTH bob bob
r SELECT 0
assert_equal "bob0" [r GET name]
r SELECT 1
assert_equal "bob1" [r GET name]
r SELECT 2
assert_equal "bob2" [r GET name]
}
test {b2 can see bob's keys} {
r AUTH b2 b2
r SELECT 0
assert_equal "bob0" [r GET name]
r SELECT 1
assert_equal "bob1" [r GET name]
r SELECT 2
assert_equal "bob2" [r GET name]
}
test {admin can select databases in different namespaces} {
r AUTH default ""
r SELECTNS 0 ::ns1
assert_equal "alice0" [r GET name]
r SELECTNS 1 ::ns1
assert_equal "alice1" [r GET name]
r SELECTNS 0 ::ns2
assert_equal "bob0" [r GET name]
r SELECTNS 1 ::ns2
assert_equal "bob1" [r GET name]
}
test {Scripts commands should not affect other namespaces} {
r AUTH alice alice
set sha [r SCRIPT LOAD "return true"]
assert_equal "1" [r SCRIPT EXISTS $sha]
r AUTH bob bob
assert_equal "0" [r SCRIPT EXISTS $sha]
set sha2 [r SCRIPT LOAD "return true"]
assert_equal $sha $sha2
r SCRIPT FLUSH
assert_equal "0" [r SCRIPT EXISTS $sha]
r AUTH alice alice
assert_equal "1" [r SCRIPT EXISTS $sha]
r EVALSHA $sha 0
}
}
-2
View File
@@ -2,7 +2,6 @@ start_server {tags {"rreplay"} overrides {active-replica yes}} {
test {RREPLAY use current db} {
r debug force-master yes
r allocate :: 4 0
r select 4
r set dbnum invalid
r rreplay "f4d5b2b5-4f07-4ee5-a4f2-5dc98507dfce" "*3\r\n\$3\r\nSET\r\n\$5\r\ndbnum\r\n\$4\r\nfour\r\n"
@@ -12,7 +11,6 @@ start_server {tags {"rreplay"} overrides {active-replica yes}} {
test {RREPLAY db different} {
r debug force-master yes
r allocate :: 2 2
r select 4
r set testkey four
r rreplay "f4d5b2b5-4f07-4ee5-a4f2-5dc98507dfce" "*3\r\n\$3\r\nSET\r\n\$7\r\ntestkey\r\n\$4\r\nbebe\r\n" 2