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
a83e3663
Commit
a83e3663
authored
Feb 28, 2022
by
Oran Agra
Browse files
Merge remote-tracking branch 'origin/unstable' into 7.0
parents
d5915a16
5860fa3d
Changes
142
Expand all
Hide whitespace changes
Inline
Side-by-side
deps/hiredis/fmacros.h
View file @
a83e3663
#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 @
a83e3663
/*
* 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 @
a83e3663
...
...
@@ -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 @
a83e3663
...
...
@@ -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 @
a83e3663
<?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 @
a83e3663
...
...
@@ -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 @
a83e3663
...
...
@@ -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 @
a83e3663
...
...
@@ -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 @
a83e3663
...
...
@@ -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 @
a83e3663
This diff is collapsed.
Click to expand it.
redis.conf
View file @
a83e3663
...
...
@@ -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 @
a83e3663
...
...
@@ -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 @
a83e3663
...
...
@@ -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 @
a83e3663
...
...
@@ -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 @
a83e3663
...
...
@@ -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 @
a83e3663
...
...
@@ -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 @
a83e3663
...
...
@@ -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 @
a83e3663
...
...
@@ -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 @
a83e3663
...
...
@@ -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))
...
...
src/commands.c
View file @
a83e3663
This diff is collapsed.
Click to expand it.
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