mirror of
https://github.com/danieldemus/openhab-addons
synced 2026-07-29 12:34:21 +02:00
[matter] Fix very slow things initialization (Thread devices) (#20904)
* [matter] Fix very slow thing initialization Use subscription cache for node init. - Serialize attribute values from the local cache instead of one remote read per attribute (huge speedup on Thread). This caused a lot of unnecessary single Thread requests and slowed everything down drastically. - Default requestFromRemote to false in serializePairedNode/ sendSerializedNode. Signed-off-by: Bernhard Kaszt <github-Bernhard@kaszt.at>
This commit is contained in:
Bernhard Kaszt
@@ -11,6 +11,16 @@ import { DclOtaUpdateService, PhysicalDeviceProperties } from "@matter/main/prot
|
||||
|
||||
const logger = Logger.get("ControllerNode");
|
||||
|
||||
// Attributes that are always read fresh from the device, even when the rest of the node is served from the
|
||||
// local subscription cache (requestFromRemote=false). softwareVersion/softwareVersionString only change on a
|
||||
// firmware update, so after a device reboots onto new firmware the cached value can be stale - and it is
|
||||
// surfaced as the Thing's firmware-version property, so it must be accurate. The cost is one tiny read of a
|
||||
// rarely-changing attribute per serialization, with a fallback to the cached value if the read fails.
|
||||
const ALWAYS_FRESH_ATTRIBUTES: Record<string, ReadonlySet<string>> = {
|
||||
BasicInformation: new Set(["softwareVersion", "softwareVersionString"]),
|
||||
BridgedDeviceBasicInformation: new Set(["softwareVersion", "softwareVersionString"]),
|
||||
};
|
||||
|
||||
function extractPhysicalProperties(node: PairedNode | undefined): PhysicalDeviceProperties | undefined {
|
||||
if (!node) return undefined;
|
||||
try {
|
||||
@@ -169,6 +179,25 @@ export class ControllerNode {
|
||||
});
|
||||
this.observers.on(updateManagerEvents.updateDone, peer => {
|
||||
logger.info(`Update done for peer `, peer);
|
||||
const nodeId = peer?.nodeId;
|
||||
if (!nodeId) {
|
||||
logger.error(`Node ID not found for peer `, peer);
|
||||
return;
|
||||
}
|
||||
// matter.js only emits updateDone once the device has already rebooted onto the new firmware and
|
||||
// re-established its session, so it is reachable now and we refresh immediately. The serialized node
|
||||
// (and thus the Thing's firmware-version property) is normally built from the local subscription cache
|
||||
// (requestFromRemote=false), which may still hold the pre-update version until the subscription has
|
||||
// re-primed; forcing a fresh remote read makes the reported version reflect the firmware just installed,
|
||||
// rebuilds the structure/channels and brings the Thing (set OFFLINE while applying) back ONLINE in one
|
||||
// step. If the device should be briefly unreachable again, sendSerializedNode triggers a reconnect on
|
||||
// failure and ALWAYS_FRESH_ATTRIBUTES corrects the version on the next serialization. OTA is rare, so a
|
||||
// one-off full read is cheap.
|
||||
try {
|
||||
this.sendSerializedNode(this.getNode(nodeId), undefined, true);
|
||||
} catch (e) {
|
||||
logger.error(`Could not refresh node ${nodeId} after OTA update: ${e}`);
|
||||
}
|
||||
});
|
||||
|
||||
// Query for updates now once
|
||||
@@ -407,9 +436,13 @@ export class ControllerNode {
|
||||
* Serializes a node and sends it to the web socket
|
||||
* @param node
|
||||
* @param endpointId Optional endpointId to serialize. If omitted, all endpoints will be serialized.
|
||||
* @param requestFromRemote When false (the default) values are taken from the local subscription cache instead of
|
||||
* reading every attribute individually from the device. Since the node is subscribed (autoSubscribe) the cache is
|
||||
* continuously kept up to date by the device, so a remote read per attribute is redundant and very expensive on
|
||||
* slow Thread meshes. Pass true only to force a fresh remote read of every attribute.
|
||||
*/
|
||||
sendSerializedNode(node: PairedNode, endpointId?: number) {
|
||||
this.serializePairedNode(node, endpointId)
|
||||
sendSerializedNode(node: PairedNode, endpointId?: number, requestFromRemote: boolean = false) {
|
||||
this.serializePairedNode(node, endpointId, requestFromRemote)
|
||||
.then(data => {
|
||||
this.ws.sendEvent(EventType.NodeData, data);
|
||||
})
|
||||
@@ -424,9 +457,13 @@ export class ControllerNode {
|
||||
* Serializes a node and returns the json object
|
||||
* @param node
|
||||
* @param endpointId Optional endpointId to serialize. If omitted, the root endpoint will be serialized.
|
||||
* @param requestFromRemote When false (the default) attribute values are served from the local subscription cache.
|
||||
* Reading every attribute remotely (true) issues one network request per attribute and is extremely slow for
|
||||
* Thread / sleepy devices; the autoSubscribe cache already holds the current values. Fabric-scoped attributes are
|
||||
* always read from the device regardless of this flag (enforced by matter.js).
|
||||
* @returns
|
||||
*/
|
||||
async serializePairedNode(node: PairedNode, endpointId?: number, requestFromRemote: boolean = true) {
|
||||
async serializePairedNode(node: PairedNode, endpointId?: number, requestFromRemote: boolean = false) {
|
||||
if (!this.commissioningController) {
|
||||
throw new Error("CommissioningController not initialized");
|
||||
}
|
||||
@@ -454,7 +491,29 @@ export class ControllerNode {
|
||||
if (/^\d+$/.test(attributeName)) continue;
|
||||
const attribute = cluster.attributes[attributeName];
|
||||
if (!attribute) continue;
|
||||
const attributeValue = await attribute.get(requestFromRemote);
|
||||
// Force a fresh remote read for attributes that must never be served stale from the cache
|
||||
// (see ALWAYS_FRESH_ATTRIBUTES), but fall back to the cached value if that read fails so one
|
||||
// flaky read of a single attribute never fails the whole node serialization.
|
||||
const forceFresh =
|
||||
!requestFromRemote && (ALWAYS_FRESH_ATTRIBUTES[cluster.name]?.has(attributeName) ?? false);
|
||||
let attributeValue: any;
|
||||
if (forceFresh) {
|
||||
try {
|
||||
attributeValue = await attribute.get(true);
|
||||
} catch (e) {
|
||||
logger.debug(`Fresh read of ${cluster.name}.${attributeName} failed, using cache: ${e}`);
|
||||
try {
|
||||
attributeValue = await attribute.get(false);
|
||||
} catch (e2) {
|
||||
logger.debug(
|
||||
`Cache read of ${cluster.name}.${attributeName} also failed, skipping: ${e2}`,
|
||||
);
|
||||
attributeValue = undefined;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
attributeValue = await attribute.get(requestFromRemote);
|
||||
}
|
||||
logger.debug(`Attribute ${attributeName} value: ${attributeValue}`);
|
||||
if (attributeValue !== undefined) {
|
||||
clusterData[attributeName] = attributeValue;
|
||||
|
||||
+3
-5
@@ -423,9 +423,7 @@ public class MatterWebsocketClient implements WebSocketListener, MatterWebsocket
|
||||
|
||||
protected CompletableFuture<JsonElement> sendMessage(String namespace, String functionName, @Nullable Object args[],
|
||||
int timeoutSeconds) {
|
||||
if (timeoutSeconds <= 0) {
|
||||
timeoutSeconds = REQUEST_TIMEOUT_SECONDS;
|
||||
}
|
||||
final int effectiveTimeoutSeconds = timeoutSeconds > 0 ? timeoutSeconds : REQUEST_TIMEOUT_SECONDS;
|
||||
CompletableFuture<JsonElement> responseFuture = new CompletableFuture<>();
|
||||
|
||||
Session session = this.session;
|
||||
@@ -446,9 +444,9 @@ public class MatterWebsocketClient implements WebSocketListener, MatterWebsocket
|
||||
CompletableFuture<JsonElement> future = pendingRequests.remove(requestId);
|
||||
if (future != null && !future.isDone()) {
|
||||
future.completeExceptionally(new TimeoutException(String.format(
|
||||
"Request %s:%s timed out after %d seconds", namespace, functionName, REQUEST_TIMEOUT_SECONDS)));
|
||||
"Request %s:%s timed out after %d seconds", namespace, functionName, effectiveTimeoutSeconds)));
|
||||
}
|
||||
}, timeoutSeconds, TimeUnit.SECONDS);
|
||||
}, effectiveTimeoutSeconds, TimeUnit.SECONDS);
|
||||
|
||||
return responseFuture;
|
||||
}
|
||||
|
||||
+5
@@ -243,6 +243,11 @@ public class ControllerHandler extends BaseBridgeHandler implements MatterClient
|
||||
requestAllNodeDataIfNeeded(message.nodeId);
|
||||
break;
|
||||
case STRUCTURECHANGED:
|
||||
// A structure change means the node's endpoints/clusters changed, so the channels must be rebuilt
|
||||
// from fresh data. Drop the "already enumerated" marker first: otherwise the data request that the
|
||||
// reconnect's Connected event triggers is skipped for sleepy nodes (see requestAllNodeDataIfNeeded),
|
||||
// leaving the Thing online but with stale channels. Same marker handling as removeNode/dispose.
|
||||
enumeratedNodes.remove(message.nodeId);
|
||||
updateNode(message.nodeId);
|
||||
break;
|
||||
case DECOMMISSIONED:
|
||||
|
||||
+56
@@ -13,11 +13,23 @@
|
||||
package org.openhab.binding.matter.internal.client;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
|
||||
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
import java.math.BigInteger;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.TimeoutException;
|
||||
|
||||
import org.eclipse.jdt.annotation.NonNullByDefault;
|
||||
import org.eclipse.jetty.websocket.api.RemoteEndpoint;
|
||||
import org.eclipse.jetty.websocket.api.Session;
|
||||
import org.eclipse.jetty.websocket.api.WebSocketPolicy;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.openhab.binding.matter.internal.client.dto.Endpoint;
|
||||
@@ -30,6 +42,7 @@ import org.openhab.binding.matter.internal.client.dto.ws.AttributeChangedMessage
|
||||
import org.openhab.binding.matter.internal.client.dto.ws.EventTriggeredMessage;
|
||||
import org.openhab.binding.matter.internal.client.dto.ws.Message;
|
||||
|
||||
import com.google.gson.JsonElement;
|
||||
import com.google.gson.JsonObject;
|
||||
|
||||
/**
|
||||
@@ -313,4 +326,47 @@ class MatterWebsocketClientTest {
|
||||
assertNotNull(occupancyBitmap);
|
||||
assertEquals(true, occupancyBitmap.occupied);
|
||||
}
|
||||
|
||||
/**
|
||||
* When sendMessage is called without an established WebSocket session, the returned future must fail fast with a
|
||||
* {@link MatterRequestException} ("No valid session") instead of hanging until the timeout. This is the path taken
|
||||
* while the node process is (re)starting or after a disconnect.
|
||||
*/
|
||||
@Test
|
||||
void testSendMessageWithoutSessionCompletesExceptionally() {
|
||||
// No onWebSocketConnect() called, so there is no active session.
|
||||
CompletableFuture<JsonElement> future = client.sendMessage("nodes", "listNodes", new Object[0], 1);
|
||||
ExecutionException ex = assertThrows(ExecutionException.class, () -> future.get(5, TimeUnit.SECONDS));
|
||||
assertInstanceOf(MatterRequestException.class, ex.getCause());
|
||||
Throwable cause = ex.getCause();
|
||||
assertNotNull(cause);
|
||||
assertTrue(cause.getMessage().contains("No valid session"));
|
||||
}
|
||||
|
||||
/**
|
||||
* A non-positive timeout must fall back to the default request timeout (REQUEST_TIMEOUT_SECONDS), preserving the
|
||||
* previous behaviour. We verify this indirectly: with a 0s timeout the request must stay pending well beyond a
|
||||
* short
|
||||
* wait (here 2s), proving 0 was not used as the actual delay.
|
||||
*/
|
||||
@Test
|
||||
void testSendMessageNonPositiveTimeoutFallsBackToDefault() {
|
||||
connectMockSession();
|
||||
// A timeout <= 0 must fall back to the default, so the request stays pending well beyond 2s.
|
||||
CompletableFuture<JsonElement> future = client.sendMessage("nodes", "listNodes", new Object[0], 0);
|
||||
assertThrows(TimeoutException.class, () -> future.get(2, TimeUnit.SECONDS));
|
||||
}
|
||||
|
||||
/**
|
||||
* Installs a mocked WebSocket session on the client so sendMessage reaches the request-scheduling path (instead of
|
||||
* failing early with "No valid session"). The remote endpoint is stubbed so the outgoing message is silently
|
||||
* discarded; no real socket or node process is involved.
|
||||
*/
|
||||
private void connectMockSession() {
|
||||
Session session = mock(Session.class);
|
||||
when(session.getPolicy()).thenReturn(WebSocketPolicy.newClientPolicy());
|
||||
RemoteEndpoint remote = mock(RemoteEndpoint.class);
|
||||
when(session.getRemote()).thenReturn(remote);
|
||||
client.onWebSocketConnect(session);
|
||||
}
|
||||
}
|
||||
|
||||
+173
@@ -0,0 +1,173 @@
|
||||
/*
|
||||
* Copyright (c) 2010-2026 Contributors to the openHAB project
|
||||
*
|
||||
* See the NOTICE file(s) distributed with this work for additional
|
||||
* information.
|
||||
*
|
||||
* This program and the accompanying materials are made available under the
|
||||
* terms of the Eclipse Public License 2.0 which is available at
|
||||
* http://www.eclipse.org/legal/epl-2.0
|
||||
*
|
||||
* SPDX-License-Identifier: EPL-2.0
|
||||
*/
|
||||
package org.openhab.binding.matter.internal.handler;
|
||||
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
import java.lang.reflect.Field;
|
||||
import java.math.BigInteger;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
|
||||
import org.eclipse.jdt.annotation.NonNullByDefault;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.Mock;
|
||||
import org.mockito.MockitoAnnotations;
|
||||
import org.openhab.binding.matter.internal.MatterFirmwareProvider;
|
||||
import org.openhab.binding.matter.internal.client.MatterWebsocketService;
|
||||
import org.openhab.binding.matter.internal.client.dto.ws.NodeState;
|
||||
import org.openhab.binding.matter.internal.client.dto.ws.NodeStateMessage;
|
||||
import org.openhab.binding.matter.internal.controller.MatterControllerClient;
|
||||
import org.openhab.binding.matter.internal.util.TranslationService;
|
||||
import org.openhab.core.thing.Bridge;
|
||||
|
||||
/**
|
||||
* Tests for {@link ControllerHandler} node refresh behavior.
|
||||
*
|
||||
* @author Bernhard Kaszt - Initial contribution
|
||||
*/
|
||||
@NonNullByDefault
|
||||
public class ControllerHandlerTest {
|
||||
|
||||
private static final BigInteger NODE_ID = BigInteger.valueOf(42);
|
||||
|
||||
@Mock
|
||||
@NonNullByDefault({})
|
||||
private Bridge bridge;
|
||||
@Mock
|
||||
@NonNullByDefault({})
|
||||
private MatterWebsocketService websocketService;
|
||||
@Mock
|
||||
@NonNullByDefault({})
|
||||
private TranslationService translationService;
|
||||
@Mock
|
||||
@NonNullByDefault({})
|
||||
private MatterFirmwareProvider firmwareProvider;
|
||||
@Mock
|
||||
@NonNullByDefault({})
|
||||
private MatterControllerClient client;
|
||||
|
||||
@NonNullByDefault({})
|
||||
private ControllerHandler handler;
|
||||
|
||||
@BeforeEach
|
||||
@SuppressWarnings("null")
|
||||
public void setUp() throws Exception {
|
||||
MockitoAnnotations.openMocks(this);
|
||||
when(bridge.getThings()).thenReturn(List.of());
|
||||
when(client.initializeNode(any(), any())).thenReturn(completedVoid());
|
||||
when(client.requestAllNodeData(any())).thenReturn(completedVoid());
|
||||
|
||||
handler = new ControllerHandler(bridge, websocketService, translationService, firmwareProvider);
|
||||
// The constructor creates a real MatterControllerClient; swap it for the mock so we can verify calls.
|
||||
setField(handler, "client", client);
|
||||
// Mark the handler ready so updateNode() does not short-circuit.
|
||||
setField(handler, "ready", Boolean.TRUE);
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
public void tearDown() {
|
||||
ControllerHandler localHandler = handler;
|
||||
if (localHandler != null) {
|
||||
localHandler.dispose();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A structure change must force a full data refresh even for a sleepy node that was already enumerated.
|
||||
* matter.js reports a structure change while the node stays connected; the binding reacts by reconnecting
|
||||
* (updateNode -> initializeNode), whose resulting Connected event drives the data request. For sleepy nodes
|
||||
* requestAllNodeDataIfNeeded skips that request, so without clearing the "enumerated" marker on the structure
|
||||
* change the channels are never rebuilt.
|
||||
*/
|
||||
@Test
|
||||
public void structureChangeRefreshesSleepyNode() throws Exception {
|
||||
linkEnumeratedNode(NODE_ID, true);
|
||||
|
||||
handler.onEvent(nodeState(NODE_ID, NodeState.STRUCTURECHANGED));
|
||||
handler.onEvent(nodeState(NODE_ID, NodeState.CONNECTED));
|
||||
|
||||
verify(client).requestAllNodeData(NODE_ID);
|
||||
}
|
||||
|
||||
/**
|
||||
* Control: a non-sleepy node is refreshed on reconnect anyway, so the structure change path must keep working
|
||||
* for it too.
|
||||
*/
|
||||
@Test
|
||||
public void structureChangeRefreshesNonSleepyNode() throws Exception {
|
||||
linkEnumeratedNode(NODE_ID, false);
|
||||
|
||||
handler.onEvent(nodeState(NODE_ID, NodeState.STRUCTURECHANGED));
|
||||
handler.onEvent(nodeState(NODE_ID, NodeState.CONNECTED));
|
||||
|
||||
verify(client).requestAllNodeData(NODE_ID);
|
||||
}
|
||||
|
||||
/**
|
||||
* Control: a plain reconnect (Connected without a preceding structure change) of an already enumerated sleepy
|
||||
* node must still skip the expensive re-enumeration - that is the optimization the structure-change fix must
|
||||
* not regress.
|
||||
*/
|
||||
@Test
|
||||
public void reconnectSkipsRefreshForSleepyNode() throws Exception {
|
||||
linkEnumeratedNode(NODE_ID, true);
|
||||
|
||||
handler.onEvent(nodeState(NODE_ID, NodeState.CONNECTED));
|
||||
|
||||
verify(client, never()).requestAllNodeData(any());
|
||||
}
|
||||
|
||||
private void linkEnumeratedNode(BigInteger nodeId, boolean sleepy) throws Exception {
|
||||
NodeHandler nodeHandler = mock(NodeHandler.class);
|
||||
when(nodeHandler.shouldRefreshOnReconnect()).thenReturn(!sleepy);
|
||||
|
||||
Map<BigInteger, NodeHandler> linkedNodes = getField(handler, "linkedNodes");
|
||||
linkedNodes.put(nodeId, nodeHandler);
|
||||
Set<BigInteger> enumeratedNodes = getField(handler, "enumeratedNodes");
|
||||
enumeratedNodes.add(nodeId);
|
||||
}
|
||||
|
||||
private static NodeStateMessage nodeState(BigInteger nodeId, NodeState state) {
|
||||
NodeStateMessage message = new NodeStateMessage();
|
||||
message.nodeId = nodeId;
|
||||
message.state = state;
|
||||
return message;
|
||||
}
|
||||
|
||||
@SuppressWarnings("null")
|
||||
private static CompletableFuture<Void> completedVoid() {
|
||||
return CompletableFuture.<Void> completedFuture(null);
|
||||
}
|
||||
|
||||
private static void setField(Object target, String name, Object value) throws Exception {
|
||||
Field field = ControllerHandler.class.getDeclaredField(name);
|
||||
field.setAccessible(true);
|
||||
field.set(target, value);
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private static <T> T getField(Object target, String name) throws Exception {
|
||||
Field field = ControllerHandler.class.getDeclaredField(name);
|
||||
field.setAccessible(true);
|
||||
return (T) field.get(target);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user