Performance improvement

Author: samson0v

thingsboard_gateway/connectors/modbus/modbus_connector.py

@@ -213,7 +213,7 @@ def __add_slave(self, slave_config):
             slave.master = master
 
             self.__slaves.append(slave)
-            slave.start()
+            self.loop.create_task(slave.run())
         except ValueError as e:
             self.__log.error("Failed to add slave: %s", e)
 
@@ -365,7 +365,7 @@ async def __convert_data(self):
 
                         batch_to_convert[batch_key].append(data)
 
-                    for (device_name, uplink_converter), data in batch_to_convert.items():
+                    for (_, uplink_converter), data in batch_to_convert.items():
                         converted_data: ConvertedData = uplink_converter.convert({}, data)
                         self.__log.trace("Converted data: %r", converted_data)
                         if len(converted_data['attributes']) or len(converted_data['telemetry']):

thingsboard_gateway/connectors/modbus/slave.py

@@ -13,8 +13,7 @@
 #     limitations under the License.
 
 import asyncio
-from threading import Thread
-from time import sleep, monotonic
+from time import monotonic
 from typing import TYPE_CHECKING, Dict, Union
 
 from _asyncio import Future
@@ -62,10 +61,8 @@
     from thingsboard_gateway.connectors.modbus.entities.master import Master
 
 
-class Slave(Thread):
+class Slave:
     def __init__(self, connector: 'ModbusConnector', logger, config):
-        super().__init__()
-        self.daemon = True
         self.stopped = False
         self._log = logger
         self.connector = connector
@@ -132,7 +129,7 @@ def __init__(self, connector: 'ModbusConnector', logger, config):
         for attr_config in self.attributes_updates_config:
             self.shared_attributes_keys.append(attr_config[TAG_PARAMETER])
 
-    def __timer(self):
+    async def __timer(self):
         if self.__master is not None:
             self.__send_callback(monotonic())
             next_poll_time = monotonic() + self.poll_period
@@ -144,7 +141,7 @@ def __timer(self):
                     next_poll_time = current_time + self.poll_period
 
                 sleep_time = max(0.0, next_poll_time - monotonic())
-                sleep(sleep_time)
+                await asyncio.sleep(sleep_time)
 
     def __send_callback(self, current_monotonic):
         self.last_polled_time = current_monotonic
@@ -154,17 +151,18 @@ def __send_callback(self, current_monotonic):
         except Exception as e:
             self._log.exception('Error sending slave callback: %s', e)
 
-    def run(self):
+    async def run(self):
         if self.uplink_converter_config.is_readable():
-            self.__timer()
+            await self.__timer()
 
     def close(self, loop):
+        self.stopped = True
+
         future = asyncio.run_coroutine_threadsafe(self.disconnect(), loop)
         try:
             future.result(timeout=5)
         except Exception as e:
             self._log.error('Failed to disconnect from %s: %s', self, e)
-        self.stopped = True
 
     def get_name(self):
         return self.device_name