Commit 819d291b authored by Oran Agra's avatar Oran Agra
Browse files

Merge remote-tracking branch 'origin/unstable' into 7.2

parents a51eb05b 936cfa46
......@@ -23,6 +23,7 @@
]
],
"command_flags": [
"LOADING",
"STALE"
],
"command_tips": [
......
{
"PSUBSCRIBE": {
"summary": "Listens for messages published to channels that match one or more patterns.",
"complexity": "O(N) where N is the number of patterns the client is already subscribed to.",
"complexity": "O(N) where N is the number of patterns to subscribe to.",
"group": "pubsub",
"since": "2.0.0",
"arity": -2,
......
{
"PUNSUBSCRIBE": {
"summary": "Stops listening to messages published to channels that match one or more patterns.",
"complexity": "O(N+M) where N is the number of patterns the client is already subscribed and M is the number of total patterns subscribed in the system (by any client).",
"complexity": "O(N) where N is the number of patterns to unsubscribe.",
"group": "pubsub",
"since": "2.0.0",
"arity": -1,
......
{
"CONFIG": {
"summary": "Configures Redis Sentinel.",
"complexity": "O(1)",
"complexity": "O(N) when N is the number of configuration parameters provided",
"group": "sentinel",
"since": "6.2.0",
"arity": -4,
"container": "SENTINEL",
"function": "sentinelCommand",
"history": [
[
"7.2.0",
"Added the ability to set and get multiple parameters in one call."
]
],
"command_flags": [
"ADMIN",
"SENTINEL",
......@@ -42,7 +48,7 @@
"type": "string"
},
"announce-port": {
"type": "integer"
"type": "string"
},
"sentinel-user": {
"type": "string"
......@@ -64,6 +70,9 @@
{
"const": "warning"
},
{
"const": "nothing"
},
{
"const": "unknown"
}
......@@ -87,6 +96,7 @@
"name":"set",
"token":"SET",
"type":"block",
"multiple": true,
"arguments":[
{
"name":"parameter",
......@@ -101,7 +111,8 @@
{
"token":"GET",
"name":"parameter",
"type":"string"
"type":"string",
"multiple": true
}
]
}
......
{
"SUNSUBSCRIBE": {
"summary": "Stops listening to messages posted to shard channels.",
"complexity": "O(N) where N is the number of clients already subscribed to a shard channel.",
"complexity": "O(N) where N is the number of shard channels to unsubscribe.",
"group": "pubsub",
"since": "7.0.0",
"arity": -1,
......
{
"UNSUBSCRIBE": {
"summary": "Stops listening to messages posted to channels.",
"complexity": "O(N) where N is the number of clients already subscribed to a channel.",
"complexity": "O(N) where N is the number of channels to unsubscribe.",
"group": "pubsub",
"since": "2.0.0",
"arity": -1,
......
......@@ -47,11 +47,11 @@
},
{
"type": "integer",
"description": "The rank of the member when 'WITHSCORES' is not used."
"description": "The rank of the member when 'WITHSCORE' is not used."
},
{
"type": "array",
"description": "The rank and score of the member when 'WITHSCORES' is used.",
"description": "The rank and score of the member when 'WITHSCORE' is used.",
"minItems": 2,
"maxItems": 2,
"items": [
......
......@@ -47,11 +47,11 @@
},
{
"type": "integer",
"description": "The rank of the member when 'WITHSCORES' is not used."
"description": "The rank of the member when 'WITHSCORE' is not used."
},
{
"type": "array",
"description": "The rank and score of the member when 'WITHSCORES' is used.",
"description": "The rank and score of the member when 'WITHSCORE' is used.",
"minItems": 2,
"maxItems": 2,
"items": [
......
......@@ -38,6 +38,7 @@
#include <glob.h>
#include <string.h>
#include <locale.h>
#include <ctype.h>
/*-----------------------------------------------------------------------------
* Config file name-value maps.
......@@ -79,6 +80,7 @@ configEnum loglevel_enum[] = {
{"verbose", LL_VERBOSE},
{"notice", LL_NOTICE},
{"warning", LL_WARNING},
{"nothing", LL_NOTHING},
{NULL,0}
};
......@@ -453,14 +455,13 @@ void loadServerConfigFromString(char *config) {
const char *err = NULL;
int linenum = 0, totlines, i;
sds *lines;
sds *argv = NULL;
int argc;
reading_config_file = 1;
lines = sdssplitlen(config,strlen(config),"\n",1,&totlines);
for (i = 0; i < totlines; i++) {
sds *argv;
int argc;
linenum = i+1;
lines[i] = sdstrim(lines[i]," \t\r\n");
......@@ -477,6 +478,7 @@ void loadServerConfigFromString(char *config) {
/* Skip this line if the resulting command vector is empty. */
if (argc == 0) {
sdsfreesplitres(argv,argc);
argv = NULL;
continue;
}
sdstolower(argv[0]);
......@@ -499,6 +501,7 @@ void loadServerConfigFromString(char *config) {
int new_argc;
new_argv = sdssplitargs(argv[1], &new_argc);
if (!config->interface.set(config, new_argv, new_argc, &err)) {
if(new_argv) sdsfreesplitres(new_argv, new_argc);
goto loaderr;
}
sdsfreesplitres(new_argv, new_argc);
......@@ -510,6 +513,7 @@ void loadServerConfigFromString(char *config) {
}
sdsfreesplitres(argv,argc);
argv = NULL;
continue;
} else {
int match = 0;
......@@ -524,6 +528,7 @@ void loadServerConfigFromString(char *config) {
}
if (match) {
sdsfreesplitres(argv,argc);
argv = NULL;
continue;
}
}
......@@ -590,6 +595,7 @@ void loadServerConfigFromString(char *config) {
err = "Bad directive or wrong number of arguments"; goto loaderr;
}
sdsfreesplitres(argv,argc);
argv = NULL;
}
if (server.logfile[0] != '\0') {
......@@ -627,6 +633,7 @@ void loadServerConfigFromString(char *config) {
return;
loaderr:
if (argv) sdsfreesplitres(argv,argc);
fprintf(stderr, "\n*** FATAL CONFIG FILE ERROR (Redis %s) ***\n",
REDIS_VERSION);
if (i < totlines) {
......@@ -1684,7 +1691,7 @@ int rewriteConfigOverwriteFile(char *configfile, sds content) {
return retval;
}
#ifdef _GNU_SOURCE
#if defined(_GNU_SOURCE) && !defined(__HAIKU__)
fd = mkostemp(tmp_conffile, O_CLOEXEC);
#else
/* There's a theoretical chance here to leak the FD if a module thread forks & execv in the middle */
......@@ -2377,6 +2384,14 @@ static int isValidShutdownOnSigFlags(int val, const char **err) {
return 1;
}
static int isValidAnnouncedNodename(char *val,const char **err) {
if (!(isValidAuxString(val,sdslen(val)))) {
*err = "Announced human node name contained invalid character";
return 0;
}
return 1;
}
static int isValidAnnouncedHostname(char *val, const char **err) {
if (strlen(val) >= NET_HOST_STR_LEN) {
*err = "Hostnames must be less than "
......@@ -2628,6 +2643,12 @@ int updateClusterHostname(const char **err) {
return 1;
}
int updateClusterHumanNodename(const char **err) {
UNUSED(err);
clusterUpdateMyselfHumanNodename();
return 1;
}
static int applyTlsCfg(const char **err) {
UNUSED(err);
......@@ -3072,7 +3093,7 @@ standardConfig static_configs[] = {
createBoolConfig("cluster-allow-replica-migration", NULL, MODIFIABLE_CONFIG, server.cluster_allow_replica_migration, 1, NULL, NULL),
createBoolConfig("replica-announced", NULL, MODIFIABLE_CONFIG, server.replica_announced, 1, NULL, NULL),
createBoolConfig("latency-tracking", NULL, MODIFIABLE_CONFIG, server.latency_tracking_enabled, 1, NULL, NULL),
createBoolConfig("aof-disable-auto-gc", NULL, MODIFIABLE_CONFIG, server.aof_disable_auto_gc, 0, NULL, updateAofAutoGCEnabled),
createBoolConfig("aof-disable-auto-gc", NULL, MODIFIABLE_CONFIG | HIDDEN_CONFIG, server.aof_disable_auto_gc, 0, NULL, updateAofAutoGCEnabled),
createBoolConfig("replica-ignore-disk-write-errors", NULL, MODIFIABLE_CONFIG, server.repl_ignore_disk_write_error, 0, NULL, NULL),
/* String Configs */
......@@ -3084,6 +3105,7 @@ standardConfig static_configs[] = {
createStringConfig("cluster-announce-ip", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, server.cluster_announce_ip, NULL, NULL, updateClusterIp),
createStringConfig("cluster-config-file", NULL, IMMUTABLE_CONFIG, ALLOW_EMPTY_STRING, server.cluster_configfile, "nodes.conf", NULL, NULL),
createStringConfig("cluster-announce-hostname", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, server.cluster_announce_hostname, NULL, isValidAnnouncedHostname, updateClusterHostname),
createStringConfig("cluster-announce-human-nodename", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, server.cluster_announce_human_nodename, NULL, isValidAnnouncedNodename, updateClusterHumanNodename),
createStringConfig("syslog-ident", NULL, IMMUTABLE_CONFIG, ALLOW_EMPTY_STRING, server.syslog_ident, "redis", NULL, NULL),
createStringConfig("dbfilename", NULL, MODIFIABLE_CONFIG | PROTECTED_CONFIG, ALLOW_EMPTY_STRING, server.rdb_filename, "dump.rdb", isValidDBfilename, NULL),
createStringConfig("appendfilename", NULL, IMMUTABLE_CONFIG, ALLOW_EMPTY_STRING, server.aof_filename, "appendonly.aof", isValidAOFfilename, NULL),
......
......@@ -114,14 +114,15 @@ typedef struct ConnectionType {
struct connection {
ConnectionType *type;
ConnectionState state;
int last_errno;
int fd;
short int flags;
short int refs;
int last_errno;
unsigned short int iovcnt;
void *private_data;
ConnectionCallbackFunc conn_handler;
ConnectionCallbackFunc write_handler;
ConnectionCallbackFunc read_handler;
int fd;
};
#define CONFIG_BINDADDR_MAX 16
......@@ -445,4 +446,9 @@ int RedisRegisterConnectionTypeSocket(void);
int RedisRegisterConnectionTypeUnix(void);
int RedisRegisterConnectionTypeTLS(void);
/* Return 1 if connection is using TLS protocol, 0 if otherwise. */
static inline int connIsTLS(connection *conn) {
return conn && conn->type == connectionTypeTls();
}
#endif /* __REDIS_CONNECTION_H */
......@@ -47,6 +47,7 @@
int expireIfNeeded(redisDb *db, robj *key, int flags);
int keyIsExpired(redisDb *db, robj *key);
static void dbSetValue(redisDb *db, robj *key, robj *val, int overwrite, dictEntry *de);
/* Update LFU when an object is accessed.
* Firstly, decrement the counter if the decrement time is reached.
......@@ -187,17 +188,28 @@ robj *lookupKeyWriteOrReply(client *c, robj *key, robj *reply) {
/* Add the key to the DB. It's up to the caller to increment the reference
* counter of the value if needed.
*
* The program is aborted if the key already exists. */
void dbAdd(redisDb *db, robj *key, robj *val) {
sds copy = sdsdup(key->ptr);
dictEntry *de = dictAddRaw(db->dict, copy, NULL);
* If the update_if_existing argument is false, the the program is aborted
* if the key already exists, otherwise, it can fall back to dbOverwite. */
static void dbAddInternal(redisDb *db, robj *key, robj *val, int update_if_existing) {
dictEntry *existing;
dictEntry *de = dictAddRaw(db->dict, key->ptr, &existing);
if (update_if_existing && existing) {
dbSetValue(db, key, val, 1, existing);
return;
}
serverAssertWithInfo(NULL, key, de != NULL);
dictSetKey(db->dict, de, sdsdup(key->ptr));
initObjectLRUOrLFU(val);
dictSetVal(db->dict, de, val);
signalKeyAsReady(db, key, val->type);
if (server.cluster_enabled) slotToKeyAddEntry(de, db);
notifyKeyspaceEvent(NOTIFY_NEW,"new",key,db->id);
}
void dbAdd(redisDb *db, robj *key, robj *val) {
dbAddInternal(db, key, val, 0);
}
/* This is a special version of dbAdd() that is used only when loading
* keys from the RDB file: the key is passed as an SDS string that is
* retained by the function (and not freed by the caller).
......@@ -212,6 +224,7 @@ void dbAdd(redisDb *db, robj *key, robj *val) {
int dbAddRDBLoad(redisDb *db, sds key, robj *val) {
dictEntry *de = dictAddRaw(db->dict, key, NULL);
if (de == NULL) return 0;
initObjectLRUOrLFU(val);
dictSetVal(db->dict, de, val);
if (server.cluster_enabled) slotToKeyAddEntry(de, db);
return 1;
......@@ -226,15 +239,16 @@ int dbAddRDBLoad(redisDb *db, sds key, robj *val) {
* replacement (in which case we need to emit deletion signals), or just an
* update of a value of an existing key (when false).
*
* The dictEntry input is optional, can be used if we already have one.
*
* The program is aborted if the key was not already present. */
static void dbSetValue(redisDb *db, robj *key, robj *val, int overwrite) {
dictEntry *de = dictFind(db->dict,key->ptr);
static void dbSetValue(redisDb *db, robj *key, robj *val, int overwrite, dictEntry *de) {
if (!de) de = dictFind(db->dict,key->ptr);
serverAssertWithInfo(NULL,key,de != NULL);
robj *old = dictGetVal(de);
if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
val->lru = old->lru;
}
val->lru = old->lru;
if (overwrite) {
/* RM_StringDMA may call dbUnshareStringValue which may free val, so we
* need to incr to retain old */
......@@ -262,7 +276,7 @@ static void dbSetValue(redisDb *db, robj *key, robj *val, int overwrite) {
/* Replace an existing key with a new value, we just replace value and don't
* emit any events */
void dbReplaceValue(redisDb *db, robj *key, robj *val) {
dbSetValue(db, key, val, 0);
dbSetValue(db, key, val, 0, NULL);
}
/* High level Set operation. This function can be used in order to set
......@@ -283,13 +297,17 @@ void setKey(client *c, redisDb *db, robj *key, robj *val, int flags) {
if (flags & SETKEY_ALREADY_EXIST)
keyfound = 1;
else if (flags & SETKEY_ADD_OR_UPDATE)
keyfound = -1;
else if (!(flags & SETKEY_DOESNT_EXIST))
keyfound = (lookupKeyWrite(db,key) != NULL);
if (!keyfound) {
dbAdd(db,key,val);
} else if (keyfound<0) {
dbAddInternal(db,key,val,1);
} else {
dbSetValue(db,key,val,1);
dbSetValue(db,key,val,1,NULL);
}
incrRefCount(val);
if (!(flags & SETKEY_KEEPTTL)) removeExpire(db,key);
......@@ -792,29 +810,70 @@ void keysCommand(client *c) {
setDeferredArrayLen(c,replylen,numkeys);
}
/* Data used by the dict scan callback. */
typedef struct {
list *keys; /* elements that collect from dict */
robj *o; /* o must be a hash/set/zset object, NULL means current db */
long long type; /* the particular type when scan the db */
sds pattern; /* pattern string, NULL means no pattern */
long sampled; /* cumulative number of keys sampled */
} scanData;
/* Helper function to compare key type in scan commands */
int objectTypeCompare(robj *o, long long target) {
if (o->type != OBJ_MODULE) {
if (o->type != target)
return 0;
else
return 1;
}
/* module type compare */
long long mt = (long long)REDISMODULE_TYPE_SIGN(((moduleValue *)o->ptr)->type->id);
if (target != -mt)
return 0;
else
return 1;
}
/* This callback is used by scanGenericCommand in order to collect elements
* returned by the dictionary iterator into a list. */
void scanCallback(void *privdata, const dictEntry *de) {
void **pd = (void**) privdata;
list *keys = pd[0];
robj *o = pd[1];
robj *key, *val = NULL;
scanData *data = (scanData *)privdata;
list *keys = data->keys;
robj *o = data->o;
sds val = NULL;
sds key = NULL;
data->sampled++;
/* o and typename can not have values at the same time. */
serverAssert(!((data->type != LLONG_MAX) && o));
/* Filter an element if it isn't the type we want. */
/* TODO: uncomment in redis 8.0
if (!o && data->type != LLONG_MAX) {
robj *rval = dictGetVal(de);
if (!objectTypeCompare(rval, data->type)) return;
}*/
/* Filter element if it does not match the pattern. */
sds keysds = dictGetKey(de);
if (data->pattern) {
if (!stringmatchlen(data->pattern, sdslen(data->pattern), keysds, sdslen(keysds), 0)) {
return;
}
}
if (o == NULL) {
sds sdskey = dictGetKey(de);
key = createStringObject(sdskey, sdslen(sdskey));
key = keysds;
} else if (o->type == OBJ_SET) {
sds keysds = dictGetKey(de);
key = createStringObject(keysds,sdslen(keysds));
key = keysds;
} else if (o->type == OBJ_HASH) {
sds sdskey = dictGetKey(de);
sds sdsval = dictGetVal(de);
key = createStringObject(sdskey,sdslen(sdskey));
val = createStringObject(sdsval,sdslen(sdsval));
key = keysds;
val = dictGetVal(de);
} else if (o->type == OBJ_ZSET) {
sds sdskey = dictGetKey(de);
key = createStringObject(sdskey,sdslen(sdskey));
val = createStringObjectFromLongDouble(*(double*)dictGetVal(de),0);
char buf[MAX_LONG_DOUBLE_CHARS];
int len = ld2string(buf, sizeof(buf), *(double *)dictGetVal(de), LD_STR_AUTO);
key = sdsdup(keysds);
val = sdsnewlen(buf, len);
} else {
serverPanic("Type not handled in SCAN callback.");
}
......@@ -842,6 +901,46 @@ int parseScanCursorOrReply(client *c, robj *o, unsigned long *cursor) {
return C_OK;
}
char *obj_type_name[OBJ_TYPE_MAX] = {
"string",
"list",
"set",
"zset",
"hash",
NULL, /* module type is special */
"stream"
};
/* Helper function to get type from a string in scan commands */
long long getObjectTypeByName(char *name) {
for (long long i = 0; i < OBJ_TYPE_MAX; i++) {
if (obj_type_name[i] && !strcasecmp(name, obj_type_name[i])) {
return i;
}
}
moduleType *mt = moduleTypeLookupModuleByNameIgnoreCase(name);
if (mt != NULL) return -(REDISMODULE_TYPE_SIGN(mt->id));
return LLONG_MAX;
}
char *getObjectTypeName(robj *o) {
if (o == NULL) {
return "none";
}
serverAssert(o->type >= 0 && o->type < OBJ_TYPE_MAX);
if (o->type == OBJ_MODULE) {
moduleValue *mv = o->ptr;
return mv->type->name;
} else {
return obj_type_name[o->type];
}
}
/* This command implements SCAN, HSCAN and SSCAN commands.
* If object 'o' is passed, then it must be a Hash, Set or Zset object, otherwise
* if 'o' is NULL the command will operate on the dictionary associated with
......@@ -855,11 +954,11 @@ int parseScanCursorOrReply(client *c, robj *o, unsigned long *cursor) {
* of every element on the Hash. */
void scanGenericCommand(client *c, robj *o, unsigned long cursor) {
int i, j;
list *keys = listCreate();
listNode *node, *nextnode;
listNode *node;
long count = 10;
sds pat = NULL;
sds typename = NULL;
long long type = LLONG_MAX;
int patlen = 0, use_pattern = 0;
dict *ht;
......@@ -878,12 +977,12 @@ void scanGenericCommand(client *c, robj *o, unsigned long cursor) {
if (getLongFromObjectOrReply(c, c->argv[i+1], &count, NULL)
!= C_OK)
{
goto cleanup;
return;
}
if (count < 1) {
addReplyErrorObject(c,shared.syntaxerr);
goto cleanup;
return;
}
i += 2;
......@@ -899,10 +998,16 @@ void scanGenericCommand(client *c, robj *o, unsigned long cursor) {
} else if (!strcasecmp(c->argv[i]->ptr, "type") && o == NULL && j >= 2) {
/* SCAN for a particular type only applies to the db dict */
typename = c->argv[i+1]->ptr;
type = getObjectTypeByName(typename);
if (type == LLONG_MAX) {
/* TODO: uncomment in redis 8.0
addReplyErrorFormat(c, "unknown type name '%s'", typename);
return; */
}
i+= 2;
} else {
addReplyErrorObject(c,shared.syntaxerr);
goto cleanup;
return;
}
}
......@@ -922,42 +1027,67 @@ void scanGenericCommand(client *c, robj *o, unsigned long cursor) {
ht = o->ptr;
} else if (o->type == OBJ_HASH && o->encoding == OBJ_ENCODING_HT) {
ht = o->ptr;
count *= 2; /* We return key / value for this type. */
} else if (o->type == OBJ_ZSET && o->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = o->ptr;
ht = zs->dict;
count *= 2; /* We return key / value for this type. */
}
list *keys = listCreate();
/* Set a free callback for the contents of the collected keys list.
* For the main keyspace dict, and when we scan a key that's dict encoded
* (we have 'ht'), we don't need to define free method because the strings
* in the list are just a shallow copy from the pointer in the dictEntry.
* When scanning a key with other encodings (e.g. listpack), we need to
* free the temporary strings we add to that list.
* The exception to the above is ZSET, where we do allocate temporary
* strings even when scanning a dict. */
if (o && (!ht || o->type == OBJ_ZSET)) {
listSetFreeMethod(keys, (void (*)(void*))sdsfree);
}
if (ht) {
void *privdata[2];
/* We set the max number of iterations to ten times the specified
* COUNT, so if the hash table is in a pathological state (very
* sparsely populated) we avoid to block too much time at the cost
* of returning no or very few elements. */
long maxiterations = count*10;
/* We pass two pointers to the callback: the list to which it will
* add new elements, and the object containing the dictionary so that
* it is possible to fetch more data in a type-dependent way. */
privdata[0] = keys;
privdata[1] = o;
/* We pass scanData which have three pointers to the callback:
* 1. data.keys: the list to which it will add new elements;
* 2. data.o: the object containing the dictionary so that
* it is possible to fetch more data in a type-dependent way;
* 3. data.type: the specified type scan in the db, LLONG_MAX means
* type matching is no needed;
* 4. data.pattern: the pattern string
* 5. data.sampled: the maxiteration limit is there in case we're
* working on an empty dict, one with a lot of empty buckets, and
* for the buckets are not empty, we need to limit the spampled number
* to prevent a long hang time caused by filtering too many keys*/
scanData data = {
.keys = keys,
.o = o,
.type = type,
.pattern = use_pattern ? pat : NULL,
.sampled = 0,
};
do {
cursor = dictScan(ht, cursor, scanCallback, privdata);
} while (cursor &&
maxiterations-- &&
listLength(keys) < (unsigned long)count);
cursor = dictScan(ht, cursor, scanCallback, &data);
} while (cursor && maxiterations-- && data.sampled < count);
} else if (o->type == OBJ_SET) {
char *str;
char buf[LONG_STR_SIZE];
size_t len;
int64_t llele;
setTypeIterator *si = setTypeInitIterator(o);
while (setTypeNext(si, &str, &len, &llele) != -1) {
if (str == NULL) {
listAddNodeTail(keys, createStringObjectFromLongLong(llele));
} else {
listAddNodeTail(keys, createStringObject(str, len));
len = ll2string(buf, sizeof(buf), llele);
}
char *key = str ? str : buf;
if (use_pattern && !stringmatchlen(pat, sdslen(pat), key, len, 0)) {
continue;
}
listAddNodeTail(keys, sdsnewlen(key, len));
}
setTypeReleaseIterator(si);
cursor = 0;
......@@ -965,72 +1095,53 @@ void scanGenericCommand(client *c, robj *o, unsigned long cursor) {
o->encoding == OBJ_ENCODING_LISTPACK)
{
unsigned char *p = lpFirst(o->ptr);
unsigned char *vstr;
int64_t vlen;
unsigned char *str;
int64_t len;
unsigned char intbuf[LP_INTBUF_SIZE];
while(p) {
vstr = lpGet(p,&vlen,intbuf);
listAddNodeTail(keys, createStringObject((char*)vstr,vlen));
p = lpNext(o->ptr,p);
str = lpGet(p, &len, intbuf);
/* point to the value */
p = lpNext(o->ptr, p);
if (use_pattern && !stringmatchlen(pat, sdslen(pat), (char *)str, len, 0)) {
/* jump to the next key/val pair */
p = lpNext(o->ptr, p);
continue;
}
/* add key object */
listAddNodeTail(keys, sdsnewlen(str, len));
/* add value object */
str = lpGet(p, &len, intbuf);
listAddNodeTail(keys, sdsnewlen(str, len));
p = lpNext(o->ptr, p);
}
cursor = 0;
} else {
serverPanic("Not handled encoding in SCAN.");
}
/* Step 3: Filter elements. */
node = listFirst(keys);
while (node) {
robj *kobj = listNodeValue(node);
nextnode = listNextNode(node);
int filter = 0;
/* Filter element if it does not match the pattern. */
if (use_pattern) {
if (sdsEncodedObject(kobj)) {
if (!stringmatchlen(pat, patlen, kobj->ptr, sdslen(kobj->ptr), 0))
filter = 1;
} else {
char buf[LONG_STR_SIZE];
int len;
serverAssert(kobj->encoding == OBJ_ENCODING_INT);
len = ll2string(buf,sizeof(buf),(long)kobj->ptr);
if (!stringmatchlen(pat, patlen, buf, len, 0)) filter = 1;
/* Step 3: Filter the expired keys */
if (o == NULL && listLength(keys)) {
robj kobj;
listIter li;
listNode *ln;
listRewind(keys, &li);
while ((ln = listNext(&li))) {
sds key = listNodeValue(ln);
initStaticStringObject(kobj, key);
/* Filter an element if it isn't the type we want. */
/* TODO: remove this in redis 8.0 */
if (typename) {
robj* typecheck = lookupKeyReadWithFlags(c->db, &kobj, LOOKUP_NOTOUCH|LOOKUP_NONOTIFY);
if (!typecheck || !objectTypeCompare(typecheck, type)) {
listDelNode(keys, ln);
}
continue;
}
}
/* Filter an element if it isn't the type we want. */
if (!filter && o == NULL && typename){
robj* typecheck = lookupKeyReadWithFlags(c->db, kobj, LOOKUP_NOTOUCH);
char* type = getObjectTypeName(typecheck);
if (strcasecmp((char*) typename, type)) filter = 1;
}
/* Filter element if it is an expired key. */
if (!filter && o == NULL && expireIfNeeded(c->db, kobj, 0)) filter = 1;
/* Remove the element and its associated value if needed. */
if (filter) {
decrRefCount(kobj);
listDelNode(keys, node);
}
/* If this is a hash or a sorted set, we have a flat list of
* key-value elements, so if this element was filtered, remove the
* value, or skip it if it was not filtered: we only match keys. */
if (o && (o->type == OBJ_ZSET || o->type == OBJ_HASH)) {
node = nextnode;
serverAssert(node); /* assertion for valgrind (avoid NPD) */
nextnode = listNextNode(node);
if (filter) {
kobj = listNodeValue(node);
decrRefCount(kobj);
listDelNode(keys, node);
if (expireIfNeeded(c->db, &kobj, 0)) {
listDelNode(keys, ln);
}
}
node = nextnode;
}
/* Step 4: Reply to the client. */
......@@ -1039,14 +1150,11 @@ void scanGenericCommand(client *c, robj *o, unsigned long cursor) {
addReplyArrayLen(c, listLength(keys));
while ((node = listFirst(keys)) != NULL) {
robj *kobj = listNodeValue(node);
addReplyBulk(c, kobj);
decrRefCount(kobj);
sds key = listNodeValue(node);
addReplyBulkCBuffer(c, key, sdslen(key));
listDelNode(keys, node);
}
cleanup:
listSetFreeMethod(keys,decrRefCountVoid);
listRelease(keys);
}
......@@ -1065,28 +1173,6 @@ void lastsaveCommand(client *c) {
addReplyLongLong(c,server.lastsave);
}
char* getObjectTypeName(robj *o) {
char* type;
if (o == NULL) {
type = "none";
} else {
switch(o->type) {
case OBJ_STRING: type = "string"; break;
case OBJ_LIST: type = "list"; break;
case OBJ_SET: type = "set"; break;
case OBJ_ZSET: type = "zset"; break;
case OBJ_HASH: type = "hash"; break;
case OBJ_STREAM: type = "stream"; break;
case OBJ_MODULE: {
moduleValue *mv = o->ptr;
type = mv->type->name;
}; break;
default: type = "unknown"; break;
}
}
return type;
}
void typeCommand(client *c) {
robj *o;
o = lookupKeyReadWithFlags(c->db,c->argv[1],LOOKUP_NOTOUCH);
......@@ -1557,9 +1643,6 @@ void swapdbCommand(client *c) {
*----------------------------------------------------------------------------*/
int removeExpire(redisDb *db, robj *key) {
/* An expire may only be removed if there is a corresponding entry in the
* main dict. Otherwise, the key will never be freed. */
serverAssertWithInfo(NULL,key,dictFind(db->dict,key->ptr) != NULL);
return dictDelete(db->expires,key->ptr) == DICT_OK;
}
......@@ -1590,9 +1673,6 @@ long long getExpire(redisDb *db, robj *key) {
if (dictSize(db->expires) == 0 ||
(de = dictFind(db->expires,key->ptr)) == NULL) return -1;
/* The entry was found in the expire dict, this means it should also
* be present in the main dict (safety check). */
serverAssertWithInfo(NULL,key,dictFind(db->dict,key->ptr) != NULL);
return dictGetSignedIntegerVal(de);
}
......@@ -1724,8 +1804,16 @@ int expireIfNeeded(redisDb *db, robj *key, int flags) {
* will have failed over and the new primary will send us the expire. */
if (isPausedActionsWithUpdate(PAUSE_ACTION_EXPIRE)) return 1;
/* The key needs to be converted from static to heap before deleted */
int static_key = key->refcount == OBJ_STATIC_REFCOUNT;
if (static_key) {
key = createStringObject(key->ptr, sdslen(key->ptr));
}
/* Delete the key */
deleteExpiredKeyAndPropagate(db,key);
if (static_key) {
decrRefCount(key);
}
return 1;
}
......@@ -1783,8 +1871,9 @@ int64_t getAllKeySpecsFlags(struct redisCommand *cmd, int inv) {
* found in other valid keyspecs.
*/
int getKeysUsingKeySpecs(struct redisCommand *cmd, robj **argv, int argc, int search_flags, getKeysResult *result) {
int j, i, k = 0, last, first, step;
int j, i, last, first, step;
keyReference *keys;
serverAssert(result->numkeys == 0); /* caller should initialize or reset it */
for (j = 0; j < cmd->key_specs_num; j++) {
keySpec *spec = cmd->key_specs + j;
......@@ -1849,7 +1938,7 @@ int getKeysUsingKeySpecs(struct redisCommand *cmd, robj **argv, int argc, int se
}
int count = ((last - first)+1);
keys = getKeysPrepareResult(result, count);
keys = getKeysPrepareResult(result, result->numkeys + count);
/* First or last is out of bounds, which indicates a syntax error */
if (last >= argc || last < first || first >= argc) {
......@@ -1870,8 +1959,9 @@ int getKeysUsingKeySpecs(struct redisCommand *cmd, robj **argv, int argc, int se
serverPanic("Redis built-in command declared keys positions not matching the arity requirements.");
}
}
keys[k].pos = i;
keys[k++].flags = spec->flags;
keys[result->numkeys].pos = i;
keys[result->numkeys].flags = spec->flags;
result->numkeys++;
}
/* Handle incomplete specs (only after we added the current spec
......@@ -1892,8 +1982,7 @@ invalid_spec:
}
}
result->numkeys = k;
return k;
return result->numkeys;
}
/* Return all the arguments that are keys in the command passed via argc / argv.
......@@ -2110,7 +2199,7 @@ void getKeysFreeResult(getKeysResult *result) {
* 'firstKeyOfs': firstkey index.
* 'keyStep': the interval of each key, usually this value is 1.
*
* The commands using this functoin have a fully defined keyspec, so returning flags isn't needed. */
* The commands using this function have a fully defined keyspec, so returning flags isn't needed. */
int genericGetKeys(int storeKeyOfs, int keyCountOfs, int firstKeyOfs, int keyStep,
robj **argv, int argc, getKeysResult *result) {
int i, num;
......
......@@ -413,9 +413,9 @@ void debugCommand(client *c) {
" Create a memory leak of the input string.",
"LOG <message>",
" Write <message> to the server log.",
"HTSTATS <dbid>",
"HTSTATS <dbid> [full]",
" Return hash table statistics of the specified Redis database.",
"HTSTATS-KEY <key>",
"HTSTATS-KEY <key> [full]",
" Like HTSTATS but for the hash table stored at <key>'s value.",
"LOADAOF",
" Flush the AOF buffers on disk and reload the AOF in memory.",
......@@ -713,7 +713,10 @@ NULL
if (getPositiveLongFromObjectOrReply(c, c->argv[2], &keys, NULL) != C_OK)
return;
dictExpand(c->db->dict,keys);
if (dictTryExpand(c->db->dict, keys) != DICT_OK) {
addReplyError(c, "OOM in dictTryExpand");
return;
}
long valsize = 0;
if ( c->argc == 5 && getPositiveLongFromObjectOrReply(c, c->argv[4], &valsize, NULL) != C_OK )
return;
......@@ -883,10 +886,11 @@ NULL
sizes = sdscatprintf(sizes,"sdshdr32:%d ",(int)sizeof(struct sdshdr32));
sizes = sdscatprintf(sizes,"sdshdr64:%d ",(int)sizeof(struct sdshdr64));
addReplyBulkSds(c,sizes);
} else if (!strcasecmp(c->argv[1]->ptr,"htstats") && c->argc == 3) {
} else if (!strcasecmp(c->argv[1]->ptr,"htstats") && c->argc >= 3) {
long dbid;
sds stats = sdsempty();
char buf[4096];
int full = 0;
if (getLongFromObjectOrReply(c, c->argv[2], &dbid, NULL) != C_OK) {
sdsfree(stats);
......@@ -897,20 +901,26 @@ NULL
addReplyError(c,"Out of range database");
return;
}
if (c->argc >= 4 && !strcasecmp(c->argv[3]->ptr,"full"))
full = 1;
stats = sdscatprintf(stats,"[Dictionary HT]\n");
dictGetStats(buf,sizeof(buf),server.db[dbid].dict);
dictGetStats(buf,sizeof(buf),server.db[dbid].dict,full);
stats = sdscat(stats,buf);
stats = sdscatprintf(stats,"[Expires HT]\n");
dictGetStats(buf,sizeof(buf),server.db[dbid].expires);
dictGetStats(buf,sizeof(buf),server.db[dbid].expires,full);
stats = sdscat(stats,buf);
addReplyVerbatim(c,stats,sdslen(stats),"txt");
sdsfree(stats);
} else if (!strcasecmp(c->argv[1]->ptr,"htstats-key") && c->argc == 3) {
} else if (!strcasecmp(c->argv[1]->ptr,"htstats-key") && c->argc >= 3) {
robj *o;
dict *ht = NULL;
int full = 0;
if (c->argc >= 4 && !strcasecmp(c->argv[3]->ptr,"full"))
full = 1;
if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.nokeyerr))
== NULL) return;
......@@ -933,7 +943,7 @@ NULL
"represented using an hash table");
} else {
char buf[4096];
dictGetStats(buf,sizeof(buf),ht);
dictGetStats(buf,sizeof(buf),ht,full);
addReplyVerbatim(c,buf,strlen(buf),"txt");
}
} else if (!strcasecmp(c->argv[1]->ptr,"change-repl-id") && c->argc == 2) {
......@@ -1211,6 +1221,8 @@ static void* getAndSetMcontextEip(ucontext_t *uc, void *eip) {
GET_SET_RETURN(uc->uc_mcontext.gregs[16], eip);
#elif defined(__ia64__) /* Linux IA64 */
GET_SET_RETURN(uc->uc_mcontext.sc_ip, eip);
#elif defined(__riscv) /* Linux RISC-V */
GET_SET_RETURN(uc->uc_mcontext.__gregs[REG_PC], eip);
#elif defined(__arm__) /* Linux ARM */
GET_SET_RETURN(uc->uc_mcontext.arm_pc, eip);
#elif defined(__aarch64__) /* Linux AArch64 */
......@@ -1445,6 +1457,49 @@ void logRegisters(ucontext_t *uc) {
(unsigned long) uc->uc_mcontext.gregs[18]
);
logStackContent((void**)uc->uc_mcontext.gregs[15]);
#elif defined(__riscv) /* Linux RISC-V */
serverLog(LL_WARNING,
"\n"
"ra:%016lx gp:%016lx\ntp:%016lx t0:%016lx\n"
"t1:%016lx t2:%016lx\ns0:%016lx s1:%016lx\n"
"a0:%016lx a1:%016lx\na2:%016lx a3:%016lx\n"
"a4:%016lx a5:%016lx\na6:%016lx a7:%016lx\n"
"s2:%016lx s3:%016lx\ns4:%016lx s5:%016lx\n"
"s6:%016lx s7:%016lx\ns8:%016lx s9:%016lx\n"
"s10:%016lx s11:%016lx\nt3:%016lx t4:%016lx\n"
"t5:%016lx t6:%016lx\n",
(unsigned long) uc->uc_mcontext.__gregs[1],
(unsigned long) uc->uc_mcontext.__gregs[3],
(unsigned long) uc->uc_mcontext.__gregs[4],
(unsigned long) uc->uc_mcontext.__gregs[5],
(unsigned long) uc->uc_mcontext.__gregs[6],
(unsigned long) uc->uc_mcontext.__gregs[7],
(unsigned long) uc->uc_mcontext.__gregs[8],
(unsigned long) uc->uc_mcontext.__gregs[9],
(unsigned long) uc->uc_mcontext.__gregs[10],
(unsigned long) uc->uc_mcontext.__gregs[11],
(unsigned long) uc->uc_mcontext.__gregs[12],
(unsigned long) uc->uc_mcontext.__gregs[13],
(unsigned long) uc->uc_mcontext.__gregs[14],
(unsigned long) uc->uc_mcontext.__gregs[15],
(unsigned long) uc->uc_mcontext.__gregs[16],
(unsigned long) uc->uc_mcontext.__gregs[17],
(unsigned long) uc->uc_mcontext.__gregs[18],
(unsigned long) uc->uc_mcontext.__gregs[19],
(unsigned long) uc->uc_mcontext.__gregs[20],
(unsigned long) uc->uc_mcontext.__gregs[21],
(unsigned long) uc->uc_mcontext.__gregs[22],
(unsigned long) uc->uc_mcontext.__gregs[23],
(unsigned long) uc->uc_mcontext.__gregs[24],
(unsigned long) uc->uc_mcontext.__gregs[25],
(unsigned long) uc->uc_mcontext.__gregs[26],
(unsigned long) uc->uc_mcontext.__gregs[27],
(unsigned long) uc->uc_mcontext.__gregs[28],
(unsigned long) uc->uc_mcontext.__gregs[29],
(unsigned long) uc->uc_mcontext.__gregs[30],
(unsigned long) uc->uc_mcontext.__gregs[31]
);
logStackContent((void**)uc->uc_mcontext.__gregs[REG_SP]);
#elif defined(__aarch64__) /* Linux AArch64 */
serverLog(LL_WARNING,
"\n"
......@@ -1803,7 +1858,7 @@ sds genClusterDebugString(sds infostring) {
infostring = sdscatprintf(infostring, "\r\n# Cluster info\r\n");
infostring = sdscatsds(infostring, genClusterInfoString());
infostring = sdscatprintf(infostring, "\n------ CLUSTER NODES OUTPUT ------\n");
infostring = sdscatsds(infostring, clusterGenNodesDescription(0, 0));
infostring = sdscatsds(infostring, clusterGenNodesDescription(NULL, 0, 0));
return infostring;
}
......
......@@ -294,9 +294,12 @@ int dictTryExpand(dict *d, unsigned long size) {
* work it does would be unbound and the function may block for a long time. */
int dictRehash(dict *d, int n) {
int empty_visits = n*10; /* Max number of empty buckets to visit. */
unsigned long s0 = DICTHT_SIZE(d->ht_size_exp[0]);
unsigned long s1 = DICTHT_SIZE(d->ht_size_exp[1]);
if (dict_can_resize == DICT_RESIZE_FORBID || !dictIsRehashing(d)) return 0;
if (dict_can_resize == DICT_RESIZE_AVOID &&
(DICTHT_SIZE(d->ht_size_exp[1]) / DICTHT_SIZE(d->ht_size_exp[0]) < dict_force_resize_ratio))
((s1 > s0 && s1 / s0 < dict_force_resize_ratio) ||
(s1 < s0 && s0 / s1 < dict_force_resize_ratio)))
{
return 0;
}
......@@ -1095,19 +1098,30 @@ unsigned int dictGetSomeKeys(dict *d, dictEntry **des, unsigned int count) {
} else {
emptylen = 0;
while (he) {
/* Collect all the elements of the buckets found non
* empty while iterating. */
*des = he;
des++;
/* Collect all the elements of the buckets found non empty while iterating.
* To avoid the issue of being unable to sample the end of a long chain,
* we utilize the Reservoir Sampling algorithm to optimize the sampling process.
* This means that even when the maximum number of samples has been reached,
* we continue sampling until we reach the end of the chain.
* See https://en.wikipedia.org/wiki/Reservoir_sampling. */
if (stored < count) {
des[stored] = he;
} else {
unsigned long r = randomULong() % (stored + 1);
if (r < count) des[r] = he;
}
he = dictGetNext(he);
stored++;
if (stored == count) return stored;
}
if (stored >= count) goto end;
}
}
i = (i+1) & maxsizemask;
}
return stored;
end:
return stored > count ? count : stored;
}
......@@ -1502,7 +1516,7 @@ dictEntry *dictFindEntryByPtrAndHash(dict *d, const void *oldptr, uint64_t hash)
/* ------------------------------- Debugging ---------------------------------*/
#define DICT_STATS_VECTLEN 50
size_t _dictGetStatsHt(char *buf, size_t bufsize, dict *d, int htidx) {
size_t _dictGetStatsHt(char *buf, size_t bufsize, dict *d, int htidx, int full) {
unsigned long i, slots = 0, chainlen, maxchainlen = 0;
unsigned long totchainlen = 0;
unsigned long clvector[DICT_STATS_VECTLEN];
......@@ -1510,7 +1524,23 @@ size_t _dictGetStatsHt(char *buf, size_t bufsize, dict *d, int htidx) {
if (d->ht_used[htidx] == 0) {
return snprintf(buf,bufsize,
"No stats available for empty dictionaries\n");
"Hash table %d stats (%s):\n"
"No stats available for empty dictionaries\n",
htidx, (htidx == 0) ? "main hash table" : "rehashing target");
}
if (!full) {
l += snprintf(buf+l,bufsize-l,
"Hash table %d stats (%s):\n"
" table size: %lu\n"
" number of elements: %lu\n",
htidx, (htidx == 0) ? "main hash table" : "rehashing target",
DICTHT_SIZE(d->ht_size_exp[htidx]), d->ht_used[htidx]);
/* Make sure there is a NULL term at the end. */
buf[bufsize-1] = '\0';
/* Unlike snprintf(), return the number of characters actually written. */
return strlen(buf);
}
/* Compute stats. */
......@@ -1557,24 +1587,25 @@ size_t _dictGetStatsHt(char *buf, size_t bufsize, dict *d, int htidx) {
i, clvector[i], ((float)clvector[i]/DICTHT_SIZE(d->ht_size_exp[htidx]))*100);
}
/* Make sure there is a NULL term at the end. */
buf[bufsize-1] = '\0';
/* Unlike snprintf(), return the number of characters actually written. */
if (bufsize) buf[bufsize-1] = '\0';
return strlen(buf);
}
void dictGetStats(char *buf, size_t bufsize, dict *d) {
void dictGetStats(char *buf, size_t bufsize, dict *d, int full) {
size_t l;
char *orig_buf = buf;
size_t orig_bufsize = bufsize;
l = _dictGetStatsHt(buf,bufsize,d,0);
buf += l;
bufsize -= l;
if (dictIsRehashing(d) && bufsize > 0) {
_dictGetStatsHt(buf,bufsize,d,1);
l = _dictGetStatsHt(buf,bufsize,d,0,full);
if (dictIsRehashing(d) && bufsize > l) {
buf += l;
bufsize -= l;
_dictGetStatsHt(buf,bufsize,d,1,full);
}
/* Make sure there is a NULL term at the end. */
if (orig_bufsize) orig_buf[orig_bufsize-1] = '\0';
orig_buf[orig_bufsize-1] = '\0';
}
/* ------------------------------- Benchmark ---------------------------------*/
......
......@@ -210,7 +210,7 @@ void dictReleaseIterator(dictIterator *iter);
dictEntry *dictGetRandomKey(dict *d);
dictEntry *dictGetFairRandomKey(dict *d);
unsigned int dictGetSomeKeys(dict *d, dictEntry **des, unsigned int count);
void dictGetStats(char *buf, size_t bufsize, dict *d);
void dictGetStats(char *buf, size_t bufsize, dict *d, int full);
uint64_t dictGenHashFunction(const void *key, size_t len);
uint64_t dictGenCaseHashFunction(const unsigned char *buf, size_t len);
void dictEmpty(dict *d, void(callback)(dict*));
......
......@@ -80,7 +80,7 @@ unsigned int getLRUClock(void) {
unsigned int LRU_CLOCK(void) {
unsigned int lruclock;
if (1000/server.hz <= LRU_CLOCK_RESOLUTION) {
atomicGet(server.lruclock,lruclock);
lruclock = server.lruclock;
} else {
lruclock = getLRUClock();
}
......
......@@ -490,7 +490,7 @@ void geoaddCommand(client *c) {
GeoHashBits hash;
geohashEncodeWGS84(xy[0], xy[1], GEO_STEP_MAX, &hash);
GeoHashFix52Bits bits = geohashAlign52Bits(hash);
robj *score = createObject(OBJ_STRING, sdsfromlonglong(bits));
robj *score = createStringObjectFromLongLongWithSds(bits);
robj *val = c->argv[longidx + i * 3 + 2];
argv[longidx+i*2] = score;
argv[longidx+1+i*2] = val;
......
......@@ -356,6 +356,7 @@ typedef struct RedisModuleCommandFilterCtx {
RedisModuleString **argv;
int argv_len;
int argc;
client *c;
} RedisModuleCommandFilterCtx;
 
typedef void (*RedisModuleCommandFilterFunc) (RedisModuleCommandFilterCtx *filter);
......@@ -620,7 +621,7 @@ void *RM_PoolAlloc(RedisModuleCtx *ctx, size_t bytes) {
* Helpers for modules API implementation
* -------------------------------------------------------------------------- */
 
client *moduleAllocTempClient(user *user) {
client *moduleAllocTempClient(void) {
client *c = NULL;
 
if (moduleTempClientCount > 0) {
......@@ -630,10 +631,8 @@ client *moduleAllocTempClient(user *user) {
} else {
c = createClient(NULL);
c->flags |= CLIENT_MODULE;
c->user = NULL; /* Root user */
}
c->user = user;
return c;
}
 
......@@ -858,7 +857,7 @@ void moduleCallCommandUnblockedHandler(client *c) {
moduleCreateContext(&ctx, module, REDISMODULE_CTX_TEMP_CLIENT);
selectDb(ctx.client, c->db->id);
 
CallReply *reply = moduleParseReply(c, &ctx);
CallReply *reply = moduleParseReply(c, NULL);
module->in_call++;
promise->on_unblocked(&ctx, reply, promise->private_data);
module->in_call--;
......@@ -879,7 +878,7 @@ void moduleCreateContext(RedisModuleCtx *out_ctx, RedisModule *module, int ctx_f
out_ctx->module = module;
out_ctx->flags = ctx_flags;
if (ctx_flags & REDISMODULE_CTX_TEMP_CLIENT)
out_ctx->client = moduleAllocTempClient(NULL);
out_ctx->client = moduleAllocTempClient();
else if (ctx_flags & REDISMODULE_CTX_NEW_CLIENT)
out_ctx->client = createClient(NULL);
 
......@@ -2999,15 +2998,15 @@ int RM_ReplyWithError(RedisModuleCtx *ctx, const char *err) {
 
/* Reply with the error create from a printf format and arguments.
*
* If the error code is already passed in the string 'fmt', the error
* code provided is used, otherwise the string "-ERR " for the generic
* error code is automatically added.
* Note that 'fmt' must contain all the error, including
* the initial error code. The function only provides the initial "-", so
* the usage is, for example:
*
* The usage is, for example:
* RedisModule_ReplyWithErrorFormat(ctx,"ERR Wrong Type: %s",type);
*
* RedisModule_ReplyWithErrorFormat(ctx, "An error: %s", "foo");
* and not just:
*
* RedisModule_ReplyWithErrorFormat(ctx, "-WRONGTYPE Wrong Type: %s", "foo");
* RedisModule_ReplyWithErrorFormat(ctx,"Wrong Type: %s",type);
*
* The function always returns REDISMODULE_OK.
*/
......@@ -3015,11 +3014,17 @@ int RM_ReplyWithErrorFormat(RedisModuleCtx *ctx, const char *fmt, ...) {
client *c = moduleGetReplyClient(ctx);
if (c == NULL) return REDISMODULE_OK;
 
int len = strlen(fmt) + 2; /* 1 for the \0 and 1 for the hyphen */
char *hyphenfmt = zmalloc(len);
snprintf(hyphenfmt, len, "-%s", fmt);
va_list ap;
va_start(ap, fmt);
addReplyErrorFormatInternal(c, 0, fmt, ap);
addReplyErrorFormatInternal(c, 0, hyphenfmt, ap);
va_end(ap);
 
zfree(hyphenfmt);
return REDISMODULE_OK;
}
 
......@@ -6038,7 +6043,7 @@ robj **moduleCreateArgvFromUserFormat(const char *cmdname, const char *fmt, int
argv[argc++] = createStringObject(buf,len);
} else if (*p == 'l') {
long long ll = va_arg(ap,long long);
argv[argc++] = createObject(OBJ_STRING,sdsfromlonglong(ll));
argv[argc++] = createStringObjectFromLongLongWithSds(ll);
} else if (*p == 'v') {
/* A vector of strings */
robj **v = va_arg(ap, void*);
......@@ -6215,20 +6220,7 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
error_as_call_replies = flags & REDISMODULE_ARGV_CALL_REPLIES_AS_ERRORS;
va_end(ap);
 
user *user = NULL;
if (flags & REDISMODULE_ARGV_RUN_AS_USER) {
user = ctx->user ? ctx->user->user : ctx->client->user;
if (!user) {
errno = ENOTSUP;
if (error_as_call_replies) {
sds msg = sdsnew("cannot run as user, no user directly attached to context or context's client");
reply = callReplyCreateError(msg, ctx);
}
return reply;
}
}
c = moduleAllocTempClient(user);
c = moduleAllocTempClient();
 
if (!(flags & REDISMODULE_ARGV_ALLOW_BLOCK)) {
/* We do not want to allow block, the module do not expect it */
......@@ -6248,6 +6240,20 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
}
if (ctx->module) ctx->module->in_call++;
 
user *user = NULL;
if (flags & REDISMODULE_ARGV_RUN_AS_USER) {
user = ctx->user ? ctx->user->user : ctx->client->user;
if (!user) {
errno = ENOTSUP;
if (error_as_call_replies) {
sds msg = sdsnew("cannot run as user, no user directly attached to context or context's client");
reply = callReplyCreateError(msg, ctx);
}
goto cleanup;
}
c->user = user;
}
/* We handle the above format error only when the client is setup so that
* we can free it normally. */
if (argv == NULL) {
......@@ -6567,7 +6573,7 @@ uint64_t moduleTypeEncodeId(const char *name, int encver) {
/* Search, in the list of exported data types of all the modules registered,
* a type with the same name as the one given. Returns the moduleType
* structure pointer if such a module is found, or NULL otherwise. */
moduleType *moduleTypeLookupModuleByName(const char *name) {
moduleType *moduleTypeLookupModuleByNameInternal(const char *name, int ignore_case) {
dictIterator *di = dictGetIterator(modules);
dictEntry *de;
 
......@@ -6579,7 +6585,9 @@ moduleType *moduleTypeLookupModuleByName(const char *name) {
listRewind(module->types,&li);
while((ln = listNext(&li))) {
moduleType *mt = ln->value;
if (memcmp(name,mt->name,sizeof(mt->name)) == 0) {
if ((!ignore_case && memcmp(name,mt->name,sizeof(mt->name)) == 0)
|| (ignore_case && !strcasecmp(name, mt->name)))
{
dictReleaseIterator(di);
return mt;
}
......@@ -6588,6 +6596,15 @@ moduleType *moduleTypeLookupModuleByName(const char *name) {
dictReleaseIterator(di);
return NULL;
}
/* Search all registered modules by name, and name is case sensitive */
moduleType *moduleTypeLookupModuleByName(const char *name) {
return moduleTypeLookupModuleByNameInternal(name, 0);
}
/* Search all registered modules by name, but case insensitive */
moduleType *moduleTypeLookupModuleByNameIgnoreCase(const char *name) {
return moduleTypeLookupModuleByNameInternal(name, 1);
}
 
/* Lookup a module by ID, with caching. This function is used during RDB
* loading. Modules exporting data types should never be able to unload, so
......@@ -7671,7 +7688,6 @@ RedisModuleBlockedClient *moduleBlockClient(RedisModuleCtx *ctx, RedisModuleCmdF
* commands from Lua or MULTI. We actually create an already aborted
* (client set to NULL) blocked client handle, and actually reply with
* an error. */
mstime_t timeout = timeout_ms ? (mstime()+timeout_ms) : 0;
bc->client = (islua || ismulti) ? NULL : c;
bc->module = ctx->module;
bc->reply_callback = reply_callback;
......@@ -7680,8 +7696,8 @@ RedisModuleBlockedClient *moduleBlockClient(RedisModuleCtx *ctx, RedisModuleCmdF
bc->disconnect_callback = NULL; /* Set by RM_SetDisconnectCallback() */
bc->free_privdata = free_privdata;
bc->privdata = privdata;
bc->reply_client = moduleAllocTempClient(NULL);
bc->thread_safe_ctx_client = moduleAllocTempClient(NULL);
bc->reply_client = moduleAllocTempClient();
bc->thread_safe_ctx_client = moduleAllocTempClient();
if (bc->client)
bc->reply_client->resp = bc->client->resp;
bc->dbid = c->db->id;
......@@ -7689,7 +7705,17 @@ RedisModuleBlockedClient *moduleBlockClient(RedisModuleCtx *ctx, RedisModuleCmdF
bc->unblocked = 0;
bc->background_timer = 0;
bc->background_duration = 0;
c->bstate.timeout = timeout;
c->bstate.timeout = 0;
if (timeout_ms) {
mstime_t now = mstime();
if (timeout_ms > LLONG_MAX - now) {
c->bstate.module_blocked_handle = NULL;
addReplyError(c, "timeout is out of range"); /* 'timeout_ms+now' would overflow */
return bc;
}
c->bstate.timeout = timeout_ms + now;
}
 
if (islua || ismulti) {
c->bstate.module_blocked_handle = NULL;
......@@ -7699,13 +7725,12 @@ RedisModuleBlockedClient *moduleBlockClient(RedisModuleCtx *ctx, RedisModuleCmdF
} else if (ctx->flags & REDISMODULE_CTX_BLOCKED_REPLY) {
c->bstate.module_blocked_handle = NULL;
addReplyError(c, "Blocking module command called from a Reply callback context");
}
else if (!auth_reply_callback && clientHasModuleAuthInProgress(c)) {
} else if (!auth_reply_callback && clientHasModuleAuthInProgress(c)) {
c->bstate.module_blocked_handle = NULL;
addReplyError(c, "Clients undergoing module based authentication can only be blocked on auth");
} else {
if (keys) {
blockForKeys(c,BLOCKED_MODULE,keys,numkeys,timeout,flags&REDISMODULE_BLOCK_UNBLOCK_DELETED);
blockForKeys(c,BLOCKED_MODULE,keys,numkeys,c->bstate.timeout,flags&REDISMODULE_BLOCK_UNBLOCK_DELETED);
} else {
blockClient(c,BLOCKED_MODULE);
}
......@@ -8242,6 +8267,11 @@ void moduleHandleBlockedClients(void) {
* properly unblocked by the module. */
bc->disconnect_callback = NULL;
unblockClient(c, 1);
/* Update the wait offset, we don't know if this blocked client propagated anything,
* currently we rather not add any API for that, so we just assume it did. */
c->woff = server.master_repl_offset;
/* Put the client in the list of clients that need to write
* if there are pending replies here. This is needed since
* during a non blocking command the client may receive output. */
......@@ -8610,8 +8640,14 @@ void firePostExecutionUnitJobs(void) {
* infinite loops by halting the execution could result in violation of the feature correctness
* and so Redis will make no attempt to protect the module from infinite loops.
*
* 'free_pd' can be NULL and in such case will not be used. */
* 'free_pd' can be NULL and in such case will not be used.
*
* Return REDISMODULE_OK on success and REDISMODULE_ERR if was called while loading data from disk (AOF or RDB) or
* if the instance is a readonly replica. */
int RM_AddPostNotificationJob(RedisModuleCtx *ctx, RedisModulePostNotificationJobFunc callback, void *privdata, void (*free_privdata)(void*)) {
if (server.loading|| (server.masterhost && server.repl_slave_ro)) {
return REDISMODULE_ERR;
}
RedisModulePostExecUnitJob *job = zmalloc(sizeof(*job));
job->module = ctx->module;
job->callback = callback;
......@@ -8919,7 +8955,7 @@ int RM_GetClusterNodeInfo(RedisModuleCtx *ctx, const char *id, char *ip, char *m
else
memset(master_id,0,REDISMODULE_NODE_ID_LEN);
}
if (port) *port = node->port;
if (port) *port = getNodeDefaultClientPort(node);
 
/* As usually we have to remap flags for modules, in order to ensure
* we can provide binary compatibility. */
......@@ -10635,7 +10671,8 @@ void moduleCallCommandFilters(client *c) {
RedisModuleCommandFilterCtx filter = {
.argv = c->argv,
.argv_len = c->argv_len,
.argc = c->argc
.argc = c->argc,
.c = c
};
 
while((ln = listNext(&li))) {
......@@ -10728,6 +10765,11 @@ int RM_CommandFilterArgDelete(RedisModuleCommandFilterCtx *fctx, int pos)
return REDISMODULE_OK;
}
 
/* Get Client ID for client that issued the command we are filtering */
unsigned long long RM_CommandFilterGetClientId(RedisModuleCommandFilterCtx *fctx) {
return fctx->c->id;
}
/* For a given pointer allocated via RedisModule_Alloc() or
* RedisModule_Realloc(), return the amount of memory allocated for it.
* Note that this may be different (larger) than the memory we allocated
......@@ -11032,12 +11074,12 @@ int RM_ScanKey(RedisModuleKey *key, RedisModuleScanCursor *cursor, RedisModuleSc
vstr = lpGetValue(p,&vlen,&vll);
robj *field = (vstr != NULL) ?
createStringObject((char*)vstr,vlen) :
createObject(OBJ_STRING,sdsfromlonglong(vll));
createStringObjectFromLongLongWithSds(vll);
p = lpNext(o->ptr,p);
vstr = lpGetValue(p,&vlen,&vll);
robj *value = (vstr != NULL) ?
createStringObject((char*)vstr,vlen) :
createObject(OBJ_STRING,sdsfromlonglong(vll));
createStringObjectFromLongLongWithSds(vll);
fn(key, field, value, privdata);
p = lpNext(o->ptr,p);
decrRefCount(field);
......@@ -13712,6 +13754,7 @@ void moduleRegisterCoreAPI(void) {
REGISTER_API(CommandFilterArgInsert);
REGISTER_API(CommandFilterArgReplace);
REGISTER_API(CommandFilterArgDelete);
REGISTER_API(CommandFilterGetClientId);
REGISTER_API(Fork);
REGISTER_API(SendChildHeartbeat);
REGISTER_API(ExitFromChild);
......
......@@ -85,10 +85,6 @@ void freeClientReplyValue(void *o) {
zfree(o);
}
int listMatchObjects(void *a, void *b) {
return equalStringObjects(a,b);
}
/* This function links the client to the global linked list of clients.
* unlinkClient() does the opposite, among other things. */
void linkClient(client *c) {
......@@ -197,7 +193,7 @@ client *createClient(connection *conn) {
c->woff = 0;
c->watched_keys = listCreate();
c->pubsub_channels = dictCreate(&objectKeyPointerValueDictType);
c->pubsub_patterns = listCreate();
c->pubsub_patterns = dictCreate(&objectKeyPointerValueDictType);
c->pubsubshard_channels = dictCreate(&objectKeyPointerValueDictType);
c->peerid = NULL;
c->sockname = NULL;
......@@ -214,8 +210,6 @@ client *createClient(connection *conn) {
c->auth_callback_privdata = NULL;
c->auth_module = NULL;
listInitNode(&c->clients_pending_write_node, c);
listSetFreeMethod(c->pubsub_patterns,decrRefCountVoid);
listSetMatchMethod(c->pubsub_patterns,listMatchObjects);
c->mem_usage_bucket = NULL;
c->mem_usage_bucket_node = NULL;
if (conn) linkClient(c);
......@@ -351,8 +345,8 @@ size_t _addReplyToBuffer(client *c, const char *s, size_t len) {
/* Adds the reply to the reply linked list.
* Note: some edits to this function need to be relayed to AddReplyFromClient. */
void _addReplyProtoToList(client *c, const char *s, size_t len) {
listNode *ln = listLast(c->reply);
void _addReplyProtoToList(client *c, list *reply_list, const char *s, size_t len) {
listNode *ln = listLast(reply_list);
clientReplyBlock *tail = ln? listNodeValue(ln): NULL;
/* Note that 'tail' may be NULL even if we have a tail node, because when
......@@ -380,13 +374,23 @@ void _addReplyProtoToList(client *c, const char *s, size_t len) {
tail->size = usable_size - sizeof(clientReplyBlock);
tail->used = len;
memcpy(tail->buf, s, len);
listAddNodeTail(c->reply, tail);
listAddNodeTail(reply_list, tail);
c->reply_bytes += tail->size;
closeClientOnOutputBufferLimitReached(c, 1);
}
}
/* The subscribe / unsubscribe command family has a push as a reply,
* or in other words, it responds with a push (or several of them
* depending on how many arguments it got), and has no reply. */
int cmdHasPushAsReply(struct redisCommand *cmd) {
if (!cmd) return 0;
return cmd->proc == subscribeCommand || cmd->proc == unsubscribeCommand ||
cmd->proc == psubscribeCommand || cmd->proc == punsubscribeCommand ||
cmd->proc == ssubscribeCommand || cmd->proc == sunsubscribeCommand;
}
void _addReplyToBufferOrList(client *c, const char *s, size_t len) {
if (c->flags & CLIENT_CLOSE_AFTER_REPLY) return;
......@@ -405,8 +409,20 @@ void _addReplyToBufferOrList(client *c, const char *s, size_t len) {
* buffer offset (see function comment) */
reqresSaveClientReplyOffset(c);
/* If we're processing a push message into the current client (i.e. executing PUBLISH
* to a channel which we are subscribed to, then we wanna postpone that message to be added
* after the command's reply (specifically important during multi-exec). the exception is
* the SUBSCRIBE command family, which (currently) have a push message instead of a proper reply.
* The check for executing_client also avoids affecting push messages that are part of eviction. */
if (c == server.current_client && (c->flags & CLIENT_PUSHING) &&
server.executing_client && !cmdHasPushAsReply(server.executing_client->cmd))
{
_addReplyProtoToList(c,server.pending_push_messages,s,len);
return;
}
size_t reply_len = _addReplyToBuffer(c,s,len);
if (len > reply_len) _addReplyProtoToList(c,s+reply_len,len-reply_len);
if (len > reply_len) _addReplyProtoToList(c,c->reply,s+reply_len,len-reply_len);
}
/* -----------------------------------------------------------------------------
......@@ -1599,7 +1615,7 @@ void freeClient(client *c) {
pubsubUnsubscribeShardAllChannels(c, 0);
pubsubUnsubscribeAllPatterns(c,0);
dictRelease(c->pubsub_channels);
listRelease(c->pubsub_patterns);
dictRelease(c->pubsub_patterns);
dictRelease(c->pubsubshard_channels);
/* Free data structures. */
......@@ -1787,8 +1803,9 @@ client *lookupClientByID(uint64_t id) {
* and 'nwritten' is an output parameter, it means how many bytes server write
* to client. */
static int _writevToClient(client *c, ssize_t *nwritten) {
struct iovec iov[IOV_MAX];
int iovcnt = 0;
int iovmax = min(IOV_MAX, c->conn->iovcnt);
struct iovec iov[iovmax];
size_t iov_bytes_len = 0;
/* If the static reply buffer is not empty,
* add it to the iov array for writev() as well. */
......@@ -1804,7 +1821,7 @@ static int _writevToClient(client *c, ssize_t *nwritten) {
listNode *next;
clientReplyBlock *o;
listRewind(c->reply, &iter);
while ((next = listNext(&iter)) && iovcnt < IOV_MAX && iov_bytes_len < NET_MAX_WRITES_PER_EVENT) {
while ((next = listNext(&iter)) && iovcnt < iovmax && iov_bytes_len < NET_MAX_WRITES_PER_EVENT) {
o = listNodeValue(next);
if (o->used == 0) { /* empty node, just release it and skip. */
c->reply_bytes -= o->size;
......@@ -2809,7 +2826,7 @@ sds catClientInfoString(sds s, client *client) {
flags,
client->db->id,
(int) dictSize(client->pubsub_channels),
(int) listLength(client->pubsub_patterns),
(int) dictSize(client->pubsub_patterns),
(int) dictSize(client->pubsubshard_channels),
(client->flags & CLIENT_MULTI) ? client->mstate.count : -1,
(unsigned long long) sdslen(client->querybuf),
......@@ -3946,6 +3963,7 @@ void flushSlavesOutputBuffers(void) {
* 3. Obviously if the slave is not ONLINE.
*/
if (slave->replstate == SLAVE_STATE_ONLINE &&
!(slave->flags & CLIENT_CLOSE_ASAP) &&
can_receive_writes &&
!slave->repl_start_cmd_stream_on_ack &&
clientHasPendingReplies(slave))
......
......@@ -46,15 +46,21 @@ robj *createObject(int type, void *ptr) {
o->encoding = OBJ_ENCODING_RAW;
o->ptr = ptr;
o->refcount = 1;
o->lru = 0;
return o;
}
void initObjectLRUOrLFU(robj *o) {
if (o->refcount == OBJ_SHARED_REFCOUNT)
return;
/* Set the LRU to the current lruclock (minutes resolution), or
* alternatively the LFU counter. */
if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
o->lru = (LFUGetTimeInMinutes()<<8) | LFU_INIT_VAL;
o->lru = (LFUGetTimeInMinutes() << 8) | LFU_INIT_VAL;
} else {
o->lru = LRU_CLOCK();
}
return o;
return;
}
/* Set a special refcount in the object to make it "shared":
......@@ -91,11 +97,7 @@ robj *createEmbeddedStringObject(const char *ptr, size_t len) {
o->encoding = OBJ_ENCODING_EMBSTR;
o->ptr = sh+1;
o->refcount = 1;
if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
o->lru = (LFUGetTimeInMinutes()<<8) | LFU_INIT_VAL;
} else {
o->lru = LRU_CLOCK();
}
o->lru = 0;
sh->len = len;
sh->alloc = len;
......@@ -140,33 +142,24 @@ robj *tryCreateStringObject(const char *ptr, size_t len) {
return tryCreateRawStringObject(ptr,len);
}
/* Create a string object from a long long value. When possible returns a
* shared integer object, or at least an integer encoded one.
*
* If valueobj is non zero, the function avoids returning a shared
* integer, because the object is going to be used as value in the Redis key
* space (for instance when the INCR command is used), so we want LFU/LRU
* values specific for each key. */
robj *createStringObjectFromLongLongWithOptions(long long value, int valueobj) {
/* Create a string object from a long long value according to the specified flag. */
#define LL2STROBJ_AUTO 0 /* automatically create the optimal string object */
#define LL2STROBJ_NO_SHARED 1 /* disallow shared objects */
#define LL2STROBJ_NO_INT_ENC 2 /* disallow integer encoded objects. */
robj *createStringObjectFromLongLongWithOptions(long long value, int flag) {
robj *o;
if (server.maxmemory == 0 ||
!(server.maxmemory_policy & MAXMEMORY_FLAG_NO_SHARED_INTEGERS))
{
/* If the maxmemory policy permits, we can still return shared integers
* even if valueobj is true. */
valueobj = 0;
}
if (value >= 0 && value < OBJ_SHARED_INTEGERS && valueobj == 0) {
if (value >= 0 && value < OBJ_SHARED_INTEGERS && flag == LL2STROBJ_AUTO) {
o = shared.integers[value];
} else {
if (value >= LONG_MIN && value <= LONG_MAX) {
if ((value >= LONG_MIN && value <= LONG_MAX) && flag != LL2STROBJ_NO_INT_ENC) {
o = createObject(OBJ_STRING, NULL);
o->encoding = OBJ_ENCODING_INT;
o->ptr = (void*)((long)value);
} else {
o = createObject(OBJ_STRING,sdsfromlonglong(value));
char buf[LONG_STR_SIZE];
int len = ll2string(buf, sizeof(buf), value);
o = createStringObject(buf, len);
}
}
return o;
......@@ -175,15 +168,27 @@ robj *createStringObjectFromLongLongWithOptions(long long value, int valueobj) {
/* Wrapper for createStringObjectFromLongLongWithOptions() always demanding
* to create a shared object if possible. */
robj *createStringObjectFromLongLong(long long value) {
return createStringObjectFromLongLongWithOptions(value,0);
return createStringObjectFromLongLongWithOptions(value, LL2STROBJ_AUTO);
}
/* Wrapper for createStringObjectFromLongLongWithOptions() avoiding a shared
* object when LFU/LRU info are needed, that is, when the object is used
* as a value in the key space, and Redis is configured to evict based on
* LFU/LRU. */
/* The function avoids returning a shared integer when LFU/LRU info
* are needed, that is, when the object is used as a value in the key
* space(for instance when the INCR command is used), and Redis is
* configured to evict based on LFU/LRU, so we want LFU/LRU values
* specific for each key. */
robj *createStringObjectFromLongLongForValue(long long value) {
return createStringObjectFromLongLongWithOptions(value,1);
if (server.maxmemory == 0 || !(server.maxmemory_policy & MAXMEMORY_FLAG_NO_SHARED_INTEGERS)) {
/* If the maxmemory policy permits, we can still return shared integers */
return createStringObjectFromLongLongWithOptions(value, LL2STROBJ_AUTO);
} else {
return createStringObjectFromLongLongWithOptions(value, LL2STROBJ_NO_SHARED);
}
}
/* Create a string object that contains an sds inside it. That means it can't be
* integer encoded (OBJ_ENCODING_INT), and it'll always be an EMBSTR type. */
robj *createStringObjectFromLongLongWithSds(long long value) {
return createStringObjectFromLongLongWithOptions(value, LL2STROBJ_NO_INT_ENC);
}
/* Create a string object from a long double. If humanfriendly is non-zero
......@@ -627,7 +632,7 @@ void trimStringObjectIfNeeded(robj *o, int trim_small_values) {
}
/* Try to encode a string object in order to save space */
robj *tryObjectEncoding(robj *o) {
robj *tryObjectEncodingEx(robj *o, int try_trim) {
long value;
sds s = o->ptr;
size_t len;
......@@ -692,12 +697,17 @@ robj *tryObjectEncoding(robj *o) {
/* We can't encode the object...
* Do the last try, and at least optimize the SDS string inside */
trimStringObjectIfNeeded(o, 0);
if (try_trim)
trimStringObjectIfNeeded(o, 0);
/* Return the original object. */
return o;
}
robj *tryObjectEncoding(robj *o) {
return tryObjectEncodingEx(o, 1);
}
/* Get a decoded version of an encoded object (returned as a new object).
* If the object is already raw-encoded just increment the ref count. */
robj *getDecodedObject(robj *o) {
......
......@@ -219,7 +219,7 @@ int serverPubsubShardSubscriptionCount(void) {
/* Return the number of channels + patterns a client is subscribed to. */
int clientSubscriptionsCount(client *c) {
return dictSize(c->pubsub_channels) + listLength(c->pubsub_patterns);
return dictSize(c->pubsub_channels) + dictSize(c->pubsub_patterns);
}
/* Return the number of shard level channels a client is subscribed to. */
......@@ -345,9 +345,8 @@ int pubsubSubscribePattern(client *c, robj *pattern) {
list *clients;
int retval = 0;
if (listSearchKey(c->pubsub_patterns,pattern) == NULL) {
if (dictAdd(c->pubsub_patterns, pattern, NULL) == DICT_OK) {
retval = 1;
listAddNodeTail(c->pubsub_patterns,pattern);
incrRefCount(pattern);
/* Add the client to the pattern -> list of clients hash table */
de = dictFind(server.pubsub_patterns,pattern);
......@@ -374,9 +373,8 @@ int pubsubUnsubscribePattern(client *c, robj *pattern, int notify) {
int retval = 0;
incrRefCount(pattern); /* Protect the object. May be the same we remove */
if ((ln = listSearchKey(c->pubsub_patterns,pattern)) != NULL) {
if (dictDelete(c->pubsub_patterns, pattern) == DICT_OK) {
retval = 1;
listDelNode(c->pubsub_patterns,ln);
/* Remove the client from the pattern -> clients list hash table */
de = dictFind(server.pubsub_patterns,pattern);
serverAssertWithInfo(c,NULL,de != NULL);
......@@ -448,16 +446,20 @@ void pubsubUnsubscribeShardChannels(robj **channels, unsigned int count) {
/* Unsubscribe from all the patterns. Return the number of patterns the
* client was subscribed from. */
int pubsubUnsubscribeAllPatterns(client *c, int notify) {
listNode *ln;
listIter li;
int count = 0;
listRewind(c->pubsub_patterns,&li);
while ((ln = listNext(&li)) != NULL) {
robj *pattern = ln->value;
if (dictSize(c->pubsub_patterns) > 0) {
dictIterator *di = dictGetSafeIterator(c->pubsub_patterns);
dictEntry *de;
count += pubsubUnsubscribePattern(c,pattern,notify);
while ((de = dictNext(di)) != NULL) {
robj *pattern = dictGetKey(de);
count += pubsubUnsubscribePattern(c, pattern, notify);
}
dictReleaseIterator(di);
}
/* We were subscribed to nothing? Still reply to the client. */
if (notify && count == 0) addReplyPubsubPatUnsubscribed(c,NULL);
return count;
}
......@@ -743,7 +745,7 @@ void sunsubscribeCommand(client *c) {
size_t pubsubMemOverhead(client *c) {
/* PubSub patterns */
size_t mem = listLength(c->pubsub_patterns) * sizeof(listNode);
size_t mem = dictMemUsage(c->pubsub_patterns);
/* Global PubSub channels */
mem += dictMemUsage(c->pubsub_channels);
/* Sharded PubSub channels */
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment