From e032dae87832d442ddce6be2e7b35a1e63293170 Mon Sep 17 00:00:00 2001 From: Copilot <198982749+Copilot@users.noreply.github.com> Date: Wed, 10 Dec 2025 07:29:57 +0100 Subject: [PATCH] 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> --- .../Opc.Ua.PubSub/IUaPubSubConnection.cs | 3 +- Libraries/Opc.Ua.PubSub/IntervalRunner.cs | 14 +-- .../Transport/MqttMetadataPublisher.cs | 7 +- .../Transport/MqttPubSubConnection.cs | 112 +++++++++--------- .../Transport/UdpDiscoveryPublisher.cs | 6 +- .../Transport/UdpDiscoverySubscriber.cs | 5 +- .../Transport/UdpPubSubConnection.cs | 12 +- Libraries/Opc.Ua.PubSub/UaPubSubConnection.cs | 2 +- Libraries/Opc.Ua.PubSub/UaPublisher.cs | 7 +- .../UdpPubSubConnectionTests.Publisher.cs | 25 ++-- 10 files changed, 100 insertions(+), 93 deletions(-) diff --git a/Libraries/Opc.Ua.PubSub/IUaPubSubConnection.cs b/Libraries/Opc.Ua.PubSub/IUaPubSubConnection.cs index b1a929828e..ffc5d9cc24 100644 --- a/Libraries/Opc.Ua.PubSub/IUaPubSubConnection.cs +++ b/Libraries/Opc.Ua.PubSub/IUaPubSubConnection.cs @@ -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 /// /// Publish the network message /// - bool PublishNetworkMessage(UaNetworkMessage networkMessage); + Task PublishNetworkMessageAsync(UaNetworkMessage networkMessage); /// /// Get flag that indicates if all the network clients are connected diff --git a/Libraries/Opc.Ua.PubSub/IntervalRunner.cs b/Libraries/Opc.Ua.PubSub/IntervalRunner.cs index f233281e43..6c74d42ff8 100644 --- a/Libraries/Opc.Ua.PubSub/IntervalRunner.cs +++ b/Libraries/Opc.Ua.PubSub/IntervalRunner.cs @@ -58,14 +58,14 @@ namespace Opc.Ua.PubSub object id, double interval, Func canExecuteFunc, - Action intervalAction, + Func intervalActionAsync, ITelemetryContext telemetry) { m_logger = telemetry.CreateLogger(); Id = id; Interval = interval; CanExecuteFunc = canExecuteFunc; - IntervalAction = intervalAction; + IntervalActionAsync = intervalActionAsync; } /// @@ -99,9 +99,9 @@ namespace Opc.Ua.PubSub public Func CanExecuteFunc { get; } /// - /// Get the action that will be executed at each interval + /// Get the async action that will be executed at each interval /// - public Action IntervalAction { get; } + public Func IntervalActionAsync { get; } /// /// 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); } } } diff --git a/Libraries/Opc.Ua.PubSub/Transport/MqttMetadataPublisher.cs b/Libraries/Opc.Ua.PubSub/Transport/MqttMetadataPublisher.cs index c8d0bc91bc..6b1693d566 100644 --- a/Libraries/Opc.Ua.PubSub/Transport/MqttMetadataPublisher.cs +++ b/Libraries/Opc.Ua.PubSub/Transport/MqttMetadataPublisher.cs @@ -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 /// /// Generate and publish the dataset MetaData message /// - 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, diff --git a/Libraries/Opc.Ua.PubSub/Transport/MqttPubSubConnection.cs b/Libraries/Opc.Ua.PubSub/Transport/MqttPubSubConnection.cs index 6366d6926b..24875ee1ca 100644 --- a/Libraries/Opc.Ua.PubSub/Transport/MqttPubSubConnection.cs +++ b/Libraries/Opc.Ua.PubSub/Transport/MqttPubSubConnection.cs @@ -248,7 +248,7 @@ namespace Opc.Ua.PubSub.Transport /// /// Publish the network message /// - public override bool PublishNetworkMessage(UaNetworkMessage networkMessage) + public override async Task 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(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(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) diff --git a/Libraries/Opc.Ua.PubSub/Transport/UdpDiscoveryPublisher.cs b/Libraries/Opc.Ua.PubSub/Transport/UdpDiscoveryPublisher.cs index 7b3431d30d..71a4211451 100644 --- a/Libraries/Opc.Ua.PubSub/Transport/UdpDiscoveryPublisher.cs +++ b/Libraries/Opc.Ua.PubSub/Transport/UdpDiscoveryPublisher.cs @@ -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); } } diff --git a/Libraries/Opc.Ua.PubSub/Transport/UdpDiscoverySubscriber.cs b/Libraries/Opc.Ua.PubSub/Transport/UdpDiscoverySubscriber.cs index c7ecb96daa..cfb9a3c997 100644 --- a/Libraries/Opc.Ua.PubSub/Transport/UdpDiscoverySubscriber.cs +++ b/Libraries/Opc.Ua.PubSub/Transport/UdpDiscoverySubscriber.cs @@ -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 /// /// Joint task to request discovery messages /// - private void RequestDiscoveryMessages() + private Task RequestDiscoveryMessagesAsync() { SendDiscoveryRequestDataSetMetaData(); SendDiscoveryRequestDataSetWriterConfiguration(); + return Task.CompletedTask; } } } diff --git a/Libraries/Opc.Ua.PubSub/Transport/UdpPubSubConnection.cs b/Libraries/Opc.Ua.PubSub/Transport/UdpPubSubConnection.cs index 2ae7bce2d5..c84cca4d65 100644 --- a/Libraries/Opc.Ua.PubSub/Transport/UdpPubSubConnection.cs +++ b/Libraries/Opc.Ua.PubSub/Transport/UdpPubSubConnection.cs @@ -416,13 +416,13 @@ namespace Opc.Ua.PubSub.Transport /// /// Publish the network message /// - public override bool PublishNetworkMessage(UaNetworkMessage networkMessage) + public override Task 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); } /// diff --git a/Libraries/Opc.Ua.PubSub/UaPubSubConnection.cs b/Libraries/Opc.Ua.PubSub/UaPubSubConnection.cs index ddc580a5c7..ed88b28472 100644 --- a/Libraries/Opc.Ua.PubSub/UaPubSubConnection.cs +++ b/Libraries/Opc.Ua.PubSub/UaPubSubConnection.cs @@ -226,7 +226,7 @@ namespace Opc.Ua.PubSub /// /// The network message that needs to be published. /// True if send was successful. - public abstract bool PublishNetworkMessage(UaNetworkMessage networkMessage); + public abstract Task PublishNetworkMessageAsync(UaNetworkMessage networkMessage); /// /// Get flag that indicates if all the network clients are connected diff --git a/Libraries/Opc.Ua.PubSub/UaPublisher.cs b/Libraries/Opc.Ua.PubSub/UaPublisher.cs index 5f85355181..cc3959b60b 100644 --- a/Libraries/Opc.Ua.PubSub/UaPublisher.cs +++ b/Libraries/Opc.Ua.PubSub/UaPublisher.cs @@ -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 /// /// Generate and publish the messages /// - 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, diff --git a/Tests/Opc.Ua.PubSub.Tests/Transport/UdpPubSubConnectionTests.Publisher.cs b/Tests/Opc.Ua.PubSub.Tests/Transport/UdpPubSubConnectionTests.Publisher.cs index 5e47879012..5add8cc9ce 100644 --- a/Tests/Opc.Ua.PubSub.Tests/Transport/UdpPubSubConnectionTests.Publisher.cs +++ b/Tests/Opc.Ua.PubSub.Tests/Transport/UdpPubSubConnectionTests.Publisher.cs @@ -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;