Skip to content
GitLab
Menu
Projects
Groups
Snippets
Loading...
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
Menu
Open sidebar
ruanhaishen
redis
Commits
2ab6fef0
Commit
2ab6fef0
authored
Feb 22, 2021
by
Oran Agra
Browse files
Merge origin/unstable into 6.2
parents
2dba1e39
8e83bcd2
Changes
87
Show whitespace changes
Inline
Side-by-side
src/t_hash.c
View file @
2ab6fef0
...
@@ -215,11 +215,8 @@ int hashTypeSet(robj *o, sds field, sds value, int flags) {
...
@@ -215,11 +215,8 @@ int hashTypeSet(robj *o, sds field, sds value, int flags) {
serverAssert
(
vptr
!=
NULL
);
serverAssert
(
vptr
!=
NULL
);
update
=
1
;
update
=
1
;
/* Delete value */
/* Replace value */
zl
=
ziplistDelete
(
zl
,
&
vptr
);
zl
=
ziplistReplace
(
zl
,
vptr
,
(
unsigned
char
*
)
value
,
/* Insert new value */
zl
=
ziplistInsert
(
zl
,
vptr
,
(
unsigned
char
*
)
value
,
sdslen
(
value
));
sdslen
(
value
));
}
}
}
}
...
@@ -759,11 +756,9 @@ void hincrbyfloatCommand(client *c) {
...
@@ -759,11 +756,9 @@ void hincrbyfloatCommand(client *c) {
/* Always replicate HINCRBYFLOAT as an HSET command with the final value
/* Always replicate HINCRBYFLOAT as an HSET command with the final value
* in order to make sure that differences in float precision or formatting
* in order to make sure that differences in float precision or formatting
* will not create differences in replicas or after an AOF restart. */
* will not create differences in replicas or after an AOF restart. */
robj
*
aux
,
*
newobj
;
robj
*
newobj
;
aux
=
createStringObject
(
"HSET"
,
4
);
newobj
=
createRawStringObject
(
buf
,
len
);
newobj
=
createRawStringObject
(
buf
,
len
);
rewriteClientCommandArgument
(
c
,
0
,
aux
);
rewriteClientCommandArgument
(
c
,
0
,
shared
.
hset
);
decrRefCount
(
aux
);
rewriteClientCommandArgument
(
c
,
3
,
newobj
);
rewriteClientCommandArgument
(
c
,
3
,
newobj
);
decrRefCount
(
newobj
);
decrRefCount
(
newobj
);
}
}
...
@@ -959,11 +954,33 @@ void hscanCommand(client *c) {
...
@@ -959,11 +954,33 @@ void hscanCommand(client *c) {
scanGenericCommand
(
c
,
o
,
cursor
);
scanGenericCommand
(
c
,
o
,
cursor
);
}
}
static
void
harndfieldReplyWithZiplist
(
client
*
c
,
unsigned
int
count
,
ziplistEntry
*
keys
,
ziplistEntry
*
vals
)
{
for
(
unsigned
long
i
=
0
;
i
<
count
;
i
++
)
{
if
(
vals
&&
c
->
resp
>
2
)
addReplyArrayLen
(
c
,
2
);
if
(
keys
[
i
].
sval
)
addReplyBulkCBuffer
(
c
,
keys
[
i
].
sval
,
keys
[
i
].
slen
);
else
addReplyBulkLongLong
(
c
,
keys
[
i
].
lval
);
if
(
vals
)
{
if
(
vals
[
i
].
sval
)
addReplyBulkCBuffer
(
c
,
vals
[
i
].
sval
,
vals
[
i
].
slen
);
else
addReplyBulkLongLong
(
c
,
vals
[
i
].
lval
);
}
}
}
/* How many times bigger should be the hash compared to the requested size
/* How many times bigger should be the hash compared to the requested size
* for us to not use the "remove elements" strategy? Read later in the
* for us to not use the "remove elements" strategy? Read later in the
* implementation for more info. */
* implementation for more info. */
#define HRANDFIELD_SUB_STRATEGY_MUL 3
#define HRANDFIELD_SUB_STRATEGY_MUL 3
/* If client is trying to ask for a very large number of random elements,
* queuing may consume an unlimited amount of memory, so we want to limit
* the number of randoms per time. */
#define HRANDFIELD_RANDOM_SAMPLE_LIMIT 1000
void
hrandfieldWithCountCommand
(
client
*
c
,
long
l
,
int
withvalues
)
{
void
hrandfieldWithCountCommand
(
client
*
c
,
long
l
,
int
withvalues
)
{
unsigned
long
count
,
size
;
unsigned
long
count
,
size
;
int
uniq
=
1
;
int
uniq
=
1
;
...
@@ -999,7 +1016,7 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
...
@@ -999,7 +1016,7 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
if
(
hash
->
encoding
==
OBJ_ENCODING_HT
)
{
if
(
hash
->
encoding
==
OBJ_ENCODING_HT
)
{
sds
key
,
value
;
sds
key
,
value
;
while
(
count
--
)
{
while
(
count
--
)
{
dictEntry
*
de
=
dictGetRandomKey
(
hash
->
ptr
);
dictEntry
*
de
=
dictGet
Fair
RandomKey
(
hash
->
ptr
);
key
=
dictGetKey
(
de
);
key
=
dictGetKey
(
de
);
value
=
dictGetVal
(
de
);
value
=
dictGetVal
(
de
);
if
(
withvalues
&&
c
->
resp
>
2
)
if
(
withvalues
&&
c
->
resp
>
2
)
...
@@ -1010,23 +1027,16 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
...
@@ -1010,23 +1027,16 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
}
}
}
else
if
(
hash
->
encoding
==
OBJ_ENCODING_ZIPLIST
)
{
}
else
if
(
hash
->
encoding
==
OBJ_ENCODING_ZIPLIST
)
{
ziplistEntry
*
keys
,
*
vals
=
NULL
;
ziplistEntry
*
keys
,
*
vals
=
NULL
;
keys
=
zmalloc
(
sizeof
(
ziplistEntry
)
*
count
);
unsigned
long
limit
,
sample_count
;
limit
=
count
>
HRANDFIELD_RANDOM_SAMPLE_LIMIT
?
HRANDFIELD_RANDOM_SAMPLE_LIMIT
:
count
;
keys
=
zmalloc
(
sizeof
(
ziplistEntry
)
*
limit
);
if
(
withvalues
)
if
(
withvalues
)
vals
=
zmalloc
(
sizeof
(
ziplistEntry
)
*
count
);
vals
=
zmalloc
(
sizeof
(
ziplistEntry
)
*
limit
);
ziplistRandomPairs
(
hash
->
ptr
,
count
,
keys
,
vals
);
while
(
count
)
{
for
(
unsigned
long
i
=
0
;
i
<
count
;
i
++
)
{
sample_count
=
count
>
limit
?
limit
:
count
;
if
(
withvalues
&&
c
->
resp
>
2
)
count
-=
sample_count
;
addReplyArrayLen
(
c
,
2
);
ziplistRandomPairs
(
hash
->
ptr
,
sample_count
,
keys
,
vals
);
if
(
keys
[
i
].
sval
)
harndfieldReplyWithZiplist
(
c
,
sample_count
,
keys
,
vals
);
addReplyBulkCBuffer
(
c
,
keys
[
i
].
sval
,
keys
[
i
].
slen
);
else
addReplyBulkLongLong
(
c
,
keys
[
i
].
lval
);
if
(
withvalues
)
{
if
(
vals
[
i
].
sval
)
addReplyBulkCBuffer
(
c
,
vals
[
i
].
sval
,
vals
[
i
].
slen
);
else
addReplyBulkLongLong
(
c
,
vals
[
i
].
lval
);
}
}
}
zfree
(
keys
);
zfree
(
keys
);
zfree
(
vals
);
zfree
(
vals
);
...
@@ -1068,6 +1078,7 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
...
@@ -1068,6 +1078,7 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
* used into CASE 4 is highly inefficient. */
* used into CASE 4 is highly inefficient. */
if
(
count
*
HRANDFIELD_SUB_STRATEGY_MUL
>
size
)
{
if
(
count
*
HRANDFIELD_SUB_STRATEGY_MUL
>
size
)
{
dict
*
d
=
dictCreate
(
&
sdsReplyDictType
,
NULL
);
dict
*
d
=
dictCreate
(
&
sdsReplyDictType
,
NULL
);
dictExpand
(
d
,
size
);
hashTypeIterator
*
hi
=
hashTypeInitIterator
(
hash
);
hashTypeIterator
*
hi
=
hashTypeInitIterator
(
hash
);
/* Add all the elements into the temporary dictionary. */
/* Add all the elements into the temporary dictionary. */
...
@@ -1119,9 +1130,25 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
...
@@ -1119,9 +1130,25 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
* to the temporary hash, trying to eventually get enough unique elements
* to the temporary hash, trying to eventually get enough unique elements
* to reach the specified count. */
* to reach the specified count. */
else
{
else
{
if
(
hash
->
encoding
==
OBJ_ENCODING_ZIPLIST
)
{
/* it is inefficient to repeatedly pick one random element from a
* ziplist. so we use this instead: */
ziplistEntry
*
keys
,
*
vals
=
NULL
;
keys
=
zmalloc
(
sizeof
(
ziplistEntry
)
*
count
);
if
(
withvalues
)
vals
=
zmalloc
(
sizeof
(
ziplistEntry
)
*
count
);
serverAssert
(
ziplistRandomPairsUnique
(
hash
->
ptr
,
count
,
keys
,
vals
)
==
count
);
harndfieldReplyWithZiplist
(
c
,
count
,
keys
,
vals
);
zfree
(
keys
);
zfree
(
vals
);
return
;
}
/* Hashtable encoding (generic implementation) */
unsigned
long
added
=
0
;
unsigned
long
added
=
0
;
ziplistEntry
key
,
value
;
ziplistEntry
key
,
value
;
dict
*
d
=
dictCreate
(
&
hashDictType
,
NULL
);
dict
*
d
=
dictCreate
(
&
hashDictType
,
NULL
);
dictExpand
(
d
,
count
);
while
(
added
<
count
)
{
while
(
added
<
count
)
{
hashTypeRandomElement
(
hash
,
size
,
&
key
,
withvalues
?
&
value
:
NULL
);
hashTypeRandomElement
(
hash
,
size
,
&
key
,
withvalues
?
&
value
:
NULL
);
...
...
src/t_list.c
View file @
2ab6fef0
...
@@ -41,10 +41,13 @@
...
@@ -41,10 +41,13 @@
void
listTypePush
(
robj
*
subject
,
robj
*
value
,
int
where
)
{
void
listTypePush
(
robj
*
subject
,
robj
*
value
,
int
where
)
{
if
(
subject
->
encoding
==
OBJ_ENCODING_QUICKLIST
)
{
if
(
subject
->
encoding
==
OBJ_ENCODING_QUICKLIST
)
{
int
pos
=
(
where
==
LIST_HEAD
)
?
QUICKLIST_HEAD
:
QUICKLIST_TAIL
;
int
pos
=
(
where
==
LIST_HEAD
)
?
QUICKLIST_HEAD
:
QUICKLIST_TAIL
;
value
=
getDecodedObject
(
value
);
if
(
value
->
encoding
==
OBJ_ENCODING_INT
)
{
size_t
len
=
sdslen
(
value
->
ptr
);
char
buf
[
32
];
quicklistPush
(
subject
->
ptr
,
value
->
ptr
,
len
,
pos
);
ll2string
(
buf
,
32
,
(
long
)
value
->
ptr
);
decrRefCount
(
value
);
quicklistPush
(
subject
->
ptr
,
buf
,
strlen
(
buf
),
pos
);
}
else
{
quicklistPush
(
subject
->
ptr
,
value
->
ptr
,
sdslen
(
value
->
ptr
),
pos
);
}
}
else
{
}
else
{
serverPanic
(
"Unknown list encoding"
);
serverPanic
(
"Unknown list encoding"
);
}
}
...
@@ -324,7 +327,6 @@ void lindexCommand(client *c) {
...
@@ -324,7 +327,6 @@ void lindexCommand(client *c) {
robj
*
o
=
lookupKeyReadOrReply
(
c
,
c
->
argv
[
1
],
shared
.
null
[
c
->
resp
]);
robj
*
o
=
lookupKeyReadOrReply
(
c
,
c
->
argv
[
1
],
shared
.
null
[
c
->
resp
]);
if
(
o
==
NULL
||
checkType
(
c
,
o
,
OBJ_LIST
))
return
;
if
(
o
==
NULL
||
checkType
(
c
,
o
,
OBJ_LIST
))
return
;
long
index
;
long
index
;
robj
*
value
=
NULL
;
if
((
getLongFromObjectOrReply
(
c
,
c
->
argv
[
2
],
&
index
,
NULL
)
!=
C_OK
))
if
((
getLongFromObjectOrReply
(
c
,
c
->
argv
[
2
],
&
index
,
NULL
)
!=
C_OK
))
return
;
return
;
...
@@ -333,12 +335,10 @@ void lindexCommand(client *c) {
...
@@ -333,12 +335,10 @@ void lindexCommand(client *c) {
quicklistEntry
entry
;
quicklistEntry
entry
;
if
(
quicklistIndex
(
o
->
ptr
,
index
,
&
entry
))
{
if
(
quicklistIndex
(
o
->
ptr
,
index
,
&
entry
))
{
if
(
entry
.
value
)
{
if
(
entry
.
value
)
{
value
=
createStringObject
((
char
*
)
entry
.
value
,
entry
.
sz
);
addReplyBulkCBuffer
(
c
,
entry
.
value
,
entry
.
sz
);
}
else
{
}
else
{
value
=
createStringObjectFrom
LongLong
(
entry
.
longval
);
addReplyBulk
LongLong
(
c
,
entry
.
longval
);
}
}
addReplyBulk
(
c
,
value
);
decrRefCount
(
value
);
}
else
{
}
else
{
addReplyNull
(
c
);
addReplyNull
(
c
);
}
}
...
...
src/t_set.c
View file @
2ab6fef0
...
@@ -499,7 +499,7 @@ void spopWithCountCommand(client *c) {
...
@@ -499,7 +499,7 @@ void spopWithCountCommand(client *c) {
* Prepare our replication argument vector. Also send the array length
* Prepare our replication argument vector. Also send the array length
* which is common to both the code paths. */
* which is common to both the code paths. */
robj
*
propargv
[
3
];
robj
*
propargv
[
3
];
propargv
[
0
]
=
createStringObject
(
"SREM"
,
4
)
;
propargv
[
0
]
=
shared
.
srem
;
propargv
[
1
]
=
c
->
argv
[
1
];
propargv
[
1
]
=
c
->
argv
[
1
];
addReplySetLen
(
c
,
count
);
addReplySetLen
(
c
,
count
);
...
@@ -590,13 +590,12 @@ void spopWithCountCommand(client *c) {
...
@@ -590,13 +590,12 @@ void spopWithCountCommand(client *c) {
* dirty counter. We don't want to propagate an SPOP command since
* dirty counter. We don't want to propagate an SPOP command since
* we propagated the command as a set of SREMs operations using
* we propagated the command as a set of SREMs operations using
* the alsoPropagate() API. */
* the alsoPropagate() API. */
decrRefCount
(
propargv
[
0
]);
preventCommandPropagation
(
c
);
preventCommandPropagation
(
c
);
signalModifiedKey
(
c
,
c
->
db
,
c
->
argv
[
1
]);
signalModifiedKey
(
c
,
c
->
db
,
c
->
argv
[
1
]);
}
}
void
spopCommand
(
client
*
c
)
{
void
spopCommand
(
client
*
c
)
{
robj
*
set
,
*
ele
,
*
aux
;
robj
*
set
,
*
ele
;
sds
sdsele
;
sds
sdsele
;
int64_t
llele
;
int64_t
llele
;
int
encoding
;
int
encoding
;
...
@@ -629,9 +628,7 @@ void spopCommand(client *c) {
...
@@ -629,9 +628,7 @@ void spopCommand(client *c) {
notifyKeyspaceEvent
(
NOTIFY_SET
,
"spop"
,
c
->
argv
[
1
],
c
->
db
->
id
);
notifyKeyspaceEvent
(
NOTIFY_SET
,
"spop"
,
c
->
argv
[
1
],
c
->
db
->
id
);
/* Replicate/AOF this command as an SREM operation */
/* Replicate/AOF this command as an SREM operation */
aux
=
createStringObject
(
"SREM"
,
4
);
rewriteClientCommandVector
(
c
,
3
,
shared
.
srem
,
c
->
argv
[
1
],
ele
);
rewriteClientCommandVector
(
c
,
3
,
aux
,
c
->
argv
[
1
],
ele
);
decrRefCount
(
aux
);
/* Add the element to the reply */
/* Add the element to the reply */
addReplyBulk
(
c
,
ele
);
addReplyBulk
(
c
,
ele
);
...
@@ -677,13 +674,13 @@ void srandmemberWithCountCommand(client *c) {
...
@@ -677,13 +674,13 @@ void srandmemberWithCountCommand(client *c) {
uniq
=
0
;
uniq
=
0
;
}
}
if
((
set
=
lookupKeyReadOrReply
(
c
,
c
->
argv
[
1
],
shared
.
empty
set
[
c
->
resp
]
))
if
((
set
=
lookupKeyReadOrReply
(
c
,
c
->
argv
[
1
],
shared
.
empty
array
))
==
NULL
||
checkType
(
c
,
set
,
OBJ_SET
))
return
;
==
NULL
||
checkType
(
c
,
set
,
OBJ_SET
))
return
;
size
=
setTypeSize
(
set
);
size
=
setTypeSize
(
set
);
/* If count is zero, serve it ASAP to avoid special cases later. */
/* If count is zero, serve it ASAP to avoid special cases later. */
if
(
count
==
0
)
{
if
(
count
==
0
)
{
addReply
(
c
,
shared
.
empty
set
[
c
->
resp
]
);
addReply
(
c
,
shared
.
empty
array
);
return
;
return
;
}
}
...
@@ -693,7 +690,7 @@ void srandmemberWithCountCommand(client *c) {
...
@@ -693,7 +690,7 @@ void srandmemberWithCountCommand(client *c) {
* structures. This case is the only one that also needs to return the
* structures. This case is the only one that also needs to return the
* elements in random order. */
* elements in random order. */
if
(
!
uniq
||
count
==
1
)
{
if
(
!
uniq
||
count
==
1
)
{
addReply
Set
Len
(
c
,
count
);
addReply
Array
Len
(
c
,
count
);
while
(
count
--
)
{
while
(
count
--
)
{
encoding
=
setTypeRandomElement
(
set
,
&
ele
,
&
llele
);
encoding
=
setTypeRandomElement
(
set
,
&
ele
,
&
llele
);
if
(
encoding
==
OBJ_ENCODING_INTSET
)
{
if
(
encoding
==
OBJ_ENCODING_INTSET
)
{
...
@@ -709,7 +706,19 @@ void srandmemberWithCountCommand(client *c) {
...
@@ -709,7 +706,19 @@ void srandmemberWithCountCommand(client *c) {
* The number of requested elements is greater than the number of
* The number of requested elements is greater than the number of
* elements inside the set: simply return the whole set. */
* elements inside the set: simply return the whole set. */
if
(
count
>=
size
)
{
if
(
count
>=
size
)
{
sunionDiffGenericCommand
(
c
,
c
->
argv
+
1
,
1
,
NULL
,
SET_OP_UNION
);
setTypeIterator
*
si
;
addReplyArrayLen
(
c
,
size
);
si
=
setTypeInitIterator
(
set
);
while
((
encoding
=
setTypeNext
(
si
,
&
ele
,
&
llele
))
!=
-
1
)
{
if
(
encoding
==
OBJ_ENCODING_INTSET
)
{
addReplyBulkLongLong
(
c
,
llele
);
}
else
{
addReplyBulkCBuffer
(
c
,
ele
,
sdslen
(
ele
));
}
size
--
;
}
setTypeReleaseIterator
(
si
);
serverAssert
(
size
==
0
);
return
;
return
;
}
}
...
@@ -730,6 +739,7 @@ void srandmemberWithCountCommand(client *c) {
...
@@ -730,6 +739,7 @@ void srandmemberWithCountCommand(client *c) {
/* Add all the elements into the temporary dictionary. */
/* Add all the elements into the temporary dictionary. */
si
=
setTypeInitIterator
(
set
);
si
=
setTypeInitIterator
(
set
);
dictExpand
(
d
,
size
);
while
((
encoding
=
setTypeNext
(
si
,
&
ele
,
&
llele
))
!=
-
1
)
{
while
((
encoding
=
setTypeNext
(
si
,
&
ele
,
&
llele
))
!=
-
1
)
{
int
retval
=
DICT_ERR
;
int
retval
=
DICT_ERR
;
...
@@ -762,6 +772,7 @@ void srandmemberWithCountCommand(client *c) {
...
@@ -762,6 +772,7 @@ void srandmemberWithCountCommand(client *c) {
unsigned
long
added
=
0
;
unsigned
long
added
=
0
;
sds
sdsele
;
sds
sdsele
;
dictExpand
(
d
,
count
);
while
(
added
<
count
)
{
while
(
added
<
count
)
{
encoding
=
setTypeRandomElement
(
set
,
&
ele
,
&
llele
);
encoding
=
setTypeRandomElement
(
set
,
&
ele
,
&
llele
);
if
(
encoding
==
OBJ_ENCODING_INTSET
)
{
if
(
encoding
==
OBJ_ENCODING_INTSET
)
{
...
@@ -784,7 +795,7 @@ void srandmemberWithCountCommand(client *c) {
...
@@ -784,7 +795,7 @@ void srandmemberWithCountCommand(client *c) {
dictIterator
*
di
;
dictIterator
*
di
;
dictEntry
*
de
;
dictEntry
*
de
;
addReply
Set
Len
(
c
,
count
);
addReply
Array
Len
(
c
,
count
);
di
=
dictGetIterator
(
d
);
di
=
dictGetIterator
(
d
);
while
((
de
=
dictNext
(
di
))
!=
NULL
)
while
((
de
=
dictNext
(
di
))
!=
NULL
)
addReplyBulkSds
(
c
,
dictGetKey
(
de
));
addReplyBulkSds
(
c
,
dictGetKey
(
de
));
...
...
src/t_stream.c
View file @
2ab6fef0
...
@@ -43,6 +43,10 @@
...
@@ -43,6 +43,10 @@
* avoid malloc allocation.*/
* avoid malloc allocation.*/
#define STREAMID_STATIC_VECTOR_LEN 8
#define STREAMID_STATIC_VECTOR_LEN 8
/* Max pre-allocation for listpack. This is done to avoid abuse of a user
* setting stream_node_max_bytes to a huge number. */
#define STREAM_LISTPACK_MAX_PRE_ALLOCATE 4096
void
streamFreeCG
(
streamCG
*
cg
);
void
streamFreeCG
(
streamCG
*
cg
);
void
streamFreeNACK
(
streamNACK
*
na
);
void
streamFreeNACK
(
streamNACK
*
na
);
size_t
streamReplyWithRangeFromConsumerPEL
(
client
*
c
,
stream
*
s
,
streamID
*
start
,
streamID
*
end
,
size_t
count
,
streamConsumer
*
consumer
);
size_t
streamReplyWithRangeFromConsumerPEL
(
client
*
c
,
stream
*
s
,
streamID
*
start
,
streamID
*
end
,
size_t
count
,
streamConsumer
*
consumer
);
...
@@ -508,8 +512,16 @@ int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_
...
@@ -508,8 +512,16 @@ int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_
{
{
lp
=
NULL
;
lp
=
NULL
;
}
else
if
(
server
.
stream_node_max_entries
)
{
}
else
if
(
server
.
stream_node_max_entries
)
{
int64_t
count
=
lpGetInteger
(
lpFirst
(
lp
));
unsigned
char
*
lp_ele
=
lpFirst
(
lp
);
if
(
count
>=
server
.
stream_node_max_entries
)
lp
=
NULL
;
/* Count both live entries and deleted ones. */
int64_t
count
=
lpGetInteger
(
lp_ele
)
+
lpGetInteger
(
lpNext
(
lp
,
lp_ele
));
if
(
count
>=
server
.
stream_node_max_entries
)
{
/* Shrink extra pre-allocated memory */
lp
=
lpShrinkToFit
(
lp
);
if
(
ri
.
data
!=
lp
)
raxInsert
(
s
->
rax
,
ri
.
key
,
ri
.
key_len
,
lp
,
NULL
);
lp
=
NULL
;
}
}
}
}
}
...
@@ -517,8 +529,17 @@ int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_
...
@@ -517,8 +529,17 @@ int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_
if
(
lp
==
NULL
)
{
if
(
lp
==
NULL
)
{
master_id
=
id
;
master_id
=
id
;
streamEncodeID
(
rax_key
,
&
id
);
streamEncodeID
(
rax_key
,
&
id
);
/* Create the listpack having the master entry ID and fields. */
/* Create the listpack having the master entry ID and fields.
lp
=
lpNew
();
* Pre-allocate some bytes when creating listpack to avoid realloc on
* every XADD. Since listpack.c uses malloc_size, it'll grow in steps,
* and won't realloc on every XADD.
* When listpack reaches max number of entries, we'll shrink the
* allocation to fit the data. */
size_t
prealloc
=
STREAM_LISTPACK_MAX_PRE_ALLOCATE
;
if
(
server
.
stream_node_max_bytes
>
0
&&
server
.
stream_node_max_bytes
<
prealloc
)
{
prealloc
=
server
.
stream_node_max_bytes
;
}
lp
=
lpNew
(
prealloc
);
lp
=
lpAppendInteger
(
lp
,
1
);
/* One item, the one we are adding. */
lp
=
lpAppendInteger
(
lp
,
1
);
/* One item, the one we are adding. */
lp
=
lpAppendInteger
(
lp
,
0
);
/* Zero deleted so far. */
lp
=
lpAppendInteger
(
lp
,
0
);
/* Zero deleted so far. */
lp
=
lpAppendInteger
(
lp
,
numfields
);
lp
=
lpAppendInteger
(
lp
,
numfields
);
...
@@ -1328,19 +1349,19 @@ void streamPropagateXCLAIM(client *c, robj *key, streamCG *group, robj *groupnam
...
@@ -1328,19 +1349,19 @@ void streamPropagateXCLAIM(client *c, robj *key, streamCG *group, robj *groupnam
* Note that JUSTID is useful in order to avoid that XCLAIM will do
* 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. */
* useless work in the slave side, trying to fetch the stream item. */
robj
*
argv
[
14
];
robj
*
argv
[
14
];
argv
[
0
]
=
createStringObject
(
"XCLAIM"
,
6
)
;
argv
[
0
]
=
shared
.
xclaim
;
argv
[
1
]
=
key
;
argv
[
1
]
=
key
;
argv
[
2
]
=
groupname
;
argv
[
2
]
=
groupname
;
argv
[
3
]
=
createStringObject
(
nack
->
consumer
->
name
,
sdslen
(
nack
->
consumer
->
name
));
argv
[
3
]
=
createStringObject
(
nack
->
consumer
->
name
,
sdslen
(
nack
->
consumer
->
name
));
argv
[
4
]
=
createStringObjectFromLongLong
(
0
)
;
argv
[
4
]
=
shared
.
integers
[
0
]
;
argv
[
5
]
=
id
;
argv
[
5
]
=
id
;
argv
[
6
]
=
createStringObject
(
"TIME"
,
4
)
;
argv
[
6
]
=
shared
.
time
;
argv
[
7
]
=
createStringObjectFromLongLong
(
nack
->
delivery_time
);
argv
[
7
]
=
createStringObjectFromLongLong
(
nack
->
delivery_time
);
argv
[
8
]
=
createStringObject
(
"RETRYCOUNT"
,
10
)
;
argv
[
8
]
=
shared
.
retrycount
;
argv
[
9
]
=
createStringObjectFromLongLong
(
nack
->
delivery_count
);
argv
[
9
]
=
createStringObjectFromLongLong
(
nack
->
delivery_count
);
argv
[
10
]
=
createStringObject
(
"FORCE"
,
5
)
;
argv
[
10
]
=
shared
.
force
;
argv
[
11
]
=
createStringObject
(
"JUSTID"
,
6
)
;
argv
[
11
]
=
shared
.
justid
;
argv
[
12
]
=
createStringObject
(
"LASTID"
,
6
)
;
argv
[
12
]
=
shared
.
lastid
;
argv
[
13
]
=
createObjectFromStreamID
(
&
group
->
last_id
);
argv
[
13
]
=
createObjectFromStreamID
(
&
group
->
last_id
);
/* We use progagate() because this code path is not always called from
/* We use progagate() because this code path is not always called from
...
@@ -1348,16 +1369,9 @@ void streamPropagateXCLAIM(client *c, robj *key, streamCG *group, robj *groupnam
...
@@ -1348,16 +1369,9 @@ void streamPropagateXCLAIM(client *c, robj *key, streamCG *group, robj *groupnam
* consumer group state, and we don't need MULTI/EXEC wrapping because
* consumer group state, and we don't need MULTI/EXEC wrapping because
* there is no message state cross-message atomicity required. */
* there is no message state cross-message atomicity required. */
propagate
(
server
.
xclaimCommand
,
c
->
db
->
id
,
argv
,
14
,
PROPAGATE_AOF
|
PROPAGATE_REPL
);
propagate
(
server
.
xclaimCommand
,
c
->
db
->
id
,
argv
,
14
,
PROPAGATE_AOF
|
PROPAGATE_REPL
);
decrRefCount
(
argv
[
0
]);
decrRefCount
(
argv
[
3
]);
decrRefCount
(
argv
[
3
]);
decrRefCount
(
argv
[
4
]);
decrRefCount
(
argv
[
6
]);
decrRefCount
(
argv
[
7
]);
decrRefCount
(
argv
[
7
]);
decrRefCount
(
argv
[
8
]);
decrRefCount
(
argv
[
9
]);
decrRefCount
(
argv
[
9
]);
decrRefCount
(
argv
[
10
]);
decrRefCount
(
argv
[
11
]);
decrRefCount
(
argv
[
12
]);
decrRefCount
(
argv
[
13
]);
decrRefCount
(
argv
[
13
]);
}
}
...
@@ -1369,8 +1383,8 @@ void streamPropagateXCLAIM(client *c, robj *key, streamCG *group, robj *groupnam
...
@@ -1369,8 +1383,8 @@ void streamPropagateXCLAIM(client *c, robj *key, streamCG *group, robj *groupnam
*/
*/
void
streamPropagateGroupID
(
client
*
c
,
robj
*
key
,
streamCG
*
group
,
robj
*
groupname
)
{
void
streamPropagateGroupID
(
client
*
c
,
robj
*
key
,
streamCG
*
group
,
robj
*
groupname
)
{
robj
*
argv
[
5
];
robj
*
argv
[
5
];
argv
[
0
]
=
createStringObject
(
"XGROUP"
,
6
)
;
argv
[
0
]
=
shared
.
xgroup
;
argv
[
1
]
=
createStringObject
(
"SETID"
,
5
)
;
argv
[
1
]
=
shared
.
setid
;
argv
[
2
]
=
key
;
argv
[
2
]
=
key
;
argv
[
3
]
=
groupname
;
argv
[
3
]
=
groupname
;
argv
[
4
]
=
createObjectFromStreamID
(
&
group
->
last_id
);
argv
[
4
]
=
createObjectFromStreamID
(
&
group
->
last_id
);
...
@@ -1380,8 +1394,6 @@ void streamPropagateGroupID(client *c, robj *key, streamCG *group, robj *groupna
...
@@ -1380,8 +1394,6 @@ void streamPropagateGroupID(client *c, robj *key, streamCG *group, robj *groupna
* consumer group state, and we don't need MULTI/EXEC wrapping because
* consumer group state, and we don't need MULTI/EXEC wrapping because
* there is no message state cross-message atomicity required. */
* there is no message state cross-message atomicity required. */
propagate
(
server
.
xgroupCommand
,
c
->
db
->
id
,
argv
,
5
,
PROPAGATE_AOF
|
PROPAGATE_REPL
);
propagate
(
server
.
xgroupCommand
,
c
->
db
->
id
,
argv
,
5
,
PROPAGATE_AOF
|
PROPAGATE_REPL
);
decrRefCount
(
argv
[
0
]);
decrRefCount
(
argv
[
1
]);
decrRefCount
(
argv
[
4
]);
decrRefCount
(
argv
[
4
]);
}
}
...
@@ -1393,8 +1405,8 @@ void streamPropagateGroupID(client *c, robj *key, streamCG *group, robj *groupna
...
@@ -1393,8 +1405,8 @@ void streamPropagateGroupID(client *c, robj *key, streamCG *group, robj *groupna
*/
*/
void
streamPropagateConsumerCreation
(
client
*
c
,
robj
*
key
,
robj
*
groupname
,
sds
consumername
)
{
void
streamPropagateConsumerCreation
(
client
*
c
,
robj
*
key
,
robj
*
groupname
,
sds
consumername
)
{
robj
*
argv
[
5
];
robj
*
argv
[
5
];
argv
[
0
]
=
createStringObject
(
"XGROUP"
,
6
)
;
argv
[
0
]
=
shared
.
xgroup
;
argv
[
1
]
=
createStringObject
(
"CREATECONSUMER"
,
14
)
;
argv
[
1
]
=
shared
.
createconsumer
;
argv
[
2
]
=
key
;
argv
[
2
]
=
key
;
argv
[
3
]
=
groupname
;
argv
[
3
]
=
groupname
;
argv
[
4
]
=
createObject
(
OBJ_STRING
,
sdsdup
(
consumername
));
argv
[
4
]
=
createObject
(
OBJ_STRING
,
sdsdup
(
consumername
));
...
@@ -1404,8 +1416,6 @@ void streamPropagateConsumerCreation(client *c, robj *key, robj *groupname, sds
...
@@ -1404,8 +1416,6 @@ void streamPropagateConsumerCreation(client *c, robj *key, robj *groupname, sds
* consumer group state, and we don't need MULTI/EXEC wrapping because
* consumer group state, and we don't need MULTI/EXEC wrapping because
* there is no message state cross-message atomicity required. */
* there is no message state cross-message atomicity required. */
propagate
(
server
.
xgroupCommand
,
c
->
db
->
id
,
argv
,
5
,
PROPAGATE_AOF
|
PROPAGATE_REPL
);
propagate
(
server
.
xgroupCommand
,
c
->
db
->
id
,
argv
,
5
,
PROPAGATE_AOF
|
PROPAGATE_REPL
);
decrRefCount
(
argv
[
0
]);
decrRefCount
(
argv
[
1
]);
decrRefCount
(
argv
[
4
]);
decrRefCount
(
argv
[
4
]);
}
}
...
@@ -1725,9 +1735,7 @@ int streamParseIntervalIDOrReply(client *c, robj *o, streamID *id, int *exclude,
...
@@ -1725,9 +1735,7 @@ int streamParseIntervalIDOrReply(client *c, robj *o, streamID *id, int *exclude,
}
}
void
streamRewriteApproxSpecifier
(
client
*
c
,
int
idx
)
{
void
streamRewriteApproxSpecifier
(
client
*
c
,
int
idx
)
{
robj
*
equal_obj
=
createStringObject
(
"="
,
1
);
rewriteClientCommandArgument
(
c
,
idx
,
shared
.
special_equals
);
rewriteClientCommandArgument
(
c
,
idx
,
equal_obj
);
decrRefCount
(
equal_obj
);
}
}
/* We propagate MAXLEN/MINID ~ <count> as MAXLEN/MINID = <resulting-len-of-stream>
/* We propagate MAXLEN/MINID ~ <count> as MAXLEN/MINID = <resulting-len-of-stream>
...
@@ -3471,7 +3479,7 @@ NULL
...
@@ -3471,7 +3479,7 @@ NULL
key
=
c
->
argv
[
2
];
key
=
c
->
argv
[
2
];
/* Lookup the key now, this is common for all the subcommands but HELP. */
/* Lookup the key now, this is common for all the subcommands but HELP. */
robj
*
o
=
lookupKey
Write
OrReply
(
c
,
key
,
shared
.
nokeyerr
);
robj
*
o
=
lookupKey
Read
OrReply
(
c
,
key
,
shared
.
nokeyerr
);
if
(
o
==
NULL
||
checkType
(
c
,
o
,
OBJ_STREAM
))
return
;
if
(
o
==
NULL
||
checkType
(
c
,
o
,
OBJ_STREAM
))
return
;
s
=
o
->
ptr
;
s
=
o
->
ptr
;
...
...
src/t_string.c
View file @
2ab6fef0
...
@@ -73,16 +73,25 @@ static int checkStringLength(client *c, long long size) {
...
@@ -73,16 +73,25 @@ static int checkStringLength(client *c, long long size) {
#define OBJ_PERSIST (1<<8)
/* Set if we need to remove the ttl */
#define OBJ_PERSIST (1<<8)
/* Set if we need to remove the ttl */
void
setGenericCommand
(
client
*
c
,
int
flags
,
robj
*
key
,
robj
*
val
,
robj
*
expire
,
int
unit
,
robj
*
ok_reply
,
robj
*
abort_reply
)
{
void
setGenericCommand
(
client
*
c
,
int
flags
,
robj
*
key
,
robj
*
val
,
robj
*
expire
,
int
unit
,
robj
*
ok_reply
,
robj
*
abort_reply
)
{
long
long
milliseconds
=
0
;
/* initialized to avoid any harmness warning */
long
long
milliseconds
=
0
,
when
=
0
;
/* initialized to avoid any harmness warning */
if
(
expire
)
{
if
(
expire
)
{
if
(
getLongLongFromObjectOrReply
(
c
,
expire
,
&
milliseconds
,
NULL
)
!=
C_OK
)
if
(
getLongLongFromObjectOrReply
(
c
,
expire
,
&
milliseconds
,
NULL
)
!=
C_OK
)
return
;
return
;
if
(
milliseconds
<=
0
)
{
if
(
milliseconds
<=
0
||
(
unit
==
UNIT_SECONDS
&&
milliseconds
>
LLONG_MAX
/
1000
))
{
addReplyErrorFormat
(
c
,
"invalid expire time in %s"
,
c
->
cmd
->
name
);
/* Negative value provided or multiplication is gonna overflow. */
addReplyErrorFormat
(
c
,
"invalid expire time in %s"
,
c
->
cmd
->
name
);
return
;
return
;
}
}
if
(
unit
==
UNIT_SECONDS
)
milliseconds
*=
1000
;
if
(
unit
==
UNIT_SECONDS
)
milliseconds
*=
1000
;
when
=
milliseconds
;
if
((
flags
&
OBJ_PX
)
||
(
flags
&
OBJ_EX
))
when
+=
mstime
();
if
(
when
<=
0
)
{
/* Overflow detected. */
addReplyErrorFormat
(
c
,
"invalid expire time in %s"
,
c
->
cmd
->
name
);
return
;
}
}
}
if
((
flags
&
OBJ_SET_NX
&&
lookupKeyWrite
(
c
->
db
,
key
)
!=
NULL
)
||
if
((
flags
&
OBJ_SET_NX
&&
lookupKeyWrite
(
c
->
db
,
key
)
!=
NULL
)
||
...
@@ -100,14 +109,7 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire,
...
@@ -100,14 +109,7 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire,
server
.
dirty
++
;
server
.
dirty
++
;
notifyKeyspaceEvent
(
NOTIFY_STRING
,
"set"
,
key
,
c
->
db
->
id
);
notifyKeyspaceEvent
(
NOTIFY_STRING
,
"set"
,
key
,
c
->
db
->
id
);
if
(
expire
)
{
if
(
expire
)
{
robj
*
exp
=
shared
.
pxat
;
setExpire
(
c
,
c
->
db
,
key
,
when
);
if
((
flags
&
OBJ_PX
)
||
(
flags
&
OBJ_EX
))
{
setExpire
(
c
,
c
->
db
,
key
,
milliseconds
+
mstime
());
exp
=
shared
.
px
;
}
else
{
setExpire
(
c
,
c
->
db
,
key
,
milliseconds
);
}
notifyKeyspaceEvent
(
NOTIFY_GENERIC
,
"expire"
,
key
,
c
->
db
->
id
);
notifyKeyspaceEvent
(
NOTIFY_GENERIC
,
"expire"
,
key
,
c
->
db
->
id
);
/* Propagate as SET Key Value PXAT millisecond-timestamp if there is EXAT/PXAT or
/* Propagate as SET Key Value PXAT millisecond-timestamp if there is EXAT/PXAT or
...
@@ -119,6 +121,7 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire,
...
@@ -119,6 +121,7 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire,
* Additional care is required while modifying the argument order. AOF relies on the
* Additional care is required while modifying the argument order. AOF relies on the
* exp argument being at index 3. (see feedAppendOnlyFile)
* exp argument being at index 3. (see feedAppendOnlyFile)
* */
* */
robj
*
exp
=
(
flags
&
OBJ_PXAT
)
||
(
flags
&
OBJ_EXAT
)
?
shared
.
pxat
:
shared
.
px
;
robj
*
millisecondObj
=
createStringObjectFromLongLong
(
milliseconds
);
robj
*
millisecondObj
=
createStringObjectFromLongLong
(
milliseconds
);
rewriteClientCommandVector
(
c
,
5
,
shared
.
set
,
key
,
val
,
exp
,
millisecondObj
);
rewriteClientCommandVector
(
c
,
5
,
shared
.
set
,
key
,
val
,
exp
,
millisecondObj
);
decrRefCount
(
millisecondObj
);
decrRefCount
(
millisecondObj
);
...
@@ -335,17 +338,26 @@ void getexCommand(client *c) {
...
@@ -335,17 +338,26 @@ void getexCommand(client *c) {
return
;
return
;
}
}
long
long
milliseconds
=
0
;
long
long
milliseconds
=
0
,
when
=
0
;
/* Validate the expiration time value first */
/* Validate the expiration time value first */
if
(
expire
)
{
if
(
expire
)
{
if
(
getLongLongFromObjectOrReply
(
c
,
expire
,
&
milliseconds
,
NULL
)
!=
C_OK
)
if
(
getLongLongFromObjectOrReply
(
c
,
expire
,
&
milliseconds
,
NULL
)
!=
C_OK
)
return
;
return
;
if
(
milliseconds
<=
0
)
{
if
(
milliseconds
<=
0
||
(
unit
==
UNIT_SECONDS
&&
milliseconds
>
LLONG_MAX
/
1000
))
{
addReplyErrorFormat
(
c
,
"invalid expire time in %s"
,
c
->
cmd
->
name
);
/* Negative value provided or multiplication is gonna overflow. */
addReplyErrorFormat
(
c
,
"invalid expire time in %s"
,
c
->
cmd
->
name
);
return
;
return
;
}
}
if
(
unit
==
UNIT_SECONDS
)
milliseconds
*=
1000
;
if
(
unit
==
UNIT_SECONDS
)
milliseconds
*=
1000
;
when
=
milliseconds
;
if
((
flags
&
OBJ_PX
)
||
(
flags
&
OBJ_EX
))
when
+=
mstime
();
if
(
when
<=
0
)
{
/* Overflow detected. */
addReplyErrorFormat
(
c
,
"invalid expire time in %s"
,
c
->
cmd
->
name
);
return
;
}
}
}
/* We need to do this before we expire the key or delete it */
/* We need to do this before we expire the key or delete it */
...
@@ -365,14 +377,9 @@ void getexCommand(client *c) {
...
@@ -365,14 +377,9 @@ void getexCommand(client *c) {
notifyKeyspaceEvent
(
NOTIFY_GENERIC
,
"del"
,
c
->
argv
[
1
],
c
->
db
->
id
);
notifyKeyspaceEvent
(
NOTIFY_GENERIC
,
"del"
,
c
->
argv
[
1
],
c
->
db
->
id
);
server
.
dirty
++
;
server
.
dirty
++
;
}
else
if
(
expire
)
{
}
else
if
(
expire
)
{
robj
*
exp
=
shared
.
pexpireat
;
setExpire
(
c
,
c
->
db
,
c
->
argv
[
1
],
when
);
if
((
flags
&
OBJ_PX
)
||
(
flags
&
OBJ_EX
))
{
/* Propagate */
setExpire
(
c
,
c
->
db
,
c
->
argv
[
1
],
milliseconds
+
mstime
());
robj
*
exp
=
(
flags
&
OBJ_PXAT
)
||
(
flags
&
OBJ_EXAT
)
?
shared
.
pexpireat
:
shared
.
pexpire
;
exp
=
shared
.
pexpire
;
}
else
{
setExpire
(
c
,
c
->
db
,
c
->
argv
[
1
],
milliseconds
);
}
robj
*
millisecondObj
=
createStringObjectFromLongLong
(
milliseconds
);
robj
*
millisecondObj
=
createStringObjectFromLongLong
(
milliseconds
);
rewriteClientCommandVector
(
c
,
3
,
exp
,
c
->
argv
[
1
],
millisecondObj
);
rewriteClientCommandVector
(
c
,
3
,
exp
,
c
->
argv
[
1
],
millisecondObj
);
decrRefCount
(
millisecondObj
);
decrRefCount
(
millisecondObj
);
...
@@ -631,7 +638,7 @@ void decrbyCommand(client *c) {
...
@@ -631,7 +638,7 @@ void decrbyCommand(client *c) {
void
incrbyfloatCommand
(
client
*
c
)
{
void
incrbyfloatCommand
(
client
*
c
)
{
long
double
incr
,
value
;
long
double
incr
,
value
;
robj
*
o
,
*
new
,
*
aux
;
robj
*
o
,
*
new
;
o
=
lookupKeyWrite
(
c
->
db
,
c
->
argv
[
1
]);
o
=
lookupKeyWrite
(
c
->
db
,
c
->
argv
[
1
]);
if
(
checkType
(
c
,
o
,
OBJ_STRING
))
return
;
if
(
checkType
(
c
,
o
,
OBJ_STRING
))
return
;
...
@@ -659,9 +666,7 @@ void incrbyfloatCommand(client *c) {
...
@@ -659,9 +666,7 @@ void incrbyfloatCommand(client *c) {
* will not create differences in replicas or after an AOF restart. */
* will not create differences in replicas or after an AOF restart. */
rewriteClientCommandArgument
(
c
,
0
,
shared
.
set
);
rewriteClientCommandArgument
(
c
,
0
,
shared
.
set
);
rewriteClientCommandArgument
(
c
,
2
,
new
);
rewriteClientCommandArgument
(
c
,
2
,
new
);
aux
=
createStringObject
(
"KEEPTTL"
,
7
);
rewriteClientCommandArgument
(
c
,
3
,
shared
.
keepttl
);
rewriteClientCommandArgument
(
c
,
3
,
aux
);
decrRefCount
(
aux
);
}
}
void
appendCommand
(
client
*
c
)
{
void
appendCommand
(
client
*
c
)
{
...
@@ -764,7 +769,7 @@ void stralgoLCS(client *c) {
...
@@ -764,7 +769,7 @@ void stralgoLCS(client *c) {
addReplyError
(
c
,
addReplyError
(
c
,
"The specified keys must contain string values"
);
"The specified keys must contain string values"
);
/* Don't cleanup the objects, we need to do that
/* Don't cleanup the objects, we need to do that
* only after calli
g
n getDecodedObject(). */
* only after callin
g
getDecodedObject(). */
obja
=
NULL
;
obja
=
NULL
;
objb
=
NULL
;
objb
=
NULL
;
goto
cleanup
;
goto
cleanup
;
...
...
src/t_zset.c
View file @
2ab6fef0
...
@@ -3956,11 +3956,33 @@ void bzpopmaxCommand(client *c) {
...
@@ -3956,11 +3956,33 @@ void bzpopmaxCommand(client *c) {
blockingGenericZpopCommand
(
c
,
ZSET_MAX
);
blockingGenericZpopCommand
(
c
,
ZSET_MAX
);
}
}
static
void
zarndmemberReplyWithZiplist
(
client
*
c
,
unsigned
int
count
,
ziplistEntry
*
keys
,
ziplistEntry
*
vals
)
{
for
(
unsigned
long
i
=
0
;
i
<
count
;
i
++
)
{
if
(
vals
&&
c
->
resp
>
2
)
addReplyArrayLen
(
c
,
2
);
if
(
keys
[
i
].
sval
)
addReplyBulkCBuffer
(
c
,
keys
[
i
].
sval
,
keys
[
i
].
slen
);
else
addReplyBulkLongLong
(
c
,
keys
[
i
].
lval
);
if
(
vals
)
{
if
(
vals
[
i
].
sval
)
{
addReplyDouble
(
c
,
zzlStrtod
(
vals
[
i
].
sval
,
vals
[
i
].
slen
));
}
else
addReplyDouble
(
c
,
vals
[
i
].
lval
);
}
}
}
/* How many times bigger should be the zset compared to the requested size
/* How many times bigger should be the zset compared to the requested size
* for us to not use the "remove elements" strategy? Read later in the
* for us to not use the "remove elements" strategy? Read later in the
* implementation for more info. */
* implementation for more info. */
#define ZRANDMEMBER_SUB_STRATEGY_MUL 3
#define ZRANDMEMBER_SUB_STRATEGY_MUL 3
/* If client is trying to ask for a very large number of random elements,
* queuing may consume an unlimited amount of memory, so we want to limit
* the number of randoms per time. */
#define ZRANDMEMBER_RANDOM_SAMPLE_LIMIT 1000
void
zrandmemberWithCountCommand
(
client
*
c
,
long
l
,
int
withscores
)
{
void
zrandmemberWithCountCommand
(
client
*
c
,
long
l
,
int
withscores
)
{
unsigned
long
count
,
size
;
unsigned
long
count
,
size
;
int
uniq
=
1
;
int
uniq
=
1
;
...
@@ -4006,23 +4028,16 @@ void zrandmemberWithCountCommand(client *c, long l, int withscores) {
...
@@ -4006,23 +4028,16 @@ void zrandmemberWithCountCommand(client *c, long l, int withscores) {
}
}
}
else
if
(
zsetobj
->
encoding
==
OBJ_ENCODING_ZIPLIST
)
{
}
else
if
(
zsetobj
->
encoding
==
OBJ_ENCODING_ZIPLIST
)
{
ziplistEntry
*
keys
,
*
vals
=
NULL
;
ziplistEntry
*
keys
,
*
vals
=
NULL
;
keys
=
zmalloc
(
sizeof
(
ziplistEntry
)
*
count
);
unsigned
long
limit
,
sample_count
;
limit
=
count
>
ZRANDMEMBER_RANDOM_SAMPLE_LIMIT
?
ZRANDMEMBER_RANDOM_SAMPLE_LIMIT
:
count
;
keys
=
zmalloc
(
sizeof
(
ziplistEntry
)
*
limit
);
if
(
withscores
)
if
(
withscores
)
vals
=
zmalloc
(
sizeof
(
ziplistEntry
)
*
count
);
vals
=
zmalloc
(
sizeof
(
ziplistEntry
)
*
limit
);
ziplistRandomPairs
(
zsetobj
->
ptr
,
count
,
keys
,
vals
);
while
(
count
)
{
for
(
unsigned
long
i
=
0
;
i
<
count
;
i
++
)
{
sample_count
=
count
>
limit
?
limit
:
count
;
if
(
withscores
&&
c
->
resp
>
2
)
count
-=
sample_count
;
addReplyArrayLen
(
c
,
2
);
ziplistRandomPairs
(
zsetobj
->
ptr
,
sample_count
,
keys
,
vals
);
if
(
keys
[
i
].
sval
)
zarndmemberReplyWithZiplist
(
c
,
sample_count
,
keys
,
vals
);
addReplyBulkCBuffer
(
c
,
keys
[
i
].
sval
,
keys
[
i
].
slen
);
else
addReplyBulkLongLong
(
c
,
keys
[
i
].
lval
);
if
(
withscores
)
{
if
(
vals
[
i
].
sval
)
{
addReplyDouble
(
c
,
zzlStrtod
(
vals
[
i
].
sval
,
vals
[
i
].
slen
));
}
else
addReplyDouble
(
c
,
vals
[
i
].
lval
);
}
}
}
zfree
(
keys
);
zfree
(
keys
);
zfree
(
vals
);
zfree
(
vals
);
...
@@ -4070,6 +4085,7 @@ void zrandmemberWithCountCommand(client *c, long l, int withscores) {
...
@@ -4070,6 +4085,7 @@ void zrandmemberWithCountCommand(client *c, long l, int withscores) {
* used into CASE 4 is highly inefficient. */
* used into CASE 4 is highly inefficient. */
if
(
count
*
ZRANDMEMBER_SUB_STRATEGY_MUL
>
size
)
{
if
(
count
*
ZRANDMEMBER_SUB_STRATEGY_MUL
>
size
)
{
dict
*
d
=
dictCreate
(
&
sdsReplyDictType
,
NULL
);
dict
*
d
=
dictCreate
(
&
sdsReplyDictType
,
NULL
);
dictExpand
(
d
,
size
);
/* Add all the elements into the temporary dictionary. */
/* Add all the elements into the temporary dictionary. */
while
(
zuiNext
(
&
src
,
&
zval
))
{
while
(
zuiNext
(
&
src
,
&
zval
))
{
sds
key
=
zuiNewSdsFromValue
(
&
zval
);
sds
key
=
zuiNewSdsFromValue
(
&
zval
);
...
@@ -4111,8 +4127,24 @@ void zrandmemberWithCountCommand(client *c, long l, int withscores) {
...
@@ -4111,8 +4127,24 @@ void zrandmemberWithCountCommand(client *c, long l, int withscores) {
* to the temporary set, trying to eventually get enough unique elements
* to the temporary set, trying to eventually get enough unique elements
* to reach the specified count. */
* to reach the specified count. */
else
{
else
{
if
(
zsetobj
->
encoding
==
OBJ_ENCODING_ZIPLIST
)
{
/* it is inefficient to repeatedly pick one random element from a
* ziplist. so we use this instead: */
ziplistEntry
*
keys
,
*
vals
=
NULL
;
keys
=
zmalloc
(
sizeof
(
ziplistEntry
)
*
count
);
if
(
withscores
)
vals
=
zmalloc
(
sizeof
(
ziplistEntry
)
*
count
);
serverAssert
(
ziplistRandomPairsUnique
(
zsetobj
->
ptr
,
count
,
keys
,
vals
)
==
count
);
zarndmemberReplyWithZiplist
(
c
,
count
,
keys
,
vals
);
zfree
(
keys
);
zfree
(
vals
);
return
;
}
/* Hashtable encoding (generic implementation) */
unsigned
long
added
=
0
;
unsigned
long
added
=
0
;
dict
*
d
=
dictCreate
(
&
hashDictType
,
NULL
);
dict
*
d
=
dictCreate
(
&
hashDictType
,
NULL
);
dictExpand
(
d
,
count
);
while
(
added
<
count
)
{
while
(
added
<
count
)
{
ziplistEntry
key
;
ziplistEntry
key
;
...
...
src/tls.c
View file @
2ab6fef0
...
@@ -350,7 +350,7 @@ ConnectionType CT_TLS;
...
@@ -350,7 +350,7 @@ ConnectionType CT_TLS;
* socket operation.
* socket operation.
*
*
* When this happens, we need to do two things:
* When this happens, we need to do two things:
* 1. Make sure we register for the even.
* 1. Make sure we register for the even
t
.
* 2. Make sure we know which handler needs to execute when the
* 2. Make sure we know which handler needs to execute when the
* event fires. That is, if we notify the caller of a write operation
* event fires. That is, if we notify the caller of a write operation
* that it blocks, and SSL asks for a read, we need to trigger the
* that it blocks, and SSL asks for a read, we need to trigger the
...
...
src/tracking.c
View file @
2ab6fef0
...
@@ -147,7 +147,7 @@ int checkPrefixCollisionsOrReply(client *c, robj **prefixes, size_t numprefix) {
...
@@ -147,7 +147,7 @@ int checkPrefixCollisionsOrReply(client *c, robj **prefixes, size_t numprefix) {
}
}
}
}
}
}
return
-
1
;
return
1
;
}
}
/* Set the client 'c' to track the prefix 'prefix'. If the client 'c' is
/* Set the client 'c' to track the prefix 'prefix'. If the client 'c' is
...
@@ -269,7 +269,7 @@ void sendTrackingMessage(client *c, char *keyname, size_t keylen, int proto) {
...
@@ -269,7 +269,7 @@ void sendTrackingMessage(client *c, char *keyname, size_t keylen, int proto) {
* are unable to send invalidation messages to the redirected
* are unable to send invalidation messages to the redirected
* connection, because the client no longer exist. */
* connection, because the client no longer exist. */
if
(
c
->
resp
>
2
)
{
if
(
c
->
resp
>
2
)
{
addReplyPushLen
(
c
,
3
);
addReplyPushLen
(
c
,
2
);
addReplyBulkCBuffer
(
c
,
"tracking-redir-broken"
,
21
);
addReplyBulkCBuffer
(
c
,
"tracking-redir-broken"
,
21
);
addReplyLongLong
(
c
,
c
->
client_tracking_redirection
);
addReplyLongLong
(
c
,
c
->
client_tracking_redirection
);
}
}
...
...
src/util.c
View file @
2ab6fef0
...
@@ -244,6 +244,33 @@ long long memtoll(const char *p, int *err) {
...
@@ -244,6 +244,33 @@ long long memtoll(const char *p, int *err) {
return
val
*
mul
;
return
val
*
mul
;
}
}
/* Search a memory buffer for any set of bytes, like strpbrk().
* Returns pointer to first found char or NULL.
*/
const
char
*
mempbrk
(
const
char
*
s
,
size_t
len
,
const
char
*
chars
,
size_t
charslen
)
{
for
(
size_t
j
=
0
;
j
<
len
;
j
++
)
{
for
(
size_t
n
=
0
;
n
<
charslen
;
n
++
)
if
(
s
[
j
]
==
chars
[
n
])
return
&
s
[
j
];
}
return
NULL
;
}
/* Modify the buffer replacing all occurrences of chars from the 'from'
* set with the corresponding char in the 'to' set. Always returns s.
*/
char
*
memmapchars
(
char
*
s
,
size_t
len
,
const
char
*
from
,
const
char
*
to
,
size_t
setlen
)
{
for
(
size_t
j
=
0
;
j
<
len
;
j
++
)
{
for
(
size_t
i
=
0
;
i
<
setlen
;
i
++
)
{
if
(
s
[
j
]
==
from
[
i
])
{
s
[
j
]
=
to
[
i
];
break
;
}
}
}
return
s
;
}
/* Return the number of digits of 'v' when converted to string in radix 10.
/* Return the number of digits of 'v' when converted to string in radix 10.
* See ll2string() for more information. */
* See ll2string() for more information. */
uint32_t
digits10
(
uint64_t
v
)
{
uint32_t
digits10
(
uint64_t
v
)
{
...
...
src/util.h
View file @
2ab6fef0
...
@@ -49,6 +49,8 @@ int stringmatchlen(const char *p, int plen, const char *s, int slen, int nocase)
...
@@ -49,6 +49,8 @@ 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
stringmatch
(
const
char
*
p
,
const
char
*
s
,
int
nocase
);
int
stringmatchlen_fuzz_test
(
void
);
int
stringmatchlen_fuzz_test
(
void
);
long
long
memtoll
(
const
char
*
p
,
int
*
err
);
long
long
memtoll
(
const
char
*
p
,
int
*
err
);
const
char
*
mempbrk
(
const
char
*
s
,
size_t
len
,
const
char
*
chars
,
size_t
charslen
);
char
*
memmapchars
(
char
*
s
,
size_t
len
,
const
char
*
from
,
const
char
*
to
,
size_t
setlen
);
uint32_t
digits10
(
uint64_t
v
);
uint32_t
digits10
(
uint64_t
v
);
uint32_t
sdigits10
(
int64_t
v
);
uint32_t
sdigits10
(
int64_t
v
);
int
ll2string
(
char
*
s
,
size_t
len
,
long
long
value
);
int
ll2string
(
char
*
s
,
size_t
len
,
long
long
value
);
...
...
src/ziplist.c
View file @
2ab6fef0
...
@@ -1265,6 +1265,42 @@ unsigned char *ziplistDeleteRange(unsigned char *zl, int index, unsigned int num
...
@@ -1265,6 +1265,42 @@ unsigned char *ziplistDeleteRange(unsigned char *zl, int index, unsigned int num
return
(
p
==
NULL
)
?
zl
:
__ziplistDelete
(
zl
,
p
,
num
);
return
(
p
==
NULL
)
?
zl
:
__ziplistDelete
(
zl
,
p
,
num
);
}
}
/* Replaces the entry at p. This is equivalent to a delete and an insert,
* but avoids some overhead when replacing a value of the same size. */
unsigned
char
*
ziplistReplace
(
unsigned
char
*
zl
,
unsigned
char
*
p
,
unsigned
char
*
s
,
unsigned
int
slen
)
{
/* get metadata of the current entry */
zlentry
entry
;
zipEntry
(
p
,
&
entry
);
/* compute length of entry to store, excluding prevlen */
unsigned
int
reqlen
;
unsigned
char
encoding
=
0
;
long
long
value
=
123456789
;
/* initialized to avoid warning. */
if
(
zipTryEncoding
(
s
,
slen
,
&
value
,
&
encoding
))
{
reqlen
=
zipIntSize
(
encoding
);
/* encoding is set */
}
else
{
reqlen
=
slen
;
/* encoding == 0 */
}
reqlen
+=
zipStoreEntryEncoding
(
NULL
,
encoding
,
slen
);
if
(
reqlen
==
entry
.
lensize
+
entry
.
len
)
{
/* Simply overwrite the element. */
p
+=
entry
.
prevrawlensize
;
p
+=
zipStoreEntryEncoding
(
p
,
encoding
,
slen
);
if
(
ZIP_IS_STR
(
encoding
))
{
memcpy
(
p
,
s
,
slen
);
}
else
{
zipSaveInteger
(
p
,
value
,
encoding
);
}
}
else
{
/* Fallback. */
zl
=
ziplistDelete
(
zl
,
&
p
);
zl
=
ziplistInsert
(
zl
,
p
,
s
,
slen
);
}
return
zl
;
}
/* Compare entry pointer to by 'p' with 'sstr' of length 'slen'. */
/* Compare entry pointer to by 'p' with 'sstr' of length 'slen'. */
/* Return 1 if equal. */
/* Return 1 if equal. */
unsigned
int
ziplistCompare
(
unsigned
char
*
p
,
unsigned
char
*
sstr
,
unsigned
int
slen
)
{
unsigned
int
ziplistCompare
(
unsigned
char
*
p
,
unsigned
char
*
sstr
,
unsigned
int
slen
)
{
...
@@ -1523,8 +1559,8 @@ void ziplistRandomPair(unsigned char *zl, unsigned long total_count, ziplistEntr
...
@@ -1523,8 +1559,8 @@ void ziplistRandomPair(unsigned char *zl, unsigned long total_count, ziplistEntr
}
}
/* int compare for qsort */
/* int compare for qsort */
int
intCompare
(
const
void
*
a
,
const
void
*
b
)
{
int
u
intCompare
(
const
void
*
a
,
const
void
*
b
)
{
return
(
*
(
int
*
)
a
-
*
(
int
*
)
b
);
return
(
*
(
unsigned
int
*
)
a
-
*
(
unsigned
int
*
)
b
);
}
}
/* Helper method to store a string into from val or lval into dest */
/* Helper method to store a string into from val or lval into dest */
...
@@ -1534,39 +1570,42 @@ static inline void ziplistSaveValue(unsigned char *val, unsigned int len, long l
...
@@ -1534,39 +1570,42 @@ static inline void ziplistSaveValue(unsigned char *val, unsigned int len, long l
dest
->
lval
=
lval
;
dest
->
lval
=
lval
;
}
}
/* Randomly select unique count of key value pairs and store into 'keys' and
/* Randomly select count of key value pairs and store into 'keys' and
* 'vals' args. The order of the picked entries is random.
* 'vals' args. The order of the picked entries is random, and the selections
* are non-unique (repetitions are possible).
* The 'vals' arg can be NULL in which case we skip these. */
* The 'vals' arg can be NULL in which case we skip these. */
void
ziplistRandomPairs
(
unsigned
char
*
zl
,
int
count
,
ziplistEntry
*
keys
,
ziplistEntry
*
vals
)
{
void
ziplistRandomPairs
(
unsigned
char
*
zl
,
unsigned
int
count
,
ziplistEntry
*
keys
,
ziplistEntry
*
vals
)
{
unsigned
char
*
p
,
*
key
,
*
value
;
unsigned
char
*
p
,
*
key
,
*
value
;
unsigned
int
klen
,
vlen
;
unsigned
int
klen
=
0
,
vlen
=
0
;
long
long
klval
,
vlval
;
long
long
klval
=
0
,
vlval
=
0
;
/* Notice: the index member must be first due to the use in uintCompare */
typedef
struct
{
typedef
struct
{
int
index
;
unsigned
int
index
;
int
order
;
unsigned
int
order
;
}
rand_pick
;
}
rand_pick
;
rand_pick
*
picks
=
zmalloc
(
sizeof
(
rand_pick
)
*
count
);
rand_pick
*
picks
=
zmalloc
(
sizeof
(
rand_pick
)
*
count
);
unsigned
long
total_size
=
ziplistLen
(
zl
)
/
2
;
unsigned
int
total_size
=
ziplistLen
(
zl
)
/
2
;
/* Avoid div by zero on corrupt ziplist */
/* Avoid div by zero on corrupt ziplist */
assert
(
total_size
);
assert
(
total_size
);
/* create a pool of random indexes (some may be duplicate). */
/* create a pool of random indexes (some may be duplicate). */
for
(
int
i
=
0
;
i
<
count
;
i
++
)
{
for
(
unsigned
int
i
=
0
;
i
<
count
;
i
++
)
{
picks
[
i
].
index
=
(
rand
()
%
total_size
)
*
2
;
/* Generate even indexes */
picks
[
i
].
index
=
(
rand
()
%
total_size
)
*
2
;
/* Generate even indexes */
/* keep track of the order we picked them */
/* keep track of the order we picked them */
picks
[
i
].
order
=
i
;
picks
[
i
].
order
=
i
;
}
}
/* sort by indexes. */
/* sort by indexes. */
qsort
(
picks
,
count
,
sizeof
(
rand_pick
),
intCompare
);
qsort
(
picks
,
count
,
sizeof
(
rand_pick
),
u
intCompare
);
/* fetch the elements form the ziplist into a output array respecting the original order. */
/* fetch the elements form the ziplist into a output array respecting the original order. */
int
zipindex
=
0
,
pickindex
=
0
;
unsigned
int
zipindex
=
0
,
pickindex
=
0
;
p
=
ziplistIndex
(
zl
,
0
);
p
=
ziplistIndex
(
zl
,
0
);
while
(
ziplistGet
(
p
,
&
key
,
&
klen
,
&
klval
)
&&
pickindex
<
count
)
{
while
(
ziplistGet
(
p
,
&
key
,
&
klen
,
&
klval
)
&&
pickindex
<
count
)
{
p
=
ziplistNext
(
zl
,
p
);
p
=
ziplistNext
(
zl
,
p
);
ziplistGet
(
p
,
&
value
,
&
vlen
,
&
vlval
);
assert
(
ziplistGet
(
p
,
&
value
,
&
vlen
,
&
vlval
)
)
;
while
(
pickindex
<
count
&&
zipindex
==
picks
[
pickindex
].
index
)
{
while
(
pickindex
<
count
&&
zipindex
==
picks
[
pickindex
].
index
)
{
int
storeorder
=
picks
[
pickindex
].
order
;
int
storeorder
=
picks
[
pickindex
].
order
;
ziplistSaveValue
(
key
,
klen
,
klval
,
&
keys
[
storeorder
]);
ziplistSaveValue
(
key
,
klen
,
klval
,
&
keys
[
storeorder
]);
...
@@ -1581,6 +1620,51 @@ void ziplistRandomPairs(unsigned char *zl, int count, ziplistEntry *keys, ziplis
...
@@ -1581,6 +1620,51 @@ void ziplistRandomPairs(unsigned char *zl, int count, ziplistEntry *keys, ziplis
zfree
(
picks
);
zfree
(
picks
);
}
}
/* Randomly select count of key value pairs and store into 'keys' and
* 'vals' args. The selections are unique (no repetitions), and the order of
* the picked entries is NOT-random.
* The 'vals' arg can be NULL in which case we skip these.
* The return value is the number of items picked which can be lower than the
* requested count if the ziplist doesn't hold enough pairs. */
unsigned
int
ziplistRandomPairsUnique
(
unsigned
char
*
zl
,
unsigned
int
count
,
ziplistEntry
*
keys
,
ziplistEntry
*
vals
)
{
unsigned
char
*
p
,
*
key
;
unsigned
int
klen
=
0
;
long
long
klval
=
0
;
unsigned
int
total_size
=
ziplistLen
(
zl
)
/
2
;
unsigned
int
index
=
0
;
if
(
count
>
total_size
)
count
=
total_size
;
/* To only iterate once, every time we try to pick a member, the probability
* we pick it is the quotient of the count left we want to pick and the
* count still we haven't visited in the dict, this way, we could make every
* member be equally picked.*/
p
=
ziplistIndex
(
zl
,
0
);
unsigned
int
picked
=
0
,
remaining
=
count
;
while
(
picked
<
count
&&
p
)
{
double
randomDouble
=
((
double
)
rand
())
/
RAND_MAX
;
double
threshold
=
((
double
)
remaining
)
/
(
total_size
-
index
);
if
(
randomDouble
<=
threshold
)
{
assert
(
ziplistGet
(
p
,
&
key
,
&
klen
,
&
klval
));
ziplistSaveValue
(
key
,
klen
,
klval
,
&
keys
[
picked
]);
p
=
ziplistNext
(
zl
,
p
);
assert
(
p
);
if
(
vals
)
{
assert
(
ziplistGet
(
p
,
&
key
,
&
klen
,
&
klval
));
ziplistSaveValue
(
key
,
klen
,
klval
,
&
vals
[
picked
]);
}
remaining
--
;
picked
++
;
}
else
{
p
=
ziplistNext
(
zl
,
p
);
assert
(
p
);
}
p
=
ziplistNext
(
zl
,
p
);
index
++
;
}
return
picked
;
}
#ifdef REDIS_TEST
#ifdef REDIS_TEST
#include <sys/time.h>
#include <sys/time.h>
#include "adlist.h"
#include "adlist.h"
...
@@ -2022,6 +2106,41 @@ int ziplistTest(int argc, char **argv) {
...
@@ -2022,6 +2106,41 @@ int ziplistTest(int argc, char **argv) {
zfree
(
zl
);
zfree
(
zl
);
}
}
printf
(
"Replace with same size:
\n
"
);
{
zl
=
createList
();
/* "hello", "foo", "quux", "1024" */
unsigned
char
*
orig_zl
=
zl
;
p
=
ziplistIndex
(
zl
,
0
);
zl
=
ziplistReplace
(
zl
,
p
,
(
unsigned
char
*
)
"zoink"
,
5
);
p
=
ziplistIndex
(
zl
,
3
);
zl
=
ziplistReplace
(
zl
,
p
,
(
unsigned
char
*
)
"yy"
,
2
);
p
=
ziplistIndex
(
zl
,
1
);
zl
=
ziplistReplace
(
zl
,
p
,
(
unsigned
char
*
)
"65536"
,
5
);
p
=
ziplistIndex
(
zl
,
0
);
assert
(
!
memcmp
((
char
*
)
p
,
"
\x00\x05
zoink"
"
\x07\xf0\x00\x00\x01
"
/* 65536 as int24 */
"
\x05\x04
quux"
"
\x06\x02
yy"
"
\xff
"
,
23
));
assert
(
zl
==
orig_zl
);
/* no reallocations have happened */
zfree
(
zl
);
printf
(
"SUCCESS
\n\n
"
);
}
printf
(
"Replace with different size:
\n
"
);
{
zl
=
createList
();
/* "hello", "foo", "quux", "1024" */
p
=
ziplistIndex
(
zl
,
1
);
zl
=
ziplistReplace
(
zl
,
p
,
(
unsigned
char
*
)
"squirrel"
,
8
);
p
=
ziplistIndex
(
zl
,
0
);
assert
(
!
strncmp
((
char
*
)
p
,
"
\x00\x05
hello"
"
\x07\x08
squirrel"
"
\x0a\x04
quux"
"
\x06\xc0\x00\x04
"
"
\xff
"
,
28
));
zfree
(
zl
);
printf
(
"SUCCESS
\n\n
"
);
}
printf
(
"Regression test for >255 byte strings:
\n
"
);
printf
(
"Regression test for >255 byte strings:
\n
"
);
{
{
char
v1
[
257
]
=
{
0
},
v2
[
257
]
=
{
0
};
char
v1
[
257
]
=
{
0
},
v2
[
257
]
=
{
0
};
...
...
src/ziplist.h
View file @
2ab6fef0
...
@@ -53,6 +53,7 @@ unsigned int ziplistGet(unsigned char *p, unsigned char **sval, unsigned int *sl
...
@@ -53,6 +53,7 @@ unsigned int ziplistGet(unsigned char *p, unsigned char **sval, unsigned int *sl
unsigned
char
*
ziplistInsert
(
unsigned
char
*
zl
,
unsigned
char
*
p
,
unsigned
char
*
s
,
unsigned
int
slen
);
unsigned
char
*
ziplistInsert
(
unsigned
char
*
zl
,
unsigned
char
*
p
,
unsigned
char
*
s
,
unsigned
int
slen
);
unsigned
char
*
ziplistDelete
(
unsigned
char
*
zl
,
unsigned
char
**
p
);
unsigned
char
*
ziplistDelete
(
unsigned
char
*
zl
,
unsigned
char
**
p
);
unsigned
char
*
ziplistDeleteRange
(
unsigned
char
*
zl
,
int
index
,
unsigned
int
num
);
unsigned
char
*
ziplistDeleteRange
(
unsigned
char
*
zl
,
int
index
,
unsigned
int
num
);
unsigned
char
*
ziplistReplace
(
unsigned
char
*
zl
,
unsigned
char
*
p
,
unsigned
char
*
s
,
unsigned
int
slen
);
unsigned
int
ziplistCompare
(
unsigned
char
*
p
,
unsigned
char
*
s
,
unsigned
int
slen
);
unsigned
int
ziplistCompare
(
unsigned
char
*
p
,
unsigned
char
*
s
,
unsigned
int
slen
);
unsigned
char
*
ziplistFind
(
unsigned
char
*
zl
,
unsigned
char
*
p
,
unsigned
char
*
vstr
,
unsigned
int
vlen
,
unsigned
int
skip
);
unsigned
char
*
ziplistFind
(
unsigned
char
*
zl
,
unsigned
char
*
p
,
unsigned
char
*
vstr
,
unsigned
int
vlen
,
unsigned
int
skip
);
unsigned
int
ziplistLen
(
unsigned
char
*
zl
);
unsigned
int
ziplistLen
(
unsigned
char
*
zl
);
...
@@ -62,7 +63,8 @@ typedef int (*ziplistValidateEntryCB)(unsigned char* p, void* userdata);
...
@@ -62,7 +63,8 @@ typedef int (*ziplistValidateEntryCB)(unsigned char* p, void* userdata);
int
ziplistValidateIntegrity
(
unsigned
char
*
zl
,
size_t
size
,
int
deep
,
int
ziplistValidateIntegrity
(
unsigned
char
*
zl
,
size_t
size
,
int
deep
,
ziplistValidateEntryCB
entry_cb
,
void
*
cb_userdata
);
ziplistValidateEntryCB
entry_cb
,
void
*
cb_userdata
);
void
ziplistRandomPair
(
unsigned
char
*
zl
,
unsigned
long
total_count
,
ziplistEntry
*
key
,
ziplistEntry
*
val
);
void
ziplistRandomPair
(
unsigned
char
*
zl
,
unsigned
long
total_count
,
ziplistEntry
*
key
,
ziplistEntry
*
val
);
void
ziplistRandomPairs
(
unsigned
char
*
zl
,
int
count
,
ziplistEntry
*
keys
,
ziplistEntry
*
vals
);
void
ziplistRandomPairs
(
unsigned
char
*
zl
,
unsigned
int
count
,
ziplistEntry
*
keys
,
ziplistEntry
*
vals
);
unsigned
int
ziplistRandomPairsUnique
(
unsigned
char
*
zl
,
unsigned
int
count
,
ziplistEntry
*
keys
,
ziplistEntry
*
vals
);
#ifdef REDIS_TEST
#ifdef REDIS_TEST
int
ziplistTest
(
int
argc
,
char
*
argv
[]);
int
ziplistTest
(
int
argc
,
char
*
argv
[]);
...
...
src/zmalloc.c
View file @
2ab6fef0
...
@@ -57,6 +57,12 @@ void zlibc_free(void *ptr) {
...
@@ -57,6 +57,12 @@ void zlibc_free(void *ptr) {
#endif
#endif
#endif
#endif
#if PREFIX_SIZE > 0
#define ASSERT_NO_SIZE_OVERFLOW(sz) assert((sz) + PREFIX_SIZE > (sz))
#else
#define ASSERT_NO_SIZE_OVERFLOW(sz)
#endif
/* Explicitly override malloc/free etc when using tcmalloc. */
/* Explicitly override malloc/free etc when using tcmalloc. */
#if defined(USE_TCMALLOC)
#if defined(USE_TCMALLOC)
#define malloc(size) tc_malloc(size)
#define malloc(size) tc_malloc(size)
...
@@ -89,6 +95,7 @@ static void (*zmalloc_oom_handler)(size_t) = zmalloc_default_oom;
...
@@ -89,6 +95,7 @@ static void (*zmalloc_oom_handler)(size_t) = zmalloc_default_oom;
/* Try allocating memory, and return NULL if failed.
/* Try allocating memory, and return NULL if failed.
* '*usable' is set to the usable size if non NULL. */
* '*usable' is set to the usable size if non NULL. */
void
*
ztrymalloc_usable
(
size_t
size
,
size_t
*
usable
)
{
void
*
ztrymalloc_usable
(
size_t
size
,
size_t
*
usable
)
{
ASSERT_NO_SIZE_OVERFLOW
(
size
);
void
*
ptr
=
malloc
(
size
+
PREFIX_SIZE
);
void
*
ptr
=
malloc
(
size
+
PREFIX_SIZE
);
if
(
!
ptr
)
return
NULL
;
if
(
!
ptr
)
return
NULL
;
...
@@ -131,6 +138,7 @@ void *zmalloc_usable(size_t size, size_t *usable) {
...
@@ -131,6 +138,7 @@ void *zmalloc_usable(size_t size, size_t *usable) {
* Currently implemented only for jemalloc. Used for online defragmentation. */
* Currently implemented only for jemalloc. Used for online defragmentation. */
#ifdef HAVE_DEFRAG
#ifdef HAVE_DEFRAG
void
*
zmalloc_no_tcache
(
size_t
size
)
{
void
*
zmalloc_no_tcache
(
size_t
size
)
{
ASSERT_NO_SIZE_OVERFLOW
(
size
);
void
*
ptr
=
mallocx
(
size
+
PREFIX_SIZE
,
MALLOCX_TCACHE_NONE
);
void
*
ptr
=
mallocx
(
size
+
PREFIX_SIZE
,
MALLOCX_TCACHE_NONE
);
if
(
!
ptr
)
zmalloc_oom_handler
(
size
);
if
(
!
ptr
)
zmalloc_oom_handler
(
size
);
update_zmalloc_stat_alloc
(
zmalloc_size
(
ptr
));
update_zmalloc_stat_alloc
(
zmalloc_size
(
ptr
));
...
@@ -147,6 +155,7 @@ void zfree_no_tcache(void *ptr) {
...
@@ -147,6 +155,7 @@ void zfree_no_tcache(void *ptr) {
/* Try allocating memory and zero it, and return NULL if failed.
/* Try allocating memory and zero it, and return NULL if failed.
* '*usable' is set to the usable size if non NULL. */
* '*usable' is set to the usable size if non NULL. */
void
*
ztrycalloc_usable
(
size_t
size
,
size_t
*
usable
)
{
void
*
ztrycalloc_usable
(
size_t
size
,
size_t
*
usable
)
{
ASSERT_NO_SIZE_OVERFLOW
(
size
);
void
*
ptr
=
calloc
(
1
,
size
+
PREFIX_SIZE
);
void
*
ptr
=
calloc
(
1
,
size
+
PREFIX_SIZE
);
if
(
ptr
==
NULL
)
return
NULL
;
if
(
ptr
==
NULL
)
return
NULL
;
...
@@ -187,6 +196,7 @@ void *zcalloc_usable(size_t size, size_t *usable) {
...
@@ -187,6 +196,7 @@ void *zcalloc_usable(size_t size, size_t *usable) {
/* Try reallocating memory, and return NULL if failed.
/* Try reallocating memory, and return NULL if failed.
* '*usable' is set to the usable size if non NULL. */
* '*usable' is set to the usable size if non NULL. */
void
*
ztryrealloc_usable
(
void
*
ptr
,
size_t
size
,
size_t
*
usable
)
{
void
*
ztryrealloc_usable
(
void
*
ptr
,
size_t
size
,
size_t
*
usable
)
{
ASSERT_NO_SIZE_OVERFLOW
(
size
);
#ifndef HAVE_MALLOC_SIZE
#ifndef HAVE_MALLOC_SIZE
void
*
realptr
;
void
*
realptr
;
#endif
#endif
...
...
tests/cluster/tests/1
8
-cluster-nodes-slots.tcl
→
tests/cluster/tests/1
9
-cluster-nodes-slots.tcl
View file @
2ab6fef0
File moved
tests/instances.tcl
View file @
2ab6fef0
...
@@ -29,6 +29,7 @@ set ::sentinel_base_port 20000
...
@@ -29,6 +29,7 @@ set ::sentinel_base_port 20000
set ::redis_base_port 30000
set ::redis_base_port 30000
set ::redis_port_count 1024
set ::redis_port_count 1024
set ::host
"127.0.0.1"
set ::host
"127.0.0.1"
set ::leaked_fds_file
[
file normalize
"tmp/leaked_fds.txt"
]
set ::pids
{}
;
# We kill everything at exit
set ::pids
{}
;
# We kill everything at exit
set ::dirs
{}
;
# We remove all the temp dirs at exit
set ::dirs
{}
;
# We remove all the temp dirs at exit
set ::run_matching
{}
;
# If non empty, only tests matching pattern are run.
set ::run_matching
{}
;
# If non empty, only tests matching pattern are run.
...
@@ -410,13 +411,13 @@ proc check_leaks instance_types {
...
@@ -410,13 +411,13 @@ proc check_leaks instance_types {
# Execute all the units inside the 'tests' directory.
# Execute all the units inside the 'tests' directory.
proc run_tests
{}
{
proc run_tests
{}
{
set sentinel_fd_leaks_file
"sentinel_fd_leaks"
if
{
[
file exists $sentinel_fd_leaks_file
]
}
{
file delete $sentinel_fd_leaks_file
}
set tests
[
lsort
[
glob ../tests/*
]]
set tests
[
lsort
[
glob ../tests/*
]]
foreach test $tests
{
foreach test $tests
{
# Remove leaked_fds file before starting
if
{
$::leaked
_fds_file !=
""
&&
[
file exists $::leaked_fds_file
]}
{
file delete $::leaked_fds_file
}
if
{
$::run
_matching ne
{}
&&
[
string match $::run_matching $test
]
== 0
}
{
if
{
$::run
_matching ne
{}
&&
[
string match $::run_matching $test
]
== 0
}
{
continue
continue
}
}
...
@@ -424,19 +425,19 @@ proc run_tests {} {
...
@@ -424,19 +425,19 @@ proc run_tests {} {
puts
[
colorstr yellow
"Testing unit:
[
lindex
[
file split $test
]
end
]
"
]
puts
[
colorstr yellow
"Testing unit:
[
lindex
[
file split $test
]
end
]
"
]
source $test
source $test
check_leaks
{
redis sentinel
}
check_leaks
{
redis sentinel
}
# Check if a leaked fds file was created and abort the test.
if
{
$::leaked
_fds_file !=
""
&&
[
file exists $::leaked_fds_file
]}
{
puts
[
colorstr red
"ERROR: Sentinel has leaked fds to scripts:"
]
puts
[
exec cat $::leaked_fds_file
]
puts
"----"
incr ::failed
}
}
}
}
}
# Print a message and exists with 0 / 1 according to zero or more failures.
# Print a message and exists with 0 / 1 according to zero or more failures.
proc end_tests
{}
{
proc end_tests
{}
{
set sentinel_fd_leaks_file
"sentinel_fd_leaks"
if
{
[
file exists $sentinel_fd_leaks_file
]
}
{
# temporarily disabling this error from failing the tests until leaks are fixed.
#puts
[
colorstr red
"WARNING: sentinel test(s) failed, there are leaked fds in sentinel:"
]
#puts
[
exec cat $sentinel_fd_leaks_file
]
#exit 1
}
if
{
$::failed
== 0
}
{
if
{
$::failed
== 0
}
{
puts
"GOOD! No errors."
puts
"GOOD! No errors."
exit 0
exit 0
...
...
tests/integration/corrupt-dump-fuzzer.tcl
View file @
2ab6fef0
...
@@ -71,6 +71,13 @@ foreach sanitize_dump {no yes} {
...
@@ -71,6 +71,13 @@ foreach sanitize_dump {no yes} {
set min_cycles 10
;
# run at least 10 cycles
set min_cycles 10
;
# run at least 10 cycles
}
}
# Don't execute this on FreeBSD due to a yet-undiscovered memory issue
# which causes tclsh to bloat.
if
{[
exec uname
]
==
"FreeBSD"
}
{
set min_cycles 1
set min_duration 1
}
test
"Fuzzer corrupt restore payloads - sanitize_dump:
$sanitize
_dump"
{
test
"Fuzzer corrupt restore payloads - sanitize_dump:
$sanitize
_dump"
{
if
{
$min
_duration * 2 > $::timeout
}
{
if
{
$min
_duration * 2 > $::timeout
}
{
fail
"insufficient timeout"
fail
"insufficient timeout"
...
...
tests/integration/rdb.tcl
View file @
2ab6fef0
...
@@ -206,10 +206,13 @@ set system_name [string tolower [exec uname -s]]
...
@@ -206,10 +206,13 @@ set system_name [string tolower [exec uname -s]]
if
{
$system
_name eq
{
linux
}}
{
if
{
$system
_name eq
{
linux
}}
{
start_server
{
overrides
{
save
""
}}
{
start_server
{
overrides
{
save
""
}}
{
test
{
Test child sending
COW
info
}
{
test
{
Test child sending info
}
{
# make sure that rdb_last_cow_size and current_cow_size are zero
(
the test using new server
)
,
# make sure that rdb_last_cow_size and current_cow_size are zero
(
the test using new server
)
,
# so that the comparisons during the test will be valid
# so that the comparisons during the test will be valid
assert
{[
s current_cow_size
]
== 0
}
assert
{[
s current_cow_size
]
== 0
}
assert
{[
s current_save_keys_processed
]
== 0
}
assert
{[
s current_save_keys_total
]
== 0
}
assert
{[
s rdb_last_cow_size
]
== 0
}
assert
{[
s rdb_last_cow_size
]
== 0
}
# using a 200us delay, the bgsave is empirically taking about 10 seconds.
# using a 200us delay, the bgsave is empirically taking about 10 seconds.
...
@@ -234,23 +237,35 @@ start_server {overrides {save ""}} {
...
@@ -234,23 +237,35 @@ start_server {overrides {save ""}} {
# start background rdb save
# start background rdb save
r bgsave
r bgsave
set current_save_keys_total
[
s current_save_keys_total
]
if
{
$::verbose
}
{
puts
"Keys before bgsave start: current_save_keys_total"
}
# on each iteration, we will write some key to the server to trigger copy-on-write, and
# on each iteration, we will write some key to the server to trigger copy-on-write, and
# wait to see that it reflected in INFO.
# wait to see that it reflected in INFO.
set iteration 1
set iteration 1
while 1
{
while 1
{
# take
a
sample before writing new data to the server
# take sample
s
before writing new data to the server
set cow_size
[
s current_cow_size
]
set cow_size
[
s current_cow_size
]
if
{
$::verbose
}
{
if
{
$::verbose
}
{
puts
"COW info before copy-on-write:
$cow
_size"
puts
"COW info before copy-on-write:
$cow
_size"
}
}
set keys_processed
[
s current_save_keys_processed
]
if
{
$::verbose
}
{
puts
"current_save_keys_processed info :
$keys
_processed"
}
# trigger copy-on-write
# trigger copy-on-write
r setrange key$iteration 0
[
string repeat B $size
]
r setrange key$iteration 0
[
string repeat B $size
]
# wait to see that current_cow_size value updated
(
as long as the child is in progress
)
# wait to see that current_cow_size value updated
(
as long as the child is in progress
)
wait_for_condition 80 100
{
wait_for_condition 80 100
{
[
s rdb_bgsave_in_progress
]
== 0 ||
[
s rdb_bgsave_in_progress
]
== 0 ||
[
s current_cow_size
]
>= $cow_size + $size
[
s current_cow_size
]
>= $cow_size + $size &&
[
s current_save_keys_processed
]
> $keys_processed &&
[
s current_fork_perc
]
> 0
}
else
{
}
else
{
if
{
$::verbose
}
{
if
{
$::verbose
}
{
puts
"COW info on fail:
[
s current_cow_size
]
"
puts
"COW info on fail:
[
s current_cow_size
]
"
...
@@ -259,6 +274,9 @@ start_server {overrides {save ""}} {
...
@@ -259,6 +274,9 @@ start_server {overrides {save ""}} {
fail
"COW info wasn't reported"
fail
"COW info wasn't reported"
}
}
# assert that $keys_processed is not greater than total keys.
assert_morethan_equal $current_save_keys_total $keys_processed
# for no accurate, stop after 2 iterations
# for no accurate, stop after 2 iterations
if
{
!$::accurate && $iteration == 2
}
{
if
{
!$::accurate && $iteration == 2
}
{
break
break
...
...
tests/integration/redis-cli.tcl
View file @
2ab6fef0
...
@@ -109,7 +109,7 @@ start_server {tags {"cli"}} {
...
@@ -109,7 +109,7 @@ start_server {tags {"cli"}} {
test_interactive_cli
"INFO response should be printed raw"
{
test_interactive_cli
"INFO response should be printed raw"
{
set lines
[
split
[
run_command $fd info
]
"
\n
"
]
set lines
[
split
[
run_command $fd info
]
"
\n
"
]
foreach line $lines
{
foreach line $lines
{
assert
[
regexp
{
^$|^#|^
[
a-z0-9_
]
+:
.+
}
$line
]
assert
[
regexp
{
^$|^#|^
[
^#:
]
+:
}
$line
]
}
}
}
}
...
...
tests/integration/replication-2.tcl
View file @
2ab6fef0
...
@@ -39,17 +39,19 @@ start_server {tags {"repl"}} {
...
@@ -39,17 +39,19 @@ start_server {tags {"repl"}} {
}
}
test
{
No write if min-slaves-max-lag is > of the slave lag
}
{
test
{
No write if min-slaves-max-lag is > of the slave lag
}
{
r -1 deferred 1
r config set min-slaves-to-write 1
r config set min-slaves-to-write 1
r config set min-slaves-max-lag 2
r config set min-slaves-max-lag 2
r -1 debug sleep 6
exec kill -SIGSTOP
[
srv -1 pid
]
assert
{[
r set foo 12345
]
eq
{
OK
}}
assert
{[
r set foo 12345
]
eq
{
OK
}}
after 4000
wait_for_condition 100 100
{
[
catch
{
r set foo 12345
}]
!= 0
}
else
{
fail
"Master didn't become readonly"
}
catch
{
r set foo 12345
}
err
catch
{
r set foo 12345
}
err
assert
{[
r -1 read
]
eq
{
OK
}}
assert_match
{
NOREPLICAS*
}
$err
r -1 deferred 0
}
set err
exec kill -SIGCONT
[
srv -1 pid
]
}
{
NOREPLICAS*
}
test
{
min-slaves-to-write is ignored by slaves
}
{
test
{
min-slaves-to-write is ignored by slaves
}
{
r config set min-slaves-to-write 1
r config set min-slaves-to-write 1
...
...
tests/modules/Makefile
View file @
2ab6fef0
...
@@ -10,6 +10,12 @@ else # Linux, others
...
@@ -10,6 +10,12 @@ else # Linux, others
SHOBJ_LDFLAGS
?=
-shared
SHOBJ_LDFLAGS
?=
-shared
endif
endif
# Needed to satisfy __stack_chk_fail_local on Linux with -m32, due to gcc
# -fstack-protector by default. Breaks on FreeBSD so we exclude it.
ifneq
($(uname_S),FreeBSD)
LIBS
=
-lc
endif
TEST_MODULES
=
\
TEST_MODULES
=
\
commandfilter.so
\
commandfilter.so
\
testrdb.so
\
testrdb.so
\
...
@@ -29,6 +35,8 @@ TEST_MODULES = \
...
@@ -29,6 +35,8 @@ TEST_MODULES = \
test_lazyfree.so
\
test_lazyfree.so
\
timer.so
\
timer.so
\
defragtest.so
\
defragtest.so
\
hash.so
\
zset.so
\
stream.so
stream.so
...
@@ -43,7 +51,7 @@ all: $(TEST_MODULES)
...
@@ -43,7 +51,7 @@ all: $(TEST_MODULES)
$(CC)
-I
../../src
$(CFLAGS)
$(SHOBJ_CFLAGS)
-fPIC
-c
$<
-o
$@
$(CC)
-I
../../src
$(CFLAGS)
$(SHOBJ_CFLAGS)
-fPIC
-c
$<
-o
$@
%.so
:
%.xo
%.so
:
%.xo
$(LD)
-o
$@
$<
$(SHOBJ_LDFLAGS)
$(LDFLAGS)
$(LIBS)
-lc
$(LD)
-o
$@
$<
$(SHOBJ_LDFLAGS)
$(LDFLAGS)
$(LIBS)
.PHONY
:
clean
.PHONY
:
clean
...
...
Prev
1
2
3
4
5
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