Unverified Commit 779af3ab authored by Yuan Wang's avatar Yuan Wang Committed by GitHub
Browse files

Dynamic event loop binding for connection structure (#13642)



The IO thread has an independent event loop, so we can no longer
hard-code the event loop to the connection, instead, we should
dynamically select the event loop for the connection.

- configure the event loop during connection creation.
- add a new interface to allow dynamic event loop binding.

For TLS connection, we need to check for any pending data on the
connection and handle it accordingly when changing connection cross IO
thread and main thread. This commit doesn't handle it, @sundb will
overall support for TLS connection later.

---------
Co-authored-by: default avatardebing.sun <debing.sun@redis.com>
parent ded8d993
......@@ -317,7 +317,7 @@ migrateCachedSocket* migrateGetSocket(client *c, robj *host, robj *port, long ti
}
/* Create the connection */
conn = connCreate(connTypeOfCluster());
conn = connCreate(server.el, connTypeOfCluster());
if (connBlockingConnect(conn, host->ptr, atoi(port->ptr), timeout)
!= C_OK) {
addReplyError(c,"-IOERR error or timeout connecting to the client");
......
......@@ -1262,7 +1262,7 @@ void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
return;
}
connection *conn = connCreateAccepted(connTypeOfCluster(), cfd, &require_auth);
connection *conn = connCreateAccepted(server.el, connTypeOfCluster(), cfd, &require_auth);
/* Make sure connection is not in an error state */
if (connGetState(conn) != CONN_STATE_ACCEPTING) {
......@@ -4583,7 +4583,7 @@ static int clusterNodeCronHandleReconnect(clusterNode *node, mstime_t handshake_
if (node->link == NULL) {
clusterLink *link = createClusterLink(node);
link->conn = connCreate(connTypeOfCluster());
link->conn = connCreate(server.el, connTypeOfCluster());
connSetPrivateData(link->conn, link);
if (connConnect(link->conn, node->ip, node->cport, server.bind_source_addr,
clusterLinkConnectHandler) == C_ERR) {
......
......@@ -60,8 +60,8 @@ typedef struct ConnectionType {
int (*listen)(connListener *listener);
/* create/shutdown/close connection */
connection* (*conn_create)(void);
connection* (*conn_create_accepted)(int fd, void *priv);
connection* (*conn_create)(struct aeEventLoop *el);
connection* (*conn_create_accepted)(struct aeEventLoop *el, int fd, void *priv);
void (*shutdown)(struct connection *conn);
void (*close)(struct connection *conn);
......@@ -70,6 +70,9 @@ typedef struct ConnectionType {
int (*blocking_connect)(struct connection *conn, const char *addr, int port, long long timeout);
int (*accept)(struct connection *conn, ConnectionCallbackFunc accept_handler);
/* event loop */
int (*rebind_event_loop)(struct connection *conn, aeEventLoop *el);
/* IO */
int (*write)(struct connection *conn, const void *data, size_t data_len);
int (*writev)(struct connection *conn, const struct iovec *iov, int iovcnt);
......@@ -98,6 +101,7 @@ struct connection {
short int refs;
unsigned short int iovcnt;
void *private_data;
struct aeEventLoop *el;
ConnectionCallbackFunc conn_handler;
ConnectionCallbackFunc write_handler;
ConnectionCallbackFunc read_handler;
......@@ -162,6 +166,12 @@ static inline int connBlockingConnect(connection *conn, const char *addr, int po
return conn->type->blocking_connect(conn, addr, port, timeout);
}
/* Rebind the connection to another event loop, read/write handlers must not
* be installed in the current event loop */
static inline int connRebindEventLoop(connection *conn, aeEventLoop *el) {
return conn->type->rebind_event_loop(conn, el);
}
/* Write to connection, behaves the same as write(2).
*
* Like write(2), a short write is possible. A -1 return indicates an error.
......@@ -379,14 +389,14 @@ ConnectionType *connectionTypeUnix(void);
int connectionIndexByType(const char *typename);
/* Create a connection of specified type */
static inline connection *connCreate(ConnectionType *ct) {
return ct->conn_create();
static inline connection *connCreate(struct aeEventLoop *el, ConnectionType *ct) {
return ct->conn_create(el);
}
/* 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);
static inline connection *connCreateAccepted(struct aeEventLoop *el, ConnectionType *ct, int fd, void *priv) {
return ct->conn_create_accepted(el, fd, priv);
}
/* Configure a connection type. A typical case is to configure TLS.
......
......@@ -2925,7 +2925,7 @@ write_error: /* Handle sendCommand() errors. */
}
int connectWithMaster(void) {
server.repl_transfer_s = connCreate(connTypeOfReplication());
server.repl_transfer_s = connCreate(server.el, connTypeOfReplication());
if (connConnect(server.repl_transfer_s, server.masterhost, server.masterport,
server.bind_source_addr, syncWithMaster) == C_ERR) {
serverLog(LL_WARNING,"Unable to connect to MASTER: %s",
......
......@@ -53,11 +53,12 @@ static ConnectionType CT_Socket;
* be embedded in different structs, not just client.
*/
static connection *connCreateSocket(void) {
static connection *connCreateSocket(struct aeEventLoop *el) {
connection *conn = zcalloc(sizeof(connection));
conn->type = &CT_Socket;
conn->fd = -1;
conn->iovcnt = IOV_MAX;
conn->el = el;
return conn;
}
......@@ -72,9 +73,9 @@ static connection *connCreateSocket(void) {
* is not in an error state (which is not possible for a socket connection,
* but could but possible with other protocols).
*/
static connection *connCreateAcceptedSocket(int fd, void *priv) {
static connection *connCreateAcceptedSocket(struct aeEventLoop *el, int fd, void *priv) {
UNUSED(priv);
connection *conn = connCreateSocket();
connection *conn = connCreateSocket(el);
conn->fd = fd;
conn->state = CONN_STATE_ACCEPTING;
return conn;
......@@ -93,7 +94,7 @@ static int connSocketConnect(connection *conn, const char *addr, int port, const
conn->state = CONN_STATE_CONNECTING;
conn->conn_handler = connect_handler;
aeCreateFileEvent(server.el, conn->fd, AE_WRITABLE,
aeCreateFileEvent(conn->el, conn->fd, AE_WRITABLE,
conn->type->ae_handler, conn);
return C_OK;
......@@ -114,7 +115,7 @@ static void connSocketShutdown(connection *conn) {
/* Close the connection and free resources. */
static void connSocketClose(connection *conn) {
if (conn->fd != -1) {
aeDeleteFileEvent(server.el,conn->fd, AE_READABLE | AE_WRITABLE);
aeDeleteFileEvent(conn->el, conn->fd, AE_READABLE | AE_WRITABLE);
close(conn->fd);
conn->fd = -1;
}
......@@ -190,6 +191,15 @@ static int connSocketAccept(connection *conn, ConnectionCallbackFunc accept_hand
return ret;
}
/* Rebind the connection to another event loop, read/write handlers must not
* be installed in the current event loop, otherwise it will cause two event
* loops to manage the same connection at the same time. */
static int connSocketRebindEventLoop(connection *conn, aeEventLoop *el) {
serverAssert(!conn->read_handler && !conn->write_handler);
conn->el = el;
return C_OK;
}
/* Register a write handler, to be called when the connection is writable.
* If NULL, the existing handler is removed.
*
......@@ -207,9 +217,9 @@ static int connSocketSetWriteHandler(connection *conn, ConnectionCallbackFunc fu
else
conn->flags &= ~CONN_FLAG_WRITE_BARRIER;
if (!conn->write_handler)
aeDeleteFileEvent(server.el,conn->fd,AE_WRITABLE);
aeDeleteFileEvent(conn->el,conn->fd,AE_WRITABLE);
else
if (aeCreateFileEvent(server.el,conn->fd,AE_WRITABLE,
if (aeCreateFileEvent(conn->el,conn->fd,AE_WRITABLE,
conn->type->ae_handler,conn) == AE_ERR) return C_ERR;
return C_OK;
}
......@@ -222,9 +232,9 @@ static int connSocketSetReadHandler(connection *conn, ConnectionCallbackFunc fun
conn->read_handler = func;
if (!conn->read_handler)
aeDeleteFileEvent(server.el,conn->fd,AE_READABLE);
aeDeleteFileEvent(conn->el,conn->fd,AE_READABLE);
else
if (aeCreateFileEvent(server.el,conn->fd,
if (aeCreateFileEvent(conn->el,conn->fd,
AE_READABLE,conn->type->ae_handler,conn) == AE_ERR) return C_ERR;
return C_OK;
}
......@@ -250,7 +260,7 @@ static void connSocketEventHandler(struct aeEventLoop *el, int fd, void *clientD
conn->state = CONN_STATE_CONNECTED;
}
if (!conn->write_handler) aeDeleteFileEvent(server.el,conn->fd,AE_WRITABLE);
if (!conn->write_handler) aeDeleteFileEvent(conn->el, conn->fd, AE_WRITABLE);
if (!callHandler(conn, conn->conn_handler)) return;
conn->conn_handler = NULL;
......@@ -291,7 +301,6 @@ static void connSocketAcceptHandler(aeEventLoop *el, int fd, void *privdata, int
int cport, cfd;
int max = server.max_new_conns_per_cycle;
char cip[NET_IP_STR_LEN];
UNUSED(el);
UNUSED(mask);
UNUSED(privdata);
......@@ -304,7 +313,7 @@ static void connSocketAcceptHandler(aeEventLoop *el, int fd, void *privdata, int
return;
}
serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport);
acceptCommonHandler(connCreateAcceptedSocket(cfd, NULL),0,cip);
acceptCommonHandler(connCreateAcceptedSocket(el,cfd,NULL), 0, cip);
}
}
......@@ -397,6 +406,9 @@ static ConnectionType CT_Socket = {
.blocking_connect = connSocketBlockingConnect,
.accept = connSocketAccept,
/* event loop */
.rebind_event_loop = connSocketRebindEventLoop,
/* IO */
.write = connSocketWrite,
.writev = connSocketWritev,
......
......@@ -435,20 +435,21 @@ typedef struct tls_connection {
listNode *pending_list_node;
} tls_connection;
static connection *createTLSConnection(int client_side) {
static connection *createTLSConnection(struct aeEventLoop *el, int client_side) {
SSL_CTX *ctx = redis_tls_ctx;
if (client_side && redis_tls_client_ctx)
ctx = redis_tls_client_ctx;
tls_connection *conn = zcalloc(sizeof(tls_connection));
conn->c.type = &CT_TLS;
conn->c.fd = -1;
conn->c.el = el;
conn->c.iovcnt = IOV_MAX;
conn->ssl = SSL_new(ctx);
return (connection *) conn;
}
static connection *connCreateTLS(void) {
return createTLSConnection(1);
static connection *connCreateTLS(struct aeEventLoop *el) {
return createTLSConnection(el, 1);
}
/* Fetch the latest OpenSSL error and store it in the connection */
......@@ -468,10 +469,11 @@ static void updateTLSError(tls_connection *conn) {
* Callers should use connGetState() and verify the created connection
* is not in an error state.
*/
static connection *connCreateAcceptedTLS(int fd, void *priv) {
static connection *connCreateAcceptedTLS(struct aeEventLoop *el, int fd, void *priv) {
int require_auth = *(int *)priv;
tls_connection *conn = (tls_connection *) createTLSConnection(0);
tls_connection *conn = (tls_connection *) createTLSConnection(el, 0);
conn->c.fd = fd;
conn->c.el = el;
conn->c.state = CONN_STATE_ACCEPTING;
if (!conn->ssl) {
......@@ -575,17 +577,17 @@ static int updateStateAfterSSLIO(tls_connection *conn, int ret_value, int update
}
static void registerSSLEvent(tls_connection *conn, WantIOType want) {
int mask = aeGetFileEvents(server.el, conn->c.fd);
int mask = aeGetFileEvents(conn->c.el, conn->c.fd);
switch (want) {
case WANT_READ:
if (mask & AE_WRITABLE) aeDeleteFileEvent(server.el, conn->c.fd, AE_WRITABLE);
if (!(mask & AE_READABLE)) aeCreateFileEvent(server.el, conn->c.fd, AE_READABLE,
if (mask & AE_WRITABLE) aeDeleteFileEvent(conn->c.el, conn->c.fd, AE_WRITABLE);
if (!(mask & AE_READABLE)) aeCreateFileEvent(conn->c.el, conn->c.fd, AE_READABLE,
tlsEventHandler, conn);
break;
case WANT_WRITE:
if (mask & AE_READABLE) aeDeleteFileEvent(server.el, conn->c.fd, AE_READABLE);
if (!(mask & AE_WRITABLE)) aeCreateFileEvent(server.el, conn->c.fd, AE_WRITABLE,
if (mask & AE_READABLE) aeDeleteFileEvent(conn->c.el, conn->c.fd, AE_READABLE);
if (!(mask & AE_WRITABLE)) aeCreateFileEvent(conn->c.el, conn->c.fd, AE_WRITABLE,
tlsEventHandler, conn);
break;
default:
......@@ -595,19 +597,19 @@ static void registerSSLEvent(tls_connection *conn, WantIOType want) {
}
static void updateSSLEvent(tls_connection *conn) {
int mask = aeGetFileEvents(server.el, conn->c.fd);
int mask = aeGetFileEvents(conn->c.el, conn->c.fd);
int need_read = conn->c.read_handler || (conn->flags & TLS_CONN_FLAG_WRITE_WANT_READ);
int need_write = conn->c.write_handler || (conn->flags & TLS_CONN_FLAG_READ_WANT_WRITE);
if (need_read && !(mask & AE_READABLE))
aeCreateFileEvent(server.el, conn->c.fd, AE_READABLE, tlsEventHandler, conn);
aeCreateFileEvent(conn->c.el, conn->c.fd, AE_READABLE, tlsEventHandler, conn);
if (!need_read && (mask & AE_READABLE))
aeDeleteFileEvent(server.el, conn->c.fd, AE_READABLE);
aeDeleteFileEvent(conn->c.el, conn->c.fd, AE_READABLE);
if (need_write && !(mask & AE_WRITABLE))
aeCreateFileEvent(server.el, conn->c.fd, AE_WRITABLE, tlsEventHandler, conn);
aeCreateFileEvent(conn->c.el, conn->c.fd, AE_WRITABLE, tlsEventHandler, conn);
if (!need_write && (mask & AE_WRITABLE))
aeDeleteFileEvent(server.el, conn->c.fd, AE_WRITABLE);
aeDeleteFileEvent(conn->c.el, conn->c.fd, AE_WRITABLE);
}
static void tlsHandleEvent(tls_connection *conn, int mask) {
......@@ -748,7 +750,6 @@ static void tlsAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask)
int cport, cfd;
int max = server.max_new_tls_conns_per_cycle;
char cip[NET_IP_STR_LEN];
UNUSED(el);
UNUSED(mask);
UNUSED(privdata);
......@@ -761,7 +762,7 @@ static void tlsAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask)
return;
}
serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport);
acceptCommonHandler(connCreateAcceptedTLS(cfd, &server.tls_auth_clients),0,cip);
acceptCommonHandler(connCreateAcceptedTLS(el,cfd,&server.tls_auth_clients), 0, cip);
}
}
......@@ -863,6 +864,13 @@ static int connTLSConnect(connection *conn_, const char *addr, int port, const c
return C_OK;
}
static int connTLSRebindEventLoop(connection *conn_, aeEventLoop *el) {
tls_connection *conn = (tls_connection *) conn_;
serverAssert(!conn->c.read_handler && !conn->c.write_handler);
conn->c.el = el;
return C_OK;
}
static int connTLSWrite(connection *conn_, const void *data, size_t data_len) {
tls_connection *conn = (tls_connection *) conn_;
int ret;
......@@ -1114,6 +1122,9 @@ static ConnectionType CT_TLS = {
.blocking_connect = connTLSBlockingConnect,
.accept = connTLSAccept,
/* event loop */
.rebind_event_loop = connTLSRebindEventLoop,
/* IO */
.read = connTLSRead,
.write = connTLSWrite,
......
......@@ -74,18 +74,19 @@ static int connUnixListen(connListener *listener) {
return C_OK;
}
static connection *connCreateUnix(void) {
static connection *connCreateUnix(struct aeEventLoop *el) {
connection *conn = zcalloc(sizeof(connection));
conn->type = &CT_Unix;
conn->fd = -1;
conn->iovcnt = IOV_MAX;
conn->el = el;
return conn;
}
static connection *connCreateAcceptedUnix(int fd, void *priv) {
static connection *connCreateAcceptedUnix(struct aeEventLoop *el, int fd, void *priv) {
UNUSED(priv);
connection *conn = connCreateUnix();
connection *conn = connCreateUnix(el);
conn->fd = fd;
conn->state = CONN_STATE_ACCEPTING;
return conn;
......@@ -107,7 +108,7 @@ static void connUnixAcceptHandler(aeEventLoop *el, int fd, void *privdata, int m
return;
}
serverLog(LL_VERBOSE,"Accepted connection to %s", server.unixsocket);
acceptCommonHandler(connCreateAcceptedUnix(cfd, NULL),CLIENT_UNIX_SOCKET,NULL);
acceptCommonHandler(connCreateAcceptedUnix(el, cfd, NULL),CLIENT_UNIX_SOCKET,NULL);
}
}
......@@ -123,6 +124,10 @@ static int connUnixAccept(connection *conn, ConnectionCallbackFunc accept_handle
return connectionTypeTcp()->accept(conn, accept_handler);
}
static int connUnixRebindEventLoop(connection *conn, aeEventLoop *el) {
return connectionTypeTcp()->rebind_event_loop(conn, el);
}
static int connUnixWrite(connection *conn, const void *data, size_t data_len) {
return connectionTypeTcp()->write(conn, data, data_len);
}
......@@ -186,6 +191,9 @@ static ConnectionType CT_Unix = {
.blocking_connect = NULL,
.accept = connUnixAccept,
/* event loop */
.rebind_event_loop = connUnixRebindEventLoop,
/* IO */
.write = connUnixWrite,
.writev = connUnixWritev,
......
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