Performance optimization of SendPacketAsync (#1624)

* performance optimization of SendPacketAsync

* Update Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

Co-authored-by: Günther Foidl <gue@korporal.at>

Co-authored-by: Günther Foidl <gue@korporal.at>

Source/MQTTnet.AspnetCore/MqttConnectionContext.cs

@@ -211,8 +211,17 @@ namespace MQTTnet.AspNetCore
                 {
                     var buffer = PacketFormatterAdapter.Encode(packet);
 
-                    WritePacketBuffer(_output, buffer);
+                    if (buffer.Payload.Count == 0)
-                    await _output.FlushAsync(cancellationToken).ConfigureAwait(false);
+                    {
+                        // zero copy
+                        // https://github.com/dotnet/runtime/blob/e31ddfdc4f574b26231233dc10c9a9c402f40590/src/libraries/System.IO.Pipelines/src/System/IO/Pipelines/StreamPipeWriter.cs#L279
+                        await _output.WriteAsync(buffer.Packet, cancellationToken).ConfigureAwait(false);
+                    }
+                    else
+                    {
+                        WritePacketBuffer(_output, buffer);
+                        await _output.FlushAsync(cancellationToken).ConfigureAwait(false);
+                    }
 
                     BytesSent += buffer.Length;
                 }

Source/MQTTnet.Benchmarks/SendPacketAsyncBenchmark.cs

@@ -0,0 +1,68 @@
+using BenchmarkDotNet.Attributes;
+using BenchmarkDotNet.Jobs;
+using MQTTnet.Formatter;
+using System;
+using System.IO;
+using System.IO.Pipelines;
+using System.Threading.Tasks;
+
+namespace MQTTnet.Benchmarks
+{
+    [SimpleJob(RuntimeMoniker.Net60)]
+    [RPlotExporter, RankColumn]
+    [MemoryDiagnoser]
+    public class SendPacketAsyncBenchmark : BaseBenchmark
+    {
+        MemoryStream stream;
+        MqttPacketBuffer buffer;
+
+        [GlobalSetup]
+        public void GlobalSetup()
+        {
+            stream = new MemoryStream(1024);
+            var packet = new ArraySegment<byte>(new byte[10]);
+            buffer = new MqttPacketBuffer(packet);
+        }
+
+        [Benchmark(Baseline = true)]
+        public async ValueTask Before()
+        {
+            stream.Position = 0;
+            var output = PipeWriter.Create(stream);
+
+            WritePacketBuffer(output, buffer);
+            await output.FlushAsync();
+        }
+
+        [Benchmark]
+        public async ValueTask After()
+        {
+            stream.Position = 0;
+            var output = PipeWriter.Create(stream);
+
+            if (buffer.Payload.Count == 0)
+            {
+                await output.WriteAsync(buffer.Packet).ConfigureAwait(false);
+            }
+            else
+            {
+                WritePacketBuffer(output, buffer);
+                await output.FlushAsync().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.FlushAsync
+
+            var span = output.GetSpan(buffer.Length);
+
+            buffer.Packet.AsSpan().CopyTo(span);
+            buffer.Payload.AsSpan().CopyTo(span.Slice(buffer.Packet.Count));
+
+            output.Advance(buffer.Length);
+        }
+    }
+}