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
d2b5a579
Unverified
Commit
d2b5a579
authored
Feb 28, 2022
by
Oran Agra
Committed by
GitHub
Feb 28, 2022
Browse files
Merge pull request #10355 from oranagra/release-7.0-rc2
Release 7.0 RC2
parents
d5915a16
10dc57ab
Changes
144
Hide whitespace changes
Inline
Side-by-side
deps/hiredis/examples/example.c
View file @
d2b5a579
...
...
@@ -2,7 +2,10 @@
#include <stdlib.h>
#include <string.h>
#include <hiredis.h>
#include <win32.h>
#ifdef _MSC_VER
#include <winsock2.h>
/* For struct timeval */
#endif
int
main
(
int
argc
,
char
**
argv
)
{
unsigned
int
j
,
isunix
=
0
;
...
...
deps/hiredis/fmacros.h
View file @
d2b5a579
#ifndef __HIREDIS_FMACRO_H
#define __HIREDIS_FMACRO_H
#ifndef _AIX
#define _XOPEN_SOURCE 600
#define _POSIX_C_SOURCE 200112L
#endif
#if defined(__APPLE__) && defined(__MACH__)
/* Enable TCP_KEEPALIVE */
...
...
deps/hiredis/fuzzing/format_command_fuzzer.c
0 → 100644
View file @
d2b5a579
/*
* Copyright (c) 2020, Salvatore Sanfilippo <antirez at gmail dot com>
* Copyright (c) 2020, Pieter Noordhuis <pcnoordhuis at gmail dot com>
* Copyright (c) 2020, Matt Stancliff <matt at genges dot com>,
* Jan-Erik Rediger <janerik at fnordig dot com>
*
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
*
* * Redistributions of source code must retain the above copyright notice,
* this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* * Neither the name of Redis nor the names of its contributors may be used
* to endorse or promote products derived from this software without
* specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*/
#include <stdlib.h>
#include <string.h>
#include "hiredis.h"
int
LLVMFuzzerTestOneInput
(
const
uint8_t
*
data
,
size_t
size
)
{
char
*
new_str
,
*
cmd
;
if
(
size
<
3
)
return
0
;
new_str
=
malloc
(
size
+
1
);
if
(
new_str
==
NULL
)
return
0
;
memcpy
(
new_str
,
data
,
size
);
new_str
[
size
]
=
'\0'
;
redisFormatCommand
(
&
cmd
,
new_str
);
if
(
cmd
!=
NULL
)
hi_free
(
cmd
);
free
(
new_str
);
return
0
;
}
deps/hiredis/hiredis.c
View file @
d2b5a579
...
...
@@ -96,6 +96,8 @@ void freeReplyObject(void *reply) {
switch
(
r
->
type
)
{
case
REDIS_REPLY_INTEGER
:
case
REDIS_REPLY_NIL
:
case
REDIS_REPLY_BOOL
:
break
;
/* Nothing to free */
case
REDIS_REPLY_ARRAY
:
case
REDIS_REPLY_MAP
:
...
...
@@ -112,6 +114,7 @@ void freeReplyObject(void *reply) {
case
REDIS_REPLY_STRING
:
case
REDIS_REPLY_DOUBLE
:
case
REDIS_REPLY_VERB
:
case
REDIS_REPLY_BIGNUM
:
hi_free
(
r
->
str
);
break
;
}
...
...
@@ -129,7 +132,8 @@ static void *createStringObject(const redisReadTask *task, char *str, size_t len
assert
(
task
->
type
==
REDIS_REPLY_ERROR
||
task
->
type
==
REDIS_REPLY_STATUS
||
task
->
type
==
REDIS_REPLY_STRING
||
task
->
type
==
REDIS_REPLY_VERB
);
task
->
type
==
REDIS_REPLY_VERB
||
task
->
type
==
REDIS_REPLY_BIGNUM
);
/* Copy string value */
if
(
task
->
type
==
REDIS_REPLY_VERB
)
{
...
...
@@ -235,12 +239,14 @@ static void *createDoubleObject(const redisReadTask *task, double value, char *s
* decimal string conversion artifacts. */
memcpy
(
r
->
str
,
str
,
len
);
r
->
str
[
len
]
=
'\0'
;
r
->
len
=
len
;
if
(
task
->
parent
)
{
parent
=
task
->
parent
->
obj
;
assert
(
parent
->
type
==
REDIS_REPLY_ARRAY
||
parent
->
type
==
REDIS_REPLY_MAP
||
parent
->
type
==
REDIS_REPLY_SET
);
parent
->
type
==
REDIS_REPLY_SET
||
parent
->
type
==
REDIS_REPLY_PUSH
);
parent
->
element
[
task
->
idx
]
=
r
;
}
return
r
;
...
...
@@ -277,7 +283,8 @@ static void *createBoolObject(const redisReadTask *task, int bval) {
parent
=
task
->
parent
->
obj
;
assert
(
parent
->
type
==
REDIS_REPLY_ARRAY
||
parent
->
type
==
REDIS_REPLY_MAP
||
parent
->
type
==
REDIS_REPLY_SET
);
parent
->
type
==
REDIS_REPLY_SET
||
parent
->
type
==
REDIS_REPLY_PUSH
);
parent
->
element
[
task
->
idx
]
=
r
;
}
return
r
;
...
...
@@ -565,13 +572,12 @@ int redisFormatCommand(char **target, const char *format, ...) {
* lengths. If the latter is set to NULL, strlen will be used to compute the
* argument lengths.
*/
int
redisFormatSdsCommandArgv
(
hisds
*
target
,
int
argc
,
const
char
**
argv
,
const
size_t
*
argvlen
)
long
long
redisFormatSdsCommandArgv
(
hisds
*
target
,
int
argc
,
const
char
**
argv
,
const
size_t
*
argvlen
)
{
hisds
cmd
,
aux
;
unsigned
long
long
totlen
;
unsigned
long
long
totlen
,
len
;
int
j
;
size_t
len
;
/* Abort on a NULL target */
if
(
target
==
NULL
)
...
...
@@ -602,7 +608,7 @@ int redisFormatSdsCommandArgv(hisds *target, int argc, const char **argv,
cmd
=
hi_sdscatfmt
(
cmd
,
"*%i
\r\n
"
,
argc
);
for
(
j
=
0
;
j
<
argc
;
j
++
)
{
len
=
argvlen
?
argvlen
[
j
]
:
strlen
(
argv
[
j
]);
cmd
=
hi_sdscatfmt
(
cmd
,
"$%
u
\r\n
"
,
len
);
cmd
=
hi_sdscatfmt
(
cmd
,
"$%
U
\r\n
"
,
len
);
cmd
=
hi_sdscatlen
(
cmd
,
argv
[
j
],
len
);
cmd
=
hi_sdscatlen
(
cmd
,
"
\r\n
"
,
sizeof
(
"
\r\n
"
)
-
1
);
}
...
...
@@ -622,11 +628,11 @@ void redisFreeSdsCommand(hisds cmd) {
* lengths. If the latter is set to NULL, strlen will be used to compute the
* argument lengths.
*/
int
redisFormatCommandArgv
(
char
**
target
,
int
argc
,
const
char
**
argv
,
const
size_t
*
argvlen
)
{
long
long
redisFormatCommandArgv
(
char
**
target
,
int
argc
,
const
char
**
argv
,
const
size_t
*
argvlen
)
{
char
*
cmd
=
NULL
;
/* final command */
in
t
pos
;
/* position in final command */
size_t
len
;
int
totlen
,
j
;
size_
t
pos
;
/* position in final command */
size_t
len
,
totlen
;
int
j
;
/* Abort on a NULL target */
if
(
target
==
NULL
)
...
...
@@ -797,6 +803,9 @@ redisContext *redisConnectWithOptions(const redisOptions *options) {
if
(
options
->
options
&
REDIS_OPT_NOAUTOFREE
)
{
c
->
flags
|=
REDIS_NO_AUTO_FREE
;
}
if
(
options
->
options
&
REDIS_OPT_NOAUTOFREEREPLIES
)
{
c
->
flags
|=
REDIS_NO_AUTO_FREE_REPLIES
;
}
/* Set any user supplied RESP3 PUSH handler or use freeReplyObject
* as a default unless specifically flagged that we don't want one. */
...
...
@@ -825,7 +834,7 @@ redisContext *redisConnectWithOptions(const redisOptions *options) {
c
->
fd
=
options
->
endpoint
.
fd
;
c
->
flags
|=
REDIS_CONNECTED
;
}
else
{
// Unknown type - FIXME - FREE
redisFree
(
c
);
return
NULL
;
}
...
...
@@ -939,13 +948,11 @@ int redisBufferRead(redisContext *c) {
return
REDIS_ERR
;
nread
=
c
->
funcs
->
read
(
c
,
buf
,
sizeof
(
buf
));
if
(
nread
>
0
)
{
if
(
redisReaderFeed
(
c
->
reader
,
buf
,
nread
)
!=
REDIS_OK
)
{
__redisSetError
(
c
,
c
->
reader
->
err
,
c
->
reader
->
errstr
);
return
REDIS_ERR
;
}
else
{
}
}
else
if
(
nread
<
0
)
{
if
(
nread
<
0
)
{
return
REDIS_ERR
;
}
if
(
nread
>
0
&&
redisReaderFeed
(
c
->
reader
,
buf
,
nread
)
!=
REDIS_OK
)
{
__redisSetError
(
c
,
c
->
reader
->
err
,
c
->
reader
->
errstr
);
return
REDIS_ERR
;
}
return
REDIS_OK
;
...
...
@@ -989,17 +996,6 @@ oom:
return
REDIS_ERR
;
}
/* Internal helper function to try and get a reply from the reader,
* or set an error in the context otherwise. */
int
redisGetReplyFromReader
(
redisContext
*
c
,
void
**
reply
)
{
if
(
redisReaderGetReply
(
c
->
reader
,
reply
)
==
REDIS_ERR
)
{
__redisSetError
(
c
,
c
->
reader
->
err
,
c
->
reader
->
errstr
);
return
REDIS_ERR
;
}
return
REDIS_OK
;
}
/* Internal helper that returns 1 if the reply was a RESP3 PUSH
* message and we handled it with a user-provided callback. */
static
int
redisHandledPushReply
(
redisContext
*
c
,
void
*
reply
)
{
...
...
@@ -1011,12 +1007,34 @@ static int redisHandledPushReply(redisContext *c, void *reply) {
return
0
;
}
/* Get a reply from our reader or set an error in the context. */
int
redisGetReplyFromReader
(
redisContext
*
c
,
void
**
reply
)
{
if
(
redisReaderGetReply
(
c
->
reader
,
reply
)
==
REDIS_ERR
)
{
__redisSetError
(
c
,
c
->
reader
->
err
,
c
->
reader
->
errstr
);
return
REDIS_ERR
;
}
return
REDIS_OK
;
}
/* Internal helper to get the next reply from our reader while handling
* any PUSH messages we encounter along the way. This is separate from
* redisGetReplyFromReader so as to not change its behavior. */
static
int
redisNextInBandReplyFromReader
(
redisContext
*
c
,
void
**
reply
)
{
do
{
if
(
redisGetReplyFromReader
(
c
,
reply
)
==
REDIS_ERR
)
return
REDIS_ERR
;
}
while
(
redisHandledPushReply
(
c
,
*
reply
));
return
REDIS_OK
;
}
int
redisGetReply
(
redisContext
*
c
,
void
**
reply
)
{
int
wdone
=
0
;
void
*
aux
=
NULL
;
/* Try to read pending replies */
if
(
redis
Get
ReplyFromReader
(
c
,
&
aux
)
==
REDIS_ERR
)
if
(
redis
NextInBand
ReplyFromReader
(
c
,
&
aux
)
==
REDIS_ERR
)
return
REDIS_ERR
;
/* For the blocking context, flush output buffer and read reply */
...
...
@@ -1032,12 +1050,8 @@ int redisGetReply(redisContext *c, void **reply) {
if
(
redisBufferRead
(
c
)
==
REDIS_ERR
)
return
REDIS_ERR
;
/* We loop here in case the user has specified a RESP3
* PUSH handler (e.g. for client tracking). */
do
{
if
(
redisGetReplyFromReader
(
c
,
&
aux
)
==
REDIS_ERR
)
return
REDIS_ERR
;
}
while
(
redisHandledPushReply
(
c
,
aux
));
if
(
redisNextInBandReplyFromReader
(
c
,
&
aux
)
==
REDIS_ERR
)
return
REDIS_ERR
;
}
while
(
aux
==
NULL
);
}
...
...
@@ -1114,7 +1128,7 @@ int redisAppendCommand(redisContext *c, const char *format, ...) {
int
redisAppendCommandArgv
(
redisContext
*
c
,
int
argc
,
const
char
**
argv
,
const
size_t
*
argvlen
)
{
hisds
cmd
;
int
len
;
long
long
len
;
len
=
redisFormatSdsCommandArgv
(
&
cmd
,
argc
,
argv
,
argvlen
);
if
(
len
==
-
1
)
{
...
...
deps/hiredis/hiredis.h
View file @
d2b5a579
...
...
@@ -47,8 +47,8 @@ typedef long long ssize_t;
#define HIREDIS_MAJOR 1
#define HIREDIS_MINOR 0
#define HIREDIS_PATCH
0
#define HIREDIS_SONAME 1.0.
0
#define HIREDIS_PATCH
3
#define HIREDIS_SONAME 1.0.
3-dev
/* Connection type can be blocking or non-blocking and is set in the
* least significant bit of the flags field in redisContext. */
...
...
@@ -80,12 +80,18 @@ typedef long long ssize_t;
/* Flag that is set when we should set SO_REUSEADDR before calling bind() */
#define REDIS_REUSEADDR 0x80
/* Flag that is set when the async connection supports push replies. */
#define REDIS_SUPPORTS_PUSH 0x100
/**
* Flag that indicates the user does not want the context to
* be automatically freed upon error
*/
#define REDIS_NO_AUTO_FREE 0x200
/* Flag that indicates the user does not want replies to be automatically freed */
#define REDIS_NO_AUTO_FREE_REPLIES 0x400
#define REDIS_KEEPALIVE_INTERVAL 15
/* seconds */
/* number of times we retry to connect in the case of EADDRNOTAVAIL and
...
...
@@ -112,7 +118,8 @@ typedef struct redisReply {
double
dval
;
/* The double when type is REDIS_REPLY_DOUBLE */
size_t
len
;
/* Length of string */
char
*
str
;
/* Used for REDIS_REPLY_ERROR, REDIS_REPLY_STRING
REDIS_REPLY_VERB, and REDIS_REPLY_DOUBLE (in additional to dval). */
REDIS_REPLY_VERB, REDIS_REPLY_DOUBLE (in additional to dval),
and REDIS_REPLY_BIGNUM. */
char
vtype
[
4
];
/* Used for REDIS_REPLY_VERB, contains the null
terminated 3 character content type, such as "txt". */
size_t
elements
;
/* number of elements, for REDIS_REPLY_ARRAY */
...
...
@@ -127,8 +134,8 @@ void freeReplyObject(void *reply);
/* Functions to format a command according to the protocol. */
int
redisvFormatCommand
(
char
**
target
,
const
char
*
format
,
va_list
ap
);
int
redisFormatCommand
(
char
**
target
,
const
char
*
format
,
...);
int
redisFormatCommandArgv
(
char
**
target
,
int
argc
,
const
char
**
argv
,
const
size_t
*
argvlen
);
int
redisFormatSdsCommandArgv
(
hisds
*
target
,
int
argc
,
const
char
**
argv
,
const
size_t
*
argvlen
);
long
long
redisFormatCommandArgv
(
char
**
target
,
int
argc
,
const
char
**
argv
,
const
size_t
*
argvlen
);
long
long
redisFormatSdsCommandArgv
(
hisds
*
target
,
int
argc
,
const
char
**
argv
,
const
size_t
*
argvlen
);
void
redisFreeCommand
(
char
*
cmd
);
void
redisFreeSdsCommand
(
hisds
cmd
);
...
...
@@ -152,6 +159,11 @@ struct redisSsl;
/* Don't automatically intercept and free RESP3 PUSH replies. */
#define REDIS_OPT_NO_PUSH_AUTOFREE 0x08
/**
* Don't automatically free replies
*/
#define REDIS_OPT_NOAUTOFREEREPLIES 0x10
/* In Unix systems a file descriptor is a regular signed int, with -1
* representing an invalid descriptor. In Windows it is a SOCKET
* (32- or 64-bit unsigned integer depending on the architecture), where
...
...
@@ -255,7 +267,7 @@ typedef struct redisContext {
}
unix_sock
;
/* For non-blocking connect */
struct
sockadr
*
saddr
;
struct
sockad
d
r
*
saddr
;
size_t
addrlen
;
/* Optional data and corresponding destructor users can use to provide
...
...
deps/hiredis/hiredis.targets
0 → 100644
View file @
d2b5a579
<?xml version="1.0" encoding="utf-8"?>
<Project
ToolsVersion=
"4.0"
xmlns=
"http://schemas.microsoft.com/developer/msbuild/2003"
>
<ItemDefinitionGroup>
<ClCompile>
<AdditionalIncludeDirectories>
$(MSBuildThisFileDirectory)\..\..\include;%(AdditionalIncludeDirectories)
</AdditionalIncludeDirectories>
</ClCompile>
<Link>
<AdditionalLibraryDirectories>
$(MSBuildThisFileDirectory)\..\..\lib;%(AdditionalLibraryDirectories)
</AdditionalLibraryDirectories>
</Link>
</ItemDefinitionGroup>
</Project>
\ No newline at end of file
deps/hiredis/hiredis_ssl.h
View file @
d2b5a579
...
...
@@ -56,7 +56,9 @@ typedef enum {
REDIS_SSL_CTX_CERT_KEY_REQUIRED
,
/* Client cert and key must both be specified or skipped */
REDIS_SSL_CTX_CA_CERT_LOAD_FAILED
,
/* Failed to load CA Certificate or CA Path */
REDIS_SSL_CTX_CLIENT_CERT_LOAD_FAILED
,
/* Failed to load client certificate */
REDIS_SSL_CTX_PRIVATE_KEY_LOAD_FAILED
/* Failed to load private key */
REDIS_SSL_CTX_PRIVATE_KEY_LOAD_FAILED
,
/* Failed to load private key */
REDIS_SSL_CTX_OS_CERTSTORE_OPEN_FAILED
,
/* Failed to open system certifcate store */
REDIS_SSL_CTX_OS_CERT_ADD_FAILED
/* Failed to add CA certificates obtained from system to the SSL context */
}
redisSSLContextError
;
/**
...
...
deps/hiredis/read.c
View file @
d2b5a579
...
...
@@ -123,29 +123,28 @@ static char *readBytes(redisReader *r, unsigned int bytes) {
/* Find pointer to \r\n. */
static
char
*
seekNewline
(
char
*
s
,
size_t
len
)
{
int
pos
=
0
;
int
_len
=
len
-
1
;
/* Position should be < len-1 because the character at "pos" should be
* followed by a \n. Note that strchr cannot be used because it doesn't
* allow to search a limited length and the buffer that is being searched
* might not have a trailing NULL character. */
while
(
pos
<
_len
)
{
while
(
pos
<
_len
&&
s
[
pos
]
!=
'\r'
)
pos
++
;
if
(
pos
==
_len
)
{
/* Not found. */
return
NULL
;
}
else
{
if
(
s
[
pos
+
1
]
==
'\n'
)
{
/* Found. */
return
s
+
pos
;
}
else
{
/* Continue searching. */
pos
++
;
}
char
*
ret
;
/* We cannot match with fewer than 2 bytes */
if
(
len
<
2
)
return
NULL
;
/* Search up to len - 1 characters */
len
--
;
/* Look for the \r */
while
((
ret
=
memchr
(
s
,
'\r'
,
len
))
!=
NULL
)
{
if
(
ret
[
1
]
==
'\n'
)
{
/* Found. */
break
;
}
/* Continue searching. */
ret
++
;
len
-=
ret
-
s
;
s
=
ret
;
}
return
NULL
;
return
ret
;
}
/* Convert a string into a long long. Returns REDIS_OK if the string could be
...
...
@@ -274,60 +273,104 @@ static int processLineItem(redisReader *r) {
if
((
p
=
readLine
(
r
,
&
len
))
!=
NULL
)
{
if
(
cur
->
type
==
REDIS_REPLY_INTEGER
)
{
long
long
v
;
if
(
string2ll
(
p
,
len
,
&
v
)
==
REDIS_ERR
)
{
__redisReaderSetError
(
r
,
REDIS_ERR_PROTOCOL
,
"Bad integer value"
);
return
REDIS_ERR
;
}
if
(
r
->
fn
&&
r
->
fn
->
createInteger
)
{
long
long
v
;
if
(
string2ll
(
p
,
len
,
&
v
)
==
REDIS_ERR
)
{
__redisReaderSetError
(
r
,
REDIS_ERR_PROTOCOL
,
"Bad integer value"
);
return
REDIS_ERR
;
}
obj
=
r
->
fn
->
createInteger
(
cur
,
v
);
}
else
{
obj
=
(
void
*
)
REDIS_REPLY_INTEGER
;
}
}
else
if
(
cur
->
type
==
REDIS_REPLY_DOUBLE
)
{
if
(
r
->
fn
&&
r
->
fn
->
createDouble
)
{
char
buf
[
326
],
*
eptr
;
double
d
;
char
buf
[
326
],
*
eptr
;
double
d
;
if
((
size_t
)
len
>=
sizeof
(
buf
))
{
if
((
size_t
)
len
>=
sizeof
(
buf
))
{
__redisReaderSetError
(
r
,
REDIS_ERR_PROTOCOL
,
"Double value is too large"
);
return
REDIS_ERR
;
}
memcpy
(
buf
,
p
,
len
);
buf
[
len
]
=
'\0'
;
if
(
len
==
3
&&
strcasecmp
(
buf
,
"inf"
)
==
0
)
{
d
=
INFINITY
;
/* Positive infinite. */
}
else
if
(
len
==
4
&&
strcasecmp
(
buf
,
"-inf"
)
==
0
)
{
d
=
-
INFINITY
;
/* Negative infinite. */
}
else
{
d
=
strtod
((
char
*
)
buf
,
&
eptr
);
/* RESP3 only allows "inf", "-inf", and finite values, while
* strtod() allows other variations on infinity, NaN,
* etc. We explicity handle our two allowed infinite cases
* above, so strtod() should only result in finite values. */
if
(
buf
[
0
]
==
'\0'
||
eptr
!=
&
buf
[
len
]
||
!
isfinite
(
d
))
{
__redisReaderSetError
(
r
,
REDIS_ERR_PROTOCOL
,
"
D
ouble value
is too large
"
);
"
Bad d
ouble value"
);
return
REDIS_ERR
;
}
}
memcpy
(
buf
,
p
,
len
);
buf
[
len
]
=
'\0'
;
if
(
strcasecmp
(
buf
,
",inf"
)
==
0
)
{
d
=
INFINITY
;
/* Positive infinite. */
}
else
if
(
strcasecmp
(
buf
,
",-inf"
)
==
0
)
{
d
=
-
INFINITY
;
/* Negative infinite. */
}
else
{
d
=
strtod
((
char
*
)
buf
,
&
eptr
);
if
(
buf
[
0
]
==
'\0'
||
eptr
[
0
]
!=
'\0'
||
isnan
(
d
))
{
__redisReaderSetError
(
r
,
REDIS_ERR_PROTOCOL
,
"Bad double value"
);
return
REDIS_ERR
;
}
}
if
(
r
->
fn
&&
r
->
fn
->
createDouble
)
{
obj
=
r
->
fn
->
createDouble
(
cur
,
d
,
buf
,
len
);
}
else
{
obj
=
(
void
*
)
REDIS_REPLY_DOUBLE
;
}
}
else
if
(
cur
->
type
==
REDIS_REPLY_NIL
)
{
if
(
len
!=
0
)
{
__redisReaderSetError
(
r
,
REDIS_ERR_PROTOCOL
,
"Bad nil value"
);
return
REDIS_ERR
;
}
if
(
r
->
fn
&&
r
->
fn
->
createNil
)
obj
=
r
->
fn
->
createNil
(
cur
);
else
obj
=
(
void
*
)
REDIS_REPLY_NIL
;
}
else
if
(
cur
->
type
==
REDIS_REPLY_BOOL
)
{
int
bval
=
p
[
0
]
==
't'
||
p
[
0
]
==
'T'
;
int
bval
;
if
(
len
!=
1
||
!
strchr
(
"tTfF"
,
p
[
0
]))
{
__redisReaderSetError
(
r
,
REDIS_ERR_PROTOCOL
,
"Bad bool value"
);
return
REDIS_ERR
;
}
bval
=
p
[
0
]
==
't'
||
p
[
0
]
==
'T'
;
if
(
r
->
fn
&&
r
->
fn
->
createBool
)
obj
=
r
->
fn
->
createBool
(
cur
,
bval
);
else
obj
=
(
void
*
)
REDIS_REPLY_BOOL
;
}
else
if
(
cur
->
type
==
REDIS_REPLY_BIGNUM
)
{
/* Ensure all characters are decimal digits (with possible leading
* minus sign). */
for
(
int
i
=
0
;
i
<
len
;
i
++
)
{
/* XXX Consider: Allow leading '+'? Error on leading '0's? */
if
(
i
==
0
&&
p
[
0
]
==
'-'
)
continue
;
if
(
p
[
i
]
<
'0'
||
p
[
i
]
>
'9'
)
{
__redisReaderSetError
(
r
,
REDIS_ERR_PROTOCOL
,
"Bad bignum value"
);
return
REDIS_ERR
;
}
}
if
(
r
->
fn
&&
r
->
fn
->
createString
)
obj
=
r
->
fn
->
createString
(
cur
,
p
,
len
);
else
obj
=
(
void
*
)
REDIS_REPLY_BIGNUM
;
}
else
{
/* Type will be error or status. */
for
(
int
i
=
0
;
i
<
len
;
i
++
)
{
if
(
p
[
i
]
==
'\r'
||
p
[
i
]
==
'\n'
)
{
__redisReaderSetError
(
r
,
REDIS_ERR_PROTOCOL
,
"Bad simple string value"
);
return
REDIS_ERR
;
}
}
if
(
r
->
fn
&&
r
->
fn
->
createString
)
obj
=
r
->
fn
->
createString
(
cur
,
p
,
len
);
else
...
...
@@ -453,7 +496,6 @@ static int processAggregateItem(redisReader *r) {
long
long
elements
;
int
root
=
0
,
len
;
/* Set error for nested multi bulks with depth > 7 */
if
(
r
->
ridx
==
r
->
tasks
-
1
)
{
if
(
redisReaderGrow
(
r
)
==
REDIS_ERR
)
return
REDIS_ERR
;
...
...
@@ -569,6 +611,9 @@ static int processItem(redisReader *r) {
case
'>'
:
cur
->
type
=
REDIS_REPLY_PUSH
;
break
;
case
'('
:
cur
->
type
=
REDIS_REPLY_BIGNUM
;
break
;
default:
__redisReaderSetErrorProtocolByte
(
r
,
*
p
);
return
REDIS_ERR
;
...
...
@@ -587,6 +632,7 @@ static int processItem(redisReader *r) {
case
REDIS_REPLY_DOUBLE
:
case
REDIS_REPLY_NIL
:
case
REDIS_REPLY_BOOL
:
case
REDIS_REPLY_BIGNUM
:
return
processLineItem
(
r
);
case
REDIS_REPLY_STRING
:
case
REDIS_REPLY_VERB
:
...
...
deps/hiredis/sds.c
View file @
d2b5a579
...
...
@@ -72,7 +72,7 @@ static inline char hi_sdsReqType(size_t string_size) {
* and 'initlen'.
* If NULL is used for 'init' the string is initialized with zero bytes.
*
* The string is always null-termined (all the hisds strings are, always) so
* The string is always null-termin
at
ed (all the hisds strings are, always) so
* even if you create an hisds string with:
*
* mystring = hi_sdsnewlen("abc",3);
...
...
@@ -415,7 +415,7 @@ hisds hi_sdscpylen(hisds s, const char *t, size_t len) {
return
s
;
}
/* Like hi_sdscpylen() but 't' must be a null-termined string so that the length
/* Like hi_sdscpylen() but 't' must be a null-termin
at
ed string so that the length
* of the string is obtained with strlen(). */
hisds
hi_sdscpy
(
hisds
s
,
const
char
*
t
)
{
return
hi_sdscpylen
(
s
,
t
,
strlen
(
t
));
...
...
deps/hiredis/ssl.c
View file @
d2b5a579
...
...
@@ -38,6 +38,7 @@
#include <string.h>
#ifdef _WIN32
#include <windows.h>
#include <wincrypt.h>
#else
#include <pthread.h>
#endif
...
...
@@ -182,6 +183,10 @@ const char *redisSSLContextGetError(redisSSLContextError error)
return
"Failed to load client certificate"
;
case
REDIS_SSL_CTX_PRIVATE_KEY_LOAD_FAILED
:
return
"Failed to load private key"
;
case
REDIS_SSL_CTX_OS_CERTSTORE_OPEN_FAILED
:
return
"Failed to open system certifcate store"
;
case
REDIS_SSL_CTX_OS_CERT_ADD_FAILED
:
return
"Failed to add CA certificates obtained from system to the SSL context"
;
default:
return
"Unknown error code"
;
}
...
...
@@ -214,6 +219,11 @@ redisSSLContext *redisCreateSSLContext(const char *cacert_filename, const char *
const
char
*
cert_filename
,
const
char
*
private_key_filename
,
const
char
*
server_name
,
redisSSLContextError
*
error
)
{
#ifdef _WIN32
HCERTSTORE
win_store
=
NULL
;
PCCERT_CONTEXT
win_ctx
=
NULL
;
#endif
redisSSLContext
*
ctx
=
hi_calloc
(
1
,
sizeof
(
redisSSLContext
));
if
(
ctx
==
NULL
)
goto
error
;
...
...
@@ -234,6 +244,31 @@ redisSSLContext *redisCreateSSLContext(const char *cacert_filename, const char *
}
if
(
capath
||
cacert_filename
)
{
#ifdef _WIN32
if
(
0
==
strcmp
(
cacert_filename
,
"wincert"
))
{
win_store
=
CertOpenSystemStore
(
NULL
,
"Root"
);
if
(
!
win_store
)
{
if
(
error
)
*
error
=
REDIS_SSL_CTX_OS_CERTSTORE_OPEN_FAILED
;
goto
error
;
}
X509_STORE
*
store
=
SSL_CTX_get_cert_store
(
ctx
->
ssl_ctx
);
while
(
win_ctx
=
CertEnumCertificatesInStore
(
win_store
,
win_ctx
))
{
X509
*
x509
=
NULL
;
x509
=
d2i_X509
(
NULL
,
(
const
unsigned
char
**
)
&
win_ctx
->
pbCertEncoded
,
win_ctx
->
cbCertEncoded
);
if
(
x509
)
{
if
((
1
!=
X509_STORE_add_cert
(
store
,
x509
))
||
(
1
!=
SSL_CTX_add_client_CA
(
ctx
->
ssl_ctx
,
x509
)))
{
if
(
error
)
*
error
=
REDIS_SSL_CTX_OS_CERT_ADD_FAILED
;
goto
error
;
}
X509_free
(
x509
);
}
}
CertFreeCertificateContext
(
win_ctx
);
CertCloseStore
(
win_store
,
0
);
}
else
#endif
if
(
!
SSL_CTX_load_verify_locations
(
ctx
->
ssl_ctx
,
cacert_filename
,
capath
))
{
if
(
error
)
*
error
=
REDIS_SSL_CTX_CA_CERT_LOAD_FAILED
;
goto
error
;
...
...
@@ -257,6 +292,10 @@ redisSSLContext *redisCreateSSLContext(const char *cacert_filename, const char *
return
ctx
;
error:
#ifdef _WIN32
CertFreeCertificateContext
(
win_ctx
);
CertCloseStore
(
win_store
,
0
);
#endif
redisFreeSSLContext
(
ctx
);
return
NULL
;
}
...
...
@@ -353,7 +392,11 @@ int redisInitiateSSLWithContext(redisContext *c, redisSSLContext *redis_ssl_ctx)
}
}
return
redisSSLConnect
(
c
,
ssl
);
if
(
redisSSLConnect
(
c
,
ssl
)
!=
REDIS_OK
)
{
goto
error
;
}
return
REDIS_OK
;
error:
if
(
ssl
)
...
...
deps/hiredis/test.c
View file @
d2b5a579
...
...
@@ -11,12 +11,17 @@
#include <signal.h>
#include <errno.h>
#include <limits.h>
#include <math.h>
#include "hiredis.h"
#include "async.h"
#ifdef HIREDIS_TEST_SSL
#include "hiredis_ssl.h"
#endif
#ifdef HIREDIS_TEST_ASYNC
#include "adapters/libevent.h"
#include <event2/event.h>
#endif
#include "net.h"
#include "win32.h"
...
...
@@ -58,6 +63,8 @@ struct pushCounters {
int
str
;
};
static
int
insecure_calloc_calls
;
#ifdef HIREDIS_TEST_SSL
redisSSLContext
*
_ssl_ctx
=
NULL
;
#endif
...
...
@@ -597,6 +604,147 @@ static void test_reply_reader(void) {
((
redisReply
*
)
reply
)
->
element
[
1
]
->
integer
==
42
);
freeReplyObject
(
reply
);
redisReaderFree
(
reader
);
test
(
"Can parse RESP3 doubles: "
);
reader
=
redisReaderCreate
();
redisReaderFeed
(
reader
,
",3.14159265358979323846
\r\n
"
,
25
);
ret
=
redisReaderGetReply
(
reader
,
&
reply
);
test_cond
(
ret
==
REDIS_OK
&&
((
redisReply
*
)
reply
)
->
type
==
REDIS_REPLY_DOUBLE
&&
fabs
(((
redisReply
*
)
reply
)
->
dval
-
3
.
14159265358979323846
)
<
0
.
00000001
&&
((
redisReply
*
)
reply
)
->
len
==
22
&&
strcmp
(((
redisReply
*
)
reply
)
->
str
,
"3.14159265358979323846"
)
==
0
);
freeReplyObject
(
reply
);
redisReaderFree
(
reader
);
test
(
"Set error on invalid RESP3 double: "
);
reader
=
redisReaderCreate
();
redisReaderFeed
(
reader
,
",3.14159
\000
265358979323846
\r\n
"
,
26
);
ret
=
redisReaderGetReply
(
reader
,
&
reply
);
test_cond
(
ret
==
REDIS_ERR
&&
strcasecmp
(
reader
->
errstr
,
"Bad double value"
)
==
0
);
freeReplyObject
(
reply
);
redisReaderFree
(
reader
);
test
(
"Correctly parses RESP3 double INFINITY: "
);
reader
=
redisReaderCreate
();
redisReaderFeed
(
reader
,
",inf
\r\n
"
,
6
);
ret
=
redisReaderGetReply
(
reader
,
&
reply
);
test_cond
(
ret
==
REDIS_OK
&&
((
redisReply
*
)
reply
)
->
type
==
REDIS_REPLY_DOUBLE
&&
isinf
(((
redisReply
*
)
reply
)
->
dval
)
&&
((
redisReply
*
)
reply
)
->
dval
>
0
);
freeReplyObject
(
reply
);
redisReaderFree
(
reader
);
test
(
"Set error when RESP3 double is NaN: "
);
reader
=
redisReaderCreate
();
redisReaderFeed
(
reader
,
",nan
\r\n
"
,
6
);
ret
=
redisReaderGetReply
(
reader
,
&
reply
);
test_cond
(
ret
==
REDIS_ERR
&&
strcasecmp
(
reader
->
errstr
,
"Bad double value"
)
==
0
);
freeReplyObject
(
reply
);
redisReaderFree
(
reader
);
test
(
"Can parse RESP3 nil: "
);
reader
=
redisReaderCreate
();
redisReaderFeed
(
reader
,
"_
\r\n
"
,
3
);
ret
=
redisReaderGetReply
(
reader
,
&
reply
);
test_cond
(
ret
==
REDIS_OK
&&
((
redisReply
*
)
reply
)
->
type
==
REDIS_REPLY_NIL
);
freeReplyObject
(
reply
);
redisReaderFree
(
reader
);
test
(
"Set error on invalid RESP3 nil: "
);
reader
=
redisReaderCreate
();
redisReaderFeed
(
reader
,
"_nil
\r\n
"
,
6
);
ret
=
redisReaderGetReply
(
reader
,
&
reply
);
test_cond
(
ret
==
REDIS_ERR
&&
strcasecmp
(
reader
->
errstr
,
"Bad nil value"
)
==
0
);
freeReplyObject
(
reply
);
redisReaderFree
(
reader
);
test
(
"Can parse RESP3 bool (true): "
);
reader
=
redisReaderCreate
();
redisReaderFeed
(
reader
,
"#t
\r\n
"
,
4
);
ret
=
redisReaderGetReply
(
reader
,
&
reply
);
test_cond
(
ret
==
REDIS_OK
&&
((
redisReply
*
)
reply
)
->
type
==
REDIS_REPLY_BOOL
&&
((
redisReply
*
)
reply
)
->
integer
);
freeReplyObject
(
reply
);
redisReaderFree
(
reader
);
test
(
"Can parse RESP3 bool (false): "
);
reader
=
redisReaderCreate
();
redisReaderFeed
(
reader
,
"#f
\r\n
"
,
4
);
ret
=
redisReaderGetReply
(
reader
,
&
reply
);
test_cond
(
ret
==
REDIS_OK
&&
((
redisReply
*
)
reply
)
->
type
==
REDIS_REPLY_BOOL
&&
!
((
redisReply
*
)
reply
)
->
integer
);
freeReplyObject
(
reply
);
redisReaderFree
(
reader
);
test
(
"Set error on invalid RESP3 bool: "
);
reader
=
redisReaderCreate
();
redisReaderFeed
(
reader
,
"#foobar
\r\n
"
,
9
);
ret
=
redisReaderGetReply
(
reader
,
&
reply
);
test_cond
(
ret
==
REDIS_ERR
&&
strcasecmp
(
reader
->
errstr
,
"Bad bool value"
)
==
0
);
freeReplyObject
(
reply
);
redisReaderFree
(
reader
);
test
(
"Can parse RESP3 map: "
);
reader
=
redisReaderCreate
();
redisReaderFeed
(
reader
,
"%2
\r\n
+first
\r\n
:123
\r\n
$6
\r\n
second
\r\n
#t
\r\n
"
,
34
);
ret
=
redisReaderGetReply
(
reader
,
&
reply
);
test_cond
(
ret
==
REDIS_OK
&&
((
redisReply
*
)
reply
)
->
type
==
REDIS_REPLY_MAP
&&
((
redisReply
*
)
reply
)
->
elements
==
4
&&
((
redisReply
*
)
reply
)
->
element
[
0
]
->
type
==
REDIS_REPLY_STATUS
&&
((
redisReply
*
)
reply
)
->
element
[
0
]
->
len
==
5
&&
!
strcmp
(((
redisReply
*
)
reply
)
->
element
[
0
]
->
str
,
"first"
)
&&
((
redisReply
*
)
reply
)
->
element
[
1
]
->
type
==
REDIS_REPLY_INTEGER
&&
((
redisReply
*
)
reply
)
->
element
[
1
]
->
integer
==
123
&&
((
redisReply
*
)
reply
)
->
element
[
2
]
->
type
==
REDIS_REPLY_STRING
&&
((
redisReply
*
)
reply
)
->
element
[
2
]
->
len
==
6
&&
!
strcmp
(((
redisReply
*
)
reply
)
->
element
[
2
]
->
str
,
"second"
)
&&
((
redisReply
*
)
reply
)
->
element
[
3
]
->
type
==
REDIS_REPLY_BOOL
&&
((
redisReply
*
)
reply
)
->
element
[
3
]
->
integer
);
freeReplyObject
(
reply
);
redisReaderFree
(
reader
);
test
(
"Can parse RESP3 set: "
);
reader
=
redisReaderCreate
();
redisReaderFeed
(
reader
,
"~5
\r\n
+orange
\r\n
$5
\r\n
apple
\r\n
#f
\r\n
:100
\r\n
:999
\r\n
"
,
40
);
ret
=
redisReaderGetReply
(
reader
,
&
reply
);
test_cond
(
ret
==
REDIS_OK
&&
((
redisReply
*
)
reply
)
->
type
==
REDIS_REPLY_SET
&&
((
redisReply
*
)
reply
)
->
elements
==
5
&&
((
redisReply
*
)
reply
)
->
element
[
0
]
->
type
==
REDIS_REPLY_STATUS
&&
((
redisReply
*
)
reply
)
->
element
[
0
]
->
len
==
6
&&
!
strcmp
(((
redisReply
*
)
reply
)
->
element
[
0
]
->
str
,
"orange"
)
&&
((
redisReply
*
)
reply
)
->
element
[
1
]
->
type
==
REDIS_REPLY_STRING
&&
((
redisReply
*
)
reply
)
->
element
[
1
]
->
len
==
5
&&
!
strcmp
(((
redisReply
*
)
reply
)
->
element
[
1
]
->
str
,
"apple"
)
&&
((
redisReply
*
)
reply
)
->
element
[
2
]
->
type
==
REDIS_REPLY_BOOL
&&
!
((
redisReply
*
)
reply
)
->
element
[
2
]
->
integer
&&
((
redisReply
*
)
reply
)
->
element
[
3
]
->
type
==
REDIS_REPLY_INTEGER
&&
((
redisReply
*
)
reply
)
->
element
[
3
]
->
integer
==
100
&&
((
redisReply
*
)
reply
)
->
element
[
4
]
->
type
==
REDIS_REPLY_INTEGER
&&
((
redisReply
*
)
reply
)
->
element
[
4
]
->
integer
==
999
);
freeReplyObject
(
reply
);
redisReaderFree
(
reader
);
test
(
"Can parse RESP3 bignum: "
);
reader
=
redisReaderCreate
();
redisReaderFeed
(
reader
,
"(3492890328409238509324850943850943825024385
\r\n
"
,
46
);
ret
=
redisReaderGetReply
(
reader
,
&
reply
);
test_cond
(
ret
==
REDIS_OK
&&
((
redisReply
*
)
reply
)
->
type
==
REDIS_REPLY_BIGNUM
&&
((
redisReply
*
)
reply
)
->
len
==
43
&&
!
strcmp
(((
redisReply
*
)
reply
)
->
str
,
"3492890328409238509324850943850943825024385"
));
freeReplyObject
(
reply
);
redisReaderFree
(
reader
);
}
static
void
test_free_null
(
void
)
{
...
...
@@ -623,6 +771,13 @@ static void *hi_calloc_fail(size_t nmemb, size_t size) {
return
NULL
;
}
static
void
*
hi_calloc_insecure
(
size_t
nmemb
,
size_t
size
)
{
(
void
)
nmemb
;
(
void
)
size
;
insecure_calloc_calls
++
;
return
(
void
*
)
0xdeadc0de
;
}
static
void
*
hi_realloc_fail
(
void
*
ptr
,
size_t
size
)
{
(
void
)
ptr
;
(
void
)
size
;
...
...
@@ -630,6 +785,8 @@ static void *hi_realloc_fail(void *ptr, size_t size) {
}
static
void
test_allocator_injection
(
void
)
{
void
*
ptr
;
hiredisAllocFuncs
ha
=
{
.
mallocFn
=
hi_malloc_fail
,
.
callocFn
=
hi_calloc_fail
,
...
...
@@ -649,6 +806,13 @@ static void test_allocator_injection(void) {
redisReader
*
reader
=
redisReaderCreate
();
test_cond
(
reader
==
NULL
);
/* Make sure hiredis itself protects against a non-overflow checking calloc */
test
(
"hiredis calloc wrapper protects against overflow: "
);
ha
.
callocFn
=
hi_calloc_insecure
;
hiredisSetAllocators
(
&
ha
);
ptr
=
hi_calloc
((
SIZE_MAX
/
sizeof
(
void
*
))
+
3
,
sizeof
(
void
*
));
test_cond
(
ptr
==
NULL
&&
insecure_calloc_calls
==
0
);
// Return allocators to default
hiredisResetAllocators
();
}
...
...
@@ -1283,6 +1447,440 @@ static void test_throughput(struct config config) {
// redisFree(c);
// }
#ifdef HIREDIS_TEST_ASYNC
struct
event_base
*
base
;
typedef
struct
TestState
{
redisOptions
*
options
;
int
checkpoint
;
int
resp3
;
int
disconnect
;
}
TestState
;
/* Helper to disconnect and stop event loop */
void
async_disconnect
(
redisAsyncContext
*
ac
)
{
redisAsyncDisconnect
(
ac
);
event_base_loopbreak
(
base
);
}
/* Testcase timeout, will trigger a failure */
void
timeout_cb
(
int
fd
,
short
event
,
void
*
arg
)
{
(
void
)
fd
;
(
void
)
event
;
(
void
)
arg
;
printf
(
"Timeout in async testing!
\n
"
);
exit
(
1
);
}
/* Unexpected call, will trigger a failure */
void
unexpected_cb
(
redisAsyncContext
*
ac
,
void
*
r
,
void
*
privdata
)
{
(
void
)
ac
;
(
void
)
r
;
printf
(
"Unexpected call: %s
\n
"
,(
char
*
)
privdata
);
exit
(
1
);
}
/* Helper function to publish a message via own client. */
void
publish_msg
(
redisOptions
*
options
,
const
char
*
channel
,
const
char
*
msg
)
{
redisContext
*
c
=
redisConnectWithOptions
(
options
);
assert
(
c
!=
NULL
);
redisReply
*
reply
=
redisCommand
(
c
,
"PUBLISH %s %s"
,
channel
,
msg
);
assert
(
reply
->
type
==
REDIS_REPLY_INTEGER
&&
reply
->
integer
==
1
);
freeReplyObject
(
reply
);
disconnect
(
c
,
0
);
}
/* Expect a reply of type INTEGER */
void
integer_cb
(
redisAsyncContext
*
ac
,
void
*
r
,
void
*
privdata
)
{
redisReply
*
reply
=
r
;
TestState
*
state
=
privdata
;
assert
(
reply
!=
NULL
&&
reply
->
type
==
REDIS_REPLY_INTEGER
);
state
->
checkpoint
++
;
if
(
state
->
disconnect
)
async_disconnect
(
ac
);
}
/* Subscribe callback for test_pubsub_handling and test_pubsub_handling_resp3:
* - a published message triggers an unsubscribe
* - a command is sent before the unsubscribe response is received. */
void
subscribe_cb
(
redisAsyncContext
*
ac
,
void
*
r
,
void
*
privdata
)
{
redisReply
*
reply
=
r
;
TestState
*
state
=
privdata
;
assert
(
reply
!=
NULL
&&
reply
->
type
==
(
state
->
resp3
?
REDIS_REPLY_PUSH
:
REDIS_REPLY_ARRAY
)
&&
reply
->
elements
==
3
);
if
(
strcmp
(
reply
->
element
[
0
]
->
str
,
"subscribe"
)
==
0
)
{
assert
(
strcmp
(
reply
->
element
[
1
]
->
str
,
"mychannel"
)
==
0
&&
reply
->
element
[
2
]
->
str
==
NULL
);
publish_msg
(
state
->
options
,
"mychannel"
,
"Hello!"
);
}
else
if
(
strcmp
(
reply
->
element
[
0
]
->
str
,
"message"
)
==
0
)
{
assert
(
strcmp
(
reply
->
element
[
1
]
->
str
,
"mychannel"
)
==
0
&&
strcmp
(
reply
->
element
[
2
]
->
str
,
"Hello!"
)
==
0
);
state
->
checkpoint
++
;
/* Unsubscribe after receiving the published message. Send unsubscribe
* which should call the callback registered during subscribe */
redisAsyncCommand
(
ac
,
unexpected_cb
,
(
void
*
)
"unsubscribe should call subscribe_cb()"
,
"unsubscribe"
);
/* Send a regular command after unsubscribing, then disconnect */
state
->
disconnect
=
1
;
redisAsyncCommand
(
ac
,
integer_cb
,
state
,
"LPUSH mylist foo"
);
}
else
if
(
strcmp
(
reply
->
element
[
0
]
->
str
,
"unsubscribe"
)
==
0
)
{
assert
(
strcmp
(
reply
->
element
[
1
]
->
str
,
"mychannel"
)
==
0
&&
reply
->
element
[
2
]
->
str
==
NULL
);
}
else
{
printf
(
"Unexpected pubsub command: %s
\n
"
,
reply
->
element
[
0
]
->
str
);
exit
(
1
);
}
}
/* Expect a reply of type ARRAY */
void
array_cb
(
redisAsyncContext
*
ac
,
void
*
r
,
void
*
privdata
)
{
redisReply
*
reply
=
r
;
TestState
*
state
=
privdata
;
assert
(
reply
!=
NULL
&&
reply
->
type
==
REDIS_REPLY_ARRAY
);
state
->
checkpoint
++
;
if
(
state
->
disconnect
)
async_disconnect
(
ac
);
}
/* Expect a NULL reply */
void
null_cb
(
redisAsyncContext
*
ac
,
void
*
r
,
void
*
privdata
)
{
(
void
)
ac
;
assert
(
r
==
NULL
);
TestState
*
state
=
privdata
;
state
->
checkpoint
++
;
}
static
void
test_pubsub_handling
(
struct
config
config
)
{
test
(
"Subscribe, handle published message and unsubscribe: "
);
/* Setup event dispatcher with a testcase timeout */
base
=
event_base_new
();
struct
event
*
timeout
=
evtimer_new
(
base
,
timeout_cb
,
NULL
);
assert
(
timeout
!=
NULL
);
evtimer_assign
(
timeout
,
base
,
timeout_cb
,
NULL
);
struct
timeval
timeout_tv
=
{.
tv_sec
=
10
};
evtimer_add
(
timeout
,
&
timeout_tv
);
/* Connect */
redisOptions
options
=
get_redis_tcp_options
(
config
);
redisAsyncContext
*
ac
=
redisAsyncConnectWithOptions
(
&
options
);
assert
(
ac
!=
NULL
&&
ac
->
err
==
0
);
redisLibeventAttach
(
ac
,
base
);
/* Start subscribe */
TestState
state
=
{.
options
=
&
options
};
redisAsyncCommand
(
ac
,
subscribe_cb
,
&
state
,
"subscribe mychannel"
);
/* Make sure non-subscribe commands are handled */
redisAsyncCommand
(
ac
,
array_cb
,
&
state
,
"PING"
);
/* Start event dispatching loop */
test_cond
(
event_base_dispatch
(
base
)
==
0
);
event_free
(
timeout
);
event_base_free
(
base
);
/* Verify test checkpoints */
assert
(
state
.
checkpoint
==
3
);
}
/* Unexpected push message, will trigger a failure */
void
unexpected_push_cb
(
redisAsyncContext
*
ac
,
void
*
r
)
{
(
void
)
ac
;
(
void
)
r
;
printf
(
"Unexpected call to the PUSH callback!
\n
"
);
exit
(
1
);
}
static
void
test_pubsub_handling_resp3
(
struct
config
config
)
{
test
(
"Subscribe, handle published message and unsubscribe using RESP3: "
);
/* Setup event dispatcher with a testcase timeout */
base
=
event_base_new
();
struct
event
*
timeout
=
evtimer_new
(
base
,
timeout_cb
,
NULL
);
assert
(
timeout
!=
NULL
);
evtimer_assign
(
timeout
,
base
,
timeout_cb
,
NULL
);
struct
timeval
timeout_tv
=
{.
tv_sec
=
10
};
evtimer_add
(
timeout
,
&
timeout_tv
);
/* Connect */
redisOptions
options
=
get_redis_tcp_options
(
config
);
redisAsyncContext
*
ac
=
redisAsyncConnectWithOptions
(
&
options
);
assert
(
ac
!=
NULL
&&
ac
->
err
==
0
);
redisLibeventAttach
(
ac
,
base
);
/* Not expecting any push messages in this test */
redisAsyncSetPushCallback
(
ac
,
unexpected_push_cb
);
/* Switch protocol */
redisAsyncCommand
(
ac
,
NULL
,
NULL
,
"HELLO 3"
);
/* Start subscribe */
TestState
state
=
{.
options
=
&
options
,
.
resp3
=
1
};
redisAsyncCommand
(
ac
,
subscribe_cb
,
&
state
,
"subscribe mychannel"
);
/* Make sure non-subscribe commands are handled in RESP3 */
redisAsyncCommand
(
ac
,
integer_cb
,
&
state
,
"LPUSH mylist foo"
);
redisAsyncCommand
(
ac
,
integer_cb
,
&
state
,
"LPUSH mylist foo"
);
redisAsyncCommand
(
ac
,
integer_cb
,
&
state
,
"LPUSH mylist foo"
);
/* Handle an array with 3 elements as a non-subscribe command */
redisAsyncCommand
(
ac
,
array_cb
,
&
state
,
"LRANGE mylist 0 2"
);
/* Start event dispatching loop */
test_cond
(
event_base_dispatch
(
base
)
==
0
);
event_free
(
timeout
);
event_base_free
(
base
);
/* Verify test checkpoints */
assert
(
state
.
checkpoint
==
6
);
}
/* Subscribe callback for test_command_timeout_during_pubsub:
* - a subscribe response triggers a published message
* - the published message triggers a command that times out
* - the command timeout triggers a disconnect */
void
subscribe_with_timeout_cb
(
redisAsyncContext
*
ac
,
void
*
r
,
void
*
privdata
)
{
redisReply
*
reply
=
r
;
TestState
*
state
=
privdata
;
/* The non-clean disconnect should trigger the
* subscription callback with a NULL reply. */
if
(
reply
==
NULL
)
{
state
->
checkpoint
++
;
event_base_loopbreak
(
base
);
return
;
}
assert
(
reply
->
type
==
(
state
->
resp3
?
REDIS_REPLY_PUSH
:
REDIS_REPLY_ARRAY
)
&&
reply
->
elements
==
3
);
if
(
strcmp
(
reply
->
element
[
0
]
->
str
,
"subscribe"
)
==
0
)
{
assert
(
strcmp
(
reply
->
element
[
1
]
->
str
,
"mychannel"
)
==
0
&&
reply
->
element
[
2
]
->
str
==
NULL
);
publish_msg
(
state
->
options
,
"mychannel"
,
"Hello!"
);
state
->
checkpoint
++
;
}
else
if
(
strcmp
(
reply
->
element
[
0
]
->
str
,
"message"
)
==
0
)
{
assert
(
strcmp
(
reply
->
element
[
1
]
->
str
,
"mychannel"
)
==
0
&&
strcmp
(
reply
->
element
[
2
]
->
str
,
"Hello!"
)
==
0
);
state
->
checkpoint
++
;
/* Send a command that will trigger a timeout */
redisAsyncCommand
(
ac
,
null_cb
,
state
,
"DEBUG SLEEP 3"
);
redisAsyncCommand
(
ac
,
null_cb
,
state
,
"LPUSH mylist foo"
);
}
else
{
printf
(
"Unexpected pubsub command: %s
\n
"
,
reply
->
element
[
0
]
->
str
);
exit
(
1
);
}
}
static
void
test_command_timeout_during_pubsub
(
struct
config
config
)
{
test
(
"Command timeout during Pub/Sub: "
);
/* Setup event dispatcher with a testcase timeout */
base
=
event_base_new
();
struct
event
*
timeout
=
evtimer_new
(
base
,
timeout_cb
,
NULL
);
assert
(
timeout
!=
NULL
);
evtimer_assign
(
timeout
,
base
,
timeout_cb
,
NULL
);
struct
timeval
timeout_tv
=
{.
tv_sec
=
10
};
evtimer_add
(
timeout
,
&
timeout_tv
);
/* Connect */
redisOptions
options
=
get_redis_tcp_options
(
config
);
redisAsyncContext
*
ac
=
redisAsyncConnectWithOptions
(
&
options
);
assert
(
ac
!=
NULL
&&
ac
->
err
==
0
);
redisLibeventAttach
(
ac
,
base
);
/* Configure a command timout */
struct
timeval
command_timeout
=
{.
tv_sec
=
2
};
redisAsyncSetTimeout
(
ac
,
command_timeout
);
/* Not expecting any push messages in this test */
redisAsyncSetPushCallback
(
ac
,
unexpected_push_cb
);
/* Switch protocol */
redisAsyncCommand
(
ac
,
NULL
,
NULL
,
"HELLO 3"
);
/* Start subscribe */
TestState
state
=
{.
options
=
&
options
,
.
resp3
=
1
};
redisAsyncCommand
(
ac
,
subscribe_with_timeout_cb
,
&
state
,
"subscribe mychannel"
);
/* Start event dispatching loop */
assert
(
event_base_dispatch
(
base
)
==
0
);
event_free
(
timeout
);
event_base_free
(
base
);
/* Verify test checkpoints */
test_cond
(
state
.
checkpoint
==
5
);
}
/* Subscribe callback for test_pubsub_multiple_channels */
void
subscribe_channel_a_cb
(
redisAsyncContext
*
ac
,
void
*
r
,
void
*
privdata
)
{
redisReply
*
reply
=
r
;
TestState
*
state
=
privdata
;
assert
(
reply
!=
NULL
&&
reply
->
type
==
REDIS_REPLY_ARRAY
&&
reply
->
elements
==
3
);
if
(
strcmp
(
reply
->
element
[
0
]
->
str
,
"subscribe"
)
==
0
)
{
assert
(
strcmp
(
reply
->
element
[
1
]
->
str
,
"A"
)
==
0
);
publish_msg
(
state
->
options
,
"A"
,
"Hello!"
);
state
->
checkpoint
++
;
}
else
if
(
strcmp
(
reply
->
element
[
0
]
->
str
,
"message"
)
==
0
)
{
assert
(
strcmp
(
reply
->
element
[
1
]
->
str
,
"A"
)
==
0
&&
strcmp
(
reply
->
element
[
2
]
->
str
,
"Hello!"
)
==
0
);
state
->
checkpoint
++
;
/* Unsubscribe to channels, including a channel X which we don't subscribe to */
redisAsyncCommand
(
ac
,
unexpected_cb
,
(
void
*
)
"unsubscribe should not call unexpected_cb()"
,
"unsubscribe B X A"
);
/* Send a regular command after unsubscribing, then disconnect */
state
->
disconnect
=
1
;
redisAsyncCommand
(
ac
,
integer_cb
,
state
,
"LPUSH mylist foo"
);
}
else
if
(
strcmp
(
reply
->
element
[
0
]
->
str
,
"unsubscribe"
)
==
0
)
{
assert
(
strcmp
(
reply
->
element
[
1
]
->
str
,
"A"
)
==
0
);
state
->
checkpoint
++
;
}
else
{
printf
(
"Unexpected pubsub command: %s
\n
"
,
reply
->
element
[
0
]
->
str
);
exit
(
1
);
}
}
/* Subscribe callback for test_pubsub_multiple_channels */
void
subscribe_channel_b_cb
(
redisAsyncContext
*
ac
,
void
*
r
,
void
*
privdata
)
{
redisReply
*
reply
=
r
;
TestState
*
state
=
privdata
;
assert
(
reply
!=
NULL
&&
reply
->
type
==
REDIS_REPLY_ARRAY
&&
reply
->
elements
==
3
);
if
(
strcmp
(
reply
->
element
[
0
]
->
str
,
"subscribe"
)
==
0
)
{
assert
(
strcmp
(
reply
->
element
[
1
]
->
str
,
"B"
)
==
0
);
state
->
checkpoint
++
;
}
else
if
(
strcmp
(
reply
->
element
[
0
]
->
str
,
"unsubscribe"
)
==
0
)
{
assert
(
strcmp
(
reply
->
element
[
1
]
->
str
,
"B"
)
==
0
);
state
->
checkpoint
++
;
}
else
{
printf
(
"Unexpected pubsub command: %s
\n
"
,
reply
->
element
[
0
]
->
str
);
exit
(
1
);
}
}
/* Test handling of multiple channels
* - subscribe to channel A and B
* - a published message on A triggers an unsubscribe of channel B, X and A
* where channel X is not subscribed to.
* - a command sent after unsubscribe triggers a disconnect */
static
void
test_pubsub_multiple_channels
(
struct
config
config
)
{
test
(
"Subscribe to multiple channels: "
);
/* Setup event dispatcher with a testcase timeout */
base
=
event_base_new
();
struct
event
*
timeout
=
evtimer_new
(
base
,
timeout_cb
,
NULL
);
assert
(
timeout
!=
NULL
);
evtimer_assign
(
timeout
,
base
,
timeout_cb
,
NULL
);
struct
timeval
timeout_tv
=
{.
tv_sec
=
10
};
evtimer_add
(
timeout
,
&
timeout_tv
);
/* Connect */
redisOptions
options
=
get_redis_tcp_options
(
config
);
redisAsyncContext
*
ac
=
redisAsyncConnectWithOptions
(
&
options
);
assert
(
ac
!=
NULL
&&
ac
->
err
==
0
);
redisLibeventAttach
(
ac
,
base
);
/* Not expecting any push messages in this test */
redisAsyncSetPushCallback
(
ac
,
unexpected_push_cb
);
/* Start subscribing to two channels */
TestState
state
=
{.
options
=
&
options
};
redisAsyncCommand
(
ac
,
subscribe_channel_a_cb
,
&
state
,
"subscribe A"
);
redisAsyncCommand
(
ac
,
subscribe_channel_b_cb
,
&
state
,
"subscribe B"
);
/* Start event dispatching loop */
assert
(
event_base_dispatch
(
base
)
==
0
);
event_free
(
timeout
);
event_base_free
(
base
);
/* Verify test checkpoints */
test_cond
(
state
.
checkpoint
==
6
);
}
/* Command callback for test_monitor() */
void
monitor_cb
(
redisAsyncContext
*
ac
,
void
*
r
,
void
*
privdata
)
{
redisReply
*
reply
=
r
;
TestState
*
state
=
privdata
;
/* NULL reply is received when BYE triggers a disconnect. */
if
(
reply
==
NULL
)
{
event_base_loopbreak
(
base
);
return
;
}
assert
(
reply
!=
NULL
&&
reply
->
type
==
REDIS_REPLY_STATUS
);
state
->
checkpoint
++
;
if
(
state
->
checkpoint
==
1
)
{
/* Response from MONITOR */
redisContext
*
c
=
redisConnectWithOptions
(
state
->
options
);
assert
(
c
!=
NULL
);
redisReply
*
reply
=
redisCommand
(
c
,
"SET first 1"
);
assert
(
reply
->
type
==
REDIS_REPLY_STATUS
);
freeReplyObject
(
reply
);
redisFree
(
c
);
}
else
if
(
state
->
checkpoint
==
2
)
{
/* Response for monitored command 'SET first 1' */
assert
(
strstr
(
reply
->
str
,
"first"
)
!=
NULL
);
redisContext
*
c
=
redisConnectWithOptions
(
state
->
options
);
assert
(
c
!=
NULL
);
redisReply
*
reply
=
redisCommand
(
c
,
"SET second 2"
);
assert
(
reply
->
type
==
REDIS_REPLY_STATUS
);
freeReplyObject
(
reply
);
redisFree
(
c
);
}
else
if
(
state
->
checkpoint
==
3
)
{
/* Response for monitored command 'SET second 2' */
assert
(
strstr
(
reply
->
str
,
"second"
)
!=
NULL
);
/* Send QUIT to disconnect */
redisAsyncCommand
(
ac
,
NULL
,
NULL
,
"QUIT"
);
}
}
/* Test handling of the monitor command
* - sends MONITOR to enable monitoring.
* - sends SET commands via separate clients to be monitored.
* - sends QUIT to stop monitoring and disconnect. */
static
void
test_monitor
(
struct
config
config
)
{
test
(
"Enable monitoring: "
);
/* Setup event dispatcher with a testcase timeout */
base
=
event_base_new
();
struct
event
*
timeout
=
evtimer_new
(
base
,
timeout_cb
,
NULL
);
assert
(
timeout
!=
NULL
);
evtimer_assign
(
timeout
,
base
,
timeout_cb
,
NULL
);
struct
timeval
timeout_tv
=
{.
tv_sec
=
10
};
evtimer_add
(
timeout
,
&
timeout_tv
);
/* Connect */
redisOptions
options
=
get_redis_tcp_options
(
config
);
redisAsyncContext
*
ac
=
redisAsyncConnectWithOptions
(
&
options
);
assert
(
ac
!=
NULL
&&
ac
->
err
==
0
);
redisLibeventAttach
(
ac
,
base
);
/* Not expecting any push messages in this test */
redisAsyncSetPushCallback
(
ac
,
unexpected_push_cb
);
/* Start monitor */
TestState
state
=
{.
options
=
&
options
};
redisAsyncCommand
(
ac
,
monitor_cb
,
&
state
,
"monitor"
);
/* Start event dispatching loop */
test_cond
(
event_base_dispatch
(
base
)
==
0
);
event_free
(
timeout
);
event_base_free
(
base
);
/* Verify test checkpoints */
assert
(
state
.
checkpoint
==
3
);
}
#endif
/* HIREDIS_TEST_ASYNC */
int
main
(
int
argc
,
char
**
argv
)
{
struct
config
cfg
=
{
.
tcp
=
{
...
...
@@ -1401,6 +1999,24 @@ int main(int argc, char **argv) {
}
#endif
#ifdef HIREDIS_TEST_ASYNC
printf
(
"
\n
Testing asynchronous API against TCP connection (%s:%d):
\n
"
,
cfg
.
tcp
.
host
,
cfg
.
tcp
.
port
);
cfg
.
type
=
CONN_TCP
;
int
major
;
redisContext
*
c
=
do_connect
(
cfg
);
get_redis_version
(
c
,
&
major
,
NULL
);
disconnect
(
c
,
0
);
test_pubsub_handling
(
cfg
);
test_pubsub_multiple_channels
(
cfg
);
test_monitor
(
cfg
);
if
(
major
>=
6
)
{
test_pubsub_handling_resp3
(
cfg
);
test_command_timeout_during_pubsub
(
cfg
);
}
#endif
/* HIREDIS_TEST_ASYNC */
if
(
test_inherit_fd
)
{
printf
(
"
\n
Testing against inherited fd (%s): "
,
cfg
.
unix_sock
.
path
);
if
(
test_unix_socket
)
{
...
...
redis.conf
View file @
d2b5a579
...
...
@@ -213,7 +213,9 @@ tcp-keepalive 300
#
# tls-client-key-file-pass secret
# Configure a DH parameters file to enable Diffie-Hellman (DH) key exchange:
# Configure a DH parameters file to enable Diffie-Hellman (DH) key exchange,
# required by older versions of OpenSSL (<3.0). Newer versions do not require
# this configuration and recommend against it.
#
# tls-dh-params-file redis.dh
...
...
@@ -641,7 +643,7 @@ repl-diskless-sync-max-replicas 0
# you risk an OOM kill.
repl
-
diskless
-
load
disabled
#
Replicas
send PINGs to
server
in a predefined interval. It's possible to
#
Master
send PINGs to
its replicas
in a predefined interval. It's possible to
# change this interval with the repl_ping_replica_period option. The default
# value is 10 seconds.
#
...
...
@@ -1678,7 +1680,7 @@ aof-timestamp-enabled no
# routing. By default this value is only shown as additional metadata in the CLUSTER SLOTS
# command, but can be changed using 'cluster-preferred-endpoint-type' config. This value is
# communicated along the clusterbus to all nodes, setting it to an empty string will remove
# the hostname and also propgate the removal.
# the hostname and also prop
a
gate the removal.
#
# cluster-announce-hostname ""
...
...
runtest-moduleapi
View file @
d2b5a579
...
...
@@ -44,6 +44,7 @@ $TCLSH tests/test_helper.tcl \
--single
unit/moduleapi/aclcheck
\
--single
unit/moduleapi/subcommands
\
--single
unit/moduleapi/reply
\
--single
unit/moduleapi/cmdintrospection
\
--single
unit/moduleapi/eventloop
\
--single
unit/moduleapi/timer
\
"
${
@
}
"
src/acl.c
View file @
d2b5a579
...
...
@@ -1531,18 +1531,15 @@ static int ACLSelectorCheckKey(aclSelector *selector, const char *key, int keyle
return
ACL_DENIED_KEY
;
}
/* Returns if a given command may possibly access channels. For this context,
* the unsubscribe commands do not have channels. */
static
int
ACLDoesCommandHaveChannels
(
struct
redisCommand
*
cmd
)
{
return
(
cmd
->
proc
==
publishCommand
||
cmd
->
proc
==
subscribeCommand
||
cmd
->
proc
==
psubscribeCommand
||
cmd
->
proc
==
spublishCommand
||
cmd
->
proc
==
ssubscribeCommand
);
}
/* Checks a channel against a provide list of channels. */
static
int
ACLCheckChannelAgainstList
(
list
*
reference
,
const
char
*
channel
,
int
channellen
,
int
literal
)
{
/* Checks a channel against a provided list of channels. The is_pattern
* argument should only be used when subscribing (not when publishing)
* and controls whether the input channel is evaluated as a channel pattern
* (like in PSUBSCRIBE) or a plain channel name (like in SUBSCRIBE).
*
* Note that a plain channel name like in PUBLISH or SUBSCRIBE can be
* matched against ACL channel patterns, but the pattern provided in PSUBSCRIBE
* can only be matched as a literal against an ACL pattern (using plain string compare). */
static
int
ACLCheckChannelAgainstList
(
list
*
reference
,
const
char
*
channel
,
int
channellen
,
int
is_pattern
)
{
listIter
li
;
listNode
*
ln
;
...
...
@@ -1550,8 +1547,10 @@ static int ACLCheckChannelAgainstList(list *reference, const char *channel, int
while
((
ln
=
listNext
(
&
li
)))
{
sds
pattern
=
listNodeValue
(
ln
);
size_t
plen
=
sdslen
(
pattern
);
if
((
literal
&&
!
strcmp
(
pattern
,
channel
))
||
(
!
literal
&&
stringmatchlen
(
pattern
,
plen
,
channel
,
channellen
,
0
)))
/* Channel patterns are matched literally against the channels in
* the list. Regular channels perform pattern matching. */
if
((
is_pattern
&&
!
strcmp
(
pattern
,
channel
))
||
(
!
is_pattern
&&
stringmatchlen
(
pattern
,
plen
,
channel
,
channellen
,
0
)))
{
return
ACL_OK
;
}
...
...
@@ -1559,28 +1558,6 @@ static int ACLCheckChannelAgainstList(list *reference, const char *channel, int
return
ACL_DENIED_CHANNEL
;
}
/* Check if the pub/sub channels of the command can be executed
* according to the ACL channels associated with the specified selector.
*
* idx and count are the index and count of channel arguments from the
* command. The literal argument controls whether the selector's ACL channels are
* evaluated as literal values or matched as glob-like patterns.
*
* If the selector can execute the command ACL_OK is returned, otherwise
* ACL_DENIED_CHANNEL. */
static
int
ACLSelectorCheckPubsubArguments
(
aclSelector
*
s
,
robj
**
argv
,
int
idx
,
int
count
,
int
literal
,
int
*
idxptr
)
{
for
(
int
j
=
idx
;
j
<
idx
+
count
;
j
++
)
{
if
(
ACLCheckChannelAgainstList
(
s
->
channels
,
argv
[
j
]
->
ptr
,
sdslen
(
argv
[
j
]
->
ptr
),
literal
!=
ACL_OK
))
{
if
(
idxptr
)
*
idxptr
=
j
;
return
ACL_DENIED_CHANNEL
;
}
}
/* If we survived all the above checks, the selector can execute the
* command. */
return
ACL_OK
;
}
/* To prevent duplicate calls to getKeysResult, a cache is maintained
* in between calls to the various selectors. */
typedef
struct
{
...
...
@@ -1645,7 +1622,7 @@ static int ACLSelectorCheckCmd(aclSelector *selector, struct redisCommand *cmd,
int
idx
=
resultidx
[
j
].
pos
;
ret
=
ACLSelectorCheckKey
(
selector
,
argv
[
idx
]
->
ptr
,
sdslen
(
argv
[
idx
]
->
ptr
),
resultidx
[
j
].
flags
);
if
(
ret
!=
ACL_OK
)
{
if
(
resultidx
)
*
keyidxptr
=
resultidx
[
j
].
pos
;
if
(
keyidxptr
)
*
keyidxptr
=
resultidx
[
j
].
pos
;
return
ret
;
}
}
...
...
@@ -1653,26 +1630,30 @@ static int ACLSelectorCheckCmd(aclSelector *selector, struct redisCommand *cmd,
/* Check if the user can execute commands explicitly touching the channels
* mentioned in the command arguments */
if
(
!
(
selector
->
flags
&
SELECTOR_FLAG_ALLCHANNELS
)
&&
ACLDoesCommandHaveChannels
(
cmd
))
{
if
(
cmd
->
proc
==
publishCommand
||
cmd
->
proc
==
spublishCommand
)
{
ret
=
ACLSelectorCheckPubsubArguments
(
selector
,
argv
,
1
,
1
,
0
,
keyidxptr
);
}
else
if
(
cmd
->
proc
==
subscribeCommand
||
cmd
->
proc
==
ssubscribeCommand
)
{
ret
=
ACLSelectorCheckPubsubArguments
(
selector
,
argv
,
1
,
argc
-
1
,
0
,
keyidxptr
);
}
else
if
(
cmd
->
proc
==
psubscribeCommand
)
{
ret
=
ACLSelectorCheckPubsubArguments
(
selector
,
argv
,
1
,
argc
-
1
,
1
,
keyidxptr
);
}
else
{
serverPanic
(
"Encountered a command declared with channels but not handled"
);
}
if
(
ret
!=
ACL_OK
)
{
/* keyidxptr is set by ACLSelectorCheckPubsubArguments */
return
ret
;
const
int
channel_flags
=
CMD_CHANNEL_PUBLISH
|
CMD_CHANNEL_SUBSCRIBE
;
if
(
!
(
selector
->
flags
&
SELECTOR_FLAG_ALLCHANNELS
)
&&
doesCommandHaveChannelsWithFlags
(
cmd
,
channel_flags
))
{
getKeysResult
channels
=
(
getKeysResult
)
GETKEYS_RESULT_INIT
;
getChannelsFromCommand
(
cmd
,
argv
,
argc
,
&
channels
);
keyReference
*
channelref
=
channels
.
keys
;
for
(
int
j
=
0
;
j
<
channels
.
numkeys
;
j
++
)
{
int
idx
=
channelref
[
j
].
pos
;
if
(
!
(
channelref
[
j
].
flags
&
channel_flags
))
continue
;
int
is_pattern
=
channelref
[
j
].
flags
&
CMD_CHANNEL_PATTERN
;
int
ret
=
ACLCheckChannelAgainstList
(
selector
->
channels
,
argv
[
idx
]
->
ptr
,
sdslen
(
argv
[
idx
]
->
ptr
),
is_pattern
);
if
(
ret
!=
ACL_OK
)
{
if
(
keyidxptr
)
*
keyidxptr
=
channelref
[
j
].
pos
;
getKeysFreeResult
(
&
channels
);
return
ret
;
}
}
getKeysFreeResult
(
&
channels
);
}
return
ACL_OK
;
}
/* Check if the key can be accessed by the client according to
* the ACLs associated with the specified user.
* the ACLs associated with the specified user according to the
* keyspec access flags.
*
* If the user can access the key, ACL_OK is returned, otherwise
* ACL_DENIED_KEY is returned. */
...
...
@@ -1699,7 +1680,7 @@ int ACLUserCheckKeyPerm(user *u, const char *key, int keylen, int flags) {
*
* If the user can access the key, ACL_OK is returned, otherwise
* ACL_DENIED_CHANNEL is returned. */
int
ACLUserCheckChannelPerm
(
user
*
u
,
sds
channel
,
int
l
iter
al
)
{
int
ACLUserCheckChannelPerm
(
user
*
u
,
sds
channel
,
int
i
s_pat
ter
n
)
{
listIter
li
;
listNode
*
ln
;
...
...
@@ -1714,7 +1695,7 @@ int ACLUserCheckChannelPerm(user *u, sds channel, int literal) {
if
(
s
->
flags
&
SELECTOR_FLAG_ALLCHANNELS
)
return
ACL_OK
;
/* Otherwise, loop over the selectors list and check each channel */
if
(
ACLCheckChannelAgainstList
(
s
->
channels
,
channel
,
sdslen
(
channel
),
l
iter
al
)
==
ACL_OK
)
{
if
(
ACLCheckChannelAgainstList
(
s
->
channels
,
channel
,
sdslen
(
channel
),
i
s_pat
ter
n
)
==
ACL_OK
)
{
return
ACL_OK
;
}
}
...
...
src/aof.c
View file @
d2b5a579
...
...
@@ -42,11 +42,13 @@
#include <sys/param.h>
void
freeClientArgv
(
client
*
c
);
off_t
getAppendOnlyFileSize
(
sds
filename
);
off_t
getBaseAndIncrAppendOnlyFilesSize
(
aofManifest
*
am
);
off_t
getAppendOnlyFileSize
(
sds
filename
,
int
*
status
);
off_t
getBaseAndIncrAppendOnlyFilesSize
(
aofManifest
*
am
,
int
*
status
);
int
getBaseAndIncrAppendOnlyFilesNum
(
aofManifest
*
am
);
int
aofFileExist
(
char
*
filename
);
int
rewriteAppendOnlyFile
(
char
*
filename
);
aofManifest
*
aofLoadManifestFromFile
(
sds
am_filepath
);
void
aofManifestFreeAndUpdate
(
aofManifest
*
am
);
/* ----------------------------------------------------------------------------
* AOF Manifest file implementation.
...
...
@@ -226,13 +228,8 @@ sds getAofManifestAsString(aofManifest *am) {
* in order to support seamless upgrades from previous versions which did not
* use them.
*/
#define MANIFEST_MAX_LINE 1024
void
aofLoadManifestFromDisk
(
void
)
{
const
char
*
err
=
NULL
;
long
long
maxseq
=
0
;
server
.
aof_manifest
=
aofManifestCreate
();
if
(
!
dirExists
(
server
.
aof_dirname
))
{
serverLog
(
LL_NOTICE
,
"The AOF directory %s doesn't exist"
,
server
.
aof_dirname
);
return
;
...
...
@@ -247,16 +244,26 @@ void aofLoadManifestFromDisk(void) {
return
;
}
aofManifest
*
am
=
aofLoadManifestFromFile
(
am_filepath
);
if
(
am
)
aofManifestFreeAndUpdate
(
am
);
sdsfree
(
am_name
);
sdsfree
(
am_filepath
);
}
/* Generic manifest loading function, used in `aofLoadManifestFromDisk` and redis-check-aof tool. */
#define MANIFEST_MAX_LINE 1024
aofManifest
*
aofLoadManifestFromFile
(
sds
am_filepath
)
{
const
char
*
err
=
NULL
;
long
long
maxseq
=
0
;
aofManifest
*
am
=
aofManifestCreate
();
FILE
*
fp
=
fopen
(
am_filepath
,
"r"
);
if
(
fp
==
NULL
)
{
serverLog
(
LL_WARNING
,
"Fatal error: can't open the AOF manifest "
"file %s for reading: %s"
,
am_
name
,
strerror
(
errno
));
"file %s for reading: %s"
,
am_
filepath
,
strerror
(
errno
));
exit
(
1
);
}
sdsfree
(
am_name
);
sdsfree
(
am_filepath
);
char
buf
[
MANIFEST_MAX_LINE
+
1
];
sds
*
argv
=
NULL
;
int
argc
;
...
...
@@ -292,14 +299,14 @@ void aofLoadManifestFromDisk(void) {
line
=
sdstrim
(
sdsnew
(
buf
),
"
\t\r\n
"
);
if
(
!
sdslen
(
line
))
{
err
=
"
The
AOF manifest file
is invalid
format"
;
err
=
"
Invalid
AOF manifest file format"
;
goto
loaderr
;
}
argv
=
sdssplitargs
(
line
,
&
argc
);
/* 'argc < 6' was done for forward compatibility. */
if
(
argv
==
NULL
||
argc
<
6
||
(
argc
%
2
))
{
err
=
"
The
AOF manifest file
is invalid
format"
;
err
=
"
Invalid
AOF manifest file format"
;
goto
loaderr
;
}
...
...
@@ -321,7 +328,7 @@ void aofLoadManifestFromDisk(void) {
/* We have to make sure we load all the information. */
if
(
!
ai
->
file_name
||
!
ai
->
file_seq
||
!
ai
->
file_type
)
{
err
=
"
The
AOF manifest file
is invalid
format"
;
err
=
"
Invalid
AOF manifest file format"
;
goto
loaderr
;
}
...
...
@@ -329,21 +336,21 @@ void aofLoadManifestFromDisk(void) {
argv
=
NULL
;
if
(
ai
->
file_type
==
AOF_FILE_TYPE_BASE
)
{
if
(
server
.
aof_manifest
->
base_aof_info
)
{
if
(
am
->
base_aof_info
)
{
err
=
"Found duplicate base file information"
;
goto
loaderr
;
}
server
.
aof_manifest
->
base_aof_info
=
ai
;
server
.
aof_manifest
->
curr_base_file_seq
=
ai
->
file_seq
;
am
->
base_aof_info
=
ai
;
am
->
curr_base_file_seq
=
ai
->
file_seq
;
}
else
if
(
ai
->
file_type
==
AOF_FILE_TYPE_HIST
)
{
listAddNodeTail
(
server
.
aof_manifest
->
history_aof_list
,
ai
);
listAddNodeTail
(
am
->
history_aof_list
,
ai
);
}
else
if
(
ai
->
file_type
==
AOF_FILE_TYPE_INCR
)
{
if
(
ai
->
file_seq
<=
maxseq
)
{
err
=
"Found a non-monotonic sequence number"
;
goto
loaderr
;
}
listAddNodeTail
(
server
.
aof_manifest
->
incr_aof_list
,
ai
);
server
.
aof_manifest
->
curr_incr_file_seq
=
ai
->
file_seq
;
listAddNodeTail
(
am
->
incr_aof_list
,
ai
);
am
->
curr_incr_file_seq
=
ai
->
file_seq
;
maxseq
=
ai
->
file_seq
;
}
else
{
err
=
"Unknown AOF file type"
;
...
...
@@ -356,7 +363,7 @@ void aofLoadManifestFromDisk(void) {
}
fclose
(
fp
);
return
;
return
am
;
loaderr:
/* Sanitizer suppression: may report a false positive if we goto loaderr
...
...
@@ -627,7 +634,7 @@ void aofUpgradePrepare(aofManifest *am) {
server
.
aof_dirname
,
strerror
(
errno
));
sdsfree
(
aof_filepath
);
exit
(
1
);
;
exit
(
1
);
}
sdsfree
(
aof_filepath
);
...
...
@@ -721,7 +728,7 @@ void aofOpenIfNeededOnServerStart(void) {
exit
(
1
);
}
server
.
aof_last_incr_size
=
getAppendOnlyFileSize
(
aof_name
);
server
.
aof_last_incr_size
=
getAppendOnlyFileSize
(
aof_name
,
NULL
);
}
int
aofFileExist
(
char
*
filename
)
{
...
...
@@ -1338,26 +1345,35 @@ int loadSingleAppendOnlyFile(char *filename) {
client
*
old_client
=
server
.
current_client
;
fakeClient
=
server
.
current_client
=
createAOFClient
();
/* Check if this AOF file has an RDB preamble. In that case we need to
* load the RDB file and later continue loading the AOF tail. */
/* Check if the AOF file is in RDB format (it may be RDB encoded base AOF
* or old style RDB-preamble AOF). In that case we need to load the RDB file
* and later continue loading the AOF tail if it is an old style RDB-preamble AOF. */
char
sig
[
5
];
/* "REDIS" */
if
(
fread
(
sig
,
1
,
5
,
fp
)
!=
5
||
memcmp
(
sig
,
"REDIS"
,
5
)
!=
0
)
{
/* No RDB
preamble
, seek back at 0 offset. */
/* No
t in
RDB
format
, seek back at 0 offset. */
if
(
fseek
(
fp
,
0
,
SEEK_SET
)
==
-
1
)
goto
readerr
;
}
else
{
/* RDB
preamble
. Pass loading the RDB functions. */
/* RDB
format
. Pass loading the RDB functions. */
rio
rdb
;
int
old_style
=
!
strcmp
(
filename
,
server
.
aof_filename
);
if
(
old_style
)
serverLog
(
LL_NOTICE
,
"Reading RDB preamble from AOF file..."
);
else
serverLog
(
LL_NOTICE
,
"Reading RDB base file on AOF loading..."
);
serverLog
(
LL_NOTICE
,
"Reading RDB preamble from AOF file..."
);
if
(
fseek
(
fp
,
0
,
SEEK_SET
)
==
-
1
)
goto
readerr
;
rioInitWithFile
(
&
rdb
,
fp
);
if
(
rdbLoadRio
(
&
rdb
,
RDBFLAGS_AOF_PREAMBLE
,
NULL
)
!=
C_OK
)
{
serverLog
(
LL_WARNING
,
"Error reading the RDB preamble of the AOF file %s, AOF loading aborted"
,
filename
);
if
(
old_style
)
serverLog
(
LL_WARNING
,
"Error reading the RDB preamble of the AOF file %s, AOF loading aborted"
,
filename
);
else
serverLog
(
LL_WARNING
,
"Error reading the RDB base file %s, AOF loading aborted"
,
filename
);
goto
readerr
;
}
else
{
loadingAbsProgress
(
ftello
(
fp
));
last_progress_report_size
=
ftello
(
fp
);
serverLog
(
LL_NOTICE
,
"Reading the remaining AOF tail..."
);
if
(
old_style
)
serverLog
(
LL_NOTICE
,
"Reading the remaining AOF tail..."
);
}
}
...
...
@@ -1517,15 +1533,15 @@ uxeof: /* Unexpected AOF end of file. */
}
}
}
serverLog
(
LL_WARNING
,
"Unexpected end of file reading the append only file %s. You can:
\
1) Make a backup of your AOF file, then use ./redis-check-aof --fix <filename
>. \
2) Alternatively you can set the 'aof-load-truncated' configuration option to yes and restart the server."
,
filename
);
serverLog
(
LL_WARNING
,
"Unexpected end of file reading the append only file %s. You can:
"
"
1) Make a backup of your AOF file, then use ./redis-check-aof --fix <filename
.manifest>. "
"
2) Alternatively you can set the 'aof-load-truncated' configuration option to yes and restart the server."
,
filename
);
ret
=
AOF_FAILED
;
goto
cleanup
;
fmterr:
/* Format error. */
serverLog
(
LL_WARNING
,
"Bad file format reading the append only file %s:
\
make a backup of your AOF file, then use ./redis-check-aof --fix <filename>"
,
filename
);
serverLog
(
LL_WARNING
,
"Bad file format reading the append only file %s:
"
"
make a backup of your AOF file, then use ./redis-check-aof --fix <filename
.manifest
>"
,
filename
);
ret
=
AOF_FAILED
;
/* fall through to cleanup. */
...
...
@@ -1540,7 +1556,7 @@ cleanup:
/* Load the AOF files according the aofManifest pointed by am. */
int
loadAppendOnlyFiles
(
aofManifest
*
am
)
{
serverAssert
(
am
!=
NULL
);
int
ret
=
C_OK
;
int
status
,
ret
=
C_OK
;
long
long
start
;
off_t
total_size
=
0
;
sds
aof_name
;
...
...
@@ -1574,7 +1590,16 @@ int loadAppendOnlyFiles(aofManifest *am) {
/* Here we calculate the total size of all BASE and INCR files in
* advance, it will be set to `server.loading_total_bytes`. */
total_size
=
getBaseAndIncrAppendOnlyFilesSize
(
am
);
total_size
=
getBaseAndIncrAppendOnlyFilesSize
(
am
,
&
status
);
if
(
status
!=
AOF_OK
)
{
/* If an AOF exists in the manifest but not on the disk, we consider this to be a fatal error. */
if
(
status
==
AOF_NOT_EXIST
)
status
=
AOF_FAILED
;
return
status
;
}
else
if
(
total_size
==
0
)
{
return
AOF_EMPTY
;
}
startLoading
(
total_size
,
RDBFLAGS_AOF_PREAMBLE
,
0
);
/* Load BASE AOF if needed. */
...
...
@@ -1590,9 +1615,8 @@ int loadAppendOnlyFiles(aofManifest *am) {
aof_name
,
(
float
)(
ustime
()
-
start
)
/
1000000
);
}
/* If an AOF exists in the manifest but not on the disk, Or the truncated
* file is not the last file, we consider this to be a fatal error. */
if
(
ret
==
AOF_NOT_EXIST
||
(
ret
==
AOF_TRUNCATED
&&
!
last_file
))
{
/* If the truncated file is not the last file, we consider this to be a fatal error. */
if
(
ret
==
AOF_TRUNCATED
&&
!
last_file
)
{
ret
=
AOF_FAILED
;
}
...
...
@@ -1620,7 +1644,11 @@ int loadAppendOnlyFiles(aofManifest *am) {
aof_name
,
(
float
)(
ustime
()
-
start
)
/
1000000
);
}
if
(
ret
==
AOF_NOT_EXIST
||
(
ret
==
AOF_TRUNCATED
&&
!
last_file
))
{
/* We know that (at least) one of the AOF files has data (total_size > 0),
* so empty incr AOF file doesn't count as a AOF_EMPTY result */
if
(
ret
==
AOF_EMPTY
)
ret
=
AOF_OK
;
if
(
ret
==
AOF_TRUNCATED
&&
!
last_file
)
{
ret
=
AOF_FAILED
;
}
...
...
@@ -1635,7 +1663,7 @@ int loadAppendOnlyFiles(aofManifest *am) {
server
.
aof_fsync_offset
=
server
.
aof_current_size
;
cleanup:
stopLoading
(
ret
==
AOF_OK
);
stopLoading
(
ret
==
AOF_OK
||
ret
==
AOF_TRUNCATED
);
return
ret
;
}
...
...
@@ -2007,10 +2035,14 @@ int rewriteStreamObject(rio *r, robj *key, robj *o) {
/* Append XSETID after XADD, make sure lastid is correct,
* in case of XDEL lastid. */
if
(
!
rioWriteBulkCount
(
r
,
'*'
,
3
)
||
if
(
!
rioWriteBulkCount
(
r
,
'*'
,
7
)
||
!
rioWriteBulkString
(
r
,
"XSETID"
,
6
)
||
!
rioWriteBulkObject
(
r
,
key
)
||
!
rioWriteBulkStreamID
(
r
,
&
s
->
last_id
))
!
rioWriteBulkStreamID
(
r
,
&
s
->
last_id
)
||
!
rioWriteBulkString
(
r
,
"ENTRIESADDED"
,
12
)
||
!
rioWriteBulkLongLong
(
r
,
s
->
entries_added
)
||
!
rioWriteBulkString
(
r
,
"MAXDELETEDID"
,
12
)
||
!
rioWriteBulkStreamID
(
r
,
&
s
->
max_deleted_entry_id
))
{
streamIteratorStop
(
&
si
);
return
0
;
...
...
@@ -2025,12 +2057,14 @@ int rewriteStreamObject(rio *r, robj *key, robj *o) {
while
(
raxNext
(
&
ri
))
{
streamCG
*
group
=
ri
.
data
;
/* Emit the XGROUP CREATE in order to create the group. */
if
(
!
rioWriteBulkCount
(
r
,
'*'
,
5
)
||
if
(
!
rioWriteBulkCount
(
r
,
'*'
,
7
)
||
!
rioWriteBulkString
(
r
,
"XGROUP"
,
6
)
||
!
rioWriteBulkString
(
r
,
"CREATE"
,
6
)
||
!
rioWriteBulkObject
(
r
,
key
)
||
!
rioWriteBulkString
(
r
,(
char
*
)
ri
.
key
,
ri
.
key_len
)
||
!
rioWriteBulkStreamID
(
r
,
&
group
->
last_id
))
!
rioWriteBulkStreamID
(
r
,
&
group
->
last_id
)
||
!
rioWriteBulkString
(
r
,
"ENTRIESREAD"
,
11
)
||
!
rioWriteBulkLongLong
(
r
,
group
->
entries_read
))
{
raxStop
(
&
ri
);
streamIteratorStop
(
&
si
);
...
...
@@ -2332,7 +2366,7 @@ int rewriteAppendOnlyFileBackground(void) {
server
.
aof_selected_db
=
-
1
;
flushAppendOnlyFile
(
1
);
if
(
openNewIncrAofForAppend
()
!=
C_OK
)
return
C_ERR
;
server
.
stat_aof_rewrites
++
;
if
((
childpid
=
redisFork
(
CHILD_TYPE_AOF
))
==
0
)
{
char
tmpfile
[
256
];
...
...
@@ -2388,7 +2422,10 @@ void aofRemoveTempFile(pid_t childpid) {
bg_unlink
(
tmpfile
);
}
off_t
getAppendOnlyFileSize
(
sds
filename
)
{
/* Get size of an AOF file.
* The status argument is an optional output argument to be filled with
* one of the AOF_ status values. */
off_t
getAppendOnlyFileSize
(
sds
filename
,
int
*
status
)
{
struct
redis_stat
sb
;
off_t
size
;
mstime_t
latency
;
...
...
@@ -2396,10 +2433,12 @@ off_t getAppendOnlyFileSize(sds filename) {
sds
aof_filepath
=
makePath
(
server
.
aof_dirname
,
filename
);
latencyStartMonitor
(
latency
);
if
(
redis_stat
(
aof_filepath
,
&
sb
)
==
-
1
)
{
if
(
status
)
*
status
=
errno
==
ENOENT
?
AOF_NOT_EXIST
:
AOF_OPEN_ERR
;
serverLog
(
LL_WARNING
,
"Unable to obtain the AOF file %s length. stat: %s"
,
filename
,
strerror
(
errno
));
size
=
0
;
}
else
{
if
(
status
)
*
status
=
AOF_OK
;
size
=
sb
.
st_size
;
}
latencyEndMonitor
(
latency
);
...
...
@@ -2408,22 +2447,27 @@ off_t getAppendOnlyFileSize(sds filename) {
return
size
;
}
off_t
getBaseAndIncrAppendOnlyFilesSize
(
aofManifest
*
am
)
{
/* Get size of all AOF files referred by the manifest (excluding history).
* The status argument is an output argument to be filled with
* one of the AOF_ status values. */
off_t
getBaseAndIncrAppendOnlyFilesSize
(
aofManifest
*
am
,
int
*
status
)
{
off_t
size
=
0
;
listNode
*
ln
;
listIter
li
;
if
(
am
->
base_aof_info
)
{
serverAssert
(
am
->
base_aof_info
->
file_type
==
AOF_FILE_TYPE_BASE
);
size
+=
getAppendOnlyFileSize
(
am
->
base_aof_info
->
file_name
);
size
+=
getAppendOnlyFileSize
(
am
->
base_aof_info
->
file_name
,
status
);
if
(
*
status
!=
AOF_OK
)
return
0
;
}
listRewind
(
am
->
incr_aof_list
,
&
li
);
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
{
aofInfo
*
ai
=
(
aofInfo
*
)
ln
->
value
;
serverAssert
(
ai
->
file_type
==
AOF_FILE_TYPE_INCR
);
size
+=
getAppendOnlyFileSize
(
ai
->
file_name
);
size
+=
getAppendOnlyFileSize
(
ai
->
file_name
,
status
);
if
(
*
status
!=
AOF_OK
)
return
0
;
}
return
size
;
...
...
@@ -2497,7 +2541,7 @@ void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
if
(
server
.
aof_fd
!=
-
1
)
{
/* AOF enabled. */
server
.
aof_selected_db
=
-
1
;
/* Make sure SELECT is re-issued */
server
.
aof_current_size
=
getAppendOnlyFileSize
(
new_base_filename
)
+
server
.
aof_last_incr_size
;
server
.
aof_current_size
=
getAppendOnlyFileSize
(
new_base_filename
,
NULL
)
+
server
.
aof_last_incr_size
;
server
.
aof_rewrite_base_size
=
server
.
aof_current_size
;
server
.
aof_fsync_offset
=
server
.
aof_current_size
;
server
.
aof_last_fsync
=
server
.
unixtime
;
...
...
src/blocked.c
View file @
d2b5a579
...
...
@@ -108,9 +108,11 @@ void blockClient(client *c, int btype) {
/* This function is called after a client has finished a blocking operation
* in order to update the total command duration, log the command into
* the Slow log if needed, and log the reply duration event if needed. */
void
updateStatsOnUnblock
(
client
*
c
,
long
blocked_us
,
long
reply_us
){
void
updateStatsOnUnblock
(
client
*
c
,
long
blocked_us
,
long
reply_us
,
int
had_errors
){
const
ustime_t
total_cmd_duration
=
c
->
duration
+
blocked_us
+
reply_us
;
c
->
lastcmd
->
microseconds
+=
total_cmd_duration
;
if
(
had_errors
)
c
->
lastcmd
->
failed_calls
++
;
if
(
server
.
latency_tracking_enabled
)
updateCommandLatencyHistogram
(
&
(
c
->
lastcmd
->
latency_histogram
),
total_cmd_duration
*
1000
);
/* Log the command into the Slow log if needed. */
...
...
@@ -314,6 +316,7 @@ void serveClientsBlockedOnListKey(robj *o, readyList *rl) {
* call. */
if
(
dstkey
)
incrRefCount
(
dstkey
);
long
long
prev_error_replies
=
server
.
stat_total_error_replies
;
client
*
old_client
=
server
.
current_client
;
server
.
current_client
=
receiver
;
monotime
replyTimer
;
...
...
@@ -322,7 +325,7 @@ void serveClientsBlockedOnListKey(robj *o, readyList *rl) {
rl
->
key
,
dstkey
,
rl
->
db
,
wherefrom
,
whereto
,
&
deleted
);
updateStatsOnUnblock
(
receiver
,
0
,
elapsedUs
(
replyTimer
));
updateStatsOnUnblock
(
receiver
,
0
,
elapsedUs
(
replyTimer
)
,
server
.
stat_total_error_replies
!=
prev_error_replies
);
unblockClient
(
receiver
);
afterCommand
(
receiver
);
server
.
current_client
=
old_client
;
...
...
@@ -366,6 +369,7 @@ void serveClientsBlockedOnSortedSetKey(robj *o, readyList *rl) {
?
1
:
0
;
int
reply_nil_when_empty
=
use_nested_array
;
long
long
prev_error_replies
=
server
.
stat_total_error_replies
;
client
*
old_client
=
server
.
current_client
;
server
.
current_client
=
receiver
;
monotime
replyTimer
;
...
...
@@ -388,7 +392,7 @@ void serveClientsBlockedOnSortedSetKey(robj *o, readyList *rl) {
decrRefCount
(
argv
[
1
]);
if
(
count
!=
-
1
)
decrRefCount
(
argv
[
2
]);
updateStatsOnUnblock
(
receiver
,
0
,
elapsedUs
(
replyTimer
));
updateStatsOnUnblock
(
receiver
,
0
,
elapsedUs
(
replyTimer
)
,
server
.
stat_total_error_replies
!=
prev_error_replies
);
unblockClient
(
receiver
);
afterCommand
(
receiver
);
server
.
current_client
=
old_client
;
...
...
@@ -421,6 +425,12 @@ void serveClientsBlockedOnStreamKey(robj *o, readyList *rl) {
bkinfo
*
bki
=
dictFetchValue
(
receiver
->
bpop
.
keys
,
rl
->
key
);
streamID
*
gt
=
&
bki
->
stream_id
;
long
long
prev_error_replies
=
server
.
stat_total_error_replies
;
client
*
old_client
=
server
.
current_client
;
server
.
current_client
=
receiver
;
monotime
replyTimer
;
elapsedStart
(
&
replyTimer
);
/* If we blocked in the context of a consumer
* group, we need to resolve the group and update the
* last ID the client is blocked for: this is needed
...
...
@@ -440,8 +450,7 @@ void serveClientsBlockedOnStreamKey(robj *o, readyList *rl) {
addReplyError
(
receiver
,
"-NOGROUP the consumer group this client "
"was blocked on no longer exists"
);
unblockClient
(
receiver
);
continue
;
goto
unblock_receiver
;
}
else
{
*
gt
=
group
->
last_id
;
}
...
...
@@ -470,10 +479,6 @@ void serveClientsBlockedOnStreamKey(robj *o, readyList *rl) {
}
}
client
*
old_client
=
server
.
current_client
;
server
.
current_client
=
receiver
;
monotime
replyTimer
;
elapsedStart
(
&
replyTimer
);
/* Emit the two elements sub-array consisting of
* the name of the stream and the data we
* extracted from it. Wrapped in a single-item
...
...
@@ -493,11 +498,13 @@ void serveClientsBlockedOnStreamKey(robj *o, readyList *rl) {
streamReplyWithRange
(
receiver
,
s
,
&
start
,
NULL
,
receiver
->
bpop
.
xread_count
,
0
,
group
,
consumer
,
noack
,
&
pi
);
updateStatsOnUnblock
(
receiver
,
0
,
elapsedUs
(
replyTimer
));
/* Note that after we unblock the client, 'gt'
* and other receiver->bpop stuff are no longer
* valid, so we must do the setup above before
* this call. */
* the unblockClient call. */
unblock_receiver:
updateStatsOnUnblock
(
receiver
,
0
,
elapsedUs
(
replyTimer
),
server
.
stat_total_error_replies
!=
prev_error_replies
);
unblockClient
(
receiver
);
afterCommand
(
receiver
);
server
.
current_client
=
old_client
;
...
...
@@ -545,12 +552,13 @@ void serveClientsBlockedOnKeyByModule(readyList *rl) {
* different modules with different triggers to consider if a key
* is ready or not. This means we can't exit the loop but need
* to continue after the first failure. */
long
long
prev_error_replies
=
server
.
stat_total_error_replies
;
client
*
old_client
=
server
.
current_client
;
server
.
current_client
=
receiver
;
monotime
replyTimer
;
elapsedStart
(
&
replyTimer
);
if
(
!
moduleTryServeClientBlockedOnKey
(
receiver
,
rl
->
key
))
continue
;
updateStatsOnUnblock
(
receiver
,
0
,
elapsedUs
(
replyTimer
));
updateStatsOnUnblock
(
receiver
,
0
,
elapsedUs
(
replyTimer
)
,
server
.
stat_total_error_replies
!=
prev_error_replies
);
moduleUnblockClient
(
receiver
);
afterCommand
(
receiver
);
server
.
current_client
=
old_client
;
...
...
src/call_reply.c
View file @
d2b5a579
...
...
@@ -60,7 +60,7 @@ struct CallReply {
double
d
;
/* Reply value for double reply. */
struct
CallReply
*
array
;
/* Array of sub-reply elements. used for set, array, map, and attribute */
}
val
;
list
*
deferred_error_list
;
/* list of errors in sds form or NULL */
struct
CallReply
*
attribute
;
/* attribute reply, NULL if not exists */
};
...
...
@@ -237,6 +237,8 @@ void freeCallReply(CallReply *rep) {
freeCallReplyInternal
(
rep
);
}
sdsfree
(
rep
->
original_proto
);
if
(
rep
->
deferred_error_list
)
listRelease
(
rep
->
deferred_error_list
);
zfree
(
rep
);
}
...
...
@@ -488,6 +490,11 @@ int callReplyIsResp3(CallReply *rep) {
return
rep
->
flags
&
REPLY_FLAG_RESP3
;
}
/* Returns a list of errors in sds form, or NULL. */
list
*
callReplyDeferredErrorList
(
CallReply
*
rep
)
{
return
rep
->
deferred_error_list
;
}
/* Create a new CallReply struct from the reply blob.
*
* The function will own the reply blob, so it must not be used or freed by
...
...
@@ -496,6 +503,9 @@ int callReplyIsResp3(CallReply *rep) {
* The reply blob will be freed when the returned CallReply struct is later
* freed using freeCallReply().
*
* The deferred_error_list is an optional list of errors that are present
* in the reply blob, if given, this function will take ownership on it.
*
* The private_data is optional and can later be accessed using
* callReplyGetPrivateData().
*
...
...
@@ -504,7 +514,7 @@ int callReplyIsResp3(CallReply *rep) {
* DESIGNED TO HANDLE USER INPUT and using it to parse invalid replies is
* unsafe.
*/
CallReply
*
callReplyCreate
(
sds
reply
,
void
*
private_data
)
{
CallReply
*
callReplyCreate
(
sds
reply
,
list
*
deferred_error_list
,
void
*
private_data
)
{
CallReply
*
res
=
zmalloc
(
sizeof
(
*
res
));
res
->
flags
=
REPLY_FLAG_ROOT
;
res
->
original_proto
=
reply
;
...
...
@@ -512,5 +522,6 @@ CallReply *callReplyCreate(sds reply, void *private_data) {
res
->
proto_len
=
sdslen
(
reply
);
res
->
private_data
=
private_data
;
res
->
attribute
=
NULL
;
res
->
deferred_error_list
=
deferred_error_list
;
return
res
;
}
src/call_reply.h
View file @
d2b5a579
...
...
@@ -34,7 +34,7 @@
typedef
struct
CallReply
CallReply
;
CallReply
*
callReplyCreate
(
sds
reply
,
void
*
private_data
);
CallReply
*
callReplyCreate
(
sds
reply
,
list
*
deferred_error_list
,
void
*
private_data
);
int
callReplyType
(
CallReply
*
rep
);
const
char
*
callReplyGetString
(
CallReply
*
rep
,
size_t
*
len
);
long
long
callReplyGetLongLong
(
CallReply
*
rep
);
...
...
@@ -51,6 +51,7 @@ const char *callReplyGetVerbatim(CallReply *rep, size_t *len, const char **forma
const
char
*
callReplyGetProto
(
CallReply
*
rep
,
size_t
*
len
);
void
*
callReplyGetPrivateData
(
CallReply
*
rep
);
int
callReplyIsResp3
(
CallReply
*
rep
);
list
*
callReplyDeferredErrorList
(
CallReply
*
rep
);
void
freeCallReply
(
CallReply
*
rep
);
#endif
/* SRC_CALL_REPLY_H_ */
src/cluster.c
View file @
d2b5a579
...
...
@@ -106,7 +106,7 @@ dictType clusterNodesDictType = {
};
/* Cluster re-addition blacklist. This maps node IDs to the time
* we can re-add this node. The goal is to avoid read
d
ing a removed
* we can re-add this node. The goal is to avoid reading a removed
* node for some time. */
dictType
clusterNodesBlackListDictType
=
{
dictSdsCaseHash
,
/* hash function */
...
...
@@ -243,10 +243,9 @@ int clusterLoadConfig(char *filename) {
if
(
hostname
)
{
*
hostname
=
'\0'
;
hostname
++
;
zfree
(
n
->
hostname
);
n
->
hostname
=
zstrdup
(
hostname
);
}
else
{
n
->
hostname
=
NULL
;
n
->
hostname
=
sdscpy
(
n
->
hostname
,
hostname
);
}
else
if
(
sdslen
(
n
->
hostname
)
!=
0
)
{
sdsclear
(
n
->
hostname
);
}
/* The plaintext port for client in a TLS cluster (n->pport) is not
...
...
@@ -570,20 +569,15 @@ void clusterUpdateMyselfIp(void) {
/* Update the hostname for the specified node with the provided C string. */
static
void
updateAnnouncedHostname
(
clusterNode
*
node
,
char
*
new
)
{
if
(
!
node
->
hostname
&&
!
new
)
{
return
;
}
/* Previous and new hostname are the same, no need to update. */
if
(
new
&&
node
->
hostname
&&
!
strcmp
(
new
,
node
->
hostname
))
{
if
(
new
&&
!
strcmp
(
new
,
node
->
hostname
))
{
return
;
}
if
(
node
->
hostname
)
zfree
(
node
->
hostname
);
if
(
new
)
{
node
->
hostname
=
zstrdup
(
new
);
}
else
{
node
->
hostname
=
NULL
;
node
->
hostname
=
sdscpy
(
node
->
hostname
,
new
);
}
else
if
(
sdslen
(
node
->
hostname
)
!=
0
)
{
sdsclear
(
node
->
hostname
)
;
}
}
...
...
@@ -959,7 +953,7 @@ clusterNode *createClusterNode(char *nodename, int flags) {
node
->
link
=
NULL
;
node
->
inbound_link
=
NULL
;
memset
(
node
->
ip
,
0
,
sizeof
(
node
->
ip
));
node
->
hostname
=
NULL
;
node
->
hostname
=
sdsempty
()
;
node
->
port
=
0
;
node
->
cport
=
0
;
node
->
pport
=
0
;
...
...
@@ -1125,7 +1119,7 @@ void freeClusterNode(clusterNode *n) {
nodename
=
sdsnewlen
(
n
->
name
,
CLUSTER_NAMELEN
);
serverAssert
(
dictDelete
(
server
.
cluster
->
nodes
,
nodename
)
==
DICT_OK
);
sdsfree
(
nodename
);
z
free
(
n
->
hostname
);
sds
free
(
n
->
hostname
);
/* Release links and associated data structures. */
if
(
n
->
link
)
freeClusterLink
(
n
->
link
);
...
...
@@ -1947,9 +1941,9 @@ static clusterMsgPingExt *getNextPingExt(clusterMsgPingExt *ext) {
* will be 8 byte padded. */
int
getHostnamePingExtSize
()
{
/* If hostname is not set, we don't send this extension */
if
(
!
myself
->
hostname
)
return
0
;
if
(
sdslen
(
myself
->
hostname
)
==
0
)
return
0
;
int
totlen
=
sizeof
(
clusterMsgPingExt
)
+
EIGHT_BYTE_ALIGN
(
s
tr
len
(
myself
->
hostname
)
+
1
);
int
totlen
=
sizeof
(
clusterMsgPingExt
)
+
EIGHT_BYTE_ALIGN
(
s
ds
len
(
myself
->
hostname
)
+
1
);
return
totlen
;
}
...
...
@@ -1958,19 +1952,18 @@ int getHostnamePingExtSize() {
* will return the amount of bytes written. */
int
writeHostnamePingExt
(
clusterMsgPingExt
**
cursor
)
{
/* If hostname is not set, we don't send this extension */
if
(
!
myself
->
hostname
)
return
0
;
if
(
sdslen
(
myself
->
hostname
)
==
0
)
return
0
;
/* Add the hostname information at the extension cursor */
clusterMsgPingExtHostname
*
ext
=
&
(
*
cursor
)
->
ext
[
0
].
hostname
;
size_t
hostname_len
=
strlen
(
myself
->
hostname
);
memcpy
(
ext
->
hostname
,
myself
->
hostname
,
hostname_len
);
memcpy
(
ext
->
hostname
,
myself
->
hostname
,
sdslen
(
myself
->
hostname
));
uint32_t
extension_size
=
getHostnamePingExtSize
();
/* Move the write cursor */
(
*
cursor
)
->
type
=
CLUSTERMSG_EXT_TYPE_HOSTNAME
;
(
*
cursor
)
->
length
=
htonl
(
extension_size
);
/* Make sure the string is NULL terminated by adding 1 */
*
cursor
=
(
clusterMsgPingExt
*
)
(
ext
->
hostname
+
EIGHT_BYTE_ALIGN
(
s
tr
len
(
myself
->
hostname
)
+
1
));
*
cursor
=
(
clusterMsgPingExt
*
)
(
ext
->
hostname
+
EIGHT_BYTE_ALIGN
(
s
ds
len
(
myself
->
hostname
)
+
1
));
return
extension_size
;
}
...
...
@@ -2975,7 +2968,7 @@ void clusterSendPing(clusterLink *link, int type) {
/* Set the initial extension position */
clusterMsgPingExt
*
cursor
=
getInitialPingExt
(
hdr
,
gossipcount
);
/* Add in the extensions */
if
(
myself
->
hostname
)
{
if
(
sdslen
(
myself
->
hostname
)
!=
0
)
{
hdr
->
mflags
[
0
]
|=
CLUSTERMSG_FLAG0_EXT_DATA
;
totlen
+=
writeHostnamePingExt
(
&
cursor
);
extensions
++
;
...
...
@@ -3959,7 +3952,8 @@ void clusterCron(void) {
iteration
++
;
/* Number of times this function was called so far. */
updateAnnouncedHostname
(
myself
,
server
.
cluster_announce_hostname
);
clusterUpdateMyselfHostname
();
/* The handshake timeout is the time after which a handshake node that was
* not turned into a normal node is removed from the nodes. Usually it is
* just the NODE_TIMEOUT value, but when NODE_TIMEOUT is too small we use
...
...
@@ -4578,7 +4572,7 @@ sds clusterGenNodeDescription(clusterNode *node, int use_pport) {
/* Node coordinates */
ci
=
sdscatlen
(
sdsempty
(),
node
->
name
,
CLUSTER_NAMELEN
);
if
(
node
->
hostname
)
{
if
(
sdslen
(
node
->
hostname
)
!=
0
)
{
ci
=
sdscatfmt
(
ci
,
" %s:%i@%i,%s "
,
node
->
ip
,
port
,
...
...
@@ -4804,7 +4798,7 @@ void addReplyClusterLinksDescription(client *c) {
const
char
*
getPreferredEndpoint
(
clusterNode
*
n
)
{
switch
(
server
.
cluster_preferred_endpoint_type
)
{
case
CLUSTER_ENDPOINT_TYPE_IP
:
return
n
->
ip
;
case
CLUSTER_ENDPOINT_TYPE_HOSTNAME
:
return
n
->
hostname
?
n
->
hostname
:
"?"
;
case
CLUSTER_ENDPOINT_TYPE_HOSTNAME
:
return
(
sdslen
(
n
->
hostname
)
!=
0
)
?
n
->
hostname
:
"?"
;
case
CLUSTER_ENDPOINT_TYPE_UNKNOWN_ENDPOINT
:
return
""
;
}
return
"unknown"
;
...
...
@@ -4898,7 +4892,7 @@ void addNodeToNodeReply(client *c, clusterNode *node) {
if
(
server
.
cluster_preferred_endpoint_type
==
CLUSTER_ENDPOINT_TYPE_IP
)
{
addReplyBulkCString
(
c
,
node
->
ip
);
}
else
if
(
server
.
cluster_preferred_endpoint_type
==
CLUSTER_ENDPOINT_TYPE_HOSTNAME
)
{
addReplyBulkCString
(
c
,
node
->
hostname
?
node
->
hostname
:
"?"
);
addReplyBulkCString
(
c
,
sdslen
(
node
->
hostname
)
!=
0
?
node
->
hostname
:
"?"
);
}
else
if
(
server
.
cluster_preferred_endpoint_type
==
CLUSTER_ENDPOINT_TYPE_UNKNOWN_ENDPOINT
)
{
addReplyNull
(
c
);
}
else
{
...
...
@@ -4921,7 +4915,7 @@ void addNodeToNodeReply(client *c, clusterNode *node) {
length
++
;
}
if
(
server
.
cluster_preferred_endpoint_type
!=
CLUSTER_ENDPOINT_TYPE_HOSTNAME
&&
node
->
hostname
)
&&
sdslen
(
node
->
hostname
)
!=
0
)
{
addReplyBulkCString
(
c
,
"hostname"
);
addReplyBulkCString
(
c
,
node
->
hostname
);
...
...
@@ -5032,7 +5026,7 @@ void clusterCommand(client *c) {
" Reset current node (default: soft)."
,
"SET-CONFIG-EPOCH <epoch>"
,
" Set config epoch of current node."
,
"SETSLOT <slot> (IMPORTING
|MIGRATING
|STABLE|NODE <node-id>)"
,
"SETSLOT <slot> (IMPORTING
<node-id>|MIGRATING <node-id>
|STABLE|NODE <node-id>)"
,
" Set slot state."
,
"REPLICAS <node-id>"
,
" Return <node-id> replicas."
,
...
...
@@ -5226,6 +5220,10 @@ NULL
(
char
*
)
c
->
argv
[
4
]
->
ptr
);
return
;
}
if
(
nodeIsSlave
(
n
))
{
addReplyError
(
c
,
"Target node is not a master"
);
return
;
}
/* If this hash slot was served by 'myself' before to switch
* make sure there are no longer local keys for this hash slot. */
if
(
server
.
cluster
->
slots
[
slot
]
==
myself
&&
n
!=
myself
)
{
...
...
@@ -5285,7 +5283,7 @@ NULL
addReplySds
(
c
,
reply
);
}
else
if
(
!
strcasecmp
(
c
->
argv
[
1
]
->
ptr
,
"info"
)
&&
c
->
argc
==
2
)
{
/* CLUSTER INFO */
char
*
statestr
[]
=
{
"ok"
,
"fail"
,
"needhelp"
};
char
*
statestr
[]
=
{
"ok"
,
"fail"
};
int
slots_assigned
=
0
,
slots_ok
=
0
,
slots_pfail
=
0
,
slots_fail
=
0
;
uint64_t
myepoch
;
int
j
;
...
...
@@ -5703,7 +5701,7 @@ int verifyDumpPayload(unsigned char *p, size_t len, uint16_t *rdbver_ptr) {
if
(
len
<
10
)
return
C_ERR
;
footer
=
p
+
(
len
-
10
);
/*
V
erify RDB version */
/*
Set and v
erify RDB version
.
*/
rdbver
=
(
footer
[
1
]
<<
8
)
|
footer
[
0
];
if
(
rdbver_ptr
)
{
*
rdbver_ptr
=
rdbver
;
...
...
src/cluster.h
View file @
d2b5a579
...
...
@@ -96,8 +96,8 @@ typedef struct clusterLink {
#define CLUSTERMSG_TYPE_UPDATE 7
/* Another node slots configuration */
#define CLUSTERMSG_TYPE_MFSTART 8
/* Pause clients for manual failover */
#define CLUSTERMSG_TYPE_MODULE 9
/* Module cluster API message. */
#define CLUSTERMSG_TYPE_
COUNT 10
/* Total number of message types.
*/
#define CLUSTERMSG_TYPE_
PUBLISHSHARD 11
/* Pub/Sub Publish shard propagation
*/
#define CLUSTERMSG_TYPE_
PUBLISHSHARD 10
/* Pub/Sub Publish shard propagation
*/
#define CLUSTERMSG_TYPE_
COUNT 11
/* Total number of message types.
*/
/* Flags that a module can set in order to prevent certain Redis Cluster
* features to be enabled. Useful when implementing a different distributed
...
...
@@ -134,8 +134,8 @@ typedef struct clusterNode {
mstime_t
repl_offset_time
;
/* Unix time we received offset for this node */
mstime_t
orphaned_time
;
/* Starting time of orphaned master condition */
long
long
repl_offset
;
/* Last known repl offset for this node. */
char
ip
[
NET_IP_STR_LEN
];
/* Latest known IP address of this node */
char
*
hostname
;
/* The known hostname for this node */
char
ip
[
NET_IP_STR_LEN
];
/* Latest known IP address of this node */
sds
hostname
;
/* The known hostname for this node */
int
port
;
/* Latest known clients port (TLS or plain). */
int
pport
;
/* Latest known clients plaintext port. Only used
if the main clients port is for TLS. */
...
...
@@ -339,8 +339,6 @@ typedef struct {
* changes in clusterMsg be caught at compile time.
*/
/* Avoid static_assert on non-C11 compilers. */
#if __STDC_VERSION__ >= 201112L
static_assert
(
offsetof
(
clusterMsg
,
sig
)
==
0
,
"unexpected field offset"
);
static_assert
(
offsetof
(
clusterMsg
,
totlen
)
==
4
,
"unexpected field offset"
);
static_assert
(
offsetof
(
clusterMsg
,
ver
)
==
8
,
"unexpected field offset"
);
...
...
@@ -362,7 +360,6 @@ static_assert(offsetof(clusterMsg, flags) == 2250, "unexpected field offset");
static_assert
(
offsetof
(
clusterMsg
,
state
)
==
2252
,
"unexpected field offset"
);
static_assert
(
offsetof
(
clusterMsg
,
mflags
)
==
2253
,
"unexpected field offset"
);
static_assert
(
offsetof
(
clusterMsg
,
data
)
==
2256
,
"unexpected field offset"
);
#endif
#define CLUSTERMSG_MIN_LEN (sizeof(clusterMsg)-sizeof(union clusterMsgData))
...
...
Prev
1
2
3
4
5
6
…
8
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