From 1b3c415dd49483c2b60ecee9a6314b2cb9102f25 Mon Sep 17 00:00:00 2001 From: Michele Tramonti Date: Wed, 29 Apr 2026 11:17:45 +0200 Subject: [PATCH] fix(plc4j/drivers/s7): stop S7HMuxImpl from leaking direct buffers in the embedded outbound queue S7HMuxImpl.encode() forwards the outbound message to the active TCP channel and also pushed an outBB.copy() into the MessageToMessageCodec out list. The embedded channel's outbound queue is never drained, so every sent message leaked one direct ByteBuf, eventually exhausting MaxDirectMemorySize on long-running applications (see issue #2248). Replace the second copy with the Unpooled.EMPTY_BUFFER singleton (Netty still requires at least one element in the out list) so no direct memory is allocated for the embedded path. Forward to the active TCP channel is unchanged. Add S7HMuxLeakTest as a regression test: it wires S7HMuxImpl exactly like S7HPlcConnection does (one EmbeddedChannel as the logical pipeline, one EmbeddedChannel as the primary TCP channel) and asserts that 2000 outbound messages are forwarded once to the TCP side and that no non-empty ByteBufs accumulate in the embedded outbound queue. fixes #2248 --- .../s7/readwrite/protocol/S7HMuxImpl.java | 14 ++- .../s7/readwrite/protocol/S7HMuxLeakTest.java | 98 +++++++++++++++++++ 2 files changed, 110 insertions(+), 2 deletions(-) create mode 100644 plc4j/drivers/s7/src/test/java/org/apache/plc4x/java/s7/readwrite/protocol/S7HMuxLeakTest.java diff --git a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7HMuxImpl.java b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7HMuxImpl.java index 6d0c3c0d7cc..a96192743ac 100644 --- a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7HMuxImpl.java +++ b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7HMuxImpl.java @@ -19,6 +19,7 @@ package org.apache.plc4x.java.s7.readwrite.protocol; import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; import io.netty.channel.Channel; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandler.Sharable; @@ -130,9 +131,18 @@ public class S7HMuxImpl extends MessageToMessageCodec implemen @Override protected void encode(ChannelHandlerContext ctx, ByteBuf outBB, List list) { if ((embedCtx == null) && (ctx.channel() instanceof EmbeddedChannel)) embedCtx = ctx; - if ((tcpChannel != null) && (embedCtx == ctx)) { - tcpChannel.writeAndFlush(outBB.copy()); + + if (ctx == embedCtx) { + if (tcpChannel != null) { + tcpChannel.writeAndFlush(outBB.copy()); + } + // Netty requires at least one element in the out list, but nothing consumes + // the EmbeddedChannel's outbound queue. Use the EMPTY_BUFFER singleton instead + // of a copied ByteBuf to avoid leaking direct memory (see issue #2248). + list.add(Unpooled.EMPTY_BUFFER); + return; } + list.add(outBB.copy()); } diff --git a/plc4j/drivers/s7/src/test/java/org/apache/plc4x/java/s7/readwrite/protocol/S7HMuxLeakTest.java b/plc4j/drivers/s7/src/test/java/org/apache/plc4x/java/s7/readwrite/protocol/S7HMuxLeakTest.java new file mode 100644 index 00000000000..2881bdefeb5 --- /dev/null +++ b/plc4j/drivers/s7/src/test/java/org/apache/plc4x/java/s7/readwrite/protocol/S7HMuxLeakTest.java @@ -0,0 +1,98 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * https://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.apache.plc4x.java.s7.readwrite.protocol; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.PooledByteBufAllocator; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.util.ResourceLeakDetector; +import org.apache.plc4x.java.s7.readwrite.configuration.S7Configuration; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * Regression test for issue #2248. + * + *

S7HMuxImpl is installed in the EmbeddedChannel pipeline of S7HPlcConnection. Its encode() + * forwards each outbound message to the active TCP channel and propagates a sentinel downstream. + * Before the fix, encode() pushed a freshly-copied direct ByteBuf into the EmbeddedChannel's + * outbound queue: nothing in plc4j ever drained that queue, so every sent message leaked one + * direct buffer until DirectMemory was exhausted (OutOfMemoryError reported on long-running + * applications). + */ +public class S7HMuxLeakTest { + + private static final int ITERATIONS = 2000; + private static final int PAYLOAD_BYTES = 256; + + @Test + void embeddedOutboundQueueShouldNotAccumulateRealByteBufs() throws Exception { + ResourceLeakDetector.Level previous = ResourceLeakDetector.getLevel(); + ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.PARANOID); + try { + S7HMuxImpl mux = new S7HMuxImpl(); + EmbeddedChannel embedded = new EmbeddedChannel(mux); + EmbeddedChannel tcp = new EmbeddedChannel(); + + mux.setEmbededhannel(embedded, new S7Configuration()); + mux.setPrimaryChannel(tcp); + + int forwardedToTcp = 0; + for (int i = 0; i < ITERATIONS; i++) { + ByteBuf out = PooledByteBufAllocator.DEFAULT.directBuffer(PAYLOAD_BYTES); + out.writeZero(PAYLOAD_BYTES); + embedded.writeOutbound(out); + + Object msg; + while ((msg = tcp.readOutbound()) != null) { + if (msg instanceof ByteBuf) { + ((ByteBuf) msg).release(); + forwardedToTcp++; + } + } + } + + assertEquals(ITERATIONS, forwardedToTcp, + "every message written to the embedded channel must be forwarded once to the TCP channel"); + + int leakedBuffers = 0; + long leakedBytes = 0; + Object pending; + while ((pending = embedded.readOutbound()) != null) { + if (pending instanceof ByteBuf) { + ByteBuf bb = (ByteBuf) pending; + if (bb.capacity() > 0) { + leakedBuffers++; + leakedBytes += bb.capacity(); + } + bb.release(); + } + } + assertEquals(0, leakedBuffers, + "EmbeddedChannel outbound queue must not accumulate non-empty ByteBufs (leaked " + + leakedBytes + " bytes across " + leakedBuffers + " buffers)"); + + embedded.finishAndReleaseAll(); + tcp.finishAndReleaseAll(); + } finally { + ResourceLeakDetector.setLevel(previous); + } + } +}