8 /* dscache.c - Disk store cache for disk store backend.
10 * When Redis is configured for using disk as backend instead of memory, the
11 * memory is used as a cache, so that recently accessed keys are taken in
12 * memory for fast read and write operations.
14 * Modified keys are marked to be flushed on disk, and will be flushed
15 * as long as the maxium configured flush time elapsed.
17 * This file implements the whole caching subsystem and contains further
22 * - The WATCH helper will be used to signal the cache system
23 * we need to flush a given key/dbid into disk, adding this key/dbid
24 * pair into a server.ds_cache_dirty linked list AND hash table (so that we
25 * don't add the same thing multiple times).
27 * - cron() checks if there are elements on this list. When there are things
28 * to flush, we create an IO Job for the I/O thread.
29 * NOTE: We disalbe object sharing when server.ds_enabled == 1 so objects
30 * that are referenced an IO job for flushing on disk are marked as
31 * o->storage == REDIS_DS_SAVING.
33 * - This is what we do on key lookup:
34 * 1) The key already exists in memory. object->storage == REDIS_DS_MEMORY
35 * or it is object->storage == REDIS_DS_DIRTY:
36 * We don't do nothing special, lookup, return value object pointer.
37 * 2) The key is in memory but object->storage == REDIS_DS_SAVING.
38 * When this happens we block waiting for the I/O thread to process
39 * this object. Then continue.
40 * 3) The key is not in memory. We block to load the key from disk.
41 * Of course the key may not be present at all on the disk store as well,
42 * in such case we just detect this condition and continue, returning
45 * - Preloading of needed keys:
46 * 1) As it was done with VM, also with this new system we try preloading
47 * keys a client is going to use. We block the client, load keys
48 * using the I/O thread, unblock the client. Same code as VM more or less.
50 * - Reclaiming memory.
51 * In cron() we detect our memory limit was reached. What we
52 * do is deleting keys that are REDIS_DS_MEMORY, using LRU.
54 * If this is not enough to return again under the memory limits we also
55 * start to flush keys that need to be synched on disk synchronously,
56 * removing it from the memory. We do this blocking as memory limit is a
57 * much "harder" barrirer in the new design.
59 * - IO thread operations are no longer stopped for sync loading/saving of
60 * things. When a key is found to be in the process of being saved
61 * we simply wait for the IO thread to end its work.
63 * Otherwise if there is to load a key without any IO thread operation
64 * just started it is blocking-loaded in the lookup function.
66 * - What happens when an object is destroyed?
68 * If o->storage == REDIS_DS_MEMORY then we simply destory the object.
69 * If o->storage == REDIS_DS_DIRTY we can still remove the object. It had
70 * changes not flushed on disk, but is being removed so
72 * if o->storage == REDIS_DS_SAVING then the object is being saved so
73 * it is impossible that its refcount == 1, must be at
74 * least two. When the object is saved the storage will
75 * be set back to DS_MEMORY.
77 * - What happens when keys are deleted?
79 * We simply schedule a key flush operation as usually, but when the
80 * IO thread will be created the object pointer will be set to NULL
81 * so the IO thread will know that the work to do is to delete the key
82 * from the disk store.
84 * - What happens with MULTI/EXEC?
88 * - If dsSet() fails on the write thread log the error and reschedule the
92 /* Virtual Memory is composed mainly of two subsystems:
93 * - Blocking Virutal Memory
94 * - Threaded Virtual Memory I/O
95 * The two parts are not fully decoupled, but functions are split among two
96 * different sections of the source code (delimited by comments) in order to
97 * make more clear what functionality is about the blocking VM and what about
98 * the threaded (not blocking) VM.
102 * Redis VM is a blocking VM (one that blocks reading swapped values from
103 * disk into memory when a value swapped out is needed in memory) that is made
104 * unblocking by trying to examine the command argument vector in order to
105 * load in background values that will likely be needed in order to exec
106 * the command. The command is executed only once all the relevant keys
107 * are loaded into memory.
109 * This basically is almost as simple of a blocking VM, but almost as parallel
110 * as a fully non-blocking VM.
113 void spawnIOThread(void);
115 /* =================== Virtual Memory - Blocking Side ====================== */
121 zmalloc_enable_thread_safeness(); /* we need thread safe zmalloc() */
123 redisLog(REDIS_NOTICE
,"Opening Disk Store: %s", server
.ds_path
);
124 /* Open Disk Store */
125 if (dsOpen() != REDIS_OK
) {
126 redisLog(REDIS_WARNING
,"Fatal error opening disk store. Exiting.");
130 /* Initialize threaded I/O for Object Cache */
131 server
.io_newjobs
= listCreate();
132 server
.io_processing
= listCreate();
133 server
.io_processed
= listCreate();
134 server
.io_ready_clients
= listCreate();
135 pthread_mutex_init(&server
.io_mutex
,NULL
);
136 server
.io_active_threads
= 0;
137 if (pipe(pipefds
) == -1) {
138 redisLog(REDIS_WARNING
,"Unable to intialized DS: pipe(2): %s. Exiting."
142 server
.io_ready_pipe_read
= pipefds
[0];
143 server
.io_ready_pipe_write
= pipefds
[1];
144 redisAssert(anetNonBlock(NULL
,server
.io_ready_pipe_read
) != ANET_ERR
);
145 /* LZF requires a lot of stack */
146 pthread_attr_init(&server
.io_threads_attr
);
147 pthread_attr_getstacksize(&server
.io_threads_attr
, &stacksize
);
149 /* Solaris may report a stacksize of 0, let's set it to 1 otherwise
150 * multiplying it by 2 in the while loop later will not really help ;) */
151 if (!stacksize
) stacksize
= 1;
153 while (stacksize
< REDIS_THREAD_STACK_SIZE
) stacksize
*= 2;
154 pthread_attr_setstacksize(&server
.io_threads_attr
, stacksize
);
155 /* Listen for events in the threaded I/O pipe */
156 if (aeCreateFileEvent(server
.el
, server
.io_ready_pipe_read
, AE_READABLE
,
157 vmThreadedIOCompletedJob
, NULL
) == AE_ERR
)
158 oom("creating file event");
160 /* Spawn our I/O thread */
164 /* Compute how good candidate the specified object is for eviction.
165 * An higher number means a better candidate. */
166 double computeObjectSwappability(robj
*o
) {
167 /* actual age can be >= minage, but not < minage. As we use wrapping
168 * 21 bit clocks with minutes resolution for the LRU. */
169 return (double) estimateObjectIdleTime(o
);
172 /* Try to free one entry from the diskstore object cache */
173 int cacheFreeOneEntry(void) {
175 struct dictEntry
*best
= NULL
;
176 double best_swappability
= 0;
177 redisDb
*best_db
= NULL
;
181 for (j
= 0; j
< server
.dbnum
; j
++) {
182 redisDb
*db
= server
.db
+j
;
183 /* Why maxtries is set to 100?
184 * Because this way (usually) we'll find 1 object even if just 1% - 2%
185 * are swappable objects */
188 if (dictSize(db
->dict
) == 0) continue;
189 for (i
= 0; i
< 5; i
++) {
193 if (maxtries
) maxtries
--;
194 de
= dictGetRandomKey(db
->dict
);
195 val
= dictGetEntryVal(de
);
196 /* Only swap objects that are currently in memory.
198 * Also don't swap shared objects: not a good idea in general and
199 * we need to ensure that the main thread does not touch the
200 * object while the I/O thread is using it, but we can't
201 * control other keys without adding additional mutex. */
202 if (val
->storage
!= REDIS_DS_MEMORY
) {
203 if (maxtries
) i
--; /* don't count this try */
206 swappability
= computeObjectSwappability(val
);
207 if (!best
|| swappability
> best_swappability
) {
209 best_swappability
= swappability
;
215 /* FIXME: If there are objects marked as DS_DIRTY or DS_SAVING
216 * let's wait for this objects to be clear and retry...
218 * Object cache vm limit is considered an hard limit. */
221 key
= dictGetEntryKey(best
);
222 val
= dictGetEntryVal(best
);
224 redisLog(REDIS_DEBUG
,"Key selected for cache eviction: %s swappability:%f",
225 key
, best_swappability
);
227 /* Delete this key from memory */
229 robj
*kobj
= createStringObject(key
,sdslen(key
));
230 dbDelete(best_db
,kobj
);
236 /* Return true if it's safe to swap out objects in a given moment.
237 * Basically we don't want to swap objects out while there is a BGSAVE
238 * or a BGAEOREWRITE running in backgroud. */
239 int dsCanTouchDiskStore(void) {
240 return (server
.bgsavechildpid
== -1 && server
.bgrewritechildpid
== -1);
243 /* =================== Virtual Memory - Threaded I/O ======================= */
245 void freeIOJob(iojob
*j
) {
246 decrRefCount(j
->key
);
247 /* j->val can be NULL if the job is about deleting the key from disk. */
248 if (j
->val
) decrRefCount(j
->val
);
252 /* Every time a thread finished a Job, it writes a byte into the write side
253 * of an unix pipe in order to "awake" the main thread, and this function
255 void vmThreadedIOCompletedJob(aeEventLoop
*el
, int fd
, void *privdata
,
259 int retval
, processed
= 0, toprocess
= -1;
262 REDIS_NOTUSED(privdata
);
264 /* For every byte we read in the read side of the pipe, there is one
265 * I/O job completed to process. */
266 while((retval
= read(fd
,buf
,1)) == 1) {
270 redisLog(REDIS_DEBUG
,"Processing I/O completed job");
272 /* Get the processed element (the oldest one) */
274 redisAssert(listLength(server
.io_processed
) != 0);
275 if (toprocess
== -1) {
276 toprocess
= (listLength(server
.io_processed
)*REDIS_MAX_COMPLETED_JOBS_PROCESSED
)/100;
277 if (toprocess
<= 0) toprocess
= 1;
279 ln
= listFirst(server
.io_processed
);
281 listDelNode(server
.io_processed
,ln
);
284 /* Post process it in the main thread, as there are things we
285 * can do just here to avoid race conditions and/or invasive locks */
286 redisLog(REDIS_DEBUG
,"COMPLETED Job type %s, key: %s",
287 (j
->type
== REDIS_IOJOB_LOAD
) ? "load" : "save",
288 (unsigned char*)j
->key
->ptr
);
289 if (j
->type
== REDIS_IOJOB_LOAD
) {
290 /* Create the key-value pair in the in-memory database */
291 if (j
->val
!= NULL
) {
292 dbAdd(j
->db
,j
->key
,j
->val
);
293 incrRefCount(j
->val
);
294 setExpire(j
->db
,j
->key
,j
->expire
);
296 /* The key does not exist. Create a negative cache entry
298 /* FIXME: add this entry into the negative cache */
300 /* Handle clients waiting for this key to be loaded. */
301 handleClientsBlockedOnSwappedKey(j
->db
,j
->key
);
303 } else if (j
->type
== REDIS_IOJOB_SAVE
) {
304 redisAssert(j
->val
->storage
== REDIS_DS_SAVING
);
305 j
->val
->storage
= REDIS_DS_MEMORY
;
309 if (processed
== toprocess
) return;
311 if (retval
< 0 && errno
!= EAGAIN
) {
312 redisLog(REDIS_WARNING
,
313 "WARNING: read(2) error in vmThreadedIOCompletedJob() %s",
318 void lockThreadedIO(void) {
319 pthread_mutex_lock(&server
.io_mutex
);
322 void unlockThreadedIO(void) {
323 pthread_mutex_unlock(&server
.io_mutex
);
326 void *IOThreadEntryPoint(void *arg
) {
331 pthread_detach(pthread_self());
333 /* Get a new job to process */
335 if (listLength(server
.io_newjobs
) == 0) {
336 /* No new jobs in queue, exit. */
341 ln
= listFirst(server
.io_newjobs
);
343 listDelNode(server
.io_newjobs
,ln
);
344 /* Add the job in the processing queue */
345 listAddNodeTail(server
.io_processing
,j
);
346 ln
= listLast(server
.io_processing
); /* We use ln later to remove it */
348 redisLog(REDIS_DEBUG
,"Thread %ld: new job type %s: %p about key '%s'",
349 (long) pthread_self(),
350 (j
->type
== REDIS_IOJOB_LOAD
) ? "load" : "save",
351 (void*)j
, (char*)j
->key
->ptr
);
353 /* Process the Job */
354 if (j
->type
== REDIS_IOJOB_LOAD
) {
357 j
->val
= dsGet(j
->db
,j
->key
,&expire
);
358 if (j
->val
) j
->expire
= expire
;
359 } else if (j
->type
== REDIS_IOJOB_SAVE
) {
360 redisAssert(j
->val
->storage
== REDIS_DS_SAVING
);
362 dsSet(j
->db
,j
->key
,j
->val
);
367 /* Done: insert the job into the processed queue */
368 redisLog(REDIS_DEBUG
,"Thread %ld completed the job: %p (key %s)",
369 (long) pthread_self(), (void*)j
, (char*)j
->key
->ptr
);
371 listDelNode(server
.io_processing
,ln
);
372 listAddNodeTail(server
.io_processed
,j
);
375 /* Signal the main thread there is new stuff to process */
376 redisAssert(write(server
.io_ready_pipe_write
,"x",1) == 1);
378 return NULL
; /* never reached */
381 void spawnIOThread(void) {
383 sigset_t mask
, omask
;
387 sigaddset(&mask
,SIGCHLD
);
388 sigaddset(&mask
,SIGHUP
);
389 sigaddset(&mask
,SIGPIPE
);
390 pthread_sigmask(SIG_SETMASK
, &mask
, &omask
);
391 while ((err
= pthread_create(&thread
,&server
.io_threads_attr
,IOThreadEntryPoint
,NULL
)) != 0) {
392 redisLog(REDIS_WARNING
,"Unable to spawn an I/O thread: %s",
396 pthread_sigmask(SIG_SETMASK
, &omask
, NULL
);
397 server
.io_active_threads
++;
400 /* We need to wait for the last thread to exit before we are able to
401 * fork() in order to BGSAVE or BGREWRITEAOF. */
402 void waitEmptyIOJobsQueue(void) {
404 int io_processed_len
;
407 if (listLength(server
.io_newjobs
) == 0 &&
408 listLength(server
.io_processing
) == 0 &&
409 server
.io_active_threads
== 0)
414 /* While waiting for empty jobs queue condition we post-process some
415 * finshed job, as I/O threads may be hanging trying to write against
416 * the io_ready_pipe_write FD but there are so much pending jobs that
418 io_processed_len
= listLength(server
.io_processed
);
420 if (io_processed_len
) {
421 vmThreadedIOCompletedJob(NULL
,server
.io_ready_pipe_read
,
422 (void*)0xdeadbeef,0);
423 usleep(1000); /* 1 millisecond */
425 usleep(10000); /* 10 milliseconds */
430 /* This function must be called while with threaded IO locked */
431 void queueIOJob(iojob
*j
) {
432 redisLog(REDIS_DEBUG
,"Queued IO Job %p type %d about key '%s'\n",
433 (void*)j
, j
->type
, (char*)j
->key
->ptr
);
434 listAddNodeTail(server
.io_newjobs
,j
);
435 if (server
.io_active_threads
< server
.vm_max_threads
)
439 void dsCreateIOJob(int type
, redisDb
*db
, robj
*key
, robj
*val
) {
442 j
= zmalloc(sizeof(*j
));
448 if (val
) incrRefCount(val
);
455 void cacheScheduleForFlush(redisDb
*db
, robj
*key
) {
459 de
= dictFind(db
->dict
,key
->ptr
);
461 robj
*val
= dictGetEntryVal(de
);
462 if (val
->storage
== REDIS_DS_DIRTY
)
465 val
->storage
= REDIS_DS_DIRTY
;
468 redisLog(REDIS_DEBUG
,"Scheduling key %s for saving",key
->ptr
);
469 dk
= zmalloc(sizeof(*dk
));
473 dk
->ctime
= time(NULL
);
474 listAddNodeTail(server
.cache_flush_queue
, key
);
477 void cacheCron(void) {
478 time_t now
= time(NULL
);
481 /* Sync stuff on disk */
482 while((ln
= listFirst(server
.cache_flush_queue
)) != NULL
) {
483 dirtykey
*dk
= ln
->value
;
485 if ((now
- dk
->ctime
) >= server
.cache_flush_delay
) {
486 struct dictEntry
*de
;
489 redisLog(REDIS_DEBUG
,"Creating IO Job to save key %s",dk
->key
->ptr
);
491 /* Lookup the key. We need to check if it's still here and
492 * possibly access to the value. */
493 de
= dictFind(dk
->db
->dict
,dk
->key
->ptr
);
495 val
= dictGetEntryVal(de
);
496 redisAssert(val
->storage
== REDIS_DS_DIRTY
);
497 val
->storage
= REDIS_DS_SAVING
;
499 /* Setting the value to NULL tells the IO thread to delete
500 * the key on disk. */
503 dsCreateIOJob(REDIS_IOJOB_SAVE
,dk
->db
,dk
->key
,val
);
504 listDelNode(server
.cache_flush_queue
,ln
);
506 break; /* too early */
510 /* Reclaim memory from the object cache */
511 while (server
.ds_enabled
&& zmalloc_used_memory() >
512 server
.cache_max_memory
)
514 if (cacheFreeOneEntry() == REDIS_ERR
) break;
518 /* ============ Virtual Memory - Blocking clients on missing keys =========== */
520 /* This function makes the clinet 'c' waiting for the key 'key' to be loaded.
521 * If the key is already in memory we don't need to block, regardless
522 * of the storage of the value object for this key:
524 * - If it's REDIS_DS_MEMORY we have the key in memory.
525 * - If it's REDIS_DS_DIRTY they key was modified, but still in memory.
526 * - if it's REDIS_DS_SAVING the key is being saved by an IO Job. When
527 * the client will lookup the key it will block if the key is still
528 * in this stage but it's more or less the best we can do.
530 * FIXME: we should try if it's actually better to suspend the client
531 * accessing an object that is being saved, and awake it only when
532 * the saving was completed.
534 * Otherwise if the key is not in memory, we block the client and start
535 * an IO Job to load it:
537 * the key is added to the io_keys list in the client structure, and also
538 * in the hash table mapping swapped keys to waiting clients, that is,
539 * server.io_waited_keys. */
540 int waitForSwappedKey(redisClient
*c
, robj
*key
) {
541 struct dictEntry
*de
;
544 /* Return ASAP if the key is in memory */
545 de
= dictFind(c
->db
->dict
,key
->ptr
);
546 if (de
!= NULL
) return 0;
548 /* Add the key to the list of keys this client is waiting for.
549 * This maps clients to keys they are waiting for. */
550 listAddNodeTail(c
->io_keys
,key
);
553 /* Add the client to the swapped keys => clients waiting map. */
554 de
= dictFind(c
->db
->io_keys
,key
);
558 /* For every key we take a list of clients blocked for it */
560 retval
= dictAdd(c
->db
->io_keys
,key
,l
);
562 redisAssert(retval
== DICT_OK
);
564 l
= dictGetEntryVal(de
);
566 listAddNodeTail(l
,c
);
568 /* Are we already loading the key from disk? If not create a job */
570 dsCreateIOJob(REDIS_IOJOB_LOAD
,c
->db
,key
,NULL
);
574 /* Preload keys for any command with first, last and step values for
575 * the command keys prototype, as defined in the command table. */
576 void waitForMultipleSwappedKeys(redisClient
*c
, struct redisCommand
*cmd
, int argc
, robj
**argv
) {
578 if (cmd
->vm_firstkey
== 0) return;
579 last
= cmd
->vm_lastkey
;
580 if (last
< 0) last
= argc
+last
;
581 for (j
= cmd
->vm_firstkey
; j
<= last
; j
+= cmd
->vm_keystep
) {
582 redisAssert(j
< argc
);
583 waitForSwappedKey(c
,argv
[j
]);
587 /* Preload keys needed for the ZUNIONSTORE and ZINTERSTORE commands.
588 * Note that the number of keys to preload is user-defined, so we need to
589 * apply a sanity check against argc. */
590 void zunionInterBlockClientOnSwappedKeys(redisClient
*c
, struct redisCommand
*cmd
, int argc
, robj
**argv
) {
594 num
= atoi(argv
[2]->ptr
);
595 if (num
> (argc
-3)) return;
596 for (i
= 0; i
< num
; i
++) {
597 waitForSwappedKey(c
,argv
[3+i
]);
601 /* Preload keys needed to execute the entire MULTI/EXEC block.
603 * This function is called by blockClientOnSwappedKeys when EXEC is issued,
604 * and will block the client when any command requires a swapped out value. */
605 void execBlockClientOnSwappedKeys(redisClient
*c
, struct redisCommand
*cmd
, int argc
, robj
**argv
) {
607 struct redisCommand
*mcmd
;
613 if (!(c
->flags
& REDIS_MULTI
)) return;
614 for (i
= 0; i
< c
->mstate
.count
; i
++) {
615 mcmd
= c
->mstate
.commands
[i
].cmd
;
616 margc
= c
->mstate
.commands
[i
].argc
;
617 margv
= c
->mstate
.commands
[i
].argv
;
619 if (mcmd
->vm_preload_proc
!= NULL
) {
620 mcmd
->vm_preload_proc(c
,mcmd
,margc
,margv
);
622 waitForMultipleSwappedKeys(c
,mcmd
,margc
,margv
);
627 /* Is this client attempting to run a command against swapped keys?
628 * If so, block it ASAP, load the keys in background, then resume it.
630 * The important idea about this function is that it can fail! If keys will
631 * still be swapped when the client is resumed, this key lookups will
632 * just block loading keys from disk. In practical terms this should only
633 * happen with SORT BY command or if there is a bug in this function.
635 * Return 1 if the client is marked as blocked, 0 if the client can
636 * continue as the keys it is going to access appear to be in memory. */
637 int blockClientOnSwappedKeys(redisClient
*c
, struct redisCommand
*cmd
) {
638 if (cmd
->vm_preload_proc
!= NULL
) {
639 cmd
->vm_preload_proc(c
,cmd
,c
->argc
,c
->argv
);
641 waitForMultipleSwappedKeys(c
,cmd
,c
->argc
,c
->argv
);
644 /* If the client was blocked for at least one key, mark it as blocked. */
645 if (listLength(c
->io_keys
)) {
646 c
->flags
|= REDIS_IO_WAIT
;
647 aeDeleteFileEvent(server
.el
,c
->fd
,AE_READABLE
);
648 server
.cache_blocked_clients
++;
655 /* Remove the 'key' from the list of blocked keys for a given client.
657 * The function returns 1 when there are no longer blocking keys after
658 * the current one was removed (and the client can be unblocked). */
659 int dontWaitForSwappedKey(redisClient
*c
, robj
*key
) {
663 struct dictEntry
*de
;
665 /* The key object might be destroyed when deleted from the c->io_keys
666 * list (and the "key" argument is physically the same object as the
667 * object inside the list), so we need to protect it. */
670 /* Remove the key from the list of keys this client is waiting for. */
671 listRewind(c
->io_keys
,&li
);
672 while ((ln
= listNext(&li
)) != NULL
) {
673 if (equalStringObjects(ln
->value
,key
)) {
674 listDelNode(c
->io_keys
,ln
);
678 redisAssert(ln
!= NULL
);
680 /* Remove the client form the key => waiting clients map. */
681 de
= dictFind(c
->db
->io_keys
,key
);
682 redisAssert(de
!= NULL
);
683 l
= dictGetEntryVal(de
);
684 ln
= listSearchKey(l
,c
);
685 redisAssert(ln
!= NULL
);
687 if (listLength(l
) == 0)
688 dictDelete(c
->db
->io_keys
,key
);
691 return listLength(c
->io_keys
) == 0;
694 /* Every time we now a key was loaded back in memory, we handle clients
695 * waiting for this key if any. */
696 void handleClientsBlockedOnSwappedKey(redisDb
*db
, robj
*key
) {
697 struct dictEntry
*de
;
702 de
= dictFind(db
->io_keys
,key
);
705 l
= dictGetEntryVal(de
);
707 /* Note: we can't use something like while(listLength(l)) as the list
708 * can be freed by the calling function when we remove the last element. */
711 redisClient
*c
= ln
->value
;
713 if (dontWaitForSwappedKey(c
,key
)) {
714 /* Put the client in the list of clients ready to go as we
715 * loaded all the keys about it. */
716 listAddNodeTail(server
.io_ready_clients
,c
);