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
commit a361d8b238c732fcbd1a4f8a2f47a2e690a6a573 Author: Sebastian Rühl <[email protected]> AuthorDate: Mon Jul 6 20:52:54 2026 +0200 fix(plc4j): single broadcast reader for shared serial ports Align the Java shared serial port with the plc4go reference design: SharedPort owns the one reader per physical device and broadcasts every chunk to per-connection subscribers (previously each instance ran its own competing reader, splitting responses randomly between connections). Acquiring with a differing configuration now fails instead of silently reusing the first one. The new WritePacer enforces inter-frame silence measured from the last write or received data, re-checked mid-wait, on shared and dedicated ports alike. Teardown is exactly-once (CAS) with fatal-error eviction that releases the device before notifying listeners, so reconnects re-open cleanly. --- RELEASE_NOTES | 8 + .../transport/serial/SerialTransportInstance.java | 231 +++++++++--- .../transport/serial/SharedPortSubscriber.java | 33 ++ .../transport/serial/SharedSerialPortManager.java | 386 ++++++++++++++++----- .../plc4x/java/transport/serial/WritePacer.java | 85 +++++ .../config/SerialTransportConfiguration.java | 4 +- .../java/transport/serial/DropOldestWriteTest.java | 53 +++ .../transport/serial/SharedModeInstanceTest.java | 160 +++++++++ .../transport/serial/SharedPortBroadcastTest.java | 218 ++++++++++++ .../serial/SharedSerialPortManagerTest.java | 14 +- .../java/transport/serial/WritePacerTest.java | 74 ++++ 11 files changed, 1129 insertions(+), 137 deletions(-) diff --git a/RELEASE_NOTES b/RELEASE_NOTES index 9d76d24faf..3ddb5ed69d 100644 --- a/RELEASE_NOTES +++ b/RELEASE_NOTES @@ -23,6 +23,11 @@ New Features between multiple connections ("reuse-port", e.g. multi-slave Modbus RTU) with broadcast reads and serialized writes, plus inter-frame write pacing ("interframe-delay") on both shared and dedicated ports. +- The Java serial transport's shared-port mode ("reuse-port") now uses a + single broadcast reader per physical port, fixing responses being split + between connections, and "interframe-delay" write pacing works on both + shared and dedicated ports (gap measured from the last write or + received data). Incompatible changes -------------------- @@ -45,6 +50,9 @@ Incompatible changes falling back to defaults. Option values are case-insensitive and accept "-" or "_" as separator (canonical forms: none, odd, even, mark, space; none, rts-cts, xon-xoff). +- Acquiring a shared Java serial port ("reuse-port") with a configuration + differing from the first connection's now fails with an error instead + of silently reusing the first configuration. - The 'plc4x' proxy driver now defaults to the TLS transport instead of plaintext TCP. Existing plaintext connections must switch to an explicit transport prefix (e.g. "plc4x:tcp://..."). 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 05422834a7..a8b3e0fc39 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 @@ -58,22 +58,32 @@ public class SerialTransportInstance extends BaseTransportInstance<SerialTranspo private final Lock readLock = new ReentrantLock(); private final Lock writeLock = new ReentrantLock(); // Only used for non-shared ports private volatile boolean open = true; + // Guards close() so two concurrent callers can't both pass the + // check-then-act on `open` and both proceed to release resources (e.g. + // double-decrementing the shared port's refcount). + private final java.util.concurrent.atomic.AtomicBoolean closeGuard = new java.util.concurrent.atomic.AtomicBoolean(false); // Async support private volatile Runnable dataListener; private volatile Consumer<Throwable> disconnectListener; - private final SerialPortDataListener serialPortDataListener; + private SerialPortDataListener serialPortDataListener; private volatile Thread readerThread; + private final WritePacer writePacer; // dedicated-path pacing; no-op pacer in shared mode + private SharedPortSubscriber sharedSubscriber; + private long droppedBytes; + private long lastWarnedDroppedBytes; public SerialTransportInstance(SharedSerialPortManager sharedSerialPortManager, String port, SerialTransportConfiguration configuration, AuditLog auditLog) throws TransportException { super(configuration, auditLog); this.sharedSerialPortManager = sharedSerialPortManager; this.ringBuffer = new RingBuffer(DEFAULT_BUFFER_SIZE); - try { - SerialPort tempPort; - SharedSerialPortManager.SharedPort tempSharedPort = null; + // Hoisted so the catch block below can see whatever got created + // before the failure, for best-effort cleanup. + SerialPort tempPort = null; + SharedSerialPortManager.SharedPort tempSharedPort = null; + try { if (configuration.reusePort) { // Use shared port manager SharedSerialPortManager.SerialPortConfig portConfig = new SharedSerialPortManager.SerialPortConfig( @@ -144,35 +154,53 @@ public class SerialTransportInstance extends BaseTransportInstance<SerialTranspo this.port = tempPort; this.sharedPort = tempSharedPort; this.outputStream = tempPort.getOutputStream(); - - // Create the serial port data listener for async I/O - // Note: Some platforms/devices don't properly support SerialPortDataListener events, - // so we use a background reader thread as a fallback - this.serialPortDataListener = new SerialPortDataListener() { - @Override - public int getListeningEvents() { - return SerialPort.LISTENING_EVENT_DATA_AVAILABLE; - } - - @Override - public void serialEvent(SerialPortEvent event) { - if (event.getEventType() != SerialPort.LISTENING_EVENT_DATA_AVAILABLE) { - return; + this.writePacer = new WritePacer(tempSharedPort != null ? 0 : configuration.interframeDelay); + + if (tempSharedPort != null) { + // Shared mode: the SharedPort owns the single reader; this + // instance only subscribes to the broadcast. + this.serialPortDataListener = null; + this.readerThread = null; + this.sharedSubscriber = new SharedPortSubscriber() { + @Override + public void onData(byte[] data, int offset, int length) { + deliverSharedData(data, offset, length); } - readFromPort(); - } - }; - // Try to register the event listener first - boolean eventListenerRegistered = tempPort.addDataListener(serialPortDataListener); - LOGGER.debug("Serial port event listener registration result: {}", eventListenerRegistered); + @Override + public void onFailure(Throwable cause) { + Consumer<Throwable> listener = disconnectListener; + if (listener != null) { + listener.accept(cause); + } + } + }; + tempSharedPort.addSubscriber(this.sharedSubscriber); + } else { + // Dedicated mode: keep the per-instance reader exactly as before. + // Note: Some platforms/devices don't properly support SerialPortDataListener events, + // so we use a background reader thread as a fallback + this.serialPortDataListener = new SerialPortDataListener() { + @Override + public int getListeningEvents() { + return SerialPort.LISTENING_EVENT_DATA_AVAILABLE; + } - // Always start a background reader thread as fallback - // This ensures data is read even on platforms where events don't fire properly - this.readerThread = new Thread(this::readerLoop, "Serial-Reader-" + port); - this.readerThread.setDaemon(true); - this.readerThread.start(); - LOGGER.debug("Serial port reader thread started"); + @Override + public void serialEvent(SerialPortEvent event) { + if (event.getEventType() != SerialPort.LISTENING_EVENT_DATA_AVAILABLE) { + return; + } + readFromPort(); + } + }; + boolean eventListenerRegistered = tempPort.addDataListener(serialPortDataListener); + LOGGER.debug("Serial port event listener registration result: {}", eventListenerRegistered); + this.readerThread = new Thread(this::readerLoop, "Serial-Reader-" + port); + this.readerThread.setDaemon(true); + this.readerThread.start(); + LOGGER.debug("Serial port reader thread started"); + } getAuditLog().write(AuditLogEventType.CONNECT, String.format( "Serial port opened on %s at %d baud, %d%s%d, flow control: %s", @@ -182,7 +210,29 @@ public class SerialTransportInstance extends BaseTransportInstance<SerialTranspo String errorMsg = String.format("Failed to create serial transport for %s - %s", port, e.getMessage()); LOGGER.error(errorMsg, e); - getAuditLog().write(AuditLogEventType.ERROR, "Error in constructor: " + errorMsg); + + // Best-effort cleanup FIRST: never leak a subscriber/refcount + // (shared) or an open port (dedicated) out of a failed + // constructor — and never let a throwing audit log skip it. + try { + if (tempSharedPort != null) { + if (this.sharedSubscriber != null) { + tempSharedPort.removeSubscriber(this.sharedSubscriber); + } + sharedSerialPortManager.releasePort(tempSharedPort); + } else if (tempPort != null) { + tempPort.closePort(); + } + } catch (Exception cleanupError) { + LOGGER.warn("Cleanup after failed serial transport construction also failed", cleanupError); + } + + try { + getAuditLog().write(AuditLogEventType.ERROR, "Error in constructor: " + errorMsg); + } catch (Exception auditError) { + LOGGER.warn("Audit log write after failed serial transport construction also failed", auditError); + } + throw new TransportException(errorMsg, e); } } @@ -318,6 +368,11 @@ public class SerialTransportInstance extends BaseTransportInstance<SerialTranspo sharedPort.lockWrite(); } else { writeLock.lock(); + try { + writePacer.awaitTurn(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } } try { @@ -339,6 +394,7 @@ public class SerialTransportInstance extends BaseTransportInstance<SerialTranspo if (sharedPort != null) { sharedPort.unlockWrite(); } else { + writePacer.noteActivity(); writeLock.unlock(); } } @@ -346,39 +402,59 @@ public class SerialTransportInstance extends BaseTransportInstance<SerialTranspo @Override public void close() throws TransportException { - if (!open) { + if (!closeGuard.compareAndSet(false, true)) { + // Already closed (or a concurrent close is in flight); avoid + // double teardown, e.g. a double decrement of the shared port's + // refcount. return; } // Set open to false first to stop the reader thread open = false; - // Stop the reader thread - if (readerThread != null) { - readerThread.interrupt(); - try { - readerThread.join(1000); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - } - - // Remove the data listener - port.removeDataListener(); - LOGGER.debug("Serial port data listener removed"); - if (sharedPort != null) { - // Don't lock for shared ports - manager handles it + // Don't lock for shared ports - manager handles it. No reader + // thread/listener to stop here: the SharedPort owns those and + // must keep serving the other subscribers. Only remove our own + // subscription. + // + // Only flip `open` under readLock (so an in-flight + // deliverSharedData() call observes the close and bails out + // cleanly). removeSubscriber() and releasePort() must run + // OUTSIDE the lock: on the last release, releasePort() -> + // SharedPort.shutdown() joins the shared reader thread, and that + // reader may be blocked inside deliverSharedData() waiting for + // this very readLock. Calling releasePort() while holding the + // lock would make every close racing live data pay the ~1s join + // timeout (mirrors the rationale already documented on + // SharedSerialPortManager.releasePort()). readLock.lock(); try { open = false; - sharedSerialPortManager.releasePort(sharedPort); - LOGGER.debug("Released shared serial port"); - getAuditLog().write(AuditLogEventType.CLOSE, "Released shared serial port"); } finally { readLock.unlock(); } + if (sharedSubscriber != null) { + sharedPort.removeSubscriber(sharedSubscriber); + } + sharedSerialPortManager.releasePort(sharedPort); + LOGGER.debug("Released shared serial port"); + getAuditLog().write(AuditLogEventType.CLOSE, "Released shared serial port"); } else { + // Stop the reader thread (dedicated only) + if (readerThread != null) { + readerThread.interrupt(); + try { + readerThread.join(1000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + // Remove the data listener (dedicated only) + port.removeDataListener(); + LOGGER.debug("Serial port data listener removed"); + // Lock for dedicated ports writeLock.lock(); try { @@ -480,6 +556,7 @@ public class SerialTransportInstance extends BaseTransportInstance<SerialTranspo ringBuffer.write(buffer, 0, bytesRead); LOGGER.debug("Read {} bytes from serial port into ring buffer, buffer now has {} bytes", bytesRead, ringBuffer.availableForReading()); + writePacer.noteActivity(); // Notify the data listener if registered Runnable listener = dataListener; @@ -492,6 +569,60 @@ public class SerialTransportInstance extends BaseTransportInstance<SerialTranspo } } + /** + * Shared-mode data path: writes a broadcast chunk into this instance's + * ring buffer with drop-OLDEST overflow semantics (mirroring plc4go) and + * wakes the codec-facing data listener. Called from the shared reader + * thread. + */ + private void deliverSharedData(byte[] data, int offset, int length) { + readLock.lock(); + try { + if (!open) { + return; + } + int dropped = writeDroppingOldest(ringBuffer, data, offset, length); + if (dropped > 0) { + droppedBytes += dropped; + if (lastWarnedDroppedBytes == 0 || droppedBytes >= 2 * lastWarnedDroppedBytes) { + lastWarnedDroppedBytes = droppedBytes; + LOGGER.warn("Ring buffer overflow on shared serial port, dropped {} oldest bytes (total dropped: {})", + dropped, droppedBytes); + } + } + Runnable listener = dataListener; + if (listener != null) { + listener.run(); + } + } finally { + readLock.unlock(); + } + } + + /** + * Writes a chunk into the ring buffer, dropping the OLDEST bytes when + * capacity would be exceeded (mirroring the plc4go shared-port ring + * semantics — RingBuffer.write alone would drop the newest). Returns + * how many bytes were dropped. + */ + static int writeDroppingOldest(RingBuffer ringBuffer, byte[] data, int offset, int length) { + int capacity = ringBuffer.capacity(); + if (length >= capacity) { + int dropped = ringBuffer.availableForReading() + (length - capacity); + ringBuffer.clear(); + ringBuffer.write(data, offset + (length - capacity), capacity); + return dropped; + } + int remaining = ringBuffer.remainingForWriting(); + int dropped = 0; + if (remaining < length) { + dropped = length - remaining; + ringBuffer.skip(dropped); + } + ringBuffer.write(data, offset, length); + return dropped; + } + /** * Background reader loop that polls for data from the serial port. * This is used as a fallback for platforms where SerialPortDataListener events diff --git a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SharedPortSubscriber.java b/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SharedPortSubscriber.java new file mode 100644 index 0000000000..52c2e08a99 --- /dev/null +++ b/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SharedPortSubscriber.java @@ -0,0 +1,33 @@ +/* + * 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.transport.serial; + +/** + * Receives the broadcast of everything a shared serial port reads. Every + * subscriber sees the FULL wire traffic (protocol codecs pick out their own + * frames), mirroring the plc4go shared-port design. + */ +interface SharedPortSubscriber { + + /** A chunk read from the physical port. Called from the reader thread. */ + void onData(byte[] data, int offset, int length); + + /** The physical port failed fatally; the shared port has been evicted. */ + void onFailure(Throwable cause); +} diff --git a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SharedSerialPortManager.java b/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SharedSerialPortManager.java index a34b98dd83..334002d068 100644 --- a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SharedSerialPortManager.java +++ b/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SharedSerialPortManager.java @@ -19,14 +19,16 @@ package org.apache.plc4x.java.transport.serial; import com.fazecast.jSerialComm.SerialPort; +import org.apache.plc4x.java.spi.transports.api.exceptions.TransportException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.HashMap; import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Function; /** * Manages shared serial ports across multiple transport instances. @@ -38,63 +40,81 @@ public class SharedSerialPortManager { private static final Logger LOGGER = LoggerFactory.getLogger(SharedSerialPortManager.class); // Map of port name -> shared port wrapper - private final Map<String, SharedPort> sharedPorts = new ConcurrentHashMap<>(); + private final Map<String, SharedPort> sharedPorts = new HashMap<>(); + + private final Function<String, SerialPort> portFactory; public SharedSerialPortManager() { - // Private constructor for singleton + this(SerialPort::getCommPort); + } + + // Visible for tests: lets unit tests supply mocked physical ports. + SharedSerialPortManager(Function<String, SerialPort> portFactory) { + this.portFactory = portFactory; } /** - * Acquires a shared serial port for the given port name and configuration. - * If a port is already open for this name, returns the existing one. - * Otherwise creates a new one with the provided configuration. + * Acquires a shared serial port. Joining an existing entry requires an + * IDENTICAL configuration; a mismatch is a hard error (no silent + * first-config-wins). */ - public SharedPort acquirePort(String portName, SerialPortConfig config) { - return sharedPorts.compute(portName, (name, existing) -> { - if (existing != null) { - existing.incrementRefCount(); - LOGGER.debug("Reusing shared serial port {} (refCount={})", - portName, existing.getRefCount()); - return existing; - } else { - SerialPort port = SerialPort.getCommPort(portName); - - // Configure port - port.setBaudRate(config.baudRate); - port.setNumDataBits(config.dataBits); - port.setNumStopBits(config.stopBits); - port.setParity(config.parity); - port.setFlowControl(config.flowControl); - - // Set timeouts - port.setComPortTimeouts( - SerialPort.TIMEOUT_READ_SEMI_BLOCKING | SerialPort.TIMEOUT_WRITE_BLOCKING, - config.readTimeout, - config.writeTimeout - ); - - // Open port - if (!port.openPort()) { - throw new RuntimeException("Failed to open serial port: " + portName); - } + public synchronized SharedPort acquirePort(String portName, SerialPortConfig config) throws TransportException { + SharedPort existing = sharedPorts.get(portName); + if (existing != null && existing.isClosed()) { + // An acquire can race the fail() window where the entry is + // dying (closed CAS already flipped) but not yet evicted from + // the map (evict() and this method both synchronize on the + // manager, but fail() may not have reached evict() yet, or a + // stale entry may still linger under this name). Treat it as + // absent and fall through to opening a fresh port. + sharedPorts.remove(portName); + existing = null; + } + if (existing != null) { + if (!existing.getConfig().matches(config)) { + throw new TransportException(String.format( + "Serial port %s is already shared with a different configuration (existing: %s, requested: %s)", + portName, existing.getConfig(), config)); + } + existing.incrementRefCount(); + LOGGER.debug("Reusing shared serial port {} (refCount={})", portName, existing.getRefCount()); + return existing; + } - // Set DTR/RTS - if (config.dtr) { - port.setDTR(); - } else { - port.clearDTR(); - } + SerialPort port = portFactory.apply(portName); - if (config.rts) { - port.setRTS(); - } else { - port.clearRTS(); - } + // Configure port + port.setBaudRate(config.baudRate); + port.setNumDataBits(config.dataBits); + port.setNumStopBits(config.stopBits); + port.setParity(config.parity); + port.setFlowControl(config.flowControl); + port.setComPortTimeouts( + SerialPort.TIMEOUT_READ_SEMI_BLOCKING | SerialPort.TIMEOUT_WRITE_BLOCKING, + config.readTimeout, + config.writeTimeout + ); - LOGGER.info("Opened shared serial port {} at {} baud", portName, config.baudRate); - return new SharedPort(port, portName, config.interframeDelay); - } - }); + if (!port.openPort()) { + throw new TransportException("Failed to open serial port: " + portName); + } + + if (config.dtr) { + port.setDTR(); + } else { + port.clearDTR(); + } + if (config.rts) { + port.setRTS(); + } else { + port.clearRTS(); + } + + LOGGER.info("Opened shared serial port {} at {} baud", portName, config.baudRate); + SharedPort sharedPort = new SharedPort(port, portName, config, this); + sharedPorts.put(portName, sharedPort); + sharedPort.startReader(); + return sharedPort; } /** @@ -103,24 +123,43 @@ public class SharedSerialPortManager { */ public void releasePort(SharedPort port) { String portName = port.getPortName(); - - sharedPorts.compute(portName, (name, existing) -> { - if (existing == null) { - LOGGER.warn("Attempted to release non-existent port: {}", portName); - return null; + boolean shouldShutdown; + synchronized (this) { + SharedPort existing = sharedPorts.get(portName); + if (existing != port) { + // The entry was already evicted (fatal failure) and possibly + // replaced by a fresh acquire under the same name. A stale + // holder releasing here must not touch the new entry. + LOGGER.warn("releasing an already-evicted shared serial port {}", portName); + return; } - int refCount = existing.decrementRefCount(); LOGGER.debug("Released shared serial port {} (refCount={})", portName, refCount); - - if (refCount <= 0) { - existing.getPort().closePort(); - LOGGER.info("Closed shared serial port {}", portName); - return null; // Remove from map + shouldShutdown = refCount <= 0; + if (shouldShutdown) { + sharedPorts.remove(portName); } + } + // shutdown() joins the reader thread; it must run OUTSIDE the + // manager lock. Otherwise, a reader concurrently inside fail() + // (which calls the synchronized manager.evict(this)) would stall the + // join while holding up this lock, letting closePort() race with the + // still-in-progress fail(). The closed-guard in SharedPort makes the + // ordering safe regardless, but staying off the lock avoids the stall. + if (shouldShutdown) { + port.shutdown(); + LOGGER.info("Closed shared serial port {}", portName); + } + } - return existing; - }); + /** + * Evicts a shared port entry after a fatal failure so that a subsequent + * acquire re-opens the physical device instead of reusing a dead entry. + */ + synchronized void evict(SharedPort port) { + if (sharedPorts.get(port.getPortName()) == port) { + sharedPorts.remove(port.getPortName()); + } } /** @@ -152,23 +191,54 @@ public class SharedSerialPortManager { this.rts = rts; this.interframeDelay = interframeDelay; } + + public boolean matches(SerialPortConfig other) { + return other != null + && baudRate == other.baudRate + && dataBits == other.dataBits + && stopBits == other.stopBits + && parity == other.parity + && flowControl == other.flowControl + && readTimeout == other.readTimeout + && writeTimeout == other.writeTimeout + && dtr == other.dtr + && rts == other.rts + && interframeDelay == other.interframeDelay; + } + + @Override + public String toString() { + return String.format( + "baud=%d dataBits=%d stopBits=%d parity=%d flowControl=%d readTimeout=%d writeTimeout=%d dtr=%b rts=%b interframeDelay=%d", + baudRate, dataBits, stopBits, parity, flowControl, readTimeout, writeTimeout, dtr, rts, interframeDelay); + } } /** * Wrapper around a SerialPort with reference counting and shared access control. */ public static class SharedPort { + private static final int READER_CHUNK = 4096; + private final SerialPort port; private final String portName; - private final int interframeDelay; + private final SerialPortConfig config; private final AtomicInteger refCount = new AtomicInteger(1); private final Lock writeLock = new ReentrantLock(); - private volatile long lastWriteTime = 0; - private SharedPort(SerialPort port, String portName, int interframeDelay) { + private final java.util.List<SharedPortSubscriber> subscribers = new java.util.concurrent.CopyOnWriteArrayList<>(); + private final WritePacer pacer; + private volatile boolean open = true; + private volatile Thread readerThread; + private final SharedSerialPortManager manager; + private final java.util.concurrent.atomic.AtomicBoolean closed = new java.util.concurrent.atomic.AtomicBoolean(false); + + private SharedPort(SerialPort port, String portName, SerialPortConfig config, SharedSerialPortManager manager) { this.port = port; this.portName = portName; - this.interframeDelay = interframeDelay; + this.config = config; + this.manager = manager; + this.pacer = new WritePacer(config.interframeDelay); } public SerialPort getPort() { @@ -179,6 +249,10 @@ public class SharedSerialPortManager { return portName; } + public SerialPortConfig getConfig() { + return config; + } + public int getRefCount() { return refCount.get(); } @@ -191,23 +265,179 @@ public class SharedSerialPortManager { return refCount.decrementAndGet(); } + /** + * True once this entry has started (or finished) tearing down, via + * either {@link #fail(Throwable)} or {@link #shutdown()}. Lets + * {@link SharedSerialPortManager#acquirePort} detect a dying entry + * that hasn't been evicted from the map yet. + */ + boolean isClosed() { + return closed.get(); + } + + void addSubscriber(SharedPortSubscriber subscriber) { + subscribers.add(subscriber); + } + + void removeSubscriber(SharedPortSubscriber subscriber) { + subscribers.remove(subscriber); + } + + /** + * Starts the single reader for this physical port: an event listener + * plus a polling fallback thread (some platforms/devices never fire + * events). Both funnel through readFromPort(), which is what + * broadcasts to subscribers — the per-port reader replaces the old + * per-instance competing readers. + */ + void startReader() { + port.addDataListener(new com.fazecast.jSerialComm.SerialPortDataListener() { + @Override + public int getListeningEvents() { + return SerialPort.LISTENING_EVENT_DATA_AVAILABLE; + } + + @Override + public void serialEvent(com.fazecast.jSerialComm.SerialPortEvent event) { + if (event.getEventType() == SerialPort.LISTENING_EVENT_DATA_AVAILABLE) { + readFromPort(); + } + } + }); + readerThread = new Thread(this::readerLoop, "Shared-Serial-Reader-" + portName); + readerThread.setDaemon(true); + readerThread.start(); + } + + private void readerLoop() { + while (open && !Thread.currentThread().isInterrupted()) { + try { + readFromPort(); + Thread.sleep(10); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } catch (Exception e) { + // An unexpected RuntimeException out of readFromPort() + // must escalate to a fatal failure instead of silently + // killing the polling thread and leaving subscribers + // waiting on a port that looks alive but has no reader. + fail(e); + return; + } + } + } + + private synchronized void readFromPort() { + if (!open) { + return; + } + int available = port.bytesAvailable(); + if (available < 0) { + fail(new java.io.IOException("Serial port " + portName + " reported an error (bytesAvailable=" + available + ")")); + return; + } + if (available == 0) { + return; + } + byte[] buffer = new byte[Math.min(available, READER_CHUNK)]; + int bytesRead = port.readBytes(buffer, buffer.length); + if (bytesRead < 0) { + fail(new java.io.IOException("Serial port " + portName + " read failed (" + bytesRead + ")")); + return; + } + if (bytesRead == 0) { + return; + } + pacer.noteActivity(); + for (SharedPortSubscriber subscriber : subscribers) { + try { + subscriber.onData(buffer, 0, bytesRead); + } catch (Exception e) { + LOGGER.warn("Shared serial subscriber threw in onData on {}", portName, e); + } + } + } + + /** + * Fatal path: evict the entry and release the physical port FIRST, + * THEN notify subscribers. + * + * The fd must be released before listeners run: a disconnect + * listener can synchronously attempt to reconnect, and jSerialComm + * holds an exclusive lock on the device, so a reopen would fail if + * this port hadn't already released it. Ordering: evict (so the + * reconnect's acquirePort() re-opens the device instead of reusing + * the dying entry) -> removeDataListener/closePort (release the fd) + * -> onFailure to all subscribers. + * + * Guarded by {@link #closed} so that a concurrent {@link #shutdown()} + * (triggered by a release racing this failure) can't result in both + * paths running: whichever loses the CAS no-ops entirely, avoiding a + * stale onFailure delivery or a double close. + */ + private void fail(Throwable cause) { + if (!closed.compareAndSet(false, true)) { + return; + } + open = false; + LOGGER.warn("Shared serial port {} failed", portName, cause); + manager.evict(this); + port.removeDataListener(); + port.closePort(); + for (SharedPortSubscriber subscriber : subscribers) { + try { + subscriber.onFailure(cause); + } catch (Exception e) { + LOGGER.warn("Shared serial subscriber threw in onFailure on {}", portName, e); + } + } + } + + /** + * Stops the reader (listener + thread) and closes the port. + * + * Ordering rationale: the port is closed BEFORE interrupting/joining + * the reader thread. A reader thread can be parked inside a native + * jSerialComm blocking read; a plain Java {@code Thread.interrupt()} + * does not unblock that native call, but closing the underlying port + * does. Once closed, the reader loop's {@code open} check (both at + * the top of {@link #readFromPort()} and in the {@code readerLoop} + * condition) causes it to exit quietly rather than treating the + * close as a fatal failure. The {@link #closed} CAS guard also makes + * this safe against a concurrent {@link #fail()}: whichever call + * wins the race performs the teardown, the other is a no-op, so + * shutdown() never has to wait on the manager lock that a + * concurrent fail()->manager.evict() might be holding. + */ + void shutdown() { + if (!closed.compareAndSet(false, true)) { + return; + } + open = false; + port.removeDataListener(); + port.closePort(); + Thread thread = readerThread; + if (thread != null) { + thread.interrupt(); + try { + thread.join(1000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + } + /** * Acquires the write lock for this port. * Ensures only one instance writes at a time and enforces interframe delay. */ public void lockWrite() { writeLock.lock(); - - // Enforce interframe delay - if (interframeDelay > 0 && lastWriteTime > 0) { - long elapsed = System.currentTimeMillis() - lastWriteTime; - if (elapsed < interframeDelay) { - try { - Thread.sleep(interframeDelay - elapsed); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - } + try { + pacer.awaitTurn(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); } } @@ -215,7 +445,7 @@ public class SharedSerialPortManager { * Releases the write lock and updates last write time. */ public void unlockWrite() { - lastWriteTime = System.currentTimeMillis(); + pacer.noteActivity(); writeLock.unlock(); } } diff --git a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/WritePacer.java b/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/WritePacer.java new file mode 100644 index 0000000000..9d0d325984 --- /dev/null +++ b/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/WritePacer.java @@ -0,0 +1,85 @@ +/* + * 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.transport.serial; + +/** + * Enforces a minimum silence gap before writes on a serial line, measured + * from the later of the last write end and the last observed read activity + * (Modbus RTU t3.5-style semantics). Mirrors plc4go's pacer. + * + * A zero (or negative) delay disables pacing. The pacer only tracks + * timestamps — it does not serialize concurrent writers; callers must hold + * their own write lock around awaitTurn() + write for the gap guarantee to + * hold. + */ +final class WritePacer { + + private final long delayMillis; + private final long delayNanos; + private final Object lock = new Object(); + private long lastActivityNanos; + + WritePacer(long delayMillis) { + this.delayMillis = delayMillis; + this.delayNanos = delayMillis * 1_000_000L; + // Initialized so a fresh pacer never waits: the first awaitTurn() + // sees "last activity" as already delayNanos in the past. + this.lastActivityNanos = System.nanoTime() - delayNanos; + } + + /** + * Records line activity (a completed write or received data). Monotone: + * never moves the timestamp backwards. Uses {@link System#nanoTime()}, + * which is immune to wall-clock adjustments (NTP, DST, manual changes) + * that would otherwise corrupt the gap computation. + */ + void noteActivity() { + if (delayMillis <= 0) { + return; + } + synchronized (lock) { + long now = System.nanoTime(); + if (now - lastActivityNanos > 0) { + lastActivityNanos = now; + } + } + } + + /** + * Sleeps until the configured gap since the last activity has elapsed. + * Recomputes after every sleep, so activity arriving during the wait + * extends the gap. + */ + void awaitTurn() throws InterruptedException { + if (delayMillis <= 0) { + return; + } + while (true) { + long ready; + synchronized (lock) { + ready = lastActivityNanos + delayNanos; + } + long waitNanos = ready - System.nanoTime(); + if (waitNanos <= 0) { + return; + } + Thread.sleep(waitNanos / 1_000_000L, (int) (waitNanos % 1_000_000L)); + } + } +} diff --git a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/config/SerialTransportConfiguration.java b/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/config/SerialTransportConfiguration.java index 41d4c6cafe..71b1a0e981 100644 --- a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/config/SerialTransportConfiguration.java +++ b/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/config/SerialTransportConfiguration.java @@ -112,10 +112,10 @@ public class SerialTransportConfiguration implements TransportConfiguration { /** * Interframe delay in milliseconds for protocols that need spacing between messages. - * Only applies when reusePort is true. + * Applies to shared and dedicated ports; the gap is measured from the last write or received data. */ @ConfigurationParameter( "interframe-delay") - @Description( "Interframe delay in milliseconds for protocols that need spacing between messages. Only applies when reusePort is true.") + @Description( "Interframe delay in milliseconds for protocols that need spacing between messages. Applies to shared and dedicated ports; the gap is measured from the last write or received data.") @IntDefaultValue(0) public int interframeDelay; } diff --git a/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/DropOldestWriteTest.java b/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/DropOldestWriteTest.java new file mode 100644 index 0000000000..82d8ed7b64 --- /dev/null +++ b/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/DropOldestWriteTest.java @@ -0,0 +1,53 @@ +/* + * 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.transport.serial; + +import org.apache.plc4x.java.spi.transports.api.RingBuffer; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.*; + +class DropOldestWriteTest { + + @Test + void writesWithoutDropWhenSpaceSuffices() { + RingBuffer buffer = new RingBuffer(8); + int dropped = SerialTransportInstance.writeDroppingOldest(buffer, new byte[]{1, 2, 3}, 0, 3); + assertEquals(0, dropped); + assertArrayEquals(new byte[]{1, 2, 3}, buffer.read(3)); + } + + @Test + void dropsOldestOnOverflow() { + RingBuffer buffer = new RingBuffer(4); + SerialTransportInstance.writeDroppingOldest(buffer, new byte[]{'a', 'b'}, 0, 2); + int dropped = SerialTransportInstance.writeDroppingOldest(buffer, new byte[]{'c', 'd', 'e', 'f'}, 0, 4); + assertEquals(2, dropped, "the two oldest bytes must be dropped"); + assertArrayEquals(new byte[]{'c', 'd', 'e', 'f'}, buffer.read(4), "newest bytes survive"); + } + + @Test + void chunkLargerThanCapacityKeepsTail() { + RingBuffer buffer = new RingBuffer(4); + byte[] big = {'0', '1', '2', '3', '4', '5', '6', '7', '8', '9'}; + int dropped = SerialTransportInstance.writeDroppingOldest(buffer, big, 0, big.length); + assertEquals(6, dropped); + assertArrayEquals(new byte[]{'6', '7', '8', '9'}, buffer.read(4), "only the tail fits"); + } +} 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 new file mode 100644 index 0000000000..dd0e0446e0 --- /dev/null +++ b/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/SharedModeInstanceTest.java @@ -0,0 +1,160 @@ +/* + * 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.transport.serial; + +import com.fazecast.jSerialComm.SerialPort; +import org.apache.plc4x.java.transport.serial.config.SerialTransportConfiguration; +import org.apache.plc4x.java.utils.auditlog.api.AuditLog; +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayOutputStream; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Function; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; + +/** + * Instance-level integration test for shared-port ("reuse-port") mode: two + * real {@link SerialTransportInstance}s backed by a single mocked physical + * port, verifying that both receive a broadcast, that closing one instance + * does not disturb the other, and that the physical port is only closed once + * the last instance releases it. + */ +class SharedModeInstanceTest { + + private static SerialTransportConfiguration config() { + SerialTransportConfiguration config = new SerialTransportConfiguration(); + config.baudRate = 9600; + config.dataBits = 8; + config.stopBits = 1; + config.parity = "NONE"; + config.flowControl = "NONE"; + config.readTimeout = 1000; + config.writeTimeout = 1000; + config.reusePort = true; + return config; + } + + /** + * Mirrors SharedPortBroadcastTest's MockPortFactory, extended with a + * one-shot "broadcast" mechanism. All stubs are attached once, at mock + * creation time (i.e. before the SharedPort's reader thread is started + * by SharedSerialPortManager.acquirePort()), so there is no live + * re-stubbing of a mock the reader thread might be concurrently calling. + */ + static final class MockPortFactory implements Function<String, SerialPort> { + final Map<String, SerialPort> created = new HashMap<>(); + private final AtomicInteger bytesAvailable = new AtomicInteger(0); + private final AtomicReference<byte[]> pendingPayload = new AtomicReference<>(); + + @Override + public SerialPort apply(String portName) { + SerialPort port = mock(SerialPort.class); + when(port.openPort()).thenReturn(true); + // SerialTransportInstance.isOpen() checks `open && port.isOpen()`; + // without this stub the mock's default `false` would make + // getNumBytesAvailable()/read() short-circuit to empty forever. + when(port.isOpen()).thenReturn(true); + when(port.getOutputStream()).thenReturn(new ByteArrayOutputStream()); + // Stub readBytes first, then bytesAvailable, both at mock-creation + // time - before the reader thread exists - so there is no window + // where the reader could observe a non-zero bytesAvailable with + // an unstubbed (default) readBytes. + when(port.readBytes(any(byte[].class), anyInt())).thenAnswer(invocation -> { + byte[] buffer = invocation.getArgument(0); + byte[] payload = pendingPayload.get(); + if (payload == null) { + return 0; + } + System.arraycopy(payload, 0, buffer, 0, payload.length); + return payload.length; + }); + when(port.bytesAvailable()).thenAnswer(inv -> bytesAvailable.getAndSet(0)); + created.put(portName, port); + return port; + } + + /** + * Arms a one-shot broadcast: the next poll of the shared reader + * thread will read exactly this payload and hand it to all + * subscribers, then go quiet again. + */ + void broadcast(byte[] payload) { + pendingPayload.set(payload); + bytesAvailable.set(payload.length); + } + } + + private static void awaitBytesAvailable(SerialTransportInstance instance, int expected) throws Exception { + long deadline = System.currentTimeMillis() + 5000; + while (System.currentTimeMillis() < deadline) { + if (instance.getNumBytesAvailable() >= expected) { + return; + } + Thread.sleep(10); + } + fail("Timed out waiting for " + expected + " bytes to become available"); + } + + @Test + void twoSharedInstancesReceiveBroadcastsAndCloseIndependently() 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()); + + assertEquals(1, factory.created.size(), "both instances must share one physical port"); + SerialPort mockPort = factory.created.get("COMSHARED"); + + byte[] payload = {0x01, 0x02, 0x03}; + factory.broadcast(payload); + + awaitBytesAvailable(first, payload.length); + awaitBytesAvailable(second, payload.length); + + assertArrayEquals(payload, first.read(payload.length), "first instance must receive the broadcast"); + assertArrayEquals(payload, second.read(payload.length), "second instance must receive the broadcast"); + + // Closing the first instance must not affect the second: no join + // stall (Finding 1) and the physical port must stay open since the + // second instance still holds a reference. + first.close(); + assertFalse(first.isOpen()); + verify(mockPort, never()).closePort(); + + byte[] secondPayload = {0x04, 0x05}; + factory.broadcast(secondPayload); + + awaitBytesAvailable(second, secondPayload.length); + assertArrayEquals(secondPayload, second.read(secondPayload.length), + "second instance must keep receiving broadcasts after the first instance closed"); + + // Closing the last instance must release the physical port. + second.close(); + assertFalse(second.isOpen()); + verify(mockPort).closePort(); + } +} diff --git a/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/SharedPortBroadcastTest.java b/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/SharedPortBroadcastTest.java new file mode 100644 index 0000000000..d826f1b4d8 --- /dev/null +++ b/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/SharedPortBroadcastTest.java @@ -0,0 +1,218 @@ +/* + * 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.transport.serial; + +import com.fazecast.jSerialComm.SerialPort; +import org.apache.plc4x.java.spi.transports.api.exceptions.TransportException; +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayOutputStream; +import java.util.HashMap; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; + +class SharedPortBroadcastTest { + + private static SharedSerialPortManager.SerialPortConfig config(int baudRate) { + return new SharedSerialPortManager.SerialPortConfig( + baudRate, 8, 1, SerialPort.NO_PARITY, SerialPort.FLOW_CONTROL_DISABLED, + 1000, 1000, false, false, 0); + } + + static final class MockPortFactory implements java.util.function.Function<String, SerialPort> { + final Map<String, SerialPort> created = new HashMap<>(); + + @Override + public SerialPort apply(String portName) { + SerialPort port = mock(SerialPort.class); + when(port.openPort()).thenReturn(true); + when(port.getOutputStream()).thenReturn(new ByteArrayOutputStream()); + when(port.bytesAvailable()).thenReturn(0); + created.put(portName, port); + return port; + } + } + + @Test + void identicalConfigSharesOnePhysicalPort() throws Exception { + MockPortFactory factory = new MockPortFactory(); + SharedSerialPortManager manager = new SharedSerialPortManager(factory); + + SharedSerialPortManager.SharedPort first = manager.acquirePort("COMX", config(9600)); + SharedSerialPortManager.SharedPort second = manager.acquirePort("COMX", config(9600)); + + assertSame(first, second, "identical config must share the entry"); + assertEquals(1, factory.created.size(), "only one physical open"); + } + + @Test + void configMismatchThrowsNamingThePort() throws Exception { + SharedSerialPortManager manager = new SharedSerialPortManager(new MockPortFactory()); + manager.acquirePort("COMY", config(9600)); + + TransportException e = assertThrows(TransportException.class, + () -> manager.acquirePort("COMY", config(19200))); + assertTrue(e.getMessage().contains("COMY"), "error must name the port: " + e.getMessage()); + } + + @Test + void lastReleaseClosesThePort() throws Exception { + MockPortFactory factory = new MockPortFactory(); + SharedSerialPortManager manager = new SharedSerialPortManager(factory); + SharedSerialPortManager.SharedPort shared = manager.acquirePort("COMZ", config(9600)); + manager.acquirePort("COMZ", config(9600)); + + manager.releasePort(shared); + verify(factory.created.get("COMZ"), never()).closePort(); + + manager.releasePort(shared); + verify(factory.created.get("COMZ")).closePort(); + + // A fresh acquire after full release opens a new physical port. + manager.acquirePort("COMZ", config(9600)); + assertEquals(1, factory.created.size(), "map replaced same-name entry"); + } + + static final class CollectingSubscriber implements SharedPortSubscriber { + final ByteArrayOutputStream received = new ByteArrayOutputStream(); + final java.util.concurrent.CountDownLatch dataLatch = new java.util.concurrent.CountDownLatch(1); + final java.util.concurrent.CountDownLatch failureLatch = new java.util.concurrent.CountDownLatch(1); + volatile Throwable failure; + + @Override + public synchronized void onData(byte[] data, int offset, int length) { + received.write(data, offset, length); + dataLatch.countDown(); + } + + @Override + public void onFailure(Throwable cause) { + failure = cause; + failureLatch.countDown(); + } + + synchronized byte[] bytes() { + return received.toByteArray(); + } + } + + @Test + void readerBroadcastsToAllSubscribers() throws Exception { + MockPortFactory factory = new MockPortFactory(); + SharedSerialPortManager manager = new SharedSerialPortManager(factory); + SharedSerialPortManager.SharedPort shared = manager.acquirePort("COMB", config(9600)); + SerialPort port = factory.created.get("COMB"); + + CollectingSubscriber first = new CollectingSubscriber(); + CollectingSubscriber second = new CollectingSubscriber(); + shared.addSubscriber(first); + shared.addSubscriber(second); + + byte[] payload = {0x01, 0x02, 0x03}; + // The shared reader thread is already running and polling the mock + // concurrently. Mockito's stubbing bookkeeping (which invocation a + // "when(...)" call attaches its answer to) is shared per-mock, not + // thread-local, so restubbing here while the reader thread is free + // to call the same mock races: a reader-thread invocation of + // bytesAvailable() can interleave with the test thread evaluating + // "when(port.readBytes(...))" and end up attached to the wrong + // invocation, corrupting the mock's stub table. Take SharedPort's + // own monitor (the same lock readFromPort() synchronizes on) to + // block the reader for the duration of the stubbing calls, so the + // reorder below is deterministic rather than merely probable. + synchronized (shared) { + when(port.readBytes(any(byte[].class), anyInt())).thenAnswer(invocation -> { + byte[] buffer = invocation.getArgument(0); + System.arraycopy(payload, 0, buffer, 0, payload.length); + return payload.length; + }); + when(port.bytesAvailable()).thenReturn(payload.length, 0); + } + + assertTrue(first.dataLatch.await(5, java.util.concurrent.TimeUnit.SECONDS), "first subscriber got data"); + assertTrue(second.dataLatch.await(5, java.util.concurrent.TimeUnit.SECONDS), "second subscriber got data"); + assertArrayEquals(payload, first.bytes()); + assertArrayEquals(payload, second.bytes()); + + manager.releasePort(shared); + } + + @Test + void fatalReadErrorNotifiesSubscribersAndEvicts() throws Exception { + MockPortFactory factory = new MockPortFactory(); + SharedSerialPortManager manager = new SharedSerialPortManager(factory); + SharedSerialPortManager.SharedPort shared = manager.acquirePort("COMF", config(9600)); + SerialPort port = factory.created.get("COMF"); + + CollectingSubscriber subscriber = new CollectingSubscriber(); + shared.addSubscriber(subscriber); + + // Stub under the SharedPort monitor: readFromPort() synchronizes on + // it, so this blocks the live reader while Mockito's (non-thread-safe) + // stub bookkeeping is mutated. readBytes is stubbed first so the + // reader can never see a non-zero bytesAvailable with an unstubbed read. + synchronized (shared) { + when(port.readBytes(any(byte[].class), anyInt())).thenReturn(-1); // fatal + when(port.bytesAvailable()).thenReturn(1); + } + + assertTrue(subscriber.failureLatch.await(5, java.util.concurrent.TimeUnit.SECONDS), "onFailure fired"); + assertNotNull(subscriber.failure); + + // Evicted: a fresh acquire opens a new physical port. + manager.acquirePort("COMF", config(9600)); + assertEquals(1, factory.created.size(), "entry was evicted and re-created under the same name"); + } + + @Test + void lastReleaseStopsReaderBeforeClosingPort() throws Exception { + MockPortFactory factory = new MockPortFactory(); + SharedSerialPortManager manager = new SharedSerialPortManager(factory); + SharedSerialPortManager.SharedPort shared = manager.acquirePort("COMS", config(9600)); + SerialPort port = factory.created.get("COMS"); + + manager.releasePort(shared); + + org.mockito.InOrder inOrder = inOrder(port); + inOrder.verify(port).removeDataListener(); + inOrder.verify(port).closePort(); + } + + @Test + void sharedWritesArePaced() throws Exception { + MockPortFactory factory = new MockPortFactory(); + SharedSerialPortManager manager = new SharedSerialPortManager(factory); + SharedSerialPortManager.SerialPortConfig paced = new SharedSerialPortManager.SerialPortConfig( + 9600, 8, 1, SerialPort.NO_PARITY, SerialPort.FLOW_CONTROL_DISABLED, + 1000, 1000, false, false, 60); + SharedSerialPortManager.SharedPort shared = manager.acquirePort("COMP", paced); + + shared.lockWrite(); + shared.unlockWrite(); + long start = System.currentTimeMillis(); + shared.lockWrite(); + shared.unlockWrite(); + assertTrue(System.currentTimeMillis() - start >= 50, + "second write must wait out the inter-frame gap"); + + manager.releasePort(shared); + } +} diff --git a/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/SharedSerialPortManagerTest.java b/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/SharedSerialPortManagerTest.java index 6c6eab1a44..b0c67c9293 100644 --- a/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/SharedSerialPortManagerTest.java +++ b/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/SharedSerialPortManagerTest.java @@ -60,7 +60,7 @@ class SharedSerialPortManagerTest { // Cleanup manager.releasePort(sharedPort); - } catch (RuntimeException e) { + } catch (Exception e) { // Port might be in use System.out.println("Could not open port (expected if in use): " + e.getMessage()); } @@ -82,7 +82,7 @@ class SharedSerialPortManagerTest { // Cleanup manager.releasePort(port1); manager.releasePort(port2); - } catch (RuntimeException e) { + } catch (Exception e) { System.out.println("Could not open port (expected if in use): " + e.getMessage()); } } @@ -108,7 +108,7 @@ class SharedSerialPortManagerTest { manager.releasePort(port); assertEquals(0, port.getRefCount()); assertFalse(port.getPort().isOpen()); - } catch (RuntimeException e) { + } catch (Exception e) { System.out.println("Could not open port (expected if in use): " + e.getMessage()); } } @@ -127,7 +127,7 @@ class SharedSerialPortManagerTest { assertFalse(port.getPort().isOpen()); assertEquals(0, port.getRefCount()); - } catch (RuntimeException e) { + } catch (Exception e) { System.out.println("Could not open port (expected if in use): " + e.getMessage()); } } @@ -151,7 +151,7 @@ class SharedSerialPortManagerTest { // Cleanup manager.releasePort(port2); - } catch (RuntimeException e) { + } catch (Exception e) { System.out.println("Could not open port (expected if in use): " + e.getMessage()); } } @@ -189,7 +189,7 @@ class SharedSerialPortManagerTest { // Cleanup manager.releasePort(port); - } catch (RuntimeException e) { + } catch (Exception e) { System.out.println("Could not open port (expected if in use): " + e.getMessage()); } } @@ -222,7 +222,7 @@ class SharedSerialPortManagerTest { // Cleanup manager.releasePort(port); - } catch (RuntimeException e) { + } catch (Exception e) { System.out.println("Could not open port (expected if in use): " + e.getMessage()); } } diff --git a/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/WritePacerTest.java b/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/WritePacerTest.java new file mode 100644 index 0000000000..39d2f8f397 --- /dev/null +++ b/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/WritePacerTest.java @@ -0,0 +1,74 @@ +/* + * 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.transport.serial; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.*; + +class WritePacerTest { + + @Test + void zeroDelayIsNoop() throws Exception { + WritePacer pacer = new WritePacer(0); + pacer.noteActivity(); + long start = System.currentTimeMillis(); + pacer.awaitTurn(); + assertTrue(System.currentTimeMillis() - start < 20, "zero delay must not wait"); + } + + @Test + void enforcesGapAfterActivity() throws Exception { + WritePacer pacer = new WritePacer(60); + pacer.noteActivity(); + long start = System.currentTimeMillis(); + pacer.awaitTurn(); + assertTrue(System.currentTimeMillis() - start >= 50, "must wait out the gap"); + } + + @Test + void noWaitWhenGapElapsed() throws Exception { + WritePacer pacer = new WritePacer(30); + pacer.noteActivity(); + Thread.sleep(40); + long start = System.currentTimeMillis(); + pacer.awaitTurn(); + assertTrue(System.currentTimeMillis() - start < 20, "gap already elapsed"); + } + + @Test + void activityDuringWaitExtendsGap() throws Exception { + WritePacer pacer = new WritePacer(80); + pacer.noteActivity(); + Thread traffic = new Thread(() -> { + try { + Thread.sleep(40); + pacer.noteActivity(); // traffic arrives while a writer waits its turn + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + traffic.start(); + long start = System.currentTimeMillis(); + pacer.awaitTurn(); + traffic.join(1000); + assertTrue(System.currentTimeMillis() - start >= 110, + "the gap must restart from the mid-wait activity (40ms + 80ms)"); + } +}
