Merge pull request #3 from YaslePoy/header-remake

Header remake
This commit is contained in:
2025-07-27 23:47:18 +03:00
committed by GitHub
38 changed files with 573 additions and 305 deletions
+55 -45
View File
@@ -31,69 +31,79 @@ Console.WriteLine("End waiting");
var totalRequests = tasks.Sum(i => i.Result); var totalRequests = tasks.Sum(i => i.Result);
Console.WriteLine($"Total requests: {totalRequests:N0}"); Console.WriteLine($"Total requests: {totalRequests:N0}");
Console.WriteLine($"Results: {totalRequests / time.TotalSeconds:N} RPS"); Console.WriteLine($"Results: {totalRequests / time.TotalSeconds:N} RPS");
File.AppendAllText("results.txt", $"[SINGLE CBOR WRITER ALLOC] {totalRequests}\r\n"); File.AppendAllText("results.txt", $"[FAST ID] {totalRequests}\r\n");
async Task<List<ILoadTest>> GetLoadEndpoints(int count) async Task<List<ILoadTest>> GetLoadEndpoints(int count)
{ {
var loads = new List<ILoadTest>(); try
for (int i = 0; i < count; i++)
{ {
var builder = Host.CreateApplicationBuilder(new HostApplicationBuilderSettings { DisableDefaults = true }); var loads = new List<ILoadTest>();
builder.Services.AddSingleton<IContextualSerializationToolKit, CborSerializationToolkit>(); for (int i = 0; i < count; i++)
builder.Services.AddSingleton<IEndPointContext, EndPointContext>();
builder.Services.AddSingleton<IRealStoreInstanceRepository, InstanceRepository>(provider =>
{ {
var repo = new InstanceRepository(provider.GetService<IRepresentationModuleProducer>()); Console.WriteLine($"Initializing {i}");
repo.FillSingletons(typeof(Program).Assembly); var builder = Host.CreateApplicationBuilder(new HostApplicationBuilderSettings { DisableDefaults = true });
return repo; builder.Services.AddSingleton<IContextualSerializationToolKit, CborSerializationToolkit>();
}); builder.Services.AddSingleton<IEndPointContext, EndPointContext>();
builder.Services.AddSingleton<IRealStoreInstanceRepository, InstanceRepository>(provider =>
{
var repo = new InstanceRepository(provider.GetService<IRepresentationModuleProducer>());
repo.FillSingletons(typeof(Program).Assembly);
return repo;
});
builder.Services.AddSingleton<IInstanceRepository, RemoteInstanceRepository>(); builder.Services.AddSingleton<IInstanceRepository, RemoteInstanceRepository>();
builder.Services.AddSingleton<IChannelInteractionModule, ChannelInteractionModule>(); builder.Services.AddSingleton<IChannelInteractionModule, ChannelInteractionModule>();
// builder.Services.AddSingleton<IUntrustedInteractionModule, UdpUntrustedInteraction>(); // builder.Services.AddSingleton<IUntrustedInteractionModule, UdpUntrustedInteraction>();
builder.Services.AddSingleton<IRepresentationModule, RepresentationModule>(); builder.Services.AddSingleton<IRepresentationModule, RepresentationModule>();
var serverEndPoint = new IPEndPoint(IPAddress.Loopback, 4567); var serverEndPoint = new IPEndPoint(IPAddress.Loopback, 4567);
builder.Services.AddSingleton<IFrontendBridge, NetworkFrontendBridge>(); builder.Services.AddSingleton<IFrontendBridge, NetworkFrontendBridge>();
builder.Services.AddOptions(); builder.Services.AddOptions();
builder.Services.Configure<GatewayOptions>(options => options.Endpoint = serverEndPoint); builder.Services.Configure<GatewayOptions>(options => options.Endpoint = serverEndPoint);
builder.Services.AddSingleton<IRepresentationModuleProducer, StaticRepresentationModuleProducer>(); builder.Services.AddSingleton<IRepresentationModuleProducer, StaticRepresentationModuleProducer>();
// builder.Services.AddSingleton<IRequestExtractor, RequestExtractor>(); // builder.Services.AddSingleton<IRequestExtractor, RequestExtractor>();
builder.Services.AddSingleton<IExecuteModule, BasicExecutionModule>(); builder.Services.AddSingleton<IExecuteModule, BasicExecutionModule>();
builder.Services.AddSingleton<IMethodRepository, CollectableMethodRepository>(p => builder.Services.AddSingleton<IMethodRepository, CollectableMethodRepository>(p =>
{ {
var methodRepo = new CollectableMethodRepository(); var methodRepo = new CollectableMethodRepository();
methodRepo.AppendInvokers(new GeneratedInvokersCollection()); methodRepo.AppendInvokers(new GeneratedInvokersCollection());
return methodRepo; return methodRepo;
}); });
builder.Services.AddSingleton<ICallIndexProvider, GeneratedCallIndexProvider>(); builder.Services.AddSingleton<ICallIndexProvider, GeneratedCallIndexProvider>();
builder.Services.AddSingleton<ICancellationRepository, CancellationRepository>(); builder.Services.AddSingleton<ICancellationRepository, CancellationRepository>();
var app = builder.Build(); var app = builder.Build();
var frontendBridge = app.Services.GetService<IFrontendBridge>()!; Console.WriteLine($"Connecting {i}");
await frontendBridge.Connect(); var frontendBridge = app.Services.GetService<IFrontendBridge>()!;
// _ = app.Services.GetService<IRequestExtractor>()!.StartExtraction(); await frontendBridge.Connect();
// _ = app.Services.GetService<IUntrustedInteractionModule>().Start(serverEndPoint); // _ = app.Services.GetService<IRequestExtractor>()!.StartExtraction();
var context = app.Services.GetService<IInstanceRepository>(); // _ = 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);
}
var singletonObject = return loads;
context.GetSingletonObject<ILoadTest>( }
app.Services.GetService<IEndPointContext>()); catch (Exception e)
loads.Add(singletonObject); {
Console.WriteLine(e);
throw;
} }
return loads;
} }
async Task<int> Requests(CancellationToken token, int id, ILoadTest load) async Task<int> Requests(CancellationToken token, int id, ILoadTest load)
{ {
try try
{ {
int count = 0; int count = 0;
while (true){ while (true)
{
if (token.IsCancellationRequested) if (token.IsCancellationRequested)
{ {
break; break;
@@ -101,7 +111,7 @@ async Task<int> Requests(CancellationToken token, int id, ILoadTest load)
await load.Next(2); await load.Next(2);
count++; count++;
} }
Console.WriteLine(id); Console.WriteLine(id);
return count; return count;
+93
View File
@@ -0,0 +1,93 @@
using System.Formats.Cbor;
using BenchmarkDotNet.Attributes;
namespace mROA.Benchmark;
[MemoryDiagnoser]
public class CborTest
{
private CborWriter _writer;
private CborReader _reader;
private Memory<byte> FlatEncoded;
private Memory<byte> ArrayEncoded;
public CborTest()
{
_writer = new CborWriter(initialCapacity:512);
_writer.WriteStartArray(2);
_writer.WriteByteString([1, 2, 3, 4, 5, 6, 7, 8]);
_writer.WriteStartArray(2);
_writer.WriteInt32(12);
_writer.WriteTextString("tralala");
_writer.WriteEndArray();
_writer.WriteEndArray();
ArrayEncoded = _writer.Encode();
_writer.Reset();
_writer.WriteStartArray(3);
_writer.WriteByteString([1, 2, 3, 4, 5, 6, 7, 8]);
_writer.WriteInt32(12);
_writer.WriteTextString("tralala");
_writer.WriteEndArray();
FlatEncoded = _writer.Encode();
}
[Benchmark]
public int FlatWrite()
{
_writer.Reset();
_writer.WriteStartArray(3);
_writer.WriteByteString([1, 2, 3, 4, 5, 6, 7, 8]);
_writer.WriteInt32(12);
_writer.WriteTextString("tralala");
_writer.WriteEndArray();
var len = _writer.Encode(FlatEncoded.Span);
return len;
}
[Benchmark]
public int ArrayWrite()
{
_writer.Reset();
_writer.WriteStartArray(2);
_writer.WriteByteString([1, 2, 3, 4, 5, 6, 7, 8]);
_writer.WriteStartArray(2);
_writer.WriteInt32(12);
_writer.WriteTextString("tralala");
_writer.WriteEndArray();
_writer.WriteEndArray();
var len = _writer.Encode(ArrayEncoded.Span);
return len;
}
[Benchmark]
public int FlatRead()
{
var reader = new CborReader(FlatEncoded);
reader.ReadStartArray();
var arr = reader.ReadByteString();
var i = reader.ReadInt32();
var text = reader.ReadTextString();
reader.ReadEndArray();
return arr.Length;
}
[Benchmark]
public int ArrayRead()
{
var reader = new CborReader(ArrayEncoded);
reader.ReadStartArray();
var arr = reader.ReadByteString();
reader.ReadStartArray();
var i = reader.ReadInt32();
var text = reader.ReadTextString();
reader.ReadEndArray();
reader.ReadEndArray();
return arr.Length;
}
}
+34 -41
View File
@@ -1,51 +1,44 @@
using System.Collections.Generic; // See https://aka.ms/new-console-template for more information
using System.Linq;
using BenchmarkDotNet.Attributes;
namespace mROA.Benchmark using System.Formats.Cbor;
using System.Runtime.CompilerServices;
using System.Runtime.InteropServices;
using BenchmarkDotNet.Running;
using mROA.Benchmark;
using mROA.Implementation;
Console.WriteLine("Hello, World!");
var test = new CborTest();
test.ArrayWrite();
BenchmarkRunner.Run<CborTest>();
public static class CborExtensions
{ {
class Program public static unsafe void WriteToCbor(this RequestId id, CborWriter writer)
{ {
static void Main(string[] args) Span<byte> span = stackalloc byte[16];
{ MemoryMarshal.Write(span, ref id);
// Console.WriteLine("Hello, World!"); writer.WriteByteString(span);
// var summary = BenchmarkRunner.Run<CollectionsSpeed>();
}
} }
public class CollectionsSpeed [MethodImpl(MethodImplOptions.AggressiveInlining)]
public static unsafe void WriteToCborInline(this RequestId id, CborWriter writer)
{ {
private const int N = 1000; Span<byte> span = stackalloc byte[16];
MemoryMarshal.Write(span, ref id);
writer.WriteByteString(span);
}
private readonly List<int> _immutable; public static void WriteToDest(this RequestId id, Span<byte> destination)
private readonly int[] _array; {
MemoryMarshal.Write(destination, ref id);
public CollectionsSpeed() }
{
_array = Enumerable.Range(0, N).ToArray();
// _immutable = [.._array];
}
[Benchmark]
public int DefaultArray()
{
var sum = 0;
for (int i = 0; i < N; i++)
{
sum += _array[i];
}
return sum;
}
[Benchmark] [MethodImpl(MethodImplOptions.AggressiveOptimization)]
public int ImmutableArray() public static unsafe void WriteToCborOpt(this RequestId id, CborWriter writer)
{ {
var sum = 0; Span<byte> span = stackalloc byte[16];
for (int i = 0; i < N; i++) MemoryMarshal.Write(span, ref id);
{ writer.WriteByteString(span);
sum += _immutable[i];
}
return sum;
}
} }
} }
+57
View File
@@ -0,0 +1,57 @@
using System.Formats.Cbor;
using BenchmarkDotNet.Attributes;
using mROA.Implementation;
[MemoryDiagnoser]
public class RequestWriter
{
private const int N = 1000;
public RequestId Id = RequestId.Generate();
private CborWriter _writer;
public RequestWriter()
{
_writer = new CborWriter(initialCapacity: 512);
}
[Benchmark(Baseline = true)]
public int DefaultCbor()
{
_writer.Reset();
_writer.WriteByteString(Id.ToByteArray());
return _writer.BytesWritten;
}
[Benchmark]
public int DirectCbor()
{
_writer.Reset();
Id.WriteToCbor(_writer);
return _writer.BytesWritten;
}
[Benchmark]
public int Stackalloc()
{
_writer.Reset();
Span<byte> span = stackalloc byte[16];
Id.WriteToDest(span);
_writer.WriteByteString(span);
return _writer.BytesWritten;
}
[Benchmark]
public int DirectCborInline()
{
_writer.Reset();
Id.WriteToCborInline(_writer);
return _writer.BytesWritten;
}
[Benchmark]
public int DirectCborOpt()
{
_writer.Reset();
Id.WriteToCborOpt(_writer);
return _writer.BytesWritten;
}
}
+14 -3
View File
@@ -2,13 +2,24 @@
<PropertyGroup> <PropertyGroup>
<OutputType>Exe</OutputType> <OutputType>Exe</OutputType>
<TargetFramework>netstandard2.1</TargetFramework> <TargetFramework>net9.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable> <Nullable>enable</Nullable>
<AllowUnsafeBlocks>true</AllowUnsafeBlocks>
</PropertyGroup> </PropertyGroup>
<ItemGroup> <ItemGroup>
<PackageReference Include="BenchmarkDotNet" Version="0.14.0" /> <ProjectReference Include="..\mROA\mROA.csproj" />
</ItemGroup>
<ItemGroup>
<Reference Include="System.Formats.Cbor">
<HintPath>..\..\..\..\.nuget\packages\system.formats.cbor\9.0.7\lib\net9.0\System.Formats.Cbor.dll</HintPath>
</Reference>
</ItemGroup>
<ItemGroup>
<PackageReference Include="BenchmarkDotNet" Version="0.15.2" />
</ItemGroup> </ItemGroup>
</Project> </Project>
+19
View File
@@ -0,0 +1,19 @@
using System;
using System.Formats.Cbor;
using System.Runtime.CompilerServices;
using System.Runtime.InteropServices;
using mROA.Implementation;
namespace mROA.Cbor
{
public static class CborExtensions
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static unsafe void WriteToCborInline(this RequestId id, CborWriter writer)
{
Span<byte> span = stackalloc byte[16];
MemoryMarshal.Write(span, ref id);
writer.WriteByteString(span);
}
}
}
+11 -16
View File
@@ -18,7 +18,7 @@ namespace mROA.Cbor
private readonly IOrdinaryStructureParser[] _parsers = private readonly IOrdinaryStructureParser[] _parsers =
{ {
new NetworkMessageHeaderParser(), new DefaultCallRequestParser(), new FinalCommandExecutionParser(), new DefaultCallRequestParser(), new FinalCommandExecutionParser(),
new FinalCommandExecutionResultlessParser() new FinalCommandExecutionResultlessParser()
}; };
@@ -27,27 +27,21 @@ namespace mROA.Cbor
private bool FindParser(Type t, out IOrdinaryStructureParser parser) private bool FindParser(Type t, out IOrdinaryStructureParser parser)
{ {
if (t == typeof(NetworkMessageHeader)) if (t == typeof(DefaultCallRequest))
{ {
parser = _parsers[0]; parser = _parsers[0];
return true; return true;
} }
if (t == typeof(DefaultCallRequest)) if (t == typeof(FinalCommandExecution<object>))
{ {
parser = _parsers[1]; parser = _parsers[1];
return true; return true;
} }
if (t == typeof(FinalCommandExecution<object>))
{
parser = _parsers[2];
return true;
}
if (t == typeof(FinalCommandExecution)) if (t == typeof(FinalCommandExecution))
{ {
parser = _parsers[3]; parser = _parsers[2];
return true; return true;
} }
@@ -116,9 +110,9 @@ namespace mROA.Cbor
return preParsed.ToObject(type, context); 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); return Convert.ChangeType(nonCasted, type);
@@ -164,8 +158,9 @@ namespace mROA.Cbor
case DateTimeOffset dto: case DateTimeOffset dto:
writer.WriteDateTimeOffset(dto); writer.WriteDateTimeOffset(dto);
break; break;
case Guid g: case RequestId g:
writer.WriteByteString(g.ToByteArray()); // writer.WriteByteString(g.ToByteArray());
g.WriteToCborInline(writer);
break; break;
case byte[] bytes: case byte[] bytes:
writer.WriteByteString(bytes); writer.WriteByteString(bytes);
@@ -270,8 +265,8 @@ namespace mROA.Cbor
return reader.ReadUInt64(); return reader.ReadUInt64();
case CborReaderState.ByteString: case CborReaderState.ByteString:
if (type == typeof(Guid)) if (type == typeof(RequestId))
return new Guid(reader.ReadByteString()); return new RequestId(reader.ReadByteString());
return reader.ReadByteString(); return reader.ReadByteString();
case CborReaderState.TextString: case CborReaderState.TextString:
return reader.ReadTextString(); return reader.ReadTextString();
+9 -31
View File
@@ -12,38 +12,14 @@ namespace mROA.Cbor
object Read(CborReader reader, IEndPointContext context, CborSerializationToolkit serialization); object Read(CborReader reader, IEndPointContext context, CborSerializationToolkit serialization);
} }
public class NetworkMessageHeaderParser : IOrdinaryStructureParser
{
public void Write(CborWriter writer, object value, IEndPointContext context, CborSerializationToolkit serialization)
{
var v = value as NetworkMessageHeader;
writer.WriteStartArray(3);
writer.WriteByteString(v.Id.ToByteArray());
writer.WriteInt32((int)v.MessageType);
writer.WriteByteString(v.Data);
writer.WriteEndArray();
}
public object Read(CborReader reader, IEndPointContext context, CborSerializationToolkit serialization)
{
reader.ReadStartArray();
var value = new NetworkMessageHeader
{
Id = new Guid(reader.ReadByteString()),
MessageType = (EMessageType)reader.ReadInt32(),
Data = reader.ReadByteString()
};
return value;
}
}
public class DefaultCallRequestParser : IOrdinaryStructureParser public class DefaultCallRequestParser : IOrdinaryStructureParser
{ {
public void Write(CborWriter writer, object value, IEndPointContext context, CborSerializationToolkit serialization) public void Write(CborWriter writer, object value, IEndPointContext context, CborSerializationToolkit serialization)
{ {
var v = (DefaultCallRequest)value; var v = (DefaultCallRequest)value;
writer.WriteStartArray(4); writer.WriteStartArray(4);
writer.WriteByteString(v.Id.ToByteArray()); v.Id.WriteToCborInline(writer);
// writer.WriteByteString(v.Id.ToByteArray());
writer.WriteInt32(v.CommandId); writer.WriteInt32(v.CommandId);
writer.WriteStartArray(1); writer.WriteStartArray(1);
writer.WriteUInt64(v.ObjectId.Flat); writer.WriteUInt64(v.ObjectId.Flat);
@@ -57,7 +33,7 @@ namespace mROA.Cbor
reader.ReadStartArray(); reader.ReadStartArray();
var value = new DefaultCallRequest var value = new DefaultCallRequest
{ {
Id = new Guid(reader.ReadByteString()), Id = new RequestId(reader.ReadByteString()),
CommandId = reader.ReadInt32(), CommandId = reader.ReadInt32(),
ObjectId = (ComplexObjectIdentifier)ComplexObjectIdentifierParser.Instance.Read(reader, context, serialization), ObjectId = (ComplexObjectIdentifier)ComplexObjectIdentifierParser.Instance.Read(reader, context, serialization),
Parameters = serialization.ReadData(reader, typeof(object[]), context) as object[] Parameters = serialization.ReadData(reader, typeof(object[]), context) as object[]
@@ -92,7 +68,8 @@ namespace mROA.Cbor
{ {
var v = (FinalCommandExecution<object>)value; var v = (FinalCommandExecution<object>)value;
writer.WriteStartArray(2); writer.WriteStartArray(2);
writer.WriteByteString(v.Id.ToByteArray()); // writer.WriteByteString(v.Id.ToByteArray());
v.Id.WriteToCborInline(writer);
serialization.WriteData(v.Result, writer, context); serialization.WriteData(v.Result, writer, context);
writer.WriteEndArray(); writer.WriteEndArray();
} }
@@ -102,7 +79,7 @@ namespace mROA.Cbor
reader.ReadStartArray(); reader.ReadStartArray();
var result = new FinalCommandExecution<object> var result = new FinalCommandExecution<object>
{ {
Id = new Guid(reader.ReadByteString()), Id = new RequestId(reader.ReadByteString()),
Result = serialization.ReadData(reader, typeof(object), context), Result = serialization.ReadData(reader, typeof(object), context),
}; };
reader.ReadEndArray(); reader.ReadEndArray();
@@ -116,7 +93,8 @@ namespace mROA.Cbor
{ {
var v = (FinalCommandExecution)value; var v = (FinalCommandExecution)value;
writer.WriteStartArray(1); writer.WriteStartArray(1);
writer.WriteByteString(v.Id.ToByteArray()); // writer.WriteByteString(v.Id.ToByteArray());
v.Id.WriteToCborInline(writer);
writer.WriteEndArray(); writer.WriteEndArray();
} }
@@ -125,7 +103,7 @@ namespace mROA.Cbor
reader.ReadStartArray(); reader.ReadStartArray();
var result = new FinalCommandExecution var result = new FinalCommandExecution
{ {
Id = new Guid(reader.ReadByteString()) Id = new RequestId(reader.ReadByteString())
}; };
reader.ReadEndArray(); reader.ReadEndArray();
return result; return result;
+1
View File
@@ -7,6 +7,7 @@
<Version>2.0.7</Version> <Version>2.0.7</Version>
<LangVersion>9</LangVersion> <LangVersion>9</LangVersion>
<PackageIcon>mroaLogo.png</PackageIcon> <PackageIcon>mroaLogo.png</PackageIcon>
<AllowUnsafeBlocks>true</AllowUnsafeBlocks>
</PropertyGroup> </PropertyGroup>
<PropertyGroup Condition=" '$(Configuration)' == 'Debug' "> <PropertyGroup Condition=" '$(Configuration)' == 'Debug' ">
+1 -1
View File
@@ -31,7 +31,7 @@ namespace mROA.Codegen
Console.WriteLine("Sending event..."); Console.WriteLine("Sending event...");
var request = new DefaultCallRequest var request = new DefaultCallRequest
{ {
Id = Guid.NewGuid(), Id = RequestId.Generate(),
CommandId = <!L commandId>, CommandId = <!L commandId>,
ObjectId = new ComplexObjectIdentifier(index, ownerId), ObjectId = new ComplexObjectIdentifier(index, ownerId),
Parameters = new object[] { <!L transferParameters> } Parameters = new object[] { <!L transferParameters> }
+24
View File
@@ -20,4 +20,28 @@ public class Identifier
Assert.Fail(); 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));
}
} }
+6 -6
View File
@@ -17,8 +17,6 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Example.Shared", "Example.S
EndProject EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Example.Frontend", "Example.Frontend\Example.Frontend.csproj", "{9BD25A13-3165-47C0-9EAA-5C59EC490E32}" Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Example.Frontend", "Example.Frontend\Example.Frontend.csproj", "{9BD25A13-3165-47C0-9EAA-5C59EC490E32}"
EndProject EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "mROA.Benchmark", "mROA.Benchmark\mROA.Benchmark.csproj", "{6868F42B-E30D-4040-AD4A-BC2A2E76D03A}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "mROA.Cbor", "mROA.Cbor\mROA.Cbor.csproj", "{6211E4EA-13FD-4EB1-8E6A-C0173DD0784A}" Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "mROA.Cbor", "mROA.Cbor\mROA.Cbor.csproj", "{6211E4EA-13FD-4EB1-8E6A-C0173DD0784A}"
EndProject EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "TotalDemo", "TotalDemo", "{FC4FA752-10A7-4D78-A7C2-9BCD8A81FB5E}" Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "TotalDemo", "TotalDemo", "{FC4FA752-10A7-4D78-A7C2-9BCD8A81FB5E}"
@@ -31,6 +29,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Functionality.Shared", "Fun
EndProject EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Test", "Test", "{8C20901F-B416-4ABC-8AA4-9059646B081B}" Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Test", "Test", "{8C20901F-B416-4ABC-8AA4-9059646B081B}"
EndProject EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "mROA.Benchmark", "mROA.Benchmark\mROA.Benchmark.csproj", "{E021E8B3-56C2-400E-A05E-523CF7831189}"
EndProject
Global Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU Debug|Any CPU = Debug|Any CPU
@@ -61,10 +61,6 @@ Global
{9BD25A13-3165-47C0-9EAA-5C59EC490E32}.Debug|Any CPU.Build.0 = Debug|Any CPU {9BD25A13-3165-47C0-9EAA-5C59EC490E32}.Debug|Any CPU.Build.0 = Debug|Any CPU
{9BD25A13-3165-47C0-9EAA-5C59EC490E32}.Release|Any CPU.ActiveCfg = Release|Any CPU {9BD25A13-3165-47C0-9EAA-5C59EC490E32}.Release|Any CPU.ActiveCfg = Release|Any CPU
{9BD25A13-3165-47C0-9EAA-5C59EC490E32}.Release|Any CPU.Build.0 = Release|Any CPU {9BD25A13-3165-47C0-9EAA-5C59EC490E32}.Release|Any CPU.Build.0 = Release|Any CPU
{6868F42B-E30D-4040-AD4A-BC2A2E76D03A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{6868F42B-E30D-4040-AD4A-BC2A2E76D03A}.Debug|Any CPU.Build.0 = Debug|Any CPU
{6868F42B-E30D-4040-AD4A-BC2A2E76D03A}.Release|Any CPU.ActiveCfg = Release|Any CPU
{6868F42B-E30D-4040-AD4A-BC2A2E76D03A}.Release|Any CPU.Build.0 = Release|Any CPU
{6211E4EA-13FD-4EB1-8E6A-C0173DD0784A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU {6211E4EA-13FD-4EB1-8E6A-C0173DD0784A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{6211E4EA-13FD-4EB1-8E6A-C0173DD0784A}.Debug|Any CPU.Build.0 = Debug|Any CPU {6211E4EA-13FD-4EB1-8E6A-C0173DD0784A}.Debug|Any CPU.Build.0 = Debug|Any CPU
{6211E4EA-13FD-4EB1-8E6A-C0173DD0784A}.Release|Any CPU.ActiveCfg = Release|Any CPU {6211E4EA-13FD-4EB1-8E6A-C0173DD0784A}.Release|Any CPU.ActiveCfg = Release|Any CPU
@@ -81,6 +77,10 @@ Global
{D9D28596-E10C-4A98-A2AA-573219467506}.Debug|Any CPU.Build.0 = Debug|Any CPU {D9D28596-E10C-4A98-A2AA-573219467506}.Debug|Any CPU.Build.0 = Debug|Any CPU
{D9D28596-E10C-4A98-A2AA-573219467506}.Release|Any CPU.ActiveCfg = Release|Any CPU {D9D28596-E10C-4A98-A2AA-573219467506}.Release|Any CPU.ActiveCfg = Release|Any CPU
{D9D28596-E10C-4A98-A2AA-573219467506}.Release|Any CPU.Build.0 = Release|Any CPU {D9D28596-E10C-4A98-A2AA-573219467506}.Release|Any CPU.Build.0 = Release|Any CPU
{E021E8B3-56C2-400E-A05E-523CF7831189}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{E021E8B3-56C2-400E-A05E-523CF7831189}.Debug|Any CPU.Build.0 = Debug|Any CPU
{E021E8B3-56C2-400E-A05E-523CF7831189}.Release|Any CPU.ActiveCfg = Release|Any CPU
{E021E8B3-56C2-400E-A05E-523CF7831189}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection EndGlobalSection
GlobalSection(SolutionProperties) = preSolution GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE HideSolutionNode = FALSE
+4 -3
View File
@@ -1,12 +1,13 @@
using System; using System;
using System.Threading; using System.Threading;
using mROA.Implementation;
namespace mROA.Abstract namespace mROA.Abstract
{ {
public interface ICancellationRepository public interface ICancellationRepository
{ {
void RegisterCancellation(Guid id, CancellationTokenSource cts); void RegisterCancellation(RequestId id, CancellationTokenSource cts);
CancellationTokenSource? GetCancellation(Guid id); CancellationTokenSource? GetCancellation(RequestId id);
void FreeCancelation(Guid id); void FreeCancellation(RequestId id);
} }
} }
+6 -6
View File
@@ -9,13 +9,13 @@ namespace mROA.Abstract
{ {
int ConnectionId { get; set; } int ConnectionId { get; set; }
IEndPointContext Context { get; set; } IEndPointContext Context { get; set; }
Channel<NetworkMessageHeader> ReceiveChanel { get; } Channel<NetworkMessage> ReceiveChanel { get; }
ChannelReader<NetworkMessageHeader> TrustedPostChanel { get; } ChannelReader<NetworkMessage> TrustedPostChanel { get; }
ChannelReader<NetworkMessageHeader> UntrustedPostChanel { get; } ChannelReader<NetworkMessage> UntrustedPostChanel { get; }
Func<bool> IsConnected { get; set; } Func<bool> IsConnected { get; set; }
ValueTask<NetworkMessageHeader> GetNextMessageReceiving(); ValueTask<NetworkMessage> GetNextMessageReceiving();
Task PostMessageAsync(NetworkMessageHeader messageHeader); Task PostMessageAsync(NetworkMessage message);
Task PostMessageUntrustedAsync(NetworkMessageHeader messageHeader); Task PostMessageUntrustedAsync(NetworkMessage message);
event Action<int> OnDisconnected; event Action<int> OnDisconnected;
Task Restart(bool sendRecovery); Task Restart(bool sendRecovery);
void PassReconnection(); void PassReconnection();
+1 -1
View File
@@ -5,6 +5,6 @@ namespace mROA.Abstract
{ {
public interface ICommandExecution : INetworkMessage public interface ICommandExecution : INetworkMessage
{ {
Guid Id { get; set; } RequestId Id { get; set; }
} }
} }
+7 -7
View File
@@ -11,21 +11,21 @@ namespace mROA.Abstract
int Id { get; } int Id { get; }
IEndPointContext Context { get; } IEndPointContext Context { get; }
Task<(object? Deserialized, EMessageType MessageType)> GetSingle(Predicate<NetworkMessageHeader> rule, Task<(object? Deserialized, EMessageType MessageType)> GetSingle(Predicate<NetworkMessage> rule,
IEndPointContext? context, CancellationToken token = default, IEndPointContext? context, CancellationToken token = default,
params Func<NetworkMessageHeader, Type?>[] converter); params Func<NetworkMessage, Type?>[] converter);
IAsyncEnumerable<(object parced, EMessageType originalType)> GetStream(Predicate<NetworkMessageHeader> rule, IAsyncEnumerable<(object parced, EMessageType originalType)> GetStream(Predicate<NetworkMessage> rule,
IEndPointContext? context, CancellationToken token = default, IEndPointContext? context, CancellationToken token = default,
params Func<NetworkMessageHeader, Type?>[] converter); 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; 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; 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; where T : notnull;
} }
} }
+2 -2
View File
@@ -8,7 +8,7 @@ namespace mROA.Abstract
{ {
Task StartExtraction(); Task StartExtraction();
void PushMessage(object parced, EMessageType originalType); void PushMessage(object parced, EMessageType originalType);
Predicate<NetworkMessageHeader> Rule { get; } Predicate<NetworkMessage> Rule { get; }
Func<NetworkMessageHeader, Type?>[] Converters { get; } Func<NetworkMessage, Type?>[] Converters { get; }
} }
} }
@@ -114,7 +114,7 @@ namespace mROA.Implementation.Backend
if (cts == null) if (cts == null)
throw new NullReferenceException("Can't find cancellation for this request"); throw new NullReferenceException("Can't find cancellation for this request");
cts.Cancel(); cts.Cancel();
_cancellationRepo.FreeCancelation(command.Id); _cancellationRepo.FreeCancellation(command.Id);
return new FinalCommandExecution return new FinalCommandExecution
{ {
@@ -164,11 +164,11 @@ namespace mROA.Implementation.Backend
{ {
Id = command.Id Id = command.Id
}; };
_cancellationRepo.FreeCancelation(command.Id); _cancellationRepo.FreeCancellation(command.Id);
if (invoker.IsTrusted) if (invoker.IsTrusted)
representationModule.PostCallMessage(command.Id, EMessageType.FinishedCommandExecution, representationModule.PostCallMessageAsync(command.Id, EMessageType.FinishedCommandExecution,
payload, context); payload, context);
}); });
@@ -195,9 +195,9 @@ namespace mROA.Implementation.Backend
Id = command.Id, Id = command.Id,
Result = finalResult Result = finalResult
}; };
_cancellationRepo.FreeCancelation(command.Id); _cancellationRepo.FreeCancellation(command.Id);
representationModule.PostCallMessage(command.Id, EMessageType.FinishedCommandExecution, representationModule.PostCallMessageAsync(command.Id, EMessageType.FinishedCommandExecution,
payload, context); payload, context);
}); });
@@ -1,5 +1,6 @@
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Collections.Specialized;
using mROA.Abstract; using mROA.Abstract;
namespace mROA.Implementation.Backend namespace mROA.Implementation.Backend
@@ -56,7 +56,7 @@ namespace mROA.Implementation.Backend
while (true) while (true)
{ {
var client = await _tcpListener.AcceptTcpClientAsync(); var client = await _tcpListener.AcceptTcpClientAsync();
_ = HandleConnection(client).ConfigureAwait(false); _ = HandleConnection(client);
} }
} }
@@ -70,7 +70,7 @@ namespace mROA.Implementation.Backend
CallIndexProvider = _callIndexProvider CallIndexProvider = _callIndexProvider
}; };
var streamExtractor = var streamExtractor =
new ChannelInteractionModule.StreamExtractor(client.GetStream(), _serialization, context); new ChannelInteractionModule.StreamExtractor(client.GetStream());
interaction.IsConnected = () => streamExtractor.IsConnected; interaction.IsConnected = () => streamExtractor.IsConnected;
streamExtractor.MessageReceived = async message => streamExtractor.MessageReceived = async message =>
{ {
@@ -98,14 +98,14 @@ namespace mROA.Implementation.Backend
} }
private void HandleNewClient(EndPointContext context, ChannelInteractionModule interaction, private void HandleNewClient(EndPointContext context, ChannelInteractionModule interaction,
ChannelInteractionModule.StreamExtractor streamExtractor, CancellationTokenSource cts, NetworkMessageHeader connectionHeader) ChannelInteractionModule.StreamExtractor streamExtractor, CancellationTokenSource cts, NetworkMessage connection)
{ {
context.HostId = 0; context.HostId = 0;
context.OwnerId = -interaction.ConnectionId; context.OwnerId = -interaction.ConnectionId;
interaction.Context = context; interaction.Context = context;
Task.Run(async () => await streamExtractor.LoopedReceive(cts.Token)); Task.Run(async () => await streamExtractor.LoopedReceive(cts.Token));
_ = streamExtractor.SendFromChannel(interaction.TrustedPostChanel, cts.Token); _ = streamExtractor.SendFromChannel(interaction.TrustedPostChanel, cts.Token);
interaction.PostMessageAsync(new NetworkMessageHeader(_serialization, interaction.PostMessageAsync(new NetworkMessage(_serialization,
new IdAssignment { Id = interaction.ConnectionId }, null)); new IdAssignment { Id = interaction.ConnectionId }, null));
_extractorsTokenSources[interaction.ConnectionId] = cts; _extractorsTokenSources[interaction.ConnectionId] = cts;
@@ -143,7 +143,7 @@ namespace mROA.Implementation.Backend
}; };
} }
private void RecoverDisconnectedClient(NetworkMessageHeader connectionRequest, private void RecoverDisconnectedClient(NetworkMessage connectionRequest,
ChannelInteractionModule.StreamExtractor streamExtractor, ChannelInteractionModule.StreamExtractor streamExtractor,
CancellationTokenSource cts) CancellationTokenSource cts)
{ {
+1 -1
View File
@@ -41,7 +41,7 @@ namespace mROA.Implementation.Backend
while (token.IsCancellationRequested == false) while (token.IsCancellationRequested == false)
{ {
var incoming = await _client.ReceiveAsync(); var incoming = await _client.ReceiveAsync();
var parsed = _serializationToolkit.Deserialize<NetworkMessageHeader>(incoming.Buffer, null); var parsed = _serializationToolkit.Deserialize<NetworkMessage>(incoming.Buffer, null);
try try
{ {
int channelId; int channelId;
+3 -3
View File
@@ -4,7 +4,7 @@ namespace mROA.Implementation
{ {
public interface ICallRequest public interface ICallRequest
{ {
Guid Id { get; } RequestId Id { get; }
int CommandId { get; } int CommandId { get; }
ComplexObjectIdentifier ObjectId { get; } ComplexObjectIdentifier ObjectId { get; }
object?[]? Parameters { get; } object?[]? Parameters { get; }
@@ -12,7 +12,7 @@ namespace mROA.Implementation
public struct DefaultCallRequest : ICallRequest public struct DefaultCallRequest : ICallRequest
{ {
public Guid Id { get; set; } public RequestId Id { get; set; }
public int CommandId { get; set; } public int CommandId { get; set; }
public ComplexObjectIdentifier ObjectId { get; set; } public ComplexObjectIdentifier ObjectId { get; set; }
@@ -26,7 +26,7 @@ namespace mROA.Implementation
public class CancelRequest : ICallRequest public class CancelRequest : ICallRequest
{ {
public Guid Id { get; set; } public RequestId Id { get; set; }
public int CommandId { get; set; } = -2; public int CommandId { get; set; } = -2;
public ComplexObjectIdentifier ObjectId { get; set; } = ComplexObjectIdentifier.Null; public ComplexObjectIdentifier ObjectId { get; set; } = ComplexObjectIdentifier.Null;
public object?[]? Parameters { get; set; } = null; public object?[]? Parameters { get; set; } = null;
@@ -8,19 +8,19 @@ namespace mROA.Implementation
{ {
public class CancellationRepository : ICancellationRepository 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); _cancellations.TryAdd(id, cts);
} }
public CancellationTokenSource? GetCancellation(Guid id) public CancellationTokenSource? GetCancellation(RequestId id)
{ {
return _cancellations.GetValueOrDefault(id); return _cancellations.GetValueOrDefault(id);
} }
public void FreeCancelation(Guid id) public void FreeCancellation(RequestId id)
{ {
_cancellations.Remove(id, out _); _cancellations.Remove(id, out _);
} }
+44 -52
View File
@@ -1,5 +1,6 @@
using System; using System;
using System.IO; using System.IO;
using System.Runtime.InteropServices;
using System.Threading; using System.Threading;
using System.Threading.Channels; using System.Threading.Channels;
using System.Threading.Tasks; using System.Threading.Tasks;
@@ -9,11 +10,11 @@ namespace mROA.Implementation
{ {
public class ChannelInteractionModule : IChannelInteractionModule public class ChannelInteractionModule : IChannelInteractionModule
{ {
private readonly ChannelReader<NetworkMessageHeader> _receiveReader; private readonly ChannelReader<NetworkMessage> _receiveReader;
private readonly ChannelWriter<NetworkMessageHeader> _trustedWriter; private readonly ChannelWriter<NetworkMessage> _trustedWriter;
private readonly ChannelWriter<NetworkMessageHeader> _untrustedWriter; private readonly ChannelWriter<NetworkMessage> _untrustedWriter;
private readonly Channel<NetworkMessageHeader> _outputTrustedChannel; private readonly Channel<NetworkMessage> _outputTrustedChannel;
private readonly Channel<NetworkMessageHeader> _outputUntrustedChannel; private readonly Channel<NetworkMessage> _outputUntrustedChannel;
private readonly IContextualSerializationToolKit _serialization; private readonly IContextualSerializationToolKit _serialization;
private bool _isConnected = true; private bool _isConnected = true;
private bool _isActive = true; private bool _isActive = true;
@@ -28,19 +29,19 @@ namespace mROA.Implementation
public ChannelInteractionModule(IContextualSerializationToolKit serialization) public ChannelInteractionModule(IContextualSerializationToolKit serialization)
{ {
_serialization = serialization; _serialization = serialization;
ReceiveChanel = Channel.CreateUnbounded<NetworkMessageHeader>(new UnboundedChannelOptions ReceiveChanel = Channel.CreateUnbounded<NetworkMessage>(new UnboundedChannelOptions
{ {
SingleReader = false, SingleReader = false,
SingleWriter = false, SingleWriter = false,
}); });
_receiveReader = ReceiveChanel.Reader; _receiveReader = ReceiveChanel.Reader;
_outputTrustedChannel = Channel.CreateUnbounded<NetworkMessageHeader>(new UnboundedChannelOptions _outputTrustedChannel = Channel.CreateUnbounded<NetworkMessage>(new UnboundedChannelOptions
{ {
SingleReader = true, SingleReader = true,
SingleWriter = true SingleWriter = true
}); });
_trustedWriter = _outputTrustedChannel.Writer; _trustedWriter = _outputTrustedChannel.Writer;
_outputUntrustedChannel = Channel.CreateUnbounded<NetworkMessageHeader>(new UnboundedChannelOptions _outputUntrustedChannel = Channel.CreateUnbounded<NetworkMessage>(new UnboundedChannelOptions
{ {
SingleReader = true, SingleReader = true,
SingleWriter = true, SingleWriter = true,
@@ -52,33 +53,33 @@ namespace mROA.Implementation
public int ConnectionId { get; set; } public int ConnectionId { get; set; }
public IEndPointContext Context { get; set; } public IEndPointContext Context { get; set; }
public Channel<NetworkMessageHeader> ReceiveChanel { get; } public Channel<NetworkMessage> ReceiveChanel { get; }
public ChannelReader<NetworkMessageHeader> TrustedPostChanel => _outputTrustedChannel.Reader; public ChannelReader<NetworkMessage> TrustedPostChanel => _outputTrustedChannel.Reader;
public ChannelReader<NetworkMessageHeader> UntrustedPostChanel => _outputUntrustedChannel.Reader; public ChannelReader<NetworkMessage> UntrustedPostChanel => _outputUntrustedChannel.Reader;
public Func<bool> IsConnected { get; set; } = () => false; public Func<bool> IsConnected { get; set; } = () => false;
public ValueTask<NetworkMessageHeader> GetNextMessageReceiving() public ValueTask<NetworkMessage> GetNextMessageReceiving()
{ {
return _receiveReader.ReadAsync(); return _receiveReader.ReadAsync();
} }
private async ValueTask<bool> PostMessageInternal(NetworkMessageHeader messageHeader) private async ValueTask<bool> PostMessageInternal(NetworkMessage message)
{ {
if (!IsConnected()) if (!IsConnected())
{ {
return false; return false;
} }
await _trustedWriter.WriteAsync(messageHeader); await _trustedWriter.WriteAsync(message);
return true; return true;
} }
public async Task PostMessageAsync(NetworkMessageHeader messageHeader) public async Task PostMessageAsync(NetworkMessage message)
{ {
while (true) while (true)
{ {
if (await PostMessageInternal(messageHeader)) if (await PostMessageInternal(message))
break; break;
if (!_isActive) if (!_isActive)
@@ -91,9 +92,9 @@ namespace mROA.Implementation
} }
} }
public async Task PostMessageUntrustedAsync(NetworkMessageHeader messageHeader) public async Task PostMessageUntrustedAsync(NetworkMessage message)
{ {
await _untrustedWriter.WriteAsync(messageHeader); await _untrustedWriter.WriteAsync(message);
} }
public event Action<int>? OnDisconnected; public event Action<int>? OnDisconnected;
@@ -103,12 +104,12 @@ namespace mROA.Implementation
if (sendRecovery) if (sendRecovery)
{ {
await PostMessageAsync( await PostMessageAsync(
new NetworkMessageHeader(_serialization, new ClientRecovery(Math.Abs(ConnectionId)), Context)); new NetworkMessage(_serialization, new ClientRecovery(Math.Abs(ConnectionId)), Context));
await ReceiveChanel.Reader.ReadAsync(); await ReceiveChanel.Reader.ReadAsync();
} }
else else
{ {
await _trustedWriter.WriteAsync(new NetworkMessageHeader()); await _trustedWriter.WriteAsync(new NetworkMessage());
} }
PassReconnection(); PassReconnection();
@@ -141,42 +142,34 @@ namespace mROA.Implementation
public class StreamExtractor public class StreamExtractor
{ {
private const int BufferSize = ushort.MaxValue; private const int BufferSize = ushort.MaxValue + 19;
private readonly Stream _ioStream; private readonly Stream _ioStream;
private readonly IContextualSerializationToolKit _serializationToolkit;
private readonly Memory<byte> _buffer = new byte[BufferSize]; private readonly Memory<byte> _buffer = new byte[BufferSize];
private readonly IEndPointContext _context;
private readonly byte[] _lenBuffer;
public StreamExtractor(Stream ioStream, IContextualSerializationToolKit serializationToolkit, public StreamExtractor(Stream ioStream)
IEndPointContext context)
{ {
_ioStream = ioStream; _ioStream = ioStream;
_serializationToolkit = serializationToolkit;
_context = context;
_lenBuffer = new byte[2];
}
public Action<NetworkMessageHeader> MessageReceived = _ => { };
private async Task<ushort> ReadMessageLength()
{
await _ioStream.ReadAsync(_lenBuffer);
var len = BitConverter.ToUInt16(_lenBuffer);
return len;
} }
public Action<NetworkMessage> MessageReceived = _ => { };
public async Task SingleReceive(CancellationToken token = default) public async Task SingleReceive(CancellationToken token = default)
{ {
var len = await ReadMessageLength(); var firstRead = await _ioStream.ReadAsync(_buffer, token);
var localSpan = _buffer[..len];
var meta = MemoryMarshal.Read<NetworkMessage.NetworkMessageMeta>(_buffer.Span);
var len = meta.BodyLength;
var readLen = firstRead - 19;
if (readLen != len)
{
var lastPart = _buffer[firstRead..(len + 19)];
await _ioStream.ReadExactlyAsync(lastPart, cancellationToken: token);
}
await _ioStream.ReadExactlyAsync(localSpan, cancellationToken: token); var message = meta.ToMessage(_buffer.Span);
var message = _serializationToolkit.Deserialize<NetworkMessageHeader>(localSpan, _context);
// _logger.LogTrace("RECV {0}", message.ToString());
MessageReceived(message); MessageReceived(message);
} }
@@ -188,18 +181,17 @@ namespace mROA.Implementation
} }
} }
private async Task Send(NetworkMessageHeader message, CancellationToken token = default) private async Task Send(NetworkMessage message, CancellationToken token = default)
{ {
var bodySpan = _buffer[2..]; var meta = message.ToMeta();
var len = _serializationToolkit.Serialize(message, bodySpan.Span, _context); MemoryMarshal.Write(_buffer.Span, ref meta);
var header = BitConverter.GetBytes((ushort)len); message.Data.CopyTo(_buffer.Span[19..]);
header.CopyTo(_buffer); var sendingSpan = _buffer[..(19 + meta.BodyLength)];
var sendingSpan = _buffer[..(len + 2)];
await _ioStream.WriteAsync(sendingSpan, token); await _ioStream.WriteAsync(sendingSpan, token);
// _logger.LogTrace("SEND {0}", message.ToString()); // _logger.LogTrace("SEND {0}", message.ToString());
} }
public async Task SendFromChannel(ChannelReader<NetworkMessageHeader> channel, public async Task SendFromChannel(ChannelReader<NetworkMessage> channel,
CancellationToken token = default) CancellationToken token = default)
{ {
while (token.IsCancellationRequested == false && IsConnected) while (token.IsCancellationRequested == false && IsConnected)
@@ -5,7 +5,7 @@ namespace mROA.Implementation.CommandExecution
{ {
public class AsyncCommandExecution : ICommandExecution public class AsyncCommandExecution : ICommandExecution
{ {
public Guid Id { get; set; } public RequestId Id { get; set; }
public EMessageType MessageType => EMessageType.Unknown; public EMessageType MessageType => EMessageType.Unknown;
} }
} }
@@ -6,7 +6,7 @@ namespace mROA.Implementation.CommandExecution
{ {
public class ExceptionCommandExecution : ICommandExecution public class ExceptionCommandExecution : ICommandExecution
{ {
public Guid Id { get; set; } public RequestId Id { get; set; }
public EMessageType MessageType => EMessageType.ExceptionCommandExecution; public EMessageType MessageType => EMessageType.ExceptionCommandExecution;
public string Exception { get; set; } public string Exception { get; set; }
@@ -5,13 +5,13 @@ namespace mROA.Implementation.CommandExecution
{ {
public struct FinalCommandExecution : ICommandExecution public struct FinalCommandExecution : ICommandExecution
{ {
public Guid Id { get; set; } public RequestId Id { get; set; }
public EMessageType MessageType => EMessageType.FinishedCommandExecution; public EMessageType MessageType => EMessageType.FinishedCommandExecution;
} }
public struct FinalCommandExecution<T> : ICommandExecution public struct FinalCommandExecution<T> : ICommandExecution
{ {
public Guid Id { get; set; } public RequestId Id { get; set; }
public EMessageType MessageType => EMessageType.FinishedCommandExecution; public EMessageType MessageType => EMessageType.FinishedCommandExecution;
public T? Result { get; set; } public T? Result { get; set; }
} }
+1 -1
View File
@@ -1,6 +1,6 @@
namespace mROA.Implementation namespace mROA.Implementation
{ {
public enum EMessageType public enum EMessageType : byte
{ {
Unknown, Unknown,
FinishedCommandExecution, FinishedCommandExecution,
@@ -28,7 +28,7 @@ namespace mROA.Implementation.Frontend
_serialization = serialization; _serialization = serialization;
_interactionModule = interactionModule; _interactionModule = interactionModule;
_rawExtractorCancellation = new CancellationTokenSource(); _rawExtractorCancellation = new CancellationTokenSource();
_currentExtractor = new ChannelInteractionModule.StreamExtractor(Stream.Null, _serialization, context); _currentExtractor = new ChannelInteractionModule.StreamExtractor(Stream.Null);
} }
public async Task Connect() public async Task Connect()
@@ -39,7 +39,7 @@ namespace mROA.Implementation.Frontend
_interactionModule.IsConnected = () => _currentExtractor.IsConnected; _interactionModule.IsConnected = () => _currentExtractor.IsConnected;
_interactionModule.OnDisconnected += _ => { Reconnect().ConfigureAwait(false); }; _interactionModule.OnDisconnected += _ => { Reconnect().ConfigureAwait(false); };
_interactionModule.PostMessageAsync(new NetworkMessageHeader(_serialization, new ClientConnect(), _context)) _interactionModule.PostMessageAsync(new NetworkMessage(_serialization, new ClientConnect(), _context))
.Wait(); .Wait();
_ = _currentExtractor.SingleReceive().ConfigureAwait(false); _ = _currentExtractor.SingleReceive().ConfigureAwait(false);
@@ -63,7 +63,7 @@ namespace mROA.Implementation.Frontend
private void PrepareExtractor() private void PrepareExtractor()
{ {
_currentExtractor = _currentExtractor =
new ChannelInteractionModule.StreamExtractor(_tcpClient.GetStream(), _serialization, _context); new ChannelInteractionModule.StreamExtractor(_tcpClient.GetStream());
_ = _currentExtractor.SendFromChannel(_interactionModule.TrustedPostChanel, _ = _currentExtractor.SendFromChannel(_interactionModule.TrustedPostChanel,
_rawExtractorCancellation.Token); _rawExtractorCancellation.Token);
@@ -95,7 +95,7 @@ namespace mROA.Implementation.Frontend
public void Disconnect() public void Disconnect()
{ {
_ = _interactionModule.PostMessageAsync(new NetworkMessageHeader(_serialization, new ClientDisconnect(), _ = _interactionModule.PostMessageAsync(new NetworkMessage(_serialization, new ClientDisconnect(),
_context)); _context));
_interactionModule.Dispose(); _interactionModule.Dispose();
_tcpClient.Dispose(); _tcpClient.Dispose();
@@ -4,7 +4,7 @@ namespace mROA.Implementation.Frontend
{ {
public class RemoteException : Exception public class RemoteException : Exception
{ {
public Guid CallRequestId; public RequestId CallRequestId;
private readonly string _error; private readonly string _error;
public RemoteException(string error) public RemoteException(string error)
@@ -56,11 +56,11 @@ namespace mROA.Implementation.Frontend
} }
} }
public Predicate<NetworkMessageHeader> Rule { get; } = m => public Predicate<NetworkMessage> Rule { get; } = m =>
m.MessageType is EMessageType.CallRequest or EMessageType.CancelRequest m.MessageType is EMessageType.CallRequest or EMessageType.CancelRequest
or EMessageType.EventRequest or EMessageType.ClientDisconnect; or EMessageType.EventRequest or EMessageType.ClientDisconnect;
public Func<NetworkMessageHeader, Type?>[] Converters { get; } = public Func<NetworkMessage, Type?>[] Converters { get; } =
{ {
m => m.MessageType == EMessageType.CallRequest ? typeof(DefaultCallRequest) : null, m => m.MessageType == EMessageType.CallRequest ? typeof(DefaultCallRequest) : null,
m => m.MessageType == EMessageType.CancelRequest ? typeof(CancelRequest) : null, m => m.MessageType == EMessageType.CancelRequest ? typeof(CancelRequest) : null,
@@ -44,7 +44,7 @@ namespace mROA.Implementation.Frontend
while (token.IsCancellationRequested == false) while (token.IsCancellationRequested == false)
{ {
var message = new Memory<byte>((await udpClient.ReceiveAsync()).Buffer); var message = new Memory<byte>((await udpClient.ReceiveAsync()).Buffer);
var parsed = _serializationToolkit.Deserialize<NetworkMessageHeader>(message, _context); var parsed = _serializationToolkit.Deserialize<NetworkMessage>(message, _context);
await writer.WriteAsync(parsed, token); await writer.WriteAsync(parsed, token);
} }
@@ -52,9 +52,9 @@ namespace mROA.Implementation.Frontend
private async Task Posting(UdpClient udpClient, CancellationToken token) private async Task Posting(UdpClient udpClient, CancellationToken token)
{ {
var initMessage = new NetworkMessageHeader var initMessage = new NetworkMessage
{ {
MessageType = EMessageType.UntrustedConnect, Id = Guid.NewGuid(), MessageType = EMessageType.UntrustedConnect, Id = RequestId.Generate(),
Data = BitConverter.GetBytes(_channelInteractionModule.ConnectionId) Data = BitConverter.GetBytes(_channelInteractionModule.ConnectionId)
}; };
+73
View File
@@ -0,0 +1,73 @@
using System;
using mROA.Abstract;
namespace mROA.Implementation
{
public class NetworkMessage
{
private bool Equals(NetworkMessage other)
{
return Id.Equals(other.Id) && MessageType == other.MessageType;
}
public override bool Equals(object? obj)
{
if (obj is null) return false;
if (ReferenceEquals(this, obj)) return true;
return obj.GetType() == GetType() && Equals((NetworkMessage)obj);
}
public NetworkMessage()
{
Data = Array.Empty<byte>();
}
public NetworkMessage(IContextualSerializationToolKit serializationToolkit,
INetworkMessage networkMessage, IEndPointContext? context)
{
MessageType = networkMessage.MessageType;
Data = serializationToolkit.Serialize(networkMessage, context);
Id = RequestId.Generate();
}
public RequestId Id { get; set; }
public EMessageType MessageType { get; set; }
public byte[] Data { get; set; }
public object Serialized { get; set; }
public IEndPointContext Context { get; set; }
public override string ToString()
{
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 RequestId Id;
public ushort BodyLength;
public byte Type;
public NetworkMessage ToMessage(Span<byte> memory)
{
var data = memory[19..][..BodyLength];
return new NetworkMessage
{
Data = data.ToArray(),
Id = Id,
MessageType = (EMessageType)Type
};
}
}
}
}
@@ -1,44 +0,0 @@
using System;
using mROA.Abstract;
namespace mROA.Implementation
{
public class NetworkMessageHeader
{
private bool Equals(NetworkMessageHeader other)
{
return Id.Equals(other.Id) && MessageType == other.MessageType;
}
public override bool Equals(object? obj)
{
if (obj is null) return false;
if (ReferenceEquals(this, obj)) return true;
return obj.GetType() == GetType() && Equals((NetworkMessageHeader)obj);
}
public NetworkMessageHeader()
{
Data = Array.Empty<byte>();
}
public NetworkMessageHeader(IContextualSerializationToolKit serializationToolkit,
INetworkMessage networkMessage, IEndPointContext? context)
{
MessageType = networkMessage.MessageType;
Data = serializationToolkit.Serialize(networkMessage, context);
Id = Guid.NewGuid();
}
public Guid Id { get; set; }
public EMessageType MessageType { get; set; }
public byte[] Data { get; set; }
public override string ToString()
{
return $" {Id}:{MessageType} [{Data.Length}]";
}
}
}
+3 -3
View File
@@ -57,7 +57,7 @@ namespace mROA.Implementation
{ {
var request = new DefaultCallRequest 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); await _representationModule.PostCallMessageAsync(request.Id, EMessageType.CallRequest, request, _context);
@@ -102,7 +102,7 @@ namespace mROA.Implementation
{ {
var request = new DefaultCallRequest 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); await _representationModule.PostCallMessageAsync(request.Id, EMessageType.CallRequest, request, _context);
@@ -149,7 +149,7 @@ namespace mROA.Implementation
{ {
var request = new DefaultCallRequest 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, await _representationModule.PostCallMessageUntrustedAsync(request.Id, EMessageType.CallRequest, request,
_context); _context);
+9 -9
View File
@@ -28,8 +28,8 @@ namespace mROA.Implementation
public IEndPointContext Context => _interaction.Context; public IEndPointContext Context => _interaction.Context;
public async Task<(object? Deserialized, EMessageType MessageType)> GetSingle( public async Task<(object? Deserialized, EMessageType MessageType)> GetSingle(
Predicate<NetworkMessageHeader> rule, IEndPointContext? context, Predicate<NetworkMessage> rule, IEndPointContext? context,
CancellationToken token = default, params Func<NetworkMessageHeader, Type?>[] converter) CancellationToken token = default, params Func<NetworkMessage, Type?>[] converter)
{ {
var writer = _interaction.ReceiveChanel.Writer; var writer = _interaction.ReceiveChanel.Writer;
var reader = _interaction.ReceiveChanel.Reader; var reader = _interaction.ReceiveChanel.Reader;
@@ -52,9 +52,9 @@ namespace mROA.Implementation
} }
public async IAsyncEnumerable<(object parced, EMessageType originalType)> GetStream( public async IAsyncEnumerable<(object parced, EMessageType originalType)> GetStream(
Predicate<NetworkMessageHeader> rule, IEndPointContext? context, Predicate<NetworkMessage> rule, IEndPointContext? context,
[EnumeratorCancellation] CancellationToken token = default, [EnumeratorCancellation] CancellationToken token = default,
params Func<NetworkMessageHeader, Type?>[] converter) params Func<NetworkMessage, Type?>[] converter)
{ {
var writer = _interaction.ReceiveChanel.Writer; var writer = _interaction.ReceiveChanel.Writer;
await foreach (var message in _interaction.ReceiveChanel.Reader.ReadAllAsync(token)) await foreach (var message in _interaction.ReceiveChanel.Reader.ReadAllAsync(token))
@@ -79,25 +79,25 @@ 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 IEndPointContext? context) where T : notnull
{ {
var serialized = _serialization.Serialize(payload, context); var serialized = _serialization.Serialize(payload, context);
await _interaction.PostMessageAsync(new NetworkMessageHeader await _interaction.PostMessageAsync(new NetworkMessage
{ Id = id, MessageType = eMessageType, Data = serialized }); { 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 where T : notnull
{ {
PostCallMessageAsync(id, eMessageType, payload, context).GetAwaiter().GetResult(); 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 IEndPointContext? context) where T : notnull
{ {
var serialized = _serialization.Serialize(payload, context); var serialized = _serialization.Serialize(payload, context);
await _interaction.PostMessageUntrustedAsync(new NetworkMessageHeader await _interaction.PostMessageUntrustedAsync(new NetworkMessage
{ Id = id, MessageType = eMessageType, Data = serialized }); { Id = id, MessageType = eMessageType, Data = serialized });
} }
} }
+2 -2
View File
@@ -5,9 +5,9 @@ namespace mROA.Implementation
public struct RequestContext public struct RequestContext
{ {
public int OwnerId { get; } 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; RequestId = requestId;
OwnerId = ownerId; OwnerId = ownerId;
+64
View File
@@ -0,0 +1,64 @@
using System;
using System.Collections.Generic;
using System.Runtime.InteropServices;
namespace mROA.Implementation
{
public struct RequestId : IEquatable<RequestId>
{
private static Random _random = new();
public ulong P0;
public ulong P1;
public static RequestId Generate()
{
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;
}
}
}