Mqttnet內存與性能改進錄

1 MQTTnet介紹

MQTTnet是一個高性能的 .NET MQTT庫，它提供MQTT客戶端和MQTT服務器的功能，支持到最新MQTT5協議版本，支持.Net Framework4.5.2版本或以上。

MQTTnet is a high performance .NET library for MQTT based communication. It provides a MQTT client and a MQTT server (broker) and supports the MQTT protocol up to version 5. It is compatible with mostly any supported .NET Framework version and CPU architecture.

2 我與MQTTnet

我有一些小型項目，需要安裝在局域網環境下的windows或linux系統，這個安裝過程需要小白也能安裝，而且每天都有可能有多份新的安裝部署的新環境，所以流行的mqtt服務器emqx可能變得不太適合我的選型，因為讓小白來大量部署它不是非常方便。

我的這個小項目主體是一個Web項目，瀏覽器用戶對象是管理員，數據的產生者是N多個廉價linux小型設備，設備使用mqtt協議高頻提交數據到后臺，后臺也需要使用mqtt協議來主動控制設備完成一些操作動作。除此之后，Web瀏覽器也需要使用mqtt over websocket來訂閱一些主題，達到監控某臺設備的實時數據目的。

經過比較，MQTTnet變成了我意向使用的mqtt庫，尤其是MQTTnet.AspNetCore子項目，基于kestrel來使用tcp或websocket做傳輸層，增加mqtt應用層協議的解析，最后讓mqtt與asp.netcore完美地融合在一起。

3 Bug發現

項目有后臺主動發送mqtt到設備以控制設備的需求，在mqttnet里有個對應的InjectApplicationMessage()擴展方法可以從server主動發送mqtt到client，但這個方法總是拋出ArgumentNullException。但如果使用InjectApplicationMessage (InjectedMqttApplicationMessage)這個基礎方法來注入mqtt消息不有異常。

經過一段時間后，閑時的我決定遷出mqttnet項目的源代碼來調試分析。最后發現是因為這個擴展方法沒有傳遞SenderClientId導致的異常，所以我決定嘗試修改并推送一個請求到mqttnet項目。

4 改進之路

經過嘗試修改一個小小bug之后，我開始認真的閱讀MQTTnet.AspNetCore的源代碼，陸續發現一些可以減少內存復制和內存分配的優化點：

ReadOnlyMemory<byte>轉為ReceivedMqttPacket過程優化；
MqttPacketBuffer發送過程的優化；
Array.Copy()的改進;
Byte[]?->?ArraySegment<byte>的優化；

4.1 避免不必要的`ReadOnlyMemory<byte>`轉為`byte[]`

原始代碼

var bodySlice = copy.Slice(0, bodyLength);
var buffer = bodySlice.GetMemory().ToArray();
var receivedMqttPacket = new ReceivedMqttPacket(fixedHeader, new ArraySegment<byte>(buffer, 0, buffer.Length), buffer.Length + 2);static ReadOnlyMemory<byte> GetMemory(this in ReadOnlySequence<byte> input)
{if (input.IsSingleSegment){return input.First;}// Should be rarereturn input.ToArray();
}

原始代碼設計了一個GetMemory()方法，目的是在兩個地方調用到。但它的一句var buffer = bodySlice.GetMemory().ToArray()，就會無條件的產生一次內存分配和一次內存拷貝。

改進代碼

var bodySlice = copy.Slice(0, bodyLength);
var bodySegment = GetArraySegment(ref bodySlice); 
var receivedMqttPacket = new ReceivedMqttPacket(fixedHeader, bodySegment, headerLength + bodyLength);static ArraySegment<byte> GetArraySegment(ref ReadOnlySequence<byte> input)
{if (input.IsSingleSegment && MemoryMarshal.TryGetArray(input.First, out var segment)){return segment;}// Should be rarevar array = input.ToArray();return new ArraySegment<byte>(array);
}

因為有其它地方的優化，GetMemory()不再需要復用，所以我們直接改為GetArraySegment()，里面使用MemoryMarshal.TryGetArray()方法嘗試從ReadOnlyMemory<byte>獲取ArraySegment<byte>對象。而mqttnet的ReceivedMqttPacket對象是支持ArraySegment<byte>類型參數的。

在我提交請求之后，@gfoidl給了很多其它特別好的性能方面的建議，有興趣的同學可以點此查看。

戲劇性的是，在我嘗試改進這個問題的時候，我發現了mqttnet的另外一個BUG：當bodySegment的Offset不是0開始的時候，mqttnet會產生異常。這足以說明，mqttnet項目從未使用Offset大于0的ArraySegment<byte>，所以這個bug才一直沒有發現。本為不是MQTTnet.AspNetCore子項目的代碼我就不改的原則，我向mqttnet提了問題：https://github.com/dotnet/MQTTnet/issues/1592 作者也很認真看待這個問題，于是自己加班解決：https://github.com/dotnet/MQTTnet/pull/1593

更戲劇性的是，我開心地合并main代碼過來驗證之后，發現作者改的BUG里又帶入了BUG！現在Offset大于0還是有問題。于是我心急啊，我決定為這個BUG中BUG提交一個修改的請求：https://github.com/dotnet/MQTTnet/pull/1598

最后，這個MemoryMarshal.TryGetArray()的優化終于提到合并，改進后CPU時間時間也減少了，內存分配更是減少了50%。

4.2?`MqttPacketBuffer`發送過程的優化

MqttPacketBuffer有兩個數據段：Pacaket段和Payload段，我看到它原始發送代碼如下：

var buffer = formatter.Encode(packet);
var msg = buffer.Join().AsMemory();
var output = _output;
var result = await output.WriteAsync(msg, cancellationToken).ConfigureAwait(false);

我也沒有經過認證思考，覺得這里可以將Pacaket段和Payload直接兩次發送即可。

var buffer = PacketFormatterAdapter.Encode(packet);
await _output.WriteAsync(buffer.Packet, cancellationToken).ConfigureAwait(false);if (buffer.Payload.Count > 0)
{ await _output.WriteAsync(buffer.Payload, cancellationToken).ConfigureAwait(false);
}

后來作者說，當mqtt over websocket時，有些客戶端在實現上沒能兼容一個mqtt包分多個websocket幀傳輸的處理，所以需要合并發送。那我就想，如果我檢測傳輸層是websocket的話再Join合并就行了，于是改為如下：

if (_isOverWebSocket == false)
{await _output.WriteAsync(buffer.Packet, cancellationToken).ConfigureAwait(false);if (buffer.Payload.Count > 0){await _output.WriteAsync(buffer.Payload, cancellationToken).ConfigureAwait(false);}
}
else
{     var bufferSegment = buffer.Join();await _output.WriteAsync(bufferSegment, cancellationToken).ConfigureAwait(false);
}

雖然覺得這個方案比之前要好了一些，但感覺Jion里的?new byte[]的分配讓我耿耿于懷。再經過幾將進改，最后的代碼如下，雖然也有拷貝，但至少已經沒有分配：

if (buffer.Payload.Count == 0)
{// zero copy// https://github.com/dotnet/runtime/blob/main/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/StreamPipeWriter.cs#L279await _output.WriteAsync(buffer.Packet, cancellationToken).ConfigureAwait(false);
}
else
{WritePacketBuffer(_output, buffer);await _output.FlushAsync(cancellationToken).ConfigureAwait(false);
}static void WritePacketBuffer(PipeWriter output, MqttPacketBuffer buffer)
{// copy MqttPacketBuffer's Packet and Payload to the same buffer block of PipeWriter// MqttPacket will be transmitted within the bounds of a WebSocket frame after PipeWriter.FlushAsyncvar span = output.GetSpan(buffer.Length);buffer.Packet.AsSpan().CopyTo(span);buffer.Payload.AsSpan().CopyTo(span.Slice(buffer.Packet.Count));output.Advance(buffer.Length);
}

4.3?`Array.Copy()`的改進

mqttnet由于要兼容很多.net框架和版本，所以往往能使用的api不多，比如在內存拷貝了，還保留了最初的Array.Copy()，我們可以較新的框架下使用更好的api來復制，最高可達25%的復制性能提升，這個改進的工作量非常小，但產出是相當的可喜啊。

[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static void Copy(byte[] source, int sourceIndex, byte[] destination, int destinationIndex, int length)
{
#if NETCOREAPP3_1_OR_GREATER || NETSTANDARD2_1source.AsSpan(sourceIndex, length).CopyTo(destination.AsSpan(destinationIndex, length));
#elif NET461_OR_GREATER || NETSTANDARD1_3_OR_GREATERunsafe{fixed (byte* pSoure = &source[sourceIndex]){fixed (byte* pDestination = &destination[destinationIndex]){System.Buffer.MemoryCopy(pSoure, pDestination, length, length);}}}
#elseArray.Copy(source, sourceIndex, destination, destinationIndex, length);
#endif
}

4.4?`Byte[]`?->?`ArraySegment<byte>`的優化

當前的mqttnet，由于歷史設計的局限原因，現在還不能創建ArraySegment<byte>或Memory<byte>作為payload的mqtt消息包。如果我們從ArrayPool申請1000字節的buffer，實際我們會得到一個到1024字節的buffer，想拿租賃的buffer的前1000字節做mqtt消息的payload，我們現在不得不再創建一個1000字節的byte[1000]?newpayload，然后拷貝buffer到newpayload。

這種局限對服務端來說弊端是很大的，我現在嘗試如何不破壞原始的byte[]支持的設計提前下，讓mqttnet也支持ArraySegment<byte>的數據發送。當然，保持兼容性的新Api加入對項目來說是一種大的變化，自然有一定的風險性。

如果你也關注這個mqttnet項目，你可以查看 https://github.com/dotnet/MQTTnet/pull/1585 這個提議，也許未來它會變成現實。