dscache.c 24.6 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 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109
/* 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 已提交
110 111
void spawnIOThread(void);

112 113
/* =================== Virtual Memory - Blocking Side  ====================== */

114
void dsInit(void) {
115 116 117
    int pipefds[2];
    size_t stacksize;

118
    zmalloc_enable_thread_safeness(); /* we need thread safe zmalloc() */
119

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

127
    /* Initialize threaded I/O for Object Cache */
128 129 130 131 132 133 134
    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) {
135
        redisLog(REDIS_WARNING,"Unable to intialized DS: pipe(2): %s. Exiting."
136 137 138 139 140 141 142 143 144
            ,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);
145 146 147 148 149

    /* 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;

150 151 152 153 154 155 156
    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");

157 158
    /* Spawn our I/O thread */
    spawnIOThread();
159 160
}

161 162
/* Compute how good candidate the specified object is for eviction.
 * An higher number means a better candidate. */
163 164 165
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. */
166
    return (double) estimateObjectIdleTime(o);
167 168
}

169 170
/* Try to free one entry from the diskstore object cache */
int cacheFreeOneEntry(void) {
171 172 173 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
    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. */
199
            if (val->storage != REDIS_DS_MEMORY) {
200 201 202 203 204 205 206 207 208 209 210
                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;
            }
        }
    }
211 212 213 214 215 216 217
    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;
    }
218 219 220
    key = dictGetEntryKey(best);
    val = dictGetEntryVal(best);

221
    redisLog(REDIS_DEBUG,"Key selected for cache eviction: %s swappability:%f",
222 223
        key, best_swappability);

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

/* 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. */
236
int dsCanTouchDiskStore(void) {
237 238 239 240 241 242 243
    return (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1);
}

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

void freeIOJob(iojob *j) {
    decrRefCount(j->key);
244 245
    /* j->val can be NULL if the job is about deleting the key from disk. */
    if (j->val) decrRefCount(j->val);
246 247 248 249 250
    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 已提交
251
 * is called. */
252 253 254 255
void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
            int mask)
{
    char buf[1];
A
antirez 已提交
256
    int retval, processed = 0, toprocess = -1;
257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279
    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 已提交
280

281 282
        /* Post process it in the main thread, as there are things we
         * can do just here to avoid race conditions and/or invasive locks */
283 284 285
        redisLog(REDIS_DEBUG,"COMPLETED Job type %s, key: %s",
            (j->type == REDIS_IOJOB_LOAD) ? "load" : "save",
            (unsigned char*)j->key->ptr);
286
        if (j->type == REDIS_IOJOB_LOAD) {
287
            /* Create the key-value pair in the in-memory database */
A
antirez 已提交
288
            dbAdd(j->db,j->key,j->val);
289
            incrRefCount(j->val);
290 291
            /* Handle clients waiting for this key to be loaded. */
            handleClientsBlockedOnSwappedKey(j->db,j->key);
292
            freeIOJob(j);
A
antirez 已提交
293 294 295
        } else if (j->type == REDIS_IOJOB_SAVE) {
            redisAssert(j->val->storage == REDIS_DS_SAVING);
            j->val->storage = REDIS_DS_MEMORY;
296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327
            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();
328 329
            sleep(1);
            continue;
330 331 332 333 334 335 336 337
        }
        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();
338 339 340 341
        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);
342 343 344

        /* Process the Job */
        if (j->type == REDIS_IOJOB_LOAD) {
345 346 347
            j->val = dsGet(j->db,j->key);
            redisAssert(j->val != NULL);
        } else if (j->type == REDIS_IOJOB_SAVE) {
348
            redisAssert(j->val->storage == REDIS_DS_SAVING);
349 350 351 352
            if (j->val)
                dsSet(j->db,j->key,j->val);
            else
                dsDel(j->db,j->key);
353 354 355 356 357 358 359 360 361 362 363 364 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
        }

        /* 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) {
409 410
            vmThreadedIOCompletedJob(NULL,server.io_ready_pipe_read,
                                                        (void*)0xdeadbeef,0);
411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426
            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();
}

427
void dsCreateIOJob(int type, redisDb *db, robj *key, robj *val) {
428 429 430
    iojob *j;

    j = zmalloc(sizeof(*j));
431
    j->type = type;
432 433 434
    j->db = db;
    j->key = key;
    incrRefCount(key);
435
    j->val = val;
436
    if (val) incrRefCount(val);
437 438 439 440 441 442

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

443 444 445 446 447 448 449 450 451 452 453 454 455
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 已提交
456
    redisLog(REDIS_DEBUG,"Scheduling key %s for saving",key->ptr);
457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476
    dk = zmalloc(sizeof(*dk));
    dk->db = db;
    dk->key = key;
    incrRefCount(key);
    dk->ctime = time(NULL);
    listAddNodeTail(server.cache_flush_queue, key);
}

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 已提交
477 478
            redisLog(REDIS_DEBUG,"Creating IO Job to save key %s",dk->key->ptr);

479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505
            /* Lookup the key. We need to check if it's still here and
             * possibly access to the value. */
            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);
        } 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;
    }
}

506 507 508
/* ============ Virtual Memory - Blocking clients on missing keys =========== */

/* This function makes the clinet 'c' waiting for the key 'key' to be loaded.
509 510 511 512 513 514 515 516
 * 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.
517
 *
518 519 520 521 522 523 524 525
 *   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
526 527 528 529 530 531
 * 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;

532
    /* Return ASAP if the key is in memory */
533
    de = dictFind(c->db->dict,key->ptr);
534
    if (de != NULL) return 0;
535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556

    /* 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 */
557 558
    if (de == NULL)
        dsCreateIOJob(REDIS_IOJOB_LOAD,c->db,key,NULL);
559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 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
    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);
636
        server.cache_blocked_clients++;
637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652
        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;

653 654 655 656 657
    /* 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);

658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677
    /* 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);

678
    decrRefCount(key);
679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707
    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);
        }
    }
}