]> git.saurik.com Git - redis.git/blob - src/dscache.c
a0c252d5eaad0c72e00b31030f724fcd4f7aba6e
[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 * - 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).
26 *
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.
32 *
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
43 * NULL from lookup.
44 *
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.
49 *
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.
53 *
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.
58 *
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.
62 *
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.
65 *
66 * - What happens when an object is destroyed?
67 *
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
71 * who cares.
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.
76 *
77 * - What happens when keys are deleted?
78 *
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.
83 *
84 * - What happens with MULTI/EXEC?
85 *
86 * Good question.
87 *
88 * - If dsSet() fails on the write thread log the error and reschedule the
89 * key for flush.
90 *
91 * - Check why INCR will not update the LRU info for the object.
92 *
93 * - Fix/Check the following race condition: a key gets a DEL so there is
94 * a write operation scheduled against this key. Later the same key will
95 * be the argument of a GET, but the write operation was still not
96 * completed (to delete the file). If the GET will be for some reason
97 * a blocking loading (via lookup) we can load the old value on memory.
98 *
99 * This problems can be fixed with negative caching. We can use it
100 * to optimize the system, but also when a key is deleted we mark
101 * it as non existing on disk as well (in a way that this cache
102 * entry can't be evicted, setting time to 0), then we avoid looking at
103 * the disk at all if the key can't be there. When an IO Job complete
104 * a deletion, we set the time of the negative caching to a non zero
105 * value so it will be evicted later.
106 *
107 * Are there other patterns like this where we load stale data?
108 */
109
110 /* Virtual Memory is composed mainly of two subsystems:
111 * - Blocking Virutal Memory
112 * - Threaded Virtual Memory I/O
113 * The two parts are not fully decoupled, but functions are split among two
114 * different sections of the source code (delimited by comments) in order to
115 * make more clear what functionality is about the blocking VM and what about
116 * the threaded (not blocking) VM.
117 *
118 * Redis VM design:
119 *
120 * Redis VM is a blocking VM (one that blocks reading swapped values from
121 * disk into memory when a value swapped out is needed in memory) that is made
122 * unblocking by trying to examine the command argument vector in order to
123 * load in background values that will likely be needed in order to exec
124 * the command. The command is executed only once all the relevant keys
125 * are loaded into memory.
126 *
127 * This basically is almost as simple of a blocking VM, but almost as parallel
128 * as a fully non-blocking VM.
129 */
130
131 void spawnIOThread(void);
132
133 /* =================== Virtual Memory - Blocking Side ====================== */
134
135 void dsInit(void) {
136 int pipefds[2];
137 size_t stacksize;
138
139 zmalloc_enable_thread_safeness(); /* we need thread safe zmalloc() */
140
141 redisLog(REDIS_NOTICE,"Opening Disk Store: %s", server.ds_path);
142 /* Open Disk Store */
143 if (dsOpen() != REDIS_OK) {
144 redisLog(REDIS_WARNING,"Fatal error opening disk store. Exiting.");
145 exit(1);
146 };
147
148 /* Initialize threaded I/O for Object Cache */
149 server.io_newjobs = listCreate();
150 server.io_processing = listCreate();
151 server.io_processed = listCreate();
152 server.io_ready_clients = listCreate();
153 pthread_mutex_init(&server.io_mutex,NULL);
154 pthread_cond_init(&server.io_condvar,NULL);
155 server.io_active_threads = 0;
156 if (pipe(pipefds) == -1) {
157 redisLog(REDIS_WARNING,"Unable to intialized DS: pipe(2): %s. Exiting."
158 ,strerror(errno));
159 exit(1);
160 }
161 server.io_ready_pipe_read = pipefds[0];
162 server.io_ready_pipe_write = pipefds[1];
163 redisAssert(anetNonBlock(NULL,server.io_ready_pipe_read) != ANET_ERR);
164 /* LZF requires a lot of stack */
165 pthread_attr_init(&server.io_threads_attr);
166 pthread_attr_getstacksize(&server.io_threads_attr, &stacksize);
167
168 /* Solaris may report a stacksize of 0, let's set it to 1 otherwise
169 * multiplying it by 2 in the while loop later will not really help ;) */
170 if (!stacksize) stacksize = 1;
171
172 while (stacksize < REDIS_THREAD_STACK_SIZE) stacksize *= 2;
173 pthread_attr_setstacksize(&server.io_threads_attr, stacksize);
174 /* Listen for events in the threaded I/O pipe */
175 if (aeCreateFileEvent(server.el, server.io_ready_pipe_read, AE_READABLE,
176 vmThreadedIOCompletedJob, NULL) == AE_ERR)
177 oom("creating file event");
178
179 /* Spawn our I/O thread */
180 spawnIOThread();
181 }
182
183 /* Compute how good candidate the specified object is for eviction.
184 * An higher number means a better candidate. */
185 double computeObjectSwappability(robj *o) {
186 /* actual age can be >= minage, but not < minage. As we use wrapping
187 * 21 bit clocks with minutes resolution for the LRU. */
188 return (double) estimateObjectIdleTime(o);
189 }
190
191 /* Try to free one entry from the diskstore object cache */
192 int cacheFreeOneEntry(void) {
193 int j, i;
194 struct dictEntry *best = NULL;
195 double best_swappability = 0;
196 redisDb *best_db = NULL;
197 robj *val;
198 sds key;
199
200 for (j = 0; j < server.dbnum; j++) {
201 redisDb *db = server.db+j;
202 /* Why maxtries is set to 100?
203 * Because this way (usually) we'll find 1 object even if just 1% - 2%
204 * are swappable objects */
205 int maxtries = 100;
206
207 if (dictSize(db->dict) == 0) continue;
208 for (i = 0; i < 5; i++) {
209 dictEntry *de;
210 double swappability;
211
212 if (maxtries) maxtries--;
213 de = dictGetRandomKey(db->dict);
214 val = dictGetEntryVal(de);
215 /* Only swap objects that are currently in memory.
216 *
217 * Also don't swap shared objects: not a good idea in general and
218 * we need to ensure that the main thread does not touch the
219 * object while the I/O thread is using it, but we can't
220 * control other keys without adding additional mutex. */
221 if (val->storage != REDIS_DS_MEMORY) {
222 if (maxtries) i--; /* don't count this try */
223 continue;
224 }
225 swappability = computeObjectSwappability(val);
226 if (!best || swappability > best_swappability) {
227 best = de;
228 best_swappability = swappability;
229 best_db = db;
230 }
231 }
232 }
233 if (best == NULL) {
234 /* FIXME: If there are objects marked as DS_DIRTY or DS_SAVING
235 * let's wait for this objects to be clear and retry...
236 *
237 * Object cache vm limit is considered an hard limit. */
238 return REDIS_ERR;
239 }
240 key = dictGetEntryKey(best);
241 val = dictGetEntryVal(best);
242
243 redisLog(REDIS_DEBUG,"Key selected for cache eviction: %s swappability:%f",
244 key, best_swappability);
245
246 /* Delete this key from memory */
247 {
248 robj *kobj = createStringObject(key,sdslen(key));
249 dbDelete(best_db,kobj);
250 decrRefCount(kobj);
251 }
252 return REDIS_OK;
253 }
254
255 /* Return true if it's safe to swap out objects in a given moment.
256 * Basically we don't want to swap objects out while there is a BGSAVE
257 * or a BGAEOREWRITE running in backgroud. */
258 int dsCanTouchDiskStore(void) {
259 return (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1);
260 }
261
262 /* =================== Virtual Memory - Threaded I/O ======================= */
263
264 void freeIOJob(iojob *j) {
265 decrRefCount(j->key);
266 /* j->val can be NULL if the job is about deleting the key from disk. */
267 if (j->val) decrRefCount(j->val);
268 zfree(j);
269 }
270
271 /* Every time a thread finished a Job, it writes a byte into the write side
272 * of an unix pipe in order to "awake" the main thread, and this function
273 * is called. */
274 void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
275 int mask)
276 {
277 char buf[1];
278 int retval, processed = 0, toprocess = -1;
279 REDIS_NOTUSED(el);
280 REDIS_NOTUSED(mask);
281 REDIS_NOTUSED(privdata);
282
283 /* For every byte we read in the read side of the pipe, there is one
284 * I/O job completed to process. */
285 while((retval = read(fd,buf,1)) == 1) {
286 iojob *j;
287 listNode *ln;
288
289 redisLog(REDIS_DEBUG,"Processing I/O completed job");
290
291 /* Get the processed element (the oldest one) */
292 lockThreadedIO();
293 redisAssert(listLength(server.io_processed) != 0);
294 if (toprocess == -1) {
295 toprocess = (listLength(server.io_processed)*REDIS_MAX_COMPLETED_JOBS_PROCESSED)/100;
296 if (toprocess <= 0) toprocess = 1;
297 }
298 ln = listFirst(server.io_processed);
299 j = ln->value;
300 listDelNode(server.io_processed,ln);
301 unlockThreadedIO();
302
303 /* Post process it in the main thread, as there are things we
304 * can do just here to avoid race conditions and/or invasive locks */
305 redisLog(REDIS_DEBUG,"COMPLETED Job type %s, key: %s",
306 (j->type == REDIS_IOJOB_LOAD) ? "load" : "save",
307 (unsigned char*)j->key->ptr);
308 if (j->type == REDIS_IOJOB_LOAD) {
309 /* Create the key-value pair in the in-memory database */
310 if (j->val != NULL) {
311 /* Note: the key may already be here if between the time
312 * this key loading was scheduled and now there was the
313 * need to blocking load the key for a key lookup. */
314 if (dbAdd(j->db,j->key,j->val) == REDIS_OK) {
315 incrRefCount(j->val);
316 if (j->expire != -1) setExpire(j->db,j->key,j->expire);
317 }
318 } else {
319 /* The key does not exist. Create a negative cache entry
320 * for this key. */
321 /* FIXME: add this entry into the negative cache */
322 }
323 /* Handle clients waiting for this key to be loaded. */
324 handleClientsBlockedOnSwappedKey(j->db,j->key);
325 freeIOJob(j);
326 } else if (j->type == REDIS_IOJOB_SAVE) {
327 if (j->val) {
328 redisAssert(j->val->storage == REDIS_DS_SAVING);
329 j->val->storage = REDIS_DS_MEMORY;
330 }
331 freeIOJob(j);
332 }
333 processed++;
334 if (processed == toprocess) return;
335 }
336 if (retval < 0 && errno != EAGAIN) {
337 redisLog(REDIS_WARNING,
338 "WARNING: read(2) error in vmThreadedIOCompletedJob() %s",
339 strerror(errno));
340 }
341 }
342
343 void lockThreadedIO(void) {
344 pthread_mutex_lock(&server.io_mutex);
345 }
346
347 void unlockThreadedIO(void) {
348 pthread_mutex_unlock(&server.io_mutex);
349 }
350
351 void *IOThreadEntryPoint(void *arg) {
352 iojob *j;
353 listNode *ln;
354 REDIS_NOTUSED(arg);
355
356 pthread_detach(pthread_self());
357 lockThreadedIO();
358 while(1) {
359 /* Wait for more work to do */
360 pthread_cond_wait(&server.io_condvar,&server.io_mutex);
361 /* Get a new job to process */
362 if (listLength(server.io_newjobs) == 0) {
363 /* No new jobs in queue, reiterate. */
364 unlockThreadedIO();
365 continue;
366 }
367 ln = listFirst(server.io_newjobs);
368 j = ln->value;
369 listDelNode(server.io_newjobs,ln);
370 /* Add the job in the processing queue */
371 listAddNodeTail(server.io_processing,j);
372 ln = listLast(server.io_processing); /* We use ln later to remove it */
373 unlockThreadedIO();
374
375 redisLog(REDIS_DEBUG,"Thread %ld: new job type %s: %p about key '%s'",
376 (long) pthread_self(),
377 (j->type == REDIS_IOJOB_LOAD) ? "load" : "save",
378 (void*)j, (char*)j->key->ptr);
379
380 /* Process the Job */
381 if (j->type == REDIS_IOJOB_LOAD) {
382 time_t expire;
383
384 j->val = dsGet(j->db,j->key,&expire);
385 if (j->val) j->expire = expire;
386 } else if (j->type == REDIS_IOJOB_SAVE) {
387 if (j->val) {
388 redisAssert(j->val->storage == REDIS_DS_SAVING);
389 dsSet(j->db,j->key,j->val);
390 } else {
391 dsDel(j->db,j->key);
392 }
393 }
394
395 /* Done: insert the job into the processed queue */
396 redisLog(REDIS_DEBUG,"Thread %ld completed the job: %p (key %s)",
397 (long) pthread_self(), (void*)j, (char*)j->key->ptr);
398
399 lockThreadedIO();
400 listDelNode(server.io_processing,ln);
401 listAddNodeTail(server.io_processed,j);
402
403 /* Signal the main thread there is new stuff to process */
404 redisAssert(write(server.io_ready_pipe_write,"x",1) == 1);
405 }
406 /* never reached, but that's the full pattern... */
407 unlockThreadedIO();
408 return NULL;
409 }
410
411 void spawnIOThread(void) {
412 pthread_t thread;
413 sigset_t mask, omask;
414 int err;
415
416 sigemptyset(&mask);
417 sigaddset(&mask,SIGCHLD);
418 sigaddset(&mask,SIGHUP);
419 sigaddset(&mask,SIGPIPE);
420 pthread_sigmask(SIG_SETMASK, &mask, &omask);
421 while ((err = pthread_create(&thread,&server.io_threads_attr,IOThreadEntryPoint,NULL)) != 0) {
422 redisLog(REDIS_WARNING,"Unable to spawn an I/O thread: %s",
423 strerror(err));
424 usleep(1000000);
425 }
426 pthread_sigmask(SIG_SETMASK, &omask, NULL);
427 server.io_active_threads++;
428 }
429
430 /* Wait that all the pending IO Jobs are processed */
431 void waitEmptyIOJobsQueue(void) {
432 while(1) {
433 int io_processed_len;
434
435 lockThreadedIO();
436 if (listLength(server.io_newjobs) == 0 &&
437 listLength(server.io_processing) == 0)
438 {
439 unlockThreadedIO();
440 return;
441 }
442 /* While waiting for empty jobs queue condition we post-process some
443 * finshed job, as I/O threads may be hanging trying to write against
444 * the io_ready_pipe_write FD but there are so much pending jobs that
445 * it's blocking. */
446 io_processed_len = listLength(server.io_processed);
447 unlockThreadedIO();
448 if (io_processed_len) {
449 vmThreadedIOCompletedJob(NULL,server.io_ready_pipe_read,
450 (void*)0xdeadbeef,0);
451 usleep(1000); /* 1 millisecond */
452 } else {
453 usleep(10000); /* 10 milliseconds */
454 }
455 }
456 }
457
458 /* Process all the IO Jobs already completed by threads but still waiting
459 * processing from the main thread. */
460 void processAllPendingIOJobs(void) {
461 while(1) {
462 int io_processed_len;
463
464 lockThreadedIO();
465 io_processed_len = listLength(server.io_processed);
466 unlockThreadedIO();
467 if (io_processed_len == 0) return;
468 vmThreadedIOCompletedJob(NULL,server.io_ready_pipe_read,
469 (void*)0xdeadbeef,0);
470 }
471 }
472
473 /* This function must be called while with threaded IO locked */
474 void queueIOJob(iojob *j) {
475 redisLog(REDIS_DEBUG,"Queued IO Job %p type %d about key '%s'\n",
476 (void*)j, j->type, (char*)j->key->ptr);
477 listAddNodeTail(server.io_newjobs,j);
478 if (server.io_active_threads < server.vm_max_threads)
479 spawnIOThread();
480 }
481
482 void dsCreateIOJob(int type, redisDb *db, robj *key, robj *val) {
483 iojob *j;
484
485 j = zmalloc(sizeof(*j));
486 j->type = type;
487 j->db = db;
488 j->key = key;
489 incrRefCount(key);
490 j->val = val;
491 if (val) incrRefCount(val);
492
493 lockThreadedIO();
494 queueIOJob(j);
495 pthread_cond_signal(&server.io_condvar);
496 unlockThreadedIO();
497 }
498
499 void cacheScheduleForFlush(redisDb *db, robj *key) {
500 dirtykey *dk;
501 dictEntry *de;
502
503 de = dictFind(db->dict,key->ptr);
504 if (de) {
505 robj *val = dictGetEntryVal(de);
506 if (val->storage == REDIS_DS_DIRTY)
507 return;
508 else
509 val->storage = REDIS_DS_DIRTY;
510 }
511
512 redisLog(REDIS_DEBUG,"Scheduling key %s for saving",key->ptr);
513 dk = zmalloc(sizeof(*dk));
514 dk->db = db;
515 dk->key = key;
516 incrRefCount(key);
517 dk->ctime = time(NULL);
518 listAddNodeTail(server.cache_flush_queue, dk);
519 }
520
521 void cacheCron(void) {
522 time_t now = time(NULL);
523 listNode *ln;
524
525 /* Sync stuff on disk */
526 while((ln = listFirst(server.cache_flush_queue)) != NULL) {
527 dirtykey *dk = ln->value;
528
529 if ((now - dk->ctime) >= server.cache_flush_delay) {
530 struct dictEntry *de;
531 robj *val;
532
533 redisLog(REDIS_DEBUG,"Creating IO Job to save key %s",dk->key->ptr);
534
535 /* Lookup the key, in order to put the current value in the IO
536 * Job and mark ti as DS_SAVING.
537 * Otherwise if the key does not exists we schedule a disk store
538 * delete operation, setting the value to NULL. */
539 de = dictFind(dk->db->dict,dk->key->ptr);
540 if (de) {
541 val = dictGetEntryVal(de);
542 redisAssert(val->storage == REDIS_DS_DIRTY);
543 val->storage = REDIS_DS_SAVING;
544 } else {
545 /* Setting the value to NULL tells the IO thread to delete
546 * the key on disk. */
547 val = NULL;
548 }
549 dsCreateIOJob(REDIS_IOJOB_SAVE,dk->db,dk->key,val);
550 listDelNode(server.cache_flush_queue,ln);
551 decrRefCount(dk->key);
552 zfree(dk);
553 } else {
554 break; /* too early */
555 }
556 }
557
558 /* Reclaim memory from the object cache */
559 while (server.ds_enabled && zmalloc_used_memory() >
560 server.cache_max_memory)
561 {
562 if (cacheFreeOneEntry() == REDIS_ERR) break;
563 }
564 }
565
566 /* ============ Negative caching for diskstore objects ====================== */
567 /* Since accesses to keys that don't exist with disk store cost us a disk
568 * access, we need to cache names of keys that do not exist but are frequently
569 * accessed. */
570 int cacheKeyMayExist(redisDb *db, robj *key) {
571 /* FIXME: for now we just always return true. */
572 return 1;
573 }
574
575 /* ============ Virtual Memory - Blocking clients on missing keys =========== */
576
577 /* This function makes the clinet 'c' waiting for the key 'key' to be loaded.
578 * If the key is already in memory we don't need to block, regardless
579 * of the storage of the value object for this key:
580 *
581 * - If it's REDIS_DS_MEMORY we have the key in memory.
582 * - If it's REDIS_DS_DIRTY they key was modified, but still in memory.
583 * - if it's REDIS_DS_SAVING the key is being saved by an IO Job. When
584 * the client will lookup the key it will block if the key is still
585 * in this stage but it's more or less the best we can do.
586 *
587 * FIXME: we should try if it's actually better to suspend the client
588 * accessing an object that is being saved, and awake it only when
589 * the saving was completed.
590 *
591 * Otherwise if the key is not in memory, we block the client and start
592 * an IO Job to load it:
593 *
594 * the key is added to the io_keys list in the client structure, and also
595 * in the hash table mapping swapped keys to waiting clients, that is,
596 * server.io_waited_keys. */
597 int waitForSwappedKey(redisClient *c, robj *key) {
598 struct dictEntry *de;
599 list *l;
600
601 /* Return ASAP if the key is in memory */
602 de = dictFind(c->db->dict,key->ptr);
603 if (de != NULL) return 0;
604
605 /* Add the key to the list of keys this client is waiting for.
606 * This maps clients to keys they are waiting for. */
607 listAddNodeTail(c->io_keys,key);
608 incrRefCount(key);
609
610 /* Add the client to the swapped keys => clients waiting map. */
611 de = dictFind(c->db->io_keys,key);
612 if (de == NULL) {
613 int retval;
614
615 /* For every key we take a list of clients blocked for it */
616 l = listCreate();
617 retval = dictAdd(c->db->io_keys,key,l);
618 incrRefCount(key);
619 redisAssert(retval == DICT_OK);
620 } else {
621 l = dictGetEntryVal(de);
622 }
623 listAddNodeTail(l,c);
624
625 /* Are we already loading the key from disk? If not create a job */
626 /* FIXME: if a given client was blocked for this key (so job already
627 * created) but the client was freed, there may be a job loading this
628 * key even if de == NULL. Does this creates some race condition?
629 *
630 * Example: after the first load the key gets a DEL that will schedule
631 * a write. But the write will happen later, the duplicated load will
632 * fire and we'll get again the key in memory. */
633 if (de == NULL)
634 dsCreateIOJob(REDIS_IOJOB_LOAD,c->db,key,NULL);
635 return 1;
636 }
637
638 /* Preload keys for any command with first, last and step values for
639 * the command keys prototype, as defined in the command table. */
640 void waitForMultipleSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) {
641 int j, last;
642 if (cmd->vm_firstkey == 0) return;
643 last = cmd->vm_lastkey;
644 if (last < 0) last = argc+last;
645 for (j = cmd->vm_firstkey; j <= last; j += cmd->vm_keystep) {
646 redisAssert(j < argc);
647 waitForSwappedKey(c,argv[j]);
648 }
649 }
650
651 /* Preload keys needed for the ZUNIONSTORE and ZINTERSTORE commands.
652 * Note that the number of keys to preload is user-defined, so we need to
653 * apply a sanity check against argc. */
654 void zunionInterBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) {
655 int i, num;
656 REDIS_NOTUSED(cmd);
657
658 num = atoi(argv[2]->ptr);
659 if (num > (argc-3)) return;
660 for (i = 0; i < num; i++) {
661 waitForSwappedKey(c,argv[3+i]);
662 }
663 }
664
665 /* Preload keys needed to execute the entire MULTI/EXEC block.
666 *
667 * This function is called by blockClientOnSwappedKeys when EXEC is issued,
668 * and will block the client when any command requires a swapped out value. */
669 void execBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) {
670 int i, margc;
671 struct redisCommand *mcmd;
672 robj **margv;
673 REDIS_NOTUSED(cmd);
674 REDIS_NOTUSED(argc);
675 REDIS_NOTUSED(argv);
676
677 if (!(c->flags & REDIS_MULTI)) return;
678 for (i = 0; i < c->mstate.count; i++) {
679 mcmd = c->mstate.commands[i].cmd;
680 margc = c->mstate.commands[i].argc;
681 margv = c->mstate.commands[i].argv;
682
683 if (mcmd->vm_preload_proc != NULL) {
684 mcmd->vm_preload_proc(c,mcmd,margc,margv);
685 } else {
686 waitForMultipleSwappedKeys(c,mcmd,margc,margv);
687 }
688 }
689 }
690
691 /* Is this client attempting to run a command against swapped keys?
692 * If so, block it ASAP, load the keys in background, then resume it.
693 *
694 * The important idea about this function is that it can fail! If keys will
695 * still be swapped when the client is resumed, this key lookups will
696 * just block loading keys from disk. In practical terms this should only
697 * happen with SORT BY command or if there is a bug in this function.
698 *
699 * Return 1 if the client is marked as blocked, 0 if the client can
700 * continue as the keys it is going to access appear to be in memory. */
701 int blockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd) {
702 if (cmd->vm_preload_proc != NULL) {
703 cmd->vm_preload_proc(c,cmd,c->argc,c->argv);
704 } else {
705 waitForMultipleSwappedKeys(c,cmd,c->argc,c->argv);
706 }
707
708 /* If the client was blocked for at least one key, mark it as blocked. */
709 if (listLength(c->io_keys)) {
710 c->flags |= REDIS_IO_WAIT;
711 aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
712 server.cache_blocked_clients++;
713 return 1;
714 } else {
715 return 0;
716 }
717 }
718
719 /* Remove the 'key' from the list of blocked keys for a given client.
720 *
721 * The function returns 1 when there are no longer blocking keys after
722 * the current one was removed (and the client can be unblocked). */
723 int dontWaitForSwappedKey(redisClient *c, robj *key) {
724 list *l;
725 listNode *ln;
726 listIter li;
727 struct dictEntry *de;
728
729 /* The key object might be destroyed when deleted from the c->io_keys
730 * list (and the "key" argument is physically the same object as the
731 * object inside the list), so we need to protect it. */
732 incrRefCount(key);
733
734 /* Remove the key from the list of keys this client is waiting for. */
735 listRewind(c->io_keys,&li);
736 while ((ln = listNext(&li)) != NULL) {
737 if (equalStringObjects(ln->value,key)) {
738 listDelNode(c->io_keys,ln);
739 break;
740 }
741 }
742 redisAssert(ln != NULL);
743
744 /* Remove the client form the key => waiting clients map. */
745 de = dictFind(c->db->io_keys,key);
746 redisAssert(de != NULL);
747 l = dictGetEntryVal(de);
748 ln = listSearchKey(l,c);
749 redisAssert(ln != NULL);
750 listDelNode(l,ln);
751 if (listLength(l) == 0)
752 dictDelete(c->db->io_keys,key);
753
754 decrRefCount(key);
755 return listLength(c->io_keys) == 0;
756 }
757
758 /* Every time we now a key was loaded back in memory, we handle clients
759 * waiting for this key if any. */
760 void handleClientsBlockedOnSwappedKey(redisDb *db, robj *key) {
761 struct dictEntry *de;
762 list *l;
763 listNode *ln;
764 int len;
765
766 de = dictFind(db->io_keys,key);
767 if (!de) return;
768
769 l = dictGetEntryVal(de);
770 len = listLength(l);
771 /* Note: we can't use something like while(listLength(l)) as the list
772 * can be freed by the calling function when we remove the last element. */
773 while (len--) {
774 ln = listFirst(l);
775 redisClient *c = ln->value;
776
777 if (dontWaitForSwappedKey(c,key)) {
778 /* Put the client in the list of clients ready to go as we
779 * loaded all the keys about it. */
780 listAddNodeTail(server.io_ready_clients,c);
781 }
782 }
783 }