Commit 2379e5db authored by Oran Agra's avatar Oran Agra Committed by antirez
Browse files

memory reporting of clients argv

track and report memory used by clients argv.
this is very usaful in case clients started sending a command and didn't
complete it. in which case the first args of the command are already
trimmed from the query buffer.
parent fe434069
...@@ -634,6 +634,7 @@ struct client *createFakeClient(void) { ...@@ -634,6 +634,7 @@ struct client *createFakeClient(void) {
c->querybuf_peak = 0; c->querybuf_peak = 0;
c->argc = 0; c->argc = 0;
c->argv = NULL; c->argv = NULL;
c->argv_bytes = 0;
c->bufpos = 0; c->bufpos = 0;
c->flags = 0; c->flags = 0;
c->btype = BLOCKED_NONE; c->btype = BLOCKED_NONE;
...@@ -657,6 +658,7 @@ void freeFakeClientArgv(struct client *c) { ...@@ -657,6 +658,7 @@ void freeFakeClientArgv(struct client *c) {
for (j = 0; j < c->argc; j++) for (j = 0; j < c->argc; j++)
decrRefCount(c->argv[j]); decrRefCount(c->argv[j]);
zfree(c->argv); zfree(c->argv);
c->argv_bytes = 0;
} }
void freeFakeClient(struct client *c) { void freeFakeClient(struct client *c) {
......
...@@ -44,7 +44,7 @@ size_t sdsZmallocSize(sds s) { ...@@ -44,7 +44,7 @@ size_t sdsZmallocSize(sds s) {
} }
/* Return the amount of memory used by the sds string at object->ptr /* Return the amount of memory used by the sds string at object->ptr
* for a string object. */ * for a string object. This includes internal fragmentation. */
size_t getStringObjectSdsUsedMemory(robj *o) { size_t getStringObjectSdsUsedMemory(robj *o) {
serverAssertWithInfo(NULL,o,o->type == OBJ_STRING); serverAssertWithInfo(NULL,o,o->type == OBJ_STRING);
switch(o->encoding) { switch(o->encoding) {
...@@ -54,6 +54,17 @@ size_t getStringObjectSdsUsedMemory(robj *o) { ...@@ -54,6 +54,17 @@ size_t getStringObjectSdsUsedMemory(robj *o) {
} }
} }
/* Return approximate memory used by the sds string at object->ptr
* for a string object. This method does not include internal fragmentation. */
size_t getStringObjectSize(robj *o) {
serverAssertWithInfo(NULL,o,o->type == OBJ_STRING);
switch(o->encoding) {
case OBJ_ENCODING_RAW: return sdsAllocSize(o->ptr);
case OBJ_ENCODING_EMBSTR: return sdsAllocSize(o->ptr);
default: return 0; /* Just integer encoding for now. */
}
}
/* Client.reply list dup and free methods. */ /* Client.reply list dup and free methods. */
void *dupClientReplyValue(void *o) { void *dupClientReplyValue(void *o) {
clientReplyBlock *old = o; clientReplyBlock *old = o;
...@@ -117,6 +128,7 @@ client *createClient(int fd) { ...@@ -117,6 +128,7 @@ client *createClient(int fd) {
c->reqtype = 0; c->reqtype = 0;
c->argc = 0; c->argc = 0;
c->argv = NULL; c->argv = NULL;
c->argv_bytes = 0;
c->cmd = c->lastcmd = NULL; c->cmd = c->lastcmd = NULL;
c->multibulklen = 0; c->multibulklen = 0;
c->bulklen = -1; c->bulklen = -1;
...@@ -757,6 +769,7 @@ static void freeClientArgv(client *c) { ...@@ -757,6 +769,7 @@ static void freeClientArgv(client *c) {
decrRefCount(c->argv[j]); decrRefCount(c->argv[j]);
c->argc = 0; c->argc = 0;
c->cmd = NULL; c->cmd = NULL;
c->argv_bytes = 0;
} }
/* Close all the slaves connections. This is useful in chained replication /* Close all the slaves connections. This is useful in chained replication
...@@ -903,6 +916,7 @@ void freeClient(client *c) { ...@@ -903,6 +916,7 @@ void freeClient(client *c) {
* and finally release the client structure itself. */ * and finally release the client structure itself. */
if (c->name) decrRefCount(c->name); if (c->name) decrRefCount(c->name);
zfree(c->argv); zfree(c->argv);
c->argv_bytes = 0;
freeClientMultiState(c); freeClientMultiState(c);
sdsfree(c->peerid); sdsfree(c->peerid);
zfree(c); zfree(c);
...@@ -1158,12 +1172,14 @@ int processInlineBuffer(client *c) { ...@@ -1158,12 +1172,14 @@ int processInlineBuffer(client *c) {
if (argc) { if (argc) {
if (c->argv) zfree(c->argv); if (c->argv) zfree(c->argv);
c->argv = zmalloc(sizeof(robj*)*argc); c->argv = zmalloc(sizeof(robj*)*argc);
c->argv_bytes = 0;
} }
/* Create redis objects for all arguments. */ /* Create redis objects for all arguments. */
for (c->argc = 0, j = 0; j < argc; j++) { for (c->argc = 0, j = 0; j < argc; j++) {
if (sdslen(argv[j])) { if (sdslen(argv[j])) {
c->argv[c->argc] = createObject(OBJ_STRING,argv[j]); c->argv[c->argc] = createObject(OBJ_STRING,argv[j]);
c->argv_bytes += getStringObjectSize(c->argv[c->argc]);
c->argc++; c->argc++;
} else { } else {
sdsfree(argv[j]); sdsfree(argv[j]);
...@@ -1256,6 +1272,7 @@ int processMultibulkBuffer(client *c) { ...@@ -1256,6 +1272,7 @@ int processMultibulkBuffer(client *c) {
/* Setup argv array on client structure */ /* Setup argv array on client structure */
if (c->argv) zfree(c->argv); if (c->argv) zfree(c->argv);
c->argv = zmalloc(sizeof(robj*)*c->multibulklen); c->argv = zmalloc(sizeof(robj*)*c->multibulklen);
c->argv_bytes = 0;
} }
serverAssertWithInfo(c,NULL,c->multibulklen > 0); serverAssertWithInfo(c,NULL,c->multibulklen > 0);
...@@ -1326,15 +1343,19 @@ int processMultibulkBuffer(client *c) { ...@@ -1326,15 +1343,19 @@ int processMultibulkBuffer(client *c) {
c->bulklen >= PROTO_MBULK_BIG_ARG && c->bulklen >= PROTO_MBULK_BIG_ARG &&
sdslen(c->querybuf) == (size_t)(c->bulklen+2)) sdslen(c->querybuf) == (size_t)(c->bulklen+2))
{ {
c->argv[c->argc++] = createObject(OBJ_STRING,c->querybuf); c->argv[c->argc] = createObject(OBJ_STRING,c->querybuf);
c->argv_bytes += getStringObjectSize(c->argv[c->argc]);
c->argc++;
sdsIncrLen(c->querybuf,-2); /* remove CRLF */ sdsIncrLen(c->querybuf,-2); /* remove CRLF */
/* Assume that if we saw a fat argument we'll see another one /* Assume that if we saw a fat argument we'll see another one
* likely... */ * likely... */
c->querybuf = sdsnewlen(SDS_NOINIT,c->bulklen+2); c->querybuf = sdsnewlen(SDS_NOINIT,c->bulklen+2);
sdsclear(c->querybuf); sdsclear(c->querybuf);
} else { } else {
c->argv[c->argc++] = c->argv[c->argc] =
createStringObject(c->querybuf+c->qb_pos,c->bulklen); createStringObject(c->querybuf+c->qb_pos,c->bulklen);
c->argv_bytes += getStringObjectSize(c->argv[c->argc]);
c->argc++;
c->qb_pos += c->bulklen+2; c->qb_pos += c->bulklen+2;
} }
c->bulklen = -1; c->bulklen = -1;
...@@ -1605,7 +1626,7 @@ sds catClientInfoString(sds s, client *client) { ...@@ -1605,7 +1626,7 @@ sds catClientInfoString(sds s, client *client) {
if (emask & AE_WRITABLE) *p++ = 'w'; if (emask & AE_WRITABLE) *p++ = 'w';
*p = '\0'; *p = '\0';
return sdscatfmt(s, return sdscatfmt(s,
"id=%U addr=%s fd=%i name=%s age=%I idle=%I flags=%s db=%i sub=%i psub=%i multi=%i qbuf=%U qbuf-free=%U obl=%U oll=%U omem=%U events=%s cmd=%s", "id=%U addr=%s fd=%i name=%s age=%I idle=%I flags=%s db=%i sub=%i psub=%i multi=%i qbuf=%U qbuf-free=%U argv=%U obl=%U oll=%U omem=%U events=%s cmd=%s",
(unsigned long long) client->id, (unsigned long long) client->id,
getClientPeerId(client), getClientPeerId(client),
client->fd, client->fd,
...@@ -1619,9 +1640,10 @@ sds catClientInfoString(sds s, client *client) { ...@@ -1619,9 +1640,10 @@ sds catClientInfoString(sds s, client *client) {
(client->flags & CLIENT_MULTI) ? client->mstate.count : -1, (client->flags & CLIENT_MULTI) ? client->mstate.count : -1,
(unsigned long long) sdslen(client->querybuf), (unsigned long long) sdslen(client->querybuf),
(unsigned long long) sdsavail(client->querybuf), (unsigned long long) sdsavail(client->querybuf),
(unsigned long long) zmalloc_size(client->argv) + client->argv_bytes,
(unsigned long long) client->bufpos, (unsigned long long) client->bufpos,
(unsigned long long) listLength(client->reply), (unsigned long long) listLength(client->reply),
(unsigned long long) getClientOutputBufferMemoryUsage(client), (unsigned long long) getClientOutputBufferMemoryUsage(client) + sizeof(client->buf),
events, events,
client->lastcmd ? client->lastcmd->name : "NULL"); client->lastcmd ? client->lastcmd->name : "NULL");
} }
...@@ -1907,6 +1929,10 @@ void rewriteClientCommandVector(client *c, int argc, ...) { ...@@ -1907,6 +1929,10 @@ void rewriteClientCommandVector(client *c, int argc, ...) {
/* Replace argv and argc with our new versions. */ /* Replace argv and argc with our new versions. */
c->argv = argv; c->argv = argv;
c->argc = argc; c->argc = argc;
c->argv_bytes = 0;
for (j = 0; j < c->argc; j++)
if (c->argv[j])
c->argv_bytes += getStringObjectSize(c->argv[j]);
c->cmd = lookupCommandOrOriginal(c->argv[0]->ptr); c->cmd = lookupCommandOrOriginal(c->argv[0]->ptr);
serverAssertWithInfo(c,NULL,c->cmd != NULL); serverAssertWithInfo(c,NULL,c->cmd != NULL);
va_end(ap); va_end(ap);
...@@ -1914,10 +1940,15 @@ void rewriteClientCommandVector(client *c, int argc, ...) { ...@@ -1914,10 +1940,15 @@ void rewriteClientCommandVector(client *c, int argc, ...) {
/* Completely replace the client command vector with the provided one. */ /* Completely replace the client command vector with the provided one. */
void replaceClientCommandVector(client *c, int argc, robj **argv) { void replaceClientCommandVector(client *c, int argc, robj **argv) {
int j;
freeClientArgv(c); freeClientArgv(c);
zfree(c->argv); zfree(c->argv);
c->argv = argv; c->argv = argv;
c->argc = argc; c->argc = argc;
c->argv_bytes = 0;
for (j = 0; j < c->argc; j++)
if (c->argv[j])
c->argv_bytes += getStringObjectSize(c->argv[j]);
c->cmd = lookupCommandOrOriginal(c->argv[0]->ptr); c->cmd = lookupCommandOrOriginal(c->argv[0]->ptr);
serverAssertWithInfo(c,NULL,c->cmd != NULL); serverAssertWithInfo(c,NULL,c->cmd != NULL);
} }
...@@ -1942,6 +1973,8 @@ void rewriteClientCommandArgument(client *c, int i, robj *newval) { ...@@ -1942,6 +1973,8 @@ void rewriteClientCommandArgument(client *c, int i, robj *newval) {
c->argv[i] = NULL; c->argv[i] = NULL;
} }
oldval = c->argv[i]; oldval = c->argv[i];
if (oldval) c->argv_bytes -= getStringObjectSize(oldval);
if (newval) c->argv_bytes += getStringObjectSize(newval);
c->argv[i] = newval; c->argv[i] = newval;
incrRefCount(newval); incrRefCount(newval);
if (oldval) decrRefCount(oldval); if (oldval) decrRefCount(oldval);
......
...@@ -986,6 +986,9 @@ struct redisMemOverhead *getMemoryOverheadData(void) { ...@@ -986,6 +986,9 @@ struct redisMemOverhead *getMemoryOverheadData(void) {
mem += getClientOutputBufferMemoryUsage(c); mem += getClientOutputBufferMemoryUsage(c);
mem += sdsAllocSize(c->querybuf); mem += sdsAllocSize(c->querybuf);
mem += sizeof(client); mem += sizeof(client);
mem += c->argv_bytes;
if (c->argv)
mem += zmalloc_size(c->argv);
} }
} }
mh->clients_slaves = mem; mh->clients_slaves = mem;
...@@ -1004,6 +1007,9 @@ struct redisMemOverhead *getMemoryOverheadData(void) { ...@@ -1004,6 +1007,9 @@ struct redisMemOverhead *getMemoryOverheadData(void) {
mem += getClientOutputBufferMemoryUsage(c); mem += getClientOutputBufferMemoryUsage(c);
mem += sdsAllocSize(c->querybuf); mem += sdsAllocSize(c->querybuf);
mem += sizeof(client); mem += sizeof(client);
mem += c->argv_bytes;
if (c->argv)
mem += zmalloc_size(c->argv);
} }
} }
mh->clients_normal = mem; mh->clients_normal = mem;
......
...@@ -719,6 +719,7 @@ typedef struct client { ...@@ -719,6 +719,7 @@ typedef struct client {
size_t querybuf_peak; /* Recent (100ms or more) peak of querybuf size. */ size_t querybuf_peak; /* Recent (100ms or more) peak of querybuf size. */
int argc; /* Num of arguments of current command. */ int argc; /* Num of arguments of current command. */
robj **argv; /* Arguments of current command. */ robj **argv; /* Arguments of current command. */
unsigned long long argv_bytes; /* Tot bytes of objects in argv list. */
struct redisCommand *cmd, *lastcmd; /* Last command executed. */ struct redisCommand *cmd, *lastcmd; /* Last command executed. */
int reqtype; /* Request protocol type: PROTO_REQ_* */ int reqtype; /* Request protocol type: PROTO_REQ_* */
int multibulklen; /* Number of multi bulk arguments left to read. */ int multibulklen; /* Number of multi bulk arguments left to read. */
......
start_server {tags {"introspection"}} { start_server {tags {"introspection"}} {
test {CLIENT LIST} { test {CLIENT LIST} {
r client list r client list
} {*addr=*:* fd=* age=* idle=* flags=N db=9 sub=0 psub=0 multi=-1 qbuf=26 qbuf-free=* obl=0 oll=0 omem=0 events=r cmd=client*} } {*addr=*:* fd=* age=* idle=* flags=N db=9 sub=0 psub=0 multi=-1 qbuf=26 qbuf-free=* argv=* obl=0 oll=0 omem=0 events=r cmd=client*}
test {MONITOR can log executed commands} { test {MONITOR can log executed commands} {
set rd [redis_deferring_client] set rd [redis_deferring_client]
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment