Modbus TCP bridge skip cycle function

io.openems.edge.bridge.modbus/readme.adoc

@@ -28,6 +28,16 @@ Read-Tasks can have two different priorities, that are defined in the ModbusProt
 - `LOW`: only one task of all defined LOW priority tasks of all components registered on the same bridge is executed per Cycle
 Write-Tasks always have `HIGH` priority, i.e. a set-point is always executed as-soon-as-possible - as long as the Component is not marked as defective
 
+=== Access interval
+
+OpenEMS's core cycle time is usually 1 second, but some modbus devices can be very slow. It's possible to lower the core cycle time but the whole system is slowed down.
+
+The parameter `intervalBetweenAccesses` can be configured to slow down the modbus bridge's access. Specifically, it defines the minimum time (in milliseconds) between 2 modbus accesses, which normally equals to the core cycle time (usually 1000ms).
+
+`intervalBetweenAccesses` is aligned to and calculated from the core cycle time. It's set to be 0 by default and any value between 0 and core cycle time will not affect the original modbus bridge logic. The right amount of core cycles to be skipped will be calculated and applied when `intervalBetweenAccesses` is set to be longer than core cycle time.
+
+Note that the real interval may be longer than configured due to the round off of core cycle time. For example, we set `intervalBetweenAccesses` to 1.5 seconds but core time is 1 second and in this case the final interval is 2 seconds.
+
 === Channels
 
 Each Modbus Bridge provides Channels for more detailed information:

io.openems.edge.bridge.modbus/src/io/openems/edge/bridge/modbus/BridgeModbusTcpImpl.java

@@ -2,13 +2,18 @@
 
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.util.Objects;
 
+import io.openems.edge.bridge.modbus.api.LogVerbosity;
+import org.osgi.service.cm.ConfigurationAdmin;
 import org.osgi.service.component.ComponentContext;
 import org.osgi.service.component.annotations.Activate;
 import org.osgi.service.component.annotations.Component;
 import org.osgi.service.component.annotations.ConfigurationPolicy;
 import org.osgi.service.component.annotations.Deactivate;
 import org.osgi.service.component.annotations.Modified;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.event.Event;
 import org.osgi.service.event.EventHandler;
 import org.osgi.service.event.propertytypes.EventTopics;
 import org.osgi.service.metatype.annotations.Designate;
@@ -24,6 +29,8 @@
 import io.openems.edge.bridge.modbus.api.BridgeModbusTcp;
 import io.openems.edge.common.component.OpenemsComponent;
 import io.openems.edge.common.event.EdgeEventConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Provides a service for connecting to, querying and writing to a Modbus/TCP
@@ -45,6 +52,13 @@ public class BridgeModbusTcpImpl extends AbstractModbusBridge
 	/** The configured IP address. */
 	private InetAddress ipAddress = null;
 	private int port;
+	private final Logger log = LoggerFactory.getLogger(BridgeModbusTcpImpl.class);
+	private boolean shouldSkip = false;
+	private int noSkipIdx = 0;
+	private long cycleIdx = 0;
+
+	@Reference
+	protected ConfigurationAdmin cm;
 
 	public BridgeModbusTcpImpl() {
 		super(//
@@ -72,6 +86,18 @@ private void modified(ComponentContext context, ConfigTcp config) throws Unknown
 	private void applyConfig(ConfigTcp config) {
 		this.setIpAddress(InetAddressUtils.parseOrNull(config.ip()));
 		this.port = config.port();
+        int coreCycleTime = 1000;
+        try {
+			// the default cycleTime is not persisted and won't be returned, only modified cycleTime will!
+			coreCycleTime = (int) (Integer) this.cm.getConfiguration("Core.Cycle").getProperties().get("cycleTime");
+        } catch (Exception e) {
+			this.logCycle("cycleTime reading failed, use default value " + coreCycleTime);
+        }
+        this.noSkipIdx = (int)Math.ceil(config.intervalBetweenAccesses() * 1.0 / coreCycleTime);
+		this.noSkipIdx = Math.max(this.noSkipIdx, 1);
+		this.logCycle("applyConfig: cycleTime=" + coreCycleTime
+			+ ", interval=" + config.intervalBetweenAccesses()
+			+ ", noSkipIdx=" + this.noSkipIdx);
 	}
 
 	@Override
@@ -127,4 +153,39 @@ public InetAddress getIpAddress() {
 	public void setIpAddress(InetAddress ipAddress) {
 		this.ipAddress = ipAddress;
 	}
+
+	private boolean isNewCycle(Event event) {
+		return Objects.equals(event.getTopic(), EdgeEventConstants.TOPIC_CYCLE_BEFORE_PROCESS_IMAGE);
+	}
+
+	private void logCycle(String msg) {
+		if (this.getLogVerbosity() == LogVerbosity.DEBUG_LOG) {
+			this.logInfo(this.log, msg);
+		}
+	}
+
+	@Override
+	public void handleEvent(Event event) {
+		if (!this.isEnabled()) {
+			return;
+		}
+
+		if (this.isNewCycle(event)) {
+			this.shouldSkip = this.cycleIdx % this.noSkipIdx != 0;
+			this.logCycle("handleEvent: Cycle " + this.cycleIdx + (this.shouldSkip ? " will" : " won't") + " be skipped");
+			this.cycleIdx++;
+		}
+		if (this.shouldSkip) {
+			return;
+		}
+
+		switch (event.getTopic()) {
+			case EdgeEventConstants.TOPIC_CYCLE_BEFORE_PROCESS_IMAGE:
+				this.worker.onBeforeProcessImage();
+				break;
+			case EdgeEventConstants.TOPIC_CYCLE_EXECUTE_WRITE:
+				this.worker.onExecuteWrite();
+				break;
+		}
+	}
 }

io.openems.edge.bridge.modbus/src/io/openems/edge/bridge/modbus/ConfigTcp.java

@@ -33,5 +33,8 @@
 	@AttributeDefinition(name = "Invalidate elements after how many read Errors?", description = "Increase this value if modbus read errors happen frequently.")
 	int invalidateElementsAfterReadErrors() default 1;
 
+	@AttributeDefinition(name = "Interval between accesses (in milliseconds)", description = "Cycle skipping will be calculated from this value and core cycle time.")
+	int intervalBetweenAccesses() default 0;
+
 	String webconsole_configurationFactory_nameHint() default "Bridge Modbus/TCP [{id}]";
 }

io.openems.edge.bridge.modbus/test/io/openems/edge/bridge/modbus/BridgeModbusTcpImplTest.java

@@ -1,5 +1,6 @@
 package io.openems.edge.bridge.modbus;
 
+import io.openems.edge.common.test.DummyConfigurationAdmin;
 import org.junit.Ignore;
 import org.junit.Test;
 
@@ -25,6 +26,9 @@
 import io.openems.edge.common.test.DummyCycle;
 import io.openems.edge.common.test.TestUtils;
 
+import java.util.Dictionary;
+import java.util.Hashtable;
+
 public class BridgeModbusTcpImplTest {
 
 	private static final String MODBUS_ID = "modbus0";
@@ -113,6 +117,127 @@ public void test() throws Exception {
 		}
 	}
 
+	@Test
+	public void testSkipInterval() throws Exception {
+		final ThrowingRunnable<Exception> sleep = () -> Thread.sleep(CYCLE_TIME);
+
+		var port = TestUtils.findRandomOpenPortOnAllLocalInterfaces();
+		ModbusSlave slave = null;
+		try {
+			/*
+			 * Open Modbus/TCP Slave
+			 */
+			slave = ModbusSlaveFactory.createTCPSlave(port, 1);
+			var processImage = new SimpleProcessImage(UNIT_ID);
+			slave.addProcessImage(UNIT_ID, processImage);
+			slave.open();
+
+			var cm = new DummyConfigurationAdmin();
+			var cfg = cm.createFactoryConfiguration("Core.Cycle", null);
+			Dictionary<String, Object> properties = new Hashtable<>();
+			properties.put("cycleTime", 100);
+			cfg.update(properties);
+
+			// interval = 0, should not change original modbus behavior
+			int numTests = 1;
+			for (int i = 0; i < numTests; i++) {
+				var sut = new BridgeModbusTcpImpl();
+				var device = new MyModbusComponent(DEVICE_ID, sut, UNIT_ID);
+				var test = new ComponentTest(sut) //
+						.addComponent(device) //
+						.addReference("cm", cm);
+
+				test.activate(MyConfigTcp.create() //
+						.setId(MODBUS_ID) //
+						.setIp("127.0.0.1") //
+						.setPort(port) //
+						.setInvalidateElementsAfterReadErrors(1) //
+						.setLogVerbosity(LogVerbosity.DEBUG_LOG) //
+						.setIntervalBetweenAccesses(0)
+						.build());
+
+				processImage.addRegister(100, new SimpleRegister(11));
+				test.next(new TestCase() //
+						.onAfterProcessImage(sleep) //
+						.output(REGISTER_100, 11) //
+						.output(MODBUS_COMMUNICATION_FAILED, false)); //
+
+				processImage.addRegister(100, new SimpleRegister(22));
+				test.next(new TestCase() //
+						.onAfterProcessImage(sleep) //
+						.output(REGISTER_100, 22) //
+						.output(MODBUS_COMMUNICATION_FAILED, false)); //
+				test.next(new TestCase() //
+						.onAfterProcessImage(sleep) //
+						.output(REGISTER_100, 22) //
+						.output(MODBUS_COMMUNICATION_FAILED, false)); //
+
+				// Important! Otherwise, new sut cannot connect to the slave (only 1 slave thread)
+				sut.deactivate();
+			}
+			// 0 <= interval < maxInterval
+			numTests = 7;
+			int maxInterval = CYCLE_TIME * 3;
+			for (int i = 0; i < numTests; i++) {
+				var sut = new BridgeModbusTcpImpl();
+				var device = new MyModbusComponent(DEVICE_ID, sut, UNIT_ID);
+				var test = new ComponentTest(sut) //
+						.addComponent(device)
+						.addReference("cm", cm);
+
+				int interval = maxInterval * i / numTests;
+				int skips = (int)Math.ceil(interval * 1.0 / CYCLE_TIME) - 1;
+
+				System.out.println("Interval=" + interval + ", skips=" + skips);
+
+				test.activate(MyConfigTcp.create()
+						.setId(MODBUS_ID)
+						.setIp("127.0.0.1")
+						.setPort(port)
+						.setInvalidateElementsAfterReadErrors(1)
+						.setLogVerbosity(LogVerbosity.DEBUG_LOG)
+						.setIntervalBetweenAccesses(interval)
+						.build());
+
+				processImage.addRegister(100, new SimpleRegister(111));
+				test.next(new TestCase()
+						.onAfterProcessImage(sleep)
+						.output(REGISTER_100, 111)
+						.output(MODBUS_COMMUNICATION_FAILED, false));
+
+				processImage.addRegister(100, new SimpleRegister(222));
+				for (int j = 0; j < skips; j++) {
+					test.next(new TestCase()
+							.onAfterProcessImage(sleep)
+							.output(REGISTER_100, 111)
+							.output(MODBUS_COMMUNICATION_FAILED, false));
+				}
+				test.next(new TestCase()
+						.onAfterProcessImage(sleep)
+						.output(REGISTER_100, 222)
+						.output(MODBUS_COMMUNICATION_FAILED, false));
+
+				processImage.addRegister(100, new SimpleRegister(333));
+				for (int j = 0; j < skips; j++) {
+					test.next(new TestCase()
+							.onAfterProcessImage(sleep)
+							.output(REGISTER_100, 222)
+							.output(MODBUS_COMMUNICATION_FAILED, false));
+				}
+				test.next(new TestCase()
+						.onAfterProcessImage(sleep)
+						.output(REGISTER_100, 333)
+						.output(MODBUS_COMMUNICATION_FAILED, false));
+
+				sut.deactivate();
+			}
+		} finally {
+			if (slave != null) {
+				slave.close();
+			}
+		}
+	}
+
 	private static class MyModbusComponent extends DummyModbusComponent {
 
 		public MyModbusComponent(String id, AbstractModbusBridge bridge, int unitId) throws OpenemsException {

io.openems.edge.bridge.modbus/test/io/openems/edge/bridge/modbus/MyConfigTcp.java

@@ -12,6 +12,7 @@ protected static class Builder {
 		private int port;
 		private LogVerbosity logVerbosity;
 		private int invalidateElementsAfterReadErrors;
+		private int intervalBetweenAccesses;
 
 		private Builder() {
 		}
@@ -41,6 +42,11 @@ public Builder setInvalidateElementsAfterReadErrors(int invalidateElementsAfterR
 			return this;
 		}
 
+		public Builder setIntervalBetweenAccesses(int intervalBetweenAccesses) {
+			this.intervalBetweenAccesses = intervalBetweenAccesses;
+			return this;
+		}
+
 		public MyConfigTcp build() {
 			return new MyConfigTcp(this);
 		}
@@ -82,4 +88,9 @@ public int invalidateElementsAfterReadErrors() {
 		return this.builder.invalidateElementsAfterReadErrors;
 	}
 
+	@Override
+	public int intervalBetweenAccesses() {
+		return this.builder.intervalBetweenAccesses;
+	}
+
 }