dscache.c 25.3 KB
Newer Older
1 2 3 4 5 6 7
#include "redis.h"

#include <fcntl.h>
#include <pthread.h>
#include <math.h>
#include <signal.h>

8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28
/* dscache.c - Disk store cache for disk store backend.
 *
 * When Redis is configured for using disk as backend instead of memory, the
 * memory is used as a cache, so that recently accessed keys are taken in
 * memory for fast read and write operations.
 *
 * Modified keys are marked to be flushed on disk, and will be flushed
 * as long as the maxium configured flush time elapsed.
 *
 * This file implements the whole caching subsystem and contains further
 * documentation. */

/* TODO:
 *
 * - The WATCH helper will be used to signal the cache system
 *   we need to flush a given key/dbid into disk, adding this key/dbid
 *   pair into a server.ds_cache_dirty linked list AND hash table (so that we
 *   don't add the same thing multiple times).
 *
 * - cron() checks if there are elements on this list. When there are things
 *   to flush, we create an IO Job for the I/O thread.
29 30 31
 *   NOTE: We disalbe object sharing when server.ds_enabled == 1 so objects
 *   that are referenced an IO job for flushing on disk are marked as
 *   o->storage == REDIS_DS_SAVING.
32 33
 *
 * - This is what we do on key lookup:
34 35
 *   1) The key already exists in memory. object->storage == REDIS_DS_MEMORY
 *      or it is object->storage == REDIS_DS_DIRTY:
36 37
 *      We don't do nothing special, lookup, return value object pointer.
 *   2) The key is in memory but object->storage == REDIS_DS_SAVING.
38 39
 *      When this happens we block waiting for the I/O thread to process
 *      this object. Then continue.
40 41 42 43 44 45 46 47 48 49
 *   3) The key is not in memory. We block to load the key from disk.
 *      Of course the key may not be present at all on the disk store as well,
 *      in such case we just detect this condition and continue, returning
 *      NULL from lookup.
 *
 * - Preloading of needed keys:
 *   1) As it was done with VM, also with this new system we try preloading
 *      keys a client is going to use. We block the client, load keys
 *      using the I/O thread, unblock the client. Same code as VM more or less.
 *
50 51 52 53
 * - Reclaiming memory.
 *   In cron() we detect our memory limit was reached. What we
 *   do is deleting keys that are REDIS_DS_MEMORY, using LRU.
 *
54 55
 *   If this is not enough to return again under the memory limits we also
 *   start to flush keys that need to be synched on disk synchronously,
56 57
 *   removing it from the memory. We do this blocking as memory limit is a
 *   much "harder" barrirer in the new design.
58 59
 *
 * - IO thread operations are no longer stopped for sync loading/saving of
60 61
 *   things. When a key is found to be in the process of being saved
 *   we simply wait for the IO thread to end its work.
62 63 64
 *
 *   Otherwise if there is to load a key without any IO thread operation
 *   just started it is blocking-loaded in the lookup function.
65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86
 *
 * - What happens when an object is destroyed?
 *
 *   If o->storage == REDIS_DS_MEMORY then we simply destory the object.
 *   If o->storage == REDIS_DS_DIRTY we can still remove the object. It had
 *                    changes not flushed on disk, but is being removed so
 *                    who cares.
 *   if o->storage == REDIS_DS_SAVING then the object is being saved so
 *                    it is impossible that its refcount == 1, must be at
 *                    least two. When the object is saved the storage will
 *                    be set back to DS_MEMORY.
 *
 * - What happens when keys are deleted?
 *
 *   We simply schedule a key flush operation as usually, but when the
 *   IO thread will be created the object pointer will be set to NULL
 *   so the IO thread will know that the work to do is to delete the key
 *   from the disk store.
 *
 * - What happens with MULTI/EXEC?
 *
 *   Good question.
87 88 89
 *
 * - If dsSet() fails on the write thread log the error and reschedule the
 *   key for flush.
90 91
 */

92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112
/* Virtual Memory is composed mainly of two subsystems:
 * - Blocking Virutal Memory
 * - Threaded Virtual Memory I/O
 * The two parts are not fully decoupled, but functions are split among two
 * different sections of the source code (delimited by comments) in order to
 * make more clear what functionality is about the blocking VM and what about
 * the threaded (not blocking) VM.
 *
 * Redis VM design:
 *
 * Redis VM is a blocking VM (one that blocks reading swapped values from
 * disk into memory when a value swapped out is needed in memory) that is made
 * unblocking by trying to examine the command argument vector in order to
 * load in background values that will likely be needed in order to exec
 * the command. The command is executed only once all the relevant keys
 * are loaded into memory.
 *
 * This basically is almost as simple of a blocking VM, but almost as parallel
 * as a fully non-blocking VM.
 */

A
antirez 已提交
113 114
void spawnIOThread(void);

115 116
/* =================== Virtual Memory - Blocking Side  ====================== */

117
void dsInit(void) {
118 119 120
    int pipefds[2];
    size_t stacksize;

121
    zmalloc_enable_thread_safeness(); /* we need thread safe zmalloc() */
122

A
antirez 已提交
123
    redisLog(REDIS_NOTICE,"Opening Disk Store: %s", server.ds_path);
124 125 126
    /* Open Disk Store */
    if (dsOpen() != REDIS_OK) {
        redisLog(REDIS_WARNING,"Fatal error opening disk store. Exiting.");
127
        exit(1);
128
    };
129

130
    /* Initialize threaded I/O for Object Cache */
131 132 133 134 135 136 137
    server.io_newjobs = listCreate();
    server.io_processing = listCreate();
    server.io_processed = listCreate();
    server.io_ready_clients = listCreate();
    pthread_mutex_init(&server.io_mutex,NULL);
    server.io_active_threads = 0;
    if (pipe(pipefds) == -1) {
138
        redisLog(REDIS_WARNING,"Unable to intialized DS: pipe(2): %s. Exiting."
139 140 141 142 143 144 145 146 147
            ,strerror(errno));
        exit(1);
    }
    server.io_ready_pipe_read = pipefds[0];
    server.io_ready_pipe_write = pipefds[1];
    redisAssert(anetNonBlock(NULL,server.io_ready_pipe_read) != ANET_ERR);
    /* LZF requires a lot of stack */
    pthread_attr_init(&server.io_threads_attr);
    pthread_attr_getstacksize(&server.io_threads_attr, &stacksize);
148 149 150 151 152

    /* Solaris may report a stacksize of 0, let's set it to 1 otherwise
     * multiplying it by 2 in the while loop later will not really help ;) */
    if (!stacksize) stacksize = 1;

153 154 155 156 157 158 159
    while (stacksize < REDIS_THREAD_STACK_SIZE) stacksize *= 2;
    pthread_attr_setstacksize(&server.io_threads_attr, stacksize);
    /* Listen for events in the threaded I/O pipe */
    if (aeCreateFileEvent(server.el, server.io_ready_pipe_read, AE_READABLE,
        vmThreadedIOCompletedJob, NULL) == AE_ERR)
        oom("creating file event");

160 161
    /* Spawn our I/O thread */
    spawnIOThread();
162 163
}

164 165
/* Compute how good candidate the specified object is for eviction.
 * An higher number means a better candidate. */
166 167 168
double computeObjectSwappability(robj *o) {
    /* actual age can be >= minage, but not < minage. As we use wrapping
     * 21 bit clocks with minutes resolution for the LRU. */
169
    return (double) estimateObjectIdleTime(o);
170 171
}

172 173
/* Try to free one entry from the diskstore object cache */
int cacheFreeOneEntry(void) {
174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201
    int j, i;
    struct dictEntry *best = NULL;
    double best_swappability = 0;
    redisDb *best_db = NULL;
    robj *val;
    sds key;

    for (j = 0; j < server.dbnum; j++) {
        redisDb *db = server.db+j;
        /* Why maxtries is set to 100?
         * Because this way (usually) we'll find 1 object even if just 1% - 2%
         * are swappable objects */
        int maxtries = 100;

        if (dictSize(db->dict) == 0) continue;
        for (i = 0; i < 5; i++) {
            dictEntry *de;
            double swappability;

            if (maxtries) maxtries--;
            de = dictGetRandomKey(db->dict);
            val = dictGetEntryVal(de);
            /* Only swap objects that are currently in memory.
             *
             * Also don't swap shared objects: not a good idea in general and
             * we need to ensure that the main thread does not touch the
             * object while the I/O thread is using it, but we can't
             * control other keys without adding additional mutex. */
202
            if (val->storage != REDIS_DS_MEMORY) {
203 204 205 206 207 208 209 210 211 212 213
                if (maxtries) i--; /* don't count this try */
                continue;
            }
            swappability = computeObjectSwappability(val);
            if (!best || swappability > best_swappability) {
                best = de;
                best_swappability = swappability;
                best_db = db;
            }
        }
    }
214 215 216 217 218 219 220
    if (best == NULL) {
        /* FIXME: If there are objects marked as DS_DIRTY or DS_SAVING
         * let's wait for this objects to be clear and retry...
         *
         * Object cache vm limit is considered an hard limit. */
        return REDIS_ERR;
    }
221 222 223
    key = dictGetEntryKey(best);
    val = dictGetEntryVal(best);

224
    redisLog(REDIS_DEBUG,"Key selected for cache eviction: %s swappability:%f",
225 226
        key, best_swappability);

227 228 229 230 231
    /* Delete this key from memory */
    {
        robj *kobj = createStringObject(key,sdslen(key));
        dbDelete(best_db,kobj);
        decrRefCount(kobj);
232
    }
233
    return REDIS_OK;
234 235 236 237 238
}

/* Return true if it's safe to swap out objects in a given moment.
 * Basically we don't want to swap objects out while there is a BGSAVE
 * or a BGAEOREWRITE running in backgroud. */
239
int dsCanTouchDiskStore(void) {
240 241 242 243 244 245 246
    return (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1);
}

/* =================== Virtual Memory - Threaded I/O  ======================= */

void freeIOJob(iojob *j) {
    decrRefCount(j->key);
247 248
    /* j->val can be NULL if the job is about deleting the key from disk. */
    if (j->val) decrRefCount(j->val);
249 250 251 252 253
    zfree(j);
}

/* Every time a thread finished a Job, it writes a byte into the write side
 * of an unix pipe in order to "awake" the main thread, and this function
A
antirez 已提交
254
 * is called. */
255 256 257 258
void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
            int mask)
{
    char buf[1];
A
antirez 已提交
259
    int retval, processed = 0, toprocess = -1;
260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282
    REDIS_NOTUSED(el);
    REDIS_NOTUSED(mask);
    REDIS_NOTUSED(privdata);

    /* For every byte we read in the read side of the pipe, there is one
     * I/O job completed to process. */
    while((retval = read(fd,buf,1)) == 1) {
        iojob *j;
        listNode *ln;

        redisLog(REDIS_DEBUG,"Processing I/O completed job");

        /* Get the processed element (the oldest one) */
        lockThreadedIO();
        redisAssert(listLength(server.io_processed) != 0);
        if (toprocess == -1) {
            toprocess = (listLength(server.io_processed)*REDIS_MAX_COMPLETED_JOBS_PROCESSED)/100;
            if (toprocess <= 0) toprocess = 1;
        }
        ln = listFirst(server.io_processed);
        j = ln->value;
        listDelNode(server.io_processed,ln);
        unlockThreadedIO();
A
antirez 已提交
283

284 285
        /* Post process it in the main thread, as there are things we
         * can do just here to avoid race conditions and/or invasive locks */
286 287 288
        redisLog(REDIS_DEBUG,"COMPLETED Job type %s, key: %s",
            (j->type == REDIS_IOJOB_LOAD) ? "load" : "save",
            (unsigned char*)j->key->ptr);
289
        if (j->type == REDIS_IOJOB_LOAD) {
290
            /* Create the key-value pair in the in-memory database */
291 292 293
            if (j->val != NULL) {
                dbAdd(j->db,j->key,j->val);
                incrRefCount(j->val);
294
                if (j->expire != -1) setExpire(j->db,j->key,j->expire);
295 296 297 298 299
            } else {
                /* The key does not exist. Create a negative cache entry
                 * for this key. */
                /* FIXME: add this entry into the negative cache */
            }
300 301
            /* Handle clients waiting for this key to be loaded. */
            handleClientsBlockedOnSwappedKey(j->db,j->key);
302
            freeIOJob(j);
A
antirez 已提交
303 304 305
        } else if (j->type == REDIS_IOJOB_SAVE) {
            redisAssert(j->val->storage == REDIS_DS_SAVING);
            j->val->storage = REDIS_DS_MEMORY;
306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337
            freeIOJob(j);
        }
        processed++;
        if (processed == toprocess) return;
    }
    if (retval < 0 && errno != EAGAIN) {
        redisLog(REDIS_WARNING,
            "WARNING: read(2) error in vmThreadedIOCompletedJob() %s",
            strerror(errno));
    }
}

void lockThreadedIO(void) {
    pthread_mutex_lock(&server.io_mutex);
}

void unlockThreadedIO(void) {
    pthread_mutex_unlock(&server.io_mutex);
}

void *IOThreadEntryPoint(void *arg) {
    iojob *j;
    listNode *ln;
    REDIS_NOTUSED(arg);

    pthread_detach(pthread_self());
    while(1) {
        /* Get a new job to process */
        lockThreadedIO();
        if (listLength(server.io_newjobs) == 0) {
            /* No new jobs in queue, exit. */
            unlockThreadedIO();
338 339
            sleep(1);
            continue;
340 341 342 343 344 345 346 347
        }
        ln = listFirst(server.io_newjobs);
        j = ln->value;
        listDelNode(server.io_newjobs,ln);
        /* Add the job in the processing queue */
        listAddNodeTail(server.io_processing,j);
        ln = listLast(server.io_processing); /* We use ln later to remove it */
        unlockThreadedIO();
348 349 350 351
        redisLog(REDIS_DEBUG,"Thread %ld: new job type %s: %p about key '%s'",
            (long) pthread_self(),
            (j->type == REDIS_IOJOB_LOAD) ? "load" : "save",
            (void*)j, (char*)j->key->ptr);
352 353 354

        /* Process the Job */
        if (j->type == REDIS_IOJOB_LOAD) {
355 356 357 358
            time_t expire;

            j->val = dsGet(j->db,j->key,&expire);
            if (j->val) j->expire = expire;
359
        } else if (j->type == REDIS_IOJOB_SAVE) {
360
            redisAssert(j->val->storage == REDIS_DS_SAVING);
361 362 363 364
            if (j->val)
                dsSet(j->db,j->key,j->val);
            else
                dsDel(j->db,j->key);
365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420
        }

        /* Done: insert the job into the processed queue */
        redisLog(REDIS_DEBUG,"Thread %ld completed the job: %p (key %s)",
            (long) pthread_self(), (void*)j, (char*)j->key->ptr);
        lockThreadedIO();
        listDelNode(server.io_processing,ln);
        listAddNodeTail(server.io_processed,j);
        unlockThreadedIO();

        /* Signal the main thread there is new stuff to process */
        redisAssert(write(server.io_ready_pipe_write,"x",1) == 1);
    }
    return NULL; /* never reached */
}

void spawnIOThread(void) {
    pthread_t thread;
    sigset_t mask, omask;
    int err;

    sigemptyset(&mask);
    sigaddset(&mask,SIGCHLD);
    sigaddset(&mask,SIGHUP);
    sigaddset(&mask,SIGPIPE);
    pthread_sigmask(SIG_SETMASK, &mask, &omask);
    while ((err = pthread_create(&thread,&server.io_threads_attr,IOThreadEntryPoint,NULL)) != 0) {
        redisLog(REDIS_WARNING,"Unable to spawn an I/O thread: %s",
            strerror(err));
        usleep(1000000);
    }
    pthread_sigmask(SIG_SETMASK, &omask, NULL);
    server.io_active_threads++;
}

/* We need to wait for the last thread to exit before we are able to
 * fork() in order to BGSAVE or BGREWRITEAOF. */
void waitEmptyIOJobsQueue(void) {
    while(1) {
        int io_processed_len;

        lockThreadedIO();
        if (listLength(server.io_newjobs) == 0 &&
            listLength(server.io_processing) == 0 &&
            server.io_active_threads == 0)
        {
            unlockThreadedIO();
            return;
        }
        /* While waiting for empty jobs queue condition we post-process some
         * finshed job, as I/O threads may be hanging trying to write against
         * the io_ready_pipe_write FD but there are so much pending jobs that
         * it's blocking. */
        io_processed_len = listLength(server.io_processed);
        unlockThreadedIO();
        if (io_processed_len) {
421 422
            vmThreadedIOCompletedJob(NULL,server.io_ready_pipe_read,
                                                        (void*)0xdeadbeef,0);
423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438
            usleep(1000); /* 1 millisecond */
        } else {
            usleep(10000); /* 10 milliseconds */
        }
    }
}

/* This function must be called while with threaded IO locked */
void queueIOJob(iojob *j) {
    redisLog(REDIS_DEBUG,"Queued IO Job %p type %d about key '%s'\n",
        (void*)j, j->type, (char*)j->key->ptr);
    listAddNodeTail(server.io_newjobs,j);
    if (server.io_active_threads < server.vm_max_threads)
        spawnIOThread();
}

439
void dsCreateIOJob(int type, redisDb *db, robj *key, robj *val) {
440 441 442
    iojob *j;

    j = zmalloc(sizeof(*j));
443
    j->type = type;
444 445 446
    j->db = db;
    j->key = key;
    incrRefCount(key);
447
    j->val = val;
448
    if (val) incrRefCount(val);
449 450 451 452 453 454

    lockThreadedIO();
    queueIOJob(j);
    unlockThreadedIO();
}

455 456 457 458 459 460 461 462 463 464 465 466 467
void cacheScheduleForFlush(redisDb *db, robj *key) {
    dirtykey *dk;
    dictEntry *de;
    
    de = dictFind(db->dict,key->ptr);
    if (de) {
        robj *val = dictGetEntryVal(de);
        if (val->storage == REDIS_DS_DIRTY)
            return;
        else
            val->storage = REDIS_DS_DIRTY;
    }

A
antirez 已提交
468
    redisLog(REDIS_DEBUG,"Scheduling key %s for saving",key->ptr);
469 470 471 472 473
    dk = zmalloc(sizeof(*dk));
    dk->db = db;
    dk->key = key;
    incrRefCount(key);
    dk->ctime = time(NULL);
A
antirez 已提交
474
    listAddNodeTail(server.cache_flush_queue, dk);
475 476 477 478 479 480 481 482 483 484 485 486 487 488
}

void cacheCron(void) {
    time_t now = time(NULL);
    listNode *ln;

    /* Sync stuff on disk */
    while((ln = listFirst(server.cache_flush_queue)) != NULL) {
        dirtykey *dk = ln->value;

        if ((now - dk->ctime) >= server.cache_flush_delay) {
            struct dictEntry *de;
            robj *val;

A
antirez 已提交
489 490
            redisLog(REDIS_DEBUG,"Creating IO Job to save key %s",dk->key->ptr);

A
antirez 已提交
491 492 493 494
            /* Lookup the key, in order to put the current value in the IO
             * Job and mark ti as DS_SAVING.
             * Otherwise if the key does not exists we schedule a disk store
             * delete operation, setting the value to NULL. */
495 496 497 498 499 500 501 502 503 504 505 506
            de = dictFind(dk->db->dict,dk->key->ptr);
            if (de) {
                val = dictGetEntryVal(de);
                redisAssert(val->storage == REDIS_DS_DIRTY);
                val->storage = REDIS_DS_SAVING;
            } else {
                /* Setting the value to NULL tells the IO thread to delete
                 * the key on disk. */
                val = NULL;
            }
            dsCreateIOJob(REDIS_IOJOB_SAVE,dk->db,dk->key,val);
            listDelNode(server.cache_flush_queue,ln);
A
antirez 已提交
507 508
            decrRefCount(dk->key);
            zfree(dk);
509 510 511 512 513 514 515 516 517 518 519 520 521
        } else {
            break; /* too early */
        }
    }

    /* Reclaim memory from the object cache */
    while (server.ds_enabled && zmalloc_used_memory() >
            server.cache_max_memory)
    {
        if (cacheFreeOneEntry() == REDIS_ERR) break;
    }
}

522 523 524
/* ============ Virtual Memory - Blocking clients on missing keys =========== */

/* This function makes the clinet 'c' waiting for the key 'key' to be loaded.
525 526 527 528 529 530 531 532
 * If the key is already in memory we don't need to block, regardless
 * of the storage of the value object for this key:
 *
 * - If it's REDIS_DS_MEMORY we have the key in memory.
 * - If it's REDIS_DS_DIRTY they key was modified, but still in memory.
 * - if it's REDIS_DS_SAVING the key is being saved by an IO Job. When
 *   the client will lookup the key it will block if the key is still
 *   in this stage but it's more or less the best we can do.
533
 *
534 535 536 537 538 539 540 541
 *   FIXME: we should try if it's actually better to suspend the client
 *   accessing an object that is being saved, and awake it only when
 *   the saving was completed.
 *
 * Otherwise if the key is not in memory, we block the client and start
 * an IO Job to load it:
 *
 * the key is added to the io_keys list in the client structure, and also
542 543 544 545 546 547
 * in the hash table mapping swapped keys to waiting clients, that is,
 * server.io_waited_keys. */
int waitForSwappedKey(redisClient *c, robj *key) {
    struct dictEntry *de;
    list *l;

548
    /* Return ASAP if the key is in memory */
549
    de = dictFind(c->db->dict,key->ptr);
550
    if (de != NULL) return 0;
551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572

    /* Add the key to the list of keys this client is waiting for.
     * This maps clients to keys they are waiting for. */
    listAddNodeTail(c->io_keys,key);
    incrRefCount(key);

    /* Add the client to the swapped keys => clients waiting map. */
    de = dictFind(c->db->io_keys,key);
    if (de == NULL) {
        int retval;

        /* For every key we take a list of clients blocked for it */
        l = listCreate();
        retval = dictAdd(c->db->io_keys,key,l);
        incrRefCount(key);
        redisAssert(retval == DICT_OK);
    } else {
        l = dictGetEntryVal(de);
    }
    listAddNodeTail(l,c);

    /* Are we already loading the key from disk? If not create a job */
573 574
    if (de == NULL)
        dsCreateIOJob(REDIS_IOJOB_LOAD,c->db,key,NULL);
575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651
    return 1;
}

/* Preload keys for any command with first, last and step values for
 * the command keys prototype, as defined in the command table. */
void waitForMultipleSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) {
    int j, last;
    if (cmd->vm_firstkey == 0) return;
    last = cmd->vm_lastkey;
    if (last < 0) last = argc+last;
    for (j = cmd->vm_firstkey; j <= last; j += cmd->vm_keystep) {
        redisAssert(j < argc);
        waitForSwappedKey(c,argv[j]);
    }
}

/* Preload keys needed for the ZUNIONSTORE and ZINTERSTORE commands.
 * Note that the number of keys to preload is user-defined, so we need to
 * apply a sanity check against argc. */
void zunionInterBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) {
    int i, num;
    REDIS_NOTUSED(cmd);

    num = atoi(argv[2]->ptr);
    if (num > (argc-3)) return;
    for (i = 0; i < num; i++) {
        waitForSwappedKey(c,argv[3+i]);
    }
}

/* Preload keys needed to execute the entire MULTI/EXEC block.
 *
 * This function is called by blockClientOnSwappedKeys when EXEC is issued,
 * and will block the client when any command requires a swapped out value. */
void execBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) {
    int i, margc;
    struct redisCommand *mcmd;
    robj **margv;
    REDIS_NOTUSED(cmd);
    REDIS_NOTUSED(argc);
    REDIS_NOTUSED(argv);

    if (!(c->flags & REDIS_MULTI)) return;
    for (i = 0; i < c->mstate.count; i++) {
        mcmd = c->mstate.commands[i].cmd;
        margc = c->mstate.commands[i].argc;
        margv = c->mstate.commands[i].argv;

        if (mcmd->vm_preload_proc != NULL) {
            mcmd->vm_preload_proc(c,mcmd,margc,margv);
        } else {
            waitForMultipleSwappedKeys(c,mcmd,margc,margv);
        }
    }
}

/* Is this client attempting to run a command against swapped keys?
 * If so, block it ASAP, load the keys in background, then resume it.
 *
 * The important idea about this function is that it can fail! If keys will
 * still be swapped when the client is resumed, this key lookups will
 * just block loading keys from disk. In practical terms this should only
 * happen with SORT BY command or if there is a bug in this function.
 *
 * Return 1 if the client is marked as blocked, 0 if the client can
 * continue as the keys it is going to access appear to be in memory. */
int blockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd) {
    if (cmd->vm_preload_proc != NULL) {
        cmd->vm_preload_proc(c,cmd,c->argc,c->argv);
    } else {
        waitForMultipleSwappedKeys(c,cmd,c->argc,c->argv);
    }

    /* If the client was blocked for at least one key, mark it as blocked. */
    if (listLength(c->io_keys)) {
        c->flags |= REDIS_IO_WAIT;
        aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
652
        server.cache_blocked_clients++;
653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668
        return 1;
    } else {
        return 0;
    }
}

/* Remove the 'key' from the list of blocked keys for a given client.
 *
 * The function returns 1 when there are no longer blocking keys after
 * the current one was removed (and the client can be unblocked). */
int dontWaitForSwappedKey(redisClient *c, robj *key) {
    list *l;
    listNode *ln;
    listIter li;
    struct dictEntry *de;

669 670 671 672 673
    /* The key object might be destroyed when deleted from the c->io_keys
     * list (and the "key" argument is physically the same object as the
     * object inside the list), so we need to protect it. */
    incrRefCount(key);

674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693
    /* Remove the key from the list of keys this client is waiting for. */
    listRewind(c->io_keys,&li);
    while ((ln = listNext(&li)) != NULL) {
        if (equalStringObjects(ln->value,key)) {
            listDelNode(c->io_keys,ln);
            break;
        }
    }
    redisAssert(ln != NULL);

    /* Remove the client form the key => waiting clients map. */
    de = dictFind(c->db->io_keys,key);
    redisAssert(de != NULL);
    l = dictGetEntryVal(de);
    ln = listSearchKey(l,c);
    redisAssert(ln != NULL);
    listDelNode(l,ln);
    if (listLength(l) == 0)
        dictDelete(c->db->io_keys,key);

694
    decrRefCount(key);
695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723
    return listLength(c->io_keys) == 0;
}

/* Every time we now a key was loaded back in memory, we handle clients
 * waiting for this key if any. */
void handleClientsBlockedOnSwappedKey(redisDb *db, robj *key) {
    struct dictEntry *de;
    list *l;
    listNode *ln;
    int len;

    de = dictFind(db->io_keys,key);
    if (!de) return;

    l = dictGetEntryVal(de);
    len = listLength(l);
    /* Note: we can't use something like while(listLength(l)) as the list
     * can be freed by the calling function when we remove the last element. */
    while (len--) {
        ln = listFirst(l);
        redisClient *c = ln->value;

        if (dontWaitForSwappedKey(c,key)) {
            /* Put the client in the list of clients ready to go as we
             * loaded all the keys about it. */
            listAddNodeTail(server.io_ready_clients,c);
        }
    }
}