Different behaviour if running subscribe_data_change and subscribe_alarms_and_conditions in one process or different processes

[ts4iot]

I'm capturing data from a Siemens WinCC server.
The alarm messages from the WinCC server are captured by subscribe_alarms_and_conditions.
Some status words of components are captured by subscribe_data_change.

Both handlers are writing the captured data to different asyncio.Queue(s).

If I run the data capturing of subscribe_alarms_and_conditions and subscribe_data_change in two separat procesess, the data capturing runs fine and looks well synchronized.

If I run both in one process it looks also fine at the beginning but than step by step the subscription with the higher load becomes dominant and after some time I got no information from the subscription with the lower load.

Any idea if this behaviour could be caused by the opcua-asyncio library?

[AndreasHeine]

you need to ensure your queue consumers to be at equal priority so both will get the chance to process a item!

[oroulet]

if you are overloading your CPU or not releasing the loop often enough you can get whatever strange issues. And yes what you describe is typical

[ts4iot]

My consuming code look like this. The send methods are transmitting data to an iot platform. How to check if I have enough free time in the main loop. CPU util is at 40%. 20% with only one subscription.

	while g_running:
		if(not g_eventqueue.empty()):
			g_logger.info("Event queue size start: %d" % g_eventqueue.qsize())	
			while(not g_eventqueue.empty()):
				send_event(await g_eventqueue.get())	
		if(not g_statusqueue.empty()):
			g_logger.info("Status queue size start: %d" % g_statusqueue.qsize())		
			while(not g_statusqueue.empty()):
				send_state(await g_statusqueue.get())	
		await g_plc.check_connection()
		await asyncio.sleep(0.1)

The event queue receives about 100 messages / second.
The status queue receives about 25 messages / second.

[AndreasHeine]

currently you process all items in one queue and after that you process all items of the second one... and with the "while queue not empty", you will never have equal priority, so as long as one queue is not empty it will not switch to the other... you should consider using one queue instead with two different itemtypes in it or process one item (if not queue not empty) from each queue in a whileloop so it will process both (one item from both) in one iteration.

[AndreasHeine]

the limit of python 3.10 asyncio is beyond 600+ msg/sec (on a i7 9750H on Win10 -> WindowsSelectorEventLoopPolicy... so on linux with ProactorEventLoop its faster) but you need streamlined code which does not block at all!

[ts4iot]

I also made some tests without sending the data to the iot platform. But the behaviour is the same.

[ts4iot]

Thank you for the help. After disabling the transmission to the iot platform and decreasing the await asyncio.sleep(0.1) to await asyncio.sleep(0.01) brings the expected result. This is a clear indication that the data transmission needs to much time and blocks the opcua processing. Currently i use paho-mqtt for the data transmission. I will now switch to asyncio-mqtt.

[AndreasHeine]

see:
https://github.com/AndreasHeine/opcua-sub-to-mqtt/blob/c9816fcf1ea0c91a22a06f680263c6df59640881/opcua_client_mqtt_publisher.py#L272