X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/b19c33d48a3b5456b87f84c85fdd85592e0c0562..e53ca04b50b86ef158a75c54ae9ee8b17e31719c:/src/networking.c diff --git a/src/networking.c b/src/networking.c index 46d49bf6..166b44c9 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 || @@ -160,7 +167,6 @@ void _addReplyStringToList(redisClient *c, char *s, size_t len) { void addReply(redisClient *c, robj *obj) { if (_installWriteEvent(c) != REDIS_OK) return; - redisAssert(!server.vm_enabled || obj->storage == REDIS_VM_MEMORY); /* This is an important place where we can avoid copy-on-write * when there is a saving child running, avoiding touching the @@ -173,6 +179,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 +279,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 +288,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 +308,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 +330,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 +386,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++) @@ -408,10 +456,17 @@ void freeClient(redisClient *c) { ln = listSearchKey(server.clients,c); redisAssert(ln != NULL); listDelNode(server.clients,ln); + /* When client was just unblocked because of a blocking operation, + * remove it from the list with unblocked clients. */ + if (c->flags & REDIS_UNBLOCKED) { + ln = listSearchKey(server.unblocked_clients,c); + redisAssert(ln != NULL); + listDelNode(server.unblocked_clients,ln); + } /* Remove from the list of clients waiting for swapped keys, or ready * to be restarted, but not yet woken up again. */ if (c->flags & REDIS_IO_WAIT) { - redisAssert(server.vm_enabled); + redisAssert(server.ds_enabled); if (listLength(c->io_keys) == 0) { ln = listSearchKey(server.io_ready_clients,c); @@ -425,7 +480,7 @@ void freeClient(redisClient *c) { dontWaitForSwappedKey(c,ln->value); } } - server.vm_blocked_clients--; + server.cache_blocked_clients--; } listRelease(c->io_keys); /* Master/slave cleanup. @@ -466,15 +521,6 @@ void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) { REDIS_NOTUSED(el); REDIS_NOTUSED(mask); - /* Use writev() if we have enough buffers to send */ - if (!server.glueoutputbuf && - listLength(c->reply) > REDIS_WRITEV_THRESHOLD && - !(c->flags & REDIS_MASTER)) - { - sendReplyToClientWritev(el, fd, privdata, mask); - return; - } - while(c->bufpos > 0 || listLength(c->reply)) { if (c->bufpos > 0) { if (c->flags & REDIS_MASTER) { @@ -541,89 +587,10 @@ 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); } } -void sendReplyToClientWritev(aeEventLoop *el, int fd, void *privdata, int mask) -{ - redisClient *c = privdata; - int nwritten = 0, totwritten = 0, objlen, willwrite; - robj *o; - struct iovec iov[REDIS_WRITEV_IOVEC_COUNT]; - int offset, ion = 0; - REDIS_NOTUSED(el); - REDIS_NOTUSED(mask); - - listNode *node; - while (listLength(c->reply)) { - offset = c->sentlen; - ion = 0; - willwrite = 0; - - /* fill-in the iov[] array */ - for(node = listFirst(c->reply); node; node = listNextNode(node)) { - o = listNodeValue(node); - objlen = sdslen(o->ptr); - - if (totwritten + objlen - offset > REDIS_MAX_WRITE_PER_EVENT) - break; - - if(ion == REDIS_WRITEV_IOVEC_COUNT) - break; /* no more iovecs */ - - iov[ion].iov_base = ((char*)o->ptr) + offset; - iov[ion].iov_len = objlen - offset; - willwrite += objlen - offset; - offset = 0; /* just for the first item */ - ion++; - } - - if(willwrite == 0) - break; - - /* write all collected blocks at once */ - if((nwritten = writev(fd, iov, ion)) < 0) { - if (errno != EAGAIN) { - redisLog(REDIS_VERBOSE, - "Error writing to client: %s", strerror(errno)); - freeClient(c); - return; - } - break; - } - - totwritten += nwritten; - offset = c->sentlen; - - /* remove written robjs from c->reply */ - while (nwritten && listLength(c->reply)) { - o = listNodeValue(listFirst(c->reply)); - objlen = sdslen(o->ptr); - - if(nwritten >= objlen - offset) { - listDelNode(c->reply, listFirst(c->reply)); - nwritten -= objlen - offset; - c->sentlen = 0; - } else { - /* partial write */ - c->sentlen += nwritten; - break; - } - offset = 0; - } - } - - if (totwritten > 0) - c->lastinteraction = time(NULL); - - if (listLength(c->reply) == 0) { - c->sentlen = 0; - aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE); - } -} - /* resetClient prepare the client to process the next command */ void resetClient(redisClient *c) { freeClientArgv(c); @@ -652,7 +619,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 +722,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 +761,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) { @@ -863,3 +826,22 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) { } processInputBuffer(c); } + +void getClientsMaxBuffers(unsigned long *longest_output_list, + unsigned long *biggest_input_buffer) { + redisClient *c; + listNode *ln; + listIter li; + unsigned long lol = 0, bib = 0; + + listRewind(server.clients,&li); + while ((ln = listNext(&li)) != NULL) { + c = listNodeValue(ln); + + if (listLength(c->reply) > lol) lol = listLength(c->reply); + if (sdslen(c->querybuf) > bib) bib = sdslen(c->querybuf); + } + *longest_output_list = lol; + *biggest_input_buffer = bib; +} +