Commit e9bb30fd authored by antirez's avatar antirez
Browse files

Experimental: new keyspace and expire algorithm.

This is an alpha quality implementation of a new keyspace representation
and a new expire algorithm for Redis.

This work is described here:

    https://gist.github.com/antirez/b2eb293819666ee104c7fcad71986eb7
parent fd0ee469
......@@ -475,6 +475,7 @@ ssize_t rdbSaveStringObject(rio *rdb, robj *obj) {
* RDB_LOAD_PLAIN: Return a plain string allocated with zmalloc()
* instead of a Redis object with an sds in it.
* RDB_LOAD_SDS: Return an SDS string instead of a Redis object.
* RDB_LOAD_KEY: Return a key object instead of a Redis object.
*
* On I/O error NULL is returned.
*/
......@@ -751,7 +752,7 @@ size_t rdbSaveStreamConsumers(rio *rdb, streamCG *cg) {
/* Save a Redis object.
* Returns -1 on error, number of bytes written on success. */
ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key) {
ssize_t rdbSaveObject(rio *rdb, robj *o, rkey *key) {
ssize_t n = 0, nwritten = 0;
if (o->type == OBJ_STRING) {
......@@ -966,7 +967,9 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key) {
RedisModuleIO io;
moduleValue *mv = o->ptr;
moduleType *mt = mv->type;
moduleInitIOContext(io,mt,rdb,key);
robj *keyobj = createStringObject(key->name,key->len);
moduleInitIOContext(io,mt,rdb,keyobj);
decrRefCount(keyobj);
/* Write the "module" identifier as prefix, so that we'll be able
* to call the right module during loading. */
......@@ -1005,7 +1008,7 @@ size_t rdbSavedObjectLen(robj *o) {
* On error -1 is returned.
* On success if the key was actually saved 1 is returned, otherwise 0
* is returned (the key was already expired). */
int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime) {
int rdbSaveKeyValuePair(rio *rdb, rkey *key, robj *val, long long expiretime) {
int savelru = server.maxmemory_policy & MAXMEMORY_FLAG_LRU;
int savelfu = server.maxmemory_policy & MAXMEMORY_FLAG_LFU;
......@@ -1017,7 +1020,7 @@ int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime) {
/* Save the LRU info. */
if (savelru) {
uint64_t idletime = estimateObjectIdleTime(val);
uint64_t idletime = estimateObjectIdleTime(key);
idletime /= 1000; /* Using seconds is enough and requires less space.*/
if (rdbSaveType(rdb,RDB_OPCODE_IDLE) == -1) return -1;
if (rdbSaveLen(rdb,idletime) == -1) return -1;
......@@ -1026,7 +1029,7 @@ int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime) {
/* Save the LFU info. */
if (savelfu) {
uint8_t buf[1];
buf[0] = LFUDecrAndReturn(val);
buf[0] = LFUDecrAndReturn(key);
/* We can encode this in exactly two bytes: the opcode and an 8
* bit counter, since the frequency is logarithmic with a 0-255 range.
* Note that we do not store the halving time because to reset it
......@@ -1037,7 +1040,8 @@ int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime) {
/* Save type, key, value */
if (rdbSaveObjectType(rdb,val) == -1) return -1;
if (rdbSaveStringObject(rdb,key) == -1) return -1;
if (rdbSaveRawString(rdb,(unsigned char*)key->name,key->len) == -1)
return -1;
if (rdbSaveObject(rdb,val,key) == -1) return -1;
return 1;
}
......@@ -1129,20 +1133,18 @@ int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi) {
* these sizes are just hints to resize the hash tables. */
uint64_t db_size, expires_size;
db_size = dictSize(db->dict);
expires_size = dictSize(db->expires);
expires_size = raxSize(db->expires);
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;
/* Iterate this DB writing every entry */
while((de = dictNext(di)) != NULL) {
sds keystr = dictGetKey(de);
robj key, *o = dictGetVal(de);
long long expire;
rkey *key = dictGetKey(de);
robj *o = dictGetVal(de);
initStaticStringObject(key,keystr);
expire = getExpire(db,&key);
if (rdbSaveKeyValuePair(rdb,&key,o,expire) == -1) goto werr;
if (rdbSaveKeyValuePair(rdb,key,o,getExpire(key)) == -1)
goto werr;
/* When this RDB is produced as part of an AOF rewrite, move
* accumulated diff from parent to child while rewriting in
......@@ -1928,7 +1930,6 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) {
if ((expires_size = rdbLoadLen(rdb,NULL)) == RDB_LENERR)
goto eoferr;
dictExpand(db->dict,db_size);
dictExpand(db->expires,expires_size);
continue; /* Read next opcode. */
} else if (type == RDB_OPCODE_AUX) {
/* AUX: generic string-string fields. Use to add state to RDB
......@@ -2034,17 +2035,13 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) {
decrRefCount(val);
} else {
/* Add the new object in the hash table */
dbAdd(db,key,val);
rkey *k = dbAdd(db,key,val);
/* Set the expire time if needed */
if (expiretime != -1) setExpire(NULL,db,key,expiretime);
if (expiretime != -1) setExpire(NULL,db,k,expiretime);
/* Set usage information (for eviction). */
objectSetLRUOrLFU(val,lfu_freq,lru_idle,lru_clock);
/* Decrement the key refcount since dbAdd() will take its
* own reference. */
decrRefCount(key);
objectSetLRUOrLFU(k,lfu_freq,lru_idle,lru_clock);
}
/* Reset the state that is key-specified and is populated by
......
......@@ -120,6 +120,7 @@
#define RDB_LOAD_ENC (1<<0)
#define RDB_LOAD_PLAIN (1<<1)
#define RDB_LOAD_SDS (1<<2)
#define RDB_LOAD_KEY (1<<3)
#define RDB_SAVE_NONE 0
#define RDB_SAVE_AOF_PREAMBLE (1<<0)
......@@ -140,11 +141,11 @@ int rdbSaveBackground(char *filename, rdbSaveInfo *rsi);
int rdbSaveToSlavesSockets(rdbSaveInfo *rsi);
void rdbRemoveTempFile(pid_t childpid);
int rdbSave(char *filename, rdbSaveInfo *rsi);
ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key);
ssize_t rdbSaveObject(rio *rdb, robj *o, rkey *key);
size_t rdbSavedObjectLen(robj *o);
robj *rdbLoadObject(int type, rio *rdb, robj *key);
void backgroundSaveDoneHandler(int exitcode, int bysignal);
int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime);
int rdbSaveKeyValuePair(rio *rdb, rkey *key, robj *val, long long expiretime);
robj *rdbLoadStringObject(rio *rdb);
ssize_t rdbSaveStringObject(rio *rdb, robj *obj);
ssize_t rdbSaveRawString(rio *rdb, unsigned char *s, size_t len);
......
......@@ -1164,7 +1164,8 @@ static int fetchClusterSlotsConfiguration(client c) {
printf("Cluster slots configuration changed, fetching new one...\n");
const char *errmsg = "Failed to update cluster slots configuration";
static dictType dtype = {
dictSdsHash, /* hash function */
dictSdsHash, /* lookup hash function */
dictSdsHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCompare, /* key compare */
......
......@@ -1987,7 +1987,8 @@ typedef struct clusterManagerLink {
} clusterManagerLink;
static dictType clusterManagerDictType = {
dictSdsHash, /* hash function */
dictSdsHash, /* lookup hash function */
dictSdsHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCompare, /* key compare */
......@@ -1996,10 +1997,12 @@ static dictType clusterManagerDictType = {
};
static dictType clusterManagerLinkDictType = {
dictSdsHash, /* hash function */
dictSdsHash, /* lookup hash function */
dictSdsHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCompare, /* key compare */
dictSdsKeyCompare, /* lookup key compare */
dictSdsKeyCompare, /* stored key compare */
dictSdsDestructor, /* key destructor */
dictListDestructor /* val destructor */
};
......@@ -6933,7 +6936,8 @@ void type_free(void* priv_data, void* val) {
}
static dictType typeinfoDictType = {
dictSdsHash, /* hash function */
dictSdsHash, /* lookup hash function */
dictSdsHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCompare, /* key compare */
......
......@@ -402,7 +402,8 @@ void dictInstancesValDestructor (void *privdata, void *obj) {
* also used for: sentinelRedisInstance->sentinels dictionary that maps
* sentinels ip:port to last seen time in Pub/Sub hello message. */
dictType instancesDictType = {
dictSdsHash, /* hash function */
dictSdsHash, /* lookup hash function */
dictSdsHash, /* store hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCompare, /* key compare */
......@@ -415,7 +416,8 @@ dictType instancesDictType = {
* This is useful into sentinelGetObjectiveLeader() function in order to
* count the votes and understand who is the leader. */
dictType leaderVotesDictType = {
dictSdsHash, /* hash function */
dictSdsHash, /* lookup hash function */
dictSdsHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCompare, /* key compare */
......@@ -425,10 +427,12 @@ dictType leaderVotesDictType = {
/* Instance renamed commands table. */
dictType renamedCommandsDictType = {
dictSdsCaseHash, /* hash function */
dictSdsCaseHash, /* lookup hash function */
dictSdsCaseHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCaseCompare, /* key compare */
dictSdsKeyCaseCompare, /* lookup key compare */
dictSdsKeyCaseCompare, /* stored key compare */
dictSdsDestructor, /* key destructor */
dictSdsDestructor /* val destructor */
};
......
......@@ -1142,16 +1142,33 @@ void dictListDestructor(void *privdata, void *val)
listRelease((list*)val);
}
int dictSdsKeyCompare(void *privdata, const void *key1,
const void *key2)
{
static int dictGenericKeyCompare(const char *key1, const char *key2, size_t len1, size_t len2) {
if (len1 != len2) return 0;
return memcmp(key1, key2, len1) == 0;
}
int dictSdsKeyCompare(void *privdata, const void *key1, const void *key2) {
int l1,l2;
DICT_NOTUSED(privdata);
l1 = sdslen((sds)key1);
l2 = sdslen((sds)key2);
if (l1 != l2) return 0;
return memcmp(key1, key2, l1) == 0;
return dictGenericKeyCompare(key1,key2,l1,l2);
}
int dictSdsRkeyKeyCompare(void *privdata, const void *key1, const void *key2){
DICT_NOTUSED(privdata);
sds sdskey = (sds)key1;
rkey *keyobj = (rkey*)key2;
return dictGenericKeyCompare(sdskey,keyobj->name,
sdslen(sdskey),keyobj->len);
}
int dictRkeyKeyCompare(void *privdata, const void *key1, const void *key2) {
DICT_NOTUSED(privdata);
rkey *k1 = (rkey*)key1;
rkey *k2 = (rkey*)key2;
return dictGenericKeyCompare(k1->name,k2->name,k1->len,k2->len);
}
/* A case insensitive version used for the command lookup table and other
......@@ -1179,6 +1196,13 @@ void dictSdsDestructor(void *privdata, void *val)
sdsfree(val);
}
void dictKeyDestructor(void *privdata, void *val)
{
DICT_NOTUSED(privdata);
freeKey(val);
}
int dictObjKeyCompare(void *privdata, const void *key1,
const void *key2)
{
......@@ -1195,6 +1219,11 @@ uint64_t dictSdsHash(const void *key) {
return dictGenHashFunction((unsigned char*)key, sdslen((char*)key));
}
uint64_t dictKeyHash(const void *keyptr) {
const rkey *key = keyptr;
return dictGenHashFunction(key->name, key->len);
}
uint64_t dictSdsCaseHash(const void *key) {
return dictGenCaseHashFunction((unsigned char*)key, sdslen((char*)key));
}
......@@ -1243,10 +1272,12 @@ uint64_t dictEncObjHash(const void *key) {
/* Generic hash table type where keys are Redis Objects, Values
* dummy pointers. */
dictType objectKeyPointerValueDictType = {
dictEncObjHash, /* hash function */
dictEncObjHash, /* lookup hash function */
dictEncObjHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictEncObjKeyCompare, /* key compare */
dictEncObjKeyCompare, /* lookup key compare */
dictEncObjKeyCompare, /* stored key compare */
dictObjectDestructor, /* key destructor */
NULL /* val destructor */
};
......@@ -1254,80 +1285,100 @@ dictType objectKeyPointerValueDictType = {
/* Like objectKeyPointerValueDictType(), but values can be destroyed, if
* not NULL, calling zfree(). */
dictType objectKeyHeapPointerValueDictType = {
dictEncObjHash, /* hash function */
dictEncObjHash, /* lookup hash function */
dictEncObjHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictEncObjKeyCompare, /* key compare */
dictEncObjKeyCompare, /* lookup key compare */
dictEncObjKeyCompare, /* stored key compare */
dictObjectDestructor, /* key destructor */
dictVanillaFree /* val destructor */
};
/* Set dictionary type. Keys are SDS strings, values are ot used. */
dictType setDictType = {
dictSdsHash, /* hash function */
dictSdsHash, /* lookup hash function */
dictSdsHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCompare, /* key compare */
dictSdsKeyCompare, /* lookup key compare */
dictSdsKeyCompare, /* stored key compare */
dictSdsDestructor, /* key destructor */
NULL /* val destructor */
};
/* Sorted sets hash (note: a skiplist is used in addition to the hash table) */
dictType zsetDictType = {
dictSdsHash, /* hash function */
dictSdsHash, /* lookup hash function */
dictSdsHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCompare, /* key compare */
dictSdsKeyCompare, /* lookup key compare */
dictSdsKeyCompare, /* stored key compare */
NULL, /* Note: SDS string shared & freed by skiplist */
NULL /* val destructor */
};
/* Db->dict, keys are sds strings, vals are Redis objects. */
/* Db->dict, keys are "rkey" key objects, vals are "robj" Redis objects.
*
* Note that this dictionary is designed to be looked up via SDS strings
* even if keys are stored as rkey structures. So there are two differet
* hash functions and two different compare functions. */
dictType dbDictType = {
dictSdsHash, /* hash function */
dictSdsHash, /* lookup hash function */
dictKeyHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCompare, /* key compare */
dictSdsDestructor, /* key destructor */
dictSdsRkeyKeyCompare, /* lookup key compare */
dictRkeyKeyCompare, /* stored key compare */
dictKeyDestructor, /* key destructor */
dictObjectDestructor /* val destructor */
};
/* server.lua_scripts sha (as sds string) -> scripts (as robj) cache. */
dictType shaScriptObjectDictType = {
dictSdsCaseHash, /* hash function */
dictSdsCaseHash, /* lookup hash function */
dictSdsCaseHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCaseCompare, /* key compare */
dictSdsKeyCaseCompare, /* lookup key compare */
dictSdsKeyCaseCompare, /* stored key compare */
dictSdsDestructor, /* key destructor */
dictObjectDestructor /* val destructor */
};
/* Db->expires */
dictType keyptrDictType = {
dictSdsHash, /* hash function */
dictSdsHash, /* look hash function */
dictSdsHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCompare, /* key compare */
dictSdsKeyCompare, /* lookup compare */
dictSdsKeyCompare, /* stored compare */
NULL, /* key destructor */
NULL /* val destructor */
};
/* Command table. sds string -> command struct pointer. */
dictType commandTableDictType = {
dictSdsCaseHash, /* hash function */
dictSdsCaseHash, /* lookup hash function */
dictSdsCaseHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCaseCompare, /* key compare */
dictSdsKeyCaseCompare, /* lookup key compare */
dictSdsKeyCaseCompare, /* stored key compare */
dictSdsDestructor, /* key destructor */
NULL /* val destructor */
};
/* Hash type hash table (note that small hashes are represented with ziplists) */
/* Hash values type (note that small hashes are represented with ziplists) */
dictType hashDictType = {
dictSdsHash, /* hash function */
dictSdsHash, /* lookup hash function */
dictSdsHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCompare, /* key compare */
dictSdsKeyCompare, /* lookup key compare */
dictSdsKeyCompare, /* stored key compare */
dictSdsDestructor, /* key destructor */
dictSdsDestructor /* val destructor */
};
......@@ -1336,10 +1387,12 @@ dictType hashDictType = {
* lists as values. It's used for blocking operations (BLPOP) and to
* map swapped keys to a list of clients waiting for this keys to be loaded. */
dictType keylistDictType = {
dictObjHash, /* hash function */
dictObjHash, /* lookup hash function */
dictObjHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictObjKeyCompare, /* key compare */
dictObjKeyCompare, /* lookup key compare */
dictObjKeyCompare, /* stored key compare */
dictObjectDestructor, /* key destructor */
dictListDestructor /* val destructor */
};
......@@ -1347,10 +1400,12 @@ dictType keylistDictType = {
/* Cluster nodes hash table, mapping nodes addresses 1.2.3.4:6379 to
* clusterNode structures. */
dictType clusterNodesDictType = {
dictSdsHash, /* hash function */
dictSdsHash, /* lookup hash function */
dictSdsHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCompare, /* key compare */
dictSdsKeyCompare, /* lookup key compare */
dictSdsKeyCompare, /* stored key compare */
dictSdsDestructor, /* key destructor */
NULL /* val destructor */
};
......@@ -1359,10 +1414,12 @@ dictType clusterNodesDictType = {
* we can re-add this node. The goal is to avoid readding a removed
* node for some time. */
dictType clusterNodesBlackListDictType = {
dictSdsCaseHash, /* hash function */
dictSdsCaseHash, /* lookup hash function */
dictSdsCaseHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCaseCompare, /* key compare */
dictSdsKeyCaseCompare, /* lookup key compare */
dictSdsKeyCaseCompare, /* stored key compare */
dictSdsDestructor, /* key destructor */
NULL /* val destructor */
};
......@@ -1371,20 +1428,24 @@ dictType clusterNodesBlackListDictType = {
* we can re-add this node. The goal is to avoid readding a removed
* node for some time. */
dictType modulesDictType = {
dictSdsCaseHash, /* hash function */
dictSdsCaseHash, /* lookup hash function */
dictSdsCaseHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCaseCompare, /* key compare */
dictSdsKeyCaseCompare, /* lookup key compare */
dictSdsKeyCaseCompare, /* stored key compare */
dictSdsDestructor, /* key destructor */
NULL /* val destructor */
};
/* Migrate cache dict type. */
dictType migrateCacheDictType = {
dictSdsHash, /* hash function */
dictSdsHash, /* lookup hash function */
dictSdsHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCompare, /* key compare */
dictSdsKeyCompare, /* lookup key compare */
dictSdsKeyCompare, /* stored key compare */
dictSdsDestructor, /* key destructor */
NULL /* val destructor */
};
......@@ -1393,10 +1454,12 @@ dictType migrateCacheDictType = {
* Keys are sds SHA1 strings, while values are not used at all in the current
* implementation. */
dictType replScriptCacheDictType = {
dictSdsCaseHash, /* hash function */
dictSdsCaseHash, /* lookup hash function */
dictSdsCaseHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCaseCompare, /* key compare */
dictSdsKeyCaseCompare, /* lookup key compare */
dictSdsKeyCaseCompare, /* stored key compare */
dictSdsDestructor, /* key destructor */
NULL /* val destructor */
};
......@@ -1415,8 +1478,6 @@ int htNeedsResize(dict *dict) {
void tryResizeHashTables(int dbid) {
if (htNeedsResize(server.db[dbid].dict))
dictResize(server.db[dbid].dict);
if (htNeedsResize(server.db[dbid].expires))
dictResize(server.db[dbid].expires);
}
/* Our hash table implementation performs rehashing incrementally while
......@@ -1432,11 +1493,6 @@ int incrementallyRehash(int dbid) {
dictRehashMilliseconds(server.db[dbid].dict,1);
return 1; /* already used our millisecond for this loop... */
}
/* Expires */
if (dictIsRehashing(server.db[dbid].expires)) {
dictRehashMilliseconds(server.db[dbid].expires,1);
return 1; /* already used our millisecond for this loop... */
}
return 0;
}
......@@ -1859,7 +1915,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
size = dictSlots(server.db[j].dict);
used = dictSize(server.db[j].dict);
vkeys = dictSize(server.db[j].expires);
vkeys = raxSize(server.db[j].expires);
if (used || vkeys) {
serverLog(LL_VERBOSE,"DB %d: %lld keys (%lld volatile) in %lld slots HT.",j,used,vkeys,size);
/* dictPrintStats(server.dict); */
......@@ -2668,7 +2724,6 @@ void resetServerStats(void) {
server.stat_numcommands = 0;
server.stat_numconnections = 0;
server.stat_expiredkeys = 0;
server.stat_expired_stale_perc = 0;
server.stat_expired_time_cap_reached_count = 0;
server.stat_evictedkeys = 0;
server.stat_keyspace_misses = 0;
......@@ -2763,7 +2818,7 @@ void initServer(void) {
/* Create the Redis databases, and initialize other internal state. */
for (j = 0; j < server.dbnum; j++) {
server.db[j].dict = dictCreate(&dbDictType,NULL);
server.db[j].expires = dictCreate(&keyptrDictType,NULL);
server.db[j].expires = raxNew();
server.db[j].blocking_keys = dictCreate(&keylistDictType,NULL);
server.db[j].ready_keys = dictCreate(&objectKeyPointerValueDictType,NULL);
server.db[j].watched_keys = dictCreate(&keylistDictType,NULL);
......@@ -4109,7 +4164,6 @@ sds genRedisInfoString(char *section) {
"sync_partial_ok:%lld\r\n"
"sync_partial_err:%lld\r\n"
"expired_keys:%lld\r\n"
"expired_stale_perc:%.2f\r\n"
"expired_time_cap_reached_count:%lld\r\n"
"evicted_keys:%lld\r\n"
"keyspace_hits:%lld\r\n"
......@@ -4135,7 +4189,6 @@ sds genRedisInfoString(char *section) {
server.stat_sync_partial_ok,
server.stat_sync_partial_err,
server.stat_expiredkeys,
server.stat_expired_stale_perc*100,
server.stat_expired_time_cap_reached_count,
server.stat_evictedkeys,
server.stat_keyspace_hits,
......@@ -4331,7 +4384,7 @@ sds genRedisInfoString(char *section) {
long long keys, vkeys;
keys = dictSize(server.db[j].dict);
vkeys = dictSize(server.db[j].expires);
vkeys = raxSize(server.db[j].expires);
if (keys || vkeys) {
info = sdscatprintf(info,
"db%d:keys=%lld,expires=%lld,avg_ttl=%lld\r\n",
......
......@@ -634,14 +634,13 @@ typedef struct RedisModuleDigest {
#define LRU_BITS 24
#define LRU_CLOCK_MAX ((1<<LRU_BITS)-1) /* Max value of obj->lru */
#define LRU_CLOCK_RESOLUTION 1000 /* LRU clock resolution in ms */
#define KEY_FLAGS_BITS (32-LRU_BITS)
#define OBJ_SHARED_REFCOUNT INT_MAX
typedef struct redisObject {
unsigned type:4;
unsigned encoding:4;
unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or
* LFU data (least significant 8 bits frequency
* and most significant 16 bits access time). */
unsigned flags:24;
int refcount;
void *ptr;
} robj;
......@@ -660,18 +659,34 @@ typedef struct redisObject {
struct evictionPoolEntry; /* Defined in evict.c */
/* This structure is used in order to represent the output buffer of a client,
* which is actually a linked list of blocks like that, that is: client->reply. */
* which is actually a linked list of clientReplyBlock blocks.
* Such list is stored in the client->reply field. */
typedef struct clientReplyBlock {
size_t size, used;
char buf[];
} clientReplyBlock;
/* Representation of a Redis key. This representation is just used in order
* to store keys in the main dictionary. Note that keys with an expire are
* also stored in a different data structure as well, stored in db->expires,
* for the expiration algorithm to work more efficiently. */
#define KEY_FLAG_EXPIRE (1<<0)
typedef struct redisKey {
uint32_t len;
unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or
* LFU data (least significant 8 bits frequency
* and most significant 16 bits access time). */
unsigned flags:KEY_FLAGS_BITS;
uint64_t expire;
char name[];
} rkey;
/* 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. */
typedef struct redisDb {
dict *dict; /* The keyspace for this DB */
dict *expires; /* Timeout of keys with a timeout set */
rax *expires; /* Sorted tree of keys with an expire set. */
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 */
......@@ -1089,7 +1104,6 @@ struct redisServer {
long long stat_numcommands; /* Number of processed commands */
long long stat_numconnections; /* Number of connections received */
long long stat_expiredkeys; /* Number of expired keys */
double stat_expired_stale_perc; /* Percentage of keys probably expired */
long long stat_expired_time_cap_reached_count; /* Early expire cylce stops.*/
long long stat_evictedkeys; /* Number of evicted keys (maxmemory) */
long long stat_keyspace_hits; /* Number of successful lookups of keys */
......@@ -1674,7 +1688,7 @@ char *strEncoding(int encoding);
int compareStringObjects(robj *a, robj *b);
int collateStringObjects(robj *a, robj *b);
int equalStringObjects(robj *a, robj *b);
unsigned long long estimateObjectIdleTime(robj *o);
unsigned long long estimateObjectIdleTime(rkey *k);
void trimStringObjectIfNeeded(robj *o);
#define sdsEncodedObject(objptr) (objptr->encoding == OBJ_ENCODING_RAW || objptr->encoding == OBJ_ENCODING_EMBSTR)
......@@ -1941,31 +1955,34 @@ void rewriteConfigRewriteLine(struct rewriteConfigState *state, const char *opti
int rewriteConfig(char *path);
/* db.c -- Keyspace access API */
int removeExpire(redisDb *db, robj *key);
int removeExpire(redisDb *db, rkey *key);
void propagateExpire(redisDb *db, robj *key, int lazy);
int expireIfNeeded(redisDb *db, robj *key);
long long getExpire(redisDb *db, robj *key);
void setExpire(client *c, redisDb *db, robj *key, long long when);
robj *lookupKey(redisDb *db, robj *key, int flags);
robj *lookupKeyRead(redisDb *db, robj *key);
robj *lookupKeyWrite(redisDb *db, robj *key);
robj *lookupKeyReadOrReply(client *c, robj *key, robj *reply);
robj *lookupKeyWriteOrReply(client *c, robj *key, robj *reply);
robj *lookupKeyReadWithFlags(redisDb *db, robj *key, int flags);
robj *objectCommandLookup(client *c, robj *key);
robj *objectCommandLookupOrReply(client *c, robj *key, robj *reply);
void objectSetLRUOrLFU(robj *val, long long lfu_freq, long long lru_idle,
int expireIfNeeded(redisDb *db, robj *keyname, rkey *key);
int expireIfNeededByName(redisDb *db, robj *keyname);
long long getExpire(rkey *key);
void setExpire(client *c, redisDb *db, rkey *key, long long when);
robj *lookupKey(redisDb *db, robj *keyname, rkey **keyptr, int flags);
robj *lookupKeyRead(redisDb *db, robj *keyname, rkey **keyptr);
robj *lookupKeyWrite(redisDb *db, robj *keyname, rkey **keyptr);
robj *lookupKeyReadOrReply(client *c, robj *keyname, rkey **keyptr, robj *reply);
robj *lookupKeyWriteOrReply(client *c, robj *keyname, rkey **keyptr, robj *reply);
robj *lookupKeyReadWithFlags(redisDb *db, robj *keyname, rkey **keyptr, int flags);
void objectSetLRUOrLFU(rkey *key, long long lfu_freq, long long lru_idle,
long long lru_clock);
robj *objectCommandLookup(client *c, robj *keyname, rkey **key);
robj *objectCommandLookupOrReply(client *c, robj *keyname, rkey **key, robj *reply);
#define LOOKUP_NONE 0
#define LOOKUP_NOTOUCH (1<<0)
void dbAdd(redisDb *db, robj *key, robj *val);
void dbOverwrite(redisDb *db, robj *key, robj *val);
void setKey(redisDb *db, robj *key, robj *val);
rkey *dbAdd(redisDb *db, robj *key, robj *val);
rkey *dbOverwrite(redisDb *db, robj *key, robj *val);
rkey *setKey(redisDb *db, robj *keyname, robj *val);
int dbExists(redisDb *db, robj *key);
robj *dbRandomKey(redisDb *db);
int dbSyncDelete(redisDb *db, robj *key);
int dbDelete(redisDb *db, robj *key);
robj *dbUnshareStringValue(redisDb *db, robj *key, robj *o);
void freeKey(rkey *key);
void removeExpireFromTree(redisDb *db, rkey *key);
#define EMPTYDB_NO_FLAGS 0 /* No flags. */
#define EMPTYDB_ASYNC (1<<0) /* Reclaim memory in another thread. */
......@@ -2043,7 +2060,7 @@ void blockForKeys(client *c, int btype, robj **keys, int numkeys, mstime_t timeo
/* expire.c -- Handling of expired keys */
void activeExpireCycle(int type);
void expireSlaveKeys(void);
void rememberSlaveKeyWithExpire(redisDb *db, robj *key);
void rememberSlaveKeyWithExpire(redisDb *db, rkey *key);
void flushSlaveKeysWithExpireList(void);
size_t getSlaveKeyWithExpireCount(void);
......@@ -2052,7 +2069,7 @@ void evictionPoolAlloc(void);
#define LFU_INIT_VAL 5
unsigned long LFUGetTimeInMinutes(void);
uint8_t LFULogIncr(uint8_t value);
unsigned long LFUDecrAndReturn(robj *o);
unsigned long LFUDecrAndReturn(rkey *k);
/* Keys hashing / comparison functions for dict.c hash tables. */
uint64_t dictSdsHash(const void *key);
......
......@@ -107,9 +107,9 @@ robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst, int writeflag)
/* Lookup substituted key */
if (!writeflag)
o = lookupKeyRead(db,keyobj);
o = lookupKeyRead(db,keyobj,NULL);
else
o = lookupKeyWrite(db,keyobj);
o = lookupKeyWrite(db,keyobj,NULL);
if (o == NULL) goto noobj;
if (fieldobj) {
......@@ -271,9 +271,9 @@ void sortCommand(client *c) {
/* Lookup the key to sort. It must be of the right types */
if (storekey)
sortval = lookupKeyRead(c->db,c->argv[1]);
sortval = lookupKeyRead(c->db,c->argv[1],NULL);
else
sortval = lookupKeyWrite(c->db,c->argv[1]);
sortval = lookupKeyWrite(c->db,c->argv[1],NULL);
if (sortval && sortval->type != OBJ_SET &&
sortval->type != OBJ_LIST &&
sortval->type != OBJ_ZSET)
......
......@@ -449,7 +449,7 @@ sds hashTypeCurrentObjectNewSds(hashTypeIterator *hi, int what) {
}
robj *hashTypeLookupWriteOrCreate(client *c, robj *key) {
robj *o = lookupKeyWrite(c->db,key);
robj *o = lookupKeyWrite(c->db,key,NULL);
if (o == NULL) {
o = createHashObject();
dbAdd(c->db,key,o);
......@@ -679,8 +679,8 @@ static void addHashFieldToReply(client *c, robj *o, sds field) {
void hgetCommand(client *c) {
robj *o;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp])) == NULL ||
checkType(c,o,OBJ_HASH)) return;
if ((o = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.null[c->resp]))
== NULL || checkType(c,o,OBJ_HASH)) return;
addHashFieldToReply(c, o, c->argv[2]->ptr);
}
......@@ -691,7 +691,7 @@ void hmgetCommand(client *c) {
/* Don't abort when the key cannot be found. Non-existing keys are empty
* hashes, where HMGET should respond with a series of null bulks. */
o = lookupKeyRead(c->db, c->argv[1]);
o = lookupKeyRead(c->db, c->argv[1], NULL);
if (o != NULL && o->type != OBJ_HASH) {
addReply(c, shared.wrongtypeerr);
return;
......@@ -707,7 +707,7 @@ void hdelCommand(client *c) {
robj *o;
int j, deleted = 0, keyremoved = 0;
if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
if ((o = lookupKeyWriteOrReply(c,c->argv[1],NULL,shared.czero)) == NULL ||
checkType(c,o,OBJ_HASH)) return;
for (j = 2; j < c->argc; j++) {
......@@ -734,7 +734,7 @@ void hdelCommand(client *c) {
void hlenCommand(client *c) {
robj *o;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
if ((o = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.czero)) == NULL ||
checkType(c,o,OBJ_HASH)) return;
addReplyLongLong(c,hashTypeLength(o));
......@@ -743,7 +743,7 @@ void hlenCommand(client *c) {
void hstrlenCommand(client *c) {
robj *o;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
if ((o = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.czero)) == NULL ||
checkType(c,o,OBJ_HASH)) return;
addReplyLongLong(c,hashTypeGetValueLength(o,c->argv[2]->ptr));
}
......@@ -772,8 +772,8 @@ void genericHgetallCommand(client *c, int flags) {
hashTypeIterator *hi;
int length, count = 0;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp])) == NULL
|| checkType(c,o,OBJ_HASH)) return;
if ((o = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.null[c->resp]))
== NULL || checkType(c,o,OBJ_HASH)) return;
/* We return a map if the user requested keys and values, like in the
* HGETALL case. Otherwise to use a flat array makes more sense. */
......@@ -817,7 +817,7 @@ void hgetallCommand(client *c) {
void hexistsCommand(client *c) {
robj *o;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
if ((o = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.czero)) == NULL ||
checkType(c,o,OBJ_HASH)) return;
addReply(c, hashTypeExists(o,c->argv[2]->ptr) ? shared.cone : shared.czero);
......@@ -828,7 +828,7 @@ void hscanCommand(client *c) {
unsigned long cursor;
if (parseScanCursorOrReply(c,c->argv[2],&cursor) == C_ERR) return;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptyscan)) == NULL ||
checkType(c,o,OBJ_HASH)) return;
if ((o = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.emptyscan))
== NULL || checkType(c,o,OBJ_HASH)) return;
scanGenericCommand(c,o,cursor);
}
......@@ -196,7 +196,7 @@ void listTypeConvert(robj *subject, int enc) {
void pushGenericCommand(client *c, int where) {
int j, pushed = 0;
robj *lobj = lookupKeyWrite(c->db,c->argv[1]);
robj *lobj = lookupKeyWrite(c->db,c->argv[1],NULL);
if (lobj && lobj->type != OBJ_LIST) {
addReply(c,shared.wrongtypeerr);
......@@ -235,8 +235,8 @@ void pushxGenericCommand(client *c, int where) {
int j, pushed = 0;
robj *subject;
if ((subject = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
checkType(c,subject,OBJ_LIST)) return;
if ((subject = lookupKeyWriteOrReply(c,c->argv[1],NULL,shared.czero))
== NULL || checkType(c,subject,OBJ_LIST)) return;
for (j = 2; j < c->argc; j++) {
listTypePush(subject,c->argv[j],where);
......@@ -277,8 +277,8 @@ void linsertCommand(client *c) {
return;
}
if ((subject = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
checkType(c,subject,OBJ_LIST)) return;
if ((subject = lookupKeyWriteOrReply(c,c->argv[1],NULL,shared.czero))
== NULL || checkType(c,subject,OBJ_LIST)) return;
/* Seek pivot from head to tail */
iter = listTypeInitIterator(subject,0,LIST_TAIL);
......@@ -306,13 +306,13 @@ void linsertCommand(client *c) {
}
void llenCommand(client *c) {
robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.czero);
robj *o = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.czero);
if (o == NULL || checkType(c,o,OBJ_LIST)) return;
addReplyLongLong(c,listTypeLength(o));
}
void lindexCommand(client *c) {
robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp]);
robj *o = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.null[c->resp]);
if (o == NULL || checkType(c,o,OBJ_LIST)) return;
long index;
robj *value = NULL;
......@@ -339,7 +339,7 @@ void lindexCommand(client *c) {
}
void lsetCommand(client *c) {
robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr);
robj *o = lookupKeyWriteOrReply(c,c->argv[1],NULL,shared.nokeyerr);
if (o == NULL || checkType(c,o,OBJ_LIST)) return;
long index;
robj *value = c->argv[3];
......@@ -365,7 +365,7 @@ void lsetCommand(client *c) {
}
void popGenericCommand(client *c, int where) {
robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.null[c->resp]);
robj *o = lookupKeyWriteOrReply(c,c->argv[1],NULL,shared.null[c->resp]);
if (o == NULL || checkType(c,o,OBJ_LIST)) return;
robj *value = listTypePop(o,where);
......@@ -402,8 +402,8 @@ void lrangeCommand(client *c) {
if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != C_OK) ||
(getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != C_OK)) return;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp])) == NULL
|| checkType(c,o,OBJ_LIST)) return;
if ((o = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.null[c->resp]))
== NULL || checkType(c,o,OBJ_LIST)) return;
llen = listTypeLength(o);
/* convert negative indexes */
......@@ -448,7 +448,7 @@ void ltrimCommand(client *c) {
if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != C_OK) ||
(getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != C_OK)) return;
if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.ok)) == NULL ||
if ((o = lookupKeyWriteOrReply(c,c->argv[1],NULL,shared.ok)) == NULL ||
checkType(c,o,OBJ_LIST)) return;
llen = listTypeLength(o);
......@@ -496,7 +496,7 @@ void lremCommand(client *c) {
if ((getLongFromObjectOrReply(c, c->argv[2], &toremove, NULL) != C_OK))
return;
subject = lookupKeyWriteOrReply(c,c->argv[1],shared.czero);
subject = lookupKeyWriteOrReply(c,c->argv[1],NULL,shared.czero);
if (subject == NULL || checkType(c,subject,OBJ_LIST)) return;
listTypeIterator *li;
......@@ -564,7 +564,7 @@ void rpoplpushHandlePush(client *c, robj *dstkey, robj *dstobj, robj *value) {
void rpoplpushCommand(client *c) {
robj *sobj, *value;
if ((sobj = lookupKeyWriteOrReply(c,c->argv[1],shared.null[c->resp]))
if ((sobj = lookupKeyWriteOrReply(c,c->argv[1],NULL,shared.null[c->resp]))
== NULL || checkType(c,sobj,OBJ_LIST)) return;
if (listTypeLength(sobj) == 0) {
......@@ -572,7 +572,7 @@ void rpoplpushCommand(client *c) {
* versions of Redis delete keys of empty lists. */
addReplyNull(c);
} else {
robj *dobj = lookupKeyWrite(c->db,c->argv[2]);
robj *dobj = lookupKeyWrite(c->db,c->argv[2],NULL);
robj *touchedkey = c->argv[1];
if (dobj && checkType(c,dobj,OBJ_LIST)) return;
......@@ -649,7 +649,7 @@ int serveClientBlockedOnList(client *receiver, robj *key, robj *dstkey, redisDb
} else {
/* BRPOPLPUSH */
robj *dstobj =
lookupKeyWrite(receiver->db,dstkey);
lookupKeyWrite(receiver->db,dstkey,NULL);
if (!(dstobj &&
checkType(receiver,dstobj,OBJ_LIST)))
{
......@@ -692,7 +692,7 @@ void blockingPopGenericCommand(client *c, int where) {
!= C_OK) return;
for (j = 1; j < c->argc-1; j++) {
o = lookupKeyWrite(c->db,c->argv[j]);
o = lookupKeyWrite(c->db,c->argv[j],NULL);
if (o != NULL) {
if (o->type != OBJ_LIST) {
addReply(c,shared.wrongtypeerr);
......@@ -753,7 +753,7 @@ void brpoplpushCommand(client *c) {
if (getTimeoutFromObjectOrReply(c,c->argv[3],&timeout,UNIT_SECONDS)
!= C_OK) return;
robj *key = lookupKeyWrite(c->db, c->argv[1]);
robj *key = lookupKeyWrite(c->db, c->argv[1], NULL);
if (key == NULL) {
if (c->flags & CLIENT_MULTI) {
......
......@@ -265,7 +265,7 @@ void saddCommand(client *c) {
robj *set;
int j, added = 0;
set = lookupKeyWrite(c->db,c->argv[1]);
set = lookupKeyWrite(c->db,c->argv[1],NULL);
if (set == NULL) {
set = setTypeCreate(c->argv[2]->ptr);
dbAdd(c->db,c->argv[1],set);
......@@ -291,7 +291,7 @@ void sremCommand(client *c) {
robj *set;
int j, deleted = 0, keyremoved = 0;
if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
if ((set = lookupKeyWriteOrReply(c,c->argv[1],NULL,shared.czero)) == NULL ||
checkType(c,set,OBJ_SET)) return;
for (j = 2; j < c->argc; j++) {
......@@ -317,8 +317,8 @@ void sremCommand(client *c) {
void smoveCommand(client *c) {
robj *srcset, *dstset, *ele;
srcset = lookupKeyWrite(c->db,c->argv[1]);
dstset = lookupKeyWrite(c->db,c->argv[2]);
srcset = lookupKeyWrite(c->db,c->argv[1],NULL);
dstset = lookupKeyWrite(c->db,c->argv[2],NULL);
ele = c->argv[3];
/* If the source key does not exist return 0 */
......@@ -373,7 +373,7 @@ void smoveCommand(client *c) {
void sismemberCommand(client *c) {
robj *set;
if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
if ((set = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.czero)) == NULL ||
checkType(c,set,OBJ_SET)) return;
if (setTypeIsMember(set,c->argv[2]->ptr))
......@@ -385,7 +385,7 @@ void sismemberCommand(client *c) {
void scardCommand(client *c) {
robj *o;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
if ((o = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.czero)) == NULL ||
checkType(c,o,OBJ_SET)) return;
addReplyLongLong(c,setTypeSize(o));
......@@ -415,7 +415,7 @@ void spopWithCountCommand(client *c) {
/* Make sure a key with the name inputted exists, and that it's type is
* indeed a set. Otherwise, return nil */
if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.null[c->resp]))
if ((set = lookupKeyWriteOrReply(c,c->argv[1],NULL,shared.null[c->resp]))
== NULL || checkType(c,set,OBJ_SET)) return;
/* If count is zero, serve an empty multibulk ASAP to avoid special
......@@ -566,7 +566,7 @@ void spopCommand(client *c) {
/* Make sure a key with the name inputted exists, and that it's type is
* indeed a set */
if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.null[c->resp]))
if ((set = lookupKeyWriteOrReply(c,c->argv[1],NULL,shared.null[c->resp]))
== NULL || checkType(c,set,OBJ_SET)) return;
/* Get a random element from the set */
......@@ -632,7 +632,7 @@ void srandmemberWithCountCommand(client *c) {
uniq = 0;
}
if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp]))
if ((set = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.null[c->resp]))
== NULL || checkType(c,set,OBJ_SET)) return;
size = setTypeSize(set);
......@@ -760,7 +760,7 @@ void srandmemberCommand(client *c) {
return;
}
if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp]))
if ((set = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.null[c->resp]))
== NULL || checkType(c,set,OBJ_SET)) return;
encoding = setTypeRandomElement(set,&ele,&llele);
......@@ -802,8 +802,8 @@ void sinterGenericCommand(client *c, robj **setkeys,
for (j = 0; j < setnum; j++) {
robj *setobj = dstkey ?
lookupKeyWrite(c->db,setkeys[j]) :
lookupKeyRead(c->db,setkeys[j]);
lookupKeyWrite(c->db,setkeys[j],NULL) :
lookupKeyRead(c->db,setkeys[j],NULL);
if (!setobj) {
zfree(sets);
if (dstkey) {
......@@ -939,8 +939,8 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
for (j = 0; j < setnum; j++) {
robj *setobj = dstkey ?
lookupKeyWrite(c->db,setkeys[j]) :
lookupKeyRead(c->db,setkeys[j]);
lookupKeyWrite(c->db,setkeys[j],NULL) :
lookupKeyRead(c->db,setkeys[j],NULL);
if (!setobj) {
sets[j] = NULL;
continue;
......@@ -1110,7 +1110,7 @@ void sscanCommand(client *c) {
unsigned long cursor;
if (parseScanCursorOrReply(c,c->argv[2],&cursor) == C_ERR) return;
if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.emptyscan)) == NULL ||
checkType(c,set,OBJ_SET)) return;
if ((set = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.emptyscan))
== NULL || checkType(c,set,OBJ_SET)) return;
scanGenericCommand(c,set,cursor);
}
......@@ -1057,7 +1057,7 @@ size_t streamReplyWithRangeFromConsumerPEL(client *c, stream *s, streamID *start
/* Look the stream at 'key' and return the corresponding stream object.
* The function creates a key setting it to an empty stream if needed. */
robj *streamTypeLookupWriteOrCreate(client *c, robj *key) {
robj *o = lookupKeyWrite(c->db,key);
robj *o = lookupKeyWrite(c->db,key,NULL);
if (o == NULL) {
o = createStreamObject();
dbAdd(c->db,key,o);
......@@ -1287,8 +1287,8 @@ void xrangeGenericCommand(client *c, int rev) {
}
/* Return the specified range to the user. */
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptyarray)) == NULL ||
checkType(c,o,OBJ_STREAM)) return;
if ((o = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.emptyarray))
== NULL || checkType(c,o,OBJ_STREAM)) return;
s = o->ptr;
......@@ -1313,7 +1313,7 @@ void xrevrangeCommand(client *c) {
/* XLEN */
void xlenCommand(client *c) {
robj *o;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL
if ((o = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.czero)) == NULL
|| checkType(c,o,OBJ_STREAM)) return;
stream *s = o->ptr;
addReplyLongLong(c,s->length);
......@@ -1411,7 +1411,7 @@ void xreadCommand(client *c) {
* starting from now. */
int id_idx = i - streams_arg - streams_count;
robj *key = c->argv[i-streams_count];
robj *o = lookupKeyRead(c->db,key);
robj *o = lookupKeyRead(c->db,key,NULL);
if (o && checkType(c,o,OBJ_STREAM)) goto cleanup;
streamCG *group = NULL;
......@@ -1469,7 +1469,7 @@ void xreadCommand(client *c) {
size_t arraylen = 0;
void *arraylen_ptr = NULL;
for (int i = 0; i < streams_count; i++) {
robj *o = lookupKeyRead(c->db,c->argv[streams_arg+i]);
robj *o = lookupKeyRead(c->db,c->argv[streams_arg+i],NULL);
if (o == NULL) continue;
stream *s = o->ptr;
streamID *gt = ids+i; /* ID must be greater than this. */
......@@ -1736,7 +1736,7 @@ NULL
/* Everything but the "HELP" option requires a key and group name. */
if (c->argc >= 4) {
o = lookupKeyWrite(c->db,c->argv[2]);
o = lookupKeyWrite(c->db,c->argv[2],NULL);
if (o) {
if (checkType(c,o,OBJ_STREAM)) return;
s = o->ptr;
......@@ -1840,7 +1840,7 @@ NULL
*
* Set the internal "last ID" of a stream. */
void xsetidCommand(client *c) {
robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr);
robj *o = lookupKeyWriteOrReply(c,c->argv[1],NULL,shared.nokeyerr);
if (o == NULL || checkType(c,o,OBJ_STREAM)) return;
stream *s = o->ptr;
......@@ -1881,7 +1881,7 @@ void xsetidCommand(client *c) {
*/
void xackCommand(client *c) {
streamCG *group = NULL;
robj *o = lookupKeyRead(c->db,c->argv[1]);
robj *o = lookupKeyRead(c->db,c->argv[1],NULL);
if (o) {
if (checkType(c,o,OBJ_STREAM)) return; /* Type error. */
group = streamLookupCG(o->ptr,c->argv[2]->ptr);
......@@ -1952,7 +1952,7 @@ void xpendingCommand(client *c) {
}
/* Lookup the key and the group inside the stream. */
robj *o = lookupKeyRead(c->db,c->argv[1]);
robj *o = lookupKeyRead(c->db,c->argv[1],NULL);
streamCG *group;
if (o && checkType(c,o,OBJ_STREAM)) return;
......@@ -2131,7 +2131,7 @@ void xpendingCommand(client *c) {
* what messages it is now in charge of. */
void xclaimCommand(client *c) {
streamCG *group = NULL;
robj *o = lookupKeyRead(c->db,c->argv[1]);
robj *o = lookupKeyRead(c->db,c->argv[1],NULL);
long long minidle; /* Minimum idle time argument. */
long long retrycount = -1; /* -1 means RETRYCOUNT option not given. */
mstime_t deliverytime = -1; /* -1 means IDLE/TIME options not given. */
......@@ -2323,7 +2323,7 @@ void xclaimCommand(client *c) {
void xdelCommand(client *c) {
robj *o;
if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL
if ((o = lookupKeyWriteOrReply(c,c->argv[1],NULL,shared.czero)) == NULL
|| checkType(c,o,OBJ_STREAM)) return;
stream *s = o->ptr;
......@@ -2368,7 +2368,7 @@ void xtrimCommand(client *c) {
/* If the key does not exist, we are ok returning zero, that is, the
* number of elements removed from the stream. */
if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL
if ((o = lookupKeyWriteOrReply(c,c->argv[1],NULL,shared.czero)) == NULL
|| checkType(c,o,OBJ_STREAM)) return;
stream *s = o->ptr;
......@@ -2460,7 +2460,7 @@ NULL
key = c->argv[2];
/* Lookup the key now, this is common for all the subcommands but HELP. */
robj *o = lookupKeyWriteOrReply(c,key,shared.nokeyerr);
robj *o = lookupKeyWriteOrReply(c,key,NULL,shared.nokeyerr);
if (o == NULL || checkType(c,o,OBJ_STREAM)) return;
s = o->ptr;
......
......@@ -64,7 +64,7 @@ static int checkStringLength(client *c, long long size) {
#define OBJ_SET_EX (1<<2) /* Set if time in seconds is given */
#define OBJ_SET_PX (1<<3) /* Set if time in ms in given */
void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, int unit, robj *ok_reply, robj *abort_reply) {
void setGenericCommand(client *c, int flags, robj *keyname, robj *val, robj *expire, int unit, robj *ok_reply, robj *abort_reply) {
long long milliseconds = 0; /* initialized to avoid any harmness warning */
if (expire) {
......@@ -77,18 +77,18 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire,
if (unit == UNIT_SECONDS) milliseconds *= 1000;
}
if ((flags & OBJ_SET_NX && lookupKeyWrite(c->db,key) != NULL) ||
(flags & OBJ_SET_XX && lookupKeyWrite(c->db,key) == NULL))
if ((flags & OBJ_SET_NX && lookupKeyWrite(c->db,keyname,NULL) != NULL) ||
(flags & OBJ_SET_XX && lookupKeyWrite(c->db,keyname,NULL) == NULL))
{
addReply(c, abort_reply ? abort_reply : shared.null[c->resp]);
return;
}
setKey(c->db,key,val);
rkey *key = setKey(c->db,keyname,val);
server.dirty++;
if (expire) setExpire(c,c->db,key,mstime()+milliseconds);
notifyKeyspaceEvent(NOTIFY_STRING,"set",key,c->db->id);
notifyKeyspaceEvent(NOTIFY_STRING,"set",keyname,c->db->id);
if (expire) notifyKeyspaceEvent(NOTIFY_GENERIC,
"expire",key,c->db->id);
"expire",keyname,c->db->id);
addReply(c, ok_reply ? ok_reply : shared.ok);
}
......@@ -157,8 +157,8 @@ void psetexCommand(client *c) {
int getGenericCommand(client *c) {
robj *o;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp])) == NULL)
return C_OK;
if ((o = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.null[c->resp]))
== NULL) return C_OK;
if (o->type != OBJ_STRING) {
addReply(c,shared.wrongtypeerr);
......@@ -194,7 +194,7 @@ void setrangeCommand(client *c) {
return;
}
o = lookupKeyWrite(c->db,c->argv[1]);
o = lookupKeyWrite(c->db,c->argv[1],NULL);
if (o == NULL) {
/* Return 0 when setting nothing on a non-existing string */
if (sdslen(value) == 0) {
......@@ -251,8 +251,8 @@ void getrangeCommand(client *c) {
return;
if (getLongLongFromObjectOrReply(c,c->argv[3],&end,NULL) != C_OK)
return;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptybulk)) == NULL ||
checkType(c,o,OBJ_STRING)) return;
if ((o = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.emptybulk)) == NULL
|| checkType(c,o,OBJ_STRING)) return;
if (o->encoding == OBJ_ENCODING_INT) {
str = llbuf;
......@@ -287,7 +287,7 @@ void mgetCommand(client *c) {
addReplyArrayLen(c,c->argc-1);
for (j = 1; j < c->argc; j++) {
robj *o = lookupKeyRead(c->db,c->argv[j]);
robj *o = lookupKeyRead(c->db,c->argv[j],NULL);
if (o == NULL) {
addReplyNull(c);
} else {
......@@ -312,7 +312,7 @@ void msetGenericCommand(client *c, int nx) {
* set anything if at least one key alerady exists. */
if (nx) {
for (j = 1; j < c->argc; j += 2) {
if (lookupKeyWrite(c->db,c->argv[j]) != NULL) {
if (lookupKeyWrite(c->db,c->argv[j],NULL) != NULL) {
addReply(c, shared.czero);
return;
}
......@@ -340,7 +340,7 @@ void incrDecrCommand(client *c, long long incr) {
long long value, oldvalue;
robj *o, *new;
o = lookupKeyWrite(c->db,c->argv[1]);
o = lookupKeyWrite(c->db,c->argv[1],NULL);
if (o != NULL && checkType(c,o,OBJ_STRING)) return;
if (getLongLongFromObjectOrReply(c,o,&value,NULL) != C_OK) return;
......@@ -400,7 +400,7 @@ void incrbyfloatCommand(client *c) {
long double incr, value;
robj *o, *new, *aux;
o = lookupKeyWrite(c->db,c->argv[1]);
o = lookupKeyWrite(c->db,c->argv[1],NULL);
if (o != NULL && checkType(c,o,OBJ_STRING)) return;
if (getLongDoubleFromObjectOrReply(c,o,&value,NULL) != C_OK ||
getLongDoubleFromObjectOrReply(c,c->argv[2],&incr,NULL) != C_OK)
......@@ -434,7 +434,7 @@ void appendCommand(client *c) {
size_t totlen;
robj *o, *append;
o = lookupKeyWrite(c->db,c->argv[1]);
o = lookupKeyWrite(c->db,c->argv[1],NULL);
if (o == NULL) {
/* Create the key */
c->argv[2] = tryObjectEncoding(c->argv[2]);
......@@ -465,7 +465,7 @@ void appendCommand(client *c) {
void strlenCommand(client *c) {
robj *o;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
if ((o = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.czero)) == NULL ||
checkType(c,o,OBJ_STRING)) return;
addReplyLongLong(c,stringObjectLen(o));
}
......@@ -1597,7 +1597,7 @@ void zaddGenericCommand(client *c, int flags) {
}
/* Lookup the key and create the sorted set if does not exist. */
zobj = lookupKeyWrite(c->db,key);
zobj = lookupKeyWrite(c->db,key,NULL);
if (zobj == NULL) {
if (xx) goto reply_to_client; /* No key + XX option: nothing to do. */
if (server.zset_max_ziplist_entries == 0 ||
......@@ -1665,7 +1665,7 @@ void zremCommand(client *c) {
robj *zobj;
int deleted = 0, keyremoved = 0, j;
if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL ||
if ((zobj = lookupKeyWriteOrReply(c,key,NULL,shared.czero)) == NULL ||
checkType(c,zobj,OBJ_ZSET)) return;
for (j = 2; j < c->argc; j++) {
......@@ -1718,7 +1718,7 @@ void zremrangeGenericCommand(client *c, int rangetype) {
}
/* Step 2: Lookup & range sanity checks if needed. */
if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL ||
if ((zobj = lookupKeyWriteOrReply(c,key,NULL,shared.czero)) == NULL ||
checkType(c,zobj,OBJ_ZSET)) goto cleanup;
if (rangetype == ZRANGE_RANK) {
......@@ -2165,7 +2165,8 @@ uint64_t dictSdsHash(const void *key);
int dictSdsKeyCompare(void *privdata, const void *key1, const void *key2);
dictType setAccumulatorDictType = {
dictSdsHash, /* hash function */
dictSdsHash, /* lookup hash function */
dictSdsHash, /* stored hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCompare, /* key compare */
......@@ -2205,7 +2206,7 @@ void zunionInterGenericCommand(client *c, robj *dstkey, int op) {
/* read keys to be used for input */
src = zcalloc(sizeof(zsetopsrc) * setnum);
for (i = 0, j = 3; i < setnum; i++, j++) {
robj *obj = lookupKeyWrite(c->db,c->argv[j]);
robj *obj = lookupKeyWrite(c->db,c->argv[j],NULL);
if (obj != NULL) {
if (obj->type != OBJ_ZSET && obj->type != OBJ_SET) {
zfree(src);
......@@ -2427,7 +2428,7 @@ void zrangeGenericCommand(client *c, int reverse) {
return;
}
if ((zobj = lookupKeyReadOrReply(c,key,shared.null[c->resp])) == NULL
if ((zobj = lookupKeyReadOrReply(c,key,NULL,shared.null[c->resp])) == NULL
|| checkType(c,zobj,OBJ_ZSET)) return;
/* Sanitize indexes. */
......@@ -2575,8 +2576,8 @@ void genericZrangebyscoreCommand(client *c, int reverse) {
}
/* Ok, lookup the key and get the range */
if ((zobj = lookupKeyReadOrReply(c,key,shared.null[c->resp])) == NULL ||
checkType(c,zobj,OBJ_ZSET)) return;
if ((zobj = lookupKeyReadOrReply(c,key,NULL,shared.null[c->resp]))
== NULL || checkType(c,zobj,OBJ_ZSET)) return;
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
unsigned char *zl = zobj->ptr;
......@@ -2730,7 +2731,7 @@ void zcountCommand(client *c) {
}
/* Lookup the sorted set */
if ((zobj = lookupKeyReadOrReply(c, key, shared.czero)) == NULL ||
if ((zobj = lookupKeyReadOrReply(c, key, NULL, shared.czero)) == NULL ||
checkType(c, zobj, OBJ_ZSET)) return;
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
......@@ -2807,7 +2808,7 @@ void zlexcountCommand(client *c) {
}
/* Lookup the sorted set */
if ((zobj = lookupKeyReadOrReply(c, key, shared.czero)) == NULL ||
if ((zobj = lookupKeyReadOrReply(c, key, NULL, shared.czero)) == NULL ||
checkType(c, zobj, OBJ_ZSET))
{
zslFreeLexRange(&range);
......@@ -2920,8 +2921,8 @@ void genericZrangebylexCommand(client *c, int reverse) {
}
/* Ok, lookup the key and get the range */
if ((zobj = lookupKeyReadOrReply(c,key,shared.null[c->resp])) == NULL ||
checkType(c,zobj,OBJ_ZSET))
if ((zobj = lookupKeyReadOrReply(c,key,NULL,shared.null[c->resp]))
== NULL || checkType(c,zobj,OBJ_ZSET))
{
zslFreeLexRange(&range);
return;
......@@ -3065,7 +3066,7 @@ void zcardCommand(client *c) {
robj *key = c->argv[1];
robj *zobj;
if ((zobj = lookupKeyReadOrReply(c,key,shared.czero)) == NULL ||
if ((zobj = lookupKeyReadOrReply(c,key,NULL,shared.czero)) == NULL ||
checkType(c,zobj,OBJ_ZSET)) return;
addReplyLongLong(c,zsetLength(zobj));
......@@ -3076,8 +3077,8 @@ void zscoreCommand(client *c) {
robj *zobj;
double score;
if ((zobj = lookupKeyReadOrReply(c,key,shared.null[c->resp])) == NULL ||
checkType(c,zobj,OBJ_ZSET)) return;
if ((zobj = lookupKeyReadOrReply(c,key,NULL,shared.null[c->resp])) == NULL
|| checkType(c,zobj,OBJ_ZSET)) return;
if (zsetScore(zobj,c->argv[2]->ptr,&score) == C_ERR) {
addReplyNull(c);
......@@ -3092,8 +3093,8 @@ void zrankGenericCommand(client *c, int reverse) {
robj *zobj;
long rank;
if ((zobj = lookupKeyReadOrReply(c,key,shared.null[c->resp])) == NULL ||
checkType(c,zobj,OBJ_ZSET)) return;
if ((zobj = lookupKeyReadOrReply(c,key,NULL,shared.null[c->resp]))
== NULL || checkType(c,zobj,OBJ_ZSET)) return;
serverAssertWithInfo(c,ele,sdsEncodedObject(ele));
rank = zsetRank(zobj,ele->ptr,reverse);
......@@ -3117,8 +3118,8 @@ void zscanCommand(client *c) {
unsigned long cursor;
if (parseScanCursorOrReply(c,c->argv[2],&cursor) == C_ERR) return;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptyscan)) == NULL ||
checkType(c,o,OBJ_ZSET)) return;
if ((o = lookupKeyReadOrReply(c,c->argv[1],NULL,shared.emptyscan))
== NULL || checkType(c,o,OBJ_ZSET)) return;
scanGenericCommand(c,o,cursor);
}
......@@ -3153,7 +3154,7 @@ void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey
idx = 0;
while (idx < keyc) {
key = keyv[idx++];
zobj = lookupKeyWrite(c->db,key);
zobj = lookupKeyWrite(c->db,key,NULL);
if (!zobj) continue;
if (checkType(c,zobj,OBJ_ZSET)) return;
break;
......@@ -3265,7 +3266,7 @@ void blockingGenericZpopCommand(client *c, int where) {
!= C_OK) return;
for (j = 1; j < c->argc-1; j++) {
o = lookupKeyWrite(c->db,c->argv[j]);
o = lookupKeyWrite(c->db,c->argv[j],NULL);
if (o != NULL) {
if (o->type != OBJ_ZSET) {
addReply(c,shared.wrongtypeerr);
......
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