fix(device): use lock whenever a device is communicated with

Lock was only used in the receive loop to synchronize among child devices.
However the initialisation step that is getting a single poll from each device and
sending commands to devices also require the same synchronization.

A previous PR attempted this, but was locking too coarsely including some periods where
we were waiting for a timeout, so caused major slowdown in device performance.

Since we added the api_lock usage on send, we no longer need a separate per device
lock to prevent multiple concurrent sends.

Issue #1054

custom_components/tuya_local/device.py

@@ -165,7 +165,6 @@ class TuyaLocalDevice(object):
         self._SINGLE_PROTO_CONNECTION_ATTEMPTS = 3
         # The number of failures from a working protocol before retrying other protocols.
         self._AUTO_FAILURE_RESET_COUNT = 10
-        self._lock = Lock()
 
     @property
     def name(self):
@@ -345,126 +344,124 @@ class TuyaLocalDevice(object):
         while self._running:
             error_count = self._api_working_protocol_failures
             force_backoff = False
-            try:
-                await self._api_lock.acquire()
-                last_cache = self._cached_state.get("updated_at", 0)
-                now = time()
-                full_poll = False
-                if (persist == self.should_poll) or (
-                    persist and (self._api.socket is None)
-                ):
-                    # use persistent connections after initial communication
-                    # has been established.  Until then, we need to rotate
-                    # the protocol version, which seems to require a fresh
-                    # connection.
-                    persist = not self.should_poll
-                    _LOGGER.debug(
-                        "%s persistant connection set to %s", self.name, persist
-                    )
-                    self._api.set_socketPersistent(persist)
-                    if self._api.parent:
-                        self._api.parent.set_socketPersistent(persist)
-                    self._last_full_poll = 0  # ensure we start with a full poll
-
-                needs_full_poll = now - self._last_full_poll > self._CACHE_TIMEOUT
-                if now - last_cache > self._CACHE_TIMEOUT or (
-                    persist and needs_full_poll
-                ):
-                    if (
-                        self._force_dps
-                        and not dps_updated
-                        and self._api_protocol_working
+            async with self._api_lock:
+                try:
+                    last_cache = self._cached_state.get("updated_at", 0)
+                    now = time()
+                    full_poll = False
+                    if (persist == self.should_poll) or (
+                        persist and (self._api.socket is None)
                     ):
-                        poll = await self._retry_on_failed_connection(
-                            lambda: self._api.updatedps(self._force_dps),
-                            f"Failed to update device dps for {self.name}",
+                        # use persistent connections after initial communication
+                        # has been established.  Until then, we need to rotate
+                        # the protocol version, which seems to require a fresh
+                        # connection.
+                        persist = not self.should_poll
+                        _LOGGER.debug(
+                            "%s persistant connection set to %s", self.name, persist
                         )
-                        dps_updated = True
-                    else:
-                        poll = await self._retry_on_failed_connection(
-                            lambda: self._api.status(),
-                            f"Failed to fetch device status for {self.name}",
-                        )
-                        dps_updated = False
-                        full_poll = True
-                    self._last_full_poll = now
-                    last_heartbeat = now  # reset heartbeat timer on full poll
-                elif persist:
-                    if now - last_heartbeat > self._HEARTBEAT_INTERVAL:
-                        await self._hass.async_add_executor_job(
-                            self._api.heartbeat,
-                            True,
-                        )
-                        last_heartbeat = now
-                    poll = await self._hass.async_add_executor_job(
-                        self._api.receive,
-                    )
-                    # Ignore Payload error 904, as 3.4 protocol devices seem to return
-                    # this when there is no new data, instead of just returning nothing.
-                    if poll and "Err" in poll and poll["Err"] == "904":
-                        poll = None
-                else:
-                    force_backoff = True
-                    poll = None
-
-                if poll:
-                    if "Err" in poll:
-                        # Limit disconnects to the errors that are caused low level
-                        # communication problems
-                        if poll["Err"] in {"901", "902", "905", "906", "914"}:
-                            force_backoff = True
-                            persist = False
-                            self._api.set_socketPersistent(False)
-                            if self._api.parent:
-                                self._api.parent.set_socketPersistent(False)
-                        # increment the error count if not done already
-                        if error_count == self._api_working_protocol_failures:
-                            self._api_working_protocol_failures += 1
-                        if self._api_working_protocol_failures == 1:
-                            _LOGGER.warning(
-                                "%s error reading: %s", self.name, poll["Error"]
+                        self._api.set_socketPersistent(persist)
+                        if self._api.parent:
+                            self._api.parent.set_socketPersistent(persist)
+                        self._last_full_poll = 0  # ensure we start with a full poll
+
+                    needs_full_poll = now - self._last_full_poll > self._CACHE_TIMEOUT
+                    if now - last_cache > self._CACHE_TIMEOUT or (
+                        persist and needs_full_poll
+                    ):
+                        if (
+                            self._force_dps
+                            and not dps_updated
+                            and self._api_protocol_working
+                        ):
+                            poll = await self._retry_on_failed_connection(
+                                lambda: self._api.updatedps(self._force_dps),
+                                f"Failed to update device dps for {self.name}",
                             )
+                            dps_updated = True
                         else:
-                            _LOGGER.debug(
-                                "%s error reading: %s", self.name, poll["Error"]
+                            poll = await self._retry_on_failed_connection(
+                                lambda: self._api.status(),
+                                f"Failed to fetch device status for {self.name}",
                             )
-                        if "Payload" in poll and poll["Payload"]:
-                            _LOGGER.debug(
-                                "%s err payload: %s",
-                                self.name,
-                                poll["Payload"],
+                            dps_updated = False
+                            full_poll = True
+                        self._last_full_poll = now
+                        last_heartbeat = now  # reset heartbeat timer on full poll
+                    elif persist:
+                        if now - last_heartbeat > self._HEARTBEAT_INTERVAL:
+                            await self._hass.async_add_executor_job(
+                                self._api.heartbeat,
+                                True,
                             )
+                            last_heartbeat = now
+                        poll = await self._hass.async_add_executor_job(
+                            self._api.receive,
+                        )
+                        # Ignore Payload error 904, as 3.4 protocol devices seem to return
+                        # this when there is no new data, instead of just returning nothing.
+                        if poll and "Err" in poll and poll["Err"] == "904":
+                            poll = None
                     else:
-                        if "dps" in poll:
-                            poll = poll["dps"]
-                        if isinstance(poll, dict):
-                            poll["full_poll"] = full_poll
-                            yield poll
+                        force_backoff = True
+                        poll = None
+
+                    if poll:
+                        if "Err" in poll:
+                            # Limit disconnects to the errors that are caused low level
+                            # communication problems
+                            if poll["Err"] in {"901", "902", "905", "906", "914"}:
+                                force_backoff = True
+                                persist = False
+                                self._api.set_socketPersistent(False)
+                                if self._api.parent:
+                                    self._api.parent.set_socketPersistent(False)
+                            # increment the error count if not done already
+                            if error_count == self._api_working_protocol_failures:
+                                self._api_working_protocol_failures += 1
+                            if self._api_working_protocol_failures == 1:
+                                _LOGGER.warning(
+                                    "%s error reading: %s", self.name, poll["Error"]
+                                )
+                            else:
+                                _LOGGER.debug(
+                                    "%s error reading: %s", self.name, poll["Error"]
+                                )
+                            if "Payload" in poll and poll["Payload"]:
+                                _LOGGER.debug(
+                                    "%s err payload: %s",
+                                    self.name,
+                                    poll["Payload"],
+                                )
+                        else:
+                            if "dps" in poll:
+                                poll = poll["dps"]
+                            if isinstance(poll, dict):
+                                poll["full_poll"] = full_poll
+                                yield poll
+
+                except CancelledError:
+                    self._running = False
+                    # Close the persistent connection when exiting the loop
+                    persist = False
+                    _LOGGER.debug("%s receive loop interrupted", self.name)
+                    self._api.set_socketPersistent(False)
+                    if self._api.parent:
+                        self._api.parent.set_socketPersistent(False)
+                    raise
+                except Exception as t:
+                    _LOGGER.exception(
+                        "%s receive loop error %s:%s",
+                        self.name,
+                        type(t).__name__,
+                        t,
+                    )
+                    persist = False
+                    self._api.set_socketPersistent(False)
+                    if self._api.parent:
+                        self._api.parent.set_socketPersistent(False)
+                    force_backoff = True
 
-            except CancelledError:
-                self._running = False
-                # Close the persistent connection when exiting the loop
-                persist = False
-                _LOGGER.debug("%s receive loop interrupted", self.name)
-                self._api.set_socketPersistent(False)
-                if self._api.parent:
-                    self._api.parent.set_socketPersistent(False)
-                raise
-            except Exception as t:
-                _LOGGER.exception(
-                    "%s receive loop error %s:%s",
-                    self.name,
-                    type(t).__name__,
-                    t,
-                )
-                persist = False
-                self._api.set_socketPersistent(False)
-                if self._api.parent:
-                    self._api.parent.set_socketPersistent(False)
-                force_backoff = True
-            finally:
-                if self._api_lock.locked():
-                    self._api_lock.release()
             if not self.has_returned_state:
                 force_backoff = True
             await asyncio.sleep(5 if force_backoff else 0.1)
@@ -536,10 +533,11 @@ class TuyaLocalDevice(object):
     async def async_refresh(self):
         _LOGGER.debug("Refreshing device state for %s", self.name)
         if not self._running:
-            await self._retry_on_failed_connection(
-                lambda: self._refresh_cached_state(),
-                f"Failed to refresh device state for {self.name}.",
-            )
+            async with self._api_lock:
+                await self._retry_on_failed_connection(
+                    lambda: self._refresh_cached_state(),
+                    f"Failed to refresh device state for {self.name}.",
+                )
 
     def get_property(self, dps_id):
         cached_state = self._get_cached_state()
@@ -648,31 +646,28 @@ class TuyaLocalDevice(object):
         await self._send_pending_updates()
 
     async def _send_pending_updates(self):
-        pending_properties = self._get_unsent_properties()
+        async with self._api_lock:
+            pending_properties = self._get_unsent_properties()
 
-        _LOGGER.debug(
-            "%s sending dps update: %s",
-            self.name,
-            log_json(pending_properties),
-        )
+            _LOGGER.debug(
+                "%s sending dps update: %s",
+                self.name,
+                log_json(pending_properties),
+            )
 
-        await self._retry_on_failed_connection(
-            lambda: self._set_values(pending_properties),
-            "Failed to update device state.",
-        )
+            await self._retry_on_failed_connection(
+                lambda: self._set_values(pending_properties),
+                "Failed to update device state.",
+            )
 
     def _set_values(self, properties):
-        try:
-            self._lock.acquire()
-            self._api.set_multiple_values(properties, nowait=True)
-            now = time()
-            self._last_connection = now
-            pending_updates = self._get_pending_updates()
-            for key in properties.keys():
-                pending_updates[key]["updated_at"] = now
-                pending_updates[key]["sent"] = True
-        finally:
-            self._lock.release()
+        self._api.set_multiple_values(properties, nowait=True)
+        now = time()
+        self._last_connection = now
+        pending_updates = self._get_pending_updates()
+        for key in properties.keys():
+            pending_updates[key]["updated_at"] = now
+            pending_updates[key]["sent"] = True
 
     async def _retry_on_failed_connection(self, func, error_message):
         if self._api_protocol_version_index is None: