Concept

SDK internals

Subscriptions

Introduction

How quickly a client learns about changes in the server depends on many factors. Besides the configuration of a subscription and its MonitoredItems (essentially pairs of node and attribute), the setup of server and client plays a major role. The hardware and software configuration is more of an administrative factor. If the implementation is based on wrong assumptions, however, this shows up during testing at best and only in production at worst. At that point, an adjusted configuration rarely works as a workaround.

This article explains how the OPC UA .NET SDK processes subscription notifications internally. It focuses on details of the OPC UA specification, the resulting processing in the SDK, and the points where your own code (user code) plugs into this processing. The article shows typical pitfalls and best practices. With this knowledge, you can improve how your application scales when it communicates over OPC UA and increase its throughput.

Scaling, timing, and threading

The same conditions apply to processing notifications, whether a subscription monitors events (as in "OPC UA Alarms & Events") or data changes. Data changes are values, status codes, and timestamps (as in "OPC UA Data Access"). The question of the best possible scaling mainly arises when you aim for the best possible data throughput. You can measure data throughput under idealized conditions, but the result says little about your own application. The reason is the large number of possible implementations and use cases, which a lab setup cannot reproduce and therefore only touches on. When communicating, a client application only sends requests and receives responses. In a server application, it is the other way around. How the application further processes requests and responses determines how well it can scale in its use case.

With timing, too, every tenth of a second rightly counts. Experience shows, however, that time measurements are often too imprecise, quickly lose the context of what they measure, and depend on too many external variables. The timing behavior derived from them therefore often describes an application at the macro level. It hardly reflects the channels, algorithms, and techniques of communication and data processing in detail. What matters most for the desired timing behavior is step-by-step data processing across processing layers that are as independent as possible. You can implement this with patterns such as the chain of responsibility, among others.

An easy-to-use API, scenario-specific objects, ready-to-run samples, and copy-and-paste development get you to your goal quickly. In terms of software design, however, the result is more of a proof of concept (PoC). In production, it does not hold up against a well-designed software architecture. Threading is where it gets difficult: a common assumption is that more threads mean more performance. Scheduling in operating systems and CPUs keeps getting better. This progress is lost, however, if applications handle the system resources they get carelessly. Deliberate and transparent threading behavior is often the key to a scalable application that offers the best possible timing behavior for its requirements.

Message processing

To see where you can improve, you need a basic understanding of how the SDK processes subscription messages. The starting point is therefore a single client, that is, one instance of the OpcClient class. It connects to an OPC UA server and creates a subscription with one MonitoredItem. A MonitoredItem is a pair of node and attribute that the subscription monitors.

As soon as the subscription and its MonitoredItem exist, the server stores in a cache all changes and events that fall within the grid of the SamplingInterval used. At the same time, the client uses a separate thread to send a Publish request within the PublishingInterval. It requests the values and events that have been added to the cache since the last Publish request. In its Publish response, the server sends all values and events stored since the last request.

In OPC UA, this package of all values and events from the cache is called a NotificationMessage and is sent as such. A NotificationMessage consists of NotificationMessageItems, one item per value or event. A NotificationMessage can contain items for several MonitoredItems, not just for one specific MonitoredItem. In response to its Publish requests, the client therefore receives a sequence of NotificationMessages.

The client splits each NotificationMessage it receives into its items and dispatches them to the EventHandlers for data changes and events of the subscription's MonitoredItems. By default, a thread from the .NET thread pool performs this dispatching. This lets the client receive and dispatch the next NotificationMessage as quickly as possible. The SDK dispatches the NotificationMessageItems of each subscription sequentially and so keeps the order of the changes in the server. The SDK raises the resulting events, that is, the calls of the EventHandlers, per subscription. Your EventHandlers of several subscriptions therefore run in parallel.

With threading, a common question is which thread dispatches the items. There are three cases:

In a worker thread

By default, the SDK dispatches NotificationMessages using threads from the .NET thread pool. The number of these threads depends on the hardware, among other things. While dispatching, the client can receive further NotificationMessages in parallel. The SDK always dispatches the items of each message sequentially. After dispatching one NotificationMessage, it is possible that the same thread immediately receives the next NotificationMessage that has arrived in the meantime. The word "possible" is deliberate: whether the .NET CLR provides a different thread or the same thread from the thread pool for dispatching is not determined.

In the UI thread

In an application with a graphical user interface, for example with Windows Forms (WinForms) or Windows Presentation Foundation (WPF), the SDK runs the EventHandlers of a subscription through the current SynchronizationContext. It also dispatches the NotificationMessageItems through this context. For this, the SDK determines the current SynchronizationContext when you create an instance of OpcSubscription and assigns it to the subscription for its entire lifetime. The SDK then dispatches the NotificationMessageItems that the subscription receives through this SynchronizationContext.

In graphical applications, the SynchronizationContext refers to the UI thread by default. This thread draws the user interface and responds to input. If the SDK dispatches the NotificationMessageItems through the UI thread, you can update the user interface directly in the EventHandler without manual synchronization. You can access controls there directly, for example to update the display with the new information. Another advantage: the SDK adds the dispatching as a job to the MessageQueue of the UI thread. The UI thread therefore also updates the user interface regularly and is not busy with dispatching alone. This keeps the user interface responsive.

In a custom thread

In some situations, you want a different thread or a thread of your choice to dispatch the NotificationMessages. For this, you can set your own scheduler for such background tasks with OpcTaskSchedule.CustomScheduler. If you set a callback there that hands the background action to a new thread, for example, dispatching explicitly takes place in a new thread. A CustomScheduler you set takes precedence over an existing SynchronizationContext (the second case).

These cases might suggest that the SDK always uses different threads. As the previous sections show, however, there is no simple yes or no answer. The answer depends on how your application sets up the SDK and on the load, that is, the number of NotificationMessages to expect. Always providing your own thread with OpcTaskSchedule.CustomScheduler only looks like a quick solution at first glance. Processing, in this case dispatching the NotificationMessages, does not necessarily get faster with more threads. One reason is the higher scheduling effort for the CPU, which degrades overall performance instead of improving it.

Communication timing

As described at the beginning, timing is relative. You therefore cannot assess it in absolute terms with simple numbers in idealized environments. To still get a feel for the timing of your application and its setup, the following cases start with the simplest networked infrastructure and become more complex step by step.

One client, one server

Computer A runs the client software, computer B the server software. The two computers are connected directly over LAN. If both computers have the same technical specifications, the client computer learns about changes in the server about as quickly as possible and can process them just as quickly.

Multiple clients, multiple servers

If N client computers connect one-to-one to N server computers, in theory the timing is the same whether there is one client computer or N. This requires each client-server pair to be connected directly. As soon as communication runs through a switch, a router, a VPN, or something similar, this alone affects the timing behavior between the client-server pairs.

One client, multiple servers

If N clients run on one client computer and connect to N server computers, many more factors that affect timing behavior add to those of the second case. Apart from the network adapter, this case only looks at a client application with N instances of OpcClient that connect to N servers. Each of these instances uses:

• its own thread to request and receive the NotificationMessages
• by default, the .NET thread pool to dispatch the NotificationMessageItems to the EventHandlers of the MonitoredItems
• for processing the information from the server:
  • either the same thread that another client used before
  • or a different thread
  • or a new thread

On a low-performance computer with few free resources, the .NET thread pool can therefore become the bottleneck. This degrades timing behavior when dispatching and processing the NotificationMessages.

Apart from the .NET thread pool, your own downstream processing chain for data changes and events plays a decisive role in timing. Actions in a thread from the .NET thread pool should ideally never block (for too long). They should therefore only work on CPU and memory and strictly avoid access to resources such as the file system, network, or databases. Such access also degrades timing behavior when further NotificationMessages are dispatched. If a worker thread (a thread from the .NET thread pool) blocks for too long, it can also starve the entire .NET thread pool. The reason: its threads also run the continuations of asynchronous operations and tasks.

Recommendations

For the best possible timing behavior, we therefore recommend running only CPU and memory operations in the handler methods of client, subscription, and MonitoredItems. Use the handler methods mainly to evaluate notifications, classify them, and add them to a queue of your own. A separate thread of your own then takes the notifications from this queue and processes them by your own criteria. This decouples processing from communication, and you determine timing behavior and scaling within your own software architecture.

Process all notifications

The following samples show what data processing in the client can look like. As always, the starting point is an active connection with a subscription:

C#

using (var client = new OpcClient("<address>")) {
    client.Connect();
    client.SubscribeDataChange("<nodeId>", HandleDataChanged);

    ...
}

VB.NET

Using client = New OpcClient("<address>")
    client.Connect()
    client.SubscribeDataChange("<nodeId>", AddressOf HandleDataChanged)

    ...
End Using

The EventHandler only fills the queue through which the threads exchange data:

C#

private static readonly BlockingCollection<string> data = new BlockingCollection<string>();

VB.NET

Private Shared ReadOnly data As New BlockingCollection(Of String)()

The BlockingCollection shown uses a ConcurrentQueue by default. In addition, it blocks as long as the collection contains no entry. If you use a ConcurrentQueue or another collection directly instead of a BlockingCollection, you must ensure two things. First, access is thread-safe, so that several threads do not access it at the same time. Second, the processing thread blocks as long as there is no entry. Otherwise, the processing thread runs at full CPU load. This is how the sample fills the queue:

C#

private static void HandleDataChanged(object sender, OpcDataChangeReceivedEventArgs e)
{
    data.Add(e.Item.Value.As<string>());
}

VB.NET

Private Shared Sub HandleDataChanged(ByVal sender As Object, ByVal e As OpcDataChangeReceivedEventArgs)
    data.Add(e.Item.Value.As(Of String)())
End Sub

The following code snippet defines and starts the thread that processes the data collection:

C#

var processDataThread = new Thread(ProcessData);
processDataThread.Start();

VB.NET

Dim processDataThread = New Thread(AddressOf ProcessData)
processDataThread.Start()

After it starts, the separate thread runs the following method. It retrieves all data from the collection and blocks the ForEach loop when there is no more data in the collection. The loop continues as soon as a new entry arrives. Only when CompleteAdding is called on the collection does the enumerator end the loop, after all pending entries have been processed.

C#

private static void ProcessData()
{
    // Using 'GetConsumingEnumerable' removes an item from the collection, consumes it through
    // the foreach loop and blocks until another item is/has being added to collection or
    // 'CompleteAdding' is used to signal the end of the data processing.
    foreach (var item in data.GetConsumingEnumerable()) {
        // Insert your data processing here.
    }
}

VB.NET

Private Shared Sub ProcessData()
    'Using 'GetConsumingEnumerable' removes an item from the collection, consumes it through
    'the For Each loop and blocks until another item is/has being added to collection or
    ''CompleteAdding' is used to signal the end of the data processing.
    For Each item As String In data.GetConsumingEnumerable()
        'Insert your data processing here.
    Next
End Sub

When you exit the application or close the connection to the server, the following code ends data processing:

C#

data.CompleteAdding();
processDataThread.Join();

VB.NET

data.CompleteAdding()
processDataThread.Join()

The sample shown processes all received data and ends processing only when the BlockingCollection contains no more entries. The call of processDataThread.Join() therefore blocks until the thread has processed all entries and emptied the data collection completely.

Process with a cancellation condition

Data processing in the client can also run only until a certain condition occurs. A CancellationToken, for example, can signal that the condition has occurred. The following snippets show a client with a cancellation condition.

Here, too, an active connection with a subscription is the prerequisite:

C#

using (var client = new OpcClient("<address>")) {
    client.Connect();
    client.SubscribeDataChange("<nodeId>", HandleDataChanged);

    ...
}

VB.NET

Using client = New OpcClient("<address>")
    client.Connect()
    client.SubscribeDataChange("<nodeId>", AddressOf HandleDataChanged)

    ...
End Using

The EventHandler still only fills the queue through which the threads exchange data:

C#

private static readonly BlockingCollection<string> data = new BlockingCollection<string>();

VB.NET

Private Shared ReadOnly data As New BlockingCollection(Of String)()

The EventHandler also still fills the queue:

C#

private static void HandleDataChanged(object sender, OpcDataChangeReceivedEventArgs e)
{
    data.Add(e.Item.Value.As<string>());
}

VB.NET

Private Shared Sub HandleDataChanged(ByVal sender As Object, ByVal e As OpcDataChangeReceivedEventArgs)
    data.Add(e.Item.Value.As(Of String)())
End Sub

The following code snippet defines and starts the thread with a CancellationToken. Through the CancellationToken, the application signals to the thread that it should stop processing any data that may still be pending. The thread's task is still to process the data from the collection:

C#

var processData = new CancellationTokenSource();
var processDataThread = new Thread(() => ProcessData(processData.Token));

processDataThread.Start();

VB.NET

Dim processDataCancellation = New CancellationTokenSource()
Dim processDataThread = New Thread(Sub() ProcessData(processDataCancellation.Token))

processDataThread.Start()

After it starts, the separate thread runs the following method. It retrieves the data from the collection and processes it. The call data.Take(...) blocks until the collection contains an entry or the CancellationToken signals cancellation. On cancellation, Take throws an OperationCanceledException. The While loop then ends because the CancellationToken signals cancellation. The loop checks this both after an exception and after processing an entry.

C#

private static void ProcessData(CancellationToken token)
{
    while (!token.IsCancellationRequested) {
        try {
            // Insert your data processing here.
            Debug.WriteLine(data.Take(token));
        }
        catch (OperationCanceledException) {
            // Can be ignored, because of the while loop will exit the method.
        }
    }
}

VB.NET

Private Shared Sub ProcessData(ByVal token As CancellationToken)
    While Not token.IsCancellationRequested
        Try
            'Insert your data processing here.
            Debug.WriteLine(data.Take(token))
        Catch ex As OperationCanceledException
            'Can be ignored, because of the while loop will exit the method.
        End Try
    End While
End Sub

When you exit the application, disconnect from the server, or want to end processing immediately because of a cancellation condition, the following code informs the thread:

C#

processData.Cancel();
processDataThread.Join();

VB.NET

processDataCancellation.Cancel()
processDataThread.Join()

The sample shown processes the data in the collection until the cancellation condition occurs. The thread no longer processes data that was received and added to the queue in the meantime. That data remains in the collection.

Consulting and support

The processing and samples described here help you implement a scalable subscription. If you process the received values and events correctly and in a way that fits your use case, you keep full control over the timing behavior of your application. If you have questions about the implementation and the right approach, our development team is happy to help at support@traeger.de.