+ server.current_client = NULL;
+}
+
+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;
+}
+
+/* Turn a Redis client into an sds string representing its state. */
+sds getClientInfoString(redisClient *client) {
+ char ip[32], flags[16], events[3], *p;
+ int port;
+ time_t now = time(NULL);
+ int emask;
+
+ anetPeerToString(client->fd,ip,&port);
+ p = flags;
+ if (client->flags & REDIS_SLAVE) {
+ if (client->flags & REDIS_MONITOR)
+ *p++ = 'O';
+ else
+ *p++ = 'S';
+ }
+ if (client->flags & REDIS_MASTER) *p++ = 'M';
+ if (client->flags & REDIS_MULTI) *p++ = 'x';
+ if (client->flags & REDIS_BLOCKED) *p++ = 'b';
+ 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';
+
+ emask = client->fd == -1 ? 0 : aeGetFileEvents(server.el,client->fd);
+ p = events;
+ if (emask & AE_READABLE) *p++ = 'r';
+ 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",
+ ip,port,client->fd,
+ (long)(now - client->ctime),
+ (long)(now - client->lastinteraction),
+ flags,
+ client->db->id,
+ (int) dictSize(client->pubsub_channels),
+ (int) listLength(client->pubsub_patterns),
+ (unsigned long) sdslen(client->querybuf),
+ (unsigned long) sdsavail(client->querybuf),
+ (unsigned long) client->bufpos,
+ (unsigned long) listLength(client->reply),
+ getClientOutputBufferMemoryUsage(client),
+ events,
+ client->lastcmd ? client->lastcmd->name : "NULL");
+}
+
+sds getAllClientsInfoString(void) {
+ listNode *ln;
+ listIter li;
+ redisClient *client;
+ sds o = sdsempty();
+
+ listRewind(server.clients,&li);
+ while ((ln = listNext(&li)) != NULL) {
+ sds cs;
+
+ client = listNodeValue(ln);
+ cs = getClientInfoString(client);
+ o = sdscatsds(o,cs);
+ sdsfree(cs);
+ o = sdscatlen(o,"\n",1);
+ }
+ return o;
+}
+
+void clientCommand(redisClient *c) {
+ listNode *ln;
+ listIter li;
+ redisClient *client;
+
+ if (!strcasecmp(c->argv[1]->ptr,"list") && c->argc == 2) {
+ sds o = getAllClientsInfoString();
+ addReplyBulkCBuffer(c,o,sdslen(o));
+ sdsfree(o);
+ } else if (!strcasecmp(c->argv[1]->ptr,"kill") && c->argc == 3) {
+ listRewind(server.clients,&li);
+ while ((ln = listNext(&li)) != NULL) {
+ char ip[32], addr[64];
+ int port;
+
+ client = listNodeValue(ln);
+ if (anetPeerToString(client->fd,ip,&port) == -1) continue;
+ snprintf(addr,sizeof(addr),"%s:%d",ip,port);
+ if (strcmp(addr,c->argv[2]->ptr) == 0) {
+ addReply(c,shared.ok);
+ if (c == client) {
+ client->flags |= REDIS_CLOSE_AFTER_REPLY;
+ } else {
+ freeClient(client);
+ }
+ return;
+ }
+ }
+ addReplyError(c,"No such client");
+ } else {
+ addReplyError(c, "Syntax error, try CLIENT (LIST | KILL ip:port)");
+ }
+}
+
+/* Rewrite the command vector of the client. All the new objects ref count
+ * is incremented. The old command vector is freed, and the old objects
+ * ref count is decremented. */
+void rewriteClientCommandVector(redisClient *c, int argc, ...) {
+ va_list ap;
+ int j;
+ robj **argv; /* The new argument vector */
+
+ argv = zmalloc(sizeof(robj*)*argc);
+ va_start(ap,argc);
+ for (j = 0; j < argc; j++) {
+ robj *a;
+
+ a = va_arg(ap, robj*);
+ argv[j] = a;
+ incrRefCount(a);
+ }
+ /* We free the objects in the original vector at the end, so we are
+ * sure that if the same objects are reused in the new vector the
+ * refcount gets incremented before it gets decremented. */
+ for (j = 0; j < c->argc; j++) decrRefCount(c->argv[j]);
+ zfree(c->argv);
+ /* Replace argv and argc with our new versions. */
+ c->argv = argv;
+ c->argc = argc;
+ c->cmd = lookupCommand(c->argv[0]->ptr);
+ redisAssertWithInfo(c,NULL,c->cmd != NULL);
+ va_end(ap);
+}
+
+/* Rewrite a single item in the command vector.
+ * The new val ref count is incremented, and the old decremented. */
+void rewriteClientCommandArgument(redisClient *c, int i, robj *newval) {
+ robj *oldval;
+
+ redisAssertWithInfo(c,NULL,i < c->argc);
+ oldval = c->argv[i];
+ c->argv[i] = newval;
+ incrRefCount(newval);
+ decrRefCount(oldval);
+
+ /* If this is the command name make sure to fix c->cmd. */
+ if (i == 0) {
+ c->cmd = lookupCommand(c->argv[0]->ptr);
+ 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 length limits. */
+unsigned long getClientOutputBufferMemoryUsage(redisClient *c) {
+ unsigned long list_item_size = sizeof(listNode)+sizeof(robj);
+
+ 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;
+}
+
+int getClientLimitClassByName(char *name) {
+ if (!strcasecmp(name,"normal")) return REDIS_CLIENT_LIMIT_CLASS_NORMAL;
+ else if (!strcasecmp(name,"slave")) return REDIS_CLIENT_LIMIT_CLASS_SLAVE;
+ else if (!strcasecmp(name,"pubsub")) return REDIS_CLIENT_LIMIT_CLASS_PUBSUB;
+ else return -1;
+}
+
+char *getClientLimitClassName(int class) {
+ switch(class) {
+ case REDIS_CLIENT_LIMIT_CLASS_NORMAL: return "normal";
+ case REDIS_CLIENT_LIMIT_CLASS_SLAVE: return "slave";
+ case REDIS_CLIENT_LIMIT_CLASS_PUBSUB: return "pubsub";
+ default: return NULL;
+ }
+}
+
+/* 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->reply_bytes == 0 || c->flags & REDIS_CLOSE_ASAP) return;
+ if (checkClientOutputBufferLimits(c)) {
+ sds client = getClientInfoString(c);
+
+ freeClientAsync(c);
+ redisLog(REDIS_WARNING,"Client %s scheduled to be closed ASAP for overcoming of output buffer limits.", client);
+ sdsfree(client);
+ }
+}
+
+/* Helper function used by freeMemoryIfNeeded() in order to flush slaves
+ * output buffers without returning control to the event loop. */
+void flushSlavesOutputBuffers(void) {
+ listIter li;
+ listNode *ln;
+
+ listRewind(server.slaves,&li);
+ while((ln = listNext(&li))) {
+ redisClient *slave = listNodeValue(ln);
+ int events;
+
+ events = aeGetFileEvents(server.el,slave->fd);
+ if (events & AE_WRITABLE &&
+ slave->replstate == REDIS_REPL_ONLINE &&
+ listLength(slave->reply))
+ {
+ sendReplyToClient(server.el,slave->fd,slave,0);
+ }
+ }
projects / redis.git / blobdiff