appendonly.aof
SHORT_TERM_TODO
redis.conf.*
+release.h
+src/transfer.sh
install: dummy
cd src && $(MAKE) $@
-$(TARGETS) clean:
+clean:
+ cd src && $(MAKE) $@
+ cd deps/hiredis && $(MAKE) $@
+ cd deps/linenoise && $(MAKE) $@
+
+$(TARGETS):
cd src && $(MAKE) $@
src/help.h:
--- /dev/null
+/hiredis-test
+/hiredis-example*
+/*.o
+/*.so
+/*.dylib
+/*.a
--- /dev/null
+Copyright (c) 2006-2009, 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.
--- /dev/null
+# Hiredis Makefile
+# Copyright (C) 2010 Salvatore Sanfilippo <antirez at gmail dot com>
+# 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
+
+uname_S := $(shell sh -c 'uname -s 2>/dev/null || echo not')
+OPTIMIZATION?=-O2
+ifeq ($(uname_S),SunOS)
+ CFLAGS?= -std=c99 -pedantic $(OPTIMIZATION) -fPIC -Wall -W -D__EXTENSIONS__ -D_XPG6
+ CCLINK?= -ldl -lnsl -lsocket -lm -lpthread
+ DYLIBNAME?=libhiredis.so
+ DYLIB_MAKE_CMD?=gcc -shared -Wl,-soname,${DYLIBNAME} -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 -Wwrite-strings $(ARCH) $(PROF)
+ CCLINK?= -lm -pthread
+ 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 -Wwrite-strings $(ARCH) $(PROF)
+ CCLINK?= -lm -pthread
+ DYLIBNAME?=libhiredis.so
+ DYLIB_MAKE_CMD?=gcc -shared -Wl,-soname,${DYLIBNAME} -o ${DYLIBNAME} ${OBJ}
+ STLIBNAME?=libhiredis.a
+ STLIB_MAKE_CMD?=ar rcs ${STLIBNAME} ${OBJ}
+endif
+CCOPT= $(CFLAGS) $(CCLINK) $(ARCH) $(PROF)
+DEBUG?= -g -ggdb
+
+PREFIX?= /usr/local
+INSTALL_INC= $(PREFIX)/include/hiredis
+INSTALL_LIB= $(PREFIX)/lib
+INSTALL= cp -a
+
+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
+example-libev.o: example-libev.c hiredis.h async.h adapters/libev.h
+example-libevent.o: example-libevent.c hiredis.h async.h adapters/libevent.h
+example.o: example.c hiredis.h
+hiredis.o: hiredis.c hiredis.h net.h sds.h util.h
+sds.o: sds.c sds.h
+test.o: test.c hiredis.h
+
+${DYLIBNAME}: ${OBJ}
+ ${DYLIB_MAKE_CMD}
+
+${STLIBNAME}: ${OBJ}
+ ${STLIB_MAKE_CMD}
+
+dynamic: ${DYLIBNAME}
+static: ${STLIBNAME}
+
+# Binaries:
+hiredis-example-libevent: example-libevent.o ${DYLIBNAME}
+ $(CC) -o $@ $(CCOPT) $(DEBUG) -L. -lhiredis -levent -Wl,-rpath,. example-libevent.c
+
+hiredis-example-libev: example-libev.o ${DYLIBNAME}
+ $(CC) -o $@ $(CCOPT) $(DEBUG) -L. -lhiredis -lev -Wl,-rpath,. example-libev.c
+
+hiredis-%: %.o ${DYLIBNAME}
+ $(CC) -o $@ $(CCOPT) $(DEBUG) -L. -lhiredis -Wl,-rpath,. $<
+
+test: hiredis-test
+ ./hiredis-test
+
+.c.o:
+ $(CC) -c $(CFLAGS) $(OBJARCH) $(DEBUG) $(COMPILE_TIME) $<
+
+clean:
+ 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)
+
+32bit:
+ @echo ""
+ @echo "WARNING: if it fails under Linux you probably need to install libc6-dev-i386"
+ @echo ""
+ make ARCH="-m32"
+
+gprof:
+ make PROF="-pg"
+
+gcov:
+ make PROF="-fprofile-arcs -ftest-coverage"
+
+noopt:
+ make OPTIMIZATION=""
--- /dev/null
+# HIREDIS
+
+Hiredis is a minimalistic C client library for the [Redis](http://redis.io/) database.
+
+It is minimalistic because it just adds minimal support for the protocol, but
+at the same time it uses an high level printf-alike API in order to make it
+much higher level than otherwise suggested by its minimal code base and the
+lack of explicit bindings for every Redis command.
+
+Apart from supporting sending commands and receiving replies, it comes with
+a reply parser that is decoupled from the I/O layer. It
+is a stream parser designed for easy reusability, which can for instance be used
+in higher level language bindings for efficient reply parsing.
+
+Hiredis only supports the binary-safe Redis protocol, so you can use it with any
+Redis version >= 1.2.0.
+
+The library comes with multiple APIs. There is the
+*synchronous API*, the *asynchronous API* and the *reply parsing API*.
+
+## UPGRADING
+
+Version 0.9.0 is a major overhaul of hiredis in every aspect. However, upgrading existing
+code using hiredis should not be a big pain. The key thing to keep in mind when
+upgrading is that hiredis >= 0.9.0 uses a `redisContext*` to keep state, in contrast to
+the stateless 0.0.1 that only has a file descriptor to work with.
+
+## Synchronous API
+
+To consume the synchronous API, there are only a few function calls that need to be introduced:
+
+ redisContext *redisConnect(const char *ip, int port);
+ void *redisCommand(redisContext *c, const char *format, ...);
+ void freeReplyObject(void *reply);
+
+### Connecting
+
+The function `redisConnect` is used to create a so-called `redisContext`. The context is where
+Hiredis holds state for a connection. The `redisContext` struct has an `error` field that is
+non-NULL when the connection is in an error state. It contains a string with a textual
+representation of the error. After trying to connect to Redis using `redisConnect` you should
+check the `error` field to see if establishing the connection was successful:
+
+ redisContext *c = redisConnect("127.0.0.1", 6379);
+ if (c->error != NULL) {
+ printf("Error: %s\n", c->error);
+ // handle error
+ }
+
+### Sending commands
+
+There are several ways to issue commands to Redis. The first that will be introduced is
+`redisCommand`. This function takes a format similar to printf. In the simplest form,
+it is used like this:
+
+ reply = redisCommand(context, "SET foo bar");
+
+The specifier `%s` interpolates a string in the command, and uses `strlen` to
+determine the length of the string:
+
+ reply = redisCommand(context, "SET foo %s", value);
+
+When you need to pass binary safe strings in a command, the `%b` specifier can be
+used. Together with a pointer to the string, it requires a `size_t` length argument
+of the string:
+
+ reply = redisCommand(context, "SET foo %b", value, valuelen);
+
+Internally, Hiredis splits the command in different arguments and will
+convert it to the protocol used to communicate with Redis.
+One or more spaces separates arguments, so you can use the specifiers
+anywhere in an argument:
+
+ reply = redisCommand("SET key:%s %s", myid, value);
+
+### Using replies
+
+The return value of `redisCommand` holds a reply when the command was
+successfully executed. When the return value is `NULL`, the `error` field
+in the context can be used to find out what was the cause of failure.
+Once an error is returned the context cannot be reused and you should set up
+a new connection.
+
+The standard replies that `redisCommand` are of the type `redisReply`. The
+`type` field in the `redisReply` should be used to test what kind of reply
+was received:
+
+* **`REDIS_REPLY_STATUS`**:
+ * The command replied with a status reply. The status string can be accessed using `reply->str`.
+ The length of this string can be accessed using `reply->len`.
+
+* **`REDIS_REPLY_ERROR`**:
+ * The command replied with an error. The error string can be accessed identical to `REDIS_REPLY_STATUS`.
+
+* **`REDIS_REPLY_INTEGER`**:
+ * The command replied with an integer. The integer value can be accessed using the
+ `reply->integer` field of type `long long`.
+
+* **`REDIS_REPLY_NIL`**:
+ * The command replied with a **nil** object. There is no data to access.
+
+* **`REDIS_REPLY_STRING`**:
+ * A bulk (string) reply. The value of the reply can be accessed using `reply->str`.
+ The length of this string can be accessed using `reply->len`.
+
+* **`REDIS_REPLY_ARRAY`**:
+ * A multi bulk reply. The number of elements in the multi bulk reply is stored in
+ `reply->elements`. Every element in the multi bulk reply is a `redisReply` object as well
+ and can be accessed via `reply->elements[..index..]`.
+ Redis may reply with nested arrays but this is fully supported.
+
+Replies should be freed using the `freeReplyObject()` function.
+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).
+
+### Cleaning up
+
+To disconnect and free the context the following function can be used:
+
+ void redisFree(redisContext *c);
+
+This function immediately closes the socket and then free's the allocations done in
+creating the context.
+
+### Sending commands (cont'd)
+
+Together with `redisCommand`, the function `redisCommandArgv` can be used to issue commands.
+It has the following prototype:
+
+ void *redisCommandArgv(redisContext *c, int argc, const char **argv, const size_t *argvlen);
+
+It takes the number of arguments `argc`, an array of strings `argv` and the lengths of the
+arguments `argvlen`. For convenience, `argvlen` may be set to `NULL` and the function will
+use `strlen(3)` on every argument to determine its length. Obviously, when any of the arguments
+need to be binary safe, the entire array of lengths `argvlen` should be provided.
+
+The return value has the same semantic as `redisCommand`.
+
+### Pipelining
+
+To explain how Hiredis supports pipelining in a blocking connection, there needs to be
+understanding of the internal execution flow.
+
+When any of the functions in the `redisCommand` family is called, Hiredis first formats the
+command according to the Redis protocol. The formatted command is then put in the output buffer
+of the context. This output buffer is dynamic, so it can hold any number of commands.
+After the command is put in the output buffer, `redisGetReply` is called. This function has the
+following two execution paths:
+
+1. The input buffer is non-empty:
+ * Try to parse a single reply from the input buffer and return it
+ * If no reply could be parsed, continue at *2*
+2. The input buffer is empty:
+ * Write the **entire** output buffer to the socket
+ * Read from the socket until a single reply could be parsed
+
+The function `redisGetReply` is exported as part of the Hiredis API and can be used when a reply
+is expected on the socket. To pipeline commands, the only things that needs to be done is
+filling up the output buffer. For this cause, two commands can be used that are identical
+to the `redisCommand` family, apart from not returning a reply:
+
+ void redisAppendCommand(redisContext *c, const char *format, ...);
+ void redisAppendCommandArgv(redisContext *c, int argc, const char **argv, const size_t *argvlen);
+
+After calling either function one or more times, `redisGetReply` can be used to receive the
+subsequent replies. The return value for this function is either `REDIS_OK` or `REDIS_ERR`, where
+the latter means an error occurred while reading a reply. Just as with the other commands,
+the `error` field in the context can be used to find out what the cause of this error is.
+
+The following examples shows a simple pipeline (resulting in only a single call to `write(2)` and
+a single call to `write(2)`):
+
+ redisReply *reply;
+ redisAppendCommand(context,"SET foo bar");
+ redisAppendCommand(context,"GET foo");
+ redisGetReply(context,&reply); // reply for SET
+ freeReplyObject(reply);
+ redisGetReply(context,&reply); // reply for GET
+ freeReplyObject(reply);
+
+This API can also be used to implement a blocking subscriber:
+
+ reply = redisCommand(context,"SUBSCRIBE foo");
+ freeReplyObject(reply);
+ while(redisGetReply(context,&reply) == REDIS_OK) {
+ // consume message
+ freeReplyObject(reply);
+ }
+
+## Asynchronous API
+
+Hiredis comes with an asynchronous API that works easily with any event library.
+Examples are bundled that show using Hiredis with [libev](http://software.schmorp.de/pkg/libev.html)
+and [libevent](http://monkey.org/~provos/libevent/).
+
+### Connecting
+
+The function `redisAsyncConnect` can be used to establish a non-blocking connection to
+Redis. It returns a pointer to the newly created `redisAsyncContext` struct. The `error` field
+should be checked after creation to see if there were errors creating the connection.
+Because the connection that will be created is non-blocking, the kernel is not able to
+instantly return if the specified host and port is able to accept a connection.
+
+ redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379);
+ if (c->error != NULL) {
+ printf("Error: %s\n", c->error);
+ // handle error
+ }
+
+The asynchronous context can hold a disconnect callback function that is called when the
+connection is disconnected (either because of an error or per user request). This function should
+have the following prototype:
+
+ void(const redisAsyncContext *c, int status);
+
+On a disconnect, the `status` argument is set to `REDIS_OK` when disconnection was initiated by the
+user, or `REDIS_ERR` when the disconnection was caused by an error. When it is `REDIS_ERR`, the `error`
+field in the context can be accessed to find out the cause of the error.
+
+The context object is always free'd after the disconnect callback fired. When a reconnect is needed,
+the disconnect callback is a good point to do so.
+
+Setting the disconnect callback can only be done once per context. For subsequent calls it will
+return `REDIS_ERR`. The function to set the disconnect callback has the following prototype:
+
+ int redisAsyncSetDisconnectCallback(redisAsyncContext *ac, redisDisconnectCallback *fn);
+
+### Sending commands and their callbacks
+
+In an asynchronous context, commands are automatically pipelined due to the nature of an event loop.
+Therefore, unlike the synchronous API, there is only a single way to send commands.
+Because commands are sent to Redis asynchronously, issuing a command requires a callback function
+that is called when the reply is received. Reply callbacks should have the following prototype:
+
+ void(redisAsyncContext *c, void *reply, void *privdata);
+
+The `privdata` argument can be used to curry arbitrary data to the callback from the point where
+the command is initially queued for execution.
+
+The functions that can be used to issue commands in an asynchronous context are:
+
+ int redisAsyncCommand(
+ redisAsyncContext *ac, redisCallbackFn *fn, void *privdata,
+ const char *format, ...);
+ int redisAsyncCommandArgv(
+ redisAsyncContext *ac, redisCallbackFn *fn, void *privdata,
+ int argc, const char **argv, const size_t *argvlen);
+
+Both functions work like their blocking counterparts. The return value is `REDIS_OK` when the command
+was successfully added to the output buffer and `REDIS_ERR` otherwise. Example: when the connection
+is being disconnected per user-request, no new commands may be added to the output buffer and `REDIS_ERR` is
+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.
+
+All pending callbacks are called with a `NULL` reply when the context encountered an error.
+
+### Disconnecting
+
+An asynchronous connection can be terminated using:
+
+ void redisAsyncDisconnect(redisAsyncContext *ac);
+
+When this function is called, the connection is **not** immediately terminated. Instead, new
+commands are no longer accepted and the connection is only terminated when all pending commands
+have been written to the socket, their respective replies have been read and their respective
+callbacks have been executed. After this, the disconnection callback is executed with the
+`REDIS_OK` status and the context object is free'd.
+
+### Hooking it up to event library *X*
+
+There are a few hooks that need to be set on the context object after it is created.
+See the `adapters/` directory for bindings to *libev* and *libevent*.
+
+## Reply parsing API
+
+To be done.
+
+## AUTHORS
+
+Hiredis was written by Salvatore Sanfilippo (antirez at gmail) and
+Pieter Noordhuis (pcnoordhuis at gmail) and is released under the BSD license.
--- /dev/null
+- add redisCommandVector()
+- add support for pipelining
--- /dev/null
+#include <sys/types.h>
+#include <ev.h>
+#include "../hiredis.h"
+#include "../async.h"
+
+typedef struct redisLibevEvents {
+ redisAsyncContext *context;
+ struct ev_loop *loop;
+ int reading, writing;
+ ev_io rev, wev;
+} redisLibevEvents;
+
+void redisLibevReadEvent(struct ev_loop *loop, ev_io *watcher, int revents) {
+ ((void)loop); ((void)revents);
+ redisLibevEvents *e = watcher->data;
+ redisAsyncHandleRead(e->context);
+}
+
+void redisLibevWriteEvent(struct ev_loop *loop, ev_io *watcher, int revents) {
+ ((void)loop); ((void)revents);
+ redisLibevEvents *e = watcher->data;
+ redisAsyncHandleWrite(e->context);
+}
+
+void redisLibevAddRead(void *privdata) {
+ redisLibevEvents *e = privdata;
+ if (!e->reading) {
+ e->reading = 1;
+ ev_io_start(e->loop,&e->rev);
+ }
+}
+
+void redisLibevDelRead(void *privdata) {
+ redisLibevEvents *e = privdata;
+ if (e->reading) {
+ e->reading = 0;
+ ev_io_stop(e->loop,&e->rev);
+ }
+}
+
+void redisLibevAddWrite(void *privdata) {
+ redisLibevEvents *e = privdata;
+ if (!e->writing) {
+ e->writing = 1;
+ ev_io_start(e->loop,&e->wev);
+ }
+}
+
+void redisLibevDelWrite(void *privdata) {
+ redisLibevEvents *e = privdata;
+ if (e->writing) {
+ e->writing = 0;
+ ev_io_stop(e->loop,&e->wev);
+ }
+}
+
+void redisLibevCleanup(void *privdata) {
+ redisLibevEvents *e = privdata;
+ redisLibevDelRead(privdata);
+ redisLibevDelWrite(privdata);
+ free(e);
+}
+
+int redisLibevAttach(redisAsyncContext *ac, struct ev_loop *loop) {
+ redisContext *c = &(ac->c);
+ redisLibevEvents *e;
+
+ /* Nothing should be attached when something is already attached */
+ if (ac->data != NULL)
+ return REDIS_ERR;
+
+ /* Create container for context and r/w events */
+ e = malloc(sizeof(*e));
+ e->context = ac;
+ e->loop = loop;
+ e->reading = e->writing = 0;
+ e->rev.data = e;
+ 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->data = e;
+
+ /* Initialize read/write events */
+ ev_io_init(&e->rev,redisLibevReadEvent,c->fd,EV_READ);
+ ev_io_init(&e->wev,redisLibevWriteEvent,c->fd,EV_WRITE);
+ return REDIS_OK;
+}
--- /dev/null
+#include <sys/types.h>
+#include <event.h>
+#include "../hiredis.h"
+#include "../async.h"
+
+typedef struct redisLibeventEvents {
+ redisAsyncContext *context;
+ struct event rev, wev;
+} redisLibeventEvents;
+
+void redisLibeventReadEvent(int fd, short event, void *arg) {
+ ((void)fd); ((void)event);
+ redisLibeventEvents *e = arg;
+ redisAsyncHandleRead(e->context);
+}
+
+void redisLibeventWriteEvent(int fd, short event, void *arg) {
+ ((void)fd); ((void)event);
+ redisLibeventEvents *e = arg;
+ redisAsyncHandleWrite(e->context);
+}
+
+void redisLibeventAddRead(void *privdata) {
+ redisLibeventEvents *e = privdata;
+ event_add(&e->rev,NULL);
+}
+
+void redisLibeventDelRead(void *privdata) {
+ redisLibeventEvents *e = privdata;
+ event_del(&e->rev);
+}
+
+void redisLibeventAddWrite(void *privdata) {
+ redisLibeventEvents *e = privdata;
+ event_add(&e->wev,NULL);
+}
+
+void redisLibeventDelWrite(void *privdata) {
+ redisLibeventEvents *e = privdata;
+ event_del(&e->wev);
+}
+
+void redisLibeventCleanup(void *privdata) {
+ redisLibeventEvents *e = privdata;
+ event_del(&e->rev);
+ event_del(&e->wev);
+ free(e);
+}
+
+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->data != NULL)
+ return REDIS_ERR;
+
+ /* Create container for context and r/w events */
+ e = malloc(sizeof(*e));
+ 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->data = e;
+
+ /* Initialize and install read/write events */
+ event_set(&e->rev,c->fd,EV_READ,redisLibeventReadEvent,e);
+ event_set(&e->wev,c->fd,EV_WRITE,redisLibeventWriteEvent,e);
+ event_base_set(base,&e->rev);
+ event_base_set(base,&e->wev);
+ return REDIS_OK;
+}
--- /dev/null
+/*
+ * Copyright (c) 2009-2010, Salvatore Sanfilippo <antirez at gmail dot com>
+ * 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 <string.h>
+#include <assert.h>
+#include "async.h"
+#include "sds.h"
+#include "util.h"
+
+/* Forward declaration of function in hiredis.c */
+void __redisAppendCommand(redisContext *c, char *cmd, size_t len);
+
+static redisAsyncContext *redisAsyncInitialize(redisContext *c) {
+ redisAsyncContext *ac = realloc(c,sizeof(redisAsyncContext));
+ /* Set all bytes in the async part of the context to 0 */
+ memset(ac+sizeof(redisContext),0,sizeof(redisAsyncContext)-sizeof(redisContext));
+ return ac;
+}
+
+/* We want the error field to be accessible directly instead of requiring
+ * an indirection to the redisContext struct. */
+static void __redisAsyncCopyError(redisAsyncContext *ac) {
+ redisContext *c = &(ac->c);
+ ac->err = c->err;
+ ac->errstr = c->errstr;
+}
+
+redisAsyncContext *redisAsyncConnect(const char *ip, int port) {
+ redisContext *c = redisConnectNonBlock(ip,port);
+ redisAsyncContext *ac = redisAsyncInitialize(c);
+ __redisAsyncCopyError(ac);
+ return ac;
+}
+
+redisAsyncContext *redisAsyncConnectUnix(const char *path) {
+ redisContext *c = redisConnectUnixNonBlock(path);
+ redisAsyncContext *ac = redisAsyncInitialize(c);
+ __redisAsyncCopyError(ac);
+ return ac;
+}
+
+int redisAsyncSetReplyObjectFunctions(redisAsyncContext *ac, redisReplyObjectFunctions *fn) {
+ redisContext *c = &(ac->c);
+ return redisSetReplyObjectFunctions(c,fn);
+}
+
+int redisAsyncSetDisconnectCallback(redisAsyncContext *ac, redisDisconnectCallback *fn) {
+ if (ac->onDisconnect == NULL) {
+ ac->onDisconnect = fn;
+ return REDIS_OK;
+ }
+ return REDIS_ERR;
+}
+
+/* Helper functions to push/shift callbacks */
+static int __redisPushCallback(redisCallbackList *list, redisCallback *source) {
+ redisCallback *cb;
+
+ /* Copy callback from stack to heap */
+ cb = calloc(1,sizeof(*cb));
+ if (!cb) redisOOM();
+ if (source != NULL) {
+ cb->fn = source->fn;
+ cb->privdata = source->privdata;
+ }
+
+ /* Store callback in list */
+ if (list->head == NULL)
+ list->head = cb;
+ if (list->tail != NULL)
+ list->tail->next = cb;
+ list->tail = cb;
+ return REDIS_OK;
+}
+
+static int __redisShiftCallback(redisCallbackList *list, redisCallback *target) {
+ redisCallback *cb = list->head;
+ if (cb != NULL) {
+ list->head = cb->next;
+ if (cb == list->tail)
+ list->tail = NULL;
+
+ /* Copy callback from heap to stack */
+ if (target != NULL)
+ memcpy(target,cb,sizeof(*cb));
+ free(cb);
+ return REDIS_OK;
+ }
+ return REDIS_ERR;
+}
+
+/* Tries to do a clean disconnect from Redis, meaning it stops new commands
+ * from being issued, but tries to flush the output buffer and execute
+ * callbacks for all remaining replies.
+ *
+ * This functions is generally called from within a callback, so the
+ * processCallbacks function will pick up the flag when there are no
+ * more replies. */
+void redisAsyncDisconnect(redisAsyncContext *ac) {
+ redisContext *c = &(ac->c);
+ c->flags |= REDIS_DISCONNECTING;
+}
+
+/* Helper function to make the disconnect happen and clean up. */
+static void __redisAsyncDisconnect(redisAsyncContext *ac) {
+ redisContext *c = &(ac->c);
+ redisCallback cb;
+ int status;
+
+ /* Make sure error is accessible if there is any */
+ __redisAsyncCopyError(ac);
+ status = (ac->err == 0) ? REDIS_OK : REDIS_ERR;
+
+ if (status == REDIS_OK) {
+ /* When the connection is cleanly disconnected, there should not
+ * be pending callbacks. */
+ assert(__redisShiftCallback(&ac->replies,NULL) == REDIS_ERR);
+ } else {
+ /* Callbacks should not be able to issue new commands. */
+ c->flags |= REDIS_DISCONNECTING;
+
+ /* Execute pending callbacks with NULL reply. */
+ while (__redisShiftCallback(&ac->replies,&cb) == REDIS_OK) {
+ if (cb.fn != NULL)
+ cb.fn(ac,NULL,cb.privdata);
+ }
+ }
+
+ /* Signal event lib to clean up */
+ if (ac->evCleanup) ac->evCleanup(ac->data);
+
+ /* Execute callback with proper status */
+ if (ac->onDisconnect) ac->onDisconnect(ac,status);
+
+ /* Cleanup self */
+ redisFree(c);
+}
+
+void redisProcessCallbacks(redisAsyncContext *ac) {
+ redisContext *c = &(ac->c);
+ redisCallback cb;
+ void *reply = NULL;
+ int status;
+
+ while((status = redisGetReply(c,&reply)) == REDIS_OK) {
+ if (reply == NULL) {
+ /* When the connection is being disconnected and there are
+ * no more replies, this is the cue to really disconnect. */
+ if (c->flags & REDIS_DISCONNECTING && sdslen(c->obuf) == 0) {
+ __redisAsyncDisconnect(ac);
+ return;
+ }
+
+ /* When the connection is not being disconnected, simply stop
+ * trying to get replies and wait for the next loop tick. */
+ break;
+ }
+
+ /* Shift callback and execute it */
+ assert(__redisShiftCallback(&ac->replies,&cb) == REDIS_OK);
+ if (cb.fn != NULL) {
+ cb.fn(ac,reply,cb.privdata);
+ } else {
+ c->fn->freeObject(reply);
+ }
+ }
+
+ /* Disconnect when there was an error reading the reply */
+ if (status != REDIS_OK)
+ __redisAsyncDisconnect(ac);
+}
+
+/* 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 (redisBufferRead(c) == REDIS_ERR) {
+ __redisAsyncDisconnect(ac);
+ } else {
+ /* Always re-schedule reads */
+ if (ac->evAddRead) ac->evAddRead(ac->data);
+ redisProcessCallbacks(ac);
+ }
+}
+
+void redisAsyncHandleWrite(redisAsyncContext *ac) {
+ redisContext *c = &(ac->c);
+ int done = 0;
+
+ if (redisBufferWrite(c,&done) == REDIS_ERR) {
+ __redisAsyncDisconnect(ac);
+ } else {
+ /* Continue writing when not done, stop writing otherwise */
+ if (!done) {
+ if (ac->evAddWrite) ac->evAddWrite(ac->data);
+ } else {
+ if (ac->evDelWrite) ac->evDelWrite(ac->data);
+ }
+
+ /* Always schedule reads when something was written */
+ if (ac->evAddRead) ac->evAddRead(ac->data);
+ }
+}
+
+/* Helper function for the redisAsyncCommand* family of functions.
+ *
+ * Write a formatted command to the output buffer and register the provided
+ * callback function with the context.
+ */
+static int __redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, char *cmd, size_t len) {
+ redisContext *c = &(ac->c);
+ redisCallback cb;
+
+ /* Don't accept new commands when the connection is lazily closed. */
+ if (c->flags & REDIS_DISCONNECTING) return REDIS_ERR;
+ __redisAppendCommand(c,cmd,len);
+
+ /* Store callback */
+ cb.fn = fn;
+ cb.privdata = privdata;
+ __redisPushCallback(&ac->replies,&cb);
+
+ /* Always schedule a write when the write buffer is non-empty */
+ if (ac->evAddWrite) ac->evAddWrite(ac->data);
+
+ return REDIS_OK;
+}
+
+int redisvAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *format, va_list ap) {
+ char *cmd;
+ int len;
+ int status;
+ len = redisvFormatCommand(&cmd,format,ap);
+ status = __redisAsyncCommand(ac,fn,privdata,cmd,len);
+ free(cmd);
+ return status;
+}
+
+int redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *format, ...) {
+ va_list ap;
+ int status;
+ va_start(ap,format);
+ status = redisvAsyncCommand(ac,fn,privdata,format,ap);
+ va_end(ap);
+ return status;
+}
+
+int redisAsyncCommandArgv(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, int argc, const char **argv, const size_t *argvlen) {
+ char *cmd;
+ int len;
+ int status;
+ len = redisFormatCommandArgv(&cmd,argc,argv,argvlen);
+ status = __redisAsyncCommand(ac,fn,privdata,cmd,len);
+ free(cmd);
+ return status;
+}
--- /dev/null
+/*
+ * Copyright (c) 2009-2010, Salvatore Sanfilippo <antirez at gmail dot com>
+ * 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 __HIREDIS_ASYNC_H
+#define __HIREDIS_ASYNC_H
+#include "hiredis.h"
+
+struct redisAsyncContext; /* need forward declaration of redisAsyncContext */
+
+/* Reply callback prototype and container */
+typedef void (redisCallbackFn)(struct redisAsyncContext*, void*, void*);
+typedef struct redisCallback {
+ struct redisCallback *next; /* simple singly linked list */
+ redisCallbackFn *fn;
+ void *privdata;
+} redisCallback;
+
+/* List of callbacks for either regular replies or pub/sub */
+typedef struct redisCallbackList {
+ redisCallback *head, *tail;
+} redisCallbackList;
+
+/* Disconnect callback prototype */
+typedef void (redisDisconnectCallback)(const struct redisAsyncContext*, int status);
+
+/* Context for an async connection to Redis */
+typedef struct redisAsyncContext {
+ /* Hold the regular context, so it can be realloc'ed. */
+ redisContext c;
+
+ /* Setup error flags so they can be used directly. */
+ int err;
+ char *errstr;
+
+ /* Called when the library expects to start reading/writing.
+ * The supplied functions should be idempotent. */
+ void (*evAddRead)(void *privdata);
+ void (*evDelRead)(void *privdata);
+ void (*evAddWrite)(void *privdata);
+ void (*evDelWrite)(void *privdata);
+ void (*evCleanup)(void *privdata);
+ void *data;
+
+ /* Called when either the connection is terminated due to an error or per
+ * user request. The status is set accordingly (REDIS_OK, REDIS_ERR). */
+ redisDisconnectCallback *onDisconnect;
+
+ /* Reply callbacks */
+ redisCallbackList replies;
+} redisAsyncContext;
+
+/* Functions that proxy to hiredis */
+redisAsyncContext *redisAsyncConnect(const char *ip, int port);
+int redisAsyncSetReplyObjectFunctions(redisAsyncContext *ac, redisReplyObjectFunctions *fn);
+int redisAsyncSetDisconnectCallback(redisAsyncContext *ac, redisDisconnectCallback *fn);
+void redisAsyncDisconnect(redisAsyncContext *ac);
+
+/* Handle read/write events */
+void redisAsyncHandleRead(redisAsyncContext *ac);
+void redisAsyncHandleWrite(redisAsyncContext *ac);
+
+/* Command functions for an async context. Write the command to the
+ * output buffer and register the provided callback. */
+int redisvAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *format, va_list ap);
+int redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *format, ...);
+int redisAsyncCommandArgv(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, int argc, const char **argv, const size_t *argvlen);
+
+#endif
--- /dev/null
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <signal.h>
+#include "hiredis.h"
+#include "async.h"
+#include "adapters/libev.h"
+
+void getCallback(redisAsyncContext *c, void *r, void *privdata) {
+ redisReply *reply = r;
+ if (reply == NULL) return;
+ printf("argv[%s]: %s\n", (char*)privdata, reply->str);
+
+ /* Disconnect after receiving the reply to GET */
+ redisAsyncDisconnect(c);
+}
+
+void disconnectCallback(const redisAsyncContext *c, int status) {
+ if (status != REDIS_OK) {
+ printf("Error: %s\n", c->errstr);
+ }
+}
+
+int main (int argc, char **argv) {
+ signal(SIGPIPE, SIG_IGN);
+ struct ev_loop *loop = ev_default_loop(0);
+
+ redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379);
+ if (c->err) {
+ /* Let *c leak for now... */
+ printf("Error: %s\n", c->errstr);
+ return 1;
+ }
+
+ redisLibevAttach(c,loop);
+ redisAsyncSetDisconnectCallback(c,disconnectCallback);
+ redisAsyncCommand(c, NULL, NULL, "SET key %b", argv[argc-1], strlen(argv[argc-1]));
+ redisAsyncCommand(c, getCallback, (char*)"end-1", "GET key");
+ ev_loop(loop, 0);
+ return 0;
+}
--- /dev/null
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <signal.h>
+#include "hiredis.h"
+#include "async.h"
+#include "adapters/libevent.h"
+
+void getCallback(redisAsyncContext *c, void *r, void *privdata) {
+ redisReply *reply = r;
+ if (reply == NULL) return;
+ printf("argv[%s]: %s\n", (char*)privdata, reply->str);
+
+ /* Disconnect after receiving the reply to GET */
+ redisAsyncDisconnect(c);
+}
+
+void disconnectCallback(const redisAsyncContext *c, int status) {
+ if (status != REDIS_OK) {
+ printf("Error: %s\n", c->errstr);
+ }
+}
+
+int main (int argc, char **argv) {
+ signal(SIGPIPE, SIG_IGN);
+ struct event_base *base = event_base_new();
+
+ redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379);
+ if (c->err) {
+ /* Let *c leak for now... */
+ printf("Error: %s\n", c->errstr);
+ return 1;
+ }
+
+ redisLibeventAttach(c,base);
+ redisAsyncSetDisconnectCallback(c,disconnectCallback);
+ redisAsyncCommand(c, NULL, NULL, "SET key %b", argv[argc-1], strlen(argv[argc-1]));
+ redisAsyncCommand(c, getCallback, (char*)"end-1", "GET key");
+ event_base_dispatch(base);
+ return 0;
+}
--- /dev/null
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "hiredis.h"
+
+int main(void) {
+ unsigned int j;
+ redisContext *c;
+ redisReply *reply;
+
+ c = redisConnect((char*)"127.0.0.1", 6379);
+ if (c->err) {
+ printf("Connection error: %s\n", c->errstr);
+ exit(1);
+ }
+
+ /* PING server */
+ reply = redisCommand(c,"PING");
+ printf("PONG: %s\n", reply->str);
+ freeReplyObject(reply);
+
+ /* Set a key */
+ reply = redisCommand(c,"SET %s %s", "foo", "hello world");
+ printf("SET: %s\n", reply->str);
+ freeReplyObject(reply);
+
+ /* Set a key using binary safe API */
+ reply = redisCommand(c,"SET %b %b", "bar", 3, "hello", 5);
+ printf("SET (binary API): %s\n", reply->str);
+ freeReplyObject(reply);
+
+ /* Try a GET and two INCR */
+ reply = redisCommand(c,"GET foo");
+ printf("GET foo: %s\n", reply->str);
+ freeReplyObject(reply);
+
+ reply = redisCommand(c,"INCR counter");
+ printf("INCR counter: %lld\n", reply->integer);
+ freeReplyObject(reply);
+ /* again ... */
+ reply = redisCommand(c,"INCR counter");
+ printf("INCR counter: %lld\n", reply->integer);
+ freeReplyObject(reply);
+
+ /* Create a list of numbers, from 0 to 9 */
+ reply = redisCommand(c,"DEL mylist");
+ freeReplyObject(reply);
+ for (j = 0; j < 10; j++) {
+ char buf[64];
+
+ snprintf(buf,64,"%d",j);
+ reply = redisCommand(c,"LPUSH mylist element-%s", buf);
+ freeReplyObject(reply);
+ }
+
+ /* Let's check what we have inside the list */
+ reply = redisCommand(c,"LRANGE mylist 0 -1");
+ if (reply->type == REDIS_REPLY_ARRAY) {
+ for (j = 0; j < reply->elements; j++) {
+ printf("%u) %s\n", j, reply->element[j]->str);
+ }
+ }
+ freeReplyObject(reply);
+
+ return 0;
+}
--- /dev/null
+#ifndef _REDIS_FMACRO_H
+#define _REDIS_FMACRO_H
+
+#define _BSD_SOURCE
+
+#ifdef __linux__
+#define _XOPEN_SOURCE 700
+#else
+#define _XOPEN_SOURCE
+#endif
+
+#define _LARGEFILE_SOURCE
+#define _FILE_OFFSET_BITS 64
+
+#endif
--- /dev/null
+/*
+ * Copyright (c) 2009-2010, Salvatore Sanfilippo <antirez at gmail dot com>
+ * 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 <string.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <assert.h>
+#include <errno.h>
+
+#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 */
+ unsigned int pos; /* buffer cursor */
+
+ redisReadTask rstack[3]; /* stack of read tasks */
+ int ridx; /* index of stack */
+} 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. */
+static redisReplyObjectFunctions defaultFunctions = {
+ createStringObject,
+ createArrayObject,
+ createIntegerObject,
+ createNilObject,
+ freeReplyObject
+};
+
+/* Create a reply object */
+static redisReply *createReplyObject(int type) {
+ redisReply *r = calloc(sizeof(*r),1);
+
+ if (!r) redisOOM();
+ r->type = type;
+ return r;
+}
+
+/* Free a reply object */
+void freeReplyObject(void *reply) {
+ redisReply *r = reply;
+ size_t j;
+
+ switch(r->type) {
+ 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);
+ break;
+ default:
+ 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 ||
+ task->type == REDIS_REPLY_STATUS ||
+ task->type == REDIS_REPLY_STRING);
+
+ /* Copy string value */
+ memcpy(value,str,len);
+ value[len] = '\0';
+ r->str = value;
+ r->len = len;
+
+ if (task->parent) {
+ redisReply *parent = task->parent;
+ assert(parent->type == REDIS_REPLY_ARRAY);
+ parent->element[task->idx] = r;
+ }
+ return r;
+}
+
+static void *createArrayObject(const redisReadTask *task, int elements) {
+ redisReply *r = createReplyObject(REDIS_REPLY_ARRAY);
+ r->elements = elements;
+ if ((r->element = calloc(sizeof(redisReply*),elements)) == NULL)
+ redisOOM();
+ if (task->parent) {
+ redisReply *parent = task->parent;
+ assert(parent->type == REDIS_REPLY_ARRAY);
+ parent->element[task->idx] = r;
+ }
+ return r;
+}
+
+static void *createIntegerObject(const redisReadTask *task, long long value) {
+ redisReply *r = createReplyObject(REDIS_REPLY_INTEGER);
+ r->integer = value;
+ if (task->parent) {
+ redisReply *parent = task->parent;
+ assert(parent->type == REDIS_REPLY_ARRAY);
+ parent->element[task->idx] = r;
+ }
+ return r;
+}
+
+static void *createNilObject(const redisReadTask *task) {
+ redisReply *r = createReplyObject(REDIS_REPLY_NIL);
+ if (task->parent) {
+ redisReply *parent = task->parent;
+ assert(parent->type == REDIS_REPLY_ARRAY);
+ parent->element[task->idx] = r;
+ }
+ return r;
+}
+
+static char *readBytes(redisReader *r, unsigned int bytes) {
+ char *p;
+ if (sdslen(r->buf)-r->pos >= bytes) {
+ p = r->buf+r->pos;
+ r->pos += bytes;
+ return p;
+ }
+ return NULL;
+}
+
+static char *seekNewline(char *s) {
+ /* Find pointer to \r\n without strstr */
+ while (s != NULL) {
+ s = strchr(s,'\r');
+ if (s != NULL) {
+ if (s[1] == '\n')
+ break;
+ else
+ s++;
+ } else {
+ break;
+ }
+ }
+ return s;
+}
+
+static char *readLine(redisReader *r, int *_len) {
+ char *p, *s;
+ int len;
+
+ p = r->buf+r->pos;
+ s = seekNewline(p);
+ if (s != NULL) {
+ len = s-(r->buf+r->pos);
+ r->pos += len+2; /* skip \r\n */
+ if (_len) *_len = len;
+ return p;
+ }
+ return NULL;
+}
+
+static void moveToNextTask(redisReader *r) {
+ redisReadTask *cur, *prv;
+ while (r->ridx >= 0) {
+ /* Return a.s.a.p. when the stack is now empty. */
+ if (r->ridx == 0) {
+ r->ridx--;
+ return;
+ }
+
+ cur = &(r->rstack[r->ridx]);
+ prv = &(r->rstack[r->ridx-1]);
+ assert(prv->type == REDIS_REPLY_ARRAY);
+ if (cur->idx == prv->elements-1) {
+ r->ridx--;
+ } else {
+ /* Reset the type because the next item can be anything */
+ assert(cur->idx < prv->elements);
+ cur->type = -1;
+ cur->elements = -1;
+ cur->idx++;
+ return;
+ }
+ }
+}
+
+static int processLineItem(redisReader *r) {
+ redisReadTask *cur = &(r->rstack[r->ridx]);
+ void *obj;
+ char *p;
+ int len;
+
+ if ((p = readLine(r,&len)) != NULL) {
+ if (r->fn) {
+ if (cur->type == REDIS_REPLY_INTEGER) {
+ obj = r->fn->createInteger(cur,strtoll(p,NULL,10));
+ } else {
+ obj = r->fn->createString(cur,p,len);
+ }
+ } else {
+ obj = (void*)(size_t)(cur->type);
+ }
+
+ /* If there is no root yet, register this object as root. */
+ if (r->reply == NULL)
+ r->reply = obj;
+ moveToNextTask(r);
+ return 0;
+ }
+ return -1;
+}
+
+static int processBulkItem(redisReader *r) {
+ redisReadTask *cur = &(r->rstack[r->ridx]);
+ void *obj = NULL;
+ char *p, *s;
+ long len;
+ unsigned long bytelen;
+
+ p = r->buf+r->pos;
+ s = seekNewline(p);
+ if (s != NULL) {
+ p = r->buf+r->pos;
+ bytelen = s-(r->buf+r->pos)+2; /* include \r\n */
+ len = strtol(p,NULL,10);
+
+ if (len < 0) {
+ /* The nil object can always be created. */
+ obj = r->fn ? r->fn->createNil(cur) :
+ (void*)REDIS_REPLY_NIL;
+ } else {
+ /* Only continue when the buffer contains the entire bulk item. */
+ bytelen += len+2; /* include \r\n */
+ if (r->pos+bytelen <= sdslen(r->buf)) {
+ obj = r->fn ? r->fn->createString(cur,s+2,len) :
+ (void*)REDIS_REPLY_STRING;
+ }
+ }
+
+ /* Proceed when obj was created. */
+ if (obj != NULL) {
+ r->pos += bytelen;
+ if (r->reply == NULL)
+ r->reply = obj;
+ moveToNextTask(r);
+ return 0;
+ }
+ }
+ return -1;
+}
+
+static int processMultiBulkItem(redisReader *r) {
+ redisReadTask *cur = &(r->rstack[r->ridx]);
+ void *obj;
+ char *p;
+ long elements;
+
+ if ((p = readLine(r,NULL)) != NULL) {
+ elements = strtol(p,NULL,10);
+ if (elements == -1) {
+ obj = r->fn ? r->fn->createNil(cur) :
+ (void*)REDIS_REPLY_NIL;
+ moveToNextTask(r);
+ } else {
+ obj = r->fn ? r->fn->createArray(cur,elements) :
+ (void*)REDIS_REPLY_ARRAY;
+
+ /* Modify task stack when there are more than 0 elements. */
+ if (elements > 0) {
+ cur->elements = elements;
+ r->ridx++;
+ r->rstack[r->ridx].type = -1;
+ r->rstack[r->ridx].elements = -1;
+ r->rstack[r->ridx].parent = obj;
+ r->rstack[r->ridx].idx = 0;
+ } else {
+ moveToNextTask(r);
+ }
+ }
+
+ /* Object was created, so we can always continue. */
+ if (r->reply == NULL)
+ r->reply = obj;
+ return 0;
+ }
+ return -1;
+}
+
+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) {
+ if ((p = readBytes(r,1)) != NULL) {
+ switch (p[0]) {
+ case '-':
+ cur->type = REDIS_REPLY_ERROR;
+ break;
+ case '+':
+ cur->type = REDIS_REPLY_STATUS;
+ break;
+ case ':':
+ cur->type = REDIS_REPLY_INTEGER;
+ break;
+ case '$':
+ cur->type = REDIS_REPLY_STRING;
+ break;
+ case '*':
+ 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;
+ }
+ } else {
+ /* could not consume 1 byte */
+ return -1;
+ }
+ }
+
+ /* process typed item */
+ switch(cur->type) {
+ case REDIS_REPLY_ERROR:
+ case REDIS_REPLY_STATUS:
+ case REDIS_REPLY_INTEGER:
+ return processLineItem(r);
+ case REDIS_REPLY_STRING:
+ return processBulkItem(r);
+ case REDIS_REPLY_ARRAY:
+ return processMultiBulkItem(r);
+ default:
+ redisSetReplyReaderError(r,sdscatprintf(sdsempty(),
+ "unknown item type '%d'", cur->type));
+ return -1;
+ }
+}
+
+void *redisReplyReaderCreate() {
+ redisReader *r = calloc(sizeof(redisReader),1);
+ r->error = NULL;
+ r->fn = &defaultFunctions;
+ r->buf = sdsempty();
+ r->ridx = -1;
+ return 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;
+}
+
+/* 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;
+}
+
+void redisReplyReaderFree(void *reader) {
+ redisReader *r = reader;
+ if (r->error != NULL)
+ sdsfree(r->error);
+ if (r->reply != NULL && r->fn)
+ 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 = 0;
+ }
+ r->ridx = -1;
+ r->error = err;
+}
+
+char *redisReplyReaderGetError(void *reader) {
+ redisReader *r = reader;
+ return r->error;
+}
+
+void redisReplyReaderFeed(void *reader, char *buf, size_t len) {
+ redisReader *r = reader;
+
+ /* Copy the provided buffer. */
+ if (buf != NULL && len >= 1)
+ r->buf = sdscatlen(r->buf,buf,len);
+}
+
+int redisReplyReaderGetReply(void *reader, void **reply) {
+ redisReader *r = reader;
+ if (reply != NULL) *reply = NULL;
+
+ /* When the buffer is empty, there will never be a reply. */
+ if (sdslen(r->buf) == 0)
+ return REDIS_OK;
+
+ /* Set first item to process when the stack is empty. */
+ if (r->ridx == -1) {
+ r->rstack[0].type = -1;
+ r->rstack[0].elements = -1;
+ r->rstack[0].parent = NULL;
+ r->rstack[0].idx = -1;
+ r->ridx = 0;
+ }
+
+ /* Process items in reply. */
+ while (r->ridx >= 0)
+ if (processItem(r) < 0)
+ break;
+
+ /* Discard the consumed part of the buffer. */
+ if (r->pos > 0) {
+ if (r->pos == sdslen(r->buf)) {
+ /* sdsrange has a quirck on this edge case. */
+ sdsfree(r->buf);
+ r->buf = sdsempty();
+ } else {
+ r->buf = sdsrange(r->buf,r->pos,sdslen(r->buf));
+ }
+ r->pos = 0;
+ }
+
+ /* Emit a reply when there is one. */
+ if (r->ridx == -1) {
+ void *aux = r->reply;
+ r->reply = NULL;
+
+ /* Destroy the buffer when it is empty and is quite large. */
+ if (sdslen(r->buf) == 0 && sdsavail(r->buf) > 16*1024) {
+ sdsfree(r->buf);
+ r->buf = sdsempty();
+ r->pos = 0;
+ }
+
+ /* Check if there actually *is* a reply. */
+ if (r->error != NULL) {
+ return REDIS_ERR;
+ } else {
+ if (reply != NULL) *reply = aux;
+ }
+ }
+ return REDIS_OK;
+}
+
+/* Calculate the number of bytes needed to represent an integer as string. */
+static int intlen(int i) {
+ int len = 0;
+ if (i < 0) {
+ len++;
+ i = -i;
+ }
+ do {
+ len++;
+ i /= 10;
+ } while(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;
+}
+
+int redisvFormatCommand(char **target, const char *format, va_list ap) {
+ size_t size;
+ const char *arg, *c = format;
+ char *cmd = NULL; /* final command */
+ int pos; /* position in final command */
+ sds current; /* current argument */
+ char **argv = NULL;
+ int argc = 0, j;
+ int totlen = 0;
+
+ /* Abort if there is not target to set */
+ if (target == NULL)
+ return -1;
+
+ /* Build the command string accordingly to protocol */
+ current = sdsempty();
+ while(*c != '\0') {
+ if (*c != '%' || c[1] == '\0') {
+ if (*c == ' ') {
+ if (sdslen(current) != 0) {
+ addArgument(current, &argv, &argc, &totlen);
+ current = sdsempty();
+ }
+ } else {
+ current = sdscatlen(current,c,1);
+ }
+ } else {
+ switch(c[1]) {
+ case 's':
+ arg = va_arg(ap,char*);
+ current = sdscat(current,arg);
+ break;
+ case 'b':
+ arg = va_arg(ap,char*);
+ size = va_arg(ap,size_t);
+ current = sdscatlen(current,arg,size);
+ break;
+ case '%':
+ cmd = sdscat(cmd,"%");
+ break;
+ }
+ c++;
+ }
+ c++;
+ }
+
+ /* Add the last argument if needed */
+ if (sdslen(current) != 0) {
+ addArgument(current, &argv, &argc, &totlen);
+ } else {
+ sdsfree(current);
+ }
+
+ /* 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();
+ 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]);
+ cmd[pos++] = '\r';
+ cmd[pos++] = '\n';
+ }
+ assert(pos == totlen);
+ free(argv);
+ cmd[totlen] = '\0';
+ *target = cmd;
+ return totlen;
+}
+
+/* Format a command according to the Redis protocol. This function
+ * takes a format similar to printf:
+ *
+ * %s represents a C null terminated string you want to interpolate
+ * %b represents a binary safe string
+ *
+ * When using %b you need to provide both the pointer to the string
+ * and the length in bytes. Examples:
+ *
+ * len = redisFormatCommand(target, "GET %s", mykey);
+ * len = redisFormatCommand(target, "SET %s %b", mykey, myval, myvallen);
+ */
+int redisFormatCommand(char **target, const char *format, ...) {
+ va_list ap;
+ int len;
+ va_start(ap,format);
+ len = redisvFormatCommand(target,format,ap);
+ va_end(ap);
+ return len;
+}
+
+/* Format a command according to the Redis protocol. This function takes the
+ * number of arguments, an array with arguments and an array with their
+ * lengths. If the latter is set to NULL, strlen will be used to compute the
+ * argument lengths.
+ */
+int redisFormatCommandArgv(char **target, int argc, const char **argv, const size_t *argvlen) {
+ char *cmd = NULL; /* final command */
+ int pos; /* position in final command */
+ size_t len;
+ int totlen, j;
+
+ /* Calculate number of bytes needed for the command */
+ 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;
+ }
+
+ /* Build the command at protocol level */
+ cmd = malloc(totlen+1);
+ if (!cmd) redisOOM();
+ pos = sprintf(cmd,"*%d\r\n",argc);
+ for (j = 0; j < argc; j++) {
+ len = argvlen ? argvlen[j] : strlen(argv[j]);
+ pos += sprintf(cmd+pos,"$%zu\r\n",len);
+ memcpy(cmd+pos,argv[j],len);
+ pos += len;
+ cmd[pos++] = '\r';
+ cmd[pos++] = '\n';
+ }
+ assert(pos == totlen);
+ cmd[totlen] = '\0';
+ *target = cmd;
+ return totlen;
+}
+
+void __redisSetError(redisContext *c, int type, const sds errstr) {
+ c->err = type;
+ if (errstr != NULL) {
+ c->errstr = errstr;
+ } else {
+ /* Only REDIS_ERR_IO may lack a description! */
+ assert(type == REDIS_ERR_IO);
+ c->errstr = sdsnew(strerror(errno));
+ }
+}
+
+static redisContext *redisContextInit() {
+ redisContext *c = calloc(sizeof(redisContext),1);
+ c->err = 0;
+ c->errstr = NULL;
+ c->obuf = sdsempty();
+ c->fn = &defaultFunctions;
+ c->reader = NULL;
+ return c;
+}
+
+void redisFree(redisContext *c) {
+ /* Disconnect before free'ing if not yet disconnected. */
+ if (c->flags & REDIS_CONNECTED)
+ close(c->fd);
+ if (c->errstr != NULL)
+ sdsfree(c->errstr);
+ if (c->obuf != NULL)
+ sdsfree(c->obuf);
+ if (c->reader != NULL)
+ redisReplyReaderFree(c->reader);
+ free(c);
+}
+
+/* Connect to a Redis instance. On error the field error in the returned
+ * context will be set to the return value of the error function.
+ * When no set of reply functions is given, the default set will be used. */
+redisContext *redisConnect(const char *ip, int port) {
+ redisContext *c = redisContextInit();
+ c->flags |= REDIS_BLOCK;
+ c->flags |= REDIS_CONNECTED;
+ redisContextConnectTcp(c,ip,port);
+ return c;
+}
+
+redisContext *redisConnectNonBlock(const char *ip, int port) {
+ redisContext *c = redisContextInit();
+ c->flags &= ~REDIS_BLOCK;
+ c->flags |= REDIS_CONNECTED;
+ redisContextConnectTcp(c,ip,port);
+ return c;
+}
+
+redisContext *redisConnectUnix(const char *path) {
+ redisContext *c = redisContextInit();
+ c->flags |= REDIS_BLOCK;
+ c->flags |= REDIS_CONNECTED;
+ redisContextConnectUnix(c,path);
+ return c;
+}
+
+redisContext *redisConnectUnixNonBlock(const char *path) {
+ redisContext *c = redisContextInit();
+ c->flags &= ~REDIS_BLOCK;
+ c->flags |= REDIS_CONNECTED;
+ redisContextConnectUnix(c,path);
+ return c;
+}
+
+/* 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.
+ *
+ * After this function is called, you may use redisContextReadReply to
+ * see if there is a reply available. */
+int redisBufferRead(redisContext *c) {
+ char buf[2048];
+ int nread = read(c->fd,buf,sizeof(buf));
+ if (nread == -1) {
+ if (errno == EAGAIN) {
+ /* Try again later */
+ } else {
+ __redisSetError(c,REDIS_ERR_IO,NULL);
+ return REDIS_ERR;
+ }
+ } else if (nread == 0) {
+ __redisSetError(c,REDIS_ERR_EOF,
+ sdsnew("Server closed the connection"));
+ return REDIS_ERR;
+ } else {
+ __redisCreateReplyReader(c);
+ redisReplyReaderFeed(c->reader,buf,nread);
+ }
+ return REDIS_OK;
+}
+
+/* Write the output buffer to the socket.
+ *
+ * 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).
+ *
+ * Returns REDIS_ERR if an error occured trying to write and sets
+ * c->error to hold the appropriate error string.
+ */
+int redisBufferWrite(redisContext *c, int *done) {
+ int nwritten;
+ if (sdslen(c->obuf) > 0) {
+ nwritten = write(c->fd,c->obuf,sdslen(c->obuf));
+ if (nwritten == -1) {
+ if (errno == EAGAIN) {
+ /* Try again later */
+ } else {
+ __redisSetError(c,REDIS_ERR_IO,NULL);
+ return REDIS_ERR;
+ }
+ } else if (nwritten > 0) {
+ if (nwritten == (signed)sdslen(c->obuf)) {
+ sdsfree(c->obuf);
+ c->obuf = sdsempty();
+ } else {
+ c->obuf = sdsrange(c->obuf,nwritten,-1);
+ }
+ }
+ }
+ if (done != NULL) *done = (sdslen(c->obuf) == 0);
+ return REDIS_OK;
+}
+
+/* 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));
+ return REDIS_ERR;
+ }
+ return REDIS_OK;
+}
+
+int redisGetReply(redisContext *c, void **reply) {
+ int wdone = 0;
+ void *aux = NULL;
+
+ /* Try to read pending replies */
+ if (redisGetReplyFromReader(c,&aux) == REDIS_ERR)
+ return REDIS_ERR;
+
+ /* For the blocking context, flush output buffer and read reply */
+ if (aux == NULL && c->flags & REDIS_BLOCK) {
+ /* Write until done */
+ do {
+ if (redisBufferWrite(c,&wdone) == REDIS_ERR)
+ return REDIS_ERR;
+ } while (!wdone);
+
+ /* Read until there is a reply */
+ do {
+ if (redisBufferRead(c) == REDIS_ERR)
+ return REDIS_ERR;
+ if (redisGetReplyFromReader(c,&aux) == REDIS_ERR)
+ return REDIS_ERR;
+ } while (aux == NULL);
+ }
+
+ /* Set reply object */
+ if (reply != NULL) *reply = aux;
+ return REDIS_OK;
+}
+
+
+/* Helper function for the redisAppendCommand* family of functions.
+ *
+ * Write a formatted command to the output buffer. When this family
+ * 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);
+}
+
+void redisvAppendCommand(redisContext *c, const char *format, va_list ap) {
+ char *cmd;
+ int len;
+ len = redisvFormatCommand(&cmd,format,ap);
+ __redisAppendCommand(c,cmd,len);
+ free(cmd);
+}
+
+void redisAppendCommand(redisContext *c, const char *format, ...) {
+ va_list ap;
+ va_start(ap,format);
+ redisvAppendCommand(c,format,ap);
+ va_end(ap);
+}
+
+void 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);
+ free(cmd);
+}
+
+/* Helper function for the redisCommand* family of functions.
+ *
+ * Write a formatted command to the output buffer. If the given context is
+ * blocking, immediately read the reply into the "reply" pointer. When the
+ * context is non-blocking, the "reply" pointer will not be used and the
+ * command is simply appended to the write buffer.
+ *
+ * Returns the reply when a reply was succesfully retrieved. Returns NULL
+ * 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);
+
+ if (c->flags & REDIS_BLOCK) {
+ if (redisGetReply(c,&aux) == REDIS_OK)
+ return aux;
+ return NULL;
+ }
+ 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;
+}
+
+void *redisCommand(redisContext *c, const char *format, ...) {
+ va_list ap;
+ void *reply = NULL;
+ va_start(ap,format);
+ reply = redisvCommand(c,format,ap);
+ va_end(ap);
+ return reply;
+}
+
+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;
+}
--- /dev/null
+/*
+ * Copyright (c) 2009-2010, Salvatore Sanfilippo <antirez at gmail dot com>
+ * 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 __HIREDIS_H
+#define __HIREDIS_H
+#include <stdio.h> /* for size_t */
+#include <stdarg.h> /* for va_list */
+
+#define HIREDIS_MAJOR 0
+#define HIREDIS_MINOR 9
+#define HIREDIS_PATCH 0
+
+#define REDIS_ERR -1
+#define REDIS_OK 0
+
+/* When an error occurs, the err flag in a context is set to hold the type of
+ * 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 */
+
+/* Connection type can be blocking or non-blocking and is set in the
+ * least significant bit of the flags field in redisContext. */
+#define REDIS_BLOCK 0x1
+
+/* Connection may be disconnected before being free'd. The second bit
+ * in the flags field is set when the context is connected. */
+#define REDIS_CONNECTED 0x2
+
+/* The async API might try to disconnect cleanly and flush the output
+ * buffer and read all subsequent replies before disconnecting.
+ * This flag means no new commands can come in and the connection
+ * should be terminated once all replies have been read. */
+#define REDIS_DISCONNECTING 0x4
+
+#define REDIS_REPLY_ERROR 0
+#define REDIS_REPLY_STRING 1
+#define REDIS_REPLY_ARRAY 2
+#define REDIS_REPLY_INTEGER 3
+#define REDIS_REPLY_NIL 4
+#define REDIS_REPLY_STATUS 5
+
+/* This is the reply object returned by redisCommand() */
+typedef struct redisReply {
+ int type; /* REDIS_REPLY_* */
+ long long integer; /* The integer when type is REDIS_REPLY_INTEGER */
+ int len; /* Length of string */
+ char *str; /* Used for both REDIS_REPLY_ERROR and REDIS_REPLY_STRING */
+ size_t elements; /* number of elements, for REDIS_REPLY_ARRAY */
+ struct redisReply **element; /* elements vector for REDIS_REPLY_ARRAY */
+} redisReply;
+
+typedef struct redisReadTask {
+ int type;
+ int elements; /* number of elements in multibulk container */
+ void *parent; /* optional pointer to parent object */
+ int idx; /* index in parent (array) object */
+} redisReadTask;
+
+typedef struct redisReplyObjectFunctions {
+ void *(*createString)(const redisReadTask*, char*, size_t);
+ void *(*createArray)(const redisReadTask*, int);
+ void *(*createInteger)(const redisReadTask*, long long);
+ void *(*createNil)(const redisReadTask*);
+ 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 */
+ int err; /* Error flags, 0 when there is no error */
+ char *errstr; /* String representation of error when applicable */
+
+ /* Function set for reply buildup and reply reader */
+ redisReplyObjectFunctions *fn;
+ void *reader;
+} redisContext;
+
+void freeReplyObject(void *reply);
+void *redisReplyReaderCreate();
+int redisReplyReaderSetReplyObjectFunctions(void *reader, redisReplyObjectFunctions *fn);
+void *redisReplyReaderGetObject(void *reader);
+char *redisReplyReaderGetError(void *reader);
+void redisReplyReaderFree(void *ptr);
+void redisReplyReaderFeed(void *reader, 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);
+
+redisContext *redisConnect(const char *ip, int port);
+redisContext *redisConnectNonBlock(const char *ip, int port);
+redisContext *redisConnectUnix(const char *path);
+redisContext *redisConnectUnixNonBlock(const char *path);
+int redisSetReplyObjectFunctions(redisContext *c, redisReplyObjectFunctions *fn);
+void redisFree(redisContext *c);
+int redisBufferRead(redisContext *c);
+int redisBufferWrite(redisContext *c, int *done);
+
+/* In a blocking context, this function first checks if there are unconsumed
+ * replies to return and returns one if so. Otherwise, it flushes the output
+ * buffer to the socket and reads until it has a reply. In a non-blocking
+ * context, it will return unconsumed replies until there are no more. */
+int redisGetReply(redisContext *c, void **reply);
+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);
+
+/* Issue a command to Redis. In a blocking context, it is identical to calling
+ * redisAppendCommand, followed by redisGetReply. The function will return
+ * NULL if there was an error in performing the request, otherwise it will
+ * return the reply. In a non-blocking context, it is identical to calling
+ * only redisAppendCommand and will always return NULL. */
+void *redisvCommand(redisContext *c, const char *format, va_list ap);
+void *redisCommand(redisContext *c, const char *format, ...);
+void *redisCommandArgv(redisContext *c, int argc, const char **argv, const size_t *argvlen);
+
+#endif
--- /dev/null
+/* Extracted from anet.c to work properly with Hiredis error reporting.
+ *
+ * Copyright (c) 2006-2010, Salvatore Sanfilippo <antirez at gmail dot com>
+ * 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 "fmacros.h"
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#include <arpa/inet.h>
+#include <unistd.h>
+#include <fcntl.h>
+#include <string.h>
+#include <netdb.h>
+#include <errno.h>
+#include <stdarg.h>
+#include <stdio.h>
+
+#include "hiredis.h"
+#include "sds.h"
+
+/* Forward declaration */
+void __redisSetError(redisContext *c, int type, sds err);
+
+static int redisCreateSocket(redisContext *c, int type) {
+ int s, on = 1;
+ if ((s = socket(type, SOCK_STREAM, 0)) == -1) {
+ __redisSetError(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);
+ return REDIS_ERR;
+ }
+ }
+ return s;
+}
+
+static int redisSetNonBlock(redisContext *c, int fd) {
+ int flags;
+
+ /* Set the socket nonblocking.
+ * 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)));
+ close(fd);
+ return REDIS_ERR;
+ }
+ if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1) {
+ __redisSetError(c,REDIS_ERR_IO,
+ sdscatprintf(sdsempty(), "fcntl(F_SETFL,O_NONBLOCK): %s", strerror(errno)));
+ close(fd);
+ return REDIS_ERR;
+ }
+ return REDIS_OK;
+}
+
+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)));
+ return REDIS_ERR;
+ }
+ return REDIS_OK;
+}
+
+int redisContextConnectTcp(redisContext *c, const char *addr, int port) {
+ int s;
+ int blocking = (c->flags & REDIS_BLOCK);
+ struct sockaddr_in sa;
+
+ if ((s = redisCreateSocket(c,AF_INET)) == REDIS_ERR)
+ return REDIS_ERR;
+ if (!blocking && redisSetNonBlock(c,s) == REDIS_ERR)
+ return REDIS_ERR;
+
+ sa.sin_family = AF_INET;
+ sa.sin_port = htons(port);
+ if (inet_aton(addr, &sa.sin_addr) == 0) {
+ struct hostent *he;
+
+ 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 (connect(s, (struct sockaddr*)&sa, sizeof(sa)) == -1) {
+ if (errno == EINPROGRESS && !blocking) {
+ /* This is ok. */
+ } else {
+ __redisSetError(c,REDIS_ERR_IO,NULL);
+ close(s);
+ return REDIS_ERR;
+ }
+ }
+
+ if (redisSetTcpNoDelay(c,s) != REDIS_OK) {
+ close(s);
+ return REDIS_ERR;
+ }
+
+ c->fd = s;
+ return REDIS_OK;
+}
+
+int redisContextConnectUnix(redisContext *c, const char *path) {
+ int s;
+ int blocking = (c->flags & REDIS_BLOCK);
+ struct sockaddr_un sa;
+
+ if ((s = redisCreateSocket(c,AF_LOCAL)) == REDIS_ERR)
+ return REDIS_ERR;
+ if (!blocking && redisSetNonBlock(c,s) != REDIS_OK)
+ return REDIS_ERR;
+
+ sa.sun_family = AF_LOCAL;
+ strncpy(sa.sun_path,path,sizeof(sa.sun_path)-1);
+ if (connect(s, (struct sockaddr*)&sa, sizeof(sa)) == -1) {
+ if (errno == EINPROGRESS && !blocking) {
+ /* This is ok. */
+ } else {
+ __redisSetError(c,REDIS_ERR_IO,NULL);
+ close(s);
+ return REDIS_ERR;
+ }
+ }
+
+ c->fd = s;
+ return REDIS_OK;
+}
--- /dev/null
+/* Extracted from anet.c to work properly with Hiredis error reporting.
+ *
+ * Copyright (c) 2006-2010, Salvatore Sanfilippo <antirez at gmail dot com>
+ * 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 __NET_H
+#define __NET_H
+
+int redisContextConnectTcp(redisContext *c, const char *addr, int port);
+int redisContextConnectUnix(redisContext *c, const char *path);
+
+#endif
--- /dev/null
+/* SDSLib, A C dynamic strings library
+ *
+ * Copyright (c) 2006-2010, Salvatore Sanfilippo <antirez at gmail dot com>
+ * 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.
+ */
+
+#define SDS_ABORT_ON_OOM
+
+#include "sds.h"
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <ctype.h>
+
+static void sdsOomAbort(void) {
+ fprintf(stderr,"SDS: Out Of Memory (SDS_ABORT_ON_OOM defined)\n");
+ abort();
+}
+
+sds sdsnewlen(const void *init, size_t initlen) {
+ struct sdshdr *sh;
+
+ sh = malloc(sizeof(struct sdshdr)+initlen+1);
+#ifdef SDS_ABORT_ON_OOM
+ if (sh == NULL) sdsOomAbort();
+#else
+ if (sh == NULL) return NULL;
+#endif
+ sh->len = initlen;
+ sh->free = 0;
+ if (initlen) {
+ if (init) memcpy(sh->buf, init, initlen);
+ else memset(sh->buf,0,initlen);
+ }
+ sh->buf[initlen] = '\0';
+ return (char*)sh->buf;
+}
+
+sds sdsempty(void) {
+ return sdsnewlen("",0);
+}
+
+sds sdsnew(const char *init) {
+ size_t initlen = (init == NULL) ? 0 : strlen(init);
+ return sdsnewlen(init, initlen);
+}
+
+size_t sdslen(const sds s) {
+ struct sdshdr *sh = (void*) (s-(sizeof(struct sdshdr)));
+ return sh->len;
+}
+
+sds sdsdup(const sds s) {
+ return sdsnewlen(s, sdslen(s));
+}
+
+void sdsfree(sds s) {
+ if (s == NULL) return;
+ free(s-sizeof(struct sdshdr));
+}
+
+size_t sdsavail(sds s) {
+ struct sdshdr *sh = (void*) (s-(sizeof(struct sdshdr)));
+ return sh->free;
+}
+
+void sdsupdatelen(sds s) {
+ struct sdshdr *sh = (void*) (s-(sizeof(struct sdshdr)));
+ int reallen = strlen(s);
+ sh->free += (sh->len-reallen);
+ sh->len = reallen;
+}
+
+static sds sdsMakeRoomFor(sds s, size_t addlen) {
+ struct sdshdr *sh, *newsh;
+ size_t free = sdsavail(s);
+ size_t len, newlen;
+
+ if (free >= addlen) return s;
+ len = sdslen(s);
+ sh = (void*) (s-(sizeof(struct sdshdr)));
+ newlen = (len+addlen)*2;
+ newsh = realloc(sh, sizeof(struct sdshdr)+newlen+1);
+#ifdef SDS_ABORT_ON_OOM
+ if (newsh == NULL) sdsOomAbort();
+#else
+ if (newsh == NULL) return NULL;
+#endif
+
+ newsh->free = newlen - len;
+ return newsh->buf;
+}
+
+sds sdscatlen(sds s, const void *t, size_t len) {
+ struct sdshdr *sh;
+ size_t curlen = sdslen(s);
+
+ s = sdsMakeRoomFor(s,len);
+ if (s == NULL) return NULL;
+ sh = (void*) (s-(sizeof(struct sdshdr)));
+ memcpy(s+curlen, t, len);
+ sh->len = curlen+len;
+ sh->free = sh->free-len;
+ s[curlen+len] = '\0';
+ return s;
+}
+
+sds sdscat(sds s, const char *t) {
+ return sdscatlen(s, t, strlen(t));
+}
+
+sds sdscpylen(sds s, char *t, size_t len) {
+ struct sdshdr *sh = (void*) (s-(sizeof(struct sdshdr)));
+ size_t totlen = sh->free+sh->len;
+
+ if (totlen < len) {
+ s = sdsMakeRoomFor(s,len-sh->len);
+ if (s == NULL) return NULL;
+ sh = (void*) (s-(sizeof(struct sdshdr)));
+ totlen = sh->free+sh->len;
+ }
+ memcpy(s, t, len);
+ s[len] = '\0';
+ sh->len = len;
+ sh->free = totlen-len;
+ return s;
+}
+
+sds sdscpy(sds s, char *t) {
+ return sdscpylen(s, t, strlen(t));
+}
+
+sds sdscatvprintf(sds s, const char *fmt, va_list ap) {
+ va_list cpy;
+ char *buf, *t;
+ size_t buflen = 16;
+
+ while(1) {
+ buf = malloc(buflen);
+#ifdef SDS_ABORT_ON_OOM
+ if (buf == NULL) sdsOomAbort();
+#else
+ if (buf == NULL) return NULL;
+#endif
+ buf[buflen-2] = '\0';
+ va_copy(cpy,ap);
+ vsnprintf(buf, buflen, fmt, cpy);
+ if (buf[buflen-2] != '\0') {
+ free(buf);
+ buflen *= 2;
+ continue;
+ }
+ break;
+ }
+ t = sdscat(s, buf);
+ free(buf);
+ return t;
+}
+
+sds sdscatprintf(sds s, const char *fmt, ...) {
+ va_list ap;
+ char *t;
+ va_start(ap, fmt);
+ t = sdscatvprintf(s,fmt,ap);
+ va_end(ap);
+ return t;
+}
+
+sds sdstrim(sds s, const char *cset) {
+ struct sdshdr *sh = (void*) (s-(sizeof(struct sdshdr)));
+ char *start, *end, *sp, *ep;
+ size_t len;
+
+ sp = start = s;
+ ep = end = s+sdslen(s)-1;
+ while(sp <= end && strchr(cset, *sp)) sp++;
+ while(ep > start && strchr(cset, *ep)) ep--;
+ len = (sp > ep) ? 0 : ((ep-sp)+1);
+ if (sh->buf != sp) memmove(sh->buf, sp, len);
+ sh->buf[len] = '\0';
+ sh->free = sh->free+(sh->len-len);
+ sh->len = len;
+ return s;
+}
+
+sds sdsrange(sds s, int start, int end) {
+ struct sdshdr *sh = (void*) (s-(sizeof(struct sdshdr)));
+ size_t newlen, len = sdslen(s);
+
+ if (len == 0) return s;
+ if (start < 0) {
+ start = len+start;
+ if (start < 0) start = 0;
+ }
+ if (end < 0) {
+ end = len+end;
+ if (end < 0) end = 0;
+ }
+ newlen = (start > end) ? 0 : (end-start)+1;
+ if (newlen != 0) {
+ if (start >= (signed)len) start = len-1;
+ if (end >= (signed)len) end = len-1;
+ newlen = (start > end) ? 0 : (end-start)+1;
+ } else {
+ start = 0;
+ }
+ if (start != 0) memmove(sh->buf, sh->buf+start, newlen);
+ sh->buf[newlen] = 0;
+ sh->free = sh->free+(sh->len-newlen);
+ sh->len = newlen;
+ return s;
+}
+
+void sdstolower(sds s) {
+ int len = sdslen(s), j;
+
+ for (j = 0; j < len; j++) s[j] = tolower(s[j]);
+}
+
+void sdstoupper(sds s) {
+ int len = sdslen(s), j;
+
+ for (j = 0; j < len; j++) s[j] = toupper(s[j]);
+}
+
+int sdscmp(sds s1, sds s2) {
+ size_t l1, l2, minlen;
+ int cmp;
+
+ l1 = sdslen(s1);
+ l2 = sdslen(s2);
+ minlen = (l1 < l2) ? l1 : l2;
+ cmp = memcmp(s1,s2,minlen);
+ if (cmp == 0) return l1-l2;
+ return cmp;
+}
+
+/* Split 's' with separator in 'sep'. An array
+ * of sds strings is returned. *count will be set
+ * by reference to the number of tokens returned.
+ *
+ * On out of memory, zero length string, zero length
+ * separator, NULL is returned.
+ *
+ * Note that 'sep' is able to split a string using
+ * a multi-character separator. For example
+ * sdssplit("foo_-_bar","_-_"); will return two
+ * elements "foo" and "bar".
+ *
+ * This version of the function is binary-safe but
+ * requires length arguments. sdssplit() is just the
+ * same function but for zero-terminated strings.
+ */
+sds *sdssplitlen(char *s, int len, char *sep, int seplen, int *count) {
+ int elements = 0, slots = 5, start = 0, j;
+
+ sds *tokens = malloc(sizeof(sds)*slots);
+#ifdef SDS_ABORT_ON_OOM
+ if (tokens == NULL) sdsOomAbort();
+#endif
+ if (seplen < 1 || len < 0 || tokens == NULL) return NULL;
+ if (len == 0) {
+ *count = 0;
+ return tokens;
+ }
+ for (j = 0; j < (len-(seplen-1)); j++) {
+ /* make sure there is room for the next element and the final one */
+ if (slots < elements+2) {
+ sds *newtokens;
+
+ slots *= 2;
+ newtokens = realloc(tokens,sizeof(sds)*slots);
+ if (newtokens == NULL) {
+#ifdef SDS_ABORT_ON_OOM
+ sdsOomAbort();
+#else
+ goto cleanup;
+#endif
+ }
+ tokens = newtokens;
+ }
+ /* search the separator */
+ if ((seplen == 1 && *(s+j) == sep[0]) || (memcmp(s+j,sep,seplen) == 0)) {
+ tokens[elements] = sdsnewlen(s+start,j-start);
+ if (tokens[elements] == NULL) {
+#ifdef SDS_ABORT_ON_OOM
+ sdsOomAbort();
+#else
+ goto cleanup;
+#endif
+ }
+ elements++;
+ start = j+seplen;
+ j = j+seplen-1; /* skip the separator */
+ }
+ }
+ /* Add the final element. We are sure there is room in the tokens array. */
+ tokens[elements] = sdsnewlen(s+start,len-start);
+ if (tokens[elements] == NULL) {
+#ifdef SDS_ABORT_ON_OOM
+ sdsOomAbort();
+#else
+ goto cleanup;
+#endif
+ }
+ elements++;
+ *count = elements;
+ return tokens;
+
+#ifndef SDS_ABORT_ON_OOM
+cleanup:
+ {
+ int i;
+ for (i = 0; i < elements; i++) sdsfree(tokens[i]);
+ free(tokens);
+ return NULL;
+ }
+#endif
+}
+
+void sdsfreesplitres(sds *tokens, int count) {
+ if (!tokens) return;
+ while(count--)
+ sdsfree(tokens[count]);
+ free(tokens);
+}
+
+sds sdsfromlonglong(long long value) {
+ char buf[32], *p;
+ unsigned long long v;
+
+ v = (value < 0) ? -value : value;
+ p = buf+31; /* point to the last character */
+ do {
+ *p-- = '0'+(v%10);
+ v /= 10;
+ } while(v);
+ if (value < 0) *p-- = '-';
+ p++;
+ return sdsnewlen(p,32-(p-buf));
+}
+
+sds sdscatrepr(sds s, char *p, size_t len) {
+ s = sdscatlen(s,"\"",1);
+ 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;
+ default:
+ if (isprint(*p))
+ s = sdscatprintf(s,"%c",*p);
+ else
+ s = sdscatprintf(s,"\\x%02x",(unsigned char)*p);
+ break;
+ }
+ p++;
+ }
+ return sdscatlen(s,"\"",1);
+}
+
+/* Split a line into arguments, where every argument can be in the
+ * following programming-language REPL-alike form:
+ *
+ * foo bar "newline are supported\n" and "\xff\x00otherstuff"
+ *
+ * The number of arguments is stored into *argc, and an array
+ * of sds is returned. The caller should sdsfree() all the returned
+ * strings and finally free() the array itself.
+ *
+ * Note that sdscatrepr() is able to convert back a string into
+ * a quoted string in the same format sdssplitargs() is able to parse.
+ */
+sds *sdssplitargs(char *line, int *argc) {
+ char *p = line;
+ char *current = NULL;
+ char **vector = NULL;
+
+ *argc = 0;
+ while(1) {
+ /* skip blanks */
+ while(*p && isspace(*p)) p++;
+ if (*p) {
+ /* get a token */
+ int inq=0; /* set to 1 if we are in "quotes" */
+ int done=0;
+
+ if (current == NULL) current = sdsempty();
+ while(!done) {
+ if (inq) {
+ if (*p == '\\' && *(p+1)) {
+ char c;
+
+ p++;
+ switch(*p) {
+ case 'n': c = '\n'; break;
+ case 'r': c = '\r'; break;
+ case 't': c = '\t'; break;
+ case 'b': c = '\b'; break;
+ case 'a': c = '\a'; break;
+ default: c = *p; break;
+ }
+ current = sdscatlen(current,&c,1);
+ } else if (*p == '"') {
+ /* closing quote must be followed by a space */
+ if (*(p+1) && !isspace(*(p+1))) goto err;
+ done=1;
+ } else if (!*p) {
+ /* unterminated quotes */
+ goto err;
+ } else {
+ current = sdscatlen(current,p,1);
+ }
+ } else {
+ switch(*p) {
+ case ' ':
+ case '\n':
+ case '\r':
+ case '\t':
+ case '\0':
+ done=1;
+ break;
+ case '"':
+ inq=1;
+ break;
+ default:
+ current = sdscatlen(current,p,1);
+ break;
+ }
+ }
+ if (*p) p++;
+ }
+ /* add the token to the vector */
+ vector = realloc(vector,((*argc)+1)*sizeof(char*));
+ vector[*argc] = current;
+ (*argc)++;
+ current = NULL;
+ } else {
+ return vector;
+ }
+ }
+
+err:
+ while((*argc)--)
+ sdsfree(vector[*argc]);
+ free(vector);
+ if (current) sdsfree(current);
+ return NULL;
+}
--- /dev/null
+/* SDSLib, A C dynamic strings library
+ *
+ * Copyright (c) 2006-2010, Salvatore Sanfilippo <antirez at gmail dot com>
+ * 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 __SDS_H
+#define __SDS_H
+
+#include <sys/types.h>
+#include <stdarg.h>
+
+typedef char *sds;
+
+struct sdshdr {
+ int len;
+ int free;
+ char buf[];
+};
+
+sds sdsnewlen(const void *init, size_t initlen);
+sds sdsnew(const char *init);
+sds sdsempty();
+size_t sdslen(const sds s);
+sds sdsdup(const sds s);
+void sdsfree(sds s);
+size_t sdsavail(sds s);
+sds sdscatlen(sds s, const void *t, size_t len);
+sds sdscat(sds s, const char *t);
+sds sdscpylen(sds s, char *t, size_t len);
+sds sdscpy(sds s, char *t);
+
+sds sdscatvprintf(sds s, const char *fmt, va_list ap);
+#ifdef __GNUC__
+sds sdscatprintf(sds s, const char *fmt, ...)
+ __attribute__((format(printf, 2, 3)));
+#else
+sds sdscatprintf(sds s, const char *fmt, ...);
+#endif
+
+sds sdstrim(sds s, const char *cset);
+sds sdsrange(sds s, int start, int end);
+void sdsupdatelen(sds s);
+int sdscmp(sds s1, sds s2);
+sds *sdssplitlen(char *s, int len, char *sep, int seplen, int *count);
+void sdsfreesplitres(sds *tokens, int count);
+void sdstolower(sds s);
+void sdstoupper(sds s);
+sds sdsfromlonglong(long long value);
+sds sdscatrepr(sds s, char *p, size_t len);
+sds *sdssplitargs(char *line, int *argc);
+
+#endif
--- /dev/null
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <strings.h>
+#include <sys/time.h>
+#include <assert.h>
+#include <unistd.h>
+#include <signal.h>
+
+#include "hiredis.h"
+
+/* 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++;}
+
+static long long usec(void) {
+ struct timeval tv;
+ gettimeofday(&tv,NULL);
+ 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);
+ exit(1);
+ }
+}
+
+static void test_format_commands() {
+ char *cmd;
+ int len;
+
+ test("Format command without interpolation: ");
+ len = redisFormatCommand(&cmd,"SET foo bar");
+ test_cond(strncmp(cmd,"*3\r\n$3\r\nSET\r\n$3\r\nfoo\r\n$3\r\nbar\r\n",len) == 0 &&
+ len == 4+4+(3+2)+4+(3+2)+4+(3+2));
+ free(cmd);
+
+ test("Format command with %%s string interpolation: ");
+ len = redisFormatCommand(&cmd,"SET %s %s","foo","bar");
+ test_cond(strncmp(cmd,"*3\r\n$3\r\nSET\r\n$3\r\nfoo\r\n$3\r\nbar\r\n",len) == 0 &&
+ len == 4+4+(3+2)+4+(3+2)+4+(3+2));
+ free(cmd);
+
+ test("Format command with %%b string interpolation: ");
+ len = redisFormatCommand(&cmd,"SET %b %b","foo",3,"b\0r",3);
+ test_cond(strncmp(cmd,"*3\r\n$3\r\nSET\r\n$3\r\nfoo\r\n$3\r\nb\0r\r\n",len) == 0 &&
+ len == 4+4+(3+2)+4+(3+2)+4+(3+2));
+ free(cmd);
+
+ const char *argv[3];
+ argv[0] = "SET";
+ argv[1] = "foo";
+ argv[2] = "bar";
+ size_t lens[3] = { 3, 3, 3 };
+ int argc = 3;
+
+ test("Format command by passing argc/argv without lengths: ");
+ len = redisFormatCommandArgv(&cmd,argc,argv,NULL);
+ test_cond(strncmp(cmd,"*3\r\n$3\r\nSET\r\n$3\r\nfoo\r\n$3\r\nbar\r\n",len) == 0 &&
+ len == 4+4+(3+2)+4+(3+2)+4+(3+2));
+ free(cmd);
+
+ test("Format command by passing argc/argv with lengths: ");
+ len = redisFormatCommandArgv(&cmd,argc,argv,lens);
+ test_cond(strncmp(cmd,"*3\r\n$3\r\nSET\r\n$3\r\nfoo\r\n$3\r\nbar\r\n",len) == 0 &&
+ len == 4+4+(3+2)+4+(3+2)+4+(3+2));
+ free(cmd);
+}
+
+static void test_blocking_connection() {
+ redisContext *c;
+ redisReply *reply;
+
+ __connect(&c);
+ test("Returns I/O error when the connection is lost: ");
+ reply = redisCommand(c,"QUIT");
+ test_cond(strcasecmp(reply->str,"OK") == 0 && redisCommand(c,"PING") == NULL);
+
+ /* Two conditions may happen, depending on the type of connection.
+ * When connected via TCP, the socket will not yet be aware of the closed
+ * connection and the write(2) call will succeed, but the read(2) will
+ * result in an EOF. When connected via Unix sockets, the socket will be
+ * immediately aware that it was closed and fail on the write(2) call. */
+ if (use_unix) {
+ fprintf(stderr,"Error: %s\n", c->errstr);
+ assert(c->err == REDIS_ERR_IO &&
+ strcmp(c->errstr,"Broken pipe") == 0);
+ } else {
+ fprintf(stderr,"Error: %s\n", c->errstr);
+ assert(c->err == REDIS_ERR_EOF &&
+ strcmp(c->errstr,"Server closed the connection") == 0);
+ }
+ freeReplyObject(reply);
+ redisFree(c);
+
+ __connect(&c); /* reconnect */
+ 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("Sorry DB 9 is not empty, test can not continue\n");
+ exit(1);
+ } else {
+ printf("DB 9 is empty... test can continue\n");
+ }
+ freeReplyObject(reply);
+
+ test("Is a able to send commands verbatim: ");
+ reply = redisCommand(c,"SET foo bar");
+ test_cond (reply->type == REDIS_REPLY_STATUS &&
+ strcasecmp(reply->str,"ok") == 0)
+ freeReplyObject(reply);
+
+ test("%%s String interpolation works: ");
+ reply = redisCommand(c,"SET %s %s","foo","hello world");
+ freeReplyObject(reply);
+ reply = redisCommand(c,"GET foo");
+ test_cond(reply->type == REDIS_REPLY_STRING &&
+ strcmp(reply->str,"hello world") == 0);
+ freeReplyObject(reply);
+
+ test("%%b String interpolation works: ");
+ reply = redisCommand(c,"SET %b %b","foo",3,"hello\x00world",11);
+ freeReplyObject(reply);
+ reply = redisCommand(c,"GET foo");
+ test_cond(reply->type == REDIS_REPLY_STRING &&
+ memcmp(reply->str,"hello\x00world",11) == 0)
+
+ test("Binary reply length is correct: ");
+ test_cond(reply->len == 11)
+ freeReplyObject(reply);
+
+ test("Can parse nil replies: ");
+ reply = redisCommand(c,"GET nokey");
+ test_cond(reply->type == REDIS_REPLY_NIL)
+ freeReplyObject(reply);
+
+ /* test 7 */
+ test("Can parse integer replies: ");
+ reply = redisCommand(c,"INCR mycounter");
+ test_cond(reply->type == REDIS_REPLY_INTEGER && reply->integer == 1)
+ freeReplyObject(reply);
+
+ test("Can parse multi bulk replies: ");
+ freeReplyObject(redisCommand(c,"LPUSH mylist foo"));
+ freeReplyObject(redisCommand(c,"LPUSH mylist bar"));
+ reply = redisCommand(c,"LRANGE mylist 0 -1");
+ test_cond(reply->type == REDIS_REPLY_ARRAY &&
+ reply->elements == 2 &&
+ !memcmp(reply->element[0]->str,"bar",3) &&
+ !memcmp(reply->element[1]->str,"foo",3))
+ freeReplyObject(reply);
+
+ /* m/e with multi bulk reply *before* other reply.
+ * specifically test ordering of reply items to parse. */
+ test("Can handle nested multi bulk replies: ");
+ freeReplyObject(redisCommand(c,"MULTI"));
+ freeReplyObject(redisCommand(c,"LRANGE mylist 0 -1"));
+ freeReplyObject(redisCommand(c,"PING"));
+ reply = (redisCommand(c,"EXEC"));
+ test_cond(reply->type == REDIS_REPLY_ARRAY &&
+ reply->elements == 2 &&
+ reply->element[0]->type == REDIS_REPLY_ARRAY &&
+ reply->element[0]->elements == 2 &&
+ !memcmp(reply->element[0]->element[0]->str,"bar",3) &&
+ !memcmp(reply->element[0]->element[1]->str,"foo",3) &&
+ reply->element[1]->type == REDIS_REPLY_STATUS &&
+ strcasecmp(reply->element[1]->str,"pong") == 0);
+ freeReplyObject(reply);
+}
+
+static void test_reply_reader() {
+ 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("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);
+}
+
+static void test_throughput() {
+ int i;
+ long long t1, t2;
+ redisContext *c = blocking_context;
+ redisReply **replies;
+
+ test("Throughput:\n");
+ for (i = 0; i < 500; i++)
+ freeReplyObject(redisCommand(c,"LPUSH mylist foo"));
+
+ replies = malloc(sizeof(redisReply*)*1000);
+ t1 = usec();
+ for (i = 0; i < 1000; i++) {
+ replies[i] = redisCommand(c,"PING");
+ assert(replies[i] != NULL && replies[i]->type == REDIS_REPLY_STATUS);
+ }
+ t2 = usec();
+ for (i = 0; i < 1000; i++) freeReplyObject(replies[i]);
+ free(replies);
+ printf("\t(1000x PING: %.2fs)\n", (t2-t1)/1000000.0);
+
+ replies = malloc(sizeof(redisReply*)*1000);
+ t1 = usec();
+ for (i = 0; i < 1000; i++) {
+ replies[i] = redisCommand(c,"LRANGE mylist 0 499");
+ assert(replies[i] != NULL && replies[i]->type == REDIS_REPLY_ARRAY);
+ assert(replies[i] != NULL && replies[i]->elements == 500);
+ }
+ t2 = usec();
+ for (i = 0; i < 1000; i++) freeReplyObject(replies[i]);
+ free(replies);
+ printf("\t(1000x LRANGE with 500 elements: %.2fs)\n", (t2-t1)/1000000.0);
+}
+
+static void cleanup() {
+ 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);
+}
+
+// static long __test_callback_flags = 0;
+// static void __test_callback(redisContext *c, void *privdata) {
+// ((void)c);
+// /* Shift to detect execution order */
+// __test_callback_flags <<= 8;
+// __test_callback_flags |= (long)privdata;
+// }
+//
+// static void __test_reply_callback(redisContext *c, redisReply *reply, void *privdata) {
+// ((void)c);
+// /* Shift to detect execution order */
+// __test_callback_flags <<= 8;
+// __test_callback_flags |= (long)privdata;
+// if (reply) freeReplyObject(reply);
+// }
+//
+// static redisContext *__connect_nonblock() {
+// /* Reset callback flags */
+// __test_callback_flags = 0;
+// return redisConnectNonBlock("127.0.0.1", 6379, NULL);
+// }
+//
+// static void test_nonblocking_connection() {
+// redisContext *c;
+// int wdone = 0;
+//
+// test("Calls command callback when command is issued: ");
+// c = __connect_nonblock();
+// redisSetCommandCallback(c,__test_callback,(void*)1);
+// redisCommand(c,"PING");
+// test_cond(__test_callback_flags == 1);
+// redisFree(c);
+//
+// test("Calls disconnect callback on redisDisconnect: ");
+// c = __connect_nonblock();
+// redisSetDisconnectCallback(c,__test_callback,(void*)2);
+// redisDisconnect(c);
+// test_cond(__test_callback_flags == 2);
+// redisFree(c);
+//
+// test("Calls disconnect callback and free callback on redisFree: ");
+// c = __connect_nonblock();
+// redisSetDisconnectCallback(c,__test_callback,(void*)2);
+// redisSetFreeCallback(c,__test_callback,(void*)4);
+// redisFree(c);
+// test_cond(__test_callback_flags == ((2 << 8) | 4));
+//
+// test("redisBufferWrite against empty write buffer: ");
+// c = __connect_nonblock();
+// test_cond(redisBufferWrite(c,&wdone) == REDIS_OK && wdone == 1);
+// redisFree(c);
+//
+// test("redisBufferWrite against not yet connected fd: ");
+// c = __connect_nonblock();
+// redisCommand(c,"PING");
+// test_cond(redisBufferWrite(c,NULL) == REDIS_ERR &&
+// strncmp(c->error,"write:",6) == 0);
+// redisFree(c);
+//
+// test("redisBufferWrite against closed fd: ");
+// c = __connect_nonblock();
+// redisCommand(c,"PING");
+// redisDisconnect(c);
+// test_cond(redisBufferWrite(c,NULL) == REDIS_ERR &&
+// strncmp(c->error,"write:",6) == 0);
+// redisFree(c);
+//
+// test("Process callbacks in the right sequence: ");
+// c = __connect_nonblock();
+// redisCommandWithCallback(c,__test_reply_callback,(void*)1,"PING");
+// redisCommandWithCallback(c,__test_reply_callback,(void*)2,"PING");
+// redisCommandWithCallback(c,__test_reply_callback,(void*)3,"PING");
+//
+// /* Write output buffer */
+// wdone = 0;
+// while(!wdone) {
+// usleep(500);
+// redisBufferWrite(c,&wdone);
+// }
+//
+// /* Read until at least one callback is executed (the 3 replies will
+// * arrive in a single packet, causing all callbacks to be executed in
+// * a single pass). */
+// while(__test_callback_flags == 0) {
+// assert(redisBufferRead(c) == REDIS_OK);
+// redisProcessCallbacks(c);
+// }
+// test_cond(__test_callback_flags == 0x010203);
+// redisFree(c);
+//
+// test("redisDisconnect executes pending callbacks with NULL reply: ");
+// c = __connect_nonblock();
+// redisSetDisconnectCallback(c,__test_callback,(void*)1);
+// redisCommandWithCallback(c,__test_reply_callback,(void*)2,"PING");
+// redisDisconnect(c);
+// test_cond(__test_callback_flags == 0x0201);
+// redisFree(c);
+// }
+
+int main(int argc, char **argv) {
+ if (argc > 1) {
+ if (strcmp(argv[1],"-s") == 0)
+ use_unix = 1;
+ }
+
+ signal(SIGPIPE, SIG_IGN);
+ test_format_commands();
+ test_blocking_connection();
+ test_reply_reader();
+ // test_nonblocking_connection();
+ test_throughput();
+ cleanup();
+
+ if (fails == 0) {
+ printf("ALL TESTS PASSED\n");
+ } else {
+ printf("*** %d TESTS FAILED ***\n", fails);
+ }
+ return 0;
+}
--- /dev/null
+/*
+ * Copyright (c) 2009-2010, Salvatore Sanfilippo <antirez at gmail dot com>
+ * 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 <stdlib.h>
+
+/* Abort on out of memory */
+static void redisOOM(void) {
+ fprintf(stderr,"Out of memory in hiredis");
+ exit(1);
+}
+
+#endif
--- /dev/null
+linenoise_example*
--- /dev/null
+linenoise_example: linenoise.h linenoise.c
+
+linenoise_example: linenoise.o example.o
+ $(CC) $(ARCH) -Wall -W -Os -g -o linenoise_example linenoise.o example.o
+
+.c.o:
+ $(CC) $(ARCH) -c -Wall -W -Os -g $<
+
+clean:
+ rm -f linenoise_example *.o
--- /dev/null
+# Linenoise
+
+A minimal, zero-config, BSD licensed, readline replacement.
+
+News: linenoise is now part of [Android](http://android.git.kernel.org/?p=platform/system/core.git;a=tree;f=liblinenoise;h=56450eaed7f783760e5e6a5993ef75cde2e29dea;hb=HEAD Android)!
+
+## Can a line editing library be 20k lines of code?
+
+Line editing with some support for history is a really important feature for command line utilities. Instead of retyping almost the same stuff again and again it's just much better to hit the up arrow and edit on syntax errors, or in order to try a slightly different command. But apparently code dealing with terminals is some sort of Black Magic: readline is 30k lines of code, libedit 20k. Is it reasonable to link small utilities to huge libraries just to get a minimal support for line editing?
+
+So what usually happens is either:
+
+ * Large programs with configure scripts disabling line editing if readline is not present in the system, or not supporting it at all since readline is GPL licensed and libedit (the BSD clone) is not as known and available as readline is (Readl world example of this problem: Tclsh).
+ * Smaller programs not using a configure script not supporting line editing at all (A problem we had with Redis-cli for instance).
+
+The result is a pollution of binaries without line editing support.
+
+So I spent more or less two hours doing a reality check resulting in this little library: is it *really* needed for a line editing library to be 20k lines of code? Apparently not, it is possibe to get a very small, zero configuration, trivial to embed library, that solves the problem. Smaller programs will just include this, supporing line editing out of the box. Larger programs may use this little library or just checking with configure if readline/libedit is available and resorting to linenoise if not.
+
+## Terminals, in 2010.
+
+Apparently almost every terminal you can happen to use today has some kind of support for VT100 alike escape sequences. So I tried to write a lib using just very basic VT100 features. The resulting library appears to work everywhere I tried to use it.
+
+Since it's so young I guess there are a few bugs, or the lib may not compile or work with some operating system, but it's a matter of a few weeks and eventually we'll get it right, and there will be no excuses for not shipping command line tools without built-in line editing support.
+
+The library is currently less than 400 lines of code. In order to use it in your project just look at the *example.c* file in the source distribution, it is trivial. Linenoise is BSD code, so you can use both in free software and commercial software.
+
+## Tested with...
+
+ * Linux text only console ($TERM = linux)
+ * Linux KDE terminal application ($TERM = xterm)
+ * Linux xterm ($TERM = xterm)
+ * Mac OS X iTerm ($TERM = xterm)
+ * Mac OS X default Terminal.app ($TERM = xterm)
+ * OpenBSD 4.5 through an OSX Terminal.app ($TERM = screen)
+ * IBM AIX 6.1
+ * FreeBSD xterm ($TERM = xterm)
+
+Please test it everywhere you can and report back!
+
+## Let's push this forward!
+
+Please fork it and add something interesting and send me a pull request. What's especially interesting are fixes, new key bindings, completion.
+
+Send feedbacks to antirez at gmail
--- /dev/null
+#include <stdio.h>
+#include <stdlib.h>
+#include "linenoise.h"
+
+int main(void) {
+ char *line;
+
+ linenoiseHistoryLoad("history.txt"); /* Load the history at startup */
+ while((line = linenoise("hello> ")) != NULL) {
+ if (line[0] != '\0') {
+ printf("echo: '%s'\n", line);
+ linenoiseHistoryAdd(line);
+ linenoiseHistorySave("history.txt"); /* Save every new entry */
+ }
+ free(line);
+ }
+ return 0;
+}
--- /dev/null
+/* linenoise.c -- guerrilla line editing library against the idea that a
+ * line editing lib needs to be 20,000 lines of C code.
+ *
+ * You can find the latest source code at:
+ *
+ * http://github.com/antirez/linenoise
+ *
+ * Does a number of crazy assumptions that happen to be true in 99.9999% of
+ * the 2010 UNIX computers around.
+ *
+ * Copyright (c) 2010, Salvatore Sanfilippo <antirez at gmail dot com>
+ * 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.
+ *
+ * References:
+ * - http://invisible-island.net/xterm/ctlseqs/ctlseqs.html
+ * - http://www.3waylabs.com/nw/WWW/products/wizcon/vt220.html
+ *
+ * Todo list:
+ * - Switch to gets() if $TERM is something we can't support.
+ * - Filter bogus Ctrl+<char> combinations.
+ * - Win32 support
+ *
+ * Bloat:
+ * - Completion?
+ * - History search like Ctrl+r in readline?
+ *
+ * List of escape sequences used by this program, we do everything just
+ * with three sequences. In order to be so cheap we may have some
+ * flickering effect with some slow terminal, but the lesser sequences
+ * the more compatible.
+ *
+ * CHA (Cursor Horizontal Absolute)
+ * Sequence: ESC [ n G
+ * Effect: moves cursor to column n
+ *
+ * EL (Erase Line)
+ * Sequence: ESC [ n K
+ * Effect: if n is 0 or missing, clear from cursor to end of line
+ * Effect: if n is 1, clear from beginning of line to cursor
+ * Effect: if n is 2, clear entire line
+ *
+ * CUF (CUrsor Forward)
+ * Sequence: ESC [ n C
+ * Effect: moves cursor forward of n chars
+ *
+ */
+
+#include <termios.h>
+#include <unistd.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <errno.h>
+#include <string.h>
+#include <stdlib.h>
+#include <sys/types.h>
+#include <sys/ioctl.h>
+#include <unistd.h>
+
+#define LINENOISE_DEFAULT_HISTORY_MAX_LEN 100
+#define LINENOISE_MAX_LINE 4096
+static char *unsupported_term[] = {"dumb","cons25",NULL};
+
+static struct termios orig_termios; /* in order to restore at exit */
+static int rawmode = 0; /* for atexit() function to check if restore is needed*/
+static int atexit_registered = 0; /* register atexit just 1 time */
+static int history_max_len = LINENOISE_DEFAULT_HISTORY_MAX_LEN;
+static int history_len = 0;
+char **history = NULL;
+
+static void linenoiseAtExit(void);
+int linenoiseHistoryAdd(const char *line);
+
+static int isUnsupportedTerm(void) {
+ char *term = getenv("TERM");
+ int j;
+
+ if (term == NULL) return 0;
+ for (j = 0; unsupported_term[j]; j++)
+ if (!strcasecmp(term,unsupported_term[j])) return 1;
+ return 0;
+}
+
+static void freeHistory(void) {
+ if (history) {
+ int j;
+
+ for (j = 0; j < history_len; j++)
+ free(history[j]);
+ free(history);
+ }
+}
+
+static int enableRawMode(int fd) {
+ struct termios raw;
+
+ if (!isatty(STDIN_FILENO)) goto fatal;
+ if (!atexit_registered) {
+ atexit(linenoiseAtExit);
+ atexit_registered = 1;
+ }
+ if (tcgetattr(fd,&orig_termios) == -1) goto fatal;
+
+ raw = orig_termios; /* modify the original mode */
+ /* input modes: no break, no CR to NL, no parity check, no strip char,
+ * no start/stop output control. */
+ raw.c_iflag &= ~(BRKINT | ICRNL | INPCK | ISTRIP | IXON);
+ /* output modes - disable post processing */
+ raw.c_oflag &= ~(OPOST);
+ /* control modes - set 8 bit chars */
+ raw.c_cflag |= (CS8);
+ /* local modes - choing off, canonical off, no extended functions,
+ * no signal chars (^Z,^C) */
+ raw.c_lflag &= ~(ECHO | ICANON | IEXTEN | ISIG);
+ /* control chars - set return condition: min number of bytes and timer.
+ * We want read to return every single byte, without timeout. */
+ raw.c_cc[VMIN] = 1; raw.c_cc[VTIME] = 0; /* 1 byte, no timer */
+
+ /* put terminal in raw mode after flushing */
+ if (tcsetattr(fd,TCSAFLUSH,&raw) < 0) goto fatal;
+ rawmode = 1;
+ return 0;
+
+fatal:
+ errno = ENOTTY;
+ return -1;
+}
+
+static void disableRawMode(int fd) {
+ /* Don't even check the return value as it's too late. */
+ if (rawmode && tcsetattr(fd,TCSAFLUSH,&orig_termios) != -1)
+ rawmode = 0;
+}
+
+/* At exit we'll try to fix the terminal to the initial conditions. */
+static void linenoiseAtExit(void) {
+ disableRawMode(STDIN_FILENO);
+ freeHistory();
+}
+
+static int getColumns(void) {
+ struct winsize ws;
+
+ if (ioctl(1, TIOCGWINSZ, &ws) == -1) return 80;
+ return ws.ws_col;
+}
+
+static void refreshLine(int fd, const char *prompt, char *buf, size_t len, size_t pos, size_t cols) {
+ char seq[64];
+ size_t plen = strlen(prompt);
+
+ while((plen+pos) >= cols) {
+ buf++;
+ len--;
+ pos--;
+ }
+ while (plen+len > cols) {
+ len--;
+ }
+
+ /* Cursor to left edge */
+ snprintf(seq,64,"\x1b[0G");
+ if (write(fd,seq,strlen(seq)) == -1) return;
+ /* Write the prompt and the current buffer content */
+ if (write(fd,prompt,strlen(prompt)) == -1) return;
+ if (write(fd,buf,len) == -1) return;
+ /* Erase to right */
+ snprintf(seq,64,"\x1b[0K");
+ if (write(fd,seq,strlen(seq)) == -1) return;
+ /* Move cursor to original position. */
+ snprintf(seq,64,"\x1b[0G\x1b[%dC", (int)(pos+plen));
+ if (write(fd,seq,strlen(seq)) == -1) return;
+}
+
+static int linenoisePrompt(int fd, char *buf, size_t buflen, const char *prompt) {
+ size_t plen = strlen(prompt);
+ size_t pos = 0;
+ size_t len = 0;
+ size_t cols = getColumns();
+ int history_index = 0;
+
+ buf[0] = '\0';
+ buflen--; /* Make sure there is always space for the nulterm */
+
+ /* The latest history entry is always our current buffer, that
+ * initially is just an empty string. */
+ linenoiseHistoryAdd("");
+
+ if (write(fd,prompt,plen) == -1) return -1;
+ while(1) {
+ char c;
+ int nread;
+ char seq[2], seq2[2];
+
+ nread = read(fd,&c,1);
+ if (nread <= 0) return len;
+ switch(c) {
+ case 13: /* enter */
+ case 4: /* ctrl-d */
+ history_len--;
+ free(history[history_len]);
+ return (len == 0 && c == 4) ? -1 : (int)len;
+ case 3: /* ctrl-c */
+ errno = EAGAIN;
+ return -1;
+ case 127: /* backspace */
+ case 8: /* ctrl-h */
+ if (pos > 0 && len > 0) {
+ memmove(buf+pos-1,buf+pos,len-pos);
+ pos--;
+ len--;
+ buf[len] = '\0';
+ refreshLine(fd,prompt,buf,len,pos,cols);
+ }
+ break;
+ case 20: /* ctrl-t */
+ if (pos > 0 && pos < len) {
+ int aux = buf[pos-1];
+ buf[pos-1] = buf[pos];
+ buf[pos] = aux;
+ if (pos != len-1) pos++;
+ refreshLine(fd,prompt,buf,len,pos,cols);
+ }
+ break;
+ case 2: /* ctrl-b */
+ goto left_arrow;
+ case 6: /* ctrl-f */
+ goto right_arrow;
+ case 16: /* ctrl-p */
+ seq[1] = 65;
+ goto up_down_arrow;
+ case 14: /* ctrl-n */
+ seq[1] = 66;
+ goto up_down_arrow;
+ break;
+ case 27: /* escape sequence */
+ if (read(fd,seq,2) == -1) break;
+ if (seq[0] == 91 && seq[1] == 68) {
+left_arrow:
+ /* left arrow */
+ if (pos > 0) {
+ pos--;
+ refreshLine(fd,prompt,buf,len,pos,cols);
+ }
+ } else if (seq[0] == 91 && seq[1] == 67) {
+right_arrow:
+ /* right arrow */
+ if (pos != len) {
+ pos++;
+ refreshLine(fd,prompt,buf,len,pos,cols);
+ }
+ } else if (seq[0] == 91 && (seq[1] == 65 || seq[1] == 66)) {
+up_down_arrow:
+ /* up and down arrow: history */
+ if (history_len > 1) {
+ /* Update the current history entry before to
+ * overwrite it with tne next one. */
+ free(history[history_len-1-history_index]);
+ history[history_len-1-history_index] = strdup(buf);
+ /* Show the new entry */
+ history_index += (seq[1] == 65) ? 1 : -1;
+ if (history_index < 0) {
+ history_index = 0;
+ break;
+ } else if (history_index >= history_len) {
+ history_index = history_len-1;
+ break;
+ }
+ strncpy(buf,history[history_len-1-history_index],buflen);
+ buf[buflen] = '\0';
+ len = pos = strlen(buf);
+ refreshLine(fd,prompt,buf,len,pos,cols);
+ }
+ } else if (seq[0] == 91 && seq[1] > 48 && seq[1] < 55) {
+ /* extended escape */
+ if (read(fd,seq2,2) == -1) break;
+ if (seq[1] == 51 && seq2[0] == 126) {
+ /* delete */
+ if (len > 0 && pos < len) {
+ memmove(buf+pos,buf+pos+1,len-pos-1);
+ len--;
+ buf[len] = '\0';
+ refreshLine(fd,prompt,buf,len,pos,cols);
+ }
+ }
+ }
+ break;
+ default:
+ if (len < buflen) {
+ if (len == pos) {
+ buf[pos] = c;
+ pos++;
+ len++;
+ buf[len] = '\0';
+ if (plen+len < cols) {
+ /* Avoid a full update of the line in the
+ * trivial case. */
+ if (write(fd,&c,1) == -1) return -1;
+ } else {
+ refreshLine(fd,prompt,buf,len,pos,cols);
+ }
+ } else {
+ memmove(buf+pos+1,buf+pos,len-pos);
+ buf[pos] = c;
+ len++;
+ pos++;
+ buf[len] = '\0';
+ refreshLine(fd,prompt,buf,len,pos,cols);
+ }
+ }
+ break;
+ case 21: /* Ctrl+u, delete the whole line. */
+ buf[0] = '\0';
+ pos = len = 0;
+ refreshLine(fd,prompt,buf,len,pos,cols);
+ break;
+ case 11: /* Ctrl+k, delete from current to end of line. */
+ buf[pos] = '\0';
+ len = pos;
+ refreshLine(fd,prompt,buf,len,pos,cols);
+ break;
+ case 1: /* Ctrl+a, go to the start of the line */
+ pos = 0;
+ refreshLine(fd,prompt,buf,len,pos,cols);
+ break;
+ case 5: /* ctrl+e, go to the end of the line */
+ pos = len;
+ refreshLine(fd,prompt,buf,len,pos,cols);
+ break;
+ }
+ }
+ return len;
+}
+
+static int linenoiseRaw(char *buf, size_t buflen, const char *prompt) {
+ int fd = STDIN_FILENO;
+ int count;
+
+ if (buflen == 0) {
+ errno = EINVAL;
+ return -1;
+ }
+ if (!isatty(STDIN_FILENO)) {
+ if (fgets(buf, buflen, stdin) == NULL) return -1;
+ count = strlen(buf);
+ if (count && buf[count-1] == '\n') {
+ count--;
+ buf[count] = '\0';
+ }
+ } else {
+ if (enableRawMode(fd) == -1) return -1;
+ count = linenoisePrompt(fd, buf, buflen, prompt);
+ disableRawMode(fd);
+ printf("\n");
+ }
+ return count;
+}
+
+char *linenoise(const char *prompt) {
+ char buf[LINENOISE_MAX_LINE];
+ int count;
+
+ if (isUnsupportedTerm()) {
+ size_t len;
+
+ printf("%s",prompt);
+ fflush(stdout);
+ if (fgets(buf,LINENOISE_MAX_LINE,stdin) == NULL) return NULL;
+ len = strlen(buf);
+ while(len && (buf[len-1] == '\n' || buf[len-1] == '\r')) {
+ len--;
+ buf[len] = '\0';
+ }
+ return strdup(buf);
+ } else {
+ count = linenoiseRaw(buf,LINENOISE_MAX_LINE,prompt);
+ if (count == -1) return NULL;
+ return strdup(buf);
+ }
+}
+
+/* Using a circular buffer is smarter, but a bit more complex to handle. */
+int linenoiseHistoryAdd(const char *line) {
+ char *linecopy;
+
+ if (history_max_len == 0) return 0;
+ if (history == NULL) {
+ history = malloc(sizeof(char*)*history_max_len);
+ if (history == NULL) return 0;
+ memset(history,0,(sizeof(char*)*history_max_len));
+ }
+ linecopy = strdup(line);
+ if (!linecopy) return 0;
+ if (history_len == history_max_len) {
+ free(history[0]);
+ memmove(history,history+1,sizeof(char*)*(history_max_len-1));
+ history_len--;
+ }
+ history[history_len] = linecopy;
+ history_len++;
+ return 1;
+}
+
+int linenoiseHistorySetMaxLen(int len) {
+ char **new;
+
+ if (len < 1) return 0;
+ if (history) {
+ int tocopy = history_len;
+
+ new = malloc(sizeof(char*)*len);
+ if (new == NULL) return 0;
+ if (len < tocopy) tocopy = len;
+ memcpy(new,history+(history_max_len-tocopy), sizeof(char*)*tocopy);
+ free(history);
+ history = new;
+ }
+ history_max_len = len;
+ if (history_len > history_max_len)
+ history_len = history_max_len;
+ return 1;
+}
+
+/* Save the history in the specified file. On success 0 is returned
+ * otherwise -1 is returned. */
+int linenoiseHistorySave(char *filename) {
+ FILE *fp = fopen(filename,"w");
+ int j;
+
+ if (fp == NULL) return -1;
+ for (j = 0; j < history_len; j++)
+ fprintf(fp,"%s\n",history[j]);
+ fclose(fp);
+ return 0;
+}
+
+/* Load the history from the specified file. If the file does not exist
+ * zero is returned and no operation is performed.
+ *
+ * If the file exists and the operation succeeded 0 is returned, otherwise
+ * on error -1 is returned. */
+int linenoiseHistoryLoad(char *filename) {
+ FILE *fp = fopen(filename,"r");
+ char buf[LINENOISE_MAX_LINE];
+
+ if (fp == NULL) return -1;
+
+ while (fgets(buf,LINENOISE_MAX_LINE,fp) != NULL) {
+ char *p;
+
+ p = strchr(buf,'\r');
+ if (!p) p = strchr(buf,'\n');
+ if (p) *p = '\0';
+ linenoiseHistoryAdd(buf);
+ }
+ fclose(fp);
+ return 0;
+}
--- /dev/null
+/* linenoise.h -- guerrilla line editing library against the idea that a
+ * line editing lib needs to be 20,000 lines of C code.
+ *
+ * See linenoise.c for more information.
+ *
+ * Copyright (c) 2010, Salvatore Sanfilippo <antirez at gmail dot com>
+ * 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 __LINENOISE_H
+#define __LINENOISE_H
+
+char *linenoise(const char *prompt);
+int linenoiseHistoryAdd(const char *line);
+int linenoiseHistorySetMaxLen(int len);
+int linenoiseHistorySave(char *filename);
+int linenoiseHistoryLoad(char *filename);
+
+#endif /* __LINENOISE_H */
# default. You can specify a custom pid file location here.
pidfile /var/run/redis.pid
-# Accept connections on the specified port, default is 6379
+# Accept connections on the specified port, default is 6379.
port 6379
# If you want you can bind a single interface, if the bind option is not
#
# bind 127.0.0.1
+# Specify the path for the unix socket that will be used to listen for
+# incoming connections. There is no default, so Redis will not listen
+# on a unix socket when not specified.
+#
+# unixsocket /tmp/redis.sock
+
# Close the connection after a client is idle for N seconds (0 to disable)
timeout 300
#
# masterauth <master-password>
+# When a slave lost the connection with the master, or when the replication
+# is still in progress, the slave can act in two different ways:
+#
+# 1) if slave-serve-stale-data is set to 'yes' (the default) the slave will
+# still reply to client requests, possibly with out of data data, or the
+# data set may just be empty if this is the first synchronization.
+#
+# 2) if slave-serve-stale data is set to 'no' the slave will reply with
+# an error "SYNC with master in progress" to all the kind of commands
+# but to INFO and SLAVEOF.
+#
+slave-serve-stale-data yes
+
################################## SECURITY ###################################
# Require clients to issue AUTH <PASSWORD> before processing any other
#
# requirepass foobared
+# Command renaming.
+#
+# It is possilbe to change the name of dangerous commands in a shared
+# environment. For instance the CONFIG command may be renamed into something
+# of hard to guess so that it will be still available for internal-use
+# tools but not available for general clients.
+#
+# Example:
+#
+# rename-command CONFIG b840fc02d524045429941cc15f59e41cb7be6c52
+#
+# It is also possilbe to completely kill a command renaming it into
+# an empty string:
+#
+# rename-command CONFIG ""
+
################################### LIMITS ####################################
# Set the max number of connected clients at the same time. By default there
# volatile-random -> remove a random key with an expire set
# allkeys->random -> remove a random key, any key
# volatile-ttl -> remove the key with the nearest expire time (minor TTL)
+# noeviction -> don't expire at all, just return an error on write operations
+#
+# Note: with all the kind of policies, Redis will return an error on write
+# operations, when there are not suitable keys for eviction.
+#
+# At the date of writing this commands are: set setnx setex append
+# incr decr rpush lpush rpushx lpushx linsert lset rpoplpush sadd
+# sinter sinterstore sunion sunionstore sdiff sdiffstore zadd zincrby
+# zunionstore zinterstore hset hsetnx hmset hincrby incrby decrby
+# getset mset msetnx exec sort
+#
+# The default is:
#
# maxmemory-policy volatile-lru
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 vm.o pubsub.o multi.o debug.o sort.o intset.o syncio.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 linenoise.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
CHECKAOFOBJ = redis-check-aof.o
CHECKDUMPPRGNAME = redis-check-dump
CHECKAOFPRGNAME = redis-check-aof
-all: redis-server redis-benchmark redis-cli redis-check-dump redis-check-aof
+all: redis-benchmark redis-cli redis-check-dump redis-check-aof redis-server
# Deps (use make dep to generate this)
adlist.o: adlist.c adlist.h zmalloc.h
zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h sha1.h
dict.o: dict.c fmacros.h dict.h zmalloc.h
intset.o: intset.c intset.h zmalloc.h
-linenoise.o: linenoise.c fmacros.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 \
zmalloc.h
redis-check-aof.o: redis-check-aof.c fmacros.h config.h
redis-check-dump.o: redis-check-dump.c lzf.h
-redis-cli.o: redis-cli.c fmacros.h version.h anet.h sds.h adlist.h \
- zmalloc.h linenoise.h
+redis-cli.o: redis-cli.c fmacros.h version.h sds.h adlist.h zmalloc.h
redis.o: redis.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
release.o: release.c release.h
@echo ""
redis-benchmark: $(BENCHOBJ)
- $(CC) -o $(BENCHPRGNAME) $(CCOPT) $(DEBUG) $(BENCHOBJ)
+ cd ../deps/hiredis && make static
+ $(CC) -o $(BENCHPRGNAME) $(CCOPT) $(DEBUG) $(BENCHOBJ) ../deps/hiredis/libhiredis.a
+
+redis-benchmark.o:
+ $(CC) -c $(CFLAGS) -I../deps/hiredis $(DEBUG) $(COMPILE_TIME) $<
redis-cli: $(CLIOBJ)
- $(CC) -o $(CLIPRGNAME) $(CCOPT) $(DEBUG) $(CLIOBJ)
+ cd ../deps/hiredis && make static ARCH="$(ARCH)"
+ cd ../deps/linenoise && make ARCH="$(ARCH)"
+ $(CC) -o $(CLIPRGNAME) $(CCOPT) $(DEBUG) $(CLIOBJ) ../deps/hiredis/libhiredis.a ../deps/linenoise/linenoise.o
+
+redis-cli.o:
+ $(CC) -c $(CFLAGS) -I../deps/hiredis -I../deps/linenoise $(DEBUG) $(COMPILE_TIME) $<
redis-check-dump: $(CHECKDUMPOBJ)
$(CC) -o $(CHECKDUMPPRGNAME) $(CCOPT) $(DEBUG) $(CHECKDUMPOBJ)
clean:
rm -rf $(PRGNAME) $(BENCHPRGNAME) $(CLIPRGNAME) $(CHECKDUMPPRGNAME) $(CHECKAOFPRGNAME) *.o *.gcda *.gcno *.gcov
+ cd ../deps/hiredis && make clean
+ cd ../deps/linenoise && make clean
dep:
$(CC) -MM *.c
#include <sys/types.h>
#include <sys/socket.h>
+#include <sys/un.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
return ANET_OK;
}
+static int anetCreateSocket(char *err, int domain) {
+ int s, on = 1;
+ if ((s = socket(domain, SOCK_STREAM, 0)) == -1) {
+ anetSetError(err, "creating socket: %s\n", strerror(errno));
+ return ANET_ERR;
+ }
+
+ /* Make sure connection-intensive things like the redis benckmark
+ * will be able to close/open sockets a zillion of times */
+ if (setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)) == -1) {
+ anetSetError(err, "setsockopt SO_REUSEADDR: %s\n", strerror(errno));
+ return ANET_ERR;
+ }
+ return s;
+}
+
#define ANET_CONNECT_NONE 0
#define ANET_CONNECT_NONBLOCK 1
static int anetTcpGenericConnect(char *err, char *addr, int port, int flags)
{
- int s, on = 1;
+ int s;
struct sockaddr_in sa;
- if ((s = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
- anetSetError(err, "creating socket: %s\n", strerror(errno));
+ if ((s = anetCreateSocket(err,AF_INET)) == ANET_ERR)
return ANET_ERR;
- }
- /* Make sure connection-intensive things like the redis benckmark
- * will be able to close/open sockets a zillion of times */
- setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
sa.sin_family = AF_INET;
sa.sin_port = htons(port);
return anetTcpGenericConnect(err,addr,port,ANET_CONNECT_NONBLOCK);
}
+int anetUnixGenericConnect(char *err, char *path, int flags)
+{
+ int s;
+ struct sockaddr_un sa;
+
+ if ((s = anetCreateSocket(err,AF_LOCAL)) == ANET_ERR)
+ return ANET_ERR;
+
+ sa.sun_family = AF_LOCAL;
+ strncpy(sa.sun_path,path,sizeof(sa.sun_path)-1);
+ if (flags & ANET_CONNECT_NONBLOCK) {
+ if (anetNonBlock(err,s) != ANET_OK)
+ return ANET_ERR;
+ }
+ if (connect(s,(struct sockaddr*)&sa,sizeof(sa)) == -1) {
+ if (errno == EINPROGRESS &&
+ flags & ANET_CONNECT_NONBLOCK)
+ return s;
+
+ anetSetError(err, "connect: %s\n", strerror(errno));
+ close(s);
+ return ANET_ERR;
+ }
+ return s;
+}
+
+int anetUnixConnect(char *err, char *path)
+{
+ return anetUnixGenericConnect(err,path,ANET_CONNECT_NONE);
+}
+
+int anetUnixNonBlockConnect(char *err, char *path)
+{
+ return anetUnixGenericConnect(err,path,ANET_CONNECT_NONBLOCK);
+}
+
/* Like read(2) but make sure 'count' is read before to return
* (unless error or EOF condition is encountered) */
int anetRead(int fd, char *buf, int count)
return totlen;
}
-int anetTcpServer(char *err, int port, char *bindaddr)
-{
- int s, on = 1;
- struct sockaddr_in sa;
-
- if ((s = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
- anetSetError(err, "socket: %s\n", strerror(errno));
+static int anetListen(char *err, int s, struct sockaddr *sa, socklen_t len) {
+ if (bind(s,sa,len) == -1) {
+ anetSetError(err, "bind: %s\n", strerror(errno));
+ close(s);
return ANET_ERR;
}
- if (setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)) == -1) {
- anetSetError(err, "setsockopt SO_REUSEADDR: %s\n", strerror(errno));
+ if (listen(s, 511) == -1) { /* the magic 511 constant is from nginx */
+ anetSetError(err, "listen: %s\n", strerror(errno));
close(s);
return ANET_ERR;
}
+ return ANET_OK;
+}
+
+int anetTcpServer(char *err, int port, char *bindaddr)
+{
+ int s;
+ struct sockaddr_in sa;
+
+ if ((s = anetCreateSocket(err,AF_INET)) == ANET_ERR)
+ return ANET_ERR;
+
memset(&sa,0,sizeof(sa));
sa.sin_family = AF_INET;
sa.sin_port = htons(port);
sa.sin_addr.s_addr = htonl(INADDR_ANY);
- if (bindaddr) {
- if (inet_aton(bindaddr, &sa.sin_addr) == 0) {
- anetSetError(err, "Invalid bind address\n");
- close(s);
- return ANET_ERR;
- }
- }
- if (bind(s, (struct sockaddr*)&sa, sizeof(sa)) == -1) {
- anetSetError(err, "bind: %s\n", strerror(errno));
+ if (bindaddr && inet_aton(bindaddr, &sa.sin_addr) == 0) {
+ anetSetError(err, "Invalid bind address\n");
close(s);
return ANET_ERR;
}
- if (listen(s, 511) == -1) { /* the magic 511 constant is from nginx */
- anetSetError(err, "listen: %s\n", strerror(errno));
- close(s);
+ if (anetListen(err,s,(struct sockaddr*)&sa,sizeof(sa)) == ANET_ERR)
return ANET_ERR;
- }
return s;
}
-int anetAccept(char *err, int serversock, char *ip, int *port)
+int anetUnixServer(char *err, char *path)
{
- int fd;
- struct sockaddr_in sa;
- unsigned int saLen;
+ int s;
+ struct sockaddr_un sa;
+
+ if ((s = anetCreateSocket(err,AF_LOCAL)) == ANET_ERR)
+ return ANET_ERR;
+
+ memset(&sa,0,sizeof(sa));
+ sa.sun_family = AF_LOCAL;
+ strncpy(sa.sun_path,path,sizeof(sa.sun_path)-1);
+ if (anetListen(err,s,(struct sockaddr*)&sa,sizeof(sa)) == ANET_ERR)
+ return ANET_ERR;
+ return s;
+}
+static int anetGenericAccept(char *err, int s, struct sockaddr *sa, socklen_t *len) {
+ int fd;
while(1) {
- saLen = sizeof(sa);
- fd = accept(serversock, (struct sockaddr*)&sa, &saLen);
+ fd = accept(s,sa,len);
if (fd == -1) {
if (errno == EINTR)
continue;
}
break;
}
+ return fd;
+}
+
+int anetTcpAccept(char *err, int s, char *ip, int *port) {
+ int fd;
+ struct sockaddr_in sa;
+ socklen_t salen = sizeof(sa);
+ if ((fd = anetGenericAccept(err,s,(struct sockaddr*)&sa,&salen)) == ANET_ERR)
+ return ANET_ERR;
+
if (ip) strcpy(ip,inet_ntoa(sa.sin_addr));
if (port) *port = ntohs(sa.sin_port);
return fd;
}
+
+int anetUnixAccept(char *err, int s) {
+ int fd;
+ struct sockaddr_un sa;
+ socklen_t salen = sizeof(sa);
+ if ((fd = anetGenericAccept(err,s,(struct sockaddr*)&sa,&salen)) == ANET_ERR)
+ return ANET_ERR;
+
+ return fd;
+}
int anetTcpConnect(char *err, char *addr, int port);
int anetTcpNonBlockConnect(char *err, char *addr, int port);
+int anetUnixConnect(char *err, char *path);
+int anetUnixNonBlockConnect(char *err, char *path);
int anetRead(int fd, char *buf, int count);
int anetResolve(char *err, char *host, char *ipbuf);
int anetTcpServer(char *err, int port, char *bindaddr);
-int anetAccept(char *err, int serversock, char *ip, int *port);
+int anetUnixServer(char *err, char *path);
+int anetTcpAccept(char *err, int serversock, char *ip, int *port);
+int anetUnixAccept(char *err, int serversock);
int anetWrite(int fd, char *buf, int count);
int anetNonBlock(char *err, int fd);
int anetTcpNoDelay(char *err, int fd);
FILE *fp = fopen(filename,"r");
struct redis_stat sb;
int appendonly = server.appendonly;
+ long loops = 0;
if (redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0)
return REDIS_ERR;
server.appendonly = 0;
fakeClient = createFakeClient();
+ startLoading(fp);
+
while(1) {
int argc, j;
unsigned long len;
struct redisCommand *cmd;
int force_swapout;
+ /* Serve the clients from time to time */
+ if (!(loops++ % 1000)) {
+ loadingProgress(ftello(fp));
+ aeProcessEvents(server.el, AE_FILE_EVENTS|AE_DONT_WAIT);
+ }
+
if (fgets(buf,sizeof(buf),fp) == NULL) {
if (feof(fp))
break;
fclose(fp);
freeFakeClient(fakeClient);
server.appendonly = appendonly;
+ stopLoading();
return REDIS_OK;
readerr:
char tmpfile[256];
if (server.vm_enabled) vmReopenSwapFile();
- close(server.fd);
+ if (server.ipfd > 0) close(server.ipfd);
+ if (server.sofd > 0) close(server.sofd);
snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) {
_exit(0);
}
/* Split into arguments */
- argv = sdssplitlen(line,sdslen(line)," ",1,&argc);
+ argv = sdssplitargs(line,&argc);
sdstolower(argv[0]);
/* Execute config directives */
}
} else if (!strcasecmp(argv[0],"bind") && argc == 2) {
server.bindaddr = zstrdup(argv[1]);
+ } else if (!strcasecmp(argv[0],"unixsocket") && argc == 2) {
+ server.unixsocket = zstrdup(argv[1]);
} else if (!strcasecmp(argv[0],"save") && argc == 3) {
int seconds = atoi(argv[1]);
int changes = atoi(argv[2]);
server.maxmemory_policy = REDIS_MAXMEMORY_ALLKEYS_LRU;
} else if (!strcasecmp(argv[1],"allkeys-random")) {
server.maxmemory_policy = REDIS_MAXMEMORY_ALLKEYS_RANDOM;
+ } else if (!strcasecmp(argv[1],"noeviction")) {
+ server.maxmemory_policy = REDIS_MAXMEMORY_NO_EVICTION;
} else {
err = "Invalid maxmemory policy";
goto loaderr;
server.replstate = REDIS_REPL_CONNECT;
} else if (!strcasecmp(argv[0],"masterauth") && argc == 2) {
server.masterauth = zstrdup(argv[1]);
+ } else if (!strcasecmp(argv[0],"slave-serve-stale-data") && argc == 2) {
+ if ((server.repl_serve_stale_data = yesnotoi(argv[1])) == -1) {
+ err = "argument must be 'yes' or 'no'"; goto loaderr;
+ }
} else if (!strcasecmp(argv[0],"glueoutputbuf") && argc == 2) {
if ((server.glueoutputbuf = yesnotoi(argv[1])) == -1) {
err = "argument must be 'yes' or 'no'"; goto loaderr;
server.vm_pages = memtoll(argv[1], NULL);
} else if (!strcasecmp(argv[0],"vm-max-threads") && argc == 2) {
server.vm_max_threads = strtoll(argv[1], NULL, 10);
- } else if (!strcasecmp(argv[0],"hash-max-zipmap-entries") && argc == 2){
+ } else if (!strcasecmp(argv[0],"hash-max-zipmap-entries") && argc == 2) {
server.hash_max_zipmap_entries = memtoll(argv[1], NULL);
- } else if (!strcasecmp(argv[0],"hash-max-zipmap-value") && argc == 2){
+ } else if (!strcasecmp(argv[0],"hash-max-zipmap-value") && argc == 2) {
server.hash_max_zipmap_value = memtoll(argv[1], NULL);
} else if (!strcasecmp(argv[0],"list-max-ziplist-entries") && argc == 2){
server.list_max_ziplist_entries = memtoll(argv[1], NULL);
- } else if (!strcasecmp(argv[0],"list-max-ziplist-value") && argc == 2){
+ } else if (!strcasecmp(argv[0],"list-max-ziplist-value") && argc == 2) {
server.list_max_ziplist_value = memtoll(argv[1], NULL);
- } else if (!strcasecmp(argv[0],"set-max-intset-entries") && argc == 2){
+ } else if (!strcasecmp(argv[0],"set-max-intset-entries") && argc == 2) {
server.set_max_intset_entries = memtoll(argv[1], NULL);
+ } else if (!strcasecmp(argv[0],"rename-command") && argc == 3) {
+ struct redisCommand *cmd = lookupCommand(argv[1]);
+ int retval;
+
+ if (!cmd) {
+ err = "No such command in rename-command";
+ goto loaderr;
+ }
+
+ /* If the target command name is the emtpy string we just
+ * remove it from the command table. */
+ retval = dictDelete(server.commands, argv[1]);
+ redisAssert(retval == DICT_OK);
+
+ /* Otherwise we re-add the command under a different name. */
+ if (sdslen(argv[2]) != 0) {
+ sds copy = sdsdup(argv[2]);
+
+ retval = dictAdd(server.commands, copy, cmd);
+ if (retval != DICT_OK) {
+ sdsfree(copy);
+ err = "Target command name already exists"; goto loaderr;
+ }
+ }
} else {
err = "Bad directive or wrong number of arguments"; goto loaderr;
}
server.maxmemory_policy = REDIS_MAXMEMORY_ALLKEYS_LRU;
} else if (!strcasecmp(o->ptr,"allkeys-random")) {
server.maxmemory_policy = REDIS_MAXMEMORY_ALLKEYS_RANDOM;
+ } else if (!strcasecmp(o->ptr,"noeviction")) {
+ server.maxmemory_policy = REDIS_MAXMEMORY_NO_EVICTION;
} else {
goto badfmt;
}
appendServerSaveParams(seconds, changes);
}
sdsfreesplitres(v,vlen);
+ } else if (!strcasecmp(c->argv[2]->ptr,"slave-serve-stale-data")) {
+ int yn = yesnotoi(o->ptr);
+
+ if (yn == -1) goto badfmt;
+ server.repl_serve_stale_data = yn;
} else {
addReplyErrorFormat(c,"Unsupported CONFIG parameter: %s",
(char*)c->argv[2]->ptr);
case REDIS_MAXMEMORY_VOLATILE_RANDOM: s = "volatile-random"; break;
case REDIS_MAXMEMORY_ALLKEYS_LRU: s = "allkeys-lru"; break;
case REDIS_MAXMEMORY_ALLKEYS_RANDOM: s = "allkeys-random"; break;
+ case REDIS_MAXMEMORY_NO_EVICTION: s = "noeviction"; break;
default: s = "unknown"; break; /* too harmless to panic */
}
addReplyBulkCString(c,"maxmemory-policy");
sdsfree(buf);
matches++;
}
+ if (stringmatch(pattern,"slave-serve-stale-data",0)) {
+ addReplyBulkCString(c,"slave-serve-stale-data");
+ addReplyBulkCString(c,server.repl_serve_stale_data ? "yes" : "no");
+ matches++;
+ }
setDeferredMultiBulkLength(c,replylen,matches*2);
}
if (de) {
robj *val = dictGetEntryVal(de);
- /* Update the access time for the aging algorithm. */
- val->lru = server.lruclock;
+ /* Update the access time for the aging algorithm.
+ * Don't do it if we have a saving child, as this will trigger
+ * a copy on write madness. */
+ if (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1)
+ val->lru = server.lruclock;
if (server.vm_enabled) {
if (val->storage == REDIS_VM_MEMORY ||
* will be consistent even if we allow write operations against expiring
* keys. */
void propagateExpire(redisDb *db, robj *key) {
- struct redisCommand *cmd;
robj *argv[2];
- cmd = lookupCommand("del");
argv[0] = createStringObject("DEL",3);
argv[1] = key;
incrRefCount(key);
if (server.appendonly)
- feedAppendOnlyFile(cmd,db->id,argv,2);
+ feedAppendOnlyFile(server.delCommand,db->id,argv,2);
if (listLength(server.slaves))
replicationFeedSlaves(server.slaves,db->id,argv,2);
#include <assert.h>
#include <limits.h>
#include <sys/time.h>
+#include <ctype.h>
#include "dict.h"
#include "zmalloc.h"
return hash;
}
+/* And a case insensitive version */
+unsigned int dictGenCaseHashFunction(const unsigned char *buf, int len) {
+ unsigned int hash = 5381;
+
+ while (len--)
+ hash = ((hash << 5) + hash) + (tolower(*buf++)); /* hash * 33 + c */
+ return hash;
+}
+
/* ----------------------------- API implementation ------------------------- */
/* Reset an hashtable already initialized with ht_init().
dictEntry *dictGetRandomKey(dict *d);
void dictPrintStats(dict *d);
unsigned int dictGenHashFunction(const unsigned char *buf, int len);
+unsigned int dictGenCaseHashFunction(const unsigned char *buf, int len);
void dictEmpty(dict *d);
void dictEnableResize(void);
void dictDisableResize(void);
+++ /dev/null
-/* linenoise.c -- guerrilla line editing library against the idea that a
- * line editing lib needs to be 20,000 lines of C code.
- *
- * You can find the latest source code at:
- *
- * http://github.com/antirez/linenoise
- *
- * Does a number of crazy assumptions that happen to be true in 99.9999% of
- * the 2010 UNIX computers around.
- *
- * Copyright (c) 2010, Salvatore Sanfilippo <antirez at gmail dot com>
- * 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.
- *
- * References:
- * - http://invisible-island.net/xterm/ctlseqs/ctlseqs.html
- * - http://www.3waylabs.com/nw/WWW/products/wizcon/vt220.html
- *
- * Todo list:
- * - Switch to gets() if $TERM is something we can't support.
- * - Filter bogus Ctrl+<char> combinations.
- * - Win32 support
- *
- * Bloat:
- * - Completion?
- * - History search like Ctrl+r in readline?
- *
- * List of escape sequences used by this program, we do everything just
- * with three sequences. In order to be so cheap we may have some
- * flickering effect with some slow terminal, but the lesser sequences
- * the more compatible.
- *
- * CHA (Cursor Horizontal Absolute)
- * Sequence: ESC [ n G
- * Effect: moves cursor to column n
- *
- * EL (Erase Line)
- * Sequence: ESC [ n K
- * Effect: if n is 0 or missing, clear from cursor to end of line
- * Effect: if n is 1, clear from beginning of line to cursor
- * Effect: if n is 2, clear entire line
- *
- * CUF (CUrsor Forward)
- * Sequence: ESC [ n C
- * Effect: moves cursor forward of n chars
- *
- */
-
-#include "fmacros.h"
-
-#include <termios.h>
-#include <unistd.h>
-#include <stdlib.h>
-#include <stdio.h>
-#include <errno.h>
-#include <string.h>
-#include <stdlib.h>
-#include <sys/types.h>
-#include <sys/ioctl.h>
-#include <unistd.h>
-
-#define LINENOISE_DEFAULT_HISTORY_MAX_LEN 100
-#define LINENOISE_MAX_LINE 4096
-static char *unsupported_term[] = {"dumb","cons25",NULL};
-
-static struct termios orig_termios; /* in order to restore at exit */
-static int rawmode = 0; /* for atexit() function to check if restore is needed*/
-static int atexit_registered = 0; /* register atexit just 1 time */
-static int history_max_len = LINENOISE_DEFAULT_HISTORY_MAX_LEN;
-static int history_len = 0;
-char **history = NULL;
-
-static void linenoiseAtExit(void);
-int linenoiseHistoryAdd(const char *line);
-
-static int isUnsupportedTerm(void) {
- char *term = getenv("TERM");
- int j;
-
- if (term == NULL) return 0;
- for (j = 0; unsupported_term[j]; j++)
- if (!strcasecmp(term,unsupported_term[j])) return 1;
- return 0;
-}
-
-static void freeHistory(void) {
- if (history) {
- int j;
-
- for (j = 0; j < history_len; j++)
- free(history[j]);
- free(history);
- }
-}
-
-static int enableRawMode(int fd) {
- struct termios raw;
-
- if (!isatty(STDIN_FILENO)) goto fatal;
- if (!atexit_registered) {
- atexit(linenoiseAtExit);
- atexit_registered = 1;
- }
- if (tcgetattr(fd,&orig_termios) == -1) goto fatal;
-
- raw = orig_termios; /* modify the original mode */
- /* input modes: no break, no CR to NL, no parity check, no strip char,
- * no start/stop output control. */
- raw.c_iflag &= ~(BRKINT | ICRNL | INPCK | ISTRIP | IXON);
- /* output modes - disable post processing */
- raw.c_oflag &= ~(OPOST);
- /* control modes - set 8 bit chars */
- raw.c_cflag |= (CS8);
- /* local modes - choing off, canonical off, no extended functions,
- * no signal chars (^Z,^C) */
- raw.c_lflag &= ~(ECHO | ICANON | IEXTEN | ISIG);
- /* control chars - set return condition: min number of bytes and timer.
- * We want read to return every single byte, without timeout. */
- raw.c_cc[VMIN] = 1; raw.c_cc[VTIME] = 0; /* 1 byte, no timer */
-
- /* put terminal in raw mode after flushing */
- if (tcsetattr(fd,TCSAFLUSH,&raw) < 0) goto fatal;
- rawmode = 1;
- return 0;
-
-fatal:
- errno = ENOTTY;
- return -1;
-}
-
-static void disableRawMode(int fd) {
- /* Don't even check the return value as it's too late. */
- if (rawmode && tcsetattr(fd,TCSAFLUSH,&orig_termios) != -1)
- rawmode = 0;
-}
-
-/* At exit we'll try to fix the terminal to the initial conditions. */
-static void linenoiseAtExit(void) {
- disableRawMode(STDIN_FILENO);
- freeHistory();
-}
-
-static int getColumns(void) {
- struct winsize ws;
-
- if (ioctl(1, TIOCGWINSZ, &ws) == -1) return 80;
- return ws.ws_col;
-}
-
-static void refreshLine(int fd, const char *prompt, char *buf, size_t len, size_t pos, size_t cols) {
- char seq[64];
- size_t plen = strlen(prompt);
-
- while((plen+pos) >= cols) {
- buf++;
- len--;
- pos--;
- }
- while (plen+len > cols) {
- len--;
- }
-
- /* Cursor to left edge */
- snprintf(seq,64,"\x1b[0G");
- if (write(fd,seq,strlen(seq)) == -1) return;
- /* Write the prompt and the current buffer content */
- if (write(fd,prompt,strlen(prompt)) == -1) return;
- if (write(fd,buf,len) == -1) return;
- /* Erase to right */
- snprintf(seq,64,"\x1b[0K");
- if (write(fd,seq,strlen(seq)) == -1) return;
- /* Move cursor to original position. */
- snprintf(seq,64,"\x1b[0G\x1b[%dC", (int)(pos+plen));
- if (write(fd,seq,strlen(seq)) == -1) return;
-}
-
-static int linenoisePrompt(int fd, char *buf, size_t buflen, const char *prompt) {
- size_t plen = strlen(prompt);
- size_t pos = 0;
- size_t len = 0;
- size_t cols = getColumns();
- int history_index = 0;
-
- buf[0] = '\0';
- buflen--; /* Make sure there is always space for the nulterm */
-
- /* The latest history entry is always our current buffer, that
- * initially is just an empty string. */
- linenoiseHistoryAdd("");
-
- if (write(fd,prompt,plen) == -1) return -1;
- while(1) {
- char c;
- int nread;
- char seq[2];
-
- nread = read(fd,&c,1);
- if (nread <= 0) return len;
- switch(c) {
- case 13: /* enter */
- case 4: /* ctrl-d */
- history_len--;
- free(history[history_len]);
- return (len == 0 && c == 4) ? -1 : (int)len;
- case 3: /* ctrl-c */
- errno = EAGAIN;
- return -1;
- case 127: /* backspace */
- case 8: /* ctrl-h */
- if (pos > 0 && len > 0) {
- memmove(buf+pos-1,buf+pos,len-pos);
- pos--;
- len--;
- buf[len] = '\0';
- refreshLine(fd,prompt,buf,len,pos,cols);
- }
- break;
- case 20: /* ctrl-t */
- if (pos > 0 && pos < len) {
- int aux = buf[pos-1];
- buf[pos-1] = buf[pos];
- buf[pos] = aux;
- if (pos != len-1) pos++;
- refreshLine(fd,prompt,buf,len,pos,cols);
- }
- break;
- case 2: /* ctrl-b */
- goto left_arrow;
- case 6: /* ctrl-f */
- goto right_arrow;
- case 16: /* ctrl-p */
- seq[1] = 65;
- goto up_down_arrow;
- case 14: /* ctrl-n */
- seq[1] = 66;
- goto up_down_arrow;
- break;
- case 27: /* escape sequence */
- if (read(fd,seq,2) == -1) break;
- if (seq[0] == 91 && seq[1] == 68) {
-left_arrow:
- /* left arrow */
- if (pos > 0) {
- pos--;
- refreshLine(fd,prompt,buf,len,pos,cols);
- }
- } else if (seq[0] == 91 && seq[1] == 67) {
-right_arrow:
- /* right arrow */
- if (pos != len) {
- pos++;
- refreshLine(fd,prompt,buf,len,pos,cols);
- }
- } else if (seq[0] == 91 && (seq[1] == 65 || seq[1] == 66)) {
-up_down_arrow:
- /* up and down arrow: history */
- if (history_len > 1) {
- /* Update the current history entry before to
- * overwrite it with tne next one. */
- free(history[history_len-1-history_index]);
- history[history_len-1-history_index] = strdup(buf);
- /* Show the new entry */
- history_index += (seq[1] == 65) ? 1 : -1;
- if (history_index < 0) {
- history_index = 0;
- break;
- } else if (history_index >= history_len) {
- history_index = history_len-1;
- break;
- }
- strncpy(buf,history[history_len-1-history_index],buflen);
- buf[buflen] = '\0';
- len = pos = strlen(buf);
- refreshLine(fd,prompt,buf,len,pos,cols);
- }
- }
- break;
- default:
- if (len < buflen) {
- if (len == pos) {
- buf[pos] = c;
- pos++;
- len++;
- buf[len] = '\0';
- if (plen+len < cols) {
- /* Avoid a full update of the line in the
- * trivial case. */
- if (write(fd,&c,1) == -1) return -1;
- } else {
- refreshLine(fd,prompt,buf,len,pos,cols);
- }
- } else {
- memmove(buf+pos+1,buf+pos,len-pos);
- buf[pos] = c;
- len++;
- pos++;
- buf[len] = '\0';
- refreshLine(fd,prompt,buf,len,pos,cols);
- }
- }
- break;
- case 21: /* Ctrl+u, delete the whole line. */
- buf[0] = '\0';
- pos = len = 0;
- refreshLine(fd,prompt,buf,len,pos,cols);
- break;
- case 11: /* Ctrl+k, delete from current to end of line. */
- buf[pos] = '\0';
- len = pos;
- refreshLine(fd,prompt,buf,len,pos,cols);
- break;
- case 1: /* Ctrl+a, go to the start of the line */
- pos = 0;
- refreshLine(fd,prompt,buf,len,pos,cols);
- break;
- case 5: /* ctrl+e, go to the end of the line */
- pos = len;
- refreshLine(fd,prompt,buf,len,pos,cols);
- break;
- }
- }
- return len;
-}
-
-static int linenoiseRaw(char *buf, size_t buflen, const char *prompt) {
- int fd = STDIN_FILENO;
- int count;
-
- if (buflen == 0) {
- errno = EINVAL;
- return -1;
- }
- if (!isatty(STDIN_FILENO)) {
- if (fgets(buf, buflen, stdin) == NULL) return -1;
- count = strlen(buf);
- if (count && buf[count-1] == '\n') {
- count--;
- buf[count] = '\0';
- }
- } else {
- if (enableRawMode(fd) == -1) return -1;
- count = linenoisePrompt(fd, buf, buflen, prompt);
- disableRawMode(fd);
- printf("\n");
- }
- return count;
-}
-
-char *linenoise(const char *prompt) {
- char buf[LINENOISE_MAX_LINE];
- int count;
-
- if (isUnsupportedTerm()) {
- size_t len;
-
- printf("%s",prompt);
- fflush(stdout);
- if (fgets(buf,LINENOISE_MAX_LINE,stdin) == NULL) return NULL;
- len = strlen(buf);
- while(len && (buf[len-1] == '\n' || buf[len-1] == '\r')) {
- len--;
- buf[len] = '\0';
- }
- return strdup(buf);
- } else {
- count = linenoiseRaw(buf,LINENOISE_MAX_LINE,prompt);
- if (count == -1) return NULL;
- return strdup(buf);
- }
-}
-
-/* Using a circular buffer is smarter, but a bit more complex to handle. */
-int linenoiseHistoryAdd(const char *line) {
- char *linecopy;
-
- if (history_max_len == 0) return 0;
- if (history == NULL) {
- history = malloc(sizeof(char*)*history_max_len);
- if (history == NULL) return 0;
- memset(history,0,(sizeof(char*)*history_max_len));
- }
- linecopy = strdup(line);
- if (!linecopy) return 0;
- if (history_len == history_max_len) {
- free(history[0]);
- memmove(history,history+1,sizeof(char*)*(history_max_len-1));
- history_len--;
- }
- history[history_len] = linecopy;
- history_len++;
- return 1;
-}
-
-int linenoiseHistorySetMaxLen(int len) {
- char **new;
-
- if (len < 1) return 0;
- if (history) {
- int tocopy = history_len;
-
- new = malloc(sizeof(char*)*len);
- if (new == NULL) return 0;
- if (len < tocopy) tocopy = len;
- memcpy(new,history+(history_max_len-tocopy), sizeof(char*)*tocopy);
- free(history);
- history = new;
- }
- history_max_len = len;
- if (history_len > history_max_len)
- history_len = history_max_len;
- return 1;
-}
-
-/* Save the history in the specified file. On success 0 is returned
- * otherwise -1 is returned. */
-int linenoiseHistorySave(char *filename) {
- FILE *fp = fopen(filename,"w");
- int j;
-
- if (fp == NULL) return -1;
- for (j = 0; j < history_len; j++)
- fprintf(fp,"%s\n",history[j]);
- fclose(fp);
- return 0;
-}
-
-/* Load the history from the specified file. If the file does not exist
- * zero is returned and no operation is performed.
- *
- * If the file exists and the operation succeeded 0 is returned, otherwise
- * on error -1 is returned. */
-int linenoiseHistoryLoad(char *filename) {
- FILE *fp = fopen(filename,"r");
- char buf[LINENOISE_MAX_LINE];
-
- if (fp == NULL) return -1;
-
- while (fgets(buf,LINENOISE_MAX_LINE,fp) != NULL) {
- char *p;
-
- p = strchr(buf,'\r');
- if (!p) p = strchr(buf,'\n');
- if (p) *p = '\0';
- linenoiseHistoryAdd(buf);
- }
- fclose(fp);
- return 0;
-}
+++ /dev/null
-/* linenoise.h -- guerrilla line editing library against the idea that a
- * line editing lib needs to be 20,000 lines of C code.
- *
- * See linenoise.c for more information.
- *
- * Copyright (c) 2010, Salvatore Sanfilippo <antirez at gmail dot com>
- * 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 __LINENOISE_H
-#define __LINENOISE_H
-
-char *linenoise(const char *prompt);
-int linenoiseHistoryAdd(const char *line);
-int linenoiseHistorySetMaxLen(int len);
-int linenoiseHistorySave(char *filename);
-int linenoiseHistoryLoad(char *filename);
-
-#endif /* __LINENOISE_H */
/* Send a MULTI command to all the slaves and AOF file. Check the execCommand
* implememntation for more information. */
void execCommandReplicateMulti(redisClient *c) {
- struct redisCommand *cmd;
robj *multistring = createStringObject("MULTI",5);
- cmd = lookupCommand("multi");
if (server.appendonly)
- feedAppendOnlyFile(cmd,c->db->id,&multistring,1);
+ feedAppendOnlyFile(server.multiCommand,c->db->id,&multistring,1);
if (listLength(server.slaves))
replicationFeedSlaves(server.slaves,c->db->id,&multistring,1);
decrRefCount(multistring);
}
}
-void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
- int cport, cfd;
- char cip[128];
+static void acceptCommonHandler(int fd) {
redisClient *c;
- REDIS_NOTUSED(el);
- REDIS_NOTUSED(mask);
- REDIS_NOTUSED(privdata);
-
- cfd = anetAccept(server.neterr, fd, cip, &cport);
- if (cfd == AE_ERR) {
- redisLog(REDIS_VERBOSE,"Accepting client connection: %s", server.neterr);
- return;
- }
- redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport);
- if ((c = createClient(cfd)) == NULL) {
+ if ((c = createClient(fd)) == NULL) {
redisLog(REDIS_WARNING,"Error allocating resoures for the client");
- close(cfd); /* May be already closed, just ingore errors */
+ close(fd); /* May be already closed, just ingore errors */
return;
}
/* If maxclient directive is set and this is one client more... close the
server.stat_numconnections++;
}
+void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
+ int cport, cfd;
+ char cip[128];
+ REDIS_NOTUSED(el);
+ REDIS_NOTUSED(mask);
+ REDIS_NOTUSED(privdata);
+
+ cfd = anetTcpAccept(server.neterr, fd, cip, &cport);
+ if (cfd == AE_ERR) {
+ redisLog(REDIS_VERBOSE,"Accepting client connection: %s", server.neterr);
+ return;
+ }
+ redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport);
+ acceptCommonHandler(cfd);
+}
+
+void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
+ int cfd;
+ REDIS_NOTUSED(el);
+ REDIS_NOTUSED(mask);
+ REDIS_NOTUSED(privdata);
+
+ cfd = anetUnixAccept(server.neterr, fd);
+ if (cfd == AE_ERR) {
+ redisLog(REDIS_VERBOSE,"Accepting client connection: %s", server.neterr);
+ return;
+ }
+ redisLog(REDIS_VERBOSE,"Accepted connection to %s", server.unixsocket);
+ acceptCommonHandler(cfd);
+}
+
+
static void freeClientArgv(redisClient *c) {
int j;
for (j = 0; j < c->argc; j++)
/* Case 2: we lost the connection with the master. */
if (c->flags & REDIS_MASTER) {
server.master = NULL;
+ /* FIXME */
server.replstate = REDIS_REPL_CONNECT;
/* Since we lost the connection with the master, we should also
* close the connection with all our slaves if we have any, so
#include <math.h>
robj *createObject(int type, void *ptr) {
- robj *o;
-
- if (server.vm_enabled) pthread_mutex_lock(&server.obj_freelist_mutex);
- if (listLength(server.objfreelist)) {
- listNode *head = listFirst(server.objfreelist);
- o = listNodeValue(head);
- listDelNode(server.objfreelist,head);
- if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex);
- } else {
- if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex);
- o = zmalloc(sizeof(*o));
- }
+ robj *o = zmalloc(sizeof(*o));
o->type = type;
o->encoding = REDIS_ENCODING_RAW;
o->ptr = ptr;
o->refcount = 1;
+
/* Set the LRU to the current lruclock (minutes resolution).
* We do this regardless of the fact VM is active as LRU is also
* used for the maxmemory directive when Redis is used as cache.
default: redisPanic("Unknown object type"); break;
}
o->ptr = NULL; /* defensive programming. We'll see NULL in traces. */
- if (server.vm_enabled) pthread_mutex_lock(&server.obj_freelist_mutex);
- if (listLength(server.objfreelist) > REDIS_OBJFREELIST_MAX ||
- !listAddNodeHead(server.objfreelist,o))
- zfree(o);
- if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex);
+ zfree(o);
}
}
#include <sys/resource.h>
#include <sys/wait.h>
#include <arpa/inet.h>
+#include <sys/stat.h>
int rdbSaveType(FILE *fp, unsigned char type) {
if (fwrite(&type,1,1,fp) == 0) return -1;
if ((childpid = fork()) == 0) {
/* Child */
if (server.vm_enabled) vmReopenSwapFile();
- close(server.fd);
+ if (server.ipfd > 0) close(server.ipfd);
+ if (server.sofd > 0) close(server.sofd);
if (rdbSave(filename) == REDIS_OK) {
_exit(0);
} else {
return o;
}
+/* Mark that we are loading in the global state and setup the fields
+ * needed to provide loading stats. */
+void startLoading(FILE *fp) {
+ struct stat sb;
+
+ /* Load the DB */
+ server.loading = 1;
+ server.loading_start_time = time(NULL);
+ if (fstat(fileno(fp), &sb) == -1) {
+ server.loading_total_bytes = 1; /* just to avoid division by zero */
+ } else {
+ server.loading_total_bytes = sb.st_size;
+ }
+}
+
+/* Refresh the loading progress info */
+void loadingProgress(off_t pos) {
+ server.loading_loaded_bytes = pos;
+}
+
+/* Loading finished */
+void stopLoading(void) {
+ server.loading = 0;
+}
+
int rdbLoad(char *filename) {
FILE *fp;
uint32_t dbid;
redisDb *db = server.db+0;
char buf[1024];
time_t expiretime, now = time(NULL);
+ long loops = 0;
fp = fopen(filename,"r");
if (!fp) return REDIS_ERR;
redisLog(REDIS_WARNING,"Can't handle RDB format version %d",rdbver);
return REDIS_ERR;
}
+
+ startLoading(fp);
while(1) {
robj *key, *val;
int force_swapout;
expiretime = -1;
+
+ /* Serve the clients from time to time */
+ if (!(loops++ % 1000)) {
+ loadingProgress(ftello(fp));
+ aeProcessEvents(server.el, AE_FILE_EVENTS|AE_DONT_WAIT);
+ }
+
/* Read type. */
if ((type = rdbLoadType(fp)) == -1) goto eoferr;
if (type == REDIS_EXPIRETIME) {
}
}
fclose(fp);
+ stopLoading();
return REDIS_OK;
eoferr: /* unexpected end of file is handled here with a fatal exit */
#include <assert.h>
#include "ae.h"
-#include "anet.h"
+#include "hiredis.h"
#include "sds.h"
#include "adlist.h"
#include "zmalloc.h"
-#define REPLY_INT 0
-#define REPLY_RETCODE 1
-#define REPLY_BULK 2
-#define REPLY_MBULK 3
-
#define CLIENT_CONNECTING 0
#define CLIENT_SENDQUERY 1
#define CLIENT_READREPLY 2
-#define MAX_LATENCY 5000
-
#define REDIS_NOTUSED(V) ((void) V)
static struct config {
aeEventLoop *el;
char *hostip;
int hostport;
+ char *hostsocket;
int keepalive;
long long start;
long long totlatency;
- int *latency;
+ long long *latency;
char *title;
list *clients;
int quiet;
} config;
typedef struct _client {
+ redisContext *context;
int state;
- int fd;
sds obuf;
- sds ibuf;
- int mbulk; /* Number of elements in an mbulk reply */
- int readlen; /* readlen == -1 means read a single line */
- int totreceived;
- unsigned int written; /* bytes of 'obuf' already written */
+ unsigned int written; /* bytes of 'obuf' already written */
int replytype;
- long long start; /* start time in milliseconds */
+ long long start; /* start time of a request */
+ long long latency; /* request latency */
} *client;
/* Prototypes */
static void createMissingClients(client c);
/* Implementation */
+static long long ustime(void) {
+ struct timeval tv;
+ long long ust;
+
+ gettimeofday(&tv, NULL);
+ ust = ((long)tv.tv_sec)*1000000;
+ ust += tv.tv_usec;
+ return ust;
+}
+
static long long mstime(void) {
struct timeval tv;
long long mst;
static void freeClient(client c) {
listNode *ln;
-
- aeDeleteFileEvent(config.el,c->fd,AE_WRITABLE);
- aeDeleteFileEvent(config.el,c->fd,AE_READABLE);
- sdsfree(c->ibuf);
+ aeDeleteFileEvent(config.el,c->context->fd,AE_WRITABLE);
+ aeDeleteFileEvent(config.el,c->context->fd,AE_READABLE);
+ redisFree(c->context);
sdsfree(c->obuf);
- close(c->fd);
zfree(c);
config.liveclients--;
ln = listSearchKey(config.clients,c);
}
static void resetClient(client c) {
- aeDeleteFileEvent(config.el,c->fd,AE_WRITABLE);
- aeDeleteFileEvent(config.el,c->fd,AE_READABLE);
- aeCreateFileEvent(config.el,c->fd, AE_WRITABLE,writeHandler,c);
- sdsfree(c->ibuf);
- c->ibuf = sdsempty();
- c->readlen = (c->replytype == REPLY_BULK ||
- c->replytype == REPLY_MBULK) ? -1 : 0;
- c->mbulk = -1;
+ aeDeleteFileEvent(config.el,c->context->fd,AE_WRITABLE);
+ aeDeleteFileEvent(config.el,c->context->fd,AE_READABLE);
+ aeCreateFileEvent(config.el,c->context->fd,AE_WRITABLE,writeHandler,c);
c->written = 0;
- c->totreceived = 0;
c->state = CLIENT_SENDQUERY;
- c->start = mstime();
- createMissingClients(c);
+ c->start = ustime();
+ c->latency = -1;
}
static void randomizeClientKey(client c) {
memcpy(p,buf,strlen(buf));
}
-static void prepareClientForReply(client c, int type) {
- if (type == REPLY_BULK) {
- c->replytype = REPLY_BULK;
- c->readlen = -1;
- } else if (type == REPLY_MBULK) {
- c->replytype = REPLY_MBULK;
- c->readlen = -1;
- c->mbulk = -1;
- } else {
- c->replytype = type;
- c->readlen = 0;
- }
-}
-
static void clientDone(client c) {
- static int last_tot_received = 1;
-
- long long latency;
- config.donerequests ++;
- latency = mstime() - c->start;
- if (latency > MAX_LATENCY) latency = MAX_LATENCY;
- config.latency[latency]++;
-
- if (config.debug && last_tot_received != c->totreceived) {
- printf("Tot bytes received: %d\n", c->totreceived);
- last_tot_received = c->totreceived;
- }
if (config.donerequests == config.requests) {
freeClient(c);
aeStop(config.el);
}
}
-/* Read a length from the buffer pointed to by *p, store the length in *len,
- * and return the number of bytes that the cursor advanced. */
-static int readLen(char *p, int *len) {
- char *tail = strstr(p,"\r\n");
- if (tail == NULL)
- return 0;
- *tail = '\0';
- *len = atoi(p+1);
- return tail+2-p;
-}
-
-static void readHandler(aeEventLoop *el, int fd, void *privdata, int mask)
-{
- char buf[1024], *p;
- int nread, pos=0, len=0;
+static void readHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
client c = privdata;
+ void *reply = NULL;
REDIS_NOTUSED(el);
REDIS_NOTUSED(fd);
REDIS_NOTUSED(mask);
- nread = read(c->fd,buf,sizeof(buf));
- if (nread == -1) {
- fprintf(stderr, "Reading from socket: %s\n", strerror(errno));
- freeClient(c);
- return;
- }
- if (nread == 0) {
- fprintf(stderr, "EOF from client\n");
- freeClient(c);
- return;
- }
- c->totreceived += nread;
- c->ibuf = sdscatlen(c->ibuf,buf,nread);
- len = sdslen(c->ibuf);
-
- if (c->replytype == REPLY_INT ||
- c->replytype == REPLY_RETCODE)
- {
- /* Check if the first line is complete. This is everything we need
- * when waiting for an integer or status code reply.*/
- if ((p = strstr(c->ibuf,"\r\n")) != NULL)
- goto done;
- } else if (c->replytype == REPLY_BULK) {
- int advance = 0;
- if (c->readlen < 0) {
- advance = readLen(c->ibuf+pos,&c->readlen);
- if (advance) {
- pos += advance;
- if (c->readlen == -1) {
- goto done;
- } else {
- /* include the trailing \r\n */
- c->readlen += 2;
- }
- } else {
- goto skip;
- }
- }
-
- int canconsume;
- if (c->readlen > 0) {
- canconsume = c->readlen > (len-pos) ? (len-pos) : c->readlen;
- c->readlen -= canconsume;
- pos += canconsume;
- }
+ /* Calculate latency only for the first read event. This means that the
+ * server already sent the reply and we need to parse it. Parsing overhead
+ * is not part of the latency, so calculate it only once, here. */
+ if (c->latency < 0) c->latency = ustime()-(c->start);
- if (c->readlen == 0)
- goto done;
- } else if (c->replytype == REPLY_MBULK) {
- int advance = 0;
- if (c->mbulk == -1) {
- advance = readLen(c->ibuf+pos,&c->mbulk);
- if (advance) {
- pos += advance;
- if (c->mbulk == -1)
- goto done;
- } else {
- goto skip;
- }
+ if (redisBufferRead(c->context) != REDIS_OK) {
+ fprintf(stderr,"Error: %s\n",c->context->errstr);
+ exit(1);
+ } else {
+ if (redisGetReply(c->context,&reply) != REDIS_OK) {
+ fprintf(stderr,"Error: %s\n",c->context->errstr);
+ exit(1);
}
-
- int canconsume;
- while(c->mbulk > 0 && pos < len) {
- if (c->readlen > 0) {
- canconsume = c->readlen > (len-pos) ? (len-pos) : c->readlen;
- c->readlen -= canconsume;
- pos += canconsume;
- if (c->readlen == 0)
- c->mbulk--;
- } else {
- advance = readLen(c->ibuf+pos,&c->readlen);
- if (advance) {
- pos += advance;
- if (c->readlen == -1) {
- c->mbulk--;
- continue;
- } else {
- /* include the trailing \r\n */
- c->readlen += 2;
- }
- } else {
- goto skip;
- }
- }
+ if (reply != NULL) {
+ if (config.donerequests < config.requests)
+ config.latency[config.donerequests++] = c->latency;
+ clientDone(c);
}
-
- if (c->mbulk == 0)
- goto done;
}
-
-skip:
- c->ibuf = sdsrange(c->ibuf,pos,-1);
- return;
-done:
- clientDone(c);
- return;
}
-static void writeHandler(aeEventLoop *el, int fd, void *privdata, int mask)
-{
+static void writeHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
client c = privdata;
REDIS_NOTUSED(el);
REDIS_NOTUSED(fd);
if (c->state == CLIENT_CONNECTING) {
c->state = CLIENT_SENDQUERY;
- c->start = mstime();
+ c->start = ustime();
+ c->latency = -1;
}
if (sdslen(c->obuf) > c->written) {
void *ptr = c->obuf+c->written;
- int len = sdslen(c->obuf) - c->written;
- int nwritten = write(c->fd, ptr, len);
+ int nwritten = write(c->context->fd,ptr,sdslen(c->obuf)-c->written);
if (nwritten == -1) {
if (errno != EPIPE)
fprintf(stderr, "Writing to socket: %s\n", strerror(errno));
}
c->written += nwritten;
if (sdslen(c->obuf) == c->written) {
- aeDeleteFileEvent(config.el,c->fd,AE_WRITABLE);
- aeCreateFileEvent(config.el,c->fd,AE_READABLE,readHandler,c);
+ aeDeleteFileEvent(config.el,c->context->fd,AE_WRITABLE);
+ aeCreateFileEvent(config.el,c->context->fd,AE_READABLE,readHandler,c);
c->state = CLIENT_READREPLY;
}
}
}
-static client createClient(void) {
+static client createClient(int replytype) {
client c = zmalloc(sizeof(struct _client));
- char err[ANET_ERR_LEN];
-
- c->fd = anetTcpNonBlockConnect(err,config.hostip,config.hostport);
- if (c->fd == ANET_ERR) {
- zfree(c);
- fprintf(stderr,"Connect: %s\n",err);
- return NULL;
+ if (config.hostsocket == NULL) {
+ c->context = redisConnectNonBlock(config.hostip,config.hostport);
+ } else {
+ c->context = redisConnectUnixNonBlock(config.hostsocket);
}
- anetTcpNoDelay(NULL,c->fd);
+ if (c->context->err) {
+ fprintf(stderr,"Could not connect to Redis at ");
+ if (config.hostsocket == NULL)
+ fprintf(stderr,"%s:%d: %s\n",config.hostip,config.hostport,c->context->errstr);
+ else
+ fprintf(stderr,"%s: %s\n",config.hostsocket,c->context->errstr);
+ exit(1);
+ }
+ c->replytype = replytype;
+ c->state = CLIENT_CONNECTING;
c->obuf = sdsempty();
- c->ibuf = sdsempty();
- c->mbulk = -1;
- c->readlen = 0;
c->written = 0;
- c->totreceived = 0;
- c->state = CLIENT_CONNECTING;
- aeCreateFileEvent(config.el, c->fd, AE_WRITABLE, writeHandler, c);
- config.liveclients++;
+ redisSetReplyObjectFunctions(c->context,NULL);
+ aeCreateFileEvent(config.el,c->context->fd,AE_WRITABLE,writeHandler,c);
listAddNodeTail(config.clients,c);
+ config.liveclients++;
return c;
}
static void createMissingClients(client c) {
while(config.liveclients < config.numclients) {
- client new = createClient();
- if (!new) continue;
+ client new = createClient(c->replytype);
sdsfree(new->obuf);
new->obuf = sdsdup(c->obuf);
if (config.randomkeys) randomizeClientKey(c);
- prepareClientForReply(new,c->replytype);
}
}
+static int compareLatency(const void *a, const void *b) {
+ return (*(long long*)a)-(*(long long*)b);
+}
+
static void showLatencyReport(void) {
- int j, seen = 0;
+ int i, curlat = 0;
float perc, reqpersec;
reqpersec = (float)config.donerequests/((float)config.totlatency/1000);
printf(" %d bytes payload\n", config.datasize);
printf(" keep alive: %d\n", config.keepalive);
printf("\n");
- for (j = 0; j <= MAX_LATENCY; j++) {
- if (config.latency[j]) {
- seen += config.latency[j];
- perc = ((float)seen*100)/config.donerequests;
- printf("%.2f%% <= %d milliseconds\n", perc, j);
+
+ qsort(config.latency,config.requests,sizeof(long long),compareLatency);
+ for (i = 0; i < config.requests; i++) {
+ if (config.latency[i]/1000 != curlat || i == (config.requests-1)) {
+ curlat = config.latency[i]/1000;
+ perc = ((float)(i+1)*100)/config.requests;
+ printf("%.2f%% <= %d milliseconds\n", perc, curlat);
}
}
printf("%.2f requests per second\n\n", reqpersec);
}
static void prepareForBenchmark(char *title) {
- memset(config.latency,0,sizeof(int)*(MAX_LATENCY+1));
config.title = title;
config.start = mstime();
config.donerequests = 0;
config.keepalive = atoi(argv[i+1]);
i++;
} else if (!strcmp(argv[i],"-h") && !lastarg) {
- char *ip = zmalloc(32);
- if (anetResolve(NULL,argv[i+1],ip) == ANET_ERR) {
- printf("Can't resolve %s\n", argv[i]);
- exit(1);
- }
- config.hostip = ip;
+ config.hostip = argv[i+1];
i++;
} else if (!strcmp(argv[i],"-p") && !lastarg) {
config.hostport = atoi(argv[i+1]);
i++;
+ } else if (!strcmp(argv[i],"-s") && !lastarg) {
+ config.hostsocket = argv[i+1];
+ i++;
} else if (!strcmp(argv[i],"-d") && !lastarg) {
config.datasize = atoi(argv[i+1]);
i++;
printf("Wrong option '%s' or option argument missing\n\n",argv[i]);
printf("Usage: redis-benchmark [-h <host>] [-p <port>] [-c <clients>] [-n <requests]> [-k <boolean>]\n\n");
printf(" -h <hostname> Server hostname (default 127.0.0.1)\n");
- printf(" -p <hostname> Server port (default 6379)\n");
+ printf(" -p <port> Server port (default 6379)\n");
+ printf(" -s <socket> Server socket (overrides host and port)\n");
printf(" -c <clients> Number of parallel connections (default 50)\n");
printf(" -n <requests> Total number of requests (default 10000)\n");
printf(" -d <size> Data size of SET/GET value in bytes (default 2)\n");
config.idlemode = 0;
config.latency = NULL;
config.clients = listCreate();
- config.latency = zmalloc(sizeof(int)*(MAX_LATENCY+1));
-
config.hostip = "127.0.0.1";
config.hostport = 6379;
+ config.hostsocket = NULL;
parseOptions(argc,argv);
+ config.latency = zmalloc(sizeof(long long)*config.requests);
if (config.keepalive == 0) {
printf("WARNING: keepalive disabled, you probably need 'echo 1 > /proc/sys/net/ipv4/tcp_tw_reuse' for Linux and 'sudo sysctl -w net.inet.tcp.msl=1000' for Mac OS X in order to use a lot of clients/requests\n");
if (config.idlemode) {
printf("Creating %d idle connections and waiting forever (Ctrl+C when done)\n", config.numclients);
prepareForBenchmark("IDLE");
- c = createClient();
- if (!c) exit(1);
+ c = createClient(0); /* will never receive a reply */
c->obuf = sdsempty();
- prepareClientForReply(c,REPLY_RETCODE); /* will never receive it */
createMissingClients(c);
aeMain(config.el);
/* and will wait for every */
do {
prepareForBenchmark("PING");
- c = createClient();
- if (!c) exit(1);
+ c = createClient(REDIS_REPLY_STATUS);
c->obuf = sdscat(c->obuf,"PING\r\n");
- prepareClientForReply(c,REPLY_RETCODE);
createMissingClients(c);
aeMain(config.el);
endBenchmark();
prepareForBenchmark("PING (multi bulk)");
- c = createClient();
- if (!c) exit(1);
+ c = createClient(REDIS_REPLY_STATUS);
c->obuf = sdscat(c->obuf,"*1\r\n$4\r\nPING\r\n");
- prepareClientForReply(c,REPLY_RETCODE);
createMissingClients(c);
aeMain(config.el);
endBenchmark();
prepareForBenchmark("MSET (10 keys, multi bulk)");
- c = createClient();
- if (!c) exit(1);
+ c = createClient(REDIS_REPLY_ARRAY);
c->obuf = sdscatprintf(c->obuf,"*%d\r\n$4\r\nMSET\r\n", 11);
{
int i;
}
zfree(data);
}
- prepareClientForReply(c,REPLY_RETCODE);
createMissingClients(c);
aeMain(config.el);
endBenchmark();
prepareForBenchmark("SET");
- c = createClient();
- if (!c) exit(1);
- c->obuf = sdscat(c->obuf,"SET foo_rand000000000000 ");
+ c = createClient(REDIS_REPLY_STATUS);
+ c->obuf = sdscat(c->obuf,"*3\r\n$3\r\nSET\r\n$20\r\nfoo_rand000000000000\r\n");
{
char *data = zmalloc(config.datasize+2);
memset(data,'x',config.datasize);
data[config.datasize] = '\r';
data[config.datasize+1] = '\n';
+ c->obuf = sdscatprintf(c->obuf,"$%d\r\n",config.datasize);
c->obuf = sdscatlen(c->obuf,data,config.datasize+2);
}
- prepareClientForReply(c,REPLY_RETCODE);
createMissingClients(c);
aeMain(config.el);
endBenchmark();
prepareForBenchmark("GET");
- c = createClient();
- if (!c) exit(1);
+ c = createClient(REDIS_REPLY_STRING);
c->obuf = sdscat(c->obuf,"GET foo_rand000000000000\r\n");
- prepareClientForReply(c,REPLY_BULK);
createMissingClients(c);
aeMain(config.el);
endBenchmark();
prepareForBenchmark("INCR");
- c = createClient();
- if (!c) exit(1);
+ c = createClient(REDIS_REPLY_INTEGER);
c->obuf = sdscat(c->obuf,"INCR counter_rand000000000000\r\n");
- prepareClientForReply(c,REPLY_INT);
createMissingClients(c);
aeMain(config.el);
endBenchmark();
prepareForBenchmark("LPUSH");
- c = createClient();
- if (!c) exit(1);
+ c = createClient(REDIS_REPLY_INTEGER);
c->obuf = sdscat(c->obuf,"LPUSH mylist bar\r\n");
- prepareClientForReply(c,REPLY_INT);
createMissingClients(c);
aeMain(config.el);
endBenchmark();
prepareForBenchmark("LPOP");
- c = createClient();
- if (!c) exit(1);
+ c = createClient(REDIS_REPLY_STRING);
c->obuf = sdscat(c->obuf,"LPOP mylist\r\n");
- prepareClientForReply(c,REPLY_BULK);
createMissingClients(c);
aeMain(config.el);
endBenchmark();
prepareForBenchmark("SADD");
- c = createClient();
- if (!c) exit(1);
+ c = createClient(REDIS_REPLY_STATUS);
c->obuf = sdscat(c->obuf,"SADD myset counter_rand000000000000\r\n");
- prepareClientForReply(c,REPLY_RETCODE);
createMissingClients(c);
aeMain(config.el);
endBenchmark();
prepareForBenchmark("SPOP");
- c = createClient();
- if (!c) exit(1);
+ c = createClient(REDIS_REPLY_STRING);
c->obuf = sdscat(c->obuf,"SPOP myset\r\n");
- prepareClientForReply(c,REPLY_BULK);
createMissingClients(c);
aeMain(config.el);
endBenchmark();
prepareForBenchmark("LPUSH (again, in order to bench LRANGE)");
- c = createClient();
- if (!c) exit(1);
+ c = createClient(REDIS_REPLY_STATUS);
c->obuf = sdscat(c->obuf,"LPUSH mylist bar\r\n");
- prepareClientForReply(c,REPLY_RETCODE);
createMissingClients(c);
aeMain(config.el);
endBenchmark();
prepareForBenchmark("LRANGE (first 100 elements)");
- c = createClient();
- if (!c) exit(1);
+ c = createClient(REDIS_REPLY_ARRAY);
c->obuf = sdscat(c->obuf,"LRANGE mylist 0 99\r\n");
- prepareClientForReply(c,REPLY_MBULK);
createMissingClients(c);
aeMain(config.el);
endBenchmark();
prepareForBenchmark("LRANGE (first 300 elements)");
- c = createClient();
- if (!c) exit(1);
+ c = createClient(REDIS_REPLY_ARRAY);
c->obuf = sdscat(c->obuf,"LRANGE mylist 0 299\r\n");
- prepareClientForReply(c,REPLY_MBULK);
createMissingClients(c);
aeMain(config.el);
endBenchmark();
prepareForBenchmark("LRANGE (first 450 elements)");
- c = createClient();
- if (!c) exit(1);
+ c = createClient(REDIS_REPLY_ARRAY);
c->obuf = sdscat(c->obuf,"LRANGE mylist 0 449\r\n");
- prepareClientForReply(c,REPLY_MBULK);
createMissingClients(c);
aeMain(config.el);
endBenchmark();
prepareForBenchmark("LRANGE (first 600 elements)");
- c = createClient();
- if (!c) exit(1);
+ c = createClient(REDIS_REPLY_ARRAY);
c->obuf = sdscat(c->obuf,"LRANGE mylist 0 599\r\n");
- prepareClientForReply(c,REPLY_MBULK);
createMissingClients(c);
aeMain(config.el);
endBenchmark();
/* display error stack */
for (i = 0; i < errors.level; i++) {
- printf("0x%08lx - %s\n", errors.offset[i], errors.error[i]);
+ printf("0x%08lx - %s\n",
+ (unsigned long) errors.offset[i], errors.error[i]);
}
}
#include <ctype.h>
#include <errno.h>
#include <sys/stat.h>
+#include <sys/time.h>
-#include "anet.h"
+#include "hiredis.h"
#include "sds.h"
-#include "adlist.h"
#include "zmalloc.h"
#include "linenoise.h"
#include "help.h"
#define REDIS_NOTUSED(V) ((void) V)
+static redisContext *context;
static struct config {
char *hostip;
int hostport;
+ char *hostsocket;
long repeat;
int dbnum;
int interactive;
char *historyfile;
} config;
-static int cliReadReply(int fd);
static void usage();
+char *redisGitSHA1(void);
-/* Connect to the client. If force is not zero the connection is performed
- * even if there is already a connected socket. */
-static int cliConnect(int force) {
- char err[ANET_ERR_LEN];
- static int fd = ANET_ERR;
-
- if (fd == ANET_ERR || force) {
- if (force) close(fd);
- fd = anetTcpConnect(err,config.hostip,config.hostport);
- if (fd == ANET_ERR) {
- fprintf(stderr, "Could not connect to Redis at %s:%d: %s", config.hostip, config.hostport, err);
- return -1;
- }
- anetTcpNoDelay(NULL,fd);
- }
- return fd;
-}
+/*------------------------------------------------------------------------------
+ * Utility functions
+ *--------------------------------------------------------------------------- */
-static sds cliReadLine(int fd) {
- sds line = sdsempty();
+static long long mstime(void) {
+ struct timeval tv;
+ long long mst;
- while(1) {
- char c;
- ssize_t ret;
-
- ret = read(fd,&c,1);
- if (ret <= 0) {
- sdsfree(line);
- return NULL;
- } else if ((ret == 0) || (c == '\n')) {
- break;
- } else {
- line = sdscatlen(line,&c,1);
- }
- }
- return sdstrim(line,"\r\n");
+ gettimeofday(&tv, NULL);
+ mst = ((long)tv.tv_sec)*1000;
+ mst += tv.tv_usec/1000;
+ return mst;
}
-static int cliReadSingleLineReply(int fd, int quiet) {
- sds reply = cliReadLine(fd);
+/*------------------------------------------------------------------------------
+ * Networking / parsing
+ *--------------------------------------------------------------------------- */
- if (reply == NULL) return 1;
- if (!quiet)
- printf("%s", reply);
- sdsfree(reply);
- return 0;
-}
+/* Send AUTH command to the server */
+static int cliAuth() {
+ redisReply *reply;
+ if (config.auth == NULL) return REDIS_OK;
-static void printStringRepr(char *s, int len) {
- printf("\"");
- while(len--) {
- switch(*s) {
- case '\\':
- case '"':
- printf("\\%c",*s);
- break;
- case '\n': printf("\\n"); break;
- case '\r': printf("\\r"); break;
- case '\t': printf("\\t"); break;
- case '\a': printf("\\a"); break;
- case '\b': printf("\\b"); break;
- default:
- if (isprint(*s))
- printf("%c",*s);
- else
- printf("\\x%02x",(unsigned char)*s);
- break;
- }
- s++;
+ reply = redisCommand(context,"AUTH %s",config.auth);
+ if (reply != NULL) {
+ freeReplyObject(reply);
+ return REDIS_OK;
}
- printf("\"");
+ return REDIS_ERR;
}
-static int cliReadBulkReply(int fd) {
- sds replylen = cliReadLine(fd);
- char *reply, crlf[2];
- int bulklen;
-
- if (replylen == NULL) return 1;
- bulklen = atoi(replylen);
- if (bulklen == -1) {
- sdsfree(replylen);
- printf("(nil)\n");
- return 0;
- }
- reply = zmalloc(bulklen);
- anetRead(fd,reply,bulklen);
- anetRead(fd,crlf,2);
- if (config.raw_output || !config.tty) {
- if (bulklen && fwrite(reply,bulklen,1,stdout) == 0) {
- zfree(reply);
- return 1;
- }
- } else {
- /* If you are producing output for the standard output we want
- * a more interesting output with quoted characters and so forth */
- printStringRepr(reply,bulklen);
+/* Send SELECT dbnum to the server */
+static int cliSelect() {
+ redisReply *reply;
+ char dbnum[16];
+ if (config.dbnum == 0) return REDIS_OK;
+
+ snprintf(dbnum,sizeof(dbnum),"%d",config.dbnum);
+ reply = redisCommand(context,"SELECT %s",dbnum);
+ if (reply != NULL) {
+ freeReplyObject(reply);
+ return REDIS_OK;
}
- zfree(reply);
- return 0;
+ return REDIS_ERR;
}
-static int cliReadMultiBulkReply(int fd) {
- sds replylen = cliReadLine(fd);
- int elements, c = 1;
- int retval = 0;
+/* Connect to the client. If force is not zero the connection is performed
+ * even if there is already a connected socket. */
+static int cliConnect(int force) {
+ if (context == NULL || force) {
+ if (context != NULL)
+ redisFree(context);
- if (replylen == NULL) return 1;
- elements = atoi(replylen);
- if (elements == -1) {
- sdsfree(replylen);
- printf("(nil)\n");
- return 0;
- }
- if (elements == 0) {
- printf("(empty list or set)\n");
- }
- while(elements--) {
- if (config.tty) printf("%d. ", c);
- if (cliReadReply(fd)) retval = 1;
- if (elements) printf("%c",config.mb_sep);
- c++;
+ if (config.hostsocket == NULL) {
+ context = redisConnect(config.hostip,config.hostport);
+ } else {
+ context = redisConnectUnix(config.hostsocket);
+ }
+
+ if (context->err) {
+ fprintf(stderr,"Could not connect to Redis at ");
+ if (config.hostsocket == NULL)
+ fprintf(stderr,"%s:%d: %s\n",config.hostip,config.hostport,context->errstr);
+ else
+ fprintf(stderr,"%s: %s\n",config.hostsocket,context->errstr);
+ redisFree(context);
+ context = NULL;
+ return REDIS_ERR;
+ }
+
+ /* Do AUTH and select the right DB. */
+ if (cliAuth() != REDIS_OK)
+ return REDIS_ERR;
+ if (cliSelect() != REDIS_OK)
+ return REDIS_ERR;
}
- return retval;
+ return REDIS_OK;
}
-static int cliReadReply(int fd) {
- char type;
- int nread;
+static void cliPrintContextErrorAndExit() {
+ if (context == NULL) return;
+ fprintf(stderr,"Error: %s\n",context->errstr);
+ exit(1);
+}
- if ((nread = anetRead(fd,&type,1)) <= 0) {
- if (config.shutdown) return 0;
- if (config.interactive &&
- (nread == 0 || (nread == -1 && errno == ECONNRESET)))
- {
- return ECONNRESET;
+static sds cliFormatReply(redisReply *r, char *prefix) {
+ sds out = sdsempty();
+ switch (r->type) {
+ case REDIS_REPLY_ERROR:
+ if (config.tty) out = sdscat(out,"(error) ");
+ out = sdscatprintf(out,"%s\n", r->str);
+ break;
+ case REDIS_REPLY_STATUS:
+ out = sdscat(out,r->str);
+ out = sdscat(out,"\n");
+ break;
+ case REDIS_REPLY_INTEGER:
+ if (config.tty) out = sdscat(out,"(integer) ");
+ out = sdscatprintf(out,"%lld\n",r->integer);
+ break;
+ case REDIS_REPLY_STRING:
+ if (config.raw_output || !config.tty) {
+ out = sdscatlen(out,r->str,r->len);
} else {
- printf("I/O error while reading from socket: %s",strerror(errno));
- exit(1);
+ /* If you are producing output for the standard output we want
+ * a more interesting output with quoted characters and so forth */
+ out = sdscatrepr(out,r->str,r->len);
+ out = sdscat(out,"\n");
}
- }
- switch(type) {
- case '-':
- if (config.tty) printf("(error) ");
- cliReadSingleLineReply(fd,0);
- return 1;
- case '+':
- return cliReadSingleLineReply(fd,0);
- case ':':
- if (config.tty) printf("(integer) ");
- return cliReadSingleLineReply(fd,0);
- case '$':
- return cliReadBulkReply(fd);
- case '*':
- return cliReadMultiBulkReply(fd);
+ break;
+ case REDIS_REPLY_NIL:
+ out = sdscat(out,"(nil)\n");
+ break;
+ case REDIS_REPLY_ARRAY:
+ if (r->elements == 0) {
+ out = sdscat(out,"(empty list or set)\n");
+ } else {
+ unsigned int i, idxlen = 0;
+ char _prefixlen[16];
+ char _prefixfmt[16];
+ sds _prefix;
+ sds tmp;
+
+ /* Calculate chars needed to represent the largest index */
+ i = r->elements;
+ do {
+ idxlen++;
+ i /= 10;
+ } while(i);
+
+ /* Prefix for nested multi bulks should grow with idxlen+2 spaces */
+ memset(_prefixlen,' ',idxlen+2);
+ _prefixlen[idxlen+2] = '\0';
+ _prefix = sdscat(sdsnew(prefix),_prefixlen);
+
+ /* Setup prefix format for every entry */
+ snprintf(_prefixfmt,sizeof(_prefixfmt),"%%s%%%dd) ",idxlen);
+
+ for (i = 0; i < r->elements; i++) {
+ /* Don't use the prefix for the first element, as the parent
+ * caller already prepended the index number. */
+ out = sdscatprintf(out,_prefixfmt,i == 0 ? "" : prefix,i+1);
+
+ /* Format the multi bulk entry */
+ tmp = cliFormatReply(r->element[i],_prefix);
+ out = sdscatlen(out,tmp,sdslen(tmp));
+ sdsfree(tmp);
+ }
+ sdsfree(_prefix);
+ }
+ break;
default:
- printf("protocol error, got '%c' as reply type byte", type);
- return 1;
+ fprintf(stderr,"Unknown reply type: %d\n", r->type);
+ exit(1);
}
+ return out;
}
-static int selectDb(int fd) {
- int retval;
- sds cmd;
- char type;
-
- if (config.dbnum == 0)
- return 0;
-
- cmd = sdsempty();
- cmd = sdscatprintf(cmd,"SELECT %d\r\n",config.dbnum);
- anetWrite(fd,cmd,sdslen(cmd));
- anetRead(fd,&type,1);
- if (type <= 0 || type != '+') return 1;
- retval = cliReadSingleLineReply(fd,1);
- if (retval) {
- return retval;
+static int cliReadReply() {
+ redisReply *reply;
+ sds out;
+
+ if (redisGetReply(context,(void**)&reply) != REDIS_OK) {
+ if (config.shutdown)
+ return REDIS_OK;
+ if (config.interactive) {
+ /* Filter cases where we should reconnect */
+ if (context->err == REDIS_ERR_IO && errno == ECONNRESET)
+ return REDIS_ERR;
+ if (context->err == REDIS_ERR_EOF)
+ return REDIS_ERR;
+ }
+ cliPrintContextErrorAndExit();
+ return REDIS_ERR; /* avoid compiler warning */
}
- return 0;
+
+ out = cliFormatReply(reply,"");
+ freeReplyObject(reply);
+ fwrite(out,sdslen(out),1,stdout);
+ sdsfree(out);
+ return REDIS_OK;
}
static int cliSendCommand(int argc, char **argv, int repeat) {
char *command = argv[0];
- int fd, j, retval = 0;
- sds cmd;
+ size_t *argvlen;
+ int j;
config.raw_output = !strcasecmp(command,"info");
if (!strcasecmp(command,"help")) {
output_help(--argc, ++argv);
- return 0;
+ return REDIS_OK;
}
if (!strcasecmp(command,"shutdown")) config.shutdown = 1;
if (!strcasecmp(command,"monitor")) config.monitor_mode = 1;
if (!strcasecmp(command,"subscribe") ||
!strcasecmp(command,"psubscribe")) config.pubsub_mode = 1;
- if ((fd = cliConnect(0)) == -1) return 1;
-
- /* Select db number */
- retval = selectDb(fd);
- if (retval) {
- fprintf(stderr,"Error setting DB num\n");
- return 1;
- }
- /* Build the command to send */
- cmd = sdscatprintf(sdsempty(),"*%d\r\n",argc);
- for (j = 0; j < argc; j++) {
- cmd = sdscatprintf(cmd,"$%lu\r\n",
- (unsigned long)sdslen(argv[j]));
- cmd = sdscatlen(cmd,argv[j],sdslen(argv[j]));
- cmd = sdscatlen(cmd,"\r\n",2);
- }
+ /* Setup argument length */
+ argvlen = malloc(argc*sizeof(size_t));
+ for (j = 0; j < argc; j++)
+ argvlen[j] = sdslen(argv[j]);
while(repeat--) {
- anetWrite(fd,cmd,sdslen(cmd));
+ redisAppendCommandArgv(context,argc,(const char**)argv,argvlen);
while (config.monitor_mode) {
- if (cliReadSingleLineReply(fd,0)) exit(1);
- printf("\n");
+ if (cliReadReply() != REDIS_OK) exit(1);
+ fflush(stdout);
}
if (config.pubsub_mode) {
- printf("Reading messages... (press Ctrl-c to quit)\n");
+ printf("Reading messages... (press Ctrl-C to quit)\n");
while (1) {
- cliReadReply(fd);
- printf("\n\n");
+ if (cliReadReply() != REDIS_OK) exit(1);
}
}
- retval = cliReadReply(fd);
- if (!config.raw_output && config.tty) printf("\n");
- if (retval) return retval;
+ if (cliReadReply() != REDIS_OK)
+ return REDIS_ERR;
}
- return 0;
+ return REDIS_OK;
}
+/*------------------------------------------------------------------------------
+ * User interface
+ *--------------------------------------------------------------------------- */
+
static int parseOptions(int argc, char **argv) {
int i;
int lastarg = i==argc-1;
if (!strcmp(argv[i],"-h") && !lastarg) {
- char *ip = zmalloc(32);
- if (anetResolve(NULL,argv[i+1],ip) == ANET_ERR) {
- printf("Can't resolve %s\n", argv[i]);
- exit(1);
- }
- config.hostip = ip;
+ config.hostip = argv[i+1];
i++;
} else if (!strcmp(argv[i],"-h") && lastarg) {
usage();
} else if (!strcmp(argv[i],"-p") && !lastarg) {
config.hostport = atoi(argv[i+1]);
i++;
+ } else if (!strcmp(argv[i],"-s") && !lastarg) {
+ config.hostsocket = argv[i+1];
+ i++;
} else if (!strcmp(argv[i],"-r") && !lastarg) {
config.repeat = strtoll(argv[i+1],NULL,10);
i++;
"automatically used as last argument.\n"
);
} else if (!strcmp(argv[i],"-v")) {
- printf("redis-cli shipped with Redis version %s\n", REDIS_VERSION);
+ printf("redis-cli shipped with Redis version %s (%s)\n", REDIS_VERSION, redisGitSHA1());
exit(0);
} else {
break;
}
static void usage() {
- fprintf(stderr, "usage: redis-cli [-iv] [-h host] [-p port] [-a authpw] [-r repeat_times] [-n db_num] cmd arg1 arg2 arg3 ... argN\n");
+ fprintf(stderr, "usage: redis-cli [-iv] [-h host] [-p port] [-s /path/to/socket] [-a authpw] [-r repeat_times] [-n db_num] cmd arg1 arg2 arg3 ... argN\n");
fprintf(stderr, "usage: echo \"argN\" | redis-cli -x [options] cmd arg1 arg2 ... arg(N-1)\n\n");
fprintf(stderr, "example: cat /etc/passwd | redis-cli -x set my_passwd\n");
fprintf(stderr, "example: redis-cli get my_passwd\n");
{
exit(0);
} else {
- int err;
-
- if ((err = cliSendCommand(argc, argv, 1)) != 0) {
- if (err == ECONNRESET) {
- printf("Reconnecting... ");
- fflush(stdout);
- if (cliConnect(1) == -1) exit(1);
- printf("OK\n");
- cliSendCommand(argc,argv,1);
- }
+ long long start_time = mstime(), elapsed;
+
+ if (cliSendCommand(argc,argv,1) != REDIS_OK) {
+ printf("Reconnecting... ");
+ fflush(stdout);
+ if (cliConnect(1) != REDIS_OK) exit(1);
+ printf("OK\n");
+
+ /* If we still cannot send the command,
+ * print error and abort. */
+ if (cliSendCommand(argc,argv,1) != REDIS_OK)
+ cliPrintContextErrorAndExit();
+ }
+ elapsed = mstime()-start_time;
+ if (elapsed >= 500) {
+ printf("(%.2fs)\n",(double)elapsed/1000);
}
}
}
config.hostip = "127.0.0.1";
config.hostport = 6379;
+ config.hostsocket = NULL;
config.repeat = 1;
config.dbnum = 0;
config.interactive = 0;
argc -= firstarg;
argv += firstarg;
- if (config.auth != NULL) {
- char *authargv[2];
- int dbnum = config.dbnum;
-
- /* We need to save the real configured database number and set it to
- * zero here, otherwise cliSendCommand() will try to perform the
- * SELECT command before the authentication, and it will fail. */
- config.dbnum = 0;
- authargv[0] = "AUTH";
- authargv[1] = config.auth;
- cliSendCommand(2, convertToSds(2, authargv), 1);
- config.dbnum = dbnum; /* restore the right DB number */
- }
+ /* Try to connect */
+ if (cliConnect(0) != REDIS_OK) exit(1);
/* Start interactive mode when no command is provided */
if (argc == 0) repl();
{"zcount",zcountCommand,4,0,NULL,1,1,1},
{"zrevrange",zrevrangeCommand,-4,0,NULL,1,1,1},
{"zcard",zcardCommand,2,0,NULL,1,1,1},
- {"zscore",zscoreCommand,3,REDIS_CMD_DENYOOM,NULL,1,1,1},
+ {"zscore",zscoreCommand,3,0,NULL,1,1,1},
{"zrank",zrankCommand,3,0,NULL,1,1,1},
{"zrevrank",zrevrankCommand,3,0,NULL,1,1,1},
{"hset",hsetCommand,4,REDIS_CMD_DENYOOM,NULL,1,1,1},
return memcmp(key1, key2, l1) == 0;
}
+/* A case insensitive version used for the command lookup table. */
+int dictSdsKeyCaseCompare(void *privdata, const void *key1,
+ const void *key2)
+{
+ DICT_NOTUSED(privdata);
+
+ return strcasecmp(key1, key2) == 0;
+}
+
void dictRedisObjectDestructor(void *privdata, void *val)
{
DICT_NOTUSED(privdata);
return dictGenHashFunction((unsigned char*)key, sdslen((char*)key));
}
+unsigned int dictSdsCaseHash(const void *key) {
+ return dictGenCaseHashFunction((unsigned char*)key, sdslen((char*)key));
+}
+
int dictEncObjKeyCompare(void *privdata, const void *key1,
const void *key2)
{
NULL /* val destructor */
};
+/* Command table. sds string -> command struct pointer. */
+dictType commandTableDictType = {
+ dictSdsCaseHash, /* hash function */
+ NULL, /* key dup */
+ NULL, /* val dup */
+ dictSdsKeyCaseCompare, /* key compare */
+ dictSdsDestructor, /* key destructor */
+ NULL /* val destructor */
+};
+
/* Hash type hash table (note that small hashes are represented with zimpaps) */
dictType hashDictType = {
dictEncObjHash, /* hash function */
while (server.vm_enabled && zmalloc_used_memory() >
server.vm_max_memory)
{
- int retval;
-
- if (tryFreeOneObjectFromFreelist() == REDIS_OK) continue;
- retval = (server.vm_max_threads == 0) ?
+ int retval = (server.vm_max_threads == 0) ?
vmSwapOneObjectBlocking() :
vmSwapOneObjectThreaded();
if (retval == REDIS_ERR && !(loops % 300) &&
}
}
- /* Check if we should connect to a MASTER */
- if (server.replstate == REDIS_REPL_CONNECT && !(loops % 10)) {
- redisLog(REDIS_NOTICE,"Connecting to MASTER...");
- if (syncWithMaster() == REDIS_OK) {
- redisLog(REDIS_NOTICE,"MASTER <-> SLAVE sync succeeded");
- if (server.appendonly) rewriteAppendOnlyFileBackground();
- }
- }
+ /* Replication cron function -- used to reconnect to master and
+ * to detect transfer failures. */
+ if (!(loops % 10)) replicationCron();
+
return 100;
}
"-ERR source and destination objects are the same\r\n"));
shared.outofrangeerr = createObject(REDIS_STRING,sdsnew(
"-ERR index out of range\r\n"));
+ shared.loadingerr = createObject(REDIS_STRING,sdsnew(
+ "-LOADING Redis is loading the dataset in memory\r\n"));
shared.space = createObject(REDIS_STRING,sdsnew(" "));
shared.colon = createObject(REDIS_STRING,sdsnew(":"));
shared.plus = createObject(REDIS_STRING,sdsnew("+"));
}
void initServerConfig() {
- server.dbnum = REDIS_DEFAULT_DBNUM;
server.port = REDIS_SERVERPORT;
+ server.bindaddr = NULL;
+ server.unixsocket = NULL;
+ server.ipfd = -1;
+ server.sofd = -1;
+ server.dbnum = REDIS_DEFAULT_DBNUM;
server.verbosity = REDIS_VERBOSE;
server.maxidletime = REDIS_MAXIDLETIME;
server.saveparams = NULL;
+ server.loading = 0;
server.logfile = NULL; /* NULL = log on standard output */
- server.bindaddr = NULL;
server.glueoutputbuf = 1;
server.daemonize = 0;
server.appendonly = 0;
server.masterport = 6379;
server.master = NULL;
server.replstate = REDIS_REPL_NONE;
+ server.repl_serve_stale_data = 1;
/* Double constants initialization */
R_Zero = 0.0;
R_PosInf = 1.0/R_Zero;
R_NegInf = -1.0/R_Zero;
R_Nan = R_Zero/R_Zero;
+
+ /* Command table -- we intiialize it here as it is part of the
+ * initial configuration, since command names may be changed via
+ * redis.conf using the rename-command directive. */
+ server.commands = dictCreate(&commandTableDictType,NULL);
+ populateCommandTable();
+ server.delCommand = lookupCommandByCString("del");
+ server.multiCommand = lookupCommandByCString("multi");
}
void initServer() {
server.clients = listCreate();
server.slaves = listCreate();
server.monitors = listCreate();
- server.objfreelist = listCreate();
createSharedObjects();
server.el = aeCreateEventLoop();
server.db = zmalloc(sizeof(redisDb)*server.dbnum);
- server.fd = anetTcpServer(server.neterr, server.port, server.bindaddr);
- if (server.fd == -1) {
- redisLog(REDIS_WARNING, "Opening TCP port: %s", server.neterr);
+ server.ipfd = anetTcpServer(server.neterr,server.port,server.bindaddr);
+ if (server.ipfd == ANET_ERR) {
+ redisLog(REDIS_WARNING, "Opening port: %s", server.neterr);
+ exit(1);
+ }
+ if (server.unixsocket != NULL) {
+ unlink(server.unixsocket); /* don't care if this fails */
+ server.sofd = anetUnixServer(server.neterr,server.unixsocket);
+ if (server.sofd == ANET_ERR) {
+ redisLog(REDIS_WARNING, "Opening socket: %s", server.neterr);
+ exit(1);
+ }
+ }
+ if (server.ipfd < 0 && server.sofd < 0) {
+ redisLog(REDIS_WARNING, "Configured to not listen anywhere, exiting.");
exit(1);
}
for (j = 0; j < server.dbnum; j++) {
server.stat_keyspace_hits = 0;
server.unixtime = time(NULL);
aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL);
- if (aeCreateFileEvent(server.el, server.fd, AE_READABLE,
- acceptHandler, NULL) == AE_ERR) oom("creating file event");
+ if (server.ipfd > 0 && aeCreateFileEvent(server.el,server.ipfd,AE_READABLE,
+ acceptTcpHandler,NULL) == AE_ERR) oom("creating file event");
+ if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE,
+ acceptUnixHandler,NULL) == AE_ERR) oom("creating file event");
if (server.appendonly) {
server.appendfd = open(server.appendfilename,O_WRONLY|O_APPEND|O_CREAT,0644);
if (server.vm_enabled) vmInit();
}
-int qsortRedisCommands(const void *r1, const void *r2) {
- return strcasecmp(
- ((struct redisCommand*)r1)->name,
- ((struct redisCommand*)r2)->name);
-}
+/* Populates the Redis Command Table starting from the hard coded list
+ * we have on top of redis.c file. */
+void populateCommandTable(void) {
+ int j;
+ int numcommands = sizeof(readonlyCommandTable)/sizeof(struct redisCommand);
-void sortCommandTable() {
- /* Copy and sort the read-only version of the command table */
- commandTable = (struct redisCommand*)zmalloc(sizeof(readonlyCommandTable));
- memcpy(commandTable,readonlyCommandTable,sizeof(readonlyCommandTable));
- qsort(commandTable,
- sizeof(readonlyCommandTable)/sizeof(struct redisCommand),
- sizeof(struct redisCommand),qsortRedisCommands);
+ for (j = 0; j < numcommands; j++) {
+ struct redisCommand *c = readonlyCommandTable+j;
+ int retval;
+
+ retval = dictAdd(server.commands, sdsnew(c->name), c);
+ assert(retval == DICT_OK);
+ }
}
/* ====================== Commands lookup and execution ===================== */
-struct redisCommand *lookupCommand(char *name) {
- struct redisCommand tmp = {name,NULL,0,0,NULL,0,0,0};
- return bsearch(
- &tmp,
- commandTable,
- sizeof(readonlyCommandTable)/sizeof(struct redisCommand),
- sizeof(struct redisCommand),
- qsortRedisCommands);
+struct redisCommand *lookupCommand(sds name) {
+ return dictFetchValue(server.commands, name);
+}
+
+struct redisCommand *lookupCommandByCString(char *s) {
+ struct redisCommand *cmd;
+ sds name = sdsnew(s);
+
+ cmd = dictFetchValue(server.commands, name);
+ sdsfree(name);
+ return cmd;
}
/* Call() is the core of Redis execution of a command */
return REDIS_OK;
}
+ /* Only allow INFO and SLAVEOF when slave-serve-stale-data is no and
+ * we are a slave with a broken link with master. */
+ if (server.masterhost && server.replstate != REDIS_REPL_CONNECTED &&
+ server.repl_serve_stale_data == 0 &&
+ cmd->proc != infoCommand && cmd->proc != slaveofCommand)
+ {
+ addReplyError(c,
+ "link with MASTER is down and slave-serve-stale-data is set to no");
+ return REDIS_OK;
+ }
+
+ /* Loading DB? Return an error if the command is not INFO */
+ if (server.loading && cmd->proc != infoCommand) {
+ addReply(c, shared.loadingerr);
+ return REDIS_OK;
+ }
+
/* Exec the command */
if (c->flags & REDIS_MULTI &&
cmd->proc != execCommand && cmd->proc != discardCommand &&
"blocked_clients:%d\r\n"
"used_memory:%zu\r\n"
"used_memory_human:%s\r\n"
+ "used_memory_rss:%zu\r\n"
"mem_fragmentation_ratio:%.2f\r\n"
"use_tcmalloc:%d\r\n"
+ "loading:%d\r\n"
+ "aof_enabled:%d\r\n"
"changes_since_last_save:%lld\r\n"
"bgsave_in_progress:%d\r\n"
"last_save_time:%ld\r\n"
server.blpop_blocked_clients,
zmalloc_used_memory(),
hmem,
+ zmalloc_get_rss(),
zmalloc_get_fragmentation_ratio(),
#ifdef USE_TCMALLOC
1,
#else
0,
#endif
+ server.loading,
+ server.appendonly,
server.dirty,
server.bgsavechildpid != -1,
server.lastsave,
"master_port:%d\r\n"
"master_link_status:%s\r\n"
"master_last_io_seconds_ago:%d\r\n"
+ "master_sync_in_progress:%d\r\n"
,server.masterhost,
server.masterport,
(server.replstate == REDIS_REPL_CONNECTED) ?
"up" : "down",
- server.master ? ((int)(time(NULL)-server.master->lastinteraction)) : -1
+ server.master ? ((int)(time(NULL)-server.master->lastinteraction)) : -1,
+ server.replstate == REDIS_REPL_TRANSFER
);
+
+ if (server.replstate == REDIS_REPL_TRANSFER) {
+ info = sdscatprintf(info,
+ "master_sync_left_bytes:%ld\r\n"
+ "master_sync_last_io_seconds_ago:%d\r\n"
+ ,(long)server.repl_transfer_left,
+ (int)(time(NULL)-server.repl_transfer_lastio)
+ );
+ }
}
if (server.vm_enabled) {
lockThreadedIO();
);
unlockThreadedIO();
}
+ if (server.loading) {
+ double perc;
+ time_t eta, elapsed;
+ off_t remaining_bytes = server.loading_total_bytes-
+ server.loading_loaded_bytes;
+
+ perc = ((double)server.loading_loaded_bytes /
+ server.loading_total_bytes) * 100;
+
+ elapsed = time(NULL)-server.loading_start_time;
+ if (elapsed == 0) {
+ eta = 1; /* A fake 1 second figure if we don't have enough info */
+ } else {
+ eta = (elapsed*remaining_bytes)/server.loading_loaded_bytes;
+ }
+
+ info = sdscatprintf(info,
+ "loading_start_time:%ld\r\n"
+ "loading_total_bytes:%llu\r\n"
+ "loading_loaded_bytes:%llu\r\n"
+ "loading_loaded_perc:%.2f\r\n"
+ "loading_eta_seconds:%ld\r\n"
+ ,(unsigned long) server.loading_start_time,
+ (unsigned long long) server.loading_total_bytes,
+ (unsigned long long) server.loading_loaded_bytes,
+ perc,
+ eta
+ );
+ }
for (j = 0; j < server.dbnum; j++) {
long long keys, vkeys;
/* ============================ Maxmemory directive ======================== */
-/* Try to free one object form the pre-allocated objects free list.
- * This is useful under low mem conditions as by default we take 1 million
- * free objects allocated. On success REDIS_OK is returned, otherwise
- * REDIS_ERR. */
-int tryFreeOneObjectFromFreelist(void) {
- robj *o;
-
- if (server.vm_enabled) pthread_mutex_lock(&server.obj_freelist_mutex);
- if (listLength(server.objfreelist)) {
- listNode *head = listFirst(server.objfreelist);
- o = listNodeValue(head);
- listDelNode(server.objfreelist,head);
- if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex);
- zfree(o);
- return REDIS_OK;
- } else {
- if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex);
- return REDIS_ERR;
- }
-}
-
/* 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:
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;
+
while (server.maxmemory && zmalloc_used_memory() > server.maxmemory) {
int j, k, freed = 0;
- /* Basic strategy -- remove objects from the free list. */
- if (tryFreeOneObjectFromFreelist() == REDIS_OK) continue;
-
for (j = 0; j < server.dbnum; j++) {
- long bestval;
+ long bestval = 0; /* just to prevent warning */
sds bestkey = NULL;
struct dictEntry *de;
redisDb *db = server.db+j;
de = dictGetRandomKey(dict);
thiskey = dictGetEntryKey(de);
+ /* When policy is volatile-lru we need an additonal lookup
+ * to locate the real key, as dict is set to db->expires. */
+ if (server.maxmemory_policy == REDIS_MAXMEMORY_VOLATILE_LRU)
+ de = dictFind(db->dict, thiskey);
o = dictGetEntryVal(de);
thisval = estimateObjectIdleTime(o);
time_t start;
initServerConfig();
- sortCommandTable();
if (argc == 2) {
if (strcmp(argv[1], "-v") == 0 ||
strcmp(argv[1], "--version") == 0) version();
if (rdbLoad(server.dbfilename) == REDIS_OK)
redisLog(REDIS_NOTICE,"DB loaded from disk: %ld seconds",time(NULL)-start);
}
- redisLog(REDIS_NOTICE,"The server is now ready to accept connections on port %d", server.port);
+ if (server.ipfd > 0)
+ redisLog(REDIS_NOTICE,"The server is now ready to accept connections on port %d", server.port);
+ if (server.sofd > 0)
+ redisLog(REDIS_NOTICE,"The server is now ready to accept connections at %s", server.unixsocket);
aeSetBeforeSleepProc(server.el,beforeSleep);
aeMain(server.el);
aeDeleteEventLoop(server.el);
#define REDIS_STATIC_ARGS 8
#define REDIS_DEFAULT_DBNUM 16
#define REDIS_CONFIGLINE_MAX 1024
-#define REDIS_OBJFREELIST_MAX 1000000 /* Max number of objects to cache */
#define REDIS_MAX_SYNC_TIME 60 /* Slave can't take more to sync */
#define REDIS_EXPIRELOOKUPS_PER_CRON 10 /* lookup 10 expires per loop */
#define REDIS_MAX_WRITE_PER_EVENT (1024*64)
/* Slave replication state - slave side */
#define REDIS_REPL_NONE 0 /* No active replication */
#define REDIS_REPL_CONNECT 1 /* Must connect to master */
-#define REDIS_REPL_CONNECTED 2 /* Connected to master */
+#define REDIS_REPL_TRANSFER 2 /* Receiving .rdb from master */
+#define REDIS_REPL_CONNECTED 3 /* Connected to master */
/* Slave replication state - from the point of view of master
* Note that in SEND_BULK and ONLINE state the slave receives new updates
#define REDIS_MAXMEMORY_VOLATILE_RANDOM 2
#define REDIS_MAXMEMORY_ALLKEYS_LRU 3
#define REDIS_MAXMEMORY_ALLKEYS_RANDOM 4
+#define REDIS_MAXMEMORY_NO_EVICTION 5
/* We can print the stacktrace, so our assert is defined this way: */
#define redisAssert(_e) ((_e)?(void)0 : (_redisAssert(#_e,__FILE__,__LINE__),_exit(1)))
robj *crlf, *ok, *err, *emptybulk, *czero, *cone, *cnegone, *pong, *space,
*colon, *nullbulk, *nullmultibulk, *queued,
*emptymultibulk, *wrongtypeerr, *nokeyerr, *syntaxerr, *sameobjecterr,
- *outofrangeerr, *plus,
+ *outofrangeerr, *loadingerr, *plus,
*select0, *select1, *select2, *select3, *select4,
*select5, *select6, *select7, *select8, *select9,
*messagebulk, *pmessagebulk, *subscribebulk, *unsubscribebulk, *mbulk3,
struct redisServer {
pthread_t mainthread;
int port;
- int fd;
+ char *bindaddr;
+ char *unixsocket;
+ int ipfd;
+ int sofd;
redisDb *db;
long long dirty; /* changes to DB from the last save */
long long dirty_before_bgsave; /* used to restore dirty on failed BGSAVE */
list *clients;
+ dict *commands; /* Command table hahs table */
+ /* RDB / AOF loading information */
+ int loading;
+ off_t loading_total_bytes;
+ off_t loading_loaded_bytes;
+ time_t loading_start_time;
+ /* Fast pointers to often looked up command */
+ struct redisCommand *delCommand, *multiCommand;
list *slaves, *monitors;
char neterr[ANET_ERR_LEN];
aeEventLoop *el;
int cronloops; /* number of times the cron function run */
- list *objfreelist; /* A list of freed objects to avoid malloc() */
time_t lastsave; /* Unix time of last save succeeede */
/* Fields used only for stats */
time_t stat_starttime; /* server start time */
struct saveparam *saveparams;
int saveparamslen;
char *logfile;
- char *bindaddr;
char *dbfilename;
char *appendfilename;
char *requirepass;
int activerehashing;
/* Replication related */
int isslave;
+ /* Slave specific fields */
char *masterauth;
char *masterhost;
int masterport;
redisClient *master; /* client that is master for this slave */
- int replstate;
+ int replstate; /* replication status if the instance is a slave */
+ off_t repl_transfer_left; /* bytes left reading .rdb */
+ int repl_transfer_s; /* slave -> master SYNC socket */
+ int repl_transfer_fd; /* slave -> master SYNC temp file descriptor */
+ char *repl_transfer_tmpfile; /* slave-> master SYNC temp file name */
+ time_t repl_transfer_lastio; /* unix time of the latest read, for timeout */
+ int repl_serve_stale_data; /* Serve stale data when link is down? */
+ /* Limits */
unsigned int maxclients;
unsigned long long maxmemory;
int maxmemory_policy;
int maxmemory_samples;
+ /* Blocked clients */
unsigned int blpop_blocked_clients;
unsigned int vm_blocked_clients;
/* Sort parameters - qsort_r() is only available under BSD so we
list *io_processed; /* List of VM I/O jobs already processed */
list *io_ready_clients; /* Clients ready to be unblocked. All keys loaded */
pthread_mutex_t io_mutex; /* lock to access io_jobs/io_done/io_thread_job */
- pthread_mutex_t obj_freelist_mutex; /* safe redis objects creation/free */
pthread_mutex_t io_swapfile_mutex; /* So we can lseek + write */
pthread_attr_t io_threads_attr; /* attributes for threads creation */
int io_active_threads; /* Number of running I/O threads */
void setDeferredMultiBulkLength(redisClient *c, void *node, long length);
void addReplySds(redisClient *c, sds s);
void processInputBuffer(redisClient *c);
-void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask);
+void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask);
+void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask);
void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask);
void addReplyBulk(redisClient *c, robj *obj);
void addReplyBulkCString(redisClient *c, char *s);
robj *tryObjectEncoding(robj *o);
robj *getDecodedObject(robj *o);
size_t stringObjectLen(robj *o);
-int tryFreeOneObjectFromFreelist(void);
robj *createStringObjectFromLongLong(long long value);
robj *createListObject(void);
robj *createZiplistObject(void);
void replicationFeedMonitors(list *monitors, int dictid, robj **argv, int argc);
int syncWithMaster(void);
void updateSlavesWaitingBgsave(int bgsaveerr);
+void replicationCron(void);
+
+/* Generic persistence functions */
+void startLoading(FILE *fp);
+void loadingProgress(off_t pos);
+void stopLoading(void);
/* RDB persistence */
int rdbLoad(char *filename);
void freeMemoryIfNeeded(void);
int processCommand(redisClient *c);
void setupSigSegvAction(void);
-struct redisCommand *lookupCommand(char *name);
+struct redisCommand *lookupCommand(sds name);
+struct redisCommand *lookupCommandByCString(char *s);
void call(redisClient *c, struct redisCommand *cmd);
int prepareForShutdown();
void redisLog(int level, const char *fmt, ...);
void updateDictResizePolicy(void);
int htNeedsResize(dict *dict);
void oom(const char *msg);
+void populateCommandTable(void);
/* Virtual Memory */
void vmInit(void);
#include <fcntl.h>
#include <sys/stat.h>
+/* ---------------------------------- MASTER -------------------------------- */
+
void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
listNode *ln;
listIter li;
}
}
+/* ----------------------------------- SLAVE -------------------------------- */
+
+/* Abort the async download of the bulk dataset while SYNC-ing with master */
+void replicationAbortSyncTransfer(void) {
+ redisAssert(server.replstate == REDIS_REPL_TRANSFER);
+
+ aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE);
+ close(server.repl_transfer_s);
+ close(server.repl_transfer_fd);
+ unlink(server.repl_transfer_tmpfile);
+ zfree(server.repl_transfer_tmpfile);
+ server.replstate = REDIS_REPL_CONNECT;
+}
+
+/* Asynchronously read the SYNC payload we receive from a master */
+void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
+ char buf[4096];
+ ssize_t nread, readlen;
+ REDIS_NOTUSED(el);
+ REDIS_NOTUSED(privdata);
+ REDIS_NOTUSED(mask);
+
+ /* If repl_transfer_left == -1 we still have to read the bulk length
+ * from the master reply. */
+ if (server.repl_transfer_left == -1) {
+ if (syncReadLine(fd,buf,1024,3600) == -1) {
+ redisLog(REDIS_WARNING,
+ "I/O error reading bulk count from MASTER: %s",
+ strerror(errno));
+ replicationAbortSyncTransfer();
+ return;
+ }
+ if (buf[0] == '-') {
+ redisLog(REDIS_WARNING,
+ "MASTER aborted replication with an error: %s",
+ buf+1);
+ replicationAbortSyncTransfer();
+ return;
+ } else if (buf[0] != '$') {
+ redisLog(REDIS_WARNING,"Bad protocol from MASTER, the first byte is not '$', are you sure the host and port are right?");
+ replicationAbortSyncTransfer();
+ return;
+ }
+ server.repl_transfer_left = strtol(buf+1,NULL,10);
+ redisLog(REDIS_NOTICE,
+ "MASTER <-> SLAVE sync: receiving %ld bytes from master",
+ server.repl_transfer_left);
+ return;
+ }
+
+ /* Read bulk data */
+ readlen = (server.repl_transfer_left < (signed)sizeof(buf)) ?
+ server.repl_transfer_left : (signed)sizeof(buf);
+ nread = read(fd,buf,readlen);
+ if (nread <= 0) {
+ redisLog(REDIS_WARNING,"I/O error trying to sync with MASTER: %s",
+ (nread == -1) ? strerror(errno) : "connection lost");
+ replicationAbortSyncTransfer();
+ return;
+ }
+ server.repl_transfer_lastio = time(NULL);
+ if (write(server.repl_transfer_fd,buf,nread) != nread) {
+ redisLog(REDIS_WARNING,"Write error or short write writing to the DB dump file needed for MASTER <-> SLAVE synchrnonization: %s", strerror(errno));
+ replicationAbortSyncTransfer();
+ return;
+ }
+ server.repl_transfer_left -= nread;
+ /* Check if the transfer is now complete */
+ if (server.repl_transfer_left == 0) {
+ if (rename(server.repl_transfer_tmpfile,server.dbfilename) == -1) {
+ redisLog(REDIS_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno));
+ replicationAbortSyncTransfer();
+ return;
+ }
+ redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Loading DB in memory");
+ emptyDb();
+ /* Before loading the DB into memory we need to delete the readable
+ * handler, otherwise it will get called recursively since
+ * rdbLoad() will call the event loop to process events from time to
+ * time for non blocking loading. */
+ aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE);
+ if (rdbLoad(server.dbfilename) != REDIS_OK) {
+ redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk");
+ replicationAbortSyncTransfer();
+ return;
+ }
+ /* Final setup of the connected slave <- master link */
+ zfree(server.repl_transfer_tmpfile);
+ close(server.repl_transfer_fd);
+ server.master = createClient(server.repl_transfer_s);
+ server.master->flags |= REDIS_MASTER;
+ server.master->authenticated = 1;
+ server.replstate = REDIS_REPL_CONNECTED;
+ redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Finished with success");
+ }
+}
+
int syncWithMaster(void) {
char buf[1024], tmpfile[256], authcmd[1024];
- long dumpsize;
int fd = anetTcpConnect(NULL,server.masterhost,server.masterport);
int dfd, maxtries = 5;
strerror(errno));
return REDIS_ERR;
}
- /* Read the bulk write count */
- if (syncReadLine(fd,buf,1024,3600) == -1) {
- close(fd);
- redisLog(REDIS_WARNING,"I/O error reading bulk count from MASTER: %s",
- strerror(errno));
- return REDIS_ERR;
- }
- if (buf[0] == '-') {
- close(fd);
- redisLog(REDIS_WARNING,"MASTER aborted replication with an error: %s",
- buf+1);
- return REDIS_ERR;
- } else if (buf[0] != '$') {
- close(fd);
- redisLog(REDIS_WARNING,"Bad protocol from MASTER, the first byte is not '$', are you sure the host and port are right?");
- return REDIS_ERR;
- }
- dumpsize = strtol(buf+1,NULL,10);
- redisLog(REDIS_NOTICE,"Receiving %ld bytes data dump from MASTER",dumpsize);
- /* Read the bulk write data on a temp file */
+
+ /* Prepare a suitable temp file for bulk transfer */
while(maxtries--) {
snprintf(tmpfile,256,
"temp-%d.%ld.rdb",(int)time(NULL),(long int)getpid());
redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno));
return REDIS_ERR;
}
- while(dumpsize) {
- int nread, nwritten;
- nread = read(fd,buf,(dumpsize < 1024)?dumpsize:1024);
- if (nread <= 0) {
- redisLog(REDIS_WARNING,"I/O error trying to sync with MASTER: %s",
- (nread == -1) ? strerror(errno) : "connection lost");
- close(fd);
- close(dfd);
- return REDIS_ERR;
- }
- nwritten = write(dfd,buf,nread);
- if (nwritten == -1) {
- redisLog(REDIS_WARNING,"Write error writing to the DB dump file needed for MASTER <-> SLAVE synchrnonization: %s", strerror(errno));
- close(fd);
- close(dfd);
- return REDIS_ERR;
- }
- dumpsize -= nread;
- }
- close(dfd);
- if (rename(tmpfile,server.dbfilename) == -1) {
- redisLog(REDIS_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno));
- unlink(tmpfile);
+ /* Setup the non blocking download of the bulk file. */
+ if (aeCreateFileEvent(server.el, fd, AE_READABLE, readSyncBulkPayload, NULL)
+ == AE_ERR)
+ {
close(fd);
+ redisLog(REDIS_WARNING,"Can't create readable event for SYNC");
return REDIS_ERR;
}
- emptyDb();
- if (rdbLoad(server.dbfilename) != REDIS_OK) {
- redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk");
- close(fd);
- return REDIS_ERR;
- }
- server.master = createClient(fd);
- server.master->flags |= REDIS_MASTER;
- server.master->authenticated = 1;
- server.replstate = REDIS_REPL_CONNECTED;
+ server.replstate = REDIS_REPL_TRANSFER;
+ server.repl_transfer_left = -1;
+ server.repl_transfer_s = fd;
+ server.repl_transfer_fd = dfd;
+ server.repl_transfer_lastio = time(NULL);
+ server.repl_transfer_tmpfile = zstrdup(tmpfile);
return REDIS_OK;
}
sdsfree(server.masterhost);
server.masterhost = NULL;
if (server.master) freeClient(server.master);
+ if (server.replstate == REDIS_REPL_TRANSFER)
+ replicationAbortSyncTransfer();
server.replstate = REDIS_REPL_NONE;
redisLog(REDIS_NOTICE,"MASTER MODE enabled (user request)");
}
server.masterhost = sdsdup(c->argv[1]->ptr);
server.masterport = atoi(c->argv[2]->ptr);
if (server.master) freeClient(server.master);
+ if (server.replstate == REDIS_REPL_TRANSFER)
+ replicationAbortSyncTransfer();
server.replstate = REDIS_REPL_CONNECT;
redisLog(REDIS_NOTICE,"SLAVE OF %s:%d enabled (user request)",
server.masterhost, server.masterport);
}
addReply(c,shared.ok);
}
+
+/* --------------------------- REPLICATION CRON ---------------------------- */
+
+#define REDIS_REPL_TRANSFER_TIMEOUT 60
+
+void replicationCron(void) {
+ /* Bulk transfer I/O timeout? */
+ if (server.masterhost && server.replstate == REDIS_REPL_TRANSFER &&
+ (time(NULL)-server.repl_transfer_lastio) > REDIS_REPL_TRANSFER_TIMEOUT)
+ {
+ redisLog(REDIS_WARNING,"Timeout receiving bulk data from MASTER...");
+ replicationAbortSyncTransfer();
+ }
+
+ /* Check if we should connect to a MASTER */
+ if (server.replstate == REDIS_REPL_CONNECT) {
+ redisLog(REDIS_NOTICE,"Connecting to MASTER...");
+ if (syncWithMaster() == REDIS_OK) {
+ redisLog(REDIS_NOTICE,"MASTER <-> SLAVE sync started: SYNC sent");
+ if (server.appendonly) rewriteAppendOnlyFileBackground();
+ }
+ }
+}
-#define REDIS_VERSION "2.1.5"
+#define REDIS_VERSION "2.1.7"
server.io_processed = listCreate();
server.io_ready_clients = listCreate();
pthread_mutex_init(&server.io_mutex,NULL);
- pthread_mutex_init(&server.obj_freelist_mutex,NULL);
pthread_mutex_init(&server.io_swapfile_mutex,NULL);
server.io_active_threads = 0;
if (pipe(pipefds) == -1) {
"pls: %2u, "
"payload %5u"
"} ",
- (long unsigned int)p,
+ (long unsigned)p,
index,
- p-zl,
+ (unsigned long) (p-zl),
entry.headersize+entry.len,
entry.headersize,
entry.prevrawlen,
p += entry.headersize;
if (ZIP_IS_STR(entry.encoding)) {
if (entry.len > 40) {
- fwrite(p,40,1,stdout);
+ if (fwrite(p,40,1,stdout) == 0) perror("fwrite");
printf("...");
} else {
- fwrite(p,entry.len,1,stdout);
+ if (entry.len &&
+ fwrite(p,entry.len,1,stdout) == 0) perror("fwrite");
}
} else {
printf("%lld", (long long) zipLoadInteger(p,entry.encoding));
printf("Pop tail: ");
if (vstr)
- fwrite(vstr,vlen,1,stdout);
+ if (vlen && fwrite(vstr,vlen,1,stdout) == 0) perror("fwrite");
else
printf("%lld", vlong);
return 1;
}
if (entry) {
- fwrite(entry,elen,1,stdout);
+ if (elen && fwrite(entry,elen,1,stdout) == 0) perror("fwrite");
printf("\n");
} else {
printf("%lld\n", value);
return 1;
}
if (entry) {
- fwrite(entry,elen,1,stdout);
+ if (elen && fwrite(entry,elen,1,stdout) == 0) perror("fwrite");
printf("\n");
} else {
printf("%lld\n", value);
return 1;
}
if (entry) {
- fwrite(entry,elen,1,stdout);
+ if (elen && fwrite(entry,elen,1,stdout) == 0) perror("fwrite");
printf("\n");
} else {
printf("%lld\n", value);
while (ziplistGet(p, &entry, &elen, &value)) {
printf("Entry: ");
if (entry) {
- fwrite(entry,elen,1,stdout);
+ if (elen && fwrite(entry,elen,1,stdout) == 0) perror("fwrite");
} else {
printf("%lld", value);
}
while (ziplistGet(p, &entry, &elen, &value)) {
printf("Entry: ");
if (entry) {
- fwrite(entry,elen,1,stdout);
+ if (elen && fwrite(entry,elen,1,stdout) == 0) perror("fwrite");
} else {
printf("%lld", value);
}
while (ziplistGet(p, &entry, &elen, &value)) {
printf("Entry: ");
if (entry) {
- fwrite(entry,elen,1,stdout);
+ if (elen && fwrite(entry,elen,1,stdout) == 0) perror("fwrite");
} else {
printf("%lld", value);
}
while (ziplistGet(p, &entry, &elen, &value)) {
printf("Entry: ");
if (entry) {
- fwrite(entry,elen,1,stdout);
+ if (elen && fwrite(entry,elen,1,stdout) == 0) perror("fwrite");
} else {
printf("%lld", value);
}
while (ziplistGet(p, &entry, &elen, &value)) {
printf("Entry: ");
if (entry) {
- fwrite(entry,elen,1,stdout);
+ if (elen && fwrite(entry,elen,1,stdout) == 0) perror("fwrite");
} else {
printf("%lld", value);
}
} else {
printf("Entry: ");
if (entry) {
- fwrite(entry,elen,1,stdout);
+ if (elen && fwrite(entry,elen,1,stdout) == 0)
+ perror("fwrite");
} else {
printf("%lld",value);
}
l = zipmapDecodeLength(p);
printf("{key %u}",l);
p += zipmapEncodeLength(NULL,l);
- fwrite(p,l,1,stdout);
+ if (l != 0 && fwrite(p,l,1,stdout) == 0) perror("fwrite");
p += l;
l = zipmapDecodeLength(p);
printf("{value %u}",l);
p += zipmapEncodeLength(NULL,l);
e = *p++;
- fwrite(p,l,1,stdout);
+ if (l != 0 && fwrite(p,l,1,stdout) == 0) perror("fwrite");
p += l+e;
if (e) {
printf("[");
zmalloc_thread_safe = 1;
}
-/* Fragmentation = RSS / allocated-bytes */
+/* Get the RSS information in an OS-specific way.
+ *
+ * WARNING: the function zmalloc_get_rss() is not designed to be fast
+ * and may not be called in the busy loops where Redis tries to release
+ * memory expiring or swapping out objects.
+ *
+ * For this kind of "fast RSS reporting" usages use instead the
+ * function RedisEstimateRSS() that is a much faster (and less precise)
+ * version of the funciton. */
#if defined(HAVE_PROCFS)
#include <unistd.h>
#include <sys/stat.h>
#include <fcntl.h>
-float zmalloc_get_fragmentation_ratio(void) {
- size_t allocated = zmalloc_used_memory();
+size_t zmalloc_get_rss(void) {
int page = sysconf(_SC_PAGESIZE);
size_t rss;
char buf[4096];
rss = strtoll(p,NULL,10);
rss *= page;
- return (float)rss/allocated;
+ return rss;
}
#elif defined(HAVE_TASKINFO)
#include <unistd.h>
#include <mach/task.h>
#include <mach/mach_init.h>
-float zmalloc_get_fragmentation_ratio(void) {
+size_t zmalloc_get_rss(void) {
task_t task = MACH_PORT_NULL;
struct task_basic_info t_info;
mach_msg_type_number_t t_info_count = TASK_BASIC_INFO_COUNT;
return 0;
task_info(task, TASK_BASIC_INFO, (task_info_t)&t_info, &t_info_count);
- return (float)t_info.resident_size/zmalloc_used_memory();
+ return t_info.resident_size;
}
#else
-float zmalloc_get_fragmentation_ratio(void) {
- return 0;
+float zmalloc_get_rss(void) {
+ /* If we can't get the RSS in an OS-specific way for this system just
+ * return the memory usage we estimated in zmalloc()..
+ *
+ * Fragmentation will appear to be always 1 (no fragmentation)
+ * of course... */
+ return zmalloc_used_memory();
}
#endif
+
+/* Fragmentation = RSS / allocated-bytes */
+float zmalloc_get_fragmentation_ratio(void) {
+ return (float)zmalloc_get_rss()/zmalloc_used_memory();
+}
size_t zmalloc_used_memory(void);
void zmalloc_enable_thread_safeness(void);
float zmalloc_get_fragmentation_ratio(void);
+size_t zmalloc_get_rss(void);
#endif /* _ZMALLOC_H */
randpath {{*}$r sadd $k $v} \
{{*}$r srem $k $v} \
{
- set otherset [findKeyWithType r set]
+ set otherset [findKeyWithType {*}$r set]
if {$otherset ne {}} {
randpath {
{*}$r sunionstore $k2 $k $otherset
randpath {{*}$r zadd $k $d $v} \
{{*}$r zrem $k $v} \
{
- set otherzset [findKeyWithType r zset]
+ set otherzset [findKeyWithType {*}$r zset]
if {$otherzset ne {}} {
randpath {
{*}$r zunionstore $k2 2 $k $otherzset
projects / redis.git / commitdiff
