diff --git a/mROA.Benchmark/ConcurrentAlloc.cs b/mROA.Benchmark/ConcurrentAlloc.cs new file mode 100644 index 0000000..a51b970 --- /dev/null +++ b/mROA.Benchmark/ConcurrentAlloc.cs @@ -0,0 +1,150 @@ +using System.Formats.Cbor; +using BenchmarkDotNet.Attributes; +using BenchmarkDotNet.Jobs; +using mROA.Implementation; + +namespace mROA.Benchmark; + +[MemoryDiagnoser] +public class ConcurrentAlloc +{ + private readonly CircularMemoryManager _cmm = new(1024); + private readonly FastCircularMemoryManager _fmm = new(); + private CborWriter _writer = new(); + + [GlobalSetup] + public void Setup() + { + _writer.WriteStartArray(2); + _writer.WriteByteString([1, 2, 3, 4, 5, 6, 7, 8]); + _writer.WriteStartArray(2); + _writer.WriteInt32(12); + _writer.WriteTextString("tralala 7EBC1458-BB53-49EA-84C9-EFECC0FC08FD"); + _writer.WriteEndArray(); + _writer.WriteEndArray(); + } + + [Benchmark(Baseline = true)] + public int CircularAllocate() + { + var writer = _writer; + var buffer = _cmm.AllocSlice(writer.BytesWritten); + writer.Encode(buffer); + + return buffer.Length; + } + + [Benchmark] + public int CircularMemoryAllocate() + { + var writer = _writer; + var buffer = _cmm.AllocMemory(writer.BytesWritten); + writer.Encode(buffer.Span); + return buffer.Length; + } + + [Benchmark] + public int FastCircularAllocate() + { + var writer = _writer; + var buffer = _fmm.Alloc(writer.BytesWritten); + writer.Encode(buffer); + return buffer.Length; + } + + [Benchmark] + public int FastMemoryCircularAllocate() + { + var writer = _writer; + var buffer = _fmm.AllocMem(writer.BytesWritten); + writer.Encode(buffer.Span); + return buffer.Length; + } + + [Benchmark] + public int HeapAllocate() + { + var writer = _writer; + var encoded = writer.Encode(); + return encoded.Length; + } +} + +public class FastCircularMemoryManager +{ + private readonly byte[] _buffer; + private long _offset; // atomic offset (in bytes) + private readonly int _mask; // если размер степени двойки — можно использовать маску + + public FastCircularMemoryManager(int size = 4096) // 4KB buffer + { + if (!IsPowerOfTwo(size)) + throw new ArgumentException("Size should be power of two for performance.", nameof(size)); + + _buffer = new byte[size]; + _offset = 0; + _mask = size - 1; // для быстрого циклического сдвига: (offset & _mask) + } + + private static bool IsPowerOfTwo(int x) => x > 0 && (x & (x - 1)) == 0; + + public Span Alloc(int size) + { + if (size > _buffer.Length) + return new byte[size]; // fallback + + long oldOffset, newOffset; + int start; + + // Atomic "bump pointer" с циклическим переполнением + do + { + oldOffset = Volatile.Read(ref _offset); + start = (int)(oldOffset & _mask); + + // Проверяем, не пересекает ли выделение границу буфера + if (start + size > _buffer.Length) + { + // Переполнение — обнуляем (циклический буфер) + newOffset = size; // сбрасываем на начало + size + start = 0; + } + else + { + newOffset = oldOffset + size; + } + } while (Interlocked.CompareExchange(ref _offset, newOffset, oldOffset) != oldOffset); + + return _buffer.AsSpan(start, size); + } + + public Memory AllocMem(int size) + { + if (size > _buffer.Length) + return new byte[size]; // fallback + + long oldOffset, newOffset; + int start; + + // Atomic "bump pointer" с циклическим переполнением + do + { + oldOffset = Volatile.Read(ref _offset); + start = (int)(oldOffset & _mask); + + // Проверяем, не пересекает ли выделение границу буфера + if (start + size > _buffer.Length) + { + // Переполнение — обнуляем (циклический буфер) + newOffset = size; // сбрасываем на начало + size + start = 0; + } + else + { + newOffset = oldOffset + size; + } + } while (Interlocked.CompareExchange(ref _offset, newOffset, oldOffset) != oldOffset); + + return _buffer.AsMemory(start, size); + } +} \ No newline at end of file diff --git a/mROA.Benchmark/Program.cs b/mROA.Benchmark/Program.cs index 560812c..0dc8d53 100644 --- a/mROA.Benchmark/Program.cs +++ b/mROA.Benchmark/Program.cs @@ -5,4 +5,4 @@ using mROA.Benchmark; Console.WriteLine("Hello, World!"); -BenchmarkRunner.Run(); \ No newline at end of file +BenchmarkRunner.Run(); \ No newline at end of file diff --git a/mROA.Test/ConcurrentTest.cs b/mROA.Test/ConcurrentTest.cs new file mode 100644 index 0000000..7efeacf --- /dev/null +++ b/mROA.Test/ConcurrentTest.cs @@ -0,0 +1,34 @@ +using System; +using System.Threading.Tasks; +using mROA.Implementation; + +namespace mROA.Test; + +[TestFixture] +public class ConcurrentTest +{ + private CircularMemoryManager _cmm; + + [SetUp] + public void Setup() + { + _cmm = new CircularMemoryManager(100); + } + + [Test] + public void ParallelAlloc() + { + Parallel.For(0, 100, i => + { + var data = _cmm.AllocSlice(1); + data.Span[0] = 1; + }); + + var total = _cmm.AllocSlice(100).Span; + if (total.IndexOf((byte)0) == -1) + { + Assert.Pass(); + } + Assert.Fail(); + } +} \ No newline at end of file diff --git a/mROA/Abstract/IExecuteModule.cs b/mROA/Abstract/IExecuteModule.cs index a7f7468..723d230 100644 --- a/mROA/Abstract/IExecuteModule.cs +++ b/mROA/Abstract/IExecuteModule.cs @@ -7,7 +7,7 @@ namespace mROA.Abstract { ICommandExecution? Execute(CallRequest command, IInstanceRepository instanceRepository, IRepresentationModule representationModule, IEndPointContext context); - ICommandExecution Cancel(CancelRequest command); + ICommandExecution Cancel(CancelRequest command); } } \ No newline at end of file diff --git a/mROA/Abstract/IRepresentationModule.cs b/mROA/Abstract/IRepresentationModule.cs index 3446380..b83acde 100644 --- a/mROA/Abstract/IRepresentationModule.cs +++ b/mROA/Abstract/IRepresentationModule.cs @@ -25,7 +25,8 @@ namespace mROA.Abstract void PostCallMessage(RequestId id, EMessageType eMessageType, T payload, IEndPointContext? context) where T : notnull; - Task PostCallMessageUntrustedAsync(RequestId id, EMessageType eMessageType, T payload, IEndPointContext? context) + Task PostCallMessageUntrustedAsync(RequestId id, EMessageType eMessageType, T payload, + IEndPointContext? context) where T : notnull; } } \ No newline at end of file diff --git a/mROA/Abstract/IRequestExtractor.cs b/mROA/Abstract/IRequestExtractor.cs index 6fc3816..2a2e01b 100644 --- a/mROA/Abstract/IRequestExtractor.cs +++ b/mROA/Abstract/IRequestExtractor.cs @@ -7,7 +7,7 @@ namespace mROA.Abstract public interface IRequestExtractor { Task StartExtraction(); - void PushMessage(object parced, EMessageType originalType); + void PushMessage(object parsed, EMessageType originalType); Predicate Rule { get; } Func[] Converters { get; } } diff --git a/mROA/Implementation/Backend/NetworkGatewayModule.cs b/mROA/Implementation/Backend/NetworkGatewayModule.cs index 27d62a0..2905d04 100644 --- a/mROA/Implementation/Backend/NetworkGatewayModule.cs +++ b/mROA/Implementation/Backend/NetworkGatewayModule.cs @@ -79,7 +79,7 @@ namespace mROA.Implementation.Backend switch (connectionRequest.MessageType) { case EMessageType.ClientConnect: - + HandleNewClient(context, interaction, streamExtractor, cts, connectionRequest); break; case EMessageType.ClientRecovery: @@ -94,7 +94,8 @@ namespace mROA.Implementation.Backend } private void HandleNewClient(EndPointContext context, ChannelInteractionModule interaction, - ChannelInteractionModule.StreamExtractor streamExtractor, CancellationTokenSource cts, NetworkMessage connection) + ChannelInteractionModule.StreamExtractor streamExtractor, CancellationTokenSource cts, + NetworkMessage connection) { context.HostId = 0; context.OwnerId = -interaction.ConnectionId; @@ -126,8 +127,8 @@ namespace mROA.Implementation.Backend { var func = converters[i]; if (func(message) is not { } t) continue; - - var deserialized = _serialization.Deserialize(message.Data, t, context); + + var deserialized = _serialization.Deserialize(message.Data, t, context)!; Task.Run(() => requestExtractor.PushMessage(deserialized, message.MessageType)); break; } diff --git a/mROA/Implementation/CallRequest.cs b/mROA/Implementation/CallRequest.cs index dcf5fc1..c054016 100644 --- a/mROA/Implementation/CallRequest.cs +++ b/mROA/Implementation/CallRequest.cs @@ -2,8 +2,6 @@ namespace mROA.Implementation { - - public struct CallRequest { public RequestId Id { get; set; } diff --git a/mROA/Implementation/ChannelInteractionModule.cs b/mROA/Implementation/ChannelInteractionModule.cs index 1f90654..4dbef5e 100644 --- a/mROA/Implementation/ChannelInteractionModule.cs +++ b/mROA/Implementation/ChannelInteractionModule.cs @@ -153,16 +153,16 @@ namespace mROA.Implementation } public Action MessageReceived = _ => { }; - + public async Task SingleReceive(CancellationToken token = default) { var firstRead = await _ioStream.ReadAsync(_buffer, token); - + var meta = MemoryMarshal.Read(_buffer.Span); - + var len = meta.BodyLength; var readLen = firstRead - 19; - + if (readLen != len) { var lastPart = _buffer[firstRead..(len + 19)]; @@ -191,9 +191,9 @@ namespace mROA.Implementation message.Data.CopyTo(_buffer.Span[19..]); var sendingSpan = _buffer[..(19 + meta.BodyLength)]; await _ioStream.WriteAsync(sendingSpan, token); - #if TRACE +#if TRACE Console.WriteLine("SEND " + message); - #endif +#endif // _logger.LogTrace("SEND {0}", message.ToString()); } diff --git a/mROA/Implementation/CircularMemoryManager.cs b/mROA/Implementation/CircularMemoryManager.cs new file mode 100644 index 0000000..9ab70c6 --- /dev/null +++ b/mROA/Implementation/CircularMemoryManager.cs @@ -0,0 +1,60 @@ +using System; +using System.Threading; + +namespace mROA.Implementation +{ + public class CircularMemoryManager + { + private readonly Memory _buffer; + private Memory _current; + private SpinLock _spinLock = new(false); + + public CircularMemoryManager(int size) + { + _buffer = new Memory(new byte[size]); + _current = _buffer; + } + + public Span AllocSlice(int size) + { + Span order; + var lockTaken = false; + try + { + _spinLock.Enter(ref lockTaken); + if (_current.Length < size) + _current = _buffer; + + order = _current.Span[..size]; + _current = _current[size..]; + } + finally + { + if (lockTaken) _spinLock.Exit(false); + } + + return order; + } + + public Memory AllocMemory(int size) + { + Memory order; + var lockTaken = false; + try + { + _spinLock.Enter(ref lockTaken); + if (_current.Length < size) + _current = _buffer; + + order = _current[..size]; + _current = _current[size..]; + } + finally + { + if (lockTaken) _spinLock.Exit(false); + } + + return order; + } + } +} \ No newline at end of file diff --git a/mROA/Implementation/CommandExecution/AsyncCommandExecution.cs b/mROA/Implementation/CommandExecution/AsyncCommandExecution.cs new file mode 100644 index 0000000..768668a --- /dev/null +++ b/mROA/Implementation/CommandExecution/AsyncCommandExecution.cs @@ -0,0 +1,11 @@ +using System; +using mROA.Abstract; + +namespace mROA.Implementation.CommandExecution +{ + public class AsyncCommandExecution : ICommandExecution + { + public RequestId Id { get; set; } + public EMessageType MessageType => EMessageType.Unknown; + } +} \ No newline at end of file diff --git a/mROA/Implementation/Frontend/RequestExtractor.cs b/mROA/Implementation/Frontend/RequestExtractor.cs index 8e68ebb..9d97bde 100644 --- a/mROA/Implementation/Frontend/RequestExtractor.cs +++ b/mROA/Implementation/Frontend/RequestExtractor.cs @@ -36,20 +36,20 @@ namespace mROA.Implementation.Frontend } } - public void PushMessage(object parced, EMessageType originalType) + public void PushMessage(object parsed, EMessageType originalType) { switch (originalType) { case EMessageType.CallRequest: - HandleCallRequest((CallRequest)parced); + HandleCallRequest((CallRequest)parsed); break; case EMessageType.ClientDisconnect: return; case EMessageType.EventRequest: - HandleEventRequest((CallRequest)parced); + HandleEventRequest((CallRequest)parsed); break; case EMessageType.CancelRequest: - HandleCancelRequest((CancelRequest)parced); + HandleCancelRequest((CancelRequest)parsed); break; default: throw new ArgumentOutOfRangeException(); @@ -76,7 +76,7 @@ namespace mROA.Implementation.Frontend private void HandleCallRequest(CallRequest request) { var result = _executeModule.Execute(request, _context.RealRepository, _representationModule, _context); - + if (result is null) { return; diff --git a/mROA/Implementation/NetworkMessage.cs b/mROA/Implementation/NetworkMessage.cs index 719bd42..37fc139 100644 --- a/mROA/Implementation/NetworkMessage.cs +++ b/mROA/Implementation/NetworkMessage.cs @@ -37,6 +37,7 @@ namespace mROA.Implementation public byte[] Data { get; set; } public object Serialized { get; set; } public IEndPointContext Context { get; set; } + public override string ToString() { return $" {Id}:{MessageType} [{Data.Length}]"; @@ -51,7 +52,7 @@ namespace mROA.Implementation Id = Id }; } - + public struct NetworkMessageMeta { public RequestId Id; diff --git a/mROA/LegacyExtentions.cs b/mROA/LegacyExtentions.cs index 44757cb..64bef79 100644 --- a/mROA/LegacyExtentions.cs +++ b/mROA/LegacyExtentions.cs @@ -9,7 +9,8 @@ namespace mROA { public static async ValueTask ReadExactlyAsync(this Stream stream, byte[] buffer, int offset, int count) { - return await stream.ReadAtLeastAsyncCore(buffer.AsMemory(offset, count), count, true, CancellationToken.None); + return await stream.ReadAtLeastAsyncCore(buffer.AsMemory(offset, count), count, true, + CancellationToken.None); } public static ValueTask ReadExactlyAsync(this Stream stream, Memory buffer,