Remove executor limit (#621)

Author: marcelveldt

matter_server/server/device_controller.py

@@ -6,7 +6,6 @@
 
 import asyncio
 from collections import deque
-from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime
 from functools import partial
 import logging
@@ -114,11 +113,9 @@ def __init__(
         self._aiozc: AsyncZeroconf | None = None
         self._fallback_node_scanner_timer: asyncio.TimerHandle | None = None
         self._fallback_node_scanner_task: asyncio.Task | None = None
-        self._sdk_executor = ThreadPoolExecutor(
-            max_workers=1, thread_name_prefix="SDKExecutor"
-        )
-        self._node_setup_throttle = asyncio.Semaphore(10)
+        self._node_setup_throttle = asyncio.Semaphore(5)
         self._mdns_event_timer: dict[str, asyncio.TimerHandle] = {}
+        self._node_lock: dict[int, asyncio.Lock] = {}
 
     async def initialize(self) -> None:
         """Async initialize of controller."""
@@ -437,14 +434,15 @@ async def open_commissioning_window(
         if discriminator is None:
             discriminator = randint(0, 4095)  # noqa: S311
 
-        sdk_result = await self._call_sdk(
-            self.chip_controller.OpenCommissioningWindow,
-            nodeid=node_id,
-            timeout=timeout,
-            iteration=iteration,
-            discriminator=discriminator,
-            option=option,
-        )
+        async with self._get_node_lock(node_id):
+            sdk_result = await self._call_sdk(
+                self.chip_controller.OpenCommissioningWindow,
+                nodeid=node_id,
+                timeout=timeout,
+                iteration=iteration,
+                discriminator=discriminator,
+                option=option,
+            )
         self._known_commissioning_params[node_id] = params = CommissioningParameters(
             setup_pin_code=sdk_result.setupPinCode,
             setup_manual_code=sdk_result.setupManualCode,
@@ -504,13 +502,14 @@ async def interview_node(self, node_id: int) -> None:
 
         try:
             LOGGER.info("Interviewing node: %s", node_id)
-            read_response: Attribute.AsyncReadTransaction.ReadResponse = (
-                await self.chip_controller.Read(
-                    nodeid=node_id,
-                    attributes="*",
-                    fabricFiltered=False,
+            async with self._get_node_lock(node_id):
+                read_response: Attribute.AsyncReadTransaction.ReadResponse = (
+                    await self.chip_controller.Read(
+                        nodeid=node_id,
+                        attributes="*",
+                        fabricFiltered=False,
+                    )
                 )
-            )
         except ChipStackError as err:
             raise NodeInterviewFailed(f"Failed to interview node {node_id}") from err
 
@@ -566,14 +565,15 @@ async def send_device_command(
         cluster_cls: Cluster = ALL_CLUSTERS[cluster_id]
         command_cls = getattr(cluster_cls.Commands, command_name)
         command = dataclass_from_dict(command_cls, payload, allow_sdk_types=True)
-        return await self.chip_controller.SendCommand(
-            nodeid=node_id,
-            endpoint=endpoint_id,
-            payload=command,
-            responseType=response_type,
-            timedRequestTimeoutMs=timed_request_timeout_ms,
-            interactionTimeoutMs=interaction_timeout_ms,
-        )
+        async with self._get_node_lock(node_id):
+            return await self.chip_controller.SendCommand(
+                nodeid=node_id,
+                endpoint=endpoint_id,
+                payload=command,
+                responseType=response_type,
+                timedRequestTimeoutMs=timed_request_timeout_ms,
+                interactionTimeoutMs=interaction_timeout_ms,
+            )
 
     @api_command(APICommand.READ_ATTRIBUTE)
     async def read_attribute(
@@ -595,21 +595,22 @@ async def read_attribute(
 
         future = self.server.loop.create_future()
         device = await self._resolve_node(node_id)
-        Attribute.Read(
-            future=future,
-            eventLoop=self.server.loop,
-            device=device.deviceProxy,
-            devCtrl=self.chip_controller,
-            attributes=[
-                Attribute.AttributePath(
-                    EndpointId=endpoint_id,
-                    ClusterId=cluster_id,
-                    AttributeId=attribute_id,
-                )
-            ],
-            fabricFiltered=fabric_filtered,
-        ).raise_on_error()
-        result: Attribute.AsyncReadTransaction.ReadResponse = await future
+        async with self._get_node_lock(node_id):
+            Attribute.Read(
+                future=future,
+                eventLoop=self.server.loop,
+                device=device.deviceProxy,
+                devCtrl=self.chip_controller,
+                attributes=[
+                    Attribute.AttributePath(
+                        EndpointId=endpoint_id,
+                        ClusterId=cluster_id,
+                        AttributeId=attribute_id,
+                    )
+                ],
+                fabricFiltered=fabric_filtered,
+            ).raise_on_error()
+            result: Attribute.AsyncReadTransaction.ReadResponse = await future
         read_atributes = parse_attributes_from_read_result(result.tlvAttributes)
         # update cached info in node attributes
         self._nodes[node_id].attributes.update(read_atributes)
@@ -639,10 +640,11 @@ async def write_attribute(
             value_type=attribute.attribute_type.Type,
             allow_sdk_types=True,
         )
-        return await self.chip_controller.WriteAttribute(
-            nodeid=node_id,
-            attributes=[(endpoint_id, attribute)],
-        )
+        async with self._get_node_lock(node_id):
+            return await self.chip_controller.WriteAttribute(
+                nodeid=node_id,
+                attributes=[(endpoint_id, attribute)],
+            )
 
     @api_command(APICommand.REMOVE_NODE)
     async def remove_node(self, node_id: int) -> None:
@@ -1001,16 +1003,17 @@ def resubscription_succeeded(
         else:
             interval_ceiling = NODE_SUBSCRIPTION_CEILING_THREAD
         self._last_subscription_attempt[node_id] = 0
-        sub: Attribute.SubscriptionTransaction = await self.chip_controller.Read(
-            node_id,
-            attributes="*",
-            events=[("*", 1)],
-            returnClusterObject=False,
-            reportInterval=(interval_floor, interval_ceiling),
-            fabricFiltered=False,
-            keepSubscriptions=True,
-            autoResubscribe=True,
-        )
+        async with self._get_node_lock(node_id):
+            sub: Attribute.SubscriptionTransaction = await self.chip_controller.Read(
+                node_id,
+                attributes="*",
+                events=[("*", 1)],
+                returnClusterObject=False,
+                reportInterval=(interval_floor, interval_ceiling),
+                fabricFiltered=False,
+                keepSubscriptions=True,
+                autoResubscribe=True,
+            )
 
         sub.SetAttributeUpdateCallback(attribute_updated_callback)
         sub.SetEventUpdateCallback(event_callback)
@@ -1050,7 +1053,7 @@ async def _call_sdk(
         return cast(
             _T,
             await self.server.loop.run_in_executor(
-                self._sdk_executor,
+                None,
                 partial(target, *args, **kwargs),
             ),
         )
@@ -1136,12 +1139,13 @@ async def _resolve_node(
                 retries,
             )
             time_start = time.time()
-            return await self._call_sdk(
-                self.chip_controller.GetConnectedDeviceSync,
-                nodeid=node_id,
-                allowPASE=False,
-                timeoutMs=None,
-            )
+            async with self._get_node_lock(node_id):
+                return await self._call_sdk(
+                    self.chip_controller.GetConnectedDeviceSync,
+                    nodeid=node_id,
+                    allowPASE=False,
+                    timeoutMs=None,
+                )
         except ChipStackError as err:
             if attempt >= retries:
                 # when we're out of retries, raise NodeNotResolving
@@ -1352,3 +1356,9 @@ def run_fallback_node_scanner() -> None:
         self._fallback_node_scanner_timer = self.server.loop.call_later(
             FALLBACK_NODE_SCANNER_INTERVAL, run_fallback_node_scanner
         )
+
+    def _get_node_lock(self, node_id: int) -> asyncio.Lock:
+        """Return lock for given node."""
+        if node_id not in self._node_lock:
+            self._node_lock[node_id] = asyncio.Lock()
+        return self._node_lock[node_id]