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