Support MPTCP for valkey-cli and benchmark (#2067)

Add '--mptcp' for both valkey-cli and valkey-benchmark to enable MPTCP.

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

Author: pizhenwei

src/cli_common.c

@@ -430,7 +430,7 @@ sds cliVersion(void) {
 }
 
 /* This is a wrapper to call valkeyConnect or valkeyConnectWithTimeout. */
-valkeyContext *valkeyConnectWrapper(enum valkeyConnectionType ct, const char *ip_or_path, int port, const struct timeval tv, int nonblock) {
+valkeyContext *valkeyConnectWrapper(enum valkeyConnectionType ct, const char *ip_or_path, int port, const struct timeval tv, int nonblock, int multipath) {
     valkeyOptions options = {0};
 
     switch (ct) {
@@ -462,5 +462,10 @@ valkeyContext *valkeyConnectWrapper(enum valkeyConnectionType ct, const char *ip
         options.options |= VALKEY_OPT_NONBLOCK;
     }
 
+    if (multipath) {
+        assert(ct == VALKEY_CONN_TCP); /* caller should make sure multipath is available for TCP only */
+        options.options |= VALKEY_OPT_MPTCP;
+    }
+
     return valkeyConnectWithOptions(&options);
 }

src/cli_common.h

@@ -53,6 +53,6 @@ sds escapeJsonString(sds s, const char *p, size_t len);
 
 sds cliVersion(void);
 
-valkeyContext *valkeyConnectWrapper(enum valkeyConnectionType ct, const char *ip_or_path, int port, const struct timeval tv, int nonblock);
+valkeyContext *valkeyConnectWrapper(enum valkeyConnectionType ct, const char *ip_or_path, int port, const struct timeval tv, int nonblock, int multipath);
 
 #endif /* __CLICOMMON_H */

src/valkey-benchmark.c

@@ -91,6 +91,7 @@ static struct config {
     enum valkeyConnectionType ct;
     cliConnInfo conn_info;
     int tls;
+    int mptcp;
     struct cliSSLconfig sslconfig;
     int numclients;
     _Atomic int liveclients;
@@ -248,7 +249,7 @@ static valkeyContext *getValkeyContext(enum valkeyConnectionType ct, const char
     valkeyContext *ctx = NULL;
     valkeyReply *reply = NULL;
     struct timeval tv = {0};
-    ctx = valkeyConnectWrapper(ct, ip_or_path, port, tv, 0);
+    ctx = valkeyConnectWrapper(ct, ip_or_path, port, tv, 0, config.mptcp);
     if (ctx == NULL || ctx->err) {
         fprintf(stderr, "Could not connect to server at ");
         char *err = (ctx != NULL ? ctx->errstr : "");
@@ -666,7 +667,7 @@ static client createClient(char *cmd, size_t len, client from, int thread_id) {
         c->cluster_node = node;
     }
 
-    c->context = valkeyConnectWrapper(config.ct, ip, port, tv, 1);
+    c->context = valkeyConnectWrapper(config.ct, ip, port, tv, 1, config.mptcp);
     if (c->context->err) {
         fprintf(stderr, "Could not connect to server at ");
         if (config.ct != VALKEY_CONN_UNIX || is_cluster_client)
@@ -1490,6 +1491,8 @@ int parseOptions(int argc, char **argv) {
             }
             config.ct = VALKEY_CONN_RDMA;
 #endif
+        } else if (!strcmp(argv[i], "--mptcp")) {
+            config.mptcp = 1;
         } else {
             /* Assume the user meant to provide an option when the arg starts
              * with a dash. We're done otherwise and should use the remainder
@@ -1602,6 +1605,7 @@ int parseOptions(int argc, char **argv) {
         "                    the 'fcall' test. (default 1)\n",
         tls_usage,
         rdma_usage,
+        " --mptcp            Enable an MPTCP connection.\n"
         " --help             Output this help and exit.\n"
         " --version          Output version and exit.\n\n"
         "Examples:\n\n"
@@ -1780,6 +1784,11 @@ int main(int argc, char **argv) {
     }
 #endif
 
+    if (config.mptcp && (config.ct != VALKEY_CONN_TCP)) {
+        fprintf(stderr, "Options --mptcp is only supported by TCP\n");
+        exit(1);
+    }
+
     if (config.cluster_mode) {
         // We only include the slot placeholder {tag} if cluster mode is enabled
         tag = ":{tag}";

src/valkey-cli.c

@@ -219,6 +219,7 @@ static struct config {
     cliConnInfo conn_info; /* conn_info.hostip is used as unix socket path on ct == VALKEY_CONN_UNIX */
     struct timeval connect_timeout;
     int tls;
+    int mptcp;
     cliSSLconfig sslconfig;
     long repeat;
     long interval;
@@ -1639,7 +1640,7 @@ static int cliConnect(int flags) {
             cliRefreshPrompt();
         }
 
-        context = valkeyConnectWrapper(config.ct, config.conn_info.hostip, config.conn_info.hostport, config.connect_timeout, 0);
+        context = valkeyConnectWrapper(config.ct, config.conn_info.hostip, config.conn_info.hostport, config.connect_timeout, 0, config.mptcp);
 
         if (!context->err && config.tls) {
             const char *err = NULL;
@@ -2526,7 +2527,7 @@ static valkeyReply *reconnectingValkeyCommand(valkeyContext *c, const char *fmt,
             fflush(stdout);
 
             valkeyFree(c);
-            c = valkeyConnectWrapper(config.ct, config.conn_info.hostip, config.conn_info.hostport, config.connect_timeout, 0);
+            c = valkeyConnectWrapper(config.ct, config.conn_info.hostip, config.conn_info.hostport, config.connect_timeout, 0, config.mptcp);
             if (!c->err && config.tls) {
                 const char *err = NULL;
                 if (cliSecureConnection(c, config.sslconfig, &err) == VALKEY_ERR && err) {
@@ -2831,6 +2832,8 @@ static int parseOptions(int argc, char **argv) {
             }
             config.ct = VALKEY_CONN_RDMA;
 #endif
+        } else if (!strcmp(argv[i], "--mptcp")) {
+            config.mptcp = 1;
         } else if (!strcmp(argv[i], "-v") || !strcmp(argv[i], "--version")) {
             sds version = cliVersion();
             printf("valkey-cli %s\n", version);
@@ -2910,6 +2913,11 @@ static int parseOptions(int argc, char **argv) {
         exit(1);
     }
 
+    if (config.mptcp && (config.ct != VALKEY_CONN_TCP)) {
+        fprintf(stderr, "Options --mptcp is only supported by TCP.\n");
+        exit(1);
+    }
+
     return i;
 }
 
@@ -3001,6 +3009,7 @@ static void usage(int err) {
             "  -6                 Prefer IPv6 over IPv4 on DNS lookup.\n"
             "%s"
             "%s"
+            "  --mptcp            Enable an MPTCP connection.\n"
             "  --raw              Use raw formatting for replies (default when STDOUT is\n"
             "                     not a tty).\n"
             "  --no-raw           Force formatted output even when STDOUT is not a tty.\n"
@@ -3985,7 +3994,7 @@ static int clusterManagerExecTransaction(clusterManagerNode *node, clusterManage
 
 static int clusterManagerNodeConnect(clusterManagerNode *node) {
     if (node->context) valkeyFree(node->context);
-    node->context = valkeyConnectWrapper(config.ct, node->ip, node->port, config.connect_timeout, 0);
+    node->context = valkeyConnectWrapper(config.ct, node->ip, node->port, config.connect_timeout, 0, config.mptcp);
     if (!node->context->err && config.tls) {
         const char *err = NULL;
         if (cliSecureConnection(node->context, config.sslconfig, &err) == VALKEY_ERR && err) {
@@ -7727,7 +7736,7 @@ static int clusterManagerCommandImport(int argc, char **argv) {
     char *reply_err = NULL;
     valkeyReply *src_reply = NULL;
     // Connect to the source node.
-    valkeyContext *src_ctx = valkeyConnectWrapper(config.ct, src_ip, src_port, config.connect_timeout, 0);
+    valkeyContext *src_ctx = valkeyConnectWrapper(config.ct, src_ip, src_port, config.connect_timeout, 0, config.mptcp);
     if (src_ctx->err) {
         success = 0;
         fprintf(stderr, "Could not connect to Valkey at %s:%d: %s.\n", src_ip, src_port, src_ctx->errstr);