commit cbc87b28ece603083065b5cb2b481f5d0d0f4eba
parent 90ae2ac2a6ff6ce6e8d30aee854ad32b39f0dad0
Author: Mikolaj Lenczewski <33129490+EnderRifter@users.noreply.github.com>
Date: Sun, 16 Feb 2020 15:48:29 +0000
Fixed huge memory issue! Yay! Udp is now somewhat more useable.
Diffstat:
5 files changed, 70 insertions(+), 25 deletions(-)
diff --git a/NetSharp/NetSharp/NetSharp.csproj b/NetSharp/NetSharp/NetSharp.csproj
@@ -18,7 +18,6 @@
<ItemGroup>
<PackageReference Include="Microsoft.CSharp" Version="4.7.0" />
- <PackageReference Include="System.Dynamic.Runtime" Version="4.3.0" />
- <PackageReference Include="System.Threading.Tasks.Dataflow" Version="4.11.0" />
+ <PackageReference Include="System.Threading.Channels" Version="4.7.0" />
</ItemGroup>
</Project>
\ No newline at end of file
diff --git a/NetSharp/NetSharp/NetSharp.xml b/NetSharp/NetSharp/NetSharp.xml
@@ -1557,6 +1557,11 @@
Holds currently connected and active clients, as well as their current received packet queues.
</summary>
</member>
+ <member name="F:NetSharp.Servers.UdpServer.clientChannelOptions">
+ <summary>
+ The options that should be applied to every channel created to handle a client.
+ </summary>
+ </member>
<member name="M:NetSharp.Servers.UdpServer.#ctor">
<inheritdoc />
</member>
@@ -1787,13 +1792,14 @@
<param name="cancellationToken">The cancellation token to use for asynchronous cancellation.</param>
<returns>The result of the receive operation.</returns>
</member>
- <member name="M:NetSharp.Utils.NetworkOperations.ReadFromAsync(System.Net.Sockets.Socket,System.Net.EndPoint,System.Net.Sockets.SocketFlags,System.Threading.CancellationToken)">
+ <member name="M:NetSharp.Utils.NetworkOperations.ReadFromAsync(System.Net.Sockets.Socket,System.Int32,System.Net.EndPoint,System.Net.Sockets.SocketFlags,System.Threading.CancellationToken)">
<summary>
Reads a datagram asynchronously from the given remote endpoint, via the given socket.
The given <see cref="T:System.Net.Sockets.SocketFlags"/> are associated with the read, and the given <see cref="T:System.Threading.CancellationToken"/>
is used to allow for asynchronous task cancellation.
</summary>
<param name="socket">The socket which should read data from the network.</param>
+ <param name="count">The number of bytes to read from the network.</param>
<param name="remoteEndPoint">The remote endpoint from which data should be read.</param>
<param name="socketFlags">The socket flags associated with the receive operation.</param>
<param name="cancellationToken">The cancellation token to use for asynchronous cancellation.</param>
diff --git a/NetSharp/NetSharp/Servers/UdpServer.cs b/NetSharp/NetSharp/Servers/UdpServer.cs
@@ -2,8 +2,8 @@
using System.Collections.Concurrent;
using System.Net;
using System.Net.Sockets;
+using System.Threading.Channels;
using System.Threading.Tasks;
-using System.Threading.Tasks.Dataflow;
using NetSharp.Interfaces;
using NetSharp.Packets;
using NetSharp.Packets.Builtin;
@@ -20,19 +20,28 @@ namespace NetSharp.Servers
/// <summary>
/// Holds currently connected and active clients, as well as their current received packet queues.
/// </summary>
- private readonly ConcurrentDictionary<EndPoint, BufferBlock<Packet>> activeClients;
+ private readonly ConcurrentDictionary<EndPoint, Channel<Packet>> activeClients;
+
+ /// <summary>
+ /// The options that should be applied to every channel created to handle a client.
+ /// </summary>
+ private static readonly UnboundedChannelOptions clientChannelOptions = new UnboundedChannelOptions
+ {
+ SingleReader = true,
+ SingleWriter = true
+ };
/// <inheritdoc />
public UdpServer() : base(SocketType.Dgram, ProtocolType.Udp, SocketOptionManager.Udp)
{
- activeClients = new ConcurrentDictionary<EndPoint, BufferBlock<Packet>>();
+ activeClients = new ConcurrentDictionary<EndPoint, Channel<Packet>>();
}
/// <inheritdoc />
protected override async Task HandleClientAsync(ClientHandlerArgs args)
{
EndPoint clientEndPoint = args.ClientEndPoint;
- BufferBlock<Packet> clientPacketBuffer = activeClients[clientEndPoint];
+ Channel<Packet> clientPacketBuffer = activeClients[clientEndPoint];
logger.LogMessage($"Initialised client handler for client socket: [Remote EP: {clientEndPoint}]");
@@ -41,7 +50,7 @@ namespace NetSharp.Servers
do
{
// receive a single raw packet from the network
- Packet rawRequest = await clientPacketBuffer.ReceiveAsync(serverShutdownCancellationTokenSource.Token);
+ Packet rawRequest = await clientPacketBuffer.Reader.ReadAsync(serverShutdownCancellationTokenSource.Token);
if (rawRequest.Equals(NullPacket) || rawRequest.Type == PacketRegistry.GetPacketId<DisconnectPacket>())
{
@@ -89,8 +98,15 @@ namespace NetSharp.Servers
logger.LogMessage($"Stopping client handler for client socket: [Remote EP: {clientEndPoint}]");
- activeClients.TryRemove(clientEndPoint, out BufferBlock<Packet> remainingPackets);
- logger.LogMessage($"Client handler has {remainingPackets.Count} packets left, which will be dropped");
+ if (activeClients.TryRemove(clientEndPoint, out Channel<Packet> packetChannel))
+ {
+ packetChannel.Writer.Complete();
+ logger.LogMessage($"Shutting down packet channel for client socket: [Remote EP: {clientEndPoint}]");
+ }
+ else
+ {
+ logger.LogMessage($"Couldn't shut down packet channel for client socket: [Remote EP: {clientEndPoint}]");
+ }
}
catch (TaskCanceledException) { logger.LogMessage("Client handling was cancelled via a task cancellation."); }
catch (OperationCanceledException) { logger.LogMessage("Client handling was cancelled via an operation cancellation."); }
@@ -130,7 +146,7 @@ namespace NetSharp.Servers
{
ClientHandlerArgs args = ClientHandlerArgs.ForUdpClientHandler(in clientEndPoint);
- activeClients.TryAdd(clientEndPoint, new BufferBlock<Packet>());
+ activeClients.TryAdd(clientEndPoint, Channel.CreateUnbounded<Packet>(clientChannelOptions));
await Task.Factory.StartNew(DoHandleClientAsync, args,
serverShutdownCancellationTokenSource.Token,
@@ -138,7 +154,7 @@ namespace NetSharp.Servers
TaskScheduler.Current);
}
- activeClients[clientEndPoint].Post(request);
+ await activeClients[clientEndPoint].Writer.WriteAsync(request);
}
OnServerStopped();
diff --git a/NetSharp/NetSharp/Utils/NetworkOperations.cs b/NetSharp/NetSharp/Utils/NetworkOperations.cs
@@ -25,7 +25,7 @@ namespace NetSharp.Utils
/// <param name="cancellationToken">The cancellation token to use for asynchronous cancellation.</param>
/// <returns>The result of the receive operation.</returns>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
- public static Task<TransmissionResult> ReadAsync(Socket socket, int count, SocketFlags socketFlags,
+ internal static Task<TransmissionResult> ReadAsync(Socket socket, int count, SocketFlags socketFlags,
CancellationToken cancellationToken)
{
return Task.Factory.StartNew(() =>
@@ -50,24 +50,30 @@ namespace NetSharp.Utils
/// is used to allow for asynchronous task cancellation.
/// </summary>
/// <param name="socket">The socket which should read data from the network.</param>
+ /// <param name="count">The number of bytes to read from the network.</param>
/// <param name="remoteEndPoint">The remote endpoint from which data should be read.</param>
/// <param name="socketFlags">The socket flags associated with the receive operation.</param>
/// <param name="cancellationToken">The cancellation token to use for asynchronous cancellation.</param>
/// <returns>The result of the receive operation.</returns>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
- public static Task<TransmissionResult> ReadFromAsync(Socket socket, EndPoint remoteEndPoint,
+ internal static Task<TransmissionResult> ReadFromAsync(Socket socket, int count, EndPoint remoteEndPoint,
SocketFlags socketFlags, CancellationToken cancellationToken)
{
return Task.Factory.StartNew(() =>
{
- byte[] byteBuffer = new byte[Constants.UdpMaxBufferSize];
-
+ byte[] byteBuffer = new byte[count];
EndPoint actualRemoteEndPoint = remoteEndPoint;
+ int receivedBytesCount = 0;
+
+ while (count > receivedBytesCount)
+ {
+ IPPacketInformation _;
- int receivedByteCount = socket.ReceiveFrom(byteBuffer, socketFlags, ref actualRemoteEndPoint);
+ receivedBytesCount += socket.ReceiveMessageFrom(byteBuffer, receivedBytesCount,
+ count - receivedBytesCount, ref socketFlags, ref actualRemoteEndPoint, out _);
+ }
- return new TransmissionResult(new Memory<byte>(byteBuffer, 0, receivedByteCount), receivedByteCount,
- actualRemoteEndPoint);
+ return new TransmissionResult(byteBuffer, receivedBytesCount, actualRemoteEndPoint);
}, cancellationToken);
}
@@ -119,10 +125,28 @@ namespace NetSharp.Utils
public static async Task<(Packet packet, TransmissionResult packetResult)> ReadPacketFromAsync(
Socket socket, EndPoint remoteEndPoint, SocketFlags socketFlags, CancellationToken cancellationToken)
{
- TransmissionResult packetResult =
- await ReadFromAsync(socket, remoteEndPoint, socketFlags, cancellationToken);
+ TransmissionResult packetHeaderResult =
+ await ReadFromAsync(socket, Packet.HeaderSize, remoteEndPoint, socketFlags, cancellationToken);
+
+ int packetSize = EndianAwareBitConverter.ToInt32(packetHeaderResult.Buffer.Span.Slice(0, sizeof(int)));
+
+ if (packetSize == 0)
+ {
+ return (Packet.Deserialise(packetHeaderResult.Buffer), packetHeaderResult);
+ }
+
+ TransmissionResult packetDataResult =
+ await ReadFromAsync(socket, packetSize, packetHeaderResult.RemoteEndPoint, socketFlags, cancellationToken);
+
+ byte[] serialisedPacket = new byte[Packet.HeaderSize + packetSize];
+
+ Memory<byte> serialisedPacketHeader = new Memory<byte>(serialisedPacket, 0, Packet.HeaderSize);
+ packetHeaderResult.Buffer.CopyTo(serialisedPacketHeader);
+
+ Memory<byte> serialisedPacketData = new Memory<byte>(serialisedPacket, Packet.HeaderSize, packetSize);
+ packetDataResult.Buffer.CopyTo(serialisedPacketData);
- return (Packet.Deserialise(packetResult.Buffer), packetResult);
+ return (Packet.Deserialise(serialisedPacket), packetDataResult);
}
/// <summary>
@@ -135,7 +159,7 @@ namespace NetSharp.Utils
/// <param name="socketFlags">The socket flags associated with the send operation.</param>
/// <param name="cancellationToken">The cancellation token to use for asynchronous cancellation.</param>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
- public static Task WriteAsync(Socket socket, ReadOnlyMemory<byte> buffer, SocketFlags socketFlags,
+ internal static Task WriteAsync(Socket socket, ReadOnlyMemory<byte> buffer, SocketFlags socketFlags,
CancellationToken cancellationToken)
{
return Task.Factory.StartNew(() =>
@@ -192,7 +216,7 @@ namespace NetSharp.Utils
/// <param name="socketFlags">The socket flags associated with the send operation.</param>
/// <param name="cancellationToken">The cancellation token to use for asynchronous cancellation.</param>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
- public static Task WriteToAsync(Socket socket, EndPoint remoteEndPoint, ReadOnlyMemory<byte> buffer,
+ internal static Task WriteToAsync(Socket socket, EndPoint remoteEndPoint, ReadOnlyMemory<byte> buffer,
SocketFlags socketFlags, CancellationToken cancellationToken)
{
return Task.Factory.StartNew(() =>
diff --git a/NetSharp/NetSharpExamples/Program.cs b/NetSharp/NetSharpExamples/Program.cs
@@ -147,7 +147,7 @@ namespace NetSharpExamples
await using Stream serverOutputStream = File.OpenWrite(serverLogFile);
using Server server = new UdpServer();
- server.ChangeLoggingStream(Console.OpenStandardOutput());
+ server.ChangeLoggingStream(Console.OpenStandardOutput(), LogLevel.Warn);
//server.ChangeLoggingStream(serverOutputStream, LogLevel.Error);
Console.WriteLine("Starting server...");
