Handle mqtt ping timeouts (#18653)
* Handle mqtt ping timeouts * Increase ping timeout * Reset ping_timeout when disconnection is requested
Stelios Fragkakis committed
Oct 2, 2024 at 18:32 UTC
934fa2e1090a05211253051aa87f443863cc44b6
7 files changed
+52
-22
src/aclk/aclk.c
+21
-9
@@ -23,7 +23,6 @@
23
int aclk_pubacks_per_conn = 0; // How many PubAcks we got since MQTT conn est.
24
int aclk_rcvd_cloud_msgs = 0;
25
int aclk_connection_counter = 0;
26
-int disconnect_req = 0;
26
27
static bool aclk_connected = false;
28
static inline void aclk_set_connected(void) {
@@ -51,7 +50,8 @@ bool aclk_online_for_nodes(void) {
50
51
int aclk_ctx_based = 0;
52
int aclk_disable_runtime = 0;
54
-int aclk_kill_link = 0;
53
+
54
+ACLK_DISCONNECT_ACTION disconnect_req = ACLK_NO_DISCONNECT;
55
56
usec_t aclk_session_us = 0;
57
time_t aclk_session_sec = 0;
@@ -301,14 +301,26 @@ static int handle_connection(mqtt_wss_client client)
301
return 1;
302
}
303
304
- if (disconnect_req || aclk_kill_link) {
305
- nd_log(NDLS_DAEMON, NDLP_NOTICE,
306
- "Going to restart connection due to disconnect_req=%s (cloud req), aclk_kill_link=%s (reclaim)",
307
- disconnect_req ? "true" : "false",
308
- aclk_kill_link ? "true" : "false");
304
+ if (disconnect_req != ACLK_NO_DISCONNECT) {
305
+ const char *reason;
306
+ switch (disconnect_req) {
307
+ case ACLK_CLOUD_DISCONNECT:
308
+ reason = "cloud request";
309
+ break;
310
+ case ACLK_PING_TIMEOUT:
311
+ reason = "ping timeout";
312
+ break;
313
+ case ACLK_RELOAD_CONF:
314
+ reason = "reclaim";
315
+ break;
316
+ default:
317
+ reason = "unknown";
318
+ break;
319
+ }
320
+
321
+ nd_log(NDLS_DAEMON, NDLP_NOTICE, "Going to restart connection due to \"%s\"", reason);
322
310
- disconnect_req = 0;
311
- aclk_kill_link = 0;
323
+ disconnect_req = ACLK_NO_DISCONNECT;
324
aclk_graceful_disconnect(client);
325
aclk_shared_state.mqtt_shutdown_msg_id = -1;
326
aclk_shared_state.mqtt_shutdown_msg_rcvd = 0;
src/aclk/aclk.h
+8
-1
@@ -11,6 +11,13 @@
11
// stable for the purposes of TBEB (truncated binary exponential backoff)
12
#define ACLK_PUBACKS_CONN_STABLE 3
13
14
+typedef enum {
15
+ ACLK_NO_DISCONNECT = 0,
16
+ ACLK_CLOUD_DISCONNECT = 1,
17
+ ACLK_RELOAD_CONF = 2,
18
+ ACLK_PING_TIMEOUT = 3
19
+} ACLK_DISCONNECT_ACTION;
20
+
21
typedef enum __attribute__((packed)) {
22
ACLK_STATUS_CONNECTED = 0,
23
ACLK_STATUS_NONE,
@@ -62,7 +69,7 @@ extern time_t aclk_session_sec;
69
extern time_t aclk_block_until;
70
71
extern int aclk_connection_counter;
65
-extern int disconnect_req;
72
+extern ACLK_DISCONNECT_ACTION disconnect_req;
73
74
void *aclk_main(void *ptr);
75
src/aclk/aclk_rx_msgs.c
+1
-1
@@ -407,7 +407,7 @@ int handle_disconnect_req(const char *msg, size_t msg_len)
407
"Cloud asks not to reconnect for %u seconds. We shall honor that request",
408
(unsigned int)cmd->reconnect_after_s);
409
}
410
- disconnect_req = 1;
410
+ disconnect_req = ACLK_CLOUD_DISCONNECT;
411
freez(cmd->error_description);
412
freez(cmd);
413
return 0;
src/aclk/mqtt_websockets/mqtt_ng.c
+1
@@ -1804,6 +1804,7 @@ static int parse_data(struct mqtt_ng_client *client)
1804
return MQTT_NG_CLIENT_PROTOCOL_ERROR;
1805
}
1806
parser->state = MQTT_PARSE_MQTT_PACKET_DONE;
1807
+ ping_timeout = 0;
1808
break;
1809
case MQTT_CPT_DISCONNECT:
1810
rc = parse_disconnect_varhdr(client);
src/aclk/mqtt_websockets/mqtt_ng.h
+1
-1
@@ -10,7 +10,7 @@
10
#define MQTT_NG_MSGGEN_MSG_TOO_BIG 3
11
12
struct mqtt_ng_client;
13
-
13
+extern time_t ping_timeout;
14
/* Converts integer to MQTT Variable Byte Integer as per 1.5.5 of MQTT 5 specs
15
* @param input value to be converted
16
* @param output pointer to memory where output will be written to. Must allow up to 4 bytes to be written.
src/aclk/mqtt_websockets/mqtt_wss_client.c
+19
-9
@@ -9,12 +9,16 @@
9
#include "mqtt_ng.h"
10
#include "ws_client.h"
11
#include "common_internal.h"
12
+#include "../aclk.h"
13
14
#define PIPE_READ_END 0
15
#define PIPE_WRITE_END 1
16
#define POLLFD_SOCKET 0
17
#define POLLFD_PIPE 1
18
19
+#define PING_TIMEOUT (60) //Expect a ping response within this time (seconds)
20
+time_t ping_timeout = 0;
21
+
22
#if (OPENSSL_VERSION_NUMBER < OPENSSL_VERSION_110) && (SSLEAY_VERSION_NUMBER >= OPENSSL_VERSION_097)
23
#include <openssl/conf.h>
24
#endif
@@ -744,13 +748,11 @@ static int handle_mqtt_internal(mqtt_wss_client client)
748
return 0;
749
}
750
747
-#define SEC_TO_MSEC 1000
748
-static long long int t_till_next_keepalive_ms(mqtt_wss_client client)
751
+static int t_till_next_keepalive_ms(mqtt_wss_client client)
752
{
753
time_t last_send = mqtt_ng_last_send_time(client->mqtt);
751
- long long int next_mqtt_keep_alive = (last_send * SEC_TO_MSEC)
752
- + (client->mqtt_keepalive * (SEC_TO_MSEC * 0.75 /* SEND IN ADVANCE */));
753
- return(next_mqtt_keep_alive - (time(NULL) * SEC_TO_MSEC));
754
+ time_t next_mqtt_keep_alive = last_send + client->mqtt_keepalive * 0.75;
755
+ return ((next_mqtt_keep_alive - now_realtime_sec()) * MSEC_PER_SEC);
756
}
757
758
#ifdef MQTT_WSS_CPUSTATS
@@ -777,10 +779,12 @@ int mqtt_wss_service(mqtt_wss_client client, int timeout_ms)
779
#endif
780
781
// Check user requested TO doesn't interfere with MQTT keep alives
780
- long long int till_next_keep_alive = t_till_next_keepalive_ms(client);
781
- if (client->mqtt_connected && (timeout_ms < 0 || timeout_ms >= till_next_keep_alive)) {
782
- timeout_ms = till_next_keep_alive;
783
- send_keepalive = 1;
782
+ if (!ping_timeout) {
783
+ int till_next_keep_alive = t_till_next_keepalive_ms(client);
784
+ if (client->mqtt_connected && (timeout_ms < 0 || timeout_ms >= till_next_keep_alive)) {
785
+ timeout_ms = till_next_keep_alive;
786
+ send_keepalive = 1;
787
+ }
788
}
789
790
#ifdef MQTT_WSS_CPUSTATS
@@ -802,11 +806,17 @@ int mqtt_wss_service(mqtt_wss_client client, int timeout_ms)
806
#endif
807
808
if (ret == 0) {
809
+ time_t now = now_realtime_sec();
810
if (send_keepalive) {
811
// otherwise we shortened the timeout ourselves to take care of
812
// MQTT keep alives
813
mqtt_ng_ping(client->mqtt);
814
+ ping_timeout = now + PING_TIMEOUT;
815
} else {
816
+ if (ping_timeout && ping_timeout < now) {
817
+ disconnect_req = ACLK_PING_TIMEOUT;
818
+ ping_timeout = 0;
819
+ }
820
// if poll timed out and user requested timeout was being used
821
// return here let user do his work and he will call us back soon
822
return 0;
src/claim/claim.c
+1
-1
@@ -148,7 +148,7 @@ bool load_claiming_state(void) {
148
if (aclk_online()) {
149
nd_log(NDLS_DAEMON, NDLP_ERR,
150
"CLAIM: agent was already connected to NC - forcing reconnection under new credentials");
151
- aclk_kill_link = 1;
151
+ disconnect_req = ACLK_RELOAD_CONF;
152
}
153
aclk_disable_runtime = 0;
154