From d98d986a6ed0a083fb3c63dddf84a2fd6548b732 Mon Sep 17 00:00:00 2001 From: "Roger A. Light" Date: Mon, 11 Nov 2024 17:06:50 +0000 Subject: [PATCH] Add pcket_buffer_size for clients. Reduce read() calls --- lib/mosquitto.c | 9 ++ lib/mosquitto_internal.h | 4 + lib/packet_mosq.c | 117 +++++++++++------- man/mosquitto.conf.5.xml | 25 ++-- mosquitto.conf | 15 ++- src/conf.c | 16 +-- src/context.c | 9 ++ src/http_serv.c | 17 +-- src/mosquitto_broker_internal.h | 4 +- .../16-config-parse-errors-without-tls.py | 4 +- 10 files changed, 133 insertions(+), 87 deletions(-) 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 + size size - 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