src/db.c

   1 #include "redis.h"
   2
   3 #include <signal.h>
   4
   5 /*-----------------------------------------------------------------------------
   6  * C-level DB API
   7  *----------------------------------------------------------------------------*/
   8
   9 robj *lookupKey(redisDb *db, robj *key) {
  10     dictEntry *de = dictFind(db->dict,key->ptr);
  11     if (de) {
  12         robj *val = dictGetEntryVal(de);
  13
  14         /* Update the access time for the aging algorithm.
  15          * Don't do it if we have a saving child, as this will trigger
  16          * a copy on write madness. */
  17         if (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1)
  18             val->lru = server.lruclock;
  19
  20         if (server.ds_enabled &&
  21             cacheScheduleIOGetFlags(db,key) & REDIS_IO_SAVEINPROG)
  22         {
  23             /* There is a save in progress for this object!
  24              * Wait for it to get out. */
  25             waitEmptyIOJobsQueue();
  26             processAllPendingIOJobs();
  27             redisAssert(!(cacheScheduleIOGetFlags(db,key) & REDIS_IO_SAVEINPROG));
  28         }
  29         server.stat_keyspace_hits++;
  30         return val;
  31     } else {
  32         time_t expire;
  33         robj *val;
  34
  35         /* Key not found in the in memory hash table, but if disk store is
  36          * enabled we may have this key on disk. If so load it in memory
  37          * in a blocking way. */
  38         if (server.ds_enabled && cacheKeyMayExist(db,key)) {
  39             if (cacheScheduleIOGetFlags(db,key) & REDIS_IO_SAVEINPROG) {
  40                 /* There is a save in progress for this object!
  41                  * Wait for it to get out. */
  42                 waitEmptyIOJobsQueue();
  43                 processAllPendingIOJobs();
  44                 redisAssert((cacheScheduleIOGetFlags(db,key) & REDIS_IO_SAVEINPROG) == 0);
  45             }
  46
  47             redisLog(REDIS_DEBUG,"Force loading key %s via lookup",
  48                 key->ptr);
  49             val = dsGet(db,key,&expire);
  50             if (val) {
  51                 int retval = dbAdd(db,key,val);
  52                 redisAssert(retval == REDIS_OK);
  53                 if (expire != -1) setExpire(db,key,expire);
  54                 server.stat_keyspace_hits++;
  55                 return val;
  56             }
  57         }
  58         server.stat_keyspace_misses++;
  59         return NULL;
  60     }
  61 }
  62
  63 robj *lookupKeyRead(redisDb *db, robj *key) {
  64     expireIfNeeded(db,key);
  65     return lookupKey(db,key);
  66 }
  67
  68 robj *lookupKeyWrite(redisDb *db, robj *key) {
  69     expireIfNeeded(db,key);
  70     return lookupKey(db,key);
  71 }
  72
  73 robj *lookupKeyReadOrReply(redisClient *c, robj *key, robj *reply) {
  74     robj *o = lookupKeyRead(c->db, key);
  75     if (!o) addReply(c,reply);
  76     return o;
  77 }
  78
  79 robj *lookupKeyWriteOrReply(redisClient *c, robj *key, robj *reply) {
  80     robj *o = lookupKeyWrite(c->db, key);
  81     if (!o) addReply(c,reply);
  82     return o;
  83 }
  84
  85 /* Add the key to the DB. If the key already exists REDIS_ERR is returned,
  86  * otherwise REDIS_OK is returned, and the caller should increment the
  87  * refcount of 'val'. */
  88 int dbAdd(redisDb *db, robj *key, robj *val) {
  89     /* Perform a lookup before adding the key, as we need to copy the
  90      * key value. */
  91     if (dictFind(db->dict, key->ptr) != NULL) {
  92         return REDIS_ERR;
  93     } else {
  94         sds copy = sdsdup(key->ptr);
  95         dictAdd(db->dict, copy, val);
  96         if (server.ds_enabled) {
  97             /* FIXME: remove entry from negative cache */
  98         }
  99         return REDIS_OK;
 100     }
 101 }
 102
 103 /* If the key does not exist, this is just like dbAdd(). Otherwise
 104  * the value associated to the key is replaced with the new one.
 105  *
 106  * On update (key already existed) 0 is returned. Otherwise 1. */
 107 int dbReplace(redisDb *db, robj *key, robj *val) {
 108     robj *oldval;
 109
 110     if ((oldval = dictFetchValue(db->dict,key->ptr)) == NULL) {
 111         sds copy = sdsdup(key->ptr);
 112         dictAdd(db->dict, copy, val);
 113         return 1;
 114     } else {
 115         dictReplace(db->dict, key->ptr, val);
 116         return 0;
 117     }
 118 }
 119
 120 int dbExists(redisDb *db, robj *key) {
 121     return dictFind(db->dict,key->ptr) != NULL;
 122 }
 123
 124 /* Return a random key, in form of a Redis object.
 125  * If there are no keys, NULL is returned.
 126  *
 127  * The function makes sure to return keys not already expired. */
 128 robj *dbRandomKey(redisDb *db) {
 129     struct dictEntry *de;
 130
 131     while(1) {
 132         sds key;
 133         robj *keyobj;
 134
 135         de = dictGetRandomKey(db->dict);
 136         if (de == NULL) return NULL;
 137
 138         key = dictGetEntryKey(de);
 139         keyobj = createStringObject(key,sdslen(key));
 140         if (dictFind(db->expires,key)) {
 141             if (expireIfNeeded(db,keyobj)) {
 142                 decrRefCount(keyobj);
 143                 continue; /* search for another key. This expired. */
 144             }
 145         }
 146         return keyobj;
 147     }
 148 }
 149
 150 /* Delete a key, value, and associated expiration entry if any, from the DB */
 151 int dbDelete(redisDb *db, robj *key) {
 152     /* If diskstore is enabled make sure to awake waiting clients for this key
 153      * as it is not really useful to wait for a key already deleted to be
 154      * loaded from disk. */
 155     if (server.ds_enabled) handleClientsBlockedOnSwappedKey(db,key);
 156
 157     /* FIXME: we should mark this key as non existing on disk in the negative
 158      * cache. */
 159
 160     /* Deleting an entry from the expires dict will not free the sds of
 161      * the key, because it is shared with the main dictionary. */
 162     if (dictSize(db->expires) > 0) dictDelete(db->expires,key->ptr);
 163     return dictDelete(db->dict,key->ptr) == DICT_OK;
 164 }
 165
 166 /* Empty the whole database */
 167 long long emptyDb() {
 168     int j;
 169     long long removed = 0;
 170
 171     for (j = 0; j < server.dbnum; j++) {
 172         removed += dictSize(server.db[j].dict);
 173         dictEmpty(server.db[j].dict);
 174         dictEmpty(server.db[j].expires);
 175     }
 176     return removed;
 177 }
 178
 179 int selectDb(redisClient *c, int id) {
 180     if (id < 0 || id >= server.dbnum)
 181         return REDIS_ERR;
 182     c->db = &server.db[id];
 183     return REDIS_OK;
 184 }
 185
 186 /*-----------------------------------------------------------------------------
 187  * Hooks for key space changes.
 188  *
 189  * Every time a key in the database is modified the function
 190  * signalModifiedKey() is called.
 191  *
 192  * Every time a DB is flushed the function signalFlushDb() is called.
 193  *----------------------------------------------------------------------------*/
 194
 195 void signalModifiedKey(redisDb *db, robj *key) {
 196     touchWatchedKey(db,key);
 197     if (server.ds_enabled)
 198         cacheScheduleIO(db,key,REDIS_IO_SAVE);
 199 }
 200
 201 void signalFlushedDb(int dbid) {
 202     touchWatchedKeysOnFlush(dbid);
 203     if (server.ds_enabled)
 204         dsFlushDb(dbid);
 205 }
 206
 207 /*-----------------------------------------------------------------------------
 208  * Type agnostic commands operating on the key space
 209  *----------------------------------------------------------------------------*/
 210
 211 void flushdbCommand(redisClient *c) {
 212     server.dirty += dictSize(c->db->dict);
 213     signalFlushedDb(c->db->id);
 214     dictEmpty(c->db->dict);
 215     dictEmpty(c->db->expires);
 216     addReply(c,shared.ok);
 217 }
 218
 219 void flushallCommand(redisClient *c) {
 220     signalFlushedDb(-1);
 221     server.dirty += emptyDb();
 222     addReply(c,shared.ok);
 223     if (server.bgsavechildpid != -1) {
 224         kill(server.bgsavechildpid,SIGKILL);
 225         rdbRemoveTempFile(server.bgsavechildpid);
 226     }
 227     rdbSave(server.dbfilename);
 228     server.dirty++;
 229 }
 230
 231 void delCommand(redisClient *c) {
 232     int deleted = 0, j;
 233
 234     for (j = 1; j < c->argc; j++) {
 235         if (server.ds_enabled) {
 236             lookupKeyRead(c->db,c->argv[j]);
 237             /* FIXME: this can be optimized a lot, no real need to load
 238              * a possibly huge value. */
 239         }
 240         if (dbDelete(c->db,c->argv[j])) {
 241             signalModifiedKey(c->db,c->argv[j]);
 242             server.dirty++;
 243             deleted++;
 244         } else if (server.ds_enabled) {
 245             if (cacheKeyMayExist(c->db,c->argv[j]) &&
 246                 dsExists(c->db,c->argv[j]))
 247             {
 248                 cacheScheduleIO(c->db,c->argv[j],REDIS_IO_SAVE);
 249                 deleted = 1;
 250             }
 251         }
 252     }
 253     addReplyLongLong(c,deleted);
 254 }
 255
 256 void existsCommand(redisClient *c) {
 257     expireIfNeeded(c->db,c->argv[1]);
 258     if (dbExists(c->db,c->argv[1])) {
 259         addReply(c, shared.cone);
 260     } else {
 261         addReply(c, shared.czero);
 262     }
 263 }
 264
 265 void selectCommand(redisClient *c) {
 266     int id = atoi(c->argv[1]->ptr);
 267
 268     if (selectDb(c,id) == REDIS_ERR) {
 269         addReplyError(c,"invalid DB index");
 270     } else {
 271         addReply(c,shared.ok);
 272     }
 273 }
 274
 275 void randomkeyCommand(redisClient *c) {
 276     robj *key;
 277
 278     if ((key = dbRandomKey(c->db)) == NULL) {
 279         addReply(c,shared.nullbulk);
 280         return;
 281     }
 282
 283     addReplyBulk(c,key);
 284     decrRefCount(key);
 285 }
 286
 287 void keysCommand(redisClient *c) {
 288     dictIterator *di;
 289     dictEntry *de;
 290     sds pattern = c->argv[1]->ptr;
 291     int plen = sdslen(pattern), allkeys;
 292     unsigned long numkeys = 0;
 293     void *replylen = addDeferredMultiBulkLength(c);
 294
 295     di = dictGetIterator(c->db->dict);
 296     allkeys = (pattern[0] == '*' && pattern[1] == '\0');
 297     while((de = dictNext(di)) != NULL) {
 298         sds key = dictGetEntryKey(de);
 299         robj *keyobj;
 300
 301         if (allkeys || stringmatchlen(pattern,plen,key,sdslen(key),0)) {
 302             keyobj = createStringObject(key,sdslen(key));
 303             if (expireIfNeeded(c->db,keyobj) == 0) {
 304                 addReplyBulk(c,keyobj);
 305                 numkeys++;
 306             }
 307             decrRefCount(keyobj);
 308         }
 309     }
 310     dictReleaseIterator(di);
 311     setDeferredMultiBulkLength(c,replylen,numkeys);
 312 }
 313
 314 void dbsizeCommand(redisClient *c) {
 315     addReplyLongLong(c,dictSize(c->db->dict));
 316 }
 317
 318 void lastsaveCommand(redisClient *c) {
 319     addReplyLongLong(c,server.lastsave);
 320 }
 321
 322 void typeCommand(redisClient *c) {
 323     robj *o;
 324     char *type;
 325
 326     o = lookupKeyRead(c->db,c->argv[1]);
 327     if (o == NULL) {
 328         type = "none";
 329     } else {
 330         switch(o->type) {
 331         case REDIS_STRING: type = "string"; break;
 332         case REDIS_LIST: type = "list"; break;
 333         case REDIS_SET: type = "set"; break;
 334         case REDIS_ZSET: type = "zset"; break;
 335         case REDIS_HASH: type = "hash"; break;
 336         default: type = "unknown"; break;
 337         }
 338     }
 339     addReplyStatus(c,type);
 340 }
 341
 342 void saveCommand(redisClient *c) {
 343     if (server.bgsavechildpid != -1) {
 344         addReplyError(c,"Background save already in progress");
 345         return;
 346     }
 347     if (rdbSave(server.dbfilename) == REDIS_OK) {
 348         addReply(c,shared.ok);
 349     } else {
 350         addReply(c,shared.err);
 351     }
 352 }
 353
 354 void bgsaveCommand(redisClient *c) {
 355     if (server.bgsavechildpid != -1) {
 356         addReplyError(c,"Background save already in progress");
 357         return;
 358     }
 359     if (rdbSaveBackground(server.dbfilename) == REDIS_OK) {
 360         addReplyStatus(c,"Background saving started");
 361     } else {
 362         addReply(c,shared.err);
 363     }
 364 }
 365
 366 void shutdownCommand(redisClient *c) {
 367     if (prepareForShutdown() == REDIS_OK)
 368         exit(0);
 369     addReplyError(c,"Errors trying to SHUTDOWN. Check logs.");
 370 }
 371
 372 void renameGenericCommand(redisClient *c, int nx) {
 373     robj *o;
 374
 375     /* To use the same key as src and dst is probably an error */
 376     if (sdscmp(c->argv[1]->ptr,c->argv[2]->ptr) == 0) {
 377         addReply(c,shared.sameobjecterr);
 378         return;
 379     }
 380
 381     if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr)) == NULL)
 382         return;
 383
 384     incrRefCount(o);
 385     if (dbAdd(c->db,c->argv[2],o) == REDIS_ERR) {
 386         if (nx) {
 387             decrRefCount(o);
 388             addReply(c,shared.czero);
 389             return;
 390         }
 391         dbReplace(c->db,c->argv[2],o);
 392     }
 393     dbDelete(c->db,c->argv[1]);
 394     signalModifiedKey(c->db,c->argv[1]);
 395     signalModifiedKey(c->db,c->argv[2]);
 396     server.dirty++;
 397     addReply(c,nx ? shared.cone : shared.ok);
 398 }
 399
 400 void renameCommand(redisClient *c) {
 401     renameGenericCommand(c,0);
 402 }
 403
 404 void renamenxCommand(redisClient *c) {
 405     renameGenericCommand(c,1);
 406 }
 407
 408 void moveCommand(redisClient *c) {
 409     robj *o;
 410     redisDb *src, *dst;
 411     int srcid;
 412
 413     /* Obtain source and target DB pointers */
 414     src = c->db;
 415     srcid = c->db->id;
 416     if (selectDb(c,atoi(c->argv[2]->ptr)) == REDIS_ERR) {
 417         addReply(c,shared.outofrangeerr);
 418         return;
 419     }
 420     dst = c->db;
 421     selectDb(c,srcid); /* Back to the source DB */
 422
 423     /* If the user is moving using as target the same
 424      * DB as the source DB it is probably an error. */
 425     if (src == dst) {
 426         addReply(c,shared.sameobjecterr);
 427         return;
 428     }
 429
 430     /* Check if the element exists and get a reference */
 431     o = lookupKeyWrite(c->db,c->argv[1]);
 432     if (!o) {
 433         addReply(c,shared.czero);
 434         return;
 435     }
 436
 437     /* Try to add the element to the target DB */
 438     if (dbAdd(dst,c->argv[1],o) == REDIS_ERR) {
 439         addReply(c,shared.czero);
 440         return;
 441     }
 442     incrRefCount(o);
 443
 444     /* OK! key moved, free the entry in the source DB */
 445     dbDelete(src,c->argv[1]);
 446     server.dirty++;
 447     addReply(c,shared.cone);
 448 }
 449
 450 /*-----------------------------------------------------------------------------
 451  * Expires API
 452  *----------------------------------------------------------------------------*/
 453
 454 int removeExpire(redisDb *db, robj *key) {
 455     /* An expire may only be removed if there is a corresponding entry in the
 456      * main dict. Otherwise, the key will never be freed. */
 457     redisAssert(dictFind(db->dict,key->ptr) != NULL);
 458     return dictDelete(db->expires,key->ptr) == DICT_OK;
 459 }
 460
 461 void setExpire(redisDb *db, robj *key, time_t when) {
 462     dictEntry *de;
 463
 464     /* Reuse the sds from the main dict in the expire dict */
 465     de = dictFind(db->dict,key->ptr);
 466     redisAssert(de != NULL);
 467     dictReplace(db->expires,dictGetEntryKey(de),(void*)when);
 468 }
 469
 470 /* Return the expire time of the specified key, or -1 if no expire
 471  * is associated with this key (i.e. the key is non volatile) */
 472 time_t getExpire(redisDb *db, robj *key) {
 473     dictEntry *de;
 474
 475     /* No expire? return ASAP */
 476     if (dictSize(db->expires) == 0 ||
 477        (de = dictFind(db->expires,key->ptr)) == NULL) return -1;
 478
 479     /* The entry was found in the expire dict, this means it should also
 480      * be present in the main dict (safety check). */
 481     redisAssert(dictFind(db->dict,key->ptr) != NULL);
 482     return (time_t) dictGetEntryVal(de);
 483 }
 484
 485 /* Propagate expires into slaves and the AOF file.
 486  * When a key expires in the master, a DEL operation for this key is sent
 487  * to all the slaves and the AOF file if enabled.
 488  *
 489  * This way the key expiry is centralized in one place, and since both
 490  * AOF and the master->slave link guarantee operation ordering, everything
 491  * will be consistent even if we allow write operations against expiring
 492  * keys. */
 493 void propagateExpire(redisDb *db, robj *key) {
 494     robj *argv[2];
 495
 496     argv[0] = createStringObject("DEL",3);
 497     argv[1] = key;
 498     incrRefCount(key);
 499
 500     if (server.appendonly)
 501         feedAppendOnlyFile(server.delCommand,db->id,argv,2);
 502     if (listLength(server.slaves))
 503         replicationFeedSlaves(server.slaves,db->id,argv,2);
 504
 505     decrRefCount(argv[0]);
 506     decrRefCount(argv[1]);
 507 }
 508
 509 int expireIfNeeded(redisDb *db, robj *key) {
 510     time_t when = getExpire(db,key);
 511
 512     /* If we are running in the context of a slave, return ASAP:
 513      * the slave key expiration is controlled by the master that will
 514      * send us synthesized DEL operations for expired keys.
 515      *
 516      * Still we try to return the right information to the caller,
 517      * that is, 0 if we think the key should be still valid, 1 if
 518      * we think the key is expired at this time. */
 519     if (server.masterhost != NULL) {
 520         return time(NULL) > when;
 521     }
 522
 523     if (when < 0) return 0;
 524
 525     /* Return when this key has not expired */
 526     if (time(NULL) <= when) return 0;
 527
 528     /* Delete the key */
 529     server.stat_expiredkeys++;
 530     propagateExpire(db,key);
 531     return dbDelete(db,key);
 532 }
 533
 534 /*-----------------------------------------------------------------------------
 535  * Expires Commands
 536  *----------------------------------------------------------------------------*/
 537
 538 void expireGenericCommand(redisClient *c, robj *key, robj *param, long offset) {
 539     dictEntry *de;
 540     long seconds;
 541
 542     if (getLongFromObjectOrReply(c, param, &seconds, NULL) != REDIS_OK) return;
 543
 544     seconds -= offset;
 545
 546     de = dictFind(c->db->dict,key->ptr);
 547     if (de == NULL) {
 548         addReply(c,shared.czero);
 549         return;
 550     }
 551     if (seconds <= 0) {
 552         if (dbDelete(c->db,key)) server.dirty++;
 553         addReply(c, shared.cone);
 554         signalModifiedKey(c->db,key);
 555         return;
 556     } else {
 557         time_t when = time(NULL)+seconds;
 558         setExpire(c->db,key,when);
 559         addReply(c,shared.cone);
 560         signalModifiedKey(c->db,key);
 561         server.dirty++;
 562         return;
 563     }
 564 }
 565
 566 void expireCommand(redisClient *c) {
 567     expireGenericCommand(c,c->argv[1],c->argv[2],0);
 568 }
 569
 570 void expireatCommand(redisClient *c) {
 571     expireGenericCommand(c,c->argv[1],c->argv[2],time(NULL));
 572 }
 573
 574 void ttlCommand(redisClient *c) {
 575     time_t expire, ttl = -1;
 576
 577     expire = getExpire(c->db,c->argv[1]);
 578     if (expire != -1) {
 579         ttl = (expire-time(NULL));
 580         if (ttl < 0) ttl = -1;
 581     }
 582     addReplyLongLong(c,(long long)ttl);
 583 }
 584
 585 void persistCommand(redisClient *c) {
 586     dictEntry *de;
 587
 588     de = dictFind(c->db->dict,c->argv[1]->ptr);
 589     if (de == NULL) {
 590         addReply(c,shared.czero);
 591     } else {
 592         if (removeExpire(c->db,c->argv[1])) {
 593             addReply(c,shared.cone);
 594             server.dirty++;
 595         } else {
 596             addReply(c,shared.czero);
 597         }
 598     }
 599 }