# Abläufe im SDK

> Konzept: Interne Abläufe des OPC UA .NET SDK bei Subscriptions: Skalierung, Timing und Threading der Nachrichten und Empfehlungen für die eigene Anwendung.

## Subscriptions

### Einführung

Wie schnell ein Client von Änderungen im Server erfährt, hängt von vielen Faktoren ab. Neben der Konfiguration einer Subscription und ihrer MonitoredItems (im Kern Paare aus Node und Attribut) spielt das Setup von Server und Client eine große Rolle. Die Hard- und Softwarekonfiguration ist dabei eher ein administrativer Faktor. Wenn die Implementierung jedoch von falschen Annahmen ausgeht, zeigt sich das im besten Fall noch in der Testphase, im schlechtesten Fall erst im Produktivbetrieb. Eine angepasste Konfiguration reicht als Workaround dann nur selten aus.

Dieser Artikel erklärt, wie das OPC UA .NET SDK Benachrichtigungen von Subscriptions intern verarbeitet. Im Fokus stehen Details der OPC-UA-Spezifikation, die daraus folgenden Abläufe im SDK und die Stellen, an denen eigener Code (User Code) in diese Abläufe eingebunden ist. Der Artikel zeigt typische Fallstricke und bewährte Vorgehensweisen (Best Practices). Damit können Sie die Skalierung Ihrer Anwendung bei der Kommunikation über OPC UA verbessern und den Durchsatz steigern.

### Skalierung, Timing und Threading

Für die Verarbeitung von Benachrichtigungen gelten dieselben Bedingungen, unabhängig davon, ob eine Subscription Ereignisse (Stichwort „OPC UA Alarme & Events“) oder Datenänderungen überwacht. Datenänderungen sind Werte, Status und Zeitstempel (Stichwort „OPC UA Data Access“). Die Frage nach der bestmöglichen **Skalierung** stellt sich vor allem beim bestmöglichen Datendurchsatz. Der Datendurchsatz lässt sich zwar unter idealisierten Bedingungen messen, sagt aber wenig über die eigene Anwendung aus. Der Grund sind die vielen möglichen Implementierungen und Anwendungsfälle, die ein Laboraufbau nicht nachstellen kann und deshalb nur streift. Bei der Kommunikation versendet eine Clientanwendung ausschließlich Anfragen (engl. Requests) und empfängt Antworten (engl. Responses). In einer Serveranwendung ist es umgekehrt. Wie die Anwendung Anfragen und Antworten weiterverarbeitet, entscheidet über ihre mögliche Skalierung im jeweiligen Anwendungsszenario.

Auch beim **Timing** zählt zu Recht jede Zehntelsekunde. Erfahrungsgemäß ist die Zeitmessung jedoch oft zu ungenau, verliert schnell den eigentlichen Kontext der Messung und hängt von zu vielen externen Variablen ab. Das daraus abgeleitete Timing-Verhalten einer Anwendung beschreibt deshalb oft eher die Makroebene. Die Kanäle, Algorithmen und Techniken der Kommunikation und Datenverarbeitung im Detail bildet es kaum ab. Entscheidend für das gewünschte Timing-Verhalten ist vor allem eine schrittweise Datenverarbeitung über möglichst unabhängige Verarbeitungsebenen. Das lässt sich unter anderem mit Mustern wie der [Zuständigkeitskette](https://de.wikipedia.org/wiki/Zust%C3%A4ndigkeitskette) umsetzen.

Eine leicht handhabbare API, szenariobezogene Objekte, sofort lauffähige Beispiele (Ready-to-Run) und Entwicklung per Copy and Paste führen schnell zum Ziel. Softwaretechnisch ist das Ergebnis jedoch eher ein Proof of Concept (PoC). Im Produktivbetrieb hält es den Vergleich mit einer durchdachten Softwarearchitektur nicht stand. Schwierig wird es beim **Threading**: Hier herrscht oft die Annahme, dass mehr Threads auch mehr Leistung bringen. Das Scheduling von Betriebssystemen und CPUs wird zwar immer besser. Dieser Fortschritt verpufft aber, wenn Anwendungen achtlos mit den bereitgestellten Systemressourcen umgehen. Ein bewusstes und transparentes Threading-Verhalten ist oft der Schlüssel zu einer skalierbaren Anwendung, die für ihre Anforderungen das bestmögliche Timing-Verhalten bietet.

### Nachrichtenverarbeitung

Um Verbesserungsmöglichkeiten zu erkennen, brauchen Sie ein Grundverständnis davon, wie das SDK Nachrichten von Subscriptions verarbeitet. Ausgangspunkt ist deshalb ein einzelner Client, also eine Instanz der Klasse `OpcClient`. Er verbindet sich erfolgreich mit einem OPC UA Server und erstellt erfolgreich eine Subscription mit einem MonitoredItem. Ein MonitoredItem ist ein Paar aus Node und Attribut, das die Subscription überwacht.

Sobald die Subscription mit ihrem MonitoredItem angelegt ist, speichert der Server alle Änderungen und Ereignisse, die in das Raster des verwendeten SamplingIntervals fallen, in einem Cache. Parallel dazu fordert der Client in einem separaten Thread innerhalb des PublishingIntervals per Publish-Request die Werte und Ereignisse an, die seit dem letzten Publish-Request neu im Cache liegen. In seiner Antwort (Publish-Response) sendet der Server alle seit der letzten Anforderung gespeicherten Werte und Ereignisse.

Dieses Paket aller Werte und Ereignisse aus dem Cache heißt in OPC UA NotificationMessage und wird als solche versendet. Eine NotificationMessage besteht aus NotificationMessageItems, ein Item je Wert oder Ereignis. Eine NotificationMessage kann Items für mehrere MonitoredItems enthalten, nicht nur für ein bestimmtes MonitoredItem. Auf seine Publish-Requests hin empfängt der Client so eine Folge von NotificationMessages.

Der Client teilt jede empfangene NotificationMessage in ihre Items auf und verteilt sie auf die `EventHandler` für Datenänderungen und Ereignisse der MonitoredItems der Subscription. Standardmäßig übernimmt ein Thread aus dem .NET-Thread-Pool diese Verteilung. So kann der Client die nächste NotificationMessage so schnell wie möglich empfangen und wieder verteilen. Die NotificationMessageItems verteilt das SDK je Subscription sequenziell und hält damit die Reihenfolge der Änderungen im Server ein. Die dabei ausgelösten Ereignisse, also die Aufrufe der `EventHandler`, löst das SDK je Subscription aus. Die eigenen `EventHandler` mehrerer Subscriptions laufen deshalb parallel.

Beim Threading stellt sich oft die Frage, welcher Thread die Items verteilt. Dabei sind drei Fälle zu unterscheiden:

#### Im Worker-Thread

Standardmäßig verteilt das SDK NotificationMessages über Threads aus dem .NET-Thread-Pool. Die Anzahl dieser Threads richtet sich unter anderem nach der Hardware. Während der Verteilung kann der Client parallel weitere NotificationMessages empfangen. Die Items jeder Message verteilt das SDK dabei stets sequenziell. Nach der Verteilung einer NotificationMessage ist es **möglich**, dass derselbe Thread direkt im Anschluss die nächste, inzwischen empfangene NotificationMessage zur Verteilung erhält. Das Wort „möglich“ ist bewusst gewählt: Ob die .NET CLR aus dem Thread-Pool einen anderen oder wieder denselben Thread für die Verteilung bereitstellt, steht nicht fest.

#### Im UI-Thread

In einer Anwendung mit grafischer Benutzeroberfläche, zum Beispiel mit Windows Forms (kurz WinForms) oder der Windows Presentation Foundation (kurz WPF), führt das SDK die `EventHandler` einer Subscription über den aktuell gültigen `SynchronizationContext` aus. Über diesen Kontext verteilt es damit auch die NotificationMessageItems. Dazu ermittelt das SDK beim Erstellen einer Instanz von `OpcSubscription` den aktuell gültigen `SynchronizationContext` und ordnet ihn der Subscription für ihre gesamte Lebensdauer zu. Über diesen `SynchronizationContext` verteilt das SDK dann die NotificationMessageItems, die die Subscription empfängt.

In grafischen Anwendungen verweist der `SynchronizationContext` standardmäßig auf den UI-Thread. Dieser Thread zeichnet die Oberfläche und reagiert auf Eingaben. Wenn das SDK die NotificationMessageItems über den UI-Thread verteilt, können Sie die Oberfläche ohne manuelle Synchronisierung direkt im `EventHandler` aktualisieren. Sie greifen dort also direkt auf Steuerelemente zu, zum Beispiel um die Anzeige mit den neuen Informationen zu aktualisieren. Ein weiterer Vorteil: Das SDK fügt die Verteilung als Job in die `MessageQueue` des UI-Threads ein. So aktualisiert der UI-Thread regelmäßig auch die Oberfläche und ist nicht ausschließlich mit der Verteilung beschäftigt. Die Oberfläche bleibt dadurch reaktionsfähig.

#### In einem eigenen Thread

In manchen Situationen soll gezielt ein anderer oder ein selbst festgelegter Thread die NotificationMessages verteilen. Dafür können Sie über `OpcTaskSchedule.CustomScheduler` einen eigenen Scheduler für solche Hintergrundaufgaben festlegen. Wenn Sie dort zum Beispiel einen Callback festlegen, der die Hintergrundaktion an einen neuen Thread übergibt, findet die Verteilung explizit in einem neuen Thread statt. Ein festgelegter `CustomScheduler` hat Vorrang vor einem vorhandenen `SynchronizationContext` (zweiter Fall).

Aus diesen Fällen ließe sich schließen, dass das SDK stets verschiedene Threads verwendet. Wie die vorigen Abschnitte zeigen, lässt sich diese Frage aber nicht einfach mit Ja oder Nein beantworten. Die Antwort hängt davon ab, wie die eigene Anwendung das SDK einrichtet, und von der Last, also der Menge der zu erwartenden NotificationMessages. Stets einen eigenen Thread über `OpcTaskSchedule.CustomScheduler` bereitzustellen, ist zudem nur auf den ersten Blick eine schnelle Lösung. Die Verarbeitung, in diesem Fall die Verteilung der NotificationMessages, wird mit mehr Threads nicht unbedingt schneller. Ein Grund ist der höhere Scheduling-Aufwand für die CPU, der die Gesamtleistung verschlechtert statt verbessert.

### Timing der Kommunikation

Wie eingangs beschrieben, ist Timing relativ. Es lässt sich deshalb nicht absolut in idealisierten Umgebungen mit einfachen Zahlen bewerten. Um dennoch ein Gefühl für das Timing der eigenen Anwendung und ihres Setups zu bekommen, beginnen die folgenden Fälle mit der einfachsten Infrastruktur mit Netzwerk und werden schrittweise komplexer.

#### Ein Client, ein Server

Rechner A führt die Client-Software aus, Rechner B die Server-Software. Beide Rechner sind direkt per LAN miteinander verbunden. Wenn beide Rechner die gleiche technische Spezifikation erfüllen, erfährt der Client-Rechner Änderungen im Server in etwa so schnell wie möglich und kann sie ebenso schnell verarbeiten.

#### Mehrere Clients, mehrere Server

Wenn sich N Client-Rechner im Verhältnis 1:1 mit N Server-Rechnern verbinden, ändert sich am Timing in der Theorie nichts, ob es nun ein oder N Client-Rechner sind. Voraussetzung ist, dass jedes Client-Server-Paar direkt miteinander verbunden ist. Sobald die Kommunikation über einen Switch, einen Router, ein VPN oder Ähnliches läuft, beeinflusst schon das allein das Timing-Verhalten zwischen den Client-Server-Paaren.

#### Ein Client, mehrere Server

Wenn N Clients auf einem Client-Rechner laufen und sich mit N Server-Rechnern verbinden, kommen zu den Faktoren aus dem zweiten Fall viele weitere hinzu, die das Timing-Verhalten beeinflussen. Abgesehen von der Netzwerkkarte betrachtet dieser Fall ausschließlich eine Clientanwendung mit N Instanzen von `OpcClient`, die sich mit N Servern verbinden. Jede dieser Instanzen verwendet:

- einen eigenen Thread, um die NotificationMessages anzufordern und zu empfangen
- standardmäßig den .NET-Thread-Pool, um die NotificationMessageItems auf die `EventHandler` der MonitoredItems zu verteilen
- für die Verarbeitung der Informationen vom Server:
  - entweder denselben Thread, den zuvor schon ein anderer Client verwendet hat
  - oder einen anderen Thread
  - oder einen neuen Thread

Auf einem leistungsschwachen Rechner mit wenig freien Ressourcen ist deshalb unter Umständen der .NET-Thread-Pool der Flaschenhals. Das verschlechtert das Timing-Verhalten bei der Verteilung und Abarbeitung der NotificationMessages.

Unabhängig vom .NET-Thread-Pool spielt vor allem die eigene, nachgelagerte Verarbeitungskette der Datenänderungen und Ereignisse eine entscheidende Rolle für das Timing. Aktionen in einem Thread aus dem .NET-Thread-Pool sollten möglichst nie (zu lange) blockieren. Sie sollten deshalb nur auf CPU und Arbeitsspeicher arbeiten und Zugriffe auf Ressourcen wie Dateisystem, Netzwerk oder Datenbanken unbedingt vermeiden. Solche Zugriffe verschlechtern ebenfalls das Timing-Verhalten bei der weiteren Verteilung von NotificationMessages. Wenn ein Worker-Thread (ein Thread aus dem .NET-Thread-Pool) zu lange blockiert, kann das außerdem den gesamten .NET-Thread-Pool aushungern (Starvation). Der Grund: Seine Threads führen auch die Continuations von asynchronen Operationen und Tasks aus.

### Empfehlungen

Für das bestmögliche **Timing-Verhalten** empfehlen wir deshalb, in den Handler-Methoden von Client, Subscription und MonitoredItems ausschließlich Operationen auf CPU und Arbeitsspeicher auszuführen. Verwenden Sie die Handler-Methoden vor allem dazu, Benachrichtigungen (engl. Notifications) zu bewerten, einzuordnen und in eine eigene Queue einzureihen. Ein separater eigener Thread holt die Benachrichtigungen dann aus dieser Queue und verarbeitet sie nach eigenen Maßstäben. So entkoppeln Sie die Verarbeitung von der Kommunikation und bestimmen Timing-Verhalten und Skalierung im Rahmen Ihrer eigenen Softwarearchitektur.

#### Alle Benachrichtigungen verarbeiten

Die folgenden Beispiele zeigen, wie die Datenverarbeitung im Client aussehen kann. Ausgangspunkt ist wie immer eine aktive Verbindung mit Subscription:

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

    ...
}
```

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

    ...
End Using
```

Der `EventHandler` füllt nur die Queue, über die die Threads Daten austauschen:

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

```

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

```

Die gezeigte `BlockingCollection` verwendet standardmäßig eine `ConcurrentQueue`. Zusätzlich blockiert sie, solange die Collection keinen Eintrag enthält. Wenn Sie statt einer `BlockingCollection` direkt eine `ConcurrentQueue` oder eine andere Auflistung verwenden, müssen Sie zwei Dinge sicherstellen. Erstens ist der Zugriff threadsicher, damit nicht mehrere Threads zugleich zugreifen. Zweitens blockiert der verarbeitende Thread, solange kein Eintrag enthalten ist. Andernfalls läuft der verarbeitende Thread mit voller CPU-Last. So füllt das Beispiel die Queue:

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

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

```

Der folgende Codeausschnitt definiert und startet den Thread, der die Auflistung der Daten verarbeitet:

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

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

Nach dem Start führt der separate Thread die folgende Methode aus. Sie ruft alle Daten der Auflistung ab und blockiert die ForEach-Schleife, wenn keine weiteren Daten in der Auflistung vorhanden sind. Die Schleife läuft weiter, sobald ein neuer Eintrag hinzukommt. Erst wenn `CompleteAdding` auf der Auflistung aufgerufen wird, beendet der Enumerator die Schleife, und zwar nachdem alle noch ausstehenden Einträge abgearbeitet sind.

```csharp
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.
    }
}
```

```vbnet
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

```

Wenn Sie die Anwendung beenden oder die Verbindung zum Server schließen, beendet der folgende Code die Datenverarbeitung:

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

```vbnet
data.CompleteAdding()
processDataThread.Join()
```

Das gezeigte Beispiel verarbeitet alle empfangenen Daten und beendet die Verarbeitung erst, wenn die `BlockingCollection` keine Einträge mehr enthält. Der Aufruf von `processDataThread.Join()` blockiert deshalb so lange, bis der Thread alle Einträge abgearbeitet und die Auflistung `data` vollständig geleert hat.

#### Verarbeitung mit Abbruchbedingung

Die Datenverarbeitung im Client kann auch nur so lange laufen, bis eine bestimmte Bedingung eintritt. Den Eintritt der Bedingung kann zum Beispiel ein `CancellationToken` signalisieren. Die folgenden Ausschnitte zeigen einen Client mit Abbruchbedingung.

Auch hier ist eine aktive Verbindung mit Subscription die Voraussetzung:

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

    ...
}
```

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

    ...
End Using
```

Der `EventHandler` füllt weiterhin nur die Queue, über die die Threads Daten austauschen:

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

```

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

```

Ebenso befüllt weiterhin der `EventHandler` die Queue:

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

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

```

Der folgende Codeausschnitt definiert und startet den Thread mit einem `CancellationToken`. Über den `CancellationToken` signalisiert die Anwendung dem Thread, dass er die Verarbeitung der möglicherweise noch ausstehenden Daten beenden soll. Die Aufgabe des Threads bleibt die Verarbeitung der Daten aus der Auflistung:

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

processDataThread.Start();
```

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

processDataThread.Start()
```

Nach dem Start führt der separate Thread die folgende Methode aus. Sie ruft die Daten der Auflistung ab und verarbeitet sie. Der Aufruf `data.Take(...)` blockiert, bis die Auflistung einen Eintrag enthält oder der `CancellationToken` den Abbruch signalisiert. Bei einem Abbruch löst `Take` eine `OperationCanceledException` aus. Anschließend endet die While-Schleife, weil der `CancellationToken` den Abbruch signalisiert. Die Schleife prüft das sowohl nach einer Exception als auch nach der Verarbeitung eines Eintrags.

```csharp
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.
        }
    }
}
```

```vbnet
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

```

Wenn Sie die Anwendung beenden, die Verbindung zum Server trennen oder die Verarbeitung wegen einer Abbruchbedingung sofort beenden wollen, informiert der folgende Code den Thread darüber:

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

```vbnet
processDataCancellation.Cancel()
processDataThread.Join()
```

Das gezeigte Beispiel verarbeitet die Daten der Auflistung, bis die Abbruchbedingung eintritt. Daten, die in der Zwischenzeit noch empfangen und in die Queue eingetragen wurden, verarbeitet der Thread danach nicht mehr. Sie verbleiben in der Auflistung.

#### Beratung und Support

Die beschriebenen Abläufe und Beispiele helfen Ihnen, eine skalierbare Subscription zu implementieren. Wenn Sie die empfangenen Werte und Ereignisse korrekt und passend zu Ihrem Anwendungsfall verarbeiten, behalten Sie die volle Kontrolle über das Timing-Verhalten Ihrer Anwendung. Bei Fragen zur Implementierung und zum richtigen Vorgehen unterstützt Sie unser Entwicklungsteam über [support@traeger.de](mailto:support@traeger.de).