Skip to content

Commit 1b1a3f3

Browse files
saknosakno
committed
Cluster 5.24.1
1 parent 0870503 commit 1b1a3f3

18 files changed

+291
-152
lines changed

‎CHANGELOG.md‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,13 @@
11
Release Notes
22
====
33

4+
# 08-23-2025
5+
<a href="https://www.nuget.org/packages/dotnext.net.cluster/5.24.1">DotNext.Net.Cluster 5.24.1</a>
6+
* Fixed stream ID inflation of the multiplexing protocol client if underlying TCP connection is unstable
7+
8+
<a href="https://www.nuget.org/packages/dotnext.aspnetcore.cluster/5.24.1">DotNext.AspNetCore.Cluster 5.24.1</a>
9+
* Updated dependencies
10+
411
# 08-19-2025
512
<a href="https://www.nuget.org/packages/dotnext/5.24.0">DotNext 5.24.0</a>
613
* Merged [258](https://github.com/dotnet/dotNext/pull/258)

‎README.md‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -65,10 +65,10 @@ Release Date: 08-19-2025
6565
<a href="https://www.nuget.org/packages/dotnext.io/5.24.0">DotNext.IO 5.24.0</a>
6666
* Improved behavioral compatibility with [Pipe](https://learn.microsoft.com/en-us/dotnet/api/system.io.pipelines.pipe) class by extension methods exposed by `PipeExtensions` class
6767

68-
<a href="https://www.nuget.org/packages/dotnext.net.cluster/5.24.0">DotNext.Net.Cluster 5.24.0</a>
68+
<a href="https://www.nuget.org/packages/dotnext.net.cluster/5.24.1">DotNext.Net.Cluster 5.24.1</a>
6969
* Added `DotNext.Net.Multiplexing` namespace that exposes simple unencrypted multiplexing protocol implementation on top of TCP. The multiplexed channel is exposed as [IDuplexPipe](https://learn.microsoft.com/en-us/dotnet/api/system.io.pipelines.iduplexpipe). The main purpose of this implementation is the efficient communication between nodes within the cluster inside the trusted LAN
7070

71-
<a href="https://www.nuget.org/packages/dotnext.aspnetcore.cluster/5.24.0">DotNext.AspNetCore.Cluster 5.24.0</a>
71+
<a href="https://www.nuget.org/packages/dotnext.aspnetcore.cluster/5.24.1">DotNext.AspNetCore.Cluster 5.24.1</a>
7272
* Updated dependencies
7373

7474
<a href="https://www.nuget.org/packages/dotnext.maintenanceservices/0.6.0">DotNext.MaintenanceServices 0.6.0</a>

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

Lines changed: 38 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -223,20 +223,55 @@ public static async Task TerminateStream()
223223
await streamCount.WaitForZero(DefaultTimeout);
224224
}
225225

226-
private sealed class StreamCountObserver() : InstrumentObserver<int, UpDownCounter<int>>(static (instr, tags) => IsStreamCount(instr))
226+
private sealed class StreamCountObserver() : InstrumentObserver<long, UpDownCounter<long>>(static (instr, tags) => IsStreamCount(instr))
227227
{
228228
private readonly TaskCompletionSource zeroReached = new();
229-
private int streamCount;
229+
private long streamCount;
230230

231231
internal static bool IsStreamCount(Instrument instrument)
232232
=> instrument is { Meter.Name: "DotNext.Net.Multiplexing.Server", Name: "streams-count" };
233233

234-
protected override void Record(int value)
234+
protected override void Record(long value)
235235
{
236236
if (Interlocked.Add(ref streamCount, value) is 0)
237237
zeroReached.TrySetResult();
238238
}
239239

240240
public Task WaitForZero(TimeSpan timeout) => zeroReached.Task.WaitAsync(timeout);
241241
}
242+
243+
[Fact]
244+
public static async Task WaitForConnectionAsync()
245+
{
246+
await using var client = new TcpMultiplexedClient(LocalEndPoint, new() { Timeout = DefaultTimeout });
247+
248+
await using var server = new TcpMultiplexedListener(LocalEndPoint, new() { Timeout = DefaultTimeout });
249+
await server.StartAsync();
250+
251+
var task = client.OpenStreamAsync().AsTask();
252+
False(task.IsCompleted);
253+
254+
await client.StartAsync();
255+
await task;
256+
True(task.IsCompletedSuccessfully);
257+
}
258+
259+
[Fact]
260+
public static async Task WaitForDisposedConnectionAsync()
261+
{
262+
Task task;
263+
await using (var client = new TcpMultiplexedClient(LocalEndPoint, new() { Timeout = DefaultTimeout }))
264+
{
265+
task = client.OpenStreamAsync().AsTask();
266+
}
267+
268+
await ThrowsAsync<ObjectDisposedException>(Func.Constant(task));
269+
}
270+
271+
[Fact]
272+
public static async Task WaitForCanceledConnectionAsync()
273+
{
274+
await using var client = new TcpMultiplexedClient(LocalEndPoint, new() { Timeout = DefaultTimeout });
275+
await ThrowsAnyAsync<OperationCanceledException>(client.OpenStreamAsync(new CancellationToken(canceled: true)).AsTask);
276+
}
242277
}

‎src/cluster/DotNext.AspNetCore.Cluster/DotNext.AspNetCore.Cluster.csproj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
<ImplicitUsings>true</ImplicitUsings>
99
<IsAotCompatible>true</IsAotCompatible>
1010
<Features>nullablePublicOnly</Features>
11-
<VersionPrefix>5.24.0</VersionPrefix>
11+
<VersionPrefix>5.24.1</VersionPrefix>
1212
<VersionSuffix></VersionSuffix>
1313
<Authors>.NET Foundation and Contributors</Authors>
1414
<Product>.NEXT Family of Libraries</Product>

‎src/cluster/DotNext.Net.Cluster/DotNext.Net.Cluster.csproj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
<Nullable>enable</Nullable>
99
<IsAotCompatible>true</IsAotCompatible>
1010
<Features>nullablePublicOnly</Features>
11-
<VersionPrefix>5.24.0</VersionPrefix>
11+
<VersionPrefix>5.24.1</VersionPrefix>
1212
<VersionSuffix></VersionSuffix>
1313
<Authors>.NET Foundation and Contributors</Authors>
1414
<Product>.NEXT Family of Libraries</Product>

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

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
using System.Diagnostics.Metrics;
2+
3+
namespace DotNext.Net.Multiplexing;
4+
5+
internal interface IStreamMetrics
6+
{
7+
static abstract UpDownCounter<long> StreamCount { get; }
8+
}

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

Lines changed: 54 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,4 @@
11
using System.Collections.Concurrent;
2-
using System.Diagnostics;
3-
using System.Diagnostics.Metrics;
42
using System.Net.Sockets;
53
using System.Runtime.CompilerServices;
64

@@ -9,79 +7,96 @@ namespace DotNext.Net.Multiplexing;
97
using Buffers;
108
using Threading;
119

12-
internal sealed class InputMultiplexer(
13-
ConcurrentDictionary<uint, MultiplexedStream> streams,
14-
AsyncAutoResetEvent writeSignal,
15-
BufferWriter<byte> framingBuffer,
16-
int flushThreshold,
17-
UpDownCounter<int> streamCounter,
18-
in TagList measurementTags,
19-
TimeSpan timeout,
20-
TimeSpan heartbeatTimeout,
21-
CancellationToken token) : Multiplexer(streams, new ConcurrentQueue<ProtocolCommand>(), streamCounter, measurementTags, token)
10+
internal sealed class InputMultiplexer<T>() : Multiplexer<T>(new(), new ConcurrentQueue<ProtocolCommand>())
11+
where T : IStreamMetrics
2212
{
23-
24-
public TimeSpan Timeout => timeout;
13+
public required TimeSpan Timeout { get; init; }
14+
15+
public required TimeSpan HeartbeatTimeout { private get; init; }
16+
17+
public required int FlushThreshold { private get; init; }
18+
19+
public required BufferWriter<byte> FramingBuffer { private get; init; }
20+
21+
public required AsyncAutoResetEvent TransportSignal { private get; init; }
2522

2623
public bool TryAddStream(uint streamId, MultiplexedStream stream)
2724
{
28-
var result = streams.TryAdd(streamId, stream);
25+
var result = Streams.TryAdd(streamId, stream);
2926
ChangeStreamCount(Unsafe.BitCast<bool, byte>(result));
3027
return result;
3128
}
32-
33-
public OutputMultiplexer CreateOutput(Memory<byte> framingBuffer, TimeSpan receiveTimeout)
34-
=> new(streams, writeSignal, commands, framingBuffer, streamCounter, measurementTags, receiveTimeout, RootToken);
3529

36-
public OutputMultiplexer CreateOutput(Memory<byte> framingBuffer, TimeSpan receiveTimeout, Func<AsyncAutoResetEvent, MultiplexedStream?> handlerFactory,
37-
CancellationToken token)
38-
=> new(streams, writeSignal, commands, framingBuffer, streamCounter, measurementTags, receiveTimeout, token)
39-
{ HandlerFactory = handlerFactory };
30+
public bool TryRemoveStream(uint streamId, MultiplexedStream stream)
31+
{
32+
var removed = Streams.TryRemove(new(streamId, stream));
33+
ChangeStreamCount(-Unsafe.BitCast<bool, byte>(removed));
34+
return removed;
35+
}
36+
37+
public OutputMultiplexer<T> CreateOutput(Memory<byte> framingBuffer, TimeSpan receiveTimeout) => new(Streams, Commands)
38+
{
39+
MeasurementTags = MeasurementTags,
40+
RootToken = RootToken,
41+
FramingBuffer = framingBuffer,
42+
Timeout = receiveTimeout,
43+
TransportSignal = TransportSignal,
44+
};
45+
46+
public OutputMultiplexer<T> CreateOutput(Memory<byte> framingBuffer, TimeSpan receiveTimeout, MultiplexedStreamFactory handlerFactory,
47+
CancellationToken token) => new(Streams, Commands)
48+
{
49+
MeasurementTags = MeasurementTags,
50+
RootToken = token,
51+
FramingBuffer = framingBuffer,
52+
Timeout = receiveTimeout,
53+
TransportSignal = TransportSignal,
54+
Factory = handlerFactory,
55+
};
4056

4157
public async Task ProcessAsync(Func<bool> condition, Socket socket)
4258
{
43-
using var enumerator = streams.GetEnumerator();
59+
using var enumerator = Streams.GetEnumerator();
4460
for (var requiresHeartbeat = false;
4561
condition();
46-
requiresHeartbeat = !await writeSignal.WaitAsync(heartbeatTimeout, RootToken).ConfigureAwait(false))
62+
requiresHeartbeat = !await TransportSignal.WaitAsync(HeartbeatTimeout, RootToken).ConfigureAwait(false))
4763
{
48-
framingBuffer.Clear(reuseBuffer: true);
64+
FramingBuffer.Clear(reuseBuffer: true);
4965

5066
// combine streams
5167
while (enumerator.MoveNext())
5268
{
5369
var (streamId, stream) = enumerator.Current;
5470

55-
if (stream.IsCompleted && streams.TryRemove(streamId, out _))
71+
if (stream.IsCompleted && TryRemoveStream(streamId, stream))
5672
{
57-
Protocol.WriteStreamClosed(framingBuffer, streamId);
58-
ChangeStreamCount(-1);
73+
Protocol.WriteStreamClosed(FramingBuffer, streamId);
5974
}
6075
else
6176
{
62-
await stream.WriteFrameAsync(framingBuffer, streamId).ConfigureAwait(false);
77+
await stream.WriteFrameAsync(FramingBuffer, streamId).ConfigureAwait(false);
6378
}
6479

6580
// write the buffer on overflow
66-
if (framingBuffer.WrittenCount >= flushThreshold)
81+
if (FramingBuffer.WrittenCount >= FlushThreshold)
6782
{
68-
await SendAsync(framingBuffer.WrittenMemory, socket).ConfigureAwait(false);
69-
framingBuffer.Clear(reuseBuffer: true);
83+
await SendAsync(FramingBuffer.WrittenMemory, socket).ConfigureAwait(false);
84+
FramingBuffer.Clear(reuseBuffer: true);
7085
}
7186
}
7287

7388
// process protocol commands
74-
commands.Serialize(framingBuffer);
89+
Commands.Serialize(FramingBuffer);
7590

76-
switch (framingBuffer.WrittenCount)
91+
switch (FramingBuffer.WrittenCount)
7792
{
7893
case 0 when requiresHeartbeat:
79-
Protocol.WriteHeartbeat(framingBuffer);
94+
Protocol.WriteHeartbeat(FramingBuffer);
8095
goto default;
8196
case 0:
8297
break;
8398
default:
84-
await SendAsync(framingBuffer.WrittenMemory, socket).ConfigureAwait(false);
99+
await SendAsync(FramingBuffer.WrittenMemory, socket).ConfigureAwait(false);
85100
break;
86101
}
87102

@@ -94,7 +109,7 @@ private async ValueTask SendAsync(ReadOnlyMemory<byte> buffer, Socket socket)
94109
{
95110
for (int bytesWritten; !buffer.IsEmpty; buffer = buffer.Slice(bytesWritten))
96111
{
97-
StartOperation(timeout);
112+
StartOperation(Timeout);
98113
try
99114
{
100115
bytesWritten = await socket.SendAsync(buffer, TimeBoundedToken).ConfigureAwait(false);
@@ -116,9 +131,9 @@ private async ValueTask SendAsync(ReadOnlyMemory<byte> buffer, Socket socket)
116131

117132
public async ValueTask CompleteAllAsync(Exception e)
118133
{
119-
foreach (var id in streams.Keys)
134+
foreach (var id in Streams.Keys)
120135
{
121-
if (streams.TryRemove(id, out var stream))
136+
if (Streams.TryRemove(id, out var stream))
122137
{
123138
await stream.CompleteTransportOutputAsync(e).ConfigureAwait(false);
124139
await stream.CompleteTransportInputAsync(e).ConfigureAwait(false);

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

Lines changed: 30 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
using System.Diagnostics.CodeAnalysis;
22
using System.IO.Pipelines;
33
using System.Net.Sockets;
4+
using System.Runtime.ExceptionServices;
45
using Microsoft.AspNetCore.Connections;
56

67
namespace DotNext.Net.Multiplexing;
@@ -18,12 +19,36 @@ partial class MultiplexedClient
1819
private readonly PipeOptions options;
1920

2021
[SuppressMessage("Usage", "CA2213", Justification = "False positive")]
21-
private readonly InputMultiplexer input;
22+
private readonly InputMultiplexer<MultiplexedClient> input;
2223

2324
[SuppressMessage("Usage", "CA2213", Justification = "False positive")]
24-
private readonly OutputMultiplexer output;
25+
private readonly OutputMultiplexer<MultiplexedClient> output;
2526
private uint streamId;
2627

28+
private void ReportConnected()
29+
=> Interlocked.Exchange(ref readiness, null)?.TrySetResult();
30+
31+
private void ReportDisconnected()
32+
{
33+
if (readiness is null)
34+
{
35+
var source = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
36+
Interlocked.CompareExchange(ref readiness, source, null);
37+
}
38+
}
39+
40+
private void ReportDisposed()
41+
{
42+
var e = new ObjectDisposedException(GetType().Name);
43+
ExceptionDispatchInfo.SetCurrentStackTrace(e);
44+
if (readiness?.TrySetException(e) is null)
45+
{
46+
var source = new TaskCompletionSource();
47+
source.SetException(e);
48+
Interlocked.CompareExchange(ref readiness, source, null);
49+
}
50+
}
51+
2752
private async Task DispatchAsync()
2853
{
2954
var socket = default(Socket);
@@ -43,7 +68,7 @@ private async Task DispatchAsync()
4368
{
4469
socket = await ConnectAsync(input.RootToken).ConfigureAwait(false);
4570
receiveLoop = output.ProcessAsync(socket);
46-
readiness.TrySetResult();
71+
ReportConnected();
4772
}
4873

4974
// send data
@@ -61,6 +86,8 @@ await input.CompleteAllAsync(new ConnectionResetException(ExceptionMessages.Conn
6186
{
6287
socket?.Dispose();
6388
await receiveLoop.ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
89+
90+
ReportDisconnected();
6491
await input.CompleteAllAsync(e).ConfigureAwait(false);
6592
}
6693
}

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

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,13 +2,15 @@
22

33
namespace DotNext.Net.Multiplexing;
44

5-
partial class MultiplexedClient
5+
partial class MultiplexedClient : IStreamMetrics
66
{
7-
private static readonly UpDownCounter<int> streamCount;
7+
private static readonly UpDownCounter<long> StreamCount;
88

99
static MultiplexedClient()
1010
{
1111
var meter = new Meter("DotNext.Net.Multiplexing.Client");
12-
streamCount = meter.CreateUpDownCounter<int>("streams-count", description: "Number of Streams");
12+
StreamCount = meter.CreateUpDownCounter<long>("streams-count", description: "Number of Streams");
1313
}
14+
15+
static UpDownCounter<long> IStreamMetrics.StreamCount => StreamCount;
1416
}

0 commit comments

Comments
 (0)