You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by cd...@apache.org on 2026/07/08 20:31:28 UTC
(plc4x) 01/02: fix: Addressed a Deadlock caused by obtaining connection
This is an automated email from the ASF dual-hosted git repository.
chrisdutz pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit cf28d43fe6f74ece501cff835bdeaec088db3fbf
Author: Christofer Dutz <ch...@c-ware.de>
AuthorDate: Wed Jul 8 22:17:11 2026 +0200
fix: Addressed a Deadlock caused by obtaining connection
closes: #2631
---
.../S7CotpTransportConfigurationTest.java | 32 ++++++++++
.../java/utils/cache/ConnectionContainer.java | 70 +++++++++++++++++++++-
.../java/utils/cache/ConnectionContainerTest.java | 45 ++++++++++++++
3 files changed, 144 insertions(+), 3 deletions(-)
diff --git a/plc4j/drivers/s7/src/test/java/org/apache/plc4x/java/s7/configuration/S7CotpTransportConfigurationTest.java b/plc4j/drivers/s7/src/test/java/org/apache/plc4x/java/s7/configuration/S7CotpTransportConfigurationTest.java
index b8310bc4c3..05125200da 100644
--- a/plc4j/drivers/s7/src/test/java/org/apache/plc4x/java/s7/configuration/S7CotpTransportConfigurationTest.java
+++ b/plc4j/drivers/s7/src/test/java/org/apache/plc4x/java/s7/configuration/S7CotpTransportConfigurationTest.java
@@ -38,6 +38,17 @@ class S7CotpTransportConfigurationTest {
.createConfiguration(S7CotpTransportConfiguration.class, params);
}
+ /**
+ * Mirrors how {@code DriverBase} actually builds the transport configuration: it parses the
+ * connection string with the transport-code ({@code cotp}) prefix, so only {@code cotp.}-prefixed
+ * parameters reach the COTP transport config. COTP addressing (rack/slot/device-group/tsap)
+ * therefore <em>must</em> be given as {@code cotp.remote-slot=...} etc.
+ */
+ private S7CotpTransportConfiguration parsePrefixed(String params) {
+ return new ConfigurationFactory()
+ .createPrefixedConfiguration(S7CotpTransportConfiguration.class, "cotp", params);
+ }
+
@Test
void defaultPort() {
assertEquals(102, new S7CotpTransportConfiguration().getDefaultPort());
@@ -59,6 +70,27 @@ class S7CotpTransportConfigurationTest {
assertEquals(0x0103, cfg.getRemoteTsap());
}
+ @Test
+ void prefixedRemoteSlotIsHonouredThroughDriverPath() {
+ // #2620, real driver path: COTP addressing must be cotp.-prefixed to reach the transport
+ // config (DriverBase parses it with the "cotp" prefix). cotp.remote-slot=3 -> 0x0103.
+ S7CotpTransportConfiguration cfg =
+ parsePrefixed("cotp.remote-rack=0&cotp.remote-slot=3&controller-type=S7_400");
+ assertEquals(3, cfg.getRemoteSlot());
+ assertEquals(0x0103, cfg.getRemoteTsap());
+ }
+
+ @Test
+ void unprefixedAddressingIsIgnoredThroughDriverPath() {
+ // The required-prefix contract: an un-prefixed remote-slot never reaches the COTP transport
+ // config, so the called TSAP falls back to the slot-0 default. This is the misconfiguration
+ // behind the original #2620 report; documented here so it can't silently regress.
+ S7CotpTransportConfiguration cfg =
+ parsePrefixed("remote-rack=0&remote-slot=3&controller-type=S7_400");
+ assertEquals(0, cfg.getRemoteSlot());
+ assertEquals(0x0100, cfg.getRemoteTsap());
+ }
+
@Test
void remoteRackAndSlotAreEncoded() {
// (rack << 4) | slot in the low byte, PG_OR_PC (0x01) in the high byte.
diff --git a/plc4j/tools/connection-cache/src/main/java/org/apache/plc4x/java/utils/cache/ConnectionContainer.java b/plc4j/tools/connection-cache/src/main/java/org/apache/plc4x/java/utils/cache/ConnectionContainer.java
index 02444a876c..aff47dbb49 100644
--- a/plc4j/tools/connection-cache/src/main/java/org/apache/plc4x/java/utils/cache/ConnectionContainer.java
+++ b/plc4j/tools/connection-cache/src/main/java/org/apache/plc4x/java/utils/cache/ConnectionContainer.java
@@ -26,6 +26,7 @@ import org.slf4j.LoggerFactory;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
@@ -49,6 +50,9 @@ import java.util.concurrent.locks.ReentrantLock;
* <p>
* Deadlock Prevention: Always acquires locks in the same order and releases them promptly.
* Timer tasks are canceled before attempting to close connections to prevent deadlocks.
+ * Every blocking I/O performed under the lock is time-bounded — connect (see {@link #connectBounded()},
+ * bounded by {@code maxWaitTimeMs}), ping ({@code pingTimeoutMs}) and close ({@code closeTimeoutMs}) — so
+ * a hung driver can never hold the lock indefinitely and starve other callers of the same connection.
*/
class ConnectionContainer {
@@ -145,7 +149,7 @@ class ConnectionContainer {
// Try to get a new connection if we haven't got one yet.
if(connection == null) {
try {
- connection = connectionFactory.get();
+ connection = connectBounded();
} catch (PlcConnectionException e) {
if (LOGGER.isTraceEnabled()) {
LOGGER.warn("Exception while getting connection for lease", e);
@@ -167,7 +171,7 @@ class ConnectionContainer {
LOGGER.debug("Connection failed validation, closing and getting new connection");
closeBounded(connection);
try {
- connection = connectionFactory.get();
+ connection = connectBounded();
// Restore tracked state (subscriptions, event listeners) on new connection
if (stateTracker.hasStateToRestore()) {
stateTracker.restoreState(connection);
@@ -276,7 +280,7 @@ class ConnectionContainer {
// If the connection was previously invalidated, get a new connection.
if (connection == null) {
try {
- connection = connectionFactory.get();
+ connection = connectBounded();
// Restore tracked state (subscriptions, event listeners) on new connection
if (stateTracker.hasStateToRestore()) {
stateTracker.restoreState(connection);
@@ -560,6 +564,66 @@ class ConnectionContainer {
closeBounded(toClose);
}
+ /**
+ * Establishes a connection without ever blocking the caller — and therefore the container lock —
+ * for longer than {@code maxWaitTimeMs}. The underlying driver {@code connect()} can otherwise hang
+ * indefinitely (e.g. a wedged TCP handshake, or a driver awaiting a {@code CompletableFuture} that
+ * never completes). Because this runs under the container lock, an unbounded connect would hold the
+ * lock forever and freeze every other operation on this connection — including waiters, who block on
+ * {@code lock.lock()} in {@link #lease()} before they ever obtain a future and so never see their own
+ * {@code maxWaitTimeMs} timeout. Bounding the connect turns "stuck forever" into "fails after
+ * {@code maxWaitTimeMs}, dead connect abandoned to a background daemon thread", honouring the wait
+ * timeout the caller was promised. A non-positive {@code maxWaitTimeMs} disables the bound (legacy
+ * synchronous connect).
+ *
+ * @return the freshly established connection (never {@code null})
+ * @throws PlcConnectionException if the connect fails or does not complete within {@code maxWaitTimeMs}
+ */
+ private PlcConnection connectBounded() throws PlcConnectionException {
+ if (maxWaitTimeMs <= 0) {
+ return connectionFactory.get();
+ }
+ CompletableFuture<PlcConnection> result = new CompletableFuture<>();
+ Thread connector = new Thread(() -> {
+ try {
+ PlcConnection established = connectionFactory.get();
+ // If the caller already gave up (timed out below), it will never use this connection —
+ // close it here instead of leaking it. complete() returns false iff we lost that race.
+ if (!result.complete(established)) {
+ closeQuietly(established);
+ }
+ } catch (Throwable t) {
+ result.completeExceptionally(t);
+ }
+ }, "cache-connect-" + connectionString);
+ connector.setDaemon(true);
+ connector.start();
+ try {
+ return result.get(maxWaitTimeMs, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ // Abandon the hanging connect. If the connector produced a connection in the tiny race
+ // window between get() timing out and this line, completeExceptionally() fails — recover
+ // and close that now-orphaned connection so it is not leaked.
+ if (!result.completeExceptionally(e)) {
+ result.thenAccept(this::closeQuietly);
+ }
+ connector.interrupt(); // best-effort; the underlying connect may not honor interruption
+ LOGGER.warn("Connection establishment did not complete within {}ms; abandoning it to the "
+ + "background and failing this attempt: {}", maxWaitTimeMs, connectionString);
+ throw new PlcConnectionException("Connection establishment did not complete within "
+ + maxWaitTimeMs + "ms: " + connectionString, e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new PlcConnectionException("Interrupted while establishing connection: " + connectionString, e);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof PlcConnectionException) {
+ throw (PlcConnectionException) cause;
+ }
+ throw new PlcConnectionException("Error establishing connection: " + connectionString, cause);
+ }
+ }
+
/**
* Closes {@code conn} without ever blocking the caller — and therefore the container lock — for
* longer than {@code closeTimeoutMs}. A {@code close()} on a wedged socket can otherwise hang
diff --git a/plc4j/tools/connection-cache/src/test/java/org/apache/plc4x/java/utils/cache/ConnectionContainerTest.java b/plc4j/tools/connection-cache/src/test/java/org/apache/plc4x/java/utils/cache/ConnectionContainerTest.java
index 6752128531..7ccc6399ca 100644
--- a/plc4j/tools/connection-cache/src/test/java/org/apache/plc4x/java/utils/cache/ConnectionContainerTest.java
+++ b/plc4j/tools/connection-cache/src/test/java/org/apache/plc4x/java/utils/cache/ConnectionContainerTest.java
@@ -731,4 +731,49 @@ class ConnectionContainerTest {
assertTrue(elapsed < 2000,
"container.close() must be bounded by the close timeout, but took " + elapsed + "ms");
}
+
+ /**
+ * Regression test for issue #2631: a hanging driver {@code connect()} must not hold the container
+ * lock indefinitely and deadlock other callers. The connect runs under the lock, so it is bounded
+ * by {@code maxWaitTimeMs}; after that the attempt fails and the lock is released, letting the next
+ * caller proceed instead of blocking forever on {@code lock.lock()} inside {@code lease()}.
+ */
+ @Test
+ void testConnectTimeout_AbandonsHangingConnect() throws Exception {
+ CountDownLatch connectStarted = new CountDownLatch(1);
+ // Simulate a wedged connect that never completes (far longer than the configured max wait).
+ ConnectionContainer container = new ConnectionContainer(
+ "test:tcp://localhost",
+ () -> {
+ connectStarted.countDown();
+ try {
+ Thread.sleep(60_000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return mockConnection;
+ },
+ scheduler,
+ 5000, // maxIdleTimeMs
+ 1000, // maxLeaseTimeMs
+ 300, // maxWaitTimeMs — bounds the connect
+ 5000, // pingTimeoutMs
+ 30000 // idlePingThresholdMs
+ );
+
+ // A first caller triggers the (hanging) connect; lease() must not block past the max wait.
+ long start = System.currentTimeMillis();
+ Future<PlcConnection> future1 = container.lease();
+ long elapsed = System.currentTimeMillis() - start;
+
+ assertTrue(connectStarted.await(1, TimeUnit.SECONDS), "connect should have been invoked");
+ assertTrue(elapsed < 2000,
+ "lease() must be bounded by the max wait time, but took " + elapsed + "ms");
+ // The bounded connect failed, so the lease future completes exceptionally rather than hanging.
+ assertTrue(future1.isDone());
+ assertThrows(java.util.concurrent.ExecutionException.class, future1::get);
+
+ // The container recovered: it is not stuck leased and a subsequent caller can proceed.
+ assertFalse(container.isLeased());
+ }
}