Compare commits

..
Author SHA1 Message Date
Vivek Saini 70959cd222 Fixed memory leaks and initialized namespace 2022-03-07 19:52:41 -05:00
Eric de Ruiter ac5c4e3a31 Added command stats per namespace + output them in the info command 2021-11-07 01:03:39 +01:00
Eric de Ruiter 160d443c23 Move script cache to namespace 2021-11-06 22:38:39 +01:00
Eric de Ruiter 6cadb8c4d0 Reset last_allocated_db after emptying all databases 2021-11-06 20:50:06 +01:00
Eric de Ruiter 0f4721654f Added last_allocated_db to prevent looping over all databases every time 2021-11-06 13:04:05 +01:00
Eric de Ruiter d0e5165306 Added SELECTNS admin command to switch to databases in other namespaces 2021-11-06 12:26:22 +01:00
Eric de Ruiter ace526fa8b forgot to commit generateAutoNamespaceName implementation 2021-11-01 10:53:04 +01:00
Eric de Ruiter c2efd10f4d added support for ::auto namespace + support for default user namespace in configuration 2021-11-01 10:49:21 +01:00
Eric de Ruiter 8434366312 redisMaster->staleKeyMap is not always initialized, only delete it when initialized 2021-10-29 15:06:50 +02:00
Eric de Ruiter 05b631e732 Namespace fixes for rdb/aof/scripting 2021-10-29 15:06:50 +02:00
Eric de Ruiter 9b98faf8ee Added database mapping + namespace to client list output + added complete database mapping to serverAssertWithInfo 2021-10-29 15:06:50 +02:00
Eric de Ruiter acda6f203e Fix flushall + emptyDbStucture + debug loadaof 2021-10-29 15:06:50 +02:00
Eric de Ruiter 71aa11992f Move master initialization out of config parsing; this caused issues as db's/namespaces are not yet initialized at that stage 2021-10-29 15:06:50 +02:00
Eric de Ruiter 4254348c59 added basic namespace tests 2021-10-29 15:06:50 +02:00
Eric de Ruiter 9adb05ce78 added actual namespace switching in ACLAuthenticateUser 2021-10-29 15:06:50 +02:00
Eric de Ruiter c8c4828f8c fix rreplay tests to also include allocate calls (this is needed as it uses debug force-master which causes the select to not allocate databases) 2021-10-29 15:06:50 +02:00
Eric de Ruiter eb311cf7d1 make sure fake replication client also has a namespace 2021-10-29 15:06:50 +02:00
Eric de Ruiter d3cd45779f fix debug htstats to work with namespaces 2021-10-29 15:06:50 +02:00
Eric de Ruiter 25e37446c9 make sure db allocation is always present when selecting a different database in aof (as the oaf might be enabled after a database is selected) 2021-10-29 15:06:50 +02:00
Eric de Ruiter b8561cf4ba implemented database allocation + acl namespace support 2021-10-29 15:06:50 +02:00
Eric de Ruiter 3c4212f87f implemented basic support for namespaces (this only adds support for the default namespace) 2021-10-29 15:06:03 +02:00
39 changed files with 1015 additions and 843 deletions

No files matched your search

-1
View File
@@ -6,7 +6,6 @@ Build-Depends:
debhelper (>= 9~),
dpkg-dev (>= 1.17.5),
systemd,
libsystemd-dev <!nocheck>,
procps <!nocheck>,
build-essential <!nocheck>,
tcl <!nocheck>,
+7 -85
View File
@@ -1,94 +1,16 @@
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/EQ-Alpha/KeyDB
Upstream-Name: keydb-pro
Source: https://github.com/JohnSully/KeyDB-Pro
Files: *
Copyright:
© 2006-2014 Salvatore Sanfilippo <antirez@gmail.com>
© 2019-2021 John Sully
© 2019-2021 EQ Alpha Technology Ltd.
Copyright: © 2006-2014 Salvatore Sanfilippo <antirez@gmail.com>
Copyright © 2019, John Sully
License: Proprietary
License: BSD-3-clause
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>
Files: debian/*
Copyright: © 2009 Chris Lamb <lamby@debian.org>
License: BSD-3-clause
License: BSD-2-clause
-1
View File
@@ -5,4 +5,3 @@ 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,7 +3,6 @@
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,7 +6,6 @@ Build-Depends:
debhelper (>= 9~),
dpkg-dev (>= 1.17.5),
systemd,
libsystemd-dev <!nocheck>,
procps <!nocheck>,
build-essential <!nocheck>,
tcl <!nocheck>,
+6 -11
View File
@@ -1,14 +1,11 @@
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/EQ-Alpha/KeyDB
Upstream-Name: keydb
Source: https://github.com/JohnSully/KeyDB
Files: *
Copyright:
© 2006-2014 Salvatore Sanfilippo <antirez@gmail.com>
© 2019-2021 John Sully
© 2019-2021 EQ Alpha Technology Ltd.
Copyright: © 2006-2014 Salvatore Sanfilippo <antirez@gmail.com>
Copyright © 2019, John Sully
License: BSD-3-clause
Files:
@@ -85,10 +82,8 @@ License: MIT
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
Files: pkg/deb/debian_dh9/*
Copyright:
© 2009 Chris Lamb <lamby@debian.org>
© 2020-2021 EQ Alpha Technology Ltd. <support@eqalpha.com>
Files: debian/*
Copyright: © 2009 Chris Lamb <lamby@debian.org>
License: BSD-3-clause
License: BSD-2-clause
-1
View File
@@ -5,4 +5,3 @@ 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,7 +9,6 @@ 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)
+1 -7
View File
@@ -25,13 +25,7 @@ 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-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/
cp $DIR/../../src/keydb-* $DIR/keydb_build/keydb_rpm/usr/bin/
# update spec file with build info
sed -i '2d' $DIR/keydb_build/keydb.spec
+38 -9
View File
@@ -33,6 +33,7 @@ extern "C" {
}
#include <fcntl.h>
#include <ctype.h>
#include <uuid/uuid.h>
/* =============================================================================
* Global state for ACLs
@@ -223,18 +224,12 @@ uint64_t ACLGetCommandCategoryFlagByName(const char *name) {
return 0; /* No match. */
}
/* Method for pattern comparison used for the user->patterns list
/* Method for passwords/pattern comparison used for the user->passwords 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);
@@ -254,12 +249,19 @@ 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,ACLListMatchSdsSecure);
listSetMatchMethod(u->passwords,ACLListMatchSds);
listSetFreeMethod(u->passwords,ACLListFreeSds);
listSetDupMethod(u->passwords,ACLListDupSds);
listSetMatchMethod(u->patterns,ACLListMatchSds);
@@ -294,6 +296,7 @@ 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);
@@ -344,6 +347,8 @@ 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) {
@@ -675,6 +680,13 @@ 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;
}
@@ -807,6 +819,9 @@ 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
@@ -1006,6 +1021,14 @@ 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);
@@ -1120,6 +1143,8 @@ 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 {
@@ -1975,7 +2000,7 @@ void aclCommand(client *c) {
return;
}
addReplyMapLen(c,5);
addReplyMapLen(c,6);
/* Flags */
addReplyBulkCString(c,"flags");
@@ -2036,6 +2061,10 @@ 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)
{
+20 -5
View File
@@ -731,10 +731,14 @@ 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 != g_pserver->aof_selected_db) {
char seldb[64];
if (dictid != -1 && dictid != g_pserver->aof_selected_db) {
char seldb[64], mappeddb[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;
@@ -767,7 +771,8 @@ void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int a
struct client *createAOFClient(void) {
struct client *c =(client*) zmalloc(sizeof(*c), MALLOC_LOCAL);
selectDb(c,0);
c->ns = g_pserver->default_namespace;
selectDbNamespaced(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;
@@ -879,7 +884,7 @@ int loadAppendOnlyFile(char *filename) {
} else {
/* RDB preamble. Pass loading the RDB functions. */
rio rdb;
rdbSaveInfo rsi;
rdbSaveInfo rsi = RDB_SAVE_INFO_INIT;
serverLog(LL_NOTICE,"Reading RDB preamble from AOF file...");
if (fseek(fp,0,SEEK_SET) == -1) goto readerr;
@@ -1546,12 +1551,22 @@ 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;
+6 -15
View File
@@ -405,10 +405,6 @@ 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).
@@ -646,11 +642,6 @@ 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;
}
@@ -691,7 +682,7 @@ void bitopCommand(client *c) {
if (fShiftOp)
{
long long newlen = (long long)maxlen + shift/CHAR_BIT;
long newlen = (long)maxlen + shift/CHAR_BIT;
if (shift > 0 && (shift % CHAR_BIT) != 0)
newlen++;
@@ -703,14 +694,14 @@ void bitopCommand(client *c) {
res = (unsigned char*) sdsnewlen(NULL,newlen);
if (shift >= 0)
{ // left shift
long long byteoffset = shift/CHAR_BIT;
long byteoffset = shift/CHAR_BIT;
memset(res, 0, byteoffset);
long long srcLen = newlen - byteoffset - ((shift % CHAR_BIT) ? 1 : 0);
long srcLen = newlen - byteoffset - ((shift % CHAR_BIT) ? 1 : 0);
// now the bitshift+copy
unsigned bitshift = shift % CHAR_BIT;
unsigned char carry = 0;
for (long long iSrc = 0; iSrc < srcLen; ++iSrc)
for (long iSrc = 0; iSrc < srcLen; ++iSrc)
{
res[byteoffset+iSrc] = (src[0][iSrc] << bitshift) | carry;
carry = src[0][iSrc] >> (CHAR_BIT - bitshift);
@@ -720,14 +711,14 @@ void bitopCommand(client *c) {
}
else
{ // right shift
long long byteoffset = -shift/CHAR_BIT;
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 long idx = 1; idx < newlen; ++idx)
for (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,6 +592,7 @@ 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);
+24 -16
View File
@@ -2196,24 +2196,32 @@ int clusterProcessPacket(clusterLink *link) {
}
} else if (type == CLUSTERMSG_TYPE_PUBLISH) {
robj *channel, *message;
uint32_t channel_len, message_len;
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);
/* Don't bother creating useless objects if there are no
* Pub/Sub subscribers. */
if (dictSize(g_pserver->pubsub_channels) ||
dictSize(g_pserver->pubsub_patterns))
if (!ns ||
dictSize(ns->pubsub_channels) ||
dictSize(ns->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,channel_len);
(char*)hdr->data.publish.msg.bulk_data+ns_len,channel_len);
message = createStringObject(
(char*)hdr->data.publish.msg.bulk_data+channel_len,
(char*)hdr->data.publish.msg.bulk_data+ns_len+channel_len,
message_len);
pubsubPublishMessage(channel,message);
pubsubPublishMessage(ns,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);
@@ -2447,10 +2455,7 @@ 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] {
/* 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);
connSetWriteHandlerWithBarrier(link->conn, clusterWriteHandler, 1);
});
}
@@ -2757,22 +2762,24 @@ void clusterBroadcastPong(int target) {
/* Send a PUBLISH message.
*
* If link is NULL, then the message is broadcasted to the whole cluster. */
void clusterSendPublish(clusterLink *link, robj *channel, robj *message) {
void clusterSendPublish(redisNamespace *ns, 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;
uint32_t channel_len, message_len, ns_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;
totlen += sizeof(clusterMsgDataPublish) - 8 + channel_len + message_len + ns_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);
@@ -2785,7 +2792,8 @@ void clusterSendPublish(clusterLink *link, robj *channel, robj *message) {
memcpy(payload,hdr,sizeof(*hdr));
hdr = (clusterMsg*) payload;
}
memcpy(hdr->data.publish.msg.bulk_data,ptrFromObj(channel),sdslen(szFromObj(channel)));
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+sdslen(szFromObj(channel)),
ptrFromObj(message),sdslen(szFromObj(message)));
@@ -2891,8 +2899,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(robj *channel, robj *message) {
clusterSendPublish(NULL, channel, message);
void clusterPropagatePublish(redisNamespace *ns, robj *channel, robj *message) {
clusterSendPublish(ns, NULL, channel, message);
}
/* -----------------------------------------------------------------------------
+1
View File
@@ -209,6 +209,7 @@ 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. */
+7 -2
View File
@@ -542,7 +542,11 @@ void loadServerConfigFromString(char *config) {
if (port < 0 || port > 65535 || *ptr != '\0') {
err= "Invalid master port"; goto loaderr;
}
replicationAddMaster(argv[1], port);
redisMasterConnInfo *masterInfo = (redisMasterConnInfo*)zmalloc(sizeof(redisMasterConnInfo), MALLOC_LOCAL);
masterInfo->ip = sdsnew(argv[1]);
masterInfo->port = port;
listAddNodeTail(g_pserver->repl_init_masters, masterInfo);
} 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";
@@ -2667,7 +2671,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),
@@ -2685,6 +2689,7 @@ 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,7 +51,6 @@ void cronCommand(client *c)
if (c->argc < (6 + numkeys)) {
addReplyError(c, "Missing arguments or numkeys is too big");
return;
}
}
+146 -11
View File
@@ -494,6 +494,13 @@ 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;
@@ -643,6 +650,88 @@ 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;
}
@@ -716,8 +805,17 @@ int getFlushCommandFlags(client *c, int *flags) {
}
/* Flushes the whole server data set. */
void flushAllDataAndResetRDB(int flags) {
g_pserver->dirty += emptyDb(-1,flags,NULL);
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);
}
}
}
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
@@ -765,7 +863,7 @@ void flushdbCommand(client *c) {
void flushallCommand(client *c) {
int flags;
if (getFlushCommandFlags(c,&flags) == C_ERR) return;
flushAllDataAndResetRDB(flags);
flushAllDataAndResetRDB(c->ns, flags);
addReply(c,shared.ok);
}
@@ -825,13 +923,48 @@ void selectCommand(client *c) {
addReplyError(c,"SELECT is not allowed in cluster mode");
return;
}
if (selectDb(c,id) == C_ERR) {
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) {
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;
@@ -1259,7 +1392,7 @@ void moveCommand(client *c) {
if (getIntFromObjectOrReply(c, c->argv[2], &dbid, NULL) != C_OK)
return;
if (selectDb(c,dbid) == C_ERR) {
if (selectDbNamespaced(c, dbid) == C_ERR) {
addReplyError(c,"DB index is out of range");
return;
}
@@ -1334,7 +1467,7 @@ void copyCommand(client *c) {
if (getIntFromObjectOrReply(c, c->argv[j+1], &dbid, NULL) != C_OK)
return;
if (selectDb(c, dbid) == C_ERR) {
if (selectDbNamespaced(c, dbid) == C_ERR) {
addReplyError(c,"DB index is out of range");
return;
}
@@ -1438,11 +1571,13 @@ 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(int id1, int id2) {
if (id1 < 0 || id1 >= cserver.dbnum ||
id2 < 0 || id2 >= cserver.dbnum) return C_ERR;
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;
if (id1 == id2) return C_OK;
redisDb *db1 = &g_pserver->db[id1], *db2 = &g_pserver->db[id2];
redisDb *db1 = allocateDb(c, id1);
redisDb *db2 = allocateDb(c, id2);
/* Swap hash tables. Note that we don't swap blocking_keys,
* ready_keys and watched_keys, since we want clients to
@@ -1492,7 +1627,7 @@ void swapdbCommand(client *c) {
return;
/* Swap... */
if (dbSwapDatabases(id1,id2) == C_ERR) {
if (dbSwapDatabases(c, id1,id2) == C_ERR) {
addReplyError(c,"DB index is out of range");
return;
} else {
+8 -4
View File
@@ -567,9 +567,11 @@ 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) {
@@ -577,7 +579,7 @@ NULL
return;
}
g_pserver->dirty = 0; /* Prevent AOF / replication */
serverLog(LL_WARNING,"Append Only File loaded by DEBUG LOADAOF");
serverLog(LL_WARNING,"Append Only File loaded by DEBUG LOADAOF done");
addReply(c,shared.ok);
} else if (!strcasecmp(szFromObj(c->argv[1]),"object") && c->argc == 3) {
dictEntry *de;
@@ -840,18 +842,18 @@ NULL
sdsfree(stats);
return;
}
if (dbid < 0 || dbid >= cserver.dbnum) {
if (dbid < 0 || dbid >= std::min(cserver.dbnum, cserver.ns_dbnum)) {
sdsfree(stats);
addReplyError(c,"Out of range database");
return;
}
stats = sdscatprintf(stats,"[Dictionary HT]\n");
dictGetStats(buf,sizeof(buf),g_pserver->db[dbid].dict);
dictGetStats(buf,sizeof(buf),c->ns->db[dbid]->dict);
stats = sdscat(stats,buf);
stats = sdscatprintf(stats,"[Expires set]\n");
g_pserver->db[dbid].setexpire->getstats(buf, sizeof(buf));
c->ns->db[dbid]->setexpire->getstats(buf, sizeof(buf));
stats = sdscat(stats, buf);
addReplyVerbatim(c,stats,sdslen(stats),"txt");
@@ -972,6 +974,7 @@ 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);
@@ -1035,6 +1038,7 @@ 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);
}
+13 -2
View File
@@ -981,8 +981,19 @@ 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 */
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);
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 += moduleDefragGlobals();
return defragged;
}
+59 -25
View File
@@ -365,7 +365,11 @@ typedef struct RedisModuleCommandFilter {
static list *moduleCommandFilters;
/* Module GIL Variables */
static readWriteLock s_moduleGIL("Module GIL");
static int s_cAcquisitionsServer = 0;
static int s_cAcquisitionsModule = 0;
static std::mutex s_mutex;
static std::condition_variable s_cv;
static std::recursive_mutex s_mutexModule;
thread_local bool g_fModuleThread = false;
typedef void (*RedisModuleForkDoneHandler) (int exitcode, int bysignal, void *user_data);
@@ -2093,10 +2097,9 @@ 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) {
UNUSED(ctx);
int receivers = pubsubPublishMessage(channel, message);
int receivers = pubsubPublishMessage(ctx->client->ns, channel, message);
if (g_pserver->cluster_enabled)
clusterPropagatePublish(channel, message);
clusterPropagatePublish(ctx->client->ns, channel, message);
return receivers;
}
@@ -2289,7 +2292,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 = selectDb(ctx->client,newid);
int retval = selectDbNamespaced(ctx->client,newid);
return (retval == C_OK) ? REDISMODULE_OK : REDISMODULE_ERR;
}
@@ -2509,7 +2512,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(async? EMPTYDB_ASYNC: EMPTYDB_NO_FLAGS);
flushAllDataAndResetRDB(NULL, async? EMPTYDB_ASYNC: EMPTYDB_NO_FLAGS);
if (g_pserver->aof_enabled && restart_aof) restartAOFAfterSYNC();
}
@@ -4211,6 +4214,7 @@ 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++;
@@ -4285,14 +4289,7 @@ 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);
@@ -5965,58 +5962,95 @@ void RM_ThreadSafeContextUnlock(RedisModuleCtx *ctx) {
// as the server thread acquisition is sufficient. If we did try to lock we would deadlock
static bool FModuleCallBackLock(bool fServerThread)
{
return !fServerThread && aeThreadOwnsLock() && !g_fModuleThread && s_moduleGIL.hasReader();
return !fServerThread && aeThreadOwnsLock() && !g_fModuleThread && s_cAcquisitionsServer > 0;
}
void moduleAcquireGIL(int fServerThread, int fExclusive) {
std::unique_lock<std::mutex> lock(s_mutex);
int *pcheck = fServerThread ? &s_cAcquisitionsModule : &s_cAcquisitionsServer;
if (FModuleCallBackLock(fServerThread)) {
return;
}
while (*pcheck > 0)
s_cv.wait(lock);
if (fServerThread)
{
s_moduleGIL.acquireRead();
++s_cAcquisitionsServer;
}
else
{
s_moduleGIL.acquireWrite(fExclusive);
// only try to acquire the mutexModule in exclusive mode
if (fExclusive){
// It is possible that another module thread holds the GIL (and s_mutexModule as a result).
// When said thread goes to release the GIL, it will wait for s_mutex, which this thread owns.
// This thread is however waiting for the GIL (and s_mutexModule) that the other thread owns.
// As a result, a deadlock has occured.
// We release the lock on s_mutex and wait until we are able to safely acquire the GIL
// in order to prevent this deadlock from occuring.
while (!s_mutexModule.try_lock())
s_cv.wait(lock);
}
++s_cAcquisitionsModule;
fModuleGILWlocked++;
}
}
int moduleTryAcquireGIL(bool fServerThread, int fExclusive) {
std::unique_lock<std::mutex> lock(s_mutex, std::defer_lock);
if (!lock.try_lock())
return 1;
int *pcheck = fServerThread ? &s_cAcquisitionsModule : &s_cAcquisitionsServer;
if (FModuleCallBackLock(fServerThread)) {
return 0;
}
if (*pcheck > 0)
return 1;
if (fServerThread)
{
if (!s_moduleGIL.tryAcquireRead())
return 1;
++s_cAcquisitionsServer;
}
else
{
if (!s_moduleGIL.tryAcquireWrite(fExclusive))
return 1;
// only try to acquire the mutexModule in exclusive mode
if (fExclusive){
if (!s_mutexModule.try_lock())
return 1;
}
++s_cAcquisitionsModule;
fModuleGILWlocked++;
}
return 0;
}
void moduleReleaseGIL(int fServerThread, int fExclusive) {
std::unique_lock<std::mutex> lock(s_mutex);
if (FModuleCallBackLock(fServerThread)) {
return;
}
if (fServerThread)
{
s_moduleGIL.releaseRead();
--s_cAcquisitionsServer;
}
else
{
s_moduleGIL.releaseWrite(fExclusive);
// only try to release the mutexModule in exclusive mode
if (fExclusive)
s_mutexModule.unlock();
--s_cAcquisitionsModule;
fModuleGILWlocked--;
}
s_cv.notify_all();
}
int moduleGILAcquiredByModule(void) {
return s_moduleGIL.hasWriter();
return fModuleGILWlocked > 0;
}
@@ -8493,7 +8527,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. */
selectDb(ctx.client, 0);
selectDbNamespaced(ctx.client, 0);
/* Event specific context and data pointer setup. */
if (eid == REDISMODULE_EVENT_CLIENT_CHANGE) {
+82 -29
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,6 +172,7 @@ 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;
@@ -1942,9 +1943,7 @@ 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)
@@ -2734,7 +2733,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 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",
"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",
(unsigned long long) client->id,
getClientPeerId(client),
getClientSockname(client),
@@ -2743,6 +2742,7 @@ 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,7 +2757,8 @@ 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->flags & CLIENT_TRACKING) ? (long long) client->client_tracking_redirection : -1,
client->ns ? client->ns->name : "(unset)");
}
sds getAllClientsInfoString(int type) {
@@ -2838,7 +2839,7 @@ void resetCommand(client *c) {
}
if (c->flags & CLIENT_TRACKING) disableTracking(c);
selectDb(c,0);
c->resp = 2;
clientSetDefaultAuth(c);
@@ -2848,6 +2849,9 @@ 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;
@@ -3861,37 +3865,86 @@ int checkClientPauseTimeoutAndReturnIfPaused(void) {
*
* The function returns the total number of events processed. */
void processEventsWhileBlocked(int iel) {
serverAssert(GlobalLocksAcquired());
int iterations = 4; /* See the function top-comment. */
int eventsCount = 0;
executeWithoutGlobalLock([&](){
int iterations = 4; /* See the function top-comment. */
try
{
ProcessingEventsWhileBlocked = 1;
while (iterations--) {
long long startval = g_pserver->events_processed_while_blocked;
long long ae_events = aeProcessEvents(g_pserver->rgthreadvar[iel].el,
AE_FILE_EVENTS|AE_DONT_WAIT|
AE_CALL_BEFORE_SLEEP|AE_CALL_AFTER_SLEEP);
/* Note that g_pserver->events_processed_while_blocked will also get
* incremeted by callbacks called by the event loop handlers. */
eventsCount += ae_events;
long long events = eventsCount - startval;
if (!events) break;
}
ProcessingEventsWhileBlocked = 0;
std::vector<client*> vecclients;
listIter li;
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)
{
client *c = (client*)listNodeValue(ln);
if (c->lock.fOwnLock()) {
serverAssert(c->flags & CLIENT_PROTECTED); // If the client is not protected we have no gurantee they won't be free'd in the event loop
c->lock.unlock();
vecclients.push_back(c);
}
catch (...)
{
ProcessingEventsWhileBlocked = 0;
throw;
}
/* Since we're about to release our lock we need to flush the repl backlog queue */
bool fReplBacklog = g_pserver->repl_batch_offStart >= 0;
if (fReplBacklog) {
flushReplBacklogToClients();
g_pserver->repl_batch_idxStart = -1;
g_pserver->repl_batch_offStart = -1;
}
long long eventsCount = 0;
aeReleaseLock();
serverAssert(!GlobalLocksAcquired());
try
{
ProcessingEventsWhileBlocked = 1;
while (iterations--) {
long long startval = g_pserver->events_processed_while_blocked;
long long ae_events = aeProcessEvents(g_pserver->rgthreadvar[iel].el,
AE_FILE_EVENTS|AE_DONT_WAIT|
AE_CALL_BEFORE_SLEEP|AE_CALL_AFTER_SLEEP);
/* Note that g_pserver->events_processed_while_blocked will also get
* incremeted by callbacks called by the event loop handlers. */
eventsCount += ae_events;
long long events = eventsCount - startval;
if (!events) break;
}
});
ProcessingEventsWhileBlocked = 0;
}
catch (...)
{
// Caller expects us to be locked so fix and rethrow
ProcessingEventsWhileBlocked = 0;
AeLocker locker;
locker.arm(nullptr);
locker.release();
for (client *c : vecclients)
c->lock.lock();
throw;
}
AeLocker locker;
locker.arm(nullptr);
locker.release();
g_pserver->events_processed_while_blocked += eventsCount;
whileBlockedCron();
// Restore it so the calling code is not confused
if (fReplBacklog) {
g_pserver->repl_batch_idxStart = g_pserver->repl_backlog_idx;
g_pserver->repl_batch_offStart = g_pserver->master_repl_offset;
}
for (client *c : vecclients)
c->lock.lock();
// If a different thread processed the shutdown we need to abort the lua command or we will hang
if (serverTL->el->stop)
throw ShutdownException();
+2 -5
View File
@@ -27,17 +27,14 @@ 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
{
if (p != nullptr)
zfree(p);
zfree(p);
}
void operator delete(void *p, std::size_t) noexcept
{
if (p != nullptr)
zfree(p);
zfree(p);
}
#endif
+6 -5
View File
@@ -101,6 +101,7 @@ 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];
@@ -108,7 +109,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;
@@ -118,24 +119,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),dbid);
len = ll2string(buf,sizeof(buf), db->mapped_id);
chan = sdscatlen(chan, buf, len);
chan = sdscatlen(chan, "__:", 3);
chan = sdscatsds(chan, szFromObj(key));
chanobj = createObject(OBJ_STRING, chan);
pubsubPublishMessage(chanobj, eventobj);
pubsubPublishMessage(db->ns, 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),dbid);
if (len == -1) len = ll2string(buf,sizeof(buf),db->mapped_id);
chan = sdscatlen(chan, buf, len);
chan = sdscatlen(chan, "__:", 3);
chan = sdscatsds(chan, szFromObj(eventobj));
chanobj = createObject(OBJ_STRING, chan);
pubsubPublishMessage(chanobj, key);
pubsubPublishMessage(db->ns, chanobj, key);
decrRefCount(chanobj);
}
decrRefCount(eventobj);
+21 -9
View File
@@ -1112,15 +1112,26 @@ struct redisMemOverhead *getMemoryOverheadData(void) {
mh->aof_buffer = mem;
mem_total+=mem;
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))));
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))));
}
}
dictReleaseIterator(di);
mh->lua_caches = mem;
mem_total+=mem;
@@ -1234,7 +1245,8 @@ sds getMemoryDoctorReport(void) {
}
/* Too many scripts are cached? */
if (dictSize(g_pserver->lua_scripts) > 1000) {
//TODO: all namespaces
if (dictSize(g_pserver->default_namespace->lua_scripts) > 1000) {
many_scripts = 1;
num_reports++;
}
+17 -16
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(g_pserver->pubsub_channels,channel);
de = dictFind(c->ns->pubsub_channels,channel);
if (de == NULL) {
clients = listCreate();
dictAdd(g_pserver->pubsub_channels,channel,clients);
dictAdd(c->ns->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(g_pserver->pubsub_channels,channel);
de = dictFind(c->ns->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(g_pserver->pubsub_channels,channel);
dictDelete(c->ns->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(g_pserver->pubsub_patterns,pattern);
de = dictFind(c->ns->pubsub_patterns,pattern);
if (de == NULL) {
clients = listCreate();
dictAdd(g_pserver->pubsub_patterns,pattern,clients);
dictAdd(c->ns->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(g_pserver->pubsub_patterns,pattern);
de = dictFind(c->ns->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(g_pserver->pubsub_patterns,pattern);
dictDelete(c->ns->pubsub_patterns,pattern);
}
}
/* Notify the client */
@@ -291,8 +291,9 @@ int pubsubUnsubscribeAllPatterns(client *c, int notify) {
}
/* Publish a message */
int pubsubPublishMessage(robj *channel, robj *message) {
int pubsubPublishMessage(redisNamespace *ns, robj *channel, robj *message) {
serverAssert(GlobalLocksAcquired());
serverAssert(ns);
int receivers = 0;
dictEntry *de;
dictIterator *di;
@@ -300,7 +301,7 @@ int pubsubPublishMessage(robj *channel, robj *message) {
listIter li;
/* Send to clients listening for that channel */
de = dictFind(g_pserver->pubsub_channels,channel);
de = dictFind(ns->pubsub_channels,channel);
if (de) {
list *list = reinterpret_cast<::list*>(dictGetVal(de));
listNode *ln;
@@ -320,7 +321,7 @@ int pubsubPublishMessage(robj *channel, robj *message) {
}
}
/* Send to clients listening to matching channels */
di = dictGetIterator(g_pserver->pubsub_patterns);
di = dictGetIterator(ns->pubsub_patterns);
if (di) {
channel = getDecodedObject(channel);
while((de = dictNext(di)) != NULL) {
@@ -423,9 +424,9 @@ void punsubscribeCommand(client *c) {
/* PUBLISH <channel> <message> */
void publishCommand(client *c) {
int receivers = pubsubPublishMessage(c->argv[1],c->argv[2]);
int receivers = pubsubPublishMessage(c->ns,c->argv[1],c->argv[2]);
if (g_pserver->cluster_enabled)
clusterPropagatePublish(c->argv[1],c->argv[2]);
clusterPropagatePublish(c->ns,c->argv[1],c->argv[2]);
else
forceCommandPropagation(c,PROPAGATE_REPL);
addReplyLongLong(c,receivers);
@@ -450,7 +451,7 @@ NULL
{
/* PUBSUB CHANNELS [<pattern>] */
sds pat = (c->argc == 2) ? NULL : szFromObj(c->argv[2]);
dictIterator *di = dictGetIterator(g_pserver->pubsub_channels);
dictIterator *di = dictGetIterator(c->ns->pubsub_channels);
dictEntry *de;
long mblen = 0;
void *replylen;
@@ -475,14 +476,14 @@ NULL
addReplyArrayLen(c,(c->argc-2)*2);
for (j = 2; j < c->argc; j++) {
list *l = (list*)dictFetchValue(g_pserver->pubsub_channels,c->argv[j]);
list *l = (list*)dictFetchValue(c->ns->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(g_pserver->pubsub_patterns));
addReplyLongLong(c,dictSize(c->ns->pubsub_patterns));
} else {
addReplySubcommandSyntaxError(c);
}
+60 -68
View File
@@ -1205,40 +1205,6 @@ int rdbSaveInfoAuxFields(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
== -1) return -1;
if (rdbSaveAuxFieldStrInt(rdb,"repl-offset",g_pserver->master_repl_offset)
== -1) return -1;
if (g_pserver->fActiveReplica && listLength(g_pserver->masters) > 0) {
sdsstring val = sdsstring(sdsempty());
listNode *ln;
listIter li;
redisMaster* mi;
listRewind(g_pserver->masters,&li);
while ((ln = listNext(&li)) != NULL) {
mi = (redisMaster*)listNodeValue(ln);
if (!mi->master) {
// If master client is not available, use info from master struct - better than nothing
serverLog(LL_NOTICE, "saving master %s", mi->master_replid);
if (mi->master_replid[0] == 0) {
// if replid is null, there's no reason to save it
continue;
}
val = val.catfmt("%s:%I:%s:%i;", mi->master_replid,
mi->master_initial_offset,
mi->masterhost,
mi->masterport);
}
else {
serverLog(LL_NOTICE, "saving master %s", mi->master->replid);
if (mi->master->replid[0] == 0) {
// if replid is null, there's no reason to save it
continue;
}
val = val.catfmt("%s:%I:%s:%i;", mi->master->replid,
mi->master->reploff,
mi->masterhost,
mi->masterport);
}
}
if (rdbSaveAuxFieldStrStr(rdb, "repl-masters",val.get()) == -1) return -1;
}
}
if (rdbSaveAuxFieldStrInt(rdb,"aof-preamble",aof_preamble) == -1) return -1;
return 1;
@@ -1317,6 +1283,7 @@ 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;
@@ -1336,7 +1303,8 @@ 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 (dictSize(d) == 0) continue;
if (db->mapped_id < 0) continue;
serverAssert(db->ns);
di = dictGetSafeIterator(d);
/* Write the SELECT DB opcode */
@@ -1350,10 +1318,13 @@ 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((de = dictNext(di)) != NULL) {
while(db_size != 0 && (de = dictNext(di)) != NULL) {
sds keystr = (sds)dictGetKey(de);
robj *o = (robj*)dictGetVal(de);
@@ -1383,15 +1354,27 @@ 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 && 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;
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. */
}
}
dictReleaseIterator(di);
di = NULL; /* So that we don't release it again on error. */
dictReleaseIterator(di_ns);
di_ns = NULL;
}
if (rdbSaveModulesAux(rdb, REDISMODULE_AUX_AFTER_RDB) == -1) goto werr;
@@ -1409,6 +1392,7 @@ 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;
}
@@ -2585,6 +2569,7 @@ 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;
@@ -2604,7 +2589,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;
@@ -2652,6 +2637,7 @@ 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
@@ -2687,28 +2673,11 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
memcpy(rsi->repl_id,ptrFromObj(auxval),CONFIG_RUN_ID_SIZE+1);
rsi->repl_id_is_set = 1;
}
} else if (!strcasecmp(szFromObj(auxkey),"repl-masters")) {
if (rsi) {
struct redisMaster mi;
char *masters = szFromObj(auxval);
char *entry = strtok(masters, ":");
while (entry != NULL) {
memcpy(mi.master_replid, entry, sizeof(mi.master_replid));
entry = strtok(NULL, ":");
mi.master_initial_offset = atoi(entry);
entry = strtok(NULL, ":");
mi.masterhost = entry;
entry = strtok(NULL, ";");
mi.masterport = atoi(entry);
entry = strtok(NULL, ":");
rsi->addMaster(mi);
}
}
} else if (!strcasecmp(szFromObj(auxkey),"repl-offset")) {
if (rsi) rsi->repl_offset = strtoll(szFromObj(auxval),NULL,10);
} else if (!strcasecmp(szFromObj(auxkey),"lua")) {
/* Load the script back in memory. */
if (luaCreateFunction(NULL,g_pserver->lua,auxval) == NULL) {
if (luaCreateFunction(lua_client, lua_client->ns->lua, auxval) == NULL) {
rdbReportCorruptRDB(
"Can't load Lua script from RDB file! "
"BODY: %s", (char*)ptrFromObj(auxval));
@@ -2761,6 +2730,25 @@ 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. */
@@ -2868,11 +2856,11 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
bool fExpiredKey = iAmMaster() && !(rdbflags&RDBFLAGS_AOF_PREAMBLE) && expiretime != -1 && expiretime < now;
if (fStaleMvccKey || fExpiredKey) {
if (fStaleMvccKey && !fExpiredKey && rsi != nullptr && rsi->masters != nullptr && rsi->masters[0].staleKeyMap != nullptr && lookupKeyRead(db, &keyobj) == nullptr) {
if (fStaleMvccKey && !fExpiredKey && rsi != nullptr && rsi->mi != nullptr && rsi->mi->staleKeyMap != nullptr && lookupKeyRead(db, &keyobj) == nullptr) {
// We have a key that we've already deleted and is not back in our database.
// We'll need to inform the sending master of the delete if it is also a replica of us
robj_sharedptr objKeyDup(createStringObject(key, sdslen(key)));
rsi->masters[0].staleKeyMap->operator[](db - g_pserver->db).push_back(objKeyDup);
rsi->mi->staleKeyMap->operator[](db - g_pserver->db).push_back(objKeyDup);
}
fLastKeyExpired = true;
sdsfree(key);
@@ -2937,6 +2925,8 @@ 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;
@@ -2983,6 +2973,8 @@ 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");
@@ -3297,8 +3289,8 @@ void bgsaveCommand(client *c) {
* is returned, and the RDB saving will not persist any replication related
* information. */
rdbSaveInfo *rdbPopulateSaveInfo(rdbSaveInfo *rsi) {
rdbSaveInfo rsi_init;
*rsi = std::move(rsi_init);
rdbSaveInfo rsi_init = RDB_SAVE_INFO_INIT;
*rsi = rsi_init;
/* If the instance is a master, we can populate the replication info
* only when repl_backlog is not NULL. If the repl_backlog is NULL,
-115
View File
@@ -1,115 +0,0 @@
#pragma once
#include <condition_variable>
class readWriteLock {
fastlock m_readLock;
fastlock m_writeLock;
std::condition_variable_any m_cv;
int m_readCount = 0;
int m_writeCount = 0;
bool m_writeWaiting = false;
public:
readWriteLock(const char *name) : m_readLock(name), m_writeLock(name) {}
void acquireRead() {
std::unique_lock<fastlock> rm(m_readLock);
while (m_writeCount > 0 || m_writeWaiting)
m_cv.wait(rm);
m_readCount++;
}
bool tryAcquireRead() {
std::unique_lock<fastlock> rm(m_readLock, std::defer_lock);
if (!rm.try_lock())
return false;
if (m_writeCount > 0 || m_writeWaiting)
return false;
m_readCount++;
return true;
}
void acquireWrite(bool exclusive = true) {
std::unique_lock<fastlock> rm(m_readLock);
m_writeWaiting = true;
while (m_readCount > 0)
m_cv.wait(rm);
if (exclusive) {
/* Another thread might have the write lock while we have the read lock
but won't be able to release it until they can acquire the read lock
so release the read lock and try again instead of waiting to avoid deadlock */
while(!m_writeLock.try_lock())
m_cv.wait(rm);
}
m_writeCount++;
m_writeWaiting = false;
}
void upgradeWrite(bool exclusive = true) {
std::unique_lock<fastlock> rm(m_readLock);
m_writeWaiting = true;
while (m_readCount > 1)
m_cv.wait(rm);
if (exclusive) {
/* Another thread might have the write lock while we have the read lock
but won't be able to release it until they can acquire the read lock
so release the read lock and try again instead of waiting to avoid deadlock */
while(!m_writeLock.try_lock())
m_cv.wait(rm);
}
m_writeCount++;
m_readCount--;
m_writeWaiting = false;
}
bool tryAcquireWrite(bool exclusive = true) {
std::unique_lock<fastlock> rm(m_readLock, std::defer_lock);
if (!rm.try_lock())
return false;
if (m_readCount > 0)
return false;
if (exclusive)
if (!m_writeLock.try_lock())
return false;
m_writeCount++;
return true;
}
void releaseRead() {
std::unique_lock<fastlock> rm(m_readLock);
serverAssert(m_readCount > 0);
m_readCount--;
m_cv.notify_all();
}
void releaseWrite(bool exclusive = true) {
std::unique_lock<fastlock> rm(m_readLock);
serverAssert(m_writeCount > 0);
if (exclusive)
m_writeLock.unlock();
m_writeCount--;
m_cv.notify_all();
}
void downgradeWrite(bool exclusive = true) {
std::unique_lock<fastlock> rm(m_readLock);
serverAssert(m_writeCount > 0);
if (exclusive)
m_writeLock.unlock();
m_writeCount--;
while (m_writeCount > 0 || m_writeWaiting)
m_cv.wait(rm);
m_readCount++;
}
bool hasReader() {
return m_readCount > 0;
}
bool hasWriter() {
return m_writeCount > 0;
}
bool writeWaiting() {
return m_writeWaiting;
}
};
+1
View File
@@ -1984,6 +1984,7 @@ 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;
+53 -50
View File
@@ -1987,7 +1987,7 @@ void readSyncBulkPayload(connection *conn) {
ssize_t nread, readlen, nwritten;
int use_diskless_load = useDisklessLoad();
dbBackup *diskless_load_backup = NULL;
rdbSaveInfo rsi;
rdbSaveInfo rsi = RDB_SAVE_INFO_INIT;
int empty_db_flags = g_pserver->repl_slave_lazy_flush ? EMPTYDB_ASYNC :
EMPTYDB_NO_FLAGS;
off_t left;
@@ -2308,7 +2308,7 @@ void readSyncBulkPayload(connection *conn) {
mi->staleKeyMap->clear();
else
mi->staleKeyMap = new (MALLOC_LOCAL) std::map<int, std::vector<robj_sharedptr>>();
rsi.addMaster(*mi);
rsi.mi = mi;
}
if (rdbLoadFile(rdb_filename,&rsi,RDBFLAGS_REPLICATION) != C_OK) {
serverLog(LL_WARNING,
@@ -2359,7 +2359,7 @@ void readSyncBulkPayload(connection *conn) {
{
mergeReplicationId(mi->master->replid);
}
else if (!g_pserver->fActiveReplica)
else
{
/* After a full resynchroniziation we use the replication ID and
* offset of the master. The secondary ID / offset are cleared since
@@ -2555,7 +2555,7 @@ int slaveTryPartialResynchronization(redisMaster *mi, connection *conn, int read
* client structure representing the master into g_pserver->master. */
mi->master_initial_offset = -1;
if (mi->cached_master) {
if (mi->cached_master && !g_pserver->fActiveReplica) {
psync_replid = mi->cached_master->replid;
snprintf(psync_offset,sizeof(psync_offset),"%lld", mi->cached_master->reploff+1);
serverLog(LL_NOTICE,"Trying a partial resynchronization (request %s:%s).", psync_replid, psync_offset);
@@ -2654,15 +2654,14 @@ int slaveTryPartialResynchronization(redisMaster *mi, connection *conn, int read
sizeof(g_pserver->replid2));
g_pserver->second_replid_offset = g_pserver->master_repl_offset+1;
if (!g_pserver->fActiveReplica) {
/* Update the cached master ID and our own primary ID to the
* new one. */
memcpy(g_pserver->replid,sznew,sizeof(g_pserver->replid));
memcpy(mi->cached_master->replid,sznew,sizeof(g_pserver->replid));
/* Update the cached master ID and our own primary ID to the
* new one. */
memcpy(g_pserver->replid,sznew,sizeof(g_pserver->replid));
memcpy(mi->cached_master->replid,sznew,sizeof(g_pserver->replid));
/* Disconnect all the sub-slaves: they need to be notified. */
/* Disconnect all the sub-slaves: they need to be notified. */
if (!g_pserver->fActiveReplica)
disconnectSlaves();
}
}
}
@@ -3023,6 +3022,18 @@ retry_connect:
disconnectSlavesExcept(mi->master_uuid); /* Force our slaves to resync with us as well. */
freeReplicationBacklog(); /* Don't allow our chained slaves to PSYNC. */
}
else
{
if (listLength(g_pserver->slaves))
{
changeReplicationId();
clearReplicationId2();
}
else
{
freeReplicationBacklog(); /* Don't allow our chained slaves to PSYNC. */
}
}
/* Fall back to SYNC if needed. Otherwise psync_result == PSYNC_FULLRESYNC
* and the g_pserver->master_replid and master_initial_offset are
@@ -3274,7 +3285,8 @@ void freeMasterInfo(redisMaster *mi)
zfree(mi->masteruser);
if (mi->repl_transfer_tmpfile)
zfree(mi->repl_transfer_tmpfile);
delete mi->staleKeyMap;
if (mi->staleKeyMap != nullptr)
delete mi->staleKeyMap;
if (mi->cached_master != nullptr)
freeClientAsync(mi->cached_master);
if (mi->master != nullptr)
@@ -3668,26 +3680,6 @@ void replicationCacheMasterUsingMyself(redisMaster *mi) {
mi->master = NULL;
}
/* This function is called when reloading master info from an RDB in Active Replica mode.
* It creates a cached master client using the info contained in the redisMaster struct.
*
* Assumes that the passed struct contains valid master info. */
void replicationCacheMasterUsingMaster(redisMaster *mi) {
if (mi->cached_master) {
freeClient(mi->cached_master);
}
replicationCreateMasterClient(mi, NULL, -1);
std::lock_guard<decltype(mi->master->lock)> lock(mi->master->lock);
memcpy(mi->master->replid, mi->master_replid, sizeof(mi->master_replid));
mi->master->reploff = mi->master_initial_offset;
unlinkClient(mi->master);
mi->cached_master = mi->master;
mi->master = NULL;
}
/* Free a cached master, called when there are no longer the conditions for
* a partial resync on reconnection. */
void replicationDiscardCachedMaster(redisMaster *mi) {
@@ -3801,10 +3793,10 @@ void refreshGoodSlavesCount(void) {
*/
/* Initialize the script cache, only called at startup. */
void replicationScriptCacheInit(void) {
g_pserver->repl_scriptcache_size = 10000;
g_pserver->repl_scriptcache_dict = dictCreate(&replScriptCacheDictType,NULL);
g_pserver->repl_scriptcache_fifo = listCreate();
void replicationScriptCacheInit(redisNamespace *ns) {
ns->repl_scriptcache_size = 10000;
ns->repl_scriptcache_dict = dictCreate(&replScriptCacheDictType,NULL);
ns->repl_scriptcache_fifo = listCreate();
}
/* Empty the script cache. Should be called every time we are no longer sure
@@ -3819,38 +3811,49 @@ void replicationScriptCacheInit(void) {
* to reclaim otherwise unused memory.
*/
void replicationScriptCacheFlush(void) {
dictEmpty(g_pserver->repl_scriptcache_dict,NULL);
listRelease(g_pserver->repl_scriptcache_fifo);
g_pserver->repl_scriptcache_fifo = listCreate();
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);
}
/* Add an entry into the script cache, if we reach max number of entries the
* oldest is removed from the list. */
void replicationScriptCacheAdd(sds sha1) {
void replicationScriptCacheAdd(redisNamespace *ns, sds sha1) {
int retval;
sds key = sdsdup(sha1);
/* Evict oldest. */
if (listLength(g_pserver->repl_scriptcache_fifo) == g_pserver->repl_scriptcache_size)
if (listLength(ns->repl_scriptcache_fifo) == ns->repl_scriptcache_size)
{
listNode *ln = listLast(g_pserver->repl_scriptcache_fifo);
listNode *ln = listLast(ns->repl_scriptcache_fifo);
sds oldest = (sds)listNodeValue(ln);
retval = dictDelete(g_pserver->repl_scriptcache_dict,oldest);
retval = dictDelete(ns->repl_scriptcache_dict,oldest);
serverAssert(retval == DICT_OK);
listDelNode(g_pserver->repl_scriptcache_fifo,ln);
listDelNode(ns->repl_scriptcache_fifo,ln);
}
/* Add current. */
retval = dictAdd(g_pserver->repl_scriptcache_dict,key,NULL);
listAddNodeHead(g_pserver->repl_scriptcache_fifo,key);
retval = dictAdd(ns->repl_scriptcache_dict,key,NULL);
listAddNodeHead(ns->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(sds sha1) {
return dictFind(g_pserver->repl_scriptcache_dict,sha1) != NULL;
int replicationScriptCacheExists(redisNamespace *ns, sds sha1) {
return dictFind(ns->repl_scriptcache_dict,sha1) != NULL;
}
/* ----------------------- SYNCHRONOUS REPLICATION --------------------------
@@ -4224,8 +4227,7 @@ 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 &&
listLength(g_pserver->repl_scriptcache_fifo) != 0)
g_pserver->aof_state == AOF_OFF)
{
replicationScriptCacheFlush();
}
@@ -4706,6 +4708,7 @@ 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;
+39 -36
View File
@@ -605,6 +605,7 @@ 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);
@@ -1144,31 +1145,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 scriptingInit(int setup) {
lua_State *lua = lua_open();
void scriptingSetup() {
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();
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) {
lua_State *lua = lua_open();
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. */
g_pserver->lua_scripts = dictCreate(&shaScriptObjectDictType,NULL);
g_pserver->lua_scripts_mem = 0;
ns->lua_scripts = dictCreate(&shaScriptObjectDictType,NULL);
ns->lua_scripts_mem = 0;
/* Register the redis commands table and fields */
lua_newtable(lua);
@@ -1316,23 +1317,23 @@ void scriptingInit(int setup) {
* to global variables. */
scriptingEnableGlobalsProtection(lua);
g_pserver->lua = lua;
ns->lua = lua;
}
/* Release resources related to Lua scripting.
* This function is used in order to reset the scripting environment. */
void scriptingRelease(int async) {
void scriptingRelease(redisNamespace *ns, int async) {
if (async)
freeLuaScriptsAsync(g_pserver->lua_scripts);
freeLuaScriptsAsync(ns->lua_scripts);
else
dictRelease(g_pserver->lua_scripts);
g_pserver->lua_scripts_mem = 0;
lua_close(g_pserver->lua);
dictRelease(ns->lua_scripts);
ns->lua_scripts_mem = 0;
lua_close(ns->lua);
}
void scriptingReset(int async) {
scriptingRelease(async);
scriptingInit(0);
void scriptingReset(redisNamespace *ns, int async) {
scriptingRelease(ns, async);
scriptingInit(ns);
}
/* Set an array of Redis String Objects as a Lua array (table) stored into a
@@ -1421,7 +1422,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(g_pserver->lua_scripts,sha)) != NULL) {
if ((de = dictFind(c->ns->lua_scripts,sha)) != NULL) {
sdsfree(sha);
return (sds)dictGetKey(de);
}
@@ -1459,9 +1460,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(g_pserver->lua_scripts,sha,body);
int retval = dictAdd(c->ns->lua_scripts,sha,body);
serverAssertWithInfo(c ? c : serverTL->lua_client,NULL,retval == DICT_OK);
g_pserver->lua_scripts_mem += sdsZmallocSize(sha) + getStringObjectSdsUsedMemory(body);
c->ns->lua_scripts_mem += sdsZmallocSize(sha) + getStringObjectSdsUsedMemory(body);
incrRefCount(body);
return sha;
}
@@ -1522,7 +1523,7 @@ void resetLuaClient(void) {
}
void evalGenericCommand(client *c, int evalsha) {
lua_State *lua = g_pserver->lua;
lua_State *lua = c->ns->lua;
char funcname[43];
long long numkeys;
long long initial_server_dirty = g_pserver->dirty;
@@ -1626,7 +1627,7 @@ void evalGenericCommand(client *c, int evalsha) {
lua_sethook(lua,luaMaskCountHook,LUA_MASKCOUNT,100000);
delhook = 1;
} else if (ldb.active) {
lua_sethook(g_pserver->lua,luaLdbLineHook,LUA_MASKLINE|LUA_MASKCOUNT,100000);
lua_sethook(c->ns->lua,luaLdbLineHook,LUA_MASKLINE|LUA_MASKCOUNT,100000);
delhook = 1;
}
@@ -1709,13 +1710,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((sds)ptrFromObj(c->argv[1]))) {
if (!replicationScriptCacheExists(c->ns, (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(g_pserver->lua_scripts,ptrFromObj(c->argv[1]));
robj *script = (robj*)dictFetchValue(c->ns->lua_scripts,ptrFromObj(c->argv[1]));
replicationScriptCacheAdd((sds)ptrFromObj(c->argv[1]));
replicationScriptCacheAdd(c->ns, (sds)ptrFromObj(c->argv[1]));
serverAssertWithInfo(c,NULL,script != NULL);
/* If the script did not produce any changes in the dataset we want
@@ -1798,7 +1799,7 @@ NULL
addReplyError(c,"SCRIPT FLUSH only support SYNC|ASYNC option");
return;
}
scriptingReset(async);
scriptingReset(c->ns, async);
addReply(c,shared.ok);
replicationScriptCacheFlush();
g_pserver->dirty++; /* Propagating this command is a good idea. */
@@ -1807,13 +1808,13 @@ NULL
addReplyArrayLen(c, c->argc-2);
for (j = 2; j < c->argc; j++) {
if (dictFind(g_pserver->lua_scripts,ptrFromObj(c->argv[j])))
if (dictFind(c->ns->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,g_pserver->lua,c->argv[2]);
sds sha = luaCreateFunction(c, c->ns->lua,c->argv[2]);
if (sha == NULL) return; /* The error was sent by luaCreateFunction(). */
addReplyBulkCBuffer(c,sha,40);
forceCommandPropagation(c,PROPAGATE_REPL|PROPAGATE_AOF);
@@ -1822,6 +1823,8 @@ 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 {
-15
View File
@@ -424,21 +424,6 @@ public:
return *this;
}
sdsstring &operator=(sdsstring &&other)
{
sds tmp = m_str;
m_str = other.m_str;
other.m_str = tmp;
return *this;
}
template<typename... Args>
sdsstring catfmt(const char *fmt, Args... args)
{
m_str = sdscatfmt(m_str, fmt, args...);
return *this;
}
sds release() {
sds sdsT = m_str;
m_str = nullptr;
+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(channel,payload);
pubsubPublishMessage(g_pserver->default_namespace, channel,payload);
decrRefCount(channel);
decrRefCount(payload);
}
+190 -168
View File
@@ -63,11 +63,9 @@
#include <algorithm>
#include <uuid/uuid.h>
#include <mutex>
#include <condition_variable>
#include "aelocker.h"
#include "motd.h"
#include "t_nhash.h"
#include "readwritelock.h"
#ifdef __linux__
#include <sys/prctl.h>
#include <sys/mman.h>
@@ -92,16 +90,10 @@ double R_Zero, R_PosInf, R_NegInf, R_Nan;
/* Global vars */
namespace GlobalHidden {
struct redisServer server; /* Server global state */
readWriteLock forkLock("Fork (global)");
}
redisServer *g_pserver = &GlobalHidden::server;
readWriteLock *g_forkLock = &GlobalHidden::forkLock;
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.
*
@@ -1177,7 +1169,15 @@ struct redisCommand redisCommandTable[] = {
{"failover",failoverCommand,-1,
"admin no-script ok-stale",
0,NULL,0,0,0,0,0,0}
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},
};
/*============================ Utility functions ============================ */
@@ -1610,6 +1610,18 @@ 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 */
@@ -2015,7 +2027,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() {
void updateCachedTime(int update_daylight_info) {
long long t = ustime();
__atomic_store(&g_pserver->ustime, &t, __ATOMIC_RELAXED);
t /= 1000;
@@ -2028,10 +2040,12 @@ void updateCachedTime() {
* 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. */
struct tm tm;
time_t ut = g_pserver->unixtime;
localtime_r(&ut,&tm);
__atomic_store(&g_pserver->daylight_active, &tm.tm_isdst, __ATOMIC_RELAXED);
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);
}
}
void checkChildrenDone(void) {
@@ -2110,8 +2124,17 @@ 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 */
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;
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);
}
if (!g_pserver->cron_malloc_stats.allocator_active)
g_pserver->cron_malloc_stats.allocator_active = g_pserver->cron_malloc_stats.allocator_resident;
@@ -2158,6 +2181,9 @@ 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. */
@@ -2415,6 +2441,7 @@ 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;
}
}
@@ -2625,15 +2652,6 @@ 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);
}
g_forkLock->releaseRead();
/* 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();
@@ -2654,8 +2672,6 @@ void afterSleep(struct aeEventLoop *eventLoop) {
Otherwise you may double acquire the GIL and cause deadlocks in the module */
if (!ProcessingEventsWhileBlocked) {
if (serverTL->modulesEnabledThisAeLoop) moduleAcquireGIL(TRUE /*fServerThread*/);
g_forkLock->acquireRead();
wakeTimeThread();
}
}
@@ -2746,6 +2762,7 @@ 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);
@@ -2754,6 +2771,7 @@ 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);
@@ -2852,7 +2870,7 @@ void initMasterInfo(redisMaster *master)
void initServerConfig(void) {
int j;
updateCachedTime();
updateCachedTime(true);
getRandomHexChars(g_pserver->runid,CONFIG_RUN_ID_SIZE);
g_pserver->runid[CONFIG_RUN_ID_SIZE] = '\0';
changeReplicationId();
@@ -2919,6 +2937,7 @@ 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;
@@ -2975,6 +2994,7 @@ 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;
@@ -3361,6 +3381,16 @@ 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
@@ -3485,13 +3515,47 @@ static void initServerThread(struct redisServerThreadVars *pvar, int fMain)
}
}
void initServer(void) {
signal(SIGHUP, SIG_IGN);
signal(SIGPIPE, SIG_IGN);
setupSignalHandlers();
makeThreadKillable();
sds generateAutoNamespaceName() {
char uuid[37];
uuid_t uuid_bin;
uuid_generate(uuid_bin);
uuid_unparse(uuid_bin, uuid);
//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++) {
@@ -3503,11 +3567,20 @@ void initServer(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;
@@ -3554,8 +3627,6 @@ 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;
@@ -3638,8 +3709,7 @@ void initServer(void) {
uuid_generate((unsigned char*)cserver.uuid);
if (g_pserver->cluster_enabled) clusterInit();
replicationScriptCacheInit();
scriptingInit(1);
scriptingSetup();
slowlogInit();
latencyMonitorInit();
@@ -3751,6 +3821,7 @@ void populateCommandTable(void) {
void resetCommandTableStats(void) {
struct redisCommand *c;
redisNamespace *ns;
dictEntry *de;
dictIterator *di;
@@ -3764,6 +3835,28 @@ 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) {
@@ -4019,6 +4112,7 @@ 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 {
@@ -4042,6 +4136,7 @@ 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
@@ -4101,6 +4196,8 @@ 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 */
@@ -4206,6 +4303,7 @@ 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;
@@ -5122,7 +5220,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->lua ? (long long)lua_gc(g_pserver->lua,LUA_GCCOUNT,0)*1024 : 0;
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
struct redisMemOverhead *mh = getMemoryOverheadData();
/* Peak memory is updated from time to time by serverCron() so it
@@ -5204,7 +5302,7 @@ sds genRedisInfoString(const char *section) {
used_memory_lua_hmem,
(long long) mh->lua_caches,
used_memory_scripts_hmem,
dictSize(g_pserver->lua_scripts),
dictSize(g_pserver->default_namespace->lua_scripts),
g_pserver->maxmemory,
maxmemory_hmem,
evict_policy,
@@ -5421,8 +5519,8 @@ sds genRedisInfoString(const char *section) {
g_pserver->stat_evictedkeys,
g_pserver->stat_keyspace_hits,
g_pserver->stat_keyspace_misses,
dictSize(g_pserver->pubsub_channels),
dictSize(g_pserver->pubsub_patterns),
dictSize(g_pserver->default_namespace->pubsub_channels), //TODO: all namespaces
dictSize(g_pserver->default_namespace->pubsub_patterns), //TODO: all namespaces
g_pserver->stat_fork_time,
g_pserver->stat_total_forks,
dictSize(g_pserver->migrate_cached_sockets),
@@ -5669,6 +5767,31 @@ 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")) {
@@ -6223,63 +6346,6 @@ void closeChildUnusedResourceAfterFork() {
cserver.pidfile = NULL;
}
void executeWithoutGlobalLock(std::function<void()> func) {
serverAssert(GlobalLocksAcquired());
std::vector<client*> vecclients;
listIter li;
listNode *ln;
listRewind(g_pserver->clients, &li);
// 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)
{
client *c = (client*)listNodeValue(ln);
if (c->lock.fOwnLock()) {
serverAssert(c->flags & CLIENT_PROTECTED || c->flags & CLIENT_EXECUTING_COMMAND); // If the client is not protected we have no gurantee they won't be free'd in the event loop
c->lock.unlock();
vecclients.push_back(c);
}
}
/* Since we're about to release our lock we need to flush the repl backlog queue */
bool fReplBacklog = g_pserver->repl_batch_offStart >= 0;
if (fReplBacklog) {
flushReplBacklogToClients();
g_pserver->repl_batch_idxStart = -1;
g_pserver->repl_batch_offStart = -1;
}
aeReleaseLock();
serverAssert(!GlobalLocksAcquired());
try {
func();
}
catch (...) {
// Caller expects us to be locked so fix and rethrow
AeLocker locker;
locker.arm(nullptr);
locker.release();
for (client *c : vecclients)
c->lock.lock();
throw;
}
AeLocker locker;
locker.arm(nullptr);
locker.release();
// Restore it so the calling code is not confused
if (fReplBacklog) {
g_pserver->repl_batch_idxStart = g_pserver->repl_backlog_idx;
g_pserver->repl_batch_offStart = g_pserver->master_repl_offset;
}
for (client *c : vecclients)
c->lock.lock();
}
/* purpose is one of CHILD_TYPE_ types */
int redisFork(int purpose) {
int childpid;
@@ -6291,10 +6357,7 @@ int redisFork(int purpose) {
openChildInfoPipe();
}
long long startWriteLock = ustime();
g_forkLock->releaseRead();
executeWithoutGlobalLock([](){ g_forkLock->acquireWrite(); });
latencyAddSampleIfNeeded("fork-lock",(ustime()-startWriteLock)/1000);
if ((childpid = fork()) == 0) {
/* Child */
g_pserver->in_fork_child = purpose;
@@ -6303,8 +6366,6 @@ int redisFork(int purpose) {
closeChildUnusedResourceAfterFork();
} else {
/* Parent */
g_forkLock->releaseWrite();
g_forkLock->acquireRead();
g_pserver->stat_total_forks++;
g_pserver->stat_fork_time = ustime()-start;
g_pserver->stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / g_pserver->stat_fork_time / (1024*1024*1024); /* GB per second. */
@@ -6367,7 +6428,7 @@ void loadDataFromDisk(void) {
if (loadAppendOnlyFile(g_pserver->aof_filename) == C_OK)
serverLog(LL_NOTICE,"DB loaded from append only file: %.3f seconds",(float)(ustime()-start)/1000000);
} else if (g_pserver->rdb_filename != NULL || g_pserver->rdb_s3bucketpath != NULL) {
rdbSaveInfo rsi;
rdbSaveInfo rsi = RDB_SAVE_INFO_INIT;
errno = 0; /* Prevent a stale value from affecting error checking */
if (rdbLoad(&rsi,RDBFLAGS_NONE) == C_OK) {
serverLog(LL_NOTICE,"DB loaded from disk: %.3f seconds",
@@ -6395,23 +6456,11 @@ void loadDataFromDisk(void) {
while ((ln = listNext(&li)))
{
redisMaster *mi = (redisMaster*)listNodeValue(ln);
if (g_pserver->fActiveReplica) {
for (size_t i = 0; i < rsi.numMasters(); i++) {
if (!strcmp(mi->masterhost, rsi.masters[i].masterhost) && mi->masterport == rsi.masters[i].masterport) {
memcpy(mi->master_replid, rsi.masters[i].master_replid, sizeof(mi->master_replid));
mi->master_initial_offset = rsi.masters[i].master_initial_offset;
replicationCacheMasterUsingMaster(mi);
serverLog(LL_NOTICE, "Cached master recovered from RDB for %s:%d", mi->masterhost, mi->masterport);
}
}
}
else {
/* If we are a replica, create a cached master from this
* information, in order to allow partial resynchronizations
* with masters. */
replicationCacheMasterUsingMyself(mi);
selectDb(mi->cached_master,rsi.repl_stream_db);
}
/* If we are a replica, create a cached master from this
* information, in order to allow partial resynchronizations
* with masters. */
replicationCacheMasterUsingMyself(mi);
selectDb(mi->cached_master,rsi.repl_stream_db);
}
}
} else if (errno != ENOENT) {
@@ -6620,47 +6669,6 @@ 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;
int cycle_count = 0;
g_forkLock->acquireRead();
while (true) {
{
std::unique_lock<std::mutex> lock(time_thread_mutex);
if (sleeping_threads >= cserver.cthreads) {
g_forkLock->releaseRead();
time_thread_cv.wait(lock);
g_forkLock->acquireRead();
cycle_count = 0;
}
}
updateCachedTime();
if (cycle_count == MAX_CYCLES_TO_HOLD_FORK_LOCK) {
g_forkLock->releaseRead();
g_forkLock->acquireRead();
cycle_count = 0;
}
#if defined(__APPLE__)
nanosleep(&delay, nullptr);
#else
clock_nanosleep(CLOCK_MONOTONIC, 0, &delay, NULL);
#endif
cycle_count++;
}
g_forkLock->releaseRead();
}
void *workerThreadMain(void *parg)
{
int iel = (int)((int64_t)parg);
@@ -6676,7 +6684,6 @@ void *workerThreadMain(void *parg)
}
moduleAcquireGIL(true); // Normally afterSleep acquires this, but that won't be called on the first run
g_forkLock->acquireRead();
aeEventLoop *el = g_pserver->rgthreadvar[iel].el;
try
{
@@ -6685,7 +6692,6 @@ void *workerThreadMain(void *parg)
catch (ShutdownException)
{
}
g_forkLock->releaseRead();
moduleReleaseGIL(true);
serverAssert(!GlobalLocksAcquired());
aeDeleteEventLoop(el);
@@ -6946,6 +6952,24 @@ 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);
@@ -6995,7 +7019,7 @@ int main(int argc, char **argv) {
__AFL_INIT();
#endif
rio rdb;
rdbSaveInfo rsi;
rdbSaveInfo rsi = RDB_SAVE_INFO_INIT;
startLoadingFile(stdin, (char*)"stdin", 0);
rioInitWithFile(&rdb,stdin);
rdbLoadRio(&rdb,0,&rsi);
@@ -7059,8 +7083,6 @@ 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
+95 -125
View File
@@ -96,7 +96,6 @@ typedef long long ustime_t; /* microsecond time type. */
#include "connection.h" /* Connection abstraction */
#include "serverassert.h"
#include "expire.h"
#include "readwritelock.h"
#define REDISMODULE_CORE 1
#include "redismodule.h" /* Redis modules API defines. */
@@ -734,9 +733,6 @@ typedef enum {
#define REDISMODULE_AUX_BEFORE_RDB (1<<0)
#define REDISMODULE_AUX_AFTER_RDB (1<<1)
/* Number of cycles before time thread gives up fork lock */
#define MAX_CYCLES_TO_HOLD_FORK_LOCK 10
struct RedisModule;
struct RedisModuleIO;
struct RedisModuleDigest;
@@ -974,6 +970,8 @@ 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. */
@@ -984,6 +982,7 @@ struct redisDb {
~redisDb();
redisNamespace *ns;
::dict *dict; /* The keyspace for this DB */
expireset *setexpire;
expireset::setiter expireitr;
@@ -991,12 +990,33 @@ 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; /* Database ID */
int id; /* Global database ID */
int mapped_id; /* Namespaced 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. */
@@ -1106,6 +1126,7 @@ 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
@@ -1145,6 +1166,7 @@ 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. */
@@ -1290,7 +1312,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,
*hdel, *zrem, *mvccrestore, *pexpirememberat, *redacted, *allocate,
*select[PROTO_SHARED_SELECT_CMDS],
*integers[OBJ_SHARED_INTEGERS],
*mbulkhdr[OBJ_SHARED_BULKHDR_LEN], /* "*<value>\r\n" */
@@ -1393,38 +1415,6 @@ struct redisMemOverhead {
} *db;
};
struct redisMaster {
char *masteruser; /* AUTH with this user and masterauth with master */
char *masterauth; /* AUTH with this password with master */
char *masterhost; /* Hostname of master */
int masterport; /* Port of master */
client *cached_master; /* Cached master to be reused for PSYNC. */
client *master;
/* The following two fields is where we store master PSYNC replid/offset
* while the PSYNC is in progress. At the end we'll copy the fields into
* the server->master client structure. */
char master_replid[CONFIG_RUN_ID_SIZE+1]; /* Master PSYNC runid. */
long long master_initial_offset; /* Master PSYNC offset. */
bool isActive = false;
int repl_state; /* Replication status if the instance is a replica */
off_t repl_transfer_size; /* Size of RDB to read from master during sync. */
off_t repl_transfer_read; /* Amount of RDB read from master during sync. */
off_t repl_transfer_last_fsync_off; /* Offset when we fsync-ed last time. */
connection *repl_transfer_s; /* Slave -> Master SYNC socket */
int repl_transfer_fd; /* Slave -> Master SYNC temp file descriptor */
char *repl_transfer_tmpfile; /* Slave-> master SYNC temp file name */
time_t repl_transfer_lastio; /* Unix time of the latest read, for timeout */
time_t repl_down_since; /* Unix time at which link with master went down */
unsigned char master_uuid[UUID_BINARY_LEN]; /* Used during sync with master, this is our master's UUID */
/* After we've connected with our master use the UUID in g_pserver->master */
uint64_t mvccLastSync;
/* During a handshake the server may have stale keys, we track these here to share once a reciprocal connection is made */
std::map<int, std::vector<robj_sharedptr>> *staleKeyMap;
int ielReplTransfer = -1;
};
/* This structure can be optionally passed to RDB save/load functions in
* order to implement additional functionalities, by storing and loading
* metadata to the RDB file.
@@ -1433,66 +1423,7 @@ struct redisMaster {
* replication in order to make sure that chained slaves (slaves of slaves)
* select the correct DB and are able to accept the stream coming from the
* top-level master. */
class rdbSaveInfo {
public:
rdbSaveInfo() {
repl_stream_db = -1;
repl_id_is_set = 0;
memcpy(repl_id, "0000000000000000000000000000000000000000", sizeof(repl_id));
repl_offset = -1;
fForceSetKey = TRUE;
mvccMinThreshold = 0;
masters = nullptr;
masterCount = 0;
}
rdbSaveInfo(const rdbSaveInfo &other) {
repl_stream_db = other.repl_stream_db;
repl_id_is_set = other.repl_id_is_set;
memcpy(repl_id, other.repl_id, sizeof(repl_id));
repl_offset = other.repl_offset;
fForceSetKey = other.fForceSetKey;
mvccMinThreshold = other.mvccMinThreshold;
masters = (struct redisMaster*)malloc(sizeof(struct redisMaster) * other.masterCount);
memcpy(masters, other.masters, sizeof(struct redisMaster) * other.masterCount);
masterCount = other.masterCount;
}
rdbSaveInfo(rdbSaveInfo &&other) : rdbSaveInfo() {
swap(*this, other);
}
rdbSaveInfo &operator=(rdbSaveInfo other) {
swap(*this, other);
return *this;
}
~rdbSaveInfo() {
free(masters);
}
friend void swap(rdbSaveInfo &first, rdbSaveInfo &second) {
std::swap(first.repl_stream_db, second.repl_stream_db);
std::swap(first.repl_id_is_set, second.repl_id_is_set);
std::swap(first.repl_id, second.repl_id);
std::swap(first.repl_offset, second.repl_offset);
std::swap(first.fForceSetKey, second.fForceSetKey);
std::swap(first.mvccMinThreshold, second.mvccMinThreshold);
std::swap(first.masters, second.masters);
std::swap(first.masterCount, second.masterCount);
}
void addMaster(const struct redisMaster &mi) {
masterCount++;
if (masters == nullptr) {
masters = (struct redisMaster*)malloc(sizeof(struct redisMaster));
}
else {
masters = (struct redisMaster*)realloc(masters, sizeof(struct redisMaster) * masterCount);
}
memcpy(masters + masterCount - 1, &mi, sizeof(struct redisMaster));
}
size_t numMasters() {
return masterCount;
}
typedef struct rdbSaveInfo {
/* Used saving and loading. */
int repl_stream_db; /* DB to select in g_pserver->master client. */
@@ -1502,11 +1433,10 @@ public:
long long repl_offset; /* Replication offset. */
int fForceSetKey;
uint64_t mvccMinThreshold;
struct redisMaster *masters;
struct redisMaster *mi;
} rdbSaveInfo;
private:
size_t masterCount;
};
#define RDB_SAVE_INFO_INIT {-1,0,"0000000000000000000000000000000000000000",-1, TRUE, 0, nullptr}
struct malloc_stats {
size_t zmalloc_used;
@@ -1595,12 +1525,48 @@ 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 */
char *masterhost; /* Hostname of master */
int masterport; /* Port of master */
client *cached_master; /* Cached master to be reused for PSYNC. */
client *master;
/* The following two fields is where we store master PSYNC replid/offset
* while the PSYNC is in progress. At the end we'll copy the fields into
* the server->master client structure. */
char master_replid[CONFIG_RUN_ID_SIZE+1]; /* Master PSYNC runid. */
long long master_initial_offset; /* Master PSYNC offset. */
bool isActive = false;
int repl_state; /* Replication status if the instance is a replica */
off_t repl_transfer_size; /* Size of RDB to read from master during sync. */
off_t repl_transfer_read; /* Amount of RDB read from master during sync. */
off_t repl_transfer_last_fsync_off; /* Offset when we fsync-ed last time. */
connection *repl_transfer_s; /* Slave -> Master SYNC socket */
int repl_transfer_fd; /* Slave -> Master SYNC temp file descriptor */
char *repl_transfer_tmpfile; /* Slave-> master SYNC temp file name */
time_t repl_transfer_lastio; /* Unix time of the latest read, for timeout */
time_t repl_down_since; /* Unix time at which link with master went down */
unsigned char master_uuid[UUID_BINARY_LEN]; /* Used during sync with master, this is our master's UUID */
/* After we've connected with our master use the UUID in g_pserver->master */
uint64_t mvccLastSync;
/* During a handshake the server may have stale keys, we track these here to share once a reciprocal connection is made */
std::map<int, std::vector<robj_sharedptr>> *staleKeyMap;
int ielReplTransfer = -1;
};
// Const vars are not changed after worker threads are launched
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). */
@@ -1616,11 +1582,12 @@ struct redisServerConst {
*zpopmaxCommand, *sremCommand, *execCommand,
*expireCommand, *pexpireCommand, *xclaimCommand,
*xgroupCommand, *rreplayCommand, *rpoplpushCommand,
*hdelCommand, *zremCommand, *lmoveCommand;
*hdelCommand, *zremCommand, *lmoveCommand, *allocateCommand;
/* 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. */
@@ -1636,6 +1603,7 @@ 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 */
@@ -1664,7 +1632,10 @@ 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. */
@@ -1899,7 +1870,8 @@ struct redisServer {
int repl_diskless_sync_delay; /* Delay to start a diskless repl BGSAVE. */
/* Replication (replica) */
list *masters;
int enable_multimaster;
list *repl_init_masters; /* temp variable to hold the master ip/port pairs to init after config parsing */
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? */
@@ -1912,10 +1884,6 @@ 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. */
@@ -1966,8 +1934,6 @@ 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 */
@@ -1994,11 +1960,8 @@ 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 */
@@ -2177,7 +2140,6 @@ typedef struct {
//extern struct redisServer server;
extern redisServer *g_pserver;
extern readWriteLock *g_forkLock;
extern struct redisServerConst cserver;
extern __thread struct redisServerThreadVars *serverTL; // thread local server vars
extern struct sharedObjectsStruct shared;
@@ -2200,6 +2162,10 @@ extern dictType sdsReplyDictType;
* Functions prototypes
*----------------------------------------------------------------------------*/
/* Namespaces */
redisNamespace *getNamespace(const char* name);
sds generateAutoNamespaceName();
/* Modules */
void moduleInitModulesSystem(void);
void moduleInitModulesSystemLast(void);
@@ -2489,10 +2455,10 @@ void resizeReplicationBacklog(long long newsize);
struct redisMaster *replicationAddMaster(char *ip, int port);
void replicationUnsetMaster(struct redisMaster *mi);
void refreshGoodSlavesCount(void);
void replicationScriptCacheInit(void);
void replicationScriptCacheInit(redisNamespace *ns);
void replicationScriptCacheFlush(void);
void replicationScriptCacheAdd(sds sha1);
int replicationScriptCacheExists(sds sha1);
void replicationScriptCacheAdd(redisNamespace *ns, sds sha1);
int replicationScriptCacheExists(redisNamespace *ns, sds sha1);
void processClientsWaitingReplicas(void);
void unblockClientWaitingReplicas(client *c);
int replicationCountAcksByOffset(long long offset);
@@ -2506,7 +2472,6 @@ void clearReplicationId2(void);
void mergeReplicationId(const char *);
void chopReplicationBacklog(void);
void replicationCacheMasterUsingMyself(struct redisMaster *mi);
void replicationCacheMasterUsingMaster(struct redisMaster *mi);
void feedReplicationBacklog(const void *ptr, size_t len);
void updateMasterAuth();
void showLatestBacklog();
@@ -2562,7 +2527,6 @@ void sendChildInfo(childInfoType info_type, size_t keys, const char *pname);
void receiveChildInfo(void);
/* Fork helpers */
void executeWithoutGlobalLock(std::function<void()> func);
int redisFork(int type);
int hasActiveChildProcess();
void resetChildState();
@@ -2701,11 +2665,12 @@ 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(void);
void updateCachedTime(int update_daylight_info);
void resetServerStats(void);
void activeDefragCycle(void);
unsigned int getLRUClock(void);
@@ -2768,7 +2733,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(robj *channel, robj *message);
int pubsubPublishMessage(redisNamespace *ns, robj *channel, robj *message);
void addReplyPubsubMessage(client *c, robj *channel, robj *msg);
/* Keyspace events notification */
@@ -2828,7 +2793,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(int flags);
void flushAllDataAndResetRDB(redisNamespace *ns, int flags);
long long dbTotalServerKeyCount();
dbBackup *backupDb(void);
void restoreDbBackup(dbBackup *buckup);
@@ -2836,6 +2801,8 @@ 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);
@@ -2875,7 +2842,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(robj *channel, robj *message);
void clusterPropagatePublish(redisNamespace *ns, 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);
@@ -2897,7 +2864,8 @@ int redis_check_rdb_main(int argc, const char **argv, FILE *fp);
int redis_check_aof_main(int argc, char **argv);
/* Scripting */
void scriptingInit(int setup);
void scriptingSetup();
void scriptingInit(redisNamespace *ns);
int ldbRemoveChild(pid_t pid);
void ldbKillForkedSessions(void);
int ldbPendingChildren(void);
@@ -2990,6 +2958,8 @@ 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
@@ -0,0 +1,4 @@
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 ::
+4 -2
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*}
} {*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=*}
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*}
} {*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=*}
test {MONITOR can log executed commands} {
set rd [redis_deferring_client]
@@ -147,6 +147,7 @@ start_server {tags {"introspection"}} {
supervised
syslog-facility
databases
databases-per-namespace
io-threads
logfile
unixsocketperm
@@ -160,6 +161,7 @@ start_server {tags {"introspection"}} {
active-replica
bind
set-proc-title
acl-namespace-default
}
if {!$::tls} {
+101
View File
@@ -0,0 +1,101 @@
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,6 +2,7 @@ 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"
@@ -11,6 +12,7 @@ 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