Commit 846d8b3e authored by Pieter Noordhuis's avatar Pieter Noordhuis
Browse files

merge antirez/smallkeys

parents 178d6903 22194a7f
...@@ -87,7 +87,7 @@ staticsymbols: ...@@ -87,7 +87,7 @@ staticsymbols:
tclsh utils/build-static-symbols.tcl > staticsymbols.h tclsh utils/build-static-symbols.tcl > staticsymbols.h
test: test:
tclsh8.5 tests/test_helper.tcl tclsh8.5 tests/test_helper.tcl --tags "${TAGS}"
bench: bench:
./redis-benchmark ./redis-benchmark
......
...@@ -4,6 +4,7 @@ Redis TODO and Roadmap ...@@ -4,6 +4,7 @@ Redis TODO and Roadmap
VERSION 2.2 TODO (Optimizations and latency) VERSION 2.2 TODO (Optimizations and latency)
============================================ ============================================
* Support for syslog(3).
* Lower the CPU usage. * Lower the CPU usage.
* Lower the RAM usage everywhere possible. * Lower the RAM usage everywhere possible.
* Specially encoded Sets (like Hashes). * Specially encoded Sets (like Hashes).
......
...@@ -121,6 +121,7 @@ ...@@ -121,6 +121,7 @@
#define REDIS_SET 2 #define REDIS_SET 2
#define REDIS_ZSET 3 #define REDIS_ZSET 3
#define REDIS_HASH 4 #define REDIS_HASH 4
#define REDIS_VMPOINTER 8
/* Objects encoding. Some kind of objects like Strings and Hashes can be /* Objects encoding. Some kind of objects like Strings and Hashes can be
* internally represented in multiple ways. The 'encoding' field of the object * internally represented in multiple ways. The 'encoding' field of the object
...@@ -133,7 +134,7 @@ ...@@ -133,7 +134,7 @@
#define REDIS_ENCODING_ZIPLIST 5 /* Encoded as ziplist */ #define REDIS_ENCODING_ZIPLIST 5 /* Encoded as ziplist */
static char* strencoding[] = { static char* strencoding[] = {
"raw", "int", "zipmap", "hashtable" "raw", "int", "hashtable", "zipmap", "list", "ziplist"
}; };
/* Object types only used for dumping to disk */ /* Object types only used for dumping to disk */
...@@ -250,30 +251,41 @@ static void _redisPanic(char *msg, char *file, int line); ...@@ -250,30 +251,41 @@ static void _redisPanic(char *msg, char *file, int line);
/* A redis object, that is a type able to hold a string / list / set */ /* A redis object, that is a type able to hold a string / list / set */
/* The VM object structure */
struct redisObjectVM {
off_t page; /* the page at witch the object is stored on disk */
off_t usedpages; /* number of pages used on disk */
time_t atime; /* Last access time */
} vm;
/* The actual Redis Object */ /* The actual Redis Object */
typedef struct redisObject { typedef struct redisObject {
void *ptr; unsigned type:4;
unsigned char type; unsigned storage:2; /* REDIS_VM_MEMORY or REDIS_VM_SWAPPING */
unsigned char encoding; unsigned encoding:4;
unsigned char storage; /* If this object is a key, where is the value? unsigned lru:22; /* lru time (relative to server.lruclock) */
* REDIS_VM_MEMORY, REDIS_VM_SWAPPED, ... */
unsigned char vtype; /* If this object is a key, and value is swapped out,
* this is the type of the swapped out object. */
int refcount; int refcount;
void *ptr;
/* VM fields, this are only allocated if VM is active, otherwise the /* VM fields, this are only allocated if VM is active, otherwise the
* object allocation function will just allocate * object allocation function will just allocate
* sizeof(redisObjct) minus sizeof(redisObjectVM), so using * sizeof(redisObjct) minus sizeof(redisObjectVM), so using
* Redis without VM active will not have any overhead. */ * Redis without VM active will not have any overhead. */
struct redisObjectVM vm;
} robj; } robj;
/* The VM pointer structure - identifies an object in the swap file.
*
* This object is stored in place of the value
* object in the main key->value hash table representing a database.
* Note that the first fields (type, storage) are the same as the redisObject
* structure so that vmPointer strucuters can be accessed even when casted
* as redisObject structures.
*
* This is useful as we don't know if a value object is or not on disk, but we
* 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 {
unsigned type:4;
unsigned storage:2; /* REDIS_VM_SWAPPED or REDIS_VM_LOADING */
unsigned notused:26;
unsigned int vtype; /* type of the object stored in the swap file */
off_t page; /* the page at witch the object is stored on disk */
off_t usedpages; /* number of pages used on disk */
} vmpointer;
/* Macro used to initalize a Redis object allocated on the stack. /* Macro used to initalize a Redis object allocated on the stack.
* Note that this macro is taken near the structure definition to make sure * Note that this macro is taken near the structure definition to make sure
* we'll update it when the structure is changed, to avoid bugs like * we'll update it when the structure is changed, to avoid bugs like
...@@ -283,7 +295,7 @@ typedef struct redisObject { ...@@ -283,7 +295,7 @@ typedef struct redisObject {
_var.type = REDIS_STRING; \ _var.type = REDIS_STRING; \
_var.encoding = REDIS_ENCODING_RAW; \ _var.encoding = REDIS_ENCODING_RAW; \
_var.ptr = _ptr; \ _var.ptr = _ptr; \
if (server.vm_enabled) _var.storage = REDIS_VM_MEMORY; \ _var.storage = REDIS_VM_MEMORY; \
} while(0); } while(0);
typedef struct redisDb { typedef struct redisDb {
...@@ -372,6 +384,7 @@ struct redisServer { ...@@ -372,6 +384,7 @@ struct redisServer {
int daemonize; int daemonize;
int appendonly; int appendonly;
int appendfsync; int appendfsync;
int no_appendfsync_on_rewrite;
int shutdown_asap; int shutdown_asap;
time_t lastfsync; time_t lastfsync;
int appendfd; int appendfd;
...@@ -452,6 +465,8 @@ struct redisServer { ...@@ -452,6 +465,8 @@ struct redisServer {
list *pubsub_patterns; /* A list of pubsub_patterns */ list *pubsub_patterns; /* A list of pubsub_patterns */
/* Misc */ /* Misc */
FILE *devnull; FILE *devnull;
unsigned lruclock:22; /* clock incrementing every minute, for LRU */
unsigned lruclock_padding:10;
}; };
typedef struct pubsubPattern { typedef struct pubsubPattern {
...@@ -544,6 +559,9 @@ typedef struct iojob { ...@@ -544,6 +559,9 @@ typedef struct iojob {
int type; /* Request type, REDIS_IOJOB_* */ int type; /* Request type, REDIS_IOJOB_* */
redisDb *db;/* Redis database */ redisDb *db;/* Redis database */
robj *key; /* This I/O request is about swapping this key */ robj *key; /* This I/O request is about swapping this key */
robj *id; /* Unique identifier of this job:
this is the object to swap for REDIS_IOREQ_*_SWAP, or the
vmpointer objct for REDIS_IOREQ_LOAD. */
robj *val; /* the value to swap for REDIS_IOREQ_*_SWAP, otherwise this robj *val; /* the value to swap for REDIS_IOREQ_*_SWAP, otherwise this
* field is populated by the I/O thread for REDIS_IOREQ_LOAD. */ * field is populated by the I/O thread for REDIS_IOREQ_LOAD. */
off_t page; /* Swap page where to read/write the object */ off_t page; /* Swap page where to read/write the object */
...@@ -577,8 +595,7 @@ static robj *getDecodedObject(robj *o); ...@@ -577,8 +595,7 @@ static robj *getDecodedObject(robj *o);
static int removeExpire(redisDb *db, robj *key); static int removeExpire(redisDb *db, robj *key);
static int expireIfNeeded(redisDb *db, robj *key); static int expireIfNeeded(redisDb *db, robj *key);
static int deleteIfVolatile(redisDb *db, robj *key); static int deleteIfVolatile(redisDb *db, robj *key);
static int deleteIfSwapped(redisDb *db, robj *key); static int dbDelete(redisDb *db, robj *key);
static int deleteKey(redisDb *db, robj *key);
static time_t getExpire(redisDb *db, robj *key); static time_t getExpire(redisDb *db, robj *key);
static int setExpire(redisDb *db, robj *key, time_t when); static int setExpire(redisDb *db, robj *key, time_t when);
static void updateSlavesWaitingBgsave(int bgsaveerr); static void updateSlavesWaitingBgsave(int bgsaveerr);
...@@ -600,8 +617,8 @@ static void unblockClientWaitingData(redisClient *c); ...@@ -600,8 +617,8 @@ static void unblockClientWaitingData(redisClient *c);
static int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele); static int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele);
static void vmInit(void); static void vmInit(void);
static void vmMarkPagesFree(off_t page, off_t count); static void vmMarkPagesFree(off_t page, off_t count);
static robj *vmLoadObject(robj *key); static robj *vmLoadObject(robj *o);
static robj *vmPreviewObject(robj *key); static robj *vmPreviewObject(robj *o);
static int vmSwapOneObjectBlocking(void); static int vmSwapOneObjectBlocking(void);
static int vmSwapOneObjectThreaded(void); static int vmSwapOneObjectThreaded(void);
static int vmCanSwapOut(void); static int vmCanSwapOut(void);
...@@ -637,7 +654,7 @@ static int compareStringObjects(robj *a, robj *b); ...@@ -637,7 +654,7 @@ static int compareStringObjects(robj *a, robj *b);
static int equalStringObjects(robj *a, robj *b); static int equalStringObjects(robj *a, robj *b);
static void usage(); static void usage();
static int rewriteAppendOnlyFileBackground(void); static int rewriteAppendOnlyFileBackground(void);
static int vmSwapObjectBlocking(robj *key, robj *val); static vmpointer *vmSwapObjectBlocking(robj *val);
static int prepareForShutdown(); static int prepareForShutdown();
static void touchWatchedKey(redisDb *db, robj *key); static void touchWatchedKey(redisDb *db, robj *key);
static void touchWatchedKeysOnFlush(int dbid); static void touchWatchedKeysOnFlush(int dbid);
...@@ -1110,7 +1127,7 @@ static void dictListDestructor(void *privdata, void *val) ...@@ -1110,7 +1127,7 @@ static void dictListDestructor(void *privdata, void *val)
listRelease((list*)val); listRelease((list*)val);
} }
static int sdsDictKeyCompare(void *privdata, const void *key1, static int dictSdsKeyCompare(void *privdata, const void *key1,
const void *key2) const void *key2)
{ {
int l1,l2; int l1,l2;
...@@ -1130,11 +1147,18 @@ static void dictRedisObjectDestructor(void *privdata, void *val) ...@@ -1130,11 +1147,18 @@ static void dictRedisObjectDestructor(void *privdata, void *val)
decrRefCount(val); decrRefCount(val);
} }
static void dictSdsDestructor(void *privdata, void *val)
{
DICT_NOTUSED(privdata);
sdsfree(val);
}
static int dictObjKeyCompare(void *privdata, const void *key1, static int dictObjKeyCompare(void *privdata, const void *key1,
const void *key2) const void *key2)
{ {
const robj *o1 = key1, *o2 = 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) { static unsigned int dictObjHash(const void *key) {
...@@ -1142,6 +1166,10 @@ static unsigned int dictObjHash(const void *key) { ...@@ -1142,6 +1166,10 @@ static unsigned int dictObjHash(const void *key) {
return dictGenHashFunction(o->ptr, sdslen((sds)o->ptr)); 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, static int dictEncObjKeyCompare(void *privdata, const void *key1,
const void *key2) const void *key2)
{ {
...@@ -1154,7 +1182,7 @@ static int dictEncObjKeyCompare(void *privdata, const void *key1, ...@@ -1154,7 +1182,7 @@ static int dictEncObjKeyCompare(void *privdata, const void *key1,
o1 = getDecodedObject(o1); o1 = getDecodedObject(o1);
o2 = getDecodedObject(o2); o2 = getDecodedObject(o2);
cmp = sdsDictKeyCompare(privdata,o1->ptr,o2->ptr); cmp = dictSdsKeyCompare(privdata,o1->ptr,o2->ptr);
decrRefCount(o1); decrRefCount(o1);
decrRefCount(o2); decrRefCount(o2);
return cmp; return cmp;
...@@ -1183,7 +1211,7 @@ static unsigned int dictEncObjHash(const void *key) { ...@@ -1183,7 +1211,7 @@ static unsigned int dictEncObjHash(const void *key) {
} }
} }
/* Sets type and expires */ /* Sets type */
static dictType setDictType = { static dictType setDictType = {
dictEncObjHash, /* hash function */ dictEncObjHash, /* hash function */
NULL, /* key dup */ NULL, /* key dup */
...@@ -1203,23 +1231,23 @@ static dictType zsetDictType = { ...@@ -1203,23 +1231,23 @@ static dictType zsetDictType = {
dictVanillaFree /* val destructor of malloc(sizeof(double)) */ dictVanillaFree /* val destructor of malloc(sizeof(double)) */
}; };
/* Db->dict */ /* Db->dict, keys are sds strings, vals are Redis objects. */
static dictType dbDictType = { static dictType dbDictType = {
dictObjHash, /* hash function */ dictSdsHash, /* hash function */
NULL, /* key dup */ NULL, /* key dup */
NULL, /* val dup */ NULL, /* val dup */
dictObjKeyCompare, /* key compare */ dictSdsKeyCompare, /* key compare */
dictRedisObjectDestructor, /* key destructor */ dictSdsDestructor, /* key destructor */
dictRedisObjectDestructor /* val destructor */ dictRedisObjectDestructor /* val destructor */
}; };
/* Db->expires */ /* Db->expires */
static dictType keyptrDictType = { static dictType keyptrDictType = {
dictObjHash, /* hash function */ dictSdsHash, /* hash function */
NULL, /* key dup */ NULL, /* key dup */
NULL, /* val dup */ NULL, /* val dup */
dictObjKeyCompare, /* key compare */ dictSdsKeyCompare, /* key compare */
dictRedisObjectDestructor, /* key destructor */ dictSdsDestructor, /* key destructor */
NULL /* val destructor */ NULL /* val destructor */
}; };
...@@ -1388,7 +1416,7 @@ void backgroundRewriteDoneHandler(int statloc) { ...@@ -1388,7 +1416,7 @@ void backgroundRewriteDoneHandler(int statloc) {
/* If append only is actually enabled... */ /* If append only is actually enabled... */
close(server.appendfd); close(server.appendfd);
server.appendfd = fd; server.appendfd = fd;
fsync(fd); if (server.appendfsync != APPENDFSYNC_NO) aof_fsync(fd);
server.appendseldb = -1; /* Make sure it will issue SELECT */ server.appendseldb = -1; /* Make sure it will issue SELECT */
redisLog(REDIS_NOTICE,"The new append only file was selected for future appends."); redisLog(REDIS_NOTICE,"The new append only file was selected for future appends.");
} else { } else {
...@@ -1434,6 +1462,19 @@ static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientD ...@@ -1434,6 +1462,19 @@ static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientD
* in objects at every object access, and accuracy is not needed. * in objects at every object access, and accuracy is not needed.
* To access a global var is faster than calling time(NULL) */ * To access a global var is faster than calling time(NULL) */
server.unixtime = time(NULL); server.unixtime = time(NULL);
/* We have just 21 bits per object for LRU information.
* So we use an (eventually wrapping) LRU clock with minutes resolution.
*
* When we need to select what object to swap, we compute the minimum
* time distance between the current lruclock and the object last access
* lruclock info. Even if clocks will wrap on overflow, there is
* the interesting property that we are sure that at least
* ABS(A-B) minutes passed between current time and timestamp B.
*
* This is not precise but we don't need at all precision, but just
* something statistically reasonable.
*/
server.lruclock = (time(NULL)/60)&((1<<21)-1);
/* We received a SIGTERM, shutting down here in a safe way, as it is /* We received a SIGTERM, shutting down here in a safe way, as it is
* not ok doing so inside the signal handler. */ * not ok doing so inside the signal handler. */
...@@ -1532,7 +1573,11 @@ static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientD ...@@ -1532,7 +1573,11 @@ static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientD
if ((de = dictGetRandomKey(db->expires)) == NULL) break; if ((de = dictGetRandomKey(db->expires)) == NULL) break;
t = (time_t) dictGetEntryVal(de); t = (time_t) dictGetEntryVal(de);
if (now > t) { if (now > t) {
deleteKey(db,dictGetEntryKey(de)); sds key = dictGetEntryKey(de);
robj *keyobj = createStringObject(key,sdslen(key));
dbDelete(db,keyobj);
decrRefCount(keyobj);
expired++; expired++;
server.stat_expiredkeys++; server.stat_expiredkeys++;
} }
...@@ -1688,6 +1733,7 @@ static void initServerConfig() { ...@@ -1688,6 +1733,7 @@ static void initServerConfig() {
server.daemonize = 0; server.daemonize = 0;
server.appendonly = 0; server.appendonly = 0;
server.appendfsync = APPENDFSYNC_EVERYSEC; server.appendfsync = APPENDFSYNC_EVERYSEC;
server.no_appendfsync_on_rewrite = 0;
server.lastfsync = time(NULL); server.lastfsync = time(NULL);
server.appendfd = -1; server.appendfd = -1;
server.appendseldb = -1; /* Make sure the first time will not match */ server.appendseldb = -1; /* Make sure the first time will not match */
...@@ -1944,6 +1990,11 @@ static void loadServerConfig(char *filename) { ...@@ -1944,6 +1990,11 @@ static void loadServerConfig(char *filename) {
} else if (!strcasecmp(argv[0],"appendfilename") && argc == 2) { } else if (!strcasecmp(argv[0],"appendfilename") && argc == 2) {
zfree(server.appendfilename); zfree(server.appendfilename);
server.appendfilename = zstrdup(argv[1]); server.appendfilename = zstrdup(argv[1]);
} else if (!strcasecmp(argv[0],"no-appendfsync-on-rewrite")
&& argc == 2) {
if ((server.no_appendfsync_on_rewrite= yesnotoi(argv[1])) == -1) {
err = "argument must be 'yes' or 'no'"; goto loaderr;
}
} else if (!strcasecmp(argv[0],"appendfsync") && argc == 2) { } else if (!strcasecmp(argv[0],"appendfsync") && argc == 2) {
if (!strcasecmp(argv[1],"no")) { if (!strcasecmp(argv[1],"no")) {
server.appendfsync = APPENDFSYNC_NO; server.appendfsync = APPENDFSYNC_NO;
...@@ -2875,6 +2926,12 @@ static void addReplyBulk(redisClient *c, robj *obj) { ...@@ -2875,6 +2926,12 @@ static void addReplyBulk(redisClient *c, robj *obj) {
addReply(c,shared.crlf); 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 */ /* In the CONFIG command we need to add vanilla C string as bulk replies */
static void addReplyBulkCString(redisClient *c, char *s) { static void addReplyBulkCString(redisClient *c, char *s) {
if (s == NULL) { if (s == NULL) {
...@@ -2934,12 +2991,9 @@ static robj *createObject(int type, void *ptr) { ...@@ -2934,12 +2991,9 @@ static robj *createObject(int type, void *ptr) {
listDelNode(server.objfreelist,head); listDelNode(server.objfreelist,head);
if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex); if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex);
} else { } else {
if (server.vm_enabled) { if (server.vm_enabled)
pthread_mutex_unlock(&server.obj_freelist_mutex); pthread_mutex_unlock(&server.obj_freelist_mutex);
o = zmalloc(sizeof(*o)); o = zmalloc(sizeof(*o));
} else {
o = zmalloc(sizeof(*o)-sizeof(struct redisObjectVM));
}
} }
o->type = type; o->type = type;
o->encoding = REDIS_ENCODING_RAW; o->encoding = REDIS_ENCODING_RAW;
...@@ -2947,10 +3001,10 @@ static robj *createObject(int type, void *ptr) { ...@@ -2947,10 +3001,10 @@ static robj *createObject(int type, void *ptr) {
o->refcount = 1; o->refcount = 1;
if (server.vm_enabled) { if (server.vm_enabled) {
/* Note that this code may run in the context of an I/O thread /* Note that this code may run in the context of an I/O thread
* and accessing to server.unixtime in theory is an error * and accessing server.lruclock in theory is an error
* (no locks). But in practice this is safe, and even if we read * (no locks). But in practice this is safe, and even if we read
* garbage Redis will not fail, as it's just a statistical info */ * garbage Redis will not fail. */
o->vm.atime = server.unixtime; o->lru = server.lruclock;
o->storage = REDIS_VM_MEMORY; o->storage = REDIS_VM_MEMORY;
} }
return o; return o;
...@@ -3072,28 +3126,31 @@ static void incrRefCount(robj *o) { ...@@ -3072,28 +3126,31 @@ static void incrRefCount(robj *o) {
static void decrRefCount(void *obj) { static void decrRefCount(void *obj) {
robj *o = obj; robj *o = obj;
if (o->refcount <= 0) redisPanic("decrRefCount against refcount <= 0"); /* Object is a swapped out value, or in the process of being loaded. */
/* Object is a key of a swapped out value, or in the process of being
* loaded. */
if (server.vm_enabled && if (server.vm_enabled &&
(o->storage == REDIS_VM_SWAPPED || o->storage == REDIS_VM_LOADING)) (o->storage == REDIS_VM_SWAPPED || o->storage == REDIS_VM_LOADING))
{ {
if (o->storage == REDIS_VM_LOADING) vmCancelThreadedIOJob(obj); vmpointer *vp = obj;
redisAssert(o->type == REDIS_STRING); if (o->storage == REDIS_VM_LOADING) vmCancelThreadedIOJob(o);
freeStringObject(o); vmMarkPagesFree(vp->page,vp->usedpages);
vmMarkPagesFree(o->vm.page,o->vm.usedpages);
pthread_mutex_lock(&server.obj_freelist_mutex);
if (listLength(server.objfreelist) > REDIS_OBJFREELIST_MAX ||
!listAddNodeHead(server.objfreelist,o))
zfree(o);
pthread_mutex_unlock(&server.obj_freelist_mutex);
server.vm_stats_swapped_objects--; server.vm_stats_swapped_objects--;
zfree(vp);
return; return;
} }
/* Object is in memory, or in the process of being swapped out. */
if (--(o->refcount) == 0) { if (o->refcount <= 0) redisPanic("decrRefCount against refcount <= 0");
/* Object is in memory, or in the process of being swapped out.
*
* If the object is being swapped out, abort the operation on
* decrRefCount even if the refcount does not drop to 0: the object
* is referenced at least two times, as value of the key AND as
* job->val in the iojob. So if we don't invalidate the iojob, when it is
* done but the relevant key was removed in the meantime, the
* complete jobs handler will not find the key about the job and the
* assert will fail. */
if (server.vm_enabled && o->storage == REDIS_VM_SWAPPING) if (server.vm_enabled && o->storage == REDIS_VM_SWAPPING)
vmCancelThreadedIOJob(obj); vmCancelThreadedIOJob(o);
if (--(o->refcount) == 0) {
switch(o->type) { switch(o->type) {
case REDIS_STRING: freeStringObject(o); break; case REDIS_STRING: freeStringObject(o); break;
case REDIS_LIST: freeListObject(o); break; case REDIS_LIST: freeListObject(o); break;
...@@ -3110,64 +3167,6 @@ static void decrRefCount(void *obj) { ...@@ -3110,64 +3167,6 @@ static void decrRefCount(void *obj) {
} }
} }
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 (key->storage == REDIS_VM_MEMORY ||
key->storage == REDIS_VM_SWAPPING)
{
/* If we were swapping the object out, stop it, this key
* was requested. */
if (key->storage == REDIS_VM_SWAPPING)
vmCancelThreadedIOJob(key);
/* Update the access time of the key for the aging algorithm. */
key->vm.atime = server.unixtime;
} else {
int notify = (key->storage == REDIS_VM_LOADING);
/* Our value was swapped on disk. Bring it at home. */
redisAssert(val == NULL);
val = vmLoadObject(key);
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) { static int checkType(redisClient *c, robj *o, int type) {
if (o->type != type) { if (o->type != type) {
addReply(c,shared.wrongtypeerr); addReply(c,shared.wrongtypeerr);
...@@ -3176,21 +3175,6 @@ static int checkType(redisClient *c, robj *o, int type) { ...@@ -3176,21 +3175,6 @@ static int checkType(redisClient *c, robj *o, int type) {
return 0; 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 /* 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 * (that is, is a number that fits into long without any other space or
* character before or after the digits). * character before or after the digits).
...@@ -3410,6 +3394,134 @@ static int getLongFromObjectOrReply(redisClient *c, robj *o, long *target, const ...@@ -3410,6 +3394,134 @@ static int getLongFromObjectOrReply(redisClient *c, robj *o, long *target, const
return REDIS_OK; 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 =========================== */ /*============================ RDB saving/loading =========================== */
static int rdbSaveType(FILE *fp, unsigned char type) { static int rdbSaveType(FILE *fp, unsigned char type) {
...@@ -3779,9 +3891,12 @@ static int rdbSave(char *filename) { ...@@ -3779,9 +3891,12 @@ static int rdbSave(char *filename) {
/* Iterate this DB writing every entry */ /* Iterate this DB writing every entry */
while((de = dictNext(di)) != NULL) { while((de = dictNext(di)) != NULL) {
robj *key = dictGetEntryKey(de); sds keystr = dictGetEntryKey(de);
robj *o = dictGetEntryVal(de); robj key, *o = dictGetEntryVal(de);
time_t expiretime = getExpire(db,key); time_t expiretime;
initStaticStringObject(key,keystr);
expiretime = getExpire(db,&key);
/* Save the expire time */ /* Save the expire time */
if (expiretime != -1) { if (expiretime != -1) {
...@@ -3792,20 +3907,20 @@ static int rdbSave(char *filename) { ...@@ -3792,20 +3907,20 @@ static int rdbSave(char *filename) {
} }
/* Save the key and associated value. This requires special /* Save the key and associated value. This requires special
* handling if the value is swapped out. */ * handling if the value is swapped out. */
if (!server.vm_enabled || key->storage == REDIS_VM_MEMORY || if (!server.vm_enabled || o->storage == REDIS_VM_MEMORY ||
key->storage == REDIS_VM_SWAPPING) { o->storage == REDIS_VM_SWAPPING) {
/* Save type, key, value */ /* Save type, key, value */
if (rdbSaveType(fp,o->type) == -1) goto werr; 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; if (rdbSaveObject(fp,o) == -1) goto werr;
} else { } else {
/* REDIS_VM_SWAPPED or REDIS_VM_LOADING */ /* REDIS_VM_SWAPPED or REDIS_VM_LOADING */
robj *po; robj *po;
/* Get a preview of the object in memory */ /* Get a preview of the object in memory */
po = vmPreviewObject(key); po = vmPreviewObject(o);
/* Save type, key, value */ /* Save type, key, value */
if (rdbSaveType(fp,key->vtype) == -1) goto werr; 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; if (rdbSaveObject(fp,po) == -1) goto werr;
/* Remove the loaded object from memory */ /* Remove the loaded object from memory */
decrRefCount(po); decrRefCount(po);
...@@ -4139,11 +4254,9 @@ static int rdbLoad(char *filename) { ...@@ -4139,11 +4254,9 @@ static int rdbLoad(char *filename) {
uint32_t dbid; uint32_t dbid;
int type, retval, rdbver; int type, retval, rdbver;
int swap_all_values = 0; int swap_all_values = 0;
dict *d = server.db[0].dict;
redisDb *db = server.db+0; redisDb *db = server.db+0;
char buf[1024]; char buf[1024];
time_t expiretime, now = time(NULL); time_t expiretime, now = time(NULL);
long long loadedkeys = 0;
fp = fopen(filename,"r"); fp = fopen(filename,"r");
if (!fp) return REDIS_ERR; if (!fp) return REDIS_ERR;
...@@ -4162,6 +4275,7 @@ static int rdbLoad(char *filename) { ...@@ -4162,6 +4275,7 @@ static int rdbLoad(char *filename) {
} }
while(1) { while(1) {
robj *key, *val; robj *key, *val;
int force_swapout;
expiretime = -1; expiretime = -1;
/* Read type. */ /* Read type. */
...@@ -4181,7 +4295,6 @@ static int rdbLoad(char *filename) { ...@@ -4181,7 +4295,6 @@ static int rdbLoad(char *filename) {
exit(1); exit(1);
} }
db = server.db+dbid; db = server.db+dbid;
d = db->dict;
continue; continue;
} }
/* Read key */ /* Read key */
...@@ -4195,12 +4308,11 @@ static int rdbLoad(char *filename) { ...@@ -4195,12 +4308,11 @@ static int rdbLoad(char *filename) {
continue; continue;
} }
/* Add the new object in the hash table */ /* Add the new object in the hash table */
retval = dictAdd(d,key,val); retval = dbAdd(db,key,val);
if (retval == DICT_ERR) { if (retval == REDIS_ERR) {
redisLog(REDIS_WARNING,"Loading DB, duplicated key (%s) found! Unrecoverable error, exiting now.", key->ptr); redisLog(REDIS_WARNING,"Loading DB, duplicated key (%s) found! Unrecoverable error, exiting now.", key->ptr);
exit(1); exit(1);
} }
loadedkeys++;
/* Set the expire time if needed */ /* Set the expire time if needed */
if (expiretime != -1) setExpire(db,key,expiretime); if (expiretime != -1) setExpire(db,key,expiretime);
...@@ -4212,23 +4324,30 @@ static int rdbLoad(char *filename) { ...@@ -4212,23 +4324,30 @@ static int rdbLoad(char *filename) {
* to random sampling, otherwise we may try to swap already * to random sampling, otherwise we may try to swap already
* swapped keys. */ * swapped keys. */
if (swap_all_values) { 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 */ /* de may be NULL since the key already expired */
if (de) { if (de) {
key = dictGetEntryKey(de); vmpointer *vp;
val = dictGetEntryVal(de); val = dictGetEntryVal(de);
if (vmSwapObjectBlocking(key,val) == REDIS_OK) { if (val->refcount == 1 &&
dictGetEntryVal(de) = NULL; (vp = vmSwapObjectBlocking(val)) != NULL)
} dictGetEntryVal(de) = vp;
} }
decrRefCount(key);
continue; 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 /* If we have still some hope of having some value fitting memory
* then we try random sampling. */ * 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) { while (zmalloc_used_memory() > server.vm_max_memory) {
if (vmSwapOneObjectBlocking() == REDIS_ERR) break; if (vmSwapOneObjectBlocking() == REDIS_ERR) break;
} }
...@@ -4258,7 +4377,7 @@ static int prepareForShutdown() { ...@@ -4258,7 +4377,7 @@ static int prepareForShutdown() {
} }
if (server.appendonly) { if (server.appendonly) {
/* Append only file: fsync() the AOF and exit */ /* Append only file: fsync() the AOF and exit */
fsync(server.appendfd); aof_fsync(server.appendfd);
if (server.vm_enabled) unlink(server.vm_swap_file); if (server.vm_enabled) unlink(server.vm_swap_file);
} else { } else {
/* Snapshotting. Perform a SYNC SAVE and exit */ /* Snapshotting. Perform a SYNC SAVE and exit */
...@@ -4317,23 +4436,16 @@ static void setGenericCommand(redisClient *c, int nx, robj *key, robj *val, robj ...@@ -4317,23 +4436,16 @@ static void setGenericCommand(redisClient *c, int nx, robj *key, robj *val, robj
touchWatchedKey(c->db,key); touchWatchedKey(c->db,key);
if (nx) deleteIfVolatile(c->db,key); if (nx) deleteIfVolatile(c->db,key);
retval = dictAdd(c->db->dict,key,val); retval = dbAdd(c->db,key,val);
if (retval == DICT_ERR) { if (retval == REDIS_ERR) {
if (!nx) { if (!nx) {
/* If the key is about a swapped value, we want a new key object dbReplace(c->db,key,val);
* 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);
incrRefCount(val); incrRefCount(val);
} else { } else {
addReply(c,shared.czero); addReply(c,shared.czero);
return; return;
} }
} else { } else {
incrRefCount(key);
incrRefCount(val); incrRefCount(val);
} }
server.dirty++; server.dirty++;
...@@ -4375,11 +4487,7 @@ static void getCommand(redisClient *c) { ...@@ -4375,11 +4487,7 @@ static void getCommand(redisClient *c) {
static void getsetCommand(redisClient *c) { static void getsetCommand(redisClient *c) {
if (getGenericCommand(c) == REDIS_ERR) return; if (getGenericCommand(c) == REDIS_ERR) return;
if (dictAdd(c->db->dict,c->argv[1],c->argv[2]) == DICT_ERR) { dbReplace(c->db,c->argv[1],c->argv[2]);
dictReplace(c->db->dict,c->argv[1],c->argv[2]);
} else {
incrRefCount(c->argv[1]);
}
incrRefCount(c->argv[2]); incrRefCount(c->argv[2]);
server.dirty++; server.dirty++;
removeExpire(c->db,c->argv[1]); removeExpire(c->db,c->argv[1]);
...@@ -4425,17 +4533,9 @@ static void msetGenericCommand(redisClient *c, int nx) { ...@@ -4425,17 +4533,9 @@ static void msetGenericCommand(redisClient *c, int nx) {
} }
for (j = 1; j < c->argc; j += 2) { for (j = 1; j < c->argc; j += 2) {
int retval;
c->argv[j+1] = tryObjectEncoding(c->argv[j+1]); c->argv[j+1] = tryObjectEncoding(c->argv[j+1]);
retval = dictAdd(c->db->dict,c->argv[j],c->argv[j+1]); dbReplace(c->db,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]); incrRefCount(c->argv[j+1]);
} else {
incrRefCount(c->argv[j]);
incrRefCount(c->argv[j+1]);
}
removeExpire(c->db,c->argv[j]); removeExpire(c->db,c->argv[j]);
} }
server.dirty += (c->argc-1)/2; server.dirty += (c->argc-1)/2;
...@@ -4452,7 +4552,6 @@ static void msetnxCommand(redisClient *c) { ...@@ -4452,7 +4552,6 @@ static void msetnxCommand(redisClient *c) {
static void incrDecrCommand(redisClient *c, long long incr) { static void incrDecrCommand(redisClient *c, long long incr) {
long long value; long long value;
int retval;
robj *o; robj *o;
o = lookupKeyWrite(c->db,c->argv[1]); o = lookupKeyWrite(c->db,c->argv[1]);
...@@ -4461,13 +4560,7 @@ static void incrDecrCommand(redisClient *c, long long incr) { ...@@ -4461,13 +4560,7 @@ static void incrDecrCommand(redisClient *c, long long incr) {
value += incr; value += incr;
o = createStringObjectFromLongLong(value); o = createStringObjectFromLongLong(value);
retval = dictAdd(c->db->dict,c->argv[1],o); dbReplace(c->db,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]);
}
server.dirty++; server.dirty++;
addReply(c,shared.colon); addReply(c,shared.colon);
addReply(c,o); addReply(c,o);
...@@ -4504,17 +4597,10 @@ static void appendCommand(redisClient *c) { ...@@ -4504,17 +4597,10 @@ static void appendCommand(redisClient *c) {
o = lookupKeyWrite(c->db,c->argv[1]); o = lookupKeyWrite(c->db,c->argv[1]);
if (o == NULL) { if (o == NULL) {
/* Create the key */ /* Create the key */
retval = dictAdd(c->db->dict,c->argv[1],c->argv[2]); retval = dbAdd(c->db,c->argv[1],c->argv[2]);
incrRefCount(c->argv[1]);
incrRefCount(c->argv[2]); incrRefCount(c->argv[2]);
totlen = stringObjectLen(c->argv[2]); totlen = stringObjectLen(c->argv[2]);
} else { } else {
dictEntry *de;
de = dictFind(c->db->dict,c->argv[1]);
assert(de != NULL);
o = dictGetEntryVal(de);
if (o->type != REDIS_STRING) { if (o->type != REDIS_STRING) {
addReply(c,shared.wrongtypeerr); addReply(c,shared.wrongtypeerr);
return; return;
...@@ -4526,7 +4612,7 @@ static void appendCommand(redisClient *c) { ...@@ -4526,7 +4612,7 @@ static void appendCommand(redisClient *c) {
o = createStringObject(decoded->ptr, sdslen(decoded->ptr)); o = createStringObject(decoded->ptr, sdslen(decoded->ptr));
decrRefCount(decoded); decrRefCount(decoded);
dictReplace(c->db->dict,c->argv[1],o); dbReplace(c->db,c->argv[1],o);
} }
/* APPEND! */ /* APPEND! */
if (c->argv[2]->encoding == REDIS_ENCODING_RAW) { if (c->argv[2]->encoding == REDIS_ENCODING_RAW) {
...@@ -4585,7 +4671,7 @@ static void delCommand(redisClient *c) { ...@@ -4585,7 +4671,7 @@ static void delCommand(redisClient *c) {
int deleted = 0, j; int deleted = 0, j;
for (j = 1; j < c->argc; 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]); touchWatchedKey(c->db,c->argv[j]);
server.dirty++; server.dirty++;
deleted++; deleted++;
...@@ -4596,7 +4682,7 @@ static void delCommand(redisClient *c) { ...@@ -4596,7 +4682,7 @@ static void delCommand(redisClient *c) {
static void existsCommand(redisClient *c) { static void existsCommand(redisClient *c) {
expireIfNeeded(c->db,c->argv[1]); 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); addReply(c, shared.cone);
} else { } else {
addReply(c, shared.czero); addReply(c, shared.czero);
...@@ -4614,27 +4700,15 @@ static void selectCommand(redisClient *c) { ...@@ -4614,27 +4700,15 @@ static void selectCommand(redisClient *c) {
} }
static void randomkeyCommand(redisClient *c) { static void randomkeyCommand(redisClient *c) {
dictEntry *de;
robj *key; robj *key;
while(1) { if ((key = dbRandomKey(c->db)) == NULL) {
de = dictGetRandomKey(c->db->dict);
if (!de || expireIfNeeded(c->db,dictGetEntryKey(de)) == 0) break;
}
if (de == NULL) {
addReply(c,shared.nullbulk); addReply(c,shared.nullbulk);
return; return;
} }
key = dictGetEntryKey(de);
if (server.vm_enabled) {
key = dupStringObject(key);
addReplyBulk(c,key); addReplyBulk(c,key);
decrRefCount(key); decrRefCount(key);
} else {
addReplyBulk(c,key);
}
} }
static void keysCommand(redisClient *c) { static void keysCommand(redisClient *c) {
...@@ -4649,15 +4723,17 @@ static void keysCommand(redisClient *c) { ...@@ -4649,15 +4723,17 @@ static void keysCommand(redisClient *c) {
addReply(c,lenobj); addReply(c,lenobj);
decrRefCount(lenobj); decrRefCount(lenobj);
while((de = dictNext(di)) != NULL) { 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') || if ((pattern[0] == '*' && pattern[1] == '\0') ||
stringmatchlen(pattern,plen,key,sdslen(key),0)) { stringmatchlen(pattern,plen,key,sdslen(key),0)) {
keyobj = createStringObject(key,sdslen(key));
if (expireIfNeeded(c->db,keyobj) == 0) { if (expireIfNeeded(c->db,keyobj) == 0) {
addReplyBulk(c,keyobj); addReplyBulk(c,keyobj);
numkeys++; numkeys++;
} }
decrRefCount(keyobj);
} }
} }
dictReleaseIterator(di); dictReleaseIterator(di);
...@@ -4740,17 +4816,15 @@ static void renameGenericCommand(redisClient *c, int nx) { ...@@ -4740,17 +4816,15 @@ static void renameGenericCommand(redisClient *c, int nx) {
incrRefCount(o); incrRefCount(o);
deleteIfVolatile(c->db,c->argv[2]); deleteIfVolatile(c->db,c->argv[2]);
if (dictAdd(c->db->dict,c->argv[2],o) == DICT_ERR) { if (dbAdd(c->db,c->argv[2],o) == REDIS_ERR) {
if (nx) { if (nx) {
decrRefCount(o); decrRefCount(o);
addReply(c,shared.czero); addReply(c,shared.czero);
return; return;
} }
dictReplace(c->db->dict,c->argv[2],o); dbReplace(c->db,c->argv[2],o);
} else {
incrRefCount(c->argv[2]);
} }
deleteKey(c->db,c->argv[1]); dbDelete(c->db,c->argv[1]);
touchWatchedKey(c->db,c->argv[2]); touchWatchedKey(c->db,c->argv[2]);
server.dirty++; server.dirty++;
addReply(c,nx ? shared.cone : shared.ok); addReply(c,nx ? shared.cone : shared.ok);
...@@ -4795,15 +4869,14 @@ static void moveCommand(redisClient *c) { ...@@ -4795,15 +4869,14 @@ static void moveCommand(redisClient *c) {
/* Try to add the element to the target DB */ /* Try to add the element to the target DB */
deleteIfVolatile(dst,c->argv[1]); deleteIfVolatile(dst,c->argv[1]);
if (dictAdd(dst->dict,c->argv[1],o) == DICT_ERR) { if (dbAdd(dst,c->argv[1],o) == REDIS_ERR) {
addReply(c,shared.czero); addReply(c,shared.czero);
return; return;
} }
incrRefCount(c->argv[1]);
incrRefCount(o); incrRefCount(o);
/* OK! key moved, free the entry in the source DB */ /* OK! key moved, free the entry in the source DB */
deleteKey(src,c->argv[1]); dbDelete(src,c->argv[1]);
server.dirty++; server.dirty++;
addReply(c,shared.cone); addReply(c,shared.cone);
} }
...@@ -5012,8 +5085,7 @@ static void pushGenericCommand(redisClient *c, int where) { ...@@ -5012,8 +5085,7 @@ static void pushGenericCommand(redisClient *c, int where) {
return; return;
} }
lobj = createZiplistObject(); lobj = createZiplistObject();
dictAdd(c->db->dict,c->argv[1],lobj); dbAdd(c->db,c->argv[1],lobj);
incrRefCount(c->argv[1]);
} else { } else {
if (lobj->type != REDIS_LIST) { if (lobj->type != REDIS_LIST) {
addReply(c,shared.wrongtypeerr); addReply(c,shared.wrongtypeerr);
...@@ -5124,7 +5196,7 @@ static void popGenericCommand(redisClient *c, int where) { ...@@ -5124,7 +5196,7 @@ static void popGenericCommand(redisClient *c, int where) {
} else { } else {
addReplyBulk(c,value); addReplyBulk(c,value);
decrRefCount(value); decrRefCount(value);
if (lLength(o) == 0) deleteKey(c->db,c->argv[1]); if (lLength(o) == 0) dbDelete(c->db,c->argv[1]);
server.dirty++; server.dirty++;
} }
} }
...@@ -5223,7 +5295,7 @@ static void ltrimCommand(redisClient *c) { ...@@ -5223,7 +5295,7 @@ static void ltrimCommand(redisClient *c) {
} else { } else {
redisPanic("Unknown list encoding"); redisPanic("Unknown list encoding");
} }
if (lLength(o) == 0) deleteKey(c->db,c->argv[1]); if (lLength(o) == 0) dbDelete(c->db,c->argv[1]);
server.dirty++; server.dirty++;
addReply(c,shared.ok); addReply(c,shared.ok);
} }
...@@ -5263,7 +5335,7 @@ static void lremCommand(redisClient *c) { ...@@ -5263,7 +5335,7 @@ static void lremCommand(redisClient *c) {
if (subject->encoding == REDIS_ENCODING_ZIPLIST) if (subject->encoding == REDIS_ENCODING_ZIPLIST)
decrRefCount(obj); decrRefCount(obj);
if (lLength(subject) == 0) deleteKey(c->db,c->argv[1]); if (lLength(subject) == 0) dbDelete(c->db,c->argv[1]);
addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",removed)); addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",removed));
} }
...@@ -5300,8 +5372,7 @@ static void rpoplpushcommand(redisClient *c) { ...@@ -5300,8 +5372,7 @@ static void rpoplpushcommand(redisClient *c) {
/* Create the list if the key does not exist */ /* Create the list if the key does not exist */
if (!dobj) { if (!dobj) {
dobj = createZiplistObject(); dobj = createZiplistObject();
dictAdd(c->db->dict,c->argv[2],dobj); dbAdd(c->db,c->argv[2],dobj);
incrRefCount(c->argv[2]);
} }
lPush(dobj,value,REDIS_HEAD); lPush(dobj,value,REDIS_HEAD);
} }
...@@ -5313,7 +5384,7 @@ static void rpoplpushcommand(redisClient *c) { ...@@ -5313,7 +5384,7 @@ static void rpoplpushcommand(redisClient *c) {
decrRefCount(value); decrRefCount(value);
/* Delete the source list when it is empty */ /* Delete the source list when it is empty */
if (lLength(sobj) == 0) deleteKey(c->db,c->argv[1]); if (lLength(sobj) == 0) dbDelete(c->db,c->argv[1]);
server.dirty++; server.dirty++;
} }
} }
...@@ -5326,8 +5397,7 @@ static void saddCommand(redisClient *c) { ...@@ -5326,8 +5397,7 @@ static void saddCommand(redisClient *c) {
set = lookupKeyWrite(c->db,c->argv[1]); set = lookupKeyWrite(c->db,c->argv[1]);
if (set == NULL) { if (set == NULL) {
set = createSetObject(); set = createSetObject();
dictAdd(c->db->dict,c->argv[1],set); dbAdd(c->db,c->argv[1],set);
incrRefCount(c->argv[1]);
} else { } else {
if (set->type != REDIS_SET) { if (set->type != REDIS_SET) {
addReply(c,shared.wrongtypeerr); addReply(c,shared.wrongtypeerr);
...@@ -5352,7 +5422,7 @@ static void sremCommand(redisClient *c) { ...@@ -5352,7 +5422,7 @@ static void sremCommand(redisClient *c) {
if (dictDelete(set->ptr,c->argv[2]) == DICT_OK) { if (dictDelete(set->ptr,c->argv[2]) == DICT_OK) {
server.dirty++; server.dirty++;
if (htNeedsResize(set->ptr)) dictResize(set->ptr); 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); addReply(c,shared.cone);
} else { } else {
addReply(c,shared.czero); addReply(c,shared.czero);
...@@ -5383,13 +5453,12 @@ static void smoveCommand(redisClient *c) { ...@@ -5383,13 +5453,12 @@ static void smoveCommand(redisClient *c) {
return; return;
} }
if (dictSize((dict*)srcset->ptr) == 0 && srcset != dstset) if (dictSize((dict*)srcset->ptr) == 0 && srcset != dstset)
deleteKey(c->db,c->argv[1]); dbDelete(c->db,c->argv[1]);
server.dirty++; server.dirty++;
/* Add the element to the destination set */ /* Add the element to the destination set */
if (!dstset) { if (!dstset) {
dstset = createSetObject(); dstset = createSetObject();
dictAdd(c->db->dict,c->argv[2],dstset); dbAdd(c->db,c->argv[2],dstset);
incrRefCount(c->argv[2]);
} }
if (dictAdd(dstset->ptr,c->argv[3],NULL) == DICT_OK) if (dictAdd(dstset->ptr,c->argv[3],NULL) == DICT_OK)
incrRefCount(c->argv[3]); incrRefCount(c->argv[3]);
...@@ -5435,7 +5504,7 @@ static void spopCommand(redisClient *c) { ...@@ -5435,7 +5504,7 @@ static void spopCommand(redisClient *c) {
addReplyBulk(c,ele); addReplyBulk(c,ele);
dictDelete(set->ptr,ele); dictDelete(set->ptr,ele);
if (htNeedsResize(set->ptr)) dictResize(set->ptr); 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++; server.dirty++;
} }
} }
...@@ -5479,7 +5548,7 @@ static void sinterGenericCommand(redisClient *c, robj **setskeys, unsigned long ...@@ -5479,7 +5548,7 @@ static void sinterGenericCommand(redisClient *c, robj **setskeys, unsigned long
if (!setobj) { if (!setobj) {
zfree(dv); zfree(dv);
if (dstkey) { if (dstkey) {
if (deleteKey(c->db,dstkey)) if (dbDelete(c->db,dstkey))
server.dirty++; server.dirty++;
addReply(c,shared.czero); addReply(c,shared.czero);
} else { } else {
...@@ -5539,10 +5608,9 @@ static void sinterGenericCommand(redisClient *c, robj **setskeys, unsigned long ...@@ -5539,10 +5608,9 @@ static void sinterGenericCommand(redisClient *c, robj **setskeys, unsigned long
if (dstkey) { if (dstkey) {
/* Store the resulting set into the target, if the intersection /* Store the resulting set into the target, if the intersection
* is not an empty set. */ * is not an empty set. */
deleteKey(c->db,dstkey); dbDelete(c->db,dstkey);
if (dictSize((dict*)dstset->ptr) > 0) { if (dictSize((dict*)dstset->ptr) > 0) {
dictAdd(c->db->dict,dstkey,dstset); dbAdd(c->db,dstkey,dstset);
incrRefCount(dstkey);
addReplyLongLong(c,dictSize((dict*)dstset->ptr)); addReplyLongLong(c,dictSize((dict*)dstset->ptr));
} else { } else {
decrRefCount(dstset); decrRefCount(dstset);
...@@ -5642,10 +5710,9 @@ static void sunionDiffGenericCommand(redisClient *c, robj **setskeys, int setsnu ...@@ -5642,10 +5710,9 @@ static void sunionDiffGenericCommand(redisClient *c, robj **setskeys, int setsnu
} else { } else {
/* If we have a target key where to store the resulting set /* If we have a target key where to store the resulting set
* create this key with the result set inside */ * create this key with the result set inside */
deleteKey(c->db,dstkey); dbDelete(c->db,dstkey);
if (dictSize((dict*)dstset->ptr) > 0) { if (dictSize((dict*)dstset->ptr) > 0) {
dictAdd(c->db->dict,dstkey,dstset); dbAdd(c->db,dstkey,dstset);
incrRefCount(dstkey);
addReplyLongLong(c,dictSize((dict*)dstset->ptr)); addReplyLongLong(c,dictSize((dict*)dstset->ptr));
} else { } else {
decrRefCount(dstset); decrRefCount(dstset);
...@@ -6001,8 +6068,7 @@ static void zaddGenericCommand(redisClient *c, robj *key, robj *ele, double scor ...@@ -6001,8 +6068,7 @@ static void zaddGenericCommand(redisClient *c, robj *key, robj *ele, double scor
zsetobj = lookupKeyWrite(c->db,key); zsetobj = lookupKeyWrite(c->db,key);
if (zsetobj == NULL) { if (zsetobj == NULL) {
zsetobj = createZsetObject(); zsetobj = createZsetObject();
dictAdd(c->db->dict,key,zsetobj); dbAdd(c->db,key,zsetobj);
incrRefCount(key);
} else { } else {
if (zsetobj->type != REDIS_ZSET) { if (zsetobj->type != REDIS_ZSET) {
addReply(c,shared.wrongtypeerr); addReply(c,shared.wrongtypeerr);
...@@ -6118,7 +6184,7 @@ static void zremCommand(redisClient *c) { ...@@ -6118,7 +6184,7 @@ static void zremCommand(redisClient *c) {
/* Delete from the hash table */ /* Delete from the hash table */
dictDelete(zs->dict,c->argv[2]); dictDelete(zs->dict,c->argv[2]);
if (htNeedsResize(zs->dict)) dictResize(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++; server.dirty++;
addReply(c,shared.cone); addReply(c,shared.cone);
} }
...@@ -6139,7 +6205,7 @@ static void zremrangebyscoreCommand(redisClient *c) { ...@@ -6139,7 +6205,7 @@ static void zremrangebyscoreCommand(redisClient *c) {
zs = zsetobj->ptr; zs = zsetobj->ptr;
deleted = zslDeleteRangeByScore(zs->zsl,min,max,zs->dict); deleted = zslDeleteRangeByScore(zs->zsl,min,max,zs->dict);
if (htNeedsResize(zs->dict)) dictResize(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; server.dirty += deleted;
addReplyLongLong(c,deleted); addReplyLongLong(c,deleted);
} }
...@@ -6177,7 +6243,7 @@ static void zremrangebyrankCommand(redisClient *c) { ...@@ -6177,7 +6243,7 @@ static void zremrangebyrankCommand(redisClient *c) {
* use 1-based rank */ * use 1-based rank */
deleted = zslDeleteRangeByRank(zs->zsl,start+1,end+1,zs->dict); deleted = zslDeleteRangeByRank(zs->zsl,start+1,end+1,zs->dict);
if (htNeedsResize(zs->dict)) dictResize(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; server.dirty += deleted;
addReplyLongLong(c, deleted); addReplyLongLong(c, deleted);
} }
...@@ -6365,10 +6431,9 @@ static void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) { ...@@ -6365,10 +6431,9 @@ static void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
redisAssert(op == REDIS_OP_INTER || op == REDIS_OP_UNION); redisAssert(op == REDIS_OP_INTER || op == REDIS_OP_UNION);
} }
deleteKey(c->db,dstkey); dbDelete(c->db,dstkey);
if (dstzset->zsl->length) { if (dstzset->zsl->length) {
dictAdd(c->db->dict,dstkey,dstobj); dbAdd(c->db,dstkey,dstobj);
incrRefCount(dstkey);
addReplyLongLong(c, dstzset->zsl->length); addReplyLongLong(c, dstzset->zsl->length);
server.dirty++; server.dirty++;
} else { } else {
...@@ -6844,8 +6909,7 @@ static robj *hashLookupWriteOrCreate(redisClient *c, robj *key) { ...@@ -6844,8 +6909,7 @@ static robj *hashLookupWriteOrCreate(redisClient *c, robj *key) {
robj *o = lookupKeyWrite(c->db,key); robj *o = lookupKeyWrite(c->db,key);
if (o == NULL) { if (o == NULL) {
o = createHashObject(); o = createHashObject();
dictAdd(c->db->dict,key,o); dbAdd(c->db,key,o);
incrRefCount(key);
} else { } else {
if (o->type != REDIS_HASH) { if (o->type != REDIS_HASH) {
addReply(c,shared.wrongtypeerr); addReply(c,shared.wrongtypeerr);
...@@ -6969,7 +7033,7 @@ static void hdelCommand(redisClient *c) { ...@@ -6969,7 +7033,7 @@ static void hdelCommand(redisClient *c) {
checkType(c,o,REDIS_HASH)) return; checkType(c,o,REDIS_HASH)) return;
if (hashDelete(o,c->argv[2])) { 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); addReply(c,shared.cone);
server.dirty++; server.dirty++;
} else { } else {
...@@ -7438,9 +7502,7 @@ static void sortCommand(redisClient *c) { ...@@ -7438,9 +7502,7 @@ static void sortCommand(redisClient *c) {
} }
} }
} }
if (dictReplace(c->db->dict,storekey,sobj)) { dbReplace(c->db,storekey,sobj);
incrRefCount(storekey);
}
/* Note: we add 1 because the DB is dirty anyway since even if the /* 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 * SORT result is empty a new key is set and maybe the old content
* replaced. */ * replaced. */
...@@ -7622,7 +7684,7 @@ static void monitorCommand(redisClient *c) { ...@@ -7622,7 +7684,7 @@ static void monitorCommand(redisClient *c) {
/* ================================= Expire ================================= */ /* ================================= Expire ================================= */
static int removeExpire(redisDb *db, robj *key) { static int removeExpire(redisDb *db, robj *key) {
if (dictDelete(db->expires,key) == DICT_OK) { if (dictDelete(db->expires,key->ptr) == DICT_OK) {
return 1; return 1;
} else { } else {
return 0; return 0;
...@@ -7630,10 +7692,11 @@ static int removeExpire(redisDb *db, robj *key) { ...@@ -7630,10 +7692,11 @@ static int removeExpire(redisDb *db, robj *key) {
} }
static int setExpire(redisDb *db, robj *key, time_t when) { 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; return 0;
} else { } else {
incrRefCount(key);
return 1; return 1;
} }
} }
...@@ -7645,7 +7708,7 @@ static time_t getExpire(redisDb *db, robj *key) { ...@@ -7645,7 +7708,7 @@ static time_t getExpire(redisDb *db, robj *key) {
/* No expire? return ASAP */ /* No expire? return ASAP */
if (dictSize(db->expires) == 0 || 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); return (time_t) dictGetEntryVal(de);
} }
...@@ -7656,16 +7719,16 @@ static int expireIfNeeded(redisDb *db, robj *key) { ...@@ -7656,16 +7719,16 @@ static int expireIfNeeded(redisDb *db, robj *key) {
/* No expire? return ASAP */ /* No expire? return ASAP */
if (dictSize(db->expires) == 0 || 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 */ /* Lookup the expire */
when = (time_t) dictGetEntryVal(de); when = (time_t) dictGetEntryVal(de);
if (time(NULL) <= when) return 0; if (time(NULL) <= when) return 0;
/* Delete the key */ /* Delete the key */
dictDelete(db->expires,key); dbDelete(db,key);
server.stat_expiredkeys++; server.stat_expiredkeys++;
return dictDelete(db->dict,key) == DICT_OK; return 1;
} }
static int deleteIfVolatile(redisDb *db, robj *key) { static int deleteIfVolatile(redisDb *db, robj *key) {
...@@ -7673,13 +7736,13 @@ static int deleteIfVolatile(redisDb *db, robj *key) { ...@@ -7673,13 +7736,13 @@ static int deleteIfVolatile(redisDb *db, robj *key) {
/* No expire? return ASAP */ /* No expire? return ASAP */
if (dictSize(db->expires) == 0 || 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 */ /* Delete the key */
server.dirty++; server.dirty++;
server.stat_expiredkeys++; server.stat_expiredkeys++;
dictDelete(db->expires,key); dictDelete(db->expires,key->ptr);
return dictDelete(db->dict,key) == DICT_OK; return dictDelete(db->dict,key->ptr) == DICT_OK;
} }
static void expireGenericCommand(redisClient *c, robj *key, robj *param, long offset) { static void expireGenericCommand(redisClient *c, robj *key, robj *param, long offset) {
...@@ -7690,13 +7753,13 @@ static void expireGenericCommand(redisClient *c, robj *key, robj *param, long of ...@@ -7690,13 +7753,13 @@ static void expireGenericCommand(redisClient *c, robj *key, robj *param, long of
seconds -= offset; seconds -= offset;
de = dictFind(c->db->dict,key); de = dictFind(c->db->dict,key->ptr);
if (de == NULL) { if (de == NULL) {
addReply(c,shared.czero); addReply(c,shared.czero);
return; return;
} }
if (seconds <= 0) { if (seconds <= 0) {
if (deleteKey(c->db,key)) server.dirty++; if (dbDelete(c->db,key)) server.dirty++;
addReply(c, shared.cone); addReply(c, shared.cone);
return; return;
} else { } else {
...@@ -8465,7 +8528,7 @@ static void freeMemoryIfNeeded(void) { ...@@ -8465,7 +8528,7 @@ static void freeMemoryIfNeeded(void) {
minttl = t; minttl = t;
} }
} }
deleteKey(server.db+j,minkey); dbDelete(server.db+j,minkey);
} }
} }
if (!freed) return; /* nothing to free... */ if (!freed) return; /* nothing to free... */
...@@ -8474,6 +8537,48 @@ static void freeMemoryIfNeeded(void) { ...@@ -8474,6 +8537,48 @@ static void freeMemoryIfNeeded(void) {
/* ============================== Append Only file ========================== */ /* ============================== Append Only file ========================== */
/* Called when the user switches from "appendonly yes" to "appendonly no"
* at runtime using the CONFIG command. */
static void stopAppendOnly(void) {
flushAppendOnlyFile();
aof_fsync(server.appendfd);
close(server.appendfd);
server.appendfd = -1;
server.appendseldb = -1;
server.appendonly = 0;
/* rewrite operation in progress? kill it, wait child exit */
if (server.bgsavechildpid != -1) {
int statloc;
if (kill(server.bgsavechildpid,SIGKILL) != -1)
wait3(&statloc,0,NULL);
/* reset the buffer accumulating changes while the child saves */
sdsfree(server.bgrewritebuf);
server.bgrewritebuf = sdsempty();
server.bgsavechildpid = -1;
}
}
/* Called when the user switches from "appendonly no" to "appendonly yes"
* at runtime using the CONFIG command. */
static int startAppendOnly(void) {
server.appendonly = 1;
server.lastfsync = time(NULL);
server.appendfd = open(server.appendfilename,O_WRONLY|O_APPEND|O_CREAT,0644);
if (server.appendfd == -1) {
redisLog(REDIS_WARNING,"Used tried to switch on AOF via CONFIG, but I can't open the AOF file: %s",strerror(errno));
return REDIS_ERR;
}
if (rewriteAppendOnlyFileBackground() == REDIS_ERR) {
server.appendonly = 0;
close(server.appendfd);
redisLog(REDIS_WARNING,"Used tried to switch on AOF via CONFIG, I can't trigger a background AOF rewrite operation. Check the above logs for more info about the error.",strerror(errno));
return REDIS_ERR;
}
return REDIS_OK;
}
/* Write the append only file buffer on disk. /* Write the append only file buffer on disk.
* *
* Since we are required to write the AOF before replying to the client, * Since we are required to write the AOF before replying to the client,
...@@ -8507,6 +8612,11 @@ static void flushAppendOnlyFile(void) { ...@@ -8507,6 +8612,11 @@ static void flushAppendOnlyFile(void) {
sdsfree(server.aofbuf); sdsfree(server.aofbuf);
server.aofbuf = sdsempty(); server.aofbuf = sdsempty();
/* Don't Fsync if no-appendfsync-on-rewrite is set to yes and we have
* childs performing heavy I/O on disk. */
if (server.no_appendfsync_on_rewrite &&
(server.bgrewritechildpid != -1 || server.bgsavechildpid != -1))
return;
/* Fsync if needed */ /* Fsync if needed */
now = time(NULL); now = time(NULL);
if (server.appendfsync == APPENDFSYNC_ALWAYS || if (server.appendfsync == APPENDFSYNC_ALWAYS ||
...@@ -8633,7 +8743,6 @@ int loadAppendOnlyFile(char *filename) { ...@@ -8633,7 +8743,6 @@ int loadAppendOnlyFile(char *filename) {
struct redisClient *fakeClient; struct redisClient *fakeClient;
FILE *fp = fopen(filename,"r"); FILE *fp = fopen(filename,"r");
struct redis_stat sb; struct redis_stat sb;
unsigned long long loadedkeys = 0;
int appendonly = server.appendonly; int appendonly = server.appendonly;
if (redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0) if (redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0)
...@@ -8656,6 +8765,7 @@ int loadAppendOnlyFile(char *filename) { ...@@ -8656,6 +8765,7 @@ int loadAppendOnlyFile(char *filename) {
char buf[128]; char buf[128];
sds argsds; sds argsds;
struct redisCommand *cmd; struct redisCommand *cmd;
int force_swapout;
if (fgets(buf,sizeof(buf),fp) == NULL) { if (fgets(buf,sizeof(buf),fp) == NULL) {
if (feof(fp)) if (feof(fp))
...@@ -8696,8 +8806,11 @@ int loadAppendOnlyFile(char *filename) { ...@@ -8696,8 +8806,11 @@ int loadAppendOnlyFile(char *filename) {
for (j = 0; j < argc; j++) decrRefCount(argv[j]); for (j = 0; j < argc; j++) decrRefCount(argv[j]);
zfree(argv); zfree(argv);
/* Handle swapping while loading big datasets when VM is on */ /* Handle swapping while loading big datasets when VM is on */
loadedkeys++; force_swapout = 0;
if (server.vm_enabled && (loadedkeys % 5000) == 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) { while (zmalloc_used_memory() > server.vm_max_memory) {
if (vmSwapOneObjectBlocking() == REDIS_ERR) break; if (vmSwapOneObjectBlocking() == REDIS_ERR) break;
} }
...@@ -8809,24 +8922,26 @@ static int rewriteAppendOnlyFile(char *filename) { ...@@ -8809,24 +8922,26 @@ static int rewriteAppendOnlyFile(char *filename) {
/* Iterate this DB writing every entry */ /* Iterate this DB writing every entry */
while((de = dictNext(di)) != NULL) { while((de = dictNext(di)) != NULL) {
robj *key, *o; sds keystr = dictGetEntryKey(de);
robj key, *o;
time_t expiretime; time_t expiretime;
int swapped; 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. /* 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 * We use a "swapped" flag to remember if we need to free the
* value object instead to just increment the ref count anyway * value object instead to just increment the ref count anyway
* in order to avoid copy-on-write of pages if we are forked() */ * in order to avoid copy-on-write of pages if we are forked() */
if (!server.vm_enabled || key->storage == REDIS_VM_MEMORY || if (!server.vm_enabled || o->storage == REDIS_VM_MEMORY ||
key->storage == REDIS_VM_SWAPPING) { o->storage == REDIS_VM_SWAPPING) {
o = dictGetEntryVal(de);
swapped = 0; swapped = 0;
} else { } else {
o = vmPreviewObject(key); o = vmPreviewObject(o);
swapped = 1; swapped = 1;
} }
expiretime = getExpire(db,key); expiretime = getExpire(db,&key);
/* Save the key and associated value */ /* Save the key and associated value */
if (o->type == REDIS_STRING) { if (o->type == REDIS_STRING) {
...@@ -8834,7 +8949,7 @@ static int rewriteAppendOnlyFile(char *filename) { ...@@ -8834,7 +8949,7 @@ static int rewriteAppendOnlyFile(char *filename) {
char cmd[]="*3\r\n$3\r\nSET\r\n"; char cmd[]="*3\r\n$3\r\nSET\r\n";
if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
/* Key and value */ /* Key and value */
if (fwriteBulkObject(fp,key) == 0) goto werr; if (fwriteBulkObject(fp,&key) == 0) goto werr;
if (fwriteBulkObject(fp,o) == 0) goto werr; if (fwriteBulkObject(fp,o) == 0) goto werr;
} else if (o->type == REDIS_LIST) { } else if (o->type == REDIS_LIST) {
/* Emit the RPUSHes needed to rebuild the list */ /* Emit the RPUSHes needed to rebuild the list */
...@@ -8848,7 +8963,7 @@ static int rewriteAppendOnlyFile(char *filename) { ...@@ -8848,7 +8963,7 @@ static int rewriteAppendOnlyFile(char *filename) {
while(ziplistGet(p,&vstr,&vlen,&vlong)) { while(ziplistGet(p,&vstr,&vlen,&vlong)) {
if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; 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 (vstr) { if (vstr) {
if (fwriteBulkString(fp,(char*)vstr,vlen) == 0) if (fwriteBulkString(fp,(char*)vstr,vlen) == 0)
goto werr; goto werr;
...@@ -8868,7 +8983,7 @@ static int rewriteAppendOnlyFile(char *filename) { ...@@ -8868,7 +8983,7 @@ static int rewriteAppendOnlyFile(char *filename) {
robj *eleobj = listNodeValue(ln); robj *eleobj = listNodeValue(ln);
if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; 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; if (fwriteBulkObject(fp,eleobj) == 0) goto werr;
} }
} else { } else {
...@@ -8885,7 +9000,7 @@ static int rewriteAppendOnlyFile(char *filename) { ...@@ -8885,7 +9000,7 @@ static int rewriteAppendOnlyFile(char *filename) {
robj *eleobj = dictGetEntryKey(de); robj *eleobj = dictGetEntryKey(de);
if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; 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; if (fwriteBulkObject(fp,eleobj) == 0) goto werr;
} }
dictReleaseIterator(di); dictReleaseIterator(di);
...@@ -8901,7 +9016,7 @@ static int rewriteAppendOnlyFile(char *filename) { ...@@ -8901,7 +9016,7 @@ static int rewriteAppendOnlyFile(char *filename) {
double *score = dictGetEntryVal(de); double *score = dictGetEntryVal(de);
if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; 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 (fwriteBulkDouble(fp,*score) == 0) goto werr;
if (fwriteBulkObject(fp,eleobj) == 0) goto werr; if (fwriteBulkObject(fp,eleobj) == 0) goto werr;
} }
...@@ -8917,7 +9032,7 @@ static int rewriteAppendOnlyFile(char *filename) { ...@@ -8917,7 +9032,7 @@ static int rewriteAppendOnlyFile(char *filename) {
while((p = zipmapNext(p,&field,&flen,&val,&vlen)) != NULL) { while((p = zipmapNext(p,&field,&flen,&val,&vlen)) != NULL) {
if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; 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) if (fwriteBulkString(fp,(char*)field,flen) == -1)
return -1; return -1;
if (fwriteBulkString(fp,(char*)val,vlen) == -1) if (fwriteBulkString(fp,(char*)val,vlen) == -1)
...@@ -8932,7 +9047,7 @@ static int rewriteAppendOnlyFile(char *filename) { ...@@ -8932,7 +9047,7 @@ static int rewriteAppendOnlyFile(char *filename) {
robj *val = dictGetEntryVal(de); robj *val = dictGetEntryVal(de);
if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; 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,field) == -1) return -1;
if (fwriteBulkObject(fp,val) == -1) return -1; if (fwriteBulkObject(fp,val) == -1) return -1;
} }
...@@ -8947,7 +9062,7 @@ static int rewriteAppendOnlyFile(char *filename) { ...@@ -8947,7 +9062,7 @@ static int rewriteAppendOnlyFile(char *filename) {
/* If this key is already expired skip it */ /* If this key is already expired skip it */
if (expiretime < now) continue; if (expiretime < now) continue;
if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; 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 (fwriteBulkLongLong(fp,expiretime) == 0) goto werr; if (fwriteBulkLongLong(fp,expiretime) == 0) goto werr;
} }
if (swapped) decrRefCount(o); if (swapped) decrRefCount(o);
...@@ -8957,7 +9072,7 @@ static int rewriteAppendOnlyFile(char *filename) { ...@@ -8957,7 +9072,7 @@ static int rewriteAppendOnlyFile(char *filename) {
/* Make sure data will not remain on the OS's output buffers */ /* Make sure data will not remain on the OS's output buffers */
fflush(fp); fflush(fp);
fsync(fileno(fp)); aof_fsync(fileno(fp));
fclose(fp); fclose(fp);
/* Use RENAME to make sure the DB file is changed atomically only /* Use RENAME to make sure the DB file is changed atomically only
...@@ -9070,50 +9185,19 @@ static void aofRemoveTempFile(pid_t childpid) { ...@@ -9070,50 +9185,19 @@ static void aofRemoveTempFile(pid_t childpid) {
* as a fully non-blocking VM. * as a fully non-blocking VM.
*/ */
/* Called when the user switches from "appendonly yes" to "appendonly no" /* =================== Virtual Memory - Blocking Side ====================== */
* at runtime using the CONFIG command. */
static void stopAppendOnly(void) {
flushAppendOnlyFile();
fsync(server.appendfd);
close(server.appendfd);
server.appendfd = -1;
server.appendseldb = -1;
server.appendonly = 0;
/* rewrite operation in progress? kill it, wait child exit */
if (server.bgsavechildpid != -1) {
int statloc;
if (kill(server.bgsavechildpid,SIGKILL) != -1) /* Create a VM pointer object. This kind of objects are used in place of
wait3(&statloc,0,NULL); * values in the key -> value hash table, for swapped out objects. */
/* reset the buffer accumulating changes while the child saves */ static vmpointer *createVmPointer(int vtype) {
sdsfree(server.bgrewritebuf); vmpointer *vp = zmalloc(sizeof(vmpointer));
server.bgrewritebuf = sdsempty();
server.bgsavechildpid = -1;
}
}
/* Called when the user switches from "appendonly no" to "appendonly yes" vp->type = REDIS_VMPOINTER;
* at runtime using the CONFIG command. */ vp->storage = REDIS_VM_SWAPPED;
static int startAppendOnly(void) { vp->vtype = vtype;
server.appendonly = 1; return vp;
server.lastfsync = time(NULL);
server.appendfd = open(server.appendfilename,O_WRONLY|O_APPEND|O_CREAT,0644);
if (server.appendfd == -1) {
redisLog(REDIS_WARNING,"Used tried to switch on AOF via CONFIG, but I can't open the AOF file: %s",strerror(errno));
return REDIS_ERR;
}
if (rewriteAppendOnlyFileBackground() == REDIS_ERR) {
server.appendonly = 0;
close(server.appendfd);
redisLog(REDIS_WARNING,"Used tried to switch on AOF via CONFIG, I can't trigger a background AOF rewrite operation. Check the above logs for more info about the error.",strerror(errno));
return REDIS_ERR;
}
return REDIS_OK;
} }
/* =================== Virtual Memory - Blocking Side ====================== */
static void vmInit(void) { static void vmInit(void) {
off_t totsize; off_t totsize;
int pipefds[2]; int pipefds[2];
...@@ -9328,30 +9412,33 @@ static int vmWriteObjectOnSwap(robj *o, off_t page) { ...@@ -9328,30 +9412,33 @@ static int vmWriteObjectOnSwap(robj *o, off_t page) {
return REDIS_OK; return REDIS_OK;
} }
/* Swap the 'val' object relative to 'key' into disk. Store all the information /* Transfers the 'val' object to disk. Store all the information
* needed to later retrieve the object into the key object. * 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 * If we can't find enough contiguous empty pages to swap the object on disk
* REDIS_ERR is returned. */ * NULL is returned. */
static int vmSwapObjectBlocking(robj *key, robj *val) { static vmpointer *vmSwapObjectBlocking(robj *val) {
off_t pages = rdbSavedObjectPages(val,NULL); off_t pages = rdbSavedObjectPages(val,NULL);
off_t page; off_t page;
vmpointer *vp;
assert(key->storage == REDIS_VM_MEMORY); assert(val->storage == REDIS_VM_MEMORY);
assert(key->refcount == 1); assert(val->refcount == 1);
if (vmFindContiguousPages(&page,pages) == REDIS_ERR) return REDIS_ERR; if (vmFindContiguousPages(&page,pages) == REDIS_ERR) return NULL;
if (vmWriteObjectOnSwap(val,page) == REDIS_ERR) return REDIS_ERR; if (vmWriteObjectOnSwap(val,page) == REDIS_ERR) return NULL;
key->vm.page = page;
key->vm.usedpages = pages; vp = createVmPointer(val->type);
key->storage = REDIS_VM_SWAPPED; vp->page = page;
key->vtype = val->type; vp->usedpages = pages;
decrRefCount(val); /* Deallocate the object from memory. */ decrRefCount(val); /* Deallocate the object from memory. */
vmMarkPagesUsed(page,pages); vmMarkPagesUsed(page,pages);
redisLog(REDIS_DEBUG,"VM: object %s swapped out at %lld (%lld pages)", redisLog(REDIS_DEBUG,"VM: object %p swapped out at %lld (%lld pages)",
(unsigned char*) key->ptr, (void*) val,
(unsigned long long) page, (unsigned long long) pages); (unsigned long long) page, (unsigned long long) pages);
server.vm_stats_swapped_objects++; server.vm_stats_swapped_objects++;
server.vm_stats_swapouts++; server.vm_stats_swapouts++;
return REDIS_OK; return vp;
} }
static robj *vmReadObjectFromSwap(off_t page, int type) { static robj *vmReadObjectFromSwap(off_t page, int type) {
...@@ -9373,46 +9460,47 @@ static robj *vmReadObjectFromSwap(off_t page, int type) { ...@@ -9373,46 +9460,47 @@ static robj *vmReadObjectFromSwap(off_t page, int type) {
return o; return o;
} }
/* Load the value object relative to the 'key' object from swap to memory. /* Load the specified object from swap to memory.
* The newly allocated object is returned. * The newly allocated object is returned.
* *
* If preview is true the unserialized object is returned to the caller but * If preview is true the unserialized object is returned to the caller but
* no changes are made to the key object, nor the pages are marked as freed */ * the pages are not marked as freed, nor the vp object is freed. */
static robj *vmGenericLoadObject(robj *key, int preview) { static robj *vmGenericLoadObject(vmpointer *vp, int preview) {
robj *val; robj *val;
redisAssert(key->storage == REDIS_VM_SWAPPED || key->storage == REDIS_VM_LOADING); redisAssert(vp->type == REDIS_VMPOINTER &&
val = vmReadObjectFromSwap(key->vm.page,key->vtype); (vp->storage == REDIS_VM_SWAPPED || vp->storage == REDIS_VM_LOADING));
val = vmReadObjectFromSwap(vp->page,vp->vtype);
if (!preview) { if (!preview) {
key->storage = REDIS_VM_MEMORY; redisLog(REDIS_DEBUG, "VM: object %p loaded from disk", (void*)vp);
key->vm.atime = server.unixtime; vmMarkPagesFree(vp->page,vp->usedpages);
vmMarkPagesFree(key->vm.page,key->vm.usedpages); zfree(vp);
redisLog(REDIS_DEBUG, "VM: object %s loaded from disk",
(unsigned char*) key->ptr);
server.vm_stats_swapped_objects--; server.vm_stats_swapped_objects--;
} else { } else {
redisLog(REDIS_DEBUG, "VM: object %s previewed from disk", redisLog(REDIS_DEBUG, "VM: object %p previewed from disk", (void*)vp);
(unsigned char*) key->ptr);
} }
server.vm_stats_swapins++; server.vm_stats_swapins++;
return val; return val;
} }
/* Plain object loading, from swap to memory */ /* Plain object loading, from swap to memory.
static robj *vmLoadObject(robj *key) { *
* 'o' is actually a redisVmPointer structure that will be freed by the call.
* The return value is the loaded object. */
static robj *vmLoadObject(robj *o) {
/* If we are loading the object in background, stop it, we /* If we are loading the object in background, stop it, we
* need to load this object synchronously ASAP. */ * need to load this object synchronously ASAP. */
if (key->storage == REDIS_VM_LOADING) if (o->storage == REDIS_VM_LOADING)
vmCancelThreadedIOJob(key); vmCancelThreadedIOJob(o);
return vmGenericLoadObject(key,0); return vmGenericLoadObject((vmpointer*)o,0);
} }
/* Just load the value on disk, without to modify the key. /* Just load the value on disk, without to modify the key.
* This is useful when we want to perform some operation on the value * This is useful when we want to perform some operation on the value
* without to really bring it from swap to memory, like while saving the * without to really bring it from swap to memory, like while saving the
* dataset or rewriting the append only log. */ * dataset or rewriting the append only log. */
static robj *vmPreviewObject(robj *key) { static robj *vmPreviewObject(robj *o) {
return vmGenericLoadObject(key,1); return vmGenericLoadObject((vmpointer*)o,1);
} }
/* How a good candidate is this object for swapping? /* How a good candidate is this object for swapping?
...@@ -9427,14 +9515,16 @@ static robj *vmPreviewObject(robj *key) { ...@@ -9427,14 +9515,16 @@ static robj *vmPreviewObject(robj *key) {
* proportionally, this is why we use the logarithm. This algorithm is * proportionally, this is why we use the logarithm. This algorithm is
* just a first try and will probably be tuned later. */ * just a first try and will probably be tuned later. */
static double computeObjectSwappability(robj *o) { static double computeObjectSwappability(robj *o) {
time_t age = server.unixtime - o->vm.atime; /* actual age can be >= minage, but not < minage. As we use wrapping
* 21 bit clocks with minutes resolution for the LRU. */
time_t minage = abs(server.lruclock - o->lru);
long asize = 0; long asize = 0;
list *l; list *l;
dict *d; dict *d;
struct dictEntry *de; struct dictEntry *de;
int z; int z;
if (age <= 0) return 0; if (minage <= 0) return 0;
switch(o->type) { switch(o->type) {
case REDIS_STRING: case REDIS_STRING:
if (o->encoding != REDIS_ENCODING_RAW) { if (o->encoding != REDIS_ENCODING_RAW) {
...@@ -9453,8 +9543,7 @@ static double computeObjectSwappability(robj *o) { ...@@ -9453,8 +9543,7 @@ static double computeObjectSwappability(robj *o) {
long elesize; long elesize;
elesize = (ele->encoding == REDIS_ENCODING_RAW) ? elesize = (ele->encoding == REDIS_ENCODING_RAW) ?
(sizeof(*o)+sdslen(ele->ptr)) : (sizeof(*o)+sdslen(ele->ptr)) : sizeof(*o);
sizeof(*o);
asize += (sizeof(listNode)+elesize)*listLength(l); asize += (sizeof(listNode)+elesize)*listLength(l);
} }
break; break;
...@@ -9472,8 +9561,7 @@ static double computeObjectSwappability(robj *o) { ...@@ -9472,8 +9561,7 @@ static double computeObjectSwappability(robj *o) {
de = dictGetRandomKey(d); de = dictGetRandomKey(d);
ele = dictGetEntryKey(de); ele = dictGetEntryKey(de);
elesize = (ele->encoding == REDIS_ENCODING_RAW) ? elesize = (ele->encoding == REDIS_ENCODING_RAW) ?
(sizeof(*o)+sdslen(ele->ptr)) : (sizeof(*o)+sdslen(ele->ptr)) : sizeof(*o);
sizeof(*o);
asize += (sizeof(struct dictEntry)+elesize)*dictSize(d); asize += (sizeof(struct dictEntry)+elesize)*dictSize(d);
if (z) asize += sizeof(zskiplistNode)*dictSize(d); if (z) asize += sizeof(zskiplistNode)*dictSize(d);
} }
...@@ -9500,18 +9588,16 @@ static double computeObjectSwappability(robj *o) { ...@@ -9500,18 +9588,16 @@ static double computeObjectSwappability(robj *o) {
de = dictGetRandomKey(d); de = dictGetRandomKey(d);
ele = dictGetEntryKey(de); ele = dictGetEntryKey(de);
elesize = (ele->encoding == REDIS_ENCODING_RAW) ? elesize = (ele->encoding == REDIS_ENCODING_RAW) ?
(sizeof(*o)+sdslen(ele->ptr)) : (sizeof(*o)+sdslen(ele->ptr)) : sizeof(*o);
sizeof(*o);
ele = dictGetEntryVal(de); ele = dictGetEntryVal(de);
elesize = (ele->encoding == REDIS_ENCODING_RAW) ? elesize = (ele->encoding == REDIS_ENCODING_RAW) ?
(sizeof(*o)+sdslen(ele->ptr)) : (sizeof(*o)+sdslen(ele->ptr)) : sizeof(*o);
sizeof(*o);
asize += (sizeof(struct dictEntry)+elesize)*dictSize(d); asize += (sizeof(struct dictEntry)+elesize)*dictSize(d);
} }
} }
break; break;
} }
return (double)age*log(1+asize); return (double)minage*log(1+asize);
} }
/* Try to swap an object that's a good candidate for swapping. /* Try to swap an object that's a good candidate for swapping.
...@@ -9525,7 +9611,8 @@ static int vmSwapOneObject(int usethreads) { ...@@ -9525,7 +9611,8 @@ static int vmSwapOneObject(int usethreads) {
struct dictEntry *best = NULL; struct dictEntry *best = NULL;
double best_swappability = 0; double best_swappability = 0;
redisDb *best_db = NULL; redisDb *best_db = NULL;
robj *key, *val; robj *val;
sds key;
for (j = 0; j < server.dbnum; j++) { for (j = 0; j < server.dbnum; j++) {
redisDb *db = server.db+j; redisDb *db = server.db+j;
...@@ -9541,16 +9628,14 @@ static int vmSwapOneObject(int usethreads) { ...@@ -9541,16 +9628,14 @@ static int vmSwapOneObject(int usethreads) {
if (maxtries) maxtries--; if (maxtries) maxtries--;
de = dictGetRandomKey(db->dict); de = dictGetRandomKey(db->dict);
key = dictGetEntryKey(de);
val = dictGetEntryVal(de); val = dictGetEntryVal(de);
/* Only swap objects that are currently in memory. /* Only swap objects that are currently in memory.
* *
* Also don't swap shared objects if threaded VM is on, as we * Also don't swap shared objects: not a good idea in general and
* try to ensure that the main thread does not touch the * we need to ensure that the main thread does not touch the
* object while the I/O thread is using it, but we can't * object while the I/O thread is using it, but we can't
* control other keys without adding additional mutex. */ * control other keys without adding additional mutex. */
if (key->storage != REDIS_VM_MEMORY || if (val->storage != REDIS_VM_MEMORY || val->refcount != 1) {
(server.vm_max_threads != 0 && val->refcount != 1)) {
if (maxtries) i--; /* don't count this try */ if (maxtries) i--; /* don't count this try */
continue; continue;
} }
...@@ -9567,21 +9652,19 @@ static int vmSwapOneObject(int usethreads) { ...@@ -9567,21 +9652,19 @@ static int vmSwapOneObject(int usethreads) {
val = dictGetEntryVal(best); val = dictGetEntryVal(best);
redisLog(REDIS_DEBUG,"Key with best swappability: %s, %f", redisLog(REDIS_DEBUG,"Key with best swappability: %s, %f",
key->ptr, best_swappability); key, best_swappability);
/* Unshare the key if needed */
if (key->refcount > 1) {
robj *newkey = dupStringObject(key);
decrRefCount(key);
key = dictGetEntryKey(best) = newkey;
}
/* Swap it */ /* Swap it */
if (usethreads) { if (usethreads) {
vmSwapObjectThreaded(key,val,best_db); robj *keyobj = createStringObject(key,sdslen(key));
vmSwapObjectThreaded(keyobj,val,best_db);
decrRefCount(keyobj);
return REDIS_OK; return REDIS_OK;
} else { } else {
if (vmSwapObjectBlocking(key,val) == REDIS_OK) { vmpointer *vp;
dictGetEntryVal(best) = NULL;
if ((vp = vmSwapObjectBlocking(val)) != NULL) {
dictGetEntryVal(best) = vp;
return REDIS_OK; return REDIS_OK;
} else { } else {
return REDIS_ERR; return REDIS_ERR;
...@@ -9604,30 +9687,20 @@ static int vmCanSwapOut(void) { ...@@ -9604,30 +9687,20 @@ static int vmCanSwapOut(void) {
return (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1); 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) {
dictEntry *de;
robj *foundkey;
if ((de = dictFind(db->dict,key)) == NULL) return 0;
foundkey = dictGetEntryKey(de);
if (foundkey->storage == REDIS_VM_MEMORY) return 0;
deleteKey(db,key);
return 1;
}
/* =================== Virtual Memory - Threaded I/O ======================= */ /* =================== Virtual Memory - Threaded I/O ======================= */
static void freeIOJob(iojob *j) { static void freeIOJob(iojob *j) {
if ((j->type == REDIS_IOJOB_PREPARE_SWAP || if ((j->type == REDIS_IOJOB_PREPARE_SWAP ||
j->type == REDIS_IOJOB_DO_SWAP || j->type == REDIS_IOJOB_DO_SWAP ||
j->type == REDIS_IOJOB_LOAD) && j->val != NULL) j->type == REDIS_IOJOB_LOAD) && j->val != NULL)
{
/* we fix the storage type, otherwise decrRefCount() will try to
* kill the I/O thread Job (that does no longer exists). */
if (j->val->storage == REDIS_VM_SWAPPING)
j->val->storage = REDIS_VM_MEMORY;
decrRefCount(j->val); decrRefCount(j->val);
/* We don't decrRefCount the j->key field as we did't incremented }
* the count creating IO Jobs. This is because the key field here is decrRefCount(j->key);
* just used as an indentifier and if a key is removed the Job should
* never be touched again. */
zfree(j); zfree(j);
} }
...@@ -9648,7 +9721,6 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata, ...@@ -9648,7 +9721,6 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
while((retval = read(fd,buf,1)) == 1) { while((retval = read(fd,buf,1)) == 1) {
iojob *j; iojob *j;
listNode *ln; listNode *ln;
robj *key;
struct dictEntry *de; struct dictEntry *de;
redisLog(REDIS_DEBUG,"Processing I/O completed job"); redisLog(REDIS_DEBUG,"Processing I/O completed job");
...@@ -9671,27 +9743,26 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata, ...@@ -9671,27 +9743,26 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
} }
/* Post process it in the main thread, as there are things we /* Post process it in the main thread, as there are things we
* can do just here to avoid race conditions and/or invasive locks */ * can do just here to avoid race conditions and/or invasive locks */
redisLog(REDIS_DEBUG,"Job %p type: %d, key at %p (%s) refcount: %d\n", (void*) j, j->type, (void*)j->key, (char*)j->key->ptr, j->key->refcount); 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);
assert(de != NULL); redisAssert(de != NULL);
key = dictGetEntryKey(de);
if (j->type == REDIS_IOJOB_LOAD) { if (j->type == REDIS_IOJOB_LOAD) {
redisDb *db; redisDb *db;
vmpointer *vp = dictGetEntryVal(de);
/* Key loaded, bring it at home */ /* Key loaded, bring it at home */
key->storage = REDIS_VM_MEMORY; vmMarkPagesFree(vp->page,vp->usedpages);
key->vm.atime = server.unixtime;
vmMarkPagesFree(key->vm.page,key->vm.usedpages);
redisLog(REDIS_DEBUG, "VM: object %s loaded from disk (threaded)", redisLog(REDIS_DEBUG, "VM: object %s loaded from disk (threaded)",
(unsigned char*) key->ptr); (unsigned char*) j->key->ptr);
server.vm_stats_swapped_objects--; server.vm_stats_swapped_objects--;
server.vm_stats_swapins++; server.vm_stats_swapins++;
dictGetEntryVal(de) = j->val; dictGetEntryVal(de) = j->val;
incrRefCount(j->val); incrRefCount(j->val);
db = j->db; db = j->db;
freeIOJob(j);
/* Handle clients waiting for this key to be loaded. */ /* Handle clients waiting for this key to be loaded. */
handleClientsBlockedOnSwappedKey(db,key); handleClientsBlockedOnSwappedKey(db,j->key);
freeIOJob(j);
zfree(vp);
} else if (j->type == REDIS_IOJOB_PREPARE_SWAP) { } else if (j->type == REDIS_IOJOB_PREPARE_SWAP) {
/* Now we know the amount of pages required to swap this object. /* Now we know the amount of pages required to swap this object.
* Let's find some space for it, and queue this task again * Let's find some space for it, and queue this task again
...@@ -9701,8 +9772,8 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata, ...@@ -9701,8 +9772,8 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
{ {
/* Ooops... no space or we can't swap as there is /* Ooops... no space or we can't swap as there is
* a fork()ed Redis trying to save stuff on disk. */ * a fork()ed Redis trying to save stuff on disk. */
j->val->storage = REDIS_VM_MEMORY; /* undo operation */
freeIOJob(j); freeIOJob(j);
key->storage = REDIS_VM_MEMORY; /* undo operation */
} else { } else {
/* Note that we need to mark this pages as used now, /* Note that we need to mark this pages as used now,
* if the job will be canceled, we'll mark them as freed * if the job will be canceled, we'll mark them as freed
...@@ -9714,28 +9785,29 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata, ...@@ -9714,28 +9785,29 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
unlockThreadedIO(); unlockThreadedIO();
} }
} else if (j->type == REDIS_IOJOB_DO_SWAP) { } else if (j->type == REDIS_IOJOB_DO_SWAP) {
robj *val; vmpointer *vp;
/* Key swapped. We can finally free some memory. */ /* Key swapped. We can finally free some memory. */
if (key->storage != REDIS_VM_SWAPPING) { if (j->val->storage != REDIS_VM_SWAPPING) {
printf("key->storage: %d\n",key->storage); vmpointer *vp = (vmpointer*) j->id;
printf("key->name: %s\n",(char*)key->ptr); printf("storage: %d\n",vp->storage);
printf("key->refcount: %d\n",key->refcount); printf("key->name: %s\n",(char*)j->key->ptr);
printf("val: %p\n",(void*)j->val); printf("val: %p\n",(void*)j->val);
printf("val->type: %d\n",j->val->type); printf("val->type: %d\n",j->val->type);
printf("val->ptr: %s\n",(char*)j->val->ptr); printf("val->ptr: %s\n",(char*)j->val->ptr);
} }
redisAssert(key->storage == REDIS_VM_SWAPPING); redisAssert(j->val->storage == REDIS_VM_SWAPPING);
val = dictGetEntryVal(de); vp = createVmPointer(j->val->type);
key->vm.page = j->page; vp->page = j->page;
key->vm.usedpages = j->pages; vp->usedpages = j->pages;
key->storage = REDIS_VM_SWAPPED; dictGetEntryVal(de) = vp;
key->vtype = j->val->type; /* Fix the storage otherwise decrRefCount will attempt to
decrRefCount(val); /* Deallocate the object from memory. */ * remove the associated I/O job */
dictGetEntryVal(de) = NULL; j->val->storage = REDIS_VM_MEMORY;
decrRefCount(j->val);
redisLog(REDIS_DEBUG, redisLog(REDIS_DEBUG,
"VM: object %s swapped out at %lld (%lld pages) (threaded)", "VM: object %s swapped out at %lld (%lld pages) (threaded)",
(unsigned char*) key->ptr, (unsigned char*) j->key->ptr,
(unsigned long long) j->page, (unsigned long long) j->pages); (unsigned long long) j->page, (unsigned long long) j->pages);
server.vm_stats_swapped_objects++; server.vm_stats_swapped_objects++;
server.vm_stats_swapouts++; server.vm_stats_swapouts++;
...@@ -9790,7 +9862,7 @@ static void vmCancelThreadedIOJob(robj *o) { ...@@ -9790,7 +9862,7 @@ static void vmCancelThreadedIOJob(robj *o) {
assert(o->storage == REDIS_VM_LOADING || o->storage == REDIS_VM_SWAPPING); assert(o->storage == REDIS_VM_LOADING || o->storage == REDIS_VM_SWAPPING);
again: again:
lockThreadedIO(); lockThreadedIO();
/* Search for a matching key in one of the queues */ /* Search for a matching object in one of the queues */
for (i = 0; i < 3; i++) { for (i = 0; i < 3; i++) {
listNode *ln; listNode *ln;
listIter li; listIter li;
...@@ -9800,9 +9872,9 @@ again: ...@@ -9800,9 +9872,9 @@ again:
iojob *job = ln->value; iojob *job = ln->value;
if (job->canceled) continue; /* Skip this, already canceled. */ if (job->canceled) continue; /* Skip this, already canceled. */
if (job->key == o) { if (job->id == o) {
redisLog(REDIS_DEBUG,"*** CANCELED %p (%s) (type %d) (LIST ID %d)\n", redisLog(REDIS_DEBUG,"*** CANCELED %p (key %s) (type %d) (LIST ID %d)\n",
(void*)job, (char*)o->ptr, job->type, i); (void*)job, (char*)job->key->ptr, job->type, i);
/* Mark the pages as free since the swap didn't happened /* Mark the pages as free since the swap didn't happened
* or happened but is now discarded. */ * or happened but is now discarded. */
if (i != 1 && job->type == REDIS_IOJOB_DO_SWAP) if (i != 1 && job->type == REDIS_IOJOB_DO_SWAP)
...@@ -9850,12 +9922,14 @@ again: ...@@ -9850,12 +9922,14 @@ again:
else if (o->storage == REDIS_VM_SWAPPING) else if (o->storage == REDIS_VM_SWAPPING)
o->storage = REDIS_VM_MEMORY; o->storage = REDIS_VM_MEMORY;
unlockThreadedIO(); unlockThreadedIO();
redisLog(REDIS_DEBUG,"*** DONE");
return; return;
} }
} }
} }
unlockThreadedIO(); unlockThreadedIO();
assert(1 != 1); /* We should never reach this */ printf("Not found: %p\n", (void*)o);
redisAssert(1 != 1); /* We should never reach this */
} }
static void *IOThreadEntryPoint(void *arg) { static void *IOThreadEntryPoint(void *arg) {
...@@ -9888,7 +9962,8 @@ static void *IOThreadEntryPoint(void *arg) { ...@@ -9888,7 +9962,8 @@ static void *IOThreadEntryPoint(void *arg) {
/* Process the Job */ /* Process the Job */
if (j->type == REDIS_IOJOB_LOAD) { if (j->type == REDIS_IOJOB_LOAD) {
j->val = vmReadObjectFromSwap(j->page,j->key->vtype); vmpointer *vp = (vmpointer*)j->id;
j->val = vmReadObjectFromSwap(j->page,vp->vtype);
} else if (j->type == REDIS_IOJOB_PREPARE_SWAP) { } else if (j->type == REDIS_IOJOB_PREPARE_SWAP) {
FILE *fp = fopen("/dev/null","w+"); FILE *fp = fopen("/dev/null","w+");
j->pages = rdbSavedObjectPages(j->val,fp); j->pages = rdbSavedObjectPages(j->val,fp);
...@@ -9984,18 +10059,16 @@ static void queueIOJob(iojob *j) { ...@@ -9984,18 +10059,16 @@ static void queueIOJob(iojob *j) {
static int vmSwapObjectThreaded(robj *key, robj *val, redisDb *db) { static int vmSwapObjectThreaded(robj *key, robj *val, redisDb *db) {
iojob *j; iojob *j;
assert(key->storage == REDIS_VM_MEMORY);
assert(key->refcount == 1);
j = zmalloc(sizeof(*j)); j = zmalloc(sizeof(*j));
j->type = REDIS_IOJOB_PREPARE_SWAP; j->type = REDIS_IOJOB_PREPARE_SWAP;
j->db = db; j->db = db;
j->key = key; j->key = key;
j->val = val; incrRefCount(key);
j->id = j->val = val;
incrRefCount(val); incrRefCount(val);
j->canceled = 0; j->canceled = 0;
j->thread = (pthread_t) -1; j->thread = (pthread_t) -1;
key->storage = REDIS_VM_SWAPPING; val->storage = REDIS_VM_SWAPPING;
lockThreadedIO(); lockThreadedIO();
queueIOJob(j); queueIOJob(j);
...@@ -10017,9 +10090,9 @@ static int waitForSwappedKey(redisClient *c, robj *key) { ...@@ -10017,9 +10090,9 @@ static int waitForSwappedKey(redisClient *c, robj *key) {
/* If the key does not exist or is already in RAM we don't need to /* If the key does not exist or is already in RAM we don't need to
* block the client at all. */ * block the client at all. */
de = dictFind(c->db->dict,key); de = dictFind(c->db->dict,key->ptr);
if (de == NULL) return 0; if (de == NULL) return 0;
o = dictGetEntryKey(de); o = dictGetEntryVal(de);
if (o->storage == REDIS_VM_MEMORY) { if (o->storage == REDIS_VM_MEMORY) {
return 0; return 0;
} else if (o->storage == REDIS_VM_SWAPPING) { } else if (o->storage == REDIS_VM_SWAPPING) {
...@@ -10053,14 +10126,16 @@ static int waitForSwappedKey(redisClient *c, robj *key) { ...@@ -10053,14 +10126,16 @@ static int waitForSwappedKey(redisClient *c, robj *key) {
/* Are we already loading the key from disk? If not create a job */ /* Are we already loading the key from disk? If not create a job */
if (o->storage == REDIS_VM_SWAPPED) { if (o->storage == REDIS_VM_SWAPPED) {
iojob *j; iojob *j;
vmpointer *vp = (vmpointer*)o;
o->storage = REDIS_VM_LOADING; o->storage = REDIS_VM_LOADING;
j = zmalloc(sizeof(*j)); j = zmalloc(sizeof(*j));
j->type = REDIS_IOJOB_LOAD; j->type = REDIS_IOJOB_LOAD;
j->db = c->db; j->db = c->db;
j->key = o; j->id = (robj*)vp;
j->key->vtype = o->vtype; j->key = key;
j->page = o->vm.page; incrRefCount(key);
j->page = vp->page;
j->val = NULL; j->val = NULL;
j->canceled = 0; j->canceled = 0;
j->thread = (pthread_t) -1; j->thread = (pthread_t) -1;
...@@ -10185,6 +10260,8 @@ static int dontWaitForSwappedKey(redisClient *c, robj *key) { ...@@ -10185,6 +10260,8 @@ static int dontWaitForSwappedKey(redisClient *c, robj *key) {
return listLength(c->io_keys) == 0; return listLength(c->io_keys) == 0;
} }
/* Every time we now a key was loaded back in memory, we handle clients
* waiting for this key if any. */
static void handleClientsBlockedOnSwappedKey(redisDb *db, robj *key) { static void handleClientsBlockedOnSwappedKey(redisDb *db, robj *key) {
struct dictEntry *de; struct dictEntry *de;
list *l; list *l;
...@@ -10243,6 +10320,11 @@ static void configSetCommand(redisClient *c) { ...@@ -10243,6 +10320,11 @@ static void configSetCommand(redisClient *c) {
} else { } else {
goto badfmt; goto badfmt;
} }
} else if (!strcasecmp(c->argv[2]->ptr,"no-appendfsync-on-rewrite")) {
int yn = yesnotoi(o->ptr);
if (yn == -1) goto badfmt;
server.no_appendfsync_on_rewrite = yn;
} else if (!strcasecmp(c->argv[2]->ptr,"appendonly")) { } else if (!strcasecmp(c->argv[2]->ptr,"appendonly")) {
int old = server.appendonly; int old = server.appendonly;
int new = yesnotoi(o->ptr); int new = yesnotoi(o->ptr);
...@@ -10358,6 +10440,11 @@ static void configGetCommand(redisClient *c) { ...@@ -10358,6 +10440,11 @@ static void configGetCommand(redisClient *c) {
addReplyBulkCString(c,server.appendonly ? "yes" : "no"); addReplyBulkCString(c,server.appendonly ? "yes" : "no");
matches++; matches++;
} }
if (stringmatch(pattern,"no-appendfsync-on-rewrite",0)) {
addReplyBulkCString(c,"no-appendfsync-on-rewrite");
addReplyBulkCString(c,server.no_appendfsync_on_rewrite ? "yes" : "no");
matches++;
}
if (stringmatch(pattern,"appendfsync",0)) { if (stringmatch(pattern,"appendfsync",0)) {
char *policy; char *policy;
...@@ -10793,7 +10880,7 @@ static void touchWatchedKeysOnFlush(int dbid) { ...@@ -10793,7 +10880,7 @@ static void touchWatchedKeysOnFlush(int dbid) {
* key exists, mark the client as dirty, as the key will be * key exists, mark the client as dirty, as the key will be
* removed. */ * removed. */
if (dbid == -1 || wk->db->id == dbid) { 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; c->flags |= REDIS_DIRTY_CAS;
} }
} }
...@@ -10904,27 +10991,22 @@ static void computeDatasetDigest(unsigned char *final) { ...@@ -10904,27 +10991,22 @@ static void computeDatasetDigest(unsigned char *final) {
/* Iterate this DB writing every entry */ /* Iterate this DB writing every entry */
while((de = dictNext(di)) != NULL) { while((de = dictNext(di)) != NULL) {
robj *key, *o, *kcopy; sds key;
robj *keyobj, *o;
time_t expiretime; time_t expiretime;
memset(digest,0,20); /* This key-val digest */ memset(digest,0,20); /* This key-val digest */
key = dictGetEntryKey(de); 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); aux = htonl(o->type);
mixDigest(digest,&aux,sizeof(aux)); mixDigest(digest,&aux,sizeof(aux));
expiretime = getExpire(db,key); expiretime = getExpire(db,keyobj);
/* Save the key and associated value */ /* Save the key and associated value */
if (o->type == REDIS_STRING) { if (o->type == REDIS_STRING) {
...@@ -10991,6 +11073,7 @@ static void computeDatasetDigest(unsigned char *final) { ...@@ -10991,6 +11073,7 @@ static void computeDatasetDigest(unsigned char *final) {
if (expiretime != -1) xorDigest(digest,"!!expire!!",10); if (expiretime != -1) xorDigest(digest,"!!expire!!",10);
/* We can finally xor the key-val digest to the final digest */ /* We can finally xor the key-val digest to the final digest */
xorDigest(final,digest,20); xorDigest(final,digest,20);
decrRefCount(keyobj);
} }
dictReleaseIterator(di); dictReleaseIterator(di);
} }
...@@ -11020,17 +11103,16 @@ static void debugCommand(redisClient *c) { ...@@ -11020,17 +11103,16 @@ static void debugCommand(redisClient *c) {
redisLog(REDIS_WARNING,"Append Only File loaded by DEBUG LOADAOF"); redisLog(REDIS_WARNING,"Append Only File loaded by DEBUG LOADAOF");
addReply(c,shared.ok); addReply(c,shared.ok);
} else if (!strcasecmp(c->argv[1]->ptr,"object") && c->argc == 3) { } else if (!strcasecmp(c->argv[1]->ptr,"object") && c->argc == 3) {
dictEntry *de = dictFind(c->db->dict,c->argv[2]); dictEntry *de = dictFind(c->db->dict,c->argv[2]->ptr);
robj *key, *val; robj *val;
if (!de) { if (!de) {
addReply(c,shared.nokeyerr); addReply(c,shared.nokeyerr);
return; return;
} }
key = dictGetEntryKey(de);
val = dictGetEntryVal(de); val = dictGetEntryVal(de);
if (!server.vm_enabled || (key->storage == REDIS_VM_MEMORY || if (!server.vm_enabled || (val->storage == REDIS_VM_MEMORY ||
key->storage == REDIS_VM_SWAPPING)) { val->storage == REDIS_VM_SWAPPING)) {
char *strenc; char *strenc;
char buf[128]; char buf[128];
...@@ -11041,23 +11123,25 @@ static void debugCommand(redisClient *c) { ...@@ -11041,23 +11123,25 @@ static void debugCommand(redisClient *c) {
strenc = buf; strenc = buf;
} }
addReplySds(c,sdscatprintf(sdsempty(), 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", "encoding:%s serializedlength:%lld\r\n",
(void*)key, key->refcount, (void*)val, val->refcount, (void*)val, val->refcount,
strenc, (long long) rdbSavedObjectLen(val,NULL))); strenc, (long long) rdbSavedObjectLen(val,NULL)));
} else { } else {
vmpointer *vp = (vmpointer*) val;
addReplySds(c,sdscatprintf(sdsempty(), addReplySds(c,sdscatprintf(sdsempty(),
"+Key at:%p refcount:%d, value swapped at: page %llu " "+Value swapped at: page %llu "
"using %llu pages\r\n", "using %llu pages\r\n",
(void*)key, key->refcount, (unsigned long long) key->vm.page, (unsigned long long) vp->page,
(unsigned long long) key->vm.usedpages)); (unsigned long long) vp->usedpages));
} }
} else if (!strcasecmp(c->argv[1]->ptr,"swapin") && c->argc == 3) { } else if (!strcasecmp(c->argv[1]->ptr,"swapin") && c->argc == 3) {
lookupKeyRead(c->db,c->argv[2]); lookupKeyRead(c->db,c->argv[2]);
addReply(c,shared.ok); addReply(c,shared.ok);
} else if (!strcasecmp(c->argv[1]->ptr,"swapout") && c->argc == 3) { } else if (!strcasecmp(c->argv[1]->ptr,"swapout") && c->argc == 3) {
dictEntry *de = dictFind(c->db->dict,c->argv[2]); dictEntry *de = dictFind(c->db->dict,c->argv[2]->ptr);
robj *key, *val; robj *val;
vmpointer *vp;
if (!server.vm_enabled) { if (!server.vm_enabled) {
addReplySds(c,sdsnew("-ERR Virtual Memory is disabled\r\n")); addReplySds(c,sdsnew("-ERR Virtual Memory is disabled\r\n"));
...@@ -11067,19 +11151,14 @@ static void debugCommand(redisClient *c) { ...@@ -11067,19 +11151,14 @@ static void debugCommand(redisClient *c) {
addReply(c,shared.nokeyerr); addReply(c,shared.nokeyerr);
return; return;
} }
key = dictGetEntryKey(de);
val = dictGetEntryVal(de); val = dictGetEntryVal(de);
/* If the key is shared we want to create a copy */
if (key->refcount > 1) {
robj *newkey = dupStringObject(key);
decrRefCount(key);
key = dictGetEntryKey(de) = newkey;
}
/* Swap it */ /* Swap it */
if (key->storage != REDIS_VM_MEMORY) { if (val->storage != REDIS_VM_MEMORY) {
addReplySds(c,sdsnew("-ERR This key is not in memory\r\n")); addReplySds(c,sdsnew("-ERR This key is not in memory\r\n"));
} else if (vmSwapObjectBlocking(key,val) == REDIS_OK) { } else if (val->refcount != 1) {
dictGetEntryVal(de) = NULL; addReplySds(c,sdsnew("-ERR Object is shared\r\n"));
} else if ((vp = vmSwapObjectBlocking(val)) != NULL) {
dictGetEntryVal(de) = vp;
addReply(c,shared.ok); addReply(c,shared.ok);
} else { } else {
addReply(c,shared.err); addReply(c,shared.err);
...@@ -11100,7 +11179,8 @@ static void debugCommand(redisClient *c) { ...@@ -11100,7 +11179,8 @@ static void debugCommand(redisClient *c) {
} }
snprintf(buf,sizeof(buf),"value:%lu",j); snprintf(buf,sizeof(buf),"value:%lu",j);
val = createStringObject(buf,strlen(buf)); val = createStringObject(buf,strlen(buf));
dictAdd(c->db->dict,key,val); dbAdd(c->db,key,val);
decrRefCount(key);
} }
addReply(c,shared.ok); addReply(c,shared.ok);
} else if (!strcasecmp(c->argv[1]->ptr,"digest") && c->argc == 2) { } else if (!strcasecmp(c->argv[1]->ptr,"digest") && c->argc == 2) {
......
...@@ -195,6 +195,26 @@ appendonly no ...@@ -195,6 +195,26 @@ appendonly no
appendfsync everysec appendfsync everysec
# appendfsync no # appendfsync no
# When the AOF fsync policy is set to always or everysec, and a background
# saving process (a background save or AOF log background rewriting) is
# performing a lot of I/O against the disk, in some Linux configurations
# Redis may block too long on the fsync() call. Note that there is no fix for
# this currently, as even performing fsync in a different thread will block
# our synchronous write(2) call.
#
# In order to mitigate this problem it's possible to use the following option
# that will prevent fsync() from being called in the main process while a
# BGSAVE or BGREWRITEAOF is in progress.
#
# This means that while another child is saving the durability of Redis is
# the same as "appendfsync none", that in pratical terms means that it is
# possible to lost up to 30 seconds of log in the worst scenario (with the
# default Linux settings).
#
# If you have latency problems turn this to "yes". Otherwise leave it as
# "no" that is the safest pick from the point of view of durability.
no-appendfsync-on-rewrite no
################################ VIRTUAL MEMORY ############################### ################################ VIRTUAL MEMORY ###############################
# Virtual Memory allows Redis to work with datasets bigger than the actual # Virtual Memory allows Redis to work with datasets bigger than the actual
......
...@@ -7,6 +7,7 @@ static struct redisFunctionSym symsTable[] = { ...@@ -7,6 +7,7 @@ static struct redisFunctionSym symsTable[] = {
{"addReplyBulk",(unsigned long)addReplyBulk}, {"addReplyBulk",(unsigned long)addReplyBulk},
{"addReplyBulkCString",(unsigned long)addReplyBulkCString}, {"addReplyBulkCString",(unsigned long)addReplyBulkCString},
{"addReplyBulkLen",(unsigned long)addReplyBulkLen}, {"addReplyBulkLen",(unsigned long)addReplyBulkLen},
{"addReplyBulkSds",(unsigned long)addReplyBulkSds},
{"addReplyDouble",(unsigned long)addReplyDouble}, {"addReplyDouble",(unsigned long)addReplyDouble},
{"addReplyLongLong",(unsigned long)addReplyLongLong}, {"addReplyLongLong",(unsigned long)addReplyLongLong},
{"addReplySds",(unsigned long)addReplySds}, {"addReplySds",(unsigned long)addReplySds},
...@@ -45,21 +46,27 @@ static struct redisFunctionSym symsTable[] = { ...@@ -45,21 +46,27 @@ static struct redisFunctionSym symsTable[] = {
{"createSortOperation",(unsigned long)createSortOperation}, {"createSortOperation",(unsigned long)createSortOperation},
{"createStringObject",(unsigned long)createStringObject}, {"createStringObject",(unsigned long)createStringObject},
{"createStringObjectFromLongLong",(unsigned long)createStringObjectFromLongLong}, {"createStringObjectFromLongLong",(unsigned long)createStringObjectFromLongLong},
{"createVmPointer",(unsigned long)createVmPointer},
{"createZsetObject",(unsigned long)createZsetObject}, {"createZsetObject",(unsigned long)createZsetObject},
{"daemonize",(unsigned long)daemonize}, {"daemonize",(unsigned long)daemonize},
{"dbAdd",(unsigned long)dbAdd},
{"dbDelete",(unsigned long)dbDelete},
{"dbExists",(unsigned long)dbExists},
{"dbRandomKey",(unsigned long)dbRandomKey},
{"dbReplace",(unsigned long)dbReplace},
{"dbsizeCommand",(unsigned long)dbsizeCommand}, {"dbsizeCommand",(unsigned long)dbsizeCommand},
{"debugCommand",(unsigned long)debugCommand}, {"debugCommand",(unsigned long)debugCommand},
{"decrCommand",(unsigned long)decrCommand}, {"decrCommand",(unsigned long)decrCommand},
{"decrRefCount",(unsigned long)decrRefCount}, {"decrRefCount",(unsigned long)decrRefCount},
{"decrbyCommand",(unsigned long)decrbyCommand}, {"decrbyCommand",(unsigned long)decrbyCommand},
{"delCommand",(unsigned long)delCommand}, {"delCommand",(unsigned long)delCommand},
{"deleteIfSwapped",(unsigned long)deleteIfSwapped},
{"deleteIfVolatile",(unsigned long)deleteIfVolatile}, {"deleteIfVolatile",(unsigned long)deleteIfVolatile},
{"deleteKey",(unsigned long)deleteKey},
{"dictEncObjKeyCompare",(unsigned long)dictEncObjKeyCompare}, {"dictEncObjKeyCompare",(unsigned long)dictEncObjKeyCompare},
{"dictListDestructor",(unsigned long)dictListDestructor}, {"dictListDestructor",(unsigned long)dictListDestructor},
{"dictObjKeyCompare",(unsigned long)dictObjKeyCompare}, {"dictObjKeyCompare",(unsigned long)dictObjKeyCompare},
{"dictRedisObjectDestructor",(unsigned long)dictRedisObjectDestructor}, {"dictRedisObjectDestructor",(unsigned long)dictRedisObjectDestructor},
{"dictSdsDestructor",(unsigned long)dictSdsDestructor},
{"dictSdsKeyCompare",(unsigned long)dictSdsKeyCompare},
{"dictVanillaFree",(unsigned long)dictVanillaFree}, {"dictVanillaFree",(unsigned long)dictVanillaFree},
{"discardCommand",(unsigned long)discardCommand}, {"discardCommand",(unsigned long)discardCommand},
{"dontWaitForSwappedKey",(unsigned long)dontWaitForSwappedKey}, {"dontWaitForSwappedKey",(unsigned long)dontWaitForSwappedKey},
...@@ -196,6 +203,7 @@ static struct redisFunctionSym symsTable[] = { ...@@ -196,6 +203,7 @@ static struct redisFunctionSym symsTable[] = {
{"pushGenericCommand",(unsigned long)pushGenericCommand}, {"pushGenericCommand",(unsigned long)pushGenericCommand},
{"qsortCompareSetsByCardinality",(unsigned long)qsortCompareSetsByCardinality}, {"qsortCompareSetsByCardinality",(unsigned long)qsortCompareSetsByCardinality},
{"qsortCompareZsetopsrcByCardinality",(unsigned long)qsortCompareZsetopsrcByCardinality}, {"qsortCompareZsetopsrcByCardinality",(unsigned long)qsortCompareZsetopsrcByCardinality},
{"qsortRedisCommands",(unsigned long)qsortRedisCommands},
{"queueIOJob",(unsigned long)queueIOJob}, {"queueIOJob",(unsigned long)queueIOJob},
{"queueMultiCommand",(unsigned long)queueMultiCommand}, {"queueMultiCommand",(unsigned long)queueMultiCommand},
{"randomkeyCommand",(unsigned long)randomkeyCommand}, {"randomkeyCommand",(unsigned long)randomkeyCommand},
...@@ -245,7 +253,6 @@ static struct redisFunctionSym symsTable[] = { ...@@ -245,7 +253,6 @@ static struct redisFunctionSym symsTable[] = {
{"scardCommand",(unsigned long)scardCommand}, {"scardCommand",(unsigned long)scardCommand},
{"sdiffCommand",(unsigned long)sdiffCommand}, {"sdiffCommand",(unsigned long)sdiffCommand},
{"sdiffstoreCommand",(unsigned long)sdiffstoreCommand}, {"sdiffstoreCommand",(unsigned long)sdiffstoreCommand},
{"sdsDictKeyCompare",(unsigned long)sdsDictKeyCompare},
{"sdscatrepr",(unsigned long)sdscatrepr}, {"sdscatrepr",(unsigned long)sdscatrepr},
{"segvHandler",(unsigned long)segvHandler}, {"segvHandler",(unsigned long)segvHandler},
{"selectCommand",(unsigned long)selectCommand}, {"selectCommand",(unsigned long)selectCommand},
...@@ -269,6 +276,7 @@ static struct redisFunctionSym symsTable[] = { ...@@ -269,6 +276,7 @@ static struct redisFunctionSym symsTable[] = {
{"slaveofCommand",(unsigned long)slaveofCommand}, {"slaveofCommand",(unsigned long)slaveofCommand},
{"smoveCommand",(unsigned long)smoveCommand}, {"smoveCommand",(unsigned long)smoveCommand},
{"sortCommand",(unsigned long)sortCommand}, {"sortCommand",(unsigned long)sortCommand},
{"sortCommandTable",(unsigned long)sortCommandTable},
{"sortCompare",(unsigned long)sortCompare}, {"sortCompare",(unsigned long)sortCompare},
{"spawnIOThread",(unsigned long)spawnIOThread}, {"spawnIOThread",(unsigned long)spawnIOThread},
{"spopCommand",(unsigned long)spopCommand}, {"spopCommand",(unsigned long)spopCommand},
...@@ -290,6 +298,7 @@ static struct redisFunctionSym symsTable[] = { ...@@ -290,6 +298,7 @@ static struct redisFunctionSym symsTable[] = {
{"syncWithMaster",(unsigned long)syncWithMaster}, {"syncWithMaster",(unsigned long)syncWithMaster},
{"syncWrite",(unsigned long)syncWrite}, {"syncWrite",(unsigned long)syncWrite},
{"touchWatchedKey",(unsigned long)touchWatchedKey}, {"touchWatchedKey",(unsigned long)touchWatchedKey},
{"touchWatchedKeysOnFlush",(unsigned long)touchWatchedKeysOnFlush},
{"tryFreeOneObjectFromFreelist",(unsigned long)tryFreeOneObjectFromFreelist}, {"tryFreeOneObjectFromFreelist",(unsigned long)tryFreeOneObjectFromFreelist},
{"tryObjectEncoding",(unsigned long)tryObjectEncoding}, {"tryObjectEncoding",(unsigned long)tryObjectEncoding},
{"tryResizeHashTables",(unsigned long)tryResizeHashTables}, {"tryResizeHashTables",(unsigned long)tryResizeHashTables},
......
set defaults [list [list appendonly yes] [list appendfilename appendonly.aof]] set defaults { appendonly {yes} appendfilename {appendonly.aof} }
set server_path [tmpdir server.aof] set server_path [tmpdir server.aof]
set aof_path "$server_path/appendonly.aof" set aof_path "$server_path/appendonly.aof"
...@@ -16,20 +16,21 @@ proc create_aof {code} { ...@@ -16,20 +16,21 @@ proc create_aof {code} {
proc start_server_aof {overrides code} { proc start_server_aof {overrides code} {
upvar defaults defaults srv srv server_path server_path upvar defaults defaults srv srv server_path server_path
set _defaults $defaults set config [concat $defaults $overrides]
set srv [start_server default.conf [lappend _defaults $overrides]] set srv [start_server [list overrides $config]]
uplevel 1 $code uplevel 1 $code
kill_server $srv kill_server $srv
} }
## Test the server doesn't start when the AOF contains an unfinished MULTI tags {"aof"} {
create_aof { ## Test the server doesn't start when the AOF contains an unfinished MULTI
create_aof {
append_to_aof [formatCommand set foo hello] append_to_aof [formatCommand set foo hello]
append_to_aof [formatCommand multi] append_to_aof [formatCommand multi]
append_to_aof [formatCommand set bar world] append_to_aof [formatCommand set bar world]
} }
start_server_aof [list dir $server_path] { start_server_aof [list dir $server_path] {
test {Unfinished MULTI: Server should not have been started} { test {Unfinished MULTI: Server should not have been started} {
is_alive $srv is_alive $srv
} {0} } {0}
...@@ -37,15 +38,15 @@ start_server_aof [list dir $server_path] { ...@@ -37,15 +38,15 @@ start_server_aof [list dir $server_path] {
test {Unfinished MULTI: Server should have logged an error} { test {Unfinished MULTI: Server should have logged an error} {
exec cat [dict get $srv stdout] | tail -n1 exec cat [dict get $srv stdout] | tail -n1
} {*Unexpected end of file reading the append only file*} } {*Unexpected end of file reading the append only file*}
} }
## Test that the server exits when the AOF contains a short read ## Test that the server exits when the AOF contains a short read
create_aof { create_aof {
append_to_aof [formatCommand set foo hello] append_to_aof [formatCommand set foo hello]
append_to_aof [string range [formatCommand set bar world] 0 end-1] append_to_aof [string range [formatCommand set bar world] 0 end-1]
} }
start_server_aof [list dir $server_path] { start_server_aof [list dir $server_path] {
test {Short read: Server should not have been started} { test {Short read: Server should not have been started} {
is_alive $srv is_alive $srv
} {0} } {0}
...@@ -53,22 +54,22 @@ start_server_aof [list dir $server_path] { ...@@ -53,22 +54,22 @@ start_server_aof [list dir $server_path] {
test {Short read: Server should have logged an error} { test {Short read: Server should have logged an error} {
exec cat [dict get $srv stdout] | tail -n1 exec cat [dict get $srv stdout] | tail -n1
} {*Bad file format reading the append only file*} } {*Bad file format reading the append only file*}
} }
## Test that redis-check-aof indeed sees this AOF is not valid ## Test that redis-check-aof indeed sees this AOF is not valid
test {Short read: Utility should confirm the AOF is not valid} { test {Short read: Utility should confirm the AOF is not valid} {
catch { catch {
exec ./redis-check-aof $aof_path exec ./redis-check-aof $aof_path
} str } str
set _ $str set _ $str
} {*not valid*} } {*not valid*}
test {Short read: Utility should be able to fix the AOF} { test {Short read: Utility should be able to fix the AOF} {
exec echo y | ./redis-check-aof --fix $aof_path exec echo y | ./redis-check-aof --fix $aof_path
} {*Successfully truncated AOF*} } {*Successfully truncated AOF*}
## Test that the server can be started using the truncated AOF ## Test that the server can be started using the truncated AOF
start_server_aof [list dir $server_path] { start_server_aof [list dir $server_path] {
test {Fixed AOF: Server should have been started} { test {Fixed AOF: Server should have been started} {
is_alive $srv is_alive $srv
} {1} } {1}
...@@ -77,4 +78,5 @@ start_server_aof [list dir $server_path] { ...@@ -77,4 +78,5 @@ start_server_aof [list dir $server_path] {
set client [redis [dict get $srv host] [dict get $srv port]] set client [redis [dict get $srv host] [dict get $srv port]]
list [$client get foo] [$client get bar] list [$client get foo] [$client get bar]
} {hello {}} } {hello {}}
}
} }
start_server default.conf {} { start_server {tags {"repl"}} {
r set mykey foo r set mykey foo
start_server default.conf {} { start_server {} {
test {Second server should have role master at first} { test {Second server should have role master at first} {
s role s role
} {master} } {master}
......
set ::global_overrides {}
set ::tags {}
proc error_and_quit {config_file error} { proc error_and_quit {config_file error} {
puts "!!COULD NOT START REDIS-SERVER\n" puts "!!COULD NOT START REDIS-SERVER\n"
puts "CONFIGURATION:" puts "CONFIGURATION:"
...@@ -27,13 +30,17 @@ proc kill_server config { ...@@ -27,13 +30,17 @@ proc kill_server config {
set pid [dict get $config pid] set pid [dict get $config pid]
# check for leaks # check for leaks
if {![dict exists $config "skipleaks"]} {
catch { catch {
if {[string match {*Darwin*} [exec uname -a]]} { if {[string match {*Darwin*} [exec uname -a]]} {
tags {"leaks"} {
test "Check for memory leaks (pid $pid)" { test "Check for memory leaks (pid $pid)" {
exec leaks $pid exec leaks $pid
} {*0 leaks*} } {*0 leaks*}
} }
} }
}
}
# kill server and wait for the process to be totally exited # kill server and wait for the process to be totally exited
while {[is_alive $config]} { while {[is_alive $config]} {
...@@ -78,9 +85,35 @@ proc ping_server {host port} { ...@@ -78,9 +85,35 @@ proc ping_server {host port} {
return $retval return $retval
} }
set ::global_overrides {} # doesn't really belong here, but highly coupled to code in start_server
proc start_server {filename overrides {code undefined}} { proc tags {tags code} {
set data [split [exec cat "tests/assets/$filename"] "\n"] set ::tags [concat $::tags $tags]
uplevel 1 $code
set ::tags [lrange $::tags 0 end-[llength $tags]]
}
proc start_server {options {code undefined}} {
# setup defaults
set baseconfig "default.conf"
set overrides {}
set tags {}
# parse options
foreach {option value} $options {
switch $option {
"config" {
set baseconfig $value }
"overrides" {
set overrides $value }
"tags" {
set tags $value
set ::tags [concat $::tags $value] }
default {
error "Unknown option $option" }
}
}
set data [split [exec cat "tests/assets/$baseconfig"] "\n"]
set config {} set config {}
foreach line $data { foreach line $data {
if {[string length $line] > 0 && [string index $line 0] ne "#"} { if {[string length $line] > 0 && [string index $line 0] ne "#"} {
...@@ -98,9 +131,7 @@ proc start_server {filename overrides {code undefined}} { ...@@ -98,9 +131,7 @@ proc start_server {filename overrides {code undefined}} {
dict set config port [incr ::port] dict set config port [incr ::port]
# apply overrides from global space and arguments # apply overrides from global space and arguments
foreach override [concat $::global_overrides $overrides] { foreach {directive arguments} [concat $::global_overrides $overrides] {
set directive [lrange $override 0 0]
set arguments [lrange $override 1 end]
dict set config $directive $arguments dict set config $directive $arguments
} }
...@@ -177,19 +208,40 @@ proc start_server {filename overrides {code undefined}} { ...@@ -177,19 +208,40 @@ proc start_server {filename overrides {code undefined}} {
lappend ::servers $srv lappend ::servers $srv
# execute provided block # execute provided block
set curnum $::testnum
catch { uplevel 1 $code } err catch { uplevel 1 $code } err
if {$curnum == $::testnum} {
# don't check for leaks when no tests were executed
dict set srv "skipleaks" 1
}
# pop the server object # pop the server object
set ::servers [lrange $::servers 0 end-1] set ::servers [lrange $::servers 0 end-1]
# allow an exception to bubble up the call chain but still kill this
# server, because we want to reuse the ports when the tests are re-run
if {$err eq "exception"} {
puts [format "Logged warnings (pid %d):" [dict get $srv "pid"]]
set warnings [warnings_from_file [dict get $srv "stdout"]]
if {[string length $warnings] > 0} {
puts "$warnings"
} else {
puts "(none)"
}
# kill this server without checking for leaks
dict set srv "skipleaks" 1
kill_server $srv kill_server $srv
error "exception"
if {[string length $err] > 0} { } elseif {[string length $err] > 0} {
puts "Error executing the suite, aborting..." puts "Error executing the suite, aborting..."
puts $err puts $err
exit 1 exit 1
} }
set ::tags [lrange $::tags 0 end-[llength $tags]]
kill_server $srv
} else { } else {
set ::tags [lrange $::tags 0 end-[llength $tags]]
set _ $srv set _ $srv
} }
} }
...@@ -3,20 +3,34 @@ set ::failed 0 ...@@ -3,20 +3,34 @@ set ::failed 0
set ::testnum 0 set ::testnum 0
proc test {name code okpattern} { proc test {name code okpattern} {
# abort if tagged with a tag to deny
foreach tag $::denytags {
if {[lsearch $::tags $tag] >= 0} {
return
}
}
# check if tagged with at least 1 tag to allow when there *is* a list
# of tags to allow, because default policy is to run everything
if {[llength $::allowtags] > 0} {
set matched 0
foreach tag $::allowtags {
if {[lsearch $::tags $tag] >= 0} {
incr matched
}
}
if {$matched < 1} {
return
}
}
incr ::testnum incr ::testnum
# if {$::testnum < $::first || $::testnum > $::last} return
puts -nonewline [format "#%03d %-68s " $::testnum $name] puts -nonewline [format "#%03d %-68s " $::testnum $name]
flush stdout flush stdout
if {[catch {set retval [uplevel 1 $code]} error]} { if {[catch {set retval [uplevel 1 $code]} error]} {
puts "ERROR\n\nLogged warnings:" puts "EXCEPTION"
foreach file [glob tests/tmp/server.[pid].*/stdout] { puts "\nCaught error: $error"
set warnings [warnings_from_file $file] error "exception"
if {[string length $warnings] > 0} {
puts $warnings
}
}
puts "Script died with $error"
exit 1
} }
if {$okpattern eq $retval || [string match $okpattern $retval]} { if {$okpattern eq $retval || [string match $okpattern $retval]} {
puts "PASSED" puts "PASSED"
......
...@@ -13,9 +13,10 @@ set ::host 127.0.0.1 ...@@ -13,9 +13,10 @@ set ::host 127.0.0.1
set ::port 16379 set ::port 16379
set ::traceleaks 0 set ::traceleaks 0
set ::valgrind 0 set ::valgrind 0
set ::denytags {}
set ::allowtags {}
proc execute_tests name { proc execute_tests name {
set cur $::testnum
source "tests/$name.tcl" source "tests/$name.tcl"
} }
...@@ -92,4 +93,31 @@ proc main {} { ...@@ -92,4 +93,31 @@ proc main {} {
cleanup cleanup
} }
main # parse arguments
for {set j 0} {$j < [llength $argv]} {incr j} {
set opt [lindex $argv $j]
set arg [lindex $argv [expr $j+1]]
if {$opt eq {--tags}} {
foreach tag $arg {
if {[string index $tag 0] eq "-"} {
lappend ::denytags [string range $tag 1 end]
} else {
lappend ::allowtags $tag
}
}
incr j
} else {
puts "Wrong argument: $opt"
exit 1
}
}
if {[catch { main } err]} {
if {[string length $err] > 0} {
# only display error when not generated by the test suite
if {$err ne "exception"} {
puts $err
}
exit 1
}
}
start_server default.conf {{requirepass foobar}} { start_server {tags {"auth"} overrides {requirepass foobar}} {
test {AUTH fails when a wrong password is given} { test {AUTH fails when a wrong password is given} {
catch {r auth wrong!} err catch {r auth wrong!} err
format $err format $err
......
start_server default.conf {} { start_server {tags {"basic"}} {
test {DEL all keys to start with a clean DB} { test {DEL all keys to start with a clean DB} {
foreach key [r keys *] {r del $key} foreach key [r keys *] {r del $key}
r dbsize r dbsize
...@@ -52,6 +52,7 @@ start_server default.conf {} { ...@@ -52,6 +52,7 @@ start_server default.conf {} {
r get foo r get foo
} [string repeat "abcd" 1000000] } [string repeat "abcd" 1000000]
tags {"slow"} {
test {Very big payload random access} { test {Very big payload random access} {
set err {} set err {}
array set payload {} array set payload {}
...@@ -92,6 +93,7 @@ start_server default.conf {} { ...@@ -92,6 +93,7 @@ start_server default.conf {} {
test {DBSIZE should be 10101 now} { test {DBSIZE should be 10101 now} {
r dbsize r dbsize
} {10101} } {10101}
}
test {INCR against non existing key} { test {INCR against non existing key} {
set res {} set res {}
......
start_server default.conf {} { start_server {tags {"cas"}} {
test {EXEC works on WATCHed key not modified} { test {EXEC works on WATCHed key not modified} {
r watch x y z r watch x y z
r watch k r watch k
......
start_server default.conf {} { start_server {tags {"expire"}} {
test {EXPIRE - don't set timeouts multiple times} { test {EXPIRE - don't set timeouts multiple times} {
r set x foobar r set x foobar
set v1 [r expire x 5] set v1 [r expire x 5]
...@@ -12,10 +12,12 @@ start_server default.conf {} { ...@@ -12,10 +12,12 @@ start_server default.conf {} {
r get x r get x
} {foobar} } {foobar}
tags {"slow"} {
test {EXPIRE - After 6 seconds the key should no longer be here} { test {EXPIRE - After 6 seconds the key should no longer be here} {
after 6000 after 6000
list [r get x] [r exists x] list [r get x] [r exists x]
} {{} 0} } {{} 0}
}
test {EXPIRE - Delete on write policy} { test {EXPIRE - Delete on write policy} {
r del x r del x
...@@ -46,10 +48,12 @@ start_server default.conf {} { ...@@ -46,10 +48,12 @@ start_server default.conf {} {
r get y r get y
} {foo} } {foo}
tags {"slow"} {
test {SETEX - Wait for the key to expire} { test {SETEX - Wait for the key to expire} {
after 3000 after 3000
r get y r get y
} {} } {}
}
test {SETEX - Wrong time parameter} { test {SETEX - Wrong time parameter} {
catch {r setex z -10 foo} e catch {r setex z -10 foo} e
......
start_server default.conf {} { start_server {} {
test {SAVE - make sure there are all the types as values} { test {SAVE - make sure there are all the types as values} {
# Wait for a background saving in progress to terminate # Wait for a background saving in progress to terminate
waitForBgsave r waitForBgsave r
...@@ -12,6 +12,7 @@ start_server default.conf {} { ...@@ -12,6 +12,7 @@ start_server default.conf {} {
r save r save
} {OK} } {OK}
tags {"slow"} {
foreach fuzztype {binary alpha compr} { foreach fuzztype {binary alpha compr} {
test "FUZZ stresser with data model $fuzztype" { test "FUZZ stresser with data model $fuzztype" {
set err 0 set err 0
...@@ -27,6 +28,7 @@ start_server default.conf {} { ...@@ -27,6 +28,7 @@ start_server default.conf {} {
set _ $err set _ $err
} {0} } {0}
} }
}
test {BGSAVE} { test {BGSAVE} {
waitForBgsave r waitForBgsave r
......
start_server default.conf {} { start_server {} {
test {Handle an empty query well} { test {Handle an empty query well} {
set fd [r channel] set fd [r channel]
puts -nonewline $fd "\r\n" puts -nonewline $fd "\r\n"
......
start_server default.conf {} { start_server {tags {"sort"}} {
test {SORT ALPHA against integer encoded strings} { test {SORT ALPHA against integer encoded strings} {
r del mylist r del mylist
r lpush mylist 2 r lpush mylist 2
...@@ -8,6 +8,8 @@ start_server default.conf {} { ...@@ -8,6 +8,8 @@ start_server default.conf {} {
r sort mylist alpha r sort mylist alpha
} {1 10 2 3} } {1 10 2 3}
tags {"slow"} {
set res {}
test {Create a random list and a random set} { test {Create a random list and a random set} {
set tosort {} set tosort {}
array set seenrand {} array set seenrand {}
...@@ -30,7 +32,6 @@ start_server default.conf {} { ...@@ -30,7 +32,6 @@ start_server default.conf {} {
lappend tosort [list $i $rint] lappend tosort [list $i $rint]
} }
set sorted [lsort -index 1 -real $tosort] set sorted [lsort -index 1 -real $tosort]
set res {}
for {set i 0} {$i < 10000} {incr i} { for {set i 0} {$i < 10000} {incr i} {
lappend res [lindex $sorted $i 0] lappend res [lindex $sorted $i 0]
} }
...@@ -132,6 +133,7 @@ start_server default.conf {} { ...@@ -132,6 +133,7 @@ start_server default.conf {} {
flush stdout flush stdout
format {} format {}
} {} } {}
}
test {SORT regression for issue #19, sorting floats} { test {SORT regression for issue #19, sorting floats} {
r flushdb r flushdb
......
start_server default.conf {} { start_server {tags {"hash"}} {
test {HSET/HLEN - Small hash creation} { test {HSET/HLEN - Small hash creation} {
array set smallhash {} array set smallhash {}
for {set i 0} {$i < 8} {incr i} { for {set i 0} {$i < 8} {incr i} {
......
start_server default.conf {} { start_server {tags {"list"}} {
test {Basic LPUSH, RPUSH, LLENGTH, LINDEX} { test {Basic LPUSH, RPUSH, LLENGTH, LINDEX} {
set res [r lpush mylist a] set res [r lpush mylist a]
append res [r lpush mylist b] append res [r lpush mylist b]
......
start_server default.conf {} { start_server {tags {"set"}} {
test {SADD, SCARD, SISMEMBER, SMEMBERS basics} { test {SADD, SCARD, SISMEMBER, SMEMBERS basics} {
r sadd myset foo r sadd myset foo
r sadd myset bar r sadd myset bar
......
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