This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
The following commit(s) were added to refs/heads/develop by this push:
new 420e7e6a39 fix(plc4j/modbus): validate response function codes and
make request-timeout cover queueing
420e7e6a39 is described below
commit 420e7e6a39eada49b96825ab243686fa6ba8ec43
Author: Sebastian Rühl <[email protected]>
AuthorDate: Tue Jul 7 13:51:54 2026 +0200
fix(plc4j/modbus): validate response function codes and make
request-timeout cover queueing
Serial connections (RTU/ASCII) correlated responses by unit id only, so a
late response from a timed-out request could complete the next request on
the same unit. Responses are now also validated against the pending
request's function code (exception frames answer any request; the generated
error PDU carries no fc). Mismatches are logged and discarded while the
pending request stays armed. Same-fc late responses remain
indistinguishable — inherent to RTU, documented.
The request timeout is now armed at submission instead of dispatch, making
it a total wall-clock budget; requests that expire while queued fail fast
with a TimeoutException and never reach the wire (small dispatch margin,
documented in the config description).
---
RELEASE_NOTES | 4 +
.../java/modbus/ascii/ModbusAsciiConnection.java | 125 ++++++++++++++++-----
.../ascii/config/ModbusAsciiConfiguration.java | 2 +-
.../plc4x/java/modbus/rtu/ModbusRtuConnection.java | 125 ++++++++++++++++-----
.../modbus/rtu/config/ModbusRtuConfiguration.java | 2 +-
.../modbus/ascii/ModbusAsciiConnectionTest.java | 31 +++--
.../rtu/ModbusRtuConnectionRequestChainTest.java | 74 +++++++++++-
.../java/modbus/rtu/ModbusRtuConnectionTest.java | 31 +++--
.../transport/serial/SerialTransportInstance.java | 3 +-
.../transport/serial/SharedModeInstanceTest.java | 29 +++++
10 files changed, 344 insertions(+), 82 deletions(-)
diff --git a/RELEASE_NOTES b/RELEASE_NOTES
index a2424b1b12..5cf58c4c37 100644
--- a/RELEASE_NOTES
+++ b/RELEASE_NOTES
@@ -38,6 +38,10 @@ New Features
unknown Modbus exception codes map to REMOTE_ERROR instead of failing;
shared serial ports dispatch each connection's callbacks on their own
thread, so one blocked callback no longer stalls the whole port.
+ Responses are additionally validated against the pending request's
+ function code (late responses from timed-out requests are discarded
+ instead of completing the wrong caller), and the request timeout now
+ covers the full time from submission, including queueing.
Incompatible changes
--------------------
diff --git
a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/ascii/ModbusAsciiConnection.java
b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/ascii/ModbusAsciiConnection.java
index cc4d702310..c97e2f641a 100644
---
a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/ascii/ModbusAsciiConnection.java
+++
b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/ascii/ModbusAsciiConnection.java
@@ -64,7 +64,7 @@ public class ModbusAsciiConnection extends
ConnectionBase<ModbusAsciiConfigurati
private static final Logger LOGGER =
LoggerFactory.getLogger(ModbusAsciiConnection.class);
private ModbusAsciiMessageCodec messageCodec;
- private final Map<Short, CompletableFuture<ModbusAsciiADU>>
pendingRequests = new ConcurrentHashMap<>();
+ private final Map<Short, PendingRequest> pendingRequests = new
ConcurrentHashMap<>();
// Serializes request/response transactions. Modbus over serial is a
// single-outstanding-transaction protocol; without this, concurrent
@@ -107,8 +107,8 @@ public class ModbusAsciiConnection extends
ConnectionBase<ModbusAsciiConfigurati
if (messageCodec != null) {
messageCodec.close();
}
- pendingRequests.values().forEach(future ->
- future.completeExceptionally(new PlcRuntimeException("Connection
closed")));
+ pendingRequests.values().forEach(pending ->
+ pending.future.completeExceptionally(new
PlcRuntimeException("Connection closed")));
pendingRequests.clear();
super.close();
LOGGER.info("Modbus ASCII connection closed");
@@ -126,7 +126,7 @@ public class ModbusAsciiConnection extends
ConnectionBase<ModbusAsciiConfigurati
int pendingCount = pendingRequests.size();
if (pendingCount > 0) {
LOGGER.warn("Failing {} pending requests due to transport
disconnect", pendingCount);
- pendingRequests.values().forEach(future ->
future.completeExceptionally(exception));
+ pendingRequests.values().forEach(pending ->
pending.future.completeExceptionally(exception));
pendingRequests.clear();
}
}
@@ -147,9 +147,31 @@ public class ModbusAsciiConnection extends
ConnectionBase<ModbusAsciiConfigurati
}
/**
- * Correlation is by unit id only; requests are serialized per connection
- * via {@link #sendRequest}, and per shared port each connection must own
- * distinct unit ids.
+ * A dispatched request awaiting its response: the future plus the
+ * request's function code for response validation.
+ */
+ private static final class PendingRequest {
+ final byte functionFlag;
+ final CompletableFuture<ModbusAsciiADU> future;
+ PendingRequest(byte functionFlag, CompletableFuture<ModbusAsciiADU>
future) {
+ this.functionFlag = functionFlag;
+ this.future = future;
+ }
+ }
+
+ /**
+ * Correlates a received frame to the pending request for its unit id.
+ * Validation: an exception frame (errorFlag) answers any pending
+ * request; a normal frame must carry the SAME function code as the
+ * pending request — mismatches (e.g. a late response arriving after
+ * its request timed out and a different operation took its place) are
+ * discarded without touching the pending entry.
+ * <p>
+ * Residual limitation: a late response with the SAME function code as
+ * the new request is physically indistinguishable — Modbus RTU/ASCII
+ * carry no transaction ids. Requests are serialized per connection and
+ * shared ports require distinct unit ids per connection, which bounds
+ * the exposure to same-fc retry patterns.
*/
private void handleIncomingMessage(ModbusAsciiADU modbusMessage) {
short address = modbusMessage.getAddress();
@@ -157,17 +179,61 @@ public class ModbusAsciiConnection extends
ConnectionBase<ModbusAsciiConfigurati
auditLog.write(AuditLogEventType.INCOMING_MESSAGE,
"Received Modbus ASCII response, address=" + address);
}
- CompletableFuture<ModbusAsciiADU> future =
pendingRequests.remove(address);
- if (future != null) {
- future.complete(modbusMessage);
- } else {
+ PendingRequest pending = pendingRequests.get(address);
+ if (pending == null) {
LOGGER.warn("Received response for unknown address: {}", address);
+ return;
}
+ boolean isException = modbusMessage.getPdu().getErrorFlag();
+ if (!isException && modbusMessage.getPdu().getFunctionFlag() !=
pending.functionFlag) {
+ LOGGER.warn("Discarding response for address {} with function code
0x{} while waiting for 0x{} (late response from a timed-out request?)",
+ address,
+ Integer.toHexString(modbusMessage.getPdu().getFunctionFlag() &
0xFF),
+ Integer.toHexString(pending.functionFlag & 0xFF));
+ return;
+ }
+ pendingRequests.remove(address, pending);
+ pending.future.complete(modbusMessage);
}
// Package-private for tests.
CompletableFuture<ModbusAsciiADU> sendRequest(ModbusAsciiADU request,
short address) {
CompletableFuture<ModbusAsciiADU> responseFuture = new
CompletableFuture<>();
+
+ // Total budget from submission: queueing + dispatch + response.
+ long timeoutMs = getConfiguration().getRequestTimeout();
+ // Explicit deadline, checked again in dispatchRequest(): the JVM-wide
+ // CompletableFuture delay scheduler is a single thread and only gets
+ // back around to an already-overdue timeout AFTER it finishes
+ // processing the previous request's own completion (cleanup +
+ // chain-continuation + submitting the next dispatch) — that hand-off
+ // to the async pool routinely wins the race against the scheduler
+ // catching up, so responseFuture.isDone() alone can still be false
+ // for a request whose enqueue-time budget has, in wall-clock terms,
+ // already run out.
+ long nowNanos = System.nanoTime();
+ long budgetNanos = TimeUnit.MILLISECONDS.toNanos(timeoutMs);
+ // A request with only a sliver of its budget left is not worth
+ // dispatching either — it has no realistic chance of a round trip
+ // completing before its own timeout fires anyway. Backing the
+ // dispatch deadline off by this margin also absorbs scheduling
+ // jitter between a predecessor's timeout firing (on the JVM-wide
+ // single-threaded delay scheduler, see above) and this request's
+ // own deadline — queued only a hair later, on the same clock —
+ // which would otherwise let a request that is, for all practical
+ // purposes, dead slip through a raw "has my deadline passed" check.
+ long minViableRemainingNanos =
Math.min(TimeUnit.MILLISECONDS.toNanos(50), budgetNanos / 4);
+ long dispatchDeadlineNanos = nowNanos + budgetNanos -
minViableRemainingNanos;
+ responseFuture.orTimeout(timeoutMs, TimeUnit.MILLISECONDS)
+ .whenComplete((result, error) -> {
+ if (error instanceof TimeoutException) {
+ // Two-arg remove semantics preserved through the holder:
+ // only clear the entry if it is still OUR request.
+ pendingRequests.computeIfPresent(address,
+ (key, pending) -> pending.future == responseFuture ?
null : pending);
+ }
+ });
+
CompletableFuture<?> previous;
synchronized (requestChainLock) {
previous = requestTail;
@@ -177,18 +243,30 @@ public class ModbusAsciiConnection extends
ConnectionBase<ModbusAsciiConfigurati
}
if (previous.isDone()) {
// Hot path: nothing queued — dispatch on the caller thread as
before.
- dispatchRequest(request, address, responseFuture);
+ dispatchRequest(request, address, responseFuture,
dispatchDeadlineNanos);
} else {
// Queued: hop to the async pool so long bursts release the
// completing thread's stack (a synchronous chain nests one
// frame per queued request and can silently overflow).
- previous.whenCompleteAsync((ignored, ignoredError) ->
dispatchRequest(request, address, responseFuture));
+ previous.whenCompleteAsync((ignored, ignoredError) ->
dispatchRequest(request, address, responseFuture, dispatchDeadlineNanos));
}
return responseFuture;
}
- private void dispatchRequest(ModbusAsciiADU request, short address,
CompletableFuture<ModbusAsciiADU> responseFuture) {
- pendingRequests.put(address, responseFuture);
+ private void dispatchRequest(ModbusAsciiADU request, short address,
CompletableFuture<ModbusAsciiADU> responseFuture, long dispatchDeadlineNanos) {
+ if (responseFuture.isDone()) {
+ // Already completed (timeout or failure) while queued.
+ return;
+ }
+ if (System.nanoTime() >= dispatchDeadlineNanos) {
+ // Fail fast: the remaining budget is below the dispatch margin,
+ // so the request is semantically dead — don't make the caller
+ // wait for the orTimeout backstop.
+ responseFuture.completeExceptionally(new TimeoutException(
+ "Request timed out while queued (remaining budget below
dispatch margin)"));
+ return;
+ }
+ pendingRequests.put(address, new
PendingRequest(request.getPdu().getFunctionFlag(), responseFuture));
try {
if (auditLog.isEnabled()) {
auditLog.write(AuditLogEventType.OUTGOING_MESSAGE,
@@ -196,29 +274,16 @@ public class ModbusAsciiConnection extends
ConnectionBase<ModbusAsciiConfigurati
}
messageCodec.send(request);
} catch (MessageCodecException e) {
- pendingRequests.remove(address);
+ pendingRequests.computeIfPresent(address, (key, pending) ->
pending.future == responseFuture ? null : pending);
responseFuture.completeExceptionally(new
PlcRuntimeException("Failed to send request", e));
- return;
} catch (RuntimeException e) {
// The responseFuture MUST complete no matter what: the request
// chain's tail hangs off it, so an unchecked throw here (e.g.
// from a custom AuditLog) would otherwise silently wedge every
// subsequent request on this connection.
- pendingRequests.remove(address, responseFuture);
+ pendingRequests.computeIfPresent(address, (key, pending) ->
pending.future == responseFuture ? null : pending);
responseFuture.completeExceptionally(e);
- return;
}
-
- long timeoutMs = getConfiguration().getRequestTimeout();
- responseFuture.orTimeout(timeoutMs, TimeUnit.MILLISECONDS)
- .whenComplete((result, error) -> {
- if (error instanceof TimeoutException) {
- // Two-arg remove: only clears our own entry, immune to
- // dependent-ordering (a late cleanup can never remove a
- // successor request's entry).
- pendingRequests.remove(address, responseFuture);
- }
- });
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
diff --git
a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/ascii/config/ModbusAsciiConfiguration.java
b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/ascii/config/ModbusAsciiConfiguration.java
index 9bba5b0aa0..4cc21b729c 100644
---
a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/ascii/config/ModbusAsciiConfiguration.java
+++
b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/ascii/config/ModbusAsciiConfiguration.java
@@ -30,7 +30,7 @@ public class ModbusAsciiConfiguration implements
Configuration {
@ConfigurationParameter("request-timeout")
@IntDefaultValue(5_000)
- @Description("Default timeout for all types of requests.")
+ @Description("Default timeout for all types of requests. The timeout
covers the full time from submission including queueing; queued requests whose
remaining budget falls below a small dispatch margin (at most a quarter of the
timeout, capped at 50 ms) fail fast instead of being sent.")
private int requestTimeout;
@ConfigurationParameter("default-unit-identifier")
diff --git
a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnection.java
b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnection.java
index f3c7165fa5..4a9e7b789a 100644
---
a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnection.java
+++
b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnection.java
@@ -64,7 +64,7 @@ public class ModbusRtuConnection extends
ConnectionBase<ModbusRtuConfiguration>
private static final Logger LOGGER =
LoggerFactory.getLogger(ModbusRtuConnection.class);
private ModbusRtuMessageCodec messageCodec;
- private final Map<Short, CompletableFuture<ModbusRtuADU>> pendingRequests
= new ConcurrentHashMap<>();
+ private final Map<Short, PendingRequest> pendingRequests = new
ConcurrentHashMap<>();
// Serializes request/response transactions. Modbus over serial is a
// single-outstanding-transaction protocol; without this, concurrent
@@ -107,8 +107,8 @@ public class ModbusRtuConnection extends
ConnectionBase<ModbusRtuConfiguration>
if (messageCodec != null) {
messageCodec.close();
}
- pendingRequests.values().forEach(future ->
- future.completeExceptionally(new PlcRuntimeException("Connection
closed")));
+ pendingRequests.values().forEach(pending ->
+ pending.future.completeExceptionally(new
PlcRuntimeException("Connection closed")));
pendingRequests.clear();
super.close();
LOGGER.info("Modbus RTU connection closed");
@@ -126,7 +126,7 @@ public class ModbusRtuConnection extends
ConnectionBase<ModbusRtuConfiguration>
int pendingCount = pendingRequests.size();
if (pendingCount > 0) {
LOGGER.warn("Failing {} pending requests due to transport
disconnect", pendingCount);
- pendingRequests.values().forEach(future ->
future.completeExceptionally(exception));
+ pendingRequests.values().forEach(pending ->
pending.future.completeExceptionally(exception));
pendingRequests.clear();
}
}
@@ -147,9 +147,31 @@ public class ModbusRtuConnection extends
ConnectionBase<ModbusRtuConfiguration>
}
/**
- * Correlation is by unit id only; requests are serialized per connection
- * via {@link #sendRequest}, and per shared port each connection must own
- * distinct unit ids.
+ * A dispatched request awaiting its response: the future plus the
+ * request's function code for response validation.
+ */
+ private static final class PendingRequest {
+ final byte functionFlag;
+ final CompletableFuture<ModbusRtuADU> future;
+ PendingRequest(byte functionFlag, CompletableFuture<ModbusRtuADU>
future) {
+ this.functionFlag = functionFlag;
+ this.future = future;
+ }
+ }
+
+ /**
+ * Correlates a received frame to the pending request for its unit id.
+ * Validation: an exception frame (errorFlag) answers any pending
+ * request; a normal frame must carry the SAME function code as the
+ * pending request — mismatches (e.g. a late response arriving after
+ * its request timed out and a different operation took its place) are
+ * discarded without touching the pending entry.
+ * <p>
+ * Residual limitation: a late response with the SAME function code as
+ * the new request is physically indistinguishable — Modbus RTU/ASCII
+ * carry no transaction ids. Requests are serialized per connection and
+ * shared ports require distinct unit ids per connection, which bounds
+ * the exposure to same-fc retry patterns.
*/
private void handleIncomingMessage(ModbusRtuADU modbusMessage) {
short address = modbusMessage.getAddress();
@@ -157,17 +179,61 @@ public class ModbusRtuConnection extends
ConnectionBase<ModbusRtuConfiguration>
auditLog.write(AuditLogEventType.INCOMING_MESSAGE,
"Received Modbus RTU response, address=" + address);
}
- CompletableFuture<ModbusRtuADU> future =
pendingRequests.remove(address);
- if (future != null) {
- future.complete(modbusMessage);
- } else {
+ PendingRequest pending = pendingRequests.get(address);
+ if (pending == null) {
LOGGER.warn("Received response for unknown address: {}", address);
+ return;
}
+ boolean isException = modbusMessage.getPdu().getErrorFlag();
+ if (!isException && modbusMessage.getPdu().getFunctionFlag() !=
pending.functionFlag) {
+ LOGGER.warn("Discarding response for address {} with function code
0x{} while waiting for 0x{} (late response from a timed-out request?)",
+ address,
+ Integer.toHexString(modbusMessage.getPdu().getFunctionFlag() &
0xFF),
+ Integer.toHexString(pending.functionFlag & 0xFF));
+ return;
+ }
+ pendingRequests.remove(address, pending);
+ pending.future.complete(modbusMessage);
}
// Package-private for tests.
CompletableFuture<ModbusRtuADU> sendRequest(ModbusRtuADU request, short
address) {
CompletableFuture<ModbusRtuADU> responseFuture = new
CompletableFuture<>();
+
+ // Total budget from submission: queueing + dispatch + response.
+ long timeoutMs = getConfiguration().getRequestTimeout();
+ // Explicit deadline, checked again in dispatchRequest(): the JVM-wide
+ // CompletableFuture delay scheduler is a single thread and only gets
+ // back around to an already-overdue timeout AFTER it finishes
+ // processing the previous request's own completion (cleanup +
+ // chain-continuation + submitting the next dispatch) — that hand-off
+ // to the async pool routinely wins the race against the scheduler
+ // catching up, so responseFuture.isDone() alone can still be false
+ // for a request whose enqueue-time budget has, in wall-clock terms,
+ // already run out.
+ long nowNanos = System.nanoTime();
+ long budgetNanos = TimeUnit.MILLISECONDS.toNanos(timeoutMs);
+ // A request with only a sliver of its budget left is not worth
+ // dispatching either — it has no realistic chance of a round trip
+ // completing before its own timeout fires anyway. Backing the
+ // dispatch deadline off by this margin also absorbs scheduling
+ // jitter between a predecessor's timeout firing (on the JVM-wide
+ // single-threaded delay scheduler, see above) and this request's
+ // own deadline — queued only a hair later, on the same clock —
+ // which would otherwise let a request that is, for all practical
+ // purposes, dead slip through a raw "has my deadline passed" check.
+ long minViableRemainingNanos =
Math.min(TimeUnit.MILLISECONDS.toNanos(50), budgetNanos / 4);
+ long dispatchDeadlineNanos = nowNanos + budgetNanos -
minViableRemainingNanos;
+ responseFuture.orTimeout(timeoutMs, TimeUnit.MILLISECONDS)
+ .whenComplete((result, error) -> {
+ if (error instanceof TimeoutException) {
+ // Two-arg remove semantics preserved through the holder:
+ // only clear the entry if it is still OUR request.
+ pendingRequests.computeIfPresent(address,
+ (key, pending) -> pending.future == responseFuture ?
null : pending);
+ }
+ });
+
CompletableFuture<?> previous;
synchronized (requestChainLock) {
previous = requestTail;
@@ -177,18 +243,30 @@ public class ModbusRtuConnection extends
ConnectionBase<ModbusRtuConfiguration>
}
if (previous.isDone()) {
// Hot path: nothing queued — dispatch on the caller thread as
before.
- dispatchRequest(request, address, responseFuture);
+ dispatchRequest(request, address, responseFuture,
dispatchDeadlineNanos);
} else {
// Queued: hop to the async pool so long bursts release the
// completing thread's stack (a synchronous chain nests one
// frame per queued request and can silently overflow).
- previous.whenCompleteAsync((ignored, ignoredError) ->
dispatchRequest(request, address, responseFuture));
+ previous.whenCompleteAsync((ignored, ignoredError) ->
dispatchRequest(request, address, responseFuture, dispatchDeadlineNanos));
}
return responseFuture;
}
- private void dispatchRequest(ModbusRtuADU request, short address,
CompletableFuture<ModbusRtuADU> responseFuture) {
- pendingRequests.put(address, responseFuture);
+ private void dispatchRequest(ModbusRtuADU request, short address,
CompletableFuture<ModbusRtuADU> responseFuture, long dispatchDeadlineNanos) {
+ if (responseFuture.isDone()) {
+ // Already completed (timeout or failure) while queued.
+ return;
+ }
+ if (System.nanoTime() >= dispatchDeadlineNanos) {
+ // Fail fast: the remaining budget is below the dispatch margin,
+ // so the request is semantically dead — don't make the caller
+ // wait for the orTimeout backstop.
+ responseFuture.completeExceptionally(new TimeoutException(
+ "Request timed out while queued (remaining budget below
dispatch margin)"));
+ return;
+ }
+ pendingRequests.put(address, new
PendingRequest(request.getPdu().getFunctionFlag(), responseFuture));
try {
if (auditLog.isEnabled()) {
auditLog.write(AuditLogEventType.OUTGOING_MESSAGE,
@@ -196,29 +274,16 @@ public class ModbusRtuConnection extends
ConnectionBase<ModbusRtuConfiguration>
}
messageCodec.send(request);
} catch (MessageCodecException e) {
- pendingRequests.remove(address);
+ pendingRequests.computeIfPresent(address, (key, pending) ->
pending.future == responseFuture ? null : pending);
responseFuture.completeExceptionally(new
PlcRuntimeException("Failed to send request", e));
- return;
} catch (RuntimeException e) {
// The responseFuture MUST complete no matter what: the request
// chain's tail hangs off it, so an unchecked throw here (e.g.
// from a custom AuditLog) would otherwise silently wedge every
// subsequent request on this connection.
- pendingRequests.remove(address, responseFuture);
+ pendingRequests.computeIfPresent(address, (key, pending) ->
pending.future == responseFuture ? null : pending);
responseFuture.completeExceptionally(e);
- return;
}
-
- long timeoutMs = getConfiguration().getRequestTimeout();
- responseFuture.orTimeout(timeoutMs, TimeUnit.MILLISECONDS)
- .whenComplete((result, error) -> {
- if (error instanceof TimeoutException) {
- // Two-arg remove: only clears our own entry, immune to
- // dependent-ordering (a late cleanup can never remove a
- // successor request's entry).
- pendingRequests.remove(address, responseFuture);
- }
- });
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
diff --git
a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/config/ModbusRtuConfiguration.java
b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/config/ModbusRtuConfiguration.java
index 1e6eca0eae..b5165dbcab 100644
---
a/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/config/ModbusRtuConfiguration.java
+++
b/plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/rtu/config/ModbusRtuConfiguration.java
@@ -30,7 +30,7 @@ public class ModbusRtuConfiguration implements Configuration {
@ConfigurationParameter("request-timeout")
@IntDefaultValue(5_000)
- @Description("Default timeout for all types of requests.")
+ @Description("Default timeout for all types of requests. The timeout
covers the full time from submission including queueing; queued requests whose
remaining budget falls below a small dispatch margin (at most a quarter of the
timeout, capped at 50 ms) fail fast instead of being sent.")
private int requestTimeout;
@ConfigurationParameter("default-unit-identifier")
diff --git
a/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/ascii/ModbusAsciiConnectionTest.java
b/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/ascii/ModbusAsciiConnectionTest.java
index 1ff97851e8..7dcccde7a4 100644
---
a/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/ascii/ModbusAsciiConnectionTest.java
+++
b/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/ascii/ModbusAsciiConnectionTest.java
@@ -34,6 +34,7 @@ import org.apache.plc4x.java.utils.auditlog.api.AuditLog;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.lang.reflect.Constructor;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.Arrays;
@@ -199,6 +200,21 @@ class ModbusAsciiConnectionTest {
assertEquals(ModbusByteOrder.LITTLE_ENDIAN, method.invoke(connection,
tag));
}
+ /**
+ * pendingRequests holds {@code PendingRequest} (function-code-tagged)
+ * holders rather than raw futures; this reflects the holder in the
+ * same way production code does (byte functionFlag, CompletableFuture
future).
+ */
+ private static Object newPendingRequest(byte functionFlag,
CompletableFuture<ModbusAsciiADU> future) throws Exception {
+ Class<?> pendingRequestClass =
Arrays.stream(ModbusAsciiConnection.class.getDeclaredClasses())
+ .filter(c -> c.getSimpleName().equals("PendingRequest"))
+ .findFirst()
+ .orElseThrow(() -> new IllegalStateException("PendingRequest
nested class not found"));
+ Constructor<?> constructor =
pendingRequestClass.getDeclaredConstructor(byte.class, CompletableFuture.class);
+ constructor.setAccessible(true);
+ return constructor.newInstance(functionFlag, future);
+ }
+
@Test
void testHandleIncomingMessage() throws Exception {
Method method =
ModbusAsciiConnection.class.getDeclaredMethod("handleIncomingMessage",
ModbusAsciiADU.class);
@@ -207,11 +223,10 @@ class ModbusAsciiConnectionTest {
Field pendingField =
ModbusAsciiConnection.class.getDeclaredField("pendingRequests");
pendingField.setAccessible(true);
@SuppressWarnings("unchecked")
- Map<Short, CompletableFuture<ModbusAsciiADU>> pendingRequests =
- (Map<Short, CompletableFuture<ModbusAsciiADU>>)
pendingField.get(connection);
+ Map<Short, Object> pendingRequests = (Map<Short, Object>)
pendingField.get(connection);
CompletableFuture<ModbusAsciiADU> future = new CompletableFuture<>();
- pendingRequests.put((short) 1, future);
+ pendingRequests.put((short) 1, newPendingRequest((byte) 0x03, future));
ModbusAsciiADU responseAdu = new ModbusAsciiADU((short) 1,
new ModbusPDUReadHoldingRegistersResponse(new byte[]{0x00, 0x2A}));
@@ -235,11 +250,10 @@ class ModbusAsciiConnectionTest {
Field pendingField =
ModbusAsciiConnection.class.getDeclaredField("pendingRequests");
pendingField.setAccessible(true);
@SuppressWarnings("unchecked")
- Map<Short, CompletableFuture<ModbusAsciiADU>> pendingRequests =
- (Map<Short, CompletableFuture<ModbusAsciiADU>>)
pendingField.get(connection);
+ Map<Short, Object> pendingRequests = (Map<Short, Object>)
pendingField.get(connection);
CompletableFuture<ModbusAsciiADU> future = new CompletableFuture<>();
- pendingRequests.put((short) 1, future);
+ pendingRequests.put((short) 1, newPendingRequest((byte) 0x03, future));
connection.close();
@@ -252,11 +266,10 @@ class ModbusAsciiConnectionTest {
Field pendingField =
ModbusAsciiConnection.class.getDeclaredField("pendingRequests");
pendingField.setAccessible(true);
@SuppressWarnings("unchecked")
- Map<Short, CompletableFuture<ModbusAsciiADU>> pendingRequests =
- (Map<Short, CompletableFuture<ModbusAsciiADU>>)
pendingField.get(connection);
+ Map<Short, Object> pendingRequests = (Map<Short, Object>)
pendingField.get(connection);
CompletableFuture<ModbusAsciiADU> future = new CompletableFuture<>();
- pendingRequests.put((short) 1, future);
+ pendingRequests.put((short) 1, newPendingRequest((byte) 0x03, future));
Method method =
ModbusAsciiConnection.class.getDeclaredMethod("onTransportDisconnected",
Throwable.class);
method.setAccessible(true);
diff --git
a/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnectionRequestChainTest.java
b/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnectionRequestChainTest.java
index 394d646602..3a93031132 100644
---
a/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnectionRequestChainTest.java
+++
b/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnectionRequestChainTest.java
@@ -18,9 +18,12 @@
*/
package org.apache.plc4x.java.modbus.rtu;
+import org.apache.plc4x.java.modbus.readwrite.ModbusErrorCode;
import org.apache.plc4x.java.modbus.readwrite.ModbusPDU;
+import org.apache.plc4x.java.modbus.readwrite.ModbusPDUError;
import
org.apache.plc4x.java.modbus.readwrite.ModbusPDUReadHoldingRegistersRequest;
import
org.apache.plc4x.java.modbus.readwrite.ModbusPDUReadHoldingRegistersResponse;
+import
org.apache.plc4x.java.modbus.readwrite.ModbusPDUWriteSingleRegisterResponse;
import org.apache.plc4x.java.modbus.readwrite.ModbusRtuADU;
import org.apache.plc4x.java.modbus.rtu.config.ModbusRtuConfiguration;
import org.apache.plc4x.java.modbus.types.ModbusByteOrder;
@@ -44,9 +47,11 @@ import java.util.function.BooleanSupplier;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -147,13 +152,80 @@ class ModbusRtuConnectionRequestChainTest {
}
}
+ @Test
+ void mismatchedFunctionCodeIsDiscardedAndPendingSurvives() throws
Exception {
+ ScriptedAsyncTransport transport = new ScriptedAsyncTransport();
+ ModbusRtuConnection connection = newConnectedConnection(transport);
+
+ CompletableFuture<ModbusRtuADU> pendingRead = connection.sendRequest(
+ readHoldingRegistersRequestAdu(1, 0x0000, 1), (short) 1); // fc
0x03
+
+ // A write-echo frame (fc 0x06) for the same unit must NOT complete it.
+ transport.deliver(wireBytes(new ModbusRtuADU((short) 1,
+ new ModbusPDUWriteSingleRegisterResponse(0x0010, 0x1234))));
+ transport.runDataListener();
+ assertFalse(pendingRead.isDone(), "mismatched fc must be discarded,
pending must survive");
+
+ // The genuine 0x03 response still completes it.
+ transport.deliver(readResponseFrame(1, new byte[]{0x11, 0x22}));
+ transport.runDataListener();
+ assertArrayEquals(new byte[]{0x11, 0x22},
+ extractRegisterBytes(pendingRead.get(2, TimeUnit.SECONDS)));
+ }
+
+ @Test
+ void exceptionResponseCompletesAnyPendingRequest() throws Exception {
+ ScriptedAsyncTransport transport = new ScriptedAsyncTransport();
+ ModbusRtuConnection connection = newConnectedConnection(transport);
+
+ CompletableFuture<ModbusRtuADU> pendingRead = connection.sendRequest(
+ readHoldingRegistersRequestAdu(1, 0x0000, 1), (short) 1);
+
+ transport.deliver(wireBytes(new ModbusRtuADU((short) 1,
+ new ModbusPDUError(ModbusErrorCode.ILLEGAL_DATA_ADDRESS))));
+ transport.runDataListener();
+
+ ModbusRtuADU response = pendingRead.get(2, TimeUnit.SECONDS);
+ assertTrue(response.getPdu().getErrorFlag(), "exception frames answer
any pending request");
+ }
+
+ @Test
+ void requestTimedOutWhileQueuedNeverReachesTheWire() throws Exception {
+ ScriptedAsyncTransport transport = new ScriptedAsyncTransport();
+ // Short request timeout so the queued request's total budget expires
+ // while the head transaction is still outstanding.
+ ModbusRtuConnection connection = newConnectedConnection(transport,
200);
+
+ // Head request dispatches and stays pending (never answered).
+ CompletableFuture<ModbusRtuADU> head = connection.sendRequest(
+ readHoldingRegistersRequestAdu(1, 0x0000, 1), (short) 1);
+ awaitTrue(() -> transport.writeCount() == 1, 2, TimeUnit.SECONDS);
+
+ // Second request queues behind it and must time out WHILE QUEUED.
+ CompletableFuture<ModbusRtuADU> queued = connection.sendRequest(
+ readHoldingRegistersRequestAdu(1, 0x0010, 1), (short) 1);
+ assertThrows(ExecutionException.class, () -> queued.get(2,
TimeUnit.SECONDS),
+ "queued request must time out from its enqueue-time clock");
+
+ // Head times out too (same 200ms clock); the chain then dispatches the
+ // dead queued request — which must be SKIPPED: no second wire write.
+ assertThrows(ExecutionException.class, () -> head.get(2,
TimeUnit.SECONDS));
+ Thread.sleep(200); // give the chain's async dispatch a beat
+ assertEquals(1, transport.writeCount(),
+ "a request that died in the queue must never reach the wire");
+ }
+
/**
* Builds a connection wired to the given fake transport, and drives it
* through the same construction/connect path as {@code
ModbusRtuConnectionTest}.
*/
private static ModbusRtuConnection
newConnectedConnection(ScriptedAsyncTransport transport) throws Exception {
+ return newConnectedConnection(transport, 5000);
+ }
+
+ private static ModbusRtuConnection
newConnectedConnection(ScriptedAsyncTransport transport, int requestTimeoutMs)
throws Exception {
ModbusRtuConfiguration config = new ModbusRtuConfiguration();
- config.setRequestTimeout(5000);
+ config.setRequestTimeout(requestTimeoutMs);
config.setDefaultUnitIdentifier(1);
config.setPingAddress("4x00001:BOOL");
config.setDefaultPayloadByteOrder(ModbusByteOrder.BIG_ENDIAN);
diff --git
a/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnectionTest.java
b/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnectionTest.java
index ebcc39f7b1..2035fd269a 100644
---
a/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnectionTest.java
+++
b/plc4j/drivers/modbus/src/test/java/org/apache/plc4x/java/modbus/rtu/ModbusRtuConnectionTest.java
@@ -34,6 +34,7 @@ import org.apache.plc4x.java.utils.auditlog.api.AuditLog;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.lang.reflect.Constructor;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.Arrays;
@@ -211,6 +212,21 @@ class ModbusRtuConnectionTest {
assertEquals(ModbusByteOrder.LITTLE_ENDIAN, method.invoke(connection,
tag));
}
+ /**
+ * pendingRequests holds {@code PendingRequest} (function-code-tagged)
+ * holders rather than raw futures; this reflects the holder in the
+ * same way production code does (byte functionFlag, CompletableFuture
future).
+ */
+ private static Object newPendingRequest(byte functionFlag,
CompletableFuture<ModbusRtuADU> future) throws Exception {
+ Class<?> pendingRequestClass =
Arrays.stream(ModbusRtuConnection.class.getDeclaredClasses())
+ .filter(c -> c.getSimpleName().equals("PendingRequest"))
+ .findFirst()
+ .orElseThrow(() -> new IllegalStateException("PendingRequest
nested class not found"));
+ Constructor<?> constructor =
pendingRequestClass.getDeclaredConstructor(byte.class, CompletableFuture.class);
+ constructor.setAccessible(true);
+ return constructor.newInstance(functionFlag, future);
+ }
+
@Test
void testHandleIncomingMessage() throws Exception {
Method method =
ModbusRtuConnection.class.getDeclaredMethod("handleIncomingMessage",
ModbusRtuADU.class);
@@ -219,11 +235,10 @@ class ModbusRtuConnectionTest {
Field pendingField =
ModbusRtuConnection.class.getDeclaredField("pendingRequests");
pendingField.setAccessible(true);
@SuppressWarnings("unchecked")
- Map<Short, CompletableFuture<ModbusRtuADU>> pendingRequests =
- (Map<Short, CompletableFuture<ModbusRtuADU>>)
pendingField.get(connection);
+ Map<Short, Object> pendingRequests = (Map<Short, Object>)
pendingField.get(connection);
CompletableFuture<ModbusRtuADU> future = new CompletableFuture<>();
- pendingRequests.put((short) 1, future);
+ pendingRequests.put((short) 1, newPendingRequest((byte) 0x03, future));
ModbusRtuADU responseAdu = new ModbusRtuADU((short) 1,
new ModbusPDUReadHoldingRegistersResponse(new byte[]{0x00, 0x2A}));
@@ -248,11 +263,10 @@ class ModbusRtuConnectionTest {
Field pendingField =
ModbusRtuConnection.class.getDeclaredField("pendingRequests");
pendingField.setAccessible(true);
@SuppressWarnings("unchecked")
- Map<Short, CompletableFuture<ModbusRtuADU>> pendingRequests =
- (Map<Short, CompletableFuture<ModbusRtuADU>>)
pendingField.get(connection);
+ Map<Short, Object> pendingRequests = (Map<Short, Object>)
pendingField.get(connection);
CompletableFuture<ModbusRtuADU> future = new CompletableFuture<>();
- pendingRequests.put((short) 1, future);
+ pendingRequests.put((short) 1, newPendingRequest((byte) 0x03, future));
connection.close();
@@ -265,11 +279,10 @@ class ModbusRtuConnectionTest {
Field pendingField =
ModbusRtuConnection.class.getDeclaredField("pendingRequests");
pendingField.setAccessible(true);
@SuppressWarnings("unchecked")
- Map<Short, CompletableFuture<ModbusRtuADU>> pendingRequests =
- (Map<Short, CompletableFuture<ModbusRtuADU>>)
pendingField.get(connection);
+ Map<Short, Object> pendingRequests = (Map<Short, Object>)
pendingField.get(connection);
CompletableFuture<ModbusRtuADU> future = new CompletableFuture<>();
- pendingRequests.put((short) 1, future);
+ pendingRequests.put((short) 1, newPendingRequest((byte) 0x03, future));
Method method =
ModbusRtuConnection.class.getDeclaredMethod("onTransportDisconnected",
Throwable.class);
method.setAccessible(true);
diff --git
a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialTransportInstance.java
b/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialTransportInstance.java
index 00e2bf2be8..7f951fced3 100644
---
a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialTransportInstance.java
+++
b/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialTransportInstance.java
@@ -54,6 +54,7 @@ public class SerialTransportInstance extends
BaseTransportInstance<SerialTranspo
// subscriber rings (64 KiB) so burst tolerance is consistent.
private static final int DEFAULT_BUFFER_SIZE = 65536;
private static final byte[] EMPTY_BYTES = new byte[0];
+ private static final java.util.concurrent.atomic.AtomicLong
DISPATCH_THREAD_COUNTER = new java.util.concurrent.atomic.AtomicLong();
private final SharedSerialPortManager sharedSerialPortManager;
private final SerialPort port;
@@ -188,7 +189,7 @@ public class SerialTransportInstance extends
BaseTransportInstance<SerialTranspo
};
tempSharedPort.addSubscriber(this.sharedSubscriber);
this.sharedDispatchExecutor =
Executors.newSingleThreadExecutor(runnable -> {
- Thread thread = new Thread(runnable,
"Serial-Shared-Dispatch-" + port);
+ Thread thread = new Thread(runnable,
"Serial-Shared-Dispatch-" + port + "-" +
DISPATCH_THREAD_COUNTER.incrementAndGet());
thread.setDaemon(true);
return thread;
});
diff --git
a/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/SharedModeInstanceTest.java
b/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/SharedModeInstanceTest.java
index 2d31f8f424..a5723d6981 100644
---
a/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/SharedModeInstanceTest.java
+++
b/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/SharedModeInstanceTest.java
@@ -185,6 +185,35 @@ class SharedModeInstanceTest {
verify(mockPort).closePort();
}
+ @Test
+ void eachSharedInstanceGetsAUniquelyNamedDispatchThread() throws Exception
{
+ MockPortFactory factory = new MockPortFactory();
+ SharedSerialPortManager manager = new SharedSerialPortManager(factory);
+
+ SerialTransportInstance first = new SerialTransportInstance(
+ manager, "COMSHARED", config(), AuditLog.builder().build());
+ SerialTransportInstance second = new SerialTransportInstance(
+ manager, "COMSHARED", config(), AuditLog.builder().build());
+
+ try {
+ AtomicReference<String> firstThread = new AtomicReference<>();
+ AtomicReference<String> secondThread = new AtomicReference<>();
+ first.registerDataListener(() -> firstThread.compareAndSet(null,
Thread.currentThread().getName()));
+ second.registerDataListener(() -> secondThread.compareAndSet(null,
Thread.currentThread().getName()));
+
+ factory.broadcast(new byte[]{0x01, 0x02, 0x03, 0x04});
+ awaitTrue(() -> firstThread.get() != null && secondThread.get() !=
null, 5, TimeUnit.SECONDS);
+
+
assertTrue(firstThread.get().startsWith("Serial-Shared-Dispatch-"),
firstThread.get());
+
assertTrue(secondThread.get().startsWith("Serial-Shared-Dispatch-"),
secondThread.get());
+ assertNotEquals(firstThread.get(), secondThread.get(),
+ "two instances on one port must have distinct dispatch
threads");
+ } finally {
+ first.close();
+ second.close();
+ }
+ }
+
/**
* Regression test for shared-mode dispatch isolation: the shared reader
* thread iterates all subscribers' onData() inline
(SharedPort.readFromPort()).
