diff --git a/client/client_shared.c b/client/client_shared.c index 91827524a4..b783122d18 100644 --- a/client/client_shared.c +++ b/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" diff --git a/client/client_shared.h b/client/client_shared.h index 7a99c554cb..b1e165bc35 100644 --- a/client/client_shared.h +++ b/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 */ diff --git a/client/pub_client.c b/client/pub_client.c index 7254734d0b..32ed6ae061 100644 --- a/client/pub_client.c +++ b/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"); diff --git a/client/rr_client.c b/client/rr_client.c index f7f090f7b1..d08155ecc6 100644 --- a/client/rr_client.c +++ b/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"); diff --git a/client/sub_client.c b/client/sub_client.c index 3e583ce59c..ef32c77a1a 100644 --- a/client/sub_client.c +++ b/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"); diff --git a/include/mosquitto/libmosquitto.h b/include/mosquitto/libmosquitto.h index 770a985fa3..cee71b8ed0 100644 --- a/include/mosquitto/libmosquitto.h +++ b/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 diff --git a/include/mosquitto/libmosquitto_options.h b/include/mosquitto/libmosquitto_options.h index 255bebdff2..942f120875 100644 --- a/include/mosquitto/libmosquitto_options.h +++ b/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. diff --git a/lib/mosquitto_internal.h b/lib/mosquitto_internal.h index 0dd27b3158..9a4cf3f92e 100644 --- a/lib/mosquitto_internal.h +++ b/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 diff --git a/lib/net_mosq.c b/lib/net_mosq.c index e55be79b83..b4132d1d8c 100644 --- a/lib/net_mosq.c +++ b/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; } diff --git a/lib/net_mosq.h b/lib/net_mosq.h index 6cd556dc0f..baa9d3d5b2 100644 --- a/lib/net_mosq.h +++ b/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 # include # include #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); diff --git a/lib/options.c b/lib/options.c index 6d1cf86909..35dfa63268 100644 --- a/lib/options.c +++ b/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){ diff --git a/man/common/option-mptcp.xml b/man/common/option-mptcp.xml new file mode 100644 index 0000000000..dd40d77c74 --- /dev/null +++ b/man/common/option-mptcp.xml @@ -0,0 +1,14 @@ + + + + + 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. + + + diff --git a/man/mosquitto.conf.5.xml b/man/mosquitto.conf.5.xml index 35423406f7..71602c7bc5 100644 --- a/man/mosquitto.conf.5.xml +++ b/man/mosquitto.conf.5.xml @@ -1603,6 +1603,26 @@ accept_protocol_versions 3, 4 Not reloaded on reload signal. + + [ true | false ] + + Set 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. + + 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. + Defaults to false. + Not reloaded on reload signal. + + port number @@ -2128,6 +2148,22 @@ openssl dhparam -out dhparam.pem 2048 aliases completely. + + [ true | false ] + + Set 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. + 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. + Defaults to false. + + [ true | false ] diff --git a/man/mosquitto_pub.1.xml b/man/mosquitto_pub.1.xml index bf50b1b0f9..dd4ec24e8a 100644 --- a/man/mosquitto_pub.1.xml +++ b/man/mosquitto_pub.1.xml @@ -53,6 +53,7 @@ bind-address + @@ -188,6 +189,7 @@ + diff --git a/man/mosquitto_rr.1.xml b/man/mosquitto_rr.1.xml index c702eb6cfc..8c2b6e3b6f 100644 --- a/man/mosquitto_rr.1.xml +++ b/man/mosquitto_rr.1.xml @@ -56,6 +56,7 @@ bind-address + @@ -226,6 +227,7 @@ + diff --git a/man/mosquitto_sub.1.xml b/man/mosquitto_sub.1.xml index 5babcc6705..af031942be 100644 --- a/man/mosquitto_sub.1.xml +++ b/man/mosquitto_sub.1.xml @@ -57,6 +57,7 @@ bind-address + @@ -237,6 +238,7 @@ + diff --git a/mosquitto.conf b/mosquitto.conf index 805296f7d7..bed7e4e2e7 100644 --- a/mosquitto.conf +++ b/mosquitto.conf @@ -297,6 +297,15 @@ # happens internally to the broker; the client will not see the prefix. #mount_point +# Set mptcp 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. +# This option is only available on Linux, with kernel 5.6 or later. If the +# running kernel does not support MPTCP, the listener falls back to plain TCP. +#mptcp false + # Choose the protocol to use when listening. # This can be either mqtt, websockets, or http_api. # @@ -959,6 +968,16 @@ # properly. #try_private true +# Set bridge_mptcp 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. +# 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. +#bridge_mptcp false + # Some MQTT brokers do not allow retained messages. MQTT v5 gives a mechanism # for brokers to tell clients that they do not support retained messages, but # this is not possible for MQTT v3.1.1 or v3.1. If you need to bridge to a diff --git a/src/bridge.c b/src/bridge.c index cb425dde91..6c6fc86554 100644 --- a/src/bridge.c +++ b/src/bridge.c @@ -91,6 +91,7 @@ static struct mosquitto *bridge__new(struct mosquitto__bridge *bridge) context__add_to_by_id(new_context); } new_context->transport = mosq_t_tcp; + new_context->mptcp = bridge->mptcp; new_context->bridge = bridge; new_context->is_bridge = true; @@ -1043,7 +1044,8 @@ void bridge_check(void) rc = net__try_connect(context->bridge->addresses[0].address, context->bridge->addresses[0].port, &context->bridge->primary_retry_sock, - context->bridge->bind_address, false); + context->bridge->bind_address, false, + context->mptcp); if(rc == 0){ COMPAT_CLOSE(context->bridge->primary_retry_sock); diff --git a/src/conf.c b/src/conf.c index 74f281e895..4290140559 100644 --- a/src/conf.c +++ b/src/conf.c @@ -1432,6 +1432,21 @@ static int config__read_file_core(struct mosquitto__config *config, bool reload, return MOSQ_ERR_INVAL; } cur_bridge->max_topic_alias = (uint16_t)tmp_int; +#else + log__printf(NULL, MOSQ_LOG_WARNING, "Warning: Bridge support not available."); +#endif + }else if(!strcmp(token, "bridge_mptcp")){ +#if defined(WITH_BRIDGE) + REQUIRE_BRIDGE(token); + if(conf__parse_bool(&token, "bridge_mptcp", &cur_bridge->mptcp, &saveptr)){ + return MOSQ_ERR_INVAL; + } +# if !defined(__linux__) + if(cur_bridge->mptcp){ + log__printf(NULL, MOSQ_LOG_WARNING, "Warning: MPTCP support is only available on Linux, 'bridge_mptcp' option ignored."); + cur_bridge->mptcp = false; + } +# endif #else log__printf(NULL, MOSQ_LOG_WARNING, "Warning: Bridge support not available."); #endif @@ -2299,6 +2314,20 @@ static int config__read_file_core(struct mosquitto__config *config, bool reload, cur_listener->mount_point); return MOSQ_ERR_INVAL; } + }else if(!strcmp(token, "mptcp")){ + if(reload){ + continue; /* Listeners are not recreated on reload. */ + } + REQUIRE_LISTENER_OR_DEFAULT_LISTENER(token); + if(conf__parse_bool(&token, "mptcp", &cur_listener->mptcp, &saveptr)){ + return MOSQ_ERR_INVAL; + } +#if !defined(__linux__) + if(cur_listener->mptcp){ + log__printf(NULL, MOSQ_LOG_WARNING, "Warning: MPTCP support is only available on Linux, 'mptcp' option ignored."); + cur_listener->mptcp = false; + } +#endif }else if(!strcmp(token, "notifications")){ #ifdef WITH_BRIDGE REQUIRE_BRIDGE(token); diff --git a/src/mosquitto_broker_internal.h b/src/mosquitto_broker_internal.h index 35055a7a4c..6444db465f 100644 --- a/src/mosquitto_broker_internal.h +++ b/src/mosquitto_broker_internal.h @@ -238,6 +238,7 @@ struct mosquitto__listener { int client_count; enum mosquitto_protocol protocol; int socket_domain; + bool mptcp; bool use_username_as_clientid; uint8_t max_qos; uint16_t max_topic_alias; @@ -554,6 +555,7 @@ struct mosquitto__bridge { time_t primary_retry; mosq_sock_t primary_retry_sock; bool round_robin; + bool mptcp; bool try_private; bool try_private_accepted; bool clean_start; diff --git a/src/net.c b/src/net.c index 84d0a9f641..ae9d0535e4 100644 --- a/src/net.c +++ b/src/net.c @@ -843,11 +843,23 @@ static int net__socket_listen_tcp(struct mosquitto__listener *listener) continue; } - sock = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol); + sock = net__socket_stream(rp->ai_family, rp->ai_socktype, rp->ai_protocol, listener->mptcp); if(sock == INVALID_SOCKET){ net__print_error(MOSQ_LOG_WARNING, "Warning: %s"); continue; } +#if defined(__linux__) && defined(SO_PROTOCOL) + if(listener->mptcp){ + int protocol = 0; + socklen_t protocol_len = sizeof(protocol); + + if(getsockopt(sock, SOL_SOCKET, SO_PROTOCOL, &protocol, &protocol_len) + || protocol != IPPROTO_MPTCP){ + + log__printf(NULL, MOSQ_LOG_WARNING, "Warning: MPTCP is not supported by the kernel, falling back to TCP."); + } + } +#endif listener->sock_count++; listener->socks = mosquitto_realloc(listener->socks, sizeof(mosq_sock_t)*(size_t)listener->sock_count); if(!listener->socks){