YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-24 22:53:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add semaphore for accessing send stream

Browse Source
This commit is contained in:
Christian Kratky
2017-09-28 09:29:47 +02:00
parent e947881aca
commit d09455f78e
2 changed files with 32 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MQTTnet.Core/Adapter/MqttChannelCommunicationAdapter.cs
+22 -20
View File
@@ -16,8 +16,7 @@ namespace MQTTnet.Core.Adapter
public class MqttChannelCommunicationAdapter : IMqttCommunicationAdapter public class MqttChannelCommunicationAdapter : IMqttCommunicationAdapter
{ {
private readonly IMqttCommunicationChannel _channel; private readonly IMqttCommunicationChannel _channel;
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
private readonly Task _waitTask = Task.FromResult(0); // this task is used to prevent overlapping write
public MqttChannelCommunicationAdapter(IMqttCommunicationChannel channel, IMqttPacketSerializer serializer) public MqttChannelCommunicationAdapter(IMqttCommunicationChannel channel, IMqttPacketSerializer serializer)
{ {
@@ -77,32 +76,31 @@ namespace MQTTnet.Core.Adapter
public async Task SendPacketsAsync(TimeSpan timeout, CancellationToken cancellationToken, IEnumerable<MqttBasePacket> packets) public async Task SendPacketsAsync(TimeSpan timeout, CancellationToken cancellationToken, IEnumerable<MqttBasePacket> packets)
{ {
await _semaphore.WaitAsync(cancellationToken);
try try
{ {
var waitTask = _waitTask;
foreach (var packet in packets) foreach (var packet in packets)
{ {
if (packet == null) { continue; } if (packet == null)
MqttTrace.Information(nameof(MqttChannelCommunicationAdapter), $"TX >>> {packet} [Timeout={timeout}]");
var writeBuffer = PacketSerializer.Serialize(packet);
waitTask = waitTask.ContinueWith(t => _channel.SendStream.WriteAsync(writeBuffer, 0, writeBuffer.Length, cancellationToken).ConfigureAwait(false), cancellationToken);
}
waitTask = waitTask.ContinueWith(t =>
{
if (timeout > TimeSpan.Zero)
{ {
return _channel.SendStream.FlushAsync(cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false); continue;
} }
return _channel.SendStream.FlushAsync(cancellationToken).ConfigureAwait(false); MqttTrace.Information(nameof(MqttChannelCommunicationAdapter), $"TX >>> {0} [Timeout={1}]", packet, timeout);
}, cancellationToken);
await waitTask; // configure await false generates stackoverflow var writeBuffer = PacketSerializer.Serialize(packet);
await _channel.SendStream.WriteAsync(writeBuffer, 0, writeBuffer.Length, cancellationToken).ConfigureAwait(false);
}
if (timeout > TimeSpan.Zero)
{
await _channel.SendStream.FlushAsync(cancellationToken).TimeoutAfter(timeout).ConfigureAwait(false);
}
else
{
await _channel.SendStream.FlushAsync(cancellationToken).ConfigureAwait(false);
}
} }
catch (TaskCanceledException) catch (TaskCanceledException)
{ {
@@ -120,6 +118,10 @@ namespace MQTTnet.Core.Adapter
{ {
throw new MqttCommunicationException(exception); throw new MqttCommunicationException(exception);
} }
finally
{
_semaphore.Release();
}
} }
public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken) public async Task<MqttBasePacket> ReceivePacketAsync(TimeSpan timeout, CancellationToken cancellationToken)
Tests/MQTTnet.TestApp.NetFramework/PerformanceTest.cs
+10 -10
View File
@@ -113,16 +113,16 @@ namespace MQTTnet.TestApp.NetFramework
.Select(i => CreateMessage()) .Select(i => CreateMessage())
.ToList(); .ToList();
////if (false) if (true)
////{ {
//// //send concurrent (test for raceconditions) //send concurrent (test for raceconditions)
//// var sendTasks = msgs var sendTasks = msgs
//// .Select(msg => PublishSingleMessage(client, msg, ref msgCount)) .Select(msg => PublishSingleMessage(client, msg, ref msgCount))
//// .ToList(); .ToList();
//// await Task.WhenAll(sendTasks); await Task.WhenAll(sendTasks);
////} }
////else else
{ {
await client.PublishAsync(msgs); await client.PublishAsync(msgs);
msgCount += msgs.Count; msgCount += msgs.Count;
@@ -134,7 +134,7 @@ namespace MQTTnet.TestApp.NetFramework
var now = DateTime.Now; var now = DateTime.Now;
if (last < now - TimeSpan.FromSeconds(1)) if (last < now - TimeSpan.FromSeconds(1))
{ {
Console.WriteLine($"sending {msgCount} inteded {msgChunkSize / interval.TotalSeconds}"); Console.WriteLine($"sending {msgCount} intended {msgChunkSize / interval.TotalSeconds}");
msgCount = 0; msgCount = 0;
last = now; last = now;
} }