X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/b19c33d48a3b5456b87f84c85fdd85592e0c0562..ba55932674cb83c7854344096e778e36394fcaba:/src/networking.c diff --git a/src/networking.c b/src/networking.c index 46d49bf6..26ee46f6 100644 --- a/src/networking.c +++ b/src/networking.c @@ -41,8 +41,10 @@ redisClient *createClient(int fd) { c->reply = listCreate(); listSetFreeMethod(c->reply,decrRefCount); listSetDupMethod(c->reply,dupClientReplyValue); - c->blocking_keys = NULL; - c->blocking_keys_num = 0; + c->bpop.keys = NULL; + c->bpop.count = 0; + c->bpop.timeout = 0; + c->bpop.target = NULL; c->io_keys = listCreate(); c->watched_keys = listCreate(); listSetFreeMethod(c->io_keys,decrRefCount); @@ -55,7 +57,12 @@ redisClient *createClient(int fd) { return c; } +/* Set the event loop to listen for write events on the client's socket. + * Typically gets called every time a reply is built. */ int _installWriteEvent(redisClient *c) { + /* When CLOSE_AFTER_REPLY is set, no more replies may be added! */ + redisAssert(!(c->flags & REDIS_CLOSE_AFTER_REPLY)); + if (c->fd <= 0) return REDIS_ERR; if (c->bufpos == 0 && listLength(c->reply) == 0 && (c->replstate == REDIS_REPL_NONE || @@ -173,6 +180,9 @@ void addReply(redisClient *c, robj *obj) { if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK) _addReplyObjectToList(c,obj); } else { + /* FIXME: convert the long into string and use _addReplyToBuffer() + * instead of calling getDecodedObject. As this place in the + * code is too performance critical. */ obj = getDecodedObject(obj); if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK) _addReplyObjectToList(c,obj); @@ -270,6 +280,7 @@ void setDeferredMultiBulkLength(redisClient *c, void *node, long length) { } } +/* Add a duble as a bulk reply */ void addReplyDouble(redisClient *c, double d) { char dbuf[128], sbuf[128]; int dlen, slen; @@ -278,6 +289,8 @@ void addReplyDouble(redisClient *c, double d) { addReplyString(c,sbuf,slen); } +/* Add a long long as integer reply or bulk len / multi bulk count. + * Basically this is used to output . */ void _addReplyLongLong(redisClient *c, long long ll, char prefix) { char buf[128]; int len; @@ -296,6 +309,7 @@ void addReplyMultiBulkLen(redisClient *c, long length) { _addReplyLongLong(c,length,'*'); } +/* Create the length prefix of a bulk reply, example: $2234 */ void addReplyBulkLen(redisClient *c, robj *obj) { size_t len; @@ -317,40 +331,43 @@ void addReplyBulkLen(redisClient *c, robj *obj) { _addReplyLongLong(c,len,'$'); } +/* Add a Redis Object as a bulk reply */ void addReplyBulk(redisClient *c, robj *obj) { addReplyBulkLen(c,obj); addReply(c,obj); addReply(c,shared.crlf); } -/* In the CONFIG command we need to add vanilla C string as bulk replies */ +/* Add a C buffer as bulk reply */ +void addReplyBulkCBuffer(redisClient *c, void *p, size_t len) { + _addReplyLongLong(c,len,'$'); + addReplyString(c,p,len); + addReply(c,shared.crlf); +} + +/* Add a C nul term string as bulk reply */ void addReplyBulkCString(redisClient *c, char *s) { if (s == NULL) { addReply(c,shared.nullbulk); } else { - robj *o = createStringObject(s,strlen(s)); - addReplyBulk(c,o); - decrRefCount(o); + addReplyBulkCBuffer(c,s,strlen(s)); } } -void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) { - int cport, cfd; - char cip[128]; - redisClient *c; - REDIS_NOTUSED(el); - REDIS_NOTUSED(mask); - REDIS_NOTUSED(privdata); +/* Add a long long as a bulk reply */ +void addReplyBulkLongLong(redisClient *c, long long ll) { + char buf[64]; + int len; - 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) { + len = ll2string(buf,64,ll); + addReplyBulkCBuffer(c,buf,len); +} + +static void acceptCommonHandler(int fd) { + redisClient *c; + 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 @@ -370,6 +387,38 @@ void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) { 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++) @@ -442,6 +491,7 @@ void freeClient(redisClient *c) { /* 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 @@ -541,7 +591,6 @@ void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) { aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE); /* Close connection after entire reply has been sent. */ - if (c->flags & REDIS_QUIT) freeClient(c); if (c->flags & REDIS_CLOSE_AFTER_REPLY) freeClient(c); } } @@ -652,7 +701,7 @@ void closeTimedoutClients(void) { redisLog(REDIS_VERBOSE,"Closing idle client"); freeClient(c); } else if (c->flags & REDIS_BLOCKED) { - if (c->blockingto != 0 && c->blockingto < now) { + if (c->bpop.timeout != 0 && c->bpop.timeout < now) { addReply(c,shared.nullmultibulk); unblockClientWaitingData(c); } @@ -755,7 +804,7 @@ int processMultibulkBuffer(redisClient *c) { bulklen = strtol(c->querybuf+pos+1,&eptr,10); tolerr = (eptr[0] != '\r'); if (tolerr || bulklen == LONG_MIN || bulklen == LONG_MAX || - bulklen < 0 || bulklen > 1024*1024*1024) + bulklen < 0 || bulklen > 512*1024*1024) { addReplyError(c,"Protocol error: invalid bulk length"); setProtocolError(c,pos); @@ -794,17 +843,13 @@ int processMultibulkBuffer(redisClient *c) { void processInputBuffer(redisClient *c) { /* Keep processing while there is something in the input buffer */ while(sdslen(c->querybuf)) { - /* Before to process the input buffer, make sure the client is not - * waitig for a blocking operation such as BLPOP. Note that the first - * iteration the client is never blocked, otherwise the processInputBuffer - * would not be called at all, but after the execution of the first commands - * in the input buffer the client may be blocked, and the "goto again" - * will try to reiterate. The following line will make it return asap. */ + /* Immediately abort if the client is in the middle of something. */ if (c->flags & REDIS_BLOCKED || c->flags & REDIS_IO_WAIT) return; - /* Never continue to process the input buffer after QUIT. After the output - * buffer is flushed (with the OK), the connection will be dropped. */ - if (c->flags & REDIS_QUIT) return; + /* REDIS_CLOSE_AFTER_REPLY closes the connection once the reply is + * written to the client. Make sure to not let the reply grow after + * this flag has been set (i.e. don't process more commands). */ + if (c->flags & REDIS_CLOSE_AFTER_REPLY) return; /* Determine request type when unknown. */ if (!c->reqtype) {