Reduced frame header size

Author: sakno

src/cluster/DotNext.Net.Cluster/Net/Multiplexing/FrameControl.cs

@@ -3,7 +3,7 @@ namespace DotNext.Net.Multiplexing;
 /// <summary>
 /// Controls the fragment behavior.
 /// </summary>
-internal enum FrameControl : ushort
+internal enum FrameControl : byte
 {
     /// <summary>
     /// The fragment contains a data chunk.
@@ -41,5 +41,5 @@ internal enum FrameControl : ushort
     /// <summary>
     /// Heartbeat system fragment.
     /// </summary>
-    Heartbeat = 1024,
+    Heartbeat = 128,
 }

src/cluster/DotNext.Net.Cluster/Net/Multiplexing/FrameHeader.cs

@@ -12,18 +12,19 @@ namespace DotNext.Net.Multiplexing;
 /// <param name="control">The fragment behavior.</param>
 /// <param name="length">The length of the fragment data.</param>
 [StructLayout(LayoutKind.Auto)]
-internal readonly struct FrameHeader(ulong id, FrameControl control, ushort length) : IBinaryFormattable<FrameHeader>
+internal readonly struct FrameHeader(uint id, FrameControl control, ushort length) : IBinaryFormattable<FrameHeader>
 {
     private const byte CurrentVersion = 0;
 
     /// <summary>
     /// All protocol-specific fragments have ID = 0.
     /// </summary>
-    public const ulong SystemStreamId = 0U;
+    public const uint SystemStreamId = 0U;
 
-    public const int Size = sizeof(byte) + sizeof(ulong) + sizeof(FrameControl) + sizeof(ushort);
+    // Version(1 byte) StreamId(4 bytes) FrameControl(1 byte) PayloadLength(2 bytes)
+    public const int Size = sizeof(byte) + sizeof(uint) + sizeof(FrameControl) + sizeof(ushort);
 
-    public ulong Id => id; // if 0, then the packet is protocol-specific
+    public uint Id => id; // if 0, then the packet is protocol-specific
     public FrameControl Control => control;
     public ushort Length => length;
 
@@ -36,7 +37,7 @@ public void Format(Span<byte> destination)
         var writer = new SpanWriter<byte>(destination);
         writer.Add(CurrentVersion);
         writer.WriteLittleEndian(id);
-        writer.WriteLittleEndian((ushort)control);
+        writer.Add((byte)control);
         writer.WriteLittleEndian(length);
     }
 
@@ -47,10 +48,10 @@ public static FrameHeader Parse(ReadOnlySpan<byte> source)
 
         if (version > CurrentVersion)
             throw new UnsupportedVersionException(version);
-        
+
         return new(
-            reader.ReadLittleEndian<ulong>(),
-            (FrameControl)reader.ReadLittleEndian<ushort>(),
+            reader.ReadLittleEndian<uint>(),
+            (FrameControl)reader.Read(),
             reader.ReadLittleEndian<ushort>());
     }
 }

src/cluster/DotNext.Net.Cluster/Net/Multiplexing/InputMultiplexer.cs

@@ -10,7 +10,7 @@ namespace DotNext.Net.Multiplexing;
 using Threading;
 
 internal sealed class InputMultiplexer(
-    ConcurrentDictionary<ulong, MultiplexedStream> streams,
+    ConcurrentDictionary<uint, MultiplexedStream> streams,
     AsyncAutoResetEvent writeSignal,
     BufferWriter<byte> framingBuffer,
     int flushThreshold,
@@ -23,7 +23,7 @@ internal sealed class InputMultiplexer(
 
     public TimeSpan Timeout => timeout;
 
-    public bool TryAddStream(ulong streamId, MultiplexedStream stream)
+    public bool TryAddStream(uint streamId, MultiplexedStream stream)
     {
         var result = streams.TryAdd(streamId, stream);
         ChangeStreamCount(Unsafe.BitCast<bool, byte>(result));

src/cluster/DotNext.Net.Cluster/Net/Multiplexing/MultiplexedClient.Dispatcher.cs

@@ -1,4 +1,3 @@
-using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
 using System.IO.Pipelines;
 using System.Net.Sockets;
@@ -23,7 +22,7 @@ partial class MultiplexedClient
 
     [SuppressMessage("Usage", "CA2213", Justification = "False positive")]
     private readonly OutputMultiplexer output;
-    private ulong streamId;
+    private uint streamId;
 
     private async Task DispatchAsync()
     {

src/cluster/DotNext.Net.Cluster/Net/Multiplexing/MultiplexedClient.cs

@@ -95,7 +95,7 @@ private async ValueTask<IDuplexPipe> OpenStreamCoreAsync(CancellationToken token
     private MultiplexedStream OpenStream()
     {
         var stream = new MultiplexedStream(options, writeSignal);
-        ulong id;
+        uint id;
         do
         {
             id = Interlocked.Increment(ref streamId);

src/cluster/DotNext.Net.Cluster/Net/Multiplexing/MultiplexedStream.TransportSideReader.cs

@@ -11,7 +11,7 @@ partial class MultiplexedStream
     // this is the maximum number of bytes that we can send without adjusting the window
     private volatile int inputWindow;
 
-    public ValueTask WriteFrameAsync(IBufferWriter<byte> writer, ulong streamId)
+    public ValueTask WriteFrameAsync(IBufferWriter<byte> writer, uint streamId)
     {
         ValueTask task;
         if (!IsTransportInputCompleted(out var appSideCompleted))
@@ -64,15 +64,15 @@ private ValueTask CloseAndCancelOutputAsync()
         return task;
     }
 
-    private ValueTask WriteFrameAsync(IBufferWriter<byte> writer, ulong streamId, in ReadResult result)
+    private ValueTask WriteFrameAsync(IBufferWriter<byte> writer, uint streamId, in ReadResult result)
     {
         AdjustWindowIfNeeded(writer, streamId);
         return WriteFrame(writer, streamId, result)
             ? CompleteTransportInputAsync()
             : ValueTask.CompletedTask;
     }
 
-    private bool WriteFrame(Span<byte> frameBuffer, ulong streamId, in ReadResult readResult, out int bytesWritten, out SequencePosition consumed)
+    private bool WriteFrame(Span<byte> frameBuffer, uint streamId, in ReadResult readResult, out int bytesWritten, out SequencePosition consumed)
     {
         Debug.Assert(frameBuffer.Length > FrameHeader.Size);
 
@@ -117,7 +117,7 @@ private bool WriteFrame(Span<byte> frameBuffer, ulong streamId, in ReadResult re
         return completed;
     }
 
-    private bool WriteFrame(IBufferWriter<byte> writer, ulong streamId, in ReadResult result)
+    private bool WriteFrame(IBufferWriter<byte> writer, uint streamId, in ReadResult result)
     {
         var buffer = writer.GetSpan(frameAndHeaderSize);
         var completed = WriteFrame(

src/cluster/DotNext.Net.Cluster/Net/Multiplexing/MultiplexedStream.TransportSideWriter.cs

@@ -77,7 +77,7 @@ void IApplicationSideStream.Consume(long count)
             transportSignal.Set();
     }
 
-    private void AdjustWindowIfNeeded(IBufferWriter<byte> writer, ulong streamId)
+    private void AdjustWindowIfNeeded(IBufferWriter<byte> writer, uint streamId)
     {
         var totalReceived = Atomic.Read(in bytesReceived);
         if (totalReceived > resumeWatermark)

src/cluster/DotNext.Net.Cluster/Net/Multiplexing/Multiplexer.cs

@@ -5,15 +5,15 @@
 namespace DotNext.Net.Multiplexing;
 
 internal abstract partial class Multiplexer(
-    ConcurrentDictionary<ulong, MultiplexedStream> streams,
+    ConcurrentDictionary<uint, MultiplexedStream> streams,
     IProducerConsumerCollection<ProtocolCommand> commands,
     UpDownCounter<int> streamCounter,
     in TagList measurementTags,
     CancellationToken token) : Disposable
 {
     protected readonly UpDownCounter<int> streamCounter = streamCounter;
     protected readonly TagList measurementTags = measurementTags;
-    protected readonly ConcurrentDictionary<ulong, MultiplexedStream> streams = streams;
+    protected readonly ConcurrentDictionary<uint, MultiplexedStream> streams = streams;
     protected readonly IProducerConsumerCollection<ProtocolCommand> commands = commands;
 
     protected void ChangeStreamCount(int delta = 1) => streamCounter.Add(delta, in measurementTags);

src/cluster/DotNext.Net.Cluster/Net/Multiplexing/OutputMultiplexer.cs

@@ -8,7 +8,7 @@ namespace DotNext.Net.Multiplexing;
 using Threading;
 
 internal sealed class OutputMultiplexer(
-    ConcurrentDictionary<ulong, MultiplexedStream> streams,
+    ConcurrentDictionary<uint, MultiplexedStream> streams,
     AsyncAutoResetEvent writeSignal,
     IProducerConsumerCollection<ProtocolCommand> commands,
     Memory<byte> framingBuffer,

src/cluster/DotNext.Net.Cluster/Net/Multiplexing/Protocol.cs

@@ -7,7 +7,7 @@ namespace DotNext.Net.Multiplexing;
 
 internal static class Protocol
 {
-    private static void WriteEmptyFrame(IBufferWriter<byte> writer, [ConstantExpected] FrameControl control, ulong streamId)
+    private static void WriteEmptyFrame(IBufferWriter<byte> writer, [ConstantExpected] FrameControl control, uint streamId)
     {
         var header = new FrameHeader(streamId, control, length: 0);
         header.Format(writer.GetSpan(FrameHeader.Size));
@@ -17,13 +17,13 @@ private static void WriteEmptyFrame(IBufferWriter<byte> writer, [ConstantExpecte
     public static void WriteHeartbeat(IBufferWriter<byte> writer)
         => WriteEmptyFrame(writer, FrameControl.Heartbeat, FrameHeader.SystemStreamId);
 
-    public static void WriteStreamRejected(IBufferWriter<byte> writer, ulong streamId)
+    public static void WriteStreamRejected(IBufferWriter<byte> writer, uint streamId)
         => WriteEmptyFrame(writer, FrameControl.StreamRejected, streamId);
 
-    public static void WriteStreamClosed(IBufferWriter<byte> writer, ulong streamId)
+    public static void WriteStreamClosed(IBufferWriter<byte> writer, uint streamId)
         => WriteEmptyFrame(writer, FrameControl.StreamClosed, streamId);
 
-    public static void WriteAdjustWindow(IBufferWriter<byte> writer, ulong streamId, int windowSize)
+    public static void WriteAdjustWindow(IBufferWriter<byte> writer, uint streamId, int windowSize)
     {
         const int messageSize = FrameHeader.Size + sizeof(int);