Skip to content

Commit e390c57

Browse files
pglombardopglombardo
authored
Client Stability & Test Suite Improvements (#134)
1 parent 3e360f9 commit e390c57

File tree

13 files changed

+352
-151
lines changed

13 files changed

+352
-151
lines changed

‎.github/workflows/build.yml

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -60,9 +60,9 @@ jobs:
6060
- name: "Dotnet Cake Build"
6161
run: dotnet cake --target=Build
6262
shell: pwsh
63-
# - name: "Dotnet Cake Test"
64-
# run: dotnet cake --target=Test
65-
# shell: pwsh
63+
- name: "Dotnet Cake Test"
64+
run: dotnet cake --target=Test
65+
shell: pwsh
6666
- name: "Dotnet Cake Pack"
6767
run: dotnet cake --target=Pack
6868
shell: pwsh

‎Benchmarks/ClientBenchmarkApp/README.md

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,3 +3,31 @@
33
The benchmarks are built with [BenchmarkDotNet](https://benchmarkdotnet.org) and can be run with:
44

55
`dotnet run ClientBenchmarkApp.csproj -c Release`
6+
7+
# Results - Mar 21, 2024
8+
9+
With release [v0.11.0](https://github.com/hivemq/hivemq-mqtt-client-dotnet/releases/tag/v0.11.0) there was a big performance improvement. All messaging performance was improved but particularly publishing a QoS level 2 message went from ~206ms down to ~1.6ms.
10+
11+
## Previous Performance
12+
13+
| Method | Mean | Error | StdDev | Median |
14+
|------------------------------------------ |-------------:|------------:|------------:|---------------:|
15+
| 'Publish a QoS 0 message' | 390.8 us | 1,842.5 us | 1,218.7 us | 5.646 us |
16+
| 'Publish a QoS 1 message' | 103,722.8 us | 4,330.0 us | 2,864.1 us | 103,536.375 us |
17+
| 'Publish a QoS 2 message' | 202,367.9 us | 26,562.9 us | 17,569.7 us | 206,959.834 us |
18+
19+
## First Pass Refactor Performance
20+
21+
| Method | Mean | Error | StdDev | Median |
22+
|------------------------------------------ |-----------:|-----------:|-----------:|-------------:|
23+
| 'Publish a QoS 0 message' | 401.9 us | 1,876.3 us | 1,241.0 us | 9.250 us |
24+
| 'Publish a QoS 1 message' | 2,140.0 us | 3,568.2 us | 2,360.1 us | 1,324.251 us |
25+
| 'Publish a QoS 2 message' | 4,217.2 us | 5,803.7 us | 3,838.8 us | 2,569.166 us |
26+
27+
## Final Refactor Performance Results (for now 👻)
28+
29+
| Method | Mean | Error | StdDev | Median |
30+
|------------------------------------------ |------------:|----------:|------------:|-------------:|
31+
| 'Publish a QoS 0 message' | 47.11 us | 139.47 us | 411.23 us | 4.875 us |
32+
| 'Publish a QoS 1 message' | 1,210.71 us | 508.64 us | 1,499.75 us | 790.645 us |
33+
| 'Publish a QoS 2 message' | 2,080.46 us | 591.38 us | 1,743.71 us | 1,653.083 us |

‎Source/HiveMQtt/Client/HiveMQClient.cs

Lines changed: 34 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -37,10 +37,12 @@ public partial class HiveMQClient : IDisposable, IHiveMQClient
3737
{
3838
private static readonly NLog.Logger Logger = NLog.LogManager.GetCurrentClassLogger();
3939

40-
private ConnectState connectState = ConnectState.Disconnected;
40+
internal ConnectState ConnectState { get; set; }
4141

4242
public HiveMQClient(HiveMQClientOptions? options = null)
4343
{
44+
this.ConnectState = ConnectState.Disconnected;
45+
4446
options ??= new HiveMQClientOptions();
4547
options.Validate();
4648

@@ -66,12 +68,12 @@ public HiveMQClient(HiveMQClientOptions? options = null)
6668
public List<Subscription> Subscriptions { get; } = new();
6769

6870
/// <inheritdoc />
69-
public bool IsConnected() => this.connectState == ConnectState.Connected;
71+
public bool IsConnected() => this.ConnectState == ConnectState.Connected;
7072

7173
/// <inheritdoc />
7274
public async Task<ConnectResult> ConnectAsync()
7375
{
74-
this.connectState = ConnectState.Connecting;
76+
this.ConnectState = ConnectState.Connecting;
7577

7678
Logger.Info("Connecting to broker at {0}:{1}", this.Options.Host, this.Options.Port);
7779

@@ -89,7 +91,7 @@ public async Task<ConnectResult> ConnectAsync()
8991
// Construct the MQTT Connect packet and queue to send
9092
var connPacket = new ConnectPacket(this.Options);
9193
Logger.Trace($"Queuing packet for send: {connPacket}");
92-
this.sendQueue.Add(connPacket);
94+
this.SendQueue.Add(connPacket);
9395

9496
// FIXME: Cancellation token and better timeout value
9597
ConnAckPacket connAck;
@@ -100,7 +102,7 @@ public async Task<ConnectResult> ConnectAsync()
100102
}
101103
catch (TimeoutException)
102104
{
103-
this.connectState = ConnectState.Disconnected;
105+
this.ConnectState = ConnectState.Disconnected;
104106
throw new HiveMQttClientException("Connect timeout. No response received in time.");
105107
}
106108
finally
@@ -111,11 +113,11 @@ public async Task<ConnectResult> ConnectAsync()
111113

112114
if (connAck.ReasonCode == ConnAckReasonCode.Success)
113115
{
114-
this.connectState = ConnectState.Connected;
116+
this.ConnectState = ConnectState.Connected;
115117
}
116118
else
117119
{
118-
this.connectState = ConnectState.Disconnected;
120+
this.ConnectState = ConnectState.Disconnected;
119121
}
120122

121123
connectResult = new ConnectResult(connAck.ReasonCode, connAck.SessionPresent, connAck.Properties);
@@ -133,7 +135,7 @@ public async Task<ConnectResult> ConnectAsync()
133135
/// <inheritdoc />
134136
public async Task<bool> DisconnectAsync(DisconnectOptions? options = null)
135137
{
136-
if (this.connectState != ConnectState.Connected)
138+
if (this.ConnectState != ConnectState.Connected)
137139
{
138140
Logger.Warn("DisconnectAsync: Client is not connected.");
139141
return false;
@@ -152,15 +154,15 @@ public async Task<bool> DisconnectAsync(DisconnectOptions? options = null)
152154
};
153155

154156
// Once this is set, no more incoming packets or outgoing will be accepted
155-
this.connectState = ConnectState.Disconnecting;
157+
this.ConnectState = ConnectState.Disconnecting;
156158

157159
var taskCompletionSource = new TaskCompletionSource<DisconnectPacket>();
158160
void TaskHandler(object? sender, OnDisconnectSentEventArgs args) => taskCompletionSource.SetResult(args.DisconnectPacket);
159161
EventHandler<OnDisconnectSentEventArgs> eventHandler = TaskHandler;
160162
this.OnDisconnectSent += eventHandler;
161163

162164
Logger.Trace($"Queuing packet for send: {disconnectPacket}");
163-
this.sendQueue.Add(disconnectPacket);
165+
this.SendQueue.Add(disconnectPacket);
164166

165167
try
166168
{
@@ -176,26 +178,35 @@ public async Task<bool> DisconnectAsync(DisconnectOptions? options = null)
176178
this.OnDisconnectSent -= eventHandler;
177179
}
178180

179-
// Close the socket
181+
this.HandleDisconnection();
182+
183+
return true;
184+
}
185+
186+
/// <summary>
187+
/// Close the socket and set the connect state to disconnected.
188+
/// </summary>
189+
private void HandleDisconnection()
190+
{
191+
Logger.Debug("HandleDisconnection: Connection lost. Handling Disconnection.");
192+
180193
this.CloseSocket();
181194

182195
// Fire the corresponding event
183-
this.AfterDisconnectEventLauncher(true);
196+
this.AfterDisconnectEventLauncher(false);
184197

185-
this.connectState = ConnectState.Disconnected;
198+
this.ConnectState = ConnectState.Disconnected;
186199

187200
// FIXME
188-
if (this.sendQueue.Count > 0)
201+
if (this.SendQueue.Count > 0)
189202
{
190-
Logger.Warn("Disconnect: Send queue not empty. Packets pending but we are disconnecting.");
203+
Logger.Warn($"HandleDisconnection: Send queue not empty. {this.SendQueue.Count} packets pending but we are disconnecting (or were disconnected).");
191204
}
192205

193206
// We only clear the send queue on explicit disconnect
194-
while (this.sendQueue.TryTake(out _))
207+
while (this.SendQueue.TryTake(out _))
195208
{
196209
}
197-
198-
return true;
199210
}
200211

201212
/// <inheritdoc />
@@ -210,7 +221,7 @@ public async Task<PublishResult> PublishAsync(MQTT5PublishMessage message)
210221
if (message.QoS == QualityOfService.AtMostOnceDelivery)
211222
{
212223
Logger.Trace($"Queuing packet for send: {publishPacket}");
213-
this.sendQueue.Add(publishPacket);
224+
this.SendQueue.Add(publishPacket);
214225
return new PublishResult(publishPacket.Message);
215226
}
216227
else if (message.QoS == QualityOfService.AtLeastOnceDelivery)
@@ -223,7 +234,7 @@ public async Task<PublishResult> PublishAsync(MQTT5PublishMessage message)
223234

224235
// Construct the MQTT Connect packet and queue to send
225236
Logger.Trace($"Queuing packet for send: {publishPacket}");
226-
this.sendQueue.Add(publishPacket);
237+
this.SendQueue.Add(publishPacket);
227238

228239
var pubAckPacket = await taskCompletionSource.Task.WaitAsync(TimeSpan.FromSeconds(120)).ConfigureAwait(false);
229240

@@ -239,7 +250,7 @@ public async Task<PublishResult> PublishAsync(MQTT5PublishMessage message)
239250
publishPacket.OnPublishQoS2Complete += eventHandler;
240251

241252
// Construct the MQTT Connect packet and queue to send
242-
this.sendQueue.Add(publishPacket);
253+
this.SendQueue.Add(publishPacket);
243254

244255
// Wait on the QoS 2 handshake
245256
var packetList = await taskCompletionSource.Task.WaitAsync(TimeSpan.FromSeconds(120)).ConfigureAwait(false);
@@ -322,7 +333,7 @@ public async Task<SubscribeResult> SubscribeAsync(SubscribeOptions options)
322333
this.OnSubAckReceived += eventHandler;
323334

324335
// Queue the constructed packet to be sent on the wire
325-
this.sendQueue.Add(subscribePacket);
336+
this.SendQueue.Add(subscribePacket);
326337

327338
SubAckPacket subAck;
328339
SubscribeResult subscribeResult;
@@ -426,7 +437,7 @@ public async Task<UnsubscribeResult> UnsubscribeAsync(UnsubscribeOptions unsubOp
426437
EventHandler<OnUnsubAckReceivedEventArgs> eventHandler = TaskHandler;
427438
this.OnUnsubAckReceived += eventHandler;
428439

429-
this.sendQueue.Add(unsubscribePacket);
440+
this.SendQueue.Add(unsubscribePacket);
430441

431442
// FIXME: Cancellation token and better timeout value
432443
UnsubAckPacket unsubAck;

0 commit comments

Comments
 (0)