Use System.Threading.Lock on NET9_0_OR_GREATER

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++;
+        }
+    }
+}

sandbox/MicroBenchmark/MicroBenchmark.csproj

@@ -2,19 +2,19 @@
 
     <PropertyGroup>
         <OutputType>Exe</OutputType>
-        <TargetFrameworks>net8.0</TargetFrameworks>
+        <TargetFrameworks>net8.0;net10.0</TargetFrameworks>
         <TargetFrameworks Condition="'$(OS)' == 'Windows_NT'">$(TargetFrameworks);net481</TargetFrameworks>
         <ImplicitUsings>enable</ImplicitUsings>
         <Nullable>enable</Nullable>
         <IsPackable>false</IsPackable>
         <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
-        <LangVersion>12</LangVersion>
+        <LangVersion>latest</LangVersion>
     </PropertyGroup>
 
     <ItemGroup>
-        <PackageReference Include="BenchmarkDotNet" Version="0.14.0" />
+        <PackageReference Include="BenchmarkDotNet" Version="0.15.8" />
         <!-- Resolve version conflict between BenchmarkDotNet dependencies on net481 -->
-        <PackageReference Include="System.Reflection.Metadata" Version="8.0.1" Condition="'$(TargetFramework)' == 'net481'" />
+        <PackageReference Include="System.Reflection.Metadata" Version="9.0.0" Condition="'$(TargetFramework)' == 'net481'" />
         <PackageReference Include="StackExchange.Redis" Version="2.8.24" />
         <PackageReference Include="ZLogger" Version="2.5.10" />
         <PackageReference Include="NATS.Client" Version="1.1.6" />

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();
+    }
+}

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<byte> _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()

src/NATS.Client.Core/Internal/ObjectPool.cs

@@ -12,7 +12,11 @@ internal sealed class ObjectPool
 {
     private static int typeId = -1; // Increment by IdentityGenerator<T>
 
+#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<T>[]
 

src/NATS.Client.Core/Internal/ReplyTask.cs

@@ -7,7 +7,11 @@ namespace NATS.Client.Core.Internal;
 
 internal sealed class ReplyTask<T> : 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; }

src/NATS.Client.Core/Internal/SubscriptionManager.cs

@@ -15,7 +15,11 @@ internal sealed class SubscriptionManager : INatsSubscriptionManager, IAsyncDisp
     private readonly ILogger<SubscriptionManager> _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<int, SidMetadata> _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

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<NatsConnection> _logger;
     private readonly ObjectPool _pool;
     private readonly CancellationTokenSource _disposedCts;

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;

src/NATS.Client.JetStream/Internal/NatsJSConsume.cs

@@ -42,7 +42,11 @@ internal class NatsJSConsume<TMsg> : 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;

src/NATS.Client.JetStream/Internal/NatsJSOrderedConsume.cs

@@ -31,7 +31,11 @@ internal class NatsJSOrderedConsume<TMsg> : 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;

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;