db.c 19.3 KB
Newer Older
1 2 3
#include "redis.h"

#include <signal.h>
4
#include <ctype.h>
5

6 7 8
void SlotToKeyAdd(robj *key);
void SlotToKeyDel(robj *key);

9 10 11 12 13 14 15
/*-----------------------------------------------------------------------------
 * C-level DB API
 *----------------------------------------------------------------------------*/

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

18 19 20
        /* Update the access time for the aging algorithm.
         * Don't do it if we have a saving child, as this will trigger
         * a copy on write madness. */
A
antirez 已提交
21
        if (server.rdb_child_pid == -1 && server.aof_child_pid == -1)
22
            val->lru = server.lruclock;
23 24 25 26 27 28 29
        return val;
    } else {
        return NULL;
    }
}

robj *lookupKeyRead(redisDb *db, robj *key) {
30 31
    robj *val;

32
    expireIfNeeded(db,key);
33 34 35 36 37 38
    val = lookupKey(db,key);
    if (val == NULL)
        server.stat_keyspace_misses++;
    else
        server.stat_keyspace_hits++;
    return val;
39 40 41
}

robj *lookupKeyWrite(redisDb *db, robj *key) {
42
    expireIfNeeded(db,key);
43 44 45 46 47 48 49 50 51 52 53 54 55 56 57
    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;
}

58 59 60 61 62 63 64 65
/* Add the key to the DB. It's up to the caller to increment the reference
 * counte of the value if needed.
 *
 * 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);

66
    redisAssertWithInfo(NULL,key,retval == REDIS_OK);
67 68 69 70 71 72 73 74 75 76
 }

/* 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);
    
77
    redisAssertWithInfo(NULL,key,de != NULL);
78
    dictReplace(db->dict, key->ptr, val);
79 80
}

81 82
/* 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.
83
 *
84 85 86 87 88 89
 * 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);
90
    } else {
91
        dbOverwrite(db,key,val);
92
    }
93 94
    incrRefCount(val);
    removeExpire(db,key);
95
    signalModifiedKey(db,key);
96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115
}

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;

116
        key = dictGetKey(de);
117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132
        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);
133 134 135 136 137
    if (dictDelete(db->dict,key->ptr) == DICT_OK) {
        return 1;
    } else {
        return 0;
    }
138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158
}

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);
    }
    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;
}

159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175
/*-----------------------------------------------------------------------------
 * 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);
}

176 177 178 179 180 181
/*-----------------------------------------------------------------------------
 * Type agnostic commands operating on the key space
 *----------------------------------------------------------------------------*/

void flushdbCommand(redisClient *c) {
    server.dirty += dictSize(c->db->dict);
182
    signalFlushedDb(c->db->id);
183 184 185 186 187 188
    dictEmpty(c->db->dict);
    dictEmpty(c->db->expires);
    addReply(c,shared.ok);
}

void flushallCommand(redisClient *c) {
189
    signalFlushedDb(-1);
190 191
    server.dirty += emptyDb();
    addReply(c,shared.ok);
A
antirez 已提交
192 193 194
    if (server.rdb_child_pid != -1) {
        kill(server.rdb_child_pid,SIGKILL);
        rdbRemoveTempFile(server.rdb_child_pid);
195
    }
196 197 198 199
    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 已提交
200
        rdbSave(server.rdb_filename);
201 202
        server.dirty = saved_dirty;
    }
203 204 205 206 207 208 209 210
    server.dirty++;
}

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

    for (j = 1; j < c->argc; j++) {
        if (dbDelete(c->db,c->argv[j])) {
211
            signalModifiedKey(c->db,c->argv[j]);
212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231
            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) {
    int id = atoi(c->argv[1]->ptr);

    if (selectDb(c,id) == REDIS_ERR) {
232
        addReplyError(c,"invalid DB index");
233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253
    } 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 已提交
254
    int plen = sdslen(pattern), allkeys;
255
    unsigned long numkeys = 0;
256
    void *replylen = addDeferredMultiBulkLength(c);
257 258

    di = dictGetIterator(c->db->dict);
A
antirez 已提交
259
    allkeys = (pattern[0] == '*' && pattern[1] == '\0');
260
    while((de = dictNext(di)) != NULL) {
261
        sds key = dictGetKey(de);
262 263
        robj *keyobj;

A
antirez 已提交
264
        if (allkeys || stringmatchlen(pattern,plen,key,sdslen(key),0)) {
265 266 267 268 269 270 271 272 273
            keyobj = createStringObject(key,sdslen(key));
            if (expireIfNeeded(c->db,keyobj) == 0) {
                addReplyBulk(c,keyobj);
                numkeys++;
            }
            decrRefCount(keyobj);
        }
    }
    dictReleaseIterator(di);
274
    setDeferredMultiBulkLength(c,replylen,numkeys);
275 276 277
}

void dbsizeCommand(redisClient *c) {
278
    addReplyLongLong(c,dictSize(c->db->dict));
279 280 281
}

void lastsaveCommand(redisClient *c) {
282
    addReplyLongLong(c,server.lastsave);
283 284 285 286 287 288 289 290
}

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

    o = lookupKeyRead(c->db,c->argv[1]);
    if (o == NULL) {
291
        type = "none";
292 293
    } else {
        switch(o->type) {
294 295 296 297 298 299
        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;
300 301
        }
    }
302
    addReplyStatus(c,type);
303 304 305
}

void shutdownCommand(redisClient *c) {
306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321
    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;
        }
    }
    if (prepareForShutdown(flags) == REDIS_OK) exit(0);
322
    addReplyError(c,"Errors trying to SHUTDOWN. Check logs.");
323 324 325 326
}

void renameGenericCommand(redisClient *c, int nx) {
    robj *o;
327
    long long expire;
328 329 330 331 332 333 334 335 336 337 338

    /* 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);
339
    expire = getExpire(c->db,c->argv[1]);
340
    if (lookupKeyWrite(c->db,c->argv[2]) != NULL) {
341 342 343 344 345
        if (nx) {
            decrRefCount(o);
            addReply(c,shared.czero);
            return;
        }
346 347
        /* Overwrite: delete the old key before creating the new one with the same name. */
        dbDelete(c->db,c->argv[2]);
348
    }
349 350
    dbAdd(c->db,c->argv[2],o);
    if (expire != -1) setExpire(c->db,c->argv[2],expire);
351
    dbDelete(c->db,c->argv[1]);
352 353
    signalModifiedKey(c->db,c->argv[1]);
    signalModifiedKey(c->db,c->argv[2]);
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
    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;

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

395 396
    /* Return zero if the key already exists in the target DB */
    if (lookupKeyWrite(dst,c->argv[1]) != NULL) {
397 398 399
        addReply(c,shared.czero);
        return;
    }
400
    dbAdd(dst,c->argv[1],o);
401 402 403 404 405 406 407 408 409 410 411 412 413 414 415
    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. */
416
    redisAssertWithInfo(NULL,key,dictFind(db->dict,key->ptr) != NULL);
A
antirez 已提交
417
    return dictDelete(db->expires,key->ptr) == DICT_OK;
418 419
}

420 421
void setExpire(redisDb *db, robj *key, long long when) {
    dictEntry *kde, *de;
422 423

    /* Reuse the sds from the main dict in the expire dict */
424 425 426 427
    kde = dictFind(db->dict,key->ptr);
    redisAssertWithInfo(NULL,key,kde != NULL);
    de = dictReplaceRaw(db->expires,dictGetKey(kde));
    dictSetSignedIntegerVal(de,when);
428 429 430 431
}

/* 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) */
432
long long getExpire(redisDb *db, robj *key) {
433 434 435 436 437 438 439 440
    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). */
441
    redisAssertWithInfo(NULL,key,dictFind(db->dict,key->ptr) != NULL);
442
    return dictGetSignedIntegerVal(de);
443 444
}

445 446 447 448 449 450 451 452 453 454 455
/* 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];

456
    argv[0] = shared.del;
457
    argv[1] = key;
458 459
    incrRefCount(argv[0]);
    incrRefCount(argv[1]);
460

461
    if (server.aof_state != REDIS_AOF_OFF)
462
        feedAppendOnlyFile(server.delCommand,db->id,argv,2);
463 464 465
    if (listLength(server.slaves))
        replicationFeedSlaves(server.slaves,db->id,argv,2);

466 467
    decrRefCount(argv[0]);
    decrRefCount(argv[1]);
468 469
}

470
int expireIfNeeded(redisDb *db, robj *key) {
471
    long long when = getExpire(db,key);
472

473 474
    if (when < 0) return 0; /* No expire for this key */

475 476 477
    /* Don't expire anything while loading. It will be done later. */
    if (server.loading) return 0;

478 479 480 481 482 483 484 485
    /* 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) {
486
        return mstime() > when;
487 488
    }

489
    /* Return when this key has not expired */
490
    if (mstime() <= when) return 0;
491 492 493

    /* Delete the key */
    server.stat_expiredkeys++;
494
    propagateExpire(db,key);
495 496 497 498 499 500 501
    return dbDelete(db,key);
}

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

502 503 504 505 506 507 508 509
/* Given an string object return true if it contains exactly the "ms"
 * or "MS" string. This is used in order to check if the last argument
 * of EXPIRE, EXPIREAT or TTL is "ms" to switch into millisecond input/output */
int stringObjectEqualsMs(robj *a) {
    char *arg = a->ptr;
    return tolower(arg[0]) == 'm' && tolower(arg[1]) == 's' && arg[2] == '\0';
}

510
void expireGenericCommand(redisClient *c, long long offset, int unit) {
511
    dictEntry *de;
512 513
    robj *key = c->argv[1], *param = c->argv[2];
    long long milliseconds;
514

515 516
    if (getLongLongFromObjectOrReply(c, param, &milliseconds, NULL) != REDIS_OK)
        return;
517

518
    if (unit == UNIT_SECONDS) milliseconds *= 1000;
519
    milliseconds += offset;
520 521 522 523 524 525

    de = dictFind(c->db->dict,key->ptr);
    if (de == NULL) {
        addReply(c,shared.czero);
        return;
    }
526 527 528 529 530 531
    /* 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. */
532
    if (milliseconds <= mstime() && !server.loading && !server.masterhost) {
533 534
        robj *aux;

535
        redisAssertWithInfo(c,key,dbDelete(c->db,key));
536 537 538 539 540 541
        server.dirty++;

        /* Replicate/AOF this as an explicit DEL. */
        aux = createStringObject("DEL",3);
        rewriteClientCommandVector(c,2,aux,key);
        decrRefCount(aux);
542
        signalModifiedKey(c->db,key);
543
        addReply(c, shared.cone);
544 545
        return;
    } else {
546
        setExpire(c->db,key,milliseconds);
547
        addReply(c,shared.cone);
548
        signalModifiedKey(c->db,key);
549
        server.dirty++;
550 551 552 553 554
        return;
    }
}

void expireCommand(redisClient *c) {
555
    expireGenericCommand(c,mstime(),UNIT_SECONDS);
556 557 558
}

void expireatCommand(redisClient *c) {
559
    expireGenericCommand(c,0,UNIT_SECONDS);
560 561
}

562
void pexpireCommand(redisClient *c) {
563
    expireGenericCommand(c,mstime(),UNIT_MILLISECONDS);
564
}
565

566
void pexpireatCommand(redisClient *c) {
567
    expireGenericCommand(c,0,UNIT_MILLISECONDS);
568 569 570 571
}

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

    expire = getExpire(c->db,c->argv[1]);
    if (expire != -1) {
575
        ttl = expire-mstime();
576 577
        if (ttl < 0) ttl = -1;
    }
578 579 580
    if (ttl == -1) {
        addReplyLongLong(c,-1);
    } else {
581
        addReplyLongLong(c,output_ms ? ttl : ((ttl+500)/1000));
582
    }
583
}
A
antirez 已提交
584

585 586 587 588 589 590 591 592
void ttlCommand(redisClient *c) {
    ttlGenericCommand(c, 0);
}

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

A
antirez 已提交
593 594 595 596 597 598 599
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 已提交
600
        if (removeExpire(c->db,c->argv[1])) {
A
antirez 已提交
601
            addReply(c,shared.cone);
A
antirez 已提交
602 603
            server.dirty++;
        } else {
A
antirez 已提交
604
            addReply(c,shared.czero);
A
antirez 已提交
605
        }
A
antirez 已提交
606 607
    }
}
608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625

/* -----------------------------------------------------------------------------
 * 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);
626
        keys[i++] = j;
627
    }
628
    *numkeys = i;
629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657
    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;
658
        return keys;
659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675
    } 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;
    }
676
    keys = zmalloc(sizeof(int)*num);
677 678 679 680
    for (i = 0; i < num; i++) keys[i] = 3+i;
    *numkeys = num;
    return keys;
}