From: antirez Date: Tue, 4 Jan 2011 18:07:15 +0000 (+0100) Subject: Merge branch 'master' into unstable X-Git-Url: https://git.saurik.com/redis.git/commitdiff_plain/6eaad66373982aa9c1c0ff6b584c542670548d0e?hp=04a2ade90dd15db2a1e685e2e045b8c0181b9eea Merge branch 'master' into unstable --- diff --git a/.gitignore b/.gitignore index 90c94621..9fad5da7 100644 --- a/.gitignore +++ b/.gitignore @@ -18,3 +18,5 @@ release.h src/transfer.sh src/configs src/redis-server.dSYM +redis.ds +src/redis.conf diff --git a/redis.conf b/redis.conf index 3efd1e59..3425a59c 100644 --- a/redis.conf +++ b/redis.conf @@ -291,77 +291,31 @@ appendfsync everysec # "no" that is the safest pick from the point of view of durability. no-appendfsync-on-rewrite no -################################ VIRTUAL MEMORY ############################### - -# Virtual Memory allows Redis to work with datasets bigger than the actual -# amount of RAM needed to hold the whole dataset in memory. -# In order to do so very used keys are taken in memory while the other keys -# are swapped into a swap file, similarly to what operating systems do -# with memory pages. -# -# To enable VM just set 'vm-enabled' to yes, and set the following three -# VM parameters accordingly to your needs. - -vm-enabled no -# vm-enabled yes - -# This is the path of the Redis swap file. As you can guess, swap files -# can't be shared by different Redis instances, so make sure to use a swap -# file for every redis process you are running. Redis will complain if the -# swap file is already in use. -# -# The best kind of storage for the Redis swap file (that's accessed at random) -# is a Solid State Disk (SSD). -# -# *** WARNING *** if you are using a shared hosting the default of putting -# the swap file under /tmp is not secure. Create a dir with access granted -# only to Redis user and configure Redis to create the swap file there. -vm-swap-file /tmp/redis.swap - -# vm-max-memory configures the VM to use at max the specified amount of -# RAM. Everything that deos not fit will be swapped on disk *if* possible, that -# is, if there is still enough contiguous space in the swap file. -# -# With vm-max-memory 0 the system will swap everything it can. Not a good -# default, just specify the max amount of RAM you can in bytes, but it's -# better to leave some margin. For instance specify an amount of RAM -# that's more or less between 60 and 80% of your free RAM. -vm-max-memory 0 - -# Redis swap files is split into pages. An object can be saved using multiple -# contiguous pages, but pages can't be shared between different objects. -# So if your page is too big, small objects swapped out on disk will waste -# a lot of space. If you page is too small, there is less space in the swap -# file (assuming you configured the same number of total swap file pages). -# -# If you use a lot of small objects, use a page size of 64 or 32 bytes. -# If you use a lot of big objects, use a bigger page size. -# If unsure, use the default :) -vm-page-size 32 - -# Number of total memory pages in the swap file. -# Given that the page table (a bitmap of free/used pages) is taken in memory, -# every 8 pages on disk will consume 1 byte of RAM. -# -# The total swap size is vm-page-size * vm-pages -# -# With the default of 32-bytes memory pages and 134217728 pages Redis will -# use a 4 GB swap file, that will use 16 MB of RAM for the page table. -# -# It's better to use the smallest acceptable value for your application, -# but the default is large in order to work in most conditions. -vm-pages 134217728 - -# Max number of VM I/O threads running at the same time. -# This threads are used to read/write data from/to swap file, since they -# also encode and decode objects from disk to memory or the reverse, a bigger -# number of threads can help with big objects even if they can't help with -# I/O itself as the physical device may not be able to couple with many -# reads/writes operations at the same time. -# -# The special value of 0 turn off threaded I/O and enables the blocking -# Virtual Memory implementation. -vm-max-threads 4 +#################################### DISK STORE ############################### + +# When disk store is active Redis works as an on-disk database, where memory +# is only used as a object cache. +# +# This mode is good for datasets that are bigger than memory, and in general +# when you want to trade speed for: +# +# - less memory used +# - immediate server restart +# - per key durability, without need for backgrond savig +# +# On the other hand, with disk store enabled MULTI/EXEC are no longer +# transactional from the point of view of the persistence on disk, that is, +# Redis transactions will still guarantee that commands are either processed +# all or nothing, but there is no guarantee that all the keys are flushed +# on disk in an atomic way. +# +# Of course with disk store enabled Redis is not as fast as it is when +# working with just the memory back end. + +diskstore-enabled no +diskstore-path redis.ds +cache-max-memory 0 +cache-flush-delay 0 ############################### ADVANCED CONFIG ############################### diff --git a/src/Makefile b/src/Makefile index 17fc435b..4d0f86db 100644 --- a/src/Makefile +++ b/src/Makefile @@ -25,7 +25,7 @@ PREFIX= /usr/local INSTALL_BIN= $(PREFIX)/bin INSTALL= cp -p -OBJ = adlist.o ae.o anet.o dict.o redis.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o vm.o pubsub.o multi.o debug.o sort.o intset.o syncio.o +OBJ = adlist.o ae.o anet.o dict.o redis.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o dscache.o pubsub.o multi.o debug.o sort.o intset.o syncio.o diskstore.o BENCHOBJ = ae.o anet.o redis-benchmark.o sds.o adlist.o zmalloc.o CLIOBJ = anet.o sds.o adlist.o redis-cli.o zmalloc.o release.o CHECKDUMPOBJ = redis-check-dump.o lzf_c.o lzf_d.o @@ -51,7 +51,6 @@ ae_select.o: ae_select.c anet.o: anet.c fmacros.h anet.h aof.o: aof.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \ zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h -chprgname.o: chprgname.c config.o: config.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \ zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h db.o: db.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \ @@ -59,6 +58,10 @@ db.o: db.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \ debug.o: debug.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \ zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h sha1.h dict.o: dict.c fmacros.h dict.h zmalloc.h +diskstore.o: diskstore.c redis.h fmacros.h config.h ae.h sds.h dict.h \ + adlist.h zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h +dscache.o: dscache.c redis.h fmacros.h config.h ae.h sds.h dict.h \ + adlist.h zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h intset.o: intset.c intset.h zmalloc.h lzf_c.o: lzf_c.c lzfP.h lzf_d.o: lzf_d.c lzfP.h @@ -73,11 +76,12 @@ pubsub.o: pubsub.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \ zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h rdb.o: rdb.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \ zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h lzf.h -redis-benchmark.o: redis-benchmark.c fmacros.h ae.h anet.h sds.h adlist.h \ - zmalloc.h +redis-benchmark.o: redis-benchmark.c fmacros.h ae.h \ + ../deps/hiredis/hiredis.h sds.h adlist.h zmalloc.h redis-check-aof.o: redis-check-aof.c fmacros.h config.h redis-check-dump.o: redis-check-dump.c lzf.h -redis-cli.o: redis-cli.c fmacros.h version.h sds.h adlist.h zmalloc.h +redis-cli.o: redis-cli.c fmacros.h version.h ../deps/hiredis/hiredis.h \ + sds.h zmalloc.h ../deps/linenoise/linenoise.h help.h redis.o: redis.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \ zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h release.o: release.c release.h @@ -87,7 +91,8 @@ sds.o: sds.c sds.h zmalloc.h sha1.o: sha1.c sha1.h sort.o: sort.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \ zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h pqsort.h -syncio.o: syncio.c +syncio.o: syncio.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \ + zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h t_hash.o: t_hash.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \ zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h t_list.o: t_list.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \ @@ -100,8 +105,6 @@ t_zset.o: t_zset.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \ zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h util.o: util.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \ zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h -vm.o: vm.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \ - zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h ziplist.o: ziplist.c zmalloc.h ziplist.h zipmap.o: zipmap.c zmalloc.h zmalloc.o: zmalloc.c config.h @@ -139,7 +142,7 @@ clean: rm -rf $(PRGNAME) $(BENCHPRGNAME) $(CLIPRGNAME) $(CHECKDUMPPRGNAME) $(CHECKAOFPRGNAME) *.o *.gcda *.gcno *.gcov dep: - $(CC) -MM *.c + $(CC) -MM *.c -I ../deps/hiredis -I ../deps/linenoise test: redis-server (cd ..; tclsh8.5 tests/test_helper.tcl --tags "${TAGS}" --file "${FILE}") diff --git a/src/aof.c b/src/aof.c index 959a5f52..f5d04a62 100644 --- a/src/aof.c +++ b/src/aof.c @@ -242,7 +242,6 @@ int loadAppendOnlyFile(char *filename) { char buf[128]; sds argsds; struct redisCommand *cmd; - int force_swapout; /* Serve the clients from time to time */ if (!(loops++ % 1000)) { @@ -286,17 +285,6 @@ int loadAppendOnlyFile(char *filename) { /* Clean up, ready for the next command */ for (j = 0; j < argc; j++) decrRefCount(argv[j]); zfree(argv); - - /* Handle swapping while loading big datasets when VM is on */ - force_swapout = 0; - if ((zmalloc_used_memory() - server.vm_max_memory) > 1024*1024*32) - force_swapout = 1; - - if (server.vm_enabled && force_swapout) { - while (zmalloc_used_memory() > server.vm_max_memory) { - if (vmSwapOneObjectBlocking() == REDIS_ERR) break; - } - } } /* This point can only be reached when EOF is reached without errors. @@ -359,22 +347,11 @@ int rewriteAppendOnlyFile(char *filename) { sds keystr = dictGetEntryKey(de); robj key, *o; time_t expiretime; - int swapped; keystr = dictGetEntryKey(de); o = dictGetEntryVal(de); initStaticStringObject(key,keystr); - /* If the value for this key is swapped, load a preview in memory. - * We use a "swapped" flag to remember if we need to free the - * value object instead to just increment the ref count anyway - * in order to avoid copy-on-write of pages if we are forked() */ - if (!server.vm_enabled || o->storage == REDIS_VM_MEMORY || - o->storage == REDIS_VM_SWAPPING) { - swapped = 0; - } else { - o = vmPreviewObject(o); - swapped = 1; - } + expiretime = getExpire(db,&key); /* Save the key and associated value */ @@ -509,7 +486,6 @@ int rewriteAppendOnlyFile(char *filename) { if (fwriteBulkObject(fp,&key) == 0) goto werr; if (fwriteBulkLongLong(fp,expiretime) == 0) goto werr; } - if (swapped) decrRefCount(o); } dictReleaseIterator(di); } @@ -553,12 +529,11 @@ int rewriteAppendOnlyFileBackground(void) { pid_t childpid; if (server.bgrewritechildpid != -1) return REDIS_ERR; - if (server.vm_enabled) waitEmptyIOJobsQueue(); + redisAssert(server.ds_enabled == 0); if ((childpid = fork()) == 0) { /* Child */ char tmpfile[256]; - if (server.vm_enabled) vmReopenSwapFile(); if (server.ipfd > 0) close(server.ipfd); if (server.sofd > 0) close(server.sofd); snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid()); diff --git a/src/config.c b/src/config.c index 79c367bf..3ba87c73 100644 --- a/src/config.c +++ b/src/config.c @@ -241,21 +241,18 @@ void loadServerConfig(char *filename) { } else if (!strcasecmp(argv[0],"dbfilename") && argc == 2) { zfree(server.dbfilename); server.dbfilename = zstrdup(argv[1]); - } else if (!strcasecmp(argv[0],"vm-enabled") && argc == 2) { - if ((server.vm_enabled = yesnotoi(argv[1])) == -1) { + } else if (!strcasecmp(argv[0],"diskstore-enabled") && argc == 2) { + if ((server.ds_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 = memtoll(argv[1],NULL); - } else if (!strcasecmp(argv[0],"vm-page-size") && argc == 2) { - server.vm_page_size = memtoll(argv[1], NULL); - } else if (!strcasecmp(argv[0],"vm-pages") && argc == 2) { - server.vm_pages = memtoll(argv[1], NULL); - } else if (!strcasecmp(argv[0],"vm-max-threads") && argc == 2) { - server.vm_max_threads = strtoll(argv[1], NULL, 10); + } else if (!strcasecmp(argv[0],"diskstore-path") && argc == 2) { + sdsfree(server.ds_path); + server.ds_path = sdsnew(argv[1]); + } else if (!strcasecmp(argv[0],"cache-max-memory") && argc == 2) { + server.cache_max_memory = memtoll(argv[1],NULL); + } else if (!strcasecmp(argv[0],"cache-flush-delay") && argc == 2) { + server.cache_flush_delay = atoi(argv[1]); + if (server.cache_flush_delay < 0) server.cache_flush_delay = 0; } else if (!strcasecmp(argv[0],"hash-max-zipmap-entries") && argc == 2) { server.hash_max_zipmap_entries = memtoll(argv[1], NULL); } else if (!strcasecmp(argv[0],"hash-max-zipmap-value") && argc == 2) { diff --git a/src/db.c b/src/db.c index aa1c14ad..1242c889 100644 --- a/src/db.c +++ b/src/db.c @@ -6,6 +6,35 @@ * C-level DB API *----------------------------------------------------------------------------*/ +/* Important notes on lookup and disk store. + * + * When disk store is enabled on lookup we can have different cases. + * + * a) The key is in memory: + * - If the key is not in IO_SAVEINPROG state we can access it. + * As if it's just IO_SAVE this means we have the key in the IO queue + * but can't be accessed by the IO thread (it requires to be + * translated into an IO Job by the cache cron function.) + * - If the key is in IO_SAVEINPROG we can't touch the key and have + * to blocking wait completion of operations. + * b) The key is not in memory: + * - If it's marked as non existing on disk as well (negative cache) + * we don't need to perform the disk access. + * - if the key MAY EXIST, but is not in memory, and it is marked as IO_SAVE + * then the key can only be a deleted one. As IO_SAVE keys are never + * evicted (dirty state), so the only possibility is that key was deleted. + * - if the key MAY EXIST we need to blocking load it. + * We check that the key is not in IO_SAVEINPROG state before accessing + * the disk object. If it is in this state, we wait. + */ + +void lookupWaitBusyKey(redisDb *db, robj *key) { + /* FIXME: wait just for this key, not everything */ + waitEmptyIOJobsQueue(); + processAllPendingIOJobs(); + redisAssert((cacheScheduleIOGetFlags(db,key) & REDIS_IO_SAVEINPROG) == 0); +} + robj *lookupKey(redisDb *db, robj *key) { dictEntry *de = dictFind(db->dict,key->ptr); if (de) { @@ -17,29 +46,52 @@ robj *lookupKey(redisDb *db, robj *key) { if (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1) val->lru = server.lruclock; - 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); - } 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); - } + if (server.ds_enabled && + cacheScheduleIOGetFlags(db,key) & REDIS_IO_SAVEINPROG) + { + /* Need to wait for the key to get unbusy */ + redisLog(REDIS_DEBUG,"Lookup found a key in SAVEINPROG state. Waiting. (Key was in the cache)"); + lookupWaitBusyKey(db,key); } server.stat_keyspace_hits++; return val; } else { + time_t expire; + robj *val; + + /* Key not found in the in memory hash table, but if disk store is + * enabled we may have this key on disk. If so load it in memory + * in a blocking way. */ + if (server.ds_enabled && cacheKeyMayExist(db,key)) { + long flags = cacheScheduleIOGetFlags(db,key); + + /* They key is not in cache, but it has a SAVE op in queue? + * The only possibility is that the key was deleted, since + * dirty keys are not evicted. */ + if (flags & REDIS_IO_SAVE) { + server.stat_keyspace_misses++; + return NULL; + } + + /* At this point we need to blocking load the key in memory. + * The first thing we do is waiting here if the key is busy. */ + if (flags & REDIS_IO_SAVEINPROG) { + redisLog(REDIS_DEBUG,"Lookup found a key in SAVEINPROG state. Waiting (while force loading)."); + lookupWaitBusyKey(db,key); + } + + redisLog(REDIS_DEBUG,"Force loading key %s via lookup", key->ptr); + val = dsGet(db,key,&expire); + if (val) { + int retval = dbAdd(db,key,val); + redisAssert(retval == REDIS_OK); + if (expire != -1) setExpire(db,key,expire); + server.stat_keyspace_hits++; + return val; + } else { + cacheSetKeyDoesNotExist(db,key); + } + } server.stat_keyspace_misses++; return NULL; } @@ -78,6 +130,7 @@ int dbAdd(redisDb *db, robj *key, robj *val) { } else { sds copy = sdsdup(key->ptr); dictAdd(db->dict, copy, val); + if (server.ds_enabled) cacheSetKeyMayExist(db,key); return REDIS_OK; } } @@ -87,14 +140,19 @@ int dbAdd(redisDb *db, robj *key, robj *val) { * * On update (key already existed) 0 is returned. Otherwise 1. */ int dbReplace(redisDb *db, robj *key, robj *val) { - if (dictFind(db->dict,key->ptr) == NULL) { + robj *oldval; + int retval; + + if ((oldval = dictFetchValue(db->dict,key->ptr)) == NULL) { sds copy = sdsdup(key->ptr); dictAdd(db->dict, copy, val); - return 1; + retval = 1; } else { dictReplace(db->dict, key->ptr, val); - return 0; + retval = 0; } + if (server.ds_enabled) cacheSetKeyMayExist(db,key); + return retval; } int dbExists(redisDb *db, robj *key) { @@ -129,11 +187,14 @@ robj *dbRandomKey(redisDb *db) { /* Delete a key, value, and associated expiration entry if any, from the DB */ int dbDelete(redisDb *db, robj *key) { - /* If VM is enabled make sure to awake waiting clients for this key: - * deleting the key will kill the I/O thread bringing the key from swap - * to memory, so the client will never be notified and unblocked if we - * don't do it now. */ - if (server.vm_enabled) handleClientsBlockedOnSwappedKey(db,key); + /* If diskstore is enabled make sure to awake waiting clients for this key + * as it is not really useful to wait for a key already deleted to be + * loaded from disk. */ + if (server.ds_enabled) { + handleClientsBlockedOnSwappedKey(db,key); + cacheSetKeyDoesNotExist(db,key); + } + /* Deleting an entry from the expires dict will not free the sds of * the key, because it is shared with the main dictionary. */ if (dictSize(db->expires) > 0) dictDelete(db->expires,key->ptr); @@ -160,27 +221,50 @@ int selectDb(redisClient *c, int id) { return REDIS_OK; } +/*----------------------------------------------------------------------------- + * Hooks for key space changes. + * + * Every time a key in the database is modified the function + * signalModifiedKey() is called. + * + * Every time a DB is flushed the function signalFlushDb() is called. + *----------------------------------------------------------------------------*/ + +void signalModifiedKey(redisDb *db, robj *key) { + touchWatchedKey(db,key); + if (server.ds_enabled) + cacheScheduleIO(db,key,REDIS_IO_SAVE); +} + +void signalFlushedDb(int dbid) { + touchWatchedKeysOnFlush(dbid); +} + /*----------------------------------------------------------------------------- * Type agnostic commands operating on the key space *----------------------------------------------------------------------------*/ void flushdbCommand(redisClient *c) { server.dirty += dictSize(c->db->dict); - touchWatchedKeysOnFlush(c->db->id); + signalFlushedDb(c->db->id); dictEmpty(c->db->dict); dictEmpty(c->db->expires); + if (server.ds_enabled) dsFlushDb(c->db->id); addReply(c,shared.ok); } void flushallCommand(redisClient *c) { - touchWatchedKeysOnFlush(-1); + signalFlushedDb(-1); server.dirty += emptyDb(); addReply(c,shared.ok); if (server.bgsavechildpid != -1) { kill(server.bgsavechildpid,SIGKILL); rdbRemoveTempFile(server.bgsavechildpid); } - rdbSave(server.dbfilename); + if (server.ds_enabled) + dsFlushDb(-1); + else + rdbSave(server.dbfilename); server.dirty++; } @@ -188,10 +272,22 @@ void delCommand(redisClient *c) { int deleted = 0, j; for (j = 1; j < c->argc; j++) { + if (server.ds_enabled) { + lookupKeyRead(c->db,c->argv[j]); + /* FIXME: this can be optimized a lot, no real need to load + * a possibly huge value. */ + } if (dbDelete(c->db,c->argv[j])) { - touchWatchedKey(c->db,c->argv[j]); + signalModifiedKey(c->db,c->argv[j]); server.dirty++; deleted++; + } else if (server.ds_enabled) { + if (cacheKeyMayExist(c->db,c->argv[j]) && + dsExists(c->db,c->argv[j])) + { + cacheScheduleIO(c->db,c->argv[j],REDIS_IO_SAVE); + deleted = 1; + } } } addReplyLongLong(c,deleted); @@ -335,8 +431,8 @@ void renameGenericCommand(redisClient *c, int nx) { dbReplace(c->db,c->argv[2],o); } dbDelete(c->db,c->argv[1]); - touchWatchedKey(c->db,c->argv[1]); - touchWatchedKey(c->db,c->argv[2]); + signalModifiedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[2]); server.dirty++; addReply(c,nx ? shared.cone : shared.ok); } @@ -495,13 +591,13 @@ void expireGenericCommand(redisClient *c, robj *key, robj *param, long offset) { if (seconds <= 0) { if (dbDelete(c->db,key)) server.dirty++; addReply(c, shared.cone); - touchWatchedKey(c->db,key); + signalModifiedKey(c->db,key); return; } else { time_t when = time(NULL)+seconds; setExpire(c->db,key,when); addReply(c,shared.cone); - touchWatchedKey(c->db,key); + signalModifiedKey(c->db,key); server.dirty++; return; } diff --git a/src/debug.c b/src/debug.c index fff8d727..47d4c85d 100644 --- a/src/debug.c +++ b/src/debug.c @@ -200,60 +200,22 @@ void debugCommand(redisClient *c) { } else if (!strcasecmp(c->argv[1]->ptr,"object") && c->argc == 3) { dictEntry *de = dictFind(c->db->dict,c->argv[2]->ptr); robj *val; + char *strenc; if (!de) { addReply(c,shared.nokeyerr); return; } val = dictGetEntryVal(de); - if (!server.vm_enabled || (val->storage == REDIS_VM_MEMORY || - val->storage == REDIS_VM_SWAPPING)) { - char *strenc; - - strenc = strEncoding(val->encoding); - addReplyStatusFormat(c, - "Value at:%p refcount:%d " - "encoding:%s serializedlength:%lld " - "lru:%d lru_seconds_idle:%lu", - (void*)val, val->refcount, - strenc, (long long) rdbSavedObjectLen(val), - val->lru, estimateObjectIdleTime(val)); - } else { - vmpointer *vp = (vmpointer*) val; - addReplyStatusFormat(c, - "Value swapped at: page %llu " - "using %llu pages", - (unsigned long long) vp->page, - (unsigned long long) vp->usedpages); - } - } else if (!strcasecmp(c->argv[1]->ptr,"swapin") && c->argc == 3) { - lookupKeyRead(c->db,c->argv[2]); - addReply(c,shared.ok); - } else if (!strcasecmp(c->argv[1]->ptr,"swapout") && c->argc == 3) { - dictEntry *de = dictFind(c->db->dict,c->argv[2]->ptr); - robj *val; - vmpointer *vp; - - if (!server.vm_enabled) { - addReplyError(c,"Virtual Memory is disabled"); - return; - } - if (!de) { - addReply(c,shared.nokeyerr); - return; - } - val = dictGetEntryVal(de); - /* Swap it */ - if (val->storage != REDIS_VM_MEMORY) { - addReplyError(c,"This key is not in memory"); - } else if (val->refcount != 1) { - addReplyError(c,"Object is shared"); - } else if ((vp = vmSwapObjectBlocking(val)) != NULL) { - dictGetEntryVal(de) = vp; - addReply(c,shared.ok); - } else { - addReply(c,shared.err); - } + strenc = strEncoding(val->encoding); + + addReplyStatusFormat(c, + "Value at:%p refcount:%d " + "encoding:%s serializedlength:%lld " + "lru:%d lru_seconds_idle:%lu", + (void*)val, val->refcount, + strenc, (long long) rdbSavedObjectLen(val), + val->lru, estimateObjectIdleTime(val)); } else if (!strcasecmp(c->argv[1]->ptr,"populate") && c->argc == 3) { long keys, j; robj *key, *val; @@ -300,10 +262,12 @@ void _redisAssert(char *estr, char *file, int line) { } void _redisPanic(char *msg, char *file, int line) { + redisLog(REDIS_WARNING,"------------------------------------------------"); redisLog(REDIS_WARNING,"!!! Software Failure. Press left mouse button to continue"); redisLog(REDIS_WARNING,"Guru Meditation: %s #%s:%d",msg,file,line); #ifdef HAVE_BACKTRACE redisLog(REDIS_WARNING,"(forcing SIGSEGV in order to print the stack trace)"); + redisLog(REDIS_WARNING,"------------------------------------------------"); *((char*)-1) = 'x'; #endif } diff --git a/src/diskstore.c b/src/diskstore.c new file mode 100644 index 00000000..7250390e --- /dev/null +++ b/src/diskstore.c @@ -0,0 +1,350 @@ +/* diskstore.c implements a very simple disk backed key-value store used + * by Redis for the "disk" backend. This implementation uses the filesystem + * to store key/value pairs. Every file represents a given key. + * + * The key path is calculated using the SHA1 of the key name. For instance + * the key "foo" is stored as a file name called: + * + * /0b/ee/0beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a33 + * + * The couples of characters from the hex output of SHA1 are also used + * to locate two two levels of directories to store the file (as most + * filesystems are not able to handle too many files in a single dir). + * + * In the end there are 65536 final directories (256 directories inside + * every 256 top level directories), so that with 1 billion of files every + * directory will contain in the average 15258 entires, that is ok with + * most filesystems implementation. + * + * Note that since Redis supports multiple databases, the actual key name + * is: + * + * /0b/ee/_0beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a33 + * + * so for instance if the key is inside DB 0: + * + * /0b/ee/0_0beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a33 + * + * The actaul implementation of this disk store is highly dependant to the + * filesystem implementation itself. This implementation may be replaced by + * a B+TREE implementation in future implementations. + * + * Data ok every key is serialized using the same format used for .rdb + * serialization. Everything is serialized on every entry: key name, + * ttl information in case of keys with an associated expire time, and the + * serialized value itself. + * + * Because the format is the same of the .rdb files it is trivial to create + * an .rdb file starting from this format just by mean of scanning the + * directories and concatenating entries, with the sole addition of an + * .rdb header at the start and the end-of-db opcode at the end. + * + * ------------------------------------------------------------------------- + * + * Copyright (c) 2010-2011, Salvatore Sanfilippo + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * * Redistributions of source code must retain the above copyright notice, + * this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * * Neither the name of Redis nor the names of its contributors may be used + * to endorse or promote products derived from this software without + * specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" + * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE + * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR + * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF + * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS + * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN + * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) + * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE + * POSSIBILITY OF SUCH DAMAGE. + */ + +#include "redis.h" +#include "sha1.h" + +#include +#include +#include + +int create256dir(char *prefix) { + char buf[1024]; + int j; + + for (j = 0; j < 256; j++) { + snprintf(buf,sizeof(buf),"%s%02x",prefix,j); + if (mkdir(buf,0755) == -1) { + redisLog(REDIS_WARNING,"Error creating dir %s for diskstore: %s", + buf,strerror(errno)); + return REDIS_ERR; + } + } + return REDIS_OK; +} + +int dsOpen(void) { + struct stat sb; + int retval, j; + char *path = server.ds_path; + char buf[1024]; + + if ((retval = stat(path,&sb) == -1) && errno != ENOENT) { + redisLog(REDIS_WARNING, "Error opening disk store at %s: %s", + path, strerror(errno)); + return REDIS_ERR; + } + + /* Directory already in place. Assume everything is ok. */ + if (retval == 0 && S_ISDIR(sb.st_mode)) { + redisLog(REDIS_NOTICE,"Disk store %s exists", path); + return REDIS_OK; + } + + /* File exists but it's not a directory */ + if (retval == 0 && !S_ISDIR(sb.st_mode)) { + redisLog(REDIS_WARNING,"Disk store at %s is not a directory", path); + return REDIS_ERR; + } + + /* New disk store, create the directory structure now, as creating + * them in a lazy way is not a good idea, after very few insertions + * we'll need most of the 65536 directories anyway. */ + redisLog(REDIS_NOTICE,"Disk store %s does not exist: creating", path); + if (mkdir(path,0755) == -1) { + redisLog(REDIS_WARNING,"Disk store init failed creating dir %s: %s", + path, strerror(errno)); + return REDIS_ERR; + } + /* Create the top level 256 directories */ + snprintf(buf,sizeof(buf),"%s/",path); + if (create256dir(buf) == REDIS_ERR) return REDIS_ERR; + + /* For every 256 top level dir, create 256 nested dirs */ + for (j = 0; j < 256; j++) { + snprintf(buf,sizeof(buf),"%s/%02x/",path,j); + if (create256dir(buf) == REDIS_ERR) return REDIS_ERR; + } + return REDIS_OK; +} + +int dsClose(void) { + return REDIS_OK; +} + +/* Convert key into full path for this object. Dirty but hopefully + * is fast enough. Returns the length of the returned path. */ +int dsKeyToPath(redisDb *db, char *buf, robj *key) { + SHA1_CTX ctx; + unsigned char hash[20]; + char hex[40], digits[] = "0123456789abcdef"; + int j, l; + char *origbuf = buf; + + SHA1Init(&ctx); + SHA1Update(&ctx,key->ptr,sdslen(key->ptr)); + SHA1Final(hash,&ctx); + + /* Convert the hash into hex format */ + for (j = 0; j < 20; j++) { + hex[j*2] = digits[(hash[j]&0xF0)>>4]; + hex[(j*2)+1] = digits[hash[j]&0x0F]; + } + + /* Create the object path. Start with server.ds_path that's the root dir */ + l = sdslen(server.ds_path); + memcpy(buf,server.ds_path,l); + buf += l; + *buf++ = '/'; + + /* Then add xx/yy/ that is the two level directories */ + buf[0] = hex[0]; + buf[1] = hex[1]; + buf[2] = '/'; + buf[3] = hex[2]; + buf[4] = hex[3]; + buf[5] = '/'; + buf += 6; + + /* Add the database number followed by _ and finall the SHA1 hex */ + l = ll2string(buf,64,db->id); + buf += l; + buf[0] = '_'; + memcpy(buf+1,hex,40); + buf[41] = '\0'; + return (buf-origbuf)+41; +} + +int dsSet(redisDb *db, robj *key, robj *val) { + char buf[1024], buf2[1024]; + FILE *fp; + int retval, len; + + len = dsKeyToPath(db,buf,key); + memcpy(buf2,buf,len); + snprintf(buf2+len,sizeof(buf2)-len,"_%ld_%ld",(long)time(NULL),(long)val); + while ((fp = fopen(buf2,"w")) == NULL) { + if (errno == ENOSPC) { + redisLog(REDIS_WARNING,"Diskstore: No space left on device. Please make room and wait 30 seconds for Redis to continue."); + sleep(30); + } else { + redisLog(REDIS_WARNING,"diskstore error opening %s: %s", + buf2, strerror(errno)); + redisPanic("Unrecoverable diskstore error. Exiting."); + } + } + if ((retval = rdbSaveKeyValuePair(fp,db,key,val,time(NULL))) == -1) + return REDIS_ERR; + fclose(fp); + if (retval == 0) { + /* Expired key. Unlink failing not critical */ + unlink(buf); + unlink(buf2); + } else { + /* Use rename for atomic updadte of value */ + if (rename(buf2,buf) == -1) { + redisLog(REDIS_WARNING,"rename(2) returned an error: %s", + strerror(errno)); + redisPanic("Unrecoverable diskstore error. Exiting."); + } + } + return REDIS_OK; +} + +robj *dsGet(redisDb *db, robj *key, time_t *expire) { + char buf[1024]; + int type; + time_t expiretime = -1; /* -1 means: no expire */ + robj *dskey; /* Key as loaded from disk. */ + robj *val; + FILE *fp; + + dsKeyToPath(db,buf,key); + fp = fopen(buf,"r"); + if (fp == NULL && errno == ENOENT) return NULL; /* No such key */ + if (fp == NULL) { + redisLog(REDIS_WARNING,"Disk store failed opening %s: %s", + buf, strerror(errno)); + goto readerr; + } + + if ((type = rdbLoadType(fp)) == -1) goto readerr; + if (type == REDIS_EXPIRETIME) { + if ((expiretime = rdbLoadTime(fp)) == -1) goto readerr; + /* We read the time so we need to read the object type again */ + if ((type = rdbLoadType(fp)) == -1) goto readerr; + } + /* Read key */ + if ((dskey = rdbLoadStringObject(fp)) == NULL) goto readerr; + /* Read value */ + if ((val = rdbLoadObject(type,fp)) == NULL) goto readerr; + fclose(fp); + + /* The key we asked, and the key returned, must be the same */ + redisAssert(equalStringObjects(key,dskey)); + + /* Check if the key already expired */ + decrRefCount(dskey); + if (expiretime != -1 && expiretime < time(NULL)) { + decrRefCount(val); + unlink(buf); /* This failing is non critical here */ + return NULL; + } + + /* Everything ok... */ + *expire = expiretime; + return val; + +readerr: + redisLog(REDIS_WARNING,"Read error reading reading %s. Corrupted key?", + buf); + redisPanic("Unrecoverable error reading from disk store"); + return NULL; /* unreached */ +} + +int dsDel(redisDb *db, robj *key) { + char buf[1024]; + + dsKeyToPath(db,buf,key); + if (unlink(buf) == -1) { + if (errno == ENOENT) { + return REDIS_ERR; + } else { + redisLog(REDIS_WARNING,"Disk store can't remove %s: %s", + buf, strerror(errno)); + redisPanic("Unrecoverable Disk store errore. Existing."); + return REDIS_ERR; /* unreached */ + } + } else { + return REDIS_OK; + } +} + +int dsExists(redisDb *db, robj *key) { + char buf[1024]; + + dsKeyToPath(db,buf,key); + return access(buf,R_OK) == 0; +} + +void dsFlushOneDir(char *path, int dbid) { + DIR *dir; + struct dirent *dp, de; + + dir = opendir(path); + if (dir == NULL) { + redisLog(REDIS_WARNING,"Disk store can't open dir %s: %s", + path, strerror(errno)); + redisPanic("Unrecoverable Disk store errore. Existing."); + } + while(1) { + char buf[1024]; + + readdir_r(dir,&de,&dp); + if (dp == NULL) break; + if (dp->d_name[0] == '.') continue; + + /* Check if we need to remove this entry accordingly to the + * DB number */ + if (dbid != -1) { + char id[64]; + char *p = strchr(dp->d_name,'_'); + int len = (p - dp->d_name); + + redisAssert(p != NULL && len < 64); + memcpy(id,dp->d_name,len); + id[len] = '\0'; + if (atoi(id) != dbid) continue; /* skip this file */ + } + + /* Finally unlink the file */ + snprintf(buf,1024,"%s/%s",path,dp->d_name); + if (unlink(buf) == -1) { + redisLog(REDIS_WARNING, + "Can't unlink %s: %s", buf, strerror(errno)); + redisPanic("Unrecoverable Disk store errore. Existing."); + } + } + closedir(dir); +} + +void dsFlushDb(int dbid) { + char buf[1024]; + int j, i; + + redisLog(REDIS_NOTICE,"Flushing diskstore DB (%d)",dbid); + for (j = 0; j < 256; j++) { + for (i = 0; i < 256; i++) { + snprintf(buf,1024,"%s/%02x/%02x",server.ds_path,j,i); + dsFlushOneDir(buf,dbid); + } + } +} diff --git a/src/dscache.c b/src/dscache.c new file mode 100644 index 00000000..901c70f0 --- /dev/null +++ b/src/dscache.c @@ -0,0 +1,979 @@ +#include "redis.h" + +#include +#include +#include +#include + +/* dscache.c - Disk store cache for disk store backend. + * + * When Redis is configured for using disk as backend instead of memory, the + * memory is used as a cache, so that recently accessed keys are taken in + * memory for fast read and write operations. + * + * Modified keys are marked to be flushed on disk, and will be flushed + * as long as the maxium configured flush time elapsed. + * + * This file implements the whole caching subsystem and contains further + * documentation. */ + +/* TODO: + * + * WARNING: most of the following todo items and design issues are no + * longer relevant with the new design. Here as a checklist to see if + * some old ideas still apply. + * + * - What happens when an object is destroyed? + * + * If the object is destroyed since semantically it was deleted or + * replaced with something new, we don't care if there was a SAVE + * job pending for it. Anyway when the IO JOb will be created we'll get + * the pointer of the current value. + * + * If the object is already a REDIS_IO_SAVEINPROG object, then it is + * impossible that we get a decrRefCount() that will reach refcount of zero + * since the object is both in the dataset and in the io job entry. + * + * - What happens with MULTI/EXEC? + * + * Good question. Without some kind of versioning with a global counter + * it is not possible to have trasactions on disk, but they are still + * useful since from the point of view of memory and client bugs it is + * a protection anyway. Also it's useful for WATCH. + * + * Btw there is to check what happens when WATCH gets combined to keys + * that gets removed from the object cache. Should be save but better + * to check. + * + * - Check if/why INCR will not update the LRU info for the object. + * + * - Fix/Check the following race condition: a key gets a DEL so there is + * a write operation scheduled against this key. Later the same key will + * be the argument of a GET, but the write operation was still not + * completed (to delete the file). If the GET will be for some reason + * a blocking loading (via lookup) we can load the old value on memory. + * + * This problems can be fixed with negative caching. We can use it + * to optimize the system, but also when a key is deleted we mark + * it as non existing on disk as well (in a way that this cache + * entry can't be evicted, setting time to 0), then we avoid looking at + * the disk at all if the key can't be there. When an IO Job complete + * a deletion, we set the time of the negative caching to a non zero + * value so it will be evicted later. + * + * Are there other patterns like this where we load stale data? + * + * Also, make sure that key preloading is ONLY done for keys that are + * not marked as cacheKeyDoesNotExist(), otherwise, again, we can load + * data from disk that should instead be deleted. + * + * - dsSet() should use rename(2) in order to avoid corruptions. + * + * - Don't add a LOAD if there is already a LOADINPROGRESS, or is this + * impossible since anyway the io_keys stuff will work as lock? + * + * - Serialize special encoded things in a raw form. + * + * - When putting IO read operations on top of the queue, do this only if + * the already-on-top operation is not a save or if it is a save that + * is scheduled for later execution. If there is a save that is ready to + * fire, let's insert the load operation just before the first save that + * is scheduled for later exection for instance. + * + * - Support MULTI/EXEC transactions via a journal file, that is played on + * startup to check if there is cleanup to do. This way we can implement + * transactions with our simple file based KV store. + */ + +/* Virtual Memory is composed mainly of two subsystems: + * - Blocking Virutal Memory + * - Threaded Virtual Memory I/O + * The two parts are not fully decoupled, but functions are split among two + * different sections of the source code (delimited by comments) in order to + * make more clear what functionality is about the blocking VM and what about + * the threaded (not blocking) VM. + * + * Redis VM design: + * + * Redis VM is a blocking VM (one that blocks reading swapped values from + * disk into memory when a value swapped out is needed in memory) that is made + * unblocking by trying to examine the command argument vector in order to + * load in background values that will likely be needed in order to exec + * the command. The command is executed only once all the relevant keys + * are loaded into memory. + * + * This basically is almost as simple of a blocking VM, but almost as parallel + * as a fully non-blocking VM. + */ + +void spawnIOThread(void); +int cacheScheduleIOPushJobs(int onlyloads); + +/* =================== Virtual Memory - Blocking Side ====================== */ + +void dsInit(void) { + int pipefds[2]; + size_t stacksize; + + zmalloc_enable_thread_safeness(); /* we need thread safe zmalloc() */ + + redisLog(REDIS_NOTICE,"Opening Disk Store: %s", server.ds_path); + /* Open Disk Store */ + if (dsOpen() != REDIS_OK) { + redisLog(REDIS_WARNING,"Fatal error opening disk store. Exiting."); + exit(1); + }; + + /* Initialize threaded I/O for Object Cache */ + server.io_newjobs = listCreate(); + server.io_processing = listCreate(); + server.io_processed = listCreate(); + server.io_ready_clients = listCreate(); + pthread_mutex_init(&server.io_mutex,NULL); + pthread_cond_init(&server.io_condvar,NULL); + server.io_active_threads = 0; + if (pipe(pipefds) == -1) { + redisLog(REDIS_WARNING,"Unable to intialized DS: pipe(2): %s. Exiting." + ,strerror(errno)); + exit(1); + } + 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); + + /* Solaris may report a stacksize of 0, let's set it to 1 otherwise + * multiplying it by 2 in the while loop later will not really help ;) */ + if (!stacksize) stacksize = 1; + + 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) + oom("creating file event"); + + /* Spawn our I/O thread */ + spawnIOThread(); +} + +/* Compute how good candidate the specified object is for eviction. + * An higher number means a better candidate. */ +double computeObjectSwappability(robj *o) { + /* actual age can be >= minage, but not < minage. As we use wrapping + * 21 bit clocks with minutes resolution for the LRU. */ + return (double) estimateObjectIdleTime(o); +} + +/* Try to free one entry from the diskstore object cache */ +int cacheFreeOneEntry(void) { + int j, i; + struct dictEntry *best = NULL; + double best_swappability = 0; + redisDb *best_db = NULL; + robj *val; + sds key; + + for (j = 0; j < server.dbnum; j++) { + redisDb *db = server.db+j; + /* 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++) { + dictEntry *de; + double swappability; + robj keyobj; + sds keystr; + + if (maxtries) maxtries--; + de = dictGetRandomKey(db->dict); + keystr = dictGetEntryKey(de); + val = dictGetEntryVal(de); + initStaticStringObject(keyobj,keystr); + + /* Don't remove objects that are currently target of a + * read or write operation. */ + if (cacheScheduleIOGetFlags(db,&keyobj) != 0) { + if (maxtries) i--; /* don't count this try */ + continue; + } + swappability = computeObjectSwappability(val); + if (!best || swappability > best_swappability) { + best = de; + best_swappability = swappability; + best_db = db; + } + } + } + if (best == NULL) { + /* Was not able to fix a single object... we should check if our + * IO queues have stuff in queue, and try to consume the queue + * otherwise we'll use an infinite amount of memory if changes to + * the dataset are faster than I/O */ + if (listLength(server.cache_io_queue) > 0) { + cacheScheduleIOPushJobs(0); + waitEmptyIOJobsQueue(); + processAllPendingIOJobs(); + return REDIS_OK; + } + /* Nothing to free at all... */ + return REDIS_ERR; + } + key = dictGetEntryKey(best); + val = dictGetEntryVal(best); + + redisLog(REDIS_DEBUG,"Key selected for cache eviction: %s swappability:%f", + key, best_swappability); + + /* Delete this key from memory */ + { + robj *kobj = createStringObject(key,sdslen(key)); + dbDelete(best_db,kobj); + decrRefCount(kobj); + } + return REDIS_OK; +} + +/* Return true if it's safe to swap out objects in a given moment. + * Basically we don't want to swap objects out while there is a BGSAVE + * or a BGAEOREWRITE running in backgroud. */ +int dsCanTouchDiskStore(void) { + return (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1); +} + +/* ==================== Disk store negative caching ======================== + * + * When disk store is enabled, we need negative caching, that is, to remember + * keys that are for sure *not* on the disk key-value store. + * + * This is usefuls because without negative caching cache misses will cost us + * a disk lookup, even if the same non existing key is accessed again and again. + * + * With negative caching we remember that the key is not on disk, so if it's + * not in memory and we have a negative cache entry, we don't try a disk + * access at all. + */ + +/* Returns true if the specified key may exists on disk, that is, we don't + * have an entry in our negative cache for this key */ +int cacheKeyMayExist(redisDb *db, robj *key) { + return dictFind(db->io_negcache,key) == NULL; +} + +/* Set the specified key as an entry that may possibily exist on disk, that is, + * remove the negative cache entry for this key if any. */ +void cacheSetKeyMayExist(redisDb *db, robj *key) { + dictDelete(db->io_negcache,key); +} + +/* Set the specified key as non existing on disk, that is, create a negative + * cache entry for this key. */ +void cacheSetKeyDoesNotExist(redisDb *db, robj *key) { + if (dictReplace(db->io_negcache,key,(void*)time(NULL))) { + incrRefCount(key); + } +} + +/* Remove one entry from negative cache using approximated LRU. */ +int negativeCacheEvictOneEntry(void) { + struct dictEntry *de; + robj *best = NULL; + redisDb *best_db = NULL; + time_t time, best_time = 0; + int j; + + for (j = 0; j < server.dbnum; j++) { + redisDb *db = server.db+j; + int i; + + if (dictSize(db->io_negcache) == 0) continue; + for (i = 0; i < 3; i++) { + de = dictGetRandomKey(db->io_negcache); + time = (time_t) dictGetEntryVal(de); + + if (best == NULL || time < best_time) { + best = dictGetEntryKey(de); + best_db = db; + best_time = time; + } + } + } + if (best) { + dictDelete(best_db->io_negcache,best); + return REDIS_OK; + } else { + return REDIS_ERR; + } +} + +/* ================== Disk store cache - Threaded I/O ====================== */ + +void freeIOJob(iojob *j) { + decrRefCount(j->key); + /* j->val can be NULL if the job is about deleting the key from disk. */ + if (j->val) decrRefCount(j->val); + zfree(j); +} + +/* Every time a thread finished a Job, it writes a byte into the write side + * of an unix pipe in order to "awake" the main thread, and this function + * is called. */ +void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata, + int mask) +{ + char buf[1]; + int retval, processed = 0, toprocess = -1; + REDIS_NOTUSED(el); + REDIS_NOTUSED(mask); + REDIS_NOTUSED(privdata); + + /* For every byte we read in the read side of the pipe, there is one + * I/O job completed to process. */ + while((retval = read(fd,buf,1)) == 1) { + iojob *j; + listNode *ln; + + redisLog(REDIS_DEBUG,"Processing I/O completed job"); + + /* Get the processed element (the oldest one) */ + lockThreadedIO(); + redisAssert(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); + unlockThreadedIO(); + + /* Post process it in the main thread, as there are things we + * can do just here to avoid race conditions and/or invasive locks */ + redisLog(REDIS_DEBUG,"COMPLETED Job type %s, key: %s", + (j->type == REDIS_IOJOB_LOAD) ? "load" : "save", + (unsigned char*)j->key->ptr); + if (j->type == REDIS_IOJOB_LOAD) { + /* Create the key-value pair in the in-memory database */ + if (j->val != NULL) { + /* Note: it's possible that the key is already in memory + * due to a blocking load operation. */ + if (dbAdd(j->db,j->key,j->val) == REDIS_OK) { + incrRefCount(j->val); + if (j->expire != -1) setExpire(j->db,j->key,j->expire); + } + } else { + /* Key not found on disk. If it is also not in memory + * as a cached object, nor there is a job writing it + * in background, we are sure the key does not exist + * currently. + * + * So we set a negative cache entry avoiding that the + * resumed client will block load what does not exist... */ + if (dictFind(j->db->dict,j->key->ptr) == NULL && + (cacheScheduleIOGetFlags(j->db,j->key) & + (REDIS_IO_SAVE|REDIS_IO_SAVEINPROG)) == 0) + { + cacheSetKeyDoesNotExist(j->db,j->key); + } + } + cacheScheduleIODelFlag(j->db,j->key,REDIS_IO_LOADINPROG); + handleClientsBlockedOnSwappedKey(j->db,j->key); + freeIOJob(j); + } else if (j->type == REDIS_IOJOB_SAVE) { + cacheScheduleIODelFlag(j->db,j->key,REDIS_IO_SAVEINPROG); + freeIOJob(j); + } + processed++; + if (processed == toprocess) return; + } + if (retval < 0 && errno != EAGAIN) { + redisLog(REDIS_WARNING, + "WARNING: read(2) error in vmThreadedIOCompletedJob() %s", + strerror(errno)); + } +} + +void lockThreadedIO(void) { + pthread_mutex_lock(&server.io_mutex); +} + +void unlockThreadedIO(void) { + pthread_mutex_unlock(&server.io_mutex); +} + +void *IOThreadEntryPoint(void *arg) { + iojob *j; + listNode *ln; + REDIS_NOTUSED(arg); + + pthread_detach(pthread_self()); + lockThreadedIO(); + while(1) { + /* Get a new job to process */ + if (listLength(server.io_newjobs) == 0) { + /* Wait for more work to do */ + pthread_cond_wait(&server.io_condvar,&server.io_mutex); + continue; + } + redisLog(REDIS_DEBUG,"%ld IO jobs to process", + listLength(server.io_newjobs)); + ln = listFirst(server.io_newjobs); + j = ln->value; + listDelNode(server.io_newjobs,ln); + /* Add the job in the processing queue */ + listAddNodeTail(server.io_processing,j); + ln = listLast(server.io_processing); /* We use ln later to remove it */ + unlockThreadedIO(); + + redisLog(REDIS_DEBUG,"Thread %ld: new job type %s: %p about key '%s'", + (long) pthread_self(), + (j->type == REDIS_IOJOB_LOAD) ? "load" : "save", + (void*)j, (char*)j->key->ptr); + + /* Process the Job */ + if (j->type == REDIS_IOJOB_LOAD) { + time_t expire; + + j->val = dsGet(j->db,j->key,&expire); + if (j->val) j->expire = expire; + } else if (j->type == REDIS_IOJOB_SAVE) { + if (j->val) { + dsSet(j->db,j->key,j->val); + } else { + dsDel(j->db,j->key); + } + } + + /* Done: insert the job into the processed queue */ + redisLog(REDIS_DEBUG,"Thread %ld completed the job: %p (key %s)", + (long) pthread_self(), (void*)j, (char*)j->key->ptr); + + lockThreadedIO(); + listDelNode(server.io_processing,ln); + listAddNodeTail(server.io_processed,j); + + /* Signal the main thread there is new stuff to process */ + redisAssert(write(server.io_ready_pipe_write,"x",1) == 1); + } + /* never reached, but that's the full pattern... */ + unlockThreadedIO(); + return NULL; +} + +void spawnIOThread(void) { + pthread_t thread; + sigset_t mask, omask; + int err; + + sigemptyset(&mask); + sigaddset(&mask,SIGCHLD); + sigaddset(&mask,SIGHUP); + sigaddset(&mask,SIGPIPE); + pthread_sigmask(SIG_SETMASK, &mask, &omask); + while ((err = pthread_create(&thread,&server.io_threads_attr,IOThreadEntryPoint,NULL)) != 0) { + redisLog(REDIS_WARNING,"Unable to spawn an I/O thread: %s", + strerror(err)); + usleep(1000000); + } + pthread_sigmask(SIG_SETMASK, &omask, NULL); + server.io_active_threads++; +} + +/* Wait that all the pending IO Jobs are processed */ +void waitEmptyIOJobsQueue(void) { + while(1) { + int io_processed_len; + + lockThreadedIO(); + if (listLength(server.io_newjobs) == 0 && + listLength(server.io_processing) == 0) + { + unlockThreadedIO(); + return; + } + /* If there are new jobs we need to signal the thread to + * process the next one. */ + redisLog(REDIS_DEBUG,"waitEmptyIOJobsQueue: new %d, processing %d", + listLength(server.io_newjobs), + listLength(server.io_processing)); + + /* FIXME: signal or not? + if (listLength(server.io_newjobs)) { + pthread_cond_signal(&server.io_condvar); + } + */ + /* 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, + (void*)0xdeadbeef,0); + /* FIXME: probably wiser to drop this sleeps. Anyway + * the contention on the IO thread will avoid we to loop + * too fast here. */ + usleep(1000); /* 1 millisecond */ + } else { + /* FIXME: same as fixme above. */ + usleep(10000); /* 10 milliseconds */ + } + } +} + +/* Process all the IO Jobs already completed by threads but still waiting + * processing from the main thread. */ +void processAllPendingIOJobs(void) { + while(1) { + int io_processed_len; + + lockThreadedIO(); + io_processed_len = listLength(server.io_processed); + unlockThreadedIO(); + if (io_processed_len == 0) return; + vmThreadedIOCompletedJob(NULL,server.io_ready_pipe_read, + (void*)0xdeadbeef,0); + } +} + +/* This function must be called while with threaded IO locked */ +void queueIOJob(iojob *j) { + redisLog(REDIS_DEBUG,"Queued IO Job %p type %d about key '%s'\n", + (void*)j, j->type, (char*)j->key->ptr); + listAddNodeTail(server.io_newjobs,j); + if (server.io_active_threads < server.vm_max_threads) + spawnIOThread(); +} + +void dsCreateIOJob(int type, redisDb *db, robj *key, robj *val) { + iojob *j; + + j = zmalloc(sizeof(*j)); + j->type = type; + j->db = db; + j->key = key; + incrRefCount(key); + j->val = val; + if (val) incrRefCount(val); + + lockThreadedIO(); + queueIOJob(j); + pthread_cond_signal(&server.io_condvar); + unlockThreadedIO(); +} + +/* ============= Disk store cache - Scheduling of IO operations ============= + * + * We use a queue and an hash table to hold the state of IO operations + * so that's fast to lookup if there is already an IO operation in queue + * for a given key. + * + * There are two types of IO operations for a given key: + * REDIS_IO_LOAD and REDIS_IO_SAVE. + * + * The function cacheScheduleIO() function pushes the specified IO operation + * in the queue, but avoid adding the same key for the same operation + * multiple times, thanks to the associated hash table. + * + * We take a set of flags per every key, so when the scheduled IO operation + * gets moved from the scheduled queue to the actual IO Jobs queue that + * is processed by the IO thread, we flag it as IO_LOADINPROG or + * IO_SAVEINPROG. + * + * So for every given key we always know if there is some IO operation + * scheduled, or in progress, for this key. + * + * NOTE: all this is very important in order to guarantee correctness of + * the Disk Store Cache. Jobs are always queued here. Load jobs are + * queued at the head for faster execution only in the case there is not + * already a write operation of some kind for this job. + * + * So we have ordering, but can do exceptions when there are no already + * operations for a given key. Also when we need to block load a given + * key, for an immediate lookup operation, we can check if the key can + * be accessed synchronously without race conditions (no IN PROGRESS + * operations for this key), otherwise we blocking wait for completion. */ + +#define REDIS_IO_LOAD 1 +#define REDIS_IO_SAVE 2 +#define REDIS_IO_LOADINPROG 4 +#define REDIS_IO_SAVEINPROG 8 + +void cacheScheduleIOAddFlag(redisDb *db, robj *key, long flag) { + struct dictEntry *de = dictFind(db->io_queued,key); + + if (!de) { + dictAdd(db->io_queued,key,(void*)flag); + incrRefCount(key); + return; + } else { + long flags = (long) dictGetEntryVal(de); + + if (flags & flag) { + redisLog(REDIS_WARNING,"Adding the same flag again: was: %ld, addede: %ld",flags,flag); + redisAssert(!(flags & flag)); + } + flags |= flag; + dictGetEntryVal(de) = (void*) flags; + } +} + +void cacheScheduleIODelFlag(redisDb *db, robj *key, long flag) { + struct dictEntry *de = dictFind(db->io_queued,key); + long flags; + + redisAssert(de != NULL); + flags = (long) dictGetEntryVal(de); + redisAssert(flags & flag); + flags &= ~flag; + if (flags == 0) { + dictDelete(db->io_queued,key); + } else { + dictGetEntryVal(de) = (void*) flags; + } +} + +int cacheScheduleIOGetFlags(redisDb *db, robj *key) { + struct dictEntry *de = dictFind(db->io_queued,key); + + return (de == NULL) ? 0 : ((long) dictGetEntryVal(de)); +} + +void cacheScheduleIO(redisDb *db, robj *key, int type) { + ioop *op; + long flags; + + if ((flags = cacheScheduleIOGetFlags(db,key)) & type) return; + + redisLog(REDIS_DEBUG,"Scheduling key %s for %s", + key->ptr, type == REDIS_IO_LOAD ? "loading" : "saving"); + cacheScheduleIOAddFlag(db,key,type); + op = zmalloc(sizeof(*op)); + op->type = type; + op->db = db; + op->key = key; + incrRefCount(key); + op->ctime = time(NULL); + + /* Give priority to load operations if there are no save already + * in queue for the same key. */ + if (type == REDIS_IO_LOAD && !(flags & REDIS_IO_SAVE)) { + listAddNodeHead(server.cache_io_queue, op); + cacheScheduleIOPushJobs(1); + } else { + /* FIXME: probably when this happens we want to at least move + * the write job about this queue on top, and set the creation time + * to a value that will force processing ASAP. */ + listAddNodeTail(server.cache_io_queue, op); + } +} + +/* Push scheduled IO operations into IO Jobs that the IO thread can process. + * If 'onlyloads' is true only IO_LOAD jobs are processed: this is useful + * since it's save to push LOAD IO jobs from any place of the code, while + * SAVE io jobs should never be pushed while we are processing a command + * (not protected by lookupKey() that will block on keys in IO_SAVEINPROG + * state. */ +#define MAX_IO_JOBS_QUEUE 100 +int cacheScheduleIOPushJobs(int onlyloads) { + time_t now = time(NULL); + listNode *ln; + int jobs, topush = 0, pushed = 0; + + /* Sync stuff on disk, but only if we have less + * than MAX_IO_JOBS_QUEUE IO jobs. */ + lockThreadedIO(); + jobs = listLength(server.io_newjobs); + unlockThreadedIO(); + + topush = MAX_IO_JOBS_QUEUE-jobs; + if (topush < 0) topush = 0; + if (topush > (signed)listLength(server.cache_io_queue)) + topush = listLength(server.cache_io_queue); + + while((ln = listFirst(server.cache_io_queue)) != NULL) { + ioop *op = ln->value; + + if (!topush) break; + topush--; + + if (op->type == REDIS_IO_LOAD || + (!onlyloads && (now - op->ctime) >= server.cache_flush_delay)) + { + struct dictEntry *de; + robj *val; + + /* Don't add a SAVE job in queue if there is already + * a save in progress for the same key. */ + if (op->type == REDIS_IO_SAVE && + cacheScheduleIOGetFlags(op->db,op->key) & REDIS_IO_SAVEINPROG) + { + /* Move the operation at the end of the list of there + * are other operations. Otherwise break, nothing to do + * here. */ + if (listLength(server.cache_io_queue) > 1) { + listDelNode(server.cache_io_queue,ln); + listAddNodeTail(server.cache_io_queue,op); + continue; + } else { + break; + } + } + + redisLog(REDIS_DEBUG,"Creating IO %s Job for key %s", + op->type == REDIS_IO_LOAD ? "load" : "save", op->key->ptr); + + if (op->type == REDIS_IO_LOAD) { + dsCreateIOJob(REDIS_IOJOB_LOAD,op->db,op->key,NULL); + } else { + /* Lookup the key, in order to put the current value in the IO + * Job. Otherwise if the key does not exists we schedule a disk + * store delete operation, setting the value to NULL. */ + de = dictFind(op->db->dict,op->key->ptr); + if (de) { + val = dictGetEntryVal(de); + } else { + /* Setting the value to NULL tells the IO thread to delete + * the key on disk. */ + val = NULL; + } + dsCreateIOJob(REDIS_IOJOB_SAVE,op->db,op->key,val); + } + /* Mark the operation as in progress. */ + cacheScheduleIODelFlag(op->db,op->key,op->type); + cacheScheduleIOAddFlag(op->db,op->key, + (op->type == REDIS_IO_LOAD) ? REDIS_IO_LOADINPROG : + REDIS_IO_SAVEINPROG); + /* Finally remove the operation from the queue. + * But we'll have trace of it in the hash table. */ + listDelNode(server.cache_io_queue,ln); + decrRefCount(op->key); + zfree(op); + pushed++; + } else { + break; /* too early */ + } + } + return pushed; +} + +void cacheCron(void) { + /* Push jobs */ + cacheScheduleIOPushJobs(0); + + /* Reclaim memory from the object cache */ + while (server.ds_enabled && zmalloc_used_memory() > + server.cache_max_memory) + { + int done = 0; + + if (cacheFreeOneEntry() == REDIS_OK) done++; + if (negativeCacheEvictOneEntry() == REDIS_OK) done++; + if (done == 0) break; /* nothing more to free */ + } +} + +/* ========== Disk store cache - Blocking clients on missing keys =========== */ + +/* This function makes the clinet 'c' waiting for the key 'key' to be loaded. + * If the key is already in memory we don't need to block. + * + * FIXME: we should try if it's actually better to suspend the client + * accessing an object that is being saved, and awake it only when + * the saving was completed. + * + * Otherwise if the key is not in memory, we block the client and start + * an IO Job to load it: + * + * 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. */ +int waitForSwappedKey(redisClient *c, robj *key) { + struct dictEntry *de; + list *l; + + /* Return ASAP if the key is in memory */ + de = dictFind(c->db->dict,key->ptr); + if (de != NULL) return 0; + + /* Don't wait for keys we are sure are not on disk either */ + if (!cacheKeyMayExist(c->db,key)) return 0; + + /* 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); + redisAssert(retval == DICT_OK); + } else { + l = dictGetEntryVal(de); + } + listAddNodeTail(l,c); + + /* Are we already loading the key from disk? If not create a job */ + if (de == NULL) + cacheScheduleIO(c->db,key,REDIS_IO_LOAD); + return 1; +} + +/* Preload keys for any command with first, last and step values for + * the command keys prototype, as defined in the command table. */ +void waitForMultipleSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) { + int j, last; + if (cmd->vm_firstkey == 0) return; + last = cmd->vm_lastkey; + if (last < 0) last = argc+last; + for (j = cmd->vm_firstkey; j <= last; j += cmd->vm_keystep) { + redisAssert(j < argc); + waitForSwappedKey(c,argv[j]); + } +} + +/* Preload keys needed for the ZUNIONSTORE and ZINTERSTORE commands. + * Note that the number of keys to preload is user-defined, so we need to + * apply a sanity check against argc. */ +void zunionInterBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) { + int i, num; + REDIS_NOTUSED(cmd); + + num = atoi(argv[2]->ptr); + if (num > (argc-3)) return; + for (i = 0; i < num; i++) { + waitForSwappedKey(c,argv[3+i]); + } +} + +/* Preload keys needed to execute the entire MULTI/EXEC block. + * + * This function is called by blockClientOnSwappedKeys when EXEC is issued, + * and will block the client when any command requires a swapped out value. */ +void execBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) { + int i, margc; + struct redisCommand *mcmd; + robj **margv; + REDIS_NOTUSED(cmd); + REDIS_NOTUSED(argc); + REDIS_NOTUSED(argv); + + if (!(c->flags & REDIS_MULTI)) return; + for (i = 0; i < c->mstate.count; i++) { + mcmd = c->mstate.commands[i].cmd; + margc = c->mstate.commands[i].argc; + margv = c->mstate.commands[i].argv; + + if (mcmd->vm_preload_proc != NULL) { + mcmd->vm_preload_proc(c,mcmd,margc,margv); + } else { + waitForMultipleSwappedKeys(c,mcmd,margc,margv); + } + } +} + +/* 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. */ +int blockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd) { + if (cmd->vm_preload_proc != NULL) { + cmd->vm_preload_proc(c,cmd,c->argc,c->argv); + } else { + waitForMultipleSwappedKeys(c,cmd,c->argc,c->argv); + } + + /* 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.cache_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). */ +int dontWaitForSwappedKey(redisClient *c, robj *key) { + list *l; + listNode *ln; + listIter li; + struct dictEntry *de; + + /* The key object might be destroyed when deleted from the c->io_keys + * list (and the "key" argument is physically the same object as the + * object inside the list), so we need to protect it. */ + incrRefCount(key); + + /* Remove the key from the list of keys this client is waiting for. */ + listRewind(c->io_keys,&li); + while ((ln = listNext(&li)) != NULL) { + if (equalStringObjects(ln->value,key)) { + listDelNode(c->io_keys,ln); + break; + } + } + redisAssert(ln != NULL); + + /* Remove the client form the key => waiting clients map. */ + de = dictFind(c->db->io_keys,key); + redisAssert(de != NULL); + l = dictGetEntryVal(de); + ln = listSearchKey(l,c); + redisAssert(ln != NULL); + listDelNode(l,ln); + if (listLength(l) == 0) + dictDelete(c->db->io_keys,key); + + decrRefCount(key); + 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. */ +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); + } + } +} diff --git a/src/networking.c b/src/networking.c index 26ee46f6..cd3e787e 100644 --- a/src/networking.c +++ b/src/networking.c @@ -167,7 +167,6 @@ void _addReplyStringToList(redisClient *c, char *s, size_t len) { void addReply(redisClient *c, robj *obj) { if (_installWriteEvent(c) != REDIS_OK) return; - redisAssert(!server.vm_enabled || obj->storage == REDIS_VM_MEMORY); /* This is an important place where we can avoid copy-on-write * when there is a saving child running, avoiding touching the @@ -460,7 +459,7 @@ void freeClient(redisClient *c) { /* Remove from the list of clients waiting for swapped keys, or ready * to be restarted, but not yet woken up again. */ if (c->flags & REDIS_IO_WAIT) { - redisAssert(server.vm_enabled); + redisAssert(server.ds_enabled); if (listLength(c->io_keys) == 0) { ln = listSearchKey(server.io_ready_clients,c); @@ -474,7 +473,7 @@ void freeClient(redisClient *c) { dontWaitForSwappedKey(c,ln->value); } } - server.vm_blocked_clients--; + server.cache_blocked_clients--; } listRelease(c->io_keys); /* Master/slave cleanup. diff --git a/src/object.c b/src/object.c index de62f504..e6b02da8 100644 --- a/src/object.c +++ b/src/object.c @@ -21,7 +21,6 @@ robj *createObject(int type, void *ptr) { /* The following is only needed if VM is active, but since the conditional * is probably more costly than initializing the field it's better to * have every field properly initialized anyway. */ - o->storage = REDIS_VM_MEMORY; return o; } @@ -32,6 +31,7 @@ robj *createStringObject(char *ptr, size_t len) { robj *createStringObjectFromLongLong(long long value) { robj *o; if (value >= 0 && value < REDIS_SHARED_INTEGERS && + !server.ds_enabled && pthread_equal(pthread_self(),server.mainthread)) { incrRefCount(shared.integers[value]); o = shared.integers[value]; @@ -160,30 +160,7 @@ void incrRefCount(robj *o) { void decrRefCount(void *obj) { robj *o = obj; - /* Object is 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)) - { - vmpointer *vp = obj; - if (o->storage == REDIS_VM_LOADING) vmCancelThreadedIOJob(o); - vmMarkPagesFree(vp->page,vp->usedpages); - server.vm_stats_swapped_objects--; - zfree(vp); - return; - } - 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) - vmCancelThreadedIOJob(o); if (--(o->refcount) == 0) { switch(o->type) { case REDIS_STRING: freeStringObject(o); break; @@ -228,16 +205,16 @@ robj *tryObjectEncoding(robj *o) { /* Ok, this object can be encoded... * * Can I use a shared object? Only if the object is inside a given - * range and if this is the main thread, since when VM is enabled we - * have the constraint that I/O thread should only handle non-shared - * objects, in order to avoid race conditions (we don't have per-object - * locking). + * range and if the back end in use is in-memory. For disk store every + * object in memory used as value should be independent. * * Note that we also avoid using shared integers when maxmemory is used - * because very object needs to have a private LRU field for the LRU + * because every object needs to have a private LRU field for the LRU * algorithm to work well. */ - if (server.maxmemory == 0 && value >= 0 && value < REDIS_SHARED_INTEGERS && - pthread_equal(pthread_self(),server.mainthread)) { + if (!server.ds_enabled && + server.maxmemory == 0 && value >= 0 && value < REDIS_SHARED_INTEGERS && + pthread_equal(pthread_self(),server.mainthread)) + { decrRefCount(o); incrRefCount(shared.integers[value]); return shared.integers[value]; diff --git a/src/rdb.c b/src/rdb.c index 5e69a324..60d0a6ce 100644 --- a/src/rdb.c +++ b/src/rdb.c @@ -395,10 +395,29 @@ off_t rdbSavedObjectLen(robj *o) { return len; } -/* Return the number of pages required to save this object in the swap file */ -off_t rdbSavedObjectPages(robj *o) { - off_t bytes = rdbSavedObjectLen(o); - return (bytes+(server.vm_page_size-1))/server.vm_page_size; +/* Save a key-value pair, with expire time, type, key, value. + * On error -1 is returned. + * On success if the key was actaully saved 1 is returned, otherwise 0 + * is returned (the key was already expired). */ +int rdbSaveKeyValuePair(FILE *fp, redisDb *db, robj *key, robj *val, + time_t now) +{ + time_t expiretime; + + expiretime = getExpire(db,key); + + /* Save the expire time */ + if (expiretime != -1) { + /* If this key is already expired skip it */ + if (expiretime < now) return 0; + if (rdbSaveType(fp,REDIS_EXPIRETIME) == -1) return -1; + if (rdbSaveTime(fp,expiretime) == -1) return -1; + } + /* Save type, key, value */ + if (rdbSaveType(fp,val->type) == -1) return -1; + if (rdbSaveStringObject(fp,key) == -1) return -1; + if (rdbSaveObject(fp,val) == -1) return -1; + return 1; } /* Save the DB on disk. Return REDIS_ERR on error, REDIS_OK on success */ @@ -410,11 +429,8 @@ int rdbSave(char *filename) { 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(); + /* FIXME: implement .rdb save for disk store properly */ + redisAssert(server.ds_enabled == 0); snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid()); fp = fopen(tmpfile,"w"); @@ -441,38 +457,9 @@ int rdbSave(char *filename) { while((de = dictNext(di)) != NULL) { sds keystr = dictGetEntryKey(de); robj key, *o = dictGetEntryVal(de); - time_t expiretime; initStaticStringObject(key,keystr); - expiretime = getExpire(db,&key); - - /* Save the expire time */ - if (expiretime != -1) { - /* If this key is already expired skip it */ - if (expiretime < now) continue; - if (rdbSaveType(fp,REDIS_EXPIRETIME) == -1) goto werr; - if (rdbSaveTime(fp,expiretime) == -1) goto werr; - } - /* Save the key and associated value. This requires special - * handling if the value is swapped out. */ - if (!server.vm_enabled || o->storage == REDIS_VM_MEMORY || - o->storage == REDIS_VM_SWAPPING) { - /* Save type, key, value */ - if (rdbSaveType(fp,o->type) == -1) goto werr; - if (rdbSaveStringObject(fp,&key) == -1) goto werr; - if (rdbSaveObject(fp,o) == -1) goto werr; - } else { - /* REDIS_VM_SWAPPED or REDIS_VM_LOADING */ - robj *po; - /* Get a preview of the object in memory */ - po = vmPreviewObject(o); - /* Save type, key, value */ - if (rdbSaveType(fp,po->type) == -1) goto werr; - if (rdbSaveStringObject(fp,&key) == -1) goto werr; - if (rdbSaveObject(fp,po) == -1) goto werr; - /* Remove the loaded object from memory */ - decrRefCount(po); - } + if (rdbSaveKeyValuePair(fp,db,&key,o,now) == -1) goto werr; } dictReleaseIterator(di); } @@ -508,11 +495,10 @@ int rdbSaveBackground(char *filename) { pid_t childpid; if (server.bgsavechildpid != -1) return REDIS_ERR; - if (server.vm_enabled) waitEmptyIOJobsQueue(); + redisAssert(server.ds_enabled == 0); server.dirty_before_bgsave = server.dirty; if ((childpid = fork()) == 0) { /* Child */ - if (server.vm_enabled) vmReopenSwapFile(); if (server.ipfd > 0) close(server.ipfd); if (server.sofd > 0) close(server.sofd); if (rdbSave(filename) == REDIS_OK) { @@ -874,7 +860,6 @@ int rdbLoad(char *filename) { FILE *fp; uint32_t dbid; int type, retval, rdbver; - int swap_all_values = 0; redisDb *db = server.db+0; char buf[1024]; time_t expiretime, now = time(NULL); @@ -899,8 +884,6 @@ int rdbLoad(char *filename) { startLoading(fp); while(1) { robj *key, *val; - int force_swapout; - expiretime = -1; /* Serve the clients from time to time */ @@ -947,44 +930,7 @@ int rdbLoad(char *filename) { /* Set the expire time if needed */ if (expiretime != -1) setExpire(db,key,expiretime); - /* Handle swapping while loading big datasets when VM is on */ - - /* If we detecter we are hopeless about fitting something in memory - * we just swap every new key on disk. Directly... - * Note that's important to check for this condition before resorting - * to random sampling, otherwise we may try to swap already - * swapped keys. */ - if (swap_all_values) { - dictEntry *de = dictFind(db->dict,key->ptr); - - /* de may be NULL since the key already expired */ - if (de) { - vmpointer *vp; - val = dictGetEntryVal(de); - - if (val->refcount == 1 && - (vp = vmSwapObjectBlocking(val)) != NULL) - dictGetEntryVal(de) = vp; - } - decrRefCount(key); - continue; - } decrRefCount(key); - - /* Flush data on disk once 32 MB of additional RAM are used... */ - force_swapout = 0; - if ((zmalloc_used_memory() - server.vm_max_memory) > 1024*1024*32) - force_swapout = 1; - - /* If we have still some hope of having some value fitting memory - * then we try random sampling. */ - if (!swap_all_values && server.vm_enabled && force_swapout) { - while (zmalloc_used_memory() > server.vm_max_memory) { - if (vmSwapOneObjectBlocking() == REDIS_ERR) break; - } - if (zmalloc_used_memory() > server.vm_max_memory) - swap_all_values = 1; /* We are already using too much mem */ - } } fclose(fp); stopLoading(); diff --git a/src/redis.c b/src/redis.c index 8f4ca961..91371a00 100644 --- a/src/redis.c +++ b/src/redis.c @@ -345,7 +345,7 @@ unsigned int dictEncObjHash(const void *key) { } } -/* Sets type */ +/* Sets type and diskstore negative caching hash table */ dictType setDictType = { dictEncObjHash, /* hash function */ NULL, /* key dup */ @@ -596,7 +596,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { } updateDictResizePolicy(); } - } else { + } else if (!server.ds_enabled) { /* If there is not a background saving in progress check if * we have to save now */ time_t now = time(NULL); @@ -618,28 +618,9 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { * in order to guarantee a strict consistency. */ if (server.masterhost == NULL) activeExpireCycle(); - /* Swap a few keys on disk if we are over the memory limit and VM - * is enbled. Try to free objects from the free list first. */ - if (vmCanSwapOut()) { - while (server.vm_enabled && zmalloc_used_memory() > - server.vm_max_memory) - { - int retval = (server.vm_max_threads == 0) ? - vmSwapOneObjectBlocking() : - vmSwapOneObjectThreaded(); - if (retval == REDIS_ERR && !(loops % 300) && - 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 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; - } - } + /* Remove a few cached objects from memory if we are over the + * configured memory limit */ + if (server.ds_enabled) cacheCron(); /* Replication cron function -- used to reconnect to master and * to detect transfer failures. */ @@ -656,8 +637,8 @@ void beforeSleep(struct aeEventLoop *eventLoop) { listNode *ln; redisClient *c; - /* Awake clients that got all the swapped keys they requested */ - if (server.vm_enabled && listLength(server.io_ready_clients)) { + /* Awake clients that got all the on disk keys they requested */ + if (server.ds_enabled && listLength(server.io_ready_clients)) { listIter li; listRewind(server.io_ready_clients,&li); @@ -668,7 +649,7 @@ void beforeSleep(struct aeEventLoop *eventLoop) { /* Resume the client. */ listDelNode(server.io_ready_clients,ln); c->flags &= (~REDIS_IO_WAIT); - server.vm_blocked_clients--; + server.cache_blocked_clients--; aeCreateFileEvent(server.el, c->fd, AE_READABLE, readQueryFromClient, c); cmd = lookupCommand(c->argv[0]->ptr); @@ -787,19 +768,17 @@ void initServerConfig() { server.maxmemory = 0; server.maxmemory_policy = REDIS_MAXMEMORY_VOLATILE_LRU; server.maxmemory_samples = 3; - 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; + server.ds_enabled = 0; + server.ds_path = sdsnew("/tmp/redis.ds"); + server.cache_max_memory = 64LL*1024*1024; /* 64 MB of RAM */ + server.cache_blocked_clients = 0; server.hash_max_zipmap_entries = REDIS_HASH_MAX_ZIPMAP_ENTRIES; server.hash_max_zipmap_value = REDIS_HASH_MAX_ZIPMAP_VALUE; server.list_max_ziplist_entries = REDIS_LIST_MAX_ZIPLIST_ENTRIES; server.list_max_ziplist_value = REDIS_LIST_MAX_ZIPLIST_VALUE; server.set_max_intset_entries = REDIS_SET_MAX_INTSET_ENTRIES; server.shutdown_asap = 0; + server.cache_flush_delay = 0; updateLRUClock(); resetServerSaveParams(); @@ -848,6 +827,8 @@ void initServer() { server.slaves = listCreate(); server.monitors = listCreate(); server.unblocked_clients = listCreate(); + server.cache_io_queue = listCreate(); + createSharedObjects(); server.el = aeCreateEventLoop(); server.db = zmalloc(sizeof(redisDb)*server.dbnum); @@ -873,8 +854,11 @@ void initServer() { server.db[j].expires = dictCreate(&keyptrDictType,NULL); server.db[j].blocking_keys = dictCreate(&keylistDictType,NULL); server.db[j].watched_keys = dictCreate(&keylistDictType,NULL); - if (server.vm_enabled) + if (server.ds_enabled) { server.db[j].io_keys = dictCreate(&keylistDictType,NULL); + server.db[j].io_negcache = dictCreate(&setDictType,NULL); + server.db[j].io_queued = dictCreate(&setDictType,NULL); + } server.db[j].id = j; } server.pubsub_channels = dictCreate(&keylistDictType,NULL); @@ -911,7 +895,7 @@ void initServer() { } } - if (server.vm_enabled) vmInit(); + if (server.ds_enabled) dsInit(); } /* Populates the Redis Command Table starting from the hard coded list @@ -1050,8 +1034,8 @@ int processCommand(redisClient *c) { queueMultiCommand(c,cmd); addReply(c,shared.queued); } else { - if (server.vm_enabled && server.vm_max_threads > 0 && - blockClientOnSwappedKeys(c,cmd)) return REDIS_ERR; + if (server.ds_enabled && blockClientOnSwappedKeys(c,cmd)) + return REDIS_ERR; call(c,cmd); } return REDIS_OK; @@ -1069,10 +1053,11 @@ int prepareForShutdown() { kill(server.bgsavechildpid,SIGKILL); rdbRemoveTempFile(server.bgsavechildpid); } - if (server.appendonly) { + if (server.ds_enabled) { + /* FIXME: flush all objects on disk */ + } else if (server.appendonly) { /* Append only file: fsync() the AOF and exit */ aof_fsync(server.appendfd); - if (server.vm_enabled) unlink(server.vm_swap_file); } else if (server.saveparamslen > 0) { /* Snapshotting. Perform a SYNC SAVE and exit */ if (rdbSave(server.dbfilename) != REDIS_OK) { @@ -1185,7 +1170,7 @@ sds genRedisInfoString(void) { "hash_max_zipmap_value:%zu\r\n" "pubsub_channels:%ld\r\n" "pubsub_patterns:%u\r\n" - "vm_enabled:%d\r\n" + "ds_enabled:%d\r\n" "role:%s\r\n" ,REDIS_VERSION, redisGitSHA1(), @@ -1228,7 +1213,7 @@ sds genRedisInfoString(void) { server.hash_max_zipmap_value, dictSize(server.pubsub_channels), listLength(server.pubsub_patterns), - server.vm_enabled != 0, + server.ds_enabled != 0, server.masterhost == NULL ? "master" : "slave" ); if (server.masterhost) { @@ -1255,33 +1240,13 @@ sds genRedisInfoString(void) { ); } } - if (server.vm_enabled) { + if (server.ds_enabled) { lockThreadedIO(); info = sdscatprintf(info, - "vm_conf_max_memory:%llu\r\n" - "vm_conf_page_size:%llu\r\n" - "vm_conf_pages:%llu\r\n" - "vm_stats_used_pages:%llu\r\n" - "vm_stats_swapped_objects:%llu\r\n" - "vm_stats_swappin_count:%llu\r\n" - "vm_stats_swappout_count:%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_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 long) server.vm_stats_used_pages, - (unsigned long long) server.vm_stats_swapped_objects, - (unsigned long long) server.vm_stats_swapins, - (unsigned long long) server.vm_stats_swapouts, - (unsigned long) listLength(server.io_newjobs), - (unsigned long) listLength(server.io_processing), - (unsigned long) listLength(server.io_processed), - (unsigned long) server.io_active_threads, - (unsigned long) server.vm_blocked_clients + "cache_max_memory:%llu\r\n" + "cache_blocked_clients:%lu\r\n" + ,(unsigned long long) server.cache_max_memory, + (unsigned long) server.cache_blocked_clients ); unlockThreadedIO(); } @@ -1533,7 +1498,9 @@ int main(int argc, char **argv) { linuxOvercommitMemoryWarning(); #endif start = time(NULL); - if (server.appendonly) { + if (server.ds_enabled) { + redisLog(REDIS_NOTICE,"DB not loaded (running with disk back end)"); + } else if (server.appendonly) { if (loadAppendOnlyFile(server.appendfilename) == REDIS_OK) redisLog(REDIS_NOTICE,"DB loaded from append only file: %ld seconds",time(NULL)-start); } else { diff --git a/src/redis.h b/src/redis.h index bc1a58c1..b15bb370 100644 --- a/src/redis.h +++ b/src/redis.h @@ -119,23 +119,14 @@ #define REDIS_RDB_ENC_INT32 2 /* 32 bit signed integer */ #define REDIS_RDB_ENC_LZF 3 /* string compressed with FASTLZ */ -/* Virtual memory object->where field. */ -#define REDIS_VM_MEMORY 0 /* The object is on memory */ -#define REDIS_VM_SWAPPED 1 /* The object is on disk */ -#define REDIS_VM_SWAPPING 2 /* Redis is swapping this object on disk */ -#define REDIS_VM_LOADING 3 /* Redis is loading this object from disk */ - -/* Virtual memory static configuration stuff. - * Check vmFindContiguousPages() to know more about this magic numbers. */ -#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. */ +/* Scheduled IO opeations flags. */ +#define REDIS_IO_LOAD 1 +#define REDIS_IO_SAVE 2 +#define REDIS_IO_LOADINPROG 4 +#define REDIS_IO_SAVEINPROG 8 + #define REDIS_MAX_COMPLETED_JOBS_PROCESSED 1 +#define REDIS_THREAD_STACK_SIZE (1024*1024*4) /* Client flags */ #define REDIS_SLAVE 1 /* This client is a slave server */ @@ -230,7 +221,7 @@ void _redisPanic(char *msg, char *file, int line); #define REDIS_LRU_CLOCK_RESOLUTION 10 /* LRU clock resolution in seconds */ typedef struct redisObject { unsigned type:4; - unsigned storage:2; /* REDIS_VM_MEMORY or REDIS_VM_SWAPPING */ + unsigned notused:2; /* Not used */ unsigned encoding:4; unsigned lru:22; /* lru time (relative to server.lruclock) */ int refcount; @@ -271,14 +262,15 @@ typedef struct vmPointer { _var.type = REDIS_STRING; \ _var.encoding = REDIS_ENCODING_RAW; \ _var.ptr = _ptr; \ - _var.storage = REDIS_VM_MEMORY; \ } while(0); typedef struct redisDb { dict *dict; /* The keyspace for this DB */ dict *expires; /* Timeout of keys with a timeout set */ dict *blocking_keys; /* Keys with clients waiting for data (BLPOP) */ - dict *io_keys; /* Keys with clients waiting for VM I/O */ + dict *io_keys; /* Keys with clients waiting for DS I/O */ + dict *io_negcache; /* Negative caching for disk store */ + dict *io_queued; /* Queued IO operations hash table */ dict *watched_keys; /* WATCHED keys for MULTI/EXEC CAS */ int id; } redisDb; @@ -440,31 +432,25 @@ struct redisServer { int maxmemory_samples; /* Blocked clients */ unsigned int bpop_blocked_clients; - unsigned int vm_blocked_clients; - list *unblocked_clients; + unsigned int cache_blocked_clients; + list *unblocked_clients; /* list of clients to unblock before next loop */ + list *cache_io_queue; /* IO operations queue */ + int cache_flush_delay; /* seconds to wait before flushing keys */ /* 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_alpha; 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; + int ds_enabled; /* backend disk in redis.conf */ + char *ds_path; /* location of the disk store on disk */ + unsigned long long cache_max_memory; /* Zip structure config */ size_t hash_max_zipmap_entries; size_t hash_max_zipmap_value; size_t list_max_ziplist_entries; size_t list_max_ziplist_value; size_t set_max_intset_entries; - /* Virtual memory state */ - FILE *vm_fp; - int vm_fd; - off_t vm_next_page; /* Next probably empty page */ - off_t vm_near_pages; /* Number of pages allocated sequentially */ - unsigned char *vm_bitmap; /* Bitmap of free/used pages */ time_t unixtime; /* Unix time sampled every second. */ /* Virtual memory I/O threads stuff */ /* An I/O thread process an element taken from the io_jobs queue and @@ -475,7 +461,7 @@ struct redisServer { list *io_processed; /* List of VM I/O jobs already processed */ 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 io_swapfile_mutex; /* So we can lseek + write */ + pthread_cond_t io_condvar; /* I/O threads conditional variable */ 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 */ @@ -560,25 +546,27 @@ typedef struct zset { zskiplist *zsl; } zset; -/* VM threaded I/O request message */ -#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 */ +/* DIsk store threaded I/O request message */ +#define REDIS_IOJOB_LOAD 0 +#define REDIS_IOJOB_SAVE 1 + typedef struct iojob { int type; /* Request type, REDIS_IOJOB_* */ redisDb *db;/* Redis database */ - 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 - * 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 pages; /* Swap pages needed to save object. PREPARE_SWAP return val */ - int canceled; /* True if this command was canceled by blocking side of VM */ - pthread_t thread; /* ID of the thread processing this entry */ + robj *key; /* This I/O request is about this key */ + robj *val; /* the value to swap for REDIS_IOJOB_SAVE, otherwise this + * field is populated by the I/O thread for REDIS_IOJOB_LOAD. */ + time_t expire; /* Expire time for this key on REDIS_IOJOB_LOAD */ } iojob; +/* IO operations scheduled -- check dscache.c for more info */ +typedef struct ioop { + int type; + redisDb *db; + robj *key; + time_t ctime; /* This is the creation time of the entry. */ +} ioop; + /* Structure to hold list iteration abstraction. */ typedef struct { robj *subject; @@ -761,6 +749,10 @@ off_t rdbSavedObjectLen(robj *o); off_t rdbSavedObjectPages(robj *o); robj *rdbLoadObject(int type, FILE *fp); void backgroundSaveDoneHandler(int statloc); +int rdbSaveKeyValuePair(FILE *fp, redisDb *db, robj *key, robj *val, time_t now); +int rdbLoadType(FILE *fp); +time_t rdbLoadTime(FILE *fp); +robj *rdbLoadStringObject(FILE *fp); /* AOF persistence */ void flushAppendOnlyFile(void); @@ -792,32 +784,38 @@ int htNeedsResize(dict *dict); void oom(const char *msg); void populateCommandTable(void); -/* Virtual Memory */ -void vmInit(void); -void vmMarkPagesFree(off_t page, off_t count); -robj *vmLoadObject(robj *o); -robj *vmPreviewObject(robj *o); -int vmSwapOneObjectBlocking(void); -int vmSwapOneObjectThreaded(void); -int vmCanSwapOut(void); +/* Disk store */ +int dsOpen(void); +int dsClose(void); +int dsSet(redisDb *db, robj *key, robj *val); +robj *dsGet(redisDb *db, robj *key, time_t *expire); +int dsDel(redisDb *db, robj *key); +int dsExists(redisDb *db, robj *key); +void dsFlushDb(int dbid); + +/* Disk Store Cache */ +void dsInit(void); void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata, int mask); -void vmCancelThreadedIOJob(robj *o); void lockThreadedIO(void); void unlockThreadedIO(void); -int vmSwapObjectThreaded(robj *key, robj *val, redisDb *db); void freeIOJob(iojob *j); void queueIOJob(iojob *j); -int vmWriteObjectOnSwap(robj *o, off_t page); -robj *vmReadObjectFromSwap(off_t page, int type); void waitEmptyIOJobsQueue(void); -void vmReopenSwapFile(void); -int vmFreePage(off_t page); +void processAllPendingIOJobs(void); void zunionInterBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv); void execBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv); int blockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd); int dontWaitForSwappedKey(redisClient *c, robj *key); void handleClientsBlockedOnSwappedKey(redisDb *db, robj *key); -vmpointer *vmSwapObjectBlocking(robj *val); +int cacheFreeOneEntry(void); +void cacheScheduleIOAddFlag(redisDb *db, robj *key, long flag); +void cacheScheduleIODelFlag(redisDb *db, robj *key, long flag); +int cacheScheduleIOGetFlags(redisDb *db, robj *key); +void cacheScheduleIO(redisDb *db, robj *key, int type); +void cacheCron(void); +int cacheKeyMayExist(redisDb *db, robj *key); +void cacheSetKeyMayExist(redisDb *db, robj *key); +void cacheSetKeyDoesNotExist(redisDb *db, robj *key); /* Set data type */ robj *setTypeCreate(robj *value); @@ -888,6 +886,8 @@ robj *dbRandomKey(redisDb *db); int dbDelete(redisDb *db, robj *key); long long emptyDb(); int selectDb(redisClient *c, int id); +void signalModifiedKey(redisDb *db, robj *key); +void signalFlushedDb(int dbid); /* Git SHA1 */ char *redisGitSHA1(void); diff --git a/src/sort.c b/src/sort.c index a44a6d63..1cf8932e 100644 --- a/src/sort.c +++ b/src/sort.c @@ -368,7 +368,7 @@ void sortCommand(redisClient *c) { * SORT result is empty a new key is set and maybe the old content * replaced. */ server.dirty += 1+outputlen; - touchWatchedKey(c->db,storekey); + signalModifiedKey(c->db,storekey); addReplyLongLong(c,outputlen); } diff --git a/src/t_hash.c b/src/t_hash.c index 488bf6b7..5e7525bd 100644 --- a/src/t_hash.c +++ b/src/t_hash.c @@ -279,7 +279,7 @@ void hsetCommand(redisClient *c) { hashTypeTryObjectEncoding(o,&c->argv[2], &c->argv[3]); update = hashTypeSet(o,c->argv[2],c->argv[3]); addReply(c, update ? shared.czero : shared.cone); - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; } @@ -294,7 +294,7 @@ void hsetnxCommand(redisClient *c) { hashTypeTryObjectEncoding(o,&c->argv[2], &c->argv[3]); hashTypeSet(o,c->argv[2],c->argv[3]); addReply(c, shared.cone); - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; } } @@ -315,7 +315,7 @@ void hmsetCommand(redisClient *c) { hashTypeSet(o,c->argv[i],c->argv[i+1]); } addReply(c, shared.ok); - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; } @@ -342,7 +342,7 @@ void hincrbyCommand(redisClient *c) { hashTypeSet(o,c->argv[2],new); decrRefCount(new); addReplyLongLong(c,value); - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; } @@ -402,7 +402,7 @@ void hdelCommand(redisClient *c) { if (hashTypeDelete(o,c->argv[2])) { if (hashTypeLength(o) == 0) dbDelete(c->db,c->argv[1]); addReply(c,shared.cone); - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; } else { addReply(c,shared.czero); diff --git a/src/t_list.c b/src/t_list.c index 8ee3b987..d5db6feb 100644 --- a/src/t_list.c +++ b/src/t_list.c @@ -274,14 +274,14 @@ void pushGenericCommand(redisClient *c, int where) { return; } if (handleClientsWaitingListPush(c,c->argv[1],c->argv[2])) { - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); addReply(c,shared.cone); return; } } listTypePush(lobj,c->argv[2],where); addReplyLongLong(c,listTypeLength(lobj)); - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; } @@ -330,7 +330,7 @@ void pushxGenericCommand(redisClient *c, robj *refval, robj *val, int where) { if (subject->encoding == REDIS_ENCODING_ZIPLIST && ziplistLen(subject->ptr) > server.list_max_ziplist_entries) listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST); - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; } else { /* Notify client of a failed insert */ @@ -339,7 +339,7 @@ void pushxGenericCommand(redisClient *c, robj *refval, robj *val, int where) { } } else { listTypePush(subject,val,where); - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; } @@ -427,7 +427,7 @@ void lsetCommand(redisClient *c) { o->ptr = ziplistInsert(o->ptr,p,value->ptr,sdslen(value->ptr)); decrRefCount(value); addReply(c,shared.ok); - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; } } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { @@ -439,7 +439,7 @@ void lsetCommand(redisClient *c) { listNodeValue(ln) = value; incrRefCount(value); addReply(c,shared.ok); - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; } } else { @@ -458,7 +458,7 @@ void popGenericCommand(redisClient *c, int where) { addReplyBulk(c,value); decrRefCount(value); if (listTypeLength(o) == 0) dbDelete(c->db,c->argv[1]); - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; } } @@ -573,7 +573,7 @@ void ltrimCommand(redisClient *c) { redisPanic("Unknown list encoding"); } if (listTypeLength(o) == 0) dbDelete(c->db,c->argv[1]); - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; addReply(c,shared.ok); } @@ -616,7 +616,7 @@ void lremCommand(redisClient *c) { if (listTypeLength(subject) == 0) dbDelete(c->db,c->argv[1]); addReplyLongLong(c,removed); - if (removed) touchWatchedKey(c->db,c->argv[1]); + if (removed) signalModifiedKey(c->db,c->argv[1]); } /* This is the semantic of this command: @@ -642,7 +642,7 @@ void rpoplpushHandlePush(redisClient *c, robj *dstkey, robj *dstobj, robj *value dstobj = createZiplistObject(); dbAdd(c->db,dstkey,dstobj); } else { - touchWatchedKey(c->db,dstkey); + signalModifiedKey(c->db,dstkey); server.dirty++; } listTypePush(dstobj,value,REDIS_HEAD); @@ -670,7 +670,7 @@ void rpoplpushCommand(redisClient *c) { /* Delete the source list when it is empty */ if (listTypeLength(sobj) == 0) dbDelete(c->db,c->argv[1]); - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; } } diff --git a/src/t_set.c b/src/t_set.c index 0b4128ad..2af383d1 100644 --- a/src/t_set.c +++ b/src/t_set.c @@ -231,7 +231,7 @@ void saddCommand(redisClient *c) { } } if (setTypeAdd(set,c->argv[2])) { - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; addReply(c,shared.cone); } else { @@ -248,7 +248,7 @@ void sremCommand(redisClient *c) { c->argv[2] = tryObjectEncoding(c->argv[2]); if (setTypeRemove(set,c->argv[2])) { if (setTypeSize(set) == 0) dbDelete(c->db,c->argv[1]); - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; addReply(c,shared.cone); } else { @@ -287,8 +287,8 @@ void smoveCommand(redisClient *c) { /* Remove the src set from the database when empty */ if (setTypeSize(srcset) == 0) dbDelete(c->db,c->argv[1]); - touchWatchedKey(c->db,c->argv[1]); - touchWatchedKey(c->db,c->argv[2]); + signalModifiedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[2]); server.dirty++; /* Create the destination set when it doesn't exist */ @@ -341,7 +341,7 @@ void spopCommand(redisClient *c) { setTypeRemove(set,ele); } if (setTypeSize(set) == 0) dbDelete(c->db,c->argv[1]); - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; } @@ -382,7 +382,7 @@ void sinterGenericCommand(redisClient *c, robj **setkeys, unsigned long setnum, zfree(sets); if (dstkey) { if (dbDelete(c->db,dstkey)) { - touchWatchedKey(c->db,dstkey); + signalModifiedKey(c->db,dstkey); server.dirty++; } addReply(c,shared.czero); @@ -486,7 +486,7 @@ void sinterGenericCommand(redisClient *c, robj **setkeys, unsigned long setnum, decrRefCount(dstset); addReply(c,shared.czero); } - touchWatchedKey(c->db,dstkey); + signalModifiedKey(c->db,dstkey); server.dirty++; } else { setDeferredMultiBulkLength(c,replylen,cardinality); @@ -578,7 +578,7 @@ void sunionDiffGenericCommand(redisClient *c, robj **setkeys, int setnum, robj * decrRefCount(dstset); addReply(c,shared.czero); } - touchWatchedKey(c->db,dstkey); + signalModifiedKey(c->db,dstkey); server.dirty++; } zfree(sets); diff --git a/src/t_string.c b/src/t_string.c index 58834289..da065e88 100644 --- a/src/t_string.c +++ b/src/t_string.c @@ -37,7 +37,7 @@ void setGenericCommand(redisClient *c, int nx, robj *key, robj *val, robj *expir } else { incrRefCount(val); } - touchWatchedKey(c->db,key); + signalModifiedKey(c->db,key); server.dirty++; removeExpire(c->db,key); if (expire) setExpire(c->db,key,time(NULL)+seconds); @@ -83,7 +83,7 @@ void getsetCommand(redisClient *c) { c->argv[2] = tryObjectEncoding(c->argv[2]); dbReplace(c->db,c->argv[1],c->argv[2]); incrRefCount(c->argv[2]); - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; removeExpire(c->db,c->argv[1]); } @@ -155,7 +155,7 @@ void setbitCommand(redisClient *c) { byteval &= ~(1 << bit); byteval |= ((on & 0x1) << bit); ((char*)o->ptr)[byte] = byteval; - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; addReply(c, bitval ? shared.cone : shared.czero); } @@ -243,7 +243,7 @@ void setrangeCommand(redisClient *c) { if (sdslen(value) > 0) { o->ptr = sdsgrowzero(o->ptr,offset+sdslen(value)); memcpy((char*)o->ptr+offset,value,sdslen(value)); - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; } addReplyLongLong(c,sdslen(o->ptr)); @@ -330,7 +330,7 @@ void msetGenericCommand(redisClient *c, int nx) { dbReplace(c->db,c->argv[j],c->argv[j+1]); incrRefCount(c->argv[j+1]); removeExpire(c->db,c->argv[j]); - touchWatchedKey(c->db,c->argv[j]); + signalModifiedKey(c->db,c->argv[j]); } server.dirty += (c->argc-1)/2; addReply(c, nx ? shared.cone : shared.ok); @@ -360,7 +360,7 @@ void incrDecrCommand(redisClient *c, long long incr) { } o = createStringObjectFromLongLong(value); dbReplace(c->db,c->argv[1],o); - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; addReply(c,shared.colon); addReply(c,o); @@ -423,7 +423,7 @@ void appendCommand(redisClient *c) { o->ptr = sdscatlen(o->ptr,append->ptr,sdslen(append->ptr)); totlen = sdslen(o->ptr); } - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; addReplyLongLong(c,totlen); } diff --git a/src/t_zset.c b/src/t_zset.c index 8139b53d..27522367 100644 --- a/src/t_zset.c +++ b/src/t_zset.c @@ -399,7 +399,7 @@ void zaddGenericCommand(redisClient *c, robj *key, robj *ele, double score, int de = dictFind(zs->dict,ele); redisAssert(de != NULL); dictGetEntryVal(de) = &znode->score; - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; if (incr) addReplyDouble(c,score); @@ -427,7 +427,7 @@ void zaddGenericCommand(redisClient *c, robj *key, robj *ele, double score, int /* Update the score in the current dict entry */ dictGetEntryVal(de) = &znode->score; - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; } if (incr) @@ -477,7 +477,7 @@ void zremCommand(redisClient *c) { dictDelete(zs->dict,c->argv[2]); if (htNeedsResize(zs->dict)) dictResize(zs->dict); if (dictSize(zs->dict) == 0) dbDelete(c->db,c->argv[1]); - touchWatchedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[1]); server.dirty++; addReply(c,shared.cone); } @@ -501,7 +501,7 @@ void zremrangebyscoreCommand(redisClient *c) { deleted = zslDeleteRangeByScore(zs->zsl,range,zs->dict); if (htNeedsResize(zs->dict)) dictResize(zs->dict); if (dictSize(zs->dict) == 0) dbDelete(c->db,c->argv[1]); - if (deleted) touchWatchedKey(c->db,c->argv[1]); + if (deleted) signalModifiedKey(c->db,c->argv[1]); server.dirty += deleted; addReplyLongLong(c,deleted); } @@ -540,7 +540,7 @@ void zremrangebyrankCommand(redisClient *c) { deleted = zslDeleteRangeByRank(zs->zsl,start+1,end+1,zs->dict); if (htNeedsResize(zs->dict)) dictResize(zs->dict); if (dictSize(zs->dict) == 0) dbDelete(c->db,c->argv[1]); - if (deleted) touchWatchedKey(c->db,c->argv[1]); + if (deleted) signalModifiedKey(c->db,c->argv[1]); server.dirty += deleted; addReplyLongLong(c, deleted); } @@ -741,14 +741,14 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) { } if (dbDelete(c->db,dstkey)) { - touchWatchedKey(c->db,dstkey); + signalModifiedKey(c->db,dstkey); touched = 1; server.dirty++; } if (dstzset->zsl->length) { dbAdd(c->db,dstkey,dstobj); addReplyLongLong(c, dstzset->zsl->length); - if (!touched) touchWatchedKey(c->db,dstkey); + if (!touched) signalModifiedKey(c->db,dstkey); server.dirty++; } else { decrRefCount(dstobj); diff --git a/src/version.h b/src/version.h index ff8473e7..3913b007 100644 --- a/src/version.h +++ b/src/version.h @@ -1 +1 @@ -#define REDIS_VERSION "2.1.8" +#define REDIS_VERSION "2.3.0" diff --git a/src/vm.c b/src/vm.c deleted file mode 100644 index ac0d92e3..00000000 --- a/src/vm.c +++ /dev/null @@ -1,1149 +0,0 @@ -#include "redis.h" - -#include -#include -#include -#include - -/* Virtual Memory is composed mainly of two subsystems: - * - Blocking Virutal Memory - * - Threaded Virtual Memory I/O - * The two parts are not fully decoupled, but functions are split among two - * different sections of the source code (delimited by comments) in order to - * make more clear what functionality is about the blocking VM and what about - * the threaded (not blocking) VM. - * - * Redis VM design: - * - * Redis VM is a blocking VM (one that blocks reading swapped values from - * disk into memory when a value swapped out is needed in memory) that is made - * unblocking by trying to examine the command argument vector in order to - * load in background values that will likely be needed in order to exec - * the command. The command is executed only once all the relevant keys - * are loaded into memory. - * - * This basically is almost as simple of a blocking VM, but almost as parallel - * as a fully non-blocking VM. - */ - -/* =================== Virtual Memory - Blocking Side ====================== */ - -/* Create a VM pointer object. This kind of objects are used in place of - * values in the key -> value hash table, for swapped out objects. */ -vmpointer *createVmPointer(int vtype) { - vmpointer *vp = zmalloc(sizeof(vmpointer)); - - vp->type = REDIS_VMPOINTER; - vp->storage = REDIS_VM_SWAPPED; - vp->vtype = vtype; - return vp; -} - -void vmInit(void) { - off_t totsize; - int pipefds[2]; - size_t stacksize; - struct flock fl; - - if (server.vm_max_threads != 0) - zmalloc_enable_thread_safeness(); /* we need thread safe zmalloc() */ - - redisLog(REDIS_NOTICE,"Using '%s' as swap file",server.vm_swap_file); - /* Try to open the old swap file, otherwise create it */ - 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, - "Can't open the swap file: %s. Exiting.", - strerror(errno)); - exit(1); - } - server.vm_fd = fileno(server.vm_fp); - /* Lock the swap file for writing, this is useful in order to avoid - * another instance to use the same swap file for a config error. */ - fl.l_type = F_WRLCK; - fl.l_whence = SEEK_SET; - fl.l_start = fl.l_len = 0; - if (fcntl(server.vm_fd,F_SETLK,&fl) == -1) { - redisLog(REDIS_WARNING, - "Can't lock the swap file at '%s': %s. Make sure it is not used by another Redis instance.", server.vm_swap_file, strerror(errno)); - exit(1); - } - /* Initialize */ - server.vm_next_page = 0; - server.vm_near_pages = 0; - server.vm_stats_used_pages = 0; - server.vm_stats_swapped_objects = 0; - server.vm_stats_swapouts = 0; - server.vm_stats_swapins = 0; - totsize = server.vm_pages*server.vm_page_size; - redisLog(REDIS_NOTICE,"Allocating %lld bytes of swap file",totsize); - if (ftruncate(server.vm_fd,totsize) == -1) { - redisLog(REDIS_WARNING,"Can't ftruncate swap file: %s. Exiting.", - strerror(errno)); - exit(1); - } else { - redisLog(REDIS_NOTICE,"Swap file allocated with success"); - } - server.vm_bitmap = zcalloc((server.vm_pages+7)/8); - redisLog(REDIS_VERBOSE,"Allocated %lld bytes page table for %lld pages", - (long long) (server.vm_pages+7)/8, server.vm_pages); - - /* Initialize threaded I/O (used by Virtual Memory) */ - server.io_newjobs = listCreate(); - server.io_processing = listCreate(); - server.io_processed = listCreate(); - server.io_ready_clients = listCreate(); - pthread_mutex_init(&server.io_mutex,NULL); - pthread_mutex_init(&server.io_swapfile_mutex,NULL); - server.io_active_threads = 0; - if (pipe(pipefds) == -1) { - redisLog(REDIS_WARNING,"Unable to intialized VM: pipe(2): %s. Exiting." - ,strerror(errno)); - exit(1); - } - 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); - - /* Solaris may report a stacksize of 0, let's set it to 1 otherwise - * multiplying it by 2 in the while loop later will not really help ;) */ - if (!stacksize) stacksize = 1; - - 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) - oom("creating file event"); -} - -/* Mark the page as used */ -void vmMarkPageUsed(off_t page) { - off_t byte = page/8; - int bit = page&7; - redisAssert(vmFreePage(page) == 1); - server.vm_bitmap[byte] |= 1<= server.vm_pages) { - this -= server.vm_pages; - if (this == 0) { - /* Just overflowed, what we found on tail is no longer - * interesting, as it's no longer contiguous. */ - numfree = 0; - } - } - if (vmFreePage(this)) { - /* This is a free page */ - numfree++; - /* Already got N free pages? Return to the caller, with success */ - if (numfree == n) { - *first = this-(n-1); - server.vm_next_page = this+1; - redisLog(REDIS_DEBUG, "FOUND CONTIGUOUS PAGES: %lld pages at %lld\n", (long long) n, (long long) *first); - return REDIS_OK; - } - } else { - /* The current one is not a free page */ - numfree = 0; - } - - /* Fast-forward if the current page is not free and we already - * searched enough near this place. */ - since_jump++; - if (!numfree && since_jump >= REDIS_VM_MAX_RANDOM_JUMP/4) { - offset += random() % REDIS_VM_MAX_RANDOM_JUMP; - since_jump = 0; - /* Note that even if we rewind after the jump, we are don't need - * to make sure numfree is set to zero as we only jump *if* it - * is set to zero. */ - } else { - /* Otherwise just check the next page */ - offset++; - } - } - return REDIS_ERR; -} - -/* Write the specified object at the specified page of the swap file */ -int vmWriteObjectOnSwap(robj *o, off_t page) { - if (server.vm_enabled) pthread_mutex_lock(&server.io_swapfile_mutex); - if (fseeko(server.vm_fp,page*server.vm_page_size,SEEK_SET) == -1) { - if (server.vm_enabled) pthread_mutex_unlock(&server.io_swapfile_mutex); - redisLog(REDIS_WARNING, - "Critical VM problem in vmWriteObjectOnSwap(): can't seek: %s", - strerror(errno)); - return REDIS_ERR; - } - rdbSaveObject(server.vm_fp,o); - fflush(server.vm_fp); - if (server.vm_enabled) pthread_mutex_unlock(&server.io_swapfile_mutex); - return REDIS_OK; -} - -/* Transfers the 'val' object to disk. Store all the information - * a 'vmpointer' object containing all the information needed to load the - * object back later is returned. - * - * If we can't find enough contiguous empty pages to swap the object on disk - * NULL is returned. */ -vmpointer *vmSwapObjectBlocking(robj *val) { - off_t pages = rdbSavedObjectPages(val); - off_t page; - vmpointer *vp; - - redisAssert(val->storage == REDIS_VM_MEMORY); - redisAssert(val->refcount == 1); - if (vmFindContiguousPages(&page,pages) == REDIS_ERR) return NULL; - if (vmWriteObjectOnSwap(val,page) == REDIS_ERR) return NULL; - - vp = createVmPointer(val->type); - vp->page = page; - vp->usedpages = pages; - decrRefCount(val); /* Deallocate the object from memory. */ - vmMarkPagesUsed(page,pages); - redisLog(REDIS_DEBUG,"VM: object %p swapped out at %lld (%lld pages)", - (void*) val, - (unsigned long long) page, (unsigned long long) pages); - server.vm_stats_swapped_objects++; - server.vm_stats_swapouts++; - return vp; -} - -robj *vmReadObjectFromSwap(off_t page, int type) { - robj *o; - - 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 vmReadObjectFromSwap(): can't seek: %s", - strerror(errno)); - _exit(1); - } - o = rdbLoadObject(type,server.vm_fp); - if (o == NULL) { - 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; -} - -/* Load the specified object from swap to memory. - * The newly allocated object is returned. - * - * If preview is true the unserialized object is returned to the caller but - * the pages are not marked as freed, nor the vp object is freed. */ -robj *vmGenericLoadObject(vmpointer *vp, int preview) { - robj *val; - - redisAssert(vp->type == REDIS_VMPOINTER && - (vp->storage == REDIS_VM_SWAPPED || vp->storage == REDIS_VM_LOADING)); - val = vmReadObjectFromSwap(vp->page,vp->vtype); - if (!preview) { - redisLog(REDIS_DEBUG, "VM: object %p loaded from disk", (void*)vp); - vmMarkPagesFree(vp->page,vp->usedpages); - zfree(vp); - server.vm_stats_swapped_objects--; - } else { - redisLog(REDIS_DEBUG, "VM: object %p previewed from disk", (void*)vp); - } - server.vm_stats_swapins++; - return val; -} - -/* Plain object loading, from swap to memory. - * - * 'o' is actually a redisVmPointer structure that will be freed by the call. - * The return value is the loaded object. */ -robj *vmLoadObject(robj *o) { - /* If we are loading the object in background, stop it, we - * need to load this object synchronously ASAP. */ - if (o->storage == REDIS_VM_LOADING) - vmCancelThreadedIOJob(o); - return vmGenericLoadObject((vmpointer*)o,0); -} - -/* Just load the value on disk, without to modify the key. - * 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 - * dataset or rewriting the append only log. */ -robj *vmPreviewObject(robj *o) { - return vmGenericLoadObject((vmpointer*)o,1); -} - -/* How a good candidate is this object for swapping? - * The better candidate it is, the greater the returned value. - * - * Currently we try to perform a fast estimation of the object size in - * memory, and combine it with aging informations. - * - * Basically swappability = idle-time * log(estimated size) - * - * Bigger objects are preferred over smaller objects, but not - * proportionally, this is why we use the logarithm. This algorithm is - * just a first try and will probably be tuned later. */ -double computeObjectSwappability(robj *o) { - /* actual age can be >= minage, but not < minage. As we use wrapping - * 21 bit clocks with minutes resolution for the LRU. */ - time_t minage = estimateObjectIdleTime(o); - long asize = 0, elesize; - robj *ele; - list *l; - listNode *ln; - dict *d; - struct dictEntry *de; - int z; - - if (minage <= 0) return 0; - switch(o->type) { - case REDIS_STRING: - if (o->encoding != REDIS_ENCODING_RAW) { - asize = sizeof(*o); - } else { - asize = sdslen(o->ptr)+sizeof(*o)+sizeof(long)*2; - } - break; - case REDIS_LIST: - if (o->encoding == REDIS_ENCODING_ZIPLIST) { - asize = sizeof(*o)+ziplistSize(o->ptr); - } else { - l = o->ptr; - ln = listFirst(l); - asize = sizeof(list); - if (ln) { - ele = ln->value; - elesize = (ele->encoding == REDIS_ENCODING_RAW) ? - (sizeof(*o)+sdslen(ele->ptr)) : sizeof(*o); - asize += (sizeof(listNode)+elesize)*listLength(l); - } - } - break; - case REDIS_SET: - case REDIS_ZSET: - z = (o->type == REDIS_ZSET); - d = z ? ((zset*)o->ptr)->dict : o->ptr; - - if (!z && o->encoding == REDIS_ENCODING_INTSET) { - intset *is = o->ptr; - asize = sizeof(*is)+is->encoding*is->length; - } else { - asize = sizeof(dict)+(sizeof(struct dictEntry*)*dictSlots(d)); - if (z) asize += sizeof(zset)-sizeof(dict); - if (dictSize(d)) { - de = dictGetRandomKey(d); - ele = dictGetEntryKey(de); - elesize = (ele->encoding == REDIS_ENCODING_RAW) ? - (sizeof(*o)+sdslen(ele->ptr)) : sizeof(*o); - asize += (sizeof(struct dictEntry)+elesize)*dictSize(d); - if (z) asize += sizeof(zskiplistNode)*dictSize(d); - } - } - break; - case REDIS_HASH: - if (o->encoding == REDIS_ENCODING_ZIPMAP) { - unsigned char *p = zipmapRewind((unsigned char*)o->ptr); - unsigned int len = zipmapLen((unsigned char*)o->ptr); - unsigned int klen, vlen; - unsigned char *key, *val; - - if ((p = zipmapNext(p,&key,&klen,&val,&vlen)) == NULL) { - klen = 0; - vlen = 0; - } - asize = len*(klen+vlen+3); - } else if (o->encoding == REDIS_ENCODING_HT) { - d = o->ptr; - asize = sizeof(dict)+(sizeof(struct dictEntry*)*dictSlots(d)); - if (dictSize(d)) { - de = dictGetRandomKey(d); - ele = dictGetEntryKey(de); - elesize = (ele->encoding == REDIS_ENCODING_RAW) ? - (sizeof(*o)+sdslen(ele->ptr)) : sizeof(*o); - ele = dictGetEntryVal(de); - elesize = (ele->encoding == REDIS_ENCODING_RAW) ? - (sizeof(*o)+sdslen(ele->ptr)) : sizeof(*o); - asize += (sizeof(struct dictEntry)+elesize)*dictSize(d); - } - } - break; - } - return (double)minage*log(1+asize); -} - -/* Try to swap an object that's a good candidate for swapping. - * Returns REDIS_OK if the object was swapped, REDIS_ERR if it's not possible - * to swap any object at all. - * - * If 'usethreaded' is true, Redis will try to swap the object in background - * using I/O threads. */ -int vmSwapOneObject(int usethreads) { - int j, i; - struct dictEntry *best = NULL; - double best_swappability = 0; - redisDb *best_db = NULL; - robj *val; - sds key; - - for (j = 0; j < server.dbnum; j++) { - redisDb *db = server.db+j; - /* 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++) { - dictEntry *de; - double swappability; - - if (maxtries) maxtries--; - de = dictGetRandomKey(db->dict); - val = dictGetEntryVal(de); - /* Only swap objects that are currently in memory. - * - * Also don't swap shared objects: not a good idea in general and - * 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 - * control other keys without adding additional mutex. */ - if (val->storage != REDIS_VM_MEMORY || val->refcount != 1) { - if (maxtries) i--; /* don't count this try */ - continue; - } - swappability = computeObjectSwappability(val); - if (!best || swappability > best_swappability) { - best = de; - best_swappability = swappability; - best_db = db; - } - } - } - if (best == NULL) return REDIS_ERR; - key = dictGetEntryKey(best); - val = dictGetEntryVal(best); - - redisLog(REDIS_DEBUG,"Key with best swappability: %s, %f", - key, best_swappability); - - /* Swap it */ - if (usethreads) { - robj *keyobj = createStringObject(key,sdslen(key)); - vmSwapObjectThreaded(keyobj,val,best_db); - decrRefCount(keyobj); - return REDIS_OK; - } else { - vmpointer *vp; - - if ((vp = vmSwapObjectBlocking(val)) != NULL) { - dictGetEntryVal(best) = vp; - return REDIS_OK; - } else { - return REDIS_ERR; - } - } -} - -int vmSwapOneObjectBlocking() { - return vmSwapOneObject(0); -} - -int vmSwapOneObjectThreaded() { - return vmSwapOneObject(1); -} - -/* Return true if it's safe to swap out objects in a given moment. - * Basically we don't want to swap objects out while there is a BGSAVE - * or a BGAEOREWRITE running in backgroud. */ -int vmCanSwapOut(void) { - return (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1); -} - -/* =================== Virtual Memory - Threaded I/O ======================= */ - -void freeIOJob(iojob *j) { - if ((j->type == REDIS_IOJOB_PREPARE_SWAP || - j->type == REDIS_IOJOB_DO_SWAP || - 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->key); - zfree(j); -} - -/* Every time a thread finished a Job, it writes a byte into the write side - * of an unix pipe in order to "awake" the main thread, and this function - * is called. - * - * Note that this is called both by the event loop, when a I/O thread - * sends a byte in the notification pipe, and is also directly called from - * waitEmptyIOJobsQueue(). - * - * In the latter case we don't want to swap more, so we use the - * "privdata" argument setting it to a not NULL value to signal this - * condition. */ -void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata, - int mask) -{ - char buf[1]; - int retval, processed = 0, toprocess = -1, trytoswap = 1; - REDIS_NOTUSED(el); - REDIS_NOTUSED(mask); - REDIS_NOTUSED(privdata); - - if (privdata != NULL) trytoswap = 0; /* check the comments above... */ - - /* For every byte we read in the read side of the pipe, there is one - * I/O job completed to process. */ - while((retval = read(fd,buf,1)) == 1) { - iojob *j; - listNode *ln; - struct dictEntry *de; - - redisLog(REDIS_DEBUG,"Processing I/O completed job"); - - /* Get the processed element (the oldest one) */ - lockThreadedIO(); - redisAssert(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); - unlockThreadedIO(); - /* If this job is marked as canceled, just ignore it */ - if (j->canceled) { - freeIOJob(j); - continue; - } - /* Post process it in the main thread, as there are things we - * can do just here to avoid race conditions and/or invasive locks */ - redisLog(REDIS_DEBUG,"COMPLETED Job type: %d, ID %p, key: %s", j->type, (void*)j->id, (unsigned char*)j->key->ptr); - de = dictFind(j->db->dict,j->key->ptr); - redisAssert(de != NULL); - if (j->type == REDIS_IOJOB_LOAD) { - redisDb *db; - vmpointer *vp = dictGetEntryVal(de); - - /* Key loaded, bring it at home */ - vmMarkPagesFree(vp->page,vp->usedpages); - redisLog(REDIS_DEBUG, "VM: object %s loaded from disk (threaded)", - (unsigned char*) j->key->ptr); - server.vm_stats_swapped_objects--; - server.vm_stats_swapins++; - dictGetEntryVal(de) = j->val; - incrRefCount(j->val); - db = j->db; - /* Handle clients waiting for this key to be loaded. */ - handleClientsBlockedOnSwappedKey(db,j->key); - freeIOJob(j); - zfree(vp); - } 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 (!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. */ - j->val->storage = REDIS_VM_MEMORY; /* undo operation */ - freeIOJob(j); - } 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); - unlockThreadedIO(); - } - } else if (j->type == REDIS_IOJOB_DO_SWAP) { - vmpointer *vp; - - /* Key swapped. We can finally free some memory. */ - if (j->val->storage != REDIS_VM_SWAPPING) { - vmpointer *vp = (vmpointer*) j->id; - printf("storage: %d\n",vp->storage); - printf("key->name: %s\n",(char*)j->key->ptr); - printf("val: %p\n",(void*)j->val); - printf("val->type: %d\n",j->val->type); - printf("val->ptr: %s\n",(char*)j->val->ptr); - } - redisAssert(j->val->storage == REDIS_VM_SWAPPING); - vp = createVmPointer(j->val->type); - vp->page = j->page; - vp->usedpages = j->pages; - dictGetEntryVal(de) = vp; - /* Fix the storage otherwise decrRefCount will attempt to - * remove the associated I/O job */ - j->val->storage = REDIS_VM_MEMORY; - decrRefCount(j->val); - redisLog(REDIS_DEBUG, - "VM: object %s swapped out at %lld (%lld pages) (threaded)", - (unsigned char*) j->key->ptr, - (unsigned long long) j->page, (unsigned long long) j->pages); - server.vm_stats_swapped_objects++; - server.vm_stats_swapouts++; - freeIOJob(j); - /* Put a few more swap requests in queue if we are still - * out of memory */ - if (trytoswap && vmCanSwapOut() && - zmalloc_used_memory() > server.vm_max_memory) - { - int more = 1; - while(more) { - lockThreadedIO(); - more = listLength(server.io_newjobs) < - (unsigned) server.vm_max_threads; - unlockThreadedIO(); - /* Don't waste CPU time if swappable objects are rare. */ - if (vmSwapOneObjectThreaded() == REDIS_ERR) { - trytoswap = 0; - break; - } - } - } - } - processed++; - if (processed == toprocess) return; - } - if (retval < 0 && errno != EAGAIN) { - redisLog(REDIS_WARNING, - "WARNING: read(2) error in vmThreadedIOCompletedJob() %s", - strerror(errno)); - } -} - -void lockThreadedIO(void) { - pthread_mutex_lock(&server.io_mutex); -} - -void unlockThreadedIO(void) { - pthread_mutex_unlock(&server.io_mutex); -} - -/* Remove the specified object from the threaded I/O queue if still not - * processed, otherwise make sure to flag it as canceled. */ -void vmCancelThreadedIOJob(robj *o) { - list *lists[3] = { - server.io_newjobs, /* 0 */ - server.io_processing, /* 1 */ - server.io_processed /* 2 */ - }; - int i; - - redisAssert(o->storage == REDIS_VM_LOADING || o->storage == REDIS_VM_SWAPPING); -again: - lockThreadedIO(); - /* Search for a matching object in one of the queues */ - for (i = 0; i < 3; i++) { - listNode *ln; - listIter li; - - listRewind(lists[i],&li); - while ((ln = listNext(&li)) != NULL) { - iojob *job = ln->value; - - if (job->canceled) continue; /* Skip this, already canceled. */ - if (job->id == o) { - redisLog(REDIS_DEBUG,"*** CANCELED %p (key %s) (type %d) (LIST ID %d)\n", - (void*)job, (char*)job->key->ptr, job->type, i); - /* Mark the pages as free since the swap didn't happened - * or happened but is now discarded. */ - if (i != 1 && job->type == REDIS_IOJOB_DO_SWAP) - 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 - * is to remove it from the queue at all */ - freeIOJob(job); - 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) - o->storage = REDIS_VM_MEMORY; - unlockThreadedIO(); - redisLog(REDIS_DEBUG,"*** DONE"); - return; - } - } - } - unlockThreadedIO(); - printf("Not found: %p\n", (void*)o); - redisAssert(1 != 1); /* We should never reach this */ -} - -void *IOThreadEntryPoint(void *arg) { - iojob *j; - listNode *ln; - REDIS_NOTUSED(arg); - - pthread_detach(pthread_self()); - while(1) { - /* Get a new job to process */ - lockThreadedIO(); - if (listLength(server.io_newjobs) == 0) { - /* No new jobs in queue, exit. */ - redisLog(REDIS_DEBUG,"Thread %ld exiting, nothing to do", - (long) pthread_self()); - server.io_active_threads--; - unlockThreadedIO(); - return NULL; - } - ln = listFirst(server.io_newjobs); - j = ln->value; - listDelNode(server.io_newjobs,ln); - /* Add the job in the processing queue */ - j->thread = pthread_self(); - listAddNodeTail(server.io_processing,j); - ln = listLast(server.io_processing); /* We use ln later to remove it */ - unlockThreadedIO(); - redisLog(REDIS_DEBUG,"Thread %ld got a new job (type %d): %p about key '%s'", - (long) pthread_self(), j->type, (void*)j, (char*)j->key->ptr); - - /* Process the Job */ - if (j->type == REDIS_IOJOB_LOAD) { - vmpointer *vp = (vmpointer*)j->id; - j->val = vmReadObjectFromSwap(j->page,vp->vtype); - } else if (j->type == REDIS_IOJOB_PREPARE_SWAP) { - j->pages = rdbSavedObjectPages(j->val); - } else if (j->type == REDIS_IOJOB_DO_SWAP) { - if (vmWriteObjectOnSwap(j->val,j->page) == REDIS_ERR) - j->canceled = 1; - } - - /* Done: insert the job into the processed queue */ - redisLog(REDIS_DEBUG,"Thread %ld completed the job: %p (key %s)", - (long) pthread_self(), (void*)j, (char*)j->key->ptr); - lockThreadedIO(); - listDelNode(server.io_processing,ln); - listAddNodeTail(server.io_processed,j); - unlockThreadedIO(); - - /* Signal the main thread there is new stuff to process */ - redisAssert(write(server.io_ready_pipe_write,"x",1) == 1); - } - return NULL; /* never reached */ -} - -void spawnIOThread(void) { - pthread_t thread; - sigset_t mask, omask; - int err; - - sigemptyset(&mask); - sigaddset(&mask,SIGCHLD); - sigaddset(&mask,SIGHUP); - sigaddset(&mask,SIGPIPE); - pthread_sigmask(SIG_SETMASK, &mask, &omask); - while ((err = pthread_create(&thread,&server.io_threads_attr,IOThreadEntryPoint,NULL)) != 0) { - redisLog(REDIS_WARNING,"Unable to spawn an I/O thread: %s", - strerror(err)); - usleep(1000000); - } - 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. */ -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, - (void*)0xdeadbeef,0); - usleep(1000); /* 1 millisecond */ - } else { - usleep(10000); /* 10 milliseconds */ - } - } -} - -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 */ -void queueIOJob(iojob *j) { - redisLog(REDIS_DEBUG,"Queued IO Job %p type %d about key '%s'\n", - (void*)j, j->type, (char*)j->key->ptr); - listAddNodeTail(server.io_newjobs,j); - if (server.io_active_threads < server.vm_max_threads) - spawnIOThread(); -} - -int vmSwapObjectThreaded(robj *key, robj *val, redisDb *db) { - iojob *j; - - j = zmalloc(sizeof(*j)); - j->type = REDIS_IOJOB_PREPARE_SWAP; - j->db = db; - j->key = key; - incrRefCount(key); - j->id = j->val = val; - incrRefCount(val); - j->canceled = 0; - j->thread = (pthread_t) -1; - val->storage = REDIS_VM_SWAPPING; - - lockThreadedIO(); - queueIOJob(j); - unlockThreadedIO(); - 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. */ -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->ptr); - if (de == NULL) return 0; - o = dictGetEntryVal(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); - redisAssert(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; - vmpointer *vp = (vmpointer*)o; - - o->storage = REDIS_VM_LOADING; - j = zmalloc(sizeof(*j)); - j->type = REDIS_IOJOB_LOAD; - j->db = c->db; - j->id = (robj*)vp; - j->key = key; - incrRefCount(key); - j->page = vp->page; - j->val = NULL; - j->canceled = 0; - j->thread = (pthread_t) -1; - lockThreadedIO(); - queueIOJob(j); - unlockThreadedIO(); - } - return 1; -} - -/* Preload keys for any command with first, last and step values for - * the command keys prototype, as defined in the command table. */ -void waitForMultipleSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) { - int j, last; - if (cmd->vm_firstkey == 0) return; - last = cmd->vm_lastkey; - if (last < 0) last = argc+last; - for (j = cmd->vm_firstkey; j <= last; j += cmd->vm_keystep) { - redisAssert(j < argc); - waitForSwappedKey(c,argv[j]); - } -} - -/* Preload keys needed for the ZUNIONSTORE and ZINTERSTORE commands. - * Note that the number of keys to preload is user-defined, so we need to - * apply a sanity check against argc. */ -void zunionInterBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) { - int i, num; - REDIS_NOTUSED(cmd); - - num = atoi(argv[2]->ptr); - if (num > (argc-3)) return; - for (i = 0; i < num; i++) { - waitForSwappedKey(c,argv[3+i]); - } -} - -/* Preload keys needed to execute the entire MULTI/EXEC block. - * - * This function is called by blockClientOnSwappedKeys when EXEC is issued, - * and will block the client when any command requires a swapped out value. */ -void execBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) { - int i, margc; - struct redisCommand *mcmd; - robj **margv; - REDIS_NOTUSED(cmd); - REDIS_NOTUSED(argc); - REDIS_NOTUSED(argv); - - if (!(c->flags & REDIS_MULTI)) return; - for (i = 0; i < c->mstate.count; i++) { - mcmd = c->mstate.commands[i].cmd; - margc = c->mstate.commands[i].argc; - margv = c->mstate.commands[i].argv; - - if (mcmd->vm_preload_proc != NULL) { - mcmd->vm_preload_proc(c,mcmd,margc,margv); - } else { - waitForMultipleSwappedKeys(c,mcmd,margc,margv); - } - } -} - -/* 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. */ -int blockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd) { - if (cmd->vm_preload_proc != NULL) { - cmd->vm_preload_proc(c,cmd,c->argc,c->argv); - } else { - waitForMultipleSwappedKeys(c,cmd,c->argc,c->argv); - } - - /* 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). */ -int dontWaitForSwappedKey(redisClient *c, robj *key) { - list *l; - listNode *ln; - listIter li; - struct dictEntry *de; - - /* The key object might be destroyed when deleted from the c->io_keys - * list (and the "key" argument is physically the same object as the - * object inside the list), so we need to protect it. */ - incrRefCount(key); - - /* Remove the key from the list of keys this client is waiting for. */ - listRewind(c->io_keys,&li); - while ((ln = listNext(&li)) != NULL) { - if (equalStringObjects(ln->value,key)) { - listDelNode(c->io_keys,ln); - break; - } - } - redisAssert(ln != NULL); - - /* Remove the client form the key => waiting clients map. */ - de = dictFind(c->db->io_keys,key); - redisAssert(de != NULL); - l = dictGetEntryVal(de); - ln = listSearchKey(l,c); - redisAssert(ln != NULL); - listDelNode(l,ln); - if (listLength(l) == 0) - dictDelete(c->db->io_keys,key); - - decrRefCount(key); - 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. */ -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); - } - } -} diff --git a/tests/assets/default.conf b/tests/assets/default.conf index 15d70ffb..04233284 100644 --- a/tests/assets/default.conf +++ b/tests/assets/default.conf @@ -86,7 +86,7 @@ dbfilename dump.rdb # Also the Append Only File will be created inside this directory. # # Note that you must specify a directory here, not a file name. -dir ./test/tmp +dir /tmp ################################# REPLICATION ################################# @@ -206,62 +206,10 @@ appendfsync everysec # To enable VM just set 'vm-enabled' to yes, and set the following three # VM parameters accordingly to your needs. -vm-enabled no -# vm-enabled yes - -# This is the path of the Redis swap file. As you can guess, swap files -# can't be shared by different Redis instances, so make sure to use a swap -# file for every redis process you are running. Redis will complain if the -# swap file is already in use. -# -# The best kind of storage for the Redis swap file (that's accessed at random) -# is a Solid State Disk (SSD). -# -# *** WARNING *** if you are using a shared hosting the default of putting -# the swap file under /tmp is not secure. Create a dir with access granted -# only to Redis user and configure Redis to create the swap file there. -vm-swap-file redis.swap - -# vm-max-memory configures the VM to use at max the specified amount of -# RAM. Everything that deos not fit will be swapped on disk *if* possible, that -# is, if there is still enough contiguous space in the swap file. -# -# With vm-max-memory 0 the system will swap everything it can. Not a good -# default, just specify the max amount of RAM you can in bytes, but it's -# better to leave some margin. For instance specify an amount of RAM -# that's more or less between 60 and 80% of your free RAM. -vm-max-memory 0 - -# Redis swap files is split into pages. An object can be saved using multiple -# contiguous pages, but pages can't be shared between different objects. -# So if your page is too big, small objects swapped out on disk will waste -# a lot of space. If you page is too small, there is less space in the swap -# file (assuming you configured the same number of total swap file pages). -# -# If you use a lot of small objects, use a page size of 64 or 32 bytes. -# If you use a lot of big objects, use a bigger page size. -# If unsure, use the default :) -vm-page-size 32 - -# Number of total memory pages in the swap file. -# Given that the page table (a bitmap of free/used pages) is taken in memory, -# every 8 pages on disk will consume 1 byte of RAM. -# -# The total swap size is vm-page-size * vm-pages -# -# 32M swap should be enough for testing. -vm-pages 1048576 - -# Max number of VM I/O threads running at the same time. -# This threads are used to read/write data from/to swap file, since they -# also encode and decode objects from disk to memory or the reverse, a bigger -# number of threads can help with big objects even if they can't help with -# I/O itself as the physical device may not be able to couple with many -# reads/writes operations at the same time. -# -# The special value of 0 turn off threaded I/O and enables the blocking -# Virtual Memory implementation. -vm-max-threads 4 +diskstore-enabled yes +diskstore-path redis.ds +cache-max-memory 0 +cache-flush-delay 0 ############################### ADVANCED CONFIG ############################### diff --git a/tests/support/server.tcl b/tests/support/server.tcl index 4f48d22d..ad711e6e 100644 --- a/tests/support/server.tcl +++ b/tests/support/server.tcl @@ -177,6 +177,9 @@ proc start_server {options {code undefined}} { if {$::valgrind} { exec valgrind --suppressions=src/valgrind.sup src/redis-server $config_file > $stdout 2> $stderr & } else { + if {$::verbose} { + puts "Logging on $stdout / $stderr" + } exec src/redis-server $config_file > $stdout 2> $stderr & } diff --git a/tests/test_helper.tcl b/tests/test_helper.tcl index 2b7a8957..8ab2f6e4 100644 --- a/tests/test_helper.tcl +++ b/tests/test_helper.tcl @@ -13,7 +13,7 @@ set ::host 127.0.0.1 set ::port 16379 set ::traceleaks 0 set ::valgrind 0 -set ::verbose 0 +set ::verbose 1 set ::denytags {} set ::allowtags {} set ::external 0; # If "1" this means, we are running against external instance