#include "redis.h"
#include <sys/uio.h>
+static void setProtocolError(redisClient *c, int pos);
+
void *dupClientReplyValue(void *o) {
incrRefCount((robj*)o);
return o;
c->authenticated = 0;
c->replstate = REDIS_REPL_NONE;
c->reply = listCreate();
+ c->reply_bytes = 0;
+ c->obuf_soft_limit_reached_time = 0;
listSetFreeMethod(c->reply,decrRefCount);
listSetDupMethod(c->reply,dupClientReplyValue);
c->bpop.keys = NULL;
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) {
+/* This function is called every time we are going to transmit new data
+ * to the client. The behavior is the following:
+ *
+ * If the client should receive new data (normal clients will) the function
+ * returns REDIS_OK, and make sure to install the write handler in our event
+ * loop so that when the socket is writable new data gets written.
+ *
+ * If the client should not receive new data, because it is a fake client
+ * or a slave, or because the setup of the write handler failed, the function
+ * returns REDIS_ERR.
+ *
+ * Typically gets called every time a reply is built, before adding more
+ * data to the clients output buffers. If the function returns REDIS_ERR no
+ * data should be appended to the output buffers. */
+int prepareClientToWrite(redisClient *c) {
if (c->flags & REDIS_LUA_CLIENT) return REDIS_OK;
- if (c->fd <= 0) return REDIS_ERR;
+ if (c->fd <= 0) return REDIS_ERR; /* Fake client */
if (c->bufpos == 0 && listLength(c->reply) == 0 &&
(c->replstate == REDIS_REPL_NONE ||
c->replstate == REDIS_REPL_ONLINE) &&
listAddNodeTail(c->reply,o);
}
}
+ c->reply_bytes += sdslen(o->ptr);
+ asyncCloseClientOnOutputBufferLimitReached(c);
}
/* This method takes responsibility over the sds. When it is no longer
return;
}
+ c->reply_bytes += sdslen(s);
if (listLength(c->reply) == 0) {
listAddNodeTail(c->reply,createObject(REDIS_STRING,s));
} else {
listAddNodeTail(c->reply,createObject(REDIS_STRING,s));
}
}
+ asyncCloseClientOnOutputBufferLimitReached(c);
}
void _addReplyStringToList(redisClient *c, char *s, size_t len) {
listAddNodeTail(c->reply,createStringObject(s,len));
}
}
+ c->reply_bytes += len;
+ asyncCloseClientOnOutputBufferLimitReached(c);
}
/* -----------------------------------------------------------------------------
* -------------------------------------------------------------------------- */
void addReply(redisClient *c, robj *obj) {
- if (_installWriteEvent(c) != REDIS_OK) return;
+ if (prepareClientToWrite(c) != REDIS_OK) return;
/* This is an important place where we can avoid copy-on-write
* when there is a saving child running, avoiding touching the
if (obj->encoding == REDIS_ENCODING_RAW) {
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. */
+ } else if (obj->encoding == REDIS_ENCODING_INT) {
+ /* Optimization: if there is room in the static buffer for 32 bytes
+ * (more than the max chars a 64 bit integer can take as string) we
+ * avoid decoding the object and go for the lower level approach. */
+ if (listLength(c->reply) == 0 && (sizeof(c->buf) - c->bufpos) >= 32) {
+ char buf[32];
+ int len;
+
+ len = ll2string(buf,sizeof(buf),(long)obj->ptr);
+ if (_addReplyToBuffer(c,buf,len) == REDIS_OK)
+ return;
+ /* else... continue with the normal code path, but should never
+ * happen actually since we verified there is room. */
+ }
obj = getDecodedObject(obj);
if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK)
_addReplyObjectToList(c,obj);
decrRefCount(obj);
+ } else {
+ redisPanic("Wrong obj->encoding in addReply()");
}
}
void addReplySds(redisClient *c, sds s) {
- if (_installWriteEvent(c) != REDIS_OK) {
+ if (prepareClientToWrite(c) != REDIS_OK) {
/* The caller expects the sds to be free'd. */
sdsfree(s);
return;
}
void addReplyString(redisClient *c, char *s, size_t len) {
- if (_installWriteEvent(c) != REDIS_OK) return;
+ if (prepareClientToWrite(c) != REDIS_OK) return;
if (_addReplyToBuffer(c,s,len) != REDIS_OK)
_addReplyStringToList(c,s,len);
}
-void _addReplyError(redisClient *c, char *s, size_t len) {
+void addReplyErrorLength(redisClient *c, char *s, size_t len) {
addReplyString(c,"-ERR ",5);
addReplyString(c,s,len);
addReplyString(c,"\r\n",2);
}
void addReplyError(redisClient *c, char *err) {
- _addReplyError(c,err,strlen(err));
+ addReplyErrorLength(c,err,strlen(err));
}
void addReplyErrorFormat(redisClient *c, const char *fmt, ...) {
for (j = 0; j < l; j++) {
if (s[j] == '\r' || s[j] == '\n') s[j] = ' ';
}
- _addReplyError(c,s,sdslen(s));
+ addReplyErrorLength(c,s,sdslen(s));
sdsfree(s);
}
-void _addReplyStatus(redisClient *c, char *s, size_t len) {
+void addReplyStatusLength(redisClient *c, char *s, size_t len) {
addReplyString(c,"+",1);
addReplyString(c,s,len);
addReplyString(c,"\r\n",2);
}
void addReplyStatus(redisClient *c, char *status) {
- _addReplyStatus(c,status,strlen(status));
+ addReplyStatusLength(c,status,strlen(status));
}
void addReplyStatusFormat(redisClient *c, const char *fmt, ...) {
va_start(ap,fmt);
sds s = sdscatvprintf(sdsempty(),fmt,ap);
va_end(ap);
- _addReplyStatus(c,s,sdslen(s));
+ addReplyStatusLength(c,s,sdslen(s));
sdsfree(s);
}
/* 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;
+ if (prepareClientToWrite(c) != REDIS_OK) return NULL;
listAddNodeTail(c->reply,createObject(REDIS_STRING,NULL));
return listLast(c->reply);
}
len = listNodeValue(ln);
len->ptr = sdscatprintf(sdsempty(),"*%ld\r\n",length);
+ c->reply_bytes += sdslen(len->ptr);
if (ln->next != NULL) {
next = listNodeValue(ln->next);
listDelNode(c->reply,ln->next);
}
}
+ asyncCloseClientOnOutputBufferLimitReached(c);
}
/* Add a duble as a bulk reply */
/* Add a long long as integer reply or bulk len / multi bulk count.
* Basically this is used to output <prefix><long long><crlf>. */
-void _addReplyLongLong(redisClient *c, long long ll, char prefix) {
+void addReplyLongLongWithPrefix(redisClient *c, long long ll, char prefix) {
char buf[128];
int len;
buf[0] = prefix;
else if (ll == 1)
addReply(c,shared.cone);
else
- _addReplyLongLong(c,ll,':');
+ addReplyLongLongWithPrefix(c,ll,':');
}
void addReplyMultiBulkLen(redisClient *c, long length) {
- _addReplyLongLong(c,length,'*');
+ addReplyLongLongWithPrefix(c,length,'*');
}
/* Create the length prefix of a bulk reply, example: $2234 */
len++;
}
}
- _addReplyLongLong(c,len,'$');
+ addReplyLongLongWithPrefix(c,len,'$');
}
/* Add a Redis Object as a bulk reply */
/* Add a C buffer as bulk reply */
void addReplyBulkCBuffer(redisClient *c, void *p, size_t len) {
- _addReplyLongLong(c,len,'$');
+ addReplyLongLongWithPrefix(c,len,'$');
addReplyString(c,p,len);
addReply(c,shared.crlf);
}
addReplyBulkCBuffer(c,buf,len);
}
+/* Copy 'src' client output buffers into 'dst' client output buffers.
+ * The function takes care of freeing the old output buffers of the
+ * destination client. */
+void copyClientOutputBuffer(redisClient *dst, redisClient *src) {
+ listRelease(dst->reply);
+ dst->reply = listDup(src->reply);
+ memcpy(dst->buf,src->buf,src->bufpos);
+ dst->bufpos = src->bufpos;
+ dst->reply_bytes = src->reply_bytes;
+}
+
static void acceptCommonHandler(int fd) {
redisClient *c;
if ((c = createClient(fd)) == NULL) {
void freeClient(redisClient *c) {
listNode *ln;
+ /* If this is marked as current client unset it */
+ if (server.current_client == c) server.current_client = NULL;
+
/* Note that if the client we are freeing is blocked into a blocking
* call, we have to set querybuf to NULL *before* to call
* unblockClientWaitingData() to avoid processInputBuffer() will get
/* Case 2: we lost the connection with the master. */
if (c->flags & REDIS_MASTER) {
server.master = NULL;
- server.replstate = REDIS_REPL_CONNECT;
+ server.repl_state = REDIS_REPL_CONNECT;
server.repl_down_since = time(NULL);
/* Since we lost the connection with the master, we should also
* close the connection with all our slaves if we have any, so
}
}
}
+
+ /* If this client was scheduled for async freeing we need to remove it
+ * from the queue. */
+ if (c->flags & REDIS_CLOSE_ASAP) {
+ ln = listSearchKey(server.clients_to_close,c);
+ redisAssert(ln != NULL);
+ listDelNode(server.clients_to_close,ln);
+ }
+
/* Release memory */
zfree(c->argv);
freeClientMultiState(c);
zfree(c);
}
+/* Schedule a client to free it at a safe time in the serverCron() function.
+ * This function is useful when we need to terminate a client but we are in
+ * a context where calling freeClient() is not possible, because the client
+ * should be valid for the continuation of the flow of the program. */
+void freeClientAsync(redisClient *c) {
+ if (c->flags & REDIS_CLOSE_ASAP) return;
+ c->flags |= REDIS_CLOSE_ASAP;
+ listAddNodeTail(server.clients_to_close,c);
+}
+
+void freeClientsInAsyncFreeQueue(void) {
+ while (listLength(server.clients_to_close)) {
+ listNode *ln = listFirst(server.clients_to_close);
+ redisClient *c = listNodeValue(ln);
+
+ c->flags &= ~REDIS_CLOSE_ASAP;
+ freeClient(c);
+ listDelNode(server.clients_to_close,ln);
+ }
+}
+
void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
redisClient *c = privdata;
int nwritten = 0, totwritten = 0, objlen;
if (c->sentlen == objlen) {
listDelNode(c->reply,listFirst(c->reply));
c->sentlen = 0;
+ c->reply_bytes -= objlen;
}
}
/* Note that we avoid to send more thank REDIS_MAX_WRITE_PER_EVENT
size_t querylen;
/* Nothing to do without a \r\n */
- if (newline == NULL)
+ if (newline == NULL) {
+ if (sdslen(c->querybuf) > REDIS_INLINE_MAX_SIZE) {
+ addReplyError(c,"Protocol error: too big inline request");
+ setProtocolError(c,0);
+ }
return REDIS_ERR;
+ }
/* Split the input buffer up to the \r\n */
querylen = newline-(c->querybuf);
/* Multi bulk length cannot be read without a \r\n */
newline = strchr(c->querybuf,'\r');
- if (newline == NULL)
+ if (newline == NULL) {
+ if (sdslen(c->querybuf) > REDIS_INLINE_MAX_SIZE) {
+ addReplyError(c,"Protocol error: too big mbulk count string");
+ setProtocolError(c,0);
+ }
return REDIS_ERR;
+ }
/* Buffer should also contain \n */
if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2))
/* Read bulk length if unknown */
if (c->bulklen == -1) {
newline = strchr(c->querybuf+pos,'\r');
- if (newline == NULL)
+ if (newline == NULL) {
+ if (sdslen(c->querybuf) > REDIS_INLINE_MAX_SIZE) {
+ addReplyError(c,"Protocol error: too big bulk count string");
+ setProtocolError(c,0);
+ }
break;
+ }
/* Buffer should also contain \n */
if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2))
if (pos) c->querybuf = sdsrange(c->querybuf,pos,-1);
/* We're done when c->multibulk == 0 */
- if (c->multibulklen == 0) {
- return REDIS_OK;
- }
+ if (c->multibulklen == 0) return REDIS_OK;
+
+ /* Still not read to process the command */
return REDIS_ERR;
}
REDIS_NOTUSED(el);
REDIS_NOTUSED(mask);
+ server.current_client = c;
readlen = REDIS_IOBUF_LEN;
/* If this is a multi bulk request, and we are processing a bulk reply
* that is large enough, try to maximize the probabilty that the query
sdsIncrLen(c->querybuf,nread);
c->lastinteraction = time(NULL);
} else {
+ server.current_client = NULL;
return;
}
if (sdslen(c->querybuf) > server.client_max_querybuf_len) {
- sds ci = getClientInfoString(c);
- redisLog(REDIS_WARNING,"Closing client that reached max query buffer length: %s", ci);
+ sds ci = getClientInfoString(c), bytes = sdsempty();
+
+ bytes = sdscatrepr(bytes,c->querybuf,64);
+ redisLog(REDIS_WARNING,"Closing client that reached max query buffer length: %s (qbuf initial bytes: %s)", ci, bytes);
sdsfree(ci);
+ sdsfree(bytes);
freeClient(c);
return;
}
processInputBuffer(c);
+ server.current_client = NULL;
}
void getClientsMaxBuffers(unsigned long *longest_output_list,
if (client->flags & REDIS_DIRTY_CAS) *p++ = 'd';
if (client->flags & REDIS_CLOSE_AFTER_REPLY) *p++ = 'c';
if (client->flags & REDIS_UNBLOCKED) *p++ = 'u';
+ if (client->flags & REDIS_CLOSE_ASAP) *p++ = 'A';
if (p == flags) *p++ = 'N';
*p++ = '\0';
if (emask & AE_WRITABLE) *p++ = 'w';
*p = '\0';
return sdscatprintf(sdsempty(),
- "addr=%s:%d fd=%d idle=%ld flags=%s db=%d sub=%d psub=%d qbuf=%lu obl=%lu oll=%lu events=%s cmd=%s",
+ "addr=%s:%d fd=%d idle=%ld flags=%s db=%d sub=%d psub=%d qbuf=%lu obl=%lu oll=%lu omem=%lu events=%s cmd=%s",
ip,port,client->fd,
(long)(now - client->lastinteraction),
flags,
(unsigned long) sdslen(client->querybuf),
(unsigned long) client->bufpos,
(unsigned long) listLength(client->reply),
+ getClientOutputBufferMemoryUsage(client),
events,
client->lastcmd ? client->lastcmd->name : "NULL");
}
listRewind(server.clients,&li);
while ((ln = listNext(&li)) != NULL) {
+ sds cs;
+
client = listNodeValue(ln);
- o = sdscatsds(o,getClientInfoString(client));
+ cs = getClientInfoString(client);
+ o = sdscatsds(o,cs);
+ sdsfree(cs);
o = sdscatlen(o,"\n",1);
}
return o;
redisAssertWithInfo(c,NULL,c->cmd != NULL);
}
}
+
+/* This function returns the number of bytes that Redis is virtually
+ * using to store the reply still not read by the client.
+ * It is "virtual" since the reply output list may contain objects that
+ * are shared and are not really using additional memory.
+ *
+ * The function returns the total sum of the length of all the objects
+ * stored in the output list, plus the memory used to allocate every
+ * list node. The static reply buffer is not taken into account since it
+ * is allocated anyway.
+ *
+ * Note: this function is very fast so can be called as many time as
+ * the caller wishes. The main usage of this function currently is
+ * enforcing the client output lenght limits. */
+unsigned long getClientOutputBufferMemoryUsage(redisClient *c) {
+ unsigned long list_item_size = sizeof(listNode);
+
+ return c->reply_bytes + (list_item_size*listLength(c->reply));
+}
+
+/* Get the class of a client, used in order to envorce limits to different
+ * classes of clients.
+ *
+ * The function will return one of the following:
+ * REDIS_CLIENT_LIMIT_CLASS_NORMAL -> Normal client
+ * REDIS_CLIENT_LIMIT_CLASS_SLAVE -> Slave or client executing MONITOR command
+ * REDIS_CLIENT_LIMIT_CLASS_PUBSUB -> Client subscribed to Pub/Sub channels
+ */
+int getClientLimitClass(redisClient *c) {
+ if (c->flags & REDIS_SLAVE) return REDIS_CLIENT_LIMIT_CLASS_SLAVE;
+ if (dictSize(c->pubsub_channels) || listLength(c->pubsub_patterns))
+ return REDIS_CLIENT_LIMIT_CLASS_PUBSUB;
+ return REDIS_CLIENT_LIMIT_CLASS_NORMAL;
+}
+
+/* The function checks if the client reached output buffer soft or hard
+ * limit, and also update the state needed to check the soft limit as
+ * a side effect.
+ *
+ * Return value: non-zero if the client reached the soft or the hard limit.
+ * Otherwise zero is returned. */
+int checkClientOutputBufferLimits(redisClient *c) {
+ int soft = 0, hard = 0, class;
+ unsigned long used_mem = getClientOutputBufferMemoryUsage(c);
+
+ class = getClientLimitClass(c);
+ if (server.client_obuf_limits[class].hard_limit_bytes &&
+ used_mem >= server.client_obuf_limits[class].hard_limit_bytes)
+ hard = 1;
+ if (server.client_obuf_limits[class].soft_limit_bytes &&
+ used_mem >= server.client_obuf_limits[class].soft_limit_bytes)
+ soft = 1;
+
+ /* We need to check if the soft limit is reached continuously for the
+ * specified amount of seconds. */
+ if (soft) {
+ if (c->obuf_soft_limit_reached_time == 0) {
+ c->obuf_soft_limit_reached_time = server.unixtime;
+ soft = 0; /* First time we see the soft limit reached */
+ } else {
+ time_t elapsed = server.unixtime - c->obuf_soft_limit_reached_time;
+
+ if (elapsed <=
+ server.client_obuf_limits[class].soft_limit_seconds) {
+ soft = 0; /* The client still did not reached the max number of
+ seconds for the soft limit to be considered
+ reached. */
+ }
+ }
+ } else {
+ c->obuf_soft_limit_reached_time = 0;
+ }
+ return soft || hard;
+}
+
+/* Asynchronously close a client if soft or hard limit is reached on the
+ * output buffer size. The caller can check if the client will be closed
+ * checking if the client REDIS_CLOSE_ASAP flag is set.
+ *
+ * Note: we need to close the client asynchronously because this function is
+ * 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) {
+ if (c->flags & REDIS_CLOSE_ASAP) return;
+ if (checkClientOutputBufferLimits(c)) {
+ sds client = getClientInfoString(c);
+
+ freeClientAsync(c);
+ redisLog(REDIS_NOTICE,"Client %s scheduled to be closed ASAP for overcoming of output buffer limits.", client);
+ sdsfree(client);
+ }
+}
projects / redis.git / blobdiff