Use atomics for mqtt statistics (#20836)
* Connection is always clean * Use atomics for statistics
Stelios Fragkakis committed
Aug 18, 2025 at 17:12 UTC
45cab0333024a01cbebac2f0663012f5c8dcba5d
3 files changed
+27
-43
src/aclk/mqtt_websockets/mqtt_ng.c
+26
-41
@@ -240,7 +240,6 @@ struct mqtt_ng_client {
240
unsigned int ping_pending:1;
241
242
struct mqtt_ng_stats stats;
243
- SPINLOCK stats_spinlock;
243
244
struct {
245
SPINLOCK spinlock;
@@ -626,7 +625,6 @@ struct mqtt_ng_client *mqtt_ng_init(struct mqtt_ng_init *settings)
625
626
client->rx_aliases = RX_ALIASES_INITIALIZE();
627
629
- spinlock_init(&client->stats_spinlock);
628
spinlock_init(&client->tx_topic_aliases.spinlock);
629
630
client->tx_topic_aliases.stoi_dict = TX_ALIASES_INITIALIZE();
@@ -872,18 +870,16 @@ static void add_packet_to_timeout_monitor_list(struct mqtt_ng_client *client, ui
870
__LINE__); \
871
} \
872
if (_rc == MQTT_NG_MSGGEN_OK) { \
875
- spinlock_lock(&client->stats_spinlock); \
876
- client->stats.tx_messages_queued++; \
877
- spinlock_unlock(&client->stats_spinlock); \
873
+ __atomic_fetch_add(&client->stats.tx_messages_queued, 1, __ATOMIC_RELAXED); \
874
} \
875
_rc; \
876
})
877
882
-mqtt_msg_data mqtt_ng_generate_connect(struct transaction_buffer *trx_buf,
883
- struct mqtt_auth_properties *auth,
884
- struct mqtt_lwt_properties *lwt,
885
- uint8_t clean_start,
886
- uint16_t keep_alive)
878
+mqtt_msg_data mqtt_ng_generate_connect(
879
+ struct transaction_buffer *trx_buf,
880
+ struct mqtt_auth_properties *auth,
881
+ struct mqtt_lwt_properties *lwt,
882
+ uint16_t keep_alive)
883
{
884
// Sanity Checks First (are given parameters correct and up to MQTT spec)
885
if (!auth->client_id) {
@@ -958,8 +954,8 @@ mqtt_msg_data mqtt_ng_generate_connect(struct transaction_buffer *trx_buf,
954
if (lwt->will_retain)
955
*connect_flags |= MQTT_CONNECT_FLAG_LWT_RETAIN;
956
}
961
- if (clean_start)
962
- *connect_flags |= MQTT_CONNECT_FLAG_CLEAN_START;
957
+
958
+ *connect_flags |= MQTT_CONNECT_FLAG_CLEAN_START;
959
960
DATA_ADVANCE(&trx_buf->hdr_buffer, 1, frag)
961
@@ -1028,23 +1024,21 @@ fail_rollback:
1024
return NULL;
1025
}
1026
1031
-int mqtt_ng_connect(struct mqtt_ng_client *client,
1032
- struct mqtt_auth_properties *auth,
1033
- struct mqtt_lwt_properties *lwt,
1034
- uint8_t clean_start,
1035
- uint16_t keep_alive)
1027
+int mqtt_ng_connect(
1028
+ struct mqtt_ng_client *client,
1029
+ struct mqtt_auth_properties *auth,
1030
+ struct mqtt_lwt_properties *lwt,
1031
+ uint16_t keep_alive)
1032
{
1033
client->client_state = MQTT_STATE_RAW;
1034
client->parser.state = MQTT_PARSE_FIXED_HEADER_PACKET_TYPE;
1035
1036
LOCK_HDR_BUFFER(&client->main_buffer);
1037
client->main_buffer.sending_frag = NULL;
1042
- if (clean_start)
1043
- buffer_purge(&client->main_buffer.hdr_buffer);
1038
+ buffer_purge(&client->main_buffer.hdr_buffer);
1039
UNLOCK_HDR_BUFFER(&client->main_buffer);
1040
1046
- if (clean_start)
1047
- destroy_timeout_monitor_list(client);
1041
+ destroy_timeout_monitor_list(client);
1042
1043
spinlock_lock(&client->tx_topic_aliases.spinlock);
1044
// according to MQTT spec topic aliases should not be persisted
@@ -1058,19 +1052,13 @@ int mqtt_ng_connect(struct mqtt_ng_client *client,
1052
mqtt_ng_destroy_rx_alias_hash(client->rx_aliases);
1053
client->rx_aliases = RX_ALIASES_INITIALIZE();
1054
1061
- client->connect_msg = mqtt_ng_generate_connect(&client->main_buffer, auth, lwt, clean_start, keep_alive);
1055
+ client->connect_msg = mqtt_ng_generate_connect(&client->main_buffer, auth, lwt, keep_alive);
1056
if (client->connect_msg == NULL)
1057
return 1;
1058
1065
- spinlock_lock(&client->stats_spinlock);
1066
- if (clean_start)
1067
- client->stats.tx_messages_queued = 1;
1068
- else
1069
- client->stats.tx_messages_queued++;
1070
-
1071
- client->stats.tx_messages_sent = 0;
1072
- client->stats.rx_messages_rcvd = 0;
1073
- spinlock_unlock(&client->stats_spinlock);
1059
+ __atomic_store_n(&client->stats.tx_messages_queued, 1, __ATOMIC_RELAXED);
1060
+ __atomic_store_n(&client->stats.tx_messages_sent, 0, __ATOMIC_RELAXED);
1061
+ __atomic_store_n(&client->stats.rx_messages_rcvd, 0, __ATOMIC_RELAXED);
1062
1063
client->client_state = MQTT_STATE_CONNECT_PENDING;
1064
return 0;
@@ -2062,11 +2050,9 @@ static int send_fragment(struct mqtt_ng_client *client) {
2050
2051
if (frag->flags & BUFFER_FRAG_MQTT_PACKET_TAIL) {
2052
client->time_of_last_send = time(NULL);
2065
- spinlock_lock(&client->stats_spinlock);
2053
if (client->main_buffer.sending_frag != &ping_frag)
2067
- client->stats.tx_messages_queued--;
2068
- client->stats.tx_messages_sent++;
2069
- spinlock_unlock(&client->stats_spinlock);
2054
+ __atomic_fetch_sub(&client->stats.tx_messages_queued, 1, __ATOMIC_RELAXED);
2055
+ __atomic_fetch_add(&client->stats.tx_messages_sent, 1, __ATOMIC_RELAXED);
2056
client->main_buffer.sending_frag = NULL;
2057
return 1;
2058
}
@@ -2101,9 +2087,7 @@ int handle_incoming_traffic(struct mqtt_ng_client *client)
2087
2088
struct mqtt_publish *pub;
2089
struct mqtt_property *prop;
2104
- spinlock_lock(&client->stats_spinlock);
2105
- client->stats.rx_messages_rcvd++;
2106
- spinlock_unlock(&client->stats_spinlock);
2090
+ __atomic_fetch_add(&client->stats.rx_messages_rcvd, 1, __ATOMIC_RELAXED);
2091
2092
uint8_t ctrl_packet_type = get_control_packet_type(client->parser.mqtt_control_packet_type);
2093
switch (ctrl_packet_type) {
@@ -2279,9 +2263,10 @@ void mqtt_ng_set_max_mem(struct mqtt_ng_client *client, size_t bytes)
2263
2264
void mqtt_ng_get_stats(struct mqtt_ng_client *client, struct mqtt_ng_stats *stats)
2265
{
2282
- spinlock_lock(&client->stats_spinlock);
2283
- memcpy(stats, &client->stats, sizeof(struct mqtt_ng_stats));
2284
- spinlock_unlock(&client->stats_spinlock);
2266
+ stats->tx_messages_queued = __atomic_load_n(&client->stats.tx_messages_queued, __ATOMIC_RELAXED);
2267
+ stats->tx_messages_sent = __atomic_load_n(&client->stats.tx_messages_sent, __ATOMIC_RELAXED);
2268
+ stats->rx_messages_rcvd = __atomic_load_n(&client->stats.rx_messages_rcvd, __ATOMIC_RELAXED);
2269
+ stats->packets_waiting_puback = __atomic_load_n(&client->stats.packets_waiting_puback, __ATOMIC_RELAXED);
2270
2271
stats->tx_bytes_queued = 0;
2272
stats->tx_buffer_reclaimable = 0;
src/aclk/mqtt_websockets/mqtt_ng.h
-1
@@ -42,7 +42,6 @@ struct mqtt_auth_properties {
42
int mqtt_ng_connect(struct mqtt_ng_client *client,
43
struct mqtt_auth_properties *auth,
44
struct mqtt_lwt_properties *lwt,
45
- uint8_t clean_start,
45
uint16_t keep_alive);
46
47
int mqtt_ng_publish(struct mqtt_ng_client *client,
src/aclk/mqtt_websockets/mqtt_wss_client.c
+1
-1
@@ -605,7 +605,7 @@ int mqtt_wss_connect(
605
lwt.will_qos = (int) (mqtt_params->will_flags & MQTT_WSS_PUB_QOSMASK);
606
lwt.will_retain = (int) mqtt_params->will_flags & MQTT_WSS_PUB_RETAIN;
607
608
- int ret = mqtt_ng_connect(client->mqtt, &auth, mqtt_params->will_msg ? &lwt : NULL, 1, client->mqtt_keepalive);
608
+ int ret = mqtt_ng_connect(client->mqtt, &auth, mqtt_params->will_msg ? &lwt : NULL, client->mqtt_keepalive);
609
if (ret) {
610
nd_log(NDLS_DAEMON, NDLP_ERR, "Error generating MQTT connect");
611
return 1;