Latency on local instance subscriptions

[eduard93]

I have a server that publishes 100 variables and randomly increments one variable every 0.01 second (it amounts to less than a hundred updates per second):

server.py

import logging
import asyncio
import sys
from random import randrange

from asyncua import ua, Server


logging.basicConfig(
    format='%(asctime)s.%(msecs)03d %(levelname)s %(module)s %(funcName)-8s %(message)s',
    level=logging.INFO,
    datefmt='%Y-%m-%d %H:%M:%S')


async def main(count: int):
    _logger = logging.getLogger('asyncua')
    # setup our server
    server = Server()
    await server.init()
    server.set_endpoint('opc.tcp://0.0.0.0:4840/freeopcua/server/')

    # setup our own namespace, not really necessary but should as spec
    uri = 'http://examples.freeopcua.github.io'
    idx = await server.register_namespace(uri)

    # populating our address space
    # server.nodes, contains links to very common nodes like objects and root
    myobj = await server.nodes.objects.add_object(idx, 'MyObject')

    myvars = []
    for i in range(count):
        myvar = await myobj.add_variable(idx, f'MyVariable{i}', randrange(100))
        await myvar.set_writable()
        myvars.append(myvar)

    _logger.info('Starting server!')
    async with server:
        while True:
            await asyncio.sleep(.01)
            myvar = myvars[randrange(count)]
            new_val = await myvar.get_value() + 0.1
            await myvar.write_value(new_val)


if __name__ == '__main__':
    count = 100
    logging.basicConfig(level=logging.DEBUG)
    asyncio.run(main(count), debug=True)

And I run a corresponding client on a same server:

client.py

import threading
from datetime import datetime
import time
import asyncio
from asyncua import Client, Node, ua


class SubscriptionHandler:
    def datachange_notification(self, node: Node, val, data):
        # store/log: data.monitored_item.Value.SourceTimestamp.isoformat() and ts()
        pass

async def main(count: int):
    async with Client(url="opc.tcp://localhost:4840/freeopcua/server/") as client:
        idx = await client.get_namespace_index(uri="http://examples.freeopcua.github.io")
        nodes = []
        for i in range(count):
            var = await client.nodes.objects.get_child([f"{idx}:MyObject", f"{idx}:MyVariable{i}"])
            nodes.append(var)

        handler = SubscriptionHandler()
        subscription = await client.create_subscription(100, handler)
        await subscription.subscribe_data_change(nodes)
        await asyncio.sleep(300)
        await subscription.delete()
        await asyncio.sleep(1)

def ts():
    return datetime.utcnow().isoformat()


if __name__ == '__main__':
    count = 100
    asyncio.run(main(count))

The problem is server reports: 2021-08-03 20:20:52.403 INFO uaprocessor forward_publish_response Server wants to send publish answer but no publish request is available,enqueuing notification, length of result queue is 19321

So the queue grows and datachange notifications are arriving later and later on the client.

I have tried to:

1. Set various poll intervals in: client.create_subscription(100, handler). I started from the default 500, and 100 seems to be the best, but I still see a growing queue and a growing processing time.
2. Split nodes into 2 subscriptions:

subscription = await client.create_subscription(100, handler)
await subscription.subscribe_data_change(nodes[:int(len(nodes)/2)])
subscription2 = await client.create_subscription(100, handler2)
await subscription2.subscribe_data_change(nodes[int(len(nodes)/2):])

That also didn't help.

Notes:

• In the Task Manager I see that both server and client processes do not consume all available CPU.
• If I change await asyncio.sleep(.01) in server.py to slow updates, i.e.: await asyncio.sleep(.05) the queue disappears.
• I have measured SubscriptionHandler:datachange_notification performance and it consistently completes under 1 ms. But even when left with pass only the server queue grows quite fast.

Am I missing something? Is it possible to speed it up somehow?

[AndreasHeine]

In the Task Manager I see that both server and client processes do not consume all available CPU.

if i remember right the GIL will never let a python process use 100% CPU usage

subscription = await client.create_subscription(100, handler)
await subscription.subscribe_data_change(nodes[:int(len(nodes)/2)])
subscription2 = await client.create_subscription(100, handler2)
await subscription2.subscribe_data_change(nodes[int(len(nodes)/2):])

it does not look like you set a queuesize at all !?

opcua-asyncio/asyncua/common/subscription.py

Line 169 in 085b00e

async def subscribe_data_change(self,

    async def subscribe_data_change(self,
                                    nodes: Union[Node, Iterable[Node]],
                                    attr=ua.AttributeIds.Value,
                                    queuesize=0,
                                    monitoring=ua.MonitoringMode.Reporting,
                                   ) -> Union[int, List[Union[int, ua.StatusCode]]]:

[eduard93]

@swamper123 yes, I mean fast communication, not realtime. Thanks for the info.

[AndreasHeine]

in what interval do you need the information? i guess its a better point to start! "realtime" is always defined from the process point of view.

[eduard93]

@AndreasHeine I need to subscribe to ~8k nodes with 100 ms latency of getting updates. On average nodes change once per second so I get 8k changes per second.
I need for the latency between server firing the event and client receiving it to be less than 100 ms.

[AndreasHeine]

the hardware makes a huge difference (single-core-performance the eventloop runs in just one python process...) another point to get it working you need to be very efficient in you python code! propably you need to implement a producer/consumer architecture (opc ua data producer process -> queue/pipe (whatever) -> consumer process/process pool)
Publishinterval 100ms and avg. 8000 DataChanges/s --> 800 DataChanges per publish
should be possible but i would try to decouple it (producer/consumer)

for sub 100ms latency python is maybe not the best language (depends on the cpu and the python code)

[swamper123]

You may want to boost your interpreter as well. You can configure the source code with the arguments --enable-optimizations --with-lto and then compile it.
(More informations: https://docs.python.org/3.11/using/configure.html#cmdoption-enable-optimizations)
This may give a boost to the application in some cases (haven't meassured it yet).

But be sure, that you don't mess up your interpreter installation, install it via "make altinstall". See the README file in the python source.

But keep in mind, Python is an interpreter language...if you want to go fast, you have to use something like c/++ or rust.