Unverified Commit 41d9eb02 authored by Oran Agra's avatar Oran Agra Committed by GitHub
Browse files

Merge: Fully abstract connection and make TLS dynamically loadable (#9320)

There are many commits in this PR, the detailed changes is described
in each commit message.

### Main changes in this PR

* Fully abstract connection type, and hide connection type specified methods.
  Ex, currently TLS class looks like:
```
static ConnectionType CT_TLS = {
    /* connection type */
    .get_type = connTLSGetType,

    /* connection type initialize & finalize & configure */
    .init = tlsInit,
    .cleanup = tlsCleanup,
    .configure = tlsConfigure,

    /* ae & accept & listen & error & address handler */
    .ae_handler = tlsEventHandler,
    .accept_handler = tlsAcceptHandler,
    .addr = connTLSAddr,
    .listen = connTLSListen,

    /* create/close connection */
    .conn_create = connCreateTLS,
    .conn_create_accepted = connCreateAcceptedTLS,
    .close = connTLSClose,

    /* connect & accept */
    .connect = connTLSConnect,
    .blocking_connect = connTLSBlockingConnect,
    .accept = connTLSAccept,

    /* IO */
    .read = connTLSRead,
    .write = connTLSWrite,
    .writev = connTLSWritev,
    .set_write_handler = connTLSSetWriteHandler,
    .set_read_handler = connTLSSetReadHandler,
    .get_last_error = connTLSGetLastError,
    .sync_write = connTLSSyncWrite,
    .sync_read = connTLSSyncRead,
    .sync_readline = connTLSSyncReadLine,

    /* pending data */
    .has_pending_data = tlsHasPendingData,
    .process_pending_data = tlsProcessPendingData,

    /* TLS specified methods */
    .get_peer_cert = connTLSGetPeerCert,
};

int RedisRegisterConnectionTypeTLS()
{
    return connTypeRegister(&CT_TLS);
}
```

* Also abstract Unix socket class. Currently, the connection framework becomes like:
```
                       uplayer
                          |
                   connection layer
                     /    |     \
                   TCP   Unix   TLS
    
```

* It's possible to build TLS as a shared library (`make BUILD_TLS=module`).
  Loading the shared library(redis-tls.so) into Redis by Redis module subsystem,
  and Redis starts to listen TLS port. Ex:
```
    ./src/redis-server --tls-port 6379 --port 0 \
        --tls-cert-file ./tests/tls/redis.crt \
        --tls-key-file ./tests/tls/redis.key \
        --tls-ca-cert-file ./tests/tls/ca.crt \
        --loadmodule src/redis-tls.so
```

### Interface changes
* RM_GetContextFlags supports a new flag: REDISMODULE_CTX_FLAGS_SERVER_STARTUP
* INFO SERVER includes a list of listeners:
```
listener0:name=tcp,bind=127.0.0.1,port=6380
listener1:name=unix,bind=/run/redis.sock
listener2:name=tls,bind=127.0.0.1,port=6379
```

### Other notes

* Fix wrong signature of RedisModuleDefragFunc, this could break
  compilation of a module, but not the ABI
* Some reordering of initialization order in server.c:
  * Move initialization of listeners to be after loading the modules
  * Config TLS after initialization of listeners
  * Init cluster after initialization of listeners
* Sentinel does not support the TLS module or any connection module
  since it uses hiredis for outbound connections, so when TLS is built as
  a module, sentinel lacks TLS support.
parents 90223759 4faddf18
...@@ -24,7 +24,8 @@ jobs: ...@@ -24,7 +24,8 @@ jobs:
steps: steps:
- uses: actions/checkout@v3 - uses: actions/checkout@v3
- name: make - name: make
run: make SANITIZER=address REDIS_CFLAGS='-Werror' # build with TLS module just for compilation coverage
run: make SANITIZER=address REDIS_CFLAGS='-Werror' BUILD_TLS=module
- name: testprep - name: testprep
run: sudo apt-get install tcl8.6 tclx -y run: sudo apt-get install tcl8.6 tclx -y
- name: test - name: test
......
...@@ -539,7 +539,7 @@ jobs: ...@@ -539,7 +539,7 @@ jobs:
run: | run: |
yum -y install centos-release-scl epel-release yum -y install centos-release-scl epel-release
yum -y install devtoolset-7 openssl-devel openssl yum -y install devtoolset-7 openssl-devel openssl
scl enable devtoolset-7 "make BUILD_TLS=yes REDIS_CFLAGS='-Werror'" scl enable devtoolset-7 "make BUILD_TLS=module REDIS_CFLAGS='-Werror'"
- name: testprep - name: testprep
run: | run: |
yum -y install tcl tcltls tclx yum -y install tcl tcltls tclx
...@@ -547,19 +547,19 @@ jobs: ...@@ -547,19 +547,19 @@ jobs:
- name: test - name: test
if: true && !contains(github.event.inputs.skiptests, 'redis') if: true && !contains(github.event.inputs.skiptests, 'redis')
run: | run: |
./runtest --accurate --verbose --dump-logs --tls --dump-logs ${{github.event.inputs.test_args}} ./runtest --accurate --verbose --dump-logs --tls-module --dump-logs ${{github.event.inputs.test_args}}
- name: module api test - name: module api test
if: true && !contains(github.event.inputs.skiptests, 'modules') if: true && !contains(github.event.inputs.skiptests, 'modules')
run: | run: |
./runtest-moduleapi --verbose --dump-logs --tls --dump-logs ${{github.event.inputs.test_args}} ./runtest-moduleapi --verbose --dump-logs --tls-module --dump-logs ${{github.event.inputs.test_args}}
- name: sentinel tests - name: sentinel tests
if: true && !contains(github.event.inputs.skiptests, 'sentinel') if: true && !contains(github.event.inputs.skiptests, 'sentinel')
run: | run: |
./runtest-sentinel --tls ${{github.event.inputs.cluster_test_args}} ./runtest-sentinel ${{github.event.inputs.cluster_test_args}}
- name: cluster tests - name: cluster tests
if: true && !contains(github.event.inputs.skiptests, 'cluster') if: true && !contains(github.event.inputs.skiptests, 'cluster')
run: | run: |
./runtest-cluster --tls ${{github.event.inputs.cluster_test_args}} ./runtest-cluster --tls-module ${{github.event.inputs.cluster_test_args}}
test-centos7-tls-no-tls: test-centos7-tls-no-tls:
runs-on: ubuntu-latest runs-on: ubuntu-latest
...@@ -582,7 +582,7 @@ jobs: ...@@ -582,7 +582,7 @@ jobs:
run: | run: |
yum -y install centos-release-scl epel-release yum -y install centos-release-scl epel-release
yum -y install devtoolset-7 openssl-devel openssl yum -y install devtoolset-7 openssl-devel openssl
scl enable devtoolset-7 "make BUILD_TLS=yes REDIS_CFLAGS='-Werror'" scl enable devtoolset-7 "make BUILD_TLS=module REDIS_CFLAGS='-Werror'"
- name: testprep - name: testprep
run: | run: |
yum -y install tcl tcltls tclx yum -y install tcl tcltls tclx
......
...@@ -9,8 +9,14 @@ Getting Started ...@@ -9,8 +9,14 @@ Getting Started
To build with TLS support you'll need OpenSSL development libraries (e.g. To build with TLS support you'll need OpenSSL development libraries (e.g.
libssl-dev on Debian/Ubuntu). libssl-dev on Debian/Ubuntu).
To build TLS support as Redis built-in:
Run `make BUILD_TLS=yes`. Run `make BUILD_TLS=yes`.
Or to build TLS as Redis module:
Run `make BUILD_TLS=module`.
Note that sentinel mode does not support TLS module.
### Tests ### Tests
To run Redis test suite with TLS, you'll need TLS support for TCL (i.e. To run Redis test suite with TLS, you'll need TLS support for TCL (i.e.
...@@ -22,16 +28,27 @@ To run Redis test suite with TLS, you'll need TLS support for TCL (i.e. ...@@ -22,16 +28,27 @@ To run Redis test suite with TLS, you'll need TLS support for TCL (i.e.
2. Run `./runtest --tls` or `./runtest-cluster --tls` to run Redis and Redis 2. Run `./runtest --tls` or `./runtest-cluster --tls` to run Redis and Redis
Cluster tests in TLS mode. Cluster tests in TLS mode.
3. Run `./runtest --tls-module` or `./runtest-cluster --tls-module` to
run Redis and Redis cluster tests in TLS mode with Redis module.
### Running manually ### Running manually
To manually run a Redis server with TLS mode (assuming `gen-test-certs.sh` was To manually run a Redis server with TLS mode (assuming `gen-test-certs.sh` was
invoked so sample certificates/keys are available): invoked so sample certificates/keys are available):
For TLS built-in mode:
./src/redis-server --tls-port 6379 --port 0 \ ./src/redis-server --tls-port 6379 --port 0 \
--tls-cert-file ./tests/tls/redis.crt \ --tls-cert-file ./tests/tls/redis.crt \
--tls-key-file ./tests/tls/redis.key \ --tls-key-file ./tests/tls/redis.key \
--tls-ca-cert-file ./tests/tls/ca.crt --tls-ca-cert-file ./tests/tls/ca.crt
For TLS module mode:
./src/redis-server --tls-port 6379 --port 0 \
--tls-cert-file ./tests/tls/redis.crt \
--tls-key-file ./tests/tls/redis.key \
--tls-ca-cert-file ./tests/tls/ca.crt \
--loadmodule src/redis-tls.so
To connect to this Redis server with `redis-cli`: To connect to this Redis server with `redis-cli`:
./src/redis-cli --tls \ ./src/redis-cli --tls \
......
...@@ -44,7 +44,7 @@ distclean: ...@@ -44,7 +44,7 @@ distclean:
.PHONY: distclean .PHONY: distclean
ifeq ($(BUILD_TLS),yes) ifneq (,$(filter $(BUILD_TLS),yes module))
HIREDIS_MAKE_FLAGS = USE_SSL=1 HIREDIS_MAKE_FLAGS = USE_SSL=1
endif endif
......
...@@ -267,24 +267,41 @@ ifeq ($(MALLOC),jemalloc) ...@@ -267,24 +267,41 @@ ifeq ($(MALLOC),jemalloc)
FINAL_LIBS := ../deps/jemalloc/lib/libjemalloc.a $(FINAL_LIBS) FINAL_LIBS := ../deps/jemalloc/lib/libjemalloc.a $(FINAL_LIBS)
endif endif
ifeq ($(BUILD_TLS),yes) # LIBSSL & LIBCRYPTO
FINAL_CFLAGS+=-DUSE_OPENSSL $(OPENSSL_CFLAGS) LIBSSL_LIBS=
FINAL_LDFLAGS+=$(OPENSSL_LDFLAGS) LIBSSL_PKGCONFIG := $(shell $(PKG_CONFIG) --exists libssl && echo $$?)
LIBSSL_PKGCONFIG := $(shell $(PKG_CONFIG) --exists libssl && echo $$?)
ifeq ($(LIBSSL_PKGCONFIG),0) ifeq ($(LIBSSL_PKGCONFIG),0)
LIBSSL_LIBS=$(shell $(PKG_CONFIG) --libs libssl) LIBSSL_LIBS=$(shell $(PKG_CONFIG) --libs libssl)
else else
LIBSSL_LIBS=-lssl LIBSSL_LIBS=-lssl
endif endif
LIBCRYPTO_PKGCONFIG := $(shell $(PKG_CONFIG) --exists libcrypto && echo $$?) LIBCRYPTO_LIBS=
LIBCRYPTO_PKGCONFIG := $(shell $(PKG_CONFIG) --exists libcrypto && echo $$?)
ifeq ($(LIBCRYPTO_PKGCONFIG),0) ifeq ($(LIBCRYPTO_PKGCONFIG),0)
LIBCRYPTO_LIBS=$(shell $(PKG_CONFIG) --libs libcrypto) LIBCRYPTO_LIBS=$(shell $(PKG_CONFIG) --libs libcrypto)
else else
LIBCRYPTO_LIBS=-lcrypto LIBCRYPTO_LIBS=-lcrypto
endif endif
BUILD_NO:=0
BUILD_YES:=1
BUILD_MODULE:=2
ifeq ($(BUILD_TLS),yes)
FINAL_CFLAGS+=-DUSE_OPENSSL=$(BUILD_YES) $(OPENSSL_CFLAGS) -DBUILD_TLS_MODULE=$(BUILD_NO)
FINAL_LDFLAGS+=$(OPENSSL_LDFLAGS)
FINAL_LIBS += ../deps/hiredis/libhiredis_ssl.a $(LIBSSL_LIBS) $(LIBCRYPTO_LIBS) FINAL_LIBS += ../deps/hiredis/libhiredis_ssl.a $(LIBSSL_LIBS) $(LIBCRYPTO_LIBS)
endif endif
TLS_MODULE=
TLS_MODULE_NAME:=redis-tls$(PROG_SUFFIX).so
TLS_MODULE_CFLAGS:=$(FINAL_CFLAGS)
ifeq ($(BUILD_TLS),module)
FINAL_CFLAGS+=-DUSE_OPENSSL=$(BUILD_MODULE) $(OPENSSL_CFLAGS)
TLS_CLIENT_LIBS = ../deps/hiredis/libhiredis_ssl.a $(LIBSSL_LIBS) $(LIBCRYPTO_LIBS)
TLS_MODULE=$(TLS_MODULE_NAME)
TLS_MODULE_CFLAGS+=-DUSE_OPENSSL=$(BUILD_MODULE) $(OPENSSL_CFLAGS) -DBUILD_TLS_MODULE=$(BUILD_MODULE)
endif
ifndef V ifndef V
define MAKE_INSTALL define MAKE_INSTALL
@printf ' %b %b\n' $(LINKCOLOR)INSTALL$(ENDCOLOR) $(BINCOLOR)$(1)$(ENDCOLOR) 1>&2 @printf ' %b %b\n' $(LINKCOLOR)INSTALL$(ENDCOLOR) $(BINCOLOR)$(1)$(ENDCOLOR) 1>&2
...@@ -316,7 +333,7 @@ endif ...@@ -316,7 +333,7 @@ endif
REDIS_SERVER_NAME=redis-server$(PROG_SUFFIX) REDIS_SERVER_NAME=redis-server$(PROG_SUFFIX)
REDIS_SENTINEL_NAME=redis-sentinel$(PROG_SUFFIX) REDIS_SENTINEL_NAME=redis-sentinel$(PROG_SUFFIX)
REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o eval.o bio.o rio.o rand.o memtest.o syscheck.o crcspeed.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o lolwut6.o acl.o tracking.o connection.o tls.o sha256.o timeout.o setcpuaffinity.o monotonic.o mt19937-64.o resp_parser.o call_reply.o script_lua.o script.o functions.o function_lua.o commands.o strl.o REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o eval.o bio.o rio.o rand.o memtest.o syscheck.o crcspeed.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o lolwut6.o acl.o tracking.o socket.o tls.o sha256.o timeout.o setcpuaffinity.o monotonic.o mt19937-64.o resp_parser.o call_reply.o script_lua.o script.o functions.o function_lua.o commands.o strl.o connection.o unix.o
REDIS_CLI_NAME=redis-cli$(PROG_SUFFIX) REDIS_CLI_NAME=redis-cli$(PROG_SUFFIX)
REDIS_CLI_OBJ=anet.o adlist.o dict.o redis-cli.o zmalloc.o release.o ae.o redisassert.o crcspeed.o crc64.o siphash.o crc16.o monotonic.o cli_common.o mt19937-64.o strl.o REDIS_CLI_OBJ=anet.o adlist.o dict.o redis-cli.o zmalloc.o release.o ae.o redisassert.o crcspeed.o crc64.o siphash.o crc16.o monotonic.o cli_common.o mt19937-64.o strl.o
REDIS_BENCHMARK_NAME=redis-benchmark$(PROG_SUFFIX) REDIS_BENCHMARK_NAME=redis-benchmark$(PROG_SUFFIX)
...@@ -325,7 +342,7 @@ REDIS_CHECK_RDB_NAME=redis-check-rdb$(PROG_SUFFIX) ...@@ -325,7 +342,7 @@ REDIS_CHECK_RDB_NAME=redis-check-rdb$(PROG_SUFFIX)
REDIS_CHECK_AOF_NAME=redis-check-aof$(PROG_SUFFIX) REDIS_CHECK_AOF_NAME=redis-check-aof$(PROG_SUFFIX)
ALL_SOURCES=$(sort $(patsubst %.o,%.c,$(REDIS_SERVER_OBJ) $(REDIS_CLI_OBJ) $(REDIS_BENCHMARK_OBJ))) ALL_SOURCES=$(sort $(patsubst %.o,%.c,$(REDIS_SERVER_OBJ) $(REDIS_CLI_OBJ) $(REDIS_BENCHMARK_OBJ)))
all: $(REDIS_SERVER_NAME) $(REDIS_SENTINEL_NAME) $(REDIS_CLI_NAME) $(REDIS_BENCHMARK_NAME) $(REDIS_CHECK_RDB_NAME) $(REDIS_CHECK_AOF_NAME) all: $(REDIS_SERVER_NAME) $(REDIS_SENTINEL_NAME) $(REDIS_CLI_NAME) $(REDIS_BENCHMARK_NAME) $(REDIS_CHECK_RDB_NAME) $(REDIS_CHECK_AOF_NAME) $(TLS_MODULE)
@echo "" @echo ""
@echo "Hint: It's a good idea to run 'make test' ;)" @echo "Hint: It's a good idea to run 'make test' ;)"
@echo "" @echo ""
...@@ -385,13 +402,17 @@ $(REDIS_CHECK_RDB_NAME): $(REDIS_SERVER_NAME) ...@@ -385,13 +402,17 @@ $(REDIS_CHECK_RDB_NAME): $(REDIS_SERVER_NAME)
$(REDIS_CHECK_AOF_NAME): $(REDIS_SERVER_NAME) $(REDIS_CHECK_AOF_NAME): $(REDIS_SERVER_NAME)
$(REDIS_INSTALL) $(REDIS_SERVER_NAME) $(REDIS_CHECK_AOF_NAME) $(REDIS_INSTALL) $(REDIS_SERVER_NAME) $(REDIS_CHECK_AOF_NAME)
# redis-tls.so
$(TLS_MODULE_NAME): $(REDIS_SERVER_NAME)
$(QUIET_CC)$(CC) -o $@ tls.c -shared -fPIC $(TLS_MODULE_CFLAGS) $(TLS_CLIENT_LIBS)
# redis-cli # redis-cli
$(REDIS_CLI_NAME): $(REDIS_CLI_OBJ) $(REDIS_CLI_NAME): $(REDIS_CLI_OBJ)
$(REDIS_LD) -o $@ $^ ../deps/hiredis/libhiredis.a ../deps/linenoise/linenoise.o $(FINAL_LIBS) $(REDIS_LD) -o $@ $^ ../deps/hiredis/libhiredis.a ../deps/linenoise/linenoise.o $(FINAL_LIBS) $(TLS_CLIENT_LIBS)
# redis-benchmark # redis-benchmark
$(REDIS_BENCHMARK_NAME): $(REDIS_BENCHMARK_OBJ) $(REDIS_BENCHMARK_NAME): $(REDIS_BENCHMARK_OBJ)
$(REDIS_LD) -o $@ $^ ../deps/hiredis/libhiredis.a ../deps/hdr_histogram/libhdrhistogram.a $(FINAL_LIBS) $(REDIS_LD) -o $@ $^ ../deps/hiredis/libhiredis.a ../deps/hdr_histogram/libhdrhistogram.a $(FINAL_LIBS) $(TLS_CLIENT_LIBS)
DEP = $(REDIS_SERVER_OBJ:%.o=%.d) $(REDIS_CLI_OBJ:%.o=%.d) $(REDIS_BENCHMARK_OBJ:%.o=%.d) DEP = $(REDIS_SERVER_OBJ:%.o=%.d) $(REDIS_CLI_OBJ:%.o=%.d) $(REDIS_BENCHMARK_OBJ:%.o=%.d)
-include $(DEP) -include $(DEP)
...@@ -410,7 +431,7 @@ commands.c: commands/*.json ../utils/generate-command-code.py ...@@ -410,7 +431,7 @@ commands.c: commands/*.json ../utils/generate-command-code.py
endif endif
clean: clean:
rm -rf $(REDIS_SERVER_NAME) $(REDIS_SENTINEL_NAME) $(REDIS_CLI_NAME) $(REDIS_BENCHMARK_NAME) $(REDIS_CHECK_RDB_NAME) $(REDIS_CHECK_AOF_NAME) *.o *.gcda *.gcno *.gcov redis.info lcov-html Makefile.dep rm -rf $(REDIS_SERVER_NAME) $(REDIS_SENTINEL_NAME) $(REDIS_CLI_NAME) $(REDIS_BENCHMARK_NAME) $(REDIS_CHECK_RDB_NAME) $(REDIS_CHECK_AOF_NAME) *.o *.gcda *.gcno *.gcov redis.info lcov-html Makefile.dep *.so
rm -f $(DEP) rm -f $(DEP)
.PHONY: clean .PHONY: clean
......
...@@ -62,6 +62,15 @@ static void anetSetError(char *err, const char *fmt, ...) ...@@ -62,6 +62,15 @@ static void anetSetError(char *err, const char *fmt, ...)
va_end(ap); va_end(ap);
} }
int anetGetError(int fd) {
int sockerr = 0;
socklen_t errlen = sizeof(sockerr);
if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &sockerr, &errlen) == -1)
sockerr = errno;
return sockerr;
}
int anetSetBlock(char *err, int fd, int non_block) { int anetSetBlock(char *err, int fd, int non_block) {
int flags; int flags;
...@@ -570,11 +579,11 @@ int anetUnixAccept(char *err, int s) { ...@@ -570,11 +579,11 @@ int anetUnixAccept(char *err, int s) {
return fd; return fd;
} }
int anetFdToString(int fd, char *ip, size_t ip_len, int *port, int fd_to_str_type) { int anetFdToString(int fd, char *ip, size_t ip_len, int *port, int remote) {
struct sockaddr_storage sa; struct sockaddr_storage sa;
socklen_t salen = sizeof(sa); socklen_t salen = sizeof(sa);
if (fd_to_str_type == FD_TO_PEER_NAME) { if (remote) {
if (getpeername(fd, (struct sockaddr *)&sa, &salen) == -1) goto error; if (getpeername(fd, (struct sockaddr *)&sa, &salen) == -1) goto error;
} else { } else {
if (getsockname(fd, (struct sockaddr *)&sa, &salen) == -1) goto error; if (getsockname(fd, (struct sockaddr *)&sa, &salen) == -1) goto error;
...@@ -618,23 +627,6 @@ error: ...@@ -618,23 +627,6 @@ error:
return -1; return -1;
} }
/* Format an IP,port pair into something easy to parse. If IP is IPv6
* (matches for ":"), the ip is surrounded by []. IP and port are just
* separated by colons. This the standard to display addresses within Redis. */
int anetFormatAddr(char *buf, size_t buf_len, char *ip, int port) {
return snprintf(buf,buf_len, strchr(ip,':') ?
"[%s]:%d" : "%s:%d", ip, port);
}
/* Like anetFormatAddr() but extract ip and port from the socket's peer/sockname. */
int anetFormatFdAddr(int fd, char *buf, size_t buf_len, int fd_to_str_type) {
char ip[INET6_ADDRSTRLEN];
int port;
anetFdToString(fd,ip,sizeof(ip),&port,fd_to_str_type);
return anetFormatAddr(buf, buf_len, ip, port);
}
/* Create a pipe buffer with given flags for read end and write end. /* Create a pipe buffer with given flags for read end and write end.
* Note that it supports the file flags defined by pipe2() and fcntl(F_SETFL), * Note that it supports the file flags defined by pipe2() and fcntl(F_SETFL),
* and one of the use cases is O_CLOEXEC|O_NONBLOCK. */ * and one of the use cases is O_CLOEXEC|O_NONBLOCK. */
......
...@@ -49,10 +49,6 @@ ...@@ -49,10 +49,6 @@
#undef ip_len #undef ip_len
#endif #endif
/* FD to address string conversion types */
#define FD_TO_PEER_NAME 0
#define FD_TO_SOCK_NAME 1
int anetTcpNonBlockConnect(char *err, const char *addr, int port); int anetTcpNonBlockConnect(char *err, const char *addr, int port);
int anetTcpNonBlockBestEffortBindConnect(char *err, const char *addr, int port, const char *source_addr); int anetTcpNonBlockBestEffortBindConnect(char *err, const char *addr, int port, const char *source_addr);
int anetResolve(char *err, char *host, char *ipbuf, size_t ipbuf_len, int flags); int anetResolve(char *err, char *host, char *ipbuf, size_t ipbuf_len, int flags);
...@@ -68,11 +64,11 @@ int anetEnableTcpNoDelay(char *err, int fd); ...@@ -68,11 +64,11 @@ int anetEnableTcpNoDelay(char *err, int fd);
int anetDisableTcpNoDelay(char *err, int fd); int anetDisableTcpNoDelay(char *err, int fd);
int anetSendTimeout(char *err, int fd, long long ms); int anetSendTimeout(char *err, int fd, long long ms);
int anetRecvTimeout(char *err, int fd, long long ms); int anetRecvTimeout(char *err, int fd, long long ms);
int anetFdToString(int fd, char *ip, size_t ip_len, int *port, int fd_to_str_type); int anetFdToString(int fd, char *ip, size_t ip_len, int *port, int remote);
int anetKeepAlive(char *err, int fd, int interval); int anetKeepAlive(char *err, int fd, int interval);
int anetFormatAddr(char *fmt, size_t fmt_len, char *ip, int port); int anetFormatAddr(char *fmt, size_t fmt_len, char *ip, int port);
int anetFormatFdAddr(int fd, char *buf, size_t buf_len, int fd_to_str_type);
int anetPipe(int fds[2], int read_flags, int write_flags); int anetPipe(int fds[2], int read_flags, int write_flags);
int anetSetSockMarkId(char *err, int fd, uint32_t id); int anetSetSockMarkId(char *err, int fd, uint32_t id);
int anetGetError(int fd);
#endif #endif
...@@ -119,6 +119,14 @@ dictType clusterNodesBlackListDictType = { ...@@ -119,6 +119,14 @@ dictType clusterNodesBlackListDictType = {
NULL /* allow to expand */ NULL /* allow to expand */
}; };
static ConnectionType *connTypeOfCluster() {
if (server.tls_cluster) {
return connectionTypeTls();
}
return connectionTypeTcp();
}
/* ----------------------------------------------------------------------------- /* -----------------------------------------------------------------------------
* Initialization * Initialization
* -------------------------------------------------------------------------- */ * -------------------------------------------------------------------------- */
...@@ -670,9 +678,6 @@ void clusterInit(void) { ...@@ -670,9 +678,6 @@ void clusterInit(void) {
} }
if (saveconf) clusterSaveConfigOrDie(1); if (saveconf) clusterSaveConfigOrDie(1);
/* We need a listening TCP port for our cluster messaging needs. */
server.cfd.count = 0;
/* Port sanity check II /* Port sanity check II
* The other handshake port check is triggered too late to stop * The other handshake port check is triggered too late to stop
* us from trying to use a too-high cluster port number. */ * us from trying to use a too-high cluster port number. */
...@@ -688,14 +693,25 @@ void clusterInit(void) { ...@@ -688,14 +693,25 @@ void clusterInit(void) {
serverLog(LL_WARNING, "No bind address is configured, but it is required for the Cluster bus."); serverLog(LL_WARNING, "No bind address is configured, but it is required for the Cluster bus.");
exit(1); exit(1);
} }
int cport = server.cluster_port ? server.cluster_port : port + CLUSTER_PORT_INCR;
if (listenToPort(cport, &server.cfd) == C_ERR ) { if (connectionIndexByType(connTypeOfCluster()->get_type(NULL)) < 0) {
serverLog(LL_WARNING, "Missing connection type %s, but it is required for the Cluster bus.", connTypeOfCluster()->get_type(NULL));
exit(1);
}
connListener *listener = &server.clistener;
listener->count = 0;
listener->bindaddr = server.bindaddr;
listener->bindaddr_count = server.bindaddr_count;
listener->port = server.cluster_port ? server.cluster_port : port + CLUSTER_PORT_INCR;
listener->ct = connTypeOfCluster();
if (connListen(listener) == C_ERR ) {
/* Note: the following log text is matched by the test suite. */ /* Note: the following log text is matched by the test suite. */
serverLog(LL_WARNING, "Failed listening on port %u (cluster), aborting.", cport); serverLog(LL_WARNING, "Failed listening on port %u (cluster), aborting.", listener->port);
exit(1); exit(1);
} }
if (createSocketAcceptHandler(&server.cfd, clusterAcceptHandler) != C_OK) { if (createSocketAcceptHandler(&server.clistener, clusterAcceptHandler) != C_OK) {
serverPanic("Unrecoverable error creating Redis Cluster socket accept handler."); serverPanic("Unrecoverable error creating Redis Cluster socket accept handler.");
} }
...@@ -865,6 +881,7 @@ void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -865,6 +881,7 @@ void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
int cport, cfd; int cport, cfd;
int max = MAX_CLUSTER_ACCEPTS_PER_CALL; int max = MAX_CLUSTER_ACCEPTS_PER_CALL;
char cip[NET_IP_STR_LEN]; char cip[NET_IP_STR_LEN];
int require_auth = TLS_CLIENT_AUTH_YES;
UNUSED(el); UNUSED(el);
UNUSED(mask); UNUSED(mask);
UNUSED(privdata); UNUSED(privdata);
...@@ -882,8 +899,7 @@ void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -882,8 +899,7 @@ void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
return; return;
} }
connection *conn = server.tls_cluster ? connection *conn = connCreateAccepted(connTypeOfCluster(), cfd, &require_auth);
connCreateAcceptedTLS(cfd, TLS_CLIENT_AUTH_YES) : connCreateAcceptedSocket(cfd);
/* Make sure connection is not in an error state */ /* Make sure connection is not in an error state */
if (connGetState(conn) != CONN_STATE_ACCEPTING) { if (connGetState(conn) != CONN_STATE_ACCEPTING) {
...@@ -1769,7 +1785,7 @@ int nodeIp2String(char *buf, clusterLink *link, char *announced_ip) { ...@@ -1769,7 +1785,7 @@ int nodeIp2String(char *buf, clusterLink *link, char *announced_ip) {
buf[NET_IP_STR_LEN-1] = '\0'; /* We are not sure the input is sane. */ buf[NET_IP_STR_LEN-1] = '\0'; /* We are not sure the input is sane. */
return C_OK; return C_OK;
} else { } else {
if (connPeerToString(link->conn, buf, NET_IP_STR_LEN, NULL) == C_ERR) { if (connAddrPeerName(link->conn, buf, NET_IP_STR_LEN, NULL) == C_ERR) {
serverLog(LL_NOTICE, "Error converting peer IP to string: %s", serverLog(LL_NOTICE, "Error converting peer IP to string: %s",
link->conn ? connGetLastError(link->conn) : "no link"); link->conn ? connGetLastError(link->conn) : "no link");
return C_ERR; return C_ERR;
...@@ -2273,7 +2289,7 @@ int clusterProcessPacket(clusterLink *link) { ...@@ -2273,7 +2289,7 @@ int clusterProcessPacket(clusterLink *link) {
{ {
char ip[NET_IP_STR_LEN]; char ip[NET_IP_STR_LEN];
if (connSockName(link->conn,ip,sizeof(ip),NULL) != -1 && if (connAddrSockName(link->conn,ip,sizeof(ip),NULL) != -1 &&
strcmp(ip,myself->ip)) strcmp(ip,myself->ip))
{ {
memcpy(myself->ip,ip,NET_IP_STR_LEN); memcpy(myself->ip,ip,NET_IP_STR_LEN);
...@@ -3969,7 +3985,7 @@ static int clusterNodeCronHandleReconnect(clusterNode *node, mstime_t handshake_ ...@@ -3969,7 +3985,7 @@ static int clusterNodeCronHandleReconnect(clusterNode *node, mstime_t handshake_
if (node->link == NULL) { if (node->link == NULL) {
clusterLink *link = createClusterLink(node); clusterLink *link = createClusterLink(node);
link->conn = server.tls_cluster ? connCreateTLS() : connCreateSocket(); link->conn = connCreate(connTypeOfCluster());
connSetPrivateData(link->conn, link); connSetPrivateData(link->conn, link);
if (connConnect(link->conn, node->ip, node->cport, server.bind_source_addr, if (connConnect(link->conn, node->ip, node->cport, server.bind_source_addr,
clusterLinkConnectHandler) == -1) { clusterLinkConnectHandler) == -1) {
...@@ -5022,7 +5038,7 @@ void addNodeToNodeReply(client *c, clusterNode *node) { ...@@ -5022,7 +5038,7 @@ void addNodeToNodeReply(client *c, clusterNode *node) {
/* Report non-TLS ports to non-TLS client in TLS cluster if available. */ /* Report non-TLS ports to non-TLS client in TLS cluster if available. */
int use_pport = (server.tls_cluster && int use_pport = (server.tls_cluster &&
c->conn && connGetType(c->conn) != CONN_TYPE_TLS); c->conn && (c->conn->type != connectionTypeTls()));
addReplyLongLong(c, use_pport && node->pport ? node->pport : node->port); addReplyLongLong(c, use_pport && node->pport ? node->pport : node->port);
addReplyBulkCBuffer(c, node->name, CLUSTER_NAMELEN); addReplyBulkCBuffer(c, node->name, CLUSTER_NAMELEN);
...@@ -5327,7 +5343,7 @@ NULL ...@@ -5327,7 +5343,7 @@ NULL
/* Report plaintext ports, only if cluster is TLS but client is known to /* Report plaintext ports, only if cluster is TLS but client is known to
* be non-TLS). */ * be non-TLS). */
int use_pport = (server.tls_cluster && int use_pport = (server.tls_cluster &&
c->conn && connGetType(c->conn) != CONN_TYPE_TLS); c->conn && (c->conn->type != connectionTypeTls()));
sds nodes = clusterGenNodesDescription(0, use_pport); sds nodes = clusterGenNodesDescription(0, use_pport);
addReplyVerbatim(c,nodes,sdslen(nodes),"txt"); addReplyVerbatim(c,nodes,sdslen(nodes),"txt");
sdsfree(nodes); sdsfree(nodes);
...@@ -5759,7 +5775,7 @@ NULL ...@@ -5759,7 +5775,7 @@ NULL
/* Use plaintext port if cluster is TLS but client is non-TLS. */ /* Use plaintext port if cluster is TLS but client is non-TLS. */
int use_pport = (server.tls_cluster && int use_pport = (server.tls_cluster &&
c->conn && connGetType(c->conn) != CONN_TYPE_TLS); c->conn && (c->conn->type != connectionTypeTls()));
addReplyArrayLen(c,n->numslaves); addReplyArrayLen(c,n->numslaves);
for (j = 0; j < n->numslaves; j++) { for (j = 0; j < n->numslaves; j++) {
sds ni = clusterGenNodeDescription(n->slaves[j], use_pport); sds ni = clusterGenNodeDescription(n->slaves[j], use_pport);
...@@ -6175,8 +6191,8 @@ migrateCachedSocket* migrateGetSocket(client *c, robj *host, robj *port, long ti ...@@ -6175,8 +6191,8 @@ migrateCachedSocket* migrateGetSocket(client *c, robj *host, robj *port, long ti
dictDelete(server.migrate_cached_sockets,dictGetKey(de)); dictDelete(server.migrate_cached_sockets,dictGetKey(de));
} }
/* Create the socket */ /* Create the connection */
conn = server.tls_cluster ? connCreateTLS() : connCreateSocket(); conn = connCreate(connTypeOfCluster());
if (connBlockingConnect(conn, host->ptr, atoi(port->ptr), timeout) if (connBlockingConnect(conn, host->ptr, atoi(port->ptr), timeout)
!= C_OK) { != C_OK) {
addReplyError(c,"-IOERR error or timeout connecting to the client"); addReplyError(c,"-IOERR error or timeout connecting to the client");
...@@ -6888,7 +6904,7 @@ void clusterRedirectClient(client *c, clusterNode *n, int hashslot, int error_co ...@@ -6888,7 +6904,7 @@ void clusterRedirectClient(client *c, clusterNode *n, int hashslot, int error_co
/* Redirect to IP:port. Include plaintext port if cluster is TLS but /* Redirect to IP:port. Include plaintext port if cluster is TLS but
* client is non-TLS. */ * client is non-TLS. */
int use_pport = (server.tls_cluster && int use_pport = (server.tls_cluster &&
c->conn && connGetType(c->conn) != CONN_TYPE_TLS); c->conn && (c->conn->type != connectionTypeTls()));
int port = use_pport && n->pport ? n->pport : n->port; int port = use_pport && n->pport ? n->pport : n->port;
addReplyErrorSds(c,sdscatprintf(sdsempty(), addReplyErrorSds(c,sdscatprintf(sdsempty(),
"-%s %d %s:%d", "-%s %d %s:%d",
......
...@@ -30,6 +30,7 @@ ...@@ -30,6 +30,7 @@
#include "server.h" #include "server.h"
#include "cluster.h" #include "cluster.h"
#include "connection.h"
#include <fcntl.h> #include <fcntl.h>
#include <sys/stat.h> #include <sys/stat.h>
...@@ -2429,7 +2430,14 @@ static int updateHZ(const char **err) { ...@@ -2429,7 +2430,14 @@ static int updateHZ(const char **err) {
} }
static int updatePort(const char **err) { static int updatePort(const char **err) {
if (changeListenPort(server.port, &server.ipfd, acceptTcpHandler) == C_ERR) { connListener *listener = listenerByType(CONN_TYPE_SOCKET);
serverAssert(listener != NULL);
listener->bindaddr = server.bindaddr;
listener->bindaddr_count = server.bindaddr_count;
listener->port = server.port;
listener->ct = connectionByType(CONN_TYPE_SOCKET);
if (changeListener(listener) == C_ERR) {
*err = "Unable to listen on this port. Check server logs."; *err = "Unable to listen on this port. Check server logs.";
return 0; return 0;
} }
...@@ -2544,12 +2552,36 @@ int updateRequirePass(const char **err) { ...@@ -2544,12 +2552,36 @@ int updateRequirePass(const char **err) {
return 1; return 1;
} }
/* applyBind affects both TCP and TLS (if enabled) together */
static int applyBind(const char **err) { static int applyBind(const char **err) {
if (changeBindAddr() == C_ERR) { connListener *tcp_listener = listenerByType(CONN_TYPE_SOCKET);
connListener *tls_listener = listenerByType(CONN_TYPE_TLS);
serverAssert(tcp_listener != NULL);
tcp_listener->bindaddr = server.bindaddr;
tcp_listener->bindaddr_count = server.bindaddr_count;
tcp_listener->port = server.port;
tcp_listener->ct = connectionByType(CONN_TYPE_SOCKET);
if (changeListener(tcp_listener) == C_ERR) {
*err = "Failed to bind to specified addresses."; *err = "Failed to bind to specified addresses.";
if (tls_listener)
closeListener(tls_listener); /* failed with TLS together */
return 0; return 0;
} }
if (server.tls_port != 0) {
serverAssert(tls_listener != NULL);
tls_listener->bindaddr = server.bindaddr;
tls_listener->bindaddr_count = server.bindaddr_count;
tls_listener->port = server.tls_port;
tls_listener->ct = connectionByType(CONN_TYPE_TLS);
if (changeListener(tls_listener) == C_ERR) {
*err = "Failed to bind to specified addresses.";
closeListener(tcp_listener); /* failed with TCP together */
return 0;
}
}
return 1; return 1;
} }
...@@ -2571,13 +2603,12 @@ int updateClusterHostname(const char **err) { ...@@ -2571,13 +2603,12 @@ int updateClusterHostname(const char **err) {
return 1; return 1;
} }
#ifdef USE_OPENSSL
static int applyTlsCfg(const char **err) { static int applyTlsCfg(const char **err) {
UNUSED(err); UNUSED(err);
/* If TLS is enabled, try to configure OpenSSL. */ /* If TLS is enabled, try to configure OpenSSL. */
if ((server.tls_port || server.tls_replication || server.tls_cluster) if ((server.tls_port || server.tls_replication || server.tls_cluster)
&& tlsConfigure(&server.tls_ctx_config) == C_ERR) { && connTypeConfigure(connectionTypeTls(), &server.tls_ctx_config, 1) == C_ERR) {
*err = "Unable to update TLS configuration. Check server logs."; *err = "Unable to update TLS configuration. Check server logs.";
return 0; return 0;
} }
...@@ -2586,12 +2617,18 @@ static int applyTlsCfg(const char **err) { ...@@ -2586,12 +2617,18 @@ static int applyTlsCfg(const char **err) {
static int applyTLSPort(const char **err) { static int applyTLSPort(const char **err) {
/* Configure TLS in case it wasn't enabled */ /* Configure TLS in case it wasn't enabled */
if (!isTlsConfigured() && tlsConfigure(&server.tls_ctx_config) == C_ERR) { if (connTypeConfigure(connectionTypeTls(), &server.tls_ctx_config, 0) == C_ERR) {
*err = "Unable to update TLS configuration. Check server logs."; *err = "Unable to update TLS configuration. Check server logs.";
return 0; return 0;
} }
if (changeListenPort(server.tls_port, &server.tlsfd, acceptTLSHandler) == C_ERR) { connListener *listener = listenerByType(CONN_TYPE_TLS);
serverAssert(listener != NULL);
listener->bindaddr = server.bindaddr;
listener->bindaddr_count = server.bindaddr_count;
listener->port = server.tls_port;
listener->ct = connectionByType(CONN_TYPE_TLS);
if (changeListener(listener) == C_ERR) {
*err = "Unable to listen on this port. Check server logs."; *err = "Unable to listen on this port. Check server logs.";
return 0; return 0;
} }
...@@ -2599,8 +2636,6 @@ static int applyTLSPort(const char **err) { ...@@ -2599,8 +2636,6 @@ static int applyTLSPort(const char **err) {
return 1; return 1;
} }
#endif /* USE_OPENSSL */
static int setConfigDirOption(standardConfig *config, sds *argv, int argc, const char **err) { static int setConfigDirOption(standardConfig *config, sds *argv, int argc, const char **err) {
UNUSED(config); UNUSED(config);
if (argc != 1) { if (argc != 1) {
...@@ -3109,7 +3144,6 @@ standardConfig static_configs[] = { ...@@ -3109,7 +3144,6 @@ standardConfig static_configs[] = {
createOffTConfig("auto-aof-rewrite-min-size", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, server.aof_rewrite_min_size, 64*1024*1024, MEMORY_CONFIG, NULL, NULL), createOffTConfig("auto-aof-rewrite-min-size", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, server.aof_rewrite_min_size, 64*1024*1024, MEMORY_CONFIG, NULL, NULL),
createOffTConfig("loading-process-events-interval-bytes", NULL, MODIFIABLE_CONFIG | HIDDEN_CONFIG, 1024, INT_MAX, server.loading_process_events_interval_bytes, 1024*1024*2, INTEGER_CONFIG, NULL, NULL), createOffTConfig("loading-process-events-interval-bytes", NULL, MODIFIABLE_CONFIG | HIDDEN_CONFIG, 1024, INT_MAX, server.loading_process_events_interval_bytes, 1024*1024*2, INTEGER_CONFIG, NULL, NULL),
#ifdef USE_OPENSSL
createIntConfig("tls-port", NULL, MODIFIABLE_CONFIG, 0, 65535, server.tls_port, 0, INTEGER_CONFIG, NULL, applyTLSPort), /* TCP port. */ createIntConfig("tls-port", NULL, MODIFIABLE_CONFIG, 0, 65535, server.tls_port, 0, INTEGER_CONFIG, NULL, applyTLSPort), /* TCP port. */
createIntConfig("tls-session-cache-size", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, server.tls_ctx_config.session_cache_size, 20*1024, INTEGER_CONFIG, NULL, applyTlsCfg), createIntConfig("tls-session-cache-size", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, server.tls_ctx_config.session_cache_size, 20*1024, INTEGER_CONFIG, NULL, applyTlsCfg),
createIntConfig("tls-session-cache-timeout", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, server.tls_ctx_config.session_cache_timeout, 300, INTEGER_CONFIG, NULL, applyTlsCfg), createIntConfig("tls-session-cache-timeout", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, server.tls_ctx_config.session_cache_timeout, 300, INTEGER_CONFIG, NULL, applyTlsCfg),
...@@ -3130,7 +3164,6 @@ standardConfig static_configs[] = { ...@@ -3130,7 +3164,6 @@ standardConfig static_configs[] = {
createStringConfig("tls-protocols", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, server.tls_ctx_config.protocols, NULL, NULL, applyTlsCfg), createStringConfig("tls-protocols", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, server.tls_ctx_config.protocols, NULL, NULL, applyTlsCfg),
createStringConfig("tls-ciphers", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, server.tls_ctx_config.ciphers, NULL, NULL, applyTlsCfg), createStringConfig("tls-ciphers", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, server.tls_ctx_config.ciphers, NULL, NULL, applyTlsCfg),
createStringConfig("tls-ciphersuites", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, server.tls_ctx_config.ciphersuites, NULL, NULL, applyTlsCfg), createStringConfig("tls-ciphersuites", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, server.tls_ctx_config.ciphersuites, NULL, NULL, applyTlsCfg),
#endif
/* Special configs */ /* Special configs */
createSpecialConfig("dir", NULL, MODIFIABLE_CONFIG | PROTECTED_CONFIG | DENY_LOADING_CONFIG, setConfigDirOption, getConfigDirOption, rewriteConfigDirOption, NULL), createSpecialConfig("dir", NULL, MODIFIABLE_CONFIG | PROTECTED_CONFIG | DENY_LOADING_CONFIG, setConfigDirOption, getConfigDirOption, rewriteConfigDirOption, NULL),
......
This diff is collapsed.
...@@ -32,12 +32,19 @@ ...@@ -32,12 +32,19 @@
#define __REDIS_CONNECTION_H #define __REDIS_CONNECTION_H
#include <errno.h> #include <errno.h>
#include <stdio.h>
#include <string.h>
#include <sys/uio.h> #include <sys/uio.h>
#include "ae.h"
#define CONN_INFO_LEN 32 #define CONN_INFO_LEN 32
#define CONN_ADDR_STR_LEN 128 /* Similar to INET6_ADDRSTRLEN, hoping to handle other protocols. */
#define MAX_ACCEPTS_PER_CALL 1000
struct aeEventLoop; struct aeEventLoop;
typedef struct connection connection; typedef struct connection connection;
typedef struct connListener connListener;
typedef enum { typedef enum {
CONN_STATE_NONE = 0, CONN_STATE_NONE = 0,
...@@ -51,27 +58,55 @@ typedef enum { ...@@ -51,27 +58,55 @@ typedef enum {
#define CONN_FLAG_CLOSE_SCHEDULED (1<<0) /* Closed scheduled by a handler */ #define CONN_FLAG_CLOSE_SCHEDULED (1<<0) /* Closed scheduled by a handler */
#define CONN_FLAG_WRITE_BARRIER (1<<1) /* Write barrier requested */ #define CONN_FLAG_WRITE_BARRIER (1<<1) /* Write barrier requested */
#define CONN_TYPE_SOCKET 1 #define CONN_TYPE_SOCKET "tcp"
#define CONN_TYPE_TLS 2 #define CONN_TYPE_UNIX "unix"
#define CONN_TYPE_TLS "tls"
#define CONN_TYPE_MAX 8 /* 8 is enough to be extendable */
typedef void (*ConnectionCallbackFunc)(struct connection *conn); typedef void (*ConnectionCallbackFunc)(struct connection *conn);
typedef struct ConnectionType { typedef struct ConnectionType {
/* connection type */
const char *(*get_type)(struct connection *conn);
/* connection type initialize & finalize & configure */
void (*init)(void); /* auto-call during register */
void (*cleanup)(void);
int (*configure)(void *priv, int reconfigure);
/* ae & accept & listen & error & address handler */
void (*ae_handler)(struct aeEventLoop *el, int fd, void *clientData, int mask); void (*ae_handler)(struct aeEventLoop *el, int fd, void *clientData, int mask);
aeFileProc *accept_handler;
int (*addr)(connection *conn, char *ip, size_t ip_len, int *port, int remote);
int (*listen)(connListener *listener);
/* create/close connection */
connection* (*conn_create)(void);
connection* (*conn_create_accepted)(int fd, void *priv);
void (*close)(struct connection *conn);
/* connect & accept */
int (*connect)(struct connection *conn, const char *addr, int port, const char *source_addr, ConnectionCallbackFunc connect_handler); int (*connect)(struct connection *conn, const char *addr, int port, const char *source_addr, ConnectionCallbackFunc connect_handler);
int (*blocking_connect)(struct connection *conn, const char *addr, int port, long long timeout);
int (*accept)(struct connection *conn, ConnectionCallbackFunc accept_handler);
/* IO */
int (*write)(struct connection *conn, const void *data, size_t data_len); int (*write)(struct connection *conn, const void *data, size_t data_len);
int (*writev)(struct connection *conn, const struct iovec *iov, int iovcnt); int (*writev)(struct connection *conn, const struct iovec *iov, int iovcnt);
int (*read)(struct connection *conn, void *buf, size_t buf_len); int (*read)(struct connection *conn, void *buf, size_t buf_len);
void (*close)(struct connection *conn);
int (*accept)(struct connection *conn, ConnectionCallbackFunc accept_handler);
int (*set_write_handler)(struct connection *conn, ConnectionCallbackFunc handler, int barrier); int (*set_write_handler)(struct connection *conn, ConnectionCallbackFunc handler, int barrier);
int (*set_read_handler)(struct connection *conn, ConnectionCallbackFunc handler); int (*set_read_handler)(struct connection *conn, ConnectionCallbackFunc handler);
const char *(*get_last_error)(struct connection *conn); const char *(*get_last_error)(struct connection *conn);
int (*blocking_connect)(struct connection *conn, const char *addr, int port, long long timeout);
ssize_t (*sync_write)(struct connection *conn, char *ptr, ssize_t size, long long timeout); ssize_t (*sync_write)(struct connection *conn, char *ptr, ssize_t size, long long timeout);
ssize_t (*sync_read)(struct connection *conn, char *ptr, ssize_t size, long long timeout); ssize_t (*sync_read)(struct connection *conn, char *ptr, ssize_t size, long long timeout);
ssize_t (*sync_readline)(struct connection *conn, char *ptr, ssize_t size, long long timeout); ssize_t (*sync_readline)(struct connection *conn, char *ptr, ssize_t size, long long timeout);
int (*get_type)(struct connection *conn);
/* pending data */
int (*has_pending_data)(void);
int (*process_pending_data)(void);
/* TLS specified methods */
sds (*get_peer_cert)(struct connection *conn);
} ConnectionType; } ConnectionType;
struct connection { struct connection {
...@@ -87,6 +122,19 @@ struct connection { ...@@ -87,6 +122,19 @@ struct connection {
int fd; int fd;
}; };
#define CONFIG_BINDADDR_MAX 16
/* Setup a listener by a connection type */
struct connListener {
int fd[CONFIG_BINDADDR_MAX];
int count;
char **bindaddr;
int bindaddr_count;
int port;
ConnectionType *ct;
void *priv; /* used by connection type specified data */
};
/* The connection module does not deal with listening and accepting sockets, /* The connection module does not deal with listening and accepting sockets,
* so we assume we have a socket when an incoming connection is created. * so we assume we have a socket when an incoming connection is created.
* *
...@@ -216,7 +264,7 @@ static inline ssize_t connSyncReadLine(connection *conn, char *ptr, ssize_t size ...@@ -216,7 +264,7 @@ static inline ssize_t connSyncReadLine(connection *conn, char *ptr, ssize_t size
} }
/* Return CONN_TYPE_* for the specified connection */ /* Return CONN_TYPE_* for the specified connection */
static inline int connGetType(connection *conn) { static inline const char *connGetType(connection *conn) {
return conn->type->get_type(conn); return conn->type->get_type(conn);
} }
...@@ -224,18 +272,77 @@ static inline int connLastErrorRetryable(connection *conn) { ...@@ -224,18 +272,77 @@ static inline int connLastErrorRetryable(connection *conn) {
return conn->last_errno == EINTR; return conn->last_errno == EINTR;
} }
connection *connCreateSocket(); /* Get address information of a connection.
connection *connCreateAcceptedSocket(int fd); * remote works as boolean type to get local/remote address */
static inline int connAddr(connection *conn, char *ip, size_t ip_len, int *port, int remote) {
if (conn && conn->type->addr) {
return conn->type->addr(conn, ip, ip_len, port, remote);
}
connection *connCreateTLS(); return -1;
connection *connCreateAcceptedTLS(int fd, int require_auth); }
void connSetPrivateData(connection *conn, void *data); /* Format an IP,port pair into something easy to parse. If IP is IPv6
void *connGetPrivateData(connection *conn); * (matches for ":"), the ip is surrounded by []. IP and port are just
int connGetState(connection *conn); * separated by colons. This the standard to display addresses within Redis. */
int connHasWriteHandler(connection *conn); static inline int formatAddr(char *buf, size_t buf_len, char *ip, int port) {
int connHasReadHandler(connection *conn); return snprintf(buf, buf_len, strchr(ip,':') ?
int connGetSocketError(connection *conn); "[%s]:%d" : "%s:%d", ip, port);
}
static inline int connFormatAddr(connection *conn, char *buf, size_t buf_len, int remote)
{
char ip[CONN_ADDR_STR_LEN];
int port;
if (connAddr(conn, ip, sizeof(ip), &port, remote) < 0) {
return -1;
}
return formatAddr(buf, buf_len, ip, port);
}
static inline int connAddrPeerName(connection *conn, char *ip, size_t ip_len, int *port) {
return connAddr(conn, ip, ip_len, port, 1);
}
static inline int connAddrSockName(connection *conn, char *ip, size_t ip_len, int *port) {
return connAddr(conn, ip, ip_len, port, 0);
}
static inline int connGetState(connection *conn) {
return conn->state;
}
/* Returns true if a write handler is registered */
static inline int connHasWriteHandler(connection *conn) {
return conn->write_handler != NULL;
}
/* Returns true if a read handler is registered */
static inline int connHasReadHandler(connection *conn) {
return conn->read_handler != NULL;
}
/* Associate a private data pointer with the connection */
static inline void connSetPrivateData(connection *conn, void *data) {
conn->private_data = data;
}
/* Get the associated private data pointer */
static inline void *connGetPrivateData(connection *conn) {
return conn->private_data;
}
/* Return a text that describes the connection, suitable for inclusion
* in CLIENT LIST and similar outputs.
*
* For sockets, we always return "fd=<fdnum>" to maintain compatibility.
*/
static inline const char *connGetInfo(connection *conn, char *buf, size_t buf_len) {
snprintf(buf, buf_len-1, "fd=%i", conn == NULL ? -1 : conn->fd);
return buf;
}
/* anet-style wrappers to conns */ /* anet-style wrappers to conns */
int connBlock(connection *conn); int connBlock(connection *conn);
...@@ -245,14 +352,81 @@ int connDisableTcpNoDelay(connection *conn); ...@@ -245,14 +352,81 @@ int connDisableTcpNoDelay(connection *conn);
int connKeepAlive(connection *conn, int interval); int connKeepAlive(connection *conn, int interval);
int connSendTimeout(connection *conn, long long ms); int connSendTimeout(connection *conn, long long ms);
int connRecvTimeout(connection *conn, long long ms); int connRecvTimeout(connection *conn, long long ms);
int connPeerToString(connection *conn, char *ip, size_t ip_len, int *port);
int connFormatFdAddr(connection *conn, char *buf, size_t buf_len, int fd_to_str_type); /* Get cert for the secure connection */
int connSockName(connection *conn, char *ip, size_t ip_len, int *port); static inline sds connGetPeerCert(connection *conn) {
const char *connGetInfo(connection *conn, char *buf, size_t buf_len); if (conn->type->get_peer_cert) {
return conn->type->get_peer_cert(conn);
/* Helpers for tls special considerations */ }
sds connTLSGetPeerCert(connection *conn);
int tlsHasPendingData(); return NULL;
int tlsProcessPendingData(); }
/* Initialize the redis connection framework */
int connTypeInitialize();
/* Register a connection type into redis connection framework */
int connTypeRegister(ConnectionType *ct);
/* Lookup a connection type by type name */
ConnectionType *connectionByType(const char *typename);
/* Fast path to get TCP connection type */
ConnectionType *connectionTypeTcp();
/* Fast path to get TLS connection type */
ConnectionType *connectionTypeTls();
/* Fast path to get Unix connection type */
ConnectionType *connectionTypeUnix();
/* Lookup the index of a connection type by type name, return -1 if not found */
int connectionIndexByType(const char *typename);
/* Create a connection of specified type */
static inline connection *connCreate(ConnectionType *ct) {
return ct->conn_create();
}
/* Create an accepted connection of specified type.
* priv is connection type specified argument */
static inline connection *connCreateAccepted(ConnectionType *ct, int fd, void *priv) {
return ct->conn_create_accepted(fd, priv);
}
/* Configure a connection type. A typical case is to configure TLS.
* priv is connection type specified,
* reconfigure is boolean type to specify if overwrite the original config */
static inline int connTypeConfigure(ConnectionType *ct, void *priv, int reconfigure) {
return ct->configure(priv, reconfigure);
}
/* Walk all the connection types and cleanup them all if possible */
void connTypeCleanupAll();
/* Test all the connection type has pending data or not. */
int connTypeHasPendingData(void);
/* walk all the connection types and process pending data for each connection type */
int connTypeProcessPendingData(void);
/* Listen on an initialized listener */
static inline int connListen(connListener *listener) {
return listener->ct->listen(listener);
}
/* Get accept_handler of a connection type */
static inline aeFileProc *connAcceptHandler(ConnectionType *ct) {
if (ct)
return ct->accept_handler;
return NULL;
}
/* Get Listeners information, note that caller should free the non-empty string */
sds getListensInfoString(sds info);
int RedisRegisterConnectionTypeSocket();
int RedisRegisterConnectionTypeUnix();
int RedisRegisterConnectionTypeTLS();
#endif /* __REDIS_CONNECTION_H */ #endif /* __REDIS_CONNECTION_H */
...@@ -11,4 +11,6 @@ test -f release.h || touch release.h ...@@ -11,4 +11,6 @@ test -f release.h || touch release.h
echo "#define REDIS_GIT_SHA1 \"$GIT_SHA1\"" > release.h echo "#define REDIS_GIT_SHA1 \"$GIT_SHA1\"" > release.h
echo "#define REDIS_GIT_DIRTY \"$GIT_DIRTY\"" >> release.h echo "#define REDIS_GIT_DIRTY \"$GIT_DIRTY\"" >> release.h
echo "#define REDIS_BUILD_ID \"$BUILD_ID\"" >> release.h echo "#define REDIS_BUILD_ID \"$BUILD_ID\"" >> release.h
echo "#include \"version.h\"" >> release.h
echo "#define REDIS_BUILD_ID_RAW REDIS_VERSION REDIS_BUILD_ID REDIS_GIT_DIRTY REDIS_GIT_SHA1" >> release.h
touch release.c # Force recompile of release.c touch release.c # Force recompile of release.c
...@@ -69,14 +69,14 @@ ...@@ -69,14 +69,14 @@
* pointers that have an API the module can call with them) * pointers that have an API the module can call with them)
* -------------------------------------------------------------------------- */ * -------------------------------------------------------------------------- */
   
typedef struct RedisModuleInfoCtx { struct RedisModuleInfoCtx {
struct RedisModule *module; struct RedisModule *module;
dict *requested_sections; dict *requested_sections;
sds info; /* info string we collected so far */ sds info; /* info string we collected so far */
int sections; /* number of sections we collected so far */ int sections; /* number of sections we collected so far */
int in_section; /* indication if we're in an active section or not */ int in_section; /* indication if we're in an active section or not */
int in_dict_field; /* indication that we're currently appending to a dict */ int in_dict_field; /* indication that we're currently appending to a dict */
} RedisModuleInfoCtx; };
   
/* This represents a shared API. Shared APIs will be used to populate /* This represents a shared API. Shared APIs will be used to populate
* the server.sharedapi dictionary, mapping names of APIs exported by * the server.sharedapi dictionary, mapping names of APIs exported by
...@@ -3327,11 +3327,11 @@ int modulePopulateClientInfoStructure(void *ci, client *client, int structver) { ...@@ -3327,11 +3327,11 @@ int modulePopulateClientInfoStructure(void *ci, client *client, int structver) {
ci1->flags |= REDISMODULE_CLIENTINFO_FLAG_TRACKING; ci1->flags |= REDISMODULE_CLIENTINFO_FLAG_TRACKING;
if (client->flags & CLIENT_BLOCKED) if (client->flags & CLIENT_BLOCKED)
ci1->flags |= REDISMODULE_CLIENTINFO_FLAG_BLOCKED; ci1->flags |= REDISMODULE_CLIENTINFO_FLAG_BLOCKED;
if (connGetType(client->conn) == CONN_TYPE_TLS) if (client->conn->type == connectionTypeTls())
ci1->flags |= REDISMODULE_CLIENTINFO_FLAG_SSL; ci1->flags |= REDISMODULE_CLIENTINFO_FLAG_SSL;
   
int port; int port;
connPeerToString(client->conn,ci1->addr,sizeof(ci1->addr),&port); connAddrPeerName(client->conn,ci1->addr,sizeof(ci1->addr),&port);
ci1->port = port; ci1->port = port;
ci1->db = client->db->id; ci1->db = client->db->id;
ci1->id = client->id; ci1->id = client->id;
...@@ -3529,6 +3529,8 @@ int RM_GetSelectedDb(RedisModuleCtx *ctx) { ...@@ -3529,6 +3529,8 @@ int RM_GetSelectedDb(RedisModuleCtx *ctx) {
* *
* * REDISMODULE_CTX_FLAGS_RESP3: Indicate the that client attached to this * * REDISMODULE_CTX_FLAGS_RESP3: Indicate the that client attached to this
* context is using RESP3. * context is using RESP3.
*
* * REDISMODULE_CTX_FLAGS_SERVER_STARTUP: The Redis instance is starting
*/ */
int RM_GetContextFlags(RedisModuleCtx *ctx) { int RM_GetContextFlags(RedisModuleCtx *ctx) {
int flags = 0; int flags = 0;
...@@ -3614,6 +3616,10 @@ int RM_GetContextFlags(RedisModuleCtx *ctx) { ...@@ -3614,6 +3616,10 @@ int RM_GetContextFlags(RedisModuleCtx *ctx) {
if (hasActiveChildProcess()) flags |= REDISMODULE_CTX_FLAGS_ACTIVE_CHILD; if (hasActiveChildProcess()) flags |= REDISMODULE_CTX_FLAGS_ACTIVE_CHILD;
if (server.in_fork_child) flags |= REDISMODULE_CTX_FLAGS_IS_CHILD; if (server.in_fork_child) flags |= REDISMODULE_CTX_FLAGS_IS_CHILD;
   
/* Non-empty server.loadmodule_queue means that Redis is starting. */
if (listLength(server.loadmodule_queue) > 0)
flags |= REDISMODULE_CTX_FLAGS_SERVER_STARTUP;
return flags; return flags;
} }
   
...@@ -8946,7 +8952,7 @@ RedisModuleString *RM_GetClientCertificate(RedisModuleCtx *ctx, uint64_t client_ ...@@ -8946,7 +8952,7 @@ RedisModuleString *RM_GetClientCertificate(RedisModuleCtx *ctx, uint64_t client_
client *c = lookupClientByID(client_id); client *c = lookupClientByID(client_id);
if (c == NULL) return NULL; if (c == NULL) return NULL;
   
sds cert = connTLSGetPeerCert(c->conn); sds cert = connGetPeerCert(c->conn);
if (!cert) return NULL; if (!cert) return NULL;
   
RedisModuleString *s = createObject(OBJ_STRING, cert); RedisModuleString *s = createObject(OBJ_STRING, cert);
...@@ -12205,13 +12211,13 @@ const char *RM_GetCurrentCommandName(RedisModuleCtx *ctx) { ...@@ -12205,13 +12211,13 @@ const char *RM_GetCurrentCommandName(RedisModuleCtx *ctx) {
/* The defrag context, used to manage state during calls to the data type /* The defrag context, used to manage state during calls to the data type
* defrag callback. * defrag callback.
*/ */
typedef struct RedisModuleDefragCtx { struct RedisModuleDefragCtx {
long defragged; long defragged;
long long int endtime; long long int endtime;
unsigned long *cursor; unsigned long *cursor;
struct redisObject *key; /* Optional name of key processed, NULL when unknown. */ struct redisObject *key; /* Optional name of key processed, NULL when unknown. */
int dbid; /* The dbid of the key being processed, -1 when unknown. */ int dbid; /* The dbid of the key being processed, -1 when unknown. */
} RedisModuleDefragCtx; };
   
/* Register a defrag callback for global data, i.e. anything that the module /* Register a defrag callback for global data, i.e. anything that the module
* may allocate that is not tied to a specific data type. * may allocate that is not tied to a specific data type.
......
...@@ -1212,7 +1212,7 @@ int islocalClient(client *c) { ...@@ -1212,7 +1212,7 @@ int islocalClient(client *c) {
/* tcp */ /* tcp */
char cip[NET_IP_STR_LEN+1] = { 0 }; char cip[NET_IP_STR_LEN+1] = { 0 };
connPeerToString(c->conn, cip, sizeof(cip)-1, NULL); connAddrPeerName(c->conn, cip, sizeof(cip)-1, NULL);
return !strcmp(cip,"127.0.0.1") || !strcmp(cip,"::1"); return !strcmp(cip,"127.0.0.1") || !strcmp(cip,"::1");
} }
...@@ -1271,8 +1271,7 @@ void clientAcceptHandler(connection *conn) { ...@@ -1271,8 +1271,7 @@ void clientAcceptHandler(connection *conn) {
c); c);
} }
#define MAX_ACCEPTS_PER_CALL 1000 void acceptCommonHandler(connection *conn, int flags, char *ip) {
static void acceptCommonHandler(connection *conn, int flags, char *ip) {
client *c; client *c;
char conninfo[100]; char conninfo[100];
UNUSED(ip); UNUSED(ip);
...@@ -1344,65 +1343,6 @@ static void acceptCommonHandler(connection *conn, int flags, char *ip) { ...@@ -1344,65 +1343,6 @@ static void acceptCommonHandler(connection *conn, int flags, char *ip) {
} }
} }
void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
int cport, cfd, max = MAX_ACCEPTS_PER_CALL;
char cip[NET_IP_STR_LEN];
UNUSED(el);
UNUSED(mask);
UNUSED(privdata);
while(max--) {
cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);
if (cfd == ANET_ERR) {
if (errno != EWOULDBLOCK)
serverLog(LL_WARNING,
"Accepting client connection: %s", server.neterr);
return;
}
serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport);
acceptCommonHandler(connCreateAcceptedSocket(cfd),0,cip);
}
}
void acceptTLSHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
int cport, cfd, max = MAX_ACCEPTS_PER_CALL;
char cip[NET_IP_STR_LEN];
UNUSED(el);
UNUSED(mask);
UNUSED(privdata);
while(max--) {
cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);
if (cfd == ANET_ERR) {
if (errno != EWOULDBLOCK)
serverLog(LL_WARNING,
"Accepting client connection: %s", server.neterr);
return;
}
serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport);
acceptCommonHandler(connCreateAcceptedTLS(cfd, server.tls_auth_clients),0,cip);
}
}
void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
int cfd, max = MAX_ACCEPTS_PER_CALL;
UNUSED(el);
UNUSED(mask);
UNUSED(privdata);
while(max--) {
cfd = anetUnixAccept(server.neterr, fd);
if (cfd == ANET_ERR) {
if (errno != EWOULDBLOCK)
serverLog(LL_WARNING,
"Accepting client connection: %s", server.neterr);
return;
}
serverLog(LL_VERBOSE,"Accepted connection to %s", server.unixsocket);
acceptCommonHandler(connCreateAcceptedSocket(cfd),CLIENT_UNIX_SOCKET,NULL);
}
}
void freeClientOriginalArgv(client *c) { void freeClientOriginalArgv(client *c) {
/* We didn't rewrite this client */ /* We didn't rewrite this client */
if (!c->original_argv) return; if (!c->original_argv) return;
...@@ -2716,13 +2656,13 @@ done: ...@@ -2716,13 +2656,13 @@ done:
* you want to relax error checking or need to display something anyway (see * you want to relax error checking or need to display something anyway (see
* anetFdToString implementation for more info). */ * anetFdToString implementation for more info). */
void genClientAddrString(client *client, char *addr, void genClientAddrString(client *client, char *addr,
size_t addr_len, int fd_to_str_type) { size_t addr_len, int remote) {
if (client->flags & CLIENT_UNIX_SOCKET) { if (client->flags & CLIENT_UNIX_SOCKET) {
/* Unix socket client. */ /* Unix socket client. */
snprintf(addr,addr_len,"%s:0",server.unixsocket); snprintf(addr,addr_len,"%s:0",server.unixsocket);
} else { } else {
/* TCP client. */ /* TCP client. */
connFormatFdAddr(client->conn,addr,addr_len,fd_to_str_type); connFormatAddr(client->conn,addr,addr_len,remote);
} }
} }
...@@ -2731,10 +2671,10 @@ void genClientAddrString(client *client, char *addr, ...@@ -2731,10 +2671,10 @@ void genClientAddrString(client *client, char *addr,
* The Peer ID never changes during the life of the client, however it * The Peer ID never changes during the life of the client, however it
* is expensive to compute. */ * is expensive to compute. */
char *getClientPeerId(client *c) { char *getClientPeerId(client *c) {
char peerid[NET_ADDR_STR_LEN]; char peerid[NET_ADDR_STR_LEN] = {0};
if (c->peerid == NULL) { if (c->peerid == NULL) {
genClientAddrString(c,peerid,sizeof(peerid),FD_TO_PEER_NAME); genClientAddrString(c,peerid,sizeof(peerid),1);
c->peerid = sdsnew(peerid); c->peerid = sdsnew(peerid);
} }
return c->peerid; return c->peerid;
...@@ -2745,10 +2685,10 @@ char *getClientPeerId(client *c) { ...@@ -2745,10 +2685,10 @@ char *getClientPeerId(client *c) {
* The Socket Name never changes during the life of the client, however it * The Socket Name never changes during the life of the client, however it
* is expensive to compute. */ * is expensive to compute. */
char *getClientSockname(client *c) { char *getClientSockname(client *c) {
char sockname[NET_ADDR_STR_LEN]; char sockname[NET_ADDR_STR_LEN] = {0};
if (c->sockname == NULL) { if (c->sockname == NULL) {
genClientAddrString(c,sockname,sizeof(sockname),FD_TO_SOCK_NAME); genClientAddrString(c,sockname,sizeof(sockname),0);
c->sockname = sdsnew(sockname); c->sockname = sdsnew(sockname);
} }
return c->sockname; return c->sockname;
......
...@@ -61,6 +61,7 @@ ...@@ -61,6 +61,7 @@
#include "help.h" /* Used for backwards-compatibility with pre-7.0 servers that don't support COMMAND DOCS. */ #include "help.h" /* Used for backwards-compatibility with pre-7.0 servers that don't support COMMAND DOCS. */
#include "anet.h" #include "anet.h"
#include "ae.h" #include "ae.h"
#include "connection.h"
#include "cli_common.h" #include "cli_common.h"
#include "mt19937-64.h" #include "mt19937-64.h"
...@@ -322,7 +323,7 @@ static void cliRefreshPrompt(void) { ...@@ -322,7 +323,7 @@ static void cliRefreshPrompt(void) {
prompt = sdscatfmt(prompt,"redis %s",config.hostsocket); prompt = sdscatfmt(prompt,"redis %s",config.hostsocket);
} else { } else {
char addr[256]; char addr[256];
anetFormatAddr(addr, sizeof(addr), config.conn_info.hostip, config.conn_info.hostport); formatAddr(addr, sizeof(addr), config.conn_info.hostip, config.conn_info.hostport);
prompt = sdscatlen(prompt,addr,strlen(addr)); prompt = sdscatlen(prompt,addr,strlen(addr));
} }
......
...@@ -162,11 +162,13 @@ typedef struct RedisModuleStreamID { ...@@ -162,11 +162,13 @@ typedef struct RedisModuleStreamID {
#define REDISMODULE_CTX_FLAGS_RESP3 (1<<22) #define REDISMODULE_CTX_FLAGS_RESP3 (1<<22)
/* Redis is currently async loading database for diskless replication. */ /* Redis is currently async loading database for diskless replication. */
#define REDISMODULE_CTX_FLAGS_ASYNC_LOADING (1<<23) #define REDISMODULE_CTX_FLAGS_ASYNC_LOADING (1<<23)
/* Redis is starting. */
#define REDISMODULE_CTX_FLAGS_SERVER_STARTUP (1<<24)
/* Next context flag, must be updated when adding new flags above! /* Next context flag, must be updated when adding new flags above!
This flag should not be used directly by the module. This flag should not be used directly by the module.
* Use RedisModule_GetContextFlagsAll instead. */ * Use RedisModule_GetContextFlagsAll instead. */
#define _REDISMODULE_CTX_FLAGS_NEXT (1<<24) #define _REDISMODULE_CTX_FLAGS_NEXT (1<<25)
/* Keyspace changes notification classes. Every class is associated with a /* Keyspace changes notification classes. Every class is associated with a
* character for configuration purposes. * character for configuration purposes.
...@@ -748,13 +750,41 @@ typedef enum { ...@@ -748,13 +750,41 @@ typedef enum {
REDISMODULE_ACL_LOG_CHANNEL /* Channel authorization failure */ REDISMODULE_ACL_LOG_CHANNEL /* Channel authorization failure */
} RedisModuleACLLogEntryReason; } RedisModuleACLLogEntryReason;
/* Incomplete structures needed by both the core and modules. */
typedef struct RedisModuleString RedisModuleString;
typedef struct RedisModuleIO RedisModuleIO;
typedef struct RedisModuleDigest RedisModuleDigest;
typedef struct RedisModuleInfoCtx RedisModuleInfoCtx;
typedef struct RedisModuleDefragCtx RedisModuleDefragCtx;
/* Function pointers needed by both the core and modules, these needs to be
* exposed since you can't cast a function pointer to (void *). */
typedef void (*RedisModuleInfoFunc)(RedisModuleInfoCtx *ctx, int for_crash_report);
typedef void (*RedisModuleDefragFunc)(RedisModuleDefragCtx *ctx);
typedef void (*RedisModuleUserChangedFunc) (uint64_t client_id, void *privdata);
/* ------------------------- End of common defines ------------------------ */ /* ------------------------- End of common defines ------------------------ */
#ifndef REDISMODULE_CORE #if defined REDISMODULE_CORE
/* Things only defined for the modules core (server), not exported to modules
* that include this file. */
#define RedisModuleString robj
#endif /* defined REDISMODULE_CORE */
#if !defined REDISMODULE_CORE && !defined REDISMODULE_CORE_MODULE
/* Things defined for modules, but not for core-modules. */
typedef long long mstime_t; typedef long long mstime_t;
typedef long long ustime_t; typedef long long ustime_t;
#endif /* !defined REDISMODULE_CORE && !defined REDISMODULE_CORE_MODULE */
/* ----------- The rest of the defines are only for modules ----------------- */
#if !defined REDISMODULE_CORE || defined REDISMODULE_CORE_MODULE
/* Things defined for modules and core-modules. */
/* Macro definitions specific to individual compilers */ /* Macro definitions specific to individual compilers */
#ifndef REDISMODULE_ATTR_UNUSED #ifndef REDISMODULE_ATTR_UNUSED
# ifdef __GNUC__ # ifdef __GNUC__
...@@ -784,21 +814,16 @@ typedef long long ustime_t; ...@@ -784,21 +814,16 @@ typedef long long ustime_t;
typedef struct RedisModuleCtx RedisModuleCtx; typedef struct RedisModuleCtx RedisModuleCtx;
typedef struct RedisModuleCommand RedisModuleCommand; typedef struct RedisModuleCommand RedisModuleCommand;
typedef struct RedisModuleKey RedisModuleKey; typedef struct RedisModuleKey RedisModuleKey;
typedef struct RedisModuleString RedisModuleString;
typedef struct RedisModuleCallReply RedisModuleCallReply; typedef struct RedisModuleCallReply RedisModuleCallReply;
typedef struct RedisModuleIO RedisModuleIO;
typedef struct RedisModuleType RedisModuleType; typedef struct RedisModuleType RedisModuleType;
typedef struct RedisModuleDigest RedisModuleDigest;
typedef struct RedisModuleBlockedClient RedisModuleBlockedClient; typedef struct RedisModuleBlockedClient RedisModuleBlockedClient;
typedef struct RedisModuleClusterInfo RedisModuleClusterInfo; typedef struct RedisModuleClusterInfo RedisModuleClusterInfo;
typedef struct RedisModuleDict RedisModuleDict; typedef struct RedisModuleDict RedisModuleDict;
typedef struct RedisModuleDictIter RedisModuleDictIter; typedef struct RedisModuleDictIter RedisModuleDictIter;
typedef struct RedisModuleCommandFilterCtx RedisModuleCommandFilterCtx; typedef struct RedisModuleCommandFilterCtx RedisModuleCommandFilterCtx;
typedef struct RedisModuleCommandFilter RedisModuleCommandFilter; typedef struct RedisModuleCommandFilter RedisModuleCommandFilter;
typedef struct RedisModuleInfoCtx RedisModuleInfoCtx;
typedef struct RedisModuleServerInfoData RedisModuleServerInfoData; typedef struct RedisModuleServerInfoData RedisModuleServerInfoData;
typedef struct RedisModuleScanCursor RedisModuleScanCursor; typedef struct RedisModuleScanCursor RedisModuleScanCursor;
typedef struct RedisModuleDefragCtx RedisModuleDefragCtx;
typedef struct RedisModuleUser RedisModuleUser; typedef struct RedisModuleUser RedisModuleUser;
typedef struct RedisModuleKeyOptCtx RedisModuleKeyOptCtx; typedef struct RedisModuleKeyOptCtx RedisModuleKeyOptCtx;
...@@ -825,11 +850,8 @@ typedef void (*RedisModuleClusterMessageReceiver)(RedisModuleCtx *ctx, const cha ...@@ -825,11 +850,8 @@ typedef void (*RedisModuleClusterMessageReceiver)(RedisModuleCtx *ctx, const cha
typedef void (*RedisModuleTimerProc)(RedisModuleCtx *ctx, void *data); typedef void (*RedisModuleTimerProc)(RedisModuleCtx *ctx, void *data);
typedef void (*RedisModuleCommandFilterFunc) (RedisModuleCommandFilterCtx *filter); typedef void (*RedisModuleCommandFilterFunc) (RedisModuleCommandFilterCtx *filter);
typedef void (*RedisModuleForkDoneHandler) (int exitcode, int bysignal, void *user_data); typedef void (*RedisModuleForkDoneHandler) (int exitcode, int bysignal, void *user_data);
typedef void (*RedisModuleInfoFunc)(RedisModuleInfoCtx *ctx, int for_crash_report);
typedef void (*RedisModuleScanCB)(RedisModuleCtx *ctx, RedisModuleString *keyname, RedisModuleKey *key, void *privdata); typedef void (*RedisModuleScanCB)(RedisModuleCtx *ctx, RedisModuleString *keyname, RedisModuleKey *key, void *privdata);
typedef void (*RedisModuleScanKeyCB)(RedisModuleKey *key, RedisModuleString *field, RedisModuleString *value, void *privdata); typedef void (*RedisModuleScanKeyCB)(RedisModuleKey *key, RedisModuleString *field, RedisModuleString *value, void *privdata);
typedef void (*RedisModuleUserChangedFunc) (uint64_t client_id, void *privdata);
typedef int (*RedisModuleDefragFunc)(RedisModuleDefragCtx *ctx);
typedef RedisModuleString * (*RedisModuleConfigGetStringFunc)(const char *name, void *privdata); typedef RedisModuleString * (*RedisModuleConfigGetStringFunc)(const char *name, void *privdata);
typedef long long (*RedisModuleConfigGetNumericFunc)(const char *name, void *privdata); typedef long long (*RedisModuleConfigGetNumericFunc)(const char *name, void *privdata);
typedef int (*RedisModuleConfigGetBoolFunc)(const char *name, void *privdata); typedef int (*RedisModuleConfigGetBoolFunc)(const char *name, void *privdata);
...@@ -1557,11 +1579,5 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -1557,11 +1579,5 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
#define RMAPI_FUNC_SUPPORTED(func) (func != NULL) #define RMAPI_FUNC_SUPPORTED(func) (func != NULL)
#else
/* Things only defined for the modules core, not exported to modules
* including this file. */
#define RedisModuleString robj
#endif /* REDISMODULE_CORE */ #endif /* REDISMODULE_CORE */
#endif /* REDISMODULE_H */ #endif /* REDISMODULE_H */
...@@ -35,7 +35,6 @@ ...@@ -35,7 +35,6 @@
#include <stdio.h> #include <stdio.h>
#include "release.h" #include "release.h"
#include "version.h"
#include "crc64.h" #include "crc64.h"
char *redisGitSHA1(void) { char *redisGitSHA1(void) {
...@@ -46,8 +45,12 @@ char *redisGitDirty(void) { ...@@ -46,8 +45,12 @@ char *redisGitDirty(void) {
return REDIS_GIT_DIRTY; return REDIS_GIT_DIRTY;
} }
const char *redisBuildIdRaw(void) {
return REDIS_BUILD_ID_RAW;
}
uint64_t redisBuildId(void) { uint64_t redisBuildId(void) {
char *buildid = REDIS_VERSION REDIS_BUILD_ID REDIS_GIT_DIRTY REDIS_GIT_SHA1; char *buildid = REDIS_BUILD_ID_RAW;
return crc64(0,(unsigned char*)buildid,strlen(buildid)); return crc64(0,(unsigned char*)buildid,strlen(buildid));
} }
......
...@@ -33,6 +33,7 @@ ...@@ -33,6 +33,7 @@
#include "cluster.h" #include "cluster.h"
#include "bio.h" #include "bio.h"
#include "functions.h" #include "functions.h"
#include "connection.h"
#include <memory.h> #include <memory.h>
#include <sys/time.h> #include <sys/time.h>
...@@ -54,6 +55,13 @@ int cancelReplicationHandshake(int reconnect); ...@@ -54,6 +55,13 @@ int cancelReplicationHandshake(int reconnect);
int RDBGeneratedByReplication = 0; int RDBGeneratedByReplication = 0;
/* --------------------------- Utility functions ---------------------------- */ /* --------------------------- Utility functions ---------------------------- */
static ConnectionType *connTypeOfReplication() {
if (server.tls_replication) {
return connectionTypeTls();
}
return connectionTypeTcp();
}
/* Return the pointer to a string representing the slave ip:listening_port /* Return the pointer to a string representing the slave ip:listening_port
* pair. Mostly useful for logging, since we want to log a slave using its * pair. Mostly useful for logging, since we want to log a slave using its
...@@ -66,11 +74,11 @@ char *replicationGetSlaveName(client *c) { ...@@ -66,11 +74,11 @@ char *replicationGetSlaveName(client *c) {
ip[0] = '\0'; ip[0] = '\0';
buf[0] = '\0'; buf[0] = '\0';
if (c->slave_addr || if (c->slave_addr ||
connPeerToString(c->conn,ip,sizeof(ip),NULL) != -1) connAddrPeerName(c->conn,ip,sizeof(ip),NULL) != -1)
{ {
char *addr = c->slave_addr ? c->slave_addr : ip; char *addr = c->slave_addr ? c->slave_addr : ip;
if (c->slave_listening_port) if (c->slave_listening_port)
anetFormatAddr(buf,sizeof(buf),addr,c->slave_listening_port); formatAddr(buf,sizeof(buf),addr,c->slave_listening_port);
else else
snprintf(buf,sizeof(buf),"%s:<unknown-replica-port>",addr); snprintf(buf,sizeof(buf),"%s:<unknown-replica-port>",addr);
} else { } else {
...@@ -2863,7 +2871,7 @@ write_error: /* Handle sendCommand() errors. */ ...@@ -2863,7 +2871,7 @@ write_error: /* Handle sendCommand() errors. */
} }
int connectWithMaster(void) { int connectWithMaster(void) {
server.repl_transfer_s = server.tls_replication ? connCreateTLS() : connCreateSocket(); server.repl_transfer_s = connCreate(connTypeOfReplication());
if (connConnect(server.repl_transfer_s, server.masterhost, server.masterport, if (connConnect(server.repl_transfer_s, server.masterhost, server.masterport,
server.bind_source_addr, syncWithMaster) == C_ERR) { server.bind_source_addr, syncWithMaster) == C_ERR) {
serverLog(LL_WARNING,"Unable to connect to MASTER: %s", serverLog(LL_WARNING,"Unable to connect to MASTER: %s",
...@@ -3156,7 +3164,7 @@ void roleCommand(client *c) { ...@@ -3156,7 +3164,7 @@ void roleCommand(client *c) {
char ip[NET_IP_STR_LEN], *slaveaddr = slave->slave_addr; char ip[NET_IP_STR_LEN], *slaveaddr = slave->slave_addr;
if (!slaveaddr) { if (!slaveaddr) {
if (connPeerToString(slave->conn,ip,sizeof(ip),NULL) == -1) if (connAddrPeerName(slave->conn,ip,sizeof(ip),NULL) == -1)
continue; continue;
slaveaddr = ip; slaveaddr = ip;
} }
...@@ -3817,7 +3825,7 @@ static client *findReplica(char *host, int port) { ...@@ -3817,7 +3825,7 @@ static client *findReplica(char *host, int port) {
char ip[NET_IP_STR_LEN], *replicaip = replica->slave_addr; char ip[NET_IP_STR_LEN], *replicaip = replica->slave_addr;
if (!replicaip) { if (!replicaip) {
if (connPeerToString(replica->conn, ip, sizeof(ip), NULL) == -1) if (connAddrPeerName(replica->conn, ip, sizeof(ip), NULL) == -1)
continue; continue;
replicaip = ip; replicaip = ip;
} }
...@@ -4048,7 +4056,7 @@ void updateFailoverStatus(void) { ...@@ -4048,7 +4056,7 @@ void updateFailoverStatus(void) {
char ip[NET_IP_STR_LEN], *replicaaddr = replica->slave_addr; char ip[NET_IP_STR_LEN], *replicaaddr = replica->slave_addr;
if (!replicaaddr) { if (!replicaaddr) {
if (connPeerToString(replica->conn,ip,sizeof(ip),NULL) == -1) if (connAddrPeerName(replica->conn,ip,sizeof(ip),NULL) == -1)
continue; continue;
replicaaddr = ip; replicaaddr = ip;
} }
......
...@@ -30,7 +30,7 @@ ...@@ -30,7 +30,7 @@
#include "server.h" #include "server.h"
#include "hiredis.h" #include "hiredis.h"
#ifdef USE_OPENSSL #if USE_OPENSSL == 1 /* BUILD_YES */
#include "openssl/ssl.h" #include "openssl/ssl.h"
#include "hiredis_ssl.h" #include "hiredis_ssl.h"
#endif #endif
...@@ -44,7 +44,7 @@ ...@@ -44,7 +44,7 @@
extern char **environ; extern char **environ;
#ifdef USE_OPENSSL #if USE_OPENSSL == 1 /* BUILD_YES */
extern SSL_CTX *redis_tls_ctx; extern SSL_CTX *redis_tls_ctx;
extern SSL_CTX *redis_tls_client_ctx; extern SSL_CTX *redis_tls_client_ctx;
#endif #endif
...@@ -848,7 +848,7 @@ void sentinelRunPendingScripts(void) { ...@@ -848,7 +848,7 @@ void sentinelRunPendingScripts(void) {
sj->pid = 0; sj->pid = 0;
} else if (pid == 0) { } else if (pid == 0) {
/* Child */ /* Child */
tlsCleanup(); connTypeCleanupAll();
execve(sj->argv[0],sj->argv,environ); execve(sj->argv[0],sj->argv,environ);
/* If we are here an error occurred. */ /* If we are here an error occurred. */
_exit(2); /* Don't retry execution. */ _exit(2); /* Don't retry execution. */
...@@ -2378,9 +2378,7 @@ void sentinelSetClientName(sentinelRedisInstance *ri, redisAsyncContext *c, char ...@@ -2378,9 +2378,7 @@ void sentinelSetClientName(sentinelRedisInstance *ri, redisAsyncContext *c, char
} }
static int instanceLinkNegotiateTLS(redisAsyncContext *context) { static int instanceLinkNegotiateTLS(redisAsyncContext *context) {
#ifndef USE_OPENSSL #if USE_OPENSSL == 1 /* BUILD_YES */
(void) context;
#else
if (!redis_tls_ctx) return C_ERR; if (!redis_tls_ctx) return C_ERR;
SSL *ssl = SSL_new(redis_tls_client_ctx ? redis_tls_client_ctx : redis_tls_ctx); SSL *ssl = SSL_new(redis_tls_client_ctx ? redis_tls_client_ctx : redis_tls_ctx);
if (!ssl) return C_ERR; if (!ssl) return C_ERR;
...@@ -2389,6 +2387,8 @@ static int instanceLinkNegotiateTLS(redisAsyncContext *context) { ...@@ -2389,6 +2387,8 @@ static int instanceLinkNegotiateTLS(redisAsyncContext *context) {
SSL_free(ssl); SSL_free(ssl);
return C_ERR; return C_ERR;
} }
#else
UNUSED(context);
#endif #endif
return C_OK; return C_OK;
} }
...@@ -3027,7 +3027,7 @@ int sentinelSendHello(sentinelRedisInstance *ri) { ...@@ -3027,7 +3027,7 @@ int sentinelSendHello(sentinelRedisInstance *ri) {
if (sentinel.announce_ip) { if (sentinel.announce_ip) {
announce_ip = sentinel.announce_ip; announce_ip = sentinel.announce_ip;
} else { } else {
if (anetFdToString(ri->link->cc->c.fd,ip,sizeof(ip),NULL,FD_TO_SOCK_NAME) == -1) if (anetFdToString(ri->link->cc->c.fd,ip,sizeof(ip),NULL,0) == -1)
return C_ERR; return C_ERR;
announce_ip = ip; announce_ip = ip;
} }
......
...@@ -1535,7 +1535,7 @@ void beforeSleep(struct aeEventLoop *eventLoop) { ...@@ -1535,7 +1535,7 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
if (ProcessingEventsWhileBlocked) { if (ProcessingEventsWhileBlocked) {
uint64_t processed = 0; uint64_t processed = 0;
processed += handleClientsWithPendingReadsUsingThreads(); processed += handleClientsWithPendingReadsUsingThreads();
processed += tlsProcessPendingData(); processed += connTypeProcessPendingData();
if (server.aof_state == AOF_ON || server.aof_state == AOF_WAIT_REWRITE) if (server.aof_state == AOF_ON || server.aof_state == AOF_WAIT_REWRITE)
flushAppendOnlyFile(0); flushAppendOnlyFile(0);
processed += handleClientsWithPendingWrites(); processed += handleClientsWithPendingWrites();
...@@ -1550,11 +1550,11 @@ void beforeSleep(struct aeEventLoop *eventLoop) { ...@@ -1550,11 +1550,11 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
/* We should handle pending reads clients ASAP after event loop. */ /* We should handle pending reads clients ASAP after event loop. */
handleClientsWithPendingReadsUsingThreads(); handleClientsWithPendingReadsUsingThreads();
/* Handle TLS pending data. (must be done before flushAppendOnlyFile) */ /* Handle pending data(typical TLS). (must be done before flushAppendOnlyFile) */
tlsProcessPendingData(); connTypeProcessPendingData();
/* If tls still has pending unread data don't sleep at all. */ /* If any connection type(typical TLS) still has pending unread data don't sleep at all. */
aeSetDontWait(server.el, tlsHasPendingData()); aeSetDontWait(server.el, connTypeHasPendingData());
/* Call the Redis Cluster before sleep function. Note that this function /* Call the Redis Cluster before sleep function. Note that this function
* may change the state of Redis Cluster (from ok to fail or vice versa), * may change the state of Redis Cluster (from ok to fail or vice versa),
...@@ -1862,9 +1862,7 @@ void initServerConfig(void) { ...@@ -1862,9 +1862,7 @@ void initServerConfig(void) {
server.bindaddr_count = CONFIG_DEFAULT_BINDADDR_COUNT; server.bindaddr_count = CONFIG_DEFAULT_BINDADDR_COUNT;
for (j = 0; j < CONFIG_DEFAULT_BINDADDR_COUNT; j++) for (j = 0; j < CONFIG_DEFAULT_BINDADDR_COUNT; j++)
server.bindaddr[j] = zstrdup(default_bindaddr[j]); server.bindaddr[j] = zstrdup(default_bindaddr[j]);
server.ipfd.count = 0; memset(server.listeners, 0x00, sizeof(server.listeners));
server.tlsfd.count = 0;
server.sofd = -1;
server.active_expire_enabled = 1; server.active_expire_enabled = 1;
server.skip_checksum_validation = 0; server.skip_checksum_validation = 0;
server.loading = 0; server.loading = 0;
...@@ -2232,7 +2230,7 @@ void checkTcpBacklogSettings(void) { ...@@ -2232,7 +2230,7 @@ void checkTcpBacklogSettings(void) {
#endif #endif
} }
void closeSocketListeners(socketFds *sfd) { void closeListener(connListener *sfd) {
int j; int j;
for (j = 0; j < sfd->count; j++) { for (j = 0; j < sfd->count; j++) {
...@@ -2247,11 +2245,11 @@ void closeSocketListeners(socketFds *sfd) { ...@@ -2247,11 +2245,11 @@ void closeSocketListeners(socketFds *sfd) {
/* Create an event handler for accepting new connections in TCP or TLS domain sockets. /* Create an event handler for accepting new connections in TCP or TLS domain sockets.
* This works atomically for all socket fds */ * This works atomically for all socket fds */
int createSocketAcceptHandler(socketFds *sfd, aeFileProc *accept_handler) { int createSocketAcceptHandler(connListener *sfd, aeFileProc *accept_handler) {
int j; int j;
for (j = 0; j < sfd->count; j++) { for (j = 0; j < sfd->count; j++) {
if (aeCreateFileEvent(server.el, sfd->fd[j], AE_READABLE, accept_handler,NULL) == AE_ERR) { if (aeCreateFileEvent(server.el, sfd->fd[j], AE_READABLE, accept_handler,sfd) == AE_ERR) {
/* Rollback */ /* Rollback */
for (j = j-1; j >= 0; j--) aeDeleteFileEvent(server.el, sfd->fd[j], AE_READABLE); for (j = j-1; j >= 0; j--) aeDeleteFileEvent(server.el, sfd->fd[j], AE_READABLE);
return C_ERR; return C_ERR;
...@@ -2264,7 +2262,8 @@ int createSocketAcceptHandler(socketFds *sfd, aeFileProc *accept_handler) { ...@@ -2264,7 +2262,8 @@ int createSocketAcceptHandler(socketFds *sfd, aeFileProc *accept_handler) {
* binding the addresses specified in the Redis server configuration. * binding the addresses specified in the Redis server configuration.
* *
* The listening file descriptors are stored in the integer array 'fds' * The listening file descriptors are stored in the integer array 'fds'
* and their number is set in '*count'. * and their number is set in '*count'. Actually @sfd should be 'listener',
* for the historical reasons, let's keep 'sfd' here.
* *
* The addresses to bind are specified in the global server.bindaddr array * The addresses to bind are specified in the global server.bindaddr array
* and their number is server.bindaddr_count. If the server configuration * and their number is server.bindaddr_count. If the server configuration
...@@ -2278,14 +2277,15 @@ int createSocketAcceptHandler(socketFds *sfd, aeFileProc *accept_handler) { ...@@ -2278,14 +2277,15 @@ int createSocketAcceptHandler(socketFds *sfd, aeFileProc *accept_handler) {
* impossible to bind, or no bind addresses were specified in the server * impossible to bind, or no bind addresses were specified in the server
* configuration but the function is not able to bind * for at least * configuration but the function is not able to bind * for at least
* one of the IPv4 or IPv6 protocols. */ * one of the IPv4 or IPv6 protocols. */
int listenToPort(int port, socketFds *sfd) { int listenToPort(connListener *sfd) {
int j; int j;
char **bindaddr = server.bindaddr; int port = sfd->port;
char **bindaddr = sfd->bindaddr;
/* If we have no bind address, we don't listen on a TCP socket */ /* If we have no bind address, we don't listen on a TCP socket */
if (server.bindaddr_count == 0) return C_OK; if (sfd->bindaddr_count == 0) return C_OK;
for (j = 0; j < server.bindaddr_count; j++) { for (j = 0; j < sfd->bindaddr_count; j++) {
char* addr = bindaddr[j]; char* addr = bindaddr[j];
int optional = *addr == '-'; int optional = *addr == '-';
if (optional) addr++; if (optional) addr++;
...@@ -2309,7 +2309,7 @@ int listenToPort(int port, socketFds *sfd) { ...@@ -2309,7 +2309,7 @@ int listenToPort(int port, socketFds *sfd) {
continue; continue;
/* Rollback successful listens before exiting */ /* Rollback successful listens before exiting */
closeSocketListeners(sfd); closeListener(sfd);
return C_ERR; return C_ERR;
} }
if (server.socket_mark_id > 0) anetSetSockMarkId(NULL, sfd->fd[sfd->count], server.socket_mark_id); if (server.socket_mark_id > 0) anetSetSockMarkId(NULL, sfd->fd[sfd->count], server.socket_mark_id);
...@@ -2446,12 +2446,6 @@ void initServer(void) { ...@@ -2446,12 +2446,6 @@ void initServer(void) {
exit(1); exit(1);
} }
if ((server.tls_port || server.tls_replication || server.tls_cluster)
&& tlsConfigure(&server.tls_ctx_config) == C_ERR) {
serverLog(LL_WARNING, "Failed to configure TLS. Check logs for more info.");
exit(1);
}
for (j = 0; j < CLIENT_MEM_USAGE_BUCKETS; j++) { for (j = 0; j < CLIENT_MEM_USAGE_BUCKETS; j++) {
server.client_mem_usage_buckets[j].mem_usage_sum = 0; server.client_mem_usage_buckets[j].mem_usage_sum = 0;
server.client_mem_usage_buckets[j].clients = listCreate(); server.client_mem_usage_buckets[j].clients = listCreate();
...@@ -2470,39 +2464,6 @@ void initServer(void) { ...@@ -2470,39 +2464,6 @@ void initServer(void) {
} }
server.db = zmalloc(sizeof(redisDb)*server.dbnum); server.db = zmalloc(sizeof(redisDb)*server.dbnum);
/* Open the TCP listening socket for the user commands. */
if (server.port != 0 &&
listenToPort(server.port,&server.ipfd) == C_ERR) {
/* Note: the following log text is matched by the test suite. */
serverLog(LL_WARNING, "Failed listening on port %u (TCP), aborting.", server.port);
exit(1);
}
if (server.tls_port != 0 &&
listenToPort(server.tls_port,&server.tlsfd) == C_ERR) {
/* Note: the following log text is matched by the test suite. */
serverLog(LL_WARNING, "Failed listening on port %u (TLS), aborting.", server.tls_port);
exit(1);
}
/* Open the listening Unix domain socket. */
if (server.unixsocket != NULL) {
unlink(server.unixsocket); /* don't care if this fails */
server.sofd = anetUnixServer(server.neterr,server.unixsocket,
(mode_t)server.unixsocketperm, server.tcp_backlog);
if (server.sofd == ANET_ERR) {
serverLog(LL_WARNING, "Failed opening Unix socket: %s", server.neterr);
exit(1);
}
anetNonBlock(NULL,server.sofd);
anetCloexec(server.sofd);
}
/* Abort if there are no listening sockets at all. */
if (server.ipfd.count == 0 && server.tlsfd.count == 0 && server.sofd < 0) {
serverLog(LL_WARNING, "Configured to not listen anywhere, exiting.");
exit(1);
}
/* Create the Redis databases, and initialize other internal state. */ /* Create the Redis databases, and initialize other internal state. */
for (j = 0; j < server.dbnum; j++) { for (j = 0; j < server.dbnum; j++) {
server.db[j].dict = dictCreate(&dbDictType); server.db[j].dict = dictCreate(&dbDictType);
...@@ -2583,18 +2544,6 @@ void initServer(void) { ...@@ -2583,18 +2544,6 @@ void initServer(void) {
exit(1); exit(1);
} }
/* Create an event handler for accepting new connections in TCP and Unix
* domain sockets. */
if (createSocketAcceptHandler(&server.ipfd, acceptTcpHandler) != C_OK) {
serverPanic("Unrecoverable error creating TCP socket accept handler.");
}
if (createSocketAcceptHandler(&server.tlsfd, acceptTLSHandler) != C_OK) {
serverPanic("Unrecoverable error creating TLS socket accept handler.");
}
if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE,
acceptUnixHandler,NULL) == AE_ERR) serverPanic("Unrecoverable error creating server.sofd file event.");
/* Register a readable event for the pipe used to awake the event loop /* Register a readable event for the pipe used to awake the event loop
* from module threads. */ * from module threads. */
if (aeCreateFileEvent(server.el, server.module_pipe[0], AE_READABLE, if (aeCreateFileEvent(server.el, server.module_pipe[0], AE_READABLE,
...@@ -2618,7 +2567,6 @@ void initServer(void) { ...@@ -2618,7 +2567,6 @@ void initServer(void) {
server.maxmemory_policy = MAXMEMORY_NO_EVICTION; server.maxmemory_policy = MAXMEMORY_NO_EVICTION;
} }
if (server.cluster_enabled) clusterInit();
scriptingInit(1); scriptingInit(1);
functionsInit(); functionsInit();
slowlogInit(); slowlogInit();
...@@ -2630,6 +2578,78 @@ void initServer(void) { ...@@ -2630,6 +2578,78 @@ void initServer(void) {
applyWatchdogPeriod(); applyWatchdogPeriod();
} }
void initListeners() {
/* Setup listeners from server config for TCP/TLS/Unix */
int conn_index;
connListener *listener;
if (server.port != 0) {
conn_index = connectionIndexByType(CONN_TYPE_SOCKET);
if (conn_index < 0)
serverPanic("Failed finding connection listener of %s", CONN_TYPE_SOCKET);
listener = &server.listeners[conn_index];
listener->bindaddr = server.bindaddr;
listener->bindaddr_count = server.bindaddr_count;
listener->port = server.port;
listener->ct = connectionByType(CONN_TYPE_SOCKET);
}
if (server.tls_port || server.tls_replication || server.tls_cluster) {
ConnectionType *ct_tls = connectionTypeTls();
if (!ct_tls) {
serverLog(LL_WARNING, "Failed finding TLS support.");
exit(1);
}
if (connTypeConfigure(ct_tls, &server.tls_ctx_config, 1) == C_ERR) {
serverLog(LL_WARNING, "Failed to configure TLS. Check logs for more info.");
exit(1);
}
}
if (server.tls_port != 0) {
conn_index = connectionIndexByType(CONN_TYPE_TLS);
if (conn_index < 0)
serverPanic("Failed finding connection listener of %s", CONN_TYPE_TLS);
listener = &server.listeners[conn_index];
listener->bindaddr = server.bindaddr;
listener->bindaddr_count = server.bindaddr_count;
listener->port = server.tls_port;
listener->ct = connectionByType(CONN_TYPE_TLS);
}
if (server.unixsocket != NULL) {
conn_index = connectionIndexByType(CONN_TYPE_UNIX);
if (conn_index < 0)
serverPanic("Failed finding connection listener of %s", CONN_TYPE_UNIX);
listener = &server.listeners[conn_index];
listener->bindaddr = &server.unixsocket;
listener->bindaddr_count = 1;
listener->ct = connectionByType(CONN_TYPE_UNIX);
listener->priv = &server.unixsocketperm; /* Unix socket specified */
}
/* create all the configured listener, and add handler to start to accept */
int listen_fds = 0;
for (int j = 0; j < CONN_TYPE_MAX; j++) {
listener = &server.listeners[j];
if (listener->ct == NULL)
continue;
if (connListen(listener) == C_ERR) {
serverLog(LL_WARNING, "Failed listening on port %u (%s), aborting.", listener->port, listener->ct->get_type(NULL));
exit(1);
}
if (createSocketAcceptHandler(listener, connAcceptHandler(listener->ct)) != C_OK)
serverPanic("Unrecoverable error creating %s listener accept handler.", listener->ct->get_type(NULL));
listen_fds += listener->count;
}
if (listen_fds == 0) {
serverLog(LL_WARNING, "Configured to not listen anywhere, exiting.");
exit(1);
}
}
/* Some steps in server initialization need to be done last (after modules /* Some steps in server initialization need to be done last (after modules
* are loaded). * are loaded).
* Specifically, creation of threads due to a race bug in ld.so, in which * Specifically, creation of threads due to a race bug in ld.so, in which
...@@ -4065,11 +4085,16 @@ void incrementErrorCount(const char *fullerr, size_t namelen) { ...@@ -4065,11 +4085,16 @@ void incrementErrorCount(const char *fullerr, size_t namelen) {
void closeListeningSockets(int unlink_unix_socket) { void closeListeningSockets(int unlink_unix_socket) {
int j; int j;
for (j = 0; j < server.ipfd.count; j++) close(server.ipfd.fd[j]); for (int i = 0; i < CONN_TYPE_MAX; i++) {
for (j = 0; j < server.tlsfd.count; j++) close(server.tlsfd.fd[j]); connListener *listener = &server.listeners[i];
if (server.sofd != -1) close(server.sofd); if (listener->ct == NULL)
continue;
for (j = 0; j < listener->count; j++) close(listener->fd[j]);
}
if (server.cluster_enabled) if (server.cluster_enabled)
for (j = 0; j < server.cfd.count; j++) close(server.cfd.fd[j]); for (j = 0; j < server.clistener.count; j++) close(server.clistener.fd[j]);
if (unlink_unix_socket && server.unixsocket) { if (unlink_unix_socket && server.unixsocket) {
serverLog(LL_NOTICE,"Removing the unix socket file."); serverLog(LL_NOTICE,"Removing the unix socket file.");
if (unlink(server.unixsocket) != 0) if (unlink(server.unixsocket) != 0)
...@@ -5409,6 +5434,9 @@ sds genRedisInfoString(dict *section_dict, int all_sections, int everything) { ...@@ -5409,6 +5434,9 @@ sds genRedisInfoString(dict *section_dict, int all_sections, int everything) {
"shutdown_in_milliseconds:%I\r\n", "shutdown_in_milliseconds:%I\r\n",
(int64_t)(server.shutdown_mstime - server.mstime)); (int64_t)(server.shutdown_mstime - server.mstime));
} }
/* get all the listeners information */
info = getListensInfoString(info);
} }
/* Clients */ /* Clients */
...@@ -5934,7 +5962,7 @@ sds genRedisInfoString(dict *section_dict, int all_sections, int everything) { ...@@ -5934,7 +5962,7 @@ sds genRedisInfoString(dict *section_dict, int all_sections, int everything) {
long lag = 0; long lag = 0;
if (!slaveip) { if (!slaveip) {
if (connPeerToString(slave->conn,ip,sizeof(ip),&port) == -1) if (connAddrPeerName(slave->conn,ip,sizeof(ip),&port) == -1)
continue; continue;
slaveip = ip; slaveip = ip;
} }
...@@ -6264,61 +6292,38 @@ void redisAsciiArt(void) { ...@@ -6264,61 +6292,38 @@ void redisAsciiArt(void) {
zfree(buf); zfree(buf);
} }
int changeBindAddr(void) { /* Get the server listener by type name */
/* Close old TCP and TLS servers */ connListener *listenerByType(const char *typename) {
closeSocketListeners(&server.ipfd); int conn_index;
closeSocketListeners(&server.tlsfd);
/* Bind to the new port */
if ((server.port != 0 && listenToPort(server.port, &server.ipfd) != C_OK) ||
(server.tls_port != 0 && listenToPort(server.tls_port, &server.tlsfd) != C_OK)) {
serverLog(LL_WARNING, "Failed to bind");
closeSocketListeners(&server.ipfd);
closeSocketListeners(&server.tlsfd);
return C_ERR;
}
/* Create TCP and TLS event handlers */
if (createSocketAcceptHandler(&server.ipfd, acceptTcpHandler) != C_OK) {
serverPanic("Unrecoverable error creating TCP socket accept handler.");
}
if (createSocketAcceptHandler(&server.tlsfd, acceptTLSHandler) != C_OK) {
serverPanic("Unrecoverable error creating TLS socket accept handler.");
}
if (server.set_proc_title) redisSetProcTitle(NULL); conn_index = connectionIndexByType(typename);
if (conn_index < 0)
return NULL;
return C_OK; return &server.listeners[conn_index];
} }
int changeListenPort(int port, socketFds *sfd, aeFileProc *accept_handler) { /* Close original listener, re-create a new listener from the updated bind address & port */
socketFds new_sfd = {{0}}; int changeListener(connListener *listener) {
/* Close old servers */ /* Close old servers */
closeSocketListeners(sfd); closeListener(listener);
/* Just close the server if port disabled */ /* Just close the server if port disabled */
if (port == 0) { if (listener->port == 0) {
if (server.set_proc_title) redisSetProcTitle(NULL); if (server.set_proc_title) redisSetProcTitle(NULL);
return C_OK; return C_OK;
} }
/* Bind to the new port */ /* Re-create listener */
if (listenToPort(port, &new_sfd) != C_OK) { if (connListen(listener) != C_OK) {
return C_ERR; return C_ERR;
} }
/* Create event handlers */ /* Create event handlers */
if (createSocketAcceptHandler(&new_sfd, accept_handler) != C_OK) { if (createSocketAcceptHandler(listener, listener->ct->accept_handler) != C_OK) {
closeSocketListeners(&new_sfd); serverPanic("Unrecoverable error creating %s accept handler.", listener->ct->get_type(NULL));
return C_ERR;
} }
/* Copy new descriptors */
sfd->count = new_sfd.count;
memcpy(sfd->fd, new_sfd.fd, sizeof(new_sfd.fd));
if (server.set_proc_title) redisSetProcTitle(NULL); if (server.set_proc_title) redisSetProcTitle(NULL);
return C_OK; return C_OK;
...@@ -6944,7 +6949,7 @@ int main(int argc, char **argv) { ...@@ -6944,7 +6949,7 @@ int main(int argc, char **argv) {
ACLInit(); /* The ACL subsystem must be initialized ASAP because the ACLInit(); /* The ACL subsystem must be initialized ASAP because the
basic networking code and client creation depends on it. */ basic networking code and client creation depends on it. */
moduleInitModulesSystem(); moduleInitModulesSystem();
tlsInit(); connTypeInitialize();
/* Store the executable path and arguments in a safe place in order /* Store the executable path and arguments in a safe place in order
* to be able to restart the server later. */ * to be able to restart the server later. */
...@@ -7089,6 +7094,16 @@ int main(int argc, char **argv) { ...@@ -7089,6 +7094,16 @@ int main(int argc, char **argv) {
if (server.set_proc_title) redisSetProcTitle(NULL); if (server.set_proc_title) redisSetProcTitle(NULL);
redisAsciiArt(); redisAsciiArt();
checkTcpBacklogSettings(); checkTcpBacklogSettings();
if (!server.sentinel_mode) {
moduleInitModulesSystemLast();
moduleLoadFromQueue();
}
ACLLoadUsersAtStartup();
initListeners();
if (server.cluster_enabled) {
clusterInit();
}
InitServerLast();
if (!server.sentinel_mode) { if (!server.sentinel_mode) {
/* Things not needed when running in Sentinel mode. */ /* Things not needed when running in Sentinel mode. */
...@@ -7117,10 +7132,6 @@ int main(int argc, char **argv) { ...@@ -7117,10 +7132,6 @@ int main(int argc, char **argv) {
} }
#endif /* __arm64__ */ #endif /* __arm64__ */
#endif /* __linux__ */ #endif /* __linux__ */
moduleInitModulesSystemLast();
moduleLoadFromQueue();
ACLLoadUsersAtStartup();
InitServerLast();
aofLoadManifestFromDisk(); aofLoadManifestFromDisk();
loadDataFromDisk(); loadDataFromDisk();
aofOpenIfNeededOnServerStart(); aofOpenIfNeededOnServerStart();
...@@ -7133,10 +7144,15 @@ int main(int argc, char **argv) { ...@@ -7133,10 +7144,15 @@ int main(int argc, char **argv) {
exit(1); exit(1);
} }
} }
if (server.ipfd.count > 0 || server.tlsfd.count > 0)
serverLog(LL_NOTICE,"Ready to accept connections"); for (j = 0; j < CONN_TYPE_MAX; j++) {
if (server.sofd > 0) connListener *listener = &server.listeners[j];
serverLog(LL_NOTICE,"The server is now ready to accept connections at %s", server.unixsocket); if (listener->ct == NULL)
continue;
serverLog(LL_NOTICE,"Ready to accept connections %s", listener->ct->get_type(NULL));
}
if (server.supervised_mode == SUPERVISED_SYSTEMD) { if (server.supervised_mode == SUPERVISED_SYSTEMD) {
if (!server.masterhost) { if (!server.masterhost) {
redisCommunicateSystemd("STATUS=Ready to accept connections\n"); redisCommunicateSystemd("STATUS=Ready to accept connections\n");
...@@ -7146,8 +7162,6 @@ int main(int argc, char **argv) { ...@@ -7146,8 +7162,6 @@ int main(int argc, char **argv) {
redisCommunicateSystemd("READY=1\n"); redisCommunicateSystemd("READY=1\n");
} }
} else { } else {
ACLLoadUsersAtStartup();
InitServerLast();
sentinelIsRunning(); sentinelIsRunning();
if (server.supervised_mode == SUPERVISED_SYSTEMD) { if (server.supervised_mode == SUPERVISED_SYSTEMD) {
redisCommunicateSystemd("STATUS=Ready to accept connections\n"); redisCommunicateSystemd("STATUS=Ready to accept connections\n");
......
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