From 0b7c349a6357eca066b135daf94c4a23a7bc02ef Mon Sep 17 00:00:00 2001 From: Alexandru Macocian Date: Thu, 30 Sep 2021 18:42:18 +0200 Subject: [PATCH] Improvements to stream reading routine --- MTSC/CommunicationPrimitives.cs | 8 ++-- MTSC/MTSC.csproj | 6 +-- MTSC/ServerSide/Handlers/FtpHandler.cs | 2 +- MTSC/ServerSide/Handlers/HttpHandler.cs | 8 ++-- .../Handlers/WebsocketRoutingHandler.cs | 2 +- MTSC/ServerSide/Server.cs | 45 +++++++------------ 6 files changed, 30 insertions(+), 41 deletions(-) diff --git a/MTSC/CommunicationPrimitives.cs b/MTSC/CommunicationPrimitives.cs index ca5573e..02d8dc4 100644 --- a/MTSC/CommunicationPrimitives.cs +++ b/MTSC/CommunicationPrimitives.cs @@ -40,7 +40,7 @@ namespace MTSC return new Message((uint)ms.Length, ms.ToArray()); } - public static Message GetMessage(ClientData client, TimeSpan ReadTimeout) + public static async Task GetMessage(ClientData client, TimeSpan ReadTimeout) { Stream stream; if (client.SslStream != null) @@ -58,9 +58,11 @@ namespace MTSC int bytesRead; do { - bytesRead = stream.Read(buffer, 0, buffer.Length); + using var cts = new CancellationTokenSource(); + cts.CancelAfter((int)ReadTimeout.TotalMilliseconds); + bytesRead = await stream.ReadAsync(buffer, 0, buffer.Length, cts.Token); ms.Write(buffer, 0, bytesRead); - } while (bytesRead > 0); + } while (bytesRead > 0 && client.TcpClient.Available > 0); return new Message((uint)ms.Length, ms.ToArray()); } diff --git a/MTSC/MTSC.csproj b/MTSC/MTSC.csproj index 65f85d3..dbecaca 100644 --- a/MTSC/MTSC.csproj +++ b/MTSC/MTSC.csproj @@ -5,13 +5,13 @@ netcoreapp2.1;net48;netstandard2.0;netcoreapp3.1;net5.0 - 4.1.1 + 4.1.2 latest Alexandru-Victor Macocian MTSC Modular TCP Server and Client - 4.1.1.0 - 4.1.1.0 + 4.1.2.0 + 4.1.2.0 true AnyCPU;x64 https://github.com/AlexMacocian/MTSC diff --git a/MTSC/ServerSide/Handlers/FtpHandler.cs b/MTSC/ServerSide/Handlers/FtpHandler.cs index bf95327..e132719 100644 --- a/MTSC/ServerSide/Handlers/FtpHandler.cs +++ b/MTSC/ServerSide/Handlers/FtpHandler.cs @@ -22,7 +22,7 @@ namespace MTSC.ServerSide.Handlers Initialized } - private List ftpModules = new(); + private readonly List ftpModules = new(); public TimeSpan ReadyDelay { get; set; } = TimeSpan.FromSeconds(1); diff --git a/MTSC/ServerSide/Handlers/HttpHandler.cs b/MTSC/ServerSide/Handlers/HttpHandler.cs index 47892b4..d535684 100644 --- a/MTSC/ServerSide/Handlers/HttpHandler.cs +++ b/MTSC/ServerSide/Handlers/HttpHandler.cs @@ -102,11 +102,11 @@ namespace MTSC.ServerSide.Handlers bool IHandler.HandleReceivedMessage(Server server, ClientData client, Message message) { // Parse the request. If the message is incomplete, return 100 and queue the message to be parsed later. - HttpRequest request = null; - byte[] messageBytes = null; + HttpRequest request; try { var trimmedMessageBytes = message.MessageBytes.TrimTrailingNullBytes(); + byte[] messageBytes; if (client.Resources.TryGetResource(out var fragmentedMessage)) { var previousBytes = fragmentedMessage.Message; @@ -147,7 +147,7 @@ namespace MTSC.ServerSide.Handlers else { client.SetAffinity(this); - this.HandleIncompleteRequest(client, server, messageBytes, partialRequest); + this.HandleIncompleteRequest(client, server, messageBytes); if (partialRequest != null && partialRequest.Headers.ContainsHeader(RequestHeaders.Expect) && partialRequest.Headers[RequestHeaders.Expect].Equals("100-continue", StringComparison.OrdinalIgnoreCase)) { @@ -287,7 +287,7 @@ namespace MTSC.ServerSide.Handlers } #endregion - private void HandleIncompleteRequest(ClientData client, Server server, byte[] messageBytes, PartialHttpRequest partialRequest = null) + private void HandleIncompleteRequest(ClientData client, Server server, byte[] messageBytes) { client.Resources.SetResource(new FragmentedMessage() { Message = messageBytes, LastReceived = DateTime.Now }); server.LogDebug("Incomplete request received!"); diff --git a/MTSC/ServerSide/Handlers/WebsocketRoutingHandler.cs b/MTSC/ServerSide/Handlers/WebsocketRoutingHandler.cs index 6e928b9..b5b82c7 100644 --- a/MTSC/ServerSide/Handlers/WebsocketRoutingHandler.cs +++ b/MTSC/ServerSide/Handlers/WebsocketRoutingHandler.cs @@ -28,7 +28,7 @@ namespace MTSC.ServerSide.Handlers Closed } #region Fields - private byte[] emptyData = new byte[0]; + private readonly byte[] emptyData = new byte[0]; private DateTime ellapsedTime = DateTime.Now; private DateTime previousHeartbeatProc = DateTime.Now; private readonly Dictionary)> moduleDictionary = diff --git a/MTSC/ServerSide/Server.cs b/MTSC/ServerSide/Server.cs index 5d76c62..0286dd2 100644 --- a/MTSC/ServerSide/Server.cs +++ b/MTSC/ServerSide/Server.cs @@ -36,7 +36,6 @@ namespace MTSC.ServerSide private readonly List exceptionHandlers = new(); private readonly List serverUsageMonitors = new(); private readonly ProducerConsumerQueue<(ClientData, byte[])> messageOutQueue = new(); - private readonly IServiceManager serviceManager = new ServiceManager(); #endregion #region Private Properties private IConsumerQueue ConsumerClientQueue { get => this.addQueue; } @@ -103,7 +102,7 @@ namespace MTSC.ServerSide /// /// for configuring and retrieving services. /// - public IServiceManager ServiceManager { get => this.serviceManager; } + public IServiceManager ServiceManager { get; } = new ServiceManager(); #endregion #region Constructors /// @@ -162,7 +161,7 @@ namespace MTSC.ServerSide where TService : TInterface where TInterface : class { - this.serviceManager.RegisterTransient(); + this.ServiceManager.RegisterTransient(); return this; } /// @@ -172,7 +171,7 @@ namespace MTSC.ServerSide public Server AddTransientService() where TService : class { - this.serviceManager.RegisterTransient(); + this.ServiceManager.RegisterTransient(); return this; } /// @@ -183,7 +182,7 @@ namespace MTSC.ServerSide where TService : TInterface where TInterface : class { - this.serviceManager.RegisterTransient(serviceFactory); + this.ServiceManager.RegisterTransient(serviceFactory); return this; } /// @@ -193,7 +192,7 @@ namespace MTSC.ServerSide public Server AddTransientService(Func serviceFactory) where TService : class { - this.serviceManager.RegisterTransient(serviceFactory); + this.ServiceManager.RegisterTransient(serviceFactory); return this; } /// @@ -204,7 +203,7 @@ namespace MTSC.ServerSide where TService : TInterface where TInterface : class { - this.serviceManager.RegisterSingleton(); + this.ServiceManager.RegisterSingleton(); return this; } /// @@ -214,7 +213,7 @@ namespace MTSC.ServerSide public Server AddSingletonService() where TService : class { - this.serviceManager.RegisterSingleton(); + this.ServiceManager.RegisterSingleton(); return this; } /// @@ -225,7 +224,7 @@ namespace MTSC.ServerSide where TService : TInterface where TInterface : class { - this.serviceManager.RegisterSingleton(serviceFactory); + this.ServiceManager.RegisterSingleton(serviceFactory); return this; } /// @@ -235,7 +234,7 @@ namespace MTSC.ServerSide public Server AddSingletonService(Func serviceFactory) where TService : class { - this.serviceManager.RegisterSingleton(serviceFactory); + this.ServiceManager.RegisterSingleton(serviceFactory); return this; } /// @@ -406,7 +405,7 @@ namespace MTSC.ServerSide public T GetService() where T : class { - return this.serviceManager.GetService(); + return this.ServiceManager.GetService(); } /// /// Get handler of provided type @@ -528,8 +527,8 @@ namespace MTSC.ServerSide toBeRunOnStartup.OnStartup(this); } - this.serviceManager.RegisterServiceManager(); - this.serviceManager.RegisterSingleton(sp => this); + this.ServiceManager.RegisterServiceManager(); + this.ServiceManager.RegisterSingleton(sp => this); DateTime startLoopTime; while (this.running) { @@ -666,13 +665,6 @@ namespace MTSC.ServerSide while (this.ConsumerMessageOutQueue.TryDequeue(out var tuple)) { (var client, var bytes) = tuple; - if (client.TcpClient.Available > 0) - { - /* - * Don't send while client is still sending - */ - continue; - } try { @@ -744,19 +736,14 @@ namespace MTSC.ServerSide if (client.TcpClient.Available > 0 && !(client as IActiveClient).ReadingData) { (client as IActiveClient).ReadingData = true; - Task.Run(() => + Task.Run(async () => { try { var timeout = this.ReadTimeout; - if (client.TcpClient.Available < 1000) - { - timeout = TimeSpan.FromMilliseconds(50); - } - - var message = CommunicationPrimitives.GetMessage(client, timeout); + var message = await CommunicationPrimitives.GetMessage(client, timeout); (client as IQueueHolder).Enqueue(message); - this.LogDebug($"Received message from {(client.TcpClient.Client.RemoteEndPoint as IPEndPoint)} Message length: {message.MessageLength}"); + this.LogDebug($"Received message from {client.TcpClient.Client.RemoteEndPoint as IPEndPoint} Message length: {message.MessageLength}"); if (this.LogMessageContents) { this.LogDebug(Encoding.UTF8.GetString(message.MessageBytes)); @@ -852,7 +839,7 @@ namespace MTSC.ServerSide this.EncryptionPolicy); client.SslStream = sslStream; - if(sslStream.AuthenticateAsServerAsync(this.certificate, this.RequestClientCertificate, this.SslProtocols, false).Wait(this.SslAuthenticationTimeout)) + if (sslStream.AuthenticateAsServerAsync(this.certificate, this.RequestClientCertificate, this.SslProtocols, false).Wait(this.SslAuthenticationTimeout)) { /* * Client authenticated in the alloted time