MQTTnet
MQTTnet copied to clipboard
dotnet
Client support for ReadOnlySequence as payload
Support ReadOnlySequence as a payload in the MQTTnet client, for the new version5 release.
The MQTTNet write method only accepts an ArraySegment which is a view into a single buffer. It is currently necessary to call ToArray to support a RecyclableMemoryStream.
As a result, perfomance is affected by additional copies and memory is fragment by the additional random allocation of a big buffer.
Starting the port to the lates code to gather feedback.
see #1917 and #1918 with the issue explained.
Hi @chkr1011, I am now looking into the option of passing a RecyclableMemoryStream (or a custom MemoryStream) into the reader to also benefit from better buffer management using ReadOnlySequence on the receiver side. Have you thought about such an option?
@mregen I never used that RecyclableMemoryStream before. The idea of not increasing the buffer every time (only doing this one time when GetBuffer is called) looks like a reasonable optimization.
Isn't this something we can also do in the current writer implementation?
Shall we finish this brach first and move the change to another branch? I am afraid that this branch will grow too much.
@mregen I never used that RecyclableMemoryStream before. The idea of not increasing the buffer every time (only doing this one time when GetBuffer is called) looks like a reasonable optimization.
Isn't this something we can also do in the current writer implementation?
Shall we finish this brach first and move the change to another branch? I am afraid that this branch will grow too much.
Hi @chkr1011, currently I am trying to work out the details which we need at the interface level to allow for the lifetime management of the ReadOnlySequence Payload using the PayloadOwner if we use buffers for the payload. It doesn't have to be RecyclablaMemoryStream, but something that can take advantage of buffers and ReadOnlySequence (My preference is currently ArrayPool based bufferning because it allows for readjustable buffer size).
For the publish side the lifetime management of the payload relatively easy because the Payload can be disposed after the PublishAsync call returns.
Its more complicated with the MqttApplicationMessageReceivedEventArgs. The buffers are allocated in the MQTT receiver and best case should be returned to the ArrayPool when the event callback retrurns.
However, I see a lot of code which puts the application message or the event class itself with the payload in some queue or list for processing. Then it may be a problem for the application to dispose the buffers when the event returns.
in this case the application needs to take ownership and ensure that the buffers are disposed once the processing is done.
in this PR I added the necessary functionality in the event and app message to get it working for that use case, but the usability is yet error prone and not a beauty. Any ideas and recommendations are highly welcome.
I agree perf optimizations in MqttBufferWriter and MqttBufferReader can be done later. There is also a double copy in ReceivedMqttPacket which can be improved, I just let it use ArrayPool for now.
I you have an idea how to better implement the TransferPayload method please let me know too.
I had also been thinking about creating a IReadOnlySequenceOwner similar to the IMemoryOwner concept for the payload, but ended up doing it rather straightforward by adding the PayloadOwner and some helpers.
Hi @chkr1011, please have a look now and provide feedback. From my side the functionality for buffer ownership handling is included. Apparently there is a breaking change in the client for the lifetime of the payload, its disposed when the event callback returns, to make efficent use of buffers. Many tests who subscribe to the events and store the messages or events are affected. To overcome the issue the application can either transfer the ownership of the payload in the application message and then dispose it when it is done, to free the buffers. Or there is an option for long living payload (retained message) to clone the application message with memory from a normal heap. There is also headroom to improve the buffer management in MQTTBufferWriter and to avoid a double copy in the receive codepath. (Its already buffered, but two times copied) Thanks, Martin
@mregen
To overcome the issue the application can either transfer the ownership of the payload in the application message and then dispose it when it is done, to free the buffers.
Do you have sample code for this?
What do you think about having an option for this. I am afraid that managing the buffers is too complicated for most users which just want to receive and send some messages and never complained about memory consumption. My idea is to copy the buffers by default so that users can deal with the messages later or in other places easily. Users who want so to optimize the memory allocations can choose this in the options and then have to copy/release the buffer on their own (including more complexity). What do you think about this idea?
There is also headroom to improve the buffer management in MQTTBufferWriter and to avoid a double copy in the receive codepath. (Its already buffered, but two times copied)
Please let us finish this branch first and continue in new branches so that I can start importing the changes from main (version 4) which happened in the meantime.
Hi @chkr1011, thanks for the feedback..
@mregen
To overcome the issue the application can either transfer the ownership of the payload in the application message and then dispose it when it is done, to free the buffers.
Do you have sample code for this?
I will provide some, also there should be some samples for using a RecyclableMemoryStream. I have been running into some trouble with this implementation myself on the reader side, so I am trying to make it safe for user. Your next statement sounds like a really good idea..
What do you think about having an option for this. I am afraid that managing the buffers is too complicated for most users which just want to receive and send some messages and never complained about memory consumption. My idea is to copy the buffers by default so that users can deal with the messages later or in other places easily. Users who want so to optimize the memory allocations can choose this in the options and then have to copy/release the buffer on their own (including more complexity). What do you think about this idea?
Yes, that is probably a good idea, so by default the apps can just recompile, once you set the option you need to do more. Where would you put such an option?
Thanks, Martin
In my opinion we should introduce a new property in the client and server options with the same naming. For example, "EnableManualPayloadManagement" or similar. Then the summary tag should contain all information required to manage the payload buffers properly.
In my opinion we should introduce a new property in the client and server options with the same naming. For example, "EnableManualPayloadManagement" or similar. Then the summary tag should contain all information required to manage the payload buffers properly.
@chkr1011 Sounds good, I can take a look today. I had some improvements for this branch to catch memory issues, I will post it for the records. But I rather then start again with a clean branch. Important for me is currently more the send path, where the lifetime can be managed externally, but it should be possible to upgrade the MQTT lib later for the receive path without breaking change, that would be good.
Hi @chkr1011, the receive codepath is getting too complicated. I created a minimal version with ReadOnlySequence support in #2046 to get started with the publish support with hopefully better chances to get accepted.