Fix thread pool exhaustion with many MQTT writer groups (#3395)

* Initial plan * Convert PubSub PublishNetworkMessage to async to prevent thread pool exhaustion Co-authored-by: romanett <7413710+romanett@users.noreply.github.com> * Fix test files to properly use async/await patterns Co-authored-by: romanett <7413710+romanett@users.noreply.github.com> * Add Async suffix to PublishNetworkMessage method Co-authored-by: marcschier <11168470+marcschier@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: romanett <7413710+romanett@users.noreply.github.com> Co-authored-by: marcschier <11168470+marcschier@users.noreply.github.com>
2026-10-26 04:19:53 +08:00 · 2025-12-10 07:29:57 +01:00
parent 965c5ec522
commit e032dae878
10 changed files with 100 additions and 93 deletions
@@ -29,6 +29,7 @@

 using System;
 using System.Collections.Generic;
+using System.Threading.Tasks;

 namespace Opc.Ua.PubSub
 {
@@ -82,7 +83,7 @@ namespace Opc.Ua.PubSub
        /// <summary>
        /// Publish the network message
        /// </summary>
-        bool PublishNetworkMessage(UaNetworkMessage networkMessage);
+        Task<bool> PublishNetworkMessageAsync(UaNetworkMessage networkMessage);

        /// <summary>
        /// Get flag that indicates if all the network clients are connected
@@ -58,14 +58,14 @@ namespace Opc.Ua.PubSub
            object id,
            double interval,
            Func<bool> canExecuteFunc,
-            Action intervalAction,
+            Func<Task> intervalActionAsync,
            ITelemetryContext telemetry)
        {
            m_logger = telemetry.CreateLogger<IntervalRunner>();
            Id = id;
            Interval = interval;
            CanExecuteFunc = canExecuteFunc;
-            IntervalAction = intervalAction;
+            IntervalActionAsync = intervalActionAsync;
        }

        /// <summary>
@@ -99,9 +99,9 @@ namespace Opc.Ua.PubSub
        public Func<bool> CanExecuteFunc { get; }

        /// <summary>
-        /// Get the action that will be executed at each interval
+        /// Get the async action that will be executed at each interval
        /// </summary>
-        public Action IntervalAction { get; }
+        public Func<Task> IntervalActionAsync { get; }

        /// <summary>
        /// Starts the IntervalRunner and makes it ready to execute the code.
@@ -206,10 +206,10 @@ namespace Opc.Ua.PubSub
                    double nextCycle = (long)m_interval * HiResClock.TicksPerMillisecond;
                    m_nextPublishTick += nextCycle;

-                    if (IntervalAction != null && CanExecuteFunc != null && CanExecuteFunc())
+                    if (IntervalActionAsync != null && CanExecuteFunc != null && CanExecuteFunc())
                    {
-                        // call on a new thread
-                        Task.Run(() => IntervalAction());
+                        // call on a new task without blocking the thread pool
+                        _ = Task.Run(IntervalActionAsync);
                    }
                }
            }
@@ -28,6 +28,7 @@
 * ======================================================================*/

 using System;
+using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;

 namespace Opc.Ua.PubSub.Transport
@@ -65,7 +66,7 @@ namespace Opc.Ua.PubSub.Transport
                dataSetWriter.DataSetWriterId,
                metaDataUpdateTime,
                CanPublish,
-                PublishMessage,
+                PublishMessageAsync,
                telemetry);
        }

@@ -103,7 +104,7 @@ namespace Opc.Ua.PubSub.Transport
        /// <summary>
        /// Generate and publish the dataset MetaData message
        /// </summary>
-        private void PublishMessage()
+        private async Task PublishMessageAsync()
        {
            try
            {
@@ -113,7 +114,7 @@ namespace Opc.Ua.PubSub.Transport
                        m_dataSetWriter);
                if (metaDataNetworkMessage != null)
                {
-                    bool success = m_parentConnection.PublishNetworkMessage(metaDataNetworkMessage);
+                    bool success = await m_parentConnection.PublishNetworkMessageAsync(metaDataNetworkMessage).ConfigureAwait(false);
                    m_logger.LogInformation(
                        "MqttMetadataPublisher Publish DataSetMetaData, DataSetWriterId:{DataSetWriterId}; success = {Success}",
                        m_dataSetWriter.DataSetWriterId,
@@ -248,7 +248,7 @@ namespace Opc.Ua.PubSub.Transport
        /// <summary>
        /// Publish the network message
        /// </summary>
-        public override bool PublishNetworkMessage(UaNetworkMessage networkMessage)
+        public override async Task<bool> PublishNetworkMessageAsync(UaNetworkMessage networkMessage)
        {
            if (networkMessage == null || m_publisherMqttClient == null)
            {
@@ -257,76 +257,78 @@ namespace Opc.Ua.PubSub.Transport

            try
            {
+                IMqttClient publisherClient;
                lock (Lock)
                {
-                    if (m_publisherMqttClient.IsConnected)
+                    publisherClient = m_publisherMqttClient;
+                }
+
+                if (publisherClient != null && publisherClient.IsConnected)
+                {
+                    // get the encoded bytes
+                    byte[] bytes = networkMessage.Encode(MessageContext);
+
+                    try
                    {
-                        // get the encoded bytes
-                        byte[] bytes = networkMessage.Encode(MessageContext);
+                        string queueName = null;
+                        BrokerTransportQualityOfService qos
+                            = BrokerTransportQualityOfService.AtLeastOnce;

-                        try
+                        // the network messages that have DataSetWriterId are either metaData messages or SingleDataSet messages and
+                        if (networkMessage.DataSetWriterId != null)
                        {
-                            string queueName = null;
-                            BrokerTransportQualityOfService qos
-                                = BrokerTransportQualityOfService.AtLeastOnce;
+                            DataSetWriterDataType dataSetWriter =
+                                networkMessage.WriterGroupConfiguration.DataSetWriters.Find(x =>
+                                    x.DataSetWriterId == networkMessage.DataSetWriterId);

-                            // the network messages that have DataSetWriterId are either metaData messages or SingleDataSet messages and
-                            if (networkMessage.DataSetWriterId != null)
+                            if (dataSetWriter != null &&
+                                ExtensionObject.ToEncodeable(dataSetWriter.TransportSettings)
+                                    is BrokerDataSetWriterTransportDataType transportSettings)
                            {
-                                DataSetWriterDataType dataSetWriter =
-                                    networkMessage.WriterGroupConfiguration.DataSetWriters.Find(x =>
-                                        x.DataSetWriterId == networkMessage.DataSetWriterId);
+                                qos = transportSettings.RequestedDeliveryGuarantee;

-                                if (dataSetWriter != null &&
-                                    ExtensionObject.ToEncodeable(dataSetWriter.TransportSettings)
-                                        is BrokerDataSetWriterTransportDataType transportSettings)
+                                queueName = networkMessage.IsMetaDataMessage
+                                    ? transportSettings.MetaDataQueueName
+                                    : transportSettings.QueueName;
+                            }
+                        }
+
+                        if (queueName == null ||
+                            qos == BrokerTransportQualityOfService.NotSpecified)
+                        {
+                            if (ExtensionObject.ToEncodeable(
+                                    networkMessage.WriterGroupConfiguration.TransportSettings)
+                                is BrokerWriterGroupTransportDataType transportSettings)
+                            {
+                                queueName ??= transportSettings.QueueName;
+                                // if the value is not specified and the value of the parent object shall be used
+                                if (qos == BrokerTransportQualityOfService.NotSpecified)
                                {
                                    qos = transportSettings.RequestedDeliveryGuarantee;
-
-                                    queueName = networkMessage.IsMetaDataMessage
-                                        ? transportSettings.MetaDataQueueName
-                                        : transportSettings.QueueName;
                                }
                            }
-
-                            if (queueName == null ||
-                                qos == BrokerTransportQualityOfService.NotSpecified)
-                            {
-                                if (ExtensionObject.ToEncodeable(
-                                        networkMessage.WriterGroupConfiguration.TransportSettings)
-                                    is BrokerWriterGroupTransportDataType transportSettings)
-                                {
-                                    queueName ??= transportSettings.QueueName;
-                                    // if the value is not specified and the value of the parent object shall be used
-                                    if (qos == BrokerTransportQualityOfService.NotSpecified)
-                                    {
-                                        qos = transportSettings.RequestedDeliveryGuarantee;
-                                    }
-                                }
-                            }
-
-                            if (!string.IsNullOrEmpty(queueName))
-                            {
-                                var message = new MqttApplicationMessage
-                                {
-                                    Topic = queueName,
-                                    PayloadSegment = new ArraySegment<byte>(bytes),
-                                    QualityOfServiceLevel = GetMqttQualityOfServiceLevel(qos),
-                                    Retain = networkMessage.IsMetaDataMessage
-                                };
-
-                                m_publisherMqttClient.PublishAsync(message).GetAwaiter()
-                                    .GetResult();
-                            }
                        }
-                        catch (Exception ex)
+
+                        if (!string.IsNullOrEmpty(queueName))
                        {
-                            m_logger.LogError(ex, "MqttPubSubConnection.PublishNetworkMessage");
-                            return false;
-                        }
+                            var message = new MqttApplicationMessage
+                            {
+                                Topic = queueName,
+                                PayloadSegment = new ArraySegment<byte>(bytes),
+                                QualityOfServiceLevel = GetMqttQualityOfServiceLevel(qos),
+                                Retain = networkMessage.IsMetaDataMessage
+                            };

-                        return true;
+                            await publisherClient.PublishAsync(message).ConfigureAwait(false);
+                        }
                    }
+                    catch (Exception ex)
+                    {
+                        m_logger.LogError(ex, "MqttPubSubConnection.PublishNetworkMessage");
+                        return false;
+                    }
+
+                    return true;
                }
            }
            catch (Exception ex)
@@ -217,7 +217,7 @@ namespace Opc.Ua.PubSub.Transport
                            "UdpDiscoveryPublisher.SendResponseDataSetMetaData before sending message for DataSetWriterId:{DataSetWriterId}",
                            message.DataSetWriterId);

-                        m_udpConnection.PublishNetworkMessage(message);
+                        m_udpConnection.PublishNetworkMessageAsync(message);
                    }
                    m_metadataWriterIdsToSend.Clear();
                }
@@ -251,7 +251,7 @@ namespace Opc.Ua.PubSub.Transport
                            "UdpDiscoveryPublisher.SendResponseDataSetWriterConfiguration Before sending message for DataSetWriterId:{DataSetWriterId}",
                            responsesMessage.DataSetWriterId);

-                        m_udpConnection.PublishNetworkMessage(responsesMessage);
+                        m_udpConnection.PublishNetworkMessageAsync(responsesMessage);
                    }
                }
            }
@@ -280,7 +280,7 @@ namespace Opc.Ua.PubSub.Transport
                m_logger.LogInformation(
                    "UdpDiscoveryPublisher.SendResponsePublisherEndpoints before sending message for PublisherEndpoints.");

-                m_udpConnection.PublishNetworkMessage(message);
+                m_udpConnection.PublishNetworkMessageAsync(message);
            }
        }

@@ -65,7 +65,7 @@ namespace Opc.Ua.PubSub.Transport
                udpConnection.PubSubConnectionConfiguration.Name,
                kInitialRequestInterval,
                CanPublish,
-                RequestDiscoveryMessages,
+                RequestDiscoveryMessagesAsync,
                telemetry);
        }

@@ -280,10 +280,11 @@ namespace Opc.Ua.PubSub.Transport
        /// <summary>
        /// Joint task to request discovery messages
        /// </summary>
-        private void RequestDiscoveryMessages()
+        private Task RequestDiscoveryMessagesAsync()
        {
            SendDiscoveryRequestDataSetMetaData();
            SendDiscoveryRequestDataSetWriterConfiguration();
+            return Task.CompletedTask;
        }
    }
 }
@@ -416,13 +416,13 @@ namespace Opc.Ua.PubSub.Transport
        /// <summary>
        /// Publish the network message
        /// </summary>
-        public override bool PublishNetworkMessage(UaNetworkMessage networkMessage)
+        public override Task<bool> PublishNetworkMessageAsync(UaNetworkMessage networkMessage)
        {
            if (networkMessage == null ||
                m_publisherUdpClients == null ||
                m_publisherUdpClients.Count == 0)
            {
-                return false;
+                return Task.FromResult(false);
            }

            try
@@ -448,20 +448,20 @@ namespace Opc.Ua.PubSub.Transport
                            catch (Exception ex)
                            {
                                m_logger.LogError(ex, "UdpPubSubConnection.PublishNetworkMessage");
-                                return false;
+                                return Task.FromResult(false);
                            }
                        }
-                        return true;
+                        return Task.FromResult(true);
                    }
                }
            }
            catch (Exception ex)
            {
                m_logger.LogError(ex, "UdpPubSubConnection.PublishNetworkMessage");
-                return false;
+                return Task.FromResult(false);
            }

-            return false;
+            return Task.FromResult(false);
        }

        /// <summary>
@@ -226,7 +226,7 @@ namespace Opc.Ua.PubSub
        /// </summary>
        /// <param name="networkMessage">The network message that needs to be published.</param>
        /// <returns>True if send was successful.</returns>
-        public abstract bool PublishNetworkMessage(UaNetworkMessage networkMessage);
+        public abstract Task<bool> PublishNetworkMessageAsync(UaNetworkMessage networkMessage);

        /// <summary>
        /// Get flag that indicates if all the network clients are connected
@@ -30,6 +30,7 @@
 using System;
 using System.Collections.Generic;
 using System.Threading;
+using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;

 namespace Opc.Ua.PubSub
@@ -68,7 +69,7 @@ namespace Opc.Ua.PubSub
                WriterGroupConfiguration.Name,
                WriterGroupConfiguration.PublishingInterval,
                CanPublish,
-                PublishMessages,
+                PublishMessagesAsync,
                telemetry);
        }

@@ -143,7 +144,7 @@ namespace Opc.Ua.PubSub
        /// <summary>
        /// Generate and publish the messages
        /// </summary>
-        private void PublishMessages()
+        private async Task PublishMessagesAsync()
        {
            try
            {
@@ -156,7 +157,7 @@ namespace Opc.Ua.PubSub
                    {
                        if (uaNetworkMessage != null)
                        {
-                            bool success = PubSubConnection.PublishNetworkMessage(uaNetworkMessage);
+                            bool success = await PubSubConnection.PublishNetworkMessageAsync(uaNetworkMessage).ConfigureAwait(false);
                            m_logger.LogDebug(
                                "UaPublisher - PublishNetworkMessage, WriterGroupId:{WriterGroupId}; success = {Success}",
                                WriterGroupConfiguration.WriterGroupId,
@@ -33,6 +33,7 @@ using System.Linq;
 using System.Net;
 using System.Net.Sockets;
 using System.Threading;
+using System.Threading.Tasks;
 using NUnit.Framework;
 using Opc.Ua.PubSub.Configuration;
 using Opc.Ua.PubSub.Transport;
@@ -49,7 +50,7 @@ namespace Opc.Ua.PubSub.Tests.Transport
 #if !CUSTOM_TESTS
        [Ignore("A network interface controller is necessary in order to run correctly.")]
 #endif
-        public void ValidateUdpPubSubConnectionNetworkMessagePublishUnicast()
+        public async Task ValidateUdpPubSubConnectionNetworkMessagePublishUnicast()
        {
            ITelemetryContext telemetry = NUnitTelemetryContext.Create();

@@ -120,7 +121,7 @@ namespace Opc.Ua.PubSub.Tests.Transport
                {
                    if (uaNetworkMessage != null)
                    {
-                        publisherConnection.PublishNetworkMessage(uaNetworkMessage);
+                        await publisherConnection.PublishNetworkMessageAsync(uaNetworkMessage).ConfigureAwait(false);
                    }
                }
            }
@@ -147,7 +148,7 @@ namespace Opc.Ua.PubSub.Tests.Transport
 #if !CUSTOM_TESTS
        [Ignore("A network interface controller is necessary in order to run correctly.")]
 #endif
-        public void ValidateUdpPubSubConnectionNetworkMessagePublishBroadcast()
+        public async Task ValidateUdpPubSubConnectionNetworkMessagePublishBroadcast()
        {
            ITelemetryContext telemetry = NUnitTelemetryContext.Create();     //Arrange
            System.Net.NetworkInformation.UnicastIPAddressInformation localhost = GetFirstNic();
@@ -217,7 +218,7 @@ namespace Opc.Ua.PubSub.Tests.Transport
                {
                    if (uaNetworkMessage != null)
                    {
-                        publisherConnection.PublishNetworkMessage(uaNetworkMessage);
+                        await publisherConnection.PublishNetworkMessageAsync(uaNetworkMessage).ConfigureAwait(false);
                    }
                }
            }
@@ -244,7 +245,7 @@ namespace Opc.Ua.PubSub.Tests.Transport
 #if !CUSTOM_TESTS
        [Ignore("A network interface controller is necessary in order to run correctly.")]
 #endif
-        public void ValidateUdpPubSubConnectionNetworkMessagePublishMulticast()
+        public async Task ValidateUdpPubSubConnectionNetworkMessagePublishMulticast()
        {
            ITelemetryContext telemetry = NUnitTelemetryContext.Create();

@@ -317,7 +318,7 @@ namespace Opc.Ua.PubSub.Tests.Transport
                {
                    if (uaNetworkMessage != null)
                    {
-                        publisherConnection.PublishNetworkMessage(uaNetworkMessage);
+                        await publisherConnection.PublishNetworkMessageAsync(uaNetworkMessage).ConfigureAwait(false);
                    }
                }
            }
@@ -345,7 +346,7 @@ namespace Opc.Ua.PubSub.Tests.Transport
 #if !CUSTOM_TESTS
        [Ignore("A network interface controller is necessary in order to run correctly.")]
 #endif
-        public void ValidateUdpPubSubConnectionNetworkMessageDiscoveryPublish_DataSetMetadata()
+        public async Task ValidateUdpPubSubConnectionNetworkMessageDiscoveryPublish_DataSetMetadata()
        {
            ITelemetryContext telemetry = NUnitTelemetryContext.Create();

@@ -424,7 +425,7 @@ namespace Opc.Ua.PubSub.Tests.Transport
                {
                    if (uaNetworkMessage != null)
                    {
-                        publisherConnection.PublishNetworkMessage(uaNetworkMessage);
+                        await publisherConnection.PublishNetworkMessageAsync(uaNetworkMessage).ConfigureAwait(false);
                    }
                }
            }
@@ -451,7 +452,7 @@ namespace Opc.Ua.PubSub.Tests.Transport
 #if !CUSTOM_TESTS
        [Ignore("A network interface controller is necessary in order to run correctly.")]
 #endif
-        public void ValidateUdpPubSubConnectionNetworkMessageDiscoveryPublish_DataSetWriterConfiguration()
+        public async Task ValidateUdpPubSubConnectionNetworkMessageDiscoveryPublish_DataSetWriterConfiguration()
        {
            ITelemetryContext telemetry = NUnitTelemetryContext.Create();

@@ -526,7 +527,7 @@ namespace Opc.Ua.PubSub.Tests.Transport

            if (networkMessage != null)
            {
-                publisherConnection.PublishNetworkMessage(networkMessage);
+                await publisherConnection.PublishNetworkMessageAsync(networkMessage).ConfigureAwait(false);
            }

            //Assert
@@ -552,7 +553,7 @@ namespace Opc.Ua.PubSub.Tests.Transport
 #if !CUSTOM_TESTS
        [Ignore("A network interface controller is necessary in order to run correctly.")]
 #endif
-        public void ValidateUdpPubSubConnectionNetworkMessageDiscoveryPublish_PublisherEndpoints()
+        public async Task ValidateUdpPubSubConnectionNetworkMessageDiscoveryPublish_PublisherEndpoints()
        {
            ITelemetryContext telemetry = NUnitTelemetryContext.Create();

@@ -654,7 +655,7 @@ namespace Opc.Ua.PubSub.Tests.Transport
            //Act
            publisherConnection.Start();

-            publisherConnection.PublishNetworkMessage(uaNetworkMessage);
+            await publisherConnection.PublishNetworkMessageAsync(uaNetworkMessage).ConfigureAwait(false);

            // Assert
            bool noMessageReceived = false;