From 78c605966883cba51fa594e7cb5500beec0b5897 Mon Sep 17 00:00:00 2001 From: JanEggers Date: Sat, 2 Jun 2018 09:31:27 +0200 Subject: [PATCH] enabled clients based on tcp to benefit from new api and added Benchmark for EndtoEnd with new API --- .../MqttClientConnectionContextFactory.cs | 34 +++++++++ .../Client}/Tcp/BufferExtensions.cs | 2 +- .../Client}/Tcp/DuplexPipe.cs | 2 +- .../Client}/Tcp/SocketAwaitable.cs | 2 +- .../Client}/Tcp/SocketReceiver.cs | 2 +- .../Client}/Tcp/SocketSender.cs | 2 +- .../Client}/Tcp/TcpConnection.cs | 45 +++++++---- .../MqttConnectionContext.cs | 6 ++ .../MQTTnet.AspnetCore/ReaderExtensions.cs | 11 +-- Source/MQTTnet/MqttFactory.cs | 8 ++ .../MQTTnet.Benchmarks.csproj | 2 + ...rocessingMqttConnectionContextBenchmark.cs | 76 +++++++++++++++++++ Tests/MQTTnet.Benchmarks/Program.cs | 4 + Tests/MQTTnet.Benchmarks/TcpPipesBenchmark.cs | 8 +- .../MQTTnet.TestApp.AspNetCore2.csproj | 1 + 15 files changed, 174 insertions(+), 31 deletions(-) create mode 100644 Frameworks/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs rename {Tests/MQTTnet.Benchmarks => Frameworks/MQTTnet.AspnetCore/Client}/Tcp/BufferExtensions.cs (93%) rename {Tests/MQTTnet.Benchmarks => Frameworks/MQTTnet.AspnetCore/Client}/Tcp/DuplexPipe.cs (96%) rename {Tests/MQTTnet.Benchmarks => Frameworks/MQTTnet.AspnetCore/Client}/Tcp/SocketAwaitable.cs (97%) rename {Tests/MQTTnet.Benchmarks => Frameworks/MQTTnet.AspnetCore/Client}/Tcp/SocketReceiver.cs (96%) rename {Tests/MQTTnet.Benchmarks => Frameworks/MQTTnet.AspnetCore/Client}/Tcp/SocketSender.cs (98%) rename {Tests/MQTTnet.Benchmarks => Frameworks/MQTTnet.AspnetCore/Client}/Tcp/TcpConnection.cs (85%) create mode 100644 Tests/MQTTnet.Benchmarks/MessageProcessingMqttConnectionContextBenchmark.cs diff --git a/Frameworks/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs b/Frameworks/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs new file mode 100644 index 00000000..e308ccd4 --- /dev/null +++ b/Frameworks/MQTTnet.AspnetCore/Client/MqttClientConnectionContextFactory.cs @@ -0,0 +1,34 @@ +using System; +using System.Net; +using MQTTnet.Adapter; +using MQTTnet.AspNetCore.Client.Tcp; +using MQTTnet.Client; +using MQTTnet.Diagnostics; +using MQTTnet.Serializer; + +namespace MQTTnet.AspNetCore.Client +{ + public class MqttClientConnectionContextFactory : IMqttClientAdapterFactory + { + public IMqttChannelAdapter CreateClientAdapter(IMqttClientOptions options, IMqttNetChildLogger logger) + { + if (options == null) throw new ArgumentNullException(nameof(options)); + + var serializer = new MqttPacketSerializer { ProtocolVersion = options.ProtocolVersion }; + + switch (options.ChannelOptions) + { + case MqttClientTcpOptions tcpOptions: + { + var endpoint = new DnsEndPoint(tcpOptions.Server, tcpOptions.GetPort()); + var tcpConnection = new TcpConnection(endpoint); + return new MqttConnectionContext(serializer, tcpConnection); + } + default: + { + throw new NotSupportedException(); + } + } + } + } +} diff --git a/Tests/MQTTnet.Benchmarks/Tcp/BufferExtensions.cs b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/BufferExtensions.cs similarity index 93% rename from Tests/MQTTnet.Benchmarks/Tcp/BufferExtensions.cs rename to Frameworks/MQTTnet.AspnetCore/Client/Tcp/BufferExtensions.cs index 879306c1..5911a3a1 100644 --- a/Tests/MQTTnet.Benchmarks/Tcp/BufferExtensions.cs +++ b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/BufferExtensions.cs @@ -1,7 +1,7 @@ using System; using System.Runtime.InteropServices; -namespace MQTTnet.Benchmarks.Tcp +namespace MQTTnet.AspNetCore.Client.Tcp { public static class BufferExtensions { diff --git a/Tests/MQTTnet.Benchmarks/Tcp/DuplexPipe.cs b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/DuplexPipe.cs similarity index 96% rename from Tests/MQTTnet.Benchmarks/Tcp/DuplexPipe.cs rename to Frameworks/MQTTnet.AspnetCore/Client/Tcp/DuplexPipe.cs index f5f3316d..e234da55 100644 --- a/Tests/MQTTnet.Benchmarks/Tcp/DuplexPipe.cs +++ b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/DuplexPipe.cs @@ -1,6 +1,6 @@ using System.IO.Pipelines; -namespace MQTTnet.Benchmarks.Tcp +namespace MQTTnet.AspNetCore.Client.Tcp { public class DuplexPipe : IDuplexPipe { diff --git a/Tests/MQTTnet.Benchmarks/Tcp/SocketAwaitable.cs b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketAwaitable.cs similarity index 97% rename from Tests/MQTTnet.Benchmarks/Tcp/SocketAwaitable.cs rename to Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketAwaitable.cs index 2271bd76..96160d13 100644 --- a/Tests/MQTTnet.Benchmarks/Tcp/SocketAwaitable.cs +++ b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketAwaitable.cs @@ -6,7 +6,7 @@ using System.Runtime.CompilerServices; using System.Threading; using System.Threading.Tasks; -namespace MQTTnet.Benchmarks.Tcp +namespace MQTTnet.AspNetCore.Client.Tcp { public class SocketAwaitable : ICriticalNotifyCompletion { diff --git a/Tests/MQTTnet.Benchmarks/Tcp/SocketReceiver.cs b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketReceiver.cs similarity index 96% rename from Tests/MQTTnet.Benchmarks/Tcp/SocketReceiver.cs rename to Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketReceiver.cs index bc8e5c03..219b7222 100644 --- a/Tests/MQTTnet.Benchmarks/Tcp/SocketReceiver.cs +++ b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketReceiver.cs @@ -2,7 +2,7 @@ using System.IO.Pipelines; using System.Net.Sockets; -namespace MQTTnet.Benchmarks.Tcp +namespace MQTTnet.AspNetCore.Client.Tcp { public class SocketReceiver { diff --git a/Tests/MQTTnet.Benchmarks/Tcp/SocketSender.cs b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs similarity index 98% rename from Tests/MQTTnet.Benchmarks/Tcp/SocketSender.cs rename to Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs index 7cb1bc1d..c8ba8326 100644 --- a/Tests/MQTTnet.Benchmarks/Tcp/SocketSender.cs +++ b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/SocketSender.cs @@ -5,7 +5,7 @@ using System.Diagnostics; using System.IO.Pipelines; using System.Net.Sockets; -namespace MQTTnet.Benchmarks.Tcp +namespace MQTTnet.AspNetCore.Client.Tcp { public class SocketSender { diff --git a/Tests/MQTTnet.Benchmarks/Tcp/TcpConnection.cs b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs similarity index 85% rename from Tests/MQTTnet.Benchmarks/Tcp/TcpConnection.cs rename to Frameworks/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs index 83cb98a8..40ba2959 100644 --- a/Tests/MQTTnet.Benchmarks/Tcp/TcpConnection.cs +++ b/Frameworks/MQTTnet.AspnetCore/Client/Tcp/TcpConnection.cs @@ -4,27 +4,35 @@ using System.IO.Pipelines; using System.Net; using System.Net.Sockets; using System.Threading.Tasks; +using Microsoft.AspNetCore.Connections; +using Microsoft.AspNetCore.Http.Features; using MQTTnet.Exceptions; -namespace MQTTnet.Benchmarks.Tcp +namespace MQTTnet.AspNetCore.Client.Tcp { - public class TcpConnection + public class TcpConnection : ConnectionContext { - private readonly Socket _socket; private volatile bool _aborted; private readonly EndPoint _endPoint; + private SocketSender _sender; + private SocketReceiver _receiver; + + private Socket _socket; private IDuplexPipe _application; - private IDuplexPipe _transport; - private readonly SocketSender _sender; - private readonly SocketReceiver _receiver; + + + public bool IsConnected { get; private set; } + + public override string ConnectionId { get; set; } + + public override IFeatureCollection Features { get; } + + public override IDictionary Items { get; set; } + public override IDuplexPipe Transport { get; set; } public TcpConnection(EndPoint endPoint) { - _socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); _endPoint = endPoint; - - _sender = new SocketSender(_socket, PipeScheduler.ThreadPool); - _receiver = new SocketReceiver(_socket, PipeScheduler.ThreadPool); } public TcpConnection(Socket socket) @@ -38,29 +46,34 @@ namespace MQTTnet.Benchmarks.Tcp public Task DisposeAsync() { - _transport?.Output.Complete(); - _transport?.Input.Complete(); + IsConnected = false; + + Transport?.Output.Complete(); + Transport?.Input.Complete(); _socket?.Dispose(); return Task.CompletedTask; } - public async Task StartAsync() + public async Task StartAsync() { - if (!_socket.Connected) + if (_socket == null) { + _socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); + _sender = new SocketSender(_socket, PipeScheduler.ThreadPool); + _receiver = new SocketReceiver(_socket, PipeScheduler.ThreadPool); await _socket.ConnectAsync(_endPoint); } var pair = DuplexPipe.CreateConnectionPair(PipeOptions.Default, PipeOptions.Default); - _transport = pair.Transport; + Transport = pair.Transport; _application = pair.Application; _ = ExecuteAsync(); - return pair.Transport; + IsConnected = true; } private async Task ExecuteAsync() diff --git a/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs index 88bfcc91..f7acfa4d 100644 --- a/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs +++ b/Frameworks/MQTTnet.AspnetCore/MqttConnectionContext.cs @@ -1,5 +1,6 @@ using Microsoft.AspNetCore.Connections; using MQTTnet.Adapter; +using MQTTnet.AspNetCore.Client.Tcp; using MQTTnet.Packets; using MQTTnet.Serializer; using System; @@ -27,6 +28,10 @@ namespace MQTTnet.AspNetCore public Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken) { + if (Connection is TcpConnection tcp && !tcp.IsConnected) + { + return tcp.StartAsync(); + } return Task.CompletedTask; } @@ -34,6 +39,7 @@ namespace MQTTnet.AspNetCore { Connection.Transport.Input.Complete(); Connection.Transport.Output.Complete(); + return Task.CompletedTask; } diff --git a/Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs b/Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs index 5b832539..04117103 100644 --- a/Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs +++ b/Frameworks/MQTTnet.AspnetCore/ReaderExtensions.cs @@ -76,13 +76,10 @@ namespace MQTTnet.AspNetCore } var bodySlice = copy.Slice(0, bodyLength); - using (var body = new MemoryStream(bodySlice.GetArray())) - { - packet = serializer.Deserialize(new ReceivedMqttPacket(fixedheader, body)); - consumed = bodySlice.End; - observed = bodySlice.End; - return true; - } + packet = serializer.Deserialize(new ReceivedMqttPacket(fixedheader, new MqttPacketBodyReader(bodySlice.GetArray()))); + consumed = bodySlice.End; + observed = bodySlice.End; + return true; } } } diff --git a/Source/MQTTnet/MqttFactory.cs b/Source/MQTTnet/MqttFactory.cs index 00438fa9..e830ad20 100644 --- a/Source/MQTTnet/MqttFactory.cs +++ b/Source/MQTTnet/MqttFactory.cs @@ -22,6 +22,14 @@ namespace MQTTnet return new MqttClient(new MqttClientAdapterFactory(), logger); } + public IMqttClient CreateMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory mqttClientAdapterFactory) + { + if (logger == null) throw new ArgumentNullException(nameof(logger)); + if (mqttClientAdapterFactory == null) throw new ArgumentNullException(nameof(mqttClientAdapterFactory)); + + return new MqttClient(mqttClientAdapterFactory, logger); + } + public IMqttServer CreateMqttServer() { var logger = new MqttNetLogger(); diff --git a/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj b/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj index 5e612fb9..2a6b91a5 100644 --- a/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj +++ b/Tests/MQTTnet.Benchmarks/MQTTnet.Benchmarks.csproj @@ -10,9 +10,11 @@ + + diff --git a/Tests/MQTTnet.Benchmarks/MessageProcessingMqttConnectionContextBenchmark.cs b/Tests/MQTTnet.Benchmarks/MessageProcessingMqttConnectionContextBenchmark.cs new file mode 100644 index 00000000..1e2fbb2b --- /dev/null +++ b/Tests/MQTTnet.Benchmarks/MessageProcessingMqttConnectionContextBenchmark.cs @@ -0,0 +1,76 @@ +using BenchmarkDotNet.Attributes; +using MQTTnet.Client; + + +using MQTTnet.AspNetCore; + +using Microsoft.AspNetCore; +using Microsoft.AspNetCore.Hosting; +using MQTTnet.Server; +using MQTTnet.Diagnostics; +using MQTTnet.AspNetCore.Client; + +namespace MQTTnet.Benchmarks +{ + [MemoryDiagnoser] + public class MessageProcessingMqttConnectionContextBenchmark + { + private IWebHost _host; + private IMqttClient _mqttClient; + private MqttApplicationMessage _message; + + [GlobalSetup] + public void Setup() + { + _host = WebHost.CreateDefaultBuilder() + .UseKestrel(o => o.ListenAnyIP(1883, l => l.UseMqtt())) + .ConfigureServices(services => { + var mqttServerOptions = new MqttServerOptionsBuilder() + .WithoutDefaultEndpoint() + .Build(); + services + .AddHostedMqttServer(mqttServerOptions) + .AddMqttConnectionHandler(); + }) + .Configure(app => { + app.UseMqttServer(s => { + + }); + }) + .Build(); + + var factory = new MqttFactory(); + _mqttClient = factory.CreateMqttClient(new MqttNetLogger(), new MqttClientConnectionContextFactory()); + + _host.StartAsync().GetAwaiter().GetResult(); + + var clientOptions = new MqttClientOptionsBuilder() + .WithTcpServer("localhost").Build(); + + _mqttClient.ConnectAsync(clientOptions).GetAwaiter().GetResult(); + + _message = new MqttApplicationMessageBuilder() + .WithTopic("A") + .Build(); + } + + [GlobalCleanup] + public void Cleanup() + { + _mqttClient.DisconnectAsync().GetAwaiter().GetResult(); + _mqttClient.Dispose(); + + _host.StopAsync().GetAwaiter().GetResult(); + _host.Dispose(); + } + + [Benchmark] + public void Send_10000_Messages() + { + for (var i = 0; i < 10000; i++) + { + _mqttClient.PublishAsync(_message).GetAwaiter().GetResult(); + } + } + } +} diff --git a/Tests/MQTTnet.Benchmarks/Program.cs b/Tests/MQTTnet.Benchmarks/Program.cs index 53069163..9407ccfe 100644 --- a/Tests/MQTTnet.Benchmarks/Program.cs +++ b/Tests/MQTTnet.Benchmarks/Program.cs @@ -16,6 +16,7 @@ namespace MQTTnet.Benchmarks Console.WriteLine("5 = ChannelAdapterBenchmark"); Console.WriteLine("6 = MqttTcpChannelBenchmark"); Console.WriteLine("7 = TcpPipesBenchmark"); + Console.WriteLine("8 = MessageProcessingMqttConnectionContextBenchmark"); var pressedKey = Console.ReadKey(true); switch (pressedKey.KeyChar) @@ -41,6 +42,9 @@ namespace MQTTnet.Benchmarks case '7': BenchmarkRunner.Run(); break; + case '8': + BenchmarkRunner.Run(new AllowNonOptimized()); + break; } Console.ReadLine(); diff --git a/Tests/MQTTnet.Benchmarks/TcpPipesBenchmark.cs b/Tests/MQTTnet.Benchmarks/TcpPipesBenchmark.cs index cb3ffbaf..4252ab36 100644 --- a/Tests/MQTTnet.Benchmarks/TcpPipesBenchmark.cs +++ b/Tests/MQTTnet.Benchmarks/TcpPipesBenchmark.cs @@ -4,7 +4,7 @@ using System.Net.Sockets; using System.Threading; using System.Threading.Tasks; using BenchmarkDotNet.Attributes; -using MQTTnet.Benchmarks.Tcp; +using MQTTnet.AspNetCore.Client.Tcp; namespace MQTTnet.Benchmarks { @@ -25,10 +25,12 @@ namespace MQTTnet.Benchmarks var clientConnection = new TcpConnection(new IPEndPoint(IPAddress.Loopback, 1883)); - _client = clientConnection.StartAsync().GetAwaiter().GetResult(); + clientConnection.StartAsync().GetAwaiter().GetResult(); + _client = clientConnection.Transport; var serverConnection = new TcpConnection(task.GetAwaiter().GetResult()); - _server = serverConnection.StartAsync().GetAwaiter().GetResult(); + serverConnection.StartAsync().GetAwaiter().GetResult(); + _server = serverConnection.Transport; } diff --git a/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj b/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj index f6ead870..a216d728 100644 --- a/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj +++ b/Tests/MQTTnet.TestApp.AspNetCore2/MQTTnet.TestApp.AspNetCore2.csproj @@ -10,6 +10,7 @@ +