From 50236c63093e8bcd6749ba34fb43c0bbedfea970 Mon Sep 17 00:00:00 2001 From: Mikhail Mitrofanov Date: Wed, 19 Feb 2025 18:51:40 +0300 Subject: [PATCH 1/2] =?UTF-8?q?=D0=9F=D0=BE=D0=B4=D0=B3=D0=BE=D1=82=D0=BE?= =?UTF-8?q?=D0=B2=D0=BA=D0=B0=20=D0=BF=D0=B5=D1=80=D0=B5=D1=85=D0=BE=D0=B4?= =?UTF-8?q?=D0=B0=20=D0=BD=D0=B0=20EventWaitHandle?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- mROA/Abstract/IInteractionModule.cs | 5 ++- mROA/Implementation/NetworkMessage.cs | 1 + .../NextGenerationInteractionModule.cs | 40 ++++++++++++++----- 3 files changed, 34 insertions(+), 12 deletions(-) diff --git a/mROA/Abstract/IInteractionModule.cs b/mROA/Abstract/IInteractionModule.cs index cb1072e..83f6ee5 100644 --- a/mROA/Abstract/IInteractionModule.cs +++ b/mROA/Abstract/IInteractionModule.cs @@ -6,9 +6,12 @@ public interface INextGenerationInteractionModule : IInjectableModule { int ConnectionId { get; } public Stream? BaseStream { get; set; } + NetworkMessage[] UnhandledMessages { get; } + NetworkMessage LastMessage { get; } + EventWaitHandle CurrentReceivingHandle { get; } + void StartInfiniteReceiving(); Task GetNextMessageReceiving(); Task PostMessage(NetworkMessage message); void HandleMessage(NetworkMessage message); - NetworkMessage[] UnhandledMessages { get; } NetworkMessage? FirstByFilter(Predicate predicate); } \ No newline at end of file diff --git a/mROA/Implementation/NetworkMessage.cs b/mROA/Implementation/NetworkMessage.cs index 9b524f8..11f8035 100644 --- a/mROA/Implementation/NetworkMessage.cs +++ b/mROA/Implementation/NetworkMessage.cs @@ -5,6 +5,7 @@ namespace mROA.Implementation; public class NetworkMessage { + public static readonly NetworkMessage Null = new() {SchemaId = MessageType.Unknown, Id = Guid.Empty, Data = []}; public Guid Id { get; init; } [JsonConverter(typeof(JsonStringEnumConverter))] public MessageType SchemaId { get; init; } diff --git a/mROA/Implementation/NextGenerationInteractionModule.cs b/mROA/Implementation/NextGenerationInteractionModule.cs index 0036eda..e96ed15 100644 --- a/mROA/Implementation/NextGenerationInteractionModule.cs +++ b/mROA/Implementation/NextGenerationInteractionModule.cs @@ -11,7 +11,9 @@ public class NextGenerationInteractionModule : INextGenerationInteractionModule private const int BufferSize = ushort.MaxValue; private readonly Memory _buffer = new byte[BufferSize]; private readonly List _messageBuffer = new (128); - + public NetworkMessage[] UnhandledMessages => _messageBuffer.ToArray(); + public NetworkMessage LastMessage { get; private set; } = NetworkMessage.Null; + public EventWaitHandle CurrentReceivingHandle { get; private set; } = new(false, EventResetMode.ManualReset); public void Inject(T dependency) { @@ -27,6 +29,19 @@ public class NextGenerationInteractionModule : INextGenerationInteractionModule } + public async void StartInfiniteReceiving() + { + while (true) + { + var message = ReceiveMessage(); + LastMessage = message; + CurrentReceivingHandle.Set(); + CurrentReceivingHandle = new EventWaitHandle(false, EventResetMode.ManualReset); + } + } + + + public Task GetNextMessageReceiving() { if (_currentReceiving != null) return _currentReceiving; @@ -56,7 +71,6 @@ public class NextGenerationInteractionModule : INextGenerationInteractionModule _messageBuffer.Remove(message); } - public NetworkMessage[] UnhandledMessages => _messageBuffer.ToArray(); public NetworkMessage? FirstByFilter(Predicate predicate) { return _messageBuffer.FirstOrDefault(m => predicate(m)); @@ -71,17 +85,21 @@ public class NextGenerationInteractionModule : INextGenerationInteractionModule throw new NullReferenceException("Serialization toolkit is null"); - // Console.WriteLine("Receiving message"); - var len = BitConverter.ToUInt16([(byte)BaseStream.ReadByte(), (byte)BaseStream.ReadByte()]); - var localSpan = _buffer.Span.Slice(0, len); - BaseStream.ReadExactly(localSpan); - - // Console.WriteLine("Receiving {0}", Encoding.Default.GetString(_buffer[..len])); - - var message = _serialization.Deserialize(localSpan); - _messageBuffer.Add(message!); + var message = ReceiveMessage(); + + _messageBuffer.Add(message); _currentReceiving = Task.Run(GetNextMessage); return message!; } + + private NetworkMessage ReceiveMessage() + { + var len = BitConverter.ToUInt16([(byte)BaseStream!.ReadByte(), (byte)BaseStream.ReadByte()]); + var localSpan = _buffer.Span.Slice(0, len); + BaseStream.ReadExactly(localSpan); + + var message = _serialization!.Deserialize(localSpan); + return message!; + } } \ No newline at end of file From 97f7fc29671e212a739d826321cf8ea1e17be292 Mon Sep 17 00:00:00 2001 From: Mikhail Mitrofanov Date: Wed, 19 Feb 2025 20:10:19 +0300 Subject: [PATCH 2/2] =?UTF-8?q?=D0=9F=D1=80=D0=BE=D0=B8=D0=B7=D0=B2=D0=BE?= =?UTF-8?q?=D0=B4=D0=B8=D1=82=D0=B5=D0=BB=D1=8C=D0=BD=D0=BE=D1=81=D1=82?= =?UTF-8?q?=D1=8C=20=D0=B2=D1=8B=D1=88=D0=B5=20=D0=BA=D1=80=D1=8B=D1=88?= =?UTF-8?q?=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Example.Frontend/Program.cs | 1 + mROA.Test/NextGenTest.cs | 1 - mROA/Abstract/IInteractionModule.cs | 2 +- mROA/Abstract/ISerialisationModule.cs | 6 +- .../Backend/NetworkGatewayModule.cs | 2 +- .../Frontend/NetworkFrontendBridge.cs | 8 +- .../Frontend/RequestExtractor.cs | 8 +- mROA/Implementation/NetworkMessage.cs | 17 +++- .../NextGenerationInteractionModule.cs | 34 +++++--- mROA/Implementation/RemoteObjectBase.cs | 17 +++- mROA/Implementation/RepresentationModule.cs | 81 ++++++++++++------- 11 files changed, 121 insertions(+), 56 deletions(-) diff --git a/Example.Frontend/Program.cs b/Example.Frontend/Program.cs index fdbd77d..1f73baa 100644 --- a/Example.Frontend/Program.cs +++ b/Example.Frontend/Program.cs @@ -71,6 +71,7 @@ var x = 0; for (int i = 0; i < iterations; i++) { x = loadSingleton.Next(x); + // Console.WriteLine(x); } timer.Stop(); diff --git a/mROA.Test/NextGenTest.cs b/mROA.Test/NextGenTest.cs index 7293877..99025f3 100644 --- a/mROA.Test/NextGenTest.cs +++ b/mROA.Test/NextGenTest.cs @@ -31,7 +31,6 @@ public class NextGenTest { _listener.Start(); _interactionModuleB.BaseStream = _listener.AcceptTcpClient().GetStream(); - foreach (var guid in guids) { _interactionModuleB.PostMessage(new NetworkMessage { Id = guid, Data = "Hello user"u8.ToArray() }); diff --git a/mROA/Abstract/IInteractionModule.cs b/mROA/Abstract/IInteractionModule.cs index 83f6ee5..9068ba1 100644 --- a/mROA/Abstract/IInteractionModule.cs +++ b/mROA/Abstract/IInteractionModule.cs @@ -5,7 +5,7 @@ namespace mROA.Abstract; public interface INextGenerationInteractionModule : IInjectableModule { int ConnectionId { get; } - public Stream? BaseStream { get; set; } + public Stream BaseStream { get; set; } NetworkMessage[] UnhandledMessages { get; } NetworkMessage LastMessage { get; } EventWaitHandle CurrentReceivingHandle { get; } diff --git a/mROA/Abstract/ISerialisationModule.cs b/mROA/Abstract/ISerialisationModule.cs index 14ea0ef..d18e78e 100644 --- a/mROA/Abstract/ISerialisationModule.cs +++ b/mROA/Abstract/ISerialisationModule.cs @@ -1,3 +1,4 @@ +using System.Windows.Input; using mROA.Implementation; using mROA.Implementation.CommandExecution; @@ -20,9 +21,10 @@ public interface ISerialisationModule : IInjectableModule public interface IRepresentationModule : IInjectableModule { int Id { get; } - Task GetMessageAsync(Guid? requestId = null, MessageType? messageType = null); + Task GetMessageAsync(Guid? requestId = null, MessageType? messageType = null, CancellationToken token = default); T GetMessage(Guid? requestId = null, MessageType? messageType = null); - Task GetRawMessage(Guid? requestId = null, MessageType? messageType = null); + T GetMessage(Predicate filter); + Task GetRawMessage(Predicate filter, CancellationToken token = default); Task PostCallMessageAsync(Guid id, MessageType messageType, T payload) where T : notnull; Task PostCallMessageAsync(Guid id, MessageType messageType, object payload, Type payloadType); diff --git a/mROA/Implementation/Backend/NetworkGatewayModule.cs b/mROA/Implementation/Backend/NetworkGatewayModule.cs index c62cfc1..5ae5206 100644 --- a/mROA/Implementation/Backend/NetworkGatewayModule.cs +++ b/mROA/Implementation/Backend/NetworkGatewayModule.cs @@ -68,7 +68,7 @@ public class NetworkGatewayModule : IGatewayModule interaction!.Inject(_serialization); interaction.BaseStream = client.GetStream(); - + interaction.StartInfiniteReceiving(); interaction.PostMessage(new NetworkMessage { Id = Guid.NewGuid(), SchemaId = MessageType.IdAssigning, diff --git a/mROA/Implementation/Frontend/NetworkFrontendBridge.cs b/mROA/Implementation/Frontend/NetworkFrontendBridge.cs index a5f4a16..5c1920b 100644 --- a/mROA/Implementation/Frontend/NetworkFrontendBridge.cs +++ b/mROA/Implementation/Frontend/NetworkFrontendBridge.cs @@ -32,12 +32,16 @@ public class NetworkFrontendBridge(IPEndPoint ipEndPoint) : IFrontendBridge _tcpClient.Connect(ipEndPoint); _interactionModule.BaseStream = _tcpClient.GetStream(); - var welcomeMessage = _interactionModule.GetNextMessageReceiving().GetAwaiter().GetResult(); + _interactionModule.StartInfiniteReceiving(); + + var handle = _interactionModule.CurrentReceivingHandle; + handle.WaitOne(); + var welcomeMessage = _interactionModule.LastMessage; if (welcomeMessage.SchemaId != MessageType.IdAssigning) { throw new Exception($"Incorrect message type. Must be IdAssigning, current : {welcomeMessage.SchemaId.ToString()}"); } - + _interactionModule.HandleMessage(welcomeMessage); TransmissionConfig.OwnershipRepository = new StaticOwnershipRepository(_serialization.Deserialize(welcomeMessage.Data)!.Id); } } \ No newline at end of file diff --git a/mROA/Implementation/Frontend/RequestExtractor.cs b/mROA/Implementation/Frontend/RequestExtractor.cs index 486a61b..7da79bd 100644 --- a/mROA/Implementation/Frontend/RequestExtractor.cs +++ b/mROA/Implementation/Frontend/RequestExtractor.cs @@ -56,13 +56,13 @@ public class RequestExtractor : IRequestExtractor try { + var lastCommandId = Guid.Empty; while (true) { - var request = - _representationModule!.GetMessage(messageType: MessageType.CallRequest); - - // Console.WriteLine("Executing {0}", request.Id); + var request = + _representationModule!.GetMessage(m => m.Id != lastCommandId && m.SchemaId == MessageType.CallRequest); + lastCommandId = request.Id; if (request.Parameter is not null) { var parameterType = _methodRepository!.GetMethod(request.CommandId).GetParameters().First() diff --git a/mROA/Implementation/NetworkMessage.cs b/mROA/Implementation/NetworkMessage.cs index 11f8035..5571251 100644 --- a/mROA/Implementation/NetworkMessage.cs +++ b/mROA/Implementation/NetworkMessage.cs @@ -1,19 +1,32 @@ using System.Text.Json.Serialization; + // ReSharper disable UnusedMember.Global namespace mROA.Implementation; public class NetworkMessage { - public static readonly NetworkMessage Null = new() {SchemaId = MessageType.Unknown, Id = Guid.Empty, Data = []}; + public static readonly NetworkMessage Null = new() { SchemaId = MessageType.Unknown, Id = Guid.Empty, Data = [] }; public Guid Id { get; init; } + [JsonConverter(typeof(JsonStringEnumConverter))] public MessageType SchemaId { get; init; } public required byte[] Data { get; init; } + + public bool IsValidMessage(Guid? requestId = null, MessageType? messageType = null) + { + return (requestId is null || Id == requestId) && + (messageType is null || SchemaId == messageType); + } } public enum MessageType { - Unknown, FinishedCommandExecution, ExceptionCommandExecution, AsyncCancelCommandExecution, CallRequest, IdAssigning + Unknown, + FinishedCommandExecution, + ExceptionCommandExecution, + AsyncCancelCommandExecution, + CallRequest, + IdAssigning } \ No newline at end of file diff --git a/mROA/Implementation/NextGenerationInteractionModule.cs b/mROA/Implementation/NextGenerationInteractionModule.cs index e96ed15..a02183c 100644 --- a/mROA/Implementation/NextGenerationInteractionModule.cs +++ b/mROA/Implementation/NextGenerationInteractionModule.cs @@ -6,14 +6,14 @@ public class NextGenerationInteractionModule : INextGenerationInteractionModule { private ISerializationToolkit? _serialization; public int ConnectionId { get; private set; } - public Stream? BaseStream { get; set; } + public Stream BaseStream { get; set; } = Stream.Null; private Task? _currentReceiving; private const int BufferSize = ushort.MaxValue; private readonly Memory _buffer = new byte[BufferSize]; private readonly List _messageBuffer = new (128); public NetworkMessage[] UnhandledMessages => _messageBuffer.ToArray(); public NetworkMessage LastMessage { get; private set; } = NetworkMessage.Null; - public EventWaitHandle CurrentReceivingHandle { get; private set; } = new(false, EventResetMode.ManualReset); + public EventWaitHandle CurrentReceivingHandle { get; private set; } = new(true, EventResetMode.ManualReset); public void Inject(T dependency) { @@ -31,12 +31,23 @@ public class NextGenerationInteractionModule : INextGenerationInteractionModule public async void StartInfiniteReceiving() { - while (true) + try { - var message = ReceiveMessage(); - LastMessage = message; - CurrentReceivingHandle.Set(); CurrentReceivingHandle = new EventWaitHandle(false, EventResetMode.ManualReset); + + await Task.Yield(); + while (true) + { + var message = ReceiveMessage(); + LastMessage = message; + _messageBuffer.Add(message); + CurrentReceivingHandle.Set(); + CurrentReceivingHandle = new EventWaitHandle(false, EventResetMode.ManualReset); + } + } + catch (Exception) + { + // ignored } } @@ -71,9 +82,9 @@ public class NextGenerationInteractionModule : INextGenerationInteractionModule _messageBuffer.Remove(message); } - public NetworkMessage? FirstByFilter(Predicate predicate) + public NetworkMessage FirstByFilter(Predicate predicate) { - return _messageBuffer.FirstOrDefault(m => predicate(m)); + return _messageBuffer.FirstOrDefault(m => predicate(m)) ?? NetworkMessage.Null; } private NetworkMessage GetNextMessage() @@ -83,19 +94,18 @@ public class NextGenerationInteractionModule : INextGenerationInteractionModule if (_serialization == null) throw new NullReferenceException("Serialization toolkit is null"); - - + var message = ReceiveMessage(); _messageBuffer.Add(message); _currentReceiving = Task.Run(GetNextMessage); - return message!; + return message; } private NetworkMessage ReceiveMessage() { - var len = BitConverter.ToUInt16([(byte)BaseStream!.ReadByte(), (byte)BaseStream.ReadByte()]); + var len = BitConverter.ToUInt16([(byte)BaseStream.ReadByte(), (byte)BaseStream.ReadByte()]); var localSpan = _buffer.Span.Slice(0, len); BaseStream.ReadExactly(localSpan); diff --git a/mROA/Implementation/RemoteObjectBase.cs b/mROA/Implementation/RemoteObjectBase.cs index 6a813f2..3794a0e 100644 --- a/mROA/Implementation/RemoteObjectBase.cs +++ b/mROA/Implementation/RemoteObjectBase.cs @@ -15,18 +15,27 @@ public abstract class RemoteObjectBase(int id, IRepresentationModule representat var request = new DefaultCallRequest { CommandId = methodId, ObjectId = id, Parameter = parameter, ParameterType = parameter?.GetType() }; await representationModule.PostCallMessageAsync(request.Id, MessageType.CallRequest, request); - + + var localTokenSource = new CancellationTokenSource(); + var successResponse = representationModule.GetMessageAsync>( - messageType: MessageType.FinishedCommandExecution, requestId: request.Id); + messageType: MessageType.FinishedCommandExecution, requestId: request.Id, token: localTokenSource.Token); var errorResponse = representationModule.GetMessageAsync( - messageType: MessageType.ExceptionCommandExecution, requestId: request.Id); + messageType: MessageType.ExceptionCommandExecution, requestId: request.Id, token: localTokenSource.Token); + Task.WaitAny(successResponse, errorResponse); + + if (successResponse.IsCompletedSuccessfully) + { + await localTokenSource.CancelAsync(); return successResponse.Result.Result!; - + } + + await localTokenSource.CancelAsync(); throw errorResponse.Result.GetException(); } diff --git a/mROA/Implementation/RepresentationModule.cs b/mROA/Implementation/RepresentationModule.cs index 4a119b9..038a9ac 100644 --- a/mROA/Implementation/RepresentationModule.cs +++ b/mROA/Implementation/RepresentationModule.cs @@ -22,47 +22,74 @@ public class RepresentationModule : IRepresentationModule public int Id => (_interaction ?? throw new NullReferenceException("Interaction is not initialized")).ConnectionId; - public async Task GetMessageAsync(Guid? requestId, MessageType? messageType) + public async Task GetMessageAsync(Guid? requestId, MessageType? messageType, + CancellationToken token = default) { if (_serialization == null) throw new NullReferenceException("Serialization toolkit is not initialized"); - - return _serialization.Deserialize(await GetRawMessage(requestId, messageType))!; + + return _serialization.Deserialize(await GetRawMessage(m => m.IsValidMessage(requestId, messageType), token))!; } public T GetMessage(Guid? requestId = null, MessageType? messageType = null) { - if (_serialization == null) - throw new NullReferenceException("Serialization toolkit is not initialized"); - - return _serialization.Deserialize(GetRawMessage(requestId, messageType).GetAwaiter().GetResult())!; + return GetMessage(m => m.IsValidMessage(requestId, messageType)); } - public async Task GetRawMessage(Guid? requestId = null, MessageType? messageType = null) + public T GetMessage(Predicate filter) { + if (_serialization == null) + throw new NullReferenceException("Serialization toolkit is not initialized"); + + return _serialization.Deserialize(GetRawMessage(filter).GetAwaiter().GetResult())!; + } + + public async Task GetRawMessage(Predicate filter, CancellationToken token = default) + { + await Task.Yield(); + if (_interaction == null) throw new NullReferenceException("Interaction toolkit is not initialized"); - - var fromBuffer = - _interaction.FirstByFilter(message => - (requestId is null || message.Id == requestId) && - (messageType is null || message.SchemaId == messageType)); - - if (fromBuffer == null) + + // Console.WriteLine( + // $"{DateTime.Now.TimeOfDay} {Environment.CurrentManagedThreadId} Representation : Reading"); + + if (filter(_interaction.LastMessage)) { - while (true) - { - var message = await _interaction.GetNextMessageReceiving(); - if ((requestId is not null && message.Id != requestId) || - (messageType is not null && message.SchemaId != messageType)) continue; - - _interaction.HandleMessage(message); - return message.Data; - } + _interaction.HandleMessage(_interaction.LastMessage); + return _interaction.LastMessage.Data; } - _interaction.HandleMessage(fromBuffer); - return fromBuffer.Data; + var message = _interaction.FirstByFilter(filter); + + if (message != NetworkMessage.Null) + { + _interaction.HandleMessage(message); + return message.Data; + } + + + while (!token.IsCancellationRequested) + { + // Console.WriteLine( + // $"{DateTime.Now.TimeOfDay} {Environment.CurrentManagedThreadId} Representation : Receiving message..."); + var handle = _interaction.CurrentReceivingHandle; + handle.WaitOne(); + + message = _interaction.LastMessage; + + // Console.WriteLine( + // $"{DateTime.Now.TimeOfDay} {Environment.CurrentManagedThreadId} Representation : Message received {message.SchemaId} - {message.Id}"); + if (!filter(message)) continue; + + message = _interaction.LastMessage; + // Console.WriteLine( + // $"{DateTime.Now.TimeOfDay} {Environment.CurrentManagedThreadId} Representation : Message received Successfully {message.SchemaId} - {message.Id}"); + _interaction.HandleMessage(message); + return message.Data; + } + + return []; } public async Task PostCallMessageAsync(Guid id, MessageType messageType, T payload) where T : notnull @@ -76,7 +103,7 @@ public class RepresentationModule : IRepresentationModule throw new NullReferenceException("Interaction toolkit is not initialized"); if (_serialization == null) throw new NullReferenceException("Serialization toolkit is not initialized"); - + await _interaction.PostMessage(new NetworkMessage { Id = id, SchemaId = messageType, Data = _serialization.Serialize(payload, payloadType) }); }