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