X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/fec5a664f8ce7a1e0f687d07079aa09230fa5846..09e2d9eeba3ff65fd60f905a5bcb0f684f7a883e:/src/redis.c?ds=sidebyside diff --git a/src/redis.c b/src/redis.c index 1a1f84f9..0990d98a 100644 --- a/src/redis.c +++ b/src/redis.c @@ -28,6 +28,7 @@ */ #include "redis.h" +#include "slowlog.h" #ifdef HAVE_BACKTRACE #include @@ -50,7 +51,6 @@ #include #include #include -#include #include /* Our shared "common" objects */ @@ -193,7 +193,8 @@ struct redisCommand redisCommandTable[] = { {"migrate",migrateCommand,6,0,NULL,0,0,0,0,0}, {"dump",dumpCommand,2,0,NULL,0,0,0,0,0}, {"object",objectCommand,-2,0,NULL,0,0,0,0,0}, - {"client",clientCommand,-2,0,NULL,0,0,0,0,0} + {"client",clientCommand,-2,0,NULL,0,0,0,0,0}, + {"slowlog",slowlogCommand,-2,0,NULL,0,0,0,0,0} }; /*============================ Utility functions ============================ */ @@ -659,22 +660,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { } updateDictResizePolicy(); } - } else if (server.bgsavethread != (pthread_t) -1) { - if (server.bgsavethread != (pthread_t) -1) { - int state; - - pthread_mutex_lock(&server.bgsavethread_mutex); - state = server.bgsavethread_state; - pthread_mutex_unlock(&server.bgsavethread_mutex); - - if (state == REDIS_BGSAVE_THREAD_DONE_OK || - state == REDIS_BGSAVE_THREAD_DONE_ERR) - { - backgroundSaveDoneHandler( - (state == REDIS_BGSAVE_THREAD_DONE_OK) ? 0 : 1, 0); - } - } - } else if (!server.ds_enabled) { + } else { time_t now = time(NULL); /* If there is not a background saving/rewrite in progress check if @@ -712,10 +698,6 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { * in order to guarantee a strict consistency. */ if (server.masterhost == NULL) activeExpireCycle(); - /* Remove a few cached objects from memory if we are over the - * configured memory limit */ - if (server.ds_enabled) cacheCron(); - /* Replication cron function -- used to reconnect to master and * to detect transfer failures. */ if (!(loops % 10)) replicationCron(); @@ -735,31 +717,6 @@ void beforeSleep(struct aeEventLoop *eventLoop) { listNode *ln; redisClient *c; - /* Awake clients that got all the on disk keys they requested */ - if (server.ds_enabled && listLength(server.io_ready_clients)) { - listIter li; - - listRewind(server.io_ready_clients,&li); - while((ln = listNext(&li))) { - c = ln->value; - struct redisCommand *cmd; - - /* Resume the client. */ - listDelNode(server.io_ready_clients,ln); - c->flags &= (~REDIS_IO_WAIT); - server.cache_blocked_clients--; - aeCreateFileEvent(server.el, c->fd, AE_READABLE, - readQueryFromClient, c); - cmd = lookupCommand(c->argv[0]->ptr); - redisAssert(cmd != NULL); - call(c,cmd); - resetClient(c); - /* There may be more data to process in the input buffer. */ - if (c->querybuf && sdslen(c->querybuf) > 0) - processInputBuffer(c); - } - } - /* Try to process pending commands for clients that were just unblocked. */ while (listLength(server.unblocked_clients)) { ln = listFirst(server.unblocked_clients); @@ -870,10 +827,6 @@ void initServerConfig() { server.maxmemory = 0; server.maxmemory_policy = REDIS_MAXMEMORY_VOLATILE_LRU; server.maxmemory_samples = 3; - server.ds_enabled = 0; - server.ds_path = sdsnew("/tmp/redis.ds"); - server.cache_max_memory = 64LL*1024*1024; /* 64 MB of RAM */ - server.cache_blocked_clients = 0; server.hash_max_zipmap_entries = REDIS_HASH_MAX_ZIPMAP_ENTRIES; server.hash_max_zipmap_value = REDIS_HASH_MAX_ZIPMAP_VALUE; server.list_max_ziplist_entries = REDIS_LIST_MAX_ZIPLIST_ENTRIES; @@ -882,7 +835,6 @@ void initServerConfig() { server.zset_max_ziplist_entries = REDIS_ZSET_MAX_ZIPLIST_ENTRIES; server.zset_max_ziplist_value = REDIS_ZSET_MAX_ZIPLIST_VALUE; server.shutdown_asap = 0; - server.cache_flush_delay = 0; server.cluster_enabled = 0; server.cluster.configfile = zstrdup("nodes.conf"); @@ -916,6 +868,10 @@ void initServerConfig() { populateCommandTable(); server.delCommand = lookupCommandByCString("del"); server.multiCommand = lookupCommandByCString("multi"); + + /* Slow log */ + server.slowlog_log_slower_than = REDIS_SLOWLOG_LOG_SLOWER_THAN; + server.slowlog_max_len = REDIS_SLOWLOG_MAX_LEN; } void initServer() { @@ -930,12 +886,10 @@ void initServer() { server.syslog_facility); } - server.mainthread = pthread_self(); server.clients = listCreate(); server.slaves = listCreate(); server.monitors = listCreate(); server.unblocked_clients = listCreate(); - server.cache_io_queue = listCreate(); createSharedObjects(); server.el = aeCreateEventLoop(); @@ -965,11 +919,6 @@ void initServer() { server.db[j].expires = dictCreate(&keyptrDictType,NULL); server.db[j].blocking_keys = dictCreate(&keylistDictType,NULL); server.db[j].watched_keys = dictCreate(&keylistDictType,NULL); - if (server.ds_enabled) { - server.db[j].io_keys = dictCreate(&keylistDictType,NULL); - server.db[j].io_negcache = dictCreate(&setDictType,NULL); - server.db[j].io_queued = dictCreate(&setDictType,NULL); - } server.db[j].id = j; } server.pubsub_channels = dictCreate(&keylistDictType,NULL); @@ -979,8 +928,6 @@ void initServer() { server.cronloops = 0; server.bgsavechildpid = -1; server.bgrewritechildpid = -1; - server.bgsavethread_state = REDIS_BGSAVE_THREAD_UNACTIVE; - server.bgsavethread = (pthread_t) -1; server.bgrewritebuf = sdsempty(); server.aofbuf = sdsempty(); server.lastsave = time(NULL); @@ -1010,8 +957,8 @@ void initServer() { } } - if (server.ds_enabled) dsInit(); if (server.cluster_enabled) clusterInit(); + slowlogInit(); srand(time(NULL)^getpid()); } @@ -1058,18 +1005,20 @@ struct redisCommand *lookupCommandByCString(char *s) { } /* Call() is the core of Redis execution of a command */ -void call(redisClient *c, struct redisCommand *cmd) { - long long dirty, start = ustime(); +void call(redisClient *c) { + long long dirty, start = ustime(), duration; dirty = server.dirty; - cmd->proc(c); + c->cmd->proc(c); dirty = server.dirty-dirty; - cmd->microseconds += ustime()-start; - cmd->calls++; + duration = ustime()-start; + c->cmd->microseconds += duration; + slowlogPushEntryIfNeeded(c->argv,c->argc,duration); + c->cmd->calls++; if (server.appendonly && dirty) - feedAppendOnlyFile(cmd,c->db->id,c->argv,c->argc); - if ((dirty || cmd->flags & REDIS_CMD_FORCE_REPLICATION) && + feedAppendOnlyFile(c->cmd,c->db->id,c->argv,c->argc); + if ((dirty || c->cmd->flags & REDIS_CMD_FORCE_REPLICATION) && listLength(server.slaves)) replicationFeedSlaves(server.slaves,c->db->id,c->argv,c->argc); if (listLength(server.monitors)) @@ -1086,8 +1035,6 @@ void call(redisClient *c, struct redisCommand *cmd) { * and other operations can be performed by the caller. Otherwise * if 0 is returned the client was destroied (i.e. after QUIT). */ int processCommand(redisClient *c) { - struct redisCommand *cmd; - /* The QUIT command is handled separately. Normal command procs will * go through checking for replication and QUIT will cause trouble * when FORCE_REPLICATION is enabled and would be implemented in @@ -1099,28 +1046,29 @@ int processCommand(redisClient *c) { } /* Now lookup the command and check ASAP about trivial error conditions - * such wrong arity, bad command name and so forth. */ - cmd = lookupCommand(c->argv[0]->ptr); - if (!cmd) { + * such as wrong arity, bad command name and so forth. */ + c->cmd = lookupCommand(c->argv[0]->ptr); + if (!c->cmd) { addReplyErrorFormat(c,"unknown command '%s'", (char*)c->argv[0]->ptr); return REDIS_OK; - } else if ((cmd->arity > 0 && cmd->arity != c->argc) || - (c->argc < -cmd->arity)) { + } else if ((c->cmd->arity > 0 && c->cmd->arity != c->argc) || + (c->argc < -c->cmd->arity)) { addReplyErrorFormat(c,"wrong number of arguments for '%s' command", - cmd->name); + c->cmd->name); return REDIS_OK; } /* Check if the user is authenticated */ - if (server.requirepass && !c->authenticated && cmd->proc != authCommand) { + if (server.requirepass && !c->authenticated && c->cmd->proc != authCommand) + { addReplyError(c,"operation not permitted"); return REDIS_OK; } /* If cluster is enabled, redirect here */ if (server.cluster_enabled && - !(cmd->getkeys_proc == NULL && cmd->firstkey == 0)) { + !(c->cmd->getkeys_proc == NULL && c->cmd->firstkey == 0)) { int hashslot; if (server.cluster.state != REDIS_CLUSTER_OK) { @@ -1128,7 +1076,7 @@ int processCommand(redisClient *c) { return REDIS_OK; } else { int ask; - clusterNode *n = getNodeByQuery(c,cmd,c->argv,c->argc,&hashslot,&ask); + clusterNode *n = getNodeByQuery(c,c->cmd,c->argv,c->argc,&hashslot,&ask); if (n == NULL) { addReplyError(c,"Multi keys request invalid in cluster"); return REDIS_OK; @@ -1147,7 +1095,7 @@ int processCommand(redisClient *c) { * keys in the dataset). If there are not the only thing we can do * is returning an error. */ if (server.maxmemory) freeMemoryIfNeeded(); - if (server.maxmemory && (cmd->flags & REDIS_CMD_DENYOOM) && + if (server.maxmemory && (c->cmd->flags & REDIS_CMD_DENYOOM) && zmalloc_used_memory() > server.maxmemory) { addReplyError(c,"command not allowed when used memory > 'maxmemory'"); @@ -1157,8 +1105,10 @@ int processCommand(redisClient *c) { /* Only allow SUBSCRIBE and UNSUBSCRIBE in the context of Pub/Sub */ if ((dictSize(c->pubsub_channels) > 0 || listLength(c->pubsub_patterns) > 0) && - cmd->proc != subscribeCommand && cmd->proc != unsubscribeCommand && - cmd->proc != psubscribeCommand && cmd->proc != punsubscribeCommand) { + c->cmd->proc != subscribeCommand && + c->cmd->proc != unsubscribeCommand && + c->cmd->proc != psubscribeCommand && + c->cmd->proc != punsubscribeCommand) { addReplyError(c,"only (P)SUBSCRIBE / (P)UNSUBSCRIBE / QUIT allowed in this context"); return REDIS_OK; } @@ -1167,7 +1117,7 @@ int processCommand(redisClient *c) { * we are a slave with a broken link with master. */ if (server.masterhost && server.replstate != REDIS_REPL_CONNECTED && server.repl_serve_stale_data == 0 && - cmd->proc != infoCommand && cmd->proc != slaveofCommand) + c->cmd->proc != infoCommand && c->cmd->proc != slaveofCommand) { addReplyError(c, "link with MASTER is down and slave-serve-stale-data is set to no"); @@ -1175,22 +1125,20 @@ int processCommand(redisClient *c) { } /* Loading DB? Return an error if the command is not INFO */ - if (server.loading && cmd->proc != infoCommand) { + if (server.loading && c->cmd->proc != infoCommand) { addReply(c, shared.loadingerr); return REDIS_OK; } /* Exec the command */ if (c->flags & REDIS_MULTI && - cmd->proc != execCommand && cmd->proc != discardCommand && - cmd->proc != multiCommand && cmd->proc != watchCommand) + c->cmd->proc != execCommand && c->cmd->proc != discardCommand && + c->cmd->proc != multiCommand && c->cmd->proc != watchCommand) { - queueMultiCommand(c,cmd); + queueMultiCommand(c); addReply(c,shared.queued); } else { - if (server.ds_enabled && blockClientOnSwappedKeys(c,cmd)) - return REDIS_ERR; - call(c,cmd); + call(c); } return REDIS_OK; } @@ -1207,9 +1155,7 @@ int prepareForShutdown() { kill(server.bgsavechildpid,SIGKILL); rdbRemoveTempFile(server.bgsavechildpid); } - if (server.ds_enabled) { - /* FIXME: flush all objects on disk */ - } else if (server.appendonly) { + if (server.appendonly) { /* Append only file: fsync() the AOF and exit */ aof_fsync(server.appendfd); } else if (server.saveparamslen > 0) { @@ -1361,22 +1307,6 @@ sds genRedisInfoString(char *section) { ); } - /* Allocation statistics */ - if (allsections || !strcasecmp(section,"allocstats")) { - if (sections++) info = sdscat(info,"\r\n"); - info = sdscat(info, "# Allocstats\r\nallocation_stats:"); - for (j = 0; j <= ZMALLOC_MAX_ALLOC_STAT; j++) { - size_t count = zmalloc_allocations_for_size(j); - if (count) { - if (info[sdslen(info)-1] != ':') info = sdscatlen(info,",",1); - info = sdscatprintf(info,"%s%d=%zu", - (j == ZMALLOC_MAX_ALLOC_STAT) ? ">=" : "", - j,count); - } - } - info = sdscat(info,"\r\n"); - } - /* Persistence */ if (allsections || defsections || !strcasecmp(section,"persistence")) { if (sections++) info = sdscat(info,"\r\n"); @@ -1391,8 +1321,7 @@ sds genRedisInfoString(char *section) { server.loading, server.appendonly, server.dirty, - server.bgsavechildpid != -1 || - server.bgsavethread != (pthread_t) -1, + server.bgsavechildpid != -1, server.lastsave, server.bgrewritechildpid != -1); @@ -1438,35 +1367,6 @@ sds genRedisInfoString(char *section) { } } - /* Diskstore */ - if (allsections || defsections || !strcasecmp(section,"diskstore")) { - if (sections++) info = sdscat(info,"\r\n"); - info = sdscatprintf(info, - "# Diskstore\r\n" - "ds_enabled:%d\r\n", - server.ds_enabled != 0); - if (server.ds_enabled) { - lockThreadedIO(); - info = sdscatprintf(info, - "cache_max_memory:%llu\r\n" - "cache_blocked_clients:%lu\r\n" - "cache_io_queue_len:%lu\r\n" - "cache_io_jobs_new:%lu\r\n" - "cache_io_jobs_processing:%lu\r\n" - "cache_io_jobs_processed:%lu\r\n" - "cache_io_ready_clients:%lu\r\n" - ,(unsigned long long) server.cache_max_memory, - (unsigned long) server.cache_blocked_clients, - (unsigned long) listLength(server.cache_io_queue), - (unsigned long) listLength(server.io_newjobs), - (unsigned long) listLength(server.io_processing), - (unsigned long) listLength(server.io_processed), - (unsigned long) listLength(server.io_ready_clients) - ); - unlockThreadedIO(); - } - } - /* Stats */ if (allsections || defsections || !strcasecmp(section,"stats")) { if (sections++) info = sdscat(info,"\r\n"); @@ -1542,8 +1442,8 @@ sds genRedisInfoString(char *section) { "# CPU\r\n" "used_cpu_sys:%.2f\r\n" "used_cpu_user:%.2f\r\n" - "used_cpu_sys_childrens:%.2f\r\n" - "used_cpu_user_childrens:%.2f\r\n", + "used_cpu_sys_children:%.2f\r\n" + "used_cpu_user_children:%.2f\r\n", (float)self_ru.ru_utime.tv_sec+(float)self_ru.ru_utime.tv_usec/1000000, (float)self_ru.ru_stime.tv_sec+(float)self_ru.ru_stime.tv_usec/1000000, (float)c_ru.ru_utime.tv_sec+(float)c_ru.ru_utime.tv_usec/1000000, @@ -1824,9 +1724,7 @@ int main(int argc, char **argv) { linuxOvercommitMemoryWarning(); #endif start = ustime(); - if (server.ds_enabled) { - redisLog(REDIS_NOTICE,"DB not loaded (running with disk back end)"); - } else if (server.appendonly) { + if (server.appendonly) { if (loadAppendOnlyFile(server.appendfilename) == REDIS_OK) redisLog(REDIS_NOTICE,"DB loaded from append only file: %.3f seconds",(float)(ustime()-start)/1000000); } else {