db.c 32.2 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 "redis.h"
31
#include "cluster.h"
32 33

#include <signal.h>
34
#include <ctype.h>
35

36 37 38
void slotToKeyAdd(robj *key);
void slotToKeyDel(robj *key);
void slotToKeyFlush(void);
39

40 41 42 43 44 45 46
/*-----------------------------------------------------------------------------
 * C-level DB API
 *----------------------------------------------------------------------------*/

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

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

robj *lookupKeyRead(redisDb *db, robj *key) {
61 62
    robj *val;

63
    expireIfNeeded(db,key);
64 65 66 67 68 69
    val = lookupKey(db,key);
    if (val == NULL)
        server.stat_keyspace_misses++;
    else
        server.stat_keyspace_hits++;
    return val;
70 71 72
}

robj *lookupKeyWrite(redisDb *db, robj *key) {
73
    expireIfNeeded(db,key);
74 75 76 77 78 79 80 81 82 83 84 85 86 87 88
    return lookupKey(db,key);
}

robj *lookupKeyReadOrReply(redisClient *c, robj *key, robj *reply) {
    robj *o = lookupKeyRead(c->db, key);
    if (!o) addReply(c,reply);
    return o;
}

robj *lookupKeyWriteOrReply(redisClient *c, robj *key, robj *reply) {
    robj *o = lookupKeyWrite(c->db, key);
    if (!o) addReply(c,reply);
    return o;
}

89
/* Add the key to the DB. It's up to the caller to increment the reference
G
guiquanz 已提交
90
 * counter of the value if needed.
91 92 93 94 95 96
 *
 * 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);

97
    redisAssertWithInfo(NULL,key,retval == REDIS_OK);
98
    if (server.cluster_enabled) slotToKeyAdd(key);
99 100 101 102 103 104 105 106 107 108
 }

/* 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) {
    struct dictEntry *de = dictFind(db->dict,key->ptr);
    
109
    redisAssertWithInfo(NULL,key,de != NULL);
110
    dictReplace(db->dict, key->ptr, val);
111 112
}

113 114
/* 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.
115
 *
116 117 118 119 120 121
 * 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);
122
    } else {
123
        dbOverwrite(db,key,val);
124
    }
125 126
    incrRefCount(val);
    removeExpire(db,key);
127
    signalModifiedKey(db,key);
128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147
}

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) {
    struct dictEntry *de;

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

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

148
        key = dictGetKey(de);
149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164
        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 */
int dbDelete(redisDb *db, robj *key) {
    /* 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);
165
    if (dictDelete(db->dict,key->ptr) == DICT_OK) {
166
        if (server.cluster_enabled) slotToKeyDel(key);
167 168 169 170
        return 1;
    } else {
        return 0;
    }
171 172 173 174 175 176 177 178 179 180 181
}

long long emptyDb() {
    int j;
    long long removed = 0;

    for (j = 0; j < server.dbnum; j++) {
        removed += dictSize(server.db[j].dict);
        dictEmpty(server.db[j].dict);
        dictEmpty(server.db[j].expires);
    }
182
    if (server.cluster_enabled) slotToKeyFlush();
183 184 185 186 187 188 189 190 191 192
    return removed;
}

int selectDb(redisClient *c, int id) {
    if (id < 0 || id >= server.dbnum)
        return REDIS_ERR;
    c->db = &server.db[id];
    return REDIS_OK;
}

193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209
/*-----------------------------------------------------------------------------
 * 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);
}

210 211 212 213 214 215
/*-----------------------------------------------------------------------------
 * Type agnostic commands operating on the key space
 *----------------------------------------------------------------------------*/

void flushdbCommand(redisClient *c) {
    server.dirty += dictSize(c->db->dict);
216
    signalFlushedDb(c->db->id);
217 218
    dictEmpty(c->db->dict);
    dictEmpty(c->db->expires);
219
    if (server.cluster_enabled) slotToKeyFlush();
220 221 222 223
    addReply(c,shared.ok);
}

void flushallCommand(redisClient *c) {
224
    signalFlushedDb(-1);
225 226
    server.dirty += emptyDb();
    addReply(c,shared.ok);
A
antirez 已提交
227
    if (server.rdb_child_pid != -1) {
228
        kill(server.rdb_child_pid,SIGUSR1);
A
antirez 已提交
229
        rdbRemoveTempFile(server.rdb_child_pid);
230
    }
231 232 233 234
    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 已提交
235
        rdbSave(server.rdb_filename);
236 237
        server.dirty = saved_dirty;
    }
238 239 240 241 242 243 244 245
    server.dirty++;
}

void delCommand(redisClient *c) {
    int deleted = 0, j;

    for (j = 1; j < c->argc; j++) {
        if (dbDelete(c->db,c->argv[j])) {
246
            signalModifiedKey(c->db,c->argv[j]);
247 248
            notifyKeyspaceEvent(REDIS_NOTIFY_GENERIC,
                "del",c->argv[j],c->db->id);
249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265
            server.dirty++;
            deleted++;
        }
    }
    addReplyLongLong(c,deleted);
}

void existsCommand(redisClient *c) {
    expireIfNeeded(c->db,c->argv[1]);
    if (dbExists(c->db,c->argv[1])) {
        addReply(c, shared.cone);
    } else {
        addReply(c, shared.czero);
    }
}

void selectCommand(redisClient *c) {
266 267 268 269 270
    long id;

    if (getLongFromObjectOrReply(c, c->argv[1], &id,
        "invalid DB index") != REDIS_OK)
        return;
271

272
    if (server.cluster_enabled && id != 0) {
A
antirez 已提交
273 274 275
        addReplyError(c,"SELECT is not allowed in cluster mode");
        return;
    }
276
    if (selectDb(c,id) == REDIS_ERR) {
277
        addReplyError(c,"invalid DB index");
278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298
    } else {
        addReply(c,shared.ok);
    }
}

void randomkeyCommand(redisClient *c) {
    robj *key;

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

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

void keysCommand(redisClient *c) {
    dictIterator *di;
    dictEntry *de;
    sds pattern = c->argv[1]->ptr;
A
antirez 已提交
299
    int plen = sdslen(pattern), allkeys;
300
    unsigned long numkeys = 0;
301
    void *replylen = addDeferredMultiBulkLength(c);
302

303
    di = dictGetSafeIterator(c->db->dict);
A
antirez 已提交
304
    allkeys = (pattern[0] == '*' && pattern[1] == '\0');
305
    while((de = dictNext(di)) != NULL) {
306
        sds key = dictGetKey(de);
307 308
        robj *keyobj;

A
antirez 已提交
309
        if (allkeys || stringmatchlen(pattern,plen,key,sdslen(key),0)) {
310 311 312 313 314 315 316 317 318
            keyobj = createStringObject(key,sdslen(key));
            if (expireIfNeeded(c->db,keyobj) == 0) {
                addReplyBulk(c,keyobj);
                numkeys++;
            }
            decrRefCount(keyobj);
        }
    }
    dictReleaseIterator(di);
319
    setDeferredMultiBulkLength(c,replylen,numkeys);
320 321
}

322 323
/* This callback is used by scanGenericCommand in order to collect elements
 * returned by the dictionary iterator into a list. */
P
Pieter Noordhuis 已提交
324
void scanCallback(void *privdata, const dictEntry *de) {
325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350
    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));
    } else if (o->type == REDIS_SET) {
        key = dictGetKey(de);
        incrRefCount(key);
    } else if (o->type == REDIS_HASH) {
        key = dictGetKey(de);
        incrRefCount(key);
        val = dictGetVal(de);
        incrRefCount(val);
    } else if (o->type == REDIS_ZSET) {
        key = dictGetKey(de);
        incrRefCount(key);
        val = createStringObjectFromLongDouble(*(double*)dictGetVal(de));
    } else {
        redisPanic("Type not handled in SCAN callback.");
    }

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

353 354 355 356 357 358 359 360 361 362 363 364
/* This command implements SCAN, HSCAN and SSCAN commands.
 * If object 'o' is passed, then it must be an Hash or Set object, otherwise
 * 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.
 *
 * In the case of an Hash object the function returns both the field and value
 * of every element on the Hash. */
void scanGenericCommand(redisClient *c, robj *o) {
P
Pieter Noordhuis 已提交
365 366
    int rv;
    int i, j;
367
    char buf[REDIS_LONGSTR_SIZE], *eptr;
P
Pieter Noordhuis 已提交
368
    list *keys = listCreate();
369
    listNode *node, *nextnode;
P
Pieter Noordhuis 已提交
370
    unsigned long cursor = 0;
371
    long count = 10;
P
Pieter Noordhuis 已提交
372
    sds pat;
373
    int patlen, use_pattern = 0;
374 375 376 377 378 379 380 381 382
    dict *ht;

    /* Object must be NULL (to iterate keys names), or the type of the object
     * must be Set, Sorted Set, or Hash. */
    redisAssert(o == NULL || o->type == REDIS_SET || o->type == REDIS_HASH ||
                o->type == REDIS_ZSET);

    /* 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 已提交
383

384 385 386 387 388 389
    /* Use strtoul() because we need an *unsigned* long, so
     * getLongLongFromObject() does not cover the whole cursor space. */
    errno = 0;
    cursor = strtoul(c->argv[i-1]->ptr, &eptr, 10);
    if (isspace(((char*)c->argv[i-1])[0]) || eptr[0] != '\0' || errno == ERANGE)
    {
P
Pieter Noordhuis 已提交
390 391 392 393
        addReplyError(c, "invalid cursor");
        goto cleanup;
    }

394
    /* Step 1: Parse options. */
P
Pieter Noordhuis 已提交
395 396 397
    while (i < c->argc) {
        j = c->argc - i;
        if (!strcasecmp(c->argv[i]->ptr, "count") && j >= 2) {
A
antirez 已提交
398 399 400
            if (getLongFromObjectOrReply(c, c->argv[i+1], &count, NULL)
                != REDIS_OK)
            {
P
Pieter Noordhuis 已提交
401 402 403 404 405 406 407 408 409
                goto cleanup;
            }

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

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

414 415 416
            /* The pattern always matches if it is exactly "*", so it is
             * equivalent to disabling it. */
            use_pattern = !(pat[0] == '*' && patlen == 1);
P
Pieter Noordhuis 已提交
417 418 419 420 421 422 423 424

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

425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460
    /* Step 2: Iterate the collection.
     *
     * Note that if the object is encoded with a ziplist, intset, or any other
     * representation that is not an hash table, we are sure that it is also
     * 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. */

    /* Handle the case of an hash table. */
    ht = NULL;
    if (o == NULL) {
        ht = c->db->dict;
    } else if (o->type == REDIS_SET && o->encoding == REDIS_ENCODING_HT) {
        ht = o->ptr;
    } else if (o->type == REDIS_HASH && o->encoding == REDIS_ENCODING_HT) {
        ht = o->ptr;
        count *= 2; /* We return key / value for this type. */
    } else if (o->type == REDIS_ZSET && o->encoding == REDIS_ENCODING_SKIPLIST) {
        zset *zs = o->ptr;
        ht = zs->dict;
        count *= 2; /* We return key / value for this type. */
    }

    if (ht) {
        void *privdata[2];

        /* 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);
        } while (cursor && listLength(keys) < count);
    } else if (o->type == REDIS_SET) {
        int pos = 0;
461
        int64_t ll;
462 463 464 465 466 467 468 469 470 471 472 473 474 475

        while(intsetGet(o->ptr,pos++,&ll))
            listAddNodeTail(keys,createStringObjectFromLongLong(ll));
    } else if (o->type == REDIS_HASH || o->type == REDIS_ZSET) {
        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 已提交
476
            p = ziplistNext(o->ptr,p);
477 478 479 480
        }
    } else {
        redisPanic("Not handled encoding in SCAN.");
    }
P
Pieter Noordhuis 已提交
481

482
    /* Step 3: Filter elements. */
483 484 485 486
    node = listFirst(keys);
    while (node) {
        robj *kobj = listNodeValue(node);
        nextnode = listNextNode(node);
487 488 489
        int filter = 0;

        /* Filter element if it does not match the pattern. */
490
        if (!filter && use_pattern) {
491 492 493 494 495 496 497 498 499 500 501 502
            if (sdsEncodedObject(kobj)) {
                if (!stringmatchlen(pat, patlen, kobj->ptr, sdslen(kobj->ptr), 0))
                    filter = 1;
            } else {
                char buf[REDIS_LONGSTR_SIZE];
                int len;

                redisAssert(kobj->encoding == REDIS_ENCODING_INT);
                len = ll2string(buf,sizeof(buf),(long)kobj->ptr);
                if (!stringmatchlen(pat, patlen, buf, len, 0)) filter = 1;
            }
        }
P
Pieter Noordhuis 已提交
503

504 505 506 507 508
        /* 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 已提交
509
            decrRefCount(kobj);
510
            listDelNode(keys, node);
A
antirez 已提交
511 512 513 514 515 516 517 518 519
        }

        /* If this is an hash or a sorted set, we have a flat list of
         * key-value elements, so if this element was filtered, remove the
         * value, or skip it if it was not filtered: we only match keys. */
        if (o && (o->type == REDIS_ZSET || o->type == REDIS_HASH)) {
            node = nextnode;
            nextnode = listNextNode(node);
            if (filter) {
520 521 522 523
                kobj = listNodeValue(node);
                decrRefCount(kobj);
                listDelNode(keys, node);
            }
P
Pieter Noordhuis 已提交
524
        }
525
        node = nextnode;
P
Pieter Noordhuis 已提交
526 527
    }

528
    /* Step 4: Reply to the client. */
P
Pieter Noordhuis 已提交
529 530 531 532 533 534
    addReplyMultiBulkLen(c, 2);
    rv = snprintf(buf, sizeof(buf), "%lu", cursor);
    redisAssert(rv < sizeof(buf));
    addReplyBulkCBuffer(c, buf, rv);

    addReplyMultiBulkLen(c, listLength(keys));
535 536
    while ((node = listFirst(keys)) != NULL) {
        robj *kobj = listNodeValue(node);
P
Pieter Noordhuis 已提交
537 538
        addReplyBulk(c, kobj);
        decrRefCount(kobj);
539
        listDelNode(keys, node);
P
Pieter Noordhuis 已提交
540 541 542
    }

cleanup:
543
    listSetFreeMethod(keys,decrRefCountVoid);
P
Pieter Noordhuis 已提交
544 545 546
    listRelease(keys);
}

547 548 549 550 551
/* The SCAN command completely relies on scanGenericCommand. */
void scanCommand(redisClient *c) {
    scanGenericCommand(c,NULL);
}

552
void dbsizeCommand(redisClient *c) {
553
    addReplyLongLong(c,dictSize(c->db->dict));
554 555 556
}

void lastsaveCommand(redisClient *c) {
557
    addReplyLongLong(c,server.lastsave);
558 559 560 561 562 563 564 565
}

void typeCommand(redisClient *c) {
    robj *o;
    char *type;

    o = lookupKeyRead(c->db,c->argv[1]);
    if (o == NULL) {
566
        type = "none";
567 568
    } else {
        switch(o->type) {
569 570 571 572 573 574
        case REDIS_STRING: type = "string"; break;
        case REDIS_LIST: type = "list"; break;
        case REDIS_SET: type = "set"; break;
        case REDIS_ZSET: type = "zset"; break;
        case REDIS_HASH: type = "hash"; break;
        default: type = "unknown"; break;
575 576
        }
    }
577
    addReplyStatus(c,type);
578 579 580
}

void shutdownCommand(redisClient *c) {
581 582 583 584 585 586 587 588 589 590 591 592 593 594 595
    int flags = 0;

    if (c->argc > 2) {
        addReply(c,shared.syntaxerr);
        return;
    } else if (c->argc == 2) {
        if (!strcasecmp(c->argv[1]->ptr,"nosave")) {
            flags |= REDIS_SHUTDOWN_NOSAVE;
        } else if (!strcasecmp(c->argv[1]->ptr,"save")) {
            flags |= REDIS_SHUTDOWN_SAVE;
        } else {
            addReply(c,shared.syntaxerr);
            return;
        }
    }
A
antirez 已提交
596 597 598 599 600 601
    /* SHUTDOWN can be called even while the server is in "loading" state.
     * When this happens we need to make sure no attempt is performed to save
     * the dataset on shutdown (otherwise it could overwrite the current DB
     * with half-read data). */
    if (server.loading)
        flags = (flags & ~REDIS_SHUTDOWN_SAVE) | REDIS_SHUTDOWN_NOSAVE;
602
    if (prepareForShutdown(flags) == REDIS_OK) exit(0);
603
    addReplyError(c,"Errors trying to SHUTDOWN. Check logs.");
604 605 606 607
}

void renameGenericCommand(redisClient *c, int nx) {
    robj *o;
608
    long long expire;
609 610 611 612 613 614 615 616 617 618 619

    /* To use the same key as src and dst is probably an error */
    if (sdscmp(c->argv[1]->ptr,c->argv[2]->ptr) == 0) {
        addReply(c,shared.sameobjecterr);
        return;
    }

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

    incrRefCount(o);
620
    expire = getExpire(c->db,c->argv[1]);
621
    if (lookupKeyWrite(c->db,c->argv[2]) != NULL) {
622 623 624 625 626
        if (nx) {
            decrRefCount(o);
            addReply(c,shared.czero);
            return;
        }
A
antirez 已提交
627 628
        /* Overwrite: delete the old key before creating the new one
         * with the same name. */
629
        dbDelete(c->db,c->argv[2]);
630
    }
631 632
    dbAdd(c->db,c->argv[2],o);
    if (expire != -1) setExpire(c->db,c->argv[2],expire);
633
    dbDelete(c->db,c->argv[1]);
634 635
    signalModifiedKey(c->db,c->argv[1]);
    signalModifiedKey(c->db,c->argv[2]);
636 637 638 639
    notifyKeyspaceEvent(REDIS_NOTIFY_GENERIC,"rename_from",
        c->argv[1],c->db->id);
    notifyKeyspaceEvent(REDIS_NOTIFY_GENERIC,"rename_to",
        c->argv[2],c->db->id);
640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656
    server.dirty++;
    addReply(c,nx ? shared.cone : shared.ok);
}

void renameCommand(redisClient *c) {
    renameGenericCommand(c,0);
}

void renamenxCommand(redisClient *c) {
    renameGenericCommand(c,1);
}

void moveCommand(redisClient *c) {
    robj *o;
    redisDb *src, *dst;
    int srcid;

A
antirez 已提交
657 658 659 660 661
    if (server.cluster_enabled) {
        addReplyError(c,"MOVE is not allowed in cluster mode");
        return;
    }

662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685
    /* Obtain source and target DB pointers */
    src = c->db;
    srcid = c->db->id;
    if (selectDb(c,atoi(c->argv[2]->ptr)) == REDIS_ERR) {
        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;
    }

686 687
    /* Return zero if the key already exists in the target DB */
    if (lookupKeyWrite(dst,c->argv[1]) != NULL) {
688 689 690
        addReply(c,shared.czero);
        return;
    }
691
    dbAdd(dst,c->argv[1],o);
692 693 694 695 696 697 698 699 700 701 702 703 704 705 706
    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. */
707
    redisAssertWithInfo(NULL,key,dictFind(db->dict,key->ptr) != NULL);
A
antirez 已提交
708
    return dictDelete(db->expires,key->ptr) == DICT_OK;
709 710
}

711 712
void setExpire(redisDb *db, robj *key, long long when) {
    dictEntry *kde, *de;
713 714

    /* Reuse the sds from the main dict in the expire dict */
715 716 717 718
    kde = dictFind(db->dict,key->ptr);
    redisAssertWithInfo(NULL,key,kde != NULL);
    de = dictReplaceRaw(db->expires,dictGetKey(kde));
    dictSetSignedIntegerVal(de,when);
719 720 721 722
}

/* 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) */
723
long long getExpire(redisDb *db, robj *key) {
724 725 726 727 728 729 730 731
    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). */
732
    redisAssertWithInfo(NULL,key,dictFind(db->dict,key->ptr) != NULL);
733
    return dictGetSignedIntegerVal(de);
734 735
}

736 737 738 739 740 741 742 743 744 745 746
/* 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];

747
    argv[0] = shared.del;
748
    argv[1] = key;
749 750
    incrRefCount(argv[0]);
    incrRefCount(argv[1]);
751

752
    if (server.aof_state != REDIS_AOF_OFF)
753
        feedAppendOnlyFile(server.delCommand,db->id,argv,2);
754
    replicationFeedSlaves(server.slaves,db->id,argv,2);
755

756 757
    decrRefCount(argv[0]);
    decrRefCount(argv[1]);
758 759
}

760
int expireIfNeeded(redisDb *db, robj *key) {
761
    long long when = getExpire(db,key);
762

763 764
    if (when < 0) return 0; /* No expire for this key */

765 766 767
    /* Don't expire anything while loading. It will be done later. */
    if (server.loading) return 0;

768 769 770 771 772 773 774 775
    /* 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.
     *
     * Still we try to return the right information to the caller, 
     * that is, 0 if we think the key should be still valid, 1 if
     * we think the key is expired at this time. */
    if (server.masterhost != NULL) {
776
        return mstime() > when;
777 778
    }

779
    /* Return when this key has not expired */
780
    if (mstime() <= when) return 0;
781 782 783

    /* Delete the key */
    server.stat_expiredkeys++;
784
    propagateExpire(db,key);
785 786
    notifyKeyspaceEvent(REDIS_NOTIFY_EXPIRED,
        "expired",key,db->id);
787 788 789 790 791 792 793
    return dbDelete(db,key);
}

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

794 795 796 797 798 799
/* 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 已提交
800
 * the argv[2] parameter. The basetime is always specified in milliseconds. */
801
void expireGenericCommand(redisClient *c, long long basetime, int unit) {
802
    robj *key = c->argv[1], *param = c->argv[2];
803
    long long when; /* unix time in milliseconds when the key will expire. */
804

805
    if (getLongLongFromObjectOrReply(c, param, &when, NULL) != REDIS_OK)
806
        return;
807

808 809
    if (unit == UNIT_SECONDS) when *= 1000;
    when += basetime;
810

811 812
    /* No key, return zero. */
    if (lookupKeyRead(c->db,key) == NULL) {
813 814 815
        addReply(c,shared.czero);
        return;
    }
816

817 818 819 820 821 822
    /* 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. */
823
    if (when <= mstime() && !server.loading && !server.masterhost) {
824 825
        robj *aux;

826
        redisAssertWithInfo(c,key,dbDelete(c->db,key));
827 828 829 830 831 832
        server.dirty++;

        /* Replicate/AOF this as an explicit DEL. */
        aux = createStringObject("DEL",3);
        rewriteClientCommandVector(c,2,aux,key);
        decrRefCount(aux);
833
        signalModifiedKey(c->db,key);
834
        notifyKeyspaceEvent(REDIS_NOTIFY_GENERIC,"del",key,c->db->id);
835
        addReply(c, shared.cone);
836 837
        return;
    } else {
838
        setExpire(c->db,key,when);
839
        addReply(c,shared.cone);
840
        signalModifiedKey(c->db,key);
841
        notifyKeyspaceEvent(REDIS_NOTIFY_GENERIC,"expire",key,c->db->id);
842
        server.dirty++;
843 844 845 846 847
        return;
    }
}

void expireCommand(redisClient *c) {
848
    expireGenericCommand(c,mstime(),UNIT_SECONDS);
849 850 851
}

void expireatCommand(redisClient *c) {
852
    expireGenericCommand(c,0,UNIT_SECONDS);
853 854
}

855
void pexpireCommand(redisClient *c) {
856
    expireGenericCommand(c,mstime(),UNIT_MILLISECONDS);
857
}
858

859
void pexpireatCommand(redisClient *c) {
860
    expireGenericCommand(c,0,UNIT_MILLISECONDS);
861 862 863 864
}

void ttlGenericCommand(redisClient *c, int output_ms) {
    long long expire, ttl = -1;
865

866
    /* If the key does not exist at all, return -2 */
A
antirez 已提交
867
    if (lookupKeyRead(c->db,c->argv[1]) == NULL) {
868 869 870 871 872
        addReplyLongLong(c,-2);
        return;
    }
    /* The key exists. Return -1 if it has no expire, or the actual
     * TTL value otherwise. */
A
antirez 已提交
873
    expire = getExpire(c->db,c->argv[1]);
874
    if (expire != -1) {
875
        ttl = expire-mstime();
A
antirez 已提交
876
        if (ttl < 0) ttl = 0;
877
    }
878 879 880
    if (ttl == -1) {
        addReplyLongLong(c,-1);
    } else {
881
        addReplyLongLong(c,output_ms ? ttl : ((ttl+500)/1000));
882
    }
883
}
A
antirez 已提交
884

885 886 887 888 889 890 891 892
void ttlCommand(redisClient *c) {
    ttlGenericCommand(c, 0);
}

void pttlCommand(redisClient *c) {
    ttlGenericCommand(c, 1);
}

A
antirez 已提交
893 894 895 896 897 898 899
void persistCommand(redisClient *c) {
    dictEntry *de;

    de = dictFind(c->db->dict,c->argv[1]->ptr);
    if (de == NULL) {
        addReply(c,shared.czero);
    } else {
A
antirez 已提交
900
        if (removeExpire(c->db,c->argv[1])) {
A
antirez 已提交
901
            addReply(c,shared.cone);
A
antirez 已提交
902 903
            server.dirty++;
        } else {
A
antirez 已提交
904
            addReply(c,shared.czero);
A
antirez 已提交
905
        }
A
antirez 已提交
906 907
    }
}
908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925

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

int *getKeysUsingCommandTable(struct redisCommand *cmd,robj **argv, int argc, int *numkeys) {
    int j, i = 0, last, *keys;
    REDIS_NOTUSED(argv);

    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) {
        redisAssert(j < argc);
926
        keys[i++] = j;
927
    }
928
    *numkeys = i;
929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957
    return keys;
}

int *getKeysFromCommand(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags) {
    if (cmd->getkeys_proc) {
        return cmd->getkeys_proc(cmd,argv,argc,numkeys,flags);
    } else {
        return getKeysUsingCommandTable(cmd,argv,argc,numkeys);
    }
}

void getKeysFreeResult(int *result) {
    zfree(result);
}

int *noPreloadGetKeys(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags) {
    if (flags & REDIS_GETKEYS_PRELOAD) {
        *numkeys = 0;
        return NULL;
    } else {
        return getKeysUsingCommandTable(cmd,argv,argc,numkeys);
    }
}

int *renameGetKeys(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags) {
    if (flags & REDIS_GETKEYS_PRELOAD) {
        int *keys = zmalloc(sizeof(int));
        *numkeys = 1;
        keys[0] = 1;
958
        return keys;
959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975
    } else {
        return getKeysUsingCommandTable(cmd,argv,argc,numkeys);
    }
}

int *zunionInterGetKeys(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags) {
    int i, num, *keys;
    REDIS_NOTUSED(cmd);
    REDIS_NOTUSED(flags);

    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;
    }
976
    keys = zmalloc(sizeof(int)*num);
977 978 979 980
    for (i = 0; i < num; i++) keys[i] = 3+i;
    *numkeys = num;
    return keys;
}
981 982 983 984

/* 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. */
985
void slotToKeyAdd(robj *key) {
986 987
    unsigned int hashslot = keyHashSlot(key->ptr,sdslen(key->ptr));

988
    zslInsert(server.cluster->slots_to_keys,hashslot,key);
989 990 991
    incrRefCount(key);
}

992
void slotToKeyDel(robj *key) {
993 994
    unsigned int hashslot = keyHashSlot(key->ptr,sdslen(key->ptr));

995
    zslDelete(server.cluster->slots_to_keys,hashslot,key);
996 997
}

998
void slotToKeyFlush(void) {
999 1000 1001 1002
    zslFree(server.cluster->slots_to_keys);
    server.cluster->slots_to_keys = zslCreate();
}

1003
unsigned int getKeysInSlot(unsigned int hashslot, robj **keys, unsigned int count) {
1004 1005
    zskiplistNode *n;
    zrangespec range;
A
antirez 已提交
1006
    int j = 0;
1007 1008 1009 1010

    range.min = range.max = hashslot;
    range.minex = range.maxex = 0;
    
1011
    n = zslFirstInRange(server.cluster->slots_to_keys, range);
A
antirez 已提交
1012 1013 1014 1015 1016
    while(n && n->score == hashslot && count--) {
        keys[j++] = n->obj;
        n = n->level[0].forward;
    }
    return j;
1017
}
1018

1019
unsigned int countKeysInSlot(unsigned int hashslot) {
1020 1021
    zskiplist *zsl = server.cluster->slots_to_keys;
    zskiplistNode *zn;
1022
    zrangespec range;
1023
    int rank, count = 0;
1024 1025 1026

    range.min = range.max = hashslot;
    range.minex = range.maxex = 0;
1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043

    /* Find first element in range */
    zn = zslFirstInRange(zsl, range);

    /* 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 */
        zn = zslLastInRange(zsl, range);

        /* 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);
        }
1044
    }
1045
    return count;
1046
}