Add TCP_NODELAY support to lib and clients.

Closes #1526. Thanks to Felix Moessbauer.

ChangeLog.txt

@@ -22,6 +22,8 @@ Client library:
  iterating over property lists.
 - mosquitto_pub now handles the MQTT v5 retain-available property by never
  setting the retain bit.
+- Added MOSQ_OPT_TCP_NODELAY, to allow disabling Nagle's algorithm on client
+ sockets. Closes #1526.
 
 Clients:
 - Add timeout return code (27) for `mosquitto_sub -W <secs>` and
@@ -36,6 +38,8 @@ Clients:
 - Add support for v5 property printing to mosquitto_sub/rr in non-JSON mode.
  Closes #1416.
 - mosquitto_sub will now exit if all subscriptions were denied.
+- Add `--nodelay` to all clients to allow them to use the MOSQ_OPT_TCP_NODELAY
+ option.
 
 
 1.6.8 - 20191128

client/client_shared.c

@@ -737,6 +737,8 @@ int client_config_line_proc(struct mosq_config *cfg, int pub_or_sub, int argc, c
  cfg->max_inflight = atoi(argv[i+1]);
  }
  i++;
+ }else if(!strcmp(argv[i], "--nodelay")){
+ cfg->tcp_nodelay = true;
  }else if(!strcmp(argv[i], "-n") || !strcmp(argv[i], "--null-message")){
  if(pub_or_sub == CLIENT_SUB){
  goto unknown_option;
@@ -1177,6 +1179,9 @@ int client_opts_set(struct mosquitto *mosq, struct mosq_config *cfg)
  }
  }
 #endif
+ if(cfg->tcp_nodelay){
+ mosquitto_int_option(mosq, MOSQ_OPT_TCP_NODELAY, 1);
+ }
  return MOSQ_ERR_SUCCESS;
 }
 

client/client_shared.h

@@ -120,6 +120,7 @@ struct mosq_config {
  mosquitto_property *will_props;
  bool have_topic_alias; /* pub */
  char *response_topic; /* rr */
+ bool tcp_nodelay;
 };
 
 int client_config_load(struct mosq_config *config, int pub_or_sub, int argc, char *argv[]);

client/pub_client.c

@@ -371,9 +371,9 @@ void print_usage(void)
  printf(" {-f file | -l | -n | -m message}\n");
  printf(" [-c] [-k keepalive] [-q qos] [-r] [--repeat N] [--repeat-delay time]\n");
 #ifdef WITH_SRV
- printf(" [-A bind_address] [-S]\n");
+ printf(" [-A bind_address] [--nodelay] [-S]\n");
 #else
- printf(" [-A bind_address]\n");
+ printf(" [-A bind_address] [--nodelay]\n");
 #endif
  printf(" [-i id] [-I id_prefix]\n");
  printf(" [-d] [--quiet]\n");
@@ -424,9 +424,11 @@ void print_usage(void)
  printf(" -V : specify the version of the MQTT protocol to use when connecting.\n");
  printf(" Can be mqttv5, mqttv311 or mqttv31. Defaults to mqttv311.\n");
  printf(" --help : display this message.\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");
  printf(" --repeat : if publish mode is -f, -m, or -s, then repeat the publish N times.\n");
  printf(" --repeat-delay : if using --repeat, wait time seconds between publishes. Defaults to 0.\n");
- printf(" --quiet : don't print error messages.\n");
  printf(" --unix : connect to a broker through a unix domain socket instead of a TCP socket,\n");
  printf(" e.g. /tmp/mosquitto.sock\n");
  printf(" --will-payload : payload for the client Will, which is sent by the broker in case of\n");

client/rr_client.c

@@ -169,9 +169,9 @@ void print_usage(void)
  printf(" [-W timeout_secs]\n");
 #endif
 #ifdef WITH_SRV
- printf(" [-A bind_address] [-S]\n");
+ printf(" [-A bind_address] [--nodelay] [-S]\n");
 #else
- printf(" [-A bind_address]\n");
+ printf(" [-A bind_address] [--nodelay]\n");
 #endif
  printf(" [-i id] [-I id_prefix]\n");
  printf(" [-d] [-N] [--quiet] [-v]\n");
@@ -218,6 +218,8 @@ void print_usage(void)
  printf(" -W : Specifies a timeout in seconds how long to wait for a response.\n");
 #endif
  printf(" --help : display this message.\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");
  printf(" JSON output format option.\n");
  printf(" --quiet : don't print error messages.\n");

client/sub_client.c

@@ -188,9 +188,9 @@ void print_usage(void)
  printf(" [-W timeout_secs]\n");
 #endif
 #ifdef WITH_SRV
- printf(" [-A bind_address] [-S]\n");
+ printf(" [-A bind_address] [--nodelay] [-S]\n");
 #else
- printf(" [-A bind_address]\n");
+ printf(" [-A bind_address] [--nodelay]\n");
 #endif
  printf(" [-i id] [-I id_prefix]\n");
  printf(" [-d] [-N] [--quiet] [-v]\n");
@@ -243,6 +243,8 @@ void print_usage(void)
  printf(" -W : Specifies a timeout in seconds how long to process incoming MQTT messages.\n");
 #endif
  printf(" --help : display this message.\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");
  printf(" JSON output format option.\n");
  printf(" --quiet : don't print error messages.\n");

lib/mosquitto.h

@@ -113,6 +113,7 @@ enum mosq_opt_t {
  MOSQ_OPT_TLS_ENGINE_KPASS_SHA1 = 8,
  MOSQ_OPT_TLS_OCSP_REQUIRED = 9,
  MOSQ_OPT_TLS_ALPN = 10,
+ MOSQ_OPT_TCP_NODELAY = 11,
 };
 
 
@@ -1432,6 +1433,12 @@ libmosq_EXPORT int mosquitto_opts_set(struct mosquitto *mosq, enum mosq_opt_t op
  * value - the option specific value.
  *
  * Options:
+ * MOSQ_OPT_TCP_NODELAY -
+ * Set to 1 to disable Nagle's algorithm on client sockets. This has
+ * the effect of reducing latency of individual messages at the
+ * potential cost of increasing the number of packets being sent.
+ * Defaults to 0, which means Nagle remains enabled.
+ *
  * 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

lib/mosquitto_internal.h

@@ -337,6 +337,7 @@ struct mosquitto {
 #endif
  uint8_t maximum_qos;
  uint8_t retain_available;
+ bool tcp_nodelay;
 
 #ifdef WITH_BROKER
  UT_hash_handle hh_id;

lib/net_mosq.c

@@ -25,6 +25,7 @@ and the Eclipse Distribution License is available at
 #ifndef WIN32
 #define _GNU_SOURCE
 #include <netdb.h>
+#include <netinet/tcp.h>
 #include <sys/socket.h>
 #include <unistd.h>
 #else
@@ -887,6 +888,13 @@ int net__socket_connect(struct mosquitto *mosq, const char *host, uint16_t port,
 
  mosq->sock = sock;
 
+ if(mosq->tcp_nodelay){
+ int flag = 1;
+ if(setsockopt(mosq->sock, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(int)) != 0){
+ log__printf(mosq, MOSQ_LOG_WARNING, "Warning: Unable to set TCP_NODELAY.");
+ }
+ }
+
 #if defined(WITH_SOCKS) && !defined(WITH_BROKER)
  if(!mosq->socks5_host)
 #endif

lib/options.c

@@ -447,6 +447,10 @@ int mosquitto_int_option(struct mosquitto *mosq, enum mosq_opt_t option, int val
 #endif
  break;
 
+ case MOSQ_OPT_TCP_NODELAY:
+ mosq->tcp_nodelay = (bool)value;
+ break;
+
  default:
  return MOSQ_ERR_INVAL;
  }

man/mosquitto_pub.1.xml

@@ -35,6 +35,7 @@
  <arg><option>-i</option> <replaceable>client-id</replaceable></arg>
  <arg><option>-I</option> <replaceable>client-id-prefix</replaceable></arg>
  <arg><option>-k</option> <replaceable>keepalive-time</replaceable></arg>
+ <arg><option>--nodelay</option></arg>
  <arg><option>-q</option> <replaceable>message-QoS</replaceable></arg>
  <arg><option>--quiet</option></arg>
  <arg><option>-r</option></arg>
@@ -356,6 +357,16 @@
  <para>Send a null (zero length) message.</para>
  </listitem>
  </varlistentry>
+ <varlistentry>
+ <term><option>--nodelay</option></term>
+ <listitem>
+ <para>Disable Nagle's algorithm for the socket. This means
+ that latency of sent messages is reduced, which is
+ particularly noticable for small, reasonably infrequent
+ messages. Using this option may result in more packets
+ being sent than would normally be necessary.</para>
+ </listitem>
+ </varlistentry>
  <varlistentry>
  <term><option>-p</option></term>
  <term><option>--port</option></term>

man/mosquitto_rr.1.xml

@@ -46,6 +46,7 @@
  <arg><option>-I</option> <replaceable>client-id-prefix</replaceable></arg>
  <arg><option>-k</option> <replaceable>keepalive-time</replaceable></arg>
  <arg><option>-N</option></arg>
+ <arg><option>--nodelay</option></arg>
  <arg><option>--pretty</option></arg>
  <arg><option>-q</option> <replaceable>message-QoS</replaceable></arg>
  <arg><option>-R</option></arg>
@@ -378,6 +379,16 @@
  <para>Send a null (zero length) request message.</para>
  </listitem>
  </varlistentry>
+ <varlistentry>
+ <term><option>--nodelay</option></term>
+ <listitem>
+ <para>Disable Nagle's algorithm for the socket. This means
+ that latency of sent messages is reduced, which is
+ particularly noticable for small, reasonably infrequent
+ messages. Using this option may result in more packets
+ being sent than would normally be necessary.</para>
+ </listitem>
+ </varlistentry>
  <varlistentry>
  <term><option>-p</option></term>
  <term><option>--port</option></term>

man/mosquitto_sub.1.xml

@@ -41,6 +41,7 @@
  <arg><option>-I</option> <replaceable>client-id-prefix</replaceable></arg>
  <arg><option>-k</option> <replaceable>keepalive-time</replaceable></arg>
  <arg><option>-N</option></arg>
+ <arg><option>--nodelay</option></arg>
  <arg><option>--pretty</option></arg>
  <arg><option>-q</option> <replaceable>message-QoS</replaceable></arg>
  <arg><option>--remove-retained</option></arg>
@@ -387,6 +388,16 @@
  <option>-v</option>.</para>
  </listitem>
  </varlistentry>
+ <varlistentry>
+ <term><option>--nodelay</option></term>
+ <listitem>
+ <para>Disable Nagle's algorithm for the socket. This means
+ that latency of sent messages is reduced, which is
+ particularly noticable for small, reasonably infrequent
+ messages. Using this option may result in more packets
+ being sent than would normally be necessary.</para>
+ </listitem>
+ </varlistentry>
  <varlistentry>
  <term><option>-p</option></term>
  <term><option>--port</option></term>