diff --git a/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java b/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java index 7e58a2cb9f..2237e06f43 100644 --- a/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java +++ b/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java @@ -19,8 +19,17 @@ import io.grpc.HttpConnectProxiedSocketAddress; import io.grpc.ManagedChannel; import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts; import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder; +import io.grpc.netty.shaded.io.netty.channel.Channel; +import io.grpc.netty.shaded.io.netty.channel.EventLoopGroup; +import io.grpc.netty.shaded.io.netty.channel.epoll.Epoll; +import io.grpc.netty.shaded.io.netty.channel.epoll.EpollEventLoopGroup; +import io.grpc.netty.shaded.io.netty.channel.epoll.EpollSocketChannel; +import io.grpc.netty.shaded.io.netty.channel.nio.NioEventLoopGroup; +import io.grpc.netty.shaded.io.netty.channel.socket.nio.NioSocketChannel; import io.grpc.netty.shaded.io.netty.handler.ssl.SslContextBuilder; +import io.grpc.netty.shaded.io.netty.util.concurrent.DefaultThreadFactory; import io.grpc.stub.StreamObserver; +import jakarta.annotation.PreDestroy; import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; @@ -84,8 +93,14 @@ public class EdgeGrpcClient implements EdgeRpcClient { private ManagedChannel channel; + private final EventLoopGroup workerGroup = createWorkerGroup(); + private StreamObserver inputStream; + private volatile boolean connected; + + private volatile boolean streamActive; + private static final ReentrantLock uplinkMsgLock = new ReentrantLock(); @Override @@ -95,7 +110,11 @@ public class EdgeGrpcClient implements EdgeRpcClient { Consumer onEdgeUpdate, Consumer onDownlink, Consumer onError) { + connected = false; + streamActive = false; NettyChannelBuilder builder = NettyChannelBuilder.forAddress(rpcHost, rpcPort) + .eventLoopGroup(workerGroup) + .channelType(channelType()) .maxInboundMessageSize(maxInboundMessageSize) .keepAliveTime(keepAliveTimeSec, TimeUnit.SECONDS) .keepAliveTimeout(keepAliveTimeoutSec, TimeUnit.SECONDS) @@ -131,6 +150,7 @@ public class EdgeGrpcClient implements EdgeRpcClient { EdgeRpcServiceGrpc.EdgeRpcServiceStub stub = EdgeRpcServiceGrpc.newStub(channel); log.info("[{}] Sending a connect request to the TB!", edgeKey); this.inputStream = stub.withCompression("gzip").handleMsgs(initOutputStream(edgeKey, onUplinkResponse, onEdgeUpdate, onDownlink, onError)); + streamActive = true; this.inputStream.onNext(RequestMsg.newBuilder() .setMsgType(RequestMsgType.CONNECT_RPC_MESSAGE) .setConnectRequestMsg(ConnectRequestMsg.newBuilder() @@ -146,6 +166,20 @@ public class EdgeGrpcClient implements EdgeRpcClient { return EdgeVersionComparator.getNewestEdgeVersion(); } + private static EventLoopGroup createWorkerGroup() { + DefaultThreadFactory threadFactory = new DefaultThreadFactory("edge-grpc-worker", true); + return Epoll.isAvailable() ? new EpollEventLoopGroup(1, threadFactory) : new NioEventLoopGroup(1, threadFactory); + } + + private static Class channelType() { + return Epoll.isAvailable() ? EpollSocketChannel.class : NioSocketChannel.class; + } + + @PreDestroy + public void destroy() { + workerGroup.shutdownGracefully(); + } + private StreamObserver initOutputStream(String edgeKey, Consumer onUplinkResponse, Consumer onEdgeUpdate, @@ -162,8 +196,10 @@ public class EdgeGrpcClient implements EdgeRpcClient { serverMaxInboundMessageSize = connectResponseMsg.getMaxInboundMessageSize(); } log.info("[{}] Configuration received: {}", edgeKey, connectResponseMsg.getConfiguration()); + connected = true; onEdgeUpdate.accept(connectResponseMsg.getConfiguration()); } else { + connected = false; log.error("[{}] Failed to establish the connection! Code: {}. Error message: {}.", edgeKey, connectResponseMsg.getResponseCode(), connectResponseMsg.getErrorMsg()); try { EdgeGrpcClient.this.disconnect(true); @@ -186,6 +222,8 @@ public class EdgeGrpcClient implements EdgeRpcClient { @Override public void onError(Throwable t) { + connected = false; + streamActive = false; log.warn("[{}] Stream was terminated due to error:", edgeKey, t); try { EdgeGrpcClient.this.disconnect(true); @@ -197,6 +235,8 @@ public class EdgeGrpcClient implements EdgeRpcClient { @Override public void onCompleted() { + connected = false; + streamActive = false; log.info("[{}] Stream was closed and completed successfully!", edgeKey); } }; @@ -204,6 +244,8 @@ public class EdgeGrpcClient implements EdgeRpcClient { @Override public void disconnect(boolean onError) throws InterruptedException { + connected = false; + streamActive = false; if (!onError) { try { if (inputStream != null) { @@ -236,10 +278,19 @@ public class EdgeGrpcClient implements EdgeRpcClient { } } + @Override + public boolean isConnected() { + return connected; + } + @Override public void sendUplinkMsg(UplinkMsg msg) { uplinkMsgLock.lock(); try { + if (!streamActive) { + log.debug("Uplink msg is skipped, the cloud session is not established: {}", msg); + return; + } this.inputStream.onNext(RequestMsg.newBuilder() .setMsgType(RequestMsgType.UPLINK_RPC_MESSAGE) .setUplinkMsg(msg) @@ -253,6 +304,10 @@ public class EdgeGrpcClient implements EdgeRpcClient { public void sendSyncRequestMsg(boolean fullSyncRequired) { uplinkMsgLock.lock(); try { + if (!streamActive) { + log.debug("Sync request msg is skipped, the cloud session is not established"); + return; + } SyncRequestMsg syncRequestMsg = SyncRequestMsg.newBuilder() .setFullSync(fullSyncRequired) .build(); @@ -269,6 +324,10 @@ public class EdgeGrpcClient implements EdgeRpcClient { public void sendDownlinkResponseMsg(DownlinkResponseMsg downlinkResponseMsg) { uplinkMsgLock.lock(); try { + if (!streamActive) { + log.debug("Downlink response msg is skipped, the cloud session is not established: {}", downlinkResponseMsg); + return; + } this.inputStream.onNext(RequestMsg.newBuilder() .setMsgType(RequestMsgType.UPLINK_RPC_MESSAGE) .setDownlinkResponseMsg(downlinkResponseMsg) diff --git a/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeRpcClient.java b/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeRpcClient.java index a1a7b5462f..55bdcac04e 100644 --- a/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeRpcClient.java +++ b/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeRpcClient.java @@ -34,6 +34,8 @@ public interface EdgeRpcClient { void disconnect(boolean onError) throws InterruptedException; + boolean isConnected(); + void sendSyncRequestMsg(boolean fullSyncRequired); void sendUplinkMsg(UplinkMsg uplinkMsg); diff --git a/common/edge-api/src/test/java/org/thingsboard/edge/rpc/EdgeGrpcClientLeakTest.java b/common/edge-api/src/test/java/org/thingsboard/edge/rpc/EdgeGrpcClientLeakTest.java new file mode 100644 index 0000000000..9fc66ed33e --- /dev/null +++ b/common/edge-api/src/test/java/org/thingsboard/edge/rpc/EdgeGrpcClientLeakTest.java @@ -0,0 +1,168 @@ +/** + * Copyright © 2016-2026 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.edge.rpc; + +import io.grpc.Server; +import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder; +import io.grpc.netty.shaded.io.netty.buffer.PooledByteBufAllocator; +import io.grpc.stub.StreamObserver; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.springframework.test.util.ReflectionTestUtils; +import org.thingsboard.server.gen.edge.v1.ConnectResponseCode; +import org.thingsboard.server.gen.edge.v1.ConnectResponseMsg; +import org.thingsboard.server.gen.edge.v1.EdgeRpcServiceGrpc; +import org.thingsboard.server.gen.edge.v1.RequestMsg; +import org.thingsboard.server.gen.edge.v1.RequestMsgType; +import org.thingsboard.server.gen.edge.v1.ResponseMsg; +import org.thingsboard.server.gen.edge.v1.UplinkMsg; + +import java.lang.reflect.Method; +import java.util.concurrent.TimeUnit; +import java.util.function.BooleanSupplier; + +import static org.junit.jupiter.api.Assertions.fail; + +class EdgeGrpcClientLeakTest { + + // The window in which the default shared event loop group would be destroyed after the channel + // terminates (SharedResourceHolder delays destruction by 1 second). If EdgeGrpcClient ever goes + // back to the shared group, writes after this window hit a terminated executor and every buffer + // committed to grpc-netty's WriteQueue is pinned forever (4112 bytes per message, silent after + // the first RejectedExecutionException). + private static final long SHARED_GROUP_DEATH_WINDOW_MS = 3000; + private static final int MSG_COUNT = 50; + private static final long AWAIT_TIMEOUT_MS = 15_000; + + private Server server; + private EdgeGrpcClient client; + + @Test + void uplinksSentAfterTransportDeathDoNotPinPooledBuffers() throws Exception { + server = NettyServerBuilder.forPort(0) + .addService(new EdgeRpcServiceGrpc.EdgeRpcServiceImplBase() { + @Override + public StreamObserver handleMsgs(StreamObserver outputStream) { + return new StreamObserver<>() { + @Override + public void onNext(RequestMsg requestMsg) { + if (requestMsg.hasConnectRequestMsg()) { + outputStream.onNext(ResponseMsg.newBuilder() + .setConnectResponseMsg(ConnectResponseMsg.newBuilder() + .setResponseCode(ConnectResponseCode.ACCEPTED) + .build()) + .build()); + } + } + + @Override + public void onError(Throwable t) { + } + + @Override + public void onCompleted() { + } + }; + } + }) + .build() + .start(); + + client = new EdgeGrpcClient(); + ReflectionTestUtils.setField(client, "rpcHost", "localhost"); + ReflectionTestUtils.setField(client, "rpcPort", server.getPort()); + ReflectionTestUtils.setField(client, "timeoutSecs", 1); + ReflectionTestUtils.setField(client, "keepAliveTimeSec", 10); + ReflectionTestUtils.setField(client, "keepAliveTimeoutSec", 5); + ReflectionTestUtils.setField(client, "maxInboundMessageSize", 4194304); + + client.connect("leakTest", "leakTest", msg -> {}, cfg -> {}, msg -> {}, e -> {}); + await("client to connect", () -> client.isConnected()); + + server.shutdownNow(); + server.awaitTermination(10, TimeUnit.SECONDS); + await("client to observe the transport death", () -> !client.isConnected()); + Thread.sleep(SHARED_GROUP_DEATH_WINDOW_MS); + + long baseline = pinnedBytes(); + @SuppressWarnings("unchecked") + StreamObserver inputStream = (StreamObserver) ReflectionTestUtils.getField(client, "inputStream"); + RequestMsg uplink = RequestMsg.newBuilder() + .setMsgType(RequestMsgType.UPLINK_RPC_MESSAGE) + .setUplinkMsg(UplinkMsg.newBuilder().setUplinkMsgId(1).build()) + .build(); + // Bypasses the connected gate on purpose: this models the check-then-act straggler (and the + // pre-gate retry loop) writing to a stream whose transport is already gone. Exceptions are + // swallowed the same way the production retry loop survives them. + for (int i = 0; i < MSG_COUNT; i++) { + try { + inputStream.onNext(uplink); + } catch (RuntimeException ignored) { + } + } + + long deadline = System.currentTimeMillis() + AWAIT_TIMEOUT_MS; + while (pinnedBytes() > baseline) { + if (System.currentTimeMillis() > deadline) { + fail("Pinned pooled memory did not return to baseline: " + (pinnedBytes() - baseline) + + " bytes retained after " + MSG_COUNT + " uplinks to a dead stream"); + } + Thread.sleep(50); + } + } + + @AfterEach + void tearDown() throws Exception { + if (client != null) { + client.disconnect(true); + client.destroy(); + } + if (server != null) { + server.shutdownNow(); + } + } + + private void await(String what, BooleanSupplier condition) throws InterruptedException { + long deadline = System.currentTimeMillis() + AWAIT_TIMEOUT_MS; + while (!condition.getAsBoolean()) { + if (System.currentTimeMillis() > deadline) { + fail("Timed out waiting for " + what); + } + Thread.sleep(50); + } + } + + // grpc-netty builds its own PooledByteBufAllocator instances instead of using + // PooledByteBufAllocator.DEFAULT, and the factory that owns them is package private - so they + // have to be pulled out reflectively. Both variants are checked so the assertion holds no matter + // which one the transport picks on this platform/version. + private static long pinnedBytes() { + try { + Class utils = Class.forName("io.grpc.netty.shaded.io.grpc.netty.Utils"); + Method getByteBufAllocator = utils.getDeclaredMethod("getByteBufAllocator", boolean.class); + getByteBufAllocator.setAccessible(true); + long total = 0; + for (boolean forceHeapBuffer : new boolean[]{false, true}) { + PooledByteBufAllocator pooled = (PooledByteBufAllocator) getByteBufAllocator.invoke(null, forceHeapBuffer); + total += pooled.pinnedDirectMemory() + pooled.pinnedHeapMemory(); + } + return total; + } catch (Exception e) { + throw new IllegalStateException("Failed to read the gRPC allocator metrics", e); + } + } + +}