Aclk improvements (netdata#20775)

* Code cleanup

* Collect packets that have time out then do cleanup

* No need to run cleanup on demand

* Add a periodic cleanup
check_packet_monitor_list_for_timeouts return true of max cleanup was performed
Keep stats of pending packets waiting puback

* Replace mqtt_wss_now_usec with now_monotonic_usec

* Cleanup packet timeput monitor list on disconnection and clean reconnection -- reset statistics as well
Keep chart of puback wait messages

* Additional statistics in aclk-state

* Set timestamp on query creation

* Query threads and MQTT ACK processing now acquire locks in the same order (pending_packets → main_buffer) to prevent AB-BA deadlock.

Author: stelfrag

src/aclk/aclk.c

@@ -1089,7 +1089,13 @@ char *aclk_state(void)
         buffer_sprintf(wb, "Yes\nClaimed Id: %s\nCloud URL: %s\nACLK Proxy: %s\nPublish Latency: %s\n", claim_id.str, cloud_base_url ? cloud_base_url : "null", aclk_proxy ? aclk_proxy : "none", latency_str);
     }
 
-    buffer_sprintf(wb, "Online: %s\nReconnect count: %d\nBanned By Cloud: %s\n", aclk_online() ? "Yes" : "No", aclk_connection_counter > 0 ? (aclk_connection_counter - 1) : 0, aclk_disable_runtime ? "Yes" : "No");
+    bool aclk_is_online = aclk_online();
+
+    struct mqtt_wss_stats aclk_stats;
+    if (aclk_is_online)
+        aclk_stats = aclk_statistics();
+
+    buffer_sprintf(wb, "Online: %s\nReconnect count: %d\nBanned By Cloud: %s\n", aclk_is_online ? "Yes" : "No", aclk_connection_counter > 0 ? (aclk_connection_counter - 1) : 0, aclk_disable_runtime ? "Yes" : "No");
     if (last_conn_time_mqtt && ((tmptr = localtime_r(&last_conn_time_mqtt, &tmbuf))) ) {
         char timebuf[26];
         strftime(timebuf, 26, "%Y-%m-%d %H:%M:%S", tmptr);
@@ -1111,8 +1117,9 @@ char *aclk_state(void)
         buffer_sprintf(wb, "Next Connection Attempt At: %s\nLast Backoff: %.3f", timebuf, last_backoff_value);
     }
 
-    if (aclk_online()) {
-        buffer_sprintf(wb, "Received Cloud MQTT Messages: %d\nMQTT Messages Confirmed by Remote Broker (PUBACKs): %d", aclk_rcvd_cloud_msgs, aclk_pubacks_per_conn);
+    if (aclk_is_online) {
+        buffer_sprintf(wb, "Received Cloud MQTT Messages: %d\nMQTT Messages Confirmed by Remote Broker (PUBACKs): %d\nPending PUBACKS: %d\n",
+                       aclk_rcvd_cloud_msgs, aclk_pubacks_per_conn, aclk_stats.mqtt.packets_waiting_puback);
 
         RRDHOST *host;
         rrd_rdlock();
@@ -1186,6 +1193,16 @@ static json_object *timestamp_to_json(const time_t *t)
 
 char *aclk_state_json(void)
 {
+
+    bool aclk_is_online = aclk_online();
+
+    struct mqtt_wss_stats aclk_stats;
+
+    if (aclk_is_online)
+        aclk_stats = aclk_statistics();
+    else
+        memset(&aclk_stats, 0, sizeof(aclk_stats));
+
     json_object *tmp, *grp, *msg = json_object_new_object();
 
     tmp = json_object_new_boolean(1);
@@ -1221,7 +1238,7 @@ char *aclk_state_json(void)
     tmp =json_object_new_int64((int64_t) latency);
     json_object_object_add(msg, "publish_latency_us", tmp);
 
-    tmp = json_object_new_boolean(aclk_online());
+    tmp = json_object_new_boolean(aclk_is_online);
     json_object_object_add(msg, "online", tmp);
 
     tmp = json_object_new_string("Protobuf");
@@ -1236,13 +1253,16 @@ char *aclk_state_json(void)
     tmp = json_object_new_int(aclk_pubacks_per_conn);
     json_object_object_add(msg, "received-mqtt-pubacks", tmp);
 
+    tmp = json_object_new_int((int32_t) aclk_stats.mqtt.packets_waiting_puback);
+    json_object_object_add(msg, "pending-mqtt-pubacks", tmp);
+
     tmp = json_object_new_int(aclk_connection_counter > 0 ? (aclk_connection_counter - 1) : 0);
     json_object_object_add(msg, "reconnect-count", tmp);
 
     json_object_object_add(msg, "last-connect-time-utc", timestamp_to_json(&last_conn_time_mqtt));
     json_object_object_add(msg, "last-connect-time-puback-utc", timestamp_to_json(&last_conn_time_appl));
     json_object_object_add(msg, "last-disconnect-time-utc", timestamp_to_json(&last_disconnect_time));
-    json_object_object_add(msg, "next-connection-attempt-utc", !aclk_online() ? timestamp_to_json(&next_connection_attempt) : NULL);
+    json_object_object_add(msg, "next-connection-attempt-utc", !aclk_is_online ? timestamp_to_json(&next_connection_attempt) : NULL);
     tmp = NULL;
     if (!aclk_online() && last_backoff_value)
         tmp = json_object_new_double(last_backoff_value);

src/aclk/aclk_query_queue.c

@@ -58,6 +58,7 @@ aclk_query_t *aclk_query_new(aclk_query_type_t type)
 {
     aclk_query_t *query = get_query();
     query->type = type;
+    now_monotonic_high_precision_timeval(&query->created_tv);
     return query;
 }
 

src/aclk/aclk_rx_msgs.c

@@ -233,12 +233,9 @@ int handle_old_proto_cmd(const char *msg, size_t msg_len)
     char *str = mallocz(msg_len+1);
     memcpy(str, msg, msg_len);
     str[msg_len] = 0;
-    if (aclk_handle_cloud_cmd_message(str)) {
-        freez(str);
-        return 1;
-    }
+    int rc = aclk_handle_cloud_cmd_message(str);
     freez(str);
-    return 0;
+    return rc;
 }
 
 int create_node_instance_result(const char *msg, size_t msg_len)

src/aclk/mqtt_websockets/common_public.h

@@ -25,6 +25,7 @@ struct mqtt_ng_stats {
     int tx_messages_queued;
     int tx_messages_sent;
     int rx_messages_rcvd;
+    int packets_waiting_puback;
     size_t tx_buffer_used;
     size_t tx_buffer_free;
     size_t tx_buffer_size;

src/aclk/mqtt_websockets/mqtt_ng.c

@@ -658,15 +658,21 @@ static void mqtt_ng_destroy_tx_alias_hash(c_rhash hash)
     c_rhash_destroy(hash);
 }
 
+static void destroy_timeout_monitor_list(struct mqtt_ng_client *client)
+{
+    spinlock_lock(&client->pending_packets.spinlock);
+    (void) JudyLFreeArray(&client->pending_packets.JudyL, PJE0);
+    spinlock_unlock(&client->pending_packets.spinlock);
+    __atomic_store_n(&client->stats.packets_waiting_puback, 0, __ATOMIC_RELAXED);
+}
+
 void mqtt_ng_destroy(struct mqtt_ng_client *client)
 {
     transaction_buffer_destroy(&client->main_buffer);
 
     mqtt_ng_destroy_tx_alias_hash(client->tx_topic_aliases.stoi_dict);
     mqtt_ng_destroy_rx_alias_hash(client->rx_aliases);
-    spinlock_lock(&client->pending_packets.spinlock);
-    (void) JudyLFreeArray(&client->pending_packets.JudyL, PJE0);
-    spinlock_unlock(&client->pending_packets.spinlock);
+    destroy_timeout_monitor_list(client);
     freez(client);
 }
 
@@ -798,11 +804,12 @@ static int optimized_add(struct header_buffer *buf, void *data, size_t data_len,
     return 0;
 }
 
-static void remove_packet_from_timeout_monitor_list(struct mqtt_ng_client *client, uint16_t packet_id)
+static void remove_packet_from_timeout_monitor_list_unsafe(struct mqtt_ng_client *client, uint16_t packet_id)
 {
-    spinlock_lock(&client->pending_packets.spinlock);
-    (void) JudyLDel(&client->pending_packets.JudyL, (Word_t) packet_id, PJE0);
-    spinlock_unlock(&client->pending_packets.spinlock);
+    int rc = JudyLDel(&client->pending_packets.JudyL, (Word_t) packet_id, PJE0);
+    // rc = 1 if the packer was deleted, so update statistics
+    if (likely(rc))
+        __atomic_fetch_sub(&client->stats.packets_waiting_puback, 1, __ATOMIC_RELAXED);
 }
 
 #define PACKET_TIMEOUT_EPOCH (1704067200L)  // Jan 1, 2024 00:00:00 UTC
@@ -813,13 +820,15 @@ static void add_packet_to_timeout_monitor_list(struct mqtt_ng_client *client, ui
     time_t now = now_realtime_sec();
     // Add it to the JudyL array
     uint32_t *Pvalue = (uint32_t *) JudyLIns(&client->pending_packets.JudyL, (Word_t) packet_id, PJE0);
-    if (!Pvalue || Pvalue == PJERR) {
+    if (Pvalue == PJERR) {
         nd_log(NDLS_DAEMON, NDLP_ERR, "Error inserting packet_id (%" PRIu16 ") into JudyL array.", packet_id);
         spinlock_unlock(&client->pending_packets.spinlock);
         return;
     }
     *Pvalue = (uint32_t) ((now - PACKET_TIMEOUT_EPOCH) + PACKET_ACK_TIMEOUT_SECS);
     spinlock_unlock(&client->pending_packets.spinlock);
+
+    __atomic_fetch_add(&client->stats.packets_waiting_puback, 1, __ATOMIC_RELAXED);
 }
 
 #define TRY_GENERATE_MESSAGE(generator_function, ...)                                                                  \
@@ -1016,6 +1025,9 @@ int mqtt_ng_connect(struct mqtt_ng_client *client,
         buffer_purge(&client->main_buffer.hdr_buffer);
     UNLOCK_HDR_BUFFER(&client->main_buffer);
 
+    if (clean_start)
+        destroy_timeout_monitor_list(client);
+
     spinlock_lock(&client->tx_topic_aliases.spinlock);
     // according to MQTT spec topic aliases should not be persisted
     // even if clean session is true
@@ -1159,13 +1171,15 @@ static void mark_message_for_gc(struct buffer_fragment *frag)
 static int mark_packet_acked(struct mqtt_ng_client *client, uint16_t packet_id)
 {
     size_t reclaimable = 0;
+    spinlock_lock(&client->pending_packets.spinlock);
     LOCK_HDR_BUFFER(&client->main_buffer);
     struct buffer_fragment *frag = BUFFER_FIRST_FRAG(&client->main_buffer.hdr_buffer);
     while (frag) {
         if ( (frag->flags & BUFFER_FRAG_MQTT_PACKET_HEAD) && frag->packet_id == packet_id) {
             if (!frag->sent) {
                 nd_log(NDLS_DAEMON, NDLP_ERR, "Received packet_id (%" PRIu16 ") belongs to MQTT packet which was not yet sent!", packet_id);
                 UNLOCK_HDR_BUFFER(&client->main_buffer);
+                spinlock_unlock(&client->pending_packets.spinlock);
                 return 1;
             }
             usec_t latency = now_monotonic_usec() - frag->sent_monotonic_ut;
@@ -1178,7 +1192,8 @@ static int mark_packet_acked(struct mqtt_ng_client *client, uint16_t packet_id)
                 transaction_buffer_garbage_collect(&client->main_buffer);
 
             UNLOCK_HDR_BUFFER(&client->main_buffer);
-            remove_packet_from_timeout_monitor_list(client, packet_id);
+            remove_packet_from_timeout_monitor_list_unsafe(client, packet_id);
+            spinlock_unlock(&client->pending_packets.spinlock);
             return 0;
         }
 
@@ -1189,25 +1204,40 @@ static int mark_packet_acked(struct mqtt_ng_client *client, uint16_t packet_id)
     }
     nd_log(NDLS_DAEMON, NDLP_ERR, "Received packet_id (%" PRIu16 ") is unknown!", packet_id);
     UNLOCK_HDR_BUFFER(&client->main_buffer);
+    spinlock_unlock(&client->pending_packets.spinlock);
     return 1;
 }
 
-static void check_packet_monitor_list_for_timeouts(struct mqtt_ng_client *client)
+#define MAX_TIMED_OUT_PACKETS (1024)
+
+static bool check_packet_monitor_list_for_timeouts(struct mqtt_ng_client *client)
 {
+    uint16_t timed_out_packets[MAX_TIMED_OUT_PACKETS];
+    size_t timed_out_count = 0;
+
     spinlock_lock(&client->pending_packets.spinlock);
     bool first_then_next = true;
     uint32_t *Pvalue;
     Word_t packet_id = 0;
     time_t now = now_realtime_sec();
+
     while ((Pvalue = (uint32_t *) JudyLFirstThenNext(client->pending_packets.JudyL, &packet_id, &first_then_next))) {
         uint32_t expire_time_delta = *Pvalue;
         if (now >= (PACKET_TIMEOUT_EPOCH + expire_time_delta)) {
-            spinlock_unlock(&client->pending_packets.spinlock);
-            (void) mark_packet_acked(client, (uint16_t) packet_id);
-            spinlock_lock(&client->pending_packets.spinlock);
+            if (timed_out_count < MAX_TIMED_OUT_PACKETS) {
+                timed_out_packets[timed_out_count++] = (uint16_t)packet_id;
+            } else
+                break;
         }
     }
     spinlock_unlock(&client->pending_packets.spinlock);
+
+    // Process timeouts outside the lock
+    for (size_t i = 0; i < timed_out_count; i++) {
+        mark_packet_acked(client, timed_out_packets[i]);
+    }
+
+    return (timed_out_count ==  MAX_TIMED_OUT_PACKETS);
 }
 
 #define PUBLISH_SP_SIZE 64
@@ -1244,11 +1274,6 @@ int mqtt_ng_publish(struct mqtt_ng_client *client,
     }
 
     int rc = TRY_GENERATE_MESSAGE(mqtt_ng_generate_publish, topic, topic_free, msg, msg_free, msg_len, publish_flags, packet_id, topic_id);
-    if (rc == MQTT_NG_MSGGEN_BUFFER_OOM) {
-        check_packet_monitor_list_for_timeouts(client);
-        rc = TRY_GENERATE_MESSAGE(mqtt_ng_generate_publish, topic, topic_free, msg, msg_free, msg_len, publish_flags, packet_id, topic_id);
-    }
-
     if (rc == MQTT_NG_MSGGEN_OK)
         add_packet_to_timeout_monitor_list(client, *packet_id);
     return rc;
@@ -1930,7 +1955,6 @@ static int parse_data(struct mqtt_ng_client *client)
                     }
                     parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
                     ping_timeout = 0;
-                    check_packet_monitor_list_for_timeouts(client);
                     break;
                 case MQTT_CPT_DISCONNECT:
                     rc = parse_disconnect_varhdr(client);
@@ -2171,6 +2195,8 @@ int handle_incoming_traffic(struct mqtt_ng_client *client)
     return rc;
 }
 
+#define PACKET_TIMEOUT_REPEAT_CHECK (60)
+
 int mqtt_ng_sync(struct mqtt_ng_client *client)
 {
     if (client->client_state == MQTT_STATE_RAW || client->client_state == MQTT_STATE_DISCONNECTED)
@@ -2179,6 +2205,15 @@ int mqtt_ng_sync(struct mqtt_ng_client *client)
     if (client->client_state == MQTT_STATE_ERROR)
         return 1;
 
+    // Check for packet timeouts and cleanup
+    static time_t last_maintenance = 0;
+    if (now_realtime_sec() - last_maintenance >= PACKET_TIMEOUT_REPEAT_CHECK) {
+        // if check packet returns true then we did max cleanup, possibly there are more packets to cleanup
+        // so do not update last_maintenance thus forcing check again
+        if (likely(!check_packet_monitor_list_for_timeouts(client)))
+            last_maintenance = now_realtime_sec();
+    }
+
     worker_is_busy(WORKER_ACLK_TRY_SEND_ALL);
 
     LOCK_HDR_BUFFER(&client->main_buffer);

src/aclk/mqtt_websockets/mqtt_wss_client.c

@@ -754,17 +754,6 @@ static int t_till_next_keepalive_ms(mqtt_wss_client client)
     return timeout_ms;
 }
 
-#ifdef MQTT_WSS_CPUSTATS
-static uint64_t mqtt_wss_now_usec(void) {
-    struct timespec ts;
-    if(clock_gettime(CLOCK_MONOTONIC, &ts) == -1) {
-        nd_log(NDLS_DAEMON, NDLP_ERR, "clock_gettime(CLOCK_MONOTONIC, &timespec) failed.");
-        return 0;
-    }
-    return (uint64_t)ts.tv_sec * USEC_PER_SEC + (ts.tv_nsec % NSEC_PER_SEC) / NSEC_PER_USEC;
-}
-#endif
-
 int mqtt_wss_service(mqtt_wss_client client, int timeout_ms)
 {
     char *ptr;
@@ -774,7 +763,7 @@ int mqtt_wss_service(mqtt_wss_client client, int timeout_ms)
 
 #ifdef MQTT_WSS_CPUSTATS
     uint64_t t2;
-    uint64_t t1 = mqtt_wss_now_usec();
+    uint64_t t1 = now_monotonic_usec();
 #endif
 
     // Check user requested TO doesn't interfere with MQTT keep alives
@@ -787,7 +776,7 @@ int mqtt_wss_service(mqtt_wss_client client, int timeout_ms)
     }
 
 #ifdef MQTT_WSS_CPUSTATS
-    t2 = mqtt_wss_now_usec();
+    t2 = now_monotonic_usec();
     client->stats.time_keepalive += t2 - t1;
 #endif
 
@@ -805,7 +794,7 @@ int mqtt_wss_service(mqtt_wss_client client, int timeout_ms)
     worker_is_busy(WORKER_ACLK_POLL_OK);
 
 #ifdef MQTT_WSS_CPUSTATS
-    t1 = mqtt_wss_now_usec();
+    t1 = now_monotonic_usec();
 #endif
 
     if (ret == 0) {
@@ -828,7 +817,7 @@ int mqtt_wss_service(mqtt_wss_client client, int timeout_ms)
     }
 
 #ifdef MQTT_WSS_CPUSTATS
-    t2 = mqtt_wss_now_usec();
+    t2 = now_monotonic_usec();
     client->stats.time_keepalive += t2 - t1;
 #endif
 
@@ -866,7 +855,7 @@ int mqtt_wss_service(mqtt_wss_client client, int timeout_ms)
     }
 
 #ifdef MQTT_WSS_CPUSTATS
-    t1 = mqtt_wss_now_usec();
+    t1 = now_monotonic_usec();
     client->stats.time_read_socket += t1 - t2;
 #endif
 
@@ -890,7 +879,7 @@ int mqtt_wss_service(mqtt_wss_client client, int timeout_ms)
     }
 
 #ifdef MQTT_WSS_CPUSTATS
-    t2 = mqtt_wss_now_usec();
+    t2 = now_monotonic_usec();
     client->stats.time_process_websocket += t2 - t1;
 #endif
 
@@ -907,7 +896,7 @@ int mqtt_wss_service(mqtt_wss_client client, int timeout_ms)
     }
 
 #ifdef MQTT_WSS_CPUSTATS
-    t1 = mqtt_wss_now_usec();
+    t1 = now_monotonic_usec();
     client->stats.time_process_mqtt += t1 - t2;
 #endif
 
@@ -945,7 +934,7 @@ int mqtt_wss_service(mqtt_wss_client client, int timeout_ms)
         util_clear_pipe(client->write_notif_pipe[PIPE_READ_END]);
 
 #ifdef MQTT_WSS_CPUSTATS
-    t2 = mqtt_wss_now_usec();
+    t2 = now_monotonic_usec();
     client->stats.time_write_socket += t2 - t1;
 #endif
 

src/daemon/pulse/pulse-network.c

@@ -315,6 +315,7 @@ void pulse_network_do(bool extended __maybe_unused) {
         if(extended) {
             static RRDSET *st_aclk_queue_size = NULL;
             static RRDDIM *rd_messages = NULL;
+            static RRDDIM *rd_puback_wait = NULL;
 
             if (unlikely(!st_aclk_queue_size)) {
                 st_aclk_queue_size = rrdset_create_localhost(
@@ -334,9 +335,11 @@ void pulse_network_do(bool extended __maybe_unused) {
                 rrdlabels_add(st_aclk_queue_size->rrdlabels, "endpoint", "aclk", RRDLABEL_SRC_AUTO);
 
                 rd_messages = rrddim_add(st_aclk_queue_size, "messages", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
+                rd_puback_wait = rrddim_add(st_aclk_queue_size, "puback wait", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
             }
 
             rrddim_set_by_pointer(st_aclk_queue_size, rd_messages, (collected_number)t.mqtt.tx_messages_queued);
+            rrddim_set_by_pointer(st_aclk_queue_size, rd_puback_wait, (collected_number)t.mqtt.packets_waiting_puback);
             rrdset_done(st_aclk_queue_size);
         }