Introduce MPTCP for replication (#1961)

Allow replicas to use MPTCP in the outgoing replication connection.

A new yes/no config is introduced `repl-mptcp`, default `no`.

For MPTCP to be used in replication, the primary needs to be configured
with `mptcp yes` and the replica with `repl-mptcp yes`. Otherwise, the
connection falls back to regular TCP.

Follow-up of #1811.

---------

Signed-off-by: zhenwei pi <pizhenwei@bytedance.com>

Author: pizhenwei

src/anet.c

@@ -441,9 +441,25 @@ static int anetCreateSocket(char *err, int domain, int type, int protocol, int f
     return s;
 }
 
+/* XXX: Until glibc 2.41, getaddrinfo with hints.ai_protocol of IPPROTO_MPTCP leads error.
+ * Use hints.ai_protocol IPPROTO_IP (0) or IPPROTO_TCP (6) to resolve address and overwrite
+ * it when MPTCP is enabled.
+ * Ref: https://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git/tree/tools/testing/selftests/net/mptcp/mptcp_connect.c
+ *      https://sourceware.org/git/?p=glibc.git;a=commit;h=a8e9022e0f829d44a818c642fc85b3bfbd26a514
+ */
+static int anetTcpGetProtocol(int is_mptcp_enabled) {
+#ifdef IPPROTO_MPTCP
+    return is_mptcp_enabled ? IPPROTO_MPTCP : IPPROTO_TCP;
+#else
+    assert(!is_mptcp_enabled);
+    return IPPROTO_TCP;
+#endif
+}
+
 #define ANET_CONNECT_NONE 0
 #define ANET_CONNECT_NONBLOCK 1
 #define ANET_CONNECT_BE_BINDING 2 /* Best effort binding. */
+#define ANET_CONNECT_MPTCP 4
 static int anetTcpGenericConnect(char *err, const char *addr, int port, const char *source_addr, int flags) {
     int s = ANET_ERR, rv;
     char portstr[6]; /* strlen("65535") + 1; */
@@ -466,9 +482,10 @@ static int anetTcpGenericConnect(char *err, const char *addr, int port, const ch
          * Make sure connection-intensive things like the benchmark tool
          * will be able to close/open sockets a zillion of times.
          */
+        int ai_protocol = anetTcpGetProtocol(flags & ANET_CONNECT_MPTCP);
         int sockflags = ANET_SOCKET_CLOEXEC | ANET_SOCKET_REUSEADDR;
         if (flags & ANET_CONNECT_NONBLOCK) sockflags |= ANET_SOCKET_NONBLOCK;
-        if ((s = anetCreateSocket(err, p->ai_family, p->ai_socktype, p->ai_protocol, sockflags)) == ANET_ERR) continue;
+        if ((s = anetCreateSocket(err, p->ai_family, p->ai_socktype, ai_protocol, sockflags)) == ANET_ERR) continue;
         if (source_addr) {
             int bound = 0;
             /* Using getaddrinfo saves us from self-determining IPv4 vs IPv6 */
@@ -528,8 +545,10 @@ int anetTcpNonBlockConnect(char *err, const char *addr, int port) {
     return anetTcpGenericConnect(err, addr, port, NULL, ANET_CONNECT_NONBLOCK);
 }
 
-int anetTcpNonBlockBestEffortBindConnect(char *err, const char *addr, int port, const char *source_addr) {
-    return anetTcpGenericConnect(err, addr, port, source_addr, ANET_CONNECT_NONBLOCK | ANET_CONNECT_BE_BINDING);
+int anetTcpNonBlockBestEffortBindConnect(char *err, const char *addr, int port, const char *source_addr, int mptcp) {
+    int flags = ANET_CONNECT_NONBLOCK | ANET_CONNECT_BE_BINDING;
+    if (mptcp) flags |= ANET_CONNECT_MPTCP;
+    return anetTcpGenericConnect(err, addr, port, source_addr, flags);
 }
 
 static int anetListen(char *err, int s, struct sockaddr *sa, socklen_t len, int backlog, mode_t perm, char *group) {
@@ -574,21 +593,6 @@ static int anetV6Only(char *err, int s) {
     return ANET_OK;
 }
 
-/* XXX: Until glibc 2.41, getaddrinfo with hints.ai_protocol of IPPROTO_MPTCP leads error.
- * Use hints.ai_protocol IPPROTO_IP (0) or IPPROTO_TCP (6) to resolve address and overwrite
- * it when MPTCP is enabled.
- * Ref: https://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git/tree/tools/testing/selftests/net/mptcp/mptcp_connect.c
- *      https://sourceware.org/git/?p=glibc.git;a=commit;h=a8e9022e0f829d44a818c642fc85b3bfbd26a514
- */
-static int anetTcpGetProtocol(int is_mptcp_enabled) {
-#ifdef IPPROTO_MPTCP
-    return is_mptcp_enabled ? IPPROTO_MPTCP : IPPROTO_TCP;
-#else
-    assert(!is_mptcp_enabled);
-    return IPPROTO_TCP;
-#endif
-}
-
 static int _anetTcpServer(char *err, int port, char *bindaddr, int af, int backlog, int mptcp) {
     int s = -1, rv;
     char _port[6]; /* strlen("65535") */

src/anet.h

@@ -52,7 +52,7 @@
 #endif
 
 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 mptcp);
 int anetResolve(char *err, char *host, char *ipbuf, size_t ipbuf_len, int flags);
 int anetTcpServer(char *err, int port, char *bindaddr, int backlog, int mptcp);
 int anetTcp6Server(char *err, int port, char *bindaddr, int backlog, int mptcp);

src/cluster_legacy.c

@@ -5315,7 +5315,7 @@ static int clusterNodeCronHandleReconnect(clusterNode *node, mstime_t now) {
         clusterLink *link = createClusterLink(node);
         link->conn = connCreate(connTypeOfCluster());
         connSetPrivateData(link->conn, link);
-        if (connConnect(link->conn, node->ip, node->cport, server.bind_source_addr, clusterLinkConnectHandler) ==
+        if (connConnect(link->conn, node->ip, node->cport, server.bind_source_addr, 0, clusterLinkConnectHandler) ==
             C_ERR) {
             /* We got a synchronous error from connect before
              * clusterSendPing() had a chance to be called.

src/config.c

@@ -3167,6 +3167,7 @@ standardConfig static_configs[] = {
     createBoolConfig("lazyfree-lazy-user-del", NULL, DEBUG_CONFIG | MODIFIABLE_CONFIG, server.lazyfree_lazy_user_del, 1, NULL, NULL),
     createBoolConfig("lazyfree-lazy-user-flush", NULL, DEBUG_CONFIG | MODIFIABLE_CONFIG, server.lazyfree_lazy_user_flush, 1, NULL, NULL),
     createBoolConfig("repl-disable-tcp-nodelay", NULL, MODIFIABLE_CONFIG, server.repl_disable_tcp_nodelay, 0, NULL, NULL),
+    createBoolConfig("repl-mptcp", NULL, IMMUTABLE_CONFIG, server.repl_mptcp, 0, isValidMptcp, NULL),
     createBoolConfig("repl-diskless-sync", NULL, DEBUG_CONFIG | MODIFIABLE_CONFIG, server.repl_diskless_sync, 1, NULL, NULL),
     createBoolConfig("dual-channel-replication-enabled", NULL, DEBUG_CONFIG | MODIFIABLE_CONFIG, server.dual_channel_replication, 0, NULL, NULL),
     createBoolConfig("aof-rewrite-incremental-fsync", NULL, MODIFIABLE_CONFIG, server.aof_rewrite_incremental_fsync, 1, NULL, NULL),

src/connection.h

@@ -94,6 +94,7 @@ typedef struct ConnectionType {
                    const char *addr,
                    int port,
                    const char *source_addr,
+                   int multipath,
                    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);
@@ -188,8 +189,9 @@ static inline int connConnect(connection *conn,
                               const char *addr,
                               int port,
                               const char *src_addr,
+                              int multipath,
                               ConnectionCallbackFunc connect_handler) {
-    return conn->type->connect(conn, addr, port, src_addr, connect_handler);
+    return conn->type->connect(conn, addr, port, src_addr, multipath, connect_handler);
 }
 
 /* Blocking connect.

src/rdma.c

@@ -1184,11 +1184,15 @@ static int connRdmaConnect(connection *conn,
                            const char *addr,
                            int port,
                            const char *src_addr,
+                           int multipath,
                            ConnectionCallbackFunc connect_handler) {
     rdma_connection *rdma_conn = (rdma_connection *)conn;
     struct rdma_cm_id *cm_id;
     RdmaContext *ctx;
 
+    /* RDMA does not support multipath, and there is no outgoing RDMA connection at the current stage */
+    assert(!multipath);
+
     if (rdmaResolveAddr(rdma_conn, addr, port, src_addr) == C_ERR) {
         return C_ERR;
     }

src/replication.c

@@ -3941,7 +3941,7 @@ void syncWithPrimary(connection *conn) {
         /* Create RDB connection */
         server.repl_rdb_transfer_s = connCreate(connTypeOfReplication());
         if (connConnect(server.repl_rdb_transfer_s, server.primary_host, server.primary_port, server.bind_source_addr,
-                        dualChannelFullSyncWithPrimary) == C_ERR) {
+                        server.repl_mptcp, dualChannelFullSyncWithPrimary) == C_ERR) {
             dualChannelServerLog(LL_WARNING, "Unable to connect to Primary: %s",
                                  connGetLastError(server.repl_transfer_s));
             connClose(server.repl_rdb_transfer_s);
@@ -3978,7 +3978,7 @@ void syncWithPrimary(connection *conn) {
 int connectWithPrimary(void) {
     server.repl_transfer_s = connCreate(connTypeOfReplication());
     if (connConnect(server.repl_transfer_s, server.primary_host, server.primary_port, server.bind_source_addr,
-                    syncWithPrimary) == C_ERR) {
+                    server.repl_mptcp, syncWithPrimary) == C_ERR) {
         serverLog(LL_WARNING, "Unable to connect to PRIMARY: %s", connGetLastError(server.repl_transfer_s));
         connClose(server.repl_transfer_s);
         server.repl_transfer_s = NULL;

src/server.h

@@ -1979,6 +1979,7 @@ struct valkeyServer {
     int repl_replica_ignore_maxmemory;  /* If true replicas do not evict. */
     time_t repl_down_since;             /* Unix time at which link with primary went down */
     int repl_disable_tcp_nodelay;       /* Disable TCP_NODELAY after SYNC? */
+    int repl_mptcp;                     /* Use Multipath TCP for replica on client side */
     int replica_priority;               /* Reported in INFO and used by Sentinel. */
     int replica_announced;              /* If true, replica is announced by Sentinel */
     int replica_announce_port;          /* Give the primary this listening port. */

src/socket.c

@@ -106,8 +106,9 @@ static int connSocketConnect(connection *conn,
                              const char *addr,
                              int port,
                              const char *src_addr,
+                             int multipath,
                              ConnectionCallbackFunc connect_handler) {
-    int fd = anetTcpNonBlockBestEffortBindConnect(NULL, addr, port, src_addr);
+    int fd = anetTcpNonBlockBestEffortBindConnect(NULL, addr, port, src_addr, multipath);
     if (fd == -1) {
         conn->state = CONN_STATE_ERROR;
         conn->last_errno = errno;

src/tls.c

@@ -880,6 +880,7 @@ static int connTLSConnect(connection *conn_,
                           const char *addr,
                           int port,
                           const char *src_addr,
+                          int multipath,
                           ConnectionCallbackFunc connect_handler) {
     tls_connection *conn = (tls_connection *)conn_;
     unsigned char addr_buf[sizeof(struct in6_addr)];
@@ -893,7 +894,7 @@ static int connTLSConnect(connection *conn_,
     }
 
     /* Initiate Socket connection first */
-    if (connectionTypeTcp()->connect(conn_, addr, port, src_addr, connect_handler) == C_ERR) return C_ERR;
+    if (connectionTypeTcp()->connect(conn_, addr, port, src_addr, multipath, connect_handler) == C_ERR) return C_ERR;
 
     /* Return now, once the socket is connected we'll initiate
      * TLS connection from the event handler.

tests/unit/introspection.tcl

@@ -921,6 +921,7 @@ start_server {tags {"introspection"}} {
             daemonize
             tcp-backlog
             mptcp
+            repl-mptcp
             always-show-logo
             syslog-enabled
             cluster-enabled

valkey.conf

@@ -830,6 +830,15 @@ dual-channel-replication-enabled no
 # be a good idea.
 repl-disable-tcp-nodelay no
 
+# Enables MPTCP for the replica's connection to the primary.
+#
+# An MPTCP connection is established between the primary and the replica if
+# the replica has set 'repl-mptcp yes' and the primary has set 'mptcp yes'.
+# Otherwise, it will automatically and implicitly fall back to a regular TCP
+# connection.
+#
+# repl-mptcp no
+
 # Set the replication backlog size. The backlog is a buffer that accumulates
 # replica data when replicas are disconnected for some time, so that when a
 # replica wants to reconnect again, often a full resync is not needed, but a