From b98417ab391cb68f8d815fb11087885e513e2f3f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?G=C3=BCnther=20Foidl?= Date: Thu, 19 Jan 2023 20:08:51 +0100 Subject: [PATCH] Avoid some full memory barriers (#1590) * Avoid some full memory barriers * Reset in stats itself * Updated MqttClientStatistics * Fixed order of members as artifact from merge-conflict --- Source/MQTTnet/Adapter/MqttChannelAdapter.cs | 29 +++++++---- Source/MQTTnet/Internal/AsyncQueue.cs | 14 +++++ Source/MQTTnet/MQTTnet.csproj | 2 +- .../Server/Internal/MqttClientStatistics.cs | 51 +++++++++++-------- 4 files changed, 63 insertions(+), 33 deletions(-) diff --git a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs index 8e08dd0f..47e28b31 100644 --- a/Source/MQTTnet/Adapter/MqttChannelAdapter.cs +++ b/Source/MQTTnet/Adapter/MqttChannelAdapter.cs @@ -33,8 +33,7 @@ namespace MQTTnet.Adapter readonly AsyncLock _syncRoot = new AsyncLock(); - long _bytesReceived; - long _bytesSent; + Statistics _statistics; // mutable struct, don't make readonly! public MqttChannelAdapter(IMqttChannel channel, MqttPacketFormatterAdapter packetFormatterAdapter, MqttPacketInspector packetInspector, IMqttNetLogger logger) { @@ -51,9 +50,9 @@ namespace MQTTnet.Adapter _logger = logger.WithSource(nameof(MqttChannelAdapter)); } - public long BytesReceived => Interlocked.Read(ref _bytesReceived); + public long BytesReceived => Volatile.Read(ref _statistics._bytesReceived); - public long BytesSent => Interlocked.Read(ref _bytesSent); + public long BytesSent => Volatile.Read(ref _statistics._bytesSent); public X509Certificate2 ClientCertificate => _channel.ClientCertificate; @@ -147,7 +146,7 @@ namespace MQTTnet.Adapter _packetInspector?.EndReceivePacket(); - Interlocked.Add(ref _bytesSent, receivedPacket.TotalLength); + Interlocked.Add(ref _statistics._bytesSent, receivedPacket.TotalLength); if (PacketFormatterAdapter.ProtocolVersion == MqttProtocolVersion.Unknown) { @@ -181,11 +180,7 @@ namespace MQTTnet.Adapter return null; } - public void ResetStatistics() - { - Interlocked.Exchange(ref _bytesReceived, 0L); - Interlocked.Exchange(ref _bytesSent, 0L); - } + public void ResetStatistics() => _statistics.Reset(); public async Task SendPacketAsync(MqttPacket packet, CancellationToken cancellationToken) { @@ -216,7 +211,7 @@ namespace MQTTnet.Adapter await _channel.WriteAsync(packetBuffer.Packet, true, cancellationToken).ConfigureAwait(false); } - Interlocked.Add(ref _bytesReceived, packetBuffer.Length); + Interlocked.Add(ref _statistics._bytesReceived, packetBuffer.Length); } catch (Exception exception) { @@ -459,5 +454,17 @@ namespace MQTTnet.Adapter throw new MqttCommunicationException(exception); } + + private struct Statistics + { + public long _bytesReceived; + public long _bytesSent; + + public void Reset() + { + Volatile.Write(ref _bytesReceived, 0); + Volatile.Write(ref _bytesSent, 0); + } + } } } \ No newline at end of file diff --git a/Source/MQTTnet/Internal/AsyncQueue.cs b/Source/MQTTnet/Internal/AsyncQueue.cs index 52381a32..22e12b38 100644 --- a/Source/MQTTnet/Internal/AsyncQueue.cs +++ b/Source/MQTTnet/Internal/AsyncQueue.cs @@ -22,7 +22,11 @@ namespace MQTTnet.Internal public void Clear() { +#if NETCOREAPP3_1_OR_GREATER + _queue.Clear(); +#else Interlocked.Exchange(ref _queue, new ConcurrentQueue()); +#endif } public void Dispose() @@ -32,6 +36,16 @@ namespace MQTTnet.Internal _signal.Dispose(); _isDisposed = true; + +#if !NETSTANDARD1_3 + if (typeof(IDisposable).IsAssignableFrom(typeof(TItem))) + { + while (_queue.TryDequeue(out TItem item)) + { + (item as IDisposable).Dispose(); + } + } +#endif } } diff --git a/Source/MQTTnet/MQTTnet.csproj b/Source/MQTTnet/MQTTnet.csproj index 12044cfe..6cbeb03d 100644 --- a/Source/MQTTnet/MQTTnet.csproj +++ b/Source/MQTTnet/MQTTnet.csproj @@ -89,4 +89,4 @@ - + \ No newline at end of file diff --git a/Source/MQTTnet/Server/Internal/MqttClientStatistics.cs b/Source/MQTTnet/Server/Internal/MqttClientStatistics.cs index 28f8e9cf..8fd96426 100644 --- a/Source/MQTTnet/Server/Internal/MqttClientStatistics.cs +++ b/Source/MQTTnet/Server/Internal/MqttClientStatistics.cs @@ -10,14 +10,14 @@ namespace MQTTnet.Server { public sealed class MqttClientStatistics { - long _receivedApplicationMessagesCount; + Statistics _statistics = new Statistics // mutable struct, don't make readonly! + { + // Start with 1 because the CONNACK packet is not counted here. + _receivedPacketsCount = 1, - // Start with 1 because the CONNACK packet is not counted here. - long _receivedPacketsCount = 1; - long _sentApplicationMessagesCount; - - // Start with 1 because the CONNECT packet is not counted here. - long _sentPacketsCount = 1; + // Start with 1 because the CONNECT packet is not counted here. + _sentPacketsCount = 1 + }; public MqttClientStatistics() { @@ -30,7 +30,6 @@ namespace MQTTnet.Server } public DateTime ConnectedTimestamp { get; } - public DateTime LastNonKeepAlivePacketReceivedTimestamp { get; private set; } /// @@ -43,13 +42,13 @@ namespace MQTTnet.Server /// public DateTime LastPacketSentTimestamp { get; private set; } - public long ReceivedApplicationMessagesCount => Interlocked.Read(ref _receivedApplicationMessagesCount); + public long SentApplicationMessagesCount => Volatile.Read(ref _statistics._sentApplicationMessagesCount); - public long ReceivedPacketsCount => Interlocked.Read(ref _receivedPacketsCount); + public long ReceivedApplicationMessagesCount => Volatile.Read(ref _statistics._receivedApplicationMessagesCount); - public long SentApplicationMessagesCount => Interlocked.Read(ref _sentApplicationMessagesCount); + public long SentPacketsCount => Volatile.Read(ref _statistics._sentPacketsCount); - public long SentPacketsCount => Interlocked.Read(ref _sentPacketsCount); + public long ReceivedPacketsCount => Volatile.Read(ref _statistics._receivedPacketsCount); public void HandleReceivedPacket(MqttPacket packet) { @@ -61,11 +60,11 @@ namespace MQTTnet.Server // This class is tracking all values from Clients perspective! LastPacketSentTimestamp = DateTime.UtcNow; - Interlocked.Increment(ref _sentPacketsCount); + Interlocked.Increment(ref _statistics._sentPacketsCount); if (packet is MqttPublishPacket) { - Interlocked.Increment(ref _sentApplicationMessagesCount); + Interlocked.Increment(ref _statistics._sentApplicationMessagesCount); } if (!(packet is MqttPingReqPacket || packet is MqttPingRespPacket)) @@ -73,6 +72,7 @@ namespace MQTTnet.Server LastNonKeepAlivePacketReceivedTimestamp = LastPacketReceivedTimestamp; } } + public void ResetStatistics() => _statistics.Reset(); public void HandleSentPacket(MqttPacket packet) { @@ -84,20 +84,29 @@ namespace MQTTnet.Server // This class is tracking all values from Clients perspective! LastPacketReceivedTimestamp = DateTime.UtcNow; - Interlocked.Increment(ref _receivedPacketsCount); + Interlocked.Increment(ref _statistics._receivedPacketsCount); if (packet is MqttPublishPacket) { - Interlocked.Increment(ref _receivedApplicationMessagesCount); + Interlocked.Increment(ref _statistics._receivedApplicationMessagesCount); } } - public void ResetStatistics() + private struct Statistics { - Interlocked.Exchange(ref _sentApplicationMessagesCount, 0); - Interlocked.Exchange(ref _receivedApplicationMessagesCount, 0); - Interlocked.Exchange(ref _sentPacketsCount, 0); - Interlocked.Exchange(ref _receivedPacketsCount, 0); + public long _receivedPacketsCount; + public long _sentPacketsCount; + + public long _receivedApplicationMessagesCount; + public long _sentApplicationMessagesCount; + + public void Reset() + { + Volatile.Write(ref _receivedPacketsCount, 0); + Volatile.Write(ref _sentPacketsCount, 0); + Volatile.Write(ref _receivedApplicationMessagesCount, 0); + Volatile.Write(ref _sentApplicationMessagesCount, 0); + } } } } \ No newline at end of file