]> git.saurik.com Git - redis.git/blob - src/dscache.c
6bfcac1d1f647766f9392197bf22166ae14fc136
[redis.git] / src / dscache.c
1 #include "redis.h"
2
3 #include <fcntl.h>
4 #include <pthread.h>
5 #include <math.h>
6 #include <signal.h>
7
8 /* dscache.c - Disk store cache for disk store backend.
9 *
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.
13 *
14 * Modified keys are marked to be flushed on disk, and will be flushed
15 * as long as the maxium configured flush time elapsed.
16 *
17 * This file implements the whole caching subsystem and contains further
18 * documentation. */
19
20 /* TODO:
21 *
22 * WARNING: most of the following todo items and design issues are no
23 * longer relevant with the new design. Here as a checklist to see if
24 * some old ideas still apply.
25 *
26 * - What happens when an object is destroyed?
27 *
28 * If the object is destroyed since semantically it was deleted or
29 * replaced with something new, we don't care if there was a SAVE
30 * job pending for it. Anyway when the IO JOb will be created we'll get
31 * the pointer of the current value.
32 *
33 * If the object is already a REDIS_IO_SAVEINPROG object, then it is
34 * impossible that we get a decrRefCount() that will reach refcount of zero
35 * since the object is both in the dataset and in the io job entry.
36 *
37 * - What happens with MULTI/EXEC?
38 *
39 * Good question. Without some kind of versioning with a global counter
40 * it is not possible to have trasactions on disk, but they are still
41 * useful since from the point of view of memory and client bugs it is
42 * a protection anyway. Also it's useful for WATCH.
43 *
44 * Btw there is to check what happens when WATCH gets combined to keys
45 * that gets removed from the object cache. Should be save but better
46 * to check.
47 *
48 * - Check if/why INCR will not update the LRU info for the object.
49 *
50 * - Fix/Check the following race condition: a key gets a DEL so there is
51 * a write operation scheduled against this key. Later the same key will
52 * be the argument of a GET, but the write operation was still not
53 * completed (to delete the file). If the GET will be for some reason
54 * a blocking loading (via lookup) we can load the old value on memory.
55 *
56 * This problems can be fixed with negative caching. We can use it
57 * to optimize the system, but also when a key is deleted we mark
58 * it as non existing on disk as well (in a way that this cache
59 * entry can't be evicted, setting time to 0), then we avoid looking at
60 * the disk at all if the key can't be there. When an IO Job complete
61 * a deletion, we set the time of the negative caching to a non zero
62 * value so it will be evicted later.
63 *
64 * Are there other patterns like this where we load stale data?
65 *
66 * Also, make sure that key preloading is ONLY done for keys that are
67 * not marked as cacheKeyDoesNotExist(), otherwise, again, we can load
68 * data from disk that should instead be deleted.
69 *
70 * - dsSet() should use rename(2) in order to avoid corruptions.
71 *
72 * - Don't add a LOAD if there is already a LOADINPROGRESS, or is this
73 * impossible since anyway the io_keys stuff will work as lock?
74 *
75 * - Serialize special encoded things in a raw form.
76 */
77
78 /* Virtual Memory is composed mainly of two subsystems:
79 * - Blocking Virutal Memory
80 * - Threaded Virtual Memory I/O
81 * The two parts are not fully decoupled, but functions are split among two
82 * different sections of the source code (delimited by comments) in order to
83 * make more clear what functionality is about the blocking VM and what about
84 * the threaded (not blocking) VM.
85 *
86 * Redis VM design:
87 *
88 * Redis VM is a blocking VM (one that blocks reading swapped values from
89 * disk into memory when a value swapped out is needed in memory) that is made
90 * unblocking by trying to examine the command argument vector in order to
91 * load in background values that will likely be needed in order to exec
92 * the command. The command is executed only once all the relevant keys
93 * are loaded into memory.
94 *
95 * This basically is almost as simple of a blocking VM, but almost as parallel
96 * as a fully non-blocking VM.
97 */
98
99 void spawnIOThread(void);
100
101 /* =================== Virtual Memory - Blocking Side ====================== */
102
103 void dsInit(void) {
104 int pipefds[2];
105 size_t stacksize;
106
107 zmalloc_enable_thread_safeness(); /* we need thread safe zmalloc() */
108
109 redisLog(REDIS_NOTICE,"Opening Disk Store: %s", server.ds_path);
110 /* Open Disk Store */
111 if (dsOpen() != REDIS_OK) {
112 redisLog(REDIS_WARNING,"Fatal error opening disk store. Exiting.");
113 exit(1);
114 };
115
116 /* Initialize threaded I/O for Object Cache */
117 server.io_newjobs = listCreate();
118 server.io_processing = listCreate();
119 server.io_processed = listCreate();
120 server.io_ready_clients = listCreate();
121 pthread_mutex_init(&server.io_mutex,NULL);
122 pthread_cond_init(&server.io_condvar,NULL);
123 server.io_active_threads = 0;
124 if (pipe(pipefds) == -1) {
125 redisLog(REDIS_WARNING,"Unable to intialized DS: pipe(2): %s. Exiting."
126 ,strerror(errno));
127 exit(1);
128 }
129 server.io_ready_pipe_read = pipefds[0];
130 server.io_ready_pipe_write = pipefds[1];
131 redisAssert(anetNonBlock(NULL,server.io_ready_pipe_read) != ANET_ERR);
132 /* LZF requires a lot of stack */
133 pthread_attr_init(&server.io_threads_attr);
134 pthread_attr_getstacksize(&server.io_threads_attr, &stacksize);
135
136 /* Solaris may report a stacksize of 0, let's set it to 1 otherwise
137 * multiplying it by 2 in the while loop later will not really help ;) */
138 if (!stacksize) stacksize = 1;
139
140 while (stacksize < REDIS_THREAD_STACK_SIZE) stacksize *= 2;
141 pthread_attr_setstacksize(&server.io_threads_attr, stacksize);
142 /* Listen for events in the threaded I/O pipe */
143 if (aeCreateFileEvent(server.el, server.io_ready_pipe_read, AE_READABLE,
144 vmThreadedIOCompletedJob, NULL) == AE_ERR)
145 oom("creating file event");
146
147 /* Spawn our I/O thread */
148 spawnIOThread();
149 }
150
151 /* Compute how good candidate the specified object is for eviction.
152 * An higher number means a better candidate. */
153 double computeObjectSwappability(robj *o) {
154 /* actual age can be >= minage, but not < minage. As we use wrapping
155 * 21 bit clocks with minutes resolution for the LRU. */
156 return (double) estimateObjectIdleTime(o);
157 }
158
159 /* Try to free one entry from the diskstore object cache */
160 int cacheFreeOneEntry(void) {
161 int j, i;
162 struct dictEntry *best = NULL;
163 double best_swappability = 0;
164 redisDb *best_db = NULL;
165 robj *val;
166 sds key;
167
168 for (j = 0; j < server.dbnum; j++) {
169 redisDb *db = server.db+j;
170 /* Why maxtries is set to 100?
171 * Because this way (usually) we'll find 1 object even if just 1% - 2%
172 * are swappable objects */
173 int maxtries = 100;
174
175 if (dictSize(db->dict) == 0) continue;
176 for (i = 0; i < 5; i++) {
177 dictEntry *de;
178 double swappability;
179 robj keyobj;
180 sds keystr;
181
182 if (maxtries) maxtries--;
183 de = dictGetRandomKey(db->dict);
184 keystr = dictGetEntryKey(de);
185 val = dictGetEntryVal(de);
186 initStaticStringObject(keyobj,keystr);
187
188 /* Don't remove objects that are currently target of a
189 * read or write operation. */
190 if (cacheScheduleIOGetFlags(db,&keyobj) != 0) {
191 if (maxtries) i--; /* don't count this try */
192 continue;
193 }
194 swappability = computeObjectSwappability(val);
195 if (!best || swappability > best_swappability) {
196 best = de;
197 best_swappability = swappability;
198 best_db = db;
199 }
200 }
201 }
202 if (best == NULL) {
203 /* FIXME: If there are objects that are in the write queue
204 * so we can't delete them we should block here, at the cost of
205 * slowness as the object cache memory limit is considered
206 * n hard limit. */
207 return REDIS_ERR;
208 }
209 key = dictGetEntryKey(best);
210 val = dictGetEntryVal(best);
211
212 redisLog(REDIS_DEBUG,"Key selected for cache eviction: %s swappability:%f",
213 key, best_swappability);
214
215 /* Delete this key from memory */
216 {
217 robj *kobj = createStringObject(key,sdslen(key));
218 dbDelete(best_db,kobj);
219 decrRefCount(kobj);
220 }
221 return REDIS_OK;
222 }
223
224 /* Return true if it's safe to swap out objects in a given moment.
225 * Basically we don't want to swap objects out while there is a BGSAVE
226 * or a BGAEOREWRITE running in backgroud. */
227 int dsCanTouchDiskStore(void) {
228 return (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1);
229 }
230
231 /* ==================== Disk store negative caching ========================
232 *
233 * When disk store is enabled, we need negative caching, that is, to remember
234 * keys that are for sure *not* on the disk key-value store.
235 *
236 * This is usefuls because without negative caching cache misses will cost us
237 * a disk lookup, even if the same non existing key is accessed again and again.
238 *
239 * With negative caching we remember that the key is not on disk, so if it's
240 * not in memory and we have a negative cache entry, we don't try a disk
241 * access at all.
242 */
243
244 /* Returns true if the specified key may exists on disk, that is, we don't
245 * have an entry in our negative cache for this key */
246 int cacheKeyMayExist(redisDb *db, robj *key) {
247 return dictFind(db->io_negcache,key) == NULL;
248 }
249
250 /* Set the specified key as an entry that may possibily exist on disk, that is,
251 * remove the negative cache entry for this key if any. */
252 void cacheSetKeyMayExist(redisDb *db, robj *key) {
253 dictDelete(db->io_negcache,key);
254 }
255
256 /* Set the specified key as non existing on disk, that is, create a negative
257 * cache entry for this key. */
258 void cacheSetKeyDoesNotExist(redisDb *db, robj *key) {
259 if (dictReplace(db->io_negcache,key,(void*)time(NULL))) {
260 incrRefCount(key);
261 }
262 }
263
264 /* Remove one entry from negative cache using approximated LRU. */
265 int negativeCacheEvictOneEntry(void) {
266 struct dictEntry *de;
267 robj *best = NULL;
268 redisDb *best_db = NULL;
269 time_t time, best_time = 0;
270 int j;
271
272 for (j = 0; j < server.dbnum; j++) {
273 redisDb *db = server.db+j;
274 int i;
275
276 if (dictSize(db->io_negcache) == 0) continue;
277 for (i = 0; i < 3; i++) {
278 de = dictGetRandomKey(db->io_negcache);
279 time = (time_t) dictGetEntryVal(de);
280
281 if (best == NULL || time < best_time) {
282 best = dictGetEntryKey(de);
283 best_db = db;
284 best_time = time;
285 }
286 }
287 }
288 if (best) {
289 dictDelete(best_db->io_negcache,best);
290 return REDIS_OK;
291 } else {
292 return REDIS_ERR;
293 }
294 }
295
296 /* ================== Disk store cache - Threaded I/O ====================== */
297
298 void freeIOJob(iojob *j) {
299 decrRefCount(j->key);
300 /* j->val can be NULL if the job is about deleting the key from disk. */
301 if (j->val) decrRefCount(j->val);
302 zfree(j);
303 }
304
305 /* Every time a thread finished a Job, it writes a byte into the write side
306 * of an unix pipe in order to "awake" the main thread, and this function
307 * is called. */
308 void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
309 int mask)
310 {
311 char buf[1];
312 int retval, processed = 0, toprocess = -1;
313 REDIS_NOTUSED(el);
314 REDIS_NOTUSED(mask);
315 REDIS_NOTUSED(privdata);
316
317 /* For every byte we read in the read side of the pipe, there is one
318 * I/O job completed to process. */
319 while((retval = read(fd,buf,1)) == 1) {
320 iojob *j;
321 listNode *ln;
322
323 redisLog(REDIS_DEBUG,"Processing I/O completed job");
324
325 /* Get the processed element (the oldest one) */
326 lockThreadedIO();
327 redisAssert(listLength(server.io_processed) != 0);
328 if (toprocess == -1) {
329 toprocess = (listLength(server.io_processed)*REDIS_MAX_COMPLETED_JOBS_PROCESSED)/100;
330 if (toprocess <= 0) toprocess = 1;
331 }
332 ln = listFirst(server.io_processed);
333 j = ln->value;
334 listDelNode(server.io_processed,ln);
335 unlockThreadedIO();
336
337 /* Post process it in the main thread, as there are things we
338 * can do just here to avoid race conditions and/or invasive locks */
339 redisLog(REDIS_DEBUG,"COMPLETED Job type %s, key: %s",
340 (j->type == REDIS_IOJOB_LOAD) ? "load" : "save",
341 (unsigned char*)j->key->ptr);
342 if (j->type == REDIS_IOJOB_LOAD) {
343 /* Create the key-value pair in the in-memory database */
344 if (j->val != NULL) {
345 /* Note: it's possible that the key is already in memory
346 * due to a blocking load operation. */
347 if (dbAdd(j->db,j->key,j->val) == REDIS_OK) {
348 incrRefCount(j->val);
349 if (j->expire != -1) setExpire(j->db,j->key,j->expire);
350 }
351 }
352 cacheScheduleIODelFlag(j->db,j->key,REDIS_IO_LOADINPROG);
353 handleClientsBlockedOnSwappedKey(j->db,j->key);
354 freeIOJob(j);
355 } else if (j->type == REDIS_IOJOB_SAVE) {
356 cacheScheduleIODelFlag(j->db,j->key,REDIS_IO_SAVEINPROG);
357 freeIOJob(j);
358 }
359 processed++;
360 if (processed == toprocess) return;
361 }
362 if (retval < 0 && errno != EAGAIN) {
363 redisLog(REDIS_WARNING,
364 "WARNING: read(2) error in vmThreadedIOCompletedJob() %s",
365 strerror(errno));
366 }
367 }
368
369 void lockThreadedIO(void) {
370 pthread_mutex_lock(&server.io_mutex);
371 }
372
373 void unlockThreadedIO(void) {
374 pthread_mutex_unlock(&server.io_mutex);
375 }
376
377 void *IOThreadEntryPoint(void *arg) {
378 iojob *j;
379 listNode *ln;
380 REDIS_NOTUSED(arg);
381
382 pthread_detach(pthread_self());
383 lockThreadedIO();
384 while(1) {
385 /* Get a new job to process */
386 if (listLength(server.io_newjobs) == 0) {
387 /* Wait for more work to do */
388 pthread_cond_wait(&server.io_condvar,&server.io_mutex);
389 continue;
390 }
391 redisLog(REDIS_DEBUG,"%ld IO jobs to process",
392 listLength(server.io_newjobs));
393 ln = listFirst(server.io_newjobs);
394 j = ln->value;
395 listDelNode(server.io_newjobs,ln);
396 /* Add the job in the processing queue */
397 listAddNodeTail(server.io_processing,j);
398 ln = listLast(server.io_processing); /* We use ln later to remove it */
399 unlockThreadedIO();
400
401 redisLog(REDIS_DEBUG,"Thread %ld: new job type %s: %p about key '%s'",
402 (long) pthread_self(),
403 (j->type == REDIS_IOJOB_LOAD) ? "load" : "save",
404 (void*)j, (char*)j->key->ptr);
405
406 /* Process the Job */
407 if (j->type == REDIS_IOJOB_LOAD) {
408 time_t expire;
409
410 j->val = dsGet(j->db,j->key,&expire);
411 if (j->val) j->expire = expire;
412 } else if (j->type == REDIS_IOJOB_SAVE) {
413 if (j->val) {
414 dsSet(j->db,j->key,j->val);
415 } else {
416 dsDel(j->db,j->key);
417 }
418 }
419
420 /* Done: insert the job into the processed queue */
421 redisLog(REDIS_DEBUG,"Thread %ld completed the job: %p (key %s)",
422 (long) pthread_self(), (void*)j, (char*)j->key->ptr);
423
424 lockThreadedIO();
425 listDelNode(server.io_processing,ln);
426 listAddNodeTail(server.io_processed,j);
427
428 /* Signal the main thread there is new stuff to process */
429 redisAssert(write(server.io_ready_pipe_write,"x",1) == 1);
430 }
431 /* never reached, but that's the full pattern... */
432 unlockThreadedIO();
433 return NULL;
434 }
435
436 void spawnIOThread(void) {
437 pthread_t thread;
438 sigset_t mask, omask;
439 int err;
440
441 sigemptyset(&mask);
442 sigaddset(&mask,SIGCHLD);
443 sigaddset(&mask,SIGHUP);
444 sigaddset(&mask,SIGPIPE);
445 pthread_sigmask(SIG_SETMASK, &mask, &omask);
446 while ((err = pthread_create(&thread,&server.io_threads_attr,IOThreadEntryPoint,NULL)) != 0) {
447 redisLog(REDIS_WARNING,"Unable to spawn an I/O thread: %s",
448 strerror(err));
449 usleep(1000000);
450 }
451 pthread_sigmask(SIG_SETMASK, &omask, NULL);
452 server.io_active_threads++;
453 }
454
455 /* Wait that all the pending IO Jobs are processed */
456 void waitEmptyIOJobsQueue(void) {
457 while(1) {
458 int io_processed_len;
459
460 lockThreadedIO();
461 if (listLength(server.io_newjobs) == 0 &&
462 listLength(server.io_processing) == 0)
463 {
464 unlockThreadedIO();
465 return;
466 }
467 /* If there are new jobs we need to signal the thread to
468 * process the next one. */
469 redisLog(REDIS_DEBUG,"waitEmptyIOJobsQueue: new %d, processing %d",
470 listLength(server.io_newjobs),
471 listLength(server.io_processing));
472 /*
473 if (listLength(server.io_newjobs)) {
474 pthread_cond_signal(&server.io_condvar);
475 }
476 */
477 /* While waiting for empty jobs queue condition we post-process some
478 * finshed job, as I/O threads may be hanging trying to write against
479 * the io_ready_pipe_write FD but there are so much pending jobs that
480 * it's blocking. */
481 io_processed_len = listLength(server.io_processed);
482 unlockThreadedIO();
483 if (io_processed_len) {
484 vmThreadedIOCompletedJob(NULL,server.io_ready_pipe_read,
485 (void*)0xdeadbeef,0);
486 usleep(1000); /* 1 millisecond */
487 } else {
488 usleep(10000); /* 10 milliseconds */
489 }
490 }
491 }
492
493 /* Process all the IO Jobs already completed by threads but still waiting
494 * processing from the main thread. */
495 void processAllPendingIOJobs(void) {
496 while(1) {
497 int io_processed_len;
498
499 lockThreadedIO();
500 io_processed_len = listLength(server.io_processed);
501 unlockThreadedIO();
502 if (io_processed_len == 0) return;
503 vmThreadedIOCompletedJob(NULL,server.io_ready_pipe_read,
504 (void*)0xdeadbeef,0);
505 }
506 }
507
508 /* This function must be called while with threaded IO locked */
509 void queueIOJob(iojob *j) {
510 redisLog(REDIS_DEBUG,"Queued IO Job %p type %d about key '%s'\n",
511 (void*)j, j->type, (char*)j->key->ptr);
512 listAddNodeTail(server.io_newjobs,j);
513 if (server.io_active_threads < server.vm_max_threads)
514 spawnIOThread();
515 }
516
517 void dsCreateIOJob(int type, redisDb *db, robj *key, robj *val) {
518 iojob *j;
519
520 j = zmalloc(sizeof(*j));
521 j->type = type;
522 j->db = db;
523 j->key = key;
524 incrRefCount(key);
525 j->val = val;
526 if (val) incrRefCount(val);
527
528 lockThreadedIO();
529 queueIOJob(j);
530 pthread_cond_signal(&server.io_condvar);
531 unlockThreadedIO();
532 }
533
534 /* ============= Disk store cache - Scheduling of IO operations =============
535 *
536 * We use a queue and an hash table to hold the state of IO operations
537 * so that's fast to lookup if there is already an IO operation in queue
538 * for a given key.
539 *
540 * There are two types of IO operations for a given key:
541 * REDIS_IO_LOAD and REDIS_IO_SAVE.
542 *
543 * The function cacheScheduleIO() function pushes the specified IO operation
544 * in the queue, but avoid adding the same key for the same operation
545 * multiple times, thanks to the associated hash table.
546 *
547 * We take a set of flags per every key, so when the scheduled IO operation
548 * gets moved from the scheduled queue to the actual IO Jobs queue that
549 * is processed by the IO thread, we flag it as IO_LOADINPROG or
550 * IO_SAVEINPROG.
551 *
552 * So for every given key we always know if there is some IO operation
553 * scheduled, or in progress, for this key.
554 *
555 * NOTE: all this is very important in order to guarantee correctness of
556 * the Disk Store Cache. Jobs are always queued here. Load jobs are
557 * queued at the head for faster execution only in the case there is not
558 * already a write operation of some kind for this job.
559 *
560 * So we have ordering, but can do exceptions when there are no already
561 * operations for a given key. Also when we need to block load a given
562 * key, for an immediate lookup operation, we can check if the key can
563 * be accessed synchronously without race conditions (no IN PROGRESS
564 * operations for this key), otherwise we blocking wait for completion. */
565
566 #define REDIS_IO_LOAD 1
567 #define REDIS_IO_SAVE 2
568 #define REDIS_IO_LOADINPROG 4
569 #define REDIS_IO_SAVEINPROG 8
570
571 void cacheScheduleIOAddFlag(redisDb *db, robj *key, long flag) {
572 struct dictEntry *de = dictFind(db->io_queued,key);
573
574 if (!de) {
575 dictAdd(db->io_queued,key,(void*)flag);
576 incrRefCount(key);
577 return;
578 } else {
579 long flags = (long) dictGetEntryVal(de);
580
581 if (flags & flag) {
582 redisLog(REDIS_WARNING,"Adding the same flag again: was: %ld, addede: %ld",flags,flag);
583 redisAssert(!(flags & flag));
584 }
585 flags |= flag;
586 dictGetEntryVal(de) = (void*) flags;
587 }
588 }
589
590 void cacheScheduleIODelFlag(redisDb *db, robj *key, long flag) {
591 struct dictEntry *de = dictFind(db->io_queued,key);
592 long flags;
593
594 redisAssert(de != NULL);
595 flags = (long) dictGetEntryVal(de);
596 redisAssert(flags & flag);
597 flags &= ~flag;
598 if (flags == 0) {
599 dictDelete(db->io_queued,key);
600 } else {
601 dictGetEntryVal(de) = (void*) flags;
602 }
603 }
604
605 int cacheScheduleIOGetFlags(redisDb *db, robj *key) {
606 struct dictEntry *de = dictFind(db->io_queued,key);
607
608 return (de == NULL) ? 0 : ((long) dictGetEntryVal(de));
609 }
610
611 void cacheScheduleIO(redisDb *db, robj *key, int type) {
612 ioop *op;
613 long flags;
614
615 if ((flags = cacheScheduleIOGetFlags(db,key)) & type) return;
616
617 redisLog(REDIS_DEBUG,"Scheduling key %s for %s",
618 key->ptr, type == REDIS_IO_LOAD ? "loading" : "saving");
619 cacheScheduleIOAddFlag(db,key,type);
620 op = zmalloc(sizeof(*op));
621 op->type = type;
622 op->db = db;
623 op->key = key;
624 incrRefCount(key);
625 op->ctime = time(NULL);
626
627 /* Give priority to load operations if there are no save already
628 * in queue for the same key. */
629 if (type == REDIS_IO_LOAD && !(flags & REDIS_IO_SAVE)) {
630 listAddNodeHead(server.cache_io_queue, op);
631 } else {
632 /* FIXME: probably when this happens we want to at least move
633 * the write job about this queue on top, and set the creation time
634 * to a value that will force processing ASAP. */
635 listAddNodeTail(server.cache_io_queue, op);
636 }
637 }
638
639 void cacheCron(void) {
640 time_t now = time(NULL);
641 listNode *ln;
642 int jobs, topush = 0;
643
644 /* Sync stuff on disk, but only if we have less than 100 IO jobs */
645 lockThreadedIO();
646 jobs = listLength(server.io_newjobs);
647 unlockThreadedIO();
648
649 topush = 100-jobs;
650 if (topush < 0) topush = 0;
651 if (topush > (signed)listLength(server.cache_io_queue))
652 topush = listLength(server.cache_io_queue);
653
654 while((ln = listFirst(server.cache_io_queue)) != NULL) {
655 ioop *op = ln->value;
656
657 if (!topush) break;
658 topush--;
659
660 if (op->type == REDIS_IO_LOAD ||
661 (now - op->ctime) >= server.cache_flush_delay)
662 {
663 struct dictEntry *de;
664 robj *val;
665
666 /* Don't add a SAVE job in queue if there is already
667 * a save in progress for the same key. */
668 if (op->type == REDIS_IO_SAVE &&
669 cacheScheduleIOGetFlags(op->db,op->key) & REDIS_IO_SAVEINPROG)
670 {
671 /* Move the operation at the end of the list of there
672 * are other operations. Otherwise break, nothing to do
673 * here. */
674 if (listLength(server.cache_io_queue) > 1) {
675 listDelNode(server.cache_io_queue,ln);
676 listAddNodeTail(server.cache_io_queue,op);
677 continue;
678 } else {
679 break;
680 }
681 }
682
683 redisLog(REDIS_DEBUG,"Creating IO %s Job for key %s",
684 op->type == REDIS_IO_LOAD ? "load" : "save", op->key->ptr);
685
686 if (op->type == REDIS_IO_LOAD) {
687 dsCreateIOJob(REDIS_IOJOB_LOAD,op->db,op->key,NULL);
688 } else {
689 /* Lookup the key, in order to put the current value in the IO
690 * Job. Otherwise if the key does not exists we schedule a disk
691 * store delete operation, setting the value to NULL. */
692 de = dictFind(op->db->dict,op->key->ptr);
693 if (de) {
694 val = dictGetEntryVal(de);
695 } else {
696 /* Setting the value to NULL tells the IO thread to delete
697 * the key on disk. */
698 val = NULL;
699 }
700 dsCreateIOJob(REDIS_IOJOB_SAVE,op->db,op->key,val);
701 }
702 /* Mark the operation as in progress. */
703 cacheScheduleIODelFlag(op->db,op->key,op->type);
704 cacheScheduleIOAddFlag(op->db,op->key,
705 (op->type == REDIS_IO_LOAD) ? REDIS_IO_LOADINPROG :
706 REDIS_IO_SAVEINPROG);
707 /* Finally remove the operation from the queue.
708 * But we'll have trace of it in the hash table. */
709 listDelNode(server.cache_io_queue,ln);
710 decrRefCount(op->key);
711 zfree(op);
712 } else {
713 break; /* too early */
714 }
715 }
716
717 /* Reclaim memory from the object cache */
718 while (server.ds_enabled && zmalloc_used_memory() >
719 server.cache_max_memory)
720 {
721 int done = 0;
722
723 if (cacheFreeOneEntry() == REDIS_OK) done++;
724 if (negativeCacheEvictOneEntry() == REDIS_OK) done++;
725 if (done == 0) break; /* nothing more to free */
726 }
727 }
728
729 /* ========== Disk store cache - Blocking clients on missing keys =========== */
730
731 /* This function makes the clinet 'c' waiting for the key 'key' to be loaded.
732 * If the key is already in memory we don't need to block.
733 *
734 * FIXME: we should try if it's actually better to suspend the client
735 * accessing an object that is being saved, and awake it only when
736 * the saving was completed.
737 *
738 * Otherwise if the key is not in memory, we block the client and start
739 * an IO Job to load it:
740 *
741 * the key is added to the io_keys list in the client structure, and also
742 * in the hash table mapping swapped keys to waiting clients, that is,
743 * server.io_waited_keys. */
744 int waitForSwappedKey(redisClient *c, robj *key) {
745 struct dictEntry *de;
746 list *l;
747
748 /* Return ASAP if the key is in memory */
749 de = dictFind(c->db->dict,key->ptr);
750 if (de != NULL) return 0;
751
752 /* Don't wait for keys we are sure are not on disk either */
753 if (!cacheKeyMayExist(c->db,key)) return 0;
754
755 /* Add the key to the list of keys this client is waiting for.
756 * This maps clients to keys they are waiting for. */
757 listAddNodeTail(c->io_keys,key);
758 incrRefCount(key);
759
760 /* Add the client to the swapped keys => clients waiting map. */
761 de = dictFind(c->db->io_keys,key);
762 if (de == NULL) {
763 int retval;
764
765 /* For every key we take a list of clients blocked for it */
766 l = listCreate();
767 retval = dictAdd(c->db->io_keys,key,l);
768 incrRefCount(key);
769 redisAssert(retval == DICT_OK);
770 } else {
771 l = dictGetEntryVal(de);
772 }
773 listAddNodeTail(l,c);
774
775 /* Are we already loading the key from disk? If not create a job */
776 if (de == NULL)
777 cacheScheduleIO(c->db,key,REDIS_IO_LOAD);
778 return 1;
779 }
780
781 /* Preload keys for any command with first, last and step values for
782 * the command keys prototype, as defined in the command table. */
783 void waitForMultipleSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) {
784 int j, last;
785 if (cmd->vm_firstkey == 0) return;
786 last = cmd->vm_lastkey;
787 if (last < 0) last = argc+last;
788 for (j = cmd->vm_firstkey; j <= last; j += cmd->vm_keystep) {
789 redisAssert(j < argc);
790 waitForSwappedKey(c,argv[j]);
791 }
792 }
793
794 /* Preload keys needed for the ZUNIONSTORE and ZINTERSTORE commands.
795 * Note that the number of keys to preload is user-defined, so we need to
796 * apply a sanity check against argc. */
797 void zunionInterBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) {
798 int i, num;
799 REDIS_NOTUSED(cmd);
800
801 num = atoi(argv[2]->ptr);
802 if (num > (argc-3)) return;
803 for (i = 0; i < num; i++) {
804 waitForSwappedKey(c,argv[3+i]);
805 }
806 }
807
808 /* Preload keys needed to execute the entire MULTI/EXEC block.
809 *
810 * This function is called by blockClientOnSwappedKeys when EXEC is issued,
811 * and will block the client when any command requires a swapped out value. */
812 void execBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) {
813 int i, margc;
814 struct redisCommand *mcmd;
815 robj **margv;
816 REDIS_NOTUSED(cmd);
817 REDIS_NOTUSED(argc);
818 REDIS_NOTUSED(argv);
819
820 if (!(c->flags & REDIS_MULTI)) return;
821 for (i = 0; i < c->mstate.count; i++) {
822 mcmd = c->mstate.commands[i].cmd;
823 margc = c->mstate.commands[i].argc;
824 margv = c->mstate.commands[i].argv;
825
826 if (mcmd->vm_preload_proc != NULL) {
827 mcmd->vm_preload_proc(c,mcmd,margc,margv);
828 } else {
829 waitForMultipleSwappedKeys(c,mcmd,margc,margv);
830 }
831 }
832 }
833
834 /* Is this client attempting to run a command against swapped keys?
835 * If so, block it ASAP, load the keys in background, then resume it.
836 *
837 * The important idea about this function is that it can fail! If keys will
838 * still be swapped when the client is resumed, this key lookups will
839 * just block loading keys from disk. In practical terms this should only
840 * happen with SORT BY command or if there is a bug in this function.
841 *
842 * Return 1 if the client is marked as blocked, 0 if the client can
843 * continue as the keys it is going to access appear to be in memory. */
844 int blockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd) {
845 if (cmd->vm_preload_proc != NULL) {
846 cmd->vm_preload_proc(c,cmd,c->argc,c->argv);
847 } else {
848 waitForMultipleSwappedKeys(c,cmd,c->argc,c->argv);
849 }
850
851 /* If the client was blocked for at least one key, mark it as blocked. */
852 if (listLength(c->io_keys)) {
853 c->flags |= REDIS_IO_WAIT;
854 aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
855 server.cache_blocked_clients++;
856 return 1;
857 } else {
858 return 0;
859 }
860 }
861
862 /* Remove the 'key' from the list of blocked keys for a given client.
863 *
864 * The function returns 1 when there are no longer blocking keys after
865 * the current one was removed (and the client can be unblocked). */
866 int dontWaitForSwappedKey(redisClient *c, robj *key) {
867 list *l;
868 listNode *ln;
869 listIter li;
870 struct dictEntry *de;
871
872 /* The key object might be destroyed when deleted from the c->io_keys
873 * list (and the "key" argument is physically the same object as the
874 * object inside the list), so we need to protect it. */
875 incrRefCount(key);
876
877 /* Remove the key from the list of keys this client is waiting for. */
878 listRewind(c->io_keys,&li);
879 while ((ln = listNext(&li)) != NULL) {
880 if (equalStringObjects(ln->value,key)) {
881 listDelNode(c->io_keys,ln);
882 break;
883 }
884 }
885 redisAssert(ln != NULL);
886
887 /* Remove the client form the key => waiting clients map. */
888 de = dictFind(c->db->io_keys,key);
889 redisAssert(de != NULL);
890 l = dictGetEntryVal(de);
891 ln = listSearchKey(l,c);
892 redisAssert(ln != NULL);
893 listDelNode(l,ln);
894 if (listLength(l) == 0)
895 dictDelete(c->db->io_keys,key);
896
897 decrRefCount(key);
898 return listLength(c->io_keys) == 0;
899 }
900
901 /* Every time we now a key was loaded back in memory, we handle clients
902 * waiting for this key if any. */
903 void handleClientsBlockedOnSwappedKey(redisDb *db, robj *key) {
904 struct dictEntry *de;
905 list *l;
906 listNode *ln;
907 int len;
908
909 de = dictFind(db->io_keys,key);
910 if (!de) return;
911
912 l = dictGetEntryVal(de);
913 len = listLength(l);
914 /* Note: we can't use something like while(listLength(l)) as the list
915 * can be freed by the calling function when we remove the last element. */
916 while (len--) {
917 ln = listFirst(l);
918 redisClient *c = ln->value;
919
920 if (dontWaitForSwappedKey(c,key)) {
921 /* Put the client in the list of clients ready to go as we
922 * loaded all the keys about it. */
923 listAddNodeTail(server.io_ready_clients,c);
924 }
925 }
926 }