Object approximated LRU algorithm enhanced / fixed / refactored. This is used for...

[redis.git] / src / networking.c

diff --git a/src/networking.c b/src/networking.c
index 26a6a4c0408d2df8d40eeaaad622e63620447c6c..632fd04769dca5a3c464b26b982d8a9a064a4faa 100644 (file)
--- a/src/networking.c
+++ b/src/networking.c
@@ -17,9 +17,18 @@ redisClient *createClient(int fd) {
     anetNonBlock(NULL,fd);
     anetTcpNoDelay(NULL,fd);
     if (!c) return NULL;
+    if (aeCreateFileEvent(server.el,fd,AE_READABLE,
+        readQueryFromClient, c) == AE_ERR)
+    {
+        close(fd);
+        zfree(c);
+        return NULL;
+    }
+
     selectDb(c,0);
     c->fd = fd;
     c->querybuf = sdsempty();
+    c->newline = NULL;
     c->argc = 0;
     c->argv = NULL;
     c->bulklen = -1;
@@ -43,17 +52,12 @@ redisClient *createClient(int fd) {
     c->pubsub_patterns = listCreate();
     listSetFreeMethod(c->pubsub_patterns,decrRefCount);
     listSetMatchMethod(c->pubsub_patterns,listMatchObjects);
-    if (aeCreateFileEvent(server.el, c->fd, AE_READABLE,
-        readQueryFromClient, c) == AE_ERR) {
-        freeClient(c);
-        return NULL;
-    }
     listAddNodeTail(server.clients,c);
     initClientMultiState(c);
     return c;
 }
 
-int _ensureFileEvent(redisClient *c) {
+int _installWriteEvent(redisClient *c) {
     if (c->fd <= 0) return REDIS_ERR;
     if (c->bufpos == 0 && listLength(c->reply) == 0 &&
         (c->replstate == REDIS_REPL_NONE ||
@@ -157,21 +161,29 @@ void _addReplyStringToList(redisClient *c, char *s, size_t len) {
 }
 
 void addReply(redisClient *c, robj *obj) {
-    if (_ensureFileEvent(c) != REDIS_OK) return;
-    if (server.vm_enabled && obj->storage != REDIS_VM_MEMORY) {
-        /* Returns a new object with refcount 1 */
-        obj = dupStringObject(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
+     * refcount field of the object if it's not needed.
+     *
+     * If the encoding is RAW and there is room in the static buffer
+     * we'll be able to send the object to the client without
+     * messing with its page. */
+    if (obj->encoding == REDIS_ENCODING_RAW) {
+        if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK)
+            _addReplyObjectToList(c,obj);
     } else {
-        /* This increments the refcount. */
         obj = getDecodedObject(obj);
+        if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK)
+            _addReplyObjectToList(c,obj);
+        decrRefCount(obj);
     }
-    if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK)
-        _addReplyObjectToList(c,obj);
-    decrRefCount(obj);
 }
 
 void addReplySds(redisClient *c, sds s) {
-    if (_ensureFileEvent(c) != REDIS_OK) {
+    if (_installWriteEvent(c) != REDIS_OK) {
         /* The caller expects the sds to be free'd. */
         sdsfree(s);
         return;
@@ -185,7 +197,7 @@ void addReplySds(redisClient *c, sds s) {
 }
 
 void addReplyString(redisClient *c, char *s, size_t len) {
-    if (_ensureFileEvent(c) != REDIS_OK) return;
+    if (_installWriteEvent(c) != REDIS_OK) return;
     if (_addReplyToBuffer(c,s,len) != REDIS_OK)
         _addReplyStringToList(c,s,len);
 }
@@ -231,7 +243,10 @@ void addReplyStatusFormat(redisClient *c, const char *fmt, ...) {
 /* Adds an empty object to the reply list that will contain the multi bulk
  * length, which is not known when this function is called. */
 void *addDeferredMultiBulkLength(redisClient *c) {
-    if (_ensureFileEvent(c) != REDIS_OK) return NULL;
+    /* Note that we install the write event here even if the object is not
+     * ready to be sent, since we are sure that before returning to the
+     * event loop setDeferredMultiBulkLength() will be called. */
+    if (_installWriteEvent(c) != REDIS_OK) return NULL;
     listAddNodeTail(c->reply,createObject(REDIS_STRING,NULL));
     return listLast(c->reply);
 }
@@ -617,6 +632,7 @@ void resetClient(redisClient *c) {
     freeClientArgv(c);
     c->bulklen = -1;
     c->multibulk = 0;
+    c->newline = NULL;
 }
 
 void closeTimedoutClients(void) {
@@ -648,6 +664,8 @@ void closeTimedoutClients(void) {
 }
 
 void processInputBuffer(redisClient *c) {
+    int seeknewline = 0;
+
 again:
     /* Before to process the input buffer, make sure the client is not
      * waitig for a blocking operation such as BLPOP. Note that the first
@@ -656,15 +674,19 @@ again:
      * 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. */
     if (c->flags & REDIS_BLOCKED || c->flags & REDIS_IO_WAIT) return;
+
+    if (seeknewline && c->bulklen == -1) c->newline = strchr(c->querybuf,'\n');
+    seeknewline = 1;
     if (c->bulklen == -1) {
         /* Read the first line of the query */
-        char *p = strchr(c->querybuf,'\n');
         size_t querylen;
 
-        if (p) {
+        if (c->newline) {
+            char *p = c->newline;
             sds query, *argv;
             int argc, j;
 
+            c->newline = NULL;
             query = c->querybuf;
             c->querybuf = sdsempty();
             querylen = 1+(p-(query));
@@ -751,8 +773,14 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
         return;
     }
     if (nread) {
+        size_t oldlen = sdslen(c->querybuf);
         c->querybuf = sdscatlen(c->querybuf, buf, nread);
         c->lastinteraction = time(NULL);
+        /* Scan this new piece of the query for the newline. We do this
+         * here in order to make sure we perform this scan just one time
+         * per piece of buffer, leading to an O(N) scan instead of O(N*N) */
+        if (c->bulklen == -1 && c->newline == NULL)
+            c->newline = strchr(c->querybuf+oldlen,'\n');
     } else {
         return;
     }