1 /* 2 * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com> 3 * All rights reserved. 4 * 5 * Redistribution and use in source and binary forms, with or without 6 * modification, are permitted provided that the following conditions are met: 7 * 8 * * Redistributions of source code must retain the above copyright notice, 9 * this list of conditions and the following disclaimer. 10 * * Redistributions in binary form must reproduce the above copyright 11 * notice, this list of conditions and the following disclaimer in the 12 * documentation and/or other materials provided with the distribution. 13 * * Neither the name of Redis nor the names of its contributors may be used 14 * to endorse or promote products derived from this software without 15 * specific prior written permission. 16 * 17 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 18 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 19 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 20 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 21 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 22 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 23 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 24 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 25 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 26 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 27 * POSSIBILITY OF SUCH DAMAGE. 28 */ 29 30 #include "server.h" 31 #include "cluster.h" 32 33 #include <signal.h> 34 #include <ctype.h> 35 36 void slotToKeyAdd(robj *key); 37 void slotToKeyDel(robj *key); 38 void slotToKeyFlush(void); 39 40 /*----------------------------------------------------------------------------- 41 * C-level DB API 42 *----------------------------------------------------------------------------*/ 43 44 robj *lookupKey(redisDb *db, robj *key) { 45 dictEntry *de = dictFind(db->dict,key->ptr); 46 if (de) { 47 robj *val = dictGetVal(de); 48 49 /* Update the access time for the ageing algorithm. 50 * Don't do it if we have a saving child, as this will trigger 51 * a copy on write madness. */ 52 if (server.rdb_child_pid == -1 && server.aof_child_pid == -1) 53 val->lru = LRU_CLOCK(); 54 return val; 55 } else { 56 return NULL; 57 } 58 } 59 60 robj *lookupKeyRead(redisDb *db, robj *key) { 61 robj *val; 62 63 if (expireIfNeeded(db,key) == 1) { 64 /* Key expired. If we are in the context of a master, expireIfNeeded() 65 * returns 0 only when the key does not exist at all, so it's save 66 * to return NULL ASAP. */ 67 if (server.masterhost == NULL) return NULL; 68 69 /* However if we are in the context of a slave, expireIfNeeded() will 70 * not really try to expire the key, it only returns information 71 * about the "logical" status of the key: key expiring is up to the 72 * master in order to have a consistent view of master's data set. 73 * 74 * However, if the command caller is not the master, and as additional 75 * safety measure, the command invoked is a read-only command, we can 76 * safely return NULL here, and provide a more consistent behavior 77 * to clients accessign expired values in a read-only fashion, that 78 * will say the key as non exisitng. 79 * 80 * Notably this covers GETs when slaves are used to scale reads. */ 81 if (server.current_client && 82 server.current_client != server.master && 83 server.current_client->cmd && 84 server.current_client->cmd->flags & CMD_READONLY) 85 { 86 return NULL; 87 } 88 } 89 val = lookupKey(db,key); 90 if (val == NULL) 91 server.stat_keyspace_misses++; 92 else 93 server.stat_keyspace_hits++; 94 return val; 95 } 96 97 robj *lookupKeyWrite(redisDb *db, robj *key) { 98 expireIfNeeded(db,key); 99 return lookupKey(db,key); 100 } 101 102 robj *lookupKeyReadOrReply(client *c, robj *key, robj *reply) { 103 robj *o = lookupKeyRead(c->db, key); 104 if (!o) addReply(c,reply); 105 return o; 106 } 107 108 robj *lookupKeyWriteOrReply(client *c, robj *key, robj *reply) { 109 robj *o = lookupKeyWrite(c->db, key); 110 if (!o) addReply(c,reply); 111 return o; 112 } 113 114 /* Add the key to the DB. It's up to the caller to increment the reference 115 * counter of the value if needed. 116 * 117 * The program is aborted if the key already exists. */ 118 void dbAdd(redisDb *db, robj *key, robj *val) { 119 sds copy = sdsdup(key->ptr); 120 int retval = dictAdd(db->dict, copy, val); 121 122 serverAssertWithInfo(NULL,key,retval == C_OK); 123 if (val->type == OBJ_LIST) signalListAsReady(db, key); 124 if (server.cluster_enabled) slotToKeyAdd(key); 125 } 126 127 /* Overwrite an existing key with a new value. Incrementing the reference 128 * count of the new value is up to the caller. 129 * This function does not modify the expire time of the existing key. 130 * 131 * The program is aborted if the key was not already present. */ 132 void dbOverwrite(redisDb *db, robj *key, robj *val) { 133 dictEntry *de = dictFind(db->dict,key->ptr); 134 135 serverAssertWithInfo(NULL,key,de != NULL); 136 dictReplace(db->dict, key->ptr, val); 137 } 138 139 /* High level Set operation. This function can be used in order to set 140 * a key, whatever it was existing or not, to a new object. 141 * 142 * 1) The ref count of the value object is incremented. 143 * 2) clients WATCHing for the destination key notified. 144 * 3) The expire time of the key is reset (the key is made persistent). */ 145 void setKey(redisDb *db, robj *key, robj *val) { 146 if (lookupKeyWrite(db,key) == NULL) { 147 dbAdd(db,key,val); 148 } else { 149 dbOverwrite(db,key,val); 150 } 151 incrRefCount(val); 152 removeExpire(db,key); 153 signalModifiedKey(db,key); 154 } 155 156 int dbExists(redisDb *db, robj *key) { 157 return dictFind(db->dict,key->ptr) != NULL; 158 } 159 160 /* Return a random key, in form of a Redis object. 161 * If there are no keys, NULL is returned. 162 * 163 * The function makes sure to return keys not already expired. */ 164 robj *dbRandomKey(redisDb *db) { 165 dictEntry *de; 166 167 while(1) { 168 sds key; 169 robj *keyobj; 170 171 de = dictGetRandomKey(db->dict); 172 if (de == NULL) return NULL; 173 174 key = dictGetKey(de); 175 keyobj = createStringObject(key,sdslen(key)); 176 if (dictFind(db->expires,key)) { 177 if (expireIfNeeded(db,keyobj)) { 178 decrRefCount(keyobj); 179 continue; /* search for another key. This expired. */ 180 } 181 } 182 return keyobj; 183 } 184 } 185 186 /* Delete a key, value, and associated expiration entry if any, from the DB */ 187 int dbDelete(redisDb *db, robj *key) { 188 /* Deleting an entry from the expires dict will not free the sds of 189 * the key, because it is shared with the main dictionary. */ 190 if (dictSize(db->expires) > 0) dictDelete(db->expires,key->ptr); 191 if (dictDelete(db->dict,key->ptr) == DICT_OK) { 192 if (server.cluster_enabled) slotToKeyDel(key); 193 return 1; 194 } else { 195 return 0; 196 } 197 } 198 199 /* Prepare the string object stored at 'key' to be modified destructively 200 * to implement commands like SETBIT or APPEND. 201 * 202 * An object is usually ready to be modified unless one of the two conditions 203 * are true: 204 * 205 * 1) The object 'o' is shared (refcount > 1), we don't want to affect 206 * other users. 207 * 2) The object encoding is not "RAW". 208 * 209 * If the object is found in one of the above conditions (or both) by the 210 * function, an unshared / not-encoded copy of the string object is stored 211 * at 'key' in the specified 'db'. Otherwise the object 'o' itself is 212 * returned. 213 * 214 * USAGE: 215 * 216 * The object 'o' is what the caller already obtained by looking up 'key' 217 * in 'db', the usage pattern looks like this: 218 * 219 * o = lookupKeyWrite(db,key); 220 * if (checkType(c,o,OBJ_STRING)) return; 221 * o = dbUnshareStringValue(db,key,o); 222 * 223 * At this point the caller is ready to modify the object, for example 224 * using an sdscat() call to append some data, or anything else. 225 */ 226 robj *dbUnshareStringValue(redisDb *db, robj *key, robj *o) { 227 serverAssert(o->type == OBJ_STRING); 228 if (o->refcount != 1 || o->encoding != OBJ_ENCODING_RAW) { 229 robj *decoded = getDecodedObject(o); 230 o = createRawStringObject(decoded->ptr, sdslen(decoded->ptr)); 231 decrRefCount(decoded); 232 dbOverwrite(db,key,o); 233 } 234 return o; 235 } 236 237 long long emptyDb(void(callback)(void*)) { 238 int j; 239 long long removed = 0; 240 241 for (j = 0; j < server.dbnum; j++) { 242 removed += dictSize(server.db[j].dict); 243 dictEmpty(server.db[j].dict,callback); 244 dictEmpty(server.db[j].expires,callback); 245 } 246 if (server.cluster_enabled) slotToKeyFlush(); 247 return removed; 248 } 249 250 int selectDb(client *c, int id) { 251 if (id < 0 || id >= server.dbnum) 252 return C_ERR; 253 c->db = &server.db[id]; 254 return C_OK; 255 } 256 257 /*----------------------------------------------------------------------------- 258 * Hooks for key space changes. 259 * 260 * Every time a key in the database is modified the function 261 * signalModifiedKey() is called. 262 * 263 * Every time a DB is flushed the function signalFlushDb() is called. 264 *----------------------------------------------------------------------------*/ 265 266 void signalModifiedKey(redisDb *db, robj *key) { 267 touchWatchedKey(db,key); 268 } 269 270 void signalFlushedDb(int dbid) { 271 touchWatchedKeysOnFlush(dbid); 272 } 273 274 /*----------------------------------------------------------------------------- 275 * Type agnostic commands operating on the key space 276 *----------------------------------------------------------------------------*/ 277 278 void flushdbCommand(client *c) { 279 server.dirty += dictSize(c->db->dict); 280 signalFlushedDb(c->db->id); 281 dictEmpty(c->db->dict,NULL); 282 dictEmpty(c->db->expires,NULL); 283 if (server.cluster_enabled) slotToKeyFlush(); 284 addReply(c,shared.ok); 285 } 286 287 void flushallCommand(client *c) { 288 signalFlushedDb(-1); 289 server.dirty += emptyDb(NULL); 290 addReply(c,shared.ok); 291 if (server.rdb_child_pid != -1) { 292 kill(server.rdb_child_pid,SIGUSR1); 293 rdbRemoveTempFile(server.rdb_child_pid); 294 } 295 if (server.saveparamslen > 0) { 296 /* Normally rdbSave() will reset dirty, but we don't want this here 297 * as otherwise FLUSHALL will not be replicated nor put into the AOF. */ 298 int saved_dirty = server.dirty; 299 rdbSave(server.rdb_filename); 300 server.dirty = saved_dirty; 301 } 302 server.dirty++; 303 } 304 305 void delCommand(client *c) { 306 int deleted = 0, j; 307 308 for (j = 1; j < c->argc; j++) { 309 expireIfNeeded(c->db,c->argv[j]); 310 if (dbDelete(c->db,c->argv[j])) { 311 signalModifiedKey(c->db,c->argv[j]); 312 notifyKeyspaceEvent(NOTIFY_GENERIC, 313 "del",c->argv[j],c->db->id); 314 server.dirty++; 315 deleted++; 316 } 317 } 318 addReplyLongLong(c,deleted); 319 } 320 321 /* EXISTS key1 key2 ... key_N. 322 * Return value is the number of keys existing. */ 323 void existsCommand(client *c) { 324 long long count = 0; 325 int j; 326 327 for (j = 1; j < c->argc; j++) { 328 expireIfNeeded(c->db,c->argv[j]); 329 if (dbExists(c->db,c->argv[j])) count++; 330 } 331 addReplyLongLong(c,count); 332 } 333 334 void selectCommand(client *c) { 335 long id; 336 337 if (getLongFromObjectOrReply(c, c->argv[1], &id, 338 "invalid DB index") != C_OK) 339 return; 340 341 if (server.cluster_enabled && id != 0) { 342 addReplyError(c,"SELECT is not allowed in cluster mode"); 343 return; 344 } 345 if (selectDb(c,id) == C_ERR) { 346 addReplyError(c,"invalid DB index"); 347 } else { 348 addReply(c,shared.ok); 349 } 350 } 351 352 void randomkeyCommand(client *c) { 353 robj *key; 354 355 if ((key = dbRandomKey(c->db)) == NULL) { 356 addReply(c,shared.nullbulk); 357 return; 358 } 359 360 addReplyBulk(c,key); 361 decrRefCount(key); 362 } 363 364 void keysCommand(client *c) { 365 dictIterator *di; 366 dictEntry *de; 367 sds pattern = c->argv[1]->ptr; 368 int plen = sdslen(pattern), allkeys; 369 unsigned long numkeys = 0; 370 void *replylen = addDeferredMultiBulkLength(c); 371 372 di = dictGetSafeIterator(c->db->dict); 373 allkeys = (pattern[0] == '*' && pattern[1] == '\0'); 374 while((de = dictNext(di)) != NULL) { 375 sds key = dictGetKey(de); 376 robj *keyobj; 377 378 if (allkeys || stringmatchlen(pattern,plen,key,sdslen(key),0)) { 379 keyobj = createStringObject(key,sdslen(key)); 380 if (expireIfNeeded(c->db,keyobj) == 0) { 381 addReplyBulk(c,keyobj); 382 numkeys++; 383 } 384 decrRefCount(keyobj); 385 } 386 } 387 dictReleaseIterator(di); 388 setDeferredMultiBulkLength(c,replylen,numkeys); 389 } 390 391 /* This callback is used by scanGenericCommand in order to collect elements 392 * returned by the dictionary iterator into a list. */ 393 void scanCallback(void *privdata, const dictEntry *de) { 394 void **pd = (void**) privdata; 395 list *keys = pd[0]; 396 robj *o = pd[1]; 397 robj *key, *val = NULL; 398 399 if (o == NULL) { 400 sds sdskey = dictGetKey(de); 401 key = createStringObject(sdskey, sdslen(sdskey)); 402 } else if (o->type == OBJ_SET) { 403 key = dictGetKey(de); 404 incrRefCount(key); 405 } else if (o->type == OBJ_HASH) { 406 key = dictGetKey(de); 407 incrRefCount(key); 408 val = dictGetVal(de); 409 incrRefCount(val); 410 } else if (o->type == OBJ_ZSET) { 411 key = dictGetKey(de); 412 incrRefCount(key); 413 val = createStringObjectFromLongDouble(*(double*)dictGetVal(de),0); 414 } else { 415 serverPanic("Type not handled in SCAN callback."); 416 } 417 418 listAddNodeTail(keys, key); 419 if (val) listAddNodeTail(keys, val); 420 } 421 422 /* Try to parse a SCAN cursor stored at object 'o': 423 * if the cursor is valid, store it as unsigned integer into *cursor and 424 * returns C_OK. Otherwise return C_ERR and send an error to the 425 * client. */ 426 int parseScanCursorOrReply(client *c, robj *o, unsigned long *cursor) { 427 char *eptr; 428 429 /* Use strtoul() because we need an *unsigned* long, so 430 * getLongLongFromObject() does not cover the whole cursor space. */ 431 errno = 0; 432 *cursor = strtoul(o->ptr, &eptr, 10); 433 if (isspace(((char*)o->ptr)[0]) || eptr[0] != '\0' || errno == ERANGE) 434 { 435 addReplyError(c, "invalid cursor"); 436 return C_ERR; 437 } 438 return C_OK; 439 } 440 441 /* This command implements SCAN, HSCAN and SSCAN commands. 442 * If object 'o' is passed, then it must be a Hash or Set object, otherwise 443 * if 'o' is NULL the command will operate on the dictionary associated with 444 * the current database. 445 * 446 * When 'o' is not NULL the function assumes that the first argument in 447 * the client arguments vector is a key so it skips it before iterating 448 * in order to parse options. 449 * 450 * In the case of a Hash object the function returns both the field and value 451 * of every element on the Hash. */ 452 void scanGenericCommand(client *c, robj *o, unsigned long cursor) { 453 int i, j; 454 list *keys = listCreate(); 455 listNode *node, *nextnode; 456 long count = 10; 457 sds pat = NULL; 458 int patlen = 0, use_pattern = 0; 459 dict *ht; 460 461 /* Object must be NULL (to iterate keys names), or the type of the object 462 * must be Set, Sorted Set, or Hash. */ 463 serverAssert(o == NULL || o->type == OBJ_SET || o->type == OBJ_HASH || 464 o->type == OBJ_ZSET); 465 466 /* Set i to the first option argument. The previous one is the cursor. */ 467 i = (o == NULL) ? 2 : 3; /* Skip the key argument if needed. */ 468 469 /* Step 1: Parse options. */ 470 while (i < c->argc) { 471 j = c->argc - i; 472 if (!strcasecmp(c->argv[i]->ptr, "count") && j >= 2) { 473 if (getLongFromObjectOrReply(c, c->argv[i+1], &count, NULL) 474 != C_OK) 475 { 476 goto cleanup; 477 } 478 479 if (count < 1) { 480 addReply(c,shared.syntaxerr); 481 goto cleanup; 482 } 483 484 i += 2; 485 } else if (!strcasecmp(c->argv[i]->ptr, "match") && j >= 2) { 486 pat = c->argv[i+1]->ptr; 487 patlen = sdslen(pat); 488 489 /* The pattern always matches if it is exactly "*", so it is 490 * equivalent to disabling it. */ 491 use_pattern = !(pat[0] == '*' && patlen == 1); 492 493 i += 2; 494 } else { 495 addReply(c,shared.syntaxerr); 496 goto cleanup; 497 } 498 } 499 500 /* Step 2: Iterate the collection. 501 * 502 * Note that if the object is encoded with a ziplist, intset, or any other 503 * representation that is not a hash table, we are sure that it is also 504 * composed of a small number of elements. So to avoid taking state we 505 * just return everything inside the object in a single call, setting the 506 * cursor to zero to signal the end of the iteration. */ 507 508 /* Handle the case of a hash table. */ 509 ht = NULL; 510 if (o == NULL) { 511 ht = c->db->dict; 512 } else if (o->type == OBJ_SET && o->encoding == OBJ_ENCODING_HT) { 513 ht = o->ptr; 514 } else if (o->type == OBJ_HASH && o->encoding == OBJ_ENCODING_HT) { 515 ht = o->ptr; 516 count *= 2; /* We return key / value for this type. */ 517 } else if (o->type == OBJ_ZSET && o->encoding == OBJ_ENCODING_SKIPLIST) { 518 zset *zs = o->ptr; 519 ht = zs->dict; 520 count *= 2; /* We return key / value for this type. */ 521 } 522 523 if (ht) { 524 void *privdata[2]; 525 /* We set the max number of iterations to ten times the specified 526 * COUNT, so if the hash table is in a pathological state (very 527 * sparsely populated) we avoid to block too much time at the cost 528 * of returning no or very few elements. */ 529 long maxiterations = count*10; 530 531 /* We pass two pointers to the callback: the list to which it will 532 * add new elements, and the object containing the dictionary so that 533 * it is possible to fetch more data in a type-dependent way. */ 534 privdata[0] = keys; 535 privdata[1] = o; 536 do { 537 cursor = dictScan(ht, cursor, scanCallback, privdata); 538 } while (cursor && 539 maxiterations-- && 540 listLength(keys) < (unsigned long)count); 541 } else if (o->type == OBJ_SET) { 542 int pos = 0; 543 int64_t ll; 544 545 while(intsetGet(o->ptr,pos++,&ll)) 546 listAddNodeTail(keys,createStringObjectFromLongLong(ll)); 547 cursor = 0; 548 } else if (o->type == OBJ_HASH || o->type == OBJ_ZSET) { 549 unsigned char *p = ziplistIndex(o->ptr,0); 550 unsigned char *vstr; 551 unsigned int vlen; 552 long long vll; 553 554 while(p) { 555 ziplistGet(p,&vstr,&vlen,&vll); 556 listAddNodeTail(keys, 557 (vstr != NULL) ? createStringObject((char*)vstr,vlen) : 558 createStringObjectFromLongLong(vll)); 559 p = ziplistNext(o->ptr,p); 560 } 561 cursor = 0; 562 } else { 563 serverPanic("Not handled encoding in SCAN."); 564 } 565 566 /* Step 3: Filter elements. */ 567 node = listFirst(keys); 568 while (node) { 569 robj *kobj = listNodeValue(node); 570 nextnode = listNextNode(node); 571 int filter = 0; 572 573 /* Filter element if it does not match the pattern. */ 574 if (!filter && use_pattern) { 575 if (sdsEncodedObject(kobj)) { 576 if (!stringmatchlen(pat, patlen, kobj->ptr, sdslen(kobj->ptr), 0)) 577 filter = 1; 578 } else { 579 char buf[LONG_STR_SIZE]; 580 int len; 581 582 serverAssert(kobj->encoding == OBJ_ENCODING_INT); 583 len = ll2string(buf,sizeof(buf),(long)kobj->ptr); 584 if (!stringmatchlen(pat, patlen, buf, len, 0)) filter = 1; 585 } 586 } 587 588 /* Filter element if it is an expired key. */ 589 if (!filter && o == NULL && expireIfNeeded(c->db, kobj)) filter = 1; 590 591 /* Remove the element and its associted value if needed. */ 592 if (filter) { 593 decrRefCount(kobj); 594 listDelNode(keys, node); 595 } 596 597 /* If this is a hash or a sorted set, we have a flat list of 598 * key-value elements, so if this element was filtered, remove the 599 * value, or skip it if it was not filtered: we only match keys. */ 600 if (o && (o->type == OBJ_ZSET || o->type == OBJ_HASH)) { 601 node = nextnode; 602 nextnode = listNextNode(node); 603 if (filter) { 604 kobj = listNodeValue(node); 605 decrRefCount(kobj); 606 listDelNode(keys, node); 607 } 608 } 609 node = nextnode; 610 } 611 612 /* Step 4: Reply to the client. */ 613 addReplyMultiBulkLen(c, 2); 614 addReplyBulkLongLong(c,cursor); 615 616 addReplyMultiBulkLen(c, listLength(keys)); 617 while ((node = listFirst(keys)) != NULL) { 618 robj *kobj = listNodeValue(node); 619 addReplyBulk(c, kobj); 620 decrRefCount(kobj); 621 listDelNode(keys, node); 622 } 623 624 cleanup: 625 listSetFreeMethod(keys,decrRefCountVoid); 626 listRelease(keys); 627 } 628 629 /* The SCAN command completely relies on scanGenericCommand. */ 630 void scanCommand(client *c) { 631 unsigned long cursor; 632 if (parseScanCursorOrReply(c,c->argv[1],&cursor) == C_ERR) return; 633 scanGenericCommand(c,NULL,cursor); 634 } 635 636 void dbsizeCommand(client *c) { 637 addReplyLongLong(c,dictSize(c->db->dict)); 638 } 639 640 void lastsaveCommand(client *c) { 641 addReplyLongLong(c,server.lastsave); 642 } 643 644 void typeCommand(client *c) { 645 robj *o; 646 char *type; 647 648 o = lookupKeyRead(c->db,c->argv[1]); 649 if (o == NULL) { 650 type = "none"; 651 } else { 652 switch(o->type) { 653 case OBJ_STRING: type = "string"; break; 654 case OBJ_LIST: type = "list"; break; 655 case OBJ_SET: type = "set"; break; 656 case OBJ_ZSET: type = "zset"; break; 657 case OBJ_HASH: type = "hash"; break; 658 default: type = "unknown"; break; 659 } 660 } 661 addReplyStatus(c,type); 662 } 663 664 void shutdownCommand(client *c) { 665 int flags = 0; 666 667 if (c->argc > 2) { 668 addReply(c,shared.syntaxerr); 669 return; 670 } else if (c->argc == 2) { 671 if (!strcasecmp(c->argv[1]->ptr,"nosave")) { 672 flags |= SHUTDOWN_NOSAVE; 673 } else if (!strcasecmp(c->argv[1]->ptr,"save")) { 674 flags |= SHUTDOWN_SAVE; 675 } else { 676 addReply(c,shared.syntaxerr); 677 return; 678 } 679 } 680 /* When SHUTDOWN is called while the server is loading a dataset in 681 * memory we need to make sure no attempt is performed to save 682 * the dataset on shutdown (otherwise it could overwrite the current DB 683 * with half-read data). 684 * 685 * Also when in Sentinel mode clear the SAVE flag and force NOSAVE. */ 686 if (server.loading || server.sentinel_mode) 687 flags = (flags & ~SHUTDOWN_SAVE) | SHUTDOWN_NOSAVE; 688 if (prepareForShutdown(flags) == C_OK) exit(0); 689 addReplyError(c,"Errors trying to SHUTDOWN. Check logs."); 690 } 691 692 void renameGenericCommand(client *c, int nx) { 693 robj *o; 694 long long expire; 695 int samekey = 0; 696 697 /* When source and dest key is the same, no operation is performed, 698 * if the key exists, however we still return an error on unexisting key. */ 699 if (sdscmp(c->argv[1]->ptr,c->argv[2]->ptr) == 0) samekey = 1; 700 701 if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr)) == NULL) 702 return; 703 704 if (samekey) { 705 addReply(c,nx ? shared.czero : shared.ok); 706 return; 707 } 708 709 incrRefCount(o); 710 expire = getExpire(c->db,c->argv[1]); 711 if (lookupKeyWrite(c->db,c->argv[2]) != NULL) { 712 if (nx) { 713 decrRefCount(o); 714 addReply(c,shared.czero); 715 return; 716 } 717 /* Overwrite: delete the old key before creating the new one 718 * with the same name. */ 719 dbDelete(c->db,c->argv[2]); 720 } 721 dbAdd(c->db,c->argv[2],o); 722 if (expire != -1) setExpire(c->db,c->argv[2],expire); 723 dbDelete(c->db,c->argv[1]); 724 signalModifiedKey(c->db,c->argv[1]); 725 signalModifiedKey(c->db,c->argv[2]); 726 notifyKeyspaceEvent(NOTIFY_GENERIC,"rename_from", 727 c->argv[1],c->db->id); 728 notifyKeyspaceEvent(NOTIFY_GENERIC,"rename_to", 729 c->argv[2],c->db->id); 730 server.dirty++; 731 addReply(c,nx ? shared.cone : shared.ok); 732 } 733 734 void renameCommand(client *c) { 735 renameGenericCommand(c,0); 736 } 737 738 void renamenxCommand(client *c) { 739 renameGenericCommand(c,1); 740 } 741 742 void moveCommand(client *c) { 743 robj *o; 744 redisDb *src, *dst; 745 int srcid; 746 long long dbid, expire; 747 748 if (server.cluster_enabled) { 749 addReplyError(c,"MOVE is not allowed in cluster mode"); 750 return; 751 } 752 753 /* Obtain source and target DB pointers */ 754 src = c->db; 755 srcid = c->db->id; 756 757 if (getLongLongFromObject(c->argv[2],&dbid) == C_ERR || 758 dbid < INT_MIN || dbid > INT_MAX || 759 selectDb(c,dbid) == C_ERR) 760 { 761 addReply(c,shared.outofrangeerr); 762 return; 763 } 764 dst = c->db; 765 selectDb(c,srcid); /* Back to the source DB */ 766 767 /* If the user is moving using as target the same 768 * DB as the source DB it is probably an error. */ 769 if (src == dst) { 770 addReply(c,shared.sameobjecterr); 771 return; 772 } 773 774 /* Check if the element exists and get a reference */ 775 o = lookupKeyWrite(c->db,c->argv[1]); 776 if (!o) { 777 addReply(c,shared.czero); 778 return; 779 } 780 expire = getExpire(c->db,c->argv[1]); 781 782 /* Return zero if the key already exists in the target DB */ 783 if (lookupKeyWrite(dst,c->argv[1]) != NULL) { 784 addReply(c,shared.czero); 785 return; 786 } 787 dbAdd(dst,c->argv[1],o); 788 if (expire != -1) setExpire(dst,c->argv[1],expire); 789 incrRefCount(o); 790 791 /* OK! key moved, free the entry in the source DB */ 792 dbDelete(src,c->argv[1]); 793 server.dirty++; 794 addReply(c,shared.cone); 795 } 796 797 /*----------------------------------------------------------------------------- 798 * Expires API 799 *----------------------------------------------------------------------------*/ 800 801 int removeExpire(redisDb *db, robj *key) { 802 /* An expire may only be removed if there is a corresponding entry in the 803 * main dict. Otherwise, the key will never be freed. */ 804 serverAssertWithInfo(NULL,key,dictFind(db->dict,key->ptr) != NULL); 805 return dictDelete(db->expires,key->ptr) == DICT_OK; 806 } 807 808 void setExpire(redisDb *db, robj *key, long long when) { 809 dictEntry *kde, *de; 810 811 /* Reuse the sds from the main dict in the expire dict */ 812 kde = dictFind(db->dict,key->ptr); 813 serverAssertWithInfo(NULL,key,kde != NULL); 814 de = dictReplaceRaw(db->expires,dictGetKey(kde)); 815 dictSetSignedIntegerVal(de,when); 816 } 817 818 /* Return the expire time of the specified key, or -1 if no expire 819 * is associated with this key (i.e. the key is non volatile) */ 820 long long getExpire(redisDb *db, robj *key) { 821 dictEntry *de; 822 823 /* No expire? return ASAP */ 824 if (dictSize(db->expires) == 0 || 825 (de = dictFind(db->expires,key->ptr)) == NULL) return -1; 826 827 /* The entry was found in the expire dict, this means it should also 828 * be present in the main dict (safety check). */ 829 serverAssertWithInfo(NULL,key,dictFind(db->dict,key->ptr) != NULL); 830 return dictGetSignedIntegerVal(de); 831 } 832 833 /* Propagate expires into slaves and the AOF file. 834 * When a key expires in the master, a DEL operation for this key is sent 835 * to all the slaves and the AOF file if enabled. 836 * 837 * This way the key expiry is centralized in one place, and since both 838 * AOF and the master->slave link guarantee operation ordering, everything 839 * will be consistent even if we allow write operations against expiring 840 * keys. */ 841 void propagateExpire(redisDb *db, robj *key) { 842 robj *argv[2]; 843 844 argv[0] = shared.del; 845 argv[1] = key; 846 incrRefCount(argv[0]); 847 incrRefCount(argv[1]); 848 849 if (server.aof_state != AOF_OFF) 850 feedAppendOnlyFile(server.delCommand,db->id,argv,2); 851 replicationFeedSlaves(server.slaves,db->id,argv,2); 852 853 decrRefCount(argv[0]); 854 decrRefCount(argv[1]); 855 } 856 857 int expireIfNeeded(redisDb *db, robj *key) { 858 mstime_t when = getExpire(db,key); 859 mstime_t now; 860 861 if (when < 0) return 0; /* No expire for this key */ 862 863 /* Don't expire anything while loading. It will be done later. */ 864 if (server.loading) return 0; 865 866 /* If we are in the context of a Lua script, we claim that time is 867 * blocked to when the Lua script started. This way a key can expire 868 * only the first time it is accessed and not in the middle of the 869 * script execution, making propagation to slaves / AOF consistent. 870 * See issue #1525 on Github for more information. */ 871 now = server.lua_caller ? server.lua_time_start : mstime(); 872 873 /* If we are running in the context of a slave, return ASAP: 874 * the slave key expiration is controlled by the master that will 875 * send us synthesized DEL operations for expired keys. 876 * 877 * Still we try to return the right information to the caller, 878 * that is, 0 if we think the key should be still valid, 1 if 879 * we think the key is expired at this time. */ 880 if (server.masterhost != NULL) return now > when; 881 882 /* Return when this key has not expired */ 883 if (now <= when) return 0; 884 885 /* Delete the key */ 886 server.stat_expiredkeys++; 887 propagateExpire(db,key); 888 notifyKeyspaceEvent(NOTIFY_EXPIRED, 889 "expired",key,db->id); 890 return dbDelete(db,key); 891 } 892 893 /*----------------------------------------------------------------------------- 894 * Expires Commands 895 *----------------------------------------------------------------------------*/ 896 897 /* This is the generic command implementation for EXPIRE, PEXPIRE, EXPIREAT 898 * and PEXPIREAT. Because the commad second argument may be relative or absolute 899 * the "basetime" argument is used to signal what the base time is (either 0 900 * for *AT variants of the command, or the current time for relative expires). 901 * 902 * unit is either UNIT_SECONDS or UNIT_MILLISECONDS, and is only used for 903 * the argv[2] parameter. The basetime is always specified in milliseconds. */ 904 void expireGenericCommand(client *c, long long basetime, int unit) { 905 robj *key = c->argv[1], *param = c->argv[2]; 906 long long when; /* unix time in milliseconds when the key will expire. */ 907 908 if (getLongLongFromObjectOrReply(c, param, &when, NULL) != C_OK) 909 return; 910 911 if (unit == UNIT_SECONDS) when *= 1000; 912 when += basetime; 913 914 /* No key, return zero. */ 915 if (lookupKeyWrite(c->db,key) == NULL) { 916 addReply(c,shared.czero); 917 return; 918 } 919 920 /* EXPIRE with negative TTL, or EXPIREAT with a timestamp into the past 921 * should never be executed as a DEL when load the AOF or in the context 922 * of a slave instance. 923 * 924 * Instead we take the other branch of the IF statement setting an expire 925 * (possibly in the past) and wait for an explicit DEL from the master. */ 926 if (when <= mstime() && !server.loading && !server.masterhost) { 927 robj *aux; 928 929 serverAssertWithInfo(c,key,dbDelete(c->db,key)); 930 server.dirty++; 931 932 /* Replicate/AOF this as an explicit DEL. */ 933 aux = createStringObject("DEL",3); 934 rewriteClientCommandVector(c,2,aux,key); 935 decrRefCount(aux); 936 signalModifiedKey(c->db,key); 937 notifyKeyspaceEvent(NOTIFY_GENERIC,"del",key,c->db->id); 938 addReply(c, shared.cone); 939 return; 940 } else { 941 setExpire(c->db,key,when); 942 addReply(c,shared.cone); 943 signalModifiedKey(c->db,key); 944 notifyKeyspaceEvent(NOTIFY_GENERIC,"expire",key,c->db->id); 945 server.dirty++; 946 return; 947 } 948 } 949 950 void expireCommand(client *c) { 951 expireGenericCommand(c,mstime(),UNIT_SECONDS); 952 } 953 954 void expireatCommand(client *c) { 955 expireGenericCommand(c,0,UNIT_SECONDS); 956 } 957 958 void pexpireCommand(client *c) { 959 expireGenericCommand(c,mstime(),UNIT_MILLISECONDS); 960 } 961 962 void pexpireatCommand(client *c) { 963 expireGenericCommand(c,0,UNIT_MILLISECONDS); 964 } 965 966 void ttlGenericCommand(client *c, int output_ms) { 967 long long expire, ttl = -1; 968 969 /* If the key does not exist at all, return -2 */ 970 if (lookupKeyRead(c->db,c->argv[1]) == NULL) { 971 addReplyLongLong(c,-2); 972 return; 973 } 974 /* The key exists. Return -1 if it has no expire, or the actual 975 * TTL value otherwise. */ 976 expire = getExpire(c->db,c->argv[1]); 977 if (expire != -1) { 978 ttl = expire-mstime(); 979 if (ttl < 0) ttl = 0; 980 } 981 if (ttl == -1) { 982 addReplyLongLong(c,-1); 983 } else { 984 addReplyLongLong(c,output_ms ? ttl : ((ttl+500)/1000)); 985 } 986 } 987 988 void ttlCommand(client *c) { 989 ttlGenericCommand(c, 0); 990 } 991 992 void pttlCommand(client *c) { 993 ttlGenericCommand(c, 1); 994 } 995 996 void persistCommand(client *c) { 997 dictEntry *de; 998 999 de = dictFind(c->db->dict,c->argv[1]->ptr); 1000 if (de == NULL) { 1001 addReply(c,shared.czero); 1002 } else { 1003 if (removeExpire(c->db,c->argv[1])) { 1004 addReply(c,shared.cone); 1005 server.dirty++; 1006 } else { 1007 addReply(c,shared.czero); 1008 } 1009 } 1010 } 1011 1012 /* ----------------------------------------------------------------------------- 1013 * API to get key arguments from commands 1014 * ---------------------------------------------------------------------------*/ 1015 1016 /* The base case is to use the keys position as given in the command table 1017 * (firstkey, lastkey, step). */ 1018 int *getKeysUsingCommandTable(struct redisCommand *cmd,robj **argv, int argc, int *numkeys) { 1019 int j, i = 0, last, *keys; 1020 UNUSED(argv); 1021 1022 if (cmd->firstkey == 0) { 1023 *numkeys = 0; 1024 return NULL; 1025 } 1026 last = cmd->lastkey; 1027 if (last < 0) last = argc+last; 1028 keys = zmalloc(sizeof(int)*((last - cmd->firstkey)+1)); 1029 for (j = cmd->firstkey; j <= last; j += cmd->keystep) { 1030 serverAssert(j < argc); 1031 keys[i++] = j; 1032 } 1033 *numkeys = i; 1034 return keys; 1035 } 1036 1037 /* Return all the arguments that are keys in the command passed via argc / argv. 1038 * 1039 * The command returns the positions of all the key arguments inside the array, 1040 * so the actual return value is an heap allocated array of integers. The 1041 * length of the array is returned by reference into *numkeys. 1042 * 1043 * 'cmd' must be point to the corresponding entry into the redisCommand 1044 * table, according to the command name in argv[0]. 1045 * 1046 * This function uses the command table if a command-specific helper function 1047 * is not required, otherwise it calls the command-specific function. */ 1048 int *getKeysFromCommand(struct redisCommand *cmd, robj **argv, int argc, int *numkeys) { 1049 if (cmd->getkeys_proc) { 1050 return cmd->getkeys_proc(cmd,argv,argc,numkeys); 1051 } else { 1052 return getKeysUsingCommandTable(cmd,argv,argc,numkeys); 1053 } 1054 } 1055 1056 /* Free the result of getKeysFromCommand. */ 1057 void getKeysFreeResult(int *result) { 1058 zfree(result); 1059 } 1060 1061 /* Helper function to extract keys from following commands: 1062 * ZUNIONSTORE <destkey> <num-keys> <key> <key> ... <key> <options> 1063 * ZINTERSTORE <destkey> <num-keys> <key> <key> ... <key> <options> */ 1064 int *zunionInterGetKeys(struct redisCommand *cmd, robj **argv, int argc, int *numkeys) { 1065 int i, num, *keys; 1066 UNUSED(cmd); 1067 1068 num = atoi(argv[2]->ptr); 1069 /* Sanity check. Don't return any key if the command is going to 1070 * reply with syntax error. */ 1071 if (num > (argc-3)) { 1072 *numkeys = 0; 1073 return NULL; 1074 } 1075 1076 /* Keys in z{union,inter}store come from two places: 1077 * argv[1] = storage key, 1078 * argv[3...n] = keys to intersect */ 1079 keys = zmalloc(sizeof(int)*(num+1)); 1080 1081 /* Add all key positions for argv[3...n] to keys[] */ 1082 for (i = 0; i < num; i++) keys[i] = 3+i; 1083 1084 /* Finally add the argv[1] key position (the storage key target). */ 1085 keys[num] = 1; 1086 *numkeys = num+1; /* Total keys = {union,inter} keys + storage key */ 1087 return keys; 1088 } 1089 1090 /* Helper function to extract keys from the following commands: 1091 * EVAL <script> <num-keys> <key> <key> ... <key> [more stuff] 1092 * EVALSHA <script> <num-keys> <key> <key> ... <key> [more stuff] */ 1093 int *evalGetKeys(struct redisCommand *cmd, robj **argv, int argc, int *numkeys) { 1094 int i, num, *keys; 1095 UNUSED(cmd); 1096 1097 num = atoi(argv[2]->ptr); 1098 /* Sanity check. Don't return any key if the command is going to 1099 * reply with syntax error. */ 1100 if (num > (argc-3)) { 1101 *numkeys = 0; 1102 return NULL; 1103 } 1104 1105 keys = zmalloc(sizeof(int)*num); 1106 *numkeys = num; 1107 1108 /* Add all key positions for argv[3...n] to keys[] */ 1109 for (i = 0; i < num; i++) keys[i] = 3+i; 1110 1111 return keys; 1112 } 1113 1114 /* Helper function to extract keys from the SORT command. 1115 * 1116 * SORT <sort-key> ... STORE <store-key> ... 1117 * 1118 * The first argument of SORT is always a key, however a list of options 1119 * follow in SQL-alike style. Here we parse just the minimum in order to 1120 * correctly identify keys in the "STORE" option. */ 1121 int *sortGetKeys(struct redisCommand *cmd, robj **argv, int argc, int *numkeys) { 1122 int i, j, num, *keys, found_store = 0; 1123 UNUSED(cmd); 1124 1125 num = 0; 1126 keys = zmalloc(sizeof(int)*2); /* Alloc 2 places for the worst case. */ 1127 1128 keys[num++] = 1; /* <sort-key> is always present. */ 1129 1130 /* Search for STORE option. By default we consider options to don't 1131 * have arguments, so if we find an unknown option name we scan the 1132 * next. However there are options with 1 or 2 arguments, so we 1133 * provide a list here in order to skip the right number of args. */ 1134 struct { 1135 char *name; 1136 int skip; 1137 } skiplist[] = { 1138 {"limit", 2}, 1139 {"get", 1}, 1140 {"by", 1}, 1141 {NULL, 0} /* End of elements. */ 1142 }; 1143 1144 for (i = 2; i < argc; i++) { 1145 for (j = 0; skiplist[j].name != NULL; j++) { 1146 if (!strcasecmp(argv[i]->ptr,skiplist[j].name)) { 1147 i += skiplist[j].skip; 1148 break; 1149 } else if (!strcasecmp(argv[i]->ptr,"store") && i+1 < argc) { 1150 /* Note: we don't increment "num" here and continue the loop 1151 * to be sure to process the *last* "STORE" option if multiple 1152 * ones are provided. This is same behavior as SORT. */ 1153 found_store = 1; 1154 keys[num] = i+1; /* <store-key> */ 1155 break; 1156 } 1157 } 1158 } 1159 *numkeys = num + found_store; 1160 return keys; 1161 } 1162 1163 /* Slot to Key API. This is used by Redis Cluster in order to obtain in 1164 * a fast way a key that belongs to a specified hash slot. This is useful 1165 * while rehashing the cluster. */ 1166 void slotToKeyAdd(robj *key) { 1167 unsigned int hashslot = keyHashSlot(key->ptr,sdslen(key->ptr)); 1168 1169 zslInsert(server.cluster->slots_to_keys,hashslot,key); 1170 incrRefCount(key); 1171 } 1172 1173 void slotToKeyDel(robj *key) { 1174 unsigned int hashslot = keyHashSlot(key->ptr,sdslen(key->ptr)); 1175 1176 zslDelete(server.cluster->slots_to_keys,hashslot,key); 1177 } 1178 1179 void slotToKeyFlush(void) { 1180 zslFree(server.cluster->slots_to_keys); 1181 server.cluster->slots_to_keys = zslCreate(); 1182 } 1183 1184 unsigned int getKeysInSlot(unsigned int hashslot, robj **keys, unsigned int count) { 1185 zskiplistNode *n; 1186 zrangespec range; 1187 int j = 0; 1188 1189 range.min = range.max = hashslot; 1190 range.minex = range.maxex = 0; 1191 1192 n = zslFirstInRange(server.cluster->slots_to_keys, &range); 1193 while(n && n->score == hashslot && count--) { 1194 keys[j++] = n->obj; 1195 n = n->level[0].forward; 1196 } 1197 return j; 1198 } 1199 1200 /* Remove all the keys in the specified hash slot. 1201 * The number of removed items is returned. */ 1202 unsigned int delKeysInSlot(unsigned int hashslot) { 1203 zskiplistNode *n; 1204 zrangespec range; 1205 int j = 0; 1206 1207 range.min = range.max = hashslot; 1208 range.minex = range.maxex = 0; 1209 1210 n = zslFirstInRange(server.cluster->slots_to_keys, &range); 1211 while(n && n->score == hashslot) { 1212 robj *key = n->obj; 1213 n = n->level[0].forward; /* Go to the next item before freeing it. */ 1214 incrRefCount(key); /* Protect the object while freeing it. */ 1215 dbDelete(&server.db[0],key); 1216 decrRefCount(key); 1217 j++; 1218 } 1219 return j; 1220 } 1221 1222 unsigned int countKeysInSlot(unsigned int hashslot) { 1223 zskiplist *zsl = server.cluster->slots_to_keys; 1224 zskiplistNode *zn; 1225 zrangespec range; 1226 int rank, count = 0; 1227 1228 range.min = range.max = hashslot; 1229 range.minex = range.maxex = 0; 1230 1231 /* Find first element in range */ 1232 zn = zslFirstInRange(zsl, &range); 1233 1234 /* Use rank of first element, if any, to determine preliminary count */ 1235 if (zn != NULL) { 1236 rank = zslGetRank(zsl, zn->score, zn->obj); 1237 count = (zsl->length - (rank - 1)); 1238 1239 /* Find last element in range */ 1240 zn = zslLastInRange(zsl, &range); 1241 1242 /* Use rank of last element, if any, to determine the actual count */ 1243 if (zn != NULL) { 1244 rank = zslGetRank(zsl, zn->score, zn->obj); 1245 count -= (zsl->length - rank); 1246 } 1247 } 1248 return count; 1249 } 1250