@cryptotaxi247 / netdata-1 / commits / 4da06fec0

Reduce memory copies when processing cloud requests (#22493)

* http_api_v2() copied each response body 3 times before MQTT publish (web_client -> local_buffer -> full_msg). Add aclk_http_msg_v2_direct() which assembles JSON header + separator + HTTP headers + body into a single buffer and transfers ownership to MQTT via the existing freefnc zero-copy path. Drops body copies from 3 to 1, and peak coexistent copies from 3 to 2. No on-wire protocol change; error paths untouched. * Address review comments * Use `json_object_to_json_string_length` to calculate JSON string length directly during serialization. * Handle message allocation errors and prevent size overflows during message composition. * Update MQTT publishing logic and simplify memory free functions - Correct the return code check for `mqtt_ng_publish` in MQTT WebSocket client. - Consolidate redundant memory free functions (`freez_aclk_publish5{a,b,c}`) into `freez_aclk_publish_msg`. - Replace `int` with `bool` for clearer payload existence checks. - Minor cleanup and improvements in message assembly and publishing paths. * Refactor MQTT publishing logic to centralize error handling and clarify ownership contracts * Document single-free invariant in `mqtt_ng_generate_publish` to prevent potential double-free scenarios. * Initialize `packet_id` to prevent potential use of uninitialized variable in MQTT publish path. * Normalize NULL pointers for `http_headers` and `body` by setting their respective lengths to 0 to prevent mismatched allocation sizes and maintain consistency during message composition. * Rebase onto master, fix compilation * Initialize `packet_id` to 0 on failure to prevent potential misuse in MQTT WebSocket client. * Clarify ownership contracts and simplify error handling in MQTT publish path, ensuring consistent cleanup on failure. * Add null check for `packet_id` in MQTT publish path to prevent potential misuse. * Clarify ownership and rollback contract for `mqtt_ng_publish` and enforce `topic_free` NULL-check to prevent asymmetric cleanup issues. * Handle non-OK return code in `aclk_tx_msgs` by mapping to `HTTP_RESP_INTERNAL_SERVER_ERROR`. * Normalize return code in `mqtt_wss_wakeup` to use `MQTT_WSS_OK` for consistency.

Stelios Fragkakis committed May 21, 2026 at 11:08 UTC 4da06fec0514d5855e4effb4ba1c052986b94e2a
6 files changed +174 -58
src/aclk/aclk_query.c
+5 -19
@@ -112,7 +112,6 @@ int http_api_v2(mqtt_wss_client client, aclk_query_t *query)
112 ND_LOG_STACK_PUSH(lgs);
113
114 int retval = 0;
115 - BUFFER *local_buffer = NULL;
115 usec_t dt_ut = 0;
116
117 int z_ret;
@@ -191,30 +190,18 @@ int http_api_v2(mqtt_wss_client client, aclk_query_t *query)
190 }
191
192 web_client_build_http_header(w);
194 - local_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE, &netdata_buffers_statistics.buffers_aclk);
195 - local_buffer->content_type = CT_APPLICATION_JSON;
196 -
197 - buffer_strcat(local_buffer, w->response.header_output->buffer);
198 -
199 - if (w->response.data->len) {
200 - if (w->response.zinitialized) {
201 - buffer_need_bytes(local_buffer, w->response.data->len);
202 - memcpy(&local_buffer->buffer[local_buffer->len], w->response.data->buffer, w->response.data->len);
203 - local_buffer->len += w->response.data->len;
204 - } else
205 - buffer_strcat(local_buffer, w->response.data->buffer);
206 - }
193
208 - // send msg.
209 - w->response.code = (short)aclk_http_msg_v2(
194 + w->response.code = (short)aclk_http_msg_v2_direct(
195 client,
196 query->callback_topic,
197 query->msg_id,
198 dt_ut,
199 query->created,
200 w->response.code,
216 - local_buffer->buffer,
217 - local_buffer->len);
201 + w->response.header_output->buffer,
202 + w->response.header_output->len,
203 + w->response.data->buffer,
204 + w->response.data->len);
205
206 cleanup:
207 web_client_log_completed_request(w, false);
@@ -223,7 +210,6 @@ cleanup:
210 pending_req_list_rm(query->msg_id);
211
212 buffer_free(z_buffer);
226 - buffer_free(local_buffer);
213 return retval;
214 }
215
src/aclk/aclk_tx_msgs.c
+93 -5
@@ -11,11 +11,17 @@
11 #pragma region aclk_tx_msgs helper functions
12 #endif
13
14 -static void freez_aclk_publish5a(void *ptr) {
14 +static void freez_aclk_publish_msg(void *ptr) {
15 freez(ptr);
16 }
17 -static void freez_aclk_publish5b(void *ptr) {
18 - freez(ptr);
17 +
18 +static bool aclk_size_add_overflow(size_t *total, size_t add)
19 +{
20 + if (add > SIZE_MAX - *total)
21 + return true;
22 +
23 + *total += add;
24 + return false;
25 }
26
27 #define ACLK_HEADER_VERSION (2)
@@ -39,7 +45,7 @@ uint16_t aclk_send_bin_message_subtopic_pid(mqtt_wss_client client, char *msg, s
45 freez(json);
46 }
47
42 - int rc = mqtt_wss_publish5(client, (char *)topic, NULL, msg, &freez_aclk_publish5a, msg_len, MQTT_WSS_PUB_QOS1, &packet_id);
48 + int rc = mqtt_wss_publish5(client, (char *)topic, NULL, msg, &freez_aclk_publish_msg, msg_len, MQTT_WSS_PUB_QOS1, &packet_id);
49 if (rc != MQTT_WSS_OK)
50 packet_id = 0;
51
@@ -77,10 +83,12 @@ static short aclk_send_message_with_bin_payload(mqtt_wss_client client, json_obj
83 memcpy(&full_msg[len], payload, payload_len);
84 }
85
80 - int rc = mqtt_wss_publish5(client, (char*)topic, NULL, full_msg, &freez_aclk_publish5b, full_msg_len, MQTT_WSS_PUB_QOS1, &packet_id);
86 + int rc = mqtt_wss_publish5(client, (char*)topic, NULL, full_msg, &freez_aclk_publish_msg, full_msg_len, MQTT_WSS_PUB_QOS1, &packet_id);
87
88 if (rc == MQTT_WSS_ERR_MSG_TOO_BIG)
89 return HTTP_RESP_CONTENT_TOO_LONG;
90 + if (rc != MQTT_WSS_OK)
91 + return HTTP_RESP_INTERNAL_SERVER_ERROR;
92
93 return 0;
94 }
@@ -198,6 +206,86 @@ short aclk_http_msg_v2(mqtt_wss_client client, const char *topic, const char *ms
206 return rc;
207 }
208
209 +short aclk_http_msg_v2_direct(mqtt_wss_client client, const char *topic, const char *msg_id,
210 + usec_t t_exec, usec_t created, short http_code,
211 + const char *http_headers, size_t http_headers_len,
212 + const char *body, size_t body_len)
213 +{
214 + if (unlikely(!topic || topic[0] != '/')) {
215 + netdata_log_error("Full topic required!");
216 + return HTTP_RESP_INTERNAL_SERVER_ERROR;
217 + }
218 +
219 + // normalize NULL-with-len so the allocation size and the memcpys stay in sync
220 + if (!http_headers)
221 + http_headers_len = 0;
222 + if (!body)
223 + body_len = 0;
224 +
225 + json_object *msg = create_hdr("http", msg_id);
226 + json_object *tmp;
227 +
228 + tmp = json_object_new_int64(t_exec);
229 + json_object_object_add(msg, "t-exec", tmp);
230 +
231 + tmp = json_object_new_int64(created);
232 + json_object_object_add(msg, "t-rx", tmp);
233 +
234 + tmp = json_object_new_int(http_code);
235 + json_object_object_add(msg, "http-code", tmp);
236 +
237 + size_t json_len;
238 + const char *json_str = json_object_to_json_string_length(msg, JSON_C_TO_STRING_PLAIN, &json_len);
239 +
240 + const size_t sep_len = sizeof(V2_BIN_PAYLOAD_SEPARATOR) - 1;
241 + const bool has_payload = (http_headers_len != 0 || body_len != 0);
242 +
243 + size_t total = json_len;
244 + if (unlikely(
245 + (has_payload && aclk_size_add_overflow(&total, sep_len)) ||
246 + aclk_size_add_overflow(&total, http_headers_len) ||
247 + aclk_size_add_overflow(&total, body_len))) {
248 + json_object_put(msg);
249 + aclk_http_msg_v2_err(client, topic, msg_id, HTTP_RESP_CONTENT_TOO_LONG, CLOUD_EC_REQ_REPLY_TOO_BIG, CLOUD_EMSG_REQ_REPLY_TOO_BIG, NULL, 0);
250 + return HTTP_RESP_CONTENT_TOO_LONG;
251 + }
252 +
253 + char *raw = mallocz(total);
254 +
255 + size_t pos = 0;
256 + memcpy(raw + pos, json_str, json_len);
257 + pos += json_len;
258 + json_object_put(msg);
259 +
260 + if (has_payload) {
261 + memcpy(raw + pos, V2_BIN_PAYLOAD_SEPARATOR, sep_len);
262 + pos += sep_len;
263 + }
264 +
265 + if (http_headers_len) {
266 + memcpy(raw + pos, http_headers, http_headers_len);
267 + pos += http_headers_len;
268 + }
269 +
270 + if (body_len)
271 + memcpy(raw + pos, body, body_len);
272 +
273 + uint16_t packet_id = 0;
274 + int rc = mqtt_wss_publish5(client, (char *)topic, NULL, raw, &freez_aclk_publish_msg, total, MQTT_WSS_PUB_QOS1, &packet_id);
275 +
276 + if (rc == MQTT_WSS_ERR_MSG_TOO_BIG) {
277 + aclk_http_msg_v2_err(client, topic, msg_id, HTTP_RESP_CONTENT_TOO_LONG, CLOUD_EC_REQ_REPLY_TOO_BIG, CLOUD_EMSG_REQ_REPLY_TOO_BIG, NULL, 0);
278 + return HTTP_RESP_CONTENT_TOO_LONG;
279 + }
280 +
281 + if (rc != MQTT_WSS_OK) {
282 + aclk_http_msg_v2_err(client, topic, msg_id, HTTP_RESP_INTERNAL_SERVER_ERROR, CLOUD_EC_SND_TIMEOUT, CLOUD_EMSG_SND_TIMEOUT, NULL, 0);
283 + return HTTP_RESP_INTERNAL_SERVER_ERROR;
284 + }
285 +
286 + return http_code;
287 +}
288 +
289 uint16_t aclk_send_agent_connection_update(mqtt_wss_client client, int reachable) {
290 size_t len;
291 uint16_t pid;
src/aclk/aclk_tx_msgs.h
+4
@@ -13,6 +13,10 @@ uint16_t aclk_send_bin_message_subtopic_pid(mqtt_wss_client client, char *msg, s
13 void aclk_http_msg_v2_err(mqtt_wss_client client, const char *topic, const char *msg_id, int http_code, int ec, const char* emsg, const char *payload, size_t payload_len);
14 short aclk_http_msg_v2(mqtt_wss_client client, const char *topic, const char *msg_id, usec_t t_exec, usec_t created,
15 short http_code, const char *payload, size_t payload_len);
16 +short aclk_http_msg_v2_direct(mqtt_wss_client client, const char *topic, const char *msg_id,
17 + usec_t t_exec, usec_t created, short http_code,
18 + const char *http_headers, size_t http_headers_len,
19 + const char *body, size_t body_len);
20
21 uint16_t aclk_send_agent_connection_update(mqtt_wss_client client, int reachable);
22 char *aclk_generate_lwt(size_t *size);
src/aclk/mqtt_websockets/mqtt_ng.c
+6 -19
@@ -1300,30 +1300,17 @@ int mqtt_ng_publish(struct mqtt_ng_client *client,
1300
1301 if (client->max_msg_size && PUBLISH_SP_SIZE + mqtt_ng_publish_size(topic, msg_len, topic_id) > client->max_msg_size) {
1302 nd_log(NDLS_DAEMON, NDLP_ERR, "Message too big for server: %zu", msg_len);
1303 - if (packet_id)
1304 - *packet_id = 0;
1305 - if (msg_free)
1306 - msg_free(msg);
1303 return MQTT_NG_MSGGEN_MSG_TOO_BIG;
1304 }
1305
1306 + // Ownership contract on failure: do NOT free msg or clear *packet_id here.
1307 + // The sole caller (mqtt_wss_publish5) owns cleanup on every non-OK return.
1308 + // On MQTT_NG_MSGGEN_OK, msg is attached to a buffer fragment and freed by
1309 + // the transaction-buffer GC after ack; *packet_id has been written by the
1310 + // generator. See the long comment in mqtt_wss_publish5 for the invariant.
1311 int rc = TRY_GENERATE_MESSAGE(mqtt_ng_generate_publish, topic, topic_free, msg, msg_free, msg_len, publish_flags, packet_id, topic_id);
1311 - if (rc == MQTT_NG_MSGGEN_OK) {
1312 + if (rc == MQTT_NG_MSGGEN_OK && packet_id)
1313 add_packet_to_timeout_monitor_list(client, *packet_id);
1313 - } else {
1314 - // generator may have written *packet_id before rolling back; clear it so callers
1315 - // don't observe a stale id on failure
1316 - if (packet_id)
1317 - *packet_id = 0;
1318 - if (msg_free) {
1319 - // mqtt_ng_generate_publish has no fail_rollback path after frag_set_external_data
1320 - // succeeds, so on non-OK return msg was never linked to a fragment and the rollback
1321 - // cannot have freed it. Free here to keep the publish-layer contract (msg freed on
1322 - // every non-OK return) holding for all callers. If a future change adds a fallible
1323 - // step after frag_set_external_data, this branch will double-free and must be revisited.
1324 - msg_free(msg);
1325 - }
1326 - }
1314 return rc;
1315 }
1316
src/aclk/mqtt_websockets/mqtt_ng.h
+17
@@ -44,6 +44,23 @@ int mqtt_ng_connect(struct mqtt_ng_client *client,
44 struct mqtt_lwt_properties *lwt,
45 uint16_t keep_alive);
46
47 +/* Publish an MQTT message.
48 + *
49 + * Ownership and cleanup on return:
50 + * - MQTT_NG_MSGGEN_OK: msg is attached to the transaction buffer and freed
51 + * via msg_free after the packet is ack'd; *packet_id is set to the queued
52 + * packet id. The caller must not free msg.
53 + * - Any non-OK return (MQTT_NG_MSGGEN_MSG_TOO_BIG, MQTT_NG_MSGGEN_BUFFER_OOM,
54 + * ...): msg is NOT consumed -- ownership stays with the caller, who must
55 + * invoke msg_free. *packet_id must be treated as undefined: the generator
56 + * may have written a transient value before rolling back, so callers that
57 + * care should reset it to a sentinel (e.g. 0) on the failure path.
58 + *
59 + * topic_free: ownership of topic is handled internally by the transaction
60 + * buffer on the success path. On failure the topic may or may not have been
61 + * attached depending on where generation failed; current callers must pass
62 + * topic_free=NULL until the rollback path is made symmetric.
63 + */
64 int mqtt_ng_publish(struct mqtt_ng_client *client,
65 char *topic,
66 free_fnc_t topic_free,
src/aclk/mqtt_websockets/mqtt_wss_client.c
+49 -15
@@ -830,32 +830,66 @@ int mqtt_wss_publish5(mqtt_wss_client client,
830 uint8_t publish_flags,
831 uint16_t *packet_id)
832 {
833 - if (client->mqtt_disconnecting) {
834 - nd_log(NDLS_DAEMON, NDLP_ERR, "mqtt_wss is disconnecting can't publish");
835 - if (msg_free)
836 - msg_free(msg);
837 - return 1;
838 - }
839 -
840 - if (!client->mqtt_connected) {
841 - nd_log(NDLS_DAEMON, NDLP_ERR, "MQTT is offline. Can't send message.");
833 + // topic_free is not yet supported: the rollback path inside mqtt_ng_publish
834 + // can free topic asymmetrically across failure modes (see contract notes in
835 + // mqtt_ng.h and the long comment below). Enforce NULL until that is fixed
836 + // so callers don't silently leak a borrowed/allocated topic on failure.
837 + internal_fatal(topic_free != NULL, "mqtt_wss_publish5: topic_free must be NULL until rollback ownership is made symmetric");
838 +
839 + const char *fail_reason = NULL;
840 + if (client->mqtt_disconnecting)
841 + fail_reason = "mqtt_wss is disconnecting can't publish";
842 + else if (!client->mqtt_connected)
843 + fail_reason = "MQTT is offline. Can't send message.";
844 +
845 + if (fail_reason) {
846 + nd_log(NDLS_DAEMON, NDLP_ERR, "%s", fail_reason);
847 + if (packet_id)
848 + *packet_id = 0;
849 if (msg_free)
850 msg_free(msg);
851 return 1;
852 }
846 - uint8_t mqtt_flags = 0;
853
848 - mqtt_flags = (publish_flags & MQTT_WSS_PUB_QOSMASK) << 1;
854 + uint8_t mqtt_flags = (publish_flags & MQTT_WSS_PUB_QOSMASK) << 1;
855 if (publish_flags & MQTT_WSS_PUB_RETAIN)
856 mqtt_flags |= MQTT_PUBLISH_RETAIN;
857
858 + // Failure-path ownership contract with mqtt_ng_publish:
859 + // - On MQTT_NG_MSGGEN_OK, msg is attached to a buffer fragment and the
860 + // transaction buffer will call msg_free after the message is ack'd.
861 + // - On any non-OK return, msg is never attached, so ownership stays with
862 + // us and we must call msg_free here.
863 + //
864 + // Single-free invariant: msg is attached to a fragment only at the
865 + // final frag_set_external_data() inside mqtt_ng_generate_publish(),
866 + // after which the function commits unconditionally -- there is no
867 + // `goto fail_rollback` between attachment and commit. Every reachable
868 + // fail_rollback site therefore runs with msg unattached, so the
869 + // rollback walks no msg-bearing fragment and msg_free() is only ever
870 + // called by us. If a future change inserts a failure exit after
871 + // attaching msg but before commit, the rollback would also invoke
872 + // msg_free and this branch would double-free; preserve the invariant
873 + // or move responsibility entirely into mqtt_ng_publish().
874 + //
875 + // - topic_free is intentionally NOT handled here. mqtt_ng_publish may
876 + // attach topic to a fragment via optimized_add() before failing, in
877 + // which case the rollback already invokes topic_free; if it fails
878 + // earlier, topic_free is never invoked at all. The current callers all
879 + // pass topic_free=NULL, so this asymmetry is harmless today.
880 int rc = mqtt_ng_publish(client->mqtt, topic, topic_free, msg, msg_free, msg_len, mqtt_flags, packet_id);
853 - if (rc == MQTT_NG_MSGGEN_MSG_TOO_BIG)
854 - return MQTT_WSS_ERR_MSG_TOO_BIG;
881 + if (rc != MQTT_NG_MSGGEN_OK) {
882 + if (packet_id)
883 + *packet_id = 0;
884 + if (msg_free)
885 + msg_free(msg);
886 + if (rc == MQTT_NG_MSGGEN_MSG_TOO_BIG)
887 + return MQTT_WSS_ERR_MSG_TOO_BIG;
888 + return rc;
889 + }
890
891 mqtt_wss_wakeup(client);
857 -
858 - return rc;
892 + return MQTT_WSS_OK;
893 }
894
895 int mqtt_wss_subscribe(mqtt_wss_client client, char *topic, int max_qos_level)