Added metrics for pending streams count

Author: sakno

src/DotNext.Tests/Net/Multiplexing/TcpMultiplexerTests.cs

@@ -223,15 +223,15 @@ public static async Task TerminateStream()
         await streamCount.WaitForZero(DefaultTimeout);
     }
 
-    private sealed class StreamCountObserver() : InstrumentObserver<int, UpDownCounter<int>>(static (instr, tags) => IsStreamCount(instr))
+    private sealed class StreamCountObserver() : InstrumentObserver<long, UpDownCounter<long>>(static (instr, tags) => IsStreamCount(instr))
     {
         private readonly TaskCompletionSource zeroReached = new();
-        private int streamCount;
+        private long streamCount;
 
         internal static bool IsStreamCount(Instrument instrument)
             => instrument is { Meter.Name: "DotNext.Net.Multiplexing.Server", Name: "streams-count" };
 
-        protected override void Record(int value)
+        protected override void Record(long value)
         {
             if (Interlocked.Add(ref streamCount, value) is 0)
                 zeroReached.TrySetResult();

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

@@ -0,0 +1,8 @@
+using System.Diagnostics;
+
+namespace DotNext.Net.Multiplexing;
+
+internal interface IStreamMetrics
+{
+    static abstract void ChangeStreamCount(long delta, in TagList measurementTags);
+}

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

@@ -1,6 +1,5 @@
 using System.Collections.Concurrent;
 using System.Diagnostics;
-using System.Diagnostics.Metrics;
 using System.Net.Sockets;
 using System.Runtime.CompilerServices;
 
@@ -9,16 +8,16 @@ namespace DotNext.Net.Multiplexing;
 using Buffers;
 using Threading;
 
-internal sealed class InputMultiplexer(
+internal sealed class InputMultiplexer<T>(
     ConcurrentDictionary<uint, MultiplexedStream> streams,
     AsyncAutoResetEvent writeSignal,
     BufferWriter<byte> framingBuffer,
     int flushThreshold,
-    UpDownCounter<int> streamCounter,
     in TagList measurementTags,
     TimeSpan timeout,
     TimeSpan heartbeatTimeout,
-    CancellationToken token) : Multiplexer(streams, new ConcurrentQueue<ProtocolCommand>(), streamCounter, measurementTags, token)
+    CancellationToken token) : Multiplexer<T>(streams, new ConcurrentQueue<ProtocolCommand>(), measurementTags, token)
+    where T : IStreamMetrics
 {
 
     public TimeSpan Timeout => timeout;
@@ -37,13 +36,13 @@ public bool TryRemoveStream(uint streamId, MultiplexedStream stream)
         return removed;
     }
 
-    public OutputMultiplexer CreateOutput(Memory<byte> framingBuffer, TimeSpan receiveTimeout)
-        => new(streams, writeSignal, commands, framingBuffer, streamCounter, measurementTags, receiveTimeout, RootToken);
+    public OutputMultiplexer<T> CreateOutput(Memory<byte> framingBuffer, TimeSpan receiveTimeout)
+        => new(streams, writeSignal, commands, framingBuffer, measurementTags, receiveTimeout, RootToken);
 
-    public OutputMultiplexer CreateOutput(Memory<byte> framingBuffer, TimeSpan receiveTimeout, Func<AsyncAutoResetEvent, MultiplexedStream?> handlerFactory,
+    public OutputMultiplexer<T> CreateOutput(Memory<byte> framingBuffer, TimeSpan receiveTimeout, MultiplexedStreamFactory handlerFactory,
         CancellationToken token)
-        => new(streams, writeSignal, commands, framingBuffer, streamCounter, measurementTags, receiveTimeout, token)
-            { HandlerFactory = handlerFactory };
+        => new(streams, writeSignal, commands, framingBuffer, measurementTags, receiveTimeout, token)
+            { Factory = handlerFactory };
 
     public async Task ProcessAsync(Func<bool> condition, Socket socket)
     {

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

@@ -19,10 +19,10 @@ partial class MultiplexedClient
     private readonly PipeOptions options;
 
     [SuppressMessage("Usage", "CA2213", Justification = "False positive")]
-    private readonly InputMultiplexer input;
+    private readonly InputMultiplexer<MultiplexedClient> input;
 
     [SuppressMessage("Usage", "CA2213", Justification = "False positive")]
-    private readonly OutputMultiplexer output;
+    private readonly OutputMultiplexer<MultiplexedClient> output;
     private uint streamId;
 
     private void ReportConnected()

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

@@ -1,14 +1,18 @@
+using System.Diagnostics;
 using System.Diagnostics.Metrics;
 
 namespace DotNext.Net.Multiplexing;
 
-partial class MultiplexedClient
+partial class MultiplexedClient : IStreamMetrics
 {
-    private static readonly UpDownCounter<int> streamCount;
+    private static readonly UpDownCounter<long> StreamCount;
 
     static MultiplexedClient()
     {
         var meter = new Meter("DotNext.Net.Multiplexing.Client");
-        streamCount = meter.CreateUpDownCounter<int>("streams-count", description: "Number of Streams");
+        StreamCount = meter.CreateUpDownCounter<long>("streams-count", description: "Number of Streams");
     }
+
+    static void IStreamMetrics.ChangeStreamCount(long delta, in TagList measurementTags)
+        => StreamCount.Add(delta, measurementTags);
 }

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

@@ -38,7 +38,6 @@ protected MultiplexedClient(Options configuration)
             writeSignal,
             framingBuffer,
             flushThreshold,
-            streamCount,
             configuration.MeasurementTags,
             configuration.Timeout,
             configuration.HeartbeatTimeout,

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

@@ -30,12 +30,11 @@ private async Task DispatchAsync(Socket socket)
         var writeSignal = new AsyncAutoResetEvent(initialState: false);
         var receiveBuffer = allocator(flushThreshold);
         var framingBuffer = new PoolingBufferWriter<byte>(allocator) { Capacity = flushThreshold };
-        var input = new InputMultiplexer(
+        var input = new InputMultiplexer<MultiplexedListener>(
             new(),
             writeSignal,
             framingBuffer,
             flushThreshold,
-            streamCount,
             CreateMeasurementTags(socket),
             timeout,
             heartbeatTimeout,

src/cluster/DotNext.Net.Cluster/Net/Multiplexing/MultiplexedListener.Metrics.cs

@@ -3,17 +3,21 @@
 
 namespace DotNext.Net.Multiplexing;
 
-partial class MultiplexedListener
+partial class MultiplexedListener : IStreamMetrics
 {
     private const string ClientAddressMeterAttribute = "dotnext.net.multiplexing.client.address";
-    
-    private static readonly UpDownCounter<int> streamCount;
+
+    private static readonly UpDownCounter<long> StreamCount, PendingStreamCount;
 
     static MultiplexedListener()
     {
         var meter = new Meter("DotNext.Net.Multiplexing.Server");
-        streamCount = meter.CreateUpDownCounter<int>("streams-count", description: "Number of Streams");
+        StreamCount = meter.CreateUpDownCounter<long>("streams-count", description: "Number of Streams");
+        PendingStreamCount = meter.CreateUpDownCounter<long>("pending-streams-count", description: "Number of Pending Streams");
     }
 
     private readonly TagList measurementTags;
+
+    static void IStreamMetrics.ChangeStreamCount(long delta, in TagList measurementTags)
+        => StreamCount.Add(delta, measurementTags);
 }

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

@@ -1,3 +1,4 @@
+using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
 using System.IO.Pipelines;
 using System.Net.Sockets;
@@ -20,7 +21,7 @@ public abstract partial class MultiplexedListener : Disposable, IAsyncDisposable
     private readonly MemoryAllocator<byte> allocator;
     private readonly int flushThreshold;
     private readonly TaskCompletionSource readiness;
-    private readonly Func<AsyncAutoResetEvent, MultiplexedStream?> streamFactory;
+    private readonly MultiplexedStreamFactory streamFactory;
     private Task listener;
 
     [SuppressMessage("Usage", "CA2213", Justification = "False positive")]
@@ -41,7 +42,7 @@ protected MultiplexedListener(Options configuration)
         });
 
         allocator = configuration.ToAllocator();
-        streamFactory = new MultiplexedStreamFactory(configuration.BufferOptions, backlog.Writer).CreateStream;
+        streamFactory = new MultiplexedStreamFactoryImpl(configuration.BufferOptions, backlog.Writer);
         measurementTags = configuration.MeasurementTags;
         flushThreshold = configuration.BufferCapacity;
         readiness = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -80,6 +81,7 @@ public async ValueTask<IDuplexPipe> AcceptAsync(CancellationToken token = defaul
             throw new ObjectDisposedException(GetType().Name);
         }
 
+        PendingStreamCount.Add(-1L);
         return result;
     }
 
@@ -188,15 +190,24 @@ protected override async ValueTask DisposeAsyncCore()
 
     /// <inheritdoc/>
     public new ValueTask DisposeAsync() => base.DisposeAsync();
-}
 
-file sealed class MultiplexedStreamFactory(PipeOptions options, ChannelWriter<MultiplexedStream> backlog)
-{
-    public MultiplexedStream? CreateStream(AsyncAutoResetEvent writeSignal)
+    private sealed class MultiplexedStreamFactoryImpl(PipeOptions options, ChannelWriter<MultiplexedStream> backlog)
     {
-        var stream = new MultiplexedStream(options, writeSignal);
-        return backlog.TryWrite(stream)
-            ? stream
-            : null;
+        private MultiplexedStream? CreateStream(AsyncAutoResetEvent writeSignal, in TagList measurementTags)
+        {
+            var stream = new MultiplexedStream(options, writeSignal);
+            if (backlog.TryWrite(stream))
+            {
+                PendingStreamCount.Add(1L, measurementTags);
+            }
+            else
+            {
+                stream = null;
+            }
+
+            return stream;
+        }
+
+        public static implicit operator MultiplexedStreamFactory(MultiplexedStreamFactoryImpl impl) => impl.CreateStream;
     }
 }

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

@@ -0,0 +1,7 @@
+using System.Diagnostics;
+
+namespace DotNext.Net.Multiplexing;
+
+using Threading;
+
+internal delegate MultiplexedStream? MultiplexedStreamFactory(AsyncAutoResetEvent transportSignal, in TagList measurementTags);