Commit 9e67df2a authored by Oran Agra's avatar Oran Agra
Browse files

rebase from unstable

parents f472bb10 bcb4d091
This diff is collapsed.
...@@ -318,13 +318,17 @@ void delCommand(redisClient *c) { ...@@ -318,13 +318,17 @@ void delCommand(redisClient *c) {
addReplyLongLong(c,deleted); addReplyLongLong(c,deleted);
} }
/* EXISTS key1 key2 ... key_N.
* Return value is the number of keys existing. */
void existsCommand(redisClient *c) { void existsCommand(redisClient *c) {
expireIfNeeded(c->db,c->argv[1]); long long count = 0;
if (dbExists(c->db,c->argv[1])) { int j;
addReply(c, shared.cone);
} else { for (j = 1; j < c->argc; j++) {
addReply(c, shared.czero); expireIfNeeded(c->db,c->argv[j]);
if (dbExists(c->db,c->argv[j])) count++;
} }
addReplyLongLong(c,count);
} }
void selectCommand(redisClient *c) { void selectCommand(redisClient *c) {
......
...@@ -425,6 +425,27 @@ void debugCommand(redisClient *c) { ...@@ -425,6 +425,27 @@ void debugCommand(redisClient *c) {
sizes = sdscatprintf(sizes,"dictentry:%d ", (int)sizeof(dictEntry)); sizes = sdscatprintf(sizes,"dictentry:%d ", (int)sizeof(dictEntry));
sizes = sdscatprintf(sizes,"sdshdr:%d", (int)sizeof(struct sdshdr)); sizes = sdscatprintf(sizes,"sdshdr:%d", (int)sizeof(struct sdshdr));
addReplyBulkSds(c,sizes); addReplyBulkSds(c,sizes);
} else if (!strcasecmp(c->argv[1]->ptr,"htstats") && c->argc == 3) {
long dbid;
sds stats = sdsempty();
char buf[4096];
if (getLongFromObjectOrReply(c, c->argv[2], &dbid, NULL) != REDIS_OK)
return;
if (dbid < 0 || dbid >= server.dbnum) {
addReplyError(c,"Out of range database");
return;
}
stats = sdscatprintf(stats,"[Dictionary HT]\n");
dictGetStats(buf,sizeof(buf),server.db[dbid].dict);
stats = sdscat(stats,buf);
stats = sdscatprintf(stats,"[Expires HT]\n");
dictGetStats(buf,sizeof(buf),server.db[dbid].expires);
stats = sdscat(stats,buf);
addReplyBulkSds(c,stats);
} else if (!strcasecmp(c->argv[1]->ptr,"jemalloc") && c->argc == 3) { } else if (!strcasecmp(c->argv[1]->ptr,"jemalloc") && c->argc == 3) {
#if defined(USE_JEMALLOC) #if defined(USE_JEMALLOC)
if (!strcasecmp(c->argv[2]->ptr, "info")) { if (!strcasecmp(c->argv[2]->ptr, "info")) {
......
...@@ -687,10 +687,10 @@ dictEntry *dictGetRandomKey(dict *d) ...@@ -687,10 +687,10 @@ dictEntry *dictGetRandomKey(dict *d)
* statistics. However the function is much faster than dictGetRandomKey() * statistics. However the function is much faster than dictGetRandomKey()
* at producing N elements. */ * at producing N elements. */
unsigned int dictGetSomeKeys(dict *d, dictEntry **des, unsigned int count) { unsigned int dictGetSomeKeys(dict *d, dictEntry **des, unsigned int count) {
unsigned int j; /* internal hash table id, 0 or 1. */ unsigned long j; /* internal hash table id, 0 or 1. */
unsigned int tables; /* 1 or 2 tables? */ unsigned long tables; /* 1 or 2 tables? */
unsigned int stored = 0, maxsizemask; unsigned long stored = 0, maxsizemask;
unsigned int maxsteps; unsigned long maxsteps;
if (dictSize(d) < count) count = dictSize(d); if (dictSize(d) < count) count = dictSize(d);
maxsteps = count*10; maxsteps = count*10;
...@@ -709,14 +709,14 @@ unsigned int dictGetSomeKeys(dict *d, dictEntry **des, unsigned int count) { ...@@ -709,14 +709,14 @@ unsigned int dictGetSomeKeys(dict *d, dictEntry **des, unsigned int count) {
maxsizemask = d->ht[1].sizemask; maxsizemask = d->ht[1].sizemask;
/* Pick a random point inside the larger table. */ /* Pick a random point inside the larger table. */
unsigned int i = random() & maxsizemask; unsigned long i = random() & maxsizemask;
unsigned int emptylen = 0; /* Continuous empty entries so far. */ unsigned long emptylen = 0; /* Continuous empty entries so far. */
while(stored < count && maxsteps--) { while(stored < count && maxsteps--) {
for (j = 0; j < tables; j++) { for (j = 0; j < tables; j++) {
/* Invariant of the dict.c rehashing: up to the indexes already /* Invariant of the dict.c rehashing: up to the indexes already
* visited in ht[0] during the rehashing, there are no populated * visited in ht[0] during the rehashing, there are no populated
* buckets, so we can skip ht[0] for indexes between 0 and idx-1. */ * buckets, so we can skip ht[0] for indexes between 0 and idx-1. */
if (tables == 2 && j == 0 && i < d->rehashidx) { if (tables == 2 && j == 0 && i < (unsigned long) d->rehashidx) {
/* Moreover, if we are currently out of range in the second /* Moreover, if we are currently out of range in the second
* table, there will be no elements in both tables up to * table, there will be no elements in both tables up to
* the current rehashing index, so we jump if possible. * the current rehashing index, so we jump if possible.
...@@ -1002,24 +1002,21 @@ void dictDisableResize(void) { ...@@ -1002,24 +1002,21 @@ void dictDisableResize(void) {
dict_can_resize = 0; dict_can_resize = 0;
} }
#if 0 /* ------------------------------- Debugging ---------------------------------*/
/* The following is code that we don't use for Redis currently, but that is part
of the library. */
/* ----------------------- Debugging ------------------------*/
#define DICT_STATS_VECTLEN 50 #define DICT_STATS_VECTLEN 50
static void _dictPrintStatsHt(dictht *ht) { size_t _dictGetStatsHt(char *buf, size_t bufsize, dictht *ht, int tableid) {
unsigned long i, slots = 0, chainlen, maxchainlen = 0; unsigned long i, slots = 0, chainlen, maxchainlen = 0;
unsigned long totchainlen = 0; unsigned long totchainlen = 0;
unsigned long clvector[DICT_STATS_VECTLEN]; unsigned long clvector[DICT_STATS_VECTLEN];
size_t l = 0;
if (ht->used == 0) { if (ht->used == 0) {
printf("No stats available for empty dictionaries\n"); return snprintf(buf,bufsize,
return; "No stats available for empty dictionaries\n");
} }
/* Compute stats. */
for (i = 0; i < DICT_STATS_VECTLEN; i++) clvector[i] = 0; for (i = 0; i < DICT_STATS_VECTLEN; i++) clvector[i] = 0;
for (i = 0; i < ht->size; i++) { for (i = 0; i < ht->size; i++) {
dictEntry *he; dictEntry *he;
...@@ -1040,89 +1037,46 @@ static void _dictPrintStatsHt(dictht *ht) { ...@@ -1040,89 +1037,46 @@ static void _dictPrintStatsHt(dictht *ht) {
if (chainlen > maxchainlen) maxchainlen = chainlen; if (chainlen > maxchainlen) maxchainlen = chainlen;
totchainlen += chainlen; totchainlen += chainlen;
} }
printf("Hash table stats:\n");
printf(" table size: %ld\n", ht->size); /* Generate human readable stats. */
printf(" number of elements: %ld\n", ht->used); l += snprintf(buf+l,bufsize-l,
printf(" different slots: %ld\n", slots); "Hash table %d stats (%s):\n"
printf(" max chain length: %ld\n", maxchainlen); " table size: %ld\n"
printf(" avg chain length (counted): %.02f\n", (float)totchainlen/slots); " number of elements: %ld\n"
printf(" avg chain length (computed): %.02f\n", (float)ht->used/slots); " different slots: %ld\n"
printf(" Chain length distribution:\n"); " max chain length: %ld\n"
" avg chain length (counted): %.02f\n"
" avg chain length (computed): %.02f\n"
" Chain length distribution:\n",
tableid, (tableid == 0) ? "main hash table" : "rehashing target",
ht->size, ht->used, slots, maxchainlen,
(float)totchainlen/slots, (float)ht->used/slots);
for (i = 0; i < DICT_STATS_VECTLEN-1; i++) { for (i = 0; i < DICT_STATS_VECTLEN-1; i++) {
if (clvector[i] == 0) continue; if (clvector[i] == 0) continue;
printf(" %s%ld: %ld (%.02f%%)\n",(i == DICT_STATS_VECTLEN-1)?">= ":"", i, clvector[i], ((float)clvector[i]/ht->size)*100); if (l >= bufsize) break;
l += snprintf(buf+l,bufsize-l,
" %s%ld: %ld (%.02f%%)\n",
(i == DICT_STATS_VECTLEN-1)?">= ":"",
i, clvector[i], ((float)clvector[i]/ht->size)*100);
} }
}
void dictPrintStats(dict *d) {
_dictPrintStatsHt(&d->ht[0]);
if (dictIsRehashing(d)) {
printf("-- Rehashing into ht[1]:\n");
_dictPrintStatsHt(&d->ht[1]);
}
}
/* ----------------------- StringCopy Hash Table Type ------------------------*/
static unsigned int _dictStringCopyHTHashFunction(const void *key)
{
return dictGenHashFunction(key, strlen(key));
}
static void *_dictStringDup(void *privdata, const void *key)
{
int len = strlen(key);
char *copy = zmalloc(len+1);
DICT_NOTUSED(privdata);
memcpy(copy, key, len); /* Unlike snprintf(), teturn the number of characters actually written. */
copy[len] = '\0'; if (bufsize) buf[bufsize-1] = '\0';
return copy; return strlen(buf);
} }
static int _dictStringCopyHTKeyCompare(void *privdata, const void *key1, void dictGetStats(char *buf, size_t bufsize, dict *d) {
const void *key2) size_t l;
{ char *orig_buf = buf;
DICT_NOTUSED(privdata); size_t orig_bufsize = bufsize;
return strcmp(key1, key2) == 0; l = _dictGetStatsHt(buf,bufsize,&d->ht[0],0);
} buf += l;
bufsize -= l;
static void _dictStringDestructor(void *privdata, void *key) if (dictIsRehashing(d) && bufsize > 0) {
{ _dictGetStatsHt(buf,bufsize,&d->ht[1],1);
DICT_NOTUSED(privdata); }
/* Make sure there is a NULL term at the end. */
zfree(key); if (orig_bufsize) orig_buf[orig_bufsize-1] = '\0';
} }
dictType dictTypeHeapStringCopyKey = {
_dictStringCopyHTHashFunction, /* hash function */
_dictStringDup, /* key dup */
NULL, /* val dup */
_dictStringCopyHTKeyCompare, /* key compare */
_dictStringDestructor, /* key destructor */
NULL /* val destructor */
};
/* This is like StringCopy but does not auto-duplicate the key.
* It's used for intepreter's shared strings. */
dictType dictTypeHeapStrings = {
_dictStringCopyHTHashFunction, /* hash function */
NULL, /* key dup */
NULL, /* val dup */
_dictStringCopyHTKeyCompare, /* key compare */
_dictStringDestructor, /* key destructor */
NULL /* val destructor */
};
/* This is like StringCopy but also automatically handle dynamic
* allocated C strings as values. */
dictType dictTypeHeapStringCopyKeyValue = {
_dictStringCopyHTHashFunction, /* hash function */
_dictStringDup, /* key dup */
_dictStringDup, /* val dup */
_dictStringCopyHTKeyCompare, /* key compare */
_dictStringDestructor, /* key destructor */
_dictStringDestructor, /* val destructor */
};
#endif
...@@ -165,7 +165,7 @@ dictEntry *dictNext(dictIterator *iter); ...@@ -165,7 +165,7 @@ dictEntry *dictNext(dictIterator *iter);
void dictReleaseIterator(dictIterator *iter); void dictReleaseIterator(dictIterator *iter);
dictEntry *dictGetRandomKey(dict *d); dictEntry *dictGetRandomKey(dict *d);
unsigned int dictGetSomeKeys(dict *d, dictEntry **des, unsigned int count); unsigned int dictGetSomeKeys(dict *d, dictEntry **des, unsigned int count);
void dictPrintStats(dict *d); void dictGetStats(char *buf, size_t bufsize, dict *d);
unsigned int dictGenHashFunction(const void *key, int len); unsigned int dictGenHashFunction(const void *key, int len);
unsigned int dictGenCaseHashFunction(const unsigned char *buf, int len); unsigned int dictGenCaseHashFunction(const unsigned char *buf, int len);
void dictEmpty(dict *d, void(callback)(void*)); void dictEmpty(dict *d, void(callback)(void*));
......
This diff is collapsed.
#ifndef __GEO_H__
#define __GEO_H__
#include "redis.h"
/* Structures used inside geo.c in order to represent points and array of
* points on the earth. */
typedef struct geoPoint {
double longitude;
double latitude;
double dist;
double score;
char *member;
} geoPoint;
typedef struct geoArray {
struct geoPoint *array;
size_t buckets;
size_t used;
} geoArray;
#endif
...@@ -248,7 +248,7 @@ sds createLatencyReport(void) { ...@@ -248,7 +248,7 @@ sds createLatencyReport(void) {
dictEntry *de; dictEntry *de;
int eventnum = 0; int eventnum = 0;
di = dictGetIterator(server.latency_events); di = dictGetSafeIterator(server.latency_events);
while((de = dictNext(di)) != NULL) { while((de = dictNext(di)) != NULL) {
char *event = dictGetKey(de); char *event = dictGetKey(de);
struct latencyTimeSeries *ts = dictGetVal(de); struct latencyTimeSeries *ts = dictGetVal(de);
......
...@@ -135,23 +135,49 @@ redisClient *createClient(int fd) { ...@@ -135,23 +135,49 @@ redisClient *createClient(int fd) {
* returns REDIS_OK, and make sure to install the write handler in our event * 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. * 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, * If the client should not receive new data, because it is a fake client
* a master, a slave not yet online, or because the setup of the write handler * (used to load AOF in memory), a master or because the setup of the write
* failed, the function returns REDIS_ERR. * handler failed, the function returns REDIS_ERR.
*
* The function may return REDIS_OK without actually installing the write
* event handler in the following cases:
*
* 1) The event handler should already be installed since the output buffer
* already contained something.
* 2) The client is a slave but not yet online, so we want to just accumulate
* writes in the buffer but not actually sending them yet.
* *
* Typically gets called every time a reply is built, before adding more * 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 to the clients output buffers. If the function returns REDIS_ERR no
* data should be appended to the output buffers. */ * data should be appended to the output buffers. */
int prepareClientToWrite(redisClient *c) { int prepareClientToWrite(redisClient *c) {
/* If it's the Lua client we always return ok without installing any
* handler since there is no socket at all. */
if (c->flags & REDIS_LUA_CLIENT) return REDIS_OK; if (c->flags & REDIS_LUA_CLIENT) return REDIS_OK;
/* Masters don't receive replies, unless REDIS_MASTER_FORCE_REPLY flag
* is set. */
if ((c->flags & REDIS_MASTER) && if ((c->flags & REDIS_MASTER) &&
!(c->flags & REDIS_MASTER_FORCE_REPLY)) return REDIS_ERR; !(c->flags & REDIS_MASTER_FORCE_REPLY)) return REDIS_ERR;
if (c->fd <= 0) return REDIS_ERR; /* Fake client */
if (c->fd <= 0) return REDIS_ERR; /* Fake client for AOF loading. */
/* Only install the handler if not already installed and, in case of
* slaves, if the client can actually receive writes. */
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 ||
c->replstate == REDIS_REPL_ONLINE) && !c->repl_put_online_on_ack && (c->replstate == REDIS_REPL_ONLINE && !c->repl_put_online_on_ack)))
aeCreateFileEvent(server.el, c->fd, AE_WRITABLE, {
sendReplyToClient, c) == AE_ERR) return REDIS_ERR; /* Try to install the write handler. */
if (aeCreateFileEvent(server.el, c->fd, AE_WRITABLE,
sendReplyToClient, c) == AE_ERR)
{
freeClientAsync(c);
return REDIS_ERR;
}
}
/* Authorize the caller to queue in the output buffer of this client. */
return REDIS_OK; return REDIS_OK;
} }
...@@ -175,7 +201,7 @@ robj *dupLastObjectIfNeeded(list *reply) { ...@@ -175,7 +201,7 @@ robj *dupLastObjectIfNeeded(list *reply) {
* Low level functions to add more data to output buffers. * Low level functions to add more data to output buffers.
* -------------------------------------------------------------------------- */ * -------------------------------------------------------------------------- */
int _addReplyToBuffer(redisClient *c, char *s, size_t len) { int _addReplyToBuffer(redisClient *c, const 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 (c->flags & REDIS_CLOSE_AFTER_REPLY) return REDIS_OK;
...@@ -255,7 +281,7 @@ void _addReplySdsToList(redisClient *c, sds s) { ...@@ -255,7 +281,7 @@ void _addReplySdsToList(redisClient *c, sds s) {
asyncCloseClientOnOutputBufferLimitReached(c); asyncCloseClientOnOutputBufferLimitReached(c);
} }
void _addReplyStringToList(redisClient *c, char *s, size_t len) { void _addReplyStringToList(redisClient *c, const char *s, size_t len) {
robj *tail; robj *tail;
if (c->flags & REDIS_CLOSE_AFTER_REPLY) return; if (c->flags & REDIS_CLOSE_AFTER_REPLY) return;
...@@ -341,19 +367,19 @@ void addReplySds(redisClient *c, sds s) { ...@@ -341,19 +367,19 @@ void addReplySds(redisClient *c, sds s) {
} }
} }
void addReplyString(redisClient *c, char *s, size_t len) { void addReplyString(redisClient *c, const char *s, size_t len) {
if (prepareClientToWrite(c) != REDIS_OK) return; if (prepareClientToWrite(c) != REDIS_OK) return;
if (_addReplyToBuffer(c,s,len) != REDIS_OK) if (_addReplyToBuffer(c,s,len) != REDIS_OK)
_addReplyStringToList(c,s,len); _addReplyStringToList(c,s,len);
} }
void addReplyErrorLength(redisClient *c, char *s, size_t len) { void addReplyErrorLength(redisClient *c, const char *s, size_t len) {
addReplyString(c,"-ERR ",5); addReplyString(c,"-ERR ",5);
addReplyString(c,s,len); addReplyString(c,s,len);
addReplyString(c,"\r\n",2); addReplyString(c,"\r\n",2);
} }
void addReplyError(redisClient *c, char *err) { void addReplyError(redisClient *c, const char *err) {
addReplyErrorLength(c,err,strlen(err)); addReplyErrorLength(c,err,strlen(err));
} }
...@@ -373,13 +399,13 @@ void addReplyErrorFormat(redisClient *c, const char *fmt, ...) { ...@@ -373,13 +399,13 @@ void addReplyErrorFormat(redisClient *c, const char *fmt, ...) {
sdsfree(s); sdsfree(s);
} }
void addReplyStatusLength(redisClient *c, char *s, size_t len) { void addReplyStatusLength(redisClient *c, const char *s, size_t len) {
addReplyString(c,"+",1); addReplyString(c,"+",1);
addReplyString(c,s,len); addReplyString(c,s,len);
addReplyString(c,"\r\n",2); addReplyString(c,"\r\n",2);
} }
void addReplyStatus(redisClient *c, char *status) { void addReplyStatus(redisClient *c, const char *status) {
addReplyStatusLength(c,status,strlen(status)); addReplyStatusLength(c,status,strlen(status));
} }
...@@ -454,10 +480,10 @@ void addReplyLongLongWithPrefix(redisClient *c, long long ll, char prefix) { ...@@ -454,10 +480,10 @@ void addReplyLongLongWithPrefix(redisClient *c, long long ll, char prefix) {
/* Things like $3\r\n or *2\r\n are emitted very often by the protocol /* Things like $3\r\n or *2\r\n are emitted very often by the protocol
* so we have a few shared objects to use if the integer is small * so we have a few shared objects to use if the integer is small
* like it is most of the times. */ * like it is most of the times. */
if (prefix == '*' && ll < REDIS_SHARED_BULKHDR_LEN) { if (prefix == '*' && ll < REDIS_SHARED_BULKHDR_LEN && ll >= 0) {
addReply(c,shared.mbulkhdr[ll]); addReply(c,shared.mbulkhdr[ll]);
return; return;
} else if (prefix == '$' && ll < REDIS_SHARED_BULKHDR_LEN) { } else if (prefix == '$' && ll < REDIS_SHARED_BULKHDR_LEN && ll >= 0) {
addReply(c,shared.bulkhdr[ll]); addReply(c,shared.bulkhdr[ll]);
return; return;
} }
...@@ -519,7 +545,7 @@ void addReplyBulk(redisClient *c, robj *obj) { ...@@ -519,7 +545,7 @@ void addReplyBulk(redisClient *c, robj *obj) {
} }
/* Add a C buffer as bulk reply */ /* Add a C buffer as bulk reply */
void addReplyBulkCBuffer(redisClient *c, void *p, size_t len) { void addReplyBulkCBuffer(redisClient *c, const void *p, size_t len) {
addReplyLongLongWithPrefix(c,len,'$'); addReplyLongLongWithPrefix(c,len,'$');
addReplyString(c,p,len); addReplyString(c,p,len);
addReply(c,shared.crlf); addReply(c,shared.crlf);
...@@ -534,7 +560,7 @@ void addReplyBulkSds(redisClient *c, sds s) { ...@@ -534,7 +560,7 @@ void addReplyBulkSds(redisClient *c, sds s) {
} }
/* Add a C nul term string as bulk reply */ /* Add a C nul term string as bulk reply */
void addReplyBulkCString(redisClient *c, char *s) { void addReplyBulkCString(redisClient *c, const char *s) {
if (s == NULL) { if (s == NULL) {
addReply(c,shared.nullbulk); addReply(c,shared.nullbulk);
} else { } else {
...@@ -779,7 +805,7 @@ void freeClient(redisClient *c) { ...@@ -779,7 +805,7 @@ void freeClient(redisClient *c) {
* a context where calling freeClient() is not possible, because the client * a context where calling freeClient() is not possible, because the client
* should be valid for the continuation of the flow of the program. */ * should be valid for the continuation of the flow of the program. */
void freeClientAsync(redisClient *c) { void freeClientAsync(redisClient *c) {
if (c->flags & REDIS_CLOSE_ASAP) return; if (c->flags & REDIS_CLOSE_ASAP || c->flags & REDIS_LUA_CLIENT) return;
c->flags |= REDIS_CLOSE_ASAP; c->flags |= REDIS_CLOSE_ASAP;
listAddNodeTail(server.clients_to_close,c); listAddNodeTail(server.clients_to_close,c);
} }
...@@ -957,7 +983,7 @@ int processInlineBuffer(redisClient *c) { ...@@ -957,7 +983,7 @@ int processInlineBuffer(redisClient *c) {
/* Helper function. Trims query buffer to make the function that processes /* Helper function. Trims query buffer to make the function that processes
* multi bulk requests idempotent. */ * multi bulk requests idempotent. */
static void setProtocolError(redisClient *c, int pos) { static void setProtocolError(redisClient *c, int pos) {
if (server.verbosity >= REDIS_VERBOSE) { if (server.verbosity <= REDIS_VERBOSE) {
sds client = catClientInfoString(sdsempty(),c); sds client = catClientInfoString(sdsempty(),c);
redisLog(REDIS_VERBOSE, redisLog(REDIS_VERBOSE,
"Protocol error from client: %s", client); "Protocol error from client: %s", client);
...@@ -1501,6 +1527,16 @@ void rewriteClientCommandVector(redisClient *c, int argc, ...) { ...@@ -1501,6 +1527,16 @@ void rewriteClientCommandVector(redisClient *c, int argc, ...) {
va_end(ap); va_end(ap);
} }
/* Completely replace the client command vector with the provided one. */
void replaceClientCommandVector(redisClient *c, int argc, robj **argv) {
freeClientArgv(c);
zfree(c->argv);
c->argv = argv;
c->argc = argc;
c->cmd = lookupCommandOrOriginal(c->argv[0]->ptr);
redisAssertWithInfo(c,NULL,c->cmd != NULL);
}
/* Rewrite a single item in the command vector. /* Rewrite a single item in the command vector.
* The new val ref count is incremented, and the old decremented. */ * The new val ref count is incremented, and the old decremented. */
void rewriteClientCommandArgument(redisClient *c, int i, robj *newval) { void rewriteClientCommandArgument(redisClient *c, int i, robj *newval) {
...@@ -1676,7 +1712,9 @@ void pauseClients(mstime_t end) { ...@@ -1676,7 +1712,9 @@ void pauseClients(mstime_t end) {
/* Return non-zero if clients are currently paused. As a side effect the /* Return non-zero if clients are currently paused. As a side effect the
* function checks if the pause time was reached and clear it. */ * function checks if the pause time was reached and clear it. */
int clientsArePaused(void) { int clientsArePaused(void) {
if (server.clients_paused && server.clients_pause_end_time < server.mstime) { if (server.clients_paused &&
server.clients_pause_end_time < server.mstime)
{
listNode *ln; listNode *ln;
listIter li; listIter li;
redisClient *c; redisClient *c;
...@@ -1689,7 +1727,10 @@ int clientsArePaused(void) { ...@@ -1689,7 +1727,10 @@ int clientsArePaused(void) {
while ((ln = listNext(&li)) != NULL) { while ((ln = listNext(&li)) != NULL) {
c = listNodeValue(ln); c = listNodeValue(ln);
if (c->flags & REDIS_SLAVE) continue; /* Don't touch slaves and blocked clients. The latter pending
* requests be processed when unblocked. */
if (c->flags & (REDIS_SLAVE|REDIS_BLOCKED)) continue;
c->flags |= REDIS_UNBLOCKED;
listAddNodeTail(server.unblocked_clients,c); listAddNodeTail(server.unblocked_clients,c);
} }
} }
......
...@@ -50,14 +50,14 @@ robj *createObject(int type, void *ptr) { ...@@ -50,14 +50,14 @@ robj *createObject(int type, void *ptr) {
/* Create a string object with encoding REDIS_ENCODING_RAW, that is a plain /* Create a string object with encoding REDIS_ENCODING_RAW, that is a plain
* string object where o->ptr points to a proper sds string. */ * string object where o->ptr points to a proper sds string. */
robj *createRawStringObject(char *ptr, size_t len) { robj *createRawStringObject(const char *ptr, size_t len) {
return createObject(REDIS_STRING,sdsnewlen(ptr,len)); return createObject(REDIS_STRING,sdsnewlen(ptr,len));
} }
/* Create a string object with encoding REDIS_ENCODING_EMBSTR, that is /* Create a string object with encoding REDIS_ENCODING_EMBSTR, that is
* an object where the sds string is actually an unmodifiable string * an object where the sds string is actually an unmodifiable string
* allocated in the same chunk as the object itself. */ * allocated in the same chunk as the object itself. */
robj *createEmbeddedStringObject(char *ptr, size_t len) { robj *createEmbeddedStringObject(const char *ptr, size_t len) {
robj *o = zmalloc(sizeof(robj)+sizeof(struct sdshdr)+len+1); robj *o = zmalloc(sizeof(robj)+sizeof(struct sdshdr)+len+1);
struct sdshdr *sh = (void*)(o+1); struct sdshdr *sh = (void*)(o+1);
...@@ -85,7 +85,7 @@ robj *createEmbeddedStringObject(char *ptr, size_t len) { ...@@ -85,7 +85,7 @@ robj *createEmbeddedStringObject(char *ptr, size_t len) {
* The current limit of 39 is chosen so that the biggest string object * The current limit of 39 is chosen so that the biggest string object
* we allocate as EMBSTR will still fit into the 64 byte arena of jemalloc. */ * we allocate as EMBSTR will still fit into the 64 byte arena of jemalloc. */
#define REDIS_ENCODING_EMBSTR_SIZE_LIMIT 39 #define REDIS_ENCODING_EMBSTR_SIZE_LIMIT 39
robj *createStringObject(char *ptr, size_t len) { robj *createStringObject(const char *ptr, size_t len) {
if (len <= REDIS_ENCODING_EMBSTR_SIZE_LIMIT) if (len <= REDIS_ENCODING_EMBSTR_SIZE_LIMIT)
return createEmbeddedStringObject(ptr,len); return createEmbeddedStringObject(ptr,len);
else else
......
...@@ -869,9 +869,9 @@ int rdbSave(char *filename) { ...@@ -869,9 +869,9 @@ int rdbSave(char *filename) {
return REDIS_OK; return REDIS_OK;
werr: werr:
redisLog(REDIS_WARNING,"Write error saving DB on disk: %s", strerror(errno));
fclose(fp); fclose(fp);
unlink(tmpfile); unlink(tmpfile);
redisLog(REDIS_WARNING,"Write error saving DB on disk: %s", strerror(errno));
return REDIS_ERR; return REDIS_ERR;
} }
......
...@@ -644,9 +644,9 @@ static int cliSendCommand(int argc, char **argv, int repeat) { ...@@ -644,9 +644,9 @@ static int cliSendCommand(int argc, char **argv, int repeat) {
output_raw = 0; output_raw = 0;
if (!strcasecmp(command,"info") || if (!strcasecmp(command,"info") ||
(argc == 3 && !strcasecmp(command,"debug") && (argc >= 2 && !strcasecmp(command,"debug") &&
(!strcasecmp(argv[1],"jemalloc") && (!strcasecmp(argv[1],"jemalloc") ||
!strcasecmp(argv[2],"info"))) || !strcasecmp(argv[1],"htstats"))) ||
(argc == 2 && !strcasecmp(command,"cluster") && (argc == 2 && !strcasecmp(command,"cluster") &&
(!strcasecmp(argv[1],"nodes") || (!strcasecmp(argv[1],"nodes") ||
!strcasecmp(argv[1],"info"))) || !strcasecmp(argv[1],"info"))) ||
......
...@@ -53,7 +53,7 @@ ...@@ -53,7 +53,7 @@
#include <sys/resource.h> #include <sys/resource.h>
#include <sys/utsname.h> #include <sys/utsname.h>
#include <locale.h> #include <locale.h>
#include <sys/sysctl.h> #include <sys/socket.h>
/* Our shared "common" objects */ /* Our shared "common" objects */
...@@ -131,7 +131,7 @@ struct redisCommand redisCommandTable[] = { ...@@ -131,7 +131,7 @@ struct redisCommand redisCommandTable[] = {
{"append",appendCommand,3,"wm",0,NULL,1,1,1,0,0}, {"append",appendCommand,3,"wm",0,NULL,1,1,1,0,0},
{"strlen",strlenCommand,2,"rF",0,NULL,1,1,1,0,0}, {"strlen",strlenCommand,2,"rF",0,NULL,1,1,1,0,0},
{"del",delCommand,-2,"w",0,NULL,1,-1,1,0,0}, {"del",delCommand,-2,"w",0,NULL,1,-1,1,0,0},
{"exists",existsCommand,2,"rF",0,NULL,1,1,1,0,0}, {"exists",existsCommand,-2,"rF",0,NULL,1,-1,1,0,0},
{"setbit",setbitCommand,4,"wm",0,NULL,1,1,1,0,0}, {"setbit",setbitCommand,4,"wm",0,NULL,1,1,1,0,0},
{"getbit",getbitCommand,3,"rF",0,NULL,1,1,1,0,0}, {"getbit",getbitCommand,3,"rF",0,NULL,1,1,1,0,0},
{"setrange",setrangeCommand,4,"wm",0,NULL,1,1,1,0,0}, {"setrange",setrangeCommand,4,"wm",0,NULL,1,1,1,0,0},
...@@ -281,9 +281,15 @@ struct redisCommand redisCommandTable[] = { ...@@ -281,9 +281,15 @@ struct redisCommand redisCommandTable[] = {
{"bitpos",bitposCommand,-3,"r",0,NULL,1,1,1,0,0}, {"bitpos",bitposCommand,-3,"r",0,NULL,1,1,1,0,0},
{"wait",waitCommand,3,"rs",0,NULL,0,0,0,0,0}, {"wait",waitCommand,3,"rs",0,NULL,0,0,0,0,0},
{"command",commandCommand,0,"rlt",0,NULL,0,0,0,0,0}, {"command",commandCommand,0,"rlt",0,NULL,0,0,0,0,0},
{"geoadd",geoaddCommand,-5,"wm",0,NULL,1,1,1,0,0},
{"georadius",georadiusCommand,-6,"r",0,NULL,1,1,1,0,0},
{"georadiusbymember",georadiusByMemberCommand,-5,"r",0,NULL,1,1,1,0,0},
{"geohash",geohashCommand,-2,"r",0,NULL,1,1,1,0,0},
{"geopos",geoposCommand,-2,"r",0,NULL,1,1,1,0,0},
{"geodist",geodistCommand,-4,"r",0,NULL,1,1,1,0,0},
{"pfselftest",pfselftestCommand,1,"r",0,NULL,0,0,0,0,0}, {"pfselftest",pfselftestCommand,1,"r",0,NULL,0,0,0,0,0},
{"pfadd",pfaddCommand,-2,"wmF",0,NULL,1,1,1,0,0}, {"pfadd",pfaddCommand,-2,"wmF",0,NULL,1,1,1,0,0},
{"pfcount",pfcountCommand,-2,"r",0,NULL,1,1,1,0,0}, {"pfcount",pfcountCommand,-2,"r",0,NULL,1,-1,1,0,0},
{"pfmerge",pfmergeCommand,-2,"wm",0,NULL,1,-1,1,0,0}, {"pfmerge",pfmergeCommand,-2,"wm",0,NULL,1,-1,1,0,0},
{"pfdebug",pfdebugCommand,-3,"w",0,NULL,0,0,0,0,0}, {"pfdebug",pfdebugCommand,-3,"w",0,NULL,0,0,0,0,0},
{"latency",latencyCommand,-2,"arslt",0,NULL,0,0,0,0,0} {"latency",latencyCommand,-2,"arslt",0,NULL,0,0,0,0,0}
...@@ -905,9 +911,12 @@ long long getInstantaneousMetric(int metric) { ...@@ -905,9 +911,12 @@ long long getInstantaneousMetric(int metric) {
return sum / REDIS_METRIC_SAMPLES; return sum / REDIS_METRIC_SAMPLES;
} }
/* Check for timeouts. Returns non-zero if the client was terminated */ /* Check for timeouts. Returns non-zero if the client was terminated.
int clientsCronHandleTimeout(redisClient *c) { * The function gets the current time in milliseconds as argument since
time_t now = server.unixtime; * it gets called multiple times in a loop, so calling gettimeofday() for
* each iteration would be costly without any actual gain. */
int clientsCronHandleTimeout(redisClient *c, mstime_t now_ms) {
time_t now = now_ms/1000;
if (server.maxidletime && if (server.maxidletime &&
!(c->flags & REDIS_SLAVE) && /* no timeout for slaves */ !(c->flags & REDIS_SLAVE) && /* no timeout for slaves */
...@@ -923,11 +932,16 @@ int clientsCronHandleTimeout(redisClient *c) { ...@@ -923,11 +932,16 @@ int clientsCronHandleTimeout(redisClient *c) {
/* Blocked OPS timeout is handled with milliseconds resolution. /* Blocked OPS timeout is handled with milliseconds resolution.
* However note that the actual resolution is limited by * However note that the actual resolution is limited by
* server.hz. */ * server.hz. */
mstime_t now_ms = mstime();
if (c->bpop.timeout != 0 && c->bpop.timeout < now_ms) { if (c->bpop.timeout != 0 && c->bpop.timeout < now_ms) {
/* Handle blocking operation specific timeout. */
replyToBlockedClientTimedOut(c); replyToBlockedClientTimedOut(c);
unblockClient(c); unblockClient(c);
} else if (server.cluster_enabled) {
/* Cluster: handle unblock & redirect of clients blocked
* into keys no longer served by this server. */
if (clusterRedirectBlockedClientIfNeeded(c))
unblockClient(c);
} }
} }
return 0; return 0;
...@@ -959,17 +973,23 @@ int clientsCronResizeQueryBuffer(redisClient *c) { ...@@ -959,17 +973,23 @@ int clientsCronResizeQueryBuffer(redisClient *c) {
return 0; return 0;
} }
#define CLIENTS_CRON_MIN_ITERATIONS 5
void clientsCron(void) { void clientsCron(void) {
/* Make sure to process at least 1/(server.hz*10) of clients per call. /* Make sure to process at least numclients/server.hz of clients
* Since this function is called server.hz times per second we are sure that * per call. Since this function is called server.hz times per second
* in the worst case we process all the clients in 10 seconds. * we are sure that in the worst case we process all the clients in 1
* In normal conditions (a reasonable number of clients) we process * second. */
* all the clients in a shorter time. */
int numclients = listLength(server.clients); int numclients = listLength(server.clients);
int iterations = numclients/(server.hz*10); int iterations = numclients/server.hz;
mstime_t now = mstime();
/* Process at least a few clients while we are at it, even if we need
* to process less than CLIENTS_CRON_MIN_ITERATIONS to meet our contract
* of processing each client once per second. */
if (iterations < CLIENTS_CRON_MIN_ITERATIONS)
iterations = (numclients < CLIENTS_CRON_MIN_ITERATIONS) ?
numclients : CLIENTS_CRON_MIN_ITERATIONS;
if (iterations < 50)
iterations = (numclients < 50) ? numclients : 50;
while(listLength(server.clients) && iterations--) { while(listLength(server.clients) && iterations--) {
redisClient *c; redisClient *c;
listNode *head; listNode *head;
...@@ -983,7 +1003,7 @@ void clientsCron(void) { ...@@ -983,7 +1003,7 @@ void clientsCron(void) {
/* The following functions do different service checks on the client. /* The following functions do different service checks on the client.
* The protocol is that they return non-zero if the client was * The protocol is that they return non-zero if the client was
* terminated. */ * terminated. */
if (clientsCronHandleTimeout(c)) continue; if (clientsCronHandleTimeout(c,now)) continue;
if (clientsCronResizeQueryBuffer(c)) continue; if (clientsCronResizeQueryBuffer(c)) continue;
} }
} }
...@@ -1260,6 +1280,12 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { ...@@ -1260,6 +1280,12 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
void beforeSleep(struct aeEventLoop *eventLoop) { void beforeSleep(struct aeEventLoop *eventLoop) {
REDIS_NOTUSED(eventLoop); REDIS_NOTUSED(eventLoop);
/* Call the Redis Cluster before sleep function. Note that this function
* may change the state of Redis Cluster (from ok to fail or vice versa),
* so it's a good idea to call it before serving the unblocked clients
* later in this function. */
if (server.cluster_enabled) clusterBeforeSleep();
/* Run a fast expire cycle (the called function will return /* Run a fast expire cycle (the called function will return
* ASAP if a fast cycle is not needed). */ * ASAP if a fast cycle is not needed). */
if (server.active_expire_enabled && server.masterhost == NULL) if (server.active_expire_enabled && server.masterhost == NULL)
...@@ -1291,9 +1317,6 @@ void beforeSleep(struct aeEventLoop *eventLoop) { ...@@ -1291,9 +1317,6 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
/* Write the AOF buffer on disk */ /* Write the AOF buffer on disk */
flushAppendOnlyFile(0); flushAppendOnlyFile(0);
/* Call the Redis Cluster before sleep function. */
if (server.cluster_enabled) clusterBeforeSleep();
} }
/* =========================== Server initialization ======================== */ /* =========================== Server initialization ======================== */
...@@ -1740,6 +1763,7 @@ void resetServerStats(void) { ...@@ -1740,6 +1763,7 @@ void resetServerStats(void) {
} }
server.stat_net_input_bytes = 0; server.stat_net_input_bytes = 0;
server.stat_net_output_bytes = 0; server.stat_net_output_bytes = 0;
server.aof_delayed_fsync = 0;
} }
void initServer(void) { void initServer(void) {
...@@ -1784,7 +1808,7 @@ void initServer(void) { ...@@ -1784,7 +1808,7 @@ void initServer(void) {
server.sofd = anetUnixServer(server.neterr,server.unixsocket, server.sofd = anetUnixServer(server.neterr,server.unixsocket,
server.unixsocketperm, server.tcp_backlog); server.unixsocketperm, server.tcp_backlog);
if (server.sofd == ANET_ERR) { if (server.sofd == ANET_ERR) {
redisLog(REDIS_WARNING, "Opening socket: %s", server.neterr); redisLog(REDIS_WARNING, "Opening Unix socket: %s", server.neterr);
exit(1); exit(1);
} }
anetNonBlock(NULL,server.sofd); anetNonBlock(NULL,server.sofd);
...@@ -2199,36 +2223,22 @@ int processCommand(redisClient *c) { ...@@ -2199,36 +2223,22 @@ int processCommand(redisClient *c) {
* 2) The command has no key arguments. */ * 2) The command has no key arguments. */
if (server.cluster_enabled && if (server.cluster_enabled &&
!(c->flags & REDIS_MASTER) && !(c->flags & REDIS_MASTER) &&
!(c->flags & REDIS_LUA_CLIENT &&
server.lua_caller->flags & REDIS_MASTER) &&
!(c->cmd->getkeys_proc == NULL && c->cmd->firstkey == 0)) !(c->cmd->getkeys_proc == NULL && c->cmd->firstkey == 0))
{ {
int hashslot; int hashslot;
if (server.cluster->state != REDIS_CLUSTER_OK) { if (server.cluster->state != REDIS_CLUSTER_OK) {
flagTransaction(c); flagTransaction(c);
addReplySds(c,sdsnew("-CLUSTERDOWN The cluster is down. Use CLUSTER INFO for more information\r\n")); clusterRedirectClient(c,NULL,0,REDIS_CLUSTER_REDIR_DOWN_STATE);
return REDIS_OK; return REDIS_OK;
} else { } else {
int error_code; int error_code;
clusterNode *n = getNodeByQuery(c,c->cmd,c->argv,c->argc,&hashslot,&error_code); clusterNode *n = getNodeByQuery(c,c->cmd,c->argv,c->argc,&hashslot,&error_code);
if (n == NULL) { if (n == NULL || n != server.cluster->myself) {
flagTransaction(c);
if (error_code == REDIS_CLUSTER_REDIR_CROSS_SLOT) {
addReplySds(c,sdsnew("-CROSSSLOT Keys in request don't hash to the same slot\r\n"));
} else if (error_code == REDIS_CLUSTER_REDIR_UNSTABLE) {
/* The request spawns mutliple keys in the same slot,
* but the slot is not "stable" currently as there is
* a migration or import in progress. */
addReplySds(c,sdsnew("-TRYAGAIN Multiple keys request during rehashing of slot\r\n"));
} else {
redisPanic("getNodeByQuery() unknown error.");
}
return REDIS_OK;
} else if (n != server.cluster->myself) {
flagTransaction(c); flagTransaction(c);
addReplySds(c,sdscatprintf(sdsempty(), clusterRedirectClient(c,n,hashslot,error_code);
"-%s %d %s:%d\r\n",
(error_code == REDIS_CLUSTER_REDIR_ASK) ? "ASK" : "MOVED",
hashslot,n->ip,n->port));
return REDIS_OK; return REDIS_OK;
} }
} }
...@@ -2745,7 +2755,7 @@ sds genRedisInfoString(char *section) { ...@@ -2745,7 +2755,7 @@ sds genRedisInfoString(char *section) {
char maxmemory_hmem[64]; char maxmemory_hmem[64];
size_t zmalloc_used = zmalloc_used_memory(); size_t zmalloc_used = zmalloc_used_memory();
size_t total_system_mem = server.system_memory_size; size_t total_system_mem = server.system_memory_size;
char *evict_policy = maxmemoryToString(); const char *evict_policy = maxmemoryToString();
long long memory_lua = (long long)lua_gc(server.lua,LUA_GCCOUNT,0)*1024; long long memory_lua = (long long)lua_gc(server.lua,LUA_GCCOUNT,0)*1024;
/* Peak memory is updated from time to time by serverCron() so it /* Peak memory is updated from time to time by serverCron() so it
......
...@@ -1053,14 +1053,14 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask); ...@@ -1053,14 +1053,14 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask);
void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask); void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask);
void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask); void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask);
void addReplyBulk(redisClient *c, robj *obj); void addReplyBulk(redisClient *c, robj *obj);
void addReplyBulkCString(redisClient *c, char *s); void addReplyBulkCString(redisClient *c, const char *s);
void addReplyBulkCBuffer(redisClient *c, void *p, size_t len); void addReplyBulkCBuffer(redisClient *c, const void *p, size_t len);
void addReplyBulkLongLong(redisClient *c, long long ll); void addReplyBulkLongLong(redisClient *c, long long ll);
void addReply(redisClient *c, robj *obj); void addReply(redisClient *c, robj *obj);
void addReplySds(redisClient *c, sds s); void addReplySds(redisClient *c, sds s);
void addReplyBulkSds(redisClient *c, sds s); void addReplyBulkSds(redisClient *c, sds s);
void addReplyError(redisClient *c, char *err); void addReplyError(redisClient *c, const char *err);
void addReplyStatus(redisClient *c, char *status); void addReplyStatus(redisClient *c, const char *status);
void addReplyDouble(redisClient *c, double d); void addReplyDouble(redisClient *c, double d);
void addReplyLongLong(redisClient *c, long long ll); void addReplyLongLong(redisClient *c, long long ll);
void addReplyMultiBulkLen(redisClient *c, long length); void addReplyMultiBulkLen(redisClient *c, long length);
...@@ -1074,6 +1074,7 @@ sds catClientInfoString(sds s, redisClient *client); ...@@ -1074,6 +1074,7 @@ sds catClientInfoString(sds s, redisClient *client);
sds getAllClientsInfoString(void); sds getAllClientsInfoString(void);
void rewriteClientCommandVector(redisClient *c, int argc, ...); void rewriteClientCommandVector(redisClient *c, int argc, ...);
void rewriteClientCommandArgument(redisClient *c, int i, robj *newval); void rewriteClientCommandArgument(redisClient *c, int i, robj *newval);
void replaceClientCommandVector(redisClient *c, int argc, robj **argv);
unsigned long getClientOutputBufferMemoryUsage(redisClient *c); unsigned long getClientOutputBufferMemoryUsage(redisClient *c);
void freeClientsInAsyncFreeQueue(void); void freeClientsInAsyncFreeQueue(void);
void asyncCloseClientOnOutputBufferLimitReached(redisClient *c); void asyncCloseClientOnOutputBufferLimitReached(redisClient *c);
...@@ -1136,9 +1137,9 @@ void freeSetObject(robj *o); ...@@ -1136,9 +1137,9 @@ void freeSetObject(robj *o);
void freeZsetObject(robj *o); void freeZsetObject(robj *o);
void freeHashObject(robj *o); void freeHashObject(robj *o);
robj *createObject(int type, void *ptr); robj *createObject(int type, void *ptr);
robj *createStringObject(char *ptr, size_t len); robj *createStringObject(const char *ptr, size_t len);
robj *createRawStringObject(char *ptr, size_t len); robj *createRawStringObject(const char *ptr, size_t len);
robj *createEmbeddedStringObject(char *ptr, size_t len); robj *createEmbeddedStringObject(const char *ptr, size_t len);
robj *dupStringObject(robj *o); robj *dupStringObject(robj *o);
int isObjectRepresentableAsLongLong(robj *o, long long *llongval); int isObjectRepresentableAsLongLong(robj *o, long long *llongval);
robj *tryObjectEncoding(robj *o); robj *tryObjectEncoding(robj *o);
...@@ -1241,6 +1242,7 @@ void zzlNext(unsigned char *zl, unsigned char **eptr, unsigned char **sptr); ...@@ -1241,6 +1242,7 @@ void zzlNext(unsigned char *zl, unsigned char **eptr, unsigned char **sptr);
void zzlPrev(unsigned char *zl, unsigned char **eptr, unsigned char **sptr); void zzlPrev(unsigned char *zl, unsigned char **eptr, unsigned char **sptr);
unsigned int zsetLength(robj *zobj); unsigned int zsetLength(robj *zobj);
void zsetConvert(robj *zobj, int encoding); void zsetConvert(robj *zobj, int encoding);
int zsetScore(robj *zobj, robj *member, double *score);
unsigned long zslGetRank(zskiplist *zsl, double score, robj *o); unsigned long zslGetRank(zskiplist *zsl, double score, robj *o);
/* Core functions */ /* Core functions */
...@@ -1275,7 +1277,7 @@ void closeListeningSockets(int unlink_unix_socket); ...@@ -1275,7 +1277,7 @@ void closeListeningSockets(int unlink_unix_socket);
void updateCachedTime(void); void updateCachedTime(void);
void resetServerStats(void); void resetServerStats(void);
unsigned int getLRUClock(void); unsigned int getLRUClock(void);
char *maxmemoryToString(void); const char *maxmemoryToString(void);
/* Set data type */ /* Set data type */
robj *setTypeCreate(robj *value); robj *setTypeCreate(robj *value);
...@@ -1328,7 +1330,7 @@ void loadServerConfig(char *filename, char *options); ...@@ -1328,7 +1330,7 @@ void loadServerConfig(char *filename, char *options);
void appendServerSaveParams(time_t seconds, int changes); void appendServerSaveParams(time_t seconds, int changes);
void resetServerSaveParams(void); void resetServerSaveParams(void);
struct rewriteConfigState; /* Forward declaration to export API. */ struct rewriteConfigState; /* Forward declaration to export API. */
void rewriteConfigRewriteLine(struct rewriteConfigState *state, char *option, sds line, int force); void rewriteConfigRewriteLine(struct rewriteConfigState *state, const char *option, sds line, int force);
int rewriteConfig(char *path); int rewriteConfig(char *path);
/* db.c -- Keyspace access API */ /* db.c -- Keyspace access API */
...@@ -1396,6 +1398,7 @@ void blockClient(redisClient *c, int btype); ...@@ -1396,6 +1398,7 @@ void blockClient(redisClient *c, int btype);
void unblockClient(redisClient *c); void unblockClient(redisClient *c);
void replyToBlockedClientTimedOut(redisClient *c); void replyToBlockedClientTimedOut(redisClient *c);
int getTimeoutFromObjectOrReply(redisClient *c, robj *object, mstime_t *timeout, int unit); int getTimeoutFromObjectOrReply(redisClient *c, robj *object, mstime_t *timeout, int unit);
void disconnectAllBlockedClients(void);
/* Git SHA1 */ /* Git SHA1 */
char *redisGitSHA1(void); char *redisGitSHA1(void);
...@@ -1556,6 +1559,14 @@ void bitcountCommand(redisClient *c); ...@@ -1556,6 +1559,14 @@ void bitcountCommand(redisClient *c);
void bitposCommand(redisClient *c); void bitposCommand(redisClient *c);
void replconfCommand(redisClient *c); void replconfCommand(redisClient *c);
void waitCommand(redisClient *c); void waitCommand(redisClient *c);
void geoencodeCommand(redisClient *c);
void geodecodeCommand(redisClient *c);
void georadiusByMemberCommand(redisClient *c);
void georadiusCommand(redisClient *c);
void geoaddCommand(redisClient *c);
void geohashCommand(redisClient *c);
void geoposCommand(redisClient *c);
void geodistCommand(redisClient *c);
void pfselftestCommand(redisClient *c); void pfselftestCommand(redisClient *c);
void pfaddCommand(redisClient *c); void pfaddCommand(redisClient *c);
void pfcountCommand(redisClient *c); void pfcountCommand(redisClient *c);
......
...@@ -652,7 +652,8 @@ void replconfCommand(redisClient *c) { ...@@ -652,7 +652,8 @@ void replconfCommand(redisClient *c) {
* *
* It does a few things: * It does a few things:
* *
* 1) Put the slave in ONLINE state. * 1) Put the slave in ONLINE state (useless when the function is called
* because state is already ONLINE but repl_put_online_on_ack is true).
* 2) Make sure the writable event is re-installed, since calling the SYNC * 2) Make sure the writable event is re-installed, since calling the SYNC
* command disables it, so that we can accumulate output buffer without * command disables it, so that we can accumulate output buffer without
* sending it to the slave. * sending it to the slave.
...@@ -660,7 +661,7 @@ void replconfCommand(redisClient *c) { ...@@ -660,7 +661,7 @@ void replconfCommand(redisClient *c) {
void putSlaveOnline(redisClient *slave) { void putSlaveOnline(redisClient *slave) {
slave->replstate = REDIS_REPL_ONLINE; slave->replstate = REDIS_REPL_ONLINE;
slave->repl_put_online_on_ack = 0; slave->repl_put_online_on_ack = 0;
slave->repl_ack_time = server.unixtime; slave->repl_ack_time = server.unixtime; /* Prevent false timeout. */
if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE,
sendReplyToClient, slave) == AE_ERR) { sendReplyToClient, slave) == AE_ERR) {
redisLog(REDIS_WARNING,"Unable to register writable event for slave bulk transfer: %s", strerror(errno)); redisLog(REDIS_WARNING,"Unable to register writable event for slave bulk transfer: %s", strerror(errno));
...@@ -773,7 +774,7 @@ void updateSlavesWaitingBgsave(int bgsaveerr, int type) { ...@@ -773,7 +774,7 @@ void updateSlavesWaitingBgsave(int bgsaveerr, int type) {
* is technically online now. */ * is technically online now. */
slave->replstate = REDIS_REPL_ONLINE; slave->replstate = REDIS_REPL_ONLINE;
slave->repl_put_online_on_ack = 1; slave->repl_put_online_on_ack = 1;
slave->repl_ack_time = server.unixtime; slave->repl_ack_time = server.unixtime; /* Timeout otherwise. */
} else { } else {
if (bgsaveerr != REDIS_OK) { if (bgsaveerr != REDIS_OK) {
freeClient(slave); freeClient(slave);
...@@ -1443,7 +1444,7 @@ error: ...@@ -1443,7 +1444,7 @@ error:
int connectWithMaster(void) { int connectWithMaster(void) {
int fd; int fd;
fd = anetTcpNonBlockBindConnect(NULL, fd = anetTcpNonBlockBestEffortBindConnect(NULL,
server.masterhost,server.masterport,REDIS_BIND_ADDR); server.masterhost,server.masterport,REDIS_BIND_ADDR);
if (fd == -1) { if (fd == -1) {
redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s", redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s",
...@@ -1505,6 +1506,7 @@ void replicationSetMaster(char *ip, int port) { ...@@ -1505,6 +1506,7 @@ void replicationSetMaster(char *ip, int port) {
server.masterhost = sdsnew(ip); server.masterhost = sdsnew(ip);
server.masterport = port; server.masterport = port;
if (server.master) freeClient(server.master); if (server.master) freeClient(server.master);
disconnectAllBlockedClients(); /* Clients blocked in master, now slave. */
disconnectSlaves(); /* Force our slaves to resync with us as well. */ disconnectSlaves(); /* Force our slaves to resync with us as well. */
replicationDiscardCachedMaster(); /* Don't try a PSYNC. */ replicationDiscardCachedMaster(); /* Don't try a PSYNC. */
freeReplicationBacklog(); /* Don't allow our chained slaves to PSYNC. */ freeReplicationBacklog(); /* Don't allow our chained slaves to PSYNC. */
......
...@@ -357,8 +357,9 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) { ...@@ -357,8 +357,9 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) {
if (cmd->flags & REDIS_CMD_WRITE) server.lua_write_dirty = 1; if (cmd->flags & REDIS_CMD_WRITE) server.lua_write_dirty = 1;
/* If this is a Redis Cluster node, we need to make sure Lua is not /* If this is a Redis Cluster node, we need to make sure Lua is not
* trying to access non-local keys. */ * trying to access non-local keys, with the exception of commands
if (server.cluster_enabled) { * received from our master. */
if (server.cluster_enabled && !(server.lua_caller->flags & REDIS_MASTER)) {
/* Duplicate relevant flags in the lua client. */ /* Duplicate relevant flags in the lua client. */
c->flags &= ~(REDIS_READONLY|REDIS_ASKING); c->flags &= ~(REDIS_READONLY|REDIS_ASKING);
c->flags |= server.lua_caller->flags & (REDIS_READONLY|REDIS_ASKING); c->flags |= server.lua_caller->flags & (REDIS_READONLY|REDIS_ASKING);
...@@ -611,11 +612,12 @@ void scriptingEnableGlobalsProtection(lua_State *lua) { ...@@ -611,11 +612,12 @@ void scriptingEnableGlobalsProtection(lua_State *lua) {
/* strict.lua from: http://metalua.luaforge.net/src/lib/strict.lua.html. /* strict.lua from: http://metalua.luaforge.net/src/lib/strict.lua.html.
* Modified to be adapted to Redis. */ * Modified to be adapted to Redis. */
s[j++]="local dbg=debug\n";
s[j++]="local mt = {}\n"; s[j++]="local mt = {}\n";
s[j++]="setmetatable(_G, mt)\n"; s[j++]="setmetatable(_G, mt)\n";
s[j++]="mt.__newindex = function (t, n, v)\n"; s[j++]="mt.__newindex = function (t, n, v)\n";
s[j++]=" if debug.getinfo(2) then\n"; s[j++]=" if dbg.getinfo(2) then\n";
s[j++]=" local w = debug.getinfo(2, \"S\").what\n"; s[j++]=" local w = dbg.getinfo(2, \"S\").what\n";
s[j++]=" if w ~= \"main\" and w ~= \"C\" then\n"; s[j++]=" if w ~= \"main\" and w ~= \"C\" then\n";
s[j++]=" error(\"Script attempted to create global variable '\"..tostring(n)..\"'\", 2)\n"; s[j++]=" error(\"Script attempted to create global variable '\"..tostring(n)..\"'\", 2)\n";
s[j++]=" end\n"; s[j++]=" end\n";
...@@ -623,11 +625,12 @@ void scriptingEnableGlobalsProtection(lua_State *lua) { ...@@ -623,11 +625,12 @@ void scriptingEnableGlobalsProtection(lua_State *lua) {
s[j++]=" rawset(t, n, v)\n"; s[j++]=" rawset(t, n, v)\n";
s[j++]="end\n"; s[j++]="end\n";
s[j++]="mt.__index = function (t, n)\n"; s[j++]="mt.__index = function (t, n)\n";
s[j++]=" if debug.getinfo(2) and debug.getinfo(2, \"S\").what ~= \"C\" then\n"; s[j++]=" if dbg.getinfo(2) and dbg.getinfo(2, \"S\").what ~= \"C\" then\n";
s[j++]=" error(\"Script attempted to access unexisting global variable '\"..tostring(n)..\"'\", 2)\n"; s[j++]=" error(\"Script attempted to access unexisting global variable '\"..tostring(n)..\"'\", 2)\n";
s[j++]=" end\n"; s[j++]=" end\n";
s[j++]=" return rawget(t, n)\n"; s[j++]=" return rawget(t, n)\n";
s[j++]="end\n"; s[j++]="end\n";
s[j++]="debug = nil\n";
s[j++]=NULL; s[j++]=NULL;
for (j = 0; s[j] != NULL; j++) code = sdscatlen(code,s[j],strlen(s[j])); for (j = 0; s[j] != NULL; j++) code = sdscatlen(code,s[j],strlen(s[j]));
...@@ -731,10 +734,11 @@ void scriptingInit(void) { ...@@ -731,10 +734,11 @@ void scriptingInit(void) {
* information about the caller, that's what makes sense from the point * information about the caller, that's what makes sense from the point
* of view of the user debugging a script. */ * of view of the user debugging a script. */
{ {
char *errh_func = "function __redis__err__handler(err)\n" char *errh_func = "local dbg = debug\n"
" local i = debug.getinfo(2,'nSl')\n" "function __redis__err__handler(err)\n"
" local i = dbg.getinfo(2,'nSl')\n"
" if i and i.what == 'C' then\n" " if i and i.what == 'C' then\n"
" i = debug.getinfo(3,'nSl')\n" " i = dbg.getinfo(3,'nSl')\n"
" end\n" " end\n"
" if i then\n" " if i then\n"
" return i.source .. ':' .. i.currentline .. ': ' .. err\n" " return i.source .. ':' .. i.currentline .. ': ' .. err\n"
...@@ -876,7 +880,7 @@ void luaSetGlobalArray(lua_State *lua, char *var, robj **elev, int elec) { ...@@ -876,7 +880,7 @@ void luaSetGlobalArray(lua_State *lua, char *var, robj **elev, int elec) {
} }
/* Define a lua function with the specified function name and body. /* Define a lua function with the specified function name and body.
* The function name musts be a 2 characters long string, since all the * The function name musts be a 42 characters long string, since all the
* functions we defined in the Lua context are in the form: * functions we defined in the Lua context are in the form:
* *
* f_<hex sha1 sum> * f_<hex sha1 sum>
......
...@@ -71,7 +71,7 @@ sds sdsempty(void) { ...@@ -71,7 +71,7 @@ sds sdsempty(void) {
return sdsnewlen("",0); return sdsnewlen("",0);
} }
/* Create a new sds string starting from a null termined C string. */ /* Create a new sds string starting from a null terminated C string. */
sds sdsnew(const char *init) { sds sdsnew(const char *init) {
size_t initlen = (init == NULL) ? 0 : strlen(init); size_t initlen = (init == NULL) ? 0 : strlen(init);
return sdsnewlen(init, initlen); return sdsnewlen(init, initlen);
...@@ -557,7 +557,7 @@ sds sdscatfmt(sds s, char const *fmt, ...) { ...@@ -557,7 +557,7 @@ sds sdscatfmt(sds s, char const *fmt, ...) {
* Example: * Example:
* *
* s = sdsnew("AA...AA.a.aa.aHelloWorld :::"); * s = sdsnew("AA...AA.a.aa.aHelloWorld :::");
* s = sdstrim(s,"A. :"); * s = sdstrim(s,"Aa. :");
* printf("%s\n", s); * printf("%s\n", s);
* *
* Output will be just "Hello World". * Output will be just "Hello World".
...@@ -1098,6 +1098,7 @@ int sdsTest(int argc, char *argv[]) { ...@@ -1098,6 +1098,7 @@ int sdsTest(int argc, char *argv[]) {
unsigned int oldfree; unsigned int oldfree;
sdsfree(x); sdsfree(x);
sdsfree(y);
x = sdsnew("0"); x = sdsnew("0");
sh = (void*) (x-(sizeof(struct sdshdr))); sh = (void*) (x-(sizeof(struct sdshdr)));
test_cond("sdsnew() free/len buffers", sh->len == 1 && sh->free == 0); test_cond("sdsnew() free/len buffers", sh->len == 1 && sh->free == 0);
...@@ -1110,6 +1111,8 @@ int sdsTest(int argc, char *argv[]) { ...@@ -1110,6 +1111,8 @@ int sdsTest(int argc, char *argv[]) {
test_cond("sdsIncrLen() -- content", x[0] == '0' && x[1] == '1'); test_cond("sdsIncrLen() -- content", x[0] == '0' && x[1] == '1');
test_cond("sdsIncrLen() -- len", sh->len == 2); test_cond("sdsIncrLen() -- len", sh->len == 2);
test_cond("sdsIncrLen() -- free", sh->free == oldfree-1); test_cond("sdsIncrLen() -- free", sh->free == oldfree-1);
sdsfree(x);
} }
} }
test_report() test_report()
......
This diff is collapsed.
...@@ -23,7 +23,7 @@ A million repetitions of "a" ...@@ -23,7 +23,7 @@ A million repetitions of "a"
#include <stdio.h> #include <stdio.h>
#include <string.h> #include <string.h>
#include <sys/types.h> /* for u_int*_t */ #include <stdint.h>
#include "solarisfixes.h" #include "solarisfixes.h"
#include "sha1.h" #include "sha1.h"
#include "config.h" #include "config.h"
...@@ -53,12 +53,12 @@ A million repetitions of "a" ...@@ -53,12 +53,12 @@ A million repetitions of "a"
/* Hash a single 512-bit block. This is the core of the algorithm. */ /* Hash a single 512-bit block. This is the core of the algorithm. */
void SHA1Transform(u_int32_t state[5], const unsigned char buffer[64]) void SHA1Transform(uint32_t state[5], const unsigned char buffer[64])
{ {
u_int32_t a, b, c, d, e; uint32_t a, b, c, d, e;
typedef union { typedef union {
unsigned char c[64]; unsigned char c[64];
u_int32_t l[16]; uint32_t l[16];
} CHAR64LONG16; } CHAR64LONG16;
#ifdef SHA1HANDSOFF #ifdef SHA1HANDSOFF
CHAR64LONG16 block[1]; /* use array to appear as a pointer */ CHAR64LONG16 block[1]; /* use array to appear as a pointer */
...@@ -128,9 +128,9 @@ void SHA1Init(SHA1_CTX* context) ...@@ -128,9 +128,9 @@ void SHA1Init(SHA1_CTX* context)
/* Run your data through this. */ /* Run your data through this. */
void SHA1Update(SHA1_CTX* context, const unsigned char* data, u_int32_t len) void SHA1Update(SHA1_CTX* context, const unsigned char* data, uint32_t len)
{ {
u_int32_t i, j; uint32_t i, j;
j = context->count[0]; j = context->count[0];
if ((context->count[0] += len << 3) < j) if ((context->count[0] += len << 3) < j)
...@@ -168,7 +168,7 @@ void SHA1Final(unsigned char digest[20], SHA1_CTX* context) ...@@ -168,7 +168,7 @@ void SHA1Final(unsigned char digest[20], SHA1_CTX* context)
for (i = 0; i < 2; i++) for (i = 0; i < 2; i++)
{ {
u_int32_t t = context->count[i]; uint32_t t = context->count[i];
int j; int j;
for (j = 0; j < 4; t >>= 8, j++) for (j = 0; j < 4; t >>= 8, j++)
......
...@@ -8,14 +8,14 @@ By Steve Reid <steve@edmweb.com> ...@@ -8,14 +8,14 @@ By Steve Reid <steve@edmweb.com>
*/ */
typedef struct { typedef struct {
u_int32_t state[5]; uint32_t state[5];
u_int32_t count[2]; uint32_t count[2];
unsigned char buffer[64]; unsigned char buffer[64];
} SHA1_CTX; } SHA1_CTX;
void SHA1Transform(u_int32_t state[5], const unsigned char buffer[64]); void SHA1Transform(uint32_t state[5], const unsigned char buffer[64]);
void SHA1Init(SHA1_CTX* context); void SHA1Init(SHA1_CTX* context);
void SHA1Update(SHA1_CTX* context, const unsigned char* data, u_int32_t len); void SHA1Update(SHA1_CTX* context, const unsigned char* data, uint32_t len);
void SHA1Final(unsigned char digest[20], SHA1_CTX* context); void SHA1Final(unsigned char digest[20], SHA1_CTX* context);
#ifdef REDIS_TEST #ifdef REDIS_TEST
......
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