diff options
| -rw-r--r-- | .github/workflows/build.yml | 7 | ||||
| -rw-r--r-- | README.md | 99 | ||||
| -rw-r--r-- | async.c | 121 | ||||
| -rw-r--r-- | async.h | 2 | ||||
| -rw-r--r-- | net.c | 20 | ||||
| -rw-r--r-- | read.c | 6 | ||||
| -rw-r--r-- | sds.c | 15 | ||||
| -rw-r--r-- | sockcompat.c | 19 | ||||
| -rw-r--r-- | test.c | 25 | ||||
| -rwxr-xr-x | test.sh | 9 |
10 files changed, 256 insertions, 67 deletions
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 362bc77..7859ff4 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -14,7 +14,8 @@ jobs: - name: Install dependencies run: | - sudo add-apt-repository -y ppa:chris-lea/redis-server + curl -fsSL https://packages.redis.io/gpg | sudo gpg --dearmor -o /usr/share/keyrings/redis-archive-keyring.gpg + echo "deb [signed-by=/usr/share/keyrings/redis-archive-keyring.gpg] https://packages.redis.io/deb $(lsb_release -cs) main" | sudo tee /etc/apt/sources.list.d/redis.list sudo apt-get update sudo apt-get install -y redis-server valgrind libevent-dev @@ -151,7 +152,8 @@ jobs: - name: Install dependencies run: | - brew install openssl redis + brew install openssl redis@6.2 + brew link redis@6.2 --force - name: Build hiredis run: USE_SSL=1 make @@ -197,6 +199,7 @@ jobs: HIREDIS_PATH: ${{ github.workspace }} run: | build_hiredis() { + git config --global --add safe.directory "$(cygpath -u $HIREDIS_PATH)" cd $(cygpath -u $HIREDIS_PATH) git clean -xfd make @@ -320,23 +320,48 @@ Redis. It returns a pointer to the newly created `redisAsyncContext` struct. The 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. +In case of error, it is the caller's responsibility to free the context using `redisAsyncFree()` *Note: A `redisAsyncContext` is not thread-safe.* +An application function creating a connection might look like this: + ```c -redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379); -if (c->err) { - printf("Error: %s\n", c->errstr); - // handle error +void appConnect(myAppData *appData) +{ + redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379); + if (c->err) { + printf("Error: %s\n", c->errstr); + // handle error + redisAsyncFree(c); + c = NULL; + } else { + appData->context = c; + appData->connecting = 1; + c->data = appData; /* store application pointer for the callbacks */ + redisAsyncSetConnectCallback(c, appOnConnect); + redisAsyncSetDisconnectCallback(c, appOnDisconnect); + } } + ``` -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 + +The asynchronous context _should_ hold a *connect* callback function that is called when the connection +attempt completes, either successfully or with an error. +It _can_ also hold a *disconnect* callback function that is called when the +connection is disconnected (either because of an error or per user request). Both callbacks should have the following prototype: ```c void(const redisAsyncContext *c, int status); ``` + +On a *connect*, the `status` argument is set to `REDIS_OK` if the connection attempt succeeded. In this +case, the context is ready to accept commands. If it is called with `REDIS_ERR` then the +connection attempt failed. The `err` field in the context can be accessed to find out the cause of the error. +After a failed connection attempt, the context object is automatically freed by the libary after calling +the connect callback. This may be a good point to create a new context and retry the connection. + 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 `err` field in the context can be accessed to find out the cause of the error. @@ -344,12 +369,45 @@ field in the context can be accessed to find out the cause of the error. The context object is always freed 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: +Setting the connect or disconnect callbacks can only be done once per context. For subsequent calls the +api will return `REDIS_ERR`. The function to set the callbacks have the following prototype: ```c +int redisAsyncSetConnectCallback(redisAsyncContext *ac, redisConnectCallback *fn); int redisAsyncSetDisconnectCallback(redisAsyncContext *ac, redisDisconnectCallback *fn); ``` -`ac->data` may be used to pass user data to this callback, the same can be done for redisConnectCallback. +`ac->data` may be used to pass user data to both of thes callbacks. An typical implementation +might look something like this: +```c +void appOnConnect(redisAsyncContext *c, int status) +{ + myAppData *appData = (myAppData*)c->data; /* get my application specific context*/ + appData->connecting = 0; + if (status == REDIS_OK) { + appData->connected = 1; + } else { + appData->connected = 0; + appData->err = c->err; + appData->context = NULL; /* avoid stale pointer when callback returns */ + } + appAttemptReconnect(); +} + +void appOnDisconnect(redisAsyncContext *c, int status) +{ + myAppData *appData = (myAppData*)c->data; /* get my application specific context*/ + appData->connected = 0; + appData->err = c->err; + appData->context = NULL; /* avoid stale pointer when callback returns */ + if (status == REDIS_OK) { + appNotifyDisconnectCompleted(mydata); + } else { + appNotifyUnexpectedDisconnect(mydata); + appAttemptReconnect(); + } +} +``` + + ### Sending commands and their callbacks In an asynchronous context, commands are automatically pipelined due to the nature of an event loop. @@ -382,6 +440,14 @@ valid for the duration of the callback. All pending callbacks are called with a `NULL` reply when the context encountered an error. +For every command issued, with the exception of **SUBSCRIBE** and **PSUBSCRIBE**, the callback is +called exactly once. Even if the context object id disconnected or deleted, every pending callback +will be called with a `NULL` reply. + +For **SUBSCRIBE** and **PSUBSCRIBE**, the callbacks may be called repeatedly until a `unsubscribe` +message arrives. This will be the last invocation of the callback. In case of error, the callbacks +may reive a final `NULL` reply instead. + ### Disconnecting An asynchronous connection can be terminated using: @@ -394,6 +460,15 @@ have been written to the socket, their respective replies have been read and the callbacks have been executed. After this, the disconnection callback is executed with the `REDIS_OK` status and the context object is freed. +The connection can be forcefully disconnected using +```c +void redisAsyncFree(redisAsyncContext *ac); +``` +In this case, nothing more is written to the socket, all pending callbacks are called with a `NULL` +reply and the disconnection callback is called with `REDIS_OK`, after which the context object +is freed. + + ### 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. @@ -549,9 +624,9 @@ ssl_context = redisCreateSSLContext( if(ssl_context == NULL || ssl_error != 0) { /* Handle error and abort... */ - /* e.g. - printf("SSL error: %s\n", - (ssl_error != 0) ? + /* e.g. + printf("SSL error: %s\n", + (ssl_error != 0) ? redisSSLContextGetError(ssl_error) : "Unknown error"); // Abort */ @@ -148,6 +148,7 @@ static redisAsyncContext *redisAsyncInitialize(redisContext *c) { ac->sub.replies.tail = NULL; ac->sub.channels = channels; ac->sub.patterns = patterns; + ac->sub.pending_unsubs = 0; return ac; oom: @@ -411,11 +412,11 @@ void redisAsyncDisconnect(redisAsyncContext *ac) { static int __redisGetSubscribeCallback(redisAsyncContext *ac, redisReply *reply, redisCallback *dstcb) { redisContext *c = &(ac->c); dict *callbacks; - redisCallback *cb; + redisCallback *cb = NULL; dictEntry *de; int pvariant; char *stype; - sds sname; + sds sname = NULL; /* Match reply with the expected format of a pushed message. * The type and number of elements (3 to 4) are specified at: @@ -432,42 +433,43 @@ static int __redisGetSubscribeCallback(redisAsyncContext *ac, redisReply *reply, callbacks = ac->sub.channels; /* Locate the right callback */ - assert(reply->element[1]->type == REDIS_REPLY_STRING); - sname = sdsnewlen(reply->element[1]->str,reply->element[1]->len); - if (sname == NULL) - goto oom; + if (reply->element[1]->type == REDIS_REPLY_STRING) { + sname = sdsnewlen(reply->element[1]->str,reply->element[1]->len); + if (sname == NULL) goto oom; - de = dictFind(callbacks,sname); - if (de != NULL) { - cb = dictGetEntryVal(de); - - /* If this is an subscribe reply decrease pending counter. */ - if (strcasecmp(stype+pvariant,"subscribe") == 0) { - cb->pending_subs -= 1; + if ((de = dictFind(callbacks,sname)) != NULL) { + cb = dictGetEntryVal(de); + memcpy(dstcb,cb,sizeof(*dstcb)); } + } - memcpy(dstcb,cb,sizeof(*dstcb)); - - /* If this is an unsubscribe message, remove it. */ - if (strcasecmp(stype+pvariant,"unsubscribe") == 0) { - if (cb->pending_subs == 0) - dictDelete(callbacks,sname); - - /* If this was the last unsubscribe message, revert to - * non-subscribe mode. */ - assert(reply->element[2]->type == REDIS_REPLY_INTEGER); - - /* Unset subscribed flag only when no pipelined pending subscribe. */ - if (reply->element[2]->integer == 0 - && dictSize(ac->sub.channels) == 0 - && dictSize(ac->sub.patterns) == 0) { - c->flags &= ~REDIS_SUBSCRIBED; - - /* Move ongoing regular command callbacks. */ - redisCallback cb; - while (__redisShiftCallback(&ac->sub.replies,&cb) == REDIS_OK) { - __redisPushCallback(&ac->replies,&cb); - } + /* If this is an subscribe reply decrease pending counter. */ + if (strcasecmp(stype+pvariant,"subscribe") == 0) { + assert(cb != NULL); + cb->pending_subs -= 1; + + } else if (strcasecmp(stype+pvariant,"unsubscribe") == 0) { + if (cb == NULL) + ac->sub.pending_unsubs -= 1; + else if (cb->pending_subs == 0) + dictDelete(callbacks,sname); + + /* If this was the last unsubscribe message, revert to + * non-subscribe mode. */ + assert(reply->element[2]->type == REDIS_REPLY_INTEGER); + + /* Unset subscribed flag only when no pipelined pending subscribe + * or pending unsubscribe replies. */ + if (reply->element[2]->integer == 0 + && dictSize(ac->sub.channels) == 0 + && dictSize(ac->sub.patterns) == 0 + && ac->sub.pending_unsubs == 0) { + c->flags &= ~REDIS_SUBSCRIBED; + + /* Move ongoing regular command callbacks. */ + redisCallback cb; + while (__redisShiftCallback(&ac->sub.replies,&cb) == REDIS_OK) { + __redisPushCallback(&ac->replies,&cb); } } } @@ -540,7 +542,7 @@ void redisProcessCallbacks(redisAsyncContext *ac) { /* Even if the context is subscribed, pending regular * callbacks will get a reply before pub/sub messages arrive. */ - redisCallback cb = {NULL, NULL, 0, NULL}; + redisCallback cb = {NULL, NULL, 0, 0, NULL}; if (__redisShiftCallback(&ac->replies,&cb) != REDIS_OK) { /* * A spontaneous reply in a not-subscribed context can be the error @@ -757,6 +759,7 @@ static int __redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void redisContext *c = &(ac->c); redisCallback cb; struct dict *cbdict; + dictIterator it; dictEntry *de; redisCallback *existcb; int pvariant, hasnext; @@ -773,6 +776,7 @@ static int __redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void cb.fn = fn; cb.privdata = privdata; cb.pending_subs = 1; + cb.unsubscribe_sent = 0; /* Find out which command will be appended. */ p = nextArgument(cmd,&cstr,&clen); @@ -812,6 +816,51 @@ static int __redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void * subscribed to one or more channels or patterns. */ if (!(c->flags & REDIS_SUBSCRIBED)) return REDIS_ERR; + if (pvariant) + cbdict = ac->sub.patterns; + else + cbdict = ac->sub.channels; + + if (hasnext) { + /* Send an unsubscribe with specific channels/patterns. + * Bookkeeping the number of expected replies */ + while ((p = nextArgument(p,&astr,&alen)) != NULL) { + sname = sdsnewlen(astr,alen); + if (sname == NULL) + goto oom; + + de = dictFind(cbdict,sname); + if (de != NULL) { + existcb = dictGetEntryVal(de); + if (existcb->unsubscribe_sent == 0) + existcb->unsubscribe_sent = 1; + else + /* Already sent, reply to be ignored */ + ac->sub.pending_unsubs += 1; + } else { + /* Not subscribed to, reply to be ignored */ + ac->sub.pending_unsubs += 1; + } + sdsfree(sname); + } + } else { + /* Send an unsubscribe without specific channels/patterns. + * Bookkeeping the number of expected replies */ + int no_subs = 1; + dictInitIterator(&it,cbdict); + while ((de = dictNext(&it)) != NULL) { + existcb = dictGetEntryVal(de); + if (existcb->unsubscribe_sent == 0) { + existcb->unsubscribe_sent = 1; + no_subs = 0; + } + } + /* Unsubscribing to all channels/patterns, where none is + * subscribed to, results in a single reply to be ignored. */ + if (no_subs == 1) + ac->sub.pending_unsubs += 1; + } + /* (P)UNSUBSCRIBE does not have its own response: every channel or * pattern that is unsubscribed will receive a message. This means we * should not append a callback function for this command. */ @@ -46,6 +46,7 @@ typedef struct redisCallback { struct redisCallback *next; /* simple singly linked list */ redisCallbackFn *fn; int pending_subs; + int unsubscribe_sent; void *privdata; } redisCallback; @@ -105,6 +106,7 @@ typedef struct redisAsyncContext { redisCallbackList replies; struct dict *channels; struct dict *patterns; + int pending_unsubs; } sub; /* Any configured RESP3 PUSH handler */ @@ -277,12 +277,28 @@ int redisCheckConnectDone(redisContext *c, int *completed) { *completed = 1; return REDIS_OK; } - switch (errno) { + int error = errno; + if (error == EINPROGRESS) { + /* must check error to see if connect failed. Get the socket error */ + int fail, so_error; + socklen_t optlen = sizeof(so_error); + fail = getsockopt(c->fd, SOL_SOCKET, SO_ERROR, &so_error, &optlen); + if (fail == 0) { + if (so_error == 0) { + /* Socket is connected! */ + *completed = 1; + return REDIS_OK; + } + /* connection error; */ + errno = so_error; + error = so_error; + } + } + switch (error) { case EISCONN: *completed = 1; return REDIS_OK; case EALREADY: - case EINPROGRESS: case EWOULDBLOCK: *completed = 0; return REDIS_OK; @@ -374,7 +374,7 @@ static int processLineItem(redisReader *r) { if (r->fn && r->fn->createString) obj = r->fn->createString(cur,p,len); else - obj = (void*)(size_t)(cur->type); + obj = (void*)(uintptr_t)(cur->type); } if (obj == NULL) { @@ -439,7 +439,7 @@ static int processBulkItem(redisReader *r) { if (r->fn && r->fn->createString) obj = r->fn->createString(cur,s+2,len); else - obj = (void*)(long)cur->type; + obj = (void*)(uintptr_t)cur->type; success = 1; } } @@ -536,7 +536,7 @@ static int processAggregateItem(redisReader *r) { if (r->fn && r->fn->createArray) obj = r->fn->createArray(cur,elements); else - obj = (void*)(long)cur->type; + obj = (void*)(uintptr_t)cur->type; if (obj == NULL) { __redisReaderSetErrorOOM(r); @@ -90,6 +90,7 @@ sds sdsnewlen(const void *init, size_t initlen) { int hdrlen = sdsHdrSize(type); unsigned char *fp; /* flags pointer. */ + if (hdrlen+initlen+1 <= initlen) return NULL; /* Catch size_t overflow */ sh = s_malloc(hdrlen+initlen+1); if (sh == NULL) return NULL; if (!init) @@ -174,7 +175,7 @@ void sdsfree(sds s) { * the output will be "6" as the string was modified but the logical length * remains 6 bytes. */ void sdsupdatelen(sds s) { - int reallen = strlen(s); + size_t reallen = strlen(s); sdssetlen(s, reallen); } @@ -196,7 +197,7 @@ void sdsclear(sds s) { sds sdsMakeRoomFor(sds s, size_t addlen) { void *sh, *newsh; size_t avail = sdsavail(s); - size_t len, newlen; + size_t len, newlen, reqlen; char type, oldtype = s[-1] & SDS_TYPE_MASK; int hdrlen; @@ -205,7 +206,8 @@ sds sdsMakeRoomFor(sds s, size_t addlen) { len = sdslen(s); sh = (char*)s-sdsHdrSize(oldtype); - newlen = (len+addlen); + reqlen = newlen = (len+addlen); + if (newlen <= len) return NULL; /* Catch size_t overflow */ if (newlen < SDS_MAX_PREALLOC) newlen *= 2; else @@ -219,6 +221,7 @@ sds sdsMakeRoomFor(sds s, size_t addlen) { if (type == SDS_TYPE_5) type = SDS_TYPE_8; hdrlen = sdsHdrSize(type); + if (hdrlen+newlen+1 <= reqlen) return NULL; /* Catch size_t overflow */ if (oldtype==type) { newsh = s_realloc(sh, hdrlen+newlen+1); if (newsh == NULL) return NULL; @@ -580,7 +583,7 @@ sds sdscatprintf(sds s, const char *fmt, ...) { */ sds sdscatfmt(sds s, char const *fmt, ...) { const char *f = fmt; - int i; + long i; va_list ap; va_start(ap,fmt); @@ -755,14 +758,14 @@ int sdsrange(sds s, ssize_t start, ssize_t end) { /* Apply tolower() to every character of the sds string 's'. */ void sdstolower(sds s) { - int len = sdslen(s), j; + size_t len = sdslen(s), j; for (j = 0; j < len; j++) s[j] = tolower(s[j]); } /* Apply toupper() to every character of the sds string 's'. */ void sdstoupper(sds s) { - int len = sdslen(s), j; + size_t len = sdslen(s), j; for (j = 0; j < len; j++) s[j] = toupper(s[j]); } diff --git a/sockcompat.c b/sockcompat.c index f99d14b..31df325 100644 --- a/sockcompat.c +++ b/sockcompat.c @@ -180,10 +180,17 @@ int win32_connect(SOCKET sockfd, const struct sockaddr *addr, socklen_t addrlen) /* For Winsock connect(), the WSAEWOULDBLOCK error means the same thing as * EINPROGRESS for POSIX connect(), so we do that translation to keep POSIX - * logic consistent. */ - if (errno == EWOULDBLOCK) { + * logic consistent. + * Additionally, WSAALREADY is can be reported as WSAEINVAL to and this is + * translated to EIO. Convert appropriately + */ + int err = errno; + if (err == EWOULDBLOCK) { errno = EINPROGRESS; } + else if (err == EIO) { + errno = EALREADY; + } return ret != SOCKET_ERROR ? ret : -1; } @@ -205,6 +212,14 @@ int win32_getsockopt(SOCKET sockfd, int level, int optname, void *optval, sockle } else { ret = getsockopt(sockfd, level, optname, (char*)optval, optlen); } + if (ret != SOCKET_ERROR && level == SOL_SOCKET && optname == SO_ERROR) { + /* translate SO_ERROR codes, if non-zero */ + int err = *(int*)optval; + if (err != 0) { + err = _wsaErrorToErrno(err); + *(int*)optval = err; + } + } _updateErrno(ret != SOCKET_ERROR); return ret != SOCKET_ERROR ? ret : -1; } @@ -914,6 +914,11 @@ static void test_resp3_push_handler(redisContext *c) { old = redisSetPushCallback(c, push_handler); test("We can set a custom RESP3 PUSH handler: "); reply = redisCommand(c, "SET key:0 val:0"); + /* We need another command because depending on the version of Redis, the + * notification may be delivered after the command's reply. */ + test_cond(reply != NULL); + freeReplyObject(reply); + reply = redisCommand(c, "PING"); test_cond(reply != NULL && reply->type == REDIS_REPLY_STATUS && pc.str == 1); freeReplyObject(reply); @@ -929,6 +934,12 @@ static void test_resp3_push_handler(redisContext *c) { assert((reply = redisCommand(c, "GET key:0")) != NULL); freeReplyObject(reply); assert((reply = redisCommand(c, "SET key:0 invalid")) != NULL); + /* Depending on Redis version, we may receive either push notification or + * status reply. Both cases are valid. */ + if (reply->type == REDIS_REPLY_STATUS) { + freeReplyObject(reply); + reply = redisCommand(c, "PING"); + } test_cond(reply->type == REDIS_REPLY_PUSH); freeReplyObject(reply); @@ -1729,10 +1740,14 @@ void subscribe_channel_a_cb(redisAsyncContext *ac, void *r, void *privdata) { strcmp(reply->element[2]->str,"Hello!") == 0); state->checkpoint++; - /* Unsubscribe to channels, including a channel X which we don't subscribe to */ + /* Unsubscribe to channels, including channel X & Z which we don't subscribe to */ redisAsyncCommand(ac,unexpected_cb, (void*)"unsubscribe should not call unexpected_cb()", - "unsubscribe B X A"); + "unsubscribe B X A A Z"); + /* Unsubscribe to patterns, none which we subscribe to */ + redisAsyncCommand(ac,unexpected_cb, + (void*)"punsubscribe should not call unexpected_cb()", + "punsubscribe"); /* Send a regular command after unsubscribing, then disconnect */ state->disconnect = 1; redisAsyncCommand(ac,integer_cb,state,"LPUSH mylist foo"); @@ -1767,8 +1782,10 @@ void subscribe_channel_b_cb(redisAsyncContext *ac, void *r, void *privdata) { /* Test handling of multiple channels * - subscribe to channel A and B - * - a published message on A triggers an unsubscribe of channel B, X and A - * where channel X is not subscribed to. + * - a published message on A triggers an unsubscribe of channel B, X, A and Z + * where channel X and Z are not subscribed to. + * - the published message also triggers an unsubscribe to patterns. Since no + * pattern is subscribed to the responded pattern element type is NIL. * - a command sent after unsubscribe triggers a disconnect */ static void test_pubsub_multiple_channels(struct config config) { test("Subscribe to multiple channels: "); @@ -5,9 +5,16 @@ REDIS_PORT=${REDIS_PORT:-56379} REDIS_SSL_PORT=${REDIS_SSL_PORT:-56443} TEST_SSL=${TEST_SSL:-0} SKIPS_AS_FAILS=${SKIPS_AS_FAILS-:0} +ENABLE_DEBUG_CMD= SSL_TEST_ARGS= SKIPS_ARG= +# We need to enable the DEBUG command for redis-server >= 7.0.0 +REDIS_MAJOR_VERSION="$(redis-server --version|awk -F'[^0-9]+' '{ print $2 }')" +if [ "$REDIS_MAJOR_VERSION" -gt "6" ]; then + ENABLE_DEBUG_CMD="enable-debug-command local" +fi + tmpdir=$(mktemp -d) PID_FILE=${tmpdir}/hiredis-test-redis.pid SOCK_FILE=${tmpdir}/hiredis-test-redis.sock @@ -49,8 +56,10 @@ cleanup() { } trap cleanup INT TERM EXIT + cat > ${tmpdir}/redis.conf <<EOF daemonize yes +${ENABLE_DEBUG_CMD} pidfile ${PID_FILE} port ${REDIS_PORT} bind 127.0.0.1 |