More System.Threading.Lock locking.

Author: MarkCiliaVincenti

src/Grpc.AspNetCore.Server/Internal/HttpContextStreamWriter.cs

@@ -33,7 +33,7 @@ internal class HttpContextStreamWriter<TResponse> : IServerStreamWriter<TRespons
     private readonly Action<TResponse, SerializationContext> _serializer;
     private readonly PipeWriter _bodyWriter;
     private readonly IHttpRequestLifetimeFeature _requestLifetimeFeature;
-    private readonly object _writeLock;
+    private readonly Lock _writeLock;
     private Task? _writeTask;
     private bool _completed;
     private long _writeCount;
@@ -42,7 +42,7 @@ public HttpContextStreamWriter(HttpContextServerCallContext context, Action<TRes
     {
         _context = context;
         _serializer = serializer;
-        _writeLock = new object();
+        _writeLock = new Lock();
 
         // Copy HttpContext values.
         // This is done to avoid a race condition when reading them from HttpContext later when running in a separate thread.
@@ -92,7 +92,8 @@ private async Task WriteCoreAsync(TResponse message, CancellationToken cancellat
                 throw new InvalidOperationException("Can't write the message because the request is complete.");
             }
 
-            lock (_writeLock)
+            _writeLock.Enter();
+            try
             {
                 // Pending writes need to be awaited first
                 if (IsWriteInProgressUnsynchronized)
@@ -103,6 +104,10 @@ private async Task WriteCoreAsync(TResponse message, CancellationToken cancellat
                 // Save write task to track whether it is complete. Must be set inside lock.
                 _writeTask = _bodyWriter.WriteStreamedMessageAsync(message, _context, _serializer, cancellationToken);
             }
+            finally
+            {
+                _writeLock.Exit();
+            }
 
             await _writeTask;
             Interlocked.Increment(ref _writeCount);

src/Grpc.HealthCheck/HealthServiceImpl.cs

@@ -41,7 +41,7 @@ public class HealthServiceImpl : Grpc.Health.V1.Health.HealthBase
     private readonly Dictionary<string, HealthCheckResponse.Types.ServingStatus> statusMap =
         new Dictionary<string, HealthCheckResponse.Types.ServingStatus>();
 
-    private readonly object watchersLock = new object();
+    private readonly Lock watchersLock = new Lock();
     private readonly Dictionary<string, List<ChannelWriter<HealthCheckResponse>>> watchers =
         new Dictionary<string, List<ChannelWriter<HealthCheckResponse>>>();
 
@@ -155,7 +155,8 @@ public override async Task Watch(HealthCheckRequest request, IServerStreamWriter
             FullMode = BoundedChannelFullMode.DropOldest
         });
 
-        lock (watchersLock)
+        watchersLock.Enter();
+        try
         {
             if (!watchers.TryGetValue(service, out List<ChannelWriter<HealthCheckResponse>>? channelWriters))
             {
@@ -165,6 +166,10 @@ public override async Task Watch(HealthCheckRequest request, IServerStreamWriter
 
             channelWriters.Add(channel.Writer);
         }
+        finally
+        {
+            watchersLock.Exit();
+        }
 
         // Watch calls run until ended by the client canceling them.
         context.CancellationToken.Register(() => {

src/Grpc.Net.Client/Balancer/Internal/BalancerHttpHandler.cs

@@ -31,7 +31,7 @@ namespace Grpc.Net.Client.Balancer.Internal;
 
 internal class BalancerHttpHandler : DelegatingHandler
 {
-    private static readonly object SetupLock = new object();
+    private static readonly Lock SetupLock = new Lock();
 
     internal const string WaitForReadyKey = "WaitForReady";
     internal const string SubchannelKey = "Subchannel";

src/Grpc.Net.Client/Balancer/Internal/ChildHandlerLoadBalancer.cs

@@ -1,4 +1,4 @@
-#region Copyright notice and license
+#region Copyright notice and license
 
 // Copyright 2019 The gRPC Authors
 //
@@ -53,7 +53,7 @@ internal sealed class ChildHandlerLoadBalancer : LoadBalancer
     private readonly IChannelControlHelper _controller;
     private readonly ServiceConfig? _initialServiceConfig;
     private readonly ConnectionManager _connectionManager;
-    private readonly object _lock = new object();
+    private readonly Lock _lock = new Lock();
 
     internal (LoadBalancer LoadBalancer, string Name)? _current;
     internal (LoadBalancer LoadBalancer, string Name)? _pending;

src/Grpc.Net.Client/Balancer/Internal/ConnectionManager.cs

@@ -30,7 +30,7 @@ internal sealed class ConnectionManager : IDisposable, IChannelControlHelper
     public static readonly BalancerAttributesKey<string> HostOverrideKey = new BalancerAttributesKey<string>("HostOverride");
     private static readonly ChannelIdProvider _channelIdProvider = new ChannelIdProvider();
 
-    private readonly object _lock;
+    private readonly Lock _lock;
     internal readonly Resolver _resolver;
     private readonly ISubchannelTransportFactory _subchannelTransportFactory;
     private readonly List<Subchannel> _subchannels;
@@ -56,7 +56,7 @@ internal ConnectionManager(
         ISubchannelTransportFactory subchannelTransportFactory,
         LoadBalancerFactory[] loadBalancerFactories)
     {
-        _lock = new object();
+        _lock = new Lock();
         _nextPickerTcs = new TaskCompletionSource<SubchannelPicker>(TaskCreationOptions.RunContinuationsAsynchronously);
         _resolverStartedTcs = new TaskCompletionSource<object?>(TaskCreationOptions.RunContinuationsAsynchronously);
         _channelId = _channelIdProvider.GetNextChannelId();
@@ -221,7 +221,8 @@ public async Task ConnectAsync(bool waitForReady, CancellationToken cancellation
         else
         {
             Task waitForReadyTask;
-            lock (_lock)
+            _lock.Enter();
+            try
             {
                 var state = State;
                 if (state == ConnectivityState.Ready)
@@ -232,6 +233,10 @@ public async Task ConnectAsync(bool waitForReady, CancellationToken cancellation
                 waitForReadyTask = WaitForStateChangedAsync(state, waitForState: ConnectivityState.Ready, cancellationToken);
                 _balancer?.RequestConnection();
             }
+            finally
+            {
+                _lock.Exit();
+            }
 
             await waitForReadyTask.ConfigureAwait(false);
         }

src/Grpc.Net.Client/Balancer/Internal/SocketConnectivitySubchannelTransport.cs

@@ -79,7 +79,7 @@ public SocketConnectivitySubchannelTransport(
         _socketConnectedTimer = NonCapturingTimer.Create(OnCheckSocketConnection, state: null, Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan);
     }
 
-    private object Lock => _subchannel.Lock;
+    private Lock Lock => _subchannel.Lock;
     public DnsEndPoint? CurrentEndPoint => _currentEndPoint;
     public TimeSpan? ConnectTimeout { get; }
     public TransportStatus TransportStatus
@@ -132,7 +132,7 @@ public void Disconnect()
 
     private void DisconnectUnsynchronized()
     {
-        Debug.Assert(Monitor.IsEntered(Lock));
+        Debug.Assert(Lock.IsHeldByCurrentThread);
         Debug.Assert(!_disposed);
 
         _initialSocket?.Dispose();
@@ -170,7 +170,8 @@ public async ValueTask<ConnectResult> TryConnectAsync(ConnectContext context, in
                 await _socketConnect(socket, currentEndPoint, context.CancellationToken).ConfigureAwait(false);
                 SocketConnectivitySubchannelTransportLog.ConnectedSocket(_logger, _subchannel.Id, currentEndPoint);
 
-                lock (Lock)
+                Lock.Enter();
+                try
                 {
                     _currentEndPoint = currentEndPoint;
                     _lastEndPointIndex = currentIndex;
@@ -184,6 +185,10 @@ public async ValueTask<ConnectResult> TryConnectAsync(ConnectContext context, in
                     // Instead, the socket timer target method reschedules the next run after it has finished.
                     _socketConnectedTimer.Change(_socketPingInterval, Timeout.InfiniteTimeSpan);
                 }
+                finally
+                {
+                    Lock.Exit();
+                }
 
                 _subchannel.UpdateConnectivityState(ConnectivityState.Ready, "Successfully connected to socket.");
                 return ConnectResult.Success;
@@ -223,13 +228,18 @@ public async ValueTask<ConnectResult> TryConnectAsync(ConnectContext context, in
         _subchannel.UpdateConnectivityState(
             ConnectivityState.TransientFailure,
             new Status(StatusCode.Unavailable, "Error connecting to subchannel.", firstConnectionError));
-        lock (Lock)
+        Lock.Enter();
+        try
         {
             if (!_disposed)
             {
                 _socketConnectedTimer.Change(Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan);
             }
         }
+        finally
+        {
+            Lock.Exit();
+        }
         return result;
     }
 
@@ -319,7 +329,8 @@ public async ValueTask<Stream> GetStreamAsync(DnsEndPoint endPoint, Cancellation
         DnsEndPoint? socketEndPoint = null;
         List<ReadOnlyMemory<byte>>? socketData = null;
         DateTime? socketCreatedTime = null;
-        lock (Lock)
+        Lock.Enter();
+        try
         {
             if (_initialSocket != null)
             {
@@ -345,6 +356,10 @@ public async ValueTask<Stream> GetStreamAsync(DnsEndPoint endPoint, Cancellation
                 _socketConnectedTimer.Change(Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan);
             }
         }
+        finally
+        {
+            Lock.Exit();
+        }
 
         if (socket != null)
         {
@@ -384,11 +399,16 @@ public async ValueTask<Stream> GetStreamAsync(DnsEndPoint endPoint, Cancellation
         // This stream wrapper intercepts dispose.
         var stream = new StreamWrapper(networkStream, OnStreamDisposed, socketData);
 
-        lock (Lock)
+        Lock.Enter();
+        try
         {
             _activeStreams.Add(new ActiveStream(endPoint, socket, stream));
             SocketConnectivitySubchannelTransportLog.StreamCreated(_logger, _subchannel.Id, endPoint, CalculateInitialSocketDataLength(socketData), _activeStreams.Count);
         }
+        finally
+        {
+            Lock.Exit();
+        }
 
         return stream;
     }

src/Grpc.Net.Client/Balancer/PollingResolver.cs

@@ -44,7 +44,7 @@ public abstract class PollingResolver : Resolver
     private bool _disposed;
     private bool _resolveSuccessful;
 
-    private readonly object _lock = new object();
+    private readonly Lock _lock = new Lock();
     private readonly CancellationTokenSource _cts = new CancellationTokenSource();
     private readonly ILogger _logger;
     private readonly IBackoffPolicyFactory? _backoffPolicyFactory;

src/Grpc.Net.Client/Balancer/Subchannel.cs

@@ -40,7 +40,7 @@ namespace Grpc.Net.Client.Balancer;
 public sealed class Subchannel : IDisposable
 {
     internal readonly List<BalancerAddress> _addresses;
-    internal readonly object Lock;
+    internal readonly Lock Lock;
     internal ISubchannelTransport Transport => _transport;
     internal string Id { get; }
 
@@ -96,7 +96,7 @@ public BalancerAddress? CurrentAddress
 
     internal Subchannel(ConnectionManager manager, IReadOnlyList<BalancerAddress> addresses)
     {
-        Lock = new object();
+        Lock = new Lock();
         _logger = manager.LoggerFactory.CreateLogger(GetType());
         _connectSemaphore = new SemaphoreSlim(1);
 
@@ -283,7 +283,7 @@ public void RequestConnection()
 
     private void CancelInProgressConnectUnsynchronized()
     {
-        Debug.Assert(Monitor.IsEntered(Lock));
+        Debug.Assert(Lock.IsHeldByCurrentThread);
 
         if (_connectContext != null && !_connectContext.Disposed)
         {
@@ -299,7 +299,7 @@ private void CancelInProgressConnectUnsynchronized()
 
     private ConnectContext GetConnectContextUnsynchronized()
     {
-        Debug.Assert(Monitor.IsEntered(Lock));
+        Debug.Assert(Lock.IsHeldByCurrentThread);
 
         // There shouldn't be a previous connect in progress, but cancel the CTS to ensure they're no longer running.
         CancelInProgressConnectUnsynchronized();
@@ -312,7 +312,8 @@ private async Task ConnectTransportAsync()
     {
         ConnectContext connectContext;
         Task? waitSemaporeTask = null;
-        lock (Lock)
+        Lock.Enter();
+        try
         {
             // Don't start connecting if the subchannel has been shutdown. Transport/semaphore will be disposed if shutdown.
             if (_state == ConnectivityState.Shutdown)
@@ -333,6 +334,10 @@ private async Task ConnectTransportAsync()
                 waitSemaporeTask = _connectSemaphore.WaitAsync(connectContext.CancellationToken);
             }
         }
+        finally
+        {
+            Lock.Exit();
+        }
 
         if (waitSemaporeTask != null)
         {
@@ -355,13 +360,18 @@ private async Task ConnectTransportAsync()
 
             for (var attempt = 0; ; attempt++)
             {
-                lock (Lock)
+                Lock.Enter();
+                try
                 {
                     if (_state == ConnectivityState.Shutdown)
                     {
                         return;
                     }
                 }
+                finally
+                {
+                    Lock.Exit();
+                }
 
                 switch (await _transport.TryConnectAsync(connectContext, attempt).ConfigureAwait(false))
                 {
@@ -425,7 +435,8 @@ private async Task ConnectTransportAsync()
         }
         finally
         {
-            lock (Lock)
+            Lock.Enter();
+            try
             {
                 // Dispose context because it might have been created with a connect timeout.
                 // Want to clean up the connect timeout timer.
@@ -438,6 +449,10 @@ private async Task ConnectTransportAsync()
                     _connectSemaphore.Release();
                 }
             }
+            finally
+            {
+                Lock.Exit();
+            }
         }
     }
 

src/Grpc.Net.Client/Grpc.Net.Client.csproj

@@ -6,7 +6,7 @@
 
     <IsGrpcPublishedPackage>true</IsGrpcPublishedPackage>
     <GenerateDocumentationFile>true</GenerateDocumentationFile>
-    <TargetFrameworks>net462;netstandard2.0;netstandard2.1;net6.0;net7.0;net8.0</TargetFrameworks>
+    <TargetFrameworks>net462;netstandard2.0;netstandard2.1;net6.0;net7.0;net8.0;net9.0</TargetFrameworks>
     <PackageReadmeFile>README.md</PackageReadmeFile>
   </PropertyGroup>
 

src/Grpc.Net.Client/GrpcChannel.cs

@@ -50,7 +50,7 @@ public sealed class GrpcChannel : ChannelBase, IDisposable
     internal const long DefaultMaxRetryBufferSize = 1024 * 1024 * 16; // 16 MB
     internal const long DefaultMaxRetryBufferPerCallSize = 1024 * 1024; // 1 MB
 
-    private readonly object _lock;
+    private readonly Lock _lock;
     private readonly ConcurrentDictionary<IMethod, GrpcMethodInfo> _methodInfoCache;
     private readonly Func<IMethod, GrpcMethodInfo> _createMethodInfoFunc;
     private readonly Dictionary<MethodKey, MethodConfig>? _serviceConfigMethods;
@@ -104,7 +104,7 @@ public sealed class GrpcChannel : ChannelBase, IDisposable
 
     internal GrpcChannel(Uri address, GrpcChannelOptions channelOptions) : base(address.Authority)
     {
-        _lock = new object();
+        _lock = new Lock();
         _methodInfoCache = new ConcurrentDictionary<IMethod, GrpcMethodInfo>();
 
         // Dispose the HTTP client/handler if...