Close connections asynchronously and improve gracefulness (#7006)

* Close connections asynchronously and improve gracefulness (#6762)

* Lock around connection cleanup so we don't call .Complete() and .CancelPendingFlush()/.CancelPendingRead() on the same pipe which results in an ObjectDisposedException

* Close connections asynchronously and improve gracefulness

Co-authored-by: ReubenBond <[email protected]>

* Adding await for GatewayListNotification

* Updating GatewayManager interfaces

Co-authored-by: John Morman <[email protected]>
Co-authored-by: ReubenBond <[email protected]>

Author: 3 people

src/Orleans.Connections.Security/Security/DuplexPipeStream.cs

@@ -20,7 +20,7 @@ public DuplexPipeStreamAdapter(IDuplexPipe duplexPipe, Func<Stream, TStream> cre
         }
 
         public DuplexPipeStreamAdapter(IDuplexPipe duplexPipe, StreamPipeReaderOptions readerOptions, StreamPipeWriterOptions writerOptions, Func<Stream, TStream> createStream) :
-            base(duplexPipe.Input, duplexPipe.Output)
+            base(duplexPipe)
         {
             var stream = createStream(this);
             Stream = stream;

src/Orleans.Connections.Security/Security/DuplexPipeStreamAdapter.cs

@@ -143,7 +143,7 @@ await sslStream.AuthenticateAsClientAsync(
                 }
                 catch (Exception ex)
                 {
-                    _logger?.LogWarning(1, ex, "Authentication failed: {Exception}", ex);
+                    _logger?.LogWarning(1, ex, "Authentication failed");
 #if NETCOREAPP
                     await sslStream.DisposeAsync();
 #else

src/Orleans.Connections.Security/Security/TlsClientConnectionMiddleware.cs

@@ -186,7 +186,7 @@ await sslStream.AuthenticateAsServerAsync(
                 }
                 catch (Exception ex)
                 {
-                    _logger?.LogWarning(1, ex, "Authentication failed: {Exception}", ex);
+                    _logger?.LogWarning(1, ex, "Authentication failed");
 #if NETCOREAPP
                     await sslStream.DisposeAsync();
 #else

src/Orleans.Connections.Security/Security/TlsServerConnectionMiddleware.cs

@@ -234,11 +234,11 @@ internal void ExpediteUpdateLiveGatewaysSnapshot()
         }
 
         [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes")]
-        public void GatewayListNotification(IEnumerable<Uri> gateways)
+        public async Task GatewayListNotification(IEnumerable<Uri> gateways)
         {
             try
             {
-                UpdateLiveGatewaysSnapshot(gateways.Select(gw => gw.ToGatewayAddress()), gatewayListProvider.MaxStaleness);
+                await UpdateLiveGatewaysSnapshot(gateways.Select(gw => gw.ToGatewayAddress()), gatewayListProvider.MaxStaleness);
             }
             catch (Exception exc)
             {
@@ -261,8 +261,7 @@ internal async Task RefreshSnapshotLiveGateways_TimerCallback(object context)
                     logger.LogDebug("Discovered {GatewayCount} gateways: {Gateways}", refreshedGateways.Count, Utils.EnumerableToString(refreshedGateways));
                 }
 
-                // the next one will grab the lock.
-                UpdateLiveGatewaysSnapshot(refreshedGateways, gatewayListProvider.MaxStaleness);
+                await UpdateLiveGatewaysSnapshot(refreshedGateways, gatewayListProvider.MaxStaleness);
             }
             catch (Exception exc)
             {
@@ -271,13 +270,14 @@ internal async Task RefreshSnapshotLiveGateways_TimerCallback(object context)
         }
 
         // This function is called asynchronously from gateway refresh timer.
-        private void UpdateLiveGatewaysSnapshot(IEnumerable<SiloAddress> refreshedGateways, TimeSpan maxStaleness)
+        private async Task UpdateLiveGatewaysSnapshot(IEnumerable<SiloAddress> refreshedGateways, TimeSpan maxStaleness)
         {
-            // this is a short lock, protecting the access to knownDead, knownMasked and cachedLiveGateways.
+            List<SiloAddress> connectionsToKeepAlive;
+
+            // This is a short lock, protecting the access to knownDead, knownMasked and cachedLiveGateways.
             lock (lockable)
             {
                 // now take whatever listProvider gave us and exclude those we think are dead.
-
                 var live = new List<SiloAddress>();
                 var now = DateTime.UtcNow;
 
@@ -340,8 +340,8 @@ private void UpdateLiveGatewaysSnapshot(IEnumerable<SiloAddress> refreshedGatewa
                 if (logger.IsEnabled(LogLevel.Information))
                 {
                     logger.Info(ErrorCode.GatewayManager_FoundKnownGateways,
-                            "Refreshed the live Gateway list. Found {0} gateways from Gateway listProvider: {1}. Picked only known live out of them. Now has {2} live Gateways: {3}. Previous refresh time was = {4}",
-                                knownGateways.Count,
+                            "Refreshed the live gateway list. Found {0} gateways from gateway list provider: {1}. Picked only known live out of them. Now has {2} live gateways: {3}. Previous refresh time was = {4}",
+                            knownGateways.Count,
                             Utils.EnumerableToString(knownGateways),
                             cachedLiveGateways.Count,
                             Utils.EnumerableToString(cachedLiveGateways),
@@ -351,13 +351,14 @@ private void UpdateLiveGatewaysSnapshot(IEnumerable<SiloAddress> refreshedGatewa
                 // Close connections to known dead connections, but keep the "masked" ones.
                 // Client will not send any new request to the "masked" connections, but might still
                 // receive responses
-                var connectionsToKeepAlive = new List<SiloAddress>(live);
+                connectionsToKeepAlive = new List<SiloAddress>(live);
                 connectionsToKeepAlive.AddRange(knownMasked.Select(e => e.Key));
-                this.CloseEvictedGatewayConnections(connectionsToKeepAlive);
             }
+
+            await this.CloseEvictedGatewayConnections(connectionsToKeepAlive);
         }
 
-        private void CloseEvictedGatewayConnections(List<SiloAddress> liveGateways)
+        private async Task CloseEvictedGatewayConnections(List<SiloAddress> liveGateways)
         {
             if (this.connectionManager == null) return;
 
@@ -381,7 +382,7 @@ private void CloseEvictedGatewayConnections(List<SiloAddress> liveGateways)
                         this.logger.LogInformation("Closing connection to {Endpoint} because it has been marked as dead", address);
                     }
 
-                    this.connectionManager.Close(address);
+                    await this.connectionManager.CloseAsync(address);
                 }
             }
         }

src/Orleans.Core/Messaging/GatewayManager.cs

@@ -36,7 +36,7 @@ public interface IGatewayListProvider
     /// </summary>
     public interface IGatewayListListener
     {
-        void GatewayListNotification(IEnumerable<Uri> gateways);
+        Task GatewayListNotification(IEnumerable<Uri> gateways);
     }
 
     /// <summary>