* POSSIBILITY OF SUCH DAMAGE.
*/
-#define REDIS_VERSION "1.3.2"
+#define REDIS_VERSION "1.3.3"
#include "fmacros.h"
#include "config.h"
#define REDIS_VM_MAX_NEAR_PAGES 65536
#define REDIS_VM_MAX_RANDOM_JUMP 4096
#define REDIS_VM_MAX_THREADS 32
+#define REDIS_THREAD_STACK_SIZE (1024*1024*4)
+/* The following is the *percentage* of completed I/O jobs to process when the
+ * handelr is called. While Virtual Memory I/O operations are performed by
+ * threads, this operations must be processed by the main thread when completed
+ * in order to take effect. */
+#define REDIS_MAX_COMPLETED_JOBS_PROCESSED 1
/* Client flags */
-#define REDIS_CLOSE 1 /* This client connection should be closed ASAP */
-#define REDIS_SLAVE 2 /* This client is a slave server */
-#define REDIS_MASTER 4 /* This client is a master server */
-#define REDIS_MONITOR 8 /* This client is a slave monitor, see MONITOR */
-#define REDIS_MULTI 16 /* This client is in a MULTI context */
-#define REDIS_BLOCKED 32 /* The client is waiting in a blocking operation */
-#define REDIS_IO_WAIT 64 /* The client is waiting for Virtual Memory I/O */
+#define REDIS_SLAVE 1 /* This client is a slave server */
+#define REDIS_MASTER 2 /* This client is a master server */
+#define REDIS_MONITOR 4 /* This client is a slave monitor, see MONITOR */
+#define REDIS_MULTI 8 /* This client is in a MULTI context */
+#define REDIS_BLOCKED 16 /* The client is waiting in a blocking operation */
+#define REDIS_IO_WAIT 32 /* The client is waiting for Virtual Memory I/O */
/* Slave replication state - slave side */
#define REDIS_REPL_NONE 0 /* No active replication */
#define APPENDFSYNC_EVERYSEC 2
/* We can print the stacktrace, so our assert is defined this way: */
-#define redisAssert(_e) ((_e)?(void)0 : (_redisAssert(#_e,__FILE__,__LINE__),exit(1)))
+#define redisAssert(_e) ((_e)?(void)0 : (_redisAssert(#_e,__FILE__,__LINE__),_exit(1)))
static void _redisAssert(char *estr, char *file, int line);
/*================================= Data types ============================== */
dict *dict; /* The keyspace for this DB */
dict *expires; /* Timeout of keys with a timeout set */
dict *blockingkeys; /* Keys with clients waiting for data (BLPOP) */
+ dict *io_keys; /* Keys with clients waiting for VM I/O */
int id;
} redisDb;
list *reply;
int sentlen;
time_t lastinteraction; /* time of the last interaction, used for timeout */
- int flags; /* REDIS_CLOSE | REDIS_SLAVE | REDIS_MONITOR */
- /* REDIS_MULTI */
+ int flags; /* REDIS_SLAVE | REDIS_MONITOR | REDIS_MULTI ... */
int slaveseldb; /* slave selected db, if this client is a slave */
int authenticated; /* when requirepass is non-NULL */
int replstate; /* replication state if this is a slave */
long repldboff; /* replication DB file offset */
off_t repldbsize; /* replication DB file size */
multiState mstate; /* MULTI/EXEC state */
- robj **blockingkeys; /* The key we waiting to terminate a blocking
+ robj **blockingkeys; /* The key we are waiting to terminate a blocking
* operation such as BLPOP. Otherwise NULL. */
int blockingkeysnum; /* Number of blocking keys */
time_t blockingto; /* Blocking operation timeout. If UNIX current time
int cronloops; /* number of times the cron function run */
list *objfreelist; /* A list of freed objects to avoid malloc() */
time_t lastsave; /* Unix time of last save succeeede */
- size_t usedmemory; /* Used memory in megabytes */
/* Fields used only for stats */
time_t stat_starttime; /* server start time */
long long stat_numcommands; /* number of processed commands */
int replstate;
unsigned int maxclients;
unsigned long long maxmemory;
- unsigned int blockedclients;
+ unsigned int blpop_blocked_clients;
+ unsigned int vm_blocked_clients;
/* Sort parameters - qsort_r() is only available under BSD so we
* have to take this state global, in order to pass it to sortCompare() */
int sort_desc;
int sort_bypattern;
/* Virtual memory configuration */
int vm_enabled;
+ char *vm_swap_file;
off_t vm_page_size;
off_t vm_pages;
unsigned long long vm_max_memory;
list *io_newjobs; /* List of VM I/O jobs yet to be processed */
list *io_processing; /* List of VM I/O jobs being processed */
list *io_processed; /* List of VM I/O jobs already processed */
- list *io_clients; /* All the clients waiting for SWAP I/O operations */
+ list *io_ready_clients; /* Clients ready to be unblocked. All keys loaded */
pthread_mutex_t io_mutex; /* lock to access io_jobs/io_done/io_thread_job */
pthread_mutex_t obj_freelist_mutex; /* safe redis objects creation/free */
pthread_mutex_t io_swapfile_mutex; /* So we can lseek + write */
+ pthread_attr_t io_threads_attr; /* attributes for threads creation */
int io_active_threads; /* Number of running I/O threads */
int vm_max_threads; /* Max number of I/O threads running at the same time */
/* Our main thread is blocked on the event loop, locking for sockets ready
#define REDIS_IOJOB_LOAD 0 /* Load from disk to memory */
#define REDIS_IOJOB_PREPARE_SWAP 1 /* Compute needed pages */
#define REDIS_IOJOB_DO_SWAP 2 /* Swap from memory to disk */
-typedef struct iojon {
+typedef struct iojob {
int type; /* Request type, REDIS_IOJOB_* */
redisDb *db;/* Redis database */
robj *key; /* This I/O request is about swapping this key */
static void initClientMultiState(redisClient *c);
static void freeClientMultiState(redisClient *c);
static void queueMultiCommand(redisClient *c, struct redisCommand *cmd);
-static void unblockClient(redisClient *c);
+static void unblockClientWaitingData(redisClient *c);
static int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele);
static void vmInit(void);
static void vmMarkPagesFree(off_t page, off_t count);
static void queueIOJob(iojob *j);
static int vmWriteObjectOnSwap(robj *o, off_t page);
static robj *vmReadObjectFromSwap(off_t page, int type);
+static void waitEmptyIOJobsQueue(void);
+static void vmReopenSwapFile(void);
+static int vmFreePage(off_t page);
+static int blockClientOnSwappedKeys(struct redisCommand *cmd, redisClient *c);
+static int dontWaitForSwappedKey(redisClient *c, robj *key);
+static void handleClientsBlockedOnSwappedKey(redisDb *db, robj *key);
+static void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask);
+static struct redisCommand *lookupCommand(char *name);
+static void call(redisClient *c, struct redisCommand *cmd);
+static void resetClient(redisClient *c);
static void authCommand(redisClient *c);
static void pingCommand(redisClient *c);
static void execCommand(redisClient *c);
static void blpopCommand(redisClient *c);
static void brpopCommand(redisClient *c);
+static void appendCommand(redisClient *c);
/*================================= Globals ================================= */
{"get",getCommand,2,REDIS_CMD_INLINE},
{"set",setCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
{"setnx",setnxCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
+ {"append",appendCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
{"del",delCommand,-2,REDIS_CMD_INLINE},
{"exists",existsCommand,2,REDIS_CMD_INLINE},
{"incr",incrCommand,2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
{"lrange",lrangeCommand,4,REDIS_CMD_INLINE},
{"ltrim",ltrimCommand,4,REDIS_CMD_INLINE},
{"lrem",lremCommand,4,REDIS_CMD_BULK},
- {"rpoplpush",rpoplpushcommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
+ {"rpoplpush",rpoplpushcommand,3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
{"sadd",saddCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
{"srem",sremCommand,3,REDIS_CMD_BULK},
{"smove",smoveCommand,4,REDIS_CMD_BULK},
va_start(ap, fmt);
if (level >= server.verbosity) {
- char *c = ".-*";
+ char *c = ".-*#";
char buf[64];
time_t now;
now = time(NULL);
strftime(buf,64,"%d %b %H:%M:%S",localtime(&now));
- fprintf(fp,"%s %c ",buf,c[level]);
+ fprintf(fp,"[%d] %s %c ",(int)getpid(),buf,c[level]);
vfprintf(fp, fmt, ap);
fprintf(fp,"\n");
fflush(fp);
static unsigned int dictEncObjHash(const void *key) {
robj *o = (robj*) key;
- o = getDecodedObject(o);
- unsigned int hash = dictGenHashFunction(o->ptr, sdslen((sds)o->ptr));
- decrRefCount(o);
- return hash;
+ if (o->encoding == REDIS_ENCODING_RAW) {
+ return dictGenHashFunction(o->ptr, sdslen((sds)o->ptr));
+ } else {
+ if (o->encoding == REDIS_ENCODING_INT) {
+ char buf[32];
+ int len;
+
+ len = snprintf(buf,32,"%ld",(long)o->ptr);
+ return dictGenHashFunction((unsigned char*)buf, len);
+ } else {
+ unsigned int hash;
+
+ o = getDecodedObject(o);
+ hash = dictGenHashFunction(o->ptr, sdslen((sds)o->ptr));
+ decrRefCount(o);
+ return hash;
+ }
+ }
}
+/* Sets type and expires */
static dictType setDictType = {
dictEncObjHash, /* hash function */
NULL, /* key dup */
NULL /* val destructor */
};
+/* Sorted sets hash (note: a skiplist is used in addition to the hash table) */
static dictType zsetDictType = {
dictEncObjHash, /* hash function */
NULL, /* key dup */
dictVanillaFree /* val destructor of malloc(sizeof(double)) */
};
+/* Db->dict */
static dictType hashDictType = {
dictObjHash, /* hash function */
NULL, /* key dup */
dictRedisObjectDestructor /* val destructor */
};
+/* Db->expires */
+static dictType keyptrDictType = {
+ dictObjHash, /* hash function */
+ NULL, /* key dup */
+ NULL, /* val dup */
+ dictObjKeyCompare, /* key compare */
+ dictRedisObjectDestructor, /* key destructor */
+ NULL /* val destructor */
+};
+
/* Keylist hash table type has unencoded redis objects as keys and
- * lists as values. It's used for blocking operations (BLPOP) */
+ * lists as values. It's used for blocking operations (BLPOP) and to
+ * map swapped keys to a list of clients waiting for this keys to be loaded. */
static dictType keylistDictType = {
dictObjHash, /* hash function */
NULL, /* key dup */
redisClient *c;
listNode *ln;
time_t now = time(NULL);
+ listIter li;
- listRewind(server.clients);
- while ((ln = listYield(server.clients)) != NULL) {
+ listRewind(server.clients,&li);
+ while ((ln = listNext(&li)) != NULL) {
c = listNodeValue(ln);
if (server.maxidletime &&
!(c->flags & REDIS_SLAVE) && /* no timeout for slaves */
} else if (c->flags & REDIS_BLOCKED) {
if (c->blockingto != 0 && c->blockingto < now) {
addReply(c,shared.nullmultibulk);
- unblockClient(c);
+ unblockClientWaitingData(c);
}
}
}
* To access a global var is faster than calling time(NULL) */
server.unixtime = time(NULL);
- /* Update the global state with the amount of used memory */
- server.usedmemory = zmalloc_used_memory();
-
/* Show some info about non-empty databases */
for (j = 0; j < server.dbnum; j++) {
long long size, used, vkeys;
redisLog(REDIS_VERBOSE,"%d clients connected (%d slaves), %zu bytes in use, %d shared objects",
listLength(server.clients)-listLength(server.slaves),
listLength(server.slaves),
- server.usedmemory,
+ zmalloc_used_memory(),
dictSize(server.sharingpool));
}
/* Close connections of timedout clients */
- if ((server.maxidletime && !(loops % 10)) || server.blockedclients)
+ if ((server.maxidletime && !(loops % 10)) || server.blpop_blocked_clients)
closeTimedoutClients();
/* Check if a background saving or AOF rewrite in progress terminated */
while (server.vm_enabled && zmalloc_used_memory() >
server.vm_max_memory)
{
+ int retval;
+
if (tryFreeOneObjectFromFreelist() == REDIS_OK) continue;
- if (vmSwapOneObjectThreaded() == REDIS_ERR) {
- if ((loops % 30) == 0 && zmalloc_used_memory() >
- (server.vm_max_memory+server.vm_max_memory/10)) {
- redisLog(REDIS_WARNING,"WARNING: vm-max-memory limit exceeded by more than 10%% but unable to swap more objects out!");
- }
+ retval = (server.vm_max_threads == 0) ?
+ vmSwapOneObjectBlocking() :
+ vmSwapOneObjectThreaded();
+ if (retval == REDIS_ERR && (loops % 30) == 0 &&
+ zmalloc_used_memory() >
+ (server.vm_max_memory+server.vm_max_memory/10))
+ {
+ redisLog(REDIS_WARNING,"WARNING: vm-max-memory limit exceeded by more than 10%% but unable to swap more objects out!");
}
- /* Note that we freed just one object, because anyway when
- * the I/O thread in charge to swap this object out will
- * do its work, the handler of completed jobs will try to swap
- * more objects if we are out of memory. */
- break;
+ /* Note that when using threade I/O we free just one object,
+ * because anyway when the I/O thread in charge to swap this
+ * object out will finish, the handler of completed jobs
+ * will try to swap more objects if we are still out of memory. */
+ if (retval == REDIS_ERR || server.vm_max_threads > 0) break;
}
}
return 1000;
}
+/* This function gets called every time Redis is entering the
+ * main loop of the event driven library, that is, before to sleep
+ * for ready file descriptors. */
+static void beforeSleep(struct aeEventLoop *eventLoop) {
+ REDIS_NOTUSED(eventLoop);
+
+ if (server.vm_enabled && listLength(server.io_ready_clients)) {
+ listIter li;
+ listNode *ln;
+
+ listRewind(server.io_ready_clients,&li);
+ while((ln = listNext(&li))) {
+ redisClient *c = ln->value;
+ struct redisCommand *cmd;
+
+ /* Resume the client. */
+ listDelNode(server.io_ready_clients,ln);
+ c->flags &= (~REDIS_IO_WAIT);
+ server.vm_blocked_clients--;
+ aeCreateFileEvent(server.el, c->fd, AE_READABLE,
+ readQueryFromClient, c);
+ cmd = lookupCommand(c->argv[0]->ptr);
+ assert(cmd != NULL);
+ call(c,cmd);
+ resetClient(c);
+ /* There may be more data to process in the input buffer. */
+ if (c->querybuf && sdslen(c->querybuf) > 0)
+ processInputBuffer(c);
+ }
+ }
+}
+
static void createSharedObjects(void) {
shared.crlf = createObject(REDIS_STRING,sdsnew("\r\n"));
shared.ok = createObject(REDIS_STRING,sdsnew("+OK\r\n"));
server.rdbcompression = 1;
server.sharingpoolsize = 1024;
server.maxclients = 0;
- server.blockedclients = 0;
+ server.blpop_blocked_clients = 0;
server.maxmemory = 0;
server.vm_enabled = 0;
+ server.vm_swap_file = zstrdup("/tmp/redis-%p.vm");
server.vm_page_size = 256; /* 256 bytes per page */
server.vm_pages = 1024*1024*100; /* 104 millions of pages */
server.vm_max_memory = 1024LL*1024*1024*1; /* 1 GB of RAM */
server.vm_max_threads = 4;
+ server.vm_blocked_clients = 0;
resetServerSaveParams();
}
for (j = 0; j < server.dbnum; j++) {
server.db[j].dict = dictCreate(&hashDictType,NULL);
- server.db[j].expires = dictCreate(&setDictType,NULL);
+ server.db[j].expires = dictCreate(&keyptrDictType,NULL);
server.db[j].blockingkeys = dictCreate(&keylistDictType,NULL);
+ if (server.vm_enabled)
+ server.db[j].io_keys = dictCreate(&keylistDictType,NULL);
server.db[j].id = j;
}
server.cronloops = 0;
server.bgrewritebuf = sdsempty();
server.lastsave = time(NULL);
server.dirty = 0;
- server.usedmemory = 0;
server.stat_numcommands = 0;
server.stat_numconnections = 0;
server.stat_starttime = time(NULL);
goto loaderr;
}
} else if (!strcasecmp(argv[0],"requirepass") && argc == 2) {
- server.requirepass = zstrdup(argv[1]);
+ server.requirepass = zstrdup(argv[1]);
} else if (!strcasecmp(argv[0],"pidfile") && argc == 2) {
- server.pidfile = zstrdup(argv[1]);
+ server.pidfile = zstrdup(argv[1]);
} else if (!strcasecmp(argv[0],"dbfilename") && argc == 2) {
- server.dbfilename = zstrdup(argv[1]);
+ server.dbfilename = zstrdup(argv[1]);
} else if (!strcasecmp(argv[0],"vm-enabled") && argc == 2) {
if ((server.vm_enabled = yesnotoi(argv[1])) == -1) {
err = "argument must be 'yes' or 'no'"; goto loaderr;
}
+ } else if (!strcasecmp(argv[0],"vm-swap-file") && argc == 2) {
+ zfree(server.vm_swap_file);
+ server.vm_swap_file = zstrdup(argv[1]);
} else if (!strcasecmp(argv[0],"vm-max-memory") && argc == 2) {
server.vm_max_memory = strtoll(argv[1], NULL, 10);
} else if (!strcasecmp(argv[0],"vm-page-size") && argc == 2) {
listNode *ln;
/* Note that if the client we are freeing is blocked into a blocking
- * call, we have to set querybuf to NULL *before* to call unblockClient()
- * to avoid processInputBuffer() will get called. Also it is important
- * to remove the file events after this, because this call adds
- * the READABLE event. */
+ * call, we have to set querybuf to NULL *before* to call
+ * unblockClientWaitingData() to avoid processInputBuffer() will get
+ * called. Also it is important to remove the file events after
+ * this, because this call adds the READABLE event. */
sdsfree(c->querybuf);
c->querybuf = NULL;
if (c->flags & REDIS_BLOCKED)
- unblockClient(c);
+ unblockClientWaitingData(c);
aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
ln = listSearchKey(server.clients,c);
redisAssert(ln != NULL);
listDelNode(server.clients,ln);
- /* Remove from the list of clients waiting for VM operations */
- if (server.vm_enabled && listLength(c->io_keys)) {
- ln = listSearchKey(server.io_clients,c);
- if (ln) listDelNode(server.io_clients,ln);
- listRelease(c->io_keys);
+ /* Remove from the list of clients waiting for swapped keys */
+ if (c->flags & REDIS_IO_WAIT && listLength(c->io_keys) == 0) {
+ ln = listSearchKey(server.io_ready_clients,c);
+ if (ln) {
+ listDelNode(server.io_ready_clients,ln);
+ server.vm_blocked_clients--;
+ }
+ }
+ while (server.vm_enabled && listLength(c->io_keys)) {
+ ln = listFirst(c->io_keys);
+ dontWaitForSwappedKey(c,ln->value);
}
listRelease(c->io_keys);
/* Other cleanup */
int copylen = 0;
char buf[GLUEREPLY_UP_TO];
listNode *ln;
+ listIter li;
robj *o;
- listRewind(c->reply);
- while((ln = listYield(c->reply))) {
+ listRewind(c->reply,&li);
+ while((ln = listNext(&li))) {
int objlen;
o = ln->value;
freeClient(c);
return 0;
}
+
+ /* Now lookup the command and check ASAP about trivial error conditions
+ * such wrong arity, bad command name and so forth. */
cmd = lookupCommand(c->argv[0]->ptr);
if (!cmd) {
addReplySds(c,
resetClient(c);
return 1;
} else if (cmd->flags & REDIS_CMD_BULK && c->bulklen == -1) {
+ /* This is a bulk command, we have to read the last argument yet. */
int bulklen = atoi(c->argv[c->argc-1]->ptr);
decrRefCount(c->argv[c->argc-1]);
c->argc++;
c->querybuf = sdsrange(c->querybuf,c->bulklen,-1);
} else {
+ /* Otherwise return... there is to read the last argument
+ * from the socket. */
return 1;
}
}
queueMultiCommand(c,cmd);
addReply(c,shared.queued);
} else {
+ if (server.vm_enabled && server.vm_max_threads > 0 &&
+ blockClientOnSwappedKeys(cmd,c)) return 1;
call(c,cmd);
}
/* Prepare the client for the next command */
- if (c->flags & REDIS_CLOSE) {
- freeClient(c);
- return 0;
- }
resetClient(c);
return 1;
}
static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int dictid, robj **argv, int argc) {
listNode *ln;
+ listIter li;
int outc = 0, j;
robj **outv;
/* (args*2)+1 is enough room for args, spaces, newlines */
* be sure to free objects if there is no slave in a replication state
* able to be feed with commands */
for (j = 0; j < outc; j++) incrRefCount(outv[j]);
- listRewind(slaves);
- while((ln = listYield(slaves))) {
+ listRewind(slaves,&li);
+ while((ln = listNext(&li))) {
redisClient *slave = ln->value;
/* Don't feed slaves that are still waiting for BGSAVE to start */
o->ptr = ptr;
o->refcount = 1;
if (server.vm_enabled) {
+ /* Note that this code may run in the context of an I/O thread
+ * and accessing to server.unixtime in theory is an error
+ * (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 */
o->vm.atime = server.unixtime;
o->storage = REDIS_VM_MEMORY;
}
static void decrRefCount(void *obj) {
robj *o = obj;
- /* Object is swapped out, 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 &&
(o->storage == REDIS_VM_SWAPPED || o->storage == REDIS_VM_LOADING))
{
/* 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;
static int rdbSaveStringObject(FILE *fp, robj *obj) {
int retval;
- if (obj->storage == REDIS_VM_MEMORY &&
- obj->encoding != REDIS_ENCODING_RAW)
- {
+ /* Avoid incr/decr ref count business when possible.
+ * This plays well with copy-on-write given that we are probably
+ * in a child process (BGSAVE). Also this makes sure key objects
+ * of swapped objects are not incRefCount-ed (an assert does not allow
+ * this in order to avoid bugs) */
+ if (obj->encoding != REDIS_ENCODING_RAW) {
obj = getDecodedObject(obj);
retval = rdbSaveStringObjectRaw(fp,obj);
decrRefCount(obj);
} else {
- /* This is a fast path when we are sure the object is not encoded.
- * Note that's any *faster* actually as we needed to add the conditional
- * but because this may happen in a background process we don't want
- * to touch the object fields with incr/decrRefCount in order to
- * preveny copy on write of pages.
- *
- * Also incrRefCount() will have a failing assert() if we try to call
- * it against an object with storage != REDIS_VM_MEMORY. */
retval = rdbSaveStringObjectRaw(fp,obj);
}
return retval;
} else if (o->type == REDIS_LIST) {
/* Save a list value */
list *list = o->ptr;
+ listIter li;
listNode *ln;
- listRewind(list);
if (rdbSaveLen(fp,listLength(list)) == -1) return -1;
- while((ln = listYield(list))) {
+ listRewind(list,&li);
+ while((ln = listNext(&li))) {
robj *eleobj = listNodeValue(ln);
if (rdbSaveStringObject(fp,eleobj) == -1) return -1;
int j;
time_t now = time(NULL);
+ /* Wait for I/O therads to terminate, just in case this is a
+ * foreground-saving, to avoid seeking the swap file descriptor at the
+ * same time. */
+ if (server.vm_enabled)
+ waitEmptyIOJobsQueue();
+
snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid());
fp = fopen(tmpfile,"w");
if (!fp) {
pid_t childpid;
if (server.bgsavechildpid != -1) return REDIS_ERR;
+ if (server.vm_enabled) waitEmptyIOJobsQueue();
if ((childpid = fork()) == 0) {
/* Child */
+ if (server.vm_enabled) vmReopenSwapFile();
close(server.fd);
if (rdbSave(filename) == REDIS_OK) {
- exit(0);
+ _exit(0);
} else {
- exit(1);
+ _exit(1);
}
} else {
/* Parent */
if ((listlen = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
o = (type == REDIS_LIST) ? createListObject() : createSetObject();
+ /* It's faster to expand the dict to the right size asap in order
+ * to avoid rehashing */
+ if (type == REDIS_SET && listlen > DICT_HT_INITIAL_SIZE)
+ dictExpand(o->ptr,listlen);
/* Load every single element of the list/set */
while(listlen--) {
robj *ele;
incrDecrCommand(c,-incr);
}
+static void appendCommand(redisClient *c) {
+ int retval;
+ size_t totlen;
+ robj *o;
+
+ o = lookupKeyWrite(c->db,c->argv[1]);
+ if (o == NULL) {
+ /* Create the key */
+ retval = dictAdd(c->db->dict,c->argv[1],c->argv[2]);
+ incrRefCount(c->argv[1]);
+ incrRefCount(c->argv[2]);
+ totlen = stringObjectLen(c->argv[2]);
+ } else {
+ dictEntry *de;
+
+ de = dictFind(c->db->dict,c->argv[1]);
+ assert(de != NULL);
+
+ o = dictGetEntryVal(de);
+ if (o->type != REDIS_STRING) {
+ addReply(c,shared.wrongtypeerr);
+ return;
+ }
+ /* If the object is specially encoded or shared we have to make
+ * a copy */
+ if (o->refcount != 1 || o->encoding != REDIS_ENCODING_RAW) {
+ robj *decoded = getDecodedObject(o);
+
+ o = createStringObject(decoded->ptr, sdslen(decoded->ptr));
+ decrRefCount(decoded);
+ dictReplace(c->db->dict,c->argv[1],o);
+ }
+ /* APPEND! */
+ if (c->argv[2]->encoding == REDIS_ENCODING_RAW) {
+ o->ptr = sdscatlen(o->ptr,
+ c->argv[2]->ptr, sdslen(c->argv[2]->ptr));
+ } else {
+ o->ptr = sdscatprintf(o->ptr, "%ld",
+ (unsigned long) c->argv[2]->ptr);
+ }
+ totlen = sdslen(o->ptr);
+ }
+ server.dirty++;
+ addReplySds(c,sdscatprintf(sdsempty(),":%lu\r\n",(unsigned long)totlen));
+}
+
/* ========================= Type agnostic commands ========================= */
static void delCommand(redisClient *c) {
if (server.appendonly) {
/* Append only file: fsync() the AOF and exit */
fsync(server.appendfd);
+ if (server.vm_enabled) unlink(server.vm_swap_file);
exit(0);
} else {
/* Snapshotting. Perform a SYNC SAVE and exit */
unlink(server.pidfile);
redisLog(REDIS_WARNING,"%zu bytes used at exit",zmalloc_used_memory());
redisLog(REDIS_WARNING,"Server exit now, bye bye...");
+ if (server.vm_enabled) unlink(server.vm_swap_file);
exit(0);
} else {
/* Ooops.. error saving! The best we can do is to continue operating.
if (sortval->type == REDIS_LIST) {
list *list = sortval->ptr;
listNode *ln;
+ listIter li;
- listRewind(list);
- while((ln = listYield(list))) {
+ listRewind(list,&li);
+ while((ln = listNext(&li))) {
robj *ele = ln->value;
vector[j].obj = ele;
vector[j].u.score = 0;
addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",outputlen));
for (j = start; j <= end; j++) {
listNode *ln;
+ listIter li;
+
if (!getop) {
addReplyBulkLen(c,vector[j].obj);
addReply(c,vector[j].obj);
addReply(c,shared.crlf);
}
- listRewind(operations);
- while((ln = listYield(operations))) {
+ listRewind(operations,&li);
+ while((ln = listNext(&li))) {
redisSortOperation *sop = ln->value;
robj *val = lookupKeyByPattern(c->db,sop->pattern,
vector[j].obj);
/* STORE option specified, set the sorting result as a List object */
for (j = start; j <= end; j++) {
listNode *ln;
+ listIter li;
+
if (!getop) {
listAddNodeTail(listPtr,vector[j].obj);
incrRefCount(vector[j].obj);
}
- listRewind(operations);
- while((ln = listYield(operations))) {
+ listRewind(operations,&li);
+ while((ln = listNext(&li))) {
redisSortOperation *sop = ln->value;
robj *val = lookupKeyByPattern(c->db,sop->pattern,
vector[j].obj);
sprintf(s,"%.2fM",d);
} else if (n < (1024LL*1024*1024*1024)) {
d = (double)n/(1024LL*1024*1024);
- sprintf(s,"%.2fM",d);
+ sprintf(s,"%.2fG",d);
}
}
int j;
char hmem[64];
- bytesToHuman(hmem,server.usedmemory);
+ bytesToHuman(hmem,zmalloc_used_memory());
info = sdscatprintf(sdsempty(),
"redis_version:%s\r\n"
"arch_bits:%s\r\n"
uptime/(3600*24),
listLength(server.clients)-listLength(server.slaves),
listLength(server.slaves),
- server.blockedclients,
- server.usedmemory,
+ server.blpop_blocked_clients,
+ zmalloc_used_memory(),
hmem,
server.dirty,
server.bgsavechildpid != -1,
);
}
if (server.vm_enabled) {
+ lockThreadedIO();
info = sdscatprintf(info,
"vm_conf_max_memory:%llu\r\n"
"vm_conf_page_size:%llu\r\n"
"vm_stats_io_newjobs_len:%lu\r\n"
"vm_stats_io_processing_len:%lu\r\n"
"vm_stats_io_processed_len:%lu\r\n"
- "vm_stats_io_waiting_clients:%lu\r\n"
+ "vm_stats_io_active_threads:%lu\r\n"
+ "vm_stats_blocked_clients:%lu\r\n"
,(unsigned long long) server.vm_max_memory,
(unsigned long long) server.vm_page_size,
(unsigned long long) server.vm_pages,
(unsigned long) listLength(server.io_newjobs),
(unsigned long) listLength(server.io_processing),
(unsigned long) listLength(server.io_processed),
- (unsigned long) listLength(server.io_clients)
+ (unsigned long) server.io_active_threads,
+ (unsigned long) server.vm_blocked_clients
);
+ unlockThreadedIO();
}
for (j = 0; j < server.dbnum; j++) {
long long keys, vkeys;
/* Mark the client as a blocked client */
c->flags |= REDIS_BLOCKED;
aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
- server.blockedclients++;
+ server.blpop_blocked_clients++;
}
/* Unblock a client that's waiting in a blocking operation such as BLPOP */
-static void unblockClient(redisClient *c) {
+static void unblockClientWaitingData(redisClient *c) {
dictEntry *de;
list *l;
int j;
zfree(c->blockingkeys);
c->blockingkeys = NULL;
c->flags &= (~REDIS_BLOCKED);
- server.blockedclients--;
+ server.blpop_blocked_clients--;
/* Ok now we are ready to get read events from socket, note that we
- * can't trap errors here as it's possible that unblockClients() is
+ * can't trap errors here as it's possible that unblockClientWaitingDatas() is
* called from freeClient() itself, and the only thing we can do
* if we failed to register the READABLE event is to kill the client.
* Still the following function should never fail in the real world as
* we are sure the file descriptor is sane, and we exit on out of mem. */
aeCreateFileEvent(server.el, c->fd, AE_READABLE, readQueryFromClient, c);
/* As a final step we want to process data if there is some command waiting
- * in the input buffer. Note that this is safe even if unblockClient()
- * gets called from freeClient() because freeClient() will be smart
- * enough to call this function *after* c->querybuf was set to NULL. */
+ * in the input buffer. Note that this is safe even if
+ * unblockClientWaitingData() gets called from freeClient() because
+ * freeClient() will be smart enough to call this function
+ * *after* c->querybuf was set to NULL. */
if (c->querybuf && sdslen(c->querybuf) > 0) processInputBuffer(c);
}
addReplyBulkLen(receiver,ele);
addReply(receiver,ele);
addReply(receiver,shared.crlf);
- unblockClient(receiver);
+ unblockClientWaitingData(receiver);
return 1;
}
* registering differences since the server forked to save */
redisClient *slave;
listNode *ln;
+ listIter li;
- listRewind(server.slaves);
- while((ln = listYield(server.slaves))) {
+ listRewind(server.slaves,&li);
+ while((ln = listNext(&li))) {
slave = ln->value;
if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) break;
}
static void updateSlavesWaitingBgsave(int bgsaveerr) {
listNode *ln;
int startbgsave = 0;
+ listIter li;
- listRewind(server.slaves);
- while((ln = listYield(server.slaves))) {
+ listRewind(server.slaves,&li);
+ while((ln = listNext(&li))) {
redisClient *slave = ln->value;
if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
}
if (startbgsave) {
if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {
- listRewind(server.slaves);
+ listIter li;
+
+ listRewind(server.slaves,&li);
redisLog(REDIS_WARNING,"SYNC failed. BGSAVE failed");
- while((ln = listYield(server.slaves))) {
+ while((ln = listNext(&li))) {
redisClient *slave = ln->value;
if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START)
char buf[128];
int decrrc = 0;
- if (obj->storage == REDIS_VM_MEMORY && obj->encoding != REDIS_ENCODING_RAW){
+ /* Avoid the incr/decr ref count business if possible to help
+ * copy-on-write (we are often in a child process when this function
+ * is called).
+ * Also makes sure that key objects don't get incrRefCount-ed when VM
+ * is enabled */
+ if (obj->encoding != REDIS_ENCODING_RAW) {
obj = getDecodedObject(obj);
decrrc = 1;
}
/* Emit the RPUSHes needed to rebuild the list */
list *list = o->ptr;
listNode *ln;
+ listIter li;
- listRewind(list);
- while((ln = listYield(list))) {
+ listRewind(list,&li);
+ while((ln = listNext(&li))) {
char cmd[]="*3\r\n$5\r\nRPUSH\r\n";
robj *eleobj = listNodeValue(ln);
pid_t childpid;
if (server.bgrewritechildpid != -1) return REDIS_ERR;
+ if (server.vm_enabled) waitEmptyIOJobsQueue();
if ((childpid = fork()) == 0) {
/* Child */
char tmpfile[256];
- close(server.fd);
+ if (server.vm_enabled) vmReopenSwapFile();
+ close(server.fd);
snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) {
- exit(0);
+ _exit(0);
} else {
- exit(1);
+ _exit(1);
}
} else {
/* Parent */
*/
/* =================== Virtual Memory - Blocking Side ====================== */
+
+/* substitute the first occurrence of '%p' with the process pid in the
+ * swap file name. */
+static void expandVmSwapFilename(void) {
+ char *p = strstr(server.vm_swap_file,"%p");
+ sds new;
+
+ if (!p) return;
+ new = sdsempty();
+ *p = '\0';
+ new = sdscat(new,server.vm_swap_file);
+ new = sdscatprintf(new,"%ld",(long) getpid());
+ new = sdscat(new,p+2);
+ zfree(server.vm_swap_file);
+ server.vm_swap_file = new;
+}
+
static void vmInit(void) {
off_t totsize;
int pipefds[2];
+ size_t stacksize;
+
+ if (server.vm_max_threads != 0)
+ zmalloc_enable_thread_safeness(); /* we need thread safe zmalloc() */
- server.vm_fp = fopen("/tmp/redisvm","w+b");
+ expandVmSwapFilename();
+ redisLog(REDIS_NOTICE,"Using '%s' as swap file",server.vm_swap_file);
+ if ((server.vm_fp = fopen(server.vm_swap_file,"r+b")) == NULL) {
+ server.vm_fp = fopen(server.vm_swap_file,"w+b");
+ }
if (server.vm_fp == NULL) {
- redisLog(REDIS_WARNING,"Impossible to open the swap file. Exiting.");
+ redisLog(REDIS_WARNING,
+ "Impossible to open the swap file: %s. Exiting.",
+ strerror(errno));
exit(1);
}
server.vm_fd = fileno(server.vm_fp);
redisLog(REDIS_VERBOSE,"Allocated %lld bytes page table for %lld pages",
(long long) (server.vm_pages+7)/8, server.vm_pages);
memset(server.vm_bitmap,0,(server.vm_pages+7)/8);
- /* Try to remove the swap file, so the OS will really delete it from the
- * file system when Redis exists. */
- unlink("/tmp/redisvm");
/* Initialize threaded I/O (used by Virtual Memory) */
server.io_newjobs = listCreate();
server.io_processing = listCreate();
server.io_processed = listCreate();
- server.io_clients = listCreate();
+ server.io_ready_clients = listCreate();
pthread_mutex_init(&server.io_mutex,NULL);
pthread_mutex_init(&server.obj_freelist_mutex,NULL);
pthread_mutex_init(&server.io_swapfile_mutex,NULL);
server.io_ready_pipe_read = pipefds[0];
server.io_ready_pipe_write = pipefds[1];
redisAssert(anetNonBlock(NULL,server.io_ready_pipe_read) != ANET_ERR);
+ /* LZF requires a lot of stack */
+ pthread_attr_init(&server.io_threads_attr);
+ pthread_attr_getstacksize(&server.io_threads_attr, &stacksize);
+ while (stacksize < REDIS_THREAD_STACK_SIZE) stacksize *= 2;
+ pthread_attr_setstacksize(&server.io_threads_attr, stacksize);
/* Listen for events in the threaded I/O pipe */
if (aeCreateFileEvent(server.el, server.io_ready_pipe_read, AE_READABLE,
vmThreadedIOCompletedJob, NULL) == AE_ERR)
static void vmMarkPageUsed(off_t page) {
off_t byte = page/8;
int bit = page&7;
+ redisAssert(vmFreePage(page) == 1);
server.vm_bitmap[byte] |= 1<<bit;
redisLog(REDIS_DEBUG,"Mark used: %lld (byte:%lld bit:%d)\n",
(long long)page, (long long)byte, bit);
static void vmMarkPageFree(off_t page) {
off_t byte = page/8;
int bit = page&7;
+ redisAssert(vmFreePage(page) == 0);
server.vm_bitmap[byte] &= ~(1<<bit);
+ redisLog(REDIS_DEBUG,"Mark free: %lld (byte:%lld bit:%d)\n",
+ (long long)page, (long long)byte, bit);
}
/* Mark N contiguous pages as free, with 'page' being the first. */
for (j = 0; j < count; j++)
vmMarkPageFree(page+j);
server.vm_stats_used_pages -= count;
+ if (server.vm_stats_used_pages > 100000000) {
+ *((char*)-1) = 'x';
+ }
}
/* Test if the page is free */
* note: I implemented this function just after watching an episode of
* Battlestar Galactica, where the hybrid was continuing to say "JUMP!"
*/
-static int vmFindContiguousPages(off_t *first, int n) {
+static int vmFindContiguousPages(off_t *first, off_t n) {
off_t base, offset = 0, since_jump = 0, numfree = 0;
if (server.vm_near_pages == REDIS_VM_MAX_NEAR_PAGES) {
while(offset < server.vm_pages) {
off_t this = base+offset;
- redisLog(REDIS_DEBUG, "THIS: %lld (%c)\n", (long long) this, vmFreePage(this) ? 'F' : 'X');
/* If we overflow, restart from page zero */
if (this >= server.vm_pages) {
this -= server.vm_pages;
numfree = 0;
}
}
+ redisLog(REDIS_DEBUG, "THIS: %lld (%c)\n", (long long) this, vmFreePage(this) ? 'F' : 'X');
if (vmFreePage(this)) {
/* This is a free page */
numfree++;
if (server.vm_enabled) pthread_mutex_lock(&server.io_swapfile_mutex);
if (fseeko(server.vm_fp,page*server.vm_page_size,SEEK_SET) == -1) {
redisLog(REDIS_WARNING,
- "Unrecoverable VM problem in vmLoadObject(): can't seek: %s",
+ "Unrecoverable VM problem in vmReadObjectFromSwap(): can't seek: %s",
strerror(errno));
- exit(1);
+ _exit(1);
}
o = rdbLoadObject(type,server.vm_fp);
if (o == NULL) {
- redisLog(REDIS_WARNING, "Unrecoverable VM problem in vmLoadObject(): can't load object from swap file: %s", strerror(errno));
- exit(1);
+ redisLog(REDIS_WARNING, "Unrecoverable VM problem in vmReadObjectFromSwap(): can't load object from swap file: %s", strerror(errno));
+ _exit(1);
}
if (server.vm_enabled) pthread_mutex_unlock(&server.io_swapfile_mutex);
return o;
static robj *vmGenericLoadObject(robj *key, int preview) {
robj *val;
- redisAssert(key->storage == REDIS_VM_SWAPPED);
+ redisAssert(key->storage == REDIS_VM_SWAPPED || key->storage == REDIS_VM_LOADING);
val = vmReadObjectFromSwap(key->vm.page,key->vtype);
if (!preview) {
key->storage = REDIS_VM_MEMORY;
for (j = 0; j < server.dbnum; j++) {
redisDb *db = server.db+j;
- int maxtries = 1000;
+ /* Why maxtries is set to 100?
+ * Because this way (usually) we'll find 1 object even if just 1% - 2%
+ * are swappable objects */
+ int maxtries = 100;
if (dictSize(db->dict) == 0) continue;
for (i = 0; i < 5; i++) {
de = dictGetRandomKey(db->dict);
key = dictGetEntryKey(de);
val = dictGetEntryVal(de);
- if (key->storage != REDIS_VM_MEMORY) {
+ /* Only swap objects that are currently in memory.
+ *
+ * Also don't swap shared objects if threaded VM is on, as we
+ * try to ensure that the main thread does not touch the
+ * object while the I/O thread is using it, but we can't
+ * control other keys without adding additional mutex. */
+ if (key->storage != REDIS_VM_MEMORY ||
+ (server.vm_max_threads != 0 && val->refcount != 1)) {
if (maxtries) i--; /* don't count this try */
continue;
}
/* =================== Virtual Memory - Threaded I/O ======================= */
static void freeIOJob(iojob *j) {
- if (j->type == REDIS_IOJOB_PREPARE_SWAP ||
- j->type == REDIS_IOJOB_DO_SWAP)
+ if ((j->type == REDIS_IOJOB_PREPARE_SWAP ||
+ j->type == REDIS_IOJOB_DO_SWAP ||
+ j->type == REDIS_IOJOB_LOAD) && j->val != NULL)
decrRefCount(j->val);
decrRefCount(j->key);
zfree(j);
int mask)
{
char buf[1];
- int retval;
+ int retval, processed = 0, toprocess = -1, trytoswap = 1;
REDIS_NOTUSED(el);
REDIS_NOTUSED(mask);
REDIS_NOTUSED(privdata);
struct dictEntry *de;
redisLog(REDIS_DEBUG,"Processing I/O completed job");
- assert(listLength(server.io_processed) != 0);
/* Get the processed element (the oldest one) */
lockThreadedIO();
+ assert(listLength(server.io_processed) != 0);
+ if (toprocess == -1) {
+ toprocess = (listLength(server.io_processed)*REDIS_MAX_COMPLETED_JOBS_PROCESSED)/100;
+ if (toprocess <= 0) toprocess = 1;
+ }
ln = listFirst(server.io_processed);
j = ln->value;
listDelNode(server.io_processed,ln);
assert(de != NULL);
key = dictGetEntryKey(de);
if (j->type == REDIS_IOJOB_LOAD) {
+ redisDb *db;
+
/* Key loaded, bring it at home */
key->storage = REDIS_VM_MEMORY;
key->vm.atime = server.unixtime;
(unsigned char*) key->ptr);
server.vm_stats_swapped_objects--;
server.vm_stats_swapins++;
+ dictGetEntryVal(de) = j->val;
+ incrRefCount(j->val);
+ db = j->db;
freeIOJob(j);
+ /* Handle clients waiting for this key to be loaded. */
+ handleClientsBlockedOnSwappedKey(db,key);
} else if (j->type == REDIS_IOJOB_PREPARE_SWAP) {
/* Now we know the amount of pages required to swap this object.
* Let's find some space for it, and queue this task again
* rebranded as REDIS_IOJOB_DO_SWAP. */
- if (vmFindContiguousPages(&j->page,j->pages) == REDIS_ERR) {
- /* Ooops... no space! */
+ if (!vmCanSwapOut() ||
+ vmFindContiguousPages(&j->page,j->pages) == REDIS_ERR)
+ {
+ /* Ooops... no space or we can't swap as there is
+ * a fork()ed Redis trying to save stuff on disk. */
freeIOJob(j);
+ key->storage = REDIS_VM_MEMORY; /* undo operation */
} else {
+ /* Note that we need to mark this pages as used now,
+ * if the job will be canceled, we'll mark them as freed
+ * again. */
+ vmMarkPagesUsed(j->page,j->pages);
j->type = REDIS_IOJOB_DO_SWAP;
lockThreadedIO();
queueIOJob(j);
key->vtype = j->val->type;
decrRefCount(val); /* Deallocate the object from memory. */
dictGetEntryVal(de) = NULL;
- vmMarkPagesUsed(j->page,j->pages);
redisLog(REDIS_DEBUG,
"VM: object %s swapped out at %lld (%lld pages) (threaded)",
(unsigned char*) key->ptr,
freeIOJob(j);
/* Put a few more swap requests in queue if we are still
* out of memory */
- if (zmalloc_used_memory() > server.vm_max_memory) {
+ if (trytoswap && vmCanSwapOut() &&
+ zmalloc_used_memory() > server.vm_max_memory)
+ {
int more = 1;
while(more) {
lockThreadedIO();
(unsigned) server.vm_max_threads;
unlockThreadedIO();
/* Don't waste CPU time if swappable objects are rare. */
- if (vmSwapOneObjectThreaded() == REDIS_ERR) break;
+ if (vmSwapOneObjectThreaded() == REDIS_ERR) {
+ trytoswap = 0;
+ break;
+ }
}
}
}
- return; /* XXX REMOVE ME */
+ processed++;
+ if (processed == toprocess) return;
}
if (retval < 0 && errno != EAGAIN) {
redisLog(REDIS_WARNING,
int i;
assert(o->storage == REDIS_VM_LOADING || o->storage == REDIS_VM_SWAPPING);
+again:
lockThreadedIO();
/* Search for a matching key in one of the queues */
for (i = 0; i < 3; i++) {
listNode *ln;
+ listIter li;
- listRewind(lists[i]);
- while ((ln = listYield(lists[i])) != NULL) {
+ listRewind(lists[i],&li);
+ while ((ln = listNext(&li)) != NULL) {
iojob *job = ln->value;
if (job->canceled) continue; /* Skip this, already canceled. */
if (compareStringObjects(job->key,o) == 0) {
- redisLog(REDIS_DEBUG,"*** CANCELED %p (%s)\n",
- (void*)job, (char*)o->ptr);
+ redisLog(REDIS_DEBUG,"*** CANCELED %p (%s) (type %d) (LIST ID %d)\n",
+ (void*)job, (char*)o->ptr, job->type, i);
+ /* Mark the pages as free since the swap didn't happened
+ * or happened but is now discarded. */
+ if (i != 1 && job->type == REDIS_IOJOB_DO_SWAP)
+ vmMarkPagesFree(job->page,job->pages);
+ /* Cancel the job. It depends on the list the job is
+ * living in. */
switch(i) {
case 0: /* io_newjobs */
/* If the job was yet not processed the best thing to do
listDelNode(lists[i],ln);
break;
case 1: /* io_processing */
+ /* Oh Shi- the thread is messing with the Job:
+ *
+ * Probably it's accessing the object if this is a
+ * PREPARE_SWAP or DO_SWAP job.
+ * If it's a LOAD job it may be reading from disk and
+ * if we don't wait for the job to terminate before to
+ * cancel it, maybe in a few microseconds data can be
+ * corrupted in this pages. So the short story is:
+ *
+ * Better to wait for the job to move into the
+ * next queue (processed)... */
+
+ /* We try again and again until the job is completed. */
+ unlockThreadedIO();
+ /* But let's wait some time for the I/O thread
+ * to finish with this job. After all this condition
+ * should be very rare. */
+ usleep(1);
+ goto again;
case 2: /* io_processed */
+ /* The job was already processed, that's easy...
+ * just mark it as canceled so that we'll ignore it
+ * when processing completed jobs. */
job->canceled = 1;
break;
}
+ /* Finally we have to adjust the storage type of the object
+ * in order to "UNDO" the operaiton. */
if (o->storage == REDIS_VM_LOADING)
o->storage = REDIS_VM_SWAPPED;
else if (o->storage == REDIS_VM_SWAPPING)
lockThreadedIO();
if (listLength(server.io_newjobs) == 0) {
/* No new jobs in queue, exit. */
- redisLog(REDIS_DEBUG,"Thread %lld exiting, nothing to do\n",
+ redisLog(REDIS_DEBUG,"Thread %lld exiting, nothing to do",
(long long) pthread_self());
server.io_active_threads--;
unlockThreadedIO();
listAddNodeTail(server.io_processing,j);
ln = listLast(server.io_processing); /* We use ln later to remove it */
unlockThreadedIO();
- redisLog(REDIS_DEBUG,"Thread %lld got a new job (type %d): %p about key '%s'\n",
+ redisLog(REDIS_DEBUG,"Thread %lld got a new job (type %d): %p about key '%s'",
(long long) pthread_self(), j->type, (void*)j, (char*)j->key->ptr);
/* Process the Job */
if (j->type == REDIS_IOJOB_LOAD) {
+ j->val = vmReadObjectFromSwap(j->page,j->key->vtype);
} else if (j->type == REDIS_IOJOB_PREPARE_SWAP) {
FILE *fp = fopen("/dev/null","w+");
j->pages = rdbSavedObjectPages(j->val,fp);
}
/* Done: insert the job into the processed queue */
- redisLog(REDIS_DEBUG,"Thread %lld completed the job: %p (key %s)\n",
+ redisLog(REDIS_DEBUG,"Thread %lld completed the job: %p (key %s)",
(long long) pthread_self(), (void*)j, (char*)j->key->ptr);
lockThreadedIO();
listDelNode(server.io_processing,ln);
static void spawnIOThread(void) {
pthread_t thread;
-
- pthread_create(&thread,NULL,IOThreadEntryPoint,NULL);
+ sigset_t mask, omask;
+
+ sigemptyset(&mask);
+ sigaddset(&mask,SIGCHLD);
+ sigaddset(&mask,SIGHUP);
+ sigaddset(&mask,SIGPIPE);
+ pthread_sigmask(SIG_SETMASK, &mask, &omask);
+ pthread_create(&thread,&server.io_threads_attr,IOThreadEntryPoint,NULL);
+ pthread_sigmask(SIG_SETMASK, &omask, NULL);
server.io_active_threads++;
}
+/* We need to wait for the last thread to exit before we are able to
+ * fork() in order to BGSAVE or BGREWRITEAOF. */
+static void waitEmptyIOJobsQueue(void) {
+ while(1) {
+ int io_processed_len;
+
+ lockThreadedIO();
+ if (listLength(server.io_newjobs) == 0 &&
+ listLength(server.io_processing) == 0 &&
+ server.io_active_threads == 0)
+ {
+ unlockThreadedIO();
+ return;
+ }
+ /* While waiting for empty jobs queue condition we post-process some
+ * finshed job, as I/O threads may be hanging trying to write against
+ * the io_ready_pipe_write FD but there are so much pending jobs that
+ * it's blocking. */
+ io_processed_len = listLength(server.io_processed);
+ unlockThreadedIO();
+ if (io_processed_len) {
+ vmThreadedIOCompletedJob(NULL,server.io_ready_pipe_read,NULL,0);
+ usleep(1000); /* 1 millisecond */
+ } else {
+ usleep(10000); /* 10 milliseconds */
+ }
+ }
+}
+
+static void vmReopenSwapFile(void) {
+ /* Note: we don't close the old one as we are in the child process
+ * and don't want to mess at all with the original file object. */
+ server.vm_fp = fopen(server.vm_swap_file,"r+b");
+ if (server.vm_fp == NULL) {
+ redisLog(REDIS_WARNING,"Can't re-open the VM swap file: %s. Exiting.",
+ server.vm_swap_file);
+ _exit(1);
+ }
+ server.vm_fd = fileno(server.vm_fp);
+}
+
/* This function must be called while with threaded IO locked */
static void queueIOJob(iojob *j) {
redisLog(REDIS_DEBUG,"Queued IO Job %p type %d about key '%s'\n",
return REDIS_OK;
}
+/* ============ Virtual Memory - Blocking clients on missing keys =========== */
+
+/* This function makes the clinet 'c' waiting for the key 'key' to be loaded.
+ * If there is not already a job loading the key, it is craeted.
+ * The key is added to the io_keys list in the client structure, and also
+ * in the hash table mapping swapped keys to waiting clients, that is,
+ * server.io_waited_keys. */
+static int waitForSwappedKey(redisClient *c, robj *key) {
+ struct dictEntry *de;
+ robj *o;
+ list *l;
+
+ /* If the key does not exist or is already in RAM we don't need to
+ * block the client at all. */
+ de = dictFind(c->db->dict,key);
+ if (de == NULL) return 0;
+ o = dictGetEntryKey(de);
+ if (o->storage == REDIS_VM_MEMORY) {
+ return 0;
+ } else if (o->storage == REDIS_VM_SWAPPING) {
+ /* We were swapping the key, undo it! */
+ vmCancelThreadedIOJob(o);
+ return 0;
+ }
+
+ /* OK: the key is either swapped, or being loaded just now. */
+
+ /* Add the key to the list of keys this client is waiting for.
+ * This maps clients to keys they are waiting for. */
+ listAddNodeTail(c->io_keys,key);
+ incrRefCount(key);
+
+ /* Add the client to the swapped keys => clients waiting map. */
+ de = dictFind(c->db->io_keys,key);
+ if (de == NULL) {
+ int retval;
+
+ /* For every key we take a list of clients blocked for it */
+ l = listCreate();
+ retval = dictAdd(c->db->io_keys,key,l);
+ incrRefCount(key);
+ assert(retval == DICT_OK);
+ } else {
+ l = dictGetEntryVal(de);
+ }
+ listAddNodeTail(l,c);
+
+ /* Are we already loading the key from disk? If not create a job */
+ if (o->storage == REDIS_VM_SWAPPED) {
+ iojob *j;
+
+ o->storage = REDIS_VM_LOADING;
+ j = zmalloc(sizeof(*j));
+ j->type = REDIS_IOJOB_LOAD;
+ j->db = c->db;
+ j->key = dupStringObject(key);
+ j->key->vtype = o->vtype;
+ j->page = o->vm.page;
+ j->val = NULL;
+ j->canceled = 0;
+ j->thread = (pthread_t) -1;
+ lockThreadedIO();
+ queueIOJob(j);
+ unlockThreadedIO();
+ }
+ return 1;
+}
+
+/* Is this client attempting to run a command against swapped keys?
+ * If so, block it ASAP, load the keys in background, then resume it.
+ *
+ * The important idea about this function is that it can fail! If keys will
+ * still be swapped when the client is resumed, this key lookups will
+ * just block loading keys from disk. In practical terms this should only
+ * happen with SORT BY command or if there is a bug in this function.
+ *
+ * Return 1 if the client is marked as blocked, 0 if the client can
+ * continue as the keys it is going to access appear to be in memory. */
+static int blockClientOnSwappedKeys(struct redisCommand *cmd, redisClient *c) {
+ if (cmd->proc == getCommand) {
+ waitForSwappedKey(c,c->argv[1]);
+ }
+ /* If the client was blocked for at least one key, mark it as blocked. */
+ if (listLength(c->io_keys)) {
+ c->flags |= REDIS_IO_WAIT;
+ aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
+ server.vm_blocked_clients++;
+ return 1;
+ } else {
+ return 0;
+ }
+}
+
+/* Remove the 'key' from the list of blocked keys for a given client.
+ *
+ * The function returns 1 when there are no longer blocking keys after
+ * the current one was removed (and the client can be unblocked). */
+static int dontWaitForSwappedKey(redisClient *c, robj *key) {
+ list *l;
+ listNode *ln;
+ listIter li;
+ struct dictEntry *de;
+
+ /* Remove the key from the list of keys this client is waiting for. */
+ listRewind(c->io_keys,&li);
+ while ((ln = listNext(&li)) != NULL) {
+ if (compareStringObjects(ln->value,key) == 0) {
+ listDelNode(c->io_keys,ln);
+ break;
+ }
+ }
+ assert(ln != NULL);
+
+ /* Remove the client form the key => waiting clients map. */
+ de = dictFind(c->db->io_keys,key);
+ assert(de != NULL);
+ l = dictGetEntryVal(de);
+ ln = listSearchKey(l,c);
+ assert(ln != NULL);
+ listDelNode(l,ln);
+ if (listLength(l) == 0)
+ dictDelete(c->db->io_keys,key);
+
+ return listLength(c->io_keys) == 0;
+}
+
+static void handleClientsBlockedOnSwappedKey(redisDb *db, robj *key) {
+ struct dictEntry *de;
+ list *l;
+ listNode *ln;
+ int len;
+
+ de = dictFind(db->io_keys,key);
+ if (!de) return;
+
+ l = dictGetEntryVal(de);
+ len = listLength(l);
+ /* Note: we can't use something like while(listLength(l)) as the list
+ * can be freed by the calling function when we remove the last element. */
+ while (len--) {
+ ln = listFirst(l);
+ redisClient *c = ln->value;
+
+ if (dontWaitForSwappedKey(c,key)) {
+ /* Put the client in the list of clients ready to go as we
+ * loaded all the keys about it. */
+ listAddNodeTail(server.io_ready_clients,c);
+ }
+ }
+}
+
/* ================================= Debugging ============================== */
static void debugCommand(redisClient *c) {
}
key = dictGetEntryKey(de);
val = dictGetEntryVal(de);
- if (server.vm_enabled && (key->storage == REDIS_VM_MEMORY ||
- key->storage == REDIS_VM_SWAPPING)) {
+ if (!server.vm_enabled || (key->storage == REDIS_VM_MEMORY ||
+ key->storage == REDIS_VM_SWAPPING)) {
addReplySds(c,sdscatprintf(sdsempty(),
"+Key at:%p refcount:%d, value at:%p refcount:%d "
"encoding:%d serializedlength:%lld\r\n",
(void*)key, key->refcount, (void*)val, val->refcount,
- val->encoding, rdbSavedObjectLen(val,NULL)));
+ val->encoding, (long long) rdbSavedObjectLen(val,NULL)));
} else {
addReplySds(c,sdscatprintf(sdsempty(),
"+Key at:%p refcount:%d, value swapped at: page %llu "
}
int main(int argc, char **argv) {
+ time_t start;
+
initServerConfig();
if (argc == 2) {
resetServerSaveParams();
#ifdef __linux__
linuxOvercommitMemoryWarning();
#endif
+ start = time(NULL);
if (server.appendonly) {
if (loadAppendOnlyFile(server.appendfilename) == REDIS_OK)
- redisLog(REDIS_NOTICE,"DB loaded from append only file");
+ redisLog(REDIS_NOTICE,"DB loaded from append only file: %ld seconds",time(NULL)-start);
} else {
if (rdbLoad(server.dbfilename) == REDIS_OK)
- redisLog(REDIS_NOTICE,"DB loaded from disk");
+ redisLog(REDIS_NOTICE,"DB loaded from disk: %ld seconds",time(NULL)-start);
}
redisLog(REDIS_NOTICE,"The server is now ready to accept connections on port %d", server.port);
+ aeSetBeforeSleepProc(server.el,beforeSleep);
aeMain(server.el);
aeDeleteEventLoop(server.el);
return 0;
}
}
/* free(messages); Don't call free() with possibly corrupted memory. */
- exit(0);
+ _exit(0);
}
static void setupSigSegvAction(void) {
projects / redis.git / blobdiff