Unverified Commit 68ceb466 authored by chendianqiang's avatar chendianqiang Committed by GitHub
Browse files

Merge pull request #1 from antirez/unstable

update
parents 3d5e2c62 1c95c075
......@@ -193,7 +193,7 @@ void sortCommand(client *c) {
long limit_start = 0, limit_count = -1, start, end;
int j, dontsort = 0, vectorlen;
int getop = 0; /* GET operation counter */
int int_convertion_error = 0;
int int_conversion_error = 0;
int syntax_error = 0;
robj *sortval, *sortby = NULL, *storekey = NULL;
redisSortObject *vector; /* Resulting vector to sort */
......@@ -447,7 +447,7 @@ void sortCommand(client *c) {
serverAssertWithInfo(c,sortval,j == vectorlen);
/* Now it's time to load the right scores in the sorting vector */
if (dontsort == 0) {
if (!dontsort) {
for (j = 0; j < vectorlen; j++) {
robj *byval;
if (sortby) {
......@@ -469,7 +469,7 @@ void sortCommand(client *c) {
if (eptr[0] != '\0' || errno == ERANGE ||
isnan(vector[j].u.score))
{
int_convertion_error = 1;
int_conversion_error = 1;
}
} else if (byval->encoding == OBJ_ENCODING_INT) {
/* Don't need to decode the object if it's
......@@ -487,9 +487,7 @@ void sortCommand(client *c) {
decrRefCount(byval);
}
}
}
if (dontsort == 0) {
server.sort_desc = desc;
server.sort_alpha = alpha;
server.sort_bypattern = sortby ? 1 : 0;
......@@ -503,7 +501,7 @@ void sortCommand(client *c) {
/* Send command output to the output buffer, performing the specified
* GET/DEL/INCR/DECR operations if any. */
outputlen = getop ? getop*(end-start+1) : end-start+1;
if (int_convertion_error) {
if (int_conversion_error) {
addReplyError(c,"One or more scores can't be converted into double");
} else if (storekey == NULL) {
/* STORE option not specified, sent the sorting result to client */
......
......@@ -108,5 +108,6 @@ streamConsumer *streamLookupConsumer(streamCG *cg, sds name, int create);
streamCG *streamCreateCG(stream *s, char *name, size_t namelen, streamID *id);
streamNACK *streamCreateNACK(streamConsumer *consumer);
void streamDecodeID(void *buf, streamID *id);
int streamCompareID(streamID *a, streamID *b);
#endif
......@@ -41,6 +41,7 @@
#define STREAM_ITEM_FLAG_SAMEFIELDS (1<<1) /* Same fields as master entry. */
void streamFreeCG(streamCG *cg);
void streamFreeNACK(streamNACK *na);
size_t streamReplyWithRangeFromConsumerPEL(client *c, stream *s, streamID *start, streamID *end, size_t count, streamConsumer *consumer);
/* -----------------------------------------------------------------------
......@@ -171,7 +172,7 @@ int streamCompareID(streamID *a, streamID *b) {
* if the ID was generated by the function. However the function may return
* C_ERR if an ID was given via 'use_id', but adding it failed since the
* current top ID is greater or equal. */
int streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id, streamID *use_id) {
int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_id, streamID *use_id) {
/* If an ID was given, check that it's greater than the last entry ID
* or return an error. */
if (use_id && streamCompareID(use_id,&s->last_id) <= 0) return C_ERR;
......@@ -206,7 +207,7 @@ int streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id,
/* Create a new listpack and radix tree node if needed. Note that when
* a new listpack is created, we populate it with a "master entry". This
* is just a set of fields that is taken as refernce in order to compress
* is just a set of fields that is taken as references in order to compress
* the stream entries that we'll add inside the listpack.
*
* Note that while we use the first added entry fields to create
......@@ -221,7 +222,7 @@ int streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id,
* +-------+---------+------------+---------+--/--+---------+---------+-+
*
* count and deleted just represent respectively the total number of
* entires inside the listpack that are valid, and marked as deleted
* entries inside the listpack that are valid, and marked as deleted
* (delted flag in the entry flags set). So the total number of items
* actually inside the listpack (both deleted and not) is count+deleted.
*
......@@ -234,10 +235,24 @@ int streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id,
*
* The "0" entry at the end is the same as the 'lp-count' entry in the
* regular stream entries (see below), and marks the fact that there are
* no more entires, when we scan the stream from right to left. */
* no more entries, when we scan the stream from right to left. */
/* First of all, check if we can append to the current macro node or
* if we need to switch to the next one. 'lp' will be set to NULL if
* the current node is full. */
if (lp != NULL) {
if (server.stream_node_max_bytes &&
lp_bytes > server.stream_node_max_bytes)
{
lp = NULL;
} else if (server.stream_node_max_entries) {
int64_t count = lpGetInteger(lpFirst(lp));
if (count > server.stream_node_max_entries) lp = NULL;
}
}
int flags = STREAM_ITEM_FLAG_NONE;
if (lp == NULL || lp_bytes > STREAM_BYTES_PER_LISTPACK) {
if (lp == NULL || lp_bytes > server.stream_node_max_bytes) {
master_id = id;
streamEncodeID(rax_key,&id);
/* Create the listpack having the master entry ID and fields. */
......@@ -245,7 +260,7 @@ int streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id,
lp = lpAppendInteger(lp,1); /* One item, the one we are adding. */
lp = lpAppendInteger(lp,0); /* Zero deleted so far. */
lp = lpAppendInteger(lp,numfields);
for (int i = 0; i < numfields; i++) {
for (int64_t i = 0; i < numfields; i++) {
sds field = argv[i*2]->ptr;
lp = lpAppend(lp,(unsigned char*)field,sdslen(field));
}
......@@ -270,10 +285,10 @@ int streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id,
/* Check if the entry we are adding, have the same fields
* as the master entry. */
int master_fields_count = lpGetInteger(lp_ele);
int64_t master_fields_count = lpGetInteger(lp_ele);
lp_ele = lpNext(lp,lp_ele);
if (numfields == master_fields_count) {
int i;
int64_t i;
for (i = 0; i < master_fields_count; i++) {
sds field = argv[i*2]->ptr;
int64_t e_len;
......@@ -317,14 +332,14 @@ int streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id,
lp = lpAppendInteger(lp,id.seq - master_id.seq);
if (!(flags & STREAM_ITEM_FLAG_SAMEFIELDS))
lp = lpAppendInteger(lp,numfields);
for (int i = 0; i < numfields; i++) {
for (int64_t i = 0; i < numfields; i++) {
sds field = argv[i*2]->ptr, value = argv[i*2+1]->ptr;
if (!(flags & STREAM_ITEM_FLAG_SAMEFIELDS))
lp = lpAppend(lp,(unsigned char*)field,sdslen(field));
lp = lpAppend(lp,(unsigned char*)value,sdslen(value));
}
/* Compute and store the lp-count field. */
int lp_count = numfields;
int64_t lp_count = numfields;
lp_count += 3; /* Add the 3 fixed fields flags + ms-diff + seq-diff. */
if (!(flags & STREAM_ITEM_FLAG_SAMEFIELDS)) {
/* If the item is not compressed, it also has the fields other than
......@@ -454,7 +469,7 @@ int64_t streamTrimByLength(stream *s, size_t maxlen, int approx) {
* iteration is from the start to the end element (inclusive), otherwise
* if rev is non-zero, the iteration is reversed.
*
* Once the iterator is initalized, we iterate like this:
* Once the iterator is initialized, we iterate like this:
*
* streamIterator myiterator;
* streamIteratorStart(&myiterator,...);
......@@ -564,7 +579,7 @@ int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
/* If we are going backward, read the number of elements this
* entry is composed of, and jump backward N times to seek
* its start. */
int lp_count = lpGetInteger(si->lp_ele);
int64_t lp_count = lpGetInteger(si->lp_ele);
if (lp_count == 0) { /* We reached the master entry. */
si->lp = NULL;
si->lp_ele = NULL;
......@@ -627,12 +642,17 @@ int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
* forward, or seek the previous entry if we are going
* backward. */
if (!si->rev) {
int to_discard = (flags & STREAM_ITEM_FLAG_SAMEFIELDS) ?
*numfields : *numfields*2;
int64_t to_discard = (flags & STREAM_ITEM_FLAG_SAMEFIELDS) ?
*numfields : *numfields*2;
for (int64_t i = 0; i < to_discard; i++)
si->lp_ele = lpNext(si->lp,si->lp_ele);
} else {
int prev_times = 4; /* flag + id ms/seq diff + numfields. */
int64_t prev_times = 4; /* flag + id ms + id seq + one more to
go back to the previous entry "count"
field. */
/* If the entry was not flagged SAMEFIELD we also read the
* number of fields, so go back one more. */
if (!(flags & STREAM_ITEM_FLAG_SAMEFIELDS)) prev_times++;
while(prev_times--) si->lp_ele = lpPrev(si->lp,si->lp_ele);
}
}
......@@ -685,10 +705,22 @@ void streamIteratorRemoveEntry(streamIterator *si, streamID *current) {
/* Change the valid/deleted entries count in the master entry. */
unsigned char *p = lpFirst(lp);
aux = lpGetInteger(p);
lp = lpReplaceInteger(lp,&p,aux-1);
p = lpNext(lp,p); /* Seek deleted field. */
aux = lpGetInteger(p);
lp = lpReplaceInteger(lp,&p,aux+1);
if (aux == 1) {
/* If this is the last element in the listpack, we can remove the whole
* node. */
lpFree(lp);
raxRemove(si->stream->rax,si->ri.key,si->ri.key_len,NULL);
} else {
/* In the base case we alter the counters of valid/deleted entries. */
lp = lpReplaceInteger(lp,&p,aux-1);
p = lpNext(lp,p); /* Seek deleted field. */
aux = lpGetInteger(p);
lp = lpReplaceInteger(lp,&p,aux+1);
}
/* Update the number of entries counter. */
si->stream->length--;
/* Re-seek the iterator to fix the now messed up state. */
streamID start, end;
......@@ -811,10 +843,10 @@ void streamPropagateXCLAIM(client *c, robj *key, robj *group, robj *id, streamNA
* given, but currently such a feature is never used by the code base that
* will always pass 'spi' and propagate when a group is passed.
*
* Note that this function is recursive in certian cases. When it's called
* Note that this function is recursive in certain cases. When it's called
* with a non NULL group and consumer argument, it may call
* streamReplyWithRangeFromConsumerPEL() in order to get entries from the
* consumer pending entires list. However such a function will then call
* consumer pending entries list. However such a function will then call
* streamReplyWithRange() in order to emit single entries (found in the
* PEL by ID) to the client. This is the use case for the STREAM_RWR_RAWENTRIES
* flag.
......@@ -867,18 +899,43 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end
/* If a group is passed, we need to create an entry in the
* PEL (pending entries list) of this group *and* this consumer.
* Note that we are sure about the fact the message is not already
* associated with some other consumer, because if we reached this
* loop the IDs the user is requesting are greater than any message
* delivered for this group. */
*
* Note that we cannot be sure about the fact the message is not
* already owned by another consumer, because the admin is able
* to change the consumer group last delivered ID using the
* XGROUP SETID command. So if we find that there is already
* a NACK for the entry, we need to associate it to the new
* consumer. */
if (group && !(flags & STREAM_RWR_NOACK)) {
unsigned char buf[sizeof(streamID)];
streamEncodeID(buf,&id);
/* Try to add a new NACK. Most of the time this will work and
* will not require extra lookups. We'll fix the problem later
* if we find that there is already a entry for this ID. */
streamNACK *nack = streamCreateNACK(consumer);
int retval = 0;
retval += raxInsert(group->pel,buf,sizeof(buf),nack,NULL);
retval += raxInsert(consumer->pel,buf,sizeof(buf),nack,NULL);
serverAssert(retval == 2); /* Make sure entry was inserted. */
int group_inserted =
raxTryInsert(group->pel,buf,sizeof(buf),nack,NULL);
int consumer_inserted =
raxTryInsert(consumer->pel,buf,sizeof(buf),nack,NULL);
/* Now we can check if the entry was already busy, and
* in that case reassign the entry to the new consumer,
* or update it if the consumer is the same as before. */
if (group_inserted == 0) {
streamFreeNACK(nack);
nack = raxFind(group->pel,buf,sizeof(buf));
serverAssert(nack != raxNotFound);
raxRemove(nack->consumer->pel,buf,sizeof(buf),NULL);
/* Update the consumer and NACK metadata. */
nack->consumer = consumer;
nack->delivery_time = mstime();
nack->delivery_count = 1;
/* Add the entry in the new consumer local PEL. */
raxInsert(consumer->pel,buf,sizeof(buf),nack,NULL);
} else if (group_inserted == 1 && consumer_inserted == 0) {
serverPanic("NACK half-created. Should not be possible.");
}
/* Propagate as XCLAIM. */
if (spi) {
......@@ -899,7 +956,7 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end
/* This is an helper function for streamReplyWithRange() when called with
* group and consumer arguments, but with a range that is referring to already
* delivered messages. In this case we just emit messages that are already
* in the history of the conusmer, fetching the IDs from its PEL.
* in the history of the consumer, fetching the IDs from its PEL.
*
* Note that this function does not have a 'rev' argument because it's not
* possible to iterate in reverse using a group. Basically this function
......@@ -1035,7 +1092,7 @@ invalid:
void xaddCommand(client *c) {
streamID id;
int id_given = 0; /* Was an ID different than "*" specified? */
long long maxlen = 0; /* 0 means no maximum length. */
long long maxlen = -1; /* If left to -1 no trimming is performed. */
int approx_maxlen = 0; /* If 1 only delete whole radix tree nodes, so
the maxium length is not applied verbatim. */
int maxlen_arg_idx = 0; /* Index of the count in MAXLEN, for rewriting. */
......@@ -1059,6 +1116,11 @@ void xaddCommand(client *c) {
}
if (getLongLongFromObjectOrReply(c,c->argv[i+1],&maxlen,NULL)
!= C_OK) return;
if (maxlen < 0) {
addReplyError(c,"The MAXLEN argument must be >= 0.");
return;
}
i++;
maxlen_arg_idx = i;
} else {
......@@ -1098,7 +1160,7 @@ void xaddCommand(client *c) {
server.dirty++;
/* Remove older elements if MAXLEN was specified. */
if (maxlen) {
if (maxlen >= 0) {
if (!streamTrimByLength(s,maxlen,approx_maxlen)) {
/* If no trimming was performed, for instance because approximated
* trimming length was specified, rewrite the MAXLEN argument
......@@ -1269,14 +1331,13 @@ void xreadCommand(client *c) {
* starting from now. */
int id_idx = i - streams_arg - streams_count;
robj *key = c->argv[i-streams_count];
robj *o;
robj *o = lookupKeyRead(c->db,key);
if (o && checkType(c,o,OBJ_STREAM)) goto cleanup;
streamCG *group = NULL;
/* If a group was specified, than we need to be sure that the
* key and group actually exist. */
if (groupname) {
o = lookupKeyRead(c->db,key);
if (o && checkType(c,o,OBJ_STREAM)) goto cleanup;
if (o == NULL ||
(group = streamLookupCG(o->ptr,groupname->ptr)) == NULL)
{
......@@ -1290,8 +1351,14 @@ void xreadCommand(client *c) {
}
if (strcmp(c->argv[i]->ptr,"$") == 0) {
o = lookupKeyRead(c->db,key);
if (o && checkType(c,o,OBJ_STREAM)) goto cleanup;
if (xreadgroup) {
addReplyError(c,"The $ ID is meaningless in the context of "
"XREADGROUP: you want to read the history of "
"this consumer by specifying a proper ID, or "
"use the > ID to get new messages. The $ ID would "
"just return an empty result set.");
goto cleanup;
}
if (o) {
stream *s = o->ptr;
ids[id_idx] = s->last_id;
......@@ -1301,13 +1368,17 @@ void xreadCommand(client *c) {
}
continue;
} else if (strcmp(c->argv[i]->ptr,">") == 0) {
if (!xreadgroup || groupname == NULL) {
if (!xreadgroup) {
addReplyError(c,"The > ID can be specified only when calling "
"XREADGROUP using the GROUP <group> "
"<consumer> option.");
goto cleanup;
}
ids[id_idx] = group->last_id;
/* We use just the maximum ID to signal this is a ">" ID, anyway
* the code handling the blocking clients will have to update the
* ID later in order to match the changing consumer group last ID. */
ids[id_idx].ms = UINT64_MAX;
ids[id_idx].seq = UINT64_MAX;
continue;
}
if (streamParseIDOrReply(c,c->argv[i],ids+id_idx,0) != C_OK)
......@@ -1322,9 +1393,36 @@ void xreadCommand(client *c) {
if (o == NULL) continue;
stream *s = o->ptr;
streamID *gt = ids+i; /* ID must be greater than this. */
if (s->last_id.ms > gt->ms ||
(s->last_id.ms == gt->ms && s->last_id.seq > gt->seq))
{
int serve_synchronously = 0;
/* Check if there are the conditions to serve the client synchronously. */
if (groups) {
/* If the consumer is blocked on a group, we always serve it
* synchronously (serving its local history) if the ID specified
* was not the special ">" ID. */
if (gt->ms != UINT64_MAX ||
gt->seq != UINT64_MAX)
{
serve_synchronously = 1;
} else {
/* We also want to serve a consumer in a consumer group
* synchronously in case the group top item delivered is smaller
* than what the stream has inside. */
streamID *last = &groups[i]->last_id;
if (streamCompareID(&s->last_id, last) > 0) {
serve_synchronously = 1;
*gt = *last;
}
}
} else {
/* For consumers without a group, we serve synchronously if we can
* actually provide at least one item from the stream. */
if (streamCompareID(&s->last_id, gt) > 0) {
serve_synchronously = 1;
}
}
if (serve_synchronously) {
arraylen++;
if (arraylen == 1) arraylen_ptr = addDeferredMultiBulkLength(c);
/* streamReplyWithRange() handles the 'start' ID as inclusive,
......@@ -1336,7 +1434,7 @@ void xreadCommand(client *c) {
/* Emit the two elements sub-array consisting of the name
* of the stream and the data we extracted from it. */
addReplyMultiBulkLen(c,2);
addReplyBulk(c,c->argv[i+streams_arg]);
addReplyBulk(c,c->argv[streams_arg+i]);
streamConsumer *consumer = NULL;
if (groups) consumer = streamLookupConsumer(groups[i],
consumername->ptr,1);
......@@ -1378,6 +1476,7 @@ void xreadCommand(client *c) {
incrRefCount(consumername);
c->bpop.xread_group = groupname;
c->bpop.xread_consumer = consumername;
c->bpop.xread_group_noack = noack;
} else {
c->bpop.xread_group = NULL;
c->bpop.xread_consumer = NULL;
......@@ -1516,14 +1615,14 @@ uint64_t streamDelConsumer(streamCG *cg, sds name) {
/* XGROUP CREATE <key> <groupname> <id or $>
* XGROUP SETID <key> <id or $>
* XGROUP DELGROUP <key> <groupname>
* XGROUP DESTROY <key> <groupname>
* XGROUP DELCONSUMER <key> <groupname> <consumername> */
void xgroupCommand(client *c) {
const char *help[] = {
"CREATE <key> <groupname> <id or $> -- Create a new consumer group.",
"SETID <key> <groupname> <id or $> -- Set the current group ID.",
"DELGROUP <key> <groupname> -- Remove the specified group.",
"DELCONSUMER <key> <groupname> <consumer> -- Remove the specified conusmer.",
"DESTROY <key> <groupname> -- Remove the specified group.",
"DELCONSUMER <key> <groupname> <consumer> -- Remove the specified consumer.",
"HELP -- Prints this help.",
NULL
};
......@@ -1535,14 +1634,13 @@ NULL
/* Lookup the key now, this is common for all the subcommands but HELP. */
if (c->argc >= 4) {
robj *o = lookupKeyWriteOrReply(c,c->argv[2],shared.nokeyerr);
if (o == NULL) return;
if (o == NULL || checkType(c,o,OBJ_STREAM)) return;
s = o->ptr;
grpname = c->argv[3]->ptr;
/* Certain subcommands require the group to exist. */
if ((cg = streamLookupCG(s,grpname)) == NULL &&
(!strcasecmp(opt,"SETID") ||
!strcasecmp(opt,"DELGROUP") ||
!strcasecmp(opt,"DELCONSUMER")))
{
addReplyErrorFormat(c, "-NOGROUP No such consumer group '%s' "
......@@ -1564,22 +1662,46 @@ NULL
if (cg) {
addReply(c,shared.ok);
server.dirty++;
notifyKeyspaceEvent(NOTIFY_STREAM,"xgroup-create",
c->argv[2],c->db->id);
} else {
addReplySds(c,
sdsnew("-BUSYGROUP Consumer Group name already exists\r\n"));
}
} else if (!strcasecmp(opt,"SETID") && c->argc == 5) {
} else if (!strcasecmp(opt,"DELGROUP") && c->argc == 4) {
streamID id;
if (!strcmp(c->argv[4]->ptr,"$")) {
id = s->last_id;
} else if (streamParseIDOrReply(c,c->argv[4],&id,0) != C_OK) {
return;
}
cg->last_id = id;
addReply(c,shared.ok);
server.dirty++;
notifyKeyspaceEvent(NOTIFY_STREAM,"xgroup-setid",c->argv[2],c->db->id);
} else if (!strcasecmp(opt,"DESTROY") && c->argc == 4) {
if (cg) {
raxRemove(s->cgroups,(unsigned char*)grpname,sdslen(grpname),NULL);
streamFreeCG(cg);
addReply(c,shared.cone);
server.dirty++;
notifyKeyspaceEvent(NOTIFY_STREAM,"xgroup-destroy",
c->argv[2],c->db->id);
} else {
addReply(c,shared.czero);
}
} else if (!strcasecmp(opt,"DELCONSUMER") && c->argc == 5) {
/* Delete the consumer and returns the number of pending messages
* that were yet associated with such a consumer. */
long long pending = streamDelConsumer(cg,c->argv[4]->ptr);
addReplyLongLong(c,pending);
server.dirty++;
notifyKeyspaceEvent(NOTIFY_STREAM,"xgroup-delconsumer",
c->argv[2],c->db->id);
} else if (!strcasecmp(opt,"HELP")) {
addReplyHelp(c, help);
} else {
addReply(c,shared.syntaxerr);
addReplySubcommandSyntaxError(c);
}
}
......@@ -1728,8 +1850,10 @@ void xpendingCommand(client *c) {
/* If a consumer name was mentioned but it does not exist, we can
* just return an empty array. */
if (consumername && consumer == NULL)
if (consumername && consumer == NULL) {
addReplyMultiBulkLen(c,0);
return;
}
rax *pel = consumer ? consumer->pel : group->pel;
unsigned char startkey[sizeof(streamID)];
......@@ -1785,7 +1909,7 @@ void xpendingCommand(client *c) {
* becomes the specified <consumer>. If the minimum idle time specified
* is zero, messages are claimed regardless of their idle time.
*
* All the messages that cannot be found inside the pending entires list
* All the messages that cannot be found inside the pending entries list
* are ignored, but in case the FORCE option is used. In that case we
* create the NACK (representing a not yet acknowledged message) entry in
* the consumer group PEL.
......@@ -1970,7 +2094,7 @@ void xclaimCommand(client *c) {
nack->delivery_time = deliverytime;
/* Set the delivery attempts counter if given. */
if (retrycount >= 0) nack->delivery_count = retrycount;
/* Add the entry in the new cosnumer local PEL. */
/* Add the entry in the new consumer local PEL. */
raxInsert(consumer->pel,buf,sizeof(buf),nack,NULL);
/* Send the reply for this entry. */
if (justid) {
......@@ -1994,32 +2118,36 @@ void xclaimCommand(client *c) {
/* XDEL <key> [<ID1> <ID2> ... <IDN>]
*
* Removes the specified entries from the stream. Returns the number
* of items actaully deleted, that may be different from the number
* of items actually deleted, that may be different from the number
* of IDs passed in case certain IDs do not exist. */
void xdelCommand(client *c) {
robj *o;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL
if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL
|| checkType(c,o,OBJ_STREAM)) return;
stream *s = o->ptr;
/* We need to sanity check the IDs passed to start. Even if not
* a big issue, it is not great that the command is only partially
* executed becuase at some point an invalid ID is parsed. */
* executed because at some point an invalid ID is parsed. */
streamID id;
for (int j = 2; j < c->argc; j++) {
if (streamParseIDOrReply(c,c->argv[j],&id,0) != C_OK) return;
}
/* Actaully apply the command. */
/* Actually apply the command. */
int deleted = 0;
for (int j = 2; j < c->argc; j++) {
streamParseIDOrReply(c,c->argv[j],&id,0); /* Retval already checked. */
deleted += streamDeleteItem(s,&id);
}
signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent(NOTIFY_STREAM,"xdel",c->argv[1],c->db->id);
server.dirty += deleted;
/* Propagate the write if needed. */
if (deleted) {
signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent(NOTIFY_STREAM,"xdel",c->argv[1],c->db->id);
server.dirty += deleted;
}
addReplyLongLong(c,deleted);
}
......@@ -2040,7 +2168,7 @@ void xtrimCommand(client *c) {
/* If the key does not exist, we are ok returning zero, that is, the
* number of elements removed from the stream. */
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL
if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL
|| checkType(c,o,OBJ_STREAM)) return;
stream *s = o->ptr;
......@@ -2093,14 +2221,12 @@ void xtrimCommand(client *c) {
/* XINFO CONSUMERS key group
* XINFO GROUPS <key>
* XINFO STREAM <key>
* XINFO <key> (alias of XINFO STREAM key)
* XINFO HELP. */
void xinfoCommand(client *c) {
const char *help[] = {
"CONSUMERS <key> <groupname> -- Show consumer groups of group <groupname>.",
"GROUPS <key> -- Show the stream consumer groups.",
"STREAM <key> -- Show information about the stream.",
"<key> -- Alias for STREAM <key>.",
"HELP -- Print this help.",
NULL
};
......@@ -2112,20 +2238,19 @@ NULL
if (!strcasecmp(c->argv[1]->ptr,"HELP")) {
addReplyHelp(c, help);
return;
} else if (c->argc < 3) {
addReplyError(c,"syntax error, try 'XINFO HELP'");
return;
}
/* Handle the fact that no subcommand means "STREAM". */
if (c->argc == 2) {
opt = "STREAM";
key = c->argv[1];
} else {
opt = c->argv[1]->ptr;
key = c->argv[2];
}
/* With the exception of HELP handled before any other sub commands, all
* the ones are in the form of "<subcommand> <key>". */
opt = c->argv[1]->ptr;
key = c->argv[2];
/* Lookup the key now, this is common for all the subcommands but HELP. */
robj *o = lookupKeyWriteOrReply(c,key,shared.nokeyerr);
if (o == NULL) return;
if (o == NULL || checkType(c,o,OBJ_STREAM)) return;
s = o->ptr;
/* Dispatch the different subcommands. */
......@@ -2171,20 +2296,20 @@ NULL
raxSeek(&ri,"^",NULL,0);
while(raxNext(&ri)) {
streamCG *cg = ri.data;
addReplyMultiBulkLen(c,6);
addReplyMultiBulkLen(c,8);
addReplyStatus(c,"name");
addReplyBulkCBuffer(c,ri.key,ri.key_len);
addReplyStatus(c,"consumers");
addReplyLongLong(c,raxSize(cg->consumers));
addReplyStatus(c,"pending");
addReplyLongLong(c,raxSize(cg->pel));
addReplyStatus(c,"last-delivered-id");
addReplyStreamID(c,&cg->last_id);
}
raxStop(&ri);
} else if (c->argc == 2 ||
(!strcasecmp(opt,"STREAM") && c->argc == 3))
{
} else if (!strcasecmp(opt,"STREAM") && c->argc == 3) {
/* XINFO STREAM <key> (or the alias XINFO <key>). */
addReplyMultiBulkLen(c,12);
addReplyMultiBulkLen(c,14);
addReplyStatus(c,"length");
addReplyLongLong(c,s->length);
addReplyStatus(c,"radix-tree-keys");
......@@ -2193,6 +2318,8 @@ NULL
addReplyLongLong(c,s->rax->numnodes);
addReplyStatus(c,"groups");
addReplyLongLong(c,s->cgroups ? raxSize(s->cgroups) : 0);
addReplyStatus(c,"last-generated-id");
addReplyStreamID(c,&s->last_id);
/* To emit the first/last entry we us the streamReplyWithRange()
* API. */
......@@ -2209,7 +2336,7 @@ NULL
STREAM_RWR_RAWENTRIES,NULL);
if (!count) addReply(c,shared.nullbulk);
} else {
addReplyError(c,"syntax error, try 'XINFO HELP'");
addReplySubcommandSyntaxError(c);
}
}
......@@ -361,7 +361,7 @@ void incrDecrCommand(client *c, long long incr) {
new = o;
o->ptr = (void*)((long)value);
} else {
new = createStringObjectFromLongLong(value);
new = createStringObjectFromLongLongForValue(value);
if (o) {
dbOverwrite(c->db,c->argv[1],new);
} else {
......
......@@ -507,7 +507,7 @@ static int zslParseRange(robj *min, robj *max, zrangespec *spec) {
* + means the max string possible
*
* If the string is valid the *dest pointer is set to the redis object
* that will be used for the comparision, and ex will be set to 0 or 1
* that will be used for the comparison, and ex will be set to 0 or 1
* respectively if the item is exclusive or inclusive. C_OK will be
* returned.
*
......@@ -1100,8 +1100,8 @@ unsigned char *zzlDeleteRangeByRank(unsigned char *zl, unsigned int start, unsig
* Common sorted set API
*----------------------------------------------------------------------------*/
unsigned int zsetLength(const robj *zobj) {
int length = -1;
unsigned long zsetLength(const robj *zobj) {
unsigned long length = 0;
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
length = zzlLength(zobj->ptr);
} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
......@@ -1878,7 +1878,7 @@ void zuiClearIterator(zsetopsrc *op) {
}
}
int zuiLength(zsetopsrc *op) {
unsigned long zuiLength(zsetopsrc *op) {
if (op->subject == NULL)
return 0;
......@@ -2085,7 +2085,11 @@ int zuiFind(zsetopsrc *op, zsetopval *val, double *score) {
}
int zuiCompareByCardinality(const void *s1, const void *s2) {
return zuiLength((zsetopsrc*)s1) - zuiLength((zsetopsrc*)s2);
unsigned long first = zuiLength((zsetopsrc*)s1);
unsigned long second = zuiLength((zsetopsrc*)s2);
if (first > second) return 1;
if (first < second) return -1;
return 0;
}
#define REDIS_AGGR_SUM 1
......@@ -2129,7 +2133,7 @@ void zunionInterGenericCommand(client *c, robj *dstkey, int op) {
zsetopsrc *src;
zsetopval zval;
sds tmp;
unsigned int maxelelen = 0;
size_t maxelelen = 0;
robj *dstobj;
zset *dstzset;
zskiplistNode *znode;
......@@ -2363,8 +2367,8 @@ void zrangeGenericCommand(client *c, int reverse) {
int withscores = 0;
long start;
long end;
int llen;
int rangelen;
long llen;
long rangelen;
if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != C_OK) ||
(getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != C_OK)) return;
......@@ -2671,7 +2675,7 @@ void zcountCommand(client *c) {
robj *key = c->argv[1];
robj *zobj;
zrangespec range;
int count = 0;
unsigned long count = 0;
/* Parse the range arguments */
if (zslParseRange(c->argv[2],c->argv[3],&range) != C_OK) {
......@@ -2748,7 +2752,7 @@ void zlexcountCommand(client *c) {
robj *key = c->argv[1];
robj *zobj;
zlexrangespec range;
int count = 0;
unsigned long count = 0;
/* Parse the range arguments */
if (zslParseLexRange(c->argv[2],c->argv[3],&range) != C_OK) {
......@@ -3163,8 +3167,8 @@ void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey
signalModifiedKey(c->db,key);
}
addReplyDouble(c,score);
addReplyBulkCBuffer(c,ele,sdslen(ele));
addReplyDouble(c,score);
sdsfree(ele);
arraylen += 2;
......@@ -3216,9 +3220,9 @@ void blockingGenericZpopCommand(client *c, int where) {
return;
} else {
if (zsetLength(o) != 0) {
/* Non empty zset, this is like a normal Z[REV]POP. */
/* Non empty zset, this is like a normal ZPOP[MIN|MAX]. */
genericZpopCommand(c,&c->argv[j],1,where,1,NULL);
/* Replicate it as an Z[REV]POP instead of BZ[REV]POP. */
/* Replicate it as an ZPOP[MIN|MAX] instead of BZPOP[MIN|MAX]. */
rewriteClientCommandVector(c,2,
where == ZSET_MAX ? shared.zpopmax : shared.zpopmin,
c->argv[j]);
......
......@@ -451,7 +451,7 @@ int string2ld(const char *s, size_t slen, long double *dp) {
/* Convert a double to a string representation. Returns the number of bytes
* required. The representation should always be parsable by strtod(3).
* This function does not support human-friendly formatting like ld2string
* does. It is intented mainly to be used inside t_zset.c when writing scores
* does. It is intended mainly to be used inside t_zset.c when writing scores
* into a ziplist representing a sorted set. */
int d2string(char *buf, size_t len, double value) {
if (isnan(value)) {
......
......@@ -27,7 +27,7 @@
* traversal.
*
* <uint16_t zllen> is the number of entries. When there are more than
* 2^16-2 entires, this value is set to 2^16-1 and we need to traverse the
* 2^16-2 entries, this value is set to 2^16-1 and we need to traverse the
* entire list to know how many items it holds.
*
* <uint8_t zlend> is a special entry representing the end of the ziplist.
......@@ -256,7 +256,7 @@
#define ZIPLIST_ENTRY_END(zl) ((zl)+intrev32ifbe(ZIPLIST_BYTES(zl))-1)
/* Increment the number of items field in the ziplist header. Note that this
* macro should never overflow the unsigned 16 bit integer, since entires are
* macro should never overflow the unsigned 16 bit integer, since entries are
* always pushed one at a time. When UINT16_MAX is reached we want the count
* to stay there to signal that a full scan is needed to get the number of
* items inside the ziplist. */
......@@ -269,7 +269,7 @@
* Note that this is not how the data is actually encoded, is just what we
* get filled by a function in order to operate more easily. */
typedef struct zlentry {
unsigned int prevrawlensize; /* Bytes used to encode the previos entry len*/
unsigned int prevrawlensize; /* Bytes used to encode the previous entry len*/
unsigned int prevrawlen; /* Previous entry len. */
unsigned int lensize; /* Bytes used to encode this entry type/len.
For example strings have a 1, 2 or 5 bytes
......@@ -431,7 +431,7 @@ unsigned int zipStorePrevEntryLength(unsigned char *p, unsigned int len) {
/* Return the length of the previous element, and the number of bytes that
* are used in order to encode the previous element length.
* 'ptr' must point to the prevlen prefix of an entry (that encodes the
* length of the previos entry in order to navigate the elements backward).
* length of the previous entry in order to navigate the elements backward).
* The length of the previous entry is stored in 'prevlen', the number of
* bytes needed to encode the previous entry length are stored in
* 'prevlensize'. */
......
......@@ -30,6 +30,7 @@
#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
/* This function provide us access to the original libc free(). This is useful
* for instance to free results obtained by backtrace_symbols(). We need
......@@ -164,7 +165,7 @@ void *zrealloc(void *ptr, size_t size) {
*((size_t*)newptr) = size;
update_zmalloc_stat_free(oldsize);
update_zmalloc_stat_alloc(size);
update_zmalloc_stat_alloc(size+PREFIX_SIZE);
return (char*)newptr+PREFIX_SIZE;
#endif
}
......@@ -181,6 +182,9 @@ size_t zmalloc_size(void *ptr) {
if (size&(sizeof(long)-1)) size += sizeof(long)-(size&(sizeof(long)-1));
return size+PREFIX_SIZE;
}
size_t zmalloc_usable(void *ptr) {
return zmalloc_usable(ptr)-PREFIX_SIZE;
}
#endif
void zfree(void *ptr) {
......@@ -379,7 +383,7 @@ size_t zmalloc_get_private_dirty(long pid) {
}
/* Returns the size of physical memory (RAM) in bytes.
* It looks ugly, but this is the cleanest way to achive cross platform results.
* It looks ugly, but this is the cleanest way to achieve cross platform results.
* Cleaned up from:
*
* http://nadeausoftware.com/articles/2012/09/c_c_tip_how_get_physical_memory_size_system
......@@ -418,7 +422,7 @@ size_t zmalloc_get_memory_size(void) {
mib[0] = CTL_HW;
#if defined(HW_REALMEM)
mib[1] = HW_REALMEM; /* FreeBSD. ----------------- */
#elif defined(HW_PYSMEM)
#elif defined(HW_PHYSMEM)
mib[1] = HW_PHYSMEM; /* Others. ------------------ */
#endif
unsigned int size = 0; /* 32-bit */
......
......@@ -63,6 +63,11 @@
#ifndef ZMALLOC_LIB
#define ZMALLOC_LIB "libc"
#ifdef __GLIBC__
#include <malloc.h>
#define HAVE_MALLOC_SIZE 1
#define zmalloc_size(p) malloc_usable_size(p)
#endif
#endif
/* We can enable the Redis defrag capabilities only if we are using Jemalloc
......@@ -93,6 +98,9 @@ void *zmalloc_no_tcache(size_t size);
#ifndef HAVE_MALLOC_SIZE
size_t zmalloc_size(void *ptr);
size_t zmalloc_usable(void *ptr);
#else
#define zmalloc_usable(p) zmalloc_size(p)
#endif
#endif /* __ZMALLOC_H */
......@@ -92,3 +92,80 @@ test "Node #10 should eventually replicate node #5" {
fail "#10 didn't became slave of #5"
}
}
source "../tests/includes/init-tests.tcl"
# Create a cluster with 3 master and 15 slaves, so that we have 5
# slaves for eatch master.
test "Create a 3 nodes cluster" {
create_cluster 3 15
}
test "Cluster is up" {
assert_cluster_state ok
}
test "The first master has actually 5 slaves" {
assert {[llength [lindex [R 0 role] 2]] == 5}
}
test {Slaves of #0 are instance #3, #6, #9, #12 and #15 as expected} {
set port0 [get_instance_attrib redis 0 port]
assert {[lindex [R 3 role] 2] == $port0}
assert {[lindex [R 6 role] 2] == $port0}
assert {[lindex [R 9 role] 2] == $port0}
assert {[lindex [R 12 role] 2] == $port0}
assert {[lindex [R 15 role] 2] == $port0}
}
test {Instance #3, #6, #9, #12 and #15 synced with the master} {
wait_for_condition 1000 50 {
[RI 3 master_link_status] eq {up} &&
[RI 6 master_link_status] eq {up} &&
[RI 9 master_link_status] eq {up} &&
[RI 12 master_link_status] eq {up} &&
[RI 15 master_link_status] eq {up}
} else {
fail "Instance #3 or #6 or #9 or #12 or #15 master link status is not up"
}
}
proc master_detected {instances} {
foreach instance [dict keys $instances] {
if {[RI $instance role] eq {master}} {
return true
}
}
return false
}
test "New Master down consecutively" {
set instances [dict create 0 1 3 1 6 1 9 1 12 1 15 1]
set loops [expr {[dict size $instances]-1}]
for {set i 0} {$i < $loops} {incr i} {
set master_id -1
foreach instance [dict keys $instances] {
if {[RI $instance role] eq {master}} {
set master_id $instance
break;
}
}
if {$master_id eq -1} {
fail "no master detected, #loop $i"
}
set instances [dict remove $instances $master_id]
kill_instance redis $master_id
wait_for_condition 1000 50 {
[master_detected $instances]
} else {
failover "No failover detected when master $master_id fails"
}
assert_cluster_state ok
}
}
......@@ -39,6 +39,25 @@ start_server [list overrides [list "dir" $server_path]] {
} {0000000000000000000000000000000000000000}
}
start_server [list overrides [list "dir" $server_path]] {
test {Test RDB stream encoding} {
for {set j 0} {$j < 1000} {incr j} {
if {rand() < 0.9} {
r xadd stream * foo $j
} else {
r xadd stream * bar $j
}
}
r xgroup create stream mygroup 0
r xreadgroup GROUP mygroup Alice COUNT 1 STREAMS stream >
set digest [r debug digest]
r debug reload
set newdigest [r debug digest]
assert {$digest eq $newdigest}
r del stream
}
}
# Helper function to start a server and kill it, just to check the error
# logged.
set defaults {}
......
......@@ -11,7 +11,7 @@ proc stop_bg_complex_data {handle} {
# partial resyncs attempts, all this while flooding the master with
# write queries.
#
# You can specifiy backlog size, ttl, delay before reconnection, test duration
# You can specify backlog size, ttl, delay before reconnection, test duration
# in seconds, and an additional condition to verify at the end.
#
# If reconnect is > 0, the test actually try to break the connection and
......
......@@ -66,3 +66,13 @@ test "SDOWN is triggered by misconfigured instance repling with errors" {
R 0 bgsave
ensure_master_up
}
# We use this test setup to also test command renaming, as a side
# effect of the master going down if we send PONG instead of PING
test "SDOWN is triggered if we rename PING to PONG" {
ensure_master_up
S 4 SENTINEL SET mymaster rename-command PING PONG
ensure_master_down
S 4 SENTINEL SET mymaster rename-command PING PING
ensure_master_up
}
......@@ -276,6 +276,12 @@ proc start_server {options {code undefined}} {
error_and_quit $config_file $line
}
if {$::wait_server} {
set msg "server started PID: [dict get $srv "pid"]. press any key to continue..."
puts $msg
read stdin 1
}
while 1 {
# check that the server actually started and is ready for connections
if {[exec grep -i "Ready to accept" | wc -l < $stdout] > 0} {
......
......@@ -375,3 +375,19 @@ proc start_write_load {host port seconds} {
proc stop_write_load {handle} {
catch {exec /bin/kill -9 $handle}
}
proc K { x y } { set x }
# Shuffle a list. From Tcl wiki. Originally from Steve Cohen that improved
# other versions. Code should be under public domain.
proc lshuffle {list} {
set n [llength $list]
while {$n>0} {
set j [expr {int(rand()*$n)}]
lappend slist [lindex $list $j]
incr n -1
set temp [lindex $list $n]
set list [lreplace [K $list [set list {}]] $j $j $temp]
}
return $slist
}
......@@ -61,6 +61,7 @@ set ::all_tests {
unit/hyperloglog
unit/lazyfree
unit/wait
unit/pendingquerybuf
}
# Index to the next test to run in the ::all_tests list.
set ::next_test 0
......@@ -82,6 +83,8 @@ set ::force_failure 0
set ::timeout 600; # 10 minutes without progresses will quit the test.
set ::last_progress [clock seconds]
set ::active_servers {} ; # Pids of active Redis instances.
set ::dont_clean 0
set ::wait_server 0
# Set to 1 when we are running in client mode. The Redis test uses a
# server-client model to run tests simultaneously. The server instance
......@@ -175,6 +178,9 @@ proc s {args} {
}
proc cleanup {} {
if {$::dont_clean} {
return
}
if {!$::quiet} {puts -nonewline "Cleanup: may take some time... "}
flush stdout
catch {exec rm -rf {*}[glob tests/tmp/redis.conf.*]}
......@@ -224,6 +230,7 @@ proc test_server_cron {} {
if {$elapsed > $::timeout} {
set err "\[[colorstr red TIMEOUT]\]: clients state report follows."
puts $err
lappend ::failed_tests $err
show_clients_state
kill_clients
force_kill_all_servers
......@@ -410,6 +417,8 @@ proc print_help_screen {} {
"--clients <num> Number of test clients (default 16)."
"--timeout <sec> Test timeout in seconds (default 10 min)."
"--force-failure Force the execution of a test that always fails."
"--dont-clean don't delete redis log files after the run"
"--wait-server wait after server is started (so that you can attach a debugger)"
"--help Print this help screen."
} "\n"]
}
......@@ -463,6 +472,10 @@ for {set j 0} {$j < [llength $argv]} {incr j} {
} elseif {$opt eq {--clients}} {
set ::numclients $arg
incr j
} elseif {$opt eq {--dont-clean}} {
set ::dont_clean 1
} elseif {$opt eq {--wait-server}} {
set ::wait_server 1
} elseif {$opt eq {--timeout}} {
set ::timeout $arg
incr j
......
......@@ -25,6 +25,39 @@ start_server {tags {"dump"}} {
assert {$ttl >= (2569591501-3000) && $ttl <= 2569591501}
r get foo
} {bar}
test {RESTORE can set an absolute expire} {
r set foo bar
set encoded [r dump foo]
r del foo
set now [clock milliseconds]
r restore foo [expr $now+3000] $encoded absttl
set ttl [r pttl foo]
assert {$ttl >= 2998 && $ttl <= 3000}
r get foo
} {bar}
test {RESTORE can set LRU} {
r set foo bar
set encoded [r dump foo]
r del foo
r config set maxmemory-policy allkeys-lru
r restore foo 0 $encoded idletime 1000
set idle [r object idletime foo]
assert {$idle >= 1000 && $idle <= 1002}
r get foo
} {bar}
test {RESTORE can set LFU} {
r set foo bar
set encoded [r dump foo]
r del foo
r config set maxmemory-policy allkeys-lfu
r restore foo 0 $encoded freq 100
set freq [r object freq foo]
assert {$freq == 100}
r get foo
} {bar}
test {RESTORE returns an error of the key already exists} {
r set foo bar
......@@ -246,7 +279,7 @@ start_server {tags {"dump"}} {
set e
} {*empty string*}
test {MIGRATE with mutliple keys migrate just existing ones} {
test {MIGRATE with multiple keys migrate just existing ones} {
set first [srv 0 client]
r set key1 "v1"
r set key2 "v2"
......
......@@ -121,7 +121,7 @@ start_server {tags {"expire"}} {
list $a $b
} {somevalue {}}
test {TTL returns tiem to live in seconds} {
test {TTL returns time to live in seconds} {
r del x
r setex x 10 somevalue
set ttl [r ttl x]
......
......@@ -142,3 +142,95 @@ start_server {tags {"maxmemory"}} {
}
}
}
proc test_slave_buffers {cmd_count payload_len limit_memory pipeline} {
start_server {tags {"maxmemory"}} {
start_server {} {
set slave [srv 0 client]
set slave_host [srv 0 host]
set slave_port [srv 0 port]
set master [srv -1 client]
set master_host [srv -1 host]
set master_port [srv -1 port]
# add 100 keys of 100k (10MB total)
for {set j 0} {$j < 100} {incr j} {
$master setrange "key:$j" 100000 asdf
}
$master config set maxmemory-policy allkeys-random
$master config set client-output-buffer-limit "slave 100000000 100000000 60"
$master config set repl-backlog-size [expr {10*1024}]
$slave slaveof $master_host $master_port
wait_for_condition 50 100 {
[s 0 master_link_status] eq {up}
} else {
fail "Replication not started."
}
# measure used memory after the slave connected and set maxmemory
set orig_used [s -1 used_memory]
set orig_client_buf [s -1 mem_clients_normal]
set orig_mem_not_counted_for_evict [s -1 mem_not_counted_for_evict]
set orig_used_no_repl [expr {$orig_used - $orig_mem_not_counted_for_evict}]
set limit [expr {$orig_used - $orig_mem_not_counted_for_evict + 20*1024}]
if {$limit_memory==1} {
$master config set maxmemory $limit
}
# put the slave to sleep
set rd_slave [redis_deferring_client]
$rd_slave debug sleep 60
# send some 10mb woth of commands that don't increase the memory usage
if {$pipeline == 1} {
set rd_master [redis_deferring_client -1]
for {set k 0} {$k < $cmd_count} {incr k} {
$rd_master setrange key:0 0 [string repeat A $payload_len]
}
for {set k 0} {$k < $cmd_count} {incr k} {
#$rd_master read
}
} else {
for {set k 0} {$k < $cmd_count} {incr k} {
$master setrange key:0 0 [string repeat A $payload_len]
}
}
set new_used [s -1 used_memory]
set slave_buf [s -1 mem_clients_slaves]
set client_buf [s -1 mem_clients_normal]
set mem_not_counted_for_evict [s -1 mem_not_counted_for_evict]
set used_no_repl [expr {$new_used - $mem_not_counted_for_evict}]
set delta [expr {($used_no_repl - $client_buf) - ($orig_used_no_repl - $orig_client_buf)}]
assert {[$master dbsize] == 100}
assert {$slave_buf > 2*1024*1024} ;# some of the data may have been pushed to the OS buffers
assert {$delta < 50*1024 && $delta > -50*1024} ;# 1 byte unaccounted for, with 1M commands will consume some 1MB
$master client kill type slave
set killed_used [s -1 used_memory]
set killed_slave_buf [s -1 mem_clients_slaves]
set killed_mem_not_counted_for_evict [s -1 mem_not_counted_for_evict]
set killed_used_no_repl [expr {$killed_used - $killed_mem_not_counted_for_evict}]
set delta_no_repl [expr {$killed_used_no_repl - $used_no_repl}]
assert {$killed_slave_buf == 0}
assert {$delta_no_repl > -50*1024 && $delta_no_repl < 50*1024} ;# 1 byte unaccounted for, with 1M commands will consume some 1MB
}
}
}
test {slave buffer are counted correctly} {
# we wanna use many small commands, and we don't wanna wait long
# so we need to use a pipeline (redis_deferring_client)
# that may cause query buffer to fill and induce eviction, so we disable it
test_slave_buffers 1000000 10 0 1
}
test {slave buffer don't induce eviction} {
# test again with fewer (and bigger) commands without pipeline, but with eviction
test_slave_buffers 100000 100 1 0
}
......@@ -41,7 +41,7 @@ start_server {tags {"defrag"}} {
test "Active defrag" {
r config set activedefrag no
r config set active-defrag-threshold-lower 5
r config set active-defrag-cycle-min 25
r config set active-defrag-cycle-min 65
r config set active-defrag-cycle-max 75
r config set active-defrag-ignore-bytes 2mb
r config set maxmemory 100mb
......@@ -66,9 +66,10 @@ start_server {tags {"defrag"}} {
}
# Wait for the active defrag to stop working.
wait_for_condition 100 100 {
wait_for_condition 150 100 {
[s active_defrag_running] eq 0
} else {
after 120 ;# serverCron only updates the info once in 100ms
puts [r info memory]
puts [r memory malloc-stats]
fail "defrag didn't stop."
......@@ -97,10 +98,15 @@ start_server {tags {"defrag"}} {
r config set active-defrag-ignore-bytes 2mb
r config set maxmemory 0
r config set list-max-ziplist-size 5 ;# list of 10k items will have 2000 quicklist nodes
r config set stream-node-max-entries 5
r hmset hash h1 v1 h2 v2 h3 v3
r lpush list a b c d
r zadd zset 0 a 1 b 2 c 3 d
r sadd set a b c d
r xadd stream * item 1 value a
r xadd stream * item 2 value b
r xgroup create stream mygroup 0
r xreadgroup GROUP mygroup Alice COUNT 1 STREAMS stream >
# create big keys with 10k items
set rd [redis_deferring_client]
......@@ -109,8 +115,9 @@ start_server {tags {"defrag"}} {
$rd lpush biglist [concat "asdfasdfasdf" $j]
$rd zadd bigzset $j [concat "asdfasdfasdf" $j]
$rd sadd bigset [concat "asdfasdfasdf" $j]
$rd xadd bigstream * item 1 value a
}
for {set j 0} {$j < 40000} {incr j} {
for {set j 0} {$j < 50000} {incr j} {
$rd read ; # Discard replies
}
......@@ -134,7 +141,7 @@ start_server {tags {"defrag"}} {
for {set j 0} {$j < 500000} {incr j} {
$rd read ; # Discard replies
}
assert {[r dbsize] == 500008}
assert {[r dbsize] == 500010}
# create some fragmentation
for {set j 0} {$j < 500000} {incr j 2} {
......@@ -143,7 +150,7 @@ start_server {tags {"defrag"}} {
for {set j 0} {$j < 500000} {incr j 2} {
$rd read ; # Discard replies
}
assert {[r dbsize] == 250008}
assert {[r dbsize] == 250010}
# start defrag
after 120 ;# serverCron only updates the info once in 100ms
......@@ -155,6 +162,7 @@ start_server {tags {"defrag"}} {
r config set latency-monitor-threshold 5
r latency reset
set digest [r debug digest]
catch {r config set activedefrag yes} e
if {![string match {DISABLED*} $e]} {
# wait for the active defrag to start working (decision once a second)
......@@ -168,6 +176,7 @@ start_server {tags {"defrag"}} {
wait_for_condition 500 100 {
[s active_defrag_running] eq 0
} else {
after 120 ;# serverCron only updates the info once in 100ms
puts [r info memory]
puts [r memory malloc-stats]
fail "defrag didn't stop."
......@@ -193,9 +202,11 @@ start_server {tags {"defrag"}} {
# due to high fragmentation, 10hz, and active-defrag-cycle-max set to 75,
# we expect max latency to be not much higher than 75ms
assert {$max_latency <= 80}
} else {
set _ ""
}
} {}
# verify the data isn't corrupted or changed
set newdigest [r debug digest]
assert {$digest eq $newdigest}
r save ;# saving an rdb iterates over all the data / pointers
} {OK}
}
}
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