#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 flags); int processActiveIOJobs(int max); /* =================== 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); pthread_mutex_init(&server.bgsavethread_mutex,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; for (i = 0; i < 5 && dictSize(db->dict); 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) { /* Not able to free 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) { redisLog(REDIS_DEBUG,"--- Busy waiting IO to reclaim memory"); cacheScheduleIOPushJobs(REDIS_IO_ASAP); processActiveIOJobs(1); 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; } /* ==================== 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. * * If privdata == NULL the function will try to put more jobs in the queue * of IO jobs to process as more room is made. privdata is equal to NULL * when the function is called from the event loop, so we want to push * more IO jobs in the queue. Instead when the function is called by * other functions that want to create a write-barrier to avoid race * conditions we don't push new jobs in the queue. */ 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); /* 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); } else if (j->type == REDIS_IOJOB_SAVE) { cacheScheduleIODelFlag(j->db,j->key,REDIS_IO_SAVEINPROG); } freeIOJob(j); processed++; if (privdata == NULL) cacheScheduleIOPushJobs(0); 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); long long start; 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 */ redisLog(REDIS_DEBUG,"[T] wait for signal"); pthread_cond_wait(&server.io_condvar,&server.io_mutex); redisLog(REDIS_DEBUG,"[T] signal received"); continue; } start = ustime(); redisLog(REDIS_DEBUG,"[T] %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,"[T] %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,j->expire); } else { dsDel(j->db,j->key); } } /* Done: insert the job into the processed queue */ redisLog(REDIS_DEBUG,"[T] %ld completed the job: %p (key %s)", (long) pthread_self(), (void*)j, (char*)j->key->ptr); redisLog(REDIS_DEBUG,"[T] lock IO"); lockThreadedIO(); redisLog(REDIS_DEBUG,"[T] IO locked"); 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); redisLog(REDIS_DEBUG,"TIME (%c): %lld\n", j->type == REDIS_IOJOB_LOAD ? 'L' : 'S', ustime()-start); } /* 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 up to 'max' pending IO Jobs are processed by the I/O thread. * From our point of view an IO job processed means that the count of * server.io_processed must increase by one. * * If max is -1, all the pending IO jobs will be processed. * * Returns the number of IO jobs processed. * * NOTE: while this may appear like a busy loop, we are actually blocked * by IO since we continuously acquire/release the IO lock. */ int processActiveIOJobs(int max) { int processed = 0; while(max == -1 || max > 0) { int io_processed_len; redisLog(REDIS_DEBUG,"[P] lock IO"); lockThreadedIO(); redisLog(REDIS_DEBUG,"Waiting IO jobs processing: new:%d proessing:%d processed:%d",listLength(server.io_newjobs),listLength(server.io_processing),listLength(server.io_processed)); if (listLength(server.io_newjobs) == 0 && listLength(server.io_processing) == 0) { /* There is nothing more to process */ redisLog(REDIS_DEBUG,"[P] Nothing to process, unlock IO, return"); unlockThreadedIO(); break; } #if 1 /* If there are new jobs we need to signal the thread to * process the next one. FIXME: drop this if useless. */ redisLog(REDIS_DEBUG,"[P] waitEmptyIOJobsQueue: new %d, processing %d, processed %d", listLength(server.io_newjobs), listLength(server.io_processing), listLength(server.io_processed)); if (listLength(server.io_newjobs)) { redisLog(REDIS_DEBUG,"[P] There are new jobs, signal"); pthread_cond_signal(&server.io_condvar); } #endif /* Check if we can process some finished job */ io_processed_len = listLength(server.io_processed); redisLog(REDIS_DEBUG,"[P] Unblock IO"); unlockThreadedIO(); redisLog(REDIS_DEBUG,"[P] Wait"); usleep(10000); if (io_processed_len) { vmThreadedIOCompletedJob(NULL,server.io_ready_pipe_read, (void*)0xdeadbeef,0); processed++; if (max != -1) max--; } } return processed; } void waitEmptyIOJobsQueue(void) { processActiveIOJobs(-1); } /* Process up to 'max' IO Jobs already completed by threads but still waiting * processing from the main thread. * * If max == -1 all the pending jobs are processed. * * The number of processed jobs is returned. */ int processPendingIOJobs(int max) { int processed = 0; while(max == -1 || max > 0) { int io_processed_len; lockThreadedIO(); io_processed_len = listLength(server.io_processed); unlockThreadedIO(); if (io_processed_len == 0) break; vmThreadedIOCompletedJob(NULL,server.io_ready_pipe_read, (void*)0xdeadbeef,0); if (max != -1) max--; processed++; } return processed; } void processAllPendingIOJobs(void) { processPendingIOJobs(-1); } /* 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); } /* Consume all the IO scheduled operations, and all the thread IO jobs * so that eventually the state of diskstore is a point-in-time snapshot. * * This is useful when we need to BGSAVE with diskstore enabled. */ void cacheForcePointInTime(void) { redisLog(REDIS_NOTICE,"Diskstore: synching on disk to reach point-in-time state."); while (listLength(server.cache_io_queue) != 0) { cacheScheduleIOPushJobs(REDIS_IO_ASAP); processActiveIOJobs(1); } waitEmptyIOJobsQueue(); processAllPendingIOJobs(); } void cacheCreateIOJob(int type, redisDb *db, robj *key, robj *val, time_t expire) { iojob *j; j = zmalloc(sizeof(*j)); j->type = type; j->db = db; j->key = key; incrRefCount(key); j->val = val; if (val) incrRefCount(val); j->expire = expire; 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(REDIS_IO_ONLYLOADS); } 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 flags include REDIS_IO_ONLYLOADS only load jobs are processed:this is * useful since it's safe 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. * * The REDIS_IO_ASAP flag tells the function to don't wait for the IO job * scheduled completion time, but just do the operation ASAP. This is useful * when we need to reclaim memory from the IO queue. */ #define MAX_IO_JOBS_QUEUE 10 int cacheScheduleIOPushJobs(int flags) { time_t now = time(NULL); listNode *ln; int jobs, topush = 0, pushed = 0; /* Don't push new jobs if there is a threaded BGSAVE in progress. */ if (server.bgsavethread != (pthread_t) -1) return 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; struct dictEntry *de; robj *val; if (!topush) break; topush--; if (op->type != REDIS_IO_LOAD && flags & REDIS_IO_ONLYLOADS) break; /* Don't execute SAVE before the scheduled time for completion */ if (op->type == REDIS_IO_SAVE && !(flags & REDIS_IO_ASAP) && (now - op->ctime) < server.cache_flush_delay) break; /* Don't add a SAVE job in the IO thread 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 if there * are other operations, so we can try to process the next one. * 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) { cacheCreateIOJob(REDIS_IOJOB_LOAD,op->db,op->key,NULL,0); } else { time_t expire = -1; /* 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); expire = getExpire(op->db,op->key); } else { /* Setting the value to NULL tells the IO thread to delete * the key on disk. */ val = NULL; } cacheCreateIOJob(REDIS_IOJOB_SAVE,op->db,op->key,val,expire); } /* 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++; } 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) { int flags = cacheScheduleIOGetFlags(c->db,key); /* It is possible that even if there are no clients waiting for * a load operation, still we have a load operation in progress. * For instance think to a client performing a GET and then * closing the connection */ if ((flags & (REDIS_IO_LOAD|REDIS_IO_LOADINPROG)) == 0) cacheScheduleIO(c->db,key,REDIS_IO_LOAD); } return 1; } /* Is this client attempting to run a command against swapped keys? * If so, block it ASAP, load the keys in background, then resume it. * * The important idea about this function is that it can fail! If keys will * still be swapped when the client is resumed, this key lookups will * just block loading keys from disk. In practical terms this should only * happen with SORT BY command or if there is a bug in this function. * * Return 1 if the client is marked as blocked, 0 if the client can * continue as the keys it is going to access appear to be in memory. */ int blockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd) { int *keyindex, numkeys, j, i; /* EXEC is a special case, we need to preload all the commands * queued into the transaction */ if (cmd->proc == execCommand) { struct redisCommand *mcmd; robj **margv; int margc; if (!(c->flags & REDIS_MULTI)) return 0; 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; keyindex = getKeysFromCommand(mcmd,margv,margc,&numkeys, REDIS_GETKEYS_PRELOAD); for (j = 0; j < numkeys; j++) { redisLog(REDIS_DEBUG,"Preloading %s", (char*)margv[keyindex[j]]->ptr); waitForSwappedKey(c,margv[keyindex[j]]); } getKeysFreeResult(keyindex); } } else { keyindex = getKeysFromCommand(cmd,c->argv,c->argc,&numkeys, REDIS_GETKEYS_PRELOAD); for (j = 0; j < numkeys; j++) { redisLog(REDIS_DEBUG,"Preloading %s", (char*)c->argv[keyindex[j]]->ptr); waitForSwappedKey(c,c->argv[keyindex[j]]); } getKeysFreeResult(keyindex); } /* 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); } } }