You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by cd...@apache.org on 2025/03/11 21:58:00 UTC
(plc4x) 02/04: feat: Implemented a feature to automatically split large chunks into multiple requests and to merge them back together.
This is an automated email from the ASF dual-hosted git repository.
cdutz pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit 06bdc37020267d0cb82fbbe79dc5cc78f5ebb7ee
Author: Christofer Dutz <ch...@c-ware.de>
AuthorDate: Tue Mar 11 22:57:10 2025 +0100
feat: Implemented a feature to automatically split large chunks into multiple requests and to merge them back together.
---
.../readwrite/optimizer/S7BlockReadOptimizer.java | 225 ++++++++++++++++++---
.../ManualS71200DriverWithBlockOptimizerTest.java | 8 +-
2 files changed, 196 insertions(+), 37 deletions(-)
diff --git a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/optimizer/S7BlockReadOptimizer.java b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/optimizer/S7BlockReadOptimizer.java
index 24ef189d71..564754f6b0 100644
--- a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/optimizer/S7BlockReadOptimizer.java
+++ b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/optimizer/S7BlockReadOptimizer.java
@@ -34,6 +34,7 @@ import org.apache.plc4x.java.spi.messages.DefaultPlcReadRequest;
import org.apache.plc4x.java.spi.messages.DefaultPlcReadResponse;
import org.apache.plc4x.java.spi.messages.utils.*;
import org.apache.plc4x.java.spi.values.DefaultPlcValueHandler;
+import org.apache.plc4x.java.spi.values.PlcNull;
import org.apache.plc4x.java.spi.values.PlcRawByteArray;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -65,19 +66,19 @@ public class S7BlockReadOptimizer extends S7Optimizer {
// Sort the tags by area
// (We can only read multiple tags in one byte array, if they are located in the same area)
- Map<String, Map<PlcTag, String>> sortedTags = new HashMap<>();
+ Map<String, Map<PlcTag, String>> sortedTagsPerArea = new HashMap<>();
for (String tagName : readRequest.getTagNames()) {
PlcTag tag = readRequest.getTag(tagName);
if(tag instanceof S7SzlTag) {
- if(!sortedTags.containsKey("SZL")) {
- sortedTags.put("SZL", createTagMap());
+ if(!sortedTagsPerArea.containsKey("SZL")) {
+ sortedTagsPerArea.put("SZL", createSortedTagMap());
}
- sortedTags.get("SZL").put(tag, tagName);
+ sortedTagsPerArea.get("SZL").put(tag, tagName);
} else if(tag instanceof S7ClkTag) {
- if(!sortedTags.containsKey("CLK")) {
- sortedTags.put("CLK", createTagMap());
+ if(!sortedTagsPerArea.containsKey("CLK")) {
+ sortedTagsPerArea.put("CLK", createSortedTagMap());
}
- sortedTags.get("CLK").put(tag, tagName);
+ sortedTagsPerArea.get("CLK").put(tag, tagName);
} else if(tag instanceof S7Tag) {
S7Tag s7Tag = (S7Tag) tag;
MemoryArea memoryArea = s7Tag.getMemoryArea();
@@ -88,10 +89,10 @@ public class S7BlockReadOptimizer extends S7Optimizer {
} else if(memoryArea == MemoryArea.INSTANCE_DATA_BLOCKS) {
areaName += s7Tag.getBlockNumber();
}
- if(!sortedTags.containsKey(areaName)) {
- sortedTags.put(areaName, createTagMap());
+ if(!sortedTagsPerArea.containsKey(areaName)) {
+ sortedTagsPerArea.put(areaName, createSortedTagMap());
}
- sortedTags.get(areaName).put(tag, tagName);
+ sortedTagsPerArea.get(areaName).put(tag, tagName);
} else {
System.out.println("Ignored");
}
@@ -102,24 +103,29 @@ public class S7BlockReadOptimizer extends S7Optimizer {
// - If two consecutive tags are more than S7_ADDRESS_ANY_SIZE bytes apart,
// reading them in one block is less efficient than reading them separately.
// Other than that, extend the byte array to include the next tag.
- LinkedHashMap<String, PlcTagItem<PlcTag>> optimizedTagMap = new LinkedHashMap<>();
- for (Map<PlcTag, String> tagList : sortedTags.values()) {
+ Map<TagNameSize, PlcTagItem<PlcTag>> optimizedTagMap = new TreeMap<>();
+ for (Map<PlcTag, String> tagList : sortedTagsPerArea.values()) {
MemoryArea currentMemoryArea = null;
int currentDataBlockNumber = -1;
int currentChunkStartByteOffset = -1;
int currentChunkEndByteOffset = -1;
- Map<PlcTag, String> currentChunkTags = createTagMap();
+ Map<PlcTag, String> currentChunkTags = createSortedTagMap();
for (PlcTag plcTag : tagList.keySet()) {
// We don't do anything for Szl and Clk tags.
if(plcTag instanceof S7SzlTag) {
- optimizedTagMap.put(tagList.get(plcTag), new DefaultPlcTagItem<>(plcTag));
+ // TODO: Implement the size
+ optimizedTagMap.put(new TagNameSize(tagList.get(plcTag), 0), new DefaultPlcTagItem<>(plcTag));
} else if (plcTag instanceof S7ClkTag) {
- optimizedTagMap.put(tagList.get(plcTag), new DefaultPlcTagItem<>(plcTag));
+ // TODO: Implement the size
+ optimizedTagMap.put(new TagNameSize(tagList.get(plcTag), 0), new DefaultPlcTagItem<>(plcTag));
}
// Var-length strings, are a performance nightmare. Trying to optimize reading them is probably not
// worth the effort. For now, we simply handle them as un-chunked tags.
else if(plcTag instanceof S7StringVarLengthTag) {
- optimizedTagMap.put(tagList.get(plcTag), new DefaultPlcTagItem<>(plcTag));
+ // A var-length string tag simply reads 2 or 4 bytes.
+ optimizedTagMap.put(new TagNameSize(tagList.get(plcTag),
+ ((S7StringVarLengthTag) plcTag).getDataType() == TransportSize.STRING ? 2 : 4),
+ new DefaultPlcTagItem<>(plcTag));
}
// Only regular tags are optimized.
@@ -132,7 +138,7 @@ public class S7BlockReadOptimizer extends S7Optimizer {
if(s7Tag instanceof S7StringFixedLengthTag) {
S7StringFixedLengthTag stringFixedLengthTag = (S7StringFixedLengthTag) s7Tag;
int bytesPerChar = (stringFixedLengthTag.getDataType() == TransportSize.WSTRING) ? 2 : 1;
- curTagSize = (2 + (stringFixedLengthTag.getStringLength() * bytesPerChar)) * s7Tag.getNumberOfElements();
+ curTagSize = ((2 * bytesPerChar) + (stringFixedLengthTag.getStringLength() * bytesPerChar)) * s7Tag.getNumberOfElements();
}
// If this is the first tag, use that as starting point.
@@ -146,17 +152,17 @@ public class S7BlockReadOptimizer extends S7Optimizer {
// split up into multiple items.
else if(currentChunkEndByteOffset + S7_ADDRESS_ANY_SIZE < s7Tag.getByteOffset()) {
// Save the current chunk.
- optimizedTagMap.put("__chunk__" + optimizedTagMap.size(),
+ optimizedTagMap.put(new TagNameSize("__chunk__" + optimizedTagMap.size(), currentChunkEndByteOffset - currentChunkStartByteOffset),
new DefaultPlcTagItem<>(
new S7TagChunk(TransportSize.BYTE, currentMemoryArea, currentDataBlockNumber,
currentChunkStartByteOffset, (byte) 0,
currentChunkEndByteOffset - currentChunkStartByteOffset,
- currentChunkTags)));
+ currentChunkTags, 0, 1, currentChunkEndByteOffset - currentChunkStartByteOffset)));
// Start a new one.
currentChunkStartByteOffset = s7Tag.getByteOffset();
currentChunkEndByteOffset = s7Tag.getByteOffset() + curTagSize;
- currentChunkTags = createTagMap();
+ currentChunkTags = createSortedTagMap();
}
// Otherwise extend the array size to include this tag.
else {
@@ -167,18 +173,74 @@ public class S7BlockReadOptimizer extends S7Optimizer {
currentChunkTags.put(s7Tag, tagList.get(s7Tag));
}
}
- // Add the new tag.
- // Save the current chunk.
- optimizedTagMap.put("__chunk__" + optimizedTagMap.size(),
+
+ // Finish the last chunk.
+ optimizedTagMap.put(new TagNameSize("__chunk__" + optimizedTagMap.size(), currentChunkEndByteOffset - currentChunkStartByteOffset),
new DefaultPlcTagItem<>(
new S7TagChunk(TransportSize.BYTE, currentMemoryArea, currentDataBlockNumber,
currentChunkStartByteOffset, (byte) 0,
currentChunkEndByteOffset - currentChunkStartByteOffset,
- currentChunkTags)));
+ currentChunkTags, 0, 1, currentChunkEndByteOffset - currentChunkStartByteOffset)));
+ }
+
+ // Go through all chunks. If there are ones larger than the max PDU size, split them up into
+ // multiple tags, that utilize the packets to the maximum.
+ final int maxRequestSize = ((S7DriverContext) driverContext).getPduSize() - (EMPTY_READ_RESPONSE_SIZE + 4);
+ Map<TagNameSize, PlcTagItem<PlcTag>> optimizedTagMap2 = new TreeMap<>();
+ for (TagNameSize tagNameSize : optimizedTagMap.keySet()) {
+ PlcTagItem<PlcTag> curTagItem = optimizedTagMap.get(tagNameSize);
+ S7Tag curTag = (S7Tag) curTagItem.getTag();
+ if(tagNameSize.getTagSize() > maxRequestSize) {
+ String curTagNameBase = tagNameSize.getTagName();
+ int curTagFragmentNumber = 0;
+ int curTagSize = tagNameSize.getTagSize();
+ int curTagOffset = curTag.getByteOffset();
+ int curPartIndex = 0;
+ int totalPartCount = (curTagSize / maxRequestSize) + 1;
+ while(curTagSize > maxRequestSize) {
+ optimizedTagMap2.put(new TagNameSize(curTagNameBase + "." + curTagFragmentNumber, maxRequestSize),
+ new DefaultPlcTagItem<>(
+ new S7TagChunk(curTag.getDataType(), curTag.getMemoryArea(), curTag.getBlockNumber(), curTagOffset, (byte) 0, maxRequestSize,
+ (curTag instanceof S7TagChunk) ? ((S7TagChunk) curTag).getChunkTags() : Collections.singletonMap(curTag, tagNameSize.getTagName()),
+ curPartIndex, totalPartCount, curTagSize)));
+
+ curTagOffset += maxRequestSize;
+ curTagSize -= maxRequestSize;
+ curTagFragmentNumber++;
+ curPartIndex++;
+ }
+ optimizedTagMap2.put(new TagNameSize(curTagNameBase + "." + curTagFragmentNumber, curTagSize),
+ new DefaultPlcTagItem<>(
+ new S7TagChunk(curTag.getDataType(), curTag.getMemoryArea(), curTag.getBlockNumber(), curTagOffset, (byte) 0, curTagSize,
+ (curTag instanceof S7TagChunk) ? ((S7TagChunk) curTag).getChunkTags() : Collections.singletonMap(curTag, tagNameSize.getTagName()),
+ curPartIndex, totalPartCount, curTagSize)));
+ }
+ // Just copy over tags that fit into a request.
+ else {
+ optimizedTagMap2.put(tagNameSize, optimizedTagMap.get(tagNameSize));
+ }
+ }
+
+ // Using the First Fit Decreasing (FFD) bin-packing algorithm try to find the ideal
+ // packing for utilizing request sizes.
+ // 1. Assign a size to each tag
+ // 2. Sort the tags by size (biggest first)
+ // 3. Repeat this, till all tags are consumed
+ // 1. Take the first packet of the list
+ // 2. If the tag itself exceeds the max request size, keep on splitting it into chunks until
+ // the rest would fit into a request. Then proceed with the rest as if it was a normal tag
+ // 2. Go through the existing list of requests and check if the current tag would fit
+ // 1. If it fits, add it to the request
+ // 2. If it doesn't fit go to the next request and check
+ // 3. If you reach the end, and it didn't fit any of the previous requests, add a new one
+ LinkedHashMap<String, PlcTagItem<PlcTag>> executableTagMap = new LinkedHashMap<>();
+ for (TagNameSize tagNameSize : optimizedTagMap2.keySet()) {
+ // TODO: Implement the algorithm above.
+ executableTagMap.put(tagNameSize.getTagName(), optimizedTagMap2.get(tagNameSize));
}
return super.processReadRequest(
- new DefaultPlcReadRequest(((DefaultPlcReadRequest) readRequest).getReader(), optimizedTagMap),
+ new DefaultPlcReadRequest(((DefaultPlcReadRequest) readRequest).getReader(), executableTagMap),
driverContext);
}
@@ -198,16 +260,68 @@ public class S7BlockReadOptimizer extends S7Optimizer {
}
// Have the upstream optimizer handle its thing.
- PlcReadResponse readResponse = super.processReadResponses(new DefaultPlcReadRequest(((DefaultPlcReadRequest) readRequest).getReader(), tags), readResponses, driverContext);
+ PlcReadResponse rawReadResponse = super.processReadResponses(new DefaultPlcReadRequest(((DefaultPlcReadRequest) readRequest).getReader(), tags), readResponses, driverContext);
+
+ // Merge together split-up chunks.
+ LinkedHashMap<String, PlcTagItem<PlcTag>> mergedTagItems = new LinkedHashMap<>();
+ Map<String, PlcResponseItem<PlcValue>> mergedValues = new LinkedHashMap<>();
+ for (String tagName : rawReadResponse.getTagNames()) {
+ PlcTag tag = rawReadResponse.getRequest().getTag(tagName);
+
+ // This indicates it's a chunk that's been split up.
+ if((tag instanceof S7TagChunk) && (((S7TagChunk) tag).getTotalPartNumber() > 1)){
+ S7TagChunk tagChunk = (S7TagChunk) tag;
+ // Only handle the first part
+ if(tagChunk.getCurPartIndex() == 0) {
+ String tagBaseName = tagName.substring(0, tagName.indexOf("."));
+
+ // Allocate a chunk of memory able to store the data of all parts.
+ byte[] chunkData = new byte[tagChunk.getTotalPartSize()];
+
+ int firstPartOffset = tagChunk.getByteOffset();
+ // Copy the payload of each chunk into the total chunk data.
+ for(int curPartIndex = 0; curPartIndex < tagChunk.getTotalPartNumber(); curPartIndex++) {
+ String curPartTagName = tagBaseName + "." + curPartIndex;
+ S7TagChunk curPartTag = (S7TagChunk) rawReadResponse.getRequest().getTag(curPartTagName);
+
+ PlcResponseCode responseCode = rawReadResponse.getResponseCode(curPartTagName);
+ // If one of the parts is not OK, all in this chunk will be not OK too.
+ if(responseCode != PlcResponseCode.OK) {
+ mergedValues.put(tagName, new DefaultPlcResponseItem<>(responseCode, new PlcNull()));
+ break;
+ }
+ PlcValue plcValue = rawReadResponse.getPlcValue(curPartTagName);
+ byte[] chunkPartData = plcValue.getRaw();
+ int partDataOffset = curPartTag.getByteOffset() - firstPartOffset;
+ System.arraycopy(chunkPartData, 0, chunkData, partDataOffset, chunkPartData.length);
+ }
+
+ // Store the merged tag.
+ mergedTagItems.put(tagBaseName, new DefaultPlcTagItem<>(
+ new S7TagChunk(tagChunk.getDataType(), tagChunk.getMemoryArea(), tagChunk.getBlockNumber(),
+ firstPartOffset, (byte) 0, tagChunk.getTotalPartSize(), tagChunk.getChunkTags(),
+ 0, 1, tagChunk.getTotalPartSize())));
+ mergedValues.put(tagBaseName, new DefaultPlcResponseItem<>(PlcResponseCode.OK, new PlcRawByteArray(chunkData)));
+ }
+ }
+ // All others are just un-split chunks
+ else {
+ PlcResponseCode responseCode = rawReadResponse.getResponseCode(tagName);
+ PlcValue plcValue = rawReadResponse.getPlcValue(tagName);
+ mergedTagItems.put(tagName, new DefaultPlcTagItem<>(tag));
+ mergedValues.put(tagName, new DefaultPlcResponseItem<>(responseCode, plcValue));
+ }
+ }
+ PlcReadResponse mergedReadResponse = new DefaultPlcReadResponse(new DefaultPlcReadRequest(((DefaultPlcReadRequest)rawReadResponse.getRequest()).getReader(), mergedTagItems), mergedValues);
// If a Tag is a normal tag, just copy it over. However, if it's a S7TagChunk, process it.
Map<String, PlcResponseItem<PlcValue>> values = new HashMap<>();
- for (String tagName : readResponse.getTagNames()) {
- PlcResponseCode responseCode = readResponse.getResponseCode(tagName);
- PlcValue plcValue = readResponse.getPlcValue(tagName);
+ for (String tagName : mergedReadResponse.getTagNames()) {
+ PlcResponseCode responseCode = mergedReadResponse.getResponseCode(tagName);
+ PlcValue plcValue = mergedReadResponse.getPlcValue(tagName);
// We need to get the request from the response as only this contains the
// chunk tags and not the original one passed in via readRequest parameter.
- PlcTag tag = readResponse.getRequest().getTag(tagName);
+ PlcTag tag = mergedReadResponse.getRequest().getTag(tagName);
// If it's not a tag-chunk, then we just add it to the response.
if(!(tag instanceof S7TagChunk)) {
@@ -244,7 +358,7 @@ public class S7BlockReadOptimizer extends S7Optimizer {
return new DefaultPlcReadResponse(readRequest, values);
}
- protected Map<PlcTag, String> createTagMap() {
+ protected Map<PlcTag, String> createSortedTagMap() {
return new TreeMap<>((tag1, tag2) -> {
if (tag1 instanceof S7SzlTag) {
S7SzlTag s7Tag1 = (S7SzlTag) tag1;
@@ -274,7 +388,6 @@ public class S7BlockReadOptimizer extends S7Optimizer {
try {
int stringLength = (tag instanceof S7StringFixedLengthTag) ? ((S7StringFixedLengthTag) tag).getStringLength() : 254;
if (tag.getNumberOfElements() == 1) {
- // TODO: Pass the type of plc into the parse function ...
return DataItem.staticParse(readBuffer, tag.getDataType().getDataProtocolId(),
s7DriverContext.getControllerType(), stringLength);
} else {
@@ -305,15 +418,61 @@ public class S7BlockReadOptimizer extends S7Optimizer {
public static class S7TagChunk extends S7Tag {
private final Map<PlcTag, String> chunkTags;
+ private final int curPartIndex;
+ private final int totalPartNumber;
+ private final int totalPartSize;
- public S7TagChunk(TransportSize dataType, MemoryArea memoryArea, int blockNumber, int byteOffset, byte bitOffset, int numElements, Map<PlcTag, String> chunkTags) {
+ public S7TagChunk(TransportSize dataType, MemoryArea memoryArea, int blockNumber, int byteOffset, byte bitOffset, int numElements, Map<PlcTag, String> chunkTags, int curPartIndex, int totalPartNumber, int totalPartSize) {
super(dataType, memoryArea, blockNumber, byteOffset, bitOffset, numElements);
this.chunkTags = chunkTags;
+ this.curPartIndex = curPartIndex;
+ this.totalPartNumber = totalPartNumber;
+ this.totalPartSize = totalPartSize;
}
public Map<PlcTag, String> getChunkTags() {
return chunkTags;
}
+
+ public int getCurPartIndex() {
+ return curPartIndex;
+ }
+
+ public int getTotalPartNumber() {
+ return totalPartNumber;
+ }
+
+ public int getTotalPartSize() {
+ return totalPartSize;
+ }
+ }
+
+ public static class TagNameSize implements Comparable<TagNameSize> {
+ private final String tagName;
+ private final int tagSize;
+
+ public TagNameSize(String tagName, int tagSize) {
+ this.tagName = tagName;
+ this.tagSize = tagSize;
+ }
+
+ public String getTagName() {
+ return tagName;
+ }
+
+ public int getTagSize() {
+ return tagSize;
+ }
+
+ @Override
+ public int compareTo(TagNameSize o) {
+ int sizeComparison = Integer.compare(tagSize, o.tagSize);
+ if(sizeComparison != 0) {
+ return sizeComparison;
+ }
+ return tagName.compareTo(o.tagName);
+ }
+
}
}
diff --git a/plc4j/drivers/s7/src/test/java/org/apache/plc4x/java/s7/readwrite/ManualS71200DriverWithBlockOptimizerTest.java b/plc4j/drivers/s7/src/test/java/org/apache/plc4x/java/s7/readwrite/ManualS71200DriverWithBlockOptimizerTest.java
index 03d3fc9cc8..17d350f468 100644
--- a/plc4j/drivers/s7/src/test/java/org/apache/plc4x/java/s7/readwrite/ManualS71200DriverWithBlockOptimizerTest.java
+++ b/plc4j/drivers/s7/src/test/java/org/apache/plc4x/java/s7/readwrite/ManualS71200DriverWithBlockOptimizerTest.java
@@ -87,10 +87,10 @@ public class ManualS71200DriverWithBlockOptimizerTest extends ManualTest {
test.addTestCase("%DB4:50:LREAL", new PlcLREAL(2.71828182846D));
test.addTestCase("%DB4:136:CHAR", new PlcCHAR("H"));
test.addTestCase("%DB4:138:WCHAR", new PlcWCHAR("w"));
- //test.addTestCase("%DB4:140:STRING(10)", new PlcSTRING("hurz"));
- //test.addTestCase("%DB4:396:WSTRING(10)", new PlcWSTRING("wolf"));
- test.addTestCase("%DB4:140:STRING", new PlcSTRING("hurz"));
- test.addTestCase("%DB4:396:WSTRING", new PlcWSTRING("wolf"));
+ test.addTestCase("%DB4:140:STRING(200)", new PlcSTRING("hurz"));
+ test.addTestCase("%DB4:396:WSTRING(100)", new PlcWSTRING("wolf"));
+ //test.addTestCase("%DB4:140:STRING", new PlcSTRING("hurz"));
+ //test.addTestCase("%DB4:396:WSTRING", new PlcWSTRING("wolf"));
test.addTestCase("%DB4:58:TIME", new PlcTIME(Duration.parse("PT1.234S")));
test.addTestCase("%DB4:70:DATE", new PlcDATE(LocalDate.parse("1998-03-28")));
test.addTestCase("%DB4:72:TIME_OF_DAY", new PlcTIME_OF_DAY(LocalTime.parse("15:36:30.123")));