Commit bfe85f7c authored by antirez's avatar antirez
Browse files

Merge branch 'unstable'

parents b8513c93 6b52ad87
...@@ -53,5 +53,6 @@ int anetWrite(int fd, char *buf, int count); ...@@ -53,5 +53,6 @@ int anetWrite(int fd, char *buf, int count);
int anetNonBlock(char *err, int fd); int anetNonBlock(char *err, int fd);
int anetTcpNoDelay(char *err, int fd); int anetTcpNoDelay(char *err, int fd);
int anetTcpKeepAlive(char *err, int fd); int anetTcpKeepAlive(char *err, int fd);
int anetPeerToString(int fd, char *ip, int *port);
#endif #endif
...@@ -284,9 +284,11 @@ int loadAppendOnlyFile(char *filename) { ...@@ -284,9 +284,11 @@ int loadAppendOnlyFile(char *filename) {
/* The fake client should not have a reply */ /* The fake client should not have a reply */
redisAssert(fakeClient->bufpos == 0 && listLength(fakeClient->reply) == 0); redisAssert(fakeClient->bufpos == 0 && listLength(fakeClient->reply) == 0);
/* Clean up, ready for the next command */ /* Clean up. Command code may have changed argv/argc so we use the
for (j = 0; j < argc; j++) decrRefCount(argv[j]); * argv/argc of the client instead of the local variables. */
zfree(argv); for (j = 0; j < fakeClient->argc; j++)
decrRefCount(fakeClient->argv[j]);
zfree(fakeClient->argv);
} }
/* This point can only be reached when EOF is reached without errors. /* This point can only be reached when EOF is reached without errors.
...@@ -346,7 +348,7 @@ int rewriteAppendOnlyFile(char *filename) { ...@@ -346,7 +348,7 @@ int rewriteAppendOnlyFile(char *filename) {
/* Iterate this DB writing every entry */ /* Iterate this DB writing every entry */
while((de = dictNext(di)) != NULL) { while((de = dictNext(di)) != NULL) {
sds keystr = dictGetEntryKey(de); sds keystr;
robj key, *o; robj key, *o;
time_t expiretime; time_t expiretime;
...@@ -429,12 +431,43 @@ int rewriteAppendOnlyFile(char *filename) { ...@@ -429,12 +431,43 @@ int rewriteAppendOnlyFile(char *filename) {
} }
} else if (o->type == REDIS_ZSET) { } else if (o->type == REDIS_ZSET) {
/* Emit the ZADDs needed to rebuild the sorted set */ /* Emit the ZADDs needed to rebuild the sorted set */
char cmd[]="*4\r\n$4\r\nZADD\r\n";
if (o->encoding == REDIS_ENCODING_ZIPLIST) {
unsigned char *zl = o->ptr;
unsigned char *eptr, *sptr;
unsigned char *vstr;
unsigned int vlen;
long long vll;
double score;
eptr = ziplistIndex(zl,0);
redisAssert(eptr != NULL);
sptr = ziplistNext(zl,eptr);
redisAssert(sptr != NULL);
while (eptr != NULL) {
redisAssert(ziplistGet(eptr,&vstr,&vlen,&vll));
score = zzlGetScore(sptr);
if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
if (fwriteBulkObject(fp,&key) == 0) goto werr;
if (fwriteBulkDouble(fp,score) == 0) goto werr;
if (vstr != NULL) {
if (fwriteBulkString(fp,(char*)vstr,vlen) == 0)
goto werr;
} else {
if (fwriteBulkLongLong(fp,vll) == 0)
goto werr;
}
zzlNext(zl,&eptr,&sptr);
}
} else if (o->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = o->ptr; zset *zs = o->ptr;
dictIterator *di = dictGetIterator(zs->dict); dictIterator *di = dictGetIterator(zs->dict);
dictEntry *de; dictEntry *de;
while((de = dictNext(di)) != NULL) { while((de = dictNext(di)) != NULL) {
char cmd[]="*4\r\n$4\r\nZADD\r\n";
robj *eleobj = dictGetEntryKey(de); robj *eleobj = dictGetEntryKey(de);
double *score = dictGetEntryVal(de); double *score = dictGetEntryVal(de);
...@@ -444,6 +477,9 @@ int rewriteAppendOnlyFile(char *filename) { ...@@ -444,6 +477,9 @@ int rewriteAppendOnlyFile(char *filename) {
if (fwriteBulkObject(fp,eleobj) == 0) goto werr; if (fwriteBulkObject(fp,eleobj) == 0) goto werr;
} }
dictReleaseIterator(di); dictReleaseIterator(di);
} else {
redisPanic("Unknown sorted set encoding");
}
} else if (o->type == REDIS_HASH) { } else if (o->type == REDIS_HASH) {
char cmd[]="*4\r\n$4\r\nHSET\r\n"; char cmd[]="*4\r\n$4\r\nHSET\r\n";
......
char *ascii_logo =
" _._ \n"
" _.-``__ ''-._ \n"
" _.-`` `. `_. ''-._ Redis %s (%s/%d) %s bit\n"
" .-`` .-```. ```\\/ _.,_ ''-._ \n"
" ( ' , .-` | `, ) Running in %s mode\n"
" |`-._`-...-` __...-.``-._|'` _.-'| Port: %d\n"
" | `-._ `._ / _.-' | PID: %ld\n"
" `-._ `-._ `-./ _.-' _.-' \n"
" |`-._`-._ `-.__.-' _.-'_.-'| \n"
" | `-._`-._ _.-'_.-' | http://redis.io \n"
" `-._ `-._`-.__.-'_.-' _.-' \n"
" |`-._`-._ `-.__.-' _.-'_.-'| \n"
" | `-._`-._ _.-'_.-' | \n"
" `-._ `-._`-.__.-'_.-' _.-' \n"
" `-._ `-.__.-' _.-' \n"
" `-._ _.-' \n"
" `-.__.-' \n\n";
This diff is collapsed.
...@@ -261,6 +261,10 @@ void loadServerConfig(char *filename) { ...@@ -261,6 +261,10 @@ void loadServerConfig(char *filename) {
server.list_max_ziplist_value = memtoll(argv[1], NULL); server.list_max_ziplist_value = memtoll(argv[1], NULL);
} else if (!strcasecmp(argv[0],"set-max-intset-entries") && argc == 2) { } else if (!strcasecmp(argv[0],"set-max-intset-entries") && argc == 2) {
server.set_max_intset_entries = memtoll(argv[1], NULL); server.set_max_intset_entries = memtoll(argv[1], NULL);
} else if (!strcasecmp(argv[0],"zset-max-ziplist-entries") && argc == 2) {
server.zset_max_ziplist_entries = memtoll(argv[1], NULL);
} else if (!strcasecmp(argv[0],"zset-max-ziplist-value") && argc == 2) {
server.zset_max_ziplist_value = memtoll(argv[1], NULL);
} else if (!strcasecmp(argv[0],"rename-command") && argc == 3) { } else if (!strcasecmp(argv[0],"rename-command") && argc == 3) {
struct redisCommand *cmd = lookupCommand(argv[1]); struct redisCommand *cmd = lookupCommand(argv[1]);
int retval; int retval;
...@@ -285,6 +289,13 @@ void loadServerConfig(char *filename) { ...@@ -285,6 +289,13 @@ void loadServerConfig(char *filename) {
err = "Target command name already exists"; goto loaderr; err = "Target command name already exists"; goto loaderr;
} }
} }
} else if (!strcasecmp(argv[0],"cluster-enabled") && argc == 2) {
if ((server.cluster_enabled = yesnotoi(argv[1])) == -1) {
err = "argument must be 'yes' or 'no'"; goto loaderr;
}
} else if (!strcasecmp(argv[0],"cluster-config-file") && argc == 2) {
zfree(server.cluster.configfile);
server.cluster.configfile = zstrdup(argv[1]);
} else { } else {
err = "Bad directive or wrong number of arguments"; goto loaderr; err = "Bad directive or wrong number of arguments"; goto loaderr;
} }
...@@ -443,6 +454,12 @@ void configSetCommand(redisClient *c) { ...@@ -443,6 +454,12 @@ void configSetCommand(redisClient *c) {
} else if (!strcasecmp(c->argv[2]->ptr,"set-max-intset-entries")) { } else if (!strcasecmp(c->argv[2]->ptr,"set-max-intset-entries")) {
if (getLongLongFromObject(o,&ll) == REDIS_ERR || ll < 0) goto badfmt; if (getLongLongFromObject(o,&ll) == REDIS_ERR || ll < 0) goto badfmt;
server.set_max_intset_entries = ll; server.set_max_intset_entries = ll;
} else if (!strcasecmp(c->argv[2]->ptr,"zset-max-ziplist-entries")) {
if (getLongLongFromObject(o,&ll) == REDIS_ERR || ll < 0) goto badfmt;
server.zset_max_ziplist_entries = ll;
} else if (!strcasecmp(c->argv[2]->ptr,"zset-max-ziplist-value")) {
if (getLongLongFromObject(o,&ll) == REDIS_ERR || ll < 0) goto badfmt;
server.zset_max_ziplist_value = ll;
} else { } else {
addReplyErrorFormat(c,"Unsupported CONFIG parameter: %s", addReplyErrorFormat(c,"Unsupported CONFIG parameter: %s",
(char*)c->argv[2]->ptr); (char*)c->argv[2]->ptr);
...@@ -594,6 +611,16 @@ void configGetCommand(redisClient *c) { ...@@ -594,6 +611,16 @@ void configGetCommand(redisClient *c) {
addReplyBulkLongLong(c,server.set_max_intset_entries); addReplyBulkLongLong(c,server.set_max_intset_entries);
matches++; matches++;
} }
if (stringmatch(pattern,"zset-max-ziplist-entries",0)) {
addReplyBulkCString(c,"zset-max-ziplist-entries");
addReplyBulkLongLong(c,server.zset_max_ziplist_entries);
matches++;
}
if (stringmatch(pattern,"zset-max-ziplist-value",0)) {
addReplyBulkCString(c,"zset-max-ziplist-value");
addReplyBulkLongLong(c,server.zset_max_ziplist_value);
matches++;
}
setDeferredMultiBulkLength(c,replylen,matches*2); setDeferredMultiBulkLength(c,replylen,matches*2);
} }
......
...@@ -10,18 +10,31 @@ ...@@ -10,18 +10,31 @@
* this expects a different allocation scheme. Therefore, *exclusively* use * this expects a different allocation scheme. Therefore, *exclusively* use
* either tcmalloc or OSX's malloc_size()! */ * either tcmalloc or OSX's malloc_size()! */
#if defined(USE_TCMALLOC) #if defined(USE_TCMALLOC)
#define REDIS_MALLOC "tcmalloc"
#include <google/tcmalloc.h> #include <google/tcmalloc.h>
#if TC_VERSION_MAJOR >= 1 && TC_VERSION_MINOR >= 6 #if TC_VERSION_MAJOR >= 1 && TC_VERSION_MINOR >= 6
#define HAVE_MALLOC_SIZE 1 #define HAVE_MALLOC_SIZE 1
#define redis_malloc_size(p) tc_malloc_size(p) #define redis_malloc_size(p) tc_malloc_size(p)
#endif #endif
#elif defined(USE_JEMALLOC)
#define REDIS_MALLOC "jemalloc"
#define JEMALLOC_MANGLE
#include <jemalloc/jemalloc.h>
#if JEMALLOC_VERSION_MAJOR >= 2 && JEMALLOC_VERSION_MINOR >= 1
#define HAVE_MALLOC_SIZE 1
#define redis_malloc_size(p) JEMALLOC_P(malloc_usable_size)(p)
#endif
#elif defined(__APPLE__) #elif defined(__APPLE__)
#include <malloc/malloc.h> #include <malloc/malloc.h>
#define HAVE_MALLOC_SIZE 1 #define HAVE_MALLOC_SIZE 1
#define redis_malloc_size(p) malloc_size(p) #define redis_malloc_size(p) malloc_size(p)
#endif #endif
/* Tefine redis_fstat to fstat or fstat64() */ #ifndef REDIS_MALLOC
#define REDIS_MALLOC "libc"
#endif
/* Define redis_fstat to fstat or fstat64() */
#if defined(__APPLE__) && !defined(MAC_OS_X_VERSION_10_6) #if defined(__APPLE__) && !defined(MAC_OS_X_VERSION_10_6)
#define redis_fstat fstat64 #define redis_fstat fstat64
#define redis_stat stat64 #define redis_stat stat64
......
#include "redis.h"
/*
* Copyright 2001-2010 Georges Menie (www.menie.org)
* Copyright 2010 Salvatore Sanfilippo (adapted to Redis coding style)
* All rights reserved.
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* * Neither the name of the University of California, Berkeley nor the
* names of its contributors may be used to endorse or promote products
* derived from this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND ANY
* EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
* DISCLAIMED. IN NO EVENT SHALL THE REGENTS AND CONTRIBUTORS BE LIABLE FOR ANY
* DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
* (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
* LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
* ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
* SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/
/* CRC16 implementation acording to CCITT standards */
static const uint16_t crc16tab[256]= {
0x0000,0x1021,0x2042,0x3063,0x4084,0x50a5,0x60c6,0x70e7,
0x8108,0x9129,0xa14a,0xb16b,0xc18c,0xd1ad,0xe1ce,0xf1ef,
0x1231,0x0210,0x3273,0x2252,0x52b5,0x4294,0x72f7,0x62d6,
0x9339,0x8318,0xb37b,0xa35a,0xd3bd,0xc39c,0xf3ff,0xe3de,
0x2462,0x3443,0x0420,0x1401,0x64e6,0x74c7,0x44a4,0x5485,
0xa56a,0xb54b,0x8528,0x9509,0xe5ee,0xf5cf,0xc5ac,0xd58d,
0x3653,0x2672,0x1611,0x0630,0x76d7,0x66f6,0x5695,0x46b4,
0xb75b,0xa77a,0x9719,0x8738,0xf7df,0xe7fe,0xd79d,0xc7bc,
0x48c4,0x58e5,0x6886,0x78a7,0x0840,0x1861,0x2802,0x3823,
0xc9cc,0xd9ed,0xe98e,0xf9af,0x8948,0x9969,0xa90a,0xb92b,
0x5af5,0x4ad4,0x7ab7,0x6a96,0x1a71,0x0a50,0x3a33,0x2a12,
0xdbfd,0xcbdc,0xfbbf,0xeb9e,0x9b79,0x8b58,0xbb3b,0xab1a,
0x6ca6,0x7c87,0x4ce4,0x5cc5,0x2c22,0x3c03,0x0c60,0x1c41,
0xedae,0xfd8f,0xcdec,0xddcd,0xad2a,0xbd0b,0x8d68,0x9d49,
0x7e97,0x6eb6,0x5ed5,0x4ef4,0x3e13,0x2e32,0x1e51,0x0e70,
0xff9f,0xefbe,0xdfdd,0xcffc,0xbf1b,0xaf3a,0x9f59,0x8f78,
0x9188,0x81a9,0xb1ca,0xa1eb,0xd10c,0xc12d,0xf14e,0xe16f,
0x1080,0x00a1,0x30c2,0x20e3,0x5004,0x4025,0x7046,0x6067,
0x83b9,0x9398,0xa3fb,0xb3da,0xc33d,0xd31c,0xe37f,0xf35e,
0x02b1,0x1290,0x22f3,0x32d2,0x4235,0x5214,0x6277,0x7256,
0xb5ea,0xa5cb,0x95a8,0x8589,0xf56e,0xe54f,0xd52c,0xc50d,
0x34e2,0x24c3,0x14a0,0x0481,0x7466,0x6447,0x5424,0x4405,
0xa7db,0xb7fa,0x8799,0x97b8,0xe75f,0xf77e,0xc71d,0xd73c,
0x26d3,0x36f2,0x0691,0x16b0,0x6657,0x7676,0x4615,0x5634,
0xd94c,0xc96d,0xf90e,0xe92f,0x99c8,0x89e9,0xb98a,0xa9ab,
0x5844,0x4865,0x7806,0x6827,0x18c0,0x08e1,0x3882,0x28a3,
0xcb7d,0xdb5c,0xeb3f,0xfb1e,0x8bf9,0x9bd8,0xabbb,0xbb9a,
0x4a75,0x5a54,0x6a37,0x7a16,0x0af1,0x1ad0,0x2ab3,0x3a92,
0xfd2e,0xed0f,0xdd6c,0xcd4d,0xbdaa,0xad8b,0x9de8,0x8dc9,
0x7c26,0x6c07,0x5c64,0x4c45,0x3ca2,0x2c83,0x1ce0,0x0cc1,
0xef1f,0xff3e,0xcf5d,0xdf7c,0xaf9b,0xbfba,0x8fd9,0x9ff8,
0x6e17,0x7e36,0x4e55,0x5e74,0x2e93,0x3eb2,0x0ed1,0x1ef0
};
uint16_t crc16(const char *buf, int len) {
int counter;
uint16_t crc = 0;
for (counter = 0; counter < len; counter++)
crc = (crc<<8) ^ crc16tab[((crc>>8) ^ *buf++)&0x00FF];
return crc;
}
...@@ -2,6 +2,9 @@ ...@@ -2,6 +2,9 @@
#include <signal.h> #include <signal.h>
void SlotToKeyAdd(robj *key);
void SlotToKeyDel(robj *key);
/*----------------------------------------------------------------------------- /*-----------------------------------------------------------------------------
* C-level DB API * C-level DB API
*----------------------------------------------------------------------------*/ *----------------------------------------------------------------------------*/
...@@ -131,6 +134,7 @@ int dbAdd(redisDb *db, robj *key, robj *val) { ...@@ -131,6 +134,7 @@ int dbAdd(redisDb *db, robj *key, robj *val) {
sds copy = sdsdup(key->ptr); sds copy = sdsdup(key->ptr);
dictAdd(db->dict, copy, val); dictAdd(db->dict, copy, val);
if (server.ds_enabled) cacheSetKeyMayExist(db,key); if (server.ds_enabled) cacheSetKeyMayExist(db,key);
if (server.cluster_enabled) SlotToKeyAdd(key);
return REDIS_OK; return REDIS_OK;
} }
} }
...@@ -146,6 +150,7 @@ int dbReplace(redisDb *db, robj *key, robj *val) { ...@@ -146,6 +150,7 @@ int dbReplace(redisDb *db, robj *key, robj *val) {
if ((oldval = dictFetchValue(db->dict,key->ptr)) == NULL) { if ((oldval = dictFetchValue(db->dict,key->ptr)) == NULL) {
sds copy = sdsdup(key->ptr); sds copy = sdsdup(key->ptr);
dictAdd(db->dict, copy, val); dictAdd(db->dict, copy, val);
if (server.cluster_enabled) SlotToKeyAdd(key);
retval = 1; retval = 1;
} else { } else {
dictReplace(db->dict, key->ptr, val); dictReplace(db->dict, key->ptr, val);
...@@ -198,7 +203,12 @@ int dbDelete(redisDb *db, robj *key) { ...@@ -198,7 +203,12 @@ int dbDelete(redisDb *db, robj *key) {
/* Deleting an entry from the expires dict will not free the sds of /* Deleting an entry from the expires dict will not free the sds of
* the key, because it is shared with the main dictionary. */ * the key, because it is shared with the main dictionary. */
if (dictSize(db->expires) > 0) dictDelete(db->expires,key->ptr); if (dictSize(db->expires) > 0) dictDelete(db->expires,key->ptr);
return dictDelete(db->dict,key->ptr) == DICT_OK; if (dictDelete(db->dict,key->ptr) == DICT_OK) {
if (server.cluster_enabled) SlotToKeyDel(key);
return 1;
} else {
return 0;
}
} }
/* Empty the whole database. /* Empty the whole database.
...@@ -307,6 +317,10 @@ void existsCommand(redisClient *c) { ...@@ -307,6 +317,10 @@ void existsCommand(redisClient *c) {
void selectCommand(redisClient *c) { void selectCommand(redisClient *c) {
int id = atoi(c->argv[1]->ptr); int id = atoi(c->argv[1]->ptr);
if (server.cluster_enabled && id != 0) {
addReplyError(c,"SELECT is not allowed in cluster mode");
return;
}
if (selectDb(c,id) == REDIS_ERR) { if (selectDb(c,id) == REDIS_ERR) {
addReplyError(c,"invalid DB index"); addReplyError(c,"invalid DB index");
} else { } else {
...@@ -428,6 +442,11 @@ void moveCommand(redisClient *c) { ...@@ -428,6 +442,11 @@ void moveCommand(redisClient *c) {
redisDb *src, *dst; redisDb *src, *dst;
int srcid; int srcid;
if (server.cluster_enabled) {
addReplyError(c,"MOVE is not allowed in cluster mode");
return;
}
/* Obtain source and target DB pointers */ /* Obtain source and target DB pointers */
src = c->db; src = c->db;
srcid = c->db->id; srcid = c->db->id;
...@@ -616,3 +635,108 @@ void persistCommand(redisClient *c) { ...@@ -616,3 +635,108 @@ void persistCommand(redisClient *c) {
} }
} }
} }
/* -----------------------------------------------------------------------------
* API to get key arguments from commands
* ---------------------------------------------------------------------------*/
int *getKeysUsingCommandTable(struct redisCommand *cmd,robj **argv, int argc, int *numkeys) {
int j, i = 0, last, *keys;
REDIS_NOTUSED(argv);
if (cmd->firstkey == 0) {
*numkeys = 0;
return NULL;
}
last = cmd->lastkey;
if (last < 0) last = argc+last;
keys = zmalloc(sizeof(int)*((last - cmd->firstkey)+1));
for (j = cmd->firstkey; j <= last; j += cmd->keystep) {
redisAssert(j < argc);
keys[i++] = j;
}
*numkeys = i;
return keys;
}
int *getKeysFromCommand(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags) {
if (cmd->getkeys_proc) {
return cmd->getkeys_proc(cmd,argv,argc,numkeys,flags);
} else {
return getKeysUsingCommandTable(cmd,argv,argc,numkeys);
}
}
void getKeysFreeResult(int *result) {
zfree(result);
}
int *noPreloadGetKeys(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags) {
if (flags & REDIS_GETKEYS_PRELOAD) {
*numkeys = 0;
return NULL;
} else {
return getKeysUsingCommandTable(cmd,argv,argc,numkeys);
}
}
int *renameGetKeys(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags) {
if (flags & REDIS_GETKEYS_PRELOAD) {
int *keys = zmalloc(sizeof(int));
*numkeys = 1;
keys[0] = 1;
return keys;
} else {
return getKeysUsingCommandTable(cmd,argv,argc,numkeys);
}
}
int *zunionInterGetKeys(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags) {
int i, num, *keys;
REDIS_NOTUSED(cmd);
REDIS_NOTUSED(flags);
num = atoi(argv[2]->ptr);
/* Sanity check. Don't return any key if the command is going to
* reply with syntax error. */
if (num > (argc-3)) {
*numkeys = 0;
return NULL;
}
keys = zmalloc(sizeof(int)*num);
for (i = 0; i < num; i++) keys[i] = 3+i;
*numkeys = num;
return keys;
}
/* Slot to Key API. This is used by Redis Cluster in order to obtain in
* a fast way a key that belongs to a specified hash slot. This is useful
* while rehashing the cluster. */
void SlotToKeyAdd(robj *key) {
unsigned int hashslot = keyHashSlot(key->ptr,sdslen(key->ptr));
zslInsert(server.cluster.slots_to_keys,hashslot,key);
incrRefCount(key);
}
void SlotToKeyDel(robj *key) {
unsigned int hashslot = keyHashSlot(key->ptr,sdslen(key->ptr));
zslDelete(server.cluster.slots_to_keys,hashslot,key);
}
unsigned int GetKeysInSlot(unsigned int hashslot, robj **keys, unsigned int count) {
zskiplistNode *n;
zrangespec range;
int j = 0;
range.min = range.max = hashslot;
range.minex = range.maxex = 0;
n = zslFirstInRange(server.cluster.slots_to_keys, range);
while(n && n->score == hashslot && count--) {
keys[j++] = n->obj;
n = n->level[0].forward;
}
return j;
}
...@@ -100,7 +100,7 @@ void computeDatasetDigest(unsigned char *final) { ...@@ -100,7 +100,7 @@ void computeDatasetDigest(unsigned char *final) {
mixDigest(digest,key,sdslen(key)); mixDigest(digest,key,sdslen(key));
/* Make sure the key is loaded if VM is active */ /* Make sure the key is loaded if VM is active */
o = lookupKeyRead(db,keyobj); o = dictGetEntryVal(de);
aux = htonl(o->type); aux = htonl(o->type);
mixDigest(digest,&aux,sizeof(aux)); mixDigest(digest,&aux,sizeof(aux));
...@@ -127,6 +127,39 @@ void computeDatasetDigest(unsigned char *final) { ...@@ -127,6 +127,39 @@ void computeDatasetDigest(unsigned char *final) {
} }
setTypeReleaseIterator(si); setTypeReleaseIterator(si);
} else if (o->type == REDIS_ZSET) { } else if (o->type == REDIS_ZSET) {
unsigned char eledigest[20];
if (o->encoding == REDIS_ENCODING_ZIPLIST) {
unsigned char *zl = o->ptr;
unsigned char *eptr, *sptr;
unsigned char *vstr;
unsigned int vlen;
long long vll;
double score;
eptr = ziplistIndex(zl,0);
redisAssert(eptr != NULL);
sptr = ziplistNext(zl,eptr);
redisAssert(sptr != NULL);
while (eptr != NULL) {
redisAssert(ziplistGet(eptr,&vstr,&vlen,&vll));
score = zzlGetScore(sptr);
memset(eledigest,0,20);
if (vstr != NULL) {
mixDigest(eledigest,vstr,vlen);
} else {
ll2string(buf,sizeof(buf),vll);
mixDigest(eledigest,buf,strlen(buf));
}
snprintf(buf,sizeof(buf),"%.17g",score);
mixDigest(eledigest,buf,strlen(buf));
xorDigest(digest,eledigest,20);
zzlNext(zl,&eptr,&sptr);
}
} else if (o->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = o->ptr; zset *zs = o->ptr;
dictIterator *di = dictGetIterator(zs->dict); dictIterator *di = dictGetIterator(zs->dict);
dictEntry *de; dictEntry *de;
...@@ -134,7 +167,6 @@ void computeDatasetDigest(unsigned char *final) { ...@@ -134,7 +167,6 @@ void computeDatasetDigest(unsigned char *final) {
while((de = dictNext(di)) != NULL) { while((de = dictNext(di)) != NULL) {
robj *eleobj = dictGetEntryKey(de); robj *eleobj = dictGetEntryKey(de);
double *score = dictGetEntryVal(de); double *score = dictGetEntryVal(de);
unsigned char eledigest[20];
snprintf(buf,sizeof(buf),"%.17g",*score); snprintf(buf,sizeof(buf),"%.17g",*score);
memset(eledigest,0,20); memset(eledigest,0,20);
...@@ -143,6 +175,9 @@ void computeDatasetDigest(unsigned char *final) { ...@@ -143,6 +175,9 @@ void computeDatasetDigest(unsigned char *final) {
xorDigest(digest,eledigest,20); xorDigest(digest,eledigest,20);
} }
dictReleaseIterator(di); dictReleaseIterator(di);
} else {
redisPanic("Unknown sorted set encoding");
}
} else if (o->type == REDIS_HASH) { } else if (o->type == REDIS_HASH) {
hashTypeIterator *hi; hashTypeIterator *hi;
robj *obj; robj *obj;
......
...@@ -245,9 +245,9 @@ int dictRehashMilliseconds(dict *d, int ms) { ...@@ -245,9 +245,9 @@ int dictRehashMilliseconds(dict *d, int ms) {
} }
/* This function performs just a step of rehashing, and only if there are /* This function performs just a step of rehashing, and only if there are
* not iterators bound to our hash table. When we have iterators in the middle * no safe iterators bound to our hash table. When we have iterators in the
* of a rehashing we can't mess with the two hash tables otherwise some element * middle of a rehashing we can't mess with the two hash tables otherwise
* can be missed or duplicated. * some element can be missed or duplicated.
* *
* This function is called by common lookup or update operations in the * This function is called by common lookup or update operations in the
* dictionary so that the hash table automatically migrates from H1 to H2 * dictionary so that the hash table automatically migrates from H1 to H2
...@@ -424,17 +424,26 @@ dictIterator *dictGetIterator(dict *d) ...@@ -424,17 +424,26 @@ dictIterator *dictGetIterator(dict *d)
iter->d = d; iter->d = d;
iter->table = 0; iter->table = 0;
iter->index = -1; iter->index = -1;
iter->safe = 0;
iter->entry = NULL; iter->entry = NULL;
iter->nextEntry = NULL; iter->nextEntry = NULL;
return iter; return iter;
} }
dictIterator *dictGetSafeIterator(dict *d) {
dictIterator *i = dictGetIterator(d);
i->safe = 1;
return i;
}
dictEntry *dictNext(dictIterator *iter) dictEntry *dictNext(dictIterator *iter)
{ {
while (1) { while (1) {
if (iter->entry == NULL) { if (iter->entry == NULL) {
dictht *ht = &iter->d->ht[iter->table]; dictht *ht = &iter->d->ht[iter->table];
if (iter->index == -1 && iter->table == 0) iter->d->iterators++; if (iter->safe && iter->index == -1 && iter->table == 0)
iter->d->iterators++;
iter->index++; iter->index++;
if (iter->index >= (signed) ht->size) { if (iter->index >= (signed) ht->size) {
if (dictIsRehashing(iter->d) && iter->table == 0) { if (dictIsRehashing(iter->d) && iter->table == 0) {
...@@ -461,7 +470,8 @@ dictEntry *dictNext(dictIterator *iter) ...@@ -461,7 +470,8 @@ dictEntry *dictNext(dictIterator *iter)
void dictReleaseIterator(dictIterator *iter) void dictReleaseIterator(dictIterator *iter)
{ {
if (!(iter->index == -1 && iter->table == 0)) iter->d->iterators--; if (iter->safe && !(iter->index == -1 && iter->table == 0))
iter->d->iterators--;
zfree(iter); zfree(iter);
} }
......
...@@ -74,10 +74,13 @@ typedef struct dict { ...@@ -74,10 +74,13 @@ typedef struct dict {
int iterators; /* number of iterators currently running */ int iterators; /* number of iterators currently running */
} dict; } dict;
/* If safe is set to 1 this is a safe iteartor, that means, you can call
* dictAdd, dictFind, and other functions against the dictionary even while
* iterating. Otherwise it is a non safe iterator, and only dictNext()
* should be called while iterating. */
typedef struct dictIterator { typedef struct dictIterator {
dict *d; dict *d;
int table; int table, index, safe;
int index;
dictEntry *entry, *nextEntry; dictEntry *entry, *nextEntry;
} dictIterator; } dictIterator;
...@@ -132,6 +135,7 @@ dictEntry * dictFind(dict *d, const void *key); ...@@ -132,6 +135,7 @@ dictEntry * dictFind(dict *d, const void *key);
void *dictFetchValue(dict *d, const void *key); void *dictFetchValue(dict *d, const void *key);
int dictResize(dict *d); int dictResize(dict *d);
dictIterator *dictGetIterator(dict *d); dictIterator *dictGetIterator(dict *d);
dictIterator *dictGetSafeIterator(dict *d);
dictEntry *dictNext(dictIterator *iter); dictEntry *dictNext(dictIterator *iter);
void dictReleaseIterator(dictIterator *iter); void dictReleaseIterator(dictIterator *iter);
dictEntry *dictGetRandomKey(dict *d); dictEntry *dictGetRandomKey(dict *d);
......
...@@ -903,59 +903,6 @@ int waitForSwappedKey(redisClient *c, robj *key) { ...@@ -903,59 +903,6 @@ int waitForSwappedKey(redisClient *c, robj *key) {
return 1; return 1;
} }
/* Preload keys for any command with first, last and step values for
* the command keys prototype, as defined in the command table. */
void waitForMultipleSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) {
int j, last;
if (cmd->vm_firstkey == 0) return;
last = cmd->vm_lastkey;
if (last < 0) last = argc+last;
for (j = cmd->vm_firstkey; j <= last; j += cmd->vm_keystep) {
redisAssert(j < argc);
waitForSwappedKey(c,argv[j]);
}
}
/* Preload keys needed for the ZUNIONSTORE and ZINTERSTORE commands.
* Note that the number of keys to preload is user-defined, so we need to
* apply a sanity check against argc. */
void zunionInterBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) {
int i, num;
REDIS_NOTUSED(cmd);
num = atoi(argv[2]->ptr);
if (num > (argc-3)) return;
for (i = 0; i < num; i++) {
waitForSwappedKey(c,argv[3+i]);
}
}
/* Preload keys needed to execute the entire MULTI/EXEC block.
*
* This function is called by blockClientOnSwappedKeys when EXEC is issued,
* and will block the client when any command requires a swapped out value. */
void execBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv) {
int i, margc;
struct redisCommand *mcmd;
robj **margv;
REDIS_NOTUSED(cmd);
REDIS_NOTUSED(argc);
REDIS_NOTUSED(argv);
if (!(c->flags & REDIS_MULTI)) return;
for (i = 0; i < c->mstate.count; i++) {
mcmd = c->mstate.commands[i].cmd;
margc = c->mstate.commands[i].argc;
margv = c->mstate.commands[i].argv;
if (mcmd->vm_preload_proc != NULL) {
mcmd->vm_preload_proc(c,mcmd,margc,margv);
} else {
waitForMultipleSwappedKeys(c,mcmd,margc,margv);
}
}
}
/* Is this client attempting to run a command against swapped keys? /* Is this client attempting to run a command against swapped keys?
* If so, block it ASAP, load the keys in background, then resume it. * If so, block it ASAP, load the keys in background, then resume it.
* *
...@@ -967,10 +914,39 @@ void execBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int ...@@ -967,10 +914,39 @@ void execBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int
* Return 1 if the client is marked as blocked, 0 if the client can * Return 1 if the client is marked as blocked, 0 if the client can
* continue as the keys it is going to access appear to be in memory. */ * continue as the keys it is going to access appear to be in memory. */
int blockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd) { int blockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd) {
if (cmd->vm_preload_proc != NULL) { int *keyindex, numkeys, j, i;
cmd->vm_preload_proc(c,cmd,c->argc,c->argv);
/* EXEC is a special case, we need to preload all the commands
* queued into the transaction */
if (cmd->proc == execCommand) {
struct redisCommand *mcmd;
robj **margv;
int margc;
if (!(c->flags & REDIS_MULTI)) return 0;
for (i = 0; i < c->mstate.count; i++) {
mcmd = c->mstate.commands[i].cmd;
margc = c->mstate.commands[i].argc;
margv = c->mstate.commands[i].argv;
keyindex = getKeysFromCommand(mcmd,margv,margc,&numkeys,
REDIS_GETKEYS_PRELOAD);
for (j = 0; j < numkeys; j++) {
redisLog(REDIS_DEBUG,"Preloading %s",
(char*)margv[keyindex[j]]->ptr);
waitForSwappedKey(c,margv[keyindex[j]]);
}
getKeysFreeResult(keyindex);
}
} else { } else {
waitForMultipleSwappedKeys(c,cmd,c->argc,c->argv); keyindex = getKeysFromCommand(cmd,c->argv,c->argc,&numkeys,
REDIS_GETKEYS_PRELOAD);
for (j = 0; j < numkeys; j++) {
redisLog(REDIS_DEBUG,"Preloading %s",
(char*)c->argv[keyindex[j]]->ptr);
waitForSwappedKey(c,c->argv[keyindex[j]]);
}
getKeysFreeResult(keyindex);
} }
/* If the client was blocked for at least one key, mark it as blocked. */ /* If the client was blocked for at least one key, mark it as blocked. */
......
...@@ -16,7 +16,6 @@ redisClient *createClient(int fd) { ...@@ -16,7 +16,6 @@ redisClient *createClient(int fd) {
anetNonBlock(NULL,fd); anetNonBlock(NULL,fd);
anetTcpNoDelay(NULL,fd); anetTcpNoDelay(NULL,fd);
if (!c) return NULL;
if (aeCreateFileEvent(server.el,fd,AE_READABLE, if (aeCreateFileEvent(server.el,fd,AE_READABLE,
readQueryFromClient, c) == AE_ERR) readQueryFromClient, c) == AE_ERR)
{ {
...@@ -60,9 +59,6 @@ redisClient *createClient(int fd) { ...@@ -60,9 +59,6 @@ redisClient *createClient(int fd) {
/* Set the event loop to listen for write events on the client's socket. /* Set the event loop to listen for write events on the client's socket.
* Typically gets called every time a reply is built. */ * Typically gets called every time a reply is built. */
int _installWriteEvent(redisClient *c) { int _installWriteEvent(redisClient *c) {
/* When CLOSE_AFTER_REPLY is set, no more replies may be added! */
redisAssert(!(c->flags & REDIS_CLOSE_AFTER_REPLY));
if (c->fd <= 0) return REDIS_ERR; if (c->fd <= 0) return REDIS_ERR;
if (c->bufpos == 0 && listLength(c->reply) == 0 && if (c->bufpos == 0 && listLength(c->reply) == 0 &&
(c->replstate == REDIS_REPL_NONE || (c->replstate == REDIS_REPL_NONE ||
...@@ -88,9 +84,15 @@ robj *dupLastObjectIfNeeded(list *reply) { ...@@ -88,9 +84,15 @@ robj *dupLastObjectIfNeeded(list *reply) {
return listNodeValue(ln); return listNodeValue(ln);
} }
/* -----------------------------------------------------------------------------
* Low level functions to add more data to output buffers.
* -------------------------------------------------------------------------- */
int _addReplyToBuffer(redisClient *c, char *s, size_t len) { int _addReplyToBuffer(redisClient *c, char *s, size_t len) {
size_t available = sizeof(c->buf)-c->bufpos; size_t available = sizeof(c->buf)-c->bufpos;
if (c->flags & REDIS_CLOSE_AFTER_REPLY) return REDIS_OK;
/* If there already are entries in the reply list, we cannot /* If there already are entries in the reply list, we cannot
* add anything more to the static buffer. */ * add anything more to the static buffer. */
if (listLength(c->reply) > 0) return REDIS_ERR; if (listLength(c->reply) > 0) return REDIS_ERR;
...@@ -105,6 +107,9 @@ int _addReplyToBuffer(redisClient *c, char *s, size_t len) { ...@@ -105,6 +107,9 @@ int _addReplyToBuffer(redisClient *c, char *s, size_t len) {
void _addReplyObjectToList(redisClient *c, robj *o) { void _addReplyObjectToList(redisClient *c, robj *o) {
robj *tail; robj *tail;
if (c->flags & REDIS_CLOSE_AFTER_REPLY) return;
if (listLength(c->reply) == 0) { if (listLength(c->reply) == 0) {
incrRefCount(o); incrRefCount(o);
listAddNodeTail(c->reply,o); listAddNodeTail(c->reply,o);
...@@ -128,6 +133,12 @@ void _addReplyObjectToList(redisClient *c, robj *o) { ...@@ -128,6 +133,12 @@ void _addReplyObjectToList(redisClient *c, robj *o) {
* needed it will be free'd, otherwise it ends up in a robj. */ * needed it will be free'd, otherwise it ends up in a robj. */
void _addReplySdsToList(redisClient *c, sds s) { void _addReplySdsToList(redisClient *c, sds s) {
robj *tail; robj *tail;
if (c->flags & REDIS_CLOSE_AFTER_REPLY) {
sdsfree(s);
return;
}
if (listLength(c->reply) == 0) { if (listLength(c->reply) == 0) {
listAddNodeTail(c->reply,createObject(REDIS_STRING,s)); listAddNodeTail(c->reply,createObject(REDIS_STRING,s));
} else { } else {
...@@ -148,6 +159,9 @@ void _addReplySdsToList(redisClient *c, sds s) { ...@@ -148,6 +159,9 @@ void _addReplySdsToList(redisClient *c, sds s) {
void _addReplyStringToList(redisClient *c, char *s, size_t len) { void _addReplyStringToList(redisClient *c, char *s, size_t len) {
robj *tail; robj *tail;
if (c->flags & REDIS_CLOSE_AFTER_REPLY) return;
if (listLength(c->reply) == 0) { if (listLength(c->reply) == 0) {
listAddNodeTail(c->reply,createStringObject(s,len)); listAddNodeTail(c->reply,createStringObject(s,len));
} else { } else {
...@@ -165,6 +179,11 @@ void _addReplyStringToList(redisClient *c, char *s, size_t len) { ...@@ -165,6 +179,11 @@ void _addReplyStringToList(redisClient *c, char *s, size_t len) {
} }
} }
/* -----------------------------------------------------------------------------
* Higher level functions to queue data on the client output buffer.
* The following functions are the ones that commands implementations will call.
* -------------------------------------------------------------------------- */
void addReply(redisClient *c, robj *obj) { void addReply(redisClient *c, robj *obj) {
if (_installWriteEvent(c) != REDIS_OK) return; if (_installWriteEvent(c) != REDIS_OK) return;
...@@ -301,6 +320,11 @@ void _addReplyLongLong(redisClient *c, long long ll, char prefix) { ...@@ -301,6 +320,11 @@ void _addReplyLongLong(redisClient *c, long long ll, char prefix) {
} }
void addReplyLongLong(redisClient *c, long long ll) { void addReplyLongLong(redisClient *c, long long ll) {
if (ll == 0)
addReply(c,shared.czero);
else if (ll == 1)
addReply(c,shared.cone);
else
_addReplyLongLong(c,ll,':'); _addReplyLongLong(c,ll,':');
} }
...@@ -395,7 +419,7 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -395,7 +419,7 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
cfd = anetTcpAccept(server.neterr, fd, cip, &cport); cfd = anetTcpAccept(server.neterr, fd, cip, &cport);
if (cfd == AE_ERR) { if (cfd == AE_ERR) {
redisLog(REDIS_VERBOSE,"Accepting client connection: %s", server.neterr); redisLog(REDIS_WARNING,"Accepting client connection: %s", server.neterr);
return; return;
} }
redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport); redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport);
...@@ -410,7 +434,7 @@ void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -410,7 +434,7 @@ void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
cfd = anetUnixAccept(server.neterr, fd); cfd = anetUnixAccept(server.neterr, fd);
if (cfd == AE_ERR) { if (cfd == AE_ERR) {
redisLog(REDIS_VERBOSE,"Accepting client connection: %s", server.neterr); redisLog(REDIS_WARNING,"Accepting client connection: %s", server.neterr);
return; return;
} }
redisLog(REDIS_VERBOSE,"Accepted connection to %s", server.unixsocket); redisLog(REDIS_VERBOSE,"Accepted connection to %s", server.unixsocket);
...@@ -502,12 +526,18 @@ void freeClient(redisClient *c) { ...@@ -502,12 +526,18 @@ void freeClient(redisClient *c) {
* close the connection with all our slaves if we have any, so * 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 * 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 * with us as well. Note that also when the slave is not connected
* to the master it will keep refusing connections by other slaves. */ * to the master it will keep refusing connections by other slaves.
*
* We do this only if server.masterhost != NULL. If it is NULL this
* means the user called SLAVEOF NO ONE and we are freeing our
* link with the master, so no need to close link with slaves. */
if (server.masterhost != NULL) {
while (listLength(server.slaves)) { while (listLength(server.slaves)) {
ln = listFirst(server.slaves); ln = listFirst(server.slaves);
freeClient((redisClient*)ln->value); freeClient((redisClient*)ln->value);
} }
} }
}
/* Release memory */ /* Release memory */
zfree(c->argv); zfree(c->argv);
freeClientMultiState(c); freeClientMultiState(c);
...@@ -670,47 +700,57 @@ static void setProtocolError(redisClient *c, int pos) { ...@@ -670,47 +700,57 @@ static void setProtocolError(redisClient *c, int pos) {
int processMultibulkBuffer(redisClient *c) { int processMultibulkBuffer(redisClient *c) {
char *newline = NULL; char *newline = NULL;
char *eptr; int pos = 0, ok;
int pos = 0, tolerr; long long ll;
long bulklen;
if (c->multibulklen == 0) { if (c->multibulklen == 0) {
/* The client should have been reset */ /* The client should have been reset */
redisAssert(c->argc == 0); redisAssert(c->argc == 0);
/* Multi bulk length cannot be read without a \r\n */ /* Multi bulk length cannot be read without a \r\n */
newline = strstr(c->querybuf,"\r\n"); newline = strchr(c->querybuf,'\r');
if (newline == NULL) if (newline == NULL)
return REDIS_ERR; return REDIS_ERR;
/* Buffer should also contain \n */
if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2))
return REDIS_ERR;
/* We know for sure there is a whole line since newline != NULL, /* We know for sure there is a whole line since newline != NULL,
* so go ahead and find out the multi bulk length. */ * so go ahead and find out the multi bulk length. */
redisAssert(c->querybuf[0] == '*'); redisAssert(c->querybuf[0] == '*');
c->multibulklen = strtol(c->querybuf+1,&eptr,10); ok = string2ll(c->querybuf+1,newline-(c->querybuf+1),&ll);
pos = (newline-c->querybuf)+2; if (!ok || ll > 1024*1024) {
if (c->multibulklen <= 0) {
c->querybuf = sdsrange(c->querybuf,pos,-1);
return REDIS_OK;
} else if (c->multibulklen > 1024*1024) {
addReplyError(c,"Protocol error: invalid multibulk length"); addReplyError(c,"Protocol error: invalid multibulk length");
setProtocolError(c,pos); setProtocolError(c,pos);
return REDIS_ERR; return REDIS_ERR;
} }
pos = (newline-c->querybuf)+2;
if (ll <= 0) {
c->querybuf = sdsrange(c->querybuf,pos,-1);
return REDIS_OK;
}
c->multibulklen = ll;
/* 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);
/* Search new newline */
newline = strstr(c->querybuf+pos,"\r\n");
} }
redisAssert(c->multibulklen > 0); redisAssert(c->multibulklen > 0);
while(c->multibulklen) { while(c->multibulklen) {
/* Read bulk length if unknown */ /* Read bulk length if unknown */
if (c->bulklen == -1) { if (c->bulklen == -1) {
newline = strstr(c->querybuf+pos,"\r\n"); newline = strchr(c->querybuf+pos,'\r');
if (newline != NULL) { if (newline == NULL)
break;
/* Buffer should also contain \n */
if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2))
break;
if (c->querybuf[pos] != '$') { if (c->querybuf[pos] != '$') {
addReplyErrorFormat(c, addReplyErrorFormat(c,
"Protocol error: expected '$', got '%c'", "Protocol error: expected '$', got '%c'",
...@@ -719,21 +759,15 @@ int processMultibulkBuffer(redisClient *c) { ...@@ -719,21 +759,15 @@ int processMultibulkBuffer(redisClient *c) {
return REDIS_ERR; return REDIS_ERR;
} }
bulklen = strtol(c->querybuf+pos+1,&eptr,10); ok = string2ll(c->querybuf+pos+1,newline-(c->querybuf+pos+1),&ll);
tolerr = (eptr[0] != '\r'); if (!ok || ll < 0 || ll > 512*1024*1024) {
if (tolerr || bulklen == LONG_MIN || bulklen == LONG_MAX ||
bulklen < 0 || bulklen > 512*1024*1024)
{
addReplyError(c,"Protocol error: invalid bulk length"); addReplyError(c,"Protocol error: invalid bulk length");
setProtocolError(c,pos); setProtocolError(c,pos);
return REDIS_ERR; return REDIS_ERR;
} }
pos += eptr-(c->querybuf+pos)+2;
c->bulklen = bulklen; pos += newline-(c->querybuf+pos)+2;
} else { c->bulklen = ll;
/* No newline in current buffer, so wait for more data */
break;
}
} }
/* Read bulk argument */ /* Read bulk argument */
...@@ -845,3 +879,70 @@ void getClientsMaxBuffers(unsigned long *longest_output_list, ...@@ -845,3 +879,70 @@ void getClientsMaxBuffers(unsigned long *longest_output_list,
*biggest_input_buffer = bib; *biggest_input_buffer = bib;
} }
void clientCommand(redisClient *c) {
listNode *ln;
listIter li;
redisClient *client;
if (!strcasecmp(c->argv[1]->ptr,"list") && c->argc == 2) {
sds o = sdsempty();
time_t now = time(NULL);
listRewind(server.clients,&li);
while ((ln = listNext(&li)) != NULL) {
char ip[32], flags[16], *p;
int port;
client = listNodeValue(ln);
if (anetPeerToString(client->fd,ip,&port) == -1) continue;
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 (p == flags) *p++ = 'N';
if (client->flags & REDIS_MULTI) *p++ = 'x';
if (client->flags & REDIS_BLOCKED) *p++ = 'b';
if (client->flags & REDIS_IO_WAIT) *p++ = 'i';
if (client->flags & REDIS_DIRTY_CAS) *p++ = 'd';
if (client->flags & REDIS_CLOSE_AFTER_REPLY) *p++ = 'c';
if (client->flags & REDIS_UNBLOCKED) *p++ = 'u';
*p++ = '\0';
o = sdscatprintf(o,
"addr=%s:%d fd=%d idle=%ld flags=%s db=%d sub=%d psub=%d\n",
ip,port,client->fd,
(long)(now - client->lastinteraction),
flags,
client->db->id,
(int) dictSize(client->pubsub_channels),
(int) listLength(client->pubsub_patterns));
}
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)");
}
}
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
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