X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/f6b32c14f4c8680d2a6b7a4d71758e76ca2c3554..8562798308391d489016b3995d438b6187b5980a:/src/redis.c diff --git a/src/redis.c b/src/redis.c index 7b023dc5..3294eea4 100644 --- a/src/redis.c +++ b/src/redis.c @@ -242,7 +242,8 @@ struct redisCommand redisCommandTable[] = { {"eval",evalCommand,-3,"wms",0,zunionInterGetKeys,0,0,0,0,0}, {"evalsha",evalShaCommand,-3,"wms",0,zunionInterGetKeys,0,0,0,0,0}, {"slowlog",slowlogCommand,-2,"r",0,NULL,0,0,0,0,0}, - {"script",scriptCommand,-2,"ras",0,NULL,0,0,0,0,0} + {"script",scriptCommand,-2,"ras",0,NULL,0,0,0,0,0}, + {"time",timeCommand,1,"rR",0,NULL,0,0,0,0,0} }; /*============================ Utility functions ============================ */ @@ -615,6 +616,31 @@ void updateLRUClock(void) { REDIS_LRU_CLOCK_MAX; } + +/* Add a sample to the operations per second array of samples. */ +void trackOperationsPerSecond(void) { + long long t = mstime() - server.ops_sec_last_sample_time; + long long ops = server.stat_numcommands - server.ops_sec_last_sample_ops; + long long ops_sec; + + ops_sec = t > 0 ? (ops*1000/t) : 0; + + server.ops_sec_samples[server.ops_sec_idx] = ops_sec; + server.ops_sec_idx = (server.ops_sec_idx+1) % REDIS_OPS_SEC_SAMPLES; + server.ops_sec_last_sample_time = mstime(); + server.ops_sec_last_sample_ops = server.stat_numcommands; +} + +/* Return the mean of all the samples. */ +long long getOperationsPerSecond(void) { + int j; + long long sum = 0; + + for (j = 0; j < REDIS_OPS_SEC_SAMPLES; j++) + sum += server.ops_sec_samples[j]; + return sum / REDIS_OPS_SEC_SAMPLES; +} + int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { int j, loops = server.cronloops; REDIS_NOTUSED(eventLoop); @@ -627,6 +653,8 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { * To access a global var is faster than calling time(NULL) */ server.unixtime = time(NULL); + trackOperationsPerSecond(); + /* We have just 22 bits per object for LRU information. * So we use an (eventually wrapping) LRU clock with 10 seconds resolution. * 2^22 bits with 10 seconds resoluton is more or less 1.5 years. @@ -832,6 +860,8 @@ void createSharedObjects(void) { "-LOADING Redis is loading the dataset in memory\r\n")); shared.slowscripterr = createObject(REDIS_STRING,sdsnew( "-BUSY Redis is busy running a script. You can only call SCRIPT KILL or SHUTDOWN NOSAVE.\r\n")); + shared.bgsaveerr = createObject(REDIS_STRING,sdsnew( + "-MISCONF Redis is configured to save RDB snapshots, but is currently not able to persist on disk. Write commands are disabled. Please check Redis logs for details about the error.\r\n")); shared.space = createObject(REDIS_STRING,sdsnew(" ")); shared.colon = createObject(REDIS_STRING,sdsnew(":")); shared.plus = createObject(REDIS_STRING,sdsnew("+")); @@ -852,6 +882,8 @@ void createSharedObjects(void) { shared.psubscribebulk = createStringObject("$10\r\npsubscribe\r\n",17); shared.punsubscribebulk = createStringObject("$12\r\npunsubscribe\r\n",19); shared.del = createStringObject("DEL",3); + shared.rpop = createStringObject("RPOP",4); + shared.lpop = createStringObject("LPOP",4); for (j = 0; j < REDIS_SHARED_INTEGERS; j++) { shared.integers[j] = createObject(REDIS_STRING,(void*)(long)j); shared.integers[j]->encoding = REDIS_ENCODING_INT; @@ -865,6 +897,8 @@ void createSharedObjects(void) { } void initServerConfig() { + getRandomHexChars(server.runid,REDIS_RUN_ID_SIZE); + server.runid[REDIS_RUN_ID_SIZE] = '\0'; server.arch_bits = (sizeof(long) == 8) ? 64 : 32; server.port = REDIS_SERVERPORT; server.bindaddr = NULL; @@ -905,8 +939,8 @@ void initServerConfig() { server.maxmemory = 0; server.maxmemory_policy = REDIS_MAXMEMORY_VOLATILE_LRU; server.maxmemory_samples = 3; - server.hash_max_zipmap_entries = REDIS_HASH_MAX_ZIPMAP_ENTRIES; - server.hash_max_zipmap_value = REDIS_HASH_MAX_ZIPMAP_VALUE; + server.hash_max_ziplist_entries = REDIS_HASH_MAX_ZIPLIST_ENTRIES; + server.hash_max_ziplist_value = REDIS_HASH_MAX_ZIPLIST_VALUE; server.list_max_ziplist_entries = REDIS_LIST_MAX_ZIPLIST_ENTRIES; server.list_max_ziplist_value = REDIS_LIST_MAX_ZIPLIST_VALUE; server.set_max_intset_entries = REDIS_SET_MAX_INTSET_ENTRIES; @@ -962,6 +996,7 @@ void initServerConfig() { populateCommandTable(); server.delCommand = lookupCommandByCString("del"); server.multiCommand = lookupCommandByCString("multi"); + server.lpushCommand = lookupCommandByCString("lpush"); /* Slow log */ server.slowlog_log_slower_than = REDIS_SLOWLOG_LOG_SLOWER_THAN; @@ -1083,7 +1118,13 @@ void initServer() { server.stat_peak_memory = 0; server.stat_fork_time = 0; server.stat_rejected_conn = 0; + memset(server.ops_sec_samples,0,sizeof(server.ops_sec_samples)); + server.ops_sec_idx = 0; + server.ops_sec_last_sample_time = mstime(); + server.ops_sec_last_sample_ops = 0; server.unixtime = time(NULL); + server.lastbgsave_status = REDIS_OK; + server.stop_writes_on_bgsave_err = 1; aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL); if (server.ipfd > 0 && aeCreateFileEvent(server.el,server.ipfd,AE_READABLE, acceptTcpHandler,NULL) == AE_ERR) oom("creating file event"); @@ -1160,6 +1201,43 @@ void resetCommandTableStats(void) { } } +/* ========================== Redis OP Array API ============================ */ + +void redisOpArrayInit(redisOpArray *oa) { + oa->ops = NULL; + oa->numops = 0; +} + +int redisOpArrayAppend(redisOpArray *oa, struct redisCommand *cmd, int dbid, + robj **argv, int argc, int target) +{ + redisOp *op; + + oa->ops = zrealloc(oa->ops,sizeof(redisOp)*(oa->numops+1)); + op = oa->ops+oa->numops; + op->cmd = cmd; + op->dbid = dbid; + op->argv = argv; + op->argc = argc; + op->target = target; + oa->numops++; + return oa->numops; +} + +void redisOpArrayFree(redisOpArray *oa) { + while(oa->numops) { + int j; + redisOp *op; + + oa->numops--; + op = oa->ops+oa->numops; + for (j = 0; j < op->argc; j++) + decrRefCount(op->argv[j]); + zfree(op->argv); + } + zfree(oa->ops); +} + /* ====================== Commands lookup and execution ===================== */ struct redisCommand *lookupCommand(sds name) { @@ -1175,10 +1253,42 @@ struct redisCommand *lookupCommandByCString(char *s) { return cmd; } +/* Propagate the specified command (in the context of the specified database id) + * to AOF, Slaves and Monitors. + * + * flags are an xor between: + * + REDIS_PROPAGATE_NONE (no propagation of command at all) + * + REDIS_PROPAGATE_AOF (propagate into the AOF file if is enabled) + * + REDIS_PROPAGATE_REPL (propagate into the replication link) + */ +void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc, + int flags) +{ + if (server.aof_state != REDIS_AOF_OFF && flags & REDIS_PROPAGATE_AOF) + feedAppendOnlyFile(cmd,dbid,argv,argc); + if (flags & REDIS_PROPAGATE_REPL && listLength(server.slaves)) + replicationFeedSlaves(server.slaves,dbid,argv,argc); +} + +/* Used inside commands to schedule the propagation of additional commands + * after the current command is propagated to AOF / Replication. */ +void alsoPropagate(struct redisCommand *cmd, int dbid, robj **argv, int argc, + int target) +{ + redisOpArrayAppend(&server.also_propagate,cmd,dbid,argv,argc,target); +} + /* Call() is the core of Redis execution of a command */ void call(redisClient *c, int flags) { long long dirty, start = ustime(), duration; + /* Sent the command to clients in MONITOR mode, only if the commands are + * not geneated from reading an AOF. */ + if (listLength(server.monitors) && !server.loading) + replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc); + + /* Call the command. */ + redisOpArrayInit(&server.also_propagate); dirty = server.dirty; c->cmd->proc(c); dirty = server.dirty-dirty; @@ -1189,20 +1299,37 @@ void call(redisClient *c, int flags) { if (server.loading && c->flags & REDIS_LUA_CLIENT) flags &= ~(REDIS_CALL_SLOWLOG | REDIS_CALL_STATS); + /* Log the command into the Slow log if needed, and populate the + * per-command statistics that we show in INFO commandstats. */ if (flags & REDIS_CALL_SLOWLOG) slowlogPushEntryIfNeeded(c->argv,c->argc,duration); if (flags & REDIS_CALL_STATS) { c->cmd->microseconds += duration; c->cmd->calls++; } + + /* Propagate the command into the AOF and replication link */ if (flags & REDIS_CALL_PROPAGATE) { - if (server.aof_state != REDIS_AOF_OFF && dirty > 0) - feedAppendOnlyFile(c->cmd,c->db->id,c->argv,c->argc); - if ((dirty > 0 || c->cmd->flags & REDIS_CMD_FORCE_REPLICATION) && - listLength(server.slaves)) - replicationFeedSlaves(server.slaves,c->db->id,c->argv,c->argc); - if (listLength(server.monitors)) - replicationFeedMonitors(server.monitors,c->db->id,c->argv,c->argc); + int flags = REDIS_PROPAGATE_NONE; + + if (c->cmd->flags & REDIS_CMD_FORCE_REPLICATION) + flags |= REDIS_PROPAGATE_REPL; + if (dirty) + flags |= (REDIS_PROPAGATE_REPL | REDIS_PROPAGATE_AOF); + if (flags != REDIS_PROPAGATE_NONE) + propagate(c->cmd,c->db->id,c->argv,c->argc,flags); + } + /* Commands such as LPUSH or BRPOPLPUSH may propagate an additional + * PUSH command. */ + if (server.also_propagate.numops) { + int j; + redisOp *rop; + + for (j = 0; j < server.also_propagate.numops; j++) { + rop = &server.also_propagate.ops[j]; + propagate(rop->cmd, rop->dbid, rop->argv, rop->argc, rop->target); + } + redisOpArrayFree(&server.also_propagate); } server.stat_numcommands++; } @@ -1284,6 +1411,16 @@ int processCommand(redisClient *c) { } } + /* Don't accept write commands if there are problems persisting on disk. */ + if (server.stop_writes_on_bgsave_err && + server.saveparamslen > 0 + && server.lastbgsave_status == REDIS_ERR && + c->cmd->flags & REDIS_CMD_WRITE) + { + addReply(c, shared.bgsaveerr); + return REDIS_OK; + } + /* Only allow SUBSCRIBE and UNSUBSCRIBE in the context of Pub/Sub */ if ((dictSize(c->pubsub_channels) > 0 || listLength(c->pubsub_patterns) > 0) && @@ -1416,6 +1553,17 @@ void echoCommand(redisClient *c) { addReplyBulk(c,c->argv[1]); } +void timeCommand(redisClient *c) { + struct timeval tv; + + /* gettimeofday() can only fail if &tv is a bad addresss so we + * don't check for errors. */ + gettimeofday(&tv,NULL); + addReplyMultiBulkLen(c,2); + addReplyBulkLongLong(c,tv.tv_sec); + addReplyBulkLongLong(c,tv.tv_usec); +} + /* Convert an amount of bytes into a human readable string in the form * of 100B, 2G, 100M, 4K, and so forth. */ void bytesToHuman(char *s, unsigned long long n) { @@ -1470,6 +1618,7 @@ sds genRedisInfoString(char *section) { "multiplexing_api:%s\r\n" "gcc_version:%d.%d.%d\r\n" "process_id:%ld\r\n" + "run_id:%s\r\n" "tcp_port:%d\r\n" "uptime_in_seconds:%ld\r\n" "uptime_in_days:%ld\r\n" @@ -1485,6 +1634,7 @@ sds genRedisInfoString(char *section) { 0,0,0, #endif (long) getpid(), + server.runid, server.port, uptime, uptime/(3600*24), @@ -1544,12 +1694,14 @@ sds genRedisInfoString(char *section) { "changes_since_last_save:%lld\r\n" "bgsave_in_progress:%d\r\n" "last_save_time:%ld\r\n" + "last_bgsave_status:%s\r\n" "bgrewriteaof_in_progress:%d\r\n", server.loading, server.aof_state != REDIS_AOF_OFF, server.dirty, server.rdb_child_pid != -1, server.lastsave, + server.lastbgsave_status == REDIS_OK ? "ok" : "err", server.aof_child_pid != -1); if (server.aof_state != REDIS_AOF_OFF) { @@ -1605,6 +1757,7 @@ sds genRedisInfoString(char *section) { "# Stats\r\n" "total_connections_received:%lld\r\n" "total_commands_processed:%lld\r\n" + "instantaneous_ops_per_sec:%lld\r\n" "rejected_connections:%lld\r\n" "expired_keys:%lld\r\n" "evicted_keys:%lld\r\n" @@ -1615,6 +1768,7 @@ sds genRedisInfoString(char *section) { "latest_fork_usec:%lld\r\n", server.stat_numconnections, server.stat_numcommands, + getOperationsPerSecond(), server.stat_rejected_conn, server.stat_expiredkeys, server.stat_evictedkeys, @@ -1802,8 +1956,8 @@ int freeMemoryIfNeeded(void) { size_t mem_used, mem_tofree, mem_freed; int slaves = listLength(server.slaves); - /* Remove the size of slaves output buffers from the count of used - * memory. */ + /* Remove the size of slaves output buffers and AOF buffer from the + * count of used memory. */ mem_used = zmalloc_used_memory(); if (slaves) { listIter li; @@ -1819,6 +1973,10 @@ int freeMemoryIfNeeded(void) { mem_used -= obuf_bytes; } } + if (server.aof_state != REDIS_AOF_OFF) { + mem_used -= sdslen(server.aof_buf); + mem_used -= sdslen(server.aof_rewrite_buf); + } /* Check if we are over the memory limit. */ if (mem_used <= server.maxmemory) return REDIS_OK; @@ -1828,7 +1986,6 @@ int freeMemoryIfNeeded(void) { /* Compute how much memory we need to free. */ mem_tofree = mem_used - server.maxmemory; - printf("USED: %zu, TOFREE: %zu\n", mem_used, mem_tofree); mem_freed = 0; while (mem_freed < mem_tofree) { int j, k, keys_freed = 0; @@ -1919,7 +2076,6 @@ int freeMemoryIfNeeded(void) { delta = (long long) zmalloc_used_memory(); dbDelete(db,keyobj); delta -= (long long) zmalloc_used_memory(); - // printf("%lld\n",delta); mem_freed += delta; server.stat_evictedkeys++; decrRefCount(keyobj); @@ -1929,27 +2085,7 @@ int freeMemoryIfNeeded(void) { * start spending so much time here that is impossible to * deliver data to the slaves fast enough, so we force the * transmission here inside the loop. */ - if (slaves) { - listIter li; - listNode *ln; - - listRewind(server.slaves,&li); - while((ln = listNext(&li))) { - redisClient *slave = listNodeValue(ln); - int events; - - events = aeGetFileEvents(server.el,slave->fd); - printf("EVENTS: %d\n", events); - if (events & AE_WRITABLE && - slave->replstate == REDIS_REPL_ONLINE && - listLength(slave->reply)) - { - printf("SLAVE %d -> %d\n", - slave->fd, (int) listLength(slave->reply)); - sendReplyToClient(server.el,slave->fd,slave,0); - } - } - } + if (slaves) flushSlavesOutputBuffers(); } } if (!keys_freed) return REDIS_ERR; /* nothing to free... */