Commit 1ac30300 authored by Guy Korland's avatar Guy Korland
Browse files

Merge branch 'unstable' of github.com:antirez/redis into unstable

parents c1455dc0 673c9d70
......@@ -402,7 +402,7 @@ void lrangeCommand(client *c) {
if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != C_OK) ||
(getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != C_OK)) return;
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp])) == NULL
if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptyarray)) == NULL
|| checkType(c,o,OBJ_LIST)) return;
llen = listTypeLength(o);
......@@ -414,7 +414,7 @@ void lrangeCommand(client *c) {
/* Invariant: start >= 0, so this test will be true when end < 0.
* The range is empty when start > end or start >= length. */
if (start > end || start >= llen) {
addReplyNull(c);
addReply(c,shared.emptyarray);
return;
}
if (end >= llen) end = llen-1;
......@@ -606,7 +606,7 @@ void rpoplpushCommand(client *c) {
* Blocking POP operations
*----------------------------------------------------------------------------*/
/* This is a helper function for handleClientsBlockedOnLists(). It's work
/* This is a helper function for handleClientsBlockedOnKeys(). It's work
* is to serve a specific client (receiver) that is blocked on 'key'
* in the context of the specified 'db', doing the following:
*
......@@ -617,7 +617,7 @@ void rpoplpushCommand(client *c) {
* the AOF and replication channel.
*
* The argument 'where' is LIST_TAIL or LIST_HEAD, and indicates if the
* 'value' element was popped fron the head (BLPOP) or tail (BRPOP) so that
* 'value' element was popped from the head (BLPOP) or tail (BRPOP) so that
* we can propagate the command properly.
*
* The function returns C_OK if we are able to serve the client, otherwise
......
......@@ -415,13 +415,13 @@ void spopWithCountCommand(client *c) {
/* Make sure a key with the name inputted exists, and that it's type is
* indeed a set. Otherwise, return nil */
if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp]))
if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.null[c->resp]))
== NULL || checkType(c,set,OBJ_SET)) return;
/* If count is zero, serve an empty multibulk ASAP to avoid special
/* If count is zero, serve an empty set ASAP to avoid special
* cases later. */
if (count == 0) {
addReplyNull(c);
addReply(c,shared.emptyset[c->resp]);
return;
}
......@@ -632,13 +632,13 @@ void srandmemberWithCountCommand(client *c) {
uniq = 0;
}
if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp]))
if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.emptyset[c->resp]))
== NULL || checkType(c,set,OBJ_SET)) return;
size = setTypeSize(set);
/* If count is zero, serve it ASAP to avoid special cases later. */
if (count == 0) {
addReplyNull(c);
addReply(c,shared.emptyset[c->resp]);
return;
}
......@@ -813,7 +813,7 @@ void sinterGenericCommand(client *c, robj **setkeys,
}
addReply(c,shared.czero);
} else {
addReplyNull(c);
addReply(c,shared.emptyset[c->resp]);
}
return;
}
......
......@@ -242,17 +242,17 @@ int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_
* the current node is full. */
if (lp != NULL) {
if (server.stream_node_max_bytes &&
lp_bytes > server.stream_node_max_bytes)
lp_bytes >= server.stream_node_max_bytes)
{
lp = NULL;
} else if (server.stream_node_max_entries) {
int64_t count = lpGetInteger(lpFirst(lp));
if (count > server.stream_node_max_entries) lp = NULL;
if (count >= server.stream_node_max_entries) lp = NULL;
}
}
int flags = STREAM_ITEM_FLAG_NONE;
if (lp == NULL || lp_bytes > server.stream_node_max_bytes) {
if (lp == NULL || lp_bytes >= server.stream_node_max_bytes) {
master_id = id;
streamEncodeID(rax_key,&id);
/* Create the listpack having the master entry ID and fields. */
......@@ -492,14 +492,14 @@ void streamIteratorStart(streamIterator *si, stream *s, streamID *start, streamI
streamEncodeID(si->start_key,start);
} else {
si->start_key[0] = 0;
si->start_key[0] = 0;
si->start_key[1] = 0;
}
if (end) {
streamEncodeID(si->end_key,end);
} else {
si->end_key[0] = UINT64_MAX;
si->end_key[0] = UINT64_MAX;
si->end_key[1] = UINT64_MAX;
}
/* Seek the correct node in the radix tree. */
......
......@@ -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
* becomes too long *before* executing zzlInsert. */
zobj->ptr = zzlInsert(zobj->ptr,ele,score);
if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries)
zsetConvert(zobj,OBJ_ENCODING_SKIPLIST);
if (sdslen(ele) > server.zset_max_ziplist_value)
if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries ||
sdslen(ele) > server.zset_max_ziplist_value)
zsetConvert(zobj,OBJ_ENCODING_SKIPLIST);
if (newscore) *newscore = score;
*flags |= ZADD_ADDED;
......@@ -2427,7 +2426,7 @@ void zrangeGenericCommand(client *c, int reverse) {
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;
/* Sanitize indexes. */
......@@ -2439,7 +2438,7 @@ void zrangeGenericCommand(client *c, int reverse) {
/* Invariant: start >= 0, so this test will be true when end < 0.
* The range is empty when start > end or start >= length. */
if (start > end || start >= llen) {
addReplyNull(c);
addReply(c,shared.emptyarray);
return;
}
if (end >= llen) end = llen-1;
......@@ -2575,7 +2574,7 @@ void genericZrangebyscoreCommand(client *c, int reverse) {
}
/* 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;
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
......@@ -2595,7 +2594,7 @@ void genericZrangebyscoreCommand(client *c, int reverse) {
/* No "first" element in the specified interval. */
if (eptr == NULL) {
addReplyNull(c);
addReply(c,shared.emptyarray);
return;
}
......@@ -2662,7 +2661,7 @@ void genericZrangebyscoreCommand(client *c, int reverse) {
/* No "first" element in the specified interval. */
if (ln == NULL) {
addReplyNull(c);
addReply(c,shared.emptyarray);
return;
}
......@@ -2920,7 +2919,7 @@ void genericZrangebylexCommand(client *c, int reverse) {
}
/* 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))
{
zslFreeLexRange(&range);
......@@ -2943,7 +2942,7 @@ void genericZrangebylexCommand(client *c, int reverse) {
/* No "first" element in the specified interval. */
if (eptr == NULL) {
addReplyNull(c);
addReply(c,shared.emptyarray);
zslFreeLexRange(&range);
return;
}
......@@ -3007,7 +3006,7 @@ void genericZrangebylexCommand(client *c, int reverse) {
/* No "first" element in the specified interval. */
if (ln == NULL) {
addReplyNull(c);
addReply(c,shared.emptyarray);
zslFreeLexRange(&range);
return;
}
......@@ -3161,7 +3160,7 @@ void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey
/* No candidate for zpopping, return empty. */
if (!zobj) {
addReplyNull(c);
addReply(c,shared.emptyarray);
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 2^24 radix trees (each tree, and the
* table itself, are allocated in a lazy way only when needed) tracking
* clients that may have certain keys in their local, client side, cache.
*
* Keys are grouped into 2^24 slots, in a way similar to Redis Cluster hash
* slots, however here the function we use is crc64, taking the least
* significant 24 bits of the output.
*
* When a client enables tracking with "CLIENT TRACKING on", each key served to
* the client is hashed to one of such slots, and Redis will remember what
* client may have keys about such slot. Later, when a key in a given slot is
* modified, all the clients that may have local copies of keys in that slot
* will receive an invalidation message. There is no distinction of database
* number: a single table is used.
*
* Clients will normally take frequently requested objects in memory, removing
* them when invalidation messages are received. A strategy clients may use is
* to just cache objects in a dictionary, associating to each cached object
* some incremental epoch, or just a timestamp. When invalidation messages are
* received clients may store, in a different table, the timestamp (or epoch)
* of the invalidation of such given slot: later when accessing objects, the
* eviction of stale objects may be performed in a lazy way by checking if the
* cached object timestamp is older than the invalidation timestamp for such
* objects.
*
* The output of the 24 bit hash function is very large (more than 16 million
* possible slots), so clients that may want to use less resources may only
* use the most significant bits instead of the full 24 bits. */
#define TRACKING_TABLE_SIZE (1<<24)
rax **TrackingTable = NULL;
unsigned long TrackingTableUsedSlots = 0;
robj *TrackingChannelName;
/* 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 (c->flags & CLIENT_TRACKING) {
server.tracking_clients--;
c->flags &= ~(CLIENT_TRACKING|CLIENT_TRACKING_BROKEN_REDIR);
}
}
/* 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) {
if (c->flags & CLIENT_TRACKING) return;
c->flags |= CLIENT_TRACKING;
c->flags &= ~CLIENT_TRACKING_BROKEN_REDIR;
c->client_tracking_redirection = redirect_to;
server.tracking_clients++;
if (TrackingTable == NULL) {
TrackingTable = zcalloc(sizeof(rax*) * TRACKING_TABLE_SIZE);
TrackingChannelName = createStringObject("__redis__:invalidate",20);
}
}
/* This function is called after the excution of a readonly command in the
* case the client 'c' has keys tracking enabled. It will populate the
* tracking ivalidation 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;
uint64_t hash = crc64(0,
(unsigned char*)sdskey,sdslen(sdskey))&(TRACKING_TABLE_SIZE-1);
if (TrackingTable[hash] == NULL) {
TrackingTable[hash] = raxNew();
TrackingTableUsedSlots++;
}
raxTryInsert(TrackingTable[hash],
(unsigned char*)&c->id,sizeof(c->id),NULL,NULL);
}
getKeysFreeResult(keys);
}
void sendTrackingMessage(client *c, long long hash) {
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);
addReplyLongLong(c,hash);
} else if (using_redirection && c->flags & CLIENT_PUBSUB) {
robj *msg = createStringObjectFromLongLong(hash);
addReplyPubsubMessage(c,TrackingChannelName,msg);
decrRefCount(msg);
}
}
/* Invalidates a caching slot: this is actually the low level implementation
* of the API that Redis calls externally, that is trackingInvalidateKey(). */
void trackingInvalidateSlot(uint64_t slot) {
if (TrackingTable == NULL || TrackingTable[slot] == NULL) return;
raxIterator ri;
raxStart(&ri,TrackingTable[slot]);
raxSeek(&ri,"^",NULL,0);
while(raxNext(&ri)) {
uint64_t id;
memcpy(&id,ri.key,ri.key_len);
client *c = lookupClientByID(id);
if (c == NULL || !(c->flags & CLIENT_TRACKING)) continue;
sendTrackingMessage(c,slot);
}
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. */
raxFree(TrackingTable[slot]);
TrackingTable[slot] = NULL;
TrackingTableUsedSlots--;
}
/* 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 || TrackingTableUsedSlots == 0) return;
sds sdskey = keyobj->ptr;
uint64_t hash = crc64(0,
(unsigned char*)sdskey,sdslen(sdskey))&(TRACKING_TABLE_SIZE-1);
trackingInvalidateSlot(hash);
}
/* This function is called when one or all the Redis databases are flushed
* (dbid == -1 in case of FLUSHALL). Caching slots are not specific for
* each DB but are global: currently what we do is sending a special
* notification to clients with tracking enabled, invalidating the caching
* slot "-1", which means, "all the keys", in order to avoid flooding clients
* with many invalidation messages for all the keys they may hold.
*
* However trying to flush the tracking table here is very costly:
* we need scanning 16 million caching slots in the table to check
* if they are used, this introduces a big delay. So what we do is to really
* flush the table in the case of FLUSHALL. When a FLUSHDB is called instead
* we just send the invalidation message to all the clients, but don't
* flush the table: it will slowly get garbage collected as more keys
* are modified in the used caching slots. */
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);
}
}
}
/* In case of FLUSHALL, reclaim all the memory used by tracking. */
if (dbid == -1 && TrackingTable) {
for (int j = 0; j < TRACKING_TABLE_SIZE && TrackingTableUsedSlots > 0; j++) {
if (TrackingTable[j] != NULL) {
raxFree(TrackingTable[j]);
TrackingTable[j] = NULL;
TrackingTableUsedSlots--;
}
}
/* If there are no clients with tracking enabled, we can even
* reclaim the memory used by the table itself. The code assumes
* the table is allocated only if there is at least one client alive
* with tracking enabled. */
if (server.tracking_clients == 0) {
zfree(TrackingTable);
TrackingTable = NULL;
}
}
}
/* Tracking forces Redis to remember information about which client may have
* keys about certian caching slots. 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: for each 16 millions of caching
* slots we may end with a radix tree containing many entries.
*
* So Redis allows the user to configure a maximum fill rate 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
* random caching slots, and send invalidation messages to clients like if
* the key was modified. */
void trackingLimitUsedSlots(void) {
static unsigned int timeout_counter = 0;
if (server.tracking_table_max_fill == 0) return; /* No limits set. */
unsigned int max_slots =
(TRACKING_TABLE_SIZE/100) * server.tracking_table_max_fill;
if (TrackingTableUsedSlots <= max_slots) {
timeout_counter = 0;
return; /* Limit not reached. */
}
/* We have to invalidate a few slots 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);
/* Let's start at a random position, and perform linear probing, in order
* to improve cache locality. However once we are able to find an used
* slot, jump again randomly, in order to avoid creating big holes in the
* table (that will make this funciton use more resourced later). */
while(effort > 0) {
unsigned int idx = rand() % TRACKING_TABLE_SIZE;
do {
effort--;
idx = (idx+1) % TRACKING_TABLE_SIZE;
if (TrackingTable[idx] != NULL) {
trackingInvalidateSlot(idx);
if (TrackingTableUsedSlots <= max_slots) {
timeout_counter = 0;
return; /* Return ASAP: we are again under the limit. */
} else {
break; /* Jump to next random position. */
}
}
} while(effort > 0);
}
timeout_counter++;
}
/* This is just used in order to access the amount of used slots in the
* tracking table. */
unsigned long long trackingGetUsedSlots(void) {
return TrackingTableUsedSlots;
}
......@@ -576,7 +576,7 @@ void zipEntry(unsigned char *p, zlentry *e) {
/* Create a new empty ziplist. */
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);
ZIPLIST_BYTES(zl) = intrev32ifbe(bytes);
ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(ZIPLIST_HEADER_SIZE);
......
......@@ -148,6 +148,10 @@ void *zrealloc(void *ptr, size_t size) {
size_t oldsize;
void *newptr;
if (size == 0 && ptr != NULL) {
zfree(ptr);
return NULL;
}
if (ptr == NULL) return zmalloc(size);
#ifdef HAVE_MALLOC_SIZE
oldsize = zmalloc_size(ptr);
......@@ -290,6 +294,26 @@ size_t zmalloc_get_rss(void) {
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
size_t zmalloc_get_rss(void) {
/* 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) {
#endif
#if defined(USE_JEMALLOC)
int zmalloc_get_allocator_info(size_t *allocated,
size_t *active,
size_t *resident) {
......@@ -323,13 +348,44 @@ int zmalloc_get_allocator_info(size_t *allocated,
je_mallctl("stats.allocated", allocated, &sz, NULL, 0);
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
int zmalloc_get_allocator_info(size_t *allocated,
size_t *active,
size_t *resident) {
*allocated = *resident = *active = 0;
return 1;
}
void set_jemalloc_bg_thread(int enable) {
((void)(enable));
}
int jemalloc_purge() {
return 0;
}
#endif
/* Get the sum of the specified field (converted form kb to bytes) in
......
......@@ -86,6 +86,8 @@ size_t zmalloc_used_memory(void);
void zmalloc_set_oom_handler(void (*oom_handler)(size_t));
size_t zmalloc_get_rss(void);
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_smap_bytes_by_field(char *field, long pid);
size_t zmalloc_get_memory_size(void);
......
......@@ -8,6 +8,7 @@ source ../instances.tcl
source ../../support/cluster.tcl ; # Redis Cluster client.
set ::instances_count 20 ; # How many instances we use at max.
set ::tlsdir "../../tls"
proc main {} {
parse_options
......
......@@ -4,6 +4,7 @@
# are preseved across iterations.
source "../tests/includes/init-tests.tcl"
source "../../../tests/support/cli.tcl"
test "Create a 5 nodes cluster" {
create_cluster 5 5
......@@ -79,6 +80,7 @@ test "Cluster consistency during live resharding" {
--cluster-to $target \
--cluster-slots 100 \
--cluster-yes \
{*}[rediscli_tls_config "../../../tests"] \
| [info nameofexecutable] \
../tests/helpers/onlydots.tcl \
&] 0]
......
......@@ -5,6 +5,7 @@
# other masters have slaves.
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
# 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" {
set output [exec \
../../../src/redis-cli --cluster rebalance \
127.0.0.1:[get_instance_attrib redis 0 port] \
{*}[rediscli_tls_config "../../../tests"] \
--cluster-weight ${master0_id}=0 >@ stdout ]
}
test "Master #0 should lose its replicas" {
......@@ -51,6 +54,7 @@ test "Resharding back some slot to master #0" {
set output [exec \
../../../src/redis-cli --cluster rebalance \
127.0.0.1:[get_instance_attrib redis 0 port] \
{*}[rediscli_tls_config "../../../tests"] \
--cluster-weight ${master0_id}=.01 \
--cluster-use-empty-masters >@ stdout]
}
......
source tests/support/redis.tcl
source tests/support/util.tcl
set ::tlsdir "tests/tls"
# This function sometimes writes sometimes blocking-reads from lists/sorted
# 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
......@@ -8,8 +10,8 @@ source tests/support/util.tcl
# 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
# blocking.
proc bg_block_op {host port db ops} {
set r [redis $host $port]
proc bg_block_op {host port db ops tls} {
set r [redis $host $port 0 $tls]
$r select $db
for {set j 0} {$j < $ops} {incr j} {
......@@ -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/util.tcl
proc bg_complex_data {host port db ops} {
set r [redis $host $port]
set ::tlsdir "tests/tls"
proc bg_complex_data {host port db ops tls} {
set r [redis $host $port 0 $tls]
$r select $db
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
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 r [redis $host $port 1]
set r [redis $host $port 0 $tls]
$r select 9
while 1 {
$r set [expr rand()] [expr rand()]
......@@ -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
set ::verbose 0
set ::valgrind 0
set ::tls 0
set ::pause_on_error 0
set ::simulate_error 0
set ::failed 0
......@@ -69,7 +70,19 @@ proc spawn_instance {type base_port count {conf {}}} {
# Write the instance config file.
set cfgfile [file join $dirname $type.conf]
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 "dir ./$dirname"
puts $cfg "logfile log.txt"
# Add additional config files
......@@ -88,7 +101,7 @@ proc spawn_instance {type base_port count {conf {}}} {
}
# 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
lappend ::${type}_instances [list \
pid $pid \
......@@ -148,6 +161,13 @@ proc parse_options {} {
set ::simulate_error 1
} elseif {$opt eq {--valgrind}} {
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"} {
puts "Hello, I'm sentinel.tcl and I run Sentinel unit tests."
puts "\nOptions:"
......@@ -492,7 +512,7 @@ proc restart_instance {type id} {
}
# 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
set_instance_attrib $type $id link $link
......
......@@ -13,8 +13,9 @@ tags {"aof"} {
# cleaned after a child responsible for an AOF rewrite exited. This buffer
# was subsequently appended to the new AOF, resulting in duplicate commands.
start_server_aof [list dir $server_path] {
set client [redis [srv host] [srv port]]
set bench [open "|src/redis-benchmark -q -p [srv port] -c 20 -n 20000 incr foo" "r+"]
set client [redis [srv host] [srv port] 0 $::tls]
set bench [open "|src/redis-benchmark -q -s [srv unixsocket] -c 20 -n 20000 incr foo" "r+"]
after 100
# Benchmark should be running by now: start background rewrite
......@@ -29,7 +30,7 @@ tags {"aof"} {
# Restart server to replay AOF
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]
}
}
......@@ -52,7 +52,7 @@ tags {"aof"} {
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" {
assert {[$client get foo] eq "5"}
......@@ -69,7 +69,7 @@ tags {"aof"} {
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" {
assert {[$client get foo] eq "6"}
......@@ -170,7 +170,7 @@ tags {"aof"} {
}
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 {
[catch {$client ping} e] == 0
} else {
......@@ -194,7 +194,7 @@ tags {"aof"} {
}
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 {
[catch {$client ping} e] == 0
} else {
......@@ -218,7 +218,7 @@ tags {"aof"} {
}
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 {
[catch {$client ping} e] == 0
} else {
......@@ -241,7 +241,7 @@ tags {"aof"} {
}
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 {
[catch {$client ping} e] == 0
} else {
......@@ -257,4 +257,35 @@ tags {"aof"} {
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 @@
# Unlike stream operations such operations are "pop" style, so they consume
# 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]
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} {
......@@ -18,9 +18,9 @@ start_server {tags {"repl"}} {
set master_port [srv -1 port]
set slave [srv 0 client]
set load_handle0 [start_bg_block_op $master_host $master_port 9 100000]
set load_handle1 [start_bg_block_op $master_host $master_port 9 100000]
set load_handle2 [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 $::tls]
set load_handle2 [start_bg_block_op $master_host $master_port 9 100000 $::tls]
test {First server should have role slave after SLAVEOF} {
$slave slaveof $master_host $master_port
......
......@@ -18,6 +18,7 @@ start_server {} {
set R($j) [srv [expr 0-$j] client]
set R_host($j) [srv [expr 0-$j] host]
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]"}
}
......@@ -36,7 +37,7 @@ start_server {} {
}
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 {
set elapsed [expr {[clock milliseconds]-$cycle_start_time}]
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