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