diff --git a/lib/mosquitto.c b/lib/mosquitto.c
index 2134c7ce..9dcda83d 100644
--- a/lib/mosquitto.c
+++ b/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);
diff --git a/lib/mosquitto_internal.h b/lib/mosquitto_internal.h
index f4fd487f..f9aa4518 100644
--- a/lib/mosquitto_internal.h
+++ b/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;
};
diff --git a/lib/packet_mosq.c b/lib/packet_mosq.c
index c4a8ce7c..6eb4f3b1 100644
--- a/lib/packet_mosq.c
+++ b/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){
diff --git a/man/mosquitto.conf.5.xml b/man/mosquitto.conf.5.xml
index fc629694..1eb7a8c8 100644
--- a/man/mosquitto.conf.5.xml
+++ b/man/mosquitto.conf.5.xml
@@ -1605,16 +1605,25 @@ accept_protocol_versions 3, 4
+ sizesize
- 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.
+
+ 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.
+
diff --git a/mosquitto.conf b/mosquitto.conf
index 5ae4d900..9295e481 100644
--- a/mosquitto.conf
+++ b/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
diff --git a/src/conf.c b/src/conf.c
index 2dfe32ac..33f87666 100644
--- a/src/conf.c
+++ b/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
diff --git a/src/context.c b/src/context.c
index ab77b4c3..ce1deccf 100644
--- a/src/context.c
+++ b/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);
diff --git a/src/http_serv.c b/src/http_serv.c
index 03ff8392..122a0314 100644
--- a/src/http_serv.c
+++ b/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);
diff --git a/src/mosquitto_broker_internal.h b/src/mosquitto_broker_internal.h
index 8618b207..fc25607e 100644
--- a/src/mosquitto_broker_internal.h
+++ b/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;
diff --git a/test/broker/16-config-parse-errors-without-tls.py b/test/broker/16-config-parse-errors-without-tls.py
index 3b444bc3..befdb79b 100755
--- a/test/broker/16-config-parse-errors-without-tls.py
+++ b/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