diff --git a/sandbox/MicroBenchmark/LockBench.cs b/sandbox/MicroBenchmark/LockBench.cs new file mode 100644 index 000000000..f4dffcc5a --- /dev/null +++ b/sandbox/MicroBenchmark/LockBench.cs @@ -0,0 +1,61 @@ +using BenchmarkDotNet.Attributes; +using BenchmarkDotNet.Jobs; + +namespace MicroBenchmark; + +// Compares object lock (all TFMs) vs System.Threading.Lock (NET9+). +// On net8.0 both methods use object, so they serve as a same-TFM baseline. +// On net10.0 SystemLock uses System.Threading.Lock whose EnterScope() path +// avoids the Monitor sync-block machinery. +[SimpleJob(RuntimeMoniker.Net80)] +[SimpleJob(RuntimeMoniker.Net10_0)] +[MemoryDiagnoser] +public class LockBench +{ + private readonly object _objectLock = new(); + +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _systemLock = new(); +#else + private readonly object _systemLock = new(); +#endif + + private int _value; + + // Uncontended empty lock/unlock -- measures raw locking overhead. + [Benchmark(Baseline = true)] + public void ObjectLock_Empty() + { + lock (_objectLock) + { + } + } + + [Benchmark] + public void SystemLock_Empty() + { + lock (_systemLock) + { + } + } + + // Uncontended lock with a trivial increment -- closer to real usage where + // the lock body does a small amount of work. + [Benchmark] + public void ObjectLock_Increment() + { + lock (_objectLock) + { + _value++; + } + } + + [Benchmark] + public void SystemLock_Increment() + { + lock (_systemLock) + { + _value++; + } + } +} diff --git a/sandbox/MicroBenchmark/MicroBenchmark.csproj b/sandbox/MicroBenchmark/MicroBenchmark.csproj index 414afa07d..5cff5ba9c 100644 --- a/sandbox/MicroBenchmark/MicroBenchmark.csproj +++ b/sandbox/MicroBenchmark/MicroBenchmark.csproj @@ -2,19 +2,19 @@ Exe - net8.0 + net8.0;net10.0 $(TargetFrameworks);net481 enable enable false true - 12 + latest - + - + diff --git a/sandbox/MicroBenchmark/PublishRuntimeBench.cs b/sandbox/MicroBenchmark/PublishRuntimeBench.cs new file mode 100644 index 000000000..fe127d6cb --- /dev/null +++ b/sandbox/MicroBenchmark/PublishRuntimeBench.cs @@ -0,0 +1,44 @@ +using BenchmarkDotNet.Attributes; +using BenchmarkDotNet.Jobs; +using NATS.Client.Core; + +#pragma warning disable CS8618 + +namespace MicroBenchmark; + +// Tight serial publish loop across runtimes. +// Measures end-to-end throughput including buffering, pipe writes, and socket I/O. +// The Lock change affects CommandWriter._lock on each publish, but the body under +// the lock dominates, so any visible delta here reflects runtime improvements broadly. +// Requires a running NATS server on localhost:4222. +[SimpleJob(RuntimeMoniker.Net80)] +[SimpleJob(RuntimeMoniker.Net10_0)] +[MemoryDiagnoser] +public class PublishRuntimeBench +{ + private const int Msgs = 100_000; + private static readonly byte[] Payload = new byte[128]; + + private NatsConnection _nats; + + [GlobalSetup] + public async Task Setup() + { + _nats = new NatsConnection(); + await _nats.ConnectAsync(); + } + + [GlobalCleanup] + public async Task Cleanup() => await _nats.DisposeAsync(); + + [Benchmark] + public async Task Publish100k() + { + for (var i = 0; i < Msgs; i++) + { + await _nats.PublishAsync("bench", Payload); + } + + await _nats.PingAsync(); + } +} diff --git a/src/NATS.Client.Core/Commands/CommandWriter.cs b/src/NATS.Client.Core/Commands/CommandWriter.cs index 364faf241..790238dee 100644 --- a/src/NATS.Client.Core/Commands/CommandWriter.cs +++ b/src/NATS.Client.Core/Commands/CommandWriter.cs @@ -36,7 +36,11 @@ internal sealed class CommandWriter : IAsyncDisposable private readonly NatsConnection _connection; private readonly ObjectPool _pool; private readonly int _arrayPoolInitialSize; +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _lock = new(); +#else private readonly object _lock = new(); +#endif private readonly CancellationTokenSource _cts; private readonly ConnectionStatsCounter _counter; private readonly Memory _consolidateMem = new byte[SendMemSize].AsMemory(); @@ -957,7 +961,11 @@ private async ValueTask UnsubscribeStateMachineAsync(bool lockHeld, int sid, int private class PartialSendFailureCounter { private const int MaxRetry = 1; +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _gate = new(); +#else private readonly object _gate = new(); +#endif private int _count; public bool Failed() diff --git a/src/NATS.Client.Core/Internal/ObjectPool.cs b/src/NATS.Client.Core/Internal/ObjectPool.cs index 0658294f2..9bb5fa4df 100644 --- a/src/NATS.Client.Core/Internal/ObjectPool.cs +++ b/src/NATS.Client.Core/Internal/ObjectPool.cs @@ -12,7 +12,11 @@ internal sealed class ObjectPool { private static int typeId = -1; // Increment by IdentityGenerator +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _gate = new(); +#else private readonly object _gate = new object(); +#endif private readonly int _poolLimit; private object[] _poolNodes = new object[4]; // ObjectPool[] diff --git a/src/NATS.Client.Core/Internal/ReplyTask.cs b/src/NATS.Client.Core/Internal/ReplyTask.cs index 83110864d..260857e05 100644 --- a/src/NATS.Client.Core/Internal/ReplyTask.cs +++ b/src/NATS.Client.Core/Internal/ReplyTask.cs @@ -7,7 +7,11 @@ namespace NATS.Client.Core.Internal; internal sealed class ReplyTask : ReplyTaskBase, IDisposable { +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _gate; +#else private readonly object _gate; +#endif private readonly ReplyTaskFactory _factory; private readonly long _id; private readonly NatsConnection _connection; @@ -25,7 +29,11 @@ public ReplyTask(ReplyTaskFactory factory, long id, string subject, NatsConnecti _deserializer = deserializer; _requestTimeout = requestTimeout; _tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); +#if NET9_0_OR_GREATER + _gate = new System.Threading.Lock(); +#else _gate = new object(); +#endif } public string Subject { get; } diff --git a/src/NATS.Client.Core/Internal/SubscriptionManager.cs b/src/NATS.Client.Core/Internal/SubscriptionManager.cs index ce0ee5bcb..c4f5d2a6f 100644 --- a/src/NATS.Client.Core/Internal/SubscriptionManager.cs +++ b/src/NATS.Client.Core/Internal/SubscriptionManager.cs @@ -15,7 +15,11 @@ internal sealed class SubscriptionManager : INatsSubscriptionManager, IAsyncDisp private readonly ILogger _logger; private readonly bool _trace; private readonly bool _debug; +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _gate = new(); +#else private readonly object _gate = new(); +#endif private readonly NatsConnection _connection; private readonly string _inboxPrefix; private readonly ConcurrentDictionary _bySid = new(); @@ -96,18 +100,17 @@ public ValueTask PublishToClientHandlersAsync(string subject, string? replyTo, i } int? orphanSid = null; + NatsSubBase? targetSub = null; lock (_gate) { if (_bySid.TryGetValue(sid, out var sidMetadata)) { - if (sidMetadata.WeakReference.TryGetTarget(out var sub)) + if (sidMetadata.WeakReference.TryGetTarget(out targetSub)) { if (_trace) { _logger.LogTrace(NatsLogEvents.Subscription, "Found subscription handler for {Subject}/{Sid}", subject, sid); } - - return sub.ReceiveAsync(subject, replyTo, headersBuffer, payloadBuffer); } else { @@ -126,6 +129,11 @@ public ValueTask PublishToClientHandlersAsync(string subject, string? replyTo, i } } + if (targetSub != null) + { + return targetSub.ReceiveAsync(subject, replyTo, headersBuffer, payloadBuffer); + } + if (orphanSid != null) { try diff --git a/src/NATS.Client.Core/NatsConnection.cs b/src/NATS.Client.Core/NatsConnection.cs index b814a3676..8d7c106ce 100644 --- a/src/NATS.Client.Core/NatsConnection.cs +++ b/src/NATS.Client.Core/NatsConnection.cs @@ -36,7 +36,11 @@ public partial class NatsConnection : INatsConnection #pragma warning disable SA1401 internal readonly ConnectionStatsCounter Counter; // allow to call from external sources #pragma warning restore SA1401 +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _gate = new(); +#else private readonly object _gate = new object(); +#endif private readonly ILogger _logger; private readonly ObjectPool _pool; private readonly CancellationTokenSource _disposedCts; diff --git a/src/NATS.Client.Core/NatsSubBase.cs b/src/NATS.Client.Core/NatsSubBase.cs index 72f3ec8c4..5918e272b 100644 --- a/src/NATS.Client.Core/NatsSubBase.cs +++ b/src/NATS.Client.Core/NatsSubBase.cs @@ -32,7 +32,11 @@ public abstract class NatsSubBase { private static readonly byte[] NoRespondersHeaderSequence = { (byte)' ', (byte)'5', (byte)'0', (byte)'3' }; private readonly ILogger _logger; +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _gate = new(); +#else private readonly object _gate = new(); +#endif private readonly bool _debug; private readonly INatsSubscriptionManager _manager; private readonly Timer? _timeoutTimer; diff --git a/src/NATS.Client.JetStream/Internal/NatsJSConsume.cs b/src/NATS.Client.JetStream/Internal/NatsJSConsume.cs index 310966c5c..79de68bdd 100644 --- a/src/NATS.Client.JetStream/Internal/NatsJSConsume.cs +++ b/src/NATS.Client.JetStream/Internal/NatsJSConsume.cs @@ -42,7 +42,11 @@ internal class NatsJSConsume : NatsSubBase private readonly long _thresholdBytes; private readonly int _maxConsecutive503Errors; +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _pendingGate = new(); +#else private readonly object _pendingGate = new(); +#endif private long _pendingMsgs; private long _pendingBytes; private int _disposed; diff --git a/src/NATS.Client.JetStream/Internal/NatsJSOrderedConsume.cs b/src/NATS.Client.JetStream/Internal/NatsJSOrderedConsume.cs index d327b9e3e..8f7990034 100644 --- a/src/NATS.Client.JetStream/Internal/NatsJSOrderedConsume.cs +++ b/src/NATS.Client.JetStream/Internal/NatsJSOrderedConsume.cs @@ -31,7 +31,11 @@ internal class NatsJSOrderedConsume : NatsSubBase private readonly long _maxBytes; private readonly long _thresholdBytes; +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _pendingGate = new(); +#else private readonly object _pendingGate = new(); +#endif private long _pendingMsgs; private long _pendingBytes; private int _disposed; diff --git a/src/NATS.Client.JetStream/NatsJSConsumer.cs b/src/NATS.Client.JetStream/NatsJSConsumer.cs index 996bfacf2..2f6825225 100644 --- a/src/NATS.Client.JetStream/NatsJSConsumer.cs +++ b/src/NATS.Client.JetStream/NatsJSConsumer.cs @@ -14,7 +14,11 @@ public class NatsJSConsumer : INatsJSConsumer private readonly NatsJSContext _context; private readonly string _stream; private readonly string _consumer; +#if NET9_0_OR_GREATER + private readonly System.Threading.Lock _pinIdLock = new(); +#else private readonly object _pinIdLock = new(); +#endif private volatile bool _deleted; private string? _pinId;