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
445aa844
Unverified
Commit
445aa844
authored
Feb 22, 2021
by
Oran Agra
Committed by
GitHub
Feb 22, 2021
Browse files
Merge Redis 6.2.0 GA
Redis 6.2.0 GA
parents
2dba1e39
f098fe31
Changes
89
Hide whitespace changes
Inline
Side-by-side
src/server.h
View file @
445aa844
...
@@ -111,8 +111,10 @@ typedef long long ustime_t; /* microsecond time type. */
...
@@ -111,8 +111,10 @@ typedef long long ustime_t; /* microsecond time type. */
#define CONFIG_DEFAULT_CLUSTER_CONFIG_FILE "nodes.conf"
#define CONFIG_DEFAULT_CLUSTER_CONFIG_FILE "nodes.conf"
#define CONFIG_DEFAULT_UNIX_SOCKET_PERM 0
#define CONFIG_DEFAULT_UNIX_SOCKET_PERM 0
#define CONFIG_DEFAULT_LOGFILE ""
#define CONFIG_DEFAULT_LOGFILE ""
#define NET_HOST_STR_LEN 256
/* Longest valid hostname */
#define NET_IP_STR_LEN 46
/* INET6_ADDRSTRLEN is 46, but we need to be sure */
#define NET_IP_STR_LEN 46
/* INET6_ADDRSTRLEN is 46, but we need to be sure */
#define NET_ADDR_STR_LEN (NET_IP_STR_LEN+32)
/* Must be enough for ip:port */
#define NET_ADDR_STR_LEN (NET_IP_STR_LEN+32)
/* Must be enough for ip:port */
#define NET_HOST_PORT_STR_LEN (NET_HOST_STR_LEN+32)
/* Must be enough for hostname:port */
#define CONFIG_BINDADDR_MAX 16
#define CONFIG_BINDADDR_MAX 16
#define CONFIG_MIN_RESERVED_FDS 32
#define CONFIG_MIN_RESERVED_FDS 32
#define CONFIG_DEFAULT_PROC_TITLE_TEMPLATE "{title} {listen-addr} {server-mode}"
#define CONFIG_DEFAULT_PROC_TITLE_TEMPLATE "{title} {listen-addr} {server-mode}"
...
@@ -901,7 +903,7 @@ typedef struct client {
...
@@ -901,7 +903,7 @@ typedef struct client {
should use. */
should use. */
char
replid
[
CONFIG_RUN_ID_SIZE
+
1
];
/* Master replication ID (if master). */
char
replid
[
CONFIG_RUN_ID_SIZE
+
1
];
/* Master replication ID (if master). */
int
slave_listening_port
;
/* As configured with: REPLCONF listening-port */
int
slave_listening_port
;
/* As configured with: REPLCONF listening-port */
char
slave_
ip
[
NET_IP_STR_LEN
];
/* Optionally given by REPLCONF ip-address */
char
*
slave_
addr
;
/* Optionally given by REPLCONF ip-address */
int
slave_capa
;
/* Slave capabilities: SLAVE_CAPA_* bitwise OR. */
int
slave_capa
;
/* Slave capabilities: SLAVE_CAPA_* bitwise OR. */
multiState
mstate
;
/* MULTI/EXEC state */
multiState
mstate
;
/* MULTI/EXEC state */
int
btype
;
/* Type of blocking op if CLIENT_BLOCKED. */
int
btype
;
/* Type of blocking op if CLIENT_BLOCKED. */
...
@@ -976,8 +978,11 @@ struct sharedObjectsStruct {
...
@@ -976,8 +978,11 @@ struct sharedObjectsStruct {
*
busykeyerr
,
*
oomerr
,
*
plus
,
*
messagebulk
,
*
pmessagebulk
,
*
subscribebulk
,
*
busykeyerr
,
*
oomerr
,
*
plus
,
*
messagebulk
,
*
pmessagebulk
,
*
subscribebulk
,
*
unsubscribebulk
,
*
psubscribebulk
,
*
punsubscribebulk
,
*
del
,
*
unlink
,
*
unsubscribebulk
,
*
psubscribebulk
,
*
punsubscribebulk
,
*
del
,
*
unlink
,
*
rpop
,
*
lpop
,
*
lpush
,
*
rpoplpush
,
*
lmove
,
*
blmove
,
*
zpopmin
,
*
zpopmax
,
*
rpop
,
*
lpop
,
*
lpush
,
*
rpoplpush
,
*
lmove
,
*
blmove
,
*
zpopmin
,
*
zpopmax
,
*
emptyscan
,
*
multi
,
*
exec
,
*
left
,
*
right
,
*
persist
,
*
set
,
*
pexpireat
,
*
emptyscan
,
*
multi
,
*
exec
,
*
left
,
*
right
,
*
hset
,
*
srem
,
*
xgroup
,
*
xclaim
,
*
pexpire
,
*
pxat
,
*
px
,
*
script
,
*
replconf
,
*
eval
,
*
persist
,
*
set
,
*
pexpireat
,
*
pexpire
,
*
time
,
*
pxat
,
*
px
,
*
retrycount
,
*
force
,
*
justid
,
*
lastid
,
*
ping
,
*
setid
,
*
keepttl
,
*
load
,
*
createconsumer
,
*
getack
,
*
special_asterick
,
*
special_equals
,
*
default_username
,
*
select
[
PROTO_SHARED_SELECT_CMDS
],
*
select
[
PROTO_SHARED_SELECT_CMDS
],
*
integers
[
OBJ_SHARED_INTEGERS
],
*
integers
[
OBJ_SHARED_INTEGERS
],
*
mbulkhdr
[
OBJ_SHARED_BULKHDR_LEN
],
/* "*<value>\r\n" */
*
mbulkhdr
[
OBJ_SHARED_BULKHDR_LEN
],
/* "*<value>\r\n" */
...
@@ -1139,6 +1144,13 @@ struct clusterState;
...
@@ -1139,6 +1144,13 @@ struct clusterState;
#define CHILD_TYPE_LDB 3
#define CHILD_TYPE_LDB 3
#define CHILD_TYPE_MODULE 4
#define CHILD_TYPE_MODULE 4
typedef
enum
childInfoType
{
CHILD_INFO_TYPE_CURRENT_INFO
,
CHILD_INFO_TYPE_AOF_COW_SIZE
,
CHILD_INFO_TYPE_RDB_COW_SIZE
,
CHILD_INFO_TYPE_MODULE_COW_SIZE
}
childInfoType
;
struct
redisServer
{
struct
redisServer
{
/* General */
/* General */
pid_t
pid
;
/* Main process pid. */
pid_t
pid
;
/* Main process pid. */
...
@@ -1267,9 +1279,12 @@ struct redisServer {
...
@@ -1267,9 +1279,12 @@ struct redisServer {
redisAtomic
long
long
stat_net_input_bytes
;
/* Bytes read from network. */
redisAtomic
long
long
stat_net_input_bytes
;
/* Bytes read from network. */
redisAtomic
long
long
stat_net_output_bytes
;
/* Bytes written to network. */
redisAtomic
long
long
stat_net_output_bytes
;
/* Bytes written to network. */
size_t
stat_current_cow_bytes
;
/* Copy on write bytes while child is active. */
size_t
stat_current_cow_bytes
;
/* Copy on write bytes while child is active. */
size_t
stat_current_save_keys_processed
;
/* Processed keys while child is active. */
size_t
stat_current_save_keys_total
;
/* Number of keys when child started. */
size_t
stat_rdb_cow_bytes
;
/* Copy on write bytes during RDB saving. */
size_t
stat_rdb_cow_bytes
;
/* Copy on write bytes during RDB saving. */
size_t
stat_aof_cow_bytes
;
/* Copy on write bytes during AOF rewrite. */
size_t
stat_aof_cow_bytes
;
/* Copy on write bytes during AOF rewrite. */
size_t
stat_module_cow_bytes
;
/* Copy on write bytes during module fork. */
size_t
stat_module_cow_bytes
;
/* Copy on write bytes during module fork. */
double
stat_module_progress
;
/* Module save progress. */
uint64_t
stat_clients_type_memory
[
CLIENT_TYPE_COUNT
];
/* Mem usage by type */
uint64_t
stat_clients_type_memory
[
CLIENT_TYPE_COUNT
];
/* Mem usage by type */
long
long
stat_unexpected_error_replies
;
/* Number of unexpected (aof-loading, replica to master, etc.) error replies */
long
long
stat_unexpected_error_replies
;
/* Number of unexpected (aof-loading, replica to master, etc.) error replies */
long
long
stat_total_error_replies
;
/* Total number of issued error replies ( command + rejected errors ) */
long
long
stat_total_error_replies
;
/* Total number of issued error replies ( command + rejected errors ) */
...
@@ -1508,8 +1523,7 @@ struct redisServer {
...
@@ -1508,8 +1523,7 @@ struct redisServer {
long
long
blocked_last_cron
;
/* Indicate the mstime of the last time we did cron jobs from a blocking operation */
long
long
blocked_last_cron
;
/* Indicate the mstime of the last time we did cron jobs from a blocking operation */
/* Pubsub */
/* Pubsub */
dict
*
pubsub_channels
;
/* Map channels to list of subscribed clients */
dict
*
pubsub_channels
;
/* Map channels to list of subscribed clients */
list
*
pubsub_patterns
;
/* A list of pubsub_patterns */
dict
*
pubsub_patterns
;
/* A dict of pubsub_patterns */
dict
*
pubsub_patterns_dict
;
/* A dict of pubsub_patterns */
int
notify_keyspace_events
;
/* Events to propagate via Pub/Sub. This is an
int
notify_keyspace_events
;
/* Events to propagate via Pub/Sub. This is an
xor of NOTIFY_... flags. */
xor of NOTIFY_... flags. */
/* Cluster */
/* Cluster */
...
@@ -1596,11 +1610,6 @@ struct redisServer {
...
@@ -1596,11 +1610,6 @@ struct redisServer {
int
failover_state
;
/* Failover state */
int
failover_state
;
/* Failover state */
};
};
typedef
struct
pubsubPattern
{
client
*
client
;
robj
*
pattern
;
}
pubsubPattern
;
#define MAX_KEYS_BUFFER 256
#define MAX_KEYS_BUFFER 256
/* A result structure for the various getkeys function calls. It lists the
/* A result structure for the various getkeys function calls. It lists the
...
@@ -2057,7 +2066,9 @@ void restartAOFAfterSYNC();
...
@@ -2057,7 +2066,9 @@ void restartAOFAfterSYNC();
/* Child info */
/* Child info */
void
openChildInfoPipe
(
void
);
void
openChildInfoPipe
(
void
);
void
closeChildInfoPipe
(
void
);
void
closeChildInfoPipe
(
void
);
void
sendChildInfo
(
int
process_type
,
int
on_exit
,
size_t
cow_size
);
void
sendChildInfoGeneric
(
childInfoType
info_type
,
size_t
keys
,
double
progress
,
char
*
pname
);
void
sendChildCowInfo
(
childInfoType
info_type
,
char
*
pname
);
void
sendChildInfo
(
childInfoType
info_type
,
size_t
keys
,
char
*
pname
);
void
receiveChildInfo
(
void
);
void
receiveChildInfo
(
void
);
/* Fork helpers */
/* Fork helpers */
...
@@ -2065,7 +2076,6 @@ int redisFork(int type);
...
@@ -2065,7 +2076,6 @@ int redisFork(int type);
int
hasActiveChildProcess
();
int
hasActiveChildProcess
();
void
resetChildState
();
void
resetChildState
();
int
isMutuallyExclusiveChildType
(
int
type
);
int
isMutuallyExclusiveChildType
(
int
type
);
void
sendChildCOWInfo
(
int
ptype
,
int
on_exit
,
char
*
pname
);
/* acl.c -- Authentication related prototypes. */
/* acl.c -- Authentication related prototypes. */
extern
rax
*
Users
;
extern
rax
*
Users
;
...
@@ -2266,8 +2276,6 @@ int hashZiplistValidateIntegrity(unsigned char *zl, size_t size, int deep);
...
@@ -2266,8 +2276,6 @@ int hashZiplistValidateIntegrity(unsigned char *zl, size_t size, int deep);
/* Pub / Sub */
/* Pub / Sub */
int
pubsubUnsubscribeAllChannels
(
client
*
c
,
int
notify
);
int
pubsubUnsubscribeAllChannels
(
client
*
c
,
int
notify
);
int
pubsubUnsubscribeAllPatterns
(
client
*
c
,
int
notify
);
int
pubsubUnsubscribeAllPatterns
(
client
*
c
,
int
notify
);
void
freePubsubPattern
(
void
*
p
);
int
listMatchPubsubPattern
(
void
*
a
,
void
*
b
);
int
pubsubPublishMessage
(
robj
*
channel
,
robj
*
message
);
int
pubsubPublishMessage
(
robj
*
channel
,
robj
*
message
);
void
addReplyPubsubMessage
(
client
*
c
,
robj
*
channel
,
robj
*
msg
);
void
addReplyPubsubMessage
(
client
*
c
,
robj
*
channel
,
robj
*
msg
);
...
@@ -2682,6 +2690,7 @@ void _serverPanic(const char *file, int line, const char *msg, ...);
...
@@ -2682,6 +2690,7 @@ void _serverPanic(const char *file, int line, const char *msg, ...);
#endif
#endif
void
serverLogObjectDebugInfo
(
const
robj
*
o
);
void
serverLogObjectDebugInfo
(
const
robj
*
o
);
void
sigsegvHandler
(
int
sig
,
siginfo_t
*
info
,
void
*
secret
);
void
sigsegvHandler
(
int
sig
,
siginfo_t
*
info
,
void
*
secret
);
const
char
*
getSafeInfoString
(
const
char
*
s
,
size_t
len
,
char
**
tmp
);
sds
genRedisInfoString
(
const
char
*
section
);
sds
genRedisInfoString
(
const
char
*
section
);
sds
genModulesInfoString
(
sds
info
);
sds
genModulesInfoString
(
sds
info
);
void
enableWatchdog
(
int
period
);
void
enableWatchdog
(
int
period
);
...
...
src/t_hash.c
View file @
445aa844
...
@@ -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 @
445aa844
...
@@ -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 @
445aa844
...
@@ -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 @
445aa844
...
@@ -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 @
445aa844
...
@@ -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 @
445aa844
...
@@ -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 @
445aa844
...
@@ -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 @
445aa844
...
@@ -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 @
445aa844
...
@@ -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 @
445aa844
...
@@ -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/version.h
View file @
445aa844
#define REDIS_VERSION "6.
1.242
"
#define REDIS_VERSION "6.
2.0
"
#define REDIS_VERSION_NUM 0x00060
1f2
#define REDIS_VERSION_NUM 0x00060
200
src/ziplist.c
View file @
445aa844
...
@@ -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 @
445aa844
...
@@ -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 @
445aa844
...
@@ -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 @
445aa844
File moved
tests/instances.tcl
View file @
445aa844
...
@@ -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 @
445aa844
...
@@ -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 @
445aa844
...
@@ -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 @
445aa844
...
@@ -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
]
}
}
}
}
...
...
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