You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by sr...@apache.org on 2022/06/27 11:07:12 UTC
[plc4x] branch develop updated: fix(plc4j): Fix bug with select returning prematurely when device not sending any data (#386)
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 d8fecd3ac fix(plc4j): Fix bug with select returning prematurely when device not sending any data (#386)
d8fecd3ac is described below
commit d8fecd3ac5cc9e38764b2b4046472179d9a2259d
Author: vmpn <vm...@users.noreply.github.com>
AuthorDate: Mon Jun 27 07:07:07 2022 -0400
fix(plc4j): Fix bug with select returning prematurely when device not sending any data (#386)
* Add support for custom handlers used in testing
* Add support for custom handlers used in testing
* Add test to demonstrate problem with current select() implementation
* Change select() and wakeup() logic to match API expectations
* Don't use Java 11 features as PR check build fails
* Fix missing import
* Add missing headers
* Declare dependency used by the new test
---
plc4j/transports/serial/pom.xml | 6 +
.../plc4x/java/transport/serial/SerialChannel.java | 8 +-
.../transport/serial/SerialChannelHandler.java | 48 -------
.../transport/serial/SerialPollingSelector.java | 17 ++-
.../java/transport/serial/SerialSocketAddress.java | 18 +++
.../base/connection/SerialChannelFactoryTest.java | 5 +-
.../plc4x/java/transport/serial/DummyHandler.java} | 48 +++++--
.../serial/IdleSerialChannelHandler.java} | 40 ++++--
.../plc4x/java/transport/serial/SelectorTest.java | 156 +++++++++++++++++++++
9 files changed, 261 insertions(+), 85 deletions(-)
diff --git a/plc4j/transports/serial/pom.xml b/plc4j/transports/serial/pom.xml
index f096589fc..0822b7ca9 100644
--- a/plc4j/transports/serial/pom.xml
+++ b/plc4j/transports/serial/pom.xml
@@ -72,6 +72,12 @@
<version>0.10.0-SNAPSHOT</version>
<scope>compile</scope>
</dependency>
+
+ <dependency>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-core</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
<plugins>
diff --git a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialChannel.java b/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialChannel.java
index a39188901..9b626ece6 100644
--- a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialChannel.java
+++ b/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialChannel.java
@@ -38,6 +38,7 @@ import java.lang.reflect.Method;
import java.net.SocketAddress;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectableChannel;
+import java.util.Optional;
import java.util.concurrent.RejectedExecutionException;
public class SerialChannel extends AbstractNioByteChannel implements DuplexChannel {
@@ -158,9 +159,10 @@ public class SerialChannel extends AbstractNioByteChannel implements DuplexChann
logger.debug("Connecting to Socket Address '{}'", ((SerialSocketAddress) remoteAddress).getIdentifier());
try {
- // A bit hacky but to make a Test Connection start the String with TEST
- if (((SerialSocketAddress) remoteAddress).getIdentifier().startsWith("TEST")) {
- comPort = SerialChannelHandler.DummyHandler.INSTANCE;
+ // A bit hacky but to support testing check for custom handler
+ final Optional<SerialChannelHandler> customHandler = ((SerialSocketAddress) remoteAddress).getHandler();
+ if (customHandler.isPresent()) {
+ comPort = customHandler.get();
} else {
comPort = new SerialChannelHandler.SerialPortHandler(remoteAddress, config);
}
diff --git a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialChannelHandler.java b/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialChannelHandler.java
index 9171bb78a..438a15d09 100644
--- a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialChannelHandler.java
+++ b/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialChannelHandler.java
@@ -64,54 +64,6 @@ public abstract class SerialChannelHandler {
*/
public abstract int write(ByteBuf buf);
- public static class DummyHandler extends SerialChannelHandler {
-
- public static final DummyHandler INSTANCE = new DummyHandler(null);
-
- private SerialSelectionKey selectionKey;
-
- public DummyHandler(SocketAddress address) {
- super(address);
- }
-
- @Override
- public boolean open() {
- return true;
- }
-
- @Override
- public String getIdentifier() {
- return null;
- }
-
- @Override
- public void registerSelectionKey(SerialSelectionKey selectionKey) {
- this.selectionKey = selectionKey;
- }
-
- @Override
- public void close() {
- // NOOP
- }
-
- @Override
- public int read(ByteBuf buf) {
- buf.writeByte(1);
- return 1;
- }
-
- @Override
- public int write(ByteBuf buf) {
- System.out.println("Haha i wrote something");
- return 1;
- }
-
- public void fireEvent(int readyOp) {
- ((SerialPollingSelector) this.selectionKey.selector())
- .addEvent(new SerialPollingSelector.SelectorEvent(this.selectionKey, readyOp));
- }
- }
-
public static class SerialPortHandler extends SerialChannelHandler {
diff --git a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialPollingSelector.java b/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialPollingSelector.java
index 02e9044b8..f30030946 100644
--- a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialPollingSelector.java
+++ b/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialPollingSelector.java
@@ -100,23 +100,32 @@ class SerialPollingSelector extends AbstractSelector {
}
this.selectPromise = new DefaultPromise<>(executor);
try {
- selectPromise.await(timeout);
+ if(timeout == 0) {
+ selectPromise.await();
+ } else {
+ selectPromise.await(timeout);
+ }
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("Was interrupted", e);
}
- return events.size();
+ final int eventCount = events.size();
+ logger.debug("returning from select with {} events", eventCount);
+ return eventCount;
}
@Override
public int select() {
- return select(10);
+ return select(0);
}
@Override
public Selector wakeup() {
+ logger.debug("being asked to wake up from select");
// throw new NotImplementedException("Not implemented for this selector, should not be needed.");
- // NOOP
+ if (!selectPromise.isDone()) {
+ selectPromise.setSuccess(null);
+ }
return this;
}
diff --git a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialSocketAddress.java b/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialSocketAddress.java
index 08fe27689..50dbf9761 100644
--- a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialSocketAddress.java
+++ b/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialSocketAddress.java
@@ -20,13 +20,31 @@ package org.apache.plc4x.java.transport.serial;
import java.net.SocketAddress;
import java.util.Objects;
+import java.util.Optional;
public class SerialSocketAddress extends SocketAddress {
private final String identifier;
+ private final Optional<SerialChannelHandler> handler;
+
public SerialSocketAddress(String identifier) {
this.identifier = identifier;
+ this.handler = Optional.empty();
+ }
+
+ /**
+ * @param identifier of the port
+ * @param handler for custom behavior. E.g. testing
+ */
+ public SerialSocketAddress(final String identifier, final SerialChannelHandler handler)
+ {
+ this.identifier = identifier;
+ this.handler = Optional.of(handler);
+ }
+
+ public Optional<SerialChannelHandler> getHandler() {
+ return handler;
}
public String getIdentifier() {
diff --git a/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/base/connection/SerialChannelFactoryTest.java b/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/base/connection/SerialChannelFactoryTest.java
index 754cb60b4..f782d6cb5 100644
--- a/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/base/connection/SerialChannelFactoryTest.java
+++ b/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/base/connection/SerialChannelFactoryTest.java
@@ -25,6 +25,7 @@ import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.handler.codec.ByteToMessageCodec;
import org.apache.plc4x.java.api.exceptions.PlcConnectionException;
+import org.apache.plc4x.java.transport.serial.DummyHandler;
import org.apache.plc4x.java.transport.serial.SerialChannel;
import org.apache.plc4x.java.transport.serial.SerialChannelFactory;
import org.apache.plc4x.java.transport.serial.SerialChannelHandler;
@@ -53,7 +54,7 @@ public class SerialChannelFactoryTest {
@Test
public void createChannel() throws PlcConnectionException, InterruptedException, UnknownHostException {
- SerialChannelFactory asdf = new SerialChannelFactory(new SerialSocketAddress("TEST-port1"));
+ SerialChannelFactory asdf = new SerialChannelFactory(new SerialSocketAddress("TEST-port1", DummyHandler.INSTANCE));
// final TcpSocketChannelFactory factory = new TcpSocketChannelFactory(InetAddress.getLocalHost(), 5432);
final Channel channel = asdf.createChannel(new ChannelInitializer<SerialChannel>() {
@Override
@@ -64,7 +65,7 @@ public class SerialChannelFactoryTest {
Thread.sleep(100);
for (int i = 1; i <= 10; i++) {
Thread.sleep(10);
- SerialChannelHandler.DummyHandler.INSTANCE.fireEvent(1);
+ DummyHandler.INSTANCE.fireEvent(1);
}
Thread.sleep(100);
channel.close().sync();
diff --git a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialSocketAddress.java b/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/DummyHandler.java
similarity index 50%
copy from plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialSocketAddress.java
copy to plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/DummyHandler.java
index 08fe27689..a38cd0df3 100644
--- a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialSocketAddress.java
+++ b/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/DummyHandler.java
@@ -18,32 +18,54 @@
*/
package org.apache.plc4x.java.transport.serial;
+import io.netty.buffer.ByteBuf;
+
import java.net.SocketAddress;
-import java.util.Objects;
-public class SerialSocketAddress extends SocketAddress {
+public class DummyHandler extends SerialChannelHandler {
+
+ public static final DummyHandler INSTANCE = new DummyHandler(null);
- private final String identifier;
+ private SerialSelectionKey selectionKey;
- public SerialSocketAddress(String identifier) {
- this.identifier = identifier;
+ public DummyHandler(SocketAddress address) {
+ super(address);
}
+ @Override
+ public boolean open() {
+ return true;
+ }
+
+ @Override
public String getIdentifier() {
- return identifier;
+ return null;
}
@Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- SerialSocketAddress that = (SerialSocketAddress) o;
- return Objects.equals(identifier, that.identifier);
+ public void registerSelectionKey(SerialSelectionKey selectionKey) {
+ this.selectionKey = selectionKey;
}
@Override
- public int hashCode() {
- return Objects.hash(identifier);
+ public void close() {
+ // NOOP
}
+ @Override
+ public int read(ByteBuf buf) {
+ buf.writeByte(1);
+ return 1;
+ }
+
+ @Override
+ public int write(ByteBuf buf) {
+ System.out.println("Haha i wrote something");
+ return 1;
+ }
+
+ public void fireEvent(int readyOp) {
+ ((SerialPollingSelector) this.selectionKey.selector())
+ .addEvent(new SerialPollingSelector.SelectorEvent(this.selectionKey, readyOp));
+ }
}
diff --git a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialSocketAddress.java b/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/IdleSerialChannelHandler.java
similarity index 57%
copy from plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialSocketAddress.java
copy to plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/IdleSerialChannelHandler.java
index 08fe27689..8f337c6bc 100644
--- a/plc4j/transports/serial/src/main/java/org/apache/plc4x/java/transport/serial/SerialSocketAddress.java
+++ b/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/IdleSerialChannelHandler.java
@@ -18,32 +18,42 @@
*/
package org.apache.plc4x.java.transport.serial;
-import java.net.SocketAddress;
-import java.util.Objects;
+import io.netty.buffer.ByteBuf;
-public class SerialSocketAddress extends SocketAddress {
+class IdleSerialChannelHandler extends SerialChannelHandler {
- private final String identifier;
+ public IdleSerialChannelHandler() {
+ super(null);
+ }
+
+ @Override
+ public boolean open() {
+ return true;
+ }
- public SerialSocketAddress(String identifier) {
- this.identifier = identifier;
+ @Override
+ String getIdentifier() {
+ return "Emulated Com Port";
}
- public String getIdentifier() {
- return identifier;
+ @Override
+ void registerSelectionKey(SerialSelectionKey selectionKey) {
+ /*
+ * We can ignore registration as we will neven trigger events
+ */
}
@Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- SerialSocketAddress that = (SerialSocketAddress) o;
- return Objects.equals(identifier, that.identifier);
+ public void close() {
}
@Override
- public int hashCode() {
- return Objects.hash(identifier);
+ public int read(ByteBuf buf) {
+ return 0;
}
+ @Override
+ public int write(ByteBuf buf) {
+ throw new UnsupportedOperationException();
+ }
}
diff --git a/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/SelectorTest.java b/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/SelectorTest.java
new file mode 100644
index 000000000..fa56f256a
--- /dev/null
+++ b/plc4j/transports/serial/src/test/java/org/apache/plc4x/java/transport/serial/SelectorTest.java
@@ -0,0 +1,156 @@
+/*
+ * 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 ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.nio.NioEventLoop;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class SelectorTest {
+
+ /*
+ * WARNING: This test is fragile as it tied to logging output behavior and format.
+ * However, I could not find my reliable hook points, and it was better than not having testing at all
+ */
+ @Disabled("This test is best executed manually as necessary due to fragility and to avoid increasing build times")
+ @Test
+ public void idleDeviceTest() throws Exception {
+ final var maxTestRunTime = Duration.ofSeconds(30);
+
+ /*
+ * Configure loggers so test's logic can determine if we have done enough loops
+ * and whether netty issues warnings that should not happen. E.g.
+ */
+ final var listAppender = new ListAppender<ILoggingEvent>();
+ {
+ final var loopLogger = LoggerFactory.getLogger(NioEventLoop.class);
+ final var selectorLogger = LoggerFactory.getLogger(SerialPollingSelector.class);
+
+ listAppender.start();
+
+ /*
+ * Enable debug mode of the loggers we need and grab copy of the output for test logic
+ */
+ Arrays.asList(loopLogger, selectorLogger)
+ .stream()
+ .map(Logger.class::cast)
+ .forEach(l -> {
+ l.setLevel(Level.DEBUG);
+ l.addAppender(listAppender);
+ });
+ ;
+ }
+
+
+ /*
+ * Setup IO loop that will check for read events to come in
+ */
+ final EventLoopGroup evenLoopGroup;
+ {
+ final var bootstrap = new Bootstrap();
+ final var commPortAddress = new SerialSocketAddress("test", new IdleSerialChannelHandler());
+ final var channelFactory = new SerialChannelFactory(commPortAddress);
+ evenLoopGroup = channelFactory.getEventLoopGroup();
+
+ bootstrap
+ .group(evenLoopGroup)
+ .channel(SerialChannel.class)
+ .handler(new SimpleChannelInboundHandler<ByteBuf>() {
+ @Override
+ protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
+ throw new UnsupportedOperationException();
+ }
+ })
+ .remoteAddress(commPortAddress);
+
+ /*
+ * Start the processing and wait for connection to be established
+ */
+ bootstrap.connect().sync();
+ }
+
+ /*
+ * Wait until 2k or more select loops happen or maximum time passes
+ */
+ final var start = Instant.now();
+
+ /*
+ * With current netty select has to return more than 512 time early to trigger warning so entry number was
+ * chosen with a little safety buffer.
+ */
+ while (listAppender.list.size() < 600
+ && Duration.between(start, Instant.now()).compareTo(maxTestRunTime) <= 0)
+ {
+ Thread.sleep(Duration.ofSeconds(1).toMillis());
+ }
+
+ /*
+ * Stop netty
+ */
+ evenLoopGroup.shutdownGracefully(0, 4, TimeUnit.SECONDS).sync();
+
+ /*
+ * Check results.
+ * Line: ....SerialPollingSelector - returning from select with 0 events"
+ * means one completed call to select ideally
+ *
+ * Line ....NioEventLoop - Selector.select() returned prematurely 512 times in a row; rebuilding Selector ...
+ * means the netty warning got tripped
+ */
+
+ final var selectorWarnings =
+ listAppender.list.stream()
+ .filter(log -> log.getFormattedMessage().contains("Selector.select() returned prematurely"))
+ .collect(Collectors.toList());
+
+ Assertions.assertEquals(0, selectorWarnings.size(), "Detected selector warnings");
+
+ /*
+ * Second check in case above breaks, given fragility of depending on logs
+ */
+ final var selectorExecutions =
+ listAppender.list.stream()
+ .filter(log -> log.getFormattedMessage().contains("returning from select"))
+ .collect(Collectors.toList());
+
+ final int maxSelectorCalls = 1;
+
+ Assertions.assertTrue(selectorExecutions.size() <= maxSelectorCalls,
+ String.format("Selector called less than %d times", maxSelectorCalls));
+ }
+
+}