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

Merge 6.2 RC3

parents ec2d1807 95338f9c
......@@ -721,20 +721,26 @@ zskiplistNode *zslLastInLexRange(zskiplist *zsl, zlexrangespec *range) {
* Ziplist-backed sorted set API
*----------------------------------------------------------------------------*/
double zzlStrtod(unsigned char *vstr, unsigned int vlen) {
char buf[128];
if (vlen > sizeof(buf))
vlen = sizeof(buf);
memcpy(buf,vstr,vlen);
buf[vlen] = '\0';
return strtod(buf,NULL);
}
double zzlGetScore(unsigned char *sptr) {
unsigned char *vstr;
unsigned int vlen;
long long vlong;
char buf[128];
double score;
serverAssert(sptr != NULL);
serverAssert(ziplistGet(sptr,&vstr,&vlen,&vlong));
if (vstr) {
memcpy(buf,vstr,vlen);
buf[vlen] = '\0';
score = strtod(buf,NULL);
score = zzlStrtod(vstr,vlen);
} else {
score = vlong;
}
......@@ -1653,6 +1659,48 @@ int zsetZiplistValidateIntegrity(unsigned char *zl, size_t size, int deep) {
return ret;
}
/* Create a new sds string from the ziplist entry. */
sds zsetSdsFromZiplistEntry(ziplistEntry *e) {
return e->sval ? sdsnewlen(e->sval, e->slen) : sdsfromlonglong(e->lval);
}
/* Reply with bulk string from the ziplist entry. */
void zsetReplyFromZiplistEntry(client *c, ziplistEntry *e) {
if (e->sval)
addReplyBulkCBuffer(c, e->sval, e->slen);
else
addReplyBulkLongLong(c, e->lval);
}
/* Return random element from a non empty zset.
* 'key' and 'val' will be set to hold the element.
* The memory in `key` is not to be freed or modified by the caller.
* 'score' can be NULL in which case it's not extracted. */
void zsetTypeRandomElement(robj *zsetobj, unsigned long zsetsize, ziplistEntry *key, double *score) {
if (zsetobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zsetobj->ptr;
dictEntry *de = dictGetFairRandomKey(zs->dict);
sds s = dictGetKey(de);
key->sval = (unsigned char*)s;
key->slen = sdslen(s);
if (score)
*score = *(double*)dictGetVal(de);
} else if (zsetobj->encoding == OBJ_ENCODING_ZIPLIST) {
ziplistEntry val;
ziplistRandomPair(zsetobj->ptr, zsetsize, key, &val);
if (score) {
if (val.sval) {
*score = zzlStrtod(val.sval,val.slen);
} else {
*score = (double)val.lval;
}
}
} else {
serverPanic("Unknown zset encoding");
}
}
/*-----------------------------------------------------------------------------
* Sorted set commands
*----------------------------------------------------------------------------*/
......@@ -2543,7 +2591,9 @@ void zunionInterDiffGenericCommand(client *c, robj *dstkey, int numkeysIndex, in
/* read keys to be used for input */
src = zcalloc(sizeof(zsetopsrc) * setnum);
for (i = 0, j = numkeysIndex+1; i < setnum; i++, j++) {
robj *obj = lookupKeyWrite(c->db,c->argv[j]);
robj *obj = dstkey ?
lookupKeyWrite(c->db,c->argv[j]) :
lookupKeyRead(c->db,c->argv[j]);
if (obj != NULL) {
if (obj->type != OBJ_ZSET && obj->type != OBJ_SET) {
zfree(src);
......@@ -2749,6 +2799,9 @@ void zunionInterDiffGenericCommand(client *c, robj *dstkey, int numkeysIndex, in
unsigned long length = dstzset->zsl->length;
zskiplist *zsl = dstzset->zsl;
zskiplistNode *zn = zsl->header->level[0].forward;
/* In case of WITHSCORES, respond with a single array in RESP2, and
* nested arrays in RESP3. We can't use a map response type since the
* client library needs to know to respect the order. */
if (withscores && c->resp == 2)
addReplyArrayLen(c, length*2);
else
......@@ -2866,6 +2919,9 @@ static void zrangeResultEmitLongLongToClient(zrange_result_handler *handler,
static void zrangeResultFinalizeClient(zrange_result_handler *handler,
size_t result_count)
{
/* In case of WITHSCORES, respond with a single array in RESP2, and
* nested arrays in RESP3. We can't use a map response type since the
* client library needs to know to respect the order. */
if (handler->withscores && (handler->client->resp == 2)) {
result_count *= 2;
}
......@@ -3071,8 +3127,8 @@ void zrevrangeCommand(client *c) {
/* This command implements ZRANGEBYSCORE, ZREVRANGEBYSCORE. */
void genericZrangebyscoreCommand(zrange_result_handler *handler,
zrangespec *range, robj *zobj, int withscores, long offset,
long limit, int reverse) {
zrangespec *range, robj *zobj, long offset, long limit,
int reverse) {
client *c = handler->client;
unsigned long rangelen = 0;
......@@ -3172,8 +3228,7 @@ void genericZrangebyscoreCommand(zrange_result_handler *handler,
}
rangelen++;
handler->emitResultFromCBuffer(handler, ln->ele, sdslen(ln->ele),
((withscores) ? ln->score : ln->score));
handler->emitResultFromCBuffer(handler, ln->ele, sdslen(ln->ele), ln->score);
/* Move to next node */
if (reverse) {
......@@ -3605,11 +3660,16 @@ void zrangeGenericCommand(zrange_result_handler *handler, int argc_start, int st
}
/* Step 3: Lookup the key and get the range. */
if (((zobj = lookupKeyReadOrReply(c, key, shared.emptyarray)) == NULL)
|| checkType(c, zobj, OBJ_ZSET)) {
zobj = handler->dstkey ?
lookupKeyWrite(c->db,key) :
lookupKeyRead(c->db,key);
if (zobj == NULL) {
addReply(c,shared.emptyarray);
goto cleanup;
}
if (checkType(c,zobj,OBJ_ZSET)) goto cleanup;
/* Step 4: Pass this to the command-specific handler. */
switch (rangetype) {
case ZRANGE_AUTO:
......@@ -3619,8 +3679,8 @@ void zrangeGenericCommand(zrange_result_handler *handler, int argc_start, int st
break;
case ZRANGE_SCORE:
genericZrangebyscoreCommand(handler, &range, zobj, opt_withscores || store,
opt_offset, opt_limit, direction == ZRANGE_DIRECTION_REVERSE);
genericZrangebyscoreCommand(handler, &range, zobj, opt_offset,
opt_limit, direction == ZRANGE_DIRECTION_REVERSE);
break;
case ZRANGE_LEX:
......@@ -3895,3 +3955,216 @@ void bzpopminCommand(client *c) {
void bzpopmaxCommand(client *c) {
blockingGenericZpopCommand(c,ZSET_MAX);
}
/* How many times bigger should be the zset compared to the requested size
* for us to not use the "remove elements" strategy? Read later in the
* implementation for more info. */
#define ZRANDMEMBER_SUB_STRATEGY_MUL 3
void zrandmemberWithCountCommand(client *c, long l, int withscores) {
unsigned long count, size;
int uniq = 1;
robj *zsetobj;
if ((zsetobj = lookupKeyReadOrReply(c, c->argv[1], shared.null[c->resp]))
== NULL || checkType(c, zsetobj, OBJ_ZSET)) return;
size = zsetLength(zsetobj);
if(l >= 0) {
count = (unsigned long) l;
} else {
count = -l;
uniq = 0;
}
/* If count is zero, serve it ASAP to avoid special cases later. */
if (count == 0) {
addReply(c,shared.emptyarray);
return;
}
/* CASE 1: The count was negative, so the extraction method is just:
* "return N random elements" sampling the whole set every time.
* This case is trivial and can be served without auxiliary data
* structures. This case is the only one that also needs to return the
* elements in random order. */
if (!uniq || count == 1) {
if (withscores && c->resp == 2)
addReplyArrayLen(c, count*2);
else
addReplyArrayLen(c, count);
if (zsetobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zsetobj->ptr;
while (count--) {
dictEntry *de = dictGetFairRandomKey(zs->dict);
sds key = dictGetKey(de);
if (withscores && c->resp > 2)
addReplyArrayLen(c,2);
addReplyBulkCBuffer(c, key, sdslen(key));
if (withscores)
addReplyDouble(c, dictGetDoubleVal(de));
}
} else if (zsetobj->encoding == OBJ_ENCODING_ZIPLIST) {
ziplistEntry *keys, *vals = NULL;
keys = zmalloc(sizeof(ziplistEntry)*count);
if (withscores)
vals = zmalloc(sizeof(ziplistEntry)*count);
ziplistRandomPairs(zsetobj->ptr, count, keys, vals);
for (unsigned long i = 0; i < count; i++) {
if (withscores && c->resp > 2)
addReplyArrayLen(c,2);
if (keys[i].sval)
addReplyBulkCBuffer(c, keys[i].sval, keys[i].slen);
else
addReplyBulkLongLong(c, keys[i].lval);
if (withscores) {
if (vals[i].sval) {
addReplyDouble(c, zzlStrtod(vals[i].sval,vals[i].slen));
} else
addReplyDouble(c, vals[i].lval);
}
}
zfree(keys);
zfree(vals);
}
return;
}
zsetopsrc src;
zsetopval zval;
src.subject = zsetobj;
src.type = zsetobj->type;
src.encoding = zsetobj->encoding;
zuiInitIterator(&src);
memset(&zval, 0, sizeof(zval));
/* Initiate reply count, RESP3 responds with nested array, RESP2 with flat one. */
long reply_size = count < size ? count : size;
if (withscores && c->resp == 2)
addReplyArrayLen(c, reply_size*2);
else
addReplyArrayLen(c, reply_size);
/* CASE 2:
* The number of requested elements is greater than the number of
* elements inside the zset: simply return the whole zset. */
if (count >= size) {
while (zuiNext(&src, &zval)) {
if (withscores && c->resp > 2)
addReplyArrayLen(c,2);
addReplyBulkSds(c, zuiNewSdsFromValue(&zval));
if (withscores)
addReplyDouble(c, zval.score);
}
return;
}
/* CASE 3:
* The number of elements inside the zset is not greater than
* ZRANDMEMBER_SUB_STRATEGY_MUL times the number of requested elements.
* In this case we create a dict from scratch with all the elements, and
* subtract random elements to reach the requested number of elements.
*
* This is done because if the number of requested elements is just
* a bit less than the number of elements in the set, the natural approach
* used into CASE 4 is highly inefficient. */
if (count*ZRANDMEMBER_SUB_STRATEGY_MUL > size) {
dict *d = dictCreate(&sdsReplyDictType, NULL);
/* Add all the elements into the temporary dictionary. */
while (zuiNext(&src, &zval)) {
sds key = zuiNewSdsFromValue(&zval);
dictEntry *de = dictAddRaw(d, key, NULL);
serverAssert(de);
if (withscores)
dictSetDoubleVal(de, zval.score);
}
serverAssert(dictSize(d) == size);
/* Remove random elements to reach the right count. */
while (size > count) {
dictEntry *de;
de = dictGetRandomKey(d);
dictUnlink(d,dictGetKey(de));
sdsfree(dictGetKey(de));
dictFreeUnlinkedEntry(d,de);
size--;
}
/* Reply with what's in the dict and release memory */
dictIterator *di;
dictEntry *de;
di = dictGetIterator(d);
while ((de = dictNext(di)) != NULL) {
if (withscores && c->resp > 2)
addReplyArrayLen(c,2);
addReplyBulkSds(c, dictGetKey(de));
if (withscores)
addReplyDouble(c, dictGetDoubleVal(de));
}
dictReleaseIterator(di);
dictRelease(d);
}
/* CASE 4: We have a big zset compared to the requested number of elements.
* In this case we can simply get random elements from the zset and add
* to the temporary set, trying to eventually get enough unique elements
* to reach the specified count. */
else {
unsigned long added = 0;
dict *d = dictCreate(&hashDictType, NULL);
while (added < count) {
ziplistEntry key;
double score;
zsetTypeRandomElement(zsetobj, size, &key, withscores ? &score: NULL);
/* Try to add the object to the dictionary. If it already exists
* free it, otherwise increment the number of objects we have
* in the result dictionary. */
sds skey = zsetSdsFromZiplistEntry(&key);
if (dictAdd(d,skey,NULL) != DICT_OK) {
sdsfree(skey);
continue;
}
added++;
if (withscores && c->resp > 2)
addReplyArrayLen(c,2);
zsetReplyFromZiplistEntry(c, &key);
if (withscores)
addReplyDouble(c, score);
}
/* Release memory */
dictRelease(d);
}
}
/* ZRANDMEMBER [<count> WITHSCORES] */
void zrandmemberCommand(client *c) {
long l;
int withscores = 0;
robj *zset;
ziplistEntry ele;
if (c->argc >= 3) {
if (getLongFromObjectOrReply(c,c->argv[2],&l,NULL) != C_OK) return;
if (c->argc > 4 || (c->argc == 4 && strcasecmp(c->argv[3]->ptr,"withscores"))) {
addReplyErrorObject(c,shared.syntaxerr);
return;
} else if (c->argc == 4)
withscores = 1;
zrandmemberWithCountCommand(c, l, withscores);
return;
}
/* Handle variant without <count> argument. Reply with simple bulk string */
if ((zset = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp]))== NULL ||
checkType(c,zset,OBJ_ZSET)) {
return;
}
zsetTypeRandomElement(zset, zsetLength(zset), &ele,NULL);
zsetReplyFromZiplistEntry(c,&ele);
}
......@@ -749,7 +749,7 @@ sds getAbsolutePath(char *filename) {
* Gets the proper timezone in a more portable fashion
* i.e timezone variables are linux specific.
*/
unsigned long getTimeZone(void) {
long getTimeZone(void) {
#if defined(__linux__) || defined(__sun)
return timezone;
#else
......@@ -758,7 +758,7 @@ unsigned long getTimeZone(void) {
gettimeofday(&tv, &tz);
return tz.tz_minuteswest * 60UL;
return tz.tz_minuteswest * 60L;
#endif
}
......
......@@ -60,7 +60,7 @@ int string2d(const char *s, size_t slen, double *dp);
int d2string(char *buf, size_t len, double value);
int ld2string(char *buf, size_t len, long double value, ld2string_mode mode);
sds getAbsolutePath(char *filename);
unsigned long getTimeZone(void);
long getTimeZone(void);
int pathIsBaseName(char *path);
#ifdef REDIS_TEST
......
#define REDIS_VERSION "6.1.241"
#define REDIS_VERSION_NUM 0x000601f1
#define REDIS_VERSION "6.1.242"
#define REDIS_VERSION_NUM 0x000601f2
......@@ -1498,6 +1498,89 @@ int ziplistValidateIntegrity(unsigned char *zl, size_t size, int deep,
return 1;
}
/* Randomly select a pair of key and value.
* total_count is a pre-computed length/2 of the ziplist (to avoid calls to ziplistLen)
* 'key' and 'val' are used to store the result key value pair.
* 'val' can be NULL if the value is not needed. */
void ziplistRandomPair(unsigned char *zl, unsigned long total_count, ziplistEntry *key, ziplistEntry *val) {
int ret;
unsigned char *p;
/* Avoid div by zero on corrupt ziplist */
assert(total_count);
/* Generate even numbers, because ziplist saved K-V pair */
int r = (rand() % total_count) * 2;
p = ziplistIndex(zl, r);
ret = ziplistGet(p, &key->sval, &key->slen, &key->lval);
assert(ret != 0);
if (!val)
return;
p = ziplistNext(zl, p);
ret = ziplistGet(p, &val->sval, &val->slen, &val->lval);
assert(ret != 0);
}
/* int compare for qsort */
int intCompare(const void *a, const void *b) {
return (*(int *) a - *(int *) b);
}
/* Helper method to store a string into from val or lval into dest */
static inline void ziplistSaveValue(unsigned char *val, unsigned int len, long long lval, ziplistEntry *dest) {
dest->sval = val;
dest->slen = len;
dest->lval = lval;
}
/* Randomly select unique count of key value pairs and store into 'keys' and
* 'vals' args. The order of the picked entries is random.
* The 'vals' arg can be NULL in which case we skip these. */
void ziplistRandomPairs(unsigned char *zl, int count, ziplistEntry *keys, ziplistEntry *vals) {
unsigned char *p, *key, *value;
unsigned int klen, vlen;
long long klval, vlval;
typedef struct {
int index;
int order;
} rand_pick;
rand_pick *picks = zmalloc(sizeof(rand_pick)*count);
unsigned long total_size = ziplistLen(zl)/2;
/* Avoid div by zero on corrupt ziplist */
assert(total_size);
/* create a pool of random indexes (some may be duplicate). */
for (int i = 0; i < count; i++) {
picks[i].index = (rand() % total_size) * 2; /* Generate even indexes */
/* keep track of the order we picked them */
picks[i].order = i;
}
/* sort by indexes. */
qsort(picks, count, sizeof(rand_pick), intCompare);
/* fetch the elements form the ziplist into a output array respecting the original order. */
int zipindex = 0, pickindex = 0;
p = ziplistIndex(zl, 0);
while (ziplistGet(p, &key, &klen, &klval) && pickindex < count) {
p = ziplistNext(zl, p);
ziplistGet(p, &value, &vlen, &vlval);
while (pickindex < count && zipindex == picks[pickindex].index) {
int storeorder = picks[pickindex].order;
ziplistSaveValue(key, klen, klval, &keys[storeorder]);
if (vals)
ziplistSaveValue(value, vlen, vlval, &vals[storeorder]);
pickindex++;
}
zipindex += 2;
p = ziplistNext(zl, p);
}
zfree(picks);
}
#ifdef REDIS_TEST
#include <sys/time.h>
#include "adlist.h"
......
......@@ -34,6 +34,15 @@
#define ZIPLIST_HEAD 0
#define ZIPLIST_TAIL 1
/* Each entry in the ziplist is either a string or an integer. */
typedef struct {
/* When string is used, it is provided with the length (slen). */
unsigned char *sval;
unsigned int slen;
/* When integer is used, 'sval' is NULL, and lval holds the value. */
long long lval;
} ziplistEntry;
unsigned char *ziplistNew(void);
unsigned char *ziplistMerge(unsigned char **first, unsigned char **second);
unsigned char *ziplistPush(unsigned char *zl, unsigned char *s, unsigned int slen, int where);
......@@ -52,6 +61,8 @@ void ziplistRepr(unsigned char *zl);
typedef int (*ziplistValidateEntryCB)(unsigned char* p, void* userdata);
int ziplistValidateIntegrity(unsigned char *zl, size_t size, int deep,
ziplistValidateEntryCB entry_cb, void *cb_userdata);
void ziplistRandomPair(unsigned char *zl, unsigned long total_count, ziplistEntry *key, ziplistEntry *val);
void ziplistRandomPairs(unsigned char *zl, int count, ziplistEntry *keys, ziplistEntry *vals);
#ifdef REDIS_TEST
int ziplistTest(int argc, char *argv[]);
......
# Optimize CLUSTER NODES command by generating all nodes slot topology firstly
source "../tests/includes/init-tests.tcl"
proc cluster_allocate_with_continuous_slots {n} {
set slot 16383
set avg [expr ($slot+1) / $n]
while {$slot >= 0} {
set node [expr $slot/$avg >= $n ? $n-1 : $slot/$avg]
lappend slots_$node $slot
incr slot -1
}
for {set j 0} {$j < $n} {incr j} {
R $j cluster addslots {*}[set slots_${j}]
}
}
proc cluster_create_with_continuous_slots {masters slaves} {
cluster_allocate_with_continuous_slots $masters
if {$slaves} {
cluster_allocate_slaves $masters $slaves
}
assert_cluster_state ok
}
test "Create a 2 nodes cluster" {
cluster_create_with_continuous_slots 2 2
}
test "Cluster should start ok" {
assert_cluster_state ok
}
set master1 [Rn 0]
set master2 [Rn 1]
test "Continuous slots distribution" {
assert_match "* 0-8191*" [$master1 CLUSTER NODES]
assert_match "* 8192-16383*" [$master2 CLUSTER NODES]
$master1 CLUSTER DELSLOTS 4096
assert_match "* 0-4095 4097-8191*" [$master1 CLUSTER NODES]
$master2 CLUSTER DELSLOTS 12288
assert_match "* 8192-12287 12289-16383*" [$master2 CLUSTER NODES]
}
test "Discontinuous slots distribution" {
# Remove middle slots
$master1 CLUSTER DELSLOTS 4092 4094
assert_match "* 0-4091 4093 4095 4097-8191*" [$master1 CLUSTER NODES]
$master2 CLUSTER DELSLOTS 12284 12286
assert_match "* 8192-12283 12285 12287 12289-16383*" [$master2 CLUSTER NODES]
# Remove head slots
$master1 CLUSTER DELSLOTS 0 2
assert_match "* 1 3-4091 4093 4095 4097-8191*" [$master1 CLUSTER NODES]
# Remove tail slots
$master2 CLUSTER DELSLOTS 16380 16382 16383
assert_match "* 8192-12283 12285 12287 12289-16379 16381*" [$master2 CLUSTER NODES]
}
......@@ -24,9 +24,11 @@ set ::simulate_error 0
set ::failed 0
set ::sentinel_instances {}
set ::redis_instances {}
set ::global_config {}
set ::sentinel_base_port 20000
set ::redis_base_port 30000
set ::redis_port_count 1024
set ::host "127.0.0.1"
set ::pids {} ; # We kill everything at exit
set ::dirs {} ; # We remove all the temp dirs at exit
set ::run_matching {} ; # If non empty, only tests matching pattern are run.
......@@ -58,10 +60,9 @@ proc exec_instance {type dirname cfgfile} {
}
# Spawn a redis or sentinel instance, depending on 'type'.
proc spawn_instance {type base_port count {conf {}}} {
proc spawn_instance {type base_port count {conf {}} {base_conf_file ""}} {
for {set j 0} {$j < $count} {incr j} {
set port [find_available_port $base_port $::redis_port_count]
# Create a directory for this instance.
set dirname "${type}_${j}"
lappend ::dirs $dirname
......@@ -70,7 +71,13 @@ proc spawn_instance {type base_port count {conf {}}} {
# Write the instance config file.
set cfgfile [file join $dirname $type.conf]
set cfg [open $cfgfile w]
if {$base_conf_file ne ""} {
file copy -- $base_conf_file $cfgfile
set cfg [open $cfgfile a+]
} else {
set cfg [open $cfgfile w]
}
if {$::tls} {
puts $cfg "tls-port $port"
puts $cfg "tls-replication yes"
......@@ -92,6 +99,9 @@ proc spawn_instance {type base_port count {conf {}}} {
foreach directive $conf {
puts $cfg $directive
}
dict for {name val} $::global_config {
puts $cfg "$name $val"
}
close $cfg
# Finally exec it and remember the pid for later cleanup.
......@@ -119,18 +129,18 @@ proc spawn_instance {type base_port count {conf {}}} {
}
# Check availability finally
if {[server_is_up 127.0.0.1 $port 100] == 0} {
if {[server_is_up $::host $port 100] == 0} {
set logfile [file join $dirname log.txt]
puts [exec tail $logfile]
abort_sentinel_test "Problems starting $type #$j: ping timeout, maybe server start failed, check $logfile"
}
# Push the instance into the right list
set link [redis 127.0.0.1 $port 0 $::tls]
set link [redis $::host $port 0 $::tls]
$link reconnect 1
lappend ::${type}_instances [list \
pid $pid \
host 127.0.0.1 \
host $::host \
port $port \
link $link \
]
......@@ -232,6 +242,9 @@ proc parse_options {} {
set ::simulate_error 1
} elseif {$opt eq {--valgrind}} {
set ::valgrind 1
} elseif {$opt eq {--host}} {
incr j
set ::host ${val}
} elseif {$opt eq {--tls}} {
package require tls 1.6
::tls::init \
......@@ -239,6 +252,10 @@ proc parse_options {} {
-certfile "$::tlsdir/client.crt" \
-keyfile "$::tlsdir/client.key"
set ::tls 1
} elseif {$opt eq {--config}} {
set val2 [lindex $::argv [expr $j+2]]
dict set ::global_config $val $val2
incr j 2
} elseif {$opt eq "--help"} {
puts "--single <pattern> Only runs tests specified by pattern."
puts "--dont-clean Keep log files on exit."
......@@ -246,6 +263,8 @@ proc parse_options {} {
puts "--fail Simulate a test failure."
puts "--valgrind Run with valgrind."
puts "--tls Run tests in TLS mode."
puts "--host <host> Use hostname instead of 127.0.0.1."
puts "--config <k> <v> Extra config argument(s)."
puts "--help Shows this help."
exit 0
} else {
......@@ -391,6 +410,11 @@ proc check_leaks instance_types {
# Execute all the units inside the 'tests' directory.
proc run_tests {} {
set sentinel_fd_leaks_file "sentinel_fd_leaks"
if { [file exists $sentinel_fd_leaks_file] } {
file delete $sentinel_fd_leaks_file
}
set tests [lsort [glob ../tests/*]]
foreach test $tests {
if {$::run_matching ne {} && [string match $::run_matching $test] == 0} {
......@@ -405,7 +429,15 @@ proc run_tests {} {
# Print a message and exists with 0 / 1 according to zero or more failures.
proc end_tests {} {
if {$::failed == 0} {
set sentinel_fd_leaks_file "sentinel_fd_leaks"
if { [file exists $sentinel_fd_leaks_file] } {
# temporarily disabling this error from failing the tests until leaks are fixed.
#puts [colorstr red "WARNING: sentinel test(s) failed, there are leaked fds in sentinel:"]
#puts [exec cat $sentinel_fd_leaks_file]
#exit 1
}
if {$::failed == 0 } {
puts "GOOD! No errors."
exit 0
} else {
......
......@@ -272,4 +272,15 @@ tags {"aof"} {
}
}
}
start_server {overrides {appendonly {yes} appendfilename {appendonly.aof}}} {
test {GETEX should not append to AOF} {
set aof [file join [lindex [r config get dir] 1] appendonly.aof]
r set foo bar
set before [file size $aof]
r getex foo
set after [file size $aof]
assert_equal $before $after
}
}
}
......@@ -507,5 +507,16 @@ test {corrupt payload: fuzzer findings - valgrind invalid read} {
}
}
test {corrupt payload: fuzzer findings - HRANDFIELD on bad ziplist} {
start_server [list overrides [list loglevel verbose use-exit-on-panic yes crash-memcheck-enabled no] ] {
r config set sanitize-dump-payload yes
r debug set-skip-checksum-validation 1
r RESTORE _int 0 "\x04\xC0\x01\x09\x00\xF6\x8A\xB6\x7A\x85\x87\x72\x4D"
catch {r HRANDFIELD _int}
assert_equal [count_log_message 0 "crashed by signal"] 0
assert_equal [count_log_message 0 "ASSERTION FAILED"] 1
}
}
} ;# tags
start_server {tags {"failover"}} {
start_server {} {
start_server {} {
set node_0 [srv 0 client]
set node_0_host [srv 0 host]
set node_0_port [srv 0 port]
set node_0_pid [srv 0 pid]
set node_1 [srv -1 client]
set node_1_host [srv -1 host]
set node_1_port [srv -1 port]
set node_1_pid [srv -1 pid]
set node_2 [srv -2 client]
set node_2_host [srv -2 host]
set node_2_port [srv -2 port]
set node_2_pid [srv -2 pid]
proc assert_digests_match {n1 n2 n3} {
assert_equal [$n1 debug digest] [$n2 debug digest]
assert_equal [$n2 debug digest] [$n3 debug digest]
}
test {failover command fails without connected replica} {
catch { $node_0 failover to $node_1_host $node_1_port } err
if {! [string match "ERR*" $err]} {
fail "failover command succeeded when replica not connected"
}
}
test {setup replication for following tests} {
$node_1 replicaof $node_0_host $node_0_port
$node_2 replicaof $node_0_host $node_0_port
wait_for_sync $node_1
wait_for_sync $node_2
}
test {failover command fails with invalid host} {
catch { $node_0 failover to invalidhost $node_1_port } err
assert_match "ERR*" $err
}
test {failover command fails with invalid port} {
catch { $node_0 failover to $node_1_host invalidport } err
assert_match "ERR*" $err
}
test {failover command fails with just force and timeout} {
catch { $node_0 FAILOVER FORCE TIMEOUT 100} err
assert_match "ERR*" $err
}
test {failover command fails when sent to a replica} {
catch { $node_1 failover to $node_1_host $node_1_port } err
assert_match "ERR*" $err
}
test {failover command fails with force without timeout} {
catch { $node_0 failover to $node_1_host $node_1_port FORCE } err
assert_match "ERR*" $err
}
test {failover command to specific replica works} {
set initial_psyncs [s -1 sync_partial_ok]
set initial_syncs [s -1 sync_full]
# Generate a delta between primary and replica
set load_handler [start_write_load $node_0_host $node_0_port 5]
exec kill -SIGSTOP [srv -1 pid]
wait_for_condition 50 100 {
[s 0 total_commands_processed] > 100
} else {
fail "Node 0 did not accept writes"
}
exec kill -SIGCONT [srv -1 pid]
# Execute the failover
$node_0 failover to $node_1_host $node_1_port
# Wait for failover to end
wait_for_condition 50 100 {
[s 0 master_failover_state] == "no-failover"
} else {
fail "Failover from node 0 to node 1 did not finish"
}
stop_write_load $load_handler
$node_2 replicaof $node_1_host $node_1_port
wait_for_sync $node_0
wait_for_sync $node_2
assert_match *slave* [$node_0 role]
assert_match *master* [$node_1 role]
assert_match *slave* [$node_2 role]
# We should accept psyncs from both nodes
assert_equal [expr [s -1 sync_partial_ok] - $initial_psyncs] 2
assert_equal [expr [s -1 sync_full] - $initial_psyncs] 0
assert_digests_match $node_0 $node_1 $node_2
}
test {failover command to any replica works} {
set initial_psyncs [s -2 sync_partial_ok]
set initial_syncs [s -2 sync_full]
wait_for_ofs_sync $node_1 $node_2
# We stop node 0 to and make sure node 2 is selected
exec kill -SIGSTOP $node_0_pid
$node_1 set CASE 1
$node_1 FAILOVER
# Wait for failover to end
wait_for_condition 50 100 {
[s -1 master_failover_state] == "no-failover"
} else {
fail "Failover from node 1 to node 2 did not finish"
}
exec kill -SIGCONT $node_0_pid
$node_0 replicaof $node_2_host $node_2_port
wait_for_sync $node_0
wait_for_sync $node_1
assert_match *slave* [$node_0 role]
assert_match *slave* [$node_1 role]
assert_match *master* [$node_2 role]
# We should accept Psyncs from both nodes
assert_equal [expr [s -2 sync_partial_ok] - $initial_psyncs] 2
assert_equal [expr [s -1 sync_full] - $initial_psyncs] 0
assert_digests_match $node_0 $node_1 $node_2
}
test {failover to a replica with force works} {
set initial_psyncs [s 0 sync_partial_ok]
set initial_syncs [s 0 sync_full]
exec kill -SIGSTOP $node_0_pid
# node 0 will never acknowledge this write
$node_2 set case 2
$node_2 failover to $node_0_host $node_0_port TIMEOUT 100 FORCE
# Wait for node 0 to give up on sync attempt and start failover
wait_for_condition 50 100 {
[s -2 master_failover_state] == "failover-in-progress"
} else {
fail "Failover from node 2 to node 0 did not timeout"
}
# Quick check that everyone is a replica, we never want a
# state where there are two masters.
assert_match *slave* [$node_1 role]
assert_match *slave* [$node_2 role]
exec kill -SIGCONT $node_0_pid
# Wait for failover to end
wait_for_condition 50 100 {
[s -2 master_failover_state] == "no-failover"
} else {
fail "Failover from node 2 to node 0 did not finish"
}
$node_1 replicaof $node_0_host $node_0_port
wait_for_sync $node_1
wait_for_sync $node_2
assert_match *master* [$node_0 role]
assert_match *slave* [$node_1 role]
assert_match *slave* [$node_2 role]
assert_equal [count_log_message -2 "time out exceeded, failing over."] 1
# We should accept both psyncs, although this is the condition we might not
# since we didn't catch up.
assert_equal [expr [s 0 sync_partial_ok] - $initial_psyncs] 2
assert_equal [expr [s 0 sync_full] - $initial_syncs] 0
assert_digests_match $node_0 $node_1 $node_2
}
test {failover with timeout aborts if replica never catches up} {
set initial_psyncs [s 0 sync_partial_ok]
set initial_syncs [s 0 sync_full]
# Stop replica so it never catches up
exec kill -SIGSTOP [srv -1 pid]
$node_0 SET CASE 1
$node_0 failover to [srv -1 host] [srv -1 port] TIMEOUT 500
# Wait for failover to end
wait_for_condition 50 20 {
[s 0 master_failover_state] == "no-failover"
} else {
fail "Failover from node_0 to replica did not finish"
}
exec kill -SIGCONT [srv -1 pid]
# We need to make sure the nodes actually sync back up
wait_for_ofs_sync $node_0 $node_1
wait_for_ofs_sync $node_0 $node_2
assert_match *master* [$node_0 role]
assert_match *slave* [$node_1 role]
assert_match *slave* [$node_2 role]
# Since we never caught up, there should be no syncs
assert_equal [expr [s 0 sync_partial_ok] - $initial_psyncs] 0
assert_equal [expr [s 0 sync_full] - $initial_syncs] 0
assert_digests_match $node_0 $node_1 $node_2
}
test {failovers can be aborted} {
set initial_psyncs [s 0 sync_partial_ok]
set initial_syncs [s 0 sync_full]
# Stop replica so it never catches up
exec kill -SIGSTOP [srv -1 pid]
$node_0 SET CASE 2
$node_0 failover to [srv -1 host] [srv -1 port] TIMEOUT 60000
assert_match [s 0 master_failover_state] "waiting-for-sync"
# Sanity check that read commands are still accepted
$node_0 GET CASE
$node_0 failover abort
assert_match [s 0 master_failover_state] "no-failover"
exec kill -SIGCONT [srv -1 pid]
# Just make sure everything is still synced
wait_for_ofs_sync $node_0 $node_1
wait_for_ofs_sync $node_0 $node_2
assert_match *master* [$node_0 role]
assert_match *slave* [$node_1 role]
assert_match *slave* [$node_2 role]
# Since we never caught up, there should be no syncs
assert_equal [expr [s 0 sync_partial_ok] - $initial_psyncs] 0
assert_equal [expr [s 0 sync_full] - $initial_syncs] 0
assert_digests_match $node_0 $node_1 $node_2
}
test {failover aborts if target rejects sync request} {
set initial_psyncs [s 0 sync_partial_ok]
set initial_syncs [s 0 sync_full]
# We block psync, so the failover will fail
$node_1 acl setuser default -psync
# We pause the target long enough to send a write command
# during the pause. This write will not be interrupted.
exec kill -SIGSTOP [srv -1 pid]
set rd [redis_deferring_client]
$rd SET FOO BAR
$node_0 failover to $node_1_host $node_1_port
exec kill -SIGCONT [srv -1 pid]
# Wait for failover to end
wait_for_condition 50 100 {
[s 0 master_failover_state] == "no-failover"
} else {
fail "Failover from node_0 to replica did not finish"
}
assert_equal [$rd read] "OK"
$rd close
# restore access to psync
$node_1 acl setuser default +psync
# We need to make sure the nodes actually sync back up
wait_for_sync $node_1
wait_for_sync $node_2
assert_match *master* [$node_0 role]
assert_match *slave* [$node_1 role]
assert_match *slave* [$node_2 role]
# We will cycle all of our replicas here and force a psync.
assert_equal [expr [s 0 sync_partial_ok] - $initial_psyncs] 2
assert_equal [expr [s 0 sync_full] - $initial_syncs] 0
assert_equal [count_log_message 0 "Failover target rejected psync request"] 1
assert_digests_match $node_0 $node_1 $node_2
}
}
}
}
tags {"rdb"} {
set server_path [tmpdir "server.rdb-encoding-test"]
# Copy RDB with different encodings in server path
......@@ -289,3 +291,5 @@ start_server {overrides {save ""}} {
}
}
} ;# system_name
} ;# tags
......@@ -5,7 +5,7 @@ proc cmdstat {cmd} {
return [cmdrstat $cmd r]
}
start_server {tags {"benchmark"}} {
start_server {tags {"benchmark network"}} {
start_server {} {
set master_host [srv 0 host]
set master_port [srv 0 port]
......
start_server {tags {"repl"}} {
start_server {tags {"repl network"}} {
start_server {} {
set master [srv -1 client]
......
......@@ -5,7 +5,7 @@ proc log_file_matches {log pattern} {
string match $pattern $content
}
start_server {tags {"repl"}} {
start_server {tags {"repl network"}} {
set slave [srv 0 client]
set slave_host [srv 0 host]
set slave_port [srv 0 port]
......
......@@ -19,6 +19,7 @@ TEST_MODULES = \
misc.so \
hooks.so \
blockonkeys.so \
blockonbackground.so \
scan.so \
datatype.so \
auth.so \
......@@ -27,7 +28,8 @@ TEST_MODULES = \
getkeys.so \
test_lazyfree.so \
timer.so \
defragtest.so
defragtest.so \
stream.so
.PHONY: all
......
#define REDISMODULE_EXPERIMENTAL_API
#include "redismodule.h"
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <time.h>
#include "assert.h"
#define UNUSED(x) (void)(x)
/* Reply callback for blocking command BLOCK.DEBUG */
int HelloBlock_Reply(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
UNUSED(argv);
UNUSED(argc);
int *myint = RedisModule_GetBlockedClientPrivateData(ctx);
return RedisModule_ReplyWithLongLong(ctx,*myint);
}
/* Timeout callback for blocking command BLOCK.DEBUG */
int HelloBlock_Timeout(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
UNUSED(argv);
UNUSED(argc);
RedisModuleBlockedClient *bc = RedisModule_GetBlockedClientHandle(ctx);
assert(RedisModule_BlockedClientMeasureTimeEnd(bc)==REDISMODULE_OK);
return RedisModule_ReplyWithSimpleString(ctx,"Request timedout");
}
/* Private data freeing callback for BLOCK.DEBUG command. */
void HelloBlock_FreeData(RedisModuleCtx *ctx, void *privdata) {
UNUSED(ctx);
RedisModule_Free(privdata);
}
/* The thread entry point that actually executes the blocking part
* of the command BLOCK.DEBUG. */
void *BlockDebug_ThreadMain(void *arg) {
void **targ = arg;
RedisModuleBlockedClient *bc = targ[0];
long long delay = (unsigned long)targ[1];
long long enable_time_track = (unsigned long)targ[2];
if (enable_time_track)
assert(RedisModule_BlockedClientMeasureTimeStart(bc)==REDISMODULE_OK);
RedisModule_Free(targ);
struct timespec ts;
ts.tv_sec = delay / 1000;
ts.tv_nsec = (delay % 1000) * 1000000;
nanosleep(&ts, NULL);
int *r = RedisModule_Alloc(sizeof(int));
*r = rand();
if (enable_time_track)
assert(RedisModule_BlockedClientMeasureTimeEnd(bc)==REDISMODULE_OK);
RedisModule_UnblockClient(bc,r);
return NULL;
}
/* The thread entry point that actually executes the blocking part
* of the command BLOCK.DEBUG. */
void *DoubleBlock_ThreadMain(void *arg) {
void **targ = arg;
RedisModuleBlockedClient *bc = targ[0];
long long delay = (unsigned long)targ[1];
assert(RedisModule_BlockedClientMeasureTimeStart(bc)==REDISMODULE_OK);
RedisModule_Free(targ);
struct timespec ts;
ts.tv_sec = delay / 1000;
ts.tv_nsec = (delay % 1000) * 1000000;
nanosleep(&ts, NULL);
int *r = RedisModule_Alloc(sizeof(int));
*r = rand();
RedisModule_BlockedClientMeasureTimeEnd(bc);
/* call again RedisModule_BlockedClientMeasureTimeStart() and
* RedisModule_BlockedClientMeasureTimeEnd and ensure that the
* total execution time is 2x the delay. */
assert(RedisModule_BlockedClientMeasureTimeStart(bc)==REDISMODULE_OK);
nanosleep(&ts, NULL);
RedisModule_BlockedClientMeasureTimeEnd(bc);
RedisModule_UnblockClient(bc,r);
return NULL;
}
void HelloBlock_Disconnected(RedisModuleCtx *ctx, RedisModuleBlockedClient *bc) {
RedisModule_Log(ctx,"warning","Blocked client %p disconnected!",
(void*)bc);
}
/* BLOCK.DEBUG <delay_ms> <timeout_ms> -- Block for <count> milliseconds, then reply with
* a random number. Timeout is the command timeout, so that you can test
* what happens when the delay is greater than the timeout. */
int HelloBlock_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc != 3) return RedisModule_WrongArity(ctx);
long long delay;
long long timeout;
if (RedisModule_StringToLongLong(argv[1],&delay) != REDISMODULE_OK) {
return RedisModule_ReplyWithError(ctx,"ERR invalid count");
}
if (RedisModule_StringToLongLong(argv[2],&timeout) != REDISMODULE_OK) {
return RedisModule_ReplyWithError(ctx,"ERR invalid count");
}
pthread_t tid;
RedisModuleBlockedClient *bc = RedisModule_BlockClient(ctx,HelloBlock_Reply,HelloBlock_Timeout,HelloBlock_FreeData,timeout);
/* Here we set a disconnection handler, however since this module will
* block in sleep() in a thread, there is not much we can do in the
* callback, so this is just to show you the API. */
RedisModule_SetDisconnectCallback(bc,HelloBlock_Disconnected);
/* Now that we setup a blocking client, we need to pass the control
* to the thread. However we need to pass arguments to the thread:
* the delay and a reference to the blocked client handle. */
void **targ = RedisModule_Alloc(sizeof(void*)*3);
targ[0] = bc;
targ[1] = (void*)(unsigned long) delay;
// pass 1 as flag to enable time tracking
targ[2] = (void*)(unsigned long) 1;
if (pthread_create(&tid,NULL,BlockDebug_ThreadMain,targ) != 0) {
RedisModule_AbortBlock(bc);
return RedisModule_ReplyWithError(ctx,"-ERR Can't start thread");
}
return REDISMODULE_OK;
}
/* BLOCK.DEBUG_NOTRACKING <delay_ms> <timeout_ms> -- Block for <count> milliseconds, then reply with
* a random number. Timeout is the command timeout, so that you can test
* what happens when the delay is greater than the timeout.
* this command does not track background time so the background time should no appear in stats*/
int HelloBlockNoTracking_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc != 3) return RedisModule_WrongArity(ctx);
long long delay;
long long timeout;
if (RedisModule_StringToLongLong(argv[1],&delay) != REDISMODULE_OK) {
return RedisModule_ReplyWithError(ctx,"ERR invalid count");
}
if (RedisModule_StringToLongLong(argv[2],&timeout) != REDISMODULE_OK) {
return RedisModule_ReplyWithError(ctx,"ERR invalid count");
}
pthread_t tid;
RedisModuleBlockedClient *bc = RedisModule_BlockClient(ctx,HelloBlock_Reply,HelloBlock_Timeout,HelloBlock_FreeData,timeout);
/* Here we set a disconnection handler, however since this module will
* block in sleep() in a thread, there is not much we can do in the
* callback, so this is just to show you the API. */
RedisModule_SetDisconnectCallback(bc,HelloBlock_Disconnected);
/* Now that we setup a blocking client, we need to pass the control
* to the thread. However we need to pass arguments to the thread:
* the delay and a reference to the blocked client handle. */
void **targ = RedisModule_Alloc(sizeof(void*)*3);
targ[0] = bc;
targ[1] = (void*)(unsigned long) delay;
// pass 0 as flag to enable time tracking
targ[2] = (void*)(unsigned long) 0;
if (pthread_create(&tid,NULL,BlockDebug_ThreadMain,targ) != 0) {
RedisModule_AbortBlock(bc);
return RedisModule_ReplyWithError(ctx,"-ERR Can't start thread");
}
return REDISMODULE_OK;
}
/* BLOCK.DOUBLE_DEBUG <delay_ms> -- Block for 2 x <count> milliseconds,
* then reply with a random number.
* This command is used to test multiple calls to RedisModule_BlockedClientMeasureTimeStart()
* and RedisModule_BlockedClientMeasureTimeEnd() within the same execution. */
int HelloDoubleBlock_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc != 2) return RedisModule_WrongArity(ctx);
long long delay;
long long timeout;
if (RedisModule_StringToLongLong(argv[1],&delay) != REDISMODULE_OK) {
return RedisModule_ReplyWithError(ctx,"ERR invalid count");
}
pthread_t tid;
RedisModuleBlockedClient *bc = RedisModule_BlockClient(ctx,HelloBlock_Reply,HelloBlock_Timeout,HelloBlock_FreeData,timeout);
/* Now that we setup a blocking client, we need to pass the control
* to the thread. However we need to pass arguments to the thread:
* the delay and a reference to the blocked client handle. */
void **targ = RedisModule_Alloc(sizeof(void*)*2);
targ[0] = bc;
targ[1] = (void*)(unsigned long) delay;
if (pthread_create(&tid,NULL,DoubleBlock_ThreadMain,targ) != 0) {
RedisModule_AbortBlock(bc);
return RedisModule_ReplyWithError(ctx,"-ERR Can't start thread");
}
return REDISMODULE_OK;
}
int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
UNUSED(argv);
UNUSED(argc);
if (RedisModule_Init(ctx,"block",1,REDISMODULE_APIVER_1)
== REDISMODULE_ERR) return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"block.debug",
HelloBlock_RedisCommand,"",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"block.double_debug",
HelloDoubleBlock_RedisCommand,"",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"block.debug_no_track",
HelloBlockNoTracking_RedisCommand,"",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
return REDISMODULE_OK;
}
......@@ -2,6 +2,7 @@
#include "redismodule.h"
#include <string.h>
#include <strings.h>
#include <assert.h>
#include <unistd.h>
......@@ -65,6 +66,8 @@ int get_fsl(RedisModuleCtx *ctx, RedisModuleString *keyname, int mode, int creat
RedisModule_CloseKey(key);
if (reply_on_failure)
RedisModule_ReplyWithError(ctx, REDISMODULE_ERRORMSG_WRONGTYPE);
RedisModuleCallReply *reply = RedisModule_Call(ctx, "INCR", "c", "fsl_wrong_type");
RedisModule_FreeCallReply(reply);
return 0;
}
......@@ -298,6 +301,154 @@ int fsl_getall(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
return REDISMODULE_OK;
}
/* Callback for blockonkeys_popall */
int blockonkeys_popall_reply_callback(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argc);
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_WRITE);
if (RedisModule_KeyType(key) == REDISMODULE_KEYTYPE_LIST) {
RedisModuleString *elem;
long len = 0;
RedisModule_ReplyWithArray(ctx, REDISMODULE_POSTPONED_ARRAY_LEN);
while ((elem = RedisModule_ListPop(key, REDISMODULE_LIST_HEAD)) != NULL) {
len++;
RedisModule_ReplyWithString(ctx, elem);
RedisModule_FreeString(ctx, elem);
}
RedisModule_ReplySetArrayLength(ctx, len);
} else {
RedisModule_ReplyWithError(ctx, "ERR Not a list");
}
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
int blockonkeys_popall_timeout_callback(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
return RedisModule_ReplyWithError(ctx, "ERR Timeout");
}
/* BLOCKONKEYS.POPALL key
*
* Blocks on an empty key for up to 3 seconds. When unblocked by a list
* operation like LPUSH, all the elements are popped and returned. Fails with an
* error on timeout. */
int blockonkeys_popall(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc != 2)
return RedisModule_WrongArity(ctx);
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_READ);
if (RedisModule_KeyType(key) == REDISMODULE_KEYTYPE_EMPTY) {
RedisModule_BlockClientOnKeys(ctx, blockonkeys_popall_reply_callback,
blockonkeys_popall_timeout_callback,
NULL, 3000, &argv[1], 1, NULL);
} else {
RedisModule_ReplyWithError(ctx, "ERR Key not empty");
}
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
/* BLOCKONKEYS.LPUSH key val [val ..]
* BLOCKONKEYS.LPUSH_UNBLOCK key val [val ..]
*
* A module equivalent of LPUSH. If the name LPUSH_UNBLOCK is used,
* RM_SignalKeyAsReady() is also called. */
int blockonkeys_lpush(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc < 3)
return RedisModule_WrongArity(ctx);
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_WRITE);
if (RedisModule_KeyType(key) != REDISMODULE_KEYTYPE_EMPTY &&
RedisModule_KeyType(key) != REDISMODULE_KEYTYPE_LIST) {
RedisModule_ReplyWithError(ctx, REDISMODULE_ERRORMSG_WRONGTYPE);
} else {
for (int i = 2; i < argc; i++) {
if (RedisModule_ListPush(key, REDISMODULE_LIST_HEAD,
argv[i]) != REDISMODULE_OK) {
RedisModule_CloseKey(key);
return RedisModule_ReplyWithError(ctx, "ERR Push failed");
}
}
}
RedisModule_CloseKey(key);
/* signal key as ready if the command is lpush_unblock */
size_t len;
const char *str = RedisModule_StringPtrLen(argv[0], &len);
if (!strncasecmp(str, "blockonkeys.lpush_unblock", len)) {
RedisModule_SignalKeyAsReady(ctx, argv[1]);
}
return RedisModule_ReplyWithSimpleString(ctx, "OK");
}
/* Callback for the BLOCKONKEYS.BLPOPN command */
int blockonkeys_blpopn_reply_callback(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argc);
long long n;
RedisModule_StringToLongLong(argv[2], &n);
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_WRITE);
int result;
if (RedisModule_KeyType(key) == REDISMODULE_KEYTYPE_LIST &&
RedisModule_ValueLength(key) >= (size_t)n) {
RedisModule_ReplyWithArray(ctx, n);
for (long i = 0; i < n; i++) {
RedisModuleString *elem = RedisModule_ListPop(key, REDISMODULE_LIST_HEAD);
RedisModule_ReplyWithString(ctx, elem);
RedisModule_FreeString(ctx, elem);
}
result = REDISMODULE_OK;
} else if (RedisModule_KeyType(key) == REDISMODULE_KEYTYPE_LIST ||
RedisModule_KeyType(key) == REDISMODULE_KEYTYPE_EMPTY) {
/* continue blocking */
result = REDISMODULE_ERR;
} else {
result = RedisModule_ReplyWithError(ctx, REDISMODULE_ERRORMSG_WRONGTYPE);
}
RedisModule_CloseKey(key);
return result;
}
int blockonkeys_blpopn_timeout_callback(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
return RedisModule_ReplyWithError(ctx, "ERR Timeout");
}
/* BLOCKONKEYS.BLPOPN key N
*
* Blocks until key has N elements and then pops them or fails after 3 seconds.
*/
int blockonkeys_blpopn(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc < 3) return RedisModule_WrongArity(ctx);
long long n;
if (RedisModule_StringToLongLong(argv[2], &n) != REDISMODULE_OK) {
return RedisModule_ReplyWithError(ctx, "ERR Invalid N");
}
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_WRITE);
int keytype = RedisModule_KeyType(key);
if (keytype != REDISMODULE_KEYTYPE_EMPTY &&
keytype != REDISMODULE_KEYTYPE_LIST) {
RedisModule_ReplyWithError(ctx, REDISMODULE_ERRORMSG_WRONGTYPE);
} else if (keytype == REDISMODULE_KEYTYPE_LIST &&
RedisModule_ValueLength(key) >= (size_t)n) {
RedisModule_ReplyWithArray(ctx, n);
for (long i = 0; i < n; i++) {
RedisModuleString *elem = RedisModule_ListPop(key, REDISMODULE_LIST_HEAD);
RedisModule_ReplyWithString(ctx, elem);
RedisModule_FreeString(ctx, elem);
}
} else {
RedisModule_BlockClientOnKeys(ctx, blockonkeys_blpopn_reply_callback,
blockonkeys_blpopn_timeout_callback,
NULL, 3000, &argv[1], 1, NULL);
}
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
......@@ -334,5 +485,21 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
if (RedisModule_CreateCommand(ctx,"fsl.getall",fsl_getall,"",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "blockonkeys.popall", blockonkeys_popall,
"", 1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "blockonkeys.lpush", blockonkeys_lpush,
"", 1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "blockonkeys.lpush_unblock", blockonkeys_lpush,
"", 1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "blockonkeys.blpopn", blockonkeys_blpopn,
"", 1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
return REDISMODULE_OK;
}
#include "redismodule.h"
#include <string.h>
#include <strings.h>
#include <assert.h>
#include <unistd.h>
#include <errno.h>
/* Command which adds a stream entry with automatic ID, like XADD *.
*
* Syntax: STREAM.ADD key field1 value1 [ field2 value2 ... ]
*
* The response is the ID of the added stream entry or an error message.
*/
int stream_add(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc < 2 || argc % 2 != 0) {
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_WRITE);
RedisModuleStreamID id;
if (RedisModule_StreamAdd(key, REDISMODULE_STREAM_ADD_AUTOID, &id,
&argv[2], (argc-2)/2) == REDISMODULE_OK) {
RedisModuleString *id_str = RedisModule_CreateStringFromStreamID(ctx, &id);
RedisModule_ReplyWithString(ctx, id_str);
RedisModule_FreeString(ctx, id_str);
} else {
RedisModule_ReplyWithError(ctx, "ERR StreamAdd failed");
}
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
/* Command which adds a stream entry N times.
*
* Syntax: STREAM.ADD key N field1 value1 [ field2 value2 ... ]
*
* Returns the number of successfully added entries.
*/
int stream_addn(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc < 3 || argc % 2 == 0) {
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
long long n, i;
if (RedisModule_StringToLongLong(argv[2], &n) == REDISMODULE_ERR) {
RedisModule_ReplyWithError(ctx, "N must be a number");
return REDISMODULE_OK;
}
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_WRITE);
for (i = 0; i < n; i++) {
if (RedisModule_StreamAdd(key, REDISMODULE_STREAM_ADD_AUTOID, NULL,
&argv[3], (argc-3)/2) == REDISMODULE_ERR)
break;
}
RedisModule_ReplyWithLongLong(ctx, i);
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
/* STREAM.DELETE key stream-id */
int stream_delete(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc != 3) return RedisModule_WrongArity(ctx);
RedisModuleStreamID id;
if (RedisModule_StringToStreamID(argv[2], &id) != REDISMODULE_OK) {
return RedisModule_ReplyWithError(ctx, "Invalid stream ID");
}
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_WRITE);
if (RedisModule_StreamDelete(key, &id) == REDISMODULE_OK) {
RedisModule_ReplyWithSimpleString(ctx, "OK");
} else {
RedisModule_ReplyWithError(ctx, "ERR StreamDelete failed");
}
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
/* STREAM.RANGE key start-id end-id
*
* Returns an array of stream items. Each item is an array on the form
* [stream-id, [field1, value1, field2, value2, ...]].
*
* A funny side-effect used for testing RM_StreamIteratorDelete() is that if any
* entry has a field named "selfdestruct", the stream entry is deleted. It is
* however included in the results of this command.
*/
int stream_range(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc != 4) {
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
RedisModuleStreamID startid, endid;
if (RedisModule_StringToStreamID(argv[2], &startid) != REDISMODULE_OK ||
RedisModule_StringToStreamID(argv[3], &endid) != REDISMODULE_OK) {
RedisModule_ReplyWithError(ctx, "Invalid stream ID");
return REDISMODULE_OK;
}
/* If startid > endid, we swap and set the reverse flag. */
int flags = 0;
if (startid.ms > endid.ms ||
(startid.ms == endid.ms && startid.seq > endid.seq)) {
RedisModuleStreamID tmp = startid;
startid = endid;
endid = tmp;
flags |= REDISMODULE_STREAM_ITERATOR_REVERSE;
}
/* Open key and start iterator. */
int openflags = REDISMODULE_READ | REDISMODULE_WRITE;
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], openflags);
if (RedisModule_StreamIteratorStart(key, flags,
&startid, &endid) != REDISMODULE_OK) {
/* Key is not a stream, etc. */
RedisModule_ReplyWithError(ctx, "ERR StreamIteratorStart failed");
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
/* Check error handling: Delete current entry when no current entry. */
assert(RedisModule_StreamIteratorDelete(key) ==
REDISMODULE_ERR);
assert(errno == ENOENT);
/* Check error handling: Fetch fields when no current entry. */
assert(RedisModule_StreamIteratorNextField(key, NULL, NULL) ==
REDISMODULE_ERR);
assert(errno == ENOENT);
/* Return array. */
RedisModule_ReplyWithArray(ctx, REDISMODULE_POSTPONED_ARRAY_LEN);
RedisModule_AutoMemory(ctx);
RedisModuleStreamID id;
long numfields;
long len = 0;
while (RedisModule_StreamIteratorNextID(key, &id,
&numfields) == REDISMODULE_OK) {
RedisModule_ReplyWithArray(ctx, 2);
RedisModuleString *id_str = RedisModule_CreateStringFromStreamID(ctx, &id);
RedisModule_ReplyWithString(ctx, id_str);
RedisModule_ReplyWithArray(ctx, numfields * 2);
int delete = 0;
RedisModuleString *field, *value;
for (long i = 0; i < numfields; i++) {
assert(RedisModule_StreamIteratorNextField(key, &field, &value) ==
REDISMODULE_OK);
RedisModule_ReplyWithString(ctx, field);
RedisModule_ReplyWithString(ctx, value);
/* check if this is a "selfdestruct" field */
size_t field_len;
const char *field_str = RedisModule_StringPtrLen(field, &field_len);
if (!strncmp(field_str, "selfdestruct", field_len)) delete = 1;
}
if (delete) {
assert(RedisModule_StreamIteratorDelete(key) == REDISMODULE_OK);
}
/* check error handling: no more fields to fetch */
assert(RedisModule_StreamIteratorNextField(key, &field, &value) ==
REDISMODULE_ERR);
assert(errno == ENOENT);
len++;
}
RedisModule_ReplySetArrayLength(ctx, len);
RedisModule_StreamIteratorStop(key);
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
/*
* STREAM.TRIM key (MAXLEN (=|~) length | MINID (=|~) id)
*/
int stream_trim(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
if (argc != 5) {
RedisModule_WrongArity(ctx);
return REDISMODULE_OK;
}
/* Parse args */
int trim_by_id = 0; /* 0 = maxlen, 1 = minid */
long long maxlen;
RedisModuleStreamID minid;
size_t arg_len;
const char *arg = RedisModule_StringPtrLen(argv[2], &arg_len);
if (!strcasecmp(arg, "minid")) {
trim_by_id = 1;
if (RedisModule_StringToStreamID(argv[4], &minid) != REDISMODULE_OK) {
RedisModule_ReplyWithError(ctx, "ERR Invalid stream ID");
return REDISMODULE_OK;
}
} else if (!strcasecmp(arg, "maxlen")) {
if (RedisModule_StringToLongLong(argv[4], &maxlen) == REDISMODULE_ERR) {
RedisModule_ReplyWithError(ctx, "ERR Maxlen must be a number");
return REDISMODULE_OK;
}
} else {
RedisModule_ReplyWithError(ctx, "ERR Invalid arguments");
return REDISMODULE_OK;
}
/* Approx or exact */
int flags;
arg = RedisModule_StringPtrLen(argv[3], &arg_len);
if (arg_len == 1 && arg[0] == '~') {
flags = REDISMODULE_STREAM_TRIM_APPROX;
} else if (arg_len == 1 && arg[0] == '=') {
flags = 0;
} else {
RedisModule_ReplyWithError(ctx, "ERR Invalid approx-or-exact mark");
return REDISMODULE_OK;
}
/* Trim */
RedisModuleKey *key = RedisModule_OpenKey(ctx, argv[1], REDISMODULE_WRITE);
long long trimmed;
if (trim_by_id) {
trimmed = RedisModule_StreamTrimByID(key, flags, &minid);
} else {
trimmed = RedisModule_StreamTrimByLength(key, flags, maxlen);
}
/* Return result */
if (trimmed < 0) {
RedisModule_ReplyWithError(ctx, "ERR Trimming failed");
} else {
RedisModule_ReplyWithLongLong(ctx, trimmed);
}
RedisModule_CloseKey(key);
return REDISMODULE_OK;
}
int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
REDISMODULE_NOT_USED(argv);
REDISMODULE_NOT_USED(argc);
if (RedisModule_Init(ctx, "stream", 1, REDISMODULE_APIVER_1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "stream.add", stream_add, "",
1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "stream.addn", stream_addn, "",
1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "stream.delete", stream_delete, "",
1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "stream.range", stream_range, "",
1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "stream.trim", stream_trim, "",
1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR;
return REDISMODULE_OK;
}
......@@ -10,7 +10,7 @@ set ::tlsdir "../../tls"
proc main {} {
parse_options
spawn_instance sentinel $::sentinel_base_port $::instances_count
spawn_instance sentinel $::sentinel_base_port $::instances_count [list "sentinel deny-scripts-reconfig no"] "../tests/includes/sentinel.conf"
spawn_instance redis $::redis_base_port $::instances_count
run_tests
cleanup
......
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