db.c 40.5 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29
/*
 * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 *   * Redistributions of source code must retain the above copyright notice,
 *     this list of conditions and the following disclaimer.
 *   * Redistributions in binary form must reproduce the above copyright
 *     notice, this list of conditions and the following disclaimer in the
 *     documentation and/or other materials provided with the distribution.
 *   * Neither the name of Redis nor the names of its contributors may be used
 *     to endorse or promote products derived from this software without
 *     specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

30
#include "server.h"
31
#include "cluster.h"
32 33

#include <signal.h>
34
#include <ctype.h>
35 36 37 38 39 40 41 42

/*-----------------------------------------------------------------------------
 * C-level DB API
 *----------------------------------------------------------------------------*/

robj *lookupKey(redisDb *db, robj *key) {
    dictEntry *de = dictFind(db->dict,key->ptr);
    if (de) {
43
        robj *val = dictGetVal(de);
44

G
guiquanz 已提交
45
        /* Update the access time for the ageing algorithm.
46 47
         * Don't do it if we have a saving child, as this will trigger
         * a copy on write madness. */
A
antirez 已提交
48
        if (server.rdb_child_pid == -1 && server.aof_child_pid == -1)
49
            val->lru = LRU_CLOCK();
50 51 52 53 54 55 56
        return val;
    } else {
        return NULL;
    }
}

robj *lookupKeyRead(redisDb *db, robj *key) {
57 58
    robj *val;

59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79
    if (expireIfNeeded(db,key) == 1) {
        /* Key expired. If we are in the context of a master, expireIfNeeded()
         * returns 0 only when the key does not exist at all, so it's save
         * to return NULL ASAP. */
        if (server.masterhost == NULL) return NULL;

        /* However if we are in the context of a slave, expireIfNeeded() will
         * not really try to expire the key, it only returns information
         * about the "logical" status of the key: key expiring is up to the
         * master in order to have a consistent view of master's data set.
         *
         * However, if the command caller is not the master, and as additional
         * safety measure, the command invoked is a read-only command, we can
         * safely return NULL here, and provide a more consistent behavior
         * to clients accessign expired values in a read-only fashion, that
         * will say the key as non exisitng.
         *
         * Notably this covers GETs when slaves are used to scale reads. */
        if (server.current_client &&
            server.current_client != server.master &&
            server.current_client->cmd &&
A
antirez 已提交
80
            server.current_client->cmd->flags & CMD_READONLY)
81 82 83 84
        {
            return NULL;
        }
    }
85 86 87 88 89 90
    val = lookupKey(db,key);
    if (val == NULL)
        server.stat_keyspace_misses++;
    else
        server.stat_keyspace_hits++;
    return val;
91 92 93
}

robj *lookupKeyWrite(redisDb *db, robj *key) {
94
    expireIfNeeded(db,key);
95 96 97
    return lookupKey(db,key);
}

98
robj *lookupKeyReadOrReply(client *c, robj *key, robj *reply) {
99 100 101 102 103
    robj *o = lookupKeyRead(c->db, key);
    if (!o) addReply(c,reply);
    return o;
}

104
robj *lookupKeyWriteOrReply(client *c, robj *key, robj *reply) {
105 106 107 108 109
    robj *o = lookupKeyWrite(c->db, key);
    if (!o) addReply(c,reply);
    return o;
}

110
/* Add the key to the DB. It's up to the caller to increment the reference
G
guiquanz 已提交
111
 * counter of the value if needed.
112 113 114 115 116 117
 *
 * The program is aborted if the key already exists. */
void dbAdd(redisDb *db, robj *key, robj *val) {
    sds copy = sdsdup(key->ptr);
    int retval = dictAdd(db->dict, copy, val);

118
    serverAssertWithInfo(NULL,key,retval == C_OK);
119
    if (val->type == OBJ_LIST) signalListAsReady(db, key);
120
    if (server.cluster_enabled) slotToKeyAdd(key);
121 122 123 124 125 126 127 128
 }

/* Overwrite an existing key with a new value. Incrementing the reference
 * count of the new value is up to the caller.
 * This function does not modify the expire time of the existing key.
 *
 * The program is aborted if the key was not already present. */
void dbOverwrite(redisDb *db, robj *key, robj *val) {
A
antirez 已提交
129
    dictEntry *de = dictFind(db->dict,key->ptr);
130

A
antirez 已提交
131
    serverAssertWithInfo(NULL,key,de != NULL);
132
    dictReplace(db->dict, key->ptr, val);
133 134
}

135 136
/* High level Set operation. This function can be used in order to set
 * a key, whatever it was existing or not, to a new object.
137
 *
138 139 140 141 142 143
 * 1) The ref count of the value object is incremented.
 * 2) clients WATCHing for the destination key notified.
 * 3) The expire time of the key is reset (the key is made persistent). */
void setKey(redisDb *db, robj *key, robj *val) {
    if (lookupKeyWrite(db,key) == NULL) {
        dbAdd(db,key,val);
144
    } else {
145
        dbOverwrite(db,key,val);
146
    }
147 148
    incrRefCount(val);
    removeExpire(db,key);
149
    signalModifiedKey(db,key);
150 151 152 153 154 155 156 157 158 159 160
}

int dbExists(redisDb *db, robj *key) {
    return dictFind(db->dict,key->ptr) != NULL;
}

/* Return a random key, in form of a Redis object.
 * If there are no keys, NULL is returned.
 *
 * The function makes sure to return keys not already expired. */
robj *dbRandomKey(redisDb *db) {
A
antirez 已提交
161
    dictEntry *de;
162 163 164 165 166 167 168 169

    while(1) {
        sds key;
        robj *keyobj;

        de = dictGetRandomKey(db->dict);
        if (de == NULL) return NULL;

170
        key = dictGetKey(de);
171 172 173 174 175 176 177 178 179 180 181 182
        keyobj = createStringObject(key,sdslen(key));
        if (dictFind(db->expires,key)) {
            if (expireIfNeeded(db,keyobj)) {
                decrRefCount(keyobj);
                continue; /* search for another key. This expired. */
            }
        }
        return keyobj;
    }
}

/* Delete a key, value, and associated expiration entry if any, from the DB */
183
int dbSyncDelete(redisDb *db, robj *key) {
184 185 186
    /* Deleting an entry from the expires dict will not free the sds of
     * the key, because it is shared with the main dictionary. */
    if (dictSize(db->expires) > 0) dictDelete(db->expires,key->ptr);
187
    if (dictDelete(db->dict,key->ptr) == DICT_OK) {
188
        if (server.cluster_enabled) slotToKeyDel(key);
189 190 191 192
        return 1;
    } else {
        return 0;
    }
193 194
}

195 196 197 198 199 200 201 202
/* This is a wrapper whose behavior depends on the Redis lazy free
 * configuration. Deletes the key synchronously or asynchronously. */
int dbDelete(redisDb *db, robj *key) {
    int async = 1; /* TODO: Fixme making this a proper option. */
    if (async) return dbAsyncDelete(db,key);
    else       return dbSyncDelete(db,key);
}

203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223
/* Prepare the string object stored at 'key' to be modified destructively
 * to implement commands like SETBIT or APPEND.
 *
 * An object is usually ready to be modified unless one of the two conditions
 * are true:
 *
 * 1) The object 'o' is shared (refcount > 1), we don't want to affect
 *    other users.
 * 2) The object encoding is not "RAW".
 *
 * If the object is found in one of the above conditions (or both) by the
 * function, an unshared / not-encoded copy of the string object is stored
 * at 'key' in the specified 'db'. Otherwise the object 'o' itself is
 * returned.
 *
 * USAGE:
 *
 * The object 'o' is what the caller already obtained by looking up 'key'
 * in 'db', the usage pattern looks like this:
 *
 * o = lookupKeyWrite(db,key);
224
 * if (checkType(c,o,OBJ_STRING)) return;
225 226 227 228 229 230
 * o = dbUnshareStringValue(db,key,o);
 *
 * At this point the caller is ready to modify the object, for example
 * using an sdscat() call to append some data, or anything else.
 */
robj *dbUnshareStringValue(redisDb *db, robj *key, robj *o) {
A
antirez 已提交
231
    serverAssert(o->type == OBJ_STRING);
232
    if (o->refcount != 1 || o->encoding != OBJ_ENCODING_RAW) {
233 234 235 236 237 238 239 240
        robj *decoded = getDecodedObject(o);
        o = createRawStringObject(decoded->ptr, sdslen(decoded->ptr));
        decrRefCount(decoded);
        dbOverwrite(db,key,o);
    }
    return o;
}

241
long long emptyDb(void(callback)(void*)) {
242 243 244 245 246
    int j;
    long long removed = 0;

    for (j = 0; j < server.dbnum; j++) {
        removed += dictSize(server.db[j].dict);
247 248
        dictEmpty(server.db[j].dict,callback);
        dictEmpty(server.db[j].expires,callback);
249
    }
250
    if (server.cluster_enabled) slotToKeyFlush();
251 252 253
    return removed;
}

254
int selectDb(client *c, int id) {
255
    if (id < 0 || id >= server.dbnum)
256
        return C_ERR;
257
    c->db = &server.db[id];
258
    return C_OK;
259 260
}

261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277
/*-----------------------------------------------------------------------------
 * Hooks for key space changes.
 *
 * Every time a key in the database is modified the function
 * signalModifiedKey() is called.
 *
 * Every time a DB is flushed the function signalFlushDb() is called.
 *----------------------------------------------------------------------------*/

void signalModifiedKey(redisDb *db, robj *key) {
    touchWatchedKey(db,key);
}

void signalFlushedDb(int dbid) {
    touchWatchedKeysOnFlush(dbid);
}

278 279 280 281
/*-----------------------------------------------------------------------------
 * Type agnostic commands operating on the key space
 *----------------------------------------------------------------------------*/

282
void flushdbCommand(client *c) {
283
    server.dirty += dictSize(c->db->dict);
284
    signalFlushedDb(c->db->id);
285 286
    dictEmpty(c->db->dict,NULL);
    dictEmpty(c->db->expires,NULL);
287
    if (server.cluster_enabled) slotToKeyFlush();
288 289 290
    addReply(c,shared.ok);
}

291
void flushallCommand(client *c) {
292
    signalFlushedDb(-1);
293
    server.dirty += emptyDb(NULL);
294
    addReply(c,shared.ok);
A
antirez 已提交
295
    if (server.rdb_child_pid != -1) {
296
        kill(server.rdb_child_pid,SIGUSR1);
A
antirez 已提交
297
        rdbRemoveTempFile(server.rdb_child_pid);
298
    }
299 300 301 302
    if (server.saveparamslen > 0) {
        /* Normally rdbSave() will reset dirty, but we don't want this here
         * as otherwise FLUSHALL will not be replicated nor put into the AOF. */
        int saved_dirty = server.dirty;
A
antirez 已提交
303
        rdbSave(server.rdb_filename);
304 305
        server.dirty = saved_dirty;
    }
306 307 308
    server.dirty++;
}

309 310 311
/* This command implements DEL and LAZYDEL. */
void delGenericCommand(client *c, int lazy) {
    int numdel = 0, j;
312 313

    for (j = 1; j < c->argc; j++) {
314
        expireIfNeeded(c->db,c->argv[j]);
315 316 317
        int deleted  = lazy ? dbAsyncDelete(c->db,c->argv[j]) :
                              dbSyncDelete(c->db,c->argv[j]);
        if (deleted) {
318
            signalModifiedKey(c->db,c->argv[j]);
A
antirez 已提交
319
            notifyKeyspaceEvent(NOTIFY_GENERIC,
320
                "del",c->argv[j],c->db->id);
321
            server.dirty++;
322
            numdel++;
323 324
        }
    }
325 326 327 328 329 330 331 332 333
    addReplyLongLong(c,numdel);
}

void delCommand(client *c) {
    delGenericCommand(c,0);
}

void unlinkCommand(client *c) {
    delGenericCommand(c,1);
334 335
}

A
antirez 已提交
336 337
/* EXISTS key1 key2 ... key_N.
 * Return value is the number of keys existing. */
338
void existsCommand(client *c) {
A
antirez 已提交
339 340 341 342 343 344
    long long count = 0;
    int j;

    for (j = 1; j < c->argc; j++) {
        expireIfNeeded(c->db,c->argv[j]);
        if (dbExists(c->db,c->argv[j])) count++;
345
    }
A
antirez 已提交
346
    addReplyLongLong(c,count);
347 348
}

349
void selectCommand(client *c) {
350 351 352
    long id;

    if (getLongFromObjectOrReply(c, c->argv[1], &id,
353
        "invalid DB index") != C_OK)
354
        return;
355

356
    if (server.cluster_enabled && id != 0) {
A
antirez 已提交
357 358 359
        addReplyError(c,"SELECT is not allowed in cluster mode");
        return;
    }
360
    if (selectDb(c,id) == C_ERR) {
361
        addReplyError(c,"invalid DB index");
362 363 364 365 366
    } else {
        addReply(c,shared.ok);
    }
}

367
void randomkeyCommand(client *c) {
368 369 370 371 372 373 374 375 376 377 378
    robj *key;

    if ((key = dbRandomKey(c->db)) == NULL) {
        addReply(c,shared.nullbulk);
        return;
    }

    addReplyBulk(c,key);
    decrRefCount(key);
}

379
void keysCommand(client *c) {
380 381 382
    dictIterator *di;
    dictEntry *de;
    sds pattern = c->argv[1]->ptr;
A
antirez 已提交
383
    int plen = sdslen(pattern), allkeys;
384
    unsigned long numkeys = 0;
385
    void *replylen = addDeferredMultiBulkLength(c);
386

387
    di = dictGetSafeIterator(c->db->dict);
A
antirez 已提交
388
    allkeys = (pattern[0] == '*' && pattern[1] == '\0');
389
    while((de = dictNext(di)) != NULL) {
390
        sds key = dictGetKey(de);
391 392
        robj *keyobj;

A
antirez 已提交
393
        if (allkeys || stringmatchlen(pattern,plen,key,sdslen(key),0)) {
394 395 396 397 398 399 400 401 402
            keyobj = createStringObject(key,sdslen(key));
            if (expireIfNeeded(c->db,keyobj) == 0) {
                addReplyBulk(c,keyobj);
                numkeys++;
            }
            decrRefCount(keyobj);
        }
    }
    dictReleaseIterator(di);
403
    setDeferredMultiBulkLength(c,replylen,numkeys);
404 405
}

406 407
/* This callback is used by scanGenericCommand in order to collect elements
 * returned by the dictionary iterator into a list. */
P
Pieter Noordhuis 已提交
408
void scanCallback(void *privdata, const dictEntry *de) {
409 410 411 412 413 414 415 416
    void **pd = (void**) privdata;
    list *keys = pd[0];
    robj *o = pd[1];
    robj *key, *val = NULL;

    if (o == NULL) {
        sds sdskey = dictGetKey(de);
        key = createStringObject(sdskey, sdslen(sdskey));
417
    } else if (o->type == OBJ_SET) {
418 419
        sds keysds = dictGetKey(de);
        key = createStringObject(keysds,sdslen(keysds));
420
    } else if (o->type == OBJ_HASH) {
421 422 423 424
        key = dictGetKey(de);
        incrRefCount(key);
        val = dictGetVal(de);
        incrRefCount(val);
425
    } else if (o->type == OBJ_ZSET) {
426 427
        key = dictGetKey(de);
        incrRefCount(key);
428
        val = createStringObjectFromLongDouble(*(double*)dictGetVal(de),0);
429
    } else {
A
antirez 已提交
430
        serverPanic("Type not handled in SCAN callback.");
431 432 433 434
    }

    listAddNodeTail(keys, key);
    if (val) listAddNodeTail(keys, val);
P
Pieter Noordhuis 已提交
435 436
}

437
/* Try to parse a SCAN cursor stored at object 'o':
438
 * if the cursor is valid, store it as unsigned integer into *cursor and
439
 * returns C_OK. Otherwise return C_ERR and send an error to the
440
 * client. */
441
int parseScanCursorOrReply(client *c, robj *o, unsigned long *cursor) {
442 443 444 445 446 447 448 449 450
    char *eptr;

    /* Use strtoul() because we need an *unsigned* long, so
     * getLongLongFromObject() does not cover the whole cursor space. */
    errno = 0;
    *cursor = strtoul(o->ptr, &eptr, 10);
    if (isspace(((char*)o->ptr)[0]) || eptr[0] != '\0' || errno == ERANGE)
    {
        addReplyError(c, "invalid cursor");
451
        return C_ERR;
452
    }
453
    return C_OK;
454 455
}

456
/* This command implements SCAN, HSCAN and SSCAN commands.
457
 * If object 'o' is passed, then it must be a Hash or Set object, otherwise
458 459 460 461 462 463 464
 * if 'o' is NULL the command will operate on the dictionary associated with
 * the current database.
 *
 * When 'o' is not NULL the function assumes that the first argument in
 * the client arguments vector is a key so it skips it before iterating
 * in order to parse options.
 *
465
 * In the case of a Hash object the function returns both the field and value
466
 * of every element on the Hash. */
467
void scanGenericCommand(client *c, robj *o, unsigned long cursor) {
P
Pieter Noordhuis 已提交
468 469
    int i, j;
    list *keys = listCreate();
470
    listNode *node, *nextnode;
471
    long count = 10;
C
clark.kang 已提交
472 473
    sds pat = NULL;
    int patlen = 0, use_pattern = 0;
474 475 476 477
    dict *ht;

    /* Object must be NULL (to iterate keys names), or the type of the object
     * must be Set, Sorted Set, or Hash. */
A
antirez 已提交
478
    serverAssert(o == NULL || o->type == OBJ_SET || o->type == OBJ_HASH ||
479
                o->type == OBJ_ZSET);
480 481 482

    /* Set i to the first option argument. The previous one is the cursor. */
    i = (o == NULL) ? 2 : 3; /* Skip the key argument if needed. */
P
Pieter Noordhuis 已提交
483

484
    /* Step 1: Parse options. */
P
Pieter Noordhuis 已提交
485 486 487
    while (i < c->argc) {
        j = c->argc - i;
        if (!strcasecmp(c->argv[i]->ptr, "count") && j >= 2) {
A
antirez 已提交
488
            if (getLongFromObjectOrReply(c, c->argv[i+1], &count, NULL)
489
                != C_OK)
A
antirez 已提交
490
            {
P
Pieter Noordhuis 已提交
491 492 493 494 495 496 497 498 499
                goto cleanup;
            }

            if (count < 1) {
                addReply(c,shared.syntaxerr);
                goto cleanup;
            }

            i += 2;
500
        } else if (!strcasecmp(c->argv[i]->ptr, "match") && j >= 2) {
P
Pieter Noordhuis 已提交
501 502 503
            pat = c->argv[i+1]->ptr;
            patlen = sdslen(pat);

504 505 506
            /* The pattern always matches if it is exactly "*", so it is
             * equivalent to disabling it. */
            use_pattern = !(pat[0] == '*' && patlen == 1);
P
Pieter Noordhuis 已提交
507 508 509 510 511 512 513 514

            i += 2;
        } else {
            addReply(c,shared.syntaxerr);
            goto cleanup;
        }
    }

515 516 517
    /* Step 2: Iterate the collection.
     *
     * Note that if the object is encoded with a ziplist, intset, or any other
518
     * representation that is not a hash table, we are sure that it is also
519 520 521 522
     * composed of a small number of elements. So to avoid taking state we
     * just return everything inside the object in a single call, setting the
     * cursor to zero to signal the end of the iteration. */

523
    /* Handle the case of a hash table. */
524 525 526
    ht = NULL;
    if (o == NULL) {
        ht = c->db->dict;
527
    } else if (o->type == OBJ_SET && o->encoding == OBJ_ENCODING_HT) {
528
        ht = o->ptr;
529
    } else if (o->type == OBJ_HASH && o->encoding == OBJ_ENCODING_HT) {
530 531
        ht = o->ptr;
        count *= 2; /* We return key / value for this type. */
532
    } else if (o->type == OBJ_ZSET && o->encoding == OBJ_ENCODING_SKIPLIST) {
533 534 535 536 537 538 539
        zset *zs = o->ptr;
        ht = zs->dict;
        count *= 2; /* We return key / value for this type. */
    }

    if (ht) {
        void *privdata[2];
540 541 542 543 544
        /* We set the max number of iterations to ten times the specified
         * COUNT, so if the hash table is in a pathological state (very
         * sparsely populated) we avoid to block too much time at the cost
         * of returning no or very few elements. */
        long maxiterations = count*10;
545 546 547 548 549 550 551 552

        /* We pass two pointers to the callback: the list to which it will
         * add new elements, and the object containing the dictionary so that
         * it is possible to fetch more data in a type-dependent way. */
        privdata[0] = keys;
        privdata[1] = o;
        do {
            cursor = dictScan(ht, cursor, scanCallback, privdata);
553 554 555
        } while (cursor &&
              maxiterations-- &&
              listLength(keys) < (unsigned long)count);
556
    } else if (o->type == OBJ_SET) {
557
        int pos = 0;
558
        int64_t ll;
559 560 561

        while(intsetGet(o->ptr,pos++,&ll))
            listAddNodeTail(keys,createStringObjectFromLongLong(ll));
562
        cursor = 0;
563
    } else if (o->type == OBJ_HASH || o->type == OBJ_ZSET) {
564 565 566 567 568 569 570 571 572 573
        unsigned char *p = ziplistIndex(o->ptr,0);
        unsigned char *vstr;
        unsigned int vlen;
        long long vll;

        while(p) {
            ziplistGet(p,&vstr,&vlen,&vll);
            listAddNodeTail(keys,
                (vstr != NULL) ? createStringObject((char*)vstr,vlen) :
                                 createStringObjectFromLongLong(vll));
A
antirez 已提交
574
            p = ziplistNext(o->ptr,p);
575
        }
576
        cursor = 0;
577
    } else {
A
antirez 已提交
578
        serverPanic("Not handled encoding in SCAN.");
579
    }
P
Pieter Noordhuis 已提交
580

581
    /* Step 3: Filter elements. */
582 583 584 585
    node = listFirst(keys);
    while (node) {
        robj *kobj = listNodeValue(node);
        nextnode = listNextNode(node);
586 587 588
        int filter = 0;

        /* Filter element if it does not match the pattern. */
589
        if (!filter && use_pattern) {
590 591 592 593
            if (sdsEncodedObject(kobj)) {
                if (!stringmatchlen(pat, patlen, kobj->ptr, sdslen(kobj->ptr), 0))
                    filter = 1;
            } else {
A
antirez 已提交
594
                char buf[LONG_STR_SIZE];
595 596
                int len;

A
antirez 已提交
597
                serverAssert(kobj->encoding == OBJ_ENCODING_INT);
598 599 600 601
                len = ll2string(buf,sizeof(buf),(long)kobj->ptr);
                if (!stringmatchlen(pat, patlen, buf, len, 0)) filter = 1;
            }
        }
P
Pieter Noordhuis 已提交
602

603 604 605 606 607
        /* Filter element if it is an expired key. */
        if (!filter && o == NULL && expireIfNeeded(c->db, kobj)) filter = 1;

        /* Remove the element and its associted value if needed. */
        if (filter) {
P
Pieter Noordhuis 已提交
608
            decrRefCount(kobj);
609
            listDelNode(keys, node);
A
antirez 已提交
610 611
        }

612
        /* If this is a hash or a sorted set, we have a flat list of
A
antirez 已提交
613 614
         * key-value elements, so if this element was filtered, remove the
         * value, or skip it if it was not filtered: we only match keys. */
615
        if (o && (o->type == OBJ_ZSET || o->type == OBJ_HASH)) {
A
antirez 已提交
616 617 618
            node = nextnode;
            nextnode = listNextNode(node);
            if (filter) {
619 620 621 622
                kobj = listNodeValue(node);
                decrRefCount(kobj);
                listDelNode(keys, node);
            }
P
Pieter Noordhuis 已提交
623
        }
624
        node = nextnode;
P
Pieter Noordhuis 已提交
625 626
    }

627
    /* Step 4: Reply to the client. */
P
Pieter Noordhuis 已提交
628
    addReplyMultiBulkLen(c, 2);
629
    addReplyBulkLongLong(c,cursor);
P
Pieter Noordhuis 已提交
630 631

    addReplyMultiBulkLen(c, listLength(keys));
632 633
    while ((node = listFirst(keys)) != NULL) {
        robj *kobj = listNodeValue(node);
P
Pieter Noordhuis 已提交
634 635
        addReplyBulk(c, kobj);
        decrRefCount(kobj);
636
        listDelNode(keys, node);
P
Pieter Noordhuis 已提交
637 638 639
    }

cleanup:
640
    listSetFreeMethod(keys,decrRefCountVoid);
P
Pieter Noordhuis 已提交
641 642 643
    listRelease(keys);
}

644
/* The SCAN command completely relies on scanGenericCommand. */
645
void scanCommand(client *c) {
646
    unsigned long cursor;
647
    if (parseScanCursorOrReply(c,c->argv[1],&cursor) == C_ERR) return;
648
    scanGenericCommand(c,NULL,cursor);
649 650
}

651
void dbsizeCommand(client *c) {
652
    addReplyLongLong(c,dictSize(c->db->dict));
653 654
}

655
void lastsaveCommand(client *c) {
656
    addReplyLongLong(c,server.lastsave);
657 658
}

659
void typeCommand(client *c) {
660 661 662 663 664
    robj *o;
    char *type;

    o = lookupKeyRead(c->db,c->argv[1]);
    if (o == NULL) {
665
        type = "none";
666 667
    } else {
        switch(o->type) {
668 669 670 671 672
        case OBJ_STRING: type = "string"; break;
        case OBJ_LIST: type = "list"; break;
        case OBJ_SET: type = "set"; break;
        case OBJ_ZSET: type = "zset"; break;
        case OBJ_HASH: type = "hash"; break;
673
        default: type = "unknown"; break;
674 675
        }
    }
676
    addReplyStatus(c,type);
677 678
}

679
void shutdownCommand(client *c) {
680 681 682 683 684 685 686
    int flags = 0;

    if (c->argc > 2) {
        addReply(c,shared.syntaxerr);
        return;
    } else if (c->argc == 2) {
        if (!strcasecmp(c->argv[1]->ptr,"nosave")) {
A
antirez 已提交
687
            flags |= SHUTDOWN_NOSAVE;
688
        } else if (!strcasecmp(c->argv[1]->ptr,"save")) {
A
antirez 已提交
689
            flags |= SHUTDOWN_SAVE;
690 691 692 693 694
        } else {
            addReply(c,shared.syntaxerr);
            return;
        }
    }
695 696
    /* When SHUTDOWN is called while the server is loading a dataset in
     * memory we need to make sure no attempt is performed to save
A
antirez 已提交
697
     * the dataset on shutdown (otherwise it could overwrite the current DB
698 699 700 701
     * with half-read data).
     *
     * Also when in Sentinel mode clear the SAVE flag and force NOSAVE. */
    if (server.loading || server.sentinel_mode)
A
antirez 已提交
702
        flags = (flags & ~SHUTDOWN_SAVE) | SHUTDOWN_NOSAVE;
703
    if (prepareForShutdown(flags) == C_OK) exit(0);
704
    addReplyError(c,"Errors trying to SHUTDOWN. Check logs.");
705 706
}

707
void renameGenericCommand(client *c, int nx) {
708
    robj *o;
709
    long long expire;
710
    int samekey = 0;
711

712 713 714
    /* When source and dest key is the same, no operation is performed,
     * if the key exists, however we still return an error on unexisting key. */
    if (sdscmp(c->argv[1]->ptr,c->argv[2]->ptr) == 0) samekey = 1;
715 716 717 718

    if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr)) == NULL)
        return;

719 720 721 722 723
    if (samekey) {
        addReply(c,nx ? shared.czero : shared.ok);
        return;
    }

724
    incrRefCount(o);
725
    expire = getExpire(c->db,c->argv[1]);
726
    if (lookupKeyWrite(c->db,c->argv[2]) != NULL) {
727 728 729 730 731
        if (nx) {
            decrRefCount(o);
            addReply(c,shared.czero);
            return;
        }
A
antirez 已提交
732 733
        /* Overwrite: delete the old key before creating the new one
         * with the same name. */
734
        dbDelete(c->db,c->argv[2]);
735
    }
736 737
    dbAdd(c->db,c->argv[2],o);
    if (expire != -1) setExpire(c->db,c->argv[2],expire);
738
    dbDelete(c->db,c->argv[1]);
739 740
    signalModifiedKey(c->db,c->argv[1]);
    signalModifiedKey(c->db,c->argv[2]);
A
antirez 已提交
741
    notifyKeyspaceEvent(NOTIFY_GENERIC,"rename_from",
742
        c->argv[1],c->db->id);
A
antirez 已提交
743
    notifyKeyspaceEvent(NOTIFY_GENERIC,"rename_to",
744
        c->argv[2],c->db->id);
745 746 747 748
    server.dirty++;
    addReply(c,nx ? shared.cone : shared.ok);
}

749
void renameCommand(client *c) {
750 751 752
    renameGenericCommand(c,0);
}

753
void renamenxCommand(client *c) {
754 755 756
    renameGenericCommand(c,1);
}

757
void moveCommand(client *c) {
758 759 760
    robj *o;
    redisDb *src, *dst;
    int srcid;
761
    long long dbid, expire;
762

A
antirez 已提交
763 764 765 766 767
    if (server.cluster_enabled) {
        addReplyError(c,"MOVE is not allowed in cluster mode");
        return;
    }

768 769 770
    /* Obtain source and target DB pointers */
    src = c->db;
    srcid = c->db->id;
M
Matt Stancliff 已提交
771

772
    if (getLongLongFromObject(c->argv[2],&dbid) == C_ERR ||
M
Matt Stancliff 已提交
773
        dbid < INT_MIN || dbid > INT_MAX ||
774
        selectDb(c,dbid) == C_ERR)
M
Matt Stancliff 已提交
775
    {
776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794
        addReply(c,shared.outofrangeerr);
        return;
    }
    dst = c->db;
    selectDb(c,srcid); /* Back to the source DB */

    /* If the user is moving using as target the same
     * DB as the source DB it is probably an error. */
    if (src == dst) {
        addReply(c,shared.sameobjecterr);
        return;
    }

    /* Check if the element exists and get a reference */
    o = lookupKeyWrite(c->db,c->argv[1]);
    if (!o) {
        addReply(c,shared.czero);
        return;
    }
795
    expire = getExpire(c->db,c->argv[1]);
796

797 798
    /* Return zero if the key already exists in the target DB */
    if (lookupKeyWrite(dst,c->argv[1]) != NULL) {
799 800 801
        addReply(c,shared.czero);
        return;
    }
802
    dbAdd(dst,c->argv[1],o);
A
antirez 已提交
803
    if (expire != -1) setExpire(dst,c->argv[1],expire);
804 805 806 807 808 809 810 811 812 813 814 815 816 817 818
    incrRefCount(o);

    /* OK! key moved, free the entry in the source DB */
    dbDelete(src,c->argv[1]);
    server.dirty++;
    addReply(c,shared.cone);
}

/*-----------------------------------------------------------------------------
 * Expires API
 *----------------------------------------------------------------------------*/

int removeExpire(redisDb *db, robj *key) {
    /* An expire may only be removed if there is a corresponding entry in the
     * main dict. Otherwise, the key will never be freed. */
A
antirez 已提交
819
    serverAssertWithInfo(NULL,key,dictFind(db->dict,key->ptr) != NULL);
A
antirez 已提交
820
    return dictDelete(db->expires,key->ptr) == DICT_OK;
821 822
}

823 824
void setExpire(redisDb *db, robj *key, long long when) {
    dictEntry *kde, *de;
825 826

    /* Reuse the sds from the main dict in the expire dict */
827
    kde = dictFind(db->dict,key->ptr);
A
antirez 已提交
828
    serverAssertWithInfo(NULL,key,kde != NULL);
829 830
    de = dictReplaceRaw(db->expires,dictGetKey(kde));
    dictSetSignedIntegerVal(de,when);
831 832 833 834
}

/* Return the expire time of the specified key, or -1 if no expire
 * is associated with this key (i.e. the key is non volatile) */
835
long long getExpire(redisDb *db, robj *key) {
836 837 838 839 840 841 842 843
    dictEntry *de;

    /* No expire? return ASAP */
    if (dictSize(db->expires) == 0 ||
       (de = dictFind(db->expires,key->ptr)) == NULL) return -1;

    /* The entry was found in the expire dict, this means it should also
     * be present in the main dict (safety check). */
A
antirez 已提交
844
    serverAssertWithInfo(NULL,key,dictFind(db->dict,key->ptr) != NULL);
845
    return dictGetSignedIntegerVal(de);
846 847
}

848 849 850 851 852 853 854 855 856 857 858
/* Propagate expires into slaves and the AOF file.
 * When a key expires in the master, a DEL operation for this key is sent
 * to all the slaves and the AOF file if enabled.
 *
 * This way the key expiry is centralized in one place, and since both
 * AOF and the master->slave link guarantee operation ordering, everything
 * will be consistent even if we allow write operations against expiring
 * keys. */
void propagateExpire(redisDb *db, robj *key) {
    robj *argv[2];

859
    argv[0] = shared.del;
860
    argv[1] = key;
861 862
    incrRefCount(argv[0]);
    incrRefCount(argv[1]);
863

A
antirez 已提交
864
    if (server.aof_state != AOF_OFF)
865
        feedAppendOnlyFile(server.delCommand,db->id,argv,2);
866
    replicationFeedSlaves(server.slaves,db->id,argv,2);
867

868 869
    decrRefCount(argv[0]);
    decrRefCount(argv[1]);
870 871
}

872
int expireIfNeeded(redisDb *db, robj *key) {
873 874
    mstime_t when = getExpire(db,key);
    mstime_t now;
875

876 877
    if (when < 0) return 0; /* No expire for this key */

878 879 880
    /* Don't expire anything while loading. It will be done later. */
    if (server.loading) return 0;

881 882 883 884 885 886 887
    /* If we are in the context of a Lua script, we claim that time is
     * blocked to when the Lua script started. This way a key can expire
     * only the first time it is accessed and not in the middle of the
     * script execution, making propagation to slaves / AOF consistent.
     * See issue #1525 on Github for more information. */
    now = server.lua_caller ? server.lua_time_start : mstime();

888 889 890 891
    /* If we are running in the context of a slave, return ASAP:
     * the slave key expiration is controlled by the master that will
     * send us synthesized DEL operations for expired keys.
     *
892
     * Still we try to return the right information to the caller,
893 894
     * that is, 0 if we think the key should be still valid, 1 if
     * we think the key is expired at this time. */
895
    if (server.masterhost != NULL) return now > when;
896

897
    /* Return when this key has not expired */
898
    if (now <= when) return 0;
899 900 901

    /* Delete the key */
    server.stat_expiredkeys++;
902
    propagateExpire(db,key);
A
antirez 已提交
903
    notifyKeyspaceEvent(NOTIFY_EXPIRED,
904
        "expired",key,db->id);
905 906 907 908 909 910 911
    return dbDelete(db,key);
}

/*-----------------------------------------------------------------------------
 * Expires Commands
 *----------------------------------------------------------------------------*/

912 913 914 915 916 917
/* This is the generic command implementation for EXPIRE, PEXPIRE, EXPIREAT
 * and PEXPIREAT. Because the commad second argument may be relative or absolute
 * the "basetime" argument is used to signal what the base time is (either 0
 * for *AT variants of the command, or the current time for relative expires).
 *
 * unit is either UNIT_SECONDS or UNIT_MILLISECONDS, and is only used for
G
guiquanz 已提交
918
 * the argv[2] parameter. The basetime is always specified in milliseconds. */
919
void expireGenericCommand(client *c, long long basetime, int unit) {
920
    robj *key = c->argv[1], *param = c->argv[2];
921
    long long when; /* unix time in milliseconds when the key will expire. */
922

923
    if (getLongLongFromObjectOrReply(c, param, &when, NULL) != C_OK)
924
        return;
925

926 927
    if (unit == UNIT_SECONDS) when *= 1000;
    when += basetime;
928

929
    /* No key, return zero. */
930
    if (lookupKeyWrite(c->db,key) == NULL) {
931 932 933
        addReply(c,shared.czero);
        return;
    }
934

935 936 937 938 939 940
    /* EXPIRE with negative TTL, or EXPIREAT with a timestamp into the past
     * should never be executed as a DEL when load the AOF or in the context
     * of a slave instance.
     *
     * Instead we take the other branch of the IF statement setting an expire
     * (possibly in the past) and wait for an explicit DEL from the master. */
941
    if (when <= mstime() && !server.loading && !server.masterhost) {
942 943
        robj *aux;

A
antirez 已提交
944
        serverAssertWithInfo(c,key,dbDelete(c->db,key));
945 946 947 948 949 950
        server.dirty++;

        /* Replicate/AOF this as an explicit DEL. */
        aux = createStringObject("DEL",3);
        rewriteClientCommandVector(c,2,aux,key);
        decrRefCount(aux);
951
        signalModifiedKey(c->db,key);
A
antirez 已提交
952
        notifyKeyspaceEvent(NOTIFY_GENERIC,"del",key,c->db->id);
953
        addReply(c, shared.cone);
954 955
        return;
    } else {
956
        setExpire(c->db,key,when);
957
        addReply(c,shared.cone);
958
        signalModifiedKey(c->db,key);
A
antirez 已提交
959
        notifyKeyspaceEvent(NOTIFY_GENERIC,"expire",key,c->db->id);
960
        server.dirty++;
961 962 963 964
        return;
    }
}

965
void expireCommand(client *c) {
966
    expireGenericCommand(c,mstime(),UNIT_SECONDS);
967 968
}

969
void expireatCommand(client *c) {
970
    expireGenericCommand(c,0,UNIT_SECONDS);
971 972
}

973
void pexpireCommand(client *c) {
974
    expireGenericCommand(c,mstime(),UNIT_MILLISECONDS);
975
}
976

977
void pexpireatCommand(client *c) {
978
    expireGenericCommand(c,0,UNIT_MILLISECONDS);
979 980
}

981
void ttlGenericCommand(client *c, int output_ms) {
982
    long long expire, ttl = -1;
983

984
    /* If the key does not exist at all, return -2 */
A
antirez 已提交
985
    if (lookupKeyRead(c->db,c->argv[1]) == NULL) {
986 987 988 989 990
        addReplyLongLong(c,-2);
        return;
    }
    /* The key exists. Return -1 if it has no expire, or the actual
     * TTL value otherwise. */
A
antirez 已提交
991
    expire = getExpire(c->db,c->argv[1]);
992
    if (expire != -1) {
993
        ttl = expire-mstime();
A
antirez 已提交
994
        if (ttl < 0) ttl = 0;
995
    }
996 997 998
    if (ttl == -1) {
        addReplyLongLong(c,-1);
    } else {
999
        addReplyLongLong(c,output_ms ? ttl : ((ttl+500)/1000));
1000
    }
1001
}
A
antirez 已提交
1002

1003
void ttlCommand(client *c) {
1004 1005 1006
    ttlGenericCommand(c, 0);
}

1007
void pttlCommand(client *c) {
1008 1009 1010
    ttlGenericCommand(c, 1);
}

1011
void persistCommand(client *c) {
A
antirez 已提交
1012 1013 1014 1015 1016 1017
    dictEntry *de;

    de = dictFind(c->db->dict,c->argv[1]->ptr);
    if (de == NULL) {
        addReply(c,shared.czero);
    } else {
A
antirez 已提交
1018
        if (removeExpire(c->db,c->argv[1])) {
A
antirez 已提交
1019
            addReply(c,shared.cone);
A
antirez 已提交
1020 1021
            server.dirty++;
        } else {
A
antirez 已提交
1022
            addReply(c,shared.czero);
A
antirez 已提交
1023
        }
A
antirez 已提交
1024 1025
    }
}
1026 1027 1028 1029 1030

/* -----------------------------------------------------------------------------
 * API to get key arguments from commands
 * ---------------------------------------------------------------------------*/

1031 1032
/* The base case is to use the keys position as given in the command table
 * (firstkey, lastkey, step). */
1033 1034
int *getKeysUsingCommandTable(struct redisCommand *cmd,robj **argv, int argc, int *numkeys) {
    int j, i = 0, last, *keys;
A
antirez 已提交
1035
    UNUSED(argv);
1036 1037 1038 1039 1040 1041 1042 1043 1044

    if (cmd->firstkey == 0) {
        *numkeys = 0;
        return NULL;
    }
    last = cmd->lastkey;
    if (last < 0) last = argc+last;
    keys = zmalloc(sizeof(int)*((last - cmd->firstkey)+1));
    for (j = cmd->firstkey; j <= last; j += cmd->keystep) {
A
antirez 已提交
1045
        serverAssert(j < argc);
1046
        keys[i++] = j;
1047
    }
1048
    *numkeys = i;
1049 1050 1051
    return keys;
}

1052 1053 1054 1055 1056 1057 1058 1059
/* Return all the arguments that are keys in the command passed via argc / argv.
 *
 * The command returns the positions of all the key arguments inside the array,
 * so the actual return value is an heap allocated array of integers. The
 * length of the array is returned by reference into *numkeys.
 *
 * 'cmd' must be point to the corresponding entry into the redisCommand
 * table, according to the command name in argv[0].
1060 1061 1062
 *
 * This function uses the command table if a command-specific helper function
 * is not required, otherwise it calls the command-specific function. */
1063
int *getKeysFromCommand(struct redisCommand *cmd, robj **argv, int argc, int *numkeys) {
1064
    if (cmd->getkeys_proc) {
1065
        return cmd->getkeys_proc(cmd,argv,argc,numkeys);
1066 1067 1068 1069 1070
    } else {
        return getKeysUsingCommandTable(cmd,argv,argc,numkeys);
    }
}

1071
/* Free the result of getKeysFromCommand. */
1072 1073 1074 1075
void getKeysFreeResult(int *result) {
    zfree(result);
}

1076 1077 1078 1079
/* Helper function to extract keys from following commands:
 * ZUNIONSTORE <destkey> <num-keys> <key> <key> ... <key> <options>
 * ZINTERSTORE <destkey> <num-keys> <key> <key> ... <key> <options> */
int *zunionInterGetKeys(struct redisCommand *cmd, robj **argv, int argc, int *numkeys) {
1080
    int i, num, *keys;
A
antirez 已提交
1081
    UNUSED(cmd);
1082 1083 1084 1085 1086 1087 1088 1089

    num = atoi(argv[2]->ptr);
    /* Sanity check. Don't return any key if the command is going to
     * reply with syntax error. */
    if (num > (argc-3)) {
        *numkeys = 0;
        return NULL;
    }
1090 1091

    /* Keys in z{union,inter}store come from two places:
1092 1093
     * argv[1] = storage key,
     * argv[3...n] = keys to intersect */
1094 1095 1096
    keys = zmalloc(sizeof(int)*(num+1));

    /* Add all key positions for argv[3...n] to keys[] */
1097
    for (i = 0; i < num; i++) keys[i] = 3+i;
1098

1099
    /* Finally add the argv[1] key position (the storage key target). */
1100 1101
    keys[num] = 1;
    *numkeys = num+1;  /* Total keys = {union,inter} keys + storage key */
1102 1103
    return keys;
}
1104

1105 1106 1107 1108 1109
/* Helper function to extract keys from the following commands:
 * EVAL <script> <num-keys> <key> <key> ... <key> [more stuff]
 * EVALSHA <script> <num-keys> <key> <key> ... <key> [more stuff] */
int *evalGetKeys(struct redisCommand *cmd, robj **argv, int argc, int *numkeys) {
    int i, num, *keys;
A
antirez 已提交
1110
    UNUSED(cmd);
1111 1112 1113 1114 1115 1116 1117 1118 1119 1120

    num = atoi(argv[2]->ptr);
    /* Sanity check. Don't return any key if the command is going to
     * reply with syntax error. */
    if (num > (argc-3)) {
        *numkeys = 0;
        return NULL;
    }

    keys = zmalloc(sizeof(int)*num);
1121
    *numkeys = num;
1122 1123 1124 1125 1126 1127 1128

    /* Add all key positions for argv[3...n] to keys[] */
    for (i = 0; i < num; i++) keys[i] = 3+i;

    return keys;
}

1129 1130 1131 1132 1133 1134 1135 1136
/* Helper function to extract keys from the SORT command.
 *
 * SORT <sort-key> ... STORE <store-key> ...
 *
 * The first argument of SORT is always a key, however a list of options
 * follow in SQL-alike style. Here we parse just the minimum in order to
 * correctly identify keys in the "STORE" option. */
int *sortGetKeys(struct redisCommand *cmd, robj **argv, int argc, int *numkeys) {
M
Matt Stancliff 已提交
1137
    int i, j, num, *keys, found_store = 0;
A
antirez 已提交
1138
    UNUSED(cmd);
1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164

    num = 0;
    keys = zmalloc(sizeof(int)*2); /* Alloc 2 places for the worst case. */

    keys[num++] = 1; /* <sort-key> is always present. */

    /* Search for STORE option. By default we consider options to don't
     * have arguments, so if we find an unknown option name we scan the
     * next. However there are options with 1 or 2 arguments, so we
     * provide a list here in order to skip the right number of args. */
    struct {
        char *name;
        int skip;
    } skiplist[] = {
        {"limit", 2},
        {"get", 1},
        {"by", 1},
        {NULL, 0} /* End of elements. */
    };

    for (i = 2; i < argc; i++) {
        for (j = 0; skiplist[j].name != NULL; j++) {
            if (!strcasecmp(argv[i]->ptr,skiplist[j].name)) {
                i += skiplist[j].skip;
                break;
            } else if (!strcasecmp(argv[i]->ptr,"store") && i+1 < argc) {
1165 1166 1167
                /* Note: we don't increment "num" here and continue the loop
                 * to be sure to process the *last* "STORE" option if multiple
                 * ones are provided. This is same behavior as SORT. */
M
Matt Stancliff 已提交
1168
                found_store = 1;
1169
                keys[num] = i+1; /* <store-key> */
1170 1171 1172 1173
                break;
            }
        }
    }
M
Matt Stancliff 已提交
1174
    *numkeys = num + found_store;
1175 1176 1177
    return keys;
}

1178 1179 1180
/* Slot to Key API. This is used by Redis Cluster in order to obtain in
 * a fast way a key that belongs to a specified hash slot. This is useful
 * while rehashing the cluster. */
1181
void slotToKeyAdd(robj *key) {
1182 1183
    unsigned int hashslot = keyHashSlot(key->ptr,sdslen(key->ptr));

1184
    zslInsert(server.cluster->slots_to_keys,hashslot,key);
1185 1186 1187
    incrRefCount(key);
}

1188
void slotToKeyDel(robj *key) {
1189 1190
    unsigned int hashslot = keyHashSlot(key->ptr,sdslen(key->ptr));

1191
    zslDelete(server.cluster->slots_to_keys,hashslot,key);
1192 1193
}

1194
void slotToKeyFlush(void) {
1195 1196 1197 1198
    zslFree(server.cluster->slots_to_keys);
    server.cluster->slots_to_keys = zslCreate();
}

1199
unsigned int getKeysInSlot(unsigned int hashslot, robj **keys, unsigned int count) {
1200 1201
    zskiplistNode *n;
    zrangespec range;
A
antirez 已提交
1202
    int j = 0;
1203 1204 1205

    range.min = range.max = hashslot;
    range.minex = range.maxex = 0;
1206

1207
    n = zslFirstInRange(server.cluster->slots_to_keys, &range);
A
antirez 已提交
1208 1209 1210 1211 1212
    while(n && n->score == hashslot && count--) {
        keys[j++] = n->obj;
        n = n->level[0].forward;
    }
    return j;
1213
}
1214

1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236
/* Remove all the keys in the specified hash slot.
 * The number of removed items is returned. */
unsigned int delKeysInSlot(unsigned int hashslot) {
    zskiplistNode *n;
    zrangespec range;
    int j = 0;

    range.min = range.max = hashslot;
    range.minex = range.maxex = 0;

    n = zslFirstInRange(server.cluster->slots_to_keys, &range);
    while(n && n->score == hashslot) {
        robj *key = n->obj;
        n = n->level[0].forward; /* Go to the next item before freeing it. */
        incrRefCount(key); /* Protect the object while freeing it. */
        dbDelete(&server.db[0],key);
        decrRefCount(key);
        j++;
    }
    return j;
}

1237
unsigned int countKeysInSlot(unsigned int hashslot) {
1238 1239
    zskiplist *zsl = server.cluster->slots_to_keys;
    zskiplistNode *zn;
1240
    zrangespec range;
1241
    int rank, count = 0;
1242 1243 1244

    range.min = range.max = hashslot;
    range.minex = range.maxex = 0;
1245 1246

    /* Find first element in range */
1247
    zn = zslFirstInRange(zsl, &range);
1248 1249 1250 1251 1252 1253 1254

    /* Use rank of first element, if any, to determine preliminary count */
    if (zn != NULL) {
        rank = zslGetRank(zsl, zn->score, zn->obj);
        count = (zsl->length - (rank - 1));

        /* Find last element in range */
1255
        zn = zslLastInRange(zsl, &range);
1256 1257 1258 1259 1260 1261

        /* Use rank of last element, if any, to determine the actual count */
        if (zn != NULL) {
            rank = zslGetRank(zsl, zn->score, zn->obj);
            count -= (zsl->length - rank);
        }
1262
    }
1263
    return count;
1264
}