Commit 45617385 authored by zhenwei pi's avatar zhenwei pi
Browse files

Use connection name of string



Suggested by Oran, use an array to store all the connection types
instead of a linked list, and use connection name of string. The index
of a connection is dynamically allocated.

Currently we support max 8 connection types, include:
- tcp
- unix socket
- tls

and RDMA is in the plan, then we have another 4 types to support, it
should be enough in a long time.

Introduce 3 functions to get connection type by a fast path:
- connectionTypeTcp()
- connectionTypeTls()
- connectionTypeUnix()

Note that connectionByType() is designed to use only in unlikely code path.
Signed-off-by: default avatarzhenwei pi <pizhenwei@bytedance.com>
parent eb94d6d3
......@@ -119,12 +119,12 @@ dictType clusterNodesBlackListDictType = {
NULL /* allow to expand */
};
static int connTypeOfCluster() {
static ConnectionType *connTypeOfCluster() {
if (server.tls_cluster) {
return CONN_TYPE_TLS;
return connectionTypeTls();
}
return CONN_TYPE_SOCKET;
return connectionTypeTcp();
}
/* -----------------------------------------------------------------------------
......@@ -5030,7 +5030,7 @@ void addNodeToNodeReply(client *c, clusterNode *node) {
/* Report non-TLS ports to non-TLS client in TLS cluster if available. */
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);
addReplyBulkCBuffer(c, node->name, CLUSTER_NAMELEN);
......@@ -5335,7 +5335,7 @@ NULL
/* Report plaintext ports, only if cluster is TLS but client is known to
* be non-TLS). */
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);
addReplyVerbatim(c,nodes,sdslen(nodes),"txt");
sdsfree(nodes);
......@@ -5767,7 +5767,7 @@ NULL
/* Use plaintext port if cluster is TLS but client is non-TLS. */
int use_pport = (server.tls_cluster &&
c->conn && connGetType(c->conn) != CONN_TYPE_TLS);
c->conn && (c->conn->type != connectionTypeTls()));
addReplyArrayLen(c,n->numslaves);
for (j = 0; j < n->numslaves; j++) {
sds ni = clusterGenNodeDescription(n->slaves[j], use_pport);
......@@ -6896,7 +6896,7 @@ void clusterRedirectClient(client *c, clusterNode *n, int hashslot, int error_co
/* Redirect to IP:port. Include plaintext port if cluster is TLS but
* client is non-TLS. */
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;
addReplyErrorSds(c,sdscatprintf(sdsempty(),
"-%s %d %s:%d",
......
......@@ -2430,7 +2430,7 @@ static int updateHZ(const char **err) {
}
static int updatePort(const char **err) {
if (changeListenPort(server.port, &server.ipfd, connAcceptHandler(CONN_TYPE_SOCKET)) == C_ERR) {
if (changeListenPort(server.port, &server.ipfd, connAcceptHandler(connectionTypeTcp())) == C_ERR) {
*err = "Unable to listen on this port. Check server logs.";
return 0;
}
......@@ -2577,7 +2577,7 @@ static int applyTlsCfg(const char **err) {
/* If TLS is enabled, try to configure OpenSSL. */
if ((server.tls_port || server.tls_replication || server.tls_cluster)
&& connTypeConfigure(CONN_TYPE_TLS, &server.tls_ctx_config, 1) == C_ERR) {
&& connTypeConfigure(connectionTypeTls(), &server.tls_ctx_config, 1) == C_ERR) {
*err = "Unable to update TLS configuration. Check server logs.";
return 0;
}
......@@ -2586,12 +2586,12 @@ static int applyTlsCfg(const char **err) {
static int applyTLSPort(const char **err) {
/* Configure TLS in case it wasn't enabled */
if (connTypeConfigure(CONN_TYPE_TLS, &server.tls_ctx_config, 0) == C_ERR) {
if (connTypeConfigure(connectionTypeTls(), &server.tls_ctx_config, 0) == C_ERR) {
*err = "Unable to update TLS configuration. Check server logs.";
return 0;
}
if (changeListenPort(server.tls_port, &server.tlsfd, connAcceptHandler(CONN_TYPE_TLS)) == C_ERR) {
if (changeListenPort(server.tls_port, &server.tlsfd, connAcceptHandler(connectionTypeTls())) == C_ERR) {
*err = "Unable to listen on this port. Check server logs.";
return 0;
}
......
......@@ -30,19 +30,24 @@
static ConnectionType *connTypes[CONN_TYPE_MAX];
int connTypeRegister(ConnectionType *ct) {
int type = ct->get_type(NULL);
/* unknown connection type as a fatal error */
if (type >= CONN_TYPE_MAX) {
serverPanic("Unsupported connection type %d", type);
}
const char *typename = ct->get_type(NULL);
ConnectionType *tmpct;
int type;
if (connTypes[type] == ct) {
serverLog(LL_WARNING, "Connection type %d already registered", type);
return C_OK;
/* find an empty slot to store the new connection type */
for (type = 0; type < CONN_TYPE_MAX; type++) {
tmpct = connTypes[type];
if (!tmpct)
break;
/* ignore case, we really don't care "tls"/"TLS" */
if (!strcasecmp(typename, tmpct->get_type(NULL))) {
serverLog(LL_WARNING, "Connection types %s already registered", typename);
return C_ERR;
}
}
serverLog(LL_VERBOSE, "Connection type %d registered", type);
serverLog(LL_VERBOSE, "Connection type %s registered", typename);
connTypes[type] = ct;
if (ct->init) {
......@@ -65,43 +70,85 @@ int connTypeInitialize() {
return C_OK;
}
ConnectionType *connectionByType(int type) {
ConnectionType *connectionByType(const char *typename) {
ConnectionType *ct;
serverAssert(type < CONN_TYPE_MAX);
for (int type = 0; type < CONN_TYPE_MAX; type++) {
ct = connTypes[type];
if (!ct)
break;
ct = connTypes[type];
if (!ct) {
serverLog(LL_WARNING, "Missing implement of connection type %d", type);
if (!strcasecmp(typename, ct->get_type(NULL)))
return ct;
}
return ct;
serverLog(LL_WARNING, "Missing implement of connection type %s", typename);
return NULL;
}
/* Cache TCP connection type, query it by string once */
ConnectionType *connectionTypeTcp() {
static ConnectionType *ct_tcp = NULL;
if (ct_tcp != NULL)
return ct_tcp;
ct_tcp = connectionByType(CONN_TYPE_SOCKET);
serverAssert(ct_tcp != NULL);
return ct_tcp;
}
void connTypeCleanup(int type) {
ConnectionType *ct = connectionByType(type);
/* Cache TLS connection type, query it by string once */
ConnectionType *connectionTypeTls() {
static ConnectionType *ct_tls = NULL;
if (ct && ct->cleanup) {
ct->cleanup();
if (ct_tls != NULL)
return ct_tls;
ct_tls = connectionByType(CONN_TYPE_TLS);
return ct_tls;
}
/* Cache Unix connection type, query it by string once */
ConnectionType *connectionTypeUnix() {
static ConnectionType *ct_unix = NULL;
if (ct_unix != NULL)
return ct_unix;
ct_unix = connectionByType(CONN_TYPE_UNIX);
return ct_unix;
}
int connectionIndexByType(const char *typename) {
ConnectionType *ct;
for (int type = 0; type < CONN_TYPE_MAX; type++) {
ct = connTypes[type];
if (!ct)
break;
if (!strcasecmp(typename, ct->get_type(NULL)))
return type;
}
return -1;
}
void connTypeCleanupAll() {
ConnectionType *ct;
int type;
for (type = 0; type < CONN_TYPE_MAX; type++) {
connTypeCleanup(type);
}
}
int connTypeConfigure(int type, void *priv, int reconfigure) {
ConnectionType *ct = connectionByType(type);
ct = connTypes[type];
if (!ct)
break;
if (ct && ct->configure) {
return ct->configure(priv, reconfigure);
if (ct->cleanup)
ct->cleanup();
}
return C_ERR;
}
/* walk all the connection types until has pending data */
......@@ -136,37 +183,3 @@ int connTypeProcessPendingData(void) {
return ret;
}
void *connTypeGetCtx(int type) {
ConnectionType *ct = connectionByType(type);
if (ct && ct->get_ctx) {
return ct->get_ctx();
}
return NULL;
}
void *connTypeGetClientCtx(int type) {
ConnectionType *ct = connectionByType(type);
if (ct && ct->get_client_ctx) {
return ct->get_client_ctx();
}
return NULL;
}
connection *connCreate(int type) {
ConnectionType *ct = connectionByType(type);
serverAssert(ct && ct->conn_create);
return ct->conn_create();
}
connection *connCreateAccepted(int type, int fd, void *priv) {
ConnectionType *ct = connectionByType(type);
serverAssert(ct && ct->conn_create_accepted);
return ct->conn_create_accepted(fd, priv);
}
......@@ -57,16 +57,16 @@ typedef enum {
#define CONN_FLAG_CLOSE_SCHEDULED (1<<0) /* Closed scheduled by a handler */
#define CONN_FLAG_WRITE_BARRIER (1<<1) /* Write barrier requested */
#define CONN_TYPE_SOCKET 0
#define CONN_TYPE_UNIX 1
#define CONN_TYPE_TLS 2
#define CONN_TYPE_MAX 3
#define CONN_TYPE_SOCKET "tcp"
#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 struct ConnectionType {
/* connection type */
int (*get_type)(struct connection *conn);
const char *(*get_type)(struct connection *conn);
/* connection type initialize & finalize & configure */
void (*init)(void); /* auto-call during register */
......@@ -251,7 +251,7 @@ static inline ssize_t connSyncReadLine(connection *conn, char *ptr, ssize_t size
}
/* 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);
}
......@@ -341,8 +341,13 @@ int connSendTimeout(connection *conn, long long ms);
int connRecvTimeout(connection *conn, long long ms);
/* Helpers for tls special considerations */
void *connTypeGetCtx(int type);
void *connTypeGetClientCtx(int type);
static inline void *connTypeGetCtx(ConnectionType *ct) {
return ct->get_ctx();
}
static inline void *connTypeGetClientCtx(ConnectionType *ct) {
return ct->get_client_ctx();
}
/* Get cert for the secure connection */
static inline sds connGetPeerCert(connection *conn) {
......@@ -359,23 +364,38 @@ int connTypeInitialize();
/* Register a connection type into redis connection framework */
int connTypeRegister(ConnectionType *ct);
/* Lookup a connection type by index */
ConnectionType *connectionByType(int type);
/* 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 */
connection *connCreate(int type);
static inline connection *connCreate(ConnectionType *ct) {
return ct->conn_create();
}
/* Create a accepted connection of specified type.
* @priv is connection type specified argument */
connection *connCreateAccepted(int type, int fd, void *priv);
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 */
int connTypeConfigure(int type, void *priv, int reconfigure);
/* Cleanup a connection type. A typical case is to cleanup TLS. */
void connTypeCleanup(int type);
static inline int connTypeConfigure(ConnectionType *ct, void *priv, int reconfigure) {
return ct->configure(priv, reconfigure);
}
/* Walk all the connection type, and cleanup them all if possible */
void connTypeCleanupAll();
......@@ -387,8 +407,7 @@ int connTypeHasPendingData(void);
int connTypeProcessPendingData(void);
/* Get accept_handler of a connection type */
static inline aeFileProc *connAcceptHandler(int type) {
ConnectionType *ct = connectionByType(type);
static inline aeFileProc *connAcceptHandler(ConnectionType *ct) {
if (ct)
return ct->accept_handler;
return NULL;
......
......@@ -3327,7 +3327,7 @@ int modulePopulateClientInfoStructure(void *ci, client *client, int structver) {
ci1->flags |= REDISMODULE_CLIENTINFO_FLAG_TRACKING;
if (client->flags & CLIENT_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;
 
int port;
......
......@@ -55,12 +55,12 @@ int cancelReplicationHandshake(int reconnect);
int RDBGeneratedByReplication = 0;
/* --------------------------- Utility functions ---------------------------- */
static int connTypeOfReplication() {
static ConnectionType *connTypeOfReplication() {
if (server.tls_replication) {
return CONN_TYPE_TLS;
return connectionTypeTls();
}
return CONN_TYPE_SOCKET;
return connectionTypeTcp();
}
/* Return the pointer to a string representing the slave ip:listening_port
......
......@@ -2376,8 +2376,8 @@ static int instanceLinkNegotiateTLS(redisAsyncContext *context) {
#ifndef USE_OPENSSL
(void) context;
#else
SSL_CTX *redis_tls_ctx = connTypeGetCtx(CONN_TYPE_TLS);
SSL_CTX *redis_tls_client_ctx = connTypeGetClientCtx(CONN_TYPE_TLS);
SSL_CTX *redis_tls_ctx = connTypeGetCtx(connectionTypeTls());
SSL_CTX *redis_tls_client_ctx = connTypeGetClientCtx(connectionTypeTls());
if (!redis_tls_ctx) return C_ERR;
SSL *ssl = SSL_new(redis_tls_client_ctx ? redis_tls_client_ctx : redis_tls_ctx);
......
......@@ -2447,7 +2447,7 @@ void initServer(void) {
}
if ((server.tls_port || server.tls_replication || server.tls_cluster)
&& connTypeConfigure(CONN_TYPE_TLS, &server.tls_ctx_config, 1) == C_ERR) {
&& connTypeConfigure(connectionTypeTls(), &server.tls_ctx_config, 1) == C_ERR) {
serverLog(LL_WARNING, "Failed to configure TLS. Check logs for more info.");
exit(1);
}
......@@ -2586,13 +2586,13 @@ void initServer(void) {
/* Create an event handler for accepting new connections in TCP and Unix
* domain sockets. */
if (createSocketAcceptHandler(&server.ipfd, connAcceptHandler(CONN_TYPE_SOCKET)) != C_OK) {
if (createSocketAcceptHandler(&server.ipfd, connAcceptHandler(connectionTypeTcp())) != C_OK) {
serverPanic("Unrecoverable error creating TCP socket accept handler.");
}
if (createSocketAcceptHandler(&server.tlsfd, connAcceptHandler(CONN_TYPE_TLS)) != C_OK) {
if (createSocketAcceptHandler(&server.tlsfd, connAcceptHandler(connectionTypeTls())) != C_OK) {
serverPanic("Unrecoverable error creating TLS socket accept handler.");
}
if (createSocketAcceptHandler(&server.sofd, connAcceptHandler(CONN_TYPE_UNIX)) != C_OK) {
if (createSocketAcceptHandler(&server.sofd, connAcceptHandler(connectionTypeUnix())) != C_OK) {
serverPanic("Unrecoverable error creating server.sofd file event.");
}
......@@ -6282,10 +6282,10 @@ int changeBindAddr(void) {
}
/* Create TCP and TLS event handlers */
if (createSocketAcceptHandler(&server.ipfd, connAcceptHandler(CONN_TYPE_SOCKET)) != C_OK) {
if (createSocketAcceptHandler(&server.ipfd, connAcceptHandler(connectionTypeTcp())) != C_OK) {
serverPanic("Unrecoverable error creating TCP socket accept handler.");
}
if (createSocketAcceptHandler(&server.tlsfd, connAcceptHandler(CONN_TYPE_TLS)) != C_OK) {
if (createSocketAcceptHandler(&server.tlsfd, connAcceptHandler(connectionTypeTls())) != C_OK) {
serverPanic("Unrecoverable error creating TLS socket accept handler.");
}
......
......@@ -363,7 +363,7 @@ static ssize_t connSocketSyncReadLine(connection *conn, char *ptr, ssize_t size,
return syncReadLine(conn->fd, ptr, size, timeout);
}
static int connSocketGetType(connection *conn) {
static const char *connSocketGetType(connection *conn) {
(void) conn;
return CONN_TYPE_SOCKET;
......
......@@ -763,7 +763,7 @@ static void connTLSClose(connection *conn_) {
conn->pending_list_node = NULL;
}
connectionByType(CONN_TYPE_SOCKET)->close(conn_);
connectionTypeTcp()->close(conn_);
}
static int connTLSAccept(connection *_conn, ConnectionCallbackFunc accept_handler) {
......@@ -802,7 +802,7 @@ static int connTLSConnect(connection *conn_, const char *addr, int port, const c
ERR_clear_error();
/* Initiate Socket connection first */
if (connectionByType(CONN_TYPE_SOCKET)->connect(conn_, addr, port, src_addr, connect_handler) == C_ERR) return C_ERR;
if (connectionTypeTcp()->connect(conn_, addr, port, src_addr, connect_handler) == C_ERR) return C_ERR;
/* Return now, once the socket is connected we'll initiate
* TLS connection from the event handler.
......@@ -965,7 +965,7 @@ static int connTLSBlockingConnect(connection *conn_, const char *addr, int port,
if (conn->c.state != CONN_STATE_NONE) return C_ERR;
/* Initiate socket blocking connect first */
if (connectionByType(CONN_TYPE_SOCKET)->blocking_connect(conn_, addr, port, timeout) == C_ERR) return C_ERR;
if (connectionTypeTcp()->blocking_connect(conn_, addr, port, timeout) == C_ERR) return C_ERR;
/* Initiate TLS connection now. We set up a send/recv timeout on the socket,
* which means the specified timeout will not be enforced accurately. */
......@@ -1034,7 +1034,7 @@ exit:
return nread;
}
static int connTLSGetType(connection *conn_) {
static const char *connTLSGetType(connection *conn_) {
(void) conn_;
return CONN_TYPE_TLS;
......@@ -1064,7 +1064,7 @@ static int tlsProcessPendingData() {
*/
static sds connTLSGetPeerCert(connection *conn_) {
tls_connection *conn = (tls_connection *) conn_;
if (conn_->type->get_type(conn_) != CONN_TYPE_TLS || !conn->ssl) return NULL;
if ((conn_->type != connectionTypeTls()) || !conn->ssl) return NULL;
X509 *cert = SSL_get_peer_certificate(conn->ssl);
if (!cert) return NULL;
......
......@@ -29,18 +29,18 @@
static ConnectionType CT_Unix;
static int connUnixGetType(connection *conn) {
(void) conn;
static const char *connUnixGetType(connection *conn) {
UNUSED(conn);
return CONN_TYPE_UNIX;
}
static void connUnixEventHandler(struct aeEventLoop *el, int fd, void *clientData, int mask) {
connectionByType(CONN_TYPE_SOCKET)->ae_handler(el, fd, clientData, mask);
connectionTypeTcp()->ae_handler(el, fd, clientData, mask);
}
static int connUnixAddr(connection *conn, char *ip, size_t ip_len, int *port, int remote) {
return connectionByType(CONN_TYPE_SOCKET)->addr(conn, ip, ip_len, port, remote);
return connectionTypeTcp()->addr(conn, ip, ip_len, port, remote);
}
static connection *connCreateUnix(void) {
......@@ -79,31 +79,31 @@ static void connUnixAcceptHandler(aeEventLoop *el, int fd, void *privdata, int m
}
static void connUnixClose(connection *conn) {
connectionByType(CONN_TYPE_SOCKET)->close(conn);
connectionTypeTcp()->close(conn);
}
static int connUnixAccept(connection *conn, ConnectionCallbackFunc accept_handler) {
return connectionByType(CONN_TYPE_SOCKET)->accept(conn, accept_handler);
return connectionTypeTcp()->accept(conn, accept_handler);
}
static int connUnixWrite(connection *conn, const void *data, size_t data_len) {
return connectionByType(CONN_TYPE_SOCKET)->write(conn, data, data_len);
return connectionTypeTcp()->write(conn, data, data_len);
}
static int connUnixWritev(connection *conn, const struct iovec *iov, int iovcnt) {
return connectionByType(CONN_TYPE_SOCKET)->writev(conn, iov, iovcnt);
return connectionTypeTcp()->writev(conn, iov, iovcnt);
}
static int connUnixRead(connection *conn, void *buf, size_t buf_len) {
return connectionByType(CONN_TYPE_SOCKET)->read(conn, buf, buf_len);
return connectionTypeTcp()->read(conn, buf, buf_len);
}
static int connUnixSetWriteHandler(connection *conn, ConnectionCallbackFunc func, int barrier) {
return connectionByType(CONN_TYPE_SOCKET)->set_write_handler(conn, func, barrier);
return connectionTypeTcp()->set_write_handler(conn, func, barrier);
}
static int connUnixSetReadHandler(connection *conn, ConnectionCallbackFunc func) {
return connectionByType(CONN_TYPE_SOCKET)->set_read_handler(conn, func);
return connectionTypeTcp()->set_read_handler(conn, func);
}
static const char *connUnixGetLastError(connection *conn) {
......
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