A reimplementation of blocking operation internals.

[redis.git] / src / networking.c

diff --git a/src/networking.c b/src/networking.c
index 75f6cf2213cb8b83f8488d01eadee13f85474d66..3bc084f7d5fc4d5f4c7c0cbf61928ff011fae8fe 100644 (file)
--- a/src/networking.c
+++ b/src/networking.c
@@ -55,6 +55,7 @@ redisClient *createClient(int fd) {
     c->ctime = c->lastinteraction = server.unixtime;
     c->authenticated = 0;
     c->replstate = REDIS_REPL_NONE;
     c->ctime = c->lastinteraction = server.unixtime;
     c->authenticated = 0;
     c->replstate = REDIS_REPL_NONE;

+    c->slave_listening_port = 0;

     c->reply = listCreate();
     c->reply_bytes = 0;
     c->obuf_soft_limit_reached_time = 0;
     c->reply = listCreate();
     c->reply_bytes = 0;
     c->obuf_soft_limit_reached_time = 0;


@@ -364,7 +365,10 @@ void setDeferredMultiBulkLength(redisClient *c, void *node, long length) {
 
         /* Only glue when the next node is non-NULL (an sds in this case) */
         if (next->ptr != NULL) {
 
         /* Only glue when the next node is non-NULL (an sds in this case) */
         if (next->ptr != NULL) {

+            c->reply_bytes -= zmalloc_size_sds(len->ptr);
+            c->reply_bytes -= zmalloc_size_sds(next->ptr);

             len->ptr = sdscatlen(len->ptr,next->ptr,sdslen(next->ptr));
             len->ptr = sdscatlen(len->ptr,next->ptr,sdslen(next->ptr));

+            c->reply_bytes += zmalloc_size_sds(len->ptr);

             listDelNode(c->reply,ln->next);
         }
     }
             listDelNode(c->reply,ln->next);
         }
     }


@@ -1085,7 +1089,7 @@ sds getClientInfoString(redisClient *client) {
     if (emask & AE_WRITABLE) *p++ = 'w';
     *p = '\0';
     return sdscatprintf(sdsempty(),
     if (emask & AE_WRITABLE) *p++ = 'w';
     *p = '\0';
     return sdscatprintf(sdsempty(),

-        "addr=%s:%d fd=%d age=%ld idle=%ld flags=%s db=%d sub=%d psub=%d qbuf=%lu qbuf-free=%lu obl=%lu oll=%lu omem=%lu events=%s cmd=%s",
+        "addr=%s:%d fd=%d age=%ld idle=%ld flags=%s db=%d sub=%d psub=%d multi=%d qbuf=%lu qbuf-free=%lu obl=%lu oll=%lu omem=%lu events=%s cmd=%s",

         ip,port,client->fd,
         (long)(server.unixtime - client->ctime),
         (long)(server.unixtime - client->lastinteraction),
         ip,port,client->fd,
         (long)(server.unixtime - client->ctime),
         (long)(server.unixtime - client->lastinteraction),


@@ -1093,6 +1097,7 @@ sds getClientInfoString(redisClient *client) {
         client->db->id,
         (int) dictSize(client->pubsub_channels),
         (int) listLength(client->pubsub_patterns),
         client->db->id,
         (int) dictSize(client->pubsub_channels),
         (int) listLength(client->pubsub_patterns),

+        (client->flags & REDIS_MULTI) ? client->mstate.count : -1,

         (unsigned long) sdslen(client->querybuf),
         (unsigned long) sdsavail(client->querybuf),
         (unsigned long) client->bufpos,
         (unsigned long) sdslen(client->querybuf),
         (unsigned long) sdsavail(client->querybuf),
         (unsigned long) client->bufpos,


@@ -1301,6 +1306,7 @@ int checkClientOutputBufferLimits(redisClient *c) {
  * called from contexts where the client can't be freed safely, i.e. from the
  * lower level functions pushing data inside the client output buffers. */
 void asyncCloseClientOnOutputBufferLimitReached(redisClient *c) {
  * called from contexts where the client can't be freed safely, i.e. from the
  * lower level functions pushing data inside the client output buffers. */
 void asyncCloseClientOnOutputBufferLimitReached(redisClient *c) {

+    redisAssert(c->reply_bytes < ULONG_MAX-(1024*64));

     if (c->reply_bytes == 0 || c->flags & REDIS_CLOSE_ASAP) return;
     if (checkClientOutputBufferLimits(c)) {
         sds client = getClientInfoString(c);
     if (c->reply_bytes == 0 || c->flags & REDIS_CLOSE_ASAP) return;
     if (checkClientOutputBufferLimits(c)) {
         sds client = getClientInfoString(c);