Skip to content
GitLab
Menu
Projects
Groups
Snippets
Loading...
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in
Toggle navigation
Menu
Open sidebar
ruanhaishen
redis
Commits
49816941
Unverified
Commit
49816941
authored
Feb 27, 2019
by
chendianqiang
Committed by
GitHub
Feb 27, 2019
Browse files
Merge pull request #2 from antirez/unstable
merge from redis
parents
68ceb466
f311a529
Changes
120
Hide whitespace changes
Inline
Side-by-side
src/t_set.c
View file @
49816941
...
...
@@ -207,7 +207,7 @@ sds setTypeNextObject(setTypeIterator *si) {
* used field with values which are easy to trap if misused. */
int
setTypeRandomElement
(
robj
*
setobj
,
sds
*
sdsele
,
int64_t
*
llele
)
{
if
(
setobj
->
encoding
==
OBJ_ENCODING_HT
)
{
dictEntry
*
de
=
dictGetRandomKey
(
setobj
->
ptr
);
dictEntry
*
de
=
dictGet
Fair
RandomKey
(
setobj
->
ptr
);
*
sdsele
=
dictGetKey
(
de
);
*
llele
=
-
123456789
;
/* Not needed. Defensive. */
}
else
if
(
setobj
->
encoding
==
OBJ_ENCODING_INTSET
)
{
...
...
@@ -415,13 +415,13 @@ void spopWithCountCommand(client *c) {
/* Make sure a key with the name inputted exists, and that it's type is
* indeed a set. Otherwise, return nil */
if
((
set
=
lookupKeyReadOrReply
(
c
,
c
->
argv
[
1
],
shared
.
emptymultibulk
))
if
((
set
=
lookupKeyReadOrReply
(
c
,
c
->
argv
[
1
],
shared
.
null
[
c
->
resp
]
))
==
NULL
||
checkType
(
c
,
set
,
OBJ_SET
))
return
;
/* If count is zero, serve an empty multibulk ASAP to avoid special
* cases later. */
if
(
count
==
0
)
{
addReply
(
c
,
shared
.
emptymultibulk
);
addReply
Null
(
c
);
return
;
}
...
...
@@ -455,7 +455,7 @@ void spopWithCountCommand(client *c) {
robj
*
propargv
[
3
];
propargv
[
0
]
=
createStringObject
(
"SREM"
,
4
);
propargv
[
1
]
=
c
->
argv
[
1
];
addReply
MultiBulk
Len
(
c
,
count
);
addReply
Set
Len
(
c
,
count
);
/* Common iteration vars. */
sds
sdsele
;
...
...
@@ -516,11 +516,7 @@ void spopWithCountCommand(client *c) {
sdsfree
(
sdsele
);
}
/* Assign the new set as the key value. */
incrRefCount
(
set
);
/* Protect the old set value. */
dbOverwrite
(
c
->
db
,
c
->
argv
[
1
],
newset
);
/* Tranfer the old set to the client and release it. */
/* Transfer the old set to the client. */
setTypeIterator
*
si
;
si
=
setTypeInitIterator
(
set
);
while
((
encoding
=
setTypeNext
(
si
,
&
sdsele
,
&
llele
))
!=
-
1
)
{
...
...
@@ -539,7 +535,9 @@ void spopWithCountCommand(client *c) {
decrRefCount
(
objele
);
}
setTypeReleaseIterator
(
si
);
decrRefCount
(
set
);
/* Assign the new set as the key value. */
dbOverwrite
(
c
->
db
,
c
->
argv
[
1
],
newset
);
}
/* Don't propagate the command itself even if we incremented the
...
...
@@ -568,8 +566,8 @@ void spopCommand(client *c) {
/* Make sure a key with the name inputted exists, and that it's type is
* indeed a set */
if
((
set
=
lookupKeyWriteOrReply
(
c
,
c
->
argv
[
1
],
shared
.
null
bulk
))
==
NULL
||
checkType
(
c
,
set
,
OBJ_SET
))
return
;
if
((
set
=
lookupKeyWriteOrReply
(
c
,
c
->
argv
[
1
],
shared
.
null
[
c
->
resp
]))
==
NULL
||
checkType
(
c
,
set
,
OBJ_SET
))
return
;
/* Get a random element from the set */
encoding
=
setTypeRandomElement
(
set
,
&
sdsele
,
&
llele
);
...
...
@@ -634,13 +632,13 @@ void srandmemberWithCountCommand(client *c) {
uniq
=
0
;
}
if
((
set
=
lookupKeyReadOrReply
(
c
,
c
->
argv
[
1
],
shared
.
emptymultibulk
))
if
((
set
=
lookupKeyReadOrReply
(
c
,
c
->
argv
[
1
],
shared
.
null
[
c
->
resp
]
))
==
NULL
||
checkType
(
c
,
set
,
OBJ_SET
))
return
;
size
=
setTypeSize
(
set
);
/* If count is zero, serve it ASAP to avoid special cases later. */
if
(
count
==
0
)
{
addReply
(
c
,
shared
.
emptymultibulk
);
addReply
Null
(
c
);
return
;
}
...
...
@@ -649,7 +647,7 @@ void srandmemberWithCountCommand(client *c) {
* This case is trivial and can be served without auxiliary data
* structures. */
if
(
!
uniq
)
{
addReply
MultiBulk
Len
(
c
,
count
);
addReply
Set
Len
(
c
,
count
);
while
(
count
--
)
{
encoding
=
setTypeRandomElement
(
set
,
&
ele
,
&
llele
);
if
(
encoding
==
OBJ_ENCODING_INTSET
)
{
...
...
@@ -739,7 +737,7 @@ void srandmemberWithCountCommand(client *c) {
dictIterator
*
di
;
dictEntry
*
de
;
addReply
MultiBulk
Len
(
c
,
count
);
addReply
Set
Len
(
c
,
count
);
di
=
dictGetIterator
(
d
);
while
((
de
=
dictNext
(
di
))
!=
NULL
)
addReplyBulk
(
c
,
dictGetKey
(
de
));
...
...
@@ -762,8 +760,8 @@ void srandmemberCommand(client *c) {
return
;
}
if
((
set
=
lookupKeyReadOrReply
(
c
,
c
->
argv
[
1
],
shared
.
null
bulk
))
==
NULL
||
checkType
(
c
,
set
,
OBJ_SET
))
return
;
if
((
set
=
lookupKeyReadOrReply
(
c
,
c
->
argv
[
1
],
shared
.
null
[
c
->
resp
]))
==
NULL
||
checkType
(
c
,
set
,
OBJ_SET
))
return
;
encoding
=
setTypeRandomElement
(
set
,
&
ele
,
&
llele
);
if
(
encoding
==
OBJ_ENCODING_INTSET
)
{
...
...
@@ -815,7 +813,7 @@ void sinterGenericCommand(client *c, robj **setkeys,
}
addReply
(
c
,
shared
.
czero
);
}
else
{
addReply
(
c
,
shared
.
emptymultibulk
);
addReply
Null
(
c
);
}
return
;
}
...
...
@@ -835,7 +833,7 @@ void sinterGenericCommand(client *c, robj **setkeys,
* to the output list and save the pointer to later modify it with the
* right length */
if
(
!
dstkey
)
{
replylen
=
addDeferred
MultiBulkLength
(
c
);
replylen
=
add
Reply
Deferred
Len
(
c
);
}
else
{
/* If we have a target key where to store the resulting set
* create this key with an empty set inside */
...
...
@@ -913,7 +911,7 @@ void sinterGenericCommand(client *c, robj **setkeys,
signalModifiedKey
(
c
->
db
,
dstkey
);
server
.
dirty
++
;
}
else
{
setDeferred
MultiBulkLength
(
c
,
replylen
,
cardinality
);
setDeferred
SetLen
(
c
,
replylen
,
cardinality
);
}
zfree
(
sets
);
}
...
...
@@ -1059,7 +1057,7 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
/* Output the content of the resulting set, if not in STORE mode */
if
(
!
dstkey
)
{
addReply
MultiBulk
Len
(
c
,
cardinality
);
addReply
Set
Len
(
c
,
cardinality
);
si
=
setTypeInitIterator
(
dstset
);
while
((
ele
=
setTypeNextObject
(
si
))
!=
NULL
)
{
addReplyBulkCBuffer
(
c
,
ele
,
sdslen
(
ele
));
...
...
src/t_stream.c
View file @
49816941
...
...
@@ -37,7 +37,7 @@
* mark the entry as deleted, or having the same field as the "master"
* entry at the start of the listpack> */
#define STREAM_ITEM_FLAG_NONE 0
/* No special flags. */
#define STREAM_ITEM_FLAG_DELETED (1<<0)
/* Entry is delted. Skip it. */
#define STREAM_ITEM_FLAG_DELETED (1<<0)
/* Entry is del
e
ted. Skip it. */
#define STREAM_ITEM_FLAG_SAMEFIELDS (1<<1)
/* Same fields as master entry. */
void
streamFreeCG
(
streamCG
*
cg
);
...
...
@@ -165,7 +165,7 @@ int streamCompareID(streamID *a, streamID *b) {
* Returns the new entry ID populating the 'added_id' structure.
*
* If 'use_id' is not NULL, the ID is not auto-generated by the function,
* but instead the passed ID is u
e
sd to add the new entry. In this case
* but instead the passed ID is us
e
d to add the new entry. In this case
* adding the entry may fail as specified later in this comment.
*
* The function returns C_OK if the item was added, this is always true
...
...
@@ -223,13 +223,13 @@ int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_
*
* count and deleted just represent respectively the total number of
* entries inside the listpack that are valid, and marked as deleted
* (delted flag in the entry flags set). So the total number of items
* (del
e
ted flag in the entry flags set). So the total number of items
* actually inside the listpack (both deleted and not) is count+deleted.
*
* The real entries will be encoded with an ID that is just the
* millisecond and sequence difference compared to the key stored at
* the radix tree node containing the listpack (delta encoding), and
* if the fields of the entry are the same as the master enty fields, the
* if the fields of the entry are the same as the master ent
r
y fields, the
* entry flags will specify this fact and the entry fields and number
* of fields will be omitted (see later in the code of this function).
*
...
...
@@ -349,7 +349,8 @@ int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_
lp
=
lpAppendInteger
(
lp
,
lp_count
);
/* Insert back into the tree in order to update the listpack pointer. */
raxInsert
(
s
->
rax
,(
unsigned
char
*
)
&
rax_key
,
sizeof
(
rax_key
),
lp
,
NULL
);
if
(
ri
.
data
!=
lp
)
raxInsert
(
s
->
rax
,(
unsigned
char
*
)
&
rax_key
,
sizeof
(
rax_key
),
lp
,
NULL
);
s
->
length
++
;
s
->
last_id
=
id
;
if
(
added_id
)
*
added_id
=
id
;
...
...
@@ -485,7 +486,7 @@ int64_t streamTrimByLength(stream *s, size_t maxlen, int approx) {
* }
* streamIteratorStop(&myiterator); */
void
streamIteratorStart
(
streamIterator
*
si
,
stream
*
s
,
streamID
*
start
,
streamID
*
end
,
int
rev
)
{
/* Intialize the iterator and translates the iteration start/stop
/* In
i
tialize the iterator and translates the iteration start/stop
* elements into a 128 big big-endian number. */
if
(
start
)
{
streamEncodeID
(
si
->
start_key
,
start
);
...
...
@@ -548,15 +549,22 @@ int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
si
->
master_fields_count
=
lpGetInteger
(
si
->
lp_ele
);
si
->
lp_ele
=
lpNext
(
si
->
lp
,
si
->
lp_ele
);
/* Seek first field. */
si
->
master_fields_start
=
si
->
lp_ele
;
/* Skip master fileds to seek the first entry. */
for
(
uint64_t
i
=
0
;
i
<
si
->
master_fields_count
;
i
++
)
si
->
lp_ele
=
lpNext
(
si
->
lp
,
si
->
lp_ele
);
/* We are now pointing the zero term of the master entry. If
* we are iterating in reverse order, we need to seek the
* end of the listpack. */
if
(
si
->
rev
)
si
->
lp_ele
=
lpLast
(
si
->
lp
);
/* We are now pointing to the first field of the master entry.
* We need to seek either the first or the last entry depending
* on the direction of the iteration. */
if
(
!
si
->
rev
)
{
/* If we are iterating in normal order, skip the master fields
* to seek the first actual entry. */
for
(
uint64_t
i
=
0
;
i
<
si
->
master_fields_count
;
i
++
)
si
->
lp_ele
=
lpNext
(
si
->
lp
,
si
->
lp_ele
);
}
else
{
/* If we are iterating in reverse direction, just seek the
* last part of the last entry in the listpack (that is, the
* fields count). */
si
->
lp_ele
=
lpLast
(
si
->
lp
);
}
}
else
if
(
si
->
rev
)
{
/* If we are iter
e
ating in the reverse order, and this is not
/* If we are iterating in the reverse order, and this is not
* the first entry emitted for this listpack, then we already
* emitted the current entry, and have to go back to the previous
* one. */
...
...
@@ -717,6 +725,10 @@ void streamIteratorRemoveEntry(streamIterator *si, streamID *current) {
p
=
lpNext
(
lp
,
p
);
/* Seek deleted field. */
aux
=
lpGetInteger
(
p
);
lp
=
lpReplaceInteger
(
lp
,
&
p
,
aux
+
1
);
/* Update the listpack with the new pointer. */
if
(
si
->
lp
!=
lp
)
raxInsert
(
si
->
stream
->
rax
,
si
->
ri
.
key
,
si
->
ri
.
key_len
,
lp
,
NULL
);
}
/* Update the number of entries counter. */
...
...
@@ -739,7 +751,7 @@ void streamIteratorRemoveEntry(streamIterator *si, streamID *current) {
}
/* Stop the stream iterator. The only cleanup we need is to free the rax
* iter
e
ator, since the stream iterator itself is supposed to be stack
* iterator, since the stream iterator itself is supposed to be stack
* allocated. */
void
streamIteratorStop
(
streamIterator
*
si
)
{
raxStop
(
&
si
->
ri
);
...
...
@@ -757,6 +769,7 @@ int streamDeleteItem(stream *s, streamID *id) {
streamIteratorRemoveEntry
(
&
si
,
&
myid
);
deleted
=
1
;
}
streamIteratorStop
(
&
si
);
return
deleted
;
}
...
...
@@ -764,8 +777,8 @@ int streamDeleteItem(stream *s, streamID *id) {
* in the standard <ms>-<seq> format, using the simple string protocol
* of REPL. */
void
addReplyStreamID
(
client
*
c
,
streamID
*
id
)
{
sds
replyid
=
sdscatfmt
(
sdsempty
(),
"
+
%U-%U
\r\n
"
,
id
->
ms
,
id
->
seq
);
addReplySds
(
c
,
replyid
);
sds
replyid
=
sdscatfmt
(
sdsempty
(),
"%U-%U"
,
id
->
ms
,
id
->
seq
);
addReply
Bulk
Sds
(
c
,
replyid
);
}
/* Similar to the above function, but just creates an object, usually useful
...
...
@@ -778,18 +791,18 @@ robj *createObjectFromStreamID(streamID *id) {
/* As a result of an explicit XCLAIM or XREADGROUP command, new entries
* are created in the pending list of the stream and consumers. We need
* to propagate this changes in the form of XCLAIM commands. */
void
streamPropagateXCLAIM
(
client
*
c
,
robj
*
key
,
robj
*
group
,
robj
*
id
,
streamNACK
*
nack
)
{
void
streamPropagateXCLAIM
(
client
*
c
,
robj
*
key
,
streamCG
*
group
,
robj
*
group
name
,
robj
*
id
,
streamNACK
*
nack
)
{
/* We need to generate an XCLAIM that will work in a idempotent fashion:
*
* XCLAIM <key> <group> <consumer> 0 <id> TIME <milliseconds-unix-time>
* RETRYCOUNT <count> FORCE JUSTID.
* RETRYCOUNT <count> FORCE JUSTID
LASTID <id>
.
*
* Note that JUSTID is useful in order to avoid that XCLAIM will do
* useless work in the slave side, trying to fetch the stream item. */
robj
*
argv
[
1
2
];
robj
*
argv
[
1
4
];
argv
[
0
]
=
createStringObject
(
"XCLAIM"
,
6
);
argv
[
1
]
=
key
;
argv
[
2
]
=
group
;
argv
[
2
]
=
group
name
;
argv
[
3
]
=
createStringObject
(
nack
->
consumer
->
name
,
sdslen
(
nack
->
consumer
->
name
));
argv
[
4
]
=
createStringObjectFromLongLong
(
0
);
argv
[
5
]
=
id
;
...
...
@@ -799,7 +812,9 @@ void streamPropagateXCLAIM(client *c, robj *key, robj *group, robj *id, streamNA
argv
[
9
]
=
createStringObjectFromLongLong
(
nack
->
delivery_count
);
argv
[
10
]
=
createStringObject
(
"FORCE"
,
5
);
argv
[
11
]
=
createStringObject
(
"JUSTID"
,
6
);
propagate
(
server
.
xclaimCommand
,
c
->
db
->
id
,
argv
,
12
,
PROPAGATE_AOF
|
PROPAGATE_REPL
);
argv
[
12
]
=
createStringObject
(
"LASTID"
,
6
);
argv
[
13
]
=
createObjectFromStreamID
(
&
group
->
last_id
);
propagate
(
server
.
xclaimCommand
,
c
->
db
->
id
,
argv
,
14
,
PROPAGATE_AOF
|
PROPAGATE_REPL
);
decrRefCount
(
argv
[
0
]);
decrRefCount
(
argv
[
3
]);
decrRefCount
(
argv
[
4
]);
...
...
@@ -809,13 +824,38 @@ void streamPropagateXCLAIM(client *c, robj *key, robj *group, robj *id, streamNA
decrRefCount
(
argv
[
9
]);
decrRefCount
(
argv
[
10
]);
decrRefCount
(
argv
[
11
]);
decrRefCount
(
argv
[
12
]);
decrRefCount
(
argv
[
13
]);
}
/* Send the specified range to the client 'c'. The range the client will
* receive is between start and end inclusive, if 'count' is non zero, no more
* than 'count' elemnets are sent. The 'end' pointer can be NULL to mean that
* we want all the elements from 'start' till the end of the stream. If 'rev'
* is non zero, elements are produced in reversed order from end to start.
/* We need this when we want to propoagate the new last-id of a consumer group
* that was consumed by XREADGROUP with the NOACK option: in that case we can't
* propagate the last ID just using the XCLAIM LASTID option, so we emit
*
* XGROUP SETID <key> <groupname> <id>
*/
void
streamPropagateGroupID
(
client
*
c
,
robj
*
key
,
streamCG
*
group
,
robj
*
groupname
)
{
robj
*
argv
[
5
];
argv
[
0
]
=
createStringObject
(
"XGROUP"
,
6
);
argv
[
1
]
=
createStringObject
(
"SETID"
,
5
);
argv
[
2
]
=
key
;
argv
[
3
]
=
groupname
;
argv
[
4
]
=
createObjectFromStreamID
(
&
group
->
last_id
);
propagate
(
server
.
xgroupCommand
,
c
->
db
->
id
,
argv
,
5
,
PROPAGATE_AOF
|
PROPAGATE_REPL
);
decrRefCount
(
argv
[
0
]);
decrRefCount
(
argv
[
1
]);
decrRefCount
(
argv
[
4
]);
}
/* Send the stream items in the specified range to the client 'c'. The range
* the client will receive is between start and end inclusive, if 'count' is
* non zero, no more than 'count' elements are sent.
*
* The 'end' pointer can be NULL to mean that we want all the elements from
* 'start' till the end of the stream. If 'rev' is non zero, elements are
* produced in reversed order from end to start.
*
* The function returns the number of entries emitted.
*
* If group and consumer are not NULL, the function performs additional work:
* 1. It updates the last delivered ID in the group in case we are
...
...
@@ -827,15 +867,15 @@ void streamPropagateXCLAIM(client *c, robj *key, robj *group, robj *id, streamNA
*
* The behavior may be modified passing non-zero flags:
*
* STREAM_RWR_NOACK: Do not cr
a
ete PEL entries, that is, the point "3" above
* STREAM_RWR_NOACK: Do not cre
a
te PEL entries, that is, the point "3" above
* is not performed.
* STREAM_RWR_RAWENTRIES: Do not emit array boundaries, but just the entries,
* and return the number of entries emitted as usually.
* This is used when the function is just used in order
* to emit data and there is some higher level logic.
*
* The final argument 'spi' (stream propagati
n
o info pointer) is a structure
* filled with information needed to propagte the command execution to AOF
* The final argument 'spi' (stream propagatio
n
info pointer) is a structure
* filled with information needed to propag
a
te the command execution to AOF
* and slaves, in the case a consumer group was passed: we need to generate
* XCLAIM commands to create the pending list into AOF/slaves in that case.
*
...
...
@@ -854,39 +894,41 @@ void streamPropagateXCLAIM(client *c, robj *key, robj *group, robj *id, streamNA
#define STREAM_RWR_NOACK (1<<0)
/* Do not create entries in the PEL. */
#define STREAM_RWR_RAWENTRIES (1<<1)
/* Do not emit protocol for array
boundaries, just the entries. */
#define STREAM_RWR_HISTORY (1<<2)
/* Only serve consumer local PEL. */
size_t
streamReplyWithRange
(
client
*
c
,
stream
*
s
,
streamID
*
start
,
streamID
*
end
,
size_t
count
,
int
rev
,
streamCG
*
group
,
streamConsumer
*
consumer
,
int
flags
,
streamPropInfo
*
spi
)
{
void
*
arraylen_ptr
=
NULL
;
size_t
arraylen
=
0
;
streamIterator
si
;
int64_t
numfields
;
streamID
id
;
int
propagate_last_id
=
0
;
/* If a group was passed, we check if the request is about messages
* never delivered so far (normally this happens when ">" ID is passed).
*
* If instead the client is asking for some history, we serve it
* using a different function, so that we return entries *solely*
* from its own PEL. This ensures each consumer will always and only
* see the history of messages delivered to it and not yet confirmed
/* If the client is asking for some history, we serve it using a
* different function, so that we return entries *solely* from its
* own PEL. This ensures each consumer will always and only see
* the history of messages delivered to it and not yet confirmed
* as delivered. */
if
(
group
&&
streamCompareID
(
start
,
&
group
->
last_id
)
<=
0
)
{
if
(
group
&&
(
flags
&
STREAM_RWR_HISTORY
)
)
{
return
streamReplyWithRangeFromConsumerPEL
(
c
,
s
,
start
,
end
,
count
,
consumer
);
}
if
(
!
(
flags
&
STREAM_RWR_RAWENTRIES
))
arraylen_ptr
=
addDeferred
MultiBulkLength
(
c
);
arraylen_ptr
=
add
Reply
Deferred
Len
(
c
);
streamIteratorStart
(
&
si
,
s
,
start
,
end
,
rev
);
while
(
streamIteratorGetID
(
&
si
,
&
id
,
&
numfields
))
{
/* Update the group last_id if needed. */
if
(
group
&&
streamCompareID
(
&
id
,
&
group
->
last_id
)
>
0
)
if
(
group
&&
streamCompareID
(
&
id
,
&
group
->
last_id
)
>
0
)
{
group
->
last_id
=
id
;
propagate_last_id
=
1
;
}
/* Emit a two elements array for each item. The first is
* the ID, the second is an array of field-value pairs. */
addReply
MultiBulk
Len
(
c
,
2
);
addReply
Array
Len
(
c
,
2
);
addReplyStreamID
(
c
,
&
id
);
addReplyMultiBulkLen
(
c
,
numfields
*
2
);
addReplyMapLen
(
c
,
numfields
);
/* Emit the field-value pairs. */
while
(
numfields
--
)
{
...
...
@@ -940,16 +982,19 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end
/* Propagate as XCLAIM. */
if
(
spi
)
{
robj
*
idarg
=
createObjectFromStreamID
(
&
id
);
streamPropagateXCLAIM
(
c
,
spi
->
keyname
,
spi
->
groupname
,
idarg
,
nack
);
streamPropagateXCLAIM
(
c
,
spi
->
keyname
,
group
,
spi
->
groupname
,
idarg
,
nack
);
decrRefCount
(
idarg
);
}
}
else
{
if
(
propagate_last_id
)
streamPropagateGroupID
(
c
,
spi
->
keyname
,
group
,
spi
->
groupname
);
}
arraylen
++
;
if
(
count
&&
count
==
arraylen
)
break
;
}
streamIteratorStop
(
&
si
);
if
(
arraylen_ptr
)
setDeferred
MultiBulkLength
(
c
,
arraylen_ptr
,
arraylen
);
if
(
arraylen_ptr
)
setDeferred
ArrayLen
(
c
,
arraylen_ptr
,
arraylen
);
return
arraylen
;
}
...
...
@@ -974,25 +1019,25 @@ size_t streamReplyWithRangeFromConsumerPEL(client *c, stream *s, streamID *start
if
(
end
)
streamEncodeID
(
endkey
,
end
);
size_t
arraylen
=
0
;
void
*
arraylen_ptr
=
addDeferred
MultiBulkLength
(
c
);
void
*
arraylen_ptr
=
add
Reply
Deferred
Len
(
c
);
raxStart
(
&
ri
,
consumer
->
pel
);
raxSeek
(
&
ri
,
">="
,
startkey
,
sizeof
(
startkey
));
while
(
raxNext
(
&
ri
)
&&
(
!
count
||
arraylen
<
count
))
{
if
(
end
&&
memcmp
(
ri
.
key
,
end
,
ri
.
key_len
)
>
0
)
break
;
streamID
thisid
;
streamDecodeID
(
ri
.
key
,
&
thisid
);
if
(
streamReplyWithRange
(
c
,
s
,
&
thisid
,
NULL
,
1
,
0
,
NULL
,
NULL
,
if
(
streamReplyWithRange
(
c
,
s
,
&
thisid
,
&
thisid
,
1
,
0
,
NULL
,
NULL
,
STREAM_RWR_RAWENTRIES
,
NULL
)
==
0
)
{
/* Note that we may have a not acknowledged entry in the PEL
* about a message that's no longer here because was removed
* by the user by other means. In that case we signal it emitting
* the ID but then a NULL entry for the fields. */
addReply
MultiBulk
Len
(
c
,
2
);
addReply
Array
Len
(
c
,
2
);
streamID
id
;
streamDecodeID
(
ri
.
key
,
&
id
);
addReplyStreamID
(
c
,
&
id
);
addReply
(
c
,
shared
.
nullmultibulk
);
addReply
NullArray
(
c
);
}
else
{
streamNACK
*
nack
=
ri
.
data
;
nack
->
delivery_time
=
mstime
();
...
...
@@ -1001,7 +1046,7 @@ size_t streamReplyWithRangeFromConsumerPEL(client *c, stream *s, streamID *start
arraylen
++
;
}
raxStop
(
&
ri
);
setDeferred
MultiBulkLength
(
c
,
arraylen_ptr
,
arraylen
);
setDeferred
ArrayLen
(
c
,
arraylen_ptr
,
arraylen
);
return
arraylen
;
}
...
...
@@ -1051,15 +1096,20 @@ int string2ull(const char *s, unsigned long long *value) {
* to the client, otherwise C_OK is returned. The ID may be in incomplete
* form, just stating the milliseconds time part of the stream. In such a case
* the missing part is set according to the value of 'missing_seq' parameter.
*
* The IDs "-" and "+" specify respectively the minimum and maximum IDs
* that can be represented.
* that can be represented. If 'strict' is set to 1, "-" and "+" will be
* treated as an invalid ID.
*
* If 'c' is set to NULL, no reply is sent to the client. */
int
streamParseIDOrReply
(
client
*
c
,
robj
*
o
,
streamID
*
id
,
uint64_t
missing_seq
)
{
int
stream
Generic
ParseIDOrReply
(
client
*
c
,
robj
*
o
,
streamID
*
id
,
uint64_t
missing_seq
,
int
strict
)
{
char
buf
[
128
];
if
(
sdslen
(
o
->
ptr
)
>
sizeof
(
buf
)
-
1
)
goto
invalid
;
memcpy
(
buf
,
o
->
ptr
,
sdslen
(
o
->
ptr
)
+
1
);
if
(
strict
&&
(
buf
[
0
]
==
'-'
||
buf
[
0
]
==
'+'
)
&&
buf
[
1
]
==
'\0'
)
goto
invalid
;
/* Handle the "-" and "+" special cases. */
if
(
buf
[
0
]
==
'-'
&&
buf
[
1
]
==
'\0'
)
{
id
->
ms
=
0
;
...
...
@@ -1088,7 +1138,33 @@ invalid:
return
C_ERR
;
}
/* XADD key [MAXLEN <count>] <ID or *> [field value] [field value] ... */
/* Wrapper for streamGenericParseIDOrReply() with 'strict' argument set to
* 0, to be used when - and + are acceptable IDs. */
int
streamParseIDOrReply
(
client
*
c
,
robj
*
o
,
streamID
*
id
,
uint64_t
missing_seq
)
{
return
streamGenericParseIDOrReply
(
c
,
o
,
id
,
missing_seq
,
0
);
}
/* Wrapper for streamGenericParseIDOrReply() with 'strict' argument set to
* 1, to be used when we want to return an error if the special IDs + or -
* are provided. */
int
streamParseStrictIDOrReply
(
client
*
c
,
robj
*
o
,
streamID
*
id
,
uint64_t
missing_seq
)
{
return
streamGenericParseIDOrReply
(
c
,
o
,
id
,
missing_seq
,
1
);
}
/* We propagate MAXLEN ~ <count> as MAXLEN = <resulting-len-of-stream>
* otherwise trimming is no longer determinsitic on replicas / AOF. */
void
streamRewriteApproxMaxlen
(
client
*
c
,
stream
*
s
,
int
maxlen_arg_idx
)
{
robj
*
maxlen_obj
=
createStringObjectFromLongLong
(
s
->
length
);
robj
*
equal_obj
=
createStringObject
(
"="
,
1
);
rewriteClientCommandArgument
(
c
,
maxlen_arg_idx
,
maxlen_obj
);
rewriteClientCommandArgument
(
c
,
maxlen_arg_idx
-
1
,
equal_obj
);
decrRefCount
(
equal_obj
);
decrRefCount
(
maxlen_obj
);
}
/* XADD key [MAXLEN [~|=] <count>] <ID or *> [field value] [field value] ... */
void
xaddCommand
(
client
*
c
)
{
streamID
id
;
int
id_given
=
0
;
/* Was an ID different than "*" specified? */
...
...
@@ -1108,11 +1184,14 @@ void xaddCommand(client *c) {
* creation. */
break
;
}
else
if
(
!
strcasecmp
(
opt
,
"maxlen"
)
&&
moreargs
)
{
approx_maxlen
=
0
;
char
*
next
=
c
->
argv
[
i
+
1
]
->
ptr
;
/* Check for the form MAXLEN ~ <count>. */
if
(
moreargs
>=
2
&&
next
[
0
]
==
'~'
&&
next
[
1
]
==
'\0'
)
{
approx_maxlen
=
1
;
i
++
;
}
else
if
(
moreargs
>=
2
&&
next
[
0
]
==
'='
&&
next
[
1
]
==
'\0'
)
{
i
++
;
}
if
(
getLongLongFromObjectOrReply
(
c
,
c
->
argv
[
i
+
1
],
&
maxlen
,
NULL
)
!=
C_OK
)
return
;
...
...
@@ -1125,7 +1204,7 @@ void xaddCommand(client *c) {
maxlen_arg_idx
=
i
;
}
else
{
/* If we are here is a syntax error or a valid ID. */
if
(
streamParseIDOrReply
(
c
,
c
->
argv
[
i
],
&
id
,
0
)
!=
C_OK
)
return
;
if
(
streamParse
Strict
IDOrReply
(
c
,
c
->
argv
[
i
],
&
id
,
0
)
!=
C_OK
)
return
;
id_given
=
1
;
break
;
}
...
...
@@ -1159,18 +1238,12 @@ void xaddCommand(client *c) {
notifyKeyspaceEvent
(
NOTIFY_STREAM
,
"xadd"
,
c
->
argv
[
1
],
c
->
db
->
id
);
server
.
dirty
++
;
/* Remove older elements if MAXLEN was specified. */
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
* as zero, so that the command is propagated without trimming. */
robj
*
zeroobj
=
createStringObjectFromLongLong
(
0
);
rewriteClientCommandArgument
(
c
,
maxlen_arg_idx
,
zeroobj
);
decrRefCount
(
zeroobj
);
}
else
{
/* Notify xtrim event if needed. */
if
(
streamTrimByLength
(
s
,
maxlen
,
approx_maxlen
))
{
notifyKeyspaceEvent
(
NOTIFY_STREAM
,
"xtrim"
,
c
->
argv
[
1
],
c
->
db
->
id
);
}
if
(
approx_maxlen
)
streamRewriteApproxMaxlen
(
c
,
s
,
maxlen_arg_idx
);
}
/* Let's rewrite the ID argument with the one actually generated for
...
...
@@ -1190,7 +1263,7 @@ void xrangeGenericCommand(client *c, int rev) {
robj
*
o
;
stream
*
s
;
streamID
startid
,
endid
;
long
long
count
=
0
;
long
long
count
=
-
1
;
robj
*
startarg
=
rev
?
c
->
argv
[
3
]
:
c
->
argv
[
2
];
robj
*
endarg
=
rev
?
c
->
argv
[
2
]
:
c
->
argv
[
3
];
...
...
@@ -1214,10 +1287,17 @@ void xrangeGenericCommand(client *c, int rev) {
}
/* Return the specified range to the user. */
if
((
o
=
lookupKeyReadOrReply
(
c
,
c
->
argv
[
1
],
shared
.
emptymultibulk
))
==
NULL
||
checkType
(
c
,
o
,
OBJ_STREAM
))
return
;
if
((
o
=
lookupKeyReadOrReply
(
c
,
c
->
argv
[
1
],
shared
.
emptyarray
))
==
NULL
||
checkType
(
c
,
o
,
OBJ_STREAM
))
return
;
s
=
o
->
ptr
;
streamReplyWithRange
(
c
,
s
,
&
startid
,
&
endid
,
count
,
rev
,
NULL
,
NULL
,
0
,
NULL
);
if
(
count
==
0
)
{
addReplyNullArray
(
c
);
}
else
{
if
(
count
==
-
1
)
count
=
0
;
streamReplyWithRange
(
c
,
s
,
&
startid
,
&
endid
,
count
,
rev
,
NULL
,
NULL
,
0
,
NULL
);
}
}
/* XRANGE key start end [COUNT <n>] */
...
...
@@ -1381,7 +1461,7 @@ void xreadCommand(client *c) {
ids
[
id_idx
].
seq
=
UINT64_MAX
;
continue
;
}
if
(
streamParseIDOrReply
(
c
,
c
->
argv
[
i
],
ids
+
id_idx
,
0
)
!=
C_OK
)
if
(
streamParse
Strict
IDOrReply
(
c
,
c
->
argv
[
i
],
ids
+
id_idx
,
0
)
!=
C_OK
)
goto
cleanup
;
}
...
...
@@ -1394,8 +1474,10 @@ void xreadCommand(client *c) {
stream
*
s
=
o
->
ptr
;
streamID
*
gt
=
ids
+
i
;
/* ID must be greater than this. */
int
serve_synchronously
=
0
;
int
serve_history
=
0
;
/* True for XREADGROUP with ID != ">". */
/* Check if there are the conditions to serve the client synchronously. */
/* 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
...
...
@@ -1404,12 +1486,13 @@ void xreadCommand(client *c) {
gt
->
seq
!=
UINT64_MAX
)
{
serve_synchronously
=
1
;
serve_history
=
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
)
{
if
(
s
->
length
&&
(
streamCompareID
(
&
s
->
last_id
,
last
)
>
0
)
)
{
serve_synchronously
=
1
;
*
gt
=
*
last
;
}
...
...
@@ -1417,14 +1500,14 @@ void xreadCommand(client *c) {
}
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
)
{
if
(
s
->
length
&&
(
streamCompareID
(
&
s
->
last_id
,
gt
)
>
0
)
)
{
serve_synchronously
=
1
;
}
}
if
(
serve_synchronously
)
{
arraylen
++
;
if
(
arraylen
==
1
)
arraylen_ptr
=
addDeferred
MultiBulkLength
(
c
);
if
(
arraylen
==
1
)
arraylen_ptr
=
add
Reply
Deferred
Len
(
c
);
/* streamReplyWithRange() handles the 'start' ID as inclusive,
* so start from the next ID, since we want only messages with
* IDs greater than start. */
...
...
@@ -1433,22 +1516,28 @@ void xreadCommand(client *c) {
/* Emit the two elements sub-array consisting of the name
* of the stream and the data we extracted from it. */
addReplyMultiBulk
Len
(
c
,
2
);
if
(
c
->
resp
==
2
)
addReplyArray
Len
(
c
,
2
);
addReplyBulk
(
c
,
c
->
argv
[
streams_arg
+
i
]);
streamConsumer
*
consumer
=
NULL
;
if
(
groups
)
consumer
=
streamLookupConsumer
(
groups
[
i
],
consumername
->
ptr
,
1
);
streamPropInfo
spi
=
{
c
->
argv
[
i
+
streams_arg
],
groupname
};
int
flags
=
0
;
if
(
noack
)
flags
|=
STREAM_RWR_NOACK
;
if
(
serve_history
)
flags
|=
STREAM_RWR_HISTORY
;
streamReplyWithRange
(
c
,
s
,
&
start
,
NULL
,
count
,
0
,
groups
?
groups
[
i
]
:
NULL
,
consumer
,
noack
,
&
spi
);
consumer
,
flags
,
&
spi
);
if
(
groups
)
server
.
dirty
++
;
}
}
/* We replied synchronously! Set the top array len and return to caller. */
if
(
arraylen
)
{
setDeferredMultiBulkLength
(
c
,
arraylen_ptr
,
arraylen
);
if
(
c
->
resp
==
2
)
setDeferredArrayLen
(
c
,
arraylen_ptr
,
arraylen
);
else
setDeferredMapLen
(
c
,
arraylen_ptr
,
arraylen
);
goto
cleanup
;
}
...
...
@@ -1457,7 +1546,7 @@ void xreadCommand(client *c) {
/* If we are inside a MULTI/EXEC and the list is empty the only thing
* we can do is treating it as a timeout (even with timeout 0). */
if
(
c
->
flags
&
CLIENT_MULTI
)
{
addReply
(
c
,
shared
.
nullmultibulk
);
addReply
NullArray
(
c
);
goto
cleanup
;
}
blockForKeys
(
c
,
BLOCKED_STREAM
,
c
->
argv
+
streams_arg
,
streams_count
,
...
...
@@ -1486,7 +1575,7 @@ void xreadCommand(client *c) {
/* No BLOCK option, nor any stream we can serve. Reply as with a
* timeout happened. */
addReply
(
c
,
shared
.
nullmultibulk
);
addReply
NullArray
(
c
);
/* Continue to cleanup... */
cleanup:
/* Cleanup. */
...
...
@@ -1613,13 +1702,14 @@ uint64_t streamDelConsumer(streamCG *cg, sds name) {
* Consumer groups commands
* ----------------------------------------------------------------------- */
/* XGROUP CREATE <key> <groupname> <id or $>
* XGROUP SETID <key> <id or $>
/* XGROUP CREATE <key> <groupname> <id or $>
[MKSTREAM]
* XGROUP SETID <key>
<groupname>
<id or $>
* 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."
,
"CREATE <key> <groupname> <id or $> [opt] -- Create a new consumer group."
,
" option MKSTREAM: create the empty stream if it does not exist."
,
"SETID <key> <groupname> <id or $> -- Set the current group ID."
,
"DESTROY <key> <groupname> -- Remove the specified group."
,
"DELCONSUMER <key> <groupname> <consumer> -- Remove the specified consumer."
,
...
...
@@ -1630,13 +1720,39 @@ NULL
sds
grpname
=
NULL
;
streamCG
*
cg
=
NULL
;
char
*
opt
=
c
->
argv
[
1
]
->
ptr
;
/* Subcommand name. */
int
mkstream
=
0
;
robj
*
o
;
/* Lookup the key now, this is common for all the subcommands but HELP. */
/* CREATE has an MKSTREAM option that creates the stream if it
* does not exist. */
if
(
c
->
argc
==
6
&&
!
strcasecmp
(
opt
,
"CREATE"
))
{
if
(
strcasecmp
(
c
->
argv
[
5
]
->
ptr
,
"MKSTREAM"
))
{
addReplySubcommandSyntaxError
(
c
);
return
;
}
mkstream
=
1
;
grpname
=
c
->
argv
[
3
]
->
ptr
;
}
/* Everything but the "HELP" option requires a key and group name. */
if
(
c
->
argc
>=
4
)
{
robj
*
o
=
lookupKeyWriteOrReply
(
c
,
c
->
argv
[
2
],
shared
.
nokeyerr
);
if
(
o
==
NULL
||
checkType
(
c
,
o
,
OBJ_STREAM
))
return
;
s
=
o
->
ptr
;
o
=
lookupKeyWrite
(
c
->
db
,
c
->
argv
[
2
]);
if
(
o
)
s
=
o
->
ptr
;
grpname
=
c
->
argv
[
3
]
->
ptr
;
}
/* Check for missing key/group. */
if
(
c
->
argc
>=
4
&&
!
mkstream
)
{
/* At this point key must exist, or there is an error. */
if
(
o
==
NULL
)
{
addReplyError
(
c
,
"The XGROUP subcommand requires the key to exist. "
"Note that for CREATE you may want to use the MKSTREAM "
"option to create an empty stream automatically."
);
return
;
}
if
(
checkType
(
c
,
o
,
OBJ_STREAM
))
return
;
/* Certain subcommands require the group to exist. */
if
((
cg
=
streamLookupCG
(
s
,
grpname
))
==
NULL
&&
...
...
@@ -1651,13 +1767,26 @@ NULL
}
/* Dispatch the different subcommands. */
if
(
!
strcasecmp
(
opt
,
"CREATE"
)
&&
c
->
argc
==
5
)
{
if
(
!
strcasecmp
(
opt
,
"CREATE"
)
&&
(
c
->
argc
==
5
||
c
->
argc
==
6
)
)
{
streamID
id
;
if
(
!
strcmp
(
c
->
argv
[
4
]
->
ptr
,
"$"
))
{
id
=
s
->
last_id
;
}
else
if
(
streamParseIDOrReply
(
c
,
c
->
argv
[
4
],
&
id
,
0
)
!=
C_OK
)
{
if
(
s
)
{
id
=
s
->
last_id
;
}
else
{
id
.
ms
=
0
;
id
.
seq
=
0
;
}
}
else
if
(
streamParseStrictIDOrReply
(
c
,
c
->
argv
[
4
],
&
id
,
0
)
!=
C_OK
)
{
return
;
}
/* Handle the MKSTREAM option now that the command can no longer fail. */
if
(
s
==
NULL
&&
mkstream
)
{
o
=
createStreamObject
();
dbAdd
(
c
->
db
,
c
->
argv
[
2
],
o
);
s
=
o
->
ptr
;
}
streamCG
*
cg
=
streamCreateCG
(
s
,
grpname
,
sdslen
(
grpname
),
&
id
);
if
(
cg
)
{
addReply
(
c
,
shared
.
ok
);
...
...
@@ -1705,6 +1834,40 @@ NULL
}
}
/* XSETID <stream> <groupname> <id>
*
* Set the internal "last ID" of a stream. */
void
xsetidCommand
(
client
*
c
)
{
robj
*
o
=
lookupKeyWriteOrReply
(
c
,
c
->
argv
[
1
],
shared
.
nokeyerr
);
if
(
o
==
NULL
||
checkType
(
c
,
o
,
OBJ_STREAM
))
return
;
stream
*
s
=
o
->
ptr
;
streamID
id
;
if
(
streamParseStrictIDOrReply
(
c
,
c
->
argv
[
2
],
&
id
,
0
)
!=
C_OK
)
return
;
/* If the stream has at least one item, we want to check that the user
* is setting a last ID that is equal or greater than the current top
* item, otherwise the fundamental ID monotonicity assumption is violated. */
if
(
s
->
length
>
0
)
{
streamID
maxid
;
streamIterator
si
;
streamIteratorStart
(
&
si
,
s
,
NULL
,
NULL
,
1
);
int64_t
numfields
;
streamIteratorGetID
(
&
si
,
&
maxid
,
&
numfields
);
streamIteratorStop
(
&
si
);
if
(
streamCompareID
(
&
id
,
&
maxid
)
<
0
)
{
addReplyError
(
c
,
"The ID specified in XSETID is smaller than the "
"target stream top item"
);
return
;
}
}
s
->
last_id
=
id
;
addReply
(
c
,
shared
.
ok
);
server
.
dirty
++
;
notifyKeyspaceEvent
(
NOTIFY_STREAM
,
"xsetid"
,
c
->
argv
[
1
],
c
->
db
->
id
);
}
/* XACK <key> <group> <id> <id> ... <id>
*
* Acknowledge a message as processed. In practical terms we just check the
...
...
@@ -1732,7 +1895,7 @@ void xackCommand(client *c) {
for
(
int
j
=
3
;
j
<
c
->
argc
;
j
++
)
{
streamID
id
;
unsigned
char
buf
[
sizeof
(
streamID
)];
if
(
streamParseIDOrReply
(
c
,
c
->
argv
[
j
],
&
id
,
0
)
!=
C_OK
)
return
;
if
(
streamParse
Strict
IDOrReply
(
c
,
c
->
argv
[
j
],
&
id
,
0
)
!=
C_OK
)
return
;
streamEncodeID
(
buf
,
&
id
);
/* Lookup the ID in the group PEL: it will have a reference to the
...
...
@@ -1750,7 +1913,7 @@ void xackCommand(client *c) {
addReplyLongLong
(
c
,
acknowledged
);
}
/* XPENDING <key> <group> [<start> <stop> <count>
]
[<consumer>]
/* XPENDING <key> <group> [<start> <stop> <count> [<consumer>]
]
*
* If start and stop are omitted, the command just outputs information about
* the amount of pending messages for the key/group pair, together with
...
...
@@ -1779,6 +1942,7 @@ void xpendingCommand(client *c) {
if
(
c
->
argc
>=
6
)
{
if
(
getLongLongFromObjectOrReply
(
c
,
c
->
argv
[
5
],
&
count
,
NULL
)
==
C_ERR
)
return
;
if
(
count
<
0
)
count
=
0
;
if
(
streamParseIDOrReply
(
c
,
c
->
argv
[
3
],
&
startid
,
0
)
==
C_ERR
)
return
;
if
(
streamParseIDOrReply
(
c
,
c
->
argv
[
4
],
&
endid
,
UINT64_MAX
)
==
C_ERR
)
...
...
@@ -1801,14 +1965,14 @@ void xpendingCommand(client *c) {
/* XPENDING <key> <group> variant. */
if
(
justinfo
)
{
addReply
MultiBulk
Len
(
c
,
4
);
addReply
Array
Len
(
c
,
4
);
/* Total number of messages in the PEL. */
addReplyLongLong
(
c
,
raxSize
(
group
->
pel
));
/* First and last IDs. */
if
(
raxSize
(
group
->
pel
)
==
0
)
{
addReply
(
c
,
shared
.
nullbulk
);
/* Start. */
addReply
(
c
,
shared
.
nullbulk
);
/* End. */
addReply
(
c
,
shared
.
nullmultibulk
);
/* Clients. */
addReply
Null
(
c
);
/* Start. */
addReply
Null
(
c
);
/* End. */
addReply
NullArray
(
c
);
/* Clients. */
}
else
{
/* Start. */
raxIterator
ri
;
...
...
@@ -1828,17 +1992,17 @@ void xpendingCommand(client *c) {
/* Consumers with pending messages. */
raxStart
(
&
ri
,
group
->
consumers
);
raxSeek
(
&
ri
,
"^"
,
NULL
,
0
);
void
*
arraylen_ptr
=
addDeferred
MultiBulkLength
(
c
);
void
*
arraylen_ptr
=
add
Reply
Deferred
Len
(
c
);
size_t
arraylen
=
0
;
while
(
raxNext
(
&
ri
))
{
streamConsumer
*
consumer
=
ri
.
data
;
if
(
raxSize
(
consumer
->
pel
)
==
0
)
continue
;
addReply
MultiBulk
Len
(
c
,
2
);
addReply
Array
Len
(
c
,
2
);
addReplyBulkCBuffer
(
c
,
ri
.
key
,
ri
.
key_len
);
addReplyBulkLongLong
(
c
,
raxSize
(
consumer
->
pel
));
arraylen
++
;
}
setDeferred
MultiBulkLength
(
c
,
arraylen_ptr
,
arraylen
);
setDeferred
ArrayLen
(
c
,
arraylen_ptr
,
arraylen
);
raxStop
(
&
ri
);
}
}
...
...
@@ -1851,7 +2015,7 @@ 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
)
{
addReply
MultiBulk
Len
(
c
,
0
);
addReply
Array
Len
(
c
,
0
);
return
;
}
...
...
@@ -1865,7 +2029,7 @@ void xpendingCommand(client *c) {
streamEncodeID
(
endkey
,
&
endid
);
raxStart
(
&
ri
,
pel
);
raxSeek
(
&
ri
,
">="
,
startkey
,
sizeof
(
startkey
));
void
*
arraylen_ptr
=
addDeferred
MultiBulkLength
(
c
);
void
*
arraylen_ptr
=
add
Reply
Deferred
Len
(
c
);
size_t
arraylen
=
0
;
while
(
count
&&
raxNext
(
&
ri
)
&&
memcmp
(
ri
.
key
,
endkey
,
ri
.
key_len
)
<=
0
)
{
...
...
@@ -1873,7 +2037,7 @@ void xpendingCommand(client *c) {
arraylen
++
;
count
--
;
addReply
MultiBulk
Len
(
c
,
4
);
addReply
Array
Len
(
c
,
4
);
/* Entry ID. */
streamID
id
;
...
...
@@ -1893,7 +2057,7 @@ void xpendingCommand(client *c) {
addReplyLongLong
(
c
,
nack
->
delivery_count
);
}
raxStop
(
&
ri
);
setDeferred
MultiBulkLength
(
c
,
arraylen_ptr
,
arraylen
);
setDeferred
ArrayLen
(
c
,
arraylen_ptr
,
arraylen
);
}
}
...
...
@@ -1952,6 +2116,14 @@ void xpendingCommand(client *c) {
* Return just an array of IDs of messages successfully claimed,
* without returning the actual message.
*
* 6. LASTID <id>:
* Update the consumer group last ID with the specified ID if the
* current last ID is smaller than the provided one.
* This is used for replication / AOF, so that when we read from a
* consumer group, the XCLAIM that gets propagated to give ownership
* to the consumer, is also used in order to update the group current
* ID.
*
* The command returns an array of messages that the user
* successfully claimed, so that the caller is able to understand
* what messages it is now in charge of. */
...
...
@@ -1988,15 +2160,17 @@ void xclaimCommand(client *c) {
* the client successfully claimed some message, so it should be
* executed in a "all or nothing" fashion. */
int
j
;
for
(
j
=
4
;
j
<
c
->
argc
;
j
++
)
{
for
(
j
=
5
;
j
<
c
->
argc
;
j
++
)
{
streamID
id
;
if
(
streamParseIDOrReply
(
NULL
,
c
->
argv
[
j
],
&
id
,
0
)
!=
C_OK
)
break
;
if
(
streamParse
Strict
IDOrReply
(
NULL
,
c
->
argv
[
j
],
&
id
,
0
)
!=
C_OK
)
break
;
}
int
last_id_arg
=
j
-
1
;
/* Next time we iterate the IDs we now the range. */
/* If we stopped because some IDs cannot be parsed, perhaps they
* are trailing options. */
time_t
now
=
mstime
();
mstime_t
now
=
mstime
();
streamID
last_id
=
{
0
,
0
};
int
propagate_last_id
=
0
;
for
(;
j
<
c
->
argc
;
j
++
)
{
int
moreargs
=
(
c
->
argc
-
1
)
-
j
;
/* Number of additional arguments. */
char
*
opt
=
c
->
argv
[
j
]
->
ptr
;
...
...
@@ -2013,27 +2187,35 @@ void xclaimCommand(client *c) {
}
else
if
(
!
strcasecmp
(
opt
,
"TIME"
)
&&
moreargs
)
{
j
++
;
if
(
getLongLongFromObjectOrReply
(
c
,
c
->
argv
[
j
],
&
deliverytime
,
"Invalid
IDL
E option argument for XCLAIM"
)
"Invalid
TIM
E option argument for XCLAIM"
)
!=
C_OK
)
return
;
}
else
if
(
!
strcasecmp
(
opt
,
"RETRYCOUNT"
)
&&
moreargs
)
{
j
++
;
if
(
getLongLongFromObjectOrReply
(
c
,
c
->
argv
[
j
],
&
retrycount
,
"Invalid
IDLE
option argument for XCLAIM"
)
"Invalid
RETRYCOUNT
option argument for XCLAIM"
)
!=
C_OK
)
return
;
}
else
if
(
!
strcasecmp
(
opt
,
"LASTID"
)
&&
moreargs
)
{
j
++
;
if
(
streamParseStrictIDOrReply
(
c
,
c
->
argv
[
j
],
&
last_id
,
0
)
!=
C_OK
)
return
;
}
else
{
addReplyErrorFormat
(
c
,
"Unrecognized XCLAIM option '%s'"
,
opt
);
return
;
}
}
if
(
streamCompareID
(
&
last_id
,
&
group
->
last_id
)
>
0
)
{
group
->
last_id
=
last_id
;
propagate_last_id
=
1
;
}
if
(
deliverytime
!=
-
1
)
{
/* If a delivery time was passed, either with IDLE or TIME, we
* do some sanity check on it, and set the deliverytime to now
* (which is a sane choice usually) if the value is bogus.
* To raise an error here is not wise because clients may compute
* the idle time doing some math startin from their local time,
* the idle time doing some math startin
g
from their local time,
* and this is not a good excuse to fail in case, for instance,
* the compute
d
time is a bit in the future from our POV. */
* the compute
r
time is a bit in the future from our POV. */
if
(
deliverytime
<
0
||
deliverytime
>
now
)
deliverytime
=
now
;
}
else
{
/* If no IDLE/TIME option was passed, we want the last delivery
...
...
@@ -2044,12 +2226,13 @@ void xclaimCommand(client *c) {
/* Do the actual claiming. */
streamConsumer
*
consumer
=
streamLookupConsumer
(
group
,
c
->
argv
[
3
]
->
ptr
,
1
);
void
*
arraylenptr
=
addDeferred
MultiBulkLength
(
c
);
void
*
arraylenptr
=
add
Reply
Deferred
Len
(
c
);
size_t
arraylen
=
0
;
for
(
int
j
=
5
;
j
<=
last_id_arg
;
j
++
)
{
streamID
id
;
unsigned
char
buf
[
sizeof
(
streamID
)];
if
(
streamParseIDOrReply
(
c
,
c
->
argv
[
j
],
&
id
,
0
)
!=
C_OK
)
return
;
if
(
streamParseStrictIDOrReply
(
c
,
c
->
argv
[
j
],
&
id
,
0
)
!=
C_OK
)
serverPanic
(
"StreamID invalid after check. Should not be possible."
);
streamEncodeID
(
buf
,
&
id
);
/* Lookup the ID in the group PEL. */
...
...
@@ -2079,8 +2262,12 @@ void xclaimCommand(client *c) {
if
(
nack
!=
raxNotFound
)
{
/* We need to check if the minimum idle time requested
* by the caller is satisfied by this entry. */
if
(
minidle
)
{
* by the caller is satisfied by this entry.
*
* Note that the nack could be created by FORCE, in this
* case there was no pre-existing entry and minidle should
* be ignored, but in that case nick->consumer is NULL. */
if
(
nack
->
consumer
&&
minidle
)
{
mstime_t
this_idle
=
now
-
nack
->
delivery_time
;
if
(
this_idle
<
minidle
)
continue
;
}
...
...
@@ -2100,17 +2287,23 @@ void xclaimCommand(client *c) {
if
(
justid
)
{
addReplyStreamID
(
c
,
&
id
);
}
else
{
streamReplyWithRange
(
c
,
o
->
ptr
,
&
id
,
NULL
,
1
,
0
,
NULL
,
NULL
,
STREAM_RWR_RAWENTRIES
,
NULL
);
size_t
emitted
=
streamReplyWithRange
(
c
,
o
->
ptr
,
&
id
,
&
id
,
1
,
0
,
NULL
,
NULL
,
STREAM_RWR_RAWENTRIES
,
NULL
);
if
(
!
emitted
)
addReplyNull
(
c
);
}
arraylen
++
;
/* Propagate this change. */
streamPropagateXCLAIM
(
c
,
c
->
argv
[
1
],
c
->
argv
[
3
],
c
->
argv
[
j
],
nack
);
streamPropagateXCLAIM
(
c
,
c
->
argv
[
1
],
group
,
c
->
argv
[
2
],
c
->
argv
[
j
],
nack
);
propagate_last_id
=
0
;
/* Will be propagated by XCLAIM itself. */
server
.
dirty
++
;
}
}
setDeferredMultiBulkLength
(
c
,
arraylenptr
,
arraylen
);
if
(
propagate_last_id
)
{
streamPropagateGroupID
(
c
,
c
->
argv
[
1
],
group
,
c
->
argv
[
2
]);
server
.
dirty
++
;
}
setDeferredArrayLen
(
c
,
arraylenptr
,
arraylen
);
preventCommandPropagation
(
c
);
}
...
...
@@ -2132,13 +2325,13 @@ void xdelCommand(client *c) {
* 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
;
if
(
streamParse
Strict
IDOrReply
(
c
,
c
->
argv
[
j
],
&
id
,
0
)
!=
C_OK
)
return
;
}
/* 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. */
streamParse
Strict
IDOrReply
(
c
,
c
->
argv
[
j
],
&
id
,
0
);
/* Retval already checked. */
deleted
+=
streamDeleteItem
(
s
,
&
id
);
}
...
...
@@ -2155,7 +2348,7 @@ void xdelCommand(client *c) {
*
* List of options:
*
* MAXLEN [~] <count>
-- Trim so that the stream will be capped at
* MAXLEN [~
|=
] <count> -- Trim so that the stream will be capped at
* the specified length. Use ~ before the
* count in order to demand approximated trimming
* (like XADD MAXLEN option).
...
...
@@ -2174,9 +2367,10 @@ void xtrimCommand(client *c) {
/* Argument parsing. */
int
trim_strategy
=
TRIM_STRATEGY_NONE
;
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. */
/* Parse options. */
int
i
=
2
;
/* Start of options. */
...
...
@@ -2184,16 +2378,25 @@ void xtrimCommand(client *c) {
int
moreargs
=
(
c
->
argc
-
1
)
-
i
;
/* Number of additional arguments. */
char
*
opt
=
c
->
argv
[
i
]
->
ptr
;
if
(
!
strcasecmp
(
opt
,
"maxlen"
)
&&
moreargs
)
{
approx_maxlen
=
0
;
trim_strategy
=
TRIM_STRATEGY_MAXLEN
;
char
*
next
=
c
->
argv
[
i
+
1
]
->
ptr
;
/* Check for the form MAXLEN ~ <count>. */
if
(
moreargs
>=
2
&&
next
[
0
]
==
'~'
&&
next
[
1
]
==
'\0'
)
{
approx_maxlen
=
1
;
i
++
;
}
else
if
(
moreargs
>=
2
&&
next
[
0
]
==
'='
&&
next
[
1
]
==
'\0'
)
{
i
++
;
}
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
{
addReply
(
c
,
shared
.
syntaxerr
);
return
;
...
...
@@ -2214,11 +2417,12 @@ void xtrimCommand(client *c) {
signalModifiedKey
(
c
->
db
,
c
->
argv
[
1
]);
notifyKeyspaceEvent
(
NOTIFY_STREAM
,
"xtrim"
,
c
->
argv
[
1
],
c
->
db
->
id
);
server
.
dirty
+=
deleted
;
if
(
approx_maxlen
)
streamRewriteApproxMaxlen
(
c
,
s
,
maxlen_arg_idx
);
}
addReplyLongLong
(
c
,
deleted
);
}
/* XINFO CONSUMERS key
group
/* XINFO CONSUMERS
<
key
> <
group
>
* XINFO GROUPS <key>
* XINFO STREAM <key>
* XINFO HELP. */
...
...
@@ -2264,7 +2468,7 @@ NULL
return
;
}
addReply
MultiBulk
Len
(
c
,
raxSize
(
cg
->
consumers
));
addReply
Array
Len
(
c
,
raxSize
(
cg
->
consumers
));
raxIterator
ri
;
raxStart
(
&
ri
,
cg
->
consumers
);
raxSeek
(
&
ri
,
"^"
,
NULL
,
0
);
...
...
@@ -2274,51 +2478,51 @@ NULL
mstime_t
idle
=
now
-
consumer
->
seen_time
;
if
(
idle
<
0
)
idle
=
0
;
addReplyM
ultiBulk
Len
(
c
,
6
);
addReply
Status
(
c
,
"name"
);
addReplyM
ap
Len
(
c
,
3
);
addReply
BulkCString
(
c
,
"name"
);
addReplyBulkCBuffer
(
c
,
consumer
->
name
,
sdslen
(
consumer
->
name
));
addReply
Status
(
c
,
"pending"
);
addReply
BulkCString
(
c
,
"pending"
);
addReplyLongLong
(
c
,
raxSize
(
consumer
->
pel
));
addReply
Status
(
c
,
"idle"
);
addReply
BulkCString
(
c
,
"idle"
);
addReplyLongLong
(
c
,
idle
);
}
raxStop
(
&
ri
);
}
else
if
(
!
strcasecmp
(
opt
,
"GROUPS"
)
&&
c
->
argc
==
3
)
{
/* XINFO GROUPS <key>. */
if
(
s
->
cgroups
==
NULL
)
{
addReply
MultiBulk
Len
(
c
,
0
);
addReply
Array
Len
(
c
,
0
);
return
;
}
addReply
MultiBulk
Len
(
c
,
raxSize
(
s
->
cgroups
));
addReply
Array
Len
(
c
,
raxSize
(
s
->
cgroups
));
raxIterator
ri
;
raxStart
(
&
ri
,
s
->
cgroups
);
raxSeek
(
&
ri
,
"^"
,
NULL
,
0
);
while
(
raxNext
(
&
ri
))
{
streamCG
*
cg
=
ri
.
data
;
addReplyM
ultiBulk
Len
(
c
,
8
);
addReply
Status
(
c
,
"name"
);
addReplyM
ap
Len
(
c
,
4
);
addReply
BulkCString
(
c
,
"name"
);
addReplyBulkCBuffer
(
c
,
ri
.
key
,
ri
.
key_len
);
addReply
Status
(
c
,
"consumers"
);
addReply
BulkCString
(
c
,
"consumers"
);
addReplyLongLong
(
c
,
raxSize
(
cg
->
consumers
));
addReply
Status
(
c
,
"pending"
);
addReply
BulkCString
(
c
,
"pending"
);
addReplyLongLong
(
c
,
raxSize
(
cg
->
pel
));
addReply
Status
(
c
,
"last-delivered-id"
);
addReply
BulkCString
(
c
,
"last-delivered-id"
);
addReplyStreamID
(
c
,
&
cg
->
last_id
);
}
raxStop
(
&
ri
);
}
else
if
(
!
strcasecmp
(
opt
,
"STREAM"
)
&&
c
->
argc
==
3
)
{
/* XINFO STREAM <key> (or the alias XINFO <key>). */
addReplyM
ultiBulk
Len
(
c
,
14
);
addReply
Status
(
c
,
"length"
);
addReplyM
ap
Len
(
c
,
7
);
addReply
BulkCString
(
c
,
"length"
);
addReplyLongLong
(
c
,
s
->
length
);
addReply
Status
(
c
,
"radix-tree-keys"
);
addReply
BulkCString
(
c
,
"radix-tree-keys"
);
addReplyLongLong
(
c
,
raxSize
(
s
->
rax
));
addReply
Status
(
c
,
"radix-tree-nodes"
);
addReply
BulkCString
(
c
,
"radix-tree-nodes"
);
addReplyLongLong
(
c
,
s
->
rax
->
numnodes
);
addReply
Status
(
c
,
"groups"
);
addReply
BulkCString
(
c
,
"groups"
);
addReplyLongLong
(
c
,
s
->
cgroups
?
raxSize
(
s
->
cgroups
)
:
0
);
addReply
Status
(
c
,
"last-generated-id"
);
addReply
BulkCString
(
c
,
"last-generated-id"
);
addReplyStreamID
(
c
,
&
s
->
last_id
);
/* To emit the first/last entry we us the streamReplyWithRange()
...
...
@@ -2327,14 +2531,14 @@ NULL
streamID
start
,
end
;
start
.
ms
=
start
.
seq
=
0
;
end
.
ms
=
end
.
seq
=
UINT64_MAX
;
addReply
Status
(
c
,
"first-entry"
);
addReply
BulkCString
(
c
,
"first-entry"
);
count
=
streamReplyWithRange
(
c
,
s
,
&
start
,
&
end
,
1
,
0
,
NULL
,
NULL
,
STREAM_RWR_RAWENTRIES
,
NULL
);
if
(
!
count
)
addReply
(
c
,
shared
.
nullbulk
);
addReply
Status
(
c
,
"last-entry"
);
if
(
!
count
)
addReply
Null
(
c
);
addReply
BulkCString
(
c
,
"last-entry"
);
count
=
streamReplyWithRange
(
c
,
s
,
&
start
,
&
end
,
1
,
1
,
NULL
,
NULL
,
STREAM_RWR_RAWENTRIES
,
NULL
);
if
(
!
count
)
addReply
(
c
,
shared
.
nullbulk
);
if
(
!
count
)
addReply
Null
(
c
);
}
else
{
addReplySubcommandSyntaxError
(
c
);
}
...
...
src/t_string.c
View file @
49816941
...
...
@@ -80,7 +80,7 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire,
if
((
flags
&
OBJ_SET_NX
&&
lookupKeyWrite
(
c
->
db
,
key
)
!=
NULL
)
||
(
flags
&
OBJ_SET_XX
&&
lookupKeyWrite
(
c
->
db
,
key
)
==
NULL
))
{
addReply
(
c
,
abort_reply
?
abort_reply
:
shared
.
null
bulk
);
addReply
(
c
,
abort_reply
?
abort_reply
:
shared
.
null
[
c
->
resp
]
);
return
;
}
setKey
(
c
->
db
,
key
,
val
);
...
...
@@ -157,7 +157,7 @@ void psetexCommand(client *c) {
int
getGenericCommand
(
client
*
c
)
{
robj
*
o
;
if
((
o
=
lookupKeyReadOrReply
(
c
,
c
->
argv
[
1
],
shared
.
null
bulk
))
==
NULL
)
if
((
o
=
lookupKeyReadOrReply
(
c
,
c
->
argv
[
1
],
shared
.
null
[
c
->
resp
]
))
==
NULL
)
return
C_OK
;
if
(
o
->
type
!=
OBJ_STRING
)
{
...
...
@@ -285,14 +285,14 @@ void getrangeCommand(client *c) {
void
mgetCommand
(
client
*
c
)
{
int
j
;
addReply
MultiBulk
Len
(
c
,
c
->
argc
-
1
);
addReply
Array
Len
(
c
,
c
->
argc
-
1
);
for
(
j
=
1
;
j
<
c
->
argc
;
j
++
)
{
robj
*
o
=
lookupKeyRead
(
c
->
db
,
c
->
argv
[
j
]);
if
(
o
==
NULL
)
{
addReply
(
c
,
shared
.
nullbulk
);
addReply
Null
(
c
);
}
else
{
if
(
o
->
type
!=
OBJ_STRING
)
{
addReply
(
c
,
shared
.
nullbulk
);
addReply
Null
(
c
);
}
else
{
addReplyBulk
(
c
,
o
);
}
...
...
@@ -301,24 +301,22 @@ void mgetCommand(client *c) {
}
void
msetGenericCommand
(
client
*
c
,
int
nx
)
{
int
j
,
busykeys
=
0
;
int
j
;
if
((
c
->
argc
%
2
)
==
0
)
{
addReplyError
(
c
,
"wrong number of arguments for MSET"
);
return
;
}
/* Handle the NX flag. The MSETNX semantic is to return zero and don't
* set
no
thing
at all
if at least one
already ke
y exists. */
* set
any
thing if at least one
key alerad
y exists. */
if
(
nx
)
{
for
(
j
=
1
;
j
<
c
->
argc
;
j
+=
2
)
{
if
(
lookupKeyWrite
(
c
->
db
,
c
->
argv
[
j
])
!=
NULL
)
{
busykeys
++
;
addReply
(
c
,
shared
.
czero
);
return
;
}
}
if
(
busykeys
)
{
addReply
(
c
,
shared
.
czero
);
return
;
}
}
for
(
j
=
1
;
j
<
c
->
argc
;
j
+=
2
)
{
...
...
src/t_zset.c
View file @
49816941
...
...
@@ -244,6 +244,61 @@ int zslDelete(zskiplist *zsl, double score, sds ele, zskiplistNode **node) {
return
0
;
/* not found */
}
/* Update the score of an elmenent inside the sorted set skiplist.
* Note that the element must exist and must match 'score'.
* This function does not update the score in the hash table side, the
* caller should take care of it.
*
* Note that this function attempts to just update the node, in case after
* the score update, the node would be exactly at the same position.
* Otherwise the skiplist is modified by removing and re-adding a new
* element, which is more costly.
*
* The function returns the updated element skiplist node pointer. */
zskiplistNode
*
zslUpdateScore
(
zskiplist
*
zsl
,
double
curscore
,
sds
ele
,
double
newscore
)
{
zskiplistNode
*
update
[
ZSKIPLIST_MAXLEVEL
],
*
x
;
int
i
;
/* We need to seek to element to update to start: this is useful anyway,
* we'll have to update or remove it. */
x
=
zsl
->
header
;
for
(
i
=
zsl
->
level
-
1
;
i
>=
0
;
i
--
)
{
while
(
x
->
level
[
i
].
forward
&&
(
x
->
level
[
i
].
forward
->
score
<
curscore
||
(
x
->
level
[
i
].
forward
->
score
==
curscore
&&
sdscmp
(
x
->
level
[
i
].
forward
->
ele
,
ele
)
<
0
)))
{
x
=
x
->
level
[
i
].
forward
;
}
update
[
i
]
=
x
;
}
/* Jump to our element: note that this function assumes that the
* element with the matching score exists. */
x
=
x
->
level
[
0
].
forward
;
serverAssert
(
x
&&
curscore
==
x
->
score
&&
sdscmp
(
x
->
ele
,
ele
)
==
0
);
/* If the node, after the score update, would be still exactly
* at the same position, we can just update the score without
* actually removing and re-inserting the element in the skiplist. */
if
((
x
->
backward
==
NULL
||
x
->
backward
->
score
<
newscore
)
&&
(
x
->
level
[
0
].
forward
==
NULL
||
x
->
level
[
0
].
forward
->
score
>
newscore
))
{
x
->
score
=
newscore
;
return
x
;
}
/* No way to reuse the old node: we need to remove and insert a new
* one at a different place. */
zslDeleteNode
(
zsl
,
x
,
update
);
zskiplistNode
*
newnode
=
zslInsert
(
zsl
,
newscore
,
x
->
ele
);
/* We reused the old node x->ele SDS string, free the node now
* since zslInsert created a new one. */
x
->
ele
=
NULL
;
zslFreeNode
(
x
);
return
newnode
;
}
int
zslValueGteMin
(
double
value
,
zrangespec
*
spec
)
{
return
spec
->
minex
?
(
value
>
spec
->
min
)
:
(
value
>=
spec
->
min
);
}
...
...
@@ -519,12 +574,12 @@ int zslParseLexRangeItem(robj *item, sds *dest, int *ex) {
switch
(
c
[
0
])
{
case
'+'
:
if
(
c
[
1
]
!=
'\0'
)
return
C_ERR
;
*
ex
=
0
;
*
ex
=
1
;
*
dest
=
shared
.
maxstring
;
return
C_OK
;
case
'-'
:
if
(
c
[
1
]
!=
'\0'
)
return
C_ERR
;
*
ex
=
0
;
*
ex
=
1
;
*
dest
=
shared
.
minstring
;
return
C_OK
;
case
'('
:
...
...
@@ -597,9 +652,8 @@ int zslIsInLexRange(zskiplist *zsl, zlexrangespec *range) {
zskiplistNode
*
x
;
/* Test for ranges that will always be empty. */
if
(
sdscmplex
(
range
->
min
,
range
->
max
)
>
1
||
(
sdscmp
(
range
->
min
,
range
->
max
)
==
0
&&
(
range
->
minex
||
range
->
maxex
)))
int
cmp
=
sdscmplex
(
range
->
min
,
range
->
max
);
if
(
cmp
>
0
||
(
cmp
==
0
&&
(
range
->
minex
||
range
->
maxex
)))
return
0
;
x
=
zsl
->
tail
;
if
(
x
==
NULL
||
!
zslLexValueGteMin
(
x
->
ele
,
range
))
...
...
@@ -872,9 +926,8 @@ int zzlIsInLexRange(unsigned char *zl, zlexrangespec *range) {
unsigned
char
*
p
;
/* Test for ranges that will always be empty. */
if
(
sdscmplex
(
range
->
min
,
range
->
max
)
>
1
||
(
sdscmp
(
range
->
min
,
range
->
max
)
==
0
&&
(
range
->
minex
||
range
->
maxex
)))
int
cmp
=
sdscmplex
(
range
->
min
,
range
->
max
);
if
(
cmp
>
0
||
(
cmp
==
0
&&
(
range
->
minex
||
range
->
maxex
)))
return
0
;
p
=
ziplistIndex
(
zl
,
-
2
);
/* Last element. */
...
...
@@ -1341,13 +1394,7 @@ int zsetAdd(robj *zobj, double score, sds ele, int *flags, double *newscore) {
/* Remove and re-insert when score changes. */
if
(
score
!=
curscore
)
{
zskiplistNode
*
node
;
serverAssert
(
zslDelete
(
zs
->
zsl
,
curscore
,
ele
,
&
node
));
znode
=
zslInsert
(
zs
->
zsl
,
score
,
node
->
ele
);
/* We reused the node->ele SDS string, free the node now
* since zslInsert created a new one. */
node
->
ele
=
NULL
;
zslFreeNode
(
node
);
znode
=
zslUpdateScore
(
zs
->
zsl
,
curscore
,
ele
,
score
);
/* Note that we did not removed the original element from
* the hash table representing the sorted set, so we just
* update the score. */
...
...
@@ -1591,7 +1638,7 @@ reply_to_client:
if
(
processed
)
addReplyDouble
(
c
,
score
);
else
addReply
(
c
,
shared
.
nullbulk
);
addReply
Null
(
c
);
}
else
{
/* ZADD. */
addReplyLongLong
(
c
,
ch
?
added
+
updated
:
added
);
}
...
...
@@ -2380,7 +2427,7 @@ void zrangeGenericCommand(client *c, int reverse) {
return
;
}
if
((
zobj
=
lookupKeyReadOrReply
(
c
,
key
,
shared
.
emptymultibulk
))
==
NULL
if
((
zobj
=
lookupKeyReadOrReply
(
c
,
key
,
shared
.
null
[
c
->
resp
]
))
==
NULL
||
checkType
(
c
,
zobj
,
OBJ_ZSET
))
return
;
/* Sanitize indexes. */
...
...
@@ -2392,14 +2439,19 @@ void zrangeGenericCommand(client *c, int reverse) {
/* Invariant: start >= 0, so this test will be true when end < 0.
* The range is empty when start > end or start >= length. */
if
(
start
>
end
||
start
>=
llen
)
{
addReply
(
c
,
shared
.
emptymultibulk
);
addReply
Null
(
c
);
return
;
}
if
(
end
>=
llen
)
end
=
llen
-
1
;
rangelen
=
(
end
-
start
)
+
1
;
/* Return the result in form of a multi-bulk reply */
addReplyMultiBulkLen
(
c
,
withscores
?
(
rangelen
*
2
)
:
rangelen
);
/* Return the result in form of a multi-bulk reply. RESP3 clients
* will receive sub arrays with score->element, while RESP2 returned
* a flat array. */
if
(
withscores
&&
c
->
resp
==
2
)
addReplyArrayLen
(
c
,
rangelen
*
2
);
else
addReplyArrayLen
(
c
,
rangelen
);
if
(
zobj
->
encoding
==
OBJ_ENCODING_ZIPLIST
)
{
unsigned
char
*
zl
=
zobj
->
ptr
;
...
...
@@ -2419,13 +2471,13 @@ void zrangeGenericCommand(client *c, int reverse) {
while
(
rangelen
--
)
{
serverAssertWithInfo
(
c
,
zobj
,
eptr
!=
NULL
&&
sptr
!=
NULL
);
serverAssertWithInfo
(
c
,
zobj
,
ziplistGet
(
eptr
,
&
vstr
,
&
vlen
,
&
vlong
));
if
(
withscores
&&
c
->
resp
>
2
)
addReplyArrayLen
(
c
,
2
);
if
(
vstr
==
NULL
)
addReplyBulkLongLong
(
c
,
vlong
);
else
addReplyBulkCBuffer
(
c
,
vstr
,
vlen
);
if
(
withscores
)
addReplyDouble
(
c
,
zzlGetScore
(
sptr
));
if
(
withscores
)
addReplyDouble
(
c
,
zzlGetScore
(
sptr
));
if
(
reverse
)
zzlPrev
(
zl
,
&
eptr
,
&
sptr
);
...
...
@@ -2453,9 +2505,9 @@ void zrangeGenericCommand(client *c, int reverse) {
while
(
rangelen
--
)
{
serverAssertWithInfo
(
c
,
zobj
,
ln
!=
NULL
);
ele
=
ln
->
ele
;
if
(
withscores
&&
c
->
resp
>
2
)
addReplyArrayLen
(
c
,
2
);
addReplyBulkCBuffer
(
c
,
ele
,
sdslen
(
ele
));
if
(
withscores
)
addReplyDouble
(
c
,
ln
->
score
);
if
(
withscores
)
addReplyDouble
(
c
,
ln
->
score
);
ln
=
reverse
?
ln
->
backward
:
ln
->
level
[
0
].
forward
;
}
}
else
{
...
...
@@ -2523,7 +2575,7 @@ void genericZrangebyscoreCommand(client *c, int reverse) {
}
/* Ok, lookup the key and get the range */
if
((
zobj
=
lookupKeyReadOrReply
(
c
,
key
,
shared
.
emptymultibulk
))
==
NULL
||
if
((
zobj
=
lookupKeyReadOrReply
(
c
,
key
,
shared
.
null
[
c
->
resp
]
))
==
NULL
||
checkType
(
c
,
zobj
,
OBJ_ZSET
))
return
;
if
(
zobj
->
encoding
==
OBJ_ENCODING_ZIPLIST
)
{
...
...
@@ -2543,7 +2595,7 @@ void genericZrangebyscoreCommand(client *c, int reverse) {
/* No "first" element in the specified interval. */
if
(
eptr
==
NULL
)
{
addReply
(
c
,
shared
.
emptymultibulk
);
addReply
Null
(
c
);
return
;
}
...
...
@@ -2554,7 +2606,7 @@ void genericZrangebyscoreCommand(client *c, int reverse) {
/* We don't know in advance how many matching elements there are in the
* list, so we push this object that will represent the multi-bulk
* length in the output buffer, and will "fix" it later */
replylen
=
addDeferred
MultiBulkLength
(
c
);
replylen
=
add
Reply
Deferred
Len
(
c
);
/* If there is an offset, just traverse the number of elements without
* checking the score because that is done in the next loop. */
...
...
@@ -2576,19 +2628,18 @@ void genericZrangebyscoreCommand(client *c, int reverse) {
if
(
!
zslValueLteMax
(
score
,
&
range
))
break
;
}
/* We know the element exists, so ziplistGet should always succeed */
/* We know the element exists, so ziplistGet should always
* succeed */
serverAssertWithInfo
(
c
,
zobj
,
ziplistGet
(
eptr
,
&
vstr
,
&
vlen
,
&
vlong
));
rangelen
++
;
if
(
withscores
&&
c
->
resp
>
2
)
addReplyArrayLen
(
c
,
2
);
if
(
vstr
==
NULL
)
{
addReplyBulkLongLong
(
c
,
vlong
);
}
else
{
addReplyBulkCBuffer
(
c
,
vstr
,
vlen
);
}
if
(
withscores
)
{
addReplyDouble
(
c
,
score
);
}
if
(
withscores
)
addReplyDouble
(
c
,
score
);
/* Move to next node */
if
(
reverse
)
{
...
...
@@ -2611,14 +2662,14 @@ void genericZrangebyscoreCommand(client *c, int reverse) {
/* No "first" element in the specified interval. */
if
(
ln
==
NULL
)
{
addReply
(
c
,
shared
.
emptymultibulk
);
addReply
Null
(
c
);
return
;
}
/* We don't know in advance how many matching elements there are in the
* list, so we push this object that will represent the multi-bulk
* length in the output buffer, and will "fix" it later */
replylen
=
addDeferred
MultiBulkLength
(
c
);
replylen
=
add
Reply
Deferred
Len
(
c
);
/* If there is an offset, just traverse the number of elements without
* checking the score because that is done in the next loop. */
...
...
@@ -2639,11 +2690,9 @@ void genericZrangebyscoreCommand(client *c, int reverse) {
}
rangelen
++
;
if
(
withscores
&&
c
->
resp
>
2
)
addReplyArrayLen
(
c
,
2
);
addReplyBulkCBuffer
(
c
,
ln
->
ele
,
sdslen
(
ln
->
ele
));
if
(
withscores
)
{
addReplyDouble
(
c
,
ln
->
score
);
}
if
(
withscores
)
addReplyDouble
(
c
,
ln
->
score
);
/* Move to next node */
if
(
reverse
)
{
...
...
@@ -2656,11 +2705,8 @@ void genericZrangebyscoreCommand(client *c, int reverse) {
serverPanic
(
"Unknown sorted set encoding"
);
}
if
(
withscores
)
{
rangelen
*=
2
;
}
setDeferredMultiBulkLength
(
c
,
replylen
,
rangelen
);
if
(
withscores
&&
c
->
resp
==
2
)
rangelen
*=
2
;
setDeferredArrayLen
(
c
,
replylen
,
rangelen
);
}
void
zrangebyscoreCommand
(
client
*
c
)
{
...
...
@@ -2871,7 +2917,7 @@ void genericZrangebylexCommand(client *c, int reverse) {
}
/* Ok, lookup the key and get the range */
if
((
zobj
=
lookupKeyReadOrReply
(
c
,
key
,
shared
.
emptymultibulk
))
==
NULL
||
if
((
zobj
=
lookupKeyReadOrReply
(
c
,
key
,
shared
.
null
[
c
->
resp
]
))
==
NULL
||
checkType
(
c
,
zobj
,
OBJ_ZSET
))
{
zslFreeLexRange
(
&
range
);
...
...
@@ -2894,7 +2940,7 @@ void genericZrangebylexCommand(client *c, int reverse) {
/* No "first" element in the specified interval. */
if
(
eptr
==
NULL
)
{
addReply
(
c
,
shared
.
emptymultibulk
);
addReply
Null
(
c
);
zslFreeLexRange
(
&
range
);
return
;
}
...
...
@@ -2906,7 +2952,7 @@ void genericZrangebylexCommand(client *c, int reverse) {
/* We don't know in advance how many matching elements there are in the
* list, so we push this object that will represent the multi-bulk
* length in the output buffer, and will "fix" it later */
replylen
=
addDeferred
MultiBulkLength
(
c
);
replylen
=
add
Reply
Deferred
Len
(
c
);
/* If there is an offset, just traverse the number of elements without
* checking the score because that is done in the next loop. */
...
...
@@ -2958,7 +3004,7 @@ void genericZrangebylexCommand(client *c, int reverse) {
/* No "first" element in the specified interval. */
if
(
ln
==
NULL
)
{
addReply
(
c
,
shared
.
emptymultibulk
);
addReply
Null
(
c
);
zslFreeLexRange
(
&
range
);
return
;
}
...
...
@@ -2966,7 +3012,7 @@ void genericZrangebylexCommand(client *c, int reverse) {
/* We don't know in advance how many matching elements there are in the
* list, so we push this object that will represent the multi-bulk
* length in the output buffer, and will "fix" it later */
replylen
=
addDeferred
MultiBulkLength
(
c
);
replylen
=
add
Reply
Deferred
Len
(
c
);
/* If there is an offset, just traverse the number of elements without
* checking the score because that is done in the next loop. */
...
...
@@ -3001,7 +3047,7 @@ void genericZrangebylexCommand(client *c, int reverse) {
}
zslFreeLexRange
(
&
range
);
setDeferred
MultiBulkLength
(
c
,
replylen
,
rangelen
);
setDeferred
ArrayLen
(
c
,
replylen
,
rangelen
);
}
void
zrangebylexCommand
(
client
*
c
)
{
...
...
@@ -3027,11 +3073,11 @@ void zscoreCommand(client *c) {
robj
*
zobj
;
double
score
;
if
((
zobj
=
lookupKeyReadOrReply
(
c
,
key
,
shared
.
null
bulk
))
==
NULL
||
if
((
zobj
=
lookupKeyReadOrReply
(
c
,
key
,
shared
.
null
[
c
->
resp
]
))
==
NULL
||
checkType
(
c
,
zobj
,
OBJ_ZSET
))
return
;
if
(
zsetScore
(
zobj
,
c
->
argv
[
2
]
->
ptr
,
&
score
)
==
C_ERR
)
{
addReply
(
c
,
shared
.
nullbulk
);
addReply
Null
(
c
);
}
else
{
addReplyDouble
(
c
,
score
);
}
...
...
@@ -3043,7 +3089,7 @@ void zrankGenericCommand(client *c, int reverse) {
robj
*
zobj
;
long
rank
;
if
((
zobj
=
lookupKeyReadOrReply
(
c
,
key
,
shared
.
null
bulk
))
==
NULL
||
if
((
zobj
=
lookupKeyReadOrReply
(
c
,
key
,
shared
.
null
[
c
->
resp
]
))
==
NULL
||
checkType
(
c
,
zobj
,
OBJ_ZSET
))
return
;
serverAssertWithInfo
(
c
,
ele
,
sdsEncodedObject
(
ele
));
...
...
@@ -3051,7 +3097,7 @@ void zrankGenericCommand(client *c, int reverse) {
if
(
rank
>=
0
)
{
addReplyLongLong
(
c
,
rank
);
}
else
{
addReply
(
c
,
shared
.
nullbulk
);
addReply
Null
(
c
);
}
}
...
...
@@ -3109,11 +3155,11 @@ void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey
/* No candidate for zpopping, return empty. */
if
(
!
zobj
)
{
addReply
(
c
,
shared
.
emptymultibulk
);
addReply
Null
(
c
);
return
;
}
void
*
arraylen_ptr
=
addDeferred
MultiBulkLength
(
c
);
void
*
arraylen_ptr
=
add
Reply
Deferred
Len
(
c
);
long
arraylen
=
0
;
/* We emit the key only for the blocking variant. */
...
...
@@ -3180,7 +3226,7 @@ void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey
}
}
while
(
--
count
);
setDeferred
MultiBulkLength
(
c
,
arraylen_ptr
,
arraylen
+
(
emitkey
!=
0
));
setDeferred
ArrayLen
(
c
,
arraylen_ptr
,
arraylen
+
(
emitkey
!=
0
));
}
/* ZPOPMIN key [<count>] */
...
...
@@ -3235,7 +3281,7 @@ void blockingGenericZpopCommand(client *c, int where) {
/* If we are inside a MULTI/EXEC and the zset is empty the only thing
* we can do is treating it as a timeout (even with timeout 0). */
if
(
c
->
flags
&
CLIENT_MULTI
)
{
addReply
(
c
,
shared
.
nullmultibulk
);
addReply
NullArray
(
c
);
return
;
}
...
...
src/util.c
View file @
49816941
...
...
@@ -39,6 +39,7 @@
#include <float.h>
#include <stdint.h>
#include <errno.h>
#include <time.h>
#include "util.h"
#include "sha1.h"
...
...
@@ -47,7 +48,7 @@
int
stringmatchlen
(
const
char
*
pattern
,
int
patternLen
,
const
char
*
string
,
int
stringLen
,
int
nocase
)
{
while
(
patternLen
)
{
while
(
patternLen
&&
stringLen
)
{
switch
(
pattern
[
0
])
{
case
'*'
:
while
(
pattern
[
1
]
==
'*'
)
{
...
...
@@ -170,6 +171,22 @@ int stringmatch(const char *pattern, const char *string, int nocase) {
return
stringmatchlen
(
pattern
,
strlen
(
pattern
),
string
,
strlen
(
string
),
nocase
);
}
/* Fuzz stringmatchlen() trying to crash it with bad input. */
int
stringmatchlen_fuzz_test
(
void
)
{
char
str
[
32
];
char
pat
[
32
];
int
cycles
=
10000000
;
int
total_matches
=
0
;
while
(
cycles
--
)
{
int
strlen
=
rand
()
%
sizeof
(
str
);
int
patlen
=
rand
()
%
sizeof
(
pat
);
for
(
int
j
=
0
;
j
<
strlen
;
j
++
)
str
[
j
]
=
rand
()
%
128
;
for
(
int
j
=
0
;
j
<
patlen
;
j
++
)
pat
[
j
]
=
rand
()
%
128
;
total_matches
+=
stringmatchlen
(
pat
,
patlen
,
str
,
strlen
,
0
);
}
return
total_matches
;
}
/* Convert a string representing an amount of memory into the number of
* bytes, so for instance memtoll("1Gb") will return 1073741824 that is
* (1024*1024*1024).
...
...
@@ -346,6 +363,7 @@ int string2ll(const char *s, size_t slen, long long *value) {
int
negative
=
0
;
unsigned
long
long
v
;
/* A zero length string is not a valid number. */
if
(
plen
==
slen
)
return
0
;
...
...
@@ -355,6 +373,8 @@ int string2ll(const char *s, size_t slen, long long *value) {
return
1
;
}
/* Handle negative numbers: just set a flag and continue like if it
* was a positive number. Later convert into negative. */
if
(
p
[
0
]
==
'-'
)
{
negative
=
1
;
p
++
;
plen
++
;
...
...
@@ -368,13 +388,11 @@ int string2ll(const char *s, size_t slen, long long *value) {
if
(
p
[
0
]
>=
'1'
&&
p
[
0
]
<=
'9'
)
{
v
=
p
[
0
]
-
'0'
;
p
++
;
plen
++
;
}
else
if
(
p
[
0
]
==
'0'
&&
slen
==
1
)
{
*
value
=
0
;
return
1
;
}
else
{
return
0
;
}
/* Parse all the other digits, checking for overflow at every step. */
while
(
plen
<
slen
&&
p
[
0
]
>=
'0'
&&
p
[
0
]
<=
'9'
)
{
if
(
v
>
(
ULLONG_MAX
/
10
))
/* Overflow. */
return
0
;
...
...
@@ -391,6 +409,8 @@ int string2ll(const char *s, size_t slen, long long *value) {
if
(
plen
<
slen
)
return
0
;
/* Convert to negative if needed, and do the final overflow check when
* converting from unsigned long long to long long. */
if
(
negative
)
{
if
(
v
>
((
unsigned
long
long
)(
-
(
LLONG_MIN
+
1
))
+
1
))
/* Overflow. */
return
0
;
...
...
@@ -602,7 +622,7 @@ void getRandomHexChars(char *p, size_t len) {
* already, this will be detected and handled correctly.
*
* The function does not try to normalize everything, but only the obvious
* case of one or more "../" appear
n
ing at the start of "filename"
* case of one or more "../" appearing at the start of "filename"
* relative path. */
sds
getAbsolutePath
(
char
*
filename
)
{
char
cwd
[
1024
];
...
...
@@ -649,6 +669,24 @@ sds getAbsolutePath(char *filename) {
return
abspath
;
}
/*
* Gets the proper timezone in a more portable fashion
* i.e timezone variables are linux specific.
*/
unsigned
long
getTimeZone
(
void
)
{
#ifdef __linux__
return
timezone
;
#else
struct
timeval
tv
;
struct
timezone
tz
;
gettimeofday
(
&
tv
,
&
tz
);
return
tz
.
tz_minuteswest
*
60UL
;
#endif
}
/* Return true if the specified path is just a file basename without any
* relative or absolute path. This function just checks that no / or \
* character exists inside the specified path, that's enough in the
...
...
src/util.h
View file @
49816941
...
...
@@ -40,6 +40,7 @@
int
stringmatchlen
(
const
char
*
p
,
int
plen
,
const
char
*
s
,
int
slen
,
int
nocase
);
int
stringmatch
(
const
char
*
p
,
const
char
*
s
,
int
nocase
);
int
stringmatchlen_fuzz_test
(
void
);
long
long
memtoll
(
const
char
*
p
,
int
*
err
);
uint32_t
digits10
(
uint64_t
v
);
uint32_t
sdigits10
(
int64_t
v
);
...
...
@@ -50,6 +51,7 @@ int string2ld(const char *s, size_t slen, long double *dp);
int
d2string
(
char
*
buf
,
size_t
len
,
double
value
);
int
ld2string
(
char
*
buf
,
size_t
len
,
long
double
value
,
int
humanfriendly
);
sds
getAbsolutePath
(
char
*
filename
);
unsigned
long
getTimeZone
(
void
);
int
pathIsBaseName
(
char
*
path
);
#ifdef REDIS_TEST
...
...
src/zmalloc.c
View file @
49816941
...
...
@@ -164,7 +164,7 @@ void *zrealloc(void *ptr, size_t size) {
if
(
!
newptr
)
zmalloc_oom_handler
(
size
);
*
((
size_t
*
)
newptr
)
=
size
;
update_zmalloc_stat_free
(
oldsize
);
update_zmalloc_stat_free
(
oldsize
+
PREFIX_SIZE
);
update_zmalloc_stat_alloc
(
size
+
PREFIX_SIZE
);
return
(
char
*
)
newptr
+
PREFIX_SIZE
;
#endif
...
...
@@ -183,7 +183,7 @@ size_t zmalloc_size(void *ptr) {
return
size
+
PREFIX_SIZE
;
}
size_t
zmalloc_usable
(
void
*
ptr
)
{
return
zmalloc_
usabl
e
(
ptr
)
-
PREFIX_SIZE
;
return
zmalloc_
siz
e
(
ptr
)
-
PREFIX_SIZE
;
}
#endif
...
...
@@ -438,4 +438,20 @@ size_t zmalloc_get_memory_size(void) {
#endif
}
#ifdef REDIS_TEST
#define UNUSED(x) ((void)(x))
int
zmalloc_test
(
int
argc
,
char
**
argv
)
{
void
*
ptr
;
UNUSED
(
argc
);
UNUSED
(
argv
);
printf
(
"Initial used memory: %zu
\n
"
,
zmalloc_used_memory
());
ptr
=
zmalloc
(
123
);
printf
(
"Allocated 123 bytes; used: %zu
\n
"
,
zmalloc_used_memory
());
ptr
=
zrealloc
(
ptr
,
456
);
printf
(
"Reallocated to 456 bytes; used: %zu
\n
"
,
zmalloc_used_memory
());
zfree
(
ptr
);
printf
(
"Freed pointer; used: %zu
\n
"
,
zmalloc_used_memory
());
return
0
;
}
#endif
src/zmalloc.h
View file @
49816941
...
...
@@ -103,4 +103,8 @@ size_t zmalloc_usable(void *ptr);
#define zmalloc_usable(p) zmalloc_size(p)
#endif
#ifdef REDIS_TEST
int
zmalloc_test
(
int
argc
,
char
**
argv
);
#endif
#endif
/* __ZMALLOC_H */
tests/integration/block-repl.tcl
View file @
49816941
...
...
@@ -49,7 +49,7 @@ start_server {tags {"repl"}} {
set fd
[
open /tmp/repldump2.txt w
]
puts -nonewline $fd $csv2
close $fd
puts
"Master -
Slave
inconsistency"
puts
"Master -
Replica
inconsistency"
puts
"Run diff -u against /tmp/repldump*.txt for more info"
}
assert_equal
[
r debug digest
]
[
r -1 debug digest
]
...
...
tests/integration/psync2-reg.tcl
View file @
49816941
...
...
@@ -29,7 +29,7 @@ start_server {} {
wait_for_condition 50 1000
{
[
$R
(
1
)
dbsize
]
== 1 &&
[
$R
(
2
)
dbsize
]
== 1
}
else
{
fail
"
Slave
s not replicating from master"
fail
"
Replica
s not replicating from master"
}
$R
(
0
)
config set repl-backlog-size 10mb
$R
(
1
)
config set repl-backlog-size 10mb
...
...
@@ -41,12 +41,12 @@ start_server {} {
set elapsed
[
expr
{[
clock milliseconds
]
-$cycle_start_time
}]
if
{
$elapsed
> $duration*1000
}
break
if
{
rand
()
< .05
}
{
test
"PSYNC2 #3899 regression: kill first
slave
"
{
test
"PSYNC2 #3899 regression: kill first
replica
"
{
$R
(
1
)
client kill type master
}
}
if
{
rand
()
< .05
}
{
test
"PSYNC2 #3899 regression: kill chained
slave
"
{
test
"PSYNC2 #3899 regression: kill chained
replica
"
{
$R
(
2
)
client kill type master
}
}
...
...
tests/integration/psync2.tcl
View file @
49816941
...
...
@@ -33,9 +33,8 @@ start_server {} {
set cycle 1
while
{([
clock seconds
]
-$start_time
)
< $duration
}
{
test
"PSYNC2: --- CYCLE
$cycle
---"
{
incr cycle
}
test
"PSYNC2: --- CYCLE
$cycle
---"
{}
incr cycle
# Create a random replication layout.
# Start with switching master
(
this simulates a failover
)
.
...
...
@@ -96,7 +95,7 @@ start_server {} {
if
{
$disconnect
}
{
$R
(
$slave
_id
)
client kill type master
if
{
$debug
_msg
}
{
puts
"+++ Breaking link for
slave
#
$slave
_id"
puts
"+++ Breaking link for
replica
#
$slave
_id"
}
}
}
...
...
@@ -139,6 +138,11 @@ start_server {} {
}
assert
{
$sum
== 4
}
}
# Limit anyway the maximum number of cycles. This is useful when the
# test is skipped via --only option of the test suite. In that case
# we don't want to see many seconds of this test being just skipped.
if
{
$cycle
> 50
}
break
}
test
"PSYNC2: Bring the master back again for next test"
{
...
...
@@ -154,7 +158,7 @@ start_server {} {
wait_for_condition 50 1000
{
[
status $R
(
$master
_id
)
connected_slaves
]
== 4
}
else
{
fail
"
Slave
not reconnecting"
fail
"
Replica
not reconnecting"
}
}
...
...
@@ -169,13 +173,13 @@ start_server {} {
wait_for_condition 50 1000
{
[
status $R
(
$master
_id
)
connected_slaves
]
== 4
}
else
{
fail
"
Slave
not reconnecting"
fail
"
Replica
not reconnecting"
}
set new_sync_count
[
status $R
(
$master
_id
)
sync_full
]
assert
{
$sync
_count == $new_sync_count
}
}
test
"PSYNC2:
Slave
RDB restart with EVALSHA in backlog issue #4483"
{
test
"PSYNC2:
Replica
RDB restart with EVALSHA in backlog issue #4483"
{
# Pick a random slave
set slave_id
[
expr
{(
$master
_id+1
)
%5
}]
set sync_count
[
status $R
(
$master
_id
)
sync_full
]
...
...
@@ -190,7 +194,7 @@ start_server {} {
wait_for_condition 50 1000
{
[
$R
(
$master
_id
)
debug digest
]
==
[
$R
(
$slave
_id
)
debug digest
]
}
else
{
fail
"
Slave
not reconnecting"
fail
"
Replica
not reconnecting"
}
# Prevent the slave from receiving master updates, and at
...
...
@@ -224,7 +228,7 @@ start_server {} {
wait_for_condition 50 1000
{
[
status $R
(
$master
_id
)
connected_slaves
]
== 4
}
else
{
fail
"
Slave
not reconnecting"
fail
"
Replica
not reconnecting"
}
set new_sync_count
[
status $R
(
$master
_id
)
sync_full
]
assert
{
$sync
_count == $new_sync_count
}
...
...
@@ -234,7 +238,7 @@ start_server {} {
wait_for_condition 50 1000
{
[
$R
(
$master
_id
)
debug digest
]
==
[
$R
(
$slave
_id
)
debug digest
]
}
else
{
fail
"Debug digest mismatch between master and
slave
in post-restart handshake"
fail
"Debug digest mismatch between master and
replica
in post-restart handshake"
}
}
...
...
tests/integration/replication-2.tcl
View file @
49816941
...
...
@@ -16,7 +16,7 @@ start_server {tags {"repl"}} {
wait_for_condition 50 100
{
[
r -1 get foo
]
eq
{
12345
}
}
else
{
fail
"Write did not reached
slave
"
fail
"Write did not reached
replica
"
}
}
...
...
@@ -34,7 +34,7 @@ start_server {tags {"repl"}} {
wait_for_condition 50 100
{
[
r -1 get foo
]
eq
{
12345
}
}
else
{
fail
"Write did not reached
slave
"
fail
"Write did not reached
replica
"
}
}
...
...
@@ -60,7 +60,7 @@ start_server {tags {"repl"}} {
wait_for_condition 50 100
{
[
r -1 get foo
]
eq
{
aaabbb
}
}
else
{
fail
"Write did not reached
slave
"
fail
"Write did not reached
replica
"
}
}
...
...
@@ -81,7 +81,7 @@ start_server {tags {"repl"}} {
set fd
[
open /tmp/repldump2.txt w
]
puts -nonewline $fd $csv2
close $fd
puts
"Master -
Slave
inconsistency"
puts
"Master -
Replica
inconsistency"
puts
"Run diff -u against /tmp/repldump*.txt for more info"
}
assert_equal
[
r debug digest
]
[
r -1 debug digest
]
...
...
tests/integration/replication-3.tcl
View file @
49816941
...
...
@@ -25,7 +25,7 @@ start_server {tags {"repl"}} {
set fd
[
open /tmp/repldump2.txt w
]
puts -nonewline $fd $csv2
close $fd
puts
"Master -
Slave
inconsistency"
puts
"Master -
Replica
inconsistency"
puts
"Run diff -u against /tmp/repldump*.txt for more info"
}
assert_equal
[
r debug digest
]
[
r -1 debug digest
]
...
...
@@ -98,7 +98,7 @@ start_server {tags {"repl"}} {
set fd
[
open /tmp/repldump2.txt w
]
puts -nonewline $fd $csv2
close $fd
puts
"Master -
Slave
inconsistency"
puts
"Master -
Replica
inconsistency"
puts
"Run diff -u against /tmp/repldump*.txt for more info"
}
...
...
tests/integration/replication-4.tcl
View file @
49816941
...
...
@@ -47,7 +47,7 @@ start_server {tags {"repl"}} {
set fd
[
open /tmp/repldump2.txt w
]
puts -nonewline $fd $csv2
close $fd
puts
"Master -
Slave
inconsistency"
puts
"Master -
Replica
inconsistency"
puts
"Run diff -u against /tmp/repldump*.txt for more info"
}
assert_equal
[
r debug digest
]
[
r -1 debug digest
]
...
...
tests/integration/replication-psync.tcl
View file @
49816941
...
...
@@ -60,7 +60,7 @@ proc test_psync {descr duration backlog_size backlog_ttl delay cond diskless rec
if
(
$reconnect
)
{
for
{
set j 0
}
{
$j
< $duration*10
}
{
incr j
}
{
after 100
# catch
{
puts
"MASTER
[
$master
dbsize
]
keys,
SLAVE
[
$slave
dbsize
]
keys"
}
# catch
{
puts
"MASTER
[
$master
dbsize
]
keys,
REPLICA
[
$slave
dbsize
]
keys"
}
if
{(
$j
% 20
)
== 0
}
{
catch
{
...
...
@@ -96,7 +96,7 @@ proc test_psync {descr duration backlog_size backlog_ttl delay cond diskless rec
set fd
[
open /tmp/repldump2.txt w
]
puts -nonewline $fd $csv2
close $fd
puts
"Master -
Slave
inconsistency"
puts
"Master -
Replica
inconsistency"
puts
"Run diff -u against /tmp/repldump*.txt for more info"
}
assert_equal
[
r debug digest
]
[
r -1 debug digest
]
...
...
tests/integration/replication.tcl
View file @
49816941
...
...
@@ -32,7 +32,7 @@ start_server {tags {"repl"}} {
wait_for_condition 50 1000
{
[
string match *handshake*
[
$slave
role
]]
}
else
{
fail
"
Slave
does not enter handshake state"
fail
"
Replica
does not enter handshake state"
}
}
...
...
@@ -45,7 +45,7 @@ start_server {tags {"repl"}} {
wait_for_condition 50 1000
{
[
log_file_matches $slave_log
"*Timeout connecting to the MASTER*"
]
}
else
{
fail
"
Slave
is not able to detect timeout"
fail
"
Replica
is not able to detect timeout"
}
}
}
...
...
@@ -66,7 +66,7 @@ start_server {tags {"repl"}} {
[
lindex
[
$A role
]
0
]
eq
{
slave
}
&&
[
string match
{
*master_link_status:up*
}
[
$A info replication
]]
}
else
{
fail
"Can't turn the instance into a
slave
"
fail
"Can't turn the instance into a
replica
"
}
}
...
...
@@ -77,7 +77,7 @@ start_server {tags {"repl"}} {
wait_for_condition 50 100
{
[
$A debug digest
]
eq
[
$B debug digest
]
}
else
{
fail
"Master and
slave
have different digest:
[
$A debug digest
]
VS
[
$B debug digest
]
"
fail
"Master and
replica
have different digest:
[
$A debug digest
]
VS
[
$B debug digest
]
"
}
}
...
...
@@ -102,10 +102,10 @@ start_server {tags {"repl"}} {
[
lindex
[
$B role
]
0
]
eq
{
slave
}
&&
[
string match
{
*master_link_status:up*
}
[
$B info replication
]]
}
else
{
fail
"Can't turn the instance into a
slave
"
fail
"Can't turn the instance into a
replica
"
}
# Push elements into the
"foo"
list of the new
slave
.
# Push elements into the
"foo"
list of the new
replica
.
# If the client is still attached to the instance, we'll get
# a desync between the two instances.
$A rpush foo a b c
...
...
@@ -116,7 +116,7 @@ start_server {tags {"repl"}} {
[
$A lrange foo 0 -1
]
eq
{
a b c
}
&&
[
$B lrange foo 0 -1
]
eq
{
a b c
}
}
else
{
fail
"Master and
slave
have different digest:
[
$A debug digest
]
VS
[
$B debug digest
]
"
fail
"Master and
replica
have different digest:
[
$A debug digest
]
VS
[
$B debug digest
]
"
}
}
}
...
...
@@ -135,7 +135,7 @@ start_server {tags {"repl"}} {
s master_link_status
}
{
down
}
test
{
The role should immediately be changed to
"
slave
"
}
{
test
{
The role should immediately be changed to
"
replica
"
}
{
s role
}
{
slave
}
...
...
@@ -154,7 +154,7 @@ start_server {tags {"repl"}} {
wait_for_condition 500 100
{
[
r 0 get mykey
]
eq
{
bar
}
}
else
{
fail
"SET on master did not propagated on
slave
"
fail
"SET on master did not propagated on
replica
"
}
}
...
...
@@ -201,7 +201,7 @@ foreach dl {no yes} {
lappend slaves
[
srv 0 client
]
start_server
{}
{
lappend slaves
[
srv 0 client
]
test
"Connect multiple
slave
s at the same time (issue #141), diskless=
$dl
"
{
test
"Connect multiple
replica
s at the same time (issue #141), diskless=
$dl
"
{
# Send SLAVEOF commands to slaves
[
lindex $slaves 0
]
slaveof $master_host $master_port
[
lindex $slaves 1
]
slaveof $master_host $master_port
...
...
@@ -220,7 +220,7 @@ foreach dl {no yes} {
}
}
if
{
$retry
== 0
}
{
error
"assertion:
Slave
s not correctly synchronized"
error
"assertion:
Replica
s not correctly synchronized"
}
# Wait that slaves acknowledge they are online so
...
...
@@ -231,7 +231,7 @@ foreach dl {no yes} {
[
lindex
[[
lindex $slaves 1
]
role
]
3
]
eq
{
connected
}
&&
[
lindex
[[
lindex $slaves 2
]
role
]
3
]
eq
{
connected
}
}
else
{
fail
"
Slave
s still not connected after some time"
fail
"
Replica
s still not connected after some time"
}
# Stop the write load
...
...
@@ -248,7 +248,7 @@ foreach dl {no yes} {
[
$master
dbsize
]
==
[[
lindex $slaves 1
]
dbsize
]
&&
[
$master
dbsize
]
==
[[
lindex $slaves 2
]
dbsize
]
}
else
{
fail
"Different number of keys between masted and
slave
after too long time."
fail
"Different number of keys between masted and
replica
after too long time."
}
# Check digests
...
...
@@ -266,3 +266,46 @@ foreach dl {no yes} {
}
}
}
start_server
{
tags
{
"repl"
}}
{
set master
[
srv 0 client
]
set master_host
[
srv 0 host
]
set master_port
[
srv 0 port
]
set load_handle0
[
start_write_load $master_host $master_port 3
]
start_server
{}
{
test
"Master stream is correctly processed while the replica has a script in -BUSY state"
{
set slave
[
srv 0 client
]
$slave config set lua-time-limit 500
$slave slaveof $master_host $master_port
# Wait for the slave to be online
wait_for_condition 500 100
{
[
lindex
[
$slave
role
]
3
]
eq
{
connected
}
}
else
{
fail
"Replica still not connected after some time"
}
# Wait some time to make sure the master is sending data
# to the slave.
after 5000
# Stop the ability of the slave to process data by sendig
# a script that will put it in BUSY state.
$slave eval
{
for i=1,3000000000 do end
}
0
# Wait some time again so that more master stream will
# be processed.
after 2000
# Stop the write load
stop_write_load $load_handle0
# number of keys
wait_for_condition 500 100
{
[
$master
debug digest
]
eq
[
$slave
debug digest
]
}
else
{
fail
"Different datasets between replica and master"
}
}
}
}
tests/sentinel/tests/00-base.tcl
View file @
49816941
...
...
@@ -17,7 +17,7 @@ test "Basic failover works if the master is down" {
wait_for_condition 1000 50
{
[
lindex
[
S $id SENTINEL GET-MASTER-ADDR-BY-NAME mymaster
]
1
]
!= $old_port
}
else
{
fail
"At least one Sentinel did not receive
d
failover info"
fail
"At least one Sentinel did not receive failover info"
}
}
restart_instance redis $master_id
...
...
@@ -108,7 +108,7 @@ test "Failover works if we configure for absolute agreement" {
wait_for_condition 1000 50
{
[
lindex
[
S $id SENTINEL GET-MASTER-ADDR-BY-NAME mymaster
]
1
]
!= $old_port
}
else
{
fail
"At least one Sentinel did not receive
d
failover info"
fail
"At least one Sentinel did not receive failover info"
}
}
restart_instance redis $master_id
...
...
tests/sentinel/tests/01-conf-update.tcl
View file @
49816941
...
...
@@ -16,7 +16,7 @@ test "We can failover with Sentinel 1 crashed" {
wait_for_condition 1000 50
{
[
lindex
[
S $id SENTINEL GET-MASTER-ADDR-BY-NAME mymaster
]
1
]
!= $old_port
}
else
{
fail
"Sentinel
$id
did not receive
d
failover info"
fail
"Sentinel
$id
did not receive failover info"
}
}
}
...
...
@@ -30,7 +30,7 @@ test "After Sentinel 1 is restarted, its config gets updated" {
wait_for_condition 1000 50
{
[
lindex
[
S 1 SENTINEL GET-MASTER-ADDR-BY-NAME mymaster
]
1
]
!= $old_port
}
else
{
fail
"Restarted Sentinel did not receive
d
failover info"
fail
"Restarted Sentinel did not receive failover info"
}
}
...
...
tests/sentinel/tests/02-slaves-reconf.tcl
View file @
49816941
...
...
@@ -36,7 +36,7 @@ proc 02_crash_and_failover {} {
wait_for_condition 1000 50
{
[
lindex
[
S $id SENTINEL GET-MASTER-ADDR-BY-NAME mymaster
]
1
]
!= $old_port
}
else
{
fail
"At least one Sentinel did not receive
d
failover info"
fail
"At least one Sentinel did not receive failover info"
}
}
restart_instance redis $master_id
...
...
tests/sentinel/tests/05-manual.tcl
View file @
49816941
...
...
@@ -12,7 +12,7 @@ test "Manual failover works" {
wait_for_condition 1000 50
{
[
lindex
[
S $id SENTINEL GET-MASTER-ADDR-BY-NAME mymaster
]
1
]
!= $old_port
}
else
{
fail
"At least one Sentinel did not receive
d
failover info"
fail
"At least one Sentinel did not receive failover info"
}
}
set addr
[
S 0 SENTINEL GET-MASTER-ADDR-BY-NAME mymaster
]
...
...
Prev
1
2
3
4
5
6
Next
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
.
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment