From 3a4456e362b6606162884312c56c12a94ba6b078 Mon Sep 17 00:00:00 2001 From: Mikhail Mitrofanov Date: Tue, 8 Apr 2025 11:48:46 +0300 Subject: [PATCH] Refactoring and prepare for interaction module reconnection --- Example.Backend/LoadTestImp.cs | 3 +- Example.Backend/Page.cs | 1 + Example.Frontend/Program.cs | 8 +- mROA/Abstract/IContextRepositoryHub.cs | 1 + mROA/Abstract/IInteractionModule.cs | 2 +- .../Backend/ContextRepository.cs | 4 - .../Backend/HubRequestExtractor.cs | 8 +- .../Backend/MultiClientContextRepository.cs | 5 + .../Backend/NetworkGatewayModule.cs | 32 ++++-- .../Frontend/NetworkFrontendBridge.cs | 6 +- mROA/Implementation/IdAssignment.cs | 4 + .../NextGenerationInteractionModule.cs | 104 ++++++++++++++++-- mROA/mROA.csproj | 2 +- 13 files changed, 143 insertions(+), 37 deletions(-) diff --git a/Example.Backend/LoadTestImp.cs b/Example.Backend/LoadTestImp.cs index a1fec60..d92c7f5 100644 --- a/Example.Backend/LoadTestImp.cs +++ b/Example.Backend/LoadTestImp.cs @@ -40,10 +40,11 @@ namespace Example.Backend Console.WriteLine("Waiting canceled"); return; } + Console.WriteLine("Waiting..."); await Task.Delay(1000); } - + Console.WriteLine("Waited until the end"); } diff --git a/Example.Backend/Page.cs b/Example.Backend/Page.cs index 7e2fb4d..d55bac8 100644 --- a/Example.Backend/Page.cs +++ b/Example.Backend/Page.cs @@ -6,6 +6,7 @@ namespace Example.Backend public class Page : IPage { public string Text; + public byte[] GetData() { return Encoding.UTF8.GetBytes(Text); diff --git a/Example.Frontend/Program.cs b/Example.Frontend/Program.cs index 75291dc..ea7090d 100644 --- a/Example.Frontend/Program.cs +++ b/Example.Frontend/Program.cs @@ -39,14 +39,14 @@ class Program TransmissionConfig.RealContextRepository = builder.GetModule(); TransmissionConfig.RemoteEndpointContextRepository = builder.GetModule(); - builder.GetModule()!.Connect(); + var frontendBridge = builder.GetModule()!; + frontendBridge.Connect(); _ = builder.GetModule()!.StartExtraction(); Console.WriteLine(TransmissionConfig.OwnershipRepository.GetOwnershipId()); var context = builder.GetModule(); var factory = context.GetSingleObject(typeof(IPrinterFactory), 0) as IPrinterFactory; -//правильный порядок команд 8-5-10-7 using (var disposingPrinter = factory.Create("Test")) { DemoCheck.CreatingPrinter = true; @@ -58,6 +58,7 @@ class Program Console.WriteLine("Printer created"); Thread.Sleep(100); + // frontendBridge.Obstacle(); var name = disposingPrinter.GetName(); DemoCheck.BasicNonParamsCall = true; Console.WriteLine("Printer name : {0}", name); @@ -114,6 +115,9 @@ class Program cts.Cancel(); Console.WriteLine($"Token state {cts.Token.IsCancellationRequested}"); DemoCheck.TaskCancelation = true; + + frontendBridge.Disconnect(); + DemoCheck.Show(); Console.ReadKey(); // diff --git a/mROA/Abstract/IContextRepositoryHub.cs b/mROA/Abstract/IContextRepositoryHub.cs index f950b8f..19a7987 100644 --- a/mROA/Abstract/IContextRepositoryHub.cs +++ b/mROA/Abstract/IContextRepositoryHub.cs @@ -3,5 +3,6 @@ namespace mROA.Abstract public interface IContextRepositoryHub { IContextRepository GetRepository(int clientId); + void FreeRepository(int clientId); } } \ No newline at end of file diff --git a/mROA/Abstract/IInteractionModule.cs b/mROA/Abstract/IInteractionModule.cs index edf66aa..0b5e657 100644 --- a/mROA/Abstract/IInteractionModule.cs +++ b/mROA/Abstract/IInteractionModule.cs @@ -15,7 +15,7 @@ namespace mROA.Abstract NetworkMessageHeader[] UnhandledMessages { get; } NetworkMessageHeader? FirstByFilter(Predicate predicate); event Action OnDisconected; - Task Restart(); + Task Restart(bool sendRecovery); } } \ No newline at end of file diff --git a/mROA/Implementation/Backend/ContextRepository.cs b/mROA/Implementation/Backend/ContextRepository.cs index 71ee389..b4ecc98 100644 --- a/mROA/Implementation/Backend/ContextRepository.cs +++ b/mROA/Implementation/Backend/ContextRepository.cs @@ -10,15 +10,11 @@ namespace mROA.Implementation.Backend { public class ContextRepository : IContextRepository { - private const int StartupSize = 1024; - private const int GrowSize = 128; public static object[] EventBinders = { }; private static int LastDebugId = -1; private int _debugId = -1; - private Task _lastIndexFinder = Task.FromResult(0); - private IRepresentationModuleProducer? _representationModuleProducer; // [CanBeNull] diff --git a/mROA/Implementation/Backend/HubRequestExtractor.cs b/mROA/Implementation/Backend/HubRequestExtractor.cs index 6529c6c..71171de 100644 --- a/mROA/Implementation/Backend/HubRequestExtractor.cs +++ b/mROA/Implementation/Backend/HubRequestExtractor.cs @@ -49,9 +49,15 @@ namespace mROA.Implementation.Backend private void HubOnOnConnected(IRepresentationModule interaction) { var extractor = CreateExtractor(interaction); - _ = extractor.StartExtraction(); + extractor.StartExtraction().ContinueWith(t => OnDisconnected(interaction)); } + private void OnDisconnected(IRepresentationModule representationModule) + { + if (_contextRepository is IContextRepositoryHub contextHub) + contextHub.FreeRepository(representationModule.Id); + } + private IRequestExtractor CreateExtractor(IRepresentationModule interaction) { var extractor = (IRequestExtractor)Activator.CreateInstance(_extractorType)!; diff --git a/mROA/Implementation/Backend/MultiClientContextRepository.cs b/mROA/Implementation/Backend/MultiClientContextRepository.cs index 065c746..a0b3893 100644 --- a/mROA/Implementation/Backend/MultiClientContextRepository.cs +++ b/mROA/Implementation/Backend/MultiClientContextRepository.cs @@ -56,6 +56,11 @@ namespace mROA.Implementation.Backend return repository; } + public void FreeRepository(int clientId) + { + _repositories.Remove(clientId); + } + private IContextRepository GetRepositoryByClientId(int clientId) { if (_repositories.TryGetValue(clientId, out var repository)) diff --git a/mROA/Implementation/Backend/NetworkGatewayModule.cs b/mROA/Implementation/Backend/NetworkGatewayModule.cs index f4585f2..6bb8cda 100644 --- a/mROA/Implementation/Backend/NetworkGatewayModule.cs +++ b/mROA/Implementation/Backend/NetworkGatewayModule.cs @@ -66,6 +66,7 @@ namespace mROA.Implementation.Backend while (true) { var client = _tcpListener.AcceptTcpClient(); + Console.WriteLine($"Client connected from {client.Client.RemoteEndPoint}"); var interaction = Activator.CreateInstance(_interactionModuleType!) as INextGenerationInteractionModule; foreach (var injectableModule in _injectableModules!) @@ -77,19 +78,26 @@ namespace mROA.Implementation.Backend var connectionRequest = interaction.GetNextMessageReceiving().GetAwaiter().GetResult()!; - if (connectionRequest.MessageType == EMessageType.ClientConnect) + switch (connectionRequest.MessageType) { - interaction.PostMessageAsync(new NetworkMessageHeader(_serialization!, - new IdAssignment { Id = -interaction.ConnectionId })); - _hub!.RegisterInteraction(interaction); - Console.WriteLine("Client registered"); - }else if (connectionRequest.MessageType == EMessageType.ClientRecovery) - { - var recoveryRequest = _serialization!.Deserialize(connectionRequest.Data)!; - var recoveryInteraction = _hub.GetInteraction(recoveryRequest.Id); - recoveryInteraction.BaseStream = client.GetStream(); - recoveryInteraction.Restart(); - Console.WriteLine($"Client {recoveryRequest.Id} reconnected"); + case EMessageType.ClientConnect: + interaction.PostMessageAsync(new NetworkMessageHeader(_serialization!, + new IdAssignment { Id = -interaction.ConnectionId })); + _hub!.RegisterInteraction(interaction); + Console.WriteLine("Client registered"); + break; + case EMessageType.ClientRecovery: + { + var recoveryRequest = _serialization!.Deserialize(connectionRequest.Data)!; + var recoveryInteraction = _hub.GetInteraction(recoveryRequest.Id); + recoveryInteraction.BaseStream = client.GetStream(); + recoveryInteraction.Restart(false); + Console.WriteLine($"Client {recoveryRequest.Id} reconnected"); + break; + } + default: + client.Close(); + break; } } } diff --git a/mROA/Implementation/Frontend/NetworkFrontendBridge.cs b/mROA/Implementation/Frontend/NetworkFrontendBridge.cs index b34042a..6aca2a1 100644 --- a/mROA/Implementation/Frontend/NetworkFrontendBridge.cs +++ b/mROA/Implementation/Frontend/NetworkFrontendBridge.cs @@ -43,9 +43,9 @@ namespace mROA.Implementation.Frontend _interactionModule.BaseStream = _tcpClient.GetStream(); - _interactionModule.OnDisconected += async id => + _interactionModule.OnDisconected += id => { - await Reconnect(); + Reconnect(); }; _ = _interactionModule.PostMessageAsync(new NetworkMessageHeader(_serialization, new ClientConnect())); var welcomeMessage = _interactionModule.GetNextMessageReceiving().GetAwaiter().GetResult(); @@ -66,7 +66,7 @@ namespace mROA.Implementation.Frontend _tcpClient = new TcpClient(); _tcpClient.Connect(_serverEndPoint); _interactionModule.BaseStream = _tcpClient.GetStream(); - await _interactionModule.Restart(); + await _interactionModule.Restart(true); } public void Obstacle() diff --git a/mROA/Implementation/IdAssignment.cs b/mROA/Implementation/IdAssignment.cs index f3b3f09..068a5ff 100644 --- a/mROA/Implementation/IdAssignment.cs +++ b/mROA/Implementation/IdAssignment.cs @@ -9,6 +9,10 @@ namespace mROA.Implementation public class ClientRecovery : INetworkMessage { + public ClientRecovery() + { + Id = 0; + } public ClientRecovery(int id) { Id = id; diff --git a/mROA/Implementation/NextGenerationInteractionModule.cs b/mROA/Implementation/NextGenerationInteractionModule.cs index c27fa52..3e5d854 100644 --- a/mROA/Implementation/NextGenerationInteractionModule.cs +++ b/mROA/Implementation/NextGenerationInteractionModule.cs @@ -10,14 +10,23 @@ namespace mROA.Implementation { public class NextGenerationInteractionModule : INextGenerationInteractionModule { + private int DebugId = new Random().Next(); private const int BufferSize = ushort.MaxValue; private readonly Memory _buffer = new byte[BufferSize]; private readonly List _messageBuffer = new(128); private Task? _currentReceiving; private ISerializationToolkit? _serialization; private Stream? _baseStream; + private bool _isRecovering; + private event Action OnReconnected; - private TaskCompletionSource _reconection = new(); + private TaskCompletionSource _reconection; + + public NextGenerationInteractionModule() + { + _reconection = new TaskCompletionSource(); + _reconection.SetResult(Stream.Null); + } public int ConnectionId { get; set; } @@ -54,6 +63,25 @@ namespace mROA.Implementation _currentReceiving = Task.Run(async () => await GetNextMessage()); return _currentReceiving; } +#pragma warning disable CS8602 // Dereference of a possibly null reference. + private async ValueTask PostMessageInternal(NetworkMessageHeader messageHeader) + { + +#if TRACE + Console.WriteLine($"{DateTime.Now.TimeOfDay} Posting message: {messageHeader.Id} - {messageHeader.MessageType} to {ConnectionId}"); +#endif + + var rawMessage = _serialization.Serialize(messageHeader); + var header = BitConverter.GetBytes((ushort)rawMessage.Length).AsMemory(0, sizeof(ushort)); + + if (!_baseStream.CanWrite) + return false; + await BaseStream.WriteAsync(header); + await BaseStream.WriteAsync(rawMessage); + return true; + } +#pragma warning restore CS8602 // Dereference of a possibly null reference. + public async Task PostMessageAsync(NetworkMessageHeader messageHeader) { @@ -65,12 +93,35 @@ namespace mROA.Implementation // Console.WriteLine("Sending {0}", JsonSerializer.Serialize(message)); + bool withError = false; + while (true) + { + if (withError) + { + Console.WriteLine("Post again"); + } - var rawMessage = _serialization.Serialize(messageHeader); - var header = BitConverter.GetBytes((ushort)rawMessage.Length).AsMemory(0, sizeof(ushort)); + if (await PostMessageInternal(messageHeader)) + break; - await BaseStream.WriteAsync(header); - await BaseStream.WriteAsync(rawMessage); + // Console.WriteLine("Try to get lock from post"); + // lock (_reconection) + // { + // Console.WriteLine("Got lock from post"); + // if (!_isRecovering) + // { + // _isRecovering = true; + // Console.WriteLine("Disconnect invoke for post"); + // OnDisconected?.Invoke(ConnectionId); + // Console.WriteLine("Disconnect invoked for post"); + // } + // } + // + // withError = true; + // Console.WriteLine("Start waiting for recovery from post"); + // _ = await _reconection.Task; + // Console.WriteLine("Connection recovered from post"); + } } public void HandleMessage(NetworkMessageHeader messageHeader) @@ -95,16 +146,40 @@ namespace mROA.Implementation if (_serialization == null) throw new NullReferenceException("Serialization toolkit is null"); + bool withError = false; + while (true) { + if (withError) + { + Console.WriteLine("Recieve again"); + } + try { return await Receive(); } catch (Exception) { - OnDisconected!.Invoke(ConnectionId); - _ = await _reconection.Task; + // Console.WriteLine("Try to get lock from receive"); + // lock (_reconection) + // { + // Console.WriteLine("Got lock from receive"); + // + // if (!_isRecovering) + // { + // _isRecovering = true; + // Console.WriteLine("Disconnect invoke"); + // OnDisconected?.Invoke(ConnectionId); + // Console.WriteLine("Disconnect invoked for receive"); + // + // } + // } + // + // withError = true; + // Console.WriteLine("Start waiting for recovery from receive"); + // _ = await _reconection.Task; + // Console.WriteLine("Connection recovered"); } } } @@ -119,7 +194,7 @@ namespace mROA.Implementation return len; } - private async Task Receive() + private async ValueTask Receive() { var len = ReadMessageLength(); var localSpan = _buffer[..len]; @@ -128,9 +203,9 @@ namespace mROA.Implementation var message = _serialization.Deserialize(localSpan.Span); #if TRACE - Console.WriteLine($"{DateTime.Now.TimeOfDay} Received Message {message.Id} - {message.SchemaId}"); + Console.WriteLine($"{DateTime.Now.TimeOfDay} Received Message {message.Id} - {message.MessageType}"); TransmissionConfig.TotalTransmittedBytes += len; - Console.WriteLine($"Total recieced bytes are {TransmissionConfig.TotalTransmittedBytes}"); + Console.WriteLine($"Total received bytes are {TransmissionConfig.TotalTransmittedBytes}"); #endif _messageBuffer.Add(message); _currentReceiving = Task.Run(async () => await GetNextMessage()); @@ -138,10 +213,15 @@ namespace mROA.Implementation return message; } - public async Task Restart() + public async Task Restart(bool sendRecovery) { - await PostMessageAsync(new NetworkMessageHeader(_serialization!, new ClientRecovery(Math.Abs(ConnectionId)))); + if (sendRecovery) + await PostMessageAsync( + new NetworkMessageHeader(_serialization!, new ClientRecovery(Math.Abs(ConnectionId)))); + _isRecovering = false; _reconection.SetResult(BaseStream!); + _reconection = new TaskCompletionSource(); + } } } \ No newline at end of file diff --git a/mROA/mROA.csproj b/mROA/mROA.csproj index 5c9aea9..7be3a6c 100644 --- a/mROA/mROA.csproj +++ b/mROA/mROA.csproj @@ -21,7 +21,7 @@ - + TRACE;