Improved ACLK memory management and shutdown sequence (#8611)
Improved ACLK memory management and shutdown sequence
Stelios Fragkakis committed
Apr 9, 2020 at 18:26 UTC
764a0676e82c89b4a4516a31a7782ea606071fa5
3 files changed
+29
-20
aclk/agent_cloud_link.c
+23
-15
@@ -187,7 +187,7 @@ biofailed:
187
* should be called with
188
*
189
* mode 0 to reset the delay
190
- * mode 1 to sleep for the calculated amount of time [0 .. ACLK_MAX_BACKOFF_DELAY * 1000] ms
190
+ * mode 1 to calculate sleep time [0 .. ACLK_MAX_BACKOFF_DELAY * 1000] ms
191
*
192
*/
193
unsigned long int aclk_reconnect_delay(int mode)
@@ -210,8 +210,6 @@ unsigned long int aclk_reconnect_delay(int mode)
210
delay = (delay * 1000) + (random() % 1000);
211
}
212
213
- // sleep_usec(USEC_PER_MS * delay);
214
-
213
return delay;
214
}
215
@@ -308,7 +306,7 @@ int aclk_queue_query(char *topic, char *data, char *msg_id, char *query, int run
306
if (tmp_query->run_after == run_after) {
307
QUERY_UNLOCK;
308
QUERY_THREAD_WAKEUP;
311
- return 1;
309
+ return 0;
310
}
311
312
if (last_query)
@@ -865,18 +863,22 @@ int aclk_process_queries()
863
static void aclk_query_thread_cleanup(void *ptr)
864
{
865
struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
868
- static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;
866
867
info("cleaning up...");
868
872
- COLLECTOR_LOCK;
873
-
869
_reset_collector_list();
870
freez(collector_list);
871
877
- COLLECTOR_UNLOCK;
872
+ // Clean memory for pending queries if any
873
+ struct aclk_query *this_query;
874
879
- static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
875
+ do {
876
+ this_query = aclk_queue_pop();
877
+ aclk_query_free(this_query);
878
+ } while (this_query);
879
+
880
+ freez(static_thread->thread);
881
+ freez(static_thread);
882
}
883
884
/**
@@ -913,7 +915,7 @@ void *aclk_query_main_thread(void *ptr)
915
if (unlikely(aclk_queue_query("on_connect", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT))) {
916
errno = 0;
917
error("ACLK failed to queue on_connect command");
916
- aclk_metadata_submitted = 0;
918
+ aclk_metadata_submitted = ACLK_METADATA_REQUIRED;
919
}
920
}
921
@@ -973,7 +975,6 @@ static void aclk_main_cleanup(void *ptr)
975
}
976
}
977
976
- info("Disconnected");
978
979
static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
980
}
@@ -1278,7 +1279,6 @@ void *aclk_main(void *ptr)
1279
{
1280
struct netdata_static_thread *query_thread;
1281
1281
- netdata_thread_cleanup_push(aclk_main_cleanup, ptr);
1282
if (!netdata_cloud_setting) {
1283
info("Killing ACLK thread -> cloud functionality has been disabled");
1284
return NULL;
@@ -1318,10 +1318,11 @@ void *aclk_main(void *ptr)
1318
sleep_usec(USEC_PER_SEC * 60);
1319
}
1320
create_publish_base_topic();
1321
- create_private_key();
1321
1322
usec_t reconnect_expiry = 0; // In usecs
1323
1324
+ netdata_thread_disable_cancelability();
1325
+
1326
while (!netdata_exit) {
1327
static int first_init = 0;
1328
size_t write_q, write_q_bytes, read_q;
@@ -1375,7 +1376,8 @@ void *aclk_main(void *ptr)
1376
}
1377
} // forever
1378
exited:
1378
- aclk_shutdown();
1379
+ // Wakeup query thread to cleanup
1380
+ QUERY_THREAD_WAKEUP;
1381
1382
freez(aclk_username);
1383
freez(aclk_password);
@@ -1384,7 +1386,7 @@ exited:
1386
if (aclk_private_key != NULL)
1387
RSA_free(aclk_private_key);
1388
1387
- netdata_thread_cleanup_pop(1);
1389
+ aclk_main_cleanup(ptr);
1390
return NULL;
1391
}
1392
@@ -1869,6 +1871,12 @@ int aclk_handle_cloud_request(char *payload)
1871
return 1;
1872
}
1873
1874
+ // Checked to be "http", not needed anymore
1875
+ if (likely(cloud_to_agent.type_id)) {
1876
+ freez(cloud_to_agent.type_id);
1877
+ cloud_to_agent.type_id = NULL;
1878
+ }
1879
+
1880
if (unlikely(aclk_submit_request(&cloud_to_agent)))
1881
debug(D_ACLK, "ACLK failed to queue incoming message (%s)", payload);
1882
aclk/agent_cloud_link.h
-4
@@ -73,16 +73,12 @@ void *aclk_main(void *ptr);
73
74
extern int aclk_send_message(char *sub_topic, char *message, char *msg_id);
75
76
-//int aclk_init();
77
-//char *get_base_topic();
78
-
76
extern char *is_agent_claimed(void);
77
extern void aclk_lws_wss_mqtt_layer_disconect_notif();
78
char *create_uuid();
79
80
// callbacks for agent cloud link
81
int aclk_subscribe(char *topic, int qos);
85
-void aclk_shutdown();
82
int cloud_to_agent_parse(JSON_ENTRY *e);
83
void aclk_disconnect();
84
void aclk_connect();
aclk/mqtt.c
+6
-1
@@ -52,7 +52,12 @@ void disconnect_callback(struct mosquitto *mosq, void *obj, int rc)
52
UNUSED(obj);
53
UNUSED(rc);
54
55
- info("Connection to cloud failed");
55
+ if (netdata_exit)
56
+ info("Connection to cloud terminated due to agent shutdown");
57
+ else {
58
+ errno = 0;
59
+ error("Connection to cloud failed");
60
+ }
61
aclk_disconnect();
62
63
aclk_lws_wss_mqtt_layer_disconect_notif();