Unverified Commit 8e0264cf authored by Michael Grunder's avatar Michael Grunder Committed by GitHub
Browse files

Allow users to replace allocator and handle OOM everywhere. (#800)

* Adds an indirection to every allocation/deallocation to allow users to 
  plug in ones of their choosing (use custom functions, jemalloc, etc).

* Gracefully handle OOM everywhere in hiredis.  This should make it possible
  for users of the library to have more flexibility in how they handle such situations.

* Changes `redisReaderTask->elements` from an `int` to a `long long` to prevent
  a possible overflow when transferring the task elements into a `redisReply`.

* Adds a configurable `max elements` member to `redisReader` that defaults to
  2^32 - 1.  This can be set to "unlimited" by setting the value to zero.
parent 83bba659
...@@ -24,14 +24,14 @@ PROJECT(hiredis VERSION "${VERSION}") ...@@ -24,14 +24,14 @@ PROJECT(hiredis VERSION "${VERSION}")
SET(ENABLE_EXAMPLES OFF CACHE BOOL "Enable building hiredis examples") SET(ENABLE_EXAMPLES OFF CACHE BOOL "Enable building hiredis examples")
SET(hiredis_sources SET(hiredis_sources
alloc.c
async.c async.c
dict.c dict.c
hiredis.c hiredis.c
net.c net.c
read.c read.c
sds.c sds.c
sockcompat.c sockcompat.c)
alloc.c)
SET(hiredis_sources ${hiredis_sources}) SET(hiredis_sources ${hiredis_sources})
......
...@@ -3,7 +3,7 @@ ...@@ -3,7 +3,7 @@
# Copyright (C) 2010-2011 Pieter Noordhuis <pcnoordhuis at gmail dot com> # Copyright (C) 2010-2011 Pieter Noordhuis <pcnoordhuis at gmail dot com>
# This file is released under the BSD license, see the COPYING file # This file is released under the BSD license, see the COPYING file
OBJ=net.o hiredis.o sds.o async.o read.o sockcompat.o alloc.o OBJ=alloc.o net.o hiredis.o sds.o async.o read.o sockcompat.o
SSL_OBJ=ssl.o SSL_OBJ=ssl.o
EXAMPLES=hiredis-example hiredis-example-libevent hiredis-example-libev hiredis-example-glib EXAMPLES=hiredis-example hiredis-example-libevent hiredis-example-libev hiredis-example-glib
ifeq ($(USE_SSL),1) ifeq ($(USE_SSL),1)
...@@ -109,16 +109,16 @@ all: $(SSL_DYLIBNAME) $(SSL_STLIBNAME) $(SSL_PKGCONFNAME) ...@@ -109,16 +109,16 @@ all: $(SSL_DYLIBNAME) $(SSL_STLIBNAME) $(SSL_PKGCONFNAME)
endif endif
# Deps (use make dep to generate this) # Deps (use make dep to generate this)
alloc.o: alloc.c fmacros.h alloc.h
async.o: async.c fmacros.h alloc.h async.h hiredis.h read.h sds.h net.h dict.c dict.h win32.h async_private.h async.o: async.c fmacros.h alloc.h async.h hiredis.h read.h sds.h net.h dict.c dict.h win32.h async_private.h
dict.o: dict.c fmacros.h alloc.h dict.h dict.o: dict.c fmacros.h alloc.h dict.h
hiredis.o: hiredis.c fmacros.h hiredis.h read.h sds.h alloc.h net.h async.h win32.h hiredis.o: hiredis.c fmacros.h hiredis.h read.h sds.h alloc.h net.h async.h win32.h
alloc.o: alloc.c alloc.h
net.o: net.c fmacros.h net.h hiredis.h read.h sds.h alloc.h sockcompat.h win32.h net.o: net.c fmacros.h net.h hiredis.h read.h sds.h alloc.h sockcompat.h win32.h
read.o: read.c fmacros.h read.h sds.h win32.h read.o: read.c fmacros.h alloc.h read.h sds.h win32.h
sds.o: sds.c sds.h sdsalloc.h sds.o: sds.c sds.h sdsalloc.h alloc.h
sockcompat.o: sockcompat.c sockcompat.h sockcompat.o: sockcompat.c sockcompat.h
ssl.o: ssl.c hiredis.h read.h sds.h alloc.h async.h async_private.h ssl.o: ssl.c hiredis.h read.h sds.h alloc.h async.h win32.h async_private.h
test.o: test.c fmacros.h hiredis.h read.h sds.h alloc.h net.h test.o: test.c fmacros.h hiredis.h read.h sds.h alloc.h net.h sockcompat.h win32.h
$(DYLIBNAME): $(OBJ) $(DYLIBNAME): $(OBJ)
$(DYLIB_MAKE_CMD) -o $(DYLIBNAME) $(OBJ) $(REAL_LDFLAGS) $(DYLIB_MAKE_CMD) -o $(DYLIBNAME) $(OBJ) $(REAL_LDFLAGS)
......
...@@ -404,6 +404,16 @@ This should be done only in order to maximize performances when working with ...@@ -404,6 +404,16 @@ This should be done only in order to maximize performances when working with
large payloads. The context should be set back to `REDIS_READER_MAX_BUF` again large payloads. The context should be set back to `REDIS_READER_MAX_BUF` again
as soon as possible in order to prevent allocation of useless memory. as soon as possible in order to prevent allocation of useless memory.
### Reader max array elements
By default the hiredis reply parser sets the maximum number of multi-bulk elements
to 2^32 - 1 or 4,294,967,295 entries. If you need to process multi-bulk replies
with more than this many elements you can set the value higher or to zero, meaning
unlimited with:
```c
context->reader->maxelements = 0;
```
## SSL/TLS Support ## SSL/TLS Support
### Building ### Building
......
...@@ -96,7 +96,7 @@ static void redisAeCleanup(void *privdata) { ...@@ -96,7 +96,7 @@ static void redisAeCleanup(void *privdata) {
redisAeEvents *e = (redisAeEvents*)privdata; redisAeEvents *e = (redisAeEvents*)privdata;
redisAeDelRead(privdata); redisAeDelRead(privdata);
redisAeDelWrite(privdata); redisAeDelWrite(privdata);
free(e); hi_free(e);
} }
static int redisAeAttach(aeEventLoop *loop, redisAsyncContext *ac) { static int redisAeAttach(aeEventLoop *loop, redisAsyncContext *ac) {
...@@ -109,6 +109,9 @@ static int redisAeAttach(aeEventLoop *loop, redisAsyncContext *ac) { ...@@ -109,6 +109,9 @@ static int redisAeAttach(aeEventLoop *loop, redisAsyncContext *ac) {
/* Create container for context and r/w events */ /* Create container for context and r/w events */
e = (redisAeEvents*)hi_malloc(sizeof(*e)); e = (redisAeEvents*)hi_malloc(sizeof(*e));
if (e == NULL)
return REDIS_ERR;
e->context = ac; e->context = ac;
e->loop = loop; e->loop = loop;
e->fd = c->fd; e->fd = c->fd;
......
...@@ -134,6 +134,9 @@ redis_source_new (redisAsyncContext *ac) ...@@ -134,6 +134,9 @@ redis_source_new (redisAsyncContext *ac)
g_return_val_if_fail(ac != NULL, NULL); g_return_val_if_fail(ac != NULL, NULL);
source = (RedisSource *)g_source_new(&source_funcs, sizeof *source); source = (RedisSource *)g_source_new(&source_funcs, sizeof *source);
if (source == NULL)
return NULL;
source->ac = ac; source->ac = ac;
source->poll_fd.fd = c->fd; source->poll_fd.fd = c->fd;
source->poll_fd.events = 0; source->poll_fd.events = 0;
......
...@@ -43,7 +43,7 @@ static void redisIvykisCleanup(void *privdata) { ...@@ -43,7 +43,7 @@ static void redisIvykisCleanup(void *privdata) {
redisIvykisEvents *e = (redisIvykisEvents*)privdata; redisIvykisEvents *e = (redisIvykisEvents*)privdata;
iv_fd_unregister(&e->fd); iv_fd_unregister(&e->fd);
free(e); hi_free(e);
} }
static int redisIvykisAttach(redisAsyncContext *ac) { static int redisIvykisAttach(redisAsyncContext *ac) {
...@@ -56,6 +56,9 @@ static int redisIvykisAttach(redisAsyncContext *ac) { ...@@ -56,6 +56,9 @@ static int redisIvykisAttach(redisAsyncContext *ac) {
/* Create container for context and r/w events */ /* Create container for context and r/w events */
e = (redisIvykisEvents*)hi_malloc(sizeof(*e)); e = (redisIvykisEvents*)hi_malloc(sizeof(*e));
if (e == NULL)
return REDIS_ERR;
e->context = ac; e->context = ac;
/* Register functions to start/stop listening for events */ /* Register functions to start/stop listening for events */
......
...@@ -116,7 +116,7 @@ static void redisLibevCleanup(void *privdata) { ...@@ -116,7 +116,7 @@ static void redisLibevCleanup(void *privdata) {
redisLibevDelRead(privdata); redisLibevDelRead(privdata);
redisLibevDelWrite(privdata); redisLibevDelWrite(privdata);
redisLibevStopTimer(privdata); redisLibevStopTimer(privdata);
free(e); hi_free(e);
} }
static void redisLibevTimeout(EV_P_ ev_timer *timer, int revents) { static void redisLibevTimeout(EV_P_ ev_timer *timer, int revents) {
...@@ -149,6 +149,9 @@ static int redisLibevAttach(EV_P_ redisAsyncContext *ac) { ...@@ -149,6 +149,9 @@ static int redisLibevAttach(EV_P_ redisAsyncContext *ac) {
/* Create container for context and r/w events */ /* Create container for context and r/w events */
e = (redisLibevEvents*)hi_calloc(1, sizeof(*e)); e = (redisLibevEvents*)hi_calloc(1, sizeof(*e));
if (e == NULL)
return REDIS_ERR;
e->context = ac; e->context = ac;
#if EV_MULTIPLICITY #if EV_MULTIPLICITY
e->loop = loop; e->loop = loop;
......
...@@ -47,7 +47,7 @@ typedef struct redisLibeventEvents { ...@@ -47,7 +47,7 @@ typedef struct redisLibeventEvents {
} redisLibeventEvents; } redisLibeventEvents;
static void redisLibeventDestroy(redisLibeventEvents *e) { static void redisLibeventDestroy(redisLibeventEvents *e) {
free(e); hi_free(e);
} }
static void redisLibeventHandler(int fd, short event, void *arg) { static void redisLibeventHandler(int fd, short event, void *arg) {
...@@ -153,6 +153,9 @@ static int redisLibeventAttach(redisAsyncContext *ac, struct event_base *base) { ...@@ -153,6 +153,9 @@ static int redisLibeventAttach(redisAsyncContext *ac, struct event_base *base) {
/* Create container for context and r/w events */ /* Create container for context and r/w events */
e = (redisLibeventEvents*)hi_calloc(1, sizeof(*e)); e = (redisLibeventEvents*)hi_calloc(1, sizeof(*e));
if (e == NULL)
return REDIS_ERR;
e->context = ac; e->context = ac;
/* Register functions to start/stop listening for events */ /* Register functions to start/stop listening for events */
......
...@@ -73,7 +73,7 @@ static void redisLibuvDelWrite(void *privdata) { ...@@ -73,7 +73,7 @@ static void redisLibuvDelWrite(void *privdata) {
static void on_close(uv_handle_t* handle) { static void on_close(uv_handle_t* handle) {
redisLibuvEvents* p = (redisLibuvEvents*)handle->data; redisLibuvEvents* p = (redisLibuvEvents*)handle->data;
free(p); hi_free(p);
} }
...@@ -98,11 +98,9 @@ static int redisLibuvAttach(redisAsyncContext* ac, uv_loop_t* loop) { ...@@ -98,11 +98,9 @@ static int redisLibuvAttach(redisAsyncContext* ac, uv_loop_t* loop) {
ac->ev.delWrite = redisLibuvDelWrite; ac->ev.delWrite = redisLibuvDelWrite;
ac->ev.cleanup = redisLibuvCleanup; ac->ev.cleanup = redisLibuvCleanup;
redisLibuvEvents* p = (redisLibuvEvents*)malloc(sizeof(*p)); redisLibuvEvents* p = (redisLibuvEvents*)hi_malloc(sizeof(*p));
if (p == NULL)
if (!p) {
return REDIS_ERR; return REDIS_ERR;
}
memset(p, 0, sizeof(*p)); memset(p, 0, sizeof(*p));
......
...@@ -27,7 +27,7 @@ static int freeRedisRunLoop(RedisRunLoop* redisRunLoop) { ...@@ -27,7 +27,7 @@ static int freeRedisRunLoop(RedisRunLoop* redisRunLoop) {
CFSocketInvalidate(redisRunLoop->socketRef); CFSocketInvalidate(redisRunLoop->socketRef);
CFRelease(redisRunLoop->socketRef); CFRelease(redisRunLoop->socketRef);
} }
free(redisRunLoop); hi_free(redisRunLoop);
} }
return REDIS_ERR; return REDIS_ERR;
} }
...@@ -80,8 +80,9 @@ static int redisMacOSAttach(redisAsyncContext *redisAsyncCtx, CFRunLoopRef runLo ...@@ -80,8 +80,9 @@ static int redisMacOSAttach(redisAsyncContext *redisAsyncCtx, CFRunLoopRef runLo
/* Nothing should be attached when something is already attached */ /* Nothing should be attached when something is already attached */
if( redisAsyncCtx->ev.data != NULL ) return REDIS_ERR; if( redisAsyncCtx->ev.data != NULL ) return REDIS_ERR;
RedisRunLoop* redisRunLoop = (RedisRunLoop*) calloc(1, sizeof(RedisRunLoop)); RedisRunLoop* redisRunLoop = (RedisRunLoop*) hi_calloc(1, sizeof(RedisRunLoop));
if( !redisRunLoop ) return REDIS_ERR; if (redisRunLoop == NULL)
return REDIS_ERR;
/* Setup redis stuff */ /* Setup redis stuff */
redisRunLoop->context = redisAsyncCtx; redisRunLoop->context = redisAsyncCtx;
......
...@@ -31,35 +31,56 @@ ...@@ -31,35 +31,56 @@
#include "fmacros.h" #include "fmacros.h"
#include "alloc.h" #include "alloc.h"
#include <string.h> #include <string.h>
#include <stdlib.h>
void *hi_malloc(size_t size) { hiredisAllocFuncs hiredisAllocFns = {
void *ptr = malloc(size); .malloc = malloc,
if (ptr == NULL) .calloc = calloc,
HIREDIS_OOM_HANDLER; .realloc = realloc,
.strdup = strdup,
.free = free,
};
/* Override hiredis' allocators with ones supplied by the user */
hiredisAllocFuncs hiredisSetAllocators(hiredisAllocFuncs *override) {
hiredisAllocFuncs orig = hiredisAllocFns;
return ptr; hiredisAllocFns = *override;
return orig;
} }
void *hi_calloc(size_t nmemb, size_t size) { /* Reset allocators to use libc defaults */
void *ptr = calloc(nmemb, size); void hiredisResetAllocators(void) {
if (ptr == NULL) hiredisAllocFns = (hiredisAllocFuncs) {
HIREDIS_OOM_HANDLER; .malloc = malloc,
.calloc = calloc,
.realloc = realloc,
.strdup = strdup,
.free = free,
};
}
#ifdef _WIN32
return ptr; void *hi_malloc(size_t size) {
return hiredisAllocFns.malloc(size);
} }
void *hi_realloc(void *ptr, size_t size) { void *hi_calloc(size_t nmemb, size_t size) {
void *newptr = realloc(ptr, size); return hiredisAllocFns.calloc(nmemb, size);
if (newptr == NULL) }
HIREDIS_OOM_HANDLER;
return newptr; void *hi_realloc(void *ptr, size_t size) {
return hiredisAllocFns.realloc(ptr, size);
} }
char *hi_strdup(const char *str) { char *hi_strdup(const char *str) {
char *newstr = strdup(str); return hiredisAllocFns.strdup(str);
if (newstr == NULL) }
HIREDIS_OOM_HANDLER;
return newstr; void hi_free(void *ptr) {
hiredisAllocFns.free(ptr);
} }
#endif
...@@ -31,20 +31,58 @@ ...@@ -31,20 +31,58 @@
#ifndef HIREDIS_ALLOC_H #ifndef HIREDIS_ALLOC_H
#define HIREDIS_ALLOC_H #define HIREDIS_ALLOC_H
#include <stdlib.h> /* for size_t */ #include <stddef.h> /* for size_t */
#ifndef HIREDIS_OOM_HANDLER
#define HIREDIS_OOM_HANDLER abort()
#endif
#ifdef __cplusplus #ifdef __cplusplus
extern "C" { extern "C" {
#endif #endif
/* Structure pointing to our actually configured allocators */
typedef struct hiredisAllocFuncs {
void *(*malloc)(size_t);
void *(*calloc)(size_t,size_t);
void *(*realloc)(void*,size_t);
char *(*strdup)(const char*);
void (*free)(void*);
} hiredisAllocFuncs;
hiredisAllocFuncs hiredisSetAllocators(hiredisAllocFuncs *ha);
void hiredisResetAllocators(void);
#ifndef _WIN32
/* Hiredis' configured allocator function pointer struct */
extern hiredisAllocFuncs hiredisAllocFns;
static inline void *hi_malloc(size_t size) {
return hiredisAllocFns.malloc(size);
}
static inline void *hi_calloc(size_t nmemb, size_t size) {
return hiredisAllocFns.calloc(nmemb, size);
}
static inline void *hi_realloc(void *ptr, size_t size) {
return hiredisAllocFns.realloc(ptr, size);
}
static inline char *hi_strdup(const char *str) {
return hiredisAllocFns.strdup(str);
}
static inline void hi_free(void *ptr) {
hiredisAllocFns.free(ptr);
}
#else
void *hi_malloc(size_t size); void *hi_malloc(size_t size);
void *hi_calloc(size_t nmemb, size_t size); void *hi_calloc(size_t nmemb, size_t size);
void *hi_realloc(void *ptr, size_t size); void *hi_realloc(void *ptr, size_t size);
char *hi_strdup(const char *str); char *hi_strdup(const char *str);
void hi_free(void *ptr);
#endif
#ifdef __cplusplus #ifdef __cplusplus
} }
......
...@@ -47,8 +47,9 @@ ...@@ -47,8 +47,9 @@
#include "async_private.h" #include "async_private.h"
/* Forward declaration of function in hiredis.c */ /* Forward declarations of hiredis.c functions */
int __redisAppendCommand(redisContext *c, const char *cmd, size_t len); int __redisAppendCommand(redisContext *c, const char *cmd, size_t len);
void __redisSetError(redisContext *c, int type, const char *str);
/* Functions managing dictionary of callbacks for pub/sub. */ /* Functions managing dictionary of callbacks for pub/sub. */
static unsigned int callbackHash(const void *key) { static unsigned int callbackHash(const void *key) {
...@@ -58,7 +59,12 @@ static unsigned int callbackHash(const void *key) { ...@@ -58,7 +59,12 @@ static unsigned int callbackHash(const void *key) {
static void *callbackValDup(void *privdata, const void *src) { static void *callbackValDup(void *privdata, const void *src) {
((void) privdata); ((void) privdata);
redisCallback *dup = hi_malloc(sizeof(*dup)); redisCallback *dup;
dup = hi_malloc(sizeof(*dup));
if (dup == NULL)
return NULL;
memcpy(dup,src,sizeof(*dup)); memcpy(dup,src,sizeof(*dup));
return dup; return dup;
} }
...@@ -80,7 +86,7 @@ static void callbackKeyDestructor(void *privdata, void *key) { ...@@ -80,7 +86,7 @@ static void callbackKeyDestructor(void *privdata, void *key) {
static void callbackValDestructor(void *privdata, void *val) { static void callbackValDestructor(void *privdata, void *val) {
((void) privdata); ((void) privdata);
free(val); hi_free(val);
} }
static dictType callbackDict = { static dictType callbackDict = {
...@@ -94,10 +100,19 @@ static dictType callbackDict = { ...@@ -94,10 +100,19 @@ static dictType callbackDict = {
static redisAsyncContext *redisAsyncInitialize(redisContext *c) { static redisAsyncContext *redisAsyncInitialize(redisContext *c) {
redisAsyncContext *ac; redisAsyncContext *ac;
dict *channels = NULL, *patterns = NULL;
ac = realloc(c,sizeof(redisAsyncContext)); channels = dictCreate(&callbackDict,NULL);
if (channels == NULL)
goto oom;
patterns = dictCreate(&callbackDict,NULL);
if (patterns == NULL)
goto oom;
ac = hi_realloc(c,sizeof(redisAsyncContext));
if (ac == NULL) if (ac == NULL)
return NULL; goto oom;
c = &(ac->c); c = &(ac->c);
...@@ -126,9 +141,14 @@ static redisAsyncContext *redisAsyncInitialize(redisContext *c) { ...@@ -126,9 +141,14 @@ static redisAsyncContext *redisAsyncInitialize(redisContext *c) {
ac->replies.tail = NULL; ac->replies.tail = NULL;
ac->sub.invalid.head = NULL; ac->sub.invalid.head = NULL;
ac->sub.invalid.tail = NULL; ac->sub.invalid.tail = NULL;
ac->sub.channels = dictCreate(&callbackDict,NULL); ac->sub.channels = channels;
ac->sub.patterns = dictCreate(&callbackDict,NULL); ac->sub.patterns = patterns;
return ac; return ac;
oom:
if (channels) dictRelease(channels);
if (patterns) dictRelease(patterns);
return NULL;
} }
/* We want the error field to be accessible directly instead of requiring /* We want the error field to be accessible directly instead of requiring
...@@ -216,7 +236,7 @@ static int __redisPushCallback(redisCallbackList *list, redisCallback *source) { ...@@ -216,7 +236,7 @@ static int __redisPushCallback(redisCallbackList *list, redisCallback *source) {
redisCallback *cb; redisCallback *cb;
/* Copy callback from stack to heap */ /* Copy callback from stack to heap */
cb = malloc(sizeof(*cb)); cb = hi_malloc(sizeof(*cb));
if (cb == NULL) if (cb == NULL)
return REDIS_ERR_OOM; return REDIS_ERR_OOM;
...@@ -244,7 +264,7 @@ static int __redisShiftCallback(redisCallbackList *list, redisCallback *target) ...@@ -244,7 +264,7 @@ static int __redisShiftCallback(redisCallbackList *list, redisCallback *target)
/* Copy callback from heap to stack */ /* Copy callback from heap to stack */
if (target != NULL) if (target != NULL)
memcpy(target,cb,sizeof(*cb)); memcpy(target,cb,sizeof(*cb));
free(cb); hi_free(cb);
return REDIS_OK; return REDIS_OK;
} }
return REDIS_ERR; return REDIS_ERR;
...@@ -275,17 +295,27 @@ static void __redisAsyncFree(redisAsyncContext *ac) { ...@@ -275,17 +295,27 @@ static void __redisAsyncFree(redisAsyncContext *ac) {
__redisRunCallback(ac,&cb,NULL); __redisRunCallback(ac,&cb,NULL);
/* Run subscription callbacks callbacks with NULL reply */ /* Run subscription callbacks callbacks with NULL reply */
if (ac->sub.channels) {
it = dictGetIterator(ac->sub.channels); it = dictGetIterator(ac->sub.channels);
if (it != NULL) {
while ((de = dictNext(it)) != NULL) while ((de = dictNext(it)) != NULL)
__redisRunCallback(ac,dictGetEntryVal(de),NULL); __redisRunCallback(ac,dictGetEntryVal(de),NULL);
dictReleaseIterator(it); dictReleaseIterator(it);
}
dictRelease(ac->sub.channels); dictRelease(ac->sub.channels);
}
if (ac->sub.patterns) {
it = dictGetIterator(ac->sub.patterns); it = dictGetIterator(ac->sub.patterns);
if (it != NULL) {
while ((de = dictNext(it)) != NULL) while ((de = dictNext(it)) != NULL)
__redisRunCallback(ac,dictGetEntryVal(de),NULL); __redisRunCallback(ac,dictGetEntryVal(de),NULL);
dictReleaseIterator(it); dictReleaseIterator(it);
}
dictRelease(ac->sub.patterns); dictRelease(ac->sub.patterns);
}
/* Signal event lib to clean up */ /* Signal event lib to clean up */
_EL_CLEANUP(ac); _EL_CLEANUP(ac);
...@@ -388,6 +418,9 @@ static int __redisGetSubscribeCallback(redisAsyncContext *ac, redisReply *reply, ...@@ -388,6 +418,9 @@ static int __redisGetSubscribeCallback(redisAsyncContext *ac, redisReply *reply,
/* Locate the right callback */ /* Locate the right callback */
assert(reply->element[1]->type == REDIS_REPLY_STRING); assert(reply->element[1]->type == REDIS_REPLY_STRING);
sname = sdsnewlen(reply->element[1]->str,reply->element[1]->len); sname = sdsnewlen(reply->element[1]->str,reply->element[1]->len);
if (sname == NULL)
goto oom;
de = dictFind(callbacks,sname); de = dictFind(callbacks,sname);
if (de != NULL) { if (de != NULL) {
cb = dictGetEntryVal(de); cb = dictGetEntryVal(de);
...@@ -421,6 +454,9 @@ static int __redisGetSubscribeCallback(redisAsyncContext *ac, redisReply *reply, ...@@ -421,6 +454,9 @@ static int __redisGetSubscribeCallback(redisAsyncContext *ac, redisReply *reply,
__redisShiftCallback(&ac->sub.invalid,dstcb); __redisShiftCallback(&ac->sub.invalid,dstcb);
} }
return REDIS_OK; return REDIS_OK;
oom:
__redisSetError(&(ac->c), REDIS_ERR_OOM, "Out of memory");
return REDIS_ERR;
} }
void redisProcessCallbacks(redisAsyncContext *ac) { void redisProcessCallbacks(redisAsyncContext *ac) {
...@@ -588,8 +624,6 @@ void redisAsyncHandleWrite(redisAsyncContext *ac) { ...@@ -588,8 +624,6 @@ void redisAsyncHandleWrite(redisAsyncContext *ac) {
c->funcs->async_write(ac); c->funcs->async_write(ac);
} }
void __redisSetError(redisContext *c, int type, const char *str);
void redisAsyncHandleTimeout(redisAsyncContext *ac) { void redisAsyncHandleTimeout(redisAsyncContext *ac) {
redisContext *c = &(ac->c); redisContext *c = &(ac->c);
redisCallback cb; redisCallback cb;
...@@ -672,6 +706,9 @@ static int __redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void ...@@ -672,6 +706,9 @@ static int __redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void
/* Add every channel/pattern to the list of subscription callbacks. */ /* Add every channel/pattern to the list of subscription callbacks. */
while ((p = nextArgument(p,&astr,&alen)) != NULL) { while ((p = nextArgument(p,&astr,&alen)) != NULL) {
sname = sdsnewlen(astr,alen); sname = sdsnewlen(astr,alen);
if (sname == NULL)
goto oom;
if (pvariant) if (pvariant)
cbdict = ac->sub.patterns; cbdict = ac->sub.patterns;
else else
...@@ -715,6 +752,9 @@ static int __redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void ...@@ -715,6 +752,9 @@ static int __redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void
_EL_ADD_WRITE(ac); _EL_ADD_WRITE(ac);
return REDIS_OK; return REDIS_OK;
oom:
__redisSetError(&(ac->c), REDIS_ERR_OOM, "Out of memory");
return REDIS_ERR;
} }
int redisvAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *format, va_list ap) { int redisvAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *format, va_list ap) {
...@@ -728,7 +768,7 @@ int redisvAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdat ...@@ -728,7 +768,7 @@ int redisvAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdat
return REDIS_ERR; return REDIS_ERR;
status = __redisAsyncCommand(ac,fn,privdata,cmd,len); status = __redisAsyncCommand(ac,fn,privdata,cmd,len);
free(cmd); hi_free(cmd);
return status; return status;
} }
...@@ -758,15 +798,21 @@ int redisAsyncFormattedCommand(redisAsyncContext *ac, redisCallbackFn *fn, void ...@@ -758,15 +798,21 @@ int redisAsyncFormattedCommand(redisAsyncContext *ac, redisCallbackFn *fn, void
return status; return status;
} }
void redisAsyncSetTimeout(redisAsyncContext *ac, struct timeval tv) { int redisAsyncSetTimeout(redisAsyncContext *ac, struct timeval tv) {
if (!ac->c.timeout) { if (!ac->c.timeout) {
ac->c.timeout = hi_calloc(1, sizeof(tv)); ac->c.timeout = hi_calloc(1, sizeof(tv));
if (ac->c.timeout == NULL) {
__redisSetError(&ac->c, REDIS_ERR_OOM, "Out of memory");
__redisAsyncCopyError(ac);
return REDIS_ERR;
} }
if (tv.tv_sec == ac->c.timeout->tv_sec &&
tv.tv_usec == ac->c.timeout->tv_usec) {
return;
} }
if (tv.tv_sec != ac->c.timeout->tv_sec ||
tv.tv_usec != ac->c.timeout->tv_usec)
{
*ac->c.timeout = tv; *ac->c.timeout = tv;
}
return REDIS_OK;
} }
...@@ -118,7 +118,7 @@ redisAsyncContext *redisAsyncConnectUnix(const char *path); ...@@ -118,7 +118,7 @@ redisAsyncContext *redisAsyncConnectUnix(const char *path);
int redisAsyncSetConnectCallback(redisAsyncContext *ac, redisConnectCallback *fn); int redisAsyncSetConnectCallback(redisAsyncContext *ac, redisConnectCallback *fn);
int redisAsyncSetDisconnectCallback(redisAsyncContext *ac, redisDisconnectCallback *fn); int redisAsyncSetDisconnectCallback(redisAsyncContext *ac, redisDisconnectCallback *fn);
void redisAsyncSetTimeout(redisAsyncContext *ac, struct timeval tv); int redisAsyncSetTimeout(redisAsyncContext *ac, struct timeval tv);
void redisAsyncDisconnect(redisAsyncContext *ac); void redisAsyncDisconnect(redisAsyncContext *ac);
void redisAsyncFree(redisAsyncContext *ac); void redisAsyncFree(redisAsyncContext *ac);
......
...@@ -73,6 +73,9 @@ static void _dictReset(dict *ht) { ...@@ -73,6 +73,9 @@ static void _dictReset(dict *ht) {
/* Create a new hash table */ /* Create a new hash table */
static dict *dictCreate(dictType *type, void *privDataPtr) { static dict *dictCreate(dictType *type, void *privDataPtr) {
dict *ht = hi_malloc(sizeof(*ht)); dict *ht = hi_malloc(sizeof(*ht));
if (ht == NULL)
return NULL;
_dictInit(ht,type,privDataPtr); _dictInit(ht,type,privDataPtr);
return ht; return ht;
} }
...@@ -98,7 +101,9 @@ static int dictExpand(dict *ht, unsigned long size) { ...@@ -98,7 +101,9 @@ static int dictExpand(dict *ht, unsigned long size) {
_dictInit(&n, ht->type, ht->privdata); _dictInit(&n, ht->type, ht->privdata);
n.size = realsize; n.size = realsize;
n.sizemask = realsize-1; n.sizemask = realsize-1;
n.table = calloc(realsize,sizeof(dictEntry*)); n.table = hi_calloc(realsize,sizeof(dictEntry*));
if (n.table == NULL)
return DICT_ERR;
/* Copy all the elements from the old to the new table: /* Copy all the elements from the old to the new table:
* note that if the old hash table is empty ht->size is zero, * note that if the old hash table is empty ht->size is zero,
...@@ -125,7 +130,7 @@ static int dictExpand(dict *ht, unsigned long size) { ...@@ -125,7 +130,7 @@ static int dictExpand(dict *ht, unsigned long size) {
} }
} }
assert(ht->used == 0); assert(ht->used == 0);
free(ht->table); hi_free(ht->table);
/* Remap the new hashtable in the old */ /* Remap the new hashtable in the old */
*ht = n; *ht = n;
...@@ -144,6 +149,9 @@ static int dictAdd(dict *ht, void *key, void *val) { ...@@ -144,6 +149,9 @@ static int dictAdd(dict *ht, void *key, void *val) {
/* Allocates the memory and stores key */ /* Allocates the memory and stores key */
entry = hi_malloc(sizeof(*entry)); entry = hi_malloc(sizeof(*entry));
if (entry == NULL)
return DICT_ERR;
entry->next = ht->table[index]; entry->next = ht->table[index];
ht->table[index] = entry; ht->table[index] = entry;
...@@ -167,6 +175,9 @@ static int dictReplace(dict *ht, void *key, void *val) { ...@@ -167,6 +175,9 @@ static int dictReplace(dict *ht, void *key, void *val) {
return 1; return 1;
/* It already exists, get the entry */ /* It already exists, get the entry */
entry = dictFind(ht, key); entry = dictFind(ht, key);
if (entry == NULL)
return 0;
/* Free the old value and set the new one */ /* Free the old value and set the new one */
/* Set the new value and free the old one. Note that it is important /* Set the new value and free the old one. Note that it is important
* to do that in this order, as the value may just be exactly the same * to do that in this order, as the value may just be exactly the same
...@@ -200,7 +211,7 @@ static int dictDelete(dict *ht, const void *key) { ...@@ -200,7 +211,7 @@ static int dictDelete(dict *ht, const void *key) {
dictFreeEntryKey(ht,de); dictFreeEntryKey(ht,de);
dictFreeEntryVal(ht,de); dictFreeEntryVal(ht,de);
free(de); hi_free(de);
ht->used--; ht->used--;
return DICT_OK; return DICT_OK;
} }
...@@ -223,13 +234,13 @@ static int _dictClear(dict *ht) { ...@@ -223,13 +234,13 @@ static int _dictClear(dict *ht) {
nextHe = he->next; nextHe = he->next;
dictFreeEntryKey(ht, he); dictFreeEntryKey(ht, he);
dictFreeEntryVal(ht, he); dictFreeEntryVal(ht, he);
free(he); hi_free(he);
ht->used--; ht->used--;
he = nextHe; he = nextHe;
} }
} }
/* Free the table and the allocated cache structure */ /* Free the table and the allocated cache structure */
free(ht->table); hi_free(ht->table);
/* Re-initialize the table */ /* Re-initialize the table */
_dictReset(ht); _dictReset(ht);
return DICT_OK; /* never fails */ return DICT_OK; /* never fails */
...@@ -238,7 +249,7 @@ static int _dictClear(dict *ht) { ...@@ -238,7 +249,7 @@ static int _dictClear(dict *ht) {
/* Clear & Release the hash table */ /* Clear & Release the hash table */
static void dictRelease(dict *ht) { static void dictRelease(dict *ht) {
_dictClear(ht); _dictClear(ht);
free(ht); hi_free(ht);
} }
static dictEntry *dictFind(dict *ht, const void *key) { static dictEntry *dictFind(dict *ht, const void *key) {
...@@ -258,6 +269,8 @@ static dictEntry *dictFind(dict *ht, const void *key) { ...@@ -258,6 +269,8 @@ static dictEntry *dictFind(dict *ht, const void *key) {
static dictIterator *dictGetIterator(dict *ht) { static dictIterator *dictGetIterator(dict *ht) {
dictIterator *iter = hi_malloc(sizeof(*iter)); dictIterator *iter = hi_malloc(sizeof(*iter));
if (iter == NULL)
return NULL;
iter->ht = ht; iter->ht = ht;
iter->index = -1; iter->index = -1;
...@@ -287,7 +300,7 @@ static dictEntry *dictNext(dictIterator *iter) { ...@@ -287,7 +300,7 @@ static dictEntry *dictNext(dictIterator *iter) {
} }
static void dictReleaseIterator(dictIterator *iter) { static void dictReleaseIterator(dictIterator *iter) {
free(iter); hi_free(iter);
} }
/* ------------------------- private functions ------------------------------ */ /* ------------------------- private functions ------------------------------ */
......
...@@ -74,7 +74,7 @@ static redisReplyObjectFunctions defaultFunctions = { ...@@ -74,7 +74,7 @@ static redisReplyObjectFunctions defaultFunctions = {
/* Create a reply object */ /* Create a reply object */
static redisReply *createReplyObject(int type) { static redisReply *createReplyObject(int type) {
redisReply *r = calloc(1,sizeof(*r)); redisReply *r = hi_calloc(1,sizeof(*r));
if (r == NULL) if (r == NULL)
return NULL; return NULL;
...@@ -101,7 +101,7 @@ void freeReplyObject(void *reply) { ...@@ -101,7 +101,7 @@ void freeReplyObject(void *reply) {
if (r->element != NULL) { if (r->element != NULL) {
for (j = 0; j < r->elements; j++) for (j = 0; j < r->elements; j++)
freeReplyObject(r->element[j]); freeReplyObject(r->element[j]);
free(r->element); hi_free(r->element);
} }
break; break;
case REDIS_REPLY_ERROR: case REDIS_REPLY_ERROR:
...@@ -109,10 +109,10 @@ void freeReplyObject(void *reply) { ...@@ -109,10 +109,10 @@ void freeReplyObject(void *reply) {
case REDIS_REPLY_STRING: case REDIS_REPLY_STRING:
case REDIS_REPLY_DOUBLE: case REDIS_REPLY_DOUBLE:
case REDIS_REPLY_VERB: case REDIS_REPLY_VERB:
free(r->str); hi_free(r->str);
break; break;
} }
free(r); hi_free(r);
} }
static void *createStringObject(const redisReadTask *task, char *str, size_t len) { static void *createStringObject(const redisReadTask *task, char *str, size_t len) {
...@@ -130,22 +130,18 @@ static void *createStringObject(const redisReadTask *task, char *str, size_t len ...@@ -130,22 +130,18 @@ static void *createStringObject(const redisReadTask *task, char *str, size_t len
/* Copy string value */ /* Copy string value */
if (task->type == REDIS_REPLY_VERB) { if (task->type == REDIS_REPLY_VERB) {
buf = malloc(len-4+1); /* Skip 4 bytes of verbatim type header. */ buf = hi_malloc(len-4+1); /* Skip 4 bytes of verbatim type header. */
if (buf == NULL) { if (buf == NULL) goto oom;
freeReplyObject(r);
return NULL;
}
memcpy(r->vtype,str,3); memcpy(r->vtype,str,3);
r->vtype[3] = '\0'; r->vtype[3] = '\0';
memcpy(buf,str+4,len-4); memcpy(buf,str+4,len-4);
buf[len-4] = '\0'; buf[len-4] = '\0';
r->len = len - 4; r->len = len - 4;
} else { } else {
buf = malloc(len+1); buf = hi_malloc(len+1);
if (buf == NULL) { if (buf == NULL) goto oom;
freeReplyObject(r);
return NULL;
}
memcpy(buf,str,len); memcpy(buf,str,len);
buf[len] = '\0'; buf[len] = '\0';
r->len = len; r->len = len;
...@@ -161,6 +157,10 @@ static void *createStringObject(const redisReadTask *task, char *str, size_t len ...@@ -161,6 +157,10 @@ static void *createStringObject(const redisReadTask *task, char *str, size_t len
parent->element[task->idx] = r; parent->element[task->idx] = r;
} }
return r; return r;
oom:
freeReplyObject(r);
return NULL;
} }
static void *createArrayObject(const redisReadTask *task, size_t elements) { static void *createArrayObject(const redisReadTask *task, size_t elements) {
...@@ -171,7 +171,7 @@ static void *createArrayObject(const redisReadTask *task, size_t elements) { ...@@ -171,7 +171,7 @@ static void *createArrayObject(const redisReadTask *task, size_t elements) {
return NULL; return NULL;
if (elements > 0) { if (elements > 0) {
r->element = calloc(elements,sizeof(redisReply*)); r->element = hi_calloc(elements,sizeof(redisReply*));
if (r->element == NULL) { if (r->element == NULL) {
freeReplyObject(r); freeReplyObject(r);
return NULL; return NULL;
...@@ -218,7 +218,7 @@ static void *createDoubleObject(const redisReadTask *task, double value, char *s ...@@ -218,7 +218,7 @@ static void *createDoubleObject(const redisReadTask *task, double value, char *s
return NULL; return NULL;
r->dval = value; r->dval = value;
r->str = malloc(len+1); r->str = hi_malloc(len+1);
if (r->str == NULL) { if (r->str == NULL) {
freeReplyObject(r); freeReplyObject(r);
return NULL; return NULL;
...@@ -322,7 +322,7 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) { ...@@ -322,7 +322,7 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) {
if (*c != '%' || c[1] == '\0') { if (*c != '%' || c[1] == '\0') {
if (*c == ' ') { if (*c == ' ') {
if (touched) { if (touched) {
newargv = realloc(curargv,sizeof(char*)*(argc+1)); newargv = hi_realloc(curargv,sizeof(char*)*(argc+1));
if (newargv == NULL) goto memory_err; if (newargv == NULL) goto memory_err;
curargv = newargv; curargv = newargv;
curargv[argc++] = curarg; curargv[argc++] = curarg;
...@@ -471,7 +471,7 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) { ...@@ -471,7 +471,7 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) {
/* Add the last argument if needed */ /* Add the last argument if needed */
if (touched) { if (touched) {
newargv = realloc(curargv,sizeof(char*)*(argc+1)); newargv = hi_realloc(curargv,sizeof(char*)*(argc+1));
if (newargv == NULL) goto memory_err; if (newargv == NULL) goto memory_err;
curargv = newargv; curargv = newargv;
curargv[argc++] = curarg; curargv[argc++] = curarg;
...@@ -487,7 +487,7 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) { ...@@ -487,7 +487,7 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) {
totlen += 1+countDigits(argc)+2; totlen += 1+countDigits(argc)+2;
/* Build the command at protocol level */ /* Build the command at protocol level */
cmd = malloc(totlen+1); cmd = hi_malloc(totlen+1);
if (cmd == NULL) goto memory_err; if (cmd == NULL) goto memory_err;
pos = sprintf(cmd,"*%d\r\n",argc); pos = sprintf(cmd,"*%d\r\n",argc);
...@@ -502,7 +502,7 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) { ...@@ -502,7 +502,7 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) {
assert(pos == totlen); assert(pos == totlen);
cmd[pos] = '\0'; cmd[pos] = '\0';
free(curargv); hi_free(curargv);
*target = cmd; *target = cmd;
return totlen; return totlen;
...@@ -518,11 +518,11 @@ cleanup: ...@@ -518,11 +518,11 @@ cleanup:
if (curargv) { if (curargv) {
while(argc--) while(argc--)
sdsfree(curargv[argc]); sdsfree(curargv[argc]);
free(curargv); hi_free(curargv);
} }
sdsfree(curarg); sdsfree(curarg);
free(cmd); hi_free(cmd);
return error_type; return error_type;
} }
...@@ -563,7 +563,7 @@ int redisFormatCommand(char **target, const char *format, ...) { ...@@ -563,7 +563,7 @@ int redisFormatCommand(char **target, const char *format, ...) {
int redisFormatSdsCommandArgv(sds *target, int argc, const char **argv, int redisFormatSdsCommandArgv(sds *target, int argc, const char **argv,
const size_t *argvlen) const size_t *argvlen)
{ {
sds cmd; sds cmd, aux;
unsigned long long totlen; unsigned long long totlen;
int j; int j;
size_t len; size_t len;
...@@ -585,9 +585,13 @@ int redisFormatSdsCommandArgv(sds *target, int argc, const char **argv, ...@@ -585,9 +585,13 @@ int redisFormatSdsCommandArgv(sds *target, int argc, const char **argv,
return -1; return -1;
/* We already know how much storage we need */ /* We already know how much storage we need */
cmd = sdsMakeRoomFor(cmd, totlen); aux = sdsMakeRoomFor(cmd, totlen);
if (cmd == NULL) if (aux == NULL) {
sdsfree(cmd);
return -1; return -1;
}
cmd = aux;
/* Construct command */ /* Construct command */
cmd = sdscatfmt(cmd, "*%i\r\n", argc); cmd = sdscatfmt(cmd, "*%i\r\n", argc);
...@@ -631,7 +635,7 @@ int redisFormatCommandArgv(char **target, int argc, const char **argv, const siz ...@@ -631,7 +635,7 @@ int redisFormatCommandArgv(char **target, int argc, const char **argv, const siz
} }
/* Build the command at protocol level */ /* Build the command at protocol level */
cmd = malloc(totlen+1); cmd = hi_malloc(totlen+1);
if (cmd == NULL) if (cmd == NULL)
return -1; return -1;
...@@ -652,7 +656,7 @@ int redisFormatCommandArgv(char **target, int argc, const char **argv, const siz ...@@ -652,7 +656,7 @@ int redisFormatCommandArgv(char **target, int argc, const char **argv, const siz
} }
void redisFreeCommand(char *cmd) { void redisFreeCommand(char *cmd) {
free(cmd); hi_free(cmd);
} }
void __redisSetError(redisContext *c, int type, const char *str) { void __redisSetError(redisContext *c, int type, const char *str) {
...@@ -678,7 +682,7 @@ redisReader *redisReaderCreate(void) { ...@@ -678,7 +682,7 @@ redisReader *redisReaderCreate(void) {
static redisContext *redisContextInit(const redisOptions *options) { static redisContext *redisContextInit(const redisOptions *options) {
redisContext *c; redisContext *c;
c = calloc(1, sizeof(*c)); c = hi_calloc(1, sizeof(*c));
if (c == NULL) if (c == NULL)
return NULL; return NULL;
...@@ -702,16 +706,16 @@ void redisFree(redisContext *c) { ...@@ -702,16 +706,16 @@ void redisFree(redisContext *c) {
sdsfree(c->obuf); sdsfree(c->obuf);
redisReaderFree(c->reader); redisReaderFree(c->reader);
free(c->tcp.host); hi_free(c->tcp.host);
free(c->tcp.source_addr); hi_free(c->tcp.source_addr);
free(c->unix_sock.path); hi_free(c->unix_sock.path);
free(c->timeout); hi_free(c->timeout);
free(c->saddr); hi_free(c->saddr);
if (c->funcs->free_privdata) { if (c->funcs->free_privdata) {
c->funcs->free_privdata(c->privdata); c->funcs->free_privdata(c->privdata);
} }
memset(c, 0xff, sizeof(*c)); memset(c, 0xff, sizeof(*c));
free(c); hi_free(c);
} }
redisFD redisFreeKeepFd(redisContext *c) { redisFD redisFreeKeepFd(redisContext *c) {
...@@ -738,6 +742,11 @@ int redisReconnect(redisContext *c) { ...@@ -738,6 +742,11 @@ int redisReconnect(redisContext *c) {
c->obuf = sdsempty(); c->obuf = sdsempty();
c->reader = redisReaderCreate(); c->reader = redisReaderCreate();
if (c->obuf == NULL || c->reader == NULL) {
__redisSetError(c, REDIS_ERR_OOM, "Out of memory");
return REDIS_ERR;
}
if (c->connection_type == REDIS_CONN_TCP) { if (c->connection_type == REDIS_CONN_TCP) {
return redisContextConnectBindTcp(c, c->tcp.host, c->tcp.port, return redisContextConnectBindTcp(c, c->tcp.host, c->tcp.port,
c->timeout, c->tcp.source_addr); c->timeout, c->tcp.source_addr);
...@@ -918,6 +927,8 @@ int redisBufferWrite(redisContext *c, int *done) { ...@@ -918,6 +927,8 @@ int redisBufferWrite(redisContext *c, int *done) {
if (nwritten == (signed)sdslen(c->obuf)) { if (nwritten == (signed)sdslen(c->obuf)) {
sdsfree(c->obuf); sdsfree(c->obuf);
c->obuf = sdsempty(); c->obuf = sdsempty();
if (c->obuf == NULL)
goto oom;
} else { } else {
sdsrange(c->obuf,nwritten,-1); sdsrange(c->obuf,nwritten,-1);
} }
...@@ -925,6 +936,10 @@ int redisBufferWrite(redisContext *c, int *done) { ...@@ -925,6 +936,10 @@ int redisBufferWrite(redisContext *c, int *done) {
} }
if (done != NULL) *done = (sdslen(c->obuf) == 0); if (done != NULL) *done = (sdslen(c->obuf) == 0);
return REDIS_OK; return REDIS_OK;
oom:
__redisSetError(c, REDIS_ERR_OOM, "Out of memory");
return REDIS_ERR;
} }
/* Internal helper function to try and get a reply from the reader, /* Internal helper function to try and get a reply from the reader,
...@@ -1015,11 +1030,11 @@ int redisvAppendCommand(redisContext *c, const char *format, va_list ap) { ...@@ -1015,11 +1030,11 @@ int redisvAppendCommand(redisContext *c, const char *format, va_list ap) {
} }
if (__redisAppendCommand(c,cmd,len) != REDIS_OK) { if (__redisAppendCommand(c,cmd,len) != REDIS_OK) {
free(cmd); hi_free(cmd);
return REDIS_ERR; return REDIS_ERR;
} }
free(cmd); hi_free(cmd);
return REDIS_OK; return REDIS_OK;
} }
......
...@@ -328,6 +328,22 @@ int redisContextSetTimeout(redisContext *c, const struct timeval tv) { ...@@ -328,6 +328,22 @@ int redisContextSetTimeout(redisContext *c, const struct timeval tv) {
return REDIS_OK; return REDIS_OK;
} }
static int _redisContextUpdateTimeout(redisContext *c, const struct timeval *timeout) {
/* Same timeval struct, short circuit */
if (c->timeout == timeout)
return REDIS_OK;
/* Allocate context timeval if we need to */
if (c->timeout == NULL) {
c->timeout = hi_malloc(sizeof(*c->timeout));
if (c->timeout == NULL)
return REDIS_ERR;
}
memcpy(c->timeout, timeout, sizeof(*c->timeout));
return REDIS_OK;
}
static int _redisContextConnectTcp(redisContext *c, const char *addr, int port, static int _redisContextConnectTcp(redisContext *c, const char *addr, int port,
const struct timeval *timeout, const struct timeval *timeout,
const char *source_addr) { const char *source_addr) {
...@@ -352,20 +368,18 @@ static int _redisContextConnectTcp(redisContext *c, const char *addr, int port, ...@@ -352,20 +368,18 @@ static int _redisContextConnectTcp(redisContext *c, const char *addr, int port,
* This is a bit ugly, but atleast it works and doesn't leak memory. * This is a bit ugly, but atleast it works and doesn't leak memory.
**/ **/
if (c->tcp.host != addr) { if (c->tcp.host != addr) {
free(c->tcp.host); hi_free(c->tcp.host);
c->tcp.host = hi_strdup(addr); c->tcp.host = hi_strdup(addr);
if (c->tcp.host == NULL)
goto oom;
} }
if (timeout) { if (timeout) {
if (c->timeout != timeout) { if (_redisContextUpdateTimeout(c, timeout) == REDIS_ERR)
if (c->timeout == NULL) goto oom;
c->timeout = hi_malloc(sizeof(struct timeval));
memcpy(c->timeout, timeout, sizeof(struct timeval));
}
} else { } else {
free(c->timeout); hi_free(c->timeout);
c->timeout = NULL; c->timeout = NULL;
} }
...@@ -375,10 +389,10 @@ static int _redisContextConnectTcp(redisContext *c, const char *addr, int port, ...@@ -375,10 +389,10 @@ static int _redisContextConnectTcp(redisContext *c, const char *addr, int port,
} }
if (source_addr == NULL) { if (source_addr == NULL) {
free(c->tcp.source_addr); hi_free(c->tcp.source_addr);
c->tcp.source_addr = NULL; c->tcp.source_addr = NULL;
} else if (c->tcp.source_addr != source_addr) { } else if (c->tcp.source_addr != source_addr) {
free(c->tcp.source_addr); hi_free(c->tcp.source_addr);
c->tcp.source_addr = hi_strdup(source_addr); c->tcp.source_addr = hi_strdup(source_addr);
} }
...@@ -442,8 +456,11 @@ addrretry: ...@@ -442,8 +456,11 @@ addrretry:
} }
/* For repeat connection */ /* For repeat connection */
free(c->saddr); hi_free(c->saddr);
c->saddr = hi_malloc(p->ai_addrlen); c->saddr = hi_malloc(p->ai_addrlen);
if (c->saddr == NULL)
goto oom;
memcpy(c->saddr, p->ai_addr, p->ai_addrlen); memcpy(c->saddr, p->ai_addr, p->ai_addrlen);
c->addrlen = p->ai_addrlen; c->addrlen = p->ai_addrlen;
...@@ -488,6 +505,8 @@ addrretry: ...@@ -488,6 +505,8 @@ addrretry:
goto error; goto error;
} }
oom:
__redisSetError(c, REDIS_ERR_OOM, "Out of memory");
error: error:
rv = REDIS_ERR; rv = REDIS_ERR;
end: end:
...@@ -521,25 +540,32 @@ int redisContextConnectUnix(redisContext *c, const char *path, const struct time ...@@ -521,25 +540,32 @@ int redisContextConnectUnix(redisContext *c, const char *path, const struct time
return REDIS_ERR; return REDIS_ERR;
c->connection_type = REDIS_CONN_UNIX; c->connection_type = REDIS_CONN_UNIX;
if (c->unix_sock.path != path) if (c->unix_sock.path != path) {
hi_free(c->unix_sock.path);
c->unix_sock.path = hi_strdup(path); c->unix_sock.path = hi_strdup(path);
if (c->unix_sock.path == NULL)
goto oom;
}
if (timeout) { if (timeout) {
if (c->timeout != timeout) { if (_redisContextUpdateTimeout(c, timeout) == REDIS_ERR)
if (c->timeout == NULL) goto oom;
c->timeout = hi_malloc(sizeof(struct timeval));
memcpy(c->timeout, timeout, sizeof(struct timeval));
}
} else { } else {
free(c->timeout); hi_free(c->timeout);
c->timeout = NULL; c->timeout = NULL;
} }
if (redisContextTimeoutMsec(c,&timeout_msec) != REDIS_OK) if (redisContextTimeoutMsec(c,&timeout_msec) != REDIS_OK)
return REDIS_ERR; return REDIS_ERR;
/* Don't leak sockaddr if we're reconnecting */
if (c->saddr) hi_free(c->saddr);
sa = (struct sockaddr_un*)(c->saddr = hi_malloc(sizeof(struct sockaddr_un))); sa = (struct sockaddr_un*)(c->saddr = hi_malloc(sizeof(struct sockaddr_un)));
if (sa == NULL)
goto oom;
c->addrlen = sizeof(struct sockaddr_un); c->addrlen = sizeof(struct sockaddr_un);
sa->sun_family = AF_UNIX; sa->sun_family = AF_UNIX;
strncpy(sa->sun_path, path, sizeof(sa->sun_path) - 1); strncpy(sa->sun_path, path, sizeof(sa->sun_path) - 1);
...@@ -564,4 +590,7 @@ int redisContextConnectUnix(redisContext *c, const char *path, const struct time ...@@ -564,4 +590,7 @@ int redisContextConnectUnix(redisContext *c, const char *path, const struct time
errno = EPROTONOSUPPORT; errno = EPROTONOSUPPORT;
return REDIS_ERR; return REDIS_ERR;
#endif /* _WIN32 */ #endif /* _WIN32 */
oom:
__redisSetError(c, REDIS_ERR_OOM, "Out of memory");
return REDIS_ERR;
} }
...@@ -42,6 +42,7 @@ ...@@ -42,6 +42,7 @@
#include <limits.h> #include <limits.h>
#include <math.h> #include <math.h>
#include "alloc.h"
#include "read.h" #include "read.h"
#include "sds.h" #include "sds.h"
#include "win32.h" #include "win32.h"
...@@ -425,7 +426,7 @@ static int redisReaderGrow(redisReader *r) { ...@@ -425,7 +426,7 @@ static int redisReaderGrow(redisReader *r) {
/* Grow our stack size */ /* Grow our stack size */
newlen = r->tasks + REDIS_READER_STACK_SIZE; newlen = r->tasks + REDIS_READER_STACK_SIZE;
aux = realloc(r->task, sizeof(*r->task) * newlen); aux = hi_realloc(r->task, sizeof(*r->task) * newlen);
if (aux == NULL) if (aux == NULL)
goto oom; goto oom;
...@@ -433,7 +434,7 @@ static int redisReaderGrow(redisReader *r) { ...@@ -433,7 +434,7 @@ static int redisReaderGrow(redisReader *r) {
/* Allocate new tasks */ /* Allocate new tasks */
for (; r->tasks < newlen; r->tasks++) { for (; r->tasks < newlen; r->tasks++) {
r->task[r->tasks] = calloc(1, sizeof(**r->task)); r->task[r->tasks] = hi_calloc(1, sizeof(**r->task));
if (r->task[r->tasks] == NULL) if (r->task[r->tasks] == NULL)
goto oom; goto oom;
} }
...@@ -467,7 +468,9 @@ static int processAggregateItem(redisReader *r) { ...@@ -467,7 +468,9 @@ static int processAggregateItem(redisReader *r) {
root = (r->ridx == 0); root = (r->ridx == 0);
if (elements < -1 || (LLONG_MAX > SIZE_MAX && elements > SIZE_MAX)) { if (elements < -1 || (LLONG_MAX > SIZE_MAX && elements > SIZE_MAX) ||
(r->maxelements > 0 && elements > r->maxelements))
{
__redisReaderSetError(r,REDIS_ERR_PROTOCOL, __redisReaderSetError(r,REDIS_ERR_PROTOCOL,
"Multi-bulk length out of range"); "Multi-bulk length out of range");
return REDIS_ERR; return REDIS_ERR;
...@@ -602,7 +605,7 @@ static int processItem(redisReader *r) { ...@@ -602,7 +605,7 @@ static int processItem(redisReader *r) {
redisReader *redisReaderCreateWithFunctions(redisReplyObjectFunctions *fn) { redisReader *redisReaderCreateWithFunctions(redisReplyObjectFunctions *fn) {
redisReader *r; redisReader *r;
r = calloc(1,sizeof(redisReader)); r = hi_calloc(1,sizeof(redisReader));
if (r == NULL) if (r == NULL)
return NULL; return NULL;
...@@ -610,22 +613,22 @@ redisReader *redisReaderCreateWithFunctions(redisReplyObjectFunctions *fn) { ...@@ -610,22 +613,22 @@ redisReader *redisReaderCreateWithFunctions(redisReplyObjectFunctions *fn) {
if (r->buf == NULL) if (r->buf == NULL)
goto oom; goto oom;
r->task = calloc(REDIS_READER_STACK_SIZE, sizeof(*r->task)); r->task = hi_calloc(REDIS_READER_STACK_SIZE, sizeof(*r->task));
if (r->task == NULL) if (r->task == NULL)
goto oom; goto oom;
for (; r->tasks < REDIS_READER_STACK_SIZE; r->tasks++) { for (; r->tasks < REDIS_READER_STACK_SIZE; r->tasks++) {
r->task[r->tasks] = calloc(1, sizeof(**r->task)); r->task[r->tasks] = hi_calloc(1, sizeof(**r->task));
if (r->task[r->tasks] == NULL) if (r->task[r->tasks] == NULL)
goto oom; goto oom;
} }
r->fn = fn; r->fn = fn;
r->maxbuf = REDIS_READER_MAX_BUF; r->maxbuf = REDIS_READER_MAX_BUF;
r->maxelements = REDIS_READER_MAX_ARRAY_ELEMENTS;
r->ridx = -1; r->ridx = -1;
return r;
return r;
oom: oom:
redisReaderFree(r); redisReaderFree(r);
return NULL; return NULL;
...@@ -640,14 +643,14 @@ void redisReaderFree(redisReader *r) { ...@@ -640,14 +643,14 @@ void redisReaderFree(redisReader *r) {
/* We know r->task[i] is allocatd if i < r->tasks */ /* We know r->task[i] is allocatd if i < r->tasks */
for (int i = 0; i < r->tasks; i++) { for (int i = 0; i < r->tasks; i++) {
free(r->task[i]); hi_free(r->task[i]);
} }
if (r->task) if (r->task)
free(r->task); hi_free(r->task);
sdsfree(r->buf); sdsfree(r->buf);
free(r); hi_free(r);
} }
int redisReaderFeed(redisReader *r, const char *buf, size_t len) { int redisReaderFeed(redisReader *r, const char *buf, size_t len) {
...@@ -663,23 +666,22 @@ int redisReaderFeed(redisReader *r, const char *buf, size_t len) { ...@@ -663,23 +666,22 @@ int redisReaderFeed(redisReader *r, const char *buf, size_t len) {
if (r->len == 0 && r->maxbuf != 0 && sdsavail(r->buf) > r->maxbuf) { if (r->len == 0 && r->maxbuf != 0 && sdsavail(r->buf) > r->maxbuf) {
sdsfree(r->buf); sdsfree(r->buf);
r->buf = sdsempty(); r->buf = sdsempty();
r->pos = 0; if (r->buf == 0) goto oom;
/* r->buf should not be NULL since we just free'd a larger one. */ r->pos = 0;
assert(r->buf != NULL);
} }
newbuf = sdscatlen(r->buf,buf,len); newbuf = sdscatlen(r->buf,buf,len);
if (newbuf == NULL) { if (newbuf == NULL) goto oom;
__redisReaderSetErrorOOM(r);
return REDIS_ERR;
}
r->buf = newbuf; r->buf = newbuf;
r->len = sdslen(r->buf); r->len = sdslen(r->buf);
} }
return REDIS_OK; return REDIS_OK;
oom:
__redisReaderSetErrorOOM(r);
return REDIS_ERR;
} }
int redisReaderGetReply(redisReader *r, void **reply) { int redisReaderGetReply(redisReader *r, void **reply) {
......
...@@ -63,7 +63,11 @@ ...@@ -63,7 +63,11 @@
#define REDIS_REPLY_BIGNUM 13 #define REDIS_REPLY_BIGNUM 13
#define REDIS_REPLY_VERB 14 #define REDIS_REPLY_VERB 14
#define REDIS_READER_MAX_BUF (1024*16) /* Default max unused reader buffer. */ /* Default max unused reader buffer. */
#define REDIS_READER_MAX_BUF (1024*16)
/* Default multi-bulk element limit */
#define REDIS_READER_MAX_ARRAY_ELEMENTS ((1LL<<32) - 1)
#ifdef __cplusplus #ifdef __cplusplus
extern "C" { extern "C" {
...@@ -71,7 +75,7 @@ extern "C" { ...@@ -71,7 +75,7 @@ extern "C" {
typedef struct redisReadTask { typedef struct redisReadTask {
int type; int type;
int elements; /* number of elements in multibulk container */ long long elements; /* number of elements in multibulk container */
int idx; /* index in parent (array) object */ int idx; /* index in parent (array) object */
void *obj; /* holds user-generated value for a read task */ void *obj; /* holds user-generated value for a read task */
struct redisReadTask *parent; /* parent task */ struct redisReadTask *parent; /* parent task */
...@@ -96,6 +100,7 @@ typedef struct redisReader { ...@@ -96,6 +100,7 @@ typedef struct redisReader {
size_t pos; /* Buffer cursor */ size_t pos; /* Buffer cursor */
size_t len; /* Buffer length */ size_t len; /* Buffer length */
size_t maxbuf; /* Max length of unused buffer */ size_t maxbuf; /* Max length of unused buffer */
long long maxelements; /* Max multi-bulk elements */
redisReadTask **task; redisReadTask **task;
int tasks; int tasks;
......
...@@ -30,6 +30,7 @@ ...@@ -30,6 +30,7 @@
* POSSIBILITY OF SUCH DAMAGE. * POSSIBILITY OF SUCH DAMAGE.
*/ */
#include "fmacros.h"
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
...@@ -219,10 +220,7 @@ sds sdsMakeRoomFor(sds s, size_t addlen) { ...@@ -219,10 +220,7 @@ sds sdsMakeRoomFor(sds s, size_t addlen) {
hdrlen = sdsHdrSize(type); hdrlen = sdsHdrSize(type);
if (oldtype==type) { if (oldtype==type) {
newsh = s_realloc(sh, hdrlen+newlen+1); newsh = s_realloc(sh, hdrlen+newlen+1);
if (newsh == NULL) { if (newsh == NULL) return NULL;
s_free(sh);
return NULL;
}
s = (char*)newsh+hdrlen; s = (char*)newsh+hdrlen;
} else { } else {
/* Since the header size changes, need to move the string forward, /* Since the header size changes, need to move the string forward,
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment