First step in decoupling reply callbacks from internals
This commit is contained in:
Pieter Noordhuis
parent
a3a405bcba
commit
298e9325d7
104
hiredis.c
104
hiredis.c
|
|
@ -639,28 +639,10 @@ static redisContext *redisContextInit(redisReplyObjectFunctions *fn) {
|
|||
c->obuf = sdsempty();
|
||||
c->fn = fn == NULL ? &defaultFunctions : fn;
|
||||
c->reader = redisReplyReaderCreate(c->fn);
|
||||
c->callbacks = calloc(sizeof(redisCallback),1);
|
||||
c->clen = 1;
|
||||
c->cpos = 0;
|
||||
return c;
|
||||
}
|
||||
|
||||
void redisDisconnect(redisContext *c) {
|
||||
int i;
|
||||
redisCallback cb;
|
||||
|
||||
/* Non-blocking context: call pending callbacks with the NULL reply */
|
||||
if (!(c->flags & REDIS_BLOCK)) {
|
||||
for (i = 0; i < c->cpos; i++) {
|
||||
cb = c->callbacks[i];
|
||||
if (cb.fn != NULL) {
|
||||
cb.fn(c,NULL,cb.privdata);
|
||||
}
|
||||
}
|
||||
/* Reset callback index */
|
||||
c->cpos = 0;
|
||||
}
|
||||
|
||||
if (c->cbDisconnect.fn != NULL)
|
||||
c->cbDisconnect.fn(c,c->cbDisconnect.privdata);
|
||||
close(c->fd);
|
||||
|
|
@ -679,8 +661,6 @@ void redisFree(redisContext *c) {
|
|||
sdsfree(c->error);
|
||||
if (c->obuf != NULL)
|
||||
sdsfree(c->obuf);
|
||||
if (c->clen > 0)
|
||||
free(c->callbacks);
|
||||
redisReplyReaderFree(c->reader);
|
||||
free(c);
|
||||
}
|
||||
|
|
@ -759,38 +739,6 @@ int redisGetReply(redisContext *c, void **reply) {
|
|||
return REDIS_OK;
|
||||
}
|
||||
|
||||
static void redisPopCallback(redisContext *c) {
|
||||
assert(c->cpos > 0);
|
||||
if (c->cpos > 1)
|
||||
memmove(&c->callbacks[0],&c->callbacks[1],(c->cpos-1)*sizeof(redisCallback));
|
||||
c->cpos--;
|
||||
}
|
||||
|
||||
int redisProcessCallbacks(redisContext *c) {
|
||||
void *reply = NULL;
|
||||
redisCallback cb;
|
||||
|
||||
/* Continue while there are callbacks */
|
||||
while(c->cpos > 0) {
|
||||
cb = c->callbacks[0];
|
||||
if (redisGetReply(c,&reply) == REDIS_ERR)
|
||||
return REDIS_ERR;
|
||||
|
||||
if (reply != NULL) {
|
||||
redisPopCallback(c);
|
||||
if (cb.fn != NULL) {
|
||||
cb.fn(c,reply,cb.privdata);
|
||||
} else {
|
||||
c->fn->freeObject(reply);
|
||||
}
|
||||
} else {
|
||||
/* Stop trying */
|
||||
break;
|
||||
}
|
||||
}
|
||||
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
|
||||
|
|
@ -853,24 +801,10 @@ static int redisCommandWriteBlock(redisContext *c, void **reply, char *str, size
|
|||
return REDIS_OK;
|
||||
}
|
||||
|
||||
static int redisCommandWriteNonBlock(redisContext *c, redisCallback *cb, char *str, size_t len) {
|
||||
static int redisCommandWriteNonBlock(redisContext *c, char *str, size_t len) {
|
||||
assert(!(c->flags & REDIS_BLOCK));
|
||||
c->obuf = sdscatlen(c->obuf,str,len);
|
||||
|
||||
/* Make sure there is space for the callback. */
|
||||
assert(c->cpos <= c->clen);
|
||||
if (c->cpos == c->clen) {
|
||||
c->clen++;
|
||||
c->callbacks = realloc(c->callbacks,c->clen*sizeof(redisCallback));
|
||||
}
|
||||
|
||||
if (cb != NULL) {
|
||||
c->callbacks[c->cpos] = *cb;
|
||||
} else {
|
||||
memset(&c->callbacks[c->cpos],0,sizeof(redisCallback));
|
||||
}
|
||||
c->cpos++;
|
||||
|
||||
/* Fire write callback */
|
||||
if (c->cbCommand.fn != NULL)
|
||||
c->cbCommand.fn(c,c->cbCommand.privdata);
|
||||
|
|
@ -880,13 +814,12 @@ static int redisCommandWriteNonBlock(redisContext *c, redisCallback *cb, char *s
|
|||
|
||||
/* 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 a
|
||||
* NULL callback will be appended to the list of callbacks.
|
||||
* 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, provided that
|
||||
* the reply build functions did not return NULL when building the reply,
|
||||
* the error field in the context will be set. */
|
||||
* otherwise. When NULL is returned in a blocking context, the error field
|
||||
* in the context will be set. */
|
||||
void *redisCommand(redisContext *c, const char *format, ...) {
|
||||
va_list ap;
|
||||
char *cmd;
|
||||
|
|
@ -902,33 +835,8 @@ void *redisCommand(redisContext *c, const char *format, ...) {
|
|||
return reply;
|
||||
}
|
||||
} else {
|
||||
redisCommandWriteNonBlock(c,NULL,cmd,len);
|
||||
redisCommandWriteNonBlock(c,cmd,len);
|
||||
}
|
||||
free(cmd);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
/* Write a formatted command to the output buffer. Registers the provided
|
||||
* callback function and argument in the callback list.
|
||||
*
|
||||
* Always returns NULL. In a non-blocking context this will never fail because
|
||||
* this function does not do any I/O. In a blocking context this function will
|
||||
* have no effect (a callback in a blocking context makes no sense). */
|
||||
void *redisCommandWithCallback(redisContext *c, redisCallbackFn *fn, void *privdata, const char *format, ...) {
|
||||
va_list ap;
|
||||
char *cmd;
|
||||
int len;
|
||||
int status;
|
||||
redisCallback cb = { fn, privdata };
|
||||
|
||||
/* This function may only be used in a non-blocking context. */
|
||||
if (c->flags & REDIS_BLOCK) return NULL;
|
||||
|
||||
va_start(ap,format);
|
||||
len = redisvFormatCommand(&cmd,format,ap);
|
||||
va_end(ap);
|
||||
|
||||
status = redisCommandWriteNonBlock(c,&cb,cmd,len);
|
||||
free(cmd);
|
||||
return NULL;
|
||||
}
|
||||
|
|
|
|||
28
hiredis.h
28
hiredis.h
|
|
@ -78,20 +78,12 @@ struct redisContext; /* need forward declaration of redisContext */
|
|||
/* Callbacks triggered on non-reply events. */
|
||||
typedef void (redisContextCallbackFn)(struct redisContext*, void*);
|
||||
|
||||
/* Reply callback prototype and container */
|
||||
typedef void (redisCallbackFn)(struct redisContext*, redisReply*, void*);
|
||||
|
||||
/* Callback containers */
|
||||
typedef struct redisContextCallback {
|
||||
redisContextCallbackFn *fn;
|
||||
void *privdata;
|
||||
} redisContextCallback;
|
||||
|
||||
typedef struct redisCallback {
|
||||
redisCallbackFn *fn;
|
||||
void *privdata;
|
||||
} redisCallback;
|
||||
|
||||
/* Context for a connection to Redis */
|
||||
typedef struct redisContext {
|
||||
int fd;
|
||||
|
|
@ -107,11 +99,6 @@ typedef struct redisContext {
|
|||
redisContextCallback cbDisconnect;
|
||||
redisContextCallback cbCommand;
|
||||
redisContextCallback cbFree;
|
||||
|
||||
/* Reply callbacks */
|
||||
redisCallback *callbacks;
|
||||
int cpos;
|
||||
int clen;
|
||||
} redisContext;
|
||||
|
||||
void freeReplyObject(void *reply);
|
||||
|
|
@ -134,7 +121,6 @@ void redisFree(redisContext *c);
|
|||
int redisBufferRead(redisContext *c);
|
||||
int redisBufferWrite(redisContext *c, int *done);
|
||||
int redisGetReply(redisContext *c, void **reply);
|
||||
int redisProcessCallbacks(redisContext *c);
|
||||
|
||||
/* The disconnect callback is called *immediately* when redisDisconnect()
|
||||
* is called. It is called only once for every redisContext (since hiredis
|
||||
|
|
@ -154,19 +140,7 @@ void redisSetFreeCallback(redisContext *c, redisContextCallbackFn *fn, void *pri
|
|||
* an error occurs, it returns NULL and you should read redisContext->error
|
||||
* to find out what's wrong. In a non-blocking context, it has the same effect
|
||||
* as calling redisCommandWithCallback() with a NULL callback, and will always
|
||||
* return NULL.
|
||||
*
|
||||
* Note: using a NULL reply for an error might conflict with custom reply
|
||||
* reader functions that have NULL as a valid return value (e.g. for the nil
|
||||
* return value). Therefore, it is recommended never to return NULL from your
|
||||
* custom reply object functions. */
|
||||
* return NULL. */
|
||||
void *redisCommand(redisContext *c, const char *format, ...);
|
||||
|
||||
/* Issue a command to Redis from a non-blocking context. The formatted command
|
||||
* is appended to the write buffer and the provided callback is registered.
|
||||
*
|
||||
* Note: when called with a blocking context, this function will not do
|
||||
* anything and immediately returns NULL. */
|
||||
void *redisCommandWithCallback(redisContext *c, redisCallbackFn *fn, void *privdata, const char *format, ...);
|
||||
|
||||
#endif
|
||||
|
|
|
|||
198
test.c
198
test.c
|
|
@ -236,110 +236,110 @@ static void cleanup() {
|
|||
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);
|
||||
}
|
||||
// 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(void) {
|
||||
test_format_commands();
|
||||
test_blocking_connection();
|
||||
test_reply_reader();
|
||||
test_nonblocking_connection();
|
||||
// test_nonblocking_connection();
|
||||
test_throughput();
|
||||
cleanup();
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue