]> git.saurik.com Git - redis.git/blob - src/dscache.c
f1ffe49123f2f01e292e0195d0e91396cecf4e83
[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
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.
99 *
100 * Redis VM design:
101 *
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.
108 *
109 * This basically is almost as simple of a blocking VM, but almost as parallel
110 * as a fully non-blocking VM.
111 */
112
113 void spawnIOThread(void);
114
115 /* =================== Virtual Memory - Blocking Side ====================== */
116
117 void dsInit(void) {
118 int pipefds[2];
119 size_t stacksize;
120
121 zmalloc_enable_thread_safeness(); /* we need thread safe zmalloc() */
122
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.");
127 exit(1);
128 };
129
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."
139 ,strerror(errno));
140 exit(1);
141 }
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);
148
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;
152
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");
159
160 /* Spawn our I/O thread */
161 spawnIOThread();
162 }
163
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);
170 }
171
172 /* Try to free one entry from the diskstore object cache */
173 int cacheFreeOneEntry(void) {
174 int j, i;
175 struct dictEntry *best = NULL;
176 double best_swappability = 0;
177 redisDb *best_db = NULL;
178 robj *val;
179 sds key;
180
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 */
186 int maxtries = 100;
187
188 if (dictSize(db->dict) == 0) continue;
189 for (i = 0; i < 5; i++) {
190 dictEntry *de;
191 double swappability;
192
193 if (maxtries) maxtries--;
194 de = dictGetRandomKey(db->dict);
195 val = dictGetEntryVal(de);
196 /* Only swap objects that are currently in memory.
197 *
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 */
204 continue;
205 }
206 swappability = computeObjectSwappability(val);
207 if (!best || swappability > best_swappability) {
208 best = de;
209 best_swappability = swappability;
210 best_db = db;
211 }
212 }
213 }
214 if (best == NULL) {
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...
217 *
218 * Object cache vm limit is considered an hard limit. */
219 return REDIS_ERR;
220 }
221 key = dictGetEntryKey(best);
222 val = dictGetEntryVal(best);
223
224 redisLog(REDIS_DEBUG,"Key selected for cache eviction: %s swappability:%f",
225 key, best_swappability);
226
227 /* Delete this key from memory */
228 {
229 robj *kobj = createStringObject(key,sdslen(key));
230 dbDelete(best_db,kobj);
231 decrRefCount(kobj);
232 }
233 return REDIS_OK;
234 }
235
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);
241 }
242
243 /* =================== Virtual Memory - Threaded I/O ======================= */
244
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);
249 zfree(j);
250 }
251
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
254 * is called. */
255 void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
256 int mask)
257 {
258 char buf[1];
259 int retval, processed = 0, toprocess = -1;
260 REDIS_NOTUSED(el);
261 REDIS_NOTUSED(mask);
262 REDIS_NOTUSED(privdata);
263
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) {
267 iojob *j;
268 listNode *ln;
269
270 redisLog(REDIS_DEBUG,"Processing I/O completed job");
271
272 /* Get the processed element (the oldest one) */
273 lockThreadedIO();
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;
278 }
279 ln = listFirst(server.io_processed);
280 j = ln->value;
281 listDelNode(server.io_processed,ln);
282 unlockThreadedIO();
283
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);
295 } else {
296 /* The key does not exist. Create a negative cache entry
297 * for this key. */
298 /* FIXME: add this entry into the negative cache */
299 }
300 /* Handle clients waiting for this key to be loaded. */
301 handleClientsBlockedOnSwappedKey(j->db,j->key);
302 freeIOJob(j);
303 } else if (j->type == REDIS_IOJOB_SAVE) {
304 redisAssert(j->val->storage == REDIS_DS_SAVING);
305 j->val->storage = REDIS_DS_MEMORY;
306 freeIOJob(j);
307 }
308 processed++;
309 if (processed == toprocess) return;
310 }
311 if (retval < 0 && errno != EAGAIN) {
312 redisLog(REDIS_WARNING,
313 "WARNING: read(2) error in vmThreadedIOCompletedJob() %s",
314 strerror(errno));
315 }
316 }
317
318 void lockThreadedIO(void) {
319 pthread_mutex_lock(&server.io_mutex);
320 }
321
322 void unlockThreadedIO(void) {
323 pthread_mutex_unlock(&server.io_mutex);
324 }
325
326 void *IOThreadEntryPoint(void *arg) {
327 iojob *j;
328 listNode *ln;
329 REDIS_NOTUSED(arg);
330
331 pthread_detach(pthread_self());
332 while(1) {
333 /* Get a new job to process */
334 lockThreadedIO();
335 if (listLength(server.io_newjobs) == 0) {
336 /* No new jobs in queue, exit. */
337 unlockThreadedIO();
338 sleep(1);
339 continue;
340 }
341 ln = listFirst(server.io_newjobs);
342 j = ln->value;
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 */
347 unlockThreadedIO();
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);
352
353 /* Process the Job */
354 if (j->type == REDIS_IOJOB_LOAD) {
355 time_t expire;
356
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);
361 if (j->val)
362 dsSet(j->db,j->key,j->val);
363 else
364 dsDel(j->db,j->key);
365 }
366
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);
370 lockThreadedIO();
371 listDelNode(server.io_processing,ln);
372 listAddNodeTail(server.io_processed,j);
373 unlockThreadedIO();
374
375 /* Signal the main thread there is new stuff to process */
376 redisAssert(write(server.io_ready_pipe_write,"x",1) == 1);
377 }
378 return NULL; /* never reached */
379 }
380
381 void spawnIOThread(void) {
382 pthread_t thread;
383 sigset_t mask, omask;
384 int err;
385
386 sigemptyset(&mask);
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",
393 strerror(err));
394 usleep(1000000);
395 }
396 pthread_sigmask(SIG_SETMASK, &omask, NULL);
397 server.io_active_threads++;
398 }
399
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) {
403 while(1) {
404 int io_processed_len;
405
406 lockThreadedIO();
407 if (listLength(server.io_newjobs) == 0 &&
408 listLength(server.io_processing) == 0 &&
409 server.io_active_threads == 0)
410 {
411 unlockThreadedIO();
412 return;
413 }
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
417 * it's blocking. */
418 io_processed_len = listLength(server.io_processed);
419 unlockThreadedIO();
420 if (io_processed_len) {
421 vmThreadedIOCompletedJob(NULL,server.io_ready_pipe_read,
422 (void*)0xdeadbeef,0);
423 usleep(1000); /* 1 millisecond */
424 } else {
425 usleep(10000); /* 10 milliseconds */
426 }
427 }
428 }
429
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)
436 spawnIOThread();
437 }
438
439 void dsCreateIOJob(int type, redisDb *db, robj *key, robj *val) {
440 iojob *j;
441
442 j = zmalloc(sizeof(*j));
443 j->type = type;
444 j->db = db;
445 j->key = key;
446 incrRefCount(key);
447 j->val = val;
448 if (val) incrRefCount(val);
449
450 lockThreadedIO();
451 queueIOJob(j);
452 unlockThreadedIO();
453 }
454
455 void cacheScheduleForFlush(redisDb *db, robj *key) {
456 dirtykey *dk;
457 dictEntry *de;
458
459 de = dictFind(db->dict,key->ptr);
460 if (de) {
461 robj *val = dictGetEntryVal(de);
462 if (val->storage == REDIS_DS_DIRTY)
463 return;
464 else
465 val->storage = REDIS_DS_DIRTY;
466 }
467
468 redisLog(REDIS_DEBUG,"Scheduling key %s for saving",key->ptr);
469 dk = zmalloc(sizeof(*dk));
470 dk->db = db;
471 dk->key = key;
472 incrRefCount(key);
473 dk->ctime = time(NULL);
474 listAddNodeTail(server.cache_flush_queue, key);
475 }
476
477 void cacheCron(void) {
478 time_t now = time(NULL);
479 listNode *ln;
480
481 /* Sync stuff on disk */
482 while((ln = listFirst(server.cache_flush_queue)) != NULL) {
483 dirtykey *dk = ln->value;
484
485 if ((now - dk->ctime) >= server.cache_flush_delay) {
486 struct dictEntry *de;
487 robj *val;
488
489 redisLog(REDIS_DEBUG,"Creating IO Job to save key %s",dk->key->ptr);
490
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);
494 if (de) {
495 val = dictGetEntryVal(de);
496 redisAssert(val->storage == REDIS_DS_DIRTY);
497 val->storage = REDIS_DS_SAVING;
498 } else {
499 /* Setting the value to NULL tells the IO thread to delete
500 * the key on disk. */
501 val = NULL;
502 }
503 dsCreateIOJob(REDIS_IOJOB_SAVE,dk->db,dk->key,val);
504 listDelNode(server.cache_flush_queue,ln);
505 } else {
506 break; /* too early */
507 }
508 }
509
510 /* Reclaim memory from the object cache */
511 while (server.ds_enabled && zmalloc_used_memory() >
512 server.cache_max_memory)
513 {
514 if (cacheFreeOneEntry() == REDIS_ERR) break;
515 }
516 }
517
518 /* ============ Virtual Memory - Blocking clients on missing keys =========== */
519
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:
523 *
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.
529 *
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.
533 *
534 * Otherwise if the key is not in memory, we block the client and start
535 * an IO Job to load it:
536 *
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;
542 list *l;
543
544 /* Return ASAP if the key is in memory */
545 de = dictFind(c->db->dict,key->ptr);
546 if (de != NULL) return 0;
547
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);
551 incrRefCount(key);
552
553 /* Add the client to the swapped keys => clients waiting map. */
554 de = dictFind(c->db->io_keys,key);
555 if (de == NULL) {
556 int retval;
557
558 /* For every key we take a list of clients blocked for it */
559 l = listCreate();
560 retval = dictAdd(c->db->io_keys,key,l);
561 incrRefCount(key);
562 redisAssert(retval == DICT_OK);
563 } else {
564 l = dictGetEntryVal(de);
565 }
566 listAddNodeTail(l,c);
567
568 /* Are we already loading the key from disk? If not create a job */
569 if (de == NULL)
570 dsCreateIOJob(REDIS_IOJOB_LOAD,c->db,key,NULL);
571 return 1;
572 }
573
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) {
577 int j, last;
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]);
584 }
585 }
586
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) {
591 int i, num;
592 REDIS_NOTUSED(cmd);
593
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]);
598 }
599 }
600
601 /* Preload keys needed to execute the entire MULTI/EXEC block.
602 *
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) {
606 int i, margc;
607 struct redisCommand *mcmd;
608 robj **margv;
609 REDIS_NOTUSED(cmd);
610 REDIS_NOTUSED(argc);
611 REDIS_NOTUSED(argv);
612
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;
618
619 if (mcmd->vm_preload_proc != NULL) {
620 mcmd->vm_preload_proc(c,mcmd,margc,margv);
621 } else {
622 waitForMultipleSwappedKeys(c,mcmd,margc,margv);
623 }
624 }
625 }
626
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.
629 *
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.
634 *
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);
640 } else {
641 waitForMultipleSwappedKeys(c,cmd,c->argc,c->argv);
642 }
643
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++;
649 return 1;
650 } else {
651 return 0;
652 }
653 }
654
655 /* Remove the 'key' from the list of blocked keys for a given client.
656 *
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) {
660 list *l;
661 listNode *ln;
662 listIter li;
663 struct dictEntry *de;
664
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. */
668 incrRefCount(key);
669
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);
675 break;
676 }
677 }
678 redisAssert(ln != NULL);
679
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);
686 listDelNode(l,ln);
687 if (listLength(l) == 0)
688 dictDelete(c->db->io_keys,key);
689
690 decrRefCount(key);
691 return listLength(c->io_keys) == 0;
692 }
693
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;
698 list *l;
699 listNode *ln;
700 int len;
701
702 de = dictFind(db->io_keys,key);
703 if (!de) return;
704
705 l = dictGetEntryVal(de);
706 len = listLength(l);
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. */
709 while (len--) {
710 ln = listFirst(l);
711 redisClient *c = ln->value;
712
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);
717 }
718 }
719 }