#define REDIS_CONFIGLINE_MAX 1024
#define REDIS_OBJFREELIST_MAX 1000000 /* Max number of objects to cache */
#define REDIS_MAX_SYNC_TIME 60 /* Slave can't take more to sync */
-#define REDIS_EXPIRELOOKUPS_PER_CRON 10 /* try to expire 10 keys/loop */
+#define REDIS_EXPIRELOOKUPS_PER_CRON 10 /* lookup 10 expires per loop */
#define REDIS_MAX_WRITE_PER_EVENT (1024*64)
#define REDIS_REQUEST_MAX_SIZE (1024*1024*256) /* max bytes in inline command */
char *requirepass;
int shareobjects;
int rdbcompression;
+ int activerehashing;
/* Replication related */
int isslave;
char *masterauth;
static void zrankCommand(redisClient *c);
static void zrevrankCommand(redisClient *c);
static void hsetCommand(redisClient *c);
-static void hmsetCommand(redisClient *c);
static void hgetCommand(redisClient *c);
+static void hmsetCommand(redisClient *c);
+static void hmgetCommand(redisClient *c);
static void hdelCommand(redisClient *c);
static void hlenCommand(redisClient *c);
static void zremrangebyrankCommand(redisClient *c);
{"zrank",zrankCommand,3,REDIS_CMD_BULK,NULL,1,1,1},
{"zrevrank",zrevrankCommand,3,REDIS_CMD_BULK,NULL,1,1,1},
{"hset",hsetCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
+ {"hget",hgetCommand,3,REDIS_CMD_BULK,NULL,1,1,1},
{"hmset",hmsetCommand,-4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
+ {"hmget",hmgetCommand,-3,REDIS_CMD_BULK,NULL,1,1,1},
{"hincrby",hincrbyCommand,4,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,NULL,1,1,1},
- {"hget",hgetCommand,3,REDIS_CMD_BULK,NULL,1,1,1},
{"hdel",hdelCommand,3,REDIS_CMD_BULK,NULL,1,1,1},
{"hlen",hlenCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
{"hkeys",hkeysCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
int j;
for (j = 0; j < server.dbnum; j++) {
- if (htNeedsResize(server.db[j].dict)) {
- redisLog(REDIS_VERBOSE,"The hash table %d is too sparse, resize it...",j);
+ if (htNeedsResize(server.db[j].dict))
dictResize(server.db[j].dict);
- redisLog(REDIS_VERBOSE,"Hash table %d resized.",j);
- }
if (htNeedsResize(server.db[j].expires))
dictResize(server.db[j].expires);
}
}
+/* Our hash table implementation performs rehashing incrementally while
+ * we write/read from the hash table. Still if the server is idle, the hash
+ * table will use two tables for a long time. So we try to use 1 millisecond
+ * of CPU time at every serverCron() loop in order to rehash some key. */
+static void incrementallyRehash(void) {
+ int j;
+
+ for (j = 0; j < server.dbnum; j++) {
+ if (dictIsRehashing(server.db[j].dict)) {
+ dictRehashMilliseconds(server.db[j].dict,1);
+ break; /* already used our millisecond for this loop... */
+ }
+ }
+}
+
/* A background saving child (BGSAVE) terminated its work. Handle this. */
void backgroundSaveDoneHandler(int statloc) {
int exitcode = WEXITSTATUS(statloc);
* if we resize the HT while there is the saving child at work actually
* a lot of memory movements in the parent will cause a lot of pages
* copied. */
- if (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1 &&
- !(loops % 10))
- {
- tryResizeHashTables();
+ if (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1) {
+ if (!(loops % 10)) tryResizeHashTables();
+ if (server.activerehashing) incrementallyRehash();
}
/* Show information about connected clients */
server.requirepass = NULL;
server.shareobjects = 0;
server.rdbcompression = 1;
+ server.activerehashing = 1;
server.maxclients = 0;
server.blpop_blocked_clients = 0;
server.maxmemory = 0;
char buf[REDIS_CONFIGLINE_MAX+1], *err = NULL;
int linenum = 0;
sds line = NULL;
- char *errormsg = "Fatal error, can't open config file '%s'";
- char *errorbuf = zmalloc(sizeof(char)*(strlen(errormsg)+strlen(filename)));
- sprintf(errorbuf, errormsg, filename);
if (filename[0] == '-' && filename[1] == '\0')
fp = stdin;
else {
if ((fp = fopen(filename,"r")) == NULL) {
- redisLog(REDIS_WARNING, errorbuf);
+ redisLog(REDIS_WARNING, "Fatal error, can't open config file '%s'", filename);
exit(1);
}
}
if ((server.rdbcompression = yesnotoi(argv[1])) == -1) {
err = "argument must be 'yes' or 'no'"; goto loaderr;
}
+ } else if (!strcasecmp(argv[0],"activerehashing") && argc == 2) {
+ if ((server.activerehashing = yesnotoi(argv[1])) == -1) {
+ err = "argument must be 'yes' or 'no'"; goto loaderr;
+ }
} else if (!strcasecmp(argv[0],"daemonize") && argc == 2) {
if ((server.daemonize = yesnotoi(argv[1])) == -1) {
err = "argument must be 'yes' or 'no'"; goto loaderr;
/* Use writev() if we have enough buffers to send */
if (!server.glueoutputbuf &&
- listLength(c->reply) > REDIS_WRITEV_THRESHOLD &&
+ listLength(c->reply) > REDIS_WRITEV_THRESHOLD &&
!(c->flags & REDIS_MASTER))
{
sendReplyToClientWritev(el, fd, privdata, mask);
o = listNodeValue(node);
objlen = sdslen(o->ptr);
- if (totwritten + objlen - offset > REDIS_MAX_WRITE_PER_EVENT)
+ if (totwritten + objlen - offset > REDIS_MAX_WRITE_PER_EVENT)
break;
if(ion == REDIS_WRITEV_IOVEC_COUNT)
}
}
- if (totwritten > 0)
+ if (totwritten > 0)
c->lastinteraction = time(NULL);
if (listLength(c->reply) == 0) {
}
/* Only allow SUBSCRIBE and UNSUBSCRIBE in the context of Pub/Sub */
- if (dictSize(c->pubsub_channels) > 0 &&
+ if ((dictSize(c->pubsub_channels) > 0 || listLength(c->pubsub_patterns) > 0)
+ &&
cmd->proc != subscribeCommand && cmd->proc != unsubscribeCommand &&
cmd->proc != psubscribeCommand && cmd->proc != punsubscribeCommand) {
addReplySds(c,sdsnew("-ERR only (P)SUBSCRIBE / (P)UNSUBSCRIBE / QUIT allowed in this context\r\n"));
if (p) {
sds query, *argv;
int argc, j;
-
+
query = c->querybuf;
c->querybuf = sdsempty();
querylen = 1+(p-(query));
char buf[32], *endptr;
long value;
int slen;
-
+
value = strtol(s, &endptr, 10);
if (endptr[0] != '\0') return REDIS_ERR;
slen = snprintf(buf,32,"%ld",value);
* If the object is already raw-encoded just increment the ref count. */
static robj *getDecodedObject(robj *o) {
robj *dec;
-
+
if (o->encoding == REDIS_ENCODING_RAW) {
incrRefCount(o);
return o;
}
}
+static int getDoubleFromObject(redisClient *c, robj *o, double *value) {
+ double parsedValue;
+ char *eptr = NULL;
+
+ if (o && o->type != REDIS_STRING) {
+ addReplySds(c,sdsnew("-ERR value is not a double\r\n"));
+ return REDIS_ERR;
+ }
+
+ if (o == NULL)
+ parsedValue = 0;
+ else if (o->encoding == REDIS_ENCODING_RAW)
+ parsedValue = strtod(o->ptr, &eptr);
+ else if (o->encoding == REDIS_ENCODING_INT)
+ parsedValue = (long)o->ptr;
+ else
+ redisAssert(1 != 1);
+
+ if (eptr != NULL && *eptr != '\0') {
+ addReplySds(c,sdsnew("-ERR value is not a double\r\n"));
+ return REDIS_ERR;
+ }
+
+ *value = parsedValue;
+
+ return REDIS_OK;
+}
+
+static int getLongLongFromObject(redisClient *c, robj *o, long long *value) {
+ long long parsedValue;
+ char *eptr = NULL;
+
+ if (o && o->type != REDIS_STRING) {
+ addReplySds(c,sdsnew("-ERR value is not an integer\r\n"));
+ return REDIS_ERR;
+ }
+
+ if (o == NULL)
+ parsedValue = 0;
+ else if (o->encoding == REDIS_ENCODING_RAW)
+ parsedValue = strtoll(o->ptr, &eptr, 10);
+ else if (o->encoding == REDIS_ENCODING_INT)
+ parsedValue = (long)o->ptr;
+ else
+ redisAssert(1 != 1);
+
+ if (eptr != NULL && *eptr != '\0') {
+ addReplySds(c,sdsnew("-ERR value is not an integer\r\n"));
+ return REDIS_ERR;
+ }
+
+ *value = parsedValue;
+
+ return REDIS_OK;
+}
+
+static int getLongFromObject(redisClient *c, robj *o, long *value) {
+ long long actualValue;
+
+ if (getLongLongFromObject(c, o, &actualValue) != REDIS_OK) return REDIS_ERR;
+
+ if (actualValue < LONG_MIN || actualValue > LONG_MAX) {
+ addReplySds(c,sdsnew("-ERR value is out of range\r\n"));
+ return REDIS_ERR;
+ }
+
+ *value = actualValue;
+
+ return REDIS_OK;
+}
+
/*============================ RDB saving/loading =========================== */
static int rdbSaveType(FILE *fp, unsigned char type) {
/* Return the number of pages required to save this object in the swap file */
static off_t rdbSavedObjectPages(robj *o, FILE *fp) {
off_t bytes = rdbSavedObjectLen(o,fp);
-
+
return (bytes+(server.vm_page_size-1))/server.vm_page_size;
}
fflush(fp);
fsync(fileno(fp));
fclose(fp);
-
+
/* Use RENAME to make sure the DB file is changed atomically only
* if the generate DB file is ok. */
if (rename(tmpfile,filename) == -1) {
static int getGenericCommand(redisClient *c) {
robj *o;
-
+
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk)) == NULL)
return REDIS_OK;
static void mgetCommand(redisClient *c) {
int j;
-
+
addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",c->argc-1));
for (j = 1; j < c->argc; j++) {
robj *o = lookupKeyRead(c->db,c->argv[j]);
long long value;
int retval;
robj *o;
-
+
o = lookupKeyWrite(c->db,c->argv[1]);
- if (o == NULL) {
- value = 0;
- } else {
- if (o->type != REDIS_STRING) {
- value = 0;
- } else {
- char *eptr;
- if (o->encoding == REDIS_ENCODING_RAW)
- value = strtoll(o->ptr, &eptr, 10);
- else if (o->encoding == REDIS_ENCODING_INT)
- value = (long)o->ptr;
- else
- redisAssert(1 != 1);
- }
- }
+ if (getLongLongFromObject(c, o, &value) != REDIS_OK) return;
value += incr;
o = createObject(REDIS_STRING,sdscatprintf(sdsempty(),"%lld",value));
}
static void incrbyCommand(redisClient *c) {
- long long incr = strtoll(c->argv[2]->ptr, NULL, 10);
+ long long incr;
+
+ if (getLongLongFromObject(c, c->argv[2], &incr) != REDIS_OK) return;
+
incrDecrCommand(c,incr);
}
static void decrbyCommand(redisClient *c) {
- long long incr = strtoll(c->argv[2]->ptr, NULL, 10);
+ long long incr;
+
+ if (getLongLongFromObject(c, c->argv[2], &incr) != REDIS_OK) return;
+
incrDecrCommand(c,-incr);
}
totlen = stringObjectLen(c->argv[2]);
} else {
dictEntry *de;
-
+
de = dictFind(c->db->dict,c->argv[1]);
assert(de != NULL);
static void selectCommand(redisClient *c) {
int id = atoi(c->argv[1]->ptr);
-
+
if (selectDb(c,id) == REDIS_ERR) {
addReplySds(c,sdsnew("-ERR invalid DB index\r\n"));
} else {
static void randomkeyCommand(redisClient *c) {
dictEntry *de;
-
+
while(1) {
de = dictGetRandomKey(c->db->dict);
if (!de || expireIfNeeded(c->db,dictGetEntryKey(de)) == 0) break;
* in the next cron() Redis will be notified that the background
* saving aborted, handling special stuff like slaves pending for
* synchronization... */
- redisLog(REDIS_WARNING,"Error trying to save the DB, can't exit");
+ redisLog(REDIS_WARNING,"Error trying to save the DB, can't exit");
addReplySds(c,
sdsnew("-ERR can't quit, problems saving the DB\r\n"));
}
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
checkType(c,o,REDIS_LIST)) return;
-
+
l = o->ptr;
addReplyUlong(c,listLength(l));
}
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
checkType(c,o,REDIS_SET)) return;
-
+
s = o->ptr;
addReplyUlong(c,dictSize(s));
}
static zskiplist *zslCreate(void) {
int j;
zskiplist *zsl;
-
+
zsl = zmalloc(sizeof(*zsl));
zsl->level = 1;
zsl->length = 0;
} else {
dictEntry *de;
double *oldscore;
-
+
/* case 2: Score update operation */
de = dictFind(zs->dict,ele);
redisAssert(de != NULL);
static void zaddCommand(redisClient *c) {
double scoreval;
- scoreval = strtod(c->argv[2]->ptr,NULL);
+ if (getDoubleFromObject(c, c->argv[2], &scoreval) != REDIS_OK) return;
+
zaddGenericCommand(c,c->argv[1],c->argv[3],scoreval,0);
}
static void zincrbyCommand(redisClient *c) {
double scoreval;
- scoreval = strtod(c->argv[2]->ptr,NULL);
+ if (getDoubleFromObject(c, c->argv[2], &scoreval) != REDIS_OK) return;
+
zaddGenericCommand(c,c->argv[1],c->argv[3],scoreval,1);
}
}
static void zremrangebyscoreCommand(redisClient *c) {
- double min = strtod(c->argv[2]->ptr,NULL);
- double max = strtod(c->argv[3]->ptr,NULL);
+ double min;
+ double max;
long deleted;
robj *zsetobj;
zset *zs;
+ if ((getDoubleFromObject(c, c->argv[2], &min) != REDIS_OK) ||
+ (getDoubleFromObject(c, c->argv[3], &max) != REDIS_OK)) return;
+
if ((zsetobj = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
checkType(c,zsetobj,REDIS_ZSET)) return;
}
static void zremrangebyrankCommand(redisClient *c) {
- int start = atoi(c->argv[2]->ptr);
- int end = atoi(c->argv[3]->ptr);
+ long start;
+ long end;
int llen;
long deleted;
robj *zsetobj;
zset *zs;
+ if ((getLongFromObject(c, c->argv[2], &start) != REDIS_OK) ||
+ (getLongFromObject(c, c->argv[3], &end) != REDIS_OK)) return;
+
if ((zsetobj = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
checkType(c,zsetobj,REDIS_ZSET)) return;
zs = zsetobj->ptr;
if (remaining >= (zsetnum + 1) && !strcasecmp(c->argv[j]->ptr,"weights")) {
j++; remaining--;
for (i = 0; i < zsetnum; i++, j++, remaining--) {
- src[i].weight = strtod(c->argv[j]->ptr, NULL);
+ if (getDoubleFromObject(c, c->argv[j], &src[i].weight) != REDIS_OK)
+ return;
}
} else if (remaining >= 2 && !strcasecmp(c->argv[j]->ptr,"aggregate")) {
j++; remaining--;
static void zrangeGenericCommand(redisClient *c, int reverse) {
robj *o;
- int start = atoi(c->argv[2]->ptr);
- int end = atoi(c->argv[3]->ptr);
+ long start;
+ long end;
int withscores = 0;
int llen;
int rangelen, j;
zskiplistNode *ln;
robj *ele;
+ if ((getLongFromObject(c, c->argv[2], &start) != REDIS_OK) ||
+ (getLongFromObject(c, c->argv[3], &end) != REDIS_OK)) return;
+
if (c->argc == 5 && !strcasecmp(c->argv[4]->ptr,"withscores")) {
withscores = 1;
} else if (c->argc >= 5) {
}
}
- incr = strtoll(c->argv[3]->ptr, NULL, 10);
+ if (getLongLongFromObject(c, c->argv[3], &incr) != REDIS_OK) return;
+
if (o->encoding == REDIS_ENCODING_ZIPMAP) {
unsigned char *zm = o->ptr;
unsigned char *zval;
}
}
+static void hmgetCommand(redisClient *c) {
+ int i;
+
+ robj *o = lookupKeyRead(c->db, c->argv[1]);
+ if (o == NULL) {
+ addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",c->argc-2));
+ for (i = 2; i < c->argc; i++) {
+ addReply(c,shared.nullbulk);
+ }
+ return;
+ } else {
+ if (o->type != REDIS_HASH) {
+ addReply(c,shared.wrongtypeerr);
+ return;
+ }
+ }
+
+ addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",c->argc-2));
+ if (o->encoding == REDIS_ENCODING_ZIPMAP) {
+ unsigned char *zm = o->ptr;
+ unsigned char *v;
+ unsigned int vlen;
+ robj *field;
+
+ for (i = 2; i < c->argc; i++) {
+ field = getDecodedObject(c->argv[i]);
+ if (zipmapGet(zm,field->ptr,sdslen(field->ptr),&v,&vlen)) {
+ addReplySds(c,sdscatprintf(sdsempty(),"$%u\r\n", vlen));
+ addReplySds(c,sdsnewlen(v,vlen));
+ addReply(c,shared.crlf);
+ } else {
+ addReply(c,shared.nullbulk);
+ }
+ decrRefCount(field);
+ }
+ } else {
+ dictEntry *de;
+
+ for (i = 2; i < c->argc; i++) {
+ de = dictFind(o->ptr,c->argv[i]);
+ if (de != NULL) {
+ addReplyBulk(c,(robj*)dictGetEntryVal(de));
+ } else {
+ addReply(c,shared.nullbulk);
+ }
+ }
+ }
+}
+
static void hdelCommand(redisClient *c) {
robj *o;
int deleted = 0;
return dictDelete(db->dict,key) == DICT_OK;
}
-static void expireGenericCommand(redisClient *c, robj *key, time_t seconds) {
+static void expireGenericCommand(redisClient *c, robj *key, robj *param, long offset) {
dictEntry *de;
+ time_t seconds;
+
+ if (getLongFromObject(c, param, &seconds) != REDIS_OK) return;
+
+ seconds -= offset;
de = dictFind(c->db->dict,key);
if (de == NULL) {
addReply(c,shared.czero);
return;
}
- if (seconds < 0) {
+ if (seconds <= 0) {
if (deleteKey(c->db,key)) server.dirty++;
addReply(c, shared.cone);
return;
}
static void expireCommand(redisClient *c) {
- expireGenericCommand(c,c->argv[1],strtol(c->argv[2]->ptr,NULL,10));
+ expireGenericCommand(c,c->argv[1],c->argv[2],0);
}
static void expireatCommand(redisClient *c) {
- expireGenericCommand(c,c->argv[1],strtol(c->argv[2]->ptr,NULL,10)-time(NULL));
+ expireGenericCommand(c,c->argv[1],c->argv[2],time(NULL));
}
static void ttlCommand(redisClient *c) {
* non-blocking POP operation */
robj *argv[2], **orig_argv;
int orig_argc;
-
+
/* We need to alter the command arguments before to call
* popGenericCommand() as the command takes a single key. */
orig_argv = c->argv;
slave->replstate = REDIS_REPL_WAIT_BGSAVE_END;
} else if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) {
struct redis_stat buf;
-
+
if (bgsaveerr != REDIS_OK) {
freeClient(slave);
redisLog(REDIS_WARNING,"SYNC failed. BGSAVE child returned an error");
fflush(fp);
fsync(fileno(fp));
fclose(fp);
-
+
/* Use RENAME to make sure the DB file is changed atomically only
* if the generate DB file is ok. */
if (rename(tmpfile,filename) == -1) {
static void expandVmSwapFilename(void) {
char *p = strstr(server.vm_swap_file,"%p");
sds new;
-
+
if (!p) return;
new = sdsempty();
*p = '\0';
}
/* Find N contiguous free pages storing the first page of the cluster in *first.
- * Returns REDIS_OK if it was able to find N contiguous pages, otherwise
+ * Returns REDIS_OK if it was able to find N contiguous pages, otherwise
* REDIS_ERR is returned.
*
* This function uses a simple algorithm: we try to allocate
* we try to find less populated places doing a forward jump of
* REDIS_VM_MAX_RANDOM_JUMP, then we start scanning again a few pages
* without hurry, and then we jump again and so forth...
- *
+ *
* This function can be improved using a free list to avoid to guess
* too much, since we could collect data about freed pages.
*
listDelNode(server.io_processing,ln);
listAddNodeTail(server.io_processed,j);
unlockThreadedIO();
-
+
/* Signal the main thread there is new stuff to process */
assert(write(server.io_ready_pipe_write,"x",1) == 1);
}
static int vmSwapObjectThreaded(robj *key, robj *val, redisDb *db) {
iojob *j;
-
+
assert(key->storage == REDIS_VM_MEMORY);
assert(key->refcount == 1);
vmCancelThreadedIOJob(o);
return 0;
}
-
+
/* OK: the key is either swapped, or being loaded just now. */
/* Add the key to the list of keys this client is waiting for.
return (void*) uc->uc_mcontext->__ss.__rip;
#else
return (void*) uc->uc_mcontext->__ss.__eip;
- #endif
+ #endif
#elif defined(__i386__) || defined(__X86_64__) || defined(__x86_64__)
return (void*) uc->uc_mcontext.gregs[REG_EIP]; /* Linux 32/64 bit */
#elif defined(__ia64__) /* Linux IA64 */
redisLog(REDIS_WARNING, "%s",infostring);
/* It's not safe to sdsfree() the returned string under memory
* corruption conditions. Let it leak as we are going to abort */
-
+
trace_size = backtrace(trace, 100);
/* overwrite sigaction with caller's address */
if (getMcontextEip(uc) != NULL) {
projects / redis.git / blobdiff