Works, but slow for many connections

2025-07-27 00:14:15 +03:00
parent be12f1da65
commit bb3c4887df
24 changed files with 209 additions and 130 deletions
@@ -35,9 +35,12 @@ File.AppendAllText("results.txt", $"[SINGLE CBOR WRITER ALLOC] {totalRequests}\r

 async Task<List<ILoadTest>> GetLoadEndpoints(int count)
 {
+    try
+    {
        var loads = new List<ILoadTest>();
        for (int i = 0; i < count; i++)
        {
+            Console.WriteLine($"Initializing {i}");
            var builder = Host.CreateApplicationBuilder(new HostApplicationBuilderSettings { DisableDefaults = true });
            builder.Services.AddSingleton<IContextualSerializationToolKit, CborSerializationToolkit>();
            builder.Services.AddSingleton<IEndPointContext, EndPointContext>();
@@ -71,29 +74,36 @@ async Task<List<ILoadTest>> GetLoadEndpoints(int count)

            var app = builder.Build();

+            Console.WriteLine($"Connecting {i}");
            var frontendBridge = app.Services.GetService<IFrontendBridge>()!;
            await frontendBridge.Connect();
            // _ = app.Services.GetService<IRequestExtractor>()!.StartExtraction();
            // _ = app.Services.GetService<IUntrustedInteractionModule>().Start(serverEndPoint);
            var context = app.Services.GetService<IInstanceRepository>();
-
+            Console.WriteLine($"Connected {i}");

            var singletonObject =
                context.GetSingletonObject<ILoadTest>(
                    app.Services.GetService<IEndPointContext>());
            loads.Add(singletonObject);
        }
+
        return loads;
+    }
+    catch (Exception e)
+    {
+        Console.WriteLine(e);
+        throw;
+    }
 }

 async Task<int> Requests(CancellationToken token, int id, ILoadTest load)
 {
    try
    {
-
-
        int count = 0;
-        while (true){
+        while (true)
+        {
            if (token.IsCancellationRequested)
            {
                break;
@@ -116,9 +116,9 @@ namespace mROA.Cbor
                return preParsed.ToObject(type, context);


-            if (type == typeof(Guid))
+            if (type == typeof(RequestId))
            {
-                return new Guid((byte[])nonCasted);
+                return new RequestId((byte[])nonCasted);
            }

            return Convert.ChangeType(nonCasted, type);
@@ -164,7 +164,7 @@ namespace mROA.Cbor
                case DateTimeOffset dto:
                    writer.WriteDateTimeOffset(dto);
                    break;
-                case Guid g:
+                case RequestId g:
                    writer.WriteByteString(g.ToByteArray());
                    break;
                case byte[] bytes:
@@ -270,8 +270,8 @@ namespace mROA.Cbor

                    return reader.ReadUInt64();
                case CborReaderState.ByteString:
-                    if (type == typeof(Guid))
-                        return new Guid(reader.ReadByteString());
+                    if (type == typeof(RequestId))
+                        return new RequestId(reader.ReadByteString());
                    return reader.ReadByteString();
                case CborReaderState.TextString:
                    return reader.ReadTextString();
@@ -29,7 +29,7 @@ namespace mROA.Cbor
            reader.ReadStartArray();
            var value = new NetworkMessage
            {
-                Id = new Guid(reader.ReadByteString()),
+                Id = new RequestId(reader.ReadByteString()),
                MessageType = (EMessageType)reader.ReadInt32(),
                Data = reader.ReadByteString()
            };
@@ -57,7 +57,7 @@ namespace mROA.Cbor
            reader.ReadStartArray();
            var value = new DefaultCallRequest
            {
-                Id = new Guid(reader.ReadByteString()),
+                Id = new RequestId(reader.ReadByteString()),
                CommandId = reader.ReadInt32(),
                ObjectId = (ComplexObjectIdentifier)ComplexObjectIdentifierParser.Instance.Read(reader, context, serialization),
                Parameters = serialization.ReadData(reader, typeof(object[]), context) as object[]
@@ -102,7 +102,7 @@ namespace mROA.Cbor
            reader.ReadStartArray();
            var result = new FinalCommandExecution<object>
            {
-                Id = new Guid(reader.ReadByteString()),
+                Id = new RequestId(reader.ReadByteString()),
                Result = serialization.ReadData(reader, typeof(object), context),
            };
             reader.ReadEndArray();
@@ -125,7 +125,7 @@ namespace mROA.Cbor
            reader.ReadStartArray();
            var result = new FinalCommandExecution
            {
-                Id = new Guid(reader.ReadByteString())
+                Id = new RequestId(reader.ReadByteString())
            };
            reader.ReadEndArray();
            return result;
@@ -31,7 +31,7 @@ namespace mROA.Codegen
                            Console.WriteLine("Sending event...");
                            var request = new DefaultCallRequest
                            {
-                                Id = Guid.NewGuid(),
+                                Id = RequestId.Generate(),
                                CommandId = <!L commandId>, 
                                ObjectId = new ComplexObjectIdentifier(index, ownerId), 
                                Parameters = new object[] { <!L transferParameters> }
@@ -20,4 +20,28 @@ public class Identifier
            Assert.Fail();
        }
    }
+
+    [Test]
+    public void RequestIdTest()
+    {
+        var id = RequestId.Generate();
+        Assert.Pass(id.ToString());
+    }
+    
+    [Test]
+    public void EqualsTest()
+    {
+        var id = RequestId.Generate();
+        var id2 = new RequestId { P0 = id.P0, P1 = id.P1 };
+        Assert.That(id2, Is.EqualTo(id));
+    }
+    
+    [Test]
+    public void ByteString()
+    {
+        var id = RequestId.Generate();
+        var binary = id.ToByteArray();
+        var reverced = new RequestId(binary);
+        Assert.That(reverced, Is.EqualTo(id));
+    }
 }
@@ -1,12 +1,13 @@
 using System;
 using System.Threading;
+using mROA.Implementation;

 namespace mROA.Abstract
 {
    public interface ICancellationRepository
    {
-        void RegisterCancellation(Guid id, CancellationTokenSource cts);
-        CancellationTokenSource? GetCancellation(Guid id);
-        void FreeCancelation(Guid id);
+        void RegisterCancellation(RequestId id, CancellationTokenSource cts);
+        CancellationTokenSource? GetCancellation(RequestId id);
+        void FreeCancellation(RequestId id);
    }
 }
@@ -5,6 +5,6 @@ namespace mROA.Abstract
 {
    public interface ICommandExecution : INetworkMessage
    {
-        Guid Id { get; set; }
+        RequestId Id { get; set; }
    }
 }
@@ -19,13 +19,13 @@ namespace mROA.Abstract
            IEndPointContext? context, CancellationToken token = default,
            params Func<NetworkMessage, Type?>[] converter);

-        Task PostCallMessageAsync<T>(Guid id, EMessageType eMessageType, T payload, IEndPointContext? context)
+        Task PostCallMessageAsync<T>(RequestId id, EMessageType eMessageType, T payload, IEndPointContext? context)
            where T : notnull;

-        void PostCallMessage<T>(Guid id, EMessageType eMessageType, T payload, IEndPointContext? context)
+        void PostCallMessage<T>(RequestId id, EMessageType eMessageType, T payload, IEndPointContext? context)
            where T : notnull;

-        Task PostCallMessageUntrustedAsync<T>(Guid id, EMessageType eMessageType, T payload, IEndPointContext? context)
+        Task PostCallMessageUntrustedAsync<T>(RequestId id, EMessageType eMessageType, T payload, IEndPointContext? context)
            where T : notnull;
    }
 }
@@ -114,7 +114,7 @@ namespace mROA.Implementation.Backend
            if (cts == null)
                throw new NullReferenceException("Can't find cancellation for this request");
            cts.Cancel();
-            _cancellationRepo.FreeCancelation(command.Id);
+            _cancellationRepo.FreeCancellation(command.Id);

            return new FinalCommandExecution
            {
@@ -164,7 +164,7 @@ namespace mROA.Implementation.Backend
                {
                    Id = command.Id
                };
-                _cancellationRepo.FreeCancelation(command.Id);
+                _cancellationRepo.FreeCancellation(command.Id);


                if (invoker.IsTrusted)
@@ -195,7 +195,7 @@ namespace mROA.Implementation.Backend
                        Id = command.Id,
                        Result = finalResult
                    };
-                    _cancellationRepo.FreeCancelation(command.Id);
+                    _cancellationRepo.FreeCancellation(command.Id);

                    representationModule.PostCallMessage(command.Id, EMessageType.FinishedCommandExecution,
                        payload, context);
@@ -56,7 +56,7 @@ namespace mROA.Implementation.Backend
            while (true)
            {
                var client = await _tcpListener.AcceptTcpClientAsync();
-                _ = HandleConnection(client).ConfigureAwait(false);
+                _ = HandleConnection(client);
            }
        }

@@ -70,7 +70,7 @@ namespace mROA.Implementation.Backend
                CallIndexProvider = _callIndexProvider
            };
            var streamExtractor =
-                new ChannelInteractionModule.StreamExtractor(client.GetStream(), _serialization, context);
+                new ChannelInteractionModule.StreamExtractor(client.GetStream());
            interaction.IsConnected = () => streamExtractor.IsConnected;
            streamExtractor.MessageReceived = async message =>
            {
@@ -4,7 +4,7 @@ namespace mROA.Implementation
 {
    public interface ICallRequest
    {
-        Guid Id { get; }
+        RequestId Id { get; }
        int CommandId { get; }
        ComplexObjectIdentifier ObjectId { get; }
        object?[]? Parameters { get; }
@@ -12,7 +12,7 @@ namespace mROA.Implementation

    public struct DefaultCallRequest : ICallRequest
    {
-        public Guid Id { get; set; }
+        public RequestId Id { get; set; }
        public int CommandId { get; set; }
        public ComplexObjectIdentifier ObjectId { get; set; }

@@ -26,7 +26,7 @@ namespace mROA.Implementation

    public class CancelRequest : ICallRequest
    {
-        public Guid Id { get; set; }
+        public RequestId Id { get; set; }
        public int CommandId { get; set; } = -2;
        public ComplexObjectIdentifier ObjectId { get; set; } = ComplexObjectIdentifier.Null;
        public object?[]? Parameters { get; set; } = null;
@@ -8,19 +8,19 @@ namespace mROA.Implementation
 {
    public class CancellationRepository : ICancellationRepository
    {
-        private readonly ConcurrentDictionary<Guid, CancellationTokenSource> _cancellations = new();
+        private readonly ConcurrentDictionary<RequestId, CancellationTokenSource> _cancellations = new();

-        public void RegisterCancellation(Guid id, CancellationTokenSource cts)
+        public void RegisterCancellation(RequestId id, CancellationTokenSource cts)
        {
            _cancellations.TryAdd(id, cts);
        }

-        public CancellationTokenSource? GetCancellation(Guid id)
+        public CancellationTokenSource? GetCancellation(RequestId id)
        {
            return _cancellations.GetValueOrDefault(id);
        }

-        public void FreeCancelation(Guid id)
+        public void FreeCancellation(RequestId id)
        {
            _cancellations.Remove(id, out _);
        }
@@ -1,5 +1,6 @@
 using System;
 using System.IO;
+using System.Runtime.InteropServices;
 using System.Threading;
 using System.Threading.Channels;
 using System.Threading.Tasks;
@@ -141,39 +142,34 @@ namespace mROA.Implementation

        public class StreamExtractor
        {
-            private const int BufferSize = ushort.MaxValue + 2;
+            private const int BufferSize = ushort.MaxValue + 19;

            private readonly Stream _ioStream;
-            private readonly IContextualSerializationToolKit _serializationToolkit;
            private readonly Memory<byte> _buffer = new byte[BufferSize];
-            private readonly IEndPointContext _context;
-            private readonly byte[] _lenBuffer;

-            public StreamExtractor(Stream ioStream, IContextualSerializationToolKit serializationToolkit,
-                IEndPointContext context)
+            public StreamExtractor(Stream ioStream)
            {
                _ioStream = ioStream;
-                _serializationToolkit = serializationToolkit;
-                _context = context;
-                _lenBuffer = new byte[2];
            }

            public Action<NetworkMessage> MessageReceived = _ => { };
            
            public async Task SingleReceive(CancellationToken token = default)
            {
-                int firstRead = _ioStream.Read(_buffer.Span);
+                var firstRead = _ioStream.Read(_buffer.Span);
                
-                var metadata = new NetworkMessage.NetworkMessageMeta(_buffer.Span);
+                var meta = MemoryMarshal.Read<NetworkMessage.NetworkMessageMeta>(_buffer.Span);
                
-                var localSpan = _buffer[2..len];
-                if (firstRead - 2 != len)
+                var len = meta.BodyLength;
+                var readLen = firstRead - 19;
+                
+                if (readLen != len)
                {
-                    localSpan = _buffer[(len + 2)..];
-                    await _ioStream.ReadExactlyAsync(localSpan, cancellationToken: token);
+                    var lastPart = _buffer[firstRead..(len + 19)];
+                    await _ioStream.ReadExactlyAsync(lastPart, cancellationToken: token);
                }

-                var message = _serializationToolkit.Deserialize<NetworkMessage>(localSpan, _context);
+                var message = meta.ToMessage(_buffer.Span);
                MessageReceived(message);
            }

@@ -187,11 +183,10 @@ namespace mROA.Implementation

            private async Task Send(NetworkMessage message, CancellationToken token = default)
            {
-                var bodySpan = _buffer[2..];
-                var len = _serializationToolkit.Serialize(message, bodySpan.Span, _context);
-                var header = BitConverter.GetBytes((ushort)len);
-                header.CopyTo(_buffer);
-                var sendingSpan = _buffer[..(len + 2)];
+                var meta = message.ToMeta();
+                MemoryMarshal.Write(_buffer.Span, ref meta);
+                message.Data.CopyTo(_buffer.Span[19..]);
+                var sendingSpan = _buffer[..(19 + meta.BodyLength)];
                await _ioStream.WriteAsync(sendingSpan, token);
                // _logger.LogTrace("SEND {0}", message.ToString());
            }
@@ -5,7 +5,7 @@ namespace mROA.Implementation.CommandExecution
 {
    public class AsyncCommandExecution : ICommandExecution
    {
-        public Guid Id { get; set; }
+        public RequestId Id { get; set; }
        public EMessageType MessageType => EMessageType.Unknown;
    }
 }
@@ -6,7 +6,7 @@ namespace mROA.Implementation.CommandExecution
 {
    public class ExceptionCommandExecution : ICommandExecution
    {
-        public Guid Id { get; set; }
+        public RequestId Id { get; set; }
        public EMessageType MessageType => EMessageType.ExceptionCommandExecution;
        public string Exception { get; set; }

@@ -5,13 +5,13 @@ namespace mROA.Implementation.CommandExecution
 {
    public struct FinalCommandExecution : ICommandExecution
    {
-        public Guid Id { get; set; }
+        public RequestId Id { get; set; }
        public EMessageType MessageType => EMessageType.FinishedCommandExecution;
    }

    public struct FinalCommandExecution<T> : ICommandExecution
    {
-        public Guid Id { get; set; }
+        public RequestId Id { get; set; }
        public EMessageType MessageType => EMessageType.FinishedCommandExecution;
        public T? Result { get; set; }
    }
@@ -28,7 +28,7 @@ namespace mROA.Implementation.Frontend
            _serialization = serialization;
            _interactionModule = interactionModule;
            _rawExtractorCancellation = new CancellationTokenSource();
-            _currentExtractor = new ChannelInteractionModule.StreamExtractor(Stream.Null, _serialization, context);
+            _currentExtractor = new ChannelInteractionModule.StreamExtractor(Stream.Null);
        }

        public async Task Connect()
@@ -63,7 +63,7 @@ namespace mROA.Implementation.Frontend
        private void PrepareExtractor()
        {
            _currentExtractor =
-                new ChannelInteractionModule.StreamExtractor(_tcpClient.GetStream(), _serialization, _context);
+                new ChannelInteractionModule.StreamExtractor(_tcpClient.GetStream());

            _ = _currentExtractor.SendFromChannel(_interactionModule.TrustedPostChanel,
                _rawExtractorCancellation.Token);
@@ -4,7 +4,7 @@ namespace mROA.Implementation.Frontend
 {
    public class RemoteException : Exception
    {
-        public Guid CallRequestId;
+        public RequestId CallRequestId;
        private readonly string _error;

        public RemoteException(string error)
@@ -54,7 +54,7 @@ namespace mROA.Implementation.Frontend
        {
            var initMessage = new NetworkMessage
            {
-                MessageType = EMessageType.UntrustedConnect, Id = Guid.NewGuid(),
+                MessageType = EMessageType.UntrustedConnect, Id = RequestId.Generate(),
                Data = BitConverter.GetBytes(_channelInteractionModule.ConnectionId)
            };

@@ -27,10 +27,10 @@ namespace mROA.Implementation
        {
            MessageType = networkMessage.MessageType;
            Data = serializationToolkit.Serialize(networkMessage, context);
-            Id = Guid.NewGuid();
+            Id = RequestId.Generate();
        }

-        public Guid Id { get; set; }
+        public RequestId Id { get; set; }

        public EMessageType MessageType { get; set; }

@@ -41,18 +41,21 @@ namespace mROA.Implementation
            return $" {Id}:{MessageType} [{Data.Length}]";
        }

+        public NetworkMessageMeta ToMeta()
+        {
+            return new NetworkMessageMeta
+            {
+                BodyLength = (ushort)(Data == null ? 0 : Data.Length),
+                Type = (byte)MessageType,
+                Id = Id
+            };
+        }
+        
        public struct NetworkMessageMeta
        {
-            public byte Type;
-            public Guid Id;
+            public RequestId Id;
            public ushort BodyLength;
-
-            public NetworkMessageMeta(ReadOnlySpan<byte> metadata)
-            {
-                Type = metadata[0];
-                Id = new Guid(metadata[1..17]);
-                BodyLength = BitConverter.ToUInt16(metadata[17..]);
-            }
+            public byte Type;

            public NetworkMessage ToMessage(ReadOnlySpan<byte> memory)
            {
@@ -57,7 +57,7 @@ namespace mROA.Implementation
        {
            var request = new DefaultCallRequest
            {
-                Id = Guid.NewGuid(), CommandId = methodId, ObjectId = _identifier, Parameters = parameters
+                Id = RequestId.Generate(), CommandId = methodId, ObjectId = _identifier, Parameters = parameters
            };

            await _representationModule.PostCallMessageAsync(request.Id, EMessageType.CallRequest, request, _context);
@@ -102,7 +102,7 @@ namespace mROA.Implementation
        {
            var request = new DefaultCallRequest
            {
-                Id = Guid.NewGuid(), CommandId = methodId, ObjectId = _identifier, Parameters = parameters
+                Id = RequestId.Generate(), CommandId = methodId, ObjectId = _identifier, Parameters = parameters
            };
            await _representationModule.PostCallMessageAsync(request.Id, EMessageType.CallRequest, request, _context);

@@ -149,7 +149,7 @@ namespace mROA.Implementation
        {
            var request = new DefaultCallRequest
            {
-                Id = Guid.NewGuid(), CommandId = methodId, ObjectId = _identifier, Parameters = parameters
+                Id = RequestId.Generate(), CommandId = methodId, ObjectId = _identifier, Parameters = parameters
            };
            await _representationModule.PostCallMessageUntrustedAsync(request.Id, EMessageType.CallRequest, request,
                _context);
@@ -79,7 +79,7 @@ namespace mROA.Implementation
            }
        }

-        public async Task PostCallMessageAsync<T>(Guid id, EMessageType eMessageType, T payload,
+        public async Task PostCallMessageAsync<T>(RequestId id, EMessageType eMessageType, T payload,
            IEndPointContext? context) where T : notnull
        {
            var serialized = _serialization.Serialize(payload, context);
@@ -87,13 +87,13 @@ namespace mROA.Implementation
                { Id = id, MessageType = eMessageType, Data = serialized });
        }

-        public void PostCallMessage<T>(Guid id, EMessageType eMessageType, T payload, IEndPointContext? context)
+        public void PostCallMessage<T>(RequestId id, EMessageType eMessageType, T payload, IEndPointContext? context)
            where T : notnull
        {
            PostCallMessageAsync(id, eMessageType, payload, context).GetAwaiter().GetResult();
        }

-        public async Task PostCallMessageUntrustedAsync<T>(Guid id, EMessageType eMessageType, T payload,
+        public async Task PostCallMessageUntrustedAsync<T>(RequestId id, EMessageType eMessageType, T payload,
            IEndPointContext? context) where T : notnull
        {
            var serialized = _serialization.Serialize(payload, context);
@@ -5,9 +5,9 @@ namespace mROA.Implementation
    public struct RequestContext
    {
        public int OwnerId { get; }
-        public Guid RequestId { get; }
+        public RequestId RequestId { get; }

-        public RequestContext(Guid requestId, int ownerId)
+        public RequestContext(RequestId requestId, int ownerId)
        {
            RequestId = requestId;
            OwnerId = ownerId;
@@ -1,18 +1,64 @@
 using System;
+using System.Collections.Generic;
+using System.Runtime.InteropServices;

 namespace mROA.Implementation
 {
-    public struct RequestId
+    public struct RequestId : IEquatable<RequestId>
    {
+        private static Random _random = new();
+
        public ulong P0;
        public ulong P1;
-        public RequestId Generate()
+
+        public static RequestId Generate()
        {
-            var guid = Guid.NewGuid();
-            var bytes = guid.ToByteArray();
-            var high = BitConverter.ToUInt64(bytes, 0);
-            var low = BitConverter.ToUInt64(bytes, 8);
-            return new RequestId{ P0 = high, P1 = low};
+            var high = (ulong)(ushort)_random.Next() << 32 | (ulong)_random.Next();
+            var low = (ulong)(ushort)_random.Next() << 32 | (ulong)_random.Next();
+            return new RequestId { P0 = high, P1 = low };
+        }
+
+        public RequestId(byte[] bytes)
+        {
+            P0 = BitConverter.ToUInt64(bytes);
+            P1 = BitConverter.ToUInt64(bytes, 8);
+        }
+
+        public bool Equals(RequestId other)
+        {
+            return P0 == other.P0 && P1 == other.P1;
+        }
+
+        public override bool Equals(object? obj)
+        {
+            return obj is RequestId other && Equals(other);
+        }
+
+        public static bool operator ==(RequestId r1, RequestId r2)
+        {
+            return r1.Equals(r2);
+        }
+
+        public static bool operator !=(RequestId r1, RequestId r2)
+        {
+            return !(r1 == r2);
+        }
+
+        public override string ToString()
+        {
+            return $"{P0:X}{P1:X}";
+        }
+
+        public override int GetHashCode()
+        {
+            return HashCode.Combine(P0, P1);
+        }
+
+        public byte[] ToByteArray()
+        {
+            var array = new byte[16];
+            MemoryMarshal.Write(array, ref this);
+            return array;
        }
    }
 }