fix(client): Adding graceful shutdown to event loop

e2e/test/iothub/service/IoTHubCertificateValidationE2ETest.cs

@@ -13,6 +13,7 @@ namespace Microsoft.Azure.Devices.E2ETests.Iothub.Service
 {
     [TestClass]
     [TestCategory("InvalidServiceCertificate")]
+    [Ignore] // Removed due to invalid cert service being shut down
     public class IoTHubCertificateValidationE2ETest : E2EMsTestBase
     {
         [LoggedTestMethod]

e2e/test/provisioning/ProvisioningCertificateValidationE2ETest.cs

@@ -14,6 +14,7 @@ namespace Microsoft.Azure.Devices.E2ETests.Provisioning
 {
     [TestClass]
     [TestCategory("InvalidServiceCertificate")]
+    [Ignore] // Removed due to invalid cert service being shut down
     public class ProvisioningCertificateValidationE2ETest : E2EMsTestBase
     {
         [LoggedTestMethod]

iothub/device/src/Transport/Mqtt/MqttTransportHandler.cs

@@ -94,10 +94,16 @@ internal sealed class MqttTransportHandler : TransportHandler, IMqttIotHubEventH
         private const string ReceiveEventMessagePrefixPattern = "devices/{0}/modules/{1}/";
 
         private static readonly int s_generationPrefixLength = Guid.NewGuid().ToString().Length;
-        private static readonly Lazy<IEventLoopGroup> s_eventLoopGroup = new Lazy<IEventLoopGroup>(GetEventLoopGroup);
+        private static readonly ReferenceCounter<IEventLoopGroup> _eventLoopGroupRefCounted = new ReferenceCounter<IEventLoopGroup>();
         private static readonly TimeSpan s_regexTimeoutMilliseconds = TimeSpan.FromMilliseconds(500);
         private static readonly TimeSpan s_defaultTwinTimeout = TimeSpan.FromSeconds(60);
 
+        // Recommended use of DotNetty is to use a single event loop group. In order to allow more flexibility we can let the caller decice if they want to 
+        // use a single event loop group, or possibly use a group per instance of the client.
+        private readonly bool _useSingleEventLoopGroup;
+        private readonly IEventLoopGroup _eventLoopGroup;
+        private readonly TimeSpan _timeoutValueForEventLoop = TimeSpan.Zero;
+
         private readonly string _generationId = Guid.NewGuid().ToString();
         private readonly string _receiveEventMessageFilter;
         private readonly string _receiveEventMessagePrefix;
@@ -178,6 +184,20 @@ internal MqttTransportHandler(
 
             _webProxy = settings.Proxy;
 
+            // If the customer chooses multiple groups we need to honor that here.
+            _useSingleEventLoopGroup = settings.UseSingleEventLoopGroup;
+
+            _timeoutValueForEventLoop = settings.GracefulEventLoopShutdownTimeout;
+
+            if (_useSingleEventLoopGroup)
+            {
+                _eventLoopGroup = _eventLoopGroupRefCounted.Create(GetEventLoopGroup);
+            }
+            else
+            {
+                _eventLoopGroup = GetEventLoopGroup();
+            }
+
             if (channelFactory != null)
             {
                 _channelFactory = channelFactory;
@@ -476,9 +496,15 @@ protected override void Dispose(bool disposing)
                     disposableChannel.Dispose();
                     _channel = null;
                 }
+
             }
         }
 
+        /// <summary>
+        /// Closes the MqttTransport
+        /// </summary>
+        /// <param name="cancellationToken">A cancellation token.</param>
+        /// <returns>An awaitable task</returns>
         public override async Task CloseAsync(CancellationToken cancellationToken)
         {
             try
@@ -491,7 +517,6 @@ public override async Task CloseAsync(CancellationToken cancellationToken)
                 if (TryStop())
                 {
                     OnTransportClosedGracefully();
-
                     await _closeRetryPolicy.ExecuteAsync(CleanUpImplAsync, cancellationToken).ConfigureAwait(true);
                 }
                 else if (State == TransportState.Error)
@@ -506,6 +531,21 @@ public override async Task CloseAsync(CancellationToken cancellationToken)
             }
         }
 
+        /// <summary>
+        /// Internal implementation of the close logic that can be reused in Dispose()
+        /// </summary>
+        /// <returns></returns>
+        private Task ShutdownEventLoopGroupInternal()
+        {
+            // If the customer chooses multiple groups we need to honor that here.
+            // The remove method will only return an object if the reference count is zero or if there is no value for the object
+            if (_useSingleEventLoopGroup && _eventLoopGroupRefCounted.Remove() == null)
+            {
+                return TaskHelpers.CompletedTask;
+            }
+            return _eventLoopGroup?.ShutdownGracefullyAsync(_timeoutValueForEventLoop, _timeoutValueForEventLoop);
+        }
+
         #endregion Client operations
 
         #region MQTT callbacks
@@ -1037,6 +1077,8 @@ private async Task OpenInternalAsync(CancellationToken cancellationToken)
                     {
                         await _channel.CloseAsync().ConfigureAwait(true);
                     }
+
+                    await ShutdownEventLoopGroupInternal();
                 });
             }
 
@@ -1209,7 +1251,7 @@ private Func<IPAddress[], int, Task<IChannel>> CreateChannelFactory(IotHubConnec
                      iotHubConnectionString.HostName);
 
                 Bootstrap bootstrap = new Bootstrap()
-                    .Group(s_eventLoopGroup.Value)
+                    .Group(_eventLoopGroup)
                     .Channel<TcpSocketChannel>()
                     .Option(ChannelOption.TcpNodelay, true)
                     .Option(ChannelOption.Allocator, UnpooledByteBufferAllocator.Default)
@@ -1319,7 +1361,7 @@ private Func<IPAddress[], int, Task<IChannel>> CreateWebSocketChannelFactory(Iot
                         new LoggingHandler(LogLevel.DEBUG),
                         _mqttIotHubAdapterFactory.Create(this, iotHubConnectionString, settings, productInfo, options));
 
-                await s_eventLoopGroup.Value.RegisterAsync(clientWebSocketChannel).ConfigureAwait(true);
+                await _eventLoopGroup.RegisterAsync(clientWebSocketChannel).ConfigureAwait(true);
 
                 return clientWebSocketChannel;
             };
@@ -1420,4 +1462,4 @@ private bool IsProxyConfigured()
                 && _webProxy != DefaultWebProxySettings.Instance;
         }
     }
-}
+}

iothub/device/src/Transport/Mqtt/MqttTransportSettings.cs

@@ -250,5 +250,21 @@ public TransportType GetTransportType()
         /// Used by Edge runtime to specify an authentication chain for Edge-to-Edge connections.
         /// </summary>
         internal string AuthenticationChain { get; set; }
+
+        /// <summary>
+        /// The DotNetty event loop GracefulShutdown timeout.
+        /// </summary>
+        /// <remarks>
+        /// The DotNetty Mqtt transport has a graceful shutdown of the event loop that can block the <see cref="DeviceClient.CloseAsync(System.Threading.CancellationToken)"/> method. This timeout allows you to configure how long the shutdown event will wait for before forcefully shutting down the event loop. Change this if you need your application to respond to the CloseAsync command faster.
+        /// </remarks>
+        public TimeSpan GracefulEventLoopShutdownTimeout { get; set; } = TimeSpan.FromSeconds(1);
+
+        /// <summary>
+        /// Sets the MqttTransport to use a single Event Loop Group.
+        /// </summary>
+        /// <remarks>
+        /// This setting is to retain legacy behavior that will use a single Event Loop Group which is the thread pool model for DotNetty. If this is set to false a new group will be created per instance of the Device Client.
+        /// </remarks>
+        public bool UseSingleEventLoopGroup { get; set; } = true;
     }
 }

provisioning/transport/mqtt/src/ProvisioningTransportHandlerMqtt.cs

@@ -32,7 +32,7 @@ namespace Microsoft.Azure.Devices.Provisioning.Client.Transport
     /// </summary>
     public class ProvisioningTransportHandlerMqtt : ProvisioningTransportHandler
     {
-        private static readonly MultithreadEventLoopGroup s_eventLoopGroup = new MultithreadEventLoopGroup();
+        private readonly MultithreadEventLoopGroup s_eventLoopGroup = new MultithreadEventLoopGroup();
 
         // TODO: Unify these constants with IoT Hub Device client.
         private const int MaxMessageSize = 256 * 1024;
@@ -169,6 +169,7 @@ protected override void Dispose(bool disposing)
         {
             if (disposing)
             {
+                s_eventLoopGroup.ShutdownGracefullyAsync(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(1));
                 _webSocketChannel?.Dispose();
                 _webSocketChannel = null;
             }