#include "redis.h"
-
#include <sys/uio.h>
void *dupClientReplyValue(void *o) {
}
redisClient *createClient(int fd) {
- redisClient *c = zmalloc(sizeof(*c));
+ redisClient *c;
+
+ /* Make sure to allocate a multiple of the page size to prevent wasting
+ * memory. A page size of 4096 is assumed here. We need to compensate
+ * for the zmalloc overhead of sizeof(size_t) bytes. */
+ size_t size = 8192-sizeof(size_t);
+ redisAssert(size > sizeof(redisClient));
+ c = zmalloc(size);
+ c->buflen = size-sizeof(redisClient);
+ c->bufpos = 0;
anetNonBlock(NULL,fd);
anetTcpNoDelay(NULL,fd);
return c;
}
-void addReply(redisClient *c, robj *obj) {
- if (listLength(c->reply) == 0 &&
+int _ensureFileEvent(redisClient *c) {
+ if (c->fd <= 0) return REDIS_ERR;
+ if (c->bufpos == 0 && listLength(c->reply) == 0 &&
(c->replstate == REDIS_REPL_NONE ||
c->replstate == REDIS_REPL_ONLINE) &&
aeCreateFileEvent(server.el, c->fd, AE_WRITABLE,
- sendReplyToClient, c) == AE_ERR) return;
+ sendReplyToClient, c) == AE_ERR) return REDIS_ERR;
+ return REDIS_OK;
+}
+
+void _addReplyObjectToList(redisClient *c, robj *obj) {
+ redisAssert(obj->type == REDIS_STRING &&
+ obj->encoding == REDIS_ENCODING_RAW);
+ listAddNodeTail(c->reply,obj);
+}
+
+void _ensureBufferInReplyList(redisClient *c) {
+ sds buffer = sdsnewlen(NULL,REDIS_REPLY_CHUNK_SIZE);
+ sdsupdatelen(buffer); /* sdsnewlen expects non-empty string */
+ listAddNodeTail(c->reply,createObject(REDIS_REPLY_NODE,buffer));
+}
+
+void _addReplyStringToBuffer(redisClient *c, char *s, size_t len) {
+ size_t available = 0;
+ redisAssert(len < REDIS_REPLY_CHUNK_THRESHOLD);
+ if (listLength(c->reply) > 0) {
+ robj *o = listNodeValue(listLast(c->reply));
+
+ /* Make sure to append to a reply node with enough bytes available. */
+ if (o->type == REDIS_REPLY_NODE) available = sdsavail(o->ptr);
+ if (o->type != REDIS_REPLY_NODE || len > available) {
+ _ensureBufferInReplyList(c);
+ _addReplyStringToBuffer(c,s,len);
+ } else {
+ o->ptr = sdscatlen(o->ptr,s,len);
+ }
+ } else {
+ available = c->buflen-c->bufpos;
+ if (len > available) {
+ _ensureBufferInReplyList(c);
+ _addReplyStringToBuffer(c,s,len);
+ } else {
+ memcpy(c->buf+c->bufpos,s,len);
+ c->bufpos += 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);
- obj->refcount = 0; /* getDecodedObject() will increment the refcount */
+ } else {
+ /* This increments the refcount. */
+ obj = getDecodedObject(obj);
+ }
+
+ if (sdslen(obj->ptr) < REDIS_REPLY_CHUNK_THRESHOLD) {
+ _addReplyStringToBuffer(c,obj->ptr,sdslen(obj->ptr));
+ decrRefCount(obj);
+ } else {
+ _addReplyObjectToList(c,obj);
}
- listAddNodeTail(c->reply,getDecodedObject(obj));
}
void addReplySds(redisClient *c, sds s) {
- robj *o = createObject(REDIS_STRING,s);
- addReply(c,o);
- decrRefCount(o);
+ if (_ensureFileEvent(c) != REDIS_OK) {
+ /* The caller expects the sds to be free'd. */
+ sdsfree(s);
+ return;
+ }
+ if (sdslen(s) < REDIS_REPLY_CHUNK_THRESHOLD) {
+ _addReplyStringToBuffer(c,s,sdslen(s));
+ sdsfree(s);
+ } else {
+ _addReplyObjectToList(c,createObject(REDIS_STRING,s));
+ }
}
-void addReplyDouble(redisClient *c, double d) {
- char buf[128];
-
- snprintf(buf,sizeof(buf),"%.17g",d);
- addReplySds(c,sdscatprintf(sdsempty(),"$%lu\r\n%s\r\n",
- (unsigned long) strlen(buf),buf));
+void addReplyString(redisClient *c, char *s, size_t len) {
+ if (_ensureFileEvent(c) != REDIS_OK) return;
+ if (len < REDIS_REPLY_CHUNK_THRESHOLD) {
+ _addReplyStringToBuffer(c,s,len);
+ } else {
+ _addReplyObjectToList(c,createStringObject(s,len));
+ }
}
-void addReplyLongLong(redisClient *c, long long ll) {
- char buf[128];
- size_t len;
+/* 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;
+ _addReplyObjectToList(c,createObject(REDIS_STRING,NULL));
+ return listLast(c->reply);
+}
- if (ll == 0) {
- addReply(c,shared.czero);
- return;
- } else if (ll == 1) {
- addReply(c,shared.cone);
- return;
+/* Populate the length object and try glueing it to the next chunk. */
+void setDeferredMultiBulkLength(redisClient *c, void *node, long length) {
+ listNode *ln = (listNode*)node;
+ robj *len, *next;
+
+ /* Abort when *node is NULL (see addDeferredMultiBulkLength). */
+ if (node == NULL) return;
+
+ len = listNodeValue(ln);
+ len->ptr = sdscatprintf(sdsempty(),"*%ld\r\n",length);
+ if (ln->next != NULL) {
+ next = listNodeValue(ln->next);
+ /* Only glue when the next node is a reply chunk. */
+ if (next->type == REDIS_REPLY_NODE) {
+ len->ptr = sdscatlen(len->ptr,next->ptr,sdslen(next->ptr));
+ listDelNode(c->reply,ln->next);
+ }
}
- buf[0] = ':';
+}
+
+void addReplyDouble(redisClient *c, double d) {
+ char dbuf[128], sbuf[128];
+ int dlen, slen;
+ dlen = snprintf(dbuf,sizeof(dbuf),"%.17g",d);
+ slen = snprintf(sbuf,sizeof(sbuf),"$%d\r\n%s\r\n",dlen,dbuf);
+ addReplyString(c,sbuf,slen);
+}
+
+void _addReplyLongLong(redisClient *c, long long ll, char prefix) {
+ char buf[128];
+ int len;
+ buf[0] = prefix;
len = ll2string(buf+1,sizeof(buf)-1,ll);
buf[len+1] = '\r';
buf[len+2] = '\n';
- addReplySds(c,sdsnewlen(buf,len+3));
+ addReplyString(c,buf,len+3);
}
-void addReplyUlong(redisClient *c, unsigned long ul) {
- char buf[128];
- size_t len;
+void addReplyLongLong(redisClient *c, long long ll) {
+ _addReplyLongLong(c,ll,':');
+}
- if (ul == 0) {
- addReply(c,shared.czero);
- return;
- } else if (ul == 1) {
- addReply(c,shared.cone);
- return;
- }
- len = snprintf(buf,sizeof(buf),":%lu\r\n",ul);
- addReplySds(c,sdsnewlen(buf,len));
+void addReplyMultiBulkLen(redisClient *c, long length) {
+ _addReplyLongLong(c,length,'*');
}
void addReplyBulkLen(redisClient *c, robj *obj) {
- size_t len, intlen;
- char buf[128];
+ size_t len;
if (obj->encoding == REDIS_ENCODING_RAW) {
len = sdslen(obj->ptr);
len++;
}
}
- buf[0] = '$';
- intlen = ll2string(buf+1,sizeof(buf)-1,(long long)len);
- buf[intlen+1] = '\r';
- buf[intlen+2] = '\n';
- addReplySds(c,sdsnewlen(buf,intlen+3));
+ _addReplyLongLong(c,len,'$');
}
void addReplyBulk(redisClient *c, robj *obj) {
ln = listSearchKey(server.clients,c);
redisAssert(ln != NULL);
listDelNode(server.clients,ln);
- /* Remove from the list of clients that are now ready to be restarted
- * after waiting for swapped keys */
- if (c->flags & REDIS_IO_WAIT && listLength(c->io_keys) == 0) {
- ln = listSearchKey(server.io_ready_clients,c);
- if (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);
+ if (listLength(c->io_keys) == 0) {
+ ln = listSearchKey(server.io_ready_clients,c);
+
+ /* When this client is waiting to be woken up (REDIS_IO_WAIT),
+ * it should be present in the list io_ready_clients */
+ redisAssert(ln != NULL);
listDelNode(server.io_ready_clients,ln);
- server.vm_blocked_clients--;
+ } else {
+ while (listLength(c->io_keys)) {
+ ln = listFirst(c->io_keys);
+ dontWaitForSwappedKey(c,ln->value);
+ }
}
- }
- /* Remove from the list of clients waiting for swapped keys */
- while (server.vm_enabled && listLength(c->io_keys)) {
- ln = listFirst(c->io_keys);
- dontWaitForSwappedKey(c,ln->value);
+ server.vm_blocked_clients--;
}
listRelease(c->io_keys);
- /* Master/slave cleanup */
+ /* Master/slave cleanup.
+ * Case 1: we lost the connection with a slave. */
if (c->flags & REDIS_SLAVE) {
if (c->replstate == REDIS_REPL_SEND_BULK && c->repldbfd != -1)
close(c->repldbfd);
redisAssert(ln != NULL);
listDelNode(l,ln);
}
+
+ /* Case 2: we lost the connection with the master. */
if (c->flags & REDIS_MASTER) {
server.master = NULL;
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
+ * when we'll resync with the master the other slaves will sync again
+ * with us as well. Note that also when the slave is not connected
+ * to the master it will keep refusing connections by other slaves. */
+ while (listLength(server.slaves)) {
+ ln = listFirst(server.slaves);
+ freeClient((redisClient*)ln->value);
+ }
}
/* Release memory */
zfree(c->argv);
zfree(c);
}
-#define GLUEREPLY_UP_TO (1024)
-static void glueReplyBuffersIfNeeded(redisClient *c) {
- int copylen = 0;
- char buf[GLUEREPLY_UP_TO];
- listNode *ln;
- listIter li;
- robj *o;
-
- listRewind(c->reply,&li);
- while((ln = listNext(&li))) {
- int objlen;
-
- o = ln->value;
- objlen = sdslen(o->ptr);
- if (copylen + objlen <= GLUEREPLY_UP_TO) {
- memcpy(buf+copylen,o->ptr,objlen);
- copylen += objlen;
- listDelNode(c->reply,ln);
- } else {
- if (copylen == 0) return;
- break;
- }
- }
- /* Now the output buffer is empty, add the new single element */
- o = createObject(REDIS_STRING,sdsnewlen(buf,copylen));
- listAddNodeHead(c->reply,o);
-}
-
void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
redisClient *c = privdata;
int nwritten = 0, totwritten = 0, objlen;
return;
}
- while(listLength(c->reply)) {
- if (server.glueoutputbuf && listLength(c->reply) > 1)
- glueReplyBuffersIfNeeded(c);
+ while(c->bufpos > 0 || listLength(c->reply)) {
+ if (c->bufpos > 0) {
+ if (c->flags & REDIS_MASTER) {
+ /* Don't reply to a master */
+ nwritten = c->bufpos - c->sentlen;
+ } else {
+ nwritten = write(fd,c->buf+c->sentlen,c->bufpos-c->sentlen);
+ if (nwritten <= 0) break;
+ }
+ c->sentlen += nwritten;
+ totwritten += nwritten;
- o = listNodeValue(listFirst(c->reply));
- objlen = sdslen(o->ptr);
+ /* If the buffer was sent, set bufpos to zero to continue with
+ * the remainder of the reply. */
+ if (c->sentlen == c->bufpos) {
+ c->bufpos = 0;
+ c->sentlen = 0;
+ }
+ } else {
+ o = listNodeValue(listFirst(c->reply));
+ objlen = sdslen(o->ptr);
- if (objlen == 0) {
- listDelNode(c->reply,listFirst(c->reply));
- continue;
- }
+ if (objlen == 0) {
+ listDelNode(c->reply,listFirst(c->reply));
+ continue;
+ }
- if (c->flags & REDIS_MASTER) {
- /* Don't reply to a master */
- nwritten = objlen - c->sentlen;
- } else {
- nwritten = write(fd, ((char*)o->ptr)+c->sentlen, objlen - c->sentlen);
- if (nwritten <= 0) break;
- }
- c->sentlen += nwritten;
- totwritten += nwritten;
- /* If we fully sent the object on head go to the next one */
- if (c->sentlen == objlen) {
- listDelNode(c->reply,listFirst(c->reply));
- c->sentlen = 0;
+ if (c->flags & REDIS_MASTER) {
+ /* Don't reply to a master */
+ nwritten = objlen - c->sentlen;
+ } else {
+ nwritten = write(fd, ((char*)o->ptr)+c->sentlen,objlen-c->sentlen);
+ if (nwritten <= 0) break;
+ }
+ c->sentlen += nwritten;
+ totwritten += nwritten;
+
+ /* If we fully sent the object on head go to the next one */
+ if (c->sentlen == objlen) {
+ listDelNode(c->reply,listFirst(c->reply));
+ c->sentlen = 0;
+ }
}
/* Note that we avoid to send more thank REDIS_MAX_WRITE_PER_EVENT
* bytes, in a single threaded server it's a good idea to serve
if (server.maxidletime &&
!(c->flags & REDIS_SLAVE) && /* no timeout for slaves */
!(c->flags & REDIS_MASTER) && /* no timeout for masters */
+ !(c->flags & REDIS_BLOCKED) && /* no timeout for BLPOP */
dictSize(c->pubsub_channels) == 0 && /* no timeout for pubsub */
listLength(c->pubsub_patterns) == 0 &&
(now - c->lastinteraction > server.maxidletime))
projects / redis.git / blobdiff