diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs index a809107f..5804d856 100644 --- a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -211,8 +211,17 @@ namespace MQTTnet.AspNetCore { var buffer = PacketFormatterAdapter.Encode(packet); - WritePacketBuffer(_output, buffer); - await _output.FlushAsync(cancellationToken).ConfigureAwait(false); + if (buffer.Payload.Count == 0) + { + // 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; } diff --git a/Source/MQTTnet.Benchmarks/SendPacketAsyncBenchmark.cs b/Source/MQTTnet.Benchmarks/SendPacketAsyncBenchmark.cs new file mode 100644 index 00000000..aa5cd038 --- /dev/null +++ b/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(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); + } + } +} \ No newline at end of file