Introduce MPTCP (valkey-io#1811)

Multipath TCP (MPTCP) is an extension of the standard TCP protocol that
allows a single transport connection to use multiple network interfaces
or paths. MPTCP is useful for applications like bandwidth aggregation,
failover, and more resilient connections.

Linux kernel starts to support MPTCP since v5.6, it's time to support
it.

The test report shows that MPTCP reduces latency by ~25% in a 1%
networking packet drop environment.

Thanks to Matthieu Baerts <[email protected]> for lots of review
suggestions.

Proposed-by: Geliang Tang <[email protected]>
Tested-by: Gang Yan <[email protected]>
Signed-off-by: zhenwei pi <[email protected]>
Signed-off-by: zhenwei pi <[email protected]>

Cc Linux kernel MPTCP maintainer @matttbe

Signed-off-by: hwware <[email protected]>

Author: pizhenwei

src/anet.c

@@ -50,6 +50,7 @@
 #include "anet.h"
 #include "config.h"
 #include "util.h"
+#include "serverassert.h"
 
 #define UNUSED(x) (void)(x)
 
@@ -573,7 +574,22 @@ static int anetV6Only(char *err, int s) {
     return ANET_OK;
 }
 
-static int _anetTcpServer(char *err, int port, char *bindaddr, int af, int backlog) {
+/* 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") */
     struct addrinfo hints, *servinfo, *p;
@@ -591,7 +607,7 @@ static int _anetTcpServer(char *err, int port, char *bindaddr, int af, int backl
         return ANET_ERR;
     }
     for (p = servinfo; p != NULL; p = p->ai_next) {
-        if ((s = socket(p->ai_family, p->ai_socktype, p->ai_protocol)) == -1) continue;
+        if ((s = socket(p->ai_family, p->ai_socktype, anetTcpGetProtocol(mptcp))) == -1) continue;
 
         if (af == AF_INET6 && anetV6Only(err, s) == ANET_ERR) goto error;
         if (anetSetReuseAddr(err, s) == ANET_ERR) goto error;
@@ -611,12 +627,12 @@ static int _anetTcpServer(char *err, int port, char *bindaddr, int af, int backl
     return s;
 }
 
-int anetTcpServer(char *err, int port, char *bindaddr, int backlog) {
-    return _anetTcpServer(err, port, bindaddr, AF_INET, backlog);
+int anetTcpServer(char *err, int port, char *bindaddr, int backlog, int mptcp) {
+    return _anetTcpServer(err, port, bindaddr, AF_INET, backlog, mptcp);
 }
 
-int anetTcp6Server(char *err, int port, char *bindaddr, int backlog) {
-    return _anetTcpServer(err, port, bindaddr, AF_INET6, backlog);
+int anetTcp6Server(char *err, int port, char *bindaddr, int backlog, int mptcp) {
+    return _anetTcpServer(err, port, bindaddr, AF_INET6, backlog, mptcp);
 }
 
 int anetUnixServer(char *err, char *path, mode_t perm, int backlog, char *group) {

src/anet.h

@@ -54,8 +54,8 @@
 int anetTcpNonBlockConnect(char *err, const char *addr, int port);
 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 anetTcpServer(char *err, int port, char *bindaddr, int backlog);
-int anetTcp6Server(char *err, int port, char *bindaddr, int backlog);
+int anetTcpServer(char *err, int port, char *bindaddr, int backlog, int mptcp);
+int anetTcp6Server(char *err, int port, char *bindaddr, int backlog, int mptcp);
 int anetUnixServer(char *err, char *path, mode_t perm, int backlog, char *group);
 int anetTcpAccept(char *err, int serversock, char *ip, size_t ip_len, int *port);
 int anetUnixAccept(char *err, int serversock);
@@ -75,4 +75,12 @@ int anetSetSockMarkId(char *err, int fd, uint32_t id);
 int anetGetError(int fd);
 int anetIsFifo(char *filepath);
 
+static inline int anetHasMptcp(void) {
+#ifdef IPPROTO_MPTCP
+    return 1;
+#else
+    return 0;
+#endif
+}
+
 #endif

src/config.c

@@ -2427,6 +2427,15 @@ static int isValidIpV6(char *val, const char **err) {
     return 1;
 }
 
+static int isValidMptcp(int val, const char **err) {
+    if (val && !anetHasMptcp()) {
+        *err = "MPTCP is not supported on this platform";
+        return 0;
+    }
+
+    return 1;
+}
+
 /* Validate specified string is a valid proc-title-template */
 static int isValidProcTitleTemplate(char *val, const char **err) {
     if (!validateProcTitleTemplate(val)) {
@@ -3277,6 +3286,7 @@ standardConfig static_configs[] = {
     createIntConfig("timeout", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, server.maxidletime, 0, INTEGER_CONFIG, NULL, NULL), /* Default client timeout: infinite */
     createIntConfig("replica-announce-port", "slave-announce-port", MODIFIABLE_CONFIG, 0, 65535, server.replica_announce_port, 0, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("tcp-backlog", NULL, IMMUTABLE_CONFIG, 0, INT_MAX, server.tcp_backlog, 511, INTEGER_CONFIG, NULL, NULL), /* TCP listen backlog. */
+    createBoolConfig("mptcp", NULL, IMMUTABLE_CONFIG, server.mptcp, 0, isValidMptcp, NULL),                                  /* Multipath TCP. */
     createIntConfig("cluster-port", NULL, IMMUTABLE_CONFIG, 0, 65535, server.cluster_port, 0, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("cluster-announce-bus-port", NULL, MODIFIABLE_CONFIG, 0, 65535, server.cluster_announce_bus_port, 0, INTEGER_CONFIG, NULL, updateClusterAnnouncedPort), /* Default: Use +10000 offset. */
     createIntConfig("cluster-announce-port", NULL, MODIFIABLE_CONFIG, 0, 65535, server.cluster_announce_port, 0, INTEGER_CONFIG, NULL, updateClusterAnnouncedPort),         /* Use server.port */

src/server.c

@@ -2631,10 +2631,10 @@ int listenToPort(connListener *sfd) {
         if (optional) addr++;
         if (strchr(addr, ':')) {
             /* Bind IPv6 address. */
-            sfd->fd[sfd->count] = anetTcp6Server(server.neterr, port, addr, server.tcp_backlog);
+            sfd->fd[sfd->count] = anetTcp6Server(server.neterr, port, addr, server.tcp_backlog, server.mptcp);
         } else {
             /* Bind IPv4 address. */
-            sfd->fd[sfd->count] = anetTcpServer(server.neterr, port, addr, server.tcp_backlog);
+            sfd->fd[sfd->count] = anetTcpServer(server.neterr, port, addr, server.tcp_backlog, server.mptcp);
         }
         if (sfd->fd[sfd->count] == ANET_ERR) {
             int net_errno = errno;

src/server.h

@@ -1613,6 +1613,7 @@ struct valkeyServer {
     int port;                              /* TCP listening port */
     int tls_port;                          /* TLS listening port */
     int tcp_backlog;                       /* TCP listen() backlog */
+    int mptcp;                             /* Use Multipath TCP */
     char *bindaddr[CONFIG_BINDADDR_MAX];   /* Addresses we should bind to */
     int bindaddr_count;                    /* Number of addresses in server.bindaddr[] */
     char *bind_source_addr;                /* Source address to bind on for outgoing connections */

tests/unit/introspection.tcl

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

valkey.conf

@@ -147,6 +147,16 @@ port 6379
 # in order to get the desired effect.
 tcp-backlog 511
 
+# Multipath TCP (MPTCP)
+#
+# MPTCP splits a single TCP connection into subflows over multiple interfaces or paths.
+# It enables bandwidth aggregation, failover, and improved reliability.
+# When set to 'yes', clients will be able to use MPTCP if requested. When not
+# requested, regular TCP can be used like before.
+# Note: MPTCP is supported in the mainline Linux kernel starting from version 5.6.
+#
+# mptcp yes
+
 # Unix socket.
 #
 # Specify the path for the Unix socket that will be used to listen for