Add pcket_buffer_size for clients.

Reduce read() calls

lib/mosquitto.c

@@ -113,6 +113,13 @@ struct mosquitto *mosquitto_new(const char *id, bool clean_start, void *userdata
 
 	mosq = (struct mosquitto *)mosquitto_calloc(1, sizeof(struct mosquitto));
 	if(mosq){
+		mosq->in_packet.packet_buffer_size = 4096;
+		mosq->in_packet.packet_buffer = mosquitto_calloc(1, mosq->in_packet.packet_buffer_size);
+		if(!mosq->in_packet.packet_buffer){
+			mosquitto_FREE(mosq);
+			errno = ENOMEM;
+			return NULL;
+		}
 		mosq->sock = INVALID_SOCKET;
 #ifdef WITH_THREADING
 #  ifndef WIN32
@@ -297,6 +304,8 @@ void mosquitto__destroy(struct mosquitto *mosq)
 	mosquitto_FREE(mosq->password);
 	mosquitto_FREE(mosq->host);
 	mosquitto_FREE(mosq->bind_address);
+	mosquitto_FREE(mosq->in_packet.packet_buffer);
+	mosq->in_packet.packet_buffer_size = 0;
 
 	mosquitto_property_free_all(&mosq->connect_properties);
 

lib/mosquitto_internal.h

@@ -182,7 +182,11 @@ struct mosquitto__packet_in{
 	uint32_t packet_length;
 	uint32_t to_process;
 	uint32_t pos;
+	uint16_t packet_buffer_to_process;
+	uint16_t packet_buffer_pos;
+	uint16_t packet_buffer_size;
 	uint8_t command;
+	uint8_t *packet_buffer;
 	int8_t remaining_count;
 };
 

lib/packet_mosq.c

@@ -349,6 +349,40 @@ int packet__write(struct mosquitto *mosq)
 }
 
 
+static int read_header(struct mosquitto *mosq, ssize_t (*func_read)(struct mosquitto *, void *, size_t))
+{
+	ssize_t read_length;
+
+	mosq->in_packet.packet_buffer_pos = 0;
+	read_length = func_read(mosq, &mosq->in_packet.packet_buffer[mosq->in_packet.packet_buffer_pos], mosq->in_packet.packet_buffer_size);
+	if(read_length > 0){
+		mosq->in_packet.packet_buffer_to_process = (uint16_t)read_length;
+#ifdef WITH_BROKER
+		metrics__int_inc(mosq_counter_bytes_received, read_length);
+#endif
+	}else{
+		if(read_length == 0){
+			return MOSQ_ERR_CONN_LOST; /* EOF */
+		}
+#ifdef WIN32
+		errno = WSAGetLastError();
+#endif
+		if(errno == EAGAIN || errno == COMPAT_EWOULDBLOCK){
+			return MOSQ_ERR_SUCCESS;
+		}else{
+			switch(errno){
+				case COMPAT_ECONNRESET:
+					return MOSQ_ERR_CONN_LOST;
+				case COMPAT_EINTR:
+					return MOSQ_ERR_SUCCESS;
+				default:
+					return MOSQ_ERR_ERRNO;
+			}
+		}
+	}
+	return MOSQ_ERR_SUCCESS;
+}
+
 int packet__read(struct mosquitto *mosq)
 {
 	uint8_t byte;
@@ -386,41 +420,27 @@ int packet__read(struct mosquitto *mosq)
 	 * be more than one byte - will need to save data pending next read if it
 	 * does fail.
 	 * Then try to read the remaining payload, where 'payload' here means the
-	 * combined variable header and actual payload. This is the most likely to
+	 * combined variable packet_buffer and actual payload. This is the most likely to
 	 * fail due to longer length, so save current data and current position.
 	 * After all data is read, send to mosquitto__handle_packet() to deal with.
 	 * Finally, free the memory and reset everything to starting conditions.
 	 */
 	if(!mosq->in_packet.command){
-		read_length = local__read(mosq, &byte, 1);
-		if(read_length == 1){
-			mosq->in_packet.command = byte;
+		if(mosq->in_packet.packet_buffer_to_process == 0){
+			rc = read_header(mosq, local__read);
+			if(rc) return rc;
+		}
+
+		if(mosq->in_packet.packet_buffer_to_process > 0){
+			mosq->in_packet.command = mosq->in_packet.packet_buffer[mosq->in_packet.packet_buffer_pos];
+			mosq->in_packet.packet_buffer_to_process--;
+			mosq->in_packet.packet_buffer_pos++;
 #ifdef WITH_BROKER
-			metrics__int_inc(mosq_counter_bytes_received, 1);
 			/* Clients must send CONNECT as their first command. */
-			if(!(mosq->bridge) && state == mosq_cs_new && (byte&0xF0) != CMD_CONNECT){
+			if(!(mosq->bridge) && state == mosq_cs_new && (mosq->in_packet.command&0xF0) != CMD_CONNECT){
 				return MOSQ_ERR_PROTOCOL;
 			}
 #endif
-		}else{
-			if(read_length == 0){
-				return MOSQ_ERR_CONN_LOST; /* EOF */
-			}
-#ifdef WIN32
-			errno = WSAGetLastError();
-#endif
-			if(errno == EAGAIN || errno == COMPAT_EWOULDBLOCK){
-				return MOSQ_ERR_SUCCESS;
-			}else{
-				switch(errno){
-					case COMPAT_ECONNRESET:
-						return MOSQ_ERR_CONN_LOST;
-					case COMPAT_EINTR:
-						return MOSQ_ERR_SUCCESS;
-					default:
-						return MOSQ_ERR_ERRNO;
-				}
-			}
 		}
 	}
 	/* remaining_count is the number of bytes that the remaining_length
@@ -434,8 +454,12 @@ int packet__read(struct mosquitto *mosq)
 	 */
 	if(mosq->in_packet.remaining_count <= 0){
 		do{
-			read_length = local__read(mosq, &byte, 1);
-			if(read_length == 1){
+			if(mosq->in_packet.packet_buffer_to_process == 0){
+				rc = read_header(mosq, local__read);
+				if(rc) return rc;
+			}
+
+			if(mosq->in_packet.packet_buffer_to_process > 0){
 				mosq->in_packet.remaining_count--;
 				/* Max 4 bytes length for remaining length as defined by protocol.
 				 * Anything more likely means a broken/malicious client.
@@ -444,28 +468,13 @@ int packet__read(struct mosquitto *mosq)
 					return MOSQ_ERR_MALFORMED_PACKET;
 				}
 
-				metrics__int_inc(mosq_counter_bytes_received, 1);
+				byte = mosq->in_packet.packet_buffer[mosq->in_packet.packet_buffer_pos];
+				mosq->in_packet.packet_buffer_pos++;
+				mosq->in_packet.packet_buffer_to_process--;
 				mosq->in_packet.remaining_length += (byte & 127) * mosq->in_packet.remaining_mult;
 				mosq->in_packet.remaining_mult *= 128;
 			}else{
-				if(read_length == 0){
-					return MOSQ_ERR_CONN_LOST; /* EOF */
-				}
-#ifdef WIN32
-				errno = WSAGetLastError();
-#endif
-				if(errno == EAGAIN || errno == COMPAT_EWOULDBLOCK){
-					return MOSQ_ERR_SUCCESS;
-				}else{
-					switch(errno){
-						case COMPAT_ECONNRESET:
-							return MOSQ_ERR_CONN_LOST;
-						case COMPAT_EINTR:
-							return MOSQ_ERR_SUCCESS;
-						default:
-							return MOSQ_ERR_ERRNO;
-					}
-				}
+				return MOSQ_ERR_SUCCESS;
 			}
 		}while((byte & 128) != 0);
 		/* We have finished reading remaining_length, so make remaining_count
@@ -518,7 +527,23 @@ int packet__read(struct mosquitto *mosq)
 			if(!mosq->in_packet.payload){
 				return MOSQ_ERR_NOMEM;
 			}
+
+			mosq->in_packet.pos = 0;
 			mosq->in_packet.to_process = mosq->in_packet.remaining_length;
+
+			if(mosq->in_packet.packet_buffer_to_process > 0){
+				uint32_t len;
+				if(mosq->in_packet.packet_buffer_to_process > mosq->in_packet.remaining_length){
+					len = mosq->in_packet.remaining_length;
+				}else{
+					len = mosq->in_packet.packet_buffer_to_process;
+				}
+				memcpy(mosq->in_packet.payload, &mosq->in_packet.packet_buffer[mosq->in_packet.packet_buffer_pos], len);
+				mosq->in_packet.packet_buffer_pos += (uint16_t)len;
+				mosq->in_packet.packet_buffer_to_process -= (uint16_t)len;
+				mosq->in_packet.pos += len;
+				mosq->in_packet.to_process -= len;
+			}
 		}
 	}
 	while(mosq->in_packet.to_process>0){

man/mosquitto.conf.5.xml

@@ -1605,16 +1605,25 @@ accept_protocol_versions 3, 4</programlisting>
 					</listitem>
 				</varlistentry>
 				<varlistentry>
+					<term><option>packet_buffer_size</option> <replaceable>size</replaceable></term>
 					<term><option>websockets_headers_size</option> <replaceable>size</replaceable></term>
 					<listitem>
-						<para>Change the websockets headers size. This is a
-							global option, it is not possible to set per
-							listener. This option sets the size of the buffer
-							used in the libwebsockets library when reading HTTP
-							headers. If you are passing large header data such
-							as cookies then you may need to increase this
-							value. If left unset, or set to 0, then the default
-							of 1024 bytes will be used.</para>
+						<para>
+                            Change the size of the buffer used when reading from
+                            the network before the size of the MQTT packet is
+                            known. Defaults to 4096. Packets received that are
+                            smaller than this value in principle only need a
+                            single read() call, making reading packets more
+                            efficient.
+						</para>
+
+						<para>
+                            This also operates as the option that sets the size
+                            of the buffer used by websockets when reading the
+                            initial header. If you are passing large header data
+                            such as cookies then you may need to increase this
+                            value.
+						</para>
 					</listitem>
 				</varlistentry>
 				<varlistentry>

mosquitto.conf

@@ -342,12 +342,15 @@
 # This does not apply globally, but on a per-listener basis.
 #use_username_as_clientid
 
-# Change the websockets headers size. This is a global option, it is not
-# possible to set per listener. This option sets the size of the buffer used in
-# the libwebsockets library when reading HTTP headers. If you are passing large
-# header data such as cookies then you may need to increase this value. If left
-# unset, or set to 0, then the default of 1024 bytes will be used.
-#websockets_headers_size
+# Change the size of the buffer used when reading from the network before the
+# size of the MQTT packet is known. Defaults to 4096. Packets received that are
+# smaller than this value in principle only need a single read() call, making
+# reading packets more efficient. 
+#
+# This also operates as the option that sets the size of the buffer used by
+# websockets when reading the initial header.  If you are passing large
+# header data such as cookies then you may need to increase this value.
+#packet_buffer_size
 
 # Enforce origin checking for websockets connections. This should be set to the
 # string of the host that you wish to allow connections from, as set as the

src/conf.c

@@ -318,9 +318,7 @@ static void config__init_reload(struct mosquitto__config *config)
 	config->set_tcp_nodelay = false;
 	config->sys_interval = 10;
 	config->upgrade_outgoing_qos = false;
-#ifdef WITH_WEBSOCKETS
-	config->websockets_headers_size = 4096;
-#endif
+	config->packet_buffer_size = 4096;
 }
 
 
@@ -2457,17 +2455,13 @@ static int config__read_file_core(struct mosquitto__config *config, bool reload,
 #if defined(WITH_WEBSOCKETS) && WITH_WEBSOCKETS == WS_IS_LWS
 					if(conf__parse_int(&token, "websockets_log_level", &config->websockets_log_level, &saveptr)) return MOSQ_ERR_INVAL;
 #endif
-				}else if(!strcmp(token, "websockets_headers_size")){
-#ifdef WITH_WEBSOCKETS
-					if(conf__parse_int(&token, "websockets_headers_size", &tmp_int, &saveptr)) return MOSQ_ERR_INVAL;
+				}else if(!strcmp(token, "websockets_headers_size") || !strcmp(token, "packet_buffer_size")){
+					if(conf__parse_int(&token, token, &tmp_int, &saveptr)) return MOSQ_ERR_INVAL;
 					if(tmp_int < 0 || tmp_int > UINT16_MAX){
-						log__printf(NULL, MOSQ_LOG_WARNING, "Error: Websockets headers size must be between 0 and 65535 inclusive.");
+						log__printf(NULL, MOSQ_LOG_WARNING, "Error: Packet buffer size must be between 0 and 65535 inclusive.");
 						return MOSQ_ERR_INVAL;
 					}
-					config->websockets_headers_size = (uint16_t)tmp_int;
-#else
-					log__printf(NULL, MOSQ_LOG_WARNING, "Warning: Websockets support not available.");
-#endif
+					config->packet_buffer_size = (uint16_t)tmp_int;
 				}else if(!strcmp(token, "websockets_origin")){
 #ifdef WITH_WEBSOCKETS
 #  if LWS_LIBRARY_VERSION_NUMBER >= 3001000 || WITH_WEBSOCKETS == WS_IS_BUILTIN

src/context.c

@@ -64,6 +64,13 @@ struct mosquitto *context__init(void)
 	context = mosquitto_calloc(1, sizeof(struct mosquitto));
 	if(!context) return NULL;
 
+	context->in_packet.packet_buffer_size = db.config->packet_buffer_size;
+	context->in_packet.packet_buffer = mosquitto_calloc(1, context->in_packet.packet_buffer_size);
+	if(!context->in_packet.packet_buffer){
+		mosquitto_FREE(context);
+		return NULL;
+	}
+
 #if defined(WITH_EPOLL) || defined(WITH_KQUEUE)
 	context->ident = id_client;
 #else
@@ -152,6 +159,8 @@ void context__cleanup(struct mosquitto *context, bool force_free)
 		bridge__cleanup(context);
 	}
 #endif
+	mosquitto_FREE(context->in_packet.packet_buffer);
+	context->in_packet.packet_buffer_size = 0;
 
 	alias__free_all(context);
 	keepalive__remove(context);

src/http_serv.c

@@ -37,10 +37,6 @@ Contributors:
 int http__context_init(struct mosquitto *context)
 {
 	context->transport = mosq_t_http;
-	context->http_request = mosquitto_calloc(1, db.config->websockets_headers_size);
-	if(context->http_request == NULL){
-		return MOSQ_ERR_NOMEM;
-	}
 
 	return MOSQ_ERR_SUCCESS;
 }
@@ -48,7 +44,7 @@ int http__context_init(struct mosquitto *context)
 
 int http__context_cleanup(struct mosquitto *context)
 {
-	mosquitto_FREE(context->http_request);
+	UNUSED(context);
 	return MOSQ_ERR_SUCCESS;
 }
 
@@ -91,8 +87,8 @@ int http__read(struct mosquitto *mosq)
 		return MOSQ_ERR_SUCCESS;
 	}
 
-	hlen = strlen(mosq->http_request);
-	read_length = net__read(mosq, &mosq->http_request[hlen], db.config->websockets_headers_size-hlen);
+	hlen = strlen((char *)mosq->in_packet.packet_buffer);
+	read_length = net__read(mosq, &mosq->in_packet.packet_buffer[hlen], mosq->in_packet.packet_buffer_size-hlen);
 	if(read_length <= 0){
 		if(read_length == 0){
 			return MOSQ_ERR_CONN_LOST; /* EOF */
@@ -114,8 +110,8 @@ int http__read(struct mosquitto *mosq)
 		}
 	}
 
-	mosq->http_request[db.config->websockets_headers_size-1] = '\0'; /* Always 0 terminate */
-	read_length = phr_parse_request(mosq->http_request, strlen(mosq->http_request),
+	mosq->in_packet.packet_buffer[mosq->in_packet.packet_buffer_size-1] = '\0'; /* Always 0 terminate */
+	read_length = phr_parse_request((char *)mosq->in_packet.packet_buffer, strlen((char *)mosq->in_packet.packet_buffer),
 			&http_method, &http_method_len,
 			&http_path, &http_path_len,
 			&http_minor_version,
@@ -131,7 +127,6 @@ int http__read(struct mosquitto *mosq)
 	}
 
 	if(strncmp(http_method, "GET", http_method_len) && strncmp(http_method, "HEAD", http_method_len)){
-		mosquitto_FREE(mosq->http_request);
 		/* FIXME Not supported - send 501 response */
 		return MOSQ_ERR_UNKNOWN;
 	}
@@ -247,7 +242,7 @@ int http__read(struct mosquitto *mosq)
 	SAFE_FREE(accept_key);
 	packet->to_process = packet->packet_length;
 
-	memset(mosq->http_request, 0, db.config->websockets_headers_size);
+	memset(mosq->in_packet.packet_buffer, 0, db.config->packet_buffer_size);
 	rc = packet__queue(mosq, packet);
 	http__context_cleanup(mosq);
 	ws__context_init(mosq);

src/mosquitto_broker_internal.h

@@ -352,9 +352,7 @@ struct mosquitto__config {
 #if defined(WITH_WEBSOCKETS) && WITH_WEBSOCKETS == WS_IS_LWS
 	int websockets_log_level;
 #endif
-#ifdef WITH_WEBSOCKETS
-	uint16_t websockets_headers_size;
-#endif
+	uint16_t packet_buffer_size;
 #ifdef WITH_BRIDGE
 	struct mosquitto__bridge **bridges;
 	int bridge_count;

test/broker/16-config-parse-errors-without-tls.py

@@ -77,8 +77,8 @@ do_test_broker_failure(conf_file, [f"listener {port}", "max_topic_alias 65536"],
 do_test_broker_failure(conf_file, [f"listener {port}", "max_topic_alias -1"], port, 3, "Error: Invalid 'max_topic_alias' value in configuration.") # Invalid value
 do_test_broker_failure(conf_file, [f"listener {port}", "max_topic_alias_broker 65536"], port, 3, "Error: Invalid 'max_topic_alias_broker' value in configuration.") # Invalid value
 do_test_broker_failure(conf_file, [f"listener {port}", "max_topic_alias_broker -1"], port, 3, "Error: Invalid 'max_topic_alias_broker' value in configuration.") # Invalid value
-do_test_broker_failure(conf_file, ["websockets_headers_size 65536"], port, 3, "Error: Websockets headers size must be between 0 and 65535 inclusive.") # Invalid 'value
-do_test_broker_failure(conf_file, ["websockets_headers_size -1"], port, 3, "Error: Websockets headers size must be between 0 and 65535 inclusive.") # Invalid value
+do_test_broker_failure(conf_file, ["websockets_headers_size 65536"], port, 3, "Error: Packet buffer size must be between 0 and 65535 inclusive.") # Invalid 'value
+do_test_broker_failure(conf_file, ["websockets_headers_size -1"], port, 3, "Error: Packet buffer size must be between 0 and 65535 inclusive.") # Invalid value
 do_test_broker_failure(conf_file, ["memory_limit -1"], port, 3, "Error: Invalid 'memory_limit' value (-1).") # Invalid value
 
 do_test_broker_failure(conf_file, ["sys_interval -1"], port, 3, "Error: Invalid 'sys_interval' value (-1).") # Invalid value