Add Multipath TCP (MPTCP) support for listeners, bridges, libmosquitto and clients

client/client_shared.c

@@ -889,6 +889,8 @@ int client_config_line_proc(struct mosq_config *cfg, int pub_or_sub, int argc, c
 				goto unknown_option;
 			}
 			cfg->message_rate = true;
+		}else if(!strcmp(argv[i], "--mptcp")){
+			cfg->mptcp = true;
 		}else if(!strcmp(argv[i], "--nodelay")){
 			cfg->tcp_nodelay = true;
 		}else if(!strcmp(argv[i], "--no-tls")){
@@ -1425,9 +1427,7 @@ static int client_tls_opts_set(struct mosquitto *mosq, struct mosq_config *cfg)
 
 int client_opts_set(struct mosquitto *mosq, struct mosq_config *cfg)
 {
-#if defined(WITH_SOCKS)
 	int rc;
-#endif
 
 	mosquitto_int_option(mosq, MOSQ_OPT_PROTOCOL_VERSION, cfg->protocol_version);
 	mosquitto_int_option(mosq, MOSQ_OPT_TRANSPORT, cfg->transport);
@@ -1462,6 +1462,13 @@ int client_opts_set(struct mosquitto *mosq, struct mosq_config *cfg)
 	if(cfg->tcp_nodelay){
 		mosquitto_int_option(mosq, MOSQ_OPT_TCP_NODELAY, 1);
 	}
+	if(cfg->mptcp){
+		rc = mosquitto_int_option(mosq, MOSQ_OPT_MPTCP, 1);
+		if(rc){
+			err_printf(cfg, "Error: MPTCP is not supported on this platform.\n");
+			return rc;
+		}
+	}
 
 	if(cfg->msg_count > 0 && cfg->msg_count < 20){
 		/* 20 is the default "receive maximum"

client/client_shared.h

@@ -141,6 +141,7 @@ struct mosq_config {
 	char *options_file;
 	bool have_topic_alias; /* pub */
 	bool tcp_nodelay;
+	bool mptcp;
 	bool no_tls;
 	bool message_rate; /* sub */
 	bool measure_latency; /* rr */

client/pub_client.c

@@ -416,9 +416,9 @@ static void print_usage(void)
 	printf("                     {-f file | -l | -n | -m message}\n");
 	printf("                     [-c] [-k keepalive] [-q qos] [-r] [--repeat N] [--repeat-delay time] [-x session-expiry]\n");
 #ifdef WITH_SRV
-	printf("                     [-A bind_address] [--nodelay] [-S]\n");
+	printf("                     [-A bind_address] [--mptcp] [--nodelay] [-S]\n");
 #else
-	printf("                     [-A bind_address] [--nodelay]\n");
+	printf("                     [-A bind_address] [--mptcp] [--nodelay]\n");
 #endif
 	printf("                     [-i id] [-I id_prefix]\n");
 	printf("                     [-d] [--quiet]\n");
@@ -484,6 +484,7 @@ static void print_usage(void)
 	printf("      seconds after the client disconnects, or use -1, 4294967295, or ∞ for a session\n");
 	printf("      that does not expire. Defaults to -1 if -c is also given, or 0 if -c not given.\n");
 	printf(" --help : display this message.\n");
+	printf(" --mptcp : use Multipath TCP to connect to the broker, if available. Linux only.\n");
 	printf(" --nodelay : disable Nagle's algorithm, to reduce socket sending latency at the possible\n");
 	printf("             expense of more packets being sent.\n");
 	printf(" --quiet : don't print error messages.\n");

client/rr_client.c

@@ -176,9 +176,9 @@ static void print_usage(void)
 	printf("                    [-W timeout_secs]\n");
 #endif
 #ifdef WITH_SRV
-	printf("                    [-A bind_address] [--nodelay] [-S]\n");
+	printf("                    [-A bind_address] [--mptcp] [--nodelay] [-S]\n");
 #else
-	printf("                    [-A bind_address] [--nodelay]\n");
+	printf("                    [-A bind_address] [--mptcp] [--nodelay]\n");
 #endif
 	printf("                    [-i id] [-I id_prefix]\n");
 	printf("                    [-d] [-N] [--quiet] [-v]\n");
@@ -240,6 +240,7 @@ static void print_usage(void)
 	printf("      seconds after the client disconnects, or use -1, 4294967295, or ∞ for a session\n");
 	printf("      that does not expire. Defaults to -1 if -c is also given, or 0 if -c not given.\n");
 	printf(" --help : display this message.\n");
+	printf(" --mptcp : use Multipath TCP to connect to the broker, if available. Linux only.\n");
 	printf(" --nodelay : disable Nagle's algorithm, to reduce socket sending latency at the possible\n");
 	printf("             expense of more packets being sent.\n");
 	printf(" --pretty : print formatted output rather than minimised output when using the\n");

client/sub_client.c

@@ -244,9 +244,9 @@ static void print_usage(void)
 	printf("                     [-W timeout_secs]\n");
 #endif
 #ifdef WITH_SRV
-	printf("                     [-A bind_address] [--nodelay] [-S]\n");
+	printf("                     [-A bind_address] [--mptcp] [--nodelay] [-S]\n");
 #else
-	printf("                     [-A bind_address] [--nodelay]\n");
+	printf("                     [-A bind_address] [--mptcp] [--nodelay]\n");
 #endif
 	printf("                     [-i id] [-I id_prefix]\n");
 	printf("                     [-d] [-N] [--quiet] [-v] [-w|--watch]\n");
@@ -316,6 +316,7 @@ static void print_usage(void)
 	printf("      seconds after the client disconnects, or use -1, 4294967295, or ∞ for a session\n");
 	printf("      that does not expire. Defaults to -1 if -c is also given, or 0 if -c not given.\n");
 	printf(" --help : display this message.\n");
+	printf(" --mptcp : use Multipath TCP to connect to the broker, if available. Linux only.\n");
 	printf(" --nodelay : disable Nagle's algorithm, to reduce socket sending latency at the possible\n");
 	printf("             expense of more packets being sent.\n");
 	printf(" --pretty : print formatted output rather than minimised output when using the\n");

include/mosquitto/libmosquitto.h

@@ -91,6 +91,7 @@ enum mosq_opt_t {
 	MOSQ_OPT_TRANSPORT = 15,
 	MOSQ_OPT_HTTP_PATH = 16,
 	MOSQ_OPT_HTTP_HEADER_SIZE = 17,
+	MOSQ_OPT_MPTCP = 18,
 };
 
 /* Struct: mosquitto_message

include/mosquitto/libmosquitto_options.h

@@ -90,6 +90,14 @@ libmosq_EXPORT int mosquitto_opts_set(struct mosquitto *mosq, enum mosq_opt_t op
  *	          packets being sent.
  *	          Defaults to 0, which means Nagle remains enabled.
  *
+ *	MOSQ_OPT_MPTCP - Set to 1 to use Multipath TCP (MPTCP) instead of plain
+ *	          TCP when connecting to the broker. This is currently only
+ *	          supported on Linux, with kernel 5.6 or later. If the running
+ *	          kernel does not support MPTCP, the connection will fall back
+ *	          to plain TCP. On other platforms this option returns
+ *	          MOSQ_ERR_NOT_SUPPORTED. Must be set before the client
+ *	          connects. Defaults to 0, which means plain TCP is used.
+ *
  *	MOSQ_OPT_PROTOCOL_VERSION - Value must be set to either MQTT_PROTOCOL_V31,
  *	          MQTT_PROTOCOL_V311, or MQTT_PROTOCOL_V5. Must be set before the
  *	          client connects.  Defaults to MQTT_PROTOCOL_V311.

lib/mosquitto_internal.h

@@ -452,6 +452,7 @@ struct mosquitto {
 	uint8_t max_qos;
 	uint8_t retain_available;
 	bool tcp_nodelay;
+	bool mptcp;
 #if defined(WITH_WEBSOCKETS) && WITH_WEBSOCKETS == WS_IS_BUILTIN
 	char *http_request;
 #endif

lib/net_mosq.c

@@ -367,7 +367,7 @@ int net__try_connect_step2(struct mosquitto *mosq, uint16_t port, mosq_sock_t *s
 	ainfo = mosq->adns->ar_result;
 
 	for(rp = ainfo; rp != NULL; rp = rp->ai_next){
-		*sock = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
+		*sock = net__socket_stream(rp->ai_family, rp->ai_socktype, rp->ai_protocol, mosq->mptcp);
 		if(*sock == INVALID_SOCKET){
 			continue;
 		}
@@ -421,7 +421,31 @@ int net__try_connect_step2(struct mosquitto *mosq, uint16_t port, mosq_sock_t *s
 #endif
 
 
-static int net__try_connect_tcp(const char *host, uint16_t port, mosq_sock_t *sock, const char *bind_address, bool blocking)
+/* Create a stream socket, optionally attempting to use MPTCP instead of
+ * plain TCP. If MPTCP is requested but not supported by the running kernel,
+ * fall back to creating a plain TCP socket. */
+mosq_sock_t net__socket_stream(int domain, int type, int protocol, bool use_mptcp)
+{
+#if defined(__linux__)
+	if(use_mptcp){
+		mosq_sock_t sock;
+
+		sock = socket(domain, type, IPPROTO_MPTCP);
+		if(sock != INVALID_SOCKET
+				|| (errno != EINVAL && errno != EPROTONOSUPPORT && errno != ENOPROTOOPT)){
+
+			return sock;
+		}
+		/* MPTCP is not available, fall through to plain TCP. */
+	}
+#else
+	UNUSED(use_mptcp);
+#endif
+	return socket(domain, type, protocol);
+}
+
+
+static int net__try_connect_tcp(const char *host, uint16_t port, mosq_sock_t *sock, const char *bind_address, bool blocking, bool use_mptcp)
 {
 	struct addrinfo hints;
 	struct addrinfo *ainfo, *rp;
@@ -452,7 +476,7 @@ static int net__try_connect_tcp(const char *host, uint16_t port, mosq_sock_t *so
 	}
 
 	for(rp = ainfo; rp != NULL; rp = rp->ai_next){
-		*sock = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
+		*sock = net__socket_stream(rp->ai_family, rp->ai_socktype, rp->ai_protocol, use_mptcp);
 		if(*sock == INVALID_SOCKET){
 			continue;
 		}
@@ -556,16 +580,18 @@ static int net__try_connect_unix(const char *host, mosq_sock_t *sock)
 #endif
 
 
-int net__try_connect(const char *host, uint16_t port, mosq_sock_t *sock, const char *bind_address, bool blocking)
+int net__try_connect(const char *host, uint16_t port, mosq_sock_t *sock, const char *bind_address, bool blocking, bool use_mptcp)
 {
 	if(port == 0){
 #ifdef WITH_UNIX_SOCKETS
+		UNUSED(use_mptcp);
 		return net__try_connect_unix(host, sock);
 #else
+		UNUSED(use_mptcp);
 		return MOSQ_ERR_NOT_SUPPORTED;
 #endif
 	}else{
-		return net__try_connect_tcp(host, port, sock, bind_address, blocking);
+		return net__try_connect_tcp(host, port, sock, bind_address, blocking, use_mptcp);
 	}
 }
 
@@ -969,7 +995,7 @@ int net__socket_connect(struct mosquitto *mosq, const char *host, uint16_t port,
 		return MOSQ_ERR_INVAL;
 	}
 
-	rc = net__try_connect(host, port, &mosq->sock, bind_address, blocking);
+	rc = net__try_connect(host, port, &mosq->sock, bind_address, blocking, mosq->mptcp);
 	if(rc > 0){
 		return rc;
 	}

lib/net_mosq.h

@@ -19,6 +19,7 @@ SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
 #define NET_MOSQ_H
 
 #ifndef WIN32
+#  include <netinet/in.h>
 #  include <sys/socket.h>
 #  include <unistd.h>
 #else
@@ -29,6 +30,13 @@ typedef SSIZE_T ssize_t;
 #  endif
 #endif
 
+#if defined(__linux__) && !defined(IPPROTO_MPTCP)
+/* Multipath TCP is supported by Linux 5.6 and later. The protocol number is
+ * part of the kernel ABI, so it is safe to define it here for the case where
+ * we are building against older headers. */
+#  define IPPROTO_MPTCP 262
+#endif
+
 #include "mosquitto_internal.h"
 #include "mosquitto.h"
 
@@ -72,7 +80,8 @@ void net__init_tls(void);
 int net__socket_connect(struct mosquitto *mosq, const char *host, uint16_t port, const char *bind_address, bool blocking);
 int net__socket_close(struct mosquitto *mosq);
 int net__socket_shutdown(struct mosquitto *mosq);
-int net__try_connect(const char *host, uint16_t port, mosq_sock_t *sock, const char *bind_address, bool blocking);
+int net__try_connect(const char *host, uint16_t port, mosq_sock_t *sock, const char *bind_address, bool blocking, bool use_mptcp);
+mosq_sock_t net__socket_stream(int domain, int type, int protocol, bool use_mptcp);
 int net__try_connect_step1(struct mosquitto *mosq, const char *host);
 int net__try_connect_step2(struct mosquitto *mosq, uint16_t port, mosq_sock_t *sock);
 int net__socket_connect_step3(struct mosquitto *mosq, const char *host);

lib/options.c

@@ -569,6 +569,14 @@ int mosquitto_int_option(struct mosquitto *mosq, enum mosq_opt_t option, int val
 			mosq->tcp_nodelay = (bool)value;
 			break;
 
+		case MOSQ_OPT_MPTCP:
+#if defined(__linux__)
+			mosq->mptcp = (bool)value;
+#else
+			return MOSQ_ERR_NOT_SUPPORTED;
+#endif
+			break;
+
 		case MOSQ_OPT_TRANSPORT:
 #if defined(WITH_WEBSOCKETS) && WITH_WEBSOCKETS == WS_IS_BUILTIN
 			if(value == mosq_t_tcp || value == mosq_t_ws){

man/common/option-mptcp.xml

@@ -0,0 +1,14 @@
+<varlistentry>
+	<term><option>--mptcp</option></term>
+	<listitem>
+		<para>
+			Use Multipath TCP (MPTCP) instead of plain TCP for the connection
+			to the broker. MPTCP allows the connection to make use of multiple
+			network paths simultaneously, which can improve throughput and
+			resilience to network failures, provided the broker also accepts
+			MPTCP connections. This option is only available on Linux, with
+			kernel 5.6 or later. If the running kernel does not support MPTCP,
+			the connection falls back to plain TCP.
+		</para>
+	</listitem>
+</varlistentry>

man/mosquitto.conf.5.xml

@@ -1603,6 +1603,26 @@ accept_protocol_versions 3, 4</programlisting>
 						<para>Not reloaded on reload signal.</para>
 					</listitem>
 				</varlistentry>
+				<varlistentry>
+					<term><option>mptcp</option> [ true | false ]</term>
+					<listitem>
+						<para>Set <option>mptcp</option> to true to make the
+							listener accept connections using Multipath TCP
+							(MPTCP) as well as plain TCP. MPTCP allows a
+							connection to make use of multiple network paths
+							simultaneously, which can improve throughput and
+							resilience to network failures for clients that
+							also use MPTCP. Clients connecting with plain TCP
+							are unaffected and continue to work as normal.
+							</para>
+						<para>This option is only available on Linux, with
+							kernel 5.6 or later. If the running kernel does
+							not support MPTCP, the listener will fall back to
+							plain TCP and a warning will be logged.</para>
+						<para>Defaults to false.</para>
+						<para>Not reloaded on reload signal.</para>
+					</listitem>
+				</varlistentry>
 				<varlistentry>
 					<term><option>port</option> <replaceable>port number</replaceable></term>
 					<listitem>
@@ -2128,6 +2148,22 @@ openssl dhparam -out dhparam.pem 2048</programlisting>
 					aliases completely.</para>
 				</listitem>
 			</varlistentry>
+			<varlistentry>
+				<term><option>bridge_mptcp</option> [ true | false ]</term>
+				<listitem>
+					<para>Set <option>bridge_mptcp</option> to true to make the
+						bridge connect to the remote broker using Multipath TCP
+						(MPTCP) instead of plain TCP. MPTCP allows the
+						connection to make use of multiple network paths
+						simultaneously, which can improve throughput and
+						resilience to network failures, provided the remote
+						broker also accepts MPTCP connections.</para>
+					<para>This option is only available on Linux, with kernel
+						5.6 or later. If the running kernel does not support
+						MPTCP, the connection falls back to plain TCP.</para>
+					<para>Defaults to false.</para>
+				</listitem>
+			</varlistentry>
 			<varlistentry>
 				<term><option>bridge_outgoing_retain</option> [ true | false ]</term>
 				<listitem>

man/mosquitto_pub.1.xml

@@ -53,6 +53,7 @@
 			</group>
 			<sbr/>
 			<arg><option>-A</option> <replaceable>bind-address</replaceable></arg>
+			<arg><option>--mptcp</option></arg>
 			<arg><option>--nodelay</option></arg>
 			<arg><option>-S</option></arg>
 			<arg><option>--ws</option></arg>
@@ -188,6 +189,7 @@
 			</varlistentry>
 			<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-payload-message.xml" />
 			<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-payload-null.xml" />
+			<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-mptcp.xml" />
 			<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-nodelay.xml" />
 			<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-no-tls.xml" />
 			<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-port.xml" />

man/mosquitto_rr.1.xml

@@ -56,6 +56,7 @@
 			</group>
 			<sbr/>
 			<arg><option>-A</option> <replaceable>bind-address</replaceable></arg>
+			<arg><option>--mptcp</option></arg>
 			<arg><option>--nodelay</option></arg>
 			<arg><option>-S</option></arg>
 			<arg><option>--ws</option></arg>
@@ -226,6 +227,7 @@
 			<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-payload-message.xml" />
 			<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-format-no-eol.xml" />
 			<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-payload-null.xml" />
+			<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-mptcp.xml" />
 			<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-nodelay.xml" />
 			<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-no-tls.xml" />
 			<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-port.xml" />

man/mosquitto_sub.1.xml

@@ -57,6 +57,7 @@
 			</group>
 			<sbr/>
 			<arg><option>-A</option> <replaceable>bind-address</replaceable></arg>
+			<arg><option>--mptcp</option></arg>
 			<arg><option>--nodelay</option></arg>
 			<arg><option>-S</option></arg>
 			<arg><option>--ws</option></arg>
@@ -237,6 +238,7 @@
 				</listitem>
 			</varlistentry>
 			<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-format-no-eol.xml" />
+			<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-mptcp.xml" />
 			<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-nodelay.xml" />
 			<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-no-tls.xml" />
 			<xi:include xmlns:xi="http://www.w3.org/2001/XInclude" href="common/option-port.xml" />