Fix crash caused by concurrent access to MQTT sink #40
Conversation
Prior to this fix messages received through a Zenoh subscription were sent to the related MQTT sink directly from the Zenoh callback. Invoking the MQTT sink in this way leads to the ntex-mqtt implementations write cache being accessed from the calling thread. At the same time it is possible for the ntex-mqtt server worker thread to also access this write cache as part of its background tasks. As the write cache is designed to be accessed in a single threaded context only this lead to an abort. To fix this an MPSC channel is introduced allowing the Zenoh subscriber callbacks to send messages to an asynchronous task running within the ntex-mqtt server which then forwards the message to MQTT. This fix was inspired by this discussion: ntex-rs/ntex-mqtt#66
|
@gmartin82 If this pull request contains a bugfix or a new feature, then please consider using |
gmartin82
changed the title
|
@gmartin82 If this pull request contains a bugfix or a new feature, then please consider using |
|
Note that while this PR fixes the crash due to concurrent access it exposes another issue which occurs when data is being forwarded from Zenoh at a high frequency. While data is successfully forwarded to connected MQTT clients ntex-mqtt isn't able to handle new MQTT connection requests or keep-alive messages. This seems to be a limitation of the current ntex-mqtt implementation. I was only able to trigger this secondary issue when sending data as fast as possible from Zenoh to MQTT. When small sleeps were introduced between publishing messages (e.g. 1 ms) everything worked as expected. |
Before this fix messages received through a Zenoh subscription were sent to the related MQTT sink directly from the Zenoh callback. Invoking the MQTT sink in this way leads to the ntex-mqtt implementations write cache being accessed from the calling thread. At the same time, it is possible for the ntex-mqtt server worker thread to also access this write cache as part of its background tasks. As the write cache is designed to be accessed in a single-threaded context only this can lead to an abort.
To fix this an MPSC channel is introduced allowing the Zenoh subscriber callbacks to send messages to an asynchronous task running within the ntex-mqtt server which then forwards the message to MQTT. This fix was inspired by this discussion: ntex-rs/ntex-mqtt#66
Fixes #39