dotnet / dotnet/MQTTnet

About the performance of the RetainedMessages

Open
#2,091 0 comments 0 reactions 0 assignees View on GitHub
feature-request
Dominant language
C#
Stars
5.1k
Forks
1.2k
PR merge metrics
No merged PRs in 30d

Description

### Describe the feature request
I am currently using MqttServer.InjectApplicationMessage to push RetainedMessages, but when the number of Topics reaches tens of thousands, the performance of Retained will drop significantly, with tens of thousands push taking several minutes. The more data there is, the lower the performance.

### Which project is your feature request related to?
- Server

### Describe the solution you'd like
I found the reason for the performance degradation in the source code of MqttRetainedMessagesManager. For each push of RetainedMessages, all RetainedMessages are converted from Dictionary to List, which is used as a parameter for RetainedMessageChangedEvent. This consumes a lot of performance. Even if I don't subscribe to RetainedMessageChangedEvent, the conversion process will still run. I tried modifying the code in this section and disabled the conversion. After doing so, the time it took to push tens of thousands of messages reduced from several minutes to about 100ms. I suggest disabling this conversion when not subscribing to RetainedMessageChangedEvent to improve the performance of RetainedMessages.

### Describe alternatives you've considered

### Additional context
Example
```
public async Task UpdateMessage(string clientId, MqttApplicationMessage applicationMessage)
{
if (applicationMessage == null)
{
throw new ArgumentNullException(nameof(applicationMessage));
}
try
{
List messagesForSave = null;
var saveIsRequired = false;
//add hasHandlers;
var hasHandlers = eventContainer.RetainedMessageChangedEvent.HasHandlers;

lock (_messages)
{
var payload = applicationMessage.Payload;
var hasPayload = payload.Length > 0;

if (!hasPayload)
{
saveIsRequired = _messages.Remove(applicationMessage.Topic);
_logger.Verbose("Client '{0}' cleared retained message for topic '{1}'.", clientId, applicationMessage.Topic);
}
else
{
if (!_messages.TryGetValue(applicationMessage.Topic, out var existingMessage))
{
_messages[applicationMessage.Topic] = applicationMessage;
saveIsRequired = true;
}
else
{
if (existingMessage.QualityOfServiceLevel != applicationMessage.QualityOfServiceLevel ||
!MqttMemoryHelper.SequenceEqual(existingMessage.Payload, payload))
{
_messages[applicationMessage.Topic] = applicationMessage;
saveIsRequired = true;
}
}

_logger.Verbose("Client '{0}' set retained message for topic '{1}'.", clientId, applicationMessage.Topic);
}

//add " && hasHandlers"
if (saveIsRequired && hasHandlers)
{
messagesForSave = new List(_messages.Values);
}
}

//add " && hasHandlers"
if (saveIsRequired && hasHandlers)
{
using (await _storageAccessLock.EnterAsync().ConfigureAwait(false))
{
var eventArgs = new RetainedMessageChangedEventArgs(clientId, applicationMessage, messagesForSave);
await _eventContainer.RetainedMessageChangedEvent.InvokeAsync(eventArgs).ConfigureAwait(false);
}
}
}
catch (Exception exception)
{
_logger.Error(exception, "Unhandled exception while handling retained messages.");
}
}
```

Contributor guide

No contributing guide indexed for this repository

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.