Mirrors/VILLASnode
1
0
Fork 0
mirror of https://git.rwth-aachen.de/acs/public/villas/node/ synced 2025-03-09 00:00:00 +01:00
Code Issues Releases Wiki Activity

improve unthreaded implementation of MQTT node type

Browse source 
- set cancel type of MQTT communication management thread to asynchronous
- fix removing node from list in mqtt_stop
- add a debug output upon invocation of pthread cancel in mqtt_type_stop
- contributes to #248
This commit is contained in:
Sonja Happ 2019-09-13 09:56:50 +02:00 • committed by Steffen Vogel
parent 3624cb9fca
commit e52d2f77ee
1 changed files with 13 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
lib/nodes/mqtt.cpp
View file

@ -37,6 +37,13 @@ static pthread_t thread;
static void * mosquitto_loop_thread(void *ctx)
{
int ret;
// set the cancel type of this thread to async
ret = pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, nullptr);
if (ret != 0) {
error("Unable to set cancel type of MQTT communication thread to asynchronous.");
return nullptr;
}
while(true){
for (unsigned i = 0; i < vlist_length(&clients); i++) {
struct node *node = (struct node *) vlist_at(&clients, i);
@ -382,14 +389,15 @@ int mqtt_stop(struct node *n)
int ret;
struct mqtt *m = (struct mqtt *) n->_vd;
// unregister client from global MQTT client list
// so that mosquitto loop is no longer invoked for this client
// important to do that before disconnecting from broker, otherwise, mosquitto thread will attempt to reconnect
vlist_remove(&clients, vlist_index(&clients, n));
ret = mosquitto_disconnect(m->client);
if (ret)
goto mosquitto_error;
// unregister client from global MQTT client list
// so that mosquitto loop is no longer invoked for this client
vlist_remove_all(&clients, n);
ret = io_destroy(&m->io);
if (ret)
return ret;
@ -437,6 +445,7 @@ int mqtt_type_stop()
ret = pthread_cancel(thread);
if (ret)
return ret;
debug( 3, "Called pthread_cancel() on MQTT communication management thread.");
ret = pthread_join(thread, nullptr);
if (ret) {