Unverified Commit 8341a7d4 authored by chendianqiang's avatar chendianqiang Committed by GitHub
Browse files

Merge pull request #3 from antirez/unstable

update
parents 49816941 e78c4e81
...@@ -46,7 +46,7 @@ static int checkStringLength(client *c, long long size) { ...@@ -46,7 +46,7 @@ static int checkStringLength(client *c, long long size) {
* options and variants. This function is called in order to implement the * options and variants. This function is called in order to implement the
* following commands: SET, SETEX, PSETEX, SETNX. * following commands: SET, SETEX, PSETEX, SETNX.
* *
* 'flags' changes the behavior of the command (NX or XX, see belove). * 'flags' changes the behavior of the command (NX or XX, see below).
* *
* 'expire' represents an expire to set in form of a Redis object as passed * 'expire' represents an expire to set in form of a Redis object as passed
* by the user. It is interpreted according to the specified 'unit'. * by the user. It is interpreted according to the specified 'unit'.
...@@ -63,6 +63,7 @@ static int checkStringLength(client *c, long long size) { ...@@ -63,6 +63,7 @@ static int checkStringLength(client *c, long long size) {
#define OBJ_SET_XX (1<<1) /* Set if key exists. */ #define OBJ_SET_XX (1<<1) /* Set if key exists. */
#define OBJ_SET_EX (1<<2) /* Set if time in seconds is given */ #define OBJ_SET_EX (1<<2) /* Set if time in seconds is given */
#define OBJ_SET_PX (1<<3) /* Set if time in ms in given */ #define OBJ_SET_PX (1<<3) /* Set if time in ms in given */
#define OBJ_SET_KEEPTTL (1<<4) /* Set and keep the ttl */
void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, int unit, robj *ok_reply, robj *abort_reply) { void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, int unit, robj *ok_reply, robj *abort_reply) {
long long milliseconds = 0; /* initialized to avoid any harmness warning */ long long milliseconds = 0; /* initialized to avoid any harmness warning */
...@@ -83,7 +84,7 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, ...@@ -83,7 +84,7 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire,
addReply(c, abort_reply ? abort_reply : shared.null[c->resp]); addReply(c, abort_reply ? abort_reply : shared.null[c->resp]);
return; return;
} }
setKey(c->db,key,val); genericSetKey(c->db,key,val,flags & OBJ_SET_KEEPTTL);
server.dirty++; server.dirty++;
if (expire) setExpire(c,c->db,key,mstime()+milliseconds); if (expire) setExpire(c,c->db,key,mstime()+milliseconds);
notifyKeyspaceEvent(NOTIFY_STRING,"set",key,c->db->id); notifyKeyspaceEvent(NOTIFY_STRING,"set",key,c->db->id);
...@@ -92,7 +93,7 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, ...@@ -92,7 +93,7 @@ void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire,
addReply(c, ok_reply ? ok_reply : shared.ok); addReply(c, ok_reply ? ok_reply : shared.ok);
} }
/* SET key value [NX] [XX] [EX <seconds>] [PX <milliseconds>] */ /* SET key value [NX] [XX] [KEEPTTL] [EX <seconds>] [PX <milliseconds>] */
void setCommand(client *c) { void setCommand(client *c) {
int j; int j;
robj *expire = NULL; robj *expire = NULL;
...@@ -113,8 +114,13 @@ void setCommand(client *c) { ...@@ -113,8 +114,13 @@ void setCommand(client *c) {
!(flags & OBJ_SET_NX)) !(flags & OBJ_SET_NX))
{ {
flags |= OBJ_SET_XX; flags |= OBJ_SET_XX;
} else if (!strcasecmp(c->argv[j]->ptr,"KEEPTTL") &&
!(flags & OBJ_SET_EX) && !(flags & OBJ_SET_PX))
{
flags |= OBJ_SET_KEEPTTL;
} else if ((a[0] == 'e' || a[0] == 'E') && } else if ((a[0] == 'e' || a[0] == 'E') &&
(a[1] == 'x' || a[1] == 'X') && a[2] == '\0' && (a[1] == 'x' || a[1] == 'X') && a[2] == '\0' &&
!(flags & OBJ_SET_KEEPTTL) &&
!(flags & OBJ_SET_PX) && next) !(flags & OBJ_SET_PX) && next)
{ {
flags |= OBJ_SET_EX; flags |= OBJ_SET_EX;
...@@ -123,6 +129,7 @@ void setCommand(client *c) { ...@@ -123,6 +129,7 @@ void setCommand(client *c) {
j++; j++;
} else if ((a[0] == 'p' || a[0] == 'P') && } else if ((a[0] == 'p' || a[0] == 'P') &&
(a[1] == 'x' || a[1] == 'X') && a[2] == '\0' && (a[1] == 'x' || a[1] == 'X') && a[2] == '\0' &&
!(flags & OBJ_SET_KEEPTTL) &&
!(flags & OBJ_SET_EX) && next) !(flags & OBJ_SET_EX) && next)
{ {
flags |= OBJ_SET_PX; flags |= OBJ_SET_PX;
...@@ -398,7 +405,7 @@ void decrbyCommand(client *c) { ...@@ -398,7 +405,7 @@ void decrbyCommand(client *c) {
void incrbyfloatCommand(client *c) { void incrbyfloatCommand(client *c) {
long double incr, value; long double incr, value;
robj *o, *new, *aux; robj *o, *new, *aux1, *aux2;
o = lookupKeyWrite(c->db,c->argv[1]); o = lookupKeyWrite(c->db,c->argv[1]);
if (o != NULL && checkType(c,o,OBJ_STRING)) return; if (o != NULL && checkType(c,o,OBJ_STRING)) return;
...@@ -424,10 +431,13 @@ void incrbyfloatCommand(client *c) { ...@@ -424,10 +431,13 @@ void incrbyfloatCommand(client *c) {
/* Always replicate INCRBYFLOAT as a SET command with the final value /* Always replicate INCRBYFLOAT as a SET command with the final value
* in order to make sure that differences in float precision or formatting * in order to make sure that differences in float precision or formatting
* will not create differences in replicas or after an AOF restart. */ * will not create differences in replicas or after an AOF restart. */
aux = createStringObject("SET",3); aux1 = createStringObject("SET",3);
rewriteClientCommandArgument(c,0,aux); rewriteClientCommandArgument(c,0,aux1);
decrRefCount(aux); decrRefCount(aux1);
rewriteClientCommandArgument(c,2,new); rewriteClientCommandArgument(c,2,new);
aux2 = createStringObject("KEEPTTL",7);
rewriteClientCommandArgument(c,3,aux2);
decrRefCount(aux2);
} }
void appendCommand(client *c) { void appendCommand(client *c) {
......
...@@ -1357,9 +1357,8 @@ int zsetAdd(robj *zobj, double score, sds ele, int *flags, double *newscore) { ...@@ -1357,9 +1357,8 @@ int zsetAdd(robj *zobj, double score, sds ele, int *flags, double *newscore) {
/* Optimize: check if the element is too large or the list /* Optimize: check if the element is too large or the list
* becomes too long *before* executing zzlInsert. */ * becomes too long *before* executing zzlInsert. */
zobj->ptr = zzlInsert(zobj->ptr,ele,score); zobj->ptr = zzlInsert(zobj->ptr,ele,score);
if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries) if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries ||
zsetConvert(zobj,OBJ_ENCODING_SKIPLIST); sdslen(ele) > server.zset_max_ziplist_value)
if (sdslen(ele) > server.zset_max_ziplist_value)
zsetConvert(zobj,OBJ_ENCODING_SKIPLIST); zsetConvert(zobj,OBJ_ENCODING_SKIPLIST);
if (newscore) *newscore = score; if (newscore) *newscore = score;
*flags |= ZADD_ADDED; *flags |= ZADD_ADDED;
...@@ -2427,7 +2426,7 @@ void zrangeGenericCommand(client *c, int reverse) { ...@@ -2427,7 +2426,7 @@ void zrangeGenericCommand(client *c, int reverse) {
return; return;
} }
if ((zobj = lookupKeyReadOrReply(c,key,shared.null[c->resp])) == NULL if ((zobj = lookupKeyReadOrReply(c,key,shared.emptyarray)) == NULL
|| checkType(c,zobj,OBJ_ZSET)) return; || checkType(c,zobj,OBJ_ZSET)) return;
/* Sanitize indexes. */ /* Sanitize indexes. */
...@@ -2439,7 +2438,7 @@ void zrangeGenericCommand(client *c, int reverse) { ...@@ -2439,7 +2438,7 @@ void zrangeGenericCommand(client *c, int reverse) {
/* Invariant: start >= 0, so this test will be true when end < 0. /* Invariant: start >= 0, so this test will be true when end < 0.
* The range is empty when start > end or start >= length. */ * The range is empty when start > end or start >= length. */
if (start > end || start >= llen) { if (start > end || start >= llen) {
addReplyNull(c); addReply(c,shared.emptyarray);
return; return;
} }
if (end >= llen) end = llen-1; if (end >= llen) end = llen-1;
...@@ -2575,7 +2574,7 @@ void genericZrangebyscoreCommand(client *c, int reverse) { ...@@ -2575,7 +2574,7 @@ void genericZrangebyscoreCommand(client *c, int reverse) {
} }
/* Ok, lookup the key and get the range */ /* Ok, lookup the key and get the range */
if ((zobj = lookupKeyReadOrReply(c,key,shared.null[c->resp])) == NULL || if ((zobj = lookupKeyReadOrReply(c,key,shared.emptyarray)) == NULL ||
checkType(c,zobj,OBJ_ZSET)) return; checkType(c,zobj,OBJ_ZSET)) return;
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) { if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
...@@ -2595,7 +2594,7 @@ void genericZrangebyscoreCommand(client *c, int reverse) { ...@@ -2595,7 +2594,7 @@ void genericZrangebyscoreCommand(client *c, int reverse) {
/* No "first" element in the specified interval. */ /* No "first" element in the specified interval. */
if (eptr == NULL) { if (eptr == NULL) {
addReplyNull(c); addReply(c,shared.emptyarray);
return; return;
} }
...@@ -2662,7 +2661,7 @@ void genericZrangebyscoreCommand(client *c, int reverse) { ...@@ -2662,7 +2661,7 @@ void genericZrangebyscoreCommand(client *c, int reverse) {
/* No "first" element in the specified interval. */ /* No "first" element in the specified interval. */
if (ln == NULL) { if (ln == NULL) {
addReplyNull(c); addReply(c,shared.emptyarray);
return; return;
} }
...@@ -2906,7 +2905,10 @@ void genericZrangebylexCommand(client *c, int reverse) { ...@@ -2906,7 +2905,10 @@ void genericZrangebylexCommand(client *c, int reverse) {
while (remaining) { while (remaining) {
if (remaining >= 3 && !strcasecmp(c->argv[pos]->ptr,"limit")) { if (remaining >= 3 && !strcasecmp(c->argv[pos]->ptr,"limit")) {
if ((getLongFromObjectOrReply(c, c->argv[pos+1], &offset, NULL) != C_OK) || if ((getLongFromObjectOrReply(c, c->argv[pos+1], &offset, NULL) != C_OK) ||
(getLongFromObjectOrReply(c, c->argv[pos+2], &limit, NULL) != C_OK)) return; (getLongFromObjectOrReply(c, c->argv[pos+2], &limit, NULL) != C_OK)) {
zslFreeLexRange(&range);
return;
}
pos += 3; remaining -= 3; pos += 3; remaining -= 3;
} else { } else {
zslFreeLexRange(&range); zslFreeLexRange(&range);
...@@ -2917,7 +2919,7 @@ void genericZrangebylexCommand(client *c, int reverse) { ...@@ -2917,7 +2919,7 @@ void genericZrangebylexCommand(client *c, int reverse) {
} }
/* Ok, lookup the key and get the range */ /* Ok, lookup the key and get the range */
if ((zobj = lookupKeyReadOrReply(c,key,shared.null[c->resp])) == NULL || if ((zobj = lookupKeyReadOrReply(c,key,shared.emptyarray)) == NULL ||
checkType(c,zobj,OBJ_ZSET)) checkType(c,zobj,OBJ_ZSET))
{ {
zslFreeLexRange(&range); zslFreeLexRange(&range);
...@@ -2940,7 +2942,7 @@ void genericZrangebylexCommand(client *c, int reverse) { ...@@ -2940,7 +2942,7 @@ void genericZrangebylexCommand(client *c, int reverse) {
/* No "first" element in the specified interval. */ /* No "first" element in the specified interval. */
if (eptr == NULL) { if (eptr == NULL) {
addReplyNull(c); addReply(c,shared.emptyarray);
zslFreeLexRange(&range); zslFreeLexRange(&range);
return; return;
} }
...@@ -3004,7 +3006,7 @@ void genericZrangebylexCommand(client *c, int reverse) { ...@@ -3004,7 +3006,7 @@ void genericZrangebylexCommand(client *c, int reverse) {
/* No "first" element in the specified interval. */ /* No "first" element in the specified interval. */
if (ln == NULL) { if (ln == NULL) {
addReplyNull(c); addReply(c,shared.emptyarray);
zslFreeLexRange(&range); zslFreeLexRange(&range);
return; return;
} }
...@@ -3140,7 +3142,10 @@ void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey ...@@ -3140,7 +3142,10 @@ void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey
if (countarg) { if (countarg) {
if (getLongFromObjectOrReply(c,countarg,&count,NULL) != C_OK) if (getLongFromObjectOrReply(c,countarg,&count,NULL) != C_OK)
return; return;
if (count < 0) count = 1; if (count <= 0) {
addReply(c,shared.emptyarray);
return;
}
} }
/* Check type and break on the first error, otherwise identify candidate. */ /* Check type and break on the first error, otherwise identify candidate. */
...@@ -3155,7 +3160,7 @@ void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey ...@@ -3155,7 +3160,7 @@ void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey
/* No candidate for zpopping, return empty. */ /* No candidate for zpopping, return empty. */
if (!zobj) { if (!zobj) {
addReplyNull(c); addReply(c,shared.emptyarray);
return; return;
} }
......
/*
* Copyright (c) 2019, Redis Labs
* 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 "server.h"
#include "connhelpers.h"
#include "adlist.h"
#ifdef USE_OPENSSL
#include <openssl/ssl.h>
#include <openssl/err.h>
#include <openssl/rand.h>
#define REDIS_TLS_PROTO_TLSv1 (1<<0)
#define REDIS_TLS_PROTO_TLSv1_1 (1<<1)
#define REDIS_TLS_PROTO_TLSv1_2 (1<<2)
#define REDIS_TLS_PROTO_TLSv1_3 (1<<3)
/* Use safe defaults */
#ifdef TLS1_3_VERSION
#define REDIS_TLS_PROTO_DEFAULT (REDIS_TLS_PROTO_TLSv1_2|REDIS_TLS_PROTO_TLSv1_3)
#else
#define REDIS_TLS_PROTO_DEFAULT (REDIS_TLS_PROTO_TLSv1_2)
#endif
extern ConnectionType CT_Socket;
SSL_CTX *redis_tls_ctx;
static int parseProtocolsConfig(const char *str) {
int i, count = 0;
int protocols = 0;
if (!str) return REDIS_TLS_PROTO_DEFAULT;
sds *tokens = sdssplitlen(str, strlen(str), " ", 1, &count);
if (!tokens) {
serverLog(LL_WARNING, "Invalid tls-protocols configuration string");
return -1;
}
for (i = 0; i < count; i++) {
if (!strcasecmp(tokens[i], "tlsv1")) protocols |= REDIS_TLS_PROTO_TLSv1;
else if (!strcasecmp(tokens[i], "tlsv1.1")) protocols |= REDIS_TLS_PROTO_TLSv1_1;
else if (!strcasecmp(tokens[i], "tlsv1.2")) protocols |= REDIS_TLS_PROTO_TLSv1_2;
else if (!strcasecmp(tokens[i], "tlsv1.3")) {
#ifdef TLS1_3_VERSION
protocols |= REDIS_TLS_PROTO_TLSv1_3;
#else
serverLog(LL_WARNING, "TLSv1.3 is specified in tls-protocols but not supported by OpenSSL.");
protocols = -1;
break;
#endif
} else {
serverLog(LL_WARNING, "Invalid tls-protocols specified. "
"Use a combination of 'TLSv1', 'TLSv1.1', 'TLSv1.2' and 'TLSv1.3'.");
protocols = -1;
break;
}
}
sdsfreesplitres(tokens, count);
return protocols;
}
/* list of connections with pending data already read from the socket, but not
* served to the reader yet. */
static list *pending_list = NULL;
void tlsInit(void) {
ERR_load_crypto_strings();
SSL_load_error_strings();
SSL_library_init();
if (!RAND_poll()) {
serverLog(LL_WARNING, "OpenSSL: Failed to seed random number generator.");
}
pending_list = listCreate();
/* Server configuration */
server.tls_auth_clients = 1; /* Secure by default */
}
/* Attempt to configure/reconfigure TLS. This operation is atomic and will
* leave the SSL_CTX unchanged if fails.
*/
int tlsConfigure(redisTLSContextConfig *ctx_config) {
char errbuf[256];
SSL_CTX *ctx = NULL;
if (!ctx_config->cert_file) {
serverLog(LL_WARNING, "No tls-cert-file configured!");
goto error;
}
if (!ctx_config->key_file) {
serverLog(LL_WARNING, "No tls-key-file configured!");
goto error;
}
if (!ctx_config->ca_cert_file && !ctx_config->ca_cert_dir) {
serverLog(LL_WARNING, "Either tls-ca-cert-file or tls-ca-cert-dir must be configured!");
goto error;
}
ctx = SSL_CTX_new(SSLv23_method());
SSL_CTX_set_options(ctx, SSL_OP_NO_SSLv2|SSL_OP_NO_SSLv3);
SSL_CTX_set_options(ctx, SSL_OP_SINGLE_DH_USE);
#ifdef SSL_OP_DONT_INSERT_EMPTY_FRAGMENTS
SSL_CTX_set_options(ctx, SSL_OP_DONT_INSERT_EMPTY_FRAGMENTS);
#endif
int protocols = parseProtocolsConfig(ctx_config->protocols);
if (protocols == -1) goto error;
if (!(protocols & REDIS_TLS_PROTO_TLSv1))
SSL_CTX_set_options(ctx, SSL_OP_NO_TLSv1);
if (!(protocols & REDIS_TLS_PROTO_TLSv1_1))
SSL_CTX_set_options(ctx, SSL_OP_NO_TLSv1_1);
#ifdef SSL_OP_NO_TLSv1_2
if (!(protocols & REDIS_TLS_PROTO_TLSv1_2))
SSL_CTX_set_options(ctx, SSL_OP_NO_TLSv1_2);
#endif
#ifdef SSL_OP_NO_TLSv1_3
if (!(protocols & REDIS_TLS_PROTO_TLSv1_3))
SSL_CTX_set_options(ctx, SSL_OP_NO_TLSv1_3);
#endif
#ifdef SSL_OP_NO_COMPRESSION
SSL_CTX_set_options(ctx, SSL_OP_NO_COMPRESSION);
#endif
#ifdef SSL_OP_NO_CLIENT_RENEGOTIATION
SSL_CTX_set_options(ssl->ctx, SSL_OP_NO_CLIENT_RENEGOTIATION);
#endif
if (ctx_config->prefer_server_ciphers)
SSL_CTX_set_options(ctx, SSL_OP_CIPHER_SERVER_PREFERENCE);
SSL_CTX_set_mode(ctx, SSL_MODE_ENABLE_PARTIAL_WRITE|SSL_MODE_ACCEPT_MOVING_WRITE_BUFFER);
SSL_CTX_set_verify(ctx, SSL_VERIFY_PEER|SSL_VERIFY_FAIL_IF_NO_PEER_CERT, NULL);
SSL_CTX_set_ecdh_auto(ctx, 1);
if (SSL_CTX_use_certificate_file(ctx, ctx_config->cert_file, SSL_FILETYPE_PEM) <= 0) {
ERR_error_string_n(ERR_get_error(), errbuf, sizeof(errbuf));
serverLog(LL_WARNING, "Failed to load certificate: %s: %s", ctx_config->cert_file, errbuf);
goto error;
}
if (SSL_CTX_use_PrivateKey_file(ctx, ctx_config->key_file, SSL_FILETYPE_PEM) <= 0) {
ERR_error_string_n(ERR_get_error(), errbuf, sizeof(errbuf));
serverLog(LL_WARNING, "Failed to load private key: %s: %s", ctx_config->key_file, errbuf);
goto error;
}
if (SSL_CTX_load_verify_locations(ctx, ctx_config->ca_cert_file, ctx_config->ca_cert_dir) <= 0) {
ERR_error_string_n(ERR_get_error(), errbuf, sizeof(errbuf));
serverLog(LL_WARNING, "Failed to configure CA certificate(s) file/directory: %s", errbuf);
goto error;
}
if (ctx_config->dh_params_file) {
FILE *dhfile = fopen(ctx_config->dh_params_file, "r");
DH *dh = NULL;
if (!dhfile) {
serverLog(LL_WARNING, "Failed to load %s: %s", ctx_config->dh_params_file, strerror(errno));
goto error;
}
dh = PEM_read_DHparams(dhfile, NULL, NULL, NULL);
fclose(dhfile);
if (!dh) {
serverLog(LL_WARNING, "%s: failed to read DH params.", ctx_config->dh_params_file);
goto error;
}
if (SSL_CTX_set_tmp_dh(ctx, dh) <= 0) {
ERR_error_string_n(ERR_get_error(), errbuf, sizeof(errbuf));
serverLog(LL_WARNING, "Failed to load DH params file: %s: %s", ctx_config->dh_params_file, errbuf);
DH_free(dh);
goto error;
}
DH_free(dh);
}
if (ctx_config->ciphers && !SSL_CTX_set_cipher_list(ctx, ctx_config->ciphers)) {
serverLog(LL_WARNING, "Failed to configure ciphers: %s", ctx_config->ciphers);
goto error;
}
#ifdef TLS1_3_VERSION
if (ctx_config->ciphersuites && !SSL_CTX_set_ciphersuites(ctx, ctx_config->ciphersuites)) {
serverLog(LL_WARNING, "Failed to configure ciphersuites: %s", ctx_config->ciphersuites);
goto error;
}
#endif
SSL_CTX_free(redis_tls_ctx);
redis_tls_ctx = ctx;
return C_OK;
error:
if (ctx) SSL_CTX_free(ctx);
return C_ERR;
}
#ifdef TLS_DEBUGGING
#define TLSCONN_DEBUG(fmt, ...) \
serverLog(LL_DEBUG, "TLSCONN: " fmt, __VA_ARGS__)
#else
#define TLSCONN_DEBUG(fmt, ...)
#endif
ConnectionType CT_TLS;
/* Normal socket connections have a simple events/handler correlation.
*
* With TLS connections we need to handle cases where during a logical read
* or write operation, the SSL library asks to block for the opposite
* socket operation.
*
* When this happens, we need to do two things:
* 1. Make sure we register for the even.
* 2. Make sure we know which handler needs to execute when the
* event fires. That is, if we notify the caller of a write operation
* that it blocks, and SSL asks for a read, we need to trigger the
* write handler again on the next read event.
*
*/
typedef enum {
WANT_READ = 1,
WANT_WRITE
} WantIOType;
#define TLS_CONN_FLAG_READ_WANT_WRITE (1<<0)
#define TLS_CONN_FLAG_WRITE_WANT_READ (1<<1)
#define TLS_CONN_FLAG_FD_SET (1<<2)
typedef struct tls_connection {
connection c;
int flags;
SSL *ssl;
char *ssl_error;
listNode *pending_list_node;
} tls_connection;
connection *connCreateTLS(void) {
tls_connection *conn = zcalloc(sizeof(tls_connection));
conn->c.type = &CT_TLS;
conn->c.fd = -1;
conn->ssl = SSL_new(redis_tls_ctx);
return (connection *) conn;
}
connection *connCreateAcceptedTLS(int fd, int require_auth) {
tls_connection *conn = (tls_connection *) connCreateTLS();
conn->c.fd = fd;
conn->c.state = CONN_STATE_ACCEPTING;
if (!require_auth) {
/* We still verify certificates if provided, but don't require them.
*/
SSL_set_verify(conn->ssl, SSL_VERIFY_PEER, NULL);
}
SSL_set_fd(conn->ssl, conn->c.fd);
SSL_set_accept_state(conn->ssl);
return (connection *) conn;
}
static void tlsEventHandler(struct aeEventLoop *el, int fd, void *clientData, int mask);
/* Process the return code received from OpenSSL>
* Update the want parameter with expected I/O.
* Update the connection's error state if a real error has occured.
* Returns an SSL error code, or 0 if no further handling is required.
*/
static int handleSSLReturnCode(tls_connection *conn, int ret_value, WantIOType *want) {
if (ret_value <= 0) {
int ssl_err = SSL_get_error(conn->ssl, ret_value);
switch (ssl_err) {
case SSL_ERROR_WANT_WRITE:
*want = WANT_WRITE;
return 0;
case SSL_ERROR_WANT_READ:
*want = WANT_READ;
return 0;
case SSL_ERROR_SYSCALL:
conn->c.last_errno = errno;
if (conn->ssl_error) zfree(conn->ssl_error);
conn->ssl_error = errno ? zstrdup(strerror(errno)) : NULL;
break;
default:
/* Error! */
conn->c.last_errno = 0;
if (conn->ssl_error) zfree(conn->ssl_error);
conn->ssl_error = zmalloc(512);
ERR_error_string_n(ERR_get_error(), conn->ssl_error, 512);
break;
}
return ssl_err;
}
return 0;
}
void registerSSLEvent(tls_connection *conn, WantIOType want) {
int mask = aeGetFileEvents(server.el, conn->c.fd);
switch (want) {
case WANT_READ:
if (mask & AE_WRITABLE) aeDeleteFileEvent(server.el, conn->c.fd, AE_WRITABLE);
if (!(mask & AE_READABLE)) aeCreateFileEvent(server.el, conn->c.fd, AE_READABLE,
tlsEventHandler, conn);
break;
case WANT_WRITE:
if (mask & AE_READABLE) aeDeleteFileEvent(server.el, conn->c.fd, AE_READABLE);
if (!(mask & AE_WRITABLE)) aeCreateFileEvent(server.el, conn->c.fd, AE_WRITABLE,
tlsEventHandler, conn);
break;
default:
serverAssert(0);
break;
}
}
void updateSSLEvent(tls_connection *conn) {
int mask = aeGetFileEvents(server.el, conn->c.fd);
int need_read = conn->c.read_handler || (conn->flags & TLS_CONN_FLAG_WRITE_WANT_READ);
int need_write = conn->c.write_handler || (conn->flags & TLS_CONN_FLAG_READ_WANT_WRITE);
if (need_read && !(mask & AE_READABLE))
aeCreateFileEvent(server.el, conn->c.fd, AE_READABLE, tlsEventHandler, conn);
if (!need_read && (mask & AE_READABLE))
aeDeleteFileEvent(server.el, conn->c.fd, AE_READABLE);
if (need_write && !(mask & AE_WRITABLE))
aeCreateFileEvent(server.el, conn->c.fd, AE_WRITABLE, tlsEventHandler, conn);
if (!need_write && (mask & AE_WRITABLE))
aeDeleteFileEvent(server.el, conn->c.fd, AE_WRITABLE);
}
static void tlsHandleEvent(tls_connection *conn, int mask) {
int ret;
TLSCONN_DEBUG("tlsEventHandler(): fd=%d, state=%d, mask=%d, r=%d, w=%d, flags=%d",
fd, conn->c.state, mask, conn->c.read_handler != NULL, conn->c.write_handler != NULL,
conn->flags);
ERR_clear_error();
switch (conn->c.state) {
case CONN_STATE_CONNECTING:
if (connGetSocketError((connection *) conn)) {
conn->c.last_errno = errno;
conn->c.state = CONN_STATE_ERROR;
} else {
if (!(conn->flags & TLS_CONN_FLAG_FD_SET)) {
SSL_set_fd(conn->ssl, conn->c.fd);
conn->flags |= TLS_CONN_FLAG_FD_SET;
}
ret = SSL_connect(conn->ssl);
if (ret <= 0) {
WantIOType want = 0;
if (!handleSSLReturnCode(conn, ret, &want)) {
registerSSLEvent(conn, want);
/* Avoid hitting UpdateSSLEvent, which knows nothing
* of what SSL_connect() wants and instead looks at our
* R/W handlers.
*/
return;
}
/* If not handled, it's an error */
conn->c.state = CONN_STATE_ERROR;
} else {
conn->c.state = CONN_STATE_CONNECTED;
}
}
if (!callHandler((connection *) conn, conn->c.conn_handler)) return;
conn->c.conn_handler = NULL;
break;
case CONN_STATE_ACCEPTING:
ret = SSL_accept(conn->ssl);
if (ret <= 0) {
WantIOType want = 0;
if (!handleSSLReturnCode(conn, ret, &want)) {
/* Avoid hitting UpdateSSLEvent, which knows nothing
* of what SSL_connect() wants and instead looks at our
* R/W handlers.
*/
registerSSLEvent(conn, want);
return;
}
/* If not handled, it's an error */
conn->c.state = CONN_STATE_ERROR;
} else {
conn->c.state = CONN_STATE_CONNECTED;
}
if (!callHandler((connection *) conn, conn->c.conn_handler)) return;
conn->c.conn_handler = NULL;
break;
case CONN_STATE_CONNECTED:
{
int call_read = ((mask & AE_READABLE) && conn->c.read_handler) ||
((mask & AE_WRITABLE) && (conn->flags & TLS_CONN_FLAG_READ_WANT_WRITE));
int call_write = ((mask & AE_WRITABLE) && conn->c.write_handler) ||
((mask & AE_READABLE) && (conn->flags & TLS_CONN_FLAG_WRITE_WANT_READ));
/* Normally we execute the readable event first, and the writable
* event laster. This is useful as sometimes we may be able
* to serve the reply of a query immediately after processing the
* query.
*
* However if WRITE_BARRIER is set in the mask, our application is
* asking us to do the reverse: never fire the writable event
* after the readable. In such a case, we invert the calls.
* This is useful when, for instance, we want to do things
* in the beforeSleep() hook, like fsynching a file to disk,
* before replying to a client. */
int invert = conn->c.flags & CONN_FLAG_WRITE_BARRIER;
if (!invert && call_read) {
conn->flags &= ~TLS_CONN_FLAG_READ_WANT_WRITE;
if (!callHandler((connection *) conn, conn->c.read_handler)) return;
}
/* Fire the writable event. */
if (call_write) {
conn->flags &= ~TLS_CONN_FLAG_WRITE_WANT_READ;
if (!callHandler((connection *) conn, conn->c.write_handler)) return;
}
/* If we have to invert the call, fire the readable event now
* after the writable one. */
if (invert && call_read) {
conn->flags &= ~TLS_CONN_FLAG_READ_WANT_WRITE;
if (!callHandler((connection *) conn, conn->c.read_handler)) return;
}
/* If SSL has pending that, already read from the socket, we're at
* risk of not calling the read handler again, make sure to add it
* to a list of pending connection that should be handled anyway. */
if ((mask & AE_READABLE)) {
if (SSL_pending(conn->ssl) > 0) {
if (!conn->pending_list_node) {
listAddNodeTail(pending_list, conn);
conn->pending_list_node = listLast(pending_list);
}
} else if (conn->pending_list_node) {
listDelNode(pending_list, conn->pending_list_node);
conn->pending_list_node = NULL;
}
}
break;
}
default:
break;
}
updateSSLEvent(conn);
}
static void tlsEventHandler(struct aeEventLoop *el, int fd, void *clientData, int mask) {
UNUSED(el);
UNUSED(fd);
tls_connection *conn = clientData;
tlsHandleEvent(conn, mask);
}
static void connTLSClose(connection *conn_) {
tls_connection *conn = (tls_connection *) conn_;
if (conn->ssl) {
SSL_free(conn->ssl);
conn->ssl = NULL;
}
if (conn->ssl_error) {
zfree(conn->ssl_error);
conn->ssl_error = NULL;
}
if (conn->pending_list_node) {
listDelNode(pending_list, conn->pending_list_node);
conn->pending_list_node = NULL;
}
CT_Socket.close(conn_);
}
static int connTLSAccept(connection *_conn, ConnectionCallbackFunc accept_handler) {
tls_connection *conn = (tls_connection *) _conn;
int ret;
if (conn->c.state != CONN_STATE_ACCEPTING) return C_ERR;
ERR_clear_error();
/* Try to accept */
conn->c.conn_handler = accept_handler;
ret = SSL_accept(conn->ssl);
if (ret <= 0) {
WantIOType want = 0;
if (!handleSSLReturnCode(conn, ret, &want)) {
registerSSLEvent(conn, want); /* We'll fire back */
return C_OK;
} else {
conn->c.state = CONN_STATE_ERROR;
return C_ERR;
}
}
conn->c.state = CONN_STATE_CONNECTED;
if (!callHandler((connection *) conn, conn->c.conn_handler)) return C_OK;
conn->c.conn_handler = NULL;
return C_OK;
}
static int connTLSConnect(connection *conn_, const char *addr, int port, const char *src_addr, ConnectionCallbackFunc connect_handler) {
tls_connection *conn = (tls_connection *) conn_;
if (conn->c.state != CONN_STATE_NONE) return C_ERR;
ERR_clear_error();
/* Initiate Socket connection first */
if (CT_Socket.connect(conn_, addr, port, src_addr, connect_handler) == C_ERR) return C_ERR;
/* Return now, once the socket is connected we'll initiate
* TLS connection from the event handler.
*/
return C_OK;
}
static int connTLSWrite(connection *conn_, const void *data, size_t data_len) {
tls_connection *conn = (tls_connection *) conn_;
int ret, ssl_err;
if (conn->c.state != CONN_STATE_CONNECTED) return -1;
ERR_clear_error();
ret = SSL_write(conn->ssl, data, data_len);
if (ret <= 0) {
WantIOType want = 0;
if (!(ssl_err = handleSSLReturnCode(conn, ret, &want))) {
if (want == WANT_READ) conn->flags |= TLS_CONN_FLAG_WRITE_WANT_READ;
updateSSLEvent(conn);
errno = EAGAIN;
return -1;
} else {
if (ssl_err == SSL_ERROR_ZERO_RETURN ||
((ssl_err == SSL_ERROR_SYSCALL && !errno))) {
conn->c.state = CONN_STATE_CLOSED;
return 0;
} else {
conn->c.state = CONN_STATE_ERROR;
return -1;
}
}
}
return ret;
}
static int connTLSRead(connection *conn_, void *buf, size_t buf_len) {
tls_connection *conn = (tls_connection *) conn_;
int ret;
int ssl_err;
if (conn->c.state != CONN_STATE_CONNECTED) return -1;
ERR_clear_error();
ret = SSL_read(conn->ssl, buf, buf_len);
if (ret <= 0) {
WantIOType want = 0;
if (!(ssl_err = handleSSLReturnCode(conn, ret, &want))) {
if (want == WANT_WRITE) conn->flags |= TLS_CONN_FLAG_READ_WANT_WRITE;
updateSSLEvent(conn);
errno = EAGAIN;
return -1;
} else {
if (ssl_err == SSL_ERROR_ZERO_RETURN ||
((ssl_err == SSL_ERROR_SYSCALL) && !errno)) {
conn->c.state = CONN_STATE_CLOSED;
return 0;
} else {
conn->c.state = CONN_STATE_ERROR;
return -1;
}
}
}
return ret;
}
static const char *connTLSGetLastError(connection *conn_) {
tls_connection *conn = (tls_connection *) conn_;
if (conn->ssl_error) return conn->ssl_error;
return NULL;
}
int connTLSSetWriteHandler(connection *conn, ConnectionCallbackFunc func, int barrier) {
conn->write_handler = func;
if (barrier)
conn->flags |= CONN_FLAG_WRITE_BARRIER;
else
conn->flags &= ~CONN_FLAG_WRITE_BARRIER;
updateSSLEvent((tls_connection *) conn);
return C_OK;
}
int connTLSSetReadHandler(connection *conn, ConnectionCallbackFunc func) {
conn->read_handler = func;
updateSSLEvent((tls_connection *) conn);
return C_OK;
}
static void setBlockingTimeout(tls_connection *conn, long long timeout) {
anetBlock(NULL, conn->c.fd);
anetSendTimeout(NULL, conn->c.fd, timeout);
anetRecvTimeout(NULL, conn->c.fd, timeout);
}
static void unsetBlockingTimeout(tls_connection *conn) {
anetNonBlock(NULL, conn->c.fd);
anetSendTimeout(NULL, conn->c.fd, 0);
anetRecvTimeout(NULL, conn->c.fd, 0);
}
static int connTLSBlockingConnect(connection *conn_, const char *addr, int port, long long timeout) {
tls_connection *conn = (tls_connection *) conn_;
int ret;
if (conn->c.state != CONN_STATE_NONE) return C_ERR;
/* Initiate socket blocking connect first */
if (CT_Socket.blocking_connect(conn_, addr, port, timeout) == C_ERR) return C_ERR;
/* Initiate TLS connection now. We set up a send/recv timeout on the socket,
* which means the specified timeout will not be enforced accurately. */
SSL_set_fd(conn->ssl, conn->c.fd);
setBlockingTimeout(conn, timeout);
if ((ret = SSL_connect(conn->ssl)) <= 0) {
conn->c.state = CONN_STATE_ERROR;
return C_ERR;
}
unsetBlockingTimeout(conn);
conn->c.state = CONN_STATE_CONNECTED;
return C_OK;
}
static ssize_t connTLSSyncWrite(connection *conn_, char *ptr, ssize_t size, long long timeout) {
tls_connection *conn = (tls_connection *) conn_;
setBlockingTimeout(conn, timeout);
SSL_clear_mode(conn->ssl, SSL_MODE_ENABLE_PARTIAL_WRITE);
int ret = SSL_write(conn->ssl, ptr, size);
SSL_set_mode(conn->ssl, SSL_MODE_ENABLE_PARTIAL_WRITE);
unsetBlockingTimeout(conn);
return ret;
}
static ssize_t connTLSSyncRead(connection *conn_, char *ptr, ssize_t size, long long timeout) {
tls_connection *conn = (tls_connection *) conn_;
setBlockingTimeout(conn, timeout);
int ret = SSL_read(conn->ssl, ptr, size);
unsetBlockingTimeout(conn);
return ret;
}
static ssize_t connTLSSyncReadLine(connection *conn_, char *ptr, ssize_t size, long long timeout) {
tls_connection *conn = (tls_connection *) conn_;
ssize_t nread = 0;
setBlockingTimeout(conn, timeout);
size--;
while(size) {
char c;
if (SSL_read(conn->ssl,&c,1) <= 0) {
nread = -1;
goto exit;
}
if (c == '\n') {
*ptr = '\0';
if (nread && *(ptr-1) == '\r') *(ptr-1) = '\0';
goto exit;
} else {
*ptr++ = c;
*ptr = '\0';
nread++;
}
size--;
}
exit:
unsetBlockingTimeout(conn);
return nread;
}
ConnectionType CT_TLS = {
.ae_handler = tlsEventHandler,
.accept = connTLSAccept,
.connect = connTLSConnect,
.blocking_connect = connTLSBlockingConnect,
.read = connTLSRead,
.write = connTLSWrite,
.close = connTLSClose,
.set_write_handler = connTLSSetWriteHandler,
.set_read_handler = connTLSSetReadHandler,
.get_last_error = connTLSGetLastError,
.sync_write = connTLSSyncWrite,
.sync_read = connTLSSyncRead,
.sync_readline = connTLSSyncReadLine,
};
int tlsHasPendingData() {
if (!pending_list)
return 0;
return listLength(pending_list) > 0;
}
void tlsProcessPendingData() {
listIter li;
listNode *ln;
listRewind(pending_list,&li);
while((ln = listNext(&li))) {
tls_connection *conn = listNodeValue(ln);
tlsHandleEvent(conn, AE_READABLE);
}
}
#else /* USE_OPENSSL */
void tlsInit(void) {
}
int tlsConfigure(redisTLSContextConfig *ctx_config) {
UNUSED(ctx_config);
return C_OK;
}
connection *connCreateTLS(void) {
return NULL;
}
connection *connCreateAcceptedTLS(int fd, int require_auth) {
UNUSED(fd);
UNUSED(require_auth);
return NULL;
}
int tlsHasPendingData() {
return 0;
}
void tlsProcessPendingData() {
}
#endif
/* tracking.c - Client side caching: keys tracking and invalidation
*
* Copyright (c) 2019, Salvatore Sanfilippo <antirez at gmail 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 "server.h"
/* The tracking table is constituted by a radix tree of keys, each pointing
* to a radix tree of client IDs, used to track the clients that may have
* certain keys in their local, client side, cache.
*
* When a client enables tracking with "CLIENT TRACKING on", each key served to
* the client is remembered in the table mapping the keys to the client IDs.
* Later, when a key is modified, all the clients that may have local copy
* of such key will receive an invalidation message.
*
* Clients will normally take frequently requested objects in memory, removing
* them when invalidation messages are received. */
rax *TrackingTable = NULL;
rax *PrefixTable = NULL;
uint64_t TrackingTableTotalItems = 0; /* Total number of IDs stored across
the whole tracking table. This gives
an hint about the total memory we
are using server side for CSC. */
robj *TrackingChannelName;
/* This is the structure that we have as value of the PrefixTable, and
* represents the list of keys modified, and the list of clients that need
* to be notified, for a given prefix. */
typedef struct bcastState {
rax *keys; /* Keys modified in the current event loop cycle. */
rax *clients; /* Clients subscribed to the notification events for this
prefix. */
} bcastState;
/* Remove the tracking state from the client 'c'. Note that there is not much
* to do for us here, if not to decrement the counter of the clients in
* tracking mode, because we just store the ID of the client in the tracking
* table, so we'll remove the ID reference in a lazy way. Otherwise when a
* client with many entries in the table is removed, it would cost a lot of
* time to do the cleanup. */
void disableTracking(client *c) {
/* If this client is in broadcasting mode, we need to unsubscribe it
* from all the prefixes it is registered to. */
if (c->flags & CLIENT_TRACKING_BCAST) {
raxIterator ri;
raxStart(&ri,c->client_tracking_prefixes);
raxSeek(&ri,"^",NULL,0);
while(raxNext(&ri)) {
bcastState *bs = raxFind(PrefixTable,ri.key,ri.key_len);
serverAssert(bs != raxNotFound);
raxRemove(bs->clients,(unsigned char*)&c,sizeof(c),NULL);
/* Was it the last client? Remove the prefix from the
* table. */
if (raxSize(bs->clients) == 0) {
raxFree(bs->clients);
raxFree(bs->keys);
zfree(bs);
raxRemove(PrefixTable,ri.key,ri.key_len,NULL);
}
}
raxStop(&ri);
raxFree(c->client_tracking_prefixes);
c->client_tracking_prefixes = NULL;
}
/* Clear flags and adjust the count. */
if (c->flags & CLIENT_TRACKING) {
server.tracking_clients--;
c->flags &= ~(CLIENT_TRACKING|CLIENT_TRACKING_BROKEN_REDIR|
CLIENT_TRACKING_BCAST);
}
}
/* Set the client 'c' to track the prefix 'prefix'. If the client 'c' is
* already registered for the specified prefix, no operation is performed. */
void enableBcastTrackingForPrefix(client *c, char *prefix, size_t plen) {
bcastState *bs = raxFind(PrefixTable,(unsigned char*)prefix,sdslen(prefix));
/* If this is the first client subscribing to such prefix, create
* the prefix in the table. */
if (bs == raxNotFound) {
bs = zmalloc(sizeof(*bs));
bs->keys = raxNew();
bs->clients = raxNew();
raxInsert(PrefixTable,(unsigned char*)prefix,plen,bs,NULL);
}
if (raxTryInsert(bs->clients,(unsigned char*)&c,sizeof(c),NULL,NULL)) {
if (c->client_tracking_prefixes == NULL)
c->client_tracking_prefixes = raxNew();
raxInsert(c->client_tracking_prefixes,
(unsigned char*)prefix,plen,NULL,NULL);
}
}
/* Enable the tracking state for the client 'c', and as a side effect allocates
* the tracking table if needed. If the 'redirect_to' argument is non zero, the
* invalidation messages for this client will be sent to the client ID
* specified by the 'redirect_to' argument. Note that if such client will
* eventually get freed, we'll send a message to the original client to
* inform it of the condition. Multiple clients can redirect the invalidation
* messages to the same client ID. */
void enableTracking(client *c, uint64_t redirect_to, int bcast, robj **prefix, size_t numprefix) {
if (!(c->flags & CLIENT_TRACKING)) server.tracking_clients++;
c->flags |= CLIENT_TRACKING;
c->flags &= ~(CLIENT_TRACKING_BROKEN_REDIR|CLIENT_TRACKING_BCAST);
c->client_tracking_redirection = redirect_to;
if (TrackingTable == NULL) {
TrackingTable = raxNew();
PrefixTable = raxNew();
TrackingChannelName = createStringObject("__redis__:invalidate",20);
}
if (bcast) {
c->flags |= CLIENT_TRACKING_BCAST;
if (numprefix == 0) enableBcastTrackingForPrefix(c,"",0);
for (size_t j = 0; j < numprefix; j++) {
sds sdsprefix = prefix[j]->ptr;
enableBcastTrackingForPrefix(c,sdsprefix,sdslen(sdsprefix));
}
}
}
/* This function is called after the execution of a readonly command in the
* case the client 'c' has keys tracking enabled. It will populate the
* tracking invalidation table according to the keys the user fetched, so that
* Redis will know what are the clients that should receive an invalidation
* message with certain groups of keys are modified. */
void trackingRememberKeys(client *c) {
int numkeys;
int *keys = getKeysFromCommand(c->cmd,c->argv,c->argc,&numkeys);
if (keys == NULL) return;
for(int j = 0; j < numkeys; j++) {
int idx = keys[j];
sds sdskey = c->argv[idx]->ptr;
rax *ids = raxFind(TrackingTable,(unsigned char*)sdskey,sdslen(sdskey));
if (ids == raxNotFound) {
ids = raxNew();
int inserted = raxTryInsert(TrackingTable,(unsigned char*)sdskey,
sdslen(sdskey),ids, NULL);
serverAssert(inserted == 1);
}
if (raxTryInsert(ids,(unsigned char*)&c->id,sizeof(c->id),NULL,NULL))
TrackingTableTotalItems++;
}
getKeysFreeResult(keys);
}
/* Given a key name, this function sends an invalidation message in the
* proper channel (depending on RESP version: PubSub or Push message) and
* to the proper client (in case fo redirection), in the context of the
* client 'c' with tracking enabled.
*
* In case the 'proto' argument is non zero, the function will assume that
* 'keyname' points to a buffer of 'keylen' bytes already expressed in the
* form of Redis RESP protocol, representing an array of keys to send
* to the client as value of the invalidation. This is used in BCAST mode
* in order to optimized the implementation to use less CPU time. */
void sendTrackingMessage(client *c, char *keyname, size_t keylen, int proto) {
int using_redirection = 0;
if (c->client_tracking_redirection) {
client *redir = lookupClientByID(c->client_tracking_redirection);
if (!redir) {
/* We need to signal to the original connection that we
* are unable to send invalidation messages to the redirected
* connection, because the client no longer exist. */
if (c->resp > 2) {
addReplyPushLen(c,3);
addReplyBulkCBuffer(c,"tracking-redir-broken",21);
addReplyLongLong(c,c->client_tracking_redirection);
}
return;
}
c = redir;
using_redirection = 1;
}
/* Only send such info for clients in RESP version 3 or more. However
* if redirection is active, and the connection we redirect to is
* in Pub/Sub mode, we can support the feature with RESP 2 as well,
* by sending Pub/Sub messages in the __redis__:invalidate channel. */
if (c->resp > 2) {
addReplyPushLen(c,2);
addReplyBulkCBuffer(c,"invalidate",10);
} else if (using_redirection && c->flags & CLIENT_PUBSUB) {
/* We use a static object to speedup things, however we assume
* that addReplyPubsubMessage() will not take a reference. */
addReplyPubsubMessage(c,TrackingChannelName,NULL);
} else {
/* If are here, the client is not using RESP3, nor is
* redirecting to another client. We can't send anything to
* it since RESP2 does not support push messages in the same
* connection. */
return;
}
/* Send the "value" part, which is the array of keys. */
if (proto) {
addReplyProto(c,keyname,keylen);
} else {
addReplyArrayLen(c,1);
addReplyBulkCBuffer(c,keyname,keylen);
}
}
/* This function is called when a key is modified in Redis and in the case
* we have at least one client with the BCAST mode enabled.
* Its goal is to set the key in the right broadcast state if the key
* matches one or more prefixes in the prefix table. Later when we
* return to the event loop, we'll send invalidation messages to the
* clients subscribed to each prefix. */
void trackingRememberKeyToBroadcast(char *keyname, size_t keylen) {
raxIterator ri;
raxStart(&ri,PrefixTable);
raxSeek(&ri,"^",NULL,0);
while(raxNext(&ri)) {
if (ri.key_len > keylen) continue;
if (ri.key_len != 0 && memcmp(ri.key,keyname,ri.key_len) != 0)
continue;
bcastState *bs = ri.data;
raxTryInsert(bs->keys,(unsigned char*)keyname,keylen,NULL,NULL);
}
raxStop(&ri);
}
/* This function is called from signalModifiedKey() or other places in Redis
* when a key changes value. In the context of keys tracking, our task here is
* to send a notification to every client that may have keys about such caching
* slot. */
void trackingInvalidateKey(robj *keyobj) {
if (TrackingTable == NULL) return;
sds sdskey = keyobj->ptr;
if (raxSize(PrefixTable) > 0)
trackingRememberKeyToBroadcast(sdskey,sdslen(sdskey));
rax *ids = raxFind(TrackingTable,(unsigned char*)sdskey,sdslen(sdskey));
if (ids == raxNotFound) return;;
raxIterator ri;
raxStart(&ri,ids);
raxSeek(&ri,"^",NULL,0);
while(raxNext(&ri)) {
uint64_t id;
memcpy(&id,ri.key,sizeof(id));
client *c = lookupClientByID(id);
/* Note that if the client is in BCAST mode, we don't want to
* send invalidation messages that were pending in the case
* previously the client was not in BCAST mode. This can happen if
* TRACKING is enabled normally, and then the client switches to
* BCAST mode. */
if (c == NULL ||
!(c->flags & CLIENT_TRACKING)||
c->flags & CLIENT_TRACKING_BCAST)
{
continue;
}
sendTrackingMessage(c,sdskey,sdslen(sdskey),0);
}
raxStop(&ri);
/* Free the tracking table: we'll create the radix tree and populate it
* again if more keys will be modified in this caching slot. */
TrackingTableTotalItems -= raxSize(ids);
raxFree(ids);
raxRemove(TrackingTable,(unsigned char*)sdskey,sdslen(sdskey),NULL);
}
/* This function is called when one or all the Redis databases are flushed
* (dbid == -1 in case of FLUSHALL). Caching keys are not specific for
* each DB but are global: currently what we do is send a special
* notification to clients with tracking enabled, invalidating the caching
* key "", which means, "all the keys", in order to avoid flooding clients
* with many invalidation messages for all the keys they may hold.
*/
void freeTrackingRadixTree(void *rt) {
raxFree(rt);
}
void trackingInvalidateKeysOnFlush(int dbid) {
if (server.tracking_clients) {
listNode *ln;
listIter li;
listRewind(server.clients,&li);
while ((ln = listNext(&li)) != NULL) {
client *c = listNodeValue(ln);
if (c->flags & CLIENT_TRACKING) {
sendTrackingMessage(c,"",1,0);
}
}
}
/* In case of FLUSHALL, reclaim all the memory used by tracking. */
if (dbid == -1 && TrackingTable) {
raxFreeWithCallback(TrackingTable,freeTrackingRadixTree);
TrackingTable = raxNew();
TrackingTableTotalItems = 0;
}
}
/* Tracking forces Redis to remember information about which client may have
* certain keys. In workloads where there are a lot of reads, but keys are
* hardly modified, the amount of information we have to remember server side
* could be a lot, with the number of keys being totally not bound.
*
* So Redis allows the user to configure a maximum number of keys for the
* invalidation table. This function makes sure that we don't go over the
* specified fill rate: if we are over, we can just evict informations about
* a random key, and send invalidation messages to clients like if the key was
* modified. */
void trackingLimitUsedSlots(void) {
static unsigned int timeout_counter = 0;
if (TrackingTable == NULL) return;
if (server.tracking_table_max_keys == 0) return; /* No limits set. */
size_t max_keys = server.tracking_table_max_keys;
if (raxSize(TrackingTable) <= max_keys) {
timeout_counter = 0;
return; /* Limit not reached. */
}
/* We have to invalidate a few keys to reach the limit again. The effort
* we do here is proportional to the number of times we entered this
* function and found that we are still over the limit. */
int effort = 100 * (timeout_counter+1);
/* We just remove one key after another by using a random walk. */
raxIterator ri;
raxStart(&ri,TrackingTable);
while(effort > 0) {
effort--;
raxSeek(&ri,"^",NULL,0);
raxRandomWalk(&ri,0);
rax *ids = ri.data;
TrackingTableTotalItems -= raxSize(ids);
raxFree(ids);
raxRemove(TrackingTable,ri.key,ri.key_len,NULL);
if (raxSize(TrackingTable) <= max_keys) {
timeout_counter = 0;
raxStop(&ri);
return; /* Return ASAP: we are again under the limit. */
}
}
/* If we reach this point, we were not able to go under the configured
* limit using the maximum effort we had for this run. */
raxStop(&ri);
timeout_counter++;
}
/* This function will run the prefixes of clients in BCAST mode and
* keys that were modified about each prefix, and will send the
* notifications to each client in each prefix. */
void trackingBroadcastInvalidationMessages(void) {
raxIterator ri, ri2;
/* Return ASAP if there is nothing to do here. */
if (TrackingTable == NULL || !server.tracking_clients) return;
raxStart(&ri,PrefixTable);
raxSeek(&ri,"^",NULL,0);
while(raxNext(&ri)) {
bcastState *bs = ri.data;
if (raxSize(bs->keys)) {
/* Create the array reply with the list of keys once, then send
* it to all the clients subscribed to this prefix. */
char buf[32];
size_t len = ll2string(buf,sizeof(buf),raxSize(bs->keys));
sds proto = sdsempty();
proto = sdsMakeRoomFor(proto,raxSize(bs->keys)*15);
proto = sdscatlen(proto,"*",1);
proto = sdscatlen(proto,buf,len);
proto = sdscatlen(proto,"\r\n",2);
raxStart(&ri2,bs->keys);
raxSeek(&ri2,"^",NULL,0);
while(raxNext(&ri2)) {
len = ll2string(buf,sizeof(buf),ri2.key_len);
proto = sdscatlen(proto,"$",1);
proto = sdscatlen(proto,buf,len);
proto = sdscatlen(proto,"\r\n",2);
proto = sdscatlen(proto,ri2.key,ri2.key_len);
proto = sdscatlen(proto,"\r\n",2);
}
raxStop(&ri2);
/* Send this array of keys to every client in the list. */
raxStart(&ri2,bs->clients);
raxSeek(&ri2,"^",NULL,0);
while(raxNext(&ri2)) {
client *c;
memcpy(&c,ri2.key,sizeof(c));
sendTrackingMessage(c,proto,sdslen(proto),1);
}
raxStop(&ri2);
/* Clean up: we can remove everything from this state, because we
* want to only track the new keys that will be accumulated starting
* from now. */
sdsfree(proto);
}
raxFree(bs->keys);
bs->keys = raxNew();
}
raxStop(&ri);
}
/* This is just used in order to access the amount of used slots in the
* tracking table. */
uint64_t trackingGetTotalItems(void) {
return TrackingTableTotalItems;
}
uint64_t trackingGetTotalKeys(void) {
if (TrackingTable == NULL) return 0;
return raxSize(TrackingTable);
}
...@@ -423,6 +423,26 @@ int string2ll(const char *s, size_t slen, long long *value) { ...@@ -423,6 +423,26 @@ int string2ll(const char *s, size_t slen, long long *value) {
return 1; return 1;
} }
/* Helper function to convert a string to an unsigned long long value.
* The function attempts to use the faster string2ll() function inside
* Redis: if it fails, strtoull() is used instead. The function returns
* 1 if the conversion happened successfully or 0 if the number is
* invalid or out of range. */
int string2ull(const char *s, unsigned long long *value) {
long long ll;
if (string2ll(s,strlen(s),&ll)) {
if (ll < 0) return 0; /* Negative values are out of range. */
*value = ll;
return 1;
}
errno = 0;
char *endptr = NULL;
*value = strtoull(s,&endptr,10);
if (errno == EINVAL || errno == ERANGE || !(*s != '\0' && *endptr == '\0'))
return 0; /* strtoull() failed. */
return 1; /* Conversion done! */
}
/* Convert a string into a long. Returns 1 if the string could be parsed into a /* Convert a string into a long. Returns 1 if the string could be parsed into a
* (non-overflowing) long, 0 otherwise. The value will be set to the parsed * (non-overflowing) long, 0 otherwise. The value will be set to the parsed
* value when appropriate. */ * value when appropriate. */
...@@ -447,17 +467,18 @@ int string2l(const char *s, size_t slen, long *lval) { ...@@ -447,17 +467,18 @@ int string2l(const char *s, size_t slen, long *lval) {
* a double: no spaces or other characters before or after the string * a double: no spaces or other characters before or after the string
* representing the number are accepted. */ * representing the number are accepted. */
int string2ld(const char *s, size_t slen, long double *dp) { int string2ld(const char *s, size_t slen, long double *dp) {
char buf[256]; char buf[MAX_LONG_DOUBLE_CHARS];
long double value; long double value;
char *eptr; char *eptr;
if (slen >= sizeof(buf)) return 0; if (slen == 0 || slen >= sizeof(buf)) return 0;
memcpy(buf,s,slen); memcpy(buf,s,slen);
buf[slen] = '\0'; buf[slen] = '\0';
errno = 0; errno = 0;
value = strtold(buf, &eptr); value = strtold(buf, &eptr);
if (isspace(buf[0]) || eptr[0] != '\0' || if (isspace(buf[0]) || eptr[0] != '\0' ||
(size_t)(eptr-buf) != slen ||
(errno == ERANGE && (errno == ERANGE &&
(value == HUGE_VAL || value == -HUGE_VAL || value == 0)) || (value == HUGE_VAL || value == -HUGE_VAL || value == 0)) ||
errno == EINVAL || errno == EINVAL ||
...@@ -468,6 +489,27 @@ int string2ld(const char *s, size_t slen, long double *dp) { ...@@ -468,6 +489,27 @@ int string2ld(const char *s, size_t slen, long double *dp) {
return 1; return 1;
} }
/* Convert a string into a double. Returns 1 if the string could be parsed
* into a (non-overflowing) double, 0 otherwise. The value will be set to
* the parsed value when appropriate.
*
* Note that this function demands that the string strictly represents
* a double: no spaces or other characters before or after the string
* representing the number are accepted. */
int string2d(const char *s, size_t slen, double *dp) {
errno = 0;
char *eptr;
*dp = strtod(s, &eptr);
if (slen == 0 ||
isspace(((const char*)s)[0]) ||
(size_t)(eptr-(char*)s) != slen ||
(errno == ERANGE &&
(*dp == HUGE_VAL || *dp == -HUGE_VAL || *dp == 0)) ||
isnan(*dp))
return 0;
return 1;
}
/* Convert a double to a string representation. Returns the number of bytes /* Convert a double to a string representation. Returns the number of bytes
* required. The representation should always be parsable by strtod(3). * required. The representation should always be parsable by strtod(3).
* This function does not support human-friendly formatting like ld2string * This function does not support human-friendly formatting like ld2string
...@@ -510,15 +552,17 @@ int d2string(char *buf, size_t len, double value) { ...@@ -510,15 +552,17 @@ int d2string(char *buf, size_t len, double value) {
return len; return len;
} }
/* Convert a long double into a string. If humanfriendly is non-zero /* Create a string object from a long double.
* it does not use exponential format and trims trailing zeroes at the end, * If mode is humanfriendly it does not use exponential format and trims trailing
* however this results in loss of precision. Otherwise exp format is used * zeroes at the end (may result in loss of precision).
* and the output of snprintf() is not modified. * If mode is default exp format is used and the output of snprintf()
* is not modified (may result in loss of precision).
* If mode is hex hexadecimal format is used (no loss of precision)
* *
* The function returns the length of the string or zero if there was not * The function returns the length of the string or zero if there was not
* enough buffer room to store it. */ * enough buffer room to store it. */
int ld2string(char *buf, size_t len, long double value, int humanfriendly) { int ld2string(char *buf, size_t len, long double value, ld2string_mode mode) {
size_t l; size_t l = 0;
if (isinf(value)) { if (isinf(value)) {
/* Libc in odd systems (Hi Solaris!) will format infinite in a /* Libc in odd systems (Hi Solaris!) will format infinite in a
...@@ -531,13 +575,23 @@ int ld2string(char *buf, size_t len, long double value, int humanfriendly) { ...@@ -531,13 +575,23 @@ int ld2string(char *buf, size_t len, long double value, int humanfriendly) {
memcpy(buf,"-inf",4); memcpy(buf,"-inf",4);
l = 4; l = 4;
} }
} else if (humanfriendly) { } else {
switch (mode) {
case LD_STR_AUTO:
l = snprintf(buf,len,"%.17Lg",value);
if (l+1 > len) return 0; /* No room. */
break;
case LD_STR_HEX:
l = snprintf(buf,len,"%La",value);
if (l+1 > len) return 0; /* No room. */
break;
case LD_STR_HUMAN:
/* We use 17 digits precision since with 128 bit floats that precision /* We use 17 digits precision since with 128 bit floats that precision
* after rounding is able to represent most small decimal numbers in a * after rounding is able to represent most small decimal numbers in a
* way that is "non surprising" for the user (that is, most small * way that is "non surprising" for the user (that is, most small
* decimal numbers will be represented in a way that when converted * decimal numbers will be represented in a way that when converted
* back into a string are exactly the same as what the user typed.) */ * back into a string are exactly the same as what the user typed.) */
l = snprintf(buf,len,"%.17Lf", value); l = snprintf(buf,len,"%.17Lf",value);
if (l+1 > len) return 0; /* No room. */ if (l+1 > len) return 0; /* No room. */
/* Now remove trailing zeroes after the '.' */ /* Now remove trailing zeroes after the '.' */
if (strchr(buf,'.') != NULL) { if (strchr(buf,'.') != NULL) {
...@@ -548,9 +602,9 @@ int ld2string(char *buf, size_t len, long double value, int humanfriendly) { ...@@ -548,9 +602,9 @@ int ld2string(char *buf, size_t len, long double value, int humanfriendly) {
} }
if (*p == '.') l--; if (*p == '.') l--;
} }
} else { break;
l = snprintf(buf,len,"%.17Lg", value); default: return 0; /* Invalid mode. */
if (l+1 > len) return 0; /* No room. */ }
} }
buf[l] = '\0'; buf[l] = '\0';
return l; return l;
......
...@@ -38,6 +38,13 @@ ...@@ -38,6 +38,13 @@
* This should be the size of the buffer given to ld2string */ * This should be the size of the buffer given to ld2string */
#define MAX_LONG_DOUBLE_CHARS 5*1024 #define MAX_LONG_DOUBLE_CHARS 5*1024
/* long double to string convertion options */
typedef enum {
LD_STR_AUTO, /* %.17Lg */
LD_STR_HUMAN, /* %.17Lf + Trimming of trailing zeros */
LD_STR_HEX /* %La */
} ld2string_mode;
int stringmatchlen(const char *p, int plen, const char *s, int slen, int nocase); int stringmatchlen(const char *p, int plen, const char *s, int slen, int nocase);
int stringmatch(const char *p, const char *s, int nocase); int stringmatch(const char *p, const char *s, int nocase);
int stringmatchlen_fuzz_test(void); int stringmatchlen_fuzz_test(void);
...@@ -46,10 +53,12 @@ uint32_t digits10(uint64_t v); ...@@ -46,10 +53,12 @@ uint32_t digits10(uint64_t v);
uint32_t sdigits10(int64_t v); uint32_t sdigits10(int64_t v);
int ll2string(char *s, size_t len, long long value); int ll2string(char *s, size_t len, long long value);
int string2ll(const char *s, size_t slen, long long *value); int string2ll(const char *s, size_t slen, long long *value);
int string2ull(const char *s, unsigned long long *value);
int string2l(const char *s, size_t slen, long *value); int string2l(const char *s, size_t slen, long *value);
int string2ld(const char *s, size_t slen, long double *dp); int string2ld(const char *s, size_t slen, long double *dp);
int string2d(const char *s, size_t slen, double *dp);
int d2string(char *buf, size_t len, double value); int d2string(char *buf, size_t len, double value);
int ld2string(char *buf, size_t len, long double value, int humanfriendly); int ld2string(char *buf, size_t len, long double value, ld2string_mode mode);
sds getAbsolutePath(char *filename); sds getAbsolutePath(char *filename);
unsigned long getTimeZone(void); unsigned long getTimeZone(void);
int pathIsBaseName(char *path); int pathIsBaseName(char *path);
......
...@@ -576,7 +576,7 @@ void zipEntry(unsigned char *p, zlentry *e) { ...@@ -576,7 +576,7 @@ void zipEntry(unsigned char *p, zlentry *e) {
/* Create a new empty ziplist. */ /* Create a new empty ziplist. */
unsigned char *ziplistNew(void) { unsigned char *ziplistNew(void) {
unsigned int bytes = ZIPLIST_HEADER_SIZE+1; unsigned int bytes = ZIPLIST_HEADER_SIZE+ZIPLIST_END_SIZE;
unsigned char *zl = zmalloc(bytes); unsigned char *zl = zmalloc(bytes);
ZIPLIST_BYTES(zl) = intrev32ifbe(bytes); ZIPLIST_BYTES(zl) = intrev32ifbe(bytes);
ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(ZIPLIST_HEADER_SIZE); ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(ZIPLIST_HEADER_SIZE);
......
...@@ -148,6 +148,10 @@ void *zrealloc(void *ptr, size_t size) { ...@@ -148,6 +148,10 @@ void *zrealloc(void *ptr, size_t size) {
size_t oldsize; size_t oldsize;
void *newptr; void *newptr;
if (size == 0 && ptr != NULL) {
zfree(ptr);
return NULL;
}
if (ptr == NULL) return zmalloc(size); if (ptr == NULL) return zmalloc(size);
#ifdef HAVE_MALLOC_SIZE #ifdef HAVE_MALLOC_SIZE
oldsize = zmalloc_size(ptr); oldsize = zmalloc_size(ptr);
...@@ -290,6 +294,26 @@ size_t zmalloc_get_rss(void) { ...@@ -290,6 +294,26 @@ size_t zmalloc_get_rss(void) {
return t_info.resident_size; return t_info.resident_size;
} }
#elif defined(__FreeBSD__)
#include <sys/types.h>
#include <sys/sysctl.h>
#include <sys/user.h>
#include <unistd.h>
size_t zmalloc_get_rss(void) {
struct kinfo_proc info;
size_t infolen = sizeof(info);
int mib[4];
mib[0] = CTL_KERN;
mib[1] = KERN_PROC;
mib[2] = KERN_PROC_PID;
mib[3] = getpid();
if (sysctl(mib, 4, &info, &infolen, NULL, 0) == 0)
return (size_t)info.ki_rssize;
return 0L;
}
#else #else
size_t zmalloc_get_rss(void) { size_t zmalloc_get_rss(void) {
/* If we can't get the RSS in an OS-specific way for this system just /* If we can't get the RSS in an OS-specific way for this system just
...@@ -302,6 +326,7 @@ size_t zmalloc_get_rss(void) { ...@@ -302,6 +326,7 @@ size_t zmalloc_get_rss(void) {
#endif #endif
#if defined(USE_JEMALLOC) #if defined(USE_JEMALLOC)
int zmalloc_get_allocator_info(size_t *allocated, int zmalloc_get_allocator_info(size_t *allocated,
size_t *active, size_t *active,
size_t *resident) { size_t *resident) {
...@@ -323,13 +348,52 @@ int zmalloc_get_allocator_info(size_t *allocated, ...@@ -323,13 +348,52 @@ int zmalloc_get_allocator_info(size_t *allocated,
je_mallctl("stats.allocated", allocated, &sz, NULL, 0); je_mallctl("stats.allocated", allocated, &sz, NULL, 0);
return 1; return 1;
} }
void set_jemalloc_bg_thread(int enable) {
/* let jemalloc do purging asynchronously, required when there's no traffic
* after flushdb */
char val = !!enable;
je_mallctl("background_thread", NULL, 0, &val, 1);
}
int jemalloc_purge() {
/* return all unused (reserved) pages to the OS */
char tmp[32];
unsigned narenas = 0;
size_t sz = sizeof(unsigned);
if (!je_mallctl("arenas.narenas", &narenas, &sz, NULL, 0)) {
sprintf(tmp, "arena.%d.purge", narenas);
if (!je_mallctl(tmp, NULL, 0, NULL, 0))
return 0;
}
return -1;
}
#else #else
int zmalloc_get_allocator_info(size_t *allocated, int zmalloc_get_allocator_info(size_t *allocated,
size_t *active, size_t *active,
size_t *resident) { size_t *resident) {
*allocated = *resident = *active = 0; *allocated = *resident = *active = 0;
return 1; return 1;
} }
void set_jemalloc_bg_thread(int enable) {
((void)(enable));
}
int jemalloc_purge() {
return 0;
}
#endif
#if defined(__APPLE__)
/* For proc_pidinfo() used later in zmalloc_get_smap_bytes_by_field().
* Note that this file cannot be included in zmalloc.h because it includes
* a Darwin queue.h file where there is a "LIST_HEAD" macro (!) defined
* conficting with Redis user code. */
#include <libproc.h>
#endif #endif
/* Get the sum of the specified field (converted form kb to bytes) in /* Get the sum of the specified field (converted form kb to bytes) in
...@@ -371,7 +435,28 @@ size_t zmalloc_get_smap_bytes_by_field(char *field, long pid) { ...@@ -371,7 +435,28 @@ size_t zmalloc_get_smap_bytes_by_field(char *field, long pid) {
return bytes; return bytes;
} }
#else #else
/* Get sum of the specified field from libproc api call.
* As there are per page value basis we need to convert
* them accordingly.
*
* Note that AnonHugePages is a no-op as THP feature
* is not supported in this platform
*/
size_t zmalloc_get_smap_bytes_by_field(char *field, long pid) { size_t zmalloc_get_smap_bytes_by_field(char *field, long pid) {
#if defined(__APPLE__)
struct proc_regioninfo pri;
if (proc_pidinfo(pid, PROC_PIDREGIONINFO, 0, &pri, PROC_PIDREGIONINFO_SIZE) ==
PROC_PIDREGIONINFO_SIZE) {
if (!strcmp(field, "Private_Dirty:")) {
return (size_t)pri.pri_pages_dirtied * 4096;
} else if (!strcmp(field, "Rss:")) {
return (size_t)pri.pri_pages_resident * 4096;
} else if (!strcmp(field, "AnonHugePages:")) {
return 0;
}
}
return 0;
#endif
((void) field); ((void) field);
((void) pid); ((void) pid);
return 0; return 0;
......
...@@ -86,6 +86,8 @@ size_t zmalloc_used_memory(void); ...@@ -86,6 +86,8 @@ size_t zmalloc_used_memory(void);
void zmalloc_set_oom_handler(void (*oom_handler)(size_t)); void zmalloc_set_oom_handler(void (*oom_handler)(size_t));
size_t zmalloc_get_rss(void); size_t zmalloc_get_rss(void);
int zmalloc_get_allocator_info(size_t *allocated, size_t *active, size_t *resident); int zmalloc_get_allocator_info(size_t *allocated, size_t *active, size_t *resident);
void set_jemalloc_bg_thread(int enable);
int jemalloc_purge();
size_t zmalloc_get_private_dirty(long pid); size_t zmalloc_get_private_dirty(long pid);
size_t zmalloc_get_smap_bytes_by_field(char *field, long pid); size_t zmalloc_get_smap_bytes_by_field(char *field, long pid);
size_t zmalloc_get_memory_size(void); size_t zmalloc_get_memory_size(void);
......
...@@ -8,6 +8,7 @@ source ../instances.tcl ...@@ -8,6 +8,7 @@ source ../instances.tcl
source ../../support/cluster.tcl ; # Redis Cluster client. source ../../support/cluster.tcl ; # Redis Cluster client.
set ::instances_count 20 ; # How many instances we use at max. set ::instances_count 20 ; # How many instances we use at max.
set ::tlsdir "../../tls"
proc main {} { proc main {} {
parse_options parse_options
......
...@@ -4,6 +4,7 @@ ...@@ -4,6 +4,7 @@
# are preseved across iterations. # are preseved across iterations.
source "../tests/includes/init-tests.tcl" source "../tests/includes/init-tests.tcl"
source "../../../tests/support/cli.tcl"
test "Create a 5 nodes cluster" { test "Create a 5 nodes cluster" {
create_cluster 5 5 create_cluster 5 5
...@@ -79,6 +80,7 @@ test "Cluster consistency during live resharding" { ...@@ -79,6 +80,7 @@ test "Cluster consistency during live resharding" {
--cluster-to $target \ --cluster-to $target \
--cluster-slots 100 \ --cluster-slots 100 \
--cluster-yes \ --cluster-yes \
{*}[rediscli_tls_config "../../../tests"] \
| [info nameofexecutable] \ | [info nameofexecutable] \
../tests/helpers/onlydots.tcl \ ../tests/helpers/onlydots.tcl \
&] 0] &] 0]
......
...@@ -5,6 +5,7 @@ ...@@ -5,6 +5,7 @@
# other masters have slaves. # other masters have slaves.
source "../tests/includes/init-tests.tcl" source "../tests/includes/init-tests.tcl"
source "../../../tests/support/cli.tcl"
# Create a cluster with 5 master and 15 slaves, to make sure there are no # Create a cluster with 5 master and 15 slaves, to make sure there are no
# empty masters and make rebalancing simpler to handle during the test. # empty masters and make rebalancing simpler to handle during the test.
...@@ -33,7 +34,9 @@ test "Resharding all the master #0 slots away from it" { ...@@ -33,7 +34,9 @@ test "Resharding all the master #0 slots away from it" {
set output [exec \ set output [exec \
../../../src/redis-cli --cluster rebalance \ ../../../src/redis-cli --cluster rebalance \
127.0.0.1:[get_instance_attrib redis 0 port] \ 127.0.0.1:[get_instance_attrib redis 0 port] \
{*}[rediscli_tls_config "../../../tests"] \
--cluster-weight ${master0_id}=0 >@ stdout ] --cluster-weight ${master0_id}=0 >@ stdout ]
} }
test "Master #0 should lose its replicas" { test "Master #0 should lose its replicas" {
...@@ -51,6 +54,7 @@ test "Resharding back some slot to master #0" { ...@@ -51,6 +54,7 @@ test "Resharding back some slot to master #0" {
set output [exec \ set output [exec \
../../../src/redis-cli --cluster rebalance \ ../../../src/redis-cli --cluster rebalance \
127.0.0.1:[get_instance_attrib redis 0 port] \ 127.0.0.1:[get_instance_attrib redis 0 port] \
{*}[rediscli_tls_config "../../../tests"] \
--cluster-weight ${master0_id}=.01 \ --cluster-weight ${master0_id}=.01 \
--cluster-use-empty-masters >@ stdout] --cluster-use-empty-masters >@ stdout]
} }
......
source tests/support/redis.tcl source tests/support/redis.tcl
source tests/support/util.tcl source tests/support/util.tcl
set ::tlsdir "tests/tls"
# This function sometimes writes sometimes blocking-reads from lists/sorted # This function sometimes writes sometimes blocking-reads from lists/sorted
# sets. There are multiple processes like this executing at the same time # sets. There are multiple processes like this executing at the same time
# so that we have some chance to trap some corner condition if there is # so that we have some chance to trap some corner condition if there is
...@@ -8,8 +10,8 @@ source tests/support/util.tcl ...@@ -8,8 +10,8 @@ source tests/support/util.tcl
# space to just a few elements, and balance the operations so that it is # space to just a few elements, and balance the operations so that it is
# unlikely that lists and zsets just get more data without ever causing # unlikely that lists and zsets just get more data without ever causing
# blocking. # blocking.
proc bg_block_op {host port db ops} { proc bg_block_op {host port db ops tls} {
set r [redis $host $port] set r [redis $host $port 0 $tls]
$r select $db $r select $db
for {set j 0} {$j < $ops} {incr j} { for {set j 0} {$j < $ops} {incr j} {
...@@ -49,4 +51,4 @@ proc bg_block_op {host port db ops} { ...@@ -49,4 +51,4 @@ proc bg_block_op {host port db ops} {
} }
} }
bg_block_op [lindex $argv 0] [lindex $argv 1] [lindex $argv 2] [lindex $argv 3] bg_block_op [lindex $argv 0] [lindex $argv 1] [lindex $argv 2] [lindex $argv 3] [lindex $argv 4]
source tests/support/redis.tcl source tests/support/redis.tcl
source tests/support/util.tcl source tests/support/util.tcl
proc bg_complex_data {host port db ops} { set ::tlsdir "tests/tls"
set r [redis $host $port]
proc bg_complex_data {host port db ops tls} {
set r [redis $host $port 0 $tls]
$r select $db $r select $db
createComplexDataset $r $ops createComplexDataset $r $ops
} }
bg_complex_data [lindex $argv 0] [lindex $argv 1] [lindex $argv 2] [lindex $argv 3] bg_complex_data [lindex $argv 0] [lindex $argv 1] [lindex $argv 2] [lindex $argv 3] [lindex $argv 4]
source tests/support/redis.tcl source tests/support/redis.tcl
proc gen_write_load {host port seconds} { set ::tlsdir "tests/tls"
proc gen_write_load {host port seconds tls} {
set start_time [clock seconds] set start_time [clock seconds]
set r [redis $host $port 1] set r [redis $host $port 1 $tls]
$r select 9 $r select 9
while 1 { while 1 {
$r set [expr rand()] [expr rand()] $r set [expr rand()] [expr rand()]
...@@ -12,4 +14,4 @@ proc gen_write_load {host port seconds} { ...@@ -12,4 +14,4 @@ proc gen_write_load {host port seconds} {
} }
} }
gen_write_load [lindex $argv 0] [lindex $argv 1] [lindex $argv 2] gen_write_load [lindex $argv 0] [lindex $argv 1] [lindex $argv 2] [lindex $argv 3]
...@@ -17,6 +17,7 @@ source ../support/test.tcl ...@@ -17,6 +17,7 @@ source ../support/test.tcl
set ::verbose 0 set ::verbose 0
set ::valgrind 0 set ::valgrind 0
set ::tls 0
set ::pause_on_error 0 set ::pause_on_error 0
set ::simulate_error 0 set ::simulate_error 0
set ::failed 0 set ::failed 0
...@@ -69,7 +70,19 @@ proc spawn_instance {type base_port count {conf {}}} { ...@@ -69,7 +70,19 @@ proc spawn_instance {type base_port count {conf {}}} {
# Write the instance config file. # Write the instance config file.
set cfgfile [file join $dirname $type.conf] set cfgfile [file join $dirname $type.conf]
set cfg [open $cfgfile w] set cfg [open $cfgfile w]
if {$::tls} {
puts $cfg "tls-port $port"
puts $cfg "tls-replication yes"
puts $cfg "tls-cluster yes"
puts $cfg "port 0"
puts $cfg [format "tls-cert-file %s/../../tls/redis.crt" [pwd]]
puts $cfg [format "tls-key-file %s/../../tls/redis.key" [pwd]]
puts $cfg [format "tls-dh-params-file %s/../../tls/redis.dh" [pwd]]
puts $cfg [format "tls-ca-cert-file %s/../../tls/ca.crt" [pwd]]
puts $cfg "loglevel debug"
} else {
puts $cfg "port $port" puts $cfg "port $port"
}
puts $cfg "dir ./$dirname" puts $cfg "dir ./$dirname"
puts $cfg "logfile log.txt" puts $cfg "logfile log.txt"
# Add additional config files # Add additional config files
...@@ -88,7 +101,7 @@ proc spawn_instance {type base_port count {conf {}}} { ...@@ -88,7 +101,7 @@ proc spawn_instance {type base_port count {conf {}}} {
} }
# Push the instance into the right list # Push the instance into the right list
set link [redis 127.0.0.1 $port] set link [redis 127.0.0.1 $port 0 $::tls]
$link reconnect 1 $link reconnect 1
lappend ::${type}_instances [list \ lappend ::${type}_instances [list \
pid $pid \ pid $pid \
...@@ -148,6 +161,13 @@ proc parse_options {} { ...@@ -148,6 +161,13 @@ proc parse_options {} {
set ::simulate_error 1 set ::simulate_error 1
} elseif {$opt eq {--valgrind}} { } elseif {$opt eq {--valgrind}} {
set ::valgrind 1 set ::valgrind 1
} elseif {$opt eq {--tls}} {
package require tls 1.6
::tls::init \
-cafile "$::tlsdir/ca.crt" \
-certfile "$::tlsdir/redis.crt" \
-keyfile "$::tlsdir/redis.key"
set ::tls 1
} elseif {$opt eq "--help"} { } elseif {$opt eq "--help"} {
puts "Hello, I'm sentinel.tcl and I run Sentinel unit tests." puts "Hello, I'm sentinel.tcl and I run Sentinel unit tests."
puts "\nOptions:" puts "\nOptions:"
...@@ -492,7 +512,7 @@ proc restart_instance {type id} { ...@@ -492,7 +512,7 @@ proc restart_instance {type id} {
} }
# Connect with it with a fresh link # Connect with it with a fresh link
set link [redis 127.0.0.1 $port] set link [redis 127.0.0.1 $port 0 $::tls]
$link reconnect 1 $link reconnect 1
set_instance_attrib $type $id link $link set_instance_attrib $type $id link $link
......
...@@ -13,8 +13,9 @@ tags {"aof"} { ...@@ -13,8 +13,9 @@ tags {"aof"} {
# cleaned after a child responsible for an AOF rewrite exited. This buffer # cleaned after a child responsible for an AOF rewrite exited. This buffer
# was subsequently appended to the new AOF, resulting in duplicate commands. # was subsequently appended to the new AOF, resulting in duplicate commands.
start_server_aof [list dir $server_path] { start_server_aof [list dir $server_path] {
set client [redis [srv host] [srv port]] set client [redis [srv host] [srv port] 0 $::tls]
set bench [open "|src/redis-benchmark -q -p [srv port] -c 20 -n 20000 incr foo" "r+"] set bench [open "|src/redis-benchmark -q -s [srv unixsocket] -c 20 -n 20000 incr foo" "r+"]
after 100 after 100
# Benchmark should be running by now: start background rewrite # Benchmark should be running by now: start background rewrite
...@@ -29,7 +30,7 @@ tags {"aof"} { ...@@ -29,7 +30,7 @@ tags {"aof"} {
# Restart server to replay AOF # Restart server to replay AOF
start_server_aof [list dir $server_path] { start_server_aof [list dir $server_path] {
set client [redis [srv host] [srv port]] set client [redis [srv host] [srv port] 0 $::tls]
assert_equal 20000 [$client get foo] assert_equal 20000 [$client get foo]
} }
} }
...@@ -52,7 +52,7 @@ tags {"aof"} { ...@@ -52,7 +52,7 @@ tags {"aof"} {
assert_equal 1 [is_alive $srv] assert_equal 1 [is_alive $srv]
} }
set client [redis [dict get $srv host] [dict get $srv port]] set client [redis [dict get $srv host] [dict get $srv port] 0 $::tls]
test "Truncated AOF loaded: we expect foo to be equal to 5" { test "Truncated AOF loaded: we expect foo to be equal to 5" {
assert {[$client get foo] eq "5"} assert {[$client get foo] eq "5"}
...@@ -69,7 +69,7 @@ tags {"aof"} { ...@@ -69,7 +69,7 @@ tags {"aof"} {
assert_equal 1 [is_alive $srv] assert_equal 1 [is_alive $srv]
} }
set client [redis [dict get $srv host] [dict get $srv port]] set client [redis [dict get $srv host] [dict get $srv port] 0 $::tls]
test "Truncated AOF loaded: we expect foo to be equal to 6 now" { test "Truncated AOF loaded: we expect foo to be equal to 6 now" {
assert {[$client get foo] eq "6"} assert {[$client get foo] eq "6"}
...@@ -170,7 +170,7 @@ tags {"aof"} { ...@@ -170,7 +170,7 @@ tags {"aof"} {
} }
test "Fixed AOF: Keyspace should contain values that were parseable" { test "Fixed AOF: Keyspace should contain values that were parseable" {
set client [redis [dict get $srv host] [dict get $srv port]] set client [redis [dict get $srv host] [dict get $srv port] 0 $::tls]
wait_for_condition 50 100 { wait_for_condition 50 100 {
[catch {$client ping} e] == 0 [catch {$client ping} e] == 0
} else { } else {
...@@ -194,7 +194,7 @@ tags {"aof"} { ...@@ -194,7 +194,7 @@ tags {"aof"} {
} }
test "AOF+SPOP: Set should have 1 member" { test "AOF+SPOP: Set should have 1 member" {
set client [redis [dict get $srv host] [dict get $srv port]] set client [redis [dict get $srv host] [dict get $srv port] 0 $::tls]
wait_for_condition 50 100 { wait_for_condition 50 100 {
[catch {$client ping} e] == 0 [catch {$client ping} e] == 0
} else { } else {
...@@ -218,7 +218,7 @@ tags {"aof"} { ...@@ -218,7 +218,7 @@ tags {"aof"} {
} }
test "AOF+SPOP: Set should have 1 member" { test "AOF+SPOP: Set should have 1 member" {
set client [redis [dict get $srv host] [dict get $srv port]] set client [redis [dict get $srv host] [dict get $srv port] 0 $::tls]
wait_for_condition 50 100 { wait_for_condition 50 100 {
[catch {$client ping} e] == 0 [catch {$client ping} e] == 0
} else { } else {
...@@ -241,7 +241,7 @@ tags {"aof"} { ...@@ -241,7 +241,7 @@ tags {"aof"} {
} }
test "AOF+EXPIRE: List should be empty" { test "AOF+EXPIRE: List should be empty" {
set client [redis [dict get $srv host] [dict get $srv port]] set client [redis [dict get $srv host] [dict get $srv port] 0 $::tls]
wait_for_condition 50 100 { wait_for_condition 50 100 {
[catch {$client ping} e] == 0 [catch {$client ping} e] == 0
} else { } else {
...@@ -257,4 +257,35 @@ tags {"aof"} { ...@@ -257,4 +257,35 @@ tags {"aof"} {
r expire x -1 r expire x -1
} }
} }
start_server {overrides {appendonly {yes} appendfilename {appendonly.aof} appendfsync always}} {
test {AOF fsync always barrier issue} {
set rd [redis_deferring_client]
# Set a sleep when aof is flushed, so that we have a chance to look
# at the aof size and detect if the response of an incr command
# arrives before the data was written (and hopefully fsynced)
# We create a big reply, which will hopefully not have room in the
# socket buffers, and will install a write handler, then we sleep
# a big and issue the incr command, hoping that the last portion of
# the output buffer write, and the processing of the incr will happen
# in the same event loop cycle.
# Since the socket buffers and timing are unpredictable, we fuzz this
# test with slightly different sizes and sleeps a few times.
for {set i 0} {$i < 10} {incr i} {
r debug aof-flush-sleep 0
r del x
r setrange x [expr {int(rand()*5000000)+10000000}] x
r debug aof-flush-sleep 500000
set aof [file join [lindex [r config get dir] 1] appendonly.aof]
set size1 [file size $aof]
$rd get x
after [expr {int(rand()*30)}]
$rd incr new_value
$rd read
$rd read
set size2 [file size $aof]
assert {$size1 != $size2}
}
}
}
} }
...@@ -2,9 +2,9 @@ ...@@ -2,9 +2,9 @@
# Unlike stream operations such operations are "pop" style, so they consume # Unlike stream operations such operations are "pop" style, so they consume
# the list or sorted set, and must be replicated correctly. # the list or sorted set, and must be replicated correctly.
proc start_bg_block_op {host port db ops} { proc start_bg_block_op {host port db ops tls} {
set tclsh [info nameofexecutable] set tclsh [info nameofexecutable]
exec $tclsh tests/helpers/bg_block_op.tcl $host $port $db $ops & exec $tclsh tests/helpers/bg_block_op.tcl $host $port $db $ops $tls &
} }
proc stop_bg_block_op {handle} { proc stop_bg_block_op {handle} {
...@@ -18,9 +18,9 @@ start_server {tags {"repl"}} { ...@@ -18,9 +18,9 @@ start_server {tags {"repl"}} {
set master_port [srv -1 port] set master_port [srv -1 port]
set slave [srv 0 client] set slave [srv 0 client]
set load_handle0 [start_bg_block_op $master_host $master_port 9 100000] set load_handle0 [start_bg_block_op $master_host $master_port 9 100000 $::tls]
set load_handle1 [start_bg_block_op $master_host $master_port 9 100000] set load_handle1 [start_bg_block_op $master_host $master_port 9 100000 $::tls]
set load_handle2 [start_bg_block_op $master_host $master_port 9 100000] set load_handle2 [start_bg_block_op $master_host $master_port 9 100000 $::tls]
test {First server should have role slave after SLAVEOF} { test {First server should have role slave after SLAVEOF} {
$slave slaveof $master_host $master_port $slave slaveof $master_host $master_port
......
...@@ -18,6 +18,7 @@ start_server {} { ...@@ -18,6 +18,7 @@ start_server {} {
set R($j) [srv [expr 0-$j] client] set R($j) [srv [expr 0-$j] client]
set R_host($j) [srv [expr 0-$j] host] set R_host($j) [srv [expr 0-$j] host]
set R_port($j) [srv [expr 0-$j] port] set R_port($j) [srv [expr 0-$j] port]
set R_unixsocket($j) [srv [expr 0-$j] unixsocket]
if {$debug_msg} {puts "Log file: [srv [expr 0-$j] stdout]"} if {$debug_msg} {puts "Log file: [srv [expr 0-$j] stdout]"}
} }
...@@ -36,7 +37,7 @@ start_server {} { ...@@ -36,7 +37,7 @@ start_server {} {
} }
set cycle_start_time [clock milliseconds] set cycle_start_time [clock milliseconds]
set bench_pid [exec src/redis-benchmark -p $R_port(0) -n 10000000 -r 1000 incr __rand_int__ > /dev/null &] set bench_pid [exec src/redis-benchmark -s $R_unixsocket(0) -n 10000000 -r 1000 incr __rand_int__ > /dev/null &]
while 1 { while 1 {
set elapsed [expr {[clock milliseconds]-$cycle_start_time}] set elapsed [expr {[clock milliseconds]-$cycle_start_time}]
if {$elapsed > $duration*1000} break if {$elapsed > $duration*1000} break
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment