Fix non-thread-safe operations in amcrest (#116859)

* Fix non-thread-safe operations in amcrest

fixes #116850

* fix locking

* fix locking

* fix locking

homeassistant/components/amcrest/__init__.py

@@ -35,7 +35,7 @@ from homeassistant.const import (
     HTTP_BASIC_AUTHENTICATION,
     Platform,
 )
-from homeassistant.core import HomeAssistant, ServiceCall
+from homeassistant.core import HomeAssistant, ServiceCall, callback
 from homeassistant.exceptions import Unauthorized, UnknownUser
 from homeassistant.helpers import discovery
 import homeassistant.helpers.config_validation as cv
@@ -177,7 +177,8 @@ class AmcrestChecker(ApiWrapper):
         """Return event flag that indicates if camera's API is responding."""
         return self._async_wrap_event_flag
 
-    def _start_recovery(self) -> None:
+    @callback
+    def _async_start_recovery(self) -> None:
         self.available_flag.clear()
         self.async_available_flag.clear()
         async_dispatcher_send(
@@ -222,41 +223,89 @@ class AmcrestChecker(ApiWrapper):
             yield
         except LoginError as ex:
             async with self._async_wrap_lock:
-                self._handle_offline(ex)
+                self._async_handle_offline(ex)
             raise
         except AmcrestError:
             async with self._async_wrap_lock:
-                self._handle_error()
+                self._async_handle_error()
             raise
         async with self._async_wrap_lock:
-            self._set_online()
+            self._async_set_online()
 
-    def _handle_offline(self, ex: Exception) -> None:
+    def _handle_offline_thread_safe(self, ex: Exception) -> bool:
+        """Handle camera offline status shared between threads and event loop.
+
+        Returns if the camera was online as a bool.
+        """
         with self._wrap_lock:
             was_online = self.available
             was_login_err = self._wrap_login_err
             self._wrap_login_err = True
         if not was_login_err:
             _LOGGER.error("%s camera offline: Login error: %s", self._wrap_name, ex)
-        if was_online:
+        return was_online
-            self._start_recovery()
 
-    def _handle_error(self) -> None:
+    def _handle_offline(self, ex: Exception) -> None:
+        """Handle camera offline status from a thread."""
+        if self._handle_offline_thread_safe(ex):
+            self._hass.loop.call_soon_threadsafe(self._async_start_recovery)
+
+    @callback
+    def _async_handle_offline(self, ex: Exception) -> None:
+        if self._handle_offline_thread_safe(ex):
+            self._async_start_recovery()
+
+    def _handle_error_thread_safe(self) -> bool:
+        """Handle camera error status shared between threads and event loop.
+
+        Returns if the camera was online and is now offline as
+        a bool.
+        """
         with self._wrap_lock:
             was_online = self.available
             errs = self._wrap_errors = self._wrap_errors + 1
             offline = not self.available
         _LOGGER.debug("%s camera errs: %i", self._wrap_name, errs)
-        if was_online and offline:
+        return was_online and offline
-            _LOGGER.error("%s camera offline: Too many errors", self._wrap_name)
-            self._start_recovery()
 
-    def _set_online(self) -> None:
+    def _handle_error(self) -> None:
+        """Handle camera error status from a thread."""
+        if self._handle_error_thread_safe():
+            _LOGGER.error("%s camera offline: Too many errors", self._wrap_name)
+            self._hass.loop.call_soon_threadsafe(self._async_start_recovery)
+
+    @callback
+    def _async_handle_error(self) -> None:
+        """Handle camera error status from the event loop."""
+        if self._handle_error_thread_safe():
+            _LOGGER.error("%s camera offline: Too many errors", self._wrap_name)
+            self._async_start_recovery()
+
+    def _set_online_thread_safe(self) -> bool:
+        """Set camera online status shared between threads and event loop.
+
+        Returns if the camera was offline as a bool.
+        """
         with self._wrap_lock:
             was_offline = not self.available
             self._wrap_errors = 0
             self._wrap_login_err = False
-        if was_offline:
+        return was_offline
+
+    def _set_online(self) -> None:
+        """Set camera online status from a thread."""
+        if self._set_online_thread_safe():
+            self._hass.loop.call_soon_threadsafe(self._async_signal_online)
+
+    @callback
+    def _async_set_online(self) -> None:
+        """Set camera online status from the event loop."""
+        if self._set_online_thread_safe():
+            self._async_signal_online()
+
+    @callback
+    def _async_signal_online(self) -> None:
+        """Signal that camera is back online."""
         assert self._unsub_recheck is not None
         self._unsub_recheck()
         self._unsub_recheck = None

homeassistant/components/amcrest/camera.py

@@ -16,7 +16,7 @@ import voluptuous as vol
 from homeassistant.components.camera import Camera, CameraEntityFeature
 from homeassistant.components.ffmpeg import FFmpegManager, get_ffmpeg_manager
 from homeassistant.const import ATTR_ENTITY_ID, CONF_NAME, STATE_OFF, STATE_ON
-from homeassistant.core import HomeAssistant
+from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.aiohttp_client import (
     async_aiohttp_proxy_stream,
@@ -325,7 +325,8 @@ class AmcrestCam(Camera):
 
     # Other Entity method overrides
 
-    async def async_on_demand_update(self) -> None:
+    @callback
+    def async_on_demand_update(self) -> None:
         """Update state."""
         self.async_schedule_update_ha_state(True)