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