Unverified Commit ec2d1807 authored by Oran Agra's avatar Oran Agra Committed by GitHub
Browse files

Merge 6.2 RC2

parents b8c67ce4 049f2f08
......@@ -646,13 +646,13 @@ dictEntry *dictGetRandomKey(dict *d)
do {
/* We are sure there are no elements in indexes from 0
* to rehashidx-1 */
h = d->rehashidx + (random() % (dictSlots(d) - d->rehashidx));
h = d->rehashidx + (randomULong() % (dictSlots(d) - d->rehashidx));
he = (h >= d->ht[0].size) ? d->ht[1].table[h - d->ht[0].size] :
d->ht[0].table[h];
} while(he == NULL);
} else {
do {
h = random() & d->ht[0].sizemask;
h = randomULong() & d->ht[0].sizemask;
he = d->ht[0].table[h];
} while(he == NULL);
}
......@@ -718,7 +718,7 @@ unsigned int dictGetSomeKeys(dict *d, dictEntry **des, unsigned int count) {
maxsizemask = d->ht[1].sizemask;
/* Pick a random point inside the larger table. */
unsigned long i = random() & maxsizemask;
unsigned long i = randomULong() & maxsizemask;
unsigned long emptylen = 0; /* Continuous empty entries so far. */
while(stored < count && maxsteps--) {
for (j = 0; j < tables; j++) {
......@@ -743,7 +743,7 @@ unsigned int dictGetSomeKeys(dict *d, dictEntry **des, unsigned int count) {
if (he == NULL) {
emptylen++;
if (emptylen >= 5 && emptylen > count) {
i = random() & maxsizemask;
i = randomULong() & maxsizemask;
emptylen = 0;
}
} else {
......@@ -1135,10 +1135,10 @@ size_t _dictGetStatsHt(char *buf, size_t bufsize, dictht *ht, int tableid) {
/* Generate human readable stats. */
l += snprintf(buf+l,bufsize-l,
"Hash table %d stats (%s):\n"
" table size: %ld\n"
" number of elements: %ld\n"
" different slots: %ld\n"
" max chain length: %ld\n"
" table size: %lu\n"
" number of elements: %lu\n"
" different slots: %lu\n"
" max chain length: %lu\n"
" avg chain length (counted): %.02f\n"
" avg chain length (computed): %.02f\n"
" Chain length distribution:\n",
......@@ -1215,7 +1215,7 @@ dictType BenchmarkDictType = {
#define end_benchmark(msg) do { \
elapsed = timeInMilliseconds()-start; \
printf(msg ": %ld items in %lld ms\n", count, elapsed); \
} while(0);
} while(0)
/* dict-benchmark [count] */
int main(int argc, char **argv) {
......@@ -1270,6 +1270,13 @@ int main(int argc, char **argv) {
}
end_benchmark("Random access of existing elements");
start_benchmark();
for (j = 0; j < count; j++) {
dictEntry *de = dictGetRandomKey(dict);
assert(de != NULL);
}
end_benchmark("Accessing random keys");
start_benchmark();
for (j = 0; j < count; j++) {
sds key = sdsfromlonglong(rand() % count);
......
......@@ -33,11 +33,14 @@
* POSSIBILITY OF SUCH DAMAGE.
*/
#include <stdint.h>
#ifndef __DICT_H
#define __DICT_H
#include "mt19937-64.h"
#include <limits.h>
#include <stdint.h>
#include <stdlib.h>
#define DICT_OK 0
#define DICT_ERR 1
......@@ -148,6 +151,13 @@ typedef void (dictScanBucketFunction)(void *privdata, dictEntry **bucketref);
#define dictSize(d) ((d)->ht[0].used+(d)->ht[1].used)
#define dictIsRehashing(d) ((d)->rehashidx != -1)
/* If our unsigned long type can store a 64 bit number, use a 64 bit PRNG. */
#if ULONG_MAX >= 0xffffffffffffffff
#define randomULong() ((unsigned long) genrand64_int64())
#else
#define randomULong() random()
#endif
/* API */
dict *dictCreate(dictType *type, void *privDataPtr);
int dictExpand(dict *d, unsigned long size);
......
......@@ -462,7 +462,7 @@ static int isSafeToPerformEvictions(void) {
/* When clients are paused the dataset should be static not just from the
* POV of clients not being able to write, but also from the POV of
* expires and evictions of keys not being performed. */
if (clientsArePaused()) return 0;
if (checkClientPauseTimeoutAndReturnIfPaused()) return 0;
return 1;
}
......
......@@ -148,7 +148,7 @@ void activeExpireCycle(int type) {
/* When clients are paused the dataset should be static not just from the
* POV of clients not being able to write, but also from the POV of
* expires and evictions of keys not being performed. */
if (clientsArePaused()) return;
if (checkClientPauseTimeoutAndReturnIfPaused()) return;
if (type == ACTIVE_EXPIRE_CYCLE_FAST) {
/* Don't start a fast cycle if the previous cycle did not exit
......
......@@ -100,8 +100,8 @@ int extractLongLatOrReply(client *c, robj **argv, double *xy) {
}
if (xy[0] < GEO_LONG_MIN || xy[0] > GEO_LONG_MAX ||
xy[1] < GEO_LAT_MIN || xy[1] > GEO_LAT_MAX) {
addReplySds(c, sdscatprintf(sdsempty(),
"-ERR invalid longitude,latitude pair %f,%f\r\n",xy[0],xy[1]));
addReplyErrorFormat(c,
"-ERR invalid longitude,latitude pair %f,%f\r\n",xy[0],xy[1]);
return C_ERR;
}
return C_OK;
......@@ -249,7 +249,7 @@ int geoAppendIfWithinShape(geoArray *ga, GeoShape *shape, double score, sds memb
* using multiple queries to the sorted set, that we later need to sort
* via qsort. Similarly we need to be able to reject points outside the search
* radius area ASAP in order to allocate and process more points than needed. */
int geoGetPointsInRange(robj *zobj, double min, double max, GeoShape *shape, geoArray *ga) {
int geoGetPointsInRange(robj *zobj, double min, double max, GeoShape *shape, geoArray *ga, unsigned long limit) {
/* minex 0 = include min in range; maxex 1 = exclude max in range */
/* That's: min <= val < max */
zrangespec range = { .min = min, .max = max, .minex = 0, .maxex = 1 };
......@@ -283,6 +283,7 @@ int geoGetPointsInRange(robj *zobj, double min, double max, GeoShape *shape, geo
sdsnewlen(vstr,vlen);
if (geoAppendIfWithinShape(ga,shape,score,member)
== C_ERR) sdsfree(member);
if (ga->used && limit && ga->used >= limit) break;
zzlNext(zl, &eptr, &sptr);
}
} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
......@@ -304,6 +305,7 @@ int geoGetPointsInRange(robj *zobj, double min, double max, GeoShape *shape, geo
ele = sdsdup(ele);
if (geoAppendIfWithinShape(ga,shape,ln->score,ele)
== C_ERR) sdsfree(ele);
if (ga->used && limit && ga->used >= limit) break;
ln = ln->level[0].forward;
}
}
......@@ -342,15 +344,15 @@ void scoresOfGeoHashBox(GeoHashBits hash, GeoHashFix52Bits *min, GeoHashFix52Bit
/* Obtain all members between the min/max of this geohash bounding box.
* Populate a geoArray of GeoPoints by calling geoGetPointsInRange().
* Return the number of points added to the array. */
int membersOfGeoHashBox(robj *zobj, GeoHashBits hash, geoArray *ga, GeoShape *shape) {
int membersOfGeoHashBox(robj *zobj, GeoHashBits hash, geoArray *ga, GeoShape *shape, unsigned long limit) {
GeoHashFix52Bits min, max;
scoresOfGeoHashBox(hash,&min,&max);
return geoGetPointsInRange(zobj, min, max, shape, ga);
return geoGetPointsInRange(zobj, min, max, shape, ga, limit);
}
/* Search all eight neighbors + self geohash box */
int membersOfAllNeighbors(robj *zobj, GeoHashRadius n, GeoShape *shape, geoArray *ga) {
int membersOfAllNeighbors(robj *zobj, GeoHashRadius n, GeoShape *shape, geoArray *ga, unsigned long limit) {
GeoHashBits neighbors[9];
unsigned int i, count = 0, last_processed = 0;
int debugmsg = 0;
......@@ -401,7 +403,8 @@ int membersOfAllNeighbors(robj *zobj, GeoHashRadius n, GeoShape *shape, geoArray
D("Skipping processing of %d, same as previous\n",i);
continue;
}
count += membersOfGeoHashBox(zobj, neighbors[i], ga, shape);
if (ga->used && limit && ga->used >= limit) break;
count += membersOfGeoHashBox(zobj, neighbors[i], ga, shape, limit);
last_processed = i;
}
return count;
......@@ -428,31 +431,45 @@ static int sort_gp_desc(const void *a, const void *b) {
* Commands
* ==================================================================== */
/* GEOADD key long lat name [long2 lat2 name2 ... longN latN nameN] */
/* GEOADD key [CH] [NX|XX] long lat name [long2 lat2 name2 ... longN latN nameN] */
void geoaddCommand(client *c) {
/* Check arguments number for sanity. */
if ((c->argc - 2) % 3 != 0) {
int xx = 0, nx = 0, longidx = 2;
int i;
/* Parse options. At the end 'longidx' is set to the argument position
* of the longitude of the first element. */
while (longidx < c->argc) {
char *opt = c->argv[longidx]->ptr;
if (!strcasecmp(opt,"nx")) nx = 1;
else if (!strcasecmp(opt,"xx")) xx = 1;
else if (!strcasecmp(opt,"ch")) {}
else break;
longidx++;
}
if ((c->argc - longidx) % 3 || (xx && nx)) {
/* Need an odd number of arguments if we got this far... */
addReplyError(c, "syntax error. Try GEOADD key [x1] [y1] [name1] "
"[x2] [y2] [name2] ... ");
addReplyErrorObject(c,shared.syntaxerr);
return;
}
int elements = (c->argc - 2) / 3;
int argc = 2+elements*2; /* ZADD key score ele ... */
/* Set up the vector for calling ZADD. */
int elements = (c->argc - longidx) / 3;
int argc = longidx+elements*2; /* ZADD key [CH] [NX|XX] score ele ... */
robj **argv = zcalloc(argc*sizeof(robj*));
argv[0] = createRawStringObject("zadd",4);
argv[1] = c->argv[1]; /* key */
incrRefCount(argv[1]);
for (i = 1; i < longidx; i++) {
argv[i] = c->argv[i];
incrRefCount(argv[i]);
}
/* Create the argument vector to call ZADD in order to add all
* the score,value pairs to the requested zset, where score is actually
* an encoded version of lat,long. */
int i;
for (i = 0; i < elements; i++) {
double xy[2];
if (extractLongLatOrReply(c, (c->argv+2)+(i*3),xy) == C_ERR) {
if (extractLongLatOrReply(c, (c->argv+longidx)+(i*3),xy) == C_ERR) {
for (i = 0; i < argc; i++)
if (argv[i]) decrRefCount(argv[i]);
zfree(argv);
......@@ -464,9 +481,9 @@ void geoaddCommand(client *c) {
geohashEncodeWGS84(xy[0], xy[1], GEO_STEP_MAX, &hash);
GeoHashFix52Bits bits = geohashAlign52Bits(hash);
robj *score = createObject(OBJ_STRING, sdsfromlonglong(bits));
robj *val = c->argv[2 + i * 3 + 2];
argv[2+i*2] = score;
argv[3+i*2] = val;
robj *val = c->argv[longidx + i * 3 + 2];
argv[longidx+i*2] = score;
argv[longidx+1+i*2] = val;
incrRefCount(val);
}
......@@ -486,12 +503,12 @@ void geoaddCommand(client *c) {
#define GEOSEARCHSTORE (1<<4) /* GEOSEARCHSTORE just accept STOREDIST option */
/* GEORADIUS key x y radius unit [WITHDIST] [WITHHASH] [WITHCOORD] [ASC|DESC]
* [COUNT count] [STORE key] [STOREDIST key]
* [COUNT count [ANY]] [STORE key] [STOREDIST key]
* GEORADIUSBYMEMBER key member radius unit ... options ...
* GEOSEARCH key [FROMMEMBER member] [FORMLOG long lat] [BYRADIUS radius unit]
* [BYBOX width height unit] [WITHCORD] [WITHDIST] [WITHASH] [COUNT count] [ASC|DESC]
* GEOSEARCHSTORE dest_key src_key [FROMMEMBER member] [FORMLOG long lat] [BYRADIUS radius unit]
* [BYBOX width height unit] [WITHCORD] [WITHDIST] [WITHASH] [COUNT count] [ASC|DESC] [STOREDIST]
* GEOSEARCH key [FROMMEMBER member] [FROMLONLAT long lat] [BYRADIUS radius unit]
* [BYBOX width height unit] [WITHCORD] [WITHDIST] [WITHASH] [COUNT count [ANY]] [ASC|DESC]
* GEOSEARCHSTORE dest_key src_key [FROMMEMBER member] [FROMLONLAT long lat] [BYRADIUS radius unit]
* [BYBOX width height unit] [WITHCORD] [WITHDIST] [WITHASH] [COUNT count [ANY]] [ASC|DESC] [STOREDIST]
* */
void georadiusGeneric(client *c, int srcKeyIndex, int flags) {
robj *storekey = NULL;
......@@ -536,7 +553,8 @@ void georadiusGeneric(client *c, int srcKeyIndex, int flags) {
int withdist = 0, withhash = 0, withcoords = 0;
int frommember = 0, fromloc = 0, byradius = 0, bybox = 0;
int sort = SORT_NONE;
long long count = 0;
int any = 0; /* any=1 means a limited search, stop as soon as enough results were found. */
long long count = 0; /* Max number of results to return. 0 means unlimited. */
if (c->argc > base_args) {
int remaining = c->argc - base_args;
for (int i = 0; i < remaining; i++) {
......@@ -547,13 +565,15 @@ void georadiusGeneric(client *c, int srcKeyIndex, int flags) {
withhash = 1;
} else if (!strcasecmp(arg, "withcoord")) {
withcoords = 1;
} else if (!strcasecmp(arg, "any")) {
any = 1;
} else if (!strcasecmp(arg, "asc")) {
sort = SORT_ASC;
} else if (!strcasecmp(arg, "desc")) {
sort = SORT_DESC;
} else if (!strcasecmp(arg, "count") && (i+1) < remaining) {
if (getLongLongFromObjectOrReply(c, c->argv[base_args+i+1],
&count, NULL) != C_OK) return;
&count, NULL) != C_OK) return;
if (count <= 0) {
addReplyError(c,"COUNT must be > 0");
return;
......@@ -620,7 +640,7 @@ void georadiusGeneric(client *c, int srcKeyIndex, int flags) {
bybox = 1;
i += 3;
} else {
addReply(c, shared.syntaxerr);
addReplyErrorObject(c,shared.syntaxerr);
return;
}
}
......@@ -648,16 +668,23 @@ void georadiusGeneric(client *c, int srcKeyIndex, int flags) {
return;
}
/* COUNT without ordering does not make much sense, force ASC
* ordering if COUNT was specified but no sorting was requested. */
if (count != 0 && sort == SORT_NONE) sort = SORT_ASC;
if (any && !count) {
addReplyErrorFormat(c, "the ANY argument requires COUNT argument");
return;
}
/* COUNT without ordering does not make much sense (we need to
* sort in order to return the closest N entries),
* force ASC ordering if COUNT was specified but no sorting was
* requested. Note that this is not needed for ANY option. */
if (count != 0 && sort == SORT_NONE && !any) sort = SORT_ASC;
/* Get all neighbor geohash boxes for our radius search */
GeoHashRadius georadius = geohashCalculateAreasByShapeWGS84(&shape);
/* Search the zset for all matching points */
geoArray *ga = geoArrayCreate();
membersOfAllNeighbors(zobj, georadius, &shape, ga);
membersOfAllNeighbors(zobj, georadius, &shape, ga, any ? count : 0);
/* If no matching results, the user gets an empty reply. */
if (ga->used == 0 && storekey == NULL) {
......@@ -902,7 +929,7 @@ void geodistCommand(client *c) {
to_meter = extractUnitOrReply(c,c->argv[4]);
if (to_meter < 0) return;
} else if (c->argc > 5) {
addReply(c,shared.syntaxerr);
addReplyErrorObject(c,shared.syntaxerr);
return;
}
......
......@@ -194,7 +194,7 @@ struct commandHelp {
8,
"2.4.0" },
{ "CLIENT PAUSE",
"timeout",
"timeout [WRITE|ALL]",
"Stop processing commands from clients for some time",
8,
"2.9.50" },
......@@ -213,11 +213,21 @@ struct commandHelp {
"Enable or disable server assisted client side caching support",
8,
"6.0.0" },
{ "CLIENT TRACKINGINFO",
"-",
"Return information about server assisted client side caching for the current connection",
8,
"6.2.0" },
{ "CLIENT UNBLOCK",
"client-id [TIMEOUT|ERROR]",
"Unblock a client blocked in a blocking command from a different connection",
8,
"5.0.0" },
{ "CLIENT UNPAUSE",
"-",
"Resume processing of clients that were paused",
8,
"6.2.0" },
{ "CLUSTER ADDSLOTS",
"slot [slot ...]",
"Assign new hash slots to receiving node",
......@@ -459,7 +469,7 @@ struct commandHelp {
9,
"1.0.0" },
{ "GEOADD",
"key longitude latitude member [longitude latitude member ...]",
"key [NX|XX] [CH] longitude latitude member [longitude latitude member ...]",
"Add one or more geospatial items in the geospatial index represented using a sorted set",
13,
"3.2.0" },
......@@ -479,22 +489,22 @@ struct commandHelp {
13,
"3.2.0" },
{ "GEORADIUS",
"key longitude latitude radius m|km|ft|mi [WITHCOORD] [WITHDIST] [WITHHASH] [COUNT count] [ASC|DESC] [STORE key] [STOREDIST key]",
"key longitude latitude radius m|km|ft|mi [WITHCOORD] [WITHDIST] [WITHHASH] [COUNT count [ANY]] [ASC|DESC] [STORE key] [STOREDIST key]",
"Query a sorted set representing a geospatial index to fetch members matching a given maximum distance from a point",
13,
"3.2.0" },
{ "GEORADIUSBYMEMBER",
"key member radius m|km|ft|mi [WITHCOORD] [WITHDIST] [WITHHASH] [COUNT count] [ASC|DESC] [STORE key] [STOREDIST key]",
"key member radius m|km|ft|mi [WITHCOORD] [WITHDIST] [WITHHASH] [COUNT count [ANY]] [ASC|DESC] [STORE key] [STOREDIST key]",
"Query a sorted set representing a geospatial index to fetch members matching a given maximum distance from a member",
13,
"3.2.0" },
{ "GEOSEARCH",
"key [FROMMEMBER member] [FROMLONLAT longitude latitude] [BYRADIUS radius m|km|ft|mi] [BYBOX width height m|km|ft|mi] [ASC|DESC] [COUNT count] [WITHCOORD] [WITHDIST] [WITHHASH]",
"key [FROMMEMBER member] [FROMLONLAT longitude latitude] [BYRADIUS radius m|km|ft|mi] [BYBOX width height m|km|ft|mi] [ASC|DESC] [COUNT count [ANY]] [WITHCOORD] [WITHDIST] [WITHHASH]",
"Query a sorted set representing a geospatial index to fetch members inside an area of a box or a circle.",
13,
"6.2" },
{ "GEOSEARCHSTORE",
"destination source [FROMMEMBER member] [FROMLONLAT longitude latitude] [BYRADIUS radius m|km|ft|mi] [BYBOX width height m|km|ft|mi] [ASC|DESC] [COUNT count] [WITHCOORD] [WITHDIST] [WITHHASH] [STOREDIST]",
"destination source [FROMMEMBER member] [FROMLONLAT longitude latitude] [BYRADIUS radius m|km|ft|mi] [BYBOX width height m|km|ft|mi] [ASC|DESC] [COUNT count [ANY]] [WITHCOORD] [WITHDIST] [WITHHASH] [STOREDIST]",
"Query a sorted set representing a geospatial index to fetch members inside an area of a box or a circle, and store the result in another key.",
13,
"6.2" },
......@@ -524,8 +534,8 @@ struct commandHelp {
5,
"2.0.0" },
{ "HELLO",
"protover [AUTH username password] [SETNAME clientname]",
"switch Redis protocol",
"[protover [AUTH username password] [SETNAME clientname]]",
"Handshake with Redis",
8,
"6.0.0" },
{ "HEXISTS",
......@@ -684,8 +694,8 @@ struct commandHelp {
9,
"5.0.0" },
{ "LPOP",
"key",
"Remove and get the first element in a list",
"key [count]",
"Remove and get the first elements in a list",
2,
"1.0.0" },
{ "LPOS",
......@@ -929,8 +939,8 @@ struct commandHelp {
9,
"2.8.12" },
{ "RPOP",
"key",
"Remove and get the last element in a list",
"key [count]",
"Remove and get the last elements in a list",
2,
"1.0.0" },
{ "RPOPLPUSH",
......@@ -1189,10 +1199,15 @@ struct commandHelp {
14,
"5.0.0" },
{ "XADD",
"key [MAXLEN [=|~] length] [NOMKSTREAM] *|ID field value [field value ...]",
"key [NOMKSTREAM] [MAXLEN|MINID [=|~] threshold [LIMIT count]] *|ID field value [field value ...]",
"Appends a new entry to a stream",
14,
"5.0.0" },
{ "XAUTOCLAIM",
"key group consumer min-idle-time start [COUNT count] [justid]",
"Changes (or acquires) ownership of messages in a consumer group, as if the messages were delivered to the specified consumer.",
14,
"6.2.0" },
{ "XCLAIM",
"key group consumer min-idle-time ID [ID ...] [IDLE ms] [TIME ms-unix-time] [RETRYCOUNT count] [force] [justid]",
"Changes (or acquires) ownership of a message in a consumer group, as if the message was delivered to the specified consumer.",
......@@ -1244,7 +1259,7 @@ struct commandHelp {
14,
"5.0.0" },
{ "XTRIM",
"key MAXLEN [=|~] length",
"key MAXLEN|MINID [=|~] threshold [LIMIT count]",
"Trims the stream to (approximately if '~' is passed) a certain size",
14,
"5.0.0" },
......@@ -1309,8 +1324,8 @@ struct commandHelp {
4,
"5.0.0" },
{ "ZRANGE",
"key start stop [WITHSCORES]",
"Return a range of members in a sorted set, by index",
"key min max [BYSCORE|BYLEX] [REV] [LIMIT offset count] [WITHSCORES]",
"Return a range of members in a sorted set",
4,
"1.2.0" },
{ "ZRANGEBYLEX",
......@@ -1323,6 +1338,11 @@ struct commandHelp {
"Return a range of members in a sorted set, by score",
4,
"1.0.5" },
{ "ZRANGESTORE",
"dst src min max [BYSCORE|BYLEX] [REV] [LIMIT offset count]",
"Store a range of members from sorted set into another key",
4,
"6.2.0" },
{ "ZRANK",
"key member",
"Determine the index of a member in a sorted set",
......
......@@ -205,7 +205,7 @@ struct hllhdr {
#define HLL_RAW 255 /* Only used internally, never exposed. */
#define HLL_MAX_ENCODING 1
static char *invalid_hll_err = "-INVALIDOBJ Corrupted HLL object detected\r\n";
static char *invalid_hll_err = "-INVALIDOBJ Corrupted HLL object detected";
/* =========================== Low level bit macros ========================= */
......@@ -1171,9 +1171,8 @@ int isHLLObjectOrReply(client *c, robj *o) {
return C_OK;
invalid:
addReplySds(c,
sdsnew("-WRONGTYPE Key is not a valid "
"HyperLogLog string value.\r\n"));
addReplyError(c,"-WRONGTYPE Key is not a valid "
"HyperLogLog string value.");
return C_ERR;
}
......@@ -1203,7 +1202,7 @@ void pfaddCommand(client *c) {
updated++;
break;
case -1:
addReplySds(c,sdsnew(invalid_hll_err));
addReplyError(c,invalid_hll_err);
return;
}
}
......@@ -1211,7 +1210,7 @@ void pfaddCommand(client *c) {
if (updated) {
signalModifiedKey(c,c->db,c->argv[1]);
notifyKeyspaceEvent(NOTIFY_STRING,"pfadd",c->argv[1],c->db->id);
server.dirty++;
server.dirty += updated;
HLL_INVALIDATE_CACHE(hdr);
}
addReply(c, updated ? shared.cone : shared.czero);
......@@ -1245,7 +1244,7 @@ void pfcountCommand(client *c) {
/* Merge with this HLL with our 'max' HLL by setting max[i]
* to MAX(max[i],hll[i]). */
if (hllMerge(registers,o) == C_ERR) {
addReplySds(c,sdsnew(invalid_hll_err));
addReplyError(c,invalid_hll_err);
return;
}
}
......@@ -1285,7 +1284,7 @@ void pfcountCommand(client *c) {
/* Recompute it and update the cached value. */
card = hllCount(hdr,&invalid);
if (invalid) {
addReplySds(c,sdsnew(invalid_hll_err));
addReplyError(c,invalid_hll_err);
return;
}
hdr->card[0] = card & 0xff;
......@@ -1332,7 +1331,7 @@ void pfmergeCommand(client *c) {
/* Merge with this HLL with our 'max' HLL by setting max[i]
* to MAX(max[i],hll[i]). */
if (hllMerge(max,o) == C_ERR) {
addReplySds(c,sdsnew(invalid_hll_err));
addReplyError(c,invalid_hll_err);
return;
}
}
......@@ -1355,7 +1354,7 @@ void pfmergeCommand(client *c) {
/* Convert the destination object to dense representation if at least
* one of the inputs was dense. */
if (use_dense && hllSparseToDense(o) == C_ERR) {
addReplySds(c,sdsnew(invalid_hll_err));
addReplyError(c,invalid_hll_err);
return;
}
......@@ -1512,7 +1511,7 @@ void pfdebugCommand(client *c) {
if (hdr->encoding == HLL_SPARSE) {
if (hllSparseToDense(o) == C_ERR) {
addReplySds(c,sdsnew(invalid_hll_err));
addReplyError(c,invalid_hll_err);
return;
}
server.dirty++; /* Force propagation on encoding change. */
......@@ -1577,7 +1576,7 @@ void pfdebugCommand(client *c) {
if (hdr->encoding == HLL_SPARSE) {
if (hllSparseToDense(o) == C_ERR) {
addReplySds(c,sdsnew(invalid_hll_err));
addReplyError(c,invalid_hll_err);
return;
}
conv = 1;
......
......@@ -358,12 +358,6 @@ static long long usec(void) {
return (((long long)tv.tv_sec)*1000000)+tv.tv_usec;
}
#define assert(_e) ((_e)?(void)0:(_assert(#_e,__FILE__,__LINE__),exit(1)))
static void _assert(char *estr, char *file, int line) {
printf("\n\n=== ASSERTION FAILED ===\n");
printf("==> %s:%d '%s' is not true\n",file,line,estr);
}
static intset *createSet(int bits, int size) {
uint64_t mask = (1<<bits)-1;
uint64_t value;
......
......@@ -584,16 +584,6 @@ sds latencyCommandGenSparkeline(char *event, struct latencyTimeSeries *ts) {
* LATENCY RESET: reset data of a specified event or all the data if no event provided.
*/
void latencyCommand(client *c) {
const char *help[] = {
"DOCTOR -- Returns a human readable latency analysis report.",
"GRAPH <event> -- Returns an ASCII latency graph for the event class.",
"HISTORY <event> -- Returns time-latency samples for the event class.",
"LATEST -- Returns the latest latency samples for all events.",
"RESET [event ...] -- Resets latency data of one or more event classes.",
" (default: reset all data for all event classes)",
"HELP -- Prints this help.",
NULL
};
struct latencyTimeSeries *ts;
if (!strcasecmp(c->argv[1]->ptr,"history") && c->argc == 3) {
......@@ -639,6 +629,20 @@ NULL
addReplyLongLong(c,resets);
}
} else if (!strcasecmp(c->argv[1]->ptr,"help") && c->argc == 2) {
const char *help[] = {
"DOCTOR",
" Return a human readable latency analysis report.",
"GRAPH <event>",
" Return an ASCII latency graph for the <event> class.",
"HISTORY <event>",
" Return time-latency samples for the <event> class.",
"LATEST",
" Return the latest latency samples for all events.",
"RESET [<event> ...]",
" Reset latency data of one or more <event> classes.",
" (default: reset all data for all event classes)",
NULL
};
addReplyHelp(c, help);
} else {
addReplySubcommandSyntaxError(c);
......
......@@ -6,6 +6,49 @@
static redisAtomic size_t lazyfree_objects = 0;
static redisAtomic size_t lazyfreed_objects = 0;
/* Release objects from the lazyfree thread. It's just decrRefCount()
* updating the count of objects to release. */
void lazyfreeFreeObject(void *args[]) {
robj *o = (robj *) args[0];
decrRefCount(o);
atomicDecr(lazyfree_objects,1);
atomicIncr(lazyfreed_objects,1);
}
/* Release a database from the lazyfree thread. The 'db' pointer is the
* database which was substituted with a fresh one in the main thread
* when the database was logically deleted. */
void lazyfreeFreeDatabase(void *args[]) {
dict *ht1 = (dict *) args[0];
dict *ht2 = (dict *) args[1];
size_t numkeys = dictSize(ht1);
dictRelease(ht1);
dictRelease(ht2);
atomicDecr(lazyfree_objects,numkeys);
atomicIncr(lazyfreed_objects,numkeys);
}
/* Release the skiplist mapping Redis Cluster keys to slots in the
* lazyfree thread. */
void lazyfreeFreeSlotsMap(void *args[]) {
rax *rt = args[0];
size_t len = rt->numele;
raxFree(rt);
atomicDecr(lazyfree_objects,len);
atomicIncr(lazyfreed_objects,len);
}
/* Release the rax mapping Redis Cluster keys to slots in the
* lazyfree thread. */
void lazyFreeTrackingTable(void *args[]) {
rax *rt = args[0];
size_t len = rt->numele;
raxFree(rt);
atomicDecr(lazyfree_objects,len);
atomicIncr(lazyfreed_objects,len);
}
/* Return the number of currently pending objects to free. */
size_t lazyfreeGetPendingObjectsCount(void) {
size_t aux;
......@@ -120,7 +163,7 @@ int dbAsyncDelete(redisDb *db, robj *key) {
* equivalent to just calling decrRefCount(). */
if (free_effort > LAZYFREE_THRESHOLD && val->refcount == 1) {
atomicIncr(lazyfree_objects,1);
bioCreateBackgroundJob(BIO_LAZY_FREE,val,NULL,NULL);
bioCreateLazyFreeJob(lazyfreeFreeObject,1, val);
dictSetVal(db->dict,de,NULL);
}
}
......@@ -141,7 +184,7 @@ void freeObjAsync(robj *key, robj *obj) {
size_t free_effort = lazyfreeGetFreeEffort(key,obj);
if (free_effort > LAZYFREE_THRESHOLD && obj->refcount == 1) {
atomicIncr(lazyfree_objects,1);
bioCreateBackgroundJob(BIO_LAZY_FREE,obj,NULL,NULL);
bioCreateLazyFreeJob(lazyfreeFreeObject,1,obj);
} else {
decrRefCount(obj);
}
......@@ -155,39 +198,17 @@ void emptyDbAsync(redisDb *db) {
db->dict = dictCreate(&dbDictType,NULL);
db->expires = dictCreate(&dbExpiresDictType,NULL);
atomicIncr(lazyfree_objects,dictSize(oldht1));
bioCreateBackgroundJob(BIO_LAZY_FREE,NULL,oldht1,oldht2);
bioCreateLazyFreeJob(lazyfreeFreeDatabase,2,oldht1,oldht2);
}
/* Release the radix tree mapping Redis Cluster keys to slots asynchronously. */
void freeSlotsToKeysMapAsync(rax *rt) {
atomicIncr(lazyfree_objects,rt->numele);
bioCreateBackgroundJob(BIO_LAZY_FREE,NULL,NULL,rt);
}
/* Release objects from the lazyfree thread. It's just decrRefCount()
* updating the count of objects to release. */
void lazyfreeFreeObjectFromBioThread(robj *o) {
decrRefCount(o);
atomicDecr(lazyfree_objects,1);
atomicIncr(lazyfreed_objects,1);
bioCreateLazyFreeJob(lazyfreeFreeSlotsMap,1,rt);
}
/* Release a database from the lazyfree thread. The 'db' pointer is the
* database which was substituted with a fresh one in the main thread
* when the database was logically deleted. */
void lazyfreeFreeDatabaseFromBioThread(dict *ht1, dict *ht2) {
size_t numkeys = dictSize(ht1);
dictRelease(ht1);
dictRelease(ht2);
atomicDecr(lazyfree_objects,numkeys);
atomicIncr(lazyfreed_objects,numkeys);
}
/* Release the radix tree mapping Redis Cluster keys to slots in the
* lazyfree thread. */
void lazyfreeFreeSlotsMapFromBioThread(rax *rt) {
size_t len = rt->numele;
raxFree(rt);
atomicDecr(lazyfree_objects,len);
atomicIncr(lazyfreed_objects,len);
/* Free an object, if the object is huge enough, free it in async way. */
void freeTrackingRadixTreeAsync(rax *tracking) {
atomicIncr(lazyfree_objects,tracking->numele);
bioCreateLazyFreeJob(lazyFreeTrackingTable,1,tracking);
}
......@@ -747,6 +747,7 @@ int64_t commandFlagsFromString(char *s) {
else if (!strcasecmp(t,"no-slowlog")) flags |= CMD_SKIP_SLOWLOG;
else if (!strcasecmp(t,"fast")) flags |= CMD_FAST;
else if (!strcasecmp(t,"no-auth")) flags |= CMD_NO_AUTH;
else if (!strcasecmp(t,"may-replicate")) flags |= CMD_MAY_REPLICATE;
else if (!strcasecmp(t,"getkeys-api")) flags |= CMD_MODULE_GETKEYS;
else if (!strcasecmp(t,"no-cluster")) flags |= CMD_MODULE_NO_CLUSTER;
else break;
......@@ -813,6 +814,8 @@ int64_t commandFlagsFromString(char *s) {
* * **"no-auth"**: This command can be run by an un-authenticated client.
* Normally this is used by a command that is used
* to authenticate a client.
* * **"may-replicate"**: This command may generate replication traffic, even
* though it's not a write command.
*/
int RM_CreateCommand(RedisModuleCtx *ctx, const char *name, RedisModuleCmdFunc cmdfunc, const char *strflags, int firstkey, int lastkey, int keystep) {
int64_t flags = strflags ? commandFlagsFromString((char*)strflags) : 0;
......@@ -851,6 +854,8 @@ int RM_CreateCommand(RedisModuleCtx *ctx, const char *name, RedisModuleCmdFunc c
cp->rediscmd->keystep = keystep;
cp->rediscmd->microseconds = 0;
cp->rediscmd->calls = 0;
cp->rediscmd->rejected_calls = 0;
cp->rediscmd->failed_calls = 0;
dictAdd(server.commands,sdsdup(cmdname),cp->rediscmd);
dictAdd(server.orig_commands,sdsdup(cmdname),cp->rediscmd);
cp->rediscmd->id = ACLGetCommandID(cmdname); /* ID used for ACL. */
......@@ -1368,18 +1373,6 @@ int RM_ReplyWithLongLong(RedisModuleCtx *ctx, long long ll) {
return REDISMODULE_OK;
}
/* Reply with an error or simple string (status message). Used to implement
* ReplyWithSimpleString() and ReplyWithError().
* The function always returns REDISMODULE_OK. */
int replyWithStatus(RedisModuleCtx *ctx, const char *msg, char *prefix) {
client *c = moduleGetReplyClient(ctx);
if (c == NULL) return REDISMODULE_OK;
addReplyProto(c,prefix,strlen(prefix));
addReplyProto(c,msg,strlen(msg));
addReplyProto(c,"\r\n",2);
return REDISMODULE_OK;
}
/* Reply with the error 'err'.
*
* Note that 'err' must contain all the error, including
......@@ -1395,7 +1388,10 @@ int replyWithStatus(RedisModuleCtx *ctx, const char *msg, char *prefix) {
* The function always returns REDISMODULE_OK.
*/
int RM_ReplyWithError(RedisModuleCtx *ctx, const char *err) {
return replyWithStatus(ctx,err,"-");
client *c = moduleGetReplyClient(ctx);
if (c == NULL) return REDISMODULE_OK;
addReplyErrorFormat(c,"-%s",err);
return REDISMODULE_OK;
}
/* Reply with a simple string (+... \r\n in RESP protocol). This replies
......@@ -1404,7 +1400,12 @@ int RM_ReplyWithError(RedisModuleCtx *ctx, const char *err) {
*
* The function always returns REDISMODULE_OK. */
int RM_ReplyWithSimpleString(RedisModuleCtx *ctx, const char *msg) {
return replyWithStatus(ctx,msg,"+");
client *c = moduleGetReplyClient(ctx);
if (c == NULL) return REDISMODULE_OK;
addReplyProto(c,"+",1);
addReplyProto(c,msg,strlen(msg));
addReplyProto(c,"\r\n",2);
return REDISMODULE_OK;
}
/* Reply with an array type of 'len' elements. However 'len' other calls
......@@ -1629,7 +1630,7 @@ void moduleReplicateMultiIfNeeded(RedisModuleCtx *ctx) {
ctx->saved_oparray = server.also_propagate;
redisOpArrayInit(&server.also_propagate);
}
execCommandPropagateMulti(ctx->client);
execCommandPropagateMulti(ctx->client->db->id);
}
/* Replicate the specified command and arguments to slaves and AOF, as effect
......@@ -2044,7 +2045,7 @@ int RM_GetContextFlags(RedisModuleCtx *ctx) {
* periodically in timer callbacks or other periodic callbacks.
*/
int RM_AvoidReplicaTraffic() {
return clientsArePaused();
return checkClientPauseTimeoutAndReturnIfPaused();
}
/* Change the currently selected DB. Returns an error if the id
......@@ -7067,33 +7068,32 @@ int RM_ScanKey(RedisModuleKey *key, RedisModuleScanCursor *cursor, RedisModuleSc
*/
int RM_Fork(RedisModuleForkDoneHandler cb, void *user_data) {
pid_t childpid;
if (hasActiveChildProcess()) {
return -1;
}
openChildInfoPipe();
if ((childpid = redisFork(CHILD_TYPE_MODULE)) == 0) {
/* Child */
redisSetProcTitle("redis-module-fork");
} else if (childpid == -1) {
closeChildInfoPipe();
serverLog(LL_WARNING,"Can't fork for module: %s", strerror(errno));
} else {
/* Parent */
server.module_child_pid = childpid;
moduleForkInfo.done_handler = cb;
moduleForkInfo.done_handler_user_data = user_data;
updateDictResizePolicy();
serverLog(LL_VERBOSE, "Module fork started pid: %ld ", (long) childpid);
}
return childpid;
}
/* The module is advised to call this function from the fork child once in a while,
* so that it can report COW memory to the parent which will be reported in INFO */
void RM_SendChildCOWInfo(void) {
sendChildCOWInfo(CHILD_TYPE_MODULE, 0, "Module fork");
}
/* Call from the child process when you want to terminate it.
* retcode will be provided to the done handler executed on the parent process.
*/
int RM_ExitFromChild(int retcode) {
sendChildCOWInfo(CHILD_TYPE_MODULE, "Module fork");
sendChildCOWInfo(CHILD_TYPE_MODULE, 1, "Module fork");
exitFromChild(retcode);
return REDISMODULE_OK;
}
......@@ -7103,22 +7103,20 @@ int RM_ExitFromChild(int retcode) {
* child or the pid does not match, return C_ERR without doing anything. */
int TerminateModuleForkChild(int child_pid, int wait) {
/* Module child should be active and pid should match. */
if (server.module_child_pid == -1 ||
server.module_child_pid != child_pid) return C_ERR;
if (server.child_type != CHILD_TYPE_MODULE ||
server.child_pid != child_pid) return C_ERR;
int statloc;
serverLog(LL_VERBOSE,"Killing running module fork child: %ld",
(long) server.module_child_pid);
if (kill(server.module_child_pid,SIGUSR1) != -1 && wait) {
while(wait4(server.module_child_pid,&statloc,0,NULL) !=
server.module_child_pid);
(long) server.child_pid);
if (kill(server.child_pid,SIGUSR1) != -1 && wait) {
while(wait4(server.child_pid,&statloc,0,NULL) !=
server.child_pid);
}
/* Reset the buffer accumulating changes while the child saves. */
server.module_child_pid = -1;
resetChildState();
moduleForkInfo.done_handler = NULL;
moduleForkInfo.done_handler_user_data = NULL;
closeChildInfoPipe();
updateDictResizePolicy();
return C_OK;
}
......@@ -7135,12 +7133,12 @@ int RM_KillForkChild(int child_pid) {
void ModuleForkDoneHandler(int exitcode, int bysignal) {
serverLog(LL_NOTICE,
"Module fork exited pid: %ld, retcode: %d, bysignal: %d",
(long) server.module_child_pid, exitcode, bysignal);
(long) server.child_pid, exitcode, bysignal);
if (moduleForkInfo.done_handler) {
moduleForkInfo.done_handler(exitcode, bysignal,
moduleForkInfo.done_handler_user_data);
}
server.module_child_pid = -1;
moduleForkInfo.done_handler = NULL;
moduleForkInfo.done_handler_user_data = NULL;
}
......@@ -7938,14 +7936,21 @@ sds genModulesInfoString(sds info) {
/* Redis MODULE command.
*
* MODULE LOAD <path> [args...] */
* MODULE LIST
* MODULE LOAD <path> [args...]
* MODULE UNLOAD <name>
*/
void moduleCommand(client *c) {
char *subcmd = c->argv[1]->ptr;
if (c->argc == 2 && !strcasecmp(subcmd,"help")) {
const char *help[] = {
"LIST -- Return a list of loaded modules.",
"LOAD <path> [arg ...] -- Load a module library from <path>.",
"UNLOAD <name> -- Unload a module.",
"LIST",
" Return a list of loaded modules.",
"LOAD <path> [<arg> ...]",
" Load a module library from <path>, passing to it any optional arguments.",
"UNLOAD <name>",
" Unload a module.",
NULL
};
addReplyHelp(c, help);
......@@ -8594,6 +8599,7 @@ void moduleRegisterCoreAPI(void) {
REGISTER_API(CommandFilterArgReplace);
REGISTER_API(CommandFilterArgDelete);
REGISTER_API(Fork);
REGISTER_API(SendChildCOWInfo);
REGISTER_API(ExitFromChild);
REGISTER_API(KillForkChild);
REGISTER_API(RegisterInfoFunc);
......
......@@ -364,7 +364,7 @@ int TestAssertIntegerReply(RedisModuleCtx *ctx, RedisModuleCallReply *reply, lon
do { \
RedisModule_Log(ctx,"warning","Testing %s", name); \
reply = RedisModule_Call(ctx,name,__VA_ARGS__); \
} while (0);
} while (0)
/* TEST.IT -- Run all the tests. */
int TestIt(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
......
/*
A C-program for MT19937-64 (2004/9/29 version).
Coded by Takuji Nishimura and Makoto Matsumoto.
This is a 64-bit version of Mersenne Twister pseudorandom number
generator.
Before using, initialize the state by using init_genrand64(seed)
or init_by_array64(init_key, key_length).
Copyright (C) 2004, Makoto Matsumoto and Takuji Nishimura,
All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions
are met:
1. Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
2. 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.
3. The names of its contributors may not be used to endorse or promote
products derived from this software without specific prior written
permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS 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 COPYRIGHT OWNER OR
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.
References:
T. Nishimura, ``Tables of 64-bit Mersenne Twisters''
ACM Transactions on Modeling and
Computer Simulation 10. (2000) 348--357.
M. Matsumoto and T. Nishimura,
``Mersenne Twister: a 623-dimensionally equidistributed
uniform pseudorandom number generator''
ACM Transactions on Modeling and
Computer Simulation 8. (Jan. 1998) 3--30.
Any feedback is very welcome.
http://www.math.hiroshima-u.ac.jp/~m-mat/MT/emt.html
email: m-mat @ math.sci.hiroshima-u.ac.jp (remove spaces)
*/
#include "mt19937-64.h"
#include <stdio.h>
#define NN 312
#define MM 156
#define MATRIX_A 0xB5026F5AA96619E9ULL
#define UM 0xFFFFFFFF80000000ULL /* Most significant 33 bits */
#define LM 0x7FFFFFFFULL /* Least significant 31 bits */
/* The array for the state vector */
static unsigned long long mt[NN];
/* mti==NN+1 means mt[NN] is not initialized */
static int mti=NN+1;
/* initializes mt[NN] with a seed */
void init_genrand64(unsigned long long seed)
{
mt[0] = seed;
for (mti=1; mti<NN; mti++)
mt[mti] = (6364136223846793005ULL * (mt[mti-1] ^ (mt[mti-1] >> 62)) + mti);
}
/* initialize by an array with array-length */
/* init_key is the array for initializing keys */
/* key_length is its length */
void init_by_array64(unsigned long long init_key[],
unsigned long long key_length)
{
unsigned long long i, j, k;
init_genrand64(19650218ULL);
i=1; j=0;
k = (NN>key_length ? NN : key_length);
for (; k; k--) {
mt[i] = (mt[i] ^ ((mt[i-1] ^ (mt[i-1] >> 62)) * 3935559000370003845ULL))
+ init_key[j] + j; /* non linear */
i++; j++;
if (i>=NN) { mt[0] = mt[NN-1]; i=1; }
if (j>=key_length) j=0;
}
for (k=NN-1; k; k--) {
mt[i] = (mt[i] ^ ((mt[i-1] ^ (mt[i-1] >> 62)) * 2862933555777941757ULL))
- i; /* non linear */
i++;
if (i>=NN) { mt[0] = mt[NN-1]; i=1; }
}
mt[0] = 1ULL << 63; /* MSB is 1; assuring non-zero initial array */
}
/* generates a random number on [0, 2^64-1]-interval */
unsigned long long genrand64_int64(void)
{
int i;
unsigned long long x;
static unsigned long long mag01[2]={0ULL, MATRIX_A};
if (mti >= NN) { /* generate NN words at one time */
/* if init_genrand64() has not been called, */
/* a default initial seed is used */
if (mti == NN+1)
init_genrand64(5489ULL);
for (i=0;i<NN-MM;i++) {
x = (mt[i]&UM)|(mt[i+1]&LM);
mt[i] = mt[i+MM] ^ (x>>1) ^ mag01[(int)(x&1ULL)];
}
for (;i<NN-1;i++) {
x = (mt[i]&UM)|(mt[i+1]&LM);
mt[i] = mt[i+(MM-NN)] ^ (x>>1) ^ mag01[(int)(x&1ULL)];
}
x = (mt[NN-1]&UM)|(mt[0]&LM);
mt[NN-1] = mt[MM-1] ^ (x>>1) ^ mag01[(int)(x&1ULL)];
mti = 0;
}
x = mt[mti++];
x ^= (x >> 29) & 0x5555555555555555ULL;
x ^= (x << 17) & 0x71D67FFFEDA60000ULL;
x ^= (x << 37) & 0xFFF7EEE000000000ULL;
x ^= (x >> 43);
return x;
}
/* generates a random number on [0, 2^63-1]-interval */
long long genrand64_int63(void)
{
return (long long)(genrand64_int64() >> 1);
}
/* generates a random number on [0,1]-real-interval */
double genrand64_real1(void)
{
return (genrand64_int64() >> 11) * (1.0/9007199254740991.0);
}
/* generates a random number on [0,1)-real-interval */
double genrand64_real2(void)
{
return (genrand64_int64() >> 11) * (1.0/9007199254740992.0);
}
/* generates a random number on (0,1)-real-interval */
double genrand64_real3(void)
{
return ((genrand64_int64() >> 12) + 0.5) * (1.0/4503599627370496.0);
}
#ifdef MT19937_64_MAIN
int main(void)
{
int i;
unsigned long long init[4]={0x12345ULL, 0x23456ULL, 0x34567ULL, 0x45678ULL}, length=4;
init_by_array64(init, length);
printf("1000 outputs of genrand64_int64()\n");
for (i=0; i<1000; i++) {
printf("%20llu ", genrand64_int64());
if (i%5==4) printf("\n");
}
printf("\n1000 outputs of genrand64_real2()\n");
for (i=0; i<1000; i++) {
printf("%10.8f ", genrand64_real2());
if (i%5==4) printf("\n");
}
return 0;
}
#endif
/*
A C-program for MT19937-64 (2004/9/29 version).
Coded by Takuji Nishimura and Makoto Matsumoto.
This is a 64-bit version of Mersenne Twister pseudorandom number
generator.
Before using, initialize the state by using init_genrand64(seed)
or init_by_array64(init_key, key_length).
Copyright (C) 2004, Makoto Matsumoto and Takuji Nishimura,
All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions
are met:
1. Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
2. 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.
3. The names of its contributors may not be used to endorse or promote
products derived from this software without specific prior written
permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS 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 COPYRIGHT OWNER OR
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.
References:
T. Nishimura, ``Tables of 64-bit Mersenne Twisters''
ACM Transactions on Modeling and
Computer Simulation 10. (2000) 348--357.
M. Matsumoto and T. Nishimura,
``Mersenne Twister: a 623-dimensionally equidistributed
uniform pseudorandom number generator''
ACM Transactions on Modeling and
Computer Simulation 8. (Jan. 1998) 3--30.
Any feedback is very welcome.
http://www.math.hiroshima-u.ac.jp/~m-mat/MT/emt.html
email: m-mat @ math.sci.hiroshima-u.ac.jp (remove spaces)
*/
#ifndef __MT19937_64_H
#define __MT19937_64_H
/* initializes mt[NN] with a seed */
void init_genrand64(unsigned long long seed);
/* initialize by an array with array-length */
/* init_key is the array for initializing keys */
/* key_length is its length */
void init_by_array64(unsigned long long init_key[],
unsigned long long key_length);
/* generates a random number on [0, 2^64-1]-interval */
unsigned long long genrand64_int64(void);
/* generates a random number on [0, 2^63-1]-interval */
long long genrand64_int63(void);
/* generates a random number on [0,1]-real-interval */
double genrand64_real1(void);
/* generates a random number on [0,1)-real-interval */
double genrand64_real2(void);
/* generates a random number on (0,1)-real-interval */
double genrand64_real3(void);
/* generates a random number on (0,1]-real-interval */
double genrand64_real4(void);
#endif
......@@ -127,15 +127,15 @@ void beforePropagateMultiOrExec(int multi) {
/* Send a MULTI command to all the slaves and AOF file. Check the execCommand
* implementation for more information. */
void execCommandPropagateMulti(client *c) {
void execCommandPropagateMulti(int dbid) {
beforePropagateMultiOrExec(1);
propagate(server.multiCommand,c->db->id,&shared.multi,1,
propagate(server.multiCommand,dbid,&shared.multi,1,
PROPAGATE_AOF|PROPAGATE_REPL);
}
void execCommandPropagateExec(client *c) {
void execCommandPropagateExec(int dbid) {
beforePropagateMultiOrExec(0);
propagate(server.execCommand,c->db->id,&shared.exec,1,
propagate(server.execCommand,dbid,&shared.exec,1,
PROPAGATE_AOF|PROPAGATE_REPL);
}
......@@ -162,7 +162,6 @@ void execCommand(client *c) {
robj **orig_argv;
int orig_argc;
struct redisCommand *orig_cmd;
int must_propagate = 0; /* Need to propagate MULTI/EXEC to AOF / slaves? */
int was_master = server.masterhost == NULL;
if (!(c->flags & CLIENT_MULTI)) {
......@@ -202,19 +201,6 @@ void execCommand(client *c) {
c->argv = c->mstate.commands[j].argv;
c->cmd = c->mstate.commands[j].cmd;
/* Propagate a MULTI request once we encounter the first command which
* is not readonly nor an administrative one.
* This way we'll deliver the MULTI/..../EXEC block as a whole and
* both the AOF and the replication link will have the same consistency
* and atomicity guarantees. */
if (!must_propagate &&
!server.loading &&
!(c->cmd->flags & (CMD_READONLY|CMD_ADMIN)))
{
execCommandPropagateMulti(c);
must_propagate = 1;
}
/* ACL permissions are also checked at the time of execution in case
* they were changed after the commands were ququed. */
int acl_errpos;
......@@ -265,7 +251,7 @@ void execCommand(client *c) {
/* Make sure the EXEC command will be propagated as well if MULTI
* was already propagated. */
if (must_propagate) {
if (server.propagate_in_transaction) {
int is_master = server.masterhost == NULL;
server.dirty++;
beforePropagateMultiOrExec(0);
......@@ -388,31 +374,36 @@ void touchWatchedKey(redisDb *db, robj *key) {
}
}
/* On FLUSHDB or FLUSHALL all the watched keys that are present before the
* flush but will be deleted as effect of the flushing operation should
* be touched. "dbid" is the DB that's getting the flush. -1 if it is
* a FLUSHALL operation (all the DBs flushed). */
void touchWatchedKeysOnFlush(int dbid) {
listIter li1, li2;
/* Set CLIENT_DIRTY_CAS to all clients of DB when DB is dirty.
* It may happen in the following situations:
* FLUSHDB, FLUSHALL, SWAPDB
*
* replaced_with: for SWAPDB, the WATCH should be invalidated if
* the key exists in either of them, and skipped only if it
* doesn't exist in both. */
void touchAllWatchedKeysInDb(redisDb *emptied, redisDb *replaced_with) {
listIter li;
listNode *ln;
/* For every client, check all the waited keys */
listRewind(server.clients,&li1);
while((ln = listNext(&li1))) {
client *c = listNodeValue(ln);
listRewind(c->watched_keys,&li2);
while((ln = listNext(&li2))) {
watchedKey *wk = listNodeValue(ln);
/* For every watched key matching the specified DB, if the
* key exists, mark the client as dirty, as the key will be
* removed. */
if (dbid == -1 || wk->db->id == dbid) {
if (dictFind(wk->db->dict, wk->key->ptr) != NULL)
c->flags |= CLIENT_DIRTY_CAS;
dictEntry *de;
if (dictSize(emptied->watched_keys) == 0) return;
dictIterator *di = dictGetSafeIterator(emptied->watched_keys);
while((de = dictNext(di)) != NULL) {
robj *key = dictGetKey(de);
list *clients = dictGetVal(de);
if (!clients) continue;
listRewind(clients,&li);
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
if (dictFind(emptied->dict, key->ptr)) {
c->flags |= CLIENT_DIRTY_CAS;
} else if (replaced_with && dictFind(replaced_with->dict, key->ptr)) {
c->flags |= CLIENT_DIRTY_CAS;
}
}
}
dictReleaseIterator(di);
}
void watchCommand(client *c) {
......
......@@ -178,6 +178,7 @@ client *createClient(connection *conn) {
c->peerid = NULL;
c->sockname = NULL;
c->client_list_node = NULL;
c->paused_list_node = NULL;
c->client_tracking_redirection = 0;
c->client_tracking_prefixes = NULL;
c->client_cron_last_memory_usage = 0;
......@@ -259,8 +260,14 @@ int prepareClientToWrite(client *c) {
if (!c->conn) return C_ERR; /* Fake client for AOF loading. */
/* Schedule the client to write the output buffers to the socket, unless
* it should already be setup to do so (it has already pending data). */
if (!clientHasPendingReplies(c)) clientInstallWriteHandler(c);
* it should already be setup to do so (it has already pending data).
*
* If CLIENT_PENDING_READ is set, we're in an IO thread and should
* not install a write handler. Instead, it will be done by
* handleClientsWithPendingReadsUsingThreads() upon return.
*/
if (!clientHasPendingReplies(c) && !(c->flags & CLIENT_PENDING_READ))
clientInstallWriteHandler(c);
/* Authorize the caller to queue in the output buffer of this client. */
return C_OK;
......@@ -399,6 +406,24 @@ void addReplyErrorLength(client *c, const char *s, size_t len) {
/* Do some actions after an error reply was sent (Log if needed, updates stats, etc.) */
void afterErrorReply(client *c, const char *s, size_t len) {
/* Increment the global error counter */
server.stat_total_error_replies++;
/* Increment the error stats
* If the string already starts with "-..." then the error prefix
* is provided by the caller ( we limit the search to 32 chars). Otherwise we use "-ERR". */
if (s[0] != '-') {
incrementErrorCount("ERR", 3);
} else {
char *spaceloc = memchr(s, ' ', len < 32 ? len : 32);
if (spaceloc) {
const size_t errEndPos = (size_t)(spaceloc - s);
incrementErrorCount(s+1, errEndPos-1);
} else {
/* Fallback to ERR if we can't retrieve the error prefix */
incrementErrorCount("ERR", 3);
}
}
/* Sometimes it could be normal that a slave replies to a master with
* an error and this function gets called. Actually the error will never
* be sent because addReply*() against master clients has no effect...
......@@ -452,9 +477,11 @@ void addReplyError(client *c, const char *err) {
}
/* See addReplyErrorLength for expectations from the input string. */
/* As a side effect the SDS string is freed. */
void addReplyErrorSds(client *c, sds err) {
addReplyErrorLength(c,err,sdslen(err));
afterErrorReply(c,err,sdslen(err));
sdsfree(err);
}
/* See addReplyErrorLength for expectations from the formatted string.
......@@ -533,13 +560,9 @@ void *addReplyDeferredLen(client *c) {
return listLast(c->reply);
}
/* Populate the length object and try gluing it to the next chunk. */
void setDeferredAggregateLen(client *c, void *node, long length, char prefix) {
serverAssert(length >= 0);
void setDeferredReply(client *c, void *node, const char *s, size_t length) {
listNode *ln = (listNode*)node;
clientReplyBlock *next;
char lenstr[128];
size_t lenstr_len = sprintf(lenstr, "%c%ld\r\n", prefix, length);
/* Abort when *node is NULL: when the client should not accept writes
* we return NULL in addReplyDeferredLen() */
......@@ -557,25 +580,39 @@ void setDeferredAggregateLen(client *c, void *node, long length, char prefix) {
* - It has enough room already allocated
* - And not too large (avoid large memmove) */
if (ln->next != NULL && (next = listNodeValue(ln->next)) &&
next->size - next->used >= lenstr_len &&
next->used < PROTO_REPLY_CHUNK_BYTES * 4) {
memmove(next->buf + lenstr_len, next->buf, next->used);
memcpy(next->buf, lenstr, lenstr_len);
next->used += lenstr_len;
next->size - next->used >= length &&
next->used < PROTO_REPLY_CHUNK_BYTES * 4)
{
memmove(next->buf + length, next->buf, next->used);
memcpy(next->buf, s, length);
next->used += length;
listDelNode(c->reply,ln);
} else {
/* Create a new node */
clientReplyBlock *buf = zmalloc(lenstr_len + sizeof(clientReplyBlock));
clientReplyBlock *buf = zmalloc(length + sizeof(clientReplyBlock));
/* Take over the allocation's internal fragmentation */
buf->size = zmalloc_usable_size(buf) - sizeof(clientReplyBlock);
buf->used = lenstr_len;
memcpy(buf->buf, lenstr, lenstr_len);
buf->used = length;
memcpy(buf->buf, s, length);
listNodeValue(ln) = buf;
c->reply_bytes += buf->size;
}
asyncCloseClientOnOutputBufferLimitReached(c);
}
/* Populate the length object and try gluing it to the next chunk. */
void setDeferredAggregateLen(client *c, void *node, long length, char prefix) {
serverAssert(length >= 0);
/* Abort when *node is NULL: when the client should not accept writes
* we return NULL in addReplyDeferredLen() */
if (node == NULL) return;
char lenstr[128];
size_t lenstr_len = sprintf(lenstr, "%c%ld\r\n", prefix, length);
setDeferredReply(c, node, lenstr, lenstr_len);
}
void setDeferredArrayLen(client *c, void *node, long length) {
setDeferredAggregateLen(c,node,length,'*');
}
......@@ -771,6 +808,14 @@ void addReplyBulkSds(client *c, sds s) {
addReply(c,shared.crlf);
}
/* Set sds to a deferred reply (for symmetry with addReplyBulkSds it also frees the sds) */
void setDeferredReplyBulkSds(client *c, void *node, sds s) {
sds reply = sdscatprintf(sdsempty(), "$%d\r\n%s\r\n", (unsigned)sdslen(s), s);
setDeferredReply(c, node, reply, sdslen(reply));
sdsfree(reply);
sdsfree(s);
}
/* Add a C null term string as bulk reply */
void addReplyBulkCString(client *c, const char *s) {
if (s == NULL) {
......@@ -829,12 +874,16 @@ void addReplyHelp(client *c, const char **help) {
sdstoupper(cmd);
addReplyStatusFormat(c,
"%s <subcommand> arg arg ... arg. Subcommands are:",cmd);
"%s <subcommand> [<arg> [value] [opt] ...]. Subcommands are:",cmd);
sdsfree(cmd);
while (help[blen]) addReplyStatus(c,help[blen++]);
blen++; /* Account for the header line(s). */
addReplyStatus(c,"HELP");
addReplyStatus(c," Prints this help.");
blen += 1; /* Account for the header. */
blen += 2; /* Account for the footer. */
setDeferredArrayLen(c,blenp,blen);
}
......@@ -850,7 +899,7 @@ void addReplySubcommandSyntaxError(client *c) {
sdsfree(cmd);
}
/* Append 'src' client output buffers into 'dst' client output buffers.
/* Append 'src' client output buffers into 'dst' client output buffers.
* This function clears the output buffers of 'src' */
void AddReplyFromClient(client *dst, client *src) {
/* If the source client contains a partial response due to client output
......@@ -1310,7 +1359,7 @@ void freeClient(client *c) {
* to keep data safe and we may delay configured 'save' for full sync. */
if (server.saveparamslen == 0 &&
c->replstate == SLAVE_STATE_WAIT_BGSAVE_END &&
server.rdb_child_pid != -1 &&
server.child_type == CHILD_TYPE_RDB &&
server.rdb_child_type == RDB_CHILD_TYPE_DISK &&
anyOtherSlaveWaitRdb(c) == 0)
{
......@@ -1686,6 +1735,7 @@ int processInlineBuffer(client *c) {
* However the is an exception: masters may send us just a newline
* to keep the connection active. */
if (querylen != 0 && c->flags & CLIENT_MASTER) {
sdsfreesplitres(argv,argc);
serverLog(LL_WARNING,"WARNING: Receiving inline protocol from master, master stream corruption? Closing the master connection and discarding the cached master.");
setProtocolError("Master using the inline protocol. Desync?",c);
return C_ERR;
......@@ -1851,7 +1901,7 @@ int processMultibulkBuffer(client *c) {
c->qb_pos = 0;
/* Hint the sds library about the amount of bytes this string is
* going to contain. */
c->querybuf = sdsMakeRoomFor(c->querybuf,ll+2);
c->querybuf = sdsMakeRoomFor(c->querybuf,ll+2-sdslen(c->querybuf));
}
}
c->bulklen = ll;
......@@ -1911,7 +1961,7 @@ void commandProcessed(client *c) {
* still be able to access the client argv and argc field.
* The client will be reset in unblockClientFromModule(). */
if (!(c->flags & CLIENT_BLOCKED) ||
c->btype != BLOCKED_MODULE)
(c->btype != BLOCKED_MODULE && c->btype != BLOCKED_PAUSE))
{
resetClient(c);
}
......@@ -1954,6 +2004,20 @@ int processCommandAndResetClient(client *c) {
return deadclient ? C_ERR : C_OK;
}
/* This function will execute any fully parsed commands pending on
* the client. Returns C_ERR if the client is no longer valid after executing
* the command, and C_OK for all other cases. */
int processPendingCommandsAndResetClient(client *c) {
if (c->flags & CLIENT_PENDING_COMMAND) {
c->flags &= ~CLIENT_PENDING_COMMAND;
if (processCommandAndResetClient(c) == C_ERR) {
return C_ERR;
}
}
return C_OK;
}
/* This function is called every time, in the client structure 'c', there is
* more query buffer to process, because we read more data from the socket
* or because a client was blocked and later reactivated, so there could be
......@@ -1961,9 +2025,6 @@ int processCommandAndResetClient(client *c) {
void processInputBuffer(client *c) {
/* Keep processing while there is something in the input buffer */
while(c->qb_pos < sdslen(c->querybuf)) {
/* Return if clients are paused. */
if (!(c->flags & CLIENT_SLAVE) && clientsArePaused()) break;
/* Immediately abort if the client is in the middle of something. */
if (c->flags & CLIENT_BLOCKED) break;
......@@ -2361,26 +2422,47 @@ void clientCommand(client *c) {
if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) {
const char *help[] = {
"ID -- Return the ID of the current connection.",
"INFO -- Return information about the current client connection.",
"GETNAME -- Return the name of the current connection.",
"KILL <ip:port> -- Kill connection made from <ip:port>.",
"KILL <option> <value> [option value ...] -- Kill connections. Options are:",
" ADDR <ip:port> -- Kill connection made from <ip:port>",
" LADDR <ip:port> -- Kill connection made to <ip:port>",
" TYPE (normal|master|replica|pubsub) -- Kill connections by type.",
" USER <username> -- Kill connections authenticated with such user.",
" SKIPME (yes|no) -- Skip killing current connection (default: yes).",
"LIST [options ...] -- Return information about client connections. Options:",
" TYPE (normal|master|replica|pubsub) -- Return clients of specified type.",
" ID id [id ...] -- Return clients of specified IDs only.",
"PAUSE <timeout> -- Suspend all Redis clients for <timout> milliseconds.",
"REPLY (on|off|skip) -- Control the replies sent to the current connection.",
"SETNAME <name> -- Assign the name <name> to the current connection.",
"UNBLOCK <clientid> [TIMEOUT|ERROR] -- Unblock the specified blocked client.",
"TRACKING (on|off) [REDIRECT <id>] [BCAST] [PREFIX first] [PREFIX second] [OPTIN] [OPTOUT]... -- Enable client keys tracking for client side caching.",
"CACHING (yes|no) -- Enable/Disable tracking of the keys for next command in OPTIN/OPTOUT mode.",
"GETREDIR -- Return the client ID we are redirecting to when tracking is enabled.",
"CACHING (YES|NO)",
" Enable/disable tracking of the keys for next command in OPTIN/OPTOUT modes.",
"GETREDIR",
" Return the client ID we are redirecting to when tracking is enabled.",
"GETNAME",
" Return the name of the current connection.",
"ID",
" Return the ID of the current connection.",
"INFO",
" Return information about the current client connection.",
"KILL <ip:port>",
" Kill connection made from <ip:port>.",
"KILL <option> <value> [<option> <value> [...]]",
" Kill connections. Options are:",
" * ADDR <ip:port>",
" Kill connection made from <ip:port>",
" * TYPE (normal|master|replica|pubsub)",
" Kill connections by type.",
" * USER <username>",
" Kill connections authenticated by <username>.",
" * SKIPME (YES|NO)",
" Skip killing current connection (default: yes).",
"LIST [options ...]",
" Return information about client connections. Options:",
" * TYPE (NORMAL|MASTER|REPLICA|PUBSUB)",
" Return clients of specified type.",
"UNPAUSE",
" Stop the current client pause, resuming traffic.",
"PAUSE <timeout> [WRITE|ALL]",
" Suspend all, or just write, clients for <timout> milliseconds.",
"REPLY (ON|OFF|SKIP)",
" Control the replies sent to the current connection.",
"SETNAME <name>",
" Assign the name <name> to the current connection.",
"UNBLOCK <clientid> [TIMEOUT|ERROR]",
" Unblock the specified blocked client.",
"TRACKING (ON|OFF) [REDIRECT <id>] [BCAST] [PREFIX <prefix> [...]]",
" [OPTIN] [OPTOUT]",
" Control server assisted client side caching.",
"TRACKINGINFO",
" Report tracking status for the current connection.",
NULL
};
addReplyHelp(c, help);
......@@ -2421,7 +2503,7 @@ NULL
}
}
} else if (c->argc != 2) {
addReply(c,shared.syntaxerr);
addReplyErrorObject(c,shared.syntaxerr);
return;
}
......@@ -2440,7 +2522,7 @@ NULL
if (!(c->flags & CLIENT_REPLY_OFF))
c->flags |= CLIENT_REPLY_SKIP_NEXT;
} else {
addReply(c,shared.syntaxerr);
addReplyErrorObject(c,shared.syntaxerr);
return;
}
} else if (!strcasecmp(c->argv[1]->ptr,"kill")) {
......@@ -2496,17 +2578,17 @@ NULL
} else if (!strcasecmp(c->argv[i+1]->ptr,"no")) {
skipme = 0;
} else {
addReply(c,shared.syntaxerr);
addReplyErrorObject(c,shared.syntaxerr);
return;
}
} else {
addReply(c,shared.syntaxerr);
addReplyErrorObject(c,shared.syntaxerr);
return;
}
i += 2;
}
} else {
addReply(c,shared.syntaxerr);
addReplyErrorObject(c,shared.syntaxerr);
return;
}
......@@ -2585,13 +2667,31 @@ NULL
addReplyBulk(c,c->name);
else
addReplyNull(c);
} else if (!strcasecmp(c->argv[1]->ptr,"pause") && c->argc == 3) {
/* CLIENT PAUSE */
} else if (!strcasecmp(c->argv[1]->ptr,"unpause") && c->argc == 2) {
/* CLIENT UNPAUSE */
unpauseClients();
addReply(c,shared.ok);
} else if (!strcasecmp(c->argv[1]->ptr,"pause") && (c->argc == 3 ||
c->argc == 4))
{
/* CLIENT PAUSE TIMEOUT [WRITE|ALL] */
long long duration;
int type = CLIENT_PAUSE_ALL;
if (c->argc == 4) {
if (!strcasecmp(c->argv[3]->ptr,"write")) {
type = CLIENT_PAUSE_WRITE;
} else if (!strcasecmp(c->argv[3]->ptr,"all")) {
type = CLIENT_PAUSE_ALL;
} else {
addReplyError(c,
"CLIENT PAUSE mode must be WRITE or ALL");
return;
}
}
if (getTimeoutFromObjectOrReply(c,c->argv[2],&duration,
UNIT_MILLISECONDS) != C_OK) return;
pauseClients(duration);
UNIT_MILLISECONDS) != C_OK) return;
pauseClients(duration, type);
addReply(c,shared.ok);
} else if (!strcasecmp(c->argv[1]->ptr,"tracking") && c->argc >= 3) {
/* CLIENT TRACKING (on|off) [REDIRECT <id>] [BCAST] [PREFIX first]
......@@ -2643,7 +2743,7 @@ NULL
prefix[numprefix++] = c->argv[j];
} else {
zfree(prefix);
addReply(c,shared.syntaxerr);
addReplyErrorObject(c,shared.syntaxerr);
return;
}
}
......@@ -2700,12 +2800,19 @@ NULL
return;
}
if (options & CLIENT_TRACKING_BCAST) {
if (!checkPrefixCollisionsOrReply(c,prefix,numprefix)) {
zfree(prefix);
return;
}
}
enableTracking(c,redir,options,prefix,numprefix);
} else if (!strcasecmp(c->argv[2]->ptr,"off")) {
disableTracking(c);
} else {
zfree(prefix);
addReply(c,shared.syntaxerr);
addReplyErrorObject(c,shared.syntaxerr);
return;
}
zfree(prefix);
......@@ -2734,7 +2841,7 @@ NULL
return;
}
} else {
addReply(c,shared.syntaxerr);
addReplyErrorObject(c,shared.syntaxerr);
return;
}
......@@ -2747,23 +2854,90 @@ NULL
} else {
addReplyLongLong(c,-1);
}
} else if (!strcasecmp(c->argv[1]->ptr,"trackinginfo") && c->argc == 2) {
addReplyMapLen(c,3);
/* Flags */
addReplyBulkCString(c,"flags");
void *arraylen_ptr = addReplyDeferredLen(c);
int numflags = 0;
addReplyBulkCString(c,c->flags & CLIENT_TRACKING ? "on" : "off");
numflags++;
if (c->flags & CLIENT_TRACKING_BCAST) {
addReplyBulkCString(c,"bcast");
numflags++;
}
if (c->flags & CLIENT_TRACKING_OPTIN) {
addReplyBulkCString(c,"optin");
numflags++;
if (c->flags & CLIENT_TRACKING_CACHING) {
addReplyBulkCString(c,"caching-yes");
numflags++;
}
}
if (c->flags & CLIENT_TRACKING_OPTOUT) {
addReplyBulkCString(c,"optout");
numflags++;
if (c->flags & CLIENT_TRACKING_CACHING) {
addReplyBulkCString(c,"caching-no");
numflags++;
}
}
if (c->flags & CLIENT_TRACKING_NOLOOP) {
addReplyBulkCString(c,"noloop");
numflags++;
}
if (c->flags & CLIENT_TRACKING_BROKEN_REDIR) {
addReplyBulkCString(c,"broken_redirect");
numflags++;
}
setDeferredSetLen(c,arraylen_ptr,numflags);
/* Redirect */
addReplyBulkCString(c,"redirect");
if (c->flags & CLIENT_TRACKING) {
addReplyLongLong(c,c->client_tracking_redirection);
} else {
addReplyLongLong(c,-1);
}
/* Prefixes */
addReplyBulkCString(c,"prefixes");
if (c->client_tracking_prefixes) {
addReplyArrayLen(c,raxSize(c->client_tracking_prefixes));
raxIterator ri;
raxStart(&ri,c->client_tracking_prefixes);
raxSeek(&ri,"^",NULL,0);
while(raxNext(&ri)) {
addReplyBulkCBuffer(c,ri.key,ri.key_len);
}
raxStop(&ri);
} else {
addReplyArrayLen(c,0);
}
} else {
addReplyErrorFormat(c, "Unknown subcommand or wrong number of arguments for '%s'. Try CLIENT HELP", (char*)c->argv[1]->ptr);
addReplySubcommandSyntaxError(c);
}
}
/* HELLO <protocol-version> [AUTH <user> <password>] [SETNAME <name>] */
/* HELLO [<protocol-version> [AUTH <user> <password>] [SETNAME <name>] ] */
void helloCommand(client *c) {
long long ver;
long long ver = 0;
int next_arg = 1;
if (getLongLongFromObject(c->argv[1],&ver) != C_OK ||
ver < 2 || ver > 3)
{
addReplyError(c,"-NOPROTO unsupported protocol version");
return;
if (c->argc >= 2) {
if (getLongLongFromObjectOrReply(c, c->argv[next_arg++], &ver,
"Protocol version is not an integer or out of range") != C_OK) {
return;
}
if (ver < 2 || ver > 3) {
addReplyError(c,"-NOPROTO unsupported protocol version");
return;
}
}
for (int j = 2; j < c->argc; j++) {
for (int j = next_arg; j < c->argc; j++) {
int moreargs = (c->argc-1) - j;
const char *opt = c->argv[j]->ptr;
if (!strcasecmp(opt,"AUTH") && moreargs >= 2) {
......@@ -2791,7 +2965,7 @@ void helloCommand(client *c) {
}
/* Let's switch to the specified RESP mode. */
c->resp = ver;
if (ver) c->resp = ver;
addReplyMapLen(c,6 + !server.sentinel_mode);
addReplyBulkCString(c,"server");
......@@ -2801,7 +2975,7 @@ void helloCommand(client *c) {
addReplyBulkCString(c,REDIS_VERSION);
addReplyBulkCString(c,"proto");
addReplyLongLong(c,ver);
addReplyLongLong(c,c->resp);
addReplyBulkCString(c,"id");
addReplyLongLong(c,c->id);
......@@ -3074,54 +3248,66 @@ void flushSlavesOutputBuffers(void) {
}
}
/* Pause clients up to the specified unixtime (in ms). While clients
* are paused no command is processed from clients, so the data set can't
* change during that time.
*
* However while this function pauses normal and Pub/Sub clients, slaves are
* still served, so this function can be used on server upgrades where it is
* required that slaves process the latest bytes from the replication stream
* before being turned to masters.
/* Pause clients up to the specified unixtime (in ms) for a given type of
* commands.
*
* A main use case of this function is to allow pausing replication traffic
* so that a failover without data loss to occur. Replicas will continue to receive
* traffic to faciliate this functionality.
*
* This function is also internally used by Redis Cluster for the manual
* failover procedure implemented by CLUSTER FAILOVER.
*
* The function always succeed, even if there is already a pause in progress.
* In such a case, the pause is extended if the duration is more than the
* time left for the previous duration. However if the duration is smaller
* than the time left for the previous pause, no change is made to the
* left duration. */
void pauseClients(mstime_t end) {
if (!server.clients_paused || end > server.clients_pause_end_time)
server.clients_pause_end_time = end;
server.clients_paused = 1;
}
/* Return non-zero if clients are currently paused. As a side effect the
* function checks if the pause time was reached and clear it. */
int clientsArePaused(void) {
if (server.clients_paused &&
server.clients_pause_end_time < server.mstime)
{
listNode *ln;
listIter li;
client *c;
* In such a case, the duration is set to the maximum and new end time and the
* type is set to the more restrictive type of pause. */
void pauseClients(mstime_t end, pause_type type) {
if (type > server.client_pause_type) {
server.client_pause_type = type;
}
server.clients_paused = 0;
if (end > server.client_pause_end_time) {
server.client_pause_end_time = end;
}
/* Put all the clients in the unblocked clients queue in order to
* force the re-processing of the input buffer if any. */
listRewind(server.clients,&li);
while ((ln = listNext(&li)) != NULL) {
c = listNodeValue(ln);
/* We allow write commands that were queued
* up before and after to execute. We need
* to track this state so that we don't assert
* in propagate(). */
if (server.in_exec) {
server.client_pause_in_transaction = 1;
}
}
/* Don't touch slaves and blocked clients.
* The latter pending requests will be processed when unblocked. */
if (c->flags & (CLIENT_SLAVE|CLIENT_BLOCKED)) continue;
queueClientForReprocessing(c);
}
/* Unpause clients and queue them for reprocessing. */
void unpauseClients(void) {
listNode *ln;
listIter li;
client *c;
server.client_pause_type = CLIENT_PAUSE_OFF;
/* Unblock all of the clients so they are reprocessed. */
listRewind(server.paused_clients,&li);
while ((ln = listNext(&li)) != NULL) {
c = listNodeValue(ln);
unblockClient(c);
}
return server.clients_paused;
}
/* Returns true if clients are paused and false otherwise. */
int areClientsPaused(void) {
return server.client_pause_type != CLIENT_PAUSE_OFF;
}
/* Checks if the current client pause has elapsed and unpause clients
* if it has. Also returns true if clients are now paused and false
* otherwise. */
int checkClientPauseTimeoutAndReturnIfPaused(void) {
if (server.client_pause_end_time < server.mstime) {
unpauseClients();
}
return areClientsPaused();
}
/* This function is called by Redis in order to process a few events from
......@@ -3499,16 +3685,20 @@ int handleClientsWithPendingReadsUsingThreads(void) {
c->flags &= ~CLIENT_PENDING_READ;
listDelNode(server.clients_pending_read,ln);
if (c->flags & CLIENT_PENDING_COMMAND) {
c->flags &= ~CLIENT_PENDING_COMMAND;
if (processCommandAndResetClient(c) == C_ERR) {
/* If the client is no longer valid, we avoid
* processing the client later. So we just go
* to the next. */
continue;
}
if (processPendingCommandsAndResetClient(c) == C_ERR) {
/* If the client is no longer valid, we avoid
* processing the client later. So we just go
* to the next. */
continue;
}
processInputBuffer(c);
/* We may have pending replies if a thread readQueryFromClient() produced
* replies and did not install a write handler (it can't).
*/
if (!(c->flags & CLIENT_PENDING_WRITE) && clientHasPendingReplies(c))
clientInstallWriteHandler(c);
}
/* Update processed count on server */
......
......@@ -404,7 +404,7 @@ robj *resetRefCount(robj *obj) {
int checkType(client *c, robj *o, int type) {
/* A NULL is considered an empty key */
if (o && o->type != type) {
addReply(c,shared.wrongtypeerr);
addReplyErrorObject(c,shared.wrongtypeerr);
return 1;
}
return 0;
......@@ -1035,7 +1035,7 @@ struct redisMemOverhead *getMemoryOverheadData(void) {
mem += dictSize(server.repl_scriptcache_dict) * sizeof(dictEntry) +
dictSlots(server.repl_scriptcache_dict) * sizeof(dictEntry*);
if (listLength(server.repl_scriptcache_fifo) > 0) {
mem += listLength(server.repl_scriptcache_fifo) * (sizeof(listNode) +
mem += listLength(server.repl_scriptcache_fifo) * (sizeof(listNode) +
sdsZmallocSize(listNodeValue(listFirst(server.repl_scriptcache_fifo))));
}
mh->lua_caches = mem;
......@@ -1256,10 +1256,18 @@ void objectCommand(client *c) {
if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) {
const char *help[] = {
"ENCODING <key> -- Return the kind of internal representation used in order to store the value associated with a key.",
"FREQ <key> -- Return the access frequency index of the key. The returned integer is proportional to the logarithm of the recent access frequency of the key.",
"IDLETIME <key> -- Return the idle time of the key, that is the approximated number of seconds elapsed since the last access to the key.",
"REFCOUNT <key> -- Return the number of references of the value associated with the specified key.",
"ENCODING <key>",
" Return the kind of internal representation used in order to store the value",
" associated with a <key>.",
"FREQ <key>",
" Return the access frequency index of the <key>. The returned integer is",
" proportional to the logarithm of the recent access frequency of the key.",
"IDLETIME <key>",
" Return the idle time of the <key>, that is the approximated number of",
" seconds elapsed since the last access to the key.",
"REFCOUNT <key>",
" Return the number of references of the value associated with the specified",
" <key>.",
NULL
};
addReplyHelp(c, help);
......@@ -1303,11 +1311,17 @@ NULL
void memoryCommand(client *c) {
if (!strcasecmp(c->argv[1]->ptr,"help") && c->argc == 2) {
const char *help[] = {
"DOCTOR - Return memory problems reports.",
"MALLOC-STATS -- Return internal statistics report from the memory allocator.",
"PURGE -- Attempt to purge dirty pages for reclamation by the allocator.",
"STATS -- Return information about the memory usage of the server.",
"USAGE <key> [SAMPLES <count>] -- Return memory in bytes used by <key> and its value. Nested values are sampled up to <count> times (default: 5).",
"DOCTOR",
" Return memory problems reports.",
"MALLOC-STATS"
" Return internal statistics report from the memory allocator.",
"PURGE",
" Attempt to purge dirty pages for reclamation by the allocator.",
"STATS",
" Return information about the memory usage of the server.",
"USAGE <key> [SAMPLES <count>]",
" Return memory in bytes used by <key> and its value. Nested values are",
" sampled up to <count> times (default: 5).",
NULL
};
addReplyHelp(c, help);
......@@ -1321,13 +1335,13 @@ NULL
if (getLongLongFromObjectOrReply(c,c->argv[j+1],&samples,NULL)
== C_ERR) return;
if (samples < 0) {
addReply(c,shared.syntaxerr);
addReplyErrorObject(c,shared.syntaxerr);
return;
}
if (samples == 0) samples = LLONG_MAX;
j++; /* skip option argument. */
} else {
addReply(c,shared.syntaxerr);
addReplyErrorObject(c,shared.syntaxerr);
return;
}
}
......@@ -1452,6 +1466,6 @@ NULL
else
addReplyError(c, "Error purging dirty pages");
} else {
addReplyErrorFormat(c, "Unknown subcommand or wrong number of arguments for '%s'. Try MEMORY HELP", (char*)c->argv[1]->ptr);
addReplySubcommandSyntaxError(c);
}
}
......@@ -455,9 +455,13 @@ void publishCommand(client *c) {
void pubsubCommand(client *c) {
if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) {
const char *help[] = {
"CHANNELS [<pattern>] -- Return the currently active channels matching a pattern (default: all).",
"NUMPAT -- Return number of subscriptions to patterns.",
"NUMSUB [channel-1 .. channel-N] -- Returns the number of subscribers for the specified channels (excluding patterns, default: none).",
"CHANNELS [<pattern>]",
" Return the currently active channels matching a <pattern> (default: '*').",
"NUMPAT",
" Return number of subscriptions to patterns.",
"NUMSUB [<channel> ...]",
" Return the number of subscribers for the specified channels, excluding",
" pattern subscriptions(default: no channels).",
NULL
};
addReplyHelp(c, help);
......
......@@ -66,10 +66,10 @@ static const size_t optimization_level[] = {4096, 8192, 16384, 32768, 65536};
#else
#define D(...) \
do { \
printf("%s:%s:%d:\t", __FILE__, __FUNCTION__, __LINE__); \
printf("%s:%s:%d:\t", __FILE__, __func__, __LINE__); \
printf(__VA_ARGS__); \
printf("\n"); \
} while (0);
} while (0)
#endif
/* Bookmarks forward declarations */
......@@ -1508,15 +1508,6 @@ void quicklistBookmarksClear(quicklist *ql) {
#include <stdint.h>
#include <sys/time.h>
#define assert(_e) \
do { \
if (!(_e)) { \
printf("\n\n=== ASSERTION FAILED ===\n"); \
printf("==> %s:%d '%s' is not true\n", __FILE__, __LINE__, #_e); \
err++; \
} \
} while (0)
#define yell(str, ...) printf("ERROR! " str "\n\n", __VA_ARGS__)
#define OK printf("\tOK\n")
......@@ -1529,7 +1520,7 @@ void quicklistBookmarksClear(quicklist *ql) {
#define ERR(x, ...) \
do { \
printf("%s:%s:%d:\t", __FILE__, __FUNCTION__, __LINE__); \
printf("%s:%s:%d:\t", __FILE__, __func__, __LINE__); \
printf("ERROR! " x "\n", __VA_ARGS__); \
err++; \
} while (0)
......@@ -1614,7 +1605,7 @@ static int _ql_verify(quicklist *ql, uint32_t len, uint32_t count,
ql_info(ql);
if (len != ql->len) {
yell("quicklist length wrong: expected %d, got %u", len, ql->len);
yell("quicklist length wrong: expected %d, got %lu", len, ql->len);
errors++;
}
......@@ -1670,7 +1661,7 @@ static int _ql_verify(quicklist *ql, uint32_t len, uint32_t count,
if (node->encoding != QUICKLIST_NODE_ENCODING_RAW) {
yell("Incorrect compression: node %d is "
"compressed at depth %d ((%u, %u); total "
"nodes: %u; size: %u; recompress: %d)",
"nodes: %lu; size: %u; recompress: %d)",
at, ql->compress, low_raw, high_raw, ql->len, node->sz,
node->recompress);
errors++;
......@@ -1680,7 +1671,7 @@ static int _ql_verify(quicklist *ql, uint32_t len, uint32_t count,
!node->attempted_compress) {
yell("Incorrect non-compression: node %d is NOT "
"compressed at depth %d ((%u, %u); total "
"nodes: %u; size: %u; recompress: %d; attempted: %d)",
"nodes: %lu; size: %u; recompress: %d; attempted: %d)",
at, ql->compress, low_raw, high_raw, ql->len, node->sz,
node->recompress, node->attempted_compress);
errors++;
......@@ -2706,7 +2697,7 @@ int quicklistTest(int argc, char *argv[]) {
if (node->encoding != QUICKLIST_NODE_ENCODING_RAW) {
ERR("Incorrect compression: node %d is "
"compressed at depth %d ((%u, %u); total "
"nodes: %u; size: %u)",
"nodes: %lu; size: %u)",
at, depth, low_raw, high_raw, ql->len,
node->sz);
}
......@@ -2714,7 +2705,7 @@ int quicklistTest(int argc, char *argv[]) {
if (node->encoding != QUICKLIST_NODE_ENCODING_LZF) {
ERR("Incorrect non-compression: node %d is NOT "
"compressed at depth %d ((%u, %u); total "
"nodes: %u; size: %u; attempted: %d)",
"nodes: %lu; size: %u; attempted: %d)",
at, depth, low_raw, high_raw, ql->len,
node->sz, node->attempted_compress);
}
......
......@@ -61,7 +61,7 @@ void raxDebugShowNode(const char *msg, raxNode *n);
#ifdef RAX_DEBUG_MSG
#define debugf(...) \
if (raxDebugMsg) { \
printf("%s:%s:%d:\t", __FILE__, __FUNCTION__, __LINE__); \
printf("%s:%s:%d:\t", __FILE__, __func__, __LINE__); \
printf(__VA_ARGS__); \
fflush(stdout); \
}
......@@ -1892,7 +1892,7 @@ void raxShow(rax *rax) {
/* Used by debugnode() macro to show info about a given node. */
void raxDebugShowNode(const char *msg, raxNode *n) {
if (raxDebugMsg == 0) return;
printf("%s: %p [%.*s] key:%d size:%d children:",
printf("%s: %p [%.*s] key:%u size:%u children:",
msg, (void*)n, (int)n->size, (char*)n->data, n->iskey, n->size);
int numcld = n->iscompr ? 1 : n->size;
raxNode **cldptr = raxNodeLastChildPtr(n) - (numcld-1);
......
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