#include "lzf.h" /* LZF compression library */
#include "pqsort.h" /* Partial qsort for SORT+LIMIT */
#include "zipmap.h" /* Compact dictionary-alike data structure */
+#include "ziplist.h" /* Compact list data structure */
#include "sha1.h" /* SHA1 is used for DEBUG DIGEST */
#include "release.h" /* Release and/or git repository information */
#define REDIS_STATIC_ARGS 8
#define REDIS_DEFAULT_DBNUM 16
#define REDIS_CONFIGLINE_MAX 1024
-#define REDIS_OBJFREELIST_MAX 0 /* Max number of objects to cache */
+#define REDIS_OBJFREELIST_MAX 1000000 /* Max number of objects to cache */
#define REDIS_MAX_SYNC_TIME 60 /* Slave can't take more to sync */
#define REDIS_EXPIRELOOKUPS_PER_CRON 10 /* lookup 10 expires per loop */
#define REDIS_MAX_WRITE_PER_EVENT (1024*64)
/* Objects encoding. Some kind of objects like Strings and Hashes can be
* internally represented in multiple ways. The 'encoding' field of the object
* is set to one of this fields for this object. */
-#define REDIS_ENCODING_RAW 0 /* Raw representation */
-#define REDIS_ENCODING_INT 1 /* Encoded as integer */
-#define REDIS_ENCODING_ZIPMAP 2 /* Encoded as zipmap */
-#define REDIS_ENCODING_HT 3 /* Encoded as an hash table */
+#define REDIS_ENCODING_RAW 0 /* Raw representation */
+#define REDIS_ENCODING_INT 1 /* Encoded as integer */
+#define REDIS_ENCODING_HT 2 /* Encoded as hash table */
+#define REDIS_ENCODING_ZIPMAP 3 /* Encoded as zipmap */
+#define REDIS_ENCODING_LIST 4 /* Encoded as zipmap */
+#define REDIS_ENCODING_ZIPLIST 5 /* Encoded as ziplist */
static char* strencoding[] = {
- "raw", "int", "zipmap", "hashtable"
+ "raw", "int", "hashtable", "zipmap", "list", "ziplist"
};
/* Object types only used for dumping to disk */
#define APPENDFSYNC_ALWAYS 1
#define APPENDFSYNC_EVERYSEC 2
-/* Hashes related defaults */
+/* Zip structure related defaults */
#define REDIS_HASH_MAX_ZIPMAP_ENTRIES 64
#define REDIS_HASH_MAX_ZIPMAP_VALUE 512
+#define REDIS_LIST_MAX_ZIPLIST_ENTRIES 1024
+#define REDIS_LIST_MAX_ZIPLIST_VALUE 32
/* We can print the stacktrace, so our assert is defined this way: */
#define redisAssert(_e) ((_e)?(void)0 : (_redisAssert(#_e,__FILE__,__LINE__),_exit(1)))
* as redisObject structures.
*
* This is useful as we don't know if a value object is or not on disk, but we
- * are always free of accessing obj->storage to check this. For vmPointer
+ * are always able to read obj->storage to check this. For vmPointer
* structures "type" is set to REDIS_VMPOINTER (even if without this field
* is still possible to check the kind of object from the value of 'storage').*/
typedef struct vmPointer {
off_t vm_page_size;
off_t vm_pages;
unsigned long long vm_max_memory;
- /* Hashes config */
+ /* Zip structure config */
size_t hash_max_zipmap_entries;
size_t hash_max_zipmap_value;
+ size_t list_max_ziplist_entries;
+ size_t list_max_ziplist_value;
/* Virtual memory state */
FILE *vm_fp;
int vm_fd;
static int removeExpire(redisDb *db, robj *key);
static int expireIfNeeded(redisDb *db, robj *key);
static int deleteIfVolatile(redisDb *db, robj *key);
-static int deleteIfSwapped(redisDb *db, robj *key);
-static int deleteKey(redisDb *db, robj *key);
+static int dbDelete(redisDb *db, robj *key);
static time_t getExpire(redisDb *db, robj *key);
static int setExpire(redisDb *db, robj *key, time_t when);
static void updateSlavesWaitingBgsave(int bgsaveerr);
static void call(redisClient *c, struct redisCommand *cmd);
static void resetClient(redisClient *c);
static void convertToRealHash(robj *o);
+static void listTypeConvert(robj *o, int enc);
static int pubsubUnsubscribeAllChannels(redisClient *c, int notify);
static int pubsubUnsubscribeAllPatterns(redisClient *c, int notify);
static void freePubsubPattern(void *p);
listRelease((list*)val);
}
-static int sdsDictKeyCompare(void *privdata, const void *key1,
+static int dictSdsKeyCompare(void *privdata, const void *key1,
const void *key2)
{
int l1,l2;
decrRefCount(val);
}
+static void dictSdsDestructor(void *privdata, void *val)
+{
+ DICT_NOTUSED(privdata);
+
+ sdsfree(val);
+}
+
static int dictObjKeyCompare(void *privdata, const void *key1,
const void *key2)
{
const robj *o1 = key1, *o2 = key2;
- return sdsDictKeyCompare(privdata,o1->ptr,o2->ptr);
+ return dictSdsKeyCompare(privdata,o1->ptr,o2->ptr);
}
static unsigned int dictObjHash(const void *key) {
return dictGenHashFunction(o->ptr, sdslen((sds)o->ptr));
}
+static unsigned int dictSdsHash(const void *key) {
+ return dictGenHashFunction((unsigned char*)key, sdslen((char*)key));
+}
+
static int dictEncObjKeyCompare(void *privdata, const void *key1,
const void *key2)
{
o1 = getDecodedObject(o1);
o2 = getDecodedObject(o2);
- cmp = sdsDictKeyCompare(privdata,o1->ptr,o2->ptr);
+ cmp = dictSdsKeyCompare(privdata,o1->ptr,o2->ptr);
decrRefCount(o1);
decrRefCount(o2);
return cmp;
}
}
-/* Sets type and expires */
+/* Sets type */
static dictType setDictType = {
dictEncObjHash, /* hash function */
NULL, /* key dup */
dictVanillaFree /* val destructor of malloc(sizeof(double)) */
};
-/* Db->dict */
+/* Db->dict, keys are sds strings, vals are Redis objects. */
static dictType dbDictType = {
- dictObjHash, /* hash function */
+ dictSdsHash, /* hash function */
NULL, /* key dup */
NULL, /* val dup */
- dictObjKeyCompare, /* key compare */
- dictRedisObjectDestructor, /* key destructor */
+ dictSdsKeyCompare, /* key compare */
+ dictSdsDestructor, /* key destructor */
dictRedisObjectDestructor /* val destructor */
};
/* Db->expires */
static dictType keyptrDictType = {
- dictObjHash, /* hash function */
+ dictSdsHash, /* hash function */
NULL, /* key dup */
NULL, /* val dup */
- dictObjKeyCompare, /* key compare */
- dictRedisObjectDestructor, /* key destructor */
+ dictSdsKeyCompare, /* key compare */
+ dictSdsDestructor, /* key destructor */
NULL /* val destructor */
};
if ((de = dictGetRandomKey(db->expires)) == NULL) break;
t = (time_t) dictGetEntryVal(de);
if (now > t) {
- deleteKey(db,dictGetEntryKey(de));
+ sds key = dictGetEntryKey(de);
+ robj *keyobj = createStringObject(key,sdslen(key));
+
+ dbDelete(db,keyobj);
+ decrRefCount(keyobj);
expired++;
server.stat_expiredkeys++;
}
server.vm_blocked_clients = 0;
server.hash_max_zipmap_entries = REDIS_HASH_MAX_ZIPMAP_ENTRIES;
server.hash_max_zipmap_value = REDIS_HASH_MAX_ZIPMAP_VALUE;
+ server.list_max_ziplist_entries = REDIS_LIST_MAX_ZIPLIST_ENTRIES;
+ server.list_max_ziplist_value = REDIS_LIST_MAX_ZIPLIST_VALUE;
server.shutdown_asap = 0;
resetServerSaveParams();
server.hash_max_zipmap_entries = memtoll(argv[1], NULL);
} else if (!strcasecmp(argv[0],"hash-max-zipmap-value") && argc == 2){
server.hash_max_zipmap_value = memtoll(argv[1], NULL);
+ } else if (!strcasecmp(argv[0],"list-max-ziplist-entries") && argc == 2){
+ server.list_max_ziplist_entries = memtoll(argv[1], NULL);
+ } else if (!strcasecmp(argv[0],"list-max-ziplist-value") && argc == 2){
+ server.list_max_ziplist_value = memtoll(argv[1], NULL);
} else {
err = "Bad directive or wrong number of arguments"; goto loaderr;
}
addReply(c,shared.crlf);
}
+static void addReplyBulkSds(redisClient *c, sds s) {
+ robj *o = createStringObject(s, sdslen(s));
+ addReplyBulk(c,o);
+ decrRefCount(o);
+}
+
/* In the CONFIG command we need to add vanilla C string as bulk replies */
static void addReplyBulkCString(redisClient *c, char *s) {
if (s == NULL) {
static robj *createListObject(void) {
list *l = listCreate();
-
+ robj *o = createObject(REDIS_LIST,l);
listSetFreeMethod(l,decrRefCount);
- return createObject(REDIS_LIST,l);
+ o->encoding = REDIS_ENCODING_LIST;
+ return o;
+}
+
+static robj *createZiplistObject(void) {
+ unsigned char *zl = ziplistNew();
+ robj *o = createObject(REDIS_LIST,zl);
+ o->encoding = REDIS_ENCODING_ZIPLIST;
+ return o;
}
static robj *createSetObject(void) {
}
static void freeListObject(robj *o) {
- listRelease((list*) o->ptr);
+ switch (o->encoding) {
+ case REDIS_ENCODING_LIST:
+ listRelease((list*) o->ptr);
+ break;
+ case REDIS_ENCODING_ZIPLIST:
+ zfree(o->ptr);
+ break;
+ default:
+ redisPanic("Unknown list encoding type");
+ }
}
static void freeSetObject(robj *o) {
}
}
-static robj *lookupKey(redisDb *db, robj *key) {
- dictEntry *de = dictFind(db->dict,key);
- if (de) {
- robj *key = dictGetEntryKey(de);
- robj *val = dictGetEntryVal(de);
-
- if (server.vm_enabled) {
- if (val->storage == REDIS_VM_MEMORY ||
- val->storage == REDIS_VM_SWAPPING)
- {
- /* If we were swapping the object out, cancel the operation */
- if (val->storage == REDIS_VM_SWAPPING)
- vmCancelThreadedIOJob(val);
- /* Update the access time of the key for the aging algorithm. */
- val->lru = server.lruclock;
- } else {
- int notify = (val->storage == REDIS_VM_LOADING);
-
- /* Our value was swapped on disk. Bring it at home. */
- redisAssert(val->type == REDIS_VMPOINTER);
- val = vmLoadObject(val);
- dictGetEntryVal(de) = val;
-
- /* Clients blocked by the VM subsystem may be waiting for
- * this key... */
- if (notify) handleClientsBlockedOnSwappedKey(db,key);
- }
- }
- return val;
- } else {
- return NULL;
- }
-}
-
-static robj *lookupKeyRead(redisDb *db, robj *key) {
- expireIfNeeded(db,key);
- return lookupKey(db,key);
-}
-
-static robj *lookupKeyWrite(redisDb *db, robj *key) {
- deleteIfVolatile(db,key);
- touchWatchedKey(db,key);
- return lookupKey(db,key);
-}
-
-static robj *lookupKeyReadOrReply(redisClient *c, robj *key, robj *reply) {
- robj *o = lookupKeyRead(c->db, key);
- if (!o) addReply(c,reply);
- return o;
-}
-
-static robj *lookupKeyWriteOrReply(redisClient *c, robj *key, robj *reply) {
- robj *o = lookupKeyWrite(c->db, key);
- if (!o) addReply(c,reply);
- return o;
-}
-
static int checkType(redisClient *c, robj *o, int type) {
if (o->type != type) {
addReply(c,shared.wrongtypeerr);
return 0;
}
-static int deleteKey(redisDb *db, robj *key) {
- int retval;
-
- /* We need to protect key from destruction: after the first dictDelete()
- * it may happen that 'key' is no longer valid if we don't increment
- * it's count. This may happen when we get the object reference directly
- * from the hash table with dictRandomKey() or dict iterators */
- incrRefCount(key);
- if (dictSize(db->expires)) dictDelete(db->expires,key);
- retval = dictDelete(db->dict,key);
- decrRefCount(key);
-
- return retval == DICT_OK;
-}
-
/* Check if the nul-terminated string 's' can be represented by a long
* (that is, is a number that fits into long without any other space or
* character before or after the digits).
return REDIS_OK;
}
+/* =========================== Keyspace access API ========================== */
+
+static robj *lookupKey(redisDb *db, robj *key) {
+ dictEntry *de = dictFind(db->dict,key->ptr);
+ if (de) {
+ robj *val = dictGetEntryVal(de);
+
+ if (server.vm_enabled) {
+ if (val->storage == REDIS_VM_MEMORY ||
+ val->storage == REDIS_VM_SWAPPING)
+ {
+ /* If we were swapping the object out, cancel the operation */
+ if (val->storage == REDIS_VM_SWAPPING)
+ vmCancelThreadedIOJob(val);
+ /* Update the access time for the aging algorithm. */
+ val->lru = server.lruclock;
+ } else {
+ int notify = (val->storage == REDIS_VM_LOADING);
+
+ /* Our value was swapped on disk. Bring it at home. */
+ redisAssert(val->type == REDIS_VMPOINTER);
+ val = vmLoadObject(val);
+ dictGetEntryVal(de) = val;
+
+ /* Clients blocked by the VM subsystem may be waiting for
+ * this key... */
+ if (notify) handleClientsBlockedOnSwappedKey(db,key);
+ }
+ }
+ return val;
+ } else {
+ return NULL;
+ }
+}
+
+static robj *lookupKeyRead(redisDb *db, robj *key) {
+ expireIfNeeded(db,key);
+ return lookupKey(db,key);
+}
+
+static robj *lookupKeyWrite(redisDb *db, robj *key) {
+ deleteIfVolatile(db,key);
+ touchWatchedKey(db,key);
+ return lookupKey(db,key);
+}
+
+static robj *lookupKeyReadOrReply(redisClient *c, robj *key, robj *reply) {
+ robj *o = lookupKeyRead(c->db, key);
+ if (!o) addReply(c,reply);
+ return o;
+}
+
+static robj *lookupKeyWriteOrReply(redisClient *c, robj *key, robj *reply) {
+ robj *o = lookupKeyWrite(c->db, key);
+ if (!o) addReply(c,reply);
+ return o;
+}
+
+/* Add the key to the DB. If the key already exists REDIS_ERR is returned,
+ * otherwise REDIS_OK is returned, and the caller should increment the
+ * refcount of 'val'. */
+static int dbAdd(redisDb *db, robj *key, robj *val) {
+ /* Perform a lookup before adding the key, as we need to copy the
+ * key value. */
+ if (dictFind(db->dict, key->ptr) != NULL) {
+ return REDIS_ERR;
+ } else {
+ sds copy = sdsdup(key->ptr);
+ dictAdd(db->dict, copy, val);
+ return REDIS_OK;
+ }
+}
+
+/* If the key does not exist, this is just like dbAdd(). Otherwise
+ * the value associated to the key is replaced with the new one.
+ *
+ * On update (key already existed) 0 is returned. Otherwise 1. */
+static int dbReplace(redisDb *db, robj *key, robj *val) {
+ if (dictFind(db->dict,key->ptr) == NULL) {
+ sds copy = sdsdup(key->ptr);
+ dictAdd(db->dict, copy, val);
+ return 1;
+ } else {
+ dictReplace(db->dict, key->ptr, val);
+ return 0;
+ }
+}
+
+static int dbExists(redisDb *db, robj *key) {
+ return dictFind(db->dict,key->ptr) != NULL;
+}
+
+/* Return a random key, in form of a Redis object.
+ * If there are no keys, NULL is returned.
+ *
+ * The function makes sure to return keys not already expired. */
+static robj *dbRandomKey(redisDb *db) {
+ struct dictEntry *de;
+
+ while(1) {
+ sds key;
+ robj *keyobj;
+
+ de = dictGetRandomKey(db->dict);
+ if (de == NULL) return NULL;
+
+ key = dictGetEntryKey(de);
+ keyobj = createStringObject(key,sdslen(key));
+ if (dictFind(db->expires,key)) {
+ if (expireIfNeeded(db,keyobj)) {
+ decrRefCount(keyobj);
+ continue; /* search for another key. This expired. */
+ }
+ }
+ return keyobj;
+ }
+}
+
+/* Delete a key, value, and associated expiration entry if any, from the DB */
+static int dbDelete(redisDb *db, robj *key) {
+ int retval;
+
+ if (dictSize(db->expires)) dictDelete(db->expires,key->ptr);
+ retval = dictDelete(db->dict,key->ptr);
+
+ return retval == DICT_OK;
+}
+
/*============================ RDB saving/loading =========================== */
static int rdbSaveType(FILE *fp, unsigned char type) {
return 0;
}
+/* Save a long long value as either an encoded string or a string. */
+static int rdbSaveLongLongAsStringObject(FILE *fp, long long value) {
+ unsigned char buf[32];
+ int enclen = rdbEncodeInteger(value,buf);
+ if (enclen > 0) {
+ if (fwrite(buf,enclen,1,fp) == 0) return -1;
+ } else {
+ /* Encode as string */
+ enclen = ll2string((char*)buf,32,value);
+ redisAssert(enclen < 32);
+ if (rdbSaveLen(fp,enclen) == -1) return -1;
+ if (fwrite(buf,enclen,1,fp) == 0) return -1;
+ }
+ return 0;
+}
+
/* Like rdbSaveStringObjectRaw() but handle encoded objects */
static int rdbSaveStringObject(FILE *fp, robj *obj) {
- int retval;
-
/* Avoid to decode the object, then encode it again, if the
* object is alrady integer encoded. */
if (obj->encoding == REDIS_ENCODING_INT) {
- long val = (long) obj->ptr;
- unsigned char buf[5];
- int enclen;
-
- if ((enclen = rdbEncodeInteger(val,buf)) > 0) {
- if (fwrite(buf,enclen,1,fp) == 0) return -1;
- return 0;
- }
- /* otherwise... fall throught and continue with the usual
- * code path. */
- }
-
- /* Avoid incr/decr ref count business when possible.
- * This plays well with copy-on-write given that we are probably
- * in a child process (BGSAVE). Also this makes sure key objects
- * of swapped objects are not incRefCount-ed (an assert does not allow
- * this in order to avoid bugs) */
- if (obj->encoding != REDIS_ENCODING_RAW) {
- obj = getDecodedObject(obj);
- retval = rdbSaveRawString(fp,obj->ptr,sdslen(obj->ptr));
- decrRefCount(obj);
+ return rdbSaveLongLongAsStringObject(fp,(long)obj->ptr);
} else {
- retval = rdbSaveRawString(fp,obj->ptr,sdslen(obj->ptr));
+ redisAssert(obj->encoding == REDIS_ENCODING_RAW);
+ return rdbSaveRawString(fp,obj->ptr,sdslen(obj->ptr));
}
- return retval;
}
/* Save a double value. Doubles are saved as strings prefixed by an unsigned
if (rdbSaveStringObject(fp,o) == -1) return -1;
} else if (o->type == REDIS_LIST) {
/* Save a list value */
- list *list = o->ptr;
- listIter li;
- listNode *ln;
-
- if (rdbSaveLen(fp,listLength(list)) == -1) return -1;
- listRewind(list,&li);
- while((ln = listNext(&li))) {
- robj *eleobj = listNodeValue(ln);
+ if (o->encoding == REDIS_ENCODING_ZIPLIST) {
+ unsigned char *p;
+ unsigned char *vstr;
+ unsigned int vlen;
+ long long vlong;
+
+ if (rdbSaveLen(fp,ziplistLen(o->ptr)) == -1) return -1;
+ p = ziplistIndex(o->ptr,0);
+ while(ziplistGet(p,&vstr,&vlen,&vlong)) {
+ if (vstr) {
+ if (rdbSaveRawString(fp,vstr,vlen) == -1)
+ return -1;
+ } else {
+ if (rdbSaveLongLongAsStringObject(fp,vlong) == -1)
+ return -1;
+ }
+ p = ziplistNext(o->ptr,p);
+ }
+ } else if (o->encoding == REDIS_ENCODING_LIST) {
+ list *list = o->ptr;
+ listIter li;
+ listNode *ln;
- if (rdbSaveStringObject(fp,eleobj) == -1) return -1;
+ if (rdbSaveLen(fp,listLength(list)) == -1) return -1;
+ listRewind(list,&li);
+ while((ln = listNext(&li))) {
+ robj *eleobj = listNodeValue(ln);
+ if (rdbSaveStringObject(fp,eleobj) == -1) return -1;
+ }
+ } else {
+ redisPanic("Unknown list encoding");
}
} else if (o->type == REDIS_SET) {
/* Save a set value */
/* Iterate this DB writing every entry */
while((de = dictNext(di)) != NULL) {
- robj *key = dictGetEntryKey(de);
- robj *o = dictGetEntryVal(de);
- time_t expiretime = getExpire(db,key);
+ sds keystr = dictGetEntryKey(de);
+ robj key, *o = dictGetEntryVal(de);
+ time_t expiretime;
+
+ initStaticStringObject(key,keystr);
+ expiretime = getExpire(db,&key);
/* Save the expire time */
if (expiretime != -1) {
o->storage == REDIS_VM_SWAPPING) {
/* Save type, key, value */
if (rdbSaveType(fp,o->type) == -1) goto werr;
- if (rdbSaveStringObject(fp,key) == -1) goto werr;
+ if (rdbSaveStringObject(fp,&key) == -1) goto werr;
if (rdbSaveObject(fp,o) == -1) goto werr;
} else {
/* REDIS_VM_SWAPPED or REDIS_VM_LOADING */
po = vmPreviewObject(o);
/* Save type, key, value */
if (rdbSaveType(fp,po->type) == -1) goto werr;
- if (rdbSaveStringObject(fp,key) == -1) goto werr;
+ if (rdbSaveStringObject(fp,&key) == -1) goto werr;
if (rdbSaveObject(fp,po) == -1) goto werr;
/* Remove the loaded object from memory */
decrRefCount(po);
/* Load a Redis object of the specified type from the specified file.
* On success a newly allocated object is returned, otherwise NULL. */
static robj *rdbLoadObject(int type, FILE *fp) {
- robj *o;
+ robj *o, *ele, *dec;
+ size_t len;
redisLog(REDIS_DEBUG,"LOADING OBJECT %d (at %d)\n",type,ftell(fp));
if (type == REDIS_STRING) {
/* Read string value */
if ((o = rdbLoadEncodedStringObject(fp)) == NULL) return NULL;
o = tryObjectEncoding(o);
- } else if (type == REDIS_LIST || type == REDIS_SET) {
- /* Read list/set value */
- uint32_t listlen;
+ } else if (type == REDIS_LIST) {
+ /* Read list value */
+ if ((len = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
+
+ /* Use a real list when there are too many entries */
+ if (len > server.list_max_ziplist_entries) {
+ o = createListObject();
+ } else {
+ o = createZiplistObject();
+ }
+
+ /* Load every single element of the list */
+ while(len--) {
+ if ((ele = rdbLoadEncodedStringObject(fp)) == NULL) return NULL;
- if ((listlen = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
- o = (type == REDIS_LIST) ? createListObject() : createSetObject();
+ /* If we are using a ziplist and the value is too big, convert
+ * the object to a real list. */
+ if (o->encoding == REDIS_ENCODING_ZIPLIST &&
+ ele->encoding == REDIS_ENCODING_RAW &&
+ sdslen(ele->ptr) > server.list_max_ziplist_value)
+ listTypeConvert(o,REDIS_ENCODING_LIST);
+
+ if (o->encoding == REDIS_ENCODING_ZIPLIST) {
+ dec = getDecodedObject(ele);
+ o->ptr = ziplistPush(o->ptr,dec->ptr,sdslen(dec->ptr),REDIS_TAIL);
+ decrRefCount(dec);
+ decrRefCount(ele);
+ } else {
+ ele = tryObjectEncoding(ele);
+ listAddNodeTail(o->ptr,ele);
+ incrRefCount(ele);
+ }
+ }
+ } else if (type == REDIS_SET) {
+ /* Read list/set value */
+ if ((len = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
+ o = createSetObject();
/* It's faster to expand the dict to the right size asap in order
* to avoid rehashing */
- if (type == REDIS_SET && listlen > DICT_HT_INITIAL_SIZE)
- dictExpand(o->ptr,listlen);
+ if (len > DICT_HT_INITIAL_SIZE)
+ dictExpand(o->ptr,len);
/* Load every single element of the list/set */
- while(listlen--) {
- robj *ele;
-
+ while(len--) {
if ((ele = rdbLoadEncodedStringObject(fp)) == NULL) return NULL;
ele = tryObjectEncoding(ele);
- if (type == REDIS_LIST) {
- listAddNodeTail((list*)o->ptr,ele);
- } else {
- dictAdd((dict*)o->ptr,ele,NULL);
- }
+ dictAdd((dict*)o->ptr,ele,NULL);
}
} else if (type == REDIS_ZSET) {
/* Read list/set value */
uint32_t dbid;
int type, retval, rdbver;
int swap_all_values = 0;
- dict *d = server.db[0].dict;
redisDb *db = server.db+0;
char buf[1024];
time_t expiretime, now = time(NULL);
- long long loadedkeys = 0;
fp = fopen(filename,"r");
if (!fp) return REDIS_ERR;
}
while(1) {
robj *key, *val;
+ int force_swapout;
expiretime = -1;
/* Read type. */
exit(1);
}
db = server.db+dbid;
- d = db->dict;
continue;
}
/* Read key */
continue;
}
/* Add the new object in the hash table */
- retval = dictAdd(d,key,val);
- if (retval == DICT_ERR) {
+ retval = dbAdd(db,key,val);
+ if (retval == REDIS_ERR) {
redisLog(REDIS_WARNING,"Loading DB, duplicated key (%s) found! Unrecoverable error, exiting now.", key->ptr);
exit(1);
}
- loadedkeys++;
/* Set the expire time if needed */
if (expiretime != -1) setExpire(db,key,expiretime);
* to random sampling, otherwise we may try to swap already
* swapped keys. */
if (swap_all_values) {
- dictEntry *de = dictFind(d,key);
+ dictEntry *de = dictFind(db->dict,key->ptr);
/* de may be NULL since the key already expired */
if (de) {
vmpointer *vp;
- key = dictGetEntryKey(de);
val = dictGetEntryVal(de);
if (val->refcount == 1 &&
(vp = vmSwapObjectBlocking(val)) != NULL)
dictGetEntryVal(de) = vp;
}
+ decrRefCount(key);
continue;
}
+ decrRefCount(key);
+
+ /* Flush data on disk once 32 MB of additional RAM are used... */
+ force_swapout = 0;
+ if ((zmalloc_used_memory() - server.vm_max_memory) > 1024*1024*32)
+ force_swapout = 1;
/* If we have still some hope of having some value fitting memory
* then we try random sampling. */
- if (!swap_all_values && server.vm_enabled && (loadedkeys % 5000) == 0) {
+ if (!swap_all_values && server.vm_enabled && force_swapout) {
while (zmalloc_used_memory() > server.vm_max_memory) {
if (vmSwapOneObjectBlocking() == REDIS_ERR) break;
}
touchWatchedKey(c->db,key);
if (nx) deleteIfVolatile(c->db,key);
- retval = dictAdd(c->db->dict,key,val);
- if (retval == DICT_ERR) {
+ retval = dbAdd(c->db,key,val);
+ if (retval == REDIS_ERR) {
if (!nx) {
- /* If the key is about a swapped value, we want a new key object
- * to overwrite the old. So we delete the old key in the database.
- * This will also make sure that swap pages about the old object
- * will be marked as free. */
- if (server.vm_enabled && deleteIfSwapped(c->db,key))
- incrRefCount(key);
- dictReplace(c->db->dict,key,val);
+ dbReplace(c->db,key,val);
incrRefCount(val);
} else {
addReply(c,shared.czero);
return;
}
} else {
- incrRefCount(key);
incrRefCount(val);
}
server.dirty++;
static void getsetCommand(redisClient *c) {
if (getGenericCommand(c) == REDIS_ERR) return;
- if (dictAdd(c->db->dict,c->argv[1],c->argv[2]) == DICT_ERR) {
- dictReplace(c->db->dict,c->argv[1],c->argv[2]);
- } else {
- incrRefCount(c->argv[1]);
- }
+ dbReplace(c->db,c->argv[1],c->argv[2]);
incrRefCount(c->argv[2]);
server.dirty++;
removeExpire(c->db,c->argv[1]);
}
for (j = 1; j < c->argc; j += 2) {
- int retval;
-
c->argv[j+1] = tryObjectEncoding(c->argv[j+1]);
- retval = dictAdd(c->db->dict,c->argv[j],c->argv[j+1]);
- if (retval == DICT_ERR) {
- dictReplace(c->db->dict,c->argv[j],c->argv[j+1]);
- incrRefCount(c->argv[j+1]);
- } else {
- incrRefCount(c->argv[j]);
- incrRefCount(c->argv[j+1]);
- }
+ dbReplace(c->db,c->argv[j],c->argv[j+1]);
+ incrRefCount(c->argv[j+1]);
removeExpire(c->db,c->argv[j]);
}
server.dirty += (c->argc-1)/2;
static void incrDecrCommand(redisClient *c, long long incr) {
long long value;
- int retval;
robj *o;
o = lookupKeyWrite(c->db,c->argv[1]);
value += incr;
o = createStringObjectFromLongLong(value);
- retval = dictAdd(c->db->dict,c->argv[1],o);
- if (retval == DICT_ERR) {
- dictReplace(c->db->dict,c->argv[1],o);
- removeExpire(c->db,c->argv[1]);
- } else {
- incrRefCount(c->argv[1]);
- }
+ dbReplace(c->db,c->argv[1],o);
server.dirty++;
addReply(c,shared.colon);
addReply(c,o);
o = lookupKeyWrite(c->db,c->argv[1]);
if (o == NULL) {
/* Create the key */
- retval = dictAdd(c->db->dict,c->argv[1],c->argv[2]);
- incrRefCount(c->argv[1]);
+ retval = dbAdd(c->db,c->argv[1],c->argv[2]);
incrRefCount(c->argv[2]);
totlen = stringObjectLen(c->argv[2]);
} else {
- dictEntry *de;
-
- de = dictFind(c->db->dict,c->argv[1]);
- assert(de != NULL);
-
- o = dictGetEntryVal(de);
if (o->type != REDIS_STRING) {
addReply(c,shared.wrongtypeerr);
return;
o = createStringObject(decoded->ptr, sdslen(decoded->ptr));
decrRefCount(decoded);
- dictReplace(c->db->dict,c->argv[1],o);
+ dbReplace(c->db,c->argv[1],o);
}
/* APPEND! */
if (c->argv[2]->encoding == REDIS_ENCODING_RAW) {
int deleted = 0, j;
for (j = 1; j < c->argc; j++) {
- if (deleteKey(c->db,c->argv[j])) {
+ if (dbDelete(c->db,c->argv[j])) {
touchWatchedKey(c->db,c->argv[j]);
server.dirty++;
deleted++;
static void existsCommand(redisClient *c) {
expireIfNeeded(c->db,c->argv[1]);
- if (dictFind(c->db->dict,c->argv[1])) {
+ if (dbExists(c->db,c->argv[1])) {
addReply(c, shared.cone);
} else {
addReply(c, shared.czero);
}
static void randomkeyCommand(redisClient *c) {
- dictEntry *de;
robj *key;
- while(1) {
- de = dictGetRandomKey(c->db->dict);
- if (!de || expireIfNeeded(c->db,dictGetEntryKey(de)) == 0) break;
- }
-
- if (de == NULL) {
+ if ((key = dbRandomKey(c->db)) == NULL) {
addReply(c,shared.nullbulk);
return;
}
- key = dictGetEntryKey(de);
- if (server.vm_enabled) {
- key = dupStringObject(key);
- addReplyBulk(c,key);
- decrRefCount(key);
- } else {
- addReplyBulk(c,key);
- }
+ addReplyBulk(c,key);
+ decrRefCount(key);
}
static void keysCommand(redisClient *c) {
addReply(c,lenobj);
decrRefCount(lenobj);
while((de = dictNext(di)) != NULL) {
- robj *keyobj = dictGetEntryKey(de);
+ sds key = dictGetEntryKey(de);
+ robj *keyobj;
- sds key = keyobj->ptr;
if ((pattern[0] == '*' && pattern[1] == '\0') ||
stringmatchlen(pattern,plen,key,sdslen(key),0)) {
+ keyobj = createStringObject(key,sdslen(key));
if (expireIfNeeded(c->db,keyobj) == 0) {
addReplyBulk(c,keyobj);
numkeys++;
}
+ decrRefCount(keyobj);
}
}
dictReleaseIterator(di);
return;
}
- if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr)) == NULL)
- return;
+ if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr)) == NULL)
+ return;
+
+ incrRefCount(o);
+ deleteIfVolatile(c->db,c->argv[2]);
+ if (dbAdd(c->db,c->argv[2],o) == REDIS_ERR) {
+ if (nx) {
+ decrRefCount(o);
+ addReply(c,shared.czero);
+ return;
+ }
+ dbReplace(c->db,c->argv[2],o);
+ }
+ dbDelete(c->db,c->argv[1]);
+ touchWatchedKey(c->db,c->argv[2]);
+ server.dirty++;
+ addReply(c,nx ? shared.cone : shared.ok);
+}
+
+static void renameCommand(redisClient *c) {
+ renameGenericCommand(c,0);
+}
+
+static void renamenxCommand(redisClient *c) {
+ renameGenericCommand(c,1);
+}
+
+static void moveCommand(redisClient *c) {
+ robj *o;
+ redisDb *src, *dst;
+ int srcid;
+
+ /* Obtain source and target DB pointers */
+ src = c->db;
+ srcid = c->db->id;
+ if (selectDb(c,atoi(c->argv[2]->ptr)) == REDIS_ERR) {
+ addReply(c,shared.outofrangeerr);
+ return;
+ }
+ dst = c->db;
+ selectDb(c,srcid); /* Back to the source DB */
+
+ /* If the user is moving using as target the same
+ * DB as the source DB it is probably an error. */
+ if (src == dst) {
+ addReply(c,shared.sameobjecterr);
+ return;
+ }
+
+ /* Check if the element exists and get a reference */
+ o = lookupKeyWrite(c->db,c->argv[1]);
+ if (!o) {
+ addReply(c,shared.czero);
+ return;
+ }
+
+ /* Try to add the element to the target DB */
+ deleteIfVolatile(dst,c->argv[1]);
+ if (dbAdd(dst,c->argv[1],o) == REDIS_ERR) {
+ addReply(c,shared.czero);
+ return;
+ }
+ incrRefCount(o);
+
+ /* OK! key moved, free the entry in the source DB */
+ dbDelete(src,c->argv[1]);
+ server.dirty++;
+ addReply(c,shared.cone);
+}
+
+/* =================================== Lists ================================ */
+
+
+/* Check the argument length to see if it requires us to convert the ziplist
+ * to a real list. Only check raw-encoded objects because integer encoded
+ * objects are never too long. */
+static void listTypeTryConversion(robj *subject, robj *value) {
+ if (subject->encoding != REDIS_ENCODING_ZIPLIST) return;
+ if (value->encoding == REDIS_ENCODING_RAW &&
+ sdslen(value->ptr) > server.list_max_ziplist_value)
+ listTypeConvert(subject,REDIS_ENCODING_LIST);
+}
+
+static void listTypePush(robj *subject, robj *value, int where) {
+ /* Check if we need to convert the ziplist */
+ listTypeTryConversion(subject,value);
+ if (subject->encoding == REDIS_ENCODING_ZIPLIST &&
+ ziplistLen(subject->ptr) > server.list_max_ziplist_entries)
+ listTypeConvert(subject,REDIS_ENCODING_LIST);
+
+ if (subject->encoding == REDIS_ENCODING_ZIPLIST) {
+ int pos = (where == REDIS_HEAD) ? ZIPLIST_HEAD : ZIPLIST_TAIL;
+ value = getDecodedObject(value);
+ subject->ptr = ziplistPush(subject->ptr,value->ptr,sdslen(value->ptr),pos);
+ decrRefCount(value);
+ } else if (subject->encoding == REDIS_ENCODING_LIST) {
+ if (where == REDIS_HEAD) {
+ listAddNodeHead(subject->ptr,value);
+ } else {
+ listAddNodeTail(subject->ptr,value);
+ }
+ incrRefCount(value);
+ } else {
+ redisPanic("Unknown list encoding");
+ }
+}
+
+static robj *listTypePop(robj *subject, int where) {
+ robj *value = NULL;
+ if (subject->encoding == REDIS_ENCODING_ZIPLIST) {
+ unsigned char *p;
+ unsigned char *vstr;
+ unsigned int vlen;
+ long long vlong;
+ int pos = (where == REDIS_HEAD) ? 0 : -1;
+ p = ziplistIndex(subject->ptr,pos);
+ if (ziplistGet(p,&vstr,&vlen,&vlong)) {
+ if (vstr) {
+ value = createStringObject((char*)vstr,vlen);
+ } else {
+ value = createStringObjectFromLongLong(vlong);
+ }
+ /* We only need to delete an element when it exists */
+ subject->ptr = ziplistDelete(subject->ptr,&p);
+ }
+ } else if (subject->encoding == REDIS_ENCODING_LIST) {
+ list *list = subject->ptr;
+ listNode *ln;
+ if (where == REDIS_HEAD) {
+ ln = listFirst(list);
+ } else {
+ ln = listLast(list);
+ }
+ if (ln != NULL) {
+ value = listNodeValue(ln);
+ incrRefCount(value);
+ listDelNode(list,ln);
+ }
+ } else {
+ redisPanic("Unknown list encoding");
+ }
+ return value;
+}
+
+static unsigned long listTypeLength(robj *subject) {
+ if (subject->encoding == REDIS_ENCODING_ZIPLIST) {
+ return ziplistLen(subject->ptr);
+ } else if (subject->encoding == REDIS_ENCODING_LIST) {
+ return listLength((list*)subject->ptr);
+ } else {
+ redisPanic("Unknown list encoding");
+ }
+}
+
+/* Structure to hold set iteration abstraction. */
+typedef struct {
+ robj *subject;
+ unsigned char encoding;
+ unsigned char direction; /* Iteration direction */
+ unsigned char *zi;
+ listNode *ln;
+} listTypeIterator;
+
+/* Structure for an entry while iterating over a list. */
+typedef struct {
+ listTypeIterator *li;
+ unsigned char *zi; /* Entry in ziplist */
+ listNode *ln; /* Entry in linked list */
+} listTypeEntry;
+
+/* Initialize an iterator at the specified index. */
+static listTypeIterator *listTypeInitIterator(robj *subject, int index, unsigned char direction) {
+ listTypeIterator *li = zmalloc(sizeof(listTypeIterator));
+ li->subject = subject;
+ li->encoding = subject->encoding;
+ li->direction = direction;
+ if (li->encoding == REDIS_ENCODING_ZIPLIST) {
+ li->zi = ziplistIndex(subject->ptr,index);
+ } else if (li->encoding == REDIS_ENCODING_LIST) {
+ li->ln = listIndex(subject->ptr,index);
+ } else {
+ redisPanic("Unknown list encoding");
+ }
+ return li;
+}
+
+/* Clean up the iterator. */
+static void listTypeReleaseIterator(listTypeIterator *li) {
+ zfree(li);
+}
- incrRefCount(o);
- deleteIfVolatile(c->db,c->argv[2]);
- if (dictAdd(c->db->dict,c->argv[2],o) == DICT_ERR) {
- if (nx) {
- decrRefCount(o);
- addReply(c,shared.czero);
- return;
+/* Stores pointer to current the entry in the provided entry structure
+ * and advances the position of the iterator. Returns 1 when the current
+ * entry is in fact an entry, 0 otherwise. */
+static int listTypeNext(listTypeIterator *li, listTypeEntry *entry) {
+ entry->li = li;
+ if (li->encoding == REDIS_ENCODING_ZIPLIST) {
+ entry->zi = li->zi;
+ if (entry->zi != NULL) {
+ if (li->direction == REDIS_TAIL)
+ li->zi = ziplistNext(li->subject->ptr,li->zi);
+ else
+ li->zi = ziplistPrev(li->subject->ptr,li->zi);
+ return 1;
+ }
+ } else if (li->encoding == REDIS_ENCODING_LIST) {
+ entry->ln = li->ln;
+ if (entry->ln != NULL) {
+ if (li->direction == REDIS_TAIL)
+ li->ln = li->ln->next;
+ else
+ li->ln = li->ln->prev;
+ return 1;
}
- dictReplace(c->db->dict,c->argv[2],o);
} else {
- incrRefCount(c->argv[2]);
+ redisPanic("Unknown list encoding");
}
- deleteKey(c->db,c->argv[1]);
- touchWatchedKey(c->db,c->argv[2]);
- server.dirty++;
- addReply(c,nx ? shared.cone : shared.ok);
+ return 0;
}
-static void renameCommand(redisClient *c) {
- renameGenericCommand(c,0);
+/* Return entry or NULL at the current position of the iterator. */
+static robj *listTypeGet(listTypeEntry *entry) {
+ listTypeIterator *li = entry->li;
+ robj *value = NULL;
+ if (li->encoding == REDIS_ENCODING_ZIPLIST) {
+ unsigned char *vstr;
+ unsigned int vlen;
+ long long vlong;
+ redisAssert(entry->zi != NULL);
+ if (ziplistGet(entry->zi,&vstr,&vlen,&vlong)) {
+ if (vstr) {
+ value = createStringObject((char*)vstr,vlen);
+ } else {
+ value = createStringObjectFromLongLong(vlong);
+ }
+ }
+ } else if (li->encoding == REDIS_ENCODING_LIST) {
+ redisAssert(entry->ln != NULL);
+ value = listNodeValue(entry->ln);
+ incrRefCount(value);
+ } else {
+ redisPanic("Unknown list encoding");
+ }
+ return value;
}
-static void renamenxCommand(redisClient *c) {
- renameGenericCommand(c,1);
+/* Compare the given object with the entry at the current position. */
+static int listTypeEqual(listTypeEntry *entry, robj *o) {
+ listTypeIterator *li = entry->li;
+ if (li->encoding == REDIS_ENCODING_ZIPLIST) {
+ redisAssert(o->encoding == REDIS_ENCODING_RAW);
+ return ziplistCompare(entry->zi,o->ptr,sdslen(o->ptr));
+ } else if (li->encoding == REDIS_ENCODING_LIST) {
+ return equalStringObjects(o,listNodeValue(entry->ln));
+ } else {
+ redisPanic("Unknown list encoding");
+ }
}
-static void moveCommand(redisClient *c) {
- robj *o;
- redisDb *src, *dst;
- int srcid;
+/* Delete the element pointed to. */
+static void listTypeDelete(listTypeEntry *entry) {
+ listTypeIterator *li = entry->li;
+ if (li->encoding == REDIS_ENCODING_ZIPLIST) {
+ unsigned char *p = entry->zi;
+ li->subject->ptr = ziplistDelete(li->subject->ptr,&p);
- /* Obtain source and target DB pointers */
- src = c->db;
- srcid = c->db->id;
- if (selectDb(c,atoi(c->argv[2]->ptr)) == REDIS_ERR) {
- addReply(c,shared.outofrangeerr);
- return;
+ /* Update position of the iterator depending on the direction */
+ if (li->direction == REDIS_TAIL)
+ li->zi = p;
+ else
+ li->zi = ziplistPrev(li->subject->ptr,p);
+ } else if (entry->li->encoding == REDIS_ENCODING_LIST) {
+ listNode *next;
+ if (li->direction == REDIS_TAIL)
+ next = entry->ln->next;
+ else
+ next = entry->ln->prev;
+ listDelNode(li->subject->ptr,entry->ln);
+ li->ln = next;
+ } else {
+ redisPanic("Unknown list encoding");
}
- dst = c->db;
- selectDb(c,srcid); /* Back to the source DB */
+}
- /* If the user is moving using as target the same
- * DB as the source DB it is probably an error. */
- if (src == dst) {
- addReply(c,shared.sameobjecterr);
- return;
- }
+static void listTypeConvert(robj *subject, int enc) {
+ listTypeIterator *li;
+ listTypeEntry entry;
+ redisAssert(subject->type == REDIS_LIST);
- /* Check if the element exists and get a reference */
- o = lookupKeyWrite(c->db,c->argv[1]);
- if (!o) {
- addReply(c,shared.czero);
- return;
- }
+ if (enc == REDIS_ENCODING_LIST) {
+ list *l = listCreate();
- /* Try to add the element to the target DB */
- deleteIfVolatile(dst,c->argv[1]);
- if (dictAdd(dst->dict,c->argv[1],o) == DICT_ERR) {
- addReply(c,shared.czero);
- return;
- }
- incrRefCount(c->argv[1]);
- incrRefCount(o);
+ /* listTypeGet returns a robj with incremented refcount */
+ li = listTypeInitIterator(subject,0,REDIS_TAIL);
+ while (listTypeNext(li,&entry)) listAddNodeTail(l,listTypeGet(&entry));
+ listTypeReleaseIterator(li);
- /* OK! key moved, free the entry in the source DB */
- deleteKey(src,c->argv[1]);
- server.dirty++;
- addReply(c,shared.cone);
+ subject->encoding = REDIS_ENCODING_LIST;
+ zfree(subject->ptr);
+ subject->ptr = l;
+ } else {
+ redisPanic("Unsupported list conversion");
+ }
}
-/* =================================== Lists ================================ */
static void pushGenericCommand(redisClient *c, int where) {
- robj *lobj;
- list *list;
-
- lobj = lookupKeyWrite(c->db,c->argv[1]);
+ robj *lobj = lookupKeyWrite(c->db,c->argv[1]);
if (lobj == NULL) {
if (handleClientsWaitingListPush(c,c->argv[1],c->argv[2])) {
addReply(c,shared.cone);
return;
}
- lobj = createListObject();
- list = lobj->ptr;
- if (where == REDIS_HEAD) {
- listAddNodeHead(list,c->argv[2]);
- } else {
- listAddNodeTail(list,c->argv[2]);
- }
- dictAdd(c->db->dict,c->argv[1],lobj);
- incrRefCount(c->argv[1]);
- incrRefCount(c->argv[2]);
+ lobj = createZiplistObject();
+ dbAdd(c->db,c->argv[1],lobj);
} else {
if (lobj->type != REDIS_LIST) {
addReply(c,shared.wrongtypeerr);
addReply(c,shared.cone);
return;
}
- list = lobj->ptr;
- if (where == REDIS_HEAD) {
- listAddNodeHead(list,c->argv[2]);
- } else {
- listAddNodeTail(list,c->argv[2]);
- }
- incrRefCount(c->argv[2]);
}
+ listTypePush(lobj,c->argv[2],where);
+ addReplyLongLong(c,listTypeLength(lobj));
server.dirty++;
- addReplyLongLong(c,listLength(list));
}
static void lpushCommand(redisClient *c) {
}
static void llenCommand(redisClient *c) {
- robj *o;
- list *l;
-
- if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
- checkType(c,o,REDIS_LIST)) return;
-
- l = o->ptr;
- addReplyUlong(c,listLength(l));
+ robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.czero);
+ if (o == NULL || checkType(c,o,REDIS_LIST)) return;
+ addReplyUlong(c,listTypeLength(o));
}
static void lindexCommand(redisClient *c) {
- robj *o;
+ robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk);
+ if (o == NULL || checkType(c,o,REDIS_LIST)) return;
int index = atoi(c->argv[2]->ptr);
- list *list;
- listNode *ln;
-
- if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
- checkType(c,o,REDIS_LIST)) return;
- list = o->ptr;
+ robj *value = NULL;
- ln = listIndex(list, index);
- if (ln == NULL) {
- addReply(c,shared.nullbulk);
+ if (o->encoding == REDIS_ENCODING_ZIPLIST) {
+ unsigned char *p;
+ unsigned char *vstr;
+ unsigned int vlen;
+ long long vlong;
+ p = ziplistIndex(o->ptr,index);
+ if (ziplistGet(p,&vstr,&vlen,&vlong)) {
+ if (vstr) {
+ value = createStringObject((char*)vstr,vlen);
+ } else {
+ value = createStringObjectFromLongLong(vlong);
+ }
+ addReplyBulk(c,value);
+ decrRefCount(value);
+ } else {
+ addReply(c,shared.nullbulk);
+ }
+ } else if (o->encoding == REDIS_ENCODING_LIST) {
+ listNode *ln = listIndex(o->ptr,index);
+ if (ln != NULL) {
+ value = listNodeValue(ln);
+ addReplyBulk(c,value);
+ } else {
+ addReply(c,shared.nullbulk);
+ }
} else {
- robj *ele = listNodeValue(ln);
- addReplyBulk(c,ele);
+ redisPanic("Unknown list encoding");
}
}
static void lsetCommand(redisClient *c) {
- robj *o;
+ robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr);
+ if (o == NULL || checkType(c,o,REDIS_LIST)) return;
int index = atoi(c->argv[2]->ptr);
- list *list;
- listNode *ln;
-
- if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr)) == NULL ||
- checkType(c,o,REDIS_LIST)) return;
- list = o->ptr;
-
- ln = listIndex(list, index);
- if (ln == NULL) {
- addReply(c,shared.outofrangeerr);
+ robj *value = c->argv[3];
+
+ listTypeTryConversion(o,value);
+ if (o->encoding == REDIS_ENCODING_ZIPLIST) {
+ unsigned char *p, *zl = o->ptr;
+ p = ziplistIndex(zl,index);
+ if (p == NULL) {
+ addReply(c,shared.outofrangeerr);
+ } else {
+ o->ptr = ziplistDelete(o->ptr,&p);
+ value = getDecodedObject(value);
+ o->ptr = ziplistInsert(o->ptr,p,value->ptr,sdslen(value->ptr));
+ decrRefCount(value);
+ addReply(c,shared.ok);
+ server.dirty++;
+ }
+ } else if (o->encoding == REDIS_ENCODING_LIST) {
+ listNode *ln = listIndex(o->ptr,index);
+ if (ln == NULL) {
+ addReply(c,shared.outofrangeerr);
+ } else {
+ decrRefCount((robj*)listNodeValue(ln));
+ listNodeValue(ln) = value;
+ incrRefCount(value);
+ addReply(c,shared.ok);
+ server.dirty++;
+ }
} else {
- robj *ele = listNodeValue(ln);
-
- decrRefCount(ele);
- listNodeValue(ln) = c->argv[3];
- incrRefCount(c->argv[3]);
- addReply(c,shared.ok);
- server.dirty++;
+ redisPanic("Unknown list encoding");
}
}
static void popGenericCommand(redisClient *c, int where) {
- robj *o;
- list *list;
- listNode *ln;
-
- if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
- checkType(c,o,REDIS_LIST)) return;
- list = o->ptr;
-
- if (where == REDIS_HEAD)
- ln = listFirst(list);
- else
- ln = listLast(list);
+ robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk);
+ if (o == NULL || checkType(c,o,REDIS_LIST)) return;
- if (ln == NULL) {
+ robj *value = listTypePop(o,where);
+ if (value == NULL) {
addReply(c,shared.nullbulk);
} else {
- robj *ele = listNodeValue(ln);
- addReplyBulk(c,ele);
- listDelNode(list,ln);
- if (listLength(list) == 0) deleteKey(c->db,c->argv[1]);
+ addReplyBulk(c,value);
+ decrRefCount(value);
+ if (listTypeLength(o) == 0) dbDelete(c->db,c->argv[1]);
server.dirty++;
}
}
}
static void lrangeCommand(redisClient *c) {
- robj *o;
+ robj *o, *value;
int start = atoi(c->argv[2]->ptr);
int end = atoi(c->argv[3]->ptr);
int llen;
int rangelen, j;
- list *list;
- listNode *ln;
- robj *ele;
+ listTypeEntry entry;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptymultibulk)) == NULL
|| checkType(c,o,REDIS_LIST)) return;
- list = o->ptr;
- llen = listLength(list);
+ llen = listTypeLength(o);
/* convert negative indexes */
if (start < 0) start = llen+start;
rangelen = (end-start)+1;
/* Return the result in form of a multi-bulk reply */
- ln = listIndex(list, start);
addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",rangelen));
+ listTypeIterator *li = listTypeInitIterator(o,start,REDIS_TAIL);
for (j = 0; j < rangelen; j++) {
- ele = listNodeValue(ln);
- addReplyBulk(c,ele);
- ln = ln->next;
+ redisAssert(listTypeNext(li,&entry));
+ value = listTypeGet(&entry);
+ addReplyBulk(c,value);
+ decrRefCount(value);
}
+ listTypeReleaseIterator(li);
}
static void ltrimCommand(redisClient *c) {
if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.ok)) == NULL ||
checkType(c,o,REDIS_LIST)) return;
- list = o->ptr;
- llen = listLength(list);
+ llen = listTypeLength(o);
/* convert negative indexes */
if (start < 0) start = llen+start;
}
/* Remove list elements to perform the trim */
- for (j = 0; j < ltrim; j++) {
- ln = listFirst(list);
- listDelNode(list,ln);
- }
- for (j = 0; j < rtrim; j++) {
- ln = listLast(list);
- listDelNode(list,ln);
+ if (o->encoding == REDIS_ENCODING_ZIPLIST) {
+ o->ptr = ziplistDeleteRange(o->ptr,0,ltrim);
+ o->ptr = ziplistDeleteRange(o->ptr,-rtrim,rtrim);
+ } else if (o->encoding == REDIS_ENCODING_LIST) {
+ list = o->ptr;
+ for (j = 0; j < ltrim; j++) {
+ ln = listFirst(list);
+ listDelNode(list,ln);
+ }
+ for (j = 0; j < rtrim; j++) {
+ ln = listLast(list);
+ listDelNode(list,ln);
+ }
+ } else {
+ redisPanic("Unknown list encoding");
}
- if (listLength(list) == 0) deleteKey(c->db,c->argv[1]);
+ if (listTypeLength(o) == 0) dbDelete(c->db,c->argv[1]);
server.dirty++;
addReply(c,shared.ok);
}
static void lremCommand(redisClient *c) {
- robj *o;
- list *list;
- listNode *ln, *next;
+ robj *subject, *obj = c->argv[3];
int toremove = atoi(c->argv[2]->ptr);
int removed = 0;
- int fromtail = 0;
+ listTypeEntry entry;
- if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
- checkType(c,o,REDIS_LIST)) return;
- list = o->ptr;
+ subject = lookupKeyWriteOrReply(c,c->argv[1],shared.czero);
+ if (subject == NULL || checkType(c,subject,REDIS_LIST)) return;
+ /* Make sure obj is raw when we're dealing with a ziplist */
+ if (subject->encoding == REDIS_ENCODING_ZIPLIST)
+ obj = getDecodedObject(obj);
+
+ listTypeIterator *li;
if (toremove < 0) {
toremove = -toremove;
- fromtail = 1;
+ li = listTypeInitIterator(subject,-1,REDIS_HEAD);
+ } else {
+ li = listTypeInitIterator(subject,0,REDIS_TAIL);
}
- ln = fromtail ? list->tail : list->head;
- while (ln) {
- robj *ele = listNodeValue(ln);
- next = fromtail ? ln->prev : ln->next;
- if (equalStringObjects(ele,c->argv[3])) {
- listDelNode(list,ln);
+ while (listTypeNext(li,&entry)) {
+ if (listTypeEqual(&entry,obj)) {
+ listTypeDelete(&entry);
server.dirty++;
removed++;
if (toremove && removed == toremove) break;
}
- ln = next;
}
- if (listLength(list) == 0) deleteKey(c->db,c->argv[1]);
+ listTypeReleaseIterator(li);
+
+ /* Clean up raw encoded object */
+ if (subject->encoding == REDIS_ENCODING_ZIPLIST)
+ decrRefCount(obj);
+
+ if (listTypeLength(subject) == 0) dbDelete(c->db,c->argv[1]);
addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",removed));
}
* as well. This command was originally proposed by Ezra Zygmuntowicz.
*/
static void rpoplpushcommand(redisClient *c) {
- robj *sobj;
- list *srclist;
- listNode *ln;
-
+ robj *sobj, *value;
if ((sobj = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
checkType(c,sobj,REDIS_LIST)) return;
- srclist = sobj->ptr;
- ln = listLast(srclist);
- if (ln == NULL) {
+ if (listTypeLength(sobj) == 0) {
addReply(c,shared.nullbulk);
} else {
robj *dobj = lookupKeyWrite(c->db,c->argv[2]);
- robj *ele = listNodeValue(ln);
- list *dstlist;
-
- if (dobj && dobj->type != REDIS_LIST) {
- addReply(c,shared.wrongtypeerr);
- return;
- }
+ if (dobj && checkType(c,dobj,REDIS_LIST)) return;
+ value = listTypePop(sobj,REDIS_TAIL);
/* Add the element to the target list (unless it's directly
* passed to some BLPOP-ing client */
- if (!handleClientsWaitingListPush(c,c->argv[2],ele)) {
- if (dobj == NULL) {
- /* Create the list if the key does not exist */
- dobj = createListObject();
- dictAdd(c->db->dict,c->argv[2],dobj);
- incrRefCount(c->argv[2]);
+ if (!handleClientsWaitingListPush(c,c->argv[2],value)) {
+ /* Create the list if the key does not exist */
+ if (!dobj) {
+ dobj = createZiplistObject();
+ dbAdd(c->db,c->argv[2],dobj);
}
- dstlist = dobj->ptr;
- listAddNodeHead(dstlist,ele);
- incrRefCount(ele);
+ listTypePush(dobj,value,REDIS_HEAD);
}
/* Send the element to the client as reply as well */
- addReplyBulk(c,ele);
+ addReplyBulk(c,value);
- /* Finally remove the element from the source list */
- listDelNode(srclist,ln);
- if (listLength(srclist) == 0) deleteKey(c->db,c->argv[1]);
+ /* listTypePop returns an object with its refcount incremented */
+ decrRefCount(value);
+
+ /* Delete the source list when it is empty */
+ if (listTypeLength(sobj) == 0) dbDelete(c->db,c->argv[1]);
server.dirty++;
}
}
set = lookupKeyWrite(c->db,c->argv[1]);
if (set == NULL) {
set = createSetObject();
- dictAdd(c->db->dict,c->argv[1],set);
- incrRefCount(c->argv[1]);
+ dbAdd(c->db,c->argv[1],set);
} else {
if (set->type != REDIS_SET) {
addReply(c,shared.wrongtypeerr);
if (dictDelete(set->ptr,c->argv[2]) == DICT_OK) {
server.dirty++;
if (htNeedsResize(set->ptr)) dictResize(set->ptr);
- if (dictSize((dict*)set->ptr) == 0) deleteKey(c->db,c->argv[1]);
+ if (dictSize((dict*)set->ptr) == 0) dbDelete(c->db,c->argv[1]);
addReply(c,shared.cone);
} else {
addReply(c,shared.czero);
return;
}
if (dictSize((dict*)srcset->ptr) == 0 && srcset != dstset)
- deleteKey(c->db,c->argv[1]);
+ dbDelete(c->db,c->argv[1]);
server.dirty++;
/* Add the element to the destination set */
if (!dstset) {
dstset = createSetObject();
- dictAdd(c->db->dict,c->argv[2],dstset);
- incrRefCount(c->argv[2]);
+ dbAdd(c->db,c->argv[2],dstset);
}
if (dictAdd(dstset->ptr,c->argv[3],NULL) == DICT_OK)
incrRefCount(c->argv[3]);
addReplyBulk(c,ele);
dictDelete(set->ptr,ele);
if (htNeedsResize(set->ptr)) dictResize(set->ptr);
- if (dictSize((dict*)set->ptr) == 0) deleteKey(c->db,c->argv[1]);
+ if (dictSize((dict*)set->ptr) == 0) dbDelete(c->db,c->argv[1]);
server.dirty++;
}
}
if (!setobj) {
zfree(dv);
if (dstkey) {
- if (deleteKey(c->db,dstkey))
+ if (dbDelete(c->db,dstkey))
server.dirty++;
addReply(c,shared.czero);
} else {
if (dstkey) {
/* Store the resulting set into the target, if the intersection
* is not an empty set. */
- deleteKey(c->db,dstkey);
+ dbDelete(c->db,dstkey);
if (dictSize((dict*)dstset->ptr) > 0) {
- dictAdd(c->db->dict,dstkey,dstset);
- incrRefCount(dstkey);
+ dbAdd(c->db,dstkey,dstset);
addReplyLongLong(c,dictSize((dict*)dstset->ptr));
} else {
decrRefCount(dstset);
} else {
/* If we have a target key where to store the resulting set
* create this key with the result set inside */
- deleteKey(c->db,dstkey);
+ dbDelete(c->db,dstkey);
if (dictSize((dict*)dstset->ptr) > 0) {
- dictAdd(c->db->dict,dstkey,dstset);
- incrRefCount(dstkey);
+ dbAdd(c->db,dstkey,dstset);
addReplyLongLong(c,dictSize((dict*)dstset->ptr));
} else {
decrRefCount(dstset);
* Returns 0 when the element cannot be found, rank otherwise.
* Note that the rank is 1-based due to the span of zsl->header to the
* first element. */
-static unsigned long zslGetRank(zskiplist *zsl, double score, robj *o) {
+static unsigned long zslistTypeGetRank(zskiplist *zsl, double score, robj *o) {
zskiplistNode *x;
unsigned long rank = 0;
int i;
}
/* Finds an element by its rank. The rank argument needs to be 1-based. */
-zskiplistNode* zslGetElementByRank(zskiplist *zsl, unsigned long rank) {
+zskiplistNode* zslistTypeGetElementByRank(zskiplist *zsl, unsigned long rank) {
zskiplistNode *x;
unsigned long traversed = 0;
int i;
zsetobj = lookupKeyWrite(c->db,key);
if (zsetobj == NULL) {
zsetobj = createZsetObject();
- dictAdd(c->db->dict,key,zsetobj);
- incrRefCount(key);
+ dbAdd(c->db,key,zsetobj);
} else {
if (zsetobj->type != REDIS_ZSET) {
addReply(c,shared.wrongtypeerr);
/* Delete from the hash table */
dictDelete(zs->dict,c->argv[2]);
if (htNeedsResize(zs->dict)) dictResize(zs->dict);
- if (dictSize(zs->dict) == 0) deleteKey(c->db,c->argv[1]);
+ if (dictSize(zs->dict) == 0) dbDelete(c->db,c->argv[1]);
server.dirty++;
addReply(c,shared.cone);
}
zs = zsetobj->ptr;
deleted = zslDeleteRangeByScore(zs->zsl,min,max,zs->dict);
if (htNeedsResize(zs->dict)) dictResize(zs->dict);
- if (dictSize(zs->dict) == 0) deleteKey(c->db,c->argv[1]);
+ if (dictSize(zs->dict) == 0) dbDelete(c->db,c->argv[1]);
server.dirty += deleted;
addReplyLongLong(c,deleted);
}
* use 1-based rank */
deleted = zslDeleteRangeByRank(zs->zsl,start+1,end+1,zs->dict);
if (htNeedsResize(zs->dict)) dictResize(zs->dict);
- if (dictSize(zs->dict) == 0) deleteKey(c->db,c->argv[1]);
+ if (dictSize(zs->dict) == 0) dbDelete(c->db,c->argv[1]);
server.dirty += deleted;
addReplyLongLong(c, deleted);
}
redisAssert(op == REDIS_OP_INTER || op == REDIS_OP_UNION);
}
- deleteKey(c->db,dstkey);
+ dbDelete(c->db,dstkey);
if (dstzset->zsl->length) {
- dictAdd(c->db->dict,dstkey,dstobj);
- incrRefCount(dstkey);
+ dbAdd(c->db,dstkey,dstobj);
addReplyLongLong(c, dstzset->zsl->length);
server.dirty++;
} else {
/* check if starting point is trivial, before searching
* the element in log(N) time */
if (reverse) {
- ln = start == 0 ? zsl->tail : zslGetElementByRank(zsl, llen-start);
+ ln = start == 0 ? zsl->tail : zslistTypeGetElementByRank(zsl, llen-start);
} else {
ln = start == 0 ?
- zsl->header->forward[0] : zslGetElementByRank(zsl, start+1);
+ zsl->header->forward[0] : zslistTypeGetElementByRank(zsl, start+1);
}
/* Return the result in form of a multi-bulk reply */
}
score = dictGetEntryVal(de);
- rank = zslGetRank(zsl, *score, c->argv[2]);
+ rank = zslistTypeGetRank(zsl, *score, c->argv[2]);
if (rank) {
if (reverse) {
addReplyLongLong(c, zsl->length - rank);
robj *o = lookupKeyWrite(c->db,key);
if (o == NULL) {
o = createHashObject();
- dictAdd(c->db->dict,key,o);
- incrRefCount(key);
+ dbAdd(c->db,key,o);
} else {
if (o->type != REDIS_HASH) {
addReply(c,shared.wrongtypeerr);
checkType(c,o,REDIS_HASH)) return;
if (hashDelete(o,c->argv[2])) {
- if (hashLength(o) == 0) deleteKey(c->db,c->argv[1]);
+ if (hashLength(o) == 0) dbDelete(c->db,c->argv[1]);
addReply(c,shared.cone);
server.dirty++;
} else {
* is optimized for speed and a bit less for readability */
static void sortCommand(redisClient *c) {
list *operations;
- int outputlen = 0;
+ unsigned int outputlen = 0;
int desc = 0, alpha = 0;
int limit_start = 0, limit_count = -1, start, end;
int j, dontsort = 0, vectorlen;
/* Load the sorting vector with all the objects to sort */
switch(sortval->type) {
- case REDIS_LIST: vectorlen = listLength((list*)sortval->ptr); break;
+ case REDIS_LIST: vectorlen = listTypeLength(sortval); break;
case REDIS_SET: vectorlen = dictSize((dict*)sortval->ptr); break;
case REDIS_ZSET: vectorlen = dictSize(((zset*)sortval->ptr)->dict); break;
default: vectorlen = 0; redisPanic("Bad SORT type"); /* Avoid GCC warning */
j = 0;
if (sortval->type == REDIS_LIST) {
- list *list = sortval->ptr;
- listNode *ln;
- listIter li;
-
- listRewind(list,&li);
- while((ln = listNext(&li))) {
- robj *ele = ln->value;
- vector[j].obj = ele;
+ listTypeIterator *li = listTypeInitIterator(sortval,0,REDIS_TAIL);
+ listTypeEntry entry;
+ while(listTypeNext(li,&entry)) {
+ vector[j].obj = listTypeGet(&entry);
vector[j].u.score = 0;
vector[j].u.cmpobj = NULL;
j++;
}
+ listTypeReleaseIterator(li);
} else {
dict *set;
dictIterator *di;
}
}
} else {
- robj *listObject = createListObject();
- list *listPtr = (list*) listObject->ptr;
+ robj *sobj = createZiplistObject();
/* STORE option specified, set the sorting result as a List object */
for (j = start; j <= end; j++) {
listIter li;
if (!getop) {
- listAddNodeTail(listPtr,vector[j].obj);
- incrRefCount(vector[j].obj);
- }
- listRewind(operations,&li);
- while((ln = listNext(&li))) {
- redisSortOperation *sop = ln->value;
- robj *val = lookupKeyByPattern(c->db,sop->pattern,
- vector[j].obj);
+ listTypePush(sobj,vector[j].obj,REDIS_TAIL);
+ } else {
+ listRewind(operations,&li);
+ while((ln = listNext(&li))) {
+ redisSortOperation *sop = ln->value;
+ robj *val = lookupKeyByPattern(c->db,sop->pattern,
+ vector[j].obj);
- if (sop->type == REDIS_SORT_GET) {
- if (!val) {
- listAddNodeTail(listPtr,createStringObject("",0));
+ if (sop->type == REDIS_SORT_GET) {
+ if (!val) val = createStringObject("",0);
+
+ /* listTypePush does an incrRefCount, so we should take care
+ * care of the incremented refcount caused by either
+ * lookupKeyByPattern or createStringObject("",0) */
+ listTypePush(sobj,val,REDIS_TAIL);
+ decrRefCount(val);
} else {
- /* We should do a incrRefCount on val because it is
- * added to the list, but also a decrRefCount because
- * it is returned by lookupKeyByPattern. This results
- * in doing nothing at all. */
- listAddNodeTail(listPtr,val);
+ /* always fails */
+ redisAssert(sop->type == REDIS_SORT_GET);
}
- } else {
- redisAssert(sop->type == REDIS_SORT_GET); /* always fails */
}
}
}
- if (dictReplace(c->db->dict,storekey,listObject)) {
- incrRefCount(storekey);
- }
+ dbReplace(c->db,storekey,sobj);
/* Note: we add 1 because the DB is dirty anyway since even if the
* SORT result is empty a new key is set and maybe the old content
* replaced. */
}
/* Cleanup */
+ if (sortval->type == REDIS_LIST)
+ for (j = 0; j < vectorlen; j++)
+ decrRefCount(vector[j].obj);
decrRefCount(sortval);
listRelease(operations);
for (j = 0; j < vectorlen; j++) {
/* ================================= Expire ================================= */
static int removeExpire(redisDb *db, robj *key) {
- if (dictDelete(db->expires,key) == DICT_OK) {
+ if (dictDelete(db->expires,key->ptr) == DICT_OK) {
return 1;
} else {
return 0;
}
static int setExpire(redisDb *db, robj *key, time_t when) {
- if (dictAdd(db->expires,key,(void*)when) == DICT_ERR) {
+ sds copy = sdsdup(key->ptr);
+ if (dictAdd(db->expires,copy,(void*)when) == DICT_ERR) {
+ sdsfree(copy);
return 0;
} else {
- incrRefCount(key);
return 1;
}
}
/* No expire? return ASAP */
if (dictSize(db->expires) == 0 ||
- (de = dictFind(db->expires,key)) == NULL) return -1;
+ (de = dictFind(db->expires,key->ptr)) == NULL) return -1;
return (time_t) dictGetEntryVal(de);
}
/* No expire? return ASAP */
if (dictSize(db->expires) == 0 ||
- (de = dictFind(db->expires,key)) == NULL) return 0;
+ (de = dictFind(db->expires,key->ptr)) == NULL) return 0;
/* Lookup the expire */
when = (time_t) dictGetEntryVal(de);
if (time(NULL) <= when) return 0;
/* Delete the key */
- dictDelete(db->expires,key);
+ dbDelete(db,key);
server.stat_expiredkeys++;
- return dictDelete(db->dict,key) == DICT_OK;
+ return 1;
}
static int deleteIfVolatile(redisDb *db, robj *key) {
/* No expire? return ASAP */
if (dictSize(db->expires) == 0 ||
- (de = dictFind(db->expires,key)) == NULL) return 0;
+ (de = dictFind(db->expires,key->ptr)) == NULL) return 0;
/* Delete the key */
server.dirty++;
server.stat_expiredkeys++;
- dictDelete(db->expires,key);
- return dictDelete(db->dict,key) == DICT_OK;
+ dictDelete(db->expires,key->ptr);
+ return dictDelete(db->dict,key->ptr) == DICT_OK;
}
static void expireGenericCommand(redisClient *c, robj *key, robj *param, long offset) {
seconds -= offset;
- de = dictFind(c->db->dict,key);
+ de = dictFind(c->db->dict,key->ptr);
if (de == NULL) {
addReply(c,shared.czero);
return;
}
if (seconds <= 0) {
- if (deleteKey(c->db,key)) server.dirty++;
+ if (dbDelete(c->db,key)) server.dirty++;
addReply(c, shared.cone);
return;
} else {
minttl = t;
}
}
- deleteKey(server.db+j,minkey);
+ dbDelete(server.db+j,minkey);
}
}
if (!freed) return; /* nothing to free... */
struct redisClient *fakeClient;
FILE *fp = fopen(filename,"r");
struct redis_stat sb;
- unsigned long long loadedkeys = 0;
int appendonly = server.appendonly;
if (redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0)
char buf[128];
sds argsds;
struct redisCommand *cmd;
+ int force_swapout;
if (fgets(buf,sizeof(buf),fp) == NULL) {
if (feof(fp))
for (j = 0; j < argc; j++) decrRefCount(argv[j]);
zfree(argv);
/* Handle swapping while loading big datasets when VM is on */
- loadedkeys++;
- if (server.vm_enabled && (loadedkeys % 5000) == 0) {
+ force_swapout = 0;
+ if ((zmalloc_used_memory() - server.vm_max_memory) > 1024*1024*32)
+ force_swapout = 1;
+
+ if (server.vm_enabled && force_swapout) {
while (zmalloc_used_memory() > server.vm_max_memory) {
if (vmSwapOneObjectBlocking() == REDIS_ERR) break;
}
exit(1);
}
-/* Write an object into a file in the bulk format $<count>\r\n<payload>\r\n */
-static int fwriteBulkObject(FILE *fp, robj *obj) {
- char buf[128];
- int decrrc = 0;
-
- /* Avoid the incr/decr ref count business if possible to help
- * copy-on-write (we are often in a child process when this function
- * is called).
- * Also makes sure that key objects don't get incrRefCount-ed when VM
- * is enabled */
- if (obj->encoding != REDIS_ENCODING_RAW) {
- obj = getDecodedObject(obj);
- decrrc = 1;
- }
- snprintf(buf,sizeof(buf),"$%ld\r\n",(long)sdslen(obj->ptr));
- if (fwrite(buf,strlen(buf),1,fp) == 0) goto err;
- if (sdslen(obj->ptr) && fwrite(obj->ptr,sdslen(obj->ptr),1,fp) == 0)
- goto err;
- if (fwrite("\r\n",2,1,fp) == 0) goto err;
- if (decrrc) decrRefCount(obj);
- return 1;
-err:
- if (decrrc) decrRefCount(obj);
- return 0;
-}
-
/* Write binary-safe string into a file in the bulkformat
* $<count>\r\n<payload>\r\n */
static int fwriteBulkString(FILE *fp, char *s, unsigned long len) {
- char buf[128];
-
- snprintf(buf,sizeof(buf),"$%ld\r\n",(unsigned long)len);
- if (fwrite(buf,strlen(buf),1,fp) == 0) return 0;
- if (len && fwrite(s,len,1,fp) == 0) return 0;
+ char cbuf[128];
+ int clen;
+ cbuf[0] = '$';
+ clen = 1+ll2string(cbuf+1,sizeof(cbuf)-1,len);
+ cbuf[clen++] = '\r';
+ cbuf[clen++] = '\n';
+ if (fwrite(cbuf,clen,1,fp) == 0) return 0;
+ if (len > 0 && fwrite(s,len,1,fp) == 0) return 0;
if (fwrite("\r\n",2,1,fp) == 0) return 0;
return 1;
}
}
/* Write a long value in bulk format $<count>\r\n<payload>\r\n */
-static int fwriteBulkLong(FILE *fp, long l) {
- char buf[128], lbuf[128];
-
- snprintf(lbuf,sizeof(lbuf),"%ld\r\n",l);
- snprintf(buf,sizeof(buf),"$%lu\r\n",(unsigned long)strlen(lbuf)-2);
- if (fwrite(buf,strlen(buf),1,fp) == 0) return 0;
- if (fwrite(lbuf,strlen(lbuf),1,fp) == 0) return 0;
+static int fwriteBulkLongLong(FILE *fp, long long l) {
+ char bbuf[128], lbuf[128];
+ unsigned int blen, llen;
+ llen = ll2string(lbuf,32,l);
+ blen = snprintf(bbuf,sizeof(bbuf),"$%u\r\n%s\r\n",llen,lbuf);
+ if (fwrite(bbuf,blen,1,fp) == 0) return 0;
return 1;
}
+/* Delegate writing an object to writing a bulk string or bulk long long. */
+static int fwriteBulkObject(FILE *fp, robj *obj) {
+ /* Avoid using getDecodedObject to help copy-on-write (we are often
+ * in a child process when this function is called). */
+ if (obj->encoding == REDIS_ENCODING_INT) {
+ return fwriteBulkLongLong(fp,(long)obj->ptr);
+ } else if (obj->encoding == REDIS_ENCODING_RAW) {
+ return fwriteBulkString(fp,obj->ptr,sdslen(obj->ptr));
+ } else {
+ redisPanic("Unknown string encoding");
+ }
+}
+
/* Write a sequence of commands able to fully rebuild the dataset into
* "filename". Used both by REWRITEAOF and BGREWRITEAOF. */
static int rewriteAppendOnlyFile(char *filename) {
/* SELECT the new DB */
if (fwrite(selectcmd,sizeof(selectcmd)-1,1,fp) == 0) goto werr;
- if (fwriteBulkLong(fp,j) == 0) goto werr;
+ if (fwriteBulkLongLong(fp,j) == 0) goto werr;
/* Iterate this DB writing every entry */
while((de = dictNext(di)) != NULL) {
- robj *key, *o;
+ sds keystr = dictGetEntryKey(de);
+ robj key, *o;
time_t expiretime;
int swapped;
- key = dictGetEntryKey(de);
+ keystr = dictGetEntryKey(de);
o = dictGetEntryVal(de);
+ initStaticStringObject(key,keystr);
/* If the value for this key is swapped, load a preview in memory.
* We use a "swapped" flag to remember if we need to free the
* value object instead to just increment the ref count anyway
o = vmPreviewObject(o);
swapped = 1;
}
- expiretime = getExpire(db,key);
+ expiretime = getExpire(db,&key);
/* Save the key and associated value */
if (o->type == REDIS_STRING) {
char cmd[]="*3\r\n$3\r\nSET\r\n";
if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
/* Key and value */
- if (fwriteBulkObject(fp,key) == 0) goto werr;
+ if (fwriteBulkObject(fp,&key) == 0) goto werr;
if (fwriteBulkObject(fp,o) == 0) goto werr;
} else if (o->type == REDIS_LIST) {
/* Emit the RPUSHes needed to rebuild the list */
- list *list = o->ptr;
- listNode *ln;
- listIter li;
+ char cmd[]="*3\r\n$5\r\nRPUSH\r\n";
+ if (o->encoding == REDIS_ENCODING_ZIPLIST) {
+ unsigned char *zl = o->ptr;
+ unsigned char *p = ziplistIndex(zl,0);
+ unsigned char *vstr;
+ unsigned int vlen;
+ long long vlong;
+
+ while(ziplistGet(p,&vstr,&vlen,&vlong)) {
+ if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
+ if (fwriteBulkObject(fp,&key) == 0) goto werr;
+ if (vstr) {
+ if (fwriteBulkString(fp,(char*)vstr,vlen) == 0)
+ goto werr;
+ } else {
+ if (fwriteBulkLongLong(fp,vlong) == 0)
+ goto werr;
+ }
+ p = ziplistNext(zl,p);
+ }
+ } else if (o->encoding == REDIS_ENCODING_LIST) {
+ list *list = o->ptr;
+ listNode *ln;
+ listIter li;
- listRewind(list,&li);
- while((ln = listNext(&li))) {
- char cmd[]="*3\r\n$5\r\nRPUSH\r\n";
- robj *eleobj = listNodeValue(ln);
+ listRewind(list,&li);
+ while((ln = listNext(&li))) {
+ robj *eleobj = listNodeValue(ln);
- if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
- if (fwriteBulkObject(fp,key) == 0) goto werr;
- if (fwriteBulkObject(fp,eleobj) == 0) goto werr;
+ if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
+ if (fwriteBulkObject(fp,&key) == 0) goto werr;
+ if (fwriteBulkObject(fp,eleobj) == 0) goto werr;
+ }
+ } else {
+ redisPanic("Unknown list encoding");
}
} else if (o->type == REDIS_SET) {
/* Emit the SADDs needed to rebuild the set */
robj *eleobj = dictGetEntryKey(de);
if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
- if (fwriteBulkObject(fp,key) == 0) goto werr;
+ if (fwriteBulkObject(fp,&key) == 0) goto werr;
if (fwriteBulkObject(fp,eleobj) == 0) goto werr;
}
dictReleaseIterator(di);
double *score = dictGetEntryVal(de);
if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
- if (fwriteBulkObject(fp,key) == 0) goto werr;
+ if (fwriteBulkObject(fp,&key) == 0) goto werr;
if (fwriteBulkDouble(fp,*score) == 0) goto werr;
if (fwriteBulkObject(fp,eleobj) == 0) goto werr;
}
while((p = zipmapNext(p,&field,&flen,&val,&vlen)) != NULL) {
if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
- if (fwriteBulkObject(fp,key) == 0) goto werr;
+ if (fwriteBulkObject(fp,&key) == 0) goto werr;
if (fwriteBulkString(fp,(char*)field,flen) == -1)
return -1;
if (fwriteBulkString(fp,(char*)val,vlen) == -1)
robj *val = dictGetEntryVal(de);
if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
- if (fwriteBulkObject(fp,key) == 0) goto werr;
+ if (fwriteBulkObject(fp,&key) == 0) goto werr;
if (fwriteBulkObject(fp,field) == -1) return -1;
if (fwriteBulkObject(fp,val) == -1) return -1;
}
/* If this key is already expired skip it */
if (expiretime < now) continue;
if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
- if (fwriteBulkObject(fp,key) == 0) goto werr;
- if (fwriteBulkLong(fp,expiretime) == 0) goto werr;
+ if (fwriteBulkObject(fp,&key) == 0) goto werr;
+ if (fwriteBulkLongLong(fp,expiretime) == 0) goto werr;
}
if (swapped) decrRefCount(o);
}
return REDIS_OK;
}
-/* Swap the 'val' object relative to 'key' into disk. Store all the information
- * needed to later retrieve the object into the key object.
+/* Transfers the 'val' object to disk. Store all the information
+ * a 'vmpointer' object containing all the information needed to load the
+ * object back later is returned.
+ *
* If we can't find enough contiguous empty pages to swap the object on disk
- * REDIS_ERR is returned. */
+ * NULL is returned. */
static vmpointer *vmSwapObjectBlocking(robj *val) {
off_t pages = rdbSavedObjectPages(val,NULL);
off_t page;
struct dictEntry *best = NULL;
double best_swappability = 0;
redisDb *best_db = NULL;
- robj *key, *val;
+ robj *val;
+ sds key;
for (j = 0; j < server.dbnum; j++) {
redisDb *db = server.db+j;
if (maxtries) maxtries--;
de = dictGetRandomKey(db->dict);
- key = dictGetEntryKey(de);
val = dictGetEntryVal(de);
/* Only swap objects that are currently in memory.
*
val = dictGetEntryVal(best);
redisLog(REDIS_DEBUG,"Key with best swappability: %s, %f",
- key->ptr, best_swappability);
+ key, best_swappability);
/* Swap it */
if (usethreads) {
- vmSwapObjectThreaded(key,val,best_db);
+ robj *keyobj = createStringObject(key,sdslen(key));
+ vmSwapObjectThreaded(keyobj,val,best_db);
+ decrRefCount(keyobj);
return REDIS_OK;
} else {
vmpointer *vp;
return (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1);
}
-/* Delete a key if swapped. Returns 1 if the key was found, was swapped
- * and was deleted. Otherwise 0 is returned. */
-static int deleteIfSwapped(redisDb *db, robj *key) {
- robj *val;
-
- if ((val = dictFetchValue(db->dict,key)) == NULL) return 0;
- if (val->storage == REDIS_VM_MEMORY) return 0;
- deleteKey(db,key);
- return 1;
-}
-
/* =================== Virtual Memory - Threaded I/O ======================= */
static void freeIOJob(iojob *j) {
/* Post process it in the main thread, as there are things we
* can do just here to avoid race conditions and/or invasive locks */
redisLog(REDIS_DEBUG,"COMPLETED Job type: %d, ID %p, key: %s", j->type, (void*)j->id, (unsigned char*)j->key->ptr);
- de = dictFind(j->db->dict,j->key);
+ de = dictFind(j->db->dict,j->key->ptr);
redisAssert(de != NULL);
if (j->type == REDIS_IOJOB_LOAD) {
redisDb *db;
if (job->canceled) continue; /* Skip this, already canceled. */
if (job->id == o) {
- redisLog(REDIS_DEBUG,"*** CANCELED %p (%s) (type %d) (LIST ID %d)\n",
- (void*)job, (char*)o->ptr, job->type, i);
+ redisLog(REDIS_DEBUG,"*** CANCELED %p (key %s) (type %d) (LIST ID %d)\n",
+ (void*)job, (char*)job->key->ptr, job->type, i);
/* Mark the pages as free since the swap didn't happened
* or happened but is now discarded. */
if (i != 1 && job->type == REDIS_IOJOB_DO_SWAP)
static int vmSwapObjectThreaded(robj *key, robj *val, redisDb *db) {
iojob *j;
- assert(key->storage == REDIS_VM_MEMORY);
-
j = zmalloc(sizeof(*j));
j->type = REDIS_IOJOB_PREPARE_SWAP;
j->db = db;
/* If the key does not exist or is already in RAM we don't need to
* block the client at all. */
- de = dictFind(c->db->dict,key);
+ de = dictFind(c->db->dict,key->ptr);
if (de == NULL) return 0;
o = dictGetEntryVal(de);
if (o->storage == REDIS_VM_MEMORY) {
* key exists, mark the client as dirty, as the key will be
* removed. */
if (dbid == -1 || wk->db->id == dbid) {
- if (dictFind(wk->db->dict, wk->key) != NULL)
+ if (dictFind(wk->db->dict, wk->key->ptr) != NULL)
c->flags |= REDIS_DIRTY_CAS;
}
}
/* Iterate this DB writing every entry */
while((de = dictNext(di)) != NULL) {
- robj *key, *o, *kcopy;
+ sds key;
+ robj *keyobj, *o;
time_t expiretime;
memset(digest,0,20); /* This key-val digest */
key = dictGetEntryKey(de);
+ keyobj = createStringObject(key,sdslen(key));
+
+ mixDigest(digest,key,sdslen(key));
+
+ /* Make sure the key is loaded if VM is active */
+ o = lookupKeyRead(db,keyobj);
- if (!server.vm_enabled) {
- mixObjectDigest(digest,key);
- o = dictGetEntryVal(de);
- } else {
- /* Don't work with the key directly as when VM is active
- * this is unsafe: TODO: fix decrRefCount to check if the
- * count really reached 0 to avoid this mess */
- kcopy = dupStringObject(key);
- mixObjectDigest(digest,kcopy);
- o = lookupKeyRead(db,kcopy);
- decrRefCount(kcopy);
- }
aux = htonl(o->type);
mixDigest(digest,&aux,sizeof(aux));
- expiretime = getExpire(db,key);
+ expiretime = getExpire(db,keyobj);
/* Save the key and associated value */
if (o->type == REDIS_STRING) {
mixObjectDigest(digest,o);
} else if (o->type == REDIS_LIST) {
- list *list = o->ptr;
- listNode *ln;
- listIter li;
-
- listRewind(list,&li);
- while((ln = listNext(&li))) {
- robj *eleobj = listNodeValue(ln);
-
+ listTypeIterator *li = listTypeInitIterator(o,0,REDIS_TAIL);
+ listTypeEntry entry;
+ while(listTypeNext(li,&entry)) {
+ robj *eleobj = listTypeGet(&entry);
mixObjectDigest(digest,eleobj);
+ decrRefCount(eleobj);
}
+ listTypeReleaseIterator(li);
} else if (o->type == REDIS_SET) {
dict *set = o->ptr;
dictIterator *di = dictGetIterator(set);
if (expiretime != -1) xorDigest(digest,"!!expire!!",10);
/* We can finally xor the key-val digest to the final digest */
xorDigest(final,digest,20);
+ decrRefCount(keyobj);
}
dictReleaseIterator(di);
}
redisLog(REDIS_WARNING,"Append Only File loaded by DEBUG LOADAOF");
addReply(c,shared.ok);
} else if (!strcasecmp(c->argv[1]->ptr,"object") && c->argc == 3) {
- dictEntry *de = dictFind(c->db->dict,c->argv[2]);
- robj *key, *val;
+ dictEntry *de = dictFind(c->db->dict,c->argv[2]->ptr);
+ robj *val;
if (!de) {
addReply(c,shared.nokeyerr);
return;
}
- key = dictGetEntryKey(de);
val = dictGetEntryVal(de);
if (!server.vm_enabled || (val->storage == REDIS_VM_MEMORY ||
val->storage == REDIS_VM_SWAPPING)) {
strenc = buf;
}
addReplySds(c,sdscatprintf(sdsempty(),
- "+Key at:%p refcount:%d, value at:%p refcount:%d "
+ "+Value at:%p refcount:%d "
"encoding:%s serializedlength:%lld\r\n",
- (void*)key, key->refcount, (void*)val, val->refcount,
+ (void*)val, val->refcount,
strenc, (long long) rdbSavedObjectLen(val,NULL)));
} else {
vmpointer *vp = (vmpointer*) val;
addReplySds(c,sdscatprintf(sdsempty(),
- "+Key at:%p refcount:%d, value swapped at: page %llu "
+ "+Value swapped at: page %llu "
"using %llu pages\r\n",
- (void*)key, key->refcount, (unsigned long long) vp->page,
+ (unsigned long long) vp->page,
(unsigned long long) vp->usedpages));
}
} else if (!strcasecmp(c->argv[1]->ptr,"swapin") && c->argc == 3) {
lookupKeyRead(c->db,c->argv[2]);
addReply(c,shared.ok);
} else if (!strcasecmp(c->argv[1]->ptr,"swapout") && c->argc == 3) {
- dictEntry *de = dictFind(c->db->dict,c->argv[2]);
- robj *key, *val;
+ dictEntry *de = dictFind(c->db->dict,c->argv[2]->ptr);
+ robj *val;
vmpointer *vp;
if (!server.vm_enabled) {
addReply(c,shared.nokeyerr);
return;
}
- key = dictGetEntryKey(de);
val = dictGetEntryVal(de);
/* Swap it */
if (val->storage != REDIS_VM_MEMORY) {
}
snprintf(buf,sizeof(buf),"value:%lu",j);
val = createStringObject(buf,strlen(buf));
- dictAdd(c->db->dict,key,val);
+ dbAdd(c->db,key,val);
+ decrRefCount(key);
}
addReply(c,shared.ok);
} else if (!strcasecmp(c->argv[1]->ptr,"digest") && c->argc == 2) {
projects / redis.git / blobdiff