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 <[email protected]> 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()); + } }
