Data loss when trying to connect to ProSys Simulation Server

[KishanKishore]

I am using the sync wrapper present in the opcua-asynio library in combination with the multiprocessing module to connect to a ProSys Simulation Server. I have around 4000 nodes configured in the server.

I have a global in memory Queue (from the multiprocessing package) to which the data is being pushed after it is received in the datachange_notification method.

I am trying to keep the execution time of the datachange_notification callback as short as possible because it has been mentioned in several places that doing time taking operations within this callback can cause problems.

Here is my Subscription Handler class.

data_values = multiprocessing.Queue()

class SubHandler(object):

    """
    Client to subscription. It will receive events from server
    """

    def datachange_notification(self, node, val, data):
        data_values.put_nowait([str(node), val])

    def event_notification(self, event):
        print("Python: New event", event)

This is my target method which is running in a separate process and subscribes to a given set of nodes:

def stream_data():
    client = Client("opc.tcp://localhost:53530/OPCUA/SimulationServer/")
    client.connect()
    nodes = []
   
   # Get first 100 nodes
    for i in range(1,100):
        nodes.append(f"ns=3;i={1000 + i}")

    node_objects = [client.get_node(i) for i in nodes]
    
    handler = SubHandler()
    sub = client.create_subscription(500, handler)
    handle = sub.subscribe_data_change(node_objects)

I ran this for exactly 1 minute with 100 nodes and then later counted the total number of values received per node (in the main process):

value_count = {}
while not data_values.empty():
        i = data_values.get(timeout=1)
        
        node = i[0]
        val = i[1]

        if node in value_count:
            value_count[node].append(val)
        else:
            value_count[node] = [val]

The expected number of points is 120 with a publishing interval of 500 ms (The Simulation Server is also publishing at the same interval). But I receive only 17-18 values for every node.

If I reduce the number of nodes to 10 or lesser then the number of data points received are around 120. Is there a problem in using this library with multiprocessing or am I doing something incorrectly ? Any suggestions are appreciated. I need to be supporting 4000 nodes per process. What would be the best design for the datachange_notofication callback to handle a large number of nodes in a subscription?

I have also noticed that currently the code here:

opcua-asyncio/asyncua/common/subscription.py

Line 115 in ee844ed

for item in datachange.MonitoredItems:

loops over all the nodes present in the MonitoredItems object. I am going to try to remove the loop and send the MonitoredItems directly to reduce the number of times the callback is called. Would that be a logical thing to try ?

[AndreasHeine]

loops over all the nodes present in the MonitoredItems object. I am going to try to remove the loop and send the MonitoredItems directly to reduce the number of times the callback is called. Would be the a logical thing to try ?

i hope you read the opc ua reference part befor send stuff directly there is a publishing interval which only sends queued notifications each x sec.
data loss can also happen if the subscription queue is to small or the publishing interval is to large for the queue size(to small is also bad to much requests on the wire)

[AndreasHeine]

In my case the sampling interval (from the ProSys simulator) is 500 milliseconds and the publishing interval that I am specifying while creating the subscription is also 500 milliseconds.

have you checked the revised publishing interval with wireshark? (client and server negotiate settings in most "create" requests) but thats a different thing

Also because the callback is being called for every node separately, if we subscribe to let us say 1000 nodes at a 500 milliseconds, is the datachange_notification method expected to execute 2000 times per second?
This number will keep increasing linearly with the number of nodes increasing.

true! each subscription has a number of monitored items and for each subscription as client you get an array of datachanges back (publishresponse with wireshark you can investigate the data) and for each item you need to call the _callback which than calls your subhandler according to your item -> thats linear for sure! but if you use the async subhandler function its concurrent

opcua-asyncio/asyncua/common/subscription.py

Line 123 in ee844ed

if asyncio.iscoroutinefunction(self._handler.datachange_notification):

in my oppinion mixing async and sync make things easy to break and its pretty easy to block the eventloop^^

if you find a better way feel free to PR (but stick to specs ^^)!!!

made a lot of tests with only 1000, but rapidly changing variables (1000 each 50ms) and i had no dataloss at all (pure asyncio no sync mix!) if you just send them in stdout and the parent (process manager) just graps them then from stdout and thows them in another process you should be clean and fine

for another question (python-opcua) i made a quick changing array to show there is no dataloss

[KishanKishore]

The RevisedPublishingInterval is 1000 milliseconds. Regarding mixing the async and sync code together, I will also try the pure async version although it is not really very suitable to my use case.

[KishanKishore]

[AndreasHeine]

i can only encurage you if you cant use pure async ->

p1: process manager reads sub 1 stdout and passes to sub2 stdin
sub1: asyncio producer process -> writes stdout
sub2: stdin -> consumer process with processpool/threadpool complex "whatever you want" logic

clean process decouple! its like grabing a fast rotating carousell for too long (if you do, you get a free flight^^)

[KishanKishore]

I will have a look at this as well although I am still confused how writing to stdout would be faster than doing an in-memory operation. From what I know printing repeatedly is not very efficient. But thanks for the advice!

[AndreasHeine]

here is the python-opcua example i used for the screenshot

this sample server has a trigger wich will be set by the client an then the server start sampling after the subscription is already made by client and then it runs for a defined time and then you can compare the output (chose an array because you can see the pattern an if there is a missing message you will see it fast)

client.py

from opcua import Client
from opcua import ua


class SubHandler(object):

    """
    Subscription Handler. To receive events from server for a subscription
    data_change and event methods are called directly from receiving thread.
    Do not do expensive, slow or network operation there. Create another 
    thread if you need to do such a thing
    """

    def datachange_notification(self, node, val, data):
        print(val)

if __name__ == "__main__":
    client = Client("opc.tcp://localhost:4840")
    client.connect()
    client.load_type_definitions()
    handler = SubHandler()
    sub = client.create_subscription(50, handler)
    handle = sub.subscribe_data_change(client.get_node("ns=2;i=1"),queuesize=1000)
    trigger = client.get_node("ns=2;i=2")
    trigger.set_value(True)

server.py

import time, copy, datetime
from opcua import Server, ua

myarray = []
for item in range(64):
    myarray.append(0)

if __name__ == "__main__":
    server = Server()
    server.set_endpoint("opc.tcp://127.0.0.1:4840/")
    idx = server.register_namespace("test")
    var = server.get_objects_node().add_variable(idx, "MyArray", ua.Variant(value=myarray, varianttype=ua.VariantType.Int64))
    trigger = server.get_objects_node().add_variable(idx, "Trigger", ua.Variant(value=False, varianttype=ua.VariantType.Boolean))
    trigger.set_writable(True)
    server.start()

    start = time.time()
    while 1:
        if trigger.get_value() == True:
            for idx, item in enumerate(myarray):
                time.sleep(0.005)
                myarray[idx]+=1
                val = copy.copy(myarray)
                dv = ua.DataValue(ua.Variant(value=val, varianttype=ua.VariantType.Int64))
                var.set_value(dv)
                print(var.get_value())
            if (time.time() - start) >= 10:
                break
        else:
            time.sleep(0.050)
    time.sleep(60)
    server.stop()

[KishanKishore]

Thanks for taking the time to write the code but I see that you have subscribed to a single node ns=2;i=1 . I am not having any problems when subscribing to a single node. The problem starts occurring when I subscribe to more than 100 nodes. But I am going to try some more things on my side and get back here.

Thanks again!

[AndreasHeine]

scale it up and test 0.005 -> 200 Notifications per second add the node 5 (or more) times^^

    handle1 = sub.subscribe_data_change(client.get_node("ns=2;i=1"),queuesize=500)
    handle2 = sub.subscribe_data_change(client.get_node("ns=2;i=1"),queuesize=500)
    handle3 = sub.subscribe_data_change(client.get_node("ns=2;i=1"),queuesize=500)
    handle4 = sub.subscribe_data_change(client.get_node("ns=2;i=1"),queuesize=500)
    handle5 = sub.subscribe_data_change(client.get_node("ns=2;i=1"),queuesize=500)

dump it in a file!

[zerox1212]

I believe you doing 200 data changes per second on a single node is a lot different than doing 1000 data changes on a 1000 nodes.

[AndreasHeine]

@zerox1212 true but you can test your subhandler!!! it does not matter if a notification came from a different node the number of subhandlercalls per second is important

[zerox1212]

@KishanKishore Just for fun, what happens if you run your test with 10 subscriptions, each with 10 nodes?

[KishanKishore]

Hey thanks for the reply. I am going to be trying that and report back here although I remember you guys discussing on the python-opcua repository that fewer subscriptions are better.

What is your view on handling a large number of nodes? I read through some of the answers that you have written on the other repository but still haven't figured out the best way to handle a large number of subscriptions.

[zerox1212]

When I was using the sync library, I was making a lot of subscription objects because I didn't know better. In theory one subscription with lots of items should be more efficient, but I never tested it. My project only had a few hundred nodes and it seem fast to me.

Another simple test would be to just remove the multiprocessing stuff and print like @AndreasHeine example. This would tell us if the problem is caused by multiprocessing or something with a large number items on one subscription.

As a final note, depending on your application you could make a single node with an array with length 4000. This no doubt would be the fastest way to read 4k values.

[zerox1212]

You might have read this, but see the last post from @oroulet : FreeOpcUa/python-opcua#701 He says to do the test I mentioned and compare speed. :)

[AndreasHeine]

hat is your view on handling a large number of nodes? I read through some of the answers that you have written on the other repository but still haven't figured out the best way to handle a large number of subscriptions.

When I was using the sync library, I was making a lot of subscription objects because I didn't know better. In theory one subscription with lots of items should be more efficient, but I never tested it. My project only had a few hundred nodes and it seem fast to me.

in "theory" having to many subscriptions can cause problems but this is up to the server capabilities and its operational limits so there is no general answer a best practice is to read the server-capabilities/operational-limits and adopt your client application to it. there is a point on which the server rejects requests (like BadToManyOperations)
it is also better to seperate between information which change a lot (which should be subscribed) and information which does not change often (which can be polled with a client.read_values(nodes) dont use node.read_value() it is one request per node which is inefficient) also a good point from @zerox1212 is to use arrays instead of single variables (if you have a lot variables of the same type you can crunch them together and also the possiblity to have array and the single variables in the addresspace thats a modelling detail which has a big impact on the performance) unfortunatly we dont support the expose_its_array feature wich i used alot with Siemen PLCs where you have an large array which also exposes each arrayelement as child of the arrayvariable...

as i mentioned before dataloss can happen if the queue is to small for the publishinterval (older values will be discarded) real dataloss will only happen if the eventloop is blocked so a incomming publishresponse-message cant be processed...

[KishanKishore]

Thanks again for your valuable suggestions. I think I might have found something that is working for my use case (within the context of multiprocessing). I like the idea of having an array with 4000 values but my requirement is different and mandates 4000 individual nodes. I edited the following code from the python-opcua library just for testing purposes.

def _call_datachange(self, datachange):
        
        try:
            self._handler.datachange_notification(datachange.MonitoredItems, self._monitoreditems_map)
        except Exception as exception:
            print(exception)

        # for item in datachange.MonitoredItems:
        #     with self._lock:
        #         if item.ClientHandle not in self._monitoreditems_map:
        #             self.logger.warning("Received a notification for unknown handle: %s", item.ClientHandle)
        #             self.has_unknown_handlers = True
        #             continue
        #         data = self._monitoreditems_map[item.ClientHandle]
        #     if hasattr(self._handler, "datachange_notification"):
        #         event_data = DataChangeNotif(data, item)
        #         try:
        #             self._handler.datachange_notification(data.node, item.Value.Value.Value, event_data)
        #         except Exception:
        #             self.logger.exception("Exception calling data change handler")
        #     elif hasattr(self._handler, "data_change"):  # deprecated API
        #         self.logger.warning("data_change method is deprecated, use datachange_notification")
        #         try:
        #             self._handler.data_change(data.server_handle, data.node, item.Value.Value.Value, data.attribute)
        #         except Exception:
        #             self.logger.exception("Exception calling deprecated data change handler")
        #     else:
        #         self.logger.error("DataChange subscription created but handler has no datachange_notification method")

Then I set the queuesize parameter as 1000. This is how my subscription handler looks like:

manager = multiprocessing.Manager()
data_queue = manager.Queue()

def datachange_notification(self, monitored_items, monitored_item_map):
        data_value = {}

        for i in monitored_items:
            node_id = str(monitored_item_map[i.ClientHandle].node)
            value = i.Value.Value.Value
            source_timestamp = str(i.Value.SourceTimestamp)

            if node_id in data_value:
                data_value[node_id].append([value, source_timestamp])
            else:
                data_value[node_id] = [[value, source_timestamp]]
        
        data_queue.put(data_value)

I don't see any data loss here although I have tested only for a brief time interval of 5 seconds for now.