Replace locker with Exchange

danielmarbach · lukebakken · commit 57346a8a8c08 · 2025-01-07T12:48:41.000-08:00
diff --git a/projects/RabbitMQ.Client/Impl/AutorecoveringChannel.cs b/projects/RabbitMQ.Client/Impl/AutorecoveringChannel.cs
@@ -48,8 +48,7 @@ internal sealed class AutorecoveringChannel : IChannel, IRecoverable
         private AutorecoveringConnection _connection;
         private RecoveryAwareChannel _innerChannel;
         private bool _disposed;
-        private bool _isDisposing;
-        private readonly object _locker = new();
+        private int _isDisposing;
 
         private ushort _prefetchCountConsumer;
         private ushort _prefetchCountGlobal;
@@ -266,20 +265,11 @@ public void Dispose()
 
         public async ValueTask DisposeAsync()
         {
-            if (_disposed)
+            if (Interlocked.Exchange(ref _isDisposing, 1) != 0)
             {
                 return;
             }
 
-            lock (_locker)
-            {
-                if (_isDisposing)
-                {
-                    return;
-                }
-                _isDisposing = true;
-            }
-
             try
             {
                 if (IsOpen)
@@ -293,7 +283,6 @@ await this.AbortAsync()
             finally
             {
                 _disposed = true;
-                _isDisposing = false;
             }
         }
 
diff --git a/projects/RabbitMQ.Client/Impl/AutorecoveringConnection.cs b/projects/RabbitMQ.Client/Impl/AutorecoveringConnection.cs
@@ -50,8 +50,7 @@ internal sealed partial class AutorecoveringConnection : IConnection
 
         private Connection _innerConnection;
         private bool _disposed;
-        private bool _isDisposing;
-        private readonly object _locker = new();
+        private int _isDisposing;
 
         private Connection InnerConnection
         {
@@ -282,20 +281,11 @@ public void Dispose()
 
         public async ValueTask DisposeAsync()
         {
-            if (_disposed)
+            if (Interlocked.Exchange(ref _isDisposing, 1) != 0)
             {
                 return;
             }
 
-            lock (_locker)
-            {
-                if (_isDisposing)
-                {
-                    return;
-                }
-                _isDisposing = true;
-            }
-
             try
             {
                 await _innerConnection.DisposeAsync()
@@ -313,7 +303,6 @@ await _innerConnection.DisposeAsync()
             finally
             {
                 _disposed = true;
-                _isDisposing = false;
             }
         }
 
diff --git a/projects/RabbitMQ.Client/Impl/Channel.cs b/projects/RabbitMQ.Client/Impl/Channel.cs
@@ -64,9 +64,7 @@ internal partial class Channel : IChannel, IRecoverable
         internal readonly IConsumerDispatcher ConsumerDispatcher;
 
         private bool _disposed;
-        private bool _isDisposing;
-
-        private readonly object _locker = new();
+        private int _isDisposing;
 
         public Channel(ISession session, CreateChannelOptions createChannelOptions)
         {
@@ -531,20 +529,11 @@ void IDisposable.Dispose()
 
         protected virtual void Dispose(bool disposing)
         {
-            if (_disposed)
+            if (Interlocked.Exchange(ref _isDisposing, 1) != 0)
             {
                 return;
             }
 
-            lock (_locker)
-            {
-                if (_isDisposing)
-                {
-                    return;
-                }
-                _isDisposing = true;
-            }
-
             if (disposing)
             {
                 try
@@ -554,10 +543,7 @@ protected virtual void Dispose(bool disposing)
                         this.AbortAsync().GetAwaiter().GetResult();
                     }
 
-                    if (_serverOriginatedChannelCloseTcs is not null)
-                    {
-                        _serverOriginatedChannelCloseTcs.Task.Wait(TimeSpan.FromSeconds(5));
-                    }
+                    _serverOriginatedChannelCloseTcs?.Task.Wait(TimeSpan.FromSeconds(5));
 
                     ConsumerDispatcher.Dispose();
                     _rpcSemaphore.Dispose();
@@ -567,7 +553,6 @@ protected virtual void Dispose(bool disposing)
                 finally
                 {
                     _disposed = true;
-                    _isDisposing = false;
                 }
             }
         }
@@ -582,20 +567,11 @@ await DisposeAsyncCore()
 
         protected virtual async ValueTask DisposeAsyncCore()
         {
-            if (_disposed)
+            if (Interlocked.Exchange(ref _isDisposing, 1) != 0)
             {
                 return;
             }
 
-            lock (_locker)
-            {
-                if (_isDisposing)
-                {
-                    return;
-                }
-                _isDisposing = true;
-            }
-
             try
             {
                 if (IsOpen)
@@ -621,7 +597,6 @@ await _outstandingPublisherConfirmationsRateLimiter.DisposeAsync()
             finally
             {
                 _disposed = true;
-                _isDisposing = false;
             }
         }
 
@@ -718,9 +693,11 @@ protected virtual ulong AdjustDeliveryTag(ulong deliveryTag)
 
         protected async Task<bool> HandleChannelCloseAsync(IncomingCommand cmd, CancellationToken cancellationToken)
         {
-            lock (_locker)
+            var serverOriginatedChannelCloseTcs = _serverOriginatedChannelCloseTcs;
+            if (serverOriginatedChannelCloseTcs is null)
             {
-                _serverOriginatedChannelCloseTcs ??= new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+                // Attempt to assign the new TCS only if _tcs is still null
+                _ = Interlocked.CompareExchange(ref _serverOriginatedChannelCloseTcs, new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously), null);
             }
 
             try
diff --git a/projects/RabbitMQ.Client/Impl/Connection.cs b/projects/RabbitMQ.Client/Impl/Connection.cs
@@ -46,8 +46,7 @@ namespace RabbitMQ.Client.Framing
     internal sealed partial class Connection : IConnection
     {
         private bool _disposed;
-        private bool _isDisposing;
-        private readonly object _locker = new();
+        private int _isDisposing;
         private volatile bool _closed;
 
         private readonly ConnectionConfig _config;
@@ -499,20 +498,11 @@ public void Dispose()
 
         public async ValueTask DisposeAsync()
         {
-            if (_disposed)
+            if (Interlocked.Exchange(ref _isDisposing, 1) != 0)
             {
                 return;
             }
 
-            lock (_locker)
-            {
-                if (_isDisposing)
-                {
-                    return;
-                }
-                _isDisposing = true;
-            }
-
             try
             {
                 if (IsOpen)
@@ -534,7 +524,6 @@ await _channel0.DisposeAsync()
             finally
             {
                 _disposed = true;
-                _isDisposing = false;
             }
         }
 

Original file line number	Diff line number	Diff line change
`@@ -48,8 +48,7 @@ internal sealed class AutorecoveringChannel : IChannel, IRecoverable`
`48`	`48`	`private AutorecoveringConnection _connection;`
`49`	`49`	`private RecoveryAwareChannel _innerChannel;`
`50`	`50`	`private bool _disposed;`
`51`		`- private bool _isDisposing;`
`52`		`- private readonly object _locker = new();`
	`51`	`+ private int _isDisposing;`
`53`	`52`
`54`	`53`	`private ushort _prefetchCountConsumer;`
`55`	`54`	`private ushort _prefetchCountGlobal;`
`@@ -266,20 +265,11 @@ public void Dispose()`
`266`	`265`
`267`	`266`	`public async ValueTask DisposeAsync()`
`268`	`267`	`{`
`269`		`- if (_disposed)`
	`268`	`+ if (Interlocked.Exchange(ref _isDisposing, 1) != 0)`
`270`	`269`	`{`
`271`	`270`	`return;`
`272`	`271`	`}`
`273`	`272`
`274`		`- lock (_locker)`
`275`		`- {`
`276`		`- if (_isDisposing)`
`277`		`- {`
`278`		`- return;`
`279`		`- }`
`280`		`- _isDisposing = true;`
`281`		`- }`
`282`		`-`
`283`	`273`	`try`
`284`	`274`	`{`
`285`	`275`	`if (IsOpen)`
`@@ -293,7 +283,6 @@ await this.AbortAsync()`
`293`	`283`	`finally`
`294`	`284`	`{`
`295`	`285`	`_disposed = true;`
`296`		`- _isDisposing = false;`
`297`	`286`	`}`
`298`	`287`	`}`
`299`	`288`
Original file line number	Diff line number	Diff line change
`@@ -50,8 +50,7 @@ internal sealed partial class AutorecoveringConnection : IConnection`
`50`	`50`
`51`	`51`	`private Connection _innerConnection;`
`52`	`52`	`private bool _disposed;`
`53`		`- private bool _isDisposing;`
`54`		`- private readonly object _locker = new();`
	`53`	`+ private int _isDisposing;`
`55`	`54`
`56`	`55`	`private Connection InnerConnection`
`57`	`56`	`{`
`@@ -282,20 +281,11 @@ public void Dispose()`
`282`	`281`
`283`	`282`	`public async ValueTask DisposeAsync()`
`284`	`283`	`{`
`285`		`- if (_disposed)`
	`284`	`+ if (Interlocked.Exchange(ref _isDisposing, 1) != 0)`
`286`	`285`	`{`
`287`	`286`	`return;`
`288`	`287`	`}`
`289`	`288`
`290`		`- lock (_locker)`
`291`		`- {`
`292`		`- if (_isDisposing)`
`293`		`- {`
`294`		`- return;`
`295`		`- }`
`296`		`- _isDisposing = true;`
`297`		`- }`
`298`		`-`
`299`	`289`	`try`
`300`	`290`	`{`
`301`	`291`	`await _innerConnection.DisposeAsync()`
`@@ -313,7 +303,6 @@ await _innerConnection.DisposeAsync()`
`313`	`303`	`finally`
`314`	`304`	`{`
`315`	`305`	`_disposed = true;`
`316`		`- _isDisposing = false;`
`317`	`306`	`}`
`318`	`307`	`}`
`319`	`308`
Original file line number	Diff line number	Diff line change
`@@ -64,9 +64,7 @@ internal partial class Channel : IChannel, IRecoverable`
`64`	`64`	`internal readonly IConsumerDispatcher ConsumerDispatcher;`
`65`	`65`
`66`	`66`	`private bool _disposed;`
`67`		`- private bool _isDisposing;`
`68`		`-`
`69`		`- private readonly object _locker = new();`
	`67`	`+ private int _isDisposing;`
`70`	`68`
`71`	`69`	`public Channel(ISession session, CreateChannelOptions createChannelOptions)`
`72`	`70`	`{`
`@@ -531,20 +529,11 @@ void IDisposable.Dispose()`
`531`	`529`
`532`	`530`	`protected virtual void Dispose(bool disposing)`
`533`	`531`	`{`
`534`		`- if (_disposed)`
	`532`	`+ if (Interlocked.Exchange(ref _isDisposing, 1) != 0)`
`535`	`533`	`{`
`536`	`534`	`return;`
`537`	`535`	`}`
`538`	`536`
`539`		`- lock (_locker)`
`540`		`- {`
`541`		`- if (_isDisposing)`
`542`		`- {`
`543`		`- return;`
`544`		`- }`
`545`		`- _isDisposing = true;`
`546`		`- }`
`547`		`-`
`548`	`537`	`if (disposing)`
`549`	`538`	`{`
`550`	`539`	`try`
`@@ -554,10 +543,7 @@ protected virtual void Dispose(bool disposing)`
`554`	`543`	`this.AbortAsync().GetAwaiter().GetResult();`
`555`	`544`	`}`
`556`	`545`
`557`		`- if (_serverOriginatedChannelCloseTcs is not null)`
`558`		`- {`
`559`		`- _serverOriginatedChannelCloseTcs.Task.Wait(TimeSpan.FromSeconds(5));`
`560`		`- }`
	`546`	`+ _serverOriginatedChannelCloseTcs?.Task.Wait(TimeSpan.FromSeconds(5));`
`561`	`547`
`562`	`548`	`ConsumerDispatcher.Dispose();`
`563`	`549`	`_rpcSemaphore.Dispose();`
`@@ -567,7 +553,6 @@ protected virtual void Dispose(bool disposing)`
`567`	`553`	`finally`
`568`	`554`	`{`
`569`	`555`	`_disposed = true;`
`570`		`- _isDisposing = false;`
`571`	`556`	`}`
`572`	`557`	`}`
`573`	`558`	`}`
`@@ -582,20 +567,11 @@ await DisposeAsyncCore()`
`582`	`567`
`583`	`568`	`protected virtual async ValueTask DisposeAsyncCore()`
`584`	`569`	`{`
`585`		`- if (_disposed)`
	`570`	`+ if (Interlocked.Exchange(ref _isDisposing, 1) != 0)`
`586`	`571`	`{`
`587`	`572`	`return;`
`588`	`573`	`}`
`589`	`574`
`590`		`- lock (_locker)`
`591`		`- {`
`592`		`- if (_isDisposing)`
`593`		`- {`
`594`		`- return;`
`595`		`- }`
`596`		`- _isDisposing = true;`
`597`		`- }`
`598`		`-`
`599`	`575`	`try`
`600`	`576`	`{`
`601`	`577`	`if (IsOpen)`
`@@ -621,7 +597,6 @@ await _outstandingPublisherConfirmationsRateLimiter.DisposeAsync()`
`621`	`597`	`finally`
`622`	`598`	`{`
`623`	`599`	`_disposed = true;`
`624`		`- _isDisposing = false;`
`625`	`600`	`}`
`626`	`601`	`}`
`627`	`602`
`@@ -718,9 +693,11 @@ protected virtual ulong AdjustDeliveryTag(ulong deliveryTag)`
`718`	`693`
`719`	`694`	`protected async Task<bool> HandleChannelCloseAsync(IncomingCommand cmd, CancellationToken cancellationToken)`
`720`	`695`	`{`
`721`		`- lock (_locker)`
	`696`	`+ var serverOriginatedChannelCloseTcs = _serverOriginatedChannelCloseTcs;`
	`697`	`+ if (serverOriginatedChannelCloseTcs is null)`
`722`	`698`	`{`
`723`		`- _serverOriginatedChannelCloseTcs ??= new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);`
	`699`	`+ // Attempt to assign the new TCS only if _tcs is still null`
	`700`	`+ _ = Interlocked.CompareExchange(ref _serverOriginatedChannelCloseTcs, new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously), null);`
`724`	`701`	`}`
`725`	`702`
`726`	`703`	`try`
Original file line number	Diff line number	Diff line change
`@@ -46,8 +46,7 @@ namespace RabbitMQ.Client.Framing`
`46`	`46`	`internal sealed partial class Connection : IConnection`
`47`	`47`	`{`
`48`	`48`	`private bool _disposed;`
`49`		`- private bool _isDisposing;`
`50`		`- private readonly object _locker = new();`
	`49`	`+ private int _isDisposing;`
`51`	`50`	`private volatile bool _closed;`
`52`	`51`
`53`	`52`	`private readonly ConnectionConfig _config;`
`@@ -499,20 +498,11 @@ public void Dispose()`
`499`	`498`
`500`	`499`	`public async ValueTask DisposeAsync()`
`501`	`500`	`{`
`502`		`- if (_disposed)`
	`501`	`+ if (Interlocked.Exchange(ref _isDisposing, 1) != 0)`
`503`	`502`	`{`
`504`	`503`	`return;`
`505`	`504`	`}`
`506`	`505`
`507`		`- lock (_locker)`
`508`		`- {`
`509`		`- if (_isDisposing)`
`510`		`- {`
`511`		`- return;`
`512`		`- }`
`513`		`- _isDisposing = true;`
`514`		`- }`
`515`		`-`
`516`	`506`	`try`
`517`	`507`	`{`
`518`	`508`	`if (IsOpen)`
`@@ -534,7 +524,6 @@ await _channel0.DisposeAsync()`
`534`	`524`	`finally`
`535`	`525`	`{`
`536`	`526`	`_disposed = true;`
`537`		`- _isDisposing = false;`
`538`	`527`	`}`
`539`	`528`	`}`
`540`	`529`