From: Salvatore Sanfilippo Date: Wed, 22 Feb 2012 16:19:54 +0000 (-0800) Subject: Merge pull request #304 from bradvoth/unstable X-Git-Url: https://git.saurik.com/redis.git/commitdiff_plain/dbb27a0a90ca3800f5be1d8170e404b9e7b9bc44?hp=bad2b8e6ec81235fae085da1c8394c8ece1ad0ea Merge pull request #304 from bradvoth/unstable Simple install script changes --- diff --git a/00-RELEASENOTES b/00-RELEASENOTES new file mode 100644 index 00000000..49221f2c --- /dev/null +++ b/00-RELEASENOTES @@ -0,0 +1,79 @@ +Redis 2.6 release notes + +Migrating from 2.4 to 2.6 +========================= + +Redis 2.4 is mostly a strict subset of 2.6. However there are a few things +that you should be aware of: + +* You can't use .rdb and AOF files generated with 2.6 into a 2.4 instance. +* 2.4 slaves can be attached to 2.6 masters, but not the contrary, and only + for the time needed to perform the version upgrade. + +There are also a few API differences, that are unlikely to cause problems, +but it is better to keep them in mind: + +* SORT now will refuse to sort in numerical mode elements that can't be parsed + as numbers. +* EXPIREs now all have millisecond resolution (but this is very unlikely to + break code that was not conceived exploting the previous resolution error + in some way.) +* INFO output is a bit different now, and contains empty lines and comments + starting with '#'. All the major clients should be already fixed to work + with the new INFO format. + +--------- +CHANGELOG +--------- + +What's new in Redis 2.6.0 +========================= + +UPGRADE URGENCY: We suggest new users to start with 2.6.0, and old users to + upgrade after some testing of the application with the new + Redis version. + +* Server side Lua scripting, see http://redis.io/commands/eval +* Virtual Memory removed (was deprecated in 2.4) +* Hardcoded limits about max number of clients removed. +* AOF low level semantics is generally more sane, and especially when used + in slaves. +* Milliseconds resolution expires, also added new commands with milliseconds + precision (PEXPIRE, PTTL, ...). +* Clinets max output buffer soft and hard limits. You can specifiy different + limits for different classes of clients (normal,pubsub,slave). +* AOF is now able to rewrite aggregate data types using variadic commands, + often producing an AOF that is faster to save, load, and is smaller in size. +* Every redis.conf directive is now accepted as a command line option for the + redis-server binary, with the same name and number of arguments. +* Hash table seed randomization for protection against collisions attacks. +* Performances improved when writing large objects to Redis. +* Significant parts of the core refactored or rewritten. New internal APIs + and core changes allowed to develop Redis Cluster on top of the new code, + however for 2.6 all the cluster code was removed, and will be released with + Redis 3.0 when it is more complete and stable. +* Redis ASCII art logo added at startup. +* Crash report on memory violation or failed asserts improved significantly + to make debugging of hard to catch bugs simpler. +* redis-benchmark improvements: ability to run selected tests, + CSV output, faster, better help. +* redis-cli improvements: --eval for comfortable development of Lua scripts. +* SHUTDOWN now supports two optional arguments: "SAVE" and "NOSAVE". +* INFO output split into sections, the command is now able to just show + pecific sections. +* New statistics about how many time a command was called, and how much + execution time it used (INFO commandstats). +* More predictable SORT behavior in edge cases. +* INCRBYFLOAT and HINCRBYFLOAT commands. + +-------------------------------------------------------------------------------- + +Credits: Where not specified the implementation and design are done by +Salvatore Sanfilippo and Pieter Noordhuis. Thanks to VMware for making all +this possible. Also many thanks to all the other contributors and the amazing +community we have. + +See commit messages for more credits. + +Cheers, +Salvatore diff --git a/INSTALL b/INSTALL index 2b9fc4a8..3083f1af 100644 --- a/INSTALL +++ b/INSTALL @@ -1,30 +1 @@ -To compile Redis, do the following: - - cd src; make - -The compilation will produce a redis-server binary. - -To install Redis, use - - make install - -and all the binaries will be installed on /usr/local/bin. - -Alternatively: - - make PREFIX=/some/other/directory install - -to have the binaries in /some/other/directory/bin. - -Run the server using the following command line: - - /path/to/redis-server - -This will start a Redis server with the default configuration. - -Otherwise if you want to provide your configuration use: - - /path/to/redis-server /path/to/redis.conf - -You can find an example redis.conf file in the root directory -of this source distribution. +See README diff --git a/MANIFESTO b/MANIFESTO new file mode 100644 index 00000000..ad94fd8e --- /dev/null +++ b/MANIFESTO @@ -0,0 +1,20 @@ +[Note: this is the Redis manifesto, for general information about + installing and running Redis read the README file instead.] + +Redis Manifesto +=============== + +1 - A DSL for Abstract Data Types. Redis is a DSL (Domain Specific Language) that manipulates abstract data types and implemented as a TCP daemon. Commands manipulate a key space where keys are binary-safe strings and values are different kinds of abstract data types. Every data type represents an abstract version of a fundamental data structure. For instance Redis Lists are an abstract representation of linked lists. In Redis, the essence of a data type isn't just the kind of operations that the data types support, but also the space and time complexity of the data type and the operations performed upon it. + +2 - Memory storage is #1. The Redis data set, composed of defined key-value pairs, is primarily stored in the computer's memory. The amount of memory in all kinds of computers, including entry-level servers, is increasing significantly each year. Memory is fast, and allows Redis to have very predictable performance. Datasets composed of 10k or 40 millions keys will perform similarly. Complex data types like Redis Sorted Sets are easy to implement and manipulate in memory with good performance, making Redis very simple. Redis will continue to explore alternative options (where data can be optionally stored on disk, say) but the main goal of the project remains the development of an in-memory database. + +3 - Fundamental data structures for a fundamental API. The Redis API is a direct consequence of fundamental data structures. APIs can often be arbitrary but not an API that resembles the nature of fundamental data structures. If we ever meet intelligent life forms from another part of the universe, they'll likely know, understand and recognize the same basic data structures we have in our computer science books. Redis will avoid intermediate layers in API, so that the complexity is obvious and more complex operations can be performed as the sum of the basic operations. + +4 - Code is like a poem; it's not just something we write to reach some practical result. Sometimes people that are far from the Redis philosophy suggest using other code written by other authors (frequently in other languages) in order to implement something Redis currently lacks. But to us this is like if Shakespeare decided to end Enrico IV using the Paradiso from the Divina Commedia. Is using any external code a bad idea? Not at all. Like in "One Thousand and One Nights" smaller self contained stories are embedded in a bigger story, we'll be happy to use beautiful self contained libraries when needed. At the same time, when writing the Redis story we're trying to write smaller stories that will fit in to other code. + +5 - We're against complexity. We believe designing systems is a fight against complexity. We'll accept to fight the complexity when it's worthwhile but we'll try hard to recognize when a small feature is not worth 1000s of lines of code. Most of the time the best way to fight complexity is by not creating it at all. + +6 - Two levels of API. The Redis API has two levels: 1) a subset of the API fits naturally into a distributed version of Redis and 2) a more complex API that supports multi-key operations. Both are useful if used judiciously but there's no way to make the more complex multi-keys API distributed in an opaque way without violating our other principles. We don't want to provide the illusion of something that will work magically when actually it can't in all cases. Instead we'll provide commands to quickly migrate keys from one instance to another to perform multi-key operations and expose the tradeoffs to the user. + +7 - We optimize for joy. We believe writing code is a lot of hard work, and the only way it can be worth is by enjoying it. When there is no longer joy in writing code, the best thing to do is stop. To prevent this, we'll avoid taking paths that will make Redis less of a joy to develop. + diff --git a/deps/hiredis/CHANGELOG.md b/deps/hiredis/CHANGELOG.md new file mode 100644 index 00000000..d41db8a6 --- /dev/null +++ b/deps/hiredis/CHANGELOG.md @@ -0,0 +1,16 @@ +### 0.10.1 + +* Makefile overhaul. Important to check out if you override one or more + variables using environment variables or via arguments to the "make" tool. + +* Issue #45: Fix potential memory leak for a multi bulk reply with 0 elements + being created by the default reply object functions. + +* Issue #43: Don't crash in an asynchronous context when Redis returns an error + reply after the connection has been made (this happens when the maximum + number of connections is reached). + +### 0.10.0 + +* See commit log. + diff --git a/deps/hiredis/COPYING b/deps/hiredis/COPYING index 3e704e3e..a5fc9739 100644 --- a/deps/hiredis/COPYING +++ b/deps/hiredis/COPYING @@ -1,10 +1,29 @@ -Copyright (c) 2006-2009, Salvatore Sanfilippo +Copyright (c) 2009-2011, Salvatore Sanfilippo +Copyright (c) 2010-2011, Pieter Noordhuis + All rights reserved. -Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + +* Redistributions of source code must retain the above copyright notice, + this list of conditions and the following disclaimer. + +* Redistributions in binary form must reproduce the above copyright notice, + this list of conditions and the following disclaimer in the documentation + and/or other materials provided with the distribution. - * Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer. - * Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution. - * Neither the name of Redis nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission. +* Neither the name of Redis nor the names of its contributors may be used + to endorse or promote products derived from this software without specific + prior written permission. -THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND +ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED +WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR +ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES +(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; +LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON +ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS +SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/deps/hiredis/Makefile b/deps/hiredis/Makefile index 2a84b9b3..16b8767b 100644 --- a/deps/hiredis/Makefile +++ b/deps/hiredis/Makefile @@ -1,115 +1,148 @@ # Hiredis Makefile -# Copyright (C) 2010 Salvatore Sanfilippo +# Copyright (C) 2010-2011 Salvatore Sanfilippo +# Copyright (C) 2010-2011 Pieter Noordhuis # This file is released under the BSD license, see the COPYING file -OBJ = net.o hiredis.o sds.o async.o -BINS = hiredis-example hiredis-test +OBJ=net.o hiredis.o sds.o async.o +BINS=hiredis-example hiredis-test +LIBNAME=libhiredis -uname_S := $(shell sh -c 'uname -s 2>/dev/null || echo not') +HIREDIS_MAJOR=0 +HIREDIS_MINOR=10 + +# Fallback to gcc when $CC is not in $PATH. +CC:=$(shell sh -c 'type $(CC) >/dev/null 2>/dev/null && echo $(CC) || echo gcc') OPTIMIZATION?=-O3 +WARNINGS=-Wall -W -Wstrict-prototypes -Wwrite-strings +DEBUG?= -g -ggdb +REAL_CFLAGS=$(OPTIMIZATION) -fPIC $(CFLAGS) $(WARNINGS) $(DEBUG) $(ARCH) +REAL_LDFLAGS=$(LDFLAGS) $(ARCH) + +DYLIBSUFFIX=so +STLIBSUFFIX=a +DYLIB_MINOR_NAME=$(LIBNAME).$(DYLIBSUFFIX).$(HIREDIS_MAJOR).$(HIREDIS_MINOR) +DYLIB_MAJOR_NAME=$(LIBNAME).$(DYLIBSUFFIX).$(HIREDIS_MAJOR) +DYLIBNAME=$(LIBNAME).$(DYLIBSUFFIX) +DYLIB_MAKE_CMD=$(CC) -shared -Wl,-soname,$(DYLIB_MINOR_NAME) -o $(DYLIBNAME) $(LDFLAGS) +STLIBNAME=$(LIBNAME).$(STLIBSUFFIX) +STLIB_MAKE_CMD=ar rcs $(STLIBNAME) + +# Platform-specific overrides +uname_S := $(shell sh -c 'uname -s 2>/dev/null || echo not') ifeq ($(uname_S),SunOS) - CFLAGS?=-std=c99 -pedantic $(OPTIMIZATION) -fPIC -Wall -W -D__EXTENSIONS__ -D_XPG6 $(ARCH) $(PROF) - CCLINK?=-ldl -lnsl -lsocket -lm -lpthread - LDFLAGS?=-L. -Wl,-R,. - DYLIBNAME?=libhiredis.so - DYLIB_MAKE_CMD?=$(CC) -G -o ${DYLIBNAME} ${OBJ} - STLIBNAME?=libhiredis.a - STLIB_MAKE_CMD?=ar rcs ${STLIBNAME} ${OBJ} -else -ifeq ($(uname_S),Darwin) - CFLAGS?=-std=c99 -pedantic $(OPTIMIZATION) -fPIC -Wall -W -Wstrict-prototypes -Wwrite-strings $(ARCH) $(PROF) - CCLINK?=-lm -pthread - LDFLAGS?=-L. -Wl,-rpath,. - OBJARCH?=-arch i386 -arch x86_64 - DYLIBNAME?=libhiredis.dylib - DYLIB_MAKE_CMD?=libtool -dynamic -o ${DYLIBNAME} -lm ${DEBUG} - ${OBJ} - STLIBNAME?=libhiredis.a - STLIB_MAKE_CMD?=libtool -static -o ${STLIBNAME} - ${OBJ} -else - CFLAGS?=-std=c99 -pedantic $(OPTIMIZATION) -fPIC -Wall -W -Wstrict-prototypes -Wwrite-strings $(ARCH) $(PROF) - CCLINK?=-lm -pthread - LDFLAGS?=-L. -Wl,-rpath,. - DYLIBNAME?=libhiredis.so - DYLIB_MAKE_CMD?=gcc -shared -Wl,-soname,${DYLIBNAME} -o ${DYLIBNAME} ${OBJ} - STLIBNAME?=libhiredis.a - STLIB_MAKE_CMD?=ar rcs ${STLIBNAME} ${OBJ} + REAL_LDFLAGS+= -ldl -lnsl -lsocket + DYLIB_MAKE_CMD=$(CC) -G -o $(DYLIBNAME) -h $(DYLIB_MINOR_NAME) $(LDFLAGS) + INSTALL= cp -r endif +ifeq ($(uname_S),Darwin) + DYLIBSUFFIX=dylib + DYLIB_MINOR_NAME=$(LIBNAME).$(HIREDIS_MAJOR).$(HIREDIS_MINOR).$(DYLIBSUFFIX) + DYLIB_MAJOR_NAME=$(LIBNAME).$(HIREDIS_MAJOR).$(DYLIBSUFFIX) + DYLIB_MAKE_CMD=$(CC) -shared -Wl,-install_name,$(DYLIB_MINOR_NAME) -o $(DYLIBNAME) $(LDFLAGS) endif -CCOPT= $(CFLAGS) $(CCLINK) -DEBUG?= -g -ggdb - -PREFIX?= /usr/local -INSTALL_INC= $(PREFIX)/include/hiredis -INSTALL_LIB= $(PREFIX)/lib -INSTALL= cp -a - -all: ${DYLIBNAME} ${BINS} +all: $(DYLIBNAME) $(BINS) # Deps (use make dep to generate this) -net.o: net.c fmacros.h net.h -async.o: async.c async.h hiredis.h sds.h util.h dict.c dict.h +net.o: net.c fmacros.h net.h hiredis.h +async.o: async.c async.h hiredis.h sds.h dict.c dict.h example.o: example.c hiredis.h -hiredis.o: hiredis.c hiredis.h net.h sds.h util.h +hiredis.o: hiredis.c fmacros.h hiredis.h net.h sds.h sds.o: sds.c sds.h test.o: test.c hiredis.h -${DYLIBNAME}: ${OBJ} - ${DYLIB_MAKE_CMD} +$(DYLIBNAME): $(OBJ) + $(DYLIB_MAKE_CMD) $(OBJ) -${STLIBNAME}: ${OBJ} - ${STLIB_MAKE_CMD} +$(STLIBNAME): $(OBJ) + $(STLIB_MAKE_CMD) $(OBJ) -dynamic: ${DYLIBNAME} -static: ${STLIBNAME} +dynamic: $(DYLIBNAME) +static: $(STLIBNAME) # Binaries: -hiredis-example-libevent: example-libevent.c adapters/libevent.h ${DYLIBNAME} - $(CC) -o $@ $(CCOPT) $(DEBUG) $(LDFLAGS) -lhiredis -levent example-libevent.c +hiredis-example-libevent: example-libevent.c adapters/libevent.h $(STLIBNAME) + $(CC) -o $@ $(REAL_CFLAGS) $(REAL_LDFLAGS) -levent example-libevent.c $(STLIBNAME) -hiredis-example-libev: example-libev.c adapters/libev.h ${DYLIBNAME} - $(CC) -o $@ $(CCOPT) $(DEBUG) $(LDFLAGS) -lhiredis -lev example-libev.c +hiredis-example-libev: example-libev.c adapters/libev.h $(STLIBNAME) + $(CC) -o $@ $(REAL_CFLAGS) $(REAL_LDFLAGS) -lev example-libev.c $(STLIBNAME) ifndef AE_DIR hiredis-example-ae: @echo "Please specify AE_DIR (e.g. /src)" @false else -hiredis-example-ae: example-ae.c adapters/ae.h ${DYLIBNAME} - $(CC) -o $@ $(CCOPT) $(DEBUG) -I$(AE_DIR) $(LDFLAGS) -lhiredis example-ae.c $(AE_DIR)/ae.o $(AE_DIR)/zmalloc.o +hiredis-example-ae: example-ae.c adapters/ae.h $(STLIBNAME) + $(CC) -o $@ $(REAL_CFLAGS) $(REAL_LDFLAGS) -I$(AE_DIR) $(AE_DIR)/ae.o $(AE_DIR)/zmalloc.o example-ae.c $(STLIBNAME) endif -hiredis-%: %.o ${DYLIBNAME} - $(CC) -o $@ $(CCOPT) $(DEBUG) $(LDFLAGS) -lhiredis $< +hiredis-%: %.o $(STLIBNAME) + $(CC) -o $@ $(REAL_LDFLAGS) $< $(STLIBNAME) test: hiredis-test ./hiredis-test +check: hiredis-test + echo \ + "daemonize yes\n" \ + "pidfile /tmp/hiredis-test-redis.pid\n" \ + "port 56379\n" \ + "bind 127.0.0.1\n" \ + "unixsocket /tmp/hiredis-test-redis.sock" \ + | redis-server - + ./hiredis-test -h 127.0.0.1 -p 56379 -s /tmp/hiredis-test-redis.sock || \ + ( kill `cat /tmp/hiredis-test-redis.pid` && false ) + kill `cat /tmp/hiredis-test-redis.pid` + .c.o: - $(CC) -c $(CFLAGS) $(OBJARCH) $(DEBUG) $(COMPILE_TIME) $< + $(CC) -std=c99 -pedantic -c $(REAL_CFLAGS) $< clean: - rm -rf ${DYLIBNAME} ${STLIBNAME} $(BINS) hiredis-example* *.o *.gcda *.gcno *.gcov + rm -rf $(DYLIBNAME) $(STLIBNAME) $(BINS) hiredis-example* *.o *.gcda *.gcno *.gcov dep: $(CC) -MM *.c -install: ${DYLIBNAME} ${STLIBNAME} - mkdir -p $(INSTALL_INC) $(INSTALL_LIB) - $(INSTALL) hiredis.h async.h adapters $(INSTALL_INC) - $(INSTALL) ${DYLIBNAME} ${STLIBNAME} $(INSTALL_LIB) +# Installation related variables and target +PREFIX?=/usr/local +INCLUDE_PATH?=include/hiredis +LIBRARY_PATH?=lib +INSTALL_INCLUDE_PATH= $(PREFIX)/$(INCLUDE_PATH) +INSTALL_LIBRARY_PATH= $(PREFIX)/$(LIBRARY_PATH) + +ifeq ($(uname_S),SunOS) + INSTALL?= cp -r +endif + +INSTALL?= cp -a + +install: $(DYLIBNAME) $(STLIBNAME) + mkdir -p $(INSTALL_INCLUDE_PATH) $(INSTALL_LIBRARY_PATH) + $(INSTALL) hiredis.h async.h adapters $(INSTALL_INCLUDE_PATH) + $(INSTALL) $(DYLIBNAME) $(INSTALL_LIBRARY_PATH)/$(DYLIB_MINOR_NAME) + cd $(INSTALL_LIBRARY_PATH) && ln -sf $(DYLIB_MINOR_NAME) $(DYLIB_MAJOR_NAME) + cd $(INSTALL_LIBRARY_PATH) && ln -sf $(DYLIB_MAJOR_NAME) $(DYLIBNAME) + $(INSTALL) $(STLIBNAME) $(INSTALL_LIBRARY_PATH) 32bit: @echo "" - @echo "WARNING: if it fails under Linux you probably need to install libc6-dev-i386" + @echo "WARNING: if this fails under Linux you probably need to install libc6-dev-i386" @echo "" - $(MAKE) ARCH="-m32" + $(MAKE) CFLAGS="-m32" LDFLAGS="-m32" gprof: - $(MAKE) PROF="-pg" + $(MAKE) CFLAGS="-pg" LDFLAGS="-pg" gcov: - $(MAKE) PROF="-fprofile-arcs -ftest-coverage" + $(MAKE) CFLAGS="-fprofile-arcs -ftest-coverage" LDFLAGS="-fprofile-arcs" + +coverage: gcov + make check + mkdir -p tmp/lcov + lcov -d . -c -o tmp/lcov/hiredis.info + genhtml --legend -o tmp/lcov/report tmp/lcov/hiredis.info noopt: $(MAKE) OPTIMIZATION="" + +.PHONY: all test check clean dep install 32bit gprof gcov noopt diff --git a/deps/hiredis/README.md b/deps/hiredis/README.md index 5a77cd38..a58101cc 100644 --- a/deps/hiredis/README.md +++ b/deps/hiredis/README.md @@ -116,6 +116,12 @@ Note that this function will take care of freeing sub-replies objects contained in arrays and nested arrays, so there is no need for the user to free the sub replies (it is actually harmful and will corrupt the memory). +**Important:** the current version of hiredis (0.10.0) free's replies when the +asynchronous API is used. This means you should not call `freeReplyObject` when +you use this API. The reply is cleaned up by hiredis _after_ the callback +returns. This behavior will probably change in future releases, so make sure to +keep an eye on the changelog when upgrading (see issue #39). + ### Cleaning up To disconnect and free the context the following function can be used: @@ -280,7 +286,8 @@ is being disconnected per user-request, no new commands may be added to the outp returned on calls to the `redisAsyncCommand` family. If the reply for a command with a `NULL` callback is read, it is immediately free'd. When the callback -for a command is non-`NULL`, it is responsible for cleaning up the reply. +for a command is non-`NULL`, the memory is free'd immediately following the callback: the reply is only +valid for the duration of the callback. All pending callbacks are called with a `NULL` reply when the context encountered an error. @@ -303,7 +310,41 @@ See the `adapters/` directory for bindings to *libev* and *libevent*. ## Reply parsing API -To be done. +Hiredis comes with a reply parsing API that makes it easy for writing higher +level language bindings. + +The reply parsing API consists of the following functions: + + redisReader *redisReaderCreate(void); + void redisReaderFree(redisReader *reader); + int redisReaderFeed(redisReader *reader, const char *buf, size_t len); + int redisReaderGetReply(redisReader *reader, void **reply); + +### Usage + +The function `redisReaderCreate` creates a `redisReader` structure that holds a +buffer with unparsed data and state for the protocol parser. + +Incoming data -- most likely from a socket -- can be placed in the internal +buffer of the `redisReader` using `redisReaderFeed`. This function will make a +copy of the buffer pointed to by `buf` for `len` bytes. This data is parsed +when `redisReaderGetReply` is called. This function returns an integer status +and a reply object (as described above) via `void **reply`. The returned status +can be either `REDIS_OK` or `REDIS_ERR`, where the latter means something went +wrong (either a protocol error, or an out of memory error). + +### Customizing replies + +The function `redisReaderGetReply` creates `redisReply` and makes the function +argument `reply` point to the created `redisReply` variable. For instance, if +the response of type `REDIS_REPLY_STATUS` then the `str` field of `redisReply` +will hold the status as a vanilla C string. However, the functions that are +responsible for creating instances of the `redisReply` can be customized by +setting the `fn` field on the `redisReader` struct. This should be done +immediately after creating the `redisReader`. + +For example, [hiredis-rb](https://github.com/pietern/hiredis-rb/blob/master/ext/hiredis_ext/reader.c) +uses customized reply object functions to create Ruby objects. ## AUTHORS diff --git a/deps/hiredis/TODO b/deps/hiredis/TODO deleted file mode 100644 index de70b947..00000000 --- a/deps/hiredis/TODO +++ /dev/null @@ -1,2 +0,0 @@ -- add redisCommandVector() -- add support for pipelining diff --git a/deps/hiredis/adapters/ae.h b/deps/hiredis/adapters/ae.h index b8b2228e..65235f80 100644 --- a/deps/hiredis/adapters/ae.h +++ b/deps/hiredis/adapters/ae.h @@ -1,3 +1,5 @@ +#ifndef __HIREDIS_AE_H__ +#define __HIREDIS_AE_H__ #include #include #include "../hiredis.h" @@ -10,21 +12,21 @@ typedef struct redisAeEvents { int reading, writing; } redisAeEvents; -void redisAeReadEvent(aeEventLoop *el, int fd, void *privdata, int mask) { +static void redisAeReadEvent(aeEventLoop *el, int fd, void *privdata, int mask) { ((void)el); ((void)fd); ((void)mask); redisAeEvents *e = (redisAeEvents*)privdata; redisAsyncHandleRead(e->context); } -void redisAeWriteEvent(aeEventLoop *el, int fd, void *privdata, int mask) { +static void redisAeWriteEvent(aeEventLoop *el, int fd, void *privdata, int mask) { ((void)el); ((void)fd); ((void)mask); redisAeEvents *e = (redisAeEvents*)privdata; redisAsyncHandleWrite(e->context); } -void redisAeAddRead(void *privdata) { +static void redisAeAddRead(void *privdata) { redisAeEvents *e = (redisAeEvents*)privdata; aeEventLoop *loop = e->loop; if (!e->reading) { @@ -33,7 +35,7 @@ void redisAeAddRead(void *privdata) { } } -void redisAeDelRead(void *privdata) { +static void redisAeDelRead(void *privdata) { redisAeEvents *e = (redisAeEvents*)privdata; aeEventLoop *loop = e->loop; if (e->reading) { @@ -42,7 +44,7 @@ void redisAeDelRead(void *privdata) { } } -void redisAeAddWrite(void *privdata) { +static void redisAeAddWrite(void *privdata) { redisAeEvents *e = (redisAeEvents*)privdata; aeEventLoop *loop = e->loop; if (!e->writing) { @@ -51,7 +53,7 @@ void redisAeAddWrite(void *privdata) { } } -void redisAeDelWrite(void *privdata) { +static void redisAeDelWrite(void *privdata) { redisAeEvents *e = (redisAeEvents*)privdata; aeEventLoop *loop = e->loop; if (e->writing) { @@ -60,19 +62,19 @@ void redisAeDelWrite(void *privdata) { } } -void redisAeCleanup(void *privdata) { +static void redisAeCleanup(void *privdata) { redisAeEvents *e = (redisAeEvents*)privdata; redisAeDelRead(privdata); redisAeDelWrite(privdata); free(e); } -int redisAeAttach(aeEventLoop *loop, redisAsyncContext *ac) { +static int redisAeAttach(aeEventLoop *loop, redisAsyncContext *ac) { redisContext *c = &(ac->c); redisAeEvents *e; /* Nothing should be attached when something is already attached */ - if (ac->_adapter_data != NULL) + if (ac->ev.data != NULL) return REDIS_ERR; /* Create container for context and r/w events */ @@ -83,13 +85,13 @@ int redisAeAttach(aeEventLoop *loop, redisAsyncContext *ac) { e->reading = e->writing = 0; /* Register functions to start/stop listening for events */ - ac->evAddRead = redisAeAddRead; - ac->evDelRead = redisAeDelRead; - ac->evAddWrite = redisAeAddWrite; - ac->evDelWrite = redisAeDelWrite; - ac->evCleanup = redisAeCleanup; - ac->_adapter_data = e; + ac->ev.addRead = redisAeAddRead; + ac->ev.delRead = redisAeDelRead; + ac->ev.addWrite = redisAeAddWrite; + ac->ev.delWrite = redisAeDelWrite; + ac->ev.cleanup = redisAeCleanup; + ac->ev.data = e; return REDIS_OK; } - +#endif diff --git a/deps/hiredis/adapters/libev.h b/deps/hiredis/adapters/libev.h index 3b9ed656..534d7436 100644 --- a/deps/hiredis/adapters/libev.h +++ b/deps/hiredis/adapters/libev.h @@ -1,3 +1,6 @@ +#ifndef __HIREDIS_LIBEV_H__ +#define __HIREDIS_LIBEV_H__ +#include #include #include #include "../hiredis.h" @@ -10,7 +13,7 @@ typedef struct redisLibevEvents { ev_io rev, wev; } redisLibevEvents; -void redisLibevReadEvent(EV_P_ ev_io *watcher, int revents) { +static void redisLibevReadEvent(EV_P_ ev_io *watcher, int revents) { #if EV_MULTIPLICITY ((void)loop); #endif @@ -20,7 +23,7 @@ void redisLibevReadEvent(EV_P_ ev_io *watcher, int revents) { redisAsyncHandleRead(e->context); } -void redisLibevWriteEvent(EV_P_ ev_io *watcher, int revents) { +static void redisLibevWriteEvent(EV_P_ ev_io *watcher, int revents) { #if EV_MULTIPLICITY ((void)loop); #endif @@ -30,7 +33,7 @@ void redisLibevWriteEvent(EV_P_ ev_io *watcher, int revents) { redisAsyncHandleWrite(e->context); } -void redisLibevAddRead(void *privdata) { +static void redisLibevAddRead(void *privdata) { redisLibevEvents *e = (redisLibevEvents*)privdata; struct ev_loop *loop = e->loop; ((void)loop); @@ -40,7 +43,7 @@ void redisLibevAddRead(void *privdata) { } } -void redisLibevDelRead(void *privdata) { +static void redisLibevDelRead(void *privdata) { redisLibevEvents *e = (redisLibevEvents*)privdata; struct ev_loop *loop = e->loop; ((void)loop); @@ -50,7 +53,7 @@ void redisLibevDelRead(void *privdata) { } } -void redisLibevAddWrite(void *privdata) { +static void redisLibevAddWrite(void *privdata) { redisLibevEvents *e = (redisLibevEvents*)privdata; struct ev_loop *loop = e->loop; ((void)loop); @@ -60,7 +63,7 @@ void redisLibevAddWrite(void *privdata) { } } -void redisLibevDelWrite(void *privdata) { +static void redisLibevDelWrite(void *privdata) { redisLibevEvents *e = (redisLibevEvents*)privdata; struct ev_loop *loop = e->loop; ((void)loop); @@ -70,19 +73,19 @@ void redisLibevDelWrite(void *privdata) { } } -void redisLibevCleanup(void *privdata) { +static void redisLibevCleanup(void *privdata) { redisLibevEvents *e = (redisLibevEvents*)privdata; redisLibevDelRead(privdata); redisLibevDelWrite(privdata); free(e); } -int redisLibevAttach(EV_P_ redisAsyncContext *ac) { +static int redisLibevAttach(EV_P_ redisAsyncContext *ac) { redisContext *c = &(ac->c); redisLibevEvents *e; /* Nothing should be attached when something is already attached */ - if (ac->_adapter_data != NULL) + if (ac->ev.data != NULL) return REDIS_ERR; /* Create container for context and r/w events */ @@ -98,12 +101,12 @@ int redisLibevAttach(EV_P_ redisAsyncContext *ac) { e->wev.data = e; /* Register functions to start/stop listening for events */ - ac->evAddRead = redisLibevAddRead; - ac->evDelRead = redisLibevDelRead; - ac->evAddWrite = redisLibevAddWrite; - ac->evDelWrite = redisLibevDelWrite; - ac->evCleanup = redisLibevCleanup; - ac->_adapter_data = e; + ac->ev.addRead = redisLibevAddRead; + ac->ev.delRead = redisLibevDelRead; + ac->ev.addWrite = redisLibevAddWrite; + ac->ev.delWrite = redisLibevDelWrite; + ac->ev.cleanup = redisLibevCleanup; + ac->ev.data = e; /* Initialize read/write events */ ev_io_init(&e->rev,redisLibevReadEvent,c->fd,EV_READ); @@ -111,3 +114,4 @@ int redisLibevAttach(EV_P_ redisAsyncContext *ac) { return REDIS_OK; } +#endif diff --git a/deps/hiredis/adapters/libevent.h b/deps/hiredis/adapters/libevent.h index dc1f5c73..4055ec0f 100644 --- a/deps/hiredis/adapters/libevent.h +++ b/deps/hiredis/adapters/libevent.h @@ -1,4 +1,5 @@ -#include +#ifndef __HIREDIS_LIBEVENT_H__ +#define __HIREDIS_LIBEVENT_H__ #include #include "../hiredis.h" #include "../async.h" @@ -8,51 +9,51 @@ typedef struct redisLibeventEvents { struct event rev, wev; } redisLibeventEvents; -void redisLibeventReadEvent(int fd, short event, void *arg) { +static void redisLibeventReadEvent(int fd, short event, void *arg) { ((void)fd); ((void)event); redisLibeventEvents *e = (redisLibeventEvents*)arg; redisAsyncHandleRead(e->context); } -void redisLibeventWriteEvent(int fd, short event, void *arg) { +static void redisLibeventWriteEvent(int fd, short event, void *arg) { ((void)fd); ((void)event); redisLibeventEvents *e = (redisLibeventEvents*)arg; redisAsyncHandleWrite(e->context); } -void redisLibeventAddRead(void *privdata) { +static void redisLibeventAddRead(void *privdata) { redisLibeventEvents *e = (redisLibeventEvents*)privdata; event_add(&e->rev,NULL); } -void redisLibeventDelRead(void *privdata) { +static void redisLibeventDelRead(void *privdata) { redisLibeventEvents *e = (redisLibeventEvents*)privdata; event_del(&e->rev); } -void redisLibeventAddWrite(void *privdata) { +static void redisLibeventAddWrite(void *privdata) { redisLibeventEvents *e = (redisLibeventEvents*)privdata; event_add(&e->wev,NULL); } -void redisLibeventDelWrite(void *privdata) { +static void redisLibeventDelWrite(void *privdata) { redisLibeventEvents *e = (redisLibeventEvents*)privdata; event_del(&e->wev); } -void redisLibeventCleanup(void *privdata) { +static void redisLibeventCleanup(void *privdata) { redisLibeventEvents *e = (redisLibeventEvents*)privdata; event_del(&e->rev); event_del(&e->wev); free(e); } -int redisLibeventAttach(redisAsyncContext *ac, struct event_base *base) { +static int redisLibeventAttach(redisAsyncContext *ac, struct event_base *base) { redisContext *c = &(ac->c); redisLibeventEvents *e; /* Nothing should be attached when something is already attached */ - if (ac->_adapter_data != NULL) + if (ac->ev.data != NULL) return REDIS_ERR; /* Create container for context and r/w events */ @@ -60,12 +61,12 @@ int redisLibeventAttach(redisAsyncContext *ac, struct event_base *base) { e->context = ac; /* Register functions to start/stop listening for events */ - ac->evAddRead = redisLibeventAddRead; - ac->evDelRead = redisLibeventDelRead; - ac->evAddWrite = redisLibeventAddWrite; - ac->evDelWrite = redisLibeventDelWrite; - ac->evCleanup = redisLibeventCleanup; - ac->_adapter_data = e; + ac->ev.addRead = redisLibeventAddRead; + ac->ev.delRead = redisLibeventDelRead; + ac->ev.addWrite = redisLibeventAddWrite; + ac->ev.delWrite = redisLibeventDelWrite; + ac->ev.cleanup = redisLibeventCleanup; + ac->ev.data = e; /* Initialize and install read/write events */ event_set(&e->rev,c->fd,EV_READ,redisLibeventReadEvent,e); @@ -74,3 +75,4 @@ int redisLibeventAttach(redisAsyncContext *ac, struct event_base *base) { event_base_set(base,&e->wev); return REDIS_OK; } +#endif diff --git a/deps/hiredis/async.c b/deps/hiredis/async.c index 76c4cc3a..f83e2f51 100644 --- a/deps/hiredis/async.c +++ b/deps/hiredis/async.c @@ -1,6 +1,6 @@ /* - * Copyright (c) 2009-2010, Salvatore Sanfilippo - * Copyright (c) 2010, Pieter Noordhuis + * Copyright (c) 2009-2011, Salvatore Sanfilippo + * Copyright (c) 2010-2011, Pieter Noordhuis * * All rights reserved. * @@ -29,14 +29,33 @@ * POSSIBILITY OF SUCH DAMAGE. */ +#include "fmacros.h" +#include #include #include #include #include +#include #include "async.h" +#include "net.h" #include "dict.c" #include "sds.h" -#include "util.h" + +#define _EL_ADD_READ(ctx) do { \ + if ((ctx)->ev.addRead) (ctx)->ev.addRead((ctx)->ev.data); \ + } while(0) +#define _EL_DEL_READ(ctx) do { \ + if ((ctx)->ev.delRead) (ctx)->ev.delRead((ctx)->ev.data); \ + } while(0) +#define _EL_ADD_WRITE(ctx) do { \ + if ((ctx)->ev.addWrite) (ctx)->ev.addWrite((ctx)->ev.data); \ + } while(0) +#define _EL_DEL_WRITE(ctx) do { \ + if ((ctx)->ev.delWrite) (ctx)->ev.delWrite((ctx)->ev.data); \ + } while(0) +#define _EL_CLEANUP(ctx) do { \ + if ((ctx)->ev.cleanup) (ctx)->ev.cleanup((ctx)->ev.data); \ + } while(0); /* Forward declaration of function in hiredis.c */ void __redisAppendCommand(redisContext *c, char *cmd, size_t len); @@ -136,11 +155,6 @@ redisAsyncContext *redisAsyncConnectUnix(const char *path) { return ac; } -int redisAsyncSetReplyObjectFunctions(redisAsyncContext *ac, redisReplyObjectFunctions *fn) { - redisContext *c = &(ac->c); - return redisSetReplyObjectFunctions(c,fn); -} - int redisAsyncSetConnectCallback(redisAsyncContext *ac, redisConnectCallback *fn) { if (ac->onConnect == NULL) { ac->onConnect = fn; @@ -148,7 +162,7 @@ int redisAsyncSetConnectCallback(redisAsyncContext *ac, redisConnectCallback *fn /* The common way to detect an established connection is to wait for * the first write event to be fired. This assumes the related event * library functions are already set. */ - if (ac->ev.addWrite) ac->ev.addWrite(ac->ev.data); + _EL_ADD_WRITE(ac); return REDIS_OK; } return REDIS_ERR; @@ -168,7 +182,6 @@ static int __redisPushCallback(redisCallbackList *list, redisCallback *source) { /* Copy callback from stack to heap */ cb = malloc(sizeof(*cb)); - if (!cb) redisOOM(); if (source != NULL) { memcpy(cb,source,sizeof(*cb)); cb->next = NULL; @@ -237,7 +250,7 @@ static void __redisAsyncFree(redisAsyncContext *ac) { dictRelease(ac->sub.patterns); /* Signal event lib to clean up */ - if (ac->ev.cleanup) ac->ev.cleanup(ac->ev.data); + _EL_CLEANUP(ac); /* Execute disconnect callback. When redisAsyncFree() initiated destroying * this context, the status will always be REDIS_OK. */ @@ -368,14 +381,27 @@ void redisProcessCallbacks(redisAsyncContext *ac) { /* Even if the context is subscribed, pending regular callbacks will * get a reply before pub/sub messages arrive. */ if (__redisShiftCallback(&ac->replies,&cb) != REDIS_OK) { - /* No more regular callbacks, the context *must* be subscribed. */ + /* A spontaneous reply in a not-subscribed context can only be the + * error reply that is sent when a new connection exceeds the + * maximum number of allowed connections on the server side. This + * is seen as an error instead of a regular reply because the + * server closes the connection after sending it. To prevent the + * error from being overwritten by an EOF error the connection is + * closed here. See issue #43. */ + if ( !(c->flags & REDIS_SUBSCRIBED) && ((redisReply*)reply)->type == REDIS_REPLY_ERROR ) { + c->err = REDIS_ERR_OTHER; + snprintf(c->errstr,sizeof(c->errstr),"%s",((redisReply*)reply)->str); + __redisAsyncDisconnect(ac); + return; + } + /* No more regular callbacks and no errors, the context *must* be subscribed. */ assert(c->flags & REDIS_SUBSCRIBED); __redisGetSubscribeCallback(ac,reply,&cb); } if (cb.fn != NULL) { __redisRunCallback(ac,&cb,reply); - c->fn->freeObject(reply); + c->reader->fn->freeObject(reply); /* Proceed with free'ing when redisAsyncFree() was called. */ if (c->flags & REDIS_FREEING) { @@ -387,7 +413,7 @@ void redisProcessCallbacks(redisAsyncContext *ac) { * or there were no callbacks to begin with. Either way, don't * abort with an error, but simply ignore it because the client * doesn't know what the server will spit out over the wire. */ - c->fn->freeObject(reply); + c->reader->fn->freeObject(reply); } } @@ -396,17 +422,48 @@ void redisProcessCallbacks(redisAsyncContext *ac) { __redisAsyncDisconnect(ac); } +/* Internal helper function to detect socket status the first time a read or + * write event fires. When connecting was not succesful, the connect callback + * is called with a REDIS_ERR status and the context is free'd. */ +static int __redisAsyncHandleConnect(redisAsyncContext *ac) { + redisContext *c = &(ac->c); + + if (redisCheckSocketError(c,c->fd) == REDIS_ERR) { + /* Try again later when connect(2) is still in progress. */ + if (errno == EINPROGRESS) + return REDIS_OK; + + if (ac->onConnect) ac->onConnect(ac,REDIS_ERR); + __redisAsyncDisconnect(ac); + return REDIS_ERR; + } + + /* Mark context as connected. */ + c->flags |= REDIS_CONNECTED; + if (ac->onConnect) ac->onConnect(ac,REDIS_OK); + return REDIS_OK; +} + /* This function should be called when the socket is readable. * It processes all replies that can be read and executes their callbacks. */ void redisAsyncHandleRead(redisAsyncContext *ac) { redisContext *c = &(ac->c); + if (!(c->flags & REDIS_CONNECTED)) { + /* Abort connect was not successful. */ + if (__redisAsyncHandleConnect(ac) != REDIS_OK) + return; + /* Try again later when the context is still not connected. */ + if (!(c->flags & REDIS_CONNECTED)) + return; + } + if (redisBufferRead(c) == REDIS_ERR) { __redisAsyncDisconnect(ac); } else { /* Always re-schedule reads */ - if (ac->ev.addRead) ac->ev.addRead(ac->ev.data); + _EL_ADD_READ(ac); redisProcessCallbacks(ac); } } @@ -415,24 +472,26 @@ void redisAsyncHandleWrite(redisAsyncContext *ac) { redisContext *c = &(ac->c); int done = 0; + if (!(c->flags & REDIS_CONNECTED)) { + /* Abort connect was not successful. */ + if (__redisAsyncHandleConnect(ac) != REDIS_OK) + return; + /* Try again later when the context is still not connected. */ + if (!(c->flags & REDIS_CONNECTED)) + return; + } + if (redisBufferWrite(c,&done) == REDIS_ERR) { __redisAsyncDisconnect(ac); } else { /* Continue writing when not done, stop writing otherwise */ - if (!done) { - if (ac->ev.addWrite) ac->ev.addWrite(ac->ev.data); - } else { - if (ac->ev.delWrite) ac->ev.delWrite(ac->ev.data); - } + if (!done) + _EL_ADD_WRITE(ac); + else + _EL_DEL_WRITE(ac); /* Always schedule reads after writes */ - if (ac->ev.addRead) ac->ev.addRead(ac->ev.data); - - /* Fire onConnect when this is the first write event. */ - if (!(c->flags & REDIS_CONNECTED)) { - c->flags |= REDIS_CONNECTED; - if (ac->onConnect) ac->onConnect(ac); - } + _EL_ADD_READ(ac); } } @@ -510,7 +569,7 @@ static int __redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void __redisAppendCommand(c,cmd,len); /* Always schedule a write when the write buffer is non-empty */ - if (ac->ev.addWrite) ac->ev.addWrite(ac->ev.data); + _EL_ADD_WRITE(ac); return REDIS_OK; } diff --git a/deps/hiredis/async.h b/deps/hiredis/async.h index ba2b6f54..268274e8 100644 --- a/deps/hiredis/async.h +++ b/deps/hiredis/async.h @@ -1,6 +1,6 @@ /* - * Copyright (c) 2009-2010, Salvatore Sanfilippo - * Copyright (c) 2010, Pieter Noordhuis + * Copyright (c) 2009-2011, Salvatore Sanfilippo + * Copyright (c) 2010-2011, Pieter Noordhuis * * All rights reserved. * @@ -55,7 +55,7 @@ typedef struct redisCallbackList { /* Connection callback prototypes */ typedef void (redisDisconnectCallback)(const struct redisAsyncContext*, int status); -typedef void (redisConnectCallback)(const struct redisAsyncContext*); +typedef void (redisConnectCallback)(const struct redisAsyncContext*, int status); /* Context for an async connection to Redis */ typedef struct redisAsyncContext { @@ -103,7 +103,6 @@ typedef struct redisAsyncContext { /* Functions that proxy to hiredis */ redisAsyncContext *redisAsyncConnect(const char *ip, int port); redisAsyncContext *redisAsyncConnectUnix(const char *path); -int redisAsyncSetReplyObjectFunctions(redisAsyncContext *ac, redisReplyObjectFunctions *fn); int redisAsyncSetConnectCallback(redisAsyncContext *ac, redisConnectCallback *fn); int redisAsyncSetDisconnectCallback(redisAsyncContext *ac, redisDisconnectCallback *fn); void redisAsyncDisconnect(redisAsyncContext *ac); diff --git a/deps/hiredis/example-ae.c b/deps/hiredis/example-ae.c index 28c34dc9..5ed34a3a 100644 --- a/deps/hiredis/example-ae.c +++ b/deps/hiredis/example-ae.c @@ -18,17 +18,20 @@ void getCallback(redisAsyncContext *c, void *r, void *privdata) { redisAsyncDisconnect(c); } -void connectCallback(const redisAsyncContext *c) { - ((void)c); - printf("connected...\n"); +void connectCallback(const redisAsyncContext *c, int status) { + if (status != REDIS_OK) { + printf("Error: %s\n", c->errstr); + return; + } + printf("Connected...\n"); } void disconnectCallback(const redisAsyncContext *c, int status) { if (status != REDIS_OK) { printf("Error: %s\n", c->errstr); + return; } - printf("disconnected...\n"); - aeStop(loop); + printf("Disconnected...\n"); } int main (int argc, char **argv) { diff --git a/deps/hiredis/example-libev.c b/deps/hiredis/example-libev.c index 8efa1e39..7894f1f4 100644 --- a/deps/hiredis/example-libev.c +++ b/deps/hiredis/example-libev.c @@ -15,16 +15,20 @@ void getCallback(redisAsyncContext *c, void *r, void *privdata) { redisAsyncDisconnect(c); } -void connectCallback(const redisAsyncContext *c) { - ((void)c); - printf("connected...\n"); +void connectCallback(const redisAsyncContext *c, int status) { + if (status != REDIS_OK) { + printf("Error: %s\n", c->errstr); + return; + } + printf("Connected...\n"); } void disconnectCallback(const redisAsyncContext *c, int status) { if (status != REDIS_OK) { printf("Error: %s\n", c->errstr); + return; } - printf("disconnected...\n"); + printf("Disconnected...\n"); } int main (int argc, char **argv) { diff --git a/deps/hiredis/example-libevent.c b/deps/hiredis/example-libevent.c index f6f8c832..9da8e02b 100644 --- a/deps/hiredis/example-libevent.c +++ b/deps/hiredis/example-libevent.c @@ -15,16 +15,20 @@ void getCallback(redisAsyncContext *c, void *r, void *privdata) { redisAsyncDisconnect(c); } -void connectCallback(const redisAsyncContext *c) { - ((void)c); - printf("connected...\n"); +void connectCallback(const redisAsyncContext *c, int status) { + if (status != REDIS_OK) { + printf("Error: %s\n", c->errstr); + return; + } + printf("Connected...\n"); } void disconnectCallback(const redisAsyncContext *c, int status) { if (status != REDIS_OK) { printf("Error: %s\n", c->errstr); + return; } - printf("disconnected...\n"); + printf("Disconnected...\n"); } int main (int argc, char **argv) { diff --git a/deps/hiredis/fmacros.h b/deps/hiredis/fmacros.h index 65f9692c..21cd9cfe 100644 --- a/deps/hiredis/fmacros.h +++ b/deps/hiredis/fmacros.h @@ -1,12 +1,14 @@ #ifndef __HIREDIS_FMACRO_H #define __HIREDIS_FMACRO_H -#ifndef _BSD_SOURCE +#if !defined(_BSD_SOURCE) #define _BSD_SOURCE #endif -#ifdef __linux__ -#define _XOPEN_SOURCE 700 +#if defined(__sun__) +#define _POSIX_C_SOURCE 200112L +#elif defined(__linux__) +#define _XOPEN_SOURCE 600 #else #define _XOPEN_SOURCE #endif diff --git a/deps/hiredis/hiredis.c b/deps/hiredis/hiredis.c index 976e94f9..e6109db8 100644 --- a/deps/hiredis/hiredis.c +++ b/deps/hiredis/hiredis.c @@ -1,6 +1,6 @@ /* - * Copyright (c) 2009-2010, Salvatore Sanfilippo - * Copyright (c) 2010, Pieter Noordhuis + * Copyright (c) 2009-2011, Salvatore Sanfilippo + * Copyright (c) 2010-2011, Pieter Noordhuis * * All rights reserved. * @@ -29,6 +29,7 @@ * POSSIBILITY OF SUCH DAMAGE. */ +#include "fmacros.h" #include #include #include @@ -39,30 +40,15 @@ #include "hiredis.h" #include "net.h" #include "sds.h" -#include "util.h" - -typedef struct redisReader { - struct redisReplyObjectFunctions *fn; - sds error; /* holds optional error */ - void *reply; /* holds temporary reply */ - - sds buf; /* read buffer */ - size_t pos; /* buffer cursor */ - size_t len; /* buffer length */ - - redisReadTask rstack[9]; /* stack of read tasks */ - int ridx; /* index of stack */ - void *privdata; /* user-settable arbitrary field */ -} redisReader; static redisReply *createReplyObject(int type); static void *createStringObject(const redisReadTask *task, char *str, size_t len); static void *createArrayObject(const redisReadTask *task, int elements); static void *createIntegerObject(const redisReadTask *task, long long value); static void *createNilObject(const redisReadTask *task); -static void redisSetReplyReaderError(redisReader *r, sds err); -/* Default set of functions to build the reply. */ +/* Default set of functions to build the reply. Keep in mind that such a + * function returning NULL is interpreted as OOM. */ static redisReplyObjectFunctions defaultFunctions = { createStringObject, createArrayObject, @@ -73,9 +59,11 @@ static redisReplyObjectFunctions defaultFunctions = { /* Create a reply object */ static redisReply *createReplyObject(int type) { - redisReply *r = malloc(sizeof(*r)); + redisReply *r = calloc(1,sizeof(*r)); + + if (r == NULL) + return NULL; - if (!r) redisOOM(); r->type = type; return r; } @@ -89,35 +77,49 @@ void freeReplyObject(void *reply) { case REDIS_REPLY_INTEGER: break; /* Nothing to free */ case REDIS_REPLY_ARRAY: - for (j = 0; j < r->elements; j++) - if (r->element[j]) freeReplyObject(r->element[j]); - free(r->element); + if (r->element != NULL) { + for (j = 0; j < r->elements; j++) + if (r->element[j] != NULL) + freeReplyObject(r->element[j]); + free(r->element); + } break; case REDIS_REPLY_ERROR: case REDIS_REPLY_STATUS: case REDIS_REPLY_STRING: - free(r->str); + if (r->str != NULL) + free(r->str); break; } free(r); } static void *createStringObject(const redisReadTask *task, char *str, size_t len) { - redisReply *r = createReplyObject(task->type); - char *value = malloc(len+1); - if (!value) redisOOM(); - assert(task->type == REDIS_REPLY_ERROR || + redisReply *r, *parent; + char *buf; + + r = createReplyObject(task->type); + if (r == NULL) + return NULL; + + buf = malloc(len+1); + if (buf == NULL) { + freeReplyObject(r); + return NULL; + } + + assert(task->type == REDIS_REPLY_ERROR || task->type == REDIS_REPLY_STATUS || task->type == REDIS_REPLY_STRING); /* Copy string value */ - memcpy(value,str,len); - value[len] = '\0'; - r->str = value; + memcpy(buf,str,len); + buf[len] = '\0'; + r->str = buf; r->len = len; if (task->parent) { - redisReply *parent = task->parent->obj; + parent = task->parent->obj; assert(parent->type == REDIS_REPLY_ARRAY); parent->element[task->idx] = r; } @@ -125,12 +127,24 @@ static void *createStringObject(const redisReadTask *task, char *str, size_t len } static void *createArrayObject(const redisReadTask *task, int elements) { - redisReply *r = createReplyObject(REDIS_REPLY_ARRAY); + redisReply *r, *parent; + + r = createReplyObject(REDIS_REPLY_ARRAY); + if (r == NULL) + return NULL; + + if (elements > 0) { + r->element = calloc(elements,sizeof(redisReply*)); + if (r->element == NULL) { + freeReplyObject(r); + return NULL; + } + } + r->elements = elements; - if ((r->element = calloc(sizeof(redisReply*),elements)) == NULL) - redisOOM(); + if (task->parent) { - redisReply *parent = task->parent->obj; + parent = task->parent->obj; assert(parent->type == REDIS_REPLY_ARRAY); parent->element[task->idx] = r; } @@ -138,10 +152,16 @@ static void *createArrayObject(const redisReadTask *task, int elements) { } static void *createIntegerObject(const redisReadTask *task, long long value) { - redisReply *r = createReplyObject(REDIS_REPLY_INTEGER); + redisReply *r, *parent; + + r = createReplyObject(REDIS_REPLY_INTEGER); + if (r == NULL) + return NULL; + r->integer = value; + if (task->parent) { - redisReply *parent = task->parent->obj; + parent = task->parent->obj; assert(parent->type == REDIS_REPLY_ARRAY); parent->element[task->idx] = r; } @@ -149,15 +169,83 @@ static void *createIntegerObject(const redisReadTask *task, long long value) { } static void *createNilObject(const redisReadTask *task) { - redisReply *r = createReplyObject(REDIS_REPLY_NIL); + redisReply *r, *parent; + + r = createReplyObject(REDIS_REPLY_NIL); + if (r == NULL) + return NULL; + if (task->parent) { - redisReply *parent = task->parent->obj; + parent = task->parent->obj; assert(parent->type == REDIS_REPLY_ARRAY); parent->element[task->idx] = r; } return r; } +static void __redisReaderSetError(redisReader *r, int type, const char *str) { + size_t len; + + if (r->reply != NULL && r->fn && r->fn->freeObject) { + r->fn->freeObject(r->reply); + r->reply = NULL; + } + + /* Clear input buffer on errors. */ + if (r->buf != NULL) { + sdsfree(r->buf); + r->buf = NULL; + r->pos = r->len = 0; + } + + /* Reset task stack. */ + r->ridx = -1; + + /* Set error. */ + r->err = type; + len = strlen(str); + len = len < (sizeof(r->errstr)-1) ? len : (sizeof(r->errstr)-1); + memcpy(r->errstr,str,len); + r->errstr[len] = '\0'; +} + +static size_t chrtos(char *buf, size_t size, char byte) { + size_t len = 0; + + switch(byte) { + case '\\': + case '"': + len = snprintf(buf,size,"\"\\%c\"",byte); + break; + case '\n': len = snprintf(buf,size,"\"\\n\""); break; + case '\r': len = snprintf(buf,size,"\"\\r\""); break; + case '\t': len = snprintf(buf,size,"\"\\t\""); break; + case '\a': len = snprintf(buf,size,"\"\\a\""); break; + case '\b': len = snprintf(buf,size,"\"\\b\""); break; + default: + if (isprint(byte)) + len = snprintf(buf,size,"\"%c\"",byte); + else + len = snprintf(buf,size,"\"\\x%02x\"",(unsigned char)byte); + break; + } + + return len; +} + +static void __redisReaderSetErrorProtocolByte(redisReader *r, char byte) { + char cbuf[8], sbuf[128]; + + chrtos(cbuf,sizeof(cbuf),byte); + snprintf(sbuf,sizeof(sbuf), + "Protocol error, got %s as reply type byte", cbuf); + __redisReaderSetError(r,REDIS_ERR_PROTOCOL,sbuf); +} + +static void __redisReaderSetErrorOOM(redisReader *r) { + __redisReaderSetError(r,REDIS_ERR_OOM,"Out of memory"); +} + static char *readBytes(redisReader *r, unsigned int bytes) { char *p; if (r->len-r->pos >= bytes) { @@ -284,12 +372,18 @@ static int processLineItem(redisReader *r) { obj = (void*)(size_t)(cur->type); } + if (obj == NULL) { + __redisReaderSetErrorOOM(r); + return REDIS_ERR; + } + /* Set reply if this is the root object. */ if (r->ridx == 0) r->reply = obj; moveToNextTask(r); - return 0; + return REDIS_OK; } - return -1; + + return REDIS_ERR; } static int processBulkItem(redisReader *r) { @@ -328,15 +422,21 @@ static int processBulkItem(redisReader *r) { /* Proceed when obj was created. */ if (success) { + if (obj == NULL) { + __redisReaderSetErrorOOM(r); + return REDIS_ERR; + } + r->pos += bytelen; /* Set reply if this is the root object. */ if (r->ridx == 0) r->reply = obj; moveToNextTask(r); - return 0; + return REDIS_OK; } } - return -1; + + return REDIS_ERR; } static int processMultiBulkItem(redisReader *r) { @@ -346,11 +446,11 @@ static int processMultiBulkItem(redisReader *r) { long elements; int root = 0; - /* Set error for nested multi bulks with depth > 1 */ + /* Set error for nested multi bulks with depth > 2 */ if (r->ridx == 8) { - redisSetReplyReaderError(r,sdscatprintf(sdsempty(), - "No support for nested multi bulk replies with depth > 7")); - return -1; + __redisReaderSetError(r,REDIS_ERR_PROTOCOL, + "No support for nested multi bulk replies with depth > 7"); + return REDIS_ERR; } if ((p = readLine(r,NULL)) != NULL) { @@ -362,6 +462,12 @@ static int processMultiBulkItem(redisReader *r) { obj = r->fn->createNil(cur); else obj = (void*)REDIS_REPLY_NIL; + + if (obj == NULL) { + __redisReaderSetErrorOOM(r); + return REDIS_ERR; + } + moveToNextTask(r); } else { if (r->fn && r->fn->createArray) @@ -369,6 +475,11 @@ static int processMultiBulkItem(redisReader *r) { else obj = (void*)REDIS_REPLY_ARRAY; + if (obj == NULL) { + __redisReaderSetErrorOOM(r); + return REDIS_ERR; + } + /* Modify task stack when there are more than 0 elements. */ if (elements > 0) { cur->elements = elements; @@ -387,15 +498,15 @@ static int processMultiBulkItem(redisReader *r) { /* Set reply if this is the root object. */ if (root) r->reply = obj; - return 0; + return REDIS_OK; } - return -1; + + return REDIS_ERR; } static int processItem(redisReader *r) { redisReadTask *cur = &(r->rstack[r->ridx]); char *p; - sds byte; /* check if we need to read type */ if (cur->type < 0) { @@ -417,15 +528,12 @@ static int processItem(redisReader *r) { cur->type = REDIS_REPLY_ARRAY; break; default: - byte = sdscatrepr(sdsempty(),p,1); - redisSetReplyReaderError(r,sdscatprintf(sdsempty(), - "Protocol error, got %s as reply type byte", byte)); - sdsfree(byte); - return -1; + __redisReaderSetErrorProtocolByte(r,*p); + return REDIS_ERR; } } else { /* could not consume 1 byte */ - return -1; + return REDIS_ERR; } } @@ -441,82 +549,44 @@ static int processItem(redisReader *r) { return processMultiBulkItem(r); default: assert(NULL); - return -1; + return REDIS_ERR; /* Avoid warning. */ } } -void *redisReplyReaderCreate(void) { - redisReader *r = calloc(sizeof(redisReader),1); - r->error = NULL; - r->fn = &defaultFunctions; - r->buf = sdsempty(); - r->ridx = -1; - return r; -} +redisReader *redisReaderCreate(void) { + redisReader *r; -/* Set the function set to build the reply. Returns REDIS_OK when there - * is no temporary object and it can be set, REDIS_ERR otherwise. */ -int redisReplyReaderSetReplyObjectFunctions(void *reader, redisReplyObjectFunctions *fn) { - redisReader *r = reader; - if (r->reply == NULL) { - r->fn = fn; - return REDIS_OK; - } - return REDIS_ERR; -} + r = calloc(sizeof(redisReader),1); + if (r == NULL) + return NULL; -/* Set the private data field that is used in the read tasks. This argument can - * be used to curry arbitrary data to the custom reply object functions. */ -int redisReplyReaderSetPrivdata(void *reader, void *privdata) { - redisReader *r = reader; - if (r->reply == NULL) { - r->privdata = privdata; - return REDIS_OK; + r->err = 0; + r->errstr[0] = '\0'; + r->fn = &defaultFunctions; + r->buf = sdsempty(); + if (r->buf == NULL) { + free(r); + return NULL; } - return REDIS_ERR; -} -/* External libraries wrapping hiredis might need access to the temporary - * variable while the reply is built up. When the reader contains an - * object in between receiving some bytes to parse, this object might - * otherwise be free'd by garbage collection. */ -void *redisReplyReaderGetObject(void *reader) { - redisReader *r = reader; - return r->reply; + r->ridx = -1; + return r; } -void redisReplyReaderFree(void *reader) { - redisReader *r = reader; - if (r->error != NULL) - sdsfree(r->error); - if (r->reply != NULL && r->fn) +void redisReaderFree(redisReader *r) { + if (r->reply != NULL && r->fn && r->fn->freeObject) r->fn->freeObject(r->reply); if (r->buf != NULL) sdsfree(r->buf); free(r); } -static void redisSetReplyReaderError(redisReader *r, sds err) { - if (r->reply != NULL) - r->fn->freeObject(r->reply); - - /* Clear remaining buffer when we see a protocol error. */ - if (r->buf != NULL) { - sdsfree(r->buf); - r->buf = sdsempty(); - r->pos = r->len = 0; - } - r->ridx = -1; - r->error = err; -} - -char *redisReplyReaderGetError(void *reader) { - redisReader *r = reader; - return r->error; -} +int redisReaderFeed(redisReader *r, const char *buf, size_t len) { + sds newbuf; -void redisReplyReaderFeed(void *reader, const char *buf, size_t len) { - redisReader *r = reader; + /* Return early when this reader is in an erroneous state. */ + if (r->err) + return REDIS_ERR; /* Copy the provided buffer. */ if (buf != NULL && len >= 1) { @@ -526,16 +596,33 @@ void redisReplyReaderFeed(void *reader, const char *buf, size_t len) { sdsfree(r->buf); r->buf = sdsempty(); r->pos = 0; + + /* r->buf should not be NULL since we just free'd a larger one. */ + assert(r->buf != NULL); } #endif - r->buf = sdscatlen(r->buf,buf,len); + + newbuf = sdscatlen(r->buf,buf,len); + if (newbuf == NULL) { + __redisReaderSetErrorOOM(r); + return REDIS_ERR; + } + + r->buf = newbuf; r->len = sdslen(r->buf); } + + return REDIS_OK; } -int redisReplyReaderGetReply(void *reader, void **reply) { - redisReader *r = reader; - if (reply != NULL) *reply = NULL; +int redisReaderGetReply(redisReader *r, void **reply) { + /* Default target pointer to NULL. */ + if (reply != NULL) + *reply = NULL; + + /* Return early when this reader is in an erroneous state. */ + if (r->err) + return REDIS_ERR; /* When the buffer is empty, there will never be a reply. */ if (r->len == 0) @@ -554,9 +641,13 @@ int redisReplyReaderGetReply(void *reader, void **reply) { /* Process items in reply. */ while (r->ridx >= 0) - if (processItem(r) < 0) + if (processItem(r) != REDIS_OK) break; + /* Return ASAP when an error occurred. */ + if (r->err) + return REDIS_ERR; + /* Discard part of the buffer when we've consumed at least 1k, to avoid * doing unnecessary calls to memmove() in sds.c. */ if (r->pos >= 1024) { @@ -567,15 +658,9 @@ int redisReplyReaderGetReply(void *reader, void **reply) { /* Emit a reply when there is one. */ if (r->ridx == -1) { - void *aux = r->reply; + if (reply != NULL) + *reply = r->reply; r->reply = NULL; - - /* Check if there actually *is* a reply. */ - if (r->error != NULL) { - return REDIS_ERR; - } else { - if (reply != NULL) *reply = aux; - } } return REDIS_OK; } @@ -594,63 +679,79 @@ static int intlen(int i) { return len; } -/* Helper function for redisvFormatCommand(). */ -static void addArgument(sds a, char ***argv, int *argc, int *totlen) { - (*argc)++; - if ((*argv = realloc(*argv, sizeof(char*)*(*argc))) == NULL) redisOOM(); - if (totlen) *totlen = *totlen+1+intlen(sdslen(a))+2+sdslen(a)+2; - (*argv)[(*argc)-1] = a; +/* Helper that calculates the bulk length given a certain string length. */ +static size_t bulklen(size_t len) { + return 1+intlen(len)+2+len+2; } int redisvFormatCommand(char **target, const char *format, va_list ap) { - size_t size; - const char *arg, *c = format; + const char *c = format; char *cmd = NULL; /* final command */ int pos; /* position in final command */ - sds current; /* current argument */ + sds curarg, newarg; /* current argument */ int touched = 0; /* was the current argument touched? */ - char **argv = NULL; - int argc = 0, j; + char **curargv = NULL, **newargv = NULL; + int argc = 0; int totlen = 0; + int j; /* Abort if there is not target to set */ if (target == NULL) return -1; /* Build the command string accordingly to protocol */ - current = sdsempty(); + curarg = sdsempty(); + if (curarg == NULL) + return -1; + while(*c != '\0') { if (*c != '%' || c[1] == '\0') { if (*c == ' ') { if (touched) { - addArgument(current, &argv, &argc, &totlen); - current = sdsempty(); + newargv = realloc(curargv,sizeof(char*)*(argc+1)); + if (newargv == NULL) goto err; + curargv = newargv; + curargv[argc++] = curarg; + totlen += bulklen(sdslen(curarg)); + + /* curarg is put in argv so it can be overwritten. */ + curarg = sdsempty(); + if (curarg == NULL) goto err; touched = 0; } } else { - current = sdscatlen(current,c,1); + newarg = sdscatlen(curarg,c,1); + if (newarg == NULL) goto err; + curarg = newarg; touched = 1; } } else { + char *arg; + size_t size; + + /* Set newarg so it can be checked even if it is not touched. */ + newarg = curarg; + switch(c[1]) { case 's': arg = va_arg(ap,char*); size = strlen(arg); if (size > 0) - current = sdscatlen(current,arg,size); + newarg = sdscatlen(curarg,arg,size); break; case 'b': arg = va_arg(ap,char*); size = va_arg(ap,size_t); if (size > 0) - current = sdscatlen(current,arg,size); + newarg = sdscatlen(curarg,arg,size); break; case '%': - current = sdscat(current,"%"); + newarg = sdscat(curarg,"%"); break; default: /* Try to detect printf format */ { + static const char intfmts[] = "diouxX"; char _format[16]; const char *_p = c+1; size_t _l = 0; @@ -672,35 +773,85 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) { while (*_p != '\0' && isdigit(*_p)) _p++; } - /* Modifiers */ - if (*_p != '\0') { - if (*_p == 'h' || *_p == 'l') { - /* Allow a single repetition for these modifiers */ - if (_p[0] == _p[1]) _p++; - _p++; + /* Copy va_list before consuming with va_arg */ + va_copy(_cpy,ap); + + /* Integer conversion (without modifiers) */ + if (strchr(intfmts,*_p) != NULL) { + va_arg(ap,int); + goto fmt_valid; + } + + /* Double conversion (without modifiers) */ + if (strchr("eEfFgGaA",*_p) != NULL) { + va_arg(ap,double); + goto fmt_valid; + } + + /* Size: char */ + if (_p[0] == 'h' && _p[1] == 'h') { + _p += 2; + if (*_p != '\0' && strchr(intfmts,*_p) != NULL) { + va_arg(ap,int); /* char gets promoted to int */ + goto fmt_valid; } + goto fmt_invalid; } - /* Conversion specifier */ - if (*_p != '\0' && strchr("diouxXeEfFgGaA",*_p) != NULL) { - _l = (_p+1)-c; - if (_l < sizeof(_format)-2) { - memcpy(_format,c,_l); - _format[_l] = '\0'; - va_copy(_cpy,ap); - current = sdscatvprintf(current,_format,_cpy); - va_end(_cpy); - - /* Update current position (note: outer blocks - * increment c twice so compensate here) */ - c = _p-1; + /* Size: short */ + if (_p[0] == 'h') { + _p += 1; + if (*_p != '\0' && strchr(intfmts,*_p) != NULL) { + va_arg(ap,int); /* short gets promoted to int */ + goto fmt_valid; } + goto fmt_invalid; } - /* Consume and discard vararg */ - va_arg(ap,void); + /* Size: long long */ + if (_p[0] == 'l' && _p[1] == 'l') { + _p += 2; + if (*_p != '\0' && strchr(intfmts,*_p) != NULL) { + va_arg(ap,long long); + goto fmt_valid; + } + goto fmt_invalid; + } + + /* Size: long */ + if (_p[0] == 'l') { + _p += 1; + if (*_p != '\0' && strchr(intfmts,*_p) != NULL) { + va_arg(ap,long); + goto fmt_valid; + } + goto fmt_invalid; + } + + fmt_invalid: + va_end(_cpy); + goto err; + + fmt_valid: + _l = (_p+1)-c; + if (_l < sizeof(_format)-2) { + memcpy(_format,c,_l); + _format[_l] = '\0'; + newarg = sdscatvprintf(curarg,_format,_cpy); + + /* Update current position (note: outer blocks + * increment c twice so compensate here) */ + c = _p-1; + } + + va_end(_cpy); + break; } } + + if (newarg == NULL) goto err; + curarg = newarg; + touched = 1; c++; } @@ -709,31 +860,55 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) { /* Add the last argument if needed */ if (touched) { - addArgument(current, &argv, &argc, &totlen); + newargv = realloc(curargv,sizeof(char*)*(argc+1)); + if (newargv == NULL) goto err; + curargv = newargv; + curargv[argc++] = curarg; + totlen += bulklen(sdslen(curarg)); } else { - sdsfree(current); + sdsfree(curarg); } + /* Clear curarg because it was put in curargv or was free'd. */ + curarg = NULL; + /* Add bytes needed to hold multi bulk count */ totlen += 1+intlen(argc)+2; /* Build the command at protocol level */ cmd = malloc(totlen+1); - if (!cmd) redisOOM(); + if (cmd == NULL) goto err; + pos = sprintf(cmd,"*%d\r\n",argc); for (j = 0; j < argc; j++) { - pos += sprintf(cmd+pos,"$%zu\r\n",sdslen(argv[j])); - memcpy(cmd+pos,argv[j],sdslen(argv[j])); - pos += sdslen(argv[j]); - sdsfree(argv[j]); + pos += sprintf(cmd+pos,"$%zu\r\n",sdslen(curargv[j])); + memcpy(cmd+pos,curargv[j],sdslen(curargv[j])); + pos += sdslen(curargv[j]); + sdsfree(curargv[j]); cmd[pos++] = '\r'; cmd[pos++] = '\n'; } assert(pos == totlen); - free(argv); - cmd[totlen] = '\0'; + cmd[pos] = '\0'; + + free(curargv); *target = cmd; return totlen; + +err: + while(argc--) + sdsfree(curargv[argc]); + free(curargv); + + if (curarg != NULL) + sdsfree(curarg); + + /* No need to check cmd since it is the last statement that can fail, + * but do it anyway to be as defensive as possible. */ + if (cmd != NULL) + free(cmd); + + return -1; } /* Format a command according to the Redis protocol. This function @@ -772,12 +947,14 @@ int redisFormatCommandArgv(char **target, int argc, const char **argv, const siz totlen = 1+intlen(argc)+2; for (j = 0; j < argc; j++) { len = argvlen ? argvlen[j] : strlen(argv[j]); - totlen += 1+intlen(len)+2+len+2; + totlen += bulklen(len); } /* Build the command at protocol level */ cmd = malloc(totlen+1); - if (!cmd) redisOOM(); + if (cmd == NULL) + return -1; + pos = sprintf(cmd,"*%d\r\n",argc); for (j = 0; j < argc; j++) { len = argvlen ? argvlen[j] : strlen(argv[j]); @@ -788,41 +965,49 @@ int redisFormatCommandArgv(char **target, int argc, const char **argv, const siz cmd[pos++] = '\n'; } assert(pos == totlen); - cmd[totlen] = '\0'; + cmd[pos] = '\0'; + *target = cmd; return totlen; } -void __redisSetError(redisContext *c, int type, const sds errstr) { +void __redisSetError(redisContext *c, int type, const char *str) { + size_t len; + c->err = type; - if (errstr != NULL) { - c->errstr = errstr; + if (str != NULL) { + len = strlen(str); + len = len < (sizeof(c->errstr)-1) ? len : (sizeof(c->errstr)-1); + memcpy(c->errstr,str,len); + c->errstr[len] = '\0'; } else { /* Only REDIS_ERR_IO may lack a description! */ assert(type == REDIS_ERR_IO); - c->errstr = sdsnew(strerror(errno)); + strerror_r(errno,c->errstr,sizeof(c->errstr)); } } static redisContext *redisContextInit(void) { - redisContext *c = calloc(sizeof(redisContext),1); + redisContext *c; + + c = calloc(1,sizeof(redisContext)); + if (c == NULL) + return NULL; + c->err = 0; - c->errstr = NULL; + c->errstr[0] = '\0'; c->obuf = sdsempty(); - c->fn = &defaultFunctions; - c->reader = NULL; + c->reader = redisReaderCreate(); return c; } void redisFree(redisContext *c) { if (c->fd > 0) close(c->fd); - if (c->errstr != NULL) - sdsfree(c->errstr); if (c->obuf != NULL) sdsfree(c->obuf); if (c->reader != NULL) - redisReplyReaderFree(c->reader); + redisReaderFree(c->reader); free(c); } @@ -878,24 +1063,6 @@ int redisSetTimeout(redisContext *c, struct timeval tv) { return REDIS_ERR; } -/* Set the replyObjectFunctions to use. Returns REDIS_ERR when the reader - * was already initialized and the function set could not be re-set. - * Return REDIS_OK when they could be set. */ -int redisSetReplyObjectFunctions(redisContext *c, redisReplyObjectFunctions *fn) { - if (c->reader != NULL) - return REDIS_ERR; - c->fn = fn; - return REDIS_OK; -} - -/* Helper function to lazily create a reply reader. */ -static void __redisCreateReplyReader(redisContext *c) { - if (c->reader == NULL) { - c->reader = redisReplyReaderCreate(); - assert(redisReplyReaderSetReplyObjectFunctions(c->reader,c->fn) == REDIS_OK); - } -} - /* Use this function to handle a read event on the descriptor. It will try * and read some bytes from the socket and feed them to the reply parser. * @@ -903,7 +1070,13 @@ static void __redisCreateReplyReader(redisContext *c) { * see if there is a reply available. */ int redisBufferRead(redisContext *c) { char buf[1024*16]; - int nread = read(c->fd,buf,sizeof(buf)); + int nread; + + /* Return early when the context has seen an error. */ + if (c->err) + return REDIS_ERR; + + nread = read(c->fd,buf,sizeof(buf)); if (nread == -1) { if (errno == EAGAIN && !(c->flags & REDIS_BLOCK)) { /* Try again later */ @@ -912,12 +1085,13 @@ int redisBufferRead(redisContext *c) { return REDIS_ERR; } } else if (nread == 0) { - __redisSetError(c,REDIS_ERR_EOF, - sdsnew("Server closed the connection")); + __redisSetError(c,REDIS_ERR_EOF,"Server closed the connection"); return REDIS_ERR; } else { - __redisCreateReplyReader(c); - redisReplyReaderFeed(c->reader,buf,nread); + if (redisReaderFeed(c->reader,buf,nread) != REDIS_OK) { + __redisSetError(c,c->reader->err,c->reader->errstr); + return REDIS_ERR; + } } return REDIS_OK; } @@ -926,13 +1100,18 @@ int redisBufferRead(redisContext *c) { * * Returns REDIS_OK when the buffer is empty, or (a part of) the buffer was * succesfully written to the socket. When the buffer is empty after the - * write operation, "wdone" is set to 1 (if given). + * write operation, "done" is set to 1 (if given). * * Returns REDIS_ERR if an error occured trying to write and sets - * c->error to hold the appropriate error string. + * c->errstr to hold the appropriate error string. */ int redisBufferWrite(redisContext *c, int *done) { int nwritten; + + /* Return early when the context has seen an error. */ + if (c->err) + return REDIS_ERR; + if (sdslen(c->obuf) > 0) { nwritten = write(c->fd,c->obuf,sdslen(c->obuf)); if (nwritten == -1) { @@ -958,10 +1137,8 @@ int redisBufferWrite(redisContext *c, int *done) { /* Internal helper function to try and get a reply from the reader, * or set an error in the context otherwise. */ int redisGetReplyFromReader(redisContext *c, void **reply) { - __redisCreateReplyReader(c); - if (redisReplyReaderGetReply(c->reader,reply) == REDIS_ERR) { - __redisSetError(c,REDIS_ERR_PROTOCOL, - sdsnew(((redisReader*)c->reader)->error)); + if (redisReaderGetReply(c->reader,reply) == REDIS_ERR) { + __redisSetError(c,c->reader->err,c->reader->errstr); return REDIS_ERR; } return REDIS_OK; @@ -1004,31 +1181,65 @@ int redisGetReply(redisContext *c, void **reply) { * is used, you need to call redisGetReply yourself to retrieve * the reply (or replies in pub/sub). */ -void __redisAppendCommand(redisContext *c, char *cmd, size_t len) { - c->obuf = sdscatlen(c->obuf,cmd,len); +int __redisAppendCommand(redisContext *c, char *cmd, size_t len) { + sds newbuf; + + newbuf = sdscatlen(c->obuf,cmd,len); + if (newbuf == NULL) { + __redisSetError(c,REDIS_ERR_OOM,"Out of memory"); + return REDIS_ERR; + } + + c->obuf = newbuf; + return REDIS_OK; } -void redisvAppendCommand(redisContext *c, const char *format, va_list ap) { +int redisvAppendCommand(redisContext *c, const char *format, va_list ap) { char *cmd; int len; + len = redisvFormatCommand(&cmd,format,ap); - __redisAppendCommand(c,cmd,len); + if (len == -1) { + __redisSetError(c,REDIS_ERR_OOM,"Out of memory"); + return REDIS_ERR; + } + + if (__redisAppendCommand(c,cmd,len) != REDIS_OK) { + free(cmd); + return REDIS_ERR; + } + free(cmd); + return REDIS_OK; } -void redisAppendCommand(redisContext *c, const char *format, ...) { +int redisAppendCommand(redisContext *c, const char *format, ...) { va_list ap; + int ret; + va_start(ap,format); - redisvAppendCommand(c,format,ap); + ret = redisvAppendCommand(c,format,ap); va_end(ap); + return ret; } -void redisAppendCommandArgv(redisContext *c, int argc, const char **argv, const size_t *argvlen) { +int redisAppendCommandArgv(redisContext *c, int argc, const char **argv, const size_t *argvlen) { char *cmd; int len; + len = redisFormatCommandArgv(&cmd,argc,argv,argvlen); - __redisAppendCommand(c,cmd,len); + if (len == -1) { + __redisSetError(c,REDIS_ERR_OOM,"Out of memory"); + return REDIS_ERR; + } + + if (__redisAppendCommand(c,cmd,len) != REDIS_OK) { + free(cmd); + return REDIS_ERR; + } + free(cmd); + return REDIS_OK; } /* Helper function for the redisCommand* family of functions. @@ -1042,26 +1253,21 @@ void redisAppendCommandArgv(redisContext *c, int argc, const char **argv, const * otherwise. When NULL is returned in a blocking context, the error field * in the context will be set. */ -static void *__redisCommand(redisContext *c, char *cmd, size_t len) { - void *aux = NULL; - __redisAppendCommand(c,cmd,len); +static void *__redisBlockForReply(redisContext *c) { + void *reply; if (c->flags & REDIS_BLOCK) { - if (redisGetReply(c,&aux) == REDIS_OK) - return aux; - return NULL; + if (redisGetReply(c,&reply) != REDIS_OK) + return NULL; + return reply; } return NULL; } void *redisvCommand(redisContext *c, const char *format, va_list ap) { - char *cmd; - int len; - void *reply = NULL; - len = redisvFormatCommand(&cmd,format,ap); - reply = __redisCommand(c,cmd,len); - free(cmd); - return reply; + if (redisvAppendCommand(c,format,ap) != REDIS_OK) + return NULL; + return __redisBlockForReply(c); } void *redisCommand(redisContext *c, const char *format, ...) { @@ -1074,11 +1280,7 @@ void *redisCommand(redisContext *c, const char *format, ...) { } void *redisCommandArgv(redisContext *c, int argc, const char **argv, const size_t *argvlen) { - char *cmd; - int len; - void *reply = NULL; - len = redisFormatCommandArgv(&cmd,argc,argv,argvlen); - reply = __redisCommand(c,cmd,len); - free(cmd); - return reply; + if (redisAppendCommandArgv(c,argc,argv,argvlen) != REDIS_OK) + return NULL; + return __redisBlockForReply(c); } diff --git a/deps/hiredis/hiredis.h b/deps/hiredis/hiredis.h index f4452091..a73f50e9 100644 --- a/deps/hiredis/hiredis.h +++ b/deps/hiredis/hiredis.h @@ -1,6 +1,6 @@ /* - * Copyright (c) 2009-2010, Salvatore Sanfilippo - * Copyright (c) 2010, Pieter Noordhuis + * Copyright (c) 2009-2011, Salvatore Sanfilippo + * Copyright (c) 2010-2011, Pieter Noordhuis * * All rights reserved. * @@ -36,8 +36,8 @@ #include /* for struct timeval */ #define HIREDIS_MAJOR 0 -#define HIREDIS_MINOR 9 -#define HIREDIS_PATCH 2 +#define HIREDIS_MINOR 10 +#define HIREDIS_PATCH 1 #define REDIS_ERR -1 #define REDIS_OK 0 @@ -46,10 +46,11 @@ * error that occured. REDIS_ERR_IO means there was an I/O error and you * should use the "errno" variable to find out what is wrong. * For other values, the "errstr" field will hold a description. */ -#define REDIS_ERR_IO 1 /* error in read or write */ -#define REDIS_ERR_EOF 3 /* eof */ -#define REDIS_ERR_PROTOCOL 4 /* protocol error */ -#define REDIS_ERR_OTHER 2 /* something else */ +#define REDIS_ERR_IO 1 /* Error in read or write */ +#define REDIS_ERR_EOF 3 /* End of file */ +#define REDIS_ERR_PROTOCOL 4 /* Protocol error */ +#define REDIS_ERR_OOM 5 /* Out of memory */ +#define REDIS_ERR_OTHER 2 /* Everything else... */ /* Connection type can be blocking or non-blocking and is set in the * least significant bit of the flags field in redisContext. */ @@ -113,36 +114,56 @@ typedef struct redisReplyObjectFunctions { void (*freeObject)(void*); } redisReplyObjectFunctions; -struct redisContext; /* need forward declaration of redisContext */ - -/* Context for a connection to Redis */ -typedef struct redisContext { - int fd; - int flags; - char *obuf; /* Write buffer */ +/* State for the protocol parser */ +typedef struct redisReader { int err; /* Error flags, 0 when there is no error */ - char *errstr; /* String representation of error when applicable */ + char errstr[128]; /* String representation of error when applicable */ - /* Function set for reply buildup and reply reader */ - redisReplyObjectFunctions *fn; - void *reader; -} redisContext; + char *buf; /* Read buffer */ + size_t pos; /* Buffer cursor */ + size_t len; /* Buffer length */ + + redisReadTask rstack[9]; + int ridx; /* Index of current read task */ + void *reply; /* Temporary reply pointer */ + redisReplyObjectFunctions *fn; + void *privdata; +} redisReader; + +/* Public API for the protocol parser. */ +redisReader *redisReaderCreate(void); +void redisReaderFree(redisReader *r); +int redisReaderFeed(redisReader *r, const char *buf, size_t len); +int redisReaderGetReply(redisReader *r, void **reply); + +/* Backwards compatibility, can be removed on big version bump. */ +#define redisReplyReaderCreate redisReaderCreate +#define redisReplyReaderFree redisReaderFree +#define redisReplyReaderFeed redisReaderFeed +#define redisReplyReaderGetReply redisReaderGetReply +#define redisReplyReaderSetPrivdata(_r, _p) (int)(((redisReader*)(_r))->privdata = (_p)) +#define redisReplyReaderGetObject(_r) (((redisReader*)(_r))->reply) +#define redisReplyReaderGetError(_r) (((redisReader*)(_r))->errstr) + +/* Function to free the reply objects hiredis returns by default. */ void freeReplyObject(void *reply); -void *redisReplyReaderCreate(void); -int redisReplyReaderSetReplyObjectFunctions(void *reader, redisReplyObjectFunctions *fn); -int redisReplyReaderSetPrivdata(void *reader, void *privdata); -void *redisReplyReaderGetObject(void *reader); -char *redisReplyReaderGetError(void *reader); -void redisReplyReaderFree(void *ptr); -void redisReplyReaderFeed(void *reader, const char *buf, size_t len); -int redisReplyReaderGetReply(void *reader, void **reply); /* Functions to format a command according to the protocol. */ int redisvFormatCommand(char **target, const char *format, va_list ap); int redisFormatCommand(char **target, const char *format, ...); int redisFormatCommandArgv(char **target, int argc, const char **argv, const size_t *argvlen); +/* Context for a connection to Redis */ +typedef struct redisContext { + int err; /* Error flags, 0 when there is no error */ + char errstr[128]; /* String representation of error when applicable */ + int fd; + int flags; + char *obuf; /* Write buffer */ + redisReader *reader; /* Protocol reader */ +} redisContext; + redisContext *redisConnect(const char *ip, int port); redisContext *redisConnectWithTimeout(const char *ip, int port, struct timeval tv); redisContext *redisConnectNonBlock(const char *ip, int port); @@ -150,7 +171,6 @@ redisContext *redisConnectUnix(const char *path); redisContext *redisConnectUnixWithTimeout(const char *path, struct timeval tv); redisContext *redisConnectUnixNonBlock(const char *path); int redisSetTimeout(redisContext *c, struct timeval tv); -int redisSetReplyObjectFunctions(redisContext *c, redisReplyObjectFunctions *fn); void redisFree(redisContext *c); int redisBufferRead(redisContext *c); int redisBufferWrite(redisContext *c, int *done); @@ -164,9 +184,9 @@ int redisGetReplyFromReader(redisContext *c, void **reply); /* Write a command to the output buffer. Use these functions in blocking mode * to get a pipeline of commands. */ -void redisvAppendCommand(redisContext *c, const char *format, va_list ap); -void redisAppendCommand(redisContext *c, const char *format, ...); -void redisAppendCommandArgv(redisContext *c, int argc, const char **argv, const size_t *argvlen); +int redisvAppendCommand(redisContext *c, const char *format, va_list ap); +int redisAppendCommand(redisContext *c, const char *format, ...); +int redisAppendCommandArgv(redisContext *c, int argc, const char **argv, const size_t *argvlen); /* Issue a command to Redis. In a blocking context, it is identical to calling * redisAppendCommand, followed by redisGetReply. The function will return diff --git a/deps/hiredis/net.c b/deps/hiredis/net.c index 438a129b..158e1dd8 100644 --- a/deps/hiredis/net.c +++ b/deps/hiredis/net.c @@ -1,7 +1,7 @@ /* Extracted from anet.c to work properly with Hiredis error reporting. * - * Copyright (c) 2006-2010, Salvatore Sanfilippo - * Copyright (c) 2010, Pieter Noordhuis + * Copyright (c) 2006-2011, Salvatore Sanfilippo + * Copyright (c) 2010-2011, Pieter Noordhuis * * All rights reserved. * @@ -49,19 +49,37 @@ #include "net.h" #include "sds.h" -/* Forward declaration */ -void __redisSetError(redisContext *c, int type, sds err); +/* Defined in hiredis.c */ +void __redisSetError(redisContext *c, int type, const char *str); + +static void __redisSetErrorFromErrno(redisContext *c, int type, const char *prefix) { + char buf[128]; + size_t len = 0; + + if (prefix != NULL) + len = snprintf(buf,sizeof(buf),"%s: ",prefix); + strerror_r(errno,buf+len,sizeof(buf)-len); + __redisSetError(c,type,buf); +} + +static int redisSetReuseAddr(redisContext *c, int fd) { + int on = 1; + if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)) == -1) { + __redisSetErrorFromErrno(c,REDIS_ERR_IO,NULL); + close(fd); + return REDIS_ERR; + } + return REDIS_OK; +} static int redisCreateSocket(redisContext *c, int type) { - int s, on = 1; + int s; if ((s = socket(type, SOCK_STREAM, 0)) == -1) { - __redisSetError(c,REDIS_ERR_IO,NULL); + __redisSetErrorFromErrno(c,REDIS_ERR_IO,NULL); return REDIS_ERR; } if (type == AF_INET) { - if (setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)) == -1) { - __redisSetError(c,REDIS_ERR_IO,NULL); - close(s); + if (redisSetReuseAddr(c,s) == REDIS_ERR) { return REDIS_ERR; } } @@ -75,8 +93,7 @@ static int redisSetBlocking(redisContext *c, int fd, int blocking) { * Note that fcntl(2) for F_GETFL and F_SETFL can't be * interrupted by a signal. */ if ((flags = fcntl(fd, F_GETFL)) == -1) { - __redisSetError(c,REDIS_ERR_IO, - sdscatprintf(sdsempty(), "fcntl(F_GETFL): %s", strerror(errno))); + __redisSetErrorFromErrno(c,REDIS_ERR_IO,"fcntl(F_GETFL)"); close(fd); return REDIS_ERR; } @@ -87,8 +104,7 @@ static int redisSetBlocking(redisContext *c, int fd, int blocking) { flags |= O_NONBLOCK; if (fcntl(fd, F_SETFL, flags) == -1) { - __redisSetError(c,REDIS_ERR_IO, - sdscatprintf(sdsempty(), "fcntl(F_SETFL): %s", strerror(errno))); + __redisSetErrorFromErrno(c,REDIS_ERR_IO,"fcntl(F_SETFL)"); close(fd); return REDIS_ERR; } @@ -98,8 +114,7 @@ static int redisSetBlocking(redisContext *c, int fd, int blocking) { static int redisSetTcpNoDelay(redisContext *c, int fd) { int yes = 1; if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes)) == -1) { - __redisSetError(c,REDIS_ERR_IO, - sdscatprintf(sdsempty(), "setsockopt(TCP_NODELAY): %s", strerror(errno))); + __redisSetErrorFromErrno(c,REDIS_ERR_IO,"setsockopt(TCP_NODELAY)"); close(fd); return REDIS_ERR; } @@ -110,8 +125,6 @@ static int redisContextWaitReady(redisContext *c, int fd, const struct timeval * struct timeval to; struct timeval *toptr = NULL; fd_set wfd; - int err; - socklen_t errlen; /* Only use timeout when not NULL. */ if (timeout != NULL) { @@ -124,101 +137,115 @@ static int redisContextWaitReady(redisContext *c, int fd, const struct timeval * FD_SET(fd, &wfd); if (select(FD_SETSIZE, NULL, &wfd, NULL, toptr) == -1) { - __redisSetError(c,REDIS_ERR_IO, - sdscatprintf(sdsempty(), "select(2): %s", strerror(errno))); + __redisSetErrorFromErrno(c,REDIS_ERR_IO,"select(2)"); close(fd); return REDIS_ERR; } if (!FD_ISSET(fd, &wfd)) { errno = ETIMEDOUT; - __redisSetError(c,REDIS_ERR_IO,NULL); + __redisSetErrorFromErrno(c,REDIS_ERR_IO,NULL); close(fd); return REDIS_ERR; } - err = 0; - errlen = sizeof(err); - if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &errlen) == -1) { - __redisSetError(c,REDIS_ERR_IO, - sdscatprintf(sdsempty(), "getsockopt(SO_ERROR): %s", strerror(errno))); - close(fd); + if (redisCheckSocketError(c, fd) != REDIS_OK) return REDIS_ERR; - } - - if (err) { - errno = err; - __redisSetError(c,REDIS_ERR_IO,NULL); - close(fd); - return REDIS_ERR; - } return REDIS_OK; } - __redisSetError(c,REDIS_ERR_IO,NULL); + __redisSetErrorFromErrno(c,REDIS_ERR_IO,NULL); close(fd); return REDIS_ERR; } +int redisCheckSocketError(redisContext *c, int fd) { + int err = 0; + socklen_t errlen = sizeof(err); + + if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &errlen) == -1) { + __redisSetErrorFromErrno(c,REDIS_ERR_IO,"getsockopt(SO_ERROR)"); + close(fd); + return REDIS_ERR; + } + + if (err) { + errno = err; + __redisSetErrorFromErrno(c,REDIS_ERR_IO,NULL); + close(fd); + return REDIS_ERR; + } + + return REDIS_OK; +} + int redisContextSetTimeout(redisContext *c, struct timeval tv) { if (setsockopt(c->fd,SOL_SOCKET,SO_RCVTIMEO,&tv,sizeof(tv)) == -1) { - __redisSetError(c,REDIS_ERR_IO, - sdscatprintf(sdsempty(), "setsockopt(SO_RCVTIMEO): %s", strerror(errno))); + __redisSetErrorFromErrno(c,REDIS_ERR_IO,"setsockopt(SO_RCVTIMEO)"); return REDIS_ERR; } if (setsockopt(c->fd,SOL_SOCKET,SO_SNDTIMEO,&tv,sizeof(tv)) == -1) { - __redisSetError(c,REDIS_ERR_IO, - sdscatprintf(sdsempty(), "setsockopt(SO_SNDTIMEO): %s", strerror(errno))); + __redisSetErrorFromErrno(c,REDIS_ERR_IO,"setsockopt(SO_SNDTIMEO)"); return REDIS_ERR; } return REDIS_OK; } int redisContextConnectTcp(redisContext *c, const char *addr, int port, struct timeval *timeout) { - int s; + int s, rv; + char _port[6]; /* strlen("65535"); */ + struct addrinfo hints, *servinfo, *p; int blocking = (c->flags & REDIS_BLOCK); - struct sockaddr_in sa; - if ((s = redisCreateSocket(c,AF_INET)) < 0) - return REDIS_ERR; - if (redisSetBlocking(c,s,0) != REDIS_OK) - return REDIS_ERR; - - sa.sin_family = AF_INET; - sa.sin_port = htons(port); - if (inet_aton(addr, &sa.sin_addr) == 0) { - struct hostent *he; + snprintf(_port, 6, "%d", port); + memset(&hints,0,sizeof(hints)); + hints.ai_family = AF_INET; + hints.ai_socktype = SOCK_STREAM; - he = gethostbyname(addr); - if (he == NULL) { - __redisSetError(c,REDIS_ERR_OTHER, - sdscatprintf(sdsempty(),"Can't resolve: %s",addr)); - close(s); - return REDIS_ERR; - } - memcpy(&sa.sin_addr, he->h_addr, sizeof(struct in_addr)); + if ((rv = getaddrinfo(addr,_port,&hints,&servinfo)) != 0) { + __redisSetError(c,REDIS_ERR_OTHER,gai_strerror(rv)); + return REDIS_ERR; } - - if (connect(s, (struct sockaddr*)&sa, sizeof(sa)) == -1) { - if (errno == EINPROGRESS && !blocking) { - /* This is ok. */ - } else { - if (redisContextWaitReady(c,s,timeout) != REDIS_OK) - return REDIS_ERR; + for (p = servinfo; p != NULL; p = p->ai_next) { + if ((s = socket(p->ai_family,p->ai_socktype,p->ai_protocol)) == -1) + continue; + + if (redisSetBlocking(c,s,0) != REDIS_OK) + goto error; + if (connect(s,p->ai_addr,p->ai_addrlen) == -1) { + if (errno == EHOSTUNREACH) { + close(s); + continue; + } else if (errno == EINPROGRESS && !blocking) { + /* This is ok. */ + } else { + if (redisContextWaitReady(c,s,timeout) != REDIS_OK) + goto error; + } } + if (blocking && redisSetBlocking(c,s,1) != REDIS_OK) + goto error; + if (redisSetTcpNoDelay(c,s) != REDIS_OK) + goto error; + + c->fd = s; + c->flags |= REDIS_CONNECTED; + rv = REDIS_OK; + goto end; + } + if (p == NULL) { + char buf[128]; + snprintf(buf,sizeof(buf),"Can't create socket: %s",strerror(errno)); + __redisSetError(c,REDIS_ERR_OTHER,buf); + goto error; } - /* Reset socket to be blocking after connect(2). */ - if (blocking && redisSetBlocking(c,s,1) != REDIS_OK) - return REDIS_ERR; - - if (redisSetTcpNoDelay(c,s) != REDIS_OK) - return REDIS_ERR; - - c->fd = s; - c->flags |= REDIS_CONNECTED; - return REDIS_OK; +error: + rv = REDIS_ERR; +end: + freeaddrinfo(servinfo); + return rv; // Need to return REDIS_OK if alright } int redisContextConnectUnix(redisContext *c, const char *path, struct timeval *timeout) { diff --git a/deps/hiredis/net.h b/deps/hiredis/net.h index e149ceba..eb8a0a1c 100644 --- a/deps/hiredis/net.h +++ b/deps/hiredis/net.h @@ -1,7 +1,7 @@ /* Extracted from anet.c to work properly with Hiredis error reporting. * - * Copyright (c) 2006-2010, Salvatore Sanfilippo - * Copyright (c) 2010, Pieter Noordhuis + * Copyright (c) 2006-2011, Salvatore Sanfilippo + * Copyright (c) 2010-2011, Pieter Noordhuis * * All rights reserved. * @@ -39,6 +39,7 @@ #define AF_LOCAL AF_UNIX #endif +int redisCheckSocketError(redisContext *c, int fd); int redisContextSetTimeout(redisContext *c, struct timeval tv); int redisContextConnectTcp(redisContext *c, const char *addr, int port, struct timeval *timeout); int redisContextConnectUnix(redisContext *c, const char *path, struct timeval *timeout); diff --git a/deps/hiredis/sds.c b/deps/hiredis/sds.c index 2bd5a81c..0af9c672 100644 --- a/deps/hiredis/sds.c +++ b/deps/hiredis/sds.c @@ -28,18 +28,18 @@ * POSSIBILITY OF SUCH DAMAGE. */ -#define SDS_ABORT_ON_OOM - #include #include #include #include #include "sds.h" +#ifdef SDS_ABORT_ON_OOM static void sdsOomAbort(void) { fprintf(stderr,"SDS: Out Of Memory (SDS_ABORT_ON_OOM defined)\n"); abort(); } +#endif sds sdsnewlen(const void *init, size_t initlen) { struct sdshdr *sh; @@ -378,17 +378,19 @@ sds sdsfromlonglong(long long value) { sds sdscatrepr(sds s, char *p, size_t len) { s = sdscatlen(s,"\"",1); + if (s == NULL) return NULL; + while(len--) { switch(*p) { case '\\': case '"': s = sdscatprintf(s,"\\%c",*p); break; - case '\n': s = sdscatlen(s,"\\n",1); break; - case '\r': s = sdscatlen(s,"\\r",1); break; - case '\t': s = sdscatlen(s,"\\t",1); break; - case '\a': s = sdscatlen(s,"\\a",1); break; - case '\b': s = sdscatlen(s,"\\b",1); break; + case '\n': s = sdscatlen(s,"\\n",2); break; + case '\r': s = sdscatlen(s,"\\r",2); break; + case '\t': s = sdscatlen(s,"\\t",2); break; + case '\a': s = sdscatlen(s,"\\a",2); break; + case '\b': s = sdscatlen(s,"\\b",2); break; default: if (isprint(*p)) s = sdscatprintf(s,"%c",*p); @@ -397,6 +399,7 @@ sds sdscatrepr(sds s, char *p, size_t len) { break; } p++; + if (s == NULL) return NULL; } return sdscatlen(s,"\"",1); } @@ -416,7 +419,7 @@ sds sdscatrepr(sds s, char *p, size_t len) { sds *sdssplitargs(char *line, int *argc) { char *p = line; char *current = NULL; - char **vector = NULL; + char **vector = NULL, **_vector = NULL; *argc = 0; while(1) { @@ -427,7 +430,11 @@ sds *sdssplitargs(char *line, int *argc) { int inq=0; /* set to 1 if we are in "quotes" */ int done=0; - if (current == NULL) current = sdsempty(); + if (current == NULL) { + current = sdsempty(); + if (current == NULL) goto err; + } + while(!done) { if (inq) { if (*p == '\\' && *(p+1)) { @@ -471,9 +478,13 @@ sds *sdssplitargs(char *line, int *argc) { } } if (*p) p++; + if (current == NULL) goto err; } /* add the token to the vector */ - vector = realloc(vector,((*argc)+1)*sizeof(char*)); + _vector = realloc(vector,((*argc)+1)*sizeof(char*)); + if (_vector == NULL) goto err; + + vector = _vector; vector[*argc] = current; (*argc)++; current = NULL; @@ -485,8 +496,8 @@ sds *sdssplitargs(char *line, int *argc) { err: while((*argc)--) sdsfree(vector[*argc]); - free(vector); - if (current) sdsfree(current); + if (vector != NULL) free(vector); + if (current != NULL) sdsfree(current); return NULL; } diff --git a/deps/hiredis/test.c b/deps/hiredis/test.c index 5724a3ea..5945b655 100644 --- a/deps/hiredis/test.c +++ b/deps/hiredis/test.c @@ -1,3 +1,4 @@ +#include "fmacros.h" #include #include #include @@ -10,10 +11,28 @@ #include "hiredis.h" +enum connection_type { + CONN_TCP, + CONN_UNIX +}; + +struct config { + enum connection_type type; + + struct { + const char *host; + int port; + } tcp; + + struct { + const char *path; + } unix; +}; + /* The following lines make up our testing "framework" :) */ static int tests = 0, fails = 0; #define test(_s) { printf("#%02d ", ++tests); printf(_s); } -#define test_cond(_c) if(_c) printf("PASSED\n"); else {printf("FAILED\n"); fails++;} +#define test_cond(_c) if(_c) printf("\033[0;32mPASSED\033[0;0m\n"); else {printf("\033[0;31mFAILED\033[0;0m\n"); fails++;} static long long usec(void) { struct timeval tv; @@ -21,15 +40,60 @@ static long long usec(void) { return (((long long)tv.tv_sec)*1000000)+tv.tv_usec; } -static int use_unix = 0; -static redisContext *blocking_context = NULL; -static void __connect(redisContext **target) { - *target = blocking_context = (use_unix ? - redisConnectUnix("/tmp/redis.sock") : redisConnect((char*)"127.0.0.1", 6379)); - if (blocking_context->err) { - printf("Connection error: %s\n", blocking_context->errstr); +static redisContext *select_database(redisContext *c) { + redisReply *reply; + + /* Switch to DB 9 for testing, now that we know we can chat. */ + reply = redisCommand(c,"SELECT 9"); + assert(reply != NULL); + freeReplyObject(reply); + + /* Make sure the DB is emtpy */ + reply = redisCommand(c,"DBSIZE"); + assert(reply != NULL); + if (reply->type == REDIS_REPLY_INTEGER && reply->integer == 0) { + /* Awesome, DB 9 is empty and we can continue. */ + freeReplyObject(reply); + } else { + printf("Database #9 is not empty, test can not continue\n"); + exit(1); + } + + return c; +} + +static void disconnect(redisContext *c) { + redisReply *reply; + + /* Make sure we're on DB 9. */ + reply = redisCommand(c,"SELECT 9"); + assert(reply != NULL); + freeReplyObject(reply); + reply = redisCommand(c,"FLUSHDB"); + assert(reply != NULL); + freeReplyObject(reply); + + /* Free the context as well. */ + redisFree(c); +} + +static redisContext *connect(struct config config) { + redisContext *c = NULL; + + if (config.type == CONN_TCP) { + c = redisConnect(config.tcp.host, config.tcp.port); + } else if (config.type == CONN_UNIX) { + c = redisConnectUnix(config.unix.path); + } else { + assert(NULL); + } + + if (c->err) { + printf("Connection error: %s\n", c->errstr); exit(1); } + + return select_database(c); } static void test_format_commands(void) { @@ -78,29 +142,43 @@ static void test_format_commands(void) { len == 4+4+(3+2)+4+(1+2)+4+(1+2)); free(cmd); - test("Format command with printf-delegation (long long): "); - len = redisFormatCommand(&cmd,"key:%08lld",1234ll); - test_cond(strncmp(cmd,"*1\r\n$12\r\nkey:00001234\r\n",len) == 0 && - len == 4+5+(12+2)); - free(cmd); - - test("Format command with printf-delegation (float): "); - len = redisFormatCommand(&cmd,"v:%06.1f",12.34f); - test_cond(strncmp(cmd,"*1\r\n$8\r\nv:0012.3\r\n",len) == 0 && - len == 4+4+(8+2)); - free(cmd); - - test("Format command with printf-delegation and extra interpolation: "); - len = redisFormatCommand(&cmd,"key:%d %b",1234,"foo",3); - test_cond(strncmp(cmd,"*2\r\n$8\r\nkey:1234\r\n$3\r\nfoo\r\n",len) == 0 && - len == 4+4+(8+2)+4+(3+2)); - free(cmd); - - test("Format command with wrong printf format and extra interpolation: "); - len = redisFormatCommand(&cmd,"key:%08p %b",1234,"foo",3); - test_cond(strncmp(cmd,"*2\r\n$6\r\nkey:8p\r\n$3\r\nfoo\r\n",len) == 0 && - len == 4+4+(6+2)+4+(3+2)); - free(cmd); + /* Vararg width depends on the type. These tests make sure that the + * width is correctly determined using the format and subsequent varargs + * can correctly be interpolated. */ +#define INTEGER_WIDTH_TEST(fmt, type) do { \ + type value = 123; \ + test("Format command with printf-delegation (" #type "): "); \ + len = redisFormatCommand(&cmd,"key:%08" fmt " str:%s", value, "hello"); \ + test_cond(strncmp(cmd,"*2\r\n$12\r\nkey:00000123\r\n$9\r\nstr:hello\r\n",len) == 0 && \ + len == 4+5+(12+2)+4+(9+2)); \ + free(cmd); \ +} while(0) + +#define FLOAT_WIDTH_TEST(type) do { \ + type value = 123.0; \ + test("Format command with printf-delegation (" #type "): "); \ + len = redisFormatCommand(&cmd,"key:%08.3f str:%s", value, "hello"); \ + test_cond(strncmp(cmd,"*2\r\n$12\r\nkey:0123.000\r\n$9\r\nstr:hello\r\n",len) == 0 && \ + len == 4+5+(12+2)+4+(9+2)); \ + free(cmd); \ +} while(0) + + INTEGER_WIDTH_TEST("d", int); + INTEGER_WIDTH_TEST("hhd", char); + INTEGER_WIDTH_TEST("hd", short); + INTEGER_WIDTH_TEST("ld", long); + INTEGER_WIDTH_TEST("lld", long long); + INTEGER_WIDTH_TEST("u", unsigned int); + INTEGER_WIDTH_TEST("hhu", unsigned char); + INTEGER_WIDTH_TEST("hu", unsigned short); + INTEGER_WIDTH_TEST("lu", unsigned long); + INTEGER_WIDTH_TEST("llu", unsigned long long); + FLOAT_WIDTH_TEST(float); + FLOAT_WIDTH_TEST(double); + + test("Format command with invalid printf format: "); + len = redisFormatCommand(&cmd,"key:%08p %b",(void*)1234,"foo",3); + test_cond(len == -1); const char *argv[3]; argv[0] = "SET"; @@ -122,42 +200,117 @@ static void test_format_commands(void) { free(cmd); } -static void test_blocking_connection(void) { +static void test_reply_reader(void) { + redisReader *reader; + void *reply; + int ret; + + test("Error handling in reply parser: "); + reader = redisReaderCreate(); + redisReaderFeed(reader,(char*)"@foo\r\n",6); + ret = redisReaderGetReply(reader,NULL); + test_cond(ret == REDIS_ERR && + strcasecmp(reader->errstr,"Protocol error, got \"@\" as reply type byte") == 0); + redisReaderFree(reader); + + /* when the reply already contains multiple items, they must be free'd + * on an error. valgrind will bark when this doesn't happen. */ + test("Memory cleanup in reply parser: "); + reader = redisReaderCreate(); + redisReaderFeed(reader,(char*)"*2\r\n",4); + redisReaderFeed(reader,(char*)"$5\r\nhello\r\n",11); + redisReaderFeed(reader,(char*)"@foo\r\n",6); + ret = redisReaderGetReply(reader,NULL); + test_cond(ret == REDIS_ERR && + strcasecmp(reader->errstr,"Protocol error, got \"@\" as reply type byte") == 0); + redisReaderFree(reader); + + test("Set error on nested multi bulks with depth > 2: "); + reader = redisReaderCreate(); + redisReaderFeed(reader,(char*)"*1\r\n",4); + redisReaderFeed(reader,(char*)"*1\r\n",4); + redisReaderFeed(reader,(char*)"*1\r\n",4); + redisReaderFeed(reader,(char*)"*1\r\n",4); + ret = redisReaderGetReply(reader,NULL); + test_cond(ret == REDIS_ERR && + strncasecmp(reader->errstr,"No support for",14) == 0); + redisReaderFree(reader); + + test("Works with NULL functions for reply: "); + reader = redisReaderCreate(); + reader->fn = NULL; + redisReaderFeed(reader,(char*)"+OK\r\n",5); + ret = redisReaderGetReply(reader,&reply); + test_cond(ret == REDIS_OK && reply == (void*)REDIS_REPLY_STATUS); + redisReaderFree(reader); + + test("Works when a single newline (\\r\\n) covers two calls to feed: "); + reader = redisReaderCreate(); + reader->fn = NULL; + redisReaderFeed(reader,(char*)"+OK\r",4); + ret = redisReaderGetReply(reader,&reply); + assert(ret == REDIS_OK && reply == NULL); + redisReaderFeed(reader,(char*)"\n",1); + ret = redisReaderGetReply(reader,&reply); + test_cond(ret == REDIS_OK && reply == (void*)REDIS_REPLY_STATUS); + redisReaderFree(reader); + + test("Don't reset state after protocol error: "); + reader = redisReaderCreate(); + reader->fn = NULL; + redisReaderFeed(reader,(char*)"x",1); + ret = redisReaderGetReply(reader,&reply); + assert(ret == REDIS_ERR); + ret = redisReaderGetReply(reader,&reply); + test_cond(ret == REDIS_ERR && reply == NULL); + redisReaderFree(reader); + + /* Regression test for issue #45 on GitHub. */ + test("Don't do empty allocation for empty multi bulk: "); + reader = redisReaderCreate(); + redisReaderFeed(reader,(char*)"*0\r\n",4); + ret = redisReaderGetReply(reader,&reply); + test_cond(ret == REDIS_OK && + ((redisReply*)reply)->type == REDIS_REPLY_ARRAY && + ((redisReply*)reply)->elements == 0); + freeReplyObject(reply); + redisReaderFree(reader); +} + +static void test_blocking_connection_errors(void) { redisContext *c; - redisReply *reply; - int major, minor; test("Returns error when host cannot be resolved: "); c = redisConnect((char*)"idontexist.local", 6379); test_cond(c->err == REDIS_ERR_OTHER && - strcmp(c->errstr,"Can't resolve: idontexist.local") == 0); + (strcmp(c->errstr,"Name or service not known") == 0 || + strcmp(c->errstr,"Can't resolve: idontexist.local") == 0)); redisFree(c); test("Returns error when the port is not open: "); - c = redisConnect((char*)"localhost", 56380); + c = redisConnect((char*)"localhost", 1); test_cond(c->err == REDIS_ERR_IO && strcmp(c->errstr,"Connection refused") == 0); redisFree(c); - __connect(&c); + test("Returns error when the unix socket path doesn't accept connections: "); + c = redisConnectUnix((char*)"/tmp/idontexist.sock"); + test_cond(c->err == REDIS_ERR_IO); /* Don't care about the message... */ + redisFree(c); +} + +static void test_blocking_connection(struct config config) { + redisContext *c; + redisReply *reply; + + c = connect(config); + test("Is able to deliver commands: "); reply = redisCommand(c,"PING"); test_cond(reply->type == REDIS_REPLY_STATUS && strcasecmp(reply->str,"pong") == 0) freeReplyObject(reply); - /* Switch to DB 9 for testing, now that we know we can chat. */ - reply = redisCommand(c,"SELECT 9"); - freeReplyObject(reply); - - /* Make sure the DB is emtpy */ - reply = redisCommand(c,"DBSIZE"); - if (reply->type != REDIS_REPLY_INTEGER || reply->integer != 0) { - printf("Database #9 is not empty, test can not continue\n"); - exit(1); - } - freeReplyObject(reply); - test("Is a able to send commands verbatim: "); reply = redisCommand(c,"SET foo bar"); test_cond (reply->type == REDIS_REPLY_STATUS && @@ -221,6 +374,17 @@ static void test_blocking_connection(void) { strcasecmp(reply->element[1]->str,"pong") == 0); freeReplyObject(reply); + disconnect(c); +} + +static void test_blocking_io_errors(struct config config) { + redisContext *c; + redisReply *reply; + void *_reply; + int major, minor; + + /* Connect to target given by config. */ + c = connect(config); { /* Find out Redis version to determine the path for the next test */ const char *field = "redis_version:"; @@ -240,7 +404,7 @@ static void test_blocking_connection(void) { /* > 2.0 returns OK on QUIT and read() should be issued once more * to know the descriptor is at EOF. */ test_cond(strcasecmp(reply->str,"OK") == 0 && - redisGetReply(c,(void**)&reply) == REDIS_ERR); + redisGetReply(c,&_reply) == REDIS_ERR); freeReplyObject(reply); } else { test_cond(reply == NULL); @@ -255,91 +419,20 @@ static void test_blocking_connection(void) { strcmp(c->errstr,"Server closed the connection") == 0); redisFree(c); - __connect(&c); + c = connect(config); test("Returns I/O error on socket timeout: "); struct timeval tv = { 0, 1000 }; assert(redisSetTimeout(c,tv) == REDIS_OK); - test_cond(redisGetReply(c,(void**)&reply) == REDIS_ERR && + test_cond(redisGetReply(c,&_reply) == REDIS_ERR && c->err == REDIS_ERR_IO && errno == EAGAIN); redisFree(c); - - /* Context should be connected */ - __connect(&c); } -static void test_reply_reader(void) { - void *reader; - void *reply; - char *err; - int ret; - - test("Error handling in reply parser: "); - reader = redisReplyReaderCreate(); - redisReplyReaderFeed(reader,(char*)"@foo\r\n",6); - ret = redisReplyReaderGetReply(reader,NULL); - err = redisReplyReaderGetError(reader); - test_cond(ret == REDIS_ERR && - strcasecmp(err,"Protocol error, got \"@\" as reply type byte") == 0); - redisReplyReaderFree(reader); - - /* when the reply already contains multiple items, they must be free'd - * on an error. valgrind will bark when this doesn't happen. */ - test("Memory cleanup in reply parser: "); - reader = redisReplyReaderCreate(); - redisReplyReaderFeed(reader,(char*)"*2\r\n",4); - redisReplyReaderFeed(reader,(char*)"$5\r\nhello\r\n",11); - redisReplyReaderFeed(reader,(char*)"@foo\r\n",6); - ret = redisReplyReaderGetReply(reader,NULL); - err = redisReplyReaderGetError(reader); - test_cond(ret == REDIS_ERR && - strcasecmp(err,"Protocol error, got \"@\" as reply type byte") == 0); - redisReplyReaderFree(reader); - - test("Set error on nested multi bulks with depth > 1: "); - reader = redisReplyReaderCreate(); - redisReplyReaderFeed(reader,(char*)"*1\r\n",4); - redisReplyReaderFeed(reader,(char*)"*1\r\n",4); - redisReplyReaderFeed(reader,(char*)"*1\r\n",4); - ret = redisReplyReaderGetReply(reader,NULL); - err = redisReplyReaderGetError(reader); - test_cond(ret == REDIS_ERR && - strncasecmp(err,"No support for",14) == 0); - redisReplyReaderFree(reader); - - test("Works with NULL functions for reply: "); - reader = redisReplyReaderCreate(); - redisReplyReaderSetReplyObjectFunctions(reader,NULL); - redisReplyReaderFeed(reader,(char*)"+OK\r\n",5); - ret = redisReplyReaderGetReply(reader,&reply); - test_cond(ret == REDIS_OK && reply == (void*)REDIS_REPLY_STATUS); - redisReplyReaderFree(reader); - - test("Works when a single newline (\\r\\n) covers two calls to feed: "); - reader = redisReplyReaderCreate(); - redisReplyReaderSetReplyObjectFunctions(reader,NULL); - redisReplyReaderFeed(reader,(char*)"+OK\r",4); - ret = redisReplyReaderGetReply(reader,&reply); - assert(ret == REDIS_OK && reply == NULL); - redisReplyReaderFeed(reader,(char*)"\n",1); - ret = redisReplyReaderGetReply(reader,&reply); - test_cond(ret == REDIS_OK && reply == (void*)REDIS_REPLY_STATUS); - redisReplyReaderFree(reader); - - test("Properly reset state after protocol error: "); - reader = redisReplyReaderCreate(); - redisReplyReaderSetReplyObjectFunctions(reader,NULL); - redisReplyReaderFeed(reader,(char*)"x",1); - ret = redisReplyReaderGetReply(reader,&reply); - assert(ret == REDIS_ERR); - ret = redisReplyReaderGetReply(reader,&reply); - test_cond(ret == REDIS_OK && reply == NULL) -} - -static void test_throughput(void) { +static void test_throughput(struct config config) { + redisContext *c = connect(config); + redisReply **replies; int i, num; long long t1, t2; - redisContext *c = blocking_context; - redisReply **replies; test("Throughput:\n"); for (i = 0; i < 500; i++) @@ -396,18 +489,8 @@ static void test_throughput(void) { for (i = 0; i < num; i++) freeReplyObject(replies[i]); free(replies); printf("\t(%dx LRANGE with 500 elements (pipelined): %.3fs)\n", num, (t2-t1)/1000000.0); -} -static void cleanup(void) { - redisContext *c = blocking_context; - redisReply *reply; - - /* Make sure we're on DB 9 */ - reply = redisCommand(c,"SELECT 9"); - assert(reply != NULL); freeReplyObject(reply); - reply = redisCommand(c,"FLUSHDB"); - assert(reply != NULL); freeReplyObject(reply); - redisFree(c); + disconnect(c); } // static long __test_callback_flags = 0; @@ -429,7 +512,7 @@ static void cleanup(void) { // static redisContext *__connect_nonblock() { // /* Reset callback flags */ // __test_callback_flags = 0; -// return redisConnectNonBlock("127.0.0.1", 6379, NULL); +// return redisConnectNonBlock("127.0.0.1", port, NULL); // } // // static void test_nonblocking_connection() { @@ -510,23 +593,62 @@ static void cleanup(void) { // } int main(int argc, char **argv) { - if (argc > 1) { - if (strcmp(argv[1],"-s") == 0) - use_unix = 1; + struct config cfg = { + .tcp = { + .host = "127.0.0.1", + .port = 6379 + }, + .unix = { + .path = "/tmp/redis.sock" + } + }; + int throughput = 1; + + /* Ignore broken pipe signal (for I/O error tests). */ + signal(SIGPIPE, SIG_IGN); + + /* Parse command line options. */ + argv++; argc--; + while (argc) { + if (argc >= 2 && !strcmp(argv[0],"-h")) { + argv++; argc--; + cfg.tcp.host = argv[0]; + } else if (argc >= 2 && !strcmp(argv[0],"-p")) { + argv++; argc--; + cfg.tcp.port = atoi(argv[0]); + } else if (argc >= 2 && !strcmp(argv[0],"-s")) { + argv++; argc--; + cfg.unix.path = argv[0]; + } else if (argc >= 1 && !strcmp(argv[0],"--skip-throughput")) { + throughput = 0; + } else { + fprintf(stderr, "Invalid argument: %s\n", argv[0]); + exit(1); + } + argv++; argc--; } - signal(SIGPIPE, SIG_IGN); test_format_commands(); - test_blocking_connection(); test_reply_reader(); - // test_nonblocking_connection(); - test_throughput(); - cleanup(); + test_blocking_connection_errors(); - if (fails == 0) { - printf("ALL TESTS PASSED\n"); - } else { + printf("\nTesting against TCP connection (%s:%d):\n", cfg.tcp.host, cfg.tcp.port); + cfg.type = CONN_TCP; + test_blocking_connection(cfg); + test_blocking_io_errors(cfg); + if (throughput) test_throughput(cfg); + + printf("\nTesting against Unix socket connection (%s):\n", cfg.unix.path); + cfg.type = CONN_UNIX; + test_blocking_connection(cfg); + test_blocking_io_errors(cfg); + if (throughput) test_throughput(cfg); + + if (fails) { printf("*** %d TESTS FAILED ***\n", fails); + return 1; } + + printf("ALL TESTS PASSED\n"); return 0; } diff --git a/deps/hiredis/util.h b/deps/hiredis/util.h deleted file mode 100644 index f1929900..00000000 --- a/deps/hiredis/util.h +++ /dev/null @@ -1,40 +0,0 @@ -/* - * Copyright (c) 2009-2010, Salvatore Sanfilippo - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * * Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * * Redistributions in binary form must reproduce the above copyright - * notice, this list of conditions and the following disclaimer in the - * documentation and/or other materials provided with the distribution. - * * Neither the name of Redis nor the names of its contributors may be used - * to endorse or promote products derived from this software without - * specific prior written permission. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - -#ifndef __UTIL_H -#define __UTIL_H -#include - -/* Abort on out of memory */ -static void redisOOM(void) { - fprintf(stderr,"Out of memory in hiredis"); - exit(1); -} - -#endif diff --git a/deps/lua/src/Makefile b/deps/lua/src/Makefile index ff666160..6caed741 100644 --- a/deps/lua/src/Makefile +++ b/deps/lua/src/Makefile @@ -27,7 +27,7 @@ CORE_O= lapi.o lcode.o ldebug.o ldo.o ldump.o lfunc.o lgc.o llex.o lmem.o \ lobject.o lopcodes.o lparser.o lstate.o lstring.o ltable.o ltm.o \ lundump.o lvm.o lzio.o strbuf.o LIB_O= lauxlib.o lbaselib.o ldblib.o liolib.o lmathlib.o loslib.o ltablib.o \ - lstrlib.o loadlib.o linit.o lua_cjson.o + lstrlib.o loadlib.o linit.o lua_cjson.o lua_struct.o LUA_T= lua LUA_O= lua.o diff --git a/deps/lua/src/lua_struct.c b/deps/lua/src/lua_struct.c new file mode 100644 index 00000000..6807c838 --- /dev/null +++ b/deps/lua/src/lua_struct.c @@ -0,0 +1,354 @@ + +#include +#include +#include +#include + + +#include "lua.h" +#include "lauxlib.h" + + +/* +** {====================================================== +** Library for packing/unpacking structures. +** $Id: struct.c,v 1.2 2008/04/18 20:06:01 roberto Exp $ +** See Copyright Notice at the end of this file +** ======================================================= +*/ +/* +** Valid formats: +** > - big endian +** < - little endian +** ![num] - alignment +** x - pading +** b/B - signed/unsigned byte +** h/H - signed/unsigned short +** l/L - signed/unsigned long +** i/In - signed/unsigned integer with size `n' (default is size of int) +** cn - sequence of `n' chars (from/to a string); when packing, n==0 means + the whole string; when unpacking, n==0 means use the previous + read number as the string length +** s - zero-terminated string +** f - float +** d - double +** ' ' - ignored +*/ + + +/* is 'x' a power of 2? */ +#define isp2(x) ((x) > 0 && ((x) & ((x) - 1)) == 0) + +/* dummy structure to get alignment requirements */ +struct cD { + char c; + double d; +}; + + +#define PADDING (sizeof(struct cD) - sizeof(double)) +#define MAXALIGN (PADDING > sizeof(int) ? PADDING : sizeof(int)) + + +/* endian options */ +#define BIG 0 +#define LITTLE 1 + + +static union { + int dummy; + char endian; +} const native = {1}; + + +typedef struct Header { + int endian; + int align; +} Header; + + +static size_t getnum (const char **fmt, size_t df) { + if (!isdigit(**fmt)) /* no number? */ + return df; /* return default value */ + else { + size_t a = 0; + do { + a = a*10 + *((*fmt)++) - '0'; + } while (isdigit(**fmt)); + return a; + } +} + + +#define defaultoptions(h) ((h)->endian = native.endian, (h)->align = 1) + + + +static size_t optsize (lua_State *L, char opt, const char **fmt) { + switch (opt) { + case 'B': case 'b': return sizeof(char); + case 'H': case 'h': return sizeof(short); + case 'L': case 'l': return sizeof(long); + case 'f': return sizeof(float); + case 'd': return sizeof(double); + case 'x': return 1; + case 'c': return getnum(fmt, 1); + case 's': case ' ': case '<': case '>': case '!': return 0; + case 'i': case 'I': { + int sz = getnum(fmt, sizeof(int)); + if (!isp2(sz)) + luaL_error(L, "integral size %d is not a power of 2", sz); + return sz; + } + default: { + const char *msg = lua_pushfstring(L, "invalid format option [%c]", opt); + return luaL_argerror(L, 1, msg); + } + } +} + + +static int gettoalign (size_t len, Header *h, int opt, size_t size) { + if (size == 0 || opt == 'c') return 0; + if (size > (size_t)h->align) size = h->align; /* respect max. alignment */ + return (size - (len & (size - 1))) & (size - 1); +} + + +static void commoncases (lua_State *L, int opt, const char **fmt, Header *h) { + switch (opt) { + case ' ': return; /* ignore white spaces */ + case '>': h->endian = BIG; return; + case '<': h->endian = LITTLE; return; + case '!': { + int a = getnum(fmt, MAXALIGN); + if (!isp2(a)) + luaL_error(L, "alignment %d is not a power of 2", a); + h->align = a; + return; + } + default: assert(0); + } +} + + +static void putinteger (lua_State *L, luaL_Buffer *b, int arg, int endian, + int size) { + lua_Number n = luaL_checknumber(L, arg); + unsigned long value; + if (n < (lua_Number)LONG_MAX) + value = (long)n; + else + value = (unsigned long)n; + if (endian == LITTLE) { + int i; + for (i = 0; i < size; i++) + luaL_addchar(b, (value >> 8*i) & 0xff); + } + else { + int i; + for (i = size - 1; i >= 0; i--) + luaL_addchar(b, (value >> 8*i) & 0xff); + } +} + + +static void correctbytes (char *b, int size, int endian) { + if (endian != native.endian) { + int i = 0; + while (i < --size) { + char temp = b[i]; + b[i++] = b[size]; + b[size] = temp; + } + } +} + + +static int b_pack (lua_State *L) { + luaL_Buffer b; + const char *fmt = luaL_checkstring(L, 1); + Header h; + int arg = 2; + size_t totalsize = 0; + defaultoptions(&h); + lua_pushnil(L); /* mark to separate arguments from string buffer */ + luaL_buffinit(L, &b); + while (*fmt != '\0') { + int opt = *fmt++; + size_t size = optsize(L, opt, &fmt); + int toalign = gettoalign(totalsize, &h, opt, size); + totalsize += toalign; + while (toalign-- > 0) luaL_putchar(&b, '\0'); + switch (opt) { + case 'b': case 'B': case 'h': case 'H': + case 'l': case 'L': case 'i': case 'I': { /* integer types */ + putinteger(L, &b, arg++, h.endian, size); + break; + } + case 'x': { + luaL_putchar(&b, '\0'); + break; + } + case 'f': { + float f = (float)luaL_checknumber(L, arg++); + correctbytes((char *)&f, size, h.endian); + luaL_addlstring(&b, (char *)&f, size); + break; + } + case 'd': { + double d = luaL_checknumber(L, arg++); + correctbytes((char *)&d, size, h.endian); + luaL_addlstring(&b, (char *)&d, size); + break; + } + case 'c': case 's': { + size_t l; + const char *s = luaL_checklstring(L, arg++, &l); + if (size == 0) size = l; + luaL_argcheck(L, l >= (size_t)size, arg, "string too short"); + luaL_addlstring(&b, s, size); + if (opt == 's') { + luaL_putchar(&b, '\0'); /* add zero at the end */ + size++; + } + break; + } + default: commoncases(L, opt, &fmt, &h); + } + totalsize += size; + } + luaL_pushresult(&b); + return 1; +} + + +static lua_Number getinteger (const char *buff, int endian, + int issigned, int size) { + unsigned long l = 0; + if (endian == BIG) { + int i; + for (i = 0; i < size; i++) + l |= (unsigned long)(unsigned char)buff[size - i - 1] << (i*8); + } + else { + int i; + for (i = 0; i < size; i++) + l |= (unsigned long)(unsigned char)buff[i] << (i*8); + } + if (!issigned) + return (lua_Number)l; + else { /* signed format */ + unsigned long mask = ~(0UL) << (size*8 - 1); + if (l & mask) /* negative value? */ + l |= mask; /* signal extension */ + return (lua_Number)(long)l; + } +} + + +static int b_unpack (lua_State *L) { + Header h; + const char *fmt = luaL_checkstring(L, 1); + size_t ld; + const char *data = luaL_checklstring(L, 2, &ld); + size_t pos = luaL_optinteger(L, 3, 1) - 1; + defaultoptions(&h); + lua_settop(L, 2); + while (*fmt) { + int opt = *fmt++; + size_t size = optsize(L, opt, &fmt); + pos += gettoalign(pos, &h, opt, size); + luaL_argcheck(L, pos+size <= ld, 2, "data string too short"); + switch (opt) { + case 'b': case 'B': case 'h': case 'H': + case 'l': case 'L': case 'i': case 'I': { /* integer types */ + int issigned = islower(opt); + lua_Number res = getinteger(data+pos, h.endian, issigned, size); + lua_pushnumber(L, res); + break; + } + case 'x': { + break; + } + case 'f': { + float f; + memcpy(&f, data+pos, size); + correctbytes((char *)&f, sizeof(f), h.endian); + lua_pushnumber(L, f); + break; + } + case 'd': { + double d; + memcpy(&d, data+pos, size); + correctbytes((char *)&d, sizeof(d), h.endian); + lua_pushnumber(L, d); + break; + } + case 'c': { + if (size == 0) { + if (!lua_isnumber(L, -1)) + luaL_error(L, "format `c0' needs a previous size"); + size = lua_tonumber(L, -1); + lua_pop(L, 1); + luaL_argcheck(L, pos+size <= ld, 2, "data string too short"); + } + lua_pushlstring(L, data+pos, size); + break; + } + case 's': { + const char *e = (const char *)memchr(data+pos, '\0', ld - pos); + if (e == NULL) + luaL_error(L, "unfinished string in data"); + size = (e - (data+pos)) + 1; + lua_pushlstring(L, data+pos, size - 1); + break; + } + default: commoncases(L, opt, &fmt, &h); + } + pos += size; + } + lua_pushinteger(L, pos + 1); + return lua_gettop(L) - 2; +} + +/* }====================================================== */ + + + +static const struct luaL_reg thislib[] = { + {"pack", b_pack}, + {"unpack", b_unpack}, + {NULL, NULL} +}; + + +LUALIB_API int luaopen_struct (lua_State *L) { + luaL_register(L, "struct", thislib); + return 1; +} + + + +/****************************************************************************** +* Copyright (C) 2010 Lua.org, PUC-Rio. All rights reserved. +* +* Permission is hereby granted, free of charge, to any person obtaining +* a copy of this software and associated documentation files (the +* "Software"), to deal in the Software without restriction, including +* without limitation the rights to use, copy, modify, merge, publish, +* distribute, sublicense, and/or sell copies of the Software, and to +* permit persons to whom the Software is furnished to do so, subject to +* the following conditions: +* +* The above copyright notice and this permission notice shall be +* included in all copies or substantial portions of the Software. +* +* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +* MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. +* IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +* CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, +* TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE +* SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. +******************************************************************************/ diff --git a/redis.conf b/redis.conf index ad0c96e9..4760f291 100644 --- a/redis.conf +++ b/redis.conf @@ -201,21 +201,27 @@ slave-serve-stale-data yes # maxclients 10000 # Don't use more memory than the specified amount of bytes. -# When the memory limit is reached Redis will try to remove keys with an -# EXPIRE set. It will try to start freeing keys that are going to expire -# in little time and preserve keys with a longer time to live. -# Redis will also try to remove objects from free lists if possible. -# -# If all this fails, Redis will start to reply with errors to commands -# that will use more memory, like SET, LPUSH, and so on, and will continue -# to reply to most read-only commands like GET. -# -# WARNING: maxmemory can be a good idea mainly if you want to use Redis as a -# 'state' server or cache, not as a real DB. When Redis is used as a real -# database the memory usage will grow over the weeks, it will be obvious if -# it is going to use too much memory in the long run, and you'll have the time -# to upgrade. With maxmemory after the limit is reached you'll start to get -# errors for write operations, and this may even lead to DB inconsistency. +# When the memory limit is reached Redis will try to remove keys +# accordingly to the eviction policy selected (see maxmemmory-policy). +# +# If Redis can't remove keys according to the policy, or if the policy is +# set to 'noeviction', Redis will start to reply with errors to commands +# that would use more memory, like SET, LPUSH, and so on, and will continue +# to reply to read-only commands like GET. +# +# This option is usually useful when using Redis as an LRU cache, or to set +# an hard memory limit for an instance (using the 'noeviction' policy). +# +# WARNING: If you have slaves attached to an instance with maxmemory on, +# the size of the output buffers needed to feed the slaves are subtracted +# from the used memory count, so that network problems / resyncs will +# not trigger a loop where keys are evicted, and in turn the output +# buffer of slaves is full with DELs of keys evicted triggering the deletion +# of more keys, and so forth until the database is completely emptied. +# +# In short... if you have slaves attached it is suggested that you set a lower +# limit for maxmemory so that there is some free RAM on the system for slave +# output buffers (but this is not needed if the policy is 'noeviction'). # # maxmemory @@ -446,6 +452,43 @@ zset-max-ziplist-value 64 # want to free memory asap when possible. activerehashing yes +# The client output buffer limits can be used to force disconnection of clients +# that are not reading data from the server fast enough for some reason (a +# common reason is that a Pub/Sub client can't consume messages as fast as the +# publisher can produce them). +# +# The limit can be set differently for the three different classes of clients: +# +# normal -> normal clients +# slave -> slave clients and MONITOR clients +# pubsub -> clients subcribed to at least one pubsub channel or pattern +# +# The syntax of every client-output-buffer-limit directive is the following: +# +# client-output-buffer-limit +# +# A client is immediately disconnected once the hard limit is reached, or if +# the soft limit is reached and remains reached for the specified number of +# seconds (continuously). +# So for instance if the hard limit is 32 megabytes and the soft limit is +# 16 megabytes / 10 seconds, the client will get disconnected immediately +# if the size of the output buffers reach 32 megabytes, but will also get +# disconnected if the client reaches 16 megabytes and continuously overcomes +# the limit for 10 seconds. +# +# By default normal clients are not limited because they don't receive data +# without asking (in a push way), but just after a request, so only +# asynchronous clients may create a scenario where data is requested faster +# than it can read. +# +# Instead there is a default limit for pubsub and slave clients, since +# subscribers and slaves receive data in a push fashion. +# +# Both the hard or the soft limit can be disabled just setting it to zero. +client-output-buffer-limit normal 0 0 0 +client-output-buffer-limit slave 256mb 64mb 60 +client-output-buffer-limit pubsub 32mb 8mb 60 + ################################## INCLUDES ################################### # Include one or more other config files here. This is useful if you diff --git a/runtest b/runtest index 2ea4d39b..0eb384c2 100755 --- a/runtest +++ b/runtest @@ -1,4 +1,4 @@ -#!/bin/bash +#!/bin/sh TCL=tclsh8.5 which $TCL if [ "$?" != "0" ] diff --git a/src/MANIFESTO b/src/MANIFESTO deleted file mode 100644 index ad94fd8e..00000000 --- a/src/MANIFESTO +++ /dev/null @@ -1,20 +0,0 @@ -[Note: this is the Redis manifesto, for general information about - installing and running Redis read the README file instead.] - -Redis Manifesto -=============== - -1 - A DSL for Abstract Data Types. Redis is a DSL (Domain Specific Language) that manipulates abstract data types and implemented as a TCP daemon. Commands manipulate a key space where keys are binary-safe strings and values are different kinds of abstract data types. Every data type represents an abstract version of a fundamental data structure. For instance Redis Lists are an abstract representation of linked lists. In Redis, the essence of a data type isn't just the kind of operations that the data types support, but also the space and time complexity of the data type and the operations performed upon it. - -2 - Memory storage is #1. The Redis data set, composed of defined key-value pairs, is primarily stored in the computer's memory. The amount of memory in all kinds of computers, including entry-level servers, is increasing significantly each year. Memory is fast, and allows Redis to have very predictable performance. Datasets composed of 10k or 40 millions keys will perform similarly. Complex data types like Redis Sorted Sets are easy to implement and manipulate in memory with good performance, making Redis very simple. Redis will continue to explore alternative options (where data can be optionally stored on disk, say) but the main goal of the project remains the development of an in-memory database. - -3 - Fundamental data structures for a fundamental API. The Redis API is a direct consequence of fundamental data structures. APIs can often be arbitrary but not an API that resembles the nature of fundamental data structures. If we ever meet intelligent life forms from another part of the universe, they'll likely know, understand and recognize the same basic data structures we have in our computer science books. Redis will avoid intermediate layers in API, so that the complexity is obvious and more complex operations can be performed as the sum of the basic operations. - -4 - Code is like a poem; it's not just something we write to reach some practical result. Sometimes people that are far from the Redis philosophy suggest using other code written by other authors (frequently in other languages) in order to implement something Redis currently lacks. But to us this is like if Shakespeare decided to end Enrico IV using the Paradiso from the Divina Commedia. Is using any external code a bad idea? Not at all. Like in "One Thousand and One Nights" smaller self contained stories are embedded in a bigger story, we'll be happy to use beautiful self contained libraries when needed. At the same time, when writing the Redis story we're trying to write smaller stories that will fit in to other code. - -5 - We're against complexity. We believe designing systems is a fight against complexity. We'll accept to fight the complexity when it's worthwhile but we'll try hard to recognize when a small feature is not worth 1000s of lines of code. Most of the time the best way to fight complexity is by not creating it at all. - -6 - Two levels of API. The Redis API has two levels: 1) a subset of the API fits naturally into a distributed version of Redis and 2) a more complex API that supports multi-key operations. Both are useful if used judiciously but there's no way to make the more complex multi-keys API distributed in an opaque way without violating our other principles. We don't want to provide the illusion of something that will work magically when actually it can't in all cases. Instead we'll provide commands to quickly migrate keys from one instance to another to perform multi-key operations and expose the tradeoffs to the user. - -7 - We optimize for joy. We believe writing code is a lot of hard work, and the only way it can be worth is by enjoying it. When there is no longer joy in writing code, the best thing to do is stop. To prevent this, we'll avoid taking paths that will make Redis less of a joy to develop. - diff --git a/src/Makefile b/src/Makefile index 659d1d7f..cca785cb 100644 --- a/src/Makefile +++ b/src/Makefile @@ -59,7 +59,7 @@ CCOPT= $(CFLAGS) $(ARCH) $(PROF) PREFIX= /usr/local INSTALL_BIN= $(PREFIX)/bin -INSTALL= cp -p +INSTALL= cp -pf CCCOLOR="\033[34m" LINKCOLOR="\033[34;1m" @@ -73,7 +73,7 @@ QUIET_CC = @printf ' %b %b\n' $(CCCOLOR)CC$(ENDCOLOR) $(SRCCOLOR)$@$(ENDCOLOR QUIET_LINK = @printf ' %b %b\n' $(LINKCOLOR)LINK$(ENDCOLOR) $(BINCOLOR)$@$(ENDCOLOR); endif -OBJ = adlist.o ae.o anet.o dict.o redis.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endian.o slowlog.o scripting.o bio.o rio.o rand.o +OBJ = adlist.o ae.o anet.o dict.o redis.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o BENCHOBJ = ae.o anet.o redis-benchmark.o sds.o adlist.o zmalloc.o CLIOBJ = anet.o sds.o adlist.o redis-cli.o zmalloc.o release.o CHECKDUMPOBJ = redis-check-dump.o lzf_c.o lzf_d.o @@ -112,8 +112,8 @@ db.o: db.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \ debug.o: debug.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \ zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h util.h sha1.h dict.o: dict.c fmacros.h dict.h zmalloc.h -endian.o: endian.c -intset.o: intset.c intset.h zmalloc.h endian.h +endianconv.o: endianconv.c +intset.o: intset.c intset.h zmalloc.h endianconv.h lzf_c.o: lzf_c.c lzfP.h lzf_d.o: lzf_d.c lzfP.h multi.o: multi.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \ @@ -163,8 +163,8 @@ t_string.o: t_string.c redis.h fmacros.h config.h ae.h sds.h dict.h \ t_zset.o: t_zset.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \ zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h util.h util.o: util.c fmacros.h util.h -ziplist.o: ziplist.c zmalloc.h util.h ziplist.h endian.h -zipmap.o: zipmap.c zmalloc.h endian.h +ziplist.o: ziplist.c zmalloc.h util.h ziplist.h endianconv.h +zipmap.o: zipmap.c zmalloc.h endianconv.h zmalloc.o: zmalloc.c config.h zmalloc.h # Clean local objects when ARCH is different @@ -175,7 +175,7 @@ else endif .make-arch: - -(cd ../deps && make $(DEPENDENCY_TARGETS) ARCH="$(ARCH)") + -(cd ../deps && $(MAKE) $(DEPENDENCY_TARGETS) ARCH="$(ARCH)") -(echo $(ARCH) > .make-arch) # Clean local objects when allocator changes diff --git a/src/adlist.c b/src/adlist.c index 015012f5..51ba03bd 100644 --- a/src/adlist.c +++ b/src/adlist.c @@ -57,7 +57,7 @@ list *listCreate(void) * This function can't fail. */ void listRelease(list *list) { - unsigned int len; + unsigned long len; listNode *current, *next; current = list->head; @@ -310,7 +310,7 @@ listNode *listSearchKey(list *list, void *key) * and so on. Negative integers are used in order to count * from the tail, -1 is the last element, -2 the penultimante * and so on. If the index is out of range NULL is returned. */ -listNode *listIndex(list *list, int index) { +listNode *listIndex(list *list, long index) { listNode *n; if (index < 0) { diff --git a/src/adlist.h b/src/adlist.h index a1209f62..36dba1ff 100644 --- a/src/adlist.h +++ b/src/adlist.h @@ -50,7 +50,7 @@ typedef struct list { void *(*dup)(void *ptr); void (*free)(void *ptr); int (*match)(void *ptr, void *key); - unsigned int len; + unsigned long len; } list; /* Functions implemented as macros */ @@ -81,7 +81,7 @@ listNode *listNext(listIter *iter); void listReleaseIterator(listIter *iter); list *listDup(list *orig); listNode *listSearchKey(list *list, void *key); -listNode *listIndex(list *list, int index); +listNode *listIndex(list *list, long index); void listRewind(list *list, listIter *li); void listRewindTail(list *list, listIter *li); diff --git a/src/ae.c b/src/ae.c index 6ddccdf7..4099b125 100644 --- a/src/ae.c +++ b/src/ae.c @@ -35,6 +35,7 @@ #include #include #include +#include #include "ae.h" #include "zmalloc.h" @@ -56,32 +57,30 @@ aeEventLoop *aeCreateEventLoop(int setsize) { aeEventLoop *eventLoop; int i; - if ((eventLoop = zmalloc(sizeof(*eventLoop))) == NULL) return NULL; - eventLoop->events = NULL; - eventLoop->fired = NULL; + if ((eventLoop = zmalloc(sizeof(*eventLoop))) == NULL) goto err; eventLoop->events = zmalloc(sizeof(aeFileEvent)*setsize); eventLoop->fired = zmalloc(sizeof(aeFiredEvent)*setsize); - if (eventLoop->events == NULL || eventLoop->fired == NULL) { - zfree(eventLoop->events); - zfree(eventLoop->fired); - zfree(eventLoop); - return NULL; - } + if (eventLoop->events == NULL || eventLoop->fired == NULL) goto err; eventLoop->setsize = setsize; eventLoop->timeEventHead = NULL; eventLoop->timeEventNextId = 0; eventLoop->stop = 0; eventLoop->maxfd = -1; eventLoop->beforesleep = NULL; - if (aeApiCreate(eventLoop) == -1) { - zfree(eventLoop); - return NULL; - } + if (aeApiCreate(eventLoop) == -1) goto err; /* Events with mask == AE_NONE are not set. So let's initialize the * vector with it. */ for (i = 0; i < setsize; i++) eventLoop->events[i].mask = AE_NONE; return eventLoop; + +err: + if (eventLoop) { + zfree(eventLoop->events); + zfree(eventLoop->fired); + zfree(eventLoop); + } + return NULL; } void aeDeleteEventLoop(aeEventLoop *eventLoop) { diff --git a/src/aof.c b/src/aof.c index 25d6f454..0bdcd9ed 100644 --- a/src/aof.c +++ b/src/aof.c @@ -295,6 +295,8 @@ struct redisClient *createFakeClient(void) { * so that Redis will not try to send replies to this client. */ c->replstate = REDIS_REPL_WAIT_BGSAVE_START; c->reply = listCreate(); + c->reply_bytes = 0; + c->obuf_soft_limit_reached_time = 0; c->watched_keys = listCreate(); listSetFreeMethod(c->reply,decrRefCount); listSetDupMethod(c->reply,dupClientReplyValue); @@ -842,7 +844,7 @@ void aofRemoveTempFile(pid_t childpid) { /* Update the server.aof_current_size filed explicitly using stat(2) * to check the size of the file. This is useful after a rewrite or after * a restart, normally the size is updated just adding the write length - * to the current lenght, that is much faster. */ + * to the current length, that is much faster. */ void aofUpdateCurrentSize(void) { struct redis_stat sb; diff --git a/src/cluster.c b/src/cluster.c index 4ccff657..85cb1198 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -613,7 +613,7 @@ int clusterProcessPacket(clusterLink *link) { } } /* Update our info about the node */ - link->node->pong_received = time(NULL); + if (link->node) link->node->pong_received = time(NULL); /* Update master/slave info */ if (sender) { @@ -804,7 +804,7 @@ void clusterBroadcastMessage(void *buf, size_t len) { /* Build the message header */ void clusterBuildMessageHdr(clusterMsg *hdr, int type) { - int totlen; + int totlen = 0; memset(hdr,0,sizeof(*hdr)); hdr->type = htons(type); diff --git a/src/config.c b/src/config.c index 3f713e3e..9d28d678 100644 --- a/src/config.c +++ b/src/config.c @@ -315,6 +315,27 @@ void loadServerConfigFromString(char *config) { server.slowlog_log_slower_than = strtoll(argv[1],NULL,10); } else if (!strcasecmp(argv[0],"slowlog-max-len") && argc == 2) { server.slowlog_max_len = strtoll(argv[1],NULL,10); + } else if (!strcasecmp(argv[0],"client-output-buffer-limit") && + argc == 5) + { + int class = getClientLimitClassByName(argv[1]); + unsigned long long hard, soft; + int soft_seconds; + + if (class == -1) { + err = "Unrecognized client limit class"; + goto loaderr; + } + hard = memtoll(argv[2],NULL); + soft = memtoll(argv[3],NULL); + soft_seconds = atoi(argv[4]); + if (soft_seconds < 0) { + err = "Negative number of seconds in soft limt is invalid"; + goto loaderr; + } + server.client_obuf_limits[class].hard_limit_bytes = hard; + server.client_obuf_limits[class].soft_limit_bytes = soft; + server.client_obuf_limits[class].soft_limit_seconds = soft_seconds; } else { err = "Bad directive or wrong number of arguments"; goto loaderr; } @@ -537,6 +558,53 @@ void configSetCommand(redisClient *c) { } else { goto badfmt; } + } else if (!strcasecmp(c->argv[2]->ptr,"client-output-buffer-limit")) { + int vlen, j; + sds *v = sdssplitlen(o->ptr,sdslen(o->ptr)," ",1,&vlen); + + /* We need a multiple of 4: */ + if (vlen % 4) { + sdsfreesplitres(v,vlen); + goto badfmt; + } + + /* Sanity check of single arguments, so that we either refuse the + * whole configuration string or accept it all, even if a single + * error in a single client class is present. */ + for (j = 0; j < vlen; j++) { + char *eptr; + long val; + + if ((j % 4) == 0) { + if (getClientLimitClassByName(v[j]) == -1) { + sdsfreesplitres(v,vlen); + goto badfmt; + } + } else { + val = strtoll(v[j], &eptr, 10); + if (eptr[0] != '\0' || val < 0) { + sdsfreesplitres(v,vlen); + goto badfmt; + } + } + } + /* Finally set the new config */ + for (j = 0; j < vlen; j += 4) { + int class; + unsigned long long hard, soft; + int soft_seconds; + + class = getClientLimitClassByName(v[j]); + hard = strtoll(v[j+1],NULL,10); + soft = strtoll(v[j+2],NULL,10); + soft_seconds = strtoll(v[j+3],NULL,10); + + server.client_obuf_limits[class].hard_limit_bytes = hard; + server.client_obuf_limits[class].soft_limit_bytes = soft; + server.client_obuf_limits[class].soft_limit_seconds = soft_seconds; + } + sdsfreesplitres(v,vlen); + } else { addReplyErrorFormat(c,"Unsupported CONFIG parameter: %s", (char*)c->argv[2]->ptr); @@ -736,6 +804,24 @@ void configGetCommand(redisClient *c) { addReplyBulkCString(c,s); matches++; } + if (stringmatch(pattern,"client-output-buffer-limit",0)) { + sds buf = sdsempty(); + int j; + + for (j = 0; j < REDIS_CLIENT_LIMIT_NUM_CLASSES; j++) { + buf = sdscatprintf(buf,"%s %llu %llu %ld", + getClientLimitClassName(j), + server.client_obuf_limits[j].hard_limit_bytes, + server.client_obuf_limits[j].soft_limit_bytes, + (long) server.client_obuf_limits[j].soft_limit_seconds); + if (j != REDIS_CLIENT_LIMIT_NUM_CLASSES-1) + buf = sdscatlen(buf," ",1); + } + addReplyBulkCString(c,"client-output-buffer-limit"); + addReplyBulkCString(c,buf); + sdsfree(buf); + matches++; + } setDeferredMultiBulkLength(c,replylen,matches*2); } diff --git a/src/config.h b/src/config.h index f38ce872..323a3ddd 100644 --- a/src/config.h +++ b/src/config.h @@ -25,7 +25,7 @@ #endif /* Test for backtrace() */ -#if defined(__APPLE__) || defined(__linux__) +#if defined(__APPLE__) || defined(__linux__) || defined(__sun) #define HAVE_BACKTRACE 1 #endif diff --git a/src/db.c b/src/db.c index f7d93e08..a0775af9 100644 --- a/src/db.c +++ b/src/db.c @@ -42,17 +42,22 @@ robj *lookupKey(redisDb *db, robj *key) { * a copy on write madness. */ if (server.rdb_child_pid == -1 && server.aof_child_pid == -1) val->lru = server.lruclock; - server.stat_keyspace_hits++; return val; } else { - server.stat_keyspace_misses++; return NULL; } } robj *lookupKeyRead(redisDb *db, robj *key) { + robj *val; + expireIfNeeded(db,key); - return lookupKey(db,key); + val = lookupKey(db,key); + if (val == NULL) + server.stat_keyspace_misses++; + else + server.stat_keyspace_hits++; + return val; } robj *lookupKeyWrite(redisDb *db, robj *key) { @@ -110,7 +115,7 @@ void setKey(redisDb *db, robj *key, robj *val) { } incrRefCount(val); removeExpire(db,key); - touchWatchedKey(db,key); + signalModifiedKey(db,key); } int dbExists(redisDb *db, robj *key) { @@ -483,9 +488,10 @@ long long getExpire(redisDb *db, robj *key) { void propagateExpire(redisDb *db, robj *key) { robj *argv[2]; - argv[0] = createStringObject("DEL",3); + argv[0] = shared.del; argv[1] = key; - incrRefCount(key); + incrRefCount(argv[0]); + incrRefCount(argv[1]); if (server.aof_state != REDIS_AOF_OFF) feedAppendOnlyFile(server.delCommand,db->id,argv,2); diff --git a/src/debug.c b/src/debug.c index 728aa8a1..a355df05 100644 --- a/src/debug.c +++ b/src/debug.c @@ -2,6 +2,12 @@ #include "sha1.h" /* SHA1 is used for DEBUG DIGEST */ #include +#include + +#ifdef HAVE_BACKTRACE +#include +#include +#endif /* HAVE_BACKTRACE */ /* ================================= Debugging ============================== */ @@ -297,6 +303,8 @@ void debugCommand(redisClient *c) { } } +/* =========================== Crash handling ============================== */ + void _redisAssert(char *estr, char *file, int line) { bugReportStart(); redisLog(REDIS_WARNING,"=== ASSERTION FAILED ==="); @@ -306,8 +314,8 @@ void _redisAssert(char *estr, char *file, int line) { server.assert_file = file; server.assert_line = line; redisLog(REDIS_WARNING,"(forcing SIGSEGV to print the bug report.)"); - *((char*)-1) = 'x'; #endif + *((char*)-1) = 'x'; } void _redisAssertPrintClientInfo(redisClient *c) { @@ -376,7 +384,272 @@ void _redisPanic(char *msg, char *file, int line) { redisLog(REDIS_WARNING,"Guru Meditation: %s #%s:%d",msg,file,line); #ifdef HAVE_BACKTRACE redisLog(REDIS_WARNING,"(forcing SIGSEGV in order to print the stack trace)"); +#endif redisLog(REDIS_WARNING,"------------------------------------------------"); *((char*)-1) = 'x'; +} + +void bugReportStart(void) { + if (server.bug_report_start == 0) { + redisLog(REDIS_WARNING, + "\n\n=== REDIS BUG REPORT START: Cut & paste starting from here ==="); + server.bug_report_start = 1; + } +} + +#ifdef HAVE_BACKTRACE +static void *getMcontextEip(ucontext_t *uc) { +#if defined(__FreeBSD__) + return (void*) uc->uc_mcontext.mc_eip; +#elif defined(__dietlibc__) + return (void*) uc->uc_mcontext.eip; +#elif defined(__APPLE__) && !defined(MAC_OS_X_VERSION_10_6) + #if __x86_64__ + return (void*) uc->uc_mcontext->__ss.__rip; + #elif __i386__ + return (void*) uc->uc_mcontext->__ss.__eip; + #else + return (void*) uc->uc_mcontext->__ss.__srr0; + #endif +#elif defined(__APPLE__) && defined(MAC_OS_X_VERSION_10_6) + #if defined(_STRUCT_X86_THREAD_STATE64) && !defined(__i386__) + return (void*) uc->uc_mcontext->__ss.__rip; + #else + return (void*) uc->uc_mcontext->__ss.__eip; + #endif +#elif defined(__i386__) + return (void*) uc->uc_mcontext.gregs[14]; /* Linux 32 */ +#elif defined(__X86_64__) || defined(__x86_64__) + return (void*) uc->uc_mcontext.gregs[16]; /* Linux 64 */ +#elif defined(__ia64__) /* Linux IA64 */ + return (void*) uc->uc_mcontext.sc_ip; +#else + return NULL; +#endif +} + +void logStackContent(void **sp) { + int i; + for (i = 15; i >= 0; i--) { + if (sizeof(long) == 4) + redisLog(REDIS_WARNING, "(%08lx) -> %08lx", sp+i, sp[i]); + else + redisLog(REDIS_WARNING, "(%016lx) -> %016lx", sp+i, sp[i]); + } +} + +void logRegisters(ucontext_t *uc) { + redisLog(REDIS_WARNING, "--- REGISTERS"); +#if defined(__APPLE__) && defined(MAC_OS_X_VERSION_10_6) + #if defined(_STRUCT_X86_THREAD_STATE64) && !defined(__i386__) + redisLog(REDIS_WARNING, + "\n" + "RAX:%016lx RBX:%016lx\nRCX:%016lx RDX:%016lx\n" + "RDI:%016lx RSI:%016lx\nRBP:%016lx RSP:%016lx\n" + "R8 :%016lx R9 :%016lx\nR10:%016lx R11:%016lx\n" + "R12:%016lx R13:%016lx\nR14:%016lx R15:%016lx\n" + "RIP:%016lx EFL:%016lx\nCS :%016lx FS:%016lx GS:%016lx", + uc->uc_mcontext->__ss.__rax, + uc->uc_mcontext->__ss.__rbx, + uc->uc_mcontext->__ss.__rcx, + uc->uc_mcontext->__ss.__rdx, + uc->uc_mcontext->__ss.__rdi, + uc->uc_mcontext->__ss.__rsi, + uc->uc_mcontext->__ss.__rbp, + uc->uc_mcontext->__ss.__rsp, + uc->uc_mcontext->__ss.__r8, + uc->uc_mcontext->__ss.__r9, + uc->uc_mcontext->__ss.__r10, + uc->uc_mcontext->__ss.__r11, + uc->uc_mcontext->__ss.__r12, + uc->uc_mcontext->__ss.__r13, + uc->uc_mcontext->__ss.__r14, + uc->uc_mcontext->__ss.__r15, + uc->uc_mcontext->__ss.__rip, + uc->uc_mcontext->__ss.__rflags, + uc->uc_mcontext->__ss.__cs, + uc->uc_mcontext->__ss.__fs, + uc->uc_mcontext->__ss.__gs + ); + logStackContent((void**)uc->uc_mcontext->__ss.__rsp); + #else + redisLog(REDIS_WARNING, + "\n" + "EAX:%08lx EBX:%08lx ECX:%08lx EDX:%08lx\n" + "EDI:%08lx ESI:%08lx EBP:%08lx ESP:%08lx\n" + "SS:%08lx EFL:%08lx EIP:%08lx CS :%08lx\n" + "DS:%08lx ES:%08lx FS :%08lx GS :%08lx", + uc->uc_mcontext->__ss.__eax, + uc->uc_mcontext->__ss.__ebx, + uc->uc_mcontext->__ss.__ecx, + uc->uc_mcontext->__ss.__edx, + uc->uc_mcontext->__ss.__edi, + uc->uc_mcontext->__ss.__esi, + uc->uc_mcontext->__ss.__ebp, + uc->uc_mcontext->__ss.__esp, + uc->uc_mcontext->__ss.__ss, + uc->uc_mcontext->__ss.__eflags, + uc->uc_mcontext->__ss.__eip, + uc->uc_mcontext->__ss.__cs, + uc->uc_mcontext->__ss.__ds, + uc->uc_mcontext->__ss.__es, + uc->uc_mcontext->__ss.__fs, + uc->uc_mcontext->__ss.__gs + ); + logStackContent((void**)uc->uc_mcontext->__ss.__esp); + #endif +#elif defined(__i386__) + redisLog(REDIS_WARNING, + "\n" + "EAX:%08lx EBX:%08lx ECX:%08lx EDX:%08lx\n" + "EDI:%08lx ESI:%08lx EBP:%08lx ESP:%08lx\n" + "SS :%08lx EFL:%08lx EIP:%08lx CS:%08lx\n" + "DS :%08lx ES :%08lx FS :%08lx GS:%08lx", + uc->uc_mcontext.gregs[11], + uc->uc_mcontext.gregs[8], + uc->uc_mcontext.gregs[10], + uc->uc_mcontext.gregs[9], + uc->uc_mcontext.gregs[4], + uc->uc_mcontext.gregs[5], + uc->uc_mcontext.gregs[6], + uc->uc_mcontext.gregs[7], + uc->uc_mcontext.gregs[18], + uc->uc_mcontext.gregs[17], + uc->uc_mcontext.gregs[14], + uc->uc_mcontext.gregs[15], + uc->uc_mcontext.gregs[3], + uc->uc_mcontext.gregs[2], + uc->uc_mcontext.gregs[1], + uc->uc_mcontext.gregs[0] + ); + logStackContent((void**)uc->uc_mcontext.gregs[7]); +#elif defined(__X86_64__) || defined(__x86_64__) + redisLog(REDIS_WARNING, + "\n" + "RAX:%016lx RBX:%016lx\nRCX:%016lx RDX:%016lx\n" + "RDI:%016lx RSI:%016lx\nRBP:%016lx RSP:%016lx\n" + "R8 :%016lx R9 :%016lx\nR10:%016lx R11:%016lx\n" + "R12:%016lx R13:%016lx\nR14:%016lx R15:%016lx\n" + "RIP:%016lx EFL:%016lx\nCSGSFS:%016lx", + uc->uc_mcontext.gregs[13], + uc->uc_mcontext.gregs[11], + uc->uc_mcontext.gregs[14], + uc->uc_mcontext.gregs[12], + uc->uc_mcontext.gregs[8], + uc->uc_mcontext.gregs[9], + uc->uc_mcontext.gregs[10], + uc->uc_mcontext.gregs[15], + uc->uc_mcontext.gregs[0], + uc->uc_mcontext.gregs[1], + uc->uc_mcontext.gregs[2], + uc->uc_mcontext.gregs[3], + uc->uc_mcontext.gregs[4], + uc->uc_mcontext.gregs[5], + uc->uc_mcontext.gregs[6], + uc->uc_mcontext.gregs[7], + uc->uc_mcontext.gregs[16], + uc->uc_mcontext.gregs[17], + uc->uc_mcontext.gregs[18] + ); + logStackContent((void**)uc->uc_mcontext.gregs[15]); +#else + redisLog(REDIS_WARNING, + " Dumping of registers not supported for this OS/arch"); #endif } + +void sigsegvHandler(int sig, siginfo_t *info, void *secret) { + void *trace[100]; + char **messages = NULL; + int i, trace_size = 0; + ucontext_t *uc = (ucontext_t*) secret; + sds infostring, clients; + struct sigaction act; + REDIS_NOTUSED(info); + + bugReportStart(); + redisLog(REDIS_WARNING, + " Redis %s crashed by signal: %d", REDIS_VERSION, sig); + redisLog(REDIS_WARNING, + " Failed assertion: %s (%s:%d)", server.assert_failed, + server.assert_file, server.assert_line); + + /* Generate the stack trace */ + trace_size = backtrace(trace, 100); + + /* overwrite sigaction with caller's address */ + if (getMcontextEip(uc) != NULL) { + trace[1] = getMcontextEip(uc); + } + messages = backtrace_symbols(trace, trace_size); + redisLog(REDIS_WARNING, "--- STACK TRACE"); + for (i=1; iargc; j++) { + robj *decoded; + + decoded = getDecodedObject(cc->argv[j]); + redisLog(REDIS_WARNING,"argv[%d]: '%s'", j, (char*)decoded->ptr); + decrRefCount(decoded); + } + /* Check if the first argument, usually a key, is found inside the + * selected DB, and if so print info about the associated object. */ + if (cc->argc >= 1) { + robj *val, *key; + dictEntry *de; + + key = getDecodedObject(cc->argv[1]); + de = dictFind(cc->db->dict, key->ptr); + if (de) { + val = dictGetVal(de); + redisLog(REDIS_WARNING,"key '%s' found in DB containing the following object:", key->ptr); + redisLogObjectDebugInfo(val); + } + decrRefCount(key); + } + } + + /* Log dump of processor registers */ + logRegisters(uc); + + redisLog(REDIS_WARNING, +"\n=== REDIS BUG REPORT END. Make sure to include from START to END. ===\n\n" +" Please report the crash opening an issue on github:\n\n" +" http://github.com/antirez/redis/issues\n\n" +); + /* free(messages); Don't call free() with possibly corrupted memory. */ + if (server.daemonize) unlink(server.pidfile); + + /* Make sure we exit with the right signal at the end. So for instance + * the core will be dumped if enabled. */ + sigemptyset (&act.sa_mask); + /* When the SA_SIGINFO flag is set in sa_flags then sa_sigaction + * is used. Otherwise, sa_handler is used */ + act.sa_flags = SA_NODEFER | SA_ONSTACK | SA_RESETHAND; + act.sa_handler = SIG_DFL; + sigaction (sig, &act, NULL); + kill(getpid(),sig); +} +#endif /* HAVE_BACKTRACE */ diff --git a/src/dict.c b/src/dict.c index a573bcd6..53e16be0 100644 --- a/src/dict.c +++ b/src/dict.c @@ -85,10 +85,20 @@ unsigned int dictIdentityHashFunction(unsigned int key) return key; } +static int dict_hash_function_seed = 5381; + +void dictSetHashFunctionSeed(unsigned int seed) { + dict_hash_function_seed = seed; +} + +unsigned int dictGetHashFunctionSeed(void) { + return dict_hash_function_seed; +} + /* Generic hash function (a popular one from Bernstein). * I tested a few and this was the best. */ unsigned int dictGenHashFunction(const unsigned char *buf, int len) { - unsigned int hash = 5381; + unsigned int hash = dict_hash_function_seed; while (len--) hash = ((hash << 5) + hash) + (*buf++); /* hash * 33 + c */ @@ -97,7 +107,7 @@ unsigned int dictGenHashFunction(const unsigned char *buf, int len) { /* And a case insensitive version */ unsigned int dictGenCaseHashFunction(const unsigned char *buf, int len) { - unsigned int hash = 5381; + unsigned int hash = dict_hash_function_seed; while (len--) hash = ((hash << 5) + hash) + (tolower(*buf++)); /* hash * 33 + c */ diff --git a/src/dict.h b/src/dict.h index 76451047..5f856953 100644 --- a/src/dict.h +++ b/src/dict.h @@ -162,6 +162,8 @@ void dictEnableResize(void); void dictDisableResize(void); int dictRehash(dict *d, int n); int dictRehashMilliseconds(dict *d, int ms); +void dictSetHashFunctionSeed(unsigned int initval); +unsigned int dictGetHashFunctionSeed(void); /* Hash table types */ extern dictType dictTypeHeapStringCopyKey; diff --git a/src/endian.c b/src/endian.c deleted file mode 100644 index aff2425a..00000000 --- a/src/endian.c +++ /dev/null @@ -1,63 +0,0 @@ -/* Toggle the 16 bit unsigned integer pointed by *p from little endian to - * big endian */ -void memrev16(void *p) { - unsigned char *x = p, t; - - t = x[0]; - x[0] = x[1]; - x[1] = t; -} - -/* Toggle the 32 bit unsigned integer pointed by *p from little endian to - * big endian */ -void memrev32(void *p) { - unsigned char *x = p, t; - - t = x[0]; - x[0] = x[3]; - x[3] = t; - t = x[1]; - x[1] = x[2]; - x[2] = t; -} - -/* Toggle the 64 bit unsigned integer pointed by *p from little endian to - * big endian */ -void memrev64(void *p) { - unsigned char *x = p, t; - - t = x[0]; - x[0] = x[7]; - x[7] = t; - t = x[1]; - x[1] = x[6]; - x[6] = t; - t = x[2]; - x[2] = x[5]; - x[5] = t; - t = x[3]; - x[3] = x[4]; - x[4] = t; -} - -#ifdef TESTMAIN -#include - -int main(void) { - char buf[32]; - - sprintf(buf,"ciaoroma"); - memrev16(buf); - printf("%s\n", buf); - - sprintf(buf,"ciaoroma"); - memrev32(buf); - printf("%s\n", buf); - - sprintf(buf,"ciaoroma"); - memrev64(buf); - printf("%s\n", buf); - - return 0; -} -#endif diff --git a/src/endian.h b/src/endian.h deleted file mode 100644 index bef82272..00000000 --- a/src/endian.h +++ /dev/null @@ -1,20 +0,0 @@ -#ifndef __ENDIAN_H -#define __ENDIAN_H - -void memrev16(void *p); -void memrev32(void *p); -void memrev64(void *p); - -/* variants of the function doing the actual convertion only if the target - * host is big endian */ -#if (BYTE_ORDER == LITTLE_ENDIAN) -#define memrev16ifbe(p) -#define memrev32ifbe(p) -#define memrev64ifbe(p) -#else -#define memrev16ifbe(p) memrev16(p) -#define memrev32ifbe(p) memrev32(p) -#define memrev64ifbe(p) memrev64(p) -#endif - -#endif diff --git a/src/endianconv.c b/src/endianconv.c new file mode 100644 index 00000000..9adf09c1 --- /dev/null +++ b/src/endianconv.c @@ -0,0 +1,124 @@ +/* endinconv.c -- Endian conversions utilities. + * + * This functions are never called directly, but always using the macros + * defined into endianconv.h, this way we define everything is a non-operation + * if the arch is already little endian. + * + * Redis tries to encode everything as little endian (but a few things that need + * to be backward compatible are still in big endian) because most of the + * production environments are little endian, and we have a lot of conversions + * in a few places because ziplists, intsets, zipmaps, need to be endian-neutral + * even in memory, since they are serialied on RDB files directly with a single + * write(2) without other additional steps. + * + * ---------------------------------------------------------------------------- + * + * Copyright (c) 2011-2012, Salvatore Sanfilippo + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * * Redistributions of source code must retain the above copyright notice, + * this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * * Neither the name of Redis nor the names of its contributors may be used + * to endorse or promote products derived from this software without + * specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" + * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE + * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR + * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF + * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS + * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN + * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) + * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE + * POSSIBILITY OF SUCH DAMAGE. + */ + + +#include + +/* Toggle the 16 bit unsigned integer pointed by *p from little endian to + * big endian */ +void memrev16(void *p) { + unsigned char *x = p, t; + + t = x[0]; + x[0] = x[1]; + x[1] = t; +} + +/* Toggle the 32 bit unsigned integer pointed by *p from little endian to + * big endian */ +void memrev32(void *p) { + unsigned char *x = p, t; + + t = x[0]; + x[0] = x[3]; + x[3] = t; + t = x[1]; + x[1] = x[2]; + x[2] = t; +} + +/* Toggle the 64 bit unsigned integer pointed by *p from little endian to + * big endian */ +void memrev64(void *p) { + unsigned char *x = p, t; + + t = x[0]; + x[0] = x[7]; + x[7] = t; + t = x[1]; + x[1] = x[6]; + x[6] = t; + t = x[2]; + x[2] = x[5]; + x[5] = t; + t = x[3]; + x[3] = x[4]; + x[4] = t; +} + +uint16_t intrev16(uint16_t v) { + memrev16(&v); + return v; +} + +uint32_t intrev32(uint32_t v) { + memrev32(&v); + return v; +} + +uint64_t intrev64(uint64_t v) { + memrev64(&v); + return v; +} + +#ifdef TESTMAIN +#include + +int main(void) { + char buf[32]; + + sprintf(buf,"ciaoroma"); + memrev16(buf); + printf("%s\n", buf); + + sprintf(buf,"ciaoroma"); + memrev32(buf); + printf("%s\n", buf); + + sprintf(buf,"ciaoroma"); + memrev64(buf); + printf("%s\n", buf); + + return 0; +} +#endif diff --git a/src/endianconv.h b/src/endianconv.h new file mode 100644 index 00000000..f76e0e6b --- /dev/null +++ b/src/endianconv.h @@ -0,0 +1,63 @@ +/* See endianconv.c top comments for more information + * + * ---------------------------------------------------------------------------- + * + * Copyright (c) 2011-2012, Salvatore Sanfilippo + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * * Redistributions of source code must retain the above copyright notice, + * this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * * Neither the name of Redis nor the names of its contributors may be used + * to endorse or promote products derived from this software without + * specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" + * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE + * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR + * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF + * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS + * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN + * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) + * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE + * POSSIBILITY OF SUCH DAMAGE. + */ + +#ifndef __ENDIANCONV_H +#define __ENDIANCONV_H + +#include + +void memrev16(void *p); +void memrev32(void *p); +void memrev64(void *p); +uint16_t intrev16(uint16_t v); +uint32_t intrev32(uint32_t v); +uint64_t intrev64(uint64_t v); + +/* variants of the function doing the actual convertion only if the target + * host is big endian */ +#if (BYTE_ORDER == LITTLE_ENDIAN) +#define memrev16ifbe(p) +#define memrev32ifbe(p) +#define memrev64ifbe(p) +#define intrev16ifbe(v) (v) +#define intrev32ifbe(v) (v) +#define intrev64ifbe(v) (v) +#else +#define memrev16ifbe(p) memrev16(p) +#define memrev32ifbe(p) memrev32(p) +#define memrev64ifbe(p) memrev64(p) +#define intrev16ifbe(v) intrev16(v) +#define intrev32ifbe(v) intrev32(v) +#define intrev64ifbe(v) intrev64(v) +#endif + +#endif diff --git a/src/intset.c b/src/intset.c index 4dd0141d..225f0e92 100644 --- a/src/intset.c +++ b/src/intset.c @@ -3,7 +3,7 @@ #include #include "intset.h" #include "zmalloc.h" -#include "endian.h" +#include "endianconv.h" /* Note that these encodings are ordered, so: * INTSET_ENC_INT16 < INTSET_ENC_INT32 < INTSET_ENC_INT64. */ @@ -44,15 +44,17 @@ static int64_t _intsetGetEncoded(intset *is, int pos, uint8_t enc) { /* Return the value at pos, using the configured encoding. */ static int64_t _intsetGet(intset *is, int pos) { - return _intsetGetEncoded(is,pos,is->encoding); + return _intsetGetEncoded(is,pos,intrev32ifbe(is->encoding)); } /* Set the value at pos, using the configured encoding. */ static void _intsetSet(intset *is, int pos, int64_t value) { - if (is->encoding == INTSET_ENC_INT64) { + uint32_t encoding = intrev32ifbe(is->encoding); + + if (encoding == INTSET_ENC_INT64) { ((int64_t*)is->contents)[pos] = value; memrev64ifbe(((int64_t*)is->contents)+pos); - } else if (is->encoding == INTSET_ENC_INT32) { + } else if (encoding == INTSET_ENC_INT32) { ((int32_t*)is->contents)[pos] = value; memrev32ifbe(((int32_t*)is->contents)+pos); } else { @@ -64,14 +66,14 @@ static void _intsetSet(intset *is, int pos, int64_t value) { /* Create an empty intset. */ intset *intsetNew(void) { intset *is = zmalloc(sizeof(intset)); - is->encoding = INTSET_ENC_INT16; + is->encoding = intrev32ifbe(INTSET_ENC_INT16); is->length = 0; return is; } /* Resize the intset */ static intset *intsetResize(intset *is, uint32_t len) { - uint32_t size = len*is->encoding; + uint32_t size = len*intrev32ifbe(is->encoding); is = zrealloc(is,sizeof(intset)+size); return is; } @@ -81,18 +83,18 @@ static intset *intsetResize(intset *is, uint32_t len) { * the value is not present in the intset and sets "pos" to the position * where "value" can be inserted. */ static uint8_t intsetSearch(intset *is, int64_t value, uint32_t *pos) { - int min = 0, max = is->length-1, mid = -1; + int min = 0, max = intrev32ifbe(is->length)-1, mid = -1; int64_t cur = -1; /* The value can never be found when the set is empty */ - if (is->length == 0) { + if (intrev32ifbe(is->length) == 0) { if (pos) *pos = 0; return 0; } else { /* Check for the case where we know we cannot find the value, * but do know the insert position. */ - if (value > _intsetGet(is,is->length-1)) { - if (pos) *pos = is->length; + if (value > _intsetGet(is,intrev32ifbe(is->length)-1)) { + if (pos) *pos = intrev32ifbe(is->length); return 0; } else if (value < _intsetGet(is,0)) { if (pos) *pos = 0; @@ -123,14 +125,14 @@ static uint8_t intsetSearch(intset *is, int64_t value, uint32_t *pos) { /* Upgrades the intset to a larger encoding and inserts the given integer. */ static intset *intsetUpgradeAndAdd(intset *is, int64_t value) { - uint8_t curenc = is->encoding; + uint8_t curenc = intrev32ifbe(is->encoding); uint8_t newenc = _intsetValueEncoding(value); - int length = is->length; + int length = intrev32ifbe(is->length); int prepend = value < 0 ? 1 : 0; /* First set new encoding and resize */ - is->encoding = newenc; - is = intsetResize(is,is->length+1); + is->encoding = intrev32ifbe(newenc); + is = intsetResize(is,intrev32ifbe(is->length)+1); /* Upgrade back-to-front so we don't overwrite values. * Note that the "prepend" variable is used to make sure we have an empty @@ -142,19 +144,21 @@ static intset *intsetUpgradeAndAdd(intset *is, int64_t value) { if (prepend) _intsetSet(is,0,value); else - _intsetSet(is,is->length,value); - is->length++; + _intsetSet(is,intrev32ifbe(is->length),value); + is->length = intrev32ifbe(intrev32ifbe(is->length)+1); return is; } static void intsetMoveTail(intset *is, uint32_t from, uint32_t to) { void *src, *dst; - uint32_t bytes = is->length-from; - if (is->encoding == INTSET_ENC_INT64) { + uint32_t bytes = intrev32ifbe(is->length)-from; + uint32_t encoding = intrev32ifbe(is->encoding); + + if (encoding == INTSET_ENC_INT64) { src = (int64_t*)is->contents+from; dst = (int64_t*)is->contents+to; bytes *= sizeof(int64_t); - } else if (is->encoding == INTSET_ENC_INT32) { + } else if (encoding == INTSET_ENC_INT32) { src = (int32_t*)is->contents+from; dst = (int32_t*)is->contents+to; bytes *= sizeof(int32_t); @@ -175,7 +179,7 @@ intset *intsetAdd(intset *is, int64_t value, uint8_t *success) { /* Upgrade encoding if necessary. If we need to upgrade, we know that * this value should be either appended (if > 0) or prepended (if < 0), * because it lies outside the range of existing values. */ - if (valenc > is->encoding) { + if (valenc > intrev32ifbe(is->encoding)) { /* This always succeeds, so we don't need to curry *success. */ return intsetUpgradeAndAdd(is,value); } else { @@ -187,12 +191,12 @@ intset *intsetAdd(intset *is, int64_t value, uint8_t *success) { return is; } - is = intsetResize(is,is->length+1); - if (pos < is->length) intsetMoveTail(is,pos,pos+1); + is = intsetResize(is,intrev32ifbe(is->length)+1); + if (pos < intrev32ifbe(is->length)) intsetMoveTail(is,pos,pos+1); } _intsetSet(is,pos,value); - is->length++; + is->length = intrev32ifbe(intrev32ifbe(is->length)+1); return is; } @@ -202,14 +206,16 @@ intset *intsetRemove(intset *is, int64_t value, int *success) { uint32_t pos; if (success) *success = 0; - if (valenc <= is->encoding && intsetSearch(is,value,&pos)) { + if (valenc <= intrev32ifbe(is->encoding) && intsetSearch(is,value,&pos)) { + uint32_t len = intrev32ifbe(is->length); + /* We know we can delete */ if (success) *success = 1; /* Overwrite value with tail and update length */ - if (pos < (is->length-1)) intsetMoveTail(is,pos+1,pos); - is = intsetResize(is,is->length-1); - is->length--; + if (pos < (len-1)) intsetMoveTail(is,pos+1,pos); + is = intsetResize(is,len-1); + is->length = intrev32ifbe(len-1); } return is; } @@ -217,18 +223,18 @@ intset *intsetRemove(intset *is, int64_t value, int *success) { /* Determine whether a value belongs to this set */ uint8_t intsetFind(intset *is, int64_t value) { uint8_t valenc = _intsetValueEncoding(value); - return valenc <= is->encoding && intsetSearch(is,value,NULL); + return valenc <= intrev32ifbe(is->encoding) && intsetSearch(is,value,NULL); } /* Return random member */ int64_t intsetRandom(intset *is) { - return _intsetGet(is,rand()%is->length); + return _intsetGet(is,rand()%intrev32ifbe(is->length)); } /* Sets the value to the value at the given position. When this position is * out of range the function returns 0, when in range it returns 1. */ uint8_t intsetGet(intset *is, uint32_t pos, int64_t *value) { - if (pos < is->length) { + if (pos < intrev32ifbe(is->length)) { *value = _intsetGet(is,pos); return 1; } @@ -237,12 +243,12 @@ uint8_t intsetGet(intset *is, uint32_t pos, int64_t *value) { /* Return intset length */ uint32_t intsetLen(intset *is) { - return is->length; + return intrev32ifbe(is->length); } /* Return intset blob size in bytes. */ size_t intsetBlobLen(intset *is) { - return sizeof(intset)+is->length*is->encoding; + return sizeof(intset)+intrev32ifbe(is->length)*intrev32ifbe(is->encoding); } #ifdef INTSET_TEST_MAIN @@ -250,7 +256,7 @@ size_t intsetBlobLen(intset *is) { void intsetRepr(intset *is) { int i; - for (i = 0; i < is->length; i++) { + for (i = 0; i < intrev32ifbe(is->length); i++) { printf("%lld\n", (uint64_t)_intsetGet(is,i)); } printf("\n"); @@ -296,11 +302,13 @@ intset *createSet(int bits, int size) { void checkConsistency(intset *is) { int i; - for (i = 0; i < (is->length-1); i++) { - if (is->encoding == INTSET_ENC_INT16) { + for (i = 0; i < (intrev32ifbe(is->length)-1); i++) { + uint32_t encoding = intrev32ifbe(is->encoding); + + if (encoding == INTSET_ENC_INT16) { int16_t *i16 = (int16_t*)is->contents; assert(i16[i] < i16[i+1]); - } else if (is->encoding == INTSET_ENC_INT32) { + } else if (encoding == INTSET_ENC_INT32) { int32_t *i32 = (int32_t*)is->contents; assert(i32[i] < i32[i+1]); } else { @@ -346,7 +354,7 @@ int main(int argc, char **argv) { is = intsetAdd(is,rand()%0x800,&success); if (success) inserts++; } - assert(is->length == inserts); + assert(intrev32ifbe(is->length) == inserts); checkConsistency(is); ok(); } @@ -354,18 +362,18 @@ int main(int argc, char **argv) { printf("Upgrade from int16 to int32: "); { is = intsetNew(); is = intsetAdd(is,32,NULL); - assert(is->encoding == INTSET_ENC_INT16); + assert(intrev32ifbe(is->encoding) == INTSET_ENC_INT16); is = intsetAdd(is,65535,NULL); - assert(is->encoding == INTSET_ENC_INT32); + assert(intrev32ifbe(is->encoding) == INTSET_ENC_INT32); assert(intsetFind(is,32)); assert(intsetFind(is,65535)); checkConsistency(is); is = intsetNew(); is = intsetAdd(is,32,NULL); - assert(is->encoding == INTSET_ENC_INT16); + assert(intrev32ifbe(is->encoding) == INTSET_ENC_INT16); is = intsetAdd(is,-65535,NULL); - assert(is->encoding == INTSET_ENC_INT32); + assert(intrev32ifbe(is->encoding) == INTSET_ENC_INT32); assert(intsetFind(is,32)); assert(intsetFind(is,-65535)); checkConsistency(is); @@ -375,18 +383,18 @@ int main(int argc, char **argv) { printf("Upgrade from int16 to int64: "); { is = intsetNew(); is = intsetAdd(is,32,NULL); - assert(is->encoding == INTSET_ENC_INT16); + assert(intrev32ifbe(is->encoding) == INTSET_ENC_INT16); is = intsetAdd(is,4294967295,NULL); - assert(is->encoding == INTSET_ENC_INT64); + assert(intrev32ifbe(is->encoding) == INTSET_ENC_INT64); assert(intsetFind(is,32)); assert(intsetFind(is,4294967295)); checkConsistency(is); is = intsetNew(); is = intsetAdd(is,32,NULL); - assert(is->encoding == INTSET_ENC_INT16); + assert(intrev32ifbe(is->encoding) == INTSET_ENC_INT16); is = intsetAdd(is,-4294967295,NULL); - assert(is->encoding == INTSET_ENC_INT64); + assert(intrev32ifbe(is->encoding) == INTSET_ENC_INT64); assert(intsetFind(is,32)); assert(intsetFind(is,-4294967295)); checkConsistency(is); @@ -396,18 +404,18 @@ int main(int argc, char **argv) { printf("Upgrade from int32 to int64: "); { is = intsetNew(); is = intsetAdd(is,65535,NULL); - assert(is->encoding == INTSET_ENC_INT32); + assert(intrev32ifbe(is->encoding) == INTSET_ENC_INT32); is = intsetAdd(is,4294967295,NULL); - assert(is->encoding == INTSET_ENC_INT64); + assert(intrev32ifbe(is->encoding) == INTSET_ENC_INT64); assert(intsetFind(is,65535)); assert(intsetFind(is,4294967295)); checkConsistency(is); is = intsetNew(); is = intsetAdd(is,65535,NULL); - assert(is->encoding == INTSET_ENC_INT32); + assert(intrev32ifbe(is->encoding) == INTSET_ENC_INT32); is = intsetAdd(is,-4294967295,NULL); - assert(is->encoding == INTSET_ENC_INT64); + assert(intrev32ifbe(is->encoding) == INTSET_ENC_INT64); assert(intsetFind(is,65535)); assert(intsetFind(is,-4294967295)); checkConsistency(is); diff --git a/src/multi.c b/src/multi.c index 1504bb09..65ec38a8 100644 --- a/src/multi.c +++ b/src/multi.c @@ -112,7 +112,7 @@ void execCommand(redisClient *c) { c->argc = c->mstate.commands[j].argc; c->argv = c->mstate.commands[j].argv; c->cmd = c->mstate.commands[j].cmd; - call(c); + call(c,REDIS_CALL_FULL); /* Commands may alter argc/argv, restore mstate. */ c->mstate.commands[j].argc = c->argc; diff --git a/src/networking.c b/src/networking.c index 8a025db3..d8fb8132 100644 --- a/src/networking.c +++ b/src/networking.c @@ -3,6 +3,14 @@ static void setProtocolError(redisClient *c, int pos); +/* To evaluate the output buffer size of a client we need to get size of + * allocated objects, however we can't used zmalloc_size() directly on sds + * strings because of the trick they use to work (the header is before the + * returned pointer), so we use this helper function. */ +size_t zmalloc_size_sds(sds s) { + return zmalloc_size(s-sizeof(struct sdshdr)); +} + void *dupClientReplyValue(void *o) { incrRefCount((robj*)o); return o; @@ -47,6 +55,8 @@ redisClient *createClient(int fd) { c->authenticated = 0; c->replstate = REDIS_REPL_NONE; c->reply = listCreate(); + c->reply_bytes = 0; + c->obuf_soft_limit_reached_time = 0; listSetFreeMethod(c->reply,decrRefCount); listSetDupMethod(c->reply,dupClientReplyValue); c->bpop.keys = NULL; @@ -65,11 +75,23 @@ redisClient *createClient(int fd) { return c; } -/* Set the event loop to listen for write events on the client's socket. - * Typically gets called every time a reply is built. */ -int _installWriteEvent(redisClient *c) { +/* This function is called every time we are going to transmit new data + * to the client. The behavior is the following: + * + * If the client should receive new data (normal clients will) the function + * returns REDIS_OK, and make sure to install the write handler in our event + * loop so that when the socket is writable new data gets written. + * + * If the client should not receive new data, because it is a fake client + * or a slave, or because the setup of the write handler failed, the function + * returns REDIS_ERR. + * + * Typically gets called every time a reply is built, before adding more + * data to the clients output buffers. If the function returns REDIS_ERR no + * data should be appended to the output buffers. */ +int prepareClientToWrite(redisClient *c) { if (c->flags & REDIS_LUA_CLIENT) return REDIS_OK; - if (c->fd <= 0) return REDIS_ERR; + if (c->fd <= 0) return REDIS_ERR; /* Fake client */ if (c->bufpos == 0 && listLength(c->reply) == 0 && (c->replstate == REDIS_REPL_NONE || c->replstate == REDIS_REPL_ONLINE) && @@ -123,6 +145,7 @@ void _addReplyObjectToList(redisClient *c, robj *o) { if (listLength(c->reply) == 0) { incrRefCount(o); listAddNodeTail(c->reply,o); + c->reply_bytes += zmalloc_size_sds(o->ptr); } else { tail = listNodeValue(listLast(c->reply)); @@ -130,13 +153,17 @@ void _addReplyObjectToList(redisClient *c, robj *o) { if (tail->ptr != NULL && sdslen(tail->ptr)+sdslen(o->ptr) <= REDIS_REPLY_CHUNK_BYTES) { + c->reply_bytes -= zmalloc_size_sds(tail->ptr); tail = dupLastObjectIfNeeded(c->reply); tail->ptr = sdscatlen(tail->ptr,o->ptr,sdslen(o->ptr)); + c->reply_bytes += zmalloc_size_sds(tail->ptr); } else { incrRefCount(o); listAddNodeTail(c->reply,o); + c->reply_bytes += zmalloc_size_sds(o->ptr); } } + asyncCloseClientOnOutputBufferLimitReached(c); } /* This method takes responsibility over the sds. When it is no longer @@ -151,6 +178,7 @@ void _addReplySdsToList(redisClient *c, sds s) { if (listLength(c->reply) == 0) { listAddNodeTail(c->reply,createObject(REDIS_STRING,s)); + c->reply_bytes += zmalloc_size_sds(s); } else { tail = listNodeValue(listLast(c->reply)); @@ -158,13 +186,17 @@ void _addReplySdsToList(redisClient *c, sds s) { if (tail->ptr != NULL && sdslen(tail->ptr)+sdslen(s) <= REDIS_REPLY_CHUNK_BYTES) { + c->reply_bytes -= zmalloc_size_sds(tail->ptr); tail = dupLastObjectIfNeeded(c->reply); tail->ptr = sdscatlen(tail->ptr,s,sdslen(s)); + c->reply_bytes += zmalloc_size_sds(tail->ptr); sdsfree(s); } else { listAddNodeTail(c->reply,createObject(REDIS_STRING,s)); + c->reply_bytes += zmalloc_size_sds(s); } } + asyncCloseClientOnOutputBufferLimitReached(c); } void _addReplyStringToList(redisClient *c, char *s, size_t len) { @@ -173,7 +205,10 @@ void _addReplyStringToList(redisClient *c, char *s, size_t len) { if (c->flags & REDIS_CLOSE_AFTER_REPLY) return; if (listLength(c->reply) == 0) { - listAddNodeTail(c->reply,createStringObject(s,len)); + robj *o = createStringObject(s,len); + + listAddNodeTail(c->reply,o); + c->reply_bytes += zmalloc_size_sds(o->ptr); } else { tail = listNodeValue(listLast(c->reply)); @@ -181,12 +216,18 @@ void _addReplyStringToList(redisClient *c, char *s, size_t len) { if (tail->ptr != NULL && sdslen(tail->ptr)+len <= REDIS_REPLY_CHUNK_BYTES) { + c->reply_bytes -= zmalloc_size_sds(tail->ptr); tail = dupLastObjectIfNeeded(c->reply); tail->ptr = sdscatlen(tail->ptr,s,len); + c->reply_bytes += zmalloc_size_sds(tail->ptr); } else { - listAddNodeTail(c->reply,createStringObject(s,len)); + robj *o = createStringObject(s,len); + + listAddNodeTail(c->reply,o); + c->reply_bytes += zmalloc_size_sds(o->ptr); } } + asyncCloseClientOnOutputBufferLimitReached(c); } /* ----------------------------------------------------------------------------- @@ -195,7 +236,7 @@ void _addReplyStringToList(redisClient *c, char *s, size_t len) { * -------------------------------------------------------------------------- */ void addReply(redisClient *c, robj *obj) { - if (_installWriteEvent(c) != REDIS_OK) return; + if (prepareClientToWrite(c) != REDIS_OK) return; /* This is an important place where we can avoid copy-on-write * when there is a saving child running, avoiding touching the @@ -207,19 +248,31 @@ void addReply(redisClient *c, robj *obj) { if (obj->encoding == REDIS_ENCODING_RAW) { if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK) _addReplyObjectToList(c,obj); - } else { - /* FIXME: convert the long into string and use _addReplyToBuffer() - * instead of calling getDecodedObject. As this place in the - * code is too performance critical. */ + } else if (obj->encoding == REDIS_ENCODING_INT) { + /* Optimization: if there is room in the static buffer for 32 bytes + * (more than the max chars a 64 bit integer can take as string) we + * avoid decoding the object and go for the lower level approach. */ + if (listLength(c->reply) == 0 && (sizeof(c->buf) - c->bufpos) >= 32) { + char buf[32]; + int len; + + len = ll2string(buf,sizeof(buf),(long)obj->ptr); + if (_addReplyToBuffer(c,buf,len) == REDIS_OK) + return; + /* else... continue with the normal code path, but should never + * happen actually since we verified there is room. */ + } obj = getDecodedObject(obj); if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK) _addReplyObjectToList(c,obj); decrRefCount(obj); + } else { + redisPanic("Wrong obj->encoding in addReply()"); } } void addReplySds(redisClient *c, sds s) { - if (_installWriteEvent(c) != REDIS_OK) { + if (prepareClientToWrite(c) != REDIS_OK) { /* The caller expects the sds to be free'd. */ sdsfree(s); return; @@ -233,19 +286,19 @@ void addReplySds(redisClient *c, sds s) { } void addReplyString(redisClient *c, char *s, size_t len) { - if (_installWriteEvent(c) != REDIS_OK) return; + if (prepareClientToWrite(c) != REDIS_OK) return; if (_addReplyToBuffer(c,s,len) != REDIS_OK) _addReplyStringToList(c,s,len); } -void _addReplyError(redisClient *c, char *s, size_t len) { +void addReplyErrorLength(redisClient *c, char *s, size_t len) { addReplyString(c,"-ERR ",5); addReplyString(c,s,len); addReplyString(c,"\r\n",2); } void addReplyError(redisClient *c, char *err) { - _addReplyError(c,err,strlen(err)); + addReplyErrorLength(c,err,strlen(err)); } void addReplyErrorFormat(redisClient *c, const char *fmt, ...) { @@ -260,18 +313,18 @@ void addReplyErrorFormat(redisClient *c, const char *fmt, ...) { for (j = 0; j < l; j++) { if (s[j] == '\r' || s[j] == '\n') s[j] = ' '; } - _addReplyError(c,s,sdslen(s)); + addReplyErrorLength(c,s,sdslen(s)); sdsfree(s); } -void _addReplyStatus(redisClient *c, char *s, size_t len) { +void addReplyStatusLength(redisClient *c, char *s, size_t len) { addReplyString(c,"+",1); addReplyString(c,s,len); addReplyString(c,"\r\n",2); } void addReplyStatus(redisClient *c, char *status) { - _addReplyStatus(c,status,strlen(status)); + addReplyStatusLength(c,status,strlen(status)); } void addReplyStatusFormat(redisClient *c, const char *fmt, ...) { @@ -279,7 +332,7 @@ void addReplyStatusFormat(redisClient *c, const char *fmt, ...) { va_start(ap,fmt); sds s = sdscatvprintf(sdsempty(),fmt,ap); va_end(ap); - _addReplyStatus(c,s,sdslen(s)); + addReplyStatusLength(c,s,sdslen(s)); sdsfree(s); } @@ -289,7 +342,7 @@ void *addDeferredMultiBulkLength(redisClient *c) { /* Note that we install the write event here even if the object is not * ready to be sent, since we are sure that before returning to the * event loop setDeferredMultiBulkLength() will be called. */ - if (_installWriteEvent(c) != REDIS_OK) return NULL; + if (prepareClientToWrite(c) != REDIS_OK) return NULL; listAddNodeTail(c->reply,createObject(REDIS_STRING,NULL)); return listLast(c->reply); } @@ -304,6 +357,7 @@ void setDeferredMultiBulkLength(redisClient *c, void *node, long length) { len = listNodeValue(ln); len->ptr = sdscatprintf(sdsempty(),"*%ld\r\n",length); + c->reply_bytes += zmalloc_size_sds(len->ptr); if (ln->next != NULL) { next = listNodeValue(ln->next); @@ -313,6 +367,7 @@ void setDeferredMultiBulkLength(redisClient *c, void *node, long length) { listDelNode(c->reply,ln->next); } } + asyncCloseClientOnOutputBufferLimitReached(c); } /* Add a duble as a bulk reply */ @@ -326,9 +381,21 @@ void addReplyDouble(redisClient *c, double d) { /* Add a long long as integer reply or bulk len / multi bulk count. * Basically this is used to output . */ -void _addReplyLongLong(redisClient *c, long long ll, char prefix) { +void addReplyLongLongWithPrefix(redisClient *c, long long ll, char prefix) { char buf[128]; int len; + + /* Things like $3\r\n or *2\r\n are emitted very often by the protocol + * so we have a few shared objects to use if the integer is small + * like it is most of the times. */ + if (prefix == '*' && ll < REDIS_SHARED_BULKHDR_LEN) { + addReply(c,shared.mbulkhdr[ll]); + return; + } else if (prefix == '$' && ll < REDIS_SHARED_BULKHDR_LEN) { + addReply(c,shared.bulkhdr[ll]); + return; + } + buf[0] = prefix; len = ll2string(buf+1,sizeof(buf)-1,ll); buf[len+1] = '\r'; @@ -342,11 +409,11 @@ void addReplyLongLong(redisClient *c, long long ll) { else if (ll == 1) addReply(c,shared.cone); else - _addReplyLongLong(c,ll,':'); + addReplyLongLongWithPrefix(c,ll,':'); } void addReplyMultiBulkLen(redisClient *c, long length) { - _addReplyLongLong(c,length,'*'); + addReplyLongLongWithPrefix(c,length,'*'); } /* Create the length prefix of a bulk reply, example: $2234 */ @@ -368,7 +435,7 @@ void addReplyBulkLen(redisClient *c, robj *obj) { len++; } } - _addReplyLongLong(c,len,'$'); + addReplyLongLongWithPrefix(c,len,'$'); } /* Add a Redis Object as a bulk reply */ @@ -380,7 +447,7 @@ void addReplyBulk(redisClient *c, robj *obj) { /* Add a C buffer as bulk reply */ void addReplyBulkCBuffer(redisClient *c, void *p, size_t len) { - _addReplyLongLong(c,len,'$'); + addReplyLongLongWithPrefix(c,len,'$'); addReplyString(c,p,len); addReply(c,shared.crlf); } @@ -411,6 +478,7 @@ void copyClientOutputBuffer(redisClient *dst, redisClient *src) { dst->reply = listDup(src->reply); memcpy(dst->buf,src->buf,src->bufpos); dst->bufpos = src->bufpos; + dst->reply_bytes = src->reply_bytes; } static void acceptCommonHandler(int fd) { @@ -552,15 +620,46 @@ void freeClient(redisClient *c) { } } } + + /* If this client was scheduled for async freeing we need to remove it + * from the queue. */ + if (c->flags & REDIS_CLOSE_ASAP) { + ln = listSearchKey(server.clients_to_close,c); + redisAssert(ln != NULL); + listDelNode(server.clients_to_close,ln); + } + /* Release memory */ zfree(c->argv); freeClientMultiState(c); zfree(c); } +/* Schedule a client to free it at a safe time in the serverCron() function. + * This function is useful when we need to terminate a client but we are in + * a context where calling freeClient() is not possible, because the client + * should be valid for the continuation of the flow of the program. */ +void freeClientAsync(redisClient *c) { + if (c->flags & REDIS_CLOSE_ASAP) return; + c->flags |= REDIS_CLOSE_ASAP; + listAddNodeTail(server.clients_to_close,c); +} + +void freeClientsInAsyncFreeQueue(void) { + while (listLength(server.clients_to_close)) { + listNode *ln = listFirst(server.clients_to_close); + redisClient *c = listNodeValue(ln); + + c->flags &= ~REDIS_CLOSE_ASAP; + freeClient(c); + listDelNode(server.clients_to_close,ln); + } +} + void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) { redisClient *c = privdata; int nwritten = 0, totwritten = 0, objlen; + size_t objmem; robj *o; REDIS_NOTUSED(el); REDIS_NOTUSED(mask); @@ -586,6 +685,7 @@ void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) { } else { o = listNodeValue(listFirst(c->reply)); objlen = sdslen(o->ptr); + objmem = zmalloc_size_sds(o->ptr); if (objlen == 0) { listDelNode(c->reply,listFirst(c->reply)); @@ -606,14 +706,20 @@ void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) { if (c->sentlen == objlen) { listDelNode(c->reply,listFirst(c->reply)); c->sentlen = 0; + c->reply_bytes -= objmem; } } - /* Note that we avoid to send more thank REDIS_MAX_WRITE_PER_EVENT + /* Note that we avoid to send more than REDIS_MAX_WRITE_PER_EVENT * bytes, in a single threaded server it's a good idea to serve * other clients as well, even if a very large request comes from * super fast link that is always able to accept data (in real world - * scenario think about 'KEYS *' against the loopback interfae) */ - if (totwritten > REDIS_MAX_WRITE_PER_EVENT) break; + * scenario think about 'KEYS *' against the loopback interface). + * + * However if we are over the maxmemory limit we ignore that and + * just deliver as much data as it is possible to deliver. */ + if (totwritten > REDIS_MAX_WRITE_PER_EVENT && + (server.maxmemory == 0 || + zmalloc_used_memory() < server.maxmemory)) break; } if (nwritten == -1) { if (errno == EAGAIN) { @@ -999,6 +1105,7 @@ sds getClientInfoString(redisClient *client) { if (client->flags & REDIS_DIRTY_CAS) *p++ = 'd'; if (client->flags & REDIS_CLOSE_AFTER_REPLY) *p++ = 'c'; if (client->flags & REDIS_UNBLOCKED) *p++ = 'u'; + if (client->flags & REDIS_CLOSE_ASAP) *p++ = 'A'; if (p == flags) *p++ = 'N'; *p++ = '\0'; @@ -1008,7 +1115,7 @@ sds getClientInfoString(redisClient *client) { if (emask & AE_WRITABLE) *p++ = 'w'; *p = '\0'; return sdscatprintf(sdsempty(), - "addr=%s:%d fd=%d idle=%ld flags=%s db=%d sub=%d psub=%d qbuf=%lu obl=%lu oll=%lu events=%s cmd=%s", + "addr=%s:%d fd=%d idle=%ld flags=%s db=%d sub=%d psub=%d qbuf=%lu obl=%lu oll=%lu omem=%lu events=%s cmd=%s", ip,port,client->fd, (long)(now - client->lastinteraction), flags, @@ -1018,6 +1125,7 @@ sds getClientInfoString(redisClient *client) { (unsigned long) sdslen(client->querybuf), (unsigned long) client->bufpos, (unsigned long) listLength(client->reply), + getClientOutputBufferMemoryUsage(client), events, client->lastcmd ? client->lastcmd->name : "NULL"); } @@ -1122,3 +1230,132 @@ void rewriteClientCommandArgument(redisClient *c, int i, robj *newval) { redisAssertWithInfo(c,NULL,c->cmd != NULL); } } + +/* This function returns the number of bytes that Redis is virtually + * using to store the reply still not read by the client. + * It is "virtual" since the reply output list may contain objects that + * are shared and are not really using additional memory. + * + * The function returns the total sum of the length of all the objects + * stored in the output list, plus the memory used to allocate every + * list node. The static reply buffer is not taken into account since it + * is allocated anyway. + * + * Note: this function is very fast so can be called as many time as + * the caller wishes. The main usage of this function currently is + * enforcing the client output length limits. */ +unsigned long getClientOutputBufferMemoryUsage(redisClient *c) { + unsigned long list_item_size = sizeof(listNode)+sizeof(robj); + + return c->reply_bytes + (list_item_size*listLength(c->reply)); +} + +/* Get the class of a client, used in order to envorce limits to different + * classes of clients. + * + * The function will return one of the following: + * REDIS_CLIENT_LIMIT_CLASS_NORMAL -> Normal client + * REDIS_CLIENT_LIMIT_CLASS_SLAVE -> Slave or client executing MONITOR command + * REDIS_CLIENT_LIMIT_CLASS_PUBSUB -> Client subscribed to Pub/Sub channels + */ +int getClientLimitClass(redisClient *c) { + if (c->flags & REDIS_SLAVE) return REDIS_CLIENT_LIMIT_CLASS_SLAVE; + if (dictSize(c->pubsub_channels) || listLength(c->pubsub_patterns)) + return REDIS_CLIENT_LIMIT_CLASS_PUBSUB; + return REDIS_CLIENT_LIMIT_CLASS_NORMAL; +} + +int getClientLimitClassByName(char *name) { + if (!strcasecmp(name,"normal")) return REDIS_CLIENT_LIMIT_CLASS_NORMAL; + else if (!strcasecmp(name,"slave")) return REDIS_CLIENT_LIMIT_CLASS_SLAVE; + else if (!strcasecmp(name,"pubsub")) return REDIS_CLIENT_LIMIT_CLASS_PUBSUB; + else return -1; +} + +char *getClientLimitClassName(int class) { + switch(class) { + case REDIS_CLIENT_LIMIT_CLASS_NORMAL: return "normal"; + case REDIS_CLIENT_LIMIT_CLASS_SLAVE: return "slave"; + case REDIS_CLIENT_LIMIT_CLASS_PUBSUB: return "pubsub"; + default: return NULL; + } +} + +/* The function checks if the client reached output buffer soft or hard + * limit, and also update the state needed to check the soft limit as + * a side effect. + * + * Return value: non-zero if the client reached the soft or the hard limit. + * Otherwise zero is returned. */ +int checkClientOutputBufferLimits(redisClient *c) { + int soft = 0, hard = 0, class; + unsigned long used_mem = getClientOutputBufferMemoryUsage(c); + + class = getClientLimitClass(c); + if (server.client_obuf_limits[class].hard_limit_bytes && + used_mem >= server.client_obuf_limits[class].hard_limit_bytes) + hard = 1; + if (server.client_obuf_limits[class].soft_limit_bytes && + used_mem >= server.client_obuf_limits[class].soft_limit_bytes) + soft = 1; + + /* We need to check if the soft limit is reached continuously for the + * specified amount of seconds. */ + if (soft) { + if (c->obuf_soft_limit_reached_time == 0) { + c->obuf_soft_limit_reached_time = server.unixtime; + soft = 0; /* First time we see the soft limit reached */ + } else { + time_t elapsed = server.unixtime - c->obuf_soft_limit_reached_time; + + if (elapsed <= + server.client_obuf_limits[class].soft_limit_seconds) { + soft = 0; /* The client still did not reached the max number of + seconds for the soft limit to be considered + reached. */ + } + } + } else { + c->obuf_soft_limit_reached_time = 0; + } + return soft || hard; +} + +/* Asynchronously close a client if soft or hard limit is reached on the + * output buffer size. The caller can check if the client will be closed + * checking if the client REDIS_CLOSE_ASAP flag is set. + * + * Note: we need to close the client asynchronously because this function is + * called from contexts where the client can't be freed safely, i.e. from the + * lower level functions pushing data inside the client output buffers. */ +void asyncCloseClientOnOutputBufferLimitReached(redisClient *c) { + if (c->reply_bytes == 0 || c->flags & REDIS_CLOSE_ASAP) return; + if (checkClientOutputBufferLimits(c)) { + sds client = getClientInfoString(c); + + freeClientAsync(c); + redisLog(REDIS_WARNING,"Client %s scheduled to be closed ASAP for overcoming of output buffer limits.", client); + sdsfree(client); + } +} + +/* Helper function used by freeMemoryIfNeeded() in order to flush slaves + * output buffers without returning control to the event loop. */ +void flushSlavesOutputBuffers(void) { + listIter li; + listNode *ln; + + listRewind(server.slaves,&li); + while((ln = listNext(&li))) { + redisClient *slave = listNodeValue(ln); + int events; + + events = aeGetFileEvents(server.el,slave->fd); + if (events & AE_WRITABLE && + slave->replstate == REDIS_REPL_ONLINE && + listLength(slave->reply)) + { + sendReplyToClient(server.el,slave->fd,slave,0); + } + } +} diff --git a/src/pubsub.c b/src/pubsub.c index 27e6f9a5..984013bb 100644 --- a/src/pubsub.c +++ b/src/pubsub.c @@ -41,7 +41,7 @@ int pubsubSubscribeChannel(redisClient *c, robj *channel) { listAddNodeTail(clients,c); } /* Notify the client */ - addReply(c,shared.mbulk3); + addReply(c,shared.mbulkhdr[3]); addReply(c,shared.subscribebulk); addReplyBulk(c,channel); addReplyLongLong(c,dictSize(c->pubsub_channels)+listLength(c->pubsub_patterns)); @@ -77,7 +77,7 @@ int pubsubUnsubscribeChannel(redisClient *c, robj *channel, int notify) { } /* Notify the client */ if (notify) { - addReply(c,shared.mbulk3); + addReply(c,shared.mbulkhdr[3]); addReply(c,shared.unsubscribebulk); addReplyBulk(c,channel); addReplyLongLong(c,dictSize(c->pubsub_channels)+ @@ -103,7 +103,7 @@ int pubsubSubscribePattern(redisClient *c, robj *pattern) { listAddNodeTail(server.pubsub_patterns,pat); } /* Notify the client */ - addReply(c,shared.mbulk3); + addReply(c,shared.mbulkhdr[3]); addReply(c,shared.psubscribebulk); addReplyBulk(c,pattern); addReplyLongLong(c,dictSize(c->pubsub_channels)+listLength(c->pubsub_patterns)); @@ -128,7 +128,7 @@ int pubsubUnsubscribePattern(redisClient *c, robj *pattern, int notify) { } /* Notify the client */ if (notify) { - addReply(c,shared.mbulk3); + addReply(c,shared.mbulkhdr[3]); addReply(c,shared.punsubscribebulk); addReplyBulk(c,pattern); addReplyLongLong(c,dictSize(c->pubsub_channels)+ @@ -188,7 +188,7 @@ int pubsubPublishMessage(robj *channel, robj *message) { while ((ln = listNext(&li)) != NULL) { redisClient *c = ln->value; - addReply(c,shared.mbulk3); + addReply(c,shared.mbulkhdr[3]); addReply(c,shared.messagebulk); addReplyBulk(c,channel); addReplyBulk(c,message); @@ -206,7 +206,7 @@ int pubsubPublishMessage(robj *channel, robj *message) { sdslen(pat->pattern->ptr), (char*)channel->ptr, sdslen(channel->ptr),0)) { - addReply(pat->client,shared.mbulk4); + addReply(pat->client,shared.mbulkhdr[4]); addReply(pat->client,shared.pmessagebulk); addReplyBulk(pat->client,pat->pattern); addReplyBulk(pat->client,channel); diff --git a/src/rdb.c b/src/rdb.c index 77e2a048..3a10e0ce 100644 --- a/src/rdb.c +++ b/src/rdb.c @@ -1026,8 +1026,12 @@ int rdbLoad(char *filename) { if ((key = rdbLoadStringObject(&rdb)) == NULL) goto eoferr; /* Read value */ if ((val = rdbLoadObject(type,&rdb)) == NULL) goto eoferr; - /* Check if the key already expired */ - if (expiretime != -1 && expiretime < now) { + /* Check if the key already expired. This function is used when loading + * an RDB file from disk, either at startup, or when an RDB was + * received from the master. In the latter case, the master is + * responsible for key expiry. If we would expire keys here, the + * snapshot taken by the master may not be reflected on the slave. */ + if (server.masterhost == NULL && expiretime != -1 && expiretime < now) { decrRefCount(key); decrRefCount(val); continue; diff --git a/src/redis-benchmark.c b/src/redis-benchmark.c index f7ca312d..28daccd4 100644 --- a/src/redis-benchmark.c +++ b/src/redis-benchmark.c @@ -267,7 +267,7 @@ static client createClient(const char *cmd, size_t len) { } } - redisSetReplyObjectFunctions(c->context,NULL); +/* redisSetReplyObjectFunctions(c->context,NULL); */ aeCreateFileEvent(config.el,c->context->fd,AE_WRITABLE,writeHandler,c); listAddNodeTail(config.clients,c); config.liveclients++; @@ -426,10 +426,10 @@ usage: " -k 1=keep alive 0=reconnect (default 1)\n" " -r Use random keys for SET/GET/INCR, random values for SADD\n" " Using this option the benchmark will get/set keys\n" -" in the form mykey_rand000000012456 instead of constant\n" +" in the form mykey_rand:000000012456 instead of constant\n" " keys, the argument determines the max\n" " number of values for the random number. For instance\n" -" if set to 10 only rand000000000000 - rand000000000009\n" +" if set to 10 only rand:000000000000 - rand:000000000009\n" " range will be allowed.\n" " -q Quiet. Just show query/sec values\n" " --csv Output in CSV format\n" @@ -446,6 +446,8 @@ usage: " $ redis-benchmark -t set -n 1000000 -r 100000000\n\n" " Benchmark 127.0.0.1:6379 for a few commands producing CSV output:\n" " $ redis-benchmark -t ping,set,get -n 100000 --csv\n\n" +" Fill a list with 10000 random elements:\n" +" $ redis-benchmark -r 10000 -n 10000 lpush mylist ele:rand:000000000000\n\n" ); exit(exit_status); } diff --git a/src/redis-check-aof.c b/src/redis-check-aof.c index ff0d1f82..3762d05e 100644 --- a/src/redis-check-aof.c +++ b/src/redis-check-aof.c @@ -9,11 +9,11 @@ #define ERROR(...) { \ char __buf[1024]; \ sprintf(__buf, __VA_ARGS__); \ - sprintf(error, "0x%08lx: %s", epos, __buf); \ + sprintf(error, "0x%16llx: %s", (long long)epos, __buf); \ } static char error[1024]; -static long epos; +static off_t epos; int consumeNewline(char *buf) { if (strncmp(buf,"\r\n",2) != 0) { @@ -25,7 +25,7 @@ int consumeNewline(char *buf) { int readLong(FILE *fp, char prefix, long *target) { char buf[128], *eptr; - epos = ftell(fp); + epos = ftello(fp); if (fgets(buf,sizeof(buf),fp) == NULL) { return 0; } @@ -39,7 +39,7 @@ int readLong(FILE *fp, char prefix, long *target) { int readBytes(FILE *fp, char *target, long length) { long real; - epos = ftell(fp); + epos = ftello(fp); real = fread(target,1,length,fp); if (real != length) { ERROR("Expected to read %ld bytes, got %ld bytes",length,real); @@ -72,13 +72,14 @@ int readArgc(FILE *fp, long *target) { return readLong(fp,'*',target); } -long process(FILE *fp) { - long argc, pos = 0; +off_t process(FILE *fp) { + long argc; + off_t pos = 0; int i, multi = 0; char *str; while(1) { - if (!multi) pos = ftell(fp); + if (!multi) pos = ftello(fp); if (!readArgc(fp, &argc)) break; for (i = 0; i < argc; i++) { @@ -148,18 +149,20 @@ int main(int argc, char **argv) { exit(1); } - long size = sb.st_size; + off_t size = sb.st_size; if (size == 0) { printf("Empty file: %s\n", filename); exit(1); } - long pos = process(fp); - long diff = size-pos; + off_t pos = process(fp); + off_t diff = size-pos; + printf("AOF analyzed: size=%lld, ok_up_to=%lld, diff=%lld\n", + (long long) size, (long long) pos, (long long) diff); if (diff > 0) { if (fix) { char buf[2]; - printf("This will shrink the AOF from %ld bytes, with %ld bytes, to %ld bytes\n",size,diff,pos); + printf("This will shrink the AOF from %lld bytes, with %lld bytes, to %lld bytes\n",(long long)size,(long long)diff,(long long)pos); printf("Continue? [y/N]: "); if (fgets(buf,sizeof(buf),stdin) == NULL || strncasecmp(buf,"y",1) != 0) { diff --git a/src/redis-cli.c b/src/redis-cli.c index 3f60a132..827e4061 100644 --- a/src/redis-cli.c +++ b/src/redis-cli.c @@ -68,7 +68,7 @@ static struct config { char *auth; int raw_output; /* output mode per command */ sds mb_delim; - char prompt[32]; + char prompt[128]; char *eval; } config; @@ -91,12 +91,19 @@ static long long mstime(void) { } static void cliRefreshPrompt(void) { - if (config.dbnum == 0) - snprintf(config.prompt,sizeof(config.prompt),"redis %s:%d> ", - config.hostip, config.hostport); + int len; + + if (config.hostsocket != NULL) + len = snprintf(config.prompt,sizeof(config.prompt),"redis %s", + config.hostsocket); else - snprintf(config.prompt,sizeof(config.prompt),"redis %s:%d[%d]> ", - config.hostip, config.hostport, config.dbnum); + len = snprintf(config.prompt,sizeof(config.prompt),"redis %s:%d", + config.hostip, config.hostport); + /* Add [dbnum] if needed */ + if (config.dbnum != 0) + len += snprintf(config.prompt+len,sizeof(config.prompt)-len,"[%d]", + config.dbnum); + snprintf(config.prompt+len,sizeof(config.prompt)-len,"> "); } /*------------------------------------------------------------------------------ @@ -501,6 +508,11 @@ static int cliSendCommand(int argc, char **argv, int repeat) { size_t *argvlen; int j, output_raw; + if (!strcasecmp(command,"help") || !strcasecmp(command,"?")) { + cliOutputHelp(--argc, ++argv); + return REDIS_OK; + } + if (context == NULL) return REDIS_ERR; output_raw = 0; @@ -515,10 +527,6 @@ static int cliSendCommand(int argc, char **argv, int repeat) { output_raw = 1; } - if (!strcasecmp(command,"help") || !strcasecmp(command,"?")) { - cliOutputHelp(--argc, ++argv); - return REDIS_OK; - } if (!strcasecmp(command,"shutdown")) config.shutdown = 1; if (!strcasecmp(command,"monitor")) config.monitor_mode = 1; if (!strcasecmp(command,"subscribe") || diff --git a/src/redis.c b/src/redis.c index 6e897d83..b4acd6ea 100644 --- a/src/redis.c +++ b/src/redis.c @@ -31,11 +31,6 @@ #include "slowlog.h" #include "bio.h" -#ifdef HAVE_BACKTRACE -#include -#include -#endif /* HAVE_BACKTRACE */ - #include #include #include @@ -107,7 +102,10 @@ struct redisCommand *commandTable; * s: command not allowed in scripts. * R: random command. Command is not deterministic, that is, the same command * with the same arguments, with the same key space, may have different - * results. For instance SPOP and RANDOMKEY are two random commands. */ + * results. For instance SPOP and RANDOMKEY are two random commands. + * S: Sort command output array if called from script, so that the output + * is deterministic. + */ struct redisCommand redisCommandTable[] = { {"get",getCommand,2,"r",0,NULL,1,1,1,0,0}, {"set",setCommand,3,"wm",0,noPreloadGetKeys,1,1,1,0,0}, @@ -150,13 +148,13 @@ struct redisCommand redisCommandTable[] = { {"scard",scardCommand,2,"r",0,NULL,1,1,1,0,0}, {"spop",spopCommand,2,"wRs",0,NULL,1,1,1,0,0}, {"srandmember",srandmemberCommand,2,"rR",0,NULL,1,1,1,0,0}, - {"sinter",sinterCommand,-2,"r",0,NULL,1,-1,1,0,0}, + {"sinter",sinterCommand,-2,"rS",0,NULL,1,-1,1,0,0}, {"sinterstore",sinterstoreCommand,-3,"wm",0,NULL,1,-1,1,0,0}, - {"sunion",sunionCommand,-2,"r",0,NULL,1,-1,1,0,0}, + {"sunion",sunionCommand,-2,"rS",0,NULL,1,-1,1,0,0}, {"sunionstore",sunionstoreCommand,-3,"wm",0,NULL,1,-1,1,0,0}, - {"sdiff",sdiffCommand,-2,"r",0,NULL,1,-1,1,0,0}, + {"sdiff",sdiffCommand,-2,"rS",0,NULL,1,-1,1,0,0}, {"sdiffstore",sdiffstoreCommand,-3,"wm",0,NULL,1,-1,1,0,0}, - {"smembers",sinterCommand,2,"r",0,NULL,1,1,1,0,0}, + {"smembers",sinterCommand,2,"rS",0,NULL,1,1,1,0,0}, {"zadd",zaddCommand,-4,"wm",0,NULL,1,1,1,0,0}, {"zincrby",zincrbyCommand,4,"wm",0,NULL,1,1,1,0,0}, {"zrem",zremCommand,-3,"w",0,NULL,1,1,1,0,0}, @@ -182,8 +180,8 @@ struct redisCommand redisCommandTable[] = { {"hincrbyfloat",hincrbyfloatCommand,4,"wm",0,NULL,1,1,1,0,0}, {"hdel",hdelCommand,-3,"w",0,NULL,1,1,1,0,0}, {"hlen",hlenCommand,2,"r",0,NULL,1,1,1,0,0}, - {"hkeys",hkeysCommand,2,"r",0,NULL,1,1,1,0,0}, - {"hvals",hvalsCommand,2,"r",0,NULL,1,1,1,0,0}, + {"hkeys",hkeysCommand,2,"rS",0,NULL,1,1,1,0,0}, + {"hvals",hvalsCommand,2,"rS",0,NULL,1,1,1,0,0}, {"hgetall",hgetallCommand,2,"r",0,NULL,1,1,1,0,0}, {"hexists",hexistsCommand,3,"r",0,NULL,1,1,1,0,0}, {"incrby",incrbyCommand,3,"wm",0,NULL,1,1,1,0,0}, @@ -201,7 +199,7 @@ struct redisCommand redisCommandTable[] = { {"expireat",expireatCommand,3,"w",0,NULL,1,1,1,0,0}, {"pexpire",pexpireCommand,3,"w",0,NULL,1,1,1,0,0}, {"pexpireat",pexpireatCommand,3,"w",0,NULL,1,1,1,0,0}, - {"keys",keysCommand,2,"r",0,NULL,0,0,0,0,0}, + {"keys",keysCommand,2,"rS",0,NULL,0,0,0,0,0}, {"dbsize",dbsizeCommand,1,"r",0,NULL,0,0,0,0,0}, {"auth",authCommand,2,"rs",0,NULL,0,0,0,0,0}, {"ping",pingCommand,1,"r",0,NULL,0,0,0,0,0}, @@ -218,7 +216,7 @@ struct redisCommand redisCommandTable[] = { {"sync",syncCommand,1,"ars",0,NULL,0,0,0,0,0}, {"flushdb",flushdbCommand,1,"w",0,NULL,0,0,0,0,0}, {"flushall",flushallCommand,1,"w",0,NULL,0,0,0,0,0}, - {"sort",sortCommand,-2,"wm",0,NULL,1,1,1,0,0}, + {"sort",sortCommand,-2,"wmS",0,NULL,1,1,1,0,0}, {"info",infoCommand,-1,"r",0,NULL,0,0,0,0,0}, {"monitor",monitorCommand,1,"ars",0,NULL,0,0,0,0,0}, {"ttl",ttlCommand,2,"r",0,NULL,1,1,1,0,0}, @@ -759,6 +757,9 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { * in order to guarantee a strict consistency. */ if (server.masterhost == NULL) activeExpireCycle(); + /* Close clients that need to be closed asynchronous */ + freeClientsInAsyncFreeQueue(); + /* Replication cron function -- used to reconnect to master and * to detect transfer failures. */ if (!(loops % 10)) replicationCron(); @@ -850,15 +851,21 @@ void createSharedObjects(void) { shared.unsubscribebulk = createStringObject("$11\r\nunsubscribe\r\n",18); shared.psubscribebulk = createStringObject("$10\r\npsubscribe\r\n",17); shared.punsubscribebulk = createStringObject("$12\r\npunsubscribe\r\n",19); - shared.mbulk3 = createStringObject("*3\r\n",4); - shared.mbulk4 = createStringObject("*4\r\n",4); + shared.del = createStringObject("DEL",3); for (j = 0; j < REDIS_SHARED_INTEGERS; j++) { shared.integers[j] = createObject(REDIS_STRING,(void*)(long)j); shared.integers[j]->encoding = REDIS_ENCODING_INT; } + for (j = 0; j < REDIS_SHARED_BULKHDR_LEN; j++) { + shared.mbulkhdr[j] = createObject(REDIS_STRING, + sdscatprintf(sdsempty(),"*%d\r\n",j)); + shared.bulkhdr[j] = createObject(REDIS_STRING, + sdscatprintf(sdsempty(),"$%d\r\n",j)); + } } void initServerConfig() { + server.arch_bits = (sizeof(long) == 8) ? 64 : 32; server.port = REDIS_SERVERPORT; server.bindaddr = NULL; server.unixsocket = NULL; @@ -931,6 +938,17 @@ void initServerConfig() { server.repl_serve_stale_data = 1; server.repl_down_since = -1; + /* Client output buffer limits */ + server.client_obuf_limits[REDIS_CLIENT_LIMIT_CLASS_NORMAL].hard_limit_bytes = 0; + server.client_obuf_limits[REDIS_CLIENT_LIMIT_CLASS_NORMAL].soft_limit_bytes = 0; + server.client_obuf_limits[REDIS_CLIENT_LIMIT_CLASS_NORMAL].soft_limit_seconds = 0; + server.client_obuf_limits[REDIS_CLIENT_LIMIT_CLASS_SLAVE].hard_limit_bytes = 1024*1024*256; + server.client_obuf_limits[REDIS_CLIENT_LIMIT_CLASS_SLAVE].soft_limit_bytes = 1024*1024*64; + server.client_obuf_limits[REDIS_CLIENT_LIMIT_CLASS_SLAVE].soft_limit_seconds = 60; + server.client_obuf_limits[REDIS_CLIENT_LIMIT_CLASS_PUBSUB].hard_limit_bytes = 1024*1024*32; + server.client_obuf_limits[REDIS_CLIENT_LIMIT_CLASS_PUBSUB].soft_limit_bytes = 1024*1024*8; + server.client_obuf_limits[REDIS_CLIENT_LIMIT_CLASS_PUBSUB].soft_limit_seconds = 60; + /* Double constants initialization */ R_Zero = 0.0; R_PosInf = 1.0/R_Zero; @@ -1007,6 +1025,7 @@ void initServer() { server.current_client = NULL; server.clients = listCreate(); + server.clients_to_close = listCreate(); server.slaves = listCreate(); server.monitors = listCreate(); server.unblocked_clients = listCreate(); @@ -1081,12 +1100,20 @@ void initServer() { } } + /* 32 bit instances are limited to 4GB of address space, so if there is + * no explicit limit in the user provided configuration we set a limit + * at 3.5GB using maxmemory with 'noeviction' policy'. This saves + * useless crashes of the Redis instance. */ + if (server.arch_bits == 32 && server.maxmemory == 0) { + redisLog(REDIS_WARNING,"Warning: 32 bit instance detected but no memory limit set. Setting 3.5 GB maxmemory limit with 'noeviction' policy now."); + server.maxmemory = 3584LL*(1024*1024); /* 3584 MB = 3.5 GB */ + server.maxmemory_policy = REDIS_MAXMEMORY_NO_EVICTION; + } + if (server.cluster_enabled) clusterInit(); scriptingInit(); slowlogInit(); bioInit(); - srand(time(NULL)^getpid()); - } /* Populates the Redis Command Table starting from the hard coded list @@ -1110,6 +1137,7 @@ void populateCommandTable(void) { case 'f': c->flags |= REDIS_CMD_FORCE_REPLICATION; break; case 's': c->flags |= REDIS_CMD_NOSCRIPT; break; case 'R': c->flags |= REDIS_CMD_RANDOM; break; + case 'S': c->flags |= REDIS_CMD_SORT_FOR_SCRIPT; break; default: redisPanic("Unsupported command flag"); break; } f++; @@ -1148,24 +1176,34 @@ struct redisCommand *lookupCommandByCString(char *s) { } /* Call() is the core of Redis execution of a command */ -void call(redisClient *c) { +void call(redisClient *c, int flags) { long long dirty, start = ustime(), duration; dirty = server.dirty; c->cmd->proc(c); dirty = server.dirty-dirty; duration = ustime()-start; - c->cmd->microseconds += duration; - slowlogPushEntryIfNeeded(c->argv,c->argc,duration); - c->cmd->calls++; - - 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); + + /* When EVAL is called loading the AOF we don't want commands called + * from Lua to go into the slowlog or to populate statistics. */ + if (server.loading && c->flags & REDIS_LUA_CLIENT) + flags &= ~(REDIS_CALL_SLOWLOG | REDIS_CALL_STATS); + + if (flags & REDIS_CALL_SLOWLOG) + slowlogPushEntryIfNeeded(c->argv,c->argc,duration); + if (flags & REDIS_CALL_STATS) { + c->cmd->microseconds += duration; + c->cmd->calls++; + } + 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); + } server.stat_numcommands++; } @@ -1237,12 +1275,13 @@ int processCommand(redisClient *c) { * First we try to free some memory if possible (if there are volatile * 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 && (c->cmd->flags & REDIS_CMD_DENYOOM) && - zmalloc_used_memory() > server.maxmemory) - { - addReplyError(c,"command not allowed when used memory > 'maxmemory'"); - return REDIS_OK; + if (server.maxmemory) { + int retval = freeMemoryIfNeeded(); + if ((c->cmd->flags & REDIS_CMD_DENYOOM) && retval == REDIS_ERR) { + addReplyError(c, + "command not allowed when used memory > 'maxmemory'"); + return REDIS_OK; + } } /* Only allow SUBSCRIBE and UNSUBSCRIBE in the context of Pub/Sub */ @@ -1294,7 +1333,7 @@ int processCommand(redisClient *c) { queueMultiCommand(c); addReply(c,shared.queued); } else { - call(c); + call(c,REDIS_CALL_FULL); } return REDIS_OK; } @@ -1427,7 +1466,7 @@ sds genRedisInfoString(char *section) { "redis_version:%s\r\n" "redis_git_sha1:%s\r\n" "redis_git_dirty:%d\r\n" - "arch_bits:%s\r\n" + "arch_bits:%d\r\n" "multiplexing_api:%s\r\n" "gcc_version:%d.%d.%d\r\n" "process_id:%ld\r\n" @@ -1438,7 +1477,7 @@ sds genRedisInfoString(char *section) { REDIS_VERSION, redisGitSHA1(), strtol(redisGitDirty(),NULL,10) > 0, - (sizeof(long) == 8) ? "64" : "32", + server.arch_bits, aeGetApiName(), #ifdef __GNUC__ __GNUC__,__GNUC_MINOR__,__GNUC_PATCHLEVEL__, @@ -1457,7 +1496,7 @@ sds genRedisInfoString(char *section) { if (sections++) info = sdscat(info,"\r\n"); info = sdscatprintf(info, "# Clients\r\n" - "connected_clients:%d\r\n" + "connected_clients:%lu\r\n" "client_longest_output_list:%lu\r\n" "client_biggest_input_buf:%lu\r\n" "blocked_clients:%d\r\n", @@ -1572,7 +1611,7 @@ sds genRedisInfoString(char *section) { "keyspace_hits:%lld\r\n" "keyspace_misses:%lld\r\n" "pubsub_channels:%ld\r\n" - "pubsub_patterns:%u\r\n" + "pubsub_patterns:%lu\r\n" "latest_fork_usec:%lld\r\n", server.stat_numconnections, server.stat_numcommands, @@ -1625,7 +1664,7 @@ sds genRedisInfoString(char *section) { } } info = sdscatprintf(info, - "connected_slaves:%d\r\n", + "connected_slaves:%lu\r\n", listLength(server.slaves)); if (listLength(server.slaves)) { int slaveid = 0; @@ -1745,23 +1784,57 @@ void monitorCommand(redisClient *c) { /* ============================ Maxmemory directive ======================== */ /* This function gets called when 'maxmemory' is set on the config file to limit - * the max memory used by the server, and we are out of memory. - * This function will try to, in order: + * the max memory used by the server, before processing a command. + * + * The goal of the function is to free enough memory to keep Redis under the + * configured memory limit. * - * - Free objects from the free list - * - Try to remove keys with an EXPIRE set + * The function starts calculating how many bytes should be freed to keep + * Redis under the limit, and enters a loop selecting the best keys to + * evict accordingly to the configured policy. * - * It is not possible to free enough memory to reach used-memory < maxmemory - * the server will start refusing commands that will enlarge even more the - * memory usage. + * If all the bytes needed to return back under the limit were freed the + * function returns REDIS_OK, otherwise REDIS_ERR is returned, and the caller + * should block the execution of commands that will result in more memory + * used by the server. */ -void freeMemoryIfNeeded(void) { - /* Remove keys accordingly to the active policy as long as we are - * over the memory limit. */ - if (server.maxmemory_policy == REDIS_MAXMEMORY_NO_EVICTION) return; +int freeMemoryIfNeeded(void) { + size_t mem_used, mem_tofree, mem_freed; + int slaves = listLength(server.slaves); + + /* 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; + listNode *ln; + + listRewind(server.slaves,&li); + while((ln = listNext(&li))) { + redisClient *slave = listNodeValue(ln); + unsigned long obuf_bytes = getClientOutputBufferMemoryUsage(slave); + if (obuf_bytes > mem_used) + mem_used = 0; + else + 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; + + if (server.maxmemory_policy == REDIS_MAXMEMORY_NO_EVICTION) + return REDIS_ERR; /* We need to free memory, but policy forbids. */ - while (server.maxmemory && zmalloc_used_memory() > server.maxmemory) { - int j, k, freed = 0; + /* Compute how much memory we need to free. */ + mem_tofree = mem_used - server.maxmemory; + mem_freed = 0; + while (mem_freed < mem_tofree) { + int j, k, keys_freed = 0; for (j = 0; j < server.dbnum; j++) { long bestval = 0; /* just to prevent warning */ @@ -1834,16 +1907,36 @@ void freeMemoryIfNeeded(void) { /* Finally remove the selected key. */ if (bestkey) { + long long delta; + robj *keyobj = createStringObject(bestkey,sdslen(bestkey)); propagateExpire(db,keyobj); + /* We compute the amount of memory freed by dbDelete() alone. + * It is possible that actually the memory needed to propagate + * the DEL in AOF and replication link is greater than the one + * we are freeing removing the key, but we can't account for + * that otherwise we would never exit the loop. + * + * AOF and Output buffer memory will be freed eventually so + * we only care about memory used by the key space. */ + delta = (long long) zmalloc_used_memory(); dbDelete(db,keyobj); + delta -= (long long) zmalloc_used_memory(); + mem_freed += delta; server.stat_evictedkeys++; decrRefCount(keyobj); - freed++; + keys_freed++; + + /* When the memory to free starts to be big enough, we may + * 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) flushSlavesOutputBuffers(); } } - if (!freed) return; /* nothing to free... */ + if (!keys_freed) return REDIS_ERR; /* nothing to free... */ } + return REDIS_OK; } /* =================================== Main! ================================ */ @@ -1933,136 +2026,6 @@ void redisAsciiArt(void) { zfree(buf); } -#ifdef HAVE_BACKTRACE -static void *getMcontextEip(ucontext_t *uc) { -#if defined(__FreeBSD__) - return (void*) uc->uc_mcontext.mc_eip; -#elif defined(__dietlibc__) - return (void*) uc->uc_mcontext.eip; -#elif defined(__APPLE__) && !defined(MAC_OS_X_VERSION_10_6) - #if __x86_64__ - return (void*) uc->uc_mcontext->__ss.__rip; - #elif __i386__ - return (void*) uc->uc_mcontext->__ss.__eip; - #else - return (void*) uc->uc_mcontext->__ss.__srr0; - #endif -#elif defined(__APPLE__) && defined(MAC_OS_X_VERSION_10_6) - #if defined(_STRUCT_X86_THREAD_STATE64) && !defined(__i386__) - return (void*) uc->uc_mcontext->__ss.__rip; - #else - return (void*) uc->uc_mcontext->__ss.__eip; - #endif -#elif defined(__i386__) - return (void*) uc->uc_mcontext.gregs[14]; /* Linux 32 */ -#elif defined(__X86_64__) || defined(__x86_64__) - return (void*) uc->uc_mcontext.gregs[16]; /* Linux 64 */ -#elif defined(__ia64__) /* Linux IA64 */ - return (void*) uc->uc_mcontext.sc_ip; -#else - return NULL; -#endif -} - -void bugReportStart(void) { - if (server.bug_report_start == 0) { - redisLog(REDIS_WARNING, - "=== REDIS BUG REPORT START: Cut & paste starting from here ==="); - server.bug_report_start = 1; - } -} - -static void sigsegvHandler(int sig, siginfo_t *info, void *secret) { - void *trace[100]; - char **messages = NULL; - int i, trace_size = 0; - ucontext_t *uc = (ucontext_t*) secret; - sds infostring, clients; - struct sigaction act; - REDIS_NOTUSED(info); - - bugReportStart(); - redisLog(REDIS_WARNING, - " Redis %s crashed by signal: %d", REDIS_VERSION, sig); - redisLog(REDIS_WARNING, - " Failed assertion: %s (%s:%d)", server.assert_failed, - server.assert_file, server.assert_line); - - /* Generate the stack trace */ - trace_size = backtrace(trace, 100); - - /* overwrite sigaction with caller's address */ - if (getMcontextEip(uc) != NULL) { - trace[1] = getMcontextEip(uc); - } - messages = backtrace_symbols(trace, trace_size); - redisLog(REDIS_WARNING, "--- STACK TRACE"); - for (i=1; iargc; j++) { - robj *decoded; - - decoded = getDecodedObject(cc->argv[j]); - redisLog(REDIS_WARNING,"argv[%d]: '%s'", j, (char*)decoded->ptr); - decrRefCount(decoded); - } - /* Check if the first argument, usually a key, is found inside the - * selected DB, and if so print info about the associated object. */ - if (cc->argc >= 1) { - robj *val, *key; - dictEntry *de; - - key = getDecodedObject(cc->argv[1]); - de = dictFind(cc->db->dict, key->ptr); - if (de) { - val = dictGetVal(de); - redisLog(REDIS_WARNING,"key '%s' found in DB containing the following object:", key->ptr); - redisLogObjectDebugInfo(val); - } - decrRefCount(key); - } - } - - redisLog(REDIS_WARNING, -"=== REDIS BUG REPORT END. Make sure to include from START to END. ===\n\n" -" Please report the crash opening an issue on github:\n\n" -" http://github.com/antirez/redis/issues\n\n" -); - /* free(messages); Don't call free() with possibly corrupted memory. */ - if (server.daemonize) unlink(server.pidfile); - - /* Make sure we exit with the right signal at the end. So for instance - * the core will be dumped if enabled. */ - sigemptyset (&act.sa_mask); - /* When the SA_SIGINFO flag is set in sa_flags then sa_sigaction - * is used. Otherwise, sa_handler is used */ - act.sa_flags = SA_NODEFER | SA_ONSTACK | SA_RESETHAND; - act.sa_handler = SIG_DFL; - sigaction (sig, &act, NULL); - kill(getpid(),sig); -} -#endif /* HAVE_BACKTRACE */ - static void sigtermHandler(int sig) { REDIS_NOTUSED(sig); @@ -2094,9 +2057,15 @@ void setupSignalHandlers(void) { int main(int argc, char **argv) { long long start; + struct timeval tv; + /* We need to initialize our libraries, and the server configuration. */ zmalloc_enable_thread_safeness(); + srand(time(NULL)^getpid()); + gettimeofday(&tv,NULL); + dictSetHashFunctionSeed(tv.tv_sec^tv.tv_usec^getpid()); initServerConfig(); + if (argc >= 2) { int j = 1; /* First option to parse in argv[] */ sds options = sdsempty(); diff --git a/src/redis.h b/src/redis.h index 1ac2a2d2..982f33fe 100644 --- a/src/redis.h +++ b/src/redis.h @@ -20,6 +20,7 @@ #include #include #include +#include #include "ae.h" /* Event driven programming library */ #include "sds.h" /* Dynamic safe strings */ @@ -45,6 +46,7 @@ #define REDIS_EXPIRELOOKUPS_PER_CRON 10 /* lookup 10 expires per loop */ #define REDIS_MAX_WRITE_PER_EVENT (1024*64) #define REDIS_SHARED_INTEGERS 10000 +#define REDIS_SHARED_BULKHDR_LEN 32 #define REDIS_MAX_LOGMSG_LEN 1024 /* Default maximum length of syslog messages */ #define REDIS_AOF_REWRITE_PERC 100 #define REDIS_AOF_REWRITE_MIN_SIZE (1024*1024) @@ -76,6 +78,7 @@ #define REDIS_CMD_PUBSUB 32 /* "p" flag */ #define REDIS_CMD_NOSCRIPT 64 /* "s" flag */ #define REDIS_CMD_RANDOM 128 /* "R" flag */ +#define REDIS_CMD_SORT_FOR_SCRIPT 256 /* "S" flag */ /* Object types */ #define REDIS_STRING 0 @@ -141,11 +144,19 @@ server.unblocked_clients */ #define REDIS_LUA_CLIENT 512 /* This is a non connected client used by Lua */ #define REDIS_ASKING 1024 /* Client issued the ASKING command */ +#define REDIS_CLOSE_ASAP 2048 /* Close this client ASAP */ /* Client request types */ #define REDIS_REQ_INLINE 1 #define REDIS_REQ_MULTIBULK 2 +/* Client classes for client limits, currently used only for + * the max-client-output-buffer limit implementation. */ +#define REDIS_CLIENT_LIMIT_CLASS_NORMAL 0 +#define REDIS_CLIENT_LIMIT_CLASS_SLAVE 1 +#define REDIS_CLIENT_LIMIT_CLASS_PUBSUB 2 +#define REDIS_CLIENT_LIMIT_NUM_CLASSES 3 + /* Slave replication state - slave side */ #define REDIS_REPL_NONE 0 /* No active replication */ #define REDIS_REPL_CONNECT 1 /* Must connect to master */ @@ -227,6 +238,13 @@ points are configured. */ #define REDIS_SHUTDOWN_NOSAVE 2 /* Don't SAVE on SHUTDOWN. */ +/* Command call flags, see call() function */ +#define REDIS_CALL_NONE 0 +#define REDIS_CALL_SLOWLOG 1 +#define REDIS_CALL_STATS 2 +#define REDIS_CALL_PROPAGATE 4 +#define REDIS_CALL_FULL (REDIS_CALL_SLOWLOG | REDIS_CALL_STATS | REDIS_CALL_PROPAGATE) + /* We can print the stacktrace, so our assert is defined this way: */ #define redisAssertWithInfo(_c,_o,_e) ((_e)?(void)0 : (_redisAssertWithInfo(_c,_o,#_e,__FILE__,__LINE__),_exit(1))) #define redisAssert(_e) ((_e)?(void)0 : (_redisAssert(#_e,__FILE__,__LINE__),_exit(1))) @@ -305,8 +323,10 @@ typedef struct redisClient { int multibulklen; /* number of multi bulk arguments left to read */ long bulklen; /* length of bulk argument in multi bulk request */ list *reply; + unsigned long reply_bytes; /* Tot bytes of objects in reply list */ int sentlen; time_t lastinteraction; /* time of the last interaction, used for timeout */ + time_t obuf_soft_limit_reached_time; int flags; /* REDIS_SLAVE | REDIS_MONITOR | REDIS_MULTI ... */ int slaveseldb; /* slave selected db, if this client is a slave */ int authenticated; /* when requirepass is non-NULL */ @@ -339,9 +359,11 @@ struct sharedObjectsStruct { *outofrangeerr, *noscripterr, *loadingerr, *slowscripterr, *plus, *select0, *select1, *select2, *select3, *select4, *select5, *select6, *select7, *select8, *select9, - *messagebulk, *pmessagebulk, *subscribebulk, *unsubscribebulk, *mbulk3, - *mbulk4, *psubscribebulk, *punsubscribebulk, - *integers[REDIS_SHARED_INTEGERS]; + *messagebulk, *pmessagebulk, *subscribebulk, *unsubscribebulk, + *psubscribebulk, *punsubscribebulk, *del, + *integers[REDIS_SHARED_INTEGERS], + *mbulkhdr[REDIS_SHARED_BULKHDR_LEN], /* "*\r\n" */ + *bulkhdr[REDIS_SHARED_BULKHDR_LEN]; /* "$\r\n" */ }; /* ZSETs use a specialized version of Skiplists */ @@ -366,6 +388,12 @@ typedef struct zset { zskiplist *zsl; } zset; +typedef struct clientBufferLimitsConfig { + unsigned long long hard_limit_bytes; + unsigned long long soft_limit_bytes; + time_t soft_limit_seconds; +} clientBufferLimitsConfig; + /*----------------------------------------------------------------------------- * Redis cluster data structures *----------------------------------------------------------------------------*/ @@ -509,6 +537,7 @@ struct redisServer { int activerehashing; /* Incremental rehash in serverCron() */ char *requirepass; /* Pass for AUTH command, or NULL */ char *pidfile; /* PID file path */ + int arch_bits; /* 32 or 64 depending on sizeof(long) */ /* Networking */ int port; /* TCP listening port */ char *bindaddr; /* Bind address or NULL */ @@ -518,6 +547,7 @@ struct redisServer { int sofd; /* Unix socket file descriptor */ int cfd; /* Cluster bus lisetning socket */ list *clients; /* List of active clients */ + list *clients_to_close; /* Clients to close asynchronously */ list *slaves, *monitors; /* List of slaves and MONITORs */ redisClient *current_client; /* Current client, only used on crash report */ char neterr[ANET_ERR_LEN]; /* Error buffer for anet.c */ @@ -551,6 +581,7 @@ struct redisServer { size_t client_max_querybuf_len; /* Limit for client query buffer length */ int dbnum; /* Total number of configured DBs */ int daemonize; /* True if running as a daemon */ + clientBufferLimitsConfig client_obuf_limits[REDIS_CLIENT_LIMIT_NUM_CLASSES]; /* AOF persistence */ int aof_state; /* REDIS_AOF_(ON|OFF|WAIT_REWRITE) */ int aof_fsync; /* Kind of fsync() policy */ @@ -607,6 +638,7 @@ struct redisServer { list *unblocked_clients; /* list of clients to unblock before next loop */ /* Sort parameters - qsort_r() is only available under BSD so we * have to take this state global, in order to pass it to sortCompare() */ + int sort_dontsort; int sort_desc; int sort_alpha; int sort_bypattern; @@ -783,6 +815,12 @@ sds getClientInfoString(redisClient *client); sds getAllClientsInfoString(void); void rewriteClientCommandVector(redisClient *c, int argc, ...); void rewriteClientCommandArgument(redisClient *c, int i, robj *newval); +unsigned long getClientOutputBufferMemoryUsage(redisClient *c); +void freeClientsInAsyncFreeQueue(void); +void asyncCloseClientOnOutputBufferLimitReached(redisClient *c); +int getClientLimitClassByName(char *name); +char *getClientLimitClassName(int class); +void flushSlavesOutputBuffers(void); #ifdef __GNUC__ void addReplyErrorFormat(redisClient *c, const char *fmt, ...) @@ -799,7 +837,7 @@ void listTypeTryConversion(robj *subject, robj *value); void listTypePush(robj *subject, robj *value, int where); robj *listTypePop(robj *subject, int where); unsigned long listTypeLength(robj *subject); -listTypeIterator *listTypeInitIterator(robj *subject, int index, unsigned char direction); +listTypeIterator *listTypeInitIterator(robj *subject, long index, unsigned char direction); void listTypeReleaseIterator(listTypeIterator *li); int listTypeNext(listTypeIterator *li, listTypeEntry *entry); robj *listTypeGet(listTypeEntry *entry); @@ -906,12 +944,12 @@ unsigned int zsetLength(robj *zobj); void zsetConvert(robj *zobj, int encoding); /* Core functions */ -void freeMemoryIfNeeded(void); +int freeMemoryIfNeeded(void); int processCommand(redisClient *c); void setupSignalHandlers(void); struct redisCommand *lookupCommand(sds name); struct redisCommand *lookupCommandByCString(char *s); -void call(redisClient *c); +void call(redisClient *c, int flags); int prepareForShutdown(); void redisLog(int level, const char *fmt, ...); void redisLogRaw(int level, const char *msg); @@ -1161,5 +1199,6 @@ void _redisAssert(char *estr, char *file, int line); void _redisPanic(char *msg, char *file, int line); void bugReportStart(void); void redisLogObjectDebugInfo(robj *o); - +void sigsegvHandler(int sig, siginfo_t *info, void *secret); +sds genRedisInfoString(char *section); #endif diff --git a/src/scripting.c b/src/scripting.c index 1503c3c9..9d41a45f 100644 --- a/src/scripting.c +++ b/src/scripting.c @@ -27,7 +27,7 @@ int redis_math_randomseed (lua_State *L); * is like a normal client that bypasses all the slow I/O paths. * * Note: in this function we do not do any sanity check as the reply is - * generated by Redis directly. This allows use to go faster. + * generated by Redis directly. This allows us to go faster. * The reply string can be altered during the parsing as it is discared * after the conversion is completed. * @@ -128,6 +128,37 @@ void luaPushError(lua_State *lua, char *error) { lua_settable(lua,-3); } +/* Sort the array currently in the stack. We do this to make the output + * of commands like KEYS or SMEMBERS something deterministic when called + * from Lua (to play well with AOf/replication). + * + * The array is sorted using table.sort itself, and assuming all the + * list elements are strings. */ +void luaSortArray(lua_State *lua) { + /* Initial Stack: array */ + lua_getglobal(lua,"table"); + lua_pushstring(lua,"sort"); + lua_gettable(lua,-2); /* Stack: array, table, table.sort */ + lua_pushvalue(lua,-3); /* Stack: array, table, table.sort, array */ + if (lua_pcall(lua,1,0,0)) { + /* Stack: array, table, error */ + + /* We are not interested in the error, we assume that the problem is + * that there are 'false' elements inside the array, so we try + * again with a slower function but able to handle this case, that + * is: table.sort(table, __redis__compare_helper) */ + lua_pop(lua,1); /* Stack: array, table */ + lua_pushstring(lua,"sort"); /* Stack: array, table, sort */ + lua_gettable(lua,-2); /* Stack: array, table, table.sort */ + lua_pushvalue(lua,-3); /* Stack: array, table, table.sort, array */ + lua_getglobal(lua,"__redis__compare_helper"); + /* Stack: array, table, table.sort, array, __redis__compare_helper */ + lua_call(lua,2,0); + } + /* Stack: array (sorted), table */ + lua_pop(lua,1); /* Stack: array (sorted) */ +} + int luaRedisGenericCommand(lua_State *lua, int raise_error) { int j, argc = lua_gettop(lua); struct redisCommand *cmd; @@ -190,7 +221,8 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) { if (cmd->flags & REDIS_CMD_WRITE) server.lua_write_dirty = 1; /* Run the command */ - cmd->proc(c); + c->cmd = cmd; + call(c,REDIS_CALL_SLOWLOG | REDIS_CALL_STATS); /* Convert the result of the Redis command into a suitable Lua type. * The first thing we need is to create a single string from the client @@ -208,6 +240,14 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) { } if (raise_error && reply[0] != '-') raise_error = 0; redisProtocolToLuaType(lua,reply); + /* Sort the output array if needed, assuming it is a non-null multi bulk + * reply as expected. */ + if ((cmd->flags & REDIS_CMD_SORT_FOR_SCRIPT) && + (reply[0] == '*' && reply[1] != '-')) { + /* Skip this step if command is SORT but output was already sorted */ + if (cmd->proc != sortCommand || server.sort_dontsort) + luaSortArray(lua); + } sdsfree(reply); cleanup: @@ -303,6 +343,7 @@ void luaLoadLib(lua_State *lua, const char *libname, lua_CFunction luafunc) { } LUALIB_API int (luaopen_cjson) (lua_State *L); +LUALIB_API int (luaopen_struct) (lua_State *L); void luaLoadLibraries(lua_State *lua) { luaLoadLib(lua, "", luaopen_base); @@ -310,7 +351,8 @@ void luaLoadLibraries(lua_State *lua) { luaLoadLib(lua, LUA_STRLIBNAME, luaopen_string); luaLoadLib(lua, LUA_MATHLIBNAME, luaopen_math); luaLoadLib(lua, LUA_DBLIBNAME, luaopen_debug); - luaLoadLib(lua, "cjson", luaopen_cjson); + luaLoadLib(lua, "cjson", luaopen_cjson); + luaLoadLib(lua, "struct", luaopen_struct); #if 0 /* Stuff that we don't load currently, for sandboxing concerns. */ luaLoadLib(lua, LUA_LOADLIBNAME, luaopen_package); @@ -381,6 +423,18 @@ void scriptingInit(void) { lua_setglobal(lua,"math"); + /* Add a helper funciton that we use to sort the multi bulk output of non + * deterministic commands, when containing 'false' elements. */ + { + char *compare_func = "function __redis__compare_helper(a,b)\n" + " if a == false then a = '' end\n" + " if b == false then b = '' end\n" + " return aptr,sdslen(body->ptr)); - funcdef = sdscatlen(funcdef,"\nend\n",5); + funcdef = sdscatlen(funcdef," end",4); if (luaL_loadbuffer(lua,funcdef,sdslen(funcdef),"func definition")) { addReplyErrorFormat(c,"Error compiling script (new function): %s\n", diff --git a/src/sort.c b/src/sort.c index 7e50582e..3f02e49a 100644 --- a/src/sort.c +++ b/src/sort.c @@ -1,5 +1,6 @@ #include "redis.h" #include "pqsort.h" /* Partial qsort for SORT+LIMIT */ +#include /* isnan() */ redisSortOperation *createSortOperation(int type, robj *pattern) { redisSortOperation *so = zmalloc(sizeof(*so)); @@ -102,7 +103,10 @@ int sortCompare(const void *s1, const void *s2) { } else if (so1->u.score < so2->u.score) { cmp = -1; } else { - cmp = 0; + /* Objects have the same score, but we don't want the comparison + * to be undefined, so we compare objects lexicographycally. + * This way the result of SORT is deterministic. */ + cmp = compareStringObjects(so1->obj,so2->obj); } } else { /* Alphanumeric sorting */ @@ -136,6 +140,7 @@ void sortCommand(redisClient *c) { long limit_start = 0, limit_count = -1, start, end; int j, dontsort = 0, vectorlen; int getop = 0; /* GET operation counter */ + int int_convertion_error = 0; robj *sortval, *sortby = NULL, *storekey = NULL; redisSortObject *vector; /* Resulting vector to sort */ @@ -198,6 +203,15 @@ void sortCommand(redisClient *c) { j++; } + /* If we have STORE we need to force sorting for deterministic output + * and replication. We use alpha sorting since this is guaranteed to + * work with any input. */ + if (storekey && dontsort) { + dontsort = 0; + alpha = 1; + sortby = NULL; + } + /* Destructively convert encoded sorted sets for SORT. */ if (sortval->type == REDIS_ZSET) zsetConvert(sortval, REDIS_ENCODING_SKIPLIST); @@ -266,7 +280,14 @@ void sortCommand(redisClient *c) { if (sortby) vector[j].u.cmpobj = getDecodedObject(byval); } else { if (byval->encoding == REDIS_ENCODING_RAW) { - vector[j].u.score = strtod(byval->ptr,NULL); + char *eptr; + + vector[j].u.score = strtod(byval->ptr,&eptr); + if (eptr[0] != '\0' || errno == ERANGE || + isnan(vector[j].u.score)) + { + int_convertion_error = 1; + } } else if (byval->encoding == REDIS_ENCODING_INT) { /* Don't need to decode the object if it's * integer-encoded (the only encoding supported) so @@ -295,6 +316,7 @@ void sortCommand(redisClient *c) { } if (end >= vectorlen) end = vectorlen-1; + server.sort_dontsort = dontsort; if (dontsort == 0) { server.sort_desc = desc; server.sort_alpha = alpha; @@ -308,7 +330,9 @@ void sortCommand(redisClient *c) { /* Send command output to the output buffer, performing the specified * GET/DEL/INCR/DECR operations if any. */ outputlen = getop ? getop*(end-start+1) : end-start+1; - if (storekey == NULL) { + if (int_convertion_error) { + addReplyError(c,"One or more scores can't be converted into double"); + } else if (storekey == NULL) { /* STORE option not specified, sent the sorting result to client */ addReplyMultiBulkLen(c,outputlen); for (j = start; j <= end; j++) { @@ -367,9 +391,14 @@ void sortCommand(redisClient *c) { } } } - if (outputlen) setKey(c->db,storekey,sobj); + if (outputlen) { + setKey(c->db,storekey,sobj); + server.dirty += outputlen; + } else if (dbDelete(c->db,storekey)) { + signalModifiedKey(c->db,storekey); + server.dirty++; + } decrRefCount(sobj); - server.dirty += outputlen; addReplyLongLong(c,outputlen); } diff --git a/src/t_hash.c b/src/t_hash.c index 8ee5485c..f97fc992 100644 --- a/src/t_hash.c +++ b/src/t_hash.c @@ -320,7 +320,7 @@ void hmsetCommand(redisClient *c) { } void hincrbyCommand(redisClient *c) { - long long value, incr; + long long value, incr, oldvalue; robj *o, *current, *new; if (getLongLongFromObjectOrReply(c,c->argv[3],&incr,NULL) != REDIS_OK) return; @@ -336,6 +336,12 @@ void hincrbyCommand(redisClient *c) { value = 0; } + oldvalue = value; + if ((incr < 0 && oldvalue < 0 && incr < (LLONG_MIN-oldvalue)) || + (incr > 0 && oldvalue > 0 && incr > (LLONG_MAX-oldvalue))) { + addReplyError(c,"increment or decrement would overflow"); + return; + } value += incr; new = createStringObjectFromLongLong(value); hashTypeTryObjectEncoding(o,&c->argv[2],NULL); diff --git a/src/t_list.c b/src/t_list.c index f856d6cd..3742ec49 100644 --- a/src/t_list.c +++ b/src/t_list.c @@ -86,7 +86,7 @@ unsigned long listTypeLength(robj *subject) { } /* Initialize an iterator at the specified index. */ -listTypeIterator *listTypeInitIterator(robj *subject, int index, unsigned char direction) { +listTypeIterator *listTypeInitIterator(robj *subject, long index, unsigned char direction) { listTypeIterator *li = zmalloc(sizeof(listTypeIterator)); li->subject = subject; li->encoding = subject->encoding; @@ -484,10 +484,7 @@ void rpopCommand(redisClient *c) { void lrangeCommand(redisClient *c) { robj *o; - long start; - long end; - int llen; - int rangelen; + long start, end, llen, rangelen; if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) || (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return; @@ -546,10 +543,7 @@ void lrangeCommand(redisClient *c) { void ltrimCommand(redisClient *c) { robj *o; - long start; - long end; - int llen; - int j, ltrim, rtrim; + long start, end, llen, j, ltrim, rtrim; list *list; listNode *ln; @@ -604,7 +598,7 @@ void lremCommand(redisClient *c) { robj *subject, *obj; obj = c->argv[3] = tryObjectEncoding(c->argv[3]); long toremove; - int removed = 0; + long removed = 0; listTypeEntry entry; if ((getLongFromObjectOrReply(c, c->argv[2], &toremove, NULL) != REDIS_OK)) diff --git a/src/t_string.c b/src/t_string.c index 2bd1646e..d6143ed2 100644 --- a/src/t_string.c +++ b/src/t_string.c @@ -344,11 +344,12 @@ void incrDecrCommand(redisClient *c, long long incr) { if (getLongLongFromObjectOrReply(c,o,&value,NULL) != REDIS_OK) return; oldvalue = value; - value += incr; - if ((incr < 0 && value > oldvalue) || (incr > 0 && value < oldvalue)) { + if ((incr < 0 && oldvalue < 0 && incr < (LLONG_MIN-oldvalue)) || + (incr > 0 && oldvalue > 0 && incr > (LLONG_MAX-oldvalue))) { addReplyError(c,"increment or decrement would overflow"); return; } + value += incr; new = createStringObjectFromLongLong(value); if (o) dbOverwrite(c->db,c->argv[1],new); diff --git a/src/t_zset.c b/src/t_zset.c index d2535c72..d482d4c2 100644 --- a/src/t_zset.c +++ b/src/t_zset.c @@ -502,7 +502,7 @@ int zzlIsInRange(unsigned char *zl, zrangespec *range) { return 0; p = ziplistIndex(zl,-1); /* Last score. */ - redisAssert(p != NULL); + if (p == NULL) return 0; /* Empty sorted set */ score = zzlGetScore(p); if (!zslValueGteMin(score,range)) return 0; @@ -1254,7 +1254,7 @@ int zuiNext(zsetopsrc *op, zsetopval *val) { if (val->flags & OPVAL_DIRTY_ROBJ) decrRefCount(val->ele); - bzero(val,sizeof(zsetopval)); + memset(val,0,sizeof(zsetopval)); if (op->type == REDIS_SET) { iterset *it = &op->iter.set; diff --git a/src/ziplist.c b/src/ziplist.c index 639e4b61..4ecd1885 100644 --- a/src/ziplist.c +++ b/src/ziplist.c @@ -69,7 +69,7 @@ #include "zmalloc.h" #include "util.h" #include "ziplist.h" -#include "endian.h" +#include "endianconv.h" #define ZIP_END 255 #define ZIP_BIGLEN 254 @@ -92,13 +92,15 @@ #define ZIPLIST_LENGTH(zl) (*((uint16_t*)((zl)+sizeof(uint32_t)*2))) #define ZIPLIST_HEADER_SIZE (sizeof(uint32_t)*2+sizeof(uint16_t)) #define ZIPLIST_ENTRY_HEAD(zl) ((zl)+ZIPLIST_HEADER_SIZE) -#define ZIPLIST_ENTRY_TAIL(zl) ((zl)+ZIPLIST_TAIL_OFFSET(zl)) -#define ZIPLIST_ENTRY_END(zl) ((zl)+ZIPLIST_BYTES(zl)-1) +#define ZIPLIST_ENTRY_TAIL(zl) ((zl)+intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))) +#define ZIPLIST_ENTRY_END(zl) ((zl)+intrev32ifbe(ZIPLIST_BYTES(zl))-1) /* We know a positive increment can only be 1 because entries can only be * pushed one at a time. */ #define ZIPLIST_INCR_LENGTH(zl,incr) { \ - if (ZIPLIST_LENGTH(zl) < UINT16_MAX) ZIPLIST_LENGTH(zl)+=incr; } + if (ZIPLIST_LENGTH(zl) < UINT16_MAX) \ + ZIPLIST_LENGTH(zl) = intrev16ifbe(intrev16ifbe(ZIPLIST_LENGTH(zl))+incr); \ +} typedef struct zlentry { unsigned int prevrawlensize, prevrawlen; @@ -302,11 +304,11 @@ static int64_t zipLoadInteger(unsigned char *p, unsigned char encoding) { ret = i16; } else if (encoding == ZIP_INT_32B) { memcpy(&i32,p,sizeof(i32)); - memrev16ifbe(&i32); + memrev32ifbe(&i32); ret = i32; } else if (encoding == ZIP_INT_64B) { memcpy(&i64,p,sizeof(i64)); - memrev16ifbe(&i64); + memrev64ifbe(&i64); ret = i64; } else { assert(NULL); @@ -335,8 +337,8 @@ static unsigned int zipRawEntryLength(unsigned char *p) { unsigned char *ziplistNew(void) { unsigned int bytes = ZIPLIST_HEADER_SIZE+1; unsigned char *zl = zmalloc(bytes); - ZIPLIST_BYTES(zl) = bytes; - ZIPLIST_TAIL_OFFSET(zl) = ZIPLIST_HEADER_SIZE; + ZIPLIST_BYTES(zl) = intrev32ifbe(bytes); + ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(ZIPLIST_HEADER_SIZE); ZIPLIST_LENGTH(zl) = 0; zl[bytes-1] = ZIP_END; return zl; @@ -345,7 +347,7 @@ unsigned char *ziplistNew(void) { /* Resize the ziplist. */ static unsigned char *ziplistResize(unsigned char *zl, unsigned int len) { zl = zrealloc(zl,len); - ZIPLIST_BYTES(zl) = len; + ZIPLIST_BYTES(zl) = intrev32ifbe(len); zl[len-1] = ZIP_END; return zl; } @@ -371,7 +373,7 @@ static unsigned char *ziplistResize(unsigned char *zl, unsigned int len) { * The pointer "p" points to the first entry that does NOT need to be * updated, i.e. consecutive fields MAY need an update. */ static unsigned char *__ziplistCascadeUpdate(unsigned char *zl, unsigned char *p) { - size_t curlen = ZIPLIST_BYTES(zl), rawlen, rawlensize; + size_t curlen = intrev32ifbe(ZIPLIST_BYTES(zl)), rawlen, rawlensize; size_t offset, noffset, extra; unsigned char *np; zlentry cur, next; @@ -401,8 +403,10 @@ static unsigned char *__ziplistCascadeUpdate(unsigned char *zl, unsigned char *p noffset = np-zl; /* Update tail offset when next element is not the tail element. */ - if ((zl+ZIPLIST_TAIL_OFFSET(zl)) != np) - ZIPLIST_TAIL_OFFSET(zl) += extra; + if ((zl+intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))) != np) { + ZIPLIST_TAIL_OFFSET(zl) = + intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+extra); + } /* Move the tail to the back. */ memmove(np+rawlensize, @@ -453,25 +457,30 @@ static unsigned char *__ziplistDelete(unsigned char *zl, unsigned char *p, unsig zipPrevEncodeLength(p-nextdiff,first.prevrawlen); /* Update offset for tail */ - ZIPLIST_TAIL_OFFSET(zl) -= totlen; + ZIPLIST_TAIL_OFFSET(zl) = + intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))-totlen); /* When the tail contains more than one entry, we need to take * "nextdiff" in account as well. Otherwise, a change in the * size of prevlen doesn't have an effect on the *tail* offset. */ tail = zipEntry(p); - if (p[tail.headersize+tail.len] != ZIP_END) - ZIPLIST_TAIL_OFFSET(zl) += nextdiff; + if (p[tail.headersize+tail.len] != ZIP_END) { + ZIPLIST_TAIL_OFFSET(zl) = + intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff); + } /* Move tail to the front of the ziplist */ - memmove(first.p,p-nextdiff,ZIPLIST_BYTES(zl)-(p-zl)-1+nextdiff); + memmove(first.p,p-nextdiff, + intrev32ifbe(ZIPLIST_BYTES(zl))-(p-zl)-1+nextdiff); } else { /* The entire tail was deleted. No need to move memory. */ - ZIPLIST_TAIL_OFFSET(zl) = (first.p-zl)-first.prevrawlen; + ZIPLIST_TAIL_OFFSET(zl) = + intrev32ifbe((first.p-zl)-first.prevrawlen); } /* Resize and update length */ offset = first.p-zl; - zl = ziplistResize(zl, ZIPLIST_BYTES(zl)-totlen+nextdiff); + zl = ziplistResize(zl, intrev32ifbe(ZIPLIST_BYTES(zl))-totlen+nextdiff); ZIPLIST_INCR_LENGTH(zl,-deleted); p = zl+offset; @@ -485,7 +494,7 @@ static unsigned char *__ziplistDelete(unsigned char *zl, unsigned char *p, unsig /* Insert item at "p". */ static unsigned char *__ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char *s, unsigned int slen) { - size_t curlen = ZIPLIST_BYTES(zl), reqlen, prevlen = 0; + size_t curlen = intrev32ifbe(ZIPLIST_BYTES(zl)), reqlen, prevlen = 0; size_t offset; int nextdiff = 0; unsigned char encoding = 0; @@ -538,17 +547,20 @@ static unsigned char *__ziplistInsert(unsigned char *zl, unsigned char *p, unsig zipPrevEncodeLength(p+reqlen,reqlen); /* Update offset for tail */ - ZIPLIST_TAIL_OFFSET(zl) += reqlen; + ZIPLIST_TAIL_OFFSET(zl) = + intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+reqlen); /* When the tail contains more than one entry, we need to take * "nextdiff" in account as well. Otherwise, a change in the * size of prevlen doesn't have an effect on the *tail* offset. */ tail = zipEntry(p+reqlen); - if (p[reqlen+tail.headersize+tail.len] != ZIP_END) - ZIPLIST_TAIL_OFFSET(zl) += nextdiff; + if (p[reqlen+tail.headersize+tail.len] != ZIP_END) { + ZIPLIST_TAIL_OFFSET(zl) = + intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff); + } } else { /* This element will be the new tail. */ - ZIPLIST_TAIL_OFFSET(zl) = p-zl; + ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(p-zl); } /* When nextdiff != 0, the raw length of the next entry has changed, so @@ -720,8 +732,8 @@ unsigned int ziplistCompare(unsigned char *p, unsigned char *sstr, unsigned int /* Return length of ziplist. */ unsigned int ziplistLen(unsigned char *zl) { unsigned int len = 0; - if (ZIPLIST_LENGTH(zl) < UINT16_MAX) { - len = ZIPLIST_LENGTH(zl); + if (intrev16ifbe(ZIPLIST_LENGTH(zl)) < UINT16_MAX) { + len = intrev16ifbe(ZIPLIST_LENGTH(zl)); } else { unsigned char *p = zl+ZIPLIST_HEADER_SIZE; while (*p != ZIP_END) { @@ -730,14 +742,14 @@ unsigned int ziplistLen(unsigned char *zl) { } /* Re-store length if small enough */ - if (len < UINT16_MAX) ZIPLIST_LENGTH(zl) = len; + if (len < UINT16_MAX) ZIPLIST_LENGTH(zl) = intrev16ifbe(len); } return len; } /* Return ziplist blob size in bytes. */ size_t ziplistBlobLen(unsigned char *zl) { - return ZIPLIST_BYTES(zl); + return intrev32ifbe(ZIPLIST_BYTES(zl)); } void ziplistRepr(unsigned char *zl) { @@ -749,9 +761,9 @@ void ziplistRepr(unsigned char *zl) { "{total bytes %d} " "{length %u}\n" "{tail offset %u}\n", - ZIPLIST_BYTES(zl), - ZIPLIST_LENGTH(zl), - ZIPLIST_TAIL_OFFSET(zl)); + intrev32ifbe(ZIPLIST_BYTES(zl)), + intrev16ifbe(ZIPLIST_LENGTH(zl)), + intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))); p = ZIPLIST_ENTRY_HEAD(zl); while(*p != ZIP_END) { entry = zipEntry(p); @@ -852,7 +864,7 @@ void stress(int pos, int num, int maxsize, int dnum) { zl = ziplistDeleteRange(zl,0,1); } printf("List size: %8d, bytes: %8d, %dx push+pop (%s): %6lld usec\n", - i,ZIPLIST_BYTES(zl),num,posstr[pos],usec()-start); + i,intrev32ifbe(ZIPLIST_BYTES(zl)),num,posstr[pos],usec()-start); zfree(zl); } } diff --git a/src/zipmap.c b/src/zipmap.c index 65ed29c8..1f11fd42 100644 --- a/src/zipmap.c +++ b/src/zipmap.c @@ -80,7 +80,7 @@ #include #include #include "zmalloc.h" -#include "endian.h" +#include "endianconv.h" #define ZIPMAP_BIGLEN 254 #define ZIPMAP_END 255 diff --git a/src/zmalloc.c b/src/zmalloc.c index 56b9140c..89f80d83 100644 --- a/src/zmalloc.c +++ b/src/zmalloc.c @@ -150,6 +150,20 @@ void *zrealloc(void *ptr, size_t size) { #endif } +/* Provide zmalloc_size() for systems where this function is not provided by + * malloc itself, given that in that case we store an header with this + * information as the first bytes of every allocation. */ +#ifndef HAVE_MALLOC_SIZE +size_t zmalloc_size(void *ptr) { + void *realptr = (char*)ptr-PREFIX_SIZE; + size_t size = *((size_t*)realptr); + /* Assume at least that all the allocations are padded at sizeof(long) by + * the underlying allocator. */ + if (size&(sizeof(long)-1)) size += sizeof(long)-(size&(sizeof(long)-1)); + return size+PREFIX_SIZE; +} +#endif + void zfree(void *ptr) { #ifndef HAVE_MALLOC_SIZE void *realptr; diff --git a/src/zmalloc.h b/src/zmalloc.h index 7ee556a3..995814c8 100644 --- a/src/zmalloc.h +++ b/src/zmalloc.h @@ -76,4 +76,8 @@ void zmalloc_enable_thread_safeness(void); float zmalloc_get_fragmentation_ratio(void); size_t zmalloc_get_rss(void); +#ifndef HAVE_MALLOC_SIZE +size_t zmalloc_size(void *ptr); +#endif + #endif /* __ZMALLOC_H */ diff --git a/tests/test_helper.tcl b/tests/test_helper.tcl index 13b59b92..ef128ae2 100644 --- a/tests/test_helper.tcl +++ b/tests/test_helper.tcl @@ -35,6 +35,7 @@ set ::all_tests { unit/scripting unit/maxmemory unit/introspection + unit/obuf-limits } # Index to the next test to run in the ::all_tests list. set ::next_test 0 diff --git a/tests/unit/cas.tcl b/tests/unit/cas.tcl index 082fba35..95766abe 100644 --- a/tests/unit/cas.tcl +++ b/tests/unit/cas.tcl @@ -25,6 +25,16 @@ start_server {tags {"cas"}} { r exec } {} + test {EXEC fail on WATCHed key modified by SORT with STORE even if the result is empty} { + r flushdb + r lpush foo bar + r watch foo + r sort emptylist store foo + r multi + r ping + r exec + } {} + test {After successful EXEC key is no longer watched} { r set x 30 r watch x diff --git a/tests/unit/introspection.tcl b/tests/unit/introspection.tcl index db21cbd0..3daa65e0 100644 --- a/tests/unit/introspection.tcl +++ b/tests/unit/introspection.tcl @@ -1,5 +1,5 @@ start_server {tags {"introspection"}} { test {CLIENT LIST} { r client list - } {*addr=*:* fd=* idle=* flags=N db=9 sub=0 psub=0 qbuf=0 obl=0 oll=0 events=r cmd=client*} + } {*addr=*:* fd=* idle=* flags=N db=9 sub=0 psub=0 qbuf=0 obl=0 oll=0 omem=0 events=r cmd=client*} } diff --git a/tests/unit/obuf-limits.tcl b/tests/unit/obuf-limits.tcl new file mode 100644 index 00000000..e965e892 --- /dev/null +++ b/tests/unit/obuf-limits.tcl @@ -0,0 +1,73 @@ +start_server {tags {"obuf-limits"}} { + test {Client output buffer hard limit is enforced} { + r config set client-output-buffer-limit {pubsub 100000 0 0} + set rd1 [redis_deferring_client] + + $rd1 subscribe foo + set reply [$rd1 read] + assert {$reply eq "subscribe foo 1"} + + set omem 0 + while 1 { + r publish foo bar + set clients [split [r client list] "\r\n"] + set c [split [lindex $clients 1] " "] + if {![regexp {omem=([0-9]+)} $c - omem]} break + if {$omem > 200000} break + } + assert {$omem >= 99000 && $omem < 200000} + $rd1 close + } + + test {Client output buffer soft limit is not enforced if time is not overreached} { + r config set client-output-buffer-limit {pubsub 0 100000 10} + set rd1 [redis_deferring_client] + + $rd1 subscribe foo + set reply [$rd1 read] + assert {$reply eq "subscribe foo 1"} + + set omem 0 + set start_time 0 + set time_elapsed 0 + while 1 { + r publish foo bar + set clients [split [r client list] "\r\n"] + set c [split [lindex $clients 1] " "] + if {![regexp {omem=([0-9]+)} $c - omem]} break + if {$omem > 100000} { + if {$start_time == 0} {set start_time [clock seconds]} + set time_elapsed [expr {[clock seconds]-$start_time}] + if {$time_elapsed >= 5} break + } + } + assert {$omem >= 100000 && $time_elapsed >= 5 && $time_elapsed <= 10} + $rd1 close + } + + test {Client output buffer soft limit is enforced if time is overreached} { + r config set client-output-buffer-limit {pubsub 0 100000 3} + set rd1 [redis_deferring_client] + + $rd1 subscribe foo + set reply [$rd1 read] + assert {$reply eq "subscribe foo 1"} + + set omem 0 + set start_time 0 + set time_elapsed 0 + while 1 { + r publish foo bar + set clients [split [r client list] "\r\n"] + set c [split [lindex $clients 1] " "] + if {![regexp {omem=([0-9]+)} $c - omem]} break + if {$omem > 100000} { + if {$start_time == 0} {set start_time [clock seconds]} + set time_elapsed [expr {[clock seconds]-$start_time}] + if {$time_elapsed >= 10} break + } + } + assert {$omem >= 100000 && $time_elapsed < 6} + $rd1 close + } +} diff --git a/tests/unit/other.tcl b/tests/unit/other.tcl index 6bdb0b3f..14fd73ac 100644 --- a/tests/unit/other.tcl +++ b/tests/unit/other.tcl @@ -141,6 +141,11 @@ start_server {tags {"other"}} { # Reload and check waitForBgrewriteaof r + # We need to wait two seconds to avoid false positives here, otherwise + # the DEBUG LOADAOF command may read a partial file. + # Another solution would be to set the fsync policy to no, since this + # prevents write() to be delayed by the completion of fsync(). + after 2000 r debug loadaof set ttl [r ttl x] assert {$ttl > 900 && $ttl <= 1000} diff --git a/tests/unit/scripting.tcl b/tests/unit/scripting.tcl index 48a25e4e..86e51c17 100644 --- a/tests/unit/scripting.tcl +++ b/tests/unit/scripting.tcl @@ -190,6 +190,30 @@ start_server {tags {"scripting"}} { [r script load "return 'loaded'"] \ [r evalsha b534286061d4b9e4026607613b95c06c06015ae8 0] } {b534286061d4b9e4026607613b95c06c06015ae8 loaded} + + test "In the context of Lua the output of random commands gets ordered" { + r del myset + r sadd myset a b c d e f g h i l m n o p q r s t u v z aa aaa azz + r eval {return redis.call('smembers','myset')} 0 + } {a aa aaa azz b c d e f g h i l m n o p q r s t u v z} + + test "SORT is normally not re-ordered by the scripting engine" { + r del myset + r sadd myset 1 2 3 4 10 + r eval {return redis.call('sort','myset','desc')} 0 + } {10 4 3 2 1} + + test "SORT BY output gets ordered by scripting" { + r del myset + r sadd myset a b c d e f g h i l m n o p q r s t u v z aa aaa azz + r eval {return redis.call('sort','myset','by','_')} 0 + } {a aa aaa azz b c d e f g h i l m n o p q r s t u v z} + + test "SORT output containing NULLs is well handled by scripting" { + r del myset + r sadd myset a b c + r eval {return redis.call('sort','myset','by','_','get','#','get','_:*')} 0 + } {{} {} {} a b c} } start_server {tags {"scripting repl"}} { diff --git a/tests/unit/sort.tcl b/tests/unit/sort.tcl index d8f93e83..ba412254 100644 --- a/tests/unit/sort.tcl +++ b/tests/unit/sort.tcl @@ -146,10 +146,50 @@ start_server { test "SORT with STORE does not create empty lists (github issue 224)" { r flushdb r lpush foo bar - r sort foo limit 10 10 store zap + r sort foo alpha limit 10 10 store zap r exists zap } {0} + test "SORT with STORE removes key if result is empty (github issue 227)" { + r flushdb + r lpush foo bar + r sort emptylist store foo + r exists foo + } {0} + + test "SORT with BY and STORE should still order output" { + r del myset mylist + r sadd myset a b c d e f g h i l m n o p q r s t u v z aa aaa azz + r sort myset alpha by _ store mylist + r lrange mylist 0 -1 + } {a aa aaa azz b c d e f g h i l m n o p q r s t u v z} + + test "SORT will complain with numerical sorting and bad doubles (1)" { + r del myset + r sadd myset 1 2 3 4 not-a-double + set e {} + catch {r sort myset} e + set e + } {*ERR*double*} + + test "SORT will complain with numerical sorting and bad doubles (2)" { + r del myset + r sadd myset 1 2 3 4 + r mset score:1 10 score:2 20 score:3 30 score:4 not-a-double + set e {} + catch {r sort myset by score:*} e + set e + } {*ERR*double*} + + test "SORT BY sub-sorts lexicographically if score is the same" { + r del myset + r sadd myset a b c d e f g h i l m n o p q r s t u v z aa aaa azz + foreach ele {a aa aaa azz b c d e f g h i l m n o p q r s t u v z} { + set score:$ele 100 + } + r sort myset by score:* + } {a aa aaa azz b c d e f g h i l m n o p q r s t u v z} + tags {"slow"} { set num 100 set res [create_random_dataset $num lpush] diff --git a/tests/unit/type/hash.tcl b/tests/unit/type/hash.tcl index 04a5f4c7..e9f7c188 100644 --- a/tests/unit/type/hash.tcl +++ b/tests/unit/type/hash.tcl @@ -318,6 +318,14 @@ start_server {tags {"hash"}} { lappend rv [string match "ERR*not an integer*" $bigerr] } {1 1} + test {HINCRBY can detect overflows} { + set e {} + r hset hash n -9223372036854775484 + assert {[r hincrby hash n -1] == -9223372036854775485} + catch {r hincrby hash n -10000} e + set e + } {*overflow*} + test {HINCRBYFLOAT against non existing database key} { r del htest list [r hincrbyfloat htest foo 2.5]